[SPARK-52659][SQL] Misleading modulo error message in ansi mode

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

1. **Updated error condition**: Changed `MOD_BY_ZERO` to `REMAINDER_BY_ZERO` in `error-conditions.json` with an appropriate error message "Remainder by zero" instead of "Mod by zero"
2. **Updated test cases**: Modified `ArithmeticExpressionSuite.scala` to separate division and remainder operation tests, ensuring that remainder operations (`Remainder` and `Pmod`) expect "Remainder by zero" error message instead of "Division by zero"

The mod function now correctly throws `REMAINDER_BY_ZERO` error with message "Remainder by zero. Use `try_mod` to tolerate divisor being 0 and return NULL instead." instead of the misleading `DIVIDE_BY_ZERO` error.

### Why are the changes needed?

In ANSI mode, when executing `spark.sql("select mod(10,0)")`, the system incorrectly throws a `DIVIDE_BY_ZERO` error with message "Division by zero".

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

Yes. This PR changes the user-facing error message for modulo operations in ANSI mode.

**Before:**
org.apache.spark.SparkArithmeticException: [DIVIDE_BY_ZERO] Division by zero. Use try_divide to tolerate divisor being 0 and return NULL instead.

**After:**
org.apache.spark.SparkArithmeticException: [REMAINDER_BY_ZERO] Remainder by zero. Use try_mod to tolerate divisor being 0 and return NULL instead.

### How was this patch tested?

Updated existing unit tests.

### Was this patch authored or co-authored using generative AI tooling?

No.

This is my first pull request. I'm writing this pull request after reviewing the documentation as much as possible, but if there's anything wrong, please let me know.

Closes #51378 from pasar6987/Misleading-modulo-error-message-in-ANSI-mode.

Lead-authored-by: 공성재 <[email protected]>
Co-authored-by: 공성재 <[email protected]>
Co-authored-by: pasar6987 <[email protected]>
Signed-off-by: Gengliang Wang <[email protected]>

Author: 2 people

common/utils/src/main/resources/error/error-conditions.json

@@ -4920,6 +4920,12 @@
     ],
     "sqlState" : "42601"
   },
+  "REMAINDER_BY_ZERO" : {
+    "message" : [
+      "Remainder by zero. Use `try_mod` to tolerate divisor being 0 and return NULL instead. If necessary set <config> to \"false\" to bypass this error."
+    ],
+    "sqlState" : "22012"
+  },
   "RENAME_SRC_PATH_NOT_FOUND" : {
     "message" : [
       "Failed to rename as <sourcePath> was not found."

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

@@ -654,7 +654,13 @@ trait DivModLike extends BinaryArithmetic {
       } else {
         if (isZero(input2)) {
           // when we reach here, failOnError must be true.
-          throw QueryExecutionErrors.divideByZeroError(getContextOrNull())
+          val context = getContextOrNull()
+          val ex = this match {
+            case _: Remainder => QueryExecutionErrors.remainderByZeroError(context)
+            case _: Pmod => QueryExecutionErrors.remainderByZeroError(context)
+            case _ => QueryExecutionErrors.divideByZeroError(context)
+          }
+          throw ex
         }
         if (checkDivideOverflow && input1 == Long.MinValue && input2 == -1) {
           throw QueryExecutionErrors.overflowInIntegralDivideError(getContextOrNull())
@@ -669,6 +675,15 @@ trait DivModLike extends BinaryArithmetic {
   /**
    * Special case handling due to division/remainder by 0 => null or ArithmeticException.
    */
+  protected def divideByZeroErrorCode(ctx: CodegenContext): String = {
+    val errorContextCode = getContextOrNullCode(ctx, failOnError)
+    this match {
+      case _: Remainder => s"QueryExecutionErrors.remainderByZeroError($errorContextCode)"
+      case _: Pmod => s"QueryExecutionErrors.remainderByZeroError($errorContextCode)"
+      case _ => s"QueryExecutionErrors.divideByZeroError($errorContextCode)"
+    }
+  }
+
   override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
     val eval1 = left.genCode(ctx)
     val eval2 = right.genCode(ctx)
@@ -706,7 +721,7 @@ trait DivModLike extends BinaryArithmetic {
     // evaluate right first as we have a chance to skip left if right is 0
     if (!left.nullable && !right.nullable) {
       val divByZero = if (failOnError) {
-        s"throw QueryExecutionErrors.divideByZeroError($errorContextCode);"
+        s"throw ${divideByZeroErrorCode(ctx)};"
       } else {
         s"${ev.isNull} = true;"
       }
@@ -724,7 +739,7 @@ trait DivModLike extends BinaryArithmetic {
     } else {
       val nullOnErrorCondition = if (failOnError) "" else s" || $isZero"
       val failOnErrorBranch = if (failOnError) {
-        s"if ($isZero) throw QueryExecutionErrors.divideByZeroError($errorContextCode);"
+        s"if ($isZero) throw ${divideByZeroErrorCode(ctx)};"
       } else {
         ""
       }
@@ -1047,7 +1062,7 @@ case class Pmod(
       } else {
         if (isZero(input2)) {
           // when we reach here, failOnError must bet true.
-          throw QueryExecutionErrors.divideByZeroError(getContextOrNull())
+          throw QueryExecutionErrors.remainderByZeroError(getContextOrNull())
         }
         pmodFunc(input1, input2)
       }
@@ -1104,7 +1119,7 @@ case class Pmod(
     // evaluate right first as we have a chance to skip left if right is 0
     if (!left.nullable && !right.nullable) {
       val divByZero = if (failOnError) {
-        s"throw QueryExecutionErrors.divideByZeroError($errorContext);"
+        s"throw QueryExecutionErrors.remainderByZeroError($errorContext);"
       } else {
         s"${ev.isNull} = true;"
       }
@@ -1121,7 +1136,7 @@ case class Pmod(
     } else {
       val nullOnErrorCondition = if (failOnError) "" else s" || $isZero"
       val failOnErrorBranch = if (failOnError) {
-        s"if ($isZero) throw QueryExecutionErrors.divideByZeroError($errorContext);"
+        s"if ($isZero) throw QueryExecutionErrors.remainderByZeroError($errorContext);"
       } else {
         ""
       }

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

@@ -205,6 +205,14 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase with ExecutionE
       summary = getSummary(context))
   }
 
+  def remainderByZeroError(context: QueryContext): ArithmeticException = {
+    new SparkArithmeticException(
+      errorClass = "REMAINDER_BY_ZERO",
+      messageParameters = Map("config" -> toSQLConf(SQLConf.ANSI_ENABLED.key)),
+      context = Array(context),
+      summary = getSummary(context))
+  }
+
   def intervalDividedByZeroError(context: QueryContext): ArithmeticException = {
     new SparkArithmeticException(
       errorClass = "INTERVAL_DIVIDED_BY_ZERO",

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/ArithmeticExpressionSuite.scala

@@ -463,7 +463,7 @@ class ArithmeticExpressionSuite extends SparkFunSuite with ExpressionEvalHelper
       }
       withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") {
         checkExceptionInExpression[ArithmeticException](
-          Remainder(left, Literal(convert(0))), "Division by zero")
+          Remainder(left, Literal(convert(0))), "Remainder by zero")
       }
     }
     checkEvaluation(Remainder(positiveShortLit, positiveShortLit), 0.toShort)
@@ -567,7 +567,7 @@ class ArithmeticExpressionSuite extends SparkFunSuite with ExpressionEvalHelper
       }
       withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") {
         checkExceptionInExpression[ArithmeticException](
-          Pmod(left, Literal(convert(0))), "Division by zero")
+          Pmod(left, Literal(convert(0))), "Remainder by zero")
       }
     }
     checkEvaluation(Pmod(Literal(-7), Literal(3)), 2)
@@ -873,12 +873,13 @@ class ArithmeticExpressionSuite extends SparkFunSuite with ExpressionEvalHelper
 
   test("SPARK-33008: division by zero on divide-like operations returns incorrect result") {
     withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") {
-      val operators: Seq[((Expression, Expression) => Expression, ((Int => Any) => Unit) => Unit)] =
+      // Test division operations
+      val divideOperators: Seq[
+        ((Expression, Expression) => Expression, ((Int => Any) => Unit) => Unit)
+      ] =
         Seq((Divide(_, _), testDecimalAndDoubleType),
-          (IntegralDivide(_, _), testDecimalAndLongType),
-          (Remainder(_, _), testNumericDataTypes),
-          (Pmod(_, _), testNumericDataTypes))
-      operators.foreach { case (operator, testTypesFn) =>
+          (IntegralDivide(_, _), testDecimalAndLongType))
+      divideOperators.foreach { case (operator, testTypesFn) =>
         testTypesFn { convert =>
           val one = Literal(convert(1))
           val zero = Literal(convert(0))
@@ -887,6 +888,22 @@ class ArithmeticExpressionSuite extends SparkFunSuite with ExpressionEvalHelper
           checkExceptionInExpression[ArithmeticException](operator(one, zero), "Division by zero")
         }
       }
+
+      // Test remainder operations
+      val remainderOperators: Seq[
+        ((Expression, Expression) => Expression, ((Int => Any) => Unit) => Unit)
+      ] =
+        Seq((Remainder(_, _), testNumericDataTypes),
+          (Pmod(_, _), testNumericDataTypes))
+      remainderOperators.foreach { case (operator, testTypesFn) =>
+        testTypesFn { convert =>
+          val one = Literal(convert(1))
+          val zero = Literal(convert(0))
+          checkEvaluation(operator(Literal.create(null, one.dataType), zero), null)
+          checkEvaluation(operator(one, Literal.create(null, zero.dataType)), null)
+          checkExceptionInExpression[ArithmeticException](operator(one, zero), "Remainder by zero")
+        }
+      }
     }
   }
 
@@ -931,12 +948,13 @@ class ArithmeticExpressionSuite extends SparkFunSuite with ExpressionEvalHelper
 
   test("SPARK-34920: error class") {
     withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") {
-      val operators: Seq[((Expression, Expression) => Expression, ((Int => Any) => Unit) => Unit)] =
+      // Test division operations
+      val divideOperators: Seq[
+        ((Expression, Expression) => Expression, ((Int => Any) => Unit) => Unit)
+      ] =
         Seq((Divide(_, _), testDecimalAndDoubleType),
-          (IntegralDivide(_, _), testDecimalAndLongType),
-          (Remainder(_, _), testNumericDataTypes),
-          (Pmod(_, _), testNumericDataTypes))
-      operators.foreach { case (operator, testTypesFn) =>
+          (IntegralDivide(_, _), testDecimalAndLongType))
+      divideOperators.foreach { case (operator, testTypesFn) =>
         testTypesFn { convert =>
           val one = Literal(convert(1))
           val zero = Literal(convert(0))
@@ -946,6 +964,23 @@ class ArithmeticExpressionSuite extends SparkFunSuite with ExpressionEvalHelper
             "Division by zero")
         }
       }
+
+      // Test remainder operations
+      val remainderOperators: Seq[
+        ((Expression, Expression) => Expression, ((Int => Any) => Unit) => Unit)
+      ] =
+        Seq((Remainder(_, _), testNumericDataTypes),
+          (Pmod(_, _), testNumericDataTypes))
+      remainderOperators.foreach { case (operator, testTypesFn) =>
+        testTypesFn { convert =>
+          val one = Literal(convert(1))
+          val zero = Literal(convert(0))
+          checkEvaluation(operator(Literal.create(null, one.dataType), zero), null)
+          checkEvaluation(operator(one, Literal.create(null, zero.dataType)), null)
+          checkExceptionInExpression[SparkArithmeticException](operator(one, zero),
+            "Remainder by zero")
+        }
+      }
     }
   }
 

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

@@ -36,7 +36,7 @@ struct<>
 -- !query output
 org.apache.spark.SparkArithmeticException
 {
-  "errorClass" : "DIVIDE_BY_ZERO",
+  "errorClass" : "REMAINDER_BY_ZERO",
   "sqlState" : "22012",
   "messageParameters" : {
     "config" : "\"spark.sql.ansi.enabled\""
@@ -58,7 +58,7 @@ struct<>
 -- !query output
 org.apache.spark.SparkArithmeticException
 {
-  "errorClass" : "DIVIDE_BY_ZERO",
+  "errorClass" : "REMAINDER_BY_ZERO",
   "sqlState" : "22012",
   "messageParameters" : {
     "config" : "\"spark.sql.ansi.enabled\""

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

@@ -496,7 +496,7 @@ struct<>
 -- !query output
 org.apache.spark.SparkArithmeticException
 {
-  "errorClass" : "DIVIDE_BY_ZERO",
+  "errorClass" : "REMAINDER_BY_ZERO",
   "sqlState" : "22012",
   "messageParameters" : {
     "config" : "\"spark.sql.ansi.enabled\""
@@ -566,7 +566,7 @@ struct<>
 -- !query output
 org.apache.spark.SparkArithmeticException
 {
-  "errorClass" : "DIVIDE_BY_ZERO",
+  "errorClass" : "REMAINDER_BY_ZERO",
   "sqlState" : "22012",
   "messageParameters" : {
     "config" : "\"spark.sql.ansi.enabled\""
@@ -588,7 +588,7 @@ struct<>
 -- !query output
 org.apache.spark.SparkArithmeticException
 {
-  "errorClass" : "DIVIDE_BY_ZERO",
+  "errorClass" : "REMAINDER_BY_ZERO",
   "sqlState" : "22012",
   "messageParameters" : {
     "config" : "\"spark.sql.ansi.enabled\""

sql/core/src/test/scala/org/apache/spark/sql/errors/QueryExecutionAnsiErrorsSuite.scala

@@ -85,6 +85,26 @@ class QueryExecutionAnsiErrorsSuite extends QueryTest
         callSitePattern = getCurrentClassCallSitePattern))
   }
 
+  test("REMAINDER_BY_ZERO: can't take modulo of an integer by zero") {
+    checkError(
+      exception = intercept[SparkArithmeticException] {
+        sql("select 6 % 0").collect()
+      },
+      condition = "REMAINDER_BY_ZERO",
+      sqlState = "22012",
+      parameters = Map("config" -> ansiConf),
+      context = ExpectedContext(fragment = "6 % 0", start = 7, stop = 11))
+
+    checkError(
+      exception = intercept[SparkArithmeticException] {
+        sql("select pmod(6, 0)").collect()
+      },
+      condition = "REMAINDER_BY_ZERO",
+      sqlState = "22012",
+      parameters = Map("config" -> ansiConf),
+      context = ExpectedContext(fragment = "pmod(6, 0)", start = 7, stop = 16))
+  }
+
   test("INTERVAL_DIVIDED_BY_ZERO: interval divided by zero") {
     checkError(
       exception = intercept[SparkArithmeticException] {