From fdc82220cb5c7e9b7ff2bbabac073820ff3df7fb Mon Sep 17 00:00:00 2001 From: Denys Tsomenko Date: Thu, 2 Jul 2026 13:51:31 +0300 Subject: [PATCH 1/3] Decimal division and AVG round half away from zero (Snowflake semantics) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Snowflake rounds the last digit of decimal division and AVG results half away from zero (verified live: 5/3 = 1.666667, -5/3 = -1.666667, 5/2000000 = 0.000003 — not half-even, not truncation). Arrow's div kernel and DecimalAverager::avg both truncate toward zero, which produced the persistent <=1 ULP parity gap on TPC-H Q1/Q8 and ClickBench Q28. - DecimalAverager::avg: correct the truncated quotient with the remainder. - BinaryExpr Divide on decimals: dispatch to a fork-local kernel that replicates arrow-arith's decimal Op::Div exactly (result precision/scale derivation, overflow and divide-by-zero errors) but rounds half away from zero. Forking arrow itself was rejected as far heavier. - decimal.slt expectations regenerated; pre-existing failures in subquery/dates/encrypted_parquet/pivot slt files are unchanged (verified identical on the clean base). - Bump the workspace sqlparser pin to the Snowflake string-escape fix rev (074bd21) so downstream consumers resolve a single sqlparser instance. Co-Authored-By: Claude Opus 4.8 (1M context) --- Cargo.lock | 14 +- Cargo.toml | 2 +- .../functions-aggregate-common/src/utils.rs | 42 ++- .../physical-expr/src/expressions/binary.rs | 15 +- .../src/expressions/binary/decimal_div.rs | 248 ++++++++++++++++++ .../sqllogictest/test_files/decimal.slt | 14 +- 6 files changed, 318 insertions(+), 17 deletions(-) create mode 100644 datafusion/physical-expr/src/expressions/binary/decimal_div.rs diff --git a/Cargo.lock b/Cargo.lock index 0976084222870..e6a2b3985e6e6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3101,7 +3101,7 @@ checksum = "0ce92ff622d6dadf7349484f42c93271a0d49b7cc4d466a936405bacbe10aa78" dependencies = [ "cfg-if", "rustix 1.0.8", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -5147,7 +5147,7 @@ dependencies = [ "once_cell", "socket2 0.5.10", "tracing", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -5543,7 +5543,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.4.15", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -6063,7 +6063,7 @@ dependencies = [ [[package]] name = "sqlparser" version = "0.61.0" -source = "git+https://github.com/Embucket/datafusion-sqlparser-rs.git?rev=582112f50be00d6452e33adfd857d6d467bb5d72#582112f50be00d6452e33adfd857d6d467bb5d72" +source = "git+https://github.com/Embucket/datafusion-sqlparser-rs.git?rev=074bd2124c2e55b43528e97890937c2698d100ae#074bd2124c2e55b43528e97890937c2698d100ae" dependencies = [ "log", "recursive", @@ -6073,7 +6073,7 @@ dependencies = [ [[package]] name = "sqlparser_derive" version = "0.5.0" -source = "git+https://github.com/Embucket/datafusion-sqlparser-rs.git?rev=582112f50be00d6452e33adfd857d6d467bb5d72#582112f50be00d6452e33adfd857d6d467bb5d72" +source = "git+https://github.com/Embucket/datafusion-sqlparser-rs.git?rev=074bd2124c2e55b43528e97890937c2698d100ae#074bd2124c2e55b43528e97890937c2698d100ae" dependencies = [ "proc-macro2", "quote", @@ -6096,7 +6096,7 @@ dependencies = [ "cfg-if", "libc", "psm", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -7233,7 +7233,7 @@ version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cf221c93e13a30d793f7645a0e7762c55d169dbb0a49671918a2319d289b10bb" dependencies = [ - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 63f4ddadc563c..bf6510839772e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -184,7 +184,7 @@ regex = "1.12" rstest = "0.26.1" serde_json = "1" sha2 = "^0.10.9" -sqlparser = { git = "https://github.com/Embucket/datafusion-sqlparser-rs.git", rev = "582112f50be00d6452e33adfd857d6d467bb5d72", features = [ +sqlparser = { git = "https://github.com/Embucket/datafusion-sqlparser-rs.git", rev = "074bd2124c2e55b43528e97890937c2698d100ae", features = [ "visitor", ] } strum = "0.28.0" diff --git a/datafusion/functions-aggregate-common/src/utils.rs b/datafusion/functions-aggregate-common/src/utils.rs index 78b8c52490c76..38129df994778 100644 --- a/datafusion/functions-aggregate-common/src/utils.rs +++ b/datafusion/functions-aggregate-common/src/utils.rs @@ -156,7 +156,24 @@ impl DecimalAverager { #[inline(always)] pub fn avg(&self, sum: T::Native, count: T::Native) -> Result { if let Ok(value) = sum.mul_checked(self.target_mul.div_wrapping(self.sum_mul)) { - let new_value = value.div_wrapping(count); + let mut new_value = value.div_wrapping(count); + + // Round half away from zero (Snowflake semantics) instead of + // truncating toward zero: if twice the remainder reaches the + // divisor, bump the quotient one step in the sign of the exact + // result. `abs_rem >= count - abs_rem` avoids overflowing 2*rem. + let rem = value.mod_wrapping(count); + if !rem.is_zero() { + let negative = rem.is_lt(T::Native::ZERO); + let abs_rem = if negative { rem.neg_wrapping() } else { rem }; + if abs_rem.is_ge(count.sub_wrapping(abs_rem)) { + new_value = if negative { + new_value.sub_wrapping(T::Native::ONE) + } else { + new_value.add_wrapping(T::Native::ONE) + }; + } + } let validate = T::validate_decimal_precision( new_value, @@ -264,3 +281,26 @@ impl GenericDistinctBuffer { estimate_memory_size::(num_elements, fixed_size).unwrap() } } + +#[cfg(test)] +mod tests { + use super::*; + use arrow::datatypes::Decimal128Type; + + #[test] + fn decimal_averager_rounds_half_away_from_zero() { + // sum scale 0 -> target scale 4 (Snowflake AVG semantics verified + // live: AVG rounds the last digit half away from zero, both signs). + let averager = DecimalAverager::::try_new(0, 10, 4).unwrap(); + // 5/3 = 1.666666... -> 1.6667, not the truncated 1.6666 + assert_eq!(averager.avg(5, 3).unwrap(), 16667); + assert_eq!(averager.avg(-5, 3).unwrap(), -16667); + // 1/32 = 0.03125: exact half at the last digit rounds away from zero + // (0.0313), where truncation and half-even both give 0.0312. + assert_eq!(averager.avg(1, 32).unwrap(), 313); + assert_eq!(averager.avg(-1, 32).unwrap(), -313); + // Exact quotients are unchanged. + assert_eq!(averager.avg(6, 3).unwrap(), 20000); + assert_eq!(averager.avg(1, 8).unwrap(), 1250); + } +} diff --git a/datafusion/physical-expr/src/expressions/binary.rs b/datafusion/physical-expr/src/expressions/binary.rs index 02628b405ec6c..c5a00cc6e752e 100644 --- a/datafusion/physical-expr/src/expressions/binary.rs +++ b/datafusion/physical-expr/src/expressions/binary.rs @@ -15,6 +15,7 @@ // specific language governing permissions and limitations // under the License. +mod decimal_div; mod kernels; use crate::PhysicalExpr; @@ -349,6 +350,16 @@ impl PhysicalExpr for BinaryExpr { Operator::Minus => return apply(&lhs, &rhs, sub_wrapping), Operator::Multiply if self.fail_on_overflow => return apply(&lhs, &rhs, mul), Operator::Multiply => return apply(&lhs, &rhs, mul_wrapping), + // Decimal division rounds half away from zero (Snowflake + // semantics); arrow's `div` kernel truncates. + Operator::Divide + if decimal_div::is_decimal_division( + &left_data_type, + &right_data_type, + ) => + { + return apply(&lhs, &rhs, decimal_div::div_half_away_from_zero); + } Operator::Divide => return apply(&lhs, &rhs, div), Operator::Modulo => return apply(&lhs, &rhs, rem), @@ -4052,8 +4063,10 @@ mod tests { Field::new("a", DataType::Int32, true), Field::new("b", DataType::Decimal128(10, 2), true), ])); + // 123/122.00 = 1.00819672... rounds half away from zero to 1.0082 + // (decimal division no longer truncates; see decimal_div module). let expect = Arc::new(create_decimal_array( - &[Some(1000000), None, Some(1008196), Some(1000000)], + &[Some(1000000), None, Some(1008197), Some(1000000)], 16, 4, )) as ArrayRef; diff --git a/datafusion/physical-expr/src/expressions/binary/decimal_div.rs b/datafusion/physical-expr/src/expressions/binary/decimal_div.rs new file mode 100644 index 0000000000000..3b95e6fe38ef7 --- /dev/null +++ b/datafusion/physical-expr/src/expressions/binary/decimal_div.rs @@ -0,0 +1,248 @@ +// 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. + +//! Decimal division that rounds half away from zero (Snowflake semantics). +//! +//! Arrow's `div` kernel truncates the decimal quotient toward zero, while +//! Snowflake rounds the last digit half away from zero (verified live: +//! `5/3 = 1.666667`, `-5/3 = -1.666667`, `5/2000000 = 0.000003`). This kernel +//! replicates arrow's decimal division exactly — the same result +//! precision/scale derivation (`result_scale = min(s1 + 4, MAX_SCALE)`), the +//! same overflow and divide-by-zero errors — but corrects the truncated +//! quotient using the remainder. + +use std::cmp::Ordering; +use std::sync::Arc; + +use arrow::array::{Array, ArrayRef, AsArray, Datum, PrimitiveArray}; +use arrow::compute::kernels::arity::try_binary; +use arrow::datatypes::{ + ArrowNativeType, ArrowNativeTypeOp, DataType, Decimal32Type, Decimal64Type, + Decimal128Type, Decimal256Type, DecimalType, +}; +use arrow::error::ArrowError; + +/// Whether `lhs / rhs` is a decimal division this kernel handles (same-width +/// decimal operands, the shape produced by DataFusion's binary type coercion). +pub(super) fn is_decimal_division(lhs: &DataType, rhs: &DataType) -> bool { + matches!( + (lhs, rhs), + (DataType::Decimal32(..), DataType::Decimal32(..)) + | (DataType::Decimal64(..), DataType::Decimal64(..)) + | (DataType::Decimal128(..), DataType::Decimal128(..)) + | (DataType::Decimal256(..), DataType::Decimal256(..)) + ) +} + +/// Divide two decimal [`Datum`], rounding the quotient half away from zero. +/// +/// Drop-in replacement for arrow's `div` on decimal inputs: identical result +/// type derivation and error behavior, only the final-digit rounding differs. +pub(super) fn div_half_away_from_zero( + lhs: &dyn Datum, + rhs: &dyn Datum, +) -> Result { + let (l, l_s) = lhs.get(); + let (r, r_s) = rhs.get(); + match (l.data_type(), r.data_type()) { + (DataType::Decimal32(..), DataType::Decimal32(..)) => { + decimal_div_op::(l, l_s, r, r_s) + } + (DataType::Decimal64(..), DataType::Decimal64(..)) => { + decimal_div_op::(l, l_s, r, r_s) + } + (DataType::Decimal128(..), DataType::Decimal128(..)) => { + decimal_div_op::(l, l_s, r, r_s) + } + (DataType::Decimal256(..), DataType::Decimal256(..)) => { + decimal_div_op::(l, l_s, r, r_s) + } + (l_t, r_t) => Err(ArrowError::InvalidArgumentError(format!( + "Invalid decimal division: {l_t} / {r_t}" + ))), + } +} + +fn decimal_div_op( + l: &dyn Array, + l_s: bool, + r: &dyn Array, + r_s: bool, +) -> Result { + let l = l.as_primitive::(); + let r = r.as_primitive::(); + + let (p1, s1, s2) = match (l.data_type(), r.data_type()) { + (DataType::Decimal32(p1, s1), DataType::Decimal32(_, s2)) + | (DataType::Decimal64(p1, s1), DataType::Decimal64(_, s2)) + | (DataType::Decimal128(p1, s1), DataType::Decimal128(_, s2)) + | (DataType::Decimal256(p1, s1), DataType::Decimal256(_, s2)) => (*p1, *s1, *s2), + _ => unreachable!("callers dispatch on matching decimal types"), + }; + + // Result type derivation identical to arrow-arith's `decimal_op` Op::Div. + let result_scale = s1.saturating_add(4).min(T::MAX_SCALE); + let mul_pow = result_scale - s1 + s2; + let result_precision = (mul_pow.saturating_add(p1 as i8) as u8).min(T::MAX_PRECISION); + + let (l_mul, r_mul) = match mul_pow.cmp(&0) { + Ordering::Greater => ( + T::Native::usize_as(10).pow_checked(mul_pow as _)?, + T::Native::ONE, + ), + Ordering::Equal => (T::Native::ONE, T::Native::ONE), + Ordering::Less => ( + T::Native::ONE, + T::Native::usize_as(10).pow_checked(mul_pow.neg_wrapping() as _)?, + ), + }; + + let op = |l: T::Native, r: T::Native| -> Result { + let ln = l.mul_checked(l_mul)?; + let rn = r.mul_checked(r_mul)?; + let mut quotient = ln.div_checked(rn)?; + // Round half away from zero: if twice the remainder reaches the + // divisor, bump the quotient one step away from zero. + // `abs_rem >= abs_rn - abs_rem` avoids overflowing 2*rem. + let rem = ln.mod_wrapping(rn); + if !rem.is_zero() { + let abs_rem = if rem.is_lt(T::Native::ZERO) { + rem.neg_wrapping() + } else { + rem + }; + let abs_rn = if rn.is_lt(T::Native::ZERO) { + rn.neg_wrapping() + } else { + rn + }; + if abs_rem.is_ge(abs_rn.sub_wrapping(abs_rem)) { + let negative = ln.is_lt(T::Native::ZERO) != rn.is_lt(T::Native::ZERO); + quotient = if negative { + quotient.sub_checked(T::Native::ONE)? + } else { + quotient.add_checked(T::Native::ONE)? + }; + } + } + Ok(quotient) + }; + + // Scalar/array dispatch mirroring arrow-arith's `try_op!` macro. + let array: PrimitiveArray = match (l_s, r_s) { + (true, true) | (false, false) => try_binary(l, r, op)?, + (true, false) => match (l.null_count() == 0).then(|| l.value(0)) { + None => PrimitiveArray::new_null(r.len()), + Some(lv) => r.try_unary(|rv| op(lv, rv))?, + }, + (false, true) => match (r.null_count() == 0).then(|| r.value(0)) { + None => PrimitiveArray::new_null(l.len()), + Some(rv) => l.try_unary(|lv| op(lv, rv))?, + }, + }; + + Ok(Arc::new(array.with_precision_and_scale( + result_precision, + result_scale, + )?)) +} + +#[cfg(test)] +mod tests { + use super::*; + use arrow::array::{Decimal128Array, Scalar}; + + fn dec128(values: Vec>, precision: u8, scale: i8) -> Decimal128Array { + Decimal128Array::from(values) + .with_precision_and_scale(precision, scale) + .unwrap() + } + + fn values(result: &ArrayRef) -> (Vec>, DataType) { + let arr = result.as_primitive::(); + (arr.iter().collect(), arr.data_type().clone()) + } + + #[test] + fn rounds_half_away_from_zero_array_array() { + // NUMBER(2,0) operands -> arrow result scale 0 + 4. + // 5/3 = 1.666666... -> 1.6667 (16667, not the truncated 16666) + // -5/3 -> -1.6667; 1/8 = 0.125 exact at scale 4 -> 1250; + // 5/4 = 1.25 exact -> 12500; 1/16 = 0.0625 -> rounds to 0.0625 (exact). + let l = dec128(vec![Some(5), Some(-5), Some(1), Some(5), None], 2, 0); + let r = dec128(vec![Some(3), Some(3), Some(8), Some(4), Some(2)], 2, 0); + let result = div_half_away_from_zero(&l, &r).unwrap(); + let (vals, dt) = values(&result); + assert_eq!(dt, DataType::Decimal128(6, 4)); + assert_eq!( + vals, + vec![Some(16667), Some(-16667), Some(1250), Some(12500), None] + ); + } + + #[test] + fn rounds_exact_half_away_from_zero_not_even() { + // 1/32 = 0.03125 exactly: the digit past result scale 4 is an exact + // half. Half-away-from-zero (Snowflake, verified live) gives 0.0313; + // truncation and half-even would both give 0.0312, so this pins the + // rounding mode, both signs. + let l = dec128(vec![Some(1), Some(-1)], 2, 0); + let r = dec128(vec![Some(32), Some(32)], 2, 0); + let result = div_half_away_from_zero(&l, &r).unwrap(); + let (vals, dt) = values(&result); + assert_eq!(dt, DataType::Decimal128(6, 4)); + assert_eq!(vals, vec![Some(313), Some(-313)]); + } + + #[test] + fn scalar_array_combos_match() { + let l = dec128(vec![Some(5)], 2, 0); + let r = dec128(vec![Some(3), Some(-3), None], 2, 0); + let scalar_l = Scalar::new(l); + let result = div_half_away_from_zero(&scalar_l, &r).unwrap(); + let (vals, _) = values(&result); + assert_eq!(vals, vec![Some(16667), Some(-16667), None]); + + let l = dec128(vec![Some(5), Some(-5), None], 2, 0); + let r = dec128(vec![Some(3)], 2, 0); + let scalar_r = Scalar::new(r); + let result = div_half_away_from_zero(&l, &scalar_r).unwrap(); + let (vals, _) = values(&result); + assert_eq!(vals, vec![Some(16667), Some(-16667), None]); + } + + #[test] + fn divide_by_zero_errors() { + let l = dec128(vec![Some(5)], 2, 0); + let r = dec128(vec![Some(0)], 2, 0); + let err = div_half_away_from_zero(&l, &r).unwrap_err(); + assert!( + err.to_string().contains("Divide by zero"), + "expected divide-by-zero, got: {err}" + ); + } + + #[test] + fn scale_is_capped_like_arrow() { + // s1 = 36 -> result scale capped at MAX_SCALE (38), matching arrow. + let l = dec128(vec![Some(10_i128.pow(36))], 38, 36); + let r = dec128(vec![Some(3)], 2, 0); + let result = div_half_away_from_zero(&l, &r).unwrap(); + let (_, dt) = values(&result); + assert_eq!(dt, DataType::Decimal128(38, 38)); + } +} diff --git a/datafusion/sqllogictest/test_files/decimal.slt b/datafusion/sqllogictest/test_files/decimal.slt index eca2c88bb5f89..2df44594166f9 100644 --- a/datafusion/sqllogictest/test_files/decimal.slt +++ b/datafusion/sqllogictest/test_files/decimal.slt @@ -121,7 +121,7 @@ Decimal128(20, 6) 0.00055 query TR select arrow_typeof(avg(c1)), avg(c1) from decimal_simple; ---- -Decimal128(14, 10) 0.0000366666 +Decimal128(14, 10) 0.0000366667 query TR @@ -399,19 +399,19 @@ select c1/c5 from decimal_simple; ---- 0.5 0.641025641 -0.7142857142 +0.7142857143 0.7352941176 0.8 0.8571428571 -0.909090909 -0.909090909 +0.9090909091 +0.9090909091 0.9375 0.9615384615 1 1 1.0526315789 -1.5151515151 -2.7272727272 +1.5151515152 +2.7272727273 query T @@ -678,7 +678,7 @@ select * from decimal256_simple where c1 > c5; query TR select arrow_typeof(avg(c1)), avg(c1) from decimal256_simple; ---- -Decimal256(54, 10) 0.0000366666 +Decimal256(54, 10) 0.0000366667 query TR select arrow_typeof(min(c1)), min(c1) from decimal256_simple where c4=false; From a8f2dfebd7b0dfb2fc6c35fbef94f7b5aeefe35e Mon Sep 17 00:00:00 2001 From: Denys Tsomenko Date: Thu, 2 Jul 2026 20:27:30 +0300 Subject: [PATCH 2/3] Point sqlparser at the merged sync-branch head (01e9e5e) The Snowflake string-escape fix was squash-merged into embucket-sync-df53.0.0-parser0.61.0 as 01e9e5e (content-identical to the previously pinned 074bd21 feature-branch rev). Co-Authored-By: Claude Opus 4.8 (1M context) --- Cargo.lock | 14 +++++++------- Cargo.toml | 2 +- 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index e6a2b3985e6e6..db8b62424e145 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3101,7 +3101,7 @@ checksum = "0ce92ff622d6dadf7349484f42c93271a0d49b7cc4d466a936405bacbe10aa78" dependencies = [ "cfg-if", "rustix 1.0.8", - "windows-sys 0.59.0", + "windows-sys 0.52.0", ] [[package]] @@ -5147,7 +5147,7 @@ dependencies = [ "once_cell", "socket2 0.5.10", "tracing", - "windows-sys 0.59.0", + "windows-sys 0.52.0", ] [[package]] @@ -5543,7 +5543,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.4.15", - "windows-sys 0.59.0", + "windows-sys 0.52.0", ] [[package]] @@ -6063,7 +6063,7 @@ dependencies = [ [[package]] name = "sqlparser" version = "0.61.0" -source = "git+https://github.com/Embucket/datafusion-sqlparser-rs.git?rev=074bd2124c2e55b43528e97890937c2698d100ae#074bd2124c2e55b43528e97890937c2698d100ae" +source = "git+https://github.com/Embucket/datafusion-sqlparser-rs.git?rev=01e9e5e11505fceaa1ca93f3258d2389527320ea#01e9e5e11505fceaa1ca93f3258d2389527320ea" dependencies = [ "log", "recursive", @@ -6073,7 +6073,7 @@ dependencies = [ [[package]] name = "sqlparser_derive" version = "0.5.0" -source = "git+https://github.com/Embucket/datafusion-sqlparser-rs.git?rev=074bd2124c2e55b43528e97890937c2698d100ae#074bd2124c2e55b43528e97890937c2698d100ae" +source = "git+https://github.com/Embucket/datafusion-sqlparser-rs.git?rev=01e9e5e11505fceaa1ca93f3258d2389527320ea#01e9e5e11505fceaa1ca93f3258d2389527320ea" dependencies = [ "proc-macro2", "quote", @@ -6096,7 +6096,7 @@ dependencies = [ "cfg-if", "libc", "psm", - "windows-sys 0.59.0", + "windows-sys 0.52.0", ] [[package]] @@ -7233,7 +7233,7 @@ version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cf221c93e13a30d793f7645a0e7762c55d169dbb0a49671918a2319d289b10bb" dependencies = [ - "windows-sys 0.59.0", + "windows-sys 0.52.0", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index bf6510839772e..c4b906cb215cb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -184,7 +184,7 @@ regex = "1.12" rstest = "0.26.1" serde_json = "1" sha2 = "^0.10.9" -sqlparser = { git = "https://github.com/Embucket/datafusion-sqlparser-rs.git", rev = "074bd2124c2e55b43528e97890937c2698d100ae", features = [ +sqlparser = { git = "https://github.com/Embucket/datafusion-sqlparser-rs.git", rev = "01e9e5e11505fceaa1ca93f3258d2389527320ea", features = [ "visitor", ] } strum = "0.28.0" From 4b9bebc98db92df035a8c1a56a57df539eb2dc66 Mon Sep 17 00:00:00 2001 From: Denys Tsomenko Date: Thu, 2 Jul 2026 21:11:48 +0300 Subject: [PATCH 3/3] Fix pre-existing sqllogictest failures: to_date(double), quoted PIVOT column, stale EXPLAIN plans MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - to_date: Float64 sat in the direct-cast arm, but arrow has no Float64 -> Date32 cast; route it through the Int64 two-step like Float16/Float32/Decimal (dates.slt:216 now passes, truncation semantics identical to the other float widths). - PIVOT: the value column was rendered with Expr::to_string, which keeps quote characters, so a quoted pivot column (FOR "2023_Q1" IN ...) never matched its schema field and was not consumed by the pivot. Normalize the identifier like the rest of the planner (pivot.slt nested-PIVOT case now passes). - subquery.slt: regenerate the two EXPLAIN expectations — the plans changed upstream (CoalesceBatchesExec deprecated in #19622, LeftMark -> RightMark swap added in #17651) and were never regenerated during the df53 sync; physical-plan shape only, logical plans unchanged. encrypted_parquet.slt still fails locally only (expects a no-keys read to error; environment-dependent, fails identically on the clean base, passes in CI). Co-Authored-By: Claude Opus 4.8 (1M context) --- datafusion/functions/src/datetime/to_date.rs | 6 +- datafusion/sql/src/relation/mod.rs | 15 ++++- .../sqllogictest/test_files/subquery.slt | 62 +++++++------------ 3 files changed, 42 insertions(+), 41 deletions(-) diff --git a/datafusion/functions/src/datetime/to_date.rs b/datafusion/functions/src/datetime/to_date.rs index f1a1f0d8fbbd9..a3d8f8b1405cc 100644 --- a/datafusion/functions/src/datetime/to_date.rs +++ b/datafusion/functions/src/datetime/to_date.rs @@ -147,7 +147,7 @@ impl ScalarUDFImpl for ToDateFunc { validate_data_types(&args, "to_date")?; } match args[0].data_type() { - Null | Int32 | Int64 | Float64 | Date32 | Date64 | Timestamp(_, _) => { + Null | Int32 | Int64 | Date32 | Date64 | Timestamp(_, _) => { args[0].cast_to(&Date32, None) } UInt8 | UInt16 | UInt32 | UInt64 | Int8 | Int16 => { @@ -174,12 +174,14 @@ impl ScalarUDFImpl for ToDateFunc { } Float16 | Float32 + | Float64 | Decimal32(_, _) | Decimal64(_, _) | Decimal128(_, _) | Decimal256(_, _) => { // The only way this makes sense is to get the Int64 value of the float - // or decimal and then cast that to Date32. + // or decimal and then cast that to Date32. (Arrow has no direct + // Float64 -> Date32 cast either, so Float64 takes this path too.) args[0].cast_to(&Int64, None)?.cast_to(&Date32, None) } Utf8View | LargeUtf8 | Utf8 => self.to_date(&args), diff --git a/datafusion/sql/src/relation/mod.rs b/datafusion/sql/src/relation/mod.rs index c891dd73534fc..367d5bdc8bfba 100644 --- a/datafusion/sql/src/relation/mod.rs +++ b/datafusion/sql/src/relation/mod.rs @@ -338,7 +338,20 @@ impl SqlToRel<'_, S> { return plan_err!("PIVOT value column is required"); } - let column_name = value_column.last().unwrap().to_string(); + // Normalize like every other identifier: rendering the expression to a + // string keeps the quote characters, so a quoted pivot column + // (`FOR "2023_Q1" IN ...`) would never match its schema field. + let column_name = match value_column.last().unwrap() { + sqlparser::ast::Expr::Identifier(ident) => { + self.ident_normalizer.normalize(ident.clone()) + } + sqlparser::ast::Expr::CompoundIdentifier(idents) => self + .ident_normalizer + .normalize(idents.last().unwrap().clone()), + other => { + return plan_err!("Unsupported PIVOT value column: {other}"); + } + }; let pivot_column = Column::new(None::<&str>, column_name); let default_on_null_expr = default_on_null diff --git a/datafusion/sqllogictest/test_files/subquery.slt b/datafusion/sqllogictest/test_files/subquery.slt index bd6b77a12b5e7..bb4744a46080e 100644 --- a/datafusion/sqllogictest/test_files/subquery.slt +++ b/datafusion/sqllogictest/test_files/subquery.slt @@ -355,20 +355,18 @@ logical_plan 11)--SubqueryAlias: o 12)----TableScan: t2 projection=[t2_id] physical_plan -01)CoalesceBatchesExec: target_batch_size=2 -02)--HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(t1_id@0, t2_id@0)] -03)----CoalescePartitionsExec -04)------CoalesceBatchesExec: target_batch_size=2 -05)--------FilterExec: NOT mark@1, projection=[t1_id@0] -06)----------CoalesceBatchesExec: target_batch_size=2 -07)------------HashJoinExec: mode=CollectLeft, join_type=LeftMark, on=[(t1_id@0, t3_id@0)] +01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(t1_id@0, t2_id@0)] +02)--CoalescePartitionsExec +03)----FilterExec: NOT mark@1, projection=[t1_id@0] +04)------HashJoinExec: mode=CollectLeft, join_type=RightMark, on=[(t3_id@0, t1_id@0)] +05)--------CoalescePartitionsExec +06)----------FilterExec: t3_int@1 = 3, projection=[t3_id@0] +07)------------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1 08)--------------DataSourceExec: partitions=1, partition_sizes=[2] -09)--------------CoalesceBatchesExec: target_batch_size=2 -10)----------------FilterExec: t3_int@1 = 3, projection=[t3_id@0] -11)------------------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1 -12)--------------------DataSourceExec: partitions=1, partition_sizes=[2] -13)----RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1 -14)------DataSourceExec: partitions=1, partition_sizes=[2] +09)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1 +10)----------DataSourceExec: partitions=1, partition_sizes=[2] +11)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1 +12)----DataSourceExec: partitions=1, partition_sizes=[2] # aggregate_case_in_subquery query III rowsort @@ -437,31 +435,19 @@ logical_plan physical_plan 01)ProjectionExec: expr=[namespace_id@0 as primary_key, max(CASE WHEN IN THEN Int64(1) ELSE Int64(0) END)@1 as is_in_most_recent_task, max(distinct_source.max_uploaded_at)@2 as max_timestamp] 02)--AggregateExec: mode=FinalPartitioned, gby=[namespace_id@0 as namespace_id], aggr=[max(CASE WHEN IN THEN Int64(1) ELSE Int64(0) END), max(distinct_source.max_uploaded_at)] -03)----CoalesceBatchesExec: target_batch_size=2 -04)------RepartitionExec: partitioning=Hash([namespace_id@0], 4), input_partitions=4 -05)--------AggregateExec: mode=Partial, gby=[namespace_id@0 as namespace_id], aggr=[max(CASE WHEN IN THEN Int64(1) ELSE Int64(0) END), max(distinct_source.max_uploaded_at)] -06)----------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1 -07)------------CoalesceBatchesExec: target_batch_size=2 -08)--------------HashJoinExec: mode=CollectLeft, join_type=LeftMark, on=[(max_task_instance@1, max(distinct_source.max_task_instance)@0)], projection=[namespace_id@0, max_uploaded_at@2, mark@3] -09)----------------CoalescePartitionsExec -10)------------------UnionExec -11)--------------------ProjectionExec: expr=[1 as namespace_id, 10 as max_task_instance, 100 as max_uploaded_at] -12)----------------------PlaceholderRowExec -13)--------------------ProjectionExec: expr=[2 as namespace_id, 20 as max_task_instance, 200 as max_uploaded_at] -14)----------------------PlaceholderRowExec -15)--------------------ProjectionExec: expr=[2 as namespace_id, 15 as max_task_instance, 150 as max_uploaded_at] -16)----------------------PlaceholderRowExec -17)----------------AggregateExec: mode=Final, gby=[], aggr=[max(distinct_source.max_task_instance)] -18)------------------CoalescePartitionsExec -19)--------------------AggregateExec: mode=Partial, gby=[], aggr=[max(distinct_source.max_task_instance)] -20)----------------------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=3 -21)------------------------UnionExec -22)--------------------------ProjectionExec: expr=[10 as max_task_instance] -23)----------------------------PlaceholderRowExec -24)--------------------------ProjectionExec: expr=[20 as max_task_instance] -25)----------------------------PlaceholderRowExec -26)--------------------------ProjectionExec: expr=[15 as max_task_instance] -27)----------------------------PlaceholderRowExec +03)----RepartitionExec: partitioning=Hash([namespace_id@0], 4), input_partitions=4 +04)------AggregateExec: mode=Partial, gby=[namespace_id@0 as namespace_id], aggr=[max(CASE WHEN IN THEN Int64(1) ELSE Int64(0) END), max(distinct_source.max_uploaded_at)] +05)--------HashJoinExec: mode=CollectLeft, join_type=RightMark, on=[(max(distinct_source.max_task_instance)@0, max_task_instance@1)], projection=[namespace_id@0, max_uploaded_at@2, mark@3] +06)----------ProjectionExec: expr=[20 as max(distinct_source.max_task_instance)] +07)------------PlaceholderRowExec +08)----------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=3 +09)------------UnionExec +10)--------------ProjectionExec: expr=[1 as namespace_id, 10 as max_task_instance, 100 as max_uploaded_at] +11)----------------PlaceholderRowExec +12)--------------ProjectionExec: expr=[2 as namespace_id, 20 as max_task_instance, 200 as max_uploaded_at] +13)----------------PlaceholderRowExec +14)--------------ProjectionExec: expr=[2 as namespace_id, 15 as max_task_instance, 150 as max_uploaded_at] +15)----------------PlaceholderRowExec query II rowsort SELECT t1_id, (SELECT sum(t2_int) FROM t2 WHERE t2.t2_id = t1.t1_id having sum(t2_int) < 3) as t2_sum from t1