DRILL-5039: NPE - CTAS PARTITION BY (<char-type-column>)

exec/java-exec/src/main/codegen/templates/NewValueFunctions.java

@@ -17,6 +17,12 @@
  */
 <@pp.dropOutputFile />
 
+<#macro reassignHolder>
+        previous.buffer = buf.reallocIfNeeded(length);
+        previous.buffer.setBytes(0, in.buffer, in.start, in.end - in.start);
+        previous.end = in.end - in.start;
+</#macro>
+
 
 <@pp.changeOutputFile name="/org/apache/drill/exec/expr/fn/impl/GNewValueFunctions.java" />
 <#include "/@includes/license.ftl" />
@@ -39,28 +45,51 @@
  */
 public class GNewValueFunctions {
 <#list vv.types as type>
-<#if type.major == "Fixed" || type.major = "Bit">
-
 <#list type.minor as minor>
 <#list vv.modes as mode>
   <#if mode.name != "Repeated">
 
 <#if !minor.class.startsWith("Decimal28") && !minor.class.startsWith("Decimal38") && !minor.class.startsWith("Interval")>
 @SuppressWarnings("unused")
-@FunctionTemplate(name = "newPartitionValue", scope = FunctionTemplate.FunctionScope.SIMPLE, nulls=NullHandling.INTERNAL)
-public static class NewValue${minor.class}${mode.prefix} implements DrillSimpleFunc{
+@FunctionTemplate(name = "newPartitionValue", scope = FunctionTemplate.FunctionScope.SIMPLE, nulls = NullHandling.INTERNAL)
+public static class NewValue${minor.class}${mode.prefix} implements DrillSimpleFunc {
 
   @Param ${mode.prefix}${minor.class}Holder in;
   @Workspace ${mode.prefix}${minor.class}Holder previous;
   @Workspace Boolean initialized;
   @Output BitHolder out;
+  <#if type.major == "VarLen">
+  @Inject DrillBuf buf;
+  </#if>
 
   public void setup() {
     initialized = false;
+    <#if type.major == "VarLen">
+    previous.buffer = buf;
+    previous.start = 0;
+    </#if>
   }
 
-  <#if mode.name == "Required">
   public void eval() {
+  <#if mode.name == "Required">
+  <#if type.major == "VarLen">
+    int length = in.end - in.start;
+
+    if (initialized) {
+      if (org.apache.drill.exec.expr.fn.impl.ByteFunctionHelpers.compare(
+          previous.buffer, 0, previous.end, in.buffer, in.start, in.end) == 0) {
+        out.value = 0;
+      } else {
+        <@reassignHolder/>
+        out.value = 1;
+      }
+    } else {
+      <@reassignHolder/>
+      out.value = 1;
+      initialized = true;
+    }
+  </#if>
+  <#if type.major == "Fixed" || type.major == "Bit">
     if (initialized) {
       if (in.value == previous.value) {
         out.value = 0;
@@ -73,10 +102,36 @@ public void eval() {
       out.value = 1;
       initialized = true;
     }
-  }
   </#if>
+  </#if> <#-- mode.name == "Required" -->
+
   <#if mode.name == "Optional">
-  public void eval() {
+  <#if type.major == "VarLen">
+    int length = in.isSet == 0 ? 0 : in.end - in.start;
+
+    if (initialized) {
+      if (previous.isSet == 0 && in.isSet == 0) {
+        out.value = 0;
+      } else if (previous.isSet != 0 && in.isSet != 0 && org.apache.drill.exec.expr.fn.impl.ByteFunctionHelpers.compare(
+          previous.buffer, 0, previous.end, in.buffer, in.start, in.end) == 0) {
+        out.value = 0;
+      } else {
+        if (in.isSet == 1) {
+          <@reassignHolder/>
+        }
+        previous.isSet = in.isSet;
+        out.value = 1;
+      }
+    } else {
+      if (in.isSet == 1) {
+        <@reassignHolder/>
+      }
+      previous.isSet = in.isSet;
+      out.value = 1;
+      initialized = true;
+    }
+  </#if>
+  <#if type.major == "Fixed" || type.major == "Bit">
     if (initialized) {
       if (in.isSet == 0 && previous.isSet == 0) {
         out.value = 0;
@@ -93,14 +148,14 @@ public void eval() {
       out.value = 1;
       initialized = true;
     }
-  }
   </#if>
+  </#if> <#-- mode.name == "Optional" -->
+  }
 }
 </#if> <#-- minor.class.startWith -->
 
 </#if> <#-- mode.name -->
 </#list>
 </#list>
-</#if> <#-- type.major -->
 </#list>
 }