feat(connectors): Delta Lake Sink Connector

Cargo.toml

@@ -33,6 +33,7 @@ members = [
     "core/configs_derive",
     "core/connectors/runtime",
     "core/connectors/sdk",
+    "core/connectors/sinks/delta_sink",
     "core/connectors/sinks/elasticsearch_sink",
     "core/connectors/sinks/http_sink",
     "core/connectors/sinks/iceberg_sink",
@@ -132,6 +133,7 @@ cyper = { version = "0.8.3", features = ["rustls"], default-features = false }
 cyper-axum = { version = "0.8.0" }
 darling = "0.23"
 dashmap = "6.1.0"
+deltalake = { version = "0.30.2", features = ["azure", "gcs", "s3"] }
 derive-new = "0.7.0"
 derive_builder = "0.20.2"
 derive_more = { version = "2.1.1", features = ["full"] }
@@ -294,6 +296,7 @@ tungstenite = "0.29.0"
 twox-hash = { version = "2.1.2", features = ["xxhash32"] }
 ulid = "1.2.1"
 ureq = "2.10"
+url = "2.5.8"
 uuid = { version = "1.23.0", features = ["v4", "v7", "fast-rng", "serde", "zerocopy"] }
 vergen-git2 = { version = "9.1.0", features = ["build", "cargo", "rustc", "si"] }
 walkdir = "2.5.0"

DEPENDENCIES.md

@@ -32,21 +32,25 @@ anstyle-query: 1.1.5, "Apache-2.0 OR MIT",
 anstyle-wincon: 3.0.11, "Apache-2.0 OR MIT",
 anyhow: 1.0.102, "Apache-2.0 OR MIT",
 apache-avro: 0.21.0, "Apache-2.0",
+ar_archive_writer: 0.5.1, "Apache-2.0 WITH LLVM-exception",
 arbitrary: 1.4.2, "Apache-2.0 OR MIT",
 arc-swap: 1.9.1, "Apache-2.0 OR MIT",
 arg_enum_proc_macro: 0.3.4, "MIT",
 argon2: 0.5.3, "Apache-2.0 OR MIT",
 array-init: 2.1.0, "Apache-2.0 OR MIT",
 arrayref: 0.3.9, "BSD-2-Clause",
 arrayvec: 0.7.6, "Apache-2.0 OR MIT",
+arrow: 57.3.0, "Apache-2.0",
 arrow-arith: 57.3.0, "Apache-2.0",
 arrow-array: 57.3.0, "Apache-2.0",
 arrow-buffer: 57.3.0, "Apache-2.0",
 arrow-cast: 57.3.0, "Apache-2.0",
+arrow-csv: 57.3.0, "Apache-2.0",
 arrow-data: 57.3.0, "Apache-2.0",
 arrow-ipc: 57.3.0, "Apache-2.0",
 arrow-json: 57.3.0, "Apache-2.0",
 arrow-ord: 57.3.0, "Apache-2.0",
+arrow-row: 57.3.0, "Apache-2.0",
 arrow-schema: 57.3.0, "Apache-2.0",
 arrow-select: 57.3.0, "Apache-2.0",
 arrow-string: 57.3.0, "Apache-2.0",
@@ -80,8 +84,27 @@ autotools: 0.2.7, "MIT",
 av-scenechange: 0.14.1, "MIT",
 av1-grain: 0.2.5, "BSD-2-Clause",
 avif-serialize: 0.8.8, "BSD-3-Clause",
+aws-config: 1.8.15, "Apache-2.0",
+aws-credential-types: 1.2.14, "Apache-2.0",
 aws-lc-rs: 1.16.2, "(Apache-2.0 OR ISC) AND ISC",
 aws-lc-sys: 0.39.1, "(Apache-2.0 OR ISC OR MIT) AND (Apache-2.0 OR ISC OR MIT-0) AND (Apache-2.0 OR ISC) AND Apache-2.0 AND BSD-3-Clause AND ISC AND MIT",
+aws-runtime: 1.7.2, "Apache-2.0",
+aws-sdk-dynamodb: 1.108.0, "Apache-2.0",
+aws-sdk-sso: 1.96.0, "Apache-2.0",
+aws-sdk-ssooidc: 1.98.0, "Apache-2.0",
+aws-sdk-sts: 1.100.0, "Apache-2.0",
+aws-sigv4: 1.4.2, "Apache-2.0",
+aws-smithy-async: 1.2.14, "Apache-2.0",
+aws-smithy-http: 0.63.6, "Apache-2.0",
+aws-smithy-http-client: 1.1.12, "Apache-2.0",
+aws-smithy-json: 0.62.5, "Apache-2.0",
+aws-smithy-observability: 0.2.6, "Apache-2.0",
+aws-smithy-query: 0.60.15, "Apache-2.0",
+aws-smithy-runtime: 1.10.3, "Apache-2.0",
+aws-smithy-runtime-api: 1.11.6, "Apache-2.0",
+aws-smithy-types: 1.4.6, "Apache-2.0",
+aws-smithy-xml: 0.60.15, "Apache-2.0",
+aws-types: 1.3.14, "Apache-2.0",
 axum: 0.8.8, "MIT",
 axum-core: 0.5.6, "MIT",
 axum-macros: 0.5.0, "MIT",
@@ -140,6 +163,7 @@ bytemuck_derive: 1.10.2, "Apache-2.0 OR MIT OR Zlib",
 byteorder: 1.5.0, "MIT OR Unlicense",
 byteorder-lite: 0.1.0, "MIT OR Unlicense",
 bytes: 1.11.1, "MIT",
+bytes-utils: 0.1.4, "Apache-2.0 OR MIT",
 bytestring: 1.5.0, "Apache-2.0 OR MIT",
 bzip2: 0.6.1, "Apache-2.0 OR MIT",
 camino: 1.2.2, "Apache-2.0 OR MIT",
@@ -156,6 +180,7 @@ chacha20: 0.10.0, "Apache-2.0 OR MIT",
 charming: 0.6.0, "Apache-2.0 OR MIT",
 charming_macros: 0.1.0, "Apache-2.0 OR MIT",
 chrono: 0.4.44, "Apache-2.0 OR MIT",
+chrono-tz: 0.10.4, "Apache-2.0 OR MIT",
 cipher: 0.4.4, "Apache-2.0 OR MIT",
 clang-sys: 1.8.1, "Apache-2.0",
 clap: 4.6.0, "Apache-2.0 OR MIT",
@@ -197,6 +222,7 @@ const-random: 0.1.18, "Apache-2.0 OR MIT",
 const-random-macro: 0.1.16, "Apache-2.0 OR MIT",
 constant_time_eq: 0.4.2, "Apache-2.0 OR CC0-1.0 OR MIT-0",
 convert_case: 0.6.0, "MIT",
+convert_case: 0.9.0, "MIT",
 convert_case: 0.10.0, "MIT",
 cooked-waker: 5.0.0, "MPL-2.0",
 cookie: 0.16.2, "Apache-2.0 OR MIT",
@@ -227,6 +253,7 @@ crypto-common: 0.1.7, "Apache-2.0 OR MIT",
 crypto-common: 0.2.1, "Apache-2.0 OR MIT",
 csv: 1.4.0, "MIT OR Unlicense",
 csv-core: 0.1.13, "MIT OR Unlicense",
+ctor: 0.6.3, "Apache-2.0 OR MIT",
 ctor: 0.8.0, "Apache-2.0 OR MIT",
 ctor-proc-macro: 0.0.7, "Apache-2.0 OR MIT",
 ctr: 0.9.2, "Apache-2.0 OR MIT",
@@ -256,6 +283,14 @@ dbus-secret-service: 4.1.0, "Apache-2.0 OR MIT",
 deadpool: 0.12.3, "Apache-2.0 OR MIT",
 deadpool-runtime: 0.1.4, "Apache-2.0 OR MIT",
 debugid: 0.8.0, "Apache-2.0",
+delta_kernel: 0.19.2, "Apache-2.0",
+delta_kernel_derive: 0.19.2, "Apache-2.0",
+deltalake: 0.30.2, "Apache-2.0",
+deltalake-aws: 0.13.1, "Apache-2.0",
+deltalake-azure: 0.13.0, "Apache-2.0",
+deltalake-core: 0.30.2, "Apache-2.0",
+deltalake-derive: 0.30.0, "Apache-2.0",
+deltalake-gcp: 0.14.0, "Apache-2.0",
 deno_core: 0.351.0, "MIT",
 deno_core_icudata: 0.74.0, "MIT",
 deno_error: 0.6.1, "MIT",
@@ -291,6 +326,7 @@ document-features: 0.2.12, "Apache-2.0 OR MIT",
 dotenvy: 0.15.7, "MIT",
 downcast: 0.11.0, "MIT",
 dtoa: 1.0.11, "Apache-2.0 OR MIT",
+dtor: 0.1.1, "Apache-2.0 OR MIT",
 dtor: 0.3.0, "Apache-2.0 OR MIT",
 dtor-proc-macro: 0.0.6, "Apache-2.0 OR MIT",
 dunce: 1.0.5, "Apache-2.0 OR CC0-1.0 OR MIT-0",
@@ -442,6 +478,7 @@ home: 0.5.12, "Apache-2.0 OR MIT",
 hostname: 0.4.2, "MIT",
 http: 0.2.12, "Apache-2.0 OR MIT",
 http: 1.4.0, "Apache-2.0 OR MIT",
+http-body: 0.4.6, "MIT",
 http-body: 1.0.1, "MIT",
 http-body-util: 0.1.3, "MIT",
 http-range: 0.1.5, "MIT",
@@ -483,6 +520,7 @@ iggy-connectors: 0.4.0, "Apache-2.0",
 iggy-mcp: 0.4.0, "Apache-2.0",
 iggy_binary_protocol: 0.10.0, "Apache-2.0",
 iggy_common: 0.10.0, "Apache-2.0",
+iggy_connector_delta_sink: 0.1.0, "Apache-2.0",
 iggy_connector_elasticsearch_sink: 0.4.0, "Apache-2.0",
 iggy_connector_elasticsearch_source: 0.4.0, "Apache-2.0",
 iggy_connector_http_sink: 0.4.0, "Apache-2.0",
@@ -652,6 +690,8 @@ objc2: 0.6.4, "MIT",
 objc2-core-foundation: 0.3.2, "Apache-2.0 OR MIT OR Zlib",
 objc2-encode: 4.1.0, "MIT",
 objc2-io-kit: 0.3.2, "Apache-2.0 OR MIT OR Zlib",
+object: 0.37.3, "Apache-2.0 OR MIT",
+object_store: 0.12.5, "Apache-2.0 OR MIT",
 octocrab: 0.49.7, "Apache-2.0 OR MIT",
 oid-registry: 0.8.1, "Apache-2.0 OR MIT",
 once_cell: 1.21.4, "Apache-2.0 OR MIT",
@@ -700,10 +740,13 @@ peg-runtime: 0.6.3, "MIT",
 pem: 3.0.6, "MIT",
 pem-rfc7468: 0.7.0, "Apache-2.0 OR MIT",
 percent-encoding: 2.3.2, "Apache-2.0 OR MIT",
+percent-encoding-rfc3986: 0.1.3, "Apache-2.0 OR MIT",
 pest: 2.8.6, "Apache-2.0 OR MIT",
 pest_derive: 2.8.6, "Apache-2.0 OR MIT",
 pest_generator: 2.8.6, "Apache-2.0 OR MIT",
 pest_meta: 2.8.6, "Apache-2.0 OR MIT",
+phf: 0.12.1, "MIT",
+phf_shared: 0.12.1, "MIT",
 pico-args: 0.5.0, "MIT",
 pin-project: 1.1.11, "Apache-2.0 OR MIT",
 pin-project-internal: 1.1.11, "Apache-2.0 OR MIT",
@@ -734,6 +777,8 @@ proc-macro-crate: 1.3.1, "Apache-2.0 OR MIT",
 proc-macro-crate: 3.3.0, "Apache-2.0 OR MIT",
 proc-macro-error: 1.0.4, "Apache-2.0 OR MIT",
 proc-macro-error-attr: 1.0.4, "Apache-2.0 OR MIT",
+proc-macro-error-attr2: 2.0.0, "Apache-2.0 OR MIT",
+proc-macro-error2: 2.0.1, "Apache-2.0 OR MIT",
 proc-macro-rules: 0.4.0, "Apache-2.0 OR MIT",
 proc-macro-rules-macros: 0.4.0, "Apache-2.0 OR MIT",
 proc-macro2: 1.0.106, "Apache-2.0 OR MIT",
@@ -748,6 +793,7 @@ prost-reflect: 0.16.3, "Apache-2.0 OR MIT",
 prost-types: 0.14.3, "Apache-2.0",
 protox: 0.9.1, "Apache-2.0 OR MIT",
 protox-parse: 0.9.0, "Apache-2.0 OR MIT",
+psm: 0.1.30, "Apache-2.0 OR MIT",
 ptr_meta: 0.1.4, "MIT",
 ptr_meta_derive: 0.1.4, "MIT",
 pxfm: 0.1.28, "Apache-2.0 OR BSD-3-Clause",
@@ -779,6 +825,8 @@ raw-cpuid: 11.6.0, "MIT",
 rayon: 1.11.0, "Apache-2.0 OR MIT",
 rayon-core: 1.13.0, "Apache-2.0 OR MIT",
 rcgen: 0.14.7, "Apache-2.0 OR MIT",
+recursive: 0.1.1, "MIT",
+recursive-proc-macro-impl: 0.1.1, "MIT",
 redox_syscall: 0.5.18, "MIT",
 redox_syscall: 0.7.4, "MIT",
 redox_users: 0.5.2, "MIT",
@@ -904,6 +952,7 @@ sourcemap: 9.3.2, "BSD-3-Clause",
 spin: 0.9.8, "MIT",
 spinning_top: 0.3.0, "Apache-2.0 OR MIT",
 spki: 0.7.3, "Apache-2.0 OR MIT",
+sqlparser: 0.59.0, "Apache-2.0",
 sqlx: 0.8.6, "Apache-2.0 OR MIT",
 sqlx-core: 0.8.6, "Apache-2.0 OR MIT",
 sqlx-macros: 0.8.6, "Apache-2.0 OR MIT",
@@ -913,6 +962,7 @@ sqlx-postgres: 0.8.6, "Apache-2.0 OR MIT",
 sqlx-sqlite: 0.8.6, "Apache-2.0 OR MIT",
 sse-stream: 0.2.1, "Apache-2.0 OR MIT",
 stable_deref_trait: 1.2.1, "Apache-2.0 OR MIT",
+stacker: 0.1.23, "Apache-2.0 OR MIT",
 static-toml: 1.3.0, "MIT",
 static_assertions: 1.1.0, "Apache-2.0 OR MIT",
 strict-num: 0.1.1, "MIT",
@@ -1051,6 +1101,8 @@ uuid: 1.23.0, "Apache-2.0 OR MIT",
 v8: 137.3.0, "MIT",
 v_frame: 0.3.9, "BSD-2-Clause",
 v_htmlescape: 0.15.8, "Apache-2.0 OR MIT",
+validator: 0.19.0, "MIT",
+validator_derive: 0.19.0, "MIT",
 valuable: 0.1.1, "MIT",
 value-trait: 0.12.1, "Apache-2.0 OR MIT",
 vcpkg: 0.2.15, "Apache-2.0 OR MIT",
@@ -1168,6 +1220,7 @@ writeable: 0.6.3, "Unicode-3.0",
 wyz: 0.5.1, "MIT",
 x509-parser: 0.18.1, "Apache-2.0 OR MIT",
 xattr: 1.6.1, "Apache-2.0 OR MIT",
+xmlparser: 0.13.6, "Apache-2.0 OR MIT",
 xmlwriter: 0.1.0, "MIT",
 y4m: 0.8.0, "MIT",
 yansi: 1.0.1, "Apache-2.0 OR MIT",
@@ -1178,6 +1231,7 @@ yew-router: 0.20.0, "Apache-2.0 OR MIT",
 yew-router-macro: 0.20.0, "Apache-2.0 OR MIT",
 yoke: 0.8.2, "Unicode-3.0",
 yoke-derive: 0.8.2, "Unicode-3.0",
+z85: 3.0.7, "Apache-2.0 OR MIT",
 zerocopy: 0.8.48, "Apache-2.0 OR BSD-2-Clause OR MIT",
 zerocopy-derive: 0.8.48, "Apache-2.0 OR BSD-2-Clause OR MIT",
 zerofrom: 0.1.7, "Unicode-3.0",

core/connectors/runtime/example_config/connectors/delta_sink.toml

@@ -0,0 +1,35 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+type = "sink"
+key = "delta"
+enabled = true
+version = 0
+name = "Delta Lake sink"
+path = "target/release/libiggy_connector_delta_sink"
+verbose = true
+
+[[streams]]
+stream = "qw"
+topics = ["records"]
+schema = "json"
+batch_length = 100
+poll_interval = "5ms"
+consumer_group = "delta_sink_connector"
+
+[plugin_config]
+table_uri = "file:///tmp/iggy_testing/iggy_delta_table"

core/connectors/sdk/src/convert.rs

@@ -50,3 +50,117 @@ pub fn owned_value_to_serde_json(value: &simd_json::OwnedValue) -> serde_json::V
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use simd_json::{OwnedValue, StaticNode};
+
+    #[test]
+    fn test_null() {
+        assert_eq!(
+            owned_value_to_serde_json(&OwnedValue::Static(StaticNode::Null)),
+            serde_json::Value::Null
+        );
+    }
+
+    #[test]
+    fn test_bool() {
+        assert_eq!(
+            owned_value_to_serde_json(&OwnedValue::Static(StaticNode::Bool(true))),
+            serde_json::Value::Bool(true)
+        );
+        assert_eq!(
+            owned_value_to_serde_json(&OwnedValue::Static(StaticNode::Bool(false))),
+            serde_json::Value::Bool(false)
+        );
+    }
+
+    #[test]
+    fn test_i64() {
+        assert_eq!(
+            owned_value_to_serde_json(&OwnedValue::Static(StaticNode::I64(-42))),
+            serde_json::json!(-42)
+        );
+    }
+
+    #[test]
+    fn test_u64() {
+        assert_eq!(
+            owned_value_to_serde_json(&OwnedValue::Static(StaticNode::U64(100))),
+            serde_json::json!(100)
+        );
+    }
+
+    #[test]
+    fn test_f64_finite() {
+        assert_eq!(
+            owned_value_to_serde_json(&OwnedValue::Static(StaticNode::F64(1.5))),
+            serde_json::json!(1.5)
+        );
+    }
+
+    #[test]
+    fn test_f64_nan_maps_to_null() {
+        assert_eq!(
+            owned_value_to_serde_json(&OwnedValue::Static(StaticNode::F64(f64::NAN))),
+            serde_json::Value::Null
+        );
+    }
+
+    #[test]
+    fn test_f64_infinity_maps_to_null() {
+        assert_eq!(
+            owned_value_to_serde_json(&OwnedValue::Static(StaticNode::F64(f64::INFINITY))),
+            serde_json::Value::Null
+        );
+        assert_eq!(
+            owned_value_to_serde_json(&OwnedValue::Static(StaticNode::F64(f64::NEG_INFINITY))),
+            serde_json::Value::Null
+        );
+    }
+
+    #[test]
+    fn test_string() {
+        assert_eq!(
+            owned_value_to_serde_json(&OwnedValue::String("hello".into())),
+            serde_json::Value::String("hello".to_string())
+        );
+    }
+
+    #[test]
+    fn test_array() {
+        let input = OwnedValue::Array(Box::new(vec![
+            OwnedValue::Static(StaticNode::Null),
+            OwnedValue::Static(StaticNode::Bool(true)),
+        ]));
+        assert_eq!(
+            owned_value_to_serde_json(&input),
+            serde_json::json!([null, true])
+        );
+    }
+
+    #[test]
+    fn test_object() {
+        let mut obj = simd_json::owned::Object::new();
+        obj.insert("k".into(), OwnedValue::Static(StaticNode::I64(1)));
+        let input = OwnedValue::Object(Box::new(obj));
+        assert_eq!(
+            owned_value_to_serde_json(&input),
+            serde_json::json!({"k": 1})
+        );
+    }
+
+    #[test]
+    fn test_nested_object() {
+        let mut inner = simd_json::owned::Object::new();
+        inner.insert("b".into(), OwnedValue::String("v".into()));
+        let mut outer = simd_json::owned::Object::new();
+        outer.insert("a".into(), OwnedValue::Object(Box::new(inner)));
+        let input = OwnedValue::Object(Box::new(outer));
+        assert_eq!(
+            owned_value_to_serde_json(&input),
+            serde_json::json!({"a": {"b": "v"}})
+        );
+    }
+}