From 449b69c0804622cc747d280d28415e93144ca272 Mon Sep 17 00:00:00 2001
From: Marco Gaido <marcogaido91@gmail.com>
Date: Mon, 22 Jan 2018 15:32:04 +0100
Subject: [PATCH 1/8] [SPARK-23179][SQL] Support option to throw exception if
 overflow occurs

---
 .../catalyst/analysis/DecimalPrecision.scala  |  14 +-
 .../analysis/StreamingJoinHelper.scala        |   2 +-
 .../expressions/decimalExpressions.scala      |  24 ++-
 .../apache/spark/sql/internal/SQLConf.scala   |  12 ++
 .../org/apache/spark/sql/types/Decimal.scala  |  21 +-
 .../expressions/DecimalExpressionSuite.scala  |  27 ++-
 .../decimalArithmeticOperations.sql           |  27 ++-
 .../decimalArithmeticOperations.sql.out       | 185 ++++++++++++++----
 8 files changed, 250 insertions(+), 62 deletions(-)
 rename sql/core/src/test/resources/sql-tests/inputs/{typeCoercion/native => }/decimalArithmeticOperations.sql (77%)
 rename sql/core/src/test/resources/sql-tests/results/{typeCoercion/native => }/decimalArithmeticOperations.sql.out (53%)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DecimalPrecision.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DecimalPrecision.scala
index ab63131b07573..48a8a0976bcd1 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DecimalPrecision.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DecimalPrecision.scala
@@ -82,6 +82,8 @@ object DecimalPrecision extends TypeCoercionRule {
     PromotePrecision(Cast(e, dataType))
   }
 
+  private def nullOnOverflow: Boolean = SQLConf.get.decimalOperationsNullOnOverflow
+
   override protected def coerceTypes(plan: LogicalPlan): LogicalPlan = plan transformUp {
     // fix decimal precision for expressions
     case q => q.transformExpressionsUp(
@@ -105,7 +107,7 @@ object DecimalPrecision extends TypeCoercionRule {
         DecimalType.bounded(max(p1 - s1, p2 - s2) + resultScale + 1, resultScale)
       }
       CheckOverflow(Add(promotePrecision(e1, resultType), promotePrecision(e2, resultType)),
-        resultType)
+        resultType, nullOnOverflow)
 
     case Subtract(e1 @ DecimalType.Expression(p1, s1), e2 @ DecimalType.Expression(p2, s2)) =>
       val resultScale = max(s1, s2)
@@ -116,7 +118,7 @@ object DecimalPrecision extends TypeCoercionRule {
         DecimalType.bounded(max(p1 - s1, p2 - s2) + resultScale + 1, resultScale)
       }
       CheckOverflow(Subtract(promotePrecision(e1, resultType), promotePrecision(e2, resultType)),
-        resultType)
+        resultType, nullOnOverflow)
 
     case Multiply(e1 @ DecimalType.Expression(p1, s1), e2 @ DecimalType.Expression(p2, s2)) =>
       val resultType = if (SQLConf.get.decimalOperationsAllowPrecisionLoss) {
@@ -126,7 +128,7 @@ object DecimalPrecision extends TypeCoercionRule {
       }
       val widerType = widerDecimalType(p1, s1, p2, s2)
       CheckOverflow(Multiply(promotePrecision(e1, widerType), promotePrecision(e2, widerType)),
-        resultType)
+        resultType, nullOnOverflow)
 
     case Divide(e1 @ DecimalType.Expression(p1, s1), e2 @ DecimalType.Expression(p2, s2)) =>
       val resultType = if (SQLConf.get.decimalOperationsAllowPrecisionLoss) {
@@ -148,7 +150,7 @@ object DecimalPrecision extends TypeCoercionRule {
       }
       val widerType = widerDecimalType(p1, s1, p2, s2)
       CheckOverflow(Divide(promotePrecision(e1, widerType), promotePrecision(e2, widerType)),
-        resultType)
+        resultType, nullOnOverflow)
 
     case Remainder(e1 @ DecimalType.Expression(p1, s1), e2 @ DecimalType.Expression(p2, s2)) =>
       val resultType = if (SQLConf.get.decimalOperationsAllowPrecisionLoss) {
@@ -159,7 +161,7 @@ object DecimalPrecision extends TypeCoercionRule {
       // resultType may have lower precision, so we cast them into wider type first.
       val widerType = widerDecimalType(p1, s1, p2, s2)
       CheckOverflow(Remainder(promotePrecision(e1, widerType), promotePrecision(e2, widerType)),
-        resultType)
+        resultType, nullOnOverflow)
 
     case Pmod(e1 @ DecimalType.Expression(p1, s1), e2 @ DecimalType.Expression(p2, s2)) =>
       val resultType = if (SQLConf.get.decimalOperationsAllowPrecisionLoss) {
@@ -170,7 +172,7 @@ object DecimalPrecision extends TypeCoercionRule {
       // resultType may have lower precision, so we cast them into wider type first.
       val widerType = widerDecimalType(p1, s1, p2, s2)
       CheckOverflow(Pmod(promotePrecision(e1, widerType), promotePrecision(e2, widerType)),
-        resultType)
+        resultType, nullOnOverflow)
 
     case b @ BinaryComparison(e1 @ DecimalType.Expression(p1, s1),
     e2 @ DecimalType.Expression(p2, s2)) if p1 != p2 || s1 != s2 =>
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/StreamingJoinHelper.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/StreamingJoinHelper.scala
index 7a0aa08289efa..c7b79df4035cd 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/StreamingJoinHelper.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/StreamingJoinHelper.scala
@@ -236,7 +236,7 @@ object StreamingJoinHelper extends PredicateHelper with Logging {
           collect(left, negate) ++ collect(right, !negate)
         case UnaryMinus(child) =>
           collect(child, !negate)
-        case CheckOverflow(child, _) =>
+        case CheckOverflow(child, _, _) =>
           collect(child, negate)
         case PromotePrecision(child) =>
           collect(child, negate)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/decimalExpressions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/decimalExpressions.scala
index db1579ba28671..2db610c7d560e 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/decimalExpressions.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/decimalExpressions.scala
@@ -80,30 +80,44 @@ case class PromotePrecision(child: Expression) extends UnaryExpression {
 
 /**
  * Rounds the decimal to given scale and check whether the decimal can fit in provided precision
- * or not, returns null if not.
+ * or not. If not, if `nullOnOverflow` is `true`, it returns `null`; otherwise an
+ * `ArithmeticException` is thrown.
  */
-case class CheckOverflow(child: Expression, dataType: DecimalType) extends UnaryExpression {
+case class CheckOverflow(
+    child: Expression,
+    dataType: DecimalType,
+    nullOnOverflow: Boolean) extends UnaryExpression {
 
   override def nullable: Boolean = true
 
   override def nullSafeEval(input: Any): Any =
-    input.asInstanceOf[Decimal].toPrecision(dataType.precision, dataType.scale)
+    input.asInstanceOf[Decimal].toPrecision(
+      dataType.precision,
+      dataType.scale,
+      Decimal.ROUND_HALF_UP,
+      nullOnOverflow)
 
   override protected def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
     nullSafeCodeGen(ctx, ev, eval => {
       val tmp = ctx.freshName("tmp")
+      val onOverflow = if (nullOnOverflow) {
+        s"${ev.isNull} = true"
+      } else {
+        s"""throw new ArithmeticException($tmp.toDebugString() + " cannot be represented as " +
+           | "Decimal(${dataType.precision}, ${dataType.scale}).")""".stripMargin
+      }
       s"""
          | Decimal $tmp = $eval.clone();
          | if ($tmp.changePrecision(${dataType.precision}, ${dataType.scale})) {
          |   ${ev.value} = $tmp;
          | } else {
-         |   ${ev.isNull} = true;
+         |   $onOverflow;
          | }
        """.stripMargin
     })
   }
 
-  override def toString: String = s"CheckOverflow($child, $dataType)"
+  override def toString: String = s"CheckOverflow($child, $dataType, $nullOnOverflow)"
 
   override def sql: String = child.sql
 }
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index cc4f4bf332459..1489de5e30a4a 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -1074,6 +1074,16 @@ object SQLConf {
       .booleanConf
       .createWithDefault(true)
 
+  val DECIMAL_OPERATIONS_NULL_ON_OVERFLOW =
+    buildConf("spark.sql.decimalOperations.nullOnOverflow")
+      .internal()
+      .doc("When true (default), if an overflow on a decimal occurs, then NULL is returned. " +
+        "Spark's older versions and Hive behave in this way. If turned to false, SQL ANSI 2011 " +
+        "specification, will be followed instead: an arithmetic exception is thrown. This is " +
+        "what most of the SQL databases do.")
+      .booleanConf
+      .createWithDefault(true)
+
   val SQL_STRING_REDACTION_PATTERN =
     ConfigBuilder("spark.sql.redaction.string.regex")
       .doc("Regex to decide which parts of strings produced by Spark contain sensitive " +
@@ -1453,6 +1463,8 @@ class SQLConf extends Serializable with Logging {
 
   def decimalOperationsAllowPrecisionLoss: Boolean = getConf(DECIMAL_OPERATIONS_ALLOW_PREC_LOSS)
 
+  def decimalOperationsNullOnOverflow: Boolean = getConf(DECIMAL_OPERATIONS_NULL_ON_OVERFLOW)
+
   def continuousStreamingExecutorQueueSize: Int = getConf(CONTINUOUS_STREAMING_EXECUTOR_QUEUE_SIZE)
 
   def continuousStreamingExecutorPollIntervalMs: Long =
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/types/Decimal.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/types/Decimal.scala
index 6da4f28b12962..a7db86c74ecca 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/types/Decimal.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/types/Decimal.scala
@@ -21,6 +21,7 @@ import java.lang.{Long => JLong}
 import java.math.{BigInteger, MathContext, RoundingMode}
 
 import org.apache.spark.annotation.InterfaceStability
+import org.apache.spark.internal.Logging
 import org.apache.spark.sql.AnalysisException
 
 /**
@@ -32,7 +33,7 @@ import org.apache.spark.sql.AnalysisException
  * - Otherwise, the decimal value is longVal / (10 ** _scale)
  */
 @InterfaceStability.Unstable
-final class Decimal extends Ordered[Decimal] with Serializable {
+final class Decimal extends Ordered[Decimal] with Serializable with Logging {
   import org.apache.spark.sql.types.Decimal._
 
   private var decimalVal: BigDecimal = null
@@ -237,14 +238,26 @@ final class Decimal extends Ordered[Decimal] with Serializable {
   /**
    * Create new `Decimal` with given precision and scale.
    *
-   * @return a non-null `Decimal` value if successful or `null` if overflow would occur.
+   * @return a non-null `Decimal` value if successful. Otherwise, if `nullOnOverflow` is true, null
+   *         is returned; if `nullOnOverflow` is false, an `ArithmeticException` is thrown.
    */
   private[sql] def toPrecision(
       precision: Int,
       scale: Int,
-      roundMode: BigDecimal.RoundingMode.Value = ROUND_HALF_UP): Decimal = {
+      roundMode: BigDecimal.RoundingMode.Value = ROUND_HALF_UP,
+      nullOnOverflow: Boolean = true): Decimal = {
     val copy = clone()
-    if (copy.changePrecision(precision, scale, roundMode)) copy else null
+    if (copy.changePrecision(precision, scale, roundMode)) {
+      copy
+    } else {
+      val message = s"$toDebugString cannot be represented as Decimal($precision, $scale)."
+      if (nullOnOverflow) {
+        logWarning(s"$message NULL is returned.")
+        null
+      } else {
+        throw new ArithmeticException(message)
+      }
+    }
   }
 
   /**
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/DecimalExpressionSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/DecimalExpressionSuite.scala
index a8f758d625a02..458191dbfd5d7 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/DecimalExpressionSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/DecimalExpressionSuite.scala
@@ -45,18 +45,27 @@ class DecimalExpressionSuite extends SparkFunSuite with ExpressionEvalHelper {
 
   test("CheckOverflow") {
     val d1 = Decimal("10.1")
-    checkEvaluation(CheckOverflow(Literal(d1), DecimalType(4, 0)), Decimal("10"))
-    checkEvaluation(CheckOverflow(Literal(d1), DecimalType(4, 1)), d1)
-    checkEvaluation(CheckOverflow(Literal(d1), DecimalType(4, 2)), d1)
-    checkEvaluation(CheckOverflow(Literal(d1), DecimalType(4, 3)), null)
+    checkEvaluation(CheckOverflow(Literal(d1), DecimalType(4, 0), true), Decimal("10"))
+    checkEvaluation(CheckOverflow(Literal(d1), DecimalType(4, 1), true), d1)
+    checkEvaluation(CheckOverflow(Literal(d1), DecimalType(4, 2), true), d1)
+    checkEvaluation(CheckOverflow(Literal(d1), DecimalType(4, 3), true), null)
+    intercept[ArithmeticException](CheckOverflow(Literal(d1), DecimalType(4, 3), false).eval())
+    intercept[ArithmeticException](checkEvaluationWithGeneratedMutableProjection(
+      CheckOverflow(Literal(d1), DecimalType(4, 3), false), null))
 
     val d2 = Decimal(101, 3, 1)
-    checkEvaluation(CheckOverflow(Literal(d2), DecimalType(4, 0)), Decimal("10"))
-    checkEvaluation(CheckOverflow(Literal(d2), DecimalType(4, 1)), d2)
-    checkEvaluation(CheckOverflow(Literal(d2), DecimalType(4, 2)), d2)
-    checkEvaluation(CheckOverflow(Literal(d2), DecimalType(4, 3)), null)
+    checkEvaluation(CheckOverflow(Literal(d2), DecimalType(4, 0), true), Decimal("10"))
+    checkEvaluation(CheckOverflow(Literal(d2), DecimalType(4, 1), true), d2)
+    checkEvaluation(CheckOverflow(Literal(d2), DecimalType(4, 2), true), d2)
+    checkEvaluation(CheckOverflow(Literal(d2), DecimalType(4, 3), true), null)
+    intercept[ArithmeticException](CheckOverflow(Literal(d2), DecimalType(4, 3), false).eval())
+    intercept[ArithmeticException](checkEvaluationWithGeneratedMutableProjection(
+      CheckOverflow(Literal(d2), DecimalType(4, 3), false), null))
 
-    checkEvaluation(CheckOverflow(Literal.create(null, DecimalType(2, 1)), DecimalType(3, 2)), null)
+    checkEvaluation(CheckOverflow(
+      Literal.create(null, DecimalType(2, 1)), DecimalType(3, 2), true), null)
+    checkEvaluation(CheckOverflow(
+      Literal.create(null, DecimalType(2, 1)), DecimalType(3, 2), false), null)
   }
 
 }
diff --git a/sql/core/src/test/resources/sql-tests/inputs/typeCoercion/native/decimalArithmeticOperations.sql b/sql/core/src/test/resources/sql-tests/inputs/decimalArithmeticOperations.sql
similarity index 77%
rename from sql/core/src/test/resources/sql-tests/inputs/typeCoercion/native/decimalArithmeticOperations.sql
rename to sql/core/src/test/resources/sql-tests/inputs/decimalArithmeticOperations.sql
index c6d8a49d4b93a..308595ebd6e61 100644
--- a/sql/core/src/test/resources/sql-tests/inputs/typeCoercion/native/decimalArithmeticOperations.sql
+++ b/sql/core/src/test/resources/sql-tests/inputs/decimalArithmeticOperations.sql
@@ -49,7 +49,6 @@ select 1e35 / 0.1;
 
 -- arithmetic operations causing a precision loss are truncated
 select 123456789123456789.1234567890 * 1.123456789123456789;
-select 0.001 / 9876543210987654321098765432109876543.2
 
 -- return NULL instead of rounding, according to old Spark versions' behavior
 set spark.sql.decimalOperations.allowPrecisionLoss=false;
@@ -75,6 +74,30 @@ select 1e35 / 0.1;
 
 -- arithmetic operations causing a precision loss return NULL
 select 123456789123456789.1234567890 * 1.123456789123456789;
-select 0.001 / 9876543210987654321098765432109876543.2
+
+-- throw an exception instead of returning NULL, according to SQL ANSI 2011
+set spark.sql.decimalOperations.nullOnOverflow=false;
+
+-- test decimal operations
+select id, a+b, a-b, a*b, a/b from decimals_test order by id;
+
+-- test operations between decimals and constants
+select id, a*10, b/10 from decimals_test order by id;
+
+-- test operations on constants
+select 10.3 * 3.0;
+select 10.3000 * 3.0;
+select 10.30000 * 30.0;
+select 10.300000000000000000 * 3.000000000000000000;
+select 10.300000000000000000 * 3.0000000000000000000;
+
+-- arithmetic operations causing an overflow throw exception
+select (5e36 + 0.1) + 5e36;
+select (-4e36 - 0.1) - 7e36;
+select 12345678901234567890.0 * 12345678901234567890.0;
+select 1e35 / 0.1;
+
+-- arithmetic operations causing a precision loss throw exception
+select 123456789123456789.1234567890 * 1.123456789123456789;
 
 drop table decimals_test;
diff --git a/sql/core/src/test/resources/sql-tests/results/typeCoercion/native/decimalArithmeticOperations.sql.out b/sql/core/src/test/resources/sql-tests/results/decimalArithmeticOperations.sql.out
similarity index 53%
rename from sql/core/src/test/resources/sql-tests/results/typeCoercion/native/decimalArithmeticOperations.sql.out
rename to sql/core/src/test/resources/sql-tests/results/decimalArithmeticOperations.sql.out
index 4d70fe19d539f..c2045d87dfe31 100644
--- a/sql/core/src/test/resources/sql-tests/results/typeCoercion/native/decimalArithmeticOperations.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/decimalArithmeticOperations.sql.out
@@ -1,5 +1,5 @@
 -- Automatically generated by SQLQueryTestSuite
--- Number of queries: 32
+-- Number of queries: 45
 
 
 -- !query 0
@@ -154,43 +154,33 @@ struct<(CAST(123456789123456789.1234567890 AS DECIMAL(36,18)) * CAST(1.123456789
 
 
 -- !query 18
-select 0.001 / 9876543210987654321098765432109876543.2
-
 set spark.sql.decimalOperations.allowPrecisionLoss=false
 -- !query 18 schema
-struct<>
+struct<key:string,value:string>
 -- !query 18 output
-org.apache.spark.sql.catalyst.parser.ParseException
-
-mismatched input 'spark' expecting <EOF>(line 3, pos 4)
-
-== SQL ==
-select 0.001 / 9876543210987654321098765432109876543.2
-
-set spark.sql.decimalOperations.allowPrecisionLoss=false
-----^^^
+spark.sql.decimalOperations.allowPrecisionLoss	false
 
 
 -- !query 19
 select id, a+b, a-b, a*b, a/b from decimals_test order by id
 -- !query 19 schema
-struct<id:int,(a + b):decimal(38,17),(a - b):decimal(38,17),(a * b):decimal(38,6),(a / b):decimal(38,6)>
+struct<id:int,(a + b):decimal(38,18),(a - b):decimal(38,18),(a * b):decimal(38,36),(a / b):decimal(38,18)>
 -- !query 19 output
-1	1099	-899	99900	0.1001
-2	24690.246	0	152402061.885129	1
-3	1234.2234567891011	-1233.9765432108989	152.358023	0.0001
-4	123456789123456790.12345678912345679	123456789123456787.87654321087654321	138698367904130467.515623	109890109097814272.043109
+1	1099	-899	NULL	0.1001001001001001
+2	24690.246	0	NULL	1
+3	1234.2234567891011	-1233.9765432108989	NULL	0.000100037913541123
+4	123456789123456790.123456789123456789	123456789123456787.876543210876543211	NULL	109890109097814272.043109406191131436
 
 
 -- !query 20
 select id, a*10, b/10 from decimals_test order by id
 -- !query 20 schema
-struct<id:int,(CAST(a AS DECIMAL(38,18)) * CAST(CAST(10 AS DECIMAL(2,0)) AS DECIMAL(38,18))):decimal(38,15),(CAST(b AS DECIMAL(38,18)) / CAST(CAST(10 AS DECIMAL(2,0)) AS DECIMAL(38,18))):decimal(38,18)>
+struct<id:int,(CAST(a AS DECIMAL(38,18)) * CAST(CAST(10 AS DECIMAL(2,0)) AS DECIMAL(38,18))):decimal(38,18),(CAST(b AS DECIMAL(38,18)) / CAST(CAST(10 AS DECIMAL(2,0)) AS DECIMAL(38,18))):decimal(38,19)>
 -- !query 20 output
 1	1000	99.9
 2	123451.23	1234.5123
 3	1.234567891011	123.41
-4	1234567891234567890	0.112345678912345679
+4	1234567891234567890	0.1123456789123456789
 
 
 -- !query 21
@@ -220,7 +210,7 @@ struct<(CAST(10.30000 AS DECIMAL(7,5)) * CAST(30.0 AS DECIMAL(7,5))):decimal(11,
 -- !query 24
 select 10.300000000000000000 * 3.000000000000000000
 -- !query 24 schema
-struct<(CAST(10.300000000000000000 AS DECIMAL(20,18)) * CAST(3.000000000000000000 AS DECIMAL(20,18))):decimal(38,34)>
+struct<(CAST(10.300000000000000000 AS DECIMAL(20,18)) * CAST(3.000000000000000000 AS DECIMAL(20,18))):decimal(38,36)>
 -- !query 24 output
 30.9
 
@@ -228,9 +218,9 @@ struct<(CAST(10.300000000000000000 AS DECIMAL(20,18)) * CAST(3.00000000000000000
 -- !query 25
 select 10.300000000000000000 * 3.0000000000000000000
 -- !query 25 schema
-struct<(CAST(10.300000000000000000 AS DECIMAL(21,19)) * CAST(3.0000000000000000000 AS DECIMAL(21,19))):decimal(38,34)>
+struct<(CAST(10.300000000000000000 AS DECIMAL(21,19)) * CAST(3.0000000000000000000 AS DECIMAL(21,19))):decimal(38,37)>
 -- !query 25 output
-30.9
+NULL
 
 
 -- !query 26
@@ -260,7 +250,7 @@ NULL
 -- !query 29
 select 1e35 / 0.1
 -- !query 29 schema
-struct<(CAST(1E+35 AS DECIMAL(37,1)) / CAST(0.1 AS DECIMAL(37,1))):decimal(38,6)>
+struct<(CAST(1E+35 AS DECIMAL(37,1)) / CAST(0.1 AS DECIMAL(37,1))):decimal(38,3)>
 -- !query 29 output
 NULL
 
@@ -268,24 +258,149 @@ NULL
 -- !query 30
 select 123456789123456789.1234567890 * 1.123456789123456789
 -- !query 30 schema
-struct<(CAST(123456789123456789.1234567890 AS DECIMAL(36,18)) * CAST(1.123456789123456789 AS DECIMAL(36,18))):decimal(38,18)>
+struct<(CAST(123456789123456789.1234567890 AS DECIMAL(36,18)) * CAST(1.123456789123456789 AS DECIMAL(36,18))):decimal(38,28)>
 -- !query 30 output
-138698367904130467.654320988515622621
+NULL
 
 
 -- !query 31
-select 0.001 / 9876543210987654321098765432109876543.2
-
-drop table decimals_test
+set spark.sql.decimalOperations.nullOnOverflow=false
 -- !query 31 schema
-struct<>
+struct<key:string,value:string>
 -- !query 31 output
-org.apache.spark.sql.catalyst.parser.ParseException
+spark.sql.decimalOperations.nullOnOverflow	false
+
+
+-- !query 32
+select id, a+b, a-b, a*b, a/b from decimals_test order by id
+-- !query 32 schema
+struct<>
+-- !query 32 output
+org.apache.spark.SparkException
+Job aborted due to stage failure: Task 1 in stage 255.0 failed 1 times, most recent failure: Lost task 1.0 in stage 255.0 (TID 9377, localhost, executor driver): java.lang.ArithmeticException: Decimal(expanded,99900.000000000000000000000000000000000,38,33}) cannot be represented as Decimal(38, 36).
+	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.processNext(Unknown Source)
+	at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
+	at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$9$$anon$1.hasNext(WholeStageCodegenExec.scala:461)
+	at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
+	at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
+	at org.apache.spark.util.random.SamplingUtils$.reservoirSampleAndCount(SamplingUtils.scala:41)
+	at org.apache.spark.RangePartitioner$$anonfun$12.apply(Partitioner.scala:300)
+	at org.apache.spark.RangePartitioner$$anonfun$12.apply(Partitioner.scala:298)
+	at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsWithIndex$1$$anonfun$apply$26.apply(RDD.scala:845)
+	at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsWithIndex$1$$anonfun$apply$26.apply(RDD.scala:845)
+	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
+	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
+	at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
+	at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
+	at org.apache.spark.scheduler.Task.run(Task.scala:109)
+	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:345)
+	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
+	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
+	at java.lang.Thread.run(Thread.java:748)
+
+Driver stacktrace:
+
+
+-- !query 33
+select id, a*10, b/10 from decimals_test order by id
+-- !query 33 schema
+struct<id:int,(CAST(a AS DECIMAL(38,18)) * CAST(CAST(10 AS DECIMAL(2,0)) AS DECIMAL(38,18))):decimal(38,18),(CAST(b AS DECIMAL(38,18)) / CAST(CAST(10 AS DECIMAL(2,0)) AS DECIMAL(38,18))):decimal(38,19)>
+-- !query 33 output
+1	1000	99.9
+2	123451.23	1234.5123
+3	1.234567891011	123.41
+4	1234567891234567890	0.1123456789123456789
+
+
+-- !query 34
+select 10.3 * 3.0
+-- !query 34 schema
+struct<(CAST(10.3 AS DECIMAL(3,1)) * CAST(3.0 AS DECIMAL(3,1))):decimal(6,2)>
+-- !query 34 output
+30.9
+
 
-mismatched input 'table' expecting <EOF>(line 3, pos 5)
+-- !query 35
+select 10.3000 * 3.0
+-- !query 35 schema
+struct<(CAST(10.3000 AS DECIMAL(6,4)) * CAST(3.0 AS DECIMAL(6,4))):decimal(9,5)>
+-- !query 35 output
+30.9
+
+
+-- !query 36
+select 10.30000 * 30.0
+-- !query 36 schema
+struct<(CAST(10.30000 AS DECIMAL(7,5)) * CAST(30.0 AS DECIMAL(7,5))):decimal(11,6)>
+-- !query 36 output
+309
+
+
+-- !query 37
+select 10.300000000000000000 * 3.000000000000000000
+-- !query 37 schema
+struct<(CAST(10.300000000000000000 AS DECIMAL(20,18)) * CAST(3.000000000000000000 AS DECIMAL(20,18))):decimal(38,36)>
+-- !query 37 output
+30.9
+
+
+-- !query 38
+select 10.300000000000000000 * 3.0000000000000000000
+-- !query 38 schema
+struct<>
+-- !query 38 output
+java.lang.ArithmeticException
+Decimal(expanded,30.900000000000000000000000000000000000,38,36}) cannot be represented as Decimal(38, 37).
+
+
+-- !query 39
+select (5e36 + 0.1) + 5e36
+-- !query 39 schema
+struct<>
+-- !query 39 output
+java.lang.ArithmeticException
+Decimal(expanded,10000000000000000000000000000000000000.1,39,1}) cannot be represented as Decimal(38, 1).
+
+
+-- !query 40
+select (-4e36 - 0.1) - 7e36
+-- !query 40 schema
+struct<>
+-- !query 40 output
+java.lang.ArithmeticException
+Decimal(expanded,-11000000000000000000000000000000000000.1,39,1}) cannot be represented as Decimal(38, 1).
 
-== SQL ==
-select 0.001 / 9876543210987654321098765432109876543.2
 
+-- !query 41
+select 12345678901234567890.0 * 12345678901234567890.0
+-- !query 41 schema
+struct<>
+-- !query 41 output
+java.lang.ArithmeticException
+Decimal(expanded,1.5241578753238836750190519987501905210E+38,38,-1}) cannot be represented as Decimal(38, 2).
+
+
+-- !query 42
+select 1e35 / 0.1
+-- !query 42 schema
+struct<>
+-- !query 42 output
+java.lang.ArithmeticException
+Decimal(expanded,1000000000000000000000000000000000000,37,0}) cannot be represented as Decimal(38, 3).
+
+
+-- !query 43
+select 123456789123456789.1234567890 * 1.123456789123456789
+-- !query 43 schema
+struct<>
+-- !query 43 output
+java.lang.ArithmeticException
+Decimal(expanded,138698367904130467.65432098851562262075,38,20}) cannot be represented as Decimal(38, 28).
+
+
+-- !query 44
 drop table decimals_test
------^^^
+-- !query 44 schema
+struct<>
+-- !query 44 output
+

From fcd665e4a23b21b4ff6e112687ef802ccbc23d0f Mon Sep 17 00:00:00 2001
From: Marco Gaido <marcogaido91@gmail.com>
Date: Mon, 22 Jan 2018 18:54:34 +0100
Subject: [PATCH 2/8] fix ut failures

---
 .../expressions/mathExpressions.scala         |  6 +-
 .../org/apache/spark/sql/types/Decimal.scala  |  4 +-
 .../expressions/DecimalExpressionSuite.scala  |  1 -
 .../inputs/decimalArithmeticOperations.sql    |  3 -
 .../decimalArithmeticOperations.sql.out       | 96 +++++++------------
 5 files changed, 39 insertions(+), 71 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/mathExpressions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/mathExpressions.scala
index d8dc0862f1141..3f11005a5ad1d 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/mathExpressions.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/mathExpressions.scala
@@ -1076,8 +1076,10 @@ abstract class RoundBase(child: Expression, scale: Expression,
     val evaluationCode = dataType match {
       case DecimalType.Fixed(_, s) =>
         s"""
-        ${ev.value} = ${ce.value}.toPrecision(${ce.value}.precision(), $s, Decimal.$modeStr());
-        ${ev.isNull} = ${ev.value} == null;"""
+           |${ev.value} = ${ce.value}.toPrecision(${ce.value}.precision(), $s,
+           |  Decimal.$modeStr(), true);
+           |${ev.isNull} = ${ev.value} == null;
+         """.stripMargin
       case ByteType =>
         if (_scale < 0) {
           s"""
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/types/Decimal.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/types/Decimal.scala
index a7db86c74ecca..d82f15eae495c 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/types/Decimal.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/types/Decimal.scala
@@ -250,9 +250,9 @@ final class Decimal extends Ordered[Decimal] with Serializable with Logging {
     if (copy.changePrecision(precision, scale, roundMode)) {
       copy
     } else {
-      val message = s"$toDebugString cannot be represented as Decimal($precision, $scale)."
+      def message = s"$toDebugString cannot be represented as Decimal($precision, $scale)."
       if (nullOnOverflow) {
-        logWarning(s"$message NULL is returned.")
+        if (log.isDebugEnabled) logDebug(s"$message NULL is returned.")
         null
       } else {
         throw new ArithmeticException(message)
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/DecimalExpressionSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/DecimalExpressionSuite.scala
index 458191dbfd5d7..89c0739b6318a 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/DecimalExpressionSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/DecimalExpressionSuite.scala
@@ -67,5 +67,4 @@ class DecimalExpressionSuite extends SparkFunSuite with ExpressionEvalHelper {
     checkEvaluation(CheckOverflow(
       Literal.create(null, DecimalType(2, 1)), DecimalType(3, 2), false), null)
   }
-
 }
diff --git a/sql/core/src/test/resources/sql-tests/inputs/decimalArithmeticOperations.sql b/sql/core/src/test/resources/sql-tests/inputs/decimalArithmeticOperations.sql
index 308595ebd6e61..a51f4dc549f28 100644
--- a/sql/core/src/test/resources/sql-tests/inputs/decimalArithmeticOperations.sql
+++ b/sql/core/src/test/resources/sql-tests/inputs/decimalArithmeticOperations.sql
@@ -78,9 +78,6 @@ select 123456789123456789.1234567890 * 1.123456789123456789;
 -- throw an exception instead of returning NULL, according to SQL ANSI 2011
 set spark.sql.decimalOperations.nullOnOverflow=false;
 
--- test decimal operations
-select id, a+b, a-b, a*b, a/b from decimals_test order by id;
-
 -- test operations between decimals and constants
 select id, a*10, b/10 from decimals_test order by id;
 
diff --git a/sql/core/src/test/resources/sql-tests/results/decimalArithmeticOperations.sql.out b/sql/core/src/test/resources/sql-tests/results/decimalArithmeticOperations.sql.out
index c2045d87dfe31..548bd5aa58c1d 100644
--- a/sql/core/src/test/resources/sql-tests/results/decimalArithmeticOperations.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/decimalArithmeticOperations.sql.out
@@ -1,5 +1,5 @@
 -- Automatically generated by SQLQueryTestSuite
--- Number of queries: 45
+-- Number of queries: 44
 
 
 -- !query 0
@@ -272,135 +272,105 @@ spark.sql.decimalOperations.nullOnOverflow	false
 
 
 -- !query 32
-select id, a+b, a-b, a*b, a/b from decimals_test order by id
+select id, a*10, b/10 from decimals_test order by id
 -- !query 32 schema
-struct<>
+struct<id:int,(CAST(a AS DECIMAL(38,18)) * CAST(CAST(10 AS DECIMAL(2,0)) AS DECIMAL(38,18))):decimal(38,18),(CAST(b AS DECIMAL(38,18)) / CAST(CAST(10 AS DECIMAL(2,0)) AS DECIMAL(38,18))):decimal(38,19)>
 -- !query 32 output
-org.apache.spark.SparkException
-Job aborted due to stage failure: Task 1 in stage 255.0 failed 1 times, most recent failure: Lost task 1.0 in stage 255.0 (TID 9377, localhost, executor driver): java.lang.ArithmeticException: Decimal(expanded,99900.000000000000000000000000000000000,38,33}) cannot be represented as Decimal(38, 36).
-	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.processNext(Unknown Source)
-	at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
-	at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$9$$anon$1.hasNext(WholeStageCodegenExec.scala:461)
-	at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
-	at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
-	at org.apache.spark.util.random.SamplingUtils$.reservoirSampleAndCount(SamplingUtils.scala:41)
-	at org.apache.spark.RangePartitioner$$anonfun$12.apply(Partitioner.scala:300)
-	at org.apache.spark.RangePartitioner$$anonfun$12.apply(Partitioner.scala:298)
-	at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsWithIndex$1$$anonfun$apply$26.apply(RDD.scala:845)
-	at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsWithIndex$1$$anonfun$apply$26.apply(RDD.scala:845)
-	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
-	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
-	at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
-	at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
-	at org.apache.spark.scheduler.Task.run(Task.scala:109)
-	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:345)
-	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
-	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
-	at java.lang.Thread.run(Thread.java:748)
-
-Driver stacktrace:
+1	1000	99.9
+2	123451.23	1234.5123
+3	1.234567891011	123.41
+4	1234567891234567890	0.1123456789123456789
 
 
 -- !query 33
-select id, a*10, b/10 from decimals_test order by id
+select 10.3 * 3.0
 -- !query 33 schema
-struct<id:int,(CAST(a AS DECIMAL(38,18)) * CAST(CAST(10 AS DECIMAL(2,0)) AS DECIMAL(38,18))):decimal(38,18),(CAST(b AS DECIMAL(38,18)) / CAST(CAST(10 AS DECIMAL(2,0)) AS DECIMAL(38,18))):decimal(38,19)>
+struct<(CAST(10.3 AS DECIMAL(3,1)) * CAST(3.0 AS DECIMAL(3,1))):decimal(6,2)>
 -- !query 33 output
-1	1000	99.9
-2	123451.23	1234.5123
-3	1.234567891011	123.41
-4	1234567891234567890	0.1123456789123456789
+30.9
 
 
 -- !query 34
-select 10.3 * 3.0
+select 10.3000 * 3.0
 -- !query 34 schema
-struct<(CAST(10.3 AS DECIMAL(3,1)) * CAST(3.0 AS DECIMAL(3,1))):decimal(6,2)>
+struct<(CAST(10.3000 AS DECIMAL(6,4)) * CAST(3.0 AS DECIMAL(6,4))):decimal(9,5)>
 -- !query 34 output
 30.9
 
 
 -- !query 35
-select 10.3000 * 3.0
+select 10.30000 * 30.0
 -- !query 35 schema
-struct<(CAST(10.3000 AS DECIMAL(6,4)) * CAST(3.0 AS DECIMAL(6,4))):decimal(9,5)>
+struct<(CAST(10.30000 AS DECIMAL(7,5)) * CAST(30.0 AS DECIMAL(7,5))):decimal(11,6)>
 -- !query 35 output
-30.9
+309
 
 
 -- !query 36
-select 10.30000 * 30.0
+select 10.300000000000000000 * 3.000000000000000000
 -- !query 36 schema
-struct<(CAST(10.30000 AS DECIMAL(7,5)) * CAST(30.0 AS DECIMAL(7,5))):decimal(11,6)>
+struct<(CAST(10.300000000000000000 AS DECIMAL(20,18)) * CAST(3.000000000000000000 AS DECIMAL(20,18))):decimal(38,36)>
 -- !query 36 output
-309
+30.9
 
 
 -- !query 37
-select 10.300000000000000000 * 3.000000000000000000
+select 10.300000000000000000 * 3.0000000000000000000
 -- !query 37 schema
-struct<(CAST(10.300000000000000000 AS DECIMAL(20,18)) * CAST(3.000000000000000000 AS DECIMAL(20,18))):decimal(38,36)>
+struct<>
 -- !query 37 output
-30.9
+java.lang.ArithmeticException
+Decimal(expanded,30.900000000000000000000000000000000000,38,36}) cannot be represented as Decimal(38, 37).
 
 
 -- !query 38
-select 10.300000000000000000 * 3.0000000000000000000
+select (5e36 + 0.1) + 5e36
 -- !query 38 schema
 struct<>
 -- !query 38 output
 java.lang.ArithmeticException
-Decimal(expanded,30.900000000000000000000000000000000000,38,36}) cannot be represented as Decimal(38, 37).
+Decimal(expanded,10000000000000000000000000000000000000.1,39,1}) cannot be represented as Decimal(38, 1).
 
 
 -- !query 39
-select (5e36 + 0.1) + 5e36
+select (-4e36 - 0.1) - 7e36
 -- !query 39 schema
 struct<>
 -- !query 39 output
 java.lang.ArithmeticException
-Decimal(expanded,10000000000000000000000000000000000000.1,39,1}) cannot be represented as Decimal(38, 1).
+Decimal(expanded,-11000000000000000000000000000000000000.1,39,1}) cannot be represented as Decimal(38, 1).
 
 
 -- !query 40
-select (-4e36 - 0.1) - 7e36
+select 12345678901234567890.0 * 12345678901234567890.0
 -- !query 40 schema
 struct<>
 -- !query 40 output
 java.lang.ArithmeticException
-Decimal(expanded,-11000000000000000000000000000000000000.1,39,1}) cannot be represented as Decimal(38, 1).
+Decimal(expanded,1.5241578753238836750190519987501905210E+38,38,-1}) cannot be represented as Decimal(38, 2).
 
 
 -- !query 41
-select 12345678901234567890.0 * 12345678901234567890.0
+select 1e35 / 0.1
 -- !query 41 schema
 struct<>
 -- !query 41 output
 java.lang.ArithmeticException
-Decimal(expanded,1.5241578753238836750190519987501905210E+38,38,-1}) cannot be represented as Decimal(38, 2).
+Decimal(expanded,1000000000000000000000000000000000000,37,0}) cannot be represented as Decimal(38, 3).
 
 
 -- !query 42
-select 1e35 / 0.1
+select 123456789123456789.1234567890 * 1.123456789123456789
 -- !query 42 schema
 struct<>
 -- !query 42 output
 java.lang.ArithmeticException
-Decimal(expanded,1000000000000000000000000000000000000,37,0}) cannot be represented as Decimal(38, 3).
+Decimal(expanded,138698367904130467.65432098851562262075,38,20}) cannot be represented as Decimal(38, 28).
 
 
 -- !query 43
-select 123456789123456789.1234567890 * 1.123456789123456789
+drop table decimals_test
 -- !query 43 schema
 struct<>
 -- !query 43 output
-java.lang.ArithmeticException
-Decimal(expanded,138698367904130467.65432098851562262075,38,20}) cannot be represented as Decimal(38, 28).
-
-
--- !query 44
-drop table decimals_test
--- !query 44 schema
-struct<>
--- !query 44 output
 

From 610a595bf61721c38edfaf29dcc161e363319423 Mon Sep 17 00:00:00 2001
From: Marco Gaido <marcogaido91@gmail.com>
Date: Tue, 23 Jan 2018 10:30:53 +0100
Subject: [PATCH 3/8] remove log

---
 .../src/main/scala/org/apache/spark/sql/types/Decimal.scala  | 5 ++---
 1 file changed, 2 insertions(+), 3 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/types/Decimal.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/types/Decimal.scala
index d82f15eae495c..ee6b025450bfb 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/types/Decimal.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/types/Decimal.scala
@@ -250,12 +250,11 @@ final class Decimal extends Ordered[Decimal] with Serializable with Logging {
     if (copy.changePrecision(precision, scale, roundMode)) {
       copy
     } else {
-      def message = s"$toDebugString cannot be represented as Decimal($precision, $scale)."
       if (nullOnOverflow) {
-        if (log.isDebugEnabled) logDebug(s"$message NULL is returned.")
         null
       } else {
-        throw new ArithmeticException(message)
+        throw new ArithmeticException(
+          s"$toDebugString cannot be represented as Decimal($precision, $scale).")
       }
     }
   }

From c73471d167c502a8d5a0eae93ec2afa23282879a Mon Sep 17 00:00:00 2001
From: Marco Gaido <marcogaido91@gmail.com>
Date: Tue, 23 Jan 2018 14:06:54 +0100
Subject: [PATCH 4/8] remove unneeded logging

---
 .../src/main/scala/org/apache/spark/sql/types/Decimal.scala    | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/types/Decimal.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/types/Decimal.scala
index ee6b025450bfb..c1051007bb9d5 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/types/Decimal.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/types/Decimal.scala
@@ -21,7 +21,6 @@ import java.lang.{Long => JLong}
 import java.math.{BigInteger, MathContext, RoundingMode}
 
 import org.apache.spark.annotation.InterfaceStability
-import org.apache.spark.internal.Logging
 import org.apache.spark.sql.AnalysisException
 
 /**
@@ -33,7 +32,7 @@ import org.apache.spark.sql.AnalysisException
  * - Otherwise, the decimal value is longVal / (10 ** _scale)
  */
 @InterfaceStability.Unstable
-final class Decimal extends Ordered[Decimal] with Serializable with Logging {
+final class Decimal extends Ordered[Decimal] with Serializable {
   import org.apache.spark.sql.types.Decimal._
 
   private var decimalVal: BigDecimal = null

From 2c8e2c75c4ef99155f205b71b60b92a8521318d1 Mon Sep 17 00:00:00 2001
From: Marco Gaido <marcogaido91@gmail.com>
Date: Tue, 23 Jan 2018 16:27:02 +0100
Subject: [PATCH 5/8] fix doc

---
 .../main/scala/org/apache/spark/sql/internal/SQLConf.scala    | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index 1489de5e30a4a..51ba54a005efa 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -1079,8 +1079,8 @@ object SQLConf {
       .internal()
       .doc("When true (default), if an overflow on a decimal occurs, then NULL is returned. " +
         "Spark's older versions and Hive behave in this way. If turned to false, SQL ANSI 2011 " +
-        "specification, will be followed instead: an arithmetic exception is thrown. This is " +
-        "what most of the SQL databases do.")
+        "specification will be followed instead: an arithmetic is thrown, as most of the SQL " +
+        "databases do.")
       .booleanConf
       .createWithDefault(true)
 

From d6bc7e9bd4176c229df09b37121388462a750a97 Mon Sep 17 00:00:00 2001
From: Marco Gaido <marcogaido91@gmail.com>
Date: Fri, 20 Apr 2018 18:18:14 +0200
Subject: [PATCH 6/8] address comment

---
 .../expressions/decimalExpressions.scala         | 16 +++-------------
 1 file changed, 3 insertions(+), 13 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/decimalExpressions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/decimalExpressions.scala
index 2db610c7d560e..e30bd0d8231ee 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/decimalExpressions.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/decimalExpressions.scala
@@ -99,20 +99,10 @@ case class CheckOverflow(
 
   override protected def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
     nullSafeCodeGen(ctx, ev, eval => {
-      val tmp = ctx.freshName("tmp")
-      val onOverflow = if (nullOnOverflow) {
-        s"${ev.isNull} = true"
-      } else {
-        s"""throw new ArithmeticException($tmp.toDebugString() + " cannot be represented as " +
-           | "Decimal(${dataType.precision}, ${dataType.scale}).")""".stripMargin
-      }
       s"""
-         | Decimal $tmp = $eval.clone();
-         | if ($tmp.changePrecision(${dataType.precision}, ${dataType.scale})) {
-         |   ${ev.value} = $tmp;
-         | } else {
-         |   $onOverflow;
-         | }
+         |${ev.value} = $eval.toPrecision(
+         |  ${dataType.precision}, ${dataType.scale}, Decimal.ROUND_HALF_UP, $nullOnOverflow);
+         |${ev.isNull} = ${ev.value} == null;
        """.stripMargin
     })
   }

From aa84034bd60413057738500564a9714dfa4b4192 Mon Sep 17 00:00:00 2001
From: Marco Gaido <marcogaido91@gmail.com>
Date: Sat, 21 Apr 2018 10:21:59 +0200
Subject: [PATCH 7/8] fix codegen

---
 .../spark/sql/catalyst/expressions/decimalExpressions.scala     | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/decimalExpressions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/decimalExpressions.scala
index e30bd0d8231ee..312e04cd297b7 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/decimalExpressions.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/decimalExpressions.scala
@@ -101,7 +101,7 @@ case class CheckOverflow(
     nullSafeCodeGen(ctx, ev, eval => {
       s"""
          |${ev.value} = $eval.toPrecision(
-         |  ${dataType.precision}, ${dataType.scale}, Decimal.ROUND_HALF_UP, $nullOnOverflow);
+         |  ${dataType.precision}, ${dataType.scale}, Decimal.ROUND_HALF_UP(), $nullOnOverflow);
          |${ev.isNull} = ${ev.value} == null;
        """.stripMargin
     })

From bc25c0dbc0e9df3adfc425de09ea1918199eb2e0 Mon Sep 17 00:00:00 2001
From: Marco Gaido <marcogaido91@gmail.com>
Date: Fri, 21 Jun 2019 23:46:56 +0200
Subject: [PATCH 8/8] fix compilation error

---
 .../org/apache/spark/sql/catalyst/encoders/RowEncoder.scala   | 2 +-
 .../main/scala/org/apache/spark/sql/internal/SQLConf.scala    | 4 ++--
 .../sql/catalyst/expressions/DecimalExpressionSuite.scala     | 4 ++--
 3 files changed, 5 insertions(+), 5 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/RowEncoder.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/RowEncoder.scala
index 3a06f8d0c50fc..afe8a23f8f150 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/RowEncoder.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/RowEncoder.scala
@@ -114,7 +114,7 @@ object RowEncoder {
         d,
         "fromDecimal",
         inputObject :: Nil,
-        returnNullable = false), d)
+        returnNullable = false), d, SQLConf.get.decimalOperationsNullOnOverflow)
 
     case StringType => createSerializerForString(inputObject)
 
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index 9e7fc06bcaaef..21ffa07f83ca4 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -1446,8 +1446,8 @@ object SQLConf {
       .internal()
       .doc("When true (default), if an overflow on a decimal occurs, then NULL is returned. " +
         "Spark's older versions and Hive behave in this way. If turned to false, SQL ANSI 2011 " +
-        "specification will be followed instead: an arithmetic is thrown, as most of the SQL " +
-        "databases do.")
+        "specification will be followed instead: an arithmetic exception is thrown, as most " +
+        "of the SQL databases do.")
       .booleanConf
       .createWithDefault(true)
 
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/DecimalExpressionSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/DecimalExpressionSuite.scala
index 89c0739b6318a..d14eceb480f5f 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/DecimalExpressionSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/DecimalExpressionSuite.scala
@@ -50,7 +50,7 @@ class DecimalExpressionSuite extends SparkFunSuite with ExpressionEvalHelper {
     checkEvaluation(CheckOverflow(Literal(d1), DecimalType(4, 2), true), d1)
     checkEvaluation(CheckOverflow(Literal(d1), DecimalType(4, 3), true), null)
     intercept[ArithmeticException](CheckOverflow(Literal(d1), DecimalType(4, 3), false).eval())
-    intercept[ArithmeticException](checkEvaluationWithGeneratedMutableProjection(
+    intercept[ArithmeticException](checkEvaluationWithMutableProjection(
       CheckOverflow(Literal(d1), DecimalType(4, 3), false), null))
 
     val d2 = Decimal(101, 3, 1)
@@ -59,7 +59,7 @@ class DecimalExpressionSuite extends SparkFunSuite with ExpressionEvalHelper {
     checkEvaluation(CheckOverflow(Literal(d2), DecimalType(4, 2), true), d2)
     checkEvaluation(CheckOverflow(Literal(d2), DecimalType(4, 3), true), null)
     intercept[ArithmeticException](CheckOverflow(Literal(d2), DecimalType(4, 3), false).eval())
-    intercept[ArithmeticException](checkEvaluationWithGeneratedMutableProjection(
+    intercept[ArithmeticException](checkEvaluationWithMutableProjection(
       CheckOverflow(Literal(d2), DecimalType(4, 3), false), null))
 
     checkEvaluation(CheckOverflow(