GH-615: Produce Avro core data types out of Arrow VSR (#638)

## What's Changed

Per discussion in #615 , here is a first take on the core producers to
generate Avro data from Arrow vectors. There are a few points I'd like
to clarify before going further:

* Nullability. Avro only understands nullable types as unions but that
is not normally how they will be if the data comes from other sources. I
have added a special NullableProducer to handle nullable vectors which
are not unions. We will need something equivalent in the consumers and
probably a setting in the AvroToArrowConfig to control it on read,
defaulting to current behaviour. I have also added special handling for
nullable unions, because unions can't be nested (i.e. you can't nest
"type | null" as a type inside a union). I can add consumers to handle
both (unions and regular types) for review, if that sounds right? At the
moment the schema for nullable fields gets quite mangled on a round
trip!

* Arrow has a lot more types than Avro, at the level of minor / vector
types. Going Avro -> Arrow we just pick the direct equivalent. Going
Arrow -> Avro, we could cast silently if there is no loss of precision.
E.g. TinyInt and SmallInt -> Int and so on. For types like e.g.
Decimal256 and LargeVarChar we could write out safely but would need
support in the consumers to read back the wider types. I could start by
adding the safe conversions now and we could come back to the wide types
in a later PR maybe?

* I have made Producer inherit AutoClosable the same as Consumer does.
Not sure if that is always what you want though - it will free the
buffers in the VSR, but you might want to keep the VSR after you are
finished with IO. Do we need something like detachValueVector() to go
with resetValueVector()? Calling close() after detach would not affect
the vector.

* Type information is inferred from the list of vectors, using minor
types. We'll also need to generate the Avro schema, the input for that
would be a list of fields. I haven't done it yet but will do if that
sounds right.

* Dictionary encoding for enums not implemented yet, I'll add it if the
rest looks good. Caveat is that dictionaries must be fixed before the
encoding starts if we are writing out the whole file in one go (i.e. if
the Avro schema is at the start of the container file). If the schema is
saved separately that limitation need not apply, we could provide the
schema once encoding is finished.

Please do let me know if this is going in the right direction + any
comments. If it is I will add the missing pieces and start the
exhaustive test coverage to mirror the consumers. Once it's done this PR
should get us to the point where we can round trip the contents of an
individual block for most data types, but it does not address the
container format.

Closes #615.

---------

Co-authored-by: Martin Traverse <[email protected]>

Author: martin-traverse

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/ArrowToAvroUtils.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/AvroBigIntProducer.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/AvroBooleanProducer.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/AvroBytesProducer.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/AvroEnumProducer.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/AvroFixedSizeBinaryProducer.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/AvroFixedSizeListProducer.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++;
+  }
+}