overflow check at expression encoder

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/SerializerBuildHelper.scala

@@ -17,14 +17,17 @@
 
 package org.apache.spark.sql.catalyst
 
-import org.apache.spark.sql.catalyst.expressions.{CreateNamedStruct, Expression, IsNull, UnsafeArrayData}
+import org.apache.spark.sql.catalyst.expressions.{CheckOverflow, CreateNamedStruct, Expression, IsNull, UnsafeArrayData}
 import org.apache.spark.sql.catalyst.expressions.objects._
 import org.apache.spark.sql.catalyst.util.{DateTimeUtils, GenericArrayData}
+import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types._
 import org.apache.spark.unsafe.types.UTF8String
 
 object SerializerBuildHelper {
 
+  private def nullOnOverflow: Boolean = SQLConf.get.decimalOperationsNullOnOverflow
+
   def createSerializerForBoolean(inputObject: Expression): Expression = {
     Invoke(inputObject, "booleanValue", BooleanType)
   }
@@ -99,25 +102,25 @@ object SerializerBuildHelper {
   }
 
   def createSerializerForJavaBigDecimal(inputObject: Expression): Expression = {
-    StaticInvoke(
+    CheckOverflow(StaticInvoke(
       Decimal.getClass,
       DecimalType.SYSTEM_DEFAULT,
       "apply",
       inputObject :: Nil,
-      returnNullable = false)
+      returnNullable = false), DecimalType.SYSTEM_DEFAULT, nullOnOverflow)
   }
 
   def createSerializerForScalaBigDecimal(inputObject: Expression): Expression = {
     createSerializerForJavaBigDecimal(inputObject)
   }
 
   def createSerializerForJavaBigInteger(inputObject: Expression): Expression = {
-    StaticInvoke(
+    CheckOverflow(StaticInvoke(
       Decimal.getClass,
       DecimalType.BigIntDecimal,
       "apply",
       inputObject :: Nil,
-      returnNullable = false)
+      returnNullable = false), DecimalType.BigIntDecimal, nullOnOverflow)
   }
 
   def createSerializerForScalaBigInt(inputObject: Expression): Expression = {