DRILL-6094: Decimal data type enhancements

Add ExprVisitors for VARDECIMAL
Modify writers/readers to support VARDECIMAL
- Added usage of VarDecimal for parquet, hive, maprdb, jdbc;
- Added options to store decimals as int32 and int64 or fixed_len_byte_array or binary;
Add UDFs for VARDECIMAL data type
- modify type inference rules
- remove UDFs for obsolete DECIMAL types
Enable DECIMAL data type by default
Add unit tests for DECIMAL data type
Fix mapping for NLJ when literal with non-primitive type is used in join conditions
Refresh protobuf C++ source files
Changes in C++ files
Add support for decimal logical type in Avro.
Add support for date, time and timestamp logical types.
Update Avro version to 1.8.2.

1 files changed, 53 insertions, 31 deletions

diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcRecordReader.java b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcRecordReader.java
index bea99c71a..5f3f7131b 100755
--- a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcRecordReader.java
+++ b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcRecordReader.java
@@ -35,9 +35,9 @@ import javax.sql.DataSource;
 import org.apache.drill.common.AutoCloseables;
 import org.apache.drill.common.exceptions.ExecutionSetupException;
 import org.apache.drill.common.exceptions.UserException;
+import org.apache.drill.common.types.TypeProtos;
 import org.apache.drill.common.types.TypeProtos.MajorType;
 import org.apache.drill.common.types.TypeProtos.MinorType;
-import org.apache.drill.common.types.Types;
 import org.apache.drill.exec.exception.SchemaChangeException;
 import org.apache.drill.exec.expr.TypeHelper;
 import org.apache.drill.exec.ops.OperatorContext;
@@ -54,6 +54,7 @@ import org.apache.drill.exec.vector.NullableTimeStampVector;
 import org.apache.drill.exec.vector.NullableTimeVector;
 import org.apache.drill.exec.vector.NullableVarBinaryVector;
 import org.apache.drill.exec.vector.NullableVarCharVector;
+import org.apache.drill.exec.vector.NullableVarDecimalVector;
 import org.apache.drill.exec.vector.ValueVector;
 
 import com.google.common.base.Charsets;
@@ -104,7 +105,7 @@ class JdbcRecordReader extends AbstractRecordReader {
         .put(java.sql.Types.BLOB, MinorType.VARBINARY)
 
         .put(java.sql.Types.NUMERIC, MinorType.FLOAT8)
-        .put(java.sql.Types.DECIMAL, MinorType.FLOAT8)
+        .put(java.sql.Types.DECIMAL, MinorType.VARDECIMAL)
         .put(java.sql.Types.REAL, MinorType.FLOAT8)
 
         .put(java.sql.Types.DATE, MinorType.DATE)
@@ -137,29 +138,45 @@ class JdbcRecordReader extends AbstractRecordReader {
 
   private Copier<?> getCopier(int jdbcType, int offset, ResultSet result, ValueVector v) {
 
-    if (v instanceof NullableBigIntVector) {
-      return new BigIntCopier(offset, result, (NullableBigIntVector.Mutator) v.getMutator());
-    } else if (v instanceof NullableFloat4Vector) {
-      return new Float4Copier(offset, result, (NullableFloat4Vector.Mutator) v.getMutator());
-    } else if (v instanceof NullableFloat8Vector) {
-      return new Float8Copier(offset, result, (NullableFloat8Vector.Mutator) v.getMutator());
-    } else if (v instanceof NullableIntVector) {
-      return new IntCopier(offset, result, (NullableIntVector.Mutator) v.getMutator());
-    } else if (v instanceof NullableVarCharVector) {
-      return new VarCharCopier(offset, result, (NullableVarCharVector.Mutator) v.getMutator());
-    } else if (v instanceof NullableVarBinaryVector) {
-      return new VarBinaryCopier(offset, result, (NullableVarBinaryVector.Mutator) v.getMutator());
-    } else if (v instanceof NullableDateVector) {
-      return new DateCopier(offset, result, (NullableDateVector.Mutator) v.getMutator());
-    } else if (v instanceof NullableTimeVector) {
-      return new TimeCopier(offset, result, (NullableTimeVector.Mutator) v.getMutator());
-    } else if (v instanceof NullableTimeStampVector) {
-      return new TimeStampCopier(offset, result, (NullableTimeStampVector.Mutator) v.getMutator());
-    } else if (v instanceof NullableBitVector) {
-      return new BitCopier(offset, result, (NullableBitVector.Mutator) v.getMutator());
+    switch (jdbcType) {
+      case java.sql.Types.BIGINT:
+        return new BigIntCopier(offset, result, (NullableBigIntVector.Mutator) v.getMutator());
+      case java.sql.Types.FLOAT:
+        return new Float4Copier(offset, result, (NullableFloat4Vector.Mutator) v.getMutator());
+      case java.sql.Types.DOUBLE:
+      case java.sql.Types.NUMERIC:
+      case java.sql.Types.REAL:
+        return new Float8Copier(offset, result, (NullableFloat8Vector.Mutator) v.getMutator());
+      case java.sql.Types.TINYINT:
+      case java.sql.Types.SMALLINT:
+      case java.sql.Types.INTEGER:
+        return new IntCopier(offset, result, (NullableIntVector.Mutator) v.getMutator());
+      case java.sql.Types.CHAR:
+      case java.sql.Types.VARCHAR:
+      case java.sql.Types.LONGVARCHAR:
+      case java.sql.Types.CLOB:
+      case java.sql.Types.NCHAR:
+      case java.sql.Types.NVARCHAR:
+      case java.sql.Types.LONGNVARCHAR:
+        return new VarCharCopier(offset, result, (NullableVarCharVector.Mutator) v.getMutator());
+      case java.sql.Types.VARBINARY:
+      case java.sql.Types.LONGVARBINARY:
+      case java.sql.Types.BLOB:
+        return new VarBinaryCopier(offset, result, (NullableVarBinaryVector.Mutator) v.getMutator());
+      case java.sql.Types.DATE:
+        return new DateCopier(offset, result, (NullableDateVector.Mutator) v.getMutator());
+      case java.sql.Types.TIME:
+        return new TimeCopier(offset, result, (NullableTimeVector.Mutator) v.getMutator());
+      case java.sql.Types.TIMESTAMP:
+        return new TimeStampCopier(offset, result, (NullableTimeStampVector.Mutator) v.getMutator());
+      case java.sql.Types.BOOLEAN:
+      case java.sql.Types.BIT:
+        return new BitCopier(offset, result, (NullableBitVector.Mutator) v.getMutator());
+      case java.sql.Types.DECIMAL:
+        return new DecimalCopier(offset, result, (NullableVarDecimalVector.Mutator) v.getMutator());
+      default:
+        throw new IllegalArgumentException("Unknown how to handle vector.");
     }
-
-    throw new IllegalArgumentException("Unknown how to handle vector.");
   }
 
   @Override
@@ -197,9 +214,14 @@ class JdbcRecordReader extends AbstractRecordReader {
           continue;
         }
 
-        final MajorType type = Types.optional(minorType);
+        final MajorType type = MajorType.newBuilder()
+            .setMode(TypeProtos.DataMode.OPTIONAL)
+            .setMinorType(minorType)
+            .setScale(scale)
+            .setPrecision(width)
+            .build();
         final MaterializedField field = MaterializedField.create(name, type);
-        final Class<? extends ValueVector> clazz = (Class<? extends ValueVector>) TypeHelper.getValueVectorClass(
+        final Class<? extends ValueVector> clazz = TypeHelper.getValueVectorClass(
             minorType, type.getMode());
         ValueVector vector = output.addField(field, clazz);
         vectorBuilder.add(vector);
@@ -225,10 +247,10 @@ class JdbcRecordReader extends AbstractRecordReader {
     int counter = 0;
     Boolean b = true;
     try {
-      while (counter < 4095 && b == true) { // loop at 4095 since nullables use one more than record count and we
+      while (counter < 4095 && b) { // loop at 4095 since nullables use one more than record count and we
                                             // allocate on powers of two.
         b = resultSet.next();
-        if(b == false) {
+        if (!b) {
             break;
         }
         for (Copier<?> c : copiers) {
@@ -335,9 +357,9 @@ class JdbcRecordReader extends AbstractRecordReader {
 
   }
 
-  private class DecimalCopier extends Copier<NullableFloat8Vector.Mutator> {
+  private class DecimalCopier extends Copier<NullableVarDecimalVector.Mutator> {
 
-    public DecimalCopier(int columnIndex, ResultSet result, NullableFloat8Vector.Mutator mutator) {
+    public DecimalCopier(int columnIndex, ResultSet result, NullableVarDecimalVector.Mutator mutator) {
       super(columnIndex, result, mutator);
     }
 
@@ -345,7 +367,7 @@ class JdbcRecordReader extends AbstractRecordReader {
     void copy(int index) throws SQLException {
       BigDecimal decimal = result.getBigDecimal(columnIndex);
       if (decimal != null) {
-        mutator.setSafe(index, decimal.doubleValue());
+        mutator.setSafe(index, decimal);
       }
     }