You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by ag...@apache.org on 2026/05/08 20:42:19 UTC
(datafusion-comet) branch main updated: fix(spark-expr): preserve scalar tag in WideDecimalBinaryExpr when both inputs are scalars (#4187)
This is an automated email from the ASF dual-hosted git repository.
andygrove pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
The following commit(s) were added to refs/heads/main by this push:
new 03e833b95 fix(spark-expr): preserve scalar tag in WideDecimalBinaryExpr when both inputs are scalars (#4187)
03e833b95 is described below
commit 03e833b955d369f994d9652026ca3c1eb641acac
Author: Tom Zeng <to...@gmail.com>
AuthorDate: Fri May 8 13:42:15 2026 -0700
fix(spark-expr): preserve scalar tag in WideDecimalBinaryExpr when both inputs are scalars (#4187)
---
.../src/math_funcs/wide_decimal_binary_expr.rs | 84 +++++++++++++++++++++-
1 file changed, 83 insertions(+), 1 deletion(-)
diff --git a/native/spark-expr/src/math_funcs/wide_decimal_binary_expr.rs b/native/spark-expr/src/math_funcs/wide_decimal_binary_expr.rs
index 644252b46..7ff78713b 100644
--- a/native/spark-expr/src/math_funcs/wide_decimal_binary_expr.rs
+++ b/native/spark-expr/src/math_funcs/wide_decimal_binary_expr.rs
@@ -184,6 +184,17 @@ impl PhysicalExpr for WideDecimalBinaryExpr {
let left_val = self.left.evaluate(batch)?;
let right_val = self.right.evaluate(batch)?;
+ // Track scalar-ness so we can return a Scalar when both inputs are scalars.
+ // Without this, a (Scalar op Scalar) result would be returned as a length-1
+ // Array, and downstream comparisons against full batches would incorrectly
+ // see two Array operands with mismatched lengths instead of (Array, Scalar),
+ // crashing arrow-ord's compare_op with "Cannot compare arrays of different
+ // lengths". This pattern appears, for example, in TPC-DS q23's BHJ filter
+ // `0.95 * scalar_subquery > ssales`.
+ let both_scalar = matches!(
+ (&left_val, &right_val),
+ (ColumnarValue::Scalar(_), ColumnarValue::Scalar(_))
+ );
let (left_arr, right_arr): (ArrayRef, ArrayRef) = match (&left_val, &right_val) {
(ColumnarValue::Array(l), ColumnarValue::Array(r)) => (Arc::clone(l), Arc::clone(r)),
(ColumnarValue::Scalar(l), ColumnarValue::Array(r)) => {
@@ -280,7 +291,16 @@ impl PhysicalExpr for WideDecimalBinaryExpr {
result
};
let result = result.with_data_type(DataType::Decimal128(p_out, s_out));
- Ok(ColumnarValue::Array(Arc::new(result)))
+ if both_scalar {
+ // Convert the length-1 result back into a Scalar so downstream
+ // expressions (binary ops, comparisons) can take the scalar fast-path
+ // and propagate the scalar tag (Datum::is_scalar) through arrow-rs
+ // kernels.
+ let scalar = datafusion::common::ScalarValue::try_from_array(&result, 0)?;
+ Ok(ColumnarValue::Scalar(scalar))
+ } else {
+ Ok(ColumnarValue::Array(Arc::new(result)))
+ }
}
fn children(&self) -> Vec<&Arc<dyn PhysicalExpr>> {
@@ -557,4 +577,66 @@ mod tests {
let arr = result.as_primitive::<Decimal128Type>();
assert_eq!(arr.value(0), 20000); // 2.0000
}
+
+ /// Regression test for the Scalar x Scalar wide-decimal evaluation path.
+ ///
+ /// When both inputs are `ColumnarValue::Scalar`, `evaluate` must return a
+ /// `ColumnarValue::Scalar` -- not a length-1 `ColumnarValue::Array`. Otherwise
+ /// downstream comparisons against full batches see two `Array` operands with
+ /// mismatched lengths and arrow-ord's `compare_op` rejects them with
+ /// "Cannot compare arrays of different lengths, got N vs 1". This pattern
+ /// appears, for example, in TPC-DS q23's BHJ filter
+ /// `0.95 * scalar_subquery > ssales`.
+ #[test]
+ fn test_scalar_scalar_returns_scalar() {
+ use datafusion::common::ScalarValue;
+ use datafusion::physical_expr::expressions::Literal;
+
+ // 0.95 * 100.00 -- the same Scalar x Scalar decimal multiply pattern that
+ // appears in TPC-DS q23's filter `0.95 * scalar_subquery > ssales`.
+ let left: Arc<dyn PhysicalExpr> =
+ Arc::new(Literal::new(ScalarValue::Decimal128(Some(95), 38, 2)));
+ let right: Arc<dyn PhysicalExpr> =
+ Arc::new(Literal::new(ScalarValue::Decimal128(Some(10000), 38, 2)));
+
+ let expr = WideDecimalBinaryExpr::new(
+ left,
+ right,
+ WideDecimalOp::Multiply,
+ 38,
+ 2,
+ EvalMode::Legacy,
+ );
+
+ // Empty schema -- both inputs are Literals so no columns are needed.
+ let batch = RecordBatch::new_empty(Arc::new(Schema::empty()));
+ match expr.evaluate(&batch).unwrap() {
+ ColumnarValue::Scalar(ScalarValue::Decimal128(Some(v), 38, 2)) => {
+ // 0.95 * 100.00 = 95.00 -> at scale 2, integer 9500
+ assert_eq!(v, 9500);
+ }
+ ColumnarValue::Scalar(other) => {
+ panic!("expected Decimal128(Some(_), 38, 2), got {other:?}");
+ }
+ ColumnarValue::Array(_) => {
+ panic!("Scalar x Scalar must return ColumnarValue::Scalar, not Array");
+ }
+ }
+ }
+
+ /// Companion test: when at least one input is an Array, the result must remain an Array.
+ /// Guards against over-eager scalar-unwrapping in the fix.
+ #[test]
+ fn test_array_input_returns_array() {
+ let batch = make_batch(
+ vec![Some(150), Some(250)],
+ 38,
+ 2,
+ vec![Some(100), Some(200)],
+ 38,
+ 2,
+ );
+ let result = eval_expr(&batch, WideDecimalOp::Add, 38, 2, EvalMode::Legacy).unwrap();
+ assert_eq!(result.len(), 2);
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org