[SPARK-38710][SQL] Use SparkArithmeticException for arithmetic overfl…

…ow runtime errors

### What changes were proposed in this pull request?

On arithmetic overflow runtime errors, Spark should throw SparkArithmeticException instead of `java.lang.ArithmeticException`

### Why are the changes needed?

Use a better error exception type.

### Does this PR introduce _any_ user-facing change?

Yes, trivial change on the exception type: on arithmetic overflow runtime errors, Spark will throw SparkArithmeticException instead of `java.lang.ArithmeticException`

### How was this patch tested?

UT

Closes #36022 from gengliangwang/ArithmeticException.

Authored-by: Gengliang Wang <gengliang@apache.org>
Signed-off-by: Gengliang Wang <gengliang@apache.org>
(cherry picked from commit b70fa24)
Signed-off-by: Gengliang Wang <gengliang@apache.org>

core/src/main/resources/error/error-classes.json

@@ -3,6 +3,10 @@
     "message" : [ "Field name %s is ambiguous and has %s matching fields in the struct." ],
     "sqlState" : "42000"
   },
+  "ARITHMETIC_OVERFLOW" : {
+    "message" : [ "%s.%s If necessary set %s to false (except for ANSI interval type) to bypass this error.%s" ],
+    "sqlState" : "22003"
+  },
   "CANNOT_CAST_DATATYPE" : {
     "message" : [ "Cannot cast %s to %s." ],
     "sqlState" : "22005"

sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala

@@ -446,9 +446,8 @@ object QueryExecutionErrors {
       hint: String = "",
       errorContext: String = ""): ArithmeticException = {
     val alternative = if (hint.nonEmpty) s" To return NULL instead, use '$hint'." else ""
-    new ArithmeticException(s"$message.$alternative If necessary set " +
-      s"${SQLConf.ANSI_ENABLED.key} to false (except for ANSI interval type) to bypass this " +
-      "error." + errorContext)
+    new SparkArithmeticException("ARITHMETIC_OVERFLOW",
+      Array(message, alternative, SQLConf.ANSI_ENABLED.key, errorContext))
   }
 
   def unaryMinusCauseOverflowError(originValue: AnyVal): ArithmeticException = {

sql/core/src/test/resources/sql-tests/results/ansi/interval.sql.out

@@ -1755,7 +1755,7 @@ select -(a) from values (interval '-2147483648 months', interval '2147483647 mon
 -- !query schema
 struct<>
 -- !query output
-java.lang.ArithmeticException
+org.apache.spark.SparkArithmeticException
 integer overflow. If necessary set spark.sql.ansi.enabled to false (except for ANSI interval type) to bypass this error.
 
 
@@ -1764,7 +1764,7 @@ select a - b from values (interval '-2147483648 months', interval '2147483647 mo
 -- !query schema
 struct<>
 -- !query output
-java.lang.ArithmeticException
+org.apache.spark.SparkArithmeticException
 integer overflow. If necessary set spark.sql.ansi.enabled to false (except for ANSI interval type) to bypass this error.
 
 
@@ -1773,7 +1773,7 @@ select b + interval '1 month' from values (interval '-2147483648 months', interv
 -- !query schema
 struct<>
 -- !query output
-java.lang.ArithmeticException
+org.apache.spark.SparkArithmeticException
 integer overflow. If necessary set spark.sql.ansi.enabled to false (except for ANSI interval type) to bypass this error.
 
 
@@ -2002,7 +2002,7 @@ SELECT (INTERVAL '-178956970-8' YEAR TO MONTH) / -1
 -- !query schema
 struct<>
 -- !query output
-java.lang.ArithmeticException
+org.apache.spark.SparkArithmeticException
 Overflow in integral divide. To return NULL instead, use 'try_divide'. If necessary set spark.sql.ansi.enabled to false (except for ANSI interval type) to bypass this error.
 == SQL(line 1, position 7) ==
 SELECT (INTERVAL '-178956970-8' YEAR TO MONTH) / -1
@@ -2014,7 +2014,7 @@ SELECT (INTERVAL '-178956970-8' YEAR TO MONTH) / -1L
 -- !query schema
 struct<>
 -- !query output
-java.lang.ArithmeticException
+org.apache.spark.SparkArithmeticException
 Overflow in integral divide. To return NULL instead, use 'try_divide'. If necessary set spark.sql.ansi.enabled to false (except for ANSI interval type) to bypass this error.
 == SQL(line 1, position 7) ==
 SELECT (INTERVAL '-178956970-8' YEAR TO MONTH) / -1L
@@ -2060,7 +2060,7 @@ SELECT (INTERVAL '-106751991 04:00:54.775808' DAY TO SECOND) / -1
 -- !query schema
 struct<>
 -- !query output
-java.lang.ArithmeticException
+org.apache.spark.SparkArithmeticException
 Overflow in integral divide. To return NULL instead, use 'try_divide'. If necessary set spark.sql.ansi.enabled to false (except for ANSI interval type) to bypass this error.
 == SQL(line 1, position 7) ==
 SELECT (INTERVAL '-106751991 04:00:54.775808' DAY TO SECOND) / -1
@@ -2072,7 +2072,7 @@ SELECT (INTERVAL '-106751991 04:00:54.775808' DAY TO SECOND) / -1L
 -- !query schema
 struct<>
 -- !query output
-java.lang.ArithmeticException
+org.apache.spark.SparkArithmeticException
 Overflow in integral divide. To return NULL instead, use 'try_divide'. If necessary set spark.sql.ansi.enabled to false (except for ANSI interval type) to bypass this error.
 == SQL(line 1, position 7) ==
 SELECT (INTERVAL '-106751991 04:00:54.775808' DAY TO SECOND) / -1L

sql/core/src/test/resources/sql-tests/results/interval.sql.out

@@ -1744,7 +1744,7 @@ select -(a) from values (interval '-2147483648 months', interval '2147483647 mon
 -- !query schema
 struct<>
 -- !query output
-java.lang.ArithmeticException
+org.apache.spark.SparkArithmeticException
 integer overflow. If necessary set spark.sql.ansi.enabled to false (except for ANSI interval type) to bypass this error.
 
 
@@ -1753,7 +1753,7 @@ select a - b from values (interval '-2147483648 months', interval '2147483647 mo
 -- !query schema
 struct<>
 -- !query output
-java.lang.ArithmeticException
+org.apache.spark.SparkArithmeticException
 integer overflow. If necessary set spark.sql.ansi.enabled to false (except for ANSI interval type) to bypass this error.
 
 
@@ -1762,7 +1762,7 @@ select b + interval '1 month' from values (interval '-2147483648 months', interv
 -- !query schema
 struct<>
 -- !query output
-java.lang.ArithmeticException
+org.apache.spark.SparkArithmeticException
 integer overflow. If necessary set spark.sql.ansi.enabled to false (except for ANSI interval type) to bypass this error.
 
 
@@ -1991,7 +1991,7 @@ SELECT (INTERVAL '-178956970-8' YEAR TO MONTH) / -1
 -- !query schema
 struct<>
 -- !query output
-java.lang.ArithmeticException
+org.apache.spark.SparkArithmeticException
 Overflow in integral divide. To return NULL instead, use 'try_divide'. If necessary set spark.sql.ansi.enabled to false (except for ANSI interval type) to bypass this error.
 == SQL(line 1, position 7) ==
 SELECT (INTERVAL '-178956970-8' YEAR TO MONTH) / -1
@@ -2003,7 +2003,7 @@ SELECT (INTERVAL '-178956970-8' YEAR TO MONTH) / -1L
 -- !query schema
 struct<>
 -- !query output
-java.lang.ArithmeticException
+org.apache.spark.SparkArithmeticException
 Overflow in integral divide. To return NULL instead, use 'try_divide'. If necessary set spark.sql.ansi.enabled to false (except for ANSI interval type) to bypass this error.
 == SQL(line 1, position 7) ==
 SELECT (INTERVAL '-178956970-8' YEAR TO MONTH) / -1L
@@ -2049,7 +2049,7 @@ SELECT (INTERVAL '-106751991 04:00:54.775808' DAY TO SECOND) / -1
 -- !query schema
 struct<>
 -- !query output
-java.lang.ArithmeticException
+org.apache.spark.SparkArithmeticException
 Overflow in integral divide. To return NULL instead, use 'try_divide'. If necessary set spark.sql.ansi.enabled to false (except for ANSI interval type) to bypass this error.
 == SQL(line 1, position 7) ==
 SELECT (INTERVAL '-106751991 04:00:54.775808' DAY TO SECOND) / -1
@@ -2061,7 +2061,7 @@ SELECT (INTERVAL '-106751991 04:00:54.775808' DAY TO SECOND) / -1L
 -- !query schema
 struct<>
 -- !query output
-java.lang.ArithmeticException
+org.apache.spark.SparkArithmeticException
 Overflow in integral divide. To return NULL instead, use 'try_divide'. If necessary set spark.sql.ansi.enabled to false (except for ANSI interval type) to bypass this error.
 == SQL(line 1, position 7) ==
 SELECT (INTERVAL '-106751991 04:00:54.775808' DAY TO SECOND) / -1L

sql/core/src/test/resources/sql-tests/results/postgreSQL/int4.sql.out

@@ -199,7 +199,7 @@ SELECT '' AS five, i.f1, i.f1 * smallint('2') AS x FROM INT4_TBL i
 -- !query schema
 struct<>
 -- !query output
-java.lang.ArithmeticException
+org.apache.spark.SparkArithmeticException
 integer overflow. If necessary set spark.sql.ansi.enabled to false (except for ANSI interval type) to bypass this error.
 == SQL(line 1, position 25) ==
 SELECT '' AS five, i.f1, i.f1 * smallint('2') AS x FROM INT4_TBL i
@@ -222,7 +222,7 @@ SELECT '' AS five, i.f1, i.f1 * int('2') AS x FROM INT4_TBL i
 -- !query schema
 struct<>
 -- !query output
-java.lang.ArithmeticException
+org.apache.spark.SparkArithmeticException
 integer overflow. If necessary set spark.sql.ansi.enabled to false (except for ANSI interval type) to bypass this error.
 == SQL(line 1, position 25) ==
 SELECT '' AS five, i.f1, i.f1 * int('2') AS x FROM INT4_TBL i
@@ -245,7 +245,7 @@ SELECT '' AS five, i.f1, i.f1 + smallint('2') AS x FROM INT4_TBL i
 -- !query schema
 struct<>
 -- !query output
-java.lang.ArithmeticException
+org.apache.spark.SparkArithmeticException
 integer overflow. If necessary set spark.sql.ansi.enabled to false (except for ANSI interval type) to bypass this error.
 == SQL(line 1, position 25) ==
 SELECT '' AS five, i.f1, i.f1 + smallint('2') AS x FROM INT4_TBL i
@@ -269,7 +269,7 @@ SELECT '' AS five, i.f1, i.f1 + int('2') AS x FROM INT4_TBL i
 -- !query schema
 struct<>
 -- !query output
-java.lang.ArithmeticException
+org.apache.spark.SparkArithmeticException
 integer overflow. If necessary set spark.sql.ansi.enabled to false (except for ANSI interval type) to bypass this error.
 == SQL(line 1, position 25) ==
 SELECT '' AS five, i.f1, i.f1 + int('2') AS x FROM INT4_TBL i
@@ -293,7 +293,7 @@ SELECT '' AS five, i.f1, i.f1 - smallint('2') AS x FROM INT4_TBL i
 -- !query schema
 struct<>
 -- !query output
-java.lang.ArithmeticException
+org.apache.spark.SparkArithmeticException
 integer overflow. If necessary set spark.sql.ansi.enabled to false (except for ANSI interval type) to bypass this error.
 == SQL(line 1, position 25) ==
 SELECT '' AS five, i.f1, i.f1 - smallint('2') AS x FROM INT4_TBL i
@@ -317,7 +317,7 @@ SELECT '' AS five, i.f1, i.f1 - int('2') AS x FROM INT4_TBL i
 -- !query schema
 struct<>
 -- !query output
-java.lang.ArithmeticException
+org.apache.spark.SparkArithmeticException
 integer overflow. If necessary set spark.sql.ansi.enabled to false (except for ANSI interval type) to bypass this error.
 == SQL(line 1, position 25) ==
 SELECT '' AS five, i.f1, i.f1 - int('2') AS x FROM INT4_TBL i

sql/core/src/test/resources/sql-tests/results/postgreSQL/int8.sql.out

@@ -391,7 +391,7 @@ SELECT '' AS three, q1, q2, q1 * q2 AS multiply FROM INT8_TBL
 -- !query schema
 struct<>
 -- !query output
-java.lang.ArithmeticException
+org.apache.spark.SparkArithmeticException
 long overflow. If necessary set spark.sql.ansi.enabled to false (except for ANSI interval type) to bypass this error.
 == SQL(line 1, position 28) ==
 SELECT '' AS three, q1, q2, q1 * q2 AS multiply FROM INT8_TBL
@@ -753,7 +753,7 @@ SELECT bigint((-9223372036854775808)) * bigint((-1))
 -- !query schema
 struct<>
 -- !query output
-java.lang.ArithmeticException
+org.apache.spark.SparkArithmeticException
 long overflow. If necessary set spark.sql.ansi.enabled to false (except for ANSI interval type) to bypass this error.
 == SQL(line 1, position 7) ==
 SELECT bigint((-9223372036854775808)) * bigint((-1))
@@ -781,7 +781,7 @@ SELECT bigint((-9223372036854775808)) * int((-1))
 -- !query schema
 struct<>
 -- !query output
-java.lang.ArithmeticException
+org.apache.spark.SparkArithmeticException
 long overflow. If necessary set spark.sql.ansi.enabled to false (except for ANSI interval type) to bypass this error.
 == SQL(line 1, position 7) ==
 SELECT bigint((-9223372036854775808)) * int((-1))
@@ -809,7 +809,7 @@ SELECT bigint((-9223372036854775808)) * smallint((-1))
 -- !query schema
 struct<>
 -- !query output
-java.lang.ArithmeticException
+org.apache.spark.SparkArithmeticException
 long overflow. If necessary set spark.sql.ansi.enabled to false (except for ANSI interval type) to bypass this error.
 == SQL(line 1, position 7) ==
 SELECT bigint((-9223372036854775808)) * smallint((-1))

sql/core/src/test/resources/sql-tests/results/postgreSQL/window_part2.sql.out

@@ -224,7 +224,7 @@ from range(9223372036854775804, 9223372036854775807) x
 -- !query schema
 struct<>
 -- !query output
-java.lang.ArithmeticException
+org.apache.spark.SparkArithmeticException
 long overflow. If necessary set spark.sql.ansi.enabled to false (except for ANSI interval type) to bypass this error.
 
 
@@ -234,7 +234,7 @@ from range(-9223372036854775806, -9223372036854775805) x
 -- !query schema
 struct<>
 -- !query output
-java.lang.ArithmeticException
+org.apache.spark.SparkArithmeticException
 long overflow. If necessary set spark.sql.ansi.enabled to false (except for ANSI interval type) to bypass this error.
 
 

sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala

@@ -1278,12 +1278,12 @@ class DataFrameAggregateSuite extends QueryTest
     val error = intercept[SparkException] {
       checkAnswer(df2.select(sum($"year-month")), Nil)
     }
-    assert(error.toString contains "java.lang.ArithmeticException: integer overflow")
+    assert(error.toString contains "SparkArithmeticException: integer overflow")
 
     val error2 = intercept[SparkException] {
       checkAnswer(df2.select(sum($"day")), Nil)
     }
-    assert(error2.toString contains "java.lang.ArithmeticException: long overflow")
+    assert(error2.toString contains "SparkArithmeticException: long overflow")
   }
 
   test("SPARK-34837: Support ANSI SQL intervals by the aggregate function `avg`") {
@@ -1412,12 +1412,12 @@ class DataFrameAggregateSuite extends QueryTest
     val error = intercept[SparkException] {
       checkAnswer(df2.select(avg($"year-month")), Nil)
     }
-    assert(error.toString contains "java.lang.ArithmeticException: integer overflow")
+    assert(error.toString contains "SparkArithmeticException: integer overflow")
 
     val error2 = intercept[SparkException] {
       checkAnswer(df2.select(avg($"day")), Nil)
     }
-    assert(error2.toString contains "java.lang.ArithmeticException: long overflow")
+    assert(error2.toString contains "SparkArithmeticException: long overflow")
 
     val df3 = intervalData.filter($"class" > 4)
     val avgDF3 = df3.select(avg($"year-month"), avg($"day"))