From 09ffef4aca80e05dd5db465308670da015ea3785 Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Tue, 4 Aug 2026 18:31:15 -0400 Subject: [PATCH 1/2] 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 +- .../src/execution/operators/iceberg_scan.rs | 2 + native/core/src/execution/planner.rs | 10 +- .../eager_page_index_reader_factory.rs | 6 +- native/shuffle/src/partitioners/traits.rs | 2 +- native/shuffle/src/writers/local/spill.rs | 34 +- .../shuffle/src/writers/partition_writer.rs | 2 +- native/spark-expr/src/agg_funcs/avg.rs | 9 +- .../spark-expr/src/agg_funcs/avg_decimal.rs | 9 +- .../spark-expr/src/agg_funcs/correlation.rs | 17 +- native/spark-expr/src/agg_funcs/covariance.rs | 17 +- .../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 +- .../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> { + datafusion::common::not_impl_err!("Input batch conversion to state not implemented") } fn evaluate(&mut self, emit_to: EmitTo) -> Result { diff --git a/native/core/src/execution/operators/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, - 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/shuffle/src/partitioners/traits.rs b/native/shuffle/src/partitioners/traits.rs index 9572b70db5..5bd41a62a5 100644 --- a/native/shuffle/src/partitioners/traits.rs +++ b/native/shuffle/src/partitioners/traits.rs @@ -19,7 +19,7 @@ use arrow::record_batch::RecordBatch; use datafusion::common::Result; #[async_trait::async_trait] -pub(crate) trait ShufflePartitioner: Send + Sync { +pub(crate) trait ShufflePartitioner: Send { /// Insert a batch into the partitioner async fn insert_batch(&mut self, batch: RecordBatch) -> Result<()>; /// Write shuffle data and shuffle index file to disk diff --git a/native/shuffle/src/writers/local/spill.rs b/native/shuffle/src/writers/local/spill.rs index 450b617421..7304f5d19d 100644 --- a/native/shuffle/src/writers/local/spill.rs +++ b/native/shuffle/src/writers/local/spill.rs @@ -19,21 +19,21 @@ use crate::metrics::ShufflePartitionerMetrics; use crate::writers::BufBatchWriter; use crate::ShuffleBlockWriter; use arrow::record_batch::RecordBatch; -use datafusion::common::DataFusionError; -use datafusion::execution::disk_manager::RefCountedTempFile; use datafusion::execution::runtime_env::RuntimeEnv; -use std::fs::{File, OpenOptions}; +use datafusion::execution::SpillFile as DfSpillFile; +use datafusion::execution::SpillWriter as DfSpillWriter; +use std::sync::Arc; -struct SpillFile { - temp_file: RefCountedTempFile, - file: File, +struct ActiveSpillFile { + temp_file: Arc, + writer: Box, } pub(crate) struct SpillWriter { shuffle_block_writer: ShuffleBlockWriter, write_buffer_size: usize, batch_size: usize, - spill_file: Option, + spill_file: Option, } impl SpillWriter { @@ -62,7 +62,7 @@ impl SpillWriter { let total_bytes_written = { let mut buf_batch_writer = BufBatchWriter::new( &mut self.shuffle_block_writer, - &mut self.spill_file.as_mut().unwrap().file, + &mut self.spill_file.as_mut().unwrap().writer, self.write_buffer_size, self.batch_size, ); @@ -90,21 +90,11 @@ impl SpillWriter { ) -> datafusion::common::Result<()> { if self.spill_file.is_none() { // Spill file is not yet created, create it - let spill_file = runtime + let temp_file = runtime .disk_manager .create_tmp_file("shuffle writer spill")?; - let spill_data = OpenOptions::new() - .write(true) - .create(true) - .truncate(true) - .open(spill_file.path()) - .map_err(|e| { - DataFusionError::Execution(format!("Error occurred while spilling {e}")) - })?; - self.spill_file = Some(SpillFile { - temp_file: spill_file, - file: spill_data, - }); + let writer = temp_file.open_writer()?; + self.spill_file = Some(ActiveSpillFile { temp_file, writer }); } Ok(()) } @@ -112,7 +102,7 @@ impl SpillWriter { pub(crate) fn path(&self) -> Option<&std::path::Path> { self.spill_file .as_ref() - .map(|spill_file| spill_file.temp_file.path()) + .and_then(|spill_file| spill_file.temp_file.path()) } #[cfg(test)] diff --git a/native/shuffle/src/writers/partition_writer.rs b/native/shuffle/src/writers/partition_writer.rs index 25b0e598df..9b415f897d 100644 --- a/native/shuffle/src/writers/partition_writer.rs +++ b/native/shuffle/src/writers/partition_writer.rs @@ -32,7 +32,7 @@ use arrow::record_batch::RecordBatch; /// ascending id order, then a single [`finish_all`](PartitionWriter::finish_all). /// /// [`LocalPartitionWriter`]: crate::writers::local::local_partition_writer::LocalPartitionWriter -pub(crate) trait PartitionWriter: Send + Sync { +pub(crate) trait PartitionWriter: Send { /// Stages the batches from `iter` for partition `pid` without finalizing it. /// /// Used to stream single-partition output and to stage multi-partition diff --git a/native/spark-expr/src/agg_funcs/avg.rs b/native/spark-expr/src/agg_funcs/avg.rs index 24a9a30991..bb5371568b 100644 --- a/native/spark-expr/src/agg_funcs/avg.rs +++ b/native/spark-expr/src/agg_funcs/avg.rs @@ -283,7 +283,6 @@ where &mut self, values: &[ArrayRef], group_indices: &[usize], - _opt_filter: Option<&arrow::array::BooleanArray>, total_num_groups: usize, ) -> Result<()> { assert_eq!(values.len(), 2, "two arguments to merge_batch"); @@ -340,6 +339,14 @@ where ]) } + fn convert_to_state( + &self, + _values: &[ArrayRef], + _opt_filter: Option<&arrow::array::BooleanArray>, + ) -> Result> { + not_impl_err!("Input batch conversion to state not implemented") + } + fn size(&self) -> usize { self.counts.capacity() * std::mem::size_of::() + self.sums.capacity() * std::mem::size_of::() diff --git a/native/spark-expr/src/agg_funcs/avg_decimal.rs b/native/spark-expr/src/agg_funcs/avg_decimal.rs index 186e027244..0e02dd03d4 100644 --- a/native/spark-expr/src/agg_funcs/avg_decimal.rs +++ b/native/spark-expr/src/agg_funcs/avg_decimal.rs @@ -543,7 +543,6 @@ impl GroupsAccumulator for AvgDecimalGroupsAccumulator { &mut self, values: &[ArrayRef], group_indices: &[usize], - _opt_filter: Option<&arrow::array::BooleanArray>, total_num_groups: usize, ) -> Result<()> { assert_eq!(values.len(), 2, "two arguments to merge_batch"); @@ -652,6 +651,14 @@ impl GroupsAccumulator for AvgDecimalGroupsAccumulator { ]) } + fn convert_to_state( + &self, + _values: &[ArrayRef], + _opt_filter: Option<&arrow::array::BooleanArray>, + ) -> Result> { + not_impl_err!("Input batch conversion to state not implemented") + } + fn size(&self) -> usize { self.counts.capacity() * std::mem::size_of::() + self.sums.capacity() * std::mem::size_of::() diff --git a/native/spark-expr/src/agg_funcs/correlation.rs b/native/spark-expr/src/agg_funcs/correlation.rs index 189b76c863..d69ac61def 100644 --- a/native/spark-expr/src/agg_funcs/correlation.rs +++ b/native/spark-expr/src/agg_funcs/correlation.rs @@ -24,7 +24,7 @@ use crate::agg_funcs::covariance::{CovarianceAccumulator, CovarianceGroupsAccumu use crate::agg_funcs::stddev::StddevAccumulator; use crate::agg_funcs::variance::VarianceGroupsAccumulator; use arrow::compute::filter; -use datafusion::common::{Result, ScalarValue}; +use datafusion::common::{not_impl_err, Result, ScalarValue}; use datafusion::logical_expr::function::{AccumulatorArgs, StateFieldsArgs}; use datafusion::logical_expr::{ Accumulator, AggregateUDFImpl, EmitTo, GroupsAccumulator, Signature, Volatility, @@ -335,7 +335,6 @@ impl GroupsAccumulator for CorrelationGroupsAccumulator { &mut self, values: &[ArrayRef], group_indices: &[usize], - opt_filter: Option<&BooleanArray>, total_num_groups: usize, ) -> Result<()> { assert_eq!(values.len(), 6, "six state columns to merge_batch"); @@ -358,11 +357,11 @@ impl GroupsAccumulator for CorrelationGroupsAccumulator { ]; self.covar - .merge_batch(&covar_state, group_indices, opt_filter, total_num_groups)?; + .merge_batch(&covar_state, group_indices, total_num_groups)?; self.var1 - .merge_batch(&var1_state, group_indices, opt_filter, total_num_groups)?; + .merge_batch(&var1_state, group_indices, total_num_groups)?; self.var2 - .merge_batch(&var2_state, group_indices, opt_filter, total_num_groups)?; + .merge_batch(&var2_state, group_indices, total_num_groups)?; Ok(()) } @@ -442,6 +441,14 @@ impl GroupsAccumulator for CorrelationGroupsAccumulator { ]) } + fn convert_to_state( + &self, + _values: &[ArrayRef], + _opt_filter: Option<&BooleanArray>, + ) -> Result> { + not_impl_err!("Input batch conversion to state not implemented") + } + fn size(&self) -> usize { self.covar.size() + self.var1.size() + self.var2.size() } diff --git a/native/spark-expr/src/agg_funcs/covariance.rs b/native/spark-expr/src/agg_funcs/covariance.rs index e5b77ad33b..548c118720 100644 --- a/native/spark-expr/src/agg_funcs/covariance.rs +++ b/native/spark-expr/src/agg_funcs/covariance.rs @@ -21,7 +21,9 @@ use arrow::array::{Array, ArrayRef, AsArray, BooleanArray, Float64Array}; use arrow::buffer::NullBuffer; use arrow::compute::cast; use arrow::datatypes::{DataType, Field, FieldRef, Float64Type}; -use datafusion::common::{downcast_value, unwrap_or_internal_err, Result, ScalarValue}; +use datafusion::common::{ + downcast_value, not_impl_err, unwrap_or_internal_err, Result, ScalarValue, +}; use datafusion::logical_expr::function::{AccumulatorArgs, StateFieldsArgs}; use datafusion::logical_expr::{ Accumulator, AggregateUDFImpl, EmitTo, GroupsAccumulator, Signature, Volatility, @@ -414,7 +416,6 @@ impl GroupsAccumulator for CovarianceGroupsAccumulator { &mut self, values: &[ArrayRef], group_indices: &[usize], - _opt_filter: Option<&BooleanArray>, total_num_groups: usize, ) -> Result<()> { assert_eq!(values.len(), 4, "four arguments to merge_batch"); @@ -467,6 +468,14 @@ impl GroupsAccumulator for CovarianceGroupsAccumulator { ]) } + fn convert_to_state( + &self, + _values: &[ArrayRef], + _opt_filter: Option<&BooleanArray>, + ) -> Result> { + not_impl_err!("Input batch conversion to state not implemented") + } + fn size(&self) -> usize { (self.counts.capacity() + self.mean1s.capacity() @@ -563,8 +572,8 @@ mod groups_tests { let rstate = right.state(EmitTo::All).unwrap(); let mut merged = pop(); - merged.merge_batch(&lstate, &[0], None, 1).unwrap(); - merged.merge_batch(&rstate, &[0], None, 1).unwrap(); + merged.merge_batch(&lstate, &[0], 1).unwrap(); + merged.merge_batch(&rstate, &[0], 1).unwrap(); let merged_result = evaluate(&mut merged)[0].unwrap(); assert!((single - merged_result).abs() < 1e-12); diff --git a/native/spark-expr/src/agg_funcs/hll_plus_plus.rs b/native/spark-expr/src/agg_funcs/hll_plus_plus.rs index 01fa13800a..3cb39f0acd 100644 --- a/native/spark-expr/src/agg_funcs/hll_plus_plus.rs +++ b/native/spark-expr/src/agg_funcs/hll_plus_plus.rs @@ -30,7 +30,7 @@ use arrow::array::{ Array, ArrayRef, AsArray, BooleanArray, Float32Array, Float64Array, Int64Array, }; use arrow::datatypes::{DataType, Field, FieldRef, Float32Type, Float64Type}; -use datafusion::common::{Result, ScalarValue}; +use datafusion::common::{not_impl_err, Result, ScalarValue}; use datafusion::logical_expr::function::{AccumulatorArgs, StateFieldsArgs}; use datafusion::logical_expr::{ Accumulator, AggregateUDFImpl, EmitTo, GroupsAccumulator, Signature, Volatility, @@ -425,7 +425,6 @@ impl GroupsAccumulator for HllPlusPlusGroupsAccumulator { &mut self, values: &[ArrayRef], group_indices: &[usize], - _opt_filter: Option<&BooleanArray>, total_num_groups: usize, ) -> Result<()> { self.resize(total_num_groups); @@ -467,6 +466,14 @@ impl GroupsAccumulator for HllPlusPlusGroupsAccumulator { Ok(columns) } + fn convert_to_state( + &self, + _values: &[ArrayRef], + _opt_filter: Option<&BooleanArray>, + ) -> Result> { + not_impl_err!("Input batch conversion to state not implemented") + } + fn size(&self) -> usize { std::mem::size_of_val(self) + self.words.capacity() * std::mem::size_of::() @@ -639,9 +646,7 @@ mod tests { for part in [&mut left, &mut right] { let state = part.state(EmitTo::All).unwrap(); let n = state[0].len(); - merged - .merge_batch(&state, &vec![0usize; n], None, 1) - .unwrap(); + merged.merge_batch(&state, &vec![0usize; n], 1).unwrap(); } let single = { let mut a = acc(9); diff --git a/native/spark-expr/src/agg_funcs/percentile.rs b/native/spark-expr/src/agg_funcs/percentile.rs index 3fddfdcb68..78e585a396 100644 --- a/native/spark-expr/src/agg_funcs/percentile.rs +++ b/native/spark-expr/src/agg_funcs/percentile.rs @@ -20,7 +20,7 @@ use arrow::array::{ }; use arrow::buffer::{OffsetBuffer, ScalarBuffer}; use arrow::datatypes::{DataType, Field, FieldRef, Float64Type}; -use datafusion::common::{internal_err, plan_err, Result, ScalarValue}; +use datafusion::common::{internal_err, not_impl_err, plan_err, Result, ScalarValue}; use datafusion::logical_expr::function::{AccumulatorArgs, StateFieldsArgs}; use datafusion::logical_expr::Volatility::Immutable; use datafusion::logical_expr::{ @@ -221,7 +221,6 @@ impl GroupsAccumulator for SparkPercentileGroupsAccumulator { &mut self, values: &[ArrayRef], group_indices: &[usize], - _opt_filter: Option<&BooleanArray>, total_num_groups: usize, ) -> Result<()> { let input_group_values = values[0].as_list::(); @@ -268,6 +267,14 @@ impl GroupsAccumulator for SparkPercentileGroupsAccumulator { Ok(Arc::new(builder.finish())) } + fn convert_to_state( + &self, + _values: &[ArrayRef], + _opt_filter: Option<&BooleanArray>, + ) -> Result> { + not_impl_err!("Input batch conversion to state not implemented") + } + fn size(&self) -> usize { self.group_values .iter() diff --git a/native/spark-expr/src/agg_funcs/stddev.rs b/native/spark-expr/src/agg_funcs/stddev.rs index 3389a86af6..bbceaa72dc 100644 --- a/native/spark-expr/src/agg_funcs/stddev.rs +++ b/native/spark-expr/src/agg_funcs/stddev.rs @@ -22,7 +22,7 @@ use arrow::array::{ArrayRef, AsArray, BooleanArray, Float64Array}; use arrow::datatypes::FieldRef; use arrow::datatypes::{DataType, Field, Float64Type}; use datafusion::common::types::NativeType; -use datafusion::common::{internal_err, Result, ScalarValue}; +use datafusion::common::{internal_err, not_impl_err, Result, ScalarValue}; use datafusion::logical_expr::function::{AccumulatorArgs, StateFieldsArgs}; use datafusion::logical_expr::{ Accumulator, AggregateUDFImpl, Coercion, EmitTo, GroupsAccumulator, Signature, Volatility, @@ -231,11 +231,10 @@ impl GroupsAccumulator for StddevGroupsAccumulator { &mut self, values: &[ArrayRef], group_indices: &[usize], - opt_filter: Option<&BooleanArray>, total_num_groups: usize, ) -> Result<()> { self.inner - .merge_batch(values, group_indices, opt_filter, total_num_groups) + .merge_batch(values, group_indices, total_num_groups) } fn evaluate(&mut self, emit_to: EmitTo) -> Result { @@ -250,6 +249,14 @@ impl GroupsAccumulator for StddevGroupsAccumulator { self.inner.state(emit_to) } + fn convert_to_state( + &self, + _values: &[ArrayRef], + _opt_filter: Option<&BooleanArray>, + ) -> Result> { + not_impl_err!("Input batch conversion to state not implemented") + } + fn size(&self) -> usize { self.inner.size() } diff --git a/native/spark-expr/src/agg_funcs/sum_decimal.rs b/native/spark-expr/src/agg_funcs/sum_decimal.rs index 4ddf13c5d6..cae0a70b42 100644 --- a/native/spark-expr/src/agg_funcs/sum_decimal.rs +++ b/native/spark-expr/src/agg_funcs/sum_decimal.rs @@ -21,7 +21,7 @@ use arrow::array::{ cast::AsArray, types::Decimal128Type, Array, ArrayRef, BooleanArray, Decimal128Array, }; use arrow::datatypes::{DataType, Field, FieldRef}; -use datafusion::common::{DataFusionError, Result as DFResult, ScalarValue}; +use datafusion::common::{not_impl_err, DataFusionError, Result as DFResult, ScalarValue}; use datafusion::logical_expr::function::{AccumulatorArgs, StateFieldsArgs}; use datafusion::logical_expr::Volatility::Immutable; use datafusion::logical_expr::{ @@ -541,14 +541,8 @@ impl GroupsAccumulator for SumDecimalGroupsAccumulator { &mut self, values: &[ArrayRef], group_indices: &[usize], - opt_filter: Option<&BooleanArray>, total_num_groups: usize, ) -> DFResult<()> { - debug_assert!( - opt_filter.is_none(), - "opt_filter is not supported in merge_batch" - ); - self.resize_helper(total_num_groups); // For decimal sum, always expect 2 arrays regardless of eval_mode @@ -609,6 +603,14 @@ impl GroupsAccumulator for SumDecimalGroupsAccumulator { Ok(()) } + fn convert_to_state( + &self, + _values: &[ArrayRef], + _opt_filter: Option<&BooleanArray>, + ) -> DFResult> { + not_impl_err!("Input batch conversion to state not implemented") + } + fn size(&self) -> usize { self.sum.capacity() * std::mem::size_of::>() + self.is_empty.capacity() * std::mem::size_of::() diff --git a/native/spark-expr/src/agg_funcs/sum_int.rs b/native/spark-expr/src/agg_funcs/sum_int.rs index 59a4371772..56a8c122c4 100644 --- a/native/spark-expr/src/agg_funcs/sum_int.rs +++ b/native/spark-expr/src/agg_funcs/sum_int.rs @@ -23,7 +23,7 @@ use arrow::array::{ use arrow::datatypes::{ ArrowNativeType, DataType, Field, FieldRef, Int16Type, Int32Type, Int64Type, Int8Type, }; -use datafusion::common::{DataFusionError, Result as DFResult, ScalarValue}; +use datafusion::common::{not_impl_err, DataFusionError, Result as DFResult, ScalarValue}; use datafusion::logical_expr::function::{AccumulatorArgs, StateFieldsArgs}; use datafusion::logical_expr::Volatility::Immutable; use datafusion::logical_expr::{ @@ -495,14 +495,8 @@ impl GroupsAccumulator for SumIntGroupsAccumulatorLegacy { &mut self, values: &[ArrayRef], group_indices: &[usize], - opt_filter: Option<&BooleanArray>, total_num_groups: usize, ) -> DFResult<()> { - debug_assert!( - opt_filter.is_none(), - "opt_filter is not supported in merge_batch" - ); - if values.len() != 1 { return Err(DataFusionError::Internal(format!( "Invalid state while merging batch. Expected 1 element but found {}", @@ -529,6 +523,14 @@ impl GroupsAccumulator for SumIntGroupsAccumulatorLegacy { Ok(()) } + fn convert_to_state( + &self, + _values: &[ArrayRef], + _opt_filter: Option<&BooleanArray>, + ) -> DFResult> { + not_impl_err!("Input batch conversion to state not implemented") + } + fn size(&self) -> usize { std::mem::size_of_val(self) } @@ -642,14 +644,8 @@ impl GroupsAccumulator for SumIntGroupsAccumulatorAnsi { &mut self, values: &[ArrayRef], group_indices: &[usize], - opt_filter: Option<&BooleanArray>, total_num_groups: usize, ) -> DFResult<()> { - debug_assert!( - opt_filter.is_none(), - "opt_filter is not supported in merge_batch" - ); - if values.len() != 1 { return Err(DataFusionError::Internal(format!( "Invalid state while merging batch. Expected 1 element but found {}", @@ -680,6 +676,14 @@ impl GroupsAccumulator for SumIntGroupsAccumulatorAnsi { Ok(()) } + fn convert_to_state( + &self, + _values: &[ArrayRef], + _opt_filter: Option<&BooleanArray>, + ) -> DFResult> { + not_impl_err!("Input batch conversion to state not implemented") + } + fn size(&self) -> usize { std::mem::size_of_val(self) } @@ -826,14 +830,8 @@ impl GroupsAccumulator for SumIntGroupsAccumulatorTry { &mut self, values: &[ArrayRef], group_indices: &[usize], - opt_filter: Option<&BooleanArray>, total_num_groups: usize, ) -> DFResult<()> { - debug_assert!( - opt_filter.is_none(), - "opt_filter is not supported in merge_batch" - ); - if values.len() != 2 { return Err(DataFusionError::Internal(format!( "Invalid state while merging batch. Expected 2 elements but found {}", @@ -886,6 +884,14 @@ impl GroupsAccumulator for SumIntGroupsAccumulatorTry { Ok(()) } + fn convert_to_state( + &self, + _values: &[ArrayRef], + _opt_filter: Option<&BooleanArray>, + ) -> DFResult> { + not_impl_err!("Input batch conversion to state not implemented") + } + fn size(&self) -> usize { std::mem::size_of_val(self) } diff --git a/native/spark-expr/src/agg_funcs/variance.rs b/native/spark-expr/src/agg_funcs/variance.rs index f1841729cf..57a8f6da50 100644 --- a/native/spark-expr/src/agg_funcs/variance.rs +++ b/native/spark-expr/src/agg_funcs/variance.rs @@ -18,7 +18,7 @@ use arrow::array::{Array, ArrayRef, AsArray, BooleanArray, Float64Array}; use arrow::buffer::NullBuffer; use arrow::datatypes::{DataType, Field, FieldRef, Float64Type}; -use datafusion::common::{downcast_value, Result, ScalarValue}; +use datafusion::common::{downcast_value, not_impl_err, Result, ScalarValue}; use datafusion::logical_expr::function::{AccumulatorArgs, StateFieldsArgs}; use datafusion::logical_expr::Volatility::Immutable; use datafusion::logical_expr::{ @@ -340,7 +340,6 @@ impl GroupsAccumulator for VarianceGroupsAccumulator { &mut self, values: &[ArrayRef], group_indices: &[usize], - _opt_filter: Option<&BooleanArray>, total_num_groups: usize, ) -> Result<()> { assert_eq!(values.len(), 3, "three arguments to merge_batch"); @@ -387,6 +386,14 @@ impl GroupsAccumulator for VarianceGroupsAccumulator { ]) } + fn convert_to_state( + &self, + _values: &[ArrayRef], + _opt_filter: Option<&BooleanArray>, + ) -> Result> { + not_impl_err!("Input batch conversion to state not implemented") + } + fn size(&self) -> usize { self.counts.capacity() * size_of::() + self.means.capacity() * size_of::() @@ -512,8 +519,8 @@ mod groups_tests { let right_state = right.state(EmitTo::All).unwrap(); let mut merged = pop_acc(); - merged.merge_batch(&left_state, &[0], None, 1).unwrap(); - merged.merge_batch(&right_state, &[0], None, 1).unwrap(); + merged.merge_batch(&left_state, &[0], 1).unwrap(); + merged.merge_batch(&right_state, &[0], 1).unwrap(); let merged_result = evaluate(&mut merged)[0].unwrap(); assert!((single_result - merged_result).abs() < 1e-12); From bcd6a3aa933b705f93d5ba8cef85aff50d5c92ac Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Tue, 4 Aug 2026 19:39:18 -0400 Subject: [PATCH 2/2] fix clippy, bump iceberg-rust commit and opendal deps --- native/Cargo.lock | 106 +++++++++++------- native/Cargo.toml | 4 +- native/core/Cargo.toml | 4 +- native/core/src/execution/operators/copy.rs | 8 +- .../src/execution/operators/parquet_writer.rs | 20 ++-- native/core/src/parquet/parquet_exec.rs | 11 +- native/core/src/parquet/parquet_support.rs | 10 +- .../src/writers/shuffle_block_writer.rs | 4 +- .../src/array_funcs/array_insert.rs | 24 ++-- .../spark-expr/src/array_funcs/array_slice.rs | 2 +- .../spark-expr/src/array_funcs/arrays_zip.rs | 6 +- .../src/array_funcs/list_extract.rs | 8 +- .../src/bloom_filter/spark_bit_array.rs | 5 +- 13 files changed, 118 insertions(+), 94 deletions(-) diff --git a/native/Cargo.lock b/native/Cargo.lock index d7a6f1be2b..aaf8287783 100644 --- a/native/Cargo.lock +++ b/native/Cargo.lock @@ -1591,6 +1591,16 @@ dependencies = [ "libc", ] +[[package]] +name = "crc-fast" +version = "1.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e75b2483e97a5a7da73ac68a05b629f9c53cff58d8ed1c77866079e18b00dba5" +dependencies = [ + "digest 0.10.7", + "spin 0.10.1", +] + [[package]] name = "crc32c" version = "0.6.8" @@ -3476,7 +3486,7 @@ dependencies = [ [[package]] name = "iceberg" version = "0.10.0" -source = "git+https://github.com/mbutrovich/iceberg-rust?rev=f9473d9790fbc6bded1551f846a8603d75f59c7b#f9473d9790fbc6bded1551f846a8603d75f59c7b" +source = "git+https://github.com/mbutrovich/iceberg-rust?rev=07de8fc8d650f7439b251d57a55613f7de693eb8#07de8fc8d650f7439b251d57a55613f7de693eb8" dependencies = [ "aes-gcm", "anyhow", @@ -3533,7 +3543,7 @@ dependencies = [ [[package]] name = "iceberg-storage-opendal" version = "0.10.0" -source = "git+https://github.com/mbutrovich/iceberg-rust?rev=f9473d9790fbc6bded1551f846a8603d75f59c7b#f9473d9790fbc6bded1551f846a8603d75f59c7b" +source = "git+https://github.com/mbutrovich/iceberg-rust?rev=07de8fc8d650f7439b251d57a55613f7de693eb8#07de8fc8d650f7439b251d57a55613f7de693eb8" dependencies = [ "anyhow", "async-trait", @@ -4449,9 +4459,9 @@ dependencies = [ [[package]] name = "object_store_opendal" -version = "0.57.0" +version = "0.58.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0eb12a624a41fce745838d0ef3701ff6c47797c13cd18ad3612fd2a3134fdbd8" +checksum = "88f165780495c17aa3ce86846600504198c3fffd99073521552751c2430fa6ac" dependencies = [ "async-trait", "bytes", @@ -4490,12 +4500,13 @@ checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381" [[package]] name = "opendal" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "96c9c85ce253ff87225e7669979d877a20c98a06604ec9d6dd5f4473e08f1ae1" +checksum = "4f20562cc7447fcc915fc5c23df305a412ea80a733c9f2fd9e2d267e2815be6d" dependencies = [ "ctor", "opendal-core", + "opendal-http-transport-reqwest", "opendal-layer-concurrent-limit", "opendal-layer-logging", "opendal-layer-retry", @@ -4510,24 +4521,22 @@ dependencies = [ [[package]] name = "opendal-core" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c4f8607c90e2c963a91467f50fb49fbc7fb3d573f88cea219ca59ccd3740b309" +checksum = "ec75551ff4cf3e57da98979f6a937aaa9ddb3915bf68cc17d03df733be6646ed" dependencies = [ "anyhow", - "base64 0.22.1", + "base64 0.23.1", "bytes", "futures", "http 1.5.0", - "http-body 1.1.0", "jiff", "log", "md-5 0.11.0", "mea", "percent-encoding", - "quick-xml 0.39.4", + "quick-xml 0.41.0", "reqsign-core", - "reqwest 0.13.4", "serde", "serde_json", "tokio", @@ -4536,11 +4545,25 @@ dependencies = [ "web-time", ] +[[package]] +name = "opendal-http-transport-reqwest" +version = "0.58.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ad4d4f19c3ce01126a30611f8e544eaa217104a278c889ac17c9374fe4f9e4ef" +dependencies = [ + "bytes", + "futures", + "http 1.5.0", + "http-body 1.1.0", + "opendal-core", + "reqwest 0.13.4", +] + [[package]] name = "opendal-layer-concurrent-limit" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0d6f81ba6960e3fae1882f253b114b21d7e444e1534f209c7737a79f6243eb6f" +checksum = "249ac5b0aa5a7a6c3737342d10456067937f9c9a6f3f02544271f7908ab91081" dependencies = [ "futures", "http 1.5.0", @@ -4550,9 +4573,9 @@ dependencies = [ [[package]] name = "opendal-layer-logging" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "58ada45c6d81d1aa4c9305d0c7d4bc317c59c85866a0908a2d75a7a978aa5ee2" +checksum = "5c75411ab00f77851ff086b686c1e9ca8175ac18c15afa2cb75b9036436cb06c" dependencies = [ "log", "opendal-core", @@ -4560,9 +4583,9 @@ dependencies = [ [[package]] name = "opendal-layer-retry" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7b2a25a718afb81fad81cb9a0580a1cb989221fa2317f888c6a37f8dad408eb7" +checksum = "80b7738bd5f233ad8da39af9b9316b9b7a4eaddd91e8e32a1e19b7030688121d" dependencies = [ "backon", "log", @@ -4571,9 +4594,9 @@ dependencies = [ [[package]] name = "opendal-layer-timeout" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1e91f731724c213af81e9d03517859c8fc47b4578e64ad61ae4f099f10fe36e3" +checksum = "a704141924500f3803c05ed871b53305d2a2f11cb5ef20160c3ee688a1857f66" dependencies = [ "opendal-core", "tokio", @@ -4581,17 +4604,18 @@ dependencies = [ [[package]] name = "opendal-service-azdls" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6dea4908d490143a9b0b7f7a790e139ff829b06a023f670455ed3d44f664b361" +checksum = "2e3c406729935fe214ce574d68681a1ff7e0b322548f14094912bdbfe50e5c53" dependencies = [ - "base64 0.22.1", + "base64 0.23.1", "bytes", "http 1.5.0", "log", + "mea", "opendal-core", "opendal-service-azure-common", - "quick-xml 0.39.4", + "quick-xml 0.41.0", "reqsign-azure-storage", "reqsign-core", "reqsign-file-read-tokio", @@ -4601,9 +4625,9 @@ dependencies = [ [[package]] name = "opendal-service-azure-common" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9b489f13c42e69d69bdd72952b634356ec43a7881a20259b38b540fcecdf4051" +checksum = "7348c88edf15af435b7be930077746b569fac5e738c1bf6a363b675e7317c9df" dependencies = [ "http 1.5.0", "opendal-core", @@ -4611,9 +4635,9 @@ dependencies = [ [[package]] name = "opendal-service-fs" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "22e89a665fef0e6bd249cf5ea47fc174b7ba892159bee4b9382528b1ca873a2c" +checksum = "826c4e17a30643b888fe983897f9a4b23b07066e1d069727a923cc8fb419a702" dependencies = [ "bytes", "log", @@ -4625,9 +4649,9 @@ dependencies = [ [[package]] name = "opendal-service-gcs" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "48de101aac565ed06af4b47903c24eafd249075553ec1fb18256751c45148d47" +checksum = "007f3fba63c21e516c956b891e96ff9892d8175662bfb781cdada9d3766a11e6" dependencies = [ "async-trait", "bytes", @@ -4635,7 +4659,7 @@ dependencies = [ "log", "opendal-core", "percent-encoding", - "quick-xml 0.39.4", + "quick-xml 0.41.0", "reqsign-core", "reqsign-file-read-tokio", "reqsign-google", @@ -4646,9 +4670,9 @@ dependencies = [ [[package]] name = "opendal-service-hdfs" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fb51ce674ce98b8b7d9ac7d2dfc9d7fc8f1bbd1da1a5fb928ecc5f13ba7dc88a" +checksum = "00570d1cc0c3f00db60bc8aaa63aa304e176285264e8674ca11897afaf509629" dependencies = [ "bytes", "futures", @@ -4661,15 +4685,15 @@ dependencies = [ [[package]] name = "opendal-service-oss" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "328fa55e8888cbdfe00826bfea2a79042422b720e8369e9e021e46121dea5ace" +checksum = "cd528ec2d49c5ca69e674ffed7b3e0686fb9cfcfea0596870de381467fda4f1b" dependencies = [ "bytes", "http 1.5.0", "log", "opendal-core", - "quick-xml 0.39.4", + "quick-xml 0.41.0", "reqsign-aliyun-oss", "reqsign-core", "reqsign-file-read-tokio", @@ -4678,18 +4702,18 @@ dependencies = [ [[package]] name = "opendal-service-s3" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "313d46c9f5ae70bca26b7c3e3fbb9b639292625f28af73aa016f47e788af9deb" +checksum = "58e80cdf192d7eff05feed747894d64f81905ac4eaf132edf7ea270abdd2d663" dependencies = [ - "base64 0.22.1", + "base64 0.23.1", "bytes", - "crc32c", + "crc-fast", "http 1.5.0", "log", "md-5 0.11.0", "opendal-core", - "quick-xml 0.39.4", + "quick-xml 0.41.0", "reqsign-aws-v4", "reqsign-core", "reqsign-file-read-tokio", diff --git a/native/Cargo.toml b/native/Cargo.toml index ab129bca0a..036efa0b16 100644 --- a/native/Cargo.toml +++ b/native/Cargo.toml @@ -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/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"] } +iceberg = { git = "https://github.com/mbutrovich/iceberg-rust", rev = "07de8fc8d650f7439b251d57a55613f7de693eb8" } +iceberg-storage-opendal = { git = "https://github.com/mbutrovich/iceberg-rust", rev = "07de8fc8d650f7439b251d57a55613f7de693eb8", features = ["opendal-memory", "opendal-fs", "opendal-s3", "opendal-gcs", "opendal-oss", "opendal-azdls"] } reqsign-core = "3" [profile.release] diff --git a/native/core/Cargo.toml b/native/core/Cargo.toml index 0a685b0f37..ff9130c8c4 100644 --- a/native/core/Cargo.toml +++ b/native/core/Cargo.toml @@ -69,9 +69,9 @@ aws-config = { workspace = true } aws-credential-types = { workspace = true } parking_lot = "0.12.5" reqwest = { version = "0.12", default-features = false, features = ["rustls-tls-native-roots", "http2"] } -object_store_opendal = { version = "0.57.0", optional = true } +object_store_opendal = { version = "0.58.0", optional = true } hdfs-sys = {version = "0.3", optional = true, features = ["hdfs_3_3"]} -opendal = { version = "0.57.0", optional = true, features = ["services-hdfs"] } +opendal = { version = "0.58.1", optional = true, features = ["services-hdfs"] } iceberg = { workspace = true } iceberg-storage-opendal = { workspace = true } reqsign-core = { workspace = true } diff --git a/native/core/src/execution/operators/copy.rs b/native/core/src/execution/operators/copy.rs index 193c385c3c..4677412622 100644 --- a/native/core/src/execution/operators/copy.rs +++ b/native/core/src/execution/operators/copy.rs @@ -37,7 +37,9 @@ pub(crate) fn copy_array(array: &dyn Array) -> ArrayRef { let mut mutable = MutableArrayData::new(vec![&data], false, capacity); - mutable.extend(0, 0, capacity); + mutable + .try_extend(0, 0, capacity) + .expect("extend failed due to offset overflow"); if matches!(array.data_type(), DataType::Dictionary(_, _)) { let copied_dict = make_array(mutable.freeze()); @@ -50,7 +52,9 @@ pub(crate) fn copy_array(array: &dyn Array) -> ArrayRef { let data = values.to_data(); let mut mutable = MutableArrayData::new(vec![&data], false, values.len()); - mutable.extend(0, 0, values.len()); + mutable + .try_extend(0, 0, values.len()) + .expect("extend failed due to offset overflow"); let copied_dict = ref_copied_dict.with_values(make_array(mutable.freeze())); Arc::new(copied_dict) diff --git a/native/core/src/execution/operators/parquet_writer.rs b/native/core/src/execution/operators/parquet_writer.rs index dbbee713ae..ca91d78744 100644 --- a/native/core/src/execution/operators/parquet_writer.rs +++ b/native/core/src/execution/operators/parquet_writer.rs @@ -92,7 +92,7 @@ enum ParquetWriter { /// The opendal::Writer is created lazily on first write #[cfg(feature = "hdfs-opendal")] Remote( - ArrowWriter>>, + Box>>>, Option, Operator, String, @@ -338,7 +338,7 @@ impl ParquetWriterExec { // HDFS writer will be created lazily on first write // Use the path from prepare_object_store_with_configs Ok(ParquetWriter::Remote( - arrow_parquet_buffer_writer, + Box::new(arrow_parquet_buffer_writer), None, op, object_store_path.to_string(), @@ -656,11 +656,9 @@ mod tests { // Create OpenDAL HDFS operator let builder = Hdfs::default().name_node(namenode); - let op = Operator::new(builder) - .map_err(|e| { - DataFusionError::Execution(format!("Failed to create HDFS operator: {}", e)) - })? - .finish(); + let op = Operator::new(builder).map_err(|e| { + DataFusionError::Execution(format!("Failed to create HDFS operator: {}", e)) + })?; let mut hdfs_writer = op.writer(output_path).await.map_err(|e| { DataFusionError::Execution(format!("Failed to create HDFS writer: {}", e)) @@ -707,11 +705,9 @@ mod tests { // Create OpenDAL HDFS operator let builder = Hdfs::default().name_node(namenode); - let op = Operator::new(builder) - .map_err(|e| { - DataFusionError::Execution(format!("Failed to create HDFS operator: {}", e)) - })? - .finish(); + let op = Operator::new(builder).map_err(|e| { + DataFusionError::Execution(format!("Failed to create HDFS operator: {}", e)) + })?; // Create a single HDFS writer for the entire file let mut hdfs_writer = op.writer(output_path).await.map_err(|e| { diff --git a/native/core/src/parquet/parquet_exec.rs b/native/core/src/parquet/parquet_exec.rs index 1308ce97fc..966e64db57 100644 --- a/native/core/src/parquet/parquet_exec.rs +++ b/native/core/src/parquet/parquet_exec.rs @@ -20,7 +20,7 @@ use crate::parquet::eager_page_index_reader_factory::EagerPageIndexReaderFactory use crate::parquet::encryption_support::{CometEncryptionConfig, ENCRYPTION_FACTORY_ID}; use crate::parquet::parquet_support::SparkParquetOptions; use crate::parquet::schema_adapter::SparkPhysicalExprAdapterFactory; -use arrow::datatypes::{Field, SchemaRef}; +use arrow::datatypes::{Field, FieldRef, SchemaRef}; use datafusion::config::{ParquetOptions, TableParquetOptions}; use datafusion::datasource::listing::PartitionedFile; use datafusion::datasource::physical_plan::{ @@ -128,13 +128,14 @@ pub(crate) fn init_datasource_exec( } _ => (Arc::clone(&required_schema), None), }; - let partition_fields: Vec<_> = partition_schema + let partition_fields: Vec = partition_schema .iter() .flat_map(|s| s.fields().iter()) - .map(|f| Arc::new(Field::new(f.name(), f.data_type().clone(), f.is_nullable())) as _) + .map(|f| Arc::new(Field::new(f.name(), f.data_type().clone(), f.is_nullable()))) .collect(); - let table_schema = - TableSchema::from_file_schema(base_schema).with_table_partition_cols(partition_fields); + let table_schema = TableSchema::builder(base_schema) + .with_table_partition_cols(partition_fields) + .build(); let mut parquet_source = ParquetSource::new(table_schema) .with_table_parquet_options(table_parquet_options) diff --git a/native/core/src/parquet/parquet_support.rs b/native/core/src/parquet/parquet_support.rs index 2ee1230ed8..56176b1637 100644 --- a/native/core/src/parquet/parquet_support.rs +++ b/native/core/src/parquet/parquet_support.rs @@ -445,12 +445,10 @@ pub(crate) fn create_hdfs_operator(url: &Url) -> Result( if pos1 <= len + 1 { // In-range insertion (including appending to end) let corrected = pos1 - 1; // 0-based insertion point - mutable_values.extend(0, start, start + corrected); - mutable_values.extend(1, row_index, row_index + 1); - mutable_values.extend(0, start + corrected, end); + mutable_values.try_extend(0, start, start + corrected)?; + mutable_values.try_extend(1, row_index, row_index + 1)?; + mutable_values.try_extend(0, start + corrected, end)?; final_len = len + 1; } else { // Beyond end: pad with nulls then insert let corrected = pos1 - 1; let padding = corrected - len; - mutable_values.extend(0, start, end); - mutable_values.extend_nulls(padding); - mutable_values.extend(1, row_index, row_index + 1); + mutable_values.try_extend(0, start, end)?; + mutable_values.try_extend_nulls(padding)?; + mutable_values.try_extend(1, row_index, row_index + 1)?; final_len = corrected + 1; // equals pos1 } } else { @@ -289,9 +289,9 @@ fn array_insert( // Legacy: -1 behaves like insert before the last element (corrected = len - k) let base_offset = if legacy_mode { 0 } else { 1 }; let corrected = len - k + base_offset; - mutable_values.extend(0, start, start + corrected); - mutable_values.extend(1, row_index, row_index + 1); - mutable_values.extend(0, start + corrected, end); + mutable_values.try_extend(0, start, start + corrected)?; + mutable_values.try_extend(1, row_index, row_index + 1)?; + mutable_values.try_extend(0, start + corrected, end)?; final_len = len + 1; } else { // Negative index beyond the start (Spark-specific behavior): @@ -300,9 +300,9 @@ fn array_insert( let base_offset = if legacy_mode { 1 } else { 0 }; let target_len = k + base_offset; let padding = target_len.saturating_sub(len + 1); - mutable_values.extend(1, row_index, row_index + 1); // insert item first - mutable_values.extend_nulls(padding); // pad nulls - mutable_values.extend(0, start, end); // append original values + mutable_values.try_extend(1, row_index, row_index + 1)?; // insert item first + mutable_values.try_extend_nulls(padding)?; // pad nulls + mutable_values.try_extend(0, start, end)?; // append original values final_len = target_len; } } diff --git a/native/spark-expr/src/array_funcs/array_slice.rs b/native/spark-expr/src/array_funcs/array_slice.rs index e9dc05330b..fcd3b467a6 100644 --- a/native/spark-expr/src/array_funcs/array_slice.rs +++ b/native/spark-expr/src/array_funcs/array_slice.rs @@ -163,7 +163,7 @@ fn slice_list( } else { let take = std::cmp::min(length_value, arr_len - zero_based_start) as usize; let begin = row_start + zero_based_start as usize; - mutable.extend(0, begin, begin + take); + mutable.try_extend(0, begin, begin + take)?; take }; diff --git a/native/spark-expr/src/array_funcs/arrays_zip.rs b/native/spark-expr/src/array_funcs/arrays_zip.rs index 6f0332d3d8..de4d74bd23 100644 --- a/native/spark-expr/src/array_funcs/arrays_zip.rs +++ b/native/spark-expr/src/array_funcs/arrays_zip.rs @@ -276,15 +276,15 @@ pub fn arrays_zip_inner(args: &[ArrayRef], names: Vec) -> Result { // Null list entry or None (Null-typed) arg — all nulls. if let Some(builder) = builders[col_idx].as_mut() { - builder.extend_nulls(max_len); + builder.try_extend_nulls(max_len)?; } } } diff --git a/native/spark-expr/src/array_funcs/list_extract.rs b/native/spark-expr/src/array_funcs/list_extract.rs index d68784ca70..0944abb987 100644 --- a/native/spark-expr/src/array_funcs/list_extract.rs +++ b/native/spark-expr/src/array_funcs/list_extract.rs @@ -287,10 +287,10 @@ fn list_extract( let len = offset_window[1].as_usize() - start; if list_array.is_null(row) { - mutable.extend_nulls(1); + mutable.try_extend_nulls(1)?; } else if let Some(index) = index { if let Some(i) = adjust_index(index, len)? { - mutable.extend(0, start + i, start + i + 1); + mutable.try_extend(0, start + i, start + i + 1)?; } else if fail_on_error { // Throw appropriate error based on whether this is element_at (one_based=true) // or GetArrayItem (one_based=false) @@ -309,11 +309,11 @@ fn list_extract( }; return Err(error_wrapper(error)); } else { - mutable.extend(1, 0, 1); + mutable.try_extend(1, 0, 1)?; } } else { // index is NULL → result is NULL - mutable.extend_nulls(1); + mutable.try_extend_nulls(1)?; } } diff --git a/native/spark-expr/src/bloom_filter/spark_bit_array.rs b/native/spark-expr/src/bloom_filter/spark_bit_array.rs index 6d43bdb942..b55567402b 100644 --- a/native/spark-expr/src/bloom_filter/spark_bit_array.rs +++ b/native/spark-expr/src/bloom_filter/spark_bit_array.rs @@ -77,8 +77,9 @@ impl SparkBitArray { pub fn merge_be_words(&mut self, incoming: &[u8]) { debug_assert_eq!(self.data.len() * 8, incoming.len()); let mut bit_count: usize = 0; - for (word, chunk) in self.data.iter_mut().zip(incoming.chunks_exact(8)) { - *word |= u64::from_be_bytes(chunk.try_into().unwrap()); + let (chunks, _) = incoming.as_chunks::<8>(); + for (word, chunk) in self.data.iter_mut().zip(chunks) { + *word |= u64::from_be_bytes(*chunk); bit_count += word.count_ones() as usize; } self.bit_count = bit_count;