[SPARK-28201][SQL] Revisit MakeDecimal behavior on overflow

Author: mgaido91

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

@@ -19,6 +19,7 @@ package org.apache.spark.sql.catalyst.expressions
 
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext, EmptyBlock, ExprCode}
+import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types._
 
 /**
@@ -46,19 +47,35 @@ case class UnscaledValue(child: Expression) extends UnaryExpression {
  */
 case class MakeDecimal(child: Expression, precision: Int, scale: Int) extends UnaryExpression {
 
+  private val nullOnOverflow = SQLConf.get.decimalOperationsNullOnOverflow
+  private lazy val doEval = if (nullOnOverflow) {
+    input: Long => new Decimal().setOrNull(input, precision, scale)
+  } else {
+    input: Long => new Decimal().set(input, precision, scale)
+  }
+
   override def dataType: DataType = DecimalType(precision, scale)
-  override def nullable: Boolean = true
+  override def nullable: Boolean = child.nullable || nullOnOverflow
   override def toString: String = s"MakeDecimal($child,$precision,$scale)"
 
-  protected override def nullSafeEval(input: Any): Any =
-    Decimal(input.asInstanceOf[Long], precision, scale)
+  protected override def nullSafeEval(input: Any): Any = doEval(input.asInstanceOf[Long])
 
   override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
     nullSafeCodeGen(ctx, ev, eval => {
+      val setMethod = if (nullOnOverflow) {
+        "setOrNull"
+      } else {
+        "set"
+      }
+      val setNull = if (nullable) {
+        s"${ev.isNull} = ${ev.value} == null;"
+      } else {
+        ""
+      }
       s"""
-        ${ev.value} = (new Decimal()).setOrNull($eval, $precision, $scale);
-        ${ev.isNull} = ${ev.value} == null;
-      """
+         |${ev.value} = (new Decimal()).$setMethod($eval, $precision, $scale);
+         |$setNull
+         |""".stripMargin
     })
   }
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/types/Decimal.scala

@@ -76,7 +76,7 @@ final class Decimal extends Ordered[Decimal] with Serializable {
    */
   def set(unscaled: Long, precision: Int, scale: Int): Decimal = {
     if (setOrNull(unscaled, precision, scale) == null) {
-      throw new IllegalArgumentException("Unscaled value too large for precision")
+      throw new ArithmeticException("Unscaled value too large for precision")
     }
     this
   }

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

@@ -18,6 +18,7 @@
 package org.apache.spark.sql.catalyst.expressions
 
 import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types.{Decimal, DecimalType, LongType}
 
 class DecimalExpressionSuite extends SparkFunSuite with ExpressionEvalHelper {
@@ -31,8 +32,24 @@ class DecimalExpressionSuite extends SparkFunSuite with ExpressionEvalHelper {
   }
 
   test("MakeDecimal") {
-    checkEvaluation(MakeDecimal(Literal(101L), 3, 1), Decimal("10.1"))
-    checkEvaluation(MakeDecimal(Literal.create(null, LongType), 3, 1), null)
+    withSQLConf(SQLConf.DECIMAL_OPERATIONS_NULL_ON_OVERFLOW.key -> "true") {
+      checkEvaluation(MakeDecimal(Literal(101L), 3, 1), Decimal("10.1"))
+      checkEvaluation(MakeDecimal(Literal.create(null, LongType), 3, 1), null)
+      val overflowExpr = MakeDecimal(Literal.create(1000L, LongType), 3, 1)
+      checkEvaluation(overflowExpr, null)
+      checkEvaluationWithMutableProjection(overflowExpr, null)
+      evaluateWithoutCodegen(overflowExpr, null)
+      checkEvaluationWithUnsafeProjection(overflowExpr, null)
+    }
+    withSQLConf(SQLConf.DECIMAL_OPERATIONS_NULL_ON_OVERFLOW.key -> "false") {
+      checkEvaluation(MakeDecimal(Literal(101L), 3, 1), Decimal("10.1"))
+      checkEvaluation(MakeDecimal(Literal.create(null, LongType), 3, 1), null)
+      val overflowExpr = MakeDecimal(Literal.create(1000L, LongType), 3, 1)
+      intercept[ArithmeticException](checkEvaluationWithMutableProjection(overflowExpr, null))
+      intercept[ArithmeticException](evaluateWithoutCodegen(overflowExpr, null))
+      intercept[ArithmeticException](checkEvaluationWithUnsafeProjection(overflowExpr, null))
+    }
+
   }
 
   test("PromotePrecision") {