DRILL-6094: Decimal data type enhancements

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

23 files changed, 555 insertions, 273 deletions

diff --git a/exec/vector/src/main/codegen/data/ValueVectorTypes.tdd b/exec/vector/src/main/codegen/data/ValueVectorTypes.tdd
index ca4653d3e..f6e1f1d09 100644
--- a/exec/vector/src/main/codegen/data/ValueVectorTypes.tdd
+++ b/exec/vector/src/main/codegen/data/ValueVectorTypes.tdd
@@ -183,7 +183,9 @@
       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}] }
+        { class: "VarDecimal", friendlyType: "BigDecimal", fields: [{name: "start", type: "int"},
+            {name: "end", type: "int"}, {name: "buffer", type: "DrillBuf"},
+            {name: "scale", type: "int", include: false}, {name: "precision", type: "int", include: false}] }
       ]
     },
     {
diff --git a/exec/vector/src/main/codegen/templates/AbstractFieldReader.java b/exec/vector/src/main/codegen/templates/AbstractFieldReader.java
index f2e1eb0dd..2ed5a3b6d 100644
--- a/exec/vector/src/main/codegen/templates/AbstractFieldReader.java
+++ b/exec/vector/src/main/codegen/templates/AbstractFieldReader.java
@@ -29,10 +29,9 @@ package org.apache.drill.exec.vector.complex.impl;
  * This class is generated using freemarker and the ${.template_name} template.
  */
 @SuppressWarnings("unused")
-abstract class AbstractFieldReader extends AbstractBaseReader implements FieldReader{
+abstract class AbstractFieldReader extends AbstractBaseReader implements FieldReader {
 
-  AbstractFieldReader(){
-    super();
+  AbstractFieldReader() {
   }
 
   /**
@@ -49,78 +48,77 @@ abstract class AbstractFieldReader extends AbstractBaseReader implements FieldRe
   <#assign safeType=friendlyType />
   <#if safeType=="byte[]"><#assign safeType="ByteArray" /></#if>
 
-  public ${friendlyType} read${safeType}(int arrayIndex){
+  public ${friendlyType} read${safeType}(int arrayIndex) {
     fail("read${safeType}(int arrayIndex)");
     return null;
   }
 
-  public ${friendlyType} read${safeType}(){
+  public ${friendlyType} read${safeType}() {
     fail("read${safeType}()");
     return null;
   }
 
   </#list>
 
-  public void copyAsValue(MapWriter writer){
+  public void copyAsValue(MapWriter writer) {
     fail("CopyAsValue MapWriter");
   }
-  public void copyAsField(String name, MapWriter writer){
+  public void copyAsField(String name, MapWriter writer) {
     fail("CopyAsField MapWriter");
   }
 
-  public void copyAsField(String name, ListWriter writer){
+  public void copyAsField(String name, ListWriter writer) {
     fail("CopyAsFieldList");
   }
 
   <#list vv.types as type><#list type.minor as minor><#assign name = minor.class?cap_first />
   <#assign boxedType = (minor.boxedType!type.boxedType) />
 
-  public void read(${name}Holder holder){
+  public void read(${name}Holder holder) {
     fail("${name}");
   }
 
-  public void read(Nullable${name}Holder holder){
+  public void read(Nullable${name}Holder holder) {
     fail("${name}");
   }
 
-  public void read(int arrayIndex, ${name}Holder holder){
+  public void read(int arrayIndex, ${name}Holder holder) {
     fail("Repeated${name}");
   }
 
-  public void read(int arrayIndex, Nullable${name}Holder holder){
+  public void read(int arrayIndex, Nullable${name}Holder holder) {
     fail("Repeated${name}");
   }
 
-  public void copyAsValue(${name}Writer writer){
+  public void copyAsValue(${name}Writer writer) {
     fail("CopyAsValue${name}");
   }
-  public void copyAsField(String name, ${name}Writer writer){
+
+  <#if minor.class == "VarDecimal">
+  public void copyAsField(String name, ${name}Writer writer, int scale, int precision) {
+  <#else>
+  public void copyAsField(String name, ${name}Writer writer) {
+  </#if>
     fail("CopyAsField${name}");
   }
   </#list></#list>
 
-  public FieldReader reader(String name){
+  public FieldReader reader(String name) {
     fail("reader(String name)");
     return null;
   }
 
-  public FieldReader reader(){
+  public FieldReader reader() {
     fail("reader()");
     return null;
-
   }
 
-  public int size(){
+  public int size() {
     fail("size()");
     return -1;
   }
 
-  private void fail(String name){
+  private void fail(String name) {
     throw new IllegalArgumentException(String.format("You tried to read a [%s] type when you are using a field reader of type [%s].", name, this.getClass().getSimpleName()));
   }
-
-
 }
-
-
-
diff --git a/exec/vector/src/main/codegen/templates/AbstractFieldWriter.java b/exec/vector/src/main/codegen/templates/AbstractFieldWriter.java
index 608420cce..dcf99d34b 100644
--- a/exec/vector/src/main/codegen/templates/AbstractFieldWriter.java
+++ b/exec/vector/src/main/codegen/templates/AbstractFieldWriter.java
@@ -65,6 +65,12 @@ abstract class AbstractFieldWriter extends AbstractBaseWriter implements FieldWr
     fail("${name}");
   }
 
+  <#if minor.class?contains("Decimal") >
+  public void write${minor.class}(BigDecimal value) {
+    fail("${name}");
+  }
+  </#if>
+
   </#list></#list>
 
   public void writeNull() {
@@ -111,11 +117,18 @@ abstract class AbstractFieldWriter extends AbstractBaseWriter implements FieldWr
   <#if lowerName == "int" ><#assign lowerName = "integer" /></#if>
   <#assign upperName = minor.class?upper_case />
   <#assign capName = minor.class?cap_first />
-  <#if minor.class?starts_with("Decimal") >
+  <#if minor.class?contains("Decimal") >
+  @Override
   public ${capName}Writer ${lowerName}(String name, int scale, int precision) {
     fail("${capName}");
     return null;
   }
+
+  @Override
+  public ${capName}Writer ${lowerName}(int scale, int precision) {
+    fail("${capName}");
+    return null;
+  }
   </#if>
 
   @Override
diff --git a/exec/vector/src/main/codegen/templates/AbstractPromotableFieldWriter.java b/exec/vector/src/main/codegen/templates/AbstractPromotableFieldWriter.java
index df7be39bd..96b45d329 100644
--- a/exec/vector/src/main/codegen/templates/AbstractPromotableFieldWriter.java
+++ b/exec/vector/src/main/codegen/templates/AbstractPromotableFieldWriter.java
@@ -75,12 +75,20 @@ abstract class AbstractPromotableFieldWriter extends AbstractFieldWriter {
 
   <#list vv.types as type><#list type.minor as minor><#assign name = minor.class?cap_first />
   <#assign fields = minor.fields!type.fields />
-  <#if !minor.class?starts_with("Decimal") >
+  <#if minor.class?contains("VarDecimal")>
+  @Override
+  public void write${minor.class}(BigDecimal value) {
+    getWriter(MinorType.${name?upper_case}).write${minor.class}(value);
+  }
+  </#if>
+
   @Override
   public void write(${name}Holder holder) {
     getWriter(MinorType.${name?upper_case}).write(holder);
   }
 
+  <#-- This condition was added to cover previous decimal functionality for new VarDecimal type -->
+  <#if !minor.class?contains("Decimal") || minor.class?contains("VarDecimal")>
   public void write${minor.class}(<#list fields as field>${field.type} ${field.name}<#if field_has_next>, </#if></#list>) {
     getWriter(MinorType.${name?upper_case}).write${minor.class}(<#list fields as field>${field.name}<#if field_has_next>, </#if></#list>);
   }
@@ -116,7 +124,19 @@ abstract class AbstractPromotableFieldWriter extends AbstractFieldWriter {
   <#if lowerName == "int" ><#assign lowerName = "integer" /></#if>
   <#assign upperName = minor.class?upper_case />
   <#assign capName = minor.class?cap_first />
-  <#if !minor.class?starts_with("Decimal") >
+  <#if minor.class?contains("Decimal")>
+
+  @Override
+  public ${capName}Writer ${lowerName}(String name, int scale, int precision) {
+    return getWriter(MinorType.MAP).${lowerName}(name, scale, precision);
+  }
+
+  @Override
+  public ${capName}Writer ${lowerName}(int scale, int precision) {
+    return getWriter(MinorType.LIST).${lowerName}(scale, precision);
+  }
+
+  <#else>
 
   @Override
   public ${capName}Writer ${lowerName}(String name) {
diff --git a/exec/vector/src/main/codegen/templates/BaseWriter.java b/exec/vector/src/main/codegen/templates/BaseWriter.java
index ad9c44e1f..e65060976 100644
--- a/exec/vector/src/main/codegen/templates/BaseWriter.java
+++ b/exec/vector/src/main/codegen/templates/BaseWriter.java
@@ -53,7 +53,7 @@ package org.apache.drill.exec.vector.complex.writer;
     <#if lowerName == "int" ><#assign lowerName = "integer" /></#if>
     <#assign upperName = minor.class?upper_case />
     <#assign capName = minor.class?cap_first />
-    <#if minor.class?starts_with("Decimal") >
+    <#if minor.class?contains("Decimal") >
     ${capName}Writer ${lowerName}(String name, int scale, int precision);
     </#if>
     ${capName}Writer ${lowerName}(String name);
@@ -78,6 +78,9 @@ package org.apache.drill.exec.vector.complex.writer;
     <#if lowerName == "int" ><#assign lowerName = "integer" /></#if>
     <#assign upperName = minor.class?upper_case />
     <#assign capName = minor.class?cap_first />
+    <#if minor.class?contains("Decimal") >
+    ${capName}Writer ${lowerName}(int scale, int precision);
+    </#if>
     ${capName}Writer ${lowerName}();
     </#list></#list>
   }
@@ -112,6 +115,7 @@ package org.apache.drill.exec.vector.complex.writer;
     VarCharWriter varChar(String name);
     Var16CharWriter var16Char(String name);
     VarDecimalWriter varDecimal(String name);
+    VarDecimalWriter varDecimal(String name, int scale, int precision);
     TinyIntWriter tinyInt(String name);
     SmallIntWriter smallInt(String name);
     IntWriter integer(String name);
diff --git a/exec/vector/src/main/codegen/templates/BasicTypeHelper.java b/exec/vector/src/main/codegen/templates/BasicTypeHelper.java
index 8160dd59a..f818c236f 100644
--- a/exec/vector/src/main/codegen/templates/BasicTypeHelper.java
+++ b/exec/vector/src/main/codegen/templates/BasicTypeHelper.java
@@ -395,7 +395,9 @@ public class BasicTypeHelper {
     MajorType type1 = v1.getField().getType();
     MajorType type2 = v2.getField().getType();
 
-    if (type1.getMinorType() != type2.getMinorType()) {
+    if (type1.getMinorType() != type2.getMinorType()
+        || type1.getScale() != type1.getScale()
+        || type1.getPrecision() != type1.getPrecision()) {
       return false;
     }
 
@@ -539,11 +541,19 @@ public class BasicTypeHelper {
     }
     <#list vv.types as type>
     <#list type.minor as minor>
+    <#if minor.class.contains("Decimal")>
+      else if (holder instanceof ${minor.class}Holder) {
+        return  getType((${minor.class}Holder) holder);
+      } else if (holder instanceof Nullable${minor.class}Holder) {
+        return  getType((Nullable${minor.class}Holder) holder);
+      }
+    <#else>
       else if (holder instanceof ${minor.class}Holder) {
         return ((${minor.class}Holder) holder).TYPE;
       } else if (holder instanceof Nullable${minor.class}Holder) {
       return ((Nullable${minor.class}Holder) holder).TYPE;
       }
+    </#if>
     </#list>
     </#list>
     else if (holder instanceof UntypedNullHolder) {
@@ -553,4 +563,31 @@ public class BasicTypeHelper {
 
   }
 
+  <#list vv.types as type>
+  <#list type.minor as minor>
+  <#if minor.class.contains("Decimal")>
+  <#list ["Nullable", "", "Repeated"] as dataMode>
+  public static MajorType getType(${dataMode}${minor.class}Holder holder) {
+    return MajorType.newBuilder()
+        .setMinorType(MinorType.${minor.class?upper_case})
+      <#if dataMode == "Nullable">
+        .setMode(DataMode.OPTIONAL)
+        .setScale(holder.scale)
+        .setPrecision(holder.precision)
+      <#elseif dataMode == "Repeated">
+        .setMode(DataMode.REPEATED)
+        .setScale(holder.vector.getField().getScale())
+        .setPrecision(holder.vector.getField().getPrecision())
+      <#else>
+        .setMode(DataMode.REQUIRED)
+        .setScale(holder.scale)
+        .setPrecision(holder.precision)
+      </#if>
+        .build();
+  }
+  </#list>
+  </#if>
+  </#list>
+  </#list>
+
 }
diff --git a/exec/vector/src/main/codegen/templates/ColumnAccessors.java b/exec/vector/src/main/codegen/templates/ColumnAccessors.java
index 6068afa22..accf1b28d 100644
--- a/exec/vector/src/main/codegen/templates/ColumnAccessors.java
+++ b/exec/vector/src/main/codegen/templates/ColumnAccessors.java
@@ -28,6 +28,8 @@
       return ValueType.INTEGER;
   <#elseif drillType == "VarChar" || drillType == "Var16Char">
       return ValueType.STRING;
+  <#elseif drillType == "VarDecimal">
+    return ValueType.DECIMAL;
   <#else>
       return ValueType.${label?upper_case};
   </#if>
@@ -65,6 +67,7 @@
 package org.apache.drill.exec.vector.accessor;
 
 import java.math.BigDecimal;
+import java.math.BigInteger;
 
 import org.apache.drill.common.types.TypeProtos.MajorType;
 import org.apache.drill.exec.vector.DateUtilities;
@@ -141,12 +144,10 @@ public class ColumnAccessors {
 
     private static final int VALUE_WIDTH = ${drillType}Vector.VALUE_WIDTH;
 
-      <#if decimal>
-    private MajorType type;
-
-      </#if>
     </#if>
     <#if decimal>
+    private MajorType type;
+
     @Override
     public void bindVector(ColumnMetadata schema, VectorAccessor va) {
       super.bindVector(schema, va);
@@ -234,6 +235,14 @@ public class ColumnAccessors {
     public String getString() {
       return new String(getBytes(${indexVar}), Charsets.UTF_16);
     }
+  <#elseif drillType == "VarDecimal">
+
+    @Override
+    public BigDecimal getDecimal() {
+      byte[] bytes = getBytes();
+      BigInteger unscaledValue = bytes.length == 0 ? BigInteger.ZERO : new BigInteger(bytes);
+      return new BigDecimal(unscaledValue, type.getScale());
+    }
   </#if>
   }
 
@@ -269,20 +278,10 @@ public class ColumnAccessors {
       </#if>
       <@getType drillType label />
       <#if ! varWidth>
-
     </#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>
+    @Override
     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). -->
@@ -347,6 +346,15 @@ public class ColumnAccessors {
       final byte bytes[] = value.getBytes(Charsets.UTF_16);
       setBytes(bytes, bytes.length);
     }
+
+    <#elseif drillType = "VarDecimal">
+
+    @Override
+    public final void setDecimal(final BigDecimal bd) {
+      byte[] barr = bd.unscaledValue().toByteArray();
+      int len = barr.length;
+      setBytes(barr, len);
+    }
     </#if>
   }
 
diff --git a/exec/vector/src/main/codegen/templates/ComplexCopier.java b/exec/vector/src/main/codegen/templates/ComplexCopier.java
index 91e3ff645..6d7c7e607 100644
--- a/exec/vector/src/main/codegen/templates/ComplexCopier.java
+++ b/exec/vector/src/main/codegen/templates/ComplexCopier.java
@@ -97,9 +97,12 @@ public class ComplexCopier {
     <#list vv.types as type><#list type.minor as minor><#assign name = minor.class?cap_first />
     <#assign fields = minor.fields!type.fields />
     <#assign uncappedName = name?uncap_first/>
-    <#if !minor.class?starts_with("Decimal")>
+    <#if !minor.class?contains("Decimal")>
     case ${name?upper_case}:
       return (FieldWriter) writer.<#if name == "Int">integer<#else>${uncappedName}</#if>(name);
+    <#elseif minor.class?contains("VarDecimal")>
+    case ${name?upper_case}:
+      return (FieldWriter) writer.${uncappedName}(name, reader.getType().getScale(), reader.getType().getPrecision());
     </#if>
     </#list></#list>
     case MAP:
@@ -116,9 +119,12 @@ public class ComplexCopier {
     <#list vv.types as type><#list type.minor as minor><#assign name = minor.class?cap_first />
     <#assign fields = minor.fields!type.fields />
     <#assign uncappedName = name?uncap_first/>
-    <#if !minor.class?starts_with("Decimal")>
+    <#if !minor.class?contains("Decimal")>
+    case ${name?upper_case}:
+      return (FieldWriter) writer.<#if name == "Int">integer<#else>${uncappedName}</#if>();
+    <#elseif minor.class?contains("VarDecimal")>
     case ${name?upper_case}:
-    return (FieldWriter) writer.<#if name == "Int">integer<#else>${uncappedName}</#if>();
+      return (FieldWriter) writer.${uncappedName}(reader.getType().getScale(), reader.getType().getPrecision());
     </#if>
     </#list></#list>
     case MAP:
diff --git a/exec/vector/src/main/codegen/templates/ComplexReaders.java b/exec/vector/src/main/codegen/templates/ComplexReaders.java
index 1f0b92b02..d72c7c696 100644
--- a/exec/vector/src/main/codegen/templates/ComplexReaders.java
+++ b/exec/vector/src/main/codegen/templates/ComplexReaders.java
@@ -53,20 +53,20 @@ public class ${nullMode}${name}ReaderImpl extends AbstractFieldReader {
   
   private final ${nullMode}${name}Vector vector;
   
-  public ${nullMode}${name}ReaderImpl(${nullMode}${name}Vector vector){
+  public ${nullMode}${name}ReaderImpl(${nullMode}${name}Vector vector) {
     super();
     this.vector = vector;
   }
 
-  public MajorType getType(){
+  public MajorType getType() {
     return vector.getField().getType();
   }
 
-  public MaterializedField getField(){
+  public MaterializedField getField() {
     return vector.getField();
   }
   
-  public boolean isSet(){
+  public boolean isSet() {
     <#if nullMode == "Nullable">
     return !vector.getAccessor().isNull(idx());
     <#else>
@@ -74,76 +74,85 @@ public class ${nullMode}${name}ReaderImpl extends AbstractFieldReader {
     </#if>
   }
 
-
-  
-  
   <#if mode == "Repeated">
 
-  public void copyAsValue(${minor.class?cap_first}Writer writer){
+  public void copyAsValue(${minor.class?cap_first}Writer writer) {
     Repeated${minor.class?cap_first}WriterImpl impl = (Repeated${minor.class?cap_first}WriterImpl) writer;
     impl.vector.copyFromSafe(idx(), impl.idx(), vector);
   }
-  
-  public void copyAsField(String name, MapWriter writer){
-    Repeated${minor.class?cap_first}WriterImpl impl = (Repeated${minor.class?cap_first}WriterImpl)  writer.list(name).${lowerName}();
+
+  <#if minor.class == "VarDecimal">
+  public void copyAsField(String name, MapWriter writer, int scale, int precision) {
+    Repeated${minor.class?cap_first}WriterImpl impl
+        = (Repeated${minor.class?cap_first}WriterImpl) writer.list(name).${lowerName}(scale, precision);
+  <#else>
+  public void copyAsField(String name, MapWriter writer) {
+    Repeated${minor.class?cap_first}WriterImpl impl = (Repeated${minor.class?cap_first}WriterImpl) writer.list(name).${lowerName}();
+  </#if>
     impl.vector.copyFromSafe(idx(), impl.idx(), vector);
   }
   
-  public int size(){
+  public int size() {
     return vector.getAccessor().getInnerValueCountAt(idx());
   }
   
-  public void read(int arrayIndex, ${minor.class?cap_first}Holder h){
+  public void read(int arrayIndex, ${minor.class?cap_first}Holder h) {
     vector.getAccessor().get(idx(), arrayIndex, h);
   }
-  public void read(int arrayIndex, Nullable${minor.class?cap_first}Holder h){
+
+  public void read(int arrayIndex, Nullable${minor.class?cap_first}Holder h) {
     vector.getAccessor().get(idx(), arrayIndex, h);
   }
   
-  public ${friendlyType} read${safeType}(int arrayIndex){
+  public ${friendlyType} read${safeType}(int arrayIndex) {
     return vector.getAccessor().getSingleObject(idx(), arrayIndex);
   }
 
   
-  public List<Object> readObject(){
+  public List<Object> readObject() {
     return (List<Object>) (Object) vector.getAccessor().getObject(idx());
   }
   
   <#else>
   
-  public void copyAsValue(${minor.class?cap_first}Writer writer){
+  public void copyAsValue(${minor.class?cap_first}Writer writer) {
     ${nullMode}${minor.class?cap_first}WriterImpl impl = (${nullMode}${minor.class?cap_first}WriterImpl) writer;
     impl.vector.copyFromSafe(idx(), impl.idx(), vector);
   }
-  
-  public void copyAsField(String name, MapWriter writer){
+
+  <#if minor.class == "VarDecimal">
+  public void copyAsField(String name, MapWriter writer, int scale, int precision) {
+    ${nullMode}${minor.class?cap_first}WriterImpl impl
+        = (${nullMode}${minor.class?cap_first}WriterImpl) writer.${lowerName}(name, scale, precision);
+<#else>
+  public void copyAsField(String name, MapWriter writer) {
     ${nullMode}${minor.class?cap_first}WriterImpl impl = (${nullMode}${minor.class?cap_first}WriterImpl) writer.${lowerName}(name);
+  </#if>
     impl.vector.copyFromSafe(idx(), impl.idx(), vector);
   }
 
   <#if nullMode != "Nullable">
-  public void read(${minor.class?cap_first}Holder h){
+  public void read(${minor.class?cap_first}Holder h) {
     vector.getAccessor().get(idx(), h);
   }
   </#if>
 
-  public void read(Nullable${minor.class?cap_first}Holder h){
+  public void read(Nullable${minor.class?cap_first}Holder h) {
     vector.getAccessor().get(idx(), h);
   }
   
-  public ${friendlyType} read${safeType}(){
+  public ${friendlyType} read${safeType}() {
     return vector.getAccessor().getObject(idx());
   }
   
-  public void copyValue(FieldWriter w){
+  public void copyValue(FieldWriter w) {
     
   }
   
-  public Object readObject(){
+  public Object readObject() {
     return vector.getAccessor().getObject(idx());
   }
 
-  
   </#if>
 }
 </#if>
@@ -155,7 +164,7 @@ package org.apache.drill.exec.vector.complex.reader;
 
 <#include "/@includes/vv_imports.ftl" />
 @SuppressWarnings("unused")
-public interface ${name}Reader extends BaseReader{
+public interface ${name}Reader extends BaseReader {
   
   <#if mode == "Repeated">
   public int size();
@@ -171,8 +180,11 @@ public interface ${name}Reader extends BaseReader{
   </#if>  
   public boolean isSet();
   public void copyAsValue(${minor.class}Writer writer);
+  <#if minor.class == "VarDecimal">
+  public void copyAsField(String name, ${minor.class}Writer writer, int scale, int precision);
+  <#else>
   public void copyAsField(String name, ${minor.class}Writer writer);
-  
+  </#if>
 }
 
 
diff --git a/exec/vector/src/main/codegen/templates/ComplexWriters.java b/exec/vector/src/main/codegen/templates/ComplexWriters.java
index 6e1d8f3e2..f46b7958d 100644
--- a/exec/vector/src/main/codegen/templates/ComplexWriters.java
+++ b/exec/vector/src/main/codegen/templates/ComplexWriters.java
@@ -96,14 +96,10 @@ public class ${eName}WriterImpl extends AbstractFieldWriter {
     vector.getMutator().setValueCount(idx()+1);
   }
 
-  <#if !(minor.class == "Decimal9" || minor.class == "Decimal18" || minor.class == "Decimal28Sparse" || minor.class == "Decimal38Sparse" || minor.class == "Decimal28Dense" || minor.class == "Decimal38Dense")>
+  <#if !minor.class?contains("Decimal") || minor.class?contains("VarDecimal")>
   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);
+    mutator.addSafe(idx()<#list fields as field><#if field.include!true>, ${field.name}</#if></#list>);
+    vector.getMutator().setValueCount(idx() + 1);
   }
   </#if>
 
@@ -125,21 +121,23 @@ public class ${eName}WriterImpl extends AbstractFieldWriter {
     vector.getMutator().setValueCount(idx()+1);
   }
 
-  <#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);
-  }
-
   <#if mode == "Nullable">
-
   public void writeNull() {
     mutator.setNull(idx());
-    vector.getMutator().setValueCount(idx()+1);
+    vector.getMutator().setValueCount(idx() + 1);
+  }
+  </#if>
+
+  <#if !(minor.class == "Decimal9" || minor.class == "Decimal18")>
+  public void write${minor.class}(<#list fields as field>${field.type} ${field.name}<#if field_has_next>, </#if></#list>) {
+    mutator.setSafe(idx()<#if mode == "Nullable">, 1</#if><#list fields as field><#if field.include!true>, ${field.name}</#if></#list>);
+    vector.getMutator().setValueCount(idx() + 1);
+  }
+
+  <#if minor.class?contains("VarDecimal")>
+  public void write${minor.class}(BigDecimal value) {
+    mutator.setSafe(idx(), value);
+    vector.getMutator().setValueCount(idx() + 1);
   }
   </#if>
   </#if>
@@ -156,9 +154,13 @@ package org.apache.drill.exec.vector.complex.writer;
 public interface ${eName}Writer extends BaseWriter {
   public void write(${minor.class}Holder h);
 
-  <#if !(minor.class == "Decimal9" || minor.class == "Decimal18" || minor.class == "Decimal28Sparse" || minor.class == "Decimal38Sparse" || minor.class == "Decimal28Dense" || minor.class == "Decimal38Dense")>
+  <#if !(minor.class == "Decimal9" || minor.class == "Decimal18")>
   public void write${minor.class}(<#list fields as field>${field.type} ${field.name}<#if field_has_next>, </#if></#list>);
   </#if>
+
+  <#if minor.class?contains("VarDecimal")>
+  public void write${minor.class}(BigDecimal value);
+  </#if>
 }
 
 </#list>
diff --git a/exec/vector/src/main/codegen/templates/HolderReaderImpl.java b/exec/vector/src/main/codegen/templates/HolderReaderImpl.java
index e46989ba5..4b7be4fd7 100644
--- a/exec/vector/src/main/codegen/templates/HolderReaderImpl.java
+++ b/exec/vector/src/main/codegen/templates/HolderReaderImpl.java
@@ -42,6 +42,7 @@ import java.math.BigDecimal;
 import java.math.BigInteger;
 
 import org.apache.drill.exec.expr.holders.*;
+import org.apache.drill.exec.expr.BasicTypeHelper;
 import org.joda.time.Period;
 
 // Source code generated using FreeMarker template ${.template_name}
@@ -96,10 +97,14 @@ public class ${holderMode}${name}HolderReaderImpl extends AbstractFieldReader {
 
   @Override
   public MajorType getType() {
-<#if holderMode == "Repeated">
-    return this.repeatedHolder.TYPE;
+<#if name?contains("Decimal")>
+    return BasicTypeHelper.getType(holder);
 <#else>
+  <#if holderMode == "Repeated">
+    return this.repeatedHolder.TYPE;
+  <#else>
     return this.holder.TYPE;
+  </#if>
 </#if>
   }
 
@@ -161,7 +166,7 @@ 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);
+      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">
@@ -236,7 +241,7 @@ 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);
+      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/ListWriters.java b/exec/vector/src/main/codegen/templates/ListWriters.java
index d7a66f75d..cab8772a7 100644
--- a/exec/vector/src/main/codegen/templates/ListWriters.java
+++ b/exec/vector/src/main/codegen/templates/ListWriters.java
@@ -156,10 +156,24 @@ public class ${mode}ListWriter extends AbstractFieldWriter {
   <#assign upperName = minor.class?upper_case />
   <#assign capName = minor.class?cap_first />
   <#if lowerName == "int" ><#assign lowerName = "integer" /></#if>
+
+  <#if minor.class?contains("Decimal") >
+  @Override
+  public ${capName}Writer ${lowerName}() {
+    // returns existing writer
+    assert mode == Mode.IN_${upperName};
+    return writer;
+  }
+
+  @Override
+  public ${capName}Writer ${lowerName}(int scale, int precision) {
+    final MajorType ${upperName}_TYPE = Types.withScaleAndPrecision(MinorType.${upperName}, DataMode.REPEATED, scale, precision);
+  <#else>
   private static final MajorType ${upperName}_TYPE = Types.repeated(MinorType.${upperName});
 
   @Override
   public ${capName}Writer ${lowerName}() {
+  </#if>
     switch (mode) {
     case INIT:
       final int vectorCount = container.size();
@@ -181,14 +195,14 @@ public class ${mode}ListWriter extends AbstractFieldWriter {
          .build(logger);
     }
   }
-  
+
   </#list></#list>
   @Override
   public MaterializedField getField() {
     return container.getField();
   }
   <#if mode == "Repeated">
-  
+
   @Override
   public void startList() {
     final RepeatedListVector list = (RepeatedListVector) container;
diff --git a/exec/vector/src/main/codegen/templates/MapWriters.java b/exec/vector/src/main/codegen/templates/MapWriters.java
index 0f0552817..9074293d7 100644
--- a/exec/vector/src/main/codegen/templates/MapWriters.java
+++ b/exec/vector/src/main/codegen/templates/MapWriters.java
@@ -197,7 +197,8 @@ public class ${mode}MapWriter extends AbstractFieldWriter {
   <#assign vectName = capName />
   <#assign vectName = "Nullable${capName}" />
 
-  <#if minor.class?starts_with("Decimal") >
+  <#if minor.class?contains("Decimal") >
+  @Override
   public ${minor.class}Writer ${lowerName}(String name) {
     // returns existing writer
     final FieldWriter writer = fields.get(name.toLowerCase());
@@ -205,6 +206,7 @@ public class ${mode}MapWriter extends AbstractFieldWriter {
     return writer;
   }
 
+  @Override
   public ${minor.class}Writer ${lowerName}(String name, int scale, int precision) {
     final MajorType ${upperName}_TYPE = Types.withScaleAndPrecision(MinorType.${upperName}, DataMode.OPTIONAL, scale, precision);
   <#else>
diff --git a/exec/vector/src/main/codegen/templates/NullReader.java b/exec/vector/src/main/codegen/templates/NullReader.java
index 32ee9b9b9..4d867baa4 100644
--- a/exec/vector/src/main/codegen/templates/NullReader.java
+++ b/exec/vector/src/main/codegen/templates/NullReader.java
@@ -31,20 +31,18 @@ import org.apache.drill.common.types.TypeProtos;
  * This class is generated using freemarker and the ${.template_name} template.
  */
 @SuppressWarnings("unused")
-public class NullReader extends AbstractBaseReader implements FieldReader{
+public class NullReader extends AbstractBaseReader implements FieldReader {
   
   public static final NullReader INSTANCE = new NullReader();
   public static final NullReader EMPTY_LIST_INSTANCE = new NullReader(Types.repeated(TypeProtos.MinorType.NULL));
   public static final NullReader EMPTY_MAP_INSTANCE = new NullReader(Types.required(TypeProtos.MinorType.MAP));
   private MajorType type;
   
-  private NullReader(){
-    super();
+  private NullReader() {
     type = Types.NULL;
   }
 
-  private NullReader(MajorType type){
-    super();
+  private NullReader(MajorType type) {
     this.type = type;
   }
 
@@ -60,63 +58,67 @@ public class NullReader extends AbstractBaseReader implements FieldReader{
   public void copyAsValue(UnionWriter writer) {}
 
   <#list vv.types as type><#list type.minor as minor><#assign name = minor.class?cap_first />
-  public void read(${name}Holder holder){
+  public void read(${name}Holder holder) {
     throw new UnsupportedOperationException("NullReader cannot write into non-nullable holder");
   }
 
-  public void read(Nullable${name}Holder holder){
+  public void read(Nullable${name}Holder holder) {
     holder.isSet = 0;
   }
 
-  public void read(int arrayIndex, ${name}Holder holder){
+  public void read(int arrayIndex, ${name}Holder holder) {
     throw new ArrayIndexOutOfBoundsException();
   }
   
-  public void copyAsValue(${minor.class}Writer writer){}
-  public void copyAsField(String name, ${minor.class}Writer writer){}
+  public void copyAsValue(${minor.class}Writer writer) {}
+  <#if minor.class == "VarDecimal">
+  public void copyAsField(String name, ${minor.class}Writer writer, int scale, int precision) {}
+  <#else>
+  public void copyAsField(String name, ${minor.class}Writer writer) {}
+  </#if>
 
-  public void read(int arrayIndex, Nullable${name}Holder holder){
+  public void read(int arrayIndex, Nullable${name}Holder holder) {
     throw new ArrayIndexOutOfBoundsException();
   }
   </#list></#list>
   
-  public int size(){
+  public int size() {
     return 0;
   }
   
-  public boolean isSet(){
+  public boolean isSet() {
     return false;
   }
   
-  public boolean next(){
+  public boolean next() {
     return false;
   }
   
-  public RepeatedMapReader map(){
+  public RepeatedMapReader map() {
     return this;
   }
   
-  public RepeatedListReader list(){
+  public RepeatedListReader list() {
     return this;
   }
   
-  public MapReader map(String name){
+  public MapReader map(String name) {
     return this;
   }
   
-  public ListReader list(String name){
+  public ListReader list(String name) {
     return this;
   }
   
-  public FieldReader reader(String name){
+  public FieldReader reader(String name) {
     return this;
   }
   
-  public FieldReader reader(){
+  public FieldReader reader() {
     return this;
   }
   
-  private void fail(String name){
+  private void fail(String name) {
     throw new IllegalArgumentException(String.format("You tried to read a %s type when you are using a ValueReader of type %s.", name, this.getClass().getSimpleName()));
   }
   
@@ -126,11 +128,11 @@ public class NullReader extends AbstractBaseReader implements FieldReader{
   <#assign safeType=friendlyType />
   <#if safeType=="byte[]"><#assign safeType="ByteArray" /></#if>
   
-  public ${friendlyType} read${safeType}(int arrayIndex){
+  public ${friendlyType} read${safeType}(int arrayIndex) {
     return null;
   }
   
-  public ${friendlyType} read${safeType}(){
+  public ${friendlyType} read${safeType}() {
     return null;
   }
   </#list>
diff --git a/exec/vector/src/main/codegen/templates/NullableValueVectors.java b/exec/vector/src/main/codegen/templates/NullableValueVectors.java
index 8be5c8b86..c9c098743 100644
--- a/exec/vector/src/main/codegen/templates/NullableValueVectors.java
+++ b/exec/vector/src/main/codegen/templates/NullableValueVectors.java
@@ -482,7 +482,7 @@ public final class ${className} extends BaseDataValueVector implements <#if type
       vAccessor.get(index, holder);
       holder.isSet = bAccessor.get(index);
 
-      <#if minor.class.startsWith("Decimal")>
+      <#if minor.class.contains("Decimal")>
       holder.scale = getField().getScale();
       holder.precision = getField().getPrecision();
       </#if>
@@ -694,17 +694,29 @@ public final class ${className} extends BaseDataValueVector implements <#if type
     }
 
     </#if>
-    <#if minor.class == "Decimal28Sparse" || minor.class == "Decimal38Sparse">
+    <#if minor.class == "Decimal28Sparse" || minor.class == "Decimal38Sparse" || minor.class == "VarDecimal">
     public void set(int index, BigDecimal value) {
+      <#if type.major == "VarLen">
+      if (index > lastSet + 1) {
+        fillEmpties(index);
+      }
+      </#if>
       bits.getMutator().set(index, 1);
       values.getMutator().set(index, value);
       setCount++;
+      <#if type.major == "VarLen">lastSet = index;</#if>
     }
 
     public void setSafe(int index, BigDecimal value) {
+      <#if type.major == "VarLen">
+      if (index > lastSet + 1) {
+        fillEmpties(index);
+      }
+      </#if>
       bits.getMutator().setSafe(index, 1);
       values.getMutator().setSafe(index, value);
       setCount++;
+      <#if type.major == "VarLen">lastSet = index;</#if>
     }
 
     </#if>
diff --git a/exec/vector/src/main/codegen/templates/RepeatedValueVectors.java b/exec/vector/src/main/codegen/templates/RepeatedValueVectors.java
index 2b2b6bd9a..e9884208f 100644
--- a/exec/vector/src/main/codegen/templates/RepeatedValueVectors.java
+++ b/exec/vector/src/main/codegen/templates/RepeatedValueVectors.java
@@ -404,20 +404,14 @@ 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>) {
+    <#if (fields?size > 1) && !(minor.class == "Decimal9" || minor.class == "Decimal18"
+        || minor.class == "Decimal28Sparse" || minor.class == "Decimal38Sparse" || minor.class == "Decimal28Dense"
+        || minor.class == "Decimal38Dense") || minor.class == "VarDecimal">
+    public void addSafe(int rowIndex<#list fields as field><#if field.include!true>, ${field.type} ${field.name}</#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);
+      values.getMutator().setSafe(nextOffset<#list fields as field><#if field.include!true>, ${field.name}</#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/UnionListWriter.java b/exec/vector/src/main/codegen/templates/UnionListWriter.java
index a0d26a020..4ea907d1b 100644
--- a/exec/vector/src/main/codegen/templates/UnionListWriter.java
+++ b/exec/vector/src/main/codegen/templates/UnionListWriter.java
@@ -81,6 +81,17 @@ public class UnionListWriter extends AbstractFieldWriter {
     return this;
   }
 
+  <#if minor.class == "VarDecimal">
+  @Override
+  public ${name}Writer <#if uncappedName == "int">integer<#else>${uncappedName}</#if>(String name, int scale, int precision) {
+    assert inMap;
+    final int nextOffset = offsets.getAccessor().get(idx() + 1);
+    vector.getMutator().setNotNull(idx());
+    writer.setPosition(nextOffset);
+    ${name}Writer ${uncappedName}Writer = writer.${uncappedName}(name, scale, precision);
+    return ${uncappedName}Writer;
+  }
+  <#else>
   @Override
   public ${name}Writer <#if uncappedName == "int">integer<#else>${uncappedName}</#if>(String name) {
     assert inMap;
@@ -91,6 +102,7 @@ public class UnionListWriter extends AbstractFieldWriter {
     return ${uncappedName}Writer;
   }
   </#if>
+  </#if>
   </#list></#list>
 
   @Override
diff --git a/exec/vector/src/main/codegen/templates/UnionWriter.java b/exec/vector/src/main/codegen/templates/UnionWriter.java
index 961810e3a..13c093acf 100644
--- a/exec/vector/src/main/codegen/templates/UnionWriter.java
+++ b/exec/vector/src/main/codegen/templates/UnionWriter.java
@@ -181,7 +181,21 @@ public class UnionWriter extends AbstractFieldWriter implements FieldWriter {
   <#if lowerName == "int" ><#assign lowerName = "integer" /></#if>
   <#assign upperName = minor.class?upper_case />
   <#assign capName = minor.class?cap_first />
-  <#if !minor.class?starts_with("Decimal")>
+  <#if minor.class == "VarDecimal">
+  @Override
+  public ${capName}Writer ${lowerName}(String name, int scale, int precision) {
+    data.getMutator().setType(idx(), MinorType.MAP);
+    getMapWriter().setPosition(idx());
+    return getMapWriter().${lowerName}(name, scale, precision);
+  }
+
+  @Override
+  public ${capName}Writer ${lowerName}(int scale, int precision) {
+    data.getMutator().setType(idx(), MinorType.LIST);
+    getListWriter().setPosition(idx());
+    return getListWriter().${lowerName}(scale, precision);
+  }
+  <#else>
   @Override
   public ${capName}Writer ${lowerName}(String name) {
     data.getMutator().setType(idx(), MinorType.MAP);
diff --git a/exec/vector/src/main/codegen/templates/ValueHolders.java b/exec/vector/src/main/codegen/templates/ValueHolders.java
index d5b4342df..7635895be 100644
--- a/exec/vector/src/main/codegen/templates/ValueHolders.java
+++ b/exec/vector/src/main/codegen/templates/ValueHolders.java
@@ -32,99 +32,90 @@ package org.apache.drill.exec.expr.holders;
 /*
  * This class is generated using freemarker and the ${.template_name} template.
  */
-public final class ${className} implements ValueHolder{
-  
+<#if minor.class.contains("Decimal") && !minor.class.contains("VarDecimal")>
+/**
+ *  Old decimal types are deprecated. Please use {@link VarDecimalHolder} holder instead.
+ */
+@Deprecated
+</#if>
+public final class ${className} implements ValueHolder {
+
+  <#if minor.class.contains("Decimal")>
+  @Deprecated
+  </#if>
   public static final MajorType TYPE = Types.${mode.name?lower_case}(MinorType.${minor.class?upper_case});
 
-    <#if mode.name == "Repeated">
+  <#if mode.name == "Repeated">
     
-    /** The first index (inclusive) into the Vector. **/
-    public int start;
+  /** The first index (inclusive) into the Vector. **/
+  public int start;
     
-    /** The last index (exclusive) into the Vector. **/
-    public int end;
+  /** The last index (exclusive) into the Vector. **/
+  public int end;
     
-    /** The Vector holding the actual values. **/
-    public ${minor.class}Vector vector;
+  /** The Vector holding the actual values. **/
+  public ${minor.class}Vector vector;
     
-    <#else>
-    public static final int WIDTH = ${type.width};
+  <#else>
+  public static final int WIDTH = ${type.width};
     
-    <#if mode.name == "Optional">public int isSet;</#if>
-    <#assign fields = minor.fields!type.fields />
-    <#list fields as field>
-    public ${field.type} ${field.name};
-    </#list>
+  <#if mode.name == "Optional">public int isSet;</#if>
+  <#assign fields = minor.fields!type.fields />
+  <#list fields as field>
+  public ${field.type} ${field.name};
+  </#list>
     
-    <#if minor.class.startsWith("Decimal")>
-    public static final int maxPrecision = ${minor.maxPrecisionDigits};
-    <#if minor.class.startsWith("Decimal28") || minor.class.startsWith("Decimal38")>
-    public static final int nDecimalDigits = ${minor.nDecimalDigits};
+  <#if minor.class.startsWith("Decimal")>
+  public static final int maxPrecision = ${minor.maxPrecisionDigits};
+  <#if minor.class.startsWith("Decimal28") || minor.class.startsWith("Decimal38")>
+  public static final int nDecimalDigits = ${minor.nDecimalDigits};
     
-    public static int getInteger(int index, int start, DrillBuf buffer) {
-      int value = buffer.getInt(start + (index * 4));
+  public static int getInteger(int index, int start, DrillBuf buffer) {
+    int value = buffer.getInt(start + (index * 4));
 
-      if (index == 0) {
-          /* the first byte contains sign bit, return value without it */
-          <#if minor.class.endsWith("Sparse")>
-          value = (value & 0x7FFFFFFF);
-          <#elseif minor.class.endsWith("Dense")>
-          value = (value & 0x0000007F);
-          </#if>
-      }
-      return value;
+    if (index == 0) {
+      /* the first byte contains sign bit, return value without it */
+      <#if minor.class.endsWith("Sparse")>
+      value = (value & 0x7FFFFFFF);
+      <#elseif minor.class.endsWith("Dense")>
+      value = (value & 0x0000007F);
+      </#if>
     }
+    return value;
+  }
 
-    public static void setInteger(int index, int value, int start, DrillBuf buffer) {
-        buffer.setInt(start + (index * 4), value);
-    }
+  public static void setInteger(int index, int value, int start, DrillBuf buffer) {
+    buffer.setInt(start + (index * 4), value);
+  }
   
-    public static void setSign(boolean sign, int start, DrillBuf buffer) {
-      // Set MSB to 1 if sign is negative
-      if (sign == true) {
-        int value = getInteger(0, start, buffer);
-        setInteger(0, (value | 0x80000000), start, buffer);
-      }
+  public static void setSign(boolean sign, int start, DrillBuf buffer) {
+    // Set MSB to 1 if sign is negative
+    if (sign == true) {
+      int value = getInteger(0, start, buffer);
+      setInteger(0, (value | 0x80000000), start, buffer);
     }
+  }
   
-    public static boolean getSign(int start, DrillBuf buffer) {
-      return ((buffer.getInt(start) & 0x80000000) != 0);
-    }
-
-    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>
+  public static boolean getSign(int start, DrillBuf buffer) {
+    return ((buffer.getInt(start) & 0x80000000) != 0);
+  }
 
-    @Deprecated
-    public int hashCode(){
-      throw new UnsupportedOperationException();
-    }
+  </#if></#if>
+  @Deprecated
+  public int hashCode() {
+    throw new UnsupportedOperationException();
+  }
 
-    /*
-     * Reason for deprecation is that ValueHolders are potential scalar replacements
-     * and hence we don't want any methods to be invoked on them.
-     */
-    @Deprecated
-    public String toString(){
-      throw new UnsupportedOperationException();
-    }
-    </#if>
-    
-    
-    
-    
+  /*
+   * Reason for deprecation is that ValueHolders are potential scalar replacements
+   * and hence we don't want any methods to be invoked on them.
+   */
+  @Deprecated
+  public String toString() {
+    throw new UnsupportedOperationException();
+  }
+  </#if>
 }
-
 </#list>
 </#list>
 </#list>
diff --git a/exec/vector/src/main/codegen/templates/VariableLengthVectors.java b/exec/vector/src/main/codegen/templates/VariableLengthVectors.java
index 8ab4c3a48..6a67772ce 100644
--- a/exec/vector/src/main/codegen/templates/VariableLengthVectors.java
+++ b/exec/vector/src/main/codegen/templates/VariableLengthVectors.java
@@ -464,6 +464,10 @@ public final class ${minor.class}Vector extends BaseDataValueVector implements V
       holder.start = oAccessor.get(index);
       holder.end = oAccessor.get(index + 1);
       holder.buffer = data;
+      <#if minor.class.contains("Decimal")>
+      holder.scale = field.getScale();
+      holder.precision = field.getPrecision();
+      </#if>
     }
 
     public void get(int index, Nullable${minor.class}Holder holder){
@@ -471,6 +475,10 @@ public final class ${minor.class}Vector extends BaseDataValueVector implements V
       holder.start = oAccessor.get(index);
       holder.end = oAccessor.get(index + 1);
       holder.buffer = data;
+      <#if minor.class.contains("Decimal")>
+      holder.scale = field.getScale();
+      holder.precision = field.getPrecision();
+      </#if>
     }
 
     <#switch minor.class>
@@ -478,7 +486,7 @@ public final class ${minor.class}Vector extends BaseDataValueVector implements V
     @Override
     public ${friendlyType} getObject(int index) {
       byte[] b = get(index);
-      BigInteger bi = b.length == 0 ? new BigInteger("0") : new BigInteger(b);
+      BigInteger bi = b.length == 0 ? BigInteger.ZERO : new BigInteger(b);
       BigDecimal bd = new BigDecimal(bi, getField().getScale());
       return bd;
     }
@@ -672,6 +680,18 @@ public final class ${minor.class}Vector extends BaseDataValueVector implements V
       offsetVector.getMutator().setSafe(index + 1, outputStart + len);
     }
 
+    <#if minor.class == "VarDecimal">
+    public void set(int index, BigDecimal value) {
+      byte[] bytes = value.unscaledValue().toByteArray();
+      set(index, bytes, 0, bytes.length);
+    }
+
+    public void setSafe(int index, BigDecimal value) {
+      byte[] bytes = value.unscaledValue().toByteArray();
+      setSafe(index, bytes, 0, bytes.length);
+    }
+    </#if>
+
     public void setSafe(int index, ${minor.class}Holder holder) {
       final int start = holder.start;
       final int end =   holder.end;
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 82809547d..a451b974f 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
@@ -17,20 +17,23 @@
  */
 package org.apache.drill.exec.util;
 
+import com.google.common.math.BigIntegerMath;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.DrillBuf;
 import io.netty.buffer.UnpooledByteBufAllocator;
 
 import java.math.BigDecimal;
 import java.math.BigInteger;
+import java.math.RoundingMode;
 import java.nio.ByteBuffer;
 import java.util.Arrays;
 
+import org.apache.drill.common.types.TypeProtos;
 import org.apache.drill.common.util.CoreDecimalUtility;
 import org.apache.drill.exec.expr.fn.impl.ByteFunctionHelpers;
 import org.apache.drill.exec.expr.holders.Decimal38SparseHolder;
 
-public class DecimalUtility extends CoreDecimalUtility{
+public class DecimalUtility extends CoreDecimalUtility {
 
   public final static int MAX_DIGITS = 9;
   public final static int MAX_DIGITS_INT = 10;
@@ -152,18 +155,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,29 +358,27 @@ 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) {
+      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;
-    }
+  /**
+   * 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);
+    return startIndex + len;
+  }
 
   public static long getDecimal18FromBigDecimal(BigDecimal input, int scale, int precision) {
     // Truncate or pad to set the input to the correct scale
@@ -451,31 +452,52 @@ public class DecimalUtility extends CoreDecimalUtility{
     buffer.setInt(start + (index * 4), value);
   }
 
-    /**
-     * 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);
-    }
+  /**
+   * 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) {
+    byte[] rightBytes = new byte[rightEnd - rightStart];
+    right.getBytes(rightStart, rightBytes, 0, rightEnd - rightStart);
+
+    return compareVarLenBytes(left, leftStart, leftEnd, leftScale, rightBytes, rightScale, absCompare);
+  }
+
+  /**
+   * Compares two VarDecimal values, still stored in Drill buffer and byte array
+   *
+   * @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 byte array
+   * @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, byte right[], int rightScale, boolean absCompare) {
+    java.math.BigDecimal bdLeft = getBigDecimalFromDrillBuf(left, leftStart, leftEnd - leftStart, leftScale);
+    java.math.BigDecimal bdRight = new BigDecimal(new BigInteger(right), 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) {
+  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;
 
@@ -782,5 +804,73 @@ public class DecimalUtility extends CoreDecimalUtility{
     }
     return cmp * invert;
   }
-}
 
+  /**
+   * Returns max length of byte array, required to store value with specified precision.
+   *
+   * @param precision the precision of value
+   *
+   * @return max length of byte array
+   */
+  public static int getMaxBytesSizeForPrecision(int precision) {
+    if (precision == 0) {
+      return 0;
+    }
+    if (precision < 300) { // normal case, use exact heuristic formula
+      return (int) Math.ceil((Math.log(Math.pow(10, precision) - 1) / Math.log(2) + 1) / Byte.SIZE);
+    } else {
+      // for values greater than 304 Math.pow(10, precision) returns Infinity, therefore 1 is neglected in Math.log()
+      return (int) Math.ceil((precision * Math.log(10) / Math.log(2) + 1) / Byte.SIZE);
+    }
+  }
+
+  /**
+   * Calculates and returns square root for specified BigDecimal
+   * with specified number of digits alter decimal point.
+   *
+   * @param in    BigDecimal which square root should be calculated
+   * @param scale number of digits alter decimal point in the result value.
+   * @return square root for specified BigDecimal
+   */
+  public static BigDecimal sqrt(BigDecimal in, int scale) {
+    // unscaled BigInteger value from specified BigDecimal with doubled number of digits after decimal point
+    // was used to calculate sqrt using Guava's BigIntegerMath.
+    BigInteger valueWithDoubleMaxPrecision =
+        in.multiply(BigDecimal.TEN.pow(scale * 2)).setScale(0, RoundingMode.HALF_UP).unscaledValue();
+    return new BigDecimal(
+        BigIntegerMath.sqrt(valueWithDoubleMaxPrecision, RoundingMode.HALF_UP), scale);
+  }
+
+  /**
+   * Checks that specified decimal minorType is obsolete.
+   *
+   * @param minorType type to check
+   * @return true if specified decimal minorType is obsolete.
+   */
+  public static boolean isObsoleteDecimalType(TypeProtos.MinorType minorType) {
+    return minorType == TypeProtos.MinorType.DECIMAL9 ||
+        minorType == TypeProtos.MinorType.DECIMAL18 ||
+        minorType == TypeProtos.MinorType.DECIMAL28SPARSE ||
+        minorType == TypeProtos.MinorType.DECIMAL38SPARSE;
+  }
+
+  /**
+   * Returns default precision for specified {@link TypeProtos.MinorType}
+   * or returns specified defaultPrecision if {@link TypeProtos.MinorType} isn't
+   * {@link TypeProtos.MinorType#INT} or {@link TypeProtos.MinorType#BIGINT}.
+   *
+   * @param minorType        type wich precision should be received
+   * @param defaultPrecision default value for precision
+   * @return default precision for specified {@link TypeProtos.MinorType}
+   */
+  public static int getDefaultPrecision(TypeProtos.MinorType minorType, int defaultPrecision) {
+    switch (minorType) {
+      case INT:
+        return MAX_DIGITS_INT;
+      case BIGINT:
+        return MAX_DIGITS_BIGINT;
+      default:
+        return defaultPrecision;
+    }
+  }
+}
diff --git a/exec/vector/src/main/java/org/apache/drill/exec/vector/ValueHolderHelper.java b/exec/vector/src/main/java/org/apache/drill/exec/vector/ValueHolderHelper.java
index 73b2b551e..d1cc3b4a1 100644
--- a/exec/vector/src/main/java/org/apache/drill/exec/vector/ValueHolderHelper.java
+++ b/exec/vector/src/main/java/org/apache/drill/exec/vector/ValueHolderHelper.java
@@ -37,6 +37,7 @@ import org.apache.drill.exec.expr.holders.NullableBitHolder;
 import org.apache.drill.exec.expr.holders.TimeHolder;
 import org.apache.drill.exec.expr.holders.TimeStampHolder;
 import org.apache.drill.exec.expr.holders.VarCharHolder;
+import org.apache.drill.exec.expr.holders.VarDecimalHolder;
 import org.apache.drill.exec.memory.BufferAllocator;
 import org.apache.drill.exec.util.DecimalUtility;
 
@@ -200,4 +201,22 @@ public class ValueHolderHelper {
 
       return dch;
   }
+
+  public static VarDecimalHolder getVarDecimalHolder(DrillBuf buf, String decimal) {
+    VarDecimalHolder dch = new VarDecimalHolder();
+
+    BigDecimal bigDecimal = new BigDecimal(decimal);
+
+    byte[] bytes = bigDecimal.unscaledValue().toByteArray();
+    int length = bytes.length;
+
+    dch.scale = bigDecimal.scale();
+    dch.precision = bigDecimal.precision();
+    dch.start = 0;
+    dch.end = length;
+    dch.buffer = buf.reallocIfNeeded(length);
+    dch.buffer.setBytes(0, bytes);
+
+    return dch;
+  }
 }
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 1fa785748..937df8603 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
@@ -229,6 +229,11 @@ public class MapOrListWriterImpl implements MapOrListWriter {
   }
 
   @Override
+  public VarDecimalWriter varDecimal(String name, int scale, int precision) {
+    return (map != null) ? map.varDecimal(name, scale, precision) : list.varDecimal(scale, precision);
+  }
+
+  @Override
   public Decimal38SparseWriter decimal38Sparse(String name) {
     return (map != null) ? map.decimal38Sparse(name) : list.decimal38Sparse();
   }