comphead commented on code in PR #6732:
URL: https://github.com/apache/datafusion-comet/pull/6732#discussion_r4231746193
##########
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};
Review Comment:
#6069 is open and approved, and it adds a fourth ANSI site in this file:
`SlidingSumIntegerAccumulator::evaluate` returns
`arithmetic_overflow_error("integer")`. Git merges the two PRs without a
conflict, but this line drops that import, so from reading the merged tree I
expect whichever lands second not to compile. That sliding frame would also
keep reporting `integer overflow` with no `try_add`, because Spark recomputes
each sliding frame through the same `Add`. Could that site use
`sum_overflow_error()` too, in whichever PR lands second? I have not built the
merge.
##########
spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala:
##########
@@ -3297,20 +3297,38 @@ 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 = {
+ val (sparkError, cometError) = checkSparkAnswerMaybeThrows(df)
Review Comment:
In ANSI mode nothing here checks that Comet ran the aggregate natively. From
reading the code, if `SUM` fell back to Spark, this would compare Spark's error
with Spark's and pass. `checkSparkError` in `CometTestBase` starts with
`checkCometOperators(stripAQEPlan(df.queryExecution.executedPlan))`, and the
`ANSI integral overflow fidelity` tests in `CometExpressionSuite` make the same
check before an identical comparison. Could the helper start with that line?
Another option is a flag on `checkSparkError` that also compares
`getMessageParameters`, which would make this helper unnecessary.
##########
spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala:
##########
@@ -3319,16 +3337,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")) {
Review Comment:
The `GROUP BY` test further down (around line 3420) still only checks that
both messages contain `ARITHMETIC_OVERFLOW`, so its grouped overflow and
underflow cases pass with or without this fix. Could its two ANSI branches call
`assertAnsiSumOverflowMatchesSpark` as well? That would also compare a grouped
underflow with Spark, which this block does not cover because it only overflows
upward.
##########
native/spark-expr/src/agg_funcs/sum_int.rs:
##########
@@ -1017,4 +1028,44 @@ 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"})
+ );
+ assert_eq!(
+ error.to_string(),
+ "[ARITHMETIC_OVERFLOW] long overflow. Use 'try_add' to
tolerate overflow and \
+ return NULL instead. If necessary set
\"spark.sql.ansi.enabled\" to \"false\" \
+ to bypass this error."
+ );
Review Comment:
This `to_string()` check repeats
`test_arithmetic_overflow_suggestion_json_and_display` in
`native/common/src/error.rs`, which already pins the exact text for `long` with
`try_add`. The `params` check above is what shows the accumulators raise the
right error. Could the full-message assertion go, so a wording change in
`SparkError` only touches one test?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
