Produce a small completion marker file when cdf-to-csv completes

rtyler · rtyler · commit 8b7d289bda1c · 2025-09-10T16:10:05.000Z
For larger data sets the write_csv() call in Datafusion can produce
multiple CSV files. In order to allow downstream event-driven processes
to know when the entirety of the "action" has completed this change now
writes `cdf-completion.json` which contains some rudimentary statistics
within.
diff --git a/Cargo.toml b/Cargo.toml
@@ -7,8 +7,8 @@ members = [
 resolver = "3"
 
 [workspace.package]
-version = "1.5.2"
-edition = "2021"
+version = "1.6.0"
+edition = "2024"
 keywords = ["deltalake", "parquet", "lambda", "delta", "sqs"]
 homepage = "https://github.com/buoyant-data/oxbow"
 repository = "https://github.com/buoyant-data/oxbow"
diff --git a/crates/oxbow-lambda-shared/src/trigger.rs b/crates/oxbow-lambda-shared/src/trigger.rs
@@ -234,7 +234,9 @@ mod tests {
             assert_eq!(ChangeType::TransactionLog { version }, change);
         }
 
-        let (change, _) = ChangeType::from_key("/mytable/_change_data/cdc-00000-924d9ac7-21a9-4121-b067-a0a6517aa8ed.c000.snappy.parquet");
+        let (change, _) = ChangeType::from_key(
+            "/mytable/_change_data/cdc-00000-924d9ac7-21a9-4121-b067-a0a6517aa8ed.c000.snappy.parquet",
+        );
         assert_eq!(ChangeType::ChangeDataFeed, change);
 
         let (change, _) = ChangeType::from_key(
diff --git a/crates/oxbow-sqs/src/lib.rs b/crates/oxbow-sqs/src/lib.rs
@@ -192,7 +192,9 @@ impl TimedConsumer {
 impl std::ops::Drop for TimedConsumer {
     fn drop(&mut self) {
         if !self.receive_handles.is_empty() {
-            error!("The TimedConsumer was not flushed before being dropped! This causes data duplication, you have to flush!");
+            error!(
+                "The TimedConsumer was not flushed before being dropped! This causes data duplication, you have to flush!"
+            );
         }
     }
 }
diff --git a/crates/oxbow/Cargo.toml b/crates/oxbow/Cargo.toml
@@ -7,7 +7,7 @@ homepage.workspace = true
 
 [dependencies]
 chrono = { workspace = true }
-deltalake = { workspace = true, default-features = false }
+deltalake = { workspace = true }
 tracing = { workspace = true }
 url = { workspace = true }
 
diff --git a/crates/oxbow/src/lib.rs b/crates/oxbow/src/lib.rs
@@ -1,18 +1,18 @@
+use deltalake::ObjectStore;
 ///
 /// The lib module contains the business logic of oxbow, regardless of the interface implementation
 ///
 use deltalake::arrow::datatypes::Schema as ArrowSchema;
 use deltalake::kernel::models::{Schema, StructField};
 use deltalake::kernel::*;
 use deltalake::logstore::ObjectStoreRef;
-use deltalake::logstore::{logstore_for, LogStoreRef};
+use deltalake::logstore::{LogStoreRef, logstore_for};
 use deltalake::operations::create::CreateBuilder;
 use deltalake::parquet::arrow::async_reader::{
     ParquetObjectReader, ParquetRecordBatchStreamBuilder,
 };
 use deltalake::parquet::file::metadata::ParquetMetaData;
 use deltalake::protocol::*;
-use deltalake::ObjectStore;
 use deltalake::{DeltaResult, DeltaTable, DeltaTableError, ObjectMeta};
 use futures::StreamExt;
 use tracing::log::*;
@@ -138,7 +138,9 @@ pub async fn discover_parquet_files(
                                 debug!("Discovered file: {:?}", meta);
                                 result.push(meta);
                             } else {
-                                warn!("Was asked to discover parquet files on what appears to already be a table, and found checkpoint files: {filename}");
+                                warn!(
+                                    "Was asked to discover parquet files on what appears to already be a table, and found checkpoint files: {filename}"
+                                );
                             }
                         }
                     }
@@ -444,34 +446,37 @@ fn coerce_field(
 ) -> deltalake::arrow::datatypes::FieldRef {
     use deltalake::arrow::datatypes::*;
     match field.data_type() {
-        DataType::Timestamp(unit, tz) => {
-            match unit {
-                TimeUnit::Nanosecond => {
-                    warn!("Given a nanosecond precision which we will cowardly pretend is microseconds");
-                    let field = Field::new(
-                        field.name(),
-                        DataType::Timestamp(TimeUnit::Microsecond, tz.clone()),
-                        field.is_nullable(),
-                    );
-                    return Arc::new(field);
-                }
-                TimeUnit::Millisecond => {
-                    warn!("I have been asked to create a table with a Timestamp(millis) column ({}) that I cannot handle. Cowardly setting the Delta schema to pretend it is a Timestamp(micros)", field.name());
-                    let field = Field::new(
-                        field.name(),
-                        DataType::Timestamp(TimeUnit::Microsecond, tz.clone()),
-                        field.is_nullable(),
-                    );
-                    return Arc::new(field);
-                }
-                _ => {}
+        DataType::Timestamp(unit, tz) => match unit {
+            TimeUnit::Nanosecond => {
+                warn!(
+                    "Given a nanosecond precision which we will cowardly pretend is microseconds"
+                );
+                let field = Field::new(
+                    field.name(),
+                    DataType::Timestamp(TimeUnit::Microsecond, tz.clone()),
+                    field.is_nullable(),
+                );
+                return Arc::new(field);
             }
-        }
+            TimeUnit::Millisecond => {
+                warn!(
+                    "I have been asked to create a table with a Timestamp(millis) column ({}) that I cannot handle. Cowardly setting the Delta schema to pretend it is a Timestamp(micros)",
+                    field.name()
+                );
+                let field = Field::new(
+                    field.name(),
+                    DataType::Timestamp(TimeUnit::Microsecond, tz.clone()),
+                    field.is_nullable(),
+                );
+                return Arc::new(field);
+            }
+            _ => {}
+        },
         DataType::List(field) => {
             let coerced = coerce_field(field.clone());
             let list_field = Field::new(field.name(), DataType::List(coerced), field.is_nullable());
             return Arc::new(list_field);
-        }
+        },
         DataType::Struct(fields) => {
             let coerced: Vec<deltalake::arrow::datatypes::FieldRef> =
                 fields.iter().map(|f| coerce_field(f.clone())).collect();
@@ -481,7 +486,7 @@ fn coerce_field(
                 field.is_nullable(),
             );
             return Arc::new(struct_field);
-        }
+        },
         _ => {}
     };
     field.clone()
diff --git a/crates/oxbow/src/write.rs b/crates/oxbow/src/write.rs
@@ -3,7 +3,7 @@ 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::{record_batch::RecordBatchWriter, DeltaWriter};
+use deltalake::writer::{DeltaWriter, record_batch::RecordBatchWriter};
 use deltalake::{DeltaResult, DeltaTable};
 
 use std::io::Cursor;
diff --git a/lambdas/auto-tag/src/main.rs b/lambdas/auto-tag/src/main.rs
@@ -1,7 +1,7 @@
 use aws_lambda_events::event::s3::S3EventRecord;
 use aws_lambda_events::event::sqs::SqsEvent;
 use aws_sdk_s3::types::{Tag, Tagging};
-use lambda_runtime::{run, service_fn, Error, LambdaEvent};
+use lambda_runtime::{Error, LambdaEvent, run, service_fn};
 use tracing::log::*;
 
 use std::collections::HashMap;
diff --git a/lambdas/cdf-to-csv/Cargo.toml b/lambdas/cdf-to-csv/Cargo.toml
@@ -11,6 +11,7 @@ chrono = { workspace = true }
 lambda_runtime = "0.14.2"
 deltalake = { workspace = true, features = ["s3", "json", "datafusion"] }
 object_store = { version = "0.12.1", features = ["cloud"]}
+serde = { workspace = true }
 serde_json = { workspace = true }
 tokio = { workspace = true }
 tracing = { workspace = true }
diff --git a/lambdas/cdf-to-csv/src/main.rs b/lambdas/cdf-to-csv/src/main.rs
@@ -7,10 +7,13 @@ use deltalake::datafusion::dataframe::DataFrameWriteOptions;
 use deltalake::datafusion::prelude::*;
 use deltalake::delta_datafusion::DeltaCdfTableProvider;
 use deltalake::{DeltaOps, DeltaResult};
-use lambda_runtime::{run, service_fn, tracing, Error, LambdaEvent};
-use object_store::prefix::PrefixStore;
+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;
 use tracing::log::*;
 use url::Url;
@@ -98,13 +101,22 @@ async fn function_handler(event: LambdaEvent<SqsEvent>) -> DeltaResult<(), Error
 
             let inserts = retrieve_inserts(&ctx).await?;
             let deletes = retrieve_deletes(&ctx).await?;
-            inserts
+
+            // write_csv will return a Vec,RecordBatch> which we can use for some rudimentary
+            // statistics
+            let inserts = inserts
                 .write_csv("cdfo://inserts", DataFrameWriteOptions::default(), None)
                 .await?;
-
-            deletes
+            let deletes = deletes
                 .write_csv("cdfo://deletes", DataFrameWriteOptions::default(), None)
                 .await?;
+
+            let completion = Completion {
+                inserts: inserts.iter().map(|rb| rb.num_rows()).sum(),
+                deletes: deletes.iter().map(|rb| rb.num_rows()).sum(),
+            };
+
+            mark_complete(store.clone(), &completion).await?;
         } else {
             warn!("Invoked but didn't find min/max trigger versions, something is fishy!");
         }
@@ -131,6 +143,7 @@ async fn retrieve_inserts(ctx: &SessionContext) -> DeltaResult<DataFrame> {
     ])?)
 }
 
+/// Compute the deletes from the change data feed associated with the [SessionContext]
 async fn retrieve_deletes(ctx: &SessionContext) -> DeltaResult<DataFrame> {
     let df = ctx
         .sql("SELECT * FROM cdf WHERE _change_type IN ('delete')")
@@ -143,12 +156,33 @@ async fn retrieve_deletes(ctx: &SessionContext) -> DeltaResult<DataFrame> {
     ])?)
 }
 
+/// Write a completion file to the given object store.
+///
+/// This is expected to be the prefix store associated with a werite
+async fn mark_complete(store: Arc<dyn ObjectStore>, completion: &Completion) -> DeltaResult<()> {
+    // Write a sentinel file once the writes have completed successfully
+    store
+        .put(
+            &Path::from("cdf-completion.json"),
+            serde_json::to_string(completion)
+                .expect("Failed to serialize Completion")
+                .into(),
+        )
+        .await?;
+    Ok(())
+}
+
+#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
+struct Completion {
+    inserts: usize,
+    deletes: usize,
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
     use futures::StreamExt;
-    use object_store::path::Path;
-    use object_store::ObjectStore;
+    use object_store::{GetResultPayload, ObjectStore};
 
     use deltalake::datafusion::{
         common::assert_batches_sorted_eq, dataframe::DataFrameWriteOptions,
@@ -161,6 +195,27 @@ mod tests {
         let ctx = SessionContext::new();
         Ok((ctx, cdf))
     }
+
+    #[tokio::test]
+    async fn test_mark_complete() -> DeltaResult<()> {
+        let store: Arc<dyn ObjectStore> = Arc::new(object_store::memory::InMemory::new());
+        let completion = Completion {
+            inserts: 1,
+            deletes: 0,
+        };
+        mark_complete(store.clone(), &completion).await?;
+        let _ = store.head(&Path::from("cdf-completion.json")).await?;
+
+        let result = store.get(&Path::from("cdf-completion.json")).await?;
+        let bytes = result.bytes().await?;
+        let s = String::from_utf8(bytes.to_vec()).expect("Failed to convert buffer");
+        let received: Completion = serde_json::from_str(&s)?;
+
+        assert_eq!(completion, received);
+
+        Ok(())
+    }
+
     #[tokio::test]
     async fn test_read_cdf_deletes() -> DeltaResult<()> {
         let (ctx, cdf) = cdf_test_setup().await?;
diff --git a/lambdas/file-loader/src/main.rs b/lambdas/file-loader/src/main.rs
@@ -5,12 +5,12 @@
 use aws_lambda_events::event::sqs::SqsEvent;
 use aws_lambda_events::s3::S3EventRecord;
 use aws_lambda_events::sqs::SqsMessage;
+use deltalake::DeltaResult;
 use deltalake::arrow::datatypes::Schema as ArrowSchema;
 use deltalake::arrow::json::reader::ReaderBuilder;
-use deltalake::writer::{record_batch::RecordBatchWriter, DeltaWriter};
-use deltalake::DeltaResult;
+use deltalake::writer::{DeltaWriter, record_batch::RecordBatchWriter};
 use lambda_runtime::tracing::{debug, error, info, trace};
-use lambda_runtime::{run, service_fn, tracing, Error, LambdaEvent};
+use lambda_runtime::{Error, LambdaEvent, run, service_fn, tracing};
 
 use oxbow_lambda_shared::*;
 use oxbow_sqs::{ConsumerConfig, TimedConsumer};
@@ -138,7 +138,9 @@ async fn function_handler(event: LambdaEvent<SqsEvent>) -> Result<(), Error> {
                     }
                 }
                 RecordType::Unknown => {
-                    error!("file-loader was invoked for a file with an unknown suffix! Ignoring: {file_record:?}");
+                    error!(
+                        "file-loader was invoked for a file with an unknown suffix! Ignoring: {file_record:?}"
+                    );
                 }
             }
         }
@@ -147,7 +149,9 @@ async fn function_handler(event: LambdaEvent<SqsEvent>) -> Result<(), Error> {
             let mbytes_to_consume: usize = str::parse(&bytes_to_consume)
                 .expect("BUFFER_MORE_BYTES_ALLOWED must be parseable as a uint64");
 
-            info!("Allocated {bytes_consumed} bytes thus far... I can only have {mbytes_to_consume}MB");
+            info!(
+                "Allocated {bytes_consumed} bytes thus far... I can only have {mbytes_to_consume}MB"
+            );
             if bytes_consumed >= (mbytes_to_consume * 1024 * 1024) {
                 info!("Finalizing after consuming {bytes_consumed} bytes of memory");
                 break;
diff --git a/lambdas/glue-create/src/main.rs b/lambdas/glue-create/src/main.rs
@@ -1,6 +1,6 @@
 use aws_lambda_events::s3::S3EventRecord;
 use aws_lambda_events::sqs::SqsEvent;
-use lambda_runtime::{run, service_fn, Error, LambdaEvent};
+use lambda_runtime::{Error, LambdaEvent, run, service_fn};
 use regex::Regex;
 use tracing::log::*;
 
diff --git a/lambdas/glue-sync/src/main.rs b/lambdas/glue-sync/src/main.rs
@@ -4,9 +4,9 @@
 ///
 use aws_lambda_events::event::sqs::SqsEvent;
 use aws_sdk_glue::types::{Column, StorageDescriptor, Table, TableInput};
-use deltalake::kernel::{DataType, PrimitiveType};
 use deltalake::DeltaTable;
-use lambda_runtime::{run, service_fn, tracing, Error, LambdaEvent};
+use deltalake::kernel::{DataType, PrimitiveType};
+use lambda_runtime::{Error, LambdaEvent, run, service_fn, tracing};
 use regex::Regex;
 use tracing::log::*;
 
diff --git a/lambdas/group-events/src/main.rs b/lambdas/group-events/src/main.rs
@@ -1,7 +1,7 @@
 use aws_lambda_events::event::sqs::SqsEvent;
 use aws_lambda_events::s3::{S3Event, S3EventRecord};
 use aws_sdk_sqs::types::SendMessageBatchRequestEntry;
-use lambda_runtime::{run, service_fn, Error, LambdaEvent};
+use lambda_runtime::{Error, LambdaEvent, run, service_fn};
 use tracing::log::*;
 use uuid::Uuid;
 
diff --git a/lambdas/oxbow/src/main.rs b/lambdas/oxbow/src/main.rs
@@ -5,7 +5,7 @@
 /// function glues that into the Lambda runtime
 use aws_lambda_events::sqs::SqsEvent;
 use deltalake::DeltaTableError;
-use lambda_runtime::{service_fn, Error, LambdaEvent};
+use lambda_runtime::{Error, LambdaEvent, service_fn};
 use serde_json::Value;
 use tracing::log::*;
 use url::Url;
diff --git a/lambdas/sqs-ingest/src/main.rs b/lambdas/sqs-ingest/src/main.rs
@@ -3,7 +3,7 @@
 /// and appending them to the configured Delta table
 ///
 use aws_lambda_events::event::sqs::{SqsEvent, SqsMessage};
-use lambda_runtime::{run, service_fn, tracing, Error, LambdaEvent};
+use lambda_runtime::{Error, LambdaEvent, run, service_fn, tracing};
 use serde::Deserialize;
 use tracing::log::*;
 
@@ -38,7 +38,9 @@ async fn function_handler(event: LambdaEvent<SqsEvent>) -> Result<(), Error> {
                 "The value of BUFFER_MORE_MESSAGES cannot be coerced into an int :thinking:",
             ));
         debug!("sqs-ingest configured to consume an additional {how_many_more} messages from SQS");
-        debug!("sqs-ingest will attempt to retrieve {how_many_more} messages in no more than {more_deadline_ms}ms to avoid timing out the function");
+        debug!(
+            "sqs-ingest will attempt to retrieve {how_many_more} messages in no more than {more_deadline_ms}ms to avoid timing out the function"
+        );
     }
 
     let mut consumer = TimedConsumer::new(
diff --git a/lambdas/webhook/src/main.rs b/lambdas/webhook/src/main.rs
@@ -1,7 +1,7 @@
 ///
 /// The webhook lambda can receive JSONL formatted events and append them to a pre-configured Delta
 /// table
-use lambda_http::{run, service_fn, tracing, Body, Error, Request, Response};
+use lambda_http::{Body, Error, Request, Response, run, service_fn, tracing};
 use tracing::log::*;
 
 use oxbow::write::*;
@@ -68,7 +68,9 @@ async fn main() -> Result<(), Error> {
     match std::env::var("DYNAMO_LOCK_TABLE_NAME") {
         Ok(_) => {}
         Err(_) => {
-            warn!("sqs-ingest SHOULD have `DYNAMO_LOCK_TABLE_NAME` set to a valid name, and should have AWS_S3_LOCKING_PROVIDER=dynamodb set so that concurrent writes can be performed safely.");
+            warn!(
+                "sqs-ingest SHOULD have `DYNAMO_LOCK_TABLE_NAME` set to a valid name, and should have AWS_S3_LOCKING_PROVIDER=dynamodb set so that concurrent writes can be performed safely."
+            );
         }
     }
 

Original file line number	Diff line number	Diff line change
`@@ -234,7 +234,9 @@ mod tests {`
`234`	`234`	`assert_eq!(ChangeType::TransactionLog { version }, change);`
`235`	`235`	`}`
`236`	`236`
`237`		`- let (change, _) = ChangeType::from_key("/mytable/_change_data/cdc-00000-924d9ac7-21a9-4121-b067-a0a6517aa8ed.c000.snappy.parquet");`
	`237`	`+ let (change, _) = ChangeType::from_key(`
	`238`	`+ "/mytable/_change_data/cdc-00000-924d9ac7-21a9-4121-b067-a0a6517aa8ed.c000.snappy.parquet",`
	`239`	`+ );`
`238`	`240`	`assert_eq!(ChangeType::ChangeDataFeed, change);`
`239`	`241`
`240`	`242`	`let (change, _) = ChangeType::from_key(`
Original file line number	Diff line number	Diff line change
`@@ -192,7 +192,9 @@ impl TimedConsumer {`
`192`	`192`	`impl std::ops::Drop for TimedConsumer {`
`193`	`193`	`fn drop(&mut self) {`
`194`	`194`	`if !self.receive_handles.is_empty() {`
`195`		`- error!("The TimedConsumer was not flushed before being dropped! This causes data duplication, you have to flush!");`
	`195`	`+ error!(`
	`196`	`+ "The TimedConsumer was not flushed before being dropped! This causes data duplication, you have to flush!"`
	`197`	`+ );`
`196`	`198`	`}`
`197`	`199`	`}`
`198`	`200`	`}`