GH-698: Improve and fix Avro read consumers

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/consumers/AvroNullableConsumer.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/logical/AvroDecimal256Consumer.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/AvroTimestampMicrosConsumer.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/AvroTimestampMicrosTzConsumer.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/AvroTimestampMillisConsumer.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/AvroTimestampMillisTzConsumer.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());
+  }
+}

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