Add delta table support

.github/workflows/ci.yml

@@ -169,4 +169,3 @@ jobs:
           fail-on-error: false
           github-token: ${{ secrets.GITHUB_TOKEN }}
           path-to-lcov: lcov.info
-          debug: true

.gitignore

@@ -33,3 +33,5 @@ pyvenv.cfg
 
 # Log files
 *.log
+
+lcov.info

Cargo.toml

@@ -32,7 +32,7 @@ actix-web = { version = "4.11.0", default-features = false }
 actix-web-httpauth = { version = "0.8.2", default-features = false }
 actix-web-metrics = { version = "0.3.0", default-features = false }
 anyhow = { version = "1.0.98", default-features = false }
-arrow = { version = "55.0", default-features = false }
+arrow = { version = "56.2.0", default-features = false }
 async-trait = { version = "0.1.88" }
 aws-lc-rs = { version = "1.13.3", default-features = false }
 base64 = { version = "0.22.1", default-features = false }
@@ -43,6 +43,8 @@ clap = { version = "4.5.42", default-features = false }
 config = { version = "0.14", default-features = false }
 const-oid = { version = "0.9.6", default-features = false }
 constant_time_eq = { version = "0.4.2" }
+dashmap = { version = "6.1.0", default-features = false }
+deltalake = { git = "https://github.com/delta-io/delta-rs.git", rev = "d30b11f673b0111dbb0f904bf89d5b917ea652ed", default-features = false }
 fail = { version = "0.5.1", default-features = false }
 futures = { version = "0.3.31", default-features = false }
 gcp-bigquery-client = { version = "0.27.0", default-features = false }
@@ -53,7 +55,7 @@ k8s-openapi = { version = "0.25.0", default-features = false }
 kube = { version = "1.1.0", default-features = false }
 metrics = { version = "0.24.2", default-features = false }
 metrics-exporter-prometheus = { version = "0.17.2", default-features = false }
-parquet = { version = "55.0", default-features = false }
+parquet = { version = "55.0.0", default-features = false }
 pg_escape = { version = "0.1.1", default-features = false }
 pin-project-lite = { version = "0.2.16", default-features = false }
 postgres-replication = { git = "https://github.com/MaterializeInc/rust-postgres", default-features = false, rev = "c4b473b478b3adfbf8667d2fbe895d8423f1290b" }
@@ -84,3 +86,7 @@ x509-cert = { version = "0.2.2", default-features = false }
 
 [profile.bench]
 debug = true
+
+[profile.dev.package]
+insta.opt-level = 3
+similar.opt-level = 3

etl-api/src/configs/destination.rs

@@ -1,3 +1,5 @@
+use std::collections::HashMap;
+
 use etl_config::SerializableSecretString;
 use etl_config::shared::DestinationConfig;
 use secrecy::ExposeSecret;
@@ -14,6 +16,7 @@ const DEFAULT_MAX_CONCURRENT_STREAMS: usize = 8;
 
 #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
 #[serde(rename_all = "snake_case")]
+#[non_exhaustive]
 pub enum FullApiDestinationConfig {
     Memory,
     BigQuery {
@@ -30,6 +33,16 @@ pub enum FullApiDestinationConfig {
         #[serde(skip_serializing_if = "Option::is_none")]
         max_concurrent_streams: Option<usize>,
     },
+    DeltaLake {
+        #[schema(example = "s3://my-bucket/my-path")]
+        base_uri: String,
+        #[schema(example = "{\"aws_access_key_id\": \"https://my-endpoint.com\"}")]
+        storage_options: Option<HashMap<String, String>>,
+        #[schema(example = "{\"my_table\": [\"date\"]}")]
+        partition_columns: Option<HashMap<String, Vec<String>>>,
+        #[schema(example = 100)]
+        optimize_after_commits: Option<u64>,
+    },
 }
 
 impl From<StoredDestinationConfig> for FullApiDestinationConfig {
@@ -49,6 +62,17 @@ impl From<StoredDestinationConfig> for FullApiDestinationConfig {
                 max_staleness_mins,
                 max_concurrent_streams: Some(max_concurrent_streams),
             },
+            StoredDestinationConfig::DeltaLake {
+                base_uri,
+                storage_options,
+                partition_columns,
+                optimize_after_commits,
+            } => Self::DeltaLake {
+                base_uri,
+                storage_options,
+                partition_columns,
+                optimize_after_commits,
+            },
         }
     }
 }
@@ -64,6 +88,12 @@ pub enum StoredDestinationConfig {
         max_staleness_mins: Option<u16>,
         max_concurrent_streams: usize,
     },
+    DeltaLake {
+        base_uri: String,
+        storage_options: Option<HashMap<String, String>>,
+        partition_columns: Option<HashMap<String, Vec<String>>>,
+        optimize_after_commits: Option<u64>,
+    },
 }
 
 impl StoredDestinationConfig {
@@ -83,6 +113,17 @@ impl StoredDestinationConfig {
                 max_staleness_mins,
                 max_concurrent_streams,
             },
+            Self::DeltaLake {
+                base_uri,
+                storage_options,
+                partition_columns,
+                optimize_after_commits,
+            } => DestinationConfig::DeltaLake {
+                base_uri,
+                storage_options,
+                partition_columns,
+                optimize_after_commits,
+            },
         }
     }
 }
@@ -105,6 +146,17 @@ impl From<FullApiDestinationConfig> for StoredDestinationConfig {
                 max_concurrent_streams: max_concurrent_streams
                     .unwrap_or(DEFAULT_MAX_CONCURRENT_STREAMS),
             },
+            FullApiDestinationConfig::DeltaLake {
+                base_uri,
+                storage_options,
+                partition_columns,
+                optimize_after_commits,
+            } => Self::DeltaLake {
+                base_uri,
+                storage_options,
+                partition_columns,
+                optimize_after_commits,
+            },
         }
     }
 }
@@ -136,12 +188,24 @@ impl Encrypt<EncryptedStoredDestinationConfig> for StoredDestinationConfig {
                     max_concurrent_streams,
                 })
             }
+            Self::DeltaLake {
+                base_uri,
+                storage_options,
+                partition_columns,
+                optimize_after_commits,
+            } => Ok(EncryptedStoredDestinationConfig::DeltaLake {
+                base_uri,
+                storage_options,
+                partition_columns,
+                optimize_after_commits,
+            }),
         }
     }
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case")]
+#[non_exhaustive]
 pub enum EncryptedStoredDestinationConfig {
     Memory,
     BigQuery {
@@ -151,6 +215,12 @@ pub enum EncryptedStoredDestinationConfig {
         max_staleness_mins: Option<u16>,
         max_concurrent_streams: usize,
     },
+    DeltaLake {
+        base_uri: String,
+        storage_options: Option<HashMap<String, String>>,
+        partition_columns: Option<HashMap<String, Vec<String>>>,
+        optimize_after_commits: Option<u64>,
+    },
 }
 
 impl Store for EncryptedStoredDestinationConfig {}
@@ -182,6 +252,17 @@ impl Decrypt<StoredDestinationConfig> for EncryptedStoredDestinationConfig {
                     max_concurrent_streams,
                 })
             }
+            Self::DeltaLake {
+                base_uri,
+                storage_options,
+                partition_columns,
+                optimize_after_commits,
+            } => Ok(StoredDestinationConfig::DeltaLake {
+                base_uri,
+                storage_options,
+                partition_columns,
+                optimize_after_commits,
+            }),
         }
     }
 }

etl-benchmarks/Cargo.toml

@@ -7,10 +7,15 @@ rust-version.workspace = true
 repository.workspace = true
 homepage.workspace = true
 
+[features]
+default = ["bigquery", "deltalake"]
+bigquery = ["etl-destinations/bigquery"]
+deltalake = ["etl-destinations/deltalake"]
+
 [dev-dependencies]
 etl = { workspace = true, features = ["test-utils"] }
 etl-config = { workspace = true }
-etl-destinations = { workspace = true, features = ["bigquery"] }
+etl-destinations = { workspace = true }
 etl-postgres = { workspace = true, features = ["sqlx"] }
 etl-telemetry = { workspace = true }
 

etl-benchmarks/Dockerfile.tpch-seeder

@@ -0,0 +1,14 @@
+# Builds an image that bundles go-tpc along with psql client tools for loading TPC-H data.
+FROM golang:1.22 AS builder
+
+ARG GO_TPC_VERSION=latest
+
+RUN apt-get update \
+    && apt-get install -y --no-install-recommends postgresql-client \
+    && rm -rf /var/lib/apt/lists/*
+
+RUN go install github.com/pingcap/go-tpc/cmd/go-tpc@${GO_TPC_VERSION}
+
+ENV PATH="/go/bin:${PATH}"
+
+ENTRYPOINT ["/bin/sh", "-c"]

etl-benchmarks/README.md

@@ -13,6 +13,7 @@ Before running benchmarks, ensure you have:
 - A Postgres database set up
 - A publication created with the tables you want to benchmark
 - For BigQuery benchmarks: GCP project, dataset, and service account key file
+- For Delta Lake benchmarks: Accessible storage URI (e.g., `s3://bucket/path`) and any required object store credentials
 
 ## Quick Start
 
@@ -55,6 +56,23 @@ cargo bench --bench table_copies --features bigquery -- --log-target terminal ru
   --bq-sa-key-file /path/to/service-account-key.json
 ```
 
+### 4. Run Delta Lake Benchmark
+
+Benchmark against a Delta Lake table store:
+
+```bash
+cargo bench --bench table_copies -- --log-target terminal run \
+  --host localhost --port 5432 --database bench \
+  --username postgres --password mypass \
+  --publication-name bench_pub \
+  --table-ids 1,2,3 \
+  --destination delta-lake \
+  --delta-base-uri s3://my-bucket/my-warehouse \
+  --delta-storage-option endpoint=http://localhost:9010 \
+  --delta-storage-option access_key_id=minio \
+  --delta-storage-option secret_access_key=minio-secret
+```
+
 ## Command Reference
 
 ### Common Parameters
@@ -68,7 +86,7 @@ cargo bench --bench table_copies --features bigquery -- --log-target terminal ru
 | `--password`         | Postgres password                        | (optional)  |
 | `--publication-name` | Publication to replicate from            | `bench_pub` |
 | `--table-ids`        | Comma-separated table IDs to replicate   | (required)  |
-| `--destination`      | Destination type (`null` or `big-query`) | `null`      |
+| `--destination`      | Destination type (`null`, `big-query`, or `delta-lake`) | `null`      |
 
 ### Performance Tuning Parameters
 
@@ -87,6 +105,13 @@ cargo bench --bench table_copies --features bigquery -- --log-target terminal ru
 | `--bq-sa-key-file`        | Service account key file path | Yes                   |
 | `--bq-max-staleness-mins` | Max staleness in minutes      | No                    |
 
+### Delta Lake Parameters
+
+| Parameter                 | Description                                      | Required for Delta Lake |
+| ------------------------- | ------------------------------------------------ | ----------------------- |
+| `--delta-base-uri`        | Base URI for Delta tables (e.g., `s3://bucket`)  | Yes                     |
+| `--delta-storage-option`  | Extra storage option in `key=value` form. Repeat per option. | No |
+
 ### Logging Options
 
 | Parameter               | Description                         |
@@ -130,3 +155,22 @@ cargo bench --bench table_copies --features bigquery -- --log-target terminal ru
 ```
 
 The benchmark will measure the time it takes to complete the initial table copy phase for all specified tables.
+
+## Local Docker Environment
+
+Start a ready-to-benchmark Postgres instance seeded with TPC-H data via Docker Compose:
+
+```bash
+cd etl-benchmarks
+docker compose up postgres tpch-seeder
+```
+
+The `tpch-seeder` service builds a lightweight image (see `Dockerfile.tpch-seeder`) that bundles the [`go-tpc`](https://github.com/pingcap/go-tpc) binary and runs the TPC-H loader after Postgres becomes healthy. Adjust credentials, port mapping, scale factor, or the go-tpc version by exporting `POSTGRES_USER`, `POSTGRES_PASSWORD`, `POSTGRES_DB`, `POSTGRES_PORT`, `TPCH_SCALE_FACTOR`, or `GO_TPC_VERSION` before launching Compose. Pass `--build` (or `--pull`) when changing `GO_TPC_VERSION` so Compose rebuilds the seeder image.
+
+To add an S3-compatible target for Delta Lake benchmarking, enable the optional `minio` profile:
+
+```bash
+docker compose --profile minio up postgres tpch-seeder minio minio-setup
+```
+
+This exposes MinIO on `http://localhost:9010` (console on `http://localhost:9011`) with credentials `minio-admin` / `minio-admin-password` and creates the bucket defined by `MINIO_BUCKET` (default `delta-dev-and-test`).