ARROW-6265: [Java] Avro adapter implement Array/Map/Fixed type

java/adapter/avro/src/main/java/org/apache/arrow/consumers/AvroArraysConsumer.java

@@ -0,0 +1,80 @@
+/*
+ * 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.consumers;
+
+import java.io.IOException;
+
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.complex.ListVector;
+import org.apache.avro.io.Decoder;
+
+/**
+ * Consumer which consume array type values from avro decoder.
+ * Write the data to {@link ListVector}.
+ */
+public class AvroArraysConsumer implements Consumer {
+
+  private final ListVector vector;
+  private final Consumer delegate;
+
+  private int currentIndex = 0;
+
+  /**
+   * Instantiate a ArrayConsumer.
+   */
+  public AvroArraysConsumer(ListVector vector, Consumer delegate) {
+    this.vector = vector;
+    this.delegate = delegate;
+  }
+
+  @Override
+  public void consume(Decoder decoder) throws IOException {
+
+    vector.startNewValue(currentIndex);
+    long totalCount = 0;
+    for (long count = decoder.readArrayStart(); count != 0; count = decoder.arrayNext()) {
+      totalCount += count;
+      for (int element = 0; element < count; element++) {
+        delegate.consume(decoder);
+      }
+    }
+    vector.endValue(currentIndex, (int) totalCount);
+    currentIndex++;
+  }
+
+  @Override
+  public void addNull() {
+    currentIndex++;
+  }
+
+  @Override
+  public void setPosition(int index) {
+    currentIndex = index;
+  }
+
+  @Override
+  public FieldVector getVector() {
+    return this.vector;
+  }
+
+  @Override
+  public void close() throws Exception {
+    vector.close();
+    delegate.close();
+  }
+}

java/adapter/avro/src/main/java/org/apache/arrow/consumers/AvroBooleanConsumer.java

@@ -21,8 +21,6 @@
 
 import org.apache.arrow.vector.BitVector;
 import org.apache.arrow.vector.FieldVector;
-import org.apache.arrow.vector.complex.impl.BitWriterImpl;
-import org.apache.arrow.vector.complex.writer.BitWriter;
 import org.apache.avro.io.Decoder;
 
 /**
@@ -31,31 +29,30 @@
  */
 public class AvroBooleanConsumer implements Consumer {
 
-  private final BitWriter writer;
   private final BitVector vector;
+  private int currentIndex = 0;
 
   /**
    * Instantiate a AvroBooleanConsumer.
    */
   public AvroBooleanConsumer(BitVector vector) {
     this.vector = vector;
-    this.writer = new BitWriterImpl(vector);
   }
 
   @Override
   public void consume(Decoder decoder) throws IOException {
-    writer.writeBit(decoder.readBoolean() ? 1 : 0);
-    writer.setPosition(writer.getPosition() + 1);
+    vector.setSafe(currentIndex, decoder.readBoolean() ? 1 : 0);
+    currentIndex++;
   }
 
   @Override
   public void addNull() {
-    writer.setPosition(writer.getPosition() + 1);
+    currentIndex++;
   }
 
   @Override
   public void setPosition(int index) {
-    writer.setPosition(index);
+    currentIndex = index;
   }
 
   @Override
@@ -65,7 +62,7 @@ public FieldVector getVector() {
 
   @Override
   public void close() throws Exception {
-    writer.close();
+    vector.close();
   }
 
 }

java/adapter/avro/src/main/java/org/apache/arrow/consumers/AvroBytesConsumer.java

@@ -22,9 +22,6 @@
 
 import org.apache.arrow.vector.FieldVector;
 import org.apache.arrow.vector.VarBinaryVector;
-import org.apache.arrow.vector.complex.impl.VarBinaryWriterImpl;
-import org.apache.arrow.vector.complex.writer.VarBinaryWriter;
-import org.apache.arrow.vector.holders.VarBinaryHolder;
 import org.apache.avro.io.Decoder;
 
 /**
@@ -33,46 +30,35 @@
  */
 public class AvroBytesConsumer implements Consumer {
 
-  private final VarBinaryWriter writer;
   private final VarBinaryVector vector;
   private ByteBuffer cacheBuffer;
 
+  private int currentIndex;
+
   /**
    * Instantiate a AvroBytesConsumer.
    */
   public AvroBytesConsumer(VarBinaryVector vector) {
     this.vector = vector;
-    this.writer = new VarBinaryWriterImpl(vector);
   }
 
   @Override
   public void consume(Decoder decoder) throws IOException {
-    writeValue(decoder);
-    writer.setPosition(writer.getPosition() + 1);
-  }
-
-  @Override
-  public void addNull() {
-    writer.setPosition(writer.getPosition() + 1);
-  }
-
-  private void writeValue(Decoder decoder) throws IOException {
-    VarBinaryHolder holder = new VarBinaryHolder();
-
     // cacheBuffer is initialized null and create in the first consume,
     // if its capacity < size to read, decoder will create a new one with new capacity.
     cacheBuffer = decoder.readBytes(cacheBuffer);
+    vector.setSafe(currentIndex, cacheBuffer, 0, cacheBuffer.limit());
+    currentIndex++;
+  }
 
-    holder.start = 0;
-    holder.end = cacheBuffer.limit();
-    holder.buffer = vector.getAllocator().buffer(cacheBuffer.limit());
-    holder.buffer.setBytes(0, cacheBuffer, 0,  cacheBuffer.limit());
-    writer.write(holder);
+  @Override
+  public void addNull() {
+    currentIndex++;
   }
 
   @Override
   public void setPosition(int index) {
-    writer.setPosition(index);
+    currentIndex = index;
   }
 
   @Override
@@ -82,6 +68,6 @@ public FieldVector getVector() {
 
   @Override
   public void close() throws Exception {
-    writer.close();
+    vector.close();
   }
 }

java/adapter/avro/src/main/java/org/apache/arrow/consumers/AvroDoubleConsumer.java

@@ -21,8 +21,6 @@
 
 import org.apache.arrow.vector.FieldVector;
 import org.apache.arrow.vector.Float8Vector;
-import org.apache.arrow.vector.complex.impl.Float8WriterImpl;
-import org.apache.arrow.vector.complex.writer.Float8Writer;
 import org.apache.avro.io.Decoder;
 
 /**
@@ -31,31 +29,30 @@
  */
 public class AvroDoubleConsumer implements Consumer {
 
-  private final Float8Writer writer;
   private final Float8Vector vector;
 
+  private int currentIndex;
+
   /**
    * Instantiate a AvroDoubleConsumer.
    */
   public AvroDoubleConsumer(Float8Vector vector) {
     this.vector = vector;
-    this.writer = new Float8WriterImpl(vector);
   }
 
   @Override
   public void consume(Decoder decoder) throws IOException {
-    writer.writeFloat8(decoder.readDouble());
-    writer.setPosition(writer.getPosition() + 1);
+    vector.setSafe(currentIndex++, decoder.readDouble());
   }
 
   @Override
   public void addNull() {
-    writer.setPosition(writer.getPosition() + 1);
+    currentIndex++;
   }
 
   @Override
   public void setPosition(int index) {
-    writer.setPosition(index);
+    currentIndex = index;
   }
 
   @Override
@@ -65,6 +62,6 @@ public FieldVector getVector() {
 
   @Override
   public void close() throws Exception {
-    writer.close();
+    vector.close();
   }
 }

java/adapter/avro/src/main/java/org/apache/arrow/consumers/AvroFixedConsumer.java

@@ -0,0 +1,70 @@
+/*
+ * 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.consumers;
+
+import java.io.IOException;
+
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.FixedSizeBinaryVector;
+import org.apache.avro.io.Decoder;
+
+/**
+ * Consumer which consume fixed type values from avro decoder.
+ * Write the data to {@link org.apache.arrow.vector.FixedSizeBinaryVector}.
+ */
+public class AvroFixedConsumer implements Consumer {
+
+  private final FixedSizeBinaryVector vector;
+  private final byte[] reuseBytes;
+
+  private int currentIndex;
+
+  /**
+   * Instantiate a AvroFixedConsumer.
+   */
+  public AvroFixedConsumer(FixedSizeBinaryVector vector, int size) {
+    this.vector = vector;
+    reuseBytes = new byte[size];
+  }
+
+  @Override
+  public void consume(Decoder decoder) throws IOException {
+    decoder.readFixed(reuseBytes);
+    vector.setSafe(currentIndex++, reuseBytes);
+  }
+
+  @Override
+  public void addNull() {
+    currentIndex++;
+  }
+
+  @Override
+  public void setPosition(int index) {
+    currentIndex = index;
+  }
+
+  @Override
+  public FieldVector getVector() {
+    return vector;
+  }
+
+  @Override
+  public void close() throws Exception {
+    vector.close();
+  }
+}

java/adapter/avro/src/main/java/org/apache/arrow/consumers/AvroFloatConsumer.java

@@ -21,8 +21,6 @@
 
 import org.apache.arrow.vector.FieldVector;
 import org.apache.arrow.vector.Float4Vector;
-import org.apache.arrow.vector.complex.impl.Float4WriterImpl;
-import org.apache.arrow.vector.complex.writer.Float4Writer;
 import org.apache.avro.io.Decoder;
 
 /**
@@ -31,31 +29,30 @@
  */
 public class AvroFloatConsumer implements Consumer {
 
-  private final Float4Writer writer;
   private final Float4Vector vector;
 
+  private int currentIndex;
+
   /**
    * Instantiate a AvroFloatConsumer.
    */
   public AvroFloatConsumer(Float4Vector vector) {
     this.vector = vector;
-    this.writer = new Float4WriterImpl(vector);
   }
 
   @Override
   public void consume(Decoder decoder) throws IOException {
-    writer.writeFloat4(decoder.readFloat());
-    writer.setPosition(writer.getPosition() + 1);
+    vector.setSafe(currentIndex++, decoder.readFloat());
   }
 
   @Override
   public void addNull() {
-    writer.setPosition(writer.getPosition() + 1);
+    currentIndex++;
   }
 
   @Override
   public void setPosition(int index) {
-    writer.setPosition(index);
+    currentIndex = index;
   }
 
   @Override
@@ -65,6 +62,6 @@ public FieldVector getVector() {
 
   @Override
   public void close() throws Exception {
-    writer.close();
+    vector.close();
   }
 }