DRILL-3229: Implement Union type vector

common/src/main/java/org/apache/drill/common/expression/fn/CastFunctions.java

@@ -36,6 +36,7 @@ public class CastFunctions {
   private static Map<String, String> CAST_FUNC_REPLACEMENT_FROM_NULLABLE = new HashMap<>();
 
   static {
+    TYPE2FUNC.put(MinorType.UNION, "castUNION");
     TYPE2FUNC.put(MinorType.BIGINT, "castBIGINT");
     TYPE2FUNC.put(MinorType.INT, "castINT");
     TYPE2FUNC.put(MinorType.BIT, "castBIT");

common/src/main/java/org/apache/drill/common/types/Types.java

@@ -143,6 +143,7 @@ public static String getSqlTypeName(final MajorType type) {
       case MAP:             return "MAP";
       case LATE:            return "ANY";
       case NULL:            return "NULL";
+      case UNION:           return "UNION";
 
       // Internal types not actually used at level of SQL types(?):
 
@@ -228,6 +229,8 @@ public static int getJdbcTypeCode(final MajorType type) {
       return java.sql.Types.VARBINARY;
     case VARCHAR:
       return java.sql.Types.VARCHAR;
+    case UNION:
+      return java.sql.Types.OTHER;
     default:
       // TODO:  This isn't really an unsupported-operation/-type case; this
       //   is an unexpected, code-out-of-sync-with-itself case, so use an
@@ -290,6 +293,7 @@ public static boolean isJdbcSignedType( final MajorType type ) {
           case LATE:
           case LIST:
           case MAP:
+          case UNION:
           case NULL:
           case TIMETZ:      // SQL TIME WITH TIME ZONE
           case TIMESTAMPTZ: // SQL TIMESTAMP WITH TIME ZONE
@@ -340,6 +344,7 @@ public static boolean isFixedWidthType(final MajorType type) {
     case VARBINARY:
     case VAR16CHAR:
     case VARCHAR:
+    case UNION:
       return false;
     default:
       return true;

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

@@ -67,6 +67,10 @@ public void copyAsField(String name, ListWriter writer){
   <#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){
+    fail("${name}");
+  }
+
   public void read(Nullable${name}Holder holder){
     fail("${name}");
   }

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

@@ -45,6 +45,16 @@ public void end() {
     throw new IllegalStateException(String.format("You tried to end when you are using a ValueWriter of type %s.", this.getClass().getSimpleName()));
   }
 
+  @Override
+  public void startList() {
+    throw new IllegalStateException(String.format("You tried to start when you are using a ValueWriter of type %s.", this.getClass().getSimpleName()));
+  }
+
+  @Override
+  public void endList() {
+    throw new IllegalStateException(String.format("You tried to end when you are using a ValueWriter of type %s.", this.getClass().getSimpleName()));
+  }
+
   <#list vv.types as type><#list type.minor as minor><#assign name = minor.class?cap_first />
   <#assign fields = minor.fields!type.fields />
   @Override

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

@@ -54,6 +54,11 @@ public FieldConverter getNewMapConverter(int fieldId, String fieldName, FieldRea
     throw new UnsupportedOperationException("Doesn't support writing Map'");
   }
 
+  @Override
+  public FieldConverter getNewUnionConverter(int fieldId, String fieldName, FieldReader reader) {
+    throw new UnsupportedOperationException("Doesn't support writing Union type'");
+  }
+
   @Override
   public FieldConverter getNewRepeatedMapConverter(int fieldId, String fieldName, FieldReader reader) {
     throw new UnsupportedOperationException("Doesn't support writing RepeatedMap");

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

@@ -33,6 +33,9 @@ public interface BaseReader extends Positionable{
   MajorType getType();
   MaterializedField getField();
   void reset();
+  void read(UnionHolder holder);
+  void read(int index, UnionHolder holder);
+  void copyAsValue(UnionWriter writer);
 
   public interface MapReader extends BaseReader, Iterable<String>{
     FieldReader reader(String name);

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

@@ -57,8 +57,8 @@ public interface MapWriter extends BaseWriter {
   }
 
   public interface ListWriter extends BaseWriter {
-    void start();
-    void end();
+    void startList();
+    void endList();
     MapWriter map();
     ListWriter list();
     void copyReader(FieldReader reader);

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

@@ -0,0 +1,134 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+<@pp.dropOutputFile />
+<@pp.changeOutputFile name="/org/apache/drill/exec/vector/complex/impl/ComplexCopier.java" />
+
+
+<#include "/@includes/license.ftl" />
+
+package org.apache.drill.exec.vector.complex.impl;
+
+<#include "/@includes/vv_imports.ftl" />
+
+/*
+ * This class is generated using freemarker and the ${.template_name} template.
+ */
+@SuppressWarnings("unused")
+public class ComplexCopier {
+
+  FieldReader in;
+  FieldWriter out;
+
+  public ComplexCopier(FieldReader in, FieldWriter out) {
+    this.in = in;
+    this.out = out;
+  }
+
+  public void write() {
+    writeValue(in, out);
+  }
+
+  private void writeValue(FieldReader reader, FieldWriter writer) {
+    final DataMode m = reader.getType().getMode();
+    final MinorType mt = reader.getType().getMinorType();
+
+    switch(m){
+    case OPTIONAL:
+    case REQUIRED:
+
+
+      switch (mt) {
+
+      case LIST:
+        writer.startList();
+        while (reader.next()) {
+          writeValue(reader.reader(), getListWriterForReader(reader.reader(), writer));
+        }
+        writer.endList();
+        break;
+      case MAP:
+        writer.start();
+        if (reader.isSet()) {
+          for(String name : reader){
+            FieldReader childReader = reader.reader(name);
+            if(childReader.isSet()){
+              writeValue(childReader, getMapWriterForReader(childReader, writer, name));
+            }
+          }
+        }
+        writer.end();
+        break;
+  <#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")>
+
+      case ${name?upper_case}:
+        if (reader.isSet()) {
+          Nullable${name}Holder ${uncappedName}Holder = new Nullable${name}Holder();
+          reader.read(${uncappedName}Holder);
+          writer.write${name}(<#list fields as field>${uncappedName}Holder.${field.name}<#if field_has_next>, </#if></#list>);
+        }
+        break;
+
+  </#if>
+  </#list></#list>
+      }
+              break;
+    }
+ }
+
+  private FieldWriter getMapWriterForReader(FieldReader reader, MapWriter writer, String name) {
+    switch (reader.getType().getMinorType()) {
+    <#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")>
+    case ${name?upper_case}:
+      return (FieldWriter) writer.<#if name == "Int">integer<#else>${uncappedName}</#if>(name);
+    </#if>
+    </#list></#list>
+    case MAP:
+      return (FieldWriter) writer.map(name);
+    case LIST:
+      return (FieldWriter) writer.list(name);
+    default:
+      throw new UnsupportedOperationException(reader.getType().toString());
+    }
+  }
+
+  private FieldWriter getListWriterForReader(FieldReader reader, ListWriter writer) {
+    switch (reader.getType().getMinorType()) {
+    <#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")>
+    case ${name?upper_case}:
+    return (FieldWriter) writer.<#if name == "Int">integer<#else>${uncappedName}</#if>();
+    </#if>
+    </#list></#list>
+    case MAP:
+      return (FieldWriter) writer.map();
+    case LIST:
+      return (FieldWriter) writer.list();
+    default:
+      throw new UnsupportedOperationException(reader.getType().toString());
+    }
+  }
+}

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

@@ -119,7 +119,13 @@ public void copyAsField(String name, MapWriter writer){
     ${nullMode}${minor.class?cap_first}WriterImpl impl = (${nullMode}${minor.class?cap_first}WriterImpl) writer.${lowerName}(name);
     impl.vector.copyFromSafe(idx(), impl.idx(), vector);
   }
-
+
+  <#if nullMode != "Nullable">
+  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){
     vector.getAccessor().get(idx(), h);
   }
@@ -157,6 +163,7 @@ public interface ${name}Reader extends BaseReader{
   public Object readObject(int arrayIndex);
   public ${friendlyType} read${safeType}(int arrayIndex);
   <#else>
+  public void read(${minor.class?cap_first}Holder h);
   public void read(Nullable${minor.class?cap_first}Holder h);
   public Object readObject();
   public ${friendlyType} read${safeType}();

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

@@ -31,6 +31,7 @@
 import org.apache.drill.exec.planner.physical.WriterPrel;
 import org.apache.drill.exec.record.VectorAccessible;
 import org.apache.drill.exec.record.VectorWrapper;
+import org.apache.drill.exec.vector.complex.impl.UnionReader;
 import org.apache.drill.exec.vector.complex.reader.FieldReader;
 
 import java.io.IOException;
@@ -118,6 +119,9 @@ public void endField() throws IOException {
   }
 
   public static FieldConverter getConverter(RecordWriter recordWriter, int fieldId, String fieldName, FieldReader reader) {
+    if (reader instanceof UnionReader) {
+      return recordWriter.getNewUnionConverter(fieldId, fieldName, reader);
+    }
     switch (reader.getType().getMinorType()) {
       case MAP:
         switch (reader.getType().getMode()) {

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

@@ -30,6 +30,7 @@
 <#assign friendlyType = (minor.friendlyType!minor.boxedType!type.boxedType) />
 <#assign safeType=friendlyType />
 <#if safeType=="byte[]"><#assign safeType="ByteArray" /></#if>
+<#assign fields = minor.fields!type.fields />
 
 <@pp.changeOutputFile name="/org/apache/drill/exec/vector/complex/impl/${holderMode}${name}HolderReaderImpl.java" />
 <#include "/@includes/license.ftl" />
@@ -116,6 +117,22 @@ public boolean isSet() {
 
   }
 
+<#if holderMode != "Repeated">
+@Override
+  public void read(${name}Holder h) {
+  <#list fields as field>
+    h.${field.name} = holder.${field.name};
+  </#list>
+  }
+
+  @Override
+  public void read(Nullable${name}Holder h) {
+  <#list fields as field>
+    h.${field.name} = holder.${field.name};
+  </#list>
+  }
+</#if>
+
 <#if holderMode == "Repeated">
   @Override
   public ${friendlyType} read${safeType}(int index){
@@ -262,6 +279,11 @@ private Object readSingleObject() {
 </#if>
   }
 
+<#if holderMode != "Repeated" && nullMode != "Nullable">
+  public void copyAsValue(${minor.class?cap_first}Writer writer){
+    writer.write(holder);
+  }
+</#if>
 }
 
 </#list>

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

@@ -181,7 +181,7 @@ public MaterializedField getField() {
 
   <#if mode == "Repeated">
 
-  public void start() {
+  public void startList() {
     final RepeatedListVector list = (RepeatedListVector) container;
     final RepeatedListVector.RepeatedMutator mutator = list.getMutator();
 
@@ -202,7 +202,7 @@ public void start() {
     }
   }
 
-  public void end() {
+  public void endList() {
     // noop, we initialize state at start rather than end.
   }
   <#else>
@@ -214,11 +214,11 @@ public void setPosition(int index) {
     }
   }
 
-  public void start() {
+  public void startList() {
     // noop
   }
 
-  public void end() {
+  public void endList() {
     // noop
   }
   </#if>