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.

19 files changed, 273 insertions, 174 deletions

diff --git a/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/json/CompareFunctionsProcessor.java b/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/json/CompareFunctionsProcessor.java
index 36e7309fe..341093947 100644
--- a/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/json/CompareFunctionsProcessor.java
+++ b/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/json/CompareFunctionsProcessor.java
@@ -31,12 +31,12 @@ import org.apache.drill.common.expression.ValueExpressions.LongExpression;
 import org.apache.drill.common.expression.ValueExpressions.QuotedString;
 import org.apache.drill.common.expression.ValueExpressions.TimeExpression;
 import org.apache.drill.common.expression.ValueExpressions.TimeStampExpression;
+import org.apache.drill.common.expression.ValueExpressions.VarDecimalExpression;
 import org.apache.drill.common.expression.visitors.AbstractExprVisitor;
 import org.joda.time.LocalTime;
 import org.ojai.Value;
 import org.ojai.types.ODate;
 import org.ojai.types.OTime;
-import org.ojai.types.OTimestamp;
 
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
@@ -173,6 +173,14 @@ class CompareFunctionsProcessor extends AbstractExprVisitor<Boolean, LogicalExpr
       return true;
     }
 
+    // MaprDB does not support decimals completely, therefore double value is used.
+    // See com.mapr.db.impl.ConditionImpl.is(FieldPath path, QueryCondition.Op op, BigDecimal value) method
+    if (valueArg instanceof VarDecimalExpression) {
+      this.value = KeyValueBuilder.initFrom(((VarDecimalExpression) valueArg).getBigDecimal().doubleValue());
+      this.path = path;
+      return true;
+    }
+
     if (valueArg instanceof TimeStampExpression) {
       // disable pushdown of TimeStampExpression type until bug 22824 is fixed.
       //
@@ -196,6 +204,7 @@ class CompareFunctionsProcessor extends AbstractExprVisitor<Boolean, LogicalExpr
         .add(LongExpression.class)
         .add(QuotedString.class)
         .add(TimeExpression.class)
+        .add(VarDecimalExpression.class)
         .build();
   }
 
diff --git a/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/json/MaprDBJsonRecordReader.java b/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/json/MaprDBJsonRecordReader.java
index 9f93e18b3..fde4d2850 100644
--- a/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/json/MaprDBJsonRecordReader.java
+++ b/contrib/format-maprdb/src/main/java/org/apache/drill/exec/store/mapr/db/json/MaprDBJsonRecordReader.java
@@ -304,7 +304,7 @@ public class MaprDBJsonRecordReader extends AbstractRecordReader {
           writeDouble(writer, fieldName, reader);
           break;
         case DECIMAL:
-          throw unsupportedError("Decimal type is currently not supported.");
+          writeDecimal(writer, fieldName, reader);
         case DATE:
           writeDate(writer, fieldName, reader);
           break;
@@ -365,6 +365,15 @@ public class MaprDBJsonRecordReader extends AbstractRecordReader {
     }
   }
 
+  private void writeDecimal(MapOrListWriterImpl writer, String fieldName, DBDocumentReaderBase reader) {
+    if (allTextMode) {
+      writeString(writer, fieldName, String.valueOf(reader.getDecimal()));
+    } else {
+      writer.varDecimal(fieldName, reader.getDecimalScale(), reader.getDecimalPrecision())
+          .writeVarDecimal(reader.getDecimal());
+    }
+  }
+
   private void writeFloat(MapOrListWriterImpl writer, String fieldName, DBDocumentReaderBase reader) {
     if (allTextMode) {
       writeString(writer, fieldName, String.valueOf(reader.getFloat()));
diff --git a/contrib/native/client/src/clientlib/decimalUtils.cpp b/contrib/native/client/src/clientlib/decimalUtils.cpp
index 6e26c5528..465eefeb5 100644
--- a/contrib/native/client/src/clientlib/decimalUtils.cpp
+++ b/contrib/native/client/src/clientlib/decimalUtils.cpp
@@ -95,6 +95,21 @@ DecimalValue getDecimalValueFromByteBuf(SlicedByteBuf& data, size_t startIndex,
     return val;
 }
 
+DecimalValue getDecimalValueFromByteBuf(SlicedByteBuf& data, size_t length, int scale) {
+
+    cpp_int decimalDigits;
+    // casts the first unsigned byte to signed to determine the sign of the value
+    decimalDigits = decimalDigits | cpp_int(static_cast<int8_t>(data.getByte(0))) << (length - 1) * 8;
+    for (int pos = length - 1; pos > 0; pos--) {
+        decimalDigits = decimalDigits | cpp_int(data.getByte(pos)) << (length - pos - 1) * 8;
+    }
+
+    DecimalValue val;
+    val.m_unscaledValue = decimalDigits;
+    val.m_scale = scale;
+    return val;
+}
+
 DecimalValue getDecimalValueFromDense(SlicedByteBuf& data, size_t startIndex, int nDecimalDigits, int scale, int maxPrecision, int width)
 {
     /* This method converts the dense representation to
diff --git a/contrib/native/client/src/clientlib/fieldmeta.cpp b/contrib/native/client/src/clientlib/fieldmeta.cpp
index 797b038eb..b48f3bc06 100644
--- a/contrib/native/client/src/clientlib/fieldmeta.cpp
+++ b/contrib/native/client/src/clientlib/fieldmeta.cpp
@@ -71,6 +71,7 @@ static const std::string& getSQLType(common::MinorType type, common::DataMode mo
     case common::DECIMAL28DENSE:
     case common::DECIMAL28SPARSE:
     case common::DECIMAL38DENSE:
+    case common::VARDECIMAL:
     case common::DECIMAL38SPARSE: return SQLDecimal;
 
     case common::VARCHAR:         return SQLVarchar;
@@ -133,6 +134,7 @@ static bool isSigned(common::MinorType type, common::DataMode mode) {
     case common::DECIMAL28DENSE:
     case common::DECIMAL38DENSE:
     case common::DECIMAL38SPARSE:
+    case common::VARDECIMAL:
 
     case common::INTERVALYEAR:
     case common::INTERVALDAY:
@@ -304,6 +306,7 @@ static uint32_t getDisplaySize(const ::common::MajorType& type) {
     case ::common::DECIMAL28SPARSE:
     case ::common::DECIMAL38DENSE:
     case ::common::DECIMAL38SPARSE:
+    case ::common::VARDECIMAL:
     case ::common::MONEY:           return 2 + precision; // precision of the column plus a sign and a decimal point
 
     case ::common::VARCHAR:
diff --git a/contrib/native/client/src/clientlib/metadata.cpp b/contrib/native/client/src/clientlib/metadata.cpp
index 637c83b33..0f1cf2eab 100644
--- a/contrib/native/client/src/clientlib/metadata.cpp
+++ b/contrib/native/client/src/clientlib/metadata.cpp
@@ -110,6 +110,7 @@ static const convert_support_set s_convertMap = boost::assign::list_of
 		(ConvertSupport(common::TINYINT, common::DECIMAL18))
 		(ConvertSupport(common::TINYINT, common::DECIMAL28SPARSE))
 		(ConvertSupport(common::TINYINT, common::DECIMAL38SPARSE))
+		(ConvertSupport(common::TINYINT, common::VARDECIMAL))
 		(ConvertSupport(common::TINYINT, common::DATE))
 		(ConvertSupport(common::TINYINT, common::TIME))
 		(ConvertSupport(common::TINYINT, common::TIMESTAMP))
@@ -128,6 +129,7 @@ static const convert_support_set s_convertMap = boost::assign::list_of
 		(ConvertSupport(common::SMALLINT, common::DECIMAL18))
 		(ConvertSupport(common::SMALLINT, common::DECIMAL28SPARSE))
 		(ConvertSupport(common::SMALLINT, common::DECIMAL38SPARSE))
+		(ConvertSupport(common::SMALLINT, common::VARDECIMAL))
 		(ConvertSupport(common::SMALLINT, common::DATE))
 		(ConvertSupport(common::SMALLINT, common::TIME))
 		(ConvertSupport(common::SMALLINT, common::TIMESTAMP))
@@ -146,6 +148,7 @@ static const convert_support_set s_convertMap = boost::assign::list_of
 		(ConvertSupport(common::INT, common::DECIMAL18))
 		(ConvertSupport(common::INT, common::DECIMAL28SPARSE))
 		(ConvertSupport(common::INT, common::DECIMAL38SPARSE))
+		(ConvertSupport(common::INT, common::VARDECIMAL))
 		(ConvertSupport(common::INT, common::DATE))
 		(ConvertSupport(common::INT, common::TIME))
 		(ConvertSupport(common::INT, common::TIMESTAMP))
@@ -164,6 +167,7 @@ static const convert_support_set s_convertMap = boost::assign::list_of
 		(ConvertSupport(common::BIGINT, common::DECIMAL18))
 		(ConvertSupport(common::BIGINT, common::DECIMAL28SPARSE))
 		(ConvertSupport(common::BIGINT, common::DECIMAL38SPARSE))
+		(ConvertSupport(common::BIGINT, common::VARDECIMAL))
 		(ConvertSupport(common::BIGINT, common::DATE))
 		(ConvertSupport(common::BIGINT, common::TIME))
 		(ConvertSupport(common::BIGINT, common::TIMESTAMP))
@@ -182,6 +186,7 @@ static const convert_support_set s_convertMap = boost::assign::list_of
 		(ConvertSupport(common::DECIMAL9, common::DECIMAL18))
 		(ConvertSupport(common::DECIMAL9, common::DECIMAL28SPARSE))
 		(ConvertSupport(common::DECIMAL9, common::DECIMAL38SPARSE))
+		(ConvertSupport(common::DECIMAL9, common::VARDECIMAL))
 		(ConvertSupport(common::DECIMAL9, common::DATE))
 		(ConvertSupport(common::DECIMAL9, common::TIME))
 		(ConvertSupport(common::DECIMAL9, common::TIMESTAMP))
@@ -200,6 +205,7 @@ static const convert_support_set s_convertMap = boost::assign::list_of
 		(ConvertSupport(common::DECIMAL18, common::DECIMAL18))
 		(ConvertSupport(common::DECIMAL18, common::DECIMAL28SPARSE))
 		(ConvertSupport(common::DECIMAL18, common::DECIMAL38SPARSE))
+		(ConvertSupport(common::DECIMAL18, common::VARDECIMAL))
 		(ConvertSupport(common::DECIMAL18, common::DATE))
 		(ConvertSupport(common::DECIMAL18, common::TIME))
 		(ConvertSupport(common::DECIMAL18, common::TIMESTAMP))
@@ -218,6 +224,7 @@ static const convert_support_set s_convertMap = boost::assign::list_of
 		(ConvertSupport(common::DECIMAL28SPARSE, common::DECIMAL18))
 		(ConvertSupport(common::DECIMAL28SPARSE, common::DECIMAL28SPARSE))
 		(ConvertSupport(common::DECIMAL28SPARSE, common::DECIMAL38SPARSE))
+		(ConvertSupport(common::DECIMAL28SPARSE, common::VARDECIMAL))
 		(ConvertSupport(common::DECIMAL28SPARSE, common::DATE))
 		(ConvertSupport(common::DECIMAL28SPARSE, common::TIME))
 		(ConvertSupport(common::DECIMAL28SPARSE, common::TIMESTAMP))
@@ -236,6 +243,7 @@ static const convert_support_set s_convertMap = boost::assign::list_of
 		(ConvertSupport(common::DECIMAL38SPARSE, common::DECIMAL18))
 		(ConvertSupport(common::DECIMAL38SPARSE, common::DECIMAL28SPARSE))
 		(ConvertSupport(common::DECIMAL38SPARSE, common::DECIMAL38SPARSE))
+		(ConvertSupport(common::DECIMAL38SPARSE, common::VARDECIMAL))
 		(ConvertSupport(common::DECIMAL38SPARSE, common::DATE))
 		(ConvertSupport(common::DECIMAL38SPARSE, common::TIME))
 		(ConvertSupport(common::DECIMAL38SPARSE, common::TIMESTAMP))
@@ -248,12 +256,32 @@ static const convert_support_set s_convertMap = boost::assign::list_of
 		(ConvertSupport(common::DECIMAL38SPARSE, common::VARBINARY))
 		(ConvertSupport(common::DECIMAL38SPARSE, common::INTERVALYEAR))
 		(ConvertSupport(common::DECIMAL38SPARSE, common::INTERVALDAY))
+		(ConvertSupport(common::VARDECIMAL, common::INT))
+		(ConvertSupport(common::VARDECIMAL, common::BIGINT))
+		(ConvertSupport(common::VARDECIMAL, common::DECIMAL9))
+		(ConvertSupport(common::VARDECIMAL, common::DECIMAL18))
+		(ConvertSupport(common::VARDECIMAL, common::DECIMAL28SPARSE))
+		(ConvertSupport(common::VARDECIMAL, common::DECIMAL38SPARSE))
+		(ConvertSupport(common::VARDECIMAL, common::VARDECIMAL))
+		(ConvertSupport(common::VARDECIMAL, common::DATE))
+		(ConvertSupport(common::VARDECIMAL, common::TIME))
+		(ConvertSupport(common::VARDECIMAL, common::TIMESTAMP))
+		(ConvertSupport(common::VARDECIMAL, common::INTERVAL))
+		(ConvertSupport(common::VARDECIMAL, common::FLOAT4))
+		(ConvertSupport(common::VARDECIMAL, common::FLOAT8))
+		(ConvertSupport(common::VARDECIMAL, common::BIT))
+		(ConvertSupport(common::VARDECIMAL, common::VARCHAR))
+		(ConvertSupport(common::VARDECIMAL, common::VAR16CHAR))
+		(ConvertSupport(common::VARDECIMAL, common::VARBINARY))
+		(ConvertSupport(common::VARDECIMAL, common::INTERVALYEAR))
+		(ConvertSupport(common::VARDECIMAL, common::INTERVALDAY))
 		(ConvertSupport(common::MONEY, common::INT))
 		(ConvertSupport(common::MONEY, common::BIGINT))
 		(ConvertSupport(common::MONEY, common::DECIMAL9))
 		(ConvertSupport(common::MONEY, common::DECIMAL18))
 		(ConvertSupport(common::MONEY, common::DECIMAL28SPARSE))
 		(ConvertSupport(common::MONEY, common::DECIMAL38SPARSE))
+		(ConvertSupport(common::MONEY, common::VARDECIMAL))
 		(ConvertSupport(common::MONEY, common::DATE))
 		(ConvertSupport(common::MONEY, common::TIME))
 		(ConvertSupport(common::MONEY, common::TIMESTAMP))
@@ -272,6 +300,7 @@ static const convert_support_set s_convertMap = boost::assign::list_of
 		(ConvertSupport(common::DATE, common::DECIMAL18))
 		(ConvertSupport(common::DATE, common::DECIMAL28SPARSE))
 		(ConvertSupport(common::DATE, common::DECIMAL38SPARSE))
+		(ConvertSupport(common::DATE, common::VARDECIMAL))
 		(ConvertSupport(common::DATE, common::DATE))
 		(ConvertSupport(common::DATE, common::TIME))
 		(ConvertSupport(common::DATE, common::TIMESTAMP))
@@ -290,6 +319,7 @@ static const convert_support_set s_convertMap = boost::assign::list_of
 		(ConvertSupport(common::TIME, common::DECIMAL18))
 		(ConvertSupport(common::TIME, common::DECIMAL28SPARSE))
 		(ConvertSupport(common::TIME, common::DECIMAL38SPARSE))
+		(ConvertSupport(common::TIME, common::VARDECIMAL))
 		(ConvertSupport(common::TIME, common::DATE))
 		(ConvertSupport(common::TIME, common::TIME))
 		(ConvertSupport(common::TIME, common::TIMESTAMP))
@@ -308,6 +338,7 @@ static const convert_support_set s_convertMap = boost::assign::list_of
 		(ConvertSupport(common::TIMESTAMPTZ, common::DECIMAL18))
 		(ConvertSupport(common::TIMESTAMPTZ, common::DECIMAL28SPARSE))
 		(ConvertSupport(common::TIMESTAMPTZ, common::DECIMAL38SPARSE))
+		(ConvertSupport(common::TIMESTAMPTZ, common::VARDECIMAL))
 		(ConvertSupport(common::TIMESTAMPTZ, common::DATE))
 		(ConvertSupport(common::TIMESTAMPTZ, common::TIME))
 		(ConvertSupport(common::TIMESTAMPTZ, common::TIMESTAMP))
@@ -326,6 +357,7 @@ static const convert_support_set s_convertMap = boost::assign::list_of
 		(ConvertSupport(common::TIMESTAMP, common::DECIMAL18))
 		(ConvertSupport(common::TIMESTAMP, common::DECIMAL28SPARSE))
 		(ConvertSupport(common::TIMESTAMP, common::DECIMAL38SPARSE))
+		(ConvertSupport(common::TIMESTAMP, common::VARDECIMAL))
 		(ConvertSupport(common::TIMESTAMP, common::DATE))
 		(ConvertSupport(common::TIMESTAMP, common::TIME))
 		(ConvertSupport(common::TIMESTAMP, common::TIMESTAMP))
@@ -344,6 +376,7 @@ static const convert_support_set s_convertMap = boost::assign::list_of
 		(ConvertSupport(common::INTERVAL, common::DECIMAL18))
 		(ConvertSupport(common::INTERVAL, common::DECIMAL28SPARSE))
 		(ConvertSupport(common::INTERVAL, common::DECIMAL38SPARSE))
+		(ConvertSupport(common::INTERVAL, common::VARDECIMAL))
 		(ConvertSupport(common::INTERVAL, common::DATE))
 		(ConvertSupport(common::INTERVAL, common::TIME))
 		(ConvertSupport(common::INTERVAL, common::TIMESTAMP))
@@ -362,6 +395,7 @@ static const convert_support_set s_convertMap = boost::assign::list_of
 		(ConvertSupport(common::FLOAT4, common::DECIMAL18))
 		(ConvertSupport(common::FLOAT4, common::DECIMAL28SPARSE))
 		(ConvertSupport(common::FLOAT4, common::DECIMAL38SPARSE))
+		(ConvertSupport(common::FLOAT4, common::VARDECIMAL))
 		(ConvertSupport(common::FLOAT4, common::DATE))
 		(ConvertSupport(common::FLOAT4, common::TIME))
 		(ConvertSupport(common::FLOAT4, common::TIMESTAMP))
@@ -380,6 +414,7 @@ static const convert_support_set s_convertMap = boost::assign::list_of
 		(ConvertSupport(common::FLOAT8, common::DECIMAL18))
 		(ConvertSupport(common::FLOAT8, common::DECIMAL28SPARSE))
 		(ConvertSupport(common::FLOAT8, common::DECIMAL38SPARSE))
+		(ConvertSupport(common::FLOAT8, common::VARDECIMAL))
 		(ConvertSupport(common::FLOAT8, common::DATE))
 		(ConvertSupport(common::FLOAT8, common::TIME))
 		(ConvertSupport(common::FLOAT8, common::TIMESTAMP))
@@ -399,6 +434,7 @@ static const convert_support_set s_convertMap = boost::assign::list_of
 		(ConvertSupport(common::BIT, common::DECIMAL18))
 		(ConvertSupport(common::BIT, common::DECIMAL28SPARSE))
 		(ConvertSupport(common::BIT, common::DECIMAL38SPARSE))
+		(ConvertSupport(common::BIT, common::VARDECIMAL))
 		(ConvertSupport(common::BIT, common::DATE))
 		(ConvertSupport(common::BIT, common::TIME))
 		(ConvertSupport(common::BIT, common::TIMESTAMP))
@@ -418,6 +454,7 @@ static const convert_support_set s_convertMap = boost::assign::list_of
 		(ConvertSupport(common::FIXEDCHAR, common::DECIMAL18))
 		(ConvertSupport(common::FIXEDCHAR, common::DECIMAL28SPARSE))
 		(ConvertSupport(common::FIXEDCHAR, common::DECIMAL38SPARSE))
+		(ConvertSupport(common::FIXEDCHAR, common::VARDECIMAL))
 		(ConvertSupport(common::FIXEDCHAR, common::DATE))
 		(ConvertSupport(common::FIXEDCHAR, common::TIME))
 		(ConvertSupport(common::FIXEDCHAR, common::TIMESTAMP))
@@ -437,6 +474,7 @@ static const convert_support_set s_convertMap = boost::assign::list_of
 		(ConvertSupport(common::FIXED16CHAR, common::DECIMAL18))
 		(ConvertSupport(common::FIXED16CHAR, common::DECIMAL28SPARSE))
 		(ConvertSupport(common::FIXED16CHAR, common::DECIMAL38SPARSE))
+		(ConvertSupport(common::FIXED16CHAR, common::VARDECIMAL))
 		(ConvertSupport(common::FIXED16CHAR, common::DATE))
 		(ConvertSupport(common::FIXED16CHAR, common::TIME))
 		(ConvertSupport(common::FIXED16CHAR, common::TIMESTAMP))
@@ -455,6 +493,7 @@ static const convert_support_set s_convertMap = boost::assign::list_of
 		(ConvertSupport(common::FIXEDBINARY, common::DECIMAL18))
 		(ConvertSupport(common::FIXEDBINARY, common::DECIMAL28SPARSE))
 		(ConvertSupport(common::FIXEDBINARY, common::DECIMAL38SPARSE))
+		(ConvertSupport(common::FIXEDBINARY, common::VARDECIMAL))
 		(ConvertSupport(common::FIXEDBINARY, common::DATE))
 		(ConvertSupport(common::FIXEDBINARY, common::TIME))
 		(ConvertSupport(common::FIXEDBINARY, common::TIMESTAMP))
@@ -474,6 +513,7 @@ static const convert_support_set s_convertMap = boost::assign::list_of
 		(ConvertSupport(common::VARCHAR, common::DECIMAL18))
 		(ConvertSupport(common::VARCHAR, common::DECIMAL28SPARSE))
 		(ConvertSupport(common::VARCHAR, common::DECIMAL38SPARSE))
+		(ConvertSupport(common::VARCHAR, common::VARDECIMAL))
 		(ConvertSupport(common::VARCHAR, common::DATE))
 		(ConvertSupport(common::VARCHAR, common::TIME))
 		(ConvertSupport(common::VARCHAR, common::TIMESTAMP))
@@ -493,6 +533,7 @@ static const convert_support_set s_convertMap = boost::assign::list_of
 		(ConvertSupport(common::VAR16CHAR, common::DECIMAL18))
 		(ConvertSupport(common::VAR16CHAR, common::DECIMAL28SPARSE))
 		(ConvertSupport(common::VAR16CHAR, common::DECIMAL38SPARSE))
+		(ConvertSupport(common::VAR16CHAR, common::VARDECIMAL))
 		(ConvertSupport(common::VAR16CHAR, common::DATE))
 		(ConvertSupport(common::VAR16CHAR, common::TIME))
 		(ConvertSupport(common::VAR16CHAR, common::TIMESTAMP))
@@ -511,6 +552,7 @@ static const convert_support_set s_convertMap = boost::assign::list_of
 		(ConvertSupport(common::VARBINARY, common::DECIMAL18))
 		(ConvertSupport(common::VARBINARY, common::DECIMAL28SPARSE))
 		(ConvertSupport(common::VARBINARY, common::DECIMAL38SPARSE))
+		(ConvertSupport(common::VARBINARY, common::VARDECIMAL))
 		(ConvertSupport(common::VARBINARY, common::DATE))
 		(ConvertSupport(common::VARBINARY, common::TIME))
 		(ConvertSupport(common::VARBINARY, common::TIMESTAMP))
@@ -529,6 +571,7 @@ static const convert_support_set s_convertMap = boost::assign::list_of
 		(ConvertSupport(common::UINT1, common::DECIMAL18))
 		(ConvertSupport(common::UINT1, common::DECIMAL28SPARSE))
 		(ConvertSupport(common::UINT1, common::DECIMAL38SPARSE))
+		(ConvertSupport(common::UINT1, common::VARDECIMAL))
 		(ConvertSupport(common::UINT1, common::DATE))
 		(ConvertSupport(common::UINT1, common::TIME))
 		(ConvertSupport(common::UINT1, common::TIMESTAMP))
@@ -547,6 +590,7 @@ static const convert_support_set s_convertMap = boost::assign::list_of
 		(ConvertSupport(common::UINT2, common::DECIMAL18))
 		(ConvertSupport(common::UINT2, common::DECIMAL28SPARSE))
 		(ConvertSupport(common::UINT2, common::DECIMAL38SPARSE))
+		(ConvertSupport(common::UINT2, common::VARDECIMAL))
 		(ConvertSupport(common::UINT2, common::DATE))
 		(ConvertSupport(common::UINT2, common::TIME))
 		(ConvertSupport(common::UINT2, common::TIMESTAMP))
@@ -565,6 +609,7 @@ static const convert_support_set s_convertMap = boost::assign::list_of
 		(ConvertSupport(common::UINT4, common::DECIMAL18))
 		(ConvertSupport(common::UINT4, common::DECIMAL28SPARSE))
 		(ConvertSupport(common::UINT4, common::DECIMAL38SPARSE))
+		(ConvertSupport(common::UINT4, common::VARDECIMAL))
 		(ConvertSupport(common::UINT4, common::DATE))
 		(ConvertSupport(common::UINT4, common::TIME))
 		(ConvertSupport(common::UINT4, common::TIMESTAMP))
@@ -583,6 +628,7 @@ static const convert_support_set s_convertMap = boost::assign::list_of
 		(ConvertSupport(common::UINT8, common::DECIMAL18))
 		(ConvertSupport(common::UINT8, common::DECIMAL28SPARSE))
 		(ConvertSupport(common::UINT8, common::DECIMAL38SPARSE))
+		(ConvertSupport(common::UINT8, common::VARDECIMAL))
 		(ConvertSupport(common::UINT8, common::DATE))
 		(ConvertSupport(common::UINT8, common::TIME))
 		(ConvertSupport(common::UINT8, common::TIMESTAMP))
@@ -601,6 +647,7 @@ static const convert_support_set s_convertMap = boost::assign::list_of
 		(ConvertSupport(common::DECIMAL28DENSE, common::DECIMAL18))
 		(ConvertSupport(common::DECIMAL28DENSE, common::DECIMAL28SPARSE))
 		(ConvertSupport(common::DECIMAL28DENSE, common::DECIMAL38SPARSE))
+		(ConvertSupport(common::DECIMAL28DENSE, common::VARDECIMAL))
 		(ConvertSupport(common::DECIMAL28DENSE, common::DATE))
 		(ConvertSupport(common::DECIMAL28DENSE, common::TIME))
 		(ConvertSupport(common::DECIMAL28DENSE, common::TIMESTAMP))
@@ -619,6 +666,7 @@ static const convert_support_set s_convertMap = boost::assign::list_of
 		(ConvertSupport(common::DECIMAL38DENSE, common::DECIMAL18))
 		(ConvertSupport(common::DECIMAL38DENSE, common::DECIMAL28SPARSE))
 		(ConvertSupport(common::DECIMAL38DENSE, common::DECIMAL38SPARSE))
+		(ConvertSupport(common::DECIMAL38DENSE, common::VARDECIMAL))
 		(ConvertSupport(common::DECIMAL38DENSE, common::DATE))
 		(ConvertSupport(common::DECIMAL38DENSE, common::TIME))
 		(ConvertSupport(common::DECIMAL38DENSE, common::TIMESTAMP))
@@ -638,6 +686,7 @@ static const convert_support_set s_convertMap = boost::assign::list_of
 		(ConvertSupport(common::DM_UNKNOWN, common::DECIMAL18))
 		(ConvertSupport(common::DM_UNKNOWN, common::DECIMAL28SPARSE))
 		(ConvertSupport(common::DM_UNKNOWN, common::DECIMAL38SPARSE))
+		(ConvertSupport(common::DM_UNKNOWN, common::VARDECIMAL))
 		(ConvertSupport(common::DM_UNKNOWN, common::DATE))
 		(ConvertSupport(common::DM_UNKNOWN, common::TIME))
 		(ConvertSupport(common::DM_UNKNOWN, common::TIMESTAMP))
@@ -656,6 +705,7 @@ static const convert_support_set s_convertMap = boost::assign::list_of
 		(ConvertSupport(common::INTERVALYEAR, common::DECIMAL18))
 		(ConvertSupport(common::INTERVALYEAR, common::DECIMAL28SPARSE))
 		(ConvertSupport(common::INTERVALYEAR, common::DECIMAL38SPARSE))
+		(ConvertSupport(common::INTERVALYEAR, common::VARDECIMAL))
 		(ConvertSupport(common::INTERVALYEAR, common::DATE))
 		(ConvertSupport(common::INTERVALYEAR, common::TIME))
 		(ConvertSupport(common::INTERVALYEAR, common::TIMESTAMP))
@@ -674,6 +724,7 @@ static const convert_support_set s_convertMap = boost::assign::list_of
 		(ConvertSupport(common::INTERVALDAY, common::DECIMAL18))
 		(ConvertSupport(common::INTERVALDAY, common::DECIMAL28SPARSE))
 		(ConvertSupport(common::INTERVALDAY, common::DECIMAL38SPARSE))
+		(ConvertSupport(common::INTERVALDAY, common::VARDECIMAL))
 		(ConvertSupport(common::INTERVALDAY, common::DATE))
 		(ConvertSupport(common::INTERVALDAY, common::TIME))
 		(ConvertSupport(common::INTERVALDAY, common::TIMESTAMP))
diff --git a/contrib/native/client/src/clientlib/recordBatch.cpp b/contrib/native/client/src/clientlib/recordBatch.cpp
index d7c196d00..1c897d694 100644
--- a/contrib/native/client/src/clientlib/recordBatch.cpp
+++ b/contrib/native/client/src/clientlib/recordBatch.cpp
@@ -201,6 +201,8 @@ ValueVectorBase* ValueVectorFactory::allocateValueVector(const Drill::FieldMetad
                     return new ValueVectorDecimal28Sparse(b,f.getValueCount(), f.getScale());
                 case common::DECIMAL38SPARSE:
                     return new ValueVectorDecimal38Sparse(b,f.getValueCount(), f.getScale());
+                case common::VARDECIMAL:
+                    return new ValueVectorVarDecimal(b, f.getValueCount(), f.getScale());
                 case common::DATE:
                     return new ValueVectorTyped<DateHolder, int64_t>(b,f.getValueCount());
                 case common::TIMESTAMP:
@@ -251,6 +253,8 @@ ValueVectorBase* ValueVectorFactory::allocateValueVector(const Drill::FieldMetad
                     return new NullableValueVectorDecimal28Sparse(b,f.getValueCount(), f.getScale());
                 case common::DECIMAL38SPARSE:
                     return new NullableValueVectorDecimal38Sparse(b,f.getValueCount(), f.getScale());
+                case common::VARDECIMAL:
+                    return new NullableValueVectorVarDecimal(b, f.getValueCount(), f.getScale());
                 case common::DATE:
                     return new NullableValueVectorTyped<DateHolder,
                            ValueVectorTyped<DateHolder, int64_t> >(b,f.getValueCount());
diff --git a/contrib/native/client/src/include/drill/decimalUtils.hpp b/contrib/native/client/src/include/drill/decimalUtils.hpp
index 2ace85772..5f9d37a99 100644
--- a/contrib/native/client/src/include/drill/decimalUtils.hpp
+++ b/contrib/native/client/src/include/drill/decimalUtils.hpp
@@ -41,6 +41,7 @@ struct DecimalValue
 
 // These functions need not be exported. They are used by the templates that return the DecimalValue class.
 DecimalValue getDecimalValueFromByteBuf(SlicedByteBuf& data, size_t startIndex, int nDecimalDigits, int scale, bool truncateScale);
+DecimalValue getDecimalValueFromByteBuf(SlicedByteBuf& data, size_t length, int scale);
 DecimalValue getDecimalValueFromDense(SlicedByteBuf& data, size_t startIndex, int nDecimalDigits, int scale, int maxPrecision, int width);
 
 inline DecimalValue getDecimalValueFromIntermediate(SlicedByteBuf& data, size_t startIndex, int nDecimalDigits, int scale)
diff --git a/contrib/native/client/src/include/drill/protobuf/Types.pb.h b/contrib/native/client/src/include/drill/protobuf/Types.pb.h
index f9200ec09..b1dec7bd6 100644
--- a/contrib/native/client/src/include/drill/protobuf/Types.pb.h
+++ b/contrib/native/client/src/include/drill/protobuf/Types.pb.h
@@ -74,11 +74,12 @@ enum MinorType {
   INTERVALDAY = 39,
   LIST = 40,
   GENERIC_OBJECT = 41,
-  UNION = 42
+  UNION = 42,
+  VARDECIMAL = 43
 };
 bool MinorType_IsValid(int value);
 const MinorType MinorType_MIN = LATE;
-const MinorType MinorType_MAX = UNION;
+const MinorType MinorType_MAX = VARDECIMAL;
 const int MinorType_ARRAYSIZE = MinorType_MAX + 1;
 
 const ::google::protobuf::EnumDescriptor* MinorType_descriptor();
diff --git a/contrib/native/client/src/include/drill/recordBatch.hpp b/contrib/native/client/src/include/drill/recordBatch.hpp
index 30287b6ad..435eb7715 100644
--- a/contrib/native/client/src/include/drill/recordBatch.hpp
+++ b/contrib/native/client/src/include/drill/recordBatch.hpp
@@ -346,7 +346,8 @@ template <int DECIMAL_DIGITS, int WIDTH_IN_BYTES, bool IS_SPARSE, int MAX_PRECIS
                     strncpy(buf, str.c_str(), nChars);
                 } else {
                     size_t idxDecimalMark = str.length() - m_scale;
-                    const std::string& decStr= str.substr(0, idxDecimalMark) + "." + str.substr(idxDecimalMark, m_scale);
+                    const std::string& decStr =
+                            (idxDecimalMark == 0 ? "0" : str.substr(0, idxDecimalMark)) + "." + str.substr(idxDecimalMark, m_scale);
                     strncpy(buf, decStr.c_str(), nChars);
                 }
                 return;
@@ -734,6 +735,49 @@ class DECLSPEC_DRILL_CLIENT ValueVectorVarChar:public ValueVectorVarWidth{
         }
 };
 
+class DECLSPEC_DRILL_CLIENT ValueVectorVarDecimal:public ValueVectorVarWidth{
+    public:
+        ValueVectorVarDecimal(SlicedByteBuf *b, size_t rowCount, size_t scale):
+            ValueVectorVarWidth(b, rowCount),
+            m_scale(scale)
+        {
+        }
+        DecimalValue get(size_t index) const {
+            size_t length = getSize(index);
+            ByteBuf_t buff = getRaw(index);
+            SlicedByteBuf intermediateData(&buff[0], 0, length);
+            return getDecimalValueFromByteBuf(intermediateData, length, this->m_scale);
+        }
+
+        void getValueAt(size_t index, char* buf, size_t nChars) const {
+            const DecimalValue& val = this->get(index);
+            std::string str = boost::lexical_cast<std::string>(val.m_unscaledValue);
+            if (str[0] == '-') {
+                str = str.substr(1);
+                while (str.length() < m_scale) {
+                    str = "0" + str;
+                }
+                str = "-" + str;
+            } else {
+                while (str.length() < m_scale) {
+                    str = "0" + str;
+                }
+            }
+            if (m_scale == 0) {
+                strncpy(buf, str.c_str(), nChars);
+            } else {
+                size_t idxDecimalMark = str.length() - m_scale;
+                const std::string& decStr =
+                        (idxDecimalMark == 0 ? "0" : str.substr(0, idxDecimalMark)) + "." + str.substr(idxDecimalMark, m_scale);
+                strncpy(buf, decStr.c_str(), nChars);
+            }
+            return;
+        }
+
+    private:
+        int32_t m_scale;
+};
+
 class DECLSPEC_DRILL_CLIENT ValueVectorVarBinary:public ValueVectorVarWidth{
     public:
         ValueVectorVarBinary(SlicedByteBuf *b, size_t rowCount):ValueVectorVarWidth(b, rowCount){
@@ -764,10 +808,11 @@ typedef ValueVectorDecimal<6, 24, true, 38>  ValueVectorDecimal38Sparse;
 
 typedef NullableValueVectorTyped<int32_t, ValueVectorDecimal9> NullableValueVectorDecimal9;
 typedef NullableValueVectorTyped<int64_t, ValueVectorDecimal18> NullableValueVectorDecimal18;
-typedef NullableValueVectorTyped<DecimalValue , ValueVectorDecimal28Dense> NullableValueVectorDecimal28Dense;
-typedef NullableValueVectorTyped<DecimalValue , ValueVectorDecimal38Dense> NullableValueVectorDecimal38Dense;
-typedef NullableValueVectorTyped<DecimalValue , ValueVectorDecimal28Sparse> NullableValueVectorDecimal28Sparse;
-typedef NullableValueVectorTyped<DecimalValue , ValueVectorDecimal38Sparse> NullableValueVectorDecimal38Sparse;
+typedef NullableValueVectorTyped<DecimalValue, ValueVectorDecimal28Dense> NullableValueVectorDecimal28Dense;
+typedef NullableValueVectorTyped<DecimalValue, ValueVectorDecimal38Dense> NullableValueVectorDecimal38Dense;
+typedef NullableValueVectorTyped<DecimalValue, ValueVectorDecimal28Sparse> NullableValueVectorDecimal28Sparse;
+typedef NullableValueVectorTyped<DecimalValue, ValueVectorDecimal38Sparse> NullableValueVectorDecimal38Sparse;
+typedef NullableValueVectorTyped<DecimalValue, ValueVectorVarDecimal> NullableValueVectorVarDecimal;
 
 typedef ValueVectorTyped<DateHolder, int64_t> ValueVectorDate;
 typedef ValueVectorTyped<DateTimeHolder, int64_t> ValueVectorTimestamp;
diff --git a/contrib/native/client/src/protobuf/Types.pb.cc b/contrib/native/client/src/protobuf/Types.pb.cc
index ec8a1c84f..675bba02d 100644
--- a/contrib/native/client/src/protobuf/Types.pb.cc
+++ b/contrib/native/client/src/protobuf/Types.pb.cc
@@ -93,7 +93,7 @@ void protobuf_AddDesc_Types_2eproto() {
     "de\030\002 \001(\0162\020.common.DataMode\022\r\n\005width\030\003 \001("
     "\005\022\021\n\tprecision\030\004 \001(\005\022\r\n\005scale\030\005 \001(\005\022\020\n\010t"
     "imeZone\030\006 \001(\005\022#\n\010sub_type\030\007 \003(\0162\021.common"
-    ".MinorType*\233\004\n\tMinorType\022\010\n\004LATE\020\000\022\007\n\003MA"
+    ".MinorType*\253\004\n\tMinorType\022\010\n\004LATE\020\000\022\007\n\003MA"
     "P\020\001\022\013\n\007TINYINT\020\003\022\014\n\010SMALLINT\020\004\022\007\n\003INT\020\005\022"
     "\n\n\006BIGINT\020\006\022\014\n\010DECIMAL9\020\007\022\r\n\tDECIMAL18\020\010"
     "\022\023\n\017DECIMAL28SPARSE\020\t\022\023\n\017DECIMAL38SPARSE"
@@ -106,10 +106,10 @@ void protobuf_AddDesc_Types_2eproto() {
     "\n\005UINT4\020\037\022\t\n\005UINT8\020 \022\022\n\016DECIMAL28DENSE\020!"
     "\022\022\n\016DECIMAL38DENSE\020\"\022\016\n\nDM_UNKNOWN\020%\022\020\n\014"
     "INTERVALYEAR\020&\022\017\n\013INTERVALDAY\020\'\022\010\n\004LIST\020"
-    "(\022\022\n\016GENERIC_OBJECT\020)\022\t\n\005UNION\020**=\n\010Data"
-    "Mode\022\017\n\013DM_OPTIONAL\020\000\022\017\n\013DM_REQUIRED\020\001\022\017"
-    "\n\013DM_REPEATED\020\002B-\n\035org.apache.drill.comm"
-    "on.typesB\nTypeProtosH\001", 862);
+    "(\022\022\n\016GENERIC_OBJECT\020)\022\t\n\005UNION\020*\022\016\n\nVARD"
+    "ECIMAL\020+*=\n\010DataMode\022\017\n\013DM_OPTIONAL\020\000\022\017\n"
+    "\013DM_REQUIRED\020\001\022\017\n\013DM_REPEATED\020\002B-\n\035org.a"
+    "pache.drill.common.typesB\nTypeProtosH\001", 878);
   ::google::protobuf::MessageFactory::InternalRegisterGeneratedFile(
     "Types.proto", &protobuf_RegisterTypes);
   MajorType::default_instance_ = new MajorType();
@@ -167,6 +167,7 @@ bool MinorType_IsValid(int value) {
     case 40:
     case 41:
     case 42:
+    case 43:
       return true;
     default:
       return false;
diff --git a/contrib/storage-hive/core/src/main/codegen/data/HiveTypes.tdd b/contrib/storage-hive/core/src/main/codegen/data/HiveTypes.tdd
index e13dc3644..864610c7b 100644
--- a/contrib/storage-hive/core/src/main/codegen/data/HiveTypes.tdd
+++ b/contrib/storage-hive/core/src/main/codegen/data/HiveTypes.tdd
@@ -104,7 +104,7 @@
       hiveType: "DECIMAL",
       hiveOI: "HiveDecimalObjectInspector",
       javaType: "org.apache.hadoop.hive.common.type.HiveDecimal",
-      drillType: "Decimal38Sparse",
+      drillType: "VarDecimal",
       needOIForDrillType: true
     },
     {
diff --git a/contrib/storage-hive/core/src/main/codegen/templates/ObjectInspectors.java b/contrib/storage-hive/core/src/main/codegen/templates/ObjectInspectors.java
index 9c4531416..a539b7f1d 100644
--- a/contrib/storage-hive/core/src/main/codegen/templates/ObjectInspectors.java
+++ b/contrib/storage-hive/core/src/main/codegen/templates/ObjectInspectors.java
@@ -100,12 +100,12 @@ public class Drill${entry.drillType}${entry.hiveOI} {
     }
 
     @Override
-    public String getPrimitiveJavaObject(Object o){
+    public String getPrimitiveJavaObject(Object o) {
     <#if mode == "Optional">
       if (o == null) {
         return null;
       }
-      final NullableVarCharHolder h = (NullableVarCharHolder)o;
+      final NullableVarCharHolder h = (NullableVarCharHolder) o;
     <#else>
       final VarCharHolder h = (VarCharHolder)o;
     </#if>
@@ -118,9 +118,9 @@ public class Drill${entry.drillType}${entry.hiveOI} {
       if (o == null) {
         return null;
       }
-      final NullableVarBinaryHolder h = (NullableVarBinaryHolder)o;
+      final NullableVarBinaryHolder h = (NullableVarBinaryHolder) o;
     <#else>
-      final VarBinaryHolder h = (VarBinaryHolder)o;
+      final VarBinaryHolder h = (VarBinaryHolder) o;
     </#if>
       final byte[] buf = new byte[h.end-h.start];
       h.buffer.getBytes(h.start, buf, 0, h.end-h.start);
@@ -133,9 +133,9 @@ public class Drill${entry.drillType}${entry.hiveOI} {
       if (o == null) {
         return null;
       }
-      final NullableVarBinaryHolder h = (NullableVarBinaryHolder)o;
+      final NullableVarBinaryHolder h = (NullableVarBinaryHolder) o;
     <#else>
-      final VarBinaryHolder h = (VarBinaryHolder)o;
+      final VarBinaryHolder h = (VarBinaryHolder) o;
     </#if>
       final byte[] buf = new byte[h.end-h.start];
       h.buffer.getBytes(h.start, buf, 0, h.end-h.start);
@@ -174,18 +174,18 @@ public class Drill${entry.drillType}${entry.hiveOI} {
       return Boolean.valueOf(((BitHolder)o).value != 0);
     </#if>
     }
-<#elseif entry.drillType == "Decimal38Sparse">
+<#elseif entry.drillType == "VarDecimal">
     @Override
-    public HiveDecimal getPrimitiveJavaObject(Object o){
+    public HiveDecimal getPrimitiveJavaObject(Object o) {
     <#if mode == "Optional">
       if (o == null) {
         return null;
       }
-      final NullableDecimal38SparseHolder h = (NullableDecimal38SparseHolder) o;
+      final NullableVarDecimalHolder h = (NullableVarDecimalHolder) o;
     <#else>
-      final Decimal38SparseHolder h = (Decimal38SparseHolder) o;
+      final VarDecimalHolder h = (VarDecimalHolder) o;
     </#if>
-      return HiveDecimal.create(DecimalUtility.getBigDecimalFromSparse(h.buffer, h.start, h.nDecimalDigits, h.scale));
+      return HiveDecimal.create(DecimalUtility.getBigDecimalFromDrillBuf(h.buffer, h.start, h.end - h.start, h.scale));
     }
 
     @Override
@@ -194,17 +194,17 @@ public class Drill${entry.drillType}${entry.hiveOI} {
       if (o == null) {
         return null;
       }
-      final NullableDecimal38SparseHolder h = (NullableDecimal38SparseHolder) o;
+      final NullableVarDecimalHolder h = (NullableVarDecimalHolder) o;
     <#else>
-      final Decimal38SparseHolder h = (Decimal38SparseHolder) o;
+      final VarDecimalHolder h = (VarDecimalHolder) o;
     </#if>
       return new HiveDecimalWritable(
-          HiveDecimal.create(DecimalUtility.getBigDecimalFromSparse(h.buffer, h.start, h.nDecimalDigits, h.scale)));
+          HiveDecimal.create(DecimalUtility.getBigDecimalFromDrillBuf(h.buffer, h.start, h.end - h.start, h.scale)));
     }
 
 <#elseif entry.drillType == "TimeStamp">
     @Override
-    public java.sql.Timestamp getPrimitiveJavaObject(Object o){
+    public java.sql.Timestamp getPrimitiveJavaObject(Object o) {
     <#if mode == "Optional">
       if (o == null) {
         return null;
@@ -237,7 +237,7 @@ public class Drill${entry.drillType}${entry.hiveOI} {
 
 <#elseif entry.drillType == "Date">
     @Override
-    public java.sql.Date getPrimitiveJavaObject(Object o){
+    public java.sql.Date getPrimitiveJavaObject(Object o) {
     <#if mode == "Optional">
       if (o == null) {
         return null;
@@ -270,11 +270,11 @@ public class Drill${entry.drillType}${entry.hiveOI} {
 
 <#else>
     @Override
-    public ${entry.javaType} get(Object o){
+    public ${entry.javaType} get(Object o) {
     <#if mode == "Optional">
-      return ((Nullable${entry.drillType}Holder)o).value;
+      return ((Nullable${entry.drillType}Holder) o).value;
     <#else>
-      return ((${entry.drillType}Holder)o).value;
+      return ((${entry.drillType}Holder) o).value;
     </#if>
     }
 
@@ -295,9 +295,9 @@ public class Drill${entry.drillType}${entry.hiveOI} {
       if (o == null) {
         return null;
       }
-      return new ${entry.javaType?cap_first}(((Nullable${entry.drillType}Holder)o).value);
+      return new ${entry.javaType?cap_first}(((Nullable${entry.drillType}Holder) o).value);
     <#else>
-      return new ${entry.javaType?cap_first}(((${entry.drillType}Holder)o).value);
+      return new ${entry.javaType?cap_first}(((${entry.drillType}Holder) o).value);
     </#if>
     }
 </#if>
@@ -310,7 +310,7 @@ public class Drill${entry.drillType}${entry.hiveOI} {
       }
       final Nullable${entry.drillType}Holder h = (Nullable${entry.drillType}Holder) o;
     <#else>
-      final ${entry.drillType}Holder h = (${entry.drillType}Holder)o;
+      final ${entry.drillType}Holder h = (${entry.drillType}Holder) o;
     </#if>
       return new ${entry.javaType?cap_first}Writable(h.value);
     }
diff --git a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/expr/fn/HiveFunctionRegistry.java b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/expr/fn/HiveFunctionRegistry.java
index f6232f680..92e7f83cc 100644
--- a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/expr/fn/HiveFunctionRegistry.java
+++ b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/expr/fn/HiveFunctionRegistry.java
@@ -32,7 +32,6 @@ import org.apache.drill.common.scanner.ClassPathScanner;
 import org.apache.drill.common.scanner.persistence.ScanResult;
 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.expr.fn.impl.hive.ObjectInspectorHelper;
 import org.apache.drill.exec.planner.sql.DrillOperatorTable;
@@ -267,9 +266,9 @@ public class HiveFunctionRegistry implements PluggableFunctionRegistry{
             .build(logger);
       }
 
-      return TypeInferenceUtils.createCalciteTypeWithNullability(
+      return TypeInferenceUtils.convertToCalciteType(
           opBinding.getTypeFactory(),
-          TypeInferenceUtils.getCalciteTypeFromDrillType(hiveFuncHolder.getReturnType().getMinorType()),
+          hiveFuncHolder.getReturnType(),
           hiveFuncHolder.getReturnType().getMode() != TypeProtos.DataMode.REQUIRED);
     }
   }
diff --git a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveFieldConverter.java b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveFieldConverter.java
index 2fab5a4d4..6caecbb82 100644
--- a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveFieldConverter.java
+++ b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveFieldConverter.java
@@ -17,27 +17,19 @@
  */
 package org.apache.drill.exec.store.hive;
 
+import java.math.RoundingMode;
 import java.util.Map;
 
-import org.apache.drill.exec.expr.holders.Decimal18Holder;
-import org.apache.drill.exec.expr.holders.Decimal28SparseHolder;
-import org.apache.drill.exec.expr.holders.Decimal38SparseHolder;
-import org.apache.drill.exec.expr.holders.Decimal9Holder;
-import org.apache.drill.exec.ops.FragmentContext;
-import org.apache.drill.exec.util.DecimalUtility;
 import org.apache.drill.exec.vector.NullableBigIntVector;
 import org.apache.drill.exec.vector.NullableBitVector;
 import org.apache.drill.exec.vector.NullableDateVector;
-import org.apache.drill.exec.vector.NullableDecimal18Vector;
-import org.apache.drill.exec.vector.NullableDecimal28SparseVector;
-import org.apache.drill.exec.vector.NullableDecimal38SparseVector;
-import org.apache.drill.exec.vector.NullableDecimal9Vector;
 import org.apache.drill.exec.vector.NullableFloat4Vector;
 import org.apache.drill.exec.vector.NullableFloat8Vector;
 import org.apache.drill.exec.vector.NullableIntVector;
 import org.apache.drill.exec.vector.NullableTimeStampVector;
 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 org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
 import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector.PrimitiveCategory;
@@ -55,7 +47,6 @@ import org.apache.hadoop.hive.serde2.objectinspector.primitive.LongObjectInspect
 import org.apache.hadoop.hive.serde2.objectinspector.primitive.ShortObjectInspector;
 import org.apache.hadoop.hive.serde2.objectinspector.primitive.StringObjectInspector;
 import org.apache.hadoop.hive.serde2.objectinspector.primitive.TimestampObjectInspector;
-import org.apache.hadoop.hive.serde2.typeinfo.DecimalTypeInfo;
 import org.apache.hadoop.hive.serde2.typeinfo.PrimitiveTypeInfo;
 import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
 import org.apache.hadoop.io.Text;
@@ -92,33 +83,18 @@ public abstract class HiveFieldConverter {
     primMap.put(PrimitiveCategory.TIMESTAMP, Timestamp.class);
     primMap.put(PrimitiveCategory.DATE, Date.class);
     primMap.put(PrimitiveCategory.CHAR, Char.class);
+    primMap.put(PrimitiveCategory.DECIMAL, VarDecimal.class);
   }
 
 
-  public static HiveFieldConverter create(TypeInfo typeInfo, FragmentContext fragmentContext)
+  public static HiveFieldConverter create(TypeInfo typeInfo)
       throws IllegalAccessException, InstantiationException {
     switch (typeInfo.getCategory()) {
       case PRIMITIVE:
         final PrimitiveCategory pCat = ((PrimitiveTypeInfo) typeInfo).getPrimitiveCategory();
-        if (pCat != PrimitiveCategory.DECIMAL) {
-          Class<? extends HiveFieldConverter> clazz = primMap.get(pCat);
-          if (clazz != null) {
-            return clazz.newInstance();
-          }
-        } else {
-          // For decimal, based on precision return appropriate converter.
-          DecimalTypeInfo decimalTypeInfo = (DecimalTypeInfo) typeInfo;
-          int precision = decimalTypeInfo.precision();
-          int scale = decimalTypeInfo.scale();
-          if (precision <= 9) {
-            return new Decimal9(precision, scale);
-          } else if (precision <= 18) {
-            return new Decimal18(precision, scale);
-          } else if (precision <= 28) {
-            return new Decimal28(precision, scale, fragmentContext);
-          } else {
-            return new Decimal38(precision, scale, fragmentContext);
-          }
+        Class<? extends HiveFieldConverter> clazz = primMap.get(pCat);
+        if (clazz != null) {
+          return clazz.newInstance();
         }
 
         throwUnsupportedHiveDataTypeError(pCat.toString());
@@ -151,75 +127,15 @@ public abstract class HiveFieldConverter {
     }
   }
 
-  public static class Decimal9 extends HiveFieldConverter {
-    private final Decimal9Holder holder = new Decimal9Holder();
-
-    public Decimal9(int precision, int scale) {
-      holder.scale = scale;
-      holder.precision = precision;
-    }
-
-    @Override
-    public void setSafeValue(ObjectInspector oi, Object hiveFieldValue, ValueVector outputVV, int outputIndex) {
-      holder.value = DecimalUtility.getDecimal9FromBigDecimal(
-          ((HiveDecimalObjectInspector)oi).getPrimitiveJavaObject(hiveFieldValue).bigDecimalValue(),
-          holder.scale, holder.precision);
-      ((NullableDecimal9Vector) outputVV).getMutator().setSafe(outputIndex, holder);
-    }
-  }
-
-  public static class Decimal18 extends HiveFieldConverter {
-    private final Decimal18Holder holder = new Decimal18Holder();
-
-    public Decimal18(int precision, int scale) {
-      holder.scale = scale;
-      holder.precision = precision;
-    }
-
-    @Override
-    public void setSafeValue(ObjectInspector oi, Object hiveFieldValue, ValueVector outputVV, int outputIndex) {
-      holder.value = DecimalUtility.getDecimal18FromBigDecimal(
-          ((HiveDecimalObjectInspector)oi).getPrimitiveJavaObject(hiveFieldValue).bigDecimalValue(),
-          holder.scale, holder.precision);
-      ((NullableDecimal18Vector) outputVV).getMutator().setSafe(outputIndex, holder);
-    }
-  }
-
-  public static class Decimal28 extends HiveFieldConverter {
-    private final Decimal28SparseHolder holder = new Decimal28SparseHolder();
-
-    public Decimal28(int precision, int scale, FragmentContext context) {
-      holder.scale = scale;
-      holder.precision = precision;
-      holder.buffer = context.getManagedBuffer(Decimal28SparseHolder.nDecimalDigits * DecimalUtility.INTEGER_SIZE);
-      holder.start = 0;
-    }
-
-    @Override
-    public void setSafeValue(ObjectInspector oi, Object hiveFieldValue, ValueVector outputVV, int outputIndex) {
-      DecimalUtility.getSparseFromBigDecimal(
-          ((HiveDecimalObjectInspector)oi).getPrimitiveJavaObject(hiveFieldValue).bigDecimalValue(),
-          holder.buffer, holder.start, holder.scale, holder.precision, Decimal28SparseHolder.nDecimalDigits);
-      ((NullableDecimal28SparseVector) outputVV).getMutator().setSafe(outputIndex, holder);
-    }
-  }
-
-  public static class Decimal38 extends HiveFieldConverter {
-    private final Decimal38SparseHolder holder = new Decimal38SparseHolder();
-
-    public Decimal38(int precision, int scale, FragmentContext context) {
-      holder.scale = scale;
-      holder.precision = precision;
-      holder.buffer = context.getManagedBuffer(Decimal38SparseHolder.nDecimalDigits * DecimalUtility.INTEGER_SIZE);
-      holder.start = 0;
-    }
-
+  public static class VarDecimal extends HiveFieldConverter {
     @Override
     public void setSafeValue(ObjectInspector oi, Object hiveFieldValue, ValueVector outputVV, int outputIndex) {
-      DecimalUtility.getSparseFromBigDecimal(
-          ((HiveDecimalObjectInspector)oi).getPrimitiveJavaObject(hiveFieldValue).bigDecimalValue(),
-          holder.buffer, holder.start, holder.scale, holder.precision, Decimal38SparseHolder.nDecimalDigits);
-      ((NullableDecimal38SparseVector) outputVV).getMutator().setSafe(outputIndex, holder);
+      ((NullableVarDecimalVector) outputVV).getMutator()
+          .setSafe(
+              outputIndex,
+              ((HiveDecimalObjectInspector) oi)
+                  .getPrimitiveJavaObject(hiveFieldValue).bigDecimalValue()
+                  .setScale(outputVV.getField().getScale(), RoundingMode.HALF_UP));
     }
   }
 
diff --git a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveUtilities.java b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveUtilities.java
index 5ff55060a..91d0567af 100644
--- a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveUtilities.java
+++ b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveUtilities.java
@@ -50,6 +50,7 @@ import org.apache.drill.exec.vector.NullableIntVector;
 import org.apache.drill.exec.vector.NullableTimeStampVector;
 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 org.apache.drill.exec.work.ExecErrorConstants;
 
@@ -80,6 +81,7 @@ import org.joda.time.DateTimeZone;
 
 import javax.annotation.Nullable;
 import java.math.BigDecimal;
+import java.math.RoundingMode;
 import java.sql.Date;
 import java.sql.Timestamp;
 import java.util.List;
@@ -293,6 +295,15 @@ public class HiveUtilities {
         }
         break;
       }
+      case VARDECIMAL: {
+        final BigDecimal value = ((HiveDecimal) val).bigDecimalValue()
+            .setScale(vector.getField().getScale(), RoundingMode.HALF_UP);
+        final NullableVarDecimalVector v = ((NullableVarDecimalVector) vector);
+        for (int i = start; i < end; i++) {
+          v.getMutator().setSafe(i, value);
+        }
+        break;
+      }
     }
   }
 
@@ -346,8 +357,7 @@ public class HiveUtilities {
               .message(ExecErrorConstants.DECIMAL_DISABLE_ERR_MSG)
               .build(logger);
         }
-        DecimalTypeInfo decimalTypeInfo = (DecimalTypeInfo) primitiveTypeInfo;
-        return DecimalUtility.getDecimalDataType(decimalTypeInfo.precision());
+        return MinorType.VARDECIMAL;
       }
       case DOUBLE:
         return TypeProtos.MinorType.FLOAT8;
diff --git a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/readers/HiveAbstractReader.java b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/readers/HiveAbstractReader.java
index a922b4ca0..d0d9ed067 100644
--- a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/readers/HiveAbstractReader.java
+++ b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/readers/HiveAbstractReader.java
@@ -227,7 +227,7 @@ public abstract class HiveAbstractReader extends AbstractRecordReader {
 
         selectedColumnObjInspectors.add(fieldOI);
         selectedColumnTypes.add(typeInfo);
-        selectedColumnFieldConverters.add(HiveFieldConverter.create(typeInfo, fragmentContext));
+        selectedColumnFieldConverters.add(HiveFieldConverter.create(typeInfo));
       }
 
       for(int i=0; i<selectedColumnNames.size(); ++i){
diff --git a/contrib/storage-hive/hive-exec-shade/pom.xml b/contrib/storage-hive/hive-exec-shade/pom.xml
index 3d4dcebc3..ad572efaf 100644
--- a/contrib/storage-hive/hive-exec-shade/pom.xml
+++ b/contrib/storage-hive/hive-exec-shade/pom.xml
@@ -117,6 +117,10 @@
               <pattern>org.apache.parquet.</pattern>
               <shadedPattern>hive.org.apache.parquet.</shadedPattern>
             </relocation>
+            <relocation>
+              <pattern>org.apache.avro.</pattern>
+              <shadedPattern>hive.org.apache.avro.</shadedPattern>
+            </relocation>
           </relocations>
           <filters>
             <filter>
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);
       }
     }
 
diff --git a/contrib/storage-mongo/src/main/java/org/apache/drill/exec/store/mongo/MongoCompareFunctionProcessor.java b/contrib/storage-mongo/src/main/java/org/apache/drill/exec/store/mongo/MongoCompareFunctionProcessor.java
index 130ea0f06..b12887d7c 100644
--- a/contrib/storage-mongo/src/main/java/org/apache/drill/exec/store/mongo/MongoCompareFunctionProcessor.java
+++ b/contrib/storage-mongo/src/main/java/org/apache/drill/exec/store/mongo/MongoCompareFunctionProcessor.java
@@ -30,6 +30,7 @@ import org.apache.drill.common.expression.ValueExpressions.IntExpression;
 import org.apache.drill.common.expression.ValueExpressions.LongExpression;
 import org.apache.drill.common.expression.ValueExpressions.QuotedString;
 import org.apache.drill.common.expression.ValueExpressions.TimeExpression;
+import org.apache.drill.common.expression.ValueExpressions.VarDecimalExpression;
 import org.apache.drill.common.expression.visitors.AbstractExprVisitor;
 
 import com.google.common.collect.ImmutableMap;
@@ -110,7 +111,7 @@ public class MongoCompareFunctionProcessor extends
   @Override
   public Boolean visitConvertExpression(ConvertExpression e,
       LogicalExpression valueArg) throws RuntimeException {
-    if (e.getConvertFunction() == ConvertExpression.CONVERT_FROM
+    if (ConvertExpression.CONVERT_FROM.equals(e.getConvertFunction())
         && e.getInput() instanceof SchemaPath) {
       String encodingType = e.getEncodingType();
       switch (encodingType) {
@@ -219,6 +220,14 @@ public class MongoCompareFunctionProcessor extends
       return true;
     }
 
+    // Mongo does not support decimals, therefore double value is used.
+    // See list of supported types in BsonValueCodecProvider.
+    if (valueArg instanceof VarDecimalExpression) {
+      this.value = ((VarDecimalExpression) valueArg).getBigDecimal().doubleValue();
+      this.path = path;
+      return true;
+    }
+
     return false;
   }
 
@@ -230,7 +239,7 @@ public class MongoCompareFunctionProcessor extends
         .add(DateExpression.class).add(DoubleExpression.class)
         .add(FloatExpression.class).add(IntExpression.class)
         .add(LongExpression.class).add(QuotedString.class)
-        .add(TimeExpression.class).build();
+        .add(TimeExpression.class).add(VarDecimalExpression.class).build();
   }
 
   private static final ImmutableMap<String, String> COMPARE_FUNCTIONS_TRANSPOSE_MAP;