This is an automated email from the ASF dual-hosted git repository. mbutrovich pushed a commit to branch df55 in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 09ffef4aca80e05dd5db465308670da015ea3785 Author: Matt Butrovich <[email protected]> AuthorDate: Tue Aug 4 18:31:15 2026 -0400 bump to datafusion apache/main latest commit, use temp iceberg-rust --- native/Cargo.lock | 304 ++++++++++----------- native/Cargo.toml | 16 +- native/core/Cargo.toml | 2 +- native/core/src/execution/columnar_to_row.rs | 13 +- native/core/src/execution/merge_as_partial.rs | 18 +- .../core/src/execution/operators/iceberg_scan.rs | 2 + native/core/src/execution/planner.rs | 10 +- .../src/parquet/eager_page_index_reader_factory.rs | 6 +- native/shuffle/src/partitioners/traits.rs | 2 +- native/shuffle/src/writers/local/spill.rs | 34 +-- native/shuffle/src/writers/partition_writer.rs | 2 +- native/spark-expr/src/agg_funcs/avg.rs | 9 +- native/spark-expr/src/agg_funcs/avg_decimal.rs | 9 +- native/spark-expr/src/agg_funcs/correlation.rs | 17 +- native/spark-expr/src/agg_funcs/covariance.rs | 17 +- native/spark-expr/src/agg_funcs/hll_plus_plus.rs | 15 +- native/spark-expr/src/agg_funcs/percentile.rs | 11 +- native/spark-expr/src/agg_funcs/stddev.rs | 13 +- native/spark-expr/src/agg_funcs/sum_decimal.rs | 16 +- native/spark-expr/src/agg_funcs/sum_int.rs | 44 +-- native/spark-expr/src/agg_funcs/variance.rs | 15 +- 21 files changed, 313 insertions(+), 262 deletions(-) diff --git a/native/Cargo.lock b/native/Cargo.lock index eeaa9da5ce..d7a6f1be2b 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,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" @@ -1839,8 +1839,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=bc8b1a7a9323f69970fe728c95f3930cfbd5ad41#bc8b1a7a9323f69970fe728c95f3930cfbd5ad41"; dependencies = [ "arrow", "arrow-schema", @@ -1873,7 +1872,7 @@ dependencies = [ "datafusion-sql", "futures", "indexmap 2.14.0", - "itertools 0.14.0", + "itertools 0.15.0", "log", "object_store", "parking_lot", @@ -1888,8 +1887,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=bc8b1a7a9323f69970fe728c95f3930cfbd5ad41#bc8b1a7a9323f69970fe728c95f3930cfbd5ad41"; dependencies = [ "arrow", "async-trait", @@ -1903,7 +1901,7 @@ dependencies = [ "datafusion-physical-plan", "datafusion-session", "futures", - "itertools 0.14.0", + "itertools 0.15.0", "log", "object_store", "parking_lot", @@ -1913,8 +1911,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=bc8b1a7a9323f69970fe728c95f3930cfbd5ad41#bc8b1a7a9323f69970fe728c95f3930cfbd5ad41"; dependencies = [ "arrow", "async-trait", @@ -1928,9 +1925,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 +2057,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 +2080,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=bc8b1a7a9323f69970fe728c95f3930cfbd5ad41#bc8b1a7a9323f69970fe728c95f3930cfbd5ad41"; dependencies = [ "arrow", "arrow-ipc", @@ -2094,9 +2091,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 +2106,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=bc8b1a7a9323f69970fe728c95f3930cfbd5ad41#bc8b1a7a9323f69970fe728c95f3930cfbd5ad41"; dependencies = [ "futures", "log", @@ -2119,8 +2116,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=bc8b1a7a9323f69970fe728c95f3930cfbd5ad41#bc8b1a7a9323f69970fe728c95f3930cfbd5ad41"; dependencies = [ "arrow", "async-compression", @@ -2140,7 +2136,7 @@ dependencies = [ "flate2", "futures", "glob", - "itertools 0.14.0", + "itertools 0.15.0", "liblzma", "log", "object_store", @@ -2155,8 +2151,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=bc8b1a7a9323f69970fe728c95f3930cfbd5ad41#bc8b1a7a9323f69970fe728c95f3930cfbd5ad41"; dependencies = [ "arrow", "arrow-ipc", @@ -2171,7 +2166,7 @@ dependencies = [ "datafusion-physical-plan", "datafusion-session", "futures", - "itertools 0.14.0", + "itertools 0.15.0", "object_store", "tokio", ] @@ -2179,8 +2174,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=bc8b1a7a9323f69970fe728c95f3930cfbd5ad41#bc8b1a7a9323f69970fe728c95f3930cfbd5ad41"; dependencies = [ "arrow", "async-trait", @@ -2202,8 +2196,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=bc8b1a7a9323f69970fe728c95f3930cfbd5ad41#bc8b1a7a9323f69970fe728c95f3930cfbd5ad41"; dependencies = [ "arrow", "async-trait", @@ -2225,10 +2218,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=bc8b1a7a9323f69970fe728c95f3930cfbd5ad41#bc8b1a7a9323f69970fe728c95f3930cfbd5ad41"; dependencies = [ "arrow", + "arrow-schema", "async-trait", "bytes", "datafusion-common", @@ -2245,7 +2238,7 @@ dependencies = [ "datafusion-pruning", "datafusion-session", "futures", - "itertools 0.14.0", + "itertools 0.15.0", "log", "object_store", "parking_lot", @@ -2256,18 +2249,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=bc8b1a7a9323f69970fe728c95f3930cfbd5ad41#bc8b1a7a9323f69970fe728c95f3930cfbd5ad41"; [[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=bc8b1a7a9323f69970fe728c95f3930cfbd5ad41#bc8b1a7a9323f69970fe728c95f3930cfbd5ad41"; dependencies = [ "arrow", "arrow-buffer", "async-trait", + "bytes", "dashmap", "datafusion-common", "datafusion-expr", @@ -2277,16 +2269,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=bc8b1a7a9323f69970fe728c95f3930cfbd5ad41#bc8b1a7a9323f69970fe728c95f3930cfbd5ad41"; dependencies = [ "arrow", "arrow-schema", @@ -2299,7 +2293,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 +2301,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=bc8b1a7a9323f69970fe728c95f3930cfbd5ad41#bc8b1a7a9323f69970fe728c95f3930cfbd5ad41"; 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=bc8b1a7a9323f69970fe728c95f3930cfbd5ad41#bc8b1a7a9323f69970fe728c95f3930cfbd5ad41"; dependencies = [ "arrow", "arrow-buffer", - "base64 0.22.1", + "base64 0.23.1", "blake2", "blake3", "chrono", @@ -2337,7 +2329,7 @@ dependencies = [ "datafusion-macros", "datafusion-physical-expr-common", "hex", - "itertools 0.14.0", + "itertools 0.15.0", "log", "md-5 0.11.0", "memchr", @@ -2351,8 +2343,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=bc8b1a7a9323f69970fe728c95f3930cfbd5ad41#bc8b1a7a9323f69970fe728c95f3930cfbd5ad41"; dependencies = [ "arrow", "datafusion-common", @@ -2363,8 +2354,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 +2363,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=bc8b1a7a9323f69970fe728c95f3930cfbd5ad41#bc8b1a7a9323f69970fe728c95f3930cfbd5ad41"; dependencies = [ "arrow", "datafusion-common", @@ -2384,8 +2374,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=bc8b1a7a9323f69970fe728c95f3930cfbd5ad41#bc8b1a7a9323f69970fe728c95f3930cfbd5ad41"; dependencies = [ "arrow", "arrow-ord", @@ -2400,7 +2389,7 @@ dependencies = [ "datafusion-macros", "datafusion-physical-expr-common", "hashbrown 0.17.1", - "itertools 0.14.0", + "itertools 0.15.0", "itoa", "log", "memchr", @@ -2409,8 +2398,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=bc8b1a7a9323f69970fe728c95f3930cfbd5ad41#bc8b1a7a9323f69970fe728c95f3930cfbd5ad41"; dependencies = [ "arrow", "async-trait", @@ -2425,8 +2413,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=bc8b1a7a9323f69970fe728c95f3930cfbd5ad41#bc8b1a7a9323f69970fe728c95f3930cfbd5ad41"; dependencies = [ "arrow", "datafusion-common", @@ -2442,8 +2429,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=bc8b1a7a9323f69970fe728c95f3930cfbd5ad41#bc8b1a7a9323f69970fe728c95f3930cfbd5ad41"; dependencies = [ "datafusion-common", "datafusion-physical-expr-common", @@ -2452,19 +2438,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=bc8b1a7a9323f69970fe728c95f3930cfbd5ad41#bc8b1a7a9323f69970fe728c95f3930cfbd5ad41"; 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=bc8b1a7a9323f69970fe728c95f3930cfbd5ad41#bc8b1a7a9323f69970fe728c95f3930cfbd5ad41"; dependencies = [ "arrow", "chrono", @@ -2473,7 +2457,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 +2466,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=bc8b1a7a9323f69970fe728c95f3930cfbd5ad41#bc8b1a7a9323f69970fe728c95f3930cfbd5ad41"; dependencies = [ "arrow", "datafusion-common", @@ -2494,7 +2477,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 +2486,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=bc8b1a7a9323f69970fe728c95f3930cfbd5ad41#bc8b1a7a9323f69970fe728c95f3930cfbd5ad41"; dependencies = [ "arrow", "datafusion-common", @@ -2512,14 +2494,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=bc8b1a7a9323f69970fe728c95f3930cfbd5ad41#bc8b1a7a9323f69970fe728c95f3930cfbd5ad41"; dependencies = [ "arrow", "chrono", @@ -2527,7 +2508,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 +2516,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=bc8b1a7a9323f69970fe728c95f3930cfbd5ad41#bc8b1a7a9323f69970fe728c95f3930cfbd5ad41"; dependencies = [ "arrow", "datafusion-common", @@ -2547,14 +2527,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=bc8b1a7a9323f69970fe728c95f3930cfbd5ad41#bc8b1a7a9323f69970fe728c95f3930cfbd5ad41"; dependencies = [ "arrow", "arrow-data", @@ -2562,6 +2542,7 @@ dependencies = [ "arrow-ord", "arrow-schema", "async-trait", + "bytes", "datafusion-common", "datafusion-common-runtime", "datafusion-execution", @@ -2575,19 +2556,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=bc8b1a7a9323f69970fe728c95f3930cfbd5ad41#bc8b1a7a9323f69970fe728c95f3930cfbd5ad41"; dependencies = [ "arrow", "datafusion-common", @@ -2602,9 +2583,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=bc8b1a7a9323f69970fe728c95f3930cfbd5ad41#bc8b1a7a9323f69970fe728c95f3930cfbd5ad41"; dependencies = [ + "arrow-schema", "async-trait", "datafusion-common", "datafusion-execution", @@ -2616,8 +2597,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=bc8b1a7a9323f69970fe728c95f3930cfbd5ad41#bc8b1a7a9323f69970fe728c95f3930cfbd5ad41"; dependencies = [ "arrow", "bigdecimal", @@ -2646,8 +2626,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=bc8b1a7a9323f69970fe728c95f3930cfbd5ad41#bc8b1a7a9323f69970fe728c95f3930cfbd5ad41"; dependencies = [ "arrow", "bigdecimal", @@ -3497,7 +3476,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=f9473d9790fbc6bded1551f846a8603d75f59c7b#f9473d9790fbc6bded1551f846a8603d75f59c7b"; dependencies = [ "aes-gcm", "anyhow", @@ -3523,6 +3502,7 @@ dependencies = [ "fastnum", "flate2", "fnv", + "form_urlencoded", "futures", "itertools 0.13.0", "moka", @@ -3553,7 +3533,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=f9473d9790fbc6bded1551f846a8603d75f59c7b#f9473d9790fbc6bded1551f846a8603d75f59c7b"; dependencies = [ "anyhow", "async-trait", @@ -3729,12 +3709,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" @@ -4794,9 +4768,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 +4800,6 @@ dependencies = [ "seq-macro", "simdutf8", "snap", - "thrift", "tokio", "twox-hash", "zstd", @@ -4834,9 +4807,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 +4823,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 +4840,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 +5479,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 +5502,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 +5518,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 +5538,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 +5576,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 +5600,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 +5611,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 +6488,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" @@ -7466,9 +7442,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..ab129bca0a 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 = "bc8b1a7a9323f69970fe728c95f3930cfbd5ad41", default-features = false, features = ["unicode_expressions", "crypto_expressions", "nested_expressions", "parquet"] } +datafusion-datasource = { git = "https://github.com/apache/datafusion.git";, rev = "bc8b1a7a9323f69970fe728c95f3930cfbd5ad41" } +datafusion-physical-expr-adapter = { git = "https://github.com/apache/datafusion.git";, rev = "bc8b1a7a9323f69970fe728c95f3930cfbd5ad41" } +datafusion-spark = { git = "https://github.com/apache/datafusion.git";, rev = "bc8b1a7a9323f69970fe728c95f3930cfbd5ad41", 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 = "f9473d9790fbc6bded1551f846a8603d75f59c7b" } +iceberg-storage-opendal = { git = "https://github.com/mbutrovich/iceberg-rust";, rev = "f9473d9790fbc6bded1551f846a8603d75f59c7b", 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..0a685b0f37 100644 --- a/native/core/Cargo.toml +++ b/native/core/Cargo.toml @@ -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 = "bc8b1a7a9323f69970fe728c95f3930cfbd5ad41" } [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 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..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<Vec<ArrayRef>> { + datafusion::common::not_impl_err!("Input batch conversion to state not implemented") } fn evaluate(&mut self, emit_to: EmitTo) -> Result<ArrayRef> { 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/planner.rs b/native/core/src/execution/planner.rs index f20dadf7f3..da308c0108 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -110,7 +110,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; @@ -1995,7 +1995,11 @@ impl PhysicalPlanner { }); let unnest_options = UnnestOptions { - preserve_nulls: explode.outer, + null_handling: if explode.outer { + NullHandling::Preserve + } else { + NullHandling::Drop + }, recursions: vec![], }; @@ -4065,6 +4069,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<dyn ObjectStore>, - metadata_cache: Arc<dyn FileMetadataCache>, + metadata_cache: Arc<FileMetadataCache>, } impl EagerPageIndexReaderFactory { - pub fn new(store: Arc<dyn ObjectStore>, metadata_cache: Arc<dyn FileMetadataCache>) -> Self { + pub fn new(store: Arc<dyn ObjectStore>, metadata_cache: Arc<FileMetadataCache>) -> Self { Self { store, metadata_cache, @@ -118,7 +118,7 @@ struct EagerPageIndexReader { store: Arc<dyn ObjectStore>, inner: ParquetObjectReader, partitioned_file: PartitionedFile, - metadata_cache: Arc<dyn FileMetadataCache>, + metadata_cache: Arc<FileMetadataCache>, metadata_size_hint: Option<usize>, } 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<dyn DfSpillFile>, + writer: Box<dyn DfSpillWriter>, } pub(crate) struct SpillWriter { shuffle_block_writer: ShuffleBlockWriter, write_buffer_size: usize, batch_size: usize, - spill_file: Option<SpillFile>, + spill_file: Option<ActiveSpillFile>, } 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/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<Vec<ArrayRef>> { + not_impl_err!("Input batch conversion to state not implemented") + } + fn size(&self) -> usize { self.counts.capacity() * std::mem::size_of::<i64>() + self.sums.capacity() * std::mem::size_of::<T>() 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<Vec<ArrayRef>> { + not_impl_err!("Input batch conversion to state not implemented") + } + fn size(&self) -> usize { self.counts.capacity() * std::mem::size_of::<i64>() + self.sums.capacity() * std::mem::size_of::<i128>() 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<Vec<ArrayRef>> { + 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<Vec<ArrayRef>> { + 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<Vec<ArrayRef>> { + 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::<i64>() @@ -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::<i32>(); @@ -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<Vec<ArrayRef>> { + 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<ArrayRef> { @@ -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<Vec<ArrayRef>> { + 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<Vec<ArrayRef>> { + not_impl_err!("Input batch conversion to state not implemented") + } + fn size(&self) -> usize { self.sum.capacity() * std::mem::size_of::<Option<i128>>() + self.is_empty.capacity() * std::mem::size_of::<bool>() 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<Vec<ArrayRef>> { + 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<Vec<ArrayRef>> { + 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<Vec<ArrayRef>> { + 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<Vec<ArrayRef>> { + not_impl_err!("Input batch conversion to state not implemented") + } + fn size(&self) -> usize { self.counts.capacity() * size_of::<f64>() + self.means.capacity() * size_of::<f64>() @@ -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); --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
