diff --git a/Cargo.lock b/Cargo.lock index ebe698d2..152a25f4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -170,6 +170,15 @@ dependencies = [ "derive_arbitrary", ] +[[package]] +name = "arc-swap" +version = "1.9.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c049c0be4daef0b145cb3555416b3b8ef5b7888a38aea1a3a155801fe7b0810b" +dependencies = [ + "rustversion", +] + [[package]] name = "arrayref" version = "0.3.9" @@ -184,9 +193,9 @@ checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50" [[package]] name = "arrow" -version = "58.3.0" +version = "59.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "378530e55cd479eda3c14eb345310799717e6f76d0c332041e8487022166b471" +checksum = "ffaaa3e009861fd829d0a24dd6f115aa8e4634324bb092147d43baafe69ca4a7" dependencies = [ "arrow-arith", "arrow-array", @@ -207,9 +216,9 @@ dependencies = [ [[package]] name = "arrow-arith" -version = "58.3.0" +version = "59.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a0ab212d2c1886e802f51c5212d78ebbcbb0bec980fff9dadc1eb8d45cd0b738" +checksum = "3ac95125e1d71c4a252b5a9c729aef111e80418f08aaa6dbabd1ba66918247fc" dependencies = [ "arrow-array", "arrow-buffer", @@ -221,9 +230,9 @@ dependencies = [ [[package]] name = "arrow-array" -version = "58.3.0" +version = "59.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cfd33d3e92f207444098c75b42de99d329562be0cf686b307b097cc52b4e999e" +checksum = "0c60c79628e9a97cb90d7a0dc3e944f216a902f837d4ecabc14d524bddbbc137" dependencies = [ "ahash", "arrow-buffer", @@ -240,9 +249,9 @@ dependencies = [ [[package]] name = "arrow-avro" -version = "58.3.0" +version = "59.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "049230728cd6e093088c8d231b4beede184e35cad7777c1505c0d5a8571f4376" +checksum = "2835d67df2b69bf5de251ee6d289f85650d41b8169dcee0f950ea88747812c32" dependencies = [ "arrow-array", "arrow-buffer", @@ -264,9 +273,9 @@ dependencies = [ [[package]] name = "arrow-buffer" -version = "58.3.0" +version = "59.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0c6cd424c2693bcdbc150d843dc9d4d137dd2de4782ce6df491ad11a3a0416c0" +checksum = "6026f638c400e9878c1b1cc05c3cfd46fbf381285916ab408678701c1df46c1a" dependencies = [ "bytes", "half", @@ -276,9 +285,9 @@ dependencies = [ [[package]] name = "arrow-cast" -version = "58.3.0" +version = "59.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4c5aefb56a2c02e9e2b30746241058b85f8983f0fcff2ba0c6d09006e1cded7f" +checksum = "c82c236c3caf8df5664284f3f1fbe89938852163998c3fdbf37e84ac220445e9" dependencies = [ "arrow-array", "arrow-buffer", @@ -298,9 +307,9 @@ dependencies = [ [[package]] name = "arrow-csv" -version = "58.3.0" +version = "59.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e94e8cf7e517657a52b91ea1263acf38c4ca62a84655d72458a3359b12ab97de" +checksum = "12714e5fb7954159af1e26d4e0d37108bcf1a2ad5ee5c5bf02a944d564d588b7" dependencies = [ "arrow-array", "arrow-cast", @@ -313,9 +322,9 @@ dependencies = [ [[package]] name = "arrow-data" -version = "58.3.0" +version = "59.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3c88210023a2bfee1896af366309a3028fc3bcbd6515fa29a7990ee1baa08ee0" +checksum = "7bd568aa70c4ec5947027b0d5caee94877433b661a0bb9e8ddceeeb5f0c9b1ab" dependencies = [ "arrow-buffer", "arrow-schema", @@ -326,9 +335,9 @@ dependencies = [ [[package]] name = "arrow-flight" -version = "58.3.0" +version = "59.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "28abfe8bf9f124e5fc83b334af4fa58f8d0323ad25312ccb2d1da50178415704" +checksum = "68365401e834743d708094927e2ca727a32d639fe900df04b936e07a36701b74" dependencies = [ "arrow-array", "arrow-buffer", @@ -346,9 +355,9 @@ dependencies = [ [[package]] name = "arrow-ipc" -version = "58.3.0" +version = "59.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "238438f0834483703d88896db6fe5a7138b2230debc31b34c0336c2996e3c64f" +checksum = "e57ee4d470eab1a021bc4b63fa2b2c15d572892bf227b0a982d3b755a6c662b5" dependencies = [ "arrow-array", "arrow-buffer", @@ -362,9 +371,9 @@ dependencies = [ [[package]] name = "arrow-json" -version = "58.3.0" +version = "59.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "205ca2119e6d679d5c133c6f30e68f027738d95ed948cf77677ea69c7800036b" +checksum = "38f47e0e7a284e1f3707a780dc8cd5451b1614e9e398ea2d9ca03c7a2fe9a9ed" dependencies = [ "arrow-array", "arrow-buffer", @@ -387,9 +396,9 @@ dependencies = [ [[package]] name = "arrow-ord" -version = "58.3.0" +version = "59.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1bffd8fd2579286a5d63bac898159873e5094a79009940bcb42bbfce4f19f1d0" +checksum = "a79cf73ad2eba8686ec2aa9bbf8671208e509025f166afc040cedbd94ffe4983" dependencies = [ "arrow-array", "arrow-buffer", @@ -400,9 +409,9 @@ dependencies = [ [[package]] name = "arrow-row" -version = "58.3.0" +version = "59.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bab5994731204603c73ba69267616c50f80780774c6bb0476f1f830625115e0c" +checksum = "cea0f7d8ed6182f14952761e2c0f989852d5aa334fcbc49f73a9f2247c25b879" dependencies = [ "arrow-array", "arrow-buffer", @@ -413,9 +422,9 @@ dependencies = [ [[package]] name = "arrow-schema" -version = "58.3.0" +version = "59.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f633dbfdf39c039ada1bf9e34c694816eb71fbb7dc78f613993b7245e078a1ed" +checksum = "80b3e786a0dd9103acd583a6fb486dbf2f3268466cc0bd571dcf34cef231c1f1" dependencies = [ "serde_core", "serde_json", @@ -423,9 +432,9 @@ dependencies = [ [[package]] name = "arrow-select" -version = "58.3.0" +version = "59.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8cd065c54172ac787cf3f2f8d4107e0d3fdc26edba76fdf4f4cc170258942222" +checksum = "067a67e0361f6c31f4a7248759f36ca4ca71b187a941ed4d49da1c7d3d4db624" dependencies = [ "ahash", "arrow-array", @@ -437,9 +446,9 @@ dependencies = [ [[package]] name = "arrow-string" -version = "58.3.0" +version = "59.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "29dd7cda3ab9692f43a2e4acc444d760cc17b12bb6d8232ddf64e9bab7c06b42" +checksum = "99bc95847f3ff62a2b03d6f8ce2e3e78f01362060549a2a311898dd442f6256d" dependencies = [ "arrow-array", "arrow-buffer", @@ -529,9 +538,9 @@ checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" [[package]] name = "aws-config" -version = "1.8.17" +version = "1.8.18" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "517aa062d8bd9015ee23d6daa5e1c1372328412fdae4e6c4c1be9b69c6ad37a2" +checksum = "e33f815b73a3899c03b380d543532e5865f230dce9678d108dc10732a8682275" dependencies = [ "aws-credential-types", "aws-runtime", @@ -644,10 +653,11 @@ dependencies = [ [[package]] name = "aws-sdk-sso" -version = "1.99.0" +version = "1.101.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9f4055e6099b2ec264abdc0d9bbfffce306c1601809275c861594779a0b04b45" +checksum = "b647baea49ff551960b904f905681e9b4765a6c4ea08631e89dc52d8bd3f5896" dependencies = [ + "arc-swap", "aws-credential-types", "aws-runtime", "aws-smithy-async", @@ -668,10 +678,11 @@ dependencies = [ [[package]] name = "aws-sdk-ssooidc" -version = "1.101.0" +version = "1.103.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "02f009ba0284c5d696425fd7b4dcc5b189f5726f4041b7a5794daecb3a68d598" +checksum = "7ae401c65ff288aa7873117fe535cd32b7b1bb0bc43751d28901a1d5f20636b9" dependencies = [ + "arc-swap", "aws-credential-types", "aws-runtime", "aws-smithy-async", @@ -692,10 +703,11 @@ dependencies = [ [[package]] name = "aws-sdk-sts" -version = "1.104.0" +version = "1.106.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6aa6622798e19e6a76b690562085dd4771c736cd48343464a53ab4ae2f2c9f84" +checksum = "4c80de7bb7d03e9ca8c9fd7b489f20f3948d3f3be91a7953591347d238115408" dependencies = [ + "arc-swap", "aws-credential-types", "aws-runtime", "aws-smithy-async", @@ -801,9 +813,9 @@ dependencies = [ [[package]] name = "aws-smithy-json" -version = "0.62.6" +version = "0.62.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "517089205f18ab4adc5a3e02888cb139bbbbb2e168eac9f396216925d1fbeaf5" +checksum = "701a947f4797e52a911e114a898667c746c39feea467bbd1abd7b3721f702ffa" dependencies = [ "aws-smithy-runtime-api", "aws-smithy-schema", @@ -857,9 +869,9 @@ dependencies = [ [[package]] name = "aws-smithy-runtime-api" -version = "1.12.1" +version = "1.12.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dc117c179ecf39a62a0a3f49f600e9ac26a7ad7dd172177999f83933af776c32" +checksum = "9db177daa6ba8afb9ee1aefcf548c907abcf52065e394ee11a92780057fe0e8c" dependencies = [ "aws-smithy-async", "aws-smithy-runtime-api-macros", @@ -897,9 +909,9 @@ dependencies = [ [[package]] name = "aws-smithy-types" -version = "1.4.8" +version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "056b66dbce2f81cc0c1e2b05bb402eb58f8a3530479d650efadd5bbae9a4050b" +checksum = "32b42fcf341259d85ca10fac9a2f6448a8ec691c6955a18e45bc3b71a85fab85" dependencies = [ "base64-simd", "bytes", @@ -1207,12 +1219,6 @@ version = "1.25.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c8efb64bd706a16a1bdde310ae86b351e4d21550d98d056f22f8a7f7a2183fec" -[[package]] -name = "byteorder" -version = "1.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b" - [[package]] name = "bytes" version = "1.11.1" @@ -1290,9 +1296,9 @@ dependencies = [ [[package]] name = "chrono" -version = "0.4.44" +version = "0.4.45" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c673075a2e0e5f4a1dde27ce9dee1ea4558c7ffe648f576438a20ca1d2acc4b0" +checksum = "1aa79e62e7697b8e29b513a68abacf485adcd1fe8284a4316c5ae868e6633327" dependencies = [ "iana-time-zone", "js-sys", @@ -1863,8 +1869,7 @@ dependencies = [ [[package]] name = "datafusion" version = "54.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "997a31e15872606a49478e670c58302094c97cb96abb0a7d60720f8e92170040" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" dependencies = [ "arrow", "arrow-schema", @@ -1900,7 +1905,7 @@ dependencies = [ "flate2", "futures", "indexmap", - "itertools 0.14.0", + "itertools 0.15.0", "liblzma", "log", "object_store", @@ -1917,8 +1922,7 @@ dependencies = [ [[package]] name = "datafusion-catalog" version = "54.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f7dd61161508f8f5fa1107774ea687bd753c22d83a32eebf963549f89de14139" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" dependencies = [ "arrow", "async-trait", @@ -1932,7 +1936,7 @@ dependencies = [ "datafusion-physical-plan", "datafusion-session", "futures", - "itertools 0.14.0", + "itertools 0.15.0", "log", "object_store", "parking_lot", @@ -1942,8 +1946,7 @@ dependencies = [ [[package]] name = "datafusion-catalog-listing" version = "54.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "897c70f871277f9ce99aa38347be0d679bbe3e617156c4d2a8378cec8a2a0891" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" dependencies = [ "arrow", "async-trait", @@ -1957,7 +1960,7 @@ dependencies = [ "datafusion-physical-expr-common", "datafusion-physical-plan", "futures", - "itertools 0.14.0", + "itertools 0.15.0", "log", "object_store", ] @@ -1965,8 +1968,7 @@ dependencies = [ [[package]] name = "datafusion-cli" version = "54.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2a9533845ab888508718c2ad540cff205d1111525a9952def8b0e584b6453784" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" dependencies = [ "arrow", "async-trait", @@ -1993,8 +1995,7 @@ dependencies = [ [[package]] name = "datafusion-common" version = "54.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "121c9ded5d87d9172319e006f2afdb9928d72dbacd6a90a458d8acb1e3b43a65" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" dependencies = [ "arrow", "arrow-ipc", @@ -2005,9 +2006,10 @@ dependencies = [ "hashbrown 0.17.1", "hex", "indexmap", - "itertools 0.14.0", + "itertools 0.15.0", "libc", "log", + "num-traits", "object_store", "parquet", "recursive", @@ -2020,8 +2022,7 @@ dependencies = [ [[package]] name = "datafusion-common-runtime" version = "54.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "981b9dae74f78ee3d9f714fb49b01919eab975461b56149510c3ba9ea11287d1" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" dependencies = [ "futures", "log", @@ -2031,8 +2032,7 @@ dependencies = [ [[package]] name = "datafusion-datasource" version = "54.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ffd7d295b2ec7c00d8a56562f41ed41062cf0af75549ed891c12a0a09eddfefe" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" dependencies = [ "arrow", "async-compression", @@ -2052,7 +2052,7 @@ dependencies = [ "flate2", "futures", "glob", - "itertools 0.14.0", + "itertools 0.15.0", "liblzma", "log", "object_store", @@ -2067,8 +2067,7 @@ dependencies = [ [[package]] name = "datafusion-datasource-arrow" version = "54.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "552b0b3f342f7ec41b3fbd70f6339dc82a30cfd0349e7f280e7852528085349f" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" dependencies = [ "arrow", "arrow-ipc", @@ -2083,7 +2082,7 @@ dependencies = [ "datafusion-physical-plan", "datafusion-session", "futures", - "itertools 0.14.0", + "itertools 0.15.0", "object_store", "tokio", ] @@ -2091,8 +2090,7 @@ dependencies = [ [[package]] name = "datafusion-datasource-avro" version = "54.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fb517d08967d536284ce70afb5fe8583133779249f2d7b90587d339741a7f195" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" dependencies = [ "arrow", "arrow-avro", @@ -2110,8 +2108,7 @@ dependencies = [ [[package]] name = "datafusion-datasource-csv" version = "54.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "68850aa426b897e879c8b87e512ea8124f1d0a2869a4e51808ddaaddf1bc0ada" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" dependencies = [ "arrow", "async-trait", @@ -2133,8 +2130,7 @@ dependencies = [ [[package]] name = "datafusion-datasource-json" version = "54.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "402f93242ae08ef99139ee2c528a49d087efe88d5c7b2c3ff5480855a40ce54f" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" dependencies = [ "arrow", "async-trait", @@ -2156,10 +2152,10 @@ dependencies = [ [[package]] name = "datafusion-datasource-parquet" version = "54.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ffd2499c1bee0eeccf6a57156105700eeeb17bc701899ac719183c4e74231450" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" dependencies = [ "arrow", + "arrow-schema", "async-trait", "bytes", "datafusion-common", @@ -2176,7 +2172,7 @@ dependencies = [ "datafusion-pruning", "datafusion-session", "futures", - "itertools 0.14.0", + "itertools 0.15.0", "log", "object_store", "parking_lot", @@ -2297,18 +2293,17 @@ dependencies = [ [[package]] name = "datafusion-doc" version = "54.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cb9e7e5d11130c48c8bd4e80c79a9772dd28ce6dc330baca9246205d245b9e2e" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" [[package]] name = "datafusion-execution" version = "54.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "37a8643ab852eb68864e1b72ae789e8066282dce48eea6347ffb0aee33d1ccc0" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" dependencies = [ "arrow", "arrow-buffer", "async-trait", + "bytes", "dashmap", "datafusion-common", "datafusion-expr", @@ -2320,14 +2315,15 @@ dependencies = [ "parquet", "rand 0.9.4", "tempfile", + "tokio", + "tokio-util", "url", ] [[package]] name = "datafusion-expr" version = "54.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6932f4d71eed9c8d9341476a2b845aadfabde5495d08dbcd8fc23881f49fa7a0" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" dependencies = [ "arrow", "arrow-schema", @@ -2340,7 +2336,7 @@ dependencies = [ "datafusion-functions-window-common", "datafusion-physical-expr-common", "indexmap", - "itertools 0.14.0", + "itertools 0.15.0", "recursive", "serde_json", "sqlparser", @@ -2349,20 +2345,18 @@ dependencies = [ [[package]] name = "datafusion-expr-common" version = "54.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0225491839a31b1f7d2cb8092c2d50792e2fe1c1724e4e6d08e011f5feaf4ed2" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" dependencies = [ "arrow", "datafusion-common", "indexmap", - "itertools 0.14.0", + "itertools 0.15.0", ] [[package]] name = "datafusion-functions" version = "54.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "14872c47bfc3d21e53ec82f57074e6987a15941c1e2f43cde4ac6ae2746634e3" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" dependencies = [ "arrow", "arrow-buffer", @@ -2379,7 +2373,7 @@ dependencies = [ "datafusion-macros", "datafusion-physical-expr-common", "hex", - "itertools 0.14.0", + "itertools 0.15.0", "log", "md-5 0.11.0", "memchr", @@ -2393,8 +2387,7 @@ dependencies = [ [[package]] name = "datafusion-functions-aggregate" version = "54.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "75a2ca14e1b609be21e657e2d3130b2f446456b08393b377bb721a33952d2e09" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" dependencies = [ "arrow", "datafusion-common", @@ -2405,7 +2398,6 @@ dependencies = [ "datafusion-macros", "datafusion-physical-expr", "datafusion-physical-expr-common", - "foldhash 0.2.0", "half", "log", "num-traits", @@ -2414,8 +2406,7 @@ dependencies = [ [[package]] name = "datafusion-functions-aggregate-common" version = "54.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1ece74ba09092d2ef9c9b54a38445450aea292a1f8b04faf531936b723a24b3c" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" dependencies = [ "arrow", "datafusion-common", @@ -2426,8 +2417,7 @@ dependencies = [ [[package]] name = "datafusion-functions-nested" version = "54.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3f3e3f9ee8ca59bf70518802107de6f1b88a9509efdc629fadc5de9d6b2d5ef5" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" dependencies = [ "arrow", "arrow-ord", @@ -2442,7 +2432,7 @@ dependencies = [ "datafusion-macros", "datafusion-physical-expr-common", "hashbrown 0.17.1", - "itertools 0.14.0", + "itertools 0.15.0", "itoa", "log", "memchr", @@ -2451,8 +2441,7 @@ dependencies = [ [[package]] name = "datafusion-functions-table" version = "54.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "89161dffc22cf2b50f9f4b1bee83b5221d3b4ed7c2e37fd7aa2b22a5297b3a26" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" dependencies = [ "arrow", "async-trait", @@ -2467,8 +2456,7 @@ dependencies = [ [[package]] name = "datafusion-functions-window" version = "54.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d7339345b226b3874037708bf5023ba1c2de705128f8457a095aae5ae9cb9c78" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" dependencies = [ "arrow", "datafusion-common", @@ -2484,8 +2472,7 @@ dependencies = [ [[package]] name = "datafusion-functions-window-common" version = "54.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fa84836dc2392df6f43d6a29d37fb56a8ebdc8b3f4e10ae8dc15861fd20278fb" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" dependencies = [ "datafusion-common", "datafusion-physical-expr-common", @@ -2494,8 +2481,7 @@ dependencies = [ [[package]] name = "datafusion-macros" version = "54.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "587164e03ad68732aa9e7bfe5686e3f25970d4c64fd4bd80790749840892dae5" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" dependencies = [ "datafusion-doc", "quote", @@ -2505,8 +2491,7 @@ dependencies = [ [[package]] name = "datafusion-optimizer" version = "54.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "77f20e8cf9e8654d92f4c16b24c487353ee5bf153ffc12d5772cd399ab8cd281" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" dependencies = [ "arrow", "chrono", @@ -2515,7 +2500,7 @@ dependencies = [ "datafusion-expr-common", "datafusion-physical-expr", "indexmap", - "itertools 0.14.0", + "itertools 0.15.0", "log", "recursive", "regex", @@ -2525,8 +2510,7 @@ dependencies = [ [[package]] name = "datafusion-physical-expr" version = "54.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f015a4a82f6f7ff7e1d8d4bf3870a936752fa38b17705dfcc14adef95aa8922c" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" dependencies = [ "arrow", "datafusion-common", @@ -2534,10 +2518,11 @@ dependencies = [ "datafusion-expr-common", "datafusion-functions-aggregate-common", "datafusion-physical-expr-common", + "datafusion-proto-models", "half", "hashbrown 0.17.1", "indexmap", - "itertools 0.14.0", + "itertools 0.15.0", "parking_lot", "petgraph", "recursive", @@ -2547,8 +2532,7 @@ dependencies = [ [[package]] name = "datafusion-physical-expr-adapter" version = "54.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "51e6ffff8acdfe54e0ea15ccf38115c4a9184433b0439f42907637928d00a235" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" dependencies = [ "arrow", "datafusion-common", @@ -2556,22 +2540,22 @@ 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 = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7967a3e171c6a4bf09474b3f7a14f1a3db13ed1714ba12156f33fcce2bba54e8" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" dependencies = [ "arrow", "chrono", "datafusion-common", "datafusion-expr-common", + "datafusion-proto-models", "hashbrown 0.17.1", "indexmap", - "itertools 0.14.0", + "itertools 0.15.0", "parking_lot", "pin-project", ] @@ -2579,8 +2563,7 @@ dependencies = [ [[package]] name = "datafusion-physical-optimizer" version = "54.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "59ff803e2a96054cb6d83f35f9e60fd4f42eac515e1932bd1b2dbc91d5fcbf36" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" dependencies = [ "arrow", "datafusion-common", @@ -2591,15 +2574,14 @@ dependencies = [ "datafusion-physical-expr-common", "datafusion-physical-plan", "datafusion-pruning", - "itertools 0.14.0", + "itertools 0.15.0", "recursive", ] [[package]] name = "datafusion-physical-plan" version = "54.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "776ee54d47d15bdb126452f9ca17b03761e3b004682914beaedd3f86eb507fbc" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" dependencies = [ "arrow", "arrow-data", @@ -2607,6 +2589,7 @@ dependencies = [ "arrow-ord", "arrow-schema", "async-trait", + "bytes", "datafusion-common", "datafusion-common-runtime", "datafusion-execution", @@ -2616,23 +2599,25 @@ 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", - "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 = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9dd15a1ba5d3af93808241065c6c44dbca8296a189845e8a587c45c07bf0ffae" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" dependencies = [ "arrow", "chrono", @@ -2651,6 +2636,7 @@ dependencies = [ "datafusion-physical-expr-common", "datafusion-physical-plan", "datafusion-proto-common", + "datafusion-proto-models", "object_store", "prost", ] @@ -2658,19 +2644,26 @@ dependencies = [ [[package]] name = "datafusion-proto-common" version = "54.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "90042982cf9462eb06a0b81f92efa4188dae871e7ea3ab8dc61aa9c9349b2530" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" dependencies = [ "arrow", "datafusion-common", "prost", ] +[[package]] +name = "datafusion-proto-models" +version = "54.0.0" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" +dependencies = [ + "datafusion-proto-common", + "prost", +] + [[package]] name = "datafusion-pruning" version = "54.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d5fb9e5774660aa69c3ba93c610f175f75b65cb8c3776edb3626de8f3a4f4ee3" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" dependencies = [ "arrow", "datafusion-common", @@ -2685,8 +2678,7 @@ dependencies = [ [[package]] name = "datafusion-session" version = "54.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "15ce715fa2a61f4623cc234bcc14a3ef6a91f189128d5b14b468a6a17cdfc417" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" dependencies = [ "async-trait", "datafusion-common", @@ -2699,8 +2691,7 @@ dependencies = [ [[package]] name = "datafusion-sql" version = "54.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6094ad36a3ed6d7ac87b20b479b2d0b118250f66cf997603829fdc65b44a7099" +source = "git+https://github.com/apache/datafusion.git?rev=01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b#01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" dependencies = [ "arrow", "bigdecimal", @@ -3817,12 +3808,6 @@ dependencies = [ "syn 2.0.117", ] -[[package]] -name = "integer-encoding" -version = "3.0.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8bb03732005da905c88227371639bf1ad885cc712789c011c31c5fb3ab3ccf02" - [[package]] name = "ipnet" version = "2.12.0" @@ -3864,6 +3849,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" @@ -4202,9 +4196,9 @@ dependencies = [ [[package]] name = "memchr" -version = "2.8.0" +version = "2.8.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f8ca58f447f06ed17d5fc4043ce1b10dd205e060fb3ce5b979b8ed8e59ff3f79" +checksum = "88904434abc2901f197fe8cc55f0445e7ded921dba5911dad2e2b39b48e663c4" [[package]] name = "memmem" @@ -4543,15 +4537,6 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "04744f49eae99ab78e0d5c0b603ab218f515ea8cfe5a456d7629ad883a3b6e7d" -[[package]] -name = "ordered-float" -version = "2.10.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "68f19d67e5a2795c94e73e0bb1cc1a7edeb2e28efd39e2e1c9b7a40c1108b11c" -dependencies = [ - "num-traits", -] - [[package]] name = "ordered-float" version = "4.6.0" @@ -4604,9 +4589,9 @@ dependencies = [ [[package]] name = "parquet" -version = "58.3.0" +version = "59.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5dafa7d01085b62a47dd0c1829550a0a36710ea9c4fe358a05a85477cec8a908" +checksum = "970dff83e97d953c827ae8176f6bf4e9f77bf62daacc01ec5df348ec5eacd913" dependencies = [ "ahash", "arrow-array", @@ -4633,7 +4618,6 @@ dependencies = [ "seq-macro", "simdutf8", "snap", - "thrift", "tokio", "twox-hash", "zstd", @@ -5643,6 +5627,7 @@ version = "1.0.150" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e8014e44b4736ed0538adeecded0fce2a272f22dc9578a7eb6b2d9993c74cfb9" dependencies = [ + "indexmap", "itoa", "memchr", "serde", @@ -6084,7 +6069,7 @@ dependencies = [ "nix 0.29.0", "num-derive", "num-traits", - "ordered-float 4.6.0", + "ordered-float", "pest", "pest_derive", "phf 0.11.3", @@ -6196,17 +6181,6 @@ dependencies = [ "cfg-if", ] -[[package]] -name = "thrift" -version = "0.17.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7e54bc85fc7faa8bc175c4bab5b92ba8d9a3ce893d0e9f42cc455c8ab16a9e09" -dependencies = [ - "byteorder", - "integer-encoding", - "ordered-float 2.10.1", -] - [[package]] name = "time" version = "0.3.47" @@ -6458,12 +6432,12 @@ checksum = "8df9b6e13f2d32c91b9bd719c00d1958837bc7dec474d94952798cc8e69eeec3" [[package]] name = "tpchgen" version = "2.0.2" -source = "git+https://github.com/clflushopt/tpchgen-rs?rev=438e9c2dbc25b2fff82c0efc08b3f13b5707874f#438e9c2dbc25b2fff82c0efc08b3f13b5707874f" +source = "git+https://github.com/clflushopt/tpchgen-rs?rev=f42708c32e7fd816b8ad8d2a880c1802a71b669f#f42708c32e7fd816b8ad8d2a880c1802a71b669f" [[package]] name = "tpchgen-arrow" version = "2.0.2" -source = "git+https://github.com/clflushopt/tpchgen-rs?rev=438e9c2dbc25b2fff82c0efc08b3f13b5707874f#438e9c2dbc25b2fff82c0efc08b3f13b5707874f" +source = "git+https://github.com/clflushopt/tpchgen-rs?rev=f42708c32e7fd816b8ad8d2a880c1802a71b669f#f42708c32e7fd816b8ad8d2a880c1802a71b669f" dependencies = [ "arrow", "tpchgen", @@ -6882,7 +6856,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5f2ab60e120fd6eaa68d9567f3226e876684639d22a4219b313ff69ec0ccd5ac" dependencies = [ "log", - "ordered-float 4.6.0", + "ordered-float", "strsim 0.11.1", "thiserror 1.0.69", "wezterm-dynamic-derive", diff --git a/Cargo.toml b/Cargo.toml index 4d6e3e7e..c461b5e9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,8 +2,8 @@ members = ["benchmarks", "cli", "console"] [workspace.dependencies] -datafusion = { version = "54.0.0", default-features = false } -datafusion-proto = { version = "54.0.0" } +datafusion = { git = "https://github.com/apache/datafusion.git", rev = "01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b", default-features = false } +datafusion-proto = { git = "https://github.com/apache/datafusion.git", rev = "01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b" } [package] name = "datafusion-distributed" @@ -23,9 +23,9 @@ datafusion = { workspace = true, features = [ "datetime_expressions", ] } datafusion-proto = { workspace = true } -arrow-flight = "58" -arrow-select = "58" -arrow-ipc = { version = "58", features = ["zstd"] } +arrow-flight = "59" +arrow-select = "59" +arrow-ipc = { version = "59", features = ["zstd"] } async-trait = "0.1.89" tokio = { version = "1.48", features = ["full"] } tonic = { version = "0.14.1", features = ["transport"] } @@ -53,8 +53,8 @@ tonic-prost = "0.14.2" # integration_tests deps insta = { version = "1.46.0", features = ["filters"], optional = true } -parquet = { version = "58", optional = true } -arrow = { version = "58", optional = true, features = ["test_utils"] } +parquet = { version = "59", optional = true } +arrow = { version = "59", optional = true, features = ["test_utils"] } hyper-util = { version = "0.1.16", optional = true } [features] @@ -73,8 +73,8 @@ sysinfo = ["dep:sysinfo"] datafusion-distributed-benchmarks = { path = "benchmarks" } structopt = "0.3" insta = { version = "1.46.0", features = ["filters"] } -parquet = "58" -arrow = { version = "58", features = ["test_utils"] } +parquet = "59" +arrow = { version = "59", features = ["test_utils"] } tokio-stream = { version = "0.1.17", features = ["sync"] } hyper-util = "0.1.16" pretty_assertions = "1.4" diff --git a/benchmarks/Cargo.toml b/benchmarks/Cargo.toml index 0a03ead0..a40f7359 100644 --- a/benchmarks/Cargo.toml +++ b/benchmarks/Cargo.toml @@ -11,7 +11,7 @@ datafusion-distributed = { path = "..", features = [ "system-metrics", ] } tokio = { version = "1.48", features = ["full"] } -parquet = { version = "58" } +parquet = { version = "59" } structopt = { version = "0.3.26" } log = "0.4.27" serde = "1.0.219" @@ -20,10 +20,10 @@ env_logger = "0.11.8" async-trait = "0.1.89" futures = "0.3.31" url = "2.5.7" -arrow = { version = "58", features = ["test_utils"] } +arrow = { version = "59", features = ["test_utils"] } tonic = { version = "0.14.1", features = ["transport"] } -tpchgen = { git = "https://github.com/clflushopt/tpchgen-rs", rev = "438e9c2dbc25b2fff82c0efc08b3f13b5707874f" } -tpchgen-arrow = { git = "https://github.com/clflushopt/tpchgen-rs", rev = "438e9c2dbc25b2fff82c0efc08b3f13b5707874f" } +tpchgen = { git = "https://github.com/clflushopt/tpchgen-rs", rev = "f42708c32e7fd816b8ad8d2a880c1802a71b669f" } +tpchgen-arrow = { git = "https://github.com/clflushopt/tpchgen-rs", rev = "f42708c32e7fd816b8ad8d2a880c1802a71b669f" } reqwest = "0.12" zip = "6.0" axum = "0.7" diff --git a/cli/Cargo.toml b/cli/Cargo.toml index 66c85565..fd8b0c0a 100644 --- a/cli/Cargo.toml +++ b/cli/Cargo.toml @@ -6,7 +6,7 @@ edition = "2024" [dependencies] datafusion = { workspace = true } datafusion-distributed = { path = "..", features = ["avro", "integration"] } -datafusion-cli = { version = "54", default-features = false } +datafusion-cli = { git = "https://github.com/apache/datafusion.git", rev = "01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b", default-features = false } tokio = { version = "1.48", features = ["full"] } clap = { version = "4", features = ["derive"] } env_logger = "0.11" diff --git a/cli/src/main.rs b/cli/src/main.rs index f3a10491..e3bea80e 100644 --- a/cli/src/main.rs +++ b/cli/src/main.rs @@ -18,7 +18,7 @@ // File mainly copied from https://github.com/apache/datafusion/blob/main/datafusion-cli/src/main.rs use clap::Parser; -use datafusion::common::config_err; +use datafusion::common::{config::ConfigNonZeroUsize, config_err}; use datafusion::config::ConfigOptions; use datafusion::error::{DataFusionError, Result}; use datafusion::execution::SessionStateBuilder; @@ -214,7 +214,7 @@ fn get_session_config(args: &Args) -> Result { if batch_size == 0 { return config_err!("batch_size must be greater than 0"); } - config_options.execution.batch_size = batch_size; + config_options.execution.batch_size = ConfigNonZeroUsize::try_new(batch_size)?; }; // use easier to understand "tree" mode by default diff --git a/console/Cargo.toml b/console/Cargo.toml index 5267842e..5f043375 100644 --- a/console/Cargo.toml +++ b/console/Cargo.toml @@ -17,7 +17,7 @@ url = "2.5.7" tokio-stream = "0.1.18" [dev-dependencies] -arrow = "58" +arrow = "59" async-trait = "0.1.89" datafusion-distributed-benchmarks = { path = "../benchmarks" } futures = "0.3.31" diff --git a/examples/custom_execution_plan.rs b/examples/custom_execution_plan.rs index e322139c..76d4d941 100644 --- a/examples/custom_execution_plan.rs +++ b/examples/custom_execution_plan.rs @@ -42,7 +42,7 @@ use datafusion_distributed::{ DistributedExt, DistributedTaskContext, SessionStateBuilderExt, TaskEstimation, TaskEstimator, WorkerQueryContext, display_plan_ascii, }; -use datafusion_proto::physical_plan::PhysicalExtensionCodec; +use datafusion_proto::physical_plan::{PhysicalExtensionCodec, PhysicalProtoConverterExtension}; use datafusion_proto::protobuf; use datafusion_proto::protobuf::proto_error; use futures::{TryStreamExt, stream}; @@ -241,6 +241,7 @@ impl PhysicalExtensionCodec for NumbersExecCodec { buf: &[u8], inputs: &[Arc], _ctx: &TaskContext, + _proto_converter: &dyn PhysicalProtoConverterExtension, ) -> Result> { if !inputs.is_empty() { return internal_err!("NumbersExec should have no children, got {}", inputs.len()); @@ -261,7 +262,12 @@ impl PhysicalExtensionCodec for NumbersExecCodec { ))) } - fn try_encode(&self, node: Arc, buf: &mut Vec) -> Result<()> { + fn try_encode( + &self, + node: Arc, + buf: &mut Vec, + _proto_converter: &dyn PhysicalProtoConverterExtension, + ) -> Result<()> { let Some(exec) = node.downcast_ref::() else { return internal_err!("Expected plan to be NumbersExec, but was {}", node.name()); }; diff --git a/examples/custom_worker_url_routing.rs b/examples/custom_worker_url_routing.rs index 4ae9735f..2cb6b9a9 100644 --- a/examples/custom_worker_url_routing.rs +++ b/examples/custom_worker_url_routing.rs @@ -44,8 +44,8 @@ use datafusion_distributed::{ DistributedConfig, DistributedExt, DistributedLeafExec, SessionStateBuilderExt, TaskEstimation, TaskEstimator, TaskRoutingContext, WorkerQueryContext, display_plan_ascii, }; -use datafusion_proto::physical_plan::PhysicalExtensionCodec; -use datafusion_proto::protobuf; +use datafusion_proto::physical_plan::{PhysicalExtensionCodec, PhysicalProtoConverterExtension}; +use datafusion_proto::{TryFromProto, protobuf}; use futures::TryStreamExt; use prost::Message; use std::error::Error; @@ -151,7 +151,7 @@ impl ExecutionPlan for CacheExec { fn hash_key(file_group: &FileGroup) -> usize { let mut hasher = DefaultHasher::new(); for file in file_group.files() { - let serialized: protobuf::PartitionedFile = file.try_into().unwrap(); + let serialized = protobuf::PartitionedFile::try_from_proto(file).unwrap(); hasher.write(&serialized.encode_to_vec()); } hasher.finish() as usize @@ -255,6 +255,7 @@ impl PhysicalExtensionCodec for CachedFileScanCodec { _buf: &[u8], inputs: &[Arc], _ctx: &TaskContext, + _proto_converter: &dyn PhysicalProtoConverterExtension, ) -> Result> { let [child] = inputs else { return internal_err!("CacheExec expects exactly 1 child, got {}", inputs.len()); @@ -262,7 +263,12 @@ impl PhysicalExtensionCodec for CachedFileScanCodec { Ok(CacheExec::new(Arc::clone(child))) } - fn try_encode(&self, node: Arc, _buf: &mut Vec) -> Result<()> { + fn try_encode( + &self, + node: Arc, + _buf: &mut Vec, + _proto_converter: &dyn PhysicalProtoConverterExtension, + ) -> Result<()> { if node.downcast_ref::().is_none() { return internal_err!("Expected CacheExec, got {}", node.name()); } diff --git a/examples/work_unit_feed.rs b/examples/work_unit_feed.rs index 790c32d4..8b70b314 100644 --- a/examples/work_unit_feed.rs +++ b/examples/work_unit_feed.rs @@ -33,7 +33,7 @@ use datafusion_distributed::{ DistributedExt, DistributedTaskContext, SessionStateBuilderExt, TaskEstimation, TaskEstimator, WorkUnitFeed, WorkUnitFeedProto, WorkUnitFeedProvider, WorkerQueryContext, display_plan_ascii, }; -use datafusion_proto::physical_plan::PhysicalExtensionCodec; +use datafusion_proto::physical_plan::{PhysicalExtensionCodec, PhysicalProtoConverterExtension}; use datafusion_proto::protobuf::proto_error; use futures::stream::BoxStream; use futures::{StreamExt, TryStreamExt}; @@ -209,6 +209,7 @@ impl PhysicalExtensionCodec for RemoteScanExecCodec { buf: &[u8], _inputs: &[Arc], _ctx: &TaskContext, + _proto_converter: &dyn PhysicalProtoConverterExtension, ) -> Result> { let p = RemoteScanProto::decode(buf).map_err(|e| proto_error(format!("{e}")))?; let feed = WorkUnitFeed::::from_proto( @@ -223,7 +224,12 @@ impl PhysicalExtensionCodec for RemoteScanExecCodec { ))) } - fn try_encode(&self, node: Arc, buf: &mut Vec) -> Result<()> { + fn try_encode( + &self, + node: Arc, + buf: &mut Vec, + _proto_converter: &dyn PhysicalProtoConverterExtension, + ) -> Result<()> { let exec = node .downcast_ref::() .ok_or_else(|| proto_error(format!("expected RemoteScanExec, got {}", node.name())))?; diff --git a/src/distributed_ext.rs b/src/distributed_ext.rs index 17852a04..2863e9d0 100644 --- a/src/distributed_ext.rs +++ b/src/distributed_ext.rs @@ -2,6 +2,7 @@ use crate::config_extension_ext::{ set_distributed_option_extension, set_distributed_option_extension_from_headers, }; use crate::distributed_planner::set_distributed_task_estimator; +use crate::filter_expressions::{PlanFilterExpressions, set_distributed_filter_expressions}; use crate::networking::{set_distributed_channel_resolver, set_distributed_worker_resolver}; use crate::passthrough_headers::set_passthrough_headers; use crate::protobuf::{set_distributed_user_codec, set_distributed_user_codec_arc}; @@ -145,18 +146,18 @@ pub trait DistributedExt: Sized { /// # use datafusion::execution::{SessionState, FunctionRegistry, SessionStateBuilder, TaskContext}; /// # use datafusion::physical_plan::ExecutionPlan; /// # use datafusion::prelude::SessionConfig; - /// # use datafusion_proto::physical_plan::PhysicalExtensionCodec; + /// # use datafusion_proto::physical_plan::{PhysicalExtensionCodec, PhysicalProtoConverterExtension}; /// # use datafusion_distributed::{DistributedExt, WorkerQueryContext}; /// /// #[derive(Debug)] /// struct CustomExecCodec; /// /// impl PhysicalExtensionCodec for CustomExecCodec { - /// fn try_decode(&self, buf: &[u8], inputs: &[Arc], ctx: &TaskContext) -> datafusion::common::Result> { + /// fn try_decode(&self, buf: &[u8], inputs: &[Arc], ctx: &TaskContext, proto_converter: &dyn PhysicalProtoConverterExtension) -> datafusion::common::Result> { /// todo!() /// } /// - /// fn try_encode(&self, node: Arc, buf: &mut Vec) -> datafusion::common::Result<()> { + /// fn try_encode(&self, node: Arc, buf: &mut Vec, proto_converter: &dyn PhysicalProtoConverterExtension) -> datafusion::common::Result<()> { /// todo!() /// } /// } @@ -577,6 +578,19 @@ pub trait DistributedExt: Sized { P: WorkUnitFeedProvider + 'static, P::WorkUnit: 'static, F: Fn(&T) -> Option<&WorkUnitFeed

> + Send + Sync + 'static; + + /// Registers a plan node type that exposes filter expressions for dynamic filter display. + /// + /// Custom plan nodes should implement [PlanFilterExpressions] and register the concrete type + /// on both the coordinator and worker session builders. + fn with_distributed_filter_expressions(self) -> Self + where + T: PlanFilterExpressions + 'static; + + /// Same as [DistributedExt::with_distributed_filter_expressions] but with an in-place mutation. + fn set_distributed_filter_expressions(&mut self) + where + T: PlanFilterExpressions + 'static; } impl DistributedExt for SessionConfig { @@ -722,6 +736,13 @@ impl DistributedExt for SessionConfig { }) } + fn set_distributed_filter_expressions(&mut self) + where + T: PlanFilterExpressions + 'static, + { + set_distributed_filter_expressions::(self) + } + delegate! { to self { #[call(set_distributed_option_extension)] @@ -804,6 +825,12 @@ impl DistributedExt for SessionConfig { P: WorkUnitFeedProvider + 'static, P::WorkUnit: 'static, F: Fn(&T) -> Option<&WorkUnitFeed

> + Send + Sync + 'static; + + #[call(set_distributed_filter_expressions)] + #[expr($;self)] + fn with_distributed_filter_expressions(mut self) -> Self + where + T: PlanFilterExpressions + 'static; } } } @@ -915,6 +942,15 @@ impl DistributedExt for SessionStateBuilder { P: WorkUnitFeedProvider + 'static, P::WorkUnit: 'static, F: Fn(&T) -> Option<&WorkUnitFeed

> + Send + Sync + 'static; + + fn set_distributed_filter_expressions(&mut self) + where + T: PlanFilterExpressions + 'static; + #[call(set_distributed_filter_expressions)] + #[expr($;self)] + fn with_distributed_filter_expressions(mut self) -> Self + where + T: PlanFilterExpressions + 'static; } } } @@ -1026,6 +1062,15 @@ impl DistributedExt for SessionState { P: WorkUnitFeedProvider + 'static, P::WorkUnit: 'static, F: Fn(&T) -> Option<&WorkUnitFeed

> + Send + Sync + 'static; + + fn set_distributed_filter_expressions(&mut self) + where + T: PlanFilterExpressions + 'static; + #[call(set_distributed_filter_expressions)] + #[expr($;self)] + fn with_distributed_filter_expressions(mut self) -> Self + where + T: PlanFilterExpressions + 'static; } } } @@ -1137,6 +1182,15 @@ impl DistributedExt for SessionContext { P: WorkUnitFeedProvider + 'static, P::WorkUnit: 'static, F: Fn(&T) -> Option<&WorkUnitFeed

> + Send + Sync + 'static; + + fn set_distributed_filter_expressions(&mut self) + where + T: PlanFilterExpressions + 'static; + #[call(set_distributed_filter_expressions)] + #[expr($;self)] + fn with_distributed_filter_expressions(self) -> Self + where + T: PlanFilterExpressions + 'static; } } } diff --git a/src/distributed_planner/distributed_config.rs b/src/distributed_planner/distributed_config.rs index d6faee39..86df41e2 100644 --- a/src/distributed_planner/distributed_config.rs +++ b/src/distributed_planner/distributed_config.rs @@ -1,4 +1,5 @@ use crate::distributed_planner::task_estimator::CombinedTaskEstimator; +use crate::filter_expressions::PlanFilterExpressionRegistry; use crate::networking::{ChannelResolverExtension, WorkerResolverExtension}; use crate::work_unit_feed::WorkUnitFeedRegistry; use crate::{TaskEstimator, WorkerResolver}; @@ -78,6 +79,9 @@ extensions_options! { /// [WorkUnitFeedRegistry] that contains a set of getters that, applied to each node in a /// plan, will return the [crate::WorkUnitFeed]s present in all nodes. pub(crate) __private_work_unit_feed_registry: WorkUnitFeedRegistry, default = WorkUnitFeedRegistry::default() + /// [PlanFilterExpressionRegistry] that contains typed extractors for filter expressions + /// present in physical plan nodes. + pub(crate) __private_filter_expression_registry: PlanFilterExpressionRegistry, default = PlanFilterExpressionRegistry::default() } } @@ -210,3 +214,13 @@ impl Debug for WorkUnitFeedRegistry { write!(f, "WorkUnitFeedRegistry") } } + +impl ConfigField for PlanFilterExpressionRegistry { + fn visit(&self, _: &mut V, _: &str, _: &'static str) { + // nothing to do. + } + + fn set(&mut self, _: &str, _: &str) -> datafusion::common::Result<()> { + not_impl_err!("Not implemented") + } +} diff --git a/src/distributed_planner/distributed_query_planner.rs b/src/distributed_planner/distributed_query_planner.rs index f5ac6168..e24aa49d 100644 --- a/src/distributed_planner/distributed_query_planner.rs +++ b/src/distributed_planner/distributed_query_planner.rs @@ -331,11 +331,11 @@ mod tests { let physical_plan_ascii = TestPlanBuilder::default() .physical_plan_as_ascii(query, false) .await; - assert_snapshot!(physical_plan_ascii, @r" + assert_snapshot!(physical_plan_ascii, @" HashJoinExec: mode=CollectLeft, join_type=Left, on=[(RainToday@1, RainToday@1)], projection=[MinTemp@0, MaxTemp@2] CoalescePartitionsExec DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet], [/testdata/weather/result-000001.parquet], [/testdata/weather/result-000002.parquet]]}, projection=[MinTemp, RainToday], file_type=parquet - DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet], [/testdata/weather/result-000001.parquet], [/testdata/weather/result-000002.parquet]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ] + DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet], [/testdata/weather/result-000001.parquet], [/testdata/weather/result-000002.parquet]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible "); } @@ -367,7 +367,7 @@ mod tests { let physical_plan_ascii = TestPlanBuilder::default() .physical_plan_as_ascii(query, false) .await; - assert_snapshot!(physical_plan_ascii, @r" + assert_snapshot!(physical_plan_ascii, @" ┌───── DistributedExec ── Tasks: t0:[p0] │ CoalescePartitionsExec │ HashJoinExec: mode=CollectLeft, join_type=Left, on=[(RainTomorrow@1, RainTomorrow@1)], projection=[MinTemp@0, MaxTemp@2] @@ -398,9 +398,9 @@ mod tests { │ FilterExec: RainToday@1 = no, projection=[MaxTemp@0, RainTomorrow@2] │ RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=3 │ DistributedLeafExec: - │ t0: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000000.parquet:.., /testdata/weather/result-000001.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday, RainTomorrow], file_type=parquet, predicate=RainToday@19 = no AND DynamicFilter [ empty ], pruning_predicate=RainToday_null_count@2 != row_count@3 AND RainToday_min@0 <= no AND no <= RainToday_max@1, required_guarantees=[RainToday in (no)] - │ t1: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000001.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday, RainTomorrow], file_type=parquet, predicate=RainToday@19 = no AND DynamicFilter [ empty ], pruning_predicate=RainToday_null_count@2 != row_count@3 AND RainToday_min@0 <= no AND no <= RainToday_max@1, required_guarantees=[RainToday in (no)] - │ t2: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000001.parquet:.., /testdata/weather/result-000002.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday, RainTomorrow], file_type=parquet, predicate=RainToday@19 = no AND DynamicFilter [ empty ], pruning_predicate=RainToday_null_count@2 != row_count@3 AND RainToday_min@0 <= no AND no <= RainToday_max@1, required_guarantees=[RainToday in (no)] + │ t0: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000000.parquet:.., /testdata/weather/result-000001.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday, RainTomorrow], file_type=parquet, predicate=RainToday@19 = no AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible, pruning_predicate=RainToday_null_count@2 != row_count@3 AND RainToday_min@0 <= no AND no <= RainToday_max@1, required_guarantees=[RainToday in (no)] + │ t1: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000001.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday, RainTomorrow], file_type=parquet, predicate=RainToday@19 = no AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible, pruning_predicate=RainToday_null_count@2 != row_count@3 AND RainToday_min@0 <= no AND no <= RainToday_max@1, required_guarantees=[RainToday in (no)] + │ t2: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000001.parquet:.., /testdata/weather/result-000002.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday, RainTomorrow], file_type=parquet, predicate=RainToday@19 = no AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible, pruning_predicate=RainToday_null_count@2 != row_count@3 AND RainToday_min@0 <= no AND no <= RainToday_max@1, required_guarantees=[RainToday in (no)] └────────────────────────────────────────────────── "); } @@ -514,7 +514,7 @@ mod tests { .num_workers(2) .physical_plan_as_ascii(query, false) .await; - assert_snapshot!(physical_plan_ascii, @r" + assert_snapshot!(physical_plan_ascii, @" ┌───── DistributedExec ── Tasks: t0:[p0] │ CoalescePartitionsExec │ [Stage 1] => NetworkCoalesceExec: output_partitions=4, input_tasks=2 @@ -524,7 +524,7 @@ mod tests { │ DistributedLeafExec: │ t0: DataSourceExec: file_groups={2 groups: [[/testdata/weather/result-000000.parquet:.., /testdata/weather/result-000001.parquet:..], [/testdata/weather/result-000001.parquet:.., /testdata/weather/result-000002.parquet:..]]}, projection=[1 as Int64(1)], file_type=parquet │ DistributedLeafExec: - │ t0: DataSourceExec: file_groups={1 group: [[/testdata/flights-1m.parquet:..]]}, projection=[1 as Int64(1)], file_type=parquet + │ t0: DataSourceExec: file_groups={2 groups: [[/testdata/flights-1m.parquet:..], [/testdata/flights-1m.parquet:..]]}, projection=[1 as Int64(1)], file_type=parquet └────────────────────────────────────────────────── "); } @@ -540,17 +540,17 @@ mod tests { .distributed_max_tasks_per_stage(2) .physical_plan_as_ascii(query, false) .await; - assert_snapshot!(physical_plan_ascii, @r" + assert_snapshot!(physical_plan_ascii, @" ┌───── DistributedExec ── Tasks: t0:[p0] │ CoalescePartitionsExec - │ [Stage 1] => NetworkCoalesceExec: output_partitions=6, input_tasks=2 + │ [Stage 1] => NetworkCoalesceExec: output_partitions=8, input_tasks=2 └────────────────────────────────────────────────── - ┌───── Stage 1 ── Tasks: t0:[p0..p2] t1:[p3..p5] + ┌───── Stage 1 ── Tasks: t0:[p0..p3] t1:[p4..p7] │ DistributedUnionExec: t0:[c0] t1:[c1] │ DistributedLeafExec: │ t0: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000000.parquet:.., /testdata/weather/result-000001.parquet:.., /testdata/weather/result-000002.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[1 as Int64(1)], file_type=parquet │ DistributedLeafExec: - │ t0: DataSourceExec: file_groups={1 group: [[/testdata/flights-1m.parquet:..]]}, projection=[1 as Int64(1)], file_type=parquet + │ t0: DataSourceExec: file_groups={4 groups: [[/testdata/flights-1m.parquet:..], [/testdata/flights-1m.parquet:..], [/testdata/flights-1m.parquet:..], [/testdata/flights-1m.parquet:..]]}, projection=[1 as Int64(1)], file_type=parquet └────────────────────────────────────────────────── "); } @@ -720,7 +720,7 @@ mod tests { .broadcast_joins(true) .physical_plan_as_ascii(query, false) .await; - assert_snapshot!(physical_plan_ascii, @r" + assert_snapshot!(physical_plan_ascii, @" ┌───── DistributedExec ── Tasks: t0:[p0] │ CoalescePartitionsExec │ [Stage 2] => NetworkCoalesceExec: output_partitions=9, input_tasks=3 @@ -730,9 +730,9 @@ mod tests { │ CoalescePartitionsExec │ [Stage 1] => NetworkBroadcastExec: partitions_per_consumer=3, stage_partitions=9, input_tasks=3 │ DistributedLeafExec: - │ t0: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000000.parquet:.., /testdata/weather/result-000001.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ] - │ t1: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000001.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ] - │ t2: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000001.parquet:.., /testdata/weather/result-000002.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ] + │ t0: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000000.parquet:.., /testdata/weather/result-000001.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible + │ t1: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000001.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible + │ t2: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000001.parquet:.., /testdata/weather/result-000002.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible └────────────────────────────────────────────────── ┌───── Stage 1 ── Tasks: t0:[p0..p8] t1:[p9..p17] t2:[p18..p26] │ BroadcastExec: input_partitions=3, consumer_tasks=3, output_partitions=9 @@ -756,7 +756,7 @@ mod tests { .broadcast_joins(true) .physical_plan_as_ascii(query, false) .await; - assert_snapshot!(physical_plan_ascii, @r" + assert_snapshot!(physical_plan_ascii, @" ┌───── DistributedExec ── Tasks: t0:[p0] │ CoalescePartitionsExec │ [Stage 3] => NetworkCoalesceExec: output_partitions=9, input_tasks=3 @@ -766,9 +766,9 @@ mod tests { │ CoalescePartitionsExec │ [Stage 2] => NetworkBroadcastExec: partitions_per_consumer=3, stage_partitions=9, input_tasks=3 │ DistributedLeafExec: - │ t0: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000000.parquet:.., /testdata/weather/result-000001.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[Rainfall, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ] - │ t1: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000001.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[Rainfall, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ] - │ t2: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000001.parquet:.., /testdata/weather/result-000002.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[Rainfall, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ] + │ t0: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000000.parquet:.., /testdata/weather/result-000001.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[Rainfall, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible + │ t1: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000001.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[Rainfall, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible + │ t2: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000001.parquet:.., /testdata/weather/result-000002.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[Rainfall, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible └────────────────────────────────────────────────── ┌───── Stage 2 ── Tasks: t0:[p0..p8] t1:[p9..p17] t2:[p18..p26] │ BroadcastExec: input_partitions=3, consumer_tasks=3, output_partitions=9 @@ -776,9 +776,9 @@ mod tests { │ CoalescePartitionsExec │ [Stage 1] => NetworkBroadcastExec: partitions_per_consumer=3, stage_partitions=9, input_tasks=3 │ DistributedLeafExec: - │ t0: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000000.parquet:.., /testdata/weather/result-000001.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ] - │ t1: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000001.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ] - │ t2: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000001.parquet:.., /testdata/weather/result-000002.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ] + │ t0: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000000.parquet:.., /testdata/weather/result-000001.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible + │ t1: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000001.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible + │ t2: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000001.parquet:.., /testdata/weather/result-000002.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible └────────────────────────────────────────────────── ┌───── Stage 1 ── Tasks: t0:[p0..p8] t1:[p9..p17] t2:[p18..p26] │ BroadcastExec: input_partitions=3, consumer_tasks=3, output_partitions=9 @@ -800,18 +800,18 @@ mod tests { let physical_plan_ascii = TestPlanBuilder::default() .physical_plan_as_ascii(query, false) .await; - assert_snapshot!(physical_plan_ascii, @r" + assert_snapshot!(physical_plan_ascii, @" HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(RainToday@1, RainToday@1)], projection=[MinTemp@0, MaxTemp@2] CoalescePartitionsExec DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet], [/testdata/weather/result-000001.parquet], [/testdata/weather/result-000002.parquet]]}, projection=[MinTemp, RainToday], file_type=parquet - DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet], [/testdata/weather/result-000001.parquet], [/testdata/weather/result-000002.parquet]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ] + DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet], [/testdata/weather/result-000001.parquet], [/testdata/weather/result-000002.parquet]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible "); let physical_plan_ascii = TestPlanBuilder::default() .broadcast_joins(true) .physical_plan_as_ascii(query, false) .await; - assert_snapshot!(physical_plan_ascii, @r" + assert_snapshot!(physical_plan_ascii, @" ┌───── DistributedExec ── Tasks: t0:[p0] │ CoalescePartitionsExec │ [Stage 2] => NetworkCoalesceExec: output_partitions=9, input_tasks=3 @@ -821,9 +821,9 @@ mod tests { │ CoalescePartitionsExec │ [Stage 1] => NetworkBroadcastExec: partitions_per_consumer=3, stage_partitions=9, input_tasks=3 │ DistributedLeafExec: - │ t0: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000000.parquet:.., /testdata/weather/result-000001.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ] - │ t1: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000001.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ] - │ t2: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000001.parquet:.., /testdata/weather/result-000002.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ] + │ t0: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000000.parquet:.., /testdata/weather/result-000001.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible + │ t1: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000001.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible + │ t2: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000001.parquet:.., /testdata/weather/result-000002.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible └────────────────────────────────────────────────── ┌───── Stage 1 ── Tasks: t0:[p0..p8] t1:[p9..p17] t2:[p18..p26] │ BroadcastExec: input_partitions=3, consumer_tasks=3, output_partitions=9 @@ -854,7 +854,7 @@ mod tests { .broadcast_joins(true) .physical_plan_as_ascii(query, false) .await; - assert_snapshot!(physical_plan_ascii, @r" + assert_snapshot!(physical_plan_ascii, @" ┌───── DistributedExec ── Tasks: t0:[p0] │ CoalescePartitionsExec │ [Stage 4] => NetworkCoalesceExec: output_partitions=9, input_tasks=3 @@ -865,17 +865,17 @@ mod tests { │ CoalescePartitionsExec │ [Stage 1] => NetworkBroadcastExec: partitions_per_consumer=3, stage_partitions=3, input_tasks=3 │ DistributedLeafExec: - │ t0: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000000.parquet:.., /testdata/weather/result-000001.parquet:.., /testdata/weather/result-000002.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ] + │ t0: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000000.parquet:.., /testdata/weather/result-000001.parquet:.., /testdata/weather/result-000002.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible │ HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(RainToday@1, RainToday@1)], projection=[MinTemp@0, MaxTemp@2] │ CoalescePartitionsExec │ [Stage 2] => NetworkBroadcastExec: partitions_per_consumer=3, stage_partitions=3, input_tasks=3 │ DistributedLeafExec: - │ t0: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000000.parquet:.., /testdata/weather/result-000001.parquet:.., /testdata/weather/result-000002.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ] + │ t0: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000000.parquet:.., /testdata/weather/result-000001.parquet:.., /testdata/weather/result-000002.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible │ HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(RainToday@1, RainToday@1)], projection=[MinTemp@0, MaxTemp@2] │ CoalescePartitionsExec │ [Stage 3] => NetworkBroadcastExec: partitions_per_consumer=3, stage_partitions=3, input_tasks=3 │ DistributedLeafExec: - │ t0: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000000.parquet:.., /testdata/weather/result-000001.parquet:.., /testdata/weather/result-000002.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ] + │ t0: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000000.parquet:.., /testdata/weather/result-000001.parquet:.., /testdata/weather/result-000002.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible └────────────────────────────────────────────────── ┌───── Stage 1 ── Tasks: t0:[p0..p2] t1:[p3..p5] t2:[p6..p8] │ BroadcastExec: input_partitions=3, consumer_tasks=1, output_partitions=3 @@ -913,7 +913,7 @@ mod tests { .distributed_task_estimator(BuildSideOneTaskEstimator) .physical_plan_as_ascii(query, false) .await; - assert_snapshot!(physical_plan_ascii, @r" + assert_snapshot!(physical_plan_ascii, @" ┌───── DistributedExec ── Tasks: t0:[p0] │ CoalescePartitionsExec │ [Stage 2] => NetworkCoalesceExec: output_partitions=9, input_tasks=3 @@ -923,9 +923,9 @@ mod tests { │ CoalescePartitionsExec │ [Stage 1] => NetworkBroadcastExec: partitions_per_consumer=3, stage_partitions=9, input_tasks=1 │ DistributedLeafExec: - │ t0: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000000.parquet:.., /testdata/weather/result-000001.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ] - │ t1: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000001.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ] - │ t2: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000001.parquet:.., /testdata/weather/result-000002.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ] + │ t0: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000000.parquet:.., /testdata/weather/result-000001.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible + │ t1: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000001.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible + │ t2: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000001.parquet:.., /testdata/weather/result-000002.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible └────────────────────────────────────────────────── ┌───── Stage 1 ── Tasks: t0:[p0..p8] │ BroadcastExec: input_partitions=3, consumer_tasks=3, output_partitions=9 diff --git a/src/distributed_planner/inject_network_boundaries.rs b/src/distributed_planner/inject_network_boundaries.rs index 37d5ee95..93a3ac2d 100644 --- a/src/distributed_planner/inject_network_boundaries.rs +++ b/src/distributed_planner/inject_network_boundaries.rs @@ -1009,10 +1009,10 @@ mod tests { .await .physical_plan_as_string(query) .await; - assert_snapshot!(physical_plan_string, @r" + assert_snapshot!(physical_plan_string, @" HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(RainToday@1, RainToday@1)], projection=[MinTemp@0, MaxTemp@2] DataSourceExec: file_groups={1 group: [[/testdata/weather/result-000000.parquet, /testdata/weather/result-000001.parquet, /testdata/weather/result-000002.parquet]]}, projection=[MinTemp, RainToday], file_type=parquet - DataSourceExec: file_groups={1 group: [[/testdata/weather/result-000000.parquet, /testdata/weather/result-000001.parquet, /testdata/weather/result-000002.parquet]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ] + DataSourceExec: file_groups={1 group: [[/testdata/weather/result-000000.parquet, /testdata/weather/result-000001.parquet, /testdata/weather/result-000002.parquet]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible "); // With target_partitions=1, there is no CoalescePartitionsExec initially diff --git a/src/distributed_planner/insert_broadcast.rs b/src/distributed_planner/insert_broadcast.rs index 33381936..9fe63d27 100644 --- a/src/distributed_planner/insert_broadcast.rs +++ b/src/distributed_planner/insert_broadcast.rs @@ -194,19 +194,19 @@ mod tests { .num_workers(4) .physical_plan_as_string(query) .await; - assert_snapshot!(physical_plan_string, @r" + assert_snapshot!(physical_plan_string, @" HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(RainToday@1, RainToday@1)], projection=[MinTemp@0, MaxTemp@2] CoalescePartitionsExec DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet], [/testdata/weather/result-000001.parquet], [/testdata/weather/result-000002.parquet]]}, projection=[MinTemp, RainToday], file_type=parquet - DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet], [/testdata/weather/result-000001.parquet], [/testdata/weather/result-000002.parquet]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ] + DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet], [/testdata/weather/result-000001.parquet], [/testdata/weather/result-000002.parquet]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible "); let plan = sql_to_plan_with_broadcast(query, true, 4).await; - assert_snapshot!(plan, @r" + assert_snapshot!(plan, @" HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(RainToday@1, RainToday@1)], projection=[MinTemp@0, MaxTemp@2] CoalescePartitionsExec BroadcastExec: input_partitions=3, consumer_tasks=1, output_partitions=3 DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet], [/testdata/weather/result-000001.parquet], [/testdata/weather/result-000002.parquet]]}, projection=[MinTemp, RainToday], file_type=parquet - DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet], [/testdata/weather/result-000001.parquet], [/testdata/weather/result-000002.parquet]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ] + DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet], [/testdata/weather/result-000001.parquet], [/testdata/weather/result-000002.parquet]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible "); } @@ -222,18 +222,18 @@ mod tests { .num_workers(4) .physical_plan_as_string(query) .await; - assert_snapshot!(physical_plan_string, @r" + assert_snapshot!(physical_plan_string, @" HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(RainToday@1, RainToday@1)], projection=[MinTemp@0, MaxTemp@2] DataSourceExec: file_groups={1 group: [[/testdata/weather/result-000000.parquet, /testdata/weather/result-000001.parquet, /testdata/weather/result-000002.parquet]]}, projection=[MinTemp, RainToday], file_type=parquet - DataSourceExec: file_groups={1 group: [[/testdata/weather/result-000000.parquet, /testdata/weather/result-000001.parquet, /testdata/weather/result-000002.parquet]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ] + DataSourceExec: file_groups={1 group: [[/testdata/weather/result-000000.parquet, /testdata/weather/result-000001.parquet, /testdata/weather/result-000002.parquet]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible "); let plan = sql_to_plan_with_broadcast(query, true, 1).await; - assert_snapshot!(plan, @r" + assert_snapshot!(plan, @" HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(RainToday@1, RainToday@1)], projection=[MinTemp@0, MaxTemp@2] CoalescePartitionsExec BroadcastExec: input_partitions=1, consumer_tasks=1, output_partitions=1 DataSourceExec: file_groups={1 group: [[/testdata/weather/result-000000.parquet, /testdata/weather/result-000001.parquet, /testdata/weather/result-000002.parquet]]}, projection=[MinTemp, RainToday], file_type=parquet - DataSourceExec: file_groups={1 group: [[/testdata/weather/result-000000.parquet, /testdata/weather/result-000001.parquet, /testdata/weather/result-000002.parquet]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ] + DataSourceExec: file_groups={1 group: [[/testdata/weather/result-000000.parquet, /testdata/weather/result-000001.parquet, /testdata/weather/result-000002.parquet]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible "); } @@ -245,11 +245,11 @@ mod tests { ON a."RainToday" = b."RainToday" "#; let plan = sql_to_plan_with_broadcast(query, true, 4).await; - assert_snapshot!(plan, @r" + assert_snapshot!(plan, @" HashJoinExec: mode=CollectLeft, join_type=Left, on=[(RainToday@1, RainToday@1)], projection=[MinTemp@0, MaxTemp@2] CoalescePartitionsExec DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet], [/testdata/weather/result-000001.parquet], [/testdata/weather/result-000002.parquet]]}, projection=[MinTemp, RainToday], file_type=parquet - DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet], [/testdata/weather/result-000001.parquet], [/testdata/weather/result-000002.parquet]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ] + DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet], [/testdata/weather/result-000001.parquet], [/testdata/weather/result-000002.parquet]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible "); } @@ -261,11 +261,11 @@ mod tests { ON a."RainToday" = b."RainToday" "#; let plan = sql_to_plan_with_broadcast(query, false, 4).await; - assert_snapshot!(plan, @r" + assert_snapshot!(plan, @" HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(RainToday@1, RainToday@1)], projection=[MinTemp@0, MaxTemp@2] CoalescePartitionsExec DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet], [/testdata/weather/result-000001.parquet], [/testdata/weather/result-000002.parquet]]}, projection=[MinTemp, RainToday], file_type=parquet - DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet], [/testdata/weather/result-000001.parquet], [/testdata/weather/result-000002.parquet]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ] + DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet], [/testdata/weather/result-000001.parquet], [/testdata/weather/result-000002.parquet]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible "); } diff --git a/src/distributed_planner/statistics/complexity_cpu.rs b/src/distributed_planner/statistics/complexity_cpu.rs index 49ed6e77..d8ac3114 100644 --- a/src/distributed_planner/statistics/complexity_cpu.rs +++ b/src/distributed_planner/statistics/complexity_cpu.rs @@ -216,19 +216,24 @@ pub(super) fn complexity_cpu(node: &Arc) -> Complexity { return n.unwrap_or(Complexity::Constant(1.)); } - // RepartitionExec with Hash: computes hash per row + take_arrays + // RepartitionExec copies rows into output partitions and may evaluate repartition keys. // https://github.com/apache/datafusion/blob/branch-52/datafusion/physical-plan/src/repartition/mod.rs if let Some(node) = node.downcast_ref::() { // It needs to copy all the data for chunking it to the different output partitions... let mut n = Complexity::Linear(LinearComplexity::AllColumns); - // And it might need to compute a hash per row based on the provided expressions; hashing a - // plain column key still costs its bytes. + // Hash/range partitioning reads the partition keys. A plain column key still costs its + // bytes because the operator hashes or compares it. match node.partitioning() { Partitioning::Hash(expressions, _) => { for expr in expressions { n = n.plus(hashed_or_sorted_key_complexity(expr)) } } + Partitioning::Range(range) => { + for expr in range.ordering() { + n = n.plus(hashed_or_sorted_key_complexity(&expr.expr)) + } + } Partitioning::RoundRobinBatch(_) => {} Partitioning::UnknownPartitioning(_) => {} }; @@ -429,8 +434,14 @@ mod tests { use crate::assert_snapshot; use crate::distributed_planner::statistics::complexity_cpu::complexity_cpu; use crate::test_utils::plans::TestPlanBuilder; + use arrow::datatypes::{DataType, Field, Schema}; use datafusion::common::tree_node::{Transformed, TreeNode}; - use datafusion::physical_plan::{ExecutionPlan, displayable}; + use datafusion::common::{ScalarValue, SplitPoint}; + use datafusion::physical_expr::{LexOrdering, PhysicalSortExpr, RangePartitioning}; + use datafusion::physical_plan::empty::EmptyExec; + use datafusion::physical_plan::expressions::Column; + use datafusion::physical_plan::repartition::RepartitionExec; + use datafusion::physical_plan::{ExecutionPlan, Partitioning, displayable}; use std::cell::RefCell; use std::sync::Arc; /* schema for the "weather" table @@ -771,6 +782,34 @@ mod tests { "); } + #[test] + fn range_repartition_exec() { + let schema = Arc::new(Schema::new(vec![Field::new( + "range_key", + DataType::Int64, + false, + )])); + let input = Arc::new(EmptyExec::new(schema).with_partitions(2)); + let ordering = LexOrdering::new([PhysicalSortExpr::new_default(Arc::new(Column::new( + "range_key", + 0, + )))]) + .expect("non-empty ordering"); + let partitioning = Partitioning::Range( + RangePartitioning::try_new( + ordering, + vec![SplitPoint::new(vec![ScalarValue::Int64(Some(10))])], + ) + .expect("valid range partitioning"), + ); + let plan = Arc::new(RepartitionExec::try_new(input, partitioning).unwrap()); + + assert_snapshot!(plan_costs(plan), @r" + O((Cols+Col0)) | RepartitionExec: partitioning=Range([range_key@0 ASC], [(10)], 2), input_partitions=2 + O(1) | EmptyExec + "); + } + // HashJoinExec in Partitioned mode: a distinct planner path from CollectLeft. The cost formula // is the same (build 2x read + key hash, probe read + key hash), now over hash-repartitioned // inputs rather than a collected left side. diff --git a/src/distributed_planner/statistics/plan_statistics.rs b/src/distributed_planner/statistics/plan_statistics.rs index d3d54b8e..9f0e363b 100644 --- a/src/distributed_planner/statistics/plan_statistics.rs +++ b/src/distributed_planner/statistics/plan_statistics.rs @@ -5,6 +5,7 @@ use datafusion::config::ConfigOptions; use datafusion::error::Result; use datafusion::execution::{SendableRecordBatchStream, TaskContext}; use datafusion::physical_plan::execution_plan::CardinalityEffect; +use datafusion::physical_plan::statistics::StatisticsArgs; use datafusion::physical_plan::{DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties}; use delegate::delegate; use itertools::Itertools; @@ -102,7 +103,7 @@ fn partition_statistics_with_children_override( let stats = Arc::clone(node) .with_new_children(statistics_wrapped_children)? - .partition_statistics(partition)?; + .statistics_with_args(&StatisticsArgs::new().with_partition(partition))?; Ok(stats.as_ref().clone()) } @@ -122,8 +123,8 @@ impl DisplayAs for StatisticsWrapper { } impl ExecutionPlan for StatisticsWrapper { - fn partition_statistics(&self, partition: Option) -> Result> { - if partition.is_some() { + fn statistics_with_args(&self, args: &StatisticsArgs) -> Result> { + if args.partition().is_some() { return plan_err!("StatisticsWrapper not prepared for partition-specific stats"); } Ok(Arc::clone(&self.stats)) diff --git a/src/execution_plans/common.rs b/src/execution_plans/common.rs index 8ea0e4af..8af615d9 100644 --- a/src/execution_plans/common.rs +++ b/src/execution_plans/common.rs @@ -22,5 +22,6 @@ pub(super) fn scale_partitioning( Partitioning::RoundRobinBatch(p) => Partitioning::RoundRobinBatch(f(*p)), Partitioning::Hash(hash, p) => Partitioning::Hash(hash.clone(), f(*p)), Partitioning::UnknownPartitioning(p) => Partitioning::UnknownPartitioning(f(*p)), + Partitioning::Range(range) => Partitioning::Range(range.clone()), } } diff --git a/src/execution_plans/distributed_leaf.rs b/src/execution_plans/distributed_leaf.rs index b9ab4d35..2d8eb6f1 100644 --- a/src/execution_plans/distributed_leaf.rs +++ b/src/execution_plans/distributed_leaf.rs @@ -2,6 +2,7 @@ use crate::DistributedTaskContext; use datafusion::common::{Result, Statistics, exec_err, not_impl_err, plan_err}; use datafusion::execution::{SendableRecordBatchStream, TaskContext}; use datafusion::physical_expr_common::metrics::MetricsSet; +use datafusion::physical_plan::statistics::StatisticsArgs; use datafusion::physical_plan::{DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties}; use std::fmt::Formatter; use std::sync::Arc; @@ -185,6 +186,10 @@ impl ExecutionPlan for DistributedLeafExec { } fn partition_statistics(&self, partition: Option) -> Result> { - self.original.partition_statistics(partition) + self.statistics_with_args(&StatisticsArgs::new().with_partition(partition)) + } + + fn statistics_with_args(&self, args: &StatisticsArgs) -> Result> { + self.original.statistics_with_args(args) } } diff --git a/src/filter_expressions.rs b/src/filter_expressions.rs new file mode 100644 index 00000000..b0b3bccf --- /dev/null +++ b/src/filter_expressions.rs @@ -0,0 +1,235 @@ +use crate::config_extension_ext::set_distributed_option_extension; +use crate::{DistributedConfig, DistributedTaskContext}; +use datafusion::arrow::datatypes::SchemaRef; +use datafusion::catalog::memory::DataSourceExec; +use datafusion::common::tree_node::TreeNode; +use datafusion::common::{HashMap, HashSet, Result, internal_datafusion_err}; +use datafusion::datasource::physical_plan::FileScanConfig; +use datafusion::physical_expr::PhysicalExpr; +use datafusion::physical_expr::expressions::DynamicFilterPhysicalExpr; +use datafusion::physical_plan::ExecutionPlan; +use datafusion::physical_plan::aggregates::AggregateExec; +use datafusion::physical_plan::filter::FilterExec; +use datafusion::physical_plan::joins::HashJoinExec; +use datafusion::physical_plan::sorts::sort::SortExec; +use datafusion::prelude::SessionConfig; +use std::any::Any; +use std::fmt::{Debug, Formatter}; +use std::marker::PhantomData; +use std::sync::Arc; + +/// A filter expression and the schema required to decode its serialized form. +#[derive(Clone, Debug)] +pub struct FilterExpression { + pub expr: Arc, + pub input_schema: SchemaRef, +} + +impl FilterExpression { + pub fn new(expr: Arc, input_schema: SchemaRef) -> Self { + Self { expr, input_schema } + } +} + +/// Extension trait for plan nodes that expose filter expressions. +pub trait PlanFilterExpressions: ExecutionPlan { + fn get_filter_expressions(&self) -> Vec; +} + +trait PlanFilterExpressionExtractor: Send + Sync { + fn get_filter_expressions( + &self, + plan: &Arc, + dt_ctx: DistributedTaskContext, + ) -> Vec; +} + +struct TypedExtractor(PhantomData); + +impl PlanFilterExpressionExtractor for TypedExtractor +where + T: PlanFilterExpressions + 'static, +{ + fn get_filter_expressions( + &self, + plan: &Arc, + _dt_ctx: DistributedTaskContext, + ) -> Vec { + plan.downcast_ref::() + .map(PlanFilterExpressions::get_filter_expressions) + .unwrap_or_default() + } +} + +/// Registry of plan node types that can expose filter expressions. +#[derive(Clone)] +pub struct PlanFilterExpressionRegistry { + extractors: Vec>, +} + +impl Default for PlanFilterExpressionRegistry { + fn default() -> Self { + let mut registry = Self { + extractors: Vec::new(), + }; + registry.register::(); + registry.register::(); + registry.register::(); + registry.register::(); + registry.register::(); + registry + } +} + +impl Debug for PlanFilterExpressionRegistry { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "PlanFilterExpressionRegistry") + } +} + +impl PlanFilterExpressionRegistry { + pub fn register(&mut self) + where + T: PlanFilterExpressions + 'static, + { + self.extractors + .push(Arc::new(TypedExtractor::(PhantomData))); + } + + pub fn get_filter_expressions( + &self, + plan: &Arc, + dt_ctx: DistributedTaskContext, + ) -> Vec { + let mut out = Vec::new(); + for extractor in &self.extractors { + out.extend(extractor.get_filter_expressions(plan, dt_ctx)); + } + out + } +} + +impl PlanFilterExpressions for FilterExec { + fn get_filter_expressions(&self) -> Vec { + vec![FilterExpression::new( + Arc::clone(self.predicate()), + self.input().schema(), + )] + } +} + +impl PlanFilterExpressions for HashJoinExec { + fn get_filter_expressions(&self) -> Vec { + self.dynamic_filter_expr() + .map(|expr| { + let expr: Arc = expr.clone(); + FilterExpression::new(expr, self.right().schema()) + }) + .into_iter() + .collect() + } +} + +impl PlanFilterExpressions for AggregateExec { + fn get_filter_expressions(&self) -> Vec { + self.dynamic_filter_expr() + .map(|expr| { + let expr: Arc = expr.clone(); + FilterExpression::new(expr, self.input().schema()) + }) + .into_iter() + .collect() + } +} + +impl PlanFilterExpressions for SortExec { + fn get_filter_expressions(&self) -> Vec { + self.dynamic_filter_expr() + .map(|expr| { + let expr: Arc = expr; + FilterExpression::new(expr, self.input().schema()) + }) + .into_iter() + .collect() + } +} + +impl PlanFilterExpressions for DataSourceExec { + fn get_filter_expressions(&self) -> Vec { + let Some(file_scan) = self.data_source().downcast_ref::() else { + return Vec::new(); + }; + let table_schema = Arc::clone(file_scan.file_source().table_schema().table_schema()); + file_scan + .file_source() + .filter() + .map(|expr| FilterExpression::new(expr, table_schema)) + .into_iter() + .collect() + } +} + +pub(crate) fn collect_dynamic_filters( + filters: impl IntoIterator, +) -> Result>> { + let mut seen = HashSet::new(); + let mut out = Vec::new(); + + for filter in filters { + filter.expr.apply(|expr| { + if expr.downcast_ref::().is_some() + && let Some(id) = expr.expression_id() + && seen.insert(id) + { + out.push(downcast_dynamic_filter(Arc::clone(expr))?); + } + Ok(datafusion::common::tree_node::TreeNodeRecursion::Continue) + })?; + } + + Ok(out) +} + +pub(crate) fn collect_dynamic_filters_by_id( + filters: impl IntoIterator, +) -> Result>> { + let mut out = HashMap::new(); + for filter in collect_dynamic_filters(filters)? { + let Some(id) = filter.expression_id() else { + continue; + }; + out.insert(id, filter); + } + Ok(out) +} + +pub(crate) fn downcast_dynamic_filter( + expr: Arc, +) -> Result> { + let expr = expr as Arc; + expr.downcast::().map_err(|_| { + internal_datafusion_err!("PhysicalExpr did not downcast to DynamicFilterPhysicalExpr") + }) +} + +pub(crate) fn set_distributed_filter_expressions(cfg: &mut SessionConfig) +where + T: PlanFilterExpressions + 'static, +{ + let opts = cfg.options_mut(); + if let Some(distributed_cfg) = opts.extensions.get_mut::() { + distributed_cfg + .__private_filter_expression_registry + .register::(); + } else { + let mut registry = PlanFilterExpressionRegistry::default(); + registry.register::(); + set_distributed_option_extension( + cfg, + DistributedConfig { + __private_filter_expression_registry: registry, + ..Default::default() + }, + ) + } +} diff --git a/src/lib.rs b/src/lib.rs index d2a3a463..61891b43 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -4,6 +4,7 @@ mod common; mod config_extension_ext; mod distributed_ext; mod execution_plans; +mod filter_expressions; mod metrics; mod passthrough_headers; mod stage; @@ -30,6 +31,9 @@ pub use execution_plans::{ BroadcastExec, DistributedLeafExec, NetworkBroadcastExec, NetworkCoalesceExec, NetworkShuffleExec, }; +pub use filter_expressions::{ + FilterExpression, PlanFilterExpressionRegistry, PlanFilterExpressions, +}; pub use metrics::{ AvgLatencyMetric, BytesCounterMetric, BytesMetricExt, DISTRIBUTED_DATAFUSION_TASK_ID_LABEL, DistributedMetricsFormat, FirstLatencyMetric, GaugeMetricExt, LatencyMetricExt, MaxGaugeMetric, diff --git a/src/metrics/proto.rs b/src/metrics/proto.rs index c114a41c..1dcfd5d2 100644 --- a/src/metrics/proto.rs +++ b/src/metrics/proto.rs @@ -123,6 +123,14 @@ pub fn df_metric_to_proto(metric: Arc) -> Result Ok(pb::Metric { + value: Some(pb::metric::Value::Gauge(pb::NamedGauge { + name: name.to_string(), + value: gauge.value() as u64 + })), + partition, + labels, + }), MetricValue::Time { name, time } => Ok(pb::Metric { value: Some(pb::metric::Value::Time(pb::NamedTime { name: name.to_string(), diff --git a/src/protobuf/distributed_codec.rs b/src/protobuf/distributed_codec.rs index 3cfec30a..8bf8a7f9 100644 --- a/src/protobuf/distributed_codec.rs +++ b/src/protobuf/distributed_codec.rs @@ -22,8 +22,8 @@ use datafusion::prelude::SessionConfig; use datafusion_proto::physical_plan::from_proto::parse_protobuf_partitioning; use datafusion_proto::physical_plan::to_proto::serialize_partitioning; use datafusion_proto::physical_plan::{ - ComposedPhysicalExtensionCodec, DefaultPhysicalProtoConverter, PhysicalExtensionCodec, - PhysicalPlanDecodeContext, + ComposedPhysicalExtensionCodec, PhysicalExtensionCodec, PhysicalPlanDecodeContext, + PhysicalProtoConverterExtension, }; use datafusion_proto::protobuf; use datafusion_proto::protobuf::proto_error; @@ -51,6 +51,7 @@ impl PhysicalExtensionCodec for DistributedCodec { buf: &[u8], inputs: &[Arc], ctx: &TaskContext, + proto_converter: &dyn PhysicalProtoConverterExtension, ) -> datafusion::common::Result> { let DistributedExecProto { node: Some(distributed_exec_node), @@ -105,12 +106,12 @@ impl PhysicalExtensionCodec for DistributedCodec { .map(|s| s.try_into()) .ok_or(proto_error("NetworkShuffleExec is missing schema"))??; - let decode_ctx = PhysicalPlanDecodeContext::new(ctx, &DistributedCodec {}); + let decode_ctx = PhysicalPlanDecodeContext::new(ctx, self); let partitioning = parse_protobuf_partitioning( partitioning.as_ref(), &decode_ctx, &schema, - &DefaultPhysicalProtoConverter {}, + proto_converter, )? .ok_or(proto_error("NetworkShuffleExec is missing partitioning"))?; @@ -130,12 +131,12 @@ impl PhysicalExtensionCodec for DistributedCodec { .map(|s| s.try_into()) .ok_or(proto_error("NetworkCoalesceExec is missing schema"))??; - let decode_ctx = PhysicalPlanDecodeContext::new(ctx, &DistributedCodec {}); + let decode_ctx = PhysicalPlanDecodeContext::new(ctx, self); let partitioning = parse_protobuf_partitioning( partitioning.as_ref(), &decode_ctx, &schema, - &DefaultPhysicalProtoConverter {}, + proto_converter, )? .ok_or(proto_error("NetworkCoalesceExec is missing partitioning"))?; @@ -155,12 +156,12 @@ impl PhysicalExtensionCodec for DistributedCodec { .map(|s| s.try_into()) .ok_or(proto_error("NetworkBroadcastExec is missing schema"))??; - let decode_ctx = PhysicalPlanDecodeContext::new(ctx, &DistributedCodec {}); + let decode_ctx = PhysicalPlanDecodeContext::new(ctx, self); let partitioning = parse_protobuf_partitioning( partitioning.as_ref(), &decode_ctx, &schema, - &DefaultPhysicalProtoConverter {}, + proto_converter, )? .ok_or(proto_error("NetworkBroadcastExec is missing partitioning"))?; @@ -236,7 +237,12 @@ impl PhysicalExtensionCodec for DistributedCodec { } } - fn try_encode(&self, node: Arc, buf: &mut Vec) -> Result<()> { + fn try_encode( + &self, + node: Arc, + buf: &mut Vec, + proto_converter: &dyn PhysicalProtoConverterExtension, + ) -> Result<()> { fn encode_stage_proto(stage: &Stage) -> Result { Ok(match stage { Stage::Local(local) => StageProto { @@ -266,7 +272,7 @@ impl PhysicalExtensionCodec for DistributedCodec { partitioning: Some(serialize_partitioning( node.properties().output_partitioning(), &DistributedCodec {}, - &DefaultPhysicalProtoConverter {}, + proto_converter, )?), input_stage: Some(encode_stage_proto(node.input_stage())?), }; @@ -282,7 +288,7 @@ impl PhysicalExtensionCodec for DistributedCodec { partitioning: Some(serialize_partitioning( node.properties().output_partitioning(), &DistributedCodec {}, - &DefaultPhysicalProtoConverter {}, + proto_converter, )?), input_stage: Some(encode_stage_proto(node.input_stage())?), }; @@ -298,7 +304,7 @@ impl PhysicalExtensionCodec for DistributedCodec { partitioning: Some(serialize_partitioning( node.properties().output_partitioning(), &DistributedCodec {}, - &DefaultPhysicalProtoConverter {}, + proto_converter, )?), input_stage: Some(encode_stage_proto(node.input_stage())?), }; @@ -537,6 +543,11 @@ mod tests { physical_expr::{Partitioning, PhysicalSortExpr, expressions::Column, expressions::col}, physical_plan::{ExecutionPlan, displayable, sorts::sort::SortExec, union::UnionExec}, }; + use datafusion_proto::physical_plan::DefaultPhysicalProtoConverter; + + fn default_proto_converter() -> DefaultPhysicalProtoConverter { + DefaultPhysicalProtoConverter {} + } fn empty_exec() -> Arc { Arc::new(EmptyExec::new(SchemaRef::new(Schema::empty()))) @@ -582,9 +593,9 @@ mod tests { Arc::new(new_network_hash_shuffle_exec(part, schema, dummy_stage())); let mut buf = Vec::new(); - codec.try_encode(plan.clone(), &mut buf)?; + codec.try_encode(plan.clone(), &mut buf, &default_proto_converter())?; - let decoded = codec.try_decode(&buf, &[], &ctx)?; + let decoded = codec.try_decode(&buf, &[], &ctx, &default_proto_converter())?; assert_eq!(repr(&plan), repr(&decoded)); Ok(()) @@ -612,9 +623,9 @@ mod tests { Arc::new(NetworkCoalesceExec::try_new(union.clone(), 1, 1)?); let mut buf = Vec::new(); - codec.try_encode(plan.clone(), &mut buf)?; + codec.try_encode(plan.clone(), &mut buf, &default_proto_converter())?; - let decoded = codec.try_decode(&buf, &[union], &ctx)?; + let decoded = codec.try_decode(&buf, &[union], &ctx, &default_proto_converter())?; assert_eq!(repr(&plan), repr(&decoded)); Ok(()) @@ -645,9 +656,9 @@ mod tests { Arc::new(NetworkCoalesceExec::try_new(sort.clone(), 1, 1)?); let mut buf = Vec::new(); - codec.try_encode(plan.clone(), &mut buf)?; + codec.try_encode(plan.clone(), &mut buf, &default_proto_converter())?; - let decoded = codec.try_decode(&buf, &[sort], &ctx)?; + let decoded = codec.try_decode(&buf, &[sort], &ctx, &default_proto_converter())?; assert_eq!(repr(&plan), repr(&decoded)); Ok(()) @@ -666,9 +677,9 @@ mod tests { )); let mut buf = Vec::new(); - codec.try_encode(plan.clone(), &mut buf)?; + codec.try_encode(plan.clone(), &mut buf, &default_proto_converter())?; - let decoded = codec.try_decode(&buf, &[], &ctx)?; + let decoded = codec.try_decode(&buf, &[], &ctx, &default_proto_converter())?; assert_eq!(repr(&plan), repr(&decoded)); Ok(()) @@ -688,9 +699,9 @@ mod tests { )); let mut buf = Vec::new(); - codec.try_encode(plan.clone(), &mut buf)?; + codec.try_encode(plan.clone(), &mut buf, &default_proto_converter())?; - let decoded = codec.try_decode(&buf, &[empty_exec()], &ctx)?; + let decoded = codec.try_decode(&buf, &[empty_exec()], &ctx, &default_proto_converter())?; assert_eq!(repr(&plan), repr(&decoded)); Ok(()) @@ -709,9 +720,9 @@ mod tests { )); let mut buf = Vec::new(); - codec.try_encode(plan.clone(), &mut buf)?; + codec.try_encode(plan.clone(), &mut buf, &default_proto_converter())?; - let decoded = codec.try_decode(&buf, &[empty_exec()], &ctx)?; + let decoded = codec.try_decode(&buf, &[empty_exec()], &ctx, &default_proto_converter())?; assert_eq!(repr(&plan), repr(&decoded)); Ok(()) @@ -733,9 +744,9 @@ mod tests { Arc::new(NetworkCoalesceExec::try_new(flight.clone(), 1, 1)?); let mut buf = Vec::new(); - codec.try_encode(plan.clone(), &mut buf)?; + codec.try_encode(plan.clone(), &mut buf, &default_proto_converter())?; - let decoded = codec.try_decode(&buf, &[flight], &ctx)?; + let decoded = codec.try_decode(&buf, &[flight], &ctx, &default_proto_converter())?; assert_eq!(repr(&plan), repr(&decoded)); Ok(()) @@ -763,9 +774,9 @@ mod tests { Arc::new(NetworkCoalesceExec::try_new(union.clone(), 1, 1)?); let mut buf = Vec::new(); - codec.try_encode(plan.clone(), &mut buf)?; + codec.try_encode(plan.clone(), &mut buf, &default_proto_converter())?; - let decoded = codec.try_decode(&buf, &[union], &ctx)?; + let decoded = codec.try_decode(&buf, &[union], &ctx, &default_proto_converter())?; assert_eq!(repr(&plan), repr(&decoded)); Ok(()) @@ -796,9 +807,9 @@ mod tests { )?); let mut buf = Vec::new(); - codec.try_encode(plan.clone(), &mut buf)?; + codec.try_encode(plan.clone(), &mut buf, &default_proto_converter())?; - let decoded = codec.try_decode(&buf, &[left, right], &ctx)?; + let decoded = codec.try_decode(&buf, &[left, right], &ctx, &default_proto_converter())?; assert_eq!(repr(&plan), repr(&decoded)); Ok(()) diff --git a/src/test_utils/routing.rs b/src/test_utils/routing.rs index 26dc6c10..af8339c7 100644 --- a/src/test_utils/routing.rs +++ b/src/test_utils/routing.rs @@ -15,7 +15,10 @@ use datafusion::{ }, prelude::Expr, }; -use datafusion_proto::{physical_plan::PhysicalExtensionCodec, protobuf::proto_error}; +use datafusion_proto::{ + physical_plan::{PhysicalExtensionCodec, PhysicalProtoConverterExtension}, + protobuf::proto_error, +}; use futures::stream; use prost::Message; use std::{fmt::Formatter, sync::Arc}; @@ -324,6 +327,7 @@ impl PhysicalExtensionCodec for URLEmitterExtensionCodec { buf: &[u8], inputs: &[Arc], _ctx: &TaskContext, + _proto_converter: &dyn PhysicalProtoConverterExtension, ) -> Result> { if !inputs.is_empty() { return internal_err!( @@ -352,7 +356,12 @@ impl PhysicalExtensionCodec for URLEmitterExtensionCodec { )) } - fn try_encode(&self, node: Arc, buf: &mut Vec) -> Result<()> { + fn try_encode( + &self, + node: Arc, + buf: &mut Vec, + _proto_converter: &dyn PhysicalProtoConverterExtension, + ) -> Result<()> { let Some(exec) = node.downcast_ref::() else { return internal_err!("Expected URLEmitterExec, but was {}", node.name()); }; diff --git a/src/test_utils/test_work_unit_feed.rs b/src/test_utils/test_work_unit_feed.rs index ce82e5c3..88d48cdb 100644 --- a/src/test_utils/test_work_unit_feed.rs +++ b/src/test_utils/test_work_unit_feed.rs @@ -21,7 +21,7 @@ use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType}; use datafusion::physical_plan::metrics::{Count, ExecutionPlanMetricsSet, MetricBuilder}; use datafusion::physical_plan::stream::RecordBatchStreamAdapter; use datafusion::physical_plan::{DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties}; -use datafusion_proto::physical_plan::PhysicalExtensionCodec; +use datafusion_proto::physical_plan::{PhysicalExtensionCodec, PhysicalProtoConverterExtension}; use datafusion_proto::protobuf::proto_error; use futures::StreamExt; use futures::stream::BoxStream; @@ -502,6 +502,7 @@ impl PhysicalExtensionCodec for TestWorkUnitFeedExecCodec { buf: &[u8], inputs: &[Arc], _ctx: &TaskContext, + _proto_converter: &dyn PhysicalProtoConverterExtension, ) -> Result> { if !inputs.is_empty() { return internal_err!( @@ -530,7 +531,12 @@ impl PhysicalExtensionCodec for TestWorkUnitFeedExecCodec { ))) } - fn try_encode(&self, node: Arc, buf: &mut Vec) -> Result<()> { + fn try_encode( + &self, + node: Arc, + buf: &mut Vec, + _proto_converter: &dyn PhysicalProtoConverterExtension, + ) -> Result<()> { let Some(exec) = node.downcast_ref::() else { return internal_err!("Expected RowGeneratorExec, but was {}", node.name()); }; diff --git a/src/test_utils/work_unit_file_scan.rs b/src/test_utils/work_unit_file_scan.rs index df3d8de3..1336bbce 100644 --- a/src/test_utils/work_unit_file_scan.rs +++ b/src/test_utils/work_unit_file_scan.rs @@ -21,8 +21,10 @@ use datafusion::physical_plan::execution_plan::SchedulingType; use datafusion::physical_plan::metrics::ExecutionPlanMetricsSet; use datafusion::physical_plan::stream::RecordBatchStreamAdapter; use datafusion::physical_plan::{DisplayFormatType, ExecutionPlan, Partitioning}; +use datafusion_proto::TryFromProto; use datafusion_proto::physical_plan::{ - AsExecutionPlan, DefaultPhysicalExtensionCodec, PhysicalExtensionCodec, + DefaultPhysicalExtensionCodec, PhysicalExtensionCodec, PhysicalPlanNodeExt, + PhysicalProtoConverterExtension, }; use datafusion_proto::protobuf as df_proto; use datafusion_proto::protobuf::proto_error; @@ -73,8 +75,7 @@ impl WorkUnitFeedProvider for FileScanWorkUnitProvider { return Ok(futures::stream::empty().boxed()); }; let stream = futures::stream::iter(file_group.files().to_vec()).map(|file| { - let file_proto: df_proto::PartitionedFile = (&file) - .try_into() + let file_proto = df_proto::PartitionedFile::try_from_proto(&file) .map_err(|e| internal_datafusion_err!("{e}"))?; Ok(FileScanWorkUnit { file: Some(file_proto), @@ -129,7 +130,7 @@ impl DataSource for WorkUnitFileScanConfig { let file = work_unit?.file.expect("missing file"); let single_file_group = df_proto::FileGroup { files: vec![file] }; - let single_file_group = FileGroup::try_from(&single_file_group)?; + let single_file_group = FileGroup::try_from_proto(&single_file_group)?; let new_config = FileScanConfigBuilder::from(inner.clone()) .with_file_groups(vec![single_file_group]) @@ -237,6 +238,7 @@ impl PhysicalExtensionCodec for WorkUnitFileScanCodec { buf: &[u8], inputs: &[Arc], ctx: &TaskContext, + proto_converter: &dyn PhysicalProtoConverterExtension, ) -> Result> { if !inputs.is_empty() { return internal_err!( @@ -249,8 +251,11 @@ impl PhysicalExtensionCodec for WorkUnitFileScanCodec { let plan_node = df_proto::PhysicalPlanNode::decode(&proto.inner[..]) .map_err(|e| proto_error(format!("Failed to decode template plan: {e}")))?; - let template_plan = - plan_node.try_into_physical_plan(ctx, &DefaultPhysicalExtensionCodec {})?; + let template_plan = plan_node.try_into_physical_plan_with_converter( + ctx, + &DefaultPhysicalExtensionCodec {}, + proto_converter, + )?; let Some(dse) = template_plan.downcast_ref::() else { return Err(proto_error( "Expected the WorkUnitFileScan template plan to be a DataSourceExec", @@ -271,7 +276,12 @@ impl PhysicalExtensionCodec for WorkUnitFileScanCodec { })) } - fn try_encode(&self, node: Arc, buf: &mut Vec) -> Result<()> { + fn try_encode( + &self, + node: Arc, + buf: &mut Vec, + proto_converter: &dyn PhysicalProtoConverterExtension, + ) -> Result<()> { let Some(dse) = node.downcast_ref::() else { return internal_err!( "Expected DataSourceExec wrapping a WorkUnitFileScanConfig, got {}", @@ -284,9 +294,10 @@ impl PhysicalExtensionCodec for WorkUnitFileScanCodec { // Encode the template DataSourceExec(FileScanConfig) as a regular // PhysicalPlanNode using DataFusion's default codec. - let plan_node = df_proto::PhysicalPlanNode::try_from_physical_plan( + let plan_node = df_proto::PhysicalPlanNode::try_from_physical_plan_with_converter( DataSourceExec::from_data_source(wfs.fsc.clone()), &DefaultPhysicalExtensionCodec {}, + proto_converter, )?; let mut inner_bytes = Vec::new(); plan_node.encode(&mut inner_bytes).map_err(|e| { diff --git a/src/worker/session_builder.rs b/src/worker/session_builder.rs index c0f0bf41..7c62c520 100644 --- a/src/worker/session_builder.rs +++ b/src/worker/session_builder.rs @@ -25,18 +25,18 @@ pub trait WorkerSessionBuilder { /// # use datafusion::error::DataFusionError; /// # use datafusion::execution::{FunctionRegistry, SessionState, SessionStateBuilder, TaskContext}; /// # use datafusion::physical_plan::ExecutionPlan; - /// # use datafusion_proto::physical_plan::PhysicalExtensionCodec; + /// # use datafusion_proto::physical_plan::{PhysicalExtensionCodec, PhysicalProtoConverterExtension}; /// # use datafusion_distributed::{DistributedExt, WorkerSessionBuilder, WorkerQueryContext}; /// /// #[derive(Debug)] /// struct CustomExecCodec; /// /// impl PhysicalExtensionCodec for CustomExecCodec { - /// fn try_decode(&self, buf: &[u8], inputs: &[Arc], ctx: &TaskContext) -> datafusion::common::Result> { + /// fn try_decode(&self, buf: &[u8], inputs: &[Arc], ctx: &TaskContext, proto_converter: &dyn PhysicalProtoConverterExtension) -> datafusion::common::Result> { /// todo!() /// } /// - /// fn try_encode(&self, node: Arc, buf: &mut Vec) -> datafusion::common::Result<()> { + /// fn try_encode(&self, node: Arc, buf: &mut Vec, proto_converter: &dyn PhysicalProtoConverterExtension) -> datafusion::common::Result<()> { /// todo!() /// } /// } diff --git a/tests/custom_config_extension.rs b/tests/custom_config_extension.rs index 0f4502fa..50ccb0fe 100644 --- a/tests/custom_config_extension.rs +++ b/tests/custom_config_extension.rs @@ -14,7 +14,9 @@ mod tests { use datafusion_distributed::test_utils::localhost::start_localhost_context; use datafusion_distributed::test_utils::parquet::register_parquet_tables; use datafusion_distributed::{DistributedExt, WorkerQueryContext}; - use datafusion_proto::physical_plan::PhysicalExtensionCodec; + use datafusion_proto::physical_plan::{ + PhysicalExtensionCodec, PhysicalProtoConverterExtension, + }; use futures::TryStreamExt; use prost::Message; use std::fmt::Formatter; @@ -201,6 +203,7 @@ mod tests { buf: &[u8], inputs: &[Arc], _ctx: &TaskContext, + _proto_converter: &dyn PhysicalProtoConverterExtension, ) -> datafusion::common::Result> { let _node = CustomConfigExtensionRequiredExecProto::decode(buf) .map_err(|err| internal_datafusion_err!("{err}"))?; @@ -221,6 +224,7 @@ mod tests { &self, _node: Arc, buf: &mut Vec, + _proto_converter: &dyn PhysicalProtoConverterExtension, ) -> datafusion::common::Result<()> { CustomConfigExtensionRequiredExecProto::default() .encode(buf) diff --git a/tests/custom_extension_codec.rs b/tests/custom_extension_codec.rs index a3eb3d9f..7c23c8d8 100644 --- a/tests/custom_extension_codec.rs +++ b/tests/custom_extension_codec.rs @@ -13,7 +13,9 @@ mod tests { use datafusion_distributed::test_utils::localhost::start_localhost_context; use datafusion_distributed::test_utils::parquet::register_parquet_tables; use datafusion_distributed::{DistributedExt, WorkerQueryContext, assert_snapshot}; - use datafusion_proto::physical_plan::PhysicalExtensionCodec; + use datafusion_proto::physical_plan::{ + PhysicalExtensionCodec, PhysicalProtoConverterExtension, + }; use datafusion_proto::protobuf::proto_error; use futures::TryStreamExt; use prost::Message; @@ -139,6 +141,7 @@ mod tests { buf: &[u8], inputs: &[Arc], _ctx: &TaskContext, + _proto_converter: &dyn PhysicalProtoConverterExtension, ) -> datafusion::common::Result> { let _node = CustomPassThroughExecProto::decode(buf) .map_err(|err| proto_error(format!("{err}")))?; @@ -157,6 +160,7 @@ mod tests { &self, node: Arc, buf: &mut Vec, + _proto_converter: &dyn PhysicalProtoConverterExtension, ) -> datafusion::common::Result<()> { let Some(_plan) = node.downcast_ref::() else { return Err(proto_error(format!( diff --git a/tests/distributed_unions.rs b/tests/distributed_unions.rs index 2843dec7..915477f5 100644 --- a/tests/distributed_unions.rs +++ b/tests/distributed_unions.rs @@ -226,7 +226,7 @@ mod tests { let physical_distributed_str = display_plan_ascii(physical_distributed.as_ref(), false); assert_snapshot!(physical_distributed_str, - @r" + @" ┌───── DistributedExec ── Tasks: t0:[p0] │ SortPreservingMergeExec: [MinTemp@0 ASC NULLS LAST, RainToday@1 ASC NULLS LAST] │ [Stage 7] => NetworkCoalesceExec: output_partitions=2, input_tasks=2 @@ -252,16 +252,16 @@ mod tests { ┌───── Stage 1 ── Tasks: t0:[p0..p2] t1:[p3..p5] t2:[p6..p8] │ FilterExec: MinTemp@0 > 10, fetch=1000000 │ DistributedLeafExec: - │ t0: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000000.parquet:.., /testdata/weather/result-000001.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MinTemp, RainToday], file_type=parquet, predicate=MinTemp@0 > 10 AND DynamicFilter [ empty ], pruning_predicate=MinTemp_null_count@1 != row_count@2 AND MinTemp_max@0 > 10, required_guarantees=[] - │ t1: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000001.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MinTemp, RainToday], file_type=parquet, predicate=MinTemp@0 > 10 AND DynamicFilter [ empty ], pruning_predicate=MinTemp_null_count@1 != row_count@2 AND MinTemp_max@0 > 10, required_guarantees=[] - │ t2: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000001.parquet:.., /testdata/weather/result-000002.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MinTemp, RainToday], file_type=parquet, predicate=MinTemp@0 > 10 AND DynamicFilter [ empty ], pruning_predicate=MinTemp_null_count@1 != row_count@2 AND MinTemp_max@0 > 10, required_guarantees=[] + │ t0: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000000.parquet:.., /testdata/weather/result-000001.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MinTemp, RainToday], file_type=parquet, predicate=MinTemp@0 > 10 AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible, pruning_predicate=MinTemp_null_count@1 != row_count@2 AND MinTemp_max@0 > 10, required_guarantees=[] + │ t1: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000001.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MinTemp, RainToday], file_type=parquet, predicate=MinTemp@0 > 10 AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible, pruning_predicate=MinTemp_null_count@1 != row_count@2 AND MinTemp_max@0 > 10, required_guarantees=[] + │ t2: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000001.parquet:.., /testdata/weather/result-000002.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MinTemp, RainToday], file_type=parquet, predicate=MinTemp@0 > 10 AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible, pruning_predicate=MinTemp_null_count@1 != row_count@2 AND MinTemp_max@0 > 10, required_guarantees=[] └────────────────────────────────────────────────── ┌───── Stage 2 ── Tasks: t0:[p0..p2] t1:[p3..p5] t2:[p6..p8] │ FilterExec: MaxTemp@0 < 30, fetch=1000000 │ DistributedLeafExec: - │ t0: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000000.parquet:.., /testdata/weather/result-000001.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=MaxTemp@1 < 30 AND DynamicFilter [ empty ], pruning_predicate=MaxTemp_null_count@1 != row_count@2 AND MaxTemp_min@0 < 30, required_guarantees=[] - │ t1: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000001.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=MaxTemp@1 < 30 AND DynamicFilter [ empty ], pruning_predicate=MaxTemp_null_count@1 != row_count@2 AND MaxTemp_min@0 < 30, required_guarantees=[] - │ t2: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000001.parquet:.., /testdata/weather/result-000002.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=MaxTemp@1 < 30 AND DynamicFilter [ empty ], pruning_predicate=MaxTemp_null_count@1 != row_count@2 AND MaxTemp_min@0 < 30, required_guarantees=[] + │ t0: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000000.parquet:.., /testdata/weather/result-000001.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=MaxTemp@1 < 30 AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible, pruning_predicate=MaxTemp_null_count@1 != row_count@2 AND MaxTemp_min@0 < 30, required_guarantees=[] + │ t1: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000001.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=MaxTemp@1 < 30 AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible, pruning_predicate=MaxTemp_null_count@1 != row_count@2 AND MaxTemp_min@0 < 30, required_guarantees=[] + │ t2: DataSourceExec: file_groups={3 groups: [[/testdata/weather/result-000000.parquet:..], [/testdata/weather/result-000001.parquet:.., /testdata/weather/result-000002.parquet:..], [/testdata/weather/result-000002.parquet:..]]}, projection=[MaxTemp, RainToday], file_type=parquet, predicate=MaxTemp@1 < 30 AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible, pruning_predicate=MaxTemp_null_count@1 != row_count@2 AND MaxTemp_min@0 < 30, required_guarantees=[] └────────────────────────────────────────────────── ┌───── Stage 6 ── Tasks: t0:[p0] t1:[p1] │ LocalLimitExec: fetch=1000000 diff --git a/tests/error_propagation.rs b/tests/error_propagation.rs index 44908e71..05289594 100644 --- a/tests/error_propagation.rs +++ b/tests/error_propagation.rs @@ -13,7 +13,9 @@ mod tests { use datafusion_distributed::test_utils::localhost::start_localhost_context; use datafusion_distributed::test_utils::parquet::register_parquet_tables; use datafusion_distributed::{DistributedExt, WorkerQueryContext}; - use datafusion_proto::physical_plan::PhysicalExtensionCodec; + use datafusion_proto::physical_plan::{ + PhysicalExtensionCodec, PhysicalProtoConverterExtension, + }; use datafusion_proto::protobuf::proto_error; use futures::{TryStreamExt, stream}; use prost::Message; @@ -146,6 +148,7 @@ mod tests { buf: &[u8], inputs: &[Arc], _ctx: &TaskContext, + _proto_converter: &dyn PhysicalProtoConverterExtension, ) -> datafusion::common::Result> { let node = ErrorThrowingExecProto::decode(buf).map_err(|err| proto_error(format!("{err}")))?; @@ -167,6 +170,7 @@ mod tests { &self, node: Arc, buf: &mut Vec, + _proto_converter: &dyn PhysicalProtoConverterExtension, ) -> datafusion::common::Result<()> { let Some(plan) = node.downcast_ref::() else { return Err(proto_error(format!( diff --git a/tests/stateful_execution_plan.rs b/tests/stateful_execution_plan.rs index 4a97cee3..ae8088bf 100644 --- a/tests/stateful_execution_plan.rs +++ b/tests/stateful_execution_plan.rs @@ -16,7 +16,9 @@ mod tests { use datafusion_distributed::{ DistributedExt, WorkerQueryContext, assert_snapshot, display_plan_ascii, }; - use datafusion_proto::physical_plan::PhysicalExtensionCodec; + use datafusion_proto::physical_plan::{ + PhysicalExtensionCodec, PhysicalProtoConverterExtension, + }; use datafusion_proto::protobuf::proto_error; use futures::TryStreamExt; use prost::Message; @@ -185,6 +187,7 @@ mod tests { buf: &[u8], inputs: &[Arc], _ctx: &TaskContext, + _proto_converter: &dyn PhysicalProtoConverterExtension, ) -> datafusion::common::Result> { let _node = PassThroughExecProto::decode(buf).map_err(|err| proto_error(format!("{err}")))?; @@ -203,6 +206,7 @@ mod tests { &self, node: Arc, buf: &mut Vec, + _proto_converter: &dyn PhysicalProtoConverterExtension, ) -> datafusion::common::Result<()> { let Some(_plan) = node.downcast_ref::() else { return Err(proto_error(format!(