diff --git a/Cargo.lock b/Cargo.lock index 6b1758610a670..87ce97e2716c1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -23,6 +23,16 @@ version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa" +[[package]] +name = "aead" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d122413f284cf2d62fb1b7db97e02edb8cda96d769b16e443a4f6195e35662b0" +dependencies = [ + "crypto-common 0.1.6", + "generic-array", +] + [[package]] name = "aes" version = "0.8.4" @@ -34,6 +44,20 @@ dependencies = [ "cpufeatures 0.2.17", ] +[[package]] +name = "aes-gcm" +version = "0.10.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "831010a0f742e1209b3bcea8fab6a8e149051ba6099432c8cb2cc117dec3ead1" +dependencies = [ + "aead", + "aes", + "cipher", + "ctr", + "ghash", + "subtle", +] + [[package]] name = "ahash" version = "0.7.8" @@ -202,7 +226,8 @@ checksum = "36fa98bc79671c7981272d91a8753a928ff6a1cd8e4f20a44c45bd5d313840bf" dependencies = [ "bigdecimal", "bon", - "digest", + "crc32fast", + "digest 0.10.7", "log", "miniz_oxide", "num-bigint", @@ -212,6 +237,7 @@ dependencies = [ "serde", "serde_bytes", "serde_json", + "snap", "strum", "strum_macros", "thiserror 2.0.18", @@ -273,16 +299,34 @@ version = "57.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cb372a7cbcac02a35d3fb7b3fc1f969ec078e871f9bb899bf00a2e1809bec8a3" dependencies = [ - "arrow-arith", - "arrow-array", - "arrow-buffer", - "arrow-cast", - "arrow-data", - "arrow-ord", - "arrow-row", - "arrow-schema", - "arrow-select", - "arrow-string", + "arrow-arith 57.1.0", + "arrow-array 57.2.0", + "arrow-buffer 57.3.0", + "arrow-cast 57.2.0", + "arrow-data 57.2.0", + "arrow-ord 57.2.0", + "arrow-row 57.1.0", + "arrow-schema 57.2.0", + "arrow-select 57.2.0", + "arrow-string 57.1.0", +] + +[[package]] +name = "arrow" +version = "58.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6cfdd0833e32a9874d2b55089333ad310c0be208aafa277385ce2461dec90be3" +dependencies = [ + "arrow-arith 58.4.0", + "arrow-array 58.4.0", + "arrow-buffer 58.4.0", + "arrow-cast 58.4.0", + "arrow-data 58.4.0", + "arrow-ord 58.4.0", + "arrow-row 58.4.0", + "arrow-schema 58.4.0", + "arrow-select 58.4.0", + "arrow-string 58.4.0", ] [[package]] @@ -291,10 +335,24 @@ version = "57.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0f377dcd19e440174596d83deb49cd724886d91060c07fec4f67014ef9d54049" dependencies = [ - "arrow-array", - "arrow-buffer", - "arrow-data", - "arrow-schema", + "arrow-array 57.2.0", + "arrow-buffer 57.3.0", + "arrow-data 57.2.0", + "arrow-schema 57.2.0", + "chrono", + "num-traits", +] + +[[package]] +name = "arrow-arith" +version = "58.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0a41203398f0eaa6f7ec8e62c0da742a21abf282c148fc157f6c35c90e29981a" +dependencies = [ + "arrow-array 58.4.0", + "arrow-buffer 58.4.0", + "arrow-data 58.4.0", + "arrow-schema 58.4.0", "chrono", "num-traits", ] @@ -306,9 +364,9 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "65ca404ea6191e06bf30956394173337fa9c35f445bd447fe6c21ab944e1a23c" dependencies = [ "ahash 0.8.12", - "arrow-buffer", - "arrow-data", - "arrow-schema", + "arrow-buffer 57.3.0", + "arrow-data 57.2.0", + "arrow-schema 57.2.0", "chrono", "half", "hashbrown 0.16.1", @@ -317,6 +375,24 @@ dependencies = [ "num-traits", ] +[[package]] +name = "arrow-array" +version = "58.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae33dad492b7df00a217563a7b0ef2874df68a0deea1b1a3acf628152f7f7a69" +dependencies = [ + "ahash 0.8.12", + "arrow-buffer 58.4.0", + "arrow-data 58.4.0", + "arrow-schema 58.4.0", + "chrono", + "half", + "hashbrown 0.17.1", + "num-complex", + "num-integer", + "num-traits", +] + [[package]] name = "arrow-buffer" version = "57.3.0" @@ -329,18 +405,30 @@ dependencies = [ "num-traits", ] +[[package]] +name = "arrow-buffer" +version = "58.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9552f96391c005e6ab449fa941420935e7e062489b12b8b1b08879b2163f5b5" +dependencies = [ + "bytes", + "half", + "num-bigint", + "num-traits", +] + [[package]] name = "arrow-cast" version = "57.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9c8e372ed52bd4ee88cc1e6c3859aa7ecea204158ac640b10e187936e7e87074" dependencies = [ - "arrow-array", - "arrow-buffer", - "arrow-data", - "arrow-ord", - "arrow-schema", - "arrow-select", + "arrow-array 57.2.0", + "arrow-buffer 57.3.0", + "arrow-data 57.2.0", + "arrow-ord 57.2.0", + "arrow-schema 57.2.0", + "arrow-select 57.2.0", "atoi", "base64 0.22.1", "chrono", @@ -351,14 +439,48 @@ dependencies = [ "ryu", ] +[[package]] +name = "arrow-cast" +version = "58.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3a8a327c9649f30d8406995f27642b68df354713cca3baaaf100f076f18d5f34" +dependencies = [ + "arrow-array 58.4.0", + "arrow-buffer 58.4.0", + "arrow-data 58.4.0", + "arrow-ord 58.4.0", + "arrow-schema 58.4.0", + "arrow-select 58.4.0", + "atoi", + "base64 0.22.1", + "chrono", + "half", + "lexical-core", + "num-traits", + "ryu", +] + [[package]] name = "arrow-data" version = "57.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bf87f4ff5fc13290aa47e499a8b669a82c5977c6a1fedce22c7f542c1fd5a597" dependencies = [ - "arrow-buffer", - "arrow-schema", + "arrow-buffer 57.3.0", + "arrow-schema 57.2.0", + "half", + "num-integer", + "num-traits", +] + +[[package]] +name = "arrow-data" +version = "58.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2b24852db04738907e06c04ea61e42fe7fda962a34513022dc0d0e754fb7976b" +dependencies = [ + "arrow-buffer 58.4.0", + "arrow-schema 58.4.0", "half", "num-integer", "num-traits", @@ -366,15 +488,15 @@ dependencies = [ [[package]] name = "arrow-ipc" -version = "57.2.0" +version = "58.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "eb3ca63edd2073fcb42ba112f8ae165df1de935627ead6e203d07c99445f2081" +checksum = "29a908a11fcfb3fb2f6730f4ac15e367bc644e419155e96238f68cf3adde572b" dependencies = [ - "arrow-array", - "arrow-buffer", - "arrow-data", - "arrow-schema", - "arrow-select", + "arrow-array 58.4.0", + "arrow-buffer 58.4.0", + "arrow-data 58.4.0", + "arrow-schema 58.4.0", + "arrow-select 58.4.0", "flatbuffers", ] @@ -384,11 +506,24 @@ version = "57.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "13c4e0530272ca755d6814218dffd04425c5b7854b87fa741d5ff848bf50aa39" dependencies = [ - "arrow-array", - "arrow-buffer", - "arrow-data", - "arrow-schema", - "arrow-select", + "arrow-array 57.2.0", + "arrow-buffer 57.3.0", + "arrow-data 57.2.0", + "arrow-schema 57.2.0", + "arrow-select 57.2.0", +] + +[[package]] +name = "arrow-ord" +version = "58.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63a083ec750f5c043f02946b4baf05fcdbb55f4560a3277055caca5cc99f3eb0" +dependencies = [ + "arrow-array 58.4.0", + "arrow-buffer 58.4.0", + "arrow-data 58.4.0", + "arrow-schema 58.4.0", + "arrow-select 58.4.0", ] [[package]] @@ -397,10 +532,23 @@ version = "57.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "169676f317157dc079cc5def6354d16db63d8861d61046d2f3883268ced6f99f" dependencies = [ - "arrow-array", - "arrow-buffer", - "arrow-data", - "arrow-schema", + "arrow-array 57.2.0", + "arrow-buffer 57.3.0", + "arrow-data 57.2.0", + "arrow-schema 57.2.0", + "half", +] + +[[package]] +name = "arrow-row" +version = "58.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "514ba0ef0d4c5896202dae736251ce415abb43a950bed570fb7981b8716c0e4c" +dependencies = [ + "arrow-array 58.4.0", + "arrow-buffer 58.4.0", + "arrow-data 58.4.0", + "arrow-schema 58.4.0", "half", ] @@ -413,6 +561,12 @@ dependencies = [ "bitflags 2.11.0", ] +[[package]] +name = "arrow-schema" +version = "58.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "21ca356ad6425cecb6eb7b28e4f659f1ee7880fbb1a16127de7dd62901efee9e" + [[package]] name = "arrow-select" version = "57.2.0" @@ -420,10 +574,24 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c96d8a1c180b44ecf2e66c9a2f2bbcb8b1b6f14e165ce46ac8bde211a363411b" dependencies = [ "ahash 0.8.12", - "arrow-array", - "arrow-buffer", - "arrow-data", - "arrow-schema", + "arrow-array 57.2.0", + "arrow-buffer 57.3.0", + "arrow-data 57.2.0", + "arrow-schema 57.2.0", + "num-traits", +] + +[[package]] +name = "arrow-select" +version = "58.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c58da39eb3d8350ad4a549e5c2bc49284dac554016c69829310350f1731b0aad" +dependencies = [ + "ahash 0.8.12", + "arrow-array 58.4.0", + "arrow-buffer 58.4.0", + "arrow-data 58.4.0", + "arrow-schema 58.4.0", "num-traits", ] @@ -433,11 +601,28 @@ version = "57.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cf35e8ef49dcf0c5f6d175edee6b8af7b45611805333129c541a8b89a0fc0534" dependencies = [ - "arrow-array", - "arrow-buffer", - "arrow-data", - "arrow-schema", - "arrow-select", + "arrow-array 57.2.0", + "arrow-buffer 57.3.0", + "arrow-data 57.2.0", + "arrow-schema 57.2.0", + "arrow-select 57.2.0", + "memchr", + "num-traits", + "regex", + "regex-syntax", +] + +[[package]] +name = "arrow-string" +version = "58.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6789b388467525e3271326b6b4915666ecfdf5142aef09779445c954b67543c" +dependencies = [ + "arrow-array 58.4.0", + "arrow-buffer 58.4.0", + "arrow-data 58.4.0", + "arrow-schema 58.4.0", + "arrow-select 58.4.0", "memchr", "num-traits", "regex", @@ -766,7 +951,7 @@ dependencies = [ "fastrand 2.3.0", "hex", "http 1.4.2", - "sha1", + "sha1 0.10.6", "time", "tokio", "tracing", @@ -908,14 +1093,14 @@ dependencies = [ "bytes", "fastrand 2.3.0", "hex", - "hmac", + "hmac 0.12.1", "http 0.2.12", "http 1.4.2", "http-body 1.0.1", "lru", "percent-encoding", "regex-lite", - "sha2", + "sha2 0.10.9", "tracing", "url", ] @@ -1031,11 +1216,11 @@ dependencies = [ "bytes", "form_urlencoded", "hex", - "hmac", + "hmac 0.12.1", "http 0.2.12", "http 1.4.2", "percent-encoding", - "sha2", + "sha2 0.10.9", "time", "tracing", ] @@ -1065,10 +1250,10 @@ dependencies = [ "http 1.4.2", "http-body 1.0.1", "http-body-util", - "md-5", + "md-5 0.10.6", "pin-project-lite", - "sha1", - "sha2", + "sha1 0.10.6", + "sha2 0.10.9", "tracing", ] @@ -1335,7 +1520,7 @@ dependencies = [ "serde_json", "serde_path_to_error", "serde_urlencoded", - "sha1", + "sha1 0.10.6", "sync_wrapper", "tokio", "tokio-tungstenite", @@ -1420,7 +1605,7 @@ dependencies = [ "dyn-clone", "futures", "getrandom 0.2.16", - "hmac", + "hmac 0.12.1", "http-types", "once_cell", "paste", @@ -1431,7 +1616,7 @@ dependencies = [ "rustc_version", "serde", "serde_json", - "sha2", + "sha2 0.10.9", "time", "tracing", "url", @@ -1693,6 +1878,15 @@ dependencies = [ "generic-array", ] +[[package]] +name = "block-buffer" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d2f6c7dbe95a6ed67ad9f18e57daf93a2f034c524b99fd2b76d18fdfeb6660aa" +dependencies = [ + "hybrid-array", +] + [[package]] name = "block-padding" version = "0.3.3" @@ -2114,7 +2308,7 @@ version = "0.4.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "773f3b9af64447d2ce9850330c473515014aa235e6a783b02db81ff39e4a3dad" dependencies = [ - "crypto-common", + "crypto-common 0.1.6", "inout", ] @@ -2188,6 +2382,12 @@ dependencies = [ "cc", ] +[[package]] +name = "cmov" +version = "0.5.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c9ea0ac24bc397ab3c98583a3c9ba74fa56b09a4449bbe172b9b1ddb016027a" + [[package]] name = "codespan-reporting" version = "0.11.1" @@ -2350,6 +2550,12 @@ version = "0.9.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2459377285ad874054d797f3ccebf984978aa39129f6eafde5cdc8315b612f8" +[[package]] +name = "const-oid" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a6ef517f0926dd24a1582492c791b6a4818a4d94e789a334894aa15b0d12f55c" + [[package]] name = "const-random" version = "0.1.18" @@ -2491,7 +2697,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2fd92aca2c6001b1bf5ba0ff84ee74ec8501b52bbef0cac80bf25a6c1d87a83d" dependencies = [ "crc", - "digest", + "digest 0.10.7", "rustversion", "spin 0.10.1", ] @@ -2659,9 +2865,19 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1bfb12502f3fc46cca1bb51ac28df9d618d813cdc3d2f25b9fe775a34af26bb3" dependencies = [ "generic-array", + "rand_core 0.6.4", "typenum", ] +[[package]] +name = "crypto-common" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ce6e4c961d6cd6c9a86db418387425e8bdeaf05b3c8bc1411e6dca4c252f1453" +dependencies = [ + "hybrid-array", +] + [[package]] name = "csv" version = "1.4.0" @@ -2709,6 +2925,24 @@ dependencies = [ "linktime-proc-macro", ] +[[package]] +name = "ctr" +version = "0.9.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0369ee1ad671834580515889b80f2ea915f23b8be8d0daa4bbaf2ac5c7590835" +dependencies = [ + "cipher", +] + +[[package]] +name = "ctutils" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d5515a3834141de9eafb9717ad39eea8247b5674e6066c404e8c4b365d2a29e" +dependencies = [ + "cmov", +] + [[package]] name = "custom-labels" version = "0.4.6" @@ -2986,7 +3220,7 @@ version = "0.7.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e7c1832837b905bbfb5101e07cc24c8deddf52f93225eee6ead5f4d63d53ddcb" dependencies = [ - "const-oid", + "const-oid 0.9.6", "pem-rfc7468", "zeroize", ] @@ -3129,12 +3363,24 @@ version = "0.10.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" dependencies = [ - "block-buffer", - "const-oid", - "crypto-common", + "block-buffer 0.10.4", + "const-oid 0.9.6", + "crypto-common 0.1.6", "subtle", ] +[[package]] +name = "digest" +version = "0.11.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f1dd6dbb5841937940781866fa1281a1ff7bd3bf827091440879f9994983d5c2" +dependencies = [ + "block-buffer 0.12.1", + "const-oid 0.10.2", + "crypto-common 0.2.2", + "ctutils", +] + [[package]] name = "dirs" version = "6.0.0" @@ -3153,7 +3399,7 @@ dependencies = [ "libc", "option-ext", "redox_users", - "windows-sys 0.59.0", + "windows-sys 0.61.1", ] [[package]] @@ -3195,7 +3441,7 @@ name = "duckdb" version = "1.4.3" source = "git+https://github.com/MaterializeInc/duckdb-rs.git?rev=752c7efe2582#752c7efe25820590f590fb0a112443b9ddd396e7" dependencies = [ - "arrow", + "arrow 57.1.0", "cast", "fallible-iterator 0.3.0", "fallible-streaming-iterator", @@ -3239,7 +3485,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ee27f32b5c5292967d2d4a9d7f1e0b0aed2c15daded5a60300e4abb9d8020bca" dependencies = [ "der", - "digest", + "digest 0.10.7", "elliptic-curve", "rfc6979", "signature", @@ -3272,7 +3518,7 @@ checksum = "b5e6043086bf7973472e0c7dff2142ea0b680d30e18d9cc40f267efbf222bd47" dependencies = [ "base16ct", "crypto-bigint", - "digest", + "digest 0.10.7", "ff", "generic-array", "group", @@ -3704,7 +3950,7 @@ dependencies = [ "reqwest 0.12.28", "serde", "serde_json", - "sha2", + "sha2 0.10.9", "socket2 0.6.5", "thiserror 2.0.18", "tokio", @@ -4112,6 +4358,16 @@ dependencies = [ "rand_core 0.10.1", ] +[[package]] +name = "ghash" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0d8a4362ccb29cb0b265253fb0a2728f592895ee6854fd9bc13f2ffda266ff1" +dependencies = [ + "opaque-debug", + "polyval", +] + [[package]] name = "gimli" version = "0.32.3" @@ -4271,6 +4527,12 @@ dependencies = [ "foldhash 0.2.0", ] +[[package]] +name = "hashbrown" +version = "0.17.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" + [[package]] name = "hashlink" version = "0.9.1" @@ -4314,7 +4576,7 @@ dependencies = [ "http 1.4.2", "httpdate", "mime", - "sha1", + "sha1 0.10.6", ] [[package]] @@ -4435,16 +4697,16 @@ version = "0.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6c49c37c09c17a53d937dfbb742eb3a961d65a994e6bcdcf37e7399d0cc8ab5e" dependencies = [ - "digest", + "digest 0.10.7", ] [[package]] -name = "home" -version = "0.5.5" +name = "hmac" +version = "0.13.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5444c27eef6923071f7ebcc33e3444508466a76f7a2b93da00ed6e19f30c1ddb" +checksum = "6303bc9732ae41b04cb554b844a762b4115a61bfaa81e3e83050991eeb56863f" dependencies = [ - "windows-sys 0.48.0", + "digest 0.11.3", ] [[package]] @@ -4551,6 +4813,15 @@ version = "2.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "135b12329e5e3ce057a9f972339ea52bc954fe1e9358ef27f95e89716fbc5424" +[[package]] +name = "hybrid-array" +version = "0.4.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "707114b52a152fa7bdb290cd7cd5912d9467273b6d74e21b8d81aca1f8533f6b" +dependencies = [ + "typenum", +] + [[package]] name = "hyper" version = "0.14.32" @@ -4712,7 +4983,7 @@ dependencies = [ "libc", "percent-encoding", "pin-project-lite", - "socket2 0.5.10", + "socket2 0.6.5", "system-configuration", "tokio", "tower-service", @@ -4736,21 +5007,22 @@ dependencies = [ [[package]] name = "iceberg" -version = "0.9.0" -source = "git+https://github.com/MaterializeInc/iceberg-rust.git?rev=dedd9231ee88ee979b648e14792878b40e74c20a#dedd9231ee88ee979b648e14792878b40e74c20a" +version = "0.10.1" +source = "git+https://github.com/MaterializeInc/iceberg-rust.git?rev=958ac5b3c318026aeb362485c593e4956aeabe95#958ac5b3c318026aeb362485c593e4956aeabe95" dependencies = [ + "aes-gcm", "anyhow", "apache-avro", "array-init", - "arrow-arith", - "arrow-array", - "arrow-buffer", - "arrow-cast", - "arrow-ord", - "arrow-row", - "arrow-schema", - "arrow-select", - "arrow-string", + "arrow-arith 58.4.0", + "arrow-array 58.4.0", + "arrow-buffer 58.4.0", + "arrow-cast 58.4.0", + "arrow-ord 58.4.0", + "arrow-row 58.4.0", + "arrow-schema 58.4.0", + "arrow-select 58.4.0", + "arrow-string 58.4.0", "as-any", "async-trait", "backon", @@ -4770,7 +5042,7 @@ dependencies = [ "once_cell", "ordered-float 4.6.0", "parquet", - "rand 0.8.5", + "rand 0.9.4", "reqwest 0.12.28", "roaring", "serde", @@ -4781,17 +5053,19 @@ dependencies = [ "serde_with", "strum", "tokio", + "tracing", "typed-builder 0.20.1", "typetag", "url", "uuid", + "zeroize", "zstd", ] [[package]] name = "iceberg-catalog-rest" -version = "0.9.0" -source = "git+https://github.com/MaterializeInc/iceberg-rust.git?rev=dedd9231ee88ee979b648e14792878b40e74c20a#dedd9231ee88ee979b648e14792878b40e74c20a" +version = "0.10.1" +source = "git+https://github.com/MaterializeInc/iceberg-rust.git?rev=958ac5b3c318026aeb362485c593e4956aeabe95#958ac5b3c318026aeb362485c593e4956aeabe95" dependencies = [ "async-trait", "chrono", @@ -4803,24 +5077,24 @@ dependencies = [ "serde_derive", "serde_json", "tokio", - "tracing", "typed-builder 0.20.1", "uuid", ] [[package]] name = "iceberg-storage-opendal" -version = "0.9.0" -source = "git+https://github.com/MaterializeInc/iceberg-rust.git?rev=dedd9231ee88ee979b648e14792878b40e74c20a#dedd9231ee88ee979b648e14792878b40e74c20a" +version = "0.10.1" +source = "git+https://github.com/MaterializeInc/iceberg-rust.git?rev=958ac5b3c318026aeb362485c593e4956aeabe95#958ac5b3c318026aeb362485c593e4956aeabe95" dependencies = [ "anyhow", "async-trait", "bytes", "cfg-if", + "futures", "iceberg", "opendal", - "reqsign", - "reqwest 0.12.28", + "reqsign-aws-v4", + "reqsign-core", "serde", "typetag", "url", @@ -5041,7 +5315,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi", "libc", - "windows-sys 0.59.0", + "windows-sys 0.61.1", ] [[package]] @@ -5124,11 +5398,13 @@ checksum = "1a3546dc96b6d42c5f24902af9e2538e82e39ad350b0c766eb3fbf2d8f3d8359" dependencies = [ "jiff-static", "jiff-tzdb-platform", + "js-sys", "log", "portable-atomic", "portable-atomic-util", "serde_core", - "windows-sys 0.59.0", + "wasm-bindgen", + "windows-sys 0.61.1", ] [[package]] @@ -5262,21 +5538,6 @@ dependencies = [ "serde_json", ] -[[package]] -name = "jsonwebtoken" -version = "9.3.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5a87cc7a48537badeae96744432de36f4be2b4a34a05a5ef32e9dd8a1c169dde" -dependencies = [ - "base64 0.22.1", - "js-sys", - "pem 3.0.6", - "ring", - "serde", - "serde_json", - "simple_asn1", -] - [[package]] name = "jsonwebtoken" version = "10.3.0" @@ -5522,7 +5783,7 @@ dependencies = [ "serde", "serde_json", "serde_with", - "sha1", + "sha1 0.10.6", ] [[package]] @@ -5845,6 +6106,15 @@ dependencies = [ "twox-hash", ] +[[package]] +name = "lz4_flex" +version = "0.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7ef0d4ed8669f8f8826eb00dc878084aa8f253506c4fd5e8f58f5bce72ddb97e" +dependencies = [ + "twox-hash", +] + [[package]] name = "maplit" version = "1.0.2" @@ -5886,7 +6156,27 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d89e7ee0cfbedfc4da3340218492196241d89eefb6dab27de5df917a6d2e78cf" dependencies = [ "cfg-if", - "digest", + "digest 0.10.7", +] + +[[package]] +name = "md-5" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69b6441f590336821bb897fb28fc622898ccceb1d6cea3fde5ea86b090c4de98" +dependencies = [ + "cfg-if", + "digest 0.11.3", +] + +[[package]] +name = "mea" +version = "0.6.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c709842c4ce65cb91e2666ad5319dfc1efc3af0d34f02075eddca9000d9f8afb" +dependencies = [ + "hashbrown 0.17.1", + "slab", ] [[package]] @@ -6047,8 +6337,8 @@ dependencies = [ "saturating", "serde", "serde_json", - "sha1", - "sha2", + "sha1 0.10.6", + "sha2 0.10.9", "thiserror 2.0.18", "uuid", "zstd", @@ -6099,7 +6389,7 @@ name = "mz-adapter" version = "0.0.0" dependencies = [ "anyhow", - "arrow", + "arrow 58.4.0", "async-stream", "async-trait", "base64 0.22.1", @@ -6178,7 +6468,7 @@ dependencies = [ "serde", "serde_json", "serde_plain", - "sha2", + "sha2 0.10.9", "smallvec", "static_assertions", "thiserror 2.0.18", @@ -6231,7 +6521,7 @@ name = "mz-arrow-util" version = "0.1.0" dependencies = [ "anyhow", - "arrow", + "arrow 58.4.0", "chrono", "dec", "half", @@ -6277,7 +6567,7 @@ dependencies = [ name = "mz-authenticator" version = "0.1.0" dependencies = [ - "jsonwebtoken 10.3.0", + "jsonwebtoken", "mz-adapter", "mz-adapter-types", "mz-auth", @@ -6383,7 +6673,7 @@ dependencies = [ "humantime", "hyper 1.9.0", "hyper-util", - "jsonwebtoken 10.3.0", + "jsonwebtoken", "launchdarkly-server-sdk", "mz-alloc", "mz-alloc-default", @@ -6494,7 +6784,7 @@ dependencies = [ "serde", "serde_json", "serde_plain", - "sha2", + "sha2 0.10.9", "static_assertions", "thiserror 2.0.18", "timely", @@ -6535,7 +6825,7 @@ name = "mz-catalog-protos" version = "0.0.0" dependencies = [ "anyhow", - "md-5", + "md-5 0.10.6", "mz-audit-log", "mz-catalog-protos", "mz-compute-types", @@ -6617,7 +6907,7 @@ dependencies = [ "semver", "serde", "serde_json", - "sha2", + "sha2 0.10.9", "tracing", "uuid", ] @@ -6963,7 +7253,7 @@ dependencies = [ "rusqlite", "serde", "serde_json", - "sha2", + "sha2 0.10.9", "strsim", "supports-color 3.0.2", "tempfile", @@ -7078,7 +7368,7 @@ dependencies = [ "insta", "ipnet", "itertools 0.14.0", - "jsonwebtoken 10.3.0", + "jsonwebtoken", "maplit", "mime", "mz-adapter", @@ -7197,10 +7487,10 @@ dependencies = [ "enum-iterator", "fallible-iterator 0.2.0", "hex", - "hmac", + "hmac 0.12.1", "insta", "itertools 0.14.0", - "md-5", + "md-5 0.10.6", "murmur2", "mz-build-tools", "mz-expr", @@ -7291,7 +7581,7 @@ dependencies = [ "clap", "derivative", "futures", - "jsonwebtoken 10.3.0", + "jsonwebtoken", "lru", "mz-auth", "mz-ore", @@ -7312,7 +7602,7 @@ dependencies = [ name = "mz-frontegg-client" version = "0.0.0" dependencies = [ - "jsonwebtoken 10.3.0", + "jsonwebtoken", "mz-frontegg-auth", "reqwest 0.12.28", "serde", @@ -7333,7 +7623,7 @@ dependencies = [ "chrono", "clap", "hyper 1.9.0", - "jsonwebtoken 10.3.0", + "jsonwebtoken", "mz-frontegg-auth", "mz-ore", "openssl", @@ -7443,7 +7733,7 @@ dependencies = [ "aws-sdk-kms", "base64 0.22.1", "clap", - "jsonwebtoken 10.3.0", + "jsonwebtoken", "mz-aws-util", "mz-ore", "pem 4.0.0", @@ -7559,7 +7849,7 @@ dependencies = [ "hex", "hex-literal", "reqwest 0.12.28", - "sha2", + "sha2 0.10.9", "tar", "walkdir", ] @@ -7571,7 +7861,7 @@ dependencies = [ "anyhow", "axum", "base64 0.22.1", - "jsonwebtoken 10.3.0", + "jsonwebtoken", "mz-ore", "openssl", "serde", @@ -7614,7 +7904,7 @@ dependencies = [ "reqwest 0.12.28", "serde", "serde_json", - "sha2", + "sha2 0.10.9", "tokio", "tracing", ] @@ -7639,7 +7929,7 @@ dependencies = [ "scopeguard", "serde", "serde_json", - "sha1", + "sha1 0.10.6", "sysinfo", "tokio", "tracing", @@ -7702,7 +7992,7 @@ dependencies = [ "semver", "serde", "serde_json", - "sha2", + "sha2 0.10.9", "thiserror 2.0.18", "tokio", "tower-http", @@ -7738,7 +8028,7 @@ dependencies = [ "itertools 0.14.0", "lgalloc", "libc", - "lz4_flex", + "lz4_flex 0.12.1", "mz-ore", "mz-ore-proc", "native-tls", @@ -7800,7 +8090,7 @@ name = "mz-persist" version = "0.0.0" dependencies = [ "anyhow", - "arrow", + "arrow 58.4.0", "async-stream", "async-trait", "aws-config", @@ -7838,7 +8128,7 @@ dependencies = [ "reqwest 0.12.28", "serde", "serde_json", - "sha2", + "sha2 0.10.9", "tempfile", "time", "timely", @@ -7857,7 +8147,7 @@ version = "26.40.0-dev.0" dependencies = [ "anyhow", "arrayvec 0.7.8", - "arrow", + "arrow 58.4.0", "async-stream", "async-trait", "bytes", @@ -7916,7 +8206,7 @@ name = "mz-persist-types" version = "0.0.0" dependencies = [ "anyhow", - "arrow", + "arrow 58.4.0", "bytes", "chrono", "hex", @@ -8016,7 +8306,7 @@ dependencies = [ "enum-kinds", "futures", "itertools 0.14.0", - "jsonwebtoken 10.3.0", + "jsonwebtoken", "mz-adapter", "mz-adapter-types", "mz-auth", @@ -8189,7 +8479,7 @@ name = "mz-repr" version = "0.0.0" dependencies = [ "anyhow", - "arrow", + "arrow 58.4.0", "bincode", "bitflags 1.3.2", "bytemuck", @@ -8569,7 +8859,7 @@ dependencies = [ "itertools 0.14.0", "junit-report", "maplit", - "md-5", + "md-5 0.10.6", "mz-adapter-types", "mz-catalog", "mz-compute-types", @@ -8632,7 +8922,7 @@ name = "mz-storage" version = "0.0.0" dependencies = [ "anyhow", - "arrow", + "arrow 58.4.0", "async-stream", "async-trait", "axum", @@ -8698,7 +8988,7 @@ dependencies = [ "serde", "serde_bytes", "serde_json", - "sha2", + "sha2 0.10.9", "sqlparser", "tempfile", "thiserror 2.0.18", @@ -8789,7 +9079,7 @@ name = "mz-storage-operators" version = "0.0.0" dependencies = [ "anyhow", - "arrow", + "arrow 58.4.0", "arrow-ipc", "async-compression", "async-stream", @@ -8839,7 +9129,7 @@ name = "mz-storage-types" version = "0.0.0" dependencies = [ "anyhow", - "arrow", + "arrow 58.4.0", "async-trait", "aws-config", "aws-credential-types", @@ -8901,6 +9191,7 @@ dependencies = [ "rand 0.9.4", "rdkafka", "regex", + "reqsign-core", "reqwest 0.12.28", "serde", "serde_json", @@ -8934,7 +9225,7 @@ name = "mz-testdrive" version = "26.40.0-dev.0" dependencies = [ "anyhow", - "arrow", + "arrow 58.4.0", "async-compression", "async-trait", "aws-credential-types", @@ -8953,7 +9244,7 @@ dependencies = [ "itertools 0.14.0", "junit-report", "maplit", - "md-5", + "md-5 0.10.6", "mysql_async", "mz-adapter", "mz-avro", @@ -9019,7 +9310,7 @@ dependencies = [ "differential-dataflow", "futures-util", "itertools 0.14.0", - "lz4_flex", + "lz4_flex 0.12.1", "mz-ore", "num-traits", "proptest", @@ -9425,7 +9716,7 @@ dependencies = [ "serde", "serde_json", "serde_path_to_error", - "sha2", + "sha2 0.10.9", "thiserror 1.0.69", "url", ] @@ -9614,6 +9905,12 @@ version = "11.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ebcec7c9c2a95cacc7cd0ecb89d8a8454eca13906f6deb55258ffff0adeb9405" +[[package]] +name = "opaque-debug" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381" + [[package]] name = "open" version = "5.4.1" @@ -9626,31 +9923,106 @@ dependencies = [ [[package]] name = "opendal" -version = "0.55.0" +version = "0.57.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d075ab8a203a6ab4bc1bce0a4b9fe486a72bf8b939037f4b78d95386384bc80a" +checksum = "96c9c85ce253ff87225e7669979d877a20c98a06604ec9d6dd5f4473e08f1ae1" +dependencies = [ + "opendal-core", + "opendal-layer-retry", + "opendal-layer-timeout", + "opendal-service-gcs", + "opendal-service-s3", +] + +[[package]] +name = "opendal-core" +version = "0.57.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c4f8607c90e2c963a91467f50fb49fbc7fb3d573f88cea219ca59ccd3740b309" dependencies = [ "anyhow", - "backon", "base64 0.22.1", "bytes", - "crc32c", "futures", - "getrandom 0.2.16", "http 1.4.2", "http-body 1.0.1", "jiff", "log", - "md-5", + "md-5 0.11.0", + "mea", "percent-encoding", - "quick-xml 0.38.4", - "reqsign", - "reqwest 0.12.28", + "quick-xml 0.39.4", + "reqsign-core", + "reqwest 0.13.3", "serde", "serde_json", "tokio", "url", "uuid", + "web-time", +] + +[[package]] +name = "opendal-layer-retry" +version = "0.57.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b2a25a718afb81fad81cb9a0580a1cb989221fa2317f888c6a37f8dad408eb7" +dependencies = [ + "backon", + "log", + "opendal-core", +] + +[[package]] +name = "opendal-layer-timeout" +version = "0.57.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e91f731724c213af81e9d03517859c8fc47b4578e64ad61ae4f099f10fe36e3" +dependencies = [ + "opendal-core", + "tokio", +] + +[[package]] +name = "opendal-service-gcs" +version = "0.57.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "48de101aac565ed06af4b47903c24eafd249075553ec1fb18256751c45148d47" +dependencies = [ + "async-trait", + "bytes", + "http 1.4.2", + "log", + "opendal-core", + "percent-encoding", + "quick-xml 0.39.4", + "reqsign-core", + "reqsign-file-read-tokio", + "reqsign-google", + "serde", + "serde_json", + "tokio", +] + +[[package]] +name = "opendal-service-s3" +version = "0.57.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "313d46c9f5ae70bca26b7c3e3fbb9b639292625f28af73aa016f47e788af9deb" +dependencies = [ + "base64 0.22.1", + "bytes", + "crc32c", + "http 1.4.2", + "log", + "md-5 0.11.0", + "opendal-core", + "quick-xml 0.39.4", + "reqsign-aws-v4", + "reqsign-core", + "reqsign-file-read-tokio", + "serde", + "url", ] [[package]] @@ -9912,7 +10284,7 @@ dependencies = [ "ecdsa", "elliptic-curve", "primeorder", - "sha2", + "sha2 0.10.9", ] [[package]] @@ -9924,7 +10296,7 @@ dependencies = [ "ecdsa", "elliptic-curve", "primeorder", - "sha2", + "sha2 0.10.9", ] [[package]] @@ -9938,7 +10310,7 @@ dependencies = [ "elliptic-curve", "primeorder", "rand_core 0.6.4", - "sha2", + "sha2 0.10.9", ] [[package]] @@ -9993,18 +10365,17 @@ dependencies = [ [[package]] name = "parquet" -version = "57.2.0" +version = "58.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5f6a2926a30477c0b95fea6c28c3072712b139337a242c2cc64817bdc20a8854" +checksum = "d298093b2dec60289dce0684c986d0f7679e9dd15771c2c65406e1aaf604a704" dependencies = [ "ahash 0.8.12", - "arrow-array", - "arrow-buffer", - "arrow-cast", - "arrow-data", + "arrow-array 58.4.0", + "arrow-buffer 58.4.0", + "arrow-data 58.4.0", "arrow-ipc", - "arrow-schema", - "arrow-select", + "arrow-schema 58.4.0", + "arrow-select 58.4.0", "base64 0.22.1", "brotli", "bytes", @@ -10012,12 +10383,13 @@ dependencies = [ "flate2", "futures", "half", - "hashbrown 0.16.1", - "lz4_flex", + "hashbrown 0.17.1", + "lz4_flex 0.13.1", "num-bigint", "num-integer", "num-traits", "paste", + "ring", "seq-macro", "simdutf8", "snap", @@ -10045,8 +10417,8 @@ version = "0.12.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8ed6a7761f76e3b9f92dfb0a60a6a6477c61024b775147ff0973a02653abaf2" dependencies = [ - "digest", - "hmac", + "digest 0.10.7", + "hmac 0.12.1", ] [[package]] @@ -10089,7 +10461,7 @@ name = "persistcli" version = "0.0.0" dependencies = [ "anyhow", - "arrow", + "arrow 58.4.0", "async-trait", "axum", "bytes", @@ -10158,7 +10530,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "602113b5b5e8621770cfd490cfd90b9f84ab29bd2b0e49ad83eb6d186cef2365" dependencies = [ "pest", - "sha2", + "sha2 0.10.9", ] [[package]] @@ -10346,7 +10718,7 @@ dependencies = [ "der", "pbkdf2", "scrypt", - "sha2", + "sha2 0.10.9", "spki", ] @@ -10411,6 +10783,18 @@ dependencies = [ "windows-sys 0.59.0", ] +[[package]] +name = "polyval" +version = "0.6.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9d1fe60d06143b2430aa532c94cfe9e29783047f06c0d7fd359a9a51b729fa25" +dependencies = [ + "cfg-if", + "cpufeatures 0.2.17", + "opaque-debug", + "universal-hash", +] + [[package]] name = "portable-atomic" version = "1.13.1" @@ -10459,11 +10843,11 @@ dependencies = [ "byteorder", "bytes", "fallible-iterator 0.2.0", - "hmac", - "md-5", + "hmac 0.12.1", + "md-5 0.10.6", "memchr", "rand 0.9.4", - "sha2", + "sha2 0.10.9", "stringprep", ] @@ -10761,7 +11145,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "03da047801ff44bb6a4d407d4860c05fd70bb81714e6b2f3812603d5b145b042" dependencies = [ "heck", - "itertools 0.10.5", + "itertools 0.14.0", "log", "multimap", "petgraph", @@ -10782,7 +11166,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf" dependencies = [ "anyhow", - "itertools 0.10.5", + "itertools 0.14.0", "proc-macro2", "quote", "syn 2.0.119", @@ -10965,9 +11349,9 @@ dependencies = [ [[package]] name = "quick-xml" -version = "0.37.5" +version = "0.39.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "331e97a1af0bf59823e6eadffe373d7b27f485be8748f71471c662c1f269b7fb" +checksum = "cdcc8dd4e2f670d309a5f0e83fe36dfdc05af317008fea29144da1a2ac858e5e" dependencies = [ "memchr", "serde", @@ -10975,9 +11359,9 @@ dependencies = [ [[package]] name = "quick-xml" -version = "0.38.4" +version = "0.41.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b66c2058c55a409d601666cffe35f04333cf1013010882cec174a7467cd4e21c" +checksum = "e660451e55124f798a69a5af3f49ccfbefbd41910eefd25caf2393e1f3473ec1" dependencies = [ "memchr", "serde", @@ -11349,33 +11733,92 @@ dependencies = [ ] [[package]] -name = "reqsign" -version = "0.16.5" +name = "reqsign-aws-core" +version = "3.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "43451dbf3590a7590684c25fb8d12ecdcc90ed3ac123433e500447c7d77ed701" +checksum = "4d63b56638bb3cc7bd376a7cdce1ba3089777a08f47e4097888f2d784cc3f46c" dependencies = [ - "anyhow", - "async-trait", - "base64 0.22.1", - "chrono", + "bytes", "form_urlencoded", - "getrandom 0.2.16", "hex", - "hmac", - "home", "http 1.4.2", - "jsonwebtoken 9.3.1", "log", "percent-encoding", - "quick-xml 0.37.5", - "rand 0.8.5", - "reqwest 0.12.28", - "rsa", + "quick-xml 0.41.0", + "reqsign-core", "rust-ini", "serde", "serde_json", - "sha1", - "sha2", + "serde_urlencoded", + "sha1 0.11.0", +] + +[[package]] +name = "reqsign-aws-v4" +version = "3.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4a0c499f4ed12d04c3d4c78fe4cb01aee22c9dae22848c14db2c6313d9df9f43" +dependencies = [ + "bytes", + "http 1.4.2", + "log", + "quick-xml 0.41.0", + "reqsign-aws-core", + "reqsign-core", + "serde", +] + +[[package]] +name = "reqsign-core" +version = "3.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f4ac1510872d9481205975d264deb39c109797e5068cc882ed9064270eaae5fa" +dependencies = [ + "anyhow", + "base64 0.23.1", + "bytes", + "futures", + "hex", + "hmac 0.13.0", + "http 1.4.2", + "jiff", + "log", + "percent-encoding", + "rsa", + "serde", + "serde_json", + "sha1 0.11.0", + "sha2 0.11.0", + "windows-sys 0.61.1", +] + +[[package]] +name = "reqsign-file-read-tokio" +version = "3.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "95c3371bfc7e5c7f9627a04133af3583fd6c28715e7c83f79db38f3b384f535f" +dependencies = [ + "anyhow", + "reqsign-core", + "tokio", +] + +[[package]] +name = "reqsign-google" +version = "3.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f81a9d38870892443489c0abb5332edfa81d5a14c437af9caef7c194897c92c1" +dependencies = [ + "bytes", + "form_urlencoded", + "http 1.4.2", + "log", + "percent-encoding", + "reqsign-aws-v4", + "reqsign-core", + "rsa", + "serde", + "serde_json", "tokio", ] @@ -11421,7 +11864,7 @@ dependencies = [ "url", "wasm-bindgen", "wasm-bindgen-futures", - "wasm-streams", + "wasm-streams 0.4.1", "web-sys", ] @@ -11453,12 +11896,14 @@ dependencies = [ "sync_wrapper", "tokio", "tokio-native-tls", + "tokio-util", "tower 0.5.3", "tower-http", "tower-service", "url", "wasm-bindgen", "wasm-bindgen-futures", + "wasm-streams 0.5.0", "web-sys", ] @@ -11525,7 +11970,7 @@ version = "0.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8dd2a808d456c4a54e300a23e9f5a67e122c3024119acbfd73e3bf664491cb2" dependencies = [ - "hmac", + "hmac 0.12.1", "subtle", ] @@ -11637,15 +12082,15 @@ version = "0.9.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b8573f03f5883dcaebdfcf4725caa1ecb9c15b2ef50c43a07b816e06799bb12d" dependencies = [ - "const-oid", - "digest", + "const-oid 0.9.6", + "digest 0.10.7", "num-bigint-dig", "num-integer", "num-traits", "pkcs1", "pkcs8", "rand_core 0.6.4", - "sha2", + "sha2 0.10.9", "signature", "spki", "subtle", @@ -11747,7 +12192,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.59.0", + "windows-sys 0.61.1", ] [[package]] @@ -11917,7 +12362,7 @@ checksum = "0516a385866c09368f0b5bcd1caff3366aace790fcd46e2bb032697bb172fd1f" dependencies = [ "pbkdf2", "salsa20", - "sha2", + "sha2 0.10.9", ] [[package]] @@ -12348,7 +12793,18 @@ checksum = "e3bf829a2d51ab4a5ddf1352d8470c140cadc8301b2ae1789db023f01cedd6ba" dependencies = [ "cfg-if", "cpufeatures 0.2.17", - "digest", + "digest 0.10.7", +] + +[[package]] +name = "sha1" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aacc4cc499359472b4abe1bf11d0b12e688af9a805fa5e3016f9a386dc2d0214" +dependencies = [ + "cfg-if", + "cpufeatures 0.3.0", + "digest 0.11.3", ] [[package]] @@ -12365,10 +12821,21 @@ checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" dependencies = [ "cfg-if", "cpufeatures 0.2.17", - "digest", + "digest 0.10.7", "sha2-asm", ] +[[package]] +name = "sha2" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "446ba717509524cb3f22f17ecc096f10f4822d76ab5c0b9822c5f9c284e825f4" +dependencies = [ + "cfg-if", + "cpufeatures 0.3.0", + "digest 0.11.3", +] + [[package]] name = "sha2-asm" version = "0.6.4" @@ -12420,7 +12887,7 @@ version = "2.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "77549399552de45a898a580c1b41d445bf730df867cc44e6c0233bbc4b8329de" dependencies = [ - "digest", + "digest 0.10.7", "rand_core 0.6.4", ] @@ -12492,12 +12959,9 @@ checksum = "56199f7ddabf13fe5074ce809e7d3f42b42ae711800501b5b16ea82ad029c39d" [[package]] name = "slab" -version = "0.4.9" +version = "0.4.12" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8f92a496fb766b417c996b9c5e57daf2f7ad3b0bebe1ccfca4856390e3d3bb67" -dependencies = [ - "autocfg", -] +checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" [[package]] name = "smallvec" @@ -12596,7 +13060,7 @@ checksum = "eb9242b9ef4108a78e8cd1a2c98e193ef372437f8c22be363075233321dd4a15" dependencies = [ "base64ct", "pem-rfc7468", - "sha2", + "sha2 0.10.9", ] [[package]] @@ -12611,7 +13075,7 @@ dependencies = [ "rand_core 0.6.4", "rsa", "sec1", - "sha2", + "sha2 0.10.9", "signature", "ssh-cipher", "ssh-encoding", @@ -12898,10 +13362,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand 2.3.0", - "getrandom 0.3.3", + "getrandom 0.4.3", "once_cell", "rustix 1.1.4", - "windows-sys 0.59.0", + "windows-sys 0.61.1", ] [[package]] @@ -12920,7 +13384,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "230a1b821ccbd75b185820a1f1ff7b14d21da1e442e22c0863ea5f08771a8874" dependencies = [ "rustix 1.1.4", - "windows-sys 0.59.0", + "windows-sys 0.61.1", ] [[package]] @@ -13796,7 +14260,7 @@ dependencies = [ "httparse", "log", "rand 0.9.4", - "sha1", + "sha1 0.10.6", "thiserror 2.0.18", ] @@ -14002,6 +14466,16 @@ version = "0.5.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "81e544489bf3d8ef66c953931f56617f423cd4b5494be343d9b9d3dda037b9a3" +[[package]] +name = "universal-hash" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc1de2c688dc15305988b563c3854064043356019f97a4b46276fe734c4f07ea" +dependencies = [ + "crypto-common 0.1.6", + "subtle", +] + [[package]] name = "unsafe-libyaml" version = "0.2.11" @@ -14302,6 +14776,19 @@ dependencies = [ "web-sys", ] +[[package]] +name = "wasm-streams" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9d1ec4f6517c9e11ae630e200b2b65d193279042e28edd4a2cda233e46670bbb" +dependencies = [ + "futures-util", + "js-sys", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", +] + [[package]] name = "wasmtimer" version = "0.4.3" @@ -14462,15 +14949,6 @@ dependencies = [ "windows-link 0.1.1", ] -[[package]] -name = "windows-sys" -version = "0.48.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "677d2418bec65e3338edb076e806bc1ec15693c5d0104683f2efe857f61056a9" -dependencies = [ - "windows-targets 0.48.0", -] - [[package]] name = "windows-sys" version = "0.52.0" diff --git a/Cargo.toml b/Cargo.toml index 41bdd81398750..9272509b7a53c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -278,8 +278,8 @@ annotate-snippets = "0.12.15" anyhow = "1.0.102" array-concat = "0.5.5" arrayvec = "0.7.6" -arrow = { version = "57", default-features = false } -arrow-ipc = "57" +arrow = { version = "58", default-features = false } +arrow-ipc = "58" askama = { version = "0.12.1", default-features = false, features = ["config", "serde-json"] } assert_cmd = "2.2.2" async-compression = { version = "0.4.27", features = ["bzip2-sys", "gzip", "tokio", "xz", "zstd"] } @@ -384,9 +384,9 @@ hyper-0-14 = { package = "hyper", version = "0.14", features = ["client", "tcp"] hyper-openssl = "0.10.2" hyper-util = "0.1.20" tower-service = "0.3.3" -iceberg = "0.9.0" -iceberg-catalog-rest = "0.9.0" -iceberg-storage-opendal = { version = "0.9.0", default-features = false, features = ["opendal-s3", "opendal-gcs"] } +iceberg = "0.10.1" +iceberg-catalog-rest = "0.10.1" +iceberg-storage-opendal = { version = "0.10.1", default-features = false, features = ["opendal-s3", "opendal-gcs"] } imbl = { version = "7.0.0", features = ["serde"] } include_dir = "0.7.4" indexmap = { version = "2.10.0", default-features = false, features = ["std"] } @@ -434,7 +434,7 @@ opentelemetry_sdk = { version = "0.32.1", features = ["experimental_trace_batch_ ordered-float = { version = "5.1.0", features = ["serde"] } os_info = "3.11.0" owo-colors = "4.3.0" -parquet = { version = "57", default-features = false, features = ["arrow", "async", "brotli", "flate2", "flate2-zlib-rs", "lz4", "snap", "zstd"] } +parquet = { version = "58", default-features = false, features = ["arrow", "async", "brotli", "flate2", "flate2-zlib-rs", "lz4", "snap", "zstd"] } paste = "1.0.15" pem = "4.0.0" phf = { version = "0.13.1", features = ["uncased"] } @@ -473,6 +473,7 @@ rdkafka = { version = "0.29.0", features = ["cmake-build", "libz-static", "ssl-v rdkafka-sys = { version = "4.3.0", features = ["cmake-build", "libz-static", "ssl-vendored", "zstd"] } regex = "1.12.3" regex-syntax = "0.8.10" +reqsign-core = "3.3.0" reqwest = { version = "0.12.28", features = ["blocking", "charset", "cookies", "default-tls", "http2", "json", "native-tls-vendored", "stream"] } reqwest-middleware = { version = "0.4.2", features = ["json"] } reqwest-retry = "0.8.0" @@ -668,10 +669,13 @@ tiberius = { git = "https://github.com/MaterializeInc/tiberius", rev="64ca594cc2 async-compression = { git = "https://github.com/MaterializeInc/async-compression.git", rev = "fe7411eb6104a02a89e2c3a76ab326dd6594214d" } # Custom iceberg features for mz -# All changes should go to the `mz_v0.9.0` branch. -iceberg = { git = "https://github.com/MaterializeInc/iceberg-rust.git", rev = "dedd9231ee88ee979b648e14792878b40e74c20a" } -iceberg-catalog-rest = { git = "https://github.com/MaterializeInc/iceberg-rust.git", rev = "dedd9231ee88ee979b648e14792878b40e74c20a" } -iceberg-storage-opendal = { git = "https://github.com/MaterializeInc/iceberg-rust.git", rev = "dedd9231ee88ee979b648e14792878b40e74c20a" } +# All changes should go to the `mz_v0.10.x` branch. +# NOTE: The `[workspace.dependencies]` version requirement above must stay +# semver-compatible with this revision's crate version, otherwise Cargo drops +# these patches into `[[patch.unused]]` and silently builds against crates.io. +iceberg = { git = "https://github.com/MaterializeInc/iceberg-rust.git", rev = "958ac5b3c318026aeb362485c593e4956aeabe95" } +iceberg-catalog-rest = { git = "https://github.com/MaterializeInc/iceberg-rust.git", rev = "958ac5b3c318026aeb362485c593e4956aeabe95" } +iceberg-storage-opendal = { git = "https://github.com/MaterializeInc/iceberg-rust.git", rev = "958ac5b3c318026aeb362485c593e4956aeabe95" } # Custom duckdb crate to support mz needs # All changes should go to the `mz_changes` branch. diff --git a/LICENSE b/LICENSE index 936bb0dc3db90..96f1f15d4fca1 100644 --- a/LICENSE +++ b/LICENSE @@ -13,7 +13,7 @@ Business Source License 1.1 Licensor: Materialize, Inc. -Licensed Work: Materialize Version 20260827 +Licensed Work: Materialize Version 20260828 The Licensed Work is © 2026 Materialize, Inc. Additional Use Grant: Within a single installation of the Licensed Work, @@ -39,7 +39,7 @@ Additional Use Grant: Within a single installation of the Licensed Work, whose definitions are controlled by such third parties. -Change Date: August 27, 2030 +Change Date: August 28, 2030 Change License: Apache License, Version 2.0 diff --git a/deny.toml b/deny.toml index 1ce5b8f5d0566..dc0ac13c81ab5 100644 --- a/deny.toml +++ b/deny.toml @@ -102,10 +102,8 @@ skip = [ { name = "socket2", version = "0.5.10" }, # Used by azure_core { name = "quick-xml", version = "0.31.0" }, - # Used by reqsign (via iceberg); opendal pulls quick-xml 0.38 - { name = "quick-xml", version = "0.37.5" }, - # Used by opendal via iceberg - { name = "quick-xml", version = "0.38.4" }, + # reqsign-aws (via iceberg) is ahead of opendal, which pulls 0.39. + { name = "quick-xml", version = "0.41.0" }, # Used by sentry { name = "reqwest", version = "0.13.3" }, # Conflicts between `bon` in apache-avro and `derive_builder` in iceberg @@ -126,15 +124,39 @@ skip = [ { name = "fallible-iterator", version = "0.3.0" }, # arrow { name = "hashbrown", version = "0.16.1" }, + { name = "hashbrown", version = "0.17.1" }, + # duckdb pins arrow 57 while the workspace is on 58, so mz-testdrive links + # both majors. They never meet: duckdb's arrow stays inside duckdb's own API. + { name = "arrow", version = "57.1.0" }, + { name = "arrow-arith", version = "57.1.0" }, + { name = "arrow-array", version = "57.2.0" }, + { name = "arrow-buffer", version = "57.3.0" }, + { name = "arrow-cast", version = "57.2.0" }, + { name = "arrow-data", version = "57.2.0" }, + { name = "arrow-ord", version = "57.2.0" }, + { name = "arrow-row", version = "57.1.0" }, + { name = "arrow-schema", version = "57.2.0" }, + { name = "arrow-select", version = "57.2.0" }, + { name = "arrow-string", version = "57.1.0" }, + # opendal 0.57 and reqsign 3.x (both via iceberg) are on the next generation + # of the RustCrypto crates, while the rest of the workspace is still on the + # current one. + { name = "base64", version = "0.23.1" }, + { name = "block-buffer", version = "0.12.1" }, + { name = "const-oid", version = "0.10.2" }, + { name = "crypto-common", version = "0.2.2" }, + { name = "digest", version = "0.11.3" }, + { name = "hmac", version = "0.13.0" }, + { name = "md-5", version = "0.11.0" }, + { name = "sha1", version = "0.11.0" }, + { name = "sha2", version = "0.11.0" }, + # Held back by mz-timely-util; iceberg and the persist crates use 0.13. + { name = "lz4_flex", version = "0.12.1" }, # Used by dynfmt; iceberg/typetag pulls in v0.4. { name = "erased-serde", version = "0.3.26" }, # gcp_auth → hyper-rustls → rustls-native-certs pulls newer versions # while native-tls still pulls older versions. { name = "core-foundation", version = "0.10.1" }, - # reqsign (via iceberg-storage-opendal / opendal) pins older deps - # than the workspace. - { name = "jsonwebtoken", version = "9.3.1" }, - { name = "quick-xml", version = "0.37.5" }, # aws-lc-rs (via jsonwebtoken 10) and ring pull different `untrusted`. { name = "untrusted", version = "0.7.1" }, # Held back by lazy_static 1.4.0 (used by num-bigint-dig). @@ -211,14 +233,20 @@ wrappers = [ "launchdarkly-server-sdk-evaluation", "launchdarkly-sdk-transport", "native-tls", - "opendal", + "opendal-core", + "opendal-layer-retry", + "opendal-service-gcs", + "opendal-service-s3", "os_info", "postgres", "pprof", "prost-build", # TODO(guswynn): switch to tracing in rdkafka "rdkafka", - "reqsign", + "reqsign-aws-core", + "reqsign-aws-v4", + "reqsign-core", + "reqsign-google", "reqwest", "rustls", "sqlparser", diff --git a/misc/python/materialize/mzcompose/helpers/iceberg.py b/misc/python/materialize/mzcompose/helpers/iceberg.py index b31391f28293d..8568914c67b3a 100644 --- a/misc/python/materialize/mzcompose/helpers/iceberg.py +++ b/misc/python/materialize/mzcompose/helpers/iceberg.py @@ -98,21 +98,36 @@ def create_polaris_catalog( secret_key: str = "", endpoint: str = "http://minio:9000", region: str = "minio", + static_credentials: bool = True, ) -> None: + """Create a Polaris catalog backed by `bucket_name` in MinIO. + + Catalog properties are returned to clients verbatim on `loadTable`, so the + `s3.access-key-id`/`s3.secret-access-key` written here become the credentials + every client uses. Pass `static_credentials=False` to leave them out, which + makes credential vending the only way a client can reach the bucket. Polaris + itself still reaches MinIO through the credentials in its environment. + """ + properties = { + "default-base-location": f"s3://{bucket_name}/", + "s3.endpoint": endpoint, + "s3.path-style-access": "true", + "s3.region": region, + } + if static_credentials: + properties["s3.access-key-id"] = username + properties["s3.secret-access-key"] = secret_key + catalog_payload = { "name": catalog_name, "type": "INTERNAL", - "properties": { - "default-base-location": f"s3://{bucket_name}/", - "s3.endpoint": endpoint, - "s3.path-style-access": "true", - "s3.access-key-id": username, - "s3.secret-access-key": secret_key, - "s3.region": region, - }, + "properties": properties, "storageConfigInfo": { "storageType": "S3", - "allowedLocations": [f"s3://{bucket_name}/*"], + # Allowed locations are prefixes, not globs. Polaris rejects the + # catalog unless `default-base-location` sits within one of them, and + # a trailing `/*` is matched literally, so it contains nothing. + "allowedLocations": [f"s3://{bucket_name}/"], "endpoint": endpoint, "endpointInternal": endpoint, "pathStyleAccess": True, @@ -124,6 +139,7 @@ def create_polaris_catalog( "curl", "-sS", "-i", + "--fail-with-body", "-X", "POST", "-H", @@ -136,6 +152,30 @@ def create_polaris_catalog( ) +def assert_polaris_catalog_exists( + c: "Composition", + access_token: str, + catalog_name: str = "default_catalog", +) -> None: + """Read the catalog back, failing here if it is missing. + + A rejected catalog creation otherwise stays invisible until whatever first + uses the warehouse reports a 404, which points the investigation at the + consumer instead of at setup. + """ + c.exec( + "polaris", + "curl", + "-sS", + "--fail-with-body", + "-X", + "GET", + "-H", + f"Authorization: Bearer {access_token}", + f"http://localhost:8181/api/management/v1/catalogs/{catalog_name}", + ) + + def create_polaris_namespace( c: "Composition", access_token: str, @@ -148,6 +188,7 @@ def create_polaris_namespace( "curl", "-sS", "-i", + "--fail-with-body", "-X", "POST", "-H", @@ -160,6 +201,31 @@ def create_polaris_namespace( ) +def grant_catalog_role_privilege( + c: "Composition", + access_token: str, + privilege: str, + catalog_name: str = "default_catalog", + catalog_role: str = "catalog_admin", +) -> None: + """Grant a catalog-level privilege to a catalog role.""" + c.exec( + "polaris", + "curl", + "-sS", + "--fail-with-body", + "-X", + "PUT", + "-H", + f"Authorization: Bearer {access_token}", + "-H", + "Content-Type: application/json", + f"http://localhost:8181/api/management/v1/catalogs/{catalog_name}/catalog-roles/{catalog_role}/grants", + "-d", + json.dumps({"type": "catalog", "privilege": privilege}), + ) + + def setup_polaris_for_iceberg( c: "Composition", bucket_name: str = "test-bucket", @@ -167,6 +233,8 @@ def setup_polaris_for_iceberg( username: str = "tduser", catalog_name: str = "default_catalog", namespace: str = "default_namespace", + vended: bool = False, + static_credentials: bool = True, ) -> tuple[str, str]: """ Set up Polaris catalog with MinIO for Iceberg sink usage. @@ -176,6 +244,17 @@ def setup_polaris_for_iceberg( 2. Creating a MinIO user with S3 permissions 3. Starting Polaris with the user's credentials 4. Creating a catalog and namespace in Polaris + + With `vended=True`, also grant the catalog role the `TABLE_READ_DATA` and + `TABLE_WRITE_DATA` privileges. These authorize credential vending, so a + client sending the `X-Iceberg-Access-Delegation: vended-credentials` header + on `loadTable`/`commit` receives temporary, table-scoped MinIO STS + credentials instead of using its own static ones. Polaris mints them via + AssumeRole against MinIO using the credentials passed to it below. + + With `static_credentials=False`, the catalog withholds the long-lived S3 + credentials it would otherwise hand every client, so vending becomes the only + path to the bucket. Combine with `vended=True` to require vending. """ from materialize.mzcompose.composition import Service @@ -211,8 +290,11 @@ def setup_polaris_for_iceberg( bucket_name=bucket_name, username=username, secret_key=key, + static_credentials=static_credentials, ) + assert_polaris_catalog_exists(c, access_token, catalog_name=catalog_name) + create_polaris_namespace( c, access_token, @@ -220,4 +302,10 @@ def setup_polaris_for_iceberg( catalog_name=catalog_name, ) + if vended: + for privilege in ("TABLE_READ_DATA", "TABLE_WRITE_DATA"): + grant_catalog_role_privilege( + c, access_token, privilege, catalog_name=catalog_name + ) + return (username, key) diff --git a/misc/python/materialize/mzcompose/services/polaris.py b/misc/python/materialize/mzcompose/services/polaris.py index 662ea809185c5..5b95216b6cd4b 100644 --- a/misc/python/materialize/mzcompose/services/polaris.py +++ b/misc/python/materialize/mzcompose/services/polaris.py @@ -18,7 +18,7 @@ def __init__( self, name: str = "polaris-bootstrap", image: str = "apache/polaris-admin-tool", - tag: str = "1.2.0-incubating", + tag: str = "1.7.0", environment: list[str] = [ "POLARIS_BOOTSTRAP_CREDENTIALS=POLARIS,root,root", "POLARIS_PERSISTENCE_TYPE=relational-jdbc", @@ -50,8 +50,9 @@ def __init__( self, name: str = "polaris", image: str = "apache/polaris", - # Fails with 1.1.0-incubating - tag: str = "1.2.0-incubating", + # Fails with 1.1.0-incubating. Releases from 1.4.0 on drop the + # `-incubating` suffix, so keep this tag in sync with the admin tool's. + tag: str = "1.7.0", # 8181: api port, 8182: management port ports: list[str | int] = [8181, 8182], environment: list[str] = [ diff --git a/src/persist-types/src/parquet.rs b/src/persist-types/src/parquet.rs index de264ff500b82..adbf302af0a99 100644 --- a/src/persist-types/src/parquet.rs +++ b/src/persist-types/src/parquet.rs @@ -176,7 +176,7 @@ pub fn encode_arrays( .set_compression(config.compression.into()) .set_writer_version(WriterVersion::PARQUET_2_0) .set_data_page_size_limit(1024 * 1024) - .set_max_row_group_size(usize::MAX) + .set_max_row_group_row_count(None) .build(); let mut writer = ArrowWriter::try_new(w, Arc::clone(&schema), Some(props))?; diff --git a/src/persist/src/indexed/columnar/parquet.rs b/src/persist/src/indexed/columnar/parquet.rs index 4727c459cd724..45957a3d8f247 100644 --- a/src/persist/src/indexed/columnar/parquet.rs +++ b/src/persist/src/indexed/columnar/parquet.rs @@ -118,7 +118,7 @@ pub fn encode_parquet_kvtd( .set_compression(cfg.compression.into()) .set_writer_version(WriterVersion::PARQUET_2_0) .set_data_page_size_limit(1024 * 1024) - .set_max_row_group_size(usize::MAX) + .set_max_row_group_row_count(None) .set_key_value_metadata(Some(vec![metadata])) .build(); diff --git a/src/sql-lexer/src/keywords.txt b/src/sql-lexer/src/keywords.txt index ad04025ccbaff..ef62195f04d1e 100644 --- a/src/sql-lexer/src/keywords.txt +++ b/src/sql-lexer/src/keywords.txt @@ -145,6 +145,7 @@ Decoding Decorrelated Default Defaults +Delegation Delete Delimited Delimiter @@ -336,6 +337,7 @@ Notices Null Nullif Nulls +Oauth2 Objects Of Offset diff --git a/src/sql-parser/src/ast/defs/ddl.rs b/src/sql-parser/src/ast/defs/ddl.rs index 42e4e9a34708c..5a591458609b7 100644 --- a/src/sql-parser/src/ast/defs/ddl.rs +++ b/src/sql-parser/src/ast/defs/ddl.rs @@ -893,6 +893,7 @@ impl_display_t!(Format); // souffrir pour être belle. #[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] pub enum ConnectionOptionName { + AccessDelegation, AccessKeyId, AssumeRoleArn, AssumeRoleSessionName, @@ -906,6 +907,7 @@ pub enum ConnectionOptionName { Endpoint, GcpConnection, Host, + Oauth2ServerUrl, Password, Port, ProgressTopic, @@ -949,7 +951,8 @@ impl ConnectionOptionName { | ConnectionOptionName::SslCertificateAuthority | ConnectionOptionName::SslKey | ConnectionOptionName::User => true, - ConnectionOptionName::AssumeRoleArn + ConnectionOptionName::AccessDelegation + | ConnectionOptionName::AssumeRoleArn | ConnectionOptionName::AssumeRoleSessionName | ConnectionOptionName::AvailabilityZones | ConnectionOptionName::AwsConnection @@ -960,6 +963,7 @@ impl ConnectionOptionName { | ConnectionOptionName::Endpoint | ConnectionOptionName::GcpConnection | ConnectionOptionName::Host + | ConnectionOptionName::Oauth2ServerUrl | ConnectionOptionName::Port | ConnectionOptionName::ProgressTopic | ConnectionOptionName::ProgressTopicReplicationFactor @@ -983,6 +987,7 @@ impl ConnectionOptionName { impl AstDisplay for ConnectionOptionName { fn fmt(&self, f: &mut AstFormatter) { f.write_str(match self { + ConnectionOptionName::AccessDelegation => "ACCESS DELEGATION", ConnectionOptionName::AccessKeyId => "ACCESS KEY ID", ConnectionOptionName::AvailabilityZones => "AVAILABILITY ZONES", ConnectionOptionName::AwsConnection => "AWS CONNECTION", @@ -994,6 +999,7 @@ impl AstDisplay for ConnectionOptionName { ConnectionOptionName::Endpoint => "ENDPOINT", ConnectionOptionName::GcpConnection => "GCP CONNECTION", ConnectionOptionName::Host => "HOST", + ConnectionOptionName::Oauth2ServerUrl => "OAUTH2 SERVER URL", ConnectionOptionName::Password => "PASSWORD", ConnectionOptionName::Port => "PORT", ConnectionOptionName::ProgressTopic => "PROGRESS TOPIC", diff --git a/src/sql-parser/src/parser.rs b/src/sql-parser/src/parser.rs index 94a6b1ea5851f..d878a936530da 100644 --- a/src/sql-parser/src/parser.rs +++ b/src/sql-parser/src/parser.rs @@ -2974,6 +2974,7 @@ impl<'a> Parser<'a> { ENDPOINT, GCP, HOST, + OAUTH2, PASSWORD, PORT, PUBLIC, @@ -2993,10 +2994,14 @@ impl<'a> Parser<'a> { USERNAME, WAREHOUSE, ])? { - ACCESS => { - self.expect_keywords(&[KEY, ID])?; - ConnectionOptionName::AccessKeyId - } + ACCESS => match self.expect_one_of_keywords(&[KEY, DELEGATION])? { + KEY => { + self.expect_keyword(ID)?; + ConnectionOptionName::AccessKeyId + } + DELEGATION => ConnectionOptionName::AccessDelegation, + _ => unreachable!(), + }, ASSUME => { self.expect_keyword(ROLE)?; match self.expect_one_of_keywords(&[ARN, SESSION])? { @@ -3031,6 +3036,10 @@ impl<'a> Parser<'a> { ConnectionOptionName::GcpConnection } HOST => ConnectionOptionName::Host, + OAUTH2 => { + self.expect_keywords(&[SERVER, URL])?; + ConnectionOptionName::Oauth2ServerUrl + } PASSWORD => ConnectionOptionName::Password, PORT => ConnectionOptionName::Port, PUBLIC => { diff --git a/src/sql-parser/tests/testdata/ddl b/src/sql-parser/tests/testdata/ddl index d2f17378e471e..7db725257479f 100644 --- a/src/sql-parser/tests/testdata/ddl +++ b/src/sql-parser/tests/testdata/ddl @@ -578,6 +578,13 @@ CREATE CONNECTION awsconn TO AWS (ACCESS KEY ID = 'id', ENDPOINT = 'endpoint', R => CreateConnection(CreateConnectionStatement { name: UnresolvedItemName([Ident("awsconn")]), connection_type: Aws, if_not_exists: false, values: [ConnectionOption { name: AccessKeyId, value: Some(Value(String("id"))) }, ConnectionOption { name: Endpoint, value: Some(Value(String("endpoint"))) }, ConnectionOption { name: Region, value: Some(Value(String("region"))) }, ConnectionOption { name: AssumeRoleArn, value: Some(Value(String("role-arn"))) }, ConnectionOption { name: SecretAccessKey, value: Some(Value(String("key"))) }, ConnectionOption { name: SessionToken, value: Some(Value(String("token"))) }], with_options: [] }) +parse-statement +CREATE CONNECTION icebergconn TO ICEBERG CATALOG (CATALOG TYPE 'rest', URL 'https://example/api/catalog', CREDENTIAL 'id:secret', OAUTH2 SERVER URL 'https://example/oidc/v1/token', SCOPE 'all-apis', WAREHOUSE 'wh') +---- +CREATE CONNECTION icebergconn TO ICEBERG CATALOG (CATALOG TYPE = 'rest', URL = 'https://example/api/catalog', CREDENTIAL = 'id:secret', OAUTH2 SERVER URL = 'https://example/oidc/v1/token', SCOPE = 'all-apis', WAREHOUSE = 'wh') +=> +CreateConnection(CreateConnectionStatement { name: UnresolvedItemName([Ident("icebergconn")]), connection_type: IcebergCatalog, if_not_exists: false, values: [ConnectionOption { name: CatalogType, value: Some(Value(String("rest"))) }, ConnectionOption { name: Url, value: Some(Value(String("https://example/api/catalog"))) }, ConnectionOption { name: Credential, value: Some(Value(String("id:secret"))) }, ConnectionOption { name: Oauth2ServerUrl, value: Some(Value(String("https://example/oidc/v1/token"))) }, ConnectionOption { name: Scope, value: Some(Value(String("all-apis"))) }, ConnectionOption { name: Warehouse, value: Some(Value(String("wh"))) }], with_options: [] }) + parse-statement CREATE CONNECTION privatelinkconn TO AWS PRIVATELINK (SERVICE NAME 'com.amazonaws.vpce.us-east-1.vpce-svc-0e123abc123198abc', AVAILABILITY ZONES ('use1-az1', 'use1-az4')) ---- @@ -3033,7 +3040,7 @@ CreateConnection(CreateConnectionStatement { name: UnresolvedItemName([Ident("my parse-statement CREATE CONNECTION my_ssh_tunnel FOR SSH TUNNEL (PUBLIC KEY 3 = nope) ---- -error: Expected one of ACCESS or ASSUME or AVAILABILITY or AWS or BROKER or BROKERS or CATALOG or CREDENTIAL or DATABASE or ENDPOINT or GCP or HOST or PASSWORD or PORT or PUBLIC or PROGRESS or REGION or REGISTRY or SASL or SCOPE or SECRET or SECURITY or SERVICE or SESSION or SSH or SSL or URL or USER or USERNAME or WAREHOUSE, found left parenthesis +error: Expected one of ACCESS or ASSUME or AVAILABILITY or AWS or BROKER or BROKERS or CATALOG or CREDENTIAL or DATABASE or ENDPOINT or GCP or HOST or OAUTH2 or PASSWORD or PORT or PUBLIC or PROGRESS or REGION or REGISTRY or SASL or SCOPE or SECRET or SECURITY or SERVICE or SESSION or SSH or SSL or URL or USER or USERNAME or WAREHOUSE, found left parenthesis CREATE CONNECTION my_ssh_tunnel FOR SSH TUNNEL (PUBLIC KEY 3 = nope) ^ diff --git a/src/sql/src/plan/statement/ddl/connection.rs b/src/sql/src/plan/statement/ddl/connection.rs index a38850b27c3c7..cf333d0b1ab93 100644 --- a/src/sql/src/plan/statement/ddl/connection.rs +++ b/src/sql/src/plan/statement/ddl/connection.rs @@ -34,11 +34,11 @@ use mz_storage_types::connections::inline::ReferencedConnection; use mz_storage_types::connections::string_or_secret::StringOrSecret; use mz_storage_types::connections::{ AwsPrivatelink, AwsPrivatelinkConnection, AwsPrivatelinkRule, CsrConnection, - CsrConnectionHttpAuth, GlueSchemaRegistryConnection, IcebergCatalogAuth, - IcebergCatalogConnection, IcebergCatalogImpl, IcebergCatalogType, KafkaConnection, - KafkaSaslConfig, KafkaTlsConfig, KafkaTopicOptions, MySqlConnection, MySqlSslMode, - PostgresConnection, RestIcebergCatalog, S3TablesRestIcebergCatalog, SqlServerConnectionDetails, - SshConnection, SshTunnel, TlsIdentity, Tunnel, + CsrConnectionHttpAuth, GlueSchemaRegistryConnection, IcebergAccessDelegation, + IcebergCatalogAuth, IcebergCatalogConnection, IcebergCatalogImpl, IcebergCatalogType, + KafkaConnection, KafkaSaslConfig, KafkaTlsConfig, KafkaTopicOptions, MySqlConnection, + MySqlSslMode, PostgresConnection, RestIcebergCatalog, S3TablesRestIcebergCatalog, + SqlServerConnectionDetails, SshConnection, SshTunnel, TlsIdentity, Tunnel, }; use crate::names::Aug; @@ -49,6 +49,7 @@ use crate::session::vars; generate_extracted_config!( ConnectionOption, + (AccessDelegation, IcebergAccessDelegation), (AccessKeyId, StringOrSecret), (AssumeRoleArn, String), (AssumeRoleSessionName, String), @@ -62,6 +63,7 @@ generate_extracted_config!( (Endpoint, String), (GcpConnection, with_options::Object), (Host, String), + (Oauth2ServerUrl, String), (Password, with_options::Secret), (Port, u16), (ProgressTopic, String), @@ -190,10 +192,12 @@ pub(super) fn validate_options_per_connection_type( User, ], CreateConnectionType::IcebergCatalog => &[ + AccessDelegation, AwsConnection, CatalogType, Credential, GcpConnection, + Oauth2ServerUrl, Scope, Url, Warehouse, @@ -697,6 +701,16 @@ impl ConnectionOptionExtracted { "invalid CONNECTION: ICEBERG s3tablesrest connections do not support GCP CONNECTION" ); } + if self.oauth2_server_url.is_some() { + sql_bail!( + "invalid CONNECTION: ICEBERG s3tablesrest connections do not support OAUTH2 SERVER URL" + ); + } + if self.access_delegation.is_some() { + sql_bail!( + "invalid CONNECTION: ICEBERG s3tablesrest connections do not support ACCESS DELEGATION" + ); + } let Some(warehouse) = warehouse else { sql_bail!( "invalid CONNECTION: ICEBERG s3tablesrest connections must specify WAREHOUSE" @@ -726,8 +740,19 @@ impl ConnectionOptionExtracted { (Some(credential), None) => IcebergCatalogAuth::OAuth { credential, scope: self.scope.clone(), + server_url: self.oauth2_server_url.clone(), }, (None, Some(gcp_connection)) => { + if self.oauth2_server_url.is_some() { + sql_bail!( + "invalid CONNECTION: OAUTH2 SERVER URL applies to CREDENTIAL auth, not GCP CONNECTION" + ); + } + if self.access_delegation.is_some() { + sql_bail!( + "invalid CONNECTION: ICEBERG GCP CONNECTION does not support ACCESS DELEGATION" + ); + } /// All BigLake Iceberg REST Catalogs use the same catalog URI. const BIGLAKE_CATALOG_URI: &str = "https://biglake.googleapis.com/iceberg/v1/restcatalog"; @@ -744,7 +769,11 @@ impl ConnectionOptionExtracted { ), }; - IcebergCatalogImpl::Rest(RestIcebergCatalog { auth, warehouse }) + IcebergCatalogImpl::Rest(RestIcebergCatalog { + auth, + warehouse, + access_delegation: self.access_delegation, + }) } }; diff --git a/src/sql/src/plan/with_options.rs b/src/sql/src/plan/with_options.rs index 71ec13e986b0f..2c7342b0ae7b6 100644 --- a/src/sql/src/plan/with_options.rs +++ b/src/sql/src/plan/with_options.rs @@ -20,8 +20,8 @@ use mz_sql_parser::ast::{ ConnectionDefaultAwsPrivatelink, Expr, Ident, KafkaBroker, KafkaMatchingBrokerRule, NetworkPolicyRuleDefinition, RefreshOptionValue, ReplicaDefinition, }; -use mz_storage_types::connections::IcebergCatalogType; use mz_storage_types::connections::string_or_secret::StringOrSecret; +use mz_storage_types::connections::{IcebergAccessDelegation, IcebergCatalogType}; use serde::{Deserialize, Serialize}; use crate::ast::{AstInfo, UnresolvedItemName, Value, WithOptionValue}; @@ -68,6 +68,33 @@ impl ImpliedValue for IcebergCatalogType { } } +impl TryFromValue> for IcebergAccessDelegation { + fn try_from_value(v: WithOptionValue) -> Result { + match String::try_from_value(v)? { + s if s.eq_ignore_ascii_case("vended-credentials") => { + Ok(IcebergAccessDelegation::VendedCredentials) + } + _ => sql_bail!("invalid iceberg access delegation, expected 'vended-credentials'"), + } + } + + fn try_into_value(self, _catalog: &dyn SessionCatalog) -> Option> { + Some(WithOptionValue::Value(Value::String( + self.as_header_value().to_string(), + ))) + } + + fn name() -> String { + "iceberg access delegation".to_string() + } +} + +impl ImpliedValue for IcebergAccessDelegation { + fn implied_value() -> Result { + sql_bail!("must provide an iceberg access delegation") + } +} + #[derive(Copy, Clone, Debug)] pub struct Secret(CatalogItemId); diff --git a/src/sql/src/pure.rs b/src/sql/src/pure.rs index 82d6f4b79f05a..25c982728708a 100644 --- a/src/sql/src/pure.rs +++ b/src/sql/src/pure.rs @@ -625,8 +625,10 @@ async fn purify_create_sink( // Now that we've validated the sink's storage creds (if they exist) // we _could_ use them to build a complete Iceberg client (both catalog and storage). // TODO(kynan): Actually use those sink-specific creds here instead of ignoring them. + // Purification only proves the catalog is reachable, so it needs no table-scoped + // storage credentials. let _catalog = connection - .connect(storage_configuration, InTask::No) + .connect(storage_configuration, InTask::No, None) .await .map_err(|e| IcebergSinkPurificationError::CatalogError(Arc::new(e)))?; } diff --git a/src/storage-operators/src/s3_oneshot_sink/parquet.rs b/src/storage-operators/src/s3_oneshot_sink/parquet.rs index 8ee191c6c7bfa..4b0bcdd459df2 100644 --- a/src/storage-operators/src/s3_oneshot_sink/parquet.rs +++ b/src/storage-operators/src/s3_oneshot_sink/parquet.rs @@ -287,7 +287,7 @@ impl ParquetFile { let props = WriterProperties::builder() // This refers to the number of rows per row-group, which we don't want the writer // to enforce since we will flush based on the byte-size of the active row group - .set_max_row_group_size(usize::MAX) + .set_max_row_group_row_count(None) // Max compatibility .set_writer_version(WriterVersion::PARQUET_1_0) .set_compression(Compression::SNAPPY) diff --git a/src/storage-types/Cargo.toml b/src/storage-types/Cargo.toml index 41fb1f7b05b6f..f5f69445fdbce 100644 --- a/src/storage-types/Cargo.toml +++ b/src/storage-types/Cargo.toml @@ -65,6 +65,7 @@ proptest-derive = { workspace = true, optional = true } prost.workspace = true rdkafka.workspace = true regex.workspace = true +reqsign-core.workspace = true reqwest.workspace = true serde.workspace = true serde_json = { workspace = true, features = ["preserve_order"] } diff --git a/src/storage-types/src/connections.rs b/src/storage-types/src/connections.rs index b5977fcaba52b..aa48f6a599c23 100644 --- a/src/storage-types/src/connections.rs +++ b/src/storage-types/src/connections.rs @@ -24,18 +24,20 @@ use aws_sigv4::sign::v4; // Aliased to avoid colliding with `mz_ccsr::tls::Identity`. use aws_smithy_runtime_api::client::identity::Identity as AwsIdentity; use base64::Engine; -use http::{HeaderName, HeaderValue}; +use http::{HeaderMap, HeaderName, HeaderValue}; use iceberg::Catalog; use iceberg::CatalogBuilder; +use iceberg::TableIdent; use iceberg::io::{ GCS_CREDENTIALS_JSON, GCS_DISABLE_CONFIG_LOAD, GCS_DISABLE_VM_METADATA, GCS_USER_PROJECT, S3_ACCESS_KEY_ID, S3_DISABLE_EC2_METADATA, S3_REGION, S3_SECRET_ACCESS_KEY, }; use iceberg_catalog_rest::{ - REST_CATALOG_PROP_URI, REST_CATALOG_PROP_WAREHOUSE, RequestAuthenticator, RestCatalogBuilder, + OAuth2TokenProvider, REST_CATALOG_PROP_URI, REST_CATALOG_PROP_WAREHOUSE, RequestAuthenticator, + RestCatalogBuilder, TokenProvider, }; use iceberg_storage_opendal::{ - AwsCredential, AwsCredentialLoad, CustomAwsCredentialLoader, OpenDalStorageFactory, + AwsCredential, CustomAwsCredentialLoader, OpenDalStorageFactory, ProvideCredential, }; use itertools::Itertools; use mz_ccsr::tls::{Certificate, Identity}; @@ -63,6 +65,7 @@ use rdkafka::ClientContext; use rdkafka::config::FromClientConfigAndContext; use rdkafka::consumer::{BaseConsumer, Consumer}; use regex::Regex; +use reqsign_core::time::Timestamp; use reqwest::Request; use serde::{Deserialize, Deserializer, Serialize}; use tokio::net; @@ -88,11 +91,22 @@ use crate::errors::{ContextCreationError, CsrConnectError}; pub mod aws; pub mod gcp; +mod iceberg_credentials; pub mod inline; pub mod string_or_secret; -const REST_CATALOG_PROP_SCOPE: &str = "scope"; -const REST_CATALOG_PROP_CREDENTIAL: &str = "credential"; +/// The OAuth2 form field naming the scopes a token is requested for. +/// +/// Materialize drives the OAuth2 exchange itself rather than through the `credential`, +/// `oauth2-server-uri`, and `scope` catalog properties, so that one token object serves both +/// catalog requests and storage-credential refreshes. +const OAUTH2_PARAM_SCOPE: &str = "scope"; + +const REST_CATALOG_PROP_OAUTH2_SERVER_URI: &str = "oauth2-server-uri"; +/// Requests catalog-vended storage credentials. `iceberg-rust` turns `header.*` catalog +/// properties into headers on every REST request, the same way the Iceberg Java client +/// carries this one. +const REST_CATALOG_PROP_ACCESS_DELEGATION: &str = "header.X-Iceberg-Access-Delegation"; /// A credential loader that wraps an aws-sdk-rust credentials provider for use with /// iceberg/OpenDAL. This allows us to provide refreshable credentials from the AWS SDK @@ -101,6 +115,7 @@ const REST_CATALOG_PROP_CREDENTIAL: &str = "credential"; /// We use this instead of OpenDAL's built-in assume role support because Materialize /// has a runtime-defined credential chain (ambient → jump role → user role with external ID) /// that can't be expressed via OpenDAL's static configuration properties. +#[derive(Debug)] struct AwsSdkCredentialLoader { /// The underlying AWS SDK credentials provider. For assume role auth, this provider /// already handles the full chain: ambient creds -> jump role -> user role. @@ -113,37 +128,59 @@ impl AwsSdkCredentialLoader { } } -#[async_trait] -impl AwsCredentialLoad for AwsSdkCredentialLoader { - async fn load_credential( +impl ProvideCredential for AwsSdkCredentialLoader { + type Credential = AwsCredential; + + async fn provide_credential( &self, - _client: reqwest::Client, - ) -> anyhow::Result> { - let creds = self - .provider - .provide_credentials() - .await - .map_err(|e| { - warn!( - error = %e.display_with_causes(), - "failed to load AWS credentials for Iceberg FileIO from SDK provider" - ); - e - }) - .context( + _ctx: &reqsign_core::Context, + ) -> reqsign_core::Result> { + let creds = self.provider.provide_credentials().await.map_err(|e| { + warn!( + error = %e.display_with_causes(), + "failed to load AWS credentials for Iceberg FileIO from SDK provider" + ); + reqsign_core::Error::credential_invalid( "failed to load AWS credentials from SDK provider for Iceberg FileIO \ (credential source may be temporarily unavailable)", - )?; + ) + .with_source(e) + })?; + + // Propagate the SDK's expiry whenever it reports one. reqsign treats a `None` expiry as + // "valid forever", so dropping it would leave OpenDAL signing with stale assume-role + // credentials rather than asking us for fresh ones. + let expires_in = creds.expiry().map(aws_expiry_to_timestamp).transpose()?; Ok(Some(AwsCredential { access_key_id: creds.access_key_id().to_string(), secret_access_key: creds.secret_access_key().to_string(), session_token: creds.session_token().map(|s| s.to_string()), - expires_in: creds.expiry().map(|t| t.into()), + expires_in, })) } } +/// Converts an AWS SDK credential expiry into reqsign's [`Timestamp`]. +/// +/// Both failure modes require a nonsensical expiry (before the Unix epoch, or beyond year +/// 292278994), so they are reported as errors rather than silently dropped, which would make the +/// credential look non-expiring. +fn aws_expiry_to_timestamp(expiry: SystemTime) -> reqsign_core::Result { + let millis = expiry + .duration_since(SystemTime::UNIX_EPOCH) + .map_err(|e| { + reqsign_core::Error::unexpected("AWS credential expiry precedes the Unix epoch") + .with_source(e) + })? + .as_millis(); + let millis = i64::try_from(millis).map_err(|e| { + reqsign_core::Error::unexpected("AWS credential expiry overflows a millisecond timestamp") + .with_source(e) + })?; + Timestamp::from_millisecond(millis) +} + /// Signs each outgoing REST-catalog request with AWS SigV4. /// /// Holds a [`SharedCredentialsProvider`] (not static `Credentials`) so each @@ -611,6 +648,13 @@ pub enum IcebergCatalogAuth { credential: StringOrSecret, /// OAuth2 scope scope: Option, + /// Where to exchange `credential` for a bearer token. + /// + /// `None` uses the endpoint the Iceberg REST specification defines relative to the + /// catalog URL, `/v1/oauth/tokens`. Catalogs that host their token endpoint + /// elsewhere, or behind an auth gateway that will not serve an unauthenticated + /// exchange, need this override. + server_url: Option, }, Gcp(GcpConnectionReference), } @@ -620,6 +664,30 @@ pub struct RestIcebergCatalog { pub auth: IcebergCatalogAuth, /// The warehouse for REST catalogs pub warehouse: Option, + /// Which form of storage-access delegation to request from the catalog, if any. + /// + /// `None` means "do not request storage-access delegation". + /// If we do not have permission to request delegated access but request it anyway, + /// a catalog can reject our whole request, + /// even if we have our own storage credentials to fall back on. + pub access_delegation: Option, +} + +/// The value Materialize sends in the Iceberg REST `X-Iceberg-Access-Delegation` +/// header, naming how the catalog should grant access to table storage. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Hash, Serialize, Deserialize)] +pub enum IcebergAccessDelegation { + /// Ask the catalog to mint temporary, table-scoped storage credentials. + VendedCredentials, +} + +impl IcebergAccessDelegation { + /// The header value, as spelled in the Iceberg REST specification. + pub fn as_header_value(&self) -> &'static str { + match self { + IcebergAccessDelegation::VendedCredentials => "vended-credentials", + } + } } #[derive(Clone, Debug, Eq, PartialEq, Hash, Serialize, Deserialize)] @@ -636,9 +704,15 @@ impl IntoInlineConnection fn into_inline_connection(self, r: R) -> IcebergCatalogAuth { match self { IcebergCatalogAuth::Gcp(x) => IcebergCatalogAuth::Gcp(x.into_inline_connection(&r)), - IcebergCatalogAuth::OAuth { credential, scope } => { - IcebergCatalogAuth::OAuth { credential, scope } - } + IcebergCatalogAuth::OAuth { + credential, + scope, + server_url, + } => IcebergCatalogAuth::OAuth { + credential, + scope, + server_url, + }, } } } @@ -650,6 +724,7 @@ impl IntoInlineConnection RestIcebergCatalog { auth: self.auth.into_inline_connection(&r), warehouse: self.warehouse, + access_delegation: self.access_delegation, } } } @@ -724,18 +799,27 @@ impl IcebergCatalogConnection { } impl IcebergCatalogConnection { + /// Connects to the catalog. + /// + /// `table` names the table this handle will be used against. It is needed only to keep + /// catalog-vended storage credentials refreshed, which the REST specification scopes to a + /// single table. Passing `None` leaves the connection on whatever credentials the catalog + /// supplies at `loadTable` time, which expire. pub async fn connect( &self, storage_configuration: &StorageConfiguration, in_task: InTask, + table: Option<&TableIdent>, ) -> Result, anyhow::Error> { match self.catalog { IcebergCatalogImpl::S3TablesRest(ref s3tables) => { + // S3 Tables signs every request with SigV4 off a refreshable AWS provider, so it + // has no vended credential to keep alive. self.connect_s3tables(s3tables, storage_configuration, in_task) .await } IcebergCatalogImpl::Rest(ref rest) => { - self.connect_rest(rest, storage_configuration, in_task) + self.connect_rest(rest, storage_configuration, in_task, table) .await } } @@ -871,15 +955,14 @@ impl IcebergCatalogConnection { // N.B. We're using the AWS credentials from the catalog connection for the storage layer // even though the sink comes with its own (unused) AWS credentials for storage. let customized_credential_load = if matches!(aws_auth, AwsAuth::AssumeRole(_)) { - Some(CustomAwsCredentialLoader::new(Arc::new( - AwsSdkCredentialLoader::new(credentials_provider), + Some(CustomAwsCredentialLoader::new(AwsSdkCredentialLoader::new( + credentials_provider, ))) } else { None }; let storage_factory = Arc::new(OpenDalStorageFactory::S3 { - configured_scheme: "s3".to_string(), customized_credential_load, }); @@ -904,6 +987,7 @@ impl IcebergCatalogConnection { rest: &RestIcebergCatalog, storage_configuration: &StorageConfiguration, in_task: InTask, + table: Option<&TableIdent>, ) -> Result, anyhow::Error> { let mut props = BTreeMap::from([( REST_CATALOG_PROP_URI.to_string(), @@ -914,11 +998,19 @@ impl IcebergCatalogConnection { props.insert(REST_CATALOG_PROP_WAREHOUSE.to_string(), warehouse.clone()); } + // One client for catalog requests, OAuth token requests, and credential refreshes, so all + // three share a connection pool. `iceberg-rust` would otherwise default to its own. + let client = reqwest::Client::new(); + // Catalog auth is configured through a combination of `props` and `.with_authenticator(...)`, // which happen at different stages of the [`RestCatalogBuilder`] -> [`RestCatalog`] // construction pipeline. let (storage_factory, custom_authenticator) = match &rest.auth { - IcebergCatalogAuth::OAuth { credential, scope } => { + IcebergCatalogAuth::OAuth { + credential, + scope, + server_url, + } => { let credential = credential .get_string( in_task, @@ -926,21 +1018,110 @@ impl IcebergCatalogConnection { ) .await .map_err(|e| anyhow!("failed to read Iceberg catalog credential: {e}"))?; - props.insert(REST_CATALOG_PROP_CREDENTIAL.to_string(), credential); - if let Some(scope) = scope { - props.insert(REST_CATALOG_PROP_SCOPE.to_string(), scope.clone()); + if let Some(server_url) = server_url { + // The OAuth2 exchange POSTs the catalog credential to this URL, so a URL + // aimed inside our own network turns the connection into a request forger + // against, say, a cloud metadata endpoint. Resolve it and reject private + // addresses, the same check every other host we dial directly gets. + // + // NOTE: the resolved addresses are only checked, not pinned. The catalog + // client offers no hook to dial a pre-resolved address, so a name that + // resolves differently between this check and the request slips through. + // Kafka and Confluent Schema Registry connections do pin theirs. + let url = Url::parse(server_url).with_context(|| { + format!("invalid OAUTH2 SERVER URL for Iceberg catalog: {server_url}") + })?; + let host = url.host_str().ok_or_else(|| { + anyhow!("OAUTH2 SERVER URL for Iceberg catalog has no host: {server_url}") + })?; + resolve_address( + host, + ENFORCE_EXTERNAL_ADDRESSES.get(storage_configuration.config_set()), + ) + .await + .with_context(|| { + format!("OAUTH2 SERVER URL for Iceberg catalog is not resolvable to an external address: {server_url}") + })?; + + props.insert( + REST_CATALOG_PROP_OAUTH2_SERVER_URI.to_string(), + server_url.clone(), + ); } + + // Materialize builds an OAuth2 provider shared across both catalog requests + // and the vended credentials refresh below. + let token_endpoint = match server_url { + Some(server_url) => server_url.clone(), + // Matches `iceberg-rust`'s default when no `oauth2-server-uri` is configured. + None => format!( + "{}/v1/oauth/tokens", + self.uri.as_str().trim_end_matches('/') + ), + }; + let (client_id, client_secret) = match credential.split_once(':') { + Some((client_id, client_secret)) => { + (Some(client_id.to_string()), client_secret.to_string()) + } + None => (None, credential), + }; + let oauth_params = BTreeMap::from([( + OAUTH2_PARAM_SCOPE.to_string(), + // The default `iceberg-rust` applies when the connection names no scope. + scope.clone().unwrap_or_else(|| "catalog".to_string()), + )]); + let token: Arc = Arc::new(OAuth2TokenProvider::new( + client.clone(), + client_id, + client_secret, + token_endpoint, + // The token request needs none of the catalog's headers, and + // `OAuth2TokenProvider` sets the form content type itself. + HeaderMap::new(), + oauth_params.into_iter().collect(), + )); + + // Installing a loader hands it sole responsibility for S3 credentials: OpenDAL + // replaces its whole provider chain, including the static keys parsed out of the + // catalog's vended `storage-credentials` props. So only install one when the + // connection asked for delegation and we know which table to refresh, and let the + // static props serve every other case. + let customized_credential_load = match (&rest.access_delegation, table) { + (Some(IcebergAccessDelegation::VendedCredentials), Some(table)) => { + let endpoint = iceberg_credentials::table_credentials_endpoint( + &self.uri, + &client, + &token, + rest.warehouse.as_deref(), + table, + ) + .await?; + Some(CustomAwsCredentialLoader::new( + iceberg_credentials::VendedCredentialLoader::new( + client.clone(), + endpoint, + Arc::clone(&token), + ), + )) + } + _ => None, + }; + ( OpenDalStorageFactory::S3 { - configured_scheme: "s3".to_string(), // When used with MinIO, Polaris returns a config with: // s3.access-key-id, s3.secret-access-key, s3.endpoint, ... - // `iceberg-rust` forwards these props to `opendal`. + // `iceberg-rust` forwards these props to `opendal`. When the catalog + // vends instead, it returns per-table `storage-credentials` that + // `iceberg-rust` wires into the same FileIO. // N.B. This is not confirmed to work with other catalog & storage implementations. - customized_credential_load: None, + customized_credential_load, }, - None, + // NOTE: We construct our own OAuth authenticator for the Catalog client instead of using the one built in. + // This means we ignore auth overrides from `/v1/config` (e.g. `oauth2-server-uri`). + // This is okay because users can set these configs from Mz SQL. + Some(iceberg_catalog_rest::BearerTokenAuthenticator::new(token)), ) } IcebergCatalogAuth::Gcp(gcp_connection_reference) => { @@ -974,8 +1155,19 @@ impl IcebergCatalogConnection { } }; - let mut catalog = - RestCatalogBuilder::default().with_storage_factory(Arc::new(storage_factory)); + // `iceberg-rust` turns `header.*` props into headers on every REST request, so + // connection asked for delegation. + // than falling back to their configured storage credentials. + if let Some(delegation) = &rest.access_delegation { + props.insert( + REST_CATALOG_PROP_ACCESS_DELEGATION.to_string(), + delegation.as_header_value().to_string(), + ); + } + + let mut catalog = RestCatalogBuilder::default() + .with_storage_factory(Arc::new(storage_factory)) + .with_client(client); if let Some(auth) = custom_authenticator { catalog = catalog.with_authenticator(Arc::new(auth)); } @@ -991,8 +1183,9 @@ impl IcebergCatalogConnection { _id: CatalogItemId, storage_configuration: &StorageConfiguration, ) -> Result<(), ConnectionValidationError> { + // Validation only lists namespaces, so it needs no table-scoped credentials. let catalog = self - .connect(storage_configuration, InTask::No) + .connect(storage_configuration, InTask::No, None) .await .map_err(|e| { ConnectionValidationError::Other(anyhow!("failed to connect to catalog: {e}")) diff --git a/src/storage-types/src/connections/iceberg_credentials.rs b/src/storage-types/src/connections/iceberg_credentials.rs new file mode 100644 index 0000000000000..0c446e8890bcc --- /dev/null +++ b/src/storage-types/src/connections/iceberg_credentials.rs @@ -0,0 +1,503 @@ +// Copyright Materialize, Inc. and contributors. All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0. + +//! Catalog-vended storage credentials for Iceberg sinks. +//! +//! A REST catalog asked for access delegation answers `loadTable` with temporary, table-scoped +//! storage credentials instead of expecting the client to hold its own. Those credentials expire, +//! and OpenDAL has no notion of that: it takes what the `FileIO` was built with and signs every +//! request with it. [`VendedCredentialLoader`] closes that gap by re-fetching from the catalog's +//! `loadCredentials` endpoint, and the rest of this module locates that endpoint. +//! +//! Locating it takes a round trip of its own. The REST specification has servers announce a +//! request prefix from their `config` endpoint (`catalogs/` for Unity Catalog, absent for +//! others), and every resource path carries it. `iceberg-rust` resolves the same value when it +//! builds the catalog but keeps it private, so this module asks the server directly. + +use std::collections::BTreeMap; +use std::sync::Arc; +use std::time::{Duration, Instant, SystemTime}; + +use anyhow::{Context, anyhow}; +use iceberg::TableIdent; +use iceberg::io::{S3_ACCESS_KEY_ID, S3_SECRET_ACCESS_KEY, S3_SESSION_TOKEN}; +use iceberg_catalog_rest::{StorageCredential, TokenProvider}; +use iceberg_storage_opendal::{AwsCredential, ProvideCredential}; +use mz_ore::error::ErrorExt; +use reqsign_core::time::Timestamp; +use reqwest::StatusCode; +use serde::Deserialize; +use tokio::sync::Mutex; +use tracing::{debug, warn}; +use url::Url; + +use crate::connections::IcebergAccessDelegation; + +/// The `X-Iceberg-Access-Delegation` header, spelled for requests Materialize issues itself +/// rather than through the catalog client, which takes it as a `header.*` catalog property. +const ICEBERG_ACCESS_DELEGATION_HEADER: &str = "X-Iceberg-Access-Delegation"; + +/// Property name for the most common way catalogs report when vended S3 credentials expire. +/// Catalogs are not required to send it. +const S3_SESSION_TOKEN_EXPIRES_AT_MS: &str = "s3.session-token-expires-at-ms"; + +/// How far ahead of a reported expiry to re-fetch a vended credential. +const VENDED_CREDENTIAL_REFRESH_BUFFER: Duration = Duration::from_secs(900); + +/// How long to trust a vended credential that reports no expiry. +/// +/// A catalog that omits [`S3_SESSION_TOKEN_EXPIRES_AT_MS`] leaves nothing to schedule against, and +/// a credential held past its real lifetime fails every S3 request until the dataflow restarts. So +/// re-fetch on a short interval instead: each one is a single REST call against a credential the +/// sink is already using. +// TODO SS-449: make this a dyncfg +const VENDED_CREDENTIAL_DEFAULT_TTL: Duration = Duration::from_secs(300); + +#[derive(Debug)] +pub(super) struct VendedCredentialLoader { + client: reqwest::Client, + credential_endpoint: Url, + token: Arc, + cached: Mutex>, +} + +impl VendedCredentialLoader { + pub(super) fn new( + client: reqwest::Client, + credential_endpoint: Url, + token: Arc, + ) -> Self { + Self { + client, + credential_endpoint, + token, + cached: Mutex::new(None), + } + } + + /// Fetches a fresh credential from the catalog, paired with the instant at which it should be + /// re-fetched. + async fn fetch(&self) -> reqsign_core::Result<(AwsCredential, Instant)> { + let token = self.token.token().await.map_err(|e| { + reqsign_core::Error::credential_invalid( + "failed to obtain a catalog token for vended Iceberg storage credentials", + ) + .with_source(e) + })?; + + let response = self + .client + .get(self.credential_endpoint.clone()) + .bearer_auth(token) + .header( + ICEBERG_ACCESS_DELEGATION_HEADER, + IcebergAccessDelegation::VendedCredentials.as_header_value(), + ) + .send() + .await + .map_err(|e| { + reqsign_core::Error::unexpected(format!( + "failed to request vended Iceberg storage credentials from {}", + self.credential_endpoint + )) + .with_source(e) + })?; + + let status = response.status(); + if !status.is_success() { + // A rejected token stays rejected until something re-mints it, and nothing else on + // this path does: the REST client's own 401 handling covers catalog requests, not + // ours. Drop it so the next attempt fetches a new one. + if status == StatusCode::UNAUTHORIZED || status == StatusCode::FORBIDDEN { + if let Err(e) = self.token.invalidate().await { + warn!( + error = %e.display_with_causes(), + "failed to invalidate catalog token after {status} from the \ + Iceberg credentials endpoint" + ); + } + } + // Safe to surface the body: only success responses carry credentials. + let body = response.text().await.unwrap_or_default(); + return Err(reqsign_core::Error::unexpected(format!( + "Iceberg catalog returned {status} for vended storage credentials at {}: {body}", + self.credential_endpoint + ))); + } + + let response: LoadCredentialsResponse = response.json().await.map_err(|e| { + reqsign_core::Error::unexpected( + "failed to parse the Iceberg catalog's vended storage credentials", + ) + .with_source(e) + })?; + + // `provide_credential` is handed no path, so the longest-prefix match the Iceberg spec + // describes is not available to us. This endpoint is scoped to a single table and in + // practice returns one credential; if a catalog returns several, the most specific + // prefix is the closest thing to a safe default. + if response.storage_credentials.len() > 1 { + debug!( + endpoint = %self.credential_endpoint, + count = response.storage_credentials.len(), + "Iceberg catalog vended multiple storage credentials; using the longest prefix" + ); + } + let credential = response + .storage_credentials + .into_iter() + .max_by_key(|credential| credential.prefix.len()) + .ok_or_else(|| { + reqsign_core::Error::credential_invalid(format!( + "Iceberg catalog vended no storage credentials at {}", + self.credential_endpoint + )) + })?; + + let missing = |prop: &str| { + reqsign_core::Error::credential_invalid(format!( + "vended Iceberg storage credential for prefix {} is missing {prop}", + credential.prefix + )) + }; + let access_key_id = credential + .config + .get(S3_ACCESS_KEY_ID) + .ok_or_else(|| missing(S3_ACCESS_KEY_ID))? + .clone(); + let secret_access_key = credential + .config + .get(S3_SECRET_ACCESS_KEY) + .ok_or_else(|| missing(S3_SECRET_ACCESS_KEY))? + .clone(); + + let expires_in = credential + .config + .get(S3_SESSION_TOKEN_EXPIRES_AT_MS) + .map(|raw| { + let millis = raw.parse::().map_err(|e| { + reqsign_core::Error::credential_invalid(format!( + "vended Iceberg storage credential for prefix {} has an unparseable \ + {S3_SESSION_TOKEN_EXPIRES_AT_MS}", + credential.prefix + )) + .with_source(e) + })?; + Timestamp::from_millisecond(millis) + }) + .transpose()?; + + Ok(( + AwsCredential { + access_key_id, + secret_access_key, + session_token: credential.config.get(S3_SESSION_TOKEN).cloned(), + expires_in, + }, + refresh_deadline(expires_in), + )) + } +} + +impl ProvideCredential for VendedCredentialLoader { + type Credential = AwsCredential; + + async fn provide_credential( + &self, + _ctx: &reqsign_core::Context, + ) -> reqsign_core::Result> { + // The lock is deliberately held across the fetch. `create_operator` builds a fresh + // OpenDAL `Operator` for every file operation, so reqsign's own credential cache never + // outlives a single call and this cache is all that stands between the sink and one + // catalog round trip per S3 request. Serializing here means a stale entry costs one + // refetch rather than one per in-flight operation. + let mut cached = self.cached.lock().await; + + if let Some((credential, refresh_at)) = &*cached + && Instant::now() < *refresh_at + { + return Ok(Some(credential.clone())); + } + + let (credential, refresh_at) = self.fetch().await?; + *cached = Some((credential.clone(), refresh_at)); + Ok(Some(credential)) + } +} + +/// Returns when a vended credential expiring at `expires_in` should be re-fetched. +/// +/// Refreshes [`VENDED_CREDENTIAL_REFRESH_BUFFER`] early to absorb clock skew and the latency of +/// the fetch itself. A credential that expires within the buffer, or has expired already, is +/// re-fetched on the next call. +fn refresh_deadline(expires_in: Option) -> Instant { + let now = Instant::now(); + let Some(expires_in) = expires_in else { + return now + VENDED_CREDENTIAL_DEFAULT_TTL; + }; + let remaining = expires_in + .as_system_time() + .duration_since(SystemTime::now()) + .unwrap_or_default(); + now + remaining.saturating_sub(VENDED_CREDENTIAL_REFRESH_BUFFER) +} + +/// The `loadCredentials` response envelope. `iceberg-rust` models the credential entries but not +/// this wrapper, because it only ever reads them out of a `loadTable` response. +#[derive(Debug, Deserialize)] +struct LoadCredentialsResponse { + #[serde(default, rename = "storage-credentials")] + storage_credentials: Vec, +} + +/// The part of the Iceberg REST `config` response Materialize reads for itself. +#[derive(Debug, Default, Deserialize)] +struct CatalogConfigResponse { + #[serde(default)] + defaults: BTreeMap, + #[serde(default)] + overrides: BTreeMap, +} + +impl CatalogConfigResponse { + /// The request prefix the server wants inserted between `/v1` and the resource path, + /// `catalogs/` for Unity Catalog and absent for catalogs that do not use one. + /// + /// Overrides win over defaults, matching how the catalog client merges the two. + fn announced_prefix(&self) -> Option<&str> { + self.overrides + .get("prefix") + .or_else(|| self.defaults.get("prefix")) + .map(String::as_str) + } +} + +/// Builds the catalog's `config` endpoint, which is where a server announces its request prefix. +/// +/// Unlike every other REST path, this one is not prefixed: the prefix is what it returns. +fn catalog_config_url(uri: &Url, warehouse: Option<&str>) -> Result { + let mut url = uri.clone(); + url.path_segments_mut() + .map_err(|_| anyhow!("Iceberg catalog URI cannot be a base: {uri}"))? + // A configured URI is as likely to be written with a trailing slash as without, and + // that empty last segment would otherwise become `//v1` in the path. + .pop_if_empty() + .extend(["v1", "config"]); + if let Some(warehouse) = warehouse { + url.query_pairs_mut().append_pair("warehouse", warehouse); + } + Ok(url) +} + +/// Builds the endpoint that vends storage credentials for `table`. +fn table_credentials_url( + uri: &Url, + prefix: Option<&str>, + table: &TableIdent, +) -> Result { + let mut url = uri.clone(); + { + let mut segments = url + .path_segments_mut() + .map_err(|_| anyhow!("Iceberg catalog URI cannot be a base: {uri}"))?; + segments.pop_if_empty().push("v1"); + // A prefix is a path fragment rather than a single segment, so it is split out instead + // of pushed whole, which would percent-encode its separators. + if let Some(prefix) = prefix { + segments.extend(prefix.split('/').filter(|s| !s.is_empty())); + } + // `to_url_string` joins multi-level namespaces with the unit separator the + // specification mandates; pushing it as one segment percent-encodes it to `%1F`. + segments + .push("namespaces") + .push(&table.namespace().to_url_string()) + .push("tables") + .push(&table.name) + .push("credentials"); + } + Ok(url) +} + +/// Resolves the REST endpoint that vends storage credentials for `table`. +/// +/// Takes a round trip to the catalog's `config` endpoint, because the resource path carries a +/// request prefix that only the server knows. +pub(super) async fn table_credentials_endpoint( + uri: &Url, + client: &reqwest::Client, + token: &Arc, + warehouse: Option<&str>, + table: &TableIdent, +) -> Result { + let config_endpoint = catalog_config_url(uri, warehouse)?; + + let bearer = token + .token() + .await + .map_err(|e| anyhow!("failed to obtain an Iceberg catalog token: {e}"))?; + let response = client + .get(config_endpoint.clone()) + .bearer_auth(bearer) + .send() + .await + .with_context(|| { + format!("failed to request Iceberg catalog config at {config_endpoint}") + })?; + let status = response.status(); + if !status.is_success() { + let body = response.text().await.unwrap_or_default(); + return Err(anyhow!( + "Iceberg catalog returned {status} for its config at {config_endpoint}: {body}" + )); + } + let config: CatalogConfigResponse = response.json().await.with_context(|| { + format!("failed to parse Iceberg catalog config from {config_endpoint}") + })?; + + table_credentials_url(uri, config.announced_prefix(), table) +} + +#[cfg(test)] +mod tests { + use iceberg::NamespaceIdent; + + use super::*; + + fn table(namespace: &[&str], name: &str) -> TableIdent { + TableIdent::new( + NamespaceIdent::from_strs(namespace).expect("valid namespace"), + name.to_string(), + ) + } + + fn config_with(defaults: &[(&str, &str)], overrides: &[(&str, &str)]) -> CatalogConfigResponse { + let to_map = |pairs: &[(&str, &str)]| { + pairs + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect::>() + }; + CatalogConfigResponse { + defaults: to_map(defaults), + overrides: to_map(overrides), + } + } + + #[mz_ore::test] + fn test_announced_prefix() { + // Overrides win, matching how the catalog client merges the two. + let config = config_with( + &[("prefix", "from-defaults")], + &[("prefix", "from-overrides")], + ); + assert_eq!(config.announced_prefix(), Some("from-overrides")); + + // Defaults are consulted only when overrides is silent, and an unrelated override does + // not suppress them. + let config = config_with(&[("prefix", "from-defaults")], &[("warehouse", "wh")]); + assert_eq!(config.announced_prefix(), Some("from-defaults")); + + // A catalog that uses no prefix announces none. + assert_eq!(config_with(&[], &[]).announced_prefix(), None); + } + + #[mz_ore::test] + fn test_catalog_config_url() { + let url = |uri: &str, warehouse: Option<&str>| { + catalog_config_url(&Url::parse(uri).expect("valid URI"), warehouse) + .expect("URI is a base") + .to_string() + }; + + // The config endpoint is never prefixed: the prefix is what it returns. + assert_eq!( + url("https://catalog.example/api", None), + "https://catalog.example/api/v1/config" + ); + + // A trailing slash must not leave an empty segment behind as `//v1`. + assert_eq!( + url("https://catalog.example/api/", None), + "https://catalog.example/api/v1/config" + ); + assert_eq!( + url("https://catalog.example", None), + "https://catalog.example/v1/config" + ); + + // The warehouse rides along as a query parameter, percent-encoded. + assert_eq!( + url("https://catalog.example", Some("my catalog")), + "https://catalog.example/v1/config?warehouse=my+catalog" + ); + + // A URI that cannot be a base has no path segments to extend. + assert!( + catalog_config_url(&Url::parse("mailto:nobody@example.com").unwrap(), None).is_err() + ); + } + + #[mz_ore::test] + fn test_table_credentials_url() { + let url = |uri: &str, prefix: Option<&str>, t: &TableIdent| { + table_credentials_url(&Url::parse(uri).expect("valid URI"), prefix, t) + .expect("URI is a base") + .to_string() + }; + + // A catalog announcing no prefix puts the resource path directly under `/v1`. + assert_eq!( + url( + "https://catalog.example", + None, + &table(&["sales"], "orders") + ), + "https://catalog.example/v1/namespaces/sales/tables/orders/credentials" + ); + + // Unity Catalog's multi-segment prefix keeps its separator rather than being encoded + // as one segment, and the base path of the configured URI is preserved. + assert_eq!( + url( + "https://dbc.cloud.databricks.com/api/2.1/unity-catalog/iceberg-rest", + Some("catalogs/sink-catalog"), + &table(&["sink-namespace"], "mz_append_test") + ), + "https://dbc.cloud.databricks.com/api/2.1/unity-catalog/iceberg-rest/v1/\ + catalogs/sink-catalog/namespaces/sink-namespace/tables/mz_append_test/credentials" + ); + + // A trailing slash on either the URI or the prefix must not produce an empty segment. + assert_eq!( + url( + "https://catalog.example/", + Some("/catalogs/main/"), + &table(&["sales"], "orders") + ), + "https://catalog.example/v1/catalogs/main/namespaces/sales/tables/orders/credentials" + ); + + // Multi-level namespaces are one segment joined by the unit separator the specification + // mandates, which percent-encodes to `%1F`. + assert_eq!( + url( + "https://catalog.example", + None, + &table(&["sales", "eu"], "orders") + ), + "https://catalog.example/v1/namespaces/sales%1Feu/tables/orders/credentials" + ); + + // Names that would otherwise change the path are escaped, not injected into it. + assert_eq!( + url("https://catalog.example", None, &table(&["a/b"], "c?d#e")), + "https://catalog.example/v1/namespaces/a%2Fb/tables/c%3Fd%23e/credentials" + ); + } +} diff --git a/src/storage/src/sink/iceberg.rs b/src/storage/src/sink/iceberg.rs index 315ba1428e822..c416e450ee33d 100644 --- a/src/storage/src/sink/iceberg.rs +++ b/src/storage/src/sink/iceberg.rs @@ -1135,9 +1135,13 @@ fn mint_batch_descriptions<'scope>( return Ok(()); } + let table_ident = TableIdent::new( + NamespaceIdent::new(connection.namespace.clone()), + connection.table.clone(), + ); let catalog = connection .catalog_connection - .connect(&storage_configuration, InTask::Yes) + .connect(&storage_configuration, InTask::Yes, Some(&table_ident)) .await .with_context(|| { format!( @@ -1471,6 +1475,38 @@ struct BoundedDataFileSet { pub data_files: Vec, } +/// Returns the base location for the table's data files, with no trailing separator. +/// +/// `configured_path` is the catalog's `write.data.path`, or `write.folder-storage.path` where +/// only the older property is set. `location` is the table's own location, used when the catalog +/// configures neither. +/// +/// The result never ends in `/`. Callers join it with a `/` and a file name, and the joined URI +/// is what lands in the manifest, so a separator left on the end here produces a manifest entry +/// naming an object that was never written. +fn data_file_location(configured_path: Option<&str>, location: &str) -> String { + // Both properties may legally end in `/`. `DefaultLocationGenerator` stores the value + // verbatim and `generate_location` appends `/` plus the file name, so `s3://b/t/data/` + // yields `s3://b/t/data//f.parquet`. OpenDAL collapses the `//` when it writes the object, + // but the Parquet writer copies the unnormalized URI into the `DataFile`, leaving the + // manifest pointing at a key that does not exist and the table unreadable to anyone else. + // The reference Iceberg location provider strips them for this reason. + if let Some(path) = configured_path { + return path.trim_end_matches('/').to_string(); + } + + // WORKAROUND: S3 Tables catalog incorrectly sets location to the metadata file path + // instead of the warehouse root. Strip off the /metadata/*.metadata.json suffix. No + // clear way to detect this properly right now, so we use heuristics. + let corrected_location = match location.rsplit_once("/metadata/") { + Some((a, b)) if b.ends_with(".metadata.json") => a, + _ => location, + }; + // Trimmed before the join, not after, or a location ending in `/` moves the doubled + // separator into the middle of the URI where a trailing trim cannot reach it. + format!("{}/data", corrected_location.trim_end_matches('/')) +} + /// Construct the envelope-specific closures that [`write_data_files`] needs. /// /// Write rows into Parquet data files bounded by batch descriptions. @@ -1510,9 +1546,11 @@ fn write_data_files<'scope, H: EnvelopeHandler + 'static>( .build_fallible(move |caps| { Box::pin(async move { let [capset]: &mut [_; 1] = caps.try_into().unwrap(); + let namespace_ident = NamespaceIdent::new(connection.namespace.clone()); + let table_ident = TableIdent::new(namespace_ident, connection.table.clone()); let catalog = connection .catalog_connection - .connect(&storage_configuration, InTask::Yes) + .connect(&storage_configuration, InTask::Yes, Some(&table_ident)) .await .with_context(|| { format!( @@ -1523,8 +1561,6 @@ fn write_data_files<'scope, H: EnvelopeHandler + 'static>( ) })?; - let namespace_ident = NamespaceIdent::new(connection.namespace.clone()); - let table_ident = TableIdent::new(namespace_ident, connection.table.clone()); while let Some(_) = table_ready_input.next().await { // Wait for table to be ready } @@ -1549,16 +1585,24 @@ fn write_data_files<'scope, H: EnvelopeHandler + 'static>( .context("Failed to merge Materialize metadata into Iceberg schema")?, ); - // WORKAROUND: S3 Tables catalog incorrectly sets location to the metadata file path - // instead of the warehouse root. Strip off the /metadata/*.metadata.json suffix. - // No clear way to detect this properly right now, so we use heuristics. - let location = table_metadata.location(); - let corrected_location = match location.rsplit_once("/metadata/") { - Some((a, b)) if b.ends_with(".metadata.json") => a, - _ => location, - }; - - let data_location = format!("{}/data", corrected_location); + // A catalog that manages where data files live advertises it through + // `write.data.path`. Honor it: catalogs backing an Iceberg table with + // their own storage layout reject a commit whose data files sit outside + // that path. Unity Catalog, for one, has to register the files in the + // Delta log that actually backs the table, and answers a commit + // referencing files under `/data` with a 500. + // + // `DefaultLocationGenerator::new` reads these same properties, but its + // fallback misses the S3 Tables correction, so choose explicitly. + let properties = table_metadata.properties(); + let configured_path = properties + .get("write.data.path") + .or_else(|| properties.get("write.folder-storage.path")); + let data_location = data_file_location( + configured_path.map(String::as_str), + table_metadata.location(), + ); + debug!(%data_location, "iceberg sink data file location"); let location_generator = DefaultLocationGenerator::with_data_location(data_location); @@ -1910,11 +1954,85 @@ where #[cfg(test)] mod tests { - use super::*; use iceberg::spec::{PrimitiveType, Type}; + use iceberg::writer::file_writer::location_generator::LocationGenerator; use mz_repr::SqlScalarType; use mz_storage_types::sinks::ICEBERG_UINT64_DECIMAL_PRECISION; + use super::*; + + /// The URI a data file is committed under, as the manifest records it. + fn manifest_uri(configured_path: Option<&str>, location: &str) -> String { + let data_location = data_file_location(configured_path, location); + DefaultLocationGenerator::with_data_location(data_location) + .generate_location(None, "part-00000.parquet") + } + + /// Asserts the URI addresses exactly one object, i.e. it survives the path normalization + /// the object store applies before writing. An empty path segment would make the manifest + /// name a key that was never written. + fn assert_addresses_one_object(uri: &str) { + let path = uri + .split_once("://") + .map(|(_scheme, path)| path) + .unwrap_or(uri); + assert!( + !path.contains("//"), + "URI has an empty path segment, so it does not name the object written: {uri}" + ); + } + + #[mz_ore::test] + fn test_data_file_location_trims_configured_path() { + // The property the catalog set is honored as-is when it carries no trailing separator. + assert_eq!( + manifest_uri(Some("s3://bucket/tbl/data"), "s3://bucket/tbl"), + "s3://bucket/tbl/data/part-00000.parquet" + ); + + // A trailing separator is valid in the property, and must not reach the manifest. + for configured in [ + "s3://bucket/tbl/data/", + "s3://bucket/tbl/data//", + "s3://bucket/tbl/data///", + ] { + let uri = manifest_uri(Some(configured), "s3://bucket/tbl"); + assert_addresses_one_object(&uri); + assert_eq!(uri, "s3://bucket/tbl/data/part-00000.parquet"); + } + } + + #[mz_ore::test] + fn test_data_file_location_trims_table_location() { + // With no property set, the data directory hangs off the table location. + assert_eq!( + manifest_uri(None, "s3://bucket/tbl"), + "s3://bucket/tbl/data/part-00000.parquet" + ); + + // A table location ending in `/` would otherwise double the separator mid-URI, where + // trimming the end of the joined string could not fix it. + let uri = manifest_uri(None, "s3://bucket/tbl/"); + assert_addresses_one_object(&uri); + assert_eq!(uri, "s3://bucket/tbl/data/part-00000.parquet"); + } + + #[mz_ore::test] + fn test_data_file_location_corrects_s3_tables_metadata_path() { + // S3 Tables reports the metadata file as the table location; the data directory has to + // hang off the warehouse root instead. + assert_eq!( + data_file_location(None, "s3://bucket/tbl/metadata/00001-abc.metadata.json"), + "s3://bucket/tbl/data" + ); + + // A path that merely contains `/metadata/` is not a metadata file and is left alone. + assert_eq!( + data_file_location(None, "s3://bucket/metadata/tbl"), + "s3://bucket/metadata/tbl/data" + ); + } + #[mz_ore::test] fn test_iceberg_type_overrides() { // UInt16 should override to Int32 @@ -2394,9 +2512,11 @@ fn commit_to_iceberg<'scope>( return Ok(()); } + let namespace_ident = NamespaceIdent::new(connection.namespace.clone()); + let table_ident = TableIdent::new(namespace_ident, connection.table.clone()); let catalog = connection .catalog_connection - .connect(&storage_configuration, InTask::Yes) + .connect(&storage_configuration, InTask::Yes, Some(&table_ident)) .await .with_context(|| { format!( @@ -2407,8 +2527,6 @@ fn commit_to_iceberg<'scope>( let mut write_handle = write_handle.await?; - let namespace_ident = NamespaceIdent::new(connection.namespace.clone()); - let table_ident = TableIdent::new(namespace_ident, connection.table.clone()); while let Some(_) = table_ready_input.next().await { // Wait for table to be ready } diff --git a/test/iceberg/mzcompose.py b/test/iceberg/mzcompose.py index d56b1fc512f23..37f1bb460a22e 100644 --- a/test/iceberg/mzcompose.py +++ b/test/iceberg/mzcompose.py @@ -15,9 +15,7 @@ import urllib.request from materialize.mzcompose.composition import Composition, Service -from materialize.mzcompose.helpers.iceberg import ( - setup_polaris_for_iceberg, -) +from materialize.mzcompose.helpers.iceberg import setup_polaris_for_iceberg from materialize.mzcompose.services.materialized import Materialized from materialize.mzcompose.services.minio import Mc, Minio from materialize.mzcompose.services.mz import Mz @@ -43,7 +41,9 @@ ] -def _setup(c: Composition) -> str: +def _setup( + c: Composition, vended: bool = False, static_credentials: bool = True +) -> str: """Start fresh and return the S3 access key.""" c.down(destroy_volumes=True) c.up( @@ -52,7 +52,9 @@ def _setup(c: Composition) -> str: Service("polaris-bootstrap", idle=True), Service("polaris", idle=True), ) - _, key = setup_polaris_for_iceberg(c) + _, key = setup_polaris_for_iceberg( + c, vended=vended, static_credentials=static_credentials + ) return key @@ -67,6 +69,25 @@ def process(name: str) -> None: c.test_parts(list(c.workflows.keys()), process) +def workflow_vended_credentials(c: Composition) -> None: + """An Iceberg sink must work against a REST catalog that only hands out + temporary, table-scoped credentials. + + The Polaris catalog is created with credential vending enabled and without + the long-lived S3 credentials it would otherwise return to clients, so + Materialize has no static credentials to fall back on. The sink can only + reach MinIO with what Polaris mints for it in response to the + `X-Iceberg-Access-Delegation: vended-credentials` request the Iceberg + catalog connection sends.""" + key = _setup(c, vended=True, static_credentials=False) + + c.run_testdrive_files( + f"--var=s3-access-key={key}", + "--var=aws-endpoint=minio:9000", + "vended-credentials.td", + ) + + def workflow_smoke(c: Composition) -> None: key = _setup(c) @@ -95,6 +116,20 @@ def workflow_gcp_connection_validation(c: Composition) -> None: ) +def workflow_oauth2_server_url(c: Composition) -> None: + """OAUTH2 SERVER URL redirects a REST catalog connection's token exchange + away from the endpoint the Iceberg specification derives from the catalog + URL. Catalogs behind an auth gateway that will not serve an unauthenticated + exchange, such as Databricks Unity Catalog, need it. This exercises + connection planning only and needs no Iceberg backend.""" + c.down(destroy_volumes=True) + c.up("materialized") + + c.run_testdrive_files( + "oauth2-server-url.td", + ) + + def workflow_mode_append(c: Composition) -> None: key = _setup(c) diff --git a/test/iceberg/oauth2-server-url.td b/test/iceberg/oauth2-server-url.td new file mode 100644 index 0000000000000..21411bbe7bf57 --- /dev/null +++ b/test/iceberg/oauth2-server-url.td @@ -0,0 +1,54 @@ +# Copyright Materialize, Inc. and contributors. All rights reserved. +# +# Use of this software is governed by the Business Source License +# included in the LICENSE file at the root of this repository. +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0. + +# OAUTH2 SERVER URL overrides where a REST catalog connection exchanges its +# CREDENTIAL for a bearer token. Without it the endpoint is derived from the +# catalog URL as `/v1/oauth/tokens`, which the Iceberg REST specification +# defines but not every catalog serves. + +# The accepted case below points at a host that does not exist, so skip the +# connect-on-create validation. Everything asserted here is decided at plan time. +$ postgres-execute connection=postgres://mz_system:materialize@${testdrive.materialize-internal-sql-addr} +ALTER SYSTEM SET enable_default_connection_validation = false + +> CREATE SECRET oauth_cred AS 'client-id:client-secret' + +> CREATE CONNECTION oauth_override TO ICEBERG CATALOG ( + CATALOG TYPE = 'REST', + URL = 'https://example.invalid/api/catalog', + CREDENTIAL = SECRET oauth_cred, + OAUTH2 SERVER URL = 'https://example.invalid/oidc/v1/token', + SCOPE = 'all-apis', + WAREHOUSE = 'wh' + ); + +> SELECT name FROM mz_connections WHERE name = 'oauth_override' +oauth_override + +# The option is only meaningful for the OAuth exchange, which s3tablesrest does +# not perform: it authenticates with SigV4 through an AWS connection instead. +! CREATE CONNECTION oauth_on_s3tables TO ICEBERG CATALOG ( + CATALOG TYPE = 's3tablesrest', + URL = 'https://example.invalid/api/catalog', + OAUTH2 SERVER URL = 'https://example.invalid/oidc/v1/token', + WAREHOUSE = 'wh' + ); +contains:s3tablesrest connections do not support OAUTH2 SERVER URL + +# Omitting it stays valid, leaving the specification-derived endpoint in place. +> CREATE CONNECTION oauth_default TO ICEBERG CATALOG ( + CATALOG TYPE = 'REST', + URL = 'https://example.invalid/api/catalog', + CREDENTIAL = SECRET oauth_cred, + WAREHOUSE = 'wh' + ); + +> DROP CONNECTION oauth_default +> DROP CONNECTION oauth_override +> DROP SECRET oauth_cred diff --git a/test/iceberg/vended-credentials.td b/test/iceberg/vended-credentials.td new file mode 100644 index 0000000000000..24a43b6d0ab8e --- /dev/null +++ b/test/iceberg/vended-credentials.td @@ -0,0 +1,71 @@ +# Copyright Materialize, Inc. and contributors. All rights reserved. +# +# Use of this software is governed by the Business Source License +# included in the LICENSE file at the root of this repository. +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0. + +# An Iceberg sink must be able to write to storage using only the temporary, +# table-scoped credentials the REST catalog vends it. +# +# The Polaris catalog backing this test is created without client-visible S3 +# credentials, so nothing in the config it returns lets Materialize reach MinIO +# on its own. The only usable credentials are the ones Polaris mints per table +# in response to the `X-Iceberg-Access-Delegation: vended-credentials` request +# that `ACCESS DELEGATION` below turns on. A sink that commits data here +# therefore exercises the vending path end to end. Were the request dropped, or +# the vended credentials not threaded into the sink's FileIO, the sink would +# fail to write rather than pass quietly. + +# Delegation is opt-in. Only 'vended-credentials' is accepted. +! CREATE CONNECTION bad_delegation TO ICEBERG CATALOG ( + CATALOG TYPE = 'REST', + URL = 'http://polaris:8181/api/catalog', + CREDENTIAL = 'root:root', + WAREHOUSE = 'default_catalog', + ACCESS DELEGATION = 'remote-signing' + ); +contains:invalid iceberg access delegation + +> CREATE CONNECTION vended_polaris TO ICEBERG CATALOG ( + CATALOG TYPE = 'REST', + URL = 'http://polaris:8181/api/catalog', + CREDENTIAL = 'root:root', + WAREHOUSE = 'default_catalog', + SCOPE = 'PRINCIPAL_ROLE:ALL', + ACCESS DELEGATION = 'vended-credentials' + ); + +> CREATE TABLE vended_src (a int, b text); + +> INSERT INTO vended_src VALUES (1, 'one'), (2, 'two'), (3, 'three'); + +> CREATE SINK vended_sink + FROM vended_src + INTO ICEBERG CATALOG CONNECTION vended_polaris ( + NAMESPACE 'default_namespace', + TABLE 'vended_table' + ) + MODE APPEND + WITH (COMMIT INTERVAL '1s'); + +# Iceberg sinks commit asynchronously; wait for at least one commit interval. +$ sleep-is-probably-flaky-i-have-justified-my-need-with-a-comment duration=10s + +# A sink denied storage credentials stalls with an error instead of staying healthy. +> SELECT error IS NULL FROM mz_internal.mz_sink_statuses WHERE name = 'vended_sink' +true + +# Read the table back with our own long-lived credentials, which are independent +# of whatever the catalog vended to Materialize. +$ duckdb-execute name=vended_iceberg +CREATE SECRET s3_secret_vended (TYPE S3, KEY_ID 'tduser', SECRET '${arg.s3-access-key}', ENDPOINT '${arg.aws-endpoint}', URL_STYLE 'path', USE_SSL false, REGION 'minio'); +SET unsafe_enable_version_guessing = true; + +$ duckdb-query name=vended_iceberg +SELECT a, b FROM iceberg_scan('s3://test-bucket/default_namespace/vended_table') ORDER BY a +1 one +2 two +3 three