diff --git a/native/Cargo.lock b/native/Cargo.lock index eeaa9da5ce..5397ccb309 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", ] @@ -232,23 +232,23 @@ 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", - "arrow-buffer", - "arrow-cast", + "arrow-arith 59.1.0", + "arrow-array 59.1.0", + "arrow-buffer 59.1.0", + "arrow-cast 59.1.0", "arrow-csv", - "arrow-data", - "arrow-ipc", + "arrow-data 59.1.0", + "arrow-ipc 59.1.0", "arrow-json", - "arrow-ord", + "arrow-ord 59.1.0", "arrow-row", - "arrow-schema", - "arrow-select", - "arrow-string", + "arrow-schema 59.1.0", + "arrow-select 59.1.0", + "arrow-string 59.1.0", ] [[package]] @@ -257,10 +257,24 @@ version = "58.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0a41203398f0eaa6f7ec8e62c0da742a21abf282c148fc157f6c35c90e29981a" dependencies = [ - "arrow-array", - "arrow-buffer", - "arrow-data", - "arrow-schema", + "arrow-array 58.4.0", + "arrow-buffer 58.4.0", + "arrow-data 58.4.0", + "arrow-schema 58.4.0", + "chrono", + "num-traits", +] + +[[package]] +name = "arrow-arith" +version = "59.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "64a13b8d3008c4e9063c597a08f46446fe3fd5789277127672d6c0bdbb43b1ff" +dependencies = [ + "arrow-array 59.1.0", + "arrow-buffer 59.1.0", + "arrow-data 59.1.0", + "arrow-schema 59.1.0", "chrono", "num-traits", ] @@ -272,9 +286,27 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ae33dad492b7df00a217563a7b0ef2874df68a0deea1b1a3acf628152f7f7a69" dependencies = [ "ahash", - "arrow-buffer", - "arrow-data", - "arrow-schema", + "arrow-buffer 58.4.0", + "arrow-data 58.4.0", + "arrow-schema 58.4.0", + "chrono", + "half", + "hashbrown 0.17.1", + "num-complex", + "num-integer", + "num-traits", +] + +[[package]] +name = "arrow-array" +version = "59.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9486151b2f0785bafc6fa04fc5c99fcb4495455662e58787ea32eaaed33c4192" +dependencies = [ + "ahash", + "arrow-buffer 59.1.0", + "arrow-data 59.1.0", + "arrow-schema 59.1.0", "chrono", "chrono-tz", "half", @@ -296,18 +328,51 @@ dependencies = [ "num-traits", ] +[[package]] +name = "arrow-buffer" +version = "59.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c4776577a87794bfdf0b4e90e2ea12454fa7738ea2823c4be5b9d1851da7b434" +dependencies = [ + "bytes", + "half", + "num-bigint", + "num-traits", +] + [[package]] name = "arrow-cast" version = "58.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3a8a327c9649f30d8406995f27642b68df354713cca3baaaf100f076f18d5f34" dependencies = [ - "arrow-array", - "arrow-buffer", - "arrow-data", - "arrow-ord", - "arrow-schema", - "arrow-select", + "arrow-array 58.4.0", + "arrow-buffer 58.4.0", + "arrow-data 58.4.0", + "arrow-ord 58.4.0", + "arrow-schema 58.4.0", + "arrow-select 58.4.0", + "atoi", + "base64 0.22.1", + "chrono", + "half", + "lexical-core", + "num-traits", + "ryu", +] + +[[package]] +name = "arrow-cast" +version = "59.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a9ad451ce4f98710828a455b96991b8f031deb2e67f5fcad6773f017e4a69c3a" +dependencies = [ + "arrow-array 59.1.0", + "arrow-buffer 59.1.0", + "arrow-data 59.1.0", + "arrow-ord 59.1.0", + "arrow-schema 59.1.0", + "arrow-select 59.1.0", "atoi", "base64 0.22.1", "chrono", @@ -320,13 +385,13 @@ 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", - "arrow-schema", + "arrow-array 59.1.0", + "arrow-cast 59.1.0", + "arrow-schema 59.1.0", "chrono", "csv", "csv-core", @@ -339,8 +404,21 @@ version = "58.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2b24852db04738907e06c04ea61e42fe7fda962a34513022dc0d0e754fb7976b" dependencies = [ - "arrow-buffer", - "arrow-schema", + "arrow-buffer 58.4.0", + "arrow-schema 58.4.0", + "half", + "num-integer", + "num-traits", +] + +[[package]] +name = "arrow-data" +version = "59.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b38fe43e2e8704360f1464e6e8cc4fc381ef02cc4fb0192afa8df1aaa0115c66" +dependencies = [ + "arrow-buffer 59.1.0", + "arrow-schema 59.1.0", "half", "num-integer", "num-traits", @@ -352,11 +430,25 @@ version = "58.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "29a908a11fcfb3fb2f6730f4ac15e367bc644e419155e96238f68cf3adde572b" dependencies = [ - "arrow-array", - "arrow-buffer", - "arrow-data", - "arrow-schema", - "arrow-select", + "arrow-array 58.4.0", + "arrow-buffer 58.4.0", + "arrow-data 58.4.0", + "arrow-schema 58.4.0", + "arrow-select 58.4.0", + "flatbuffers", +] + +[[package]] +name = "arrow-ipc" +version = "59.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29dac499fcbc6ba74ee0324057821d381929a48526a3966bd9dffb44aa06d98c" +dependencies = [ + "arrow-array 59.1.0", + "arrow-buffer 59.1.0", + "arrow-data 59.1.0", + "arrow-schema 59.1.0", + "arrow-select 59.1.0", "flatbuffers", "lz4_flex 0.13.1", "zstd", @@ -364,16 +456,16 @@ 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", - "arrow-cast", - "arrow-ord", - "arrow-schema", - "arrow-select", + "arrow-array 59.1.0", + "arrow-buffer 59.1.0", + "arrow-cast 59.1.0", + "arrow-ord 59.1.0", + "arrow-schema 59.1.0", + "arrow-select 59.1.0", "chrono", "half", "indexmap 2.14.0", @@ -393,23 +485,36 @@ version = "58.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "63a083ec750f5c043f02946b4baf05fcdbb55f4560a3277055caca5cc99f3eb0" dependencies = [ - "arrow-array", - "arrow-buffer", - "arrow-data", - "arrow-schema", - "arrow-select", + "arrow-array 58.4.0", + "arrow-buffer 58.4.0", + "arrow-data 58.4.0", + "arrow-schema 58.4.0", + "arrow-select 58.4.0", +] + +[[package]] +name = "arrow-ord" +version = "59.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0e13dbdc2a9c053c10c7baa6e30faee04a180aa7ce88e471835850ce37abd20b" +dependencies = [ + "arrow-array 59.1.0", + "arrow-buffer 59.1.0", + "arrow-data 59.1.0", + "arrow-schema 59.1.0", + "arrow-select 59.1.0", ] [[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", - "arrow-data", - "arrow-schema", + "arrow-array 59.1.0", + "arrow-buffer 59.1.0", + "arrow-data 59.1.0", + "arrow-schema 59.1.0", "half", ] @@ -418,6 +523,12 @@ name = "arrow-schema" version = "58.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "21ca356ad6425cecb6eb7b28e4f659f1ee7880fbb1a16127de7dd62901efee9e" + +[[package]] +name = "arrow-schema" +version = "59.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d9e4969dc350d571766247143ab36a5187d095d3d3690970408bc630d47c69e5" dependencies = [ "bitflags 2.13.1", "serde_core", @@ -431,10 +542,24 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c58da39eb3d8350ad4a549e5c2bc49284dac554016c69829310350f1731b0aad" dependencies = [ "ahash", - "arrow-array", - "arrow-buffer", - "arrow-data", - "arrow-schema", + "arrow-array 58.4.0", + "arrow-buffer 58.4.0", + "arrow-data 58.4.0", + "arrow-schema 58.4.0", + "num-traits", +] + +[[package]] +name = "arrow-select" +version = "59.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "402770dba90865359d98d1ef92ef16e23d75c0cca9c2c880c8a05468b7743bf9" +dependencies = [ + "ahash", + "arrow-array 59.1.0", + "arrow-buffer 59.1.0", + "arrow-data 59.1.0", + "arrow-schema 59.1.0", "num-traits", ] @@ -444,11 +569,28 @@ version = "58.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6789b388467525e3271326b6b4915666ecfdf5142aef09779445c954b67543c" dependencies = [ - "arrow-array", - "arrow-buffer", - "arrow-data", - "arrow-schema", - "arrow-select", + "arrow-array 58.4.0", + "arrow-buffer 58.4.0", + "arrow-data 58.4.0", + "arrow-schema 58.4.0", + "arrow-select 58.4.0", + "memchr", + "num-traits", + "regex", + "regex-syntax", +] + +[[package]] +name = "arrow-string" +version = "59.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a2b0afbb8b9016700938291123df30838b89decc3213dba00852021988b170d3" +dependencies = [ + "arrow-array 59.1.0", + "arrow-buffer 59.1.0", + "arrow-data 59.1.0", + "arrow-schema 59.1.0", + "arrow-select 59.1.0", "memchr", "num-traits", "regex", @@ -1059,9 +1201,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" @@ -1839,11 +1981,10 @@ 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?rev=179b32c9b60103d9c4e6a4364f10f6286c963904#179b32c9b60103d9c4e6a4364f10f6286c963904" dependencies = [ "arrow", - "arrow-schema", + "arrow-schema 59.1.0", "async-trait", "chrono", "datafusion-catalog", @@ -1873,11 +2014,11 @@ dependencies = [ "datafusion-sql", "futures", "indexmap 2.14.0", - "itertools 0.14.0", + "itertools 0.15.0", "log", "object_store", "parking_lot", - "parquet", + "parquet 59.1.0", "sqlparser", "tempfile", "tokio", @@ -1888,8 +2029,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?rev=179b32c9b60103d9c4e6a4364f10f6286c963904#179b32c9b60103d9c4e6a4364f10f6286c963904" dependencies = [ "arrow", "async-trait", @@ -1903,7 +2043,7 @@ dependencies = [ "datafusion-physical-plan", "datafusion-session", "futures", - "itertools 0.14.0", + "itertools 0.15.0", "log", "object_store", "parking_lot", @@ -1913,8 +2053,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?rev=179b32c9b60103d9c4e6a4364f10f6286c963904#179b32c9b60103d9c4e6a4364f10f6286c963904" dependencies = [ "arrow", "async-trait", @@ -1928,9 +2067,10 @@ dependencies = [ "datafusion-physical-expr-common", "datafusion-physical-plan", "futures", - "itertools 0.14.0", + "itertools 0.15.0", "log", "object_store", + "percent-encoding", ] [[package]] @@ -1972,7 +2112,7 @@ dependencies = [ "once_cell", "opendal", "parking_lot", - "parquet", + "parquet 59.1.0", "paste", "pprof", "procfs", @@ -2011,7 +2151,7 @@ dependencies = [ "jni 0.22.4", "lazy_static", "once_cell", - "parquet", + "parquet 59.1.0", "paste", "prost", "regex", @@ -2046,7 +2186,7 @@ dependencies = [ "jni 0.21.1", "log", "lz4_flex 0.14.0", - "parquet", + "parquet 59.1.0", "simd-adler32", "snap", "tempfile", @@ -2059,7 +2199,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,23 +2222,23 @@ 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?rev=179b32c9b60103d9c4e6a4364f10f6286c963904#179b32c9b60103d9c4e6a4364f10f6286c963904" dependencies = [ "arrow", - "arrow-ipc", - "arrow-schema", + "arrow-ipc 59.1.0", + "arrow-schema 59.1.0", "chrono", "foldhash 0.2.0", "half", "hashbrown 0.17.1", "hex", "indexmap 2.14.0", - "itertools 0.14.0", + "itertools 0.15.0", "libc", "log", + "num-traits", "object_store", - "parquet", + "parquet 59.1.0", "sqlparser", "tokio", "uuid", @@ -2108,8 +2248,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?rev=179b32c9b60103d9c4e6a4364f10f6286c963904#179b32c9b60103d9c4e6a4364f10f6286c963904" dependencies = [ "futures", "log", @@ -2119,8 +2258,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?rev=179b32c9b60103d9c4e6a4364f10f6286c963904#179b32c9b60103d9c4e6a4364f10f6286c963904" dependencies = [ "arrow", "async-compression", @@ -2140,7 +2278,7 @@ dependencies = [ "flate2", "futures", "glob", - "itertools 0.14.0", + "itertools 0.15.0", "liblzma", "log", "object_store", @@ -2155,11 +2293,10 @@ 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?rev=179b32c9b60103d9c4e6a4364f10f6286c963904#179b32c9b60103d9c4e6a4364f10f6286c963904" dependencies = [ "arrow", - "arrow-ipc", + "arrow-ipc 59.1.0", "async-trait", "bytes", "datafusion-common", @@ -2171,7 +2308,7 @@ dependencies = [ "datafusion-physical-plan", "datafusion-session", "futures", - "itertools 0.14.0", + "itertools 0.15.0", "object_store", "tokio", ] @@ -2179,8 +2316,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?rev=179b32c9b60103d9c4e6a4364f10f6286c963904#179b32c9b60103d9c4e6a4364f10f6286c963904" dependencies = [ "arrow", "async-trait", @@ -2202,8 +2338,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?rev=179b32c9b60103d9c4e6a4364f10f6286c963904#179b32c9b60103d9c4e6a4364f10f6286c963904" dependencies = [ "arrow", "async-trait", @@ -2225,10 +2360,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?rev=179b32c9b60103d9c4e6a4364f10f6286c963904#179b32c9b60103d9c4e6a4364f10f6286c963904" dependencies = [ "arrow", + "arrow-schema 59.1.0", "async-trait", "bytes", "datafusion-common", @@ -2245,29 +2380,28 @@ dependencies = [ "datafusion-pruning", "datafusion-session", "futures", - "itertools 0.14.0", + "itertools 0.15.0", "log", "object_store", "parking_lot", - "parquet", + "parquet 59.1.0", "tokio", ] [[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?rev=179b32c9b60103d9c4e6a4364f10f6286c963904#179b32c9b60103d9c4e6a4364f10f6286c963904" [[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?rev=179b32c9b60103d9c4e6a4364f10f6286c963904#179b32c9b60103d9c4e6a4364f10f6286c963904" dependencies = [ "arrow", - "arrow-buffer", + "arrow-buffer 59.1.0", "async-trait", + "bytes", "dashmap", "datafusion-common", "datafusion-expr", @@ -2276,20 +2410,22 @@ dependencies = [ "log", "object_store", "parking_lot", - "parquet", + "parquet 59.1.0", + "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?rev=179b32c9b60103d9c4e6a4364f10f6286c963904#179b32c9b60103d9c4e6a4364f10f6286c963904" dependencies = [ "arrow", - "arrow-schema", + "arrow-schema 59.1.0", "async-trait", "chrono", "datafusion-common", @@ -2299,7 +2435,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 +2443,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?rev=179b32c9b60103d9c4e6a4364f10f6286c963904#179b32c9b60103d9c4e6a4364f10f6286c963904" 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?rev=179b32c9b60103d9c4e6a4364f10f6286c963904#179b32c9b60103d9c4e6a4364f10f6286c963904" dependencies = [ "arrow", - "arrow-buffer", - "base64 0.22.1", + "arrow-buffer 59.1.0", + "base64 0.23.1", "blake2", "blake3", "chrono", @@ -2337,7 +2471,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 +2485,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?rev=179b32c9b60103d9c4e6a4364f10f6286c963904#179b32c9b60103d9c4e6a4364f10f6286c963904" dependencies = [ "arrow", "datafusion-common", @@ -2363,8 +2496,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 +2505,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?rev=179b32c9b60103d9c4e6a4364f10f6286c963904#179b32c9b60103d9c4e6a4364f10f6286c963904" dependencies = [ "arrow", "datafusion-common", @@ -2384,11 +2516,10 @@ 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?rev=179b32c9b60103d9c4e6a4364f10f6286c963904#179b32c9b60103d9c4e6a4364f10f6286c963904" dependencies = [ "arrow", - "arrow-ord", + "arrow-ord 59.1.0", "datafusion-common", "datafusion-doc", "datafusion-execution", @@ -2400,7 +2531,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 +2540,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?rev=179b32c9b60103d9c4e6a4364f10f6286c963904#179b32c9b60103d9c4e6a4364f10f6286c963904" dependencies = [ "arrow", "async-trait", @@ -2425,8 +2555,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?rev=179b32c9b60103d9c4e6a4364f10f6286c963904#179b32c9b60103d9c4e6a4364f10f6286c963904" dependencies = [ "arrow", "datafusion-common", @@ -2442,8 +2571,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?rev=179b32c9b60103d9c4e6a4364f10f6286c963904#179b32c9b60103d9c4e6a4364f10f6286c963904" dependencies = [ "datafusion-common", "datafusion-physical-expr-common", @@ -2452,19 +2580,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?rev=179b32c9b60103d9c4e6a4364f10f6286c963904#179b32c9b60103d9c4e6a4364f10f6286c963904" 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?rev=179b32c9b60103d9c4e6a4364f10f6286c963904#179b32c9b60103d9c4e6a4364f10f6286c963904" dependencies = [ "arrow", "chrono", @@ -2473,7 +2599,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 +2608,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?rev=179b32c9b60103d9c4e6a4364f10f6286c963904#179b32c9b60103d9c4e6a4364f10f6286c963904" dependencies = [ "arrow", "datafusion-common", @@ -2494,7 +2619,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 +2628,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?rev=179b32c9b60103d9c4e6a4364f10f6286c963904#179b32c9b60103d9c4e6a4364f10f6286c963904" dependencies = [ "arrow", "datafusion-common", @@ -2512,14 +2636,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?rev=179b32c9b60103d9c4e6a4364f10f6286c963904#179b32c9b60103d9c4e6a4364f10f6286c963904" dependencies = [ "arrow", "chrono", @@ -2527,7 +2650,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 +2658,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?rev=179b32c9b60103d9c4e6a4364f10f6286c963904#179b32c9b60103d9c4e6a4364f10f6286c963904" dependencies = [ "arrow", "datafusion-common", @@ -2547,21 +2669,22 @@ 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?rev=179b32c9b60103d9c4e6a4364f10f6286c963904#179b32c9b60103d9c4e6a4364f10f6286c963904" dependencies = [ "arrow", - "arrow-data", - "arrow-ipc", - "arrow-ord", - "arrow-schema", + "arrow-data 59.1.0", + "arrow-ipc 59.1.0", + "arrow-ord 59.1.0", + "arrow-schema 59.1.0", "async-trait", + "bytes", "datafusion-common", "datafusion-common-runtime", "datafusion-execution", @@ -2575,19 +2698,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?rev=179b32c9b60103d9c4e6a4364f10f6286c963904#179b32c9b60103d9c4e6a4364f10f6286c963904" dependencies = [ "arrow", "datafusion-common", @@ -2602,9 +2725,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?rev=179b32c9b60103d9c4e6a4364f10f6286c963904#179b32c9b60103d9c4e6a4364f10f6286c963904" dependencies = [ + "arrow-schema 59.1.0", "async-trait", "datafusion-common", "datafusion-execution", @@ -2616,8 +2739,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?rev=179b32c9b60103d9c4e6a4364f10f6286c963904#179b32c9b60103d9c4e6a4364f10f6286c963904" dependencies = [ "arrow", "bigdecimal", @@ -2646,8 +2768,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?rev=179b32c9b60103d9c4e6a4364f10f6286c963904#179b32c9b60103d9c4e6a4364f10f6286c963904" dependencies = [ "arrow", "bigdecimal", @@ -3503,14 +3624,14 @@ dependencies = [ "anyhow", "apache-avro", "array-init", - "arrow-arith", - "arrow-array", - "arrow-buffer", - "arrow-cast", - "arrow-ord", - "arrow-schema", - "arrow-select", - "arrow-string", + "arrow-arith 58.4.0", + "arrow-array 58.4.0", + "arrow-buffer 58.4.0", + "arrow-cast 58.4.0", + "arrow-ord 58.4.0", + "arrow-schema 58.4.0", + "arrow-select 58.4.0", + "arrow-string 58.4.0", "as-any", "async-trait", "backon", @@ -3529,7 +3650,7 @@ dependencies = [ "murmur3", "once_cell", "ordered-float 4.6.0", - "parquet", + "parquet 58.4.0", "rand 0.9.5", "reqwest 0.12.28", "roaring", @@ -4799,12 +4920,48 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d298093b2dec60289dce0684c986d0f7679e9dd15771c2c65406e1aaf604a704" dependencies = [ "ahash", - "arrow-array", - "arrow-buffer", - "arrow-data", - "arrow-ipc", - "arrow-schema", - "arrow-select", + "arrow-array 58.4.0", + "arrow-buffer 58.4.0", + "arrow-data 58.4.0", + "arrow-ipc 58.4.0", + "arrow-schema 58.4.0", + "arrow-select 58.4.0", + "base64 0.22.1", + "brotli", + "bytes", + "chrono", + "flate2", + "futures", + "half", + "hashbrown 0.17.1", + "lz4_flex 0.13.1", + "num-bigint", + "num-integer", + "num-traits", + "paste", + "ring", + "seq-macro", + "simdutf8", + "snap", + "thrift", + "tokio", + "twox-hash", + "zstd", +] + +[[package]] +name = "parquet" +version = "59.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5302d4da74d6596a1f11f9928767995b53bca657cbeea1e4e8c5074f8a1157dd" +dependencies = [ + "ahash", + "arrow-array 59.1.0", + "arrow-buffer 59.1.0", + "arrow-data 59.1.0", + "arrow-ipc 59.1.0", + "arrow-schema 59.1.0", + "arrow-select 59.1.0", "base64 0.22.1", "brotli", "bytes", @@ -4826,7 +4983,6 @@ dependencies = [ "seq-macro", "simdutf8", "snap", - "thrift", "tokio", "twox-hash", "zstd", @@ -4834,12 +4990,12 @@ 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", + "arrow-schema 59.1.0", "chrono", "half", "indexmap 2.14.0", @@ -4850,12 +5006,12 @@ 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", + "arrow-schema 59.1.0", "chrono", "half", "indexmap 2.14.0", @@ -4867,11 +5023,11 @@ 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", + "arrow-schema 59.1.0", "base64 0.22.1", "chrono", "parquet-variant", @@ -5506,9 +5662,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 +5685,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 +5701,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 +5721,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 +5759,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 +5783,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 +5794,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", @@ -7466,9 +7636,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..0113f3602d 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", rev = "179b32c9b60103d9c4e6a4364f10f6286c963904", default-features = false, features = ["unicode_expressions", "crypto_expressions", "nested_expressions", "parquet"] } +datafusion-datasource = { git = "https://github.com/apache/datafusion", rev = "179b32c9b60103d9c4e6a4364f10f6286c963904" } +datafusion-physical-expr-adapter = { git = "https://github.com/apache/datafusion", rev = "179b32c9b60103d9c4e6a4364f10f6286c963904" } +datafusion-spark = { git = "https://github.com/apache/datafusion", rev = "179b32c9b60103d9c4e6a4364f10f6286c963904", features = ["core"] } datafusion-comet-spark-expr = { path = "spark-expr" } datafusion-comet-common = { path = "common" } datafusion-comet-jni-bridge = { path = "jni-bridge" } diff --git a/native/core/Cargo.toml b/native/core/Cargo.toml index 7c88b260c3..2a8930f925 100644 --- a/native/core/Cargo.toml +++ b/native/core/Cargo.toml @@ -72,8 +72,8 @@ reqwest = { version = "0.12", default-features = false, features = ["rustls-tls- object_store_opendal = { version = "0.57.0", optional = true } hdfs-sys = {version = "0.3", optional = true, features = ["hdfs_3_3"]} opendal = { version = "0.57.0", optional = true, features = ["services-hdfs"] } -iceberg = { workspace = true } -iceberg-storage-opendal = { workspace = true } +iceberg = { workspace = true, optional = true } +iceberg-storage-opendal = { workspace = true, optional = true } reqsign-core = { workspace = true } serde_json = "1.0" uuid = "1.23.3" @@ -91,12 +91,17 @@ 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", rev = "179b32c9b60103d9c4e6a4364f10f6286c963904" } [features] backtrace = ["datafusion/backtrace"] default = ["hdfs-opendal"] hdfs-opendal = ["opendal", "object_store_opendal", "hdfs-sys"] +# The native Iceberg scan. Off by default while this branch tracks DataFusion main: +# iceberg-rust is still on arrow 58, so its `RecordBatch` does not unify with the +# arrow 59 that DataFusion main requires. Re-enable (and make this a default +# feature again) once iceberg-rust upgrades to arrow 59. +iceberg-scan = ["iceberg", "iceberg-storage-opendal"] jemalloc = ["tikv-jemallocator", "tikv-jemalloc-ctl"] # exclude optional packages from cargo machete verifications diff --git a/native/core/src/cloud/s3/credential_bridge.rs b/native/core/src/cloud/s3/credential_bridge.rs index 14e4bc49d9..94f1a95736 100644 --- a/native/core/src/cloud/s3/credential_bridge.rs +++ b/native/core/src/cloud/s3/credential_bridge.rs @@ -22,16 +22,21 @@ use crate::execution::operators::ExecutionError; use crate::jvm_bridge::{jni_new_global_ref, jni_static_call, JVMClasses}; use async_trait::async_trait; +#[cfg(feature = "iceberg-scan")] use iceberg_storage_opendal::AwsCredential as IcebergAwsCredential; use jni::objects::{Global, JFieldID, JObject, JString, JValue}; use jni::signature::{Primitive, ReturnType}; use jni::strings::JNIString; use jni::sys::jint; +#[cfg(feature = "iceberg-scan")] use log::warn; use object_store::aws::AwsCredential; use object_store::CredentialProvider; +#[cfg(feature = "iceberg-scan")] use once_cell::sync::OnceCell; +#[cfg(feature = "iceberg-scan")] use reqsign_core::time::Timestamp; +#[cfg(feature = "iceberg-scan")] use reqsign_core::{ Context, Error as ReqsignError, ErrorKind as ReqsignErrorKind, ProvideCredential as IcebergProvideCredential, @@ -39,14 +44,17 @@ use reqsign_core::{ use std::collections::HashMap; use std::fmt; use std::sync::Arc; +#[cfg(feature = "iceberg-scan")] use std::time::Duration; /// Cap on opendal's credential cache when the provider does not report an expiry. Prevents the /// executor from holding a stale credential for the entire job lifetime. +#[cfg(feature = "iceberg-scan")] const DEFAULT_EXPIRY_WHEN_UNKNOWN: Duration = Duration::from_secs(300); /// Once-per-process latch for the "missing expiry" warning. Bridges are per-scan, so a per-bridge /// latch would re-log on every scan. +#[cfg(feature = "iceberg-scan")] static WARNED_MISSING_EXPIRY: OnceCell<()> = OnceCell::new(); /// Access intent forwarded to the Java SPI. Ordinal must match the JVM `CometS3AccessMode` enum. @@ -269,7 +277,9 @@ struct RawCredentials { access_key_id: String, secret_access_key: String, session_token: Option, - /// Absolute expiry. `0` means the provider did not report one. + /// Absolute expiry. `0` means the provider did not report one. Only consumed by the + /// Iceberg credential path. + #[cfg_attr(not(feature = "iceberg-scan"), allow(dead_code))] expiration_epoch_millis: i64, } @@ -290,6 +300,7 @@ impl CredentialProvider for CometS3CredentialBridge { } } +#[cfg(feature = "iceberg-scan")] impl IcebergProvideCredential for CometS3CredentialBridge { type Credential = IcebergAwsCredential; diff --git a/native/core/src/execution/columnar_to_row.rs b/native/core/src/execution/columnar_to_row.rs index 14e115cba0..06f5bbd78d 100644 --- a/native/core/src/execution/columnar_to_row.rs +++ b/native/core/src/execution/columnar_to_row.rs @@ -2499,11 +2499,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..1196592350 100644 --- a/native/core/src/execution/merge_as_partial.rs +++ b/native/core/src/execution/merge_as_partial.rs @@ -207,23 +207,22 @@ impl GroupsAccumulator for MergeAsPartialGroupsAccumulator { &mut self, values: &[ArrayRef], group_indices: &[usize], - opt_filter: Option<&BooleanArray>, + _opt_filter: Option<&BooleanArray>, total_num_groups: usize, ) -> Result<()> { // Redirect update to merge — this is the key trick. 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 evaluate(&mut self, emit_to: EmitTo) -> Result { @@ -234,6 +233,18 @@ impl GroupsAccumulator for MergeAsPartialGroupsAccumulator { self.inner.state(emit_to) } + fn convert_to_state( + &self, + values: &[ArrayRef], + _opt_filter: Option<&BooleanArray>, + ) -> Result> { + // The input to this accumulator is already the inner accumulator's intermediate + // state (that is the point of redirecting update to merge), so the state for a + // group of one row is that row itself. The filter is ignored here for the same + // reason it is ignored in `update_batch`. + Ok(values.to_vec()) + } + fn size(&self) -> usize { self.inner.size() } diff --git a/native/core/src/execution/operators/copy.rs b/native/core/src/execution/operators/copy.rs index 193c385c3c..ca0ae6ddfc 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("copy_array: extend within existing array cannot 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("copy_array: extend within existing array cannot 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/mod.rs b/native/core/src/execution/operators/mod.rs index 6fdc3b0486..fbc80ac290 100644 --- a/native/core/src/execution/operators/mod.rs +++ b/native/core/src/execution/operators/mod.rs @@ -21,6 +21,7 @@ pub use crate::errors::ExecutionError; pub use aligned_stream_reader::*; pub use copy::*; +#[cfg(feature = "iceberg-scan")] pub use iceberg_scan::*; pub use scan::*; @@ -28,6 +29,7 @@ mod aligned_stream_reader; mod copy; mod expand; pub use expand::ExpandExec; +#[cfg(feature = "iceberg-scan")] mod iceberg_scan; mod parquet_writer; pub use parquet_writer::{ParquetCompression, ParquetWriterExec}; diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index d4c97c509d..151dd56f65 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -23,6 +23,7 @@ pub mod operator_registry; use crate::execution::operators::init_csv_datasource_exec; use crate::execution::operators::AlignedArrowStreamReader; +#[cfg(feature = "iceberg-scan")] use crate::execution::operators::IcebergScanExec; use crate::execution::{ expressions::list_empty_to_null::ListEmptyToNullExpr, @@ -80,6 +81,7 @@ use datafusion_comet_spark_expr::{ SparkBloomFilterVersion, SparkPercentile, SumInteger, ToCsv, }; use datafusion_spark::function::aggregate::collect::{SparkCollectList, SparkCollectSet}; +#[cfg(feature = "iceberg-scan")] use iceberg::expr::Bind; use crate::execution::operators::ExecutionError::GeneralError; @@ -1700,6 +1702,14 @@ impl PhysicalPlanner { Arc::new(SparkPlan::new(spark_plan.plan_id, Arc::new(scan), vec![])), )) } + #[cfg(not(feature = "iceberg-scan"))] + OpStruct::IcebergScan(_) => Err(GeneralError( + "Native Iceberg scan is not available in this build: the `iceberg-scan` \ + feature is disabled because iceberg-rust does not yet support arrow 59. \ + Disable spark.comet.scan.impl=native_iceberg_compat to fall back to Spark." + .into(), + )), + #[cfg(feature = "iceberg-scan")] OpStruct::IcebergScan(scan) => { // Extract common data and single partition's file tasks // Per-partition injection happens in Scala before sending to native @@ -2050,10 +2060,7 @@ impl PhysicalPlanner { depth: 1, }); - let unnest_options = UnnestOptions { - preserve_nulls: explode.outer, - recursions: vec![], - }; + let unnest_options = UnnestOptions::new().with_preserve_nulls(explode.outer); let unnest_exec = Arc::new(UnnestExec::new( project_exec, @@ -3742,6 +3749,7 @@ fn align_shuffle_writer_input( .map_err(|e| ExecutionError::DataFusionError(e.to_string())) } +#[cfg(feature = "iceberg-scan")] /// Converts a protobuf PartitionValue to an iceberg Literal. /// fn partition_value_to_literal( @@ -3795,6 +3803,7 @@ fn partition_value_to_literal( Ok(Some(literal)) } +#[cfg(feature = "iceberg-scan")] /// Decodes an unscaled decimal (two's-complement big-endian) into i128. fn decimal_bytes_to_i128(bytes: &[u8]) -> Result { if bytes.len() > 16 { @@ -3815,6 +3824,7 @@ fn decimal_bytes_to_i128(bytes: &[u8]) -> Result { Ok(i128::from_be_bytes(buf)) } +#[cfg(feature = "iceberg-scan")] /// Converts a protobuf PartitionData to an iceberg Struct. /// /// Uses the existing Struct::from_iter() API from iceberg-rust to construct the struct @@ -3832,6 +3842,7 @@ fn partition_data_to_struct( Ok(iceberg::spec::Struct::from_iter(literals)) } +#[cfg(feature = "iceberg-scan")] /// Converts protobuf FileScanTasks from Scala into iceberg-rust FileScanTask objects. /// /// Each task contains a residual predicate that is used for row-group level filtering @@ -4391,6 +4402,7 @@ fn literal_to_array_ref( // Iceberg Residual Predicate Conversion // ============================================================================ +#[cfg(feature = "iceberg-scan")] /// Converts a serialized Iceberg residual predicate into an iceberg-rust `Predicate` for row-group /// pruning. This is only a pruning hint -- the post-scan CometFilter enforces correctness -- so any /// node or literal that cannot be represented degrades to `None` (no pushdown) rather than an @@ -4462,6 +4474,7 @@ fn iceberg_predicate_to_predicate( } } +#[cfg(feature = "iceberg-scan")] /// Combines the two children of a logical residual node (And/Or), returning `None` unless both /// converted. A missing child is not expected: the Scala serde emits a logical node only when both /// children convert, and Rust decodes every node it emits, so both sides always convert for a @@ -4488,6 +4501,7 @@ fn combine_logical( } } +#[cfg(feature = "iceberg-scan")] /// Converts a serialized `IcebergLiteral` into an iceberg-rust `Datum` for predicate pushdown. /// Returns `None` for null and for byte-array-backed types (decimal/uuid/fixed/binary), which /// iceberg-rust cannot use in the page index yet; the driver does not emit those for predicates, @@ -4560,6 +4574,7 @@ mod tests { use crate::execution::operators::ExecutionError; use crate::execution::planner::literal_to_array_ref; + #[cfg(feature = "iceberg-scan")] use crate::execution::planner::parse_file_scan_tasks_from_common; use crate::parquet::parquet_support::SparkParquetOptions; use crate::parquet::schema_adapter::SparkPhysicalExprAdapterFactory; @@ -5645,6 +5660,7 @@ mod tests { }); } + #[cfg(feature = "iceberg-scan")] #[test] fn test_metadata_field_id_constants_match_iceberg_rust() { // These constants are duplicated in Scala (CometIcebergNativeScan.MetadataFieldIds) @@ -5662,6 +5678,7 @@ mod tests { ); } + #[cfg(feature = "iceberg-scan")] #[test] fn test_unified_partition_type_merges_specs_by_descending_spec_id() { use iceberg::spec::{NestedField, PartitionSpec, PrimitiveType, Type}; @@ -5768,6 +5785,7 @@ mod tests { assert_eq!(fields[1].name, "category"); } + #[cfg(feature = "iceberg-scan")] #[test] fn test_unified_partition_type_tolerates_unparseable_spec() { // Regression for TestForwardCompatibility.testSparkCanReadUnknownTransform: a spec that 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/shuffle/src/writers/local/spill.rs b/native/shuffle/src/writers/local/spill.rs index 450b617421..e12b68a005 100644 --- a/native/shuffle/src/writers/local/spill.rs +++ b/native/shuffle/src/writers/local/spill.rs @@ -20,12 +20,13 @@ 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 datafusion::execution::SpillFile as DfSpillFile; use std::fs::{File, OpenOptions}; +use std::sync::Arc; struct SpillFile { - temp_file: RefCountedTempFile, + temp_file: Arc, file: File, } @@ -97,7 +98,11 @@ impl SpillWriter { .write(true) .create(true) .truncate(true) - .open(spill_file.path()) + .open(spill_file.path().ok_or_else(|| { + DataFusionError::Execution( + "Spill file backend does not expose a local path".to_string(), + ) + })?) .map_err(|e| { DataFusionError::Execution(format!("Error occurred while spilling {e}")) })?; @@ -112,7 +117,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/shuffle_block_writer.rs b/native/shuffle/src/writers/shuffle_block_writer.rs index 7b6846b3ba..646754f2d2 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,12 +159,12 @@ 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 ipc_write_context = IpcWriteContext::default(); let (encoded_dictionaries, encoded_batch) = data_gen.encode( batch, &mut dictionary_tracker, &self.write_options, - &mut compression_context, + &mut ipc_write_context, )?; debug_assert!(encoded_dictionaries.is_empty()); diff --git a/native/spark-expr/src/agg_funcs/avg.rs b/native/spark-expr/src/agg_funcs/avg.rs index 24a9a30991..048119bca2 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,43 @@ where ]) } + fn convert_to_state( + &self, + values: &[ArrayRef], + opt_filter: Option<&arrow::array::BooleanArray>, + ) -> Result> { + assert_eq!(values.len(), 1, "single argument to convert_to_state"); + let values = values[0].as_primitive::(); + let num_rows = values.len(); + + // Each row becomes its own group, so its partial sum is the value itself and + // its partial count is 1. Null and filtered-out rows contribute nothing, the + // same way `update_batch` skips them. + let mut sums = vec![T::default_value(); num_rows]; + let mut counts = vec![0i64; num_rows]; + for (idx, (sum, count)) in sums.iter_mut().zip(counts.iter_mut()).enumerate() { + if let Some(f) = opt_filter { + if !f.is_valid(idx) || !f.value(idx) { + continue; + } + } + if values.is_null(idx) { + continue; + } + *sum = values.value(idx); + *count = 1; + } + + let sums = PrimitiveArray::::new(sums.into(), None) + .with_data_type(self.return_data_type.clone()); + let counts = Int64Array::new(counts.into(), None); + + Ok(vec![ + Arc::new(sums) as ArrayRef, + Arc::new(counts) as ArrayRef, + ]) + } + 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..23198cd6c1 100644 --- a/native/spark-expr/src/agg_funcs/avg_decimal.rs +++ b/native/spark-expr/src/agg_funcs/avg_decimal.rs @@ -19,7 +19,7 @@ use arrow::array::{ builder::PrimitiveBuilder, cast::AsArray, types::{Decimal128Type, Int64Type}, - Array, ArrayRef, Decimal128Array, Int64Array, PrimitiveArray, + Array, ArrayRef, BooleanArray, Decimal128Array, Int64Array, PrimitiveArray, }; use arrow::datatypes::{DataType, Field, FieldRef}; use arrow::{array::BooleanBufferBuilder, buffer::NullBuffer, compute::sum}; @@ -30,6 +30,7 @@ use datafusion::logical_expr::{ use datafusion::physical_expr::expressions::format_state_name; use std::sync::Arc; +use super::convert_to_state::convert_to_state_per_row; use crate::utils::{build_bool_state, is_valid_decimal_precision, unlikely}; use crate::{decimal_sum_overflow_error, EvalMode, SparkErrorWithContext}; use arrow::array::ArrowNativeTypeOp; @@ -543,7 +544,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 +652,28 @@ impl GroupsAccumulator for AvgDecimalGroupsAccumulator { ]) } + fn convert_to_state( + &self, + values: &[ArrayRef], + opt_filter: Option<&BooleanArray>, + ) -> Result> { + convert_to_state_per_row( + Self::new( + &self.return_data_type, + &self.sum_data_type, + self.target_precision, + self.target_scale, + self.sum_precision, + self.sum_scale, + self.eval_mode, + self.expr_id, + Arc::clone(&self.registry), + ), + values, + opt_filter, + ) + } + 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/convert_to_state.rs b/native/spark-expr/src/agg_funcs/convert_to_state.rs new file mode 100644 index 0000000000..6655200579 --- /dev/null +++ b/native/spark-expr/src/agg_funcs/convert_to_state.rs @@ -0,0 +1,43 @@ +// 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, BooleanArray}; +use datafusion::common::Result; +use datafusion::logical_expr::{EmitTo, GroupsAccumulator}; + +/// Implements [`GroupsAccumulator::convert_to_state`] by treating every input row as +/// its own group. +/// +/// DataFusion calls `convert_to_state` when the Partial phase of a multi-phase +/// aggregation is not reducing cardinality enough to be worth maintaining a hash +/// table, and it instead forwards intermediate state straight to the next phase. +/// +/// `accumulator` must be a freshly created accumulator of the same shape as the one +/// being converted, so that the state columns produced here match those produced by +/// [`GroupsAccumulator::state`]. Row `i` of the input is accumulated into group `i`, +/// which also means null and filtered-out rows are skipped exactly as +/// [`GroupsAccumulator::update_batch`] skips them. +pub(super) fn convert_to_state_per_row( + mut accumulator: A, + values: &[ArrayRef], + opt_filter: Option<&BooleanArray>, +) -> Result> { + let num_rows = values[0].len(); + let group_indices: Vec = (0..num_rows).collect(); + accumulator.update_batch(values, &group_indices, opt_filter, num_rows)?; + accumulator.state(EmitTo::All) +} diff --git a/native/spark-expr/src/agg_funcs/correlation.rs b/native/spark-expr/src/agg_funcs/correlation.rs index 189b76c863..de70003d4d 100644 --- a/native/spark-expr/src/agg_funcs/correlation.rs +++ b/native/spark-expr/src/agg_funcs/correlation.rs @@ -20,6 +20,7 @@ use arrow::compute::{and, is_not_null}; use arrow::datatypes::{DataType, Field, FieldRef}; use std::sync::Arc; +use crate::agg_funcs::convert_to_state::convert_to_state_per_row; use crate::agg_funcs::covariance::{CovarianceAccumulator, CovarianceGroupsAccumulator}; use crate::agg_funcs::stddev::StddevAccumulator; use crate::agg_funcs::variance::VarianceGroupsAccumulator; @@ -335,7 +336,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 +358,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 +442,14 @@ impl GroupsAccumulator for CorrelationGroupsAccumulator { ]) } + fn convert_to_state( + &self, + values: &[ArrayRef], + opt_filter: Option<&BooleanArray>, + ) -> Result> { + convert_to_state_per_row(Self::new(self.null_on_divide_by_zero), values, opt_filter) + } + 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..1dcf0a0872 100644 --- a/native/spark-expr/src/agg_funcs/covariance.rs +++ b/native/spark-expr/src/agg_funcs/covariance.rs @@ -17,6 +17,7 @@ * under the License. */ +use super::convert_to_state::convert_to_state_per_row; use arrow::array::{Array, ArrayRef, AsArray, BooleanArray, Float64Array}; use arrow::buffer::NullBuffer; use arrow::compute::cast; @@ -414,7 +415,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 +467,18 @@ impl GroupsAccumulator for CovarianceGroupsAccumulator { ]) } + fn convert_to_state( + &self, + values: &[ArrayRef], + opt_filter: Option<&BooleanArray>, + ) -> Result> { + convert_to_state_per_row( + Self::new(self.stats_type, self.null_on_divide_by_zero), + values, + opt_filter, + ) + } + fn size(&self) -> usize { (self.counts.capacity() + self.mean1s.capacity() @@ -563,8 +575,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..307142e78d 100644 --- a/native/spark-expr/src/agg_funcs/hll_plus_plus.rs +++ b/native/spark-expr/src/agg_funcs/hll_plus_plus.rs @@ -23,6 +23,7 @@ //! `aggBufferSchema`, and the cardinality estimate uses the same bias-correction tables, so //! results are bit-identical to Spark. +use crate::agg_funcs::convert_to_state::convert_to_state_per_row; use crate::agg_funcs::hll_plus_plus_const::{BIAS_DATA, RAW_ESTIMATE_DATA, THRESHOLDS}; use crate::hash_funcs::create_xxhash64_hashes; use crate::math_funcs::internal::normalize_float; @@ -425,7 +426,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 +467,14 @@ impl GroupsAccumulator for HllPlusPlusGroupsAccumulator { Ok(columns) } + fn convert_to_state( + &self, + values: &[ArrayRef], + opt_filter: Option<&BooleanArray>, + ) -> Result> { + convert_to_state_per_row(Self::new(self.p), values, opt_filter) + } + fn size(&self) -> usize { std::mem::size_of_val(self) + self.words.capacity() * std::mem::size_of::() @@ -639,9 +647,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/mod.rs b/native/spark-expr/src/agg_funcs/mod.rs index 9f7d0f5c07..c765530e72 100644 --- a/native/spark-expr/src/agg_funcs/mod.rs +++ b/native/spark-expr/src/agg_funcs/mod.rs @@ -18,6 +18,7 @@ mod approx_percentile; mod avg; mod avg_decimal; +mod convert_to_state; mod correlation; mod covariance; mod hll_plus_plus; diff --git a/native/spark-expr/src/agg_funcs/percentile.rs b/native/spark-expr/src/agg_funcs/percentile.rs index 3fddfdcb68..d25f588daf 100644 --- a/native/spark-expr/src/agg_funcs/percentile.rs +++ b/native/spark-expr/src/agg_funcs/percentile.rs @@ -15,6 +15,7 @@ // specific language governing permissions and limitations // under the License. +use super::convert_to_state::convert_to_state_per_row; use arrow::array::{ Array, ArrayRef, AsArray, BooleanArray, Float64Array, Float64Builder, ListArray, }; @@ -221,7 +222,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 +268,14 @@ impl GroupsAccumulator for SparkPercentileGroupsAccumulator { Ok(Arc::new(builder.finish())) } + fn convert_to_state( + &self, + values: &[ArrayRef], + opt_filter: Option<&BooleanArray>, + ) -> Result> { + convert_to_state_per_row(Self::new(self.percentile), values, opt_filter) + } + 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..9b9a5c0cbb 100644 --- a/native/spark-expr/src/agg_funcs/stddev.rs +++ b/native/spark-expr/src/agg_funcs/stddev.rs @@ -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,15 @@ impl GroupsAccumulator for StddevGroupsAccumulator { self.inner.state(emit_to) } + fn convert_to_state( + &self, + values: &[ArrayRef], + opt_filter: Option<&BooleanArray>, + ) -> Result> { + // State is identical to variance, so the inner accumulator can produce it. + self.inner.convert_to_state(values, opt_filter) + } + 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..4a2098c78f 100644 --- a/native/spark-expr/src/agg_funcs/sum_decimal.rs +++ b/native/spark-expr/src/agg_funcs/sum_decimal.rs @@ -15,6 +15,7 @@ // specific language governing permissions and limitations // under the License. +use super::convert_to_state::convert_to_state_per_row; use crate::utils::is_valid_decimal_precision; use crate::{decimal_sum_overflow_error, EvalMode, SparkErrorWithContext}; use arrow::array::{ @@ -541,14 +542,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 +604,24 @@ impl GroupsAccumulator for SumDecimalGroupsAccumulator { Ok(()) } + fn convert_to_state( + &self, + values: &[ArrayRef], + opt_filter: Option<&BooleanArray>, + ) -> DFResult> { + convert_to_state_per_row( + Self::new( + self.result_type.clone(), + self.precision, + self.eval_mode, + self.expr_id, + Arc::clone(&self.registry), + ), + values, + opt_filter, + ) + } + 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..99fb2b7953 100644 --- a/native/spark-expr/src/agg_funcs/sum_int.rs +++ b/native/spark-expr/src/agg_funcs/sum_int.rs @@ -15,6 +15,7 @@ // specific language governing permissions and limitations // under the License. +use super::convert_to_state::convert_to_state_per_row; use crate::{arithmetic_overflow_error, EvalMode}; use arrow::array::{ as_primitive_array, cast::AsArray, Array, ArrayRef, ArrowNativeTypeOp, ArrowPrimitiveType, @@ -495,14 +496,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 +524,14 @@ impl GroupsAccumulator for SumIntGroupsAccumulatorLegacy { Ok(()) } + fn convert_to_state( + &self, + values: &[ArrayRef], + opt_filter: Option<&BooleanArray>, + ) -> DFResult> { + convert_to_state_per_row(Self::new(), values, opt_filter) + } + fn size(&self) -> usize { std::mem::size_of_val(self) } @@ -642,14 +645,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 +677,14 @@ impl GroupsAccumulator for SumIntGroupsAccumulatorAnsi { Ok(()) } + fn convert_to_state( + &self, + values: &[ArrayRef], + opt_filter: Option<&BooleanArray>, + ) -> DFResult> { + convert_to_state_per_row(Self::new(), values, opt_filter) + } + fn size(&self) -> usize { std::mem::size_of_val(self) } @@ -826,14 +831,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 +885,14 @@ impl GroupsAccumulator for SumIntGroupsAccumulatorTry { Ok(()) } + fn convert_to_state( + &self, + values: &[ArrayRef], + opt_filter: Option<&BooleanArray>, + ) -> DFResult> { + convert_to_state_per_row(Self::new(), values, opt_filter) + } + 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..ea5b0bbb33 100644 --- a/native/spark-expr/src/agg_funcs/variance.rs +++ b/native/spark-expr/src/agg_funcs/variance.rs @@ -15,6 +15,7 @@ // specific language governing permissions and limitations // under the License. +use super::convert_to_state::convert_to_state_per_row; use arrow::array::{Array, ArrayRef, AsArray, BooleanArray, Float64Array}; use arrow::buffer::NullBuffer; use arrow::datatypes::{DataType, Field, FieldRef, Float64Type}; @@ -340,7 +341,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 +387,18 @@ impl GroupsAccumulator for VarianceGroupsAccumulator { ]) } + fn convert_to_state( + &self, + values: &[ArrayRef], + opt_filter: Option<&BooleanArray>, + ) -> Result> { + convert_to_state_per_row( + Self::new(self.stats_type, self.null_on_divide_by_zero), + values, + opt_filter, + ) + } + fn size(&self) -> usize { self.counts.capacity() * size_of::() + self.means.capacity() * size_of::() @@ -512,8 +524,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)?; } }