From 05879d1f5a0a44afd39ecb972fd3bfae16c8849f Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Tue, 6 Aug 2024 16:56:18 -0600 Subject: [PATCH 1/5] upgrade df version and disable skip partial agg --- native/Cargo.lock | 144 ++++++++++-------- native/Cargo.toml | 14 +- .../core/src/execution/datafusion/planner.rs | 23 +-- 3 files changed, 99 insertions(+), 82 deletions(-) diff --git a/native/Cargo.lock b/native/Cargo.lock index ef3d3cf2e9c..581beef0608 100644 --- a/native/Cargo.lock +++ b/native/Cargo.lock @@ -84,9 +84,9 @@ checksum = "4b46cbb362ab8752921c97e041f5e366ee6297bd428a31275b9fcf1e380f7299" [[package]] name = "anstyle" -version = "1.0.7" +version = "1.0.8" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "038dfcf04a5feb68e9c60b21c9625a54c2c0616e79b72b0fd87075a056ae1d1b" +checksum = "1bec1de6f59aedf83baf9ff929c98f2ad654b97c9510f4e70cf6f661d49fd5b1" [[package]] name = "anyhow" @@ -457,9 +457,9 @@ checksum = "79296716171880943b8470b5f8d03aa55eb2e645a4874bdbb28adb49162e012c" [[package]] name = "bytemuck" -version = "1.16.1" +version = "1.16.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b236fc92302c97ed75b38da1f4917b5cdda4984745740f153a5d3059e48d725e" +checksum = "102087e286b4677862ea56cf8fc58bb2cdfa8725c40ffb80fe3a008eb7f2fc83" [[package]] name = "byteorder" @@ -469,9 +469,9 @@ checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b" [[package]] name = "bytes" -version = "1.6.1" +version = "1.7.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a12916984aab3fa6e39d655a33e09c0071eb36d6ab3aea5c2d78551f1df6d952" +checksum = "8318a53db07bb3f8dca91a600466bdb3f2eaadeedfdbcf02e1accbad9271ba50" [[package]] name = "cast" @@ -481,9 +481,9 @@ checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5" [[package]] name = "cc" -version = "1.1.6" +version = "1.1.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2aba8f4e9906c7ce3c73463f62a7f0c65183ada1a2d47e397cc8810827f9694f" +checksum = "26a5c3fd7bfa1ce3897a3a3501d362b2d87b7f2583ebcb4a949ec25911025cbc" dependencies = [ "jobserver", "libc", @@ -586,18 +586,18 @@ dependencies = [ [[package]] name = "clap" -version = "4.5.9" +version = "4.5.13" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "64acc1846d54c1fe936a78dc189c34e28d3f5afc348403f28ecf53660b9b8462" +checksum = "0fbb260a053428790f3de475e304ff84cdbc4face759ea7a3e64c1edd938a7fc" dependencies = [ "clap_builder", ] [[package]] name = "clap_builder" -version = "4.5.9" +version = "4.5.13" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6fb8393d67ba2e7bfaf28a23458e4e2b543cc73a99595511eb207fdb8aede942" +checksum = "64b17d7ea74e9f833c7dbf2cbe4fb12ff26783eda4782a8975b72f895c9b4d99" dependencies = [ "anstyle", "clap_lex", @@ -605,9 +605,9 @@ dependencies = [ [[package]] name = "clap_lex" -version = "0.7.1" +version = "0.7.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4b82cf0babdbd58558212896d1a4272303a57bdb245c2bf1147185fb45640e70" +checksum = "1462739cb27611015575c0c11df5df7601141071f07518d56fcc1be504cbec97" [[package]] name = "combine" @@ -804,7 +804,7 @@ dependencies = [ [[package]] name = "datafusion" version = "40.0.0" -source = "git+https://github.com/apache/datafusion.git?rev=c6f0d3c#c6f0d3cac93ef1436313160f1dba878745d693bb" +source = "git+https://github.com/apache/datafusion.git?rev=1c98e6e#1c98e6e640090600d36ce51f87919fea7b6eeed2" dependencies = [ "ahash", "arrow", @@ -851,7 +851,7 @@ dependencies = [ [[package]] name = "datafusion-catalog" version = "40.0.0" -source = "git+https://github.com/apache/datafusion.git?rev=c6f0d3c#c6f0d3cac93ef1436313160f1dba878745d693bb" +source = "git+https://github.com/apache/datafusion.git?rev=1c98e6e#1c98e6e640090600d36ce51f87919fea7b6eeed2" dependencies = [ "arrow-schema", "async-trait", @@ -948,7 +948,7 @@ dependencies = [ [[package]] name = "datafusion-common" version = "40.0.0" -source = "git+https://github.com/apache/datafusion.git?rev=c6f0d3c#c6f0d3cac93ef1436313160f1dba878745d693bb" +source = "git+https://github.com/apache/datafusion.git?rev=1c98e6e#1c98e6e640090600d36ce51f87919fea7b6eeed2" dependencies = [ "ahash", "arrow", @@ -968,7 +968,7 @@ dependencies = [ [[package]] name = "datafusion-common-runtime" version = "40.0.0" -source = "git+https://github.com/apache/datafusion.git?rev=c6f0d3c#c6f0d3cac93ef1436313160f1dba878745d693bb" +source = "git+https://github.com/apache/datafusion.git?rev=1c98e6e#1c98e6e640090600d36ce51f87919fea7b6eeed2" dependencies = [ "tokio", ] @@ -976,7 +976,7 @@ dependencies = [ [[package]] name = "datafusion-execution" version = "40.0.0" -source = "git+https://github.com/apache/datafusion.git?rev=c6f0d3c#c6f0d3cac93ef1436313160f1dba878745d693bb" +source = "git+https://github.com/apache/datafusion.git?rev=1c98e6e#1c98e6e640090600d36ce51f87919fea7b6eeed2" dependencies = [ "arrow", "chrono", @@ -996,7 +996,7 @@ dependencies = [ [[package]] name = "datafusion-expr" version = "40.0.0" -source = "git+https://github.com/apache/datafusion.git?rev=c6f0d3c#c6f0d3cac93ef1436313160f1dba878745d693bb" +source = "git+https://github.com/apache/datafusion.git?rev=1c98e6e#1c98e6e640090600d36ce51f87919fea7b6eeed2" dependencies = [ "ahash", "arrow", @@ -1014,7 +1014,7 @@ dependencies = [ [[package]] name = "datafusion-functions" version = "40.0.0" -source = "git+https://github.com/apache/datafusion.git?rev=c6f0d3c#c6f0d3cac93ef1436313160f1dba878745d693bb" +source = "git+https://github.com/apache/datafusion.git?rev=1c98e6e#1c98e6e640090600d36ce51f87919fea7b6eeed2" dependencies = [ "arrow", "arrow-buffer", @@ -1040,7 +1040,7 @@ dependencies = [ [[package]] name = "datafusion-functions-aggregate" version = "40.0.0" -source = "git+https://github.com/apache/datafusion.git?rev=c6f0d3c#c6f0d3cac93ef1436313160f1dba878745d693bb" +source = "git+https://github.com/apache/datafusion.git?rev=1c98e6e#1c98e6e640090600d36ce51f87919fea7b6eeed2" dependencies = [ "ahash", "arrow", @@ -1057,7 +1057,7 @@ dependencies = [ [[package]] name = "datafusion-optimizer" version = "40.0.0" -source = "git+https://github.com/apache/datafusion.git?rev=c6f0d3c#c6f0d3cac93ef1436313160f1dba878745d693bb" +source = "git+https://github.com/apache/datafusion.git?rev=1c98e6e#1c98e6e640090600d36ce51f87919fea7b6eeed2" dependencies = [ "arrow", "async-trait", @@ -1076,7 +1076,7 @@ dependencies = [ [[package]] name = "datafusion-physical-expr" version = "40.0.0" -source = "git+https://github.com/apache/datafusion.git?rev=c6f0d3c#c6f0d3cac93ef1436313160f1dba878745d693bb" +source = "git+https://github.com/apache/datafusion.git?rev=1c98e6e#1c98e6e640090600d36ce51f87919fea7b6eeed2" dependencies = [ "ahash", "arrow", @@ -1105,7 +1105,7 @@ dependencies = [ [[package]] name = "datafusion-physical-expr-common" version = "40.0.0" -source = "git+https://github.com/apache/datafusion.git?rev=c6f0d3c#c6f0d3cac93ef1436313160f1dba878745d693bb" +source = "git+https://github.com/apache/datafusion.git?rev=1c98e6e#1c98e6e640090600d36ce51f87919fea7b6eeed2" dependencies = [ "ahash", "arrow", @@ -1118,7 +1118,7 @@ dependencies = [ [[package]] name = "datafusion-physical-optimizer" version = "40.0.0" -source = "git+https://github.com/apache/datafusion.git?rev=c6f0d3c#c6f0d3cac93ef1436313160f1dba878745d693bb" +source = "git+https://github.com/apache/datafusion.git?rev=1c98e6e#1c98e6e640090600d36ce51f87919fea7b6eeed2" dependencies = [ "datafusion-common", "datafusion-execution", @@ -1129,7 +1129,7 @@ dependencies = [ [[package]] name = "datafusion-physical-plan" version = "40.0.0" -source = "git+https://github.com/apache/datafusion.git?rev=c6f0d3c#c6f0d3cac93ef1436313160f1dba878745d693bb" +source = "git+https://github.com/apache/datafusion.git?rev=1c98e6e#1c98e6e640090600d36ce51f87919fea7b6eeed2" dependencies = [ "ahash", "arrow", @@ -1162,7 +1162,7 @@ dependencies = [ [[package]] name = "datafusion-sql" version = "40.0.0" -source = "git+https://github.com/apache/datafusion.git?rev=c6f0d3c#c6f0d3cac93ef1436313160f1dba878745d693bb" +source = "git+https://github.com/apache/datafusion.git?rev=1c98e6e#1c98e6e640090600d36ce51f87919fea7b6eeed2" dependencies = [ "arrow", "arrow-array", @@ -1276,9 +1276,9 @@ dependencies = [ [[package]] name = "flate2" -version = "1.0.30" +version = "1.0.31" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5f54427cfd1c7829e2a139fcefea601bf088ebca651d2bf53ebc600eac295dae" +checksum = "7f211bbe8e69bbd0cfdea405084f128ae8b4aaa6b0b522fc8f2b009084797920" dependencies = [ "crc32fast", "miniz_oxide", @@ -1525,9 +1525,9 @@ dependencies = [ [[package]] name = "indexmap" -version = "2.2.6" +version = "2.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "168fb715dda47215e360912c096649d23d58bf392ac62f73919e831745e40f26" +checksum = "de3fc2e30ba82dd1b3911c8de1ffc143c74a914a14e99514d7637e3099df5ea0" dependencies = [ "equivalent", "hashbrown", @@ -1535,9 +1535,9 @@ dependencies = [ [[package]] name = "inferno" -version = "0.11.20" +version = "0.11.21" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7c77a3ae7d4761b9c64d2c030f70746ceb8cfba32dce0325a56792e0a4816c31" +checksum = "232929e1d75fe899576a3d5c7416ad0d88dbfbb3c3d6aa00873a7408a50ddb88" dependencies = [ "ahash", "indexmap", @@ -1658,9 +1658,9 @@ checksum = "8eaf4bc02d17cbdd7ff4c7438cafcdf7fb9a4613313ad11b4f8fefe7d3fa0130" [[package]] name = "jobserver" -version = "0.1.31" +version = "0.1.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d2b099aaa34a9751c5bf0878add70444e1ed2dd73f347be99003d4577277de6e" +checksum = "48d1dbcbbeb6a7fec7e059840aa538bd62aaccf972c7346c4d9d2059312853d0" dependencies = [ "libc", ] @@ -2020,9 +2020,9 @@ dependencies = [ [[package]] name = "object" -version = "0.36.1" +version = "0.36.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "081b846d1d56ddfc18fdf1a922e4f6e07a11768ea1b92dec44e42b72712ccfce" +checksum = "3f203fa8daa7bb185f760ae12bd8e097f63d17041dcdcaf675ac54cdf863170e" dependencies = [ "memchr", ] @@ -2248,9 +2248,12 @@ dependencies = [ [[package]] name = "ppv-lite86" -version = "0.2.17" +version = "0.2.20" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5b40af805b3121feab8a3c29f04d8ad262fa8e0561883e7653e024ae4479e6de" +checksum = "77957b295656769bb8ad2b6a6b09d897d94f05c41b069aede1fcdaa675eaea04" +dependencies = [ + "zerocopy", +] [[package]] name = "proc-macro2" @@ -2416,9 +2419,9 @@ dependencies = [ [[package]] name = "regex" -version = "1.10.5" +version = "1.10.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b91213439dad192326a0d7c6ee3955910425f441d7038e0d6933b0aec5c4517f" +checksum = "4219d74c6b67a3654a9fbebc4b419e22126d13d2f3c4a07ee0cb61ff79a79619" dependencies = [ "aho-corasick", "memchr", @@ -2445,9 +2448,9 @@ checksum = "7a66a03ae7c801facd77a29370b4faec201768915ac14a721ba36f20bc9c209b" [[package]] name = "rgb" -version = "0.8.45" +version = "0.8.47" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ade4539f42266ded9e755c605bdddf546242b2c961b03b06a7375260788a0523" +checksum = "e12bc8d2f72df26a5d3178022df33720fbede0d31d82c7291662eff89836994d" dependencies = [ "bytemuck", ] @@ -2551,11 +2554,12 @@ dependencies = [ [[package]] name = "serde_json" -version = "1.0.120" +version = "1.0.122" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4e0d21c9a8cae1235ad58a00c11cb40d4b1e5c784f1ef2c537876ed6ffd8b7c5" +checksum = "784b6203951c57ff748476b126ccb5e8e2959a5c19e5c617ab1956be3dbc68da" dependencies = [ "itoa", + "memchr", "ryu", "serde", ] @@ -2708,9 +2712,9 @@ checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" [[package]] name = "symbolic-common" -version = "12.9.2" +version = "12.10.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "71297dc3e250f7dbdf8adb99e235da783d690f5819fdeb4cce39d9cfb0aca9f1" +checksum = "16629323a4ec5268ad23a575110a724ad4544aae623451de600c747bf87b36cf" dependencies = [ "debugid", "memmap2", @@ -2720,9 +2724,9 @@ dependencies = [ [[package]] name = "symbolic-demangle" -version = "12.9.2" +version = "12.10.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "424fa2c9bf2c862891b9cfd354a752751a6730fd838a4691e7f6c2c7957b9daf" +checksum = "48c043a45f08f41187414592b3ceb53fb0687da57209cc77401767fb69d5b596" dependencies = [ "cpp_demangle", "rustc-demangle", @@ -2753,14 +2757,15 @@ dependencies = [ [[package]] name = "tempfile" -version = "3.10.1" +version = "3.12.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "85b77fafb263dd9d05cbeac119526425676db3784113aa9295c88498cbf8bff1" +checksum = "04cbcdd0c794ebb0d4cf35e88edd2f7d2c4c3e9a5a6dab322839b321c6a87a64" dependencies = [ "cfg-if", "fastrand", + "once_cell", "rustix", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -2840,22 +2845,21 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" [[package]] name = "tokio" -version = "1.38.1" +version = "1.39.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "eb2caba9f80616f438e09748d5acda951967e1ea58508ef53d9c6402485a46df" +checksum = "daa4fb1bc778bd6f04cbfc4bb2d06a7396a8f299dc33ea1900cedaa316f467b1" dependencies = [ "backtrace", "bytes", - "num_cpus", "pin-project-lite", "tokio-macros", ] [[package]] name = "tokio-macros" -version = "2.3.0" +version = "2.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5f5ae998a069d4b5aba8ee9dad856af7d520c3699e6159b185c2acd48155d39a" +checksum = "693d596312e88961bc67d7f1f97af8a70227d9f90c31bba5806eec004978d752" dependencies = [ "proc-macro2", "quote", @@ -2989,9 +2993,9 @@ dependencies = [ [[package]] name = "version_check" -version = "0.9.4" +version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "49874b5167b65d7193b8aba1567f5c7d93d001cafc34600cee003eda787e483f" +checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" [[package]] name = "walkdir" @@ -3103,11 +3107,11 @@ checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6" [[package]] name = "winapi-util" -version = "0.1.8" +version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4d4cc384e1e73b93bafa6fb4f1df8c41695c8a91cf9c4c64358067d15a7b6c6b" +checksum = "cf221c93e13a30d793f7645a0e7762c55d169dbb0a49671918a2319d289b10bb" dependencies = [ - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -3143,6 +3147,15 @@ dependencies = [ "windows-targets 0.52.6", ] +[[package]] +name = "windows-sys" +version = "0.59.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e38bc4d79ed67fd075bcc251a1c39b32a1776bbe92e5bef1f0bf1f8c531853b" +dependencies = [ + "windows-targets 0.52.6", +] + [[package]] name = "windows-targets" version = "0.42.2" @@ -3270,6 +3283,7 @@ version = "0.7.35" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1b9b4fd18abc82b8136838da5d50bae7bdea537c574d8dc1a34ed098d6c166f0" dependencies = [ + "byteorder", "zerocopy-derive", ] @@ -3305,9 +3319,9 @@ dependencies = [ [[package]] name = "zstd-sys" -version = "2.0.12+zstd.1.5.6" +version = "2.0.13+zstd.1.5.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0a4e40c320c3cb459d9a9ff6de98cff88f4751ee9275d140e2be94a2b74e4c13" +checksum = "38ff0f21cfee8f97d94cef41359e0c89aa6113028ab0291aa8ca0038995a95aa" dependencies = [ "cc", "pkg-config", diff --git a/native/Cargo.toml b/native/Cargo.toml index a9b7d4fbdf5..4acad47b4a0 100644 --- a/native/Cargo.toml +++ b/native/Cargo.toml @@ -39,13 +39,13 @@ arrow-buffer = { version = "52.2.0" } arrow-data = { version = "52.2.0" } arrow-schema = { version = "52.2.0" } parquet = { version = "52.2.0", default-features = false, features = ["experimental"] } -datafusion-common = { git = "https://github.com/apache/datafusion.git", rev = "f4e519f" } -datafusion = { default-features = false, git = "https://github.com/apache/datafusion.git", rev = "f4e519f", features = ["unicode_expressions", "crypto_expressions"] } -datafusion-functions = { git = "https://github.com/apache/datafusion.git", rev = "f4e519f", features = ["crypto_expressions"] } -datafusion-expr = { git = "https://github.com/apache/datafusion.git", rev = "f4e519f", default-features = false } -datafusion-physical-plan = { git = "https://github.com/apache/datafusion.git", rev = "f4e519f", default-features = false } -datafusion-physical-expr-common = { git = "https://github.com/apache/datafusion.git", rev = "f4e519f", default-features = false } -datafusion-physical-expr = { git = "https://github.com/apache/datafusion.git", rev = "f4e519f", default-features = false } +datafusion-common = { git = "https://github.com/apache/datafusion.git", rev = "1c98e6e" } +datafusion = { default-features = false, git = "https://github.com/apache/datafusion.git", rev = "1c98e6e", features = ["unicode_expressions", "crypto_expressions"] } +datafusion-functions = { git = "https://github.com/apache/datafusion.git", rev = "1c98e6e", features = ["crypto_expressions"] } +datafusion-expr = { git = "https://github.com/apache/datafusion.git", rev = "1c98e6e", default-features = false } +datafusion-physical-plan = { git = "https://github.com/apache/datafusion.git", rev = "1c98e6e", default-features = false } +datafusion-physical-expr-common = { git = "https://github.com/apache/datafusion.git", rev = "1c98e6e", default-features = false } +datafusion-physical-expr = { git = "https://github.com/apache/datafusion.git", rev = "1c98e6e", default-features = false } datafusion-comet-spark-expr = { path = "spark-expr", version = "0.2.0" } datafusion-comet-proto = { path = "proto", version = "0.2.0" } chrono = { version = "0.4", default-features = false, features = ["clock"] } diff --git a/native/core/src/execution/datafusion/planner.rs b/native/core/src/execution/datafusion/planner.rs index 836503aa515..eb1a7ff6f8d 100644 --- a/native/core/src/execution/datafusion/planner.rs +++ b/native/core/src/execution/datafusion/planner.rs @@ -111,6 +111,7 @@ use jni::objects::GlobalRef; use num::{BigInt, ToPrimitive}; use std::cmp::max; use std::{collections::HashMap, sync::Arc}; +use datafusion::prelude::SessionConfig; // For clippy error on type_complexity. type ExecResult = Result; @@ -138,7 +139,9 @@ pub struct PhysicalPlanner { impl Default for PhysicalPlanner { fn default() -> Self { - let session_ctx = Arc::new(SessionContext::new()); + let cfg = SessionConfig::new() + .set("datafusion.execution.skip_partial_aggregation_probe_ratio_threshold", ScalarValue::Float64(Some(1.1))); + let session_ctx = Arc::new(SessionContext::new_with_config(cfg)); let execution_props = ExecutionProps::new(); Self { exec_context_id: TEST_EXEC_CONTEXT_ID, @@ -1269,7 +1272,7 @@ impl PhysicalPlanner { &[], &[], schema.as_ref(), - "count", + Some("count".to_string()), false, false, ) @@ -1286,7 +1289,7 @@ impl PhysicalPlanner { &[], &[], schema.as_ref(), - "min", + Some("min".to_string()), false, false, ) @@ -1303,7 +1306,7 @@ impl PhysicalPlanner { &[], &[], schema.as_ref(), - "max", + Some("max".to_string()), false, false, ) @@ -1328,7 +1331,7 @@ impl PhysicalPlanner { &[], &[], schema.as_ref(), - "sum", + Some("sum".to_string()), false, false, ) @@ -1366,7 +1369,7 @@ impl PhysicalPlanner { &[], &[], &schema, - "first", + Some("first".to_string()), false, false, ) @@ -1382,7 +1385,7 @@ impl PhysicalPlanner { &[], &[], &schema, - "last", + Some("last".to_string()), false, false, ) @@ -1397,7 +1400,7 @@ impl PhysicalPlanner { &[], &[], &schema, - "bit_and", + Some("bit_and".to_string()), false, false, ) @@ -1412,7 +1415,7 @@ impl PhysicalPlanner { &[], &[], &schema, - "bit_or", + Some("bit_or".to_string()), false, false, ) @@ -1427,7 +1430,7 @@ impl PhysicalPlanner { &[], &[], &schema, - "bit_xor", + Some("bit_xor".to_string()), false, false, ) From 9c9f2d8d0c90c6e027503c13b18b7a721ed79dc1 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Tue, 6 Aug 2024 16:58:17 -0600 Subject: [PATCH 2/5] add comment --- native/core/src/execution/datafusion/planner.rs | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/native/core/src/execution/datafusion/planner.rs b/native/core/src/execution/datafusion/planner.rs index eb1a7ff6f8d..3fc9341f239 100644 --- a/native/core/src/execution/datafusion/planner.rs +++ b/native/core/src/execution/datafusion/planner.rs @@ -55,6 +55,7 @@ use datafusion::functions_aggregate::min_max::min_udaf; use datafusion::functions_aggregate::sum::sum_udaf; use datafusion::physical_plan::windows::BoundedWindowAggExec; use datafusion::physical_plan::InputOrderMode; +use datafusion::prelude::SessionConfig; use datafusion::{ arrow::{compute::SortOptions, datatypes::SchemaRef}, common::DataFusionError, @@ -111,7 +112,6 @@ use jni::objects::GlobalRef; use num::{BigInt, ToPrimitive}; use std::cmp::max; use std::{collections::HashMap, sync::Arc}; -use datafusion::prelude::SessionConfig; // For clippy error on type_complexity. type ExecResult = Result; @@ -139,8 +139,13 @@ pub struct PhysicalPlanner { impl Default for PhysicalPlanner { fn default() -> Self { - let cfg = SessionConfig::new() - .set("datafusion.execution.skip_partial_aggregation_probe_ratio_threshold", ScalarValue::Float64(Some(1.1))); + // DataFusion partial aggregates can emit duplicate rows so we disable the + // skip partial aggregation feature because this is not compatible with Spark's + // use of parital aggregates. + let cfg = SessionConfig::new().set( + "datafusion.execution.skip_partial_aggregation_probe_ratio_threshold", + ScalarValue::Float64(Some(1.1)), + ); let session_ctx = Arc::new(SessionContext::new_with_config(cfg)); let execution_props = ExecutionProps::new(); Self { From cc7205b143df46c791b4d0f346028acd3ab8f56f Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Tue, 6 Aug 2024 17:22:28 -0600 Subject: [PATCH 3/5] Save --- native/Cargo.lock | 28 ++-- native/Cargo.toml | 14 +- .../spark/sql/CometTPCDSQuerySuite.scala | 145 +----------------- 3 files changed, 24 insertions(+), 163 deletions(-) diff --git a/native/Cargo.lock b/native/Cargo.lock index 581beef0608..b34ed54bc78 100644 --- a/native/Cargo.lock +++ b/native/Cargo.lock @@ -804,7 +804,7 @@ dependencies = [ [[package]] name = "datafusion" version = "40.0.0" -source = "git+https://github.com/apache/datafusion.git?rev=1c98e6e#1c98e6e640090600d36ce51f87919fea7b6eeed2" +source = "git+https://github.com/apache/datafusion.git?rev=bddb641#bddb6415a50746d2803dd908d19c3758952d74f9" dependencies = [ "ahash", "arrow", @@ -851,7 +851,7 @@ dependencies = [ [[package]] name = "datafusion-catalog" version = "40.0.0" -source = "git+https://github.com/apache/datafusion.git?rev=1c98e6e#1c98e6e640090600d36ce51f87919fea7b6eeed2" +source = "git+https://github.com/apache/datafusion.git?rev=bddb641#bddb6415a50746d2803dd908d19c3758952d74f9" dependencies = [ "arrow-schema", "async-trait", @@ -948,7 +948,7 @@ dependencies = [ [[package]] name = "datafusion-common" version = "40.0.0" -source = "git+https://github.com/apache/datafusion.git?rev=1c98e6e#1c98e6e640090600d36ce51f87919fea7b6eeed2" +source = "git+https://github.com/apache/datafusion.git?rev=bddb641#bddb6415a50746d2803dd908d19c3758952d74f9" dependencies = [ "ahash", "arrow", @@ -968,7 +968,7 @@ dependencies = [ [[package]] name = "datafusion-common-runtime" version = "40.0.0" -source = "git+https://github.com/apache/datafusion.git?rev=1c98e6e#1c98e6e640090600d36ce51f87919fea7b6eeed2" +source = "git+https://github.com/apache/datafusion.git?rev=bddb641#bddb6415a50746d2803dd908d19c3758952d74f9" dependencies = [ "tokio", ] @@ -976,7 +976,7 @@ dependencies = [ [[package]] name = "datafusion-execution" version = "40.0.0" -source = "git+https://github.com/apache/datafusion.git?rev=1c98e6e#1c98e6e640090600d36ce51f87919fea7b6eeed2" +source = "git+https://github.com/apache/datafusion.git?rev=bddb641#bddb6415a50746d2803dd908d19c3758952d74f9" dependencies = [ "arrow", "chrono", @@ -996,7 +996,7 @@ dependencies = [ [[package]] name = "datafusion-expr" version = "40.0.0" -source = "git+https://github.com/apache/datafusion.git?rev=1c98e6e#1c98e6e640090600d36ce51f87919fea7b6eeed2" +source = "git+https://github.com/apache/datafusion.git?rev=bddb641#bddb6415a50746d2803dd908d19c3758952d74f9" dependencies = [ "ahash", "arrow", @@ -1014,7 +1014,7 @@ dependencies = [ [[package]] name = "datafusion-functions" version = "40.0.0" -source = "git+https://github.com/apache/datafusion.git?rev=1c98e6e#1c98e6e640090600d36ce51f87919fea7b6eeed2" +source = "git+https://github.com/apache/datafusion.git?rev=bddb641#bddb6415a50746d2803dd908d19c3758952d74f9" dependencies = [ "arrow", "arrow-buffer", @@ -1040,7 +1040,7 @@ dependencies = [ [[package]] name = "datafusion-functions-aggregate" version = "40.0.0" -source = "git+https://github.com/apache/datafusion.git?rev=1c98e6e#1c98e6e640090600d36ce51f87919fea7b6eeed2" +source = "git+https://github.com/apache/datafusion.git?rev=bddb641#bddb6415a50746d2803dd908d19c3758952d74f9" dependencies = [ "ahash", "arrow", @@ -1057,7 +1057,7 @@ dependencies = [ [[package]] name = "datafusion-optimizer" version = "40.0.0" -source = "git+https://github.com/apache/datafusion.git?rev=1c98e6e#1c98e6e640090600d36ce51f87919fea7b6eeed2" +source = "git+https://github.com/apache/datafusion.git?rev=bddb641#bddb6415a50746d2803dd908d19c3758952d74f9" dependencies = [ "arrow", "async-trait", @@ -1076,7 +1076,7 @@ dependencies = [ [[package]] name = "datafusion-physical-expr" version = "40.0.0" -source = "git+https://github.com/apache/datafusion.git?rev=1c98e6e#1c98e6e640090600d36ce51f87919fea7b6eeed2" +source = "git+https://github.com/apache/datafusion.git?rev=bddb641#bddb6415a50746d2803dd908d19c3758952d74f9" dependencies = [ "ahash", "arrow", @@ -1105,7 +1105,7 @@ dependencies = [ [[package]] name = "datafusion-physical-expr-common" version = "40.0.0" -source = "git+https://github.com/apache/datafusion.git?rev=1c98e6e#1c98e6e640090600d36ce51f87919fea7b6eeed2" +source = "git+https://github.com/apache/datafusion.git?rev=bddb641#bddb6415a50746d2803dd908d19c3758952d74f9" dependencies = [ "ahash", "arrow", @@ -1118,7 +1118,7 @@ dependencies = [ [[package]] name = "datafusion-physical-optimizer" version = "40.0.0" -source = "git+https://github.com/apache/datafusion.git?rev=1c98e6e#1c98e6e640090600d36ce51f87919fea7b6eeed2" +source = "git+https://github.com/apache/datafusion.git?rev=bddb641#bddb6415a50746d2803dd908d19c3758952d74f9" dependencies = [ "datafusion-common", "datafusion-execution", @@ -1129,7 +1129,7 @@ dependencies = [ [[package]] name = "datafusion-physical-plan" version = "40.0.0" -source = "git+https://github.com/apache/datafusion.git?rev=1c98e6e#1c98e6e640090600d36ce51f87919fea7b6eeed2" +source = "git+https://github.com/apache/datafusion.git?rev=bddb641#bddb6415a50746d2803dd908d19c3758952d74f9" dependencies = [ "ahash", "arrow", @@ -1162,7 +1162,7 @@ dependencies = [ [[package]] name = "datafusion-sql" version = "40.0.0" -source = "git+https://github.com/apache/datafusion.git?rev=1c98e6e#1c98e6e640090600d36ce51f87919fea7b6eeed2" +source = "git+https://github.com/apache/datafusion.git?rev=bddb641#bddb6415a50746d2803dd908d19c3758952d74f9" dependencies = [ "arrow", "arrow-array", diff --git a/native/Cargo.toml b/native/Cargo.toml index 4acad47b4a0..8e51396535e 100644 --- a/native/Cargo.toml +++ b/native/Cargo.toml @@ -39,13 +39,13 @@ arrow-buffer = { version = "52.2.0" } arrow-data = { version = "52.2.0" } arrow-schema = { version = "52.2.0" } parquet = { version = "52.2.0", default-features = false, features = ["experimental"] } -datafusion-common = { git = "https://github.com/apache/datafusion.git", rev = "1c98e6e" } -datafusion = { default-features = false, git = "https://github.com/apache/datafusion.git", rev = "1c98e6e", features = ["unicode_expressions", "crypto_expressions"] } -datafusion-functions = { git = "https://github.com/apache/datafusion.git", rev = "1c98e6e", features = ["crypto_expressions"] } -datafusion-expr = { git = "https://github.com/apache/datafusion.git", rev = "1c98e6e", default-features = false } -datafusion-physical-plan = { git = "https://github.com/apache/datafusion.git", rev = "1c98e6e", default-features = false } -datafusion-physical-expr-common = { git = "https://github.com/apache/datafusion.git", rev = "1c98e6e", default-features = false } -datafusion-physical-expr = { git = "https://github.com/apache/datafusion.git", rev = "1c98e6e", default-features = false } +datafusion-common = { git = "https://github.com/apache/datafusion.git", rev = "bddb641" } +datafusion = { default-features = false, git = "https://github.com/apache/datafusion.git", rev = "bddb641", features = ["unicode_expressions", "crypto_expressions"] } +datafusion-functions = { git = "https://github.com/apache/datafusion.git", rev = "bddb641", features = ["crypto_expressions"] } +datafusion-expr = { git = "https://github.com/apache/datafusion.git", rev = "bddb641", default-features = false } +datafusion-physical-plan = { git = "https://github.com/apache/datafusion.git", rev = "bddb641", default-features = false } +datafusion-physical-expr-common = { git = "https://github.com/apache/datafusion.git", rev = "bddb641", default-features = false } +datafusion-physical-expr = { git = "https://github.com/apache/datafusion.git", rev = "bddb641", default-features = false } datafusion-comet-spark-expr = { path = "spark-expr", version = "0.2.0" } datafusion-comet-proto = { path = "proto", version = "0.2.0" } chrono = { version = "0.4", default-features = false, features = ["clock"] } diff --git a/spark/src/test/scala/org/apache/spark/sql/CometTPCDSQuerySuite.scala b/spark/src/test/scala/org/apache/spark/sql/CometTPCDSQuerySuite.scala index 864f2c03e0c..80a6b3e0584 100644 --- a/spark/src/test/scala/org/apache/spark/sql/CometTPCDSQuerySuite.scala +++ b/spark/src/test/scala/org/apache/spark/sql/CometTPCDSQuerySuite.scala @@ -27,149 +27,9 @@ import org.apache.comet.CometConf class CometTPCDSQuerySuite extends { - val tpcdsAllQueries: Seq[String] = Seq( - "q1", - "q2", - "q3", - "q4", - "q5", - "q6", - "q7", - "q8", - "q9", - "q10", - "q11", - "q12", - "q13", - "q14a", - "q14b", - "q15", - "q16", - "q17", - "q18", - "q19", - "q20", - "q21", - "q22", - "q23a", - "q23b", - "q24a", - "q24b", - "q25", - "q26", - "q27", - "q28", - "q29", - "q30", - "q31", - "q32", - "q33", - "q34", - "q35", - "q36", - "q37", - "q38", - // TODO: https://github.com/apache/datafusion-comet/issues/392 - // comment out 39a and 39b for now because the expected result for stddev failed: - // expected: 1.5242630430075292, actual: 1.524263043007529. - // Will change the comparison logic to detect floating-point numbers and compare - // with epsilon - // "q39a", - // "q39b", - "q40", - "q41", - "q42", - "q43", - "q44", - "q45", - "q46", - "q47", - "q48", - "q49", - "q50", - "q51", - "q52", - "q53", - "q54", - "q55", - "q56", - "q57", - "q58", - "q59", - "q60", - "q61", - "q62", - "q63", - "q64", - "q65", - "q66", - "q67", - "q68", - "q69", - "q70", - "q71", - "q72", - "q73", - "q74", - "q75", - "q76", - "q77", - "q78", - "q79", - "q80", - "q81", - "q82", - "q83", - "q84", - "q85", - "q86", - "q87", - "q88", - "q89", - "q90", - "q91", - "q92", - "q93", - "q94", - "q95", - "q96", - "q97", - "q98", - "q99") + val tpcdsAllQueries: Seq[String] = Seq("q97") - val tpcdsAllQueriesV2_7_0: Seq[String] = Seq( - "q5a", - "q6", - "q10a", - "q11", - "q12", - "q14", - "q14a", - "q18a", - "q20", - "q22", - "q22a", - "q24", - "q27a", - "q34", - "q35", - "q35a", - "q36a", - "q47", - "q49", - "q51a", - "q57", - "q64", - "q67a", - "q70a", - "q72", - "q74", - "q75", - "q77a", - "q78", - "q80a", - "q86a", - "q98") + val tpcdsAllQueriesV2_7_0: Seq[String] = Seq() override val tpcdsQueries: Seq[String] = tpcdsAllQueries @@ -184,6 +44,7 @@ class CometTPCDSQuerySuite "spark.shuffle.manager", "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager") conf.set(CometConf.COMET_ENABLED.key, "true") + conf.set(CometConf.COMET_DEBUG_ENABLED.key, "true") conf.set(CometConf.COMET_EXEC_ENABLED.key, "true") conf.set(CometConf.COMET_EXEC_ALL_OPERATOR_ENABLED.key, "true") conf.set(CometConf.COMET_EXEC_SHUFFLE_ENABLED.key, "true") From c5e023e8b48622e26cd78a5275a0b6c8832cd2b7 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Tue, 6 Aug 2024 17:51:31 -0600 Subject: [PATCH 4/5] Revert debug changes --- .../core/src/execution/datafusion/planner.rs | 10 +- native/core/src/execution/jni_api.rs | 11 +- .../spark/sql/CometTPCDSQuerySuite.scala | 145 +++++++++++++++++- 3 files changed, 153 insertions(+), 13 deletions(-) diff --git a/native/core/src/execution/datafusion/planner.rs b/native/core/src/execution/datafusion/planner.rs index 3fc9341f239..a16ceda8c39 100644 --- a/native/core/src/execution/datafusion/planner.rs +++ b/native/core/src/execution/datafusion/planner.rs @@ -55,7 +55,6 @@ use datafusion::functions_aggregate::min_max::min_udaf; use datafusion::functions_aggregate::sum::sum_udaf; use datafusion::physical_plan::windows::BoundedWindowAggExec; use datafusion::physical_plan::InputOrderMode; -use datafusion::prelude::SessionConfig; use datafusion::{ arrow::{compute::SortOptions, datatypes::SchemaRef}, common::DataFusionError, @@ -139,14 +138,7 @@ pub struct PhysicalPlanner { impl Default for PhysicalPlanner { fn default() -> Self { - // DataFusion partial aggregates can emit duplicate rows so we disable the - // skip partial aggregation feature because this is not compatible with Spark's - // use of parital aggregates. - let cfg = SessionConfig::new().set( - "datafusion.execution.skip_partial_aggregation_probe_ratio_threshold", - ScalarValue::Float64(Some(1.1)), - ); - let session_ctx = Arc::new(SessionContext::new_with_config(cfg)); + let session_ctx = Arc::new(SessionContext::new()); let execution_props = ExecutionProps::new(); Self { exec_context_id: TEST_EXEC_CONTEXT_ID, diff --git a/native/core/src/execution/jni_api.rs b/native/core/src/execution/jni_api.rs index 285d63e957b..2c11d7f19f5 100644 --- a/native/core/src/execution/jni_api.rs +++ b/native/core/src/execution/jni_api.rs @@ -53,6 +53,7 @@ use crate::{ jvm_bridge::{jni_new_global_ref, JVMClasses}, }; use datafusion_comet_proto::spark_operator::Operator; +use datafusion_common::ScalarValue; use futures::stream::StreamExt; use jni::{ objects::GlobalRef, @@ -219,7 +220,15 @@ fn prepare_datafusion_session_context( // Get Datafusion configuration from Spark Execution context // can be configured in Comet Spark JVM using Spark --conf parameters // e.g: spark-shell --conf spark.datafusion.sql_parser.parse_float_as_decimal=true - let mut session_config = SessionConfig::new().with_batch_size(batch_size); + let mut session_config = SessionConfig::new() + .with_batch_size(batch_size) + // DataFusion partial aggregates can emit duplicate rows so we disable the + // skip partial aggregation feature because this is not compatible with Spark's + // use of partial aggregates. + .set( + "datafusion.execution.skip_partial_aggregation_probe_ratio_threshold", + ScalarValue::Float64(Some(1.1)), + ); for (key, value) in conf.iter().filter(|(k, _)| k.starts_with("datafusion.")) { session_config = session_config.set_str(key, value); diff --git a/spark/src/test/scala/org/apache/spark/sql/CometTPCDSQuerySuite.scala b/spark/src/test/scala/org/apache/spark/sql/CometTPCDSQuerySuite.scala index 80a6b3e0584..864f2c03e0c 100644 --- a/spark/src/test/scala/org/apache/spark/sql/CometTPCDSQuerySuite.scala +++ b/spark/src/test/scala/org/apache/spark/sql/CometTPCDSQuerySuite.scala @@ -27,9 +27,149 @@ import org.apache.comet.CometConf class CometTPCDSQuerySuite extends { - val tpcdsAllQueries: Seq[String] = Seq("q97") + val tpcdsAllQueries: Seq[String] = Seq( + "q1", + "q2", + "q3", + "q4", + "q5", + "q6", + "q7", + "q8", + "q9", + "q10", + "q11", + "q12", + "q13", + "q14a", + "q14b", + "q15", + "q16", + "q17", + "q18", + "q19", + "q20", + "q21", + "q22", + "q23a", + "q23b", + "q24a", + "q24b", + "q25", + "q26", + "q27", + "q28", + "q29", + "q30", + "q31", + "q32", + "q33", + "q34", + "q35", + "q36", + "q37", + "q38", + // TODO: https://github.com/apache/datafusion-comet/issues/392 + // comment out 39a and 39b for now because the expected result for stddev failed: + // expected: 1.5242630430075292, actual: 1.524263043007529. + // Will change the comparison logic to detect floating-point numbers and compare + // with epsilon + // "q39a", + // "q39b", + "q40", + "q41", + "q42", + "q43", + "q44", + "q45", + "q46", + "q47", + "q48", + "q49", + "q50", + "q51", + "q52", + "q53", + "q54", + "q55", + "q56", + "q57", + "q58", + "q59", + "q60", + "q61", + "q62", + "q63", + "q64", + "q65", + "q66", + "q67", + "q68", + "q69", + "q70", + "q71", + "q72", + "q73", + "q74", + "q75", + "q76", + "q77", + "q78", + "q79", + "q80", + "q81", + "q82", + "q83", + "q84", + "q85", + "q86", + "q87", + "q88", + "q89", + "q90", + "q91", + "q92", + "q93", + "q94", + "q95", + "q96", + "q97", + "q98", + "q99") - val tpcdsAllQueriesV2_7_0: Seq[String] = Seq() + val tpcdsAllQueriesV2_7_0: Seq[String] = Seq( + "q5a", + "q6", + "q10a", + "q11", + "q12", + "q14", + "q14a", + "q18a", + "q20", + "q22", + "q22a", + "q24", + "q27a", + "q34", + "q35", + "q35a", + "q36a", + "q47", + "q49", + "q51a", + "q57", + "q64", + "q67a", + "q70a", + "q72", + "q74", + "q75", + "q77a", + "q78", + "q80a", + "q86a", + "q98") override val tpcdsQueries: Seq[String] = tpcdsAllQueries @@ -44,7 +184,6 @@ class CometTPCDSQuerySuite "spark.shuffle.manager", "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager") conf.set(CometConf.COMET_ENABLED.key, "true") - conf.set(CometConf.COMET_DEBUG_ENABLED.key, "true") conf.set(CometConf.COMET_EXEC_ENABLED.key, "true") conf.set(CometConf.COMET_EXEC_ALL_OPERATOR_ENABLED.key, "true") conf.set(CometConf.COMET_EXEC_SHUFFLE_ENABLED.key, "true") From 5a847fe8a5b0e4209935d27d21222bf56db312b9 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Tue, 6 Aug 2024 17:55:54 -0600 Subject: [PATCH 5/5] add comment --- native/core/src/execution/jni_api.rs | 3 +++ 1 file changed, 3 insertions(+) diff --git a/native/core/src/execution/jni_api.rs b/native/core/src/execution/jni_api.rs index 2c11d7f19f5..eb5f698b40d 100644 --- a/native/core/src/execution/jni_api.rs +++ b/native/core/src/execution/jni_api.rs @@ -227,6 +227,9 @@ fn prepare_datafusion_session_context( // use of partial aggregates. .set( "datafusion.execution.skip_partial_aggregation_probe_ratio_threshold", + // this is the threshold of number of groups / number of rows and the + // maximum value is 1.0, so we set the threshold a little higher just + // to be safe ScalarValue::Float64(Some(1.1)), );