diff --git a/Cargo.lock b/Cargo.lock index a08a49adb9..1e39204d76 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -67,7 +67,7 @@ dependencies = [ "actix-rt", "actix-service", "actix-utils", - "base64", + "base64 0.22.1", "bitflags 2.13.1", "brotli", "bytes", @@ -425,7 +425,7 @@ dependencies = [ "digest 0.10.7", "log", "miniz_oxide", - "num-bigint", + "num-bigint 0.4.8", "quad-rand", "rand 0.9.5", "regex-lite", @@ -525,17 +525,38 @@ dependencies = [ "arrow-array 58.3.0", "arrow-buffer 58.3.0", "arrow-cast 58.3.0", - "arrow-csv", + "arrow-csv 58.3.0", "arrow-data 58.3.0", "arrow-ipc 58.3.0", "arrow-json 58.3.0", "arrow-ord 58.3.0", - "arrow-row", + "arrow-row 58.3.0", "arrow-schema 58.3.0", "arrow-select 58.3.0", "arrow-string 58.3.0", ] +[[package]] +name = "arrow" +version = "59.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "61d285d16bce7d0be61912f7928342b673067b6b7d7ef6cc179258ba7de1fecf" +dependencies = [ + "arrow-arith 59.2.0", + "arrow-array 59.2.0", + "arrow-buffer 59.2.0", + "arrow-cast 59.2.0", + "arrow-csv 59.2.0", + "arrow-data 59.2.0", + "arrow-ipc 59.2.0", + "arrow-json 59.2.0", + "arrow-ord 59.2.0", + "arrow-row 59.2.0", + "arrow-schema 59.2.0", + "arrow-select 59.2.0", + "arrow-string 59.2.0", +] + [[package]] name = "arrow-arith" version = "57.3.1" @@ -564,6 +585,20 @@ dependencies = [ "num-traits", ] +[[package]] +name = "arrow-arith" +version = "59.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "757ef1836251e88222542a7da2623bc1c9cb9e20afefa6db2c41e79991cd91d4" +dependencies = [ + "arrow-array 59.2.0", + "arrow-buffer 59.2.0", + "arrow-data 59.2.0", + "arrow-schema 59.2.0", + "chrono", + "num-traits", +] + [[package]] name = "arrow-array" version = "57.3.1" @@ -601,6 +636,25 @@ dependencies = [ "num-traits", ] +[[package]] +name = "arrow-array" +version = "59.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bc9a4a4b2b5ecd0e04df03471661cb61f28bed3c7fd50994715129b01b2edb97" +dependencies = [ + "ahash 0.8.12", + "arrow-buffer 59.2.0", + "arrow-data 59.2.0", + "arrow-schema 59.2.0", + "chrono", + "half", + "hashbrown 0.17.1", + "libc", + "num-complex", + "num-integer", + "num-traits", +] + [[package]] name = "arrow-buffer" version = "57.3.1" @@ -609,7 +663,7 @@ checksum = "0d18b89b4c4f4811d0858175e79541fe98e33e18db3b011708bc287b1240593f" dependencies = [ "bytes", "half", - "num-bigint", + "num-bigint 0.4.8", "num-traits", ] @@ -621,7 +675,19 @@ checksum = "0c6cd424c2693bcdbc150d843dc9d4d137dd2de4782ce6df491ad11a3a0416c0" dependencies = [ "bytes", "half", - "num-bigint", + "num-bigint 0.4.8", + "num-traits", +] + +[[package]] +name = "arrow-buffer" +version = "59.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c12b576ef18c1deb80925a248b25ad84f419198d791b8e293fc6aaa60441fe90" +dependencies = [ + "bytes", + "half", + "num-bigint 0.5.1", "num-traits", ] @@ -638,7 +704,7 @@ dependencies = [ "arrow-schema 57.3.1", "arrow-select 57.3.1", "atoi", - "base64", + "base64 0.22.1", "chrono", "half", "lexical-core", @@ -659,7 +725,28 @@ dependencies = [ "arrow-schema 58.3.0", "arrow-select 58.3.0", "atoi", - "base64", + "base64 0.22.1", + "chrono", + "half", + "lexical-core", + "num-traits", + "ryu", +] + +[[package]] +name = "arrow-cast" +version = "59.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "68338a9096a5dc9bc11927c58c43a8526d96bf6abd2012ef6c0c9f505991cc79" +dependencies = [ + "arrow-array 59.2.0", + "arrow-buffer 59.2.0", + "arrow-data 59.2.0", + "arrow-ord 59.2.0", + "arrow-schema 59.2.0", + "arrow-select 59.2.0", + "atoi", + "base64 0.23.1", "chrono", "half", "lexical-core", @@ -682,6 +769,21 @@ dependencies = [ "regex", ] +[[package]] +name = "arrow-csv" +version = "59.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "25011b52b346407d497ef0030e12b45e4f2d0cc279efc09c4f3d09106db30e36" +dependencies = [ + "arrow-array 59.2.0", + "arrow-cast 59.2.0", + "arrow-schema 59.2.0", + "chrono", + "csv", + "csv-core", + "regex", +] + [[package]] name = "arrow-data" version = "57.3.1" @@ -708,6 +810,19 @@ dependencies = [ "num-traits", ] +[[package]] +name = "arrow-data" +version = "59.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "723fe4aeed7604e00b9883a465af4ff0a0e6c44c03e41a68c3d1cbc403e0e44d" +dependencies = [ + "arrow-buffer 59.2.0", + "arrow-schema 59.2.0", + "half", + "num-integer", + "num-traits", +] + [[package]] name = "arrow-ipc" version = "57.3.1" @@ -736,6 +851,22 @@ dependencies = [ "flatbuffers", ] +[[package]] +name = "arrow-ipc" +version = "59.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "149437b14371f5b9ec60f5ddc751483ae99d7a7072653c0075e5e469156eea7b" +dependencies = [ + "arrow-array 59.2.0", + "arrow-buffer 59.2.0", + "arrow-data 59.2.0", + "arrow-schema 59.2.0", + "arrow-select 59.2.0", + "flatbuffers", + "lz4_flex 0.14.0", + "zstd", +] + [[package]] name = "arrow-json" version = "57.3.1" @@ -785,6 +916,31 @@ dependencies = [ "simdutf8", ] +[[package]] +name = "arrow-json" +version = "59.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f18b9123ccfec418a663f821c9a034af339711678c11ffe00d3ec07da5ff9f7e" +dependencies = [ + "arrow-array 59.2.0", + "arrow-buffer 59.2.0", + "arrow-cast 59.2.0", + "arrow-ord 59.2.0", + "arrow-schema 59.2.0", + "arrow-select 59.2.0", + "chrono", + "half", + "indexmap 2.14.0", + "itoa", + "lexical-core", + "memchr", + "num-traits", + "ryu", + "serde_core", + "serde_json", + "simdutf8", +] + [[package]] name = "arrow-ord" version = "57.3.1" @@ -811,6 +967,19 @@ dependencies = [ "arrow-select 58.3.0", ] +[[package]] +name = "arrow-ord" +version = "59.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6c08dff0686cf23ca4f562803f191ccbeb726dbae6309cd4b4aaf65e0f2c979" +dependencies = [ + "arrow-array 59.2.0", + "arrow-buffer 59.2.0", + "arrow-data 59.2.0", + "arrow-schema 59.2.0", + "arrow-select 59.2.0", +] + [[package]] name = "arrow-row" version = "58.3.0" @@ -824,6 +993,19 @@ dependencies = [ "half", ] +[[package]] +name = "arrow-row" +version = "59.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bbec439386df71ad570e6758a946111322b9e9dc8db83b5527321f0b4c9119c2" +dependencies = [ + "arrow-array 59.2.0", + "arrow-buffer 59.2.0", + "arrow-data 59.2.0", + "arrow-schema 59.2.0", + "half", +] + [[package]] name = "arrow-schema" version = "57.3.1" @@ -841,6 +1023,15 @@ dependencies = [ "serde_core", ] +[[package]] +name = "arrow-schema" +version = "59.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6fed2ca0d1eade57e811cbe73b98ad50cc08a1183e13b2d2aa43a7df593f40e" +dependencies = [ + "bitflags 2.13.1", +] + [[package]] name = "arrow-select" version = "57.3.1" @@ -869,6 +1060,20 @@ dependencies = [ "num-traits", ] +[[package]] +name = "arrow-select" +version = "59.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "466b19cf75130b891dc1b23a84b343c714c62c64c9c62e365c76aa0ff90a53fb" +dependencies = [ + "ahash 0.8.12", + "arrow-array 59.2.0", + "arrow-buffer 59.2.0", + "arrow-data 59.2.0", + "arrow-schema 59.2.0", + "num-traits", +] + [[package]] name = "arrow-string" version = "57.3.1" @@ -903,6 +1108,23 @@ dependencies = [ "regex-syntax", ] +[[package]] +name = "arrow-string" +version = "59.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c838a25bb3691e919e0f617616ac51a4ff8517a952e29ca133cf0c22b2ce65b1" +dependencies = [ + "arrow-array 59.2.0", + "arrow-buffer 59.2.0", + "arrow-data 59.2.0", + "arrow-schema 59.2.0", + "arrow-select 59.2.0", + "memchr", + "num-traits", + "regex", + "regex-syntax", +] + [[package]] name = "as-any" version = "0.3.2" @@ -1284,7 +1506,7 @@ version = "0.30.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "16e2cdb6d5ed835199484bb92bb8b3edd526effe995c61732580439c1a67e2e9" dependencies = [ - "base64", + "base64 0.22.1", "http 1.4.2", "log", "rustls", @@ -1916,6 +2138,12 @@ version = "0.22.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" +[[package]] +name = "base64" +version = "0.23.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac07cdecf99051d9a5238b80f35af32cdeba5b336e55d957b318b50137e18da5" + [[package]] name = "base64-simd" version = "0.8.0" @@ -2019,7 +2247,7 @@ checksum = "4d6867f1565b3aad85681f1015055b087fcfd840d6aeee6eee7f2da317603695" dependencies = [ "autocfg", "libm", - "num-bigint", + "num-bigint 0.4.8", "num-integer", "num-traits", "serde", @@ -2234,7 +2462,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ee04c4c84f1f811b017f2fbb7dd8815c976e7ca98593de9c1e2afad0f636bff4" dependencies = [ "async-stream", - "base64", + "base64 0.22.1", "bitflags 2.13.1", "bollard-buildkit-proto", "bollard-stubs", @@ -2291,7 +2519,7 @@ version = "1.52.1-rc.29.1.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0f0a8ca8799131c1837d1282c3f81f31e76ceb0ce426e04a7fe1ccee3287c066" dependencies = [ - "base64", + "base64 0.22.1", "bollard-buildkit-proto", "bytes", "prost", @@ -2387,7 +2615,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7969a9ba84b0ff843813e7249eed1678d9b6607ce5a3b8f0a47af3fcf7978e6e" dependencies = [ "ahash 0.8.12", - "base64", + "base64 0.22.1", "bitvec", "getrandom 0.2.17", "getrandom 0.3.4", @@ -2432,7 +2660,7 @@ version = "0.22.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2235eb320cd7178862a32dd111bd0c0f71a368e393add4914c50129add478eab" dependencies = [ - "arrow", + "arrow 58.3.0", "buoyant_kernel_derive", "bytes", "chrono", @@ -3727,7 +3955,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5f8c8abaf78cfe3cb7838d7cfcf75e4e3bf02fe3e54571e0a1f10bb485cf8fbf" dependencies = [ "async-stream", - "base64", + "base64 0.22.1", "cfg-if", "cfg_aliases", "compio", @@ -3970,6 +4198,17 @@ dependencies = [ "thiserror 2.0.19", ] +[[package]] +name = "delegate" +version = "0.13.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "780eb241654bf097afb00fc5f054a09b687dad862e485fdcf8399bb056565370" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "deltalake" version = "0.32.4" @@ -4031,7 +4270,7 @@ version = "0.32.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4588e95ff3b2ccdba56d9ec262bd3467c0593000f729402528706f62be8be1ca" dependencies = [ - "arrow", + "arrow 58.3.0", "arrow-arith 58.3.0", "arrow-array 58.3.0", "arrow-buffer 58.3.0", @@ -4039,7 +4278,7 @@ dependencies = [ "arrow-ipc 58.3.0", "arrow-json 58.3.0", "arrow-ord 58.3.0", - "arrow-row", + "arrow-row 58.3.0", "arrow-schema 58.3.0", "arrow-select 58.3.0", "async-trait", @@ -4236,7 +4475,7 @@ dependencies = [ "asn1-rs", "displaydoc", "nom 7.1.3", - "num-bigint", + "num-bigint 0.4.8", "num-traits", "rusticata-macros", ] @@ -4466,7 +4705,7 @@ version = "1.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "29547a1dc60885a552306986316bc9701ba120c1a8db6769fa68691529ad373d" dependencies = [ - "base64", + "base64 0.22.1", "serde", "serde_json", ] @@ -4578,7 +4817,7 @@ version = "9.1.0-alpha.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "12bb303aa6e1d28c0c86b6fbfe484fd0fd3f512629aeed1ac4f6b85f81d9834a" dependencies = [ - "base64", + "base64 0.22.1", "bytes", "dyn-clone", "flate2", @@ -5018,6 +5257,45 @@ dependencies = [ "spin", ] +[[package]] +name = "fluss-rs" +version = "1.0.0" +source = "git+https://github.com/seokjin0414/fluss.git?rev=70882755ebefefcdb5e285c4a817903ba728cac2#70882755ebefefcdb5e285c4a817903ba728cac2" +dependencies = [ + "arrow 59.2.0", + "arrow-schema 59.2.0", + "bigdecimal", + "bitvec", + "byteorder", + "bytes", + "clap", + "crc32c", + "dashmap", + "delegate", + "futures", + "jiff", + "linked-hash-map", + "log", + "metrics", + "opendal", + "ordered-float 5.3.0", + "parking_lot", + "parse-display 0.10.0", + "prost", + "rand 0.9.5", + "scopeguard", + "serde", + "serde_json", + "snafu", + "strum 0.26.3", + "strum_macros 0.26.4", + "tempfile", + "thiserror 1.0.69", + "tokio", + "url", + "uuid", +] + [[package]] name = "fnv" version = "1.0.7" @@ -6324,7 +6602,7 @@ version = "0.1.20" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0" dependencies = [ - "base64", + "base64 0.22.1", "bytes", "futures-channel", "futures-util", @@ -6402,7 +6680,7 @@ dependencies = [ "as-any", "async-trait", "backon", - "base64", + "base64 0.22.1", "bimap", "bytes", "chrono", @@ -6822,7 +7100,7 @@ dependencies = [ "aes-gcm", "async-broadcast", "async-trait", - "base64", + "base64 0.22.1", "blake3", "bon", "byte-unit", @@ -6890,7 +7168,7 @@ name = "iggy_connector_doris_sink" version = "0.2.0-edge.4" dependencies = [ "async-trait", - "base64", + "base64 0.22.1", "blake3", "bytes", "humantime", @@ -6910,7 +7188,7 @@ name = "iggy_connector_elasticsearch_sink" version = "0.5.0-edge.4" dependencies = [ "async-trait", - "base64", + "base64 0.22.1", "dashmap", "elasticsearch", "iggy_common", @@ -6941,12 +7219,29 @@ dependencies = [ "tracing", ] +[[package]] +name = "iggy_connector_fluss_sink" +version = "0.4.1-edge.1" +dependencies = [ + "arrow 59.2.0", + "async-trait", + "fluss-rs", + "iggy_common", + "iggy_connector_sdk", + "secrecy", + "serde", + "serde_json", + "thiserror 2.0.19", + "tokio", + "tracing", +] + [[package]] name = "iggy_connector_http_sink" version = "0.5.0-edge.4" dependencies = [ "async-trait", - "base64", + "base64 0.22.1", "bytes", "humantime", "iggy_connector_sdk", @@ -6988,7 +7283,7 @@ version = "0.5.0-edge.4" dependencies = [ "async-trait", "axum", - "base64", + "base64 0.22.1", "bytes", "iggy_common", "iggy_connector_sdk", @@ -7010,7 +7305,7 @@ dependencies = [ "ahash 0.8.12", "async-trait", "axum", - "base64", + "base64 0.22.1", "chrono", "csv", "dashmap", @@ -7034,7 +7329,7 @@ name = "iggy_connector_meilisearch_sink" version = "0.5.0-edge.4" dependencies = [ "async-trait", - "base64", + "base64 0.22.1", "iggy_common", "iggy_connector_sdk", "meilisearch-sdk", @@ -7087,7 +7382,7 @@ name = "iggy_connector_postgres_source" version = "0.5.0-edge.4" dependencies = [ "async-trait", - "base64", + "base64 0.22.1", "chrono", "dashmap", "futures", @@ -7140,7 +7435,7 @@ name = "iggy_connector_s3_sink" version = "0.5.0-edge.4" dependencies = [ "async-trait", - "base64", + "base64 0.22.1", "byte-unit", "chrono", "dashmap", @@ -7163,7 +7458,7 @@ dependencies = [ "anyhow", "apache-avro", "async-trait", - "base64", + "base64 0.22.1", "dashmap", "flatbuffers", "http 1.4.2", @@ -7209,7 +7504,7 @@ name = "iggy_connector_surrealdb_sink" version = "0.5.0-edge.4" dependencies = [ "async-trait", - "base64", + "base64 0.22.1", "bytes", "iggy_common", "iggy_connector_sdk", @@ -7417,7 +7712,7 @@ version = "0.0.1" dependencies = [ "assert_cmd", "async-trait", - "base64", + "base64 0.22.1", "bon", "bytemuck", "bytes", @@ -7430,6 +7725,7 @@ dependencies = [ "deltalake", "dtor 1.0.5", "figment", + "fluss-rs", "futures", "harness_derive", "humantime", @@ -7577,10 +7873,12 @@ dependencies = [ "jiff-core", "jiff-static", "jiff-tzdb-platform", + "js-sys", "log", "portable-atomic", "portable-atomic-util", "serde_core", + "wasm-bindgen", "windows-link 0.2.1", ] @@ -7710,7 +8008,7 @@ version = "10.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "eba32bfb4ffdeaca3e34431072faf01745c9b26d25504aa7a6cf5684334fc4fc" dependencies = [ - "base64", + "base64 0.22.1", "ed25519-dalek", "getrandom 0.2.17", "hmac 0.12.1", @@ -8066,6 +8364,9 @@ name = "log" version = "0.4.33" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad" +dependencies = [ + "value-bag", +] [[package]] name = "logos" @@ -8424,6 +8725,16 @@ dependencies = [ "tracing", ] +[[package]] +name = "metrics" +version = "0.24.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "89550ee9f79e88fef3119de263694973a8adb26c21d75322164fb8c493039fe2" +dependencies = [ + "portable-atomic", + "rapidhash", +] + [[package]] name = "miette" version = "7.6.0" @@ -8584,7 +8895,7 @@ version = "3.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b814038f367d212f55de0a630cb35102a9b8ca23785a86955d62c0087c93846d" dependencies = [ - "base64", + "base64 0.22.1", "bitflags 2.13.1", "bson", "derive-where", @@ -8790,7 +9101,7 @@ version = "0.4.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "35bd024e8b2ff75562e5f34e7f4905839deb4b22955ef5e73d2fea1b9813cb23" dependencies = [ - "num-bigint", + "num-bigint 0.4.8", "num-complex", "num-integer", "num-iter", @@ -8810,6 +9121,16 @@ dependencies = [ "serde", ] +[[package]] +name = "num-bigint" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93e7820bc0a80a0238e650327316f929ba18d5be054b647490a3a6a339f3e7c0" +dependencies = [ + "num-integer", + "num-traits", +] + [[package]] name = "num-bigint-dig" version = "0.8.6" @@ -8893,7 +9214,7 @@ version = "0.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f83d14da390562dca69fc84082e73e548e1ad308d24accdedd2720017cb37824" dependencies = [ - "num-bigint", + "num-bigint 0.4.8", "num-integer", "num-traits", ] @@ -9000,7 +9321,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "622acbc9100d3c10e2ee15804b0caa40e55c933d5aa53814cd520805b7958a49" dependencies = [ "async-trait", - "base64", + "base64 0.22.1", "bytes", "chrono", "form_urlencoded", @@ -9041,7 +9362,7 @@ checksum = "27be39870c558e1fbc5a5f8d4a29aa916268c1dbcb9d467f2a9bbab35598060e" dependencies = [ "arc-swap", "async-trait", - "base64", + "base64 0.22.1", "bytes", "cargo_metadata", "cfg-if", @@ -9107,7 +9428,7 @@ checksum = "d075ab8a203a6ab4bc1bce0a4b9fe486a72bf8b939037f4b78d95386384bc80a" dependencies = [ "anyhow", "backon", - "base64", + "base64 0.22.1", "bytes", "crc32c", "futures", @@ -9254,6 +9575,17 @@ dependencies = [ "num-traits", ] +[[package]] +name = "ordered-float" +version = "5.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b7d950ca161dc355eaf28f82b11345ed76c6e1f6eb1f4f4479e0323b9e2fbd0e" +dependencies = [ + "num-traits", + "rand 0.8.7", + "serde", +] + [[package]] name = "ordered-multimap" version = "0.7.3" @@ -9357,7 +9689,7 @@ dependencies = [ "arrow-ipc 57.3.1", "arrow-schema 57.3.1", "arrow-select 57.3.1", - "base64", + "base64 0.22.1", "brotli", "bytes", "chrono", @@ -9366,7 +9698,7 @@ dependencies = [ "half", "hashbrown 0.16.1", "lz4_flex 0.12.2", - "num-bigint", + "num-bigint 0.4.8", "num-integer", "num-traits", "paste", @@ -9392,7 +9724,7 @@ dependencies = [ "arrow-ipc 58.3.0", "arrow-schema 58.3.0", "arrow-select 58.3.0", - "base64", + "base64 0.22.1", "brotli", "bytes", "chrono", @@ -9401,7 +9733,7 @@ dependencies = [ "half", "hashbrown 0.17.1", "lz4_flex 0.13.1", - "num-bigint", + "num-bigint 0.4.8", "num-integer", "num-traits", "object_store", @@ -9421,7 +9753,18 @@ version = "0.9.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "914a1c2265c98e2446911282c6ac86d8524f495792c38c5bd884f80499c7538a" dependencies = [ - "parse-display-derive", + "parse-display-derive 0.9.1", + "regex", + "regex-syntax", +] + +[[package]] +name = "parse-display" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "287d8d3ebdce117b8539f59411e4ed9ec226e0a4153c7f55495c6070d68e6f72" +dependencies = [ + "parse-display-derive 0.10.0", "regex", "regex-syntax", ] @@ -9440,6 +9783,20 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "parse-display-derive" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7fc048687be30d79502dea2f623d052f3a074012c6eac41726b7ab17213616b1" +dependencies = [ + "proc-macro2", + "quote", + "regex", + "regex-syntax", + "structmeta", + "syn 2.0.119", +] + [[package]] name = "partitions" version = "0.1.0" @@ -9568,7 +9925,7 @@ version = "3.0.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d30c53c26bc5b31a98cd02d20f25a7c8567146caf63ed593a9d87b2775291be" dependencies = [ - "base64", + "base64 0.22.1", "serde_core", ] @@ -10352,6 +10709,7 @@ dependencies = [ "libc", "rand_chacha 0.3.1", "rand_core 0.6.4", + "serde", ] [[package]] @@ -10402,6 +10760,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c" dependencies = [ "getrandom 0.2.17", + "serde", ] [[package]] @@ -10437,6 +10796,15 @@ dependencies = [ "rand_core 0.10.1", ] +[[package]] +name = "rapidhash" +version = "4.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5da7e78a036ce858e8d55b7e7dc8ba3a88b78350fd2155d3591bbd966b58589e" +dependencies = [ + "rustversion", +] + [[package]] name = "rav1e" version = "0.8.1" @@ -10648,7 +11016,7 @@ checksum = "43451dbf3590a7590684c25fb8d12ecdcc90ed3ac123433e500447c7d77ed701" dependencies = [ "anyhow", "async-trait", - "base64", + "base64 0.22.1", "chrono", "form_urlencoded", "getrandom 0.2.17", @@ -10675,7 +11043,7 @@ version = "0.12.28" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "eddd3ca559203180a307f12d114c268abf583f59b03cb906fd0b3ff8646c1147" dependencies = [ - "base64", + "base64 0.22.1", "bytes", "futures-core", "futures-util", @@ -10718,7 +11086,7 @@ version = "0.13.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "219c5811de6525e5416c7d5d53bb656d3afdbc6c5af816e0802bcfa42dbdc1c3" dependencies = [ - "base64", + "base64 0.22.1", "bytes", "encoding_rs", "futures-channel", @@ -10916,7 +11284,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "14db48ee17a9ba61810ab1a9c1beb7d06d8136ae39ac25a1137f10d357af01af" dependencies = [ "async-trait", - "base64", + "base64 0.22.1", "bytes", "chrono", "futures", @@ -11078,7 +11446,7 @@ dependencies = [ "async-trait", "aws-creds", "aws-region", - "base64", + "base64 0.22.1", "bytes", "cfg-if", "futures-util", @@ -11582,6 +11950,15 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "serde_fmt" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e497af288b3b95d067a23a4f749f2861121ffcb2f6d8379310dcda040c345ed" +dependencies = [ + "serde_core", +] + [[package]] name = "serde_json" version = "1.0.151" @@ -11655,7 +12032,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d4f284b4d521591b17ddee01aff830dd005a04476f7862aca9298c038d00fb7e" dependencies = [ "deno_error", - "num-bigint", + "num-bigint 0.4.8", "serde", "smallvec", "thiserror 2.0.19", @@ -11668,7 +12045,7 @@ version = "3.21.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "76a5c54c7310e7b8b9577c286d7e399ddd876c3e12b3ed917a8aabc4b96e9e8c" dependencies = [ - "base64", + "base64 0.22.1", "bs58", "chrono", "hex", @@ -12039,7 +12416,7 @@ version = "0.6.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0d585997b0ac10be3c5ee635f1bab02d512760d14b7c468801ac8a01d9ae5f1d" dependencies = [ - "num-bigint", + "num-bigint 0.4.8", "num-traits", "thiserror 2.0.19", "time", @@ -12258,7 +12635,7 @@ version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "05b44e85bf579a8eeb4ceaa77a3a523baf2bf0e9bac7e40f405d537b5d2d5ccb" dependencies = [ - "base64", + "base64 0.22.1", "bytes", "cfg-if", "chrono", @@ -12364,7 +12741,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "87a2bdd6e83f6b3ea525ca9fee568030508b58355a43d0b2c1674d5f79dcd65e" dependencies = [ "atoi", - "base64", + "base64 0.22.1", "bitflags 2.13.1", "byteorder", "chrono", @@ -12527,6 +12904,12 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "strum" +version = "0.26.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fec0f0aef304996cf250b31b5a10dee7980c85da9d759361292b8bca5a18f06" + [[package]] name = "strum" version = "0.27.2" @@ -12545,6 +12928,19 @@ dependencies = [ "strum_macros 0.28.0", ] +[[package]] +name = "strum_macros" +version = "0.26.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4c6bee85a5a24955dc440386795aa378cd9cf82acd5f764469152d2270e581be" +dependencies = [ + "heck", + "proc-macro2", + "quote", + "rustversion", + "syn 2.0.119", +] + [[package]] name = "strum_macros" version = "0.27.2" @@ -12575,6 +12971,84 @@ version = "2.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" +[[package]] +name = "sval" +version = "2.20.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5fb9efbae90f97301f4d25f3be63dfd99d6b7af9d088228a52ec960d649b2e7d" + +[[package]] +name = "sval_buffer" +version = "2.20.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff5c0280ea0af40b3a1fd0b532680b5067482ffb81412ee66ec04b1d9952b49a" +dependencies = [ + "sval", + "sval_ref", +] + +[[package]] +name = "sval_dynamic" +version = "2.20.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "59b9067f2e68f58e110cf8019a268057e3791cf74b0fdb1b3c7c6e49104f44e6" +dependencies = [ + "sval", +] + +[[package]] +name = "sval_fmt" +version = "2.20.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "96ebdbf0e4b175884aa587fcf551c16dabe245ea04235abdd8cbbd160b27ed5a" +dependencies = [ + "itoa", + "ryu", + "sval", +] + +[[package]] +name = "sval_json" +version = "2.20.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e448d9fa216a6c16670b28d624fbbcf5c04e41eb187bd7c52e01222ffa72a12" +dependencies = [ + "itoa", + "ryu", + "sval", +] + +[[package]] +name = "sval_nested" +version = "2.20.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "021de5b5c26efd544c694cef9b8a9abe8633481bf3be1ea145a18a740818b291" +dependencies = [ + "sval", + "sval_buffer", + "sval_ref", +] + +[[package]] +name = "sval_ref" +version = "2.20.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "54ef5ffec8bc52ded04ee424ab8d959e25e64fd40a48a16d21f8991ce824c1bb" +dependencies = [ + "sval", +] + +[[package]] +name = "sval_serde" +version = "2.20.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b983833a8a2390f89ebcf9b9acd06883017014b4ffd72ee28e0c9de6852039f5" +dependencies = [ + "serde_core", + "sval", + "sval_nested", +] + [[package]] name = "svgtypes" version = "0.15.3" @@ -12878,7 +13352,7 @@ dependencies = [ "itertools 0.14.0", "log", "memchr", - "parse-display", + "parse-display 0.9.1", "pin-project-lite", "reqwest 0.13.4", "serde", @@ -13282,7 +13756,7 @@ checksum = "ac2a5518c70fa84342385732db33fb3f44bc4cc748936eb5833d2df34d6445ef" dependencies = [ "async-trait", "axum", - "base64", + "base64 0.22.1", "bytes", "h2 0.4.15", "http 1.4.2", @@ -13844,7 +14318,7 @@ version = "3.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dea7109cdcd5864d4eeb1b58a1648dc9bf520360d7af16ec26d0a9354bafcfc0" dependencies = [ - "base64", + "base64 0.22.1", "flate2", "log", "percent-encoding", @@ -13861,7 +14335,7 @@ version = "0.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e994ba84b0bd1b1b0cf92878b7ef898a5c1760108fe7b6010327e274917a808c" dependencies = [ - "base64", + "base64 0.22.1", "http 1.4.2", "httparse", "log", @@ -13892,7 +14366,7 @@ version = "0.45.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "80be9b06fbae3b8b303400ab20778c80bbaf338f563afe567cf3c9eea17b47ef" dependencies = [ - "base64", + "base64 0.22.1", "data-url", "flate2", "fontdb", @@ -14020,6 +14494,42 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" +[[package]] +name = "value-bag" +version = "1.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ef73bfbaf3216cb59c205d7176bee1194e0d84348979da31f4a71fefe3c2054e" +dependencies = [ + "value-bag-serde1", + "value-bag-sval2", +] + +[[package]] +name = "value-bag-serde1" +version = "1.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5b92170db3db8a6354f12a5b7f13a5453928433e08fc46aee51eedfa8f7a28a1" +dependencies = [ + "erased-serde", + "serde_core", + "serde_fmt", +] + +[[package]] +name = "value-bag-sval2" +version = "1.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0bf9832097ca044466ae3f1aa43a943d4e450b7c3b8cdf65363875df07da79a0" +dependencies = [ + "sval", + "sval_buffer", + "sval_dynamic", + "sval_fmt", + "sval_json", + "sval_ref", + "sval_serde", +] + [[package]] name = "value-trait" version = "0.12.2" @@ -14782,7 +15292,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "08db1edfb05d9b3c1542e521aea074442088292f00b5f28e435c714a98f85031" dependencies = [ "assert-json-diff", - "base64", + "base64 0.22.1", "deadpool", "futures", "http 1.4.2", diff --git a/Cargo.toml b/Cargo.toml index 7cefb5fdf7..87aec5cea4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -37,6 +37,7 @@ members = [ "core/connectors/sinks/delta_sink", "core/connectors/sinks/doris_sink", "core/connectors/sinks/elasticsearch_sink", + "core/connectors/sinks/fluss_sink", "core/connectors/sinks/http_sink", "core/connectors/sinks/iceberg_sink", "core/connectors/sinks/influxdb_sink", @@ -176,6 +177,7 @@ figment = { version = "0.10.19", features = ["toml", "env"] } file-operation = "0.8.28" flatbuffers = "25.12.19" flume = "0.12.0" +fluss-rs = { git = "https://github.com/seokjin0414/fluss.git", rev = "70882755ebefefcdb5e285c4a817903ba728cac2" } fs2 = "0.4.3" futures = "0.3.33" futures-core = { version = "0.3.33", default-features = false } diff --git a/core/connectors/sinks/fluss_sink/Cargo.toml b/core/connectors/sinks/fluss_sink/Cargo.toml new file mode 100644 index 0000000000..fc1d1b98cd --- /dev/null +++ b/core/connectors/sinks/fluss_sink/Cargo.toml @@ -0,0 +1,48 @@ +# 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. + +[package] +name = "iggy_connector_fluss_sink" +version = "0.4.1-edge.1" +description = "Iggy Apache Fluss sink connector for storing stream messages into Apache Fluss storage" +edition = "2024" +license = "Apache-2.0" +keywords = ["iggy", "messaging", "streaming", "fluss", "sink"] +categories = ["command-line-utilities", "database", "network-programming"] +homepage = "https://iggy.apache.org" +documentation = "https://iggy.apache.org/docs" +repository = "https://github.com/apache/iggy" +readme = "README.md" +publish = false + +[lib] +crate-type = ["cdylib", "lib"] + +[dependencies] +arrow = "59.2.0" +async-trait = { workspace = true } +fluss-rs = { workspace = true } +iggy_common = { workspace = true } +iggy_connector_sdk = { workspace = true } +secrecy = { workspace = true } +serde = { workspace = true } +thiserror = { workspace = true } +tokio = { workspace = true } +tracing = { workspace = true } + +[dev-dependencies] +serde_json = { workspace = true } diff --git a/core/connectors/sinks/fluss_sink/README.md b/core/connectors/sinks/fluss_sink/README.md new file mode 100644 index 0000000000..857ad7ddc4 --- /dev/null +++ b/core/connectors/sinks/fluss_sink/README.md @@ -0,0 +1,236 @@ +# Apache Fluss Sink Connector + +The Apache Fluss sink connector consumes messages from Apache Iggy streams and +appends them to an [Apache Fluss](https://fluss.apache.org/) log table. It can +create the target table automatically, preserve selected Iggy metadata, and +store payloads as either Fluss `BYTES` or `STRING`. + +The connector uses the +[Fluss Rust client](https://clients.fluss.apache.org/user-guide/rust/api-reference/) +and exposes its writer, connection, and security configuration. + +## Features + +- Appends Iggy messages to a Fluss log table. +- Creates the target table on demand. +- Flushes all pending writes before each consumed batch completes. +- Supports Fluss writer retries, idempotence, buffering, and backpressure. +- Supports `PLAINTEXT` and SASL `PLAIN` client configuration. +- Optionally stores Iggy checksum, origin timestamp, and stream metadata. +- Stores payloads as Fluss `BYTES` or `STRING`. + +## Build + +From the repository root: + +```bash +cargo build --release -p iggy_connector_fluss_sink +``` + +The connector runtime loads the resulting dynamic library from +`target/release/`. Adjust the `path` setting for the working directory and +operating system used by the runtime. + +## Configuration + +The following example uses the default plugin settings explicitly. See +[`config.toml`](config.toml) for the complete configuration file. + +```toml +type = "sink" +key = "fluss" +enabled = true +version = 0 +name = "Fluss sink" +path = "target/release/libiggy_connector_fluss_sink" +verbose = false + +[[streams]] +stream = "user_events" +topics = ["users", "orders"] +schema = "json" +batch_length = 100 +poll_interval = "5ms" +consumer_group = "fluss_sink" + +[plugin_config] +bootstrap_servers = "127.0.0.1:9123" +target_database = "fluss" +target_table = "iggy_messages" +auto_create_table = true +include_metadata = true +include_checksum = true +include_origin_timestamp = true +payload_format = "json" +``` + +All plugin fields have defaults. Existing configurations remain valid when new +fields are added because missing fields use the connector defaults. + +### Connector settings + +| Field | Type | Default | Description | +| --- | --- | --- | --- | +| `target_database` | string | `"fluss"` | Target Fluss database. The database must already exist. | +| `target_table` | string | `"iggy_messages"` | Target Fluss table. | +| `auto_create_table` | bool | `true` | Create the target table if it does not exist before writing a batch. Existing tables are left unchanged. | +| `include_metadata` | bool | `true` | Add the Iggy offset, timestamp, stream, topic, and partition columns. | +| `include_checksum` | bool | `true` | Add the Iggy message checksum column. | +| `include_origin_timestamp` | bool | `true` | Add the Iggy origin timestamp column. | +| `payload_format` | enum | `"json"` | Payload storage format: `bytea`, `json`, or `text`. | + +### Fluss writer and connection settings + +| Field | Type | Default | Description | +| --- | --- | --- | --- | +| `bootstrap_servers` | string | `"127.0.0.1:9123"` | Fluss coordinator address. | +| `writer_request_max_size` | i32 | `10485760` | Maximum writer request size in bytes. | +| `writer_acks` | string | `"all"` | Required acknowledgements. `"all"` waits for all required replicas. | +| `writer_retries` | i32 | `2147483647` | Maximum retries for transient writer failures. | +| `writer_batch_size` | i32 | `2097152` | Target Fluss writer batch size in bytes. | +| `writer_batch_timeout_ms` | i64 | `100` | Maximum time to wait for a writer batch to fill before sending it. | +| `writer_bucket_no_key_assigner` | enum | `"sticky"` | Bucket selection for tables without bucket keys: `sticky` or `round_robin`. | +| `writer_enable_idempotence` | bool | `true` | Add writer IDs and per-bucket sequence numbers so Fluss can deduplicate retried batches. | +| `writer_max_inflight_requests_per_bucket` | usize | `5` | Maximum unacknowledged requests per bucket. Idempotent writes require a value no greater than `5`. | +| `writer_buffer_memory_size` | usize | `67108864` | Total memory in bytes available for buffered write batches. | +| `writer_buffer_wait_timeout_ms` | string | `"18446744073709551615"` | Maximum time to wait for writer buffer memory. This is a string because the default is `u64::MAX`, which TOML integers cannot represent. | +| `connect_timeout_ms` | u64 | `120000` | TCP connection timeout in milliseconds. | + +Idempotent writes require `writer_acks = "all"` or `"-1"`, +`writer_retries > 0`, and +`writer_max_inflight_requests_per_bucket <= 5`. + +### Security settings + +| Field | Type | Default | Description | +| --- | --- | --- | --- | +| `security_protocol` | string | `"PLAINTEXT"` | Use `"PLAINTEXT"` without authentication or `"sasl"` for SASL authentication. Matching is case-insensitive. | +| `security_sasl_mechanism` | string | `"PLAIN"` | SASL mechanism. The pinned Fluss client supports only `PLAIN`. | +| `security_sasl_username` | string | `""` | SASL username. Required when `security_protocol = "sasl"`. | +| `security_sasl_password` | string | `""` | SASL password. Required when `security_protocol = "sasl"` and omitted from serialized connector configuration. | + +Example: + +```toml +[plugin_config] +bootstrap_servers = "fluss.example.com:9123" +security_protocol = "sasl" +security_sasl_mechanism = "PLAIN" +security_sasl_username = "iggy" +security_sasl_password = "replace-with-secret" +``` + +## Payload formats + +| Value | Fluss type | Behavior | +| --- | --- | --- | +| `bytea` | `BYTES` | Serializes the Iggy payload to bytes and preserves it in a binary column. | +| `json` | `STRING` | Serializes the payload to bytes and stores the resulting UTF-8 string. | +| `text` | `STRING` | Serializes the payload to bytes and stores the resulting UTF-8 string. | + +`json` and `text` currently use the same Fluss schema and row conversion. The +sink does not parse or validate JSON itself. Configure the Iggy stream with +`schema = "json"` when JSON validation is required before the sink receives the +message. + +Any Iggy payload variant can be written with `bytea`. The `json` and `text` +formats reject payload bytes that are not valid UTF-8. + +## Generated table schema + +When `auto_create_table = true`, the connector creates an append-only Fluss log +table without a primary key. Columns are generated in the following order: + +| Column | Fluss type | Included when | +| --- | --- | --- | +| `id` | `STRING` | Always | +| `checksum` | `DECIMAL(20, 0)` | `include_checksum = true` | +| `iggy_offset` | `DECIMAL(20, 0)` | `include_metadata = true` | +| `iggy_timestamp` | `TIMESTAMP_LTZ(6)` | `include_metadata = true` | +| `iggy_stream` | `STRING` | `include_metadata = true` | +| `iggy_topic` | `STRING` | `include_metadata = true` | +| `iggy_partition_id` | `BIGINT` | `include_metadata = true` | +| `iggy_origin_timestamp` | `TIMESTAMP_LTZ(6)` | `include_origin_timestamp = true` | +| `payload` | `BYTES` or `STRING` | Always | + +Message IDs are encoded as 32-character lowercase hexadecimal strings. +`DECIMAL(20, 0)` preserves the complete unsigned 64-bit range for offsets and +checksums. Timestamps are interpreted as microseconds since the Unix epoch. + +### Manual table creation with Flink SQL + +When `auto_create_table = false`, create the database and table through a +[Fluss catalog in Flink SQL](https://fluss.apache.org/docs/engine-flink/getting-started/) +before starting the connector. The following definition matches the default +`target_database = "fluss"`, `target_table = "iggy_messages"`, and +`payload_format = "json"` settings: + +```sql +USE CATALOG fluss_catalog; + +CREATE DATABASE IF NOT EXISTS `fluss`; +USE `fluss`; + +CREATE TABLE `iggy_messages` ( + `id` STRING COMMENT 'Apache Iggy message ID', + `checksum` DECIMAL(20, 0) COMMENT 'Apache Iggy message checksum', + `iggy_offset` DECIMAL(20, 0) COMMENT 'Apache Iggy message offset', + `iggy_timestamp` TIMESTAMP_LTZ(6) + COMMENT 'Apache Iggy message timestamp', + `iggy_stream` STRING COMMENT 'Apache Iggy stream name', + `iggy_topic` STRING COMMENT 'Apache Iggy topic name', + `iggy_partition_id` BIGINT COMMENT 'Apache Iggy partition ID', + `iggy_origin_timestamp` TIMESTAMP_LTZ(6) + COMMENT 'Apache Iggy message origin timestamp', + `payload` STRING COMMENT 'Apache Iggy message payload' +) +COMMENT 'Stores Apache Iggy messages written by the Fluss sink connector'; +``` + +Replace `fluss_catalog` with the name of the Fluss catalog configured in the +Flink SQL client. If `payload_format = "bytea"`, define `payload` as `BYTES` +instead of `STRING`. + +Omit `checksum` when `include_checksum = false`. Omit `iggy_offset`, +`iggy_timestamp`, `iggy_stream`, `iggy_topic`, and `iggy_partition_id` when +`include_metadata = false`. Omit `iggy_origin_timestamp` when +`include_origin_timestamp = false`. Keep the remaining columns in the order +shown above. + +Message headers are not stored. + +The connector does not migrate or alter existing tables. A manually created +table must use the same column order and compatible Fluss data types. +Tables created by connector versions that used `STRING` for checksum, offset, +and timestamps must be recreated or migrated before using this schema. + +## Write behavior + +For every batch received from the Iggy connector runtime, the sink: + +1. Creates the table if `auto_create_table` is enabled and the table is missing. +2. Opens an append writer for the target table. +3. Converts each Iggy message to a Fluss row. +4. Appends every row and flushes the writer. + +The effective message count per call is controlled by the stream +`batch_length`. Fluss may combine those rows into byte-sized writer batches +according to `writer_batch_size` and `writer_batch_timeout_ms`. + +## Limitations + +- The connector writes append-only log tables and does not support primary-key + upserts. +- The target database is not created automatically. +- Existing table schemas are not migrated. +- The pinned `fluss-rs` 0.1 client does not expose a public graceful connection + shutdown method. Each consumed batch is flushed before returning, and + connector shutdown currently releases the client by dropping it. + +## Testing + +Run the Fluss sink unit tests from the repository root: + +```bash +cargo test -p iggy_connector_fluss_sink +``` diff --git a/core/connectors/sinks/fluss_sink/config.toml b/core/connectors/sinks/fluss_sink/config.toml new file mode 100644 index 0000000000..163f998d63 --- /dev/null +++ b/core/connectors/sinks/fluss_sink/config.toml @@ -0,0 +1,62 @@ +# 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 = "fluss" +enabled = true +version = 0 +name = "Fluss sink" +path = "../../target/release/libiggy_connector_fluss_sink" +verbose = false + +[[streams]] +stream = "user_events" +topics = ["users", "orders"] +schema = "json" +batch_length = 100 +poll_interval = "5ms" +consumer_group = "fluss_sink" + +[plugin_config] +# iggy's fluss table settings +target_database = "fluss" +target_table = "iggy_messages" +auto_create_table = true +include_metadata = true +include_checksum = true +include_origin_timestamp = true +# payload_format = "bytea" +payload_format = "json" +use_arrow_batch = true + +# fluss connection settings, passed to the fluss connector +bootstrap_servers = "127.0.0.1:9123" +writer_request_max_size = 10485760 +writer_acks = "all" +writer_retries = 2147483647 +writer_batch_size = 2097152 +writer_bucket_no_key_assigner = "sticky" +writer_batch_timeout_ms = 100 +writer_enable_idempotence = true +writer_max_inflight_requests_per_bucket = 5 +writer_buffer_memory_size = 67108864 +writer_buffer_wait_timeout_ms = "18446744073709551615" +connect_timeout_ms = 120000 +security_protocol = "PLAINTEXT" +security_sasl_mechanism = "PLAIN" +security_sasl_username = "" +security_sasl_password = "" diff --git a/core/connectors/sinks/fluss_sink/src/config.rs b/core/connectors/sinks/fluss_sink/src/config.rs new file mode 100644 index 0000000000..aa47c9d32f --- /dev/null +++ b/core/connectors/sinks/fluss_sink/src/config.rs @@ -0,0 +1,232 @@ +// 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. + +use fluss::config::{Config as FlussConfig, NoKeyAssigner}; +use iggy_connector_sdk::Error; +use secrecy::{ExposeSecret, SecretString}; +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum PayloadFormat { + Bytea, + #[default] + Json, + Text, +} + +#[derive(Debug, Serialize, Deserialize, Clone)] +#[serde(default)] +pub struct FlussSinkConfig { + pub bootstrap_servers: String, + pub writer_request_max_size: i32, + pub writer_acks: String, + pub writer_retries: i32, + pub writer_batch_size: i32, + pub writer_bucket_no_key_assigner: NoKeyAssigner, + pub writer_batch_timeout_ms: i64, + pub writer_enable_idempotence: bool, + pub writer_max_inflight_requests_per_bucket: usize, + pub writer_buffer_memory_size: usize, + pub writer_buffer_wait_timeout_ms: String, + pub connect_timeout_ms: u64, + pub security_protocol: String, + pub security_sasl_mechanism: String, + pub security_sasl_username: String, + #[serde(serialize_with = "iggy_common::serde_secret::serialize_secret")] + pub security_sasl_password: SecretString, + pub target_database: String, + pub target_table: String, + pub auto_create_table: bool, + pub include_metadata: bool, + pub include_checksum: bool, + pub include_origin_timestamp: bool, + pub payload_format: PayloadFormat, + pub use_arrow_batch: bool, +} + +impl Default for FlussSinkConfig { + fn default() -> Self { + let fluss_config = FlussConfig::default(); + Self { + bootstrap_servers: fluss_config.bootstrap_servers, + writer_request_max_size: fluss_config.writer_request_max_size, + writer_acks: fluss_config.writer_acks, + writer_retries: fluss_config.writer_retries, + writer_batch_size: fluss_config.writer_batch_size, + writer_bucket_no_key_assigner: fluss_config.writer_bucket_no_key_assigner, + writer_batch_timeout_ms: fluss_config.writer_batch_timeout_ms, + writer_enable_idempotence: fluss_config.writer_enable_idempotence, + writer_max_inflight_requests_per_bucket: fluss_config + .writer_max_inflight_requests_per_bucket, + writer_buffer_memory_size: fluss_config.writer_buffer_memory_size, + writer_buffer_wait_timeout_ms: fluss_config.writer_buffer_wait_timeout_ms.to_string(), + connect_timeout_ms: fluss_config.connect_timeout_ms, + security_protocol: fluss_config.security_protocol, + security_sasl_mechanism: fluss_config.security_sasl_mechanism, + security_sasl_username: fluss_config.security_sasl_username, + security_sasl_password: fluss_config.security_sasl_password.into(), + target_database: "fluss".to_string(), + target_table: "iggy_messages".to_string(), + auto_create_table: true, + include_metadata: true, + include_checksum: true, + include_origin_timestamp: true, + payload_format: PayloadFormat::default(), + use_arrow_batch: false, + } + } +} + +impl TryFrom<&FlussSinkConfig> for FlussConfig { + type Error = Error; + + fn try_from(config: &FlussSinkConfig) -> Result { + let writer_buffer_wait_timeout_ms = + config + .writer_buffer_wait_timeout_ms + .parse() + .map_err(|error| { + Error::InvalidConfigValue(format!( + "invalid writer_buffer_wait_timeout_ms '{}': {error}", + config.writer_buffer_wait_timeout_ms + )) + })?; + + Ok(Self { + bootstrap_servers: config.bootstrap_servers.clone(), + writer_request_max_size: config.writer_request_max_size, + writer_acks: config.writer_acks.clone(), + writer_retries: config.writer_retries, + writer_batch_size: config.writer_batch_size, + writer_bucket_no_key_assigner: config.writer_bucket_no_key_assigner, + writer_batch_timeout_ms: config.writer_batch_timeout_ms, + writer_enable_idempotence: config.writer_enable_idempotence, + writer_max_inflight_requests_per_bucket: config.writer_max_inflight_requests_per_bucket, + writer_buffer_memory_size: config.writer_buffer_memory_size, + writer_buffer_wait_timeout_ms, + connect_timeout_ms: config.connect_timeout_ms, + security_protocol: config.security_protocol.clone(), + security_sasl_mechanism: config.security_sasl_mechanism.clone(), + security_sasl_username: config.security_sasl_username.clone(), + security_sasl_password: config.security_sasl_password.expose_secret().to_string(), + ..FlussConfig::default() + }) + } +} + +#[cfg(test)] +mod tests { + use fluss::config::Config as FlussConfig; + use iggy_connector_sdk::Error; + use serde_json::json; + + use super::{FlussSinkConfig, PayloadFormat}; + + #[test] + fn given_default_sink_config_when_converting_should_match_fluss_defaults() { + let sink_config = FlussSinkConfig::default(); + let fluss_config = FlussConfig::try_from(&sink_config).expect("Sink config should convert"); + let actual = serde_json::to_value(fluss_config).expect("Fluss config should serialize"); + let expected = + serde_json::to_value(FlussConfig::default()).expect("Fluss config should serialize"); + + assert_eq!(actual, expected); + } + + #[test] + fn given_existing_sink_config_when_deserializing_should_apply_fluss_defaults() { + let config: FlussSinkConfig = serde_json::from_value(json!({ + "bootstrap_servers": "localhost:9123", + "target_database": "analytics", + "target_table": "events", + "auto_create_table": true, + "include_metadata": true, + "include_checksum": true, + "include_origin_timestamp": true, + "payload_format": "json" + })) + .expect("Existing Fluss sink config should deserialize"); + + assert_eq!(config.writer_batch_size, 2 * 1024 * 1024); + assert_eq!(config.writer_buffer_wait_timeout_ms, u64::MAX.to_string()); + assert_eq!(config.payload_format, PayloadFormat::Json); + assert_eq!(config.target_database, "analytics"); + assert_eq!(config.target_table, "events"); + } + + #[test] + fn given_supported_payload_formats_when_deserializing_should_return_matching_variants() { + for (value, expected) in [ + ("bytea", PayloadFormat::Bytea), + ("json", PayloadFormat::Json), + ("text", PayloadFormat::Text), + ] { + let config: FlussSinkConfig = serde_json::from_value(json!({ + "payload_format": value + })) + .expect("Supported payload format should deserialize"); + + assert_eq!(config.payload_format, expected); + } + } + + #[test] + fn given_unsupported_payload_format_when_deserializing_should_fail() { + let config = serde_json::from_value::(json!({ + "payload_format": "xml" + })); + + assert!(config.is_err()); + } + + #[test] + fn given_u64_max_as_string_when_converting_should_parse_value() { + let config: FlussSinkConfig = serde_json::from_value(json!({ + "writer_buffer_wait_timeout_ms": u64::MAX.to_string() + })) + .expect("String-encoded u64 should deserialize"); + let fluss_config = FlussConfig::try_from(&config).expect("Sink config should convert"); + + assert_eq!(fluss_config.writer_buffer_wait_timeout_ms, u64::MAX); + } + + #[test] + fn given_numeric_buffer_wait_timeout_when_deserializing_should_fail() { + let config = serde_json::from_value::(json!({ + "writer_buffer_wait_timeout_ms": 100 + })); + + assert!(config.is_err()); + } + + #[test] + fn given_invalid_buffer_wait_timeout_when_converting_should_return_invalid_config_value() { + let config = FlussSinkConfig { + writer_buffer_wait_timeout_ms: "invalid".to_string(), + ..FlussSinkConfig::default() + }; + + let error = FlussConfig::try_from(&config).expect_err("Invalid value should fail"); + + assert!(matches!( + error, + Error::InvalidConfigValue(message) + if message.contains("invalid writer_buffer_wait_timeout_ms 'invalid'") + )); + } +} diff --git a/core/connectors/sinks/fluss_sink/src/lib.rs b/core/connectors/sinks/fluss_sink/src/lib.rs new file mode 100644 index 0000000000..ceab5054fd --- /dev/null +++ b/core/connectors/sinks/fluss_sink/src/lib.rs @@ -0,0 +1,158 @@ +// 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. + +use async_trait::async_trait; +use fluss::metadata::TablePath; +use iggy_connector_sdk::{ + ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector, +}; +use tokio::sync::Mutex; +use tracing::{debug, info}; + +use crate::{schema::FlussTableLayout, writer::FlussWriter}; + +mod config; +mod schema; +mod writer; +pub use config::{FlussSinkConfig, PayloadFormat}; + +sink_connector!(FlussSink); + +#[derive(Debug)] +struct State { + invocations_count: u64, + insertion_errors: u64, + messages_processed: u64, +} + +#[derive(Debug)] +pub struct FlussSink { + id: u32, + state: Mutex, + fluss_writer: writer::FlussWriter, + fluss_config: FlussSinkConfig, + table_layout: Option, + table_path: TablePath, +} + +impl FlussSink { + pub fn new(id: u32, config: FlussSinkConfig) -> Self { + let table_path = + TablePath::new(config.target_database.clone(), config.target_table.clone()); + Self { + id, + state: Mutex::new(State { + invocations_count: 0, + messages_processed: 0, + insertion_errors: 0, + }), + fluss_writer: FlussWriter::new(config.clone()), + fluss_config: config, + table_layout: None, + table_path, + } + } +} + +#[async_trait] +impl Sink for FlussSink { + async fn open(&mut self) -> Result<(), Error> { + let table_layout = FlussTableLayout::from_config(&self.fluss_config); + self.fluss_writer.connect().await.map_err(Error::from)?; + + self.fluss_writer + .ensure_table_exists(&self.table_path, &table_layout) + .await + .map_err(Error::from)?; + + self.table_layout = Some(table_layout); + info!("Opened Fluss sink connector ID: {}", self.id); + Ok(()) + } + + async fn consume( + &self, + topic_metadata: &TopicMetadata, + messages_metadata: MessagesMetadata, + messages: Vec, + ) -> Result<(), Error> { + let invocation = { + let mut state = self.state.lock().await; + state.invocations_count += 1; + state.invocations_count + }; + + debug!( + "Fluss sink connector ID: {} received: {} messages, schema: {}, stream: {}, topic: {}, partition_id: {}, current_offset: {}, invocation: {}", + self.id, + messages.len(), + messages_metadata.schema, + topic_metadata.stream, + topic_metadata.topic, + messages_metadata.partition_id, + messages_metadata.current_offset, + invocation + ); + + let table_layout = self + .table_layout + .as_ref() + .ok_or_else(|| Error::InitError("Fluss table layout is not initialized".to_string()))?; + + let result = if self.fluss_config.use_arrow_batch { + self.fluss_writer + .write_to_table_arrow( + &self.table_path, + messages_metadata, + messages, + topic_metadata, + table_layout, + ) + .await + } else { + self.fluss_writer + .write_to_table( + &self.table_path, + messages_metadata, + messages, + topic_metadata, + table_layout, + ) + .await + }; + + match result { + Ok(r) => { + let mut state = self.state.lock().await; + state.insertion_errors += r.insertion_errors; + state.messages_processed += r.messages_processed; + Ok(()) + } + + Err(error) => Err(error.into()), + } + } + + async fn close(&mut self) -> Result<(), Error> { + let state = self.state.lock().await; + info!( + "Fluss sink ID: {} processed {} messages with {} errors", + self.id, state.messages_processed, state.insertion_errors + ); + self.fluss_writer.close().await.map_err(Into::into) + } +} diff --git a/core/connectors/sinks/fluss_sink/src/schema.rs b/core/connectors/sinks/fluss_sink/src/schema.rs new file mode 100644 index 0000000000..0f6d828b5b --- /dev/null +++ b/core/connectors/sinks/fluss_sink/src/schema.rs @@ -0,0 +1,977 @@ +// 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. + +use std::{string::FromUtf8Error, sync::Arc}; + +use arrow::array::{ + ArrayBuilder, ArrayRef, BinaryBuilder, Decimal128Builder, Int64Array, StringArray, + StringBuilder, StructArray, TimestampMicrosecondBuilder, +}; +use arrow::datatypes::{DataType, Field}; +use fluss::{ + error::Error as FlussError, + metadata::{Column, TableDescriptor}, + row::{Datum, GenericRow}, +}; +use iggy_connector_sdk::{ConsumedMessage, Error as IggyError}; +use thiserror::Error; + +use crate::{FlussSinkConfig, PayloadFormat}; + +const UNSIGNED_64_DECIMAL_PRECISION: u8 = 20; +const TIMESTAMP_PRECISION: u32 = 6; + +#[derive(Debug, Error)] +pub(crate) enum Error { + #[error(transparent)] + Fluss(Box), + #[error( + "Failed to convert Iggy message ID {message_id} field '{field}' with value {value} to Fluss DECIMAL(20, 0): {source}" + )] + DecimalConversion { + message_id: u128, + field: &'static str, + value: i128, + #[source] + source: Box, + }, + #[error( + "Failed to convert Iggy message ID {message_id} field '{field}' with value {value} to Fluss TIMESTAMP_LTZ(6): {reason}" + )] + TimestampConversion { + message_id: u128, + field: &'static str, + value: i128, + reason: String, + }, + #[error( + "Payload from Iggy message ID {message_id} is not valid UTF-8 for a Fluss STRING column: {source}" + )] + InvalidPayloadUtf8 { + message_id: u128, + #[source] + source: FromUtf8Error, + }, + #[error("Failed to convert payload from Iggy message ID {message_id} to bytes: {source}")] + PayloadBytesConversion { + message_id: u128, + #[source] + source: IggyError, + }, +} + +impl From for Error { + fn from(error: FlussError) -> Self { + Self::Fluss(Box::new(error)) + } +} + +impl From for IggyError { + fn from(error: Error) -> Self { + let message = error.to_string(); + match error { + Error::Fluss(_) => Self::CannotStoreData(message), + Error::InvalidPayloadUtf8 { .. } => Self::Serialization(message), + Error::DecimalConversion { .. } + | Error::TimestampConversion { .. } + | Error::PayloadBytesConversion { .. } => Self::InvalidRecordValue(message), + } + } +} + +enum EncodedPayload { + String(String), + Binary(Vec), +} + +pub struct EncodedMessage { + id: String, + offset: i128, + checksum: i128, + timestamp: i64, + origin_timestamp: i64, + payload: EncodedPayload, +} + +impl EncodedMessage { + fn try_into(message: ConsumedMessage, payload_format: PayloadFormat) -> Result { + let message_id = message.id; + Ok(Self { + id: string_from_id(message.id), + offset: i128::from(message.offset), + checksum: i128::from(message.checksum), + timestamp: i64::try_from(message.timestamp).map_err(|error| { + Error::TimestampConversion { + message_id, + field: "timestamp", + value: i128::from(message.timestamp), + reason: error.to_string(), + } + })?, + origin_timestamp: i64::try_from(message.origin_timestamp).map_err(|error| { + Error::TimestampConversion { + message_id, + field: "origin_timestamp", + value: i128::from(message.origin_timestamp), + reason: error.to_string(), + } + })?, + payload: match payload_format { + PayloadFormat::Bytea => { + let payload = message + .payload + .try_into_vec() + .map_err(|source| Error::PayloadBytesConversion { message_id, source })?; + EncodedPayload::Binary(payload) + } + PayloadFormat::Json | PayloadFormat::Text => { + let payload_binary = message + .payload + .try_into_vec() + .map_err(|source| Error::PayloadBytesConversion { message_id, source })?; + let payload_string = String::from_utf8(payload_binary) + .map_err(|source| Error::InvalidPayloadUtf8 { message_id, source })?; + EncodedPayload::String(payload_string) + } + }, + }) + } + pub fn encode_all( + messages: impl IntoIterator, + payload_format: PayloadFormat, + ) -> impl Iterator> { + messages + .into_iter() + .map(move |m| Self::try_into(m, payload_format)) + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ColumnKind { + MessageId(&'static str), + Checksum(&'static str), + MessageTimestamp(&'static str), + OriginTimestamp(&'static str), + MessageOffset(&'static str), + Stream(&'static str), + Topic(&'static str), + PartitionId(&'static str), +} + +impl From for Column { + fn from(column: ColumnKind) -> Self { + match column { + ColumnKind::MessageId(name) => Column::new(name, fluss::metadata::DataTypes::string()) + .with_comment("Apache Iggy message ID"), + ColumnKind::Checksum(name) => Column::new( + name, + fluss::metadata::DataTypes::decimal(UNSIGNED_64_DECIMAL_PRECISION as u32, 0), + ) + .with_comment("Apache Iggy message checksum"), + ColumnKind::MessageTimestamp(name) => Column::new( + name, + fluss::metadata::DataTypes::timestamp_ltz_with_precision(TIMESTAMP_PRECISION), + ) + .with_comment("Apache Iggy message timestamp"), + ColumnKind::OriginTimestamp(name) => Column::new( + name, + fluss::metadata::DataTypes::timestamp_ltz_with_precision(TIMESTAMP_PRECISION), + ) + .with_comment("Apache Iggy message origin timestamp"), + ColumnKind::MessageOffset(name) => Column::new( + name, + fluss::metadata::DataTypes::decimal(UNSIGNED_64_DECIMAL_PRECISION as u32, 0), + ) + .with_comment("Apache Iggy message offset"), + ColumnKind::Stream(name) => Column::new(name, fluss::metadata::DataTypes::string()) + .with_comment("Apache Iggy stream name"), + ColumnKind::Topic(name) => Column::new(name, fluss::metadata::DataTypes::string()) + .with_comment("Apache Iggy topic name"), + ColumnKind::PartitionId(name) => { + Column::new(name, fluss::metadata::DataTypes::bigint()) + .with_comment("Apache Iggy partition ID") + } + } + } +} + +#[derive(Debug)] +pub struct RowContext { + pub stream: String, + pub topic: String, + pub partition_id: u32, +} + +fn string_from_id(id: u128) -> String { + format!("{:032x}", id) +} + +fn timestamp_from_micros(epoch_micros: i64) -> Result { + let epoch_millis = epoch_micros / 1_000; + let nanos_of_millisecond = ((epoch_micros % 1_000) * 1_000) as i32; + fluss::row::TimestampLtz::from_millis_nanos(epoch_millis, nanos_of_millisecond) +} + +pub struct ArrowRowBuilder { + id_builder: StringBuilder, + checksum_builder: Decimal128Builder, + message_offset_builder: Decimal128Builder, + message_timestamp_builder: TimestampMicrosecondBuilder, + origin_timestamp_builder: TimestampMicrosecondBuilder, + payload_string_builder: StringBuilder, + payload_binary_builder: BinaryBuilder, + payload_format: PayloadFormat, + extra_columns: Vec, + context: RowContext, +} + +impl ArrowRowBuilder { + fn new( + payload_format: PayloadFormat, + extra_columns: Vec, + context: RowContext, + ) -> Self { + Self { + id_builder: StringBuilder::new(), + checksum_builder: Decimal128Builder::new() + .with_data_type(DataType::Decimal128(UNSIGNED_64_DECIMAL_PRECISION, 0)), + message_offset_builder: Decimal128Builder::new() + .with_data_type(DataType::Decimal128(UNSIGNED_64_DECIMAL_PRECISION, 0)), + message_timestamp_builder: TimestampMicrosecondBuilder::new().with_timezone("UTC"), + origin_timestamp_builder: TimestampMicrosecondBuilder::new().with_timezone("UTC"), + payload_binary_builder: BinaryBuilder::new(), + payload_string_builder: StringBuilder::new(), + payload_format, + extra_columns, + context, + } + } + + pub(crate) fn append(&mut self, message: EncodedMessage) { + let EncodedMessage { + id, + offset, + checksum, + timestamp, + origin_timestamp, + payload, + } = message; + + for column in &self.extra_columns { + match column { + ColumnKind::MessageId(_) => self.id_builder.append_value(&id), + ColumnKind::Checksum(_) => self.checksum_builder.append_value(checksum), + ColumnKind::MessageOffset(_) => self.message_offset_builder.append_value(offset), + ColumnKind::MessageTimestamp(_) => { + self.message_timestamp_builder.append_value(timestamp) + } + ColumnKind::OriginTimestamp(_) => { + self.origin_timestamp_builder.append_value(origin_timestamp) + } + ColumnKind::Stream(_) | ColumnKind::Topic(_) | ColumnKind::PartitionId(_) => {} + } + } + + match payload { + EncodedPayload::Binary(payload) => self.payload_binary_builder.append_value(payload), + EncodedPayload::String(payload) => self.payload_string_builder.append_value(payload), + } + } + + pub(crate) fn finish(&mut self) -> StructArray { + let mut cols: Vec<(Arc, ArrayRef)> = + Vec::with_capacity(self.extra_columns.len() + 1); + let len = match self.payload_format { + PayloadFormat::Bytea => self.payload_binary_builder.len(), + PayloadFormat::Json | PayloadFormat::Text => self.payload_string_builder.len(), + }; + + for column in &self.extra_columns { + let (name, array): (&str, ArrayRef) = match column { + ColumnKind::MessageId(name) => (name, Arc::new(self.id_builder.finish())), + ColumnKind::Checksum(name) => (name, Arc::new(self.checksum_builder.finish())), + ColumnKind::MessageOffset(name) => { + (name, Arc::new(self.message_offset_builder.finish())) + } + ColumnKind::MessageTimestamp(name) => { + (name, Arc::new(self.message_timestamp_builder.finish())) + } + ColumnKind::OriginTimestamp(name) => { + (name, Arc::new(self.origin_timestamp_builder.finish())) + } + ColumnKind::Stream(name) => ( + name, + Arc::new(StringArray::new_repeated(self.context.stream.as_str(), len)), + ), + ColumnKind::Topic(name) => ( + name, + Arc::new(StringArray::new_repeated(self.context.topic.as_str(), len)), + ), + ColumnKind::PartitionId(name) => ( + name, + Arc::new(Int64Array::from_value( + i64::from(self.context.partition_id), + len, + )), + ), + }; + let field = Arc::new(Field::new(name, array.data_type().clone(), false)); + cols.push((field, array)); + } + + let payload: ArrayRef = match self.payload_format { + PayloadFormat::Bytea => Arc::new(self.payload_binary_builder.finish()), + PayloadFormat::Json | PayloadFormat::Text => { + Arc::new(self.payload_string_builder.finish()) + } + }; + let payload_field = Arc::new(Field::new("payload", payload.data_type().clone(), true)); + cols.push((payload_field, payload)); + + StructArray::from(cols) + } +} + +#[derive(Debug)] +pub struct FlussTableLayout { + extra_columns: Vec, + payload_format: PayloadFormat, + primary_key_columns: Vec, +} + +impl FlussTableLayout { + pub fn from_config(config: &FlussSinkConfig) -> Self { + let mut columns: Vec = Vec::with_capacity(10); + columns.push(ColumnKind::MessageId("id")); + + if config.include_checksum { + columns.push(ColumnKind::Checksum("checksum")); + }; + + if config.include_metadata { + columns.extend([ + ColumnKind::MessageOffset("iggy_offset"), + ColumnKind::MessageTimestamp("iggy_timestamp"), + ColumnKind::Stream("iggy_stream"), + ColumnKind::Topic("iggy_topic"), + ColumnKind::PartitionId("iggy_partition_id"), + ]); + }; + + if config.include_origin_timestamp { + columns.push(ColumnKind::OriginTimestamp("iggy_origin_timestamp")); + } + + Self { + extra_columns: columns, + payload_format: config.payload_format, + primary_key_columns: Vec::new(), + } + } + + fn build_schema(&self) -> Result { + let payload_column = match self.payload_format { + PayloadFormat::Bytea => Column::new("payload", fluss::metadata::DataTypes::bytes()) + .with_comment("Apache Iggy message payload"), + + PayloadFormat::Json | PayloadFormat::Text => { + Column::new("payload", fluss::metadata::DataTypes::string()) + .with_comment("Apache Iggy message payload") + } + }; + + let columns: Vec = self + .extra_columns + .iter() + .copied() + .map(Into::into) + .chain(std::iter::once(payload_column)) + .collect(); + + let mut schema_builder = fluss::metadata::Schema::builder().with_columns(columns); + if !self.primary_key_columns.is_empty() { + schema_builder = schema_builder.primary_key(self.primary_key_columns.clone()); + } + Ok(schema_builder.build()?) + } + + pub(crate) fn build_table_descriptor(&self) -> Result { + let schema = self.build_schema()?; + Ok(TableDescriptor::builder() + .comment("Stores Apache Iggy messages written by the Fluss sink connector") + .schema(schema) + .build()?) + } + + pub(crate) fn convert_to_generic_row<'a>( + &self, + message: ConsumedMessage, + context: &'a RowContext, + ) -> Result, Error> { + let message_id = message.id; + let mut values: Vec = Vec::with_capacity(self.extra_columns.len()); + let encoded_message = EncodedMessage::try_into(message, self.payload_format)?; + for column in &self.extra_columns { + let value: Datum = match column { + ColumnKind::MessageId(_) => encoded_message.id.clone().into(), + ColumnKind::Checksum(_) => fluss::row::Decimal::from_arrow_decimal128( + encoded_message.checksum, + 0, + UNSIGNED_64_DECIMAL_PRECISION as u32, + 0, + ) + .map_err(|source| Error::DecimalConversion { + message_id, + field: "checksum", + value: encoded_message.checksum, + source: Box::new(source), + }) + .map(Into::into)?, + ColumnKind::MessageTimestamp(_) => timestamp_from_micros(encoded_message.timestamp) + .map_err(|source| Error::TimestampConversion { + message_id, + field: "timestamp", + value: i128::from(encoded_message.timestamp), + reason: source.to_string(), + }) + .map(Into::into)?, + + ColumnKind::OriginTimestamp(_) => { + timestamp_from_micros(encoded_message.origin_timestamp) + .map_err(|source| Error::TimestampConversion { + message_id, + field: "origin_timestamp", + value: i128::from(encoded_message.origin_timestamp), + reason: source.to_string(), + }) + .map(Into::into)? + } + ColumnKind::MessageOffset(_) => fluss::row::Decimal::from_arrow_decimal128( + encoded_message.offset, + 0, + UNSIGNED_64_DECIMAL_PRECISION as u32, + 0, + ) + .map_err(|source| Error::DecimalConversion { + message_id, + field: "offset", + value: encoded_message.offset, + source: Box::new(source), + }) + .map(Into::into)?, + ColumnKind::Stream(_) => context.stream.as_str().into(), + ColumnKind::Topic(_) => context.topic.as_str().into(), + ColumnKind::PartitionId(_) => i64::from(context.partition_id).into(), + }; + values.push(value); + } + + let payload = match encoded_message.payload { + EncodedPayload::String(s) => Datum::from(s), + EncodedPayload::Binary(b) => Datum::from(b), + }; + + values.push(payload); + + Ok(GenericRow::from_data(values)) + } + + pub(crate) fn create_arrow_builder(&self, context: RowContext) -> ArrowRowBuilder { + ArrowRowBuilder::new(self.payload_format, self.extra_columns.clone(), context) + } +} + +#[cfg(test)] +mod tests { + use arrow::array::{ + Array, ArrayRef, BinaryArray, Decimal128Array, Int64Array, StringArray, StructArray, + TimestampMicrosecondArray, + }; + use fluss::{ + metadata::{Column, DataTypes}, + row::Datum, + }; + use iggy_connector_sdk::{ConsumedMessage, Error, Payload, Schema}; + + use super::{ + ColumnKind, EncodedMessage, Error as SchemaError, FlussTableLayout, RowContext, + TIMESTAMP_PRECISION, UNSIGNED_64_DECIMAL_PRECISION, timestamp_from_micros, + }; + use crate::{FlussSinkConfig, PayloadFormat}; + + const MESSAGE_TIMESTAMP: i64 = 1_700_000_000_123_456; + const ORIGIN_TIMESTAMP: u64 = 1_700_000_000_120_789; + + fn test_config(payload_format: PayloadFormat) -> FlussSinkConfig { + FlussSinkConfig { + payload_format, + ..FlussSinkConfig::default() + } + } + + fn config_without_optional_columns(payload_format: PayloadFormat) -> FlussSinkConfig { + FlussSinkConfig { + include_checksum: false, + include_metadata: false, + include_origin_timestamp: false, + ..test_config(payload_format) + } + } + + fn test_message(payload: Payload) -> ConsumedMessage { + ConsumedMessage { + id: 101, + offset: 202, + checksum: 303, + timestamp: MESSAGE_TIMESTAMP as u64, + origin_timestamp: ORIGIN_TIMESTAMP, + headers: None, + payload, + } + } + + fn test_context() -> RowContext { + RowContext { + stream: "orders".to_string(), + topic: "created".to_string(), + partition_id: 7, + } + } + + fn arrow_column(columns: &[ArrayRef], index: usize) -> &T { + columns[index] + .as_any() + .downcast_ref::() + .expect("Arrow column should have the expected type") + } + + fn build_arrow_rows(layout: &FlussTableLayout, messages: Vec) -> StructArray { + let mut builder = layout.create_arrow_builder(test_context()); + for message in EncodedMessage::encode_all(messages, layout.payload_format) { + let message = message.expect("Message should encode"); + builder.append(message); + } + builder.finish() + } + + #[test] + fn given_microsecond_values_when_converting_to_timestamp_should_preserve_precision() { + let cases = [ + (0, 0, 0), + (999, 0, 999_000), + (1_000, 1, 0), + (1_001, 1, 1_000), + (MESSAGE_TIMESTAMP, 1_700_000_000_123, 456_000), + (i64::MAX, 9_223_372_036_854_775, 807_000), + ]; + + for (value, expected_millis, expected_nanos) in cases { + let timestamp = timestamp_from_micros(value) + .expect("microsecond value within the i64 range should convert"); + + assert_eq!(timestamp.get_epoch_millisecond(), expected_millis); + assert_eq!(timestamp.get_nano_of_millisecond(), expected_nanos); + } + } + + #[test] + fn given_i64_max_timestamps_when_encoding_message_should_preserve_values() { + let mut message = test_message(Payload::Raw(vec![1])); + message.timestamp = i64::MAX as u64; + message.origin_timestamp = i64::MAX as u64; + + let encoded_message = EncodedMessage::try_into(message, PayloadFormat::Json) + .expect("Timestamps within the i64 range should encode"); + + assert_eq!(encoded_message.timestamp, i64::MAX); + assert_eq!(encoded_message.origin_timestamp, i64::MAX); + } + + #[test] + fn given_all_extra_columns_when_converting_messages_should_preserve_array_contents() { + let layout = FlussTableLayout::from_config(&test_config(PayloadFormat::Text)); + let first_message = test_message(Payload::Text("first".to_string())); + let mut second_message = test_message(Payload::Text("second".to_string())); + second_message.id = 102; + second_message.checksum = 304; + second_message.offset = 203; + second_message.timestamp = MESSAGE_TIMESTAMP as u64 + 1; + second_message.origin_timestamp = ORIGIN_TIMESTAMP + 1; + + let rows = build_arrow_rows(&layout, vec![first_message, second_message]); + let columns = rows.columns(); + + assert_eq!(columns.len(), layout.extra_columns.len() + 1); + + let ids = arrow_column::(columns, 0); + assert_eq!( + ids.iter().collect::>(), + vec![ + Some("00000000000000000000000000000065"), + Some("00000000000000000000000000000066"), + ] + ); + + let checksums = arrow_column::(columns, 1); + assert_eq!( + checksums.iter().collect::>(), + vec![Some(303), Some(304)] + ); + + let offsets = arrow_column::(columns, 2); + assert_eq!( + offsets.iter().collect::>(), + vec![Some(202), Some(203)] + ); + + let timestamps = arrow_column::(columns, 3); + assert_eq!( + timestamps.iter().collect::>(), + vec![Some(MESSAGE_TIMESTAMP), Some(MESSAGE_TIMESTAMP + 1)] + ); + + let streams = arrow_column::(columns, 4); + assert_eq!( + streams.iter().collect::>(), + vec![Some("orders"), Some("orders")] + ); + + let topics = arrow_column::(columns, 5); + assert_eq!( + topics.iter().collect::>(), + vec![Some("created"), Some("created")] + ); + + let partition_ids = arrow_column::(columns, 6); + assert_eq!( + partition_ids.iter().collect::>(), + vec![Some(7), Some(7)] + ); + + let origin_timestamps = arrow_column::(columns, 7); + assert_eq!( + origin_timestamps.iter().collect::>(), + vec![ + Some(ORIGIN_TIMESTAMP as i64), + Some(ORIGIN_TIMESTAMP as i64 + 1), + ] + ); + + let payloads = arrow_column::(columns, 8); + assert_eq!( + payloads.iter().collect::>(), + vec![Some("first"), Some("second")] + ); + } + + #[test] + fn given_binary_messages_when_building_minimal_arrow_columns_should_preserve_payloads() { + let config = config_without_optional_columns(PayloadFormat::Bytea); + let layout = FlussTableLayout::from_config(&config); + let first_message = test_message(Payload::Raw(vec![0, 127, 255])); + let mut second_message = test_message(Payload::Raw(vec![1, 2, 3])); + second_message.id = 102; + + let rows = build_arrow_rows(&layout, vec![first_message, second_message]); + let columns = rows.columns(); + + assert_eq!(columns.len(), 2); + + let ids = arrow_column::(columns, 0); + assert_eq!(ids.value(0), "00000000000000000000000000000065"); + assert_eq!(ids.value(1), "00000000000000000000000000000066"); + + let payloads = arrow_column::(columns, 1); + assert_eq!(payloads.value(0), &[0, 127, 255]); + assert_eq!(payloads.value(1), &[1, 2, 3]); + } + + #[test] + fn given_invalid_utf8_before_valid_message_when_encoding_arrow_rows_should_skip_invalid_row() { + let config = config_without_optional_columns(PayloadFormat::Text); + let layout = FlussTableLayout::from_config(&config); + let mut builder = layout.create_arrow_builder(test_context()); + let mut valid_message = test_message(Payload::Text("valid".to_string())); + valid_message.id = 102; + let messages = [test_message(Payload::Raw(vec![0xff])), valid_message]; + let mut errors = 0; + for message in EncodedMessage::encode_all(messages, layout.payload_format) { + match message { + Ok(message) => builder.append(message), + Err(error) => { + assert!(matches!( + error, + SchemaError::InvalidPayloadUtf8 { + message_id: 101, + .. + } + )); + errors += 1; + } + } + } + + let rows = builder.finish(); + assert_eq!(errors, 1); + assert_eq!(rows.len(), 1); + let ids = arrow_column::(rows.columns(), 0); + assert_eq!(ids.value(0), "00000000000000000000000000000066"); + let payloads = arrow_column::(rows.columns(), 1); + assert_eq!(payloads.value(0), "valid"); + } + + #[test] + fn given_timestamp_above_i64_max_when_encoding_message_should_return_conversion_error() { + let mut message = test_message(Payload::Text("payload".to_string())); + message.timestamp = i64::MAX as u64 + 1; + + let error = EncodedMessage::try_into(message, PayloadFormat::Text) + .err() + .expect("Timestamp above the i64 range should fail"); + + assert!(matches!( + error, + SchemaError::TimestampConversion { + message_id: 101, + field: "timestamp", + value, + .. + } if value == i128::from(i64::MAX) + 1 + )); + } + + #[test] + fn given_default_config_when_building_layout_should_include_all_columns_in_order() { + let layout = FlussTableLayout::from_config(&FlussSinkConfig::default()); + + assert_eq!( + layout.extra_columns, + [ + ColumnKind::MessageId("id"), + ColumnKind::Checksum("checksum"), + ColumnKind::MessageOffset("iggy_offset"), + ColumnKind::MessageTimestamp("iggy_timestamp"), + ColumnKind::Stream("iggy_stream"), + ColumnKind::Topic("iggy_topic"), + ColumnKind::PartitionId("iggy_partition_id"), + ColumnKind::OriginTimestamp("iggy_origin_timestamp"), + ] + ); + } + + #[test] + fn given_optional_columns_disabled_when_building_layout_should_only_include_id_and_payload() { + let config = config_without_optional_columns(PayloadFormat::Bytea); + let layout = FlussTableLayout::from_config(&config); + + assert_eq!(layout.extra_columns, [ColumnKind::MessageId("id")]); + } + + #[test] + fn given_payload_formats_when_building_schema_should_use_matching_payload_types() { + for (payload_format, expected_data_type) in [ + (PayloadFormat::Bytea, DataTypes::bytes()), + (PayloadFormat::Json, DataTypes::string()), + (PayloadFormat::Text, DataTypes::string()), + ] { + let config = config_without_optional_columns(payload_format); + let layout = FlussTableLayout::from_config(&config); + let schema = layout.build_schema().expect("Schema should build"); + let payload_column = schema + .columns() + .last() + .expect("Schema should contain a payload column"); + + assert_eq!(payload_column.name(), "payload"); + assert_eq!(payload_column.data_type(), &expected_data_type); + } + } + + #[test] + fn given_default_layout_when_building_descriptor_should_include_schema_metadata() { + let layout = FlussTableLayout::from_config(&FlussSinkConfig::default()); + let descriptor = layout + .build_table_descriptor() + .expect("Table descriptor should build"); + + assert_eq!( + descriptor.schema().columns(), + [ + Column::new("id", DataTypes::string()) + .with_comment("Apache Iggy message ID") + .with_id(0), + Column::new( + "checksum", + DataTypes::decimal(UNSIGNED_64_DECIMAL_PRECISION as u32, 0), + ) + .with_comment("Apache Iggy message checksum") + .with_id(1), + Column::new( + "iggy_offset", + DataTypes::decimal(UNSIGNED_64_DECIMAL_PRECISION as u32, 0), + ) + .with_comment("Apache Iggy message offset") + .with_id(2), + Column::new( + "iggy_timestamp", + DataTypes::timestamp_ltz_with_precision(TIMESTAMP_PRECISION), + ) + .with_comment("Apache Iggy message timestamp") + .with_id(3), + Column::new("iggy_stream", DataTypes::string()) + .with_comment("Apache Iggy stream name") + .with_id(4), + Column::new("iggy_topic", DataTypes::string()) + .with_comment("Apache Iggy topic name") + .with_id(5), + Column::new("iggy_partition_id", DataTypes::bigint()) + .with_comment("Apache Iggy partition ID") + .with_id(6), + Column::new( + "iggy_origin_timestamp", + DataTypes::timestamp_ltz_with_precision(TIMESTAMP_PRECISION), + ) + .with_comment("Apache Iggy message origin timestamp") + .with_id(7), + Column::new("payload", DataTypes::string()) + .with_comment("Apache Iggy message payload") + .with_id(8) + ] + ); + assert_eq!( + descriptor.comment(), + Some("Stores Apache Iggy messages written by the Fluss sink connector") + ); + assert!(!descriptor.has_primary_key()); + } + + #[test] + fn given_primary_key_columns_when_building_schema_should_set_primary_key() { + let layout = FlussTableLayout { + extra_columns: vec![ColumnKind::MessageId("id")], + payload_format: PayloadFormat::Text, + primary_key_columns: vec!["id".to_string()], + }; + + let schema = layout.build_schema().expect("Schema should build"); + + assert_eq!(schema.primary_key_column_names(), ["id"]); + } + + #[test] + fn given_text_payload_when_building_row_should_store_payload_as_string() { + let config = config_without_optional_columns(PayloadFormat::Text); + let layout = FlussTableLayout::from_config(&config); + let message = test_message(Payload::Text("hello Fluss".to_string())); + + let context = test_context(); + let row = layout + .convert_to_generic_row(message, &context) + .expect("Text row should build"); + + assert_eq!( + row.values, + [ + Datum::from("00000000000000000000000000000065".to_string()), + Datum::from("hello Fluss".to_string()), + ] + ); + } + + #[test] + fn given_json_payload_when_building_row_should_serialize_payload_as_string() { + let config = config_without_optional_columns(PayloadFormat::Json); + let layout = FlussTableLayout::from_config(&config); + let payload = Schema::Json + .try_into_payload(br#"{"event":"created"}"#.to_vec()) + .expect("JSON payload should decode"); + let message = test_message(payload); + let context = test_context(); + let row = layout + .convert_to_generic_row(message, &context) + .expect("JSON row should build"); + + assert_eq!( + row.values, + [ + Datum::from("00000000000000000000000000000065".to_string()), + Datum::from(r#"{"event":"created"}"#.to_string()), + ] + ); + } + + #[test] + fn given_max_unsigned_values_when_building_row_should_preserve_id_offset_and_checksum() { + let layout = FlussTableLayout::from_config(&test_config(PayloadFormat::Bytea)); + let mut message = test_message(Payload::Raw(vec![1])); + let context = test_context(); + message.id = u128::MAX; + message.offset = u64::MAX; + message.checksum = u64::MAX; + + let row = layout + .convert_to_generic_row(message, &context) + .expect("Unsigned values should build"); + + assert_eq!(row.values[0].as_str(), "ffffffffffffffffffffffffffffffff"); + assert_eq!(row.values[1].as_decimal().to_string(), u64::MAX.to_string()); + assert_eq!(row.values[2].as_decimal().to_string(), u64::MAX.to_string()); + } + + #[test] + fn given_origin_timestamp_above_i64_max_when_encoding_message_should_return_conversion_error() { + let mut message = test_message(Payload::Raw(vec![1])); + message.origin_timestamp = i64::MAX as u64 + 1; + + let error = EncodedMessage::try_into(message, PayloadFormat::Bytea) + .err() + .expect("Origin timestamp above the i64 range should fail"); + + assert!(matches!( + error, + SchemaError::TimestampConversion { + message_id: 101, + field: "origin_timestamp", + value, + .. + } if value == i128::from(i64::MAX) + 1 + )); + } + + #[test] + fn given_invalid_utf8_when_building_string_row_should_return_serialization_error() { + let config = config_without_optional_columns(PayloadFormat::Text); + let layout = FlussTableLayout::from_config(&config); + let message = test_message(Payload::Raw(vec![0xFF])); + + let error = layout + .convert_to_generic_row(message, &test_context()) + .expect_err("Invalid UTF-8 payload should fail"); + + assert!(matches!( + &error, + SchemaError::InvalidPayloadUtf8 { + message_id: 101, + .. + } + )); + + let error: Error = error.into(); + + assert!(matches!(error, Error::Serialization(_))); + } +} diff --git a/core/connectors/sinks/fluss_sink/src/writer.rs b/core/connectors/sinks/fluss_sink/src/writer.rs new file mode 100644 index 0000000000..611d91f646 --- /dev/null +++ b/core/connectors/sinks/fluss_sink/src/writer.rs @@ -0,0 +1,419 @@ +// 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. + +use std::{ + fmt::{self, Display, Formatter}, + time::Duration, +}; + +use arrow::array::RecordBatch; +use fluss::{ + client::{AppendWriter, FlussConnection}, + error::Error as FlussError, + metadata::{TableDescriptor, TablePath}, +}; +use iggy_connector_sdk::{ + ConsumedMessage, Error as ConnectorError, MessagesMetadata, TopicMetadata, +}; +use thiserror::Error; +use tracing::error; + +use crate::{ + FlussSinkConfig, + schema::{EncodedMessage, Error, FlussTableLayout, RowContext}, +}; + +#[derive(Debug, Error)] +pub(crate) enum WriterError { + #[error(transparent)] + Connector(ConnectorError), + #[error(transparent)] + Schema(Error), + #[error("Fluss connection is not initialized")] + ConnectionNotInitialized, + #[error("Failed to connect to Fluss: {source}")] + Connect { + #[source] + source: Box, + }, + #[error("Invalid Fluss writer configuration: {source}")] + InvalidWriterConfig { + #[source] + source: Box, + }, + #[error("Failed to close Fluss connection: {source}")] + CloseConnection { + #[source] + source: Box, + }, + #[error("Failed to get Fluss admin client: {source}")] + GetAdminClient { + #[source] + source: Box, + }, + #[error("Failed to create Fluss table '{table_path}': {source}")] + CreateTable { + table_path: TablePath, + #[source] + source: Box, + }, + #[error("Failed to build Fluss table descriptor: {source}")] + BuildTableDescriptor { + #[source] + source: Box, + }, + #[error("Failed to get Fluss table '{table_path}': {source}")] + GetTable { + table_path: TablePath, + #[source] + source: Box, + }, + #[error("Failed to create appender for Fluss table '{table_path}': {source}")] + CreateAppender { + table_path: TablePath, + #[source] + source: Box, + }, + #[error("Failed to create writer for Fluss table '{table_path}': {source}")] + CreateWriter { + table_path: TablePath, + #[source] + source: Box, + }, + #[error("Failed to append Arrow batch to Fluss table '{table_path}': {source}")] + AppendArrowBatch { + table_path: TablePath, + #[source] + source: Box, + }, + #[error("Failed to flush rows to Fluss table '{table_path}': {source}")] + FlushRows { + table_path: TablePath, + #[source] + source: Box, + }, +} + +impl From for WriterError { + fn from(error: ConnectorError) -> Self { + Self::Connector(error) + } +} + +impl From for WriterError { + fn from(error: Error) -> Self { + Self::Schema(error) + } +} + +impl From for ConnectorError { + fn from(error: WriterError) -> Self { + let message = error.to_string(); + match error { + WriterError::Connector(source) => source, + WriterError::Schema(source) => source.into(), + WriterError::ConnectionNotInitialized | WriterError::Connect { .. } => { + Self::InitError(message) + } + WriterError::InvalidWriterConfig { .. } => Self::InvalidConfigValue(message), + WriterError::CloseConnection { .. } => Self::Connection(message), + WriterError::GetAdminClient { .. } + | WriterError::CreateTable { .. } + | WriterError::BuildTableDescriptor { .. } + | WriterError::GetTable { .. } + | WriterError::CreateAppender { .. } + | WriterError::CreateWriter { .. } + | WriterError::AppendArrowBatch { .. } + | WriterError::FlushRows { .. } => Self::CannotStoreData(message), + } + } +} + +#[derive(Default)] +pub struct TableWriteResult { + pub insertion_errors: u64, + pub messages_processed: u64, +} + +pub struct FlussWriter { + connection: Option, + config: FlussSinkConfig, +} + +impl Display for FlussWriter { + fn fmt(&self, formatter: &mut Formatter) -> std::fmt::Result { + write!(formatter, "FlussWriter") + } +} + +impl fmt::Debug for FlussWriter { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("FlussWriter") + .finish_non_exhaustive() + } +} + +impl FlussWriter { + pub fn new(config: FlussSinkConfig) -> Self { + Self { + config, + connection: None, + } + } + + pub async fn connect(&mut self) -> Result<(), WriterError> { + let config = fluss::config::Config::try_from(&self.config)?; + let connection = + FlussConnection::new(config) + .await + .map_err(|source| WriterError::Connect { + source: Box::new(source), + })?; + connection.get_or_create_writer_client().map_err(|source| { + WriterError::InvalidWriterConfig { + source: Box::new(source), + } + })?; + self.connection = Some(connection); + Ok(()) + } + + pub async fn close(&self) -> Result<(), WriterError> { + self.get_connection()? + .get_or_create_writer_client() + .map_err(|source| WriterError::InvalidWriterConfig { + source: Box::new(source), + })? + .close(Duration::from_secs(30)) + .await + .map_err(|source| WriterError::CloseConnection { + source: Box::new(source), + })?; + Ok(()) + } + + pub async fn ensure_table_exists( + &self, + table_path: &TablePath, + table_layout: &FlussTableLayout, + ) -> Result<(), WriterError> { + if self.config.auto_create_table { + let table_descriptor = table_layout.build_table_descriptor().map_err(|source| { + WriterError::BuildTableDescriptor { + source: Box::new(source), + } + })?; + + self.create_table_if_not_exists(table_path, &table_descriptor) + .await?; + } + Ok(()) + } + + pub async fn write_to_table_arrow( + &self, + table_path: &TablePath, + messages_metadata: MessagesMetadata, + messages: Vec, + topic_metadata: &TopicMetadata, + table_layout: &FlussTableLayout, + ) -> Result { + let mut result = TableWriteResult::default(); + let context = RowContext { + topic: topic_metadata.topic.clone(), + stream: topic_metadata.stream.clone(), + partition_id: messages_metadata.partition_id, + }; + + let writer = self.create_writer(table_path).await?; + let mut builder = table_layout.create_arrow_builder(context); + for m in EncodedMessage::encode_all(messages, self.config.payload_format) { + let message = match m { + Ok(m) => m, + Err(err) => { + error!( + "Can not convert iggy message to row, skipping message because of error: [{}]", + err + ); + result.insertion_errors += 1; + continue; + } + }; + builder.append(message); + result.messages_processed += 1; + } + writer + .append_arrow_batch(RecordBatch::from(builder.finish())) + .map_err(|source| WriterError::AppendArrowBatch { + table_path: table_path.clone(), + source: Box::new(source), + })?; + self.flush(&writer, table_path).await?; + + Ok(result) + } + + pub async fn write_to_table( + &self, + table_path: &TablePath, + messages_metadata: MessagesMetadata, + messages: Vec, + topic_metadata: &TopicMetadata, + table_layout: &FlussTableLayout, + ) -> Result { + let mut stat = TableWriteResult::default(); + let writer = self.create_writer(table_path).await?; + let context = RowContext { + topic: topic_metadata.topic.clone(), + stream: topic_metadata.stream.clone(), + partition_id: messages_metadata.partition_id, + }; + + for message in messages { + let message_id = message.id; + let row = match table_layout.convert_to_generic_row(message, &context) { + Ok(row) => row, + Err(e) => { + error!( + "Can not convert iggy message to row, skipping message because of error: [{}]", + e + ); + stat.insertion_errors += 1; + continue; + } + }; + match writer.append(&row) { + Ok(_) => stat.messages_processed += 1, + Err(e) => { + error!( + "Can not append row, skipping message [{}] because of fluss error: [{}]", + message_id, e + ); + stat.insertion_errors += 1; + } + }; + } + self.flush(&writer, table_path).await?; + Ok(stat) + } + + fn get_connection(&self) -> Result<&FlussConnection, WriterError> { + self.connection + .as_ref() + .ok_or(WriterError::ConnectionNotInitialized) + } + + async fn create_table_if_not_exists( + &self, + table_path: &TablePath, + table_descriptor: &TableDescriptor, + ) -> Result<(), WriterError> { + self.get_connection()? + .get_admin() + .map_err(|source| WriterError::GetAdminClient { + source: Box::new(source), + })? + .create_table(table_path, table_descriptor, true) + .await + .map_err(|source| WriterError::CreateTable { + table_path: table_path.clone(), + source: Box::new(source), + }) + } + + async fn create_writer(&self, table_path: &TablePath) -> Result { + let table = self + .get_connection()? + .get_table(table_path) + .await + .map_err(|source| WriterError::GetTable { + table_path: table_path.clone(), + source: Box::new(source), + })?; + + table + .new_append() + .map_err(|source| WriterError::CreateAppender { + table_path: table_path.clone(), + source: Box::new(source), + })? + .create_writer() + .map_err(|source| WriterError::CreateWriter { + table_path: table_path.clone(), + source: Box::new(source), + }) + } + + async fn flush( + &self, + writer: &AppendWriter, + table_path: &TablePath, + ) -> Result<(), WriterError> { + writer + .flush() + .await + .map_err(|source| WriterError::FlushRows { + table_path: table_path.clone(), + source: Box::new(source), + }) + } +} + +#[cfg(test)] +mod tests { + use fluss::{error::Error as FlussError, metadata::TablePath}; + use iggy_connector_sdk::Error as ConnectorError; + + use super::WriterError; + + #[test] + fn given_connector_error_when_converting_should_preserve_original_variant() { + let expected = ConnectorError::InvalidConfigValue("invalid value".to_string()); + + let actual: ConnectorError = WriterError::Connector(expected.clone()).into(); + + assert_eq!(actual, expected); + } + + #[test] + fn given_missing_connection_when_converting_should_return_init_error() { + let error: ConnectorError = WriterError::ConnectionNotInitialized.into(); + + assert_eq!( + error, + ConnectorError::InitError("Fluss connection is not initialized".to_string()) + ); + } + + #[test] + fn given_arrow_append_failure_when_converting_should_return_cannot_store_data() { + let error: ConnectorError = WriterError::AppendArrowBatch { + table_path: TablePath::new("fluss", "iggy_messages"), + source: Box::new(FlussError::WriterClosed { + message: "writer closed".to_string(), + }), + } + .into(); + + assert!(matches!( + error, + ConnectorError::CannotStoreData(message) + if message.contains("fluss.iggy_messages") && message.contains("writer closed") + )); + } +} diff --git a/core/integration/Cargo.toml b/core/integration/Cargo.toml index a487e3d043..a367286756 100644 --- a/core/integration/Cargo.toml +++ b/core/integration/Cargo.toml @@ -49,6 +49,7 @@ ctor = { workspace = true } deltalake = { workspace = true } dtor = { workspace = true } figment = { workspace = true } +fluss-rs = { workspace = true } futures = { workspace = true } harness_derive = { workspace = true } humantime = { workspace = true } diff --git a/core/integration/tests/connectors/fixtures/fluss/cluster.rs b/core/integration/tests/connectors/fixtures/fluss/cluster.rs new file mode 100644 index 0000000000..75ecaca8de --- /dev/null +++ b/core/integration/tests/connectors/fixtures/fluss/cluster.rs @@ -0,0 +1,228 @@ +// 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. + +use std::fmt::{Display, Formatter}; +use std::net::TcpListener; + +use fluss::client::FlussConnection; +use integration::harness::TestBinaryError; +use testcontainers_modules::testcontainers::core::{IntoContainerPort, WaitFor}; +use testcontainers_modules::testcontainers::runners::AsyncRunner; +use testcontainers_modules::testcontainers::{ContainerAsync, GenericImage, ImageExt}; + +use crate::connectors::fixtures; + +const FLUSS_IMAGE: &str = "apache/fluss"; +const ZOOKEEPER_IMAGE: &str = "zookeeper"; +const ZOOKEEPER_VERSION: &str = "3.9.2"; +const ZOOKEEPER_PORT: u16 = 2181; +const FLUSS_CLIENT_PORT: u16 = 9123; +const CONNECTION_RETRY: u16 = 3; +const CONNECTION_RETRY_DELAY_S: u64 = 5; + +struct CoordinatorProperties { + zookeeper_address: String, + container_name: String, + advertised_port: u16, +} + +struct TabletServerProperties { + zookeeper_address: String, + container_name: String, + advertised_port: u16, + tablet_server_id: u32, +} + +impl Display for CoordinatorProperties { + fn fmt(&self, formatter: &mut Formatter) -> std::fmt::Result { + write!( + formatter, + "zookeeper.address: {}\n\ + bind.listeners: INTERNAL://{}:0, CLIENT://{}:{}\n\ + advertised.listeners: CLIENT://localhost:{}\n\ + internal.listener.name: INTERNAL\n\ + remote.data.dir: /tmp/fluss/remote-data", + self.zookeeper_address, + self.container_name, + self.container_name, + FLUSS_CLIENT_PORT, + self.advertised_port, + ) + } +} + +impl Display for TabletServerProperties { + fn fmt(&self, formatter: &mut Formatter) -> std::fmt::Result { + write!( + formatter, + "zookeeper.address: {}\n\ + bind.listeners: INTERNAL://{}:0, CLIENT://{}:{}\n\ + advertised.listeners: CLIENT://localhost:{}\n\ + internal.listener.name: INTERNAL\n\ + tablet-server.id: {}\n\ + kv.snapshot.interval: 0s\n\ + data.dir: /tmp/fluss/data/tablet-server-{}\n\ + remote.data.dir: /tmp/fluss/remote-data", + self.zookeeper_address, + self.container_name, + self.container_name, + FLUSS_CLIENT_PORT, + self.advertised_port, + self.tablet_server_id, + self.tablet_server_id, + ) + } +} + +pub struct FlussCluster { + #[allow(dead_code)] + zookeeper: ContainerAsync, + #[allow(dead_code)] + coordinator_server: ContainerAsync, + #[allow(dead_code)] + tablet_server: ContainerAsync, + pub coordinator_address: String, + #[allow(dead_code)] + pub fluss_version: String, +} + +impl FlussCluster { + pub async fn new(fluss_version: &str) -> Result { + Self::start(fluss_version).await + } + + pub async fn get_connection(&self) -> Result { + let config = fluss::config::Config { + bootstrap_servers: self.coordinator_address.clone(), + ..fluss::config::Config::default() + }; + + FlussConnection::new(config).await.map_err(|error| { + super::fixture_error(format!("Failed to create Fluss connection: {error}")) + }) + } + + async fn wait_for_fluss_to_become_healthy(&self) -> Result<(), TestBinaryError> { + let mut attempts = 0; + loop { + match self.get_connection().await { + Ok(_) => return Ok(()), + Err(error) => { + attempts += 1; + if attempts >= CONNECTION_RETRY { + return Err(super::fixture_error(format!( + "Failed to establish Fluss connection after {} attempts: {}", + CONNECTION_RETRY, error + ))); + } + tokio::time::sleep(std::time::Duration::from_secs(CONNECTION_RETRY_DELAY_S)) + .await; + } + } + } + } + + async fn start(fluss_version: &str) -> Result { + let network = fixtures::unique_container_name("fluss-network"); + let zookeeper_name = fixtures::unique_container_name("fluss-zookeeper"); + let coordinator_name = fixtures::unique_container_name("fluss-coordinator"); + let tablet_name = fixtures::unique_container_name("fluss-tablet-0"); + let coordinator_host_port = available_host_port()?; + let tablet_host_port = available_host_port_except(coordinator_host_port)?; + + let zookeeper = GenericImage::new(ZOOKEEPER_IMAGE, ZOOKEEPER_VERSION) + .with_exposed_port(ZOOKEEPER_PORT.tcp()) + .with_wait_for(WaitFor::message_on_stdout("Started AdminServer")) + .with_network(&network) + .with_container_name(&zookeeper_name) + .start() + .await + .map_err(|error| super::fixture_error(format!("Failed to start ZooKeeper: {error}")))?; + + let zookeeper_address = format!("{zookeeper_name}:{ZOOKEEPER_PORT}"); + let coordinator_properties = CoordinatorProperties { + zookeeper_address: zookeeper_address.clone(), + container_name: coordinator_name.clone(), + advertised_port: coordinator_host_port, + }; + let coordinator_server = GenericImage::new(FLUSS_IMAGE, fluss_version) + .with_exposed_port(FLUSS_CLIENT_PORT.tcp()) + .with_wait_for(WaitFor::Nothing) + .with_network(&network) + .with_container_name(&coordinator_name) + .with_env_var("FLUSS_PROPERTIES", coordinator_properties.to_string()) + .with_cmd(["coordinatorServer"]) + .with_mapped_port(coordinator_host_port, FLUSS_CLIENT_PORT.tcp()) + .start() + .await + .map_err(|error| { + super::fixture_error(format!("Failed to start Fluss coordinator server: {error}")) + })?; + + let tablet_properties = TabletServerProperties { + zookeeper_address: zookeeper_address.clone(), + container_name: tablet_name.clone(), + advertised_port: tablet_host_port, + tablet_server_id: 0, + }; + let tablet_server = GenericImage::new(FLUSS_IMAGE, fluss_version) + .with_exposed_port(FLUSS_CLIENT_PORT.tcp()) + .with_wait_for(WaitFor::Nothing) + .with_network(&network) + .with_container_name(&tablet_name) + .with_env_var("FLUSS_PROPERTIES", tablet_properties.to_string()) + .with_cmd(["tabletServer"]) + .with_mapped_port(tablet_host_port, FLUSS_CLIENT_PORT.tcp()) + .start() + .await + .map_err(|error| { + super::fixture_error(format!("Failed to start Fluss tablet server: {error}")) + })?; + + let result = Self { + fluss_version: fluss_version.to_string(), + zookeeper, + coordinator_server, + tablet_server, + coordinator_address: format!("localhost:{coordinator_host_port}"), + }; + + result.wait_for_fluss_to_become_healthy().await?; + + Ok(result) + } +} + +fn available_host_port() -> Result { + let listener = TcpListener::bind(("127.0.0.1", 0)) + .map_err(|error| super::fixture_error(format!("Failed to reserve a host port: {error}")))?; + listener + .local_addr() + .map(|address| address.port()) + .map_err(|error| { + super::fixture_error(format!("Failed to read the reserved host port: {error}")) + }) +} + +fn available_host_port_except(excluded_port: u16) -> Result { + loop { + let host_port = available_host_port()?; + if host_port != excluded_port { + return Ok(host_port); + } + } +} diff --git a/core/integration/tests/connectors/fixtures/fluss/mod.rs b/core/integration/tests/connectors/fixtures/fluss/mod.rs new file mode 100644 index 0000000000..16643c2303 --- /dev/null +++ b/core/integration/tests/connectors/fixtures/fluss/mod.rs @@ -0,0 +1,29 @@ +// 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. + +mod cluster; +mod sink; + +use integration::harness::TestBinaryError; +pub use sink::FlussSinkFixture; + +fn fixture_error(message: String) -> TestBinaryError { + TestBinaryError::FixtureSetup { + fixture_type: "FlussCluster".to_string(), + message, + } +} diff --git a/core/integration/tests/connectors/fixtures/fluss/sink.rs b/core/integration/tests/connectors/fixtures/fluss/sink.rs new file mode 100644 index 0000000000..16ad434114 --- /dev/null +++ b/core/integration/tests/connectors/fixtures/fluss/sink.rs @@ -0,0 +1,232 @@ +// 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. + +use std::{collections::HashMap, time::Duration}; + +use crate::connectors::fixtures::fluss::fixture_error; + +use super::cluster::FlussCluster; +use async_trait::async_trait; +use fluss::record::ScanBatch; +use fluss::row::ColumnarRow; +use fluss::{ + client::{EARLIEST_OFFSET, FlussConnection}, + metadata::{TableInfo, TablePath}, +}; +use integration::harness::{TestBinaryError, TestFixture, seeds}; +use tokio::time::{Instant, timeout_at}; + +const DEFAULT_FLUSS_VERSION: &str = "0.9.1-incubating"; +const DEFAULT_SINK_DB: &str = "fluss"; +const DEFAULT_SINK_TABLE: &str = "iggy_messages"; + +const ENV_SINK_BOOTSTRAP_SERVERS: &str = + "IGGY_CONNECTORS_SINK_FLUSS_PLUGIN_CONFIG_BOOTSTRAP_SERVERS"; +const ENV_SINK_TARGET_TABLE: &str = "IGGY_CONNECTORS_SINK_FLUSS_PLUGIN_CONFIG_TARGET_TABLE"; +const ENV_SINK_STREAMS_0_STREAM: &str = "IGGY_CONNECTORS_SINK_FLUSS_STREAMS_0_STREAM"; +const ENV_SINK_STREAMS_0_TOPICS: &str = "IGGY_CONNECTORS_SINK_FLUSS_STREAMS_0_TOPICS"; +const ENV_SINK_STREAMS_0_SCHEMA: &str = "IGGY_CONNECTORS_SINK_FLUSS_STREAMS_0_SCHEMA"; +const ENV_SINK_STREAMS_0_CONSUMER_GROUP: &str = + "IGGY_CONNECTORS_SINK_FLUSS_STREAMS_0_CONSUMER_GROUP"; +const ENV_SINK_PATH: &str = "IGGY_CONNECTORS_SINK_FLUSS_PATH"; + +fn create_test_table_path() -> TablePath { + TablePath::new(DEFAULT_SINK_DB, DEFAULT_SINK_TABLE) +} + +pub struct FlussSinkFixture { + cluster: FlussCluster, +} + +impl FlussSinkFixture { + pub async fn get_fluss_connection(&self) -> Result { + self.cluster.get_connection().await + } + + pub async fn get_test_table(&self) -> Result { + let connection = self.get_fluss_connection().await?; + let admin = connection + .get_admin() + .map_err(|error| fixture_error(format!("Failed to get Fluss admin: {error}")))?; + + admin + .get_table_info(&create_test_table_path()) + .await + .map_err(|error| fixture_error(format!("Failed to get Fluss test table: {error}"))) + } + + pub async fn read_from_test_table( + &self, + timeout: u64, + ) -> Result, TestBinaryError> { + let connection = self.get_fluss_connection().await?; + let table_path = create_test_table_path(); + let table = connection + .get_table(&table_path) + .await + .map_err(|e| fixture_error(format!("Failed to get table: {}", e)))?; + + let log_scanner = table + .new_scan() + .create_log_scanner() + .map_err(|e| fixture_error(format!("Failed to create log scanner: {}", e)))?; + + log_scanner + .subscribe(0, EARLIEST_OFFSET) + .await + .map_err(|e| fixture_error(format!("Failed to subscribe to log scanner: {}", e)))?; + + let deadline = Instant::now() + Duration::from_secs(timeout); + + let mut rows: Vec = Vec::new(); + + loop { + let records = match timeout_at(deadline, log_scanner.poll(Duration::from_secs(5))).await + { + Ok(Err(e)) => { + return Err(fixture_error(format!("Failed to poll log scanner: {}", e))); + } + Ok(Ok(records)) => records, + Err(_) => break, + }; + + for record in records { + rows.push(record.row); + } + } + + Ok(rows) + } + + pub async fn read_from_test_table_arrow_batch( + &self, + timeout: u64, + ) -> Result, TestBinaryError> { + let connection = self.get_fluss_connection().await?; + let table_path = create_test_table_path(); + let table = connection + .get_table(&table_path) + .await + .map_err(|e| fixture_error(format!("Failed to get table: {}", e)))?; + + let log_scanner = table + .new_scan() + .create_record_batch_log_scanner() + .map_err(|e| fixture_error(format!("Failed to create log scanner: {}", e)))?; + + log_scanner + .subscribe(0, EARLIEST_OFFSET) + .await + .map_err(|e| fixture_error(format!("Failed to subscribe to log scanner: {}", e)))?; + + let deadline = Instant::now() + Duration::from_secs(timeout); + + let mut rows: Vec = Vec::new(); + + loop { + let records = match timeout_at(deadline, log_scanner.poll(Duration::from_secs(5))).await + { + Ok(Err(e)) => { + return Err(fixture_error(format!("Failed to poll log scanner: {}", e))); + } + Ok(Ok(records)) => records, + Err(_) => break, + }; + + rows.extend(records); + } + + Ok(rows) + } + + pub async fn check_if_test_table_exists(&self) -> Result { + let connection = self.get_fluss_connection().await?; + let admin = connection + .get_admin() + .map_err(|e| fixture_error(format!("Error getting Fluss admin instance {}", e)))?; + let exists = admin + .table_exists(&create_test_table_path()) + .await + .map_err(|e| fixture_error(format!("Error checking if table exists {}", e)))?; + Ok(exists) + } + + pub async fn wait_for_test_table(&self, timeout: u64) -> Result<(), TestBinaryError> { + let deadline = Instant::now() + Duration::from_secs(timeout); + loop { + let timeout_at = timeout_at(deadline, self.check_if_test_table_exists()).await; + match timeout_at { + Ok(Ok(false)) => {} + Ok(Ok(true)) => return Ok(()), + Ok(Err(e)) => { + return Err(fixture_error( + format!("Checking the table has failed with: {}", e).to_string(), + )); + } + Err(_) => { + return Err(fixture_error( + format!( + "Test table was not created within the timeout of {}s", + timeout + ) + .to_string(), + )); + } + }; + + tokio::time::sleep(Duration::from_millis(50)).await; + } + } +} + +#[async_trait] +impl TestFixture for FlussSinkFixture { + async fn setup() -> Result { + let cluster = FlussCluster::new(DEFAULT_FLUSS_VERSION).await?; + Ok(Self { cluster }) + } + + fn connectors_runtime_envs(&self) -> HashMap { + HashMap::from([ + ( + ENV_SINK_BOOTSTRAP_SERVERS.to_string(), + self.cluster.coordinator_address.clone(), + ), + ( + ENV_SINK_TARGET_TABLE.to_string(), + DEFAULT_SINK_TABLE.to_string(), + ), + ( + ENV_SINK_STREAMS_0_STREAM.to_string(), + seeds::names::STREAM.to_string(), + ), + ( + ENV_SINK_STREAMS_0_TOPICS.to_string(), + format!("[{}]", seeds::names::TOPIC), + ), + (ENV_SINK_STREAMS_0_SCHEMA.to_string(), "json".to_string()), + ( + ENV_SINK_STREAMS_0_CONSUMER_GROUP.to_string(), + seeds::names::CONSUMER_GROUP.to_string(), + ), + ( + ENV_SINK_PATH.to_string(), + "../../target/debug/libiggy_connector_fluss_sink".to_string(), + ), + ]) + } +} diff --git a/core/integration/tests/connectors/fixtures/mod.rs b/core/integration/tests/connectors/fixtures/mod.rs index 5750cb9a92..4e2fb115e4 100644 --- a/core/integration/tests/connectors/fixtures/mod.rs +++ b/core/integration/tests/connectors/fixtures/mod.rs @@ -21,6 +21,7 @@ mod clickhouse; mod delta; mod doris; mod elasticsearch; +mod fluss; mod http; mod iceberg; mod influxdb; @@ -57,6 +58,7 @@ pub use doris::{ DorisSinkMaxFilterRatioFixture, DorisSinkPreCreatedFixture, }; pub use elasticsearch::{ElasticsearchSinkFixture, ElasticsearchSourcePreCreatedFixture}; +pub use fluss::FlussSinkFixture; pub use http::{ HttpSinkIndividualFixture, HttpSinkJsonArrayFixture, HttpSinkMultiTopicFixture, HttpSinkNdjsonFixture, HttpSinkNoMetadataFixture, HttpSinkRawFixture, diff --git a/core/integration/tests/connectors/fluss/fluss_sink.rs b/core/integration/tests/connectors/fluss/fluss_sink.rs new file mode 100644 index 0000000000..451193ed42 --- /dev/null +++ b/core/integration/tests/connectors/fluss/fluss_sink.rs @@ -0,0 +1,367 @@ +// 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. + +use crate::connectors::fixtures::FlussSinkFixture; +use crate::connectors::{TestMessage, create_test_messages}; +use bytes::Bytes; +use fluss::metadata::{Column, DataTypes, Schema}; +use fluss::row::{DataGetters, Decimal, TimestampLtz}; +use iggy::prelude::{Consumer, IggyMessage, Partitioning, PollingStrategy}; +use iggy_common::Identifier; +use iggy_common::MessageClient; +use integration::harness::seeds; +use integration::iggy_harness; + +const TEST_MESSAGE_COUNT: usize = 10; +const ROW_COMPARISON_MESSAGE_COUNT: u32 = 3; +const ID_COLUMN_INDEX: usize = 0; +const CHECKSUM_COLUMN_INDEX: usize = 1; +const IGGY_OFFSET_COLUMN_INDEX: usize = 2; +const IGGY_TIMESTAMP_COLUMN_INDEX: usize = 3; +const IGGY_STREAM_COLUMN_INDEX: usize = 4; +const IGGY_TOPIC_COLUMN_INDEX: usize = 5; +const IGGY_PARTITION_ID_COLUMN_INDEX: usize = 6; +const IGGY_ORIGIN_TIMESTAMP_COLUMN_INDEX: usize = 7; +const PAYLOAD_COLUMN_INDEX: usize = 8; +const UNSIGNED_64_DECIMAL_PRECISION: u32 = 20; +const TIMESTAMP_PRECISION: u32 = 6; +const WAIT_TIMEOUT_S: u64 = 10; + +fn expected_decimal(value: u64) -> Decimal { + Decimal::from_arrow_decimal128(i128::from(value), 0, UNSIGNED_64_DECIMAL_PRECISION, 0) + .expect("Unsigned 64-bit value should fit in DECIMAL(20, 0)") +} + +fn expected_timestamp(value: u64) -> TimestampLtz { + let epoch_micros = i64::try_from(value).expect("Test timestamp should fit in i64"); + TimestampLtz::from_millis_nanos( + epoch_micros / 1_000, + ((epoch_micros % 1_000) * 1_000) as i32, + ) + .expect("Test timestamp should convert to Fluss TIMESTAMP_LTZ(6)") +} + +fn expected_sink_schema() -> Schema { + Schema::builder() + .with_columns(vec![ + Column::new("id", DataTypes::string()).with_comment("Apache Iggy message ID"), + Column::new("checksum", DataTypes::decimal(20, 0)) + .with_comment("Apache Iggy message checksum"), + Column::new("iggy_offset", DataTypes::decimal(20, 0)) + .with_comment("Apache Iggy message offset"), + Column::new("iggy_timestamp", DataTypes::timestamp_ltz_with_precision(6)) + .with_comment("Apache Iggy message timestamp"), + Column::new("iggy_stream", DataTypes::string()).with_comment("Apache Iggy stream name"), + Column::new("iggy_topic", DataTypes::string()).with_comment("Apache Iggy topic name"), + Column::new("iggy_partition_id", DataTypes::bigint()) + .with_comment("Apache Iggy partition ID"), + Column::new( + "iggy_origin_timestamp", + DataTypes::timestamp_ltz_with_precision(6), + ) + .with_comment("Apache Iggy message origin timestamp"), + Column::new("payload", DataTypes::string()).with_comment("Apache Iggy message payload"), + ]) + .build() + .expect("Expected Fluss sink schema should build") +} + +#[iggy_harness( + server(connectors_runtime(config_path = "tests/connectors/fluss/sink.toml")), + seed = seeds::connector_stream +)] +async fn sink_should_create_test_table_with_expected_schema( + harness: &TestHarness, + fixture: FlussSinkFixture, +) { + let client = harness + .root_client() + .await + .expect("Root client should be available"); + + let stream_id: Identifier = seeds::names::STREAM + .try_into() + .expect("Stream identifier should be valid"); + let topic_id: Identifier = seeds::names::TOPIC + .try_into() + .expect("Topic identifier should be valid"); + + let messages_data = create_test_messages(1); + let mut messages: Vec = messages_data + .iter() + .enumerate() + .map(|(i, msg)| { + let payload = serde_json::to_vec(msg).expect("Failed to serialize message"); + IggyMessage::builder() + .id((i + 1) as u128) + .payload(Bytes::from(payload)) + .build() + .expect("Failed to build message") + }) + .collect(); + + client + .send_messages( + &stream_id, + &topic_id, + &Partitioning::partition_id(0), + &mut messages, + ) + .await + .expect("Failed to send messages"); + + fixture + .wait_for_test_table(WAIT_TIMEOUT_S) + .await + .expect("Fluss test table should be created"); + let table = fixture + .get_test_table() + .await + .expect("Fluss test table should be available"); + + assert_eq!(table.schema, expected_sink_schema()); +} + +#[iggy_harness( + server(connectors_runtime(config_path = "tests/connectors/fluss/sink.toml")), + seed = seeds::connector_stream +)] +async fn sink_should_write_message_to_test_table(harness: &TestHarness, fixture: FlussSinkFixture) { + let client = harness.root_client().await.unwrap(); + + let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap(); + let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap(); + + let messages_data = create_test_messages(TEST_MESSAGE_COUNT); + let mut messages: Vec = messages_data + .iter() + .enumerate() + .map(|(i, msg)| { + let payload = serde_json::to_vec(msg).expect("Failed to serialize message"); + IggyMessage::builder() + .id((i + 1) as u128) + .payload(Bytes::from(payload)) + .build() + .expect("Failed to build message") + }) + .collect(); + + client + .send_messages( + &stream_id, + &topic_id, + &Partitioning::partition_id(0), + &mut messages, + ) + .await + .expect("Failed to send messages"); + + fixture + .wait_for_test_table(WAIT_TIMEOUT_S) + .await + .expect("Table has not found in time"); + + let messages = fixture + .read_from_test_table(WAIT_TIMEOUT_S) + .await + .expect("read messages"); + + assert_eq!(messages.len(), TEST_MESSAGE_COUNT); +} + +#[iggy_harness( + server(connectors_runtime(config_path = "tests/connectors/fluss/sink.toml")), + seed = seeds::connector_stream +)] +async fn sink_should_preserve_extra_fields_and_payload_in_rows( + harness: &TestHarness, + fixture: FlussSinkFixture, +) { + let client = harness + .root_client() + .await + .expect("Root client should be available"); + + let stream_id: Identifier = seeds::names::STREAM + .try_into() + .expect("Stream identifier should be valid"); + let topic_id: Identifier = seeds::names::TOPIC + .try_into() + .expect("Topic identifier should be valid"); + + let expected_messages = create_test_messages(ROW_COMPARISON_MESSAGE_COUNT as usize); + let mut messages: Vec = expected_messages + .iter() + .enumerate() + .map(|(index, message)| { + let payload = + serde_json::to_vec(message).expect("Test message payload should serialize"); + IggyMessage::builder() + .id((index + 1) as u128) + .payload(Bytes::from(payload)) + .build() + .expect("Iggy message should build") + }) + .collect(); + + client + .send_messages( + &stream_id, + &topic_id, + &Partitioning::partition_id(0), + &mut messages, + ) + .await + .expect("Messages should be sent"); + + let polled_messages = client + .poll_messages( + &stream_id, + &topic_id, + Some(0), + &Consumer::default(), + &PollingStrategy::offset(0), + ROW_COMPARISON_MESSAGE_COUNT, + false, + ) + .await + .expect("Messages should be readable from Iggy"); + + fixture + .wait_for_test_table(WAIT_TIMEOUT_S) + .await + .expect("Fluss test table should be created"); + let rows = fixture + .read_from_test_table(WAIT_TIMEOUT_S) + .await + .expect("Fluss test table rows should be readable"); + + assert_eq!(rows.len(), expected_messages.len()); + assert_eq!(polled_messages.messages.len(), expected_messages.len()); + + for ((row, source_message), expected_message) in rows + .iter() + .zip(&polled_messages.messages) + .zip(&expected_messages) + { + assert_eq!( + row.get_string(ID_COLUMN_INDEX) + .expect("ID column should contain a string"), + format!("{:032x}", source_message.header.id) + ); + assert_eq!( + row.get_decimal(CHECKSUM_COLUMN_INDEX, 20, 0) + .expect("Checksum column should contain a decimal"), + expected_decimal(source_message.header.checksum) + ); + assert_eq!( + row.get_decimal(IGGY_OFFSET_COLUMN_INDEX, 20, 0) + .expect("Iggy offset column should contain a decimal"), + expected_decimal(source_message.header.offset) + ); + assert_eq!( + row.get_timestamp_ltz(IGGY_TIMESTAMP_COLUMN_INDEX, TIMESTAMP_PRECISION) + .expect("Iggy timestamp column should contain a timestamp"), + expected_timestamp(source_message.header.timestamp) + ); + assert_eq!( + row.get_string(IGGY_STREAM_COLUMN_INDEX) + .expect("Iggy stream column should contain a string"), + seeds::names::STREAM + ); + assert_eq!( + row.get_string(IGGY_TOPIC_COLUMN_INDEX) + .expect("Iggy topic column should contain a string"), + seeds::names::TOPIC + ); + assert_eq!( + row.get_long(IGGY_PARTITION_ID_COLUMN_INDEX) + .expect("Iggy partition ID column should contain a bigint"), + 0 + ); + assert_eq!( + row.get_timestamp_ltz(IGGY_ORIGIN_TIMESTAMP_COLUMN_INDEX, TIMESTAMP_PRECISION) + .expect("Iggy origin timestamp column should contain a timestamp"), + expected_timestamp(source_message.header.origin_timestamp) + ); + + let payload = row + .get_string(PAYLOAD_COLUMN_INDEX) + .expect("Payload column should contain a string"); + let actual_message: TestMessage = + serde_json::from_str(payload).expect("Payload should contain a test message"); + assert_eq!(&actual_message, expected_message); + } +} + +#[iggy_harness( + server(connectors_runtime(config_path = "tests/connectors/fluss/sink.toml")), + seed = seeds::connector_stream +)] +async fn sink_should_use_arrow_for_appending_messages( + harness: &TestHarness, + fixture: FlussSinkFixture, +) { + let client = harness + .root_client() + .await + .expect("Root client should be available"); + + let stream_id: Identifier = seeds::names::STREAM + .try_into() + .expect("Stream identifier should be valid"); + let topic_id: Identifier = seeds::names::TOPIC + .try_into() + .expect("Topic identifier should be valid"); + + let expected_messages = create_test_messages(ROW_COMPARISON_MESSAGE_COUNT as usize); + let mut messages: Vec = expected_messages + .iter() + .enumerate() + .map(|(index, message)| { + let payload = + serde_json::to_vec(message).expect("Test message payload should serialize"); + IggyMessage::builder() + .id((index + 1) as u128) + .payload(Bytes::from(payload)) + .build() + .expect("Iggy message should build") + }) + .collect(); + + client + .send_messages( + &stream_id, + &topic_id, + &Partitioning::partition_id(0), + &mut messages, + ) + .await + .expect("Messages should be sent"); + + fixture + .wait_for_test_table(WAIT_TIMEOUT_S) + .await + .expect("Fluss test table should be created"); + let rows = fixture + .read_from_test_table_arrow_batch(WAIT_TIMEOUT_S) + .await + .expect("Fluss test table rows should be readable"); + + let records_number: usize = rows.iter().map(|row| row.num_records()).sum(); + assert_eq!(records_number, expected_messages.len()); +} diff --git a/core/integration/tests/connectors/fluss/mod.rs b/core/integration/tests/connectors/fluss/mod.rs new file mode 100644 index 0000000000..842231fee8 --- /dev/null +++ b/core/integration/tests/connectors/fluss/mod.rs @@ -0,0 +1,18 @@ +// 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. + +mod fluss_sink; diff --git a/core/integration/tests/connectors/fluss/sink.toml b/core/integration/tests/connectors/fluss/sink.toml new file mode 100644 index 0000000000..53914b6ab7 --- /dev/null +++ b/core/integration/tests/connectors/fluss/sink.toml @@ -0,0 +1,20 @@ +# 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. + +[connectors] +config_type = "local" +config_dir = "../connectors/sinks/fluss_sink" diff --git a/core/integration/tests/connectors/mod.rs b/core/integration/tests/connectors/mod.rs index 08b794fe8b..4c225ac42b 100644 --- a/core/integration/tests/connectors/mod.rs +++ b/core/integration/tests/connectors/mod.rs @@ -21,6 +21,7 @@ mod delta; mod doris; mod elasticsearch; mod fixtures; +mod fluss; mod http; mod http_config_provider; mod iceberg; @@ -35,7 +36,6 @@ mod runtime; mod s3; mod stdout; mod surrealdb; - use iggy_common::IggyTimestamp; use serde::{Deserialize, Serialize};