diff --git a/Cargo.lock b/Cargo.lock index 497c3288c06..76e3f55ed3a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -651,6 +651,12 @@ version = "2.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3d62b7694a562cdf5a74227903507c56ab2cc8bdd1f781ed5cb4cf9c9f810bfc" +[[package]] +name = "arraydeque" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d902e3d592a523def97af8f317b08ce16b7ab854c1985a0c671e6f15cebc236" + [[package]] name = "arrayref" version = "0.3.9" @@ -1959,6 +1965,9 @@ name = "bitflags" version = "2.10.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "812e12b5285cc515a9c72a5c1d3b6d46a19dac5acfef5265968c166106e31dd3" +dependencies = [ + "serde_core", +] [[package]] name = "bitvec" @@ -2537,6 +2546,19 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "config" +version = "0.14.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "68578f196d2a33ff61b27fae256c3164f65e36382648e30666dde05b8cc9dfdf" +dependencies = [ + "nom", + "pathdiff", + "serde", + "serde_json", + "yaml-rust2", +] + [[package]] name = "console" version = "0.15.11" @@ -3776,7 +3798,7 @@ dependencies = [ "itertools 0.14.0", "libc", "memory-stats", - "metrics", + "metrics 0.23.1", "mimalloc-rust-sys", "nix 0.29.0", "num", @@ -3878,6 +3900,9 @@ dependencies = [ "dyn-clone", "enum-map", "erased-serde", + "etl", + "etl-config", + "etl-postgres", "feldera-adapterlib", "feldera-datagen", "feldera-iceberg", @@ -3919,7 +3944,7 @@ dependencies = [ "parquet", "postgres", "postgres-openssl", - "postgres-types", + "postgres-types 0.2.9", "pretty_assertions", "proptest", "proptest-derive", @@ -3951,7 +3976,7 @@ dependencies = [ "thread-id", "threadpool", "tokio", - "tokio-postgres", + "tokio-postgres 0.7.13", "tokio-stream", "tokio-util", "tracing", @@ -4029,7 +4054,7 @@ dependencies = [ "deadpool 0.12.3", "getrandom 0.2.16", "tokio", - "tokio-postgres", + "tokio-postgres 0.7.13", "tracing", ] @@ -4278,10 +4303,23 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e7c1832837b905bbfb5101e07cc24c8deddf52f93225eee6ead5f4d63d53ddcb" dependencies = [ "const-oid", + "der_derive", + "flagset", "pem-rfc7468", "zeroize", ] +[[package]] +name = "der_derive" +version = "0.7.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8034092389675178f570469e6c3b0465d3d30b4505c294a6550db47f3c17ad18" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.117", +] + [[package]] name = "deranged" version = "0.5.6" @@ -4755,6 +4793,70 @@ dependencies = [ "windows-sys 0.48.0", ] +[[package]] +name = "etl" +version = "0.1.0" +source = "git+https://github.com/supabase/etl?rev=05cb11ae#05cb11ae9145a81dc97c1907a6d9778566946db6" +dependencies = [ + "aws-lc-rs", + "byteorder", + "bytes", + "chrono", + "const-oid", + "etl-config", + "etl-postgres", + "fail", + "futures", + "metrics 0.24.3", + "pg_escape", + "pin-project-lite", + "postgres-replication", + "rustls 0.23.37", + "serde", + "serde_json", + "sqlx", + "sysinfo 0.38.4", + "tokio", + "tokio-postgres 0.7.11", + "tokio-rustls 0.26.2", + "tokio-stream", + "tracing", + "uuid", + "x509-cert", +] + +[[package]] +name = "etl-config" +version = "0.1.0" +source = "git+https://github.com/supabase/etl?rev=05cb11ae#05cb11ae9145a81dc97c1907a6d9778566946db6" +dependencies = [ + "config", + "secrecy", + "serde", + "sqlx", + "thiserror 2.0.17", + "tokio-postgres 0.7.11", + "url", +] + +[[package]] +name = "etl-postgres" +version = "0.1.0" +source = "git+https://github.com/supabase/etl?rev=05cb11ae#05cb11ae9145a81dc97c1907a6d9778566946db6" +dependencies = [ + "bytes", + "chrono", + "etl-config", + "pg_escape", + "serde", + "serde_json", + "sqlx", + "thiserror 2.0.17", + "tokio", + "tokio-postgres 0.7.11", + "tracing", +] + [[package]] name = "event-listener" version = "2.5.3" @@ -4792,6 +4894,17 @@ dependencies = [ "once_cell", ] +[[package]] +name = "fail" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fe5e43d0f78a42ad591453aedb1d7ae631ce7ee445c7643691055a9ed8d3b01c" +dependencies = [ + "log", + "once_cell", + "rand 0.8.5", +] + [[package]] name = "fake" version = "2.10.0" @@ -5216,7 +5329,7 @@ dependencies = [ "rkyv", "serde", "serde_json", - "tokio-postgres", + "tokio-postgres 0.7.13", ] [[package]] @@ -5277,6 +5390,12 @@ version = "0.5.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d674e81391d1e1ab681a28d99df07927c6d4aa5b027d7da16ba32d1d21ecd99" +[[package]] +name = "flagset" +version = "0.4.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b7ac824320a75a52197e8f2d787f6a38b6718bb6897a35142d749af3c0e8f4fe" + [[package]] name = "flatbuffers" version = "25.2.10" @@ -5304,6 +5423,17 @@ version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8bf7cc16383c4b8d58b9905a8509f02926ce3058053c056376248d958c9df1e8" +[[package]] +name = "flume" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "da0e4dd2a88388a1f4ccc7c9ce104604dab68d9f408dc34cd45823d5a9069095" +dependencies = [ + "futures-core", + "futures-sink", + "spin 0.9.8", +] + [[package]] name = "fnv" version = "1.0.7" @@ -5838,6 +5968,15 @@ dependencies = [ "foldhash 0.2.0", ] +[[package]] +name = "hashlink" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e8094feaf31ff591f651a2664fb9cfd92bba7a60ce3197265e9482ebe753c8f7" +dependencies = [ + "hashbrown 0.14.5", +] + [[package]] name = "hashlink" version = "0.10.0" @@ -6721,6 +6860,9 @@ name = "lazy_static" version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" +dependencies = [ + "spin 0.9.8", +] [[package]] name = "leb128fmt" @@ -6842,6 +6984,16 @@ dependencies = [ "redox_syscall 0.5.12", ] +[[package]] +name = "libsqlite3-sys" +version = "0.30.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2e99fb7a497b1e3339bc746195567ed8d3e24945ecd636e3619d20b9de9e9149" +dependencies = [ + "pkg-config", + "vcpkg", +] + [[package]] name = "libz-sys" version = "1.1.22" @@ -7069,6 +7221,16 @@ dependencies = [ "portable-atomic", ] +[[package]] +name = "metrics" +version = "0.24.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5d5312e9ba3771cfa961b585728215e3d972c950a3eed9252aa093d6301277e8" +dependencies = [ + "ahash 0.8.12", + "portable-atomic", +] + [[package]] name = "metrics-process" version = "2.4.0" @@ -7342,6 +7504,22 @@ dependencies = [ "serde", ] +[[package]] +name = "num-bigint-dig" +version = "0.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e661dda6640fad38e827a6d4a310ff4763082116fe217f279885c97f511bb0b7" +dependencies = [ + "lazy_static", + "libm", + "num-integer", + "num-iter", + "num-traits", + "rand 0.8.5", + "smallvec", + "zeroize", +] + [[package]] name = "num-complex" version = "0.4.6" @@ -7467,13 +7645,23 @@ checksum = "830b246a0e5f20af87141b25c173cd1b609bd7779a4617d6ec582abaf90870f3" [[package]] name = "objc2-core-foundation" -version = "0.3.1" +version = "0.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1c10c2894a6fed806ade6027bcd50662746363a9589d3ec9d9bef30a4e4bc166" +checksum = "2a180dd8642fa45cdb7dd721cd4c11b1cadd4929ce112ebd8b9f5803cc79d536" dependencies = [ "bitflags 2.10.0", ] +[[package]] +name = "objc2-io-kit" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "33fafba39597d6dc1fb709123dfa8289d39406734be322956a69f0931c73bb15" +dependencies = [ + "libc", + "objc2-core-foundation", +] + [[package]] name = "object" version = "0.36.7" @@ -7941,6 +8129,12 @@ version = "0.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "498a099351efa4becc6a19c72aa9270598e8fd274ca47052e37455241c88b696" +[[package]] +name = "pathdiff" +version = "0.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df94ce210e5bc13cb6651479fa48d14f601d9858cfe0467f43ae157023b938d3" + [[package]] name = "pbkdf2" version = "0.12.2" @@ -8012,7 +8206,16 @@ checksum = "9e024e787d4b99c9f2c1e7f84e0da059ad4ca935e60916a830b5a3e0030d5aa1" dependencies = [ "rust-ini 0.18.0", "thiserror 1.0.69", - "tokio-postgres", + "tokio-postgres 0.7.13", +] + +[[package]] +name = "pg_escape" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "44c7bc82ccbe2c7ef7ceed38dcac90d7ff46681e061e9d7310cbcd409113e303" +dependencies = [ + "phf 0.11.3", ] [[package]] @@ -8021,6 +8224,7 @@ version = "0.11.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1fd6780a80ae0c52cc120a26a1a42c1ae51b247a253e4e06113d23d2c2edd078" dependencies = [ + "phf_macros", "phf_shared 0.11.3", ] @@ -8033,6 +8237,29 @@ dependencies = [ "phf_shared 0.12.1", ] +[[package]] +name = "phf_generator" +version = "0.11.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3c80231409c20246a13fddb31776fb942c38553c51e871f8cbd687a4cfb5843d" +dependencies = [ + "phf_shared 0.11.3", + "rand 0.8.5", +] + +[[package]] +name = "phf_macros" +version = "0.11.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f84ac04429c13a7ff43785d75ad27569f2951ce0ffd30a3321230db2fc727216" +dependencies = [ + "phf_generator", + "phf_shared 0.11.3", + "proc-macro2", + "quote", + "syn 2.0.117", +] + [[package]] name = "phf_shared" version = "0.11.3" @@ -8148,7 +8375,7 @@ dependencies = [ "thiserror 2.0.17", "tikv-jemallocator", "tokio", - "tokio-postgres", + "tokio-postgres 0.7.13", "tokio-stream", "tracing", "tracing-subscriber", @@ -8173,6 +8400,17 @@ dependencies = [ "futures-io", ] +[[package]] +name = "pkcs1" +version = "0.7.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c8ffb9f10fa047879315e6625af03c164b16962a5368d724ed16323b68ace47f" +dependencies = [ + "der 0.7.10", + "pkcs8 0.10.2", + "spki 0.7.3", +] + [[package]] name = "pkcs8" version = "0.9.0" @@ -8281,7 +8519,7 @@ dependencies = [ "futures-util", "log", "tokio", - "tokio-postgres", + "tokio-postgres 0.7.13", ] [[package]] @@ -8305,7 +8543,7 @@ dependencies = [ "native-tls", "tokio", "tokio-native-tls", - "tokio-postgres", + "tokio-postgres 0.7.13", ] [[package]] @@ -8317,7 +8555,24 @@ dependencies = [ "openssl", "tokio", "tokio-openssl", - "tokio-postgres", + "tokio-postgres 0.7.13", +] + +[[package]] +name = "postgres-protocol" +version = "0.6.7" +source = "git+https://github.com/MaterializeInc/rust-postgres?rev=c4b473b478b3adfbf8667d2fbe895d8423f1290b#c4b473b478b3adfbf8667d2fbe895d8423f1290b" +dependencies = [ + "base64 0.22.1", + "byteorder", + "bytes", + "fallible-iterator", + "hmac", + "md-5", + "memchr", + "rand 0.8.5", + "sha2", + "stringprep", ] [[package]] @@ -8338,6 +8593,35 @@ dependencies = [ "stringprep", ] +[[package]] +name = "postgres-replication" +version = "0.6.7" +source = "git+https://github.com/MaterializeInc/rust-postgres?rev=c4b473b478b3adfbf8667d2fbe895d8423f1290b#c4b473b478b3adfbf8667d2fbe895d8423f1290b" +dependencies = [ + "byteorder", + "bytes", + "futures-util", + "memchr", + "pin-project-lite", + "postgres-protocol 0.6.7", + "postgres-types 0.2.7", + "tokio-postgres 0.7.11", +] + +[[package]] +name = "postgres-types" +version = "0.2.7" +source = "git+https://github.com/MaterializeInc/rust-postgres?rev=c4b473b478b3adfbf8667d2fbe895d8423f1290b#c4b473b478b3adfbf8667d2fbe895d8423f1290b" +dependencies = [ + "bytes", + "chrono", + "fallible-iterator", + "postgres-protocol 0.6.7", + "serde", + "serde_json", + "uuid", +] + [[package]] name = "postgres-types" version = "0.2.9" @@ -8348,7 +8632,7 @@ dependencies = [ "chrono", "fallible-iterator", "postgres-derive", - "postgres-protocol", + "postgres-protocol 0.6.8", "serde", "serde_json", "uuid", @@ -9299,7 +9583,7 @@ dependencies = [ "thiserror 2.0.17", "time", "tokio", - "tokio-postgres", + "tokio-postgres 0.7.13", "toml", "url", "walkdir", @@ -9703,6 +9987,26 @@ version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4e27ee8bb91ca0adcf0ecb116293afa12d393f9c2b9b9cd54d33e8078fe19839" +[[package]] +name = "rsa" +version = "0.9.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8573f03f5883dcaebdfcf4725caa1ecb9c15b2ef50c43a07b816e06799bb12d" +dependencies = [ + "const-oid", + "digest", + "num-bigint-dig", + "num-integer", + "num-traits", + "pkcs1", + "pkcs8 0.10.2", + "rand_core 0.6.4", + "signature 2.2.0", + "spki 0.7.3", + "subtle", + "zeroize", +] + [[package]] name = "rstar" version = "0.11.0" @@ -10169,6 +10473,16 @@ dependencies = [ "zeroize", ] +[[package]] +name = "secrecy" +version = "0.10.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e891af845473308773346dc847b2c23ee78fe442e0472ac50e22a18a93d3ae5a" +dependencies = [ + "serde", + "zeroize", +] + [[package]] name = "security-framework" version = "2.11.1" @@ -10899,7 +11213,9 @@ checksum = "1fefb893899429669dcdd979aff487bd78f4064e5e7907e4269081e0ef7d97dc" dependencies = [ "sqlx-core", "sqlx-macros", + "sqlx-mysql", "sqlx-postgres", + "sqlx-sqlite", ] [[package]] @@ -10919,13 +11235,14 @@ dependencies = [ "futures-io", "futures-util", "hashbrown 0.15.3", - "hashlink", + "hashlink 0.10.0", "indexmap 2.13.0", "log", "memchr", "native-tls", "once_cell", "percent-encoding", + "rustls 0.23.37", "serde", "serde_json", "sha2", @@ -10935,6 +11252,7 @@ dependencies = [ "tokio-stream", "tracing", "url", + "webpki-roots 0.26.11", ] [[package]] @@ -10967,12 +11285,56 @@ dependencies = [ "serde_json", "sha2", "sqlx-core", + "sqlx-mysql", "sqlx-postgres", + "sqlx-sqlite", "syn 2.0.117", "tokio", "url", ] +[[package]] +name = "sqlx-mysql" +version = "0.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aa003f0038df784eb8fecbbac13affe3da23b45194bd57dba231c8f48199c526" +dependencies = [ + "atoi", + "base64 0.22.1", + "bitflags 2.10.0", + "byteorder", + "bytes", + "crc", + "digest", + "dotenvy", + "either", + "futures-channel", + "futures-core", + "futures-io", + "futures-util", + "generic-array", + "hex", + "hkdf", + "hmac", + "itoa", + "log", + "md-5", + "memchr", + "once_cell", + "percent-encoding", + "rand 0.8.5", + "rsa", + "serde", + "sha1", + "sha2", + "smallvec", + "sqlx-core", + "stringprep", + "thiserror 2.0.17", + "tracing", + "whoami", +] + [[package]] name = "sqlx-postgres" version = "0.8.6" @@ -11010,6 +11372,30 @@ dependencies = [ "whoami", ] +[[package]] +name = "sqlx-sqlite" +version = "0.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c2d12fe70b2c1b4401038055f90f151b78208de1f9f89a7dbfd41587a10c3eea" +dependencies = [ + "atoi", + "flume", + "futures-channel", + "futures-core", + "futures-executor", + "futures-intrusive", + "futures-util", + "libsqlite3-sys", + "log", + "percent-encoding", + "serde", + "serde_urlencoded", + "sqlx-core", + "thiserror 2.0.17", + "tracing", + "url", +] + [[package]] name = "stable_deref_trait" version = "1.2.0" @@ -11201,6 +11587,20 @@ dependencies = [ "windows 0.57.0", ] +[[package]] +name = "sysinfo" +version = "0.38.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92ab6a2f8bfe508deb3c6406578252e491d299cbbf3bc0529ecc3313aee4a52f" +dependencies = [ + "libc", + "memchr", + "ntapi", + "objc2-core-foundation", + "objc2-io-kit", + "windows 0.62.2", +] + [[package]] name = "system-configuration" version = "0.5.1" @@ -11637,6 +12037,31 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-postgres" +version = "0.7.11" +source = "git+https://github.com/MaterializeInc/rust-postgres?rev=c4b473b478b3adfbf8667d2fbe895d8423f1290b#c4b473b478b3adfbf8667d2fbe895d8423f1290b" +dependencies = [ + "async-trait", + "byteorder", + "bytes", + "fallible-iterator", + "futures-channel", + "futures-util", + "log", + "parking_lot 0.12.4", + "percent-encoding", + "phf 0.11.3", + "pin-project-lite", + "postgres-protocol 0.6.7", + "postgres-types 0.2.7", + "rand 0.8.5", + "socket2 0.5.10", + "tokio", + "tokio-util", + "whoami", +] + [[package]] name = "tokio-postgres" version = "0.7.13" @@ -11654,8 +12079,8 @@ dependencies = [ "percent-encoding", "phf 0.11.3", "pin-project-lite", - "postgres-protocol", - "postgres-types", + "postgres-protocol 0.6.8", + "postgres-types 0.2.9", "rand 0.9.2", "socket2 0.5.10", "tokio", @@ -11711,9 +12136,9 @@ dependencies = [ [[package]] name = "tokio-stream" -version = "0.1.17" +version = "0.1.18" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "eca58d7bba4a75707817a2c44174253f9236b2d5fbd055602e9d5c07c139a047" +checksum = "32da49809aab5c3bc678af03902d4ccddea2a87d028d86392a4b1560c6906c70" dependencies = [ "futures-core", "pin-project-lite", @@ -12255,14 +12680,15 @@ dependencies = [ [[package]] name = "url" -version = "2.5.7" +version = "2.5.8" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "08bc136a29a3d1758e07a9cca267be308aeebf5cfd5a10f3f67ab2097683ef5b" +checksum = "ff67a8a4397373c3ef660812acab3268222035010ab8680ec4215f38ba3d0eed" dependencies = [ "form_urlencoded", "idna", "percent-encoding", "serde", + "serde_derive", ] [[package]] @@ -12428,7 +12854,7 @@ dependencies = [ "regex", "rustc_version", "rustversion", - "sysinfo", + "sysinfo 0.34.2", "time", "vergen-lib", ] @@ -12806,11 +13232,23 @@ version = "0.61.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c5ee8f3d025738cb02bad7868bbb5f8a6327501e870bf51f1b455b0a2454a419" dependencies = [ - "windows-collections", + "windows-collections 0.2.0", "windows-core 0.61.2", - "windows-future", + "windows-future 0.2.1", "windows-link 0.1.1", - "windows-numerics", + "windows-numerics 0.2.0", +] + +[[package]] +name = "windows" +version = "0.62.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "527fadee13e0c05939a6a05d5bd6eec6cd2e3dbd648b9f8e447c6518133d8580" +dependencies = [ + "windows-collections 0.3.2", + "windows-core 0.62.2", + "windows-future 0.3.2", + "windows-numerics 0.3.1", ] [[package]] @@ -12822,6 +13260,15 @@ dependencies = [ "windows-core 0.61.2", ] +[[package]] +name = "windows-collections" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "23b2d95af1a8a14a3c7367e1ed4fc9c20e0a26e79551b1454d72583c97cc6610" +dependencies = [ + "windows-core 0.62.2", +] + [[package]] name = "windows-core" version = "0.57.0" @@ -12853,13 +13300,26 @@ version = "0.61.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c0fdd3ddb90610c7638aa2b3a3ab2904fb9e5cdbecc643ddb3647212781c4ae3" dependencies = [ - "windows-implement 0.60.0", - "windows-interface 0.59.1", + "windows-implement 0.60.2", + "windows-interface 0.59.3", "windows-link 0.1.1", "windows-result 0.3.4", "windows-strings 0.4.2", ] +[[package]] +name = "windows-core" +version = "0.62.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8e83a14d34d0623b51dce9581199302a221863196a1dde71a7663a4c2be9deb" +dependencies = [ + "windows-implement 0.60.2", + "windows-interface 0.59.3", + "windows-link 0.2.1", + "windows-result 0.4.1", + "windows-strings 0.5.1", +] + [[package]] name = "windows-future" version = "0.2.1" @@ -12868,7 +13328,18 @@ checksum = "fc6a41e98427b19fe4b73c550f060b59fa592d7d686537eebf9385621bfbad8e" dependencies = [ "windows-core 0.61.2", "windows-link 0.1.1", - "windows-threading", + "windows-threading 0.1.0", +] + +[[package]] +name = "windows-future" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e1d6f90251fe18a279739e78025bd6ddc52a7e22f921070ccdc67dde84c605cb" +dependencies = [ + "windows-core 0.62.2", + "windows-link 0.2.1", + "windows-threading 0.2.1", ] [[package]] @@ -12895,9 +13366,9 @@ dependencies = [ [[package]] name = "windows-implement" -version = "0.60.0" +version = "0.60.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a47fddd13af08290e67f4acabf4b459f647552718f683a7b415d290ac744a836" +checksum = "053e2e040ab57b9dc951b72c264860db7eb3b0200ba345b4e4c3b14f67855ddf" dependencies = [ "proc-macro2", "quote", @@ -12928,9 +13399,9 @@ dependencies = [ [[package]] name = "windows-interface" -version = "0.59.1" +version = "0.59.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bd9211b69f8dcdfa817bfd14bf1c97c9188afa36f4750130fcdf3f400eca9fa8" +checksum = "3f316c4a2570ba26bbec722032c4099d8c8bc095efccdc15688708623367e358" dependencies = [ "proc-macro2", "quote", @@ -12959,6 +13430,16 @@ dependencies = [ "windows-link 0.1.1", ] +[[package]] +name = "windows-numerics" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e2e40844ac143cdb44aead537bbf727de9b044e107a0f1220392177d15b0f26" +dependencies = [ + "windows-core 0.62.2", + "windows-link 0.2.1", +] + [[package]] name = "windows-registry" version = "0.5.2" @@ -12997,6 +13478,15 @@ dependencies = [ "windows-link 0.1.1", ] +[[package]] +name = "windows-result" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7781fa89eaf60850ac3d2da7af8e5242a5ea78d1a11c49bf2910bb5a73853eb5" +dependencies = [ + "windows-link 0.2.1", +] + [[package]] name = "windows-strings" version = "0.1.0" @@ -13016,6 +13506,15 @@ dependencies = [ "windows-link 0.1.1", ] +[[package]] +name = "windows-strings" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7837d08f69c77cf6b07689544538e017c1bfcf57e34b4c0ff58e6c2cd3b37091" +dependencies = [ + "windows-link 0.2.1", +] + [[package]] name = "windows-sys" version = "0.48.0" @@ -13118,6 +13617,15 @@ dependencies = [ "windows-link 0.1.1", ] +[[package]] +name = "windows-threading" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3949bd5b99cafdf1c7ca86b43ca564028dfe27d66958f2470940f73d86d75b37" +dependencies = [ + "windows-link 0.2.1", +] + [[package]] name = "windows_aarch64_gnullvm" version = "0.48.5" @@ -13411,6 +13919,17 @@ dependencies = [ "tap", ] +[[package]] +name = "x509-cert" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1301e935010a701ae5f8655edc0ad17c44bad3ac5ce8c39185f75453b720ae94" +dependencies = [ + "const-oid", + "der 0.7.10", + "spki 0.7.3", +] + [[package]] name = "xattr" version = "1.5.0" @@ -13442,6 +13961,17 @@ dependencies = [ "lzma-sys", ] +[[package]] +name = "yaml-rust2" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8902160c4e6f2fb145dbe9d6760a75e3c9522d8bf796ed7047c85919ac7115f8" +dependencies = [ + "arraydeque", + "encoding_rs", + "hashlink 0.8.4", +] + [[package]] name = "yansi" version = "1.0.1" diff --git a/Cargo.toml b/Cargo.toml index b1c76c46d33..84321c0225f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -268,7 +268,7 @@ tikv-jemallocator = "0.6.0" time = { version = "0.3.47", features = ["serde", "serde-well-known"] } tokio = "1.50.0" tokio-postgres = "0.7" -tokio-stream = "0.1.15" +tokio-stream = "0.1.18" tokio-util = "0.7.11" tracing = "0.1.40" tracing-subscriber = "0.3.20" @@ -286,6 +286,9 @@ zip = "6.0.0" zstd = "0.12.0" backtrace = "0.3.75" parking_lot = "0.12.4" +etl = { git = "https://github.com/supabase/etl", rev = "05cb11ae" } +etl-config = { git = "https://github.com/supabase/etl", rev = "05cb11ae" } +etl-postgres = { git = "https://github.com/supabase/etl", rev = "05cb11ae" } [workspace.metadata.release] release = false diff --git a/crates/adapterlib/src/transport.rs b/crates/adapterlib/src/transport.rs index 06d786367d2..e8086057ef2 100644 --- a/crates/adapterlib/src/transport.rs +++ b/crates/adapterlib/src/transport.rs @@ -3,6 +3,7 @@ use chrono::{DateTime, Utc}; use dyn_clone::DynClone; use feldera_types::adapter_stats::ConnectorHealth; use feldera_types::config::FtModel; +use feldera_types::coordination::Completion; use feldera_types::program_schema::Relation; use rmpv::{Value as RmpValue, ext::Error as RmpDecodeError}; use serde::Deserialize; @@ -766,6 +767,19 @@ pub trait InputConsumer: Send + Sync + DynClone { /// so connectors that have no custom metrics need not override it. fn set_custom_metrics(&self, _metrics: Arc) {} + /// Returns a watch receiver that tracks completion of pipeline steps. + /// + /// The receiver yields [`Completion`] values whose `total_completed_steps` + /// field indicates how many steps have been fully processed (circuit + /// execution + all output connectors). + /// + /// Input adapters that need to defer acknowledgment until data is durably + /// processed (e.g., CDC adapters controlling a replication slot) can use + /// this to detect when their data has been consumed. + /// + /// Return `None` if the consumer does not support completion tracking. + fn completion_watcher(&self) -> Option>; + /// Endpoint failed. /// /// Reports that the endpoint failed and that it will not queue any more diff --git a/crates/adapters/Cargo.toml b/crates/adapters/Cargo.toml index 7a890f8a466..2e998652ea2 100644 --- a/crates/adapters/Cargo.toml +++ b/crates/adapters/Cargo.toml @@ -24,6 +24,7 @@ default = [ "with-pubsub", "with-redis", "with-nats", + "with-postgres-cdc", ] with-kafka = ["rdkafka"] with-deltalake = ["deltalake", "deltalake-catalog-unity"] @@ -52,6 +53,7 @@ iceberg-tests-glue = [] iceberg-tests-rest = [] fips = ["rustls/fips"] bench-mode = [] +with-postgres-cdc = ["etl", "etl-config", "etl-postgres"] [dependencies] feldera-types = { workspace = true } @@ -201,6 +203,9 @@ zip = { workspace = true } smallvec = { workspace = true } delta_kernel = { workspace = true } flate2 = { workspace = true } +etl = { workspace = true, optional = true } +etl-config = { workspace = true, optional = true } +etl-postgres = { workspace = true, optional = true } [package.metadata.cargo-machete] ignored = ["num-traits"] diff --git a/crates/adapters/src/controller.rs b/crates/adapters/src/controller.rs index 54bcb38d5b8..0d3b9939b01 100644 --- a/crates/adapters/src/controller.rs +++ b/crates/adapters/src/controller.rs @@ -7260,6 +7260,10 @@ impl InputConsumer for InputProbe { self.transaction_in_progress.store(false, Ordering::Release); } + fn completion_watcher(&self) -> Option> { + Some(self.controller.status.completion_notifier.subscribe()) + } + fn error(&self, fatal: bool, error: AnyError, tag: Option<&str>) { self.controller.input_transport_error( self.endpoint_id, diff --git a/crates/adapters/src/integrated.rs b/crates/adapters/src/integrated.rs index 377763acfbb..fff8ec49b2e 100644 --- a/crates/adapters/src/integrated.rs +++ b/crates/adapters/src/integrated.rs @@ -9,6 +9,8 @@ use std::sync::Weak; pub mod delta_table; mod postgres; +#[cfg(feature = "with-postgres-cdc")] +use crate::integrated::postgres::PostgresCdcInputEndpoint; use crate::integrated::postgres::PostgresInputEndpoint; pub use crate::integrated::postgres::PostgresOutputEndpoint; @@ -99,6 +101,12 @@ pub fn create_integrated_input_endpoint( TransportConfig::PostgresInput(config) => { Box::new(PostgresInputEndpoint::new(endpoint_name, config, consumer)) } + #[cfg(feature = "with-postgres-cdc")] + TransportConfig::PostgresCdcInput(config) => Box::new(PostgresCdcInputEndpoint::new( + endpoint_name, + config, + consumer, + )), transport => { return Err(ControllerError::unknown_input_transport( endpoint_name, diff --git a/crates/adapters/src/integrated/postgres.rs b/crates/adapters/src/integrated/postgres.rs index 3bc1efea9b7..a413972568d 100644 --- a/crates/adapters/src/integrated/postgres.rs +++ b/crates/adapters/src/integrated/postgres.rs @@ -5,8 +5,14 @@ mod output_macros; mod prepared_statements; mod tls; +#[cfg(feature = "with-postgres-cdc")] +pub(crate) mod cdc_input; + #[cfg(test)] mod test; pub use input::PostgresInputEndpoint; pub use output::PostgresOutputEndpoint; + +#[cfg(feature = "with-postgres-cdc")] +pub use cdc_input::PostgresCdcInputEndpoint; diff --git a/crates/adapters/src/integrated/postgres/cdc_input.rs b/crates/adapters/src/integrated/postgres/cdc_input.rs new file mode 100644 index 00000000000..06f15ea5b2c --- /dev/null +++ b/crates/adapters/src/integrated/postgres/cdc_input.rs @@ -0,0 +1,1493 @@ +use crate::transport::{ + InputEndpoint, InputQueue, InputReaderCommand, IntegratedInputEndpoint, NonFtInputReaderCommand, +}; +use crate::{ControllerError, InputConsumer, InputReader, PipelineState, RecordFormat}; +use anyhow::{Result as AnyResult, anyhow}; +use chrono::Utc; +use dbsp::circuit::tokio::TOKIO; +use etl::config::{ + BatchConfig, InvalidatedSlotBehavior, MemoryBackpressureConfig, PgConnectionConfig, + PipelineConfig, TableSyncCopyConfig, TcpKeepaliveConfig, TlsConfig, +}; +use etl::destination::Destination; +use etl::destination::async_result::{ + TruncateTableResult, WriteEventsResult, WriteTableRowsResult, +}; +use etl::error::EtlResult; +use etl::pipeline::Pipeline; +use etl::store::both::postgres::PostgresStore; +use etl::types::{ArrayCell, Cell, Event, TableId, TableRow}; +use feldera_adapterlib::catalog::{DeCollectionStream, InputCollectionHandle}; +use feldera_adapterlib::format::ParseError; +use feldera_adapterlib::transport::{Resume, Watermark}; +use feldera_types::config::FtModel; +use feldera_types::coordination::Completion; +use feldera_types::format::json::JsonFlavor; +use feldera_types::transport::postgres::PostgresCdcReaderConfig; +use serde_json::{Value, json}; +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; +use std::thread; +use tokio::select; +use tokio::sync::mpsc; +use tokio::sync::watch::{Receiver, Sender, channel}; +use tracing::{debug, error, info, warn}; +use url::Url; +use xxhash_rust::xxh3::xxh3_64; + +/// Deferred async result senders waiting for step completion. +type DeferredSenders = Vec>; + +/// Integrated input connector that reads from Postgres via logical replication (CDC). +pub struct PostgresCdcInputEndpoint { + inner: Arc, +} + +impl PostgresCdcInputEndpoint { + pub fn new( + endpoint_name: &str, + config: &PostgresCdcReaderConfig, + consumer: Box, + ) -> Self { + Self { + inner: Arc::new(PostgresCdcInputInner::new( + endpoint_name, + config.clone(), + consumer, + )), + } + } +} + +impl InputEndpoint for PostgresCdcInputEndpoint { + fn fault_tolerance(&self) -> Option { + Some(FtModel::AtLeastOnce) + } +} + +impl IntegratedInputEndpoint for PostgresCdcInputEndpoint { + fn open( + self: Box, + input_handle: &InputCollectionHandle, + _resume_info: Option, + ) -> AnyResult> { + Ok(Box::new(PostgresCdcInputReader::new( + &self.inner, + input_handle, + )?)) + } +} + +struct PostgresCdcInputReader { + sender: Sender, + inner: Arc, +} + +impl PostgresCdcInputReader { + fn new( + endpoint: &Arc, + input_handle: &InputCollectionHandle, + ) -> AnyResult { + let (sender, receiver) = channel(PipelineState::Paused); + let endpoint_clone = endpoint.clone(); + + let (init_status_sender, init_status_receiver) = + tokio::sync::oneshot::channel::>(); + + let input_stream = input_handle + .handle + .configure_deserializer(RecordFormat::Json(JsonFlavor::Datagen))?; + + thread::Builder::new() + .name("postgres-cdc-input-tokio-wrapper".to_string()) + .spawn(move || { + TOKIO.block_on(async { + let _ = endpoint_clone + .worker_task(input_stream, receiver, init_status_sender) + .await; + }) + }) + .expect("failed to create Postgres CDC input connector thread"); + + init_status_receiver.blocking_recv().map_err(|_| { + ControllerError::input_transport_error( + &endpoint.endpoint_name, + true, + anyhow!("worker thread terminated unexpectedly during initialization"), + ) + })??; + + Ok(Self { + sender, + inner: endpoint.clone(), + }) + } +} + +impl InputReader for PostgresCdcInputReader { + fn as_any(self: Arc) -> Arc { + self + } + + fn request(&self, command: InputReaderCommand) { + match command.as_nonft().unwrap() { + NonFtInputReaderCommand::Queue => { + // Flush queue to circuit, collecting timestamps for watermarks. + let (buffer_size, _hasher, flushed) = self.inner.queue.flush_with_aux(); + + let watermarks: Vec = flushed + .iter() + .map(|(ts, _)| Watermark::new(*ts, None)) + .collect(); + + // Build resume metadata so Feldera can checkpoint our position. + // The actual resume state is managed by etl's PostgresStore; + // we just need a stable identifier so the controller knows we + // support resumption. + let resume_metadata = json!({ + "pipeline_id": self.inner.pipeline_id, + }); + let resume = Resume::Seek { + seek: resume_metadata, + }; + + // Report data to controller with resume metadata (must be + // called exactly once per Queue command). + self.inner + .consumer + .extended(buffer_size, Some(resume), watermarks); + + // Take any deferred senders that write_events stored. + let senders: DeferredSenders = + std::mem::take(&mut *self.inner.pending_senders.lock().unwrap()); + + if !senders.is_empty() { + if let Some(tx) = self.inner.completion_tx.as_ref() { + // Read completed steps AFTER flush — the step containing + // our data hasn't completed yet, so waiting for this value + // to increase guarantees our data was processed. + let completed = self + .inner + .completion_rx + .as_ref() + .map(|rx| rx.borrow().total_completed_steps) + .unwrap_or(0); + let _ = tx.send((completed, senders)); + } else { + // No completion tracking — fire immediately. + for sender in senders { + sender.send(Ok(())); + } + } + } + } + NonFtInputReaderCommand::Transition(state) => drop(self.sender.send_replace(state)), + } + } + + fn is_closed(&self) -> bool { + self.inner.queue.is_empty() && self.sender.is_closed() + } +} + +impl Drop for PostgresCdcInputReader { + fn drop(&mut self) { + self.disconnect(); + } +} + +struct PostgresCdcInputInner { + endpoint_name: String, + config: PostgresCdcReaderConfig, + consumer: Box, + queue: Arc, + /// Deterministic pipeline ID used for replication slot naming and resume. + pipeline_id: u64, + /// Deferred async result senders from `write_events`, waiting to be paired + /// with a step number during the next `Queue` command. + pending_senders: Arc>, + /// Watch receiver for step completion (cloned into the Queue handler). + completion_rx: Option>, + /// Sender for passing (completed_at_flush, senders) to the background task. + /// Created once at construction time if completion tracking is available. + completion_tx: Option>, + /// Receiver half, taken once by worker_task_inner to spawn the background task. + completion_task_rx: Mutex>>, +} + +impl PostgresCdcInputInner { + fn new( + endpoint_name: &str, + config: PostgresCdcReaderConfig, + consumer: Box, + ) -> Self { + let queue = Arc::new(InputQueue::new(consumer.clone())); + let completion_rx = consumer.completion_watcher(); + + // Deterministic pipeline ID from config — stable across Rust versions (xxh3). + // Excludes password and other volatile fields so that rotating the password + // doesn't change the pipeline_id (which would orphan the replication slot + // and stored etl state, forcing a full re-snapshot). + let pipeline_id = { + let identity = + stable_connection_identity(&config.uri, &config.publication, &config.source_table); + xxh3_64(identity.as_bytes()) + }; + + let (completion_tx, completion_task_rx) = if completion_rx.is_some() { + let (tx, rx) = mpsc::unbounded_channel(); + (Some(tx), Some(rx)) + } else { + (None, None) + }; + + Self { + endpoint_name: endpoint_name.to_string(), + config, + consumer, + queue, + pipeline_id, + pending_senders: Arc::new(Mutex::new(Vec::new())), + completion_rx, + completion_tx, + completion_task_rx: Mutex::new(completion_task_rx), + } + } + + async fn worker_task( + self: Arc, + input_stream: Box, + receiver: Receiver, + init_status_sender: tokio::sync::oneshot::Sender>, + ) { + let mut receiver_clone = receiver.clone(); + select! { + _ = self.clone().worker_task_inner(input_stream, receiver, init_status_sender) => { + debug!("postgres_cdc {}: worker task terminated", &self.endpoint_name); + } + _ = receiver_clone.wait_for(|state| state == &PipelineState::Terminated) => { + debug!("postgres_cdc {}: received termination command; worker task canceled", + &self.endpoint_name); + } + } + } + + async fn worker_task_inner( + self: Arc, + input_stream: Box, + mut receiver: Receiver, + init_status_sender: tokio::sync::oneshot::Sender>, + ) { + let pg_conn = match parse_pg_uri(&self.config.uri) { + Ok(conn) => conn, + Err(e) => { + let _ = + init_status_sender.send(Err(ControllerError::invalid_transport_configuration( + &self.endpoint_name, + &format!("failed to parse Postgres URI: {e}"), + ))); + return; + } + }; + + let pipeline_config = PipelineConfig { + id: self.pipeline_id, + publication_name: self.config.publication.clone(), + pg_connection: pg_conn.clone(), + batch: BatchConfig::default(), + table_error_retry_delay_ms: PipelineConfig::DEFAULT_TABLE_ERROR_RETRY_DELAY_MS, + table_error_retry_max_attempts: PipelineConfig::DEFAULT_TABLE_ERROR_RETRY_MAX_ATTEMPTS, + max_table_sync_workers: PipelineConfig::DEFAULT_MAX_TABLE_SYNC_WORKERS, + max_copy_connections_per_table: PipelineConfig::DEFAULT_MAX_COPY_CONNECTIONS_PER_TABLE, + memory_refresh_interval_ms: PipelineConfig::DEFAULT_MEMORY_REFRESH_INTERVAL_MS, + memory_backpressure: Some(MemoryBackpressureConfig::default()), + table_sync_copy: TableSyncCopyConfig::IncludeAllTables, + invalidated_slot_behavior: InvalidatedSlotBehavior::default(), + }; + + // Use PostgresStore to persist table replication phases across restarts. + // This allows etl to resume from the replication slot position instead of + // re-snapshotting the entire table on restart. + let store = match PostgresStore::new(self.pipeline_id, pg_conn).await { + Ok(store) => store, + Err(e) => { + let _ = init_status_sender.send(Err(ControllerError::input_transport_error( + &self.endpoint_name, + true, + anyhow!("failed to initialize PostgresStore: {e}"), + ))); + return; + } + }; + + // Spawn the completion watcher background task if tracking is available. + // The channel was created in new(); we take the receiver here. + let completion_handle = match ( + self.consumer.completion_watcher(), + self.completion_task_rx.lock().unwrap().take(), + ) { + (Some(watcher), Some(rx)) => Some(tokio::spawn(completion_watcher_task( + watcher, + rx, + self.endpoint_name.clone(), + ))), + _ => None, + }; + + let pending_senders = if self.completion_rx.is_some() { + Some(Arc::clone(&self.pending_senders)) + } else { + None + }; + + let destination = FelderaDestination { + input_stream: Arc::new(Mutex::new(input_stream)), + queue: Arc::clone(&self.queue), + source_table: self.config.source_table.clone(), + endpoint_name: self.endpoint_name.clone(), + relation_cache: Arc::new(Mutex::new(HashMap::new())), + pending_senders, + }; + + let mut pipeline = Pipeline::new(pipeline_config, store, destination); + + match pipeline.start().await { + Ok(()) => { + info!( + "postgres_cdc {}: etl pipeline started for publication '{}', table '{}'", + &self.endpoint_name, &self.config.publication, &self.config.source_table, + ); + let _ = init_status_sender.send(Ok(())); + } + Err(e) => { + let _ = init_status_sender.send(Err(ControllerError::input_transport_error( + &self.endpoint_name, + true, + anyhow!("failed to start etl pipeline: {e}"), + ))); + return; + } + } + + wait_running(&mut receiver).await; + + if let Err(e) = pipeline.wait().await { + error!( + "postgres_cdc {}: etl pipeline error: {e}", + &self.endpoint_name + ); + self.consumer.error(true, anyhow!(e), None); + } + + // Shut down the completion watcher task. + if let Some(handle) = completion_handle { + handle.abort(); + let _ = handle.await; + } + + self.consumer.eoi(); + } +} + +/// Relation metadata cached from WAL Relation events. +#[derive(Clone, Debug)] +struct RelationInfo { + table_name: String, + schema_name: String, + column_names: Vec, +} + +/// etl Destination implementation that pushes data into a Feldera DeCollectionStream. +#[derive(Clone)] +struct FelderaDestination { + input_stream: Arc>>, + queue: Arc, + source_table: String, + endpoint_name: String, + relation_cache: Arc>>, + /// Deferred async result senders. If `Some`, write_events stores senders here + /// instead of firing them immediately. The Queue handler picks them up. + pending_senders: Option>>, +} + +impl Destination for FelderaDestination { + fn name() -> &'static str { + "feldera" + } + + async fn truncate_table( + &self, + table_id: TableId, + async_result: TruncateTableResult<()>, + ) -> EtlResult<()> { + warn!( + "postgres_cdc {}: truncate_table called for table_id={}, ignoring", + &self.endpoint_name, table_id + ); + async_result.send(Ok(())); + Ok(()) + } + + async fn write_table_rows( + &self, + table_id: TableId, + table_rows: Vec, + async_result: WriteTableRowsResult<()>, + ) -> EtlResult<()> { + let is_target = { + let cache = self.relation_cache.lock().unwrap(); + if let Some(info) = cache.get(&u32::from(table_id)) { + self.is_target_table(&info.schema_name, &info.table_name) + } else { + // During snapshot, we may not have relation info yet. + // Accept rows and let the data flow through. + true + } + }; + + if !is_target { + return Ok(()); + } + + let column_names: Vec = { + let cache = self.relation_cache.lock().unwrap(); + cache + .get(&u32::from(table_id)) + .map(|info| info.column_names.clone()) + .unwrap_or_else(|| { + // During snapshot without relation info, generate fallback names + // based on the first row's column count. + let n = table_rows.first().map(|r| r.values().len()).unwrap_or(0); + (0..n).map(|i| format!("col_{i}")).collect() + }) + }; + + let mut stream = self.input_stream.lock().unwrap(); + let mut bytes = 0; + let mut errors = Vec::new(); + let timestamp = Utc::now(); + + for row in &table_rows { + let cells = row.values(); + let json_value = row_to_json(cells, &column_names); + + let json_str = json_value.to_string(); + if let Err(e) = stream.insert(json_str.as_bytes(), &None) { + errors.push(ParseError::text_event_error( + "Failed to deserialize CDC snapshot row", + e, + 0, + Some(&json_str), + None, + )); + } + bytes += json_str.len(); + + if bytes >= 2 * 1024 * 1024 { + self.queue.push((stream.take_all(), errors), timestamp); + bytes = 0; + errors = Vec::new(); + } + } + + if bytes > 0 || !errors.is_empty() { + self.queue.push((stream.take_all(), errors), timestamp); + } + + async_result.send(Ok(())); + Ok(()) + } + + async fn write_events( + &self, + events: Vec, + async_result: WriteEventsResult<()>, + ) -> EtlResult<()> { + let mut stream = self.input_stream.lock().unwrap(); + let mut bytes = 0; + let mut errors = Vec::new(); + let timestamp = Utc::now(); + + for event in &events { + match event { + Event::Relation(rel) => { + let table_schema = &rel.table_schema; + let info = RelationInfo { + table_name: table_schema.name.name.clone(), + schema_name: table_schema.name.schema.clone(), + column_names: table_schema + .column_schemas + .iter() + .map(|c| c.name.clone()) + .collect(), + }; + debug!( + "postgres_cdc {}: relation event for {}.{} (id={})", + &self.endpoint_name, &info.schema_name, &info.table_name, table_schema.id, + ); + self.relation_cache + .lock() + .unwrap() + .insert(u32::from(table_schema.id), info); + } + Event::Insert(insert) => { + if !self.is_target_table_by_id(u32::from(insert.table_id)) { + continue; + } + if let Some(cols) = self.get_column_names(u32::from(insert.table_id)) { + let json_value = row_to_json(insert.table_row.values(), &cols); + let json_str = json_value.to_string(); + if let Err(e) = stream.insert(json_str.as_bytes(), &None) { + errors.push(ParseError::text_event_error( + "Failed to deserialize CDC insert", + e, + 0, + Some(&json_str), + None, + )); + } + bytes += json_str.len(); + } + } + Event::Update(update) => { + if !self.is_target_table_by_id(u32::from(update.table_id)) { + continue; + } + if let Some(cols) = self.get_column_names(u32::from(update.table_id)) { + // Delete old row if available + if let Some((_full, old_row)) = &update.old_table_row { + let old_json = row_to_json(old_row.values(), &cols); + let old_str = old_json.to_string(); + if let Err(e) = stream.delete(old_str.as_bytes(), &None) { + errors.push(ParseError::text_event_error( + "Failed to deserialize CDC update (old)", + e, + 0, + Some(&old_str), + None, + )); + } + bytes += old_str.len(); + } + // Insert new row + let new_json = row_to_json(update.table_row.values(), &cols); + let new_str = new_json.to_string(); + if let Err(e) = stream.insert(new_str.as_bytes(), &None) { + errors.push(ParseError::text_event_error( + "Failed to deserialize CDC update (new)", + e, + 0, + Some(&new_str), + None, + )); + } + bytes += new_str.len(); + } + } + Event::Delete(delete) => { + if !self.is_target_table_by_id(u32::from(delete.table_id)) { + continue; + } + if let Some(cols) = self.get_column_names(u32::from(delete.table_id)) + && let Some((_full, old_row)) = &delete.old_table_row + { + let old_json = row_to_json(old_row.values(), &cols); + let old_str = old_json.to_string(); + if let Err(e) = stream.delete(old_str.as_bytes(), &None) { + errors.push(ParseError::text_event_error( + "Failed to deserialize CDC delete", + e, + 0, + Some(&old_str), + None, + )); + } + bytes += old_str.len(); + } + } + Event::Truncate(_) => { + warn!( + "postgres_cdc {}: received TRUNCATE event, ignoring", + &self.endpoint_name + ); + } + Event::Begin(_) | Event::Commit(_) | Event::Unsupported => {} + } + + if bytes >= 2 * 1024 * 1024 { + self.queue.push((stream.take_all(), errors), timestamp); + bytes = 0; + errors = Vec::new(); + } + } + + if bytes > 0 || !errors.is_empty() { + self.queue.push((stream.take_all(), errors), timestamp); + } + + // Defer or fire the async result. + if let Some(ref pending) = self.pending_senders { + pending.lock().unwrap().push(async_result); + } else { + async_result.send(Ok(())); + } + + Ok(()) + } +} + +impl FelderaDestination { + fn is_target_table(&self, schema_name: &str, table_name: &str) -> bool { + let qualified = format!("{schema_name}.{table_name}"); + self.source_table == qualified + || self.source_table == table_name + || self.source_table == format!("\"{schema_name}\".\"{table_name}\"") + } + + fn is_target_table_by_id(&self, table_id: u32) -> bool { + let cache = self.relation_cache.lock().unwrap(); + if let Some(info) = cache.get(&table_id) { + self.is_target_table(&info.schema_name, &info.table_name) + } else { + false + } + } + + fn get_column_names(&self, table_id: u32) -> Option> { + let cache = self.relation_cache.lock().unwrap(); + cache.get(&table_id).map(|info| info.column_names.clone()) + } +} + +/// Convert a row of cells to a JSON object using the given column names. +fn row_to_json(cells: &[Cell], column_names: &[String]) -> Value { + let mut map = serde_json::Map::new(); + for (i, cell) in cells.iter().enumerate() { + let col_name = column_names + .get(i) + .cloned() + .unwrap_or_else(|| format!("col_{i}")); + map.insert(col_name, cell_to_json(cell)); + } + Value::Object(map) +} + +/// Convert an etl Cell to a serde_json Value. +fn cell_to_json(cell: &Cell) -> Value { + match cell { + Cell::Null => Value::Null, + Cell::Bool(b) => json!(b), + Cell::String(s) => json!(s), + Cell::I16(n) => json!(n), + Cell::I32(n) => json!(n), + Cell::U32(n) => json!(n), + Cell::I64(n) => json!(n), + Cell::F32(f) => { + if f.is_nan() || f.is_infinite() { + Value::Null + } else { + json!(f) + } + } + Cell::F64(f) => { + if f.is_nan() || f.is_infinite() { + Value::Null + } else { + json!(f) + } + } + Cell::Numeric(n) => { + // Preserve precision by encoding as string. + json!(n.to_string()) + } + Cell::Date(d) => json!(d.to_string()), + Cell::Time(t) => json!(t.to_string()), + Cell::Timestamp(ts) => json!(ts.format("%Y-%m-%dT%H:%M:%S%.f").to_string()), + Cell::TimestampTz(ts) => json!(ts.to_rfc3339()), + Cell::Uuid(u) => json!(u.to_string()), + Cell::Json(j) => j.clone(), + Cell::Bytes(b) => { + // Encode as byte array to match Datagen's BinaryFormat::Array. + json!(b) + } + Cell::Array(arr) => array_cell_to_json(arr), + } +} + +/// Convert an etl ArrayCell to a JSON array. +fn array_cell_to_json(arr: &ArrayCell) -> Value { + match arr { + ArrayCell::Bool(v) => json!(v), + ArrayCell::String(v) => json!(v), + ArrayCell::I16(v) => json!(v), + ArrayCell::I32(v) => json!(v), + ArrayCell::U32(v) => json!(v), + ArrayCell::I64(v) => json!(v), + ArrayCell::F32(v) => { + let vals: Vec = v + .iter() + .map(|opt| match opt { + Some(f) if f.is_nan() || f.is_infinite() => Value::Null, + Some(f) => json!(f), + None => Value::Null, + }) + .collect(); + Value::Array(vals) + } + ArrayCell::F64(v) => { + let vals: Vec = v + .iter() + .map(|opt| match opt { + Some(f) if f.is_nan() || f.is_infinite() => Value::Null, + Some(f) => json!(f), + None => Value::Null, + }) + .collect(); + Value::Array(vals) + } + ArrayCell::Numeric(v) => { + let vals: Vec = v + .iter() + .map(|opt| match opt { + Some(n) => json!(n.to_string()), + None => Value::Null, + }) + .collect(); + Value::Array(vals) + } + ArrayCell::Date(v) => { + let vals: Vec = v + .iter() + .map(|opt| match opt { + Some(d) => json!(d.to_string()), + None => Value::Null, + }) + .collect(); + Value::Array(vals) + } + ArrayCell::Time(v) => { + let vals: Vec = v + .iter() + .map(|opt| match opt { + Some(t) => json!(t.to_string()), + None => Value::Null, + }) + .collect(); + Value::Array(vals) + } + ArrayCell::Timestamp(v) => { + let vals: Vec = v + .iter() + .map(|opt| match opt { + Some(ts) => json!(ts.format("%Y-%m-%dT%H:%M:%S%.f").to_string()), + None => Value::Null, + }) + .collect(); + Value::Array(vals) + } + ArrayCell::TimestampTz(v) => { + let vals: Vec = v + .iter() + .map(|opt| match opt { + Some(ts) => json!(ts.to_rfc3339()), + None => Value::Null, + }) + .collect(); + Value::Array(vals) + } + ArrayCell::Uuid(v) => { + let vals: Vec = v + .iter() + .map(|opt| match opt { + Some(u) => json!(u.to_string()), + None => Value::Null, + }) + .collect(); + Value::Array(vals) + } + ArrayCell::Json(v) => { + let vals: Vec = v + .iter() + .map(|opt| match opt { + Some(j) => j.clone(), + None => Value::Null, + }) + .collect(); + Value::Array(vals) + } + ArrayCell::Bytes(v) => { + let vals: Vec = v + .iter() + .map(|opt| match opt { + Some(b) => json!(b), + None => Value::Null, + }) + .collect(); + Value::Array(vals) + } + } +} + +/// Background task that watches for Feldera step completion and fires deferred +/// ETL async result senders when the step containing their data has completed. +/// +/// Each entry is `(completed_at_flush, senders)` where `completed_at_flush` is +/// the value of `total_completed_steps` at the time the data was flushed to the +/// circuit. The data is processed once `total_completed_steps > completed_at_flush`. +async fn completion_watcher_task( + mut completion_rx: tokio::sync::watch::Receiver, + mut pending_rx: mpsc::UnboundedReceiver<(u64, DeferredSenders)>, + endpoint_name: String, +) { + let mut waiting: Vec<(u64, DeferredSenders)> = Vec::new(); + + loop { + tokio::select! { + result = completion_rx.changed() => { + if result.is_err() { + break; // Sender dropped (pipeline shutting down) + } + let completed = completion_rx.borrow().total_completed_steps; + fire_completed(&mut waiting, completed); + } + maybe_entry = pending_rx.recv() => { + match maybe_entry { + Some((step_at_flush, senders)) => { + let completed = completion_rx.borrow().total_completed_steps; + if completed > step_at_flush { + // Already complete — fire immediately. + for sender in senders { + sender.send(Ok(())); + } + } else { + waiting.push((step_at_flush, senders)); + } + } + None => break, // Channel closed + } + } + } + } + + // On shutdown, remaining senders are dropped. AsyncResult's Drop impl + // sends an error to the ETL side, causing it to shut down gracefully. + debug!( + "postgres_cdc {endpoint_name}: completion watcher exiting with {} pending entries", + waiting.len() + ); +} + +/// Fires deferred senders whose data has been fully processed. +fn fire_completed(waiting: &mut Vec<(u64, DeferredSenders)>, completed_steps: u64) { + waiting.retain_mut(|(step_at_flush, senders)| { + if completed_steps > *step_at_flush { + for sender in senders.drain(..) { + sender.send(Ok(())); + } + false + } else { + true + } + }); +} + +/// Build a stable identity string for pipeline_id hashing. +/// +/// Extracts host/port/database from the URI (excludes password, username, +/// and query parameters) combined with publication and source_table. +/// Falls back to the raw URI if parsing fails — the pipeline will likely +/// fail startup shortly after anyway with a clearer error. +fn stable_connection_identity(uri: &str, publication: &str, source_table: &str) -> String { + let (host, port, db) = match Url::parse(uri) { + Ok(url) => { + let host = url.host_str().unwrap_or("").to_string(); + let port = url.port().unwrap_or(5432); + let db = url.path().trim_start_matches('/').to_string(); + (host, port, db) + } + Err(_) => return format!("{uri}\0{publication}\0{source_table}"), + }; + format!("{host}:{port}/{db}\0{publication}\0{source_table}") +} + +/// Parse a Postgres URI into etl's PgConnectionConfig. +fn parse_pg_uri(uri: &str) -> AnyResult { + let url = Url::parse(uri)?; + + let host = url + .host_str() + .ok_or_else(|| anyhow!("missing host in URI"))? + .to_string(); + let port = url.port().unwrap_or(5432); + let username = url.username().to_string(); + if username.is_empty() { + return Err(anyhow!("missing username in URI")); + } + let password = url.password().map(|p| p.to_string().into()); + let name = url.path().trim_start_matches('/').to_string(); + if name.is_empty() { + return Err(anyhow!("missing database name in URI")); + } + + Ok(PgConnectionConfig { + host, + port, + name, + username, + password, + tls: TlsConfig::disabled(), + keepalive: TcpKeepaliveConfig::default(), + }) +} + +/// Block until the state is `Running`. +async fn wait_running(receiver: &mut Receiver) { + let _ = receiver + .wait_for(|state| state == &PipelineState::Running) + .await; +} + +#[cfg(test)] +mod tests { + use super::*; + use chrono::{NaiveDate, NaiveDateTime, NaiveTime, TimeZone, Utc}; + use etl::types::PgNumeric; + use serde_json::json; + use std::str::FromStr; + + // ----------------------------------------------------------------------- + // cell_to_json unit tests + // ----------------------------------------------------------------------- + + #[test] + fn test_cell_null() { + assert_eq!(cell_to_json(&Cell::Null), Value::Null); + } + + #[test] + fn test_cell_bool() { + assert_eq!(cell_to_json(&Cell::Bool(true)), json!(true)); + assert_eq!(cell_to_json(&Cell::Bool(false)), json!(false)); + } + + #[test] + fn test_cell_string() { + assert_eq!(cell_to_json(&Cell::String("hello".into())), json!("hello")); + assert_eq!(cell_to_json(&Cell::String("".into())), json!("")); + // Unicode + assert_eq!( + cell_to_json(&Cell::String("caf\u{00e9}".into())), + json!("caf\u{00e9}") + ); + } + + #[test] + fn test_cell_integers() { + assert_eq!(cell_to_json(&Cell::I16(42)), json!(42)); + assert_eq!(cell_to_json(&Cell::I16(-1)), json!(-1)); + assert_eq!(cell_to_json(&Cell::I32(100_000)), json!(100_000)); + assert_eq!( + cell_to_json(&Cell::U32(4_000_000_000)), + json!(4_000_000_000u64) + ); + assert_eq!(cell_to_json(&Cell::I64(i64::MAX)), json!(i64::MAX)); + assert_eq!(cell_to_json(&Cell::I64(i64::MIN)), json!(i64::MIN)); + } + + #[test] + fn test_cell_f32() { + assert_eq!(cell_to_json(&Cell::F32(3.14)), json!(3.14f32)); + // NaN and infinity produce null + assert_eq!(cell_to_json(&Cell::F32(f32::NAN)), Value::Null); + assert_eq!(cell_to_json(&Cell::F32(f32::INFINITY)), Value::Null); + assert_eq!(cell_to_json(&Cell::F32(f32::NEG_INFINITY)), Value::Null); + } + + #[test] + fn test_cell_f64() { + assert_eq!(cell_to_json(&Cell::F64(2.718)), json!(2.718f64)); + assert_eq!(cell_to_json(&Cell::F64(f64::NAN)), Value::Null); + assert_eq!(cell_to_json(&Cell::F64(f64::INFINITY)), Value::Null); + assert_eq!(cell_to_json(&Cell::F64(f64::NEG_INFINITY)), Value::Null); + } + + #[test] + fn test_cell_numeric() { + let n = PgNumeric::from_str("123.456").unwrap(); + let v = cell_to_json(&Cell::Numeric(n)); + assert_eq!(v, json!("123.456")); + } + + #[test] + fn test_cell_date() { + let d = NaiveDate::from_ymd_opt(2024, 6, 15).unwrap(); + let v = cell_to_json(&Cell::Date(d)); + assert_eq!(v, json!("2024-06-15")); + } + + #[test] + fn test_cell_time() { + let t = NaiveTime::from_hms_opt(14, 30, 0).unwrap(); + let v = cell_to_json(&Cell::Time(t)); + assert_eq!(v, json!("14:30:00")); + } + + #[test] + fn test_cell_timestamp() { + let dt = NaiveDateTime::new( + NaiveDate::from_ymd_opt(2024, 1, 1).unwrap(), + NaiveTime::from_hms_opt(12, 0, 0).unwrap(), + ); + let v = cell_to_json(&Cell::Timestamp(dt)); + assert_eq!(v, json!("2024-01-01T12:00:00")); + } + + #[test] + fn test_cell_timestamptz() { + let dt = Utc.with_ymd_and_hms(2024, 1, 1, 12, 0, 0).unwrap(); + let v = cell_to_json(&Cell::TimestampTz(dt)); + // RFC 3339 format + assert_eq!(v, json!("2024-01-01T12:00:00+00:00")); + } + + #[test] + fn test_cell_uuid() { + let u = uuid::Uuid::parse_str("550e8400-e29b-41d4-a716-446655440000").unwrap(); + let v = cell_to_json(&Cell::Uuid(u)); + assert_eq!(v, json!("550e8400-e29b-41d4-a716-446655440000")); + } + + #[test] + fn test_cell_json() { + let j = json!({"key": "value", "num": 42}); + let v = cell_to_json(&Cell::Json(j.clone())); + assert_eq!(v, j); + } + + #[test] + fn test_cell_bytes() { + let v = cell_to_json(&Cell::Bytes(vec![0xde, 0xad, 0xbe, 0xef])); + assert_eq!(v, json!([0xde, 0xad, 0xbe, 0xef])); + } + + #[test] + fn test_cell_bytes_empty() { + let v = cell_to_json(&Cell::Bytes(vec![])); + assert_eq!(v, json!([])); + } + + // ----------------------------------------------------------------------- + // array_cell_to_json unit tests + // ----------------------------------------------------------------------- + + #[test] + fn test_array_bool() { + let arr = ArrayCell::Bool(vec![Some(true), Some(false), None]); + let v = array_cell_to_json(&arr); + assert_eq!(v, json!([true, false, null])); + } + + #[test] + fn test_array_string() { + let arr = ArrayCell::String(vec![Some("a".into()), None, Some("b".into())]); + let v = array_cell_to_json(&arr); + assert_eq!(v, json!(["a", null, "b"])); + } + + #[test] + fn test_array_i16() { + let arr = ArrayCell::I16(vec![Some(1), Some(-2), None]); + let v = array_cell_to_json(&arr); + assert_eq!(v, json!([1, -2, null])); + } + + #[test] + fn test_array_i32() { + let arr = ArrayCell::I32(vec![Some(100), None]); + let v = array_cell_to_json(&arr); + assert_eq!(v, json!([100, null])); + } + + #[test] + fn test_array_i64() { + let arr = ArrayCell::I64(vec![Some(i64::MAX), None, Some(0)]); + let v = array_cell_to_json(&arr); + assert_eq!(v, json!([i64::MAX, null, 0])); + } + + #[test] + fn test_array_f32_with_special() { + let arr = ArrayCell::F32(vec![Some(1.5), None, Some(f32::NAN), Some(f32::INFINITY)]); + let v = array_cell_to_json(&arr); + assert_eq!(v, json!([1.5f32, null, null, null])); + } + + #[test] + fn test_array_f64_with_special() { + let arr = ArrayCell::F64(vec![Some(2.5), Some(f64::NEG_INFINITY), None]); + let v = array_cell_to_json(&arr); + assert_eq!(v, json!([2.5f64, null, null])); + } + + #[test] + fn test_array_numeric() { + let n = PgNumeric::from_str("99.99").unwrap(); + let arr = ArrayCell::Numeric(vec![Some(n), None]); + let v = array_cell_to_json(&arr); + assert_eq!(v, json!(["99.99", null])); + } + + #[test] + fn test_array_date() { + let d = NaiveDate::from_ymd_opt(2024, 12, 25).unwrap(); + let arr = ArrayCell::Date(vec![Some(d), None]); + let v = array_cell_to_json(&arr); + assert_eq!(v, json!(["2024-12-25", null])); + } + + #[test] + fn test_array_time() { + let t = NaiveTime::from_hms_opt(8, 30, 0).unwrap(); + let arr = ArrayCell::Time(vec![Some(t), None]); + let v = array_cell_to_json(&arr); + assert_eq!(v, json!(["08:30:00", null])); + } + + #[test] + fn test_array_timestamp() { + let dt = NaiveDateTime::new( + NaiveDate::from_ymd_opt(2024, 6, 1).unwrap(), + NaiveTime::from_hms_opt(0, 0, 0).unwrap(), + ); + let arr = ArrayCell::Timestamp(vec![Some(dt), None]); + let v = array_cell_to_json(&arr); + assert_eq!(v, json!(["2024-06-01T00:00:00", null])); + } + + #[test] + fn test_array_timestamptz() { + let dt = Utc.with_ymd_and_hms(2024, 6, 1, 0, 0, 0).unwrap(); + let arr = ArrayCell::TimestampTz(vec![Some(dt), None]); + let v = array_cell_to_json(&arr); + assert_eq!(v, json!(["2024-06-01T00:00:00+00:00", null])); + } + + #[test] + fn test_array_uuid() { + let u = uuid::Uuid::parse_str("12345678-1234-1234-1234-123456789abc").unwrap(); + let arr = ArrayCell::Uuid(vec![Some(u), None]); + let v = array_cell_to_json(&arr); + assert_eq!(v, json!(["12345678-1234-1234-1234-123456789abc", null])); + } + + #[test] + fn test_array_json() { + let j = json!({"a": 1}); + let arr = ArrayCell::Json(vec![Some(j.clone()), None]); + let v = array_cell_to_json(&arr); + assert_eq!(v, json!([{"a": 1}, null])); + } + + #[test] + fn test_array_bytes() { + let arr = ArrayCell::Bytes(vec![Some(vec![0xca, 0xfe]), None, Some(vec![])]); + let v = array_cell_to_json(&arr); + assert_eq!(v, json!([[0xca, 0xfe], null, []])); + } + + // ----------------------------------------------------------------------- + // row_to_json unit tests + // ----------------------------------------------------------------------- + + #[test] + fn test_row_to_json_basic() { + let cells = vec![Cell::I32(1), Cell::String("hello".into()), Cell::Bool(true)]; + let cols = vec!["id".into(), "name".into(), "active".into()]; + let v = row_to_json(&cells, &cols); + assert_eq!(v, json!({"id": 1, "name": "hello", "active": true})); + } + + #[test] + fn test_row_to_json_with_null() { + let cells = vec![Cell::I32(42), Cell::Null]; + let cols = vec!["id".into(), "value".into()]; + let v = row_to_json(&cells, &cols); + assert_eq!(v, json!({"id": 42, "value": null})); + } + + #[test] + fn test_row_to_json_more_cells_than_columns() { + // Extra cells get auto-generated column names + let cells = vec![Cell::I32(1), Cell::I32(2), Cell::I32(3)]; + let cols = vec!["a".into(), "b".into()]; + let v = row_to_json(&cells, &cols); + assert_eq!(v, json!({"a": 1, "b": 2, "col_2": 3})); + } + + #[test] + fn test_row_to_json_all_types() { + let d = NaiveDate::from_ymd_opt(2024, 3, 15).unwrap(); + let t = NaiveTime::from_hms_opt(10, 30, 0).unwrap(); + let ts = NaiveDateTime::new(d, t); + let tstz = Utc.with_ymd_and_hms(2024, 3, 15, 10, 30, 0).unwrap(); + let u = uuid::Uuid::parse_str("550e8400-e29b-41d4-a716-446655440000").unwrap(); + let n = PgNumeric::from_str("123.45").unwrap(); + + let cells = vec![ + Cell::Bool(true), + Cell::I16(16), + Cell::I32(32), + Cell::U32(4_000_000_000), + Cell::I64(64), + Cell::F32(1.5), + Cell::F64(2.5), + Cell::Numeric(n), + Cell::String("text".into()), + Cell::Date(d), + Cell::Time(t), + Cell::Timestamp(ts), + Cell::TimestampTz(tstz), + Cell::Uuid(u), + Cell::Json(json!({"key": "val"})), + Cell::Bytes(vec![0xab, 0xcd]), + Cell::Null, + Cell::Array(ArrayCell::I32(vec![Some(1), Some(2), None])), + ]; + let cols: Vec = vec![ + "bool_col", + "i16_col", + "i32_col", + "u32_col", + "i64_col", + "f32_col", + "f64_col", + "numeric_col", + "text_col", + "date_col", + "time_col", + "ts_col", + "tstz_col", + "uuid_col", + "json_col", + "bytes_col", + "null_col", + "arr_col", + ] + .into_iter() + .map(String::from) + .collect(); + + let v = row_to_json(&cells, &cols); + assert_eq!(v["bool_col"], json!(true)); + assert_eq!(v["i16_col"], json!(16)); + assert_eq!(v["i32_col"], json!(32)); + assert_eq!(v["u32_col"], json!(4_000_000_000u64)); + assert_eq!(v["i64_col"], json!(64)); + assert_eq!(v["f32_col"], json!(1.5f32)); + assert_eq!(v["f64_col"], json!(2.5f64)); + assert_eq!(v["numeric_col"], json!("123.45")); + assert_eq!(v["text_col"], json!("text")); + assert_eq!(v["date_col"], json!("2024-03-15")); + assert_eq!(v["time_col"], json!("10:30:00")); + assert_eq!(v["ts_col"], json!("2024-03-15T10:30:00")); + assert_eq!(v["tstz_col"], json!("2024-03-15T10:30:00+00:00")); + assert_eq!(v["uuid_col"], json!("550e8400-e29b-41d4-a716-446655440000")); + assert_eq!(v["json_col"], json!({"key": "val"})); + assert_eq!(v["bytes_col"], json!([0xab, 0xcd])); + assert_eq!(v["null_col"], Value::Null); + assert_eq!(v["arr_col"], json!([1, 2, null])); + } + + // ----------------------------------------------------------------------- + // stable_connection_identity unit tests + // ----------------------------------------------------------------------- + + #[test] + fn test_stable_identity_excludes_password() { + let id1 = stable_connection_identity( + "postgres://user:oldpass@localhost:5432/db", + "pub", + "public.tbl", + ); + let id2 = stable_connection_identity( + "postgres://user:newpass@localhost:5432/db", + "pub", + "public.tbl", + ); + assert_eq!( + id1, id2, + "rotating the password should not change the stable identity" + ); + } + + #[test] + fn test_stable_identity_excludes_username() { + let id1 = + stable_connection_identity("postgres://alice@localhost:5432/db", "pub", "public.tbl"); + let id2 = + stable_connection_identity("postgres://bob@localhost:5432/db", "pub", "public.tbl"); + assert_eq!(id1, id2); + } + + #[test] + fn test_stable_identity_changes_with_host() { + let id1 = stable_connection_identity("postgres://u:p@host1:5432/db", "pub", "public.tbl"); + let id2 = stable_connection_identity("postgres://u:p@host2:5432/db", "pub", "public.tbl"); + assert_ne!(id1, id2); + } + + #[test] + fn test_stable_identity_changes_with_publication() { + let id1 = stable_connection_identity("postgres://u:p@host:5432/db", "pub1", "public.tbl"); + let id2 = stable_connection_identity("postgres://u:p@host:5432/db", "pub2", "public.tbl"); + assert_ne!(id1, id2); + } + + #[test] + fn test_stable_identity_invalid_uri_falls_back() { + // Parse failure falls back to using the raw URI — identity is still + // deterministic, just less robust to URI format variations. + let id = stable_connection_identity("not a valid uri", "pub", "tbl"); + assert!(id.contains("not a valid uri")); + } + + // ----------------------------------------------------------------------- + // parse_pg_uri unit tests + // ----------------------------------------------------------------------- + + #[test] + fn test_parse_pg_uri_basic() { + let config = parse_pg_uri("postgres://user:pass@localhost:5432/mydb").unwrap(); + assert_eq!(config.host, "localhost"); + assert_eq!(config.port, 5432); + assert_eq!(config.username, "user"); + assert!(config.password.is_some()); + assert_eq!(config.name, "mydb"); + } + + #[test] + fn test_parse_pg_uri_default_port() { + let config = parse_pg_uri("postgres://user:pass@host.example.com/testdb").unwrap(); + assert_eq!(config.port, 5432); + assert_eq!(config.host, "host.example.com"); + } + + #[test] + fn test_parse_pg_uri_no_password() { + let config = parse_pg_uri("postgres://user@localhost/mydb").unwrap(); + assert!(config.password.is_none()); + } + + #[test] + fn test_parse_pg_uri_missing_username() { + let result = parse_pg_uri("postgres://localhost/mydb"); + assert!(result.is_err()); + } + + #[test] + fn test_parse_pg_uri_missing_database() { + let result = parse_pg_uri("postgres://user:pass@localhost"); + assert!(result.is_err()); + } + + #[test] + fn test_parse_pg_uri_custom_port() { + let config = parse_pg_uri("postgres://user:pass@db.host:15432/mydb").unwrap(); + assert_eq!(config.port, 15432); + } + + #[test] + fn test_parse_pg_uri_invalid_scheme() { + let result = parse_pg_uri("not_a_uri"); + assert!(result.is_err()); + } + + // ----------------------------------------------------------------------- + // RelationInfo / target table matching tests (direct struct construction) + // ----------------------------------------------------------------------- + + #[test] + fn test_relation_info_clone() { + let info = RelationInfo { + table_name: "orders".to_string(), + schema_name: "public".to_string(), + column_names: vec!["id".to_string(), "name".to_string()], + }; + let cloned = info.clone(); + assert_eq!(cloned.table_name, "orders"); + assert_eq!(cloned.schema_name, "public"); + assert_eq!(cloned.column_names.len(), 2); + } + + /// Test the is_target_table logic extracted for direct verification. + /// This mirrors FelderaDestination::is_target_table without needing to + /// construct the full struct. + fn target_table_matches(source_table: &str, schema_name: &str, table_name: &str) -> bool { + let qualified = format!("{schema_name}.{table_name}"); + source_table == qualified + || source_table == table_name + || source_table == format!("\"{schema_name}\".\"{table_name}\"") + } + + #[test] + fn test_target_table_unqualified() { + assert!(target_table_matches("orders", "public", "orders")); + assert!(!target_table_matches("orders", "public", "users")); + } + + #[test] + fn test_target_table_qualified() { + assert!(target_table_matches("public.orders", "public", "orders")); + assert!(!target_table_matches("other.orders", "public", "orders")); + } + + #[test] + fn test_target_table_quoted() { + assert!(target_table_matches( + "\"public\".\"orders\"", + "public", + "orders" + )); + assert!(!target_table_matches( + "\"other\".\"orders\"", + "public", + "orders" + )); + } + + #[test] + fn test_target_table_different_schema() { + assert!(!target_table_matches("myschema.orders", "public", "orders")); + assert!(target_table_matches( + "myschema.orders", + "myschema", + "orders" + )); + } + + #[test] + fn test_relation_cache_lookup() { + let cache: HashMap = HashMap::from([( + 42, + RelationInfo { + table_name: "orders".to_string(), + schema_name: "public".to_string(), + column_names: vec!["id".to_string(), "amount".to_string()], + }, + )]); + // Simulate is_target_table_by_id + let source_table = "public.orders"; + let info42 = cache.get(&42).unwrap(); + assert!(target_table_matches( + source_table, + &info42.schema_name, + &info42.table_name, + )); + assert!(cache.get(&99).is_none()); + + // Simulate get_column_names + let cols = cache.get(&42).map(|info| info.column_names.clone()); + assert_eq!(cols, Some(vec!["id".to_string(), "amount".to_string()])); + assert_eq!(cache.get(&99).map(|info| info.column_names.clone()), None); + } +} diff --git a/crates/adapters/src/integrated/postgres/test.rs b/crates/adapters/src/integrated/postgres/test.rs index 733595bd901..25ba7adc362 100644 --- a/crates/adapters/src/integrated/postgres/test.rs +++ b/crates/adapters/src/integrated/postgres/test.rs @@ -2,7 +2,7 @@ use dbsp::{Runtime, utils::Tup1}; use feldera_sqllib::Variant; use feldera_types::{ deserialize_table_record, - program_schema::{Relation, SqlIdentifier}, + program_schema::{ColumnType, Field, Relation, SqlIdentifier}, serde_with_context::{SerializeWithContext, SqlSerdeConfig}, serialize_table_record, transport::postgres::{PostgresTlsConfig, PostgresWriteMode}, @@ -2114,3 +2114,976 @@ fn test_pg_input_tls() { .execute(&format!("DROP TABLE {table_name}"), &[]) .unwrap(); } + +// =================================================================== +// Postgres CDC input connector integration tests +// +// These tests require: +// - wal_level=logical on the Postgres instance +// - The connecting user must have REPLICATION privilege +// - The `with-postgres-cdc` feature must be enabled +// +// They are gated behind #[ignore] so they don't run in normal CI. +// Run with: cargo test -p dbsp_adapters --features with-postgres-cdc -- --ignored +// =================================================================== + +#[cfg(feature = "with-postgres-cdc")] +mod cdc_tests { + use super::*; + use crate::test::wait; + use pg::pg_connect; + + /// Helper: creates a table, publication, and sets REPLICA IDENTITY FULL. + /// Returns a connected client for further DML operations. + /// On drop, cleans up the publication and table. + struct CdcTestTable { + client: postgres::Client, + table_name: String, + publication_name: String, + } + + impl CdcTestTable { + fn new_simple(table_name: &str, publication_name: &str, url: &str) -> Self { + let mut client = pg_connect(url, &None); + + // Clean up any leftover objects from previous runs. + let _ = client.execute( + &format!("DROP PUBLICATION IF EXISTS {publication_name}"), + &[], + ); + let _ = client.execute(&format!("DROP TABLE IF EXISTS {table_name}"), &[]); + + client + .execute( + &format!( + r#"CREATE TABLE {table_name} ( + id INTEGER PRIMARY KEY, + b BOOLEAN NOT NULL, + i BIGINT, + s VARCHAR NOT NULL +)"# + ), + &[], + ) + .expect("failed to create CDC test table"); + + client + .execute( + &format!("ALTER TABLE {table_name} REPLICA IDENTITY FULL"), + &[], + ) + .expect("failed to set REPLICA IDENTITY FULL"); + + client + .execute( + &format!("CREATE PUBLICATION {publication_name} FOR TABLE {table_name}"), + &[], + ) + .expect("failed to create publication"); + + CdcTestTable { + client, + table_name: table_name.to_string(), + publication_name: publication_name.to_string(), + } + } + + /// Creates a table with many Postgres types for data type coverage testing. + fn new_all_types(table_name: &str, publication_name: &str, url: &str) -> Self { + let mut client = pg_connect(url, &None); + + let _ = client.execute( + &format!("DROP PUBLICATION IF EXISTS {publication_name}"), + &[], + ); + let _ = client.execute(&format!("DROP TABLE IF EXISTS {table_name}"), &[]); + + client + .execute( + &format!( + r#"CREATE TABLE {table_name} ( + id SERIAL PRIMARY KEY, + col_text TEXT, + col_integer INTEGER, + col_bigint BIGINT, + col_boolean BOOLEAN, + col_real REAL, + col_double DOUBLE PRECISION, + col_date DATE, + col_time TIME, + col_timestamp TIMESTAMP, + col_timestamptz TIMESTAMPTZ, + col_uuid UUID, + col_jsonb JSONB, + col_bytea BYTEA, + col_numeric NUMERIC(10,2), + col_smallint SMALLINT, + col_int_array INTEGER[] +)"# + ), + &[], + ) + .expect("failed to create CDC all-types test table"); + + client + .execute( + &format!("ALTER TABLE {table_name} REPLICA IDENTITY FULL"), + &[], + ) + .expect("failed to set REPLICA IDENTITY FULL"); + + client + .execute( + &format!("CREATE PUBLICATION {publication_name} FOR TABLE {table_name}"), + &[], + ) + .expect("failed to create publication"); + + CdcTestTable { + client, + table_name: table_name.to_string(), + publication_name: publication_name.to_string(), + } + } + + fn execute(&mut self, query: &str) { + self.client + .execute(query, &[]) + .unwrap_or_else(|e| panic!("failed to execute '{query}': {e}")); + } + } + + impl Drop for CdcTestTable { + fn drop(&mut self) { + // Drop replication slots that etl may have created for this publication. + // etl creates slots with names based on the publication name. + let slots: Vec = self + .client + .query( + "SELECT slot_name FROM pg_replication_slots WHERE slot_name LIKE $1", + &[&format!("%{}%", &self.publication_name)], + ) + .unwrap_or_default() + .into_iter() + .map(|r| r.get::<_, String>("slot_name")) + .collect(); + for slot in slots { + let _ = self + .client + .execute(&format!("SELECT pg_drop_replication_slot('{slot}')"), &[]); + } + let _ = self.client.execute( + &format!("DROP PUBLICATION IF EXISTS {}", self.publication_name), + &[], + ); + let _ = self + .client + .execute(&format!("DROP TABLE IF EXISTS {}", self.table_name), &[]); + } + } + + /// Build a test circuit that reads from CDC input and writes to a file output. + /// The input schema matches the "simple" table (id, b, i, s). + fn cdc_simple_test_circuit( + url: &str, + publication: &str, + source_table: &str, + output_path: &Path, + ) -> (Controller, crossbeam::channel::Receiver) { + let schema = TestStruct::schema(); + let config = serde_json::from_value(json!({ + "name": "cdc_test", + "workers": 1, + "inputs": { + "cdc_in": { + "stream": "test_input1", + "transport": { + "name": "postgres_cdc_input", + "config": { + "uri": url, + "publication": publication, + "source_table": source_table, + }, + }, + }, + }, + "outputs": { + "test_output1": { + "stream": "test_output1", + "transport": { + "name": "file_output", + "config": { + "path": output_path, + } + }, + "format": { + "name": "json", + "config": { + "update_format": "insert_delete", + "array": false, + } + } + } + } + })) + .unwrap(); + + let (err_sender, err_receiver) = crossbeam::channel::unbounded(); + let controller = Controller::with_test_config( + move |workers| { + Ok({ + let (circuit, catalog) = Runtime::init_circuit(workers, move |circuit| { + let mut catalog = Catalog::new(); + let (input, hinput) = circuit.add_input_zset::(); + + let input_schema = serde_json::to_string(&Relation::new( + "test_input1".into(), + schema.clone(), + false, + BTreeMap::new(), + )) + .unwrap(); + + let output_schema = serde_json::to_string(&Relation::new( + "test_output1".into(), + schema, + false, + BTreeMap::new(), + )) + .unwrap(); + + catalog.register_materialized_input_zset::<_, TestStruct>( + input.clone(), + hinput, + &input_schema, + ); + + catalog.register_materialized_output_zset::<_, TestStruct>( + input, + &output_schema, + ); + + Ok(catalog) + }) + .unwrap(); + (circuit, Box::new(catalog)) + }) + }, + &config, + Box::new(move |e, _| { + let msg = format!("cdc_test: error: {e}"); + println!("{msg}"); + err_sender.send(msg).unwrap() + }), + ) + .unwrap(); + + (controller, err_receiver) + } + + /// Helper struct for the all-types test. Uses simple JSON deserialization + /// since the CDC connector produces JSON. + #[derive( + Debug, + Default, + PartialEq, + Eq, + PartialOrd, + Ord, + serde::Serialize, + serde::Deserialize, + Clone, + Hash, + size_of::SizeOf, + rkyv::Archive, + rkyv::Serialize, + rkyv::Deserialize, + feldera_macros::IsNone, + )] + #[archive_attr(derive(Ord, Eq, PartialEq, PartialOrd))] + struct CdcAllTypesStruct { + id: i32, + col_text: Option, + col_integer: Option, + col_bigint: Option, + col_boolean: Option, + col_real: Option, + col_double: Option, + col_date: Option, + col_time: Option, + col_timestamp: Option, + col_timestamptz: Option, + col_uuid: Option, + col_jsonb: Option, + col_bytea: Option, + col_numeric: Option, + col_smallint: Option, + col_int_array: Option>>, + } + + feldera_types::deserialize_table_record!(CdcAllTypesStruct["CdcAllTypesStruct", Variant, 17] { + (id, "id", false, i32, |_| None), + (col_text, "col_text", true, Option, |_| Some(None)), + (col_integer, "col_integer", true, Option, |_| Some(None)), + (col_bigint, "col_bigint", true, Option, |_| Some(None)), + (col_boolean, "col_boolean", true, Option, |_| Some(None)), + (col_real, "col_real", true, Option, |_| Some(None)), + (col_double, "col_double", true, Option, |_| Some(None)), + (col_date, "col_date", true, Option, |_| Some(None)), + (col_time, "col_time", true, Option, |_| Some(None)), + (col_timestamp, "col_timestamp", true, Option, |_| Some(None)), + (col_timestamptz, "col_timestamptz", true, Option, |_| Some(None)), + (col_uuid, "col_uuid", true, Option, |_| Some(None)), + (col_jsonb, "col_jsonb", true, Option, |_| Some(None)), + (col_bytea, "col_bytea", true, Option, |_| Some(None)), + (col_numeric, "col_numeric", true, Option, |_| Some(None)), + (col_smallint, "col_smallint", true, Option, |_| Some(None)), + (col_int_array, "col_int_array", true, Option>>, |_| Some(None)) + }); + + feldera_types::serialize_table_record!(CdcAllTypesStruct[17]{ + id["id"]: i32, + col_text["col_text"]: Option, + col_integer["col_integer"]: Option, + col_bigint["col_bigint"]: Option, + col_boolean["col_boolean"]: Option, + col_real["col_real"]: Option, + col_double["col_double"]: Option, + col_date["col_date"]: Option, + col_time["col_time"]: Option, + col_timestamp["col_timestamp"]: Option, + col_timestamptz["col_timestamptz"]: Option, + col_uuid["col_uuid"]: Option, + col_jsonb["col_jsonb"]: Option, + col_bytea["col_bytea"]: Option, + col_numeric["col_numeric"]: Option, + col_smallint["col_smallint"]: Option, + col_int_array["col_int_array"]: Option>> + }); + + impl CdcAllTypesStruct { + fn schema() -> Vec { + vec![ + Field::new("id".into(), ColumnType::int(false)), + Field::new("col_text".into(), ColumnType::varchar(true)), + Field::new("col_integer".into(), ColumnType::int(true)), + Field::new("col_bigint".into(), ColumnType::bigint(true)), + Field::new("col_boolean".into(), ColumnType::boolean(true)), + Field::new("col_real".into(), ColumnType::real(true)), + Field::new("col_double".into(), ColumnType::double(true)), + Field::new("col_date".into(), ColumnType::varchar(true)), + Field::new("col_time".into(), ColumnType::varchar(true)), + Field::new("col_timestamp".into(), ColumnType::varchar(true)), + Field::new("col_timestamptz".into(), ColumnType::varchar(true)), + Field::new("col_uuid".into(), ColumnType::varchar(true)), + Field::new("col_jsonb".into(), ColumnType::variant(true)), + Field::new("col_bytea".into(), ColumnType::varchar(true)), + Field::new("col_numeric".into(), ColumnType::varchar(true)), + Field::new("col_smallint".into(), ColumnType::smallint(true)), + Field::new( + "col_int_array".into(), + ColumnType::array(true, ColumnType::int(true)), + ), + ] + } + } + + fn cdc_all_types_test_circuit( + url: &str, + publication: &str, + source_table: &str, + output_path: &Path, + ) -> (Controller, crossbeam::channel::Receiver) { + let schema = CdcAllTypesStruct::schema(); + let config = serde_json::from_value(json!({ + "name": "cdc_all_types_test", + "workers": 1, + "inputs": { + "cdc_in": { + "stream": "test_input1", + "transport": { + "name": "postgres_cdc_input", + "config": { + "uri": url, + "publication": publication, + "source_table": source_table, + }, + }, + }, + }, + "outputs": { + "test_output1": { + "stream": "test_output1", + "transport": { + "name": "file_output", + "config": { + "path": output_path, + } + }, + "format": { + "name": "json", + "config": { + "update_format": "insert_delete", + "array": false, + } + } + } + } + })) + .unwrap(); + + let (err_sender, err_receiver) = crossbeam::channel::unbounded(); + let controller = Controller::with_test_config( + move |workers| { + Ok({ + let (circuit, catalog) = Runtime::init_circuit(workers, move |circuit| { + let mut catalog = Catalog::new(); + let (input, hinput) = circuit.add_input_zset::(); + + let input_schema = serde_json::to_string(&Relation::new( + "test_input1".into(), + schema.clone(), + false, + BTreeMap::new(), + )) + .unwrap(); + + let output_schema = serde_json::to_string(&Relation::new( + "test_output1".into(), + schema, + false, + BTreeMap::new(), + )) + .unwrap(); + + catalog.register_materialized_input_zset::<_, CdcAllTypesStruct>( + input.clone(), + hinput, + &input_schema, + ); + + catalog.register_materialized_output_zset::<_, CdcAllTypesStruct>( + input, + &output_schema, + ); + + Ok(catalog) + }) + .unwrap(); + (circuit, Box::new(catalog)) + }) + }, + &config, + Box::new(move |e, _| { + let msg = format!("cdc_all_types_test: error: {e}"); + println!("{msg}"); + err_sender.send(msg).unwrap() + }), + ) + .unwrap(); + + (controller, err_receiver) + } + + /// Helper: read output file lines as JSON values. + fn read_output_json(path: &Path) -> Vec { + let content = std::fs::read_to_string(path).unwrap_or_default(); + content + .lines() + .filter(|l| !l.is_empty()) + .map(|l| serde_json::from_str(l).unwrap()) + .collect() + } + + /// Helper: count the number of "insert" entries in the output. + fn count_inserts(rows: &[serde_json::Value]) -> usize { + rows.iter().filter(|r| r.get("insert").is_some()).count() + } + + /// Helper: count the number of "delete" entries in the output. + fn count_deletes(rows: &[serde_json::Value]) -> usize { + rows.iter().filter(|r| r.get("delete").is_some()).count() + } + + // ------------------------------------------------------------------- + // Test 1: Basic CDC insert test + // ------------------------------------------------------------------- + + /// Tests that the CDC connector picks up rows inserted into a Postgres table + /// via the initial snapshot and/or logical replication stream. + /// + /// Requires: wal_level=logical, user with REPLICATION privilege. + #[test] + #[serial] + #[ignore] + fn test_cdc_basic_insert() { + let url = postgres_url(); + let table_name = "cdc_test_basic_insert"; + let publication = "cdc_pub_basic_insert"; + + // Pre-insert some rows before starting the pipeline (tests snapshot). + let mut table = CdcTestTable::new_simple(table_name, publication, &url); + table.execute(&format!( + "INSERT INTO {table_name} VALUES (1, true, NULL, 'hello')" + )); + table.execute(&format!( + "INSERT INTO {table_name} VALUES (2, false, 42, 'world')" + )); + + let output_file = NamedTempFile::new().unwrap(); + let output_path = output_file.path().to_owned(); + + let (controller, err_receiver) = cdc_simple_test_circuit( + &url, + publication, + &format!("public.{table_name}"), + &output_path, + ); + + controller.start(); + + // Wait for the snapshot data to appear in the output. + wait( + || { + let rows = read_output_json(&output_path); + count_inserts(&rows) >= 2 || !err_receiver.is_empty() + }, + 60_000, + ) + .expect("timeout: CDC basic insert test did not receive snapshot rows"); + + assert!(err_receiver.is_empty(), "unexpected errors in CDC pipeline"); + + // Now insert more rows while the pipeline is running (tests replication). + table.execute(&format!( + "INSERT INTO {table_name} VALUES (3, true, 100, 'streaming')" + )); + + wait( + || { + let rows = read_output_json(&output_path); + count_inserts(&rows) >= 3 || !err_receiver.is_empty() + }, + 60_000, + ) + .expect("timeout: CDC basic insert test did not receive streamed row"); + + let rows = read_output_json(&output_path); + assert!(count_inserts(&rows) >= 3); + + // Verify the content of the rows + let inserts: Vec<&serde_json::Value> = + rows.iter().filter_map(|r| r.get("insert")).collect(); + + // Check that we have rows with ids 1, 2, 3 + let ids: Vec = inserts + .iter() + .filter_map(|r| r.get("id").and_then(|v| v.as_i64())) + .collect(); + assert!(ids.contains(&1), "missing row with id=1"); + assert!(ids.contains(&2), "missing row with id=2"); + assert!(ids.contains(&3), "missing row with id=3"); + + controller.stop().unwrap(); + } + + // ------------------------------------------------------------------- + // Test 2: Data type coverage test + // ------------------------------------------------------------------- + + /// Tests CDC replication of all common Postgres data types. + /// + /// Requires: wal_level=logical, user with REPLICATION privilege. + #[test] + #[serial] + #[ignore] + fn test_cdc_all_data_types() { + let url = postgres_url(); + let table_name = "cdc_test_all_types"; + let publication = "cdc_pub_all_types"; + + let mut table = CdcTestTable::new_all_types(table_name, publication, &url); + + // Insert a row with all types populated. + table.execute(&format!( + r#"INSERT INTO {table_name} ( + id, col_text, col_integer, col_bigint, col_boolean, + col_real, col_double, col_date, col_time, + col_timestamp, col_timestamptz, col_uuid, col_jsonb, + col_bytea, col_numeric, col_smallint, col_int_array + ) VALUES ( + 1, 'hello world', 42, 9876543210, true, + 3.14, 2.718281828, '2024-06-15', '14:30:00', + '2024-01-01 12:00:00', '2024-01-01 12:00:00+00', '550e8400-e29b-41d4-a716-446655440000', + '{{"key": "value", "nested": {{"a": 1}}}}', + E'\\xDEADBEEF', 12345.67, 7, ARRAY[1, 2, 3] + )"# + )); + + // Insert a row with NULLs for nullable columns. + table.execute(&format!(r#"INSERT INTO {table_name} (id) VALUES (2)"#)); + + let output_file = NamedTempFile::new().unwrap(); + let output_path = output_file.path().to_owned(); + + let (controller, err_receiver) = cdc_all_types_test_circuit( + &url, + publication, + &format!("public.{table_name}"), + &output_path, + ); + + controller.start(); + + // Wait for both rows to appear. + wait( + || { + let rows = read_output_json(&output_path); + count_inserts(&rows) >= 2 || !err_receiver.is_empty() + }, + 60_000, + ) + .expect("timeout: CDC all-types test did not receive rows"); + + assert!( + err_receiver.is_empty(), + "unexpected errors in CDC all-types pipeline" + ); + + let rows = read_output_json(&output_path); + let inserts: Vec<&serde_json::Value> = + rows.iter().filter_map(|r| r.get("insert")).collect(); + + // Find the fully-populated row (id=1). + let row1 = inserts + .iter() + .find(|r| r.get("id").and_then(|v| v.as_i64()) == Some(1)) + .expect("missing row with id=1"); + + // Verify types are present and correctly encoded. + assert_eq!(row1["col_text"], json!("hello world")); + assert_eq!(row1["col_integer"], json!(42)); + assert_eq!(row1["col_bigint"], json!(9876543210i64)); + assert_eq!(row1["col_boolean"], json!(true)); + // Float values: compare approximately + assert!(row1["col_real"].as_f64().unwrap() > 3.13); + assert!(row1["col_real"].as_f64().unwrap() < 3.15); + assert!(row1["col_double"].as_f64().unwrap() > 2.71); + assert!(row1["col_double"].as_f64().unwrap() < 2.72); + // Date, time, timestamp are encoded as strings + assert!( + row1["col_date"].as_str().unwrap().contains("2024-06-15"), + "col_date mismatch: {:?}", + row1["col_date"] + ); + assert!( + row1["col_time"].as_str().unwrap().contains("14:30"), + "col_time mismatch: {:?}", + row1["col_time"] + ); + assert!( + row1["col_timestamp"] + .as_str() + .unwrap() + .contains("2024-01-01"), + "col_timestamp mismatch: {:?}", + row1["col_timestamp"] + ); + assert!( + row1["col_timestamptz"].as_str().unwrap().contains("2024"), + "col_timestamptz mismatch: {:?}", + row1["col_timestamptz"] + ); + // UUID + assert_eq!( + row1["col_uuid"].as_str().unwrap(), + "550e8400-e29b-41d4-a716-446655440000" + ); + // JSONB - should be passed through as JSON + assert!(row1["col_jsonb"].is_object() || row1["col_jsonb"].is_string()); + // BYTEA - encoded as hex string + assert!( + row1["col_bytea"].as_str().is_some(), + "col_bytea should be a string: {:?}", + row1["col_bytea"] + ); + // NUMERIC - encoded as string to preserve precision + assert!( + row1["col_numeric"].as_str().is_some(), + "col_numeric should be a string: {:?}", + row1["col_numeric"] + ); + assert_eq!(row1["col_smallint"], json!(7)); + // Integer array + assert!(row1["col_int_array"].is_array()); + let arr = row1["col_int_array"].as_array().unwrap(); + assert_eq!(arr.len(), 3); + + // Find the NULL row (id=2). + let row2 = inserts + .iter() + .find(|r| r.get("id").and_then(|v| v.as_i64()) == Some(2)) + .expect("missing row with id=2"); + + assert!(row2["col_text"].is_null()); + assert!(row2["col_integer"].is_null()); + assert!(row2["col_bigint"].is_null()); + assert!(row2["col_boolean"].is_null()); + assert!(row2["col_uuid"].is_null()); + assert!(row2["col_jsonb"].is_null()); + assert!(row2["col_bytea"].is_null()); + + controller.stop().unwrap(); + } + + // ------------------------------------------------------------------- + // Test 3: Update and delete test + // ------------------------------------------------------------------- + + /// Tests that the CDC connector correctly captures UPDATE and DELETE operations. + /// + /// Requires: wal_level=logical, user with REPLICATION privilege, REPLICA IDENTITY FULL. + #[test] + #[serial] + #[ignore] + fn test_cdc_update_delete() { + let url = postgres_url(); + let table_name = "cdc_test_upd_del"; + let publication = "cdc_pub_upd_del"; + + let mut table = CdcTestTable::new_simple(table_name, publication, &url); + + let output_file = NamedTempFile::new().unwrap(); + let output_path = output_file.path().to_owned(); + + let (controller, err_receiver) = cdc_simple_test_circuit( + &url, + publication, + &format!("public.{table_name}"), + &output_path, + ); + + controller.start(); + + // Give the pipeline a moment to start and establish the replication connection. + std::thread::sleep(std::time::Duration::from_secs(2)); + + // Phase 1: Insert rows. + table.execute(&format!( + "INSERT INTO {table_name} VALUES (1, true, 10, 'alpha')" + )); + table.execute(&format!( + "INSERT INTO {table_name} VALUES (2, false, 20, 'beta')" + )); + + wait( + || { + let rows = read_output_json(&output_path); + count_inserts(&rows) >= 2 || !err_receiver.is_empty() + }, + 60_000, + ) + .expect("timeout: CDC update/delete test did not receive initial inserts"); + + // Phase 2: Update a row. + table.execute(&format!( + "UPDATE {table_name} SET s = 'alpha_updated', i = 11 WHERE id = 1" + )); + + // An UPDATE with REPLICA IDENTITY FULL should produce a delete of the old + // row and an insert of the new row in Feldera's insert/delete model. + wait( + || { + let rows = read_output_json(&output_path); + // We should see at least one delete (the old row) and one more insert + // (the new row) beyond the initial 2 inserts. + let ins = count_inserts(&rows); + let dels = count_deletes(&rows); + (ins >= 3 && dels >= 1) || !err_receiver.is_empty() + }, + 60_000, + ) + .expect("timeout: CDC update/delete test did not receive update events"); + + // Phase 3: Delete a row. + table.execute(&format!("DELETE FROM {table_name} WHERE id = 2")); + + wait( + || { + let rows = read_output_json(&output_path); + let dels = count_deletes(&rows); + dels >= 2 || !err_receiver.is_empty() + }, + 60_000, + ) + .expect("timeout: CDC update/delete test did not receive delete event"); + + assert!( + err_receiver.is_empty(), + "unexpected errors in CDC update/delete pipeline" + ); + + let rows = read_output_json(&output_path); + + // Verify the delete for id=2 is present. + let deletes: Vec<&serde_json::Value> = + rows.iter().filter_map(|r| r.get("delete")).collect(); + let deleted_ids: Vec = deletes + .iter() + .filter_map(|r| r.get("id").and_then(|v| v.as_i64())) + .collect(); + assert!( + deleted_ids.contains(&2), + "missing delete for id=2, got deletes: {:?}", + deleted_ids + ); + + // Verify the updated row (id=1, s='alpha_updated') is present in inserts. + let inserts: Vec<&serde_json::Value> = + rows.iter().filter_map(|r| r.get("insert")).collect(); + let updated = inserts.iter().find(|r| { + r.get("id").and_then(|v| v.as_i64()) == Some(1) + && r.get("s").and_then(|v| v.as_str()) == Some("alpha_updated") + }); + assert!( + updated.is_some(), + "missing updated row (id=1, s='alpha_updated') in inserts" + ); + + controller.stop().unwrap(); + } + + // ------------------------------------------------------------------- + // Test 4: Restart/resume test + // ------------------------------------------------------------------- + + /// Tests that the pipeline resumes from the replication slot position after + /// restart, instead of re-snapshotting the entire table. + /// + /// Requires: wal_level=logical, user with REPLICATION privilege. + #[test] + #[serial] + #[ignore] + fn test_cdc_restart_resumes_from_slot() { + let url = postgres_url(); + let table_name = "cdc_test_restart"; + let publication = "cdc_pub_restart"; + + // Insert initial rows. + let mut table = CdcTestTable::new_simple(table_name, publication, &url); + table.execute(&format!( + "INSERT INTO {table_name} VALUES (1, true, NULL, 'first')" + )); + + // --- First pipeline run --- + let output_file_1 = NamedTempFile::new().unwrap(); + let output_path_1 = output_file_1.path().to_owned(); + + let (controller_1, err_receiver_1) = cdc_simple_test_circuit( + &url, + publication, + &format!("public.{table_name}"), + &output_path_1, + ); + + controller_1.start(); + + // Wait for the snapshot row to appear. + wait( + || { + let rows = read_output_json(&output_path_1); + count_inserts(&rows) >= 1 || !err_receiver_1.is_empty() + }, + 60_000, + ) + .expect("timeout: first run did not receive snapshot row"); + + assert!( + err_receiver_1.is_empty(), + "unexpected errors in first pipeline run" + ); + + // Insert a second row while the pipeline is running. + table.execute(&format!( + "INSERT INTO {table_name} VALUES (2, false, 42, 'second')" + )); + + wait( + || { + let rows = read_output_json(&output_path_1); + count_inserts(&rows) >= 2 || !err_receiver_1.is_empty() + }, + 60_000, + ) + .expect("timeout: first run did not receive streamed row"); + + // Stop the first pipeline. + controller_1.stop().unwrap(); + + // Small delay to let the replication slot become inactive. + std::thread::sleep(std::time::Duration::from_secs(2)); + + // --- Second pipeline run (restart) --- + let output_file_2 = NamedTempFile::new().unwrap(); + let output_path_2 = output_file_2.path().to_owned(); + + let (controller_2, err_receiver_2) = cdc_simple_test_circuit( + &url, + publication, + &format!("public.{table_name}"), + &output_path_2, + ); + + controller_2.start(); + + // Insert a third row in the restarted pipeline. + table.execute(&format!( + "INSERT INTO {table_name} VALUES (3, true, 99, 'third')" + )); + + // Wait for the third row to appear. + wait( + || { + let rows = read_output_json(&output_path_2); + count_inserts(&rows) >= 1 || !err_receiver_2.is_empty() + }, + 60_000, + ) + .expect("timeout: second run did not receive any rows"); + + assert!( + err_receiver_2.is_empty(), + "unexpected errors in second pipeline run" + ); + + let rows = read_output_json(&output_path_2); + let inserts: Vec<&serde_json::Value> = + rows.iter().filter_map(|r| r.get("insert")).collect(); + + let ids: Vec = inserts + .iter() + .filter_map(|r| r.get("id").and_then(|v| v.as_i64())) + .collect(); + + // The third row (id=3) must be present — it was inserted after restart. + assert!( + ids.contains(&3), + "missing row with id=3 after restart, got ids: {ids:?}" + ); + + // Row id=1 should NOT be re-snapshotted in the second run's output. + // If PostgresStore is working correctly, etl resumes from the slot + // position and skips the snapshot phase. + // + // Note: some replay of recent rows (id=2) is acceptable under + // at-least-once semantics, but a full re-snapshot (id=1 appearing) + // indicates the stored state was not preserved. + assert!( + !ids.contains(&1), + "row id=1 was re-snapshotted after restart — PostgresStore state \ + was not preserved. Got ids: {ids:?}" + ); + + controller_2.stop().unwrap(); + } +} diff --git a/crates/adapters/src/test/mock_input_consumer.rs b/crates/adapters/src/test/mock_input_consumer.rs index ea0061314f9..8d5633f7ad8 100644 --- a/crates/adapters/src/test/mock_input_consumer.rs +++ b/crates/adapters/src/test/mock_input_consumer.rs @@ -132,6 +132,12 @@ impl InputConsumer for MockInputConsumer { } fn update_connector_health(&self, _health: ConnectorHealth) {} + + fn completion_watcher( + &self, + ) -> Option> { + None + } } pub struct MockInputParserState { diff --git a/crates/adapters/src/transport.rs b/crates/adapters/src/transport.rs index 52f63bcea8c..c112e832dc6 100644 --- a/crates/adapters/src/transport.rs +++ b/crates/adapters/src/transport.rs @@ -117,6 +117,7 @@ pub fn input_transport_config_to_endpoint( | TransportConfig::DeltaTableInput(_) | TransportConfig::DeltaTableOutput(_) | TransportConfig::PostgresInput(_) + | TransportConfig::PostgresCdcInput(_) | TransportConfig::PostgresOutput(_) | TransportConfig::HttpOutput | TransportConfig::RedisOutput(_) diff --git a/crates/adapters/src/transport/kafka/ft/test.rs b/crates/adapters/src/transport/kafka/ft/test.rs index 79ea0c38559..bc2bb9ff9c2 100644 --- a/crates/adapters/src/transport/kafka/ft/test.rs +++ b/crates/adapters/src/transport/kafka/ft/test.rs @@ -750,6 +750,12 @@ impl InputConsumer for DummyInputConsumer { }); } + fn completion_watcher( + &self, + ) -> Option> { + None + } + fn error(&self, fatal: bool, error: AnyError, _tag: Option<&'static str>) { info!("error: {error}"); self.called(ConsumerCall::Error(fatal)); diff --git a/crates/feldera-types/src/config.rs b/crates/feldera-types/src/config.rs index 33254934128..68117df7d42 100644 --- a/crates/feldera-types/src/config.rs +++ b/crates/feldera-types/src/config.rs @@ -18,7 +18,9 @@ use crate::transport::iceberg::IcebergReaderConfig; use crate::transport::kafka::{KafkaInputConfig, KafkaOutputConfig}; use crate::transport::nats::NatsInputConfig; use crate::transport::nexmark::NexmarkInputConfig; -use crate::transport::postgres::{PostgresReaderConfig, PostgresWriterConfig}; +use crate::transport::postgres::{ + PostgresCdcReaderConfig, PostgresReaderConfig, PostgresWriterConfig, +}; use crate::transport::pubsub::PubSubInputConfig; use crate::transport::redis::RedisOutputConfig; use crate::transport::s3::S3InputConfig; @@ -1619,6 +1621,7 @@ pub enum TransportConfig { // Prevent rust from complaining about large size difference between enum variants. IcebergInput(Box), PostgresInput(PostgresReaderConfig), + PostgresCdcInput(PostgresCdcReaderConfig), PostgresOutput(PostgresWriterConfig), Datagen(DatagenInputConfig), Nexmark(NexmarkInputConfig), @@ -1646,6 +1649,7 @@ impl TransportConfig { TransportConfig::DeltaTableOutput(_) => "delta_table_output".to_string(), TransportConfig::IcebergInput(_) => "iceberg_input".to_string(), TransportConfig::PostgresInput(_) => "postgres_input".to_string(), + TransportConfig::PostgresCdcInput(_) => "postgres_cdc_input".to_string(), TransportConfig::PostgresOutput(_) => "postgres_output".to_string(), TransportConfig::Datagen(_) => "datagen".to_string(), TransportConfig::Nexmark(_) => "nexmark".to_string(), diff --git a/crates/feldera-types/src/transport/postgres.rs b/crates/feldera-types/src/transport/postgres.rs index fabb660e52c..49d2e57ae1f 100644 --- a/crates/feldera-types/src/transport/postgres.rs +++ b/crates/feldera-types/src/transport/postgres.rs @@ -67,6 +67,31 @@ impl PostgresTlsConfig { } } +/// Postgres CDC input connector configuration. +/// +/// Uses logical replication to capture ongoing changes from a Postgres database. +/// Requires a pre-created publication and a user with REPLICATION privilege. +/// Tables must have primary keys and `REPLICA IDENTITY FULL` is recommended +/// for UPDATE/DELETE support. +#[derive(Debug, Clone, Eq, PartialEq, Deserialize, Serialize, ToSchema)] +pub struct PostgresCdcReaderConfig { + /// Postgres connection URI. The user must have REPLICATION privilege. + /// See: + pub uri: String, + + /// Name of the pre-created Postgres publication to replicate from. + pub publication: String, + + /// Postgres table to replicate (e.g. "public.orders"). + /// Must be included in the publication. + pub source_table: String, + + /// TLS/SSL configuration. + #[serde(flatten)] + #[schema(inline)] + pub tls: PostgresTlsConfig, +} + /// Postgres input connector configuration. #[derive(Debug, Clone, Eq, PartialEq, Deserialize, Serialize, ToSchema)] pub struct PostgresReaderConfig { diff --git a/crates/pipeline-manager/src/api/main.rs b/crates/pipeline-manager/src/api/main.rs index b5be91f5146..9abefa408ac 100644 --- a/crates/pipeline-manager/src/api/main.rs +++ b/crates/pipeline-manager/src/api/main.rs @@ -403,6 +403,7 @@ It contains the following fields: feldera_types::transport::iceberg::RestCatalogConfig, feldera_types::transport::iceberg::GlueCatalogConfig, feldera_types::transport::postgres::PostgresReaderConfig, + feldera_types::transport::postgres::PostgresCdcReaderConfig, feldera_types::transport::postgres::PostgresWriterConfig, feldera_types::transport::postgres::PostgresWriteMode, feldera_types::transport::postgres::PostgresTlsConfig, diff --git a/openapi.json b/openapi.json index ccf7814276c..e920685301f 100644 --- a/openapi.json +++ b/openapi.json @@ -11213,6 +11213,79 @@ } } }, + "PostgresCdcReaderConfig": { + "allOf": [ + { + "type": "object", + "description": "TLS/SSL configuration for PostgreSQL connectors.", + "properties": { + "ssl_ca_location": { + "type": "string", + "description": "Path to a file containing a sequence of CA certificates in PEM format.", + "nullable": true + }, + "ssl_ca_pem": { + "type": "string", + "description": "A sequence of CA certificates in PEM format.", + "nullable": true + }, + "ssl_certificate_chain_location": { + "type": "string", + "description": "The path to the certificate chain file.\nThe file must contain a sequence of PEM-formatted certificates,\nthe first being the leaf certificate, and the remainder forming\nthe chain of certificates up to and including the trusted root certificate.", + "nullable": true + }, + "ssl_client_key": { + "type": "string", + "description": "The client certificate key in PEM format.", + "nullable": true + }, + "ssl_client_key_location": { + "type": "string", + "description": "Path to the client certificate key.", + "nullable": true + }, + "ssl_client_location": { + "type": "string", + "description": "Path to the client certificate.", + "nullable": true + }, + "ssl_client_pem": { + "type": "string", + "description": "The client certificate in PEM format.", + "nullable": true + }, + "verify_hostname": { + "type": "boolean", + "description": "True to enable hostname verification when using TLS. True by default.", + "nullable": true + } + } + }, + { + "type": "object", + "required": [ + "uri", + "publication", + "source_table" + ], + "properties": { + "publication": { + "type": "string", + "description": "Name of the pre-created Postgres publication to replicate from." + }, + "source_table": { + "type": "string", + "description": "Postgres table to replicate (e.g. \"public.orders\").\nMust be included in the publication." + }, + "uri": { + "type": "string", + "description": "Postgres connection URI. The user must have REPLICATION privilege.\nSee: " + } + } + } + ], + "description": "Postgres CDC input connector configuration.\n\nUses logical replication to capture ongoing changes from a Postgres database.\nRequires a pre-created publication and a user with REPLICATION privilege.\nTables must have primary keys and `REPLICA IDENTITY FULL` is recommended\nfor UPDATE/DELETE support." + }, "PostgresReaderConfig": { "allOf": [ { @@ -13529,6 +13602,24 @@ } } }, + { + "type": "object", + "required": [ + "name", + "config" + ], + "properties": { + "config": { + "$ref": "#/components/schemas/PostgresCdcReaderConfig" + }, + "name": { + "type": "string", + "enum": [ + "postgres_cdc_input" + ] + } + } + }, { "type": "object", "required": [