diff --git a/native/Cargo.lock b/native/Cargo.lock index eeaa9da5ce..4562386f2c 100644 --- a/native/Cargo.lock +++ b/native/Cargo.lock @@ -68,9 +68,9 @@ dependencies = [ [[package]] name = "aho-corasick" -version = "1.1.4" +version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ddd31a130427c27518df266943a5308ed92d4b226cc639f5a8f1002816174301" +checksum = "c982642fa9e8606056828ee9a8505737230110bb1099153c79efe865c59d12ba" dependencies = [ "memchr", ] @@ -107,9 +107,9 @@ checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923" [[package]] name = "android_system_properties" -version = "0.1.5" +version = "0.1.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "819e7219dbd41043ac279b19830f2efc897156490d7fd6ea916720117ee66311" +checksum = "ae221649c9976a6f6c56ae1facf410f3ddb33cc661c4b7b61020a912d4237fbc" dependencies = [ "libc", ] @@ -232,9 +232,9 @@ checksum = "d3fb67a6e08acf24fdeccbac2cb6ac4305825bd1f117462e0e6f2f193345ad56" [[package]] name = "arrow" -version = "58.4.0" +version = "59.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6cfdd0833e32a9874d2b55089333ad310c0be208aafa277385ce2461dec90be3" +checksum = "b952ca5a8046ad741b60f142d6eca4aeebcad615694202bc64c5341f23e32c5b" dependencies = [ "arrow-arith", "arrow-array", @@ -253,9 +253,9 @@ dependencies = [ [[package]] name = "arrow-arith" -version = "58.4.0" +version = "59.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0a41203398f0eaa6f7ec8e62c0da742a21abf282c148fc157f6c35c90e29981a" +checksum = "64a13b8d3008c4e9063c597a08f46446fe3fd5789277127672d6c0bdbb43b1ff" dependencies = [ "arrow-array", "arrow-buffer", @@ -267,9 +267,9 @@ dependencies = [ [[package]] name = "arrow-array" -version = "58.4.0" +version = "59.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ae33dad492b7df00a217563a7b0ef2874df68a0deea1b1a3acf628152f7f7a69" +checksum = "9486151b2f0785bafc6fa04fc5c99fcb4495455662e58787ea32eaaed33c4192" dependencies = [ "ahash", "arrow-buffer", @@ -286,9 +286,9 @@ dependencies = [ [[package]] name = "arrow-buffer" -version = "58.4.0" +version = "59.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b9552f96391c005e6ab449fa941420935e7e062489b12b8b1b08879b2163f5b5" +checksum = "c4776577a87794bfdf0b4e90e2ea12454fa7738ea2823c4be5b9d1851da7b434" dependencies = [ "bytes", "half", @@ -298,9 +298,9 @@ dependencies = [ [[package]] name = "arrow-cast" -version = "58.4.0" +version = "59.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3a8a327c9649f30d8406995f27642b68df354713cca3baaaf100f076f18d5f34" +checksum = "a9ad451ce4f98710828a455b96991b8f031deb2e67f5fcad6773f017e4a69c3a" dependencies = [ "arrow-array", "arrow-buffer", @@ -320,9 +320,9 @@ dependencies = [ [[package]] name = "arrow-csv" -version = "58.4.0" +version = "59.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "af0dd6d90d1955e9f9a014c1e563ee8aeffc21909085d25623e1da44d96eca26" +checksum = "8aa7bf96d6141a7bcca2eed57c7c9767d2a2175281857b8a7b68308992864784" dependencies = [ "arrow-array", "arrow-cast", @@ -335,9 +335,9 @@ dependencies = [ [[package]] name = "arrow-data" -version = "58.4.0" +version = "59.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2b24852db04738907e06c04ea61e42fe7fda962a34513022dc0d0e754fb7976b" +checksum = "b38fe43e2e8704360f1464e6e8cc4fc381ef02cc4fb0192afa8df1aaa0115c66" dependencies = [ "arrow-buffer", "arrow-schema", @@ -348,9 +348,9 @@ dependencies = [ [[package]] name = "arrow-ipc" -version = "58.4.0" +version = "59.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "29a908a11fcfb3fb2f6730f4ac15e367bc644e419155e96238f68cf3adde572b" +checksum = "29dac499fcbc6ba74ee0324057821d381929a48526a3966bd9dffb44aa06d98c" dependencies = [ "arrow-array", "arrow-buffer", @@ -364,9 +364,9 @@ dependencies = [ [[package]] name = "arrow-json" -version = "58.4.0" +version = "59.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b8a96aed3931c076adee39ec2a40d8219fc7f09e79bcdaca1df16272993e1e14" +checksum = "0fe05e916ddc50f4c7a363cd69c0ef5894fcee063517e9a0b8582f0c56746af6" dependencies = [ "arrow-array", "arrow-buffer", @@ -389,9 +389,9 @@ dependencies = [ [[package]] name = "arrow-ord" -version = "58.4.0" +version = "59.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "63a083ec750f5c043f02946b4baf05fcdbb55f4560a3277055caca5cc99f3eb0" +checksum = "0e13dbdc2a9c053c10c7baa6e30faee04a180aa7ce88e471835850ce37abd20b" dependencies = [ "arrow-array", "arrow-buffer", @@ -402,9 +402,9 @@ dependencies = [ [[package]] name = "arrow-row" -version = "58.4.0" +version = "59.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "514ba0ef0d4c5896202dae736251ce415abb43a950bed570fb7981b8716c0e4c" +checksum = "4d5a1f8c733d15260b305683472ee8ad89c62cbd706703ca873b90d051b41592" dependencies = [ "arrow-array", "arrow-buffer", @@ -415,9 +415,9 @@ dependencies = [ [[package]] name = "arrow-schema" -version = "58.4.0" +version = "59.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "21ca356ad6425cecb6eb7b28e4f659f1ee7880fbb1a16127de7dd62901efee9e" +checksum = "d9e4969dc350d571766247143ab36a5187d095d3d3690970408bc630d47c69e5" dependencies = [ "bitflags 2.13.1", "serde_core", @@ -426,9 +426,9 @@ dependencies = [ [[package]] name = "arrow-select" -version = "58.4.0" +version = "59.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c58da39eb3d8350ad4a549e5c2bc49284dac554016c69829310350f1731b0aad" +checksum = "402770dba90865359d98d1ef92ef16e23d75c0cca9c2c880c8a05468b7743bf9" dependencies = [ "ahash", "arrow-array", @@ -440,9 +440,9 @@ dependencies = [ [[package]] name = "arrow-string" -version = "58.4.0" +version = "59.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b6789b388467525e3271326b6b4915666ecfdf5142aef09779445c954b67543c" +checksum = "a2b0afbb8b9016700938291123df30838b89decc3213dba00852021988b170d3" dependencies = [ "arrow-array", "arrow-buffer", @@ -1059,9 +1059,9 @@ checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" [[package]] name = "base64" -version = "0.23.0" +version = "0.23.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b25655df2c3cdd83c5e5b293b88acd880332b2ddadd7c30ac43144fdc0033da9" +checksum = "ac07cdecf99051d9a5238b80f35af32cdeba5b336e55d957b318b50137e18da5" [[package]] name = "base64-simd" @@ -1122,9 +1122,9 @@ dependencies = [ [[package]] name = "blake3" -version = "1.8.5" +version = "1.8.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0aa83c34e62843d924f905e0f5c866eb1dd6545fc4d719e803d9ba6030371fce" +checksum = "76ae7bad254120e9e4c63bafc385310756f90c484eac0e36b8317cf09cb92a77" dependencies = [ "arrayref", "arrayvec", @@ -1591,6 +1591,16 @@ dependencies = [ "libc", ] +[[package]] +name = "crc-fast" +version = "1.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e75b2483e97a5a7da73ac68a05b629f9c53cff58d8ed1c77866079e18b00dba5" +dependencies = [ + "digest 0.10.7", + "spin 0.10.1", +] + [[package]] name = "crc32c" version = "0.6.8" @@ -1839,8 +1849,7 @@ dependencies = [ [[package]] name = "datafusion" version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "754ef4e8f073922a26f5b23133b9db4829342362b09be0bc94309cf261c2f098" +source = "git+https://github.com/apache/datafusion.git?rev=70c26a06716eb552f0fc1d2115957a4089504f11#70c26a06716eb552f0fc1d2115957a4089504f11" dependencies = [ "arrow", "arrow-schema", @@ -1873,7 +1882,7 @@ dependencies = [ "datafusion-sql", "futures", "indexmap 2.14.0", - "itertools 0.14.0", + "itertools 0.15.0", "log", "object_store", "parking_lot", @@ -1888,8 +1897,7 @@ dependencies = [ [[package]] name = "datafusion-catalog" version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "06afd1e38dd27bbb1258685a1fc6524df6aff4e07b25b393a47de59635178d99" +source = "git+https://github.com/apache/datafusion.git?rev=70c26a06716eb552f0fc1d2115957a4089504f11#70c26a06716eb552f0fc1d2115957a4089504f11" dependencies = [ "arrow", "async-trait", @@ -1903,7 +1911,7 @@ dependencies = [ "datafusion-physical-plan", "datafusion-session", "futures", - "itertools 0.14.0", + "itertools 0.15.0", "log", "object_store", "parking_lot", @@ -1913,8 +1921,7 @@ dependencies = [ [[package]] name = "datafusion-catalog-listing" version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f0668fb32c12065ec242be0e5b4bc62bd7a06a0be3ecd83791ef877e4be67e02" +source = "git+https://github.com/apache/datafusion.git?rev=70c26a06716eb552f0fc1d2115957a4089504f11#70c26a06716eb552f0fc1d2115957a4089504f11" dependencies = [ "arrow", "async-trait", @@ -1928,9 +1935,10 @@ dependencies = [ "datafusion-physical-expr-common", "datafusion-physical-plan", "futures", - "itertools 0.14.0", + "itertools 0.15.0", "log", "object_store", + "percent-encoding", ] [[package]] @@ -2059,7 +2067,7 @@ name = "datafusion-comet-spark-expr" version = "1.1.0" dependencies = [ "arrow", - "base64 0.23.0", + "base64 0.23.1", "chrono", "chrono-tz", "criterion", @@ -2082,8 +2090,7 @@ dependencies = [ [[package]] name = "datafusion-common" version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ca43b263cdff57042cfa8fb817fb3469f4878933380dccff25f5e793580abbf9" +source = "git+https://github.com/apache/datafusion.git?rev=70c26a06716eb552f0fc1d2115957a4089504f11#70c26a06716eb552f0fc1d2115957a4089504f11" dependencies = [ "arrow", "arrow-ipc", @@ -2094,9 +2101,10 @@ dependencies = [ "hashbrown 0.17.1", "hex", "indexmap 2.14.0", - "itertools 0.14.0", + "itertools 0.15.0", "libc", "log", + "num-traits", "object_store", "parquet", "sqlparser", @@ -2108,8 +2116,7 @@ dependencies = [ [[package]] name = "datafusion-common-runtime" version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "05f0ba2b864792bdca4d76c59a1de0ab6e1b61946596b9936888dbd6360035f2" +source = "git+https://github.com/apache/datafusion.git?rev=70c26a06716eb552f0fc1d2115957a4089504f11#70c26a06716eb552f0fc1d2115957a4089504f11" dependencies = [ "futures", "log", @@ -2119,8 +2126,7 @@ dependencies = [ [[package]] name = "datafusion-datasource" version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b840a8bce0bcbf5afad02946d438591e7c373f7afccaf3d874c04485772514dd" +source = "git+https://github.com/apache/datafusion.git?rev=70c26a06716eb552f0fc1d2115957a4089504f11#70c26a06716eb552f0fc1d2115957a4089504f11" dependencies = [ "arrow", "async-compression", @@ -2140,7 +2146,7 @@ dependencies = [ "flate2", "futures", "glob", - "itertools 0.14.0", + "itertools 0.15.0", "liblzma", "log", "object_store", @@ -2155,8 +2161,7 @@ dependencies = [ [[package]] name = "datafusion-datasource-arrow" version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a24cc0b9cf6e367f27f27406eff13abf48a11b72446aaa40b3105c0ded5c17d9" +source = "git+https://github.com/apache/datafusion.git?rev=70c26a06716eb552f0fc1d2115957a4089504f11#70c26a06716eb552f0fc1d2115957a4089504f11" dependencies = [ "arrow", "arrow-ipc", @@ -2171,7 +2176,7 @@ dependencies = [ "datafusion-physical-plan", "datafusion-session", "futures", - "itertools 0.14.0", + "itertools 0.15.0", "object_store", "tokio", ] @@ -2179,8 +2184,7 @@ dependencies = [ [[package]] name = "datafusion-datasource-csv" version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e1abe56b2a7a2d1d6de5117dd1a203181e267f28529faa5da546947621b697d7" +source = "git+https://github.com/apache/datafusion.git?rev=70c26a06716eb552f0fc1d2115957a4089504f11#70c26a06716eb552f0fc1d2115957a4089504f11" dependencies = [ "arrow", "async-trait", @@ -2202,8 +2206,7 @@ dependencies = [ [[package]] name = "datafusion-datasource-json" version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9c3e467f0611ad7bdd5aad17c63c9bb6182d04e5282e5496d897ea2b49c024ba" +source = "git+https://github.com/apache/datafusion.git?rev=70c26a06716eb552f0fc1d2115957a4089504f11#70c26a06716eb552f0fc1d2115957a4089504f11" dependencies = [ "arrow", "async-trait", @@ -2225,10 +2228,10 @@ dependencies = [ [[package]] name = "datafusion-datasource-parquet" version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4cc35b92cd560082155e80d9c826929c852d3c51543f4affd3a51c464a0aab3a" +source = "git+https://github.com/apache/datafusion.git?rev=70c26a06716eb552f0fc1d2115957a4089504f11#70c26a06716eb552f0fc1d2115957a4089504f11" dependencies = [ "arrow", + "arrow-schema", "async-trait", "bytes", "datafusion-common", @@ -2245,7 +2248,7 @@ dependencies = [ "datafusion-pruning", "datafusion-session", "futures", - "itertools 0.14.0", + "itertools 0.15.0", "log", "object_store", "parking_lot", @@ -2256,18 +2259,17 @@ dependencies = [ [[package]] name = "datafusion-doc" version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d69bb69d8769e34f76839c960dbde24c1ac0c885a79b6c3c2287bdc56ec67891" +source = "git+https://github.com/apache/datafusion.git?rev=70c26a06716eb552f0fc1d2115957a4089504f11#70c26a06716eb552f0fc1d2115957a4089504f11" [[package]] name = "datafusion-execution" version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d8eac0a09bc8d263f52025cad9e001da4d8138d633fa288edda4d06b1772eae6" +source = "git+https://github.com/apache/datafusion.git?rev=70c26a06716eb552f0fc1d2115957a4089504f11#70c26a06716eb552f0fc1d2115957a4089504f11" dependencies = [ "arrow", "arrow-buffer", "async-trait", + "bytes", "dashmap", "datafusion-common", "datafusion-expr", @@ -2277,16 +2279,18 @@ dependencies = [ "object_store", "parking_lot", "parquet", + "pin-project-lite", "rand 0.9.5", "tempfile", + "tokio", + "tokio-util", "url", ] [[package]] name = "datafusion-expr" version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "eeb14d374767ee0fc62dc79a5ba8bcf8a63c14e993c7d992d0e63adfa23d77d3" +source = "git+https://github.com/apache/datafusion.git?rev=70c26a06716eb552f0fc1d2115957a4089504f11#70c26a06716eb552f0fc1d2115957a4089504f11" dependencies = [ "arrow", "arrow-schema", @@ -2299,7 +2303,7 @@ dependencies = [ "datafusion-functions-window-common", "datafusion-physical-expr-common", "indexmap 2.14.0", - "itertools 0.14.0", + "itertools 0.15.0", "serde_json", "sqlparser", ] @@ -2307,24 +2311,22 @@ dependencies = [ [[package]] name = "datafusion-expr-common" version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b7b19a8c95522bee8cbb313d74263b85e355d2b52f42e67ef5694bf5de9e9356" +source = "git+https://github.com/apache/datafusion.git?rev=70c26a06716eb552f0fc1d2115957a4089504f11#70c26a06716eb552f0fc1d2115957a4089504f11" dependencies = [ "arrow", "datafusion-common", "indexmap 2.14.0", - "itertools 0.14.0", + "itertools 0.15.0", ] [[package]] name = "datafusion-functions" version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5f64c983bbbdcb729d921a2b2ac3375598719b5cc0c30345ad664936f3176fc7" +source = "git+https://github.com/apache/datafusion.git?rev=70c26a06716eb552f0fc1d2115957a4089504f11#70c26a06716eb552f0fc1d2115957a4089504f11" dependencies = [ "arrow", "arrow-buffer", - "base64 0.22.1", + "base64 0.23.1", "blake2", "blake3", "chrono", @@ -2337,7 +2339,7 @@ dependencies = [ "datafusion-macros", "datafusion-physical-expr-common", "hex", - "itertools 0.14.0", + "itertools 0.15.0", "log", "md-5 0.11.0", "memchr", @@ -2351,8 +2353,7 @@ dependencies = [ [[package]] name = "datafusion-functions-aggregate" version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "89bc17041e424a47ed062f43df24d84aab8b57c4c3221e5c1a5eef46d6c5718b" +source = "git+https://github.com/apache/datafusion.git?rev=70c26a06716eb552f0fc1d2115957a4089504f11#70c26a06716eb552f0fc1d2115957a4089504f11" dependencies = [ "arrow", "datafusion-common", @@ -2363,8 +2364,8 @@ dependencies = [ "datafusion-macros", "datafusion-physical-expr", "datafusion-physical-expr-common", - "foldhash 0.2.0", "half", + "hashbrown 0.17.1", "log", "num-traits", ] @@ -2372,8 +2373,7 @@ dependencies = [ [[package]] name = "datafusion-functions-aggregate-common" version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "97dd2a9e865c6108059f5b37b77934f84b50bfb108f837bd0e5c9536e03f0545" +source = "git+https://github.com/apache/datafusion.git?rev=70c26a06716eb552f0fc1d2115957a4089504f11#70c26a06716eb552f0fc1d2115957a4089504f11" dependencies = [ "arrow", "datafusion-common", @@ -2384,8 +2384,7 @@ dependencies = [ [[package]] name = "datafusion-functions-nested" version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "75f0bdfeef16d96417b9632ef855645376b242e9006a126dfd0bedfc54a93f5f" +source = "git+https://github.com/apache/datafusion.git?rev=70c26a06716eb552f0fc1d2115957a4089504f11#70c26a06716eb552f0fc1d2115957a4089504f11" dependencies = [ "arrow", "arrow-ord", @@ -2400,7 +2399,7 @@ dependencies = [ "datafusion-macros", "datafusion-physical-expr-common", "hashbrown 0.17.1", - "itertools 0.14.0", + "itertools 0.15.0", "itoa", "log", "memchr", @@ -2409,8 +2408,7 @@ dependencies = [ [[package]] name = "datafusion-functions-table" version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f4e4941673c917819616877e9993da4503e4f4739812be0bc32c5356184c6383" +source = "git+https://github.com/apache/datafusion.git?rev=70c26a06716eb552f0fc1d2115957a4089504f11#70c26a06716eb552f0fc1d2115957a4089504f11" dependencies = [ "arrow", "async-trait", @@ -2425,8 +2423,7 @@ dependencies = [ [[package]] name = "datafusion-functions-window" version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "12dd2e16c12b84b6f6b41b19f55b366dd1c46876bb35b86896c6349067379e8d" +source = "git+https://github.com/apache/datafusion.git?rev=70c26a06716eb552f0fc1d2115957a4089504f11#70c26a06716eb552f0fc1d2115957a4089504f11" dependencies = [ "arrow", "datafusion-common", @@ -2442,8 +2439,7 @@ dependencies = [ [[package]] name = "datafusion-functions-window-common" version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4cdc5e4b6f8b6ef823cc1c761f85088ad4c884fe8df64df3cbcc6b2b84698441" +source = "git+https://github.com/apache/datafusion.git?rev=70c26a06716eb552f0fc1d2115957a4089504f11#70c26a06716eb552f0fc1d2115957a4089504f11" dependencies = [ "datafusion-common", "datafusion-physical-expr-common", @@ -2452,19 +2448,17 @@ dependencies = [ [[package]] name = "datafusion-macros" version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1a3614234dd93578c92428cb4f408e020874f0d2b7e6c90c928d9d28b5df2ceb" +source = "git+https://github.com/apache/datafusion.git?rev=70c26a06716eb552f0fc1d2115957a4089504f11#70c26a06716eb552f0fc1d2115957a4089504f11" dependencies = [ "datafusion-doc", "quote", - "syn 2.0.119", + "syn 3.0.3", ] [[package]] name = "datafusion-optimizer" version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a0635620b050b81bb92764e99250868f654e2cd5ad1bece413283b3f73c83179" +source = "git+https://github.com/apache/datafusion.git?rev=70c26a06716eb552f0fc1d2115957a4089504f11#70c26a06716eb552f0fc1d2115957a4089504f11" dependencies = [ "arrow", "chrono", @@ -2473,7 +2467,7 @@ dependencies = [ "datafusion-expr-common", "datafusion-physical-expr", "indexmap 2.14.0", - "itertools 0.14.0", + "itertools 0.15.0", "log", "regex", "regex-syntax", @@ -2482,8 +2476,7 @@ dependencies = [ [[package]] name = "datafusion-physical-expr" version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8cabf7a86eb70b816729e33c81bf7767c936ee1226f607a114f5dac2decac8d0" +source = "git+https://github.com/apache/datafusion.git?rev=70c26a06716eb552f0fc1d2115957a4089504f11#70c26a06716eb552f0fc1d2115957a4089504f11" dependencies = [ "arrow", "datafusion-common", @@ -2494,7 +2487,7 @@ dependencies = [ "half", "hashbrown 0.17.1", "indexmap 2.14.0", - "itertools 0.14.0", + "itertools 0.15.0", "parking_lot", "petgraph", "tokio", @@ -2503,8 +2496,7 @@ dependencies = [ [[package]] name = "datafusion-physical-expr-adapter" version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "de222e04f7e6744501555a54ab0abe26bfdfebee380af79a9bdc175704246859" +source = "git+https://github.com/apache/datafusion.git?rev=70c26a06716eb552f0fc1d2115957a4089504f11#70c26a06716eb552f0fc1d2115957a4089504f11" dependencies = [ "arrow", "datafusion-common", @@ -2512,14 +2504,13 @@ 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.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "72d0d0057fc5a502d45c870cb6d47c66eb7bdd5edb1bd71ad6f3f724975ac2a8" +source = "git+https://github.com/apache/datafusion.git?rev=70c26a06716eb552f0fc1d2115957a4089504f11#70c26a06716eb552f0fc1d2115957a4089504f11" dependencies = [ "arrow", "chrono", @@ -2527,7 +2518,7 @@ dependencies = [ "datafusion-expr-common", "hashbrown 0.17.1", "indexmap 2.14.0", - "itertools 0.14.0", + "itertools 0.15.0", "parking_lot", "pin-project", ] @@ -2535,8 +2526,7 @@ dependencies = [ [[package]] name = "datafusion-physical-optimizer" version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "86046eed10950c5f9aaed9acfd148e9bd2e1dfdfe4f9aef607d1447b271e4183" +source = "git+https://github.com/apache/datafusion.git?rev=70c26a06716eb552f0fc1d2115957a4089504f11#70c26a06716eb552f0fc1d2115957a4089504f11" dependencies = [ "arrow", "datafusion-common", @@ -2547,14 +2537,14 @@ dependencies = [ "datafusion-physical-expr-common", "datafusion-physical-plan", "datafusion-pruning", - "itertools 0.14.0", + "datafusion-session", + "itertools 0.15.0", ] [[package]] name = "datafusion-physical-plan" version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9bc84da934c903407ba297971ebcc020c4c1a38aafd765d6c144c76eff3fa6a1" +source = "git+https://github.com/apache/datafusion.git?rev=70c26a06716eb552f0fc1d2115957a4089504f11#70c26a06716eb552f0fc1d2115957a4089504f11" dependencies = [ "arrow", "arrow-data", @@ -2562,6 +2552,7 @@ dependencies = [ "arrow-ord", "arrow-schema", "async-trait", + "bytes", "datafusion-common", "datafusion-common-runtime", "datafusion-execution", @@ -2575,19 +2566,19 @@ dependencies = [ "half", "hashbrown 0.17.1", "indexmap 2.14.0", - "itertools 0.14.0", + "itertools 0.15.0", "log", "num-traits", "parking_lot", "pin-project-lite", + "serde_json", "tokio", ] [[package]] name = "datafusion-pruning" version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "eb63eeac6de19be40f487b65dd84e546195f783c5a9928618e0c4f2a3569b0d7" +source = "git+https://github.com/apache/datafusion.git?rev=70c26a06716eb552f0fc1d2115957a4089504f11#70c26a06716eb552f0fc1d2115957a4089504f11" dependencies = [ "arrow", "datafusion-common", @@ -2602,9 +2593,9 @@ dependencies = [ [[package]] name = "datafusion-session" version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5f961d209177f91bd014db5cbb2c33b7d28a2597b9003e77f17aeb712964315a" +source = "git+https://github.com/apache/datafusion.git?rev=70c26a06716eb552f0fc1d2115957a4089504f11#70c26a06716eb552f0fc1d2115957a4089504f11" dependencies = [ + "arrow-schema", "async-trait", "datafusion-common", "datafusion-execution", @@ -2616,8 +2607,7 @@ dependencies = [ [[package]] name = "datafusion-spark" version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a1ccd16a6949503e56c084df1b90c8889db826ec9347d2f0f51a7837d6fa011e" +source = "git+https://github.com/apache/datafusion.git?rev=70c26a06716eb552f0fc1d2115957a4089504f11#70c26a06716eb552f0fc1d2115957a4089504f11" dependencies = [ "arrow", "bigdecimal", @@ -2646,8 +2636,7 @@ dependencies = [ [[package]] name = "datafusion-sql" version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b1d71cb454da682b2af7488e1fc1ddd72ee1b28f19297b8ccad73f0a21ee9a69" +source = "git+https://github.com/apache/datafusion.git?rev=70c26a06716eb552f0fc1d2115957a4089504f11#70c26a06716eb552f0fc1d2115957a4089504f11" dependencies = [ "arrow", "bigdecimal", @@ -3497,7 +3486,7 @@ dependencies = [ [[package]] name = "iceberg" version = "0.10.0" -source = "git+https://github.com/apache/iceberg-rust?rev=3d84c81353b1b23b6e4ae8eea8f8a021cc6927a7#3d84c81353b1b23b6e4ae8eea8f8a021cc6927a7" +source = "git+https://github.com/mbutrovich/iceberg-rust?rev=07de8fc8d650f7439b251d57a55613f7de693eb8#07de8fc8d650f7439b251d57a55613f7de693eb8" dependencies = [ "aes-gcm", "anyhow", @@ -3523,6 +3512,7 @@ dependencies = [ "fastnum", "flate2", "fnv", + "form_urlencoded", "futures", "itertools 0.13.0", "moka", @@ -3553,7 +3543,7 @@ dependencies = [ [[package]] name = "iceberg-storage-opendal" version = "0.10.0" -source = "git+https://github.com/apache/iceberg-rust?rev=3d84c81353b1b23b6e4ae8eea8f8a021cc6927a7#3d84c81353b1b23b6e4ae8eea8f8a021cc6927a7" +source = "git+https://github.com/mbutrovich/iceberg-rust?rev=07de8fc8d650f7439b251d57a55613f7de693eb8#07de8fc8d650f7439b251d57a55613f7de693eb8" dependencies = [ "anyhow", "async-trait", @@ -3729,12 +3719,6 @@ dependencies = [ "generic-array", ] -[[package]] -name = "integer-encoding" -version = "3.0.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8bb03732005da905c88227371639bf1ad885cc712789c011c31c5fb3ab3ccf02" - [[package]] name = "inventory" version = "0.3.24" @@ -4475,9 +4459,9 @@ dependencies = [ [[package]] name = "object_store_opendal" -version = "0.57.0" +version = "0.58.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0eb12a624a41fce745838d0ef3701ff6c47797c13cd18ad3612fd2a3134fdbd8" +checksum = "88f165780495c17aa3ce86846600504198c3fffd99073521552751c2430fa6ac" dependencies = [ "async-trait", "bytes", @@ -4516,12 +4500,13 @@ checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381" [[package]] name = "opendal" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "96c9c85ce253ff87225e7669979d877a20c98a06604ec9d6dd5f4473e08f1ae1" +checksum = "4f20562cc7447fcc915fc5c23df305a412ea80a733c9f2fd9e2d267e2815be6d" dependencies = [ "ctor", "opendal-core", + "opendal-http-transport-reqwest", "opendal-layer-concurrent-limit", "opendal-layer-logging", "opendal-layer-retry", @@ -4536,24 +4521,22 @@ dependencies = [ [[package]] name = "opendal-core" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c4f8607c90e2c963a91467f50fb49fbc7fb3d573f88cea219ca59ccd3740b309" +checksum = "ec75551ff4cf3e57da98979f6a937aaa9ddb3915bf68cc17d03df733be6646ed" dependencies = [ "anyhow", - "base64 0.22.1", + "base64 0.23.1", "bytes", "futures", "http 1.5.0", - "http-body 1.1.0", "jiff", "log", "md-5 0.11.0", "mea", "percent-encoding", - "quick-xml 0.39.4", + "quick-xml 0.41.0", "reqsign-core", - "reqwest 0.13.4", "serde", "serde_json", "tokio", @@ -4562,11 +4545,25 @@ dependencies = [ "web-time", ] +[[package]] +name = "opendal-http-transport-reqwest" +version = "0.58.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ad4d4f19c3ce01126a30611f8e544eaa217104a278c889ac17c9374fe4f9e4ef" +dependencies = [ + "bytes", + "futures", + "http 1.5.0", + "http-body 1.1.0", + "opendal-core", + "reqwest 0.13.4", +] + [[package]] name = "opendal-layer-concurrent-limit" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0d6f81ba6960e3fae1882f253b114b21d7e444e1534f209c7737a79f6243eb6f" +checksum = "249ac5b0aa5a7a6c3737342d10456067937f9c9a6f3f02544271f7908ab91081" dependencies = [ "futures", "http 1.5.0", @@ -4576,9 +4573,9 @@ dependencies = [ [[package]] name = "opendal-layer-logging" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "58ada45c6d81d1aa4c9305d0c7d4bc317c59c85866a0908a2d75a7a978aa5ee2" +checksum = "5c75411ab00f77851ff086b686c1e9ca8175ac18c15afa2cb75b9036436cb06c" dependencies = [ "log", "opendal-core", @@ -4586,9 +4583,9 @@ dependencies = [ [[package]] name = "opendal-layer-retry" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7b2a25a718afb81fad81cb9a0580a1cb989221fa2317f888c6a37f8dad408eb7" +checksum = "80b7738bd5f233ad8da39af9b9316b9b7a4eaddd91e8e32a1e19b7030688121d" dependencies = [ "backon", "log", @@ -4597,9 +4594,9 @@ dependencies = [ [[package]] name = "opendal-layer-timeout" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1e91f731724c213af81e9d03517859c8fc47b4578e64ad61ae4f099f10fe36e3" +checksum = "a704141924500f3803c05ed871b53305d2a2f11cb5ef20160c3ee688a1857f66" dependencies = [ "opendal-core", "tokio", @@ -4607,17 +4604,18 @@ dependencies = [ [[package]] name = "opendal-service-azdls" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6dea4908d490143a9b0b7f7a790e139ff829b06a023f670455ed3d44f664b361" +checksum = "2e3c406729935fe214ce574d68681a1ff7e0b322548f14094912bdbfe50e5c53" dependencies = [ - "base64 0.22.1", + "base64 0.23.1", "bytes", "http 1.5.0", "log", + "mea", "opendal-core", "opendal-service-azure-common", - "quick-xml 0.39.4", + "quick-xml 0.41.0", "reqsign-azure-storage", "reqsign-core", "reqsign-file-read-tokio", @@ -4627,9 +4625,9 @@ dependencies = [ [[package]] name = "opendal-service-azure-common" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9b489f13c42e69d69bdd72952b634356ec43a7881a20259b38b540fcecdf4051" +checksum = "7348c88edf15af435b7be930077746b569fac5e738c1bf6a363b675e7317c9df" dependencies = [ "http 1.5.0", "opendal-core", @@ -4637,9 +4635,9 @@ dependencies = [ [[package]] name = "opendal-service-fs" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "22e89a665fef0e6bd249cf5ea47fc174b7ba892159bee4b9382528b1ca873a2c" +checksum = "826c4e17a30643b888fe983897f9a4b23b07066e1d069727a923cc8fb419a702" dependencies = [ "bytes", "log", @@ -4651,9 +4649,9 @@ dependencies = [ [[package]] name = "opendal-service-gcs" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "48de101aac565ed06af4b47903c24eafd249075553ec1fb18256751c45148d47" +checksum = "007f3fba63c21e516c956b891e96ff9892d8175662bfb781cdada9d3766a11e6" dependencies = [ "async-trait", "bytes", @@ -4661,7 +4659,7 @@ dependencies = [ "log", "opendal-core", "percent-encoding", - "quick-xml 0.39.4", + "quick-xml 0.41.0", "reqsign-core", "reqsign-file-read-tokio", "reqsign-google", @@ -4672,9 +4670,9 @@ dependencies = [ [[package]] name = "opendal-service-hdfs" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fb51ce674ce98b8b7d9ac7d2dfc9d7fc8f1bbd1da1a5fb928ecc5f13ba7dc88a" +checksum = "00570d1cc0c3f00db60bc8aaa63aa304e176285264e8674ca11897afaf509629" dependencies = [ "bytes", "futures", @@ -4687,15 +4685,15 @@ dependencies = [ [[package]] name = "opendal-service-oss" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "328fa55e8888cbdfe00826bfea2a79042422b720e8369e9e021e46121dea5ace" +checksum = "cd528ec2d49c5ca69e674ffed7b3e0686fb9cfcfea0596870de381467fda4f1b" dependencies = [ "bytes", "http 1.5.0", "log", "opendal-core", - "quick-xml 0.39.4", + "quick-xml 0.41.0", "reqsign-aliyun-oss", "reqsign-core", "reqsign-file-read-tokio", @@ -4704,18 +4702,18 @@ dependencies = [ [[package]] name = "opendal-service-s3" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "313d46c9f5ae70bca26b7c3e3fbb9b639292625f28af73aa016f47e788af9deb" +checksum = "58e80cdf192d7eff05feed747894d64f81905ac4eaf132edf7ea270abdd2d663" dependencies = [ - "base64 0.22.1", + "base64 0.23.1", "bytes", - "crc32c", + "crc-fast", "http 1.5.0", "log", "md-5 0.11.0", "opendal-core", - "quick-xml 0.39.4", + "quick-xml 0.41.0", "reqsign-aws-v4", "reqsign-core", "reqsign-file-read-tokio", @@ -4794,9 +4792,9 @@ dependencies = [ [[package]] name = "parquet" -version = "58.4.0" +version = "59.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d298093b2dec60289dce0684c986d0f7679e9dd15771c2c65406e1aaf604a704" +checksum = "5302d4da74d6596a1f11f9928767995b53bca657cbeea1e4e8c5074f8a1157dd" dependencies = [ "ahash", "arrow-array", @@ -4826,7 +4824,6 @@ dependencies = [ "seq-macro", "simdutf8", "snap", - "thrift", "tokio", "twox-hash", "zstd", @@ -4834,9 +4831,9 @@ dependencies = [ [[package]] name = "parquet-variant" -version = "58.4.0" +version = "59.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3fc70e87931167a4a3fde2ee923023a0624367691e3fd503476a1084dda5a054" +checksum = "57db04c7f8598c9e12c935a7e953a441009189da3f1d504b63c5dc3346a4d59b" dependencies = [ "arrow", "arrow-schema", @@ -4850,9 +4847,9 @@ dependencies = [ [[package]] name = "parquet-variant-compute" -version = "58.4.0" +version = "59.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "823a9ecee8fd83a68f7165ef13acc840c4d7ed995838ba99d4714b20a2b5e780" +checksum = "9ecebd5a05d4837104ae401977f5d8ad816045e01b367035c416df3c39540613" dependencies = [ "arrow", "arrow-schema", @@ -4867,9 +4864,9 @@ dependencies = [ [[package]] name = "parquet-variant-json" -version = "58.4.0" +version = "59.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6f37a91177e2dddb10333546952fcf2b7674b97ebd9449432f24b883d6ea8108" +checksum = "e5fc56a025dc3d5537140fc95d80879750fc1aa4324cf2452f3d1471854c2f8d" dependencies = [ "arrow-schema", "base64 0.22.1", @@ -5506,9 +5503,9 @@ dependencies = [ [[package]] name = "regex-automata" -version = "0.4.16" +version = "0.4.18" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8fcfdb36bda0c880c5931cdc7a2bcdc8ba4556847b9d912bca70bc94708711ad" +checksum = "ad8553b9b26413251cbf30e620595c7a41b3887f03da04579c0e6b0d6a06b4b2" dependencies = [ "aho-corasick", "memchr", @@ -5529,9 +5526,9 @@ checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" [[package]] name = "reqsign-aliyun-oss" -version = "3.1.2" +version = "3.1.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9c0f9f69a519dd6958c4b43606bb8e1278cdc76d611fc8fed4b796eee548dc0f" +checksum = "a5e6d659fcdbca6fe2d7ef109c2e28499b7be80501f1bb86c10caf5ec8ac1219" dependencies = [ "anyhow", "form_urlencoded", @@ -5545,12 +5542,11 @@ dependencies = [ ] [[package]] -name = "reqsign-aws-v4" +name = "reqsign-aws-core" version = "3.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cc883bc56889f3e4a419265c87facea222a921debc5c6f15c7fd8b68ec4b36b2" +checksum = "e4af084e1f3cbf3e67e0c972765399bce54ecec804cceba46b39a8331f3c1bff" dependencies = [ - "anyhow", "bytes", "form_urlencoded", "hex", @@ -5566,14 +5562,29 @@ dependencies = [ "sha1 0.11.0", ] +[[package]] +name = "reqsign-aws-v4" +version = "3.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4ac5b3b7cefa28933792b439186459f77f19f9b6edbeab41b8b187150361a206" +dependencies = [ + "bytes", + "http 1.5.0", + "log", + "quick-xml 0.41.0", + "reqsign-aws-core", + "reqsign-core", + "serde", +] + [[package]] name = "reqsign-azure-storage" -version = "3.1.1" +version = "3.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a6ebd8524185ce9c64063e3095f83968acfa90922f00c601a4a0f3aca15b077e" +checksum = "2824e7da3c2cc42ac3406c674eb57c89127fdcd97f3a73c608cfc680505ea134" dependencies = [ "anyhow", - "base64 0.22.1", + "base64 0.23.1", "bytes", "form_urlencoded", "http 1.5.0", @@ -5589,12 +5600,12 @@ dependencies = [ [[package]] name = "reqsign-core" -version = "3.2.0" +version = "3.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7e38b44697c60a823705ccef85cb04d8e0527c9d16ed7c58bf1c6395bdd24ceb" +checksum = "c07dd510b1e1b9b241883e483358147fb2ed2d497a7b39b065ba61eb93deceb0" dependencies = [ "anyhow", - "base64 0.22.1", + "base64 0.23.1", "bytes", "futures", "hex", @@ -5613,9 +5624,9 @@ dependencies = [ [[package]] name = "reqsign-file-read-tokio" -version = "3.0.3" +version = "3.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "688ff0ae421b8d4b92b53fdafaf53df2de28f428a9962edcf21702990b26f74b" +checksum = "663d9d55abd0df0830ef0ae43708297cc1371cf4e8ca91f3ac813c309cca8c98" dependencies = [ "anyhow", "reqsign-core", @@ -5624,9 +5635,9 @@ dependencies = [ [[package]] name = "reqsign-google" -version = "3.0.3" +version = "3.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a96da0b579b846d358090cb06b9e3c2ad1375529efbe3e0c45f96bd7bcf043ea" +checksum = "4080a227f82a09f68540ecd028622065d7ac4c0bcb8727a25bdcfc0526235792" dependencies = [ "form_urlencoded", "http 1.5.0", @@ -6501,17 +6512,6 @@ dependencies = [ "windows-sys 0.61.2", ] -[[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 = "tikv-jemalloc-ctl" version = "0.6.1" @@ -7386,18 +7386,18 @@ dependencies = [ [[package]] name = "zerocopy" -version = "0.8.55" +version = "0.8.56" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b5a105cd7b140f6eeec8acff2ea38135d3cab283ada58540f629fe51e46696eb" +checksum = "556764e583adb45a9f8d413c2a147fa7e8d821e48e12b14fd560b607998b75eb" dependencies = [ "zerocopy-derive", ] [[package]] name = "zerocopy-derive" -version = "0.8.55" +version = "0.8.56" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0fe976fb70c78cd64cccfe3a6fc142244e8a77b70959b30faf9d0ac37ee228eb" +checksum = "f2ab42fc20575779bd240faa45f94a74256f755c0fa9e89f0ede20d91d0cdfc1" dependencies = [ "proc-macro2", "quote", @@ -7466,9 +7466,9 @@ dependencies = [ [[package]] name = "zlib-rs" -version = "0.6.6" +version = "0.6.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b142a20ec14a91d5bc708c1dc21b080c550113d8aa77afa29635673a65dd02c5" +checksum = "34b31d188d9d685a4f9c7b46d6e36631b07058d2cfe190267adce54dc230bf12" [[package]] name = "zmij" diff --git a/native/Cargo.toml b/native/Cargo.toml index a4ad01a3f9..77037b19a7 100644 --- a/native/Cargo.toml +++ b/native/Cargo.toml @@ -34,14 +34,14 @@ edition = "2021" rust-version = "1.88" [workspace.dependencies] -arrow = { version = "58.4.0", features = ["prettyprint", "ffi", "chrono-tz"] } +arrow = { version = "59.1.0", features = ["prettyprint", "ffi", "chrono-tz"] } async-trait = { version = "0.1" } bytes = { version = "1.11.1" } -parquet = { version = "58.4.0", default-features = false, features = ["experimental"] } -datafusion = { version = "54.1.0", default-features = false, features = ["unicode_expressions", "crypto_expressions", "nested_expressions", "parquet"] } -datafusion-datasource = { version = "54.1.0" } -datafusion-physical-expr-adapter = { version = "54.1.0" } -datafusion-spark = { version = "54.1.0", features = ["core"] } +parquet = { version = "59.1.0", default-features = false, features = ["experimental"] } +datafusion = { git = "https://github.com/apache/datafusion.git", rev = "70c26a06716eb552f0fc1d2115957a4089504f11", default-features = false, features = ["unicode_expressions", "crypto_expressions", "nested_expressions", "parquet"] } +datafusion-datasource = { git = "https://github.com/apache/datafusion.git", rev = "70c26a06716eb552f0fc1d2115957a4089504f11" } +datafusion-physical-expr-adapter = { git = "https://github.com/apache/datafusion.git", rev = "70c26a06716eb552f0fc1d2115957a4089504f11" } +datafusion-spark = { git = "https://github.com/apache/datafusion.git", rev = "70c26a06716eb552f0fc1d2115957a4089504f11", features = ["core"] } datafusion-comet-spark-expr = { path = "spark-expr" } datafusion-comet-common = { path = "common" } datafusion-comet-jni-bridge = { path = "jni-bridge" } @@ -58,8 +58,8 @@ object_store = { version = "0.13.2", features = ["gcp", "azure", "aws", "http"] url = "2.2" aws-config = "1.8.18" aws-credential-types = "1.2.13" -iceberg = { git = "https://github.com/apache/iceberg-rust", rev = "3d84c81353b1b23b6e4ae8eea8f8a021cc6927a7" } -iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust", rev = "3d84c81353b1b23b6e4ae8eea8f8a021cc6927a7", features = ["opendal-memory", "opendal-fs", "opendal-s3", "opendal-gcs", "opendal-oss", "opendal-azdls"] } +iceberg = { git = "https://github.com/mbutrovich/iceberg-rust", rev = "07de8fc8d650f7439b251d57a55613f7de693eb8" } +iceberg-storage-opendal = { git = "https://github.com/mbutrovich/iceberg-rust", rev = "07de8fc8d650f7439b251d57a55613f7de693eb8", features = ["opendal-memory", "opendal-fs", "opendal-s3", "opendal-gcs", "opendal-oss", "opendal-azdls"] } reqsign-core = "3" [profile.release] diff --git a/native/core/Cargo.toml b/native/core/Cargo.toml index 7c88b260c3..3cb7374edc 100644 --- a/native/core/Cargo.toml +++ b/native/core/Cargo.toml @@ -69,9 +69,9 @@ aws-config = { workspace = true } aws-credential-types = { workspace = true } parking_lot = "0.12.5" reqwest = { version = "0.12", default-features = false, features = ["rustls-tls-native-roots", "http2"] } -object_store_opendal = { version = "0.57.0", optional = true } +object_store_opendal = { version = "0.58.0", optional = true } hdfs-sys = {version = "0.3", optional = true, features = ["hdfs_3_3"]} -opendal = { version = "0.57.0", optional = true, features = ["services-hdfs"] } +opendal = { version = "0.58.1", optional = true, features = ["services-hdfs"] } iceberg = { workspace = true } iceberg-storage-opendal = { workspace = true } reqsign-core = { workspace = true } @@ -91,7 +91,7 @@ jni = { version = "0.22.4", features = ["invocation"] } lazy_static = "1.4" assertables = "10" hex = "0.4.3" -datafusion-functions-nested = { version = "54.1.0" } +datafusion-functions-nested = { git = "https://github.com/apache/datafusion.git", rev = "70c26a06716eb552f0fc1d2115957a4089504f11" } [features] backtrace = ["datafusion/backtrace"] diff --git a/native/core/src/execution/columnar_to_row.rs b/native/core/src/execution/columnar_to_row.rs index a1321a9def..74defde672 100644 --- a/native/core/src/execution/columnar_to_row.rs +++ b/native/core/src/execution/columnar_to_row.rs @@ -2491,11 +2491,14 @@ mod tests { let schema = vec![DataType::FixedSizeBinary(3)]; let mut ctx = ColumnarToRowContext::new(schema, 100); - let array: ArrayRef = Arc::new(FixedSizeBinaryArray::from(vec![ - Some(&[1u8, 2, 3][..]), - Some(&[4u8, 5, 6][..]), - None, // Test null handling - ])); + let array: ArrayRef = Arc::new( + FixedSizeBinaryArray::try_from(vec![ + Some(&[1u8, 2, 3][..]), + Some(&[4u8, 5, 6][..]), + None, // Test null handling + ]) + .unwrap(), + ); let arrays = vec![array]; let (ptr, offsets, lengths) = ctx.convert(&arrays, 3).unwrap(); diff --git a/native/core/src/execution/merge_as_partial.rs b/native/core/src/execution/merge_as_partial.rs index 5ea26115bf..3ad3492e1e 100644 --- a/native/core/src/execution/merge_as_partial.rs +++ b/native/core/src/execution/merge_as_partial.rs @@ -210,20 +210,30 @@ impl GroupsAccumulator for MergeAsPartialGroupsAccumulator { opt_filter: Option<&BooleanArray>, total_num_groups: usize, ) -> Result<()> { - // Redirect update to merge — this is the key trick. + // Redirect update to merge — this is the key trick. Spark's PartialMerge mode + // never applies a filter (filters apply once, at the Partial stage), so there's + // nothing to forward into merge_batch, which no longer accepts one. + debug_assert!(opt_filter.is_none()); self.inner - .merge_batch(values, group_indices, opt_filter, total_num_groups) + .merge_batch(values, group_indices, total_num_groups) } fn merge_batch( &mut self, values: &[ArrayRef], group_indices: &[usize], - opt_filter: Option<&BooleanArray>, total_num_groups: usize, ) -> Result<()> { self.inner - .merge_batch(values, group_indices, opt_filter, total_num_groups) + .merge_batch(values, group_indices, total_num_groups) + } + + fn convert_to_state( + &self, + _values: &[ArrayRef], + _opt_filter: Option<&BooleanArray>, + ) -> Result> { + datafusion::common::not_impl_err!("Input batch conversion to state not implemented") } fn evaluate(&mut self, emit_to: EmitTo) -> Result { diff --git a/native/core/src/execution/operators/copy.rs b/native/core/src/execution/operators/copy.rs index 193c385c3c..4677412622 100644 --- a/native/core/src/execution/operators/copy.rs +++ b/native/core/src/execution/operators/copy.rs @@ -37,7 +37,9 @@ pub(crate) fn copy_array(array: &dyn Array) -> ArrayRef { let mut mutable = MutableArrayData::new(vec![&data], false, capacity); - mutable.extend(0, 0, capacity); + mutable + .try_extend(0, 0, capacity) + .expect("extend failed due to offset overflow"); if matches!(array.data_type(), DataType::Dictionary(_, _)) { let copied_dict = make_array(mutable.freeze()); @@ -50,7 +52,9 @@ pub(crate) fn copy_array(array: &dyn Array) -> ArrayRef { let data = values.to_data(); let mut mutable = MutableArrayData::new(vec![&data], false, values.len()); - mutable.extend(0, 0, values.len()); + mutable + .try_extend(0, 0, values.len()) + .expect("extend failed due to offset overflow"); let copied_dict = ref_copied_dict.with_values(make_array(mutable.freeze())); Arc::new(copied_dict) diff --git a/native/core/src/execution/operators/iceberg_scan.rs b/native/core/src/execution/operators/iceberg_scan.rs index e727294fd9..4dca1a3298 100644 --- a/native/core/src/execution/operators/iceberg_scan.rs +++ b/native/core/src/execution/operators/iceberg_scan.rs @@ -651,6 +651,8 @@ mod tests { start: 0, length: 0, record_count: None, + first_row_id: None, + data_sequence_number: None, data_file_path: "data.parquet".to_string(), data_file_format: DataFileFormat::Parquet, schema: Arc::new(Schema::builder().build().unwrap()), diff --git a/native/core/src/execution/operators/parquet_writer.rs b/native/core/src/execution/operators/parquet_writer.rs index dbbee713ae..ca91d78744 100644 --- a/native/core/src/execution/operators/parquet_writer.rs +++ b/native/core/src/execution/operators/parquet_writer.rs @@ -92,7 +92,7 @@ enum ParquetWriter { /// The opendal::Writer is created lazily on first write #[cfg(feature = "hdfs-opendal")] Remote( - ArrowWriter>>, + Box>>>, Option, Operator, String, @@ -338,7 +338,7 @@ impl ParquetWriterExec { // HDFS writer will be created lazily on first write // Use the path from prepare_object_store_with_configs Ok(ParquetWriter::Remote( - arrow_parquet_buffer_writer, + Box::new(arrow_parquet_buffer_writer), None, op, object_store_path.to_string(), @@ -656,11 +656,9 @@ mod tests { // Create OpenDAL HDFS operator let builder = Hdfs::default().name_node(namenode); - let op = Operator::new(builder) - .map_err(|e| { - DataFusionError::Execution(format!("Failed to create HDFS operator: {}", e)) - })? - .finish(); + let op = Operator::new(builder).map_err(|e| { + DataFusionError::Execution(format!("Failed to create HDFS operator: {}", e)) + })?; let mut hdfs_writer = op.writer(output_path).await.map_err(|e| { DataFusionError::Execution(format!("Failed to create HDFS writer: {}", e)) @@ -707,11 +705,9 @@ mod tests { // Create OpenDAL HDFS operator let builder = Hdfs::default().name_node(namenode); - let op = Operator::new(builder) - .map_err(|e| { - DataFusionError::Execution(format!("Failed to create HDFS operator: {}", e)) - })? - .finish(); + let op = Operator::new(builder).map_err(|e| { + DataFusionError::Execution(format!("Failed to create HDFS operator: {}", e)) + })?; // Create a single HDFS writer for the entire file let mut hdfs_writer = op.writer(output_path).await.map_err(|e| { diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index 85851b17ca..3a9d4707db 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -111,7 +111,7 @@ use arrow::array::{ use arrow::buffer::{BooleanBuffer, NullBuffer, OffsetBuffer}; use arrow::row::{OwnedRow, RowConverter, SortField}; use datafusion::common::utils::SingleRowListArrayBuilder; -use datafusion::common::UnnestOptions; +use datafusion::common::{NullHandling, UnnestOptions}; use datafusion::physical_plan::filter::FilterExec; use datafusion::physical_plan::joins::NestedLoopJoinExec; use datafusion::physical_plan::limit::GlobalLimitExec; @@ -2069,7 +2069,11 @@ impl PhysicalPlanner { }); let unnest_options = UnnestOptions { - preserve_nulls: explode.outer, + null_handling: if explode.outer { + NullHandling::Preserve + } else { + NullHandling::Drop + }, recursions: vec![], }; @@ -4139,6 +4143,8 @@ fn parse_file_scan_tasks_from_common( start: proto_task.start, length: proto_task.length, record_count: proto_task.record_count, + first_row_id: None, + data_sequence_number: None, data_file_format, schema: schema_ref, project_field_ids, diff --git a/native/core/src/parquet/eager_page_index_reader_factory.rs b/native/core/src/parquet/eager_page_index_reader_factory.rs index 278814c4bf..d3c854093c 100644 --- a/native/core/src/parquet/eager_page_index_reader_factory.rs +++ b/native/core/src/parquet/eager_page_index_reader_factory.rs @@ -68,11 +68,11 @@ use std::sync::Arc; #[derive(Debug)] pub struct EagerPageIndexReaderFactory { store: Arc, - metadata_cache: Arc, + metadata_cache: Arc, } impl EagerPageIndexReaderFactory { - pub fn new(store: Arc, metadata_cache: Arc) -> Self { + pub fn new(store: Arc, metadata_cache: Arc) -> Self { Self { store, metadata_cache, @@ -118,7 +118,7 @@ struct EagerPageIndexReader { store: Arc, inner: ParquetObjectReader, partitioned_file: PartitionedFile, - metadata_cache: Arc, + metadata_cache: Arc, metadata_size_hint: Option, } diff --git a/native/core/src/parquet/parquet_exec.rs b/native/core/src/parquet/parquet_exec.rs index 1308ce97fc..966e64db57 100644 --- a/native/core/src/parquet/parquet_exec.rs +++ b/native/core/src/parquet/parquet_exec.rs @@ -20,7 +20,7 @@ use crate::parquet::eager_page_index_reader_factory::EagerPageIndexReaderFactory use crate::parquet::encryption_support::{CometEncryptionConfig, ENCRYPTION_FACTORY_ID}; use crate::parquet::parquet_support::SparkParquetOptions; use crate::parquet::schema_adapter::SparkPhysicalExprAdapterFactory; -use arrow::datatypes::{Field, SchemaRef}; +use arrow::datatypes::{Field, FieldRef, SchemaRef}; use datafusion::config::{ParquetOptions, TableParquetOptions}; use datafusion::datasource::listing::PartitionedFile; use datafusion::datasource::physical_plan::{ @@ -128,13 +128,14 @@ pub(crate) fn init_datasource_exec( } _ => (Arc::clone(&required_schema), None), }; - let partition_fields: Vec<_> = partition_schema + let partition_fields: Vec = partition_schema .iter() .flat_map(|s| s.fields().iter()) - .map(|f| Arc::new(Field::new(f.name(), f.data_type().clone(), f.is_nullable())) as _) + .map(|f| Arc::new(Field::new(f.name(), f.data_type().clone(), f.is_nullable()))) .collect(); - let table_schema = - TableSchema::from_file_schema(base_schema).with_table_partition_cols(partition_fields); + let table_schema = TableSchema::builder(base_schema) + .with_table_partition_cols(partition_fields) + .build(); let mut parquet_source = ParquetSource::new(table_schema) .with_table_parquet_options(table_parquet_options) diff --git a/native/core/src/parquet/parquet_support.rs b/native/core/src/parquet/parquet_support.rs index 2ee1230ed8..56176b1637 100644 --- a/native/core/src/parquet/parquet_support.rs +++ b/native/core/src/parquet/parquet_support.rs @@ -445,12 +445,10 @@ pub(crate) fn create_hdfs_operator(url: &Url) -> Result bool { schema.fields().iter().any(|f| parse_field_id(f).is_some()) } +/// Returns true when casting `physical_type` to `target_type` is a *pure* structural +/// narrowing (dropping unrequested struct/list fields, no leaf-level value reinterpretation) +/// that DataFusion's own `datafusion_common::nested_struct::cast_column` already computes +/// identically to Comet's `spark_parquet_convert`. `ColumnarValue::cast_to` (the function a +/// plain, un-swapped `CastExpr` runs at execution time; see +/// datafusion/expr-common/src/columnar_value.rs) routes to that same function whenever +/// `datafusion_common::nested_struct::requires_nested_struct_cast` holds, matching arbitrary +/// struct/list fields by name, null-filling missing target fields, and dropping extra source +/// fields, exactly the shape apache/datafusion-comet#4859 needs pruned. Confirmed +/// byte-identical to `spark_parquet_convert` for the covered shapes by +/// `test::nested_struct_narrowing_cast_matches_datafusion_generic_cast`. +/// +/// When this returns `true`, `replace_with_spark_cast` leaves DataFusion's `CastExpr` in +/// place instead of swapping in `CometCastColumnExpr`, so DataFusion's leaf-pruning +/// (`build_projection_read_plan`'s cast-clipping, apache/datafusion#24090) can see the cast +/// and read only the requested Parquet leaves, instead of falling back to a full-column read +/// because it can't recognize `CometCastColumnExpr`. +/// +/// This is deliberately an allow list, not a deny list: it only recurses through the two +/// container shapes `nested_struct::cast_column` actually implements (Struct, List / +/// LargeList), and requires every leaf it bottoms out at to be an *exact* type match. Pruning +/// (the only case this predicate needs to cover, see apache/datafusion-comet#4859) only +/// changes which struct/list fields are kept, never a leaf's type, so exact-match leaves are +/// sufficient. A deny list here (enumerate every case where Comet's nested cast differs from +/// Arrow's, allow everything else) would fail open: a future addition to +/// `parquet_convert_array` that this predicate does not know to also exclude would silently +/// start producing wrong results instead of just missing an optimization. +fn is_pure_structural_narrowing( + physical_type: &DataType, + target_type: &DataType, + parquet_options: &SparkParquetOptions, +) -> bool { + match (physical_type, target_type) { + (DataType::Struct(source_fields), DataType::Struct(target_fields)) => { + // Comet matches by Parquet field id first when the target carries one; + // DataFusion's generic cast has no field-id concept, so any field-id-bearing + // target field is a potential divergence. + if parquet_options.use_field_id + && target_fields.iter().any(|f| parse_field_id(f).is_some()) + { + return false; + } + target_fields.iter().all(|target_field| { + // Require an *exact* (case-sensitive) name match for every target field. + // `nested_struct::cast_column` always matches by exact name; Comet + // additionally matches case-insensitively when `case_sensitive` is false, + // which could resolve a field DataFusion would instead treat as missing (and + // null-fill). Requiring an exact match sidesteps that divergence regardless + // of the `case_sensitive` setting, and also sidesteps the missing-field + // nullability divergence: DataFusion errors when a non-nullable target field + // is missing from the source, whereas Comet null-fills unconditionally. + source_fields + .iter() + .find(|f| f.name() == target_field.name()) + .is_some_and(|source_field| { + is_pure_structural_narrowing( + source_field.data_type(), + target_field.data_type(), + parquet_options, + ) + }) + }) + } + (DataType::List(source_item), DataType::List(target_item)) + | (DataType::LargeList(source_item), DataType::LargeList(target_item)) => { + is_pure_structural_narrowing( + source_item.data_type(), + target_item.data_type(), + parquet_options, + ) + } + // Map is excluded structurally, not by the equality check below: `replace_with_spark_cast` + // only reaches this predicate after its own top-level `physical_type == target_type` + // check, but `is_pure_structural_narrowing` is tested (and may be called) independently + // of that guard, so an equal Map must not slip through here either. + (DataType::Map(_, _), _) | (_, DataType::Map(_, _)) => false, + // Every other shape, including Dictionary and any leaf-level type change (timestamp + // tz relabeling, `nanosAsLong`, decimal/numeric promotion, and so on), must be an + // exact match. Comet's nested cast handling for those shapes carries Spark-specific + // value or matching semantics that `nested_struct::cast_column` does not replicate, + // and none of them arise from pruning alone. + _ => physical_type == target_type, + } +} + /// Remap physical schema field names to match logical schema field names. Mirrors Spark's /// `clipParquetGroupFields`: prefer ID match for any logical field that carries a /// `PARQUET:field_id`, fall back to case-insensitive name match otherwise. @@ -878,6 +964,39 @@ impl SparkPhysicalExprAdapter { )); } + // Leave DataFusion's `CastExpr` in place for a pure structural narrowing instead + // of swapping in `CometCastColumnExpr` (see `is_pure_structural_narrowing`'s doc + // comment for why that's safe and apache/datafusion-comet#4859 for why it matters). + // + // `ColumnarValue::cast_to` does not always route a `true` result here through + // `nested_struct::cast_column`: for a List wrapping a Struct-free, leaf-unchanged + // interior (e.g. `List(Int32)` differing only in the item field's name or + // nullability, the "read simple ARRAY fields" shape), + // `nested_struct::requires_nested_struct_cast` is false and it falls to Arrow's + // plain `cast_with_options` instead. Both dispatch targets independently implement + // the same "recast values, rewrap with the target field" pattern for List, so they + // agree regardless of which one runs; confirmed by + // `test::list_of_unchanged_leaf_with_differing_item_metadata_matches_columnar_value_cast_to` + // alongside `test::nested_struct_narrowing_cast_matches_datafusion_generic_cast` + // for the Struct-pruning shape that always does route through + // `nested_struct::cast_column`. + if is_pure_structural_narrowing(physical_type, target_type, &self.parquet_options) { + // Not an assertion that `nested_struct::cast_column` specifically runs (it may + // not, see above), only that *some* DataFusion cast path can actually perform + // this pair, so a bug in the predicate surfaces here at plan time instead of + // as an opaque `ArrowError` deep inside a parquet read. + debug_assert!( + datafusion::common::nested_struct::requires_nested_struct_cast( + physical_type, + target_type + ) || can_cast_types(physical_type, target_type), + "is_pure_structural_narrowing({physical_type:?}, {target_type:?}) returned \ + true but neither nested_struct::cast_column nor Arrow's plain cast can \ + perform this pair" + ); + return Ok(Transformed::no(expr)); + } + // Same-shape complex casts, timestamp tz relabel (e.g. Timestamp(us, None) // -> Timestamp(us, Some("UTC")) for INT96 reads), and Timestamp -> Int64 // (Spark's `nanosAsLong`) need spark_parquet_convert: it handles nested @@ -1102,7 +1221,9 @@ impl PhysicalExpr for RejectOnNonEmpty { #[cfg(test)] mod test { use crate::parquet::parquet_support::SparkParquetOptions; - use crate::parquet::schema_adapter::SparkPhysicalExprAdapterFactory; + use crate::parquet::schema_adapter::{ + is_pure_structural_narrowing, SparkPhysicalExprAdapterFactory, + }; use arrow::array::UInt32Array; use arrow::array::{ BinaryArray, Date32Array, Decimal128Array, Float32Array, Float64Array, Int32Array, @@ -1117,12 +1238,13 @@ mod test { use datafusion::datasource::source::DataSourceExec; use datafusion::execution::object_store::ObjectStoreUrl; use datafusion::execution::TaskContext; + use datafusion::physical_expr::PhysicalExpr; use datafusion::physical_plan::ExecutionPlan; use datafusion_comet_spark_expr::test_common::file_util::get_temp_filename; use datafusion_comet_spark_expr::EvalMode; use datafusion_physical_expr_adapter::PhysicalExprAdapterFactory; use futures::StreamExt; - use parquet::arrow::ArrowWriter; + use parquet::arrow::{ArrowWriter, PARQUET_FIELD_ID_META_KEY}; use std::fs::File; use std::sync::Arc; @@ -1775,4 +1897,444 @@ mod test { "Expected duplicate field error, got: {err_msg}" ); } + + /// #4859 investigation: for a pure structural narrowing of a nested column (dropping + /// unrequested struct fields, no leaf-type promotion, no tz relabeling, no field-id + /// matching, case-sensitive), does Comet's `CometCastColumnExpr` (via + /// `spark_parquet_convert`) produce the same result as DataFusion's own + /// `datafusion_common::nested_struct::cast_column`, which is what a plain, un-swapped + /// `CastExpr` would run at execution time (see `ColumnarValue::cast_to` -> + /// `cast_array_by_name` in datafusion/expr-common/src/columnar_value.rs)? If so, that + /// class of cast could be left as DataFusion's own `CastExpr` (like the identity-cast + /// unwrap in #4730) so DataFusion's #24090 leaf-pruning can see it. + #[test] + fn nested_struct_narrowing_cast_matches_datafusion_generic_cast() -> Result<(), DataFusionError> + { + use crate::parquet::parquet_support::spark_parquet_convert; + use arrow::array::{ListArray, StructArray}; + use arrow::buffer::OffsetBuffer; + use arrow::compute::CastOptions; + use arrow::datatypes::Fields; + use datafusion::physical_plan::ColumnarValue; + + // events: array> -> array>, + // mirroring the physical vs. logical (Spark-pruned) schema in the #4859 repro. + let id_array: Arc = + Arc::new(Int64Array::from(vec![1i64, 2, 3, 4])); + let payload_array: Arc = + Arc::new(StringArray::from(vec!["a", "b", "c", "d"])); + let from_struct_fields: Fields = vec![ + Field::new("id", DataType::Int64, true), + Field::new("payload", DataType::Utf8, true), + ] + .into(); + let struct_array = StructArray::new( + from_struct_fields.clone(), + vec![id_array, payload_array], + None, + ); + let from_item_field = Arc::new(Field::new( + "element", + DataType::Struct(from_struct_fields), + true, + )); + let offsets = OffsetBuffer::new(vec![0, 2, 4].into()); + let list_array: Arc = Arc::new(ListArray::new( + Arc::clone(&from_item_field), + offsets, + Arc::new(struct_array), + None, + )); + + let to_struct_fields: Fields = vec![Field::new("id", DataType::Int64, true)].into(); + let to_item_field = Arc::new(Field::new( + "element", + DataType::Struct(to_struct_fields), + true, + )); + let target_type = DataType::List(Arc::clone(&to_item_field)); + + let spark_parquet_options = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false); + let comet_result = spark_parquet_convert( + ColumnarValue::Array(Arc::clone(&list_array)), + &target_type, + &spark_parquet_options, + )?; + let comet_array = match comet_result { + ColumnarValue::Array(a) => a, + _ => panic!("expected array"), + }; + + let df_array = datafusion::common::nested_struct::cast_column( + &list_array, + &target_type, + &CastOptions::default(), + )?; + + assert_eq!( + format!("{comet_array:?}"), + format!("{df_array:?}"), + "Comet's CometCastColumnExpr / spark_parquet_convert and DataFusion's own \ + nested_struct::cast_column disagree for pure struct-narrowing" + ); + Ok(()) + } + + /// #4859 follow-up: `List(Int32)` with an unchanged leaf type but differing item-field + /// metadata (Parquet's inferred "element" name, non-nullable vs. Spark's generic "item" + /// name, nullable) does NOT trigger `requires_nested_struct_cast` (the item type itself, + /// Int32, is not Struct/List), so `ColumnarValue::cast_to` (what a plain `CastExpr` runs + /// at execution time) takes the *other* branch: Arrow's plain `cast_with_options`, not + /// `nested_struct::cast_column`. This is the exact shape from the "read simple ARRAY + /// fields" CometNativeReaderSuite failure. Compares against `ColumnarValue::cast_to` + /// directly (the real call site), not `nested_struct::cast_column`, since that is what + /// actually executes here. + #[test] + fn list_of_unchanged_leaf_with_differing_item_metadata_matches_columnar_value_cast_to( + ) -> Result<(), DataFusionError> { + use crate::parquet::parquet_support::spark_parquet_convert; + use arrow::array::{Int32Array, ListArray}; + use arrow::buffer::OffsetBuffer; + use datafusion::physical_plan::ColumnarValue; + + let values: Arc = Arc::new(Int32Array::from(vec![1, 2, 3, 4])); + let physical_item_field = Arc::new(Field::new("element", DataType::Int32, false)); + let offsets = OffsetBuffer::new(vec![0, 2, 4].into()); + let list_array: Arc = Arc::new(ListArray::new( + Arc::clone(&physical_item_field), + offsets, + values, + None, + )); + + let target_item_field = Arc::new(Field::new("item", DataType::Int32, true)); + let target_type = DataType::List(Arc::clone(&target_item_field)); + + let spark_parquet_options = default_options(); + let comet_result = spark_parquet_convert( + ColumnarValue::Array(Arc::clone(&list_array)), + &target_type, + &spark_parquet_options, + )?; + let comet_array = match comet_result { + ColumnarValue::Array(a) => a, + _ => panic!("expected array"), + }; + + let df_result = + ColumnarValue::Array(Arc::clone(&list_array)).cast_to(&target_type, None)?; + let df_array = match df_result { + ColumnarValue::Array(a) => a, + _ => panic!("expected array"), + }; + + assert_eq!( + format!("{comet_array:?}"), + format!("{df_array:?}"), + "Comet's spark_parquet_convert and DataFusion's ColumnarValue::cast_to disagree \ + for a List whose leaf type is unchanged but whose item field metadata differs" + ); + Ok(()) + } + + fn struct_type(fields: Vec<(&str, DataType)>) -> DataType { + DataType::Struct( + fields + .into_iter() + .map(|(name, dt)| Field::new(name, dt, true)) + .collect(), + ) + } + + fn struct_type_with_field_id(fields: Vec<(&str, DataType, i32)>) -> DataType { + DataType::Struct( + fields + .into_iter() + .map(|(name, dt, id)| { + Field::new(name, dt, true).with_metadata(std::collections::HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + id.to_string(), + )])) + }) + .collect(), + ) + } + + fn list_type(item: DataType) -> DataType { + DataType::List(Arc::new(Field::new("element", item, true))) + } + + fn default_options() -> SparkParquetOptions { + SparkParquetOptions::new(EvalMode::Legacy, "UTC", false) + } + + /// Dropping a struct field by exact name, including through nested struct-in-struct and + /// list-of-struct, is the actual #4859 pruning shape and must be allowed. + #[test] + fn is_pure_structural_narrowing_allows_struct_and_list_field_drop() { + let opts = default_options(); + + let physical = struct_type(vec![("id", DataType::Int64), ("payload", DataType::Utf8)]); + let target = struct_type(vec![("id", DataType::Int64)]); + assert!(is_pure_structural_narrowing(&physical, &target, &opts)); + + let physical = list_type(struct_type(vec![ + ("id", DataType::Int64), + ( + "inner", + struct_type(vec![("a", DataType::Int64), ("blob", DataType::Utf8)]), + ), + ])); + let target = list_type(struct_type(vec![( + "inner", + struct_type(vec![("a", DataType::Int64)]), + )])); + assert!(is_pure_structural_narrowing(&physical, &target, &opts)); + } + + /// A target field with no exact-name match in the source (here, only a case-insensitive + /// one) must be denied: `nested_struct::cast_column` always matches by exact name and + /// would null-fill this field instead of resolving it the way Comet's case-insensitive + /// matching does. + #[test] + fn is_pure_structural_narrowing_denies_case_insensitive_only_match() { + let opts = default_options(); + let physical = struct_type(vec![("ID", DataType::Int64)]); + let target = struct_type(vec![("id", DataType::Int64)]); + assert!(!is_pure_structural_narrowing(&physical, &target, &opts)); + } + + /// A target field with no name match at all must be denied: DataFusion's + /// `nested_struct::cast_column` null-fills it unconditionally, while Comet's behavior + /// additionally depends on `return_null_struct_if_all_fields_missing` at the struct level. + #[test] + fn is_pure_structural_narrowing_denies_missing_target_field() { + let opts = default_options(); + let physical = struct_type(vec![("id", DataType::Int64)]); + let target = struct_type(vec![("id", DataType::Int64), ("payload", DataType::Utf8)]); + assert!(!is_pure_structural_narrowing(&physical, &target, &opts)); + } + + /// When `use_field_id` is set and the target struct carries Parquet field ids, Comet + /// matches by id first; DataFusion's generic cast has no field-id concept at all, so this + /// must be denied even though the names also happen to match. + #[test] + fn is_pure_structural_narrowing_denies_field_id_matching() { + let mut opts = default_options(); + opts.use_field_id = true; + let physical = struct_type_with_field_id(vec![("id", DataType::Int64, 1)]); + let target = struct_type_with_field_id(vec![("id", DataType::Int64, 1)]); + assert!(!is_pure_structural_narrowing(&physical, &target, &opts)); + } + + /// Map value narrowing has no equivalent in `nested_struct::cast_column` (it has no Map + /// arm), so it must stay on the `CometCastColumnExpr` path regardless of nesting depth. + #[test] + fn is_pure_structural_narrowing_denies_map() { + let opts = default_options(); + let physical = DataType::Map( + Arc::new(Field::new( + "entries", + struct_type(vec![("k", DataType::Int64), ("v", DataType::Utf8)]), + false, + )), + false, + ); + let target = physical.clone(); + // Even a no-op Map "narrowing" (target == physical) must not be routed through this + // predicate; Map is excluded structurally, not by an equality shortcut. + assert!(!is_pure_structural_narrowing(&physical, &target, &opts)); + } + + /// A leaf-level type change (timestamp tz relabeling, `nanosAsLong`, decimal promotion, + /// and so on) never arises from pruning alone and must be denied: Comet's nested cast + /// handling for those shapes carries Spark-specific semantics `nested_struct::cast_column` + /// does not replicate. + #[test] + fn is_pure_structural_narrowing_denies_leaf_type_change() { + let opts = default_options(); + let physical = struct_type(vec![("id", DataType::Int32)]); + let target = struct_type(vec![("id", DataType::Int64)]); + assert!(!is_pure_structural_narrowing(&physical, &target, &opts)); + } + + /// NTZ -> LTZ timestamp relabeling (INT96 reads) is a metadata-only reinterpretation + /// Comet applies via a raw `reinterpret_cast`; Arrow's generic timestamp cast performs an + /// actual timezone conversion instead, changing the value. Locked in as its own test + /// because it is the subtlest denial: both sides are "just" Timestamp, so it is exactly + /// the shape a future refactor might accidentally fold into the equality catch-all. + #[test] + fn is_pure_structural_narrowing_denies_timestamp_ntz_to_ltz_relabel() { + use arrow::datatypes::TimeUnit; + let opts = default_options(); + let physical = struct_type(vec![( + "ts", + DataType::Timestamp(TimeUnit::Microsecond, None), + )]); + let target = struct_type(vec![( + "ts", + DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())), + )]); + assert!(!is_pure_structural_narrowing(&physical, &target, &opts)); + } + + /// Dictionary-encoded columns get Comet's own dictionary-preserving or + /// dictionary-flattening handling; Arrow's generic cast does not special-case this, so a + /// Dictionary value narrowing must be denied like Map, Timestamp relabeling, and any other + /// leaf-level divergence. + #[test] + fn is_pure_structural_narrowing_denies_dictionary() { + let opts = default_options(); + let physical = struct_type(vec![( + "d", + DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8)), + )]); + let target = struct_type(vec![( + "d", + DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::LargeUtf8)), + )]); + assert!(!is_pure_structural_narrowing(&physical, &target, &opts)); + } + + /// Build a `SparkPhysicalExprAdapter` over a single "events" column and run the real + /// `PhysicalExprAdapter::rewrite` pipeline on a plain `Column` reference to it, mirroring + /// exactly what DataFusion's parquet opener does with the un-adapted projection + /// (`opener/mod.rs`'s `rewriter.rewrite(p)`). Returns the rewritten expression so callers + /// can assert on its concrete type. + fn rewrite_events_column( + physical_type: DataType, + target_type: DataType, + opts: SparkParquetOptions, + ) -> Result, DataFusionError> { + use datafusion::physical_expr::expressions::Column; + + let physical_schema = + Arc::new(Schema::new(vec![Field::new("events", physical_type, true)])); + let logical_schema = Arc::new(Schema::new(vec![Field::new("events", target_type, true)])); + + let factory = SparkPhysicalExprAdapterFactory::new(opts, None); + let adapter = factory.create(Arc::clone(&logical_schema), Arc::clone(&physical_schema))?; + + let col_expr: Arc = Arc::new(Column::new("events", 0)); + adapter.rewrite(col_expr) + } + + /// End-to-end check of the actual wiring, not just the predicate: routing a pure + /// structural narrowing through the real `PhysicalExprAdapter::rewrite` pipeline (the + /// same path DataFusion's opener calls with the plain, un-adapted projection Column) must + /// leave DataFusion's own `CastExpr` in place rather than swapping in + /// `CometCastColumnExpr`, or DataFusion's leaf-pruning (#24090) would never see it. + #[test] + fn replace_with_spark_cast_preserves_cast_expr_for_pure_narrowing( + ) -> Result<(), DataFusionError> { + use crate::parquet::cast_column::CometCastColumnExpr; + use datafusion::physical_expr::expressions::CastExpr; + + let physical_type = struct_type(vec![("id", DataType::Int64), ("payload", DataType::Utf8)]); + let target_type = struct_type(vec![("id", DataType::Int64)]); + let rewritten = rewrite_events_column(physical_type, target_type, default_options())?; + + assert!( + rewritten.downcast_ref::().is_some(), + "expected DataFusion's own CastExpr to survive a pure structural narrowing, got: \ + {rewritten}" + ); + assert!( + rewritten.downcast_ref::().is_none(), + "CometCastColumnExpr should not appear for a pure structural narrowing" + ); + Ok(()) + } + + /// Companion to the previous test: a narrowing the predicate must deny (here, a + /// case-insensitive-only field match) has to still get `CometCastColumnExpr`'s correctness + /// handling through the real pipeline, not just fall through untouched. + #[test] + fn replace_with_spark_cast_wraps_cast_expr_when_not_pure_narrowing( + ) -> Result<(), DataFusionError> { + use crate::parquet::cast_column::CometCastColumnExpr; + + let physical_type = struct_type(vec![("ID", DataType::Int64)]); + let target_type = struct_type(vec![("id", DataType::Int64)]); + let mut opts = default_options(); + opts.case_sensitive = false; + let rewritten = rewrite_events_column(physical_type, target_type, opts)?; + + assert!( + rewritten.downcast_ref::().is_some(), + "expected CometCastColumnExpr for a case-insensitive-only match, got: {rewritten}" + ); + Ok(()) + } + + /// `use_field_id` being enabled session-wide must not deny a struct whose fields simply + /// don't carry Parquet field-id metadata (e.g. a plain Parquet file read with Iceberg + /// field-id matching turned on for the table in general). The field-id bail must key off + /// whether *this* struct's fields actually carry ids, not the config flag alone. + #[test] + fn is_pure_structural_narrowing_allows_use_field_id_enabled_without_field_id_metadata() { + let mut opts = default_options(); + opts.use_field_id = true; + let physical = struct_type(vec![("id", DataType::Int64), ("payload", DataType::Utf8)]); + let target = struct_type(vec![("id", DataType::Int64)]); + assert!(is_pure_structural_narrowing(&physical, &target, &opts)); + } + + /// The field-id bail must apply at every nesting level it is reached, not just the + /// outermost struct: an inner struct carrying field ids must deny the whole cast even + /// though the outer struct matches cleanly by name. + #[test] + fn is_pure_structural_narrowing_denies_field_id_matching_at_nested_level() { + let mut opts = default_options(); + opts.use_field_id = true; + let physical = struct_type(vec![( + "outer", + struct_type_with_field_id(vec![("id", DataType::Int64, 1)]), + )]); + let target = struct_type(vec![( + "outer", + struct_type_with_field_id(vec![("id", DataType::Int64, 1)]), + )]); + assert!(!is_pure_structural_narrowing(&physical, &target, &opts)); + } + + /// A target struct with zero field-name overlap against the source is denied, distinct + /// from (and in addition to) the partial-overlap case in + /// `is_pure_structural_narrowing_denies_missing_target_field`: DataFusion's + /// `validate_struct_compatibility` hard-errors on zero overlap, while Comet null-fills + /// every field, governed by `return_null_struct_if_all_fields_missing`. + #[test] + fn is_pure_structural_narrowing_denies_zero_field_overlap() { + let opts = default_options(); + let physical = struct_type(vec![("left", DataType::Int64)]); + let target = struct_type(vec![("right", DataType::Int64)]); + assert!(!is_pure_structural_narrowing(&physical, &target, &opts)); + } + + /// `case_sensitive = true` with an exact-case match must still be allowed: the predicate's + /// exact-name requirement does not depend on the `case_sensitive` flag's value, only on + /// whether the names actually match exactly. + #[test] + fn is_pure_structural_narrowing_allows_case_sensitive_true_with_exact_match() { + let mut opts = default_options(); + opts.case_sensitive = true; + let physical = struct_type(vec![("id", DataType::Int64), ("payload", DataType::Utf8)]); + let target = struct_type(vec![("id", DataType::Int64)]); + assert!(is_pure_structural_narrowing(&physical, &target, &opts)); + } + + /// `case_sensitive = true` with a case-differing name is denied, same as the + /// `case_sensitive = false` case. This is conservative rather than strictly necessary: with + /// `case_sensitive = true`, Comet's own matching would also fail to match "ID" to "id" and + /// null-fill it, agreeing with DataFusion. Denying here only costs a missed optimization, + /// never a wrong result, and keeps the predicate's exact-match rule uniform regardless of + /// `case_sensitive`. + #[test] + fn is_pure_structural_narrowing_denies_case_mismatch_even_when_case_sensitive_true() { + let mut opts = default_options(); + opts.case_sensitive = true; + let physical = struct_type(vec![("ID", DataType::Int64)]); + let target = struct_type(vec![("id", DataType::Int64)]); + assert!(!is_pure_structural_narrowing(&physical, &target, &opts)); + } } diff --git a/native/shuffle/src/partitioners/traits.rs b/native/shuffle/src/partitioners/traits.rs index 9572b70db5..5bd41a62a5 100644 --- a/native/shuffle/src/partitioners/traits.rs +++ b/native/shuffle/src/partitioners/traits.rs @@ -19,7 +19,7 @@ use arrow::record_batch::RecordBatch; use datafusion::common::Result; #[async_trait::async_trait] -pub(crate) trait ShufflePartitioner: Send + Sync { +pub(crate) trait ShufflePartitioner: Send { /// Insert a batch into the partitioner async fn insert_batch(&mut self, batch: RecordBatch) -> Result<()>; /// Write shuffle data and shuffle index file to disk diff --git a/native/shuffle/src/writers/local/spill.rs b/native/shuffle/src/writers/local/spill.rs index 450b617421..7304f5d19d 100644 --- a/native/shuffle/src/writers/local/spill.rs +++ b/native/shuffle/src/writers/local/spill.rs @@ -19,21 +19,21 @@ use crate::metrics::ShufflePartitionerMetrics; use crate::writers::BufBatchWriter; use crate::ShuffleBlockWriter; use arrow::record_batch::RecordBatch; -use datafusion::common::DataFusionError; -use datafusion::execution::disk_manager::RefCountedTempFile; use datafusion::execution::runtime_env::RuntimeEnv; -use std::fs::{File, OpenOptions}; +use datafusion::execution::SpillFile as DfSpillFile; +use datafusion::execution::SpillWriter as DfSpillWriter; +use std::sync::Arc; -struct SpillFile { - temp_file: RefCountedTempFile, - file: File, +struct ActiveSpillFile { + temp_file: Arc, + writer: Box, } pub(crate) struct SpillWriter { shuffle_block_writer: ShuffleBlockWriter, write_buffer_size: usize, batch_size: usize, - spill_file: Option, + spill_file: Option, } impl SpillWriter { @@ -62,7 +62,7 @@ impl SpillWriter { let total_bytes_written = { let mut buf_batch_writer = BufBatchWriter::new( &mut self.shuffle_block_writer, - &mut self.spill_file.as_mut().unwrap().file, + &mut self.spill_file.as_mut().unwrap().writer, self.write_buffer_size, self.batch_size, ); @@ -90,21 +90,11 @@ impl SpillWriter { ) -> datafusion::common::Result<()> { if self.spill_file.is_none() { // Spill file is not yet created, create it - let spill_file = runtime + let temp_file = runtime .disk_manager .create_tmp_file("shuffle writer spill")?; - let spill_data = OpenOptions::new() - .write(true) - .create(true) - .truncate(true) - .open(spill_file.path()) - .map_err(|e| { - DataFusionError::Execution(format!("Error occurred while spilling {e}")) - })?; - self.spill_file = Some(SpillFile { - temp_file: spill_file, - file: spill_data, - }); + let writer = temp_file.open_writer()?; + self.spill_file = Some(ActiveSpillFile { temp_file, writer }); } Ok(()) } @@ -112,7 +102,7 @@ impl SpillWriter { pub(crate) fn path(&self) -> Option<&std::path::Path> { self.spill_file .as_ref() - .map(|spill_file| spill_file.temp_file.path()) + .and_then(|spill_file| spill_file.temp_file.path()) } #[cfg(test)] diff --git a/native/shuffle/src/writers/partition_writer.rs b/native/shuffle/src/writers/partition_writer.rs index 25b0e598df..9b415f897d 100644 --- a/native/shuffle/src/writers/partition_writer.rs +++ b/native/shuffle/src/writers/partition_writer.rs @@ -32,7 +32,7 @@ use arrow::record_batch::RecordBatch; /// ascending id order, then a single [`finish_all`](PartitionWriter::finish_all). /// /// [`LocalPartitionWriter`]: crate::writers::local::local_partition_writer::LocalPartitionWriter -pub(crate) trait PartitionWriter: Send + Sync { +pub(crate) trait PartitionWriter: Send { /// Stages the batches from `iter` for partition `pid` without finalizing it. /// /// Used to stream single-partition output and to stage multi-partition diff --git a/native/shuffle/src/writers/shuffle_block_writer.rs b/native/shuffle/src/writers/shuffle_block_writer.rs index 7b6846b3ba..04da9ed96c 100644 --- a/native/shuffle/src/writers/shuffle_block_writer.rs +++ b/native/shuffle/src/writers/shuffle_block_writer.rs @@ -18,7 +18,7 @@ use arrow::array::RecordBatch; use arrow::datatypes::{DataType, Schema, SchemaRef}; use arrow::ipc::writer::{ - write_message, CompressionContext, DictionaryTracker, IpcDataGenerator, IpcWriteOptions, + write_message, DictionaryTracker, IpcDataGenerator, IpcWriteContext, IpcWriteOptions, StreamWriter, }; use arrow::ipc::MetadataVersion; @@ -159,7 +159,7 @@ impl ShuffleBlockWriter { // Fast path: reuse the pre-encoded schema message and write the record batch manually. let data_gen = IpcDataGenerator::default(); let mut dictionary_tracker = DictionaryTracker::new(true); - let mut compression_context = CompressionContext::default(); + let mut compression_context = IpcWriteContext::default(); let (encoded_dictionaries, encoded_batch) = data_gen.encode( batch, &mut dictionary_tracker, diff --git a/native/spark-expr/src/agg_funcs/avg.rs b/native/spark-expr/src/agg_funcs/avg.rs index 24a9a30991..bb5371568b 100644 --- a/native/spark-expr/src/agg_funcs/avg.rs +++ b/native/spark-expr/src/agg_funcs/avg.rs @@ -283,7 +283,6 @@ where &mut self, values: &[ArrayRef], group_indices: &[usize], - _opt_filter: Option<&arrow::array::BooleanArray>, total_num_groups: usize, ) -> Result<()> { assert_eq!(values.len(), 2, "two arguments to merge_batch"); @@ -340,6 +339,14 @@ where ]) } + fn convert_to_state( + &self, + _values: &[ArrayRef], + _opt_filter: Option<&arrow::array::BooleanArray>, + ) -> Result> { + not_impl_err!("Input batch conversion to state not implemented") + } + fn size(&self) -> usize { self.counts.capacity() * std::mem::size_of::() + self.sums.capacity() * std::mem::size_of::() diff --git a/native/spark-expr/src/agg_funcs/avg_decimal.rs b/native/spark-expr/src/agg_funcs/avg_decimal.rs index 186e027244..0e02dd03d4 100644 --- a/native/spark-expr/src/agg_funcs/avg_decimal.rs +++ b/native/spark-expr/src/agg_funcs/avg_decimal.rs @@ -543,7 +543,6 @@ impl GroupsAccumulator for AvgDecimalGroupsAccumulator { &mut self, values: &[ArrayRef], group_indices: &[usize], - _opt_filter: Option<&arrow::array::BooleanArray>, total_num_groups: usize, ) -> Result<()> { assert_eq!(values.len(), 2, "two arguments to merge_batch"); @@ -652,6 +651,14 @@ impl GroupsAccumulator for AvgDecimalGroupsAccumulator { ]) } + fn convert_to_state( + &self, + _values: &[ArrayRef], + _opt_filter: Option<&arrow::array::BooleanArray>, + ) -> Result> { + not_impl_err!("Input batch conversion to state not implemented") + } + fn size(&self) -> usize { self.counts.capacity() * std::mem::size_of::() + self.sums.capacity() * std::mem::size_of::() diff --git a/native/spark-expr/src/agg_funcs/correlation.rs b/native/spark-expr/src/agg_funcs/correlation.rs index 189b76c863..d69ac61def 100644 --- a/native/spark-expr/src/agg_funcs/correlation.rs +++ b/native/spark-expr/src/agg_funcs/correlation.rs @@ -24,7 +24,7 @@ use crate::agg_funcs::covariance::{CovarianceAccumulator, CovarianceGroupsAccumu use crate::agg_funcs::stddev::StddevAccumulator; use crate::agg_funcs::variance::VarianceGroupsAccumulator; use arrow::compute::filter; -use datafusion::common::{Result, ScalarValue}; +use datafusion::common::{not_impl_err, Result, ScalarValue}; use datafusion::logical_expr::function::{AccumulatorArgs, StateFieldsArgs}; use datafusion::logical_expr::{ Accumulator, AggregateUDFImpl, EmitTo, GroupsAccumulator, Signature, Volatility, @@ -335,7 +335,6 @@ impl GroupsAccumulator for CorrelationGroupsAccumulator { &mut self, values: &[ArrayRef], group_indices: &[usize], - opt_filter: Option<&BooleanArray>, total_num_groups: usize, ) -> Result<()> { assert_eq!(values.len(), 6, "six state columns to merge_batch"); @@ -358,11 +357,11 @@ impl GroupsAccumulator for CorrelationGroupsAccumulator { ]; self.covar - .merge_batch(&covar_state, group_indices, opt_filter, total_num_groups)?; + .merge_batch(&covar_state, group_indices, total_num_groups)?; self.var1 - .merge_batch(&var1_state, group_indices, opt_filter, total_num_groups)?; + .merge_batch(&var1_state, group_indices, total_num_groups)?; self.var2 - .merge_batch(&var2_state, group_indices, opt_filter, total_num_groups)?; + .merge_batch(&var2_state, group_indices, total_num_groups)?; Ok(()) } @@ -442,6 +441,14 @@ impl GroupsAccumulator for CorrelationGroupsAccumulator { ]) } + fn convert_to_state( + &self, + _values: &[ArrayRef], + _opt_filter: Option<&BooleanArray>, + ) -> Result> { + not_impl_err!("Input batch conversion to state not implemented") + } + fn size(&self) -> usize { self.covar.size() + self.var1.size() + self.var2.size() } diff --git a/native/spark-expr/src/agg_funcs/covariance.rs b/native/spark-expr/src/agg_funcs/covariance.rs index e5b77ad33b..548c118720 100644 --- a/native/spark-expr/src/agg_funcs/covariance.rs +++ b/native/spark-expr/src/agg_funcs/covariance.rs @@ -21,7 +21,9 @@ use arrow::array::{Array, ArrayRef, AsArray, BooleanArray, Float64Array}; use arrow::buffer::NullBuffer; use arrow::compute::cast; use arrow::datatypes::{DataType, Field, FieldRef, Float64Type}; -use datafusion::common::{downcast_value, unwrap_or_internal_err, Result, ScalarValue}; +use datafusion::common::{ + downcast_value, not_impl_err, unwrap_or_internal_err, Result, ScalarValue, +}; use datafusion::logical_expr::function::{AccumulatorArgs, StateFieldsArgs}; use datafusion::logical_expr::{ Accumulator, AggregateUDFImpl, EmitTo, GroupsAccumulator, Signature, Volatility, @@ -414,7 +416,6 @@ impl GroupsAccumulator for CovarianceGroupsAccumulator { &mut self, values: &[ArrayRef], group_indices: &[usize], - _opt_filter: Option<&BooleanArray>, total_num_groups: usize, ) -> Result<()> { assert_eq!(values.len(), 4, "four arguments to merge_batch"); @@ -467,6 +468,14 @@ impl GroupsAccumulator for CovarianceGroupsAccumulator { ]) } + fn convert_to_state( + &self, + _values: &[ArrayRef], + _opt_filter: Option<&BooleanArray>, + ) -> Result> { + not_impl_err!("Input batch conversion to state not implemented") + } + fn size(&self) -> usize { (self.counts.capacity() + self.mean1s.capacity() @@ -563,8 +572,8 @@ mod groups_tests { let rstate = right.state(EmitTo::All).unwrap(); let mut merged = pop(); - merged.merge_batch(&lstate, &[0], None, 1).unwrap(); - merged.merge_batch(&rstate, &[0], None, 1).unwrap(); + merged.merge_batch(&lstate, &[0], 1).unwrap(); + merged.merge_batch(&rstate, &[0], 1).unwrap(); let merged_result = evaluate(&mut merged)[0].unwrap(); assert!((single - merged_result).abs() < 1e-12); diff --git a/native/spark-expr/src/agg_funcs/hll_plus_plus.rs b/native/spark-expr/src/agg_funcs/hll_plus_plus.rs index 01fa13800a..3cb39f0acd 100644 --- a/native/spark-expr/src/agg_funcs/hll_plus_plus.rs +++ b/native/spark-expr/src/agg_funcs/hll_plus_plus.rs @@ -30,7 +30,7 @@ use arrow::array::{ Array, ArrayRef, AsArray, BooleanArray, Float32Array, Float64Array, Int64Array, }; use arrow::datatypes::{DataType, Field, FieldRef, Float32Type, Float64Type}; -use datafusion::common::{Result, ScalarValue}; +use datafusion::common::{not_impl_err, Result, ScalarValue}; use datafusion::logical_expr::function::{AccumulatorArgs, StateFieldsArgs}; use datafusion::logical_expr::{ Accumulator, AggregateUDFImpl, EmitTo, GroupsAccumulator, Signature, Volatility, @@ -425,7 +425,6 @@ impl GroupsAccumulator for HllPlusPlusGroupsAccumulator { &mut self, values: &[ArrayRef], group_indices: &[usize], - _opt_filter: Option<&BooleanArray>, total_num_groups: usize, ) -> Result<()> { self.resize(total_num_groups); @@ -467,6 +466,14 @@ impl GroupsAccumulator for HllPlusPlusGroupsAccumulator { Ok(columns) } + fn convert_to_state( + &self, + _values: &[ArrayRef], + _opt_filter: Option<&BooleanArray>, + ) -> Result> { + not_impl_err!("Input batch conversion to state not implemented") + } + fn size(&self) -> usize { std::mem::size_of_val(self) + self.words.capacity() * std::mem::size_of::() @@ -639,9 +646,7 @@ mod tests { for part in [&mut left, &mut right] { let state = part.state(EmitTo::All).unwrap(); let n = state[0].len(); - merged - .merge_batch(&state, &vec![0usize; n], None, 1) - .unwrap(); + merged.merge_batch(&state, &vec![0usize; n], 1).unwrap(); } let single = { let mut a = acc(9); diff --git a/native/spark-expr/src/agg_funcs/percentile.rs b/native/spark-expr/src/agg_funcs/percentile.rs index 3fddfdcb68..78e585a396 100644 --- a/native/spark-expr/src/agg_funcs/percentile.rs +++ b/native/spark-expr/src/agg_funcs/percentile.rs @@ -20,7 +20,7 @@ use arrow::array::{ }; use arrow::buffer::{OffsetBuffer, ScalarBuffer}; use arrow::datatypes::{DataType, Field, FieldRef, Float64Type}; -use datafusion::common::{internal_err, plan_err, Result, ScalarValue}; +use datafusion::common::{internal_err, not_impl_err, plan_err, Result, ScalarValue}; use datafusion::logical_expr::function::{AccumulatorArgs, StateFieldsArgs}; use datafusion::logical_expr::Volatility::Immutable; use datafusion::logical_expr::{ @@ -221,7 +221,6 @@ impl GroupsAccumulator for SparkPercentileGroupsAccumulator { &mut self, values: &[ArrayRef], group_indices: &[usize], - _opt_filter: Option<&BooleanArray>, total_num_groups: usize, ) -> Result<()> { let input_group_values = values[0].as_list::(); @@ -268,6 +267,14 @@ impl GroupsAccumulator for SparkPercentileGroupsAccumulator { Ok(Arc::new(builder.finish())) } + fn convert_to_state( + &self, + _values: &[ArrayRef], + _opt_filter: Option<&BooleanArray>, + ) -> Result> { + not_impl_err!("Input batch conversion to state not implemented") + } + fn size(&self) -> usize { self.group_values .iter() diff --git a/native/spark-expr/src/agg_funcs/stddev.rs b/native/spark-expr/src/agg_funcs/stddev.rs index 3389a86af6..bbceaa72dc 100644 --- a/native/spark-expr/src/agg_funcs/stddev.rs +++ b/native/spark-expr/src/agg_funcs/stddev.rs @@ -22,7 +22,7 @@ use arrow::array::{ArrayRef, AsArray, BooleanArray, Float64Array}; use arrow::datatypes::FieldRef; use arrow::datatypes::{DataType, Field, Float64Type}; use datafusion::common::types::NativeType; -use datafusion::common::{internal_err, Result, ScalarValue}; +use datafusion::common::{internal_err, not_impl_err, Result, ScalarValue}; use datafusion::logical_expr::function::{AccumulatorArgs, StateFieldsArgs}; use datafusion::logical_expr::{ Accumulator, AggregateUDFImpl, Coercion, EmitTo, GroupsAccumulator, Signature, Volatility, @@ -231,11 +231,10 @@ impl GroupsAccumulator for StddevGroupsAccumulator { &mut self, values: &[ArrayRef], group_indices: &[usize], - opt_filter: Option<&BooleanArray>, total_num_groups: usize, ) -> Result<()> { self.inner - .merge_batch(values, group_indices, opt_filter, total_num_groups) + .merge_batch(values, group_indices, total_num_groups) } fn evaluate(&mut self, emit_to: EmitTo) -> Result { @@ -250,6 +249,14 @@ impl GroupsAccumulator for StddevGroupsAccumulator { self.inner.state(emit_to) } + fn convert_to_state( + &self, + _values: &[ArrayRef], + _opt_filter: Option<&BooleanArray>, + ) -> Result> { + not_impl_err!("Input batch conversion to state not implemented") + } + fn size(&self) -> usize { self.inner.size() } diff --git a/native/spark-expr/src/agg_funcs/sum_decimal.rs b/native/spark-expr/src/agg_funcs/sum_decimal.rs index 4ddf13c5d6..cae0a70b42 100644 --- a/native/spark-expr/src/agg_funcs/sum_decimal.rs +++ b/native/spark-expr/src/agg_funcs/sum_decimal.rs @@ -21,7 +21,7 @@ use arrow::array::{ cast::AsArray, types::Decimal128Type, Array, ArrayRef, BooleanArray, Decimal128Array, }; use arrow::datatypes::{DataType, Field, FieldRef}; -use datafusion::common::{DataFusionError, Result as DFResult, ScalarValue}; +use datafusion::common::{not_impl_err, DataFusionError, Result as DFResult, ScalarValue}; use datafusion::logical_expr::function::{AccumulatorArgs, StateFieldsArgs}; use datafusion::logical_expr::Volatility::Immutable; use datafusion::logical_expr::{ @@ -541,14 +541,8 @@ impl GroupsAccumulator for SumDecimalGroupsAccumulator { &mut self, values: &[ArrayRef], group_indices: &[usize], - opt_filter: Option<&BooleanArray>, total_num_groups: usize, ) -> DFResult<()> { - debug_assert!( - opt_filter.is_none(), - "opt_filter is not supported in merge_batch" - ); - self.resize_helper(total_num_groups); // For decimal sum, always expect 2 arrays regardless of eval_mode @@ -609,6 +603,14 @@ impl GroupsAccumulator for SumDecimalGroupsAccumulator { Ok(()) } + fn convert_to_state( + &self, + _values: &[ArrayRef], + _opt_filter: Option<&BooleanArray>, + ) -> DFResult> { + not_impl_err!("Input batch conversion to state not implemented") + } + fn size(&self) -> usize { self.sum.capacity() * std::mem::size_of::>() + self.is_empty.capacity() * std::mem::size_of::() diff --git a/native/spark-expr/src/agg_funcs/sum_int.rs b/native/spark-expr/src/agg_funcs/sum_int.rs index 59a4371772..56a8c122c4 100644 --- a/native/spark-expr/src/agg_funcs/sum_int.rs +++ b/native/spark-expr/src/agg_funcs/sum_int.rs @@ -23,7 +23,7 @@ use arrow::array::{ use arrow::datatypes::{ ArrowNativeType, DataType, Field, FieldRef, Int16Type, Int32Type, Int64Type, Int8Type, }; -use datafusion::common::{DataFusionError, Result as DFResult, ScalarValue}; +use datafusion::common::{not_impl_err, DataFusionError, Result as DFResult, ScalarValue}; use datafusion::logical_expr::function::{AccumulatorArgs, StateFieldsArgs}; use datafusion::logical_expr::Volatility::Immutable; use datafusion::logical_expr::{ @@ -495,14 +495,8 @@ impl GroupsAccumulator for SumIntGroupsAccumulatorLegacy { &mut self, values: &[ArrayRef], group_indices: &[usize], - opt_filter: Option<&BooleanArray>, total_num_groups: usize, ) -> DFResult<()> { - debug_assert!( - opt_filter.is_none(), - "opt_filter is not supported in merge_batch" - ); - if values.len() != 1 { return Err(DataFusionError::Internal(format!( "Invalid state while merging batch. Expected 1 element but found {}", @@ -529,6 +523,14 @@ impl GroupsAccumulator for SumIntGroupsAccumulatorLegacy { Ok(()) } + fn convert_to_state( + &self, + _values: &[ArrayRef], + _opt_filter: Option<&BooleanArray>, + ) -> DFResult> { + not_impl_err!("Input batch conversion to state not implemented") + } + fn size(&self) -> usize { std::mem::size_of_val(self) } @@ -642,14 +644,8 @@ impl GroupsAccumulator for SumIntGroupsAccumulatorAnsi { &mut self, values: &[ArrayRef], group_indices: &[usize], - opt_filter: Option<&BooleanArray>, total_num_groups: usize, ) -> DFResult<()> { - debug_assert!( - opt_filter.is_none(), - "opt_filter is not supported in merge_batch" - ); - if values.len() != 1 { return Err(DataFusionError::Internal(format!( "Invalid state while merging batch. Expected 1 element but found {}", @@ -680,6 +676,14 @@ impl GroupsAccumulator for SumIntGroupsAccumulatorAnsi { Ok(()) } + fn convert_to_state( + &self, + _values: &[ArrayRef], + _opt_filter: Option<&BooleanArray>, + ) -> DFResult> { + not_impl_err!("Input batch conversion to state not implemented") + } + fn size(&self) -> usize { std::mem::size_of_val(self) } @@ -826,14 +830,8 @@ impl GroupsAccumulator for SumIntGroupsAccumulatorTry { &mut self, values: &[ArrayRef], group_indices: &[usize], - opt_filter: Option<&BooleanArray>, total_num_groups: usize, ) -> DFResult<()> { - debug_assert!( - opt_filter.is_none(), - "opt_filter is not supported in merge_batch" - ); - if values.len() != 2 { return Err(DataFusionError::Internal(format!( "Invalid state while merging batch. Expected 2 elements but found {}", @@ -886,6 +884,14 @@ impl GroupsAccumulator for SumIntGroupsAccumulatorTry { Ok(()) } + fn convert_to_state( + &self, + _values: &[ArrayRef], + _opt_filter: Option<&BooleanArray>, + ) -> DFResult> { + not_impl_err!("Input batch conversion to state not implemented") + } + fn size(&self) -> usize { std::mem::size_of_val(self) } diff --git a/native/spark-expr/src/agg_funcs/variance.rs b/native/spark-expr/src/agg_funcs/variance.rs index f1841729cf..57a8f6da50 100644 --- a/native/spark-expr/src/agg_funcs/variance.rs +++ b/native/spark-expr/src/agg_funcs/variance.rs @@ -18,7 +18,7 @@ use arrow::array::{Array, ArrayRef, AsArray, BooleanArray, Float64Array}; use arrow::buffer::NullBuffer; use arrow::datatypes::{DataType, Field, FieldRef, Float64Type}; -use datafusion::common::{downcast_value, Result, ScalarValue}; +use datafusion::common::{downcast_value, not_impl_err, Result, ScalarValue}; use datafusion::logical_expr::function::{AccumulatorArgs, StateFieldsArgs}; use datafusion::logical_expr::Volatility::Immutable; use datafusion::logical_expr::{ @@ -340,7 +340,6 @@ impl GroupsAccumulator for VarianceGroupsAccumulator { &mut self, values: &[ArrayRef], group_indices: &[usize], - _opt_filter: Option<&BooleanArray>, total_num_groups: usize, ) -> Result<()> { assert_eq!(values.len(), 3, "three arguments to merge_batch"); @@ -387,6 +386,14 @@ impl GroupsAccumulator for VarianceGroupsAccumulator { ]) } + fn convert_to_state( + &self, + _values: &[ArrayRef], + _opt_filter: Option<&BooleanArray>, + ) -> Result> { + not_impl_err!("Input batch conversion to state not implemented") + } + fn size(&self) -> usize { self.counts.capacity() * size_of::() + self.means.capacity() * size_of::() @@ -512,8 +519,8 @@ mod groups_tests { let right_state = right.state(EmitTo::All).unwrap(); let mut merged = pop_acc(); - merged.merge_batch(&left_state, &[0], None, 1).unwrap(); - merged.merge_batch(&right_state, &[0], None, 1).unwrap(); + merged.merge_batch(&left_state, &[0], 1).unwrap(); + merged.merge_batch(&right_state, &[0], 1).unwrap(); let merged_result = evaluate(&mut merged)[0].unwrap(); assert!((single_result - merged_result).abs() < 1e-12); diff --git a/native/spark-expr/src/array_funcs/array_insert.rs b/native/spark-expr/src/array_funcs/array_insert.rs index e056c108e0..16954f1243 100644 --- a/native/spark-expr/src/array_funcs/array_insert.rs +++ b/native/spark-expr/src/array_funcs/array_insert.rs @@ -266,17 +266,17 @@ fn array_insert( if pos1 <= len + 1 { // In-range insertion (including appending to end) let corrected = pos1 - 1; // 0-based insertion point - mutable_values.extend(0, start, start + corrected); - mutable_values.extend(1, row_index, row_index + 1); - mutable_values.extend(0, start + corrected, end); + mutable_values.try_extend(0, start, start + corrected)?; + mutable_values.try_extend(1, row_index, row_index + 1)?; + mutable_values.try_extend(0, start + corrected, end)?; final_len = len + 1; } else { // Beyond end: pad with nulls then insert let corrected = pos1 - 1; let padding = corrected - len; - mutable_values.extend(0, start, end); - mutable_values.extend_nulls(padding); - mutable_values.extend(1, row_index, row_index + 1); + mutable_values.try_extend(0, start, end)?; + mutable_values.try_extend_nulls(padding)?; + mutable_values.try_extend(1, row_index, row_index + 1)?; final_len = corrected + 1; // equals pos1 } } else { @@ -289,9 +289,9 @@ fn array_insert( // Legacy: -1 behaves like insert before the last element (corrected = len - k) let base_offset = if legacy_mode { 0 } else { 1 }; let corrected = len - k + base_offset; - mutable_values.extend(0, start, start + corrected); - mutable_values.extend(1, row_index, row_index + 1); - mutable_values.extend(0, start + corrected, end); + mutable_values.try_extend(0, start, start + corrected)?; + mutable_values.try_extend(1, row_index, row_index + 1)?; + mutable_values.try_extend(0, start + corrected, end)?; final_len = len + 1; } else { // Negative index beyond the start (Spark-specific behavior): @@ -300,9 +300,9 @@ fn array_insert( let base_offset = if legacy_mode { 1 } else { 0 }; let target_len = k + base_offset; let padding = target_len.saturating_sub(len + 1); - mutable_values.extend(1, row_index, row_index + 1); // insert item first - mutable_values.extend_nulls(padding); // pad nulls - mutable_values.extend(0, start, end); // append original values + mutable_values.try_extend(1, row_index, row_index + 1)?; // insert item first + mutable_values.try_extend_nulls(padding)?; // pad nulls + mutable_values.try_extend(0, start, end)?; // append original values final_len = target_len; } } diff --git a/native/spark-expr/src/array_funcs/array_slice.rs b/native/spark-expr/src/array_funcs/array_slice.rs index e9dc05330b..fcd3b467a6 100644 --- a/native/spark-expr/src/array_funcs/array_slice.rs +++ b/native/spark-expr/src/array_funcs/array_slice.rs @@ -163,7 +163,7 @@ fn slice_list( } else { let take = std::cmp::min(length_value, arr_len - zero_based_start) as usize; let begin = row_start + zero_based_start as usize; - mutable.extend(0, begin, begin + take); + mutable.try_extend(0, begin, begin + take)?; take }; diff --git a/native/spark-expr/src/array_funcs/arrays_zip.rs b/native/spark-expr/src/array_funcs/arrays_zip.rs index 6f0332d3d8..de4d74bd23 100644 --- a/native/spark-expr/src/array_funcs/arrays_zip.rs +++ b/native/spark-expr/src/array_funcs/arrays_zip.rs @@ -276,15 +276,15 @@ pub fn arrays_zip_inner(args: &[ArrayRef], names: Vec) -> Result { // Null list entry or None (Null-typed) arg — all nulls. if let Some(builder) = builders[col_idx].as_mut() { - builder.extend_nulls(max_len); + builder.try_extend_nulls(max_len)?; } } } diff --git a/native/spark-expr/src/array_funcs/list_extract.rs b/native/spark-expr/src/array_funcs/list_extract.rs index d68784ca70..0944abb987 100644 --- a/native/spark-expr/src/array_funcs/list_extract.rs +++ b/native/spark-expr/src/array_funcs/list_extract.rs @@ -287,10 +287,10 @@ fn list_extract( let len = offset_window[1].as_usize() - start; if list_array.is_null(row) { - mutable.extend_nulls(1); + mutable.try_extend_nulls(1)?; } else if let Some(index) = index { if let Some(i) = adjust_index(index, len)? { - mutable.extend(0, start + i, start + i + 1); + mutable.try_extend(0, start + i, start + i + 1)?; } else if fail_on_error { // Throw appropriate error based on whether this is element_at (one_based=true) // or GetArrayItem (one_based=false) @@ -309,11 +309,11 @@ fn list_extract( }; return Err(error_wrapper(error)); } else { - mutable.extend(1, 0, 1); + mutable.try_extend(1, 0, 1)?; } } else { // index is NULL → result is NULL - mutable.extend_nulls(1); + mutable.try_extend_nulls(1)?; } } diff --git a/native/spark-expr/src/bloom_filter/spark_bit_array.rs b/native/spark-expr/src/bloom_filter/spark_bit_array.rs index 6d43bdb942..b55567402b 100644 --- a/native/spark-expr/src/bloom_filter/spark_bit_array.rs +++ b/native/spark-expr/src/bloom_filter/spark_bit_array.rs @@ -77,8 +77,9 @@ impl SparkBitArray { pub fn merge_be_words(&mut self, incoming: &[u8]) { debug_assert_eq!(self.data.len() * 8, incoming.len()); let mut bit_count: usize = 0; - for (word, chunk) in self.data.iter_mut().zip(incoming.chunks_exact(8)) { - *word |= u64::from_be_bytes(chunk.try_into().unwrap()); + let (chunks, _) = incoming.as_chunks::<8>(); + for (word, chunk) in self.data.iter_mut().zip(chunks) { + *word |= u64::from_be_bytes(*chunk); bit_count += word.count_ones() as usize; } self.bit_count = bit_count; diff --git a/native/spark-expr/src/comet_scalar_funcs.rs b/native/spark-expr/src/comet_scalar_funcs.rs index 50732959c8..0173d02d35 100644 --- a/native/spark-expr/src/comet_scalar_funcs.rs +++ b/native/spark-expr/src/comet_scalar_funcs.rs @@ -23,6 +23,7 @@ use crate::math_funcs::checked_arithmetic::{checked_add, checked_div, checked_mu use crate::math_funcs::log::spark_log; use crate::math_funcs::modulo_expr::spark_modulo; use crate::math_funcs::pow::spark_pow; +use crate::math_funcs::sqrt::spark_sqrt; use crate::{ spark_ceil, spark_day_name, spark_decimal_div, spark_decimal_integral_div, spark_floor, spark_isnan, spark_lpad, spark_make_decimal, spark_month_name, spark_read_side_padding, @@ -213,6 +214,10 @@ pub fn create_comet_physical_fun_with_eval_mode( let func = Arc::new(spark_pow); make_comet_scalar_udf!("pow", func, without data_type) } + "spark_sqrt" => { + let func = Arc::new(spark_sqrt); + make_comet_scalar_udf!("spark_sqrt", func, without data_type) + } "base64" => { let func = Arc::new(crate::string_funcs::spark_base64); make_comet_scalar_udf!("base64", func, without data_type) diff --git a/native/spark-expr/src/lib.rs b/native/spark-expr/src/lib.rs index 42ab43361a..099f23b347 100644 --- a/native/spark-expr/src/lib.rs +++ b/native/spark-expr/src/lib.rs @@ -86,9 +86,9 @@ pub use json_funcs::{FromJson, ToJson}; pub use math_funcs::{ checked_add, checked_div, checked_mul, checked_sub, create_modulo_expr, create_negate_expr, spark_ceil, spark_decimal_div, spark_decimal_integral_div, spark_floor, spark_log, - spark_make_decimal, spark_pow, spark_round, spark_unhex, spark_unscaled_value, CheckOverflow, - DecimalRescaleCheckOverflow, NegativeExpr, NormalizeNaNAndZero, WideDecimalBinaryExpr, - WideDecimalOp, + spark_make_decimal, spark_pow, spark_round, spark_sqrt, spark_unhex, spark_unscaled_value, + CheckOverflow, DecimalRescaleCheckOverflow, NegativeExpr, NormalizeNaNAndZero, + WideDecimalBinaryExpr, WideDecimalOp, }; pub use query_context::{create_query_context_map, QueryContext, QueryContextMap}; pub use string_funcs::*; diff --git a/native/spark-expr/src/math_funcs/mod.rs b/native/spark-expr/src/math_funcs/mod.rs index 56809a01fb..7a7c6a9753 100644 --- a/native/spark-expr/src/math_funcs/mod.rs +++ b/native/spark-expr/src/math_funcs/mod.rs @@ -26,6 +26,7 @@ pub mod modulo_expr; mod negative; pub(crate) mod pow; mod round; +pub(crate) mod sqrt; pub(crate) mod unhex; mod utils; mod wide_decimal_binary_expr; @@ -41,5 +42,6 @@ pub use modulo_expr::create_modulo_expr; pub use negative::{create_negate_expr, NegativeExpr}; pub use pow::spark_pow; pub use round::spark_round; +pub use sqrt::spark_sqrt; pub use unhex::spark_unhex; pub use wide_decimal_binary_expr::{WideDecimalBinaryExpr, WideDecimalOp}; diff --git a/native/spark-expr/src/math_funcs/sqrt.rs b/native/spark-expr/src/math_funcs/sqrt.rs new file mode 100644 index 0000000000..1397386eb9 --- /dev/null +++ b/native/spark-expr/src/math_funcs/sqrt.rs @@ -0,0 +1,90 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use arrow::array::{ArrayRef, AsArray, Float64Array}; +use arrow::datatypes::Float64Type; +use datafusion::common::{DataFusionError, ScalarValue}; +use datafusion::physical_plan::ColumnarValue; +use std::sync::Arc; + +/// Spark-compatible `sqrt`, matching `java.lang.Math.sqrt`: a negative input produces +/// `NaN` rather than an error, unlike DataFusion's own `sqrt`. +pub fn spark_sqrt(args: &[ColumnarValue]) -> Result { + if args.len() != 1 { + return Err(DataFusionError::Internal(format!( + "spark_sqrt requires 1 argument, got {}", + args.len() + ))); + } + + match &args[0] { + ColumnarValue::Array(array) => { + let values = array.as_primitive_opt::().ok_or_else(|| { + DataFusionError::Internal(format!( + "spark_sqrt expected Float64, got {:?}", + array.data_type() + )) + })?; + let result: Float64Array = values.unary(|v| v.sqrt()); + Ok(ColumnarValue::Array(Arc::new(result) as ArrayRef)) + } + ColumnarValue::Scalar(ScalarValue::Float64(v)) => Ok(ColumnarValue::Scalar( + ScalarValue::Float64(v.map(f64::sqrt)), + )), + ColumnarValue::Scalar(other) => Err(DataFusionError::Internal(format!( + "spark_sqrt expected Float64 scalar, got {other:?}", + ))), + } +} + +#[cfg(test)] +mod test { + use super::*; + use arrow::array::{Array, Float64Array}; + + #[test] + fn test_spark_sqrt_negative_is_nan() { + let input = Float64Array::from(vec![Some(4.0), Some(-1.0), Some(0.0), None]); + let result = spark_sqrt(&[ColumnarValue::Array(Arc::new(input))]).unwrap(); + let ColumnarValue::Array(result) = result else { + unreachable!() + }; + let result = result.as_primitive::(); + assert_eq!(result.value(0), 2.0); + assert!(result.value(1).is_nan()); + assert_eq!(result.value(2), 0.0); + assert!(result.is_null(3)); + } + + #[test] + fn test_spark_sqrt_scalar_negative_is_nan() { + let result = + spark_sqrt(&[ColumnarValue::Scalar(ScalarValue::Float64(Some(-1.0)))]).unwrap(); + let ColumnarValue::Scalar(ScalarValue::Float64(Some(result))) = result else { + unreachable!() + }; + assert!(result.is_nan()); + } + + #[test] + fn test_spark_sqrt_scalar_null() { + let result = spark_sqrt(&[ColumnarValue::Scalar(ScalarValue::Float64(None))]).unwrap(); + let ColumnarValue::Scalar(ScalarValue::Float64(None)) = result else { + unreachable!() + }; + } +} diff --git a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala index 4299eb14ba..9baa12533d 100644 --- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala +++ b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala @@ -153,7 +153,7 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim { classOf[Signum] -> CometScalarFunction("signum"), classOf[Sin] -> CometScalarFunction("sin"), classOf[Sinh] -> CometScalarFunction("sinh"), - classOf[Sqrt] -> CometScalarFunction("sqrt"), + classOf[Sqrt] -> CometSqrt, classOf[Subtract] -> CometSubtract, classOf[Tan] -> CometScalarFunction("tan"), classOf[Tanh] -> CometScalarFunction("tanh"), diff --git a/spark/src/main/scala/org/apache/comet/serde/math.scala b/spark/src/main/scala/org/apache/comet/serde/math.scala index a0ea76cc06..65e2bf673e 100644 --- a/spark/src/main/scala/org/apache/comet/serde/math.scala +++ b/spark/src/main/scala/org/apache/comet/serde/math.scala @@ -19,7 +19,7 @@ package org.apache.comet.serde -import org.apache.spark.sql.catalyst.expressions.{Abs, Add, Atan2, Attribute, BRound, Ceil, CheckOverflow, Conv, Expression, Floor, Hex, Hypot, If, LessThanOrEqual, Literal, Log, Log10, Log1p, Log2, Logarithm, NaNvl, Pmod, Pow, UnaryPositive, Unhex, WidthBucket} +import org.apache.spark.sql.catalyst.expressions.{Abs, Add, Atan2, Attribute, BRound, Ceil, CheckOverflow, Conv, Expression, Floor, Hex, Hypot, If, LessThanOrEqual, Literal, Log, Log10, Log1p, Log2, Logarithm, NaNvl, Pmod, Pow, Sqrt, UnaryPositive, Unhex, WidthBucket} import org.apache.spark.sql.types.{DecimalType, DoubleType, NumericType} import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, scalarFunctionExprToProto, scalarFunctionExprToProtoWithReturnType, serializeDataType} @@ -216,6 +216,23 @@ object CometPow extends CometExpressionSerde[Pow] { } } +// Uses a custom spark_sqrt UDF because DataFusion's own `sqrt` errors on negative +// input, while Spark's Sqrt (a plain wrapper around java.lang.Math.sqrt) returns NaN. +// spark_sqrt is a Comet-only name with no DataFusion builtin counterpart, so the +// return type must be set explicitly here to skip the session registry lookup that +// scalarFunctionExprToProto would otherwise trigger (see CometLogarithm/spark_log). +object CometSqrt extends CometExpressionSerde[Sqrt] { + override def convert( + expr: Sqrt, + inputs: Seq[Attribute], + binding: Boolean): Option[ExprOuterClass.Expr] = { + val childExpr = exprToProtoInternal(expr.child, inputs, binding) + val optExpr = + scalarFunctionExprToProtoWithReturnType("spark_sqrt", DoubleType, false, childExpr) + optExpr + } +} + sealed trait MathExprBase { protected def nullIfNegative(expression: Expression): Expression = { val zero = Literal.default(expression.dataType) diff --git a/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala index fd7d4ee3f4..4feda01d89 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala @@ -29,12 +29,12 @@ import org.apache.parquet.hadoop.api.WriteSupport import org.apache.parquet.hadoop.api.WriteSupport.WriteContext import org.apache.parquet.io.api.RecordConsumer import org.apache.parquet.schema.MessageTypeParser -import org.apache.spark.sql.{CometTestBase, Row} +import org.apache.spark.sql.{CometTestBase, DataFrame, Row} import org.apache.spark.sql.comet.CometNativeScanExec import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper import org.apache.spark.sql.functions.{array, col} import org.apache.spark.sql.internal.SQLConf -import org.apache.spark.sql.types.{IntegerType, LongType, NullType, StringType, StructType} +import org.apache.spark.sql.types._ import org.apache.comet.CometConf import org.apache.comet.CometSparkSessionExtensions.isSpark41Plus @@ -794,6 +794,7 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper class Builder extends ParquetWriter.Builder[RecordConsumer => Unit, Builder](new Path(path)) { override def getWriteSupport(conf: Configuration): WriteSupport[RecordConsumer => Unit] = writeSupport + override def self(): Builder = this } val writer = new Builder().build() @@ -997,6 +998,82 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper } } } + + /** Runs `df` and returns the summed `bytes_scanned` metric across its native scans. */ + private def bytesScanned(df: DataFrame): Long = { + val (_, cometPlan) = checkSparkAnswerAndOperator(df) + val scans = cometPlan.collect { case n: CometNativeScanExec => n } + assert(scans.nonEmpty, "Expected a CometNativeScanExec") + scans.map(_.metrics("bytes_scanned").value).sum + } + + test("issue #4859: native scan honors pruned nested ReadSchema for array") { + // Regression repro: the native scan pushes only a top-level projection vector into + // DataFusion, so a plain Column("events") ref yields ProjectionMask::roots and the + // whole nested column (all leaves) is read even though Spark's pruned ReadSchema asks + // for a single leaf. Selecting only the small `id` leaf must scan far fewer bytes than + // also selecting the large `payload` leaf; before the fix they are ~equal. + withTempPath { path => + withSQLConf( + SQLConf.LEAF_NODE_DEFAULT_PARALLELISM.key -> "1", + SQLConf.NESTED_SCHEMA_PRUNING_ENABLED.key -> "true") { + // events: array>. `payload` is a distinct, + // high-entropy uuid per row so it dominates the on-disk column size and resists + // compression; `id` is a sequential long that compresses to almost nothing. + spark + .range(0, 20000) + .selectExpr("array(named_struct('id', id, 'payload', uuid())) as events") + .repartition(1) + .write + .mode("overwrite") + .parquet(path.toString) + + val idOnly = + bytesScanned(spark.read.parquet(path.toString).selectExpr("events.id")) + val idAndPayload = + bytesScanned( + spark.read.parquet(path.toString).selectExpr("events.id", "events.payload")) + + assert( + idOnly < idAndPayload / 2, + "native scan read the whole nested column despite pruned ReadSchema " + + s"(bytes_scanned: events.id=$idOnly, events.id+payload=$idAndPayload)") + } + } + } + + test("issue #4859: leaf pruning through array> drops siblings at each level") { + withTempPath { path => + withSQLConf( + SQLConf.LEAF_NODE_DEFAULT_PARALLELISM.key -> "1", + SQLConf.NESTED_SCHEMA_PRUNING_ENABLED.key -> "true") { + // events: array>>. `blob` is the dominant + // high-entropy leaf. Selecting events.inner.a must prune both the top-level + // sibling `id` and the inner sibling `blob`. + spark + .range(0, 20000) + .selectExpr( + "array(named_struct('id', id, 'inner', named_struct('a', id, 'blob', uuid()))) as events") + .repartition(1) + .write + .mode("overwrite") + .parquet(path.toString) + + val aOnly = + bytesScanned(spark.read.parquet(path.toString).selectExpr("events.inner.a")) + val aAndBlob = + bytesScanned( + spark.read + .parquet(path.toString) + .selectExpr("events.inner.a", "events.inner.blob")) + + assert( + aOnly < aAndBlob / 2, + "native scan read unrequested nested leaves through array> " + + s"(bytes_scanned: events.inner.a=$aOnly, events.inner.a+blob=$aAndBlob)") + } + } + } } private class DirectWriteSupport(schema: org.apache.parquet.schema.MessageType) diff --git a/spark/src/test/scala/org/apache/spark/sql/CometTPCHQuerySuite.scala b/spark/src/test/scala/org/apache/spark/sql/CometTPCHQuerySuite.scala index 0f94292709..b92ab0d7d9 100644 --- a/spark/src/test/scala/org/apache/spark/sql/CometTPCHQuerySuite.scala +++ b/spark/src/test/scala/org/apache/spark/sql/CometTPCHQuerySuite.scala @@ -91,7 +91,7 @@ class CometTPCHQuerySuite extends QueryTest with TPCBase with ShimCometTPCHQuery conf.set(CometConf.COMET_NATIVE_SCAN_ENABLED.key, "true") conf.set(CometConf.COMET_SHUFFLE_ENABLED.key, "true") conf.set(MEMORY_OFFHEAP_ENABLED.key, "true") - conf.set(MEMORY_OFFHEAP_SIZE.key, "2g") + conf.set(MEMORY_OFFHEAP_SIZE.key, "4g") } protected override def createSparkSession: TestSparkSession = {