some cleanup

Author: CTTY

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

@@ -20,12 +20,12 @@
 use arrow_array::RecordBatch;
 use itertools::Itertools;
 
-use crate::Result;
 use crate::spec::{DEFAULT_PARTITION_SPEC_ID, DataContentType, DataFile, PartitionKey, Struct};
 use crate::writer::file_writer::FileWriterBuilder;
 use crate::writer::file_writer::location_generator::{FileNameGenerator, LocationGenerator};
 use crate::writer::file_writer::rolling_writer::RollingFileWriter;
 use crate::writer::{CurrentFileStatus, IcebergWriter, IcebergWriterBuilder};
+use crate::{Error, ErrorKind, Result};
 
 /// Builder for `DataFileWriter`.
 #[derive(Clone, Debug)]
@@ -83,26 +83,32 @@ impl<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator> IcebergWr
     }
 
     async fn close(&mut self) -> Result<Vec<DataFile>> {
-        let writer = self.inner_writer.take().unwrap();
-        Ok(writer
-            .close()
-            .await?
-            .into_iter()
-            .map(|mut res| {
-                res.content(DataContentType::Data);
-                res.partition(
-                    self.partition_key
-                        .as_ref()
-                        .map_or(Struct::empty(), |pk| pk.data().clone()),
-                );
-                res.partition_spec_id(
-                    self.partition_key
-                        .as_ref()
-                        .map_or(DEFAULT_PARTITION_SPEC_ID, |pk| pk.spec().spec_id()),
-                );
-                res.build().expect("Guaranteed to be valid")
-            })
-            .collect_vec())
+        if let Some(writer) = self.inner_writer.take() {
+            Ok(writer
+                .close()
+                .await?
+                .into_iter()
+                .map(|mut res| {
+                    res.content(DataContentType::Data);
+                    res.partition(
+                        self.partition_key
+                            .as_ref()
+                            .map_or(Struct::empty(), |pk| pk.data().clone()),
+                    );
+                    res.partition_spec_id(
+                        self.partition_key
+                            .as_ref()
+                            .map_or(DEFAULT_PARTITION_SPEC_ID, |pk| pk.spec().spec_id()),
+                    );
+                    res.build().expect("Guaranteed to be valid")
+                })
+                .collect_vec())
+        } else {
+            Err(Error::new(
+                ErrorKind::Unexpected,
+                "Data file writer has been closed.",
+            ))
+        }
     }
 }
 

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

@@ -25,64 +25,6 @@ use crate::writer::CurrentFileStatus;
 use crate::writer::file_writer::location_generator::{FileNameGenerator, LocationGenerator};
 use crate::writer::file_writer::{FileWriter, FileWriterBuilder};
 use crate::{Error, ErrorKind, Result};
-//
-// /// Builder for creating a `RollingFileWriter` that rolls over to a new file
-// /// when the data size exceeds a target threshold.
-// #[derive(Clone)]
-// pub struct RollingFileWriterBuilder<
-//     B: FileWriterBuilder,
-//     L: LocationGenerator,
-//     F: FileNameGenerator,
-// > {
-//     inner_builder: B,
-//     target_file_size: usize,
-//     file_io: FileIO,
-//     location_generator: L,
-//     file_name_generator: F,
-// }
-//
-// impl<B: FileWriterBuilder, L: LocationGenerator, F: FileNameGenerator>
-//     RollingFileWriterBuilder<B, L, F>
-// {
-//     /// Creates a new `RollingFileWriterBuilder` with the specified inner builder and target size.
-//     ///
-//     /// # Arguments
-//     ///
-//     /// * `inner_builder` - The builder for the underlying file writer
-//     /// * `target_file_size` - The target size in bytes before rolling over to a new file
-//     ///
-//     /// NOTE: The `target_file_size` does not exactly reflect the final size on physical storage.
-//     /// This is because the input size is based on the Arrow in-memory format and cannot precisely control rollover behavior.
-//     /// The actual file size on disk is expected to be slightly larger than `target_file_size`.
-//     pub fn new(
-//         inner_builder: B,
-//         target_file_size: usize,
-//         file_io: FileIO,
-//         location_generator: L,
-//         file_name_generator: F,
-//     ) -> Self {
-//         Self {
-//             inner_builder,
-//             target_file_size,
-//             file_io,
-//             location_generator,
-//             file_name_generator,
-//         }
-//     }
-// }
-
-// impl<B: FileWriterBuilder> FileWriterBuilder for RollingFileWriterBuilder<B> {
-//     type R = RollingFileWriter<B>;
-//
-//     async fn build(self) -> Result<Self::R> {
-//         Ok(RollingFileWriter {
-//             inner: None,
-//             inner_builder: self.inner_builder,
-//             target_file_size: self.target_file_size,
-//             data_file_builders: vec![],
-//         })
-//     }
-// }
 
 /// A writer that automatically rolls over to a new file when the data size
 /// exceeds a target threshold.

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

@@ -21,16 +21,6 @@ use crate::Result;
 use crate::spec::PartitionKey;
 use crate::writer::{DefaultInput, DefaultOutput};
 
-#[async_trait::async_trait]
-pub trait PartitioningWriterBuilder<I = DefaultInput, O = DefaultOutput>:
-    Send + Clone + 'static
-{
-    /// todo doc
-    type R: PartitioningWriter<I, O>;
-    /// todo doc
-    async fn build(self) -> Result<Self::R>;
-}
-
 #[async_trait::async_trait]
 pub trait PartitioningWriter<I = DefaultInput, O = DefaultOutput>: Send + 'static {
     /// todo doc