fix : cast_operands_to_decimal_type_to_fix_arithmetic_overflow (#1996)

Author: coderfender

spark/src/main/scala/org/apache/comet/serde/arithmetic.scala

@@ -21,7 +21,7 @@ package org.apache.comet.serde
 
 import scala.math.min
 
-import org.apache.spark.sql.catalyst.expressions.{Add, Attribute, Divide, EqualTo, EvalMode, Expression, If, IntegralDivide, Literal, Multiply, Remainder, Subtract}
+import org.apache.spark.sql.catalyst.expressions.{Add, Attribute, Cast, Divide, EqualTo, EvalMode, Expression, If, IntegralDivide, Literal, Multiply, Remainder, Subtract}
 import org.apache.spark.sql.types.{ByteType, DataType, DecimalType, DoubleType, FloatType, IntegerType, LongType, ShortType}
 
 import org.apache.comet.CometSparkSessionExtensions.withInfo
@@ -201,7 +201,6 @@ object CometIntegralDivide extends CometExpressionSerde with MathBase {
       inputs: Seq[Attribute],
       binding: Boolean): Option[ExprOuterClass.Expr] = {
     val div = expr.asInstanceOf[IntegralDivide]
-    val rightExpr = nullIfWhenPrimitive(div.right)
 
     if (!supportedDataType(div.left.dataType)) {
       withInfo(div, s"Unsupported datatype ${div.left.dataType}")
@@ -212,17 +211,28 @@ object CometIntegralDivide extends CometExpressionSerde with MathBase {
       return None
     }
 
-    val dataType = (div.left.dataType, div.right.dataType) match {
+//    Precision is set to 19 (max precision for a numerical data type except DecimalType)
+
+    val left =
+      if (div.left.dataType.isInstanceOf[DecimalType]) div.left
+      else Cast(div.left, DecimalType(19, 0))
+    val right =
+      if (div.right.dataType.isInstanceOf[DecimalType]) div.right
+      else Cast(div.right, DecimalType(19, 0))
+
+    val rightExpr = nullIfWhenPrimitive(right)
+
+    val dataType = (left.dataType, right.dataType) match {
       case (l: DecimalType, r: DecimalType) =>
         // copy from IntegralDivide.resultDecimalType
         val intDig = l.precision - l.scale + r.scale
         DecimalType(min(if (intDig == 0) 1 else intDig, DecimalType.MAX_PRECISION), 0)
-      case _ => div.left.dataType
+      case _ => left.dataType
     }
 
     val divideExpr = createMathExpression(
       expr,
-      div.left,
+      left,
       rightExpr,
       inputs,
       binding,

spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala

@@ -113,6 +113,18 @@ class CometExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper {
     }
   }
 
+  test("Integral Division Overflow Handling Matches Spark Behavior") {
+    withTable("t1") {
+      withSQLConf(CometConf.COMET_CAST_ALLOW_INCOMPATIBLE.key -> "true") {
+        val value = Long.MinValue
+        sql("create table t1(c1 long, c2 short) using parquet")
+        sql(s"insert into t1 values($value, -1)")
+        val res = sql("select c1 div c2 from t1 order by c1")
+        checkSparkAnswerAndOperator(res)
+      }
+    }
+  }
+
   test("basic data type support") {
     Seq(true, false).foreach { dictionaryEnabled =>
       withTempDir { dir =>
@@ -2686,7 +2698,9 @@ class CometExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper {
 
   test("test integral divide") {
     // this test requires native_comet scan due to unsigned u8/u16 issue
-    withSQLConf(CometConf.COMET_NATIVE_SCAN_IMPL.key -> CometConf.SCAN_NATIVE_COMET) {
+    withSQLConf(
+      CometConf.COMET_NATIVE_SCAN_IMPL.key -> CometConf.SCAN_NATIVE_COMET,
+      CometConf.COMET_CAST_ALLOW_INCOMPATIBLE.key -> "true") {
       Seq(true, false).foreach { dictionaryEnabled =>
         withTempDir { dir =>
           val path1 = new Path(dir.toURI.toString, "test1.parquet")