Feat/deltalake s3

.github/workflows/general.yml

@@ -50,7 +50,7 @@ jobs:
       #     key: sqlx
       - name: Run non-api tests
         run: |
-          cargo test --workspace --all-features --exclude api
+          cargo test --workspace --all-features --exclude api --jobs 1
 
   fmt:
     name: Rustfmt
@@ -109,4 +109,4 @@ jobs:
         with:
           file: ./replicator/Dockerfile
           push: true
-          tags: ${{ vars.DOCKERHUB_USERNAME }}/replicator:${{ github.head_ref || github.ref_name }}.${{ github.sha }}
+          tags: ${{ vars.DOCKERHUB_USERNAME }}/replicator:${{ github.head_ref || github.ref_name }}.${{ github.sha }}

.vscode/settings.json

@@ -1,3 +1,3 @@
 {
-  "rust-analyzer.cargo.features": ["duckdb", "bigquery", "stdout"]
+  "rust-analyzer.cargo.features": ["duckdb", "bigquery", "stdout", "delta"]
 }

Cargo.toml

@@ -9,7 +9,11 @@ actix-web = { version = "4", default-features = false }
 actix-web-httpauth = { version = "0.8.2", default-features = false }
 anyhow = { version = "1.0", default-features = false }
 async-trait = { version = "0.1" }
+aws-credential-types = { version = "1.2.0", default-features = false }
+aws-config = { version = "1.5.10", default-features = false }
 aws-lc-rs = { version = "1.8.1", default-features = false }
+aws-sdk-sts = { version = "1.39.0", default-features = false }
+aws-types = { version = "1.3.3", default-features = false }
 base64 = { version = "0.22.1", default-features = false }
 bigdecimal = { version = "0.4.6", default-features = false }
 bytes = { version = "1.0" }
@@ -18,6 +22,8 @@ chrono = { version = "0.4", default-features = false }
 clap = { version = "4.5", default-features = false }
 config = { version = "0.14", default-features = false }
 constant_time_eq = { version = "0.3.1" }
+deltalake = { version = "0.22.0", default-features = false }
+deltalake-aws = { version = "0.3.0", default-features = false }
 duckdb = { version = "1.0", default-features = false, features = ["bundled"] }
 futures = { version = "0.3.31", default-features = false }
 # gcp-bigquery-client = { version = "0.24.1", default-features = false }
@@ -48,7 +54,6 @@ tracing-subscriber = { version = "0.3", default-features = false }
 utoipa = { version = "4.2.3", default-features = false }
 utoipa-swagger-ui = { version = "7.1.0", default-features = false }
 uuid = { version = "1.10.0", default-features = false }
-deltalake = {version="0.22.0",default-features = false}
 
 
 # [patch."https://github.com/imor/gcp-bigquery-client"]

README.md

@@ -125,6 +125,7 @@ The repository is a cargo workspace. Each of the individual sub-folders are crat
 - [x] Add BigQuery Sink
 - [x] Add DuckDb Sink
 - [x] Add MotherDuck Sink
+- [x] Add Deltalake Sink (local/s3)
 - [ ] Add Snowflake Sink
 - [ ] Add ClickHouse Sink
 - [ ] Many more to come...

pg_replicate/Cargo.toml

@@ -23,12 +23,20 @@ required-features = ["delta"]
 
 [dependencies]
 async-trait = { workspace = true }
+aws-config = { workspace = true, optional = true }
+aws-credential-types = { workspace = true, optional = true }
+aws-sdk-sts = { workspace = true, features = ["rustls"], optional = true }
+aws-types = { workspace = true, optional = true }
 bigdecimal = { workspace = true, features = ["std"] }
 bytes = { workspace = true }
 byteorder = { workspace = true }
 chrono = { workspace = true, features = ["serde"] }
 duckdb = { workspace = true, optional = true }
-deltalake = { workspace = true,features=["datafusion"],optional=true }
+deltalake = { workspace = true, features = [
+    "datafusion",
+    "s3",
+], optional = true }
+deltalake-aws = { workspace = true, optional = true }
 futures = { workspace = true }
 gcp-bigquery-client = { workspace = true, optional = true, features = [
     "rust-tls",
@@ -66,7 +74,14 @@ tracing-subscriber = { workspace = true, default-features = true, features = [
 bigquery = ["dep:gcp-bigquery-client", "dep:prost"]
 duckdb = ["dep:duckdb"]
 stdout = []
-delta = ["dep:deltalake"]
+delta = [
+    "dep:deltalake",
+    "dep:aws-credential-types",
+    "dep:aws-sdk-sts",
+    "dep:aws-types",
+    "dep:aws-config",
+    "dep:deltalake-aws",
+]
 # When enabled converts unknown types to bytes
 unknown_types_to_bytes = []
 default = ["unknown_types_to_bytes"]

pg_replicate/examples/delta.rs

@@ -51,7 +51,9 @@ struct DbArgs {
 
 #[derive(Debug, Args)]
 struct DeltaArgs {
-    /// Path to the Delta Lake for saving data from the database
+    /// The path to the Delta Lake where data from the database will be saved.
+    /// Use `file:///home/data/datalake` for saving data to local file storage.
+    /// Use `s3://datalake` for saving data to an S3 bucket.
     #[arg(long)]
     delta_path: String,
 }