Add delta table support

.config/nextest.toml

@@ -0,0 +1,6 @@
+[profile.default]
+retries = 2
+slow-timeout = { period = "15s", terminate-after = 4 }
+
+[profile.no-bigquery]
+default-filter = "not (package(etl-destinations) and binary(/bigquery_pipeline/))"

.github/workflows/ci.yml

@@ -90,19 +90,14 @@ jobs:
           sudo apt-get install libpq-dev -y
           ./etl-api/scripts/run_migrations.sh
 
+      - name: Install cargo-nextest
+        uses: taiki-e/install-action@v2
+        with:
+          tool: cargo-nextest
+
       - name: Run Tests
         run: |
-          cargo test \
-            --workspace \
-            --all-features \
-            --no-fail-fast \
-            --exclude etl-destinations \
-          && \
-          cargo test \
-            -p etl-destinations \
-            --no-default-features \
-            --no-fail-fast \
-            --features iceberg
+          cargo nextest run --all-features --no-fail-fast --profile no-bigquery
 
   test-full:
     name: Tests (Full)
@@ -121,6 +116,8 @@ jobs:
 
       - name: Set up Rust
         uses: dtolnay/rust-toolchain@1.88.0
+        with:
+          components: llvm-tools-preview
 
       - name: Cache Cargo
         uses: Swatinem/rust-cache@v2
@@ -149,8 +146,10 @@ jobs:
           sudo apt-get install libpq-dev -y
           ./etl-api/scripts/run_migrations.sh
 
-      - name: Install cargo-llvm-cov
-        uses: taiki-e/install-action@cargo-llvm-cov
+      - name: Install cargo-llvm-cov and cargo-nextest
+        uses: taiki-e/install-action@v2
+        with:
+          tool: cargo-llvm-cov,cargo-nextest
 
       - name: Set up BigQuery Credentials
         run: |
@@ -161,15 +160,12 @@ jobs:
       - name: Generate Code Coverage
         id: coverage
         run: |
-          cargo llvm-cov test \
-            --workspace --no-fail-fast \
-            --all-features \
-            --lcov --output-path lcov.info
+          cargo llvm-cov nextest --no-fail-fast \
+            --all-features --lcov --output-path lcov.info
 
       - name: Upload Coverage to Coveralls
         uses: coverallsapp/github-action@v2
         with:
           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

AGENTS.md

@@ -6,8 +6,8 @@
 - Tests live per crate (`src` unit tests, `tests` integration); benches in `etl-benchmarks/benches/`.
 
 ## Build and Test
-- Build: `cargo build --workspace --all-targets --all-features`.
-- Lint/format: `cargo fmt`; `cargo clippy --all-targets --all-features -- -D warnings`.
+- Build: `just build`.
+- Lint/format: `just fmt; just lint`.
 - Use `ENABLE_TRACING=1` when running integration tests to see the logs.
 - Use `RUST_LOG=[log-level]` if you need to see the logs with a specific log level.
 

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

docs/how-to/configure-postgres.md

@@ -107,15 +107,17 @@ Publications define which tables and operations to replicate.
 
 ```sql
 -- Create publication for specific tables
-CREATE PUBLICATION my_publication FOR TABLE users, orders;
+CREATE PUBLICATION my_publication FOR TABLE users, orders WITH (publish_via_partition_root = true);
 
 -- Create publication for all tables (use with caution)
-CREATE PUBLICATION all_tables FOR ALL TABLES;
+CREATE PUBLICATION all_tables FOR ALL TABLES WITH (publish_via_partition_root = true);
 
 -- Include only specific operations
-CREATE PUBLICATION inserts_only FOR TABLE users WITH (publish = 'insert');
+CREATE PUBLICATION inserts_only FOR TABLE users WITH (publish = 'insert') WITH (publish_via_partition_root = true);
 ```
 
+`publish_via_partition_root` allows Postgres to treat [partitioned tables](https://www.postgresql.org/docs/current/sql-createpublication.html#SQL-CREATEPUBLICATION-PARAMS-WITH-PUBLISH-VIA-PARTITION-ROOT) as one table in the eyes of logical replication.
+
 ### Managing Publications
 
 ```sql

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-api/src/db/publications.rs

@@ -43,6 +43,9 @@ pub async fn create_publication(
         }
     }
 
+    // Ensure partitioned tables publish via ancestor/root schema for logical replication
+    query.push_str(" with (publish_via_partition_root = true)");
+
     pool.execute(query.as_str()).await?;
     Ok(())
 }

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"]