Arrow batches structured types

src/main/java/net/snowflake/client/core/arrow/fullvectorconverters/ArrowFullVectorConverter.java

@@ -33,6 +33,8 @@ static Types.MinorType deduceType(ValueVector vector) {
             }
             break;
           }
+        case VECTOR:
+          return Types.MinorType.FIXED_SIZE_LIST;
         case TIME:
           return Types.MinorType.TIMEMILLI;
         case TIMESTAMP_LTZ:
@@ -112,6 +114,19 @@ static FieldVector convert(
             return new BigIntVectorConverter(allocator, vector, context, session, idx).convert();
           case DECIMAL:
             return new DecimalVectorConverter(allocator, vector, context, session, idx).convert();
+          case STRUCT:
+            return new StructVectorConverter(allocator, vector, context, session, idx, null)
+                .convert();
+          case LIST:
+            return new ListVectorConverter(allocator, vector, context, session, idx, null)
+                .convert();
+          case VARCHAR:
+            return new VarCharVectorConverter(allocator, vector, context, session, idx).convert();
+          case MAP:
+            return new MapVectorConverter(allocator, vector, context, session, idx, null).convert();
+          case FIXED_SIZE_LIST:
+            return new FixedSizeListVectorConverter(allocator, vector, context, session, idx, null)
+                .convert();
         }
       }
     } catch (SFException ex) {

src/main/java/net/snowflake/client/core/arrow/fullvectorconverters/FixedSizeListVectorConverter.java

@@ -0,0 +1,63 @@
+package net.snowflake.client.core.arrow.fullvectorconverters;
+
+import java.util.ArrayList;
+import net.snowflake.client.core.DataConversionContext;
+import net.snowflake.client.core.SFBaseSession;
+import net.snowflake.client.core.SFException;
+import net.snowflake.client.core.SnowflakeJdbcInternalApi;
+import net.snowflake.client.jdbc.SnowflakeSQLException;
+import org.apache.arrow.memory.ArrowBuf;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.ValueVector;
+import org.apache.arrow.vector.complex.FixedSizeListVector;
+import org.apache.arrow.vector.types.pojo.Field;
+
+@SnowflakeJdbcInternalApi
+public class FixedSizeListVectorConverter implements ArrowFullVectorConverter {
+  protected RootAllocator allocator;
+  protected ValueVector vector;
+  protected DataConversionContext context;
+  protected SFBaseSession session;
+  protected int idx;
+  protected Object valueTargetType;
+
+  FixedSizeListVectorConverter(
+      RootAllocator allocator,
+      ValueVector vector,
+      DataConversionContext context,
+      SFBaseSession session,
+      int idx,
+      Object valueTargetType) {
+    this.allocator = allocator;
+    this.vector = vector;
+    this.context = context;
+    this.session = session;
+    this.idx = idx;
+    this.valueTargetType = valueTargetType;
+  }
+
+  @Override
+  public FieldVector convert() throws SFException, SnowflakeSQLException {
+    FixedSizeListVector listVector = (FixedSizeListVector) vector;
+    FieldVector dataVector = listVector.getDataVector();
+    FieldVector convertedDataVector =
+        ArrowFullVectorConverter.convert(
+            allocator, dataVector, context, session, 0, valueTargetType);
+    FixedSizeListVector convertedListVector =
+        FixedSizeListVector.empty(listVector.getName(), listVector.getListSize(), allocator);
+    ArrayList<Field> fields = new ArrayList<>();
+    fields.add(convertedDataVector.getField());
+    convertedListVector.initializeChildrenFromFields(fields);
+    convertedListVector.allocateNew();
+    convertedListVector.setValueCount(listVector.getValueCount());
+    ArrowBuf validityBuffer = listVector.getValidityBuffer();
+    convertedListVector
+        .getValidityBuffer()
+        .setBytes(0L, validityBuffer, 0L, validityBuffer.capacity());
+    convertedDataVector.makeTransferPair(convertedListVector.getDataVector()).transfer();
+
+    vector.close();
+    return convertedListVector;
+  }
+}

src/main/java/net/snowflake/client/core/arrow/fullvectorconverters/ListVectorConverter.java

@@ -0,0 +1,69 @@
+package net.snowflake.client.core.arrow.fullvectorconverters;
+
+import java.util.ArrayList;
+import net.snowflake.client.core.DataConversionContext;
+import net.snowflake.client.core.SFBaseSession;
+import net.snowflake.client.core.SFException;
+import net.snowflake.client.core.SnowflakeJdbcInternalApi;
+import net.snowflake.client.jdbc.SnowflakeSQLException;
+import org.apache.arrow.memory.ArrowBuf;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.ValueVector;
+import org.apache.arrow.vector.complex.ListVector;
+import org.apache.arrow.vector.types.pojo.Field;
+
+@SnowflakeJdbcInternalApi
+public class ListVectorConverter implements ArrowFullVectorConverter {
+  protected RootAllocator allocator;
+  protected ValueVector vector;
+  protected DataConversionContext context;
+  protected SFBaseSession session;
+  protected int idx;
+  protected Object valueTargetType;
+
+  ListVectorConverter(
+      RootAllocator allocator,
+      ValueVector vector,
+      DataConversionContext context,
+      SFBaseSession session,
+      int idx,
+      Object valueTargetType) {
+    this.allocator = allocator;
+    this.vector = vector;
+    this.context = context;
+    this.session = session;
+    this.idx = idx;
+    this.valueTargetType = valueTargetType;
+  }
+
+  protected ListVector initVector(String name, Field field) {
+    ListVector convertedListVector = ListVector.empty(name, allocator);
+    ArrayList<Field> fields = new ArrayList<>();
+    fields.add(field);
+    convertedListVector.initializeChildrenFromFields(fields);
+    return convertedListVector;
+  }
+
+  @Override
+  public FieldVector convert() throws SFException, SnowflakeSQLException {
+    ListVector listVector = (ListVector) vector;
+    FieldVector dataVector = listVector.getDataVector();
+    FieldVector convertedDataVector =
+        ArrowFullVectorConverter.convert(
+            allocator, dataVector, context, session, 0, valueTargetType);
+    ListVector convertedListVector = initVector(vector.getName(), dataVector.getField());
+    convertedListVector.allocateNew();
+    convertedListVector.setValueCount(listVector.getValueCount());
+    convertedListVector.getOffsetBuffer().setBytes(0, listVector.getOffsetBuffer());
+    ArrowBuf validityBuffer = listVector.getValidityBuffer();
+    convertedListVector
+        .getValidityBuffer()
+        .setBytes(0L, validityBuffer, 0L, validityBuffer.capacity());
+    convertedListVector.setLastSet(listVector.getLastSet());
+    convertedDataVector.makeTransferPair(convertedListVector.getDataVector()).transfer();
+
+    vector.close();
+    return convertedListVector;
+  }
+}

src/main/java/net/snowflake/client/core/arrow/fullvectorconverters/MapVectorConverter.java

@@ -0,0 +1,34 @@
+package net.snowflake.client.core.arrow.fullvectorconverters;
+
+import java.util.ArrayList;
+import net.snowflake.client.core.DataConversionContext;
+import net.snowflake.client.core.SFBaseSession;
+import net.snowflake.client.core.SnowflakeJdbcInternalApi;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.ValueVector;
+import org.apache.arrow.vector.complex.ListVector;
+import org.apache.arrow.vector.complex.MapVector;
+import org.apache.arrow.vector.types.pojo.Field;
+
+@SnowflakeJdbcInternalApi
+public class MapVectorConverter extends ListVectorConverter {
+
+  MapVectorConverter(
+      RootAllocator allocator,
+      ValueVector vector,
+      DataConversionContext context,
+      SFBaseSession session,
+      int idx,
+      Object valueTargetType) {
+    super(allocator, vector, context, session, idx, valueTargetType);
+  }
+
+  @Override
+  protected ListVector initVector(String name, Field field) {
+    MapVector convertedMapVector = MapVector.empty(name, allocator, false);
+    ArrayList<Field> fields = new ArrayList<>();
+    fields.add(field);
+    convertedMapVector.initializeChildrenFromFields(fields);
+    return convertedMapVector;
+  }
+}

src/main/java/net/snowflake/client/core/arrow/fullvectorconverters/StructVectorConverter.java

@@ -0,0 +1,75 @@
+package net.snowflake.client.core.arrow.fullvectorconverters;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import net.snowflake.client.core.DataConversionContext;
+import net.snowflake.client.core.SFBaseSession;
+import net.snowflake.client.core.SFException;
+import net.snowflake.client.core.SnowflakeJdbcInternalApi;
+import net.snowflake.client.jdbc.SnowflakeSQLException;
+import org.apache.arrow.memory.ArrowBuf;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.ValueVector;
+import org.apache.arrow.vector.complex.StructVector;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.util.TransferPair;
+
+@SnowflakeJdbcInternalApi
+public class StructVectorConverter implements ArrowFullVectorConverter {
+  protected RootAllocator allocator;
+  protected ValueVector vector;
+  protected DataConversionContext context;
+  protected SFBaseSession session;
+  protected int idx;
+  protected Map<String, Object> targetTypes;
+
+  StructVectorConverter(
+      RootAllocator allocator,
+      ValueVector vector,
+      DataConversionContext context,
+      SFBaseSession session,
+      int idx,
+      Map<String, Object> targetTypes) {
+    this.allocator = allocator;
+    this.vector = vector;
+    this.context = context;
+    this.session = session;
+    this.idx = idx;
+    this.targetTypes = targetTypes;
+  }
+
+  public FieldVector convert() throws SFException, SnowflakeSQLException {
+    StructVector structVector = (StructVector) vector;
+    List<FieldVector> childVectors = structVector.getChildrenFromFields();
+    List<FieldVector> convertedVectors = new ArrayList<>();
+    for (FieldVector childVector : childVectors) {
+      Object targetType = null;
+      if (targetTypes != null) {
+        targetType = targetTypes.get(childVector.getName());
+      }
+      convertedVectors.add(
+          ArrowFullVectorConverter.convert(
+              allocator, childVector, context, session, idx, targetType));
+    }
+
+    List<Field> convertedFields =
+        convertedVectors.stream().map(ValueVector::getField).collect(Collectors.toList());
+    StructVector converted = StructVector.empty(vector.getName(), allocator);
+    converted.allocateNew();
+    converted.initializeChildrenFromFields(convertedFields);
+    for (FieldVector convertedVector : convertedVectors) {
+      TransferPair transferPair =
+          convertedVector.makeTransferPair(converted.getChild(convertedVector.getName()));
+      transferPair.transfer();
+    }
+    ArrowBuf validityBuffer = structVector.getValidityBuffer();
+    converted.getValidityBuffer().setBytes(0L, validityBuffer, 0L, validityBuffer.capacity());
+    converted.setValueCount(vector.getValueCount());
+
+    vector.close();
+    return converted;
+  }
+}

src/main/java/net/snowflake/client/core/arrow/fullvectorconverters/VarCharVectorConverter.java

@@ -0,0 +1,41 @@
+package net.snowflake.client.core.arrow.fullvectorconverters;
+
+import net.snowflake.client.core.DataConversionContext;
+import net.snowflake.client.core.SFBaseSession;
+import net.snowflake.client.core.SFException;
+import net.snowflake.client.core.SnowflakeJdbcInternalApi;
+import net.snowflake.client.core.arrow.ArrowVectorConverter;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.ValueVector;
+import org.apache.arrow.vector.VarCharVector;
+import org.apache.arrow.vector.util.Text;
+
+@SnowflakeJdbcInternalApi
+public class VarCharVectorConverter extends SimpleArrowFullVectorConverter<VarCharVector> {
+  public VarCharVectorConverter(
+      RootAllocator allocator,
+      ValueVector vector,
+      DataConversionContext context,
+      SFBaseSession session,
+      int idx) {
+    super(allocator, vector, context, session, idx);
+  }
+
+  @Override
+  protected boolean matchingType() {
+    return (vector instanceof VarCharVector);
+  }
+
+  @Override
+  protected VarCharVector initVector() {
+    VarCharVector resultVector = new VarCharVector(vector.getName(), allocator);
+    resultVector.allocateNew(vector.getValueCount());
+    return resultVector;
+  }
+
+  @Override
+  protected void convertValue(ArrowVectorConverter from, VarCharVector to, int idx)
+      throws SFException {
+    to.set(idx, new Text(from.toString(idx)));
+  }
+}