feat: write command improvements

rust/Cargo.toml

@@ -13,12 +13,18 @@ readme = "README.md"
 edition = "2021"
 
 [dependencies]
-arrow = { version = "37.0.0", optional = true }
+arrow = { version = "37", optional = true }
+arrow-array = { version = "37", optional = true }
+arrow-cast = { version = "37", optional = true }
+arrow-schema = { version = "37", optional = true }
 async-trait = "0.1"
 bytes = "1"
 chrono = { version = "0.4.22", default-features = false, features = ["clock"] }
 cfg-if = "1"
-datafusion-objectstore-hdfs = { version = "0.1.3", default-features = false, features = ["hdfs3", "try_spawn_blocking"], optional = true }
+datafusion-objectstore-hdfs = { version = "0.1.3", default-features = false, features = [
+    "hdfs3",
+    "try_spawn_blocking",
+], optional = true }
 errno = "0.3"
 futures = "0.3"
 itertools = "0.10"
@@ -91,6 +97,7 @@ glibc_version = { path = "../glibc_version", version = "0.1" }
 
 [features]
 azure = ["object_store/azure"]
+arrow = ["dep:arrow", "arrow-array", "arrow-cast", "arrow-schema"]
 default = ["arrow", "parquet"]
 datafusion = [
     "dep:datafusion",

rust/src/operations/filesystem_check.rs

@@ -32,7 +32,7 @@ use url::{ParseError, Url};
 #[derive(Debug)]
 pub struct FileSystemCheckBuilder {
     /// A snapshot of the to-be-checked table's state
-    state: DeltaTableState,
+    snapshot: DeltaTableState,
     /// Delta object store for handling data files
     store: Arc<DeltaObjectStore>,
     /// Don't remove actions to the table log. Just determine which files can be removed
@@ -70,7 +70,7 @@ impl FileSystemCheckBuilder {
     /// Create a new [`FileSystemCheckBuilder`]
     pub fn new(store: Arc<DeltaObjectStore>, state: DeltaTableState) -> Self {
         FileSystemCheckBuilder {
-            state,
+            snapshot: state,
             store,
             dry_run: false,
         }
@@ -84,10 +84,10 @@ impl FileSystemCheckBuilder {
 
     async fn create_fsck_plan(&self) -> DeltaResult<FileSystemCheckPlan> {
         let mut files_relative: HashMap<&str, &Add> =
-            HashMap::with_capacity(self.state.files().len());
+            HashMap::with_capacity(self.snapshot.files().len());
         let store = self.store.clone();
 
-        for active in self.state.files() {
+        for active in self.snapshot.files() {
             if is_absolute_path(&active.path)? {
                 return Err(DeltaTableError::Generic(
                     "Filesystem check does not support absolute paths".to_string(),
@@ -174,16 +174,16 @@ impl std::future::IntoFuture for FileSystemCheckBuilder {
             let plan = this.create_fsck_plan().await?;
             if this.dry_run {
                 return Ok((
-                    DeltaTable::new_with_state(this.store, this.state),
+                    DeltaTable::new_with_state(this.store, this.snapshot),
                     FileSystemCheckMetrics {
                         files_removed: plan.files_to_remove.into_iter().map(|f| f.path).collect(),
                         dry_run: true,
                     },
                 ));
             }
 
-            let metrics = plan.execute(&this.state).await?;
-            let mut table = DeltaTable::new_with_state(this.store, this.state);
+            let metrics = plan.execute(&this.snapshot).await?;
+            let mut table = DeltaTable::new_with_state(this.store, this.snapshot);
             table.update().await?;
             Ok((table, metrics))
         })

rust/src/operations/mod.rs

@@ -109,9 +109,7 @@ impl DeltaOps {
     #[cfg(feature = "datafusion")]
     #[must_use]
     pub fn write(self, batches: impl IntoIterator<Item = RecordBatch>) -> WriteBuilder {
-        WriteBuilder::default()
-            .with_input_batches(batches)
-            .with_object_store(self.0.object_store())
+        WriteBuilder::new(self.0.object_store(), self.0.state).with_input_batches(batches)
     }
 
     /// Vacuum stale files from delta table