GH-698: Improve and fix Avro read consumers (#718)

## What's Changed

This PR relates to #698 and is the second in a series intended to
provide full Avro read / write support in native Java. It adds
round-trip tests for both schemas (Arrow schema -> Avro -> Arrow) and
data (Arrow VSR -> Avro block -> Arrow VSR). It also adds a number of
fixes and improvements to the Avro Consumers so that data arrives back
in its original form after a round trip. The main changes are:

* Added a top level method in AvroToArrow to convert Avro schema
directly to Arrow schema (this may exist elsewhere, but is needed to
provide an API that matches the logic of this implementation)
* Avro unions of [ type, null ] or [ null, type ] now have special
handling, these are interpreted as a single nullable type rather than a
union. Setting legacyMode = false in the AvroToArrowConfig object is
required to enable this behaviour, otherwise unions are interpreted
literally. Unions with more than 2 elements are always interpreted
literally (but, per #108, in practice Java's current Union
implementation is probably not usable with Avro atm).
* Added support for new logical types (decimal 256, timestamp nano and 3
local timestamp types)
* Existing timestamp-mills and timestamp-micros times now interpreted as
zone-aware (previously they were interpreted as local, but now the local
timestamp types are interpreted as local - I think this is correct per
the [Avro
spec](https://avro.apache.org/docs/1.12.0/specification/#timestamps)).
Requires setting legacyMode = false.
* Removed namespaces from generated Arrow field names in complex types.
E.g. the Avro field myNamepsace.outerRecord.structField.intField should
be called just "intField" inside the Arrow struct. This doesn't affect
the skip field logic, which still works using the qualified names. This
requires setting legacyMode = false.
* Remove unexpected metadata in generated Arrow fields (empty alias
lists and attributes interpreted as part of the field schema). This
requires setting legacyMode = false.
* Use the expected child vector names for Arrow LIST and MAP types when
reading. For LIST, the default child vector is called "$data$" which is
illegal in Avro, so the child field name is also changed to "item" in
the producers. This requires setting legacyMode = false.

Breaking changes have been removed from this PR.

Per discussion below, all breaking changes are now behind a "legacyMode"
flag in the AvroToArrowConfig object, which is enabled by default in all
the original code paths.

Closes #698 .

This change is meant to allow for round trip of schemas and individual
Avro data blocks (one Avro data block -> one VSR). File-level
capabilities are not included. I have not included anything to recycle
the VSR as part of the read API, this feels like it belongs with the
file-level piece. Also I have not done anything specific for enums /
dict encoding as of yet.

Author: martin-traverse

adapter/avro/src/main/java/org/apache/arrow/adapter/avro/ArrowToAvroUtils.java

@@ -332,7 +332,17 @@ private static <T> T buildBaseTypeSchema(
 
       case List:
       case FixedSizeList:
-        return buildArraySchema(builder.array(), field, namespace);
+        // Arrow uses "$data$" as the field name for list items, that is not a valid Avro name
+        Field itemField = field.getChildren().get(0);
+        if (ListVector.DATA_VECTOR_NAME.equals(itemField.getName())) {
+          Field safeItemField =
+              new Field("item", itemField.getFieldType(), itemField.getChildren());
+          Field safeListField =
+              new Field(field.getName(), field.getFieldType(), List.of(safeItemField));
+          return buildArraySchema(builder.array(), safeListField, namespace);
+        } else {
+          return buildArraySchema(builder.array(), field, namespace);
+        }
 
       case Map:
         return buildMapSchema(builder.map(), field, namespace);

adapter/avro/src/main/java/org/apache/arrow/adapter/avro/AvroToArrow.java

@@ -59,4 +59,23 @@ public static AvroToArrowVectorIterator avroToArrowIterator(
 
     return AvroToArrowVectorIterator.create(decoder, schema, config);
   }
+
+  /**
+   * Convert an Avro schema to its Arrow equivalent.
+   *
+   * <p>The resulting set of Arrow fields matches what would be set in the VSR after calling
+   * avroToArrow() or avroToArrowIterator(), respecting the configuration in the config parameter.
+   *
+   * @param schema The Avro schema to convert
+   * @param config Configuration options for conversion
+   * @return The equivalent Arrow schema
+   */
+  public static org.apache.arrow.vector.types.pojo.Schema avroToAvroSchema(
+      Schema schema, AvroToArrowConfig config) {
+
+    Preconditions.checkNotNull(schema, "Avro schema object cannot be null");
+    Preconditions.checkNotNull(config, "config cannot be null");
+
+    return AvroToArrowUtils.createArrowSchema(schema, config);
+  }
 }

adapter/avro/src/main/java/org/apache/arrow/adapter/avro/AvroToArrowConfig.java

@@ -41,6 +41,12 @@ public class AvroToArrowConfig {
   /** The field names which to skip when reading decoder values. */
   private final Set<String> skipFieldNames;
 
+  /**
+   * Use legacy-mode to keep compatibility with old behavior (pre-2025), enabled by default. This
+   * affects how the AvroToArrow code interprets the Avro schema.
+   */
+  private final boolean legacyMode;
+
   /**
    * Instantiate an instance.
    *
@@ -64,6 +70,37 @@ public class AvroToArrowConfig {
     this.targetBatchSize = targetBatchSize;
     this.provider = provider;
     this.skipFieldNames = skipFieldNames;
+
+    // Default values for optional parameters
+    legacyMode = true; // Keep compatibility with old behavior by default
+  }
+
+  /**
+   * Instantiate an instance.
+   *
+   * @param allocator The memory allocator to construct the Arrow vectors with.
+   * @param targetBatchSize The maximum rowCount to read each time when partially convert data.
+   * @param provider The dictionary provider used for enum type, adapter will update this provider.
+   * @param skipFieldNames Field names which to skip.
+   * @param legacyMode Keep compatibility with old behavior (pre-2025)
+   */
+  AvroToArrowConfig(
+      BufferAllocator allocator,
+      int targetBatchSize,
+      DictionaryProvider.MapDictionaryProvider provider,
+      Set<String> skipFieldNames,
+      boolean legacyMode) {
+
+    Preconditions.checkArgument(
+        targetBatchSize == AvroToArrowVectorIterator.NO_LIMIT_BATCH_SIZE || targetBatchSize > 0,
+        "invalid targetBatchSize: %s",
+        targetBatchSize);
+
+    this.allocator = allocator;
+    this.targetBatchSize = targetBatchSize;
+    this.provider = provider;
+    this.skipFieldNames = skipFieldNames;
+    this.legacyMode = legacyMode;
   }
 
   public BufferAllocator getAllocator() {
@@ -81,4 +118,8 @@ public DictionaryProvider.MapDictionaryProvider getProvider() {
   public Set<String> getSkipFieldNames() {
     return skipFieldNames;
   }
+
+  public boolean isLegacyMode() {
+    return legacyMode;
+  }
 }

adapter/avro/src/main/java/org/apache/arrow/adapter/avro/AvroToArrowUtils.java

@@ -0,0 +1,82 @@
+/*
+ * 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.consumers;
+
+import java.io.IOException;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.avro.io.Decoder;
+
+/**
+ * Consumer wrapper which consumes nullable type values from avro decoder. Write the data to the
+ * underlying {@link FieldVector}.
+ *
+ * @param <T> The vector within consumer or its delegate.
+ */
+public class AvroNullableConsumer<T extends FieldVector> extends BaseAvroConsumer<T> {
+
+  private final Consumer<T> delegate;
+  private final int nullIndex;
+
+  /** Instantiate a AvroNullableConsumer. */
+  @SuppressWarnings("unchecked")
+  public AvroNullableConsumer(Consumer<T> delegate, int nullIndex) {
+    super((T) delegate.getVector());
+    this.delegate = delegate;
+    this.nullIndex = nullIndex;
+  }
+
+  @Override
+  public void consume(Decoder decoder) throws IOException {
+    int typeIndex = decoder.readInt();
+    if (typeIndex == nullIndex) {
+      decoder.readNull();
+      delegate.addNull();
+    } else {
+      delegate.consume(decoder);
+    }
+    currentIndex++;
+  }
+
+  @Override
+  public void addNull() {
+    // Can be called by containers of nullable types
+    delegate.addNull();
+    currentIndex++;
+  }
+
+  @Override
+  public void setPosition(int index) {
+    if (index < 0 || index > vector.getValueCount()) {
+      throw new IllegalArgumentException("Index out of bounds");
+    }
+    delegate.setPosition(index);
+    super.setPosition(index);
+  }
+
+  @Override
+  public boolean resetValueVector(T vector) {
+    boolean delegateOk = delegate.resetValueVector(vector);
+    boolean thisOk = super.resetValueVector(vector);
+    return thisOk && delegateOk;
+  }
+
+  @Override
+  public void close() throws Exception {
+    super.close();
+    delegate.close();
+  }
+}

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

@@ -0,0 +1,74 @@
+/*
+ * 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.consumers.logical;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import org.apache.arrow.adapter.avro.consumers.BaseAvroConsumer;
+import org.apache.arrow.util.Preconditions;
+import org.apache.arrow.vector.Decimal256Vector;
+import org.apache.avro.io.Decoder;
+
+/**
+ * Consumer which consume 256-bit decimal type values from avro decoder. Write the data to {@link
+ * Decimal256Vector}.
+ */
+public abstract class AvroDecimal256Consumer extends BaseAvroConsumer<Decimal256Vector> {
+
+  protected AvroDecimal256Consumer(Decimal256Vector vector) {
+    super(vector);
+  }
+
+  /** Consumer for decimal logical type with 256 bit width and original bytes type. */
+  public static class BytesDecimal256Consumer extends AvroDecimal256Consumer {
+
+    private ByteBuffer cacheBuffer;
+
+    /** Instantiate a BytesDecimal256Consumer. */
+    public BytesDecimal256Consumer(Decimal256Vector vector) {
+      super(vector);
+    }
+
+    @Override
+    public void consume(Decoder decoder) throws IOException {
+      cacheBuffer = decoder.readBytes(cacheBuffer);
+      byte[] bytes = new byte[cacheBuffer.limit()];
+      Preconditions.checkArgument(bytes.length <= 32, "Decimal bytes length should <= 32.");
+      cacheBuffer.get(bytes);
+      vector.setBigEndian(currentIndex++, bytes);
+    }
+  }
+
+  /** Consumer for decimal logical type with 256 bit width and original fixed type. */
+  public static class FixedDecimal256Consumer extends AvroDecimal256Consumer {
+
+    private final byte[] reuseBytes;
+
+    /** Instantiate a FixedDecimal256Consumer. */
+    public FixedDecimal256Consumer(Decimal256Vector vector, int size) {
+      super(vector);
+      Preconditions.checkArgument(size <= 32, "Decimal bytes length should <= 32.");
+      reuseBytes = new byte[size];
+    }
+
+    @Override
+    public void consume(Decoder decoder) throws IOException {
+      decoder.readFixed(reuseBytes);
+      vector.setBigEndian(currentIndex++, reuseBytes);
+    }
+  }
+}

adapter/avro/src/main/java/org/apache/arrow/adapter/avro/consumers/logical/AvroDecimal256Consumer.java

@@ -22,7 +22,7 @@
 import org.apache.avro.io.Decoder;
 
 /**
- * Consumer which consume date timestamp-micro values from avro decoder. Write the data to {@link
+ * Consumer which consumes local-timestamp-micros values from avro decoder. Write the data to {@link
  * TimeStampMicroVector}.
  */
 public class AvroTimestampMicrosConsumer extends BaseAvroConsumer<TimeStampMicroVector> {

adapter/avro/src/main/java/org/apache/arrow/adapter/avro/consumers/logical/AvroTimestampMicrosConsumer.java

@@ -0,0 +1,39 @@
+/*
+ * 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.consumers.logical;
+
+import java.io.IOException;
+import org.apache.arrow.adapter.avro.consumers.BaseAvroConsumer;
+import org.apache.arrow.vector.TimeStampMicroTZVector;
+import org.apache.avro.io.Decoder;
+
+/**
+ * Consumer which consumes timestamp-micros values from avro decoder. Write the data to {@link
+ * TimeStampMicroTZVector}.
+ */
+public class AvroTimestampMicrosTzConsumer extends BaseAvroConsumer<TimeStampMicroTZVector> {
+
+  /** Instantiate a AvroTimestampMicrosTzConsumer. */
+  public AvroTimestampMicrosTzConsumer(TimeStampMicroTZVector vector) {
+    super(vector);
+  }
+
+  @Override
+  public void consume(Decoder decoder) throws IOException {
+    vector.set(currentIndex++, decoder.readLong());
+  }
+}

adapter/avro/src/main/java/org/apache/arrow/adapter/avro/consumers/logical/AvroTimestampMicrosTzConsumer.java

@@ -22,7 +22,7 @@
 import org.apache.avro.io.Decoder;
 
 /**
- * Consumer which consume date timestamp-millis values from avro decoder. Write the data to {@link
+ * Consumer which consume local-timestamp-millis values from avro decoder. Write the data to {@link
  * TimeStampMilliVector}.
  */
 public class AvroTimestampMillisConsumer extends BaseAvroConsumer<TimeStampMilliVector> {

adapter/avro/src/main/java/org/apache/arrow/adapter/avro/consumers/logical/AvroTimestampMillisConsumer.java

@@ -0,0 +1,39 @@
+/*
+ * 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.consumers.logical;
+
+import java.io.IOException;
+import org.apache.arrow.adapter.avro.consumers.BaseAvroConsumer;
+import org.apache.arrow.vector.TimeStampMilliTZVector;
+import org.apache.avro.io.Decoder;
+
+/**
+ * Consumer which consume timestamp-millis values from avro decoder. Write the data to {@link
+ * TimeStampMilliTZVector}.
+ */
+public class AvroTimestampMillisTzConsumer extends BaseAvroConsumer<TimeStampMilliTZVector> {
+
+  /** Instantiate a AvroTimestampMillisTzConsumer. */
+  public AvroTimestampMillisTzConsumer(TimeStampMilliTZVector vector) {
+    super(vector);
+  }
+
+  @Override
+  public void consume(Decoder decoder) throws IOException {
+    vector.set(currentIndex++, decoder.readLong());
+  }
+}