GH-615: Produce Avro core data types out of Arrow VSR

adapter/avro/src/main/java/module-info.java

@@ -18,6 +18,8 @@
 module org.apache.arrow.adapter.avro {
   exports org.apache.arrow.adapter.avro.consumers;
   exports org.apache.arrow.adapter.avro.consumers.logical;
+  exports org.apache.arrow.adapter.avro.producers;
+  exports org.apache.arrow.adapter.avro.producers.logical;
   exports org.apache.arrow.adapter.avro;
 
   requires org.apache.arrow.memory.core;

adapter/avro/src/main/java/org/apache/arrow/adapter/avro/producers/AvroBigIntProducer.java

@@ -0,0 +1,52 @@
+/*
+ * 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.adapter.avro.producers;
+
+import java.io.IOException;
+import org.apache.arrow.vector.BaseFixedWidthVector;
+import org.apache.arrow.vector.BigIntVector;
+import org.apache.avro.io.Encoder;
+
+/**
+ * Producer that produces long values from a {@link BigIntVector}, writes data to an Avro encoder.
+ *
+ * <p>Logical types are also supported, for vectors derived from {@link BaseFixedWidthVector} where
+ * the internal representation matches BigIntVector and requires no conversion.
+ */
+public class AvroBigIntProducer extends BaseAvroProducer<BaseFixedWidthVector> {
+
+  /** Instantiate an AvroBigIntProducer. */
+  public AvroBigIntProducer(BigIntVector vector) {
+    super(vector);
+  }
+
+  /** Protected constructor for logical types with a long representation. */
+  protected AvroBigIntProducer(BaseFixedWidthVector vector) {
+    super(vector);
+    if (vector.getTypeWidth() != BigIntVector.TYPE_WIDTH) {
+      throw new IllegalArgumentException(
+          "AvroBigIntProducer requires type width = " + BigIntVector.TYPE_WIDTH);
+    }
+  }
+
+  @Override
+  public void produce(Encoder encoder) throws IOException {
+    long value = vector.getDataBuffer().getLong(currentIndex * (long) BigIntVector.TYPE_WIDTH);
+    encoder.writeLong(value);
+    currentIndex++;
+  }
+}

adapter/avro/src/main/java/org/apache/arrow/adapter/avro/producers/AvroBooleanProducer.java

@@ -0,0 +1,38 @@
+/*
+ * 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.adapter.avro.producers;
+
+import java.io.IOException;
+import org.apache.arrow.vector.BitVector;
+import org.apache.avro.io.Encoder;
+
+/**
+ * Producer that produces boolean values from a {@link BitVector}, writes data to an Avro encoder.
+ */
+public class AvroBooleanProducer extends BaseAvroProducer<BitVector> {
+
+  /** Instantiate am AvroBooleanProducer. */
+  public AvroBooleanProducer(BitVector vector) {
+    super(vector);
+  }
+
+  @Override
+  public void produce(Encoder encoder) throws IOException {
+    int bitValue = vector.get(currentIndex++);
+    encoder.writeBoolean(bitValue != 0);
+  }
+}

adapter/avro/src/main/java/org/apache/arrow/adapter/avro/producers/AvroBytesProducer.java

@@ -0,0 +1,45 @@
+/*
+ * 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.adapter.avro.producers;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import org.apache.arrow.vector.VarBinaryVector;
+import org.apache.avro.io.Encoder;
+
+/**
+ * Producer that produces byte array values from a {@link VarBinaryVector}, writes data to an Avro
+ * encoder.
+ */
+public class AvroBytesProducer extends BaseAvroProducer<VarBinaryVector> {
+
+  /** Instantiate an AvroBytesProducer. */
+  public AvroBytesProducer(VarBinaryVector vector) {
+    super(vector);
+  }
+
+  @Override
+  public void produce(Encoder encoder) throws IOException {
+    // The nio ByteBuffer is created once per call, but underlying data is not copied
+    long offset = vector.getStartOffset(currentIndex);
+    long endOffset = vector.getEndOffset(currentIndex);
+    int length = (int) (endOffset - offset);
+    ByteBuffer nioBuffer = vector.getDataBuffer().nioBuffer(offset, length);
+    encoder.writeBytes(nioBuffer);
+    currentIndex++;
+  }
+}

adapter/avro/src/main/java/org/apache/arrow/adapter/avro/producers/AvroEnumProducer.java

@@ -0,0 +1,38 @@
+/*
+ * 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.adapter.avro.producers;
+
+import java.io.IOException;
+import org.apache.arrow.vector.IntVector;
+import org.apache.avro.io.Encoder;
+
+/**
+ * Producer that produces enum values from a dictionary-encoded {@link IntVector}, writes data to an
+ * Avro encoder.
+ */
+public class AvroEnumProducer extends BaseAvroProducer<IntVector> {
+
+  /** Instantiate an AvroEnumProducer. */
+  public AvroEnumProducer(IntVector vector) {
+    super(vector);
+  }
+
+  @Override
+  public void produce(Encoder encoder) throws IOException {
+    encoder.writeEnum(vector.get(currentIndex++));
+  }
+}

adapter/avro/src/main/java/org/apache/arrow/adapter/avro/producers/AvroFixedSizeBinaryProducer.java

@@ -0,0 +1,54 @@
+/*
+ * 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.adapter.avro.producers;
+
+import java.io.IOException;
+import org.apache.arrow.vector.BaseFixedWidthVector;
+import org.apache.arrow.vector.FixedSizeBinaryVector;
+import org.apache.avro.io.Encoder;
+
+/**
+ * Producer that produces fixed-size binary values from a {@link FixedSizeBinaryVector}, writes data
+ * to an Avro encoder.
+ *
+ * <p>Logical types are also supported, for vectors derived from {@link BaseFixedWidthVector} where
+ * the internal representation is fixed width bytes and requires no conversion.
+ */
+public class AvroFixedSizeBinaryProducer extends BaseAvroProducer<BaseFixedWidthVector> {
+
+  private final byte[] reuseBytes;
+
+  /** Instantiate an AvroFixedSizeBinaryProducer. */
+  public AvroFixedSizeBinaryProducer(FixedSizeBinaryVector vector) {
+    super(vector);
+    reuseBytes = new byte[vector.getTypeWidth()];
+  }
+
+  /** Protected constructor for logical types with a fixed width representation. */
+  protected AvroFixedSizeBinaryProducer(BaseFixedWidthVector vector) {
+    super(vector);
+    reuseBytes = new byte[vector.getTypeWidth()];
+  }
+
+  @Override
+  public void produce(Encoder encoder) throws IOException {
+    long offset = (long) currentIndex * vector.getTypeWidth();
+    vector.getDataBuffer().getBytes(offset, reuseBytes);
+    encoder.writeFixed(reuseBytes);
+    currentIndex++;
+  }
+}

adapter/avro/src/main/java/org/apache/arrow/adapter/avro/producers/AvroFixedSizeListProducer.java

@@ -0,0 +1,78 @@
+/*
+ * 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.adapter.avro.producers;
+
+import java.io.IOException;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.complex.FixedSizeListVector;
+import org.apache.avro.io.Encoder;
+
+/**
+ * Producer that produces array values from a {@link FixedSizeListVector}, writes data to an avro
+ * encoder.
+ */
+public class AvroFixedSizeListProducer extends BaseAvroProducer<FixedSizeListVector> {
+
+  private final Producer<? extends FieldVector> delegate;
+
+  /** Instantiate an AvroFixedSizeListProducer. */
+  public AvroFixedSizeListProducer(
+      FixedSizeListVector vector, Producer<? extends FieldVector> delegate) {
+    super(vector);
+    this.delegate = delegate;
+  }
+
+  @Override
+  public void produce(Encoder encoder) throws IOException {
+
+    encoder.writeArrayStart();
+    encoder.setItemCount(vector.getListSize());
+
+    for (int i = 0; i < vector.getListSize(); i++) {
+      encoder.startItem();
+      delegate.produce(encoder);
+    }
+
+    encoder.writeArrayEnd();
+    currentIndex++;
+  }
+
+  @Override
+  public void skipNull() {
+    super.skipNull();
+    // Child vector contains a fixed number of elements for each entry
+    int childIndex = currentIndex * vector.getListSize();
+    delegate.setPosition(childIndex);
+  }
+
+  @Override
+  public void setPosition(int index) {
+    if (index < 0 || index > vector.getValueCount()) {
+      throw new IllegalArgumentException("Index out of bounds");
+    }
+    super.setPosition(index);
+    // Child vector contains a fixed number of elements for each entry
+    int childIndex = currentIndex * vector.getListSize();
+    delegate.setPosition(childIndex);
+  }
+
+  @Override
+  @SuppressWarnings("unchecked")
+  public void resetValueVector(FixedSizeListVector vector) {
+    ((Producer<FieldVector>) delegate).resetValueVector(vector.getDataVector());
+  }
+}

adapter/avro/src/main/java/org/apache/arrow/adapter/avro/producers/AvroFloat2Producer.java

@@ -0,0 +1,40 @@
+/*
+ * 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.adapter.avro.producers;
+
+import java.io.IOException;
+import org.apache.arrow.memory.util.Float16;
+import org.apache.arrow.vector.Float2Vector;
+import org.apache.avro.io.Encoder;
+
+/**
+ * Producer that produces float values from a {@link Float2Vector}, writes data to an Avro encoder.
+ */
+public class AvroFloat2Producer extends BaseAvroProducer<Float2Vector> {
+
+  /** Instantiate an AvroFloat2Producer. */
+  public AvroFloat2Producer(Float2Vector vector) {
+    super(vector);
+  }
+
+  @Override
+  public void produce(Encoder encoder) throws IOException {
+    short rawValue = vector.getDataBuffer().getShort(currentIndex * (long) Float2Vector.TYPE_WIDTH);
+    encoder.writeFloat(Float16.toFloat(rawValue));
+    currentIndex++;
+  }
+}