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,9 @@ public class AvroToArrowConfig {
   /** The field names which to skip when reading decoder values. */
   private final Set<String> skipFieldNames;
 
+  /** Interpret unions where one type is null as a nullable field. */
+  private final boolean handleNullable;
+
   /**
    * Instantiate an instance.
    *
@@ -64,6 +67,36 @@ public class AvroToArrowConfig {
     this.targetBatchSize = targetBatchSize;
     this.provider = provider;
     this.skipFieldNames = skipFieldNames;
+
+    // Default values for optional parameters
+    handleNullable = false; // Match original 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.
+   */
+  AvroToArrowConfig(
+      BufferAllocator allocator,
+      int targetBatchSize,
+      DictionaryProvider.MapDictionaryProvider provider,
+      Set<String> skipFieldNames,
+      boolean handleNullable) {
+
+    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.handleNullable = handleNullable;
   }
 
   public BufferAllocator getAllocator() {
@@ -81,4 +114,8 @@ public DictionaryProvider.MapDictionaryProvider getProvider() {
   public Set<String> getSkipFieldNames() {
     return skipFieldNames;
   }
+
+  public boolean isHandleNullable() {
+    return handleNullable;
+  }
 }