[SPARK-29096][SQL] The exact math method should be called only when there is a corresponding function in Math

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala

@@ -1275,7 +1275,7 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
           if ($longValue == ($integralType) $longValue) {
             $evPrim = ($integralType) $longValue;
           } else {
-            throw new ArithmeticException("Casting $c to $integralType causes overflow");
+            throw new ArithmeticException("Casting " + $c + " to $integralType causes overflow");
           }
         """
     } else {
@@ -1300,7 +1300,7 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
         if ($c == ($integralType) $c) {
           $evPrim = ($integralType) $c;
         } else {
-          throw new ArithmeticException("Casting $c to $integralType causes overflow");
+          throw new ArithmeticException("Casting " + $c + " to $integralType causes overflow");
         }
       """
   }
@@ -1335,7 +1335,7 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
         if ($mathClass.floor($c) <= $max && $mathClass.ceil($c) >= $min) {
           $evPrim = ($integralType) $c;
         } else {
-          throw new ArithmeticException("Casting $c to $integralType causes overflow");
+          throw new ArithmeticException("Casting " + $c + " to $integralType causes overflow");
         }
       """
   }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/arithmetic.scala

@@ -150,9 +150,10 @@ abstract class BinaryArithmetic extends BinaryOperator with NullIntolerant {
   def calendarIntervalMethod: String =
     sys.error("BinaryArithmetics must override either calendarIntervalMethod or genCode")
 
-  /** Name of the function for the exact version of this expression in [[Math]]. */
-  def exactMathMethod: String =
-    sys.error("BinaryArithmetics must override either exactMathMethod or genCode")
+  // Name of the function for the exact version of this expression in [[Math]].
+  // If the option "spark.sql.failOnIntegralTypeOverflow" is enabled and there is corresponding
+  // function in [[Math]], the exact function will be called instead of evaluation with [[symbol]].
+  def exactMathMethod: Option[String] = None
 
   override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = dataType match {
     case _: DecimalType =>
@@ -182,9 +183,9 @@ abstract class BinaryArithmetic extends BinaryOperator with NullIntolerant {
       })
     case IntegerType | LongType =>
       nullSafeCodeGen(ctx, ev, (eval1, eval2) => {
-        val operation = if (checkOverflow) {
+        val operation = if (checkOverflow && exactMathMethod.isDefined) {
           val mathClass = classOf[Math].getName
-          s"$mathClass.$exactMathMethod($eval1, $eval2)"
+          s"$mathClass.${exactMathMethod.get}($eval1, $eval2)"
         } else {
           s"$eval1 $symbol $eval2"
         }
@@ -235,7 +236,7 @@ case class Add(left: Expression, right: Expression) extends BinaryArithmetic {
     }
   }
 
-  override def exactMathMethod: String = "addExact"
+  override def exactMathMethod: Option[String] = Some("addExact")
 }
 
 @ExpressionDescription(
@@ -265,7 +266,7 @@ case class Subtract(left: Expression, right: Expression) extends BinaryArithmeti
     }
   }
 
-  override def exactMathMethod: String = "subtractExact"
+  override def exactMathMethod: Option[String] = Some("subtractExact")
 }
 
 @ExpressionDescription(
@@ -286,7 +287,7 @@ case class Multiply(left: Expression, right: Expression) extends BinaryArithmeti
 
   protected override def nullSafeEval(input1: Any, input2: Any): Any = numeric.times(input1, input2)
 
-  override def exactMathMethod: String = "multiplyExact"
+  override def exactMathMethod: Option[String] = Some("multiplyExact")
 }
 
 // Common base trait for Divide and Remainder, since these two classes are almost identical

sql/core/src/test/resources/sql-tests/results/pgSQL/float4.sql.out

@@ -334,9 +334,10 @@ struct<CAST(CAST(-2147483648.5 AS FLOAT) AS INT):int>
 -- !query 37
 SELECT int(float('-2147483900'))
 -- !query 37 schema
-struct<CAST(CAST(-2147483900 AS FLOAT) AS INT):int>
+struct<>
 -- !query 37 output
--2147483648
+java.lang.ArithmeticException
+Casting -2.1474839E9 to int causes overflow.
 
 
 -- !query 38
@@ -366,9 +367,10 @@ struct<CAST(CAST(-9223372036854775808.5 AS FLOAT) AS BIGINT):bigint>
 -- !query 41
 SELECT bigint(float('-9223380000000000000'))
 -- !query 41 schema
-struct<CAST(CAST(-9223380000000000000 AS FLOAT) AS BIGINT):bigint>
+struct<>
 -- !query 41 output
--9223372036854775808
+java.lang.ArithmeticException
+Casting -9.22338E18 to int causes overflow.
 
 
 -- !query 42

sql/core/src/test/resources/sql-tests/results/pgSQL/float8.sql.out

@@ -826,9 +826,10 @@ struct<CAST(CAST(-9223372036854775808.5 AS DOUBLE) AS BIGINT):bigint>
 -- !query 93
 SELECT bigint(double('-9223372036854780000'))
 -- !query 93 schema
-struct<CAST(CAST(-9223372036854780000 AS DOUBLE) AS BIGINT):bigint>
+struct<>
 -- !query 93 output
--9223372036854775808
+java.lang.ArithmeticException
+Casting -9.22337203685478E18 to long causes overflow.
 
 
 -- !query 94

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

@@ -205,13 +205,10 @@ struct<three:string,f1:int>
 -- !query 22
 SELECT '' AS five, i.f1, i.f1 * smallint('2') AS x FROM INT4_TBL i
 -- !query 22 schema
-struct<five:string,f1:int,x:int>
+struct<>
 -- !query 22 output
-	-123456	-246912
-	-2147483647	2
-	0	0
-	123456	246912
-	2147483647	-2
+java.lang.ArithmeticException
+integer overflow
 
 
 -- !query 23
@@ -228,13 +225,10 @@ struct<five:string,f1:int,x:int>
 -- !query 24
 SELECT '' AS five, i.f1, i.f1 * int('2') AS x FROM INT4_TBL i
 -- !query 24 schema
-struct<five:string,f1:int,x:int>
+struct<>
 -- !query 24 output
-	-123456	-246912
-	-2147483647	2
-	0	0
-	123456	246912
-	2147483647	-2
+java.lang.ArithmeticException
+integer overflow
 
 
 -- !query 25
@@ -251,13 +245,10 @@ struct<five:string,f1:int,x:int>
 -- !query 26
 SELECT '' AS five, i.f1, i.f1 + smallint('2') AS x FROM INT4_TBL i
 -- !query 26 schema
-struct<five:string,f1:int,x:int>
+struct<>
 -- !query 26 output
-	-123456	-123454
-	-2147483647	-2147483645
-	0	2
-	123456	123458
-	2147483647	-2147483647
+java.lang.ArithmeticException
+integer overflow
 
 
 -- !query 27
@@ -275,13 +266,10 @@ struct<five:string,f1:int,x:int>
 -- !query 28
 SELECT '' AS five, i.f1, i.f1 + int('2') AS x FROM INT4_TBL i
 -- !query 28 schema
-struct<five:string,f1:int,x:int>
+struct<>
 -- !query 28 output
-	-123456	-123454
-	-2147483647	-2147483645
-	0	2
-	123456	123458
-	2147483647	-2147483647
+java.lang.ArithmeticException
+integer overflow
 
 
 -- !query 29
@@ -299,13 +287,10 @@ struct<five:string,f1:int,x:int>
 -- !query 30
 SELECT '' AS five, i.f1, i.f1 - smallint('2') AS x FROM INT4_TBL i
 -- !query 30 schema
-struct<five:string,f1:int,x:int>
+struct<>
 -- !query 30 output
-	-123456	-123458
-	-2147483647	2147483647
-	0	-2
-	123456	123454
-	2147483647	2147483645
+java.lang.ArithmeticException
+integer overflow
 
 
 -- !query 31
@@ -323,13 +308,10 @@ struct<five:string,f1:int,x:int>
 -- !query 32
 SELECT '' AS five, i.f1, i.f1 - int('2') AS x FROM INT4_TBL i
 -- !query 32 schema
-struct<five:string,f1:int,x:int>
+struct<>
 -- !query 32 output
-	-123456	-123458
-	-2147483647	2147483647
-	0	-2
-	123456	123454
-	2147483647	2147483645
+java.lang.ArithmeticException
+integer overflow
 
 
 -- !query 33

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

@@ -389,13 +389,10 @@ struct<five:string,q1:bigint,q2:bigint,minus:bigint>
 -- !query 40
 SELECT '' AS three, q1, q2, q1 * q2 AS multiply FROM INT8_TBL
 -- !query 40 schema
-struct<three:string,q1:bigint,q2:bigint,multiply:bigint>
+struct<>
 -- !query 40 output
-	123	456	56088
-	123	4567890123456789	561850485185185047
-	4567890123456789	-4567890123456789	-4868582358072306617
-	4567890123456789	123	561850485185185047
-	4567890123456789	4567890123456789	4868582358072306617
+java.lang.ArithmeticException
+long overflow
 
 
 -- !query 41
@@ -604,12 +601,10 @@ struct<q1:int>
 -- !query 60
 SELECT CAST(q1 AS int) FROM int8_tbl WHERE q2 <> 456
 -- !query 60 schema
-struct<q1:int>
+struct<>
 -- !query 60 output
--869367531
--869367531
--869367531
-123
+java.lang.ArithmeticException
+Casting 4567890123456789 to int causes overflow
 
 
 -- !query 61
@@ -623,12 +618,10 @@ struct<q1:smallint>
 -- !query 62
 SELECT CAST(q1 AS smallint) FROM int8_tbl WHERE q2 <> 456
 -- !query 62 schema
-struct<q1:smallint>
+struct<>
 -- !query 62 output
--32491
--32491
--32491
-123
+java.lang.ArithmeticException
+Casting 4567890123456789 to short causes overflow
 
 
 -- !query 63
@@ -662,9 +655,10 @@ struct<CAST(CAST(36854775807.0 AS FLOAT) AS BIGINT):bigint>
 -- !query 66
 SELECT CAST(double('922337203685477580700.0') AS bigint)
 -- !query 66 schema
-struct<CAST(CAST(922337203685477580700.0 AS DOUBLE) AS BIGINT):bigint>
+struct<>
 -- !query 66 output
-9223372036854775807
+java.lang.ArithmeticException
+Casting 9.223372036854776E20 to long causes overflow.
 
 
 -- !query 67
@@ -728,17 +722,19 @@ struct<CAST(shiftleft(CAST(-1 AS BIGINT), 63) AS STRING):string>
 -- !query 72
 SELECT string(int(shiftleft(bigint(-1), 63))+1)
 -- !query 72 schema
-struct<CAST((CAST(shiftleft(CAST(-1 AS BIGINT), 63) AS INT) + 1) AS STRING):string>
+struct<>
 -- !query 72 output
-1
+java.lang.ArithmeticException
+Casting -9223372036854775808 to int causes overflow.
 
 
 -- !query 73
 SELECT bigint((-9223372036854775808)) * bigint((-1))
 -- !query 73 schema
-struct<(CAST(-9223372036854775808 AS BIGINT) * CAST(-1 AS BIGINT)):bigint>
+struct<>
 -- !query 73 output
--9223372036854775808
+java.lang.ArithmeticException
+long overflow
 
 
 -- !query 74
@@ -760,9 +756,10 @@ struct<(CAST(-9223372036854775808 AS BIGINT) % CAST(-1 AS BIGINT)):bigint>
 -- !query 76
 SELECT bigint((-9223372036854775808)) * int((-1))
 -- !query 76 schema
-struct<(CAST(-9223372036854775808 AS BIGINT) * CAST(CAST(-1 AS INT) AS BIGINT)):bigint>
+struct<>
 -- !query 76 output
--9223372036854775808
+java.lang.ArithmeticException
+long overflow
 
 
 -- !query 77
@@ -784,9 +781,10 @@ struct<(CAST(-9223372036854775808 AS BIGINT) % CAST(CAST(-1 AS INT) AS BIGINT)):
 -- !query 79
 SELECT bigint((-9223372036854775808)) * smallint((-1))
 -- !query 79 schema
-struct<(CAST(-9223372036854775808 AS BIGINT) * CAST(CAST(-1 AS SMALLINT) AS BIGINT)):bigint>
+struct<>
 -- !query 79 output
--9223372036854775808
+java.lang.ArithmeticException
+long overflow
 
 
 -- !query 80

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

@@ -22,6 +22,7 @@ import java.util.{Locale, TimeZone}
 
 import scala.util.control.NonFatal
 
+import org.apache.spark.SparkException
 import org.apache.spark.sql.catalyst.planning.PhysicalOperation
 import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.plans.logical.sql.{DescribeColumnStatement, DescribeTableStatement}
@@ -308,6 +309,7 @@ class SQLQueryTestSuite extends QueryTest with SharedSparkSession {
         localSparkSession.conf.set(SQLConf.CROSS_JOINS_ENABLED.key, true)
         localSparkSession.conf.set(SQLConf.ANSI_SQL_PARSER.key, true)
         localSparkSession.conf.set(SQLConf.PREFER_INTEGRAL_DIVISION.key, true)
+        localSparkSession.conf.set(SQLConf.FAIL_ON_INTEGRAL_TYPE_OVERFLOW.key, true)
       case _ =>
     }
 
@@ -413,6 +415,9 @@ class SQLQueryTestSuite extends QueryTest with SharedSparkSession {
         // with a generic pattern "###".
         val msg = if (a.plan.nonEmpty) a.getSimpleMessage else a.getMessage
         (StructType(Seq.empty), Seq(a.getClass.getName, msg.replaceAll("#\\d+", "#x")))
+      case s: SparkException =>
+        val cause = s.getCause
+        (StructType(Seq.empty), Seq(cause.getClass.getName, cause.getMessage))
       case NonFatal(e) =>
         // If there is an exception, put the exception class followed by the message.
         (StructType(Seq.empty), Seq(e.getClass.getName, e.getMessage))