Only resorts to FIXED_DECIMAL when the value can be held in a long

Author: liancheng

sql/core/src/main/scala/org/apache/spark/sql/columnar/ColumnAccessor.scala

@@ -129,7 +129,7 @@ private[sql] object ColumnAccessor {
       case BinaryType => new BinaryColumnAccessor(dup)
       case DateType => new DateColumnAccessor(dup)
       case TimestampType => new TimestampColumnAccessor(dup)
-      case DecimalType.Fixed(precision, scale) =>
+      case DecimalType.Fixed(precision, scale) if precision < 19 =>
         new FixedDecimalColumnAccessor(dup, precision, scale)
       case _ => new GenericColumnAccessor(dup)
     }

sql/core/src/main/scala/org/apache/spark/sql/columnar/ColumnBuilder.scala

@@ -161,7 +161,7 @@ private[sql] object ColumnBuilder {
       case BinaryType => new BinaryColumnBuilder
       case DateType => new DateColumnBuilder
       case TimestampType => new TimestampColumnBuilder
-      case DecimalType.Fixed(precision, scale) =>
+      case DecimalType.Fixed(precision, scale) if precision < 19 =>
         new FixedDecimalColumnBuilder(precision, scale)
       case _ => new GenericColumnBuilder
     }

sql/core/src/main/scala/org/apache/spark/sql/columnar/ColumnType.scala

@@ -454,7 +454,8 @@ private[sql] object ColumnType {
       case BinaryType => BINARY
       case DateType => DATE
       case TimestampType => TIMESTAMP
-      case DecimalType.Fixed(precision, scale) => FIXED_DECIMAL(precision, scale)
+      case DecimalType.Fixed(precision, scale) if precision < 19 =>
+        FIXED_DECIMAL(precision, scale)
       case _ => GENERIC
     }
   }

sql/core/src/test/scala/org/apache/spark/sql/columnar/ColumnTypeSuite.scala

@@ -216,4 +216,16 @@ class ColumnTypeSuite extends FunSuite with Logging {
     if (sb.nonEmpty) sb.setLength(sb.length - 1)
     sb.toString()
   }
+
+  test("column type for decimal types with different precision") {
+    (1 to 18).foreach { i =>
+      assertResult(FIXED_DECIMAL(i, 0)) {
+        ColumnType(DecimalType(i, 0))
+      }
+    }
+
+    assertResult(GENERIC) {
+      ColumnType(DecimalType(19, 0))
+    }
+  }
 }