Support for @inject in aggregation functions

7 files changed, 257 insertions, 70 deletions

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 05f5cb085..54cec1000 100644
--- a/exec/java-exec/src/main/codegen/templates/Decimal/DecimalAggrTypeFunctions1.java
+++ b/exec/java-exec/src/main/codegen/templates/Decimal/DecimalAggrTypeFunctions1.java
@@ -66,6 +66,8 @@ public static class ${type.inputType}${aggrtype.className} implements DrillAggFu
   @Workspace IntHolder outputScale;
   <#elseif type.outputType.endsWith("Sparse")>
   @Inject DrillBuf buffer;
+  @Workspace IntHolder scale;
+  @Workspace IntHolder precision;
   @Workspace ObjectHolder value;
   <#else>
   @Workspace ${type.runningType}Holder value;
@@ -78,24 +80,24 @@ public static class ${type.inputType}${aggrtype.className} implements DrillAggFu
     value.value = 0;
 	<#elseif aggrtype.funcName == "max" || aggrtype.funcName == "min">
     <#if type.outputType.endsWith("Sparse")>
+    scale.value = 0;
+    precision.value = 0;
     value = new ObjectHolder();
-    ${type.runningType}Holder tmp = new ${type.runningType}Holder();
+    //${type.runningType}Holder tmp = new ${type.runningType}Holder();
+    byte[] byteArray = new byte[${type.runningType}Holder.WIDTH];
+    org.apache.hadoop.io.Text tmp = new org.apache.hadoop.io.Text(byteArray);
     value.obj = tmp;
-    buffer = buffer.reallocIfNeeded(tmp.WIDTH);
-    tmp.buffer = buffer;
-    tmp.start  = 0;
     <#if aggrtype.funcName == "max">
-    for (int i = 0; i < tmp.nDecimalDigits; i++) {
-      tmp.setInteger(i, 0xFFFFFFFF, tmp.start, tmp.buffer);
+    for (int i = 0; i < ${type.runningType}Holder.nDecimalDigits; i++) {
+      org.apache.drill.exec.expr.fn.impl.ByteFunctionHelpers.setInteger(tmp.getBytes(), i, 0xFFFFFFFF);
     }
-    tmp.setSign(true, tmp.start, tmp.buffer);
+    org.apache.drill.exec.expr.fn.impl.ByteFunctionHelpers.setSign(tmp.getBytes(), true);
     <#elseif aggrtype.funcName == "min">
-    for (int i = 0; i < tmp.nDecimalDigits; i++) {
-      tmp.setInteger(i, 0x7FFFFFFF, tmp.start, tmp.buffer);
+    for (int i = 0; i < ${type.runningType}Holder.nDecimalDigits; i++) {
+      org.apache.drill.exec.expr.fn.impl.ByteFunctionHelpers.setInteger(tmp.getBytes(), i, 0x7FFFFFFF);
     }
     // Set sign to be positive so initial value is maximum
-    tmp.setSign(false, tmp.start, tmp.buffer);
-    tmp.precision = ${type.runningType}Holder.maxPrecision;
+    org.apache.drill.exec.expr.fn.impl.ByteFunctionHelpers.setSign(tmp.getBytes(), false);
     </#if>
     <#elseif type.outputType == "Decimal9" || type.outputType == "Decimal18">
     value = new ${type.runningType}Holder();
@@ -124,32 +126,34 @@ public static class ${type.inputType}${aggrtype.className} implements DrillAggFu
     value.value++;
     <#elseif aggrtype.funcName == "max">
     <#if type.outputType.endsWith("Sparse")>
-      ${type.runningType}Holder tmp = (${type.runningType}Holder) value.obj;
-      int cmp = org.apache.drill.exec.util.DecimalUtility.compareSparseBytes(in.buffer, in.start, in.getSign(in.start, in.buffer),
-      in.scale, in.precision, tmp.buffer,
-      tmp.start, tmp.getSign(tmp.start, tmp.buffer), tmp.precision,
-      tmp.scale, in.WIDTH, in.nDecimalDigits, false);
+      //${type.runningType}Holder tmp = (${type.runningType}Holder) value.obj;
+      org.apache.hadoop.io.Text tmp = (org.apache.hadoop.io.Text) value.obj;
+      int cmp = org.apache.drill.exec.util.DecimalUtility.compareSparseSamePrecScale(in.buffer, in.start, tmp.getBytes(), tmp.getLength());
     if (cmp == 1) {
-      in.buffer.getBytes(in.start, tmp.buffer, 0, tmp.WIDTH);
-      tmp.setSign(in.getSign(in.start, in.buffer), tmp.start, tmp.buffer);
-      tmp.scale = in.scale;
-      tmp.precision = in.precision;
+      //in.buffer.getBytes(in.start, tmp.getBytes(), 0, ${type.runningType}Holder.WIDTH);
+      for (int i = 0; i < ${type.runningType}Holder.nDecimalDigits; i++) {
+        org.apache.drill.exec.expr.fn.impl.ByteFunctionHelpers.setInteger(tmp.getBytes(), i, ${type.runningType}Holder.getInteger(i, in.start, in.buffer));
+      }
+      org.apache.drill.exec.expr.fn.impl.ByteFunctionHelpers.setSign(tmp.getBytes(), in.getSign(in.start, in.buffer));
+      scale.value = in.scale;
+      precision.value = in.precision;
     }
     <#elseif type.outputType == "Decimal9" || type.outputType == "Decimal18">
     value.value = Math.max(value.value, in.value);
     </#if>
     <#elseif aggrtype.funcName == "min">
     <#if type.outputType.endsWith("Sparse")>
-    ${type.runningType}Holder tmp = (${type.runningType}Holder) value.obj;
-    int cmp = org.apache.drill.exec.util.DecimalUtility.compareSparseBytes(in.buffer, in.start, in.getSign(in.start, in.buffer),
-      in.scale, in.precision, tmp.buffer,
-      tmp.start, tmp.getSign(tmp.start, tmp.buffer), tmp.precision,
-      tmp.scale, in.WIDTH, in.nDecimalDigits, false);
+    //${type.runningType}Holder tmp = (${type.runningType}Holder) value.obj;
+    org.apache.hadoop.io.Text tmp = (org.apache.hadoop.io.Text) value.obj;
+    int cmp = org.apache.drill.exec.util.DecimalUtility.compareSparseSamePrecScale(in.buffer, in.start, tmp.getBytes(), tmp.getLength());
     if (cmp == -1) {
-      in.buffer.getBytes(in.start, tmp.buffer, 0, tmp.WIDTH);
-      tmp.setSign(in.getSign(in.start, in.buffer), tmp.start, tmp.buffer);
-      tmp.scale = in.scale;
-      tmp.precision = in.precision;
+      //in.buffer.getBytes(in.start, tmp.getBytes(), 0, ${type.runningType}Holder.WIDTH);
+      for (int i = 0; i < ${type.runningType}Holder.nDecimalDigits; i++) {
+        org.apache.drill.exec.expr.fn.impl.ByteFunctionHelpers.setInteger(tmp.getBytes(), i, ${type.runningType}Holder.getInteger(i, in.start, in.buffer));
+      }
+      org.apache.drill.exec.expr.fn.impl.ByteFunctionHelpers.setSign(tmp.getBytes(), in.getSign(in.start, in.buffer));
+      scale.value = in.scale;
+      precision.value = in.precision;
     }
     <#elseif type.outputType == "Decimal9" || type.outputType == "Decimal18">
     value.value = Math.min(value.value, in.value);
@@ -183,12 +187,17 @@ public static class ${type.inputType}${aggrtype.className} implements DrillAggFu
     org.apache.drill.exec.util.DecimalUtility.getSparseFromBigDecimal((java.math.BigDecimal) value.obj, out.buffer, out.start, out.scale, out.precision, out.nDecimalDigits);
    <#else>
     <#if type.outputType.endsWith("Sparse")>
-    ${type.runningType}Holder tmp = (${type.runningType}Holder) value.obj;
-    out.buffer = tmp.buffer;
-    out.start = tmp.start;
-    out.setSign(tmp.getSign(tmp.start, tmp.buffer), out.start, out.buffer);
-    out.scale = tmp.scale;
-    out.precision = tmp.precision;
+    org.apache.hadoop.io.Text tmp = (org.apache.hadoop.io.Text) value.obj;
+    buffer = buffer.reallocIfNeeded(tmp.getLength());
+    //buffer.setBytes(0, tmp.getBytes(), 0, tmp.getLength());
+    for (int i = 0; i < ${type.runningType}Holder.nDecimalDigits; i++) {
+      ${type.runningType}Holder.setInteger(i, org.apache.drill.exec.expr.fn.impl.ByteFunctionHelpers.getInteger(tmp.getBytes(), i), 0, buffer);
+    }
+    out.buffer = buffer;
+    out.start = 0;
+    out.setSign(org.apache.drill.exec.expr.fn.impl.ByteFunctionHelpers.getSign(tmp.getBytes()), out.start, out.buffer);
+    out.scale = scale.value;
+    out.precision = precision.value;
     <#elseif type.outputType == "Decimal9" || type.outputType == "Decimal18">
     out.value = value.value;
     out.scale = value.scale;
@@ -204,17 +213,17 @@ public static class ${type.inputType}${aggrtype.className} implements DrillAggFu
 	  value.value = 0;
 	<#elseif aggrtype.funcName == "max" || aggrtype.funcName == "min">
     <#if type.outputType.endsWith("Sparse")>
-  	value = new ObjectHolder();
-    ${type.runningType}Holder tmp = new ${type.runningType}Holder();
-    value.obj = tmp;
-    for (int i = 0; i < tmp.nDecimalDigits; i++) {
-      tmp.setInteger(i, 0xFFFFFFFF, tmp.start, tmp.buffer);
+    org.apache.hadoop.io.Text tmp = (org.apache.hadoop.io.Text) value.obj;
+    for (int i = 0; i < ${type.runningType}Holder.nDecimalDigits; i++) {
+      org.apache.drill.exec.expr.fn.impl.ByteFunctionHelpers.setInteger(tmp.getBytes(), i, 0xFFFFFFFF);
     }
+    scale.value = 0;
+    precision.value = 0;
     <#if aggrtype.funcName == "min">
     // Set sign to be positive so initial value is maximum
-    tmp.setSign(false, tmp.start, tmp.buffer);
+    org.apache.drill.exec.expr.fn.impl.ByteFunctionHelpers.setSign(tmp.getBytes(), false);
     <#elseif aggrtype.funcName == "max">
-    tmp.setSign(true, tmp.start, tmp.buffer);
+    org.apache.drill.exec.expr.fn.impl.ByteFunctionHelpers.setSign(tmp.getBytes(), true);
     </#if>
     <#elseif type.outputType == "Decimal9" || type.outputType == "Decimal18">
     value = new ${type.runningType}Holder();
diff --git a/exec/java-exec/src/main/codegen/templates/VarCharAggrFunctions1.java b/exec/java-exec/src/main/codegen/templates/VarCharAggrFunctions1.java
index 4b9c43bd3..ea5662400 100644
--- a/exec/java-exec/src/main/codegen/templates/VarCharAggrFunctions1.java
+++ b/exec/java-exec/src/main/codegen/templates/VarCharAggrFunctions1.java
@@ -74,10 +74,7 @@ public static class ${type.inputType}${aggrtype.className} implements DrillAggFu
     init = new UInt1Holder();
     init.value = 0;
     value = new ObjectHolder();
-    ${type.runningType}Holder tmp = new ${type.runningType}Holder();
-    tmp.start = 0;
-    tmp.end = 0;
-    tmp.buffer = null;
+    org.apache.drill.exec.expr.fn.impl.DrillByteArray tmp = new org.apache.drill.exec.expr.fn.impl.DrillByteArray();
     value.obj = tmp;
 
     <#else>
@@ -96,7 +93,7 @@ public static class ${type.inputType}${aggrtype.className} implements DrillAggFu
       }
     </#if>
     <#if aggrtype.funcName == "max" || aggrtype.funcName == "min">
-    ${type.runningType}Holder tmp = (${type.runningType}Holder)value.obj;
+    org.apache.drill.exec.expr.fn.impl.DrillByteArray tmp = (org.apache.drill.exec.expr.fn.impl.DrillByteArray) value.obj;
     int cmp = 0;
     boolean swap = false;
 
@@ -106,7 +103,8 @@ public static class ${type.inputType}${aggrtype.className} implements DrillAggFu
       swap = true;
     } else {
       // Compare the bytes
-      int compare = org.apache.drill.exec.expr.fn.impl.ByteFunctionHelpers.compare(in.buffer.memoryAddress(), in.start, in.end, tmp.buffer.memoryAddress(), tmp.start, tmp.end);
+      cmp = org.apache.drill.exec.expr.fn.impl.ByteFunctionHelpers.compare(in.buffer.memoryAddress(), in.start, in.end, tmp.getBytes(), 0, tmp.getLength());
+
 
       <#if aggrtype.className == "Min">
       swap = (cmp == -1);
@@ -115,11 +113,15 @@ public static class ${type.inputType}${aggrtype.className} implements DrillAggFu
       </#if>
     }
     if (swap) {
-      int length = in.end - in.start;
-      this.buf = tmp.buffer = buf.reallocIfNeeded(in.end - in.start);
-      in.buffer.getBytes(in.start, tmp.buffer, 0, length);
-      tmp.end = length;
-      
+      int inputLength = in.end - in.start;
+      if (tmp.getLength() >= inputLength) {
+        in.buffer.getBytes(in.start, tmp.getBytes(), 0, inputLength);
+        tmp.setLength(inputLength);
+      } else {
+        byte[] tempArray = new byte[in.end - in.start];
+        in.buffer.getBytes(in.start, tempArray, 0, in.end - in.start);
+        tmp.setBytes(tempArray);
+      }
     }
     <#else>
     value.value++;
@@ -132,10 +134,12 @@ public static class ${type.inputType}${aggrtype.className} implements DrillAggFu
   @Override
   public void output() {
     <#if aggrtype.funcName == "max" || aggrtype.funcName == "min">
-    ${type.runningType}Holder tmp = (${type.runningType}Holder)value.obj;
-    out.start  = tmp.start;
-    out.end    = tmp.end;
-    out.buffer = tmp.buffer;
+    org.apache.drill.exec.expr.fn.impl.DrillByteArray tmp = (org.apache.drill.exec.expr.fn.impl.DrillByteArray) value.obj;
+    buf = buf.reallocIfNeeded(tmp.getLength());
+    buf.setBytes(0, tmp.getBytes(), 0, tmp.getLength());
+    out.start  = 0;
+    out.end    = tmp.getLength();
+    out.buffer = buf;
     <#else>
     out.value = value.value;
     </#if>
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/DrillAggFuncHolder.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/DrillAggFuncHolder.java
index 5281c37a7..c60d4c9a9 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/DrillAggFuncHolder.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/DrillAggFuncHolder.java
@@ -98,11 +98,15 @@ class DrillAggFuncHolder extends DrillFuncHolder{
         JVar sizeVar = setupBlock.decl(g.getModel().INT, "vectorSize", JExpr.lit(Integer.MAX_VALUE));
         JClass mathClass = g.getModel().ref(Math.class);
         for (int id = 0; id<workspaceVars.length; id ++) {
-          setupBlock.assign(sizeVar,mathClass.staticInvoke("min").arg(sizeVar).arg(g.getWorkspaceVectors().get(workspaceVars[id]).invoke("getValueCapacity")));
+          if (!workspaceVars[id].isInject()) {
+            setupBlock.assign(sizeVar,mathClass.staticInvoke("min").arg(sizeVar).arg(g.getWorkspaceVectors().get(workspaceVars[id]).invoke("getValueCapacity")));
+          }
         }
 
         for(int i =0 ; i < workspaceVars.length; i++) {
-          setupBlock.assign(workspaceJVars[i], JExpr._new(g.getHolderType(workspaceVars[i].majorType)));
+          if (!workspaceVars[i].isInject()) {
+            setupBlock.assign(workspaceJVars[i], JExpr._new(g.getHolderType(workspaceVars[i].majorType)));
+          }
         }
 
         //Use for loop to initialize entries in the workspace vectors.
@@ -148,18 +152,23 @@ class DrillAggFuncHolder extends DrillFuncHolder{
     JVar[] workspaceJVars = new JVar[workspaceVars.length];
 
     for(int i =0 ; i < workspaceVars.length; i++){
-      Preconditions.checkState(Types.isFixedWidthType(workspaceVars[i].majorType), String.format("Workspace variable '%s' in aggregation function '%s' is not allowed to have variable length type.", workspaceVars[i].name, registeredNames[0]));
-      Preconditions.checkState(workspaceVars[i].majorType.getMode()==DataMode.REQUIRED, String.format("Workspace variable '%s' in aggregation function '%s' is not allowed to have null or repeated type.", workspaceVars[i].name, registeredNames[0]));
+      if (workspaceVars[i].isInject() == true) {
+        workspaceJVars[i] = g.declareClassField("work", g.getModel()._ref(workspaceVars[i].type));
+        g.getBlock(BlockType.SETUP).assign(workspaceJVars[i], g.getMappingSet().getIncoming().invoke("getContext").invoke("getManagedBuffer"));
+      } else {
+        Preconditions.checkState(Types.isFixedWidthType(workspaceVars[i].majorType), String.format("Workspace variable '%s' in aggregation function '%s' is not allowed to have variable length type.", workspaceVars[i].name, registeredNames[0]));
+        Preconditions.checkState(workspaceVars[i].majorType.getMode()==DataMode.REQUIRED, String.format("Workspace variable '%s' in aggregation function '%s' is not allowed to have null or repeated type.", workspaceVars[i].name, registeredNames[0]));
 
-      //workspaceJVars[i] = g.declareClassField("work", g.getHolderType(workspaceVars[i].majorType), JExpr._new(g.getHolderType(workspaceVars[i].majorType)));
-      workspaceJVars[i] = g.declareClassField("work", g.getHolderType(workspaceVars[i].majorType));
+        //workspaceJVars[i] = g.declareClassField("work", g.getHolderType(workspaceVars[i].majorType), JExpr._new(g.getHolderType(workspaceVars[i].majorType)));
+        workspaceJVars[i] = g.declareClassField("work", g.getHolderType(workspaceVars[i].majorType));
 
-      //Declare a workspace vector for the workspace var.
-      TypedFieldId typedFieldId = new TypedFieldId(workspaceVars[i].majorType, g.getWorkspaceTypes().size());
-      JVar vv  = g.declareVectorValueSetupAndMember(g.getMappingSet().getWorkspace(), typedFieldId);
+        //Declare a workspace vector for the workspace var.
+        TypedFieldId typedFieldId = new TypedFieldId(workspaceVars[i].majorType, g.getWorkspaceTypes().size());
+        JVar vv  = g.declareVectorValueSetupAndMember(g.getMappingSet().getWorkspace(), typedFieldId);
 
-      g.getWorkspaceTypes().add(typedFieldId);
-      g.getWorkspaceVectors().put(workspaceVars[i], vv);
+        g.getWorkspaceTypes().add(typedFieldId);
+        g.getWorkspaceVectors().put(workspaceVars[i], vv);
+      }
     }
     return workspaceJVars;
   }
@@ -200,6 +209,11 @@ class DrillAggFuncHolder extends DrillFuncHolder{
 
     JVar[] internalVars = new JVar[workspaceJVars.length];
     for(int i =0; i < workspaceJVars.length; i++){
+
+      if (workspaceVars[i].isInject()) {
+        internalVars[i] = sub.decl(g.getModel()._ref(workspaceVars[i].type), workspaceVars[i].name, workspaceJVars[i]);
+        continue;
+      }
       //sub.assign(workspaceJVars[i], JExpr._new(g.getHolderType(workspaceVars[i].majorType)));
       //Access workspaceVar through workspace vector.
       JInvocation getValueAccessor = g.getWorkspaceVectors().get(workspaceVars[i]).invoke("getAccessor").invoke("get");
@@ -219,6 +233,10 @@ class DrillAggFuncHolder extends DrillFuncHolder{
     for(int i =0; i < workspaceJVars.length; i++){
       sub.assign(workspaceJVars[i], internalVars[i]);
 
+      // Injected buffers are not stored as vectors skip storing them in vectors
+      if (workspaceVars[i].isInject()) {
+        continue;
+      }
       //Change workspaceVar through workspace vector.
       JInvocation setMeth;
       if (Types.usesHolderForGet(workspaceVars[i].majorType)) {
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/ByteFunctionHelpers.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/ByteFunctionHelpers.java
index be1d9e6fc..b5241395e 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/ByteFunctionHelpers.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/ByteFunctionHelpers.java
@@ -21,6 +21,7 @@ package org.apache.drill.exec.expr.fn.impl;
 import com.google.common.primitives.UnsignedLongs;
 
 import io.netty.util.internal.PlatformDependent;
+import org.apache.drill.exec.util.DecimalUtility;
 
 public class ByteFunctionHelpers {
   static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ByteFunctionHelpers.class);
@@ -88,9 +89,73 @@ public class ByteFunctionHelpers {
 
     if (lLen == rLen) return 0;
 
-    return lLen > rLen ? 1 : 0;
+    return lLen > rLen ? 1 : -1;
 
   }
 
+  public static final int compare(final long laddr, int lStart, int lEnd, final byte[] right, int rStart, final int rEnd) {
+    int lLen = lEnd - lStart;
+    int rLen = rEnd - rStart;
+    int n = Math.min(rLen, lLen);
+    long lPos = laddr + lStart;
+    int rPos = rStart;
+
+
+
+    while (n-- != 0) {
+      byte leftByte = PlatformDependent.getByte(lPos);
+      byte rightByte = right[rPos];
+      if (leftByte != rightByte) {
+        return ((leftByte & 0xFF) - (rightByte & 0xFF)) > 0 ? 1 : -1;
+      }
+      lPos++;
+      rPos++;
+    }
+
+    if (lLen == rLen) return 0;
+
+    return lLen > rLen ? 1 : -1;
+
+  }
+  // Get the big endian integer
+  public static int getInteger(byte[] b, int index) {
+    int startIndex = index * DecimalUtility.integerSize;
+
+    if (index == 0) {
+      return (b[startIndex + 3] & 0xFF) |
+             (b[startIndex + 2] & 0xFF) << 8 |
+             (b[startIndex + 1] & 0xFF) << 16 |
+             (b[startIndex] & 0x7F) << 24;
+    }
+
+    return ((b[startIndex + 3] & 0xFF) |
+        (b[startIndex + 2] & 0xFF) << 8 |
+        (b[startIndex + 1] & 0xFF) << 16 |
+        (b[startIndex] & 0xFF) << 24);
+
+    }
+
+  // Set the big endian bytes for the input integer
+  public static void setInteger(byte[] b, int index, int value) {
+    int startIndex = index * DecimalUtility.integerSize;
+    b[startIndex] = (byte) ((value >> 24) & 0xFF);
+    b[startIndex + 1] = (byte) ((value >> 16) & 0xFF);
+    b[startIndex + 2] = (byte) ((value >> 8) & 0xFF);
+    b[startIndex + 3] = (byte) ((value) & 0xFF);
+  }
+
+  // Set the sign in a sparse decimal representation
+  public static void setSign(byte[] b, boolean sign) {
+    int value = getInteger(b, 0);
+    if (sign == true) {
+      setInteger(b, 0, value | 0x80000000);
+    } else {
+      setInteger(b, 0, value & 0x7FFFFFFF);
+    }
+  }
 
+  // Get the sign
+  public static boolean getSign(byte[] b) {
+    return ((b[0] & 0x80) > 0);
+  }
 }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/DrillByteArray.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/DrillByteArray.java
new file mode 100644
index 000000000..507db6746
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/DrillByteArray.java
@@ -0,0 +1,60 @@
+/**
+ * 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.expr.fn.impl;
+
+import io.netty.buffer.ByteBuf;
+
+public class DrillByteArray {
+  private byte[] bytes;
+  private int length;
+
+  public DrillByteArray() {
+    this.bytes = new byte[0];
+    this.length = 0;
+  }
+
+  public DrillByteArray(byte[] bytes, int length) {
+    this.bytes = bytes;
+    this.length = length;
+  }
+
+  public DrillByteArray(byte[] bytes) {
+    this(bytes, bytes.length);
+  }
+
+  public void setLength(int length) {
+    this.length = length;
+  }
+
+  public byte[] getBytes() {
+    return this.bytes;
+  }
+
+  public int getLength() {
+    return this.length;
+  }
+
+  public void setBytes(byte[] bytes) {
+    setBytes(bytes, bytes.length);
+  }
+
+  public void setBytes(byte[] bytes, int length) {
+    this.bytes = bytes;
+    this.length = length;
+  }
+}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/StringFunctionHelpers.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/StringFunctionHelpers.java
index d429a2800..b514adb1f 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/StringFunctionHelpers.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/StringFunctionHelpers.java
@@ -33,7 +33,7 @@ public class StringFunctionHelpers {
   public static void initCap(int start, int end, DrillBuf inBuf, DrillBuf outBuf) {
     boolean capNext = true;
     int out = 0;
-    for (int id = start; id < end; id++; out++) {
+    for (int id = start; id < end; id++, out++) {
       byte currentByte = inBuf.getByte(id);
 
       // 'A - Z' : 0x41 - 0x5A
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/util/DecimalUtility.java b/exec/java-exec/src/main/java/org/apache/drill/exec/util/DecimalUtility.java
index 8e752eb56..3d0a9de9c 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/util/DecimalUtility.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/util/DecimalUtility.java
@@ -25,6 +25,8 @@ 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{
 
@@ -688,5 +690,34 @@ public class DecimalUtility extends CoreDecimalUtility{
     int index = nDecimalDigits - roundUp(scale);
     return (int) (adjustScaleDivide(data.getInt(start + (index * integerSize)), MAX_DIGITS - 1));
   }
+
+  public static int compareSparseSamePrecScale(DrillBuf left, int lStart, byte[] right, int length) {
+    // check the sign first
+    boolean lSign = (left.getInt(lStart) & 0x80000000) != 0;
+    boolean rSign = ByteFunctionHelpers.getSign(right);
+    int cmp = 0;
+
+    if (lSign != rSign) {
+      return (lSign == false) ? 1 : -1;
+    }
+
+    // invert the comparison if we are comparing negative numbers
+    int invert = (lSign == true) ? -1 : 1;
+
+    // compare byte by byte
+    int n = 0;
+    int lPos = lStart;
+    int rPos = 0;
+    while (n < length/4) {
+      int leftInt = Decimal38SparseHolder.getInteger(n, lStart, left);
+      int rightInt = ByteFunctionHelpers.getInteger(right, n);
+      if (leftInt != rightInt) {
+        cmp =  (leftInt - rightInt ) > 0 ? 1 : -1;
+        break;
+      }
+      n++;
+    }
+    return cmp * invert;
+  }
 }