Upgrade oxbow to deltalake 0.28

There are some upstream changes to incorporate delta_kernel which need
to be accounted for.

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

Author: rtyler

Cargo.toml

@@ -7,7 +7,7 @@ members = [
 resolver = "3"
 
 [workspace.package]
-version = "1.6.0"
+version = "1.6.1"
 edition = "2024"
 keywords = ["deltalake", "parquet", "lambda", "delta", "sqs"]
 homepage = "https://github.com/buoyant-data/oxbow"
@@ -20,10 +20,11 @@ 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.28.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" }
+mimalloc = { version = "0.1", features = ["v3"] }
 tokio = { version = "=1", features = ["macros"] }
 regex = "=1"
 serde = "1"
@@ -44,4 +45,3 @@ opt-level = "z"
 [profile.dist]
 inherits = "release"
 lto = "thin"
-

crates/oxbow/src/lib.rs

@@ -3,6 +3,8 @@ 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::kernel::engine::arrow_conversion::TryIntoArrow;
+use deltalake::table::config::TablePropertiesExt;
 use deltalake::writer::{DeltaWriter, record_batch::RecordBatchWriter};
 use deltalake::{DeltaResult, DeltaTable};
 
@@ -47,7 +49,7 @@ pub async fn append_values(
 ) -> DeltaResult<DeltaTable> {
     let schema = table.get_schema()?;
     debug!("Attempting to append values with schema: {schema:?}");
-    let schema = ArrowSchema::try_from(schema)?;
+    let schema: ArrowSchema = schema.try_into_arrow()?;
 
     let mut writer = RecordBatchWriter::for_table(&table)?;
     let mut written = false;
@@ -129,6 +131,7 @@ mod tests {
     use super::*;
     use deltalake::kernel::DataType;
     use deltalake::*;
+    use std::num::NonZero;
 
     async fn setup_test_table() -> DeltaResult<DeltaTable> {
         DeltaOps::try_from_uri("memory://")
@@ -157,12 +160,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: ArrowSchema = table.snapshot()?.schema().try_into_arrow()?;
 
         let json = deltalake::arrow::json::ReaderBuilder::new(schema.into()).build(reader)?;
 
         let table = append_batches(table, json).await?;
-        assert_eq!(table.version(), 1);
+        assert_eq!(table.version(), Some(1));
         Ok(())
     }
 
@@ -178,7 +181,7 @@ mod tests {
             .await
             .expect("Failed to do nothing");
 
-        assert_eq!(table.version(), 1);
+        assert_eq!(table.version(), Some(1));
         Ok(())
     }
 
@@ -191,8 +194,8 @@ mod tests {
         "#;
 
         let cursor = Cursor::new(jsonl);
-        let schema = table.get_schema()?;
-        let schema = ArrowSchema::try_from(schema)?;
+        let schema = table.snapshot()?.schema();
+        let schema: ArrowSchema = schema.try_into_arrow()?;
         let mut reader = ReaderBuilder::new(schema.into()).build(cursor).unwrap();
 
         while let Some(Ok(batch)) = reader.next() {
@@ -215,7 +218,7 @@ mod tests {
             .await
             .expect("Failed to do nothing");
 
-        assert_eq!(table.version(), 1);
+        assert_eq!(table.version(), Some(1));
         Ok(())
     }
 
@@ -235,7 +238,10 @@ 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!(
+                NonZero::new(100).unwrap(),
+                state.table_config().checkpoint_interval()
+            );
         }
 
         use deltalake::Path;
@@ -245,7 +251,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,8 +9,8 @@ homepage.workspace = true
 aws_lambda_events = { workspace = true, default-features = false, features = ["sqs"] }
 chrono = { workspace = true }
 lambda_runtime = "0.14.2"
-deltalake = { workspace = true, features = ["s3", "json", "datafusion"] }
 object_store = { version = "0.12.1", features = ["cloud"]}
+deltalake = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 tokio = { workspace = true }

lambdas/cdf-to-csv/Cargo.toml

@@ -6,12 +6,12 @@ use aws_lambda_events::event::sqs::SqsEvent;
 use deltalake::datafusion::dataframe::DataFrameWriteOptions;
 use deltalake::datafusion::prelude::*;
 use deltalake::delta_datafusion::DeltaCdfTableProvider;
+use deltalake::logstore::object_store::ObjectStore;
+use deltalake::logstore::object_store::aws::AmazonS3Builder;
+use deltalake::logstore::object_store::path::Path;
+use deltalake::logstore::object_store::prefix::PrefixStore;
 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::*;
 use serde::{Deserialize, Serialize};
 use std::sync::Arc;
@@ -62,15 +62,15 @@ async fn function_handler(event: LambdaEvent<SqsEvent>) -> DeltaResult<(), Error
             // the context is unique to a triggered table
             let ctx = SessionContext::new();
             let store: Arc<dyn ObjectStore> = Arc::new(
-                object_store::aws::AmazonS3Builder::from_env()
+                AmazonS3Builder::from_env()
                     .with_url(&destination)
                     .build()
                     .expect("Failed to create an output object_store"),
             );
 
             let table = deltalake::open_table(trigger.location().as_str()).await?;
             info!(
-                "Loaded a table for {} at version {}",
+                "Loaded a table for {} at version {:?}",
                 trigger.location().as_str(),
                 table.version()
             );
@@ -181,8 +181,9 @@ struct Completion {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use deltalake::logstore::object_store::memory::InMemory;
+    use deltalake::logstore::object_store::path::Path;
     use futures::StreamExt;
-    use object_store::{GetResultPayload, ObjectStore};
 
     use deltalake::datafusion::{
         common::assert_batches_sorted_eq, dataframe::DataFrameWriteOptions,
@@ -268,7 +269,7 @@ mod tests {
     #[tokio::test]
     async fn test_write_csv() -> DeltaResult<()> {
         let (ctx, cdf) = cdf_test_setup().await?;
-        let store: Arc<dyn ObjectStore> = Arc::new(object_store::memory::InMemory::new());
+        let store: Arc<dyn ObjectStore> = Arc::new(InMemory::new());
         let insert_store = Arc::new(PrefixStore::new(store.clone(), "inserts"));
         let delete_store = Arc::new(PrefixStore::new(store.clone(), "deletes"));
         ctx.register_object_store(&Url::parse("cdfo://inserts").unwrap(), insert_store.clone());

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

@@ -14,6 +14,7 @@ oxbow-sqs = { path = "../../crates/oxbow-sqs" }
 anyhow = { workspace = true }
 aws_lambda_events = { workspace = true, default-features = false, features = ["sqs"] }
 deltalake = { workspace = true }
+mimalloc = { workspace = true }
 tokio = { workspace = true }
 aws-config = "1.5.10"
 aws-sdk-s3 = "1.60.0"

lambdas/file-loader/Cargo.toml

@@ -8,13 +8,19 @@ use aws_lambda_events::sqs::SqsMessage;
 use deltalake::DeltaResult;
 use deltalake::arrow::datatypes::Schema as ArrowSchema;
 use deltalake::arrow::json::reader::ReaderBuilder;
+use deltalake::kernel::engine::arrow_conversion::TryIntoArrow;
 use deltalake::writer::{DeltaWriter, record_batch::RecordBatchWriter};
 use lambda_runtime::tracing::{debug, error, info, trace};
 use lambda_runtime::{Error, LambdaEvent, run, service_fn, tracing};
 
 use oxbow_lambda_shared::*;
 use oxbow_sqs::{ConsumerConfig, TimedConsumer};
 
+use mimalloc::MiMalloc;
+
+#[global_allocator]
+static GLOBAL: MiMalloc = MiMalloc;
+
 use std::env;
 use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 
@@ -124,7 +130,7 @@ async fn function_handler(event: LambdaEvent<SqsEvent>) -> Result<(), Error> {
                         .send()
                         .await?;
                     info!("Attempting to read bytes from {file_record:?}");
-                    let schema = ArrowSchema::try_from(table.get_schema()?)?;
+                    let schema: ArrowSchema = table.snapshot()?.schema().try_into_arrow()?;
 
                     let mut json = ReaderBuilder::new(schema.into()).build_decoder()?;
                     while let Some(bytes) = response.body.try_next().await? {

lambdas/file-loader/src/main.rs

@@ -94,7 +94,7 @@ 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().unwrap().to_string(),
                             );
                             Some(updated)
                         }