From 13235b8f253e7511bebe8522588f0ab8204d78f7 Mon Sep 17 00:00:00 2001 From: Gene Bordegaray Date: Wed, 19 Aug 2026 13:51:56 -0400 Subject: [PATCH] Upgrade DataFusion to 55 --- .github/workflows/bindings_python_ci.yml | 2 +- .github/workflows/ci.yml | 8 +- .github/workflows/public-api.yml | 2 +- Cargo.lock | 513 ++++++++++-------- Cargo.toml | 16 +- crates/catalog/rest/src/catalog.rs | 2 +- crates/integrations/datafusion/public-api.txt | 30 +- .../datafusion/src/physical_plan/commit.rs | 108 +++- .../src/physical_plan/metadata_scan.rs | 28 +- .../datafusion/src/physical_plan/scan.rs | 28 +- .../datafusion/src/physical_plan/write.rs | 103 +++- .../integrations/datafusion/src/table/mod.rs | 6 +- .../src/table/table_provider_factory.rs | 38 +- .../tests/integration_datafusion_test.rs | 7 +- 14 files changed, 592 insertions(+), 299 deletions(-) diff --git a/.github/workflows/bindings_python_ci.yml b/.github/workflows/bindings_python_ci.yml index 6980eb909a..aa816de247 100644 --- a/.github/workflows/bindings_python_ci.yml +++ b/.github/workflows/bindings_python_ci.yml @@ -88,7 +88,7 @@ jobs: uses: ./.github/actions/setup-builder - name: Cache Rust artifacts if: runner.os != 'Linux' - uses: swatinem/rust-cache@e18b497796c12c097a38f9edb9d0641fb99eee32 # v2 + uses: swatinem/rust-cache@6323deb102c322ba6fcbdcafc7e3dddab59af2b6 # v2.9.2 with: key: bindings-python save-if: ${{ github.event_name == 'push' && github.ref == 'refs/heads/main' }} diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d3c1713484..e6c827a95e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -122,7 +122,7 @@ jobs: uses: ./.github/actions/setup-builder - name: Cache Rust artifacts - uses: swatinem/rust-cache@e18b497796c12c097a38f9edb9d0641fb99eee32 # v2 + uses: swatinem/rust-cache@6323deb102c322ba6fcbdcafc7e3dddab59af2b6 # v2.9.2 with: save-if: ${{ github.event_name == 'push' && github.ref == 'refs/heads/main' }} @@ -148,7 +148,7 @@ jobs: uses: ./.github/actions/setup-builder - name: Cache Rust artifacts - uses: swatinem/rust-cache@e18b497796c12c097a38f9edb9d0641fb99eee32 # v2 + uses: swatinem/rust-cache@6323deb102c322ba6fcbdcafc7e3dddab59af2b6 # v2.9.2 with: save-if: ${{ github.event_name == 'push' && github.ref == 'refs/heads/main' }} @@ -188,7 +188,7 @@ jobs: uses: ./.github/actions/setup-builder - name: Cache Rust artifacts - uses: swatinem/rust-cache@e18b497796c12c097a38f9edb9d0641fb99eee32 # v2 + uses: swatinem/rust-cache@6323deb102c322ba6fcbdcafc7e3dddab59af2b6 # v2.9.2 with: save-if: ${{ github.event_name == 'push' && github.ref == 'refs/heads/main' }} @@ -222,7 +222,7 @@ jobs: repo-token: ${{ secrets.GITHUB_TOKEN }} - name: Cache Rust artifacts - uses: swatinem/rust-cache@e18b497796c12c097a38f9edb9d0641fb99eee32 # v2 + uses: swatinem/rust-cache@6323deb102c322ba6fcbdcafc7e3dddab59af2b6 # v2.9.2 with: key: ${{ matrix.test-suite.name }} save-if: ${{ github.event_name == 'push' && github.ref == 'refs/heads/main' }} diff --git a/.github/workflows/public-api.yml b/.github/workflows/public-api.yml index f5149d30d4..a8d1030560 100644 --- a/.github/workflows/public-api.yml +++ b/.github/workflows/public-api.yml @@ -44,7 +44,7 @@ jobs: uses: ./.github/actions/setup-builder - name: Cache Rust artifacts - uses: swatinem/rust-cache@e18b497796c12c097a38f9edb9d0641fb99eee32 # v2 + uses: swatinem/rust-cache@6323deb102c322ba6fcbdcafc7e3dddab59af2b6 # v2.9.2 with: save-if: ${{ github.event_name == 'push' && github.ref == 'refs/heads/main' }} diff --git a/Cargo.lock b/Cargo.lock index 2387b5b439..7010c8cbea 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -165,7 +165,7 @@ dependencies = [ "digest 0.10.7", "log", "miniz_oxide", - "num-bigint", + "num-bigint 0.4.8", "quad-rand", "rand 0.9.4", "regex-lite", @@ -227,9 +227,9 @@ checksum = "d3fb67a6e08acf24fdeccbac2cb6ac4305825bd1f117462e0e6f2f193345ad56" [[package]] name = "arrow" -version = "59.1.0" +version = "59.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b952ca5a8046ad741b60f142d6eca4aeebcad615694202bc64c5341f23e32c5b" +checksum = "61d285d16bce7d0be61912f7928342b673067b6b7d7ef6cc179258ba7de1fecf" dependencies = [ "arrow-arith", "arrow-array", @@ -249,9 +249,9 @@ dependencies = [ [[package]] name = "arrow-arith" -version = "59.1.0" +version = "59.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "64a13b8d3008c4e9063c597a08f46446fe3fd5789277127672d6c0bdbb43b1ff" +checksum = "757ef1836251e88222542a7da2623bc1c9cb9e20afefa6db2c41e79991cd91d4" dependencies = [ "arrow-array", "arrow-buffer", @@ -263,9 +263,9 @@ dependencies = [ [[package]] name = "arrow-array" -version = "59.1.0" +version = "59.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9486151b2f0785bafc6fa04fc5c99fcb4495455662e58787ea32eaaed33c4192" +checksum = "bc9a4a4b2b5ecd0e04df03471661cb61f28bed3c7fd50994715129b01b2edb97" dependencies = [ "ahash", "arrow-buffer", @@ -275,6 +275,7 @@ dependencies = [ "chrono-tz", "half", "hashbrown 0.17.1", + "libc", "num-complex", "num-integer", "num-traits", @@ -282,9 +283,9 @@ dependencies = [ [[package]] name = "arrow-avro" -version = "59.1.0" +version = "59.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2e4f9b23a0d7b613acb59fa20bdbe0f80ffdae6411498378340b3915e45f5b84" +checksum = "9fb45cd6bd2b25c0965793b83200eaca82214273a8030fbbc2d783e4c7c65a61" dependencies = [ "arrow-array", "arrow-buffer", @@ -306,21 +307,21 @@ dependencies = [ [[package]] name = "arrow-buffer" -version = "59.1.0" +version = "59.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c4776577a87794bfdf0b4e90e2ea12454fa7738ea2823c4be5b9d1851da7b434" +checksum = "c12b576ef18c1deb80925a248b25ad84f419198d791b8e293fc6aaa60441fe90" dependencies = [ "bytes", "half", - "num-bigint", + "num-bigint 0.5.1", "num-traits", ] [[package]] name = "arrow-cast" -version = "59.1.0" +version = "59.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a9ad451ce4f98710828a455b96991b8f031deb2e67f5fcad6773f017e4a69c3a" +checksum = "68338a9096a5dc9bc11927c58c43a8526d96bf6abd2012ef6c0c9f505991cc79" dependencies = [ "arrow-array", "arrow-buffer", @@ -329,7 +330,7 @@ dependencies = [ "arrow-schema", "arrow-select", "atoi", - "base64", + "base64 0.23.1", "chrono", "comfy-table", "half", @@ -340,9 +341,9 @@ dependencies = [ [[package]] name = "arrow-csv" -version = "59.1.0" +version = "59.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8aa7bf96d6141a7bcca2eed57c7c9767d2a2175281857b8a7b68308992864784" +checksum = "25011b52b346407d497ef0030e12b45e4f2d0cc279efc09c4f3d09106db30e36" dependencies = [ "arrow-array", "arrow-cast", @@ -355,9 +356,9 @@ dependencies = [ [[package]] name = "arrow-data" -version = "59.1.0" +version = "59.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b38fe43e2e8704360f1464e6e8cc4fc381ef02cc4fb0192afa8df1aaa0115c66" +checksum = "723fe4aeed7604e00b9883a465af4ff0a0e6c44c03e41a68c3d1cbc403e0e44d" dependencies = [ "arrow-buffer", "arrow-schema", @@ -368,9 +369,9 @@ dependencies = [ [[package]] name = "arrow-ipc" -version = "59.1.0" +version = "59.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "29dac499fcbc6ba74ee0324057821d381929a48526a3966bd9dffb44aa06d98c" +checksum = "149437b14371f5b9ec60f5ddc751483ae99d7a7072653c0075e5e469156eea7b" dependencies = [ "arrow-array", "arrow-buffer", @@ -378,15 +379,15 @@ dependencies = [ "arrow-schema", "arrow-select", "flatbuffers", - "lz4_flex", + "lz4_flex 0.14.0", "zstd", ] [[package]] name = "arrow-json" -version = "59.1.0" +version = "59.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0fe05e916ddc50f4c7a363cd69c0ef5894fcee063517e9a0b8582f0c56746af6" +checksum = "f18b9123ccfec418a663f821c9a034af339711678c11ffe00d3ec07da5ff9f7e" dependencies = [ "arrow-array", "arrow-buffer", @@ -409,9 +410,9 @@ dependencies = [ [[package]] name = "arrow-ord" -version = "59.1.0" +version = "59.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0e13dbdc2a9c053c10c7baa6e30faee04a180aa7ce88e471835850ce37abd20b" +checksum = "e6c08dff0686cf23ca4f562803f191ccbeb726dbae6309cd4b4aaf65e0f2c979" dependencies = [ "arrow-array", "arrow-buffer", @@ -422,9 +423,9 @@ dependencies = [ [[package]] name = "arrow-pyarrow" -version = "59.1.0" +version = "59.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bf8d967bdece4fa5a0199706730175b3df3448b87e350250a86eb6c22639e445" +checksum = "c196ecc25b3a8dcbc1d842f2619cee653dcfa2fb8b56a291bc0481c3cf5c3821" dependencies = [ "arrow-array", "arrow-data", @@ -434,9 +435,9 @@ dependencies = [ [[package]] name = "arrow-row" -version = "59.1.0" +version = "59.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4d5a1f8c733d15260b305683472ee8ad89c62cbd706703ca873b90d051b41592" +checksum = "bbec439386df71ad570e6758a946111322b9e9dc8db83b5527321f0b4c9119c2" dependencies = [ "arrow-array", "arrow-buffer", @@ -447,9 +448,9 @@ dependencies = [ [[package]] name = "arrow-schema" -version = "59.1.0" +version = "59.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d9e4969dc350d571766247143ab36a5187d095d3d3690970408bc630d47c69e5" +checksum = "e6fed2ca0d1eade57e811cbe73b98ad50cc08a1183e13b2d2aa43a7df593f40e" dependencies = [ "bitflags 2.13.0", "serde_core", @@ -458,9 +459,9 @@ dependencies = [ [[package]] name = "arrow-select" -version = "59.1.0" +version = "59.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "402770dba90865359d98d1ef92ef16e23d75c0cca9c2c880c8a05468b7743bf9" +checksum = "466b19cf75130b891dc1b23a84b343c714c62c64c9c62e365c76aa0ff90a53fb" dependencies = [ "ahash", "arrow-array", @@ -472,9 +473,9 @@ dependencies = [ [[package]] name = "arrow-string" -version = "59.1.0" +version = "59.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a2b0afbb8b9016700938291123df30838b89decc3213dba00852021988b170d3" +checksum = "c838a25bb3691e919e0f617616ac51a4ff8517a952e29ca133cf0c22b2ce65b1" dependencies = [ "arrow-array", "arrow-buffer", @@ -552,7 +553,7 @@ checksum = "3b43422f69d8ff38f95f1b2bb76517c91589a924d1559a0e935d7c8ce0274c11" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -563,7 +564,7 @@ checksum = "9035ad2d096bed7955a320ee7e2230574d28fd3c3a0f186cbea1ff3c7eed5dbb" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -994,7 +995,7 @@ checksum = "221eaa237ddf1ca79b60d1372aad77e47f9c0ea5b3ce5099da8c61d027dc77b3" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -1086,6 +1087,12 @@ version = "0.22.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" +[[package]] +name = "base64" +version = "0.23.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac07cdecf99051d9a5238b80f35af32cdeba5b336e55d957b318b50137e18da5" + [[package]] name = "base64-simd" version = "0.8.0" @@ -1110,7 +1117,7 @@ checksum = "4d6867f1565b3aad85681f1015055b087fcfd840d6aeee6eee7f2da317603695" dependencies = [ "autocfg", "libm", - "num-bigint", + "num-bigint 0.4.8", "num-integer", "num-traits", "serde", @@ -1219,7 +1226,7 @@ dependencies = [ "proc-macro2", "quote", "rustversion", - "syn", + "syn 2.0.118", ] [[package]] @@ -1286,7 +1293,7 @@ checksum = "f9abbd1bc6865053c427f7198e6af43bfdedc55ab791faed4fbd361d789575ff" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -1438,7 +1445,7 @@ dependencies = [ "heck", "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -1832,7 +1839,7 @@ dependencies = [ "proc-macro2", "quote", "strsim", - "syn", + "syn 2.0.118", ] [[package]] @@ -1845,7 +1852,7 @@ dependencies = [ "proc-macro2", "quote", "strsim", - "syn", + "syn 2.0.118", ] [[package]] @@ -1856,7 +1863,7 @@ checksum = "fc34b93ccb385b40dc71c6fceac4b2ad23662c7eeb248cf10d529b7e055b6ead" dependencies = [ "darling_core 0.20.11", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -1867,7 +1874,7 @@ checksum = "ac3984ec7bd6cfa798e62b4a642426a5be0e68f9401cfc2a01e3fa9ea2fcdb8d" dependencies = [ "darling_core 0.23.0", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -1886,8 +1893,8 @@ dependencies = [ [[package]] name = "datafusion" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "arrow", "arrow-schema", @@ -1923,7 +1930,7 @@ dependencies = [ "flate2", "futures", "indexmap 2.14.0", - "itertools 0.14.0", + "itertools 0.15.0", "liblzma", "log", "object_store", @@ -1939,8 +1946,8 @@ dependencies = [ [[package]] name = "datafusion-catalog" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "arrow", "async-trait", @@ -1954,7 +1961,7 @@ dependencies = [ "datafusion-physical-plan", "datafusion-session", "futures", - "itertools 0.14.0", + "itertools 0.15.0", "log", "object_store", "parking_lot", @@ -1963,8 +1970,8 @@ dependencies = [ [[package]] name = "datafusion-catalog-listing" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "arrow", "async-trait", @@ -1978,15 +1985,16 @@ dependencies = [ "datafusion-physical-expr-common", "datafusion-physical-plan", "futures", - "itertools 0.14.0", + "itertools 0.15.0", "log", "object_store", + "percent-encoding", ] [[package]] name = "datafusion-cli" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "arrow", "async-trait", @@ -2012,8 +2020,8 @@ dependencies = [ [[package]] name = "datafusion-common" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "arrow", "arrow-ipc", @@ -2024,9 +2032,10 @@ dependencies = [ "hashbrown 0.17.1", "hex", "indexmap 2.14.0", - "itertools 0.14.0", + "itertools 0.15.0", "libc", "log", + "num-traits", "object_store", "parquet", "recursive", @@ -2038,8 +2047,8 @@ dependencies = [ [[package]] name = "datafusion-common-runtime" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "futures", "log", @@ -2048,8 +2057,8 @@ dependencies = [ [[package]] name = "datafusion-datasource" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "arrow", "async-compression", @@ -2065,11 +2074,12 @@ dependencies = [ "datafusion-physical-expr-adapter", "datafusion-physical-expr-common", "datafusion-physical-plan", + "datafusion-proto-models", "datafusion-session", "flate2", "futures", "glob", - "itertools 0.14.0", + "itertools 0.15.0", "liblzma", "log", "object_store", @@ -2083,8 +2093,8 @@ dependencies = [ [[package]] name = "datafusion-datasource-arrow" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "arrow", "arrow-ipc", @@ -2097,17 +2107,18 @@ dependencies = [ "datafusion-expr", "datafusion-physical-expr-common", "datafusion-physical-plan", + "datafusion-proto-models", "datafusion-session", "futures", - "itertools 0.14.0", + "itertools 0.15.0", "object_store", "tokio", ] [[package]] name = "datafusion-datasource-avro" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "arrow", "arrow-avro", @@ -2124,8 +2135,8 @@ dependencies = [ [[package]] name = "datafusion-datasource-csv" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "arrow", "async-trait", @@ -2137,6 +2148,7 @@ dependencies = [ "datafusion-expr", "datafusion-physical-expr-common", "datafusion-physical-plan", + "datafusion-proto-models", "datafusion-session", "futures", "object_store", @@ -2146,8 +2158,8 @@ dependencies = [ [[package]] name = "datafusion-datasource-json" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "arrow", "async-trait", @@ -2159,6 +2171,7 @@ dependencies = [ "datafusion-expr", "datafusion-physical-expr-common", "datafusion-physical-plan", + "datafusion-proto-models", "datafusion-session", "futures", "object_store", @@ -2168,10 +2181,11 @@ dependencies = [ [[package]] name = "datafusion-datasource-parquet" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "arrow", + "arrow-schema", "async-trait", "bytes", "datafusion-common", @@ -2185,10 +2199,11 @@ dependencies = [ "datafusion-physical-expr-adapter", "datafusion-physical-expr-common", "datafusion-physical-plan", + "datafusion-proto-models", "datafusion-pruning", "datafusion-session", "futures", - "itertools 0.14.0", + "itertools 0.15.0", "log", "object_store", "parking_lot", @@ -2198,17 +2213,18 @@ dependencies = [ [[package]] name = "datafusion-doc" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" [[package]] name = "datafusion-execution" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "arrow", "arrow-buffer", "async-trait", + "bytes", "dashmap", "datafusion-common", "datafusion-expr", @@ -2218,15 +2234,18 @@ dependencies = [ "object_store", "parking_lot", "parquet", + "pin-project-lite", "rand 0.9.4", "tempfile", + "tokio", + "tokio-util", "url", ] [[package]] name = "datafusion-expr" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "arrow", "arrow-schema", @@ -2238,8 +2257,10 @@ dependencies = [ "datafusion-functions-aggregate-common", "datafusion-functions-window-common", "datafusion-physical-expr-common", + "datafusion-proto-common", + "datafusion-proto-models", "indexmap 2.14.0", - "itertools 0.14.0", + "itertools 0.15.0", "recursive", "serde_json", "sqlparser", @@ -2247,19 +2268,19 @@ dependencies = [ [[package]] name = "datafusion-expr-common" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "arrow", "datafusion-common", "indexmap 2.14.0", - "itertools 0.14.0", + "itertools 0.15.0", ] [[package]] name = "datafusion-ffi" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "arrow", "arrow-schema", @@ -2290,12 +2311,12 @@ dependencies = [ [[package]] name = "datafusion-functions" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "arrow", "arrow-buffer", - "base64", + "base64 0.23.1", "blake2", "blake3", "chrono", @@ -2308,7 +2329,7 @@ dependencies = [ "datafusion-macros", "datafusion-physical-expr-common", "hex", - "itertools 0.14.0", + "itertools 0.15.0", "log", "md-5 0.11.0", "memchr", @@ -2321,8 +2342,8 @@ dependencies = [ [[package]] name = "datafusion-functions-aggregate" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "arrow", "datafusion-common", @@ -2333,16 +2354,16 @@ dependencies = [ "datafusion-macros", "datafusion-physical-expr", "datafusion-physical-expr-common", - "foldhash 0.2.0", "half", + "hashbrown 0.17.1", "log", "num-traits", ] [[package]] name = "datafusion-functions-aggregate-common" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "arrow", "datafusion-common", @@ -2352,8 +2373,8 @@ dependencies = [ [[package]] name = "datafusion-functions-nested" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "arrow", "arrow-ord", @@ -2368,7 +2389,7 @@ dependencies = [ "datafusion-macros", "datafusion-physical-expr-common", "hashbrown 0.17.1", - "itertools 0.14.0", + "itertools 0.15.0", "itoa", "log", "memchr", @@ -2376,8 +2397,8 @@ dependencies = [ [[package]] name = "datafusion-functions-table" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "arrow", "async-trait", @@ -2391,8 +2412,8 @@ dependencies = [ [[package]] name = "datafusion-functions-window" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "arrow", "datafusion-common", @@ -2407,8 +2428,8 @@ dependencies = [ [[package]] name = "datafusion-functions-window-common" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "datafusion-common", "datafusion-physical-expr-common", @@ -2416,18 +2437,18 @@ dependencies = [ [[package]] name = "datafusion-macros" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "datafusion-doc", "quote", - "syn", + "syn 3.0.3", ] [[package]] name = "datafusion-optimizer" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "arrow", "chrono", @@ -2436,7 +2457,7 @@ dependencies = [ "datafusion-expr-common", "datafusion-physical-expr", "indexmap 2.14.0", - "itertools 0.14.0", + "itertools 0.15.0", "log", "recursive", "regex", @@ -2445,8 +2466,8 @@ dependencies = [ [[package]] name = "datafusion-physical-expr" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "arrow", "datafusion-common", @@ -2454,10 +2475,11 @@ dependencies = [ "datafusion-expr-common", "datafusion-functions-aggregate-common", "datafusion-physical-expr-common", + "datafusion-proto-models", "half", "hashbrown 0.17.1", "indexmap 2.14.0", - "itertools 0.14.0", + "itertools 0.15.0", "parking_lot", "petgraph", "recursive", @@ -2466,8 +2488,8 @@ dependencies = [ [[package]] name = "datafusion-physical-expr-adapter" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "arrow", "datafusion-common", @@ -2475,29 +2497,30 @@ dependencies = [ "datafusion-functions", "datafusion-physical-expr", "datafusion-physical-expr-common", - "itertools 0.14.0", + "itertools 0.15.0", ] [[package]] name = "datafusion-physical-expr-common" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "arrow", "chrono", "datafusion-common", "datafusion-expr-common", + "datafusion-proto-models", "hashbrown 0.17.1", "indexmap 2.14.0", - "itertools 0.14.0", + "itertools 0.15.0", "parking_lot", "pin-project", ] [[package]] name = "datafusion-physical-optimizer" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "arrow", "datafusion-common", @@ -2508,14 +2531,15 @@ dependencies = [ "datafusion-physical-expr-common", "datafusion-physical-plan", "datafusion-pruning", - "itertools 0.14.0", + "datafusion-session", + "itertools 0.15.0", "recursive", ] [[package]] name = "datafusion-physical-plan" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "arrow", "arrow-data", @@ -2523,6 +2547,7 @@ dependencies = [ "arrow-ord", "arrow-schema", "async-trait", + "bytes", "datafusion-common", "datafusion-common-runtime", "datafusion-execution", @@ -2532,25 +2557,27 @@ dependencies = [ "datafusion-functions-window-common", "datafusion-physical-expr", "datafusion-physical-expr-common", + "datafusion-proto-common", + "datafusion-proto-models", "futures", "half", "hashbrown 0.17.1", "indexmap 2.14.0", - "itertools 0.14.0", + "itertools 0.15.0", "log", "num-traits", "parking_lot", "pin-project-lite", + "serde_json", "tokio", ] [[package]] name = "datafusion-proto" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "arrow", - "chrono", "datafusion-catalog", "datafusion-catalog-listing", "datafusion-common", @@ -2566,24 +2593,35 @@ dependencies = [ "datafusion-physical-expr-common", "datafusion-physical-plan", "datafusion-proto-common", + "datafusion-proto-models", "object_store", "prost", ] [[package]] name = "datafusion-proto-common" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "arrow", "datafusion-common", "prost", ] +[[package]] +name = "datafusion-proto-models" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" +dependencies = [ + "datafusion-common", + "datafusion-proto-common", + "prost", +] + [[package]] name = "datafusion-pruning" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "arrow", "datafusion-common", @@ -2597,9 +2635,10 @@ dependencies = [ [[package]] name = "datafusion-session" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ + "arrow-schema", "async-trait", "datafusion-common", "datafusion-execution", @@ -2610,8 +2649,8 @@ dependencies = [ [[package]] name = "datafusion-spark" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "arrow", "bigdecimal", @@ -2639,8 +2678,8 @@ dependencies = [ [[package]] name = "datafusion-sql" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "arrow", "bigdecimal", @@ -2653,12 +2692,13 @@ dependencies = [ "recursive", "regex", "sqlparser", + "stacker", ] [[package]] name = "datafusion-sqllogictest" -version = "54.0.0" -source = "git+https://github.com/DataDog/datafusion?rev=4a6f1bade38a83cfa6c44e5dc6ae62a838089aad#4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" +version = "55.0.0" +source = "git+https://github.com/DataDog/datafusion?rev=d5552342012888b7d1a3ab88d92e3d292fc0cde0#d5552342012888b7d1a3ab88d92e3d292fc0cde0" dependencies = [ "arrow", "async-trait", @@ -2669,7 +2709,7 @@ dependencies = [ "futures", "half", "indicatif", - "itertools 0.14.0", + "itertools 0.15.0", "log", "object_store", "sqllogictest", @@ -2698,7 +2738,7 @@ dependencies = [ "defmt-parser", "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -2748,7 +2788,7 @@ dependencies = [ "darling 0.20.11", "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -2758,7 +2798,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ab63b0e2bf4d5928aff72e83a7dace85d7bba5fe12dcc3c5a572d78caffd3f3c" dependencies = [ "derive_builder_core", - "syn", + "syn 2.0.118", ] [[package]] @@ -2820,7 +2860,7 @@ checksum = "1ac70aa55017e108007fbaf5aa0f54b021c98f92ff8af59d42eda9da96e3dd4f" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -2886,7 +2926,7 @@ dependencies = [ "enum-ordinalize", "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -2927,7 +2967,7 @@ checksum = "42e528e2d34ba8a67a1a650b86beae8ef69fc5fdb638016f386b973226590432" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -3226,7 +3266,7 @@ checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -3349,7 +3389,7 @@ checksum = "53010ccb100b96a67bc32c0175f0ed1426b31b655d562898e57325f81c023ac0" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -3659,7 +3699,7 @@ version = "0.1.20" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0" dependencies = [ - "base64", + "base64 0.22.1", "bytes", "futures-channel", "futures-util", @@ -3721,7 +3761,7 @@ dependencies = [ "as-any", "async-trait", "backon", - "base64", + "base64 0.22.1", "bimap", "bytes", "chrono", @@ -4209,6 +4249,15 @@ dependencies = [ "either", ] +[[package]] +name = "itertools" +version = "0.15.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b4baf93f58d4425749ca49a51c50ebab072c5df6994d08fed93541c331481dc" +dependencies = [ + "either", +] + [[package]] name = "itoa" version = "1.0.18" @@ -4241,7 +4290,7 @@ checksum = "e165e897f662d428f3cd3828a919dbe067c2d42bb1031eede74ef9d27ecdedd2" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -4286,7 +4335,7 @@ dependencies = [ "quote", "rustc_version", "simd_cesu8", - "syn", + "syn 2.0.118", ] [[package]] @@ -4305,7 +4354,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "38c0b942f458fe50cdac086d2f946512305e5631e720728f2a61aabcd47a6264" dependencies = [ "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -4563,6 +4612,15 @@ dependencies = [ "twox-hash", ] +[[package]] +name = "lz4_flex" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ecbdfe44b1bd960b68170b417450a628c43f7cf56bb3c5317e61cb230ee7f226" +dependencies = [ + "twox-hash", +] + [[package]] name = "marrow" version = "0.2.7" @@ -4713,7 +4771,7 @@ dependencies = [ "cfg-if 1.0.4", "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -4787,7 +4845,7 @@ checksum = "b40e46c845ac234bcba19db7ab252bc2778cbadd516a466d2f12b1580852d136" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -4807,7 +4865,7 @@ checksum = "4568f25ccbd45ab5d5603dc34318c1ec56b117531781260002151b8530a9f931" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -4885,6 +4943,16 @@ dependencies = [ "serde", ] +[[package]] +name = "num-bigint" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93e7820bc0a80a0238e650327316f929ba18d5be054b647490a3a6a339f3e7c0" +dependencies = [ + "num-integer", + "num-traits", +] + [[package]] name = "num-bigint-dig" version = "0.8.6" @@ -4964,7 +5032,7 @@ dependencies = [ "proc-macro-crate", "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -5011,7 +5079,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "622acbc9100d3c10e2ee15804b0caa40e55c933d5aa53814cd520805b7958a49" dependencies = [ "async-trait", - "base64", + "base64 0.22.1", "bytes", "chrono", "form_urlencoded", @@ -5091,7 +5159,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c4f8607c90e2c963a91467f50fb49fbc7fb3d573f88cea219ca59ccd3740b309" dependencies = [ "anyhow", - "base64", + "base64 0.22.1", "bytes", "futures", "http 1.4.2", @@ -5139,7 +5207,7 @@ version = "0.57.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6dea4908d490143a9b0b7f7a790e139ff829b06a023f670455ed3d44f664b361" dependencies = [ - "base64", + "base64 0.22.1", "bytes", "http 1.4.2", "log", @@ -5238,7 +5306,7 @@ version = "0.57.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "313d46c9f5ae70bca26b7c3e3fbb9b639292625f28af73aa016f47e788af9deb" dependencies = [ - "base64", + "base64 0.22.1", "bytes", "crc32c", "http 1.4.2", @@ -5347,9 +5415,9 @@ dependencies = [ [[package]] name = "parquet" -version = "59.1.0" +version = "59.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5302d4da74d6596a1f11f9928767995b53bca657cbeea1e4e8c5074f8a1157dd" +checksum = "7065842956a20c2a536924ce8e4d9955f7422451511b9eb7500d7bfe5077e59c" dependencies = [ "ahash", "arrow-array", @@ -5358,7 +5426,7 @@ dependencies = [ "arrow-ipc", "arrow-schema", "arrow-select", - "base64", + "base64 0.23.1", "brotli", "bytes", "chrono", @@ -5366,12 +5434,11 @@ dependencies = [ "futures", "half", "hashbrown 0.17.1", - "lz4_flex", - "num-bigint", + "lz4_flex 0.14.0", + "num-bigint 0.5.1", "num-integer", "num-traits", "object_store", - "paste", "ring", "seq-macro", "simdutf8", @@ -5403,7 +5470,7 @@ version = "3.0.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d30c53c26bc5b31a98cd02d20f25a7c8567146caf63ed593a9d87b2775291be" dependencies = [ - "base64", + "base64 0.22.1", "serde_core", ] @@ -5491,7 +5558,7 @@ checksum = "c96395f0a926bc13b1c17622aaddda1ecb55d49c8f1bf9777e4d877800a43f8b" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -5650,7 +5717,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "479ca8adacdd7ce8f1fb39ce9ecccbfe93a3f1344b3d0d97f20bc0196208f62b" dependencies = [ "proc-macro2", - "syn", + "syn 2.0.118", ] [[package]] @@ -5691,7 +5758,7 @@ dependencies = [ "itertools 0.14.0", "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -5721,7 +5788,7 @@ checksum = "7347867d0a7e1208d93b46767be83e2b8f978c3dad35f775ac8d8847551d6fe1" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -5779,7 +5846,7 @@ dependencies = [ "proc-macro2", "pyo3-macros-backend", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -5791,7 +5858,7 @@ dependencies = [ "heck", "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -6021,7 +6088,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "76009fbe0614077fc1a2ce255e3a1881a2e3a3527097d5dc6d8212c585e7e38b" dependencies = [ "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -6079,7 +6146,7 @@ checksum = "b7186006dcb21920990093f30e3dea63b7d6e977bf1256be20c3563a5db070da" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -6175,7 +6242,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "62b96928e73ad984de1d99e382749d09e5dab7dd707b767974f7e40aa926b82f" dependencies = [ "anyhow", - "base64", + "base64 0.22.1", "bytes", "form_urlencoded", "http 1.4.2", @@ -6196,7 +6263,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a5fa5cb48808693614d1701fcd3db0b30fa292e0f18e122ae068b6d32eaeed3f" dependencies = [ "anyhow", - "base64", + "base64 0.22.1", "bytes", "form_urlencoded", "futures", @@ -6249,7 +6316,7 @@ version = "0.12.28" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "eddd3ca559203180a307f12d114c268abf583f59b03cb906fd0b3ff8646c1147" dependencies = [ - "base64", + "base64 0.22.1", "bytes", "futures-core", "futures-util", @@ -6291,7 +6358,7 @@ version = "0.13.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "219c5811de6525e5416c7d5d53bb656d3afdbc6c5af816e0802bcfa42dbdc1c3" dependencies = [ - "base64", + "base64 0.22.1", "bytes", "futures-core", "futures-util", @@ -6379,7 +6446,7 @@ checksum = "c0ed1a78a1b19d184b0daa629dd9a024573173ec7d485b287cb369fb3607cc1c" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -6438,7 +6505,7 @@ dependencies = [ "regex", "relative-path", "rustc_version", - "syn", + "syn 2.0.118", "unicode-ident", ] @@ -6762,7 +6829,7 @@ checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -6771,6 +6838,7 @@ version = "1.0.150" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e8014e44b4736ed0538adeecded0fce2a272f22dc9578a7eb6b2d9993c74cfb9" dependencies = [ + "indexmap 2.14.0", "itoa", "memchr", "serde", @@ -6786,7 +6854,7 @@ checksum = "175ee3e80ae9982737ca543e96133087cbd9a485eecc3bc4de9c1a37b47ea59c" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -6816,7 +6884,7 @@ version = "3.21.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "76a5c54c7310e7b8b9577c286d7e399ddd876c3e12b3ed917a8aabc4b96e9e8c" dependencies = [ - "base64", + "base64 0.22.1", "bs58", "chrono", "hex", @@ -6839,7 +6907,7 @@ dependencies = [ "darling 0.23.0", "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -7124,7 +7192,7 @@ checksum = "a6dd45d8fc1c79299bfbb7190e42ccbbdf6a5f52e4a6ad98d92357ea965bd289" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -7146,7 +7214,7 @@ version = "0.8.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ee6798b1838b6a0f69c007c133b8df5866302197e404e8b6ee8ed3e3a5e68dc6" dependencies = [ - "base64", + "base64 0.22.1", "bytes", "crc", "crossbeam-queue", @@ -7186,7 +7254,7 @@ dependencies = [ "quote", "sqlx-core", "sqlx-macros-core", - "syn", + "syn 2.0.118", ] [[package]] @@ -7207,7 +7275,7 @@ dependencies = [ "sha2 0.10.9", "sqlx-core", "sqlx-sqlite", - "syn", + "syn 2.0.118", "tokio", "url", ] @@ -7219,7 +7287,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "aa003f0038df784eb8fecbbac13affe3da23b45194bd57dba231c8f48199c526" dependencies = [ "atoi", - "base64", + "base64 0.22.1", "bitflags 2.13.0", "byteorder", "bytes", @@ -7260,7 +7328,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "db58fcd5a53cf07c184b154801ff91347e4c30d17a3562a635ff028ad5deda46" dependencies = [ "atoi", - "base64", + "base64 0.22.1", "bitflags 2.13.0", "byteorder", "crc", @@ -7345,7 +7413,7 @@ dependencies = [ "proc-macro-crate", "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -7418,7 +7486,7 @@ dependencies = [ "heck", "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -7430,7 +7498,7 @@ dependencies = [ "heck", "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -7466,6 +7534,17 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "syn" +version = "3.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53e9bae58849f64dfa4f5d5ae372c8341f7305f82a3868709269343628b659a3" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + [[package]] name = "sync_wrapper" version = "1.0.2" @@ -7483,7 +7562,7 @@ checksum = "728a70f3dbaf5bab7f0c4b1ac8d7ae5ea60a4b5549c8a5914361c99147a709d2" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -7578,7 +7657,7 @@ checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -7589,7 +7668,7 @@ checksum = "ebc4ee7f67670e9b64d05fa4253e753e016c6c95ff35b89b7941d6b856dec1d5" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -7690,7 +7769,7 @@ checksum = "385a6cb71ab9ab790c5fe8d67f1645e6c450a7ce006a33de03daa956cf70a496" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -7888,7 +7967,7 @@ checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -7975,7 +8054,7 @@ checksum = "3c36781cc0e46a83726d9879608e4cf6c2505237e263a8eb8c24502989cfdb28" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -8011,7 +8090,7 @@ checksum = "cf808357c6ed7e13ba0f3277ec8d8f21b2d501274895104263985330c726c1c5" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -8309,7 +8388,7 @@ dependencies = [ "bumpalo", "proc-macro2", "quote", - "syn", + "syn 2.0.118", "wasm-bindgen-shared", ] @@ -8502,7 +8581,7 @@ checksum = "053e2e040ab57b9dc951b72c264860db7eb3b0200ba345b4e4c3b14f67855ddf" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -8513,7 +8592,7 @@ checksum = "3f316c4a2570ba26bbec722032c4099d8c8bc095efccdc15688708623367e358" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -8766,7 +8845,7 @@ checksum = "3e1e496dcbe6a09017acdfaf48e1a646735e7ff5b2a49e2c7e081cca77a59bc8" dependencies = [ "anyhow", "async-trait", - "base64", + "base64 0.22.1", "bytes", "clap", "crc32fast", @@ -8803,7 +8882,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cb838aa8eb67d730af301584cf003caad407487606058292a6750711b603fbee" dependencies = [ "async-trait", - "base64", + "base64 0.22.1", "blake3", "bytemuck", "bytes", @@ -8816,7 +8895,7 @@ dependencies = [ "heapify", "itertools 0.14.0", "lazy_static", - "lz4_flex", + "lz4_flex 0.13.1", "more-asserts", "rand 0.10.2", "regex", @@ -8936,7 +9015,7 @@ checksum = "de844c262c8848816172cef550288e7dc6c7b7814b4ee56b3e1553f275f1858e" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", "synstructure", ] @@ -8957,7 +9036,7 @@ checksum = "4714fd92cf900833d49538023a9b3915155210801d1c1169eba513b2addefd71" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -8977,7 +9056,7 @@ checksum = "11532158c46691caf0f2593ea8358fed6bbf68a0315e80aae9bd41fbade684a1" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", "synstructure", ] @@ -9017,7 +9096,7 @@ checksum = "625dc425cab0dca6dc3c3319506e6593dcb08a9f387ea3b284dbd52a92c40555" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 05f7f0cda1..3f7ae9de74 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -74,10 +74,10 @@ cfg-if = "1" chrono = "0.4.41" clap = { version = "4.5.48", features = ["derive", "cargo"] } dashmap = "6" -datafusion = "54.0.0" -datafusion-cli = "54.0.0" -datafusion-ffi = "54.0.0" -datafusion-sqllogictest = "54.0.0" +datafusion = "55.0.0" +datafusion-cli = "55.0.0" +datafusion-ffi = "55.0.0" +datafusion-sqllogictest = "55.0.0" derive_builder = "0.20" dirs = "6" enum-ordinalize = "4.3.0" @@ -161,7 +161,7 @@ zeroize = "1.7" zstd = "0.13.3" [patch.crates-io] -datafusion = { git = "https://github.com/DataDog/datafusion", rev = "4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" } -datafusion-cli = { git = "https://github.com/DataDog/datafusion", rev = "4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" } -datafusion-ffi = { git = "https://github.com/DataDog/datafusion", rev = "4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" } -datafusion-sqllogictest = { git = "https://github.com/DataDog/datafusion", rev = "4a6f1bade38a83cfa6c44e5dc6ae62a838089aad" } +datafusion = { git = "https://github.com/DataDog/datafusion", rev = "d5552342012888b7d1a3ab88d92e3d292fc0cde0" } +datafusion-cli = { git = "https://github.com/DataDog/datafusion", rev = "d5552342012888b7d1a3ab88d92e3d292fc0cde0" } +datafusion-ffi = { git = "https://github.com/DataDog/datafusion", rev = "d5552342012888b7d1a3ab88d92e3d292fc0cde0" } +datafusion-sqllogictest = { git = "https://github.com/DataDog/datafusion", rev = "d5552342012888b7d1a3ab88d92e3d292fc0cde0" } diff --git a/crates/catalog/rest/src/catalog.rs b/crates/catalog/rest/src/catalog.rs index b26a087236..af108b10cf 100644 --- a/crates/catalog/rest/src/catalog.rs +++ b/crates/catalog/rest/src/catalog.rs @@ -567,7 +567,7 @@ impl RestCatalog { } /// All requests and expected responses are derived from the REST catalog API spec: -/// https://github.com/apache/iceberg/blob/main/open-api/rest-catalog-open-api.yaml +/// #[async_trait] impl Catalog for RestCatalog { async fn list_namespaces( diff --git a/crates/integrations/datafusion/public-api.txt b/crates/integrations/datafusion/public-api.txt index 3d95574d8c..9793fc9c4f 100644 --- a/crates/integrations/datafusion/public-api.txt +++ b/crates/integrations/datafusion/public-api.txt @@ -7,7 +7,7 @@ impl core::clone::Clone for iceberg_datafusion::metadata_table::IcebergMetadataT pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::clone(&self) -> iceberg_datafusion::metadata_table::IcebergMetadataTableProvider impl core::fmt::Debug for iceberg_datafusion::metadata_table::IcebergMetadataTableProvider pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result -impl datafusion_catalog::table::TableProvider for iceberg_datafusion::metadata_table::IcebergMetadataTableProvider +impl datafusion_session::table::TableProvider for iceberg_datafusion::metadata_table::IcebergMetadataTableProvider pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::scan<'life0, 'life1, 'life2, 'life3, 'async_trait>(&'life0 self, _state: &'life1 dyn datafusion_session::session::Session, _projection: core::option::Option<&'life2 alloc::vec::Vec>, _filters: &'life3 [datafusion_expr::expr::Expr], _limit: core::option::Option) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::schema(&self) -> arrow_schema::schema::SchemaRef pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::table_type(&self) -> datafusion_expr::table_source::TableType @@ -38,10 +38,12 @@ pub fn iceberg_datafusion::physical_plan::IcebergTableScan::fmt(&self, f: &mut c impl datafusion_physical_plan::display::DisplayAs for iceberg_datafusion::physical_plan::IcebergTableScan pub fn iceberg_datafusion::physical_plan::IcebergTableScan::fmt_as(&self, _t: datafusion_physical_plan::display::DisplayFormatType, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result impl datafusion_physical_plan::execution_plan::ExecutionPlan for iceberg_datafusion::physical_plan::IcebergTableScan +pub fn iceberg_datafusion::physical_plan::IcebergTableScan::apply_expressions(&self, _f: &mut dyn core::ops::function::FnMut(&alloc::sync::Arc) -> datafusion_common::error::Result) -> datafusion_common::error::Result pub fn iceberg_datafusion::physical_plan::IcebergTableScan::children(&self) -> alloc::vec::Vec<&alloc::sync::Arc<(dyn datafusion_physical_plan::execution_plan::ExecutionPlan + 'static)>> pub fn iceberg_datafusion::physical_plan::IcebergTableScan::execute(&self, partition: usize, _context: alloc::sync::Arc) -> datafusion_common::error::Result pub fn iceberg_datafusion::physical_plan::IcebergTableScan::name(&self) -> &str pub fn iceberg_datafusion::physical_plan::IcebergTableScan::properties(&self) -> &alloc::sync::Arc +pub fn iceberg_datafusion::physical_plan::IcebergTableScan::replace_children(self: alloc::sync::Arc, children: alloc::vec::Vec>, _options: datafusion_physical_plan::execution_plan::ReplaceChildrenOptions) -> datafusion_common::error::Result> pub fn iceberg_datafusion::physical_plan::IcebergTableScan::with_new_children(self: alloc::sync::Arc, children: alloc::vec::Vec>) -> datafusion_common::error::Result> pub fn iceberg_datafusion::physical_plan::convert_filters_to_predicate(filters: &[datafusion_expr::expr::Expr]) -> core::option::Option pub fn iceberg_datafusion::physical_plan::project_with_partition(input: alloc::sync::Arc, table: &iceberg::table::Table) -> datafusion_common::error::Result> @@ -54,7 +56,7 @@ impl core::clone::Clone for iceberg_datafusion::metadata_table::IcebergMetadataT pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::clone(&self) -> iceberg_datafusion::metadata_table::IcebergMetadataTableProvider impl core::fmt::Debug for iceberg_datafusion::metadata_table::IcebergMetadataTableProvider pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result -impl datafusion_catalog::table::TableProvider for iceberg_datafusion::metadata_table::IcebergMetadataTableProvider +impl datafusion_session::table::TableProvider for iceberg_datafusion::metadata_table::IcebergMetadataTableProvider pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::scan<'life0, 'life1, 'life2, 'life3, 'async_trait>(&'life0 self, _state: &'life1 dyn datafusion_session::session::Session, _projection: core::option::Option<&'life2 alloc::vec::Vec>, _filters: &'life3 [datafusion_expr::expr::Expr], _limit: core::option::Option) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::schema(&self) -> arrow_schema::schema::SchemaRef pub fn iceberg_datafusion::metadata_table::IcebergMetadataTableProvider::table_type(&self) -> datafusion_expr::table_source::TableType @@ -67,8 +69,8 @@ impl core::default::Default for iceberg_datafusion::table_provider_factory::Iceb pub fn iceberg_datafusion::table_provider_factory::IcebergTableProviderFactory::default() -> iceberg_datafusion::table_provider_factory::IcebergTableProviderFactory impl core::fmt::Debug for iceberg_datafusion::table_provider_factory::IcebergTableProviderFactory pub fn iceberg_datafusion::table_provider_factory::IcebergTableProviderFactory::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result -impl datafusion_catalog::table::TableProviderFactory for iceberg_datafusion::table_provider_factory::IcebergTableProviderFactory -pub fn iceberg_datafusion::table_provider_factory::IcebergTableProviderFactory::create<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _state: &'life1 dyn datafusion_session::session::Session, cmd: &'life2 datafusion_expr::logical_plan::ddl::CreateExternalTable) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait +impl datafusion_session::table::TableProviderFactory for iceberg_datafusion::table_provider_factory::IcebergTableProviderFactory +pub fn iceberg_datafusion::table_provider_factory::IcebergTableProviderFactory::create<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _state: &'life1 dyn datafusion_session::session::Session, cmd: &'life2 datafusion_expr::logical_plan::ddl::CreateExternalTable) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait pub struct iceberg_datafusion::table::IcebergStaticTableProvider impl iceberg_datafusion::IcebergStaticTableProvider pub async fn iceberg_datafusion::IcebergStaticTableProvider::try_new_from_table(table: iceberg::table::Table) -> iceberg::error::Result @@ -77,7 +79,7 @@ impl core::clone::Clone for iceberg_datafusion::IcebergStaticTableProvider pub fn iceberg_datafusion::IcebergStaticTableProvider::clone(&self) -> iceberg_datafusion::IcebergStaticTableProvider impl core::fmt::Debug for iceberg_datafusion::IcebergStaticTableProvider pub fn iceberg_datafusion::IcebergStaticTableProvider::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result -impl datafusion_catalog::table::TableProvider for iceberg_datafusion::IcebergStaticTableProvider +impl datafusion_session::table::TableProvider for iceberg_datafusion::IcebergStaticTableProvider pub fn iceberg_datafusion::IcebergStaticTableProvider::insert_into<'life0, 'life1, 'async_trait>(&'life0 self, _state: &'life1 dyn datafusion_session::session::Session, _input: alloc::sync::Arc, _insert_op: datafusion_expr::logical_plan::dml::InsertOp) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait pub fn iceberg_datafusion::IcebergStaticTableProvider::scan<'life0, 'life1, 'life2, 'life3, 'async_trait>(&'life0 self, state: &'life1 dyn datafusion_session::session::Session, projection: core::option::Option<&'life2 alloc::vec::Vec>, filters: &'life3 [datafusion_expr::expr::Expr], limit: core::option::Option) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait pub fn iceberg_datafusion::IcebergStaticTableProvider::schema(&self) -> arrow_schema::schema::SchemaRef @@ -91,7 +93,7 @@ impl core::clone::Clone for iceberg_datafusion::IcebergTableProvider pub fn iceberg_datafusion::IcebergTableProvider::clone(&self) -> iceberg_datafusion::IcebergTableProvider impl core::fmt::Debug for iceberg_datafusion::IcebergTableProvider pub fn iceberg_datafusion::IcebergTableProvider::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result -impl datafusion_catalog::table::TableProvider for iceberg_datafusion::IcebergTableProvider +impl datafusion_session::table::TableProvider for iceberg_datafusion::IcebergTableProvider pub fn iceberg_datafusion::IcebergTableProvider::insert_into<'life0, 'life1, 'async_trait>(&'life0 self, state: &'life1 dyn datafusion_session::session::Session, input: alloc::sync::Arc, _insert_op: datafusion_expr::logical_plan::dml::InsertOp) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait pub fn iceberg_datafusion::IcebergTableProvider::scan<'life0, 'life1, 'life2, 'life3, 'async_trait>(&'life0 self, state: &'life1 dyn datafusion_session::session::Session, projection: core::option::Option<&'life2 alloc::vec::Vec>, filters: &'life3 [datafusion_expr::expr::Expr], limit: core::option::Option) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait pub fn iceberg_datafusion::IcebergTableProvider::schema(&self) -> arrow_schema::schema::SchemaRef @@ -106,16 +108,16 @@ impl core::default::Default for iceberg_datafusion::table_provider_factory::Iceb pub fn iceberg_datafusion::table_provider_factory::IcebergTableProviderFactory::default() -> iceberg_datafusion::table_provider_factory::IcebergTableProviderFactory impl core::fmt::Debug for iceberg_datafusion::table_provider_factory::IcebergTableProviderFactory pub fn iceberg_datafusion::table_provider_factory::IcebergTableProviderFactory::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result -impl datafusion_catalog::table::TableProviderFactory for iceberg_datafusion::table_provider_factory::IcebergTableProviderFactory -pub fn iceberg_datafusion::table_provider_factory::IcebergTableProviderFactory::create<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _state: &'life1 dyn datafusion_session::session::Session, cmd: &'life2 datafusion_expr::logical_plan::ddl::CreateExternalTable) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait +impl datafusion_session::table::TableProviderFactory for iceberg_datafusion::table_provider_factory::IcebergTableProviderFactory +pub fn iceberg_datafusion::table_provider_factory::IcebergTableProviderFactory::create<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _state: &'life1 dyn datafusion_session::session::Session, cmd: &'life2 datafusion_expr::logical_plan::ddl::CreateExternalTable) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait pub struct iceberg_datafusion::IcebergCatalogProvider impl iceberg_datafusion::IcebergCatalogProvider pub async fn iceberg_datafusion::IcebergCatalogProvider::try_new(client: alloc::sync::Arc) -> iceberg::error::Result pub async fn iceberg_datafusion::IcebergCatalogProvider::try_new_with_runtime(client: alloc::sync::Arc, runtime: iceberg::runtime::Runtime) -> iceberg::error::Result impl core::fmt::Debug for iceberg_datafusion::IcebergCatalogProvider pub fn iceberg_datafusion::IcebergCatalogProvider::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result -impl datafusion_catalog::catalog::CatalogProvider for iceberg_datafusion::IcebergCatalogProvider -pub fn iceberg_datafusion::IcebergCatalogProvider::schema(&self, name: &str) -> core::option::Option> +impl datafusion_session::catalog::CatalogProvider for iceberg_datafusion::IcebergCatalogProvider +pub fn iceberg_datafusion::IcebergCatalogProvider::schema(&self, name: &str) -> core::option::Option> pub fn iceberg_datafusion::IcebergCatalogProvider::schema_names(&self) -> alloc::vec::Vec #[non_exhaustive] pub struct iceberg_datafusion::IcebergDataFusionConfig pub iceberg_datafusion::IcebergDataFusionConfig::enable_eager_scan_planning: bool @@ -144,7 +146,7 @@ impl core::clone::Clone for iceberg_datafusion::IcebergStaticTableProvider pub fn iceberg_datafusion::IcebergStaticTableProvider::clone(&self) -> iceberg_datafusion::IcebergStaticTableProvider impl core::fmt::Debug for iceberg_datafusion::IcebergStaticTableProvider pub fn iceberg_datafusion::IcebergStaticTableProvider::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result -impl datafusion_catalog::table::TableProvider for iceberg_datafusion::IcebergStaticTableProvider +impl datafusion_session::table::TableProvider for iceberg_datafusion::IcebergStaticTableProvider pub fn iceberg_datafusion::IcebergStaticTableProvider::insert_into<'life0, 'life1, 'async_trait>(&'life0 self, _state: &'life1 dyn datafusion_session::session::Session, _input: alloc::sync::Arc, _insert_op: datafusion_expr::logical_plan::dml::InsertOp) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait pub fn iceberg_datafusion::IcebergStaticTableProvider::scan<'life0, 'life1, 'life2, 'life3, 'async_trait>(&'life0 self, state: &'life1 dyn datafusion_session::session::Session, projection: core::option::Option<&'life2 alloc::vec::Vec>, filters: &'life3 [datafusion_expr::expr::Expr], limit: core::option::Option) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait pub fn iceberg_datafusion::IcebergStaticTableProvider::schema(&self) -> arrow_schema::schema::SchemaRef @@ -158,7 +160,7 @@ impl core::clone::Clone for iceberg_datafusion::IcebergTableProvider pub fn iceberg_datafusion::IcebergTableProvider::clone(&self) -> iceberg_datafusion::IcebergTableProvider impl core::fmt::Debug for iceberg_datafusion::IcebergTableProvider pub fn iceberg_datafusion::IcebergTableProvider::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result -impl datafusion_catalog::table::TableProvider for iceberg_datafusion::IcebergTableProvider +impl datafusion_session::table::TableProvider for iceberg_datafusion::IcebergTableProvider pub fn iceberg_datafusion::IcebergTableProvider::insert_into<'life0, 'life1, 'async_trait>(&'life0 self, state: &'life1 dyn datafusion_session::session::Session, input: alloc::sync::Arc, _insert_op: datafusion_expr::logical_plan::dml::InsertOp) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait pub fn iceberg_datafusion::IcebergTableProvider::scan<'life0, 'life1, 'life2, 'life3, 'async_trait>(&'life0 self, state: &'life1 dyn datafusion_session::session::Session, projection: core::option::Option<&'life2 alloc::vec::Vec>, filters: &'life3 [datafusion_expr::expr::Expr], limit: core::option::Option) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait pub fn iceberg_datafusion::IcebergTableProvider::schema(&self) -> arrow_schema::schema::SchemaRef @@ -172,7 +174,7 @@ impl core::default::Default for iceberg_datafusion::table_provider_factory::Iceb pub fn iceberg_datafusion::table_provider_factory::IcebergTableProviderFactory::default() -> iceberg_datafusion::table_provider_factory::IcebergTableProviderFactory impl core::fmt::Debug for iceberg_datafusion::table_provider_factory::IcebergTableProviderFactory pub fn iceberg_datafusion::table_provider_factory::IcebergTableProviderFactory::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result -impl datafusion_catalog::table::TableProviderFactory for iceberg_datafusion::table_provider_factory::IcebergTableProviderFactory -pub fn iceberg_datafusion::table_provider_factory::IcebergTableProviderFactory::create<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _state: &'life1 dyn datafusion_session::session::Session, cmd: &'life2 datafusion_expr::logical_plan::ddl::CreateExternalTable) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait +impl datafusion_session::table::TableProviderFactory for iceberg_datafusion::table_provider_factory::IcebergTableProviderFactory +pub fn iceberg_datafusion::table_provider_factory::IcebergTableProviderFactory::create<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _state: &'life1 dyn datafusion_session::session::Session, cmd: &'life2 datafusion_expr::logical_plan::ddl::CreateExternalTable) -> core::pin::Pin>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait pub fn iceberg_datafusion::from_datafusion_error(error: datafusion_common::error::DataFusionError) -> iceberg::error::Error pub fn iceberg_datafusion::to_datafusion_error(error: iceberg::error::Error) -> datafusion_common::error::DataFusionError diff --git a/crates/integrations/datafusion/src/physical_plan/commit.rs b/crates/integrations/datafusion/src/physical_plan/commit.rs index 1dee391354..044ecf04b6 100644 --- a/crates/integrations/datafusion/src/physical_plan/commit.rs +++ b/crates/integrations/datafusion/src/physical_plan/commit.rs @@ -22,12 +22,16 @@ use datafusion::arrow::array::{Array, ArrayRef, RecordBatch, StringArray, UInt64 use datafusion::arrow::datatypes::{ DataType, Field, Schema as ArrowSchema, SchemaRef as ArrowSchemaRef, }; +use datafusion::common::tree_node::TreeNodeRecursion; use datafusion::common::{DataFusionError, Result as DFResult}; use datafusion::execution::{SendableRecordBatchStream, TaskContext}; -use datafusion::physical_expr::{EquivalenceProperties, Partitioning}; +use datafusion::physical_expr::{EquivalenceProperties, Partitioning, PhysicalExpr}; use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType}; use datafusion::physical_plan::stream::RecordBatchStreamAdapter; -use datafusion::physical_plan::{DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties}; +use datafusion::physical_plan::{ + ChildrenPropertiesMode, DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties, + ReplaceChildrenOptions, +}; use futures::StreamExt; use iceberg::Catalog; use iceberg::spec::{DataFile, deserialize_data_file_from_json}; @@ -128,6 +132,13 @@ impl ExecutionPlan for IcebergCommitExec { "IcebergCommitExec" } + fn apply_expressions( + &self, + _f: &mut dyn FnMut(&Arc) -> DFResult, + ) -> DFResult { + Ok(TreeNodeRecursion::Continue) + } + fn properties(&self) -> &Arc { &self.plan_properties } @@ -144,9 +155,10 @@ impl ExecutionPlan for IcebergCommitExec { vec![false] } - fn with_new_children( + fn replace_children( self: Arc, - children: Vec>, + mut children: Vec>, + options: ReplaceChildrenOptions, ) -> DFResult> { if children.len() != 1 { return Err(DataFusionError::Internal(format!( @@ -155,12 +167,33 @@ impl ExecutionPlan for IcebergCommitExec { ))); } - Ok(Arc::new(IcebergCommitExec::new( - self.table.clone(), - self.catalog.clone(), - children[0].clone(), - self.schema.clone(), - ))) + let input = children.swap_remove(0); + Ok(match options.children_properties { + ChildrenPropertiesMode::Keep => Arc::new(Self { + table: self.table.clone(), + catalog: Arc::clone(&self.catalog), + input, + schema: Arc::clone(&self.schema), + count_schema: Arc::clone(&self.count_schema), + plan_properties: Arc::clone(&self.plan_properties), + }), + ChildrenPropertiesMode::Recompute => Arc::new(Self::new( + self.table.clone(), + Arc::clone(&self.catalog), + input, + Arc::clone(&self.schema), + )), + }) + } + + fn with_new_children( + self: Arc, + children: Vec>, + ) -> DFResult> { + self.replace_children( + children, + ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute), + ) } fn execute( @@ -276,7 +309,10 @@ mod tests { use datafusion::physical_plan::common::collect; use datafusion::physical_plan::execution_plan::Boundedness; use datafusion::physical_plan::stream::RecordBatchStreamAdapter; - use datafusion::physical_plan::{DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties}; + use datafusion::physical_plan::{ + ChildrenPropertiesMode, DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties, + ReplaceChildrenOptions, + }; use datafusion::prelude::*; use futures::StreamExt; use iceberg::memory::{MEMORY_CATALOG_WAREHOUSE, MemoryCatalogBuilder}; @@ -326,6 +362,13 @@ mod tests { "MockWriteExec" } + fn apply_expressions( + &self, + _f: &mut dyn FnMut(&Arc) -> DFResult, + ) -> DFResult { + Ok(TreeNodeRecursion::Continue) + } + fn schema(&self) -> Arc { self.schema.clone() } @@ -338,13 +381,24 @@ mod tests { vec![] } - fn with_new_children( + fn replace_children( self: Arc, _children: Vec>, + _options: ReplaceChildrenOptions, ) -> datafusion::common::Result> { Ok(self) } + fn with_new_children( + self: Arc, + children: Vec>, + ) -> datafusion::common::Result> { + self.replace_children( + children, + ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute), + ) + } + fn execute( &self, _partition: usize, @@ -451,7 +505,8 @@ mod tests { )?; // Create a mock execution plan that returns the serialized data files - let input_exec = Arc::new(MockWriteExec::new(vec![data_file1_json, data_file2_json])); + let input_exec = Arc::new(MockWriteExec::new(vec![data_file1_json, data_file2_json])) + as Arc; // Create the IcebergCommitExec let arrow_schema = Arc::new(ArrowSchema::new(vec![Field::new( @@ -460,12 +515,35 @@ mod tests { false, )])); - let commit_exec = - IcebergCommitExec::new(table.clone(), catalog.clone(), input_exec, arrow_schema); + let commit_exec = Arc::new(IcebergCommitExec::new( + table.clone(), + catalog.clone(), + input_exec, + arrow_schema, + )); // Verify Execution Plan schema matches the count schema assert_eq!(commit_exec.schema(), IcebergCommitExec::make_count_schema()); + let replacement_input = Arc::new(MockWriteExec::new(vec![])) as Arc; + let original_properties = Arc::clone(commit_exec.properties()); + let kept_properties = Arc::clone(&commit_exec).replace_children( + vec![Arc::clone(&replacement_input)], + ReplaceChildrenOptions::new(ChildrenPropertiesMode::Keep), + )?; + assert!(Arc::ptr_eq( + &original_properties, + kept_properties.properties() + )); + let recomputed_properties = Arc::clone(&commit_exec).replace_children( + vec![replacement_input], + ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute), + )?; + assert!(!Arc::ptr_eq( + &original_properties, + recomputed_properties.properties() + )); + // Execute the commit exec let task_ctx = Arc::new(TaskContext::default()); let stream = commit_exec.execute(0, task_ctx)?; diff --git a/crates/integrations/datafusion/src/physical_plan/metadata_scan.rs b/crates/integrations/datafusion/src/physical_plan/metadata_scan.rs index 340af9ba7e..e61e49a0d0 100644 --- a/crates/integrations/datafusion/src/physical_plan/metadata_scan.rs +++ b/crates/integrations/datafusion/src/physical_plan/metadata_scan.rs @@ -18,10 +18,14 @@ use std::sync::Arc; use datafusion::catalog::TableProvider; -use datafusion::physical_expr::EquivalenceProperties; +use datafusion::common::tree_node::TreeNodeRecursion; +use datafusion::physical_expr::{EquivalenceProperties, PhysicalExpr}; use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType}; use datafusion::physical_plan::stream::RecordBatchStreamAdapter; -use datafusion::physical_plan::{DisplayAs, ExecutionPlan, Partitioning, PlanProperties}; +use datafusion::physical_plan::{ + ChildrenPropertiesMode, DisplayAs, ExecutionPlan, Partitioning, PlanProperties, + ReplaceChildrenOptions, +}; use futures::TryStreamExt; use crate::metadata_table::IcebergMetadataTableProvider; @@ -62,6 +66,13 @@ impl ExecutionPlan for IcebergMetadataScan { "IcebergMetadataScan" } + fn apply_expressions( + &self, + _f: &mut dyn FnMut(&Arc) -> datafusion::error::Result, + ) -> datafusion::error::Result { + Ok(TreeNodeRecursion::Continue) + } + fn properties(&self) -> &Arc { &self.properties } @@ -70,13 +81,24 @@ impl ExecutionPlan for IcebergMetadataScan { vec![] } - fn with_new_children( + fn replace_children( self: std::sync::Arc, _children: Vec>, + _options: ReplaceChildrenOptions, ) -> datafusion::error::Result> { Ok(self) } + fn with_new_children( + self: std::sync::Arc, + children: Vec>, + ) -> datafusion::error::Result> { + self.replace_children( + children, + ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute), + ) + } + fn execute( &self, _partition: usize, diff --git a/crates/integrations/datafusion/src/physical_plan/scan.rs b/crates/integrations/datafusion/src/physical_plan/scan.rs index 753b6af206..2d5ea728a1 100644 --- a/crates/integrations/datafusion/src/physical_plan/scan.rs +++ b/crates/integrations/datafusion/src/physical_plan/scan.rs @@ -21,12 +21,16 @@ use std::vec; use datafusion::arrow::array::RecordBatch; use datafusion::arrow::datatypes::SchemaRef as ArrowSchemaRef; +use datafusion::common::tree_node::TreeNodeRecursion; use datafusion::error::Result as DFResult; use datafusion::execution::{SendableRecordBatchStream, TaskContext}; -use datafusion::physical_expr::EquivalenceProperties; +use datafusion::physical_expr::{EquivalenceProperties, PhysicalExpr}; use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType}; use datafusion::physical_plan::stream::RecordBatchStreamAdapter; -use datafusion::physical_plan::{DisplayAs, ExecutionPlan, Partitioning, PlanProperties}; +use datafusion::physical_plan::{ + ChildrenPropertiesMode, DisplayAs, ExecutionPlan, Partitioning, PlanProperties, + ReplaceChildrenOptions, +}; use futures::{Stream, TryStreamExt}; use iceberg::expr::Predicate; use iceberg::scan::{FileScanTask, FileScanTaskStream}; @@ -122,13 +126,21 @@ impl ExecutionPlan for IcebergTableScan { "IcebergTableScan" } + fn apply_expressions( + &self, + _f: &mut dyn FnMut(&Arc) -> DFResult, + ) -> DFResult { + Ok(TreeNodeRecursion::Continue) + } + fn children(&self) -> Vec<&Arc> { vec![] } - fn with_new_children( + fn replace_children( self: Arc, children: Vec>, + _options: ReplaceChildrenOptions, ) -> DFResult> { if !children.is_empty() { return Err(datafusion::common::DataFusionError::Internal( @@ -139,6 +151,16 @@ impl ExecutionPlan for IcebergTableScan { Ok(self) } + fn with_new_children( + self: Arc, + children: Vec>, + ) -> DFResult> { + self.replace_children( + children, + ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute), + ) + } + fn properties(&self) -> &Arc { &self.plan_properties } diff --git a/crates/integrations/datafusion/src/physical_plan/write.rs b/crates/integrations/datafusion/src/physical_plan/write.rs index a7d771ec1b..ee4e912331 100644 --- a/crates/integrations/datafusion/src/physical_plan/write.rs +++ b/crates/integrations/datafusion/src/physical_plan/write.rs @@ -24,14 +24,15 @@ use datafusion::arrow::datatypes::{ DataType, Field, Schema as ArrowSchema, SchemaRef as ArrowSchemaRef, }; use datafusion::common::Result as DFResult; +use datafusion::common::tree_node::TreeNodeRecursion; use datafusion::error::DataFusionError; use datafusion::execution::{SendableRecordBatchStream, TaskContext}; -use datafusion::physical_expr::{EquivalenceProperties, Partitioning}; +use datafusion::physical_expr::{EquivalenceProperties, Partitioning, PhysicalExpr}; use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType}; use datafusion::physical_plan::stream::RecordBatchStreamAdapter; use datafusion::physical_plan::{ - DisplayAs, DisplayFormatType, ExecutionPlan, ExecutionPlanProperties, PlanProperties, - execute_input_stream, + ChildrenPropertiesMode, DisplayAs, DisplayFormatType, ExecutionPlan, ExecutionPlanProperties, + PlanProperties, ReplaceChildrenOptions, execute_input_stream, }; use futures::StreamExt; use iceberg::arrow::FieldMatchMode; @@ -66,13 +67,14 @@ pub(crate) struct IcebergWriteExec { } impl IcebergWriteExec { - pub fn new(table: Table, input: Arc, schema: ArrowSchemaRef) -> Self { - let plan_properties = Self::compute_properties(&input, schema); + pub fn new(table: Table, input: Arc) -> Self { + let result_schema = Self::make_result_schema(); + let plan_properties = Self::compute_properties(&input, Arc::clone(&result_schema)); Self { table, input, - result_schema: Self::make_result_schema(), + result_schema, plan_properties, } } @@ -137,6 +139,13 @@ impl ExecutionPlan for IcebergWriteExec { "IcebergWriteExec" } + fn apply_expressions( + &self, + _f: &mut dyn FnMut(&Arc) -> DFResult, + ) -> DFResult { + Ok(TreeNodeRecursion::Continue) + } + /// Prevents the introduction of additional `RepartitionExec` and processing input in parallel. fn benefits_from_input_partitioning(&self) -> Vec { vec![false] @@ -155,9 +164,10 @@ impl ExecutionPlan for IcebergWriteExec { vec![&self.input] } - fn with_new_children( + fn replace_children( self: Arc, - children: Vec>, + mut children: Vec>, + options: ReplaceChildrenOptions, ) -> DFResult> { if children.len() != 1 { return Err(DataFusionError::Internal(format!( @@ -166,11 +176,26 @@ impl ExecutionPlan for IcebergWriteExec { ))); } - Ok(Arc::new(Self::new( - self.table.clone(), - Arc::clone(&children[0]), - self.schema(), - ))) + let input = children.swap_remove(0); + Ok(match options.children_properties { + ChildrenPropertiesMode::Keep => Arc::new(Self { + table: self.table.clone(), + input, + result_schema: Arc::clone(&self.result_schema), + plan_properties: Arc::clone(&self.plan_properties), + }), + ChildrenPropertiesMode::Recompute => Arc::new(Self::new(self.table.clone(), input)), + }) + } + + fn with_new_children( + self: Arc, + children: Vec>, + ) -> DFResult> { + self.replace_children( + children, + ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute), + ) } /// Executes the write operation for the given partition. @@ -315,7 +340,10 @@ mod tests { use datafusion::physical_expr::{EquivalenceProperties, Partitioning}; use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType}; use datafusion::physical_plan::stream::RecordBatchStreamAdapter; - use datafusion::physical_plan::{DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties}; + use datafusion::physical_plan::{ + ChildrenPropertiesMode, DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties, + ReplaceChildrenOptions, + }; use futures::{StreamExt, stream}; use iceberg::memory::{MEMORY_CATALOG_WAREHOUSE, MemoryCatalogBuilder}; use iceberg::spec::{ @@ -374,6 +402,13 @@ mod tests { "MockExecutionPlan" } + fn apply_expressions( + &self, + _f: &mut dyn FnMut(&Arc) -> DFResult, + ) -> DFResult { + Ok(TreeNodeRecursion::Continue) + } + fn properties(&self) -> &Arc { &self.properties } @@ -382,13 +417,24 @@ mod tests { vec![] } - fn with_new_children( + fn replace_children( self: Arc, _children: Vec>, + _options: ReplaceChildrenOptions, ) -> DFResult> { Ok(self) } + fn with_new_children( + self: Arc, + children: Vec>, + ) -> DFResult> { + self.replace_children( + children, + ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute), + ) + } + fn execute( &self, _partition: usize, @@ -494,7 +540,32 @@ mod tests { ])); // 4. Create IcebergWriteExec - let write_exec = IcebergWriteExec::new(table.clone(), input_plan, arrow_schema); + let write_exec = Arc::new(IcebergWriteExec::new(table.clone(), input_plan)); + assert_eq!(write_exec.schema(), IcebergWriteExec::make_result_schema()); + + let replacement_input = + Arc::new(MockExecutionPlan::new(arrow_schema, vec![])) as Arc; + let original_properties = Arc::clone(write_exec.properties()); + let kept_properties = Arc::clone(&write_exec) + .replace_children( + vec![Arc::clone(&replacement_input)], + ReplaceChildrenOptions::new(ChildrenPropertiesMode::Keep), + ) + .expect("replacing a write child with unchanged properties should succeed"); + assert!(Arc::ptr_eq( + &original_properties, + kept_properties.properties() + )); + let recomputed_properties = Arc::clone(&write_exec) + .replace_children( + vec![replacement_input], + ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute), + ) + .expect("replacing a write child should succeed"); + assert!(!Arc::ptr_eq( + &original_properties, + recomputed_properties.properties() + )); // 5. Execute the plan let task_ctx = Arc::new(TaskContext::default()); diff --git a/crates/integrations/datafusion/src/table/mod.rs b/crates/integrations/datafusion/src/table/mod.rs index 9f5500c781..e1db524879 100644 --- a/crates/integrations/datafusion/src/table/mod.rs +++ b/crates/integrations/datafusion/src/table/mod.rs @@ -295,11 +295,7 @@ impl TableProvider for IcebergTableProvider { sort_by_partition(repartitioned_plan)? }; - let write_plan = Arc::new(IcebergWriteExec::new( - table.clone(), - write_input, - self.schema.clone(), - )); + let write_plan = Arc::new(IcebergWriteExec::new(table.clone(), write_input)); // Merge the outputs of write_plan into one so we can commit all files together let coalesce_partitions = Arc::new(CoalescePartitionsExec::new(write_plan)); diff --git a/crates/integrations/datafusion/src/table/table_provider_factory.rs b/crates/integrations/datafusion/src/table/table_provider_factory.rs index 9692340cac..f3733ac260 100644 --- a/crates/integrations/datafusion/src/table/table_provider_factory.rs +++ b/crates/integrations/datafusion/src/table/table_provider_factory.rs @@ -21,9 +21,9 @@ use std::sync::Arc; use async_trait::async_trait; use datafusion::catalog::{Session, TableProvider, TableProviderFactory}; +use datafusion::common::TableReference; use datafusion::error::Result as DFResult; use datafusion::logical_expr::CreateExternalTable; -use datafusion::sql::TableReference; use iceberg::io::{FileIOBuilder, LocalFsStorageFactory, StorageFactory}; use iceberg::table::StaticTable; use iceberg::{Error, ErrorKind, Result, TableIdent}; @@ -41,9 +41,9 @@ use crate::to_datafusion_error; /// ``` /// use std::sync::Arc; /// +/// use datafusion::common::TableReference; /// use datafusion::execution::session_state::SessionStateBuilder; /// use datafusion::prelude::*; -/// use datafusion::sql::TableReference; /// use iceberg_datafusion::IcebergTableProviderFactory; /// /// #[tokio::main] @@ -123,10 +123,9 @@ impl TableProviderFactory for IcebergTableProviderFactory { _state: &dyn Session, cmd: &CreateExternalTable, ) -> DFResult> { - check_cmd(cmd).map_err(to_datafusion_error)?; + let metadata_file_path = check_cmd(cmd).map_err(to_datafusion_error)?; let table_name = &cmd.name; - let metadata_file_path = &cmd.location; let options = &cmd.options; let table_name_with_ns = complement_namespace_if_necessary(table_name); @@ -154,9 +153,10 @@ impl TableProviderFactory for IcebergTableProviderFactory { } } -fn check_cmd(cmd: &CreateExternalTable) -> Result<()> { +fn check_cmd(cmd: &CreateExternalTable) -> Result<&str> { let CreateExternalTable { schema, + locations, table_partition_cols, order_exprs, constraints, @@ -164,6 +164,13 @@ fn check_cmd(cmd: &CreateExternalTable) -> Result<()> { .. } = cmd; + let [metadata_file_path] = locations.as_slice() else { + return Err(Error::new( + ErrorKind::FeatureUnsupported, + "Iceberg external tables require exactly one metadata file location.", + )); + }; + // Check if any of the fields violate the constraints in a single condition let is_invalid = !schema.fields().is_empty() || !table_partition_cols.is_empty() @@ -178,7 +185,7 @@ fn check_cmd(cmd: &CreateExternalTable) -> Result<()> { )); } - Ok(()) + Ok(metadata_file_path) } /// Complements the namespace of a table name if necessary. @@ -218,12 +225,11 @@ mod tests { use datafusion::arrow::datatypes::{DataType, Field, Schema}; use datafusion::catalog::TableProviderFactory; - use datafusion::common::{Constraints, DFSchema}; + use datafusion::common::{Constraints, DFSchema, TableReference}; use datafusion::execution::session_state::SessionStateBuilder; use datafusion::logical_expr::CreateExternalTable; use datafusion::parquet::arrow::PARQUET_FIELD_ID_META_KEY; use datafusion::prelude::SessionContext; - use datafusion::sql::TableReference; use super::*; @@ -257,7 +263,7 @@ mod tests { CreateExternalTable { name: TableReference::partial("static_ns", "static_table"), - location: metadata_file_path, + locations: vec![metadata_file_path], schema: Arc::new(DFSchema::empty()), file_type: "iceberg".to_string(), options: Default::default(), @@ -273,6 +279,20 @@ mod tests { } } + #[test] + fn test_multiple_locations_are_rejected() { + let mut cmd = create_external_table_cmd(); + cmd.locations.push("another/metadata.json".to_string()); + + let error = check_cmd(&cmd).expect_err("multiple locations should be rejected"); + assert_eq!(error.kind(), ErrorKind::FeatureUnsupported); + assert!( + error + .to_string() + .contains("exactly one metadata file location") + ); + } + #[tokio::test] async fn test_schema_of_created_table() { let factory = IcebergTableProviderFactory::new(); diff --git a/crates/integrations/datafusion/tests/integration_datafusion_test.rs b/crates/integrations/datafusion/tests/integration_datafusion_test.rs index ac3a8c4963..1644eec56b 100644 --- a/crates/integrations/datafusion/tests/integration_datafusion_test.rs +++ b/crates/integrations/datafusion/tests/integration_datafusion_test.rs @@ -28,7 +28,7 @@ use datafusion::arrow::record_batch::RecordBatch; use datafusion::datasource::MemTable; use datafusion::execution::context::SessionContext; use datafusion::parquet::arrow::PARQUET_FIELD_ID_META_KEY; -use datafusion::physical_plan::ExecutionPlan; +use datafusion::physical_plan::{ChildrenPropertiesMode, ExecutionPlan, ReplaceChildrenOptions}; use datafusion::prelude::SessionConfig; use expect_test::expect; use iceberg::io::LocalFsStorageFactory; @@ -508,7 +508,10 @@ async fn test_iceberg_table_scan_rejects_non_empty_children() -> Result<()> { let child = Arc::new(EmptyExec::new(scan_plan.schema())) as Arc; let error = scan_plan - .with_new_children(vec![child]) + .replace_children( + vec![child], + ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute), + ) .expect_err("IcebergTableScan should reject children"); assert!(