Upgrade the entire oxbow suite to delta-rs 0.29.x

This release has newer performance improvements on log replay and should
noticeably reduce memory consumption of the lambdas with larger tables

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

Author: rtyler

Cargo.toml

@@ -7,7 +7,7 @@ members = [
 resolver = "3"
 
 [workspace.package]
-version = "1.7.2"
+version = "1.8.0"
 edition = "2024"
 keywords = ["deltalake", "parquet", "lambda", "delta", "sqs"]
 homepage = "https://github.com/buoyant-data/oxbow"
@@ -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.26.2", features = ["s3", "json", "datafusion"] }
+deltalake = { version = "0.29.0", 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" }

cli/Cargo.toml

@@ -10,6 +10,7 @@ anyhow = { workspace = true }
 tokio = { workspace = true, features = ["full"] }
 tracing = { workspace = true }
 tracing-subscriber = { workspace = true }
+url = { workspace = true }
 
 gumdrop = "=0.8"
 oxbow = { path = "../crates/oxbow" }

cli/src/main.rs

@@ -4,6 +4,7 @@
 
 use gumdrop::Options;
 use tracing::log::*;
+use url::Url;
 
 /*
  * Flags is a structure for managing command linke parameters
@@ -39,7 +40,7 @@ async fn main() -> Result<(), anyhow::Error> {
     info!("Starting oxbow");
     let flags = Flags::parse_args_default_or_exit();
     debug!("Options as read: {:?}", flags);
-    let location = table_location(&flags)?;
+    let location = Url::parse(&table_location(&flags)?)?;
     info!("Using the table location of: {:?}", location);
 
     oxbow::convert(&location, None)

crates/oxbow/src/lib.rs

@@ -4,14 +4,19 @@
 ///
 use dynamodb_lock::Region;
 use tracing::log::*;
+use url::Url;
 
 use std::collections::HashMap;
 
 ///
 ///Wrapper aroudn [deltalake::open_table] which will open the table with the appropriate storage
 ///options needed for locking
 pub async fn open_table(table_uri: &str) -> deltalake::DeltaResult<deltalake::DeltaTable> {
-    deltalake::open_table_with_storage_options(&table_uri, storage_options(table_uri)).await
+    deltalake::open_table_with_storage_options(
+        Url::parse(table_uri).expect("Fail"),
+        storage_options(table_uri),
+    )
+    .await
 }
 
 /// Default storage options for using with `deltalake` calls

crates/oxbow/src/lock.rs

@@ -1,6 +1,5 @@
 use chrono::prelude::*;
 use deltalake::arrow::array::RecordBatch;
-use deltalake::arrow::datatypes::Schema as ArrowSchema;
 use deltalake::arrow::error::ArrowError;
 use deltalake::arrow::json::reader::ReaderBuilder;
 use deltalake::writer::{DeltaWriter, record_batch::RecordBatchWriter};
@@ -45,18 +44,15 @@ pub async fn append_values(
     mut table: DeltaTable,
     values: impl IntoIterator<Item: AsRef<str>>,
 ) -> DeltaResult<DeltaTable> {
-    let schema = table.get_schema()?;
+    let schema = table.snapshot()?.snapshot().arrow_schema();
     debug!("Attempting to append values with schema: {schema:?}");
-    let schema = ArrowSchema::try_from(schema)?;
 
     let mut writer = RecordBatchWriter::for_table(&table)?;
     let mut written = false;
 
     for value in values {
         let cursor: Cursor<&str> = Cursor::new(value.as_ref());
-        let reader = ReaderBuilder::new(schema.clone().into())
-            .build(cursor)
-            .unwrap();
+        let reader = ReaderBuilder::new(schema.clone()).build(cursor).unwrap();
 
         for res in reader {
             match res {
@@ -131,8 +127,7 @@ mod tests {
     use deltalake::*;
 
     async fn setup_test_table() -> DeltaResult<DeltaTable> {
-        DeltaOps::try_from_uri("memory://")
-            .await?
+        DeltaOps::new_in_memory()
             .create()
             .with_table_name("test")
             .with_column("id", DataType::INTEGER, true, None)
@@ -146,8 +141,7 @@ mod tests {
         use std::fs::File;
         use std::io::BufReader;
 
-        let table = DeltaOps::try_from_uri("memory://")
-            .await?
+        let table = DeltaOps::new_in_memory()
             .create()
             .with_table_name("test")
             .with_column("current", DataType::BOOLEAN, true, None)
@@ -157,12 +151,12 @@ mod tests {
 
         let buf = File::open("../../tests/data/senators.jsonl")?;
         let reader = BufReader::new(buf);
-        let schema = ArrowSchema::try_from(table.get_schema()?)?;
+        let schema = table.snapshot()?.snapshot().arrow_schema();
 
-        let json = deltalake::arrow::json::ReaderBuilder::new(schema.into()).build(reader)?;
+        let json = deltalake::arrow::json::ReaderBuilder::new(schema).build(reader)?;
 
         let table = append_batches(table, json).await?;
-        assert_eq!(table.version(), 1);
+        assert_eq!(table.version(), Some(1));
         Ok(())
     }
 
@@ -178,7 +172,7 @@ mod tests {
             .await
             .expect("Failed to do nothing");
 
-        assert_eq!(table.version(), 1);
+        assert_eq!(table.version(), Some(1));
         Ok(())
     }
 
@@ -191,9 +185,8 @@ mod tests {
         "#;
 
         let cursor = Cursor::new(jsonl);
-        let schema = table.get_schema()?;
-        let schema = ArrowSchema::try_from(schema)?;
-        let mut reader = ReaderBuilder::new(schema.into()).build(cursor).unwrap();
+        let schema = table.snapshot()?.snapshot().arrow_schema();
+        let mut reader = ReaderBuilder::new(schema).build(cursor).unwrap();
 
         while let Some(Ok(batch)) = reader.next() {
             let batch = augment_with_ds(batch)?;
@@ -215,12 +208,13 @@ mod tests {
             .await
             .expect("Failed to do nothing");
 
-        assert_eq!(table.version(), 1);
+        assert_eq!(table.version(), Some(1));
         Ok(())
     }
 
     #[tokio::test]
     async fn test_append_values_checkpoint() -> DeltaResult<()> {
+        use deltalake::table::config::TablePropertiesExt;
         let mut table = setup_test_table().await?;
 
         for _ in 0..101 {
@@ -235,7 +229,7 @@ mod tests {
 
         if let Some(state) = table.state.as_ref() {
             // The default is expected to be 100
-            assert_eq!(100, state.table_config().checkpoint_interval());
+            assert_eq!(100, state.table_config().checkpoint_interval().get());
         }
 
         use deltalake::Path;
@@ -245,7 +239,7 @@ mod tests {
             .await?;
 
         assert_ne!(0, checkpoint.size);
-        assert_eq!(table.version(), 101);
+        assert_eq!(table.version(), Some(101));
         Ok(())
     }
 }

crates/oxbow/src/write.rs

@@ -9,7 +9,6 @@ use deltalake::delta_datafusion::DeltaCdfTableProvider;
 use deltalake::{DeltaOps, DeltaResult};
 use lambda_runtime::{Error, LambdaEvent, run, service_fn, tracing};
 use object_store::ObjectStore;
-use object_store::PutPayload;
 use object_store::path::Path;
 use object_store::prefix::PrefixStore;
 use oxbow_lambda_shared::*;
@@ -68,9 +67,9 @@ async fn function_handler(event: LambdaEvent<SqsEvent>) -> DeltaResult<(), Error
                     .expect("Failed to create an output object_store"),
             );
 
-            let table = deltalake::open_table(trigger.location().as_str()).await?;
+            let table = deltalake::open_table(trigger.location().clone()).await?;
             info!(
-                "Loaded a table for {} at version {}",
+                "Loaded a table for {} at version {:?}",
                 trigger.location().as_str(),
                 table.version()
             );
@@ -182,15 +181,20 @@ struct Completion {
 mod tests {
     use super::*;
     use futures::StreamExt;
-    use object_store::{GetResultPayload, ObjectStore};
+    use object_store::ObjectStore;
 
     use deltalake::datafusion::{
         common::assert_batches_sorted_eq, dataframe::DataFrameWriteOptions,
     };
     use deltalake::operations::load_cdf::CdfLoadBuilder;
 
     async fn cdf_test_setup() -> DeltaResult<(SessionContext, CdfLoadBuilder)> {
-        let table = deltalake::open_table("../../tests/data/hive/checkpoint-cdf-table/").await?;
+        let canonical = std::fs::canonicalize("../../tests/data/hive/checkpoint-cdf-table")
+            .expect("Failed to canonicalize");
+        let table = deltalake::open_table(
+            Url::from_file_path(canonical).expect("Failed to find the Url for the CDF table"),
+        )
+        .await?;
         let cdf = DeltaOps::from(table).load_cdf().with_starting_version(3);
         let ctx = SessionContext::new();
         Ok((ctx, cdf))

lambdas/cdf-to-csv/src/main.rs

@@ -281,7 +281,7 @@ mod tests {
 
         table.load().await?;
 
-        assert_eq!(table.version(), version);
+        assert_eq!(table.version(), Some(version));
 
         let ctx = SessionContext::new();
         ctx.register_table("test", Arc::new(table))?;

lambdas/file-loader/src/main.rs

@@ -9,6 +9,7 @@ use deltalake::kernel::{DataType, PrimitiveType};
 use lambda_runtime::{Error, LambdaEvent, run, service_fn, tracing};
 use regex::Regex;
 use tracing::log::*;
+use url::Url;
 
 use oxbow_lambda_shared::*;
 use std::collections::HashMap;
@@ -84,8 +85,8 @@ async fn function_handler(event: LambdaEvent<SqsEvent>) -> Result<(), Error> {
         {
             if let Some(table_path) = key_is_delta_table(&key) {
                 debug!("loading table from {table_path}");
-                let uri = format!("s3://{bucket}/{table_path}");
-                let delta_table = deltalake::open_table(&uri).await?;
+                let uri = Url::parse(&format!("s3://{bucket}/{table_path}"))?;
+                let delta_table = deltalake::open_table(uri).await?;
 
                 if let Some(descriptor) = storage_descriptor_from(&delta_table, &glue_table) {
                     let parameters = match glue_table.parameters {
@@ -94,7 +95,10 @@ async fn function_handler(event: LambdaEvent<SqsEvent>) -> Result<(), Error> {
                             let mut updated = original.clone();
                             updated.insert(
                                 "delta.lastUpdateVersion".into(),
-                                delta_table.version().to_string(),
+                                delta_table
+                                    .version()
+                                    .expect("Failed to load a table version")
+                                    .to_string(),
                             );
                             Some(updated)
                         }
@@ -258,8 +262,9 @@ fn to_glue_type(data_type: &DataType) -> String {
 /// Generate the [Column] for Glue [aws_sdk_glue::types::StorageDescriptor] based on the provided
 /// [DeltaTable]
 fn glue_columns_for(table: &DeltaTable) -> Vec<Column> {
-    if let Ok(schema) = table.get_schema() {
-        return schema
+    if let Ok(snapshot) = table.snapshot() {
+        return snapshot
+            .schema()
             .fields()
             .map(|field| {
                 Column::builder()
@@ -315,8 +320,9 @@ mod tests {
 
     #[tokio::test]
     async fn test_glue_columns() {
-        let table = "../../tests/data/hive/deltatbl-non-partitioned";
-        let table = deltalake::open_table(&table)
+        let table =
+            std::fs::canonicalize("../../tests/data/hive/deltatbl-non-partitioned").unwrap();
+        let table = deltalake::open_table(Url::from_file_path(table).unwrap())
             .await
             .expect("Failed to open table");
 

lambdas/glue-sync/src/main.rs

@@ -153,6 +153,7 @@ mod tests {
         assert_eq!(4, event.records.len());
 
         let fifos = segmented_by_prefix(&event.records, Some(4)).expect("Failed to segment");
+        println!("fifos: {fifos:#?}");
         assert!(
             fifos.keys().len() > 2,
             "The segmented test file should have been distributed to more than just two keys"