DRILL-4184: Support variable length decimal fields in parquet

69 files changed, 1739 insertions, 98 deletions

diff --git a/common/src/main/java/org/apache/drill/common/types/Types.java b/common/src/main/java/org/apache/drill/common/types/Types.java
index 8f5d1f920..ae269252f 100644
--- a/common/src/main/java/org/apache/drill/common/types/Types.java
+++ b/common/src/main/java/org/apache/drill/common/types/Types.java
@@ -73,6 +73,7 @@ public class Types {
 
     switch(type.getMinorType()) {
     case BIGINT:
+    case VARDECIMAL:
     case DECIMAL38SPARSE:
     case DECIMAL38DENSE:
     case DECIMAL28SPARSE:
@@ -120,6 +121,7 @@ public class Types {
       case FLOAT4:          return "FLOAT";
       case FLOAT8:          return "DOUBLE";
 
+      case VARDECIMAL:
       case DECIMAL9:
       case DECIMAL18:
       case DECIMAL28DENSE:
@@ -238,6 +240,7 @@ public class Types {
           case INTERVALYEAR:    // SQL INTERVAL w/YEAR and/or MONTH
           case INTERVALDAY:     // SQL INTERVAL w/DAY, HOUR, MINUTE and/or SECOND
           // Not-yet seen/verified signed types:
+          case VARDECIMAL:      // SQL DECIMAL (if used)
           case DECIMAL9:        // SQL DECIMAL (if used)
           case DECIMAL18:       // SQL DECIMAL (if used)
           case DECIMAL28SPARSE: // SQL DECIMAL (if used)
@@ -398,6 +401,7 @@ public class Types {
     case VAR16CHAR:
     case VARCHAR:
     case UNION:
+    case VARDECIMAL:
       return false;
     default:
       return true;
@@ -633,6 +637,8 @@ public class Types {
         return "float";
       case FLOAT8:
         return "double";
+      case VARDECIMAL:
+        return "decimal";
       case DECIMAL9:
         return "decimal";
       case DECIMAL18:
diff --git a/common/src/main/java/org/apache/drill/common/util/CoreDecimalUtility.java b/common/src/main/java/org/apache/drill/common/util/CoreDecimalUtility.java
index 33c1f48de..4b4b950ff 100644
--- a/common/src/main/java/org/apache/drill/common/util/CoreDecimalUtility.java
+++ b/common/src/main/java/org/apache/drill/common/util/CoreDecimalUtility.java
@@ -82,9 +82,11 @@ public class CoreDecimalUtility {
   }
 
   public static boolean isDecimalType(TypeProtos.MinorType minorType) {
-    return minorType == TypeProtos.MinorType.DECIMAL9 ||
-           minorType == TypeProtos.MinorType.DECIMAL18 ||
-           minorType == TypeProtos.MinorType.DECIMAL28SPARSE ||
-           minorType == TypeProtos.MinorType.DECIMAL38SPARSE;
+    if (minorType == TypeProtos.MinorType.VARDECIMAL || minorType == TypeProtos.MinorType.DECIMAL9 ||
+        minorType == TypeProtos.MinorType.DECIMAL18 || minorType == TypeProtos.MinorType.DECIMAL28SPARSE ||
+        minorType == TypeProtos.MinorType.DECIMAL38SPARSE) {
+      return true;
+    }
+    return false;
   }
 }
diff --git a/exec/java-exec/src/main/codegen/data/CastHigh.tdd b/exec/java-exec/src/main/codegen/data/CastHigh.tdd
index 54c337ddf..dc4830b1b 100644
--- a/exec/java-exec/src/main/codegen/data/CastHigh.tdd
+++ b/exec/java-exec/src/main/codegen/data/CastHigh.tdd
@@ -24,5 +24,6 @@
     {value: false, from: "Decimal18"},
     {value: false, from: "Decimal28Sparse"},
     {value: false, from: "Decimal38Sparse"},
+    {value: false, from: "VarDecimal"}
   ]
 } 
diff --git a/exec/java-exec/src/main/codegen/data/Casts.tdd b/exec/java-exec/src/main/codegen/data/Casts.tdd
index 307c0f263..63fa0e4cb 100644
--- a/exec/java-exec/src/main/codegen/data/Casts.tdd
+++ b/exec/java-exec/src/main/codegen/data/Casts.tdd
@@ -102,6 +102,9 @@
     {from: "Decimal28Sparse", to: "Decimal38Dense", major: "DecimalSparseDecimalDense", arraySize: "4"},
     {from: "Decimal38Sparse", to: "Decimal38Dense", major: "DecimalSparseDecimalDense", arraySize: "4"},
 
+    {from: "Decimal28Sparse", to: "VarDecimal", major: "DecimalSparseVarDecimal", arraySize: "6"},
+    {from: "Decimal38Sparse", to: "VarDecimal", major: "DecimalSparseVarDecimal", arraySize: "6"},
+
     {from: "Decimal28Dense", to: "Decimal38Dense", major: "DecimalSimilar", arraySize: "4"},
     {from: "Decimal28Sparse", to: "Decimal38Sparse", major: "DecimalSimilar", arraySize: "6"},
 
@@ -109,21 +112,25 @@
     {from: "Int", to: "Decimal18", major: "IntDecimal", javatype: "long"},
     {from: "Int", to: "Decimal28Sparse", major: "IntDecimal", arraySize: "5"},
     {from: "Int", to: "Decimal38Sparse", major: "IntDecimal", arraySize: "6"},
+    {from: "Int", to: "VarDecimal", major: "IntDecimal", arraySize: "6"},
 
     {from: "BigInt", to: "Decimal9", major: "BigIntDecimal", javatype: "int"},
     {from: "BigInt", to: "Decimal18", major: "BigIntDecimal", javatype: "long"},
     {from: "BigInt", to: "Decimal28Sparse", major: "BigIntDecimal", arraySize: "5"},
     {from: "BigInt", to: "Decimal38Sparse", major: "BigIntDecimal", arraySize: "6"},
+    {from: "BigInt", to: "VarDecimal", major: "BigIntDecimal", arraySize: "6"},
 
     {from: "Decimal9", to: "Int", major: "DecimalSimpleInt", javatype: "int"},
     {from: "Decimal18", to: "Int", major: "DecimalSimpleInt", javatype: "int"},
     {from: "Decimal28Sparse", to: "Int", major: "DecimalComplexInt", javatype: "int"},
     {from: "Decimal38Sparse", to: "Int", major: "DecimalComplexInt", javatype: "int"},
+    {from: "VarDecimal", to: "Int", major: "DecimalComplexInt", javatype: "int"},
 
     {from: "Decimal9", to: "BigInt", major: "DecimalSimpleBigInt", javatype: "long"},
     {from: "Decimal18", to: "BigInt", major: "DecimalSimpleBigInt", javatype: "long"},
     {from: "Decimal28Sparse", to: "BigInt", major: "DecimalComplexBigInt", javatype: "long"},
     {from: "Decimal38Sparse", to: "BigInt", major: "DecimalComplexBigInt", javatype: "long"},
+    {from: "VarDecimal", to: "BigInt", major: "DecimalComplexBigInt", javatype: "long"},
 
     {from: "Decimal9", to: "Float4", major: "DecimalSimpleFloat", javatype: "float"},
     {from: "Decimal18", to: "Float4", major: "DecimalSimpleFloat", javatype: "float"},
@@ -131,16 +138,19 @@
     {from: "Decimal28Dense", to: "Float4", major: "DecimalComplexFloat", javatype: "float"},
     {from: "Decimal38Sparse", to: "Float4", major: "DecimalComplexFloat", javatype: "float"},
     {from: "Decimal38Dense", to: "Float4", major: "DecimalComplexFloat", javatype: "float"},
+    {from: "VarDecimal", to: "Float4", major: "DecimalComplexFloat", javatype: "float"},
 
     {from: "Float4", to: "Decimal9", major: "FloatDecimalSimple", javatype: "int"},
     {from: "Float4", to: "Decimal18", major: "FloatDecimalSimple", javatype: "long"},
     {from: "Float4", to: "Decimal28Sparse", major: "FloatDecimalComplex", arraySize: "5"},
     {from: "Float4", to: "Decimal38Sparse", major: "FloatDecimalComplex", arraySize: "6"},
+    {from: "Float4", to: "VarDecimal", major: "FloatDecimalComplex", arraySize: "6"},
 
     {from: "Float8", to: "Decimal9", major: "DoubleDecimalSimple", javatype: "int"},
     {from: "Float8", to: "Decimal18", major: "DoubleDecimalSimple", javatype: "long"},
     {from: "Float8", to: "Decimal28Sparse", major: "DoubleDecimalComplex", arraySize: "5"},
     {from: "Float8", to: "Decimal38Sparse", major: "DoubleDecimalComplex", arraySize: "6"}
+    {from: "Float8", to: "VarDecimal", major: "DoubleDecimalComplex", arraySize: "6"}
 
     {from: "Decimal9", to: "Float8", major: "DecimalSimpleDouble", javatype: "double"},
     {from: "Decimal18", to: "Float8", major: "DecimalSimpleDouble", javatype: "double"},
@@ -148,16 +158,19 @@
     {from: "Decimal28Dense", to: "Float8", major: "DecimalComplexDouble", javatype: "double"},
     {from: "Decimal38Sparse", to: "Float8", major: "DecimalComplexDouble", javatype: "double"},
     {from: "Decimal38Dense", to: "Float8", major: "DecimalComplexDouble", javatype: "double"},
+    {from: "VarDecimal", to: "Float8", major: "DecimalComplexDouble", javatype: "double"},
 
     {from: "VarChar", to: "Decimal9", major: "VarCharDecimalSimple", javatype: "int"},
     {from: "VarChar", to: "Decimal18", major: "VarCharDecimalSimple", javatype: "long"},
     {from: "VarChar", to: "Decimal28Sparse", major: "VarCharDecimalComplex", arraySize: "5"},
     {from: "VarChar", to: "Decimal38Sparse", major: "VarCharDecimalComplex", arraySize: "6"},
+    {from: "VarChar", to: "VarDecimal", major: "VarCharDecimalComplex", arraySize: "6"},
 
     {from: "Decimal9", to: "VarChar", major: "DecimalSimpleVarChar", bufferSize: "11", javatype: "int"},
     {from: "Decimal18", to: "VarChar", major: "DecimalSimpleVarChar", bufferSize: "20", javatype: "long"},
     {from: "Decimal28Sparse", to: "VarChar", major: "DecimalComplexVarChar", bufferSize: "30", arraySize: "5"},
     {from: "Decimal38Sparse", to: "VarChar", major: "DecimalComplexVarChar", bufferSize: "40", arraySize: "6"},
+    {from: "VarDecimal", to: "VarChar", major: "DecimalComplexVarChar", bufferSize: "40", arraySize: "6"},
 
     {from: "Decimal18", to: "Decimal9", major: "DownwardDecimalSimpleDecimalSimple", javatype: "int"},
 
@@ -177,6 +190,7 @@
     {from: "VarChar", to: "NullableDecimal18", major: "EmptyStringVarCharDecimalSimple", javatype: "long"},
     {from: "VarChar", to: "NullableDecimal28Sparse", major: "EmptyStringVarCharDecimalComplex", arraySize: "5"},
     {from: "VarChar", to: "NullableDecimal38Sparse", major: "EmptyStringVarCharDecimalComplex", arraySize: "6"},
+    {from: "VarChar", to: "NullableVarDecimal", major: "EmptyStringVarCharDecimalComplex", arraySize: "6"},
 
     {from: "NullableVarChar", to: "NullableInt", major: "EmptyString", javaType:"Integer", primeType:"int"},
     {from: "NullableVarChar", to: "NullableBigInt", major: "EmptyString", javaType: "Long", primeType: "long"},
@@ -187,6 +201,7 @@
     {from: "NullableVarChar", to: "NullableDecimal18", major: "EmptyStringVarCharDecimalSimple", javatype: "long"},
     {from: "NullableVarChar", to: "NullableDecimal28Sparse", major: "EmptyStringVarCharDecimalComplex", arraySize: "5"},
     {from: "NullableVarChar", to: "NullableDecimal38Sparse", major: "EmptyStringVarCharDecimalComplex", arraySize: "6"},
+    {from: "NullableVarChar", to: "NullableVarDecimal", major: "EmptyStringVarCharDecimalComplex", arraySize: "6"},
 
     {from: "NullableVar16Char", to: "NullableInt", major: "EmptyString", javaType:"Integer", primeType:"int"},
     {from: "NullableVar16Char", to: "NullableBigInt", major: "EmptyString", javaType: "Long", primeType: "long"},
@@ -197,6 +212,7 @@
     {from: "NullableVar16Char", to: "NullableDecimal18", major: "EmptyStringVarCharDecimalSimple", javatype: "long"},
     {from: "NullableVar16Char", to: "NullableDecimal28Sparse", major: "EmptyStringVarCharDecimalComplex", arraySize: "5"},
     {from: "NullableVar16Char", to: "NullableDecimal38Sparse", major: "EmptyStringVarCharDecimalComplex", arraySize: "6"},
+    {from: "NullableVar16Char", to: "NullableVarDecimal", major: "EmptyStringVarCharDecimalComplex", arraySize: "6"},
 
     {from: "NullableVarBinary", to: "NullableInt", major: "EmptyString", javaType:"Integer", primeType:"int"},
     {from: "NullableVarBinary", to: "NullableBigInt", major: "EmptyString", javaType: "Long", primeType: "long"},
@@ -207,5 +223,6 @@
     {from: "NullableVarBinary", to: "NullableDecimal18", major: "EmptyStringVarCharDecimalSimple", javatype: "long"},
     {from: "NullableVarBinary", to: "NullableDecimal28Sparse", major: "EmptyStringVarCharDecimalComplex", arraySize: "5"},
     {from: "NullableVarBinary", to: "NullableDecimal38Sparse", major: "EmptyStringVarCharDecimalComplex", arraySize: "6"},
+    {from: "NullableVarBinary", to: "NullableVarDecimal", major: "EmptyStringVarCharDecimalComplex", arraySize: "6"},
   ]
 } 
diff --git a/exec/java-exec/src/main/codegen/data/ComparisonTypesDecimal.tdd b/exec/java-exec/src/main/codegen/data/ComparisonTypesDecimal.tdd
index 423fe895d..db73825d9 100644
--- a/exec/java-exec/src/main/codegen/data/ComparisonTypesDecimal.tdd
+++ b/exec/java-exec/src/main/codegen/data/ComparisonTypesDecimal.tdd
@@ -13,8 +13,11 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+#
+# TODO: the assumption that VarDecimal takes 6 bytes might not always apply
 {
   { decimalTypes: [
+      {name: "VarDecimal", storage: "6"},
       {name: "Decimal28Sparse", storage: "5"},
       {name: "Decimal38Sparse", storage: "6"},
       {name: "Decimal28Dense", storage: "4"},
@@ -23,4 +26,4 @@
       {name: "Decimal18", storage: "long"}
     ]
   }
-}
\ No newline at end of file
+}
diff --git a/exec/java-exec/src/main/codegen/data/CountAggrTypes.tdd b/exec/java-exec/src/main/codegen/data/CountAggrTypes.tdd
index aec73d1a0..188517a68 100644
--- a/exec/java-exec/src/main/codegen/data/CountAggrTypes.tdd
+++ b/exec/java-exec/src/main/codegen/data/CountAggrTypes.tdd
@@ -39,6 +39,7 @@
     "VarBinary"
     "NullableVarBinary",
     "RepeatedVarBinary"
+    "VarDecimal",
     "Decimal9",
     "NullableDecimal9",
     "RepeatedDecimal9",
diff --git a/exec/java-exec/src/main/codegen/data/DecimalAggrTypes1.tdd b/exec/java-exec/src/main/codegen/data/DecimalAggrTypes1.tdd
index 92e93bb83..f24629140 100644
--- a/exec/java-exec/src/main/codegen/data/DecimalAggrTypes1.tdd
+++ b/exec/java-exec/src/main/codegen/data/DecimalAggrTypes1.tdd
@@ -24,7 +24,9 @@
       {inputType: "Decimal28Sparse", outputType: "Decimal28Sparse", runningType: "Decimal28Sparse"},
       {inputType: "NullableDecimal28Sparse", outputType: "Decimal28Sparse", runningType: "Decimal28Sparse"},
       {inputType: "Decimal38Sparse", outputType: "Decimal38Sparse", runningType: "Decimal38Sparse"},
-      {inputType: "NullableDecimal38Sparse", outputType: "Decimal38Sparse", runningType: "Decimal38Sparse"}
+      {inputType: "NullableDecimal38Sparse", outputType: "Decimal38Sparse", runningType: "Decimal38Sparse"},
+      {inputType: "VarDecimal", outputType: "VarDecimal", runningType: "VarDecimal"},
+      {inputType: "NullableVarDecimal", outputType: "VarDecimal", runningType: "VarDecimal"}
      ]
    },
    {className: "Min", funcName: "min", types: [
@@ -35,7 +37,9 @@
       {inputType: "Decimal28Sparse", outputType: "Decimal28Sparse", runningType: "Decimal28Sparse"},
       {inputType: "NullableDecimal28Sparse", outputType: "Decimal28Sparse", runningType: "Decimal28Sparse"},
       {inputType: "Decimal38Sparse", outputType: "Decimal38Sparse", runningType: "Decimal38Sparse"},
-      {inputType: "NullableDecimal38Sparse", outputType: "Decimal38Sparse", runningType: "Decimal38Sparse"}
+      {inputType: "NullableDecimal38Sparse", outputType: "Decimal38Sparse", runningType: "Decimal38Sparse"},
+      {inputType: "VarDecimal", outputType: "VarDecimal", runningType: "VarDecimal"},
+      {inputType: "NullableVarDecimal", outputType: "VarDecimal", runningType: "VarDecimal"}
      ]
    }
       {className: "Sum", funcName: "sum", types: [
@@ -46,7 +50,9 @@
         {inputType: "Decimal28Sparse", outputType: "Decimal38Sparse", major: "Numeric"},
         {inputType: "NullableDecimal28Sparse", outputType: "Decimal38Sparse", major: "Numeric"},
         {inputType: "Decimal38Sparse", outputType: "Decimal38Sparse", major: "Numeric"},
-        {inputType: "NullableDecimal38Sparse", outputType: "Decimal38Sparse", major: "Numeric"}
+        {inputType: "NullableDecimal38Sparse", outputType: "Decimal38Sparse", major: "Numeric"},
+        {inputType: "VarDecimal", outputType: "VarDecimal", major: "Numeric"},
+        {inputType: "NullableVarDecimal", outputType: "VarDecimal", major: "Numeric"}
         ]
       }
    ]
diff --git a/exec/java-exec/src/main/codegen/data/DecimalAggrTypes2.tdd b/exec/java-exec/src/main/codegen/data/DecimalAggrTypes2.tdd
index 5aa8b7fba..faa0afd9f 100644
--- a/exec/java-exec/src/main/codegen/data/DecimalAggrTypes2.tdd
+++ b/exec/java-exec/src/main/codegen/data/DecimalAggrTypes2.tdd
@@ -24,7 +24,9 @@
         {inputType: "Decimal28Sparse", outputType: "Decimal38Sparse", countRunningType: "BigInt", major: "Numeric"},
         {inputType: "NullableDecimal28Sparse", outputType: "Decimal38Sparse", countRunningType: "BigInt", major: "Numeric"},
         {inputType: "Decimal38Sparse", outputType: "Decimal38Sparse", countRunningType: "BigInt", major: "Numeric"},
-        {inputType: "NullableDecimal38Sparse", outputType: "Decimal38Sparse", countRunningType: "BigInt", major: "Numeric"}
+        {inputType: "NullableDecimal38Sparse", outputType: "Decimal38Sparse", countRunningType: "BigInt", major: "Numeric"},
+        {inputType: "VarDecimal", outputType: "VarDecimal", countRunningType: "BigInt", major: "Numeric"},
+        {inputType: "NullableVarDecimal", outputType: "VarDecimal", countRunningType: "BigInt", major: "Numeric"}
       ]
     }
   ]
diff --git a/exec/java-exec/src/main/codegen/data/NumericTypes.tdd b/exec/java-exec/src/main/codegen/data/NumericTypes.tdd
index 90b432e23..afc0ee817 100644
--- a/exec/java-exec/src/main/codegen/data/NumericTypes.tdd
+++ b/exec/java-exec/src/main/codegen/data/NumericTypes.tdd
@@ -15,7 +15,7 @@
 # limitations under the License.
 
  {
-   numeric: ["Int", "BigInt", "TinyInt", "SmallInt", "UInt1", "UInt2", "UInt4", "UInt8", "Float4", "Float8",
+   numeric: ["Int", "BigInt", "TinyInt", "SmallInt", "UInt1", "UInt2", "UInt4", "UInt8", "Float4", "Float8", "VarDecimal",
              "Decimal9", "Decimal18", "Decimal28Dense", "Decimal28Sparse", "Decimal38Dense", "Decimal38Sparse"],
 
   numericFunctions: [
diff --git a/exec/java-exec/src/main/codegen/templates/DateIntervalFunctionTemplates/ToTimeStampFunction.java b/exec/java-exec/src/main/codegen/templates/DateIntervalFunctionTemplates/ToTimeStampFunction.java
index 8dacca865..63c761497 100644
--- a/exec/java-exec/src/main/codegen/templates/DateIntervalFunctionTemplates/ToTimeStampFunction.java
+++ b/exec/java-exec/src/main/codegen/templates/DateIntervalFunctionTemplates/ToTimeStampFunction.java
@@ -59,8 +59,11 @@ public class G${numerics}ToTimeStamp implements DrillSimpleFunc {
     public void eval() {
         long inputMillis = 0;
 
-        <#if (numerics.startsWith("Decimal"))>
-        <#if (numerics == "Decimal9") || (numerics == "Decimal18")>
+        <#if (numerics.contains("Decimal"))>
+        <#if (numerics == "VarDecimal")>
+        java.math.BigDecimal input = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(left.buffer, left.start, left.end - left.start, left.scale);
+        inputMillis = input.multiply(new java.math.BigDecimal(1000)).longValue();
+        <#elseif (numerics == "Decimal9") || (numerics == "Decimal18")>
         java.math.BigInteger value = java.math.BigInteger.valueOf(left.value);
         value = value.multiply(millisConstant);
         inputMillis = (new java.math.BigDecimal(value, left.scale)).longValue();
diff --git a/exec/java-exec/src/main/codegen/templates/Decimal/CastDecimalFloat.java b/exec/java-exec/src/main/codegen/templates/Decimal/CastDecimalFloat.java
index 1acbe5ebd..fe794094f 100644
--- a/exec/java-exec/src/main/codegen/templates/Decimal/CastDecimalFloat.java
+++ b/exec/java-exec/src/main/codegen/templates/Decimal/CastDecimalFloat.java
@@ -97,7 +97,9 @@ public class Cast${type.from}${type.to} implements DrillSimpleFunc {
 
     public void eval() {
 
-        <#if type.from == "Decimal28Dense" || type.from == "Decimal38Dense">
+        <#if type.from == "VarDecimal">
+        java.math.BigDecimal bigDecimal = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(in.buffer, in.start, in.end - in.start, in.scale);
+        <#elseif type.from == "Decimal28Dense" || type.from == "Decimal38Dense">
         java.math.BigDecimal bigDecimal = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDense(in.buffer, in.start, in.nDecimalDigits, in.scale, in.maxPrecision, in.WIDTH);
         <#else>
         java.math.BigDecimal bigDecimal = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(in.buffer, in.start, in.nDecimalDigits, in.scale, true);
diff --git a/exec/java-exec/src/main/codegen/templates/Decimal/CastDecimalInt.java b/exec/java-exec/src/main/codegen/templates/Decimal/CastDecimalInt.java
index 37f35f381..3641818a9 100644
--- a/exec/java-exec/src/main/codegen/templates/Decimal/CastDecimalInt.java
+++ b/exec/java-exec/src/main/codegen/templates/Decimal/CastDecimalInt.java
@@ -101,6 +101,12 @@ public class Cast${type.from}${type.to} implements DrillSimpleFunc {
 
     public void eval() {
 
+<#if type.from.contains("VarDecimal")>
+        java.math.BigDecimal bd = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(in.buffer, in.start, in.end - in.start, in.scale);
+        bd.setScale(0, java.math.BigDecimal.ROUND_HALF_UP);
+        long lval = bd.longValue();   // round off to nearest integer
+        out.value = (${type.javatype}) lval;
+<#else>
         int carry = (org.apache.drill.exec.util.DecimalUtility.getFirstFractionalDigit(in.buffer, in.scale, in.start, in.nDecimalDigits) > 4) ? 1 : 0;
 
         // Get the index, where the integer part of the decimal ends
@@ -116,8 +122,9 @@ public class Cast${type.from}${type.to} implements DrillSimpleFunc {
         if (in.getSign(in.start, in.buffer) == true) {
             out.value *= -1;
         }
+</#if>
     }
 }
 
 </#if> <#-- type.major -->
-</#list>
\ No newline at end of file
+</#list>
diff --git a/exec/java-exec/src/main/codegen/templates/Decimal/CastDecimalSparseVarDecimal.java b/exec/java-exec/src/main/codegen/templates/Decimal/CastDecimalSparseVarDecimal.java
new file mode 100644
index 000000000..ae068c377
--- /dev/null
+++ b/exec/java-exec/src/main/codegen/templates/Decimal/CastDecimalSparseVarDecimal.java
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+<@pp.dropOutputFile />
+
+<#list cast.types as type>
+<#if type.major == "DecimalSparseVarDecimal">
+
+<@pp.changeOutputFile name="/org/apache/drill/exec/expr/fn/impl/gcast/Cast${type.from}${type.to}.java" />
+
+<#include "/@includes/license.ftl" />
+
+package org.apache.drill.exec.expr.fn.impl.gcast;
+
+<#include "/@includes/vv_imports.ftl" />
+
+import org.apache.drill.exec.expr.DrillSimpleFunc;
+import org.apache.drill.exec.expr.annotations.FunctionTemplate;
+import org.apache.drill.exec.expr.annotations.FunctionTemplate.NullHandling;
+import org.apache.drill.exec.expr.annotations.Output;
+import org.apache.drill.exec.expr.annotations.Param;
+import org.apache.drill.exec.expr.holders.*;
+import org.apache.drill.exec.record.RecordBatch;
+import org.apache.drill.exec.expr.annotations.Workspace;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.DrillBuf;
+
+import java.nio.ByteBuffer;
+
+/*
+ * This class is generated using freemarker and the ${.template_name} template.
+ */
+
+@SuppressWarnings("unused")
+@FunctionTemplate(name = "cast${type.to?upper_case}",
+    scope = FunctionTemplate.FunctionScope.SIMPLE,
+    returnType = FunctionTemplate.ReturnType.DECIMAL_CAST,
+    nulls = NullHandling.NULL_IF_NULL)
+public class Cast${type.from}${type.to} implements DrillSimpleFunc{
+
+    @Param ${type.from}Holder in;
+    @Inject DrillBuf buffer;
+    @Param BigIntHolder precision;
+    @Param BigIntHolder scale;
+    @Output ${type.to}Holder out;
+
+    public void setup() {
+        // VarDecimal size in bytes is determined once the input BigDecimal is known
+    }
+
+    public void eval() {
+        java.math.BigDecimal bd = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(in.buffer, in.start, in.nDecimalDigits, in.scale, true);
+        out.scale = (int) scale.value;
+        int len = bd.unscaledValue().toByteArray().length;
+        buffer = buffer.reallocIfNeeded(len);
+        out.start = 0;
+        out.buffer = buffer;
+        org.apache.drill.exec.util.DecimalUtility.getVarDecimalFromBigDecimal(bd, out.buffer, out.start);
+        out.end = out.start + len;
+    }
+}
+</#if>
+</#list>
diff --git a/exec/java-exec/src/main/codegen/templates/Decimal/CastDecimalVarchar.java b/exec/java-exec/src/main/codegen/templates/Decimal/CastDecimalVarchar.java
index e4c221ee8..c4ac6d7cf 100644
--- a/exec/java-exec/src/main/codegen/templates/Decimal/CastDecimalVarchar.java
+++ b/exec/java-exec/src/main/codegen/templates/Decimal/CastDecimalVarchar.java
@@ -157,6 +157,14 @@ public class Cast${type.from}${type.to} implements DrillSimpleFunc {
 
     public void eval() {
 
+<#if type.from.contains("VarDecimal")>
+        java.math.BigDecimal bigDecimal = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(in.buffer, in.start, in.end - in.start, in.scale);
+        String str = bigDecimal.toString();
+        out.buffer = buffer;
+        out.start = 0;
+        out.end = Math.min((int)len.value, str.length());
+        out.buffer.setBytes(0, String.valueOf(str.substring(0, out.end)).getBytes());
+<#else>
         StringBuilder str = new StringBuilder();
         int index = 0;
         int fractionalStartIndex = ${type.arraySize} - org.apache.drill.exec.util.DecimalUtility.roundUp(in.scale);
@@ -228,7 +236,8 @@ public class Cast${type.from}${type.to} implements DrillSimpleFunc {
         out.start = 0;
         out.end = Math.min((int)len.value, str.length()); // truncate if target type has length smaller than that of input's string
         out.buffer.setBytes(0, String.valueOf(str.substring(0,out.end)).getBytes());
+</#if>
     }
 }
 </#if> <#-- type.major -->
-</#list>
\ No newline at end of file
+</#list>
diff --git a/exec/java-exec/src/main/codegen/templates/Decimal/CastFloatDecimal.java b/exec/java-exec/src/main/codegen/templates/Decimal/CastFloatDecimal.java
index 95508a1ed..ba8f6510e 100644
--- a/exec/java-exec/src/main/codegen/templates/Decimal/CastFloatDecimal.java
+++ b/exec/java-exec/src/main/codegen/templates/Decimal/CastFloatDecimal.java
@@ -75,9 +75,24 @@ public class Cast${type.from}${type.to} implements DrillSimpleFunc {
     public void eval() {
 
         out.scale = (int) scale.value;
+
+        <#if !type.to.endsWith("VarDecimal")>
         out.precision = (int) precision.value;
+        </#if>
 
-        <#if type.major == "FloatDecimalComplex" || type.major == "DoubleDecimalComplex">
+        <#if type.to.endsWith("VarDecimal")>
+        // TODO: the entire set of logic, above, appears to be using BigDecimal constructors completely wrong!!!!
+        // for the time being, this new logic is added for VarDecimal only.
+        double d = in.value;
+        for (int i = 0; i < out.scale; ++i) {  // loop to compute unscaled value
+            d *= 10.0;
+        }
+        long lval = (long)(d >= 0.0 ? d + 0.5 : d - 0.5);  // round off unscaled integer, from float8 to nearest integer
+        java.math.BigInteger bi = new java.math.BigInteger(Long.toString(lval));
+        java.math.BigDecimal bd = new java.math.BigDecimal(bi, out.scale);
+        int len = org.apache.drill.exec.util.DecimalUtility.getVarDecimalFromBigDecimal(bd, out.buffer, out.start);
+        out.end = out.start + len;
+        <#elseif type.major == "FloatDecimalComplex" || type.major == "DoubleDecimalComplex">
         out.start = 0;
         out.buffer = buffer;
 
diff --git a/exec/java-exec/src/main/codegen/templates/Decimal/CastIntDecimal.java b/exec/java-exec/src/main/codegen/templates/Decimal/CastIntDecimal.java
index abacf4900..15304daf5 100644
--- a/exec/java-exec/src/main/codegen/templates/Decimal/CastIntDecimal.java
+++ b/exec/java-exec/src/main/codegen/templates/Decimal/CastIntDecimal.java
@@ -54,7 +54,7 @@ import java.nio.ByteBuffer;
 public class Cast${type.from}${type.to} implements DrillSimpleFunc {
 
     @Param ${type.from}Holder in;
-    <#if type.to.startsWith("Decimal28") || type.to.startsWith("Decimal38")>
+    <#if type.to.startsWith("Decimal28") || type.to.startsWith("Decimal38") || type.to.endsWith("VarDecimal")>
     @Inject DrillBuf buffer;
     </#if>
     @Param BigIntHolder precision;
@@ -71,15 +71,31 @@ public class Cast${type.from}${type.to} implements DrillSimpleFunc {
 
     public void eval() {
         out.scale = (int) scale.value;
+
+        <#if !type.to.endsWith("VarDecimal")>
         out.precision = (int) precision.value;
+        </#if>
 
-        <#if type.to == "Decimal9" || type.to == "Decimal18">
+        <#if type.to.endsWith("VarDecimal")>
+        out.start = 0;
+        out.buffer = buffer;
+        StringBuilder sb = new StringBuilder();
+        sb.append(Long.toString((long)in.value));
+        for (int i = 0; i < out.scale; ++i) {  // add 0's to get unscaled integer
+            sb.append("0");
+        }
+        java.math.BigInteger bi = new java.math.BigInteger(sb.toString());
+        java.math.BigDecimal bd = new java.math.BigDecimal(bi, out.scale);
+        int len = org.apache.drill.exec.util.DecimalUtility.getVarDecimalFromBigDecimal(bd, out.buffer, out.start);
+        out.end = out.start + len;
+        <#elseif type.to == "Decimal9" || type.to == "Decimal18">
         out.value = (${type.javatype}) in.value;
 
         // converting from integer to decimal, pad zeroes if scale is non zero
         out.value = (${type.javatype}) org.apache.drill.exec.util.DecimalUtility.adjustScaleMultiply(out.value, (int) scale.value);
 
         <#else>
+
         out.start = 0;
         out.buffer = buffer;
 
@@ -107,4 +123,4 @@ public class Cast${type.from}${type.to} implements DrillSimpleFunc {
     }
 }
 </#if> <#-- type.major -->
-</#list>
\ No newline at end of file
+</#list>
diff --git a/exec/java-exec/src/main/codegen/templates/Decimal/CastVarCharDecimal.java b/exec/java-exec/src/main/codegen/templates/Decimal/CastVarCharDecimal.java
index a4341e98c..851d424af 100644
--- a/exec/java-exec/src/main/codegen/templates/Decimal/CastVarCharDecimal.java
+++ b/exec/java-exec/src/main/codegen/templates/Decimal/CastVarCharDecimal.java
@@ -91,6 +91,25 @@ public class CastEmptyString${type.from}To${type.to} implements DrillSimpleFunc
 
         // Assign the scale and precision
         out.scale = (int) scale.value;
+
+        <#if type.to.endsWith("VarDecimal")>
+
+        // VarDecimal gets its own cast logic
+        byte[] buf = new byte[in.end - in.start];
+        buffer.getBytes(in.start, buf, 0, in.end - in.start);
+        String s = new String(buf, Charsets.UTF_8);
+        Double d = Double.valueOf(s);
+        for (int i = 0; i < out.scale; ++i) {  // loop to compute unscaled value
+            d *= 10.0;
+        }
+        long lval = (long)(d >= 0.0 ? d + 0.5 : d - 0.5);  // round off unscaled integer, and hope it's close enough
+        java.math.BigInteger bi = new java.math.BigInteger(Long.toString(lval));
+        java.math.BigDecimal bd = new java.math.BigDecimal(bi, out.scale);
+        int len = org.apache.drill.exec.util.DecimalUtility.getVarDecimalFromBigDecimal(bd, out.buffer, out.start);
+        out.end = out.start + len;
+
+        <#else>   // !VarDecimal
+
         out.precision = (int) precision.value;
 
         int readIndex = in.start;
@@ -200,6 +219,7 @@ public class CastEmptyString${type.from}To${type.to} implements DrillSimpleFunc
         if (negative == true) {
             out.value *= -1;
         }
+        </#if>
     }
 }
 
@@ -279,6 +299,25 @@ public class CastEmptyString${type.from}To${type.to} implements DrillSimpleFunc
         out.start  = 0;
 
         out.scale = (int) scale.value;
+
+        <#if type.to.endsWith("VarDecimal")>
+
+        // VarDecimal gets its own cast logic
+        byte[] buf = new byte[in.end - in.start];
+        buffer.getBytes(in.start, buf, 0, in.end - in.start);
+        String s = new String(buf, Charsets.UTF_8);
+        Double d = Double.valueOf(s);
+        for (int i = 0; i < out.scale; ++i) {  // loop to compute unscaled value
+            d *= 10.0;
+        }
+        long lval = (long)(d >= 0.0 ? d + 0.5 : d - 0.5);  // round off unscaled integer, and hope it's close enough
+        java.math.BigInteger bi = new java.math.BigInteger(Long.toString(lval));
+        java.math.BigDecimal bd = new java.math.BigDecimal(bi, out.scale);
+        int len = org.apache.drill.exec.util.DecimalUtility.getVarDecimalFromBigDecimal(bd, out.buffer, out.start);
+        out.end = out.start + len;
+
+        <#else>
+
         out.precision = (int) precision.value;
         boolean sign = false;
 
@@ -487,7 +526,8 @@ public class CastEmptyString${type.from}To${type.to} implements DrillSimpleFunc
             } while (carry > 0 && decimalBufferIndex >= 0);
         }
         out.setSign(sign, out.start, out.buffer);
+        </#if>
     }
 }
 </#if> <#-- type.major -->
-</#list>
\ No newline at end of file
+</#list>
diff --git a/exec/java-exec/src/main/codegen/templates/Decimal/DecimalAggrTypeFunctions1.java b/exec/java-exec/src/main/codegen/templates/Decimal/DecimalAggrTypeFunctions1.java
index 55ce0772e..893438ff4 100644
--- a/exec/java-exec/src/main/codegen/templates/Decimal/DecimalAggrTypeFunctions1.java
+++ b/exec/java-exec/src/main/codegen/templates/Decimal/DecimalAggrTypeFunctions1.java
@@ -167,7 +167,7 @@ public static class ${type.inputType}${aggrtype.className} implements DrillAggFu
    <#if type.inputType.endsWith("Decimal9") || type.inputType.endsWith("Decimal18")>
     java.math.BigDecimal currentValue = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromPrimitiveTypes(in.value, in.scale, in.precision);
     <#else>
-    java.math.BigDecimal currentValue = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromSparse(in.buffer, in.start, in.nDecimalDigits, in.scale);
+    java.math.BigDecimal currentValue = in.getBigDecimal();
     </#if>
     value.obj = ((java.math.BigDecimal)(value.obj)).add(currentValue);
     if (outputScale.value == Integer.MIN_VALUE) {
@@ -187,9 +187,14 @@ public static class ${type.inputType}${aggrtype.className} implements DrillAggFu
     out.buffer = buffer;
     out.start  = 0;
     out.scale = outputScale.value;
-    out.precision = 38;
     value.obj = ((java.math.BigDecimal) (value.obj)).setScale(out.scale, java.math.BigDecimal.ROUND_HALF_UP);
+<#if type.inputType.contains("VarDecimal")>
+    int len = org.apache.drill.exec.util.DecimalUtility.getVarDecimalFromBigDecimal((java.math.BigDecimal) value.obj, out.buffer, out.start);
+    out.end = out.start + len;
+<#else>
+    out.precision = 38;
     org.apache.drill.exec.util.DecimalUtility.getSparseFromBigDecimal((java.math.BigDecimal) value.obj, out.buffer, out.start, out.scale, out.precision, out.nDecimalDigits);
+</#if>
    <#else>
     <#if type.outputType.endsWith("Sparse")>
     org.apache.drill.exec.util.Text tmp = (org.apache.drill.exec.util.Text) value.obj;
diff --git a/exec/java-exec/src/main/codegen/templates/Decimal/DecimalAggrTypeFunctions2.java b/exec/java-exec/src/main/codegen/templates/Decimal/DecimalAggrTypeFunctions2.java
index e2da4282b..df1eb7e81 100644
--- a/exec/java-exec/src/main/codegen/templates/Decimal/DecimalAggrTypeFunctions2.java
+++ b/exec/java-exec/src/main/codegen/templates/Decimal/DecimalAggrTypeFunctions2.java
@@ -93,7 +93,7 @@ public static class ${type.inputType}${aggrtype.className} implements DrillAggFu
    <#if type.inputType.endsWith("Decimal9") || type.inputType.endsWith("Decimal18")>
     java.math.BigDecimal currentValue = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromPrimitiveTypes(in.value, in.scale, in.precision);
     <#else>
-    java.math.BigDecimal currentValue = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromSparse(in.buffer, in.start, in.nDecimalDigits, in.scale);
+    java.math.BigDecimal currentValue = in.getBigDecimal();
     </#if>
     value.obj = ((java.math.BigDecimal)(value.obj)).add(currentValue);
     if (outputScale.value == Integer.MIN_VALUE) {
@@ -109,9 +109,11 @@ public static class ${type.inputType}${aggrtype.className} implements DrillAggFu
     out.buffer = buffer;
     out.start  = 0;
     out.scale = outputScale.value;
-    out.precision = 38;
     java.math.BigDecimal average = ((java.math.BigDecimal)(value.obj)).divide(java.math.BigDecimal.valueOf(count.value, 0), out.scale, java.math.BigDecimal.ROUND_HALF_UP);
+<#if !type.inputType.contains("VarDecimal")>
+    out.precision = 38;
     org.apache.drill.exec.util.DecimalUtility.getSparseFromBigDecimal(average, out.buffer, out.start, out.scale, out.precision, out.nDecimalDigits);
+</#if>
   }
 
   @Override
diff --git a/exec/java-exec/src/main/codegen/templates/Decimal/DecimalFunctions.java b/exec/java-exec/src/main/codegen/templates/Decimal/DecimalFunctions.java
index 721a9d49c..53af71f62 100644
--- a/exec/java-exec/src/main/codegen/templates/Decimal/DecimalFunctions.java
+++ b/exec/java-exec/src/main/codegen/templates/Decimal/DecimalFunctions.java
@@ -85,6 +85,15 @@ import org.apache.drill.exec.expr.annotations.Workspace;
           } // outside
 </#macro>
 
+<#macro varCompareBlock leftType rightType absCompare output nullCompare nullComparesHigh>
+        outside:
+        {
+          <@compareNullsSubblock leftType=leftType rightType=rightType output=output breakTarget="outside" nullCompare=nullCompare nullComparesHigh=nullComparesHigh />
+
+          ${output} = org.apache.drill.exec.util.DecimalUtility.compareVarLenBytes(left.buffer, left.start, left.end, left.scale, right.buffer, right.start, right.end, right.scale, ${absCompare});
+        } // outside
+</#macro>
+
 <#macro adjustScale javaType leftType rightType>
 
             // Adjust the scale of the two inputs to be the same
@@ -101,7 +110,625 @@ import org.apache.drill.exec.expr.annotations.Workspace;
 <#-- For each DECIMAL... type (in DecimalTypes.tdd) ... -->
 <#list comparisonTypesDecimal.decimalTypes as type>
 
-<#if type.name.endsWith("Sparse")>
+<#if type.name.endsWith("VarDecimal")>
+
+<@pp.changeOutputFile name="/org/apache/drill/exec/expr/fn/impl/${type.name}Functions.java" />
+
+<#include "/@includes/license.ftl" />
+
+package org.apache.drill.exec.expr.fn.impl;
+
+<#include "/@includes/vv_imports.ftl" />
+
+import org.apache.drill.exec.expr.DrillSimpleFunc;
+import org.apache.drill.exec.expr.annotations.FunctionTemplate;
+import org.apache.drill.exec.expr.annotations.FunctionTemplate.NullHandling;
+import org.apache.drill.exec.expr.annotations.Output;
+import org.apache.drill.exec.expr.annotations.Param;
+import org.apache.drill.exec.expr.annotations.Workspace;
+import org.apache.drill.exec.expr.fn.FunctionGenerationHelper;
+import org.apache.drill.exec.expr.holders.*;
+import org.apache.drill.exec.record.RecordBatch;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.DrillBuf;
+
+import java.nio.ByteBuffer;
+
+@SuppressWarnings("unused")
+public class ${type.name}Functions {
+    private static void initBuffer(DrillBuf buffer) {
+        // for VarDecimal, this method of setting initial size is actually only a very rough heuristic.
+        int size = (${type.storage} * (org.apache.drill.exec.util.DecimalUtility.INTEGER_SIZE));
+        buffer = buffer.reallocIfNeeded(size);
+     }
+
+    @FunctionTemplate(name = "subtract",
+        scope = FunctionTemplate.FunctionScope.SIMPLE,
+        returnType = FunctionTemplate.ReturnType.DECIMAL_ADD_SCALE,
+        nulls = NullHandling.NULL_IF_NULL,
+        checkPrecisionRange = true)
+    public static class ${type.name}SubtractFunction implements DrillSimpleFunc {
+
+        @Param ${type.name}Holder left;
+        @Param ${type.name}Holder right;
+        @Inject DrillBuf buffer;
+        @Workspace int outputScale;
+        @Output ${type.name}Holder result;
+
+        public void setup() {
+            initBuffer(buffer);
+        }
+
+        public void eval(){
+            outputScale=Math.max(left.scale,right.scale);   // output scale is the maximum of the two scales, to represent result exactly
+            result.scale=outputScale;
+            result.buffer=buffer;
+            result.start = 0;
+
+            java.math.BigDecimal leftInput=org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(left.buffer,left.start,left.end-left.start,left.scale);
+            java.math.BigDecimal rightInput=org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(right.buffer,right.start,right.end-right.start,right.scale);
+            java.math.BigDecimal addResult=leftInput.subtract(rightInput);
+
+            // Set the scale
+            addResult.setScale(result.scale,java.math.BigDecimal.ROUND_HALF_UP);
+
+            byte[]bytes=addResult.unscaledValue().toByteArray();
+            int len=bytes.length;
+            buffer=buffer.reallocIfNeeded(len);
+            buffer.setBytes(0,bytes);
+            result.end=len;
+        }
+    }
+
+    @FunctionTemplate(name = "add",
+        scope = FunctionTemplate.FunctionScope.SIMPLE,
+        returnType = FunctionTemplate.ReturnType.DECIMAL_ADD_SCALE,
+        nulls = NullHandling.NULL_IF_NULL,
+        checkPrecisionRange = true)
+    public static class ${type.name}AddFunction implements DrillSimpleFunc {
+
+        @Param ${type.name}Holder left;
+        @Param ${type.name}Holder right;
+        @Workspace int outputScale;
+        @Inject DrillBuf buffer;
+        @Output ${type.name}Holder result;
+
+        public void setup() {
+            initBuffer(buffer);
+       }
+
+        public void eval() {
+            outputScale=Math.max(left.scale,right.scale);   // output scale is the maximum of the two scales, to represent result exactly
+            result.scale=outputScale;
+            result.buffer=buffer;
+            result.start = 0;
+
+            java.math.BigDecimal leftInput=org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(left.buffer,left.start,left.end-left.start,left.scale);
+            java.math.BigDecimal rightInput=org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(right.buffer,right.start,right.end-right.start,right.scale);
+            java.math.BigDecimal addResult=leftInput.add(rightInput);
+
+            // Set the scale
+            addResult.setScale(result.scale,java.math.BigDecimal.ROUND_HALF_UP);
+
+            byte[]bytes=addResult.unscaledValue().toByteArray();
+            int len=bytes.length;
+            buffer=buffer.reallocIfNeeded(len);
+            buffer.setBytes(0,bytes);
+            result.end=len;
+        }
+    }
+
+    @FunctionTemplate(name = "multiply",
+        scope = FunctionTemplate.FunctionScope.SIMPLE,
+        returnType = FunctionTemplate.ReturnType.DECIMAL_SUM_SCALE,
+        nulls = NullHandling.NULL_IF_NULL,
+        checkPrecisionRange = true)
+    public static class ${type.name}MultiplyFunction implements DrillSimpleFunc {
+
+        @Param ${type.name}Holder left;
+        @Param ${type.name}Holder right;
+        @Inject DrillBuf buffer;
+        @Workspace int outputScale;
+        @Output ${type.name}Holder result;
+
+        public void setup() {
+            initBuffer(buffer);
+        }
+
+        public void eval() {
+            outputScale=left.scale + right.scale;   // sum of scales is guaranteed to represent multiplication result exactly
+            result.scale=outputScale;
+            result.buffer=buffer;
+            result.start = 0;
+
+            java.math.BigDecimal leftInput=org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(left.buffer,left.start,left.end-left.start,left.scale);
+            java.math.BigDecimal rightInput=org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(right.buffer,right.start,right.end-right.start,right.scale);
+            java.math.BigDecimal output=leftInput.multiply(rightInput);
+
+            // Set the scale
+            output.setScale(result.scale,java.math.BigDecimal.ROUND_HALF_UP);
+
+            byte[]bytes=output.unscaledValue().toByteArray();
+            int len=bytes.length;
+            buffer=buffer.reallocIfNeeded(len);
+            buffer.setBytes(0,bytes);
+            result.end=len;
+        }
+    }
+
+    @FunctionTemplate(name = "exact_divide",
+        scope = FunctionTemplate.FunctionScope.SIMPLE,
+        returnType = FunctionTemplate.ReturnType.DECIMAL_DIV_SCALE,
+        nulls = NullHandling.NULL_IF_NULL,
+        checkPrecisionRange = true)
+    public static class ${type.name}DivideFunction implements DrillSimpleFunc {
+        @Param ${type.name}Holder left;
+        @Param ${type.name}Holder right;
+        @Output ${type.name}Holder result;
+        @Inject DrillBuf buffer;
+        @Workspace int outputScale;
+
+        public void setup() {
+            initBuffer(buffer);
+        }
+
+        public void eval() {
+            outputScale = left.scale + right.scale;   // preferred scale is heuristically set to sum of scales, even though rounding may be required
+            result.scale = outputScale;
+            result.buffer = buffer;
+            result.start = 0;
+
+            java.math.BigDecimal numerator = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(left.buffer, left.start, left.end - left.start, left.scale);
+            java.math.BigDecimal denominator = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(right.buffer, right.start, right.end - right.start, right.scale);
+
+            java.math.BigDecimal output = numerator.divide(denominator, (int) result.scale, java.math.BigDecimal.ROUND_HALF_UP);
+
+            byte[]bytes=output.unscaledValue().toByteArray();
+            int len=bytes.length;
+            buffer=buffer.reallocIfNeeded(len);
+            buffer.setBytes(0,bytes);
+            result.end=len;
+        }
+    }
+
+    @FunctionTemplate(name = "mod",
+        scope = FunctionTemplate.FunctionScope.SIMPLE,
+        returnType = FunctionTemplate.ReturnType.DECIMAL_MOD_SCALE,
+        nulls = NullHandling.NULL_IF_NULL,
+        checkPrecisionRange = true)
+    public static class ${type.name}ModFunction implements DrillSimpleFunc {
+
+        @Param ${type.name}Holder left;
+        @Param ${type.name}Holder right;
+        @Output ${type.name}Holder result;
+        @Inject DrillBuf buffer;
+        @Workspace int outputScale;
+
+        public void setup() {
+            initBuffer(buffer);
+        }
+
+        public void eval() {
+            outputScale = Math.max(left.scale,right.scale);   // for mod, as for add & subtract, use larger of the two input scales
+            result.scale = outputScale;
+            result.buffer = buffer;
+            result.start = 0;
+
+            java.math.BigDecimal numerator = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(left.buffer, left.start, left.end - left.start, left.scale);
+            java.math.BigDecimal denominator = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(right.buffer, right.start, right.end - right.start, right.scale);
+
+            java.math.BigDecimal output = numerator.remainder(denominator);
+            output.setScale(result.scale, java.math.BigDecimal.ROUND_HALF_UP);
+
+            byte[]bytes=output.unscaledValue().toByteArray();
+            int len=bytes.length;
+            buffer=buffer.reallocIfNeeded(len);
+            buffer.setBytes(0,bytes);
+            result.end=len;
+        }
+    }
+
+    @FunctionTemplate(name = "abs",
+        scope = FunctionTemplate.FunctionScope.SIMPLE,
+        returnType = FunctionTemplate.ReturnType.DECIMAL_MAX_SCALE,
+        nulls = NullHandling.NULL_IF_NULL)
+    public static class ${type.name}AbsFunction implements DrillSimpleFunc {
+
+        @Param  ${type.name}Holder in;
+        @Output ${type.name}Holder result;
+        @Inject DrillBuf buffer;
+
+        public void setup() {
+            initBuffer(buffer);
+        }
+
+        public void eval() {
+            result.scale = in.scale;
+            result.buffer = buffer;
+            result.start = 0;
+
+            java.math.BigDecimal bd = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(in.buffer, in.start, in.end - in.start, in.scale);
+            java.math.BigDecimal output = bd.abs();
+            output.setScale(result.scale, java.math.BigDecimal.ROUND_HALF_UP);
+
+            byte[]bytes=output.unscaledValue().toByteArray();
+            int len=bytes.length;
+            buffer=buffer.reallocIfNeeded(len);
+            buffer.setBytes(0,bytes);
+            result.end=len;
+        }
+    }
+
+    @FunctionTemplate(name = "sign", scope = FunctionTemplate.FunctionScope.SIMPLE, nulls = NullHandling.NULL_IF_NULL)
+    public static class ${type.name}SignFunction implements DrillSimpleFunc {
+
+        @Param ${type.name}Holder in;
+        @Output IntHolder out;
+
+        public void setup() {}
+
+        public void eval() {
+            java.math.BigDecimal bd = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(in.buffer, in.start, in.end - in.start, in.scale);
+            out.value = bd.signum();
+        }
+    }
+
+    @FunctionTemplate(names = {"ceil", "ceiling"},
+        scope = FunctionTemplate.FunctionScope.SIMPLE,
+        returnType = FunctionTemplate.ReturnType.DECIMAL_ZERO_SCALE,
+        nulls = NullHandling.NULL_IF_NULL)
+    public static class ${type.name}CeilFunction implements DrillSimpleFunc {
+
+        @Param  ${type.name}Holder in;
+        @Output ${type.name}Holder out;
+        @Inject DrillBuf buffer;
+
+        public void setup() {
+            initBuffer(buffer);
+        }
+
+        public void eval() {
+          out.scale = 0;
+          out.buffer = buffer;
+          out.start = 0;
+          java.math.BigDecimal bd = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(in.buffer, in.start, in.end - in.start, in.scale);
+          bd.setScale(0, java.math.BigDecimal.ROUND_CEILING);
+          byte[] bytes=bd.unscaledValue().toByteArray();
+          int len=bytes.length;
+          buffer=buffer.reallocIfNeeded(len);
+          buffer.setBytes(0,bytes);
+          out.end=len;
+        }
+    }
+
+    @FunctionTemplate(name = "floor",
+        scope = FunctionTemplate.FunctionScope.SIMPLE,
+        returnType = FunctionTemplate.ReturnType.DECIMAL_ZERO_SCALE,
+        nulls = NullHandling.NULL_IF_NULL)
+    public static class ${type.name}FloorFunction implements DrillSimpleFunc {
+
+        @Param  ${type.name}Holder in;
+        @Output ${type.name}Holder out;
+        @Inject DrillBuf buffer;
+
+        public void setup() {
+            initBuffer(buffer);
+        }
+
+        public void eval() {
+            out.scale = 0;
+            out.buffer = buffer;
+            out.start = 0;
+            java.math.BigDecimal bd = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(in.buffer, in.start, in.end - in.start, in.scale);
+            bd.setScale(0, java.math.BigDecimal.ROUND_FLOOR);
+            byte[] bytes=bd.unscaledValue().toByteArray();
+            int len=bytes.length;
+            buffer=buffer.reallocIfNeeded(len);
+            buffer.setBytes(0,bytes);
+            out.end=len;
+        }
+    }
+
+    @FunctionTemplate(names = {"trunc", "truncate"},
+        scope = FunctionTemplate.FunctionScope.SIMPLE,
+        returnType = FunctionTemplate.ReturnType.DECIMAL_ZERO_SCALE,
+        nulls = NullHandling.NULL_IF_NULL)
+    public static class ${type.name}TruncateFunction implements DrillSimpleFunc {
+
+        @Param  ${type.name}Holder in;
+        @Output ${type.name}Holder out;
+        @Inject DrillBuf buffer;
+
+        public void setup() {
+            initBuffer(buffer);
+        }
+
+        public void eval() {
+            out.scale = 0;
+            out.buffer = buffer;
+            out.start = 0;
+            java.math.BigDecimal bd = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(in.buffer, in.start, in.end - in.start, in.scale);
+            bd.setScale(0, java.math.BigDecimal.ROUND_DOWN);
+            byte[] bytes=bd.unscaledValue().toByteArray();
+            int len=bytes.length;
+            buffer=buffer.reallocIfNeeded(len);
+            buffer.setBytes(0,bytes);
+            out.end=len;
+        }
+    }
+
+    @FunctionTemplate(names = {"trunc", "truncate"},
+        scope = FunctionTemplate.FunctionScope.SIMPLE,
+        returnType = FunctionTemplate.ReturnType.DECIMAL_SET_SCALE,
+        nulls = NullHandling.NULL_IF_NULL)
+    public static class ${type.name}TruncateScaleFunction implements DrillSimpleFunc {
+
+        @Param  ${type.name}Holder left;
+        @Param  IntHolder right;
+        @Output ${type.name}Holder result;
+        @Inject DrillBuf buffer;
+
+        public void setup() {
+            initBuffer(buffer);
+        }
+
+        public void eval() {
+            result.scale = left.scale;
+            result.buffer = left.buffer;
+            result.start = left.start;
+            result.end = left.end;
+
+            // compute truncated value iff the indicated scale is less than the input's scale.
+            // otherwise, we just copy the input (left) to the result, with the same scale.
+            if (right.value < left.scale) {
+                result.buffer = buffer;
+                result.start = 0;
+                result.scale = right.value;
+                java.math.BigDecimal bd = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(left.buffer, left.start, left.end - left.start, left.scale);
+                bd.setScale(result.scale, java.math.BigDecimal.ROUND_DOWN);
+                byte[] bytes=bd.unscaledValue().toByteArray();
+                int len=bytes.length;
+                buffer=buffer.reallocIfNeeded(len);
+                buffer.setBytes(0,bytes);
+                result.end=len;
+            }
+        }
+    }
+
+    @FunctionTemplate(name = "round",
+        scope = FunctionTemplate.FunctionScope.SIMPLE,
+        returnType = FunctionTemplate.ReturnType.DECIMAL_ZERO_SCALE,
+        nulls = NullHandling.NULL_IF_NULL)
+    public static class ${type.name}RoundFunction implements DrillSimpleFunc {
+
+        @Param  ${type.name}Holder in;
+        @Output ${type.name}Holder out;
+        @Inject DrillBuf buffer;
+
+        public void setup() {
+            initBuffer(buffer);
+        }
+
+        public void eval() {
+            out.scale = 0;
+            out.buffer = buffer;
+            out.start = 0;
+            java.math.BigDecimal bd = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(in.buffer, in.start, in.end - in.start, in.scale);
+            bd.setScale(0, java.math.BigDecimal.ROUND_HALF_UP);
+            byte[] bytes=bd.unscaledValue().toByteArray();
+            int len=bytes.length;
+            buffer=buffer.reallocIfNeeded(len);
+            buffer.setBytes(0,bytes);
+            out.end=len;
+        }
+    }
+
+    @FunctionTemplate(name = "round",
+        scope = FunctionTemplate.FunctionScope.SIMPLE,
+        returnType = FunctionTemplate.ReturnType.DECIMAL_SET_SCALE,
+        nulls = NullHandling.NULL_IF_NULL)
+    public static class ${type.name}RoundScaleFunction implements DrillSimpleFunc {
+
+        @Param  ${type.name}Holder left;
+        @Param  IntHolder right;
+        @Output ${type.name}Holder result;
+        @Inject DrillBuf buffer;
+
+        public void setup() {
+            initBuffer(buffer);
+        }
+
+        public void eval() {
+            result.scale = right.value;
+            result.buffer = buffer;
+            result.start = 0;
+            java.math.BigDecimal bd = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(left.buffer, left.start, left.end - left.start, left.scale);
+            bd.setScale(result.scale, java.math.BigDecimal.ROUND_HALF_UP);
+            byte[] bytes=bd.unscaledValue().toByteArray();
+            int len=bytes.length;
+            buffer=buffer.reallocIfNeeded(len);
+            buffer.setBytes(0,bytes);
+            result.end=len;
+        }
+    }
+
+ <#-- Handle 2 x 2 combinations of nullable and non-nullable arguments. -->
+ <#list ["Nullable${type.name}", "${type.name}"] as leftType >
+ <#list ["Nullable${type.name}", "${type.name}"] as rightType >
+
+  <#-- Comparison function for sorting and grouping relational operators
+       (not for comparison expression operators (=, <, etc.)). -->
+  @FunctionTemplate(name = FunctionGenerationHelper.COMPARE_TO_NULLS_HIGH,
+      scope = FunctionTemplate.FunctionScope.SIMPLE,
+      returnType = FunctionTemplate.ReturnType.DECIMAL_MAX_SCALE,
+      nulls = NullHandling.INTERNAL)
+  public static class GCompare${leftType}Vs${rightType}NullHigh implements DrillSimpleFunc {
+
+    @Param ${leftType}Holder left;
+    @Param ${rightType}Holder right;
+    @Output IntHolder out;
+
+    public void setup() {}
+
+    public void eval() {
+      <@varCompareBlock leftType=leftType rightType=rightType absCompare="false" output="out.value" nullCompare=true nullComparesHigh=true />
+    }
+  }
+
+
+
+  <#-- Comparison function for sorting and grouping relational operators
+        (not for comparison expression operators (=, <, etc.)). -->
+  @FunctionTemplate(name = FunctionGenerationHelper.COMPARE_TO_NULLS_LOW,
+      scope = FunctionTemplate.FunctionScope.SIMPLE,
+      returnType = FunctionTemplate.ReturnType.DECIMAL_MAX_SCALE,
+      nulls = NullHandling.INTERNAL)
+  public static class GCompare${leftType}Vs${rightType}NullLow implements DrillSimpleFunc {
+
+    @Param ${leftType}Holder left;
+    @Param ${rightType}Holder right;
+    @Output IntHolder out;
+
+    public void setup() {}
+
+    public void eval() {
+      <@varCompareBlock leftType=leftType rightType=rightType absCompare="false" output="out.value" nullCompare=true nullComparesHigh=false />
+    }
+  }
+
+ </#list>
+ </#list>
+
+    <#-- Comparison function for comparison expression operator (=, &lt;, etc.),
+         not for sorting and grouping relational operators.) -->
+    @FunctionTemplate(name = "less than",
+        scope = FunctionTemplate.FunctionScope.SIMPLE,
+        returnType = FunctionTemplate.ReturnType.DECIMAL_MAX_SCALE,
+        nulls = NullHandling.NULL_IF_NULL)
+    public static class ${type.name}LessThan implements DrillSimpleFunc {
+
+        @Param ${type.name}Holder left;
+        @Param ${type.name}Holder right;
+        @Output BitHolder out;
+        public void setup() {}
+
+        public void eval() {
+            int cmp;
+            <@varCompareBlock leftType="leftType" rightType="rightType" absCompare="false" output="cmp" nullCompare=false nullComparesHigh=false />
+            out.value = cmp == -1 ? 1 : 0;
+        }
+    }
+
+
+    // TODO:  RESOLVE:  Here there are spaces in function template names, but
+    // elsewhere there are underlines.  Are things being looked up correctly?
+    <#-- Comparison function for comparison expression operator (=, &lt;, etc.),
+         not for sorting and grouping relational operators.) -->
+    @FunctionTemplate(name = "less than or equal to",
+        scope = FunctionTemplate.FunctionScope.SIMPLE,
+        returnType = FunctionTemplate.ReturnType.DECIMAL_MAX_SCALE,
+        nulls = NullHandling.NULL_IF_NULL)
+    public static class ${type.name}LessThanEq implements DrillSimpleFunc {
+
+        @Param ${type.name}Holder left;
+        @Param ${type.name}Holder right;
+        @Output BitHolder out;
+        public void setup() {}
+
+        public void eval() {
+            int cmp;
+            <@varCompareBlock leftType="leftType" rightType="rightType" absCompare="false" output="cmp" nullCompare=false nullComparesHigh=false />
+            out.value = cmp < 1 ? 1 : 0;
+        }
+    }
+
+
+    <#-- Comparison function for comparison expression operator (=, &lt;, etc.),
+         not for sorting and grouping relational operators.) -->
+    @FunctionTemplate(name = "greater than",
+        scope = FunctionTemplate.FunctionScope.SIMPLE,
+        returnType = FunctionTemplate.ReturnType.DECIMAL_MAX_SCALE,
+        nulls = NullHandling.NULL_IF_NULL)
+    public static class ${type.name}GreaterThan implements DrillSimpleFunc {
+
+        @Param ${type.name}Holder left;
+        @Param ${type.name}Holder right;
+        @Output BitHolder out;
+        public void setup() {}
+
+        public void eval() {
+            int cmp;
+            <@varCompareBlock leftType="leftType" rightType="rightType" absCompare="false" output="cmp" nullCompare=false nullComparesHigh=false />
+            out.value = cmp == 1 ? 1 : 0;
+        }
+    }
+
+
+    <#-- Comparison function for comparison expression operator (=, &lt;, etc.),
+         not for sorting and grouping relational operators.) -->
+    @FunctionTemplate(name = "greater than or equal to",
+        scope = FunctionTemplate.FunctionScope.SIMPLE,
+        returnType = FunctionTemplate.ReturnType.DECIMAL_MAX_SCALE,
+        nulls = NullHandling.NULL_IF_NULL)
+    public static class ${type.name}GreaterThanEq implements DrillSimpleFunc {
+
+        @Param ${type.name}Holder left;
+        @Param ${type.name}Holder right;
+        @Output BitHolder out;
+        public void setup() {}
+
+        public void eval() {
+            int cmp;
+            <@varCompareBlock leftType="leftType" rightType="rightType" absCompare="false" output="cmp" nullCompare=false nullComparesHigh=false />
+            out.value = cmp > -1 ? 1 : 0;
+        }
+    }
+
+
+    <#-- Comparison function for comparison expression operator (=, &lt;, etc.),
+         not for sorting and grouping relational operators.) -->
+    @FunctionTemplate(name = "Equal",
+        scope = FunctionTemplate.FunctionScope.SIMPLE,
+        returnType = FunctionTemplate.ReturnType.DECIMAL_MAX_SCALE,
+        nulls = NullHandling.NULL_IF_NULL)
+    public static class ${type.name}Equal implements DrillSimpleFunc {
+
+        @Param ${type.name}Holder left;
+        @Param ${type.name}Holder right;
+        @Output BitHolder out;
+        public void setup() {}
+
+        public void eval() {
+            int cmp;
+            <@varCompareBlock leftType="leftType" rightType="rightType" absCompare="false" output="cmp" nullCompare=false nullComparesHigh=false />
+            out.value = cmp == 0 ? 1 : 0;
+        }
+    }
+
+
+    <#-- Comparison function for comparison expression operator (=, &lt;, etc.),
+         not for sorting and grouping relational operators.) -->
+    @FunctionTemplate(name = "not equal",
+        scope = FunctionTemplate.FunctionScope.SIMPLE,
+        returnType = FunctionTemplate.ReturnType.DECIMAL_MAX_SCALE,
+        nulls = NullHandling.NULL_IF_NULL)
+    public static class ${type.name}NotEqual implements DrillSimpleFunc {
+
+        @Param ${type.name}Holder left;
+        @Param ${type.name}Holder right;
+        @Output BitHolder out;
+        public void setup() {}
+
+        public void eval() {
+            int cmp;
+            <@varCompareBlock leftType="leftType" rightType="rightType" absCompare="false" output="cmp" nullCompare=false nullComparesHigh=false />
+            out.value = cmp != 0 ? 1 : 0;
+        }
+    }
+}
+
+<#elseif type.name.endsWith("Sparse")>
 
 <@pp.changeOutputFile name="/org/apache/drill/exec/expr/fn/impl/${type.name}Functions.java" />
 
diff --git a/exec/java-exec/src/main/codegen/templates/JsonOutputRecordWriter.java b/exec/java-exec/src/main/codegen/templates/JsonOutputRecordWriter.java
index 2bd3d0c50..a3bbc9b9a 100644
--- a/exec/java-exec/src/main/codegen/templates/JsonOutputRecordWriter.java
+++ b/exec/java-exec/src/main/codegen/templates/JsonOutputRecordWriter.java
@@ -93,6 +93,7 @@ public abstract class JSONOutputRecordWriter extends AbstractRecordWriter implem
   <#case "Decimal28Dense">
   <#case "Decimal38Dense">
   <#case "Decimal38Sparse">
+  <#case "VarDecimal">
     <#assign typeName = "Decimal">
     <#break>
   <#case "Float4">
diff --git a/exec/java-exec/src/main/codegen/templates/NumericToCharFunctions.java b/exec/java-exec/src/main/codegen/templates/NumericToCharFunctions.java
index 638095015..3fc45dffb 100644
--- a/exec/java-exec/src/main/codegen/templates/NumericToCharFunctions.java
+++ b/exec/java-exec/src/main/codegen/templates/NumericToCharFunctions.java
@@ -67,7 +67,10 @@ public class G${type}ToChar implements DrillSimpleFunc {
 
     public void eval() {
 
-        <#if type == "Decimal9" || type == "Decimal18">
+        <#if type == "VarDecimal">
+        java.math.BigDecimal bigDecimal = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(left.buffer, left.start, left.end - left.start, left.scale);
+        String str = outputFormat.format(bigDecimal);
+        <#elseif type == "Decimal9" || type == "Decimal18">
         java.math.BigDecimal bigDecimal = new java.math.BigDecimal(java.math.BigInteger.valueOf(left.value), left.scale);
         String str = outputFormat.format(bigDecimal);
         <#elseif type == "Decimal28Sparse" || type == "Decimal38Sparse">
diff --git a/exec/java-exec/src/main/codegen/templates/ParquetTypeHelper.java b/exec/java-exec/src/main/codegen/templates/ParquetTypeHelper.java
index fc7100b8d..c4166f9df 100644
--- a/exec/java-exec/src/main/codegen/templates/ParquetTypeHelper.java
+++ b/exec/java-exec/src/main/codegen/templates/ParquetTypeHelper.java
@@ -141,6 +141,7 @@ public class ParquetTypeHelper {
       case DECIMAL28DENSE:
       case DECIMAL38SPARSE:
       case DECIMAL38DENSE:
+      case VARDECIMAL:
         return new DecimalMetadata(field.getPrecision(), field.getScale());
       default:
         return null;
diff --git a/exec/java-exec/src/main/codegen/templates/SqlAccessors.java b/exec/java-exec/src/main/codegen/templates/SqlAccessors.java
index 29f1ac5d0..fe650ac9c 100644
--- a/exec/java-exec/src/main/codegen/templates/SqlAccessors.java
+++ b/exec/java-exec/src/main/codegen/templates/SqlAccessors.java
@@ -126,6 +126,25 @@ public class ${name}Accessor extends AbstractSqlAccessor {
     }
       <#break>
 
+    <#case "VarDecimal">
+
+    @Override
+    public String getString(int index){
+<#if mode=="Nullable">
+        if(ac.isNull(index)){
+          return null;
+        }
+</#if>
+        try {
+          BigDecimal bd=getBigDecimal(index);
+          return bd.toString();
+        } catch (Throwable t) {
+          // TODO: we cannot throw the exception, so return its string representation, for now.  This should be handled better.
+          return t.toString();
+        }
+    }
+      <#break>
+
     <#case "VarChar">
 
     @Override
@@ -228,7 +247,7 @@ public class ${name}Accessor extends AbstractSqlAccessor {
     return String.valueOf(ac.getAsStringBuilder(index));
   }
 
-  <#elseif minor.class.startsWith("Decimal")>
+  <#elseif minor.class.contains("Decimal")>
 
   @Override
   public BigDecimal getBigDecimal(int index) {
@@ -345,4 +364,4 @@ public class ${name}Accessor extends AbstractSqlAccessor {
 }
 </#list>
 </#list>
-</#list>
\ No newline at end of file
+</#list>
diff --git a/exec/java-exec/src/main/codegen/templates/StringOutputRecordWriter.java b/exec/java-exec/src/main/codegen/templates/StringOutputRecordWriter.java
index cee157bba..5fe90ceed 100644
--- a/exec/java-exec/src/main/codegen/templates/StringOutputRecordWriter.java
+++ b/exec/java-exec/src/main/codegen/templates/StringOutputRecordWriter.java
@@ -140,13 +140,13 @@ public abstract class StringOutputRecordWriter extends AbstractRecordWriter {
         minor.class == "Decimal28Dense" ||
         minor.class == "Decimal38Dense" ||
         minor.class == "Decimal28Sparse" ||
-        minor.class == "Decimal38Sparse">
-
+        minor.class == "Decimal38Sparse" ||
+        minor.class == "VarChar" ||
+        minor.class == "Var16Char" ||
+        minor.class == "VarBinary" ||
+        minor.class == "VarDecimal">
     // TODO: error check
     addField(fieldId, reader.readObject().toString());
-
-  <#elseif minor.class == "VarChar" || minor.class == "Var16Char" || minor.class == "VarBinary">
-    addField(fieldId, reader.readObject().toString());
   <#else>
     throw new UnsupportedOperationException(String.format("Unsupported field type: %s"),
       holder.getCanonicalClass());
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/ExpressionTreeMaterializer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/ExpressionTreeMaterializer.java
index f1b50c95d..6e8e31592 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/ExpressionTreeMaterializer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/ExpressionTreeMaterializer.java
@@ -917,6 +917,13 @@ public class ExpressionTreeMaterializer {
         }
         return false;
 
+      case VARDECIMAL:
+          // for VARDECIMAL, precision does not matter (it's variable precision, to fit the number), but scale matters
+          if (to.getScale() == from.getScale()) {
+              return true;
+          }
+          return false;
+
       case FIXED16CHAR:
       case FIXEDBINARY:
       case FIXEDCHAR:
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/GetSetVectorHelper.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/GetSetVectorHelper.java
index fc9398385..8ceb3d95c 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/GetSetVectorHelper.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/GetSetVectorHelper.java
@@ -81,6 +81,14 @@ public class GetSetVectorHelper {
         eval.assign(out.getHolder().ref("start"), JExpr.lit(TypeHelper.getSize(type)).mul(indexVariable));
         eval.assign(out.getHolder().ref("buffer"), vector.invoke("getBuffer"));
         return;
+      case VARDECIMAL: {
+        eval.assign(out.getHolder().ref("buffer"), vector.invoke("getBuffer"));
+        JVar se = eval.decl(model.LONG, "startEnd", getValueAccessor.invoke("getStartEnd").arg(indexVariable));
+        eval.assign(out.getHolder().ref("start"), JExpr.cast(model._ref(int.class), se));
+        eval.assign(out.getHolder().ref("end"), JExpr.cast(model._ref(int.class), se.shr(JExpr.lit(32))));
+        eval.assign(out.getHolder().ref("scale"), vector.invoke("getField").invoke("getScale"));
+        return;
+      }
       case INTERVAL:{
         JVar start = eval.decl(model.INT, "start", JExpr.lit(TypeHelper.getSize(type)).mul(indexVariable));
         JVar data = eval.decl(model.ref(DrillBuf.class), "data", vector.invoke("getBuffer"));
@@ -88,22 +96,22 @@ public class GetSetVectorHelper {
         eval.assign(out.getHolder().ref("days"), data.invoke("getInt").arg(start.plus(JExpr.lit(4))));
         eval.assign(out.getHolder().ref("milliseconds"), data.invoke("getInt").arg(start.plus(JExpr.lit(8))));
         return;
-      }
+        }
       case INTERVALDAY: {
         JVar start = eval.decl(model.INT, "start", JExpr.lit(TypeHelper.getSize(type)).mul(indexVariable));
         eval.assign(out.getHolder().ref("days"), vector.invoke("getBuffer").invoke("getInt").arg(start));
         eval.assign(out.getHolder().ref("milliseconds"), vector.invoke("getBuffer").invoke("getInt").arg(start.plus(JExpr.lit(4))));
         return;
-      }
+        }
       case VAR16CHAR:
       case VARBINARY:
-      case VARCHAR:
+      case VARCHAR: {
          eval.assign(out.getHolder().ref("buffer"), vector.invoke("getBuffer"));
          JVar se = eval.decl(model.LONG, "startEnd", getValueAccessor.invoke("getStartEnd").arg(indexVariable));
          eval.assign(out.getHolder().ref("start"), JExpr.cast(model._ref(int.class), se));
          eval.assign(out.getHolder().ref("end"), JExpr.cast(model._ref(int.class), se.shr(JExpr.lit(32))));
         return;
-
+        }
       }
     }
 
@@ -164,6 +172,7 @@ public class GetSetVectorHelper {
       case VAR16CHAR:
       case VARBINARY:
       case VARCHAR:
+      case VARDECIMAL:
         return setMethod //
             .arg(in.f("start")) //
             .arg(in.f("end")) //
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Hash32AsDouble.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Hash32AsDouble.java
index e0db999be..e134c9d84 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Hash32AsDouble.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Hash32AsDouble.java
@@ -25,6 +25,8 @@ import org.apache.drill.exec.expr.annotations.FunctionTemplate.FunctionScope;
 import org.apache.drill.exec.expr.holders.BigIntHolder;
 import org.apache.drill.exec.expr.holders.Decimal18Holder;
 import org.apache.drill.exec.expr.holders.Decimal28SparseHolder;
+import org.apache.drill.exec.expr.holders.VarDecimalHolder;
+import org.apache.drill.exec.expr.holders.NullableVarDecimalHolder;
 import org.apache.drill.exec.expr.holders.Decimal38SparseHolder;
 import org.apache.drill.exec.expr.holders.Decimal9Holder;
 import org.apache.drill.exec.expr.holders.Float4Holder;
@@ -260,6 +262,44 @@ public class Hash32AsDouble {
   }
 
   @FunctionTemplate(name = "hash32AsDouble", scope = FunctionScope.SIMPLE, nulls = FunctionTemplate.NullHandling.INTERNAL)
+  public static class VarDecimalHash implements DrillSimpleFunc {
+    @Param
+    VarDecimalHolder in;
+    @Output
+    IntHolder out;
+
+    public void setup() {
+    }
+
+    public void eval() {
+      java.math.BigDecimal input = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(in.buffer,
+              in.start, in.end - in.start, in.scale);
+      out.value = org.apache.drill.exec.expr.fn.impl.HashHelper.hash32(input.doubleValue(), 0);
+    }
+  }
+
+  @FunctionTemplate(name = "hash32AsDouble", scope = FunctionScope.SIMPLE, nulls = FunctionTemplate.NullHandling.INTERNAL)
+  public static class NullableVarDecimalHash implements DrillSimpleFunc {
+    @Param
+    NullableVarDecimalHolder in;
+    @Output
+    IntHolder out;
+
+    public void setup() {
+    }
+
+    public void eval() {
+      if (in.isSet == 0) {
+        out.value = 0;
+      } else {
+        java.math.BigDecimal input = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(in.buffer,
+                in.start, in.end - in.start, in.scale);
+        out.value = org.apache.drill.exec.expr.fn.impl.HashHelper.hash32(input.doubleValue(), 0);
+      }
+    }
+  }
+
+  @FunctionTemplate(name = "hash32AsDouble", scope = FunctionScope.SIMPLE, nulls = FunctionTemplate.NullHandling.INTERNAL)
   public static class Decimal28Hash implements DrillSimpleFunc {
     @Param
     Decimal28SparseHolder in;
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Hash32Functions.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Hash32Functions.java
index c04dd958d..a6d24f5a7 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Hash32Functions.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Hash32Functions.java
@@ -27,6 +27,8 @@ import org.apache.drill.exec.expr.holders.BitHolder;
 import org.apache.drill.exec.expr.holders.DateHolder;
 import org.apache.drill.exec.expr.holders.Decimal18Holder;
 import org.apache.drill.exec.expr.holders.Decimal28SparseHolder;
+import org.apache.drill.exec.expr.holders.VarDecimalHolder;
+import org.apache.drill.exec.expr.holders.NullableVarDecimalHolder;
 import org.apache.drill.exec.expr.holders.Decimal38SparseHolder;
 import org.apache.drill.exec.expr.holders.Decimal9Holder;
 import org.apache.drill.exec.expr.holders.Float4Holder;
@@ -53,6 +55,8 @@ import org.apache.drill.exec.expr.holders.Var16CharHolder;
 import org.apache.drill.exec.expr.holders.VarBinaryHolder;
 import org.apache.drill.exec.expr.holders.VarCharHolder;
 
+import java.math.BigDecimal;
+
 public class Hash32Functions {
 
   @FunctionTemplate(names = {"hash", "hash32"}, scope = FunctionScope.SIMPLE, nulls = FunctionTemplate.NullHandling.INTERNAL )
@@ -429,6 +433,58 @@ public class Hash32Functions {
   }
 
   @FunctionTemplate(names = {"hash", "hash32"}, scope = FunctionScope.SIMPLE, nulls = FunctionTemplate.NullHandling.INTERNAL)
+  public static class VarDecimalHash implements DrillSimpleFunc {
+    @Param  VarDecimalHolder in;
+    @Output IntHolder out;
+
+    public void setup() {
+    }
+
+    public void eval() {
+      java.math.BigDecimal bd = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(in.buffer,
+              in.start, in.end - in.start, in.scale);
+      out.value = hashBigDecimal(bd, 0);
+    }
+  }
+
+  public static int hashBigDecimal(BigDecimal bd, int seed) {
+    java.math.BigInteger bi = bd.unscaledValue();
+    byte[] barr = bi.toByteArray();
+    int shiftCount = 0;
+    int xor = 0;
+    for (int i = 0; i < barr.length; ++i) {
+      byte b = barr[i];
+      int ival = (int) b;
+      ival <<= shiftCount;
+      ++shiftCount;
+      if (shiftCount > 3) {
+        shiftCount = 0;
+      }
+      xor = xor ^ ival;
+    }
+    return org.apache.drill.exec.expr.fn.impl.HashHelper.hash32(xor, seed);
+  }
+
+  @FunctionTemplate(names = {"hash", "hash32"}, scope = FunctionScope.SIMPLE, nulls = FunctionTemplate.NullHandling.INTERNAL)
+  public static class NullableVarDecimalHash implements DrillSimpleFunc {
+    @Param  NullableVarDecimalHolder in;
+    @Output IntHolder out;
+
+    public void setup() {
+    }
+
+    public void eval() {
+      if (in.isSet == 0) {
+        out.value = 0;
+      } else {
+        java.math.BigDecimal bd = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(in.buffer,
+                in.start, in.end - in.start, in.scale);
+        out.value = hashBigDecimal(bd, 0);
+      }
+    }
+  }
+
+  @FunctionTemplate(names = {"hash", "hash32"}, scope = FunctionScope.SIMPLE, nulls = FunctionTemplate.NullHandling.INTERNAL)
   public static class Decimal28Hash implements DrillSimpleFunc {
     @Param  Decimal28SparseHolder in;
     @Output IntHolder out;
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Hash32FunctionsWithSeed.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Hash32FunctionsWithSeed.java
index 65fb0eec9..cda47ff7c 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Hash32FunctionsWithSeed.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Hash32FunctionsWithSeed.java
@@ -27,6 +27,8 @@ import org.apache.drill.exec.expr.holders.BitHolder;
 import org.apache.drill.exec.expr.holders.DateHolder;
 import org.apache.drill.exec.expr.holders.Decimal18Holder;
 import org.apache.drill.exec.expr.holders.Decimal28SparseHolder;
+import org.apache.drill.exec.expr.holders.VarDecimalHolder;
+import org.apache.drill.exec.expr.holders.NullableVarDecimalHolder;
 import org.apache.drill.exec.expr.holders.Decimal38SparseHolder;
 import org.apache.drill.exec.expr.holders.Decimal9Holder;
 import org.apache.drill.exec.expr.holders.Float4Holder;
@@ -480,6 +482,42 @@ public class Hash32FunctionsWithSeed {
   }
 
   @FunctionTemplate(name = "hash32", scope = FunctionScope.SIMPLE, nulls = FunctionTemplate.NullHandling.INTERNAL)
+  public static class VarDecimalHash implements DrillSimpleFunc {
+    @Param  VarDecimalHolder in;
+    @Param IntHolder seed;
+    @Output IntHolder out;
+
+    public void setup() {
+    }
+
+    public void eval() {
+      java.math.BigDecimal bd = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(in.buffer,
+              in.start, in.end - in.start, in.scale);
+      out.value = Hash32Functions.hashBigDecimal(bd, seed.value);
+    }
+  }
+
+  @FunctionTemplate(name = "hash32", scope = FunctionScope.SIMPLE, nulls = FunctionTemplate.NullHandling.INTERNAL)
+  public static class NullableVarDecimalHash implements DrillSimpleFunc {
+    @Param  NullableVarDecimalHolder in;
+    @Param IntHolder seed;
+    @Output IntHolder out;
+
+    public void setup() {
+    }
+
+    public void eval() {
+      if (in.isSet == 0) {
+        out.value = seed.value;
+      } else {
+        java.math.BigDecimal bd = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(in.buffer,
+                in.start, in.end - in.start, in.scale);
+        out.value = Hash32Functions.hashBigDecimal(bd, seed.value);
+      }
+    }
+  }
+
+  @FunctionTemplate(name = "hash32", scope = FunctionScope.SIMPLE, nulls = FunctionTemplate.NullHandling.INTERNAL)
   public static class Decimal28Hash implements DrillSimpleFunc {
     @Param  Decimal28SparseHolder in;
     @Param IntHolder seed;
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Hash32WithSeedAsDouble.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Hash32WithSeedAsDouble.java
index 038d917d9..140293517 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Hash32WithSeedAsDouble.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Hash32WithSeedAsDouble.java
@@ -23,14 +23,20 @@ import org.apache.drill.exec.expr.annotations.FunctionTemplate.FunctionScope;
 import org.apache.drill.exec.expr.annotations.Output;
 import org.apache.drill.exec.expr.annotations.Param;
 import org.apache.drill.exec.expr.holders.BigIntHolder;
+import org.apache.drill.exec.expr.holders.BitHolder;
+import org.apache.drill.exec.expr.holders.DateHolder;
 import org.apache.drill.exec.expr.holders.Decimal18Holder;
 import org.apache.drill.exec.expr.holders.Decimal28SparseHolder;
+import org.apache.drill.exec.expr.holders.VarDecimalHolder;
+import org.apache.drill.exec.expr.holders.NullableVarDecimalHolder;
 import org.apache.drill.exec.expr.holders.Decimal38SparseHolder;
 import org.apache.drill.exec.expr.holders.Decimal9Holder;
 import org.apache.drill.exec.expr.holders.Float4Holder;
 import org.apache.drill.exec.expr.holders.Float8Holder;
 import org.apache.drill.exec.expr.holders.IntHolder;
 import org.apache.drill.exec.expr.holders.NullableBigIntHolder;
+import org.apache.drill.exec.expr.holders.NullableBitHolder;
+import org.apache.drill.exec.expr.holders.NullableDateHolder;
 import org.apache.drill.exec.expr.holders.NullableDecimal18Holder;
 import org.apache.drill.exec.expr.holders.NullableDecimal28SparseHolder;
 import org.apache.drill.exec.expr.holders.NullableDecimal38SparseHolder;
@@ -38,6 +44,16 @@ import org.apache.drill.exec.expr.holders.NullableDecimal9Holder;
 import org.apache.drill.exec.expr.holders.NullableFloat4Holder;
 import org.apache.drill.exec.expr.holders.NullableFloat8Holder;
 import org.apache.drill.exec.expr.holders.NullableIntHolder;
+import org.apache.drill.exec.expr.holders.NullableTimeHolder;
+import org.apache.drill.exec.expr.holders.NullableTimeStampHolder;
+import org.apache.drill.exec.expr.holders.NullableVar16CharHolder;
+import org.apache.drill.exec.expr.holders.NullableVarBinaryHolder;
+import org.apache.drill.exec.expr.holders.NullableVarCharHolder;
+import org.apache.drill.exec.expr.holders.TimeHolder;
+import org.apache.drill.exec.expr.holders.TimeStampHolder;
+import org.apache.drill.exec.expr.holders.Var16CharHolder;
+import org.apache.drill.exec.expr.holders.VarBinaryHolder;
+import org.apache.drill.exec.expr.holders.VarCharHolder;
 
 /**
  * hash32 with seed function definitions for numeric data types. These functions cast the input numeric value to a
@@ -262,6 +278,43 @@ public class Hash32WithSeedAsDouble {
   }
 
   @FunctionTemplate(name = "hash32AsDouble", scope = FunctionScope.SIMPLE, nulls = FunctionTemplate.NullHandling.INTERNAL)
+  public static class VarDecimalHash implements DrillSimpleFunc {
+    @Param
+    VarDecimalHolder in;
+    @Param IntHolder seed;
+    @Output IntHolder out;
+
+    public void setup() {
+    }
+
+    public void eval() {
+      java.math.BigDecimal bd = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(in.buffer,
+              in.start, in.end - in.start, in.scale);
+      out.value = Hash32Functions.hashBigDecimal(bd, seed.value);
+    }
+  }
+
+  @FunctionTemplate(name = "hash32AsDouble", scope = FunctionScope.SIMPLE, nulls = FunctionTemplate.NullHandling.INTERNAL)
+  public static class NullableVarDecimalHash implements DrillSimpleFunc {
+    @Param  NullableVarDecimalHolder in;
+    @Param IntHolder seed;
+    @Output IntHolder out;
+
+    public void setup() {
+    }
+
+    public void eval() {
+      if (in.isSet == 0) {
+        out.value = seed.value;
+      } else {
+        java.math.BigDecimal bd = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(in.buffer,
+                in.start, in.end - in.start, in.scale);
+        out.value = Hash32Functions.hashBigDecimal(bd, seed.value);
+      }
+    }
+  }
+
+  @FunctionTemplate(name = "hash32AsDouble", scope = FunctionScope.SIMPLE, nulls = FunctionTemplate.NullHandling.INTERNAL)
   public static class Decimal28Hash implements DrillSimpleFunc {
     @Param  Decimal28SparseHolder in;
     @Param IntHolder seed;
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Hash64AsDouble.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Hash64AsDouble.java
index 3329664b5..cfda6c8e4 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Hash64AsDouble.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Hash64AsDouble.java
@@ -23,14 +23,20 @@ import org.apache.drill.exec.expr.annotations.Output;
 import org.apache.drill.exec.expr.annotations.Param;
 import org.apache.drill.exec.expr.annotations.FunctionTemplate.FunctionScope;
 import org.apache.drill.exec.expr.holders.BigIntHolder;
+import org.apache.drill.exec.expr.holders.BitHolder;
+import org.apache.drill.exec.expr.holders.DateHolder;
 import org.apache.drill.exec.expr.holders.Decimal18Holder;
 import org.apache.drill.exec.expr.holders.Decimal28SparseHolder;
+import org.apache.drill.exec.expr.holders.VarDecimalHolder;
+import org.apache.drill.exec.expr.holders.NullableVarDecimalHolder;
 import org.apache.drill.exec.expr.holders.Decimal38SparseHolder;
 import org.apache.drill.exec.expr.holders.Decimal9Holder;
 import org.apache.drill.exec.expr.holders.Float4Holder;
 import org.apache.drill.exec.expr.holders.Float8Holder;
 import org.apache.drill.exec.expr.holders.IntHolder;
 import org.apache.drill.exec.expr.holders.NullableBigIntHolder;
+import org.apache.drill.exec.expr.holders.NullableBitHolder;
+import org.apache.drill.exec.expr.holders.NullableDateHolder;
 import org.apache.drill.exec.expr.holders.NullableDecimal18Holder;
 import org.apache.drill.exec.expr.holders.NullableDecimal28SparseHolder;
 import org.apache.drill.exec.expr.holders.NullableDecimal38SparseHolder;
@@ -38,6 +44,16 @@ import org.apache.drill.exec.expr.holders.NullableDecimal9Holder;
 import org.apache.drill.exec.expr.holders.NullableFloat4Holder;
 import org.apache.drill.exec.expr.holders.NullableFloat8Holder;
 import org.apache.drill.exec.expr.holders.NullableIntHolder;
+import org.apache.drill.exec.expr.holders.NullableTimeHolder;
+import org.apache.drill.exec.expr.holders.NullableTimeStampHolder;
+import org.apache.drill.exec.expr.holders.NullableVar16CharHolder;
+import org.apache.drill.exec.expr.holders.NullableVarBinaryHolder;
+import org.apache.drill.exec.expr.holders.NullableVarCharHolder;
+import org.apache.drill.exec.expr.holders.TimeHolder;
+import org.apache.drill.exec.expr.holders.TimeStampHolder;
+import org.apache.drill.exec.expr.holders.Var16CharHolder;
+import org.apache.drill.exec.expr.holders.VarBinaryHolder;
+import org.apache.drill.exec.expr.holders.VarCharHolder;
 
 /*
  * Class contains hash64 function definitions for different data types.
@@ -264,6 +280,44 @@ public class Hash64AsDouble {
   }
 
   @FunctionTemplate(name = "hash64AsDouble", scope = FunctionScope.SIMPLE, nulls = FunctionTemplate.NullHandling.INTERNAL)
+  public static class VarDecimalHash implements DrillSimpleFunc {
+    @Param
+    VarDecimalHolder in;
+    @Output
+    BigIntHolder out;
+
+    public void setup() {
+    }
+
+    public void eval() {
+      java.math.BigDecimal bd = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(in.buffer,
+              in.start, in.end - in.start, in.scale);
+      out.value = Hash32Functions.hashBigDecimal(bd, 0);
+    }
+  }
+
+  @FunctionTemplate(name = "hash64AsDouble", scope = FunctionScope.SIMPLE, nulls = FunctionTemplate.NullHandling.INTERNAL)
+  public static class NullableVarDecimalHash implements DrillSimpleFunc {
+    @Param
+    NullableVarDecimalHolder in;
+    @Output
+    BigIntHolder out;
+
+    public void setup() {
+    }
+
+    public void eval() {
+      if (in.isSet == 0) {
+        out.value = 0;
+      } else {
+        java.math.BigDecimal bd = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(in.buffer,
+                in.start, in.end - in.start, in.scale);
+        out.value = Hash32Functions.hashBigDecimal(bd, 0);
+      }
+    }
+  }
+
+  @FunctionTemplate(name = "hash64AsDouble", scope = FunctionScope.SIMPLE, nulls = FunctionTemplate.NullHandling.INTERNAL)
   public static class Decimal28Hash implements DrillSimpleFunc {
     @Param
     Decimal28SparseHolder in;
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Hash64Functions.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Hash64Functions.java
index dc508cb97..e94b133a4 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Hash64Functions.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Hash64Functions.java
@@ -27,6 +27,8 @@ import org.apache.drill.exec.expr.holders.BitHolder;
 import org.apache.drill.exec.expr.holders.DateHolder;
 import org.apache.drill.exec.expr.holders.Decimal18Holder;
 import org.apache.drill.exec.expr.holders.Decimal28SparseHolder;
+import org.apache.drill.exec.expr.holders.VarDecimalHolder;
+import org.apache.drill.exec.expr.holders.NullableVarDecimalHolder;
 import org.apache.drill.exec.expr.holders.Decimal38SparseHolder;
 import org.apache.drill.exec.expr.holders.Decimal9Holder;
 import org.apache.drill.exec.expr.holders.Float4Holder;
@@ -53,6 +55,8 @@ import org.apache.drill.exec.expr.holders.Var16CharHolder;
 import org.apache.drill.exec.expr.holders.VarBinaryHolder;
 import org.apache.drill.exec.expr.holders.VarCharHolder;
 
+import java.math.BigDecimal;
+
 /*
  * Class contains hash64 function definitions for different data types.
  */
@@ -479,6 +483,40 @@ public class Hash64Functions {
   }
 
   @FunctionTemplate(name = "hash64", scope = FunctionScope.SIMPLE, nulls = FunctionTemplate.NullHandling.INTERNAL)
+  public static class VarDecimalHash implements DrillSimpleFunc {
+    @Param
+    VarDecimalHolder in;
+    @Output
+    BigIntHolder out;
+
+    public void setup() {
+    }
+
+    public void eval() {
+      out.value = org.apache.drill.exec.expr.fn.impl.HashHelper.hash64(in.start, in.end, in.buffer, 0);
+    }
+  }
+
+  @FunctionTemplate(name = "hash64", scope = FunctionScope.SIMPLE, nulls = FunctionTemplate.NullHandling.INTERNAL)
+  public static class NullableVarDecimalHash implements DrillSimpleFunc {
+    @Param
+    NullableVarDecimalHolder in;
+    @Output
+    BigIntHolder out;
+
+    public void setup() {
+    }
+
+    public void eval() {
+      if (in.isSet == 0) {
+        out.value = 0;
+      } else {
+        out.value = org.apache.drill.exec.expr.fn.impl.HashHelper.hash64(in.start, in.end, in.buffer, 0);
+      }
+    }
+  }
+
+  @FunctionTemplate(name = "hash64", scope = FunctionScope.SIMPLE, nulls = FunctionTemplate.NullHandling.INTERNAL)
   public static class Decimal28Hash implements DrillSimpleFunc {
     @Param
     Decimal28SparseHolder in;
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Hash64FunctionsWithSeed.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Hash64FunctionsWithSeed.java
index 47f903d2b..9568e9950 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Hash64FunctionsWithSeed.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Hash64FunctionsWithSeed.java
@@ -27,6 +27,9 @@ import org.apache.drill.exec.expr.holders.BitHolder;
 import org.apache.drill.exec.expr.holders.DateHolder;
 import org.apache.drill.exec.expr.holders.Decimal18Holder;
 import org.apache.drill.exec.expr.holders.Decimal28SparseHolder;
+import org.apache.drill.exec.expr.holders.VarDecimalHolder;
+import org.apache.drill.exec.expr.holders.NullableVarDecimalHolder;
+
 import org.apache.drill.exec.expr.holders.Decimal38SparseHolder;
 import org.apache.drill.exec.expr.holders.Decimal9Holder;
 import org.apache.drill.exec.expr.holders.Float4Holder;
@@ -480,6 +483,40 @@ public class Hash64FunctionsWithSeed {
   }
 
   @FunctionTemplate(name = "hash64", scope = FunctionScope.SIMPLE, nulls = FunctionTemplate.NullHandling.INTERNAL)
+  public static class VarDecimalHash implements DrillSimpleFunc {
+    @Param  VarDecimalHolder in;
+    @Param BigIntHolder seed;
+    @Output BigIntHolder out;
+
+
+    public void setup() {
+    }
+
+    public void eval() {
+      out.value = org.apache.drill.exec.expr.fn.impl.HashHelper.hash64(in.start, in.start + Decimal28SparseHolder.WIDTH, in.buffer, seed.value);
+    }
+  }
+
+  @FunctionTemplate(name = "hash64", scope = FunctionScope.SIMPLE, nulls = FunctionTemplate.NullHandling.INTERNAL)
+  public static class NullableVarDecimalHash implements DrillSimpleFunc {
+    @Param  NullableVarDecimalHolder in;
+    @Param BigIntHolder seed;
+    @Output BigIntHolder out;
+
+
+    public void setup() {
+    }
+
+    public void eval() {
+      if (in.isSet == 0) {
+        out.value = seed.value;
+      } else {
+        out.value = org.apache.drill.exec.expr.fn.impl.HashHelper.hash64(in.start, in.start + NullableDecimal28SparseHolder.WIDTH, in.buffer, seed.value);
+      }
+    }
+  }
+
+  @FunctionTemplate(name = "hash64", scope = FunctionScope.SIMPLE, nulls = FunctionTemplate.NullHandling.INTERNAL)
   public static class Decimal28Hash implements DrillSimpleFunc {
     @Param  Decimal28SparseHolder in;
     @Param BigIntHolder seed;
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Hash64WithSeedAsDouble.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Hash64WithSeedAsDouble.java
index cce6ea018..13edf33c3 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Hash64WithSeedAsDouble.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Hash64WithSeedAsDouble.java
@@ -25,6 +25,8 @@ import org.apache.drill.exec.expr.annotations.Param;
 import org.apache.drill.exec.expr.holders.BigIntHolder;
 import org.apache.drill.exec.expr.holders.Decimal18Holder;
 import org.apache.drill.exec.expr.holders.Decimal28SparseHolder;
+import org.apache.drill.exec.expr.holders.VarDecimalHolder;
+import org.apache.drill.exec.expr.holders.NullableVarDecimalHolder;
 import org.apache.drill.exec.expr.holders.Decimal38SparseHolder;
 import org.apache.drill.exec.expr.holders.Decimal9Holder;
 import org.apache.drill.exec.expr.holders.Float4Holder;
@@ -266,6 +268,44 @@ public class Hash64WithSeedAsDouble {
   }
 
   @FunctionTemplate(name = "hash64AsDouble", scope = FunctionScope.SIMPLE, nulls = FunctionTemplate.NullHandling.INTERNAL)
+  public static class VarDecimalHash implements DrillSimpleFunc {
+    @Param  VarDecimalHolder in;
+    @Param BigIntHolder seed;
+    @Output BigIntHolder out;
+
+
+    public void setup() {
+    }
+
+    public void eval() {
+      java.math.BigDecimal input = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(in.buffer,
+              in.start, in.end - in.start, in.scale);
+      out.value = org.apache.drill.exec.expr.fn.impl.HashHelper.hash64(input.doubleValue(), seed.value);
+    }
+  }
+
+  @FunctionTemplate(name = "hash64AsDouble", scope = FunctionScope.SIMPLE, nulls = FunctionTemplate.NullHandling.INTERNAL)
+  public static class NullableVarDecimalHash implements DrillSimpleFunc {
+    @Param  NullableVarDecimalHolder in;
+    @Param BigIntHolder seed;
+    @Output BigIntHolder out;
+
+
+    public void setup() {
+    }
+
+    public void eval() {
+      if (in.isSet == 0) {
+        out.value = seed.value;
+      } else {
+        java.math.BigDecimal input = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(in.buffer,
+                in.start, in.end - in.start, in.scale);
+        out.value = org.apache.drill.exec.expr.fn.impl.HashHelper.hash64(input.doubleValue(), seed.value);
+      }
+    }
+  }
+
+  @FunctionTemplate(name = "hash64AsDouble", scope = FunctionScope.SIMPLE, nulls = FunctionTemplate.NullHandling.INTERNAL)
   public static class Decimal28Hash implements DrillSimpleFunc {
     @Param  Decimal28SparseHolder in;
     @Param BigIntHolder seed;
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/UnionFunctions.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/UnionFunctions.java
index 0aa9d4854..6b2b7aabb 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/UnionFunctions.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/UnionFunctions.java
@@ -115,6 +115,7 @@ public class UnionFunctions {
     case DECIMAL18:
     case DECIMAL28SPARSE:
     case DECIMAL38SPARSE:
+    case VARDECIMAL:
     case FLOAT4:
     case FLOAT8:
       return 0;
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillConstExecutor.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillConstExecutor.java
index 68beea250..40382a738 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillConstExecutor.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillConstExecutor.java
@@ -66,6 +66,7 @@ import org.apache.drill.exec.expr.holders.TimeHolder;
 import org.apache.drill.exec.expr.holders.TimeStampHolder;
 import org.apache.drill.exec.expr.holders.ValueHolder;
 import org.apache.drill.exec.expr.holders.VarCharHolder;
+import org.apache.drill.exec.expr.holders.VarDecimalHolder;
 import org.apache.drill.exec.ops.UdfUtilities;
 import org.apache.calcite.rel.type.RelDataTypeFactory;
 import org.apache.calcite.rex.RexBuilder;
@@ -243,6 +244,17 @@ public class DrillConstExecutor implements RexExecutor {
                 TypeInferenceUtils.createCalciteTypeWithNullability(typeFactory, SqlTypeName.DECIMAL, newCall.getType().isNullable()),
                 false);
             }
+            case VARDECIMAL: {
+              VarDecimalHolder varDecimalOut = (VarDecimalHolder)output;
+              return rexBuilder.makeLiteral(
+                      org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(
+                              varDecimalOut.buffer,
+                              varDecimalOut.start,
+                              varDecimalOut.end - varDecimalOut.start,
+                              varDecimalOut.scale),
+                      TypeInferenceUtils.createCalciteTypeWithNullability(typeFactory, SqlTypeName.DECIMAL, newCall.getType().isNullable()),
+                      false);
+            }
             case DECIMAL28SPARSE: {
               DrillBuf buffer;
               int start;
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillOptiq.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillOptiq.java
index ab073becf..d1c22388d 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillOptiq.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillOptiq.java
@@ -314,6 +314,8 @@ public class DrillOptiq {
         int precision = call.getType().getPrecision();
         int scale = call.getType().getScale();
 
+        // TODO: this may eventually need changes for VARDECIMAL
+
         if (precision <= 9) {
           castType = TypeProtos.MajorType.newBuilder().setMinorType(MinorType.DECIMAL9).setPrecision(precision).setScale(scale).build();
         } else if (precision <= 18) {
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/TypeInferenceUtils.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/TypeInferenceUtils.java
index 078094b5a..c6c93209a 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/TypeInferenceUtils.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/TypeInferenceUtils.java
@@ -69,6 +69,7 @@ public class TypeInferenceUtils {
       .put(TypeProtos.MinorType.DECIMAL18, SqlTypeName.DECIMAL)
       .put(TypeProtos.MinorType.DECIMAL28SPARSE, SqlTypeName.DECIMAL)
       .put(TypeProtos.MinorType.DECIMAL38SPARSE, SqlTypeName.DECIMAL)
+      .put(TypeProtos.MinorType.VARDECIMAL, SqlTypeName.DECIMAL)
       .put(TypeProtos.MinorType.TIME, SqlTypeName.TIME)
       .put(TypeProtos.MinorType.TIMESTAMP, SqlTypeName.TIMESTAMP)
       .put(TypeProtos.MinorType.VARBINARY, SqlTypeName.VARBINARY)
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/resolver/ResolverTypePrecedence.java b/exec/java-exec/src/main/java/org/apache/drill/exec/resolver/ResolverTypePrecedence.java
index 5ec36c77c..f3df7b647 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/resolver/ResolverTypePrecedence.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/resolver/ResolverTypePrecedence.java
@@ -67,6 +67,7 @@ public class ResolverTypePrecedence {
     precedenceMap.put(MinorType.DECIMAL28SPARSE, i += 2);
     precedenceMap.put(MinorType.DECIMAL38DENSE, i += 2);
     precedenceMap.put(MinorType.DECIMAL38SPARSE, i += 2);
+    precedenceMap.put(MinorType.VARDECIMAL, i += 2);
     precedenceMap.put(MinorType.FLOAT8, i += 2);
     precedenceMap.put(MinorType.DATE, i += 2);
     precedenceMap.put(MinorType.TIMESTAMP, i += 2);
@@ -112,6 +113,7 @@ public class ResolverTypePrecedence {
     rule.add(MinorType.DECIMAL28DENSE);
     rule.add(MinorType.DECIMAL38SPARSE);
     rule.add(MinorType.DECIMAL38DENSE);
+    rule.add(MinorType.VARDECIMAL);
     rule.add(MinorType.MONEY);
     rule.add(MinorType.FLOAT4);
     rule.add(MinorType.FLOAT8);
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/resolver/TypeCastRules.java b/exec/java-exec/src/main/java/org/apache/drill/exec/resolver/TypeCastRules.java
index 0105ba015..d90c16c32 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/resolver/TypeCastRules.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/resolver/TypeCastRules.java
@@ -67,6 +67,7 @@ public class TypeCastRules {
     rule.add(MinorType.DECIMAL28DENSE);
     rule.add(MinorType.DECIMAL38SPARSE);
     rule.add(MinorType.DECIMAL38DENSE);
+    rule.add(MinorType.VARDECIMAL);
     rule.add(MinorType.MONEY);
     rule.add(MinorType.FLOAT4);
     rule.add(MinorType.FLOAT8);
@@ -95,6 +96,7 @@ public class TypeCastRules {
     rule.add(MinorType.DECIMAL28DENSE);
     rule.add(MinorType.DECIMAL38SPARSE);
     rule.add(MinorType.DECIMAL38DENSE);
+    rule.add(MinorType.VARDECIMAL);
     rule.add(MinorType.MONEY);
     rule.add(MinorType.FLOAT4);
     rule.add(MinorType.FLOAT8);
@@ -123,6 +125,7 @@ public class TypeCastRules {
     rule.add(MinorType.DECIMAL28DENSE);
     rule.add(MinorType.DECIMAL38SPARSE);
     rule.add(MinorType.DECIMAL38DENSE);
+    rule.add(MinorType.VARDECIMAL);
     rule.add(MinorType.MONEY);
     rule.add(MinorType.FLOAT4);
     rule.add(MinorType.FLOAT8);
@@ -151,6 +154,7 @@ public class TypeCastRules {
     rule.add(MinorType.DECIMAL28DENSE);
     rule.add(MinorType.DECIMAL38SPARSE);
     rule.add(MinorType.DECIMAL38DENSE);
+    rule.add(MinorType.VARDECIMAL);
     rule.add(MinorType.MONEY);
     rule.add(MinorType.FLOAT4);
     rule.add(MinorType.FLOAT8);
@@ -207,6 +211,7 @@ public class TypeCastRules {
     rule.add(MinorType.DECIMAL28DENSE);
     rule.add(MinorType.DECIMAL38SPARSE);
     rule.add(MinorType.DECIMAL38DENSE);
+    rule.add(MinorType.VARDECIMAL);
     rule.add(MinorType.MONEY);
     rule.add(MinorType.FLOAT4);
     rule.add(MinorType.FLOAT8);
@@ -235,6 +240,7 @@ public class TypeCastRules {
     rule.add(MinorType.DECIMAL28DENSE);
     rule.add(MinorType.DECIMAL38SPARSE);
     rule.add(MinorType.DECIMAL38DENSE);
+    rule.add(MinorType.VARDECIMAL);
     rule.add(MinorType.MONEY);
     rule.add(MinorType.FLOAT4);
     rule.add(MinorType.FLOAT8);
@@ -263,6 +269,7 @@ public class TypeCastRules {
     rule.add(MinorType.DECIMAL28DENSE);
     rule.add(MinorType.DECIMAL38SPARSE);
     rule.add(MinorType.DECIMAL38DENSE);
+    rule.add(MinorType.VARDECIMAL);   // TODO: implement conversion between VARDECIMAL and other types??
     rule.add(MinorType.MONEY);
     rule.add(MinorType.FLOAT4);
     rule.add(MinorType.FLOAT8);
@@ -291,6 +298,7 @@ public class TypeCastRules {
     rule.add(MinorType.DECIMAL28DENSE);
     rule.add(MinorType.DECIMAL38SPARSE);
     rule.add(MinorType.DECIMAL38DENSE);
+    rule.add(MinorType.VARDECIMAL);   // TODO: implement conversion between VARDECIMAL and other types??
     rule.add(MinorType.MONEY);
     rule.add(MinorType.FLOAT4);
     rule.add(MinorType.FLOAT8);
@@ -319,6 +327,7 @@ public class TypeCastRules {
     rule.add(MinorType.DECIMAL28DENSE);
     rule.add(MinorType.DECIMAL38SPARSE);
     rule.add(MinorType.DECIMAL38DENSE);
+    rule.add(MinorType.VARDECIMAL);   // TODO: implement conversion between VARDECIMAL and other types??
     rule.add(MinorType.MONEY);
     rule.add(MinorType.FLOAT4);
     rule.add(MinorType.FLOAT8);
@@ -331,6 +340,35 @@ public class TypeCastRules {
     rule.add(MinorType.VARBINARY);
     rules.put(MinorType.DECIMAL28SPARSE, rule);
 
+    /** VARDECIMAL cast able from **/
+    rule = new HashSet<MinorType>();
+    rule.add(MinorType.TINYINT);
+    rule.add(MinorType.SMALLINT);
+    rule.add(MinorType.INT);
+    rule.add(MinorType.BIGINT);
+    rule.add(MinorType.UINT1);
+    rule.add(MinorType.UINT2);
+    rule.add(MinorType.UINT4);
+    rule.add(MinorType.UINT8);
+    rule.add(MinorType.DECIMAL9);   // TODO: implement conversion between VARDECIMAL and other types??
+    rule.add(MinorType.DECIMAL18);   // TODO: implement conversion between VARDECIMAL and other types??
+    rule.add(MinorType.DECIMAL28SPARSE);   // TODO: implement conversion between VARDECIMAL and other types??
+    rule.add(MinorType.DECIMAL28DENSE);   // TODO: implement conversion between VARDECIMAL and other types??
+    rule.add(MinorType.DECIMAL38SPARSE);   // TODO: implement conversion between VARDECIMAL and other types??
+    rule.add(MinorType.DECIMAL38DENSE);   // TODO: implement conversion between VARDECIMAL and other types??
+    rule.add(MinorType.VARDECIMAL);
+    rule.add(MinorType.MONEY);
+    rule.add(MinorType.FLOAT4);
+    rule.add(MinorType.FLOAT8);
+    rule.add(MinorType.BIT);
+    rule.add(MinorType.FIXEDCHAR);
+    rule.add(MinorType.FIXED16CHAR);
+    rule.add(MinorType.FIXEDBINARY);
+    rule.add(MinorType.VARCHAR);
+    rule.add(MinorType.VAR16CHAR);
+    rule.add(MinorType.VARBINARY);
+    rules.put(MinorType.VARDECIMAL, rule);
+
     /** DECIMAL38Dense cast able from **/
     rule = new HashSet<>();
     rule.add(MinorType.TINYINT);
@@ -347,6 +385,7 @@ public class TypeCastRules {
     rule.add(MinorType.DECIMAL28DENSE);
     rule.add(MinorType.DECIMAL38SPARSE);
     rule.add(MinorType.DECIMAL38DENSE);
+    rule.add(MinorType.VARDECIMAL);   // TODO: implement conversion between VARDECIMAL and other types??
     rule.add(MinorType.MONEY);
     rule.add(MinorType.FLOAT4);
     rule.add(MinorType.FLOAT8);
@@ -376,6 +415,7 @@ public class TypeCastRules {
     rule.add(MinorType.DECIMAL28DENSE);
     rule.add(MinorType.DECIMAL38SPARSE);
     rule.add(MinorType.DECIMAL38DENSE);
+    rule.add(MinorType.VARDECIMAL);   // TODO: implement conversion between VARDECIMAL and other types??
     rule.add(MinorType.MONEY);
     rule.add(MinorType.FLOAT4);
     rule.add(MinorType.FLOAT8);
@@ -404,6 +444,7 @@ public class TypeCastRules {
     rule.add(MinorType.DECIMAL28DENSE);
     rule.add(MinorType.DECIMAL38SPARSE);
     rule.add(MinorType.DECIMAL38DENSE);
+    rule.add(MinorType.VARDECIMAL);
     rule.add(MinorType.MONEY);
     rule.add(MinorType.FLOAT4);
     rule.add(MinorType.FLOAT8);
@@ -539,6 +580,7 @@ public class TypeCastRules {
     rule.add(MinorType.DECIMAL28DENSE);
     rule.add(MinorType.DECIMAL38SPARSE);
     rule.add(MinorType.DECIMAL38DENSE);
+    rule.add(MinorType.VARDECIMAL);
     rule.add(MinorType.MONEY);
     rule.add(MinorType.FLOAT4);
     rule.add(MinorType.BIT);
@@ -564,6 +606,7 @@ public class TypeCastRules {
     rule.add(MinorType.DECIMAL28DENSE);
     rule.add(MinorType.DECIMAL38SPARSE);
     rule.add(MinorType.DECIMAL38DENSE);
+    rule.add(MinorType.VARDECIMAL);
     rule.add(MinorType.MONEY);
     rule.add(MinorType.FLOAT4);
     rule.add(MinorType.FLOAT8);
@@ -603,6 +646,7 @@ public class TypeCastRules {
     rule.add(MinorType.DECIMAL28DENSE);
     rule.add(MinorType.DECIMAL38SPARSE);
     rule.add(MinorType.DECIMAL38DENSE);
+    rule.add(MinorType.VARDECIMAL);
     rule.add(MinorType.TIMESTAMP);
     rule.add(MinorType.FLOAT4);
     rule.add(MinorType.FLOAT8);
@@ -638,6 +682,7 @@ public class TypeCastRules {
     rule.add(MinorType.DECIMAL28DENSE);
     rule.add(MinorType.DECIMAL38SPARSE);
     rule.add(MinorType.DECIMAL38DENSE);
+    rule.add(MinorType.VARDECIMAL);
     rule.add(MinorType.MONEY);
     rule.add(MinorType.TIMESTAMPTZ);
     rule.add(MinorType.FLOAT4);
@@ -674,6 +719,7 @@ public class TypeCastRules {
     rule.add(MinorType.DECIMAL28DENSE);
     rule.add(MinorType.DECIMAL38SPARSE);
     rule.add(MinorType.DECIMAL38DENSE);
+    rule.add(MinorType.VARDECIMAL);
     rule.add(MinorType.MONEY);
     rule.add(MinorType.TIMESTAMPTZ);
     rule.add(MinorType.FLOAT4);
@@ -701,6 +747,7 @@ public class TypeCastRules {
     rule.add(MinorType.DECIMAL28DENSE);
     rule.add(MinorType.DECIMAL38SPARSE);
     rule.add(MinorType.DECIMAL38DENSE);
+    rule.add(MinorType.VARDECIMAL);
     rule.add(MinorType.MONEY);
     rule.add(MinorType.TIMESTAMPTZ);
     rule.add(MinorType.FLOAT4);
@@ -737,6 +784,7 @@ public class TypeCastRules {
     rule.add(MinorType.DECIMAL28DENSE);
     rule.add(MinorType.DECIMAL38SPARSE);
     rule.add(MinorType.DECIMAL38DENSE);
+    rule.add(MinorType.VARDECIMAL);
     rule.add(MinorType.MONEY);
     rule.add(MinorType.FLOAT4);
     rule.add(MinorType.FLOAT8);
@@ -772,6 +820,7 @@ public class TypeCastRules {
     rule.add(MinorType.DECIMAL28DENSE);
     rule.add(MinorType.DECIMAL38SPARSE);
     rule.add(MinorType.DECIMAL38DENSE);
+    rule.add(MinorType.VARDECIMAL);
     rule.add(MinorType.MONEY);
     rule.add(MinorType.TIMESTAMP);
     rule.add(MinorType.TIMESTAMPTZ);
@@ -1012,6 +1061,7 @@ public class TypeCastRules {
       case DECIMAL18:
       case DECIMAL28SPARSE:
       case DECIMAL38SPARSE:
+      case VARDECIMAL:
       case FLOAT4:
       case FLOAT8:
         return true;
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/columnreaders/ColumnReaderFactory.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/columnreaders/ColumnReaderFactory.java
index 7edfa00e1..5fbd20f41 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/columnreaders/ColumnReaderFactory.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/columnreaders/ColumnReaderFactory.java
@@ -20,6 +20,8 @@ package org.apache.drill.exec.store.parquet.columnreaders;
 import org.apache.drill.common.exceptions.ExecutionSetupException;
 import org.apache.drill.exec.ExecConstants;
 import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.vector.VarDecimalVector;
+import org.apache.drill.exec.vector.NullableVarDecimalVector;
 import org.apache.drill.exec.vector.BigIntVector;
 import org.apache.drill.exec.vector.BitVector;
 import org.apache.drill.exec.vector.DateVector;
@@ -226,7 +228,9 @@ public class ColumnReaderFactory {
           case ENUM:
             return new VarLengthColumnReaders.VarCharColumn(parentReader, allocateSize, descriptor, columnChunkMetaData, fixedLength, (VarCharVector) v, schemaElement);
           case DECIMAL:
-            if (v instanceof Decimal28SparseVector) {
+            if (v instanceof VarDecimalVector) {
+              return new VarLengthColumnReaders.VarDecimalColumn(parentReader, allocateSize, descriptor, columnChunkMetaData, fixedLength, (VarDecimalVector) v, schemaElement);
+            } else if (v instanceof Decimal28SparseVector) {
               return new VarLengthColumnReaders.Decimal28Column(parentReader, allocateSize, descriptor, columnChunkMetaData, fixedLength, (Decimal28SparseVector) v, schemaElement);
             } else if (v instanceof Decimal38SparseVector) {
               return new VarLengthColumnReaders.Decimal38Column(parentReader, allocateSize, descriptor, columnChunkMetaData, fixedLength, (Decimal38SparseVector) v, schemaElement);
@@ -244,7 +248,9 @@ public class ColumnReaderFactory {
           case ENUM:
             return new VarLengthColumnReaders.NullableVarCharColumn(parentReader, allocateSize, descriptor, columnChunkMetaData, fixedLength, (NullableVarCharVector) v, schemaElement);
           case DECIMAL:
-            if (v instanceof NullableDecimal28SparseVector) {
+            if (v instanceof NullableVarDecimalVector) {
+              return new VarLengthColumnReaders.NullableVarDecimalColumn(parentReader, allocateSize, descriptor, columnChunkMetaData, fixedLength, (NullableVarDecimalVector) v, schemaElement);
+            } else if (v instanceof NullableDecimal28SparseVector) {
               return new VarLengthColumnReaders.NullableDecimal28Column(parentReader, allocateSize, descriptor, columnChunkMetaData, fixedLength, (NullableDecimal28SparseVector) v, schemaElement);
             } else if (v instanceof NullableDecimal38SparseVector) {
               return new VarLengthColumnReaders.NullableDecimal38Column(parentReader, allocateSize, descriptor, columnChunkMetaData, fixedLength, (NullableDecimal38SparseVector) v, schemaElement);
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/columnreaders/ParquetToDrillTypeConverter.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/columnreaders/ParquetToDrillTypeConverter.java
index fc1c4af05..d6cf5dbf2 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/columnreaders/ParquetToDrillTypeConverter.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/columnreaders/ParquetToDrillTypeConverter.java
@@ -29,18 +29,20 @@ import org.apache.parquet.format.SchemaElement;
 import org.apache.parquet.schema.PrimitiveType;
 
 import static com.google.common.base.Preconditions.checkArgument;
+import static org.apache.parquet.format.Type.BYTE_ARRAY;
 
 public class ParquetToDrillTypeConverter {
 
-
-  private static TypeProtos.MinorType getDecimalType(int precision) {
-    return precision <= 28 ? TypeProtos.MinorType.DECIMAL28SPARSE : MinorType.DECIMAL38SPARSE;
+  private static TypeProtos.MinorType getDecimalType(SchemaElement schemaElement) {
+    if (schemaElement.getType() == BYTE_ARRAY) {
+      return TypeProtos.MinorType.VARDECIMAL;   // variable width decimal uses new VARDECIMAL type
+    }
+    return schemaElement.getPrecision() <= 28 ? TypeProtos.MinorType.DECIMAL28SPARSE : MinorType.DECIMAL38SPARSE;
   }
 
   private static TypeProtos.MinorType getMinorType(PrimitiveType.PrimitiveTypeName primitiveTypeName, int length,
-                                                   ConvertedType convertedType, int precision, OptionManager options) {
-
-
+                                                   ConvertedType convertedType, SchemaElement schemaElement,
+                                                   OptionManager options) {
     switch (primitiveTypeName) {
       case BINARY:
         if (convertedType == null) {
@@ -52,7 +54,7 @@ public class ParquetToDrillTypeConverter {
             return (TypeProtos.MinorType.VARCHAR);
           case DECIMAL:
             ParquetReaderUtility.checkDecimalTypeEnabled(options);
-            return (getDecimalType(precision));
+            return (getDecimalType(schemaElement));
           default:
             return (TypeProtos.MinorType.VARBINARY);
         }
@@ -119,7 +121,7 @@ public class ParquetToDrillTypeConverter {
           return TypeProtos.MinorType.VARBINARY;
         } else if (convertedType == ConvertedType.DECIMAL) {
           ParquetReaderUtility.checkDecimalTypeEnabled(options);
-          return getDecimalType(precision);
+          return getDecimalType(schemaElement);
         } else if (convertedType == ConvertedType.INTERVAL) {
           return TypeProtos.MinorType.INTERVAL;
         }
@@ -131,17 +133,14 @@ public class ParquetToDrillTypeConverter {
   public static TypeProtos.MajorType toMajorType(PrimitiveType.PrimitiveTypeName primitiveTypeName, int length,
                                           TypeProtos.DataMode mode, SchemaElement schemaElement,
                                           OptionManager options) {
-    return toMajorType(primitiveTypeName, length, mode, schemaElement.getConverted_type(),
-        schemaElement.getPrecision(), schemaElement.getScale(), options);
-  }
-
-  public static TypeProtos.MajorType toMajorType(PrimitiveType.PrimitiveTypeName primitiveTypeName, int length,
-      TypeProtos.DataMode mode, ConvertedType convertedType, int precision, int scale,
-      OptionManager options) {
-    MinorType minorType = getMinorType(primitiveTypeName, length, convertedType, precision, options);
+    ConvertedType convertedType = schemaElement.getConverted_type();
+    MinorType minorType = getMinorType(primitiveTypeName, length, convertedType, schemaElement, options);
     TypeProtos.MajorType.Builder typeBuilder = TypeProtos.MajorType.newBuilder().setMinorType(minorType).setMode(mode);
 
     if (CoreDecimalUtility.isDecimalType(minorType)) {
+      int precision = schemaElement.getPrecision();
+      int scale = schemaElement.getScale();
+
       typeBuilder.setPrecision(precision).setScale(scale);
     }
     return typeBuilder.build();
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/columnreaders/VarLengthColumnReaders.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/columnreaders/VarLengthColumnReaders.java
index b3f511f00..2d8b4747a 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/columnreaders/VarLengthColumnReaders.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/columnreaders/VarLengthColumnReaders.java
@@ -26,6 +26,8 @@ import org.apache.drill.common.exceptions.ExecutionSetupException;
 import org.apache.drill.exec.expr.holders.Decimal28SparseHolder;
 import org.apache.drill.exec.expr.holders.Decimal38SparseHolder;
 import org.apache.drill.exec.util.DecimalUtility;
+import org.apache.drill.exec.vector.VarDecimalVector;
+import org.apache.drill.exec.vector.NullableVarDecimalVector;
 import org.apache.drill.exec.vector.Decimal28SparseVector;
 import org.apache.drill.exec.vector.Decimal38SparseVector;
 import org.apache.drill.exec.vector.NullableDecimal28SparseVector;
@@ -35,6 +37,7 @@ import org.apache.drill.exec.vector.NullableVarCharVector;
 import org.apache.drill.exec.vector.VarBinaryVector;
 import org.apache.drill.exec.vector.VarCharVector;
 
+
 import org.apache.parquet.column.ColumnDescriptor;
 import org.apache.parquet.format.SchemaElement;
 import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData;
@@ -42,6 +45,77 @@ import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData;
 public class VarLengthColumnReaders {
   static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(VarLengthColumnReaders.class);
 
+  /* DAO 7/28/2016 replace fixed width decimal columns for variable width parquet column, and instead use variable
+   * width decimal column (new code, VarDecimal*)
+   */
+  public static class VarDecimalColumn extends VarLengthValuesColumn<VarDecimalVector> {
+
+    protected VarDecimalVector varDecimalVector;
+    protected VarDecimalVector.Mutator mutator;
+
+    VarDecimalColumn(ParquetRecordReader parentReader, int allocateSize, ColumnDescriptor descriptor,
+                    ColumnChunkMetaData columnChunkMetaData, boolean fixedLength, VarDecimalVector v,
+                    SchemaElement schemaElement) throws ExecutionSetupException {
+      super(parentReader, allocateSize, descriptor, columnChunkMetaData, fixedLength, v, schemaElement);
+      this.varDecimalVector = v;
+      this.mutator = v.getMutator();
+    }
+
+    @Override
+    public boolean setSafe(int index, DrillBuf value, int start, int length) {
+      if (index >= varDecimalVector.getValueCapacity()) {
+        return false;
+      }
+      if (usingDictionary) {
+        currDictValToWrite = pageReader.dictionaryValueReader.readBytes();
+        ByteBuffer buf = currDictValToWrite.toByteBuffer();
+        mutator.setSafe(index, buf, buf.position(), currDictValToWrite.length());
+      } else {
+        mutator.setSafe(index, start, start + length, value);
+      }
+      return true;
+    }
+
+    @Override
+    public int capacity() {
+      return varDecimalVector.getBuffer().capacity();
+    }
+  }
+
+  public static class NullableVarDecimalColumn extends NullableVarLengthValuesColumn<NullableVarDecimalVector> {
+
+    protected NullableVarDecimalVector nullableVarDecimalVector;
+    protected NullableVarDecimalVector.Mutator mutator;
+
+    NullableVarDecimalColumn(ParquetRecordReader parentReader, int allocateSize, ColumnDescriptor descriptor,
+                            ColumnChunkMetaData columnChunkMetaData, boolean fixedLength, NullableVarDecimalVector v,
+                            SchemaElement schemaElement) throws ExecutionSetupException {
+      super(parentReader, allocateSize, descriptor, columnChunkMetaData, fixedLength, v, schemaElement);
+      nullableVarDecimalVector = v;
+      this.mutator = v.getMutator();
+    }
+
+    @Override
+    public boolean setSafe(int index, DrillBuf value, int start, int length) {
+      if (index >= nullableVarDecimalVector.getValueCapacity()) {
+        return false;
+      }
+      if (usingDictionary) {
+        ByteBuffer buf = currDictValToWrite.toByteBuffer();
+        mutator.setSafe(index, buf, buf.position(), currDictValToWrite.length());
+      } else {
+        mutator.setSafe(index, 1, start, start + length, value);
+      }
+      return true;
+    }
+
+    @Override
+    public int capacity() {
+      return nullableVarDecimalVector.getBuffer().capacity();
+    }
+  }
+  /* end DAO  VarDecimal changes */
+
   public static class Decimal28Column extends VarLengthValuesColumn<Decimal28SparseVector> {
 
     protected Decimal28SparseVector decimal28Vector;
@@ -164,6 +238,7 @@ public class VarLengthColumnReaders {
       return nullableDecimal38Vector.getBuffer().capacity();
     }
   }
+  /* DAO */
 
   public static class VarCharColumn extends VarLengthValuesColumn<VarCharVector> {
 
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet2/DrillParquetGroupConverter.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet2/DrillParquetGroupConverter.java
index 4363a7a45..f451ee28b 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet2/DrillParquetGroupConverter.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet2/DrillParquetGroupConverter.java
@@ -44,6 +44,7 @@ import org.apache.drill.exec.expr.holders.TimeHolder;
 import org.apache.drill.exec.expr.holders.TimeStampHolder;
 import org.apache.drill.exec.expr.holders.VarBinaryHolder;
 import org.apache.drill.exec.expr.holders.VarCharHolder;
+import org.apache.drill.exec.expr.holders.VarDecimalHolder;
 import org.apache.drill.exec.physical.impl.OutputMutator;
 import org.apache.drill.exec.server.options.OptionManager;
 import org.apache.drill.exec.store.parquet.ParquetReaderUtility;
@@ -66,6 +67,7 @@ import org.apache.drill.exec.vector.complex.writer.TimeStampWriter;
 import org.apache.drill.exec.vector.complex.writer.TimeWriter;
 import org.apache.drill.exec.vector.complex.writer.VarBinaryWriter;
 import org.apache.drill.exec.vector.complex.writer.VarCharWriter;
+import org.apache.drill.exec.vector.complex.writer.VarDecimalWriter;
 import org.apache.parquet.io.api.Binary;
 import org.apache.parquet.io.api.Converter;
 import org.apache.parquet.io.api.GroupConverter;
@@ -285,21 +287,22 @@ public class DrillParquetGroupConverter extends GroupConverter {
             VarCharWriter writer = type.getRepetition() == Repetition.REPEATED ? mapWriter.list(name).varChar() : mapWriter.varChar(name);
             return new DrillVarCharConverter(writer, mutator.getManagedBuffer());
           }
-          //TODO not sure if BINARY/DECIMAL is actually supported
+          // TODO not sure if BINARY/DECIMAL is actually supported.
+          // See DRILL-4184 and DRILL-4834. Support for this is added using new VarDecimal type.
           case DECIMAL: {
             ParquetReaderUtility.checkDecimalTypeEnabled(options);
             DecimalMetadata metadata = type.getDecimalMetadata();
+            VarDecimalWriter writer = type.getRepetition() == Repetition.REPEATED ? mapWriter.list(name).varDecimal() : mapWriter.varDecimal(name);
+            return new DrillBinaryToVarDecimalConverter(writer, metadata.getPrecision(), metadata.getScale(), mutator.getManagedBuffer());
+            /* DAO 7/28/2016 this has previously used the following logic for fixed-width decimal vectors
             if (metadata.getPrecision() <= 28) {
-              Decimal28SparseWriter writer = type.getRepetition() == Repetition.REPEATED
-                  ? mapWriter.list(name).decimal28Sparse()
-                  : mapWriter.decimal28Sparse(name, metadata.getScale(), metadata.getPrecision());
+              Decimal28SparseWriter writer = type.getRepetition() == Repetition.REPEATED ? mapWriter.list(name).decimal28Sparse() : mapWriter.decimal28Sparse(name, metadata.getScale(), metadata.getPrecision());
               return new DrillBinaryToDecimal28Converter(writer, metadata.getPrecision(), metadata.getScale(), mutator.getManagedBuffer());
             } else {
-              Decimal38SparseWriter writer = type.getRepetition() == Repetition.REPEATED
-                  ? mapWriter.list(name).decimal38Sparse()
-                  : mapWriter.decimal38Sparse(name, metadata.getScale(), metadata.getPrecision());
+              Decimal38SparseWriter writer = type.getRepetition() == Repetition.REPEATED ? mapWriter.list(name).decimal38Sparse() : mapWriter.decimal38Sparse(name, metadata.getScale(), metadata.getPrecision());
               return new DrillBinaryToDecimal38Converter(writer, metadata.getPrecision(), metadata.getScale(), mutator.getManagedBuffer());
             }
+            */
           }
           default: {
             throw new UnsupportedOperationException("Unsupported type " + type.getOriginalType());
@@ -581,6 +584,32 @@ public class DrillParquetGroupConverter extends GroupConverter {
     }
   }
 
+  public static class DrillBinaryToVarDecimalConverter extends PrimitiveConverter {
+    private VarDecimalWriter writer;
+    private VarDecimalHolder holder = new VarDecimalHolder();
+    private DrillBuf buf;
+
+    public DrillBinaryToVarDecimalConverter(VarDecimalWriter writer, int precision, int scale,  DrillBuf buf) {
+      this.writer = writer;
+      holder.scale = scale;
+    }
+
+    @Override
+    public void addBinary(Binary value) {
+      // NOTE: the following commented out line can be used later to produce a BigDecimal number from the binary value,
+      // with the scale information added from holder.scale.  For variable width decimal vectors, unlike for fixed width ones,
+      // we don't need to convert to a "sparse" representation (line after next, commented out) as the binary array is already
+      // "sparse" in the sense that it should only use a number of bytes needed to encode this particular numerical value.
+      //BigDecimal bigDecimal = DecimalUtility.getBigDecimalFromByteArray(value.getBytes(), 0, value.length(), holder.scale);
+      //DecimalUtility.getSparseFromBigDecimal(bigDecimal, buf, 0, holder.scale, holder.precision, Decimal28SparseHolder.nDecimalDigits);
+      holder.buffer = buf = buf.reallocIfNeeded(value.length());
+      buf.setBytes(0, value.toByteBuffer());
+      holder.start = 0;
+      holder.end = value.length();
+      writer.write(holder);
+    }
+  }
+
   public static class DrillBinaryToDecimal28Converter extends PrimitiveConverter {
     private Decimal28SparseWriter writer;
     private Decimal28SparseHolder holder = new Decimal28SparseHolder();
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/vector/complex/MapUtility.java b/exec/java-exec/src/main/java/org/apache/drill/exec/vector/complex/MapUtility.java
index 72c094a83..9a35b234a 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/vector/complex/MapUtility.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/vector/complex/MapUtility.java
@@ -130,6 +130,13 @@ public class MapUtility {
             fieldReader.copyAsValue(mapWriter.decimal38Sparse(MappifyUtility.fieldValue));
           }
           break;
+        case VARDECIMAL:
+          if (repeated) {
+            fieldReader.copyAsValue(mapWriter.list(MappifyUtility.fieldValue).varDecimal());
+          } else {
+            fieldReader.copyAsValue(mapWriter.varDecimal(MappifyUtility.fieldValue));
+          }
+          break;
         case DATE:
           if (repeated) {
             fieldReader.copyAsValue(mapWriter.list(MappifyUtility.fieldValue).date());
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/vector/complex/fn/JsonWriter.java b/exec/java-exec/src/main/java/org/apache/drill/exec/vector/complex/fn/JsonWriter.java
index 7377184a8..4d844023d 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/vector/complex/fn/JsonWriter.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/vector/complex/fn/JsonWriter.java
@@ -110,6 +110,7 @@ public class JsonWriter {
       case DECIMAL38SPARSE:
       case DECIMAL9:
       case DECIMAL18:
+      case VARDECIMAL:
         gen.writeDecimal(reader);
         break;
 
@@ -220,6 +221,7 @@ public class JsonWriter {
       case DECIMAL38SPARSE:
       case DECIMAL9:
       case DECIMAL18:
+      case VARDECIMAL:
         for(int i = 0; i < reader.size(); i++){
           gen.writeDecimal(i, reader);
         }
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/store/parquet/TestVarlenDecimal.java b/exec/java-exec/src/test/java/org/apache/drill/exec/store/parquet/TestVarlenDecimal.java
new file mode 100644
index 000000000..4eda21a7e
--- /dev/null
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/store/parquet/TestVarlenDecimal.java
@@ -0,0 +1,77 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.drill.exec.store.parquet;
+
+import org.apache.drill.test.BaseTestQuery;
+import org.apache.drill.exec.planner.physical.PlannerSettings;
+import org.apache.drill.exec.proto.UserBitShared;
+import org.junit.BeforeClass;
+import org.junit.AfterClass;
+import org.junit.Test;
+
+public class TestVarlenDecimal extends BaseTestQuery {
+  // enable decimal data type
+  @BeforeClass
+  public static void enableDecimalDataType() throws Exception {
+    test(String.format("alter session set `%s` = true", PlannerSettings.ENABLE_DECIMAL_DATA_TYPE_KEY));
+  }
+
+  // disable decimal data type
+  @AfterClass
+  public static void disableDecimalDataType() throws Exception {
+    test(String.format("alter session set `%s` = false", PlannerSettings.ENABLE_DECIMAL_DATA_TYPE_KEY));
+  }
+
+  private static final String DATAFILE = "cp.`parquet/varlenDecimal.parquet`";
+
+  @Test
+  public void testNullCount() throws Exception {
+    String query = String.format("select count(*) as c from %s where department_id is null", DATAFILE);
+    testBuilder()
+        .sqlQuery(query)
+        .unOrdered()
+        .baselineColumns("c")
+        .baselineValues(1L)
+        .build()
+        .run();
+  }
+
+  @Test
+  public void testNotNullCount() throws Exception {
+    String query = String.format("select count(*) as c from %s where department_id is not null", DATAFILE);
+    testBuilder()
+        .sqlQuery(query)
+        .unOrdered()
+        .baselineColumns("c")
+        .baselineValues(106L)
+        .build()
+        .run();
+  }
+
+  @Test
+  public void testSimpleQuery() throws Exception {
+    String query = String.format("select cast(department_id as bigint) as c from %s where cast(employee_id as decimal) = 170", DATAFILE);
+    testBuilder()
+        .sqlQuery(query)
+        .unOrdered()
+        .baselineColumns("c")
+        .baselineValues(80L)
+        .build()
+        .run();
+  }
+}
diff --git a/exec/java-exec/src/test/java/org/apache/drill/test/TestBuilder.java b/exec/java-exec/src/test/java/org/apache/drill/test/TestBuilder.java
index 798764794..696b6b2bd 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/test/TestBuilder.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/test/TestBuilder.java
@@ -433,6 +433,7 @@ public class TestBuilder {
   private String getDecimalPrecisionScaleInfo(TypeProtos.MajorType type) {
     String precision = "";
     switch(type.getMinorType()) {
+      case VARDECIMAL:
       case DECIMAL18:
       case DECIMAL28SPARSE:
       case DECIMAL38SPARSE:
diff --git a/exec/java-exec/src/test/resources/parquet/varlenDecimal.parquet b/exec/java-exec/src/test/resources/parquet/varlenDecimal.parquet
new file mode 100755
index 000000000..c531ef13e
--- /dev/null
+++ b/exec/java-exec/src/test/resources/parquet/varlenDecimal.parquet
diff --git a/exec/jdbc/src/main/java/org/apache/drill/jdbc/impl/TypeConvertingSqlAccessor.java b/exec/jdbc/src/main/java/org/apache/drill/jdbc/impl/TypeConvertingSqlAccessor.java
index 204d7d862..f8ac6bfef 100644
--- a/exec/jdbc/src/main/java/org/apache/drill/jdbc/impl/TypeConvertingSqlAccessor.java
+++ b/exec/jdbc/src/main/java/org/apache/drill/jdbc/impl/TypeConvertingSqlAccessor.java
@@ -588,6 +588,7 @@ class TypeConvertingSqlAccessor implements SqlAccessor {
       case DECIMAL18:
       case DECIMAL28SPARSE:
       case DECIMAL38SPARSE:
+      case VARDECIMAL:
         result = innerAccessor.getBigDecimal( rowOffset );
         break;
 
diff --git a/exec/vector/src/main/codegen/data/ValueVectorTypes.tdd b/exec/vector/src/main/codegen/data/ValueVectorTypes.tdd
index 4d719b4f5..ca4653d3e 100644
--- a/exec/vector/src/main/codegen/data/ValueVectorTypes.tdd
+++ b/exec/vector/src/main/codegen/data/ValueVectorTypes.tdd
@@ -178,6 +178,15 @@
       ]
     },
     {
+      major: "VarLen",
+      width: 4,
+      javaType: "int",
+      boxedType: "DrillBuf",
+      minor: [
+        { class: "VarDecimal", friendlyType: "BigDecimal", fields: [{name: "start", type: "int"}, {name: "end", type: "int"}, {name: "buffer", type: "DrillBuf"}, {name: "scale", type: "int", include: false}] }
+      ]
+    },
+    {
       major: "Bit",
       width: 1,
       javaType: "int",
diff --git a/exec/vector/src/main/codegen/templates/BaseWriter.java b/exec/vector/src/main/codegen/templates/BaseWriter.java
index c65a75809..ad9c44e1f 100644
--- a/exec/vector/src/main/codegen/templates/BaseWriter.java
+++ b/exec/vector/src/main/codegen/templates/BaseWriter.java
@@ -111,6 +111,7 @@ package org.apache.drill.exec.vector.complex.writer;
     UInt8Writer uInt8(String name);
     VarCharWriter varChar(String name);
     Var16CharWriter var16Char(String name);
+    VarDecimalWriter varDecimal(String name);
     TinyIntWriter tinyInt(String name);
     SmallIntWriter smallInt(String name);
     IntWriter integer(String name);
diff --git a/exec/vector/src/main/codegen/templates/ColumnAccessors.java b/exec/vector/src/main/codegen/templates/ColumnAccessors.java
index d0a2ace6f..6068afa22 100644
--- a/exec/vector/src/main/codegen/templates/ColumnAccessors.java
+++ b/exec/vector/src/main/codegen/templates/ColumnAccessors.java
@@ -112,9 +112,9 @@ public class ColumnAccessors {
     <#if accessorType=="BigDecimal">
       <#assign label="Decimal">
     </#if>
-    <#assign varWidth = drillType == "VarChar" || drillType == "Var16Char" || drillType == "VarBinary" />
+    <#assign varWidth = drillType == "VarChar" || drillType == "Var16Char" || drillType == "VarBinary"  || drillType == "VarDecimal"/>
     <#assign decimal = drillType == "Decimal9" || drillType == "Decimal18" ||
-                       drillType == "Decimal28Sparse" || drillType == "Decimal38Sparse" />
+                       drillType == "Decimal28Sparse" || drillType == "Decimal38Sparse"  || drillType == "VarDecimal"/>
     <#if varWidth>
       <#assign accessorType = "byte[]">
       <#assign label = "Bytes">
@@ -135,15 +135,15 @@ public class ColumnAccessors {
 
     <#if varWidth>
   public static class ${drillType}ColumnReader extends BaseVarWidthReader {
-   
+
     <#else>
   public static class ${drillType}ColumnReader extends BaseFixedWidthReader {
-    
+
     private static final int VALUE_WIDTH = ${drillType}Vector.VALUE_WIDTH;
 
       <#if decimal>
     private MajorType type;
-    
+
       </#if>
     </#if>
     <#if decimal>
@@ -190,12 +190,12 @@ public class ColumnAccessors {
     <#elseif drillType == "IntervalDay">
       final int offset = ${getOffset};
       return DateUtilities.fromIntervalDay(
-          buf.getInt(offset), 
+          buf.getInt(offset),
           buf.getInt(offset + ${minor.millisecondsOffset}));
     <#elseif drillType == "Interval">
       final int offset = ${getOffset};
       return DateUtilities.fromInterval(
-          buf.getInt(offset), 
+          buf.getInt(offset),
           buf.getInt(offset + ${minor.daysOffset}),
           buf.getInt(offset + ${minor.millisecondsOffset}));
     <#elseif drillType == "Decimal28Sparse" || drillType == "Decimal38Sparse">
@@ -241,9 +241,9 @@ public class ColumnAccessors {
   public static class ${drillType}ColumnWriter extends BaseVarWidthWriter {
       <#else>
   public static class ${drillType}ColumnWriter extends BaseFixedWidthWriter {
-    
+
     private static final int VALUE_WIDTH = ${drillType}Vector.VALUE_WIDTH;
-    
+
         <#if decimal>
     private MajorType type;
         </#if>
@@ -272,7 +272,17 @@ public class ColumnAccessors {
 
     </#if>
     @Override
+    <#if drillType = "VarDecimal">
+    public final void setDecimal(final BigDecimal bd) {
+      byte[] barr = bd.unscaledValue().toByteArray();
+      int len = barr.length;
+      setBytes(barr, len);
+    }
+
+    public final void setBytes(final byte[] value, int len) {
+    <#else>
     public final void set${label}(final ${accessorType} value${putArgs}) {
+    </#if>
       <#-- Must compute the write offset first; can't be inline because the
            writeOffset() function has a side effect of possibly changing the buffer
            address (bufAddr). -->
@@ -355,7 +365,7 @@ import org.apache.drill.exec.vector.accessor.reader.BaseScalarReader;
 import org.apache.drill.exec.vector.accessor.writer.BaseScalarWriter;
 
 public class ColumnAccessorUtils {
-  
+
   private ColumnAccessorUtils() { }
 
 <@build vv.types "Required" "Reader" />
diff --git a/exec/vector/src/main/codegen/templates/ComplexWriters.java b/exec/vector/src/main/codegen/templates/ComplexWriters.java
index cfa049d12..6e1d8f3e2 100644
--- a/exec/vector/src/main/codegen/templates/ComplexWriters.java
+++ b/exec/vector/src/main/codegen/templates/ComplexWriters.java
@@ -98,7 +98,11 @@ public class ${eName}WriterImpl extends AbstractFieldWriter {
 
   <#if !(minor.class == "Decimal9" || minor.class == "Decimal18" || minor.class == "Decimal28Sparse" || minor.class == "Decimal38Sparse" || minor.class == "Decimal28Dense" || minor.class == "Decimal38Dense")>
   public void write${minor.class}(<#list fields as field>${field.type} ${field.name}<#if field_has_next>, </#if></#list>) {
+    <#if minor.class == "VarDecimal">
+    mutator.addSafe(idx(), <#list fields as field><#if field.name == "scale"><#break></#if>${field.name}<#if field_has_next && fields[field_index+1].name != "scale" >, </#if></#list>);
+    <#else>
     mutator.addSafe(idx(), <#list fields as field>${field.name}<#if field_has_next>, </#if></#list>);
+    </#if>
     vector.getMutator().setValueCount(idx()+1);
   }
   </#if>
@@ -123,7 +127,11 @@ public class ${eName}WriterImpl extends AbstractFieldWriter {
 
   <#if !(minor.class == "Decimal9" || minor.class == "Decimal18" || minor.class == "Decimal28Sparse" || minor.class == "Decimal38Sparse" || minor.class == "Decimal28Dense" || minor.class == "Decimal38Dense")>
   public void write${minor.class}(<#list fields as field>${field.type} ${field.name}<#if field_has_next>, </#if></#list>) {
+    <#if minor.class == "VarDecimal">
+    mutator.setSafe(idx(), <#if mode == "Nullable">1, </#if><#list fields as field><#if field.name == "scale"><#break></#if>${field.name}<#if field_has_next && fields[field_index+1].name != "scale" >, </#if></#list>);
+    <#else>
     mutator.setSafe(idx(), <#if mode == "Nullable">1, </#if><#list fields as field>${field.name}<#if field_has_next>, </#if></#list>);
+    </#if>
     vector.getMutator().setValueCount(idx()+1);
   }
 
diff --git a/exec/vector/src/main/codegen/templates/HolderReaderImpl.java b/exec/vector/src/main/codegen/templates/HolderReaderImpl.java
index 0eca723ce..e46989ba5 100644
--- a/exec/vector/src/main/codegen/templates/HolderReaderImpl.java
+++ b/exec/vector/src/main/codegen/templates/HolderReaderImpl.java
@@ -160,6 +160,8 @@ public class ${holderMode}${name}HolderReaderImpl extends AbstractFieldReader {
 
 <#if minor.class == "VarBinary">
       return value;
+<#elseif minor.class == "VarDecimal">
+      return org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(holder.buffer, holder.start, holder.end-holder.start, holder.scale);
 <#elseif minor.class == "Var16Char">
       return new String(value);
 <#elseif minor.class == "VarChar">
@@ -233,6 +235,8 @@ public class ${holderMode}${name}HolderReaderImpl extends AbstractFieldReader {
 
 <#if minor.class == "VarBinary">
       return value;
+<#elseif minor.class == "VarDecimal">
+      return org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(holder.buffer, holder.start, holder.end-holder.start, holder.scale);
 <#elseif minor.class == "Var16Char">
       return new String(value);
 <#elseif minor.class == "VarChar">
diff --git a/exec/vector/src/main/codegen/templates/RepeatedValueVectors.java b/exec/vector/src/main/codegen/templates/RepeatedValueVectors.java
index 412f498c2..2b2b6bd9a 100644
--- a/exec/vector/src/main/codegen/templates/RepeatedValueVectors.java
+++ b/exec/vector/src/main/codegen/templates/RepeatedValueVectors.java
@@ -405,11 +405,19 @@ public final class Repeated${minor.class}Vector extends BaseRepeatedValueVector
     }
 
     <#if (fields?size > 1) && !(minor.class == "Decimal9" || minor.class == "Decimal18" || minor.class == "Decimal28Sparse" || minor.class == "Decimal38Sparse" || minor.class == "Decimal28Dense" || minor.class == "Decimal38Dense")>
+    <#if minor.class == "VarDecimal">
+    public void addSafe(int arrayIndex, <#list fields as field><#if field.name == "scale"><#break></#if>${field.type} ${field.name}<#if field_has_next && fields[field_index+1].name != "scale" >, </#if></#list>) {
+      int nextOffset = offsets.getAccessor().get(arrayIndex+1);
+      values.getMutator().setSafe(nextOffset, <#list fields as field><#if field.name == "scale"><#break></#if>${field.name}<#if field_has_next && fields[field_index+1].name != "scale">, </#if></#list>);
+      offsets.getMutator().setSafe(arrayIndex+1, nextOffset+1);
+    }
+    <#else>
     public void addSafe(int rowIndex, <#list fields as field>${field.type} ${field.name}<#if field_has_next>, </#if></#list>) {
       final int nextOffset = offsets.getAccessor().get(rowIndex+1);
       values.getMutator().setSafe(nextOffset, <#list fields as field>${field.name}<#if field_has_next>, </#if></#list>);
       offsets.getMutator().setSafe(rowIndex+1, nextOffset+1);
     }
+    </#if>
 
     </#if>
     <#if minor.class == "Decimal28Sparse" || minor.class == "Decimal38Sparse">
diff --git a/exec/vector/src/main/codegen/templates/UnionReader.java b/exec/vector/src/main/codegen/templates/UnionReader.java
index 40ad89b82..54276f573 100644
--- a/exec/vector/src/main/codegen/templates/UnionReader.java
+++ b/exec/vector/src/main/codegen/templates/UnionReader.java
@@ -34,14 +34,14 @@ package org.apache.drill.exec.vector.complex.impl;
 @SuppressWarnings("unused")
 public class UnionReader extends AbstractFieldReader {
 
-  private BaseReader[] readers = new BaseReader[43];
+  private BaseReader[] readers = new BaseReader[44];
   public UnionVector data;
   
   public UnionReader(UnionVector data) {
     this.data = data;
   }
 
-  private static MajorType[] TYPES = new MajorType[43];
+  private static MajorType[] TYPES = new MajorType[44];
 
   static {
     for (MinorType minorType : MinorType.values()) {
diff --git a/exec/vector/src/main/codegen/templates/ValueHolders.java b/exec/vector/src/main/codegen/templates/ValueHolders.java
index 9982bd419..d5b4342df 100644
--- a/exec/vector/src/main/codegen/templates/ValueHolders.java
+++ b/exec/vector/src/main/codegen/templates/ValueHolders.java
@@ -90,9 +90,20 @@ public final class ${className} implements ValueHolder{
     public static boolean getSign(int start, DrillBuf buffer) {
       return ((buffer.getInt(start) & 0x80000000) != 0);
     }
-    </#if></#if>
 
-    public MajorType getType() {return TYPE;}
+    public java.math.BigDecimal getBigDecimal() {
+        java.math.BigDecimal currentValue = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromSparse(buffer, start, nDecimalDigits, scale);
+        return currentValue;
+    }
+    </#if></#if>
+    
+    <#if minor.class.startsWith("VarDecimal")>
+    public java.math.BigDecimal getBigDecimal() {
+        //System.out.println("valueHolder start " + start + " end " + " end " + " scale " + scale);
+        java.math.BigDecimal currentValue = org.apache.drill.exec.util.DecimalUtility.getBigDecimalFromDrillBuf(buffer, start, end-start, scale);
+        return currentValue;
+    }
+    </#if>
 
     @Deprecated
     public int hashCode(){
@@ -116,4 +127,4 @@ public final class ${className} implements ValueHolder{
 
 </#list>
 </#list>
-</#list>
\ No newline at end of file
+</#list>
diff --git a/exec/vector/src/main/codegen/templates/VariableLengthVectors.java b/exec/vector/src/main/codegen/templates/VariableLengthVectors.java
index 876d6880d..8ab4c3a48 100644
--- a/exec/vector/src/main/codegen/templates/VariableLengthVectors.java
+++ b/exec/vector/src/main/codegen/templates/VariableLengthVectors.java
@@ -474,6 +474,15 @@ public final class ${minor.class}Vector extends BaseDataValueVector implements V
     }
 
     <#switch minor.class>
+    <#case "VarDecimal">
+    @Override
+    public ${friendlyType} getObject(int index) {
+      byte[] b = get(index);
+      BigInteger bi = b.length == 0 ? new BigInteger("0") : new BigInteger(b);
+      BigDecimal bd = new BigDecimal(bi, getField().getScale());
+      return bd;
+    }
+    <#break>
     <#case "VarChar">
     @Override
     public ${friendlyType} getObject(int index) {
diff --git a/exec/vector/src/main/java/org/apache/drill/exec/util/DecimalUtility.java b/exec/vector/src/main/java/org/apache/drill/exec/util/DecimalUtility.java
index 914d68dee..82809547d 100644
--- a/exec/vector/src/main/java/org/apache/drill/exec/util/DecimalUtility.java
+++ b/exec/vector/src/main/java/org/apache/drill/exec/util/DecimalUtility.java
@@ -152,18 +152,18 @@ public class DecimalUtility extends CoreDecimalUtility{
     return getBigDecimalFromDrillBuf(data, startIndex, nDecimalDigits, scale, false);
   }
 
-  public static BigDecimal getBigDecimalFromSparse(DrillBuf data, int startIndex, int nDecimalDigits, int scale) {
+    public static BigDecimal getBigDecimalFromSparse(DrillBuf data, int startIndex, int nDecimalDigits, int scale) {
 
-    // In the sparse representation we pad the scale with zeroes for ease of arithmetic, need to truncate
-    return getBigDecimalFromDrillBuf(data, startIndex, nDecimalDigits, scale, true);
-  }
+        // In the sparse representation we pad the scale with zeroes for ease of arithmetic, need to truncate
+        return getBigDecimalFromDrillBuf(data, startIndex, nDecimalDigits, scale, true);
+    }
 
-  public static BigDecimal getBigDecimalFromDrillBuf(DrillBuf bytebuf, int start, int length, int scale) {
-    byte[] value = new byte[length];
-    bytebuf.getBytes(start, value, 0, length);
-    BigInteger unscaledValue = new BigInteger(value);
-    return new BigDecimal(unscaledValue, scale);
-  }
+    public static BigDecimal getBigDecimalFromDrillBuf(DrillBuf bytebuf, int start, int length, int scale) {
+      byte[] value = new byte[length];
+      bytebuf.getBytes(start, value, 0, length);
+      BigInteger unscaledValue = new BigInteger(value);
+      return new BigDecimal(unscaledValue, scale);
+    }
 
   public static BigDecimal getBigDecimalFromByteBuffer(ByteBuffer bytebuf, int start, int length, int scale) {
     byte[] value = new byte[length];
@@ -355,11 +355,29 @@ public class DecimalUtility extends CoreDecimalUtility{
         scale -= MAX_DIGITS;
     }
 
-    // Set the negative sign
-    if (sign == true) {
-        data.setInt(startIndex, data.getInt(startIndex) | 0x80000000);
+        // Set the negative sign
+        if (sign == true) {
+            data.setInt(startIndex, data.getInt(startIndex) | 0x80000000);
+        }
+
+    }
+
+    /**
+     * Converts from an input BigDecimal into varying width "VarDecimal" representation.
+     * The object that manages the "data" is assumed to already have the proper scale set,
+     * matching that of input.scale().
+     * @param input input decimal number to be stored
+     * @param data destination buffer to store the byte array representation of input
+     * @param startIndex starting index in data to hold the bytes
+     * @return startIndex + length of bytes stored (i.e., the next startIndex in the data buffer)
+     */
+    public static int getVarDecimalFromBigDecimal(BigDecimal input, ByteBuf data, int startIndex) {
+        byte[] bytes = input.unscaledValue().toByteArray();
+        int len = bytes.length;
+        data.setBytes(startIndex, bytes);
+        //System.out.println("getVarDecimal start " + startIndex + " len " + len + " value " + input);
+        return startIndex + len;
     }
-  }
 
   public static long getDecimal18FromBigDecimal(BigDecimal input, int scale, int precision) {
     // Truncate or pad to set the input to the correct scale
@@ -433,7 +451,31 @@ public class DecimalUtility extends CoreDecimalUtility{
     buffer.setInt(start + (index * 4), value);
   }
 
-  public static int compareSparseBytes(DrillBuf left, int leftStart, boolean leftSign, int leftScale, int leftPrecision, DrillBuf right, int rightStart, boolean rightSign, int rightPrecision, int rightScale, int width, int nDecimalDigits, boolean absCompare) {
+    /**
+     * Compares two VarDecimal values, still stored in their respective Drill buffers
+     * @param left left value Drill buffer
+     * @param leftStart start offset of left value
+     * @param leftEnd end offset of left value
+     * @param leftScale scale of left value
+     * @param right right value Drill buffer
+     * @param rightStart start offset of right value
+     * @param rightEnd end offset of right value
+     * @param rightScale scale of right value
+     * @param absCompare comparison of absolute values is done iff this is true
+     * @return 1 if left > right, 0 if left = right, -1 if left < right.  two values that are numerically equal, but with different
+     * scales (e.g., 2.00 and 2), are considered equal.
+     */
+    public static int compareVarLenBytes(DrillBuf left, int leftStart, int leftEnd, int leftScale, DrillBuf right, int rightStart, int rightEnd, int rightScale, boolean absCompare) {
+        java.math.BigDecimal bdLeft = getBigDecimalFromDrillBuf(left, leftStart, leftEnd - leftStart, leftScale);
+        java.math.BigDecimal bdRight = getBigDecimalFromDrillBuf(right, rightStart, rightEnd - rightStart, rightScale);
+        if (absCompare) {
+            bdLeft = bdLeft.abs();
+            bdRight = bdRight.abs();
+        }
+        return bdLeft.compareTo(bdRight);
+    }
+
+    public static int compareSparseBytes(DrillBuf left, int leftStart, boolean leftSign, int leftScale, int leftPrecision, DrillBuf right, int rightStart, boolean rightSign, int rightPrecision, int rightScale, int width, int nDecimalDigits, boolean absCompare) {
 
     int invert = 1;
 
diff --git a/exec/vector/src/main/java/org/apache/drill/exec/vector/complex/impl/MapOrListWriterImpl.java b/exec/vector/src/main/java/org/apache/drill/exec/vector/complex/impl/MapOrListWriterImpl.java
index b074abd17..1fa785748 100644
--- a/exec/vector/src/main/java/org/apache/drill/exec/vector/complex/impl/MapOrListWriterImpl.java
+++ b/exec/vector/src/main/java/org/apache/drill/exec/vector/complex/impl/MapOrListWriterImpl.java
@@ -28,6 +28,7 @@ import org.apache.drill.exec.vector.complex.writer.Decimal28SparseWriter;
 import org.apache.drill.exec.vector.complex.writer.Decimal38DenseWriter;
 import org.apache.drill.exec.vector.complex.writer.Decimal38SparseWriter;
 import org.apache.drill.exec.vector.complex.writer.Decimal9Writer;
+import org.apache.drill.exec.vector.complex.writer.VarDecimalWriter;
 import org.apache.drill.exec.vector.complex.writer.Float4Writer;
 import org.apache.drill.exec.vector.complex.writer.Float8Writer;
 import org.apache.drill.exec.vector.complex.writer.IntWriter;
@@ -223,6 +224,11 @@ public class MapOrListWriterImpl implements MapOrListWriter {
   }
 
   @Override
+  public VarDecimalWriter varDecimal(String name) {
+    return (map != null) ? map.varDecimal(name) : list.varDecimal();
+  }
+
+  @Override
   public Decimal38SparseWriter decimal38Sparse(String name) {
     return (map != null) ? map.decimal38Sparse(name) : list.decimal38Sparse();
   }
diff --git a/logical/src/main/antlr3/org/apache/drill/common/expression/parser/ExprLexer.g b/logical/src/main/antlr3/org/apache/drill/common/expression/parser/ExprLexer.g
index a881b5cb0..02979da26 100644
--- a/logical/src/main/antlr3/org/apache/drill/common/expression/parser/ExprLexer.g
+++ b/logical/src/main/antlr3/org/apache/drill/common/expression/parser/ExprLexer.g
@@ -62,6 +62,7 @@ DECIMAL28DENSE : 'decimal28dense' | 'DECIMAL28DENSE';
 DECIMAL28SPARSE : 'decimal28sparse' | 'DECIMAL28SPARSE';
 DECIMAL38DENSE : 'decimal38dense' | 'DECIMAL38DENSE';
 DECIMAL38SPARSE : 'decimal38sparse' | 'DECIMAL38SPARSE';
+VARDECIMAL : 'varDecimal' | 'VARDECIMAL';
 Or       : 'or' | 'OR' | 'Or';
 And      : 'and' | 'AND' ;
 Equals   : '==' | '=';
diff --git a/logical/src/main/antlr3/org/apache/drill/common/expression/parser/ExprParser.g b/logical/src/main/antlr3/org/apache/drill/common/expression/parser/ExprParser.g
index 12048b0c4..e73bdea00 100644
--- a/logical/src/main/antlr3/org/apache/drill/common/expression/parser/ExprParser.g
+++ b/logical/src/main/antlr3/org/apache/drill/common/expression/parser/ExprParser.g
@@ -119,6 +119,7 @@ numType returns [MajorType type]
 	| DECIMAL28SPARSE OParen precision Comma scale CParen { $type = TypeProtos.MajorType.newBuilder().setMinorType(TypeProtos.MinorType.DECIMAL28SPARSE).setMode(DataMode.REQUIRED).setPrecision($precision.value.intValue()).setScale($scale.value.intValue()).build(); }
 	| DECIMAL38DENSE OParen precision Comma scale CParen { $type = TypeProtos.MajorType.newBuilder().setMinorType(TypeProtos.MinorType.DECIMAL38DENSE).setMode(DataMode.REQUIRED).setPrecision($precision.value.intValue()).setScale($scale.value.intValue()).build(); }
 	| DECIMAL38SPARSE OParen precision Comma scale CParen { $type = TypeProtos.MajorType.newBuilder().setMinorType(TypeProtos.MinorType.DECIMAL38SPARSE).setMode(DataMode.REQUIRED).setPrecision($precision.value.intValue()).setScale($scale.value.intValue()).build(); }
+	| VARDECIMAL OParen precision Comma scale CParen { $type = TypeProtos.MajorType.newBuilder().setMinorType(TypeProtos.MinorType.VARDECIMAL).setMode(DataMode.REQUIRED).setPrecision($precision.value.intValue()).setScale($scale.value.intValue()).build(); }
 	;
 
 charType returns [MajorType type]
diff --git a/logical/src/main/java/org/apache/drill/common/expression/ExpressionStringBuilder.java b/logical/src/main/java/org/apache/drill/common/expression/ExpressionStringBuilder.java
index 34736df32..e5aec87ea 100644
--- a/logical/src/main/java/org/apache/drill/common/expression/ExpressionStringBuilder.java
+++ b/logical/src/main/java/org/apache/drill/common/expression/ExpressionStringBuilder.java
@@ -292,7 +292,7 @@ public class ExpressionStringBuilder extends AbstractExprVisitor<Void, StringBui
     case DECIMAL28SPARSE:
     case DECIMAL38DENSE:
     case DECIMAL38SPARSE:
-
+    case VARDECIMAL:  // TODO: precision might not be appropriate for VARDECIMAL, as it is "one-size-fits-all"
       // add scale and precision
       sb.append("(");
       sb.append(mt.getPrecision());
diff --git a/logical/src/main/java/org/apache/drill/common/expression/ValueExpressions.java b/logical/src/main/java/org/apache/drill/common/expression/ValueExpressions.java
index 527b6b73b..ef12d2f45 100644
--- a/logical/src/main/java/org/apache/drill/common/expression/ValueExpressions.java
+++ b/logical/src/main/java/org/apache/drill/common/expression/ValueExpressions.java
@@ -366,6 +366,8 @@ public class ValueExpressions {
 
   }
 
+  // TODO: is a class also needed for VARDECIMAL?
+
   public static class Decimal38Expression extends LogicalExpressionBase {
 
     private BigDecimal bigDecimal;
diff --git a/logical/src/main/java/org/apache/drill/common/expression/fn/CastFunctions.java b/logical/src/main/java/org/apache/drill/common/expression/fn/CastFunctions.java
index 62f9f3d7d..b5ed5b6a2 100644
--- a/logical/src/main/java/org/apache/drill/common/expression/fn/CastFunctions.java
+++ b/logical/src/main/java/org/apache/drill/common/expression/fn/CastFunctions.java
@@ -67,6 +67,7 @@ public class CastFunctions {
     TYPE2FUNC.put(MinorType.DECIMAL28DENSE, "castDECIMAL28DENSE");
     TYPE2FUNC.put(MinorType.DECIMAL38SPARSE, "castDECIMAL38SPARSE");
     TYPE2FUNC.put(MinorType.DECIMAL38DENSE, "castDECIMAL38DENSE");
+    TYPE2FUNC.put(MinorType.VARDECIMAL, "castVARDECIMAL");
 
     CAST_FUNC_REPLACEMENT_NEEDED.add(TYPE2FUNC.get(MinorType.INT));
     CAST_FUNC_REPLACEMENT_NEEDED.add(TYPE2FUNC.get(MinorType.BIGINT));
@@ -76,6 +77,7 @@ public class CastFunctions {
     CAST_FUNC_REPLACEMENT_NEEDED.add(TYPE2FUNC.get(MinorType.DECIMAL18));
     CAST_FUNC_REPLACEMENT_NEEDED.add(TYPE2FUNC.get(MinorType.DECIMAL28SPARSE));
     CAST_FUNC_REPLACEMENT_NEEDED.add(TYPE2FUNC.get(MinorType.DECIMAL38SPARSE));
+    CAST_FUNC_REPLACEMENT_NEEDED.add(TYPE2FUNC.get(MinorType.VARDECIMAL));
 
     CAST_FUNC_REPLACEMENT_FROM_NONNULLABLE_VARCHAR.put(TYPE2FUNC.get(MinorType.INT), "castEmptyStringVarCharToNullableINT");
     CAST_FUNC_REPLACEMENT_FROM_NONNULLABLE_VARCHAR.put(TYPE2FUNC.get(MinorType.BIGINT), "castEmptyStringVarCharToNullableBIGINT");
@@ -85,6 +87,7 @@ public class CastFunctions {
     CAST_FUNC_REPLACEMENT_FROM_NONNULLABLE_VARCHAR.put(TYPE2FUNC.get(MinorType.DECIMAL18), "castEmptyStringVarCharToNullableDECIMAL18");
     CAST_FUNC_REPLACEMENT_FROM_NONNULLABLE_VARCHAR.put(TYPE2FUNC.get(MinorType.DECIMAL28SPARSE), "castEmptyStringVarCharToNullableDECIMAL28SPARSE");
     CAST_FUNC_REPLACEMENT_FROM_NONNULLABLE_VARCHAR.put(TYPE2FUNC.get(MinorType.DECIMAL38SPARSE), "castEmptyStringVarCharToNullableDECIMAL38SPARSE");
+    CAST_FUNC_REPLACEMENT_FROM_NONNULLABLE_VARCHAR.put(TYPE2FUNC.get(MinorType.VARDECIMAL), "castEmptyStringVarCharToNullableVARDECIMAL");
 
     CAST_FUNC_REPLACEMENT_FROM_NONNULLABLE_VAR16CHAR.put(TYPE2FUNC.get(MinorType.INT), "castEmptyStringVar16CharToNullableINT");
     CAST_FUNC_REPLACEMENT_FROM_NONNULLABLE_VAR16CHAR.put(TYPE2FUNC.get(MinorType.BIGINT), "castEmptyStringVar16CharToNullableBIGINT");
@@ -94,6 +97,7 @@ public class CastFunctions {
     CAST_FUNC_REPLACEMENT_FROM_NONNULLABLE_VAR16CHAR.put(TYPE2FUNC.get(MinorType.DECIMAL18), "castEmptyStringVar16CharToNullableDECIMAL18");
     CAST_FUNC_REPLACEMENT_FROM_NONNULLABLE_VAR16CHAR.put(TYPE2FUNC.get(MinorType.DECIMAL28SPARSE), "castEmptyStringVar16CharToNullableDECIMAL28SPARSE");
     CAST_FUNC_REPLACEMENT_FROM_NONNULLABLE_VAR16CHAR.put(TYPE2FUNC.get(MinorType.DECIMAL38SPARSE), "castEmptyStringVar16CharToNullableDECIMAL38SPARSE");
+    CAST_FUNC_REPLACEMENT_FROM_NONNULLABLE_VAR16CHAR.put(TYPE2FUNC.get(MinorType.VARDECIMAL), "castEmptyStringVar16CharToNullableVARDECIMAL");
 
     CAST_FUNC_REPLACEMENT_FROM_NONNULLABLE_VARBINARY.put(TYPE2FUNC.get(MinorType.INT), "castEmptyStringVarBinaryToNullableINT");
     CAST_FUNC_REPLACEMENT_FROM_NONNULLABLE_VARBINARY.put(TYPE2FUNC.get(MinorType.BIGINT), "castEmptyStringVarBinaryToNullableBIGINT");
@@ -103,6 +107,7 @@ public class CastFunctions {
     CAST_FUNC_REPLACEMENT_FROM_NONNULLABLE_VARBINARY.put(TYPE2FUNC.get(MinorType.DECIMAL18), "castEmptyStringVarBinaryToNullableDECIMAL18");
     CAST_FUNC_REPLACEMENT_FROM_NONNULLABLE_VARBINARY.put(TYPE2FUNC.get(MinorType.DECIMAL28SPARSE), "castEmptyStringVarBinaryToNullableDECIMAL28SPARSE");
     CAST_FUNC_REPLACEMENT_FROM_NONNULLABLE_VARBINARY.put(TYPE2FUNC.get(MinorType.DECIMAL38SPARSE), "castEmptyStringVarBinaryToNullableDECIMAL38SPARSE");
+    CAST_FUNC_REPLACEMENT_FROM_NONNULLABLE_VARBINARY.put(TYPE2FUNC.get(MinorType.VARDECIMAL), "castEmptyStringVarBinaryToNullableVARDECIMAL");
 
     CAST_FUNC_REPLACEMENT_FROM_NULLABLE_VARCHAR.put(TYPE2FUNC.get(MinorType.INT), "castEmptyStringNullableVarCharToNullableINT");
     CAST_FUNC_REPLACEMENT_FROM_NULLABLE_VARCHAR.put(TYPE2FUNC.get(MinorType.BIGINT), "castEmptyStringNullableVarCharToNullableBIGINT");
@@ -112,6 +117,7 @@ public class CastFunctions {
     CAST_FUNC_REPLACEMENT_FROM_NULLABLE_VARCHAR.put(TYPE2FUNC.get(MinorType.DECIMAL18), "castEmptyStringNullableVarCharToNullableDECIMAL18");
     CAST_FUNC_REPLACEMENT_FROM_NULLABLE_VARCHAR.put(TYPE2FUNC.get(MinorType.DECIMAL28SPARSE), "castEmptyStringNullableVarCharToNullableDECIMAL28SPARSE");
     CAST_FUNC_REPLACEMENT_FROM_NULLABLE_VARCHAR.put(TYPE2FUNC.get(MinorType.DECIMAL38SPARSE), "castEmptyStringNullableVarCharToNullableDECIMAL38SPARSE");
+    CAST_FUNC_REPLACEMENT_FROM_NULLABLE_VARCHAR.put(TYPE2FUNC.get(MinorType.VARDECIMAL), "castEmptyStringNullableVarCharToNullableVARDECIMAL");
 
     CAST_FUNC_REPLACEMENT_FROM_NULLABLE_VAR16CHAR.put(TYPE2FUNC.get(MinorType.INT), "castEmptyStringNullableVar16CharToNullableINT");
     CAST_FUNC_REPLACEMENT_FROM_NULLABLE_VAR16CHAR.put(TYPE2FUNC.get(MinorType.BIGINT), "castEmptyStringNullableVar16CharToNullableBIGINT");
@@ -121,6 +127,7 @@ public class CastFunctions {
     CAST_FUNC_REPLACEMENT_FROM_NULLABLE_VAR16CHAR.put(TYPE2FUNC.get(MinorType.DECIMAL18), "castEmptyStringNullableVar16CharToNullableDECIMAL18");
     CAST_FUNC_REPLACEMENT_FROM_NULLABLE_VAR16CHAR.put(TYPE2FUNC.get(MinorType.DECIMAL28SPARSE), "castEmptyStringNullableVar16CharToNullableDECIMAL28SPARSE");
     CAST_FUNC_REPLACEMENT_FROM_NULLABLE_VAR16CHAR.put(TYPE2FUNC.get(MinorType.DECIMAL38SPARSE), "castEmptyStringNullableVar16CharToNullableDECIMAL38SPARSE");
+    CAST_FUNC_REPLACEMENT_FROM_NULLABLE_VAR16CHAR.put(TYPE2FUNC.get(MinorType.VARDECIMAL), "castEmptyStringNullableVar16CharToNullableVARDECIMAL");
 
     CAST_FUNC_REPLACEMENT_FROM_NULLABLE_VARBINARY.put(TYPE2FUNC.get(MinorType.INT), "castEmptyStringNullableVarBinaryToNullableINT");
     CAST_FUNC_REPLACEMENT_FROM_NULLABLE_VARBINARY.put(TYPE2FUNC.get(MinorType.BIGINT), "castEmptyStringNullableVarBinaryToNullableBIGINT");
@@ -130,6 +137,7 @@ public class CastFunctions {
     CAST_FUNC_REPLACEMENT_FROM_NULLABLE_VARBINARY.put(TYPE2FUNC.get(MinorType.DECIMAL18), "castEmptyStringNullableVarBinaryToNullableDECIMAL18");
     CAST_FUNC_REPLACEMENT_FROM_NULLABLE_VARBINARY.put(TYPE2FUNC.get(MinorType.DECIMAL28SPARSE), "castEmptyStringNullableVarBinaryToNullableDECIMAL28SPARSE");
     CAST_FUNC_REPLACEMENT_FROM_NULLABLE_VARBINARY.put(TYPE2FUNC.get(MinorType.DECIMAL38SPARSE), "castEmptyStringNullableVarBinaryToNullableDECIMAL38SPARSE");
+    CAST_FUNC_REPLACEMENT_FROM_NULLABLE_VARBINARY.put(TYPE2FUNC.get(MinorType.VARDECIMAL), "castEmptyStringNullableVarBinaryToNullableVARDECIMAL");
   }
 
   /**
diff --git a/protocol/src/main/java/org/apache/drill/common/types/TypeProtos.java b/protocol/src/main/java/org/apache/drill/common/types/TypeProtos.java
index b02298a70..c22b135fa 100644
--- a/protocol/src/main/java/org/apache/drill/common/types/TypeProtos.java
+++ b/protocol/src/main/java/org/apache/drill/common/types/TypeProtos.java
@@ -321,6 +321,14 @@ public final class TypeProtos {
      * <code>UNION = 42;</code>
      */
     UNION(37, 42),
+    /**
+     * <code>VARDECIMAL = 43;</code>
+     *
+     * <pre>
+     * variable width decimal (arbitrary precision)
+     * </pre>
+     */
+    VARDECIMAL(38, 43),
     ;
 
     /**
@@ -614,6 +622,14 @@ public final class TypeProtos {
      * <code>UNION = 42;</code>
      */
     public static final int UNION_VALUE = 42;
+    /**
+     * <code>VARDECIMAL = 43;</code>
+     *
+     * <pre>
+     * variable width decimal (arbitrary precision)
+     * </pre>
+     */
+    public static final int VARDECIMAL_VALUE = 43;
 
 
     public final int getNumber() { return value; }
@@ -658,6 +674,7 @@ public final class TypeProtos {
         case 40: return LIST;
         case 41: return GENERIC_OBJECT;
         case 42: return UNION;
+        case 43: return VARDECIMAL;
         default: return null;
       }
     }
@@ -1984,7 +2001,7 @@ public final class TypeProtos {
       "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*\225\004\n\tMinorType\022\010\n\004LATE\020\000\022\007\n\003MA" +
+      ".MinorType*\245\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" +
@@ -1997,10 +2014,10 @@ public final class TypeProtos {
       "\n\005UINT4\020\037\022\t\n\005UINT8\020 \022\022\n\016DECIMAL28DENSE\020!" +
       "\022\022\n\016DECIMAL38DENSE\020\"\022\010\n\004NULL\020%\022\020\n\014INTERV" +
       "ALYEAR\020&\022\017\n\013INTERVALDAY\020\'\022\010\n\004LIST\020(\022\022\n\016G" +
-      "ENERIC_OBJECT\020)\022\t\n\005UNION\020**4\n\010DataMode\022\014" +
-      "\n\010OPTIONAL\020\000\022\014\n\010REQUIRED\020\001\022\014\n\010REPEATED\020\002",
-      "B-\n\035org.apache.drill.common.typesB\nTypeP" +
-      "rotosH\001"
+      "ENERIC_OBJECT\020)\022\t\n\005UNION\020*\022\016\n\nVARDECIMAL" +
+      "\020+*4\n\010DataMode\022\014\n\010OPTIONAL\020\000\022\014\n\010REQUIRED",
+      "\020\001\022\014\n\010REPEATED\020\002B-\n\035org.apache.drill.com" +
+      "mon.typesB\nTypeProtosH\001"
     };
     com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner =
       new com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner() {
diff --git a/protocol/src/main/protobuf/Types.proto b/protocol/src/main/protobuf/Types.proto
index b2b29f085..6e4b66e86 100644
--- a/protocol/src/main/protobuf/Types.proto
+++ b/protocol/src/main/protobuf/Types.proto
@@ -65,6 +65,7 @@ enum MinorType {
     LIST = 40;
     GENERIC_OBJECT = 41;
     UNION = 42;
+    VARDECIMAL = 43;  // variable width decimal (arbitrary precision)
 }
 
 message MajorType {