Upgrade to delta-rs 0.29.2 and pull CommitProperties through for writers

Closes #39

Signed-off-by: R. Tyler Croy <rtyler@buoyantdata.com>

Author: rtyler

Cargo.toml

@@ -20,7 +20,7 @@ anyhow = "=1"
 chrono = "0.4"
 aws_lambda_events = { version = "0.15", default-features = false, features = ["sns", "sqs", "s3"] }
 # The datafusion feature is required to support invariants which may be in error, but is required as of currently released 0.18.2
-deltalake = { version = "0.29.0", features = ["s3", "json", "datafusion"] }
+deltalake = { version = "0.29.2", features = ["s3", "json", "datafusion"] }
 #deltalake = { git = "https://github.com/delta-io/delta-rs", branch = "main", features = ["s3", "json", "datafusion"]}
 #deltalake = { path = "../../delta-io/delta-rs/crates/deltalake", features = ["s3", "json", "datafusion"]}
 futures = { version = "0.3" }

crates/oxbow/src/lib.rs

@@ -4,6 +4,7 @@
 use deltalake::arrow::datatypes::Schema as ArrowSchema;
 use deltalake::kernel::engine::arrow_conversion::TryFromArrow;
 use deltalake::kernel::schema::{Schema, StructField};
+use deltalake::kernel::transaction::{CommitBuilder, CommitProperties};
 use deltalake::logstore::{LogStoreRef, ObjectStoreRef, StorageConfig, logstore_for};
 use deltalake::operations::create::CreateBuilder;
 use deltalake::parquet::arrow::async_reader::{
@@ -207,18 +208,21 @@ pub async fn create_table_with(
         .await
 }
 
+/// Return the common default [CommitProperties] to be used by issuing commits in oxbow
+pub(crate) fn default_commit_properties() -> CommitProperties {
+    CommitProperties::default()
+        // Turn off cleanup of expired logs. After each commit, versions are
+        // scanned (ListObject on s3) to clean up expired entries. This creates
+        // significant overhead when there are many versions on each commit.
+        .with_cleanup_expired_logs(Some(false))
+}
+
 /// Commit the given [Action]s to the [DeltaTable]
 pub async fn commit_to_table(actions: &[Action], table: &DeltaTable) -> DeltaResult<i64> {
-    use deltalake::kernel::transaction::{CommitBuilder, CommitProperties};
     if actions.is_empty() {
         return table.version().ok_or(DeltaTableError::NotInitialized);
     }
-    let commit = CommitProperties::default()
-        // Turn off cleanup of expired logs. After each commit, versions are
-        // scanned (ListObject on s3) to clean up expired entries. This creates
-        // significant overhead when there are many versions on each commit.
-        .with_cleanup_expired_logs(Some(false));
-    let pre_commit = CommitBuilder::from(commit)
+    let pre_commit = CommitBuilder::from(default_commit_properties())
         .with_actions(actions.to_vec())
         .build(
             Some(table.snapshot()?),
@@ -1226,7 +1230,6 @@ mod tests {
         let table_url = Url::from_file_path(&table_path).expect("Failed to parse local path");
         let store =
             logstore_for(table_url, StorageConfig::default()).expect("Failed to get object store");
-
         let files = discover_parquet_files(store.object_store(None).clone())
             .await
             .expect("Failed to discover parquet files");

crates/oxbow/src/write.rs

@@ -17,7 +17,8 @@ pub async fn append_batches(
     mut table: DeltaTable,
     batches: impl IntoIterator<Item = Result<RecordBatch, ArrowError>>,
 ) -> DeltaResult<DeltaTable> {
-    let mut writer = RecordBatchWriter::for_table(&table)?;
+    let mut writer = RecordBatchWriter::for_table(&table)?
+        .with_commit_properties(super::default_commit_properties());
     let mut written = false;
 
     trace!("Iterating through batches to write");
@@ -47,7 +48,8 @@ pub async fn append_values(
     let schema = table.snapshot()?.snapshot().arrow_schema();
     debug!("Attempting to append values with schema: {schema:?}");
 
-    let mut writer = RecordBatchWriter::for_table(&table)?;
+    let mut writer = RecordBatchWriter::for_table(&table)?
+        .with_commit_properties(super::default_commit_properties());
     let mut written = false;
 
     for value in values {

lambdas/group-events/src/main.rs

@@ -152,7 +152,7 @@ mod tests {
         let event: S3Event = serde_json::from_str(&buf).expect("Failed to parse");
         assert_eq!(4, event.records.len());
 
-        let fifos = segmented_by_prefix(&event.records, Some(4)).expect("Failed to segment");
+        let fifos = segmented_by_prefix(&event.records, Some(8)).expect("Failed to segment");
         println!("fifos: {fifos:#?}");
         assert!(
             fifos.keys().len() > 2,