DeviousCardi commented on code in PR #6732:
URL: https://github.com/apache/datafusion-comet/pull/6732#discussion_r4233272235
##########
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:
Agreed. I can't change that site here, because it only exists in #6069.
Whichever PR lands second will handle it:
- If #6069 merges first, I'll rebase this one and switch
`SlidingSumIntegerAccumulator::evaluate` to `sum_overflow_error()`, which also
fixes the import.
- If this one merges first, #6069 will need the same change on rebase, and
I'm happy to send it as a follow-up.
##########
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:
Good catch, thanks. The helper now starts with
`checkCometOperators(stripAQEPlan(df.queryExecution.executedPlan))`, so a SUM
that fell back to Spark fails the test instead of passing (61f490f).
##########
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:
Done in 61f490f. Both ANSI branches of the `GROUP BY` test now call
`assertAnsiSumOverflowMatchesSpark`, so the grouped overflow and the grouped
underflow are both compared with Spark's message parameters. All 12
`CometAggregateSuite ANSI support` tests pass locally on the default profile.
##########
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:
Removed in 61f490f. The `params` check stays, and the exact wording is now
pinned only in `native/common/src/error.rs`.
--
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]
