support_ansi_mode_integral_divide

Author: Bhargava Vadlamani

native/core/src/execution/planner.rs

@@ -287,8 +287,6 @@ impl PhysicalPlanner {
                 )
             }
             ExprStruct::IntegralDivide(expr) => {
-                // TODO respect eval mode
-                // https://github.com/apache/datafusion-comet/issues/533
                 let eval_mode = from_protobuf_eval_mode(expr.eval_mode)?;
                 self.create_binary_expr_with_options(
                     expr.left.as_ref().unwrap(),
@@ -987,11 +985,12 @@ impl PhysicalPlanner {
                 } else {
                     "decimal_div"
                 };
-                let fun_expr = create_comet_physical_fun(
+                let fun_expr = create_comet_physical_fun_with_eval_mode(
                     func_name,
                     data_type.clone(),
                     &self.session_ctx.state(),
                     None,
+                    eval_mode,
                 )?;
                 Ok(Arc::new(ScalarFunctionExpr::new(
                     func_name,

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

@@ -200,7 +200,7 @@ object CometIntegralDivide extends CometExpressionSerde[IntegralDivide] with Mat
 
     val rightExpr = if (expr.evalMode != EvalMode.ANSI) nullIfWhenPrimitive(right) else right
 
-    val dataType = (left.dataType, right.dataType) match {
+    val dataType = (left.dataType, rightExpr.dataType) match {
       case (l: DecimalType, r: DecimalType) =>
         // copy from IntegralDivide.resultDecimalType
         val intDig = l.precision - l.scale + r.scale

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

@@ -2988,20 +2988,29 @@ class CometExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper {
   }
 
   test("ANSI support for integral divide (division by zero)") {
-    val data = Seq((Integer.MIN_VALUE, 0))
-    withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") {
-      withParquetTable(data, "tbl") {
-        val res = spark.sql("""
-                              |SELECT
-                              |  _1 div _2
-                              |  from tbl
-                              |  """.stripMargin)
-
-        checkSparkMaybeThrows(res) match {
-          case (Some(sparkExc), Some(cometExc)) =>
-            assert(cometExc.getMessage.contains(DIVIDE_BY_ZERO_EXCEPTION_MSG))
-            assert(sparkExc.getMessage.contains("Division by zero"))
-          case _ => fail("Exception should be thrown")
+    val data = Seq((Integer.MAX_VALUE, 0))
+    Seq("true", "false").foreach { p =>
+      withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") {
+        withParquetTable(data, "tbl") {
+          val res = spark.sql("""
+                |SELECT
+                |  _1 div _2
+                |  from tbl
+                |  """.stripMargin)
+
+          checkSparkMaybeThrows(res) match {
+            case (Some(sparkException), Some(cometException)) =>
+              assert(sparkException.getMessage.contains(DIVIDE_BY_ZERO_EXCEPTION_MSG))
+              assert(cometException.getMessage.contains(DIVIDE_BY_ZERO_EXCEPTION_MSG))
+            case (None, None) => checkSparkAnswerAndOperator(res)
+            case (None, Some(ex)) =>
+              fail(
+                "Comet threw an exception but Spark did not. Comet exception: " + ex.getMessage)
+            case (Some(sparkException), None) =>
+              fail(
+                "Spark threw an exception but Comet did not. Spark exception: " +
+                  sparkException.getMessage)
+          }
         }
       }
     }