refactor: use iceberg-rust rolling writer directly

Cargo.toml

@@ -18,17 +18,17 @@ datafusion = "52.2"
 
 # Local workspace members
 futures = "0.3.17"
-iceberg = { git = "https://github.com/risingwavelabs/iceberg-rust.git", rev = "72b0729b94435a958554e009a940502a3ebeb88a", features = [
+iceberg = { git = "https://github.com/risingwavelabs/iceberg-rust.git", rev = "baaa9c7b2deb3e744db21712e4b6ced5891a6012", features = [
     "storage-s3",
     "storage-gcs",
     "storage-azblob",
     "storage-azdls",
 ] }
-iceberg-catalog-glue = { git = "https://github.com/risingwavelabs/iceberg-rust.git", rev = "72b0729b94435a958554e009a940502a3ebeb88a" }
-iceberg-catalog-memory = { git = "https://github.com/risingwavelabs/iceberg-rust.git", rev = "72b0729b94435a958554e009a940502a3ebeb88a" }
-iceberg-catalog-rest = { git = "https://github.com/risingwavelabs/iceberg-rust.git", rev = "72b0729b94435a958554e009a940502a3ebeb88a" }
+iceberg-catalog-glue = { git = "https://github.com/risingwavelabs/iceberg-rust.git", rev = "baaa9c7b2deb3e744db21712e4b6ced5891a6012" }
+iceberg-catalog-memory = { git = "https://github.com/risingwavelabs/iceberg-rust.git", rev = "baaa9c7b2deb3e744db21712e4b6ced5891a6012" }
+iceberg-catalog-rest = { git = "https://github.com/risingwavelabs/iceberg-rust.git", rev = "baaa9c7b2deb3e744db21712e4b6ced5891a6012" }
 iceberg-compaction-core = { path = "./core" }
-iceberg-datafusion = { git = "https://github.com/risingwavelabs/iceberg-rust.git", rev = "72b0729b94435a958554e009a940502a3ebeb88a" }
+iceberg-datafusion = { git = "https://github.com/risingwavelabs/iceberg-rust.git", rev = "baaa9c7b2deb3e744db21712e4b6ced5891a6012" }
 
 parquet = { version = "57.1", features = ["async"] }
 port_scanner = "0.1.5"

core/src/config/mod.rs

@@ -360,10 +360,32 @@ pub struct CompactionExecutionConfig {
     #[builder(default = "DEFAULT_NORMALIZED_COLUMN_IDENTIFIERS")]
     pub enable_normalized_column_identifiers: bool,
 
+    /// Deprecated: this setting is no longer used after switching to the upstream
+    /// `RollingFileWriter`.
+    ///
+    /// It remains temporarily for backward compatibility and will be removed in a
+    /// future change.
+    #[deprecated(
+        note = "unused after switching to the upstream RollingFileWriter; this field is now a no-op and will be removed in a future change"
+    )]
     #[builder(default = "DEFAULT_ENABLE_DYNAMIC_SIZE_ESTIMATION")]
+    #[builder_setter_attr(deprecated(
+        note = "unused after switching to the upstream RollingFileWriter; this setter is now a no-op and will be removed in a future change"
+    ))]
     pub enable_dynamic_size_estimation: bool,
 
+    /// Deprecated: this setting is no longer used after switching to the upstream
+    /// `RollingFileWriter`.
+    ///
+    /// It remains temporarily for backward compatibility and will be removed in a
+    /// future change.
+    #[deprecated(
+        note = "unused after switching to the upstream RollingFileWriter; this field is now a no-op and will be removed in a future change"
+    )]
     #[builder(default = "DEFAULT_SIZE_ESTIMATION_SMOOTHING_FACTOR")]
+    #[builder_setter_attr(deprecated(
+        note = "unused after switching to the upstream RollingFileWriter; this setter is now a no-op and will be removed in a future change"
+    ))]
     pub size_estimation_smoothing_factor: f64,
 
     /// Maximum concurrent compaction plans in `compact()` method.

core/src/executor/datafusion/mod.rs

@@ -38,7 +38,6 @@ use super::{CompactionExecutor, RewriteFilesStat};
 use crate::CompactionError;
 use crate::config::CompactionExecutionConfig;
 use crate::error::Result;
-use crate::executor::iceberg_writer::rolling_iceberg_writer;
 pub mod datafusion_processor;
 use super::{RewriteFilesRequest, RewriteFilesResponse};
 pub mod file_scan_task_table_provider;
@@ -167,6 +166,14 @@ pub fn build_iceberg_data_file_writer(
     partition_spec: Arc<PartitionSpec>,
     execution_config: Arc<CompactionExecutionConfig>,
 ) -> Result<Box<dyn IcebergWriter>> {
+    let target_file_size =
+        usize::try_from(execution_config.target_file_size_bytes).map_err(|_| {
+            CompactionError::Config(format!(
+                "target_file_size_bytes {} exceeds platform usize",
+                execution_config.target_file_size_bytes
+            ))
+        })?;
+
     let data_file_builder = {
         let parquet_writer_builder = ParquetWriterBuilder::new(
             execution_config.write_parquet_properties.clone(),
@@ -180,27 +187,18 @@ pub fn build_iceberg_data_file_writer(
             iceberg::spec::DataFileFormat::Parquet,
         );
 
-        // Noop wrapper for `DataFileWriterBuilder`
         let rolling_writer_builder = RollingFileWriterBuilder::new(
             parquet_writer_builder,
-            usize::MAX, // No rolling based on row count
+            target_file_size,
             file_io,
             location_generator,
             file_name_generator,
-        );
+        )
+        .with_max_concurrent_closes(execution_config.max_concurrent_closes);
 
         DataFileWriterBuilder::new(rolling_writer_builder)
     };
 
-    let rolling_iceberg_writer_builder =
-        rolling_iceberg_writer::RollingIcebergWriterBuilder::new(data_file_builder)
-            .with_target_file_size(execution_config.target_file_size_bytes)
-            .with_max_concurrent_closes(execution_config.max_concurrent_closes)
-            .with_dynamic_size_estimation(execution_config.enable_dynamic_size_estimation)
-            .with_size_estimation_smoothing_factor(
-                execution_config.size_estimation_smoothing_factor,
-            );
-
     let partition_splitter = if partition_spec.is_unpartitioned() {
         None
     } else {
@@ -211,7 +209,7 @@ pub fn build_iceberg_data_file_writer(
     };
 
     let iceberg_task_writer = TaskWriter::new_with_partition_splitter(
-        rolling_iceberg_writer_builder,
+        data_file_builder,
         true,
         schema,
         partition_spec,