refactor(writer): Make writer builders non-consuming in build

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

@@ -47,13 +47,13 @@ where
 #[async_trait::async_trait]
 impl<B, L, F> IcebergWriterBuilder for DataFileWriterBuilder<B, L, F>
 where
-    B: FileWriterBuilder,
-    L: LocationGenerator,
-    F: FileNameGenerator,
+    B: FileWriterBuilder + Sync,
+    L: LocationGenerator + Sync,
+    F: FileNameGenerator + Sync,
 {
     type R = DataFileWriter<B, L, F>;
 
-    async fn build(self, partition_key: Option<PartitionKey>) -> Result<Self::R> {
+    async fn build(&self, partition_key: Option<PartitionKey>) -> Result<Self::R> {
         Ok(DataFileWriter {
             inner: Some(self.inner.clone().build()),
             partition_key,

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

@@ -117,17 +117,17 @@ impl EqualityDeleteWriterConfig {
 #[async_trait::async_trait]
 impl<B, L, F> IcebergWriterBuilder for EqualityDeleteFileWriterBuilder<B, L, F>
 where
-    B: FileWriterBuilder,
-    L: LocationGenerator,
-    F: FileNameGenerator,
+    B: FileWriterBuilder + Sync,
+    L: LocationGenerator + Sync,
+    F: FileNameGenerator + Sync,
 {
     type R = EqualityDeleteFileWriter<B, L, F>;
 
-    async fn build(self, partition_key: Option<PartitionKey>) -> Result<Self::R> {
+    async fn build(&self, partition_key: Option<PartitionKey>) -> Result<Self::R> {
         Ok(EqualityDeleteFileWriter {
             inner: Some(self.inner.clone().build()),
-            projector: self.config.projector,
-            equality_ids: self.config.equality_ids,
+            projector: self.config.projector.clone(),
+            equality_ids: self.config.equality_ids.clone(),
             partition_key,
         })
     }

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

@@ -40,7 +40,7 @@ pub trait FileWriterBuilder<O = DefaultOutput>: Send + Clone + 'static {
     /// The associated file writer type.
     type R: FileWriter<O>;
     /// Build file writer.
-    fn build(self, output_file: OutputFile) -> impl Future<Output = Result<Self::R>> + Send;
+    fn build(&self, output_file: OutputFile) -> impl Future<Output = Result<Self::R>> + Send;
 }
 
 /// File writer focus on writing record batch to different physical file format.(Such as parquet. orc)

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

@@ -81,11 +81,11 @@ impl ParquetWriterBuilder {
 impl FileWriterBuilder for ParquetWriterBuilder {
     type R = ParquetWriter;
 
-    async fn build(self, output_file: OutputFile) -> Result<Self::R> {
+    async fn build(&self, output_file: OutputFile) -> Result<Self::R> {
         Ok(ParquetWriter {
             schema: self.schema.clone(),
             inner_writer: None,
-            writer_properties: self.props,
+            writer_properties: self.props.clone(),
             current_row_num: 0,
             output_file,
             nan_value_count_visitor: NanValueCountVisitor::new_with_match_mode(self.match_mode),

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

@@ -103,15 +103,15 @@ where
     }
 
     /// Build a new [`RollingFileWriter`].
-    pub fn build(self) -> RollingFileWriter<B, L, F> {
+    pub fn build(&self) -> RollingFileWriter<B, L, F> {
         RollingFileWriter {
             inner: None,
-            inner_builder: self.inner_builder,
+            inner_builder: self.inner_builder.clone(),
             target_file_size: self.target_file_size,
             data_file_builders: vec![],
-            file_io: self.file_io,
-            location_generator: self.location_generator,
-            file_name_generator: self.file_name_generator,
+            file_io: self.file_io.clone(),
+            location_generator: self.location_generator.clone(),
+            file_name_generator: self.file_name_generator.clone(),
         }
     }
 }

crates/iceberg/src/writer/mod.rs

@@ -145,10 +145,10 @@
 //! }
 //!
 //! #[async_trait::async_trait]
-//! impl<B: IcebergWriterBuilder> IcebergWriterBuilder for LatencyRecordWriterBuilder<B> {
+//! impl<B: IcebergWriterBuilder + Sync> IcebergWriterBuilder for LatencyRecordWriterBuilder<B> {
 //!     type R = LatencyRecordWriter<B::R>;
 //!
-//!     async fn build(self, partition_key: Option<PartitionKey>) -> Result<Self::R> {
+//!     async fn build(&self, partition_key: Option<PartitionKey>) -> Result<Self::R> {
 //!         Ok(LatencyRecordWriter {
 //!             inner_writer: self.inner_writer_builder.build(partition_key).await?,
 //!         })
@@ -404,7 +404,7 @@ pub trait IcebergWriterBuilder<I = DefaultInput, O = DefaultOutput>:
     /// The associated writer type.
     type R: IcebergWriter<I, O>;
     /// Build the iceberg writer with an optional partition key.
-    async fn build(self, partition_key: Option<PartitionKey>) -> Result<Self::R>;
+    async fn build(&self, partition_key: Option<PartitionKey>) -> Result<Self::R>;
 }
 
 /// The iceberg writer used to write data to iceberg table.