Expose some configuration for cdf-to-csv to customize output CSVs

Some database engines need specific formattings to handle the conversion
of nulls or timestamps

Author: rtyler

Cargo.toml

@@ -7,7 +7,7 @@ members = [
 resolver = "3"
 
 [workspace.package]
-version = "1.8.5"
+version = "1.8.6"
 edition = "2024"
 keywords = ["deltalake", "parquet", "lambda", "delta", "sqs"]
 homepage = "https://github.com/buoyant-data/oxbow"

lambdas/cdf-to-csv/README.adoc

@@ -62,6 +62,18 @@ s3://csvs/
 |
 | Prefix for output CSV files with deleted rows (requires primary key set)
 
+| `CSV_NULL_CHARACTER`
+|
+| Character to use as the null character in CSV output
+
+| `CSV_TIMESTAMP_FORMAT`
+| `%Y-%m-%dT%H:%M:%S%.f`
+| Format string for `timestamp` fields
+
+| `CSV_TIMESTAMP_TZ_FORMAT`
+|
+| Timestamp format for timestamp with timezone arrays. When empty ISO 8601 format is used.
+
 | `RUST_LOG`
 | `error`
 | Set the log level, e.g. `info`, `warn`, `error`. Can be scoped to specific modules, i.e. `oxbow=debug`

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

@@ -3,7 +3,9 @@
 //! produce CSV files to ingest into Aurora for Change Data Feeds
 
 use aws_lambda_events::event::sqs::SqsEvent;
+use deltalake::datafusion::config::CsvOptions;
 use deltalake::datafusion::dataframe::DataFrameWriteOptions;
+use deltalake::datafusion::datasource::physical_plan::CsvOpener;
 use deltalake::datafusion::prelude::*;
 use deltalake::delta_datafusion::DeltaCdfTableProvider;
 use deltalake::{DeltaOps, DeltaResult};
@@ -105,10 +107,18 @@ async fn function_handler(event: LambdaEvent<SqsEvent>) -> DeltaResult<(), Error
             // 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)
+                .write_csv(
+                    "cdfo://inserts",
+                    DataFrameWriteOptions::default(),
+                    Some(csv_options()),
+                )
                 .await?;
             let deletes = deletes
-                .write_csv("cdfo://deletes", DataFrameWriteOptions::default(), None)
+                .write_csv(
+                    "cdfo://deletes",
+                    DataFrameWriteOptions::default(),
+                    Some(csv_options()),
+                )
                 .await?;
 
             let completion = Completion {
@@ -196,6 +206,16 @@ fn escape_dataframe(input: DataFrame) -> DeltaResult<DataFrame> {
     Ok(df)
 }
 
+/// Return the computed [CsvOptions] for writing output files
+fn csv_options() -> CsvOptions {
+    CsvOptions {
+        null_value: std::env::var("CSV_NULL_CHARACTER").ok(),
+        timestamp_format: std::env::var("CSV_TIMESTAMP_FORMAT").ok(),
+        timestamp_tz_format: std::env::var("CSV_TIMESTAMP_TZ_FORMAT").ok(),
+        ..Default::default()
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use std::fs::File;
@@ -301,7 +321,7 @@ mod tests {
         ctx.register_object_store(&Url::parse("cdfo://deletes").unwrap(), delete_store.clone());
 
         let mut stream = store.list(None);
-        while let Some(Ok(_entry)) = stream.next().await {
+        if let Some(Ok(_entry)) = stream.next().await {
             unreachable!("The InMemory object store should be empty before the test begins");
         }
 
@@ -310,10 +330,18 @@ mod tests {
         let change_data = retrieve_inserts(&ctx).await?;
         let deletes = retrieve_deletes(&ctx).await?;
         change_data
-            .write_csv("cdfo://inserts", DataFrameWriteOptions::default(), None)
+            .write_csv(
+                "cdfo://inserts",
+                DataFrameWriteOptions::default(),
+                Some(csv_options()),
+            )
             .await?;
         deletes
-            .write_csv("cdfo://deletes", DataFrameWriteOptions::default(), None)
+            .write_csv(
+                "cdfo://deletes",
+                DataFrameWriteOptions::default(),
+                Some(csv_options()),
+            )
             .await?;
 
         let mut stream = store.list(None);
@@ -356,7 +384,7 @@ mod tests {
         df.write_csv(
             tempfile,
             DataFrameWriteOptions::default(),
-            Some(CsvOptions::default()),
+            Some(csv_options()),
         )
         .await?;