apacheGH-725: Added ExtensionReader (apache#726)

## What's Changed
ExtensionReader was added to support reading extension types from a
complex vector.
It contains **read(ExtensionHolder)** method for reading to the holder.
And **readObject** - for reading the value explicitly.

Closes apache#725.

(cherry picked from commit e6da71e871a3678a737a88dbe79491b4111496b4)

Author: xxlaykxx

java/vector/src/main/codegen/templates/AbstractFieldReader.java

@@ -108,6 +108,23 @@ public void copyAsField(String name, ${name}Writer writer) {
   }
 
   </#list></#list>
+
+  public void read(ExtensionHolder holder) {
+    fail("Extension");
+  }
+
+  public void read(int arrayIndex, ExtensionHolder holder) {
+    fail("RepeatedExtension");
+  }
+
+  public void copyAsValue(AbstractExtensionTypeWriter writer) {
+    fail("CopyAsValueExtension");
+  }
+
+  public void copyAsField(String name, AbstractExtensionTypeWriter writer) {
+    fail("CopyAsFieldExtension");
+  }
+
   public FieldReader reader(String name) {
     fail("reader(String name)");
     return null;

java/vector/src/main/codegen/templates/BaseReader.java

@@ -44,17 +44,17 @@ public interface BaseReader extends Positionable{
   public interface StructReader extends BaseReader, Iterable<String>{
     FieldReader reader(String name);
   }
-  
+
   public interface RepeatedStructReader extends StructReader{
     boolean next();
     int size();
     void copyAsValue(StructWriter writer);
   }
-  
+
   public interface ListReader extends BaseReader{
-    FieldReader reader(); 
+    FieldReader reader();
   }
-  
+
   public interface RepeatedListReader extends ListReader{
     boolean next();
     int size();
@@ -70,16 +70,15 @@ public interface RepeatedMapReader extends MapReader{
     int size();
     void copyAsValue(MapWriter writer);
   }
-  
-  public interface ScalarReader extends  
-  <#list vv.types as type><#list type.minor as minor><#assign name = minor.class?cap_first /> ${name}Reader, </#list></#list> 
+
+  public interface ScalarReader extends
+  <#list vv.types as type><#list type.minor as minor><#assign name = minor.class?cap_first /> ${name}Reader, </#list></#list>
   BaseReader {}
-  
+
   interface ComplexReader{
     StructReader rootAsStruct();
     ListReader rootAsList();
     boolean rootIsStruct();
     boolean ok();
   }
 }
-

java/vector/src/main/codegen/templates/NullReader.java

@@ -33,12 +33,12 @@
  */
 @SuppressWarnings("unused")
 public class NullReader extends AbstractBaseReader implements FieldReader{
-  
+
   public static final NullReader INSTANCE = new NullReader();
   public static final NullReader EMPTY_LIST_INSTANCE = new NullReader(MinorType.NULL);
   public static final NullReader EMPTY_STRUCT_INSTANCE = new NullReader(MinorType.STRUCT);
   private MinorType type;
-  
+
   private NullReader(){
     super();
     type = MinorType.NULL;
@@ -77,71 +77,72 @@ public void read(Nullable${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 read(int arrayIndex, Nullable${name}Holder holder){
     throw new ArrayIndexOutOfBoundsException();
   }
   </#list></#list>
-  
+
+  public void read(ExtensionHolder holder) {
+    holder.isSet = 0;
+  }
+
   public int size(){
     return 0;
   }
-  
+
   public boolean isSet(){
     return false;
   }
-  
+
   public boolean next(){
     return false;
   }
-  
+
   public RepeatedStructReader struct(){
     return this;
   }
-  
+
   public RepeatedListReader list(){
     return this;
   }
-  
+
   public StructReader struct(String name){
     return this;
   }
-  
+
   public ListReader list(String name){
     return this;
   }
-  
+
   public FieldReader reader(String name){
     return this;
   }
-  
+
   public FieldReader reader(){
     return this;
   }
-  
+
   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()));
   }
-  
+
   <#list ["Object", "BigDecimal", "Short", "Integer", "Long", "Boolean",
           "LocalDateTime", "Duration", "Period", "Double", "Float",
           "Character", "Text", "String", "Byte", "byte[]", "PeriodDuration"] as friendlyType>
   <#assign safeType=friendlyType />
   <#if safeType=="byte[]"><#assign safeType="ByteArray" /></#if>
-  
+
   public ${friendlyType} read${safeType}(int arrayIndex){
     return null;
   }
-  
+
   public ${friendlyType} read${safeType}(){
     return null;
   }
   </#list>
-  
-}
-
-
 
+}

java/vector/src/main/codegen/templates/PromotableWriter.java

@@ -550,6 +550,10 @@ public void addExtensionTypeWriterFactory(ExtensionTypeWriterFactory factory) {
     getWriter(MinorType.EXTENSIONTYPE).addExtensionTypeWriterFactory(factory);
   }
 
+  public void addExtensionTypeWriterFactory(ExtensionTypeWriterFactory factory, ArrowType arrowType) {
+    getWriter(MinorType.EXTENSIONTYPE, arrowType).addExtensionTypeWriterFactory(factory);
+  }
+
   @Override
   public void allocate() {
     getWriter().allocate();

java/vector/src/main/codegen/templates/UnionListWriter.java

@@ -53,6 +53,7 @@ public class Union${listName}Writer extends AbstractFieldWriter {
   private boolean inStruct = false;
   private boolean listStarted = false;
   private String structName;
+  private ArrowType extensionType;
   <#if listName == "LargeList" || listName == "LargeListView">
   private static final long OFFSET_WIDTH = 8;
   <#else>
@@ -203,9 +204,9 @@ public MapWriter map(String name, boolean keysSorted) {
 
   @Override
   public ExtensionWriter extension(ArrowType arrowType) {
+    this.extensionType = arrowType;
     return this;
   }
-  
   @Override
   public ExtensionWriter extension(String name, ArrowType arrowType) {
     ExtensionWriter extensionWriter = writer.extension(name, arrowType);
@@ -342,12 +343,11 @@ public void writeExtension(Object value) {
   
   @Override
   public void addExtensionTypeWriterFactory(ExtensionTypeWriterFactory var1) {
-    writer.addExtensionTypeWriterFactory(var1);
+    writer.addExtensionTypeWriterFactory(var1, extensionType);
   }
   
   public void write(ExtensionHolder var1) {
     writer.write(var1);
-    writer.setPosition(writer.idx() + 1);
   }
 
   <#list vv.types as type>

java/vector/src/main/java/org/apache/arrow/vector/complex/reader/ExtensionReader.java

@@ -0,0 +1,44 @@
+/*
+ * 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.arrow.vector.complex.reader;
+
+import org.apache.arrow.vector.holders.ExtensionHolder;
+
+/** Interface for reading extension types. Extends the functionality of {@link BaseReader}. */
+public interface ExtensionReader extends BaseReader {
+
+  /**
+   * Reads to the given extension holder.
+   *
+   * @param holder the {@link ExtensionHolder} to read
+   */
+  void read(ExtensionHolder holder);
+
+  /**
+   * Reads and returns an object representation of the extension type.
+   *
+   * @return the object representation of the extension type
+   */
+  Object readObject();
+
+  /**
+   * Checks if the current value is set.
+   *
+   * @return true if the value is set, false otherwise
+   */
+  boolean isSet();
+}

java/vector/src/test/java/org/apache/arrow/vector/TestListVector.java

@@ -23,16 +23,22 @@
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
+import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.UUID;
 import org.apache.arrow.memory.ArrowBuf;
 import org.apache.arrow.memory.BufferAllocator;
 import org.apache.arrow.util.AutoCloseables;
 import org.apache.arrow.vector.complex.BaseRepeatedValueVector;
 import org.apache.arrow.vector.complex.ListVector;
+import org.apache.arrow.vector.complex.impl.UnionListReader;
 import org.apache.arrow.vector.complex.impl.UnionListWriter;
+import org.apache.arrow.vector.complex.impl.UuidWriterFactory;
 import org.apache.arrow.vector.complex.reader.FieldReader;
+import org.apache.arrow.vector.complex.writer.BaseWriter.ExtensionWriter;
+import org.apache.arrow.vector.holder.UuidHolder;
 import org.apache.arrow.vector.holders.DurationHolder;
 import org.apache.arrow.vector.holders.FixedSizeBinaryHolder;
 import org.apache.arrow.vector.holders.TimeStampMilliTZHolder;
@@ -41,6 +47,7 @@
 import org.apache.arrow.vector.types.pojo.ArrowType;
 import org.apache.arrow.vector.types.pojo.Field;
 import org.apache.arrow.vector.types.pojo.FieldType;
+import org.apache.arrow.vector.types.pojo.UuidType;
 import org.apache.arrow.vector.util.TransferPair;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
@@ -1198,6 +1205,71 @@ public void testGetTransferPairWithField() {
     }
   }
 
+  @Test
+  public void testListVectorWithExtensionType() throws Exception {
+    final FieldType type = FieldType.nullable(new UuidType());
+    try (final ListVector inVector = new ListVector("list", allocator, type, null)) {
+      UnionListWriter writer = inVector.getWriter();
+      writer.allocate();
+      writer.setPosition(0);
+      UUID u1 = UUID.randomUUID();
+      UUID u2 = UUID.randomUUID();
+      writer.startList();
+      ExtensionWriter extensionWriter = writer.extension(new UuidType());
+      extensionWriter.addExtensionTypeWriterFactory(new UuidWriterFactory());
+      extensionWriter.writeExtension(u1);
+      extensionWriter.writeExtension(u2);
+      writer.endList();
+
+      writer.setValueCount(1);
+
+      FieldReader reader = inVector.getReader();
+      assertTrue(reader.isSet(), "shouldn't be null");
+      Object result = inVector.getObject(0);
+      ArrayList<UUID> resultSet = (ArrayList<UUID>) result;
+      assertEquals(2, resultSet.size());
+      assertEquals(u1, resultSet.get(0));
+      assertEquals(u2, resultSet.get(1));
+    }
+  }
+
+  @Test
+  public void testListVectorReaderForExtensionType() throws Exception {
+    final FieldType type = FieldType.nullable(new UuidType());
+    try (final ListVector inVector = new ListVector("list", allocator, type, null)) {
+      UnionListWriter writer = inVector.getWriter();
+      writer.allocate();
+      writer.setPosition(0);
+      UUID u1 = UUID.randomUUID();
+      UUID u2 = UUID.randomUUID();
+      writer.startList();
+      ExtensionWriter extensionWriter = writer.extension(new UuidType());
+      extensionWriter.addExtensionTypeWriterFactory(new UuidWriterFactory());
+      extensionWriter.writeExtension(u1);
+      extensionWriter.writeExtension(u2);
+      writer.endList();
+
+      writer.setValueCount(1);
+
+      UnionListReader reader = inVector.getReader();
+      assertTrue(reader.isSet(), "shouldn't be null");
+      reader.setPosition(0);
+      reader.next();
+      FieldReader uuidReader = reader.reader();
+      UuidHolder holder = new UuidHolder();
+      uuidReader.read(holder);
+      ByteBuffer bb = ByteBuffer.wrap(holder.value);
+      UUID actualUuid = new UUID(bb.getLong(), bb.getLong());
+      assertEquals(u1, actualUuid);
+      reader.next();
+      uuidReader = reader.reader();
+      uuidReader.read(holder);
+      bb = ByteBuffer.wrap(holder.value);
+      actualUuid = new UUID(bb.getLong(), bb.getLong());
+      assertEquals(u2, actualUuid);
+    }
+  }
+
   private void writeIntValues(UnionListWriter writer, int[] values) {
     writer.startList();
     for (int v : values) {

java/vector/src/test/java/org/apache/arrow/vector/UuidVector.java

@@ -20,6 +20,9 @@
 import java.util.UUID;
 import org.apache.arrow.memory.BufferAllocator;
 import org.apache.arrow.memory.util.hash.ArrowBufHasher;
+import org.apache.arrow.vector.complex.impl.UuidReaderImpl;
+import org.apache.arrow.vector.complex.reader.FieldReader;
+import org.apache.arrow.vector.holder.UuidHolder;
 import org.apache.arrow.vector.types.pojo.Field;
 import org.apache.arrow.vector.types.pojo.FieldType;
 import org.apache.arrow.vector.types.pojo.UuidType;
@@ -79,11 +82,21 @@ public TransferPair makeTransferPair(ValueVector to) {
     return new TransferImpl((UuidVector) to);
   }
 
+  @Override
+  protected FieldReader getReaderImpl() {
+    return new UuidReaderImpl(this);
+  }
+
   public void setSafe(int index, byte[] value) {
     getUnderlyingVector().setIndexDefined(index);
     getUnderlyingVector().setSafe(index, value);
   }
 
+  public void get(int index, UuidHolder holder) {
+    holder.value = getUnderlyingVector().get(index);
+    holder.isSet = 1;
+  }
+
   public class TransferImpl implements TransferPair {
     UuidVector to;
     ValueVector targetUnderlyingVector;