remove config in writer builder interface

Author: ZENOTME

crates/iceberg/src/writer/base_writer/data_file_writer.rs

@@ -22,45 +22,34 @@ use itertools::Itertools;
 
 use crate::spec::{DataContentType, DataFile, Struct};
 use crate::writer::file_writer::{FileWriter, FileWriterBuilder};
-use crate::writer::{CurrentFileStatus, IcebergWriter, IcebergWriterBuilder};
+use crate::writer::{CurrentWriterStatus, IcebergWriter, IcebergWriterBuilder};
 use crate::Result;
 
 /// Builder for `DataFileWriter`.
 #[derive(Clone)]
 pub struct DataFileWriterBuilder<B: FileWriterBuilder> {
     inner: B,
+    partition_value: Option<Struct>,
 }
 
 impl<B: FileWriterBuilder> DataFileWriterBuilder<B> {
     /// Create a new `DataFileWriterBuilder` using a `FileWriterBuilder`.
-    pub fn new(inner: B) -> Self {
-        Self { inner }
-    }
-}
-
-/// Config for `DataFileWriter`.
-pub struct DataFileWriterConfig {
-    partition_value: Struct,
-}
-
-impl DataFileWriterConfig {
-    /// Create a new `DataFileWriterConfig` with partition value.
-    pub fn new(partition_value: Option<Struct>) -> Self {
+    pub fn new(inner: B, partition_value: Option<Struct>) -> Self {
         Self {
-            partition_value: partition_value.unwrap_or(Struct::empty()),
+            inner,
+            partition_value,
         }
     }
 }
 
 #[async_trait::async_trait]
 impl<B: FileWriterBuilder> IcebergWriterBuilder for DataFileWriterBuilder<B> {
     type R = DataFileWriter<B>;
-    type C = DataFileWriterConfig;
 
-    async fn build(self, config: Self::C) -> Result<Self::R> {
+    async fn build(self) -> Result<Self::R> {
         Ok(DataFileWriter {
             inner_writer: Some(self.inner.clone().build().await?),
-            partition_value: config.partition_value,
+            partition_value: self.partition_value.unwrap_or(Struct::empty()),
         })
     }
 }
@@ -92,7 +81,7 @@ impl<B: FileWriterBuilder> IcebergWriter for DataFileWriter<B> {
     }
 }
 
-impl<B: FileWriterBuilder> CurrentFileStatus for DataFileWriter<B> {
+impl<B: FileWriterBuilder> CurrentWriterStatus for DataFileWriter<B> {
     fn current_file_path(&self) -> String {
         self.inner_writer.as_ref().unwrap().current_file_path()
     }
@@ -115,9 +104,7 @@ mod test {
 
     use crate::io::FileIOBuilder;
     use crate::spec::{DataContentType, DataFileFormat, Schema, Struct};
-    use crate::writer::base_writer::data_file_writer::{
-        DataFileWriterBuilder, DataFileWriterConfig,
-    };
+    use crate::writer::base_writer::data_file_writer::DataFileWriterBuilder;
     use crate::writer::file_writer::location_generator::test::MockLocationGenerator;
     use crate::writer::file_writer::location_generator::DefaultFileNameGenerator;
     use crate::writer::file_writer::ParquetWriterBuilder;
@@ -135,14 +122,12 @@ mod test {
 
         let pw = ParquetWriterBuilder::new(
             WriterProperties::builder().build(),
-            Arc::new(Schema::builder().build().unwrap()),
             file_io.clone(),
             location_gen,
             file_name_gen,
+            Arc::new(Schema::builder().build().unwrap()),
         );
-        let mut data_file_writer = DataFileWriterBuilder::new(pw)
-            .build(DataFileWriterConfig::new(None))
-            .await?;
+        let mut data_file_writer = DataFileWriterBuilder::new(pw, None).build().await?;
 
         let data_file = data_file_writer.close().await.unwrap();
         assert_eq!(data_file.len(), 1);

crates/iceberg/src/writer/base_writer/equality_delete_writer.rs

@@ -35,16 +35,18 @@ use crate::{Error, ErrorKind, Result};
 #[derive(Clone)]
 pub struct EqualityDeleteFileWriterBuilder<B: FileWriterBuilder> {
     inner: B,
+    config: EqualityDeleteWriterConfig,
 }
 
 impl<B: FileWriterBuilder> EqualityDeleteFileWriterBuilder<B> {
     /// Create a new `EqualityDeleteFileWriterBuilder` using a `FileWriterBuilder`.
-    pub fn new(inner: B) -> Self {
-        Self { inner }
+    pub fn new(inner: B, config: EqualityDeleteWriterConfig) -> Self {
+        Self { inner, config }
     }
 }
 
 /// Config for `EqualityDeleteWriter`.
+#[derive(Clone)]
 pub struct EqualityDeleteWriterConfig {
     // Field ids used to determine row equality in equality delete files.
     equality_ids: Vec<i32>,
@@ -108,14 +110,13 @@ impl EqualityDeleteWriterConfig {
 #[async_trait::async_trait]
 impl<B: FileWriterBuilder> IcebergWriterBuilder for EqualityDeleteFileWriterBuilder<B> {
     type R = EqualityDeleteFileWriter<B>;
-    type C = EqualityDeleteWriterConfig;
 
-    async fn build(self, config: Self::C) -> Result<Self::R> {
+    async fn build(self) -> Result<Self::R> {
         Ok(EqualityDeleteFileWriter {
             inner_writer: Some(self.inner.clone().build().await?),
-            projector: config.projector,
-            equality_ids: config.equality_ids,
-            partition_value: config.partition_value,
+            projector: self.config.projector,
+            equality_ids: self.config.equality_ids,
+            partition_value: self.config.partition_value,
         })
     }
 }
@@ -386,20 +387,19 @@ mod test {
             EqualityDeleteWriterConfig::new(equality_ids, Arc::new(schema), None).unwrap();
         let delete_schema =
             arrow_schema_to_schema(equality_config.projected_arrow_schema_ref()).unwrap();
-        let projector = equality_config.projector.clone();
 
         // prepare writer
         let pb = ParquetWriterBuilder::new(
             WriterProperties::builder().build(),
-            Arc::new(delete_schema),
             file_io.clone(),
             location_gen,
             file_name_gen,
+            Arc::new(delete_schema),
         );
-
-        let mut equality_delete_writer = EqualityDeleteFileWriterBuilder::new(pb)
-            .build(equality_config)
+        let mut equality_delete_writer = EqualityDeleteFileWriterBuilder::new(pb, equality_config)
+            .build()
             .await?;
+        let projector = equality_delete_writer.projector.clone();
 
         // write
         equality_delete_writer.write(to_write.clone()).await?;
@@ -556,13 +556,13 @@ mod test {
 
         let pb = ParquetWriterBuilder::new(
             WriterProperties::builder().build(),
-            Arc::new(delete_schema),
             file_io.clone(),
             location_gen,
             file_name_gen,
+            Arc::new(delete_schema),
         );
-        let mut equality_delete_writer = EqualityDeleteFileWriterBuilder::new(pb)
-            .build(config)
+        let mut equality_delete_writer = EqualityDeleteFileWriterBuilder::new(pb, config)
+            .build()
             .await?;
 
         // prepare data
@@ -640,10 +640,12 @@ mod test {
             Some(b""),
             Some(b"zzzz"),
         ])) as ArrayRef;
-        let to_write = RecordBatch::try_new(delete_arrow_schema.clone(), vec![
-            col0, col1, col2, col3, col4, col5, col6, col7, col8, col9, col10, col11, col12, col13,
-        ])
-        .unwrap();
+        let to_write =
+            RecordBatch::try_new(Arc::new(schema_to_arrow_schema(&schema).unwrap()), vec![
+                col0, col1, col2, col3, col4, col5, col6, col7, col8, col9, col10, col11, col12,
+                col13,
+            ])
+            .unwrap();
         equality_delete_writer.write(to_write.clone()).await?;
         let res = equality_delete_writer.close().await?;
         assert_eq!(res.len(), 1);

crates/iceberg/src/writer/file_writer/mod.rs

@@ -20,7 +20,7 @@
 use arrow_array::RecordBatch;
 use futures::Future;
 
-use super::CurrentFileStatus;
+use super::CurrentWriterStatus;
 use crate::spec::DataFileBuilder;
 use crate::Result;
 
@@ -41,7 +41,7 @@ pub trait FileWriterBuilder<O = DefaultOutput>: Send + Clone + 'static {
 }
 
 /// File writer focus on writing record batch to different physical file format.(Such as parquet. orc)
-pub trait FileWriter<O = DefaultOutput>: Send + CurrentFileStatus + 'static {
+pub trait FileWriter<O = DefaultOutput>: Send + CurrentWriterStatus + 'static {
     /// Write record batch to file.
     fn write(&mut self, batch: &RecordBatch) -> impl Future<Output = Result<()>> + Send;
     /// Close file writer.

crates/iceberg/src/writer/file_writer/parquet_writer.rs

@@ -42,36 +42,35 @@ use crate::spec::{
     visit_schema, DataFileBuilder, DataFileFormat, Datum, ListType, MapType, NestedFieldRef,
     PrimitiveType, Schema, SchemaRef, SchemaVisitor, StructType, Type,
 };
-use crate::writer::CurrentFileStatus;
+use crate::writer::CurrentWriterStatus;
 use crate::{Error, ErrorKind, Result};
 
 /// ParquetWriterBuilder is used to builder a [`ParquetWriter`]
 #[derive(Clone)]
 pub struct ParquetWriterBuilder<T: LocationGenerator, F: FileNameGenerator> {
     props: WriterProperties,
-    schema: SchemaRef,
-
     file_io: FileIO,
     location_generator: T,
     file_name_generator: F,
+    schema: SchemaRef,
 }
 
 impl<T: LocationGenerator, F: FileNameGenerator> ParquetWriterBuilder<T, F> {
     /// Create a new `ParquetWriterBuilder`
     /// To construct the write result, the schema should contain the `PARQUET_FIELD_ID_META_KEY` metadata for each field.
     pub fn new(
         props: WriterProperties,
-        schema: SchemaRef,
         file_io: FileIO,
         location_generator: T,
         file_name_generator: F,
+        schema: SchemaRef,
     ) -> Self {
         Self {
             props,
-            schema,
             file_io,
             location_generator,
             file_name_generator,
+            schema,
         }
     }
 }
@@ -80,20 +79,19 @@ impl<T: LocationGenerator, F: FileNameGenerator> FileWriterBuilder for ParquetWr
     type R = ParquetWriter;
 
     async fn build(self) -> crate::Result<Self::R> {
-        let arrow_schema: ArrowSchemaRef = Arc::new(self.schema.as_ref().try_into()?);
         let written_size = Arc::new(AtomicI64::new(0));
         let out_file = self.file_io.new_output(
             self.location_generator
                 .generate_location(&self.file_name_generator.generate_file_name()),
         )?;
         let inner_writer = TrackWriter::new(out_file.writer().await?, written_size.clone());
         let async_writer = AsyncFileWriter::new(inner_writer);
-        let writer =
-            AsyncArrowWriter::try_new(async_writer, arrow_schema.clone(), Some(self.props))
-                .map_err(|err| {
-                    Error::new(ErrorKind::Unexpected, "Failed to build parquet writer.")
-                        .with_source(err)
-                })?;
+        let arrow_schema: ArrowSchemaRef = Arc::new(self.schema.as_ref().try_into()?);
+        let writer = AsyncArrowWriter::try_new(async_writer, arrow_schema, Some(self.props))
+            .map_err(|err| {
+                Error::new(ErrorKind::Unexpected, "Failed to build parquet writer.")
+                    .with_source(err)
+            })?;
 
         Ok(ParquetWriter {
             schema: self.schema.clone(),
@@ -422,7 +420,7 @@ impl FileWriter for ParquetWriter {
     }
 }
 
-impl CurrentFileStatus for ParquetWriter {
+impl CurrentWriterStatus for ParquetWriter {
     fn current_file_path(&self) -> String {
         self.out_file.location().to_string()
     }
@@ -667,10 +665,10 @@ mod tests {
         // write data
         let mut pw = ParquetWriterBuilder::new(
             WriterProperties::builder().build(),
-            Arc::new(to_write.schema().as_ref().try_into().unwrap()),
             file_io.clone(),
             location_gen,
             file_name_gen,
+            Arc::new(to_write.schema().as_ref().try_into().unwrap()),
         )
         .build()
         .await?;
@@ -863,10 +861,10 @@ mod tests {
         // write data
         let mut pw = ParquetWriterBuilder::new(
             WriterProperties::builder().build(),
-            Arc::new(schema),
             file_io.clone(),
             location_gen,
             file_name_gen,
+            Arc::new(schema),
         )
         .build()
         .await?;
@@ -1053,10 +1051,10 @@ mod tests {
         // write data
         let mut pw = ParquetWriterBuilder::new(
             WriterProperties::builder().build(),
-            Arc::new(schema),
             file_io.clone(),
             loccation_gen,
             file_name_gen,
+            Arc::new(schema),
         )
         .build()
         .await?;

crates/iceberg/src/writer/mod.rs

@@ -63,10 +63,8 @@ pub trait IcebergWriterBuilder<I = DefaultInput, O = DefaultOutput>:
 {
     /// The associated writer type.
     type R: IcebergWriter<I, O>;
-    /// The associated writer config type used to build the writer.
-    type C;
     /// Build the iceberg writer.
-    async fn build(self, config: Self::C) -> Result<Self::R>;
+    async fn build(self) -> Result<Self::R>;
 }
 
 /// The iceberg writer used to write data to iceberg table.
@@ -83,7 +81,7 @@ pub trait IcebergWriter<I = DefaultInput, O = DefaultOutput>: Send + 'static {
 
 /// The current file status of iceberg writer. It implement for the writer which write a single
 /// file.
-pub trait CurrentFileStatus {
+pub trait CurrentWriterStatus {
     /// Get the current file path.
     fn current_file_path(&self) -> String;
     /// Get the current file row number.

crates/integration_tests/tests/append_data_file_test.rs

@@ -24,7 +24,7 @@ use arrow_array::{ArrayRef, BooleanArray, Int32Array, RecordBatch, StringArray};
 use futures::TryStreamExt;
 use iceberg::spec::{NestedField, PrimitiveType, Schema, Type};
 use iceberg::transaction::Transaction;
-use iceberg::writer::base_writer::data_file_writer::{DataFileWriterBuilder, DataFileWriterConfig};
+use iceberg::writer::base_writer::data_file_writer::DataFileWriterBuilder;
 use iceberg::writer::file_writer::location_generator::{
     DefaultFileNameGenerator, DefaultLocationGenerator,
 };
@@ -92,16 +92,13 @@ async fn test_append_data_file() {
     );
     let parquet_writer_builder = ParquetWriterBuilder::new(
         WriterProperties::default(),
-        table.metadata().current_schema().clone(),
         table.file_io().clone(),
         location_generator.clone(),
         file_name_generator.clone(),
+        table.metadata().current_schema().clone(),
     );
-    let data_file_writer_builder = DataFileWriterBuilder::new(parquet_writer_builder);
-    let mut data_file_writer = data_file_writer_builder
-        .build(DataFileWriterConfig::new(None))
-        .await
-        .unwrap();
+    let data_file_writer_builder = DataFileWriterBuilder::new(parquet_writer_builder, None);
+    let mut data_file_writer = data_file_writer_builder.build().await.unwrap();
     let col1 = StringArray::from(vec![Some("foo"), Some("bar"), None, Some("baz")]);
     let col2 = Int32Array::from(vec![Some(1), Some(2), Some(3), Some(4)]);
     let col3 = BooleanArray::from(vec![Some(true), Some(false), None, Some(false)]);

crates/integration_tests/tests/conflict_commit_test.rs

@@ -24,7 +24,7 @@ use arrow_array::{ArrayRef, BooleanArray, Int32Array, RecordBatch, StringArray};
 use futures::TryStreamExt;
 use iceberg::spec::{NestedField, PrimitiveType, Schema, Type};
 use iceberg::transaction::Transaction;
-use iceberg::writer::base_writer::data_file_writer::{DataFileWriterBuilder, DataFileWriterConfig};
+use iceberg::writer::base_writer::data_file_writer::DataFileWriterBuilder;
 use iceberg::writer::file_writer::location_generator::{
     DefaultFileNameGenerator, DefaultLocationGenerator,
 };
@@ -91,16 +91,13 @@ async fn test_append_data_file_conflict() {
     );
     let parquet_writer_builder = ParquetWriterBuilder::new(
         WriterProperties::default(),
-        table.metadata().current_schema().clone(),
         table.file_io().clone(),
         location_generator.clone(),
         file_name_generator.clone(),
+        table.metadata().current_schema().clone(),
     );
-    let data_file_writer_builder = DataFileWriterBuilder::new(parquet_writer_builder);
-    let mut data_file_writer = data_file_writer_builder
-        .build(DataFileWriterConfig::new(None))
-        .await
-        .unwrap();
+    let data_file_writer_builder = DataFileWriterBuilder::new(parquet_writer_builder, None);
+    let mut data_file_writer = data_file_writer_builder.build().await.unwrap();
     let col1 = StringArray::from(vec![Some("foo"), Some("bar"), None, Some("baz")]);
     let col2 = Int32Array::from(vec![Some(1), Some(2), Some(3), Some(4)]);
     let col3 = BooleanArray::from(vec![Some(true), Some(false), None, Some(false)]);