You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/16 13:57:01 UTC
(datafusion-comet) branch gh-readonly-queue/main/pr-6732-3504a2b6d514ae928ba39862bff30aa47b1cf92a created (now 49883ccc7a)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-6732-3504a2b6d514ae928ba39862bff30aa47b1cf92a
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
at 49883ccc7a fix: include try_add suggestion in ANSI SUM overflow error (#6732)
This branch includes the following new commits:
new 49883ccc7a fix: include try_add suggestion in ANSI SUM overflow error (#6732)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion-comet) 01/01: fix: include try_add suggestion in ANSI SUM overflow error (#6732)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-6732-3504a2b6d514ae928ba39862bff30aa47b1cf92a
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 49883ccc7a4074c45006df6e63a9fdd6e768b086
Author: DeviousCardi <11...@users.noreply.github.com>
AuthorDate: Fri Oct 16 13:56:39 2026 +0000
fix: include try_add suggestion in ANSI SUM overflow error (#6732)
* fix: include try_add suggestion in ANSI SUM overflow error
Spark's integral SUM returns LONG and adds through Add in both its
update and merge expressions, so an ANSI overflow is reported as
"long overflow" with the "Use 'try_add' ..." suggestion. Comet's
SumInteger raised "integer overflow" with no suggestion at its three
overflow sites (ungrouped update/merge, grouped update, grouped merge).
Raise ArithmeticOverflow { from_type: "long", function_name: "try_add" }
at all three, and make the CometAggregateSuite SUM test compare the
structured error with Spark's, including the merge path.
Closes #6661
* test: drop literal 'long overflow' check from ANSI SUM overflow test
Spark 4.2 normalizes the 'long overflow' message to 'overflow' in
ExecutionErrors.arithmeticOverflowError, so the literal check failed on
the spark-4.2 profile even though Comet's error matches Spark's. The
getMessageParameters comparison with Spark already covers the wording.
Co-Authored-By: Claude Opus 5.5 (1M context) <no...@anthropic.com>
* test: check native execution in ANSI SUM overflow tests
- assertAnsiSumOverflowMatchesSpark now checks the plan ran on Comet
operators first, so a SUM that fell back to Spark can't pass by
comparing Spark's error with itself.
- The GROUP BY test's ANSI overflow and underflow branches use the
helper, so they compare the error parameters with Spark's instead of
only checking for ARITHMETIC_OVERFLOW in both messages.
- Drop the Rust full-message assertion; native/common/src/error.rs
already pins that text, and the params check covers the accumulators.
Co-Authored-By: Claude Opus 5.5 (1M context) <no...@anthropic.com>
---------
Co-authored-by: Claude Opus 5.5 (1M context) <no...@anthropic.com>
---
native/spark-expr/src/agg_funcs/sum_int.rs | 63 +++++++++++++---
.../apache/comet/exec/CometAggregateSuite.scala | 88 +++++++++++++++-------
2 files changed, 115 insertions(+), 36 deletions(-)
diff --git a/native/spark-expr/src/agg_funcs/sum_int.rs b/native/spark-expr/src/agg_funcs/sum_int.rs
index 56a8c122c4..d3e367af72 100644
--- a/native/spark-expr/src/agg_funcs/sum_int.rs
+++ b/native/spark-expr/src/agg_funcs/sum_int.rs
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-use crate::{arithmetic_overflow_error, EvalMode};
+use crate::{EvalMode, SparkError};
use arrow::array::{
as_primitive_array, cast::AsArray, Array, ArrayRef, ArrowNativeTypeOp, ArrowPrimitiveType,
BooleanArray, Int64Array, PrimitiveArray,
@@ -31,6 +31,17 @@ use datafusion::logical_expr::{
};
use std::sync::Arc;
+/// Spark's integral `SUM` always returns `LONG` and adds through `Add` in both its update and
+/// merge expressions, so an ANSI overflow is reported by `MathUtils.addExact(Long, Long)` as a
+/// `long` overflow carrying the `try_add` suggestion.
+fn sum_overflow_error() -> DataFusionError {
+ SparkError::ArithmeticOverflow {
+ from_type: "long".to_string(),
+ function_name: "try_add".to_string(),
+ }
+ .into()
+}
+
#[derive(Debug, PartialEq, Eq, Hash)]
pub struct SumInteger {
signature: Signature,
@@ -199,9 +210,7 @@ impl Accumulator for SumIntegerAccumulatorAnsi {
int_array.value(i)
))
})?;
- sum = v
- .add_checked(sum)
- .map_err(|_| DataFusionError::from(arithmetic_overflow_error("integer")))?;
+ sum = v.add_checked(sum).map_err(|_| sum_overflow_error())?;
}
}
Ok(sum)
@@ -574,10 +583,12 @@ impl GroupsAccumulator for SumIntGroupsAccumulatorAnsi {
let v = int_array.value(i).to_i64().ok_or_else(|| {
DataFusionError::Internal("Failed to convert value to i64".to_string())
})?;
- sums[group_index] =
- Some(sums[group_index].unwrap_or(0).add_checked(v).map_err(|_| {
- DataFusionError::from(arithmetic_overflow_error("integer"))
- })?);
+ sums[group_index] = Some(
+ sums[group_index]
+ .unwrap_or(0)
+ .add_checked(v)
+ .map_err(|_| sum_overflow_error())?,
+ );
}
}
Ok(())
@@ -669,7 +680,7 @@ impl GroupsAccumulator for SumIntGroupsAccumulatorAnsi {
self.sums[group_index]
.unwrap()
.add_checked(that_sum)
- .map_err(|_| DataFusionError::from(arithmetic_overflow_error("integer")))?,
+ .map_err(|_| sum_overflow_error())?,
);
}
}
@@ -1017,4 +1028,38 @@ mod tests {
acc.merge_batch(&[states]).unwrap();
assert_eq!(acc.evaluate().unwrap(), ScalarValue::Int64(Some(60)));
}
+
+ /// Spark reports an integral SUM overflow as `long overflow` with the `try_add` suggestion,
+ /// on the update and merge paths of both the scalar and the grouped ANSI accumulators.
+ #[test]
+ fn test_ansi_overflow_matches_spark_long_add() {
+ fn assert_spark_long_add_overflow(error: DataFusionError) {
+ let DataFusionError::External(error) = error else {
+ panic!("Expected structured Spark error, got {error:?}")
+ };
+ let error = error.downcast_ref::<SparkError>().unwrap();
+ let json: serde_json::Value = serde_json::from_str(&error.to_json()).unwrap();
+ assert_eq!(json["errorClass"], "ARITHMETIC_OVERFLOW");
+ assert_eq!(
+ json["params"],
+ serde_json::json!({"fromType": "long", "functionName": "try_add"})
+ );
+ }
+
+ for (first, second) in [(i64::MAX, 1), (i64::MIN, -1)] {
+ let batch = || -> ArrayRef { Arc::new(Int64Array::from(vec![first, second])) };
+
+ let mut acc = SumIntegerAccumulatorAnsi::new();
+ assert_spark_long_add_overflow(acc.update_batch(&[batch()]).unwrap_err());
+ let mut acc = SumIntegerAccumulatorAnsi::new();
+ assert_spark_long_add_overflow(acc.merge_batch(&[batch()]).unwrap_err());
+
+ let mut acc = SumIntGroupsAccumulatorAnsi::new();
+ assert_spark_long_add_overflow(
+ acc.update_batch(&[batch()], &[0, 0], None, 1).unwrap_err(),
+ );
+ let mut acc = SumIntGroupsAccumulatorAnsi::new();
+ assert_spark_long_add_overflow(acc.merge_batch(&[batch()], &[0, 0], 1).unwrap_err());
+ }
+ }
}
diff --git a/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala
index a9095b913f..24e70b2357 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala
@@ -3945,6 +3945,32 @@ class CometAggregateSuite extends CometTestBase with AdaptiveSparkPlanHelper {
(1 to 50).flatMap(_ => Seq((maxDec38_0, 1)))
}
+ /**
+ * Spark's integral SUM adds through `Add`, so an ANSI overflow carries the `try_add`
+ * suggestion. Compare the structured error, not just its error class. The overflow wording is
+ * left to the message parameters: Spark 4.2 normalizes `long overflow` to `overflow`.
+ */
+ private def assertAnsiSumOverflowMatchesSpark(df: DataFrame): Unit = {
+ // Without this, a SUM that fell back to Spark would compare Spark's error with itself.
+ checkCometOperators(stripAQEPlan(df.queryExecution.executedPlan))
+ val (sparkError, cometError) = checkSparkAnswerMaybeThrows(df)
+ def structured(error: Option[Throwable]): SparkThrowable with Throwable = {
+ val failure = error.getOrElse(fail("Expected SUM overflow in ANSI mode"))
+ causeChain(failure)
+ .collect { case e: SparkThrowable with Throwable => e }
+ .lastOption
+ .getOrElse(fail(s"Expected SparkThrowable: $failure"))
+ }
+ val expected = structured(sparkError)
+ val actual = structured(cometError)
+ assert(expected.getErrorClass == "ARITHMETIC_OVERFLOW")
+ assert(actual.getClass == expected.getClass)
+ assert(actual.getErrorClass == expected.getErrorClass)
+ assert(actual.getSqlState == expected.getSqlState)
+ assert(actual.getMessageParameters == expected.getMessageParameters)
+ assert(actual.getMessage.contains("try_add"))
+ }
+
test("ANSI support - SUM function") {
Seq(true, false).foreach { ansiEnabled =>
withSQLConf(SQLConf.ANSI_ENABLED.key -> ansiEnabled.toString) {
@@ -3952,13 +3978,7 @@ class CometAggregateSuite extends CometTestBase with AdaptiveSparkPlanHelper {
withParquetTable(Seq((Long.MaxValue, 1L), (100L, 1L)), "tbl") {
val res = sql("SELECT SUM(_1) FROM tbl")
if (ansiEnabled) {
- checkSparkAnswerMaybeThrows(res) match {
- case (Some(sparkExc), Some(cometExc)) =>
- // make sure that the error message throws overflow exception only
- assert(sparkExc.getMessage.contains("ARITHMETIC_OVERFLOW"))
- assert(cometExc.getMessage.contains("ARITHMETIC_OVERFLOW"))
- case _ => fail("Exception should be thrown for Long overflow in ANSI mode")
- }
+ assertAnsiSumOverflowMatchesSpark(res)
} else {
checkSparkAnswerAndOperator(res)
}
@@ -3967,16 +3987,42 @@ class CometAggregateSuite extends CometTestBase with AdaptiveSparkPlanHelper {
withParquetTable(Seq((Long.MinValue, 1L), (-100L, 1L)), "tbl") {
val res = sql("SELECT SUM(_1) FROM tbl")
if (ansiEnabled) {
- checkSparkAnswerMaybeThrows(res) match {
- case (Some(sparkExc), Some(cometExc)) =>
- assert(sparkExc.getMessage.contains("ARITHMETIC_OVERFLOW"))
- assert(cometExc.getMessage.contains("ARITHMETIC_OVERFLOW"))
- case _ => fail("Exception should be thrown for Long underflow in ANSI mode")
- }
+ assertAnsiSumOverflowMatchesSpark(res)
} else {
checkSparkAnswerAndOperator(res)
}
}
+ // Overflow only when the partial sums of two scan partitions are merged, ungrouped and
+ // grouped. A large open cost keeps the two files in separate scan partitions.
+ withSQLConf(SQLConf.FILES_OPEN_COST_IN_BYTES.key -> (128L * 1024 * 1024).toString) {
+ withTempView("tbl") {
+ withTempPath { dir =>
+ Seq((Long.MaxValue, 1), (1L, 1)).foreach { row =>
+ spark
+ .createDataFrame(Seq(row))
+ .write
+ .mode("append")
+ .parquet(dir.getCanonicalPath)
+ }
+ spark.read.parquet(dir.getCanonicalPath).createOrReplaceTempView("tbl")
+ // One row per scan partition, so no partial update can overflow: the error
+ // below can only come from merging the two partial sums.
+ val rowsPerPartition =
+ spark.table("tbl").rdd.mapPartitions(rows => Iterator(rows.size)).collect()
+ assert(rowsPerPartition.toSeq == Seq(1, 1), rowsPerPartition.mkString(","))
+ for (query <- Seq(
+ "SELECT SUM(_1) FROM tbl",
+ "SELECT _2, SUM(_1) FROM tbl GROUP BY _2")) {
+ val res = sql(query)
+ if (ansiEnabled) {
+ assertAnsiSumOverflowMatchesSpark(res)
+ } else {
+ checkSparkAnswerAndOperator(res)
+ }
+ }
+ }
+ }
+ }
// Test Int SUM (should not overflow)
withParquetTable(Seq((Int.MaxValue, 1), (Int.MaxValue, 1), (100, 1)), "tbl") {
val res = sql("SELECT SUM(_1) FROM tbl")
@@ -4029,13 +4075,7 @@ class CometAggregateSuite extends CometTestBase with AdaptiveSparkPlanHelper {
"tbl") {
val res = sql("SELECT _2, SUM(_1) FROM tbl GROUP BY _2").repartition(2)
if (ansiEnabled) {
- checkSparkAnswerMaybeThrows(res) match {
- case (Some(sparkExc), Some(cometExc)) =>
- assert(sparkExc.getMessage.contains("ARITHMETIC_OVERFLOW"))
- assert(cometExc.getMessage.contains("ARITHMETIC_OVERFLOW"))
- case _ =>
- fail("Exception should be thrown for Long overflow with GROUP BY in ANSI mode")
- }
+ assertAnsiSumOverflowMatchesSpark(res)
} else {
checkSparkAnswerAndOperator(res)
}
@@ -4046,13 +4086,7 @@ class CometAggregateSuite extends CometTestBase with AdaptiveSparkPlanHelper {
"tbl") {
val res = sql("SELECT _2, SUM(_1) FROM tbl GROUP BY _2")
if (ansiEnabled) {
- checkSparkAnswerMaybeThrows(res) match {
- case (Some(sparkExc), Some(cometExc)) =>
- assert(sparkExc.getMessage.contains("ARITHMETIC_OVERFLOW"))
- assert(cometExc.getMessage.contains("ARITHMETIC_OVERFLOW"))
- case _ =>
- fail("Exception should be thrown for Long underflow with GROUP BY in ANSI mode")
- }
+ assertAnsiSumOverflowMatchesSpark(res)
} else {
checkSparkAnswerAndOperator(res)
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org