diff --git a/native/Cargo.lock b/native/Cargo.lock index f5ecb09d663..d3c7fc735c9 100644 --- a/native/Cargo.lock +++ b/native/Cargo.lock @@ -188,7 +188,7 @@ dependencies = [ "digest 0.10.7", "log", "miniz_oxide", - "num-bigint", + "num-bigint 0.4.8", "quad-rand", "rand 0.9.5", "regex-lite", @@ -232,9 +232,9 @@ checksum = "d3fb67a6e08acf24fdeccbac2cb6ac4305825bd1f117462e0e6f2f193345ad56" [[package]] name = "arrow" -version = "58.4.0" +version = "59.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6cfdd0833e32a9874d2b55089333ad310c0be208aafa277385ce2461dec90be3" +checksum = "61d285d16bce7d0be61912f7928342b673067b6b7d7ef6cc179258ba7de1fecf" dependencies = [ "arrow-arith", "arrow-array", @@ -253,9 +253,9 @@ dependencies = [ [[package]] name = "arrow-arith" -version = "58.4.0" +version = "59.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0a41203398f0eaa6f7ec8e62c0da742a21abf282c148fc157f6c35c90e29981a" +checksum = "757ef1836251e88222542a7da2623bc1c9cb9e20afefa6db2c41e79991cd91d4" dependencies = [ "arrow-array", "arrow-buffer", @@ -267,9 +267,9 @@ dependencies = [ [[package]] name = "arrow-array" -version = "58.4.0" +version = "59.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ae33dad492b7df00a217563a7b0ef2874df68a0deea1b1a3acf628152f7f7a69" +checksum = "bc9a4a4b2b5ecd0e04df03471661cb61f28bed3c7fd50994715129b01b2edb97" dependencies = [ "ahash", "arrow-buffer", @@ -279,6 +279,7 @@ dependencies = [ "chrono-tz", "half", "hashbrown 0.17.1", + "libc", "num-complex", "num-integer", "num-traits", @@ -286,21 +287,21 @@ dependencies = [ [[package]] name = "arrow-buffer" -version = "58.4.0" +version = "59.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b9552f96391c005e6ab449fa941420935e7e062489b12b8b1b08879b2163f5b5" +checksum = "c12b576ef18c1deb80925a248b25ad84f419198d791b8e293fc6aaa60441fe90" dependencies = [ "bytes", "half", - "num-bigint", + "num-bigint 0.5.1", "num-traits", ] [[package]] name = "arrow-cast" -version = "58.4.0" +version = "59.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3a8a327c9649f30d8406995f27642b68df354713cca3baaaf100f076f18d5f34" +checksum = "68338a9096a5dc9bc11927c58c43a8526d96bf6abd2012ef6c0c9f505991cc79" dependencies = [ "arrow-array", "arrow-buffer", @@ -309,7 +310,7 @@ dependencies = [ "arrow-schema", "arrow-select", "atoi", - "base64 0.22.1", + "base64 0.23.0", "chrono", "comfy-table", "half", @@ -320,9 +321,9 @@ dependencies = [ [[package]] name = "arrow-csv" -version = "58.4.0" +version = "59.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "af0dd6d90d1955e9f9a014c1e563ee8aeffc21909085d25623e1da44d96eca26" +checksum = "25011b52b346407d497ef0030e12b45e4f2d0cc279efc09c4f3d09106db30e36" dependencies = [ "arrow-array", "arrow-cast", @@ -335,9 +336,9 @@ dependencies = [ [[package]] name = "arrow-data" -version = "58.4.0" +version = "59.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2b24852db04738907e06c04ea61e42fe7fda962a34513022dc0d0e754fb7976b" +checksum = "723fe4aeed7604e00b9883a465af4ff0a0e6c44c03e41a68c3d1cbc403e0e44d" dependencies = [ "arrow-buffer", "arrow-schema", @@ -348,9 +349,9 @@ dependencies = [ [[package]] name = "arrow-ipc" -version = "58.4.0" +version = "59.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "29a908a11fcfb3fb2f6730f4ac15e367bc644e419155e96238f68cf3adde572b" +checksum = "149437b14371f5b9ec60f5ddc751483ae99d7a7072653c0075e5e469156eea7b" dependencies = [ "arrow-array", "arrow-buffer", @@ -358,15 +359,15 @@ dependencies = [ "arrow-schema", "arrow-select", "flatbuffers", - "lz4_flex 0.13.1", + "lz4_flex", "zstd", ] [[package]] name = "arrow-json" -version = "58.4.0" +version = "59.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b8a96aed3931c076adee39ec2a40d8219fc7f09e79bcdaca1df16272993e1e14" +checksum = "f18b9123ccfec418a663f821c9a034af339711678c11ffe00d3ec07da5ff9f7e" dependencies = [ "arrow-array", "arrow-buffer", @@ -389,9 +390,9 @@ dependencies = [ [[package]] name = "arrow-ord" -version = "58.4.0" +version = "59.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "63a083ec750f5c043f02946b4baf05fcdbb55f4560a3277055caca5cc99f3eb0" +checksum = "e6c08dff0686cf23ca4f562803f191ccbeb726dbae6309cd4b4aaf65e0f2c979" dependencies = [ "arrow-array", "arrow-buffer", @@ -402,9 +403,9 @@ dependencies = [ [[package]] name = "arrow-row" -version = "58.4.0" +version = "59.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "514ba0ef0d4c5896202dae736251ce415abb43a950bed570fb7981b8716c0e4c" +checksum = "bbec439386df71ad570e6758a946111322b9e9dc8db83b5527321f0b4c9119c2" dependencies = [ "arrow-array", "arrow-buffer", @@ -415,9 +416,9 @@ dependencies = [ [[package]] name = "arrow-schema" -version = "58.4.0" +version = "59.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "21ca356ad6425cecb6eb7b28e4f659f1ee7880fbb1a16127de7dd62901efee9e" +checksum = "e6fed2ca0d1eade57e811cbe73b98ad50cc08a1183e13b2d2aa43a7df593f40e" dependencies = [ "bitflags 2.13.1", "serde_core", @@ -426,9 +427,9 @@ dependencies = [ [[package]] name = "arrow-select" -version = "58.4.0" +version = "59.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c58da39eb3d8350ad4a549e5c2bc49284dac554016c69829310350f1731b0aad" +checksum = "466b19cf75130b891dc1b23a84b343c714c62c64c9c62e365c76aa0ff90a53fb" dependencies = [ "ahash", "arrow-array", @@ -440,9 +441,9 @@ dependencies = [ [[package]] name = "arrow-string" -version = "58.4.0" +version = "59.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b6789b388467525e3271326b6b4915666ecfdf5142aef09779445c954b67543c" +checksum = "c838a25bb3691e919e0f617616ac51a4ff8517a952e29ca133cf0c22b2ce65b1" dependencies = [ "arrow-array", "arrow-buffer", @@ -1087,7 +1088,7 @@ checksum = "4d6867f1565b3aad85681f1015055b087fcfd840d6aeee6eee7f2da317603695" dependencies = [ "autocfg", "libm", - "num-bigint", + "num-bigint 0.4.8", "num-integer", "num-traits", "serde", @@ -1591,6 +1592,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" @@ -1838,9 +1849,8 @@ dependencies = [ [[package]] name = "datafusion" -version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "754ef4e8f073922a26f5b23133b9db4829342362b09be0bc94309cf261c2f098" +version = "55.0.0" +source = "git+https://github.com/apache/datafusion.git?tag=55.0.0-rc2#209fd9406890d97d48c0fc396ce37a5e363270b4" dependencies = [ "arrow", "arrow-schema", @@ -1873,7 +1883,7 @@ dependencies = [ "datafusion-sql", "futures", "indexmap 2.14.0", - "itertools 0.14.0", + "itertools 0.15.0", "log", "object_store", "parking_lot", @@ -1887,9 +1897,8 @@ dependencies = [ [[package]] name = "datafusion-catalog" -version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "06afd1e38dd27bbb1258685a1fc6524df6aff4e07b25b393a47de59635178d99" +version = "55.0.0" +source = "git+https://github.com/apache/datafusion.git?tag=55.0.0-rc2#209fd9406890d97d48c0fc396ce37a5e363270b4" dependencies = [ "arrow", "async-trait", @@ -1903,7 +1912,7 @@ dependencies = [ "datafusion-physical-plan", "datafusion-session", "futures", - "itertools 0.14.0", + "itertools 0.15.0", "log", "object_store", "parking_lot", @@ -1912,9 +1921,8 @@ dependencies = [ [[package]] name = "datafusion-catalog-listing" -version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f0668fb32c12065ec242be0e5b4bc62bd7a06a0be3ecd83791ef877e4be67e02" +version = "55.0.0" +source = "git+https://github.com/apache/datafusion.git?tag=55.0.0-rc2#209fd9406890d97d48c0fc396ce37a5e363270b4" dependencies = [ "arrow", "async-trait", @@ -1928,9 +1936,10 @@ dependencies = [ "datafusion-physical-expr-common", "datafusion-physical-plan", "futures", - "itertools 0.14.0", + "itertools 0.15.0", "log", "object_store", + "percent-encoding", ] [[package]] @@ -1970,7 +1979,7 @@ dependencies = [ "object_store", "object_store_opendal", "once_cell", - "opendal 0.58.1", + "opendal", "parking_lot", "parquet", "paste", @@ -2045,7 +2054,7 @@ dependencies = [ "itertools 0.15.0", "jni 0.21.1", "log", - "lz4_flex 0.14.0", + "lz4_flex", "parquet", "simd-adler32", "snap", @@ -2082,9 +2091,8 @@ dependencies = [ [[package]] name = "datafusion-common" -version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ca43b263cdff57042cfa8fb817fb3469f4878933380dccff25f5e793580abbf9" +version = "55.0.0" +source = "git+https://github.com/apache/datafusion.git?tag=55.0.0-rc2#209fd9406890d97d48c0fc396ce37a5e363270b4" dependencies = [ "arrow", "arrow-ipc", @@ -2095,9 +2103,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,9 +2117,8 @@ dependencies = [ [[package]] name = "datafusion-common-runtime" -version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "05f0ba2b864792bdca4d76c59a1de0ab6e1b61946596b9936888dbd6360035f2" +version = "55.0.0" +source = "git+https://github.com/apache/datafusion.git?tag=55.0.0-rc2#209fd9406890d97d48c0fc396ce37a5e363270b4" dependencies = [ "futures", "log", @@ -2119,9 +2127,8 @@ dependencies = [ [[package]] name = "datafusion-datasource" -version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b840a8bce0bcbf5afad02946d438591e7c373f7afccaf3d874c04485772514dd" +version = "55.0.0" +source = "git+https://github.com/apache/datafusion.git?tag=55.0.0-rc2#209fd9406890d97d48c0fc396ce37a5e363270b4" dependencies = [ "arrow", "async-compression", @@ -2141,7 +2148,7 @@ dependencies = [ "flate2", "futures", "glob", - "itertools 0.14.0", + "itertools 0.15.0", "liblzma", "log", "object_store", @@ -2155,9 +2162,8 @@ dependencies = [ [[package]] name = "datafusion-datasource-arrow" -version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a24cc0b9cf6e367f27f27406eff13abf48a11b72446aaa40b3105c0ded5c17d9" +version = "55.0.0" +source = "git+https://github.com/apache/datafusion.git?tag=55.0.0-rc2#209fd9406890d97d48c0fc396ce37a5e363270b4" dependencies = [ "arrow", "arrow-ipc", @@ -2172,16 +2178,15 @@ dependencies = [ "datafusion-physical-plan", "datafusion-session", "futures", - "itertools 0.14.0", + "itertools 0.15.0", "object_store", "tokio", ] [[package]] name = "datafusion-datasource-csv" -version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e1abe56b2a7a2d1d6de5117dd1a203181e267f28529faa5da546947621b697d7" +version = "55.0.0" +source = "git+https://github.com/apache/datafusion.git?tag=55.0.0-rc2#209fd9406890d97d48c0fc396ce37a5e363270b4" dependencies = [ "arrow", "async-trait", @@ -2202,9 +2207,8 @@ dependencies = [ [[package]] name = "datafusion-datasource-json" -version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9c3e467f0611ad7bdd5aad17c63c9bb6182d04e5282e5496d897ea2b49c024ba" +version = "55.0.0" +source = "git+https://github.com/apache/datafusion.git?tag=55.0.0-rc2#209fd9406890d97d48c0fc396ce37a5e363270b4" dependencies = [ "arrow", "async-trait", @@ -2225,11 +2229,11 @@ dependencies = [ [[package]] name = "datafusion-datasource-parquet" -version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4cc35b92cd560082155e80d9c826929c852d3c51543f4affd3a51c464a0aab3a" +version = "55.0.0" +source = "git+https://github.com/apache/datafusion.git?tag=55.0.0-rc2#209fd9406890d97d48c0fc396ce37a5e363270b4" dependencies = [ "arrow", + "arrow-schema", "async-trait", "bytes", "datafusion-common", @@ -2246,7 +2250,7 @@ dependencies = [ "datafusion-pruning", "datafusion-session", "futures", - "itertools 0.14.0", + "itertools 0.15.0", "log", "object_store", "parking_lot", @@ -2256,19 +2260,18 @@ dependencies = [ [[package]] name = "datafusion-doc" -version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d69bb69d8769e34f76839c960dbde24c1ac0c885a79b6c3c2287bdc56ec67891" +version = "55.0.0" +source = "git+https://github.com/apache/datafusion.git?tag=55.0.0-rc2#209fd9406890d97d48c0fc396ce37a5e363270b4" [[package]] name = "datafusion-execution" -version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d8eac0a09bc8d263f52025cad9e001da4d8138d633fa288edda4d06b1772eae6" +version = "55.0.0" +source = "git+https://github.com/apache/datafusion.git?tag=55.0.0-rc2#209fd9406890d97d48c0fc396ce37a5e363270b4" dependencies = [ "arrow", "arrow-buffer", "async-trait", + "bytes", "dashmap", "datafusion-common", "datafusion-expr", @@ -2278,16 +2281,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" +version = "55.0.0" +source = "git+https://github.com/apache/datafusion.git?tag=55.0.0-rc2#209fd9406890d97d48c0fc396ce37a5e363270b4" dependencies = [ "arrow", "arrow-schema", @@ -2300,32 +2305,30 @@ dependencies = [ "datafusion-functions-window-common", "datafusion-physical-expr-common", "indexmap 2.14.0", - "itertools 0.14.0", + "itertools 0.15.0", "serde_json", "sqlparser", ] [[package]] name = "datafusion-expr-common" -version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b7b19a8c95522bee8cbb313d74263b85e355d2b52f42e67ef5694bf5de9e9356" +version = "55.0.0" +source = "git+https://github.com/apache/datafusion.git?tag=55.0.0-rc2#209fd9406890d97d48c0fc396ce37a5e363270b4" 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" +version = "55.0.0" +source = "git+https://github.com/apache/datafusion.git?tag=55.0.0-rc2#209fd9406890d97d48c0fc396ce37a5e363270b4" dependencies = [ "arrow", "arrow-buffer", - "base64 0.22.1", + "base64 0.23.0", "blake2", "blake3", "chrono", @@ -2338,7 +2341,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,9 +2354,8 @@ dependencies = [ [[package]] name = "datafusion-functions-aggregate" -version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "89bc17041e424a47ed062f43df24d84aab8b57c4c3221e5c1a5eef46d6c5718b" +version = "55.0.0" +source = "git+https://github.com/apache/datafusion.git?tag=55.0.0-rc2#209fd9406890d97d48c0fc396ce37a5e363270b4" dependencies = [ "arrow", "datafusion-common", @@ -2364,17 +2366,16 @@ dependencies = [ "datafusion-macros", "datafusion-physical-expr", "datafusion-physical-expr-common", - "foldhash 0.2.0", "half", + "hashbrown 0.17.1", "log", "num-traits", ] [[package]] name = "datafusion-functions-aggregate-common" -version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "97dd2a9e865c6108059f5b37b77934f84b50bfb108f837bd0e5c9536e03f0545" +version = "55.0.0" +source = "git+https://github.com/apache/datafusion.git?tag=55.0.0-rc2#209fd9406890d97d48c0fc396ce37a5e363270b4" dependencies = [ "arrow", "datafusion-common", @@ -2384,9 +2385,8 @@ dependencies = [ [[package]] name = "datafusion-functions-nested" -version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "75f0bdfeef16d96417b9632ef855645376b242e9006a126dfd0bedfc54a93f5f" +version = "55.0.0" +source = "git+https://github.com/apache/datafusion.git?tag=55.0.0-rc2#209fd9406890d97d48c0fc396ce37a5e363270b4" dependencies = [ "arrow", "arrow-ord", @@ -2401,7 +2401,7 @@ dependencies = [ "datafusion-macros", "datafusion-physical-expr-common", "hashbrown 0.17.1", - "itertools 0.14.0", + "itertools 0.15.0", "itoa", "log", "memchr", @@ -2409,9 +2409,8 @@ dependencies = [ [[package]] name = "datafusion-functions-table" -version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f4e4941673c917819616877e9993da4503e4f4739812be0bc32c5356184c6383" +version = "55.0.0" +source = "git+https://github.com/apache/datafusion.git?tag=55.0.0-rc2#209fd9406890d97d48c0fc396ce37a5e363270b4" dependencies = [ "arrow", "async-trait", @@ -2425,9 +2424,8 @@ dependencies = [ [[package]] name = "datafusion-functions-window" -version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "12dd2e16c12b84b6f6b41b19f55b366dd1c46876bb35b86896c6349067379e8d" +version = "55.0.0" +source = "git+https://github.com/apache/datafusion.git?tag=55.0.0-rc2#209fd9406890d97d48c0fc396ce37a5e363270b4" dependencies = [ "arrow", "datafusion-common", @@ -2442,9 +2440,8 @@ dependencies = [ [[package]] name = "datafusion-functions-window-common" -version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4cdc5e4b6f8b6ef823cc1c761f85088ad4c884fe8df64df3cbcc6b2b84698441" +version = "55.0.0" +source = "git+https://github.com/apache/datafusion.git?tag=55.0.0-rc2#209fd9406890d97d48c0fc396ce37a5e363270b4" dependencies = [ "datafusion-common", "datafusion-physical-expr-common", @@ -2452,20 +2449,18 @@ dependencies = [ [[package]] name = "datafusion-macros" -version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1a3614234dd93578c92428cb4f408e020874f0d2b7e6c90c928d9d28b5df2ceb" +version = "55.0.0" +source = "git+https://github.com/apache/datafusion.git?tag=55.0.0-rc2#209fd9406890d97d48c0fc396ce37a5e363270b4" 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" +version = "55.0.0" +source = "git+https://github.com/apache/datafusion.git?tag=55.0.0-rc2#209fd9406890d97d48c0fc396ce37a5e363270b4" dependencies = [ "arrow", "chrono", @@ -2474,7 +2469,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,9 +2477,8 @@ dependencies = [ [[package]] name = "datafusion-physical-expr" -version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8cabf7a86eb70b816729e33c81bf7767c936ee1226f607a114f5dac2decac8d0" +version = "55.0.0" +source = "git+https://github.com/apache/datafusion.git?tag=55.0.0-rc2#209fd9406890d97d48c0fc396ce37a5e363270b4" dependencies = [ "arrow", "datafusion-common", @@ -2495,7 +2489,7 @@ dependencies = [ "half", "hashbrown 0.17.1", "indexmap 2.14.0", - "itertools 0.14.0", + "itertools 0.15.0", "parking_lot", "petgraph", "tokio", @@ -2503,9 +2497,8 @@ dependencies = [ [[package]] name = "datafusion-physical-expr-adapter" -version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "de222e04f7e6744501555a54ab0abe26bfdfebee380af79a9bdc175704246859" +version = "55.0.0" +source = "git+https://github.com/apache/datafusion.git?tag=55.0.0-rc2#209fd9406890d97d48c0fc396ce37a5e363270b4" dependencies = [ "arrow", "datafusion-common", @@ -2513,14 +2506,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" +version = "55.0.0" +source = "git+https://github.com/apache/datafusion.git?tag=55.0.0-rc2#209fd9406890d97d48c0fc396ce37a5e363270b4" dependencies = [ "arrow", "chrono", @@ -2528,16 +2520,15 @@ dependencies = [ "datafusion-expr-common", "hashbrown 0.17.1", "indexmap 2.14.0", - "itertools 0.14.0", + "itertools 0.15.0", "parking_lot", "pin-project", ] [[package]] name = "datafusion-physical-optimizer" -version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "86046eed10950c5f9aaed9acfd148e9bd2e1dfdfe4f9aef607d1447b271e4183" +version = "55.0.0" +source = "git+https://github.com/apache/datafusion.git?tag=55.0.0-rc2#209fd9406890d97d48c0fc396ce37a5e363270b4" dependencies = [ "arrow", "datafusion-common", @@ -2548,14 +2539,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" +version = "55.0.0" +source = "git+https://github.com/apache/datafusion.git?tag=55.0.0-rc2#209fd9406890d97d48c0fc396ce37a5e363270b4" dependencies = [ "arrow", "arrow-data", @@ -2563,6 +2554,7 @@ dependencies = [ "arrow-ord", "arrow-schema", "async-trait", + "bytes", "datafusion-common", "datafusion-common-runtime", "datafusion-execution", @@ -2576,19 +2568,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" +version = "55.0.0" +source = "git+https://github.com/apache/datafusion.git?tag=55.0.0-rc2#209fd9406890d97d48c0fc396ce37a5e363270b4" dependencies = [ "arrow", "datafusion-common", @@ -2602,10 +2594,10 @@ dependencies = [ [[package]] name = "datafusion-session" -version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5f961d209177f91bd014db5cbb2c33b7d28a2597b9003e77f17aeb712964315a" +version = "55.0.0" +source = "git+https://github.com/apache/datafusion.git?tag=55.0.0-rc2#209fd9406890d97d48c0fc396ce37a5e363270b4" dependencies = [ + "arrow-schema", "async-trait", "datafusion-common", "datafusion-execution", @@ -2616,9 +2608,8 @@ dependencies = [ [[package]] name = "datafusion-spark" -version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a1ccd16a6949503e56c084df1b90c8889db826ec9347d2f0f51a7837d6fa011e" +version = "55.0.0" +source = "git+https://github.com/apache/datafusion.git?tag=55.0.0-rc2#209fd9406890d97d48c0fc396ce37a5e363270b4" dependencies = [ "arrow", "bigdecimal", @@ -2646,9 +2637,8 @@ dependencies = [ [[package]] name = "datafusion-sql" -version = "54.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b1d71cb454da682b2af7488e1fc1ddd72ee1b28f19297b8ccad73f0a21ee9a69" +version = "55.0.0" +source = "git+https://github.com/apache/datafusion.git?tag=55.0.0-rc2#209fd9406890d97d48c0fc396ce37a5e363270b4" dependencies = [ "arrow", "bigdecimal", @@ -2893,7 +2883,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -3498,7 +3488,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/apache/iceberg-rust?rev=1b1806cf301dd85455b771d88dc163ef66b6a15b#1b1806cf301dd85455b771d88dc163ef66b6a15b" dependencies = [ "aes-gcm", "anyhow", @@ -3524,6 +3514,7 @@ dependencies = [ "fastnum", "flate2", "fnv", + "form_urlencoded", "futures", "itertools 0.13.0", "moka", @@ -3554,7 +3545,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/apache/iceberg-rust?rev=1b1806cf301dd85455b771d88dc163ef66b6a15b#1b1806cf301dd85455b771d88dc163ef66b6a15b" dependencies = [ "anyhow", "async-trait", @@ -3562,7 +3553,7 @@ dependencies = [ "cfg-if", "futures", "iceberg", - "opendal 0.57.0", + "opendal", "reqsign-aws-v4", "reqsign-core", "serde", @@ -3730,12 +3721,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" @@ -3759,7 +3744,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi", "libc", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -4178,15 +4163,6 @@ version = "0.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154" -[[package]] -name = "lz4_flex" -version = "0.13.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7ef0d4ed8669f8f8826eb00dc878084aa8f253506c4fd5e8f58f5bce72ddb97e" -dependencies = [ - "twox-hash", -] - [[package]] name = "lz4_flex" version = "0.14.0" @@ -4325,7 +4301,7 @@ version = "0.4.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "35bd024e8b2ff75562e5f34e7f4905839deb4b22955ef5e73d2fea1b9813cb23" dependencies = [ - "num-bigint", + "num-bigint 0.4.8", "num-complex", "num-integer", "num-iter", @@ -4344,6 +4320,16 @@ dependencies = [ "serde", ] +[[package]] +name = "num-bigint" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93e7820bc0a80a0238e650327316f929ba18d5be054b647490a3a6a339f3e7c0" +dependencies = [ + "num-integer", + "num-traits", +] + [[package]] name = "num-bigint-dig" version = "0.8.6" @@ -4410,7 +4396,7 @@ version = "0.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f83d14da390562dca69fc84082e73e548e1ad308d24accdedd2720017cb37824" dependencies = [ - "num-bigint", + "num-bigint 0.4.8", "num-integer", "num-traits", ] @@ -4486,7 +4472,7 @@ dependencies = [ "futures", "mea", "object_store", - "opendal 0.58.1", + "opendal", "pin-project", "tokio", ] @@ -4517,67 +4503,25 @@ 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 0.57.0", - "opendal-layer-concurrent-limit 0.57.0", - "opendal-layer-logging 0.57.0", - "opendal-layer-retry 0.57.0", - "opendal-layer-timeout 0.57.0", + "opendal-core", + "opendal-http-transport-reqwest", + "opendal-layer-concurrent-limit", + "opendal-layer-logging", + "opendal-layer-retry", + "opendal-layer-timeout", "opendal-service-azdls", "opendal-service-fs", "opendal-service-gcs", + "opendal-service-hdfs", "opendal-service-oss", "opendal-service-s3", ] -[[package]] -name = "opendal" -version = "0.58.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4f20562cc7447fcc915fc5c23df305a412ea80a733c9f2fd9e2d267e2815be6d" -dependencies = [ - "ctor", - "opendal-core 0.58.1", - "opendal-http-transport-reqwest", - "opendal-layer-concurrent-limit 0.58.1", - "opendal-layer-logging 0.58.1", - "opendal-layer-retry 0.58.1", - "opendal-layer-timeout 0.58.1", - "opendal-service-hdfs", -] - -[[package]] -name = "opendal-core" -version = "0.57.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c4f8607c90e2c963a91467f50fb49fbc7fb3d573f88cea219ca59ccd3740b309" -dependencies = [ - "anyhow", - "base64 0.22.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", - "reqsign-core", - "reqwest 0.13.4", - "serde", - "serde_json", - "tokio", - "url", - "uuid", - "web-time", -] - [[package]] name = "opendal-core" version = "0.58.1" @@ -4595,6 +4539,7 @@ dependencies = [ "mea", "percent-encoding", "quick-xml 0.41.0", + "reqsign-core", "serde", "serde_json", "tokio", @@ -4613,22 +4558,10 @@ dependencies = [ "futures", "http 1.5.0", "http-body 1.1.0", - "opendal-core 0.58.1", + "opendal-core", "reqwest 0.13.4", ] -[[package]] -name = "opendal-layer-concurrent-limit" -version = "0.57.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0d6f81ba6960e3fae1882f253b114b21d7e444e1534f209c7737a79f6243eb6f" -dependencies = [ - "futures", - "http 1.5.0", - "mea", - "opendal-core 0.57.0", -] - [[package]] name = "opendal-layer-concurrent-limit" version = "0.58.1" @@ -4638,17 +4571,7 @@ dependencies = [ "futures", "http 1.5.0", "mea", - "opendal-core 0.58.1", -] - -[[package]] -name = "opendal-layer-logging" -version = "0.57.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "58ada45c6d81d1aa4c9305d0c7d4bc317c59c85866a0908a2d75a7a978aa5ee2" -dependencies = [ - "log", - "opendal-core 0.57.0", + "opendal-core", ] [[package]] @@ -4658,18 +4581,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5c75411ab00f77851ff086b686c1e9ca8175ac18c15afa2cb75b9036436cb06c" dependencies = [ "log", - "opendal-core 0.58.1", -] - -[[package]] -name = "opendal-layer-retry" -version = "0.57.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7b2a25a718afb81fad81cb9a0580a1cb989221fa2317f888c6a37f8dad408eb7" -dependencies = [ - "backon", - "log", - "opendal-core 0.57.0", + "opendal-core", ] [[package]] @@ -4680,17 +4592,7 @@ checksum = "80b7738bd5f233ad8da39af9b9316b9b7a4eaddd91e8e32a1e19b7030688121d" dependencies = [ "backon", "log", - "opendal-core 0.58.1", -] - -[[package]] -name = "opendal-layer-timeout" -version = "0.57.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1e91f731724c213af81e9d03517859c8fc47b4578e64ad61ae4f099f10fe36e3" -dependencies = [ - "opendal-core 0.57.0", - "tokio", + "opendal-core", ] [[package]] @@ -4699,23 +4601,24 @@ version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a704141924500f3803c05ed871b53305d2a2f11cb5ef20160c3ee688a1857f66" dependencies = [ - "opendal-core 0.58.1", + "opendal-core", "tokio", ] [[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.0", "bytes", "http 1.5.0", "log", - "opendal-core 0.57.0", + "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", @@ -4725,23 +4628,23 @@ 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 0.57.0", + "opendal-core", ] [[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", - "opendal-core 0.57.0", + "opendal-core", "serde", "tokio", "xattr", @@ -4749,17 +4652,17 @@ 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", "http 1.5.0", "log", - "opendal-core 0.57.0", + "opendal-core", "percent-encoding", - "quick-xml 0.39.4", + "quick-xml 0.41.0", "reqsign-core", "reqsign-file-read-tokio", "reqsign-google", @@ -4778,22 +4681,22 @@ dependencies = [ "futures", "hdrs", "log", - "opendal-core 0.58.1", + "opendal-core", "serde", "tokio", ] [[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 0.57.0", - "quick-xml 0.39.4", + "opendal-core", + "quick-xml 0.41.0", "reqsign-aliyun-oss", "reqsign-core", "reqsign-file-read-tokio", @@ -4802,18 +4705,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.0", "bytes", - "crc32c", + "crc-fast", "http 1.5.0", "log", "md-5 0.11.0", - "opendal-core 0.57.0", - "quick-xml 0.39.4", + "opendal-core", + "quick-xml 0.41.0", "reqsign-aws-v4", "reqsign-core", "reqsign-file-read-tokio", @@ -4892,9 +4795,9 @@ dependencies = [ [[package]] name = "parquet" -version = "58.4.0" +version = "59.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d298093b2dec60289dce0684c986d0f7679e9dd15771c2c65406e1aaf604a704" +checksum = "7065842956a20c2a536924ce8e4d9955f7422451511b9eb7500d7bfe5077e59c" dependencies = [ "ahash", "arrow-array", @@ -4903,7 +4806,7 @@ dependencies = [ "arrow-ipc", "arrow-schema", "arrow-select", - "base64 0.22.1", + "base64 0.23.0", "brotli", "bytes", "chrono", @@ -4911,20 +4814,18 @@ dependencies = [ "futures", "half", "hashbrown 0.17.1", - "lz4_flex 0.13.1", - "num-bigint", + "lz4_flex", + "num-bigint 0.5.1", "num-integer", "num-traits", "object_store", "parquet-variant", "parquet-variant-compute", "parquet-variant-json", - "paste", "ring", "seq-macro", "simdutf8", "snap", - "thrift", "tokio", "twox-hash", "zstd", @@ -4932,9 +4833,9 @@ dependencies = [ [[package]] name = "parquet-variant" -version = "58.4.0" +version = "59.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3fc70e87931167a4a3fde2ee923023a0624367691e3fd503476a1084dda5a054" +checksum = "3f7e5fff3ed0c07514a7fb8bee3f2ea5a53f36939410ecac4a466620213539a8" dependencies = [ "arrow", "arrow-schema", @@ -4948,9 +4849,9 @@ dependencies = [ [[package]] name = "parquet-variant-compute" -version = "58.4.0" +version = "59.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "823a9ecee8fd83a68f7165ef13acc840c4d7ed995838ba99d4714b20a2b5e780" +checksum = "ba4d3de89dab8d1aaaf601ae8d71bd07ea88cfca9efc1df5815b982c30f631e1" dependencies = [ "arrow", "arrow-schema", @@ -4965,12 +4866,12 @@ dependencies = [ [[package]] name = "parquet-variant-json" -version = "58.4.0" +version = "59.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6f37a91177e2dddb10333546952fcf2b7674b97ebd9449432f24b883d6ea8108" +checksum = "fb19dfe1bd24c17addd761ba4f7000f615e2fa12525871c7baa835dbb3d7f147" dependencies = [ "arrow-schema", - "base64 0.22.1", + "base64 0.23.0", "chrono", "parquet-variant", "serde_json", @@ -5433,7 +5334,7 @@ dependencies = [ "once_cell", "socket2", "tracing", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -5913,7 +5814,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -5971,7 +5872,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -6546,7 +6447,7 @@ dependencies = [ "getrandom 0.4.3", "once_cell", "rustix", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -6599,17 +6500,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" @@ -7215,7 +7105,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] diff --git a/native/Cargo.toml b/native/Cargo.toml index a4ad01a3f97..b89041f1793 100644 --- a/native/Cargo.toml +++ b/native/Cargo.toml @@ -31,17 +31,17 @@ license = "Apache-2.0" edition = "2021" # Comet uses the same minimum Rust version as DataFusion -rust-version = "1.88" +rust-version = "1.94.0" [workspace.dependencies] -arrow = { version = "58.4.0", features = ["prettyprint", "ffi", "chrono-tz"] } +arrow = { version = "59.2.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.2.0", default-features = false, features = ["experimental"] } +datafusion = { git = "https://github.com/apache/datafusion.git", tag = "55.0.0-rc2", default-features = false, features = ["unicode_expressions", "crypto_expressions", "nested_expressions", "parquet"] } +datafusion-datasource = { git = "https://github.com/apache/datafusion.git", tag = "55.0.0-rc2" } +datafusion-physical-expr-adapter = { git = "https://github.com/apache/datafusion.git", tag = "55.0.0-rc2" } +datafusion-spark = { git = "https://github.com/apache/datafusion.git", tag = "55.0.0-rc2", 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/apache/iceberg-rust", rev = "1b1806cf301dd85455b771d88dc163ef66b6a15b" } +iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust", rev = "1b1806cf301dd85455b771d88dc163ef66b6a15b", 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 66e4f7683d2..fc2bd355ea5 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", tag = "55.0.0-rc2" } [features] backtrace = ["datafusion/backtrace"] diff --git a/native/core/src/debug/debug_batch_stream.rs b/native/core/src/debug/debug_batch_stream.rs index 0c5289b1984..f7ab7228212 100644 --- a/native/core/src/debug/debug_batch_stream.rs +++ b/native/core/src/debug/debug_batch_stream.rs @@ -21,6 +21,7 @@ use std::sync::Arc; use arrow::array::RecordBatch; use arrow::datatypes::{DataType, Schema}; +use datafusion::common::tree_node::TreeNodeRecursion; use datafusion::common::Result; use datafusion::execution::SendableRecordBatchStream; use datafusion::logical_expr::ColumnarValue; @@ -87,6 +88,14 @@ impl datafusion::physical_plan::ExecutionPlan for DebugExecutionDataStream { fn children(&self) -> Vec<&Arc> { vec![&self.inner] } + /// `inner` is reported as a child, so its expressions are visited by the tree walk. This + /// wrapper owns none of its own. + fn apply_expressions( + &self, + _f: &mut dyn FnMut(&Arc) -> Result, + ) -> Result { + Ok(TreeNodeRecursion::Continue) + } fn with_new_children( self: Arc, children: Vec>, diff --git a/native/core/src/execution/columnar_to_row.rs b/native/core/src/execution/columnar_to_row.rs index a1321a9defc..74defde6729 100644 --- a/native/core/src/execution/columnar_to_row.rs +++ b/native/core/src/execution/columnar_to_row.rs @@ -2491,11 +2491,14 @@ mod tests { let schema = vec![DataType::FixedSizeBinary(3)]; let mut ctx = ColumnarToRowContext::new(schema, 100); - let array: ArrayRef = Arc::new(FixedSizeBinaryArray::from(vec![ - Some(&[1u8, 2, 3][..]), - Some(&[4u8, 5, 6][..]), - None, // Test null handling - ])); + let array: ArrayRef = Arc::new( + FixedSizeBinaryArray::try_from(vec![ + Some(&[1u8, 2, 3][..]), + Some(&[4u8, 5, 6][..]), + None, // Test null handling + ]) + .unwrap(), + ); let arrays = vec![array]; let (ptr, offsets, lengths) = ctx.convert(&arrays, 3).unwrap(); diff --git a/native/core/src/execution/merge_as_partial.rs b/native/core/src/execution/merge_as_partial.rs index 5ea26115bf2..3ad3492e1ed 100644 --- a/native/core/src/execution/merge_as_partial.rs +++ b/native/core/src/execution/merge_as_partial.rs @@ -210,20 +210,30 @@ impl GroupsAccumulator for MergeAsPartialGroupsAccumulator { opt_filter: Option<&BooleanArray>, total_num_groups: usize, ) -> Result<()> { - // Redirect update to merge — this is the key trick. + // Redirect update to merge — this is the key trick. Spark's PartialMerge mode + // never applies a filter (filters apply once, at the Partial stage), so there's + // nothing to forward into merge_batch, which no longer accepts one. + debug_assert!(opt_filter.is_none()); self.inner - .merge_batch(values, group_indices, opt_filter, total_num_groups) + .merge_batch(values, group_indices, total_num_groups) } fn merge_batch( &mut self, values: &[ArrayRef], group_indices: &[usize], - opt_filter: Option<&BooleanArray>, total_num_groups: usize, ) -> Result<()> { self.inner - .merge_batch(values, group_indices, opt_filter, total_num_groups) + .merge_batch(values, group_indices, total_num_groups) + } + + fn convert_to_state( + &self, + _values: &[ArrayRef], + _opt_filter: Option<&BooleanArray>, + ) -> Result> { + datafusion::common::not_impl_err!("Input batch conversion to state not implemented") } fn evaluate(&mut self, emit_to: EmitTo) -> Result { diff --git a/native/core/src/execution/operators/copy.rs b/native/core/src/execution/operators/copy.rs index 193c385c3ce..46774126221 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/expand.rs b/native/core/src/execution/operators/expand.rs index 28d0c702254..8406529d6fa 100644 --- a/native/core/src/execution/operators/expand.rs +++ b/native/core/src/execution/operators/expand.rs @@ -17,8 +17,10 @@ use arrow::array::RecordBatch; use arrow::datatypes::{Field, Schema, SchemaRef}; +use datafusion::common::tree_node::TreeNodeRecursion; use datafusion::common::DataFusionError; use datafusion::physical_expr::{EquivalenceProperties, PhysicalExpr}; +use datafusion::physical_plan::apply_expression_roots; use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType}; use datafusion::{ execution::TaskContext, @@ -144,6 +146,13 @@ impl ExecutionPlan for ExpandExec { vec![&self.child] } + fn apply_expressions( + &self, + f: &mut dyn FnMut(&Arc) -> datafusion::common::Result, + ) -> datafusion::common::Result { + apply_expression_roots(self.projections.iter().flatten(), f) + } + fn with_new_children( self: Arc, children: Vec>, diff --git a/native/core/src/execution/operators/iceberg_scan.rs b/native/core/src/execution/operators/iceberg_scan.rs index e727294fd9e..f9d1ec908ba 100644 --- a/native/core/src/execution/operators/iceberg_scan.rs +++ b/native/core/src/execution/operators/iceberg_scan.rs @@ -25,6 +25,7 @@ use std::task::{Context, Poll}; use arrow::array::{ArrayRef, RecordBatch, RecordBatchOptions}; use arrow::datatypes::SchemaRef; +use datafusion::common::tree_node::TreeNodeRecursion; use datafusion::common::{DataFusionError, Result as DFResult}; use datafusion::execution::{RecordBatchStream, SendableRecordBatchStream, TaskContext}; use datafusion::physical_expr::expressions::Column; @@ -143,6 +144,15 @@ impl ExecutionPlan for IcebergScanExec { vec![] } + /// The projection expressions are derived per data file inside the stream, not held on the + /// node, so there is nothing to visit here. + fn apply_expressions( + &self, + _f: &mut dyn FnMut(&Arc) -> DFResult, + ) -> DFResult { + Ok(TreeNodeRecursion::Continue) + } + fn with_new_children( self: Arc, _children: Vec>, @@ -651,6 +661,8 @@ mod tests { start: 0, length: 0, record_count: None, + first_row_id: None, + data_sequence_number: None, data_file_path: "data.parquet".to_string(), data_file_format: DataFileFormat::Parquet, schema: Arc::new(Schema::builder().build().unwrap()), diff --git a/native/core/src/execution/operators/parquet_writer.rs b/native/core/src/execution/operators/parquet_writer.rs index ee0d25d6d12..315a849fc31 100644 --- a/native/core/src/execution/operators/parquet_writer.rs +++ b/native/core/src/execution/operators/parquet_writer.rs @@ -37,9 +37,10 @@ use arrow::datatypes::{Schema, SchemaRef}; use arrow::record_batch::RecordBatch; use async_trait::async_trait; use datafusion::{ + common::tree_node::TreeNodeRecursion, error::{DataFusionError, Result}, execution::context::TaskContext, - physical_expr::EquivalenceProperties, + physical_expr::{EquivalenceProperties, PhysicalExpr}, physical_plan::{ execution_plan::{Boundedness, EmissionType}, metrics::{ExecutionPlanMetricsSet, MetricsSet}, @@ -85,14 +86,14 @@ impl ParquetCompression { /// Enum representing different types of Arrow writers based on storage backend enum ParquetWriter { /// Writer for local file system - LocalFile(ArrowWriter), + LocalFile(Box>), /// Writer for HDFS or other remote storage (writes to in-memory buffer) /// Contains the arrow writer, HDFS operator, and destination path /// an Arrow writer writes to in-memory buffer the data converted to Parquet format /// The opendal::Writer is created lazily on first write #[cfg(feature = "hdfs-opendal")] Remote( - ArrowWriter>>, + Box>>>, Option, Box, String, @@ -338,7 +339,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, Box::new(op), object_store_path.to_string(), @@ -390,7 +391,7 @@ impl ParquetWriterExec { let writer = ArrowWriter::try_new(file, schema, Some(props)).map_err(|e| { DataFusionError::Execution(format!("Failed to create local file writer: {}", e)) })?; - Ok(ParquetWriter::LocalFile(writer)) + Ok(ParquetWriter::LocalFile(Box::new(writer))) } } else { // Unsupported storage scheme @@ -439,6 +440,13 @@ impl ExecutionPlan for ParquetWriterExec { vec![&self.input] } + fn apply_expressions( + &self, + _f: &mut dyn FnMut(&Arc) -> Result, + ) -> Result { + Ok(TreeNodeRecursion::Continue) + } + fn with_new_children( self: Arc, children: Vec>, diff --git a/native/core/src/execution/operators/sample.rs b/native/core/src/execution/operators/sample.rs index a26bc0d3942..a26d4a5cc30 100644 --- a/native/core/src/execution/operators/sample.rs +++ b/native/core/src/execution/operators/sample.rs @@ -18,8 +18,10 @@ use arrow::array::BooleanArray; use arrow::buffer::BooleanBuffer; use arrow::compute::filter_record_batch; +use datafusion::common::tree_node::TreeNodeRecursion; use datafusion::common::Result; use datafusion::execution::TaskContext; +use datafusion::physical_expr::PhysicalExpr; use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType}; use datafusion::physical_plan::stream::RecordBatchStreamAdapter; use datafusion::physical_plan::{ @@ -96,6 +98,13 @@ impl ExecutionPlan for SampleExec { vec![&self.input] } + fn apply_expressions( + &self, + _f: &mut dyn FnMut(&Arc) -> Result, + ) -> Result { + Ok(TreeNodeRecursion::Continue) + } + fn with_new_children( self: Arc, children: Vec>, diff --git a/native/core/src/execution/operators/scan.rs b/native/core/src/execution/operators/scan.rs index 409d0642847..f05a703dabc 100644 --- a/native/core/src/execution/operators/scan.rs +++ b/native/core/src/execution/operators/scan.rs @@ -20,6 +20,7 @@ use crate::{errors::CometError, execution::planner::TEST_EXEC_CONTEXT_ID}; use arrow::array::{ArrayRef, RecordBatch, RecordBatchOptions}; use arrow::compute::{cast_with_options, CastOptions}; use arrow::datatypes::{DataType, Field, Schema, SchemaRef}; +use datafusion::common::tree_node::TreeNodeRecursion; use datafusion::common::{arrow_datafusion_err, DataFusionError, Result as DataFusionResult}; use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType}; use datafusion::physical_plan::metrics::{ @@ -187,6 +188,13 @@ impl ExecutionPlan for ScanExec { vec![] } + fn apply_expressions( + &self, + _f: &mut dyn FnMut(&Arc) -> DataFusionResult, + ) -> DataFusionResult { + Ok(TreeNodeRecursion::Continue) + } + fn with_new_children( self: Arc, _: Vec>, diff --git a/native/core/src/execution/operators/shuffle_scan.rs b/native/core/src/execution/operators/shuffle_scan.rs index 71cfc29ae39..fd484f399dd 100644 --- a/native/core/src/execution/operators/shuffle_scan.rs +++ b/native/core/src/execution/operators/shuffle_scan.rs @@ -24,6 +24,7 @@ use crate::{ }; use arrow::array::ArrayRef; use arrow::datatypes::{DataType, Field, Schema, SchemaRef}; +use datafusion::common::tree_node::TreeNodeRecursion; use datafusion::common::Result as DataFusionResult; use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType}; use datafusion::physical_plan::metrics::{ @@ -229,6 +230,13 @@ impl ExecutionPlan for ShuffleScanExec { vec![] } + fn apply_expressions( + &self, + _f: &mut dyn FnMut(&Arc) -> DataFusionResult, + ) -> DataFusionResult { + Ok(TreeNodeRecursion::Continue) + } + fn with_new_children( self: Arc, _: Vec>, diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index c179c3b57c5..db07ee7a8ce 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -111,7 +111,7 @@ use arrow::array::{ use arrow::buffer::{BooleanBuffer, NullBuffer, OffsetBuffer}; use arrow::row::{OwnedRow, RowConverter, SortField}; use datafusion::common::utils::SingleRowListArrayBuilder; -use datafusion::common::UnnestOptions; +use datafusion::common::{NullHandling, UnnestOptions}; use datafusion::physical_plan::filter::FilterExec; use datafusion::physical_plan::joins::NestedLoopJoinExec; use datafusion::physical_plan::limit::GlobalLimitExec; @@ -2074,7 +2074,11 @@ impl PhysicalPlanner { }); let unnest_options = UnnestOptions { - preserve_nulls: explode.outer, + null_handling: if explode.outer { + NullHandling::Preserve + } else { + NullHandling::Drop + }, recursions: vec![], }; @@ -4144,6 +4148,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 278814c4bf9..22a94d04a19 100644 --- a/native/core/src/parquet/eager_page_index_reader_factory.rs +++ b/native/core/src/parquet/eager_page_index_reader_factory.rs @@ -56,10 +56,11 @@ use datafusion::execution::cache::cache_manager::FileMetadataCache; use datafusion::physical_plan::metrics::ExecutionPlanMetricsSet; use datafusion_datasource::PartitionedFile; use futures::future::BoxFuture; -use futures::FutureExt; -use object_store::ObjectStore; +use futures::{FutureExt, TryFutureExt}; +use object_store::{ObjectStore, ObjectStoreExt}; use parquet::arrow::arrow_reader::ArrowReaderOptions; -use parquet::arrow::async_reader::{AsyncFileReader, ParquetObjectReader}; +use parquet::arrow::async_reader::AsyncFileReader; +use parquet::errors::ParquetError; use parquet::file::metadata::{PageIndexPolicy, ParquetMetaData}; use std::fmt::Debug; use std::ops::Range; @@ -68,11 +69,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, @@ -93,19 +94,10 @@ impl ParquetFileReaderFactory for EagerPageIndexReaderFactory { partitioned_file.object_meta.location.as_ref(), metrics, ); - let mut inner = ParquetObjectReader::new( - Arc::clone(&self.store), - partitioned_file.object_meta.location.clone(), - ) - .with_file_size(partitioned_file.object_meta.size); - if let Some(hint) = metadata_size_hint { - inner = inner.with_footer_size_hint(hint); - } Ok(Box::new(EagerPageIndexReader { file_metrics, store: Arc::clone(&self.store), - inner, partitioned_file, metadata_cache: Arc::clone(&self.metadata_cache), metadata_size_hint, @@ -113,12 +105,14 @@ impl ParquetFileReaderFactory for EagerPageIndexReaderFactory { } } +/// Reads bytes straight off the `ObjectStore`, the same way DataFusion's own `ParquetFileReader` +/// does, and overrides only the metadata fetch. `ParquetFileReader::new` is crate-private, so the +/// byte-range plumbing is duplicated here rather than delegated. struct EagerPageIndexReader { file_metrics: ParquetFileMetrics, store: Arc, - inner: ParquetObjectReader, partitioned_file: PartitionedFile, - metadata_cache: Arc, + metadata_cache: Arc, metadata_size_hint: Option, } @@ -126,7 +120,10 @@ impl AsyncFileReader for EagerPageIndexReader { fn get_bytes(&mut self, range: Range) -> BoxFuture<'_, parquet::errors::Result> { let bytes_scanned = range.end - range.start; self.file_metrics.bytes_scanned.add(bytes_scanned as usize); - self.inner.get_bytes(range) + self.store + .get_range(&self.partitioned_file.object_meta.location, range) + .map_err(|e| ParquetError::External(Box::new(e))) + .boxed() } fn get_byte_ranges( @@ -138,7 +135,13 @@ impl AsyncFileReader for EagerPageIndexReader { { let total: u64 = ranges.iter().map(|r| r.end - r.start).sum(); self.file_metrics.bytes_scanned.add(total as usize); - self.inner.get_byte_ranges(ranges) + async move { + self.store + .get_ranges(&self.partitioned_file.object_meta.location, &ranges) + .await + .map_err(|e| ParquetError::External(Box::new(e))) + } + .boxed() } fn get_metadata<'a>( diff --git a/native/core/src/parquet/parquet_exec.rs b/native/core/src/parquet/parquet_exec.rs index 1308ce97fca..966e64db573 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/schema_adapter.rs b/native/core/src/parquet/schema_adapter.rs index c6586b4681e..ab3b9fec30d 100644 --- a/native/core/src/parquet/schema_adapter.rs +++ b/native/core/src/parquet/schema_adapter.rs @@ -18,6 +18,7 @@ use crate::parquet::cast_column::CometCastColumnExpr; use crate::parquet::parquet_support::{spark_parquet_convert, SparkParquetOptions}; use arrow::array::new_empty_array; +use arrow::compute::can_cast_types; use arrow::datatypes::{DataType, Field, FieldRef, Schema, SchemaRef}; use arrow::record_batch::RecordBatch; use datafusion::common::tree_node::{Transformed, TransformedResult, TreeNode}; @@ -76,6 +77,91 @@ fn schema_has_field_ids(schema: &SchemaRef) -> bool { schema.fields().iter().any(|f| parse_field_id(f).is_some()) } +/// Returns true when casting `physical_type` to `target_type` is a *pure* structural +/// narrowing (dropping unrequested struct/list fields, no leaf-level value reinterpretation) +/// that DataFusion's own `datafusion_common::nested_struct::cast_column` already computes +/// identically to Comet's `spark_parquet_convert`. `ColumnarValue::cast_to` (the function a +/// plain, un-swapped `CastExpr` runs at execution time; see +/// datafusion/expr-common/src/columnar_value.rs) routes to that same function whenever +/// `datafusion_common::nested_struct::requires_nested_struct_cast` holds, matching arbitrary +/// struct/list fields by name, null-filling missing target fields, and dropping extra source +/// fields, exactly the shape apache/datafusion-comet#4859 needs pruned. Confirmed +/// byte-identical to `spark_parquet_convert` for the covered shapes by +/// `test::nested_struct_narrowing_cast_matches_datafusion_generic_cast`. +/// +/// When this returns `true`, `replace_with_spark_cast` leaves DataFusion's `CastExpr` in +/// place instead of swapping in `CometCastColumnExpr`, so DataFusion's leaf-pruning +/// (`build_projection_read_plan`'s cast-clipping, apache/datafusion#24090) can see the cast +/// and read only the requested Parquet leaves, instead of falling back to a full-column read +/// because it can't recognize `CometCastColumnExpr`. +/// +/// This is deliberately an allow list, not a deny list: it only recurses through the two +/// container shapes `nested_struct::cast_column` actually implements (Struct, List / +/// LargeList), and requires every leaf it bottoms out at to be an *exact* type match. Pruning +/// (the only case this predicate needs to cover, see apache/datafusion-comet#4859) only +/// changes which struct/list fields are kept, never a leaf's type, so exact-match leaves are +/// sufficient. A deny list here (enumerate every case where Comet's nested cast differs from +/// Arrow's, allow everything else) would fail open: a future addition to +/// `parquet_convert_array` that this predicate does not know to also exclude would silently +/// start producing wrong results instead of just missing an optimization. +fn is_pure_structural_narrowing( + physical_type: &DataType, + target_type: &DataType, + parquet_options: &SparkParquetOptions, +) -> bool { + match (physical_type, target_type) { + (DataType::Struct(source_fields), DataType::Struct(target_fields)) => { + // Comet matches by Parquet field id first when the target carries one; + // DataFusion's generic cast has no field-id concept, so any field-id-bearing + // target field is a potential divergence. + if parquet_options.use_field_id + && target_fields.iter().any(|f| parse_field_id(f).is_some()) + { + return false; + } + target_fields.iter().all(|target_field| { + // Require an *exact* (case-sensitive) name match for every target field. + // `nested_struct::cast_column` always matches by exact name; Comet + // additionally matches case-insensitively when `case_sensitive` is false, + // which could resolve a field DataFusion would instead treat as missing (and + // null-fill). Requiring an exact match sidesteps that divergence regardless + // of the `case_sensitive` setting, and also sidesteps the missing-field + // nullability divergence: DataFusion errors when a non-nullable target field + // is missing from the source, whereas Comet null-fills unconditionally. + source_fields + .iter() + .find(|f| f.name() == target_field.name()) + .is_some_and(|source_field| { + is_pure_structural_narrowing( + source_field.data_type(), + target_field.data_type(), + parquet_options, + ) + }) + }) + } + (DataType::List(source_item), DataType::List(target_item)) + | (DataType::LargeList(source_item), DataType::LargeList(target_item)) => { + is_pure_structural_narrowing( + source_item.data_type(), + target_item.data_type(), + parquet_options, + ) + } + // Map is excluded structurally, not by the equality check below: `replace_with_spark_cast` + // only reaches this predicate after its own top-level `physical_type == target_type` + // check, but `is_pure_structural_narrowing` is tested (and may be called) independently + // of that guard, so an equal Map must not slip through here either. + (DataType::Map(_, _), _) | (_, DataType::Map(_, _)) => false, + // Every other shape, including Dictionary and any leaf-level type change (timestamp + // tz relabeling, `nanosAsLong`, decimal/numeric promotion, and so on), must be an + // exact match. Comet's nested cast handling for those shapes carries Spark-specific + // value or matching semantics that `nested_struct::cast_column` does not replicate, + // and none of them arise from pruning alone. + _ => physical_type == target_type, + } +} + /// Remap physical schema field names to match logical schema field names. Mirrors Spark's /// `clipParquetGroupFields`: prefer ID match for any logical field that carries a /// `PARQUET:field_id`, fall back to case-insensitive name match otherwise. @@ -878,6 +964,39 @@ impl SparkPhysicalExprAdapter { )); } + // Leave DataFusion's `CastExpr` in place for a pure structural narrowing instead + // of swapping in `CometCastColumnExpr` (see `is_pure_structural_narrowing`'s doc + // comment for why that's safe and apache/datafusion-comet#4859 for why it matters). + // + // `ColumnarValue::cast_to` does not always route a `true` result here through + // `nested_struct::cast_column`: for a List wrapping a Struct-free, leaf-unchanged + // interior (e.g. `List(Int32)` differing only in the item field's name or + // nullability, the "read simple ARRAY fields" shape), + // `nested_struct::requires_nested_struct_cast` is false and it falls to Arrow's + // plain `cast_with_options` instead. Both dispatch targets independently implement + // the same "recast values, rewrap with the target field" pattern for List, so they + // agree regardless of which one runs; confirmed by + // `test::list_of_unchanged_leaf_with_differing_item_metadata_matches_columnar_value_cast_to` + // alongside `test::nested_struct_narrowing_cast_matches_datafusion_generic_cast` + // for the Struct-pruning shape that always does route through + // `nested_struct::cast_column`. + if is_pure_structural_narrowing(physical_type, target_type, &self.parquet_options) { + // Not an assertion that `nested_struct::cast_column` specifically runs (it may + // not, see above), only that *some* DataFusion cast path can actually perform + // this pair, so a bug in the predicate surfaces here at plan time instead of + // as an opaque `ArrowError` deep inside a parquet read. + debug_assert!( + datafusion::common::nested_struct::requires_nested_struct_cast( + physical_type, + target_type + ) || can_cast_types(physical_type, target_type), + "is_pure_structural_narrowing({physical_type:?}, {target_type:?}) returned \ + true but neither nested_struct::cast_column nor Arrow's plain cast can \ + perform this pair" + ); + return Ok(Transformed::no(expr)); + } + // Same-shape complex casts, timestamp tz relabel (e.g. Timestamp(us, None) // -> Timestamp(us, Some("UTC")) for INT96 reads), and Timestamp -> Int64 // (Spark's `nanosAsLong`) need spark_parquet_convert: it handles nested @@ -1102,7 +1221,9 @@ impl PhysicalExpr for RejectOnNonEmpty { #[cfg(test)] mod test { use crate::parquet::parquet_support::SparkParquetOptions; - use crate::parquet::schema_adapter::SparkPhysicalExprAdapterFactory; + use crate::parquet::schema_adapter::{ + is_pure_structural_narrowing, SparkPhysicalExprAdapterFactory, + }; use arrow::array::UInt32Array; use arrow::array::{ BinaryArray, Date32Array, Decimal128Array, Float32Array, Float64Array, Int32Array, @@ -1117,12 +1238,13 @@ mod test { use datafusion::datasource::source::DataSourceExec; use datafusion::execution::object_store::ObjectStoreUrl; use datafusion::execution::TaskContext; + use datafusion::physical_expr::PhysicalExpr; use datafusion::physical_plan::ExecutionPlan; use datafusion_comet_spark_expr::test_common::file_util::get_temp_filename; use datafusion_comet_spark_expr::EvalMode; use datafusion_physical_expr_adapter::PhysicalExprAdapterFactory; use futures::StreamExt; - use parquet::arrow::ArrowWriter; + use parquet::arrow::{ArrowWriter, PARQUET_FIELD_ID_META_KEY}; use std::fs::File; use std::sync::Arc; @@ -1775,4 +1897,444 @@ mod test { "Expected duplicate field error, got: {err_msg}" ); } + + /// #4859 investigation: for a pure structural narrowing of a nested column (dropping + /// unrequested struct fields, no leaf-type promotion, no tz relabeling, no field-id + /// matching, case-sensitive), does Comet's `CometCastColumnExpr` (via + /// `spark_parquet_convert`) produce the same result as DataFusion's own + /// `datafusion_common::nested_struct::cast_column`, which is what a plain, un-swapped + /// `CastExpr` would run at execution time (see `ColumnarValue::cast_to` -> + /// `cast_array_by_name` in datafusion/expr-common/src/columnar_value.rs)? If so, that + /// class of cast could be left as DataFusion's own `CastExpr` (like the identity-cast + /// unwrap in #4730) so DataFusion's #24090 leaf-pruning can see it. + #[test] + fn nested_struct_narrowing_cast_matches_datafusion_generic_cast() -> Result<(), DataFusionError> + { + use crate::parquet::parquet_support::spark_parquet_convert; + use arrow::array::{ListArray, StructArray}; + use arrow::buffer::OffsetBuffer; + use arrow::compute::CastOptions; + use arrow::datatypes::Fields; + use datafusion::physical_plan::ColumnarValue; + + // events: array> -> array>, + // mirroring the physical vs. logical (Spark-pruned) schema in the #4859 repro. + let id_array: Arc = + Arc::new(Int64Array::from(vec![1i64, 2, 3, 4])); + let payload_array: Arc = + Arc::new(StringArray::from(vec!["a", "b", "c", "d"])); + let from_struct_fields: Fields = vec![ + Field::new("id", DataType::Int64, true), + Field::new("payload", DataType::Utf8, true), + ] + .into(); + let struct_array = StructArray::new( + from_struct_fields.clone(), + vec![id_array, payload_array], + None, + ); + let from_item_field = Arc::new(Field::new( + "element", + DataType::Struct(from_struct_fields), + true, + )); + let offsets = OffsetBuffer::new(vec![0, 2, 4].into()); + let list_array: Arc = Arc::new(ListArray::new( + Arc::clone(&from_item_field), + offsets, + Arc::new(struct_array), + None, + )); + + let to_struct_fields: Fields = vec![Field::new("id", DataType::Int64, true)].into(); + let to_item_field = Arc::new(Field::new( + "element", + DataType::Struct(to_struct_fields), + true, + )); + let target_type = DataType::List(Arc::clone(&to_item_field)); + + let spark_parquet_options = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false); + let comet_result = spark_parquet_convert( + ColumnarValue::Array(Arc::clone(&list_array)), + &target_type, + &spark_parquet_options, + )?; + let comet_array = match comet_result { + ColumnarValue::Array(a) => a, + _ => panic!("expected array"), + }; + + let df_array = datafusion::common::nested_struct::cast_column( + &list_array, + &target_type, + &CastOptions::default(), + )?; + + assert_eq!( + format!("{comet_array:?}"), + format!("{df_array:?}"), + "Comet's CometCastColumnExpr / spark_parquet_convert and DataFusion's own \ + nested_struct::cast_column disagree for pure struct-narrowing" + ); + Ok(()) + } + + /// #4859 follow-up: `List(Int32)` with an unchanged leaf type but differing item-field + /// metadata (Parquet's inferred "element" name, non-nullable vs. Spark's generic "item" + /// name, nullable) does NOT trigger `requires_nested_struct_cast` (the item type itself, + /// Int32, is not Struct/List), so `ColumnarValue::cast_to` (what a plain `CastExpr` runs + /// at execution time) takes the *other* branch: Arrow's plain `cast_with_options`, not + /// `nested_struct::cast_column`. This is the exact shape from the "read simple ARRAY + /// fields" CometNativeReaderSuite failure. Compares against `ColumnarValue::cast_to` + /// directly (the real call site), not `nested_struct::cast_column`, since that is what + /// actually executes here. + #[test] + fn list_of_unchanged_leaf_with_differing_item_metadata_matches_columnar_value_cast_to( + ) -> Result<(), DataFusionError> { + use crate::parquet::parquet_support::spark_parquet_convert; + use arrow::array::{Int32Array, ListArray}; + use arrow::buffer::OffsetBuffer; + use datafusion::physical_plan::ColumnarValue; + + let values: Arc = Arc::new(Int32Array::from(vec![1, 2, 3, 4])); + let physical_item_field = Arc::new(Field::new("element", DataType::Int32, false)); + let offsets = OffsetBuffer::new(vec![0, 2, 4].into()); + let list_array: Arc = Arc::new(ListArray::new( + Arc::clone(&physical_item_field), + offsets, + values, + None, + )); + + let target_item_field = Arc::new(Field::new("item", DataType::Int32, true)); + let target_type = DataType::List(Arc::clone(&target_item_field)); + + let spark_parquet_options = default_options(); + let comet_result = spark_parquet_convert( + ColumnarValue::Array(Arc::clone(&list_array)), + &target_type, + &spark_parquet_options, + )?; + let comet_array = match comet_result { + ColumnarValue::Array(a) => a, + _ => panic!("expected array"), + }; + + let df_result = + ColumnarValue::Array(Arc::clone(&list_array)).cast_to(&target_type, None)?; + let df_array = match df_result { + ColumnarValue::Array(a) => a, + _ => panic!("expected array"), + }; + + assert_eq!( + format!("{comet_array:?}"), + format!("{df_array:?}"), + "Comet's spark_parquet_convert and DataFusion's ColumnarValue::cast_to disagree \ + for a List whose leaf type is unchanged but whose item field metadata differs" + ); + Ok(()) + } + + fn struct_type(fields: Vec<(&str, DataType)>) -> DataType { + DataType::Struct( + fields + .into_iter() + .map(|(name, dt)| Field::new(name, dt, true)) + .collect(), + ) + } + + fn struct_type_with_field_id(fields: Vec<(&str, DataType, i32)>) -> DataType { + DataType::Struct( + fields + .into_iter() + .map(|(name, dt, id)| { + Field::new(name, dt, true).with_metadata(std::collections::HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + id.to_string(), + )])) + }) + .collect(), + ) + } + + fn list_type(item: DataType) -> DataType { + DataType::List(Arc::new(Field::new("element", item, true))) + } + + fn default_options() -> SparkParquetOptions { + SparkParquetOptions::new(EvalMode::Legacy, "UTC", false) + } + + /// Dropping a struct field by exact name, including through nested struct-in-struct and + /// list-of-struct, is the actual #4859 pruning shape and must be allowed. + #[test] + fn is_pure_structural_narrowing_allows_struct_and_list_field_drop() { + let opts = default_options(); + + let physical = struct_type(vec![("id", DataType::Int64), ("payload", DataType::Utf8)]); + let target = struct_type(vec![("id", DataType::Int64)]); + assert!(is_pure_structural_narrowing(&physical, &target, &opts)); + + let physical = list_type(struct_type(vec![ + ("id", DataType::Int64), + ( + "inner", + struct_type(vec![("a", DataType::Int64), ("blob", DataType::Utf8)]), + ), + ])); + let target = list_type(struct_type(vec![( + "inner", + struct_type(vec![("a", DataType::Int64)]), + )])); + assert!(is_pure_structural_narrowing(&physical, &target, &opts)); + } + + /// A target field with no exact-name match in the source (here, only a case-insensitive + /// one) must be denied: `nested_struct::cast_column` always matches by exact name and + /// would null-fill this field instead of resolving it the way Comet's case-insensitive + /// matching does. + #[test] + fn is_pure_structural_narrowing_denies_case_insensitive_only_match() { + let opts = default_options(); + let physical = struct_type(vec![("ID", DataType::Int64)]); + let target = struct_type(vec![("id", DataType::Int64)]); + assert!(!is_pure_structural_narrowing(&physical, &target, &opts)); + } + + /// A target field with no name match at all must be denied: DataFusion's + /// `nested_struct::cast_column` null-fills it unconditionally, while Comet's behavior + /// additionally depends on `return_null_struct_if_all_fields_missing` at the struct level. + #[test] + fn is_pure_structural_narrowing_denies_missing_target_field() { + let opts = default_options(); + let physical = struct_type(vec![("id", DataType::Int64)]); + let target = struct_type(vec![("id", DataType::Int64), ("payload", DataType::Utf8)]); + assert!(!is_pure_structural_narrowing(&physical, &target, &opts)); + } + + /// When `use_field_id` is set and the target struct carries Parquet field ids, Comet + /// matches by id first; DataFusion's generic cast has no field-id concept at all, so this + /// must be denied even though the names also happen to match. + #[test] + fn is_pure_structural_narrowing_denies_field_id_matching() { + let mut opts = default_options(); + opts.use_field_id = true; + let physical = struct_type_with_field_id(vec![("id", DataType::Int64, 1)]); + let target = struct_type_with_field_id(vec![("id", DataType::Int64, 1)]); + assert!(!is_pure_structural_narrowing(&physical, &target, &opts)); + } + + /// Map value narrowing has no equivalent in `nested_struct::cast_column` (it has no Map + /// arm), so it must stay on the `CometCastColumnExpr` path regardless of nesting depth. + #[test] + fn is_pure_structural_narrowing_denies_map() { + let opts = default_options(); + let physical = DataType::Map( + Arc::new(Field::new( + "entries", + struct_type(vec![("k", DataType::Int64), ("v", DataType::Utf8)]), + false, + )), + false, + ); + let target = physical.clone(); + // Even a no-op Map "narrowing" (target == physical) must not be routed through this + // predicate; Map is excluded structurally, not by an equality shortcut. + assert!(!is_pure_structural_narrowing(&physical, &target, &opts)); + } + + /// A leaf-level type change (timestamp tz relabeling, `nanosAsLong`, decimal promotion, + /// and so on) never arises from pruning alone and must be denied: Comet's nested cast + /// handling for those shapes carries Spark-specific semantics `nested_struct::cast_column` + /// does not replicate. + #[test] + fn is_pure_structural_narrowing_denies_leaf_type_change() { + let opts = default_options(); + let physical = struct_type(vec![("id", DataType::Int32)]); + let target = struct_type(vec![("id", DataType::Int64)]); + assert!(!is_pure_structural_narrowing(&physical, &target, &opts)); + } + + /// NTZ -> LTZ timestamp relabeling (INT96 reads) is a metadata-only reinterpretation + /// Comet applies via a raw `reinterpret_cast`; Arrow's generic timestamp cast performs an + /// actual timezone conversion instead, changing the value. Locked in as its own test + /// because it is the subtlest denial: both sides are "just" Timestamp, so it is exactly + /// the shape a future refactor might accidentally fold into the equality catch-all. + #[test] + fn is_pure_structural_narrowing_denies_timestamp_ntz_to_ltz_relabel() { + use arrow::datatypes::TimeUnit; + let opts = default_options(); + let physical = struct_type(vec![( + "ts", + DataType::Timestamp(TimeUnit::Microsecond, None), + )]); + let target = struct_type(vec![( + "ts", + DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())), + )]); + assert!(!is_pure_structural_narrowing(&physical, &target, &opts)); + } + + /// Dictionary-encoded columns get Comet's own dictionary-preserving or + /// dictionary-flattening handling; Arrow's generic cast does not special-case this, so a + /// Dictionary value narrowing must be denied like Map, Timestamp relabeling, and any other + /// leaf-level divergence. + #[test] + fn is_pure_structural_narrowing_denies_dictionary() { + let opts = default_options(); + let physical = struct_type(vec![( + "d", + DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8)), + )]); + let target = struct_type(vec![( + "d", + DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::LargeUtf8)), + )]); + assert!(!is_pure_structural_narrowing(&physical, &target, &opts)); + } + + /// Build a `SparkPhysicalExprAdapter` over a single "events" column and run the real + /// `PhysicalExprAdapter::rewrite` pipeline on a plain `Column` reference to it, mirroring + /// exactly what DataFusion's parquet opener does with the un-adapted projection + /// (`opener/mod.rs`'s `rewriter.rewrite(p)`). Returns the rewritten expression so callers + /// can assert on its concrete type. + fn rewrite_events_column( + physical_type: DataType, + target_type: DataType, + opts: SparkParquetOptions, + ) -> Result, DataFusionError> { + use datafusion::physical_expr::expressions::Column; + + let physical_schema = + Arc::new(Schema::new(vec![Field::new("events", physical_type, true)])); + let logical_schema = Arc::new(Schema::new(vec![Field::new("events", target_type, true)])); + + let factory = SparkPhysicalExprAdapterFactory::new(opts, None); + let adapter = factory.create(Arc::clone(&logical_schema), Arc::clone(&physical_schema))?; + + let col_expr: Arc = Arc::new(Column::new("events", 0)); + adapter.rewrite(col_expr) + } + + /// End-to-end check of the actual wiring, not just the predicate: routing a pure + /// structural narrowing through the real `PhysicalExprAdapter::rewrite` pipeline (the + /// same path DataFusion's opener calls with the plain, un-adapted projection Column) must + /// leave DataFusion's own `CastExpr` in place rather than swapping in + /// `CometCastColumnExpr`, or DataFusion's leaf-pruning (#24090) would never see it. + #[test] + fn replace_with_spark_cast_preserves_cast_expr_for_pure_narrowing( + ) -> Result<(), DataFusionError> { + use crate::parquet::cast_column::CometCastColumnExpr; + use datafusion::physical_expr::expressions::CastExpr; + + let physical_type = struct_type(vec![("id", DataType::Int64), ("payload", DataType::Utf8)]); + let target_type = struct_type(vec![("id", DataType::Int64)]); + let rewritten = rewrite_events_column(physical_type, target_type, default_options())?; + + assert!( + rewritten.downcast_ref::().is_some(), + "expected DataFusion's own CastExpr to survive a pure structural narrowing, got: \ + {rewritten}" + ); + assert!( + rewritten.downcast_ref::().is_none(), + "CometCastColumnExpr should not appear for a pure structural narrowing" + ); + Ok(()) + } + + /// Companion to the previous test: a narrowing the predicate must deny (here, a + /// case-insensitive-only field match) has to still get `CometCastColumnExpr`'s correctness + /// handling through the real pipeline, not just fall through untouched. + #[test] + fn replace_with_spark_cast_wraps_cast_expr_when_not_pure_narrowing( + ) -> Result<(), DataFusionError> { + use crate::parquet::cast_column::CometCastColumnExpr; + + let physical_type = struct_type(vec![("ID", DataType::Int64)]); + let target_type = struct_type(vec![("id", DataType::Int64)]); + let mut opts = default_options(); + opts.case_sensitive = false; + let rewritten = rewrite_events_column(physical_type, target_type, opts)?; + + assert!( + rewritten.downcast_ref::().is_some(), + "expected CometCastColumnExpr for a case-insensitive-only match, got: {rewritten}" + ); + Ok(()) + } + + /// `use_field_id` being enabled session-wide must not deny a struct whose fields simply + /// don't carry Parquet field-id metadata (e.g. a plain Parquet file read with Iceberg + /// field-id matching turned on for the table in general). The field-id bail must key off + /// whether *this* struct's fields actually carry ids, not the config flag alone. + #[test] + fn is_pure_structural_narrowing_allows_use_field_id_enabled_without_field_id_metadata() { + let mut opts = default_options(); + opts.use_field_id = true; + let physical = struct_type(vec![("id", DataType::Int64), ("payload", DataType::Utf8)]); + let target = struct_type(vec![("id", DataType::Int64)]); + assert!(is_pure_structural_narrowing(&physical, &target, &opts)); + } + + /// The field-id bail must apply at every nesting level it is reached, not just the + /// outermost struct: an inner struct carrying field ids must deny the whole cast even + /// though the outer struct matches cleanly by name. + #[test] + fn is_pure_structural_narrowing_denies_field_id_matching_at_nested_level() { + let mut opts = default_options(); + opts.use_field_id = true; + let physical = struct_type(vec![( + "outer", + struct_type_with_field_id(vec![("id", DataType::Int64, 1)]), + )]); + let target = struct_type(vec![( + "outer", + struct_type_with_field_id(vec![("id", DataType::Int64, 1)]), + )]); + assert!(!is_pure_structural_narrowing(&physical, &target, &opts)); + } + + /// A target struct with zero field-name overlap against the source is denied, distinct + /// from (and in addition to) the partial-overlap case in + /// `is_pure_structural_narrowing_denies_missing_target_field`: DataFusion's + /// `validate_struct_compatibility` hard-errors on zero overlap, while Comet null-fills + /// every field, governed by `return_null_struct_if_all_fields_missing`. + #[test] + fn is_pure_structural_narrowing_denies_zero_field_overlap() { + let opts = default_options(); + let physical = struct_type(vec![("left", DataType::Int64)]); + let target = struct_type(vec![("right", DataType::Int64)]); + assert!(!is_pure_structural_narrowing(&physical, &target, &opts)); + } + + /// `case_sensitive = true` with an exact-case match must still be allowed: the predicate's + /// exact-name requirement does not depend on the `case_sensitive` flag's value, only on + /// whether the names actually match exactly. + #[test] + fn is_pure_structural_narrowing_allows_case_sensitive_true_with_exact_match() { + let mut opts = default_options(); + opts.case_sensitive = true; + let physical = struct_type(vec![("id", DataType::Int64), ("payload", DataType::Utf8)]); + let target = struct_type(vec![("id", DataType::Int64)]); + assert!(is_pure_structural_narrowing(&physical, &target, &opts)); + } + + /// `case_sensitive = true` with a case-differing name is denied, same as the + /// `case_sensitive = false` case. This is conservative rather than strictly necessary: with + /// `case_sensitive = true`, Comet's own matching would also fail to match "ID" to "id" and + /// null-fill it, agreeing with DataFusion. Denying here only costs a missed optimization, + /// never a wrong result, and keeps the predicate's exact-match rule uniform regardless of + /// `case_sensitive`. + #[test] + fn is_pure_structural_narrowing_denies_case_mismatch_even_when_case_sensitive_true() { + let mut opts = default_options(); + opts.case_sensitive = true; + let physical = struct_type(vec![("ID", DataType::Int64)]); + let target = struct_type(vec![("id", DataType::Int64)]); + assert!(!is_pure_structural_narrowing(&physical, &target, &opts)); + } } diff --git a/native/shuffle/src/partitioners/traits.rs b/native/shuffle/src/partitioners/traits.rs index 9572b70db5b..5bd41a62a50 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/schema_align.rs b/native/shuffle/src/schema_align.rs index 33ce490b69a..e07e92e5820 100644 --- a/native/shuffle/src/schema_align.rs +++ b/native/shuffle/src/schema_align.rs @@ -36,8 +36,9 @@ use arrow::array::RecordBatch; use arrow::datatypes::{Field, Schema, SchemaRef}; +use datafusion::common::tree_node::TreeNodeRecursion; use datafusion::common::DataFusionError; -use datafusion::physical_expr::EquivalenceProperties; +use datafusion::physical_expr::{EquivalenceProperties, PhysicalExpr}; use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType}; use datafusion::{ execution::TaskContext, @@ -175,6 +176,15 @@ impl ExecutionPlan for SchemaAlignExec { vec![&self.child] } + /// The per-column casts are described by `column_actions`, not by `PhysicalExpr`s, so there is + /// nothing to visit here. + fn apply_expressions( + &self, + _f: &mut dyn FnMut(&Arc) -> datafusion::common::Result, + ) -> datafusion::common::Result { + Ok(TreeNodeRecursion::Continue) + } + fn with_new_children( self: Arc, children: Vec>, diff --git a/native/shuffle/src/shuffle_writer.rs b/native/shuffle/src/shuffle_writer.rs index 630e0e33431..6e87e4a281d 100644 --- a/native/shuffle/src/shuffle_writer.rs +++ b/native/shuffle/src/shuffle_writer.rs @@ -26,9 +26,10 @@ use crate::writers::LocalPartitionWriter; use crate::{CometPartitioning, CompressionCodec, ShuffleBlockWriter}; use async_trait::async_trait; use datafusion::common::exec_datafusion_err; -use datafusion::physical_expr::{EquivalenceProperties, Partitioning}; +use datafusion::common::tree_node::TreeNodeRecursion; +use datafusion::physical_expr::{EquivalenceProperties, Partitioning, PhysicalExpr}; use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType}; -use datafusion::physical_plan::EmptyRecordBatchStream; +use datafusion::physical_plan::{apply_expression_roots, EmptyRecordBatchStream}; use datafusion::{ arrow::datatypes::SchemaRef, error::Result, @@ -144,6 +145,21 @@ impl ExecutionPlan for ShuffleWriterExec { vec![&self.input] } + fn apply_expressions( + &self, + f: &mut dyn FnMut(&Arc) -> Result, + ) -> Result { + match &self.partitioning { + CometPartitioning::Hash(exprs, _) => apply_expression_roots(exprs, f), + CometPartitioning::RangePartitioning(ordering, _, _, _) => { + apply_expression_roots(ordering.iter().map(|sort_expr| &sort_expr.expr), f) + } + CometPartitioning::SinglePartition | CometPartitioning::RoundRobin(_, _) => { + Ok(TreeNodeRecursion::Continue) + } + } + } + fn with_new_children( self: Arc, children: Vec>, diff --git a/native/shuffle/src/writers/local/spill.rs b/native/shuffle/src/writers/local/spill.rs index 450b617421f..7304f5d19d9 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 25b0e598df8..9b415f897da 100644 --- a/native/shuffle/src/writers/partition_writer.rs +++ b/native/shuffle/src/writers/partition_writer.rs @@ -32,7 +32,7 @@ use arrow::record_batch::RecordBatch; /// ascending id order, then a single [`finish_all`](PartitionWriter::finish_all). /// /// [`LocalPartitionWriter`]: crate::writers::local::local_partition_writer::LocalPartitionWriter -pub(crate) trait PartitionWriter: Send + Sync { +pub(crate) trait PartitionWriter: Send { /// Stages the batches from `iter` for partition `pid` without finalizing it. /// /// Used to stream single-partition output and to stage multi-partition diff --git a/native/shuffle/src/writers/shuffle_block_writer.rs b/native/shuffle/src/writers/shuffle_block_writer.rs index 7b6846b3ba7..04da9ed96ca 100644 --- a/native/shuffle/src/writers/shuffle_block_writer.rs +++ b/native/shuffle/src/writers/shuffle_block_writer.rs @@ -18,7 +18,7 @@ use arrow::array::RecordBatch; use arrow::datatypes::{DataType, Schema, SchemaRef}; use arrow::ipc::writer::{ - write_message, CompressionContext, DictionaryTracker, IpcDataGenerator, IpcWriteOptions, + write_message, DictionaryTracker, IpcDataGenerator, IpcWriteContext, IpcWriteOptions, StreamWriter, }; use arrow::ipc::MetadataVersion; @@ -159,7 +159,7 @@ impl ShuffleBlockWriter { // Fast path: reuse the pre-encoded schema message and write the record batch manually. let data_gen = IpcDataGenerator::default(); let mut dictionary_tracker = DictionaryTracker::new(true); - let mut compression_context = CompressionContext::default(); + let mut compression_context = IpcWriteContext::default(); let (encoded_dictionaries, encoded_batch) = data_gen.encode( batch, &mut dictionary_tracker, diff --git a/native/spark-expr/src/agg_funcs/avg.rs b/native/spark-expr/src/agg_funcs/avg.rs index 24a9a309911..bb5371568b5 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 186e0272445..0e02dd03d41 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 189b76c8636..d69ac61def7 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 e5b77ad33bc..548c1187208 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 01fa13800a5..3cb39f0acd7 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 3fddfdcb684..78e585a3965 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 3389a86af68..bbceaa72dcd 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 4ddf13c5d69..cae0a70b422 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 59a4371772e..56a8c122c41 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 f1841729cf0..57a8f6da501 100644 --- a/native/spark-expr/src/agg_funcs/variance.rs +++ b/native/spark-expr/src/agg_funcs/variance.rs @@ -18,7 +18,7 @@ use arrow::array::{Array, ArrayRef, AsArray, BooleanArray, Float64Array}; use arrow::buffer::NullBuffer; use arrow::datatypes::{DataType, Field, FieldRef, Float64Type}; -use datafusion::common::{downcast_value, Result, ScalarValue}; +use datafusion::common::{downcast_value, not_impl_err, Result, ScalarValue}; use datafusion::logical_expr::function::{AccumulatorArgs, StateFieldsArgs}; use datafusion::logical_expr::Volatility::Immutable; use datafusion::logical_expr::{ @@ -340,7 +340,6 @@ impl GroupsAccumulator for VarianceGroupsAccumulator { &mut self, values: &[ArrayRef], group_indices: &[usize], - _opt_filter: Option<&BooleanArray>, total_num_groups: usize, ) -> Result<()> { assert_eq!(values.len(), 3, "three arguments to merge_batch"); @@ -387,6 +386,14 @@ impl GroupsAccumulator for VarianceGroupsAccumulator { ]) } + fn convert_to_state( + &self, + _values: &[ArrayRef], + _opt_filter: Option<&BooleanArray>, + ) -> Result> { + not_impl_err!("Input batch conversion to state not implemented") + } + fn size(&self) -> usize { self.counts.capacity() * size_of::() + self.means.capacity() * size_of::() @@ -512,8 +519,8 @@ mod groups_tests { let right_state = right.state(EmitTo::All).unwrap(); let mut merged = pop_acc(); - merged.merge_batch(&left_state, &[0], None, 1).unwrap(); - merged.merge_batch(&right_state, &[0], None, 1).unwrap(); + merged.merge_batch(&left_state, &[0], 1).unwrap(); + merged.merge_batch(&right_state, &[0], 1).unwrap(); let merged_result = evaluate(&mut merged)[0].unwrap(); assert!((single_result - merged_result).abs() < 1e-12); diff --git a/native/spark-expr/src/array_funcs/array_insert.rs b/native/spark-expr/src/array_funcs/array_insert.rs index e056c108e05..16954f12437 100644 --- a/native/spark-expr/src/array_funcs/array_insert.rs +++ b/native/spark-expr/src/array_funcs/array_insert.rs @@ -266,17 +266,17 @@ fn array_insert( if pos1 <= len + 1 { // In-range insertion (including appending to end) let corrected = pos1 - 1; // 0-based insertion point - mutable_values.extend(0, start, start + corrected); - mutable_values.extend(1, row_index, row_index + 1); - mutable_values.extend(0, start + corrected, end); + mutable_values.try_extend(0, start, start + corrected)?; + mutable_values.try_extend(1, row_index, row_index + 1)?; + mutable_values.try_extend(0, start + corrected, end)?; final_len = len + 1; } else { // Beyond end: pad with nulls then insert let corrected = pos1 - 1; let padding = corrected - len; - mutable_values.extend(0, start, end); - mutable_values.extend_nulls(padding); - mutable_values.extend(1, row_index, row_index + 1); + mutable_values.try_extend(0, start, end)?; + mutable_values.try_extend_nulls(padding)?; + mutable_values.try_extend(1, row_index, row_index + 1)?; final_len = corrected + 1; // equals pos1 } } else { @@ -289,9 +289,9 @@ fn array_insert( // Legacy: -1 behaves like insert before the last element (corrected = len - k) let base_offset = if legacy_mode { 0 } else { 1 }; let corrected = len - k + base_offset; - mutable_values.extend(0, start, start + corrected); - mutable_values.extend(1, row_index, row_index + 1); - mutable_values.extend(0, start + corrected, end); + mutable_values.try_extend(0, start, start + corrected)?; + mutable_values.try_extend(1, row_index, row_index + 1)?; + mutable_values.try_extend(0, start + corrected, end)?; final_len = len + 1; } else { // Negative index beyond the start (Spark-specific behavior): @@ -300,9 +300,9 @@ fn array_insert( let base_offset = if legacy_mode { 1 } else { 0 }; let target_len = k + base_offset; let padding = target_len.saturating_sub(len + 1); - mutable_values.extend(1, row_index, row_index + 1); // insert item first - mutable_values.extend_nulls(padding); // pad nulls - mutable_values.extend(0, start, end); // append original values + mutable_values.try_extend(1, row_index, row_index + 1)?; // insert item first + mutable_values.try_extend_nulls(padding)?; // pad nulls + mutable_values.try_extend(0, start, end)?; // append original values final_len = target_len; } } diff --git a/native/spark-expr/src/array_funcs/array_slice.rs b/native/spark-expr/src/array_funcs/array_slice.rs index e9dc05330be..fcd3b467a6c 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 6f0332d3d86..de4d74bd23d 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 d68784ca70a..0944abb987f 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 6d43bdb942f..b55567402be 100644 --- a/native/spark-expr/src/bloom_filter/spark_bit_array.rs +++ b/native/spark-expr/src/bloom_filter/spark_bit_array.rs @@ -77,8 +77,9 @@ impl SparkBitArray { pub fn merge_be_words(&mut self, incoming: &[u8]) { debug_assert_eq!(self.data.len() * 8, incoming.len()); let mut bit_count: usize = 0; - for (word, chunk) in self.data.iter_mut().zip(incoming.chunks_exact(8)) { - *word |= u64::from_be_bytes(chunk.try_into().unwrap()); + let (chunks, _) = incoming.as_chunks::<8>(); + for (word, chunk) in self.data.iter_mut().zip(chunks) { + *word |= u64::from_be_bytes(*chunk); bit_count += word.count_ones() as usize; } self.bit_count = bit_count; diff --git a/native/spark-expr/src/comet_scalar_funcs.rs b/native/spark-expr/src/comet_scalar_funcs.rs index e90e01de953..6f1d663f099 100644 --- a/native/spark-expr/src/comet_scalar_funcs.rs +++ b/native/spark-expr/src/comet_scalar_funcs.rs @@ -23,6 +23,7 @@ use crate::math_funcs::checked_arithmetic::{checked_add, checked_div, checked_mu use crate::math_funcs::log::spark_log; use crate::math_funcs::modulo_expr::spark_modulo; use crate::math_funcs::pow::spark_pow; +use crate::math_funcs::sqrt::spark_sqrt; use crate::{ spark_ceil, spark_day_name, spark_decimal_div, spark_decimal_integral_div, spark_floor, spark_isnan, spark_lpad, spark_make_decimal, spark_month_name, spark_read_side_padding, @@ -213,6 +214,10 @@ pub fn create_comet_physical_fun_with_eval_mode( let func = Arc::new(spark_pow); make_comet_scalar_udf!("pow", func, without data_type) } + "spark_sqrt" => { + let func = Arc::new(spark_sqrt); + make_comet_scalar_udf!("spark_sqrt", func, without data_type) + } "base64" => { let func = Arc::new(crate::string_funcs::spark_base64); make_comet_scalar_udf!("base64", func, without data_type) diff --git a/native/spark-expr/src/lib.rs b/native/spark-expr/src/lib.rs index 2b5c29befc6..29b20e2878e 100644 --- a/native/spark-expr/src/lib.rs +++ b/native/spark-expr/src/lib.rs @@ -87,9 +87,9 @@ pub use json_funcs::{FromJson, ToJson}; pub use math_funcs::{ checked_add, checked_div, checked_mul, checked_sub, create_modulo_expr, create_negate_expr, spark_ceil, spark_decimal_div, spark_decimal_integral_div, spark_floor, spark_log, - spark_make_decimal, spark_pow, spark_round, spark_unhex, spark_unscaled_value, CheckOverflow, - DecimalRescaleCheckOverflow, NegativeExpr, NormalizeNaNAndZero, WideDecimalBinaryExpr, - WideDecimalOp, + spark_make_decimal, spark_pow, spark_round, spark_sqrt, spark_unhex, spark_unscaled_value, + CheckOverflow, DecimalRescaleCheckOverflow, NegativeExpr, NormalizeNaNAndZero, + WideDecimalBinaryExpr, WideDecimalOp, }; pub use query_context::{create_query_context_map, QueryContext, QueryContextMap}; pub use string_funcs::*; diff --git a/native/spark-expr/src/math_funcs/mod.rs b/native/spark-expr/src/math_funcs/mod.rs index 56809a01fbc..7a7c6a97539 100644 --- a/native/spark-expr/src/math_funcs/mod.rs +++ b/native/spark-expr/src/math_funcs/mod.rs @@ -26,6 +26,7 @@ pub mod modulo_expr; mod negative; pub(crate) mod pow; mod round; +pub(crate) mod sqrt; pub(crate) mod unhex; mod utils; mod wide_decimal_binary_expr; @@ -41,5 +42,6 @@ pub use modulo_expr::create_modulo_expr; pub use negative::{create_negate_expr, NegativeExpr}; pub use pow::spark_pow; pub use round::spark_round; +pub use sqrt::spark_sqrt; pub use unhex::spark_unhex; pub use wide_decimal_binary_expr::{WideDecimalBinaryExpr, WideDecimalOp}; diff --git a/native/spark-expr/src/math_funcs/sqrt.rs b/native/spark-expr/src/math_funcs/sqrt.rs new file mode 100644 index 00000000000..1397386eb9c --- /dev/null +++ b/native/spark-expr/src/math_funcs/sqrt.rs @@ -0,0 +1,90 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use arrow::array::{ArrayRef, AsArray, Float64Array}; +use arrow::datatypes::Float64Type; +use datafusion::common::{DataFusionError, ScalarValue}; +use datafusion::physical_plan::ColumnarValue; +use std::sync::Arc; + +/// Spark-compatible `sqrt`, matching `java.lang.Math.sqrt`: a negative input produces +/// `NaN` rather than an error, unlike DataFusion's own `sqrt`. +pub fn spark_sqrt(args: &[ColumnarValue]) -> Result { + if args.len() != 1 { + return Err(DataFusionError::Internal(format!( + "spark_sqrt requires 1 argument, got {}", + args.len() + ))); + } + + match &args[0] { + ColumnarValue::Array(array) => { + let values = array.as_primitive_opt::().ok_or_else(|| { + DataFusionError::Internal(format!( + "spark_sqrt expected Float64, got {:?}", + array.data_type() + )) + })?; + let result: Float64Array = values.unary(|v| v.sqrt()); + Ok(ColumnarValue::Array(Arc::new(result) as ArrayRef)) + } + ColumnarValue::Scalar(ScalarValue::Float64(v)) => Ok(ColumnarValue::Scalar( + ScalarValue::Float64(v.map(f64::sqrt)), + )), + ColumnarValue::Scalar(other) => Err(DataFusionError::Internal(format!( + "spark_sqrt expected Float64 scalar, got {other:?}", + ))), + } +} + +#[cfg(test)] +mod test { + use super::*; + use arrow::array::{Array, Float64Array}; + + #[test] + fn test_spark_sqrt_negative_is_nan() { + let input = Float64Array::from(vec![Some(4.0), Some(-1.0), Some(0.0), None]); + let result = spark_sqrt(&[ColumnarValue::Array(Arc::new(input))]).unwrap(); + let ColumnarValue::Array(result) = result else { + unreachable!() + }; + let result = result.as_primitive::(); + assert_eq!(result.value(0), 2.0); + assert!(result.value(1).is_nan()); + assert_eq!(result.value(2), 0.0); + assert!(result.is_null(3)); + } + + #[test] + fn test_spark_sqrt_scalar_negative_is_nan() { + let result = + spark_sqrt(&[ColumnarValue::Scalar(ScalarValue::Float64(Some(-1.0)))]).unwrap(); + let ColumnarValue::Scalar(ScalarValue::Float64(Some(result))) = result else { + unreachable!() + }; + assert!(result.is_nan()); + } + + #[test] + fn test_spark_sqrt_scalar_null() { + let result = spark_sqrt(&[ColumnarValue::Scalar(ScalarValue::Float64(None))]).unwrap(); + let ColumnarValue::Scalar(ScalarValue::Float64(None)) = result else { + unreachable!() + }; + } +} diff --git a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala index 85601a9e0c9..11f15f84786 100644 --- a/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala +++ b/spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala @@ -153,7 +153,7 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim { classOf[Signum] -> CometScalarFunction("signum"), classOf[Sin] -> CometScalarFunction("sin"), classOf[Sinh] -> CometScalarFunction("sinh"), - classOf[Sqrt] -> CometScalarFunction("sqrt"), + classOf[Sqrt] -> CometSqrt, classOf[Subtract] -> CometSubtract, classOf[Tan] -> CometScalarFunction("tan"), classOf[Tanh] -> CometScalarFunction("tanh"), diff --git a/spark/src/main/scala/org/apache/comet/serde/math.scala b/spark/src/main/scala/org/apache/comet/serde/math.scala index a0ea76cc061..65e2bf673e9 100644 --- a/spark/src/main/scala/org/apache/comet/serde/math.scala +++ b/spark/src/main/scala/org/apache/comet/serde/math.scala @@ -19,7 +19,7 @@ package org.apache.comet.serde -import org.apache.spark.sql.catalyst.expressions.{Abs, Add, Atan2, Attribute, BRound, Ceil, CheckOverflow, Conv, Expression, Floor, Hex, Hypot, If, LessThanOrEqual, Literal, Log, Log10, Log1p, Log2, Logarithm, NaNvl, Pmod, Pow, UnaryPositive, Unhex, WidthBucket} +import org.apache.spark.sql.catalyst.expressions.{Abs, Add, Atan2, Attribute, BRound, Ceil, CheckOverflow, Conv, Expression, Floor, Hex, Hypot, If, LessThanOrEqual, Literal, Log, Log10, Log1p, Log2, Logarithm, NaNvl, Pmod, Pow, Sqrt, UnaryPositive, Unhex, WidthBucket} import org.apache.spark.sql.types.{DecimalType, DoubleType, NumericType} import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, scalarFunctionExprToProto, scalarFunctionExprToProtoWithReturnType, serializeDataType} @@ -216,6 +216,23 @@ object CometPow extends CometExpressionSerde[Pow] { } } +// Uses a custom spark_sqrt UDF because DataFusion's own `sqrt` errors on negative +// input, while Spark's Sqrt (a plain wrapper around java.lang.Math.sqrt) returns NaN. +// spark_sqrt is a Comet-only name with no DataFusion builtin counterpart, so the +// return type must be set explicitly here to skip the session registry lookup that +// scalarFunctionExprToProto would otherwise trigger (see CometLogarithm/spark_log). +object CometSqrt extends CometExpressionSerde[Sqrt] { + override def convert( + expr: Sqrt, + inputs: Seq[Attribute], + binding: Boolean): Option[ExprOuterClass.Expr] = { + val childExpr = exprToProtoInternal(expr.child, inputs, binding) + val optExpr = + scalarFunctionExprToProtoWithReturnType("spark_sqrt", DoubleType, false, childExpr) + optExpr + } +} + sealed trait MathExprBase { protected def nullIfNegative(expression: Expression): Expression = { val zero = Literal.default(expression.dataType) diff --git a/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala index fd7d4ee3f46..4feda01d898 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometNativeReaderSuite.scala @@ -29,12 +29,12 @@ import org.apache.parquet.hadoop.api.WriteSupport import org.apache.parquet.hadoop.api.WriteSupport.WriteContext import org.apache.parquet.io.api.RecordConsumer import org.apache.parquet.schema.MessageTypeParser -import org.apache.spark.sql.{CometTestBase, Row} +import org.apache.spark.sql.{CometTestBase, DataFrame, Row} import org.apache.spark.sql.comet.CometNativeScanExec import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper import org.apache.spark.sql.functions.{array, col} import org.apache.spark.sql.internal.SQLConf -import org.apache.spark.sql.types.{IntegerType, LongType, NullType, StringType, StructType} +import org.apache.spark.sql.types._ import org.apache.comet.CometConf import org.apache.comet.CometSparkSessionExtensions.isSpark41Plus @@ -794,6 +794,7 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper class Builder extends ParquetWriter.Builder[RecordConsumer => Unit, Builder](new Path(path)) { override def getWriteSupport(conf: Configuration): WriteSupport[RecordConsumer => Unit] = writeSupport + override def self(): Builder = this } val writer = new Builder().build() @@ -997,6 +998,82 @@ class CometNativeReaderSuite extends CometTestBase with AdaptiveSparkPlanHelper } } } + + /** Runs `df` and returns the summed `bytes_scanned` metric across its native scans. */ + private def bytesScanned(df: DataFrame): Long = { + val (_, cometPlan) = checkSparkAnswerAndOperator(df) + val scans = cometPlan.collect { case n: CometNativeScanExec => n } + assert(scans.nonEmpty, "Expected a CometNativeScanExec") + scans.map(_.metrics("bytes_scanned").value).sum + } + + test("issue #4859: native scan honors pruned nested ReadSchema for array") { + // Regression repro: the native scan pushes only a top-level projection vector into + // DataFusion, so a plain Column("events") ref yields ProjectionMask::roots and the + // whole nested column (all leaves) is read even though Spark's pruned ReadSchema asks + // for a single leaf. Selecting only the small `id` leaf must scan far fewer bytes than + // also selecting the large `payload` leaf; before the fix they are ~equal. + withTempPath { path => + withSQLConf( + SQLConf.LEAF_NODE_DEFAULT_PARALLELISM.key -> "1", + SQLConf.NESTED_SCHEMA_PRUNING_ENABLED.key -> "true") { + // events: array>. `payload` is a distinct, + // high-entropy uuid per row so it dominates the on-disk column size and resists + // compression; `id` is a sequential long that compresses to almost nothing. + spark + .range(0, 20000) + .selectExpr("array(named_struct('id', id, 'payload', uuid())) as events") + .repartition(1) + .write + .mode("overwrite") + .parquet(path.toString) + + val idOnly = + bytesScanned(spark.read.parquet(path.toString).selectExpr("events.id")) + val idAndPayload = + bytesScanned( + spark.read.parquet(path.toString).selectExpr("events.id", "events.payload")) + + assert( + idOnly < idAndPayload / 2, + "native scan read the whole nested column despite pruned ReadSchema " + + s"(bytes_scanned: events.id=$idOnly, events.id+payload=$idAndPayload)") + } + } + } + + test("issue #4859: leaf pruning through array> drops siblings at each level") { + withTempPath { path => + withSQLConf( + SQLConf.LEAF_NODE_DEFAULT_PARALLELISM.key -> "1", + SQLConf.NESTED_SCHEMA_PRUNING_ENABLED.key -> "true") { + // events: array>>. `blob` is the dominant + // high-entropy leaf. Selecting events.inner.a must prune both the top-level + // sibling `id` and the inner sibling `blob`. + spark + .range(0, 20000) + .selectExpr( + "array(named_struct('id', id, 'inner', named_struct('a', id, 'blob', uuid()))) as events") + .repartition(1) + .write + .mode("overwrite") + .parquet(path.toString) + + val aOnly = + bytesScanned(spark.read.parquet(path.toString).selectExpr("events.inner.a")) + val aAndBlob = + bytesScanned( + spark.read + .parquet(path.toString) + .selectExpr("events.inner.a", "events.inner.blob")) + + assert( + aOnly < aAndBlob / 2, + "native scan read unrequested nested leaves through array> " + + s"(bytes_scanned: events.inner.a=$aOnly, events.inner.a+blob=$aAndBlob)") + } + } + } } private class DirectWriteSupport(schema: org.apache.parquet.schema.MessageType) diff --git a/spark/src/test/scala/org/apache/spark/sql/CometTPCHQuerySuite.scala b/spark/src/test/scala/org/apache/spark/sql/CometTPCHQuerySuite.scala index 0f942927091..b92ab0d7d94 100644 --- a/spark/src/test/scala/org/apache/spark/sql/CometTPCHQuerySuite.scala +++ b/spark/src/test/scala/org/apache/spark/sql/CometTPCHQuerySuite.scala @@ -91,7 +91,7 @@ class CometTPCHQuerySuite extends QueryTest with TPCBase with ShimCometTPCHQuery conf.set(CometConf.COMET_NATIVE_SCAN_ENABLED.key, "true") conf.set(CometConf.COMET_SHUFFLE_ENABLED.key, "true") conf.set(MEMORY_OFFHEAP_ENABLED.key, "true") - conf.set(MEMORY_OFFHEAP_SIZE.key, "2g") + conf.set(MEMORY_OFFHEAP_SIZE.key, "4g") } protected override def createSparkSession: TestSparkSession = {