GH-834: Fix Avro nullable consumer to use readIndex for unions

Author: Kaustav-Sarkar

.gitignore

@@ -32,3 +32,6 @@ cmake_install.cmake
 dependency-reduced-pom.xml
 install_manifest.txt
 target/
+.claude/
+CLAUDE.md
+*.class

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

@@ -431,7 +431,7 @@ private static Consumer createSkipConsumer(Schema schema) {
       case UNION:
         List<Consumer> unionDelegates =
             schema.getTypes().stream().map(s -> createSkipConsumer(s)).collect(Collectors.toList());
-        skipFunction = decoder -> unionDelegates.get(decoder.readInt()).consume(decoder);
+        skipFunction = decoder -> unionDelegates.get(decoder.readIndex()).consume(decoder);
 
         break;
       case ARRAY:

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

@@ -41,7 +41,7 @@ public AvroNullableConsumer(Consumer<T> delegate, int nullIndex) {
 
   @Override
   public void consume(Decoder decoder) throws IOException {
-    int typeIndex = decoder.readInt();
+    int typeIndex = decoder.readIndex();
     if (typeIndex == nullIndex) {
       decoder.readNull();
       delegate.addNull();

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

@@ -17,17 +17,17 @@
 package org.apache.arrow.adapter.avro.consumers;
 
 import java.io.IOException;
-import java.nio.ByteBuffer;
 import org.apache.arrow.vector.VarCharVector;
 import org.apache.avro.io.Decoder;
+import org.apache.avro.util.Utf8;
 
 /**
  * Consumer which consume string type values from avro decoder. Write the data to {@link
  * VarCharVector}.
  */
 public class AvroStringConsumer extends BaseAvroConsumer<VarCharVector> {
 
-  private ByteBuffer cacheBuffer;
+  private Utf8 cachedUtf8;
 
   /** Instantiate a AvroStringConsumer. */
   public AvroStringConsumer(VarCharVector vector) {
@@ -36,9 +36,7 @@ public AvroStringConsumer(VarCharVector vector) {
 
   @Override
   public void consume(Decoder decoder) throws IOException {
-    // cacheBuffer is initialized null and create in the first consume,
-    // if its capacity < size to read, decoder will create a new one with new capacity.
-    cacheBuffer = decoder.readBytes(cacheBuffer);
-    vector.setSafe(currentIndex++, cacheBuffer, 0, cacheBuffer.limit());
+    cachedUtf8 = decoder.readString(cachedUtf8);
+    vector.setSafe(currentIndex++, cachedUtf8.getBytes(), 0, cachedUtf8.getByteLength());
   }
 }

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

@@ -42,7 +42,7 @@ public AvroUnionsConsumer(UnionVector vector, Consumer[] delegates, Types.MinorT
 
   @Override
   public void consume(Decoder decoder) throws IOException {
-    int fieldIndex = decoder.readInt();
+    int fieldIndex = decoder.readIndex();
 
     ensureInnerVectorCapacity(currentIndex + 1, fieldIndex);
     Consumer delegate = delegates[fieldIndex];

adapter/avro/src/test/java/org/apache/arrow/adapter/avro/AvroToArrowTest.java

@@ -18,6 +18,8 @@
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
@@ -33,7 +35,13 @@
 import org.apache.arrow.vector.complex.StructVector;
 import org.apache.avro.Schema;
 import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericDatumWriter;
 import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.io.BinaryEncoder;
+import org.apache.avro.io.DatumWriter;
+import org.apache.avro.io.Decoder;
+import org.apache.avro.io.DecoderFactory;
+import org.apache.avro.io.EncoderFactory;
 import org.junit.jupiter.api.Test;
 
 public class AvroToArrowTest extends AvroTestBase {
@@ -474,4 +482,27 @@ public void testNullableUnionType() throws Exception {
 
     checkPrimitiveResult(expected, vector);
   }
+
+  @Test
+  public void testNullableStringTypeWithValidatingDecoder() throws Exception {
+    Schema schema = getSchema("test_nullable_string.avsc");
+
+    GenericRecord record = new GenericData.Record(schema);
+    record.put(0, "hello");
+
+    ByteArrayOutputStream out = new ByteArrayOutputStream();
+    BinaryEncoder encoder = new EncoderFactory().directBinaryEncoder(out, null);
+    DatumWriter<Object> writer = new GenericDatumWriter<>(schema);
+    writer.write(record, encoder);
+    encoder.flush();
+
+    ByteArrayInputStream in = new ByteArrayInputStream(out.toByteArray());
+    Decoder binaryDecoder = DecoderFactory.get().directBinaryDecoder(in, null);
+    Decoder validatingDecoder = DecoderFactory.get().validatingDecoder(schema, binaryDecoder);
+
+    VectorSchemaRoot root = AvroToArrow.avroToArrow(schema, validatingDecoder, config);
+    assertEquals(1, root.getRowCount());
+    FieldVector vector = root.getVector("f0");
+    assertEquals("hello", vector.getObject(0).toString());
+  }
 }