AVRO-4258: [java] remove some allocations from decoder.getBytes() and Utf8 handling

lang/java/avro/src/main/java/org/apache/avro/Conversions.java

@@ -190,8 +190,7 @@ public BigDecimal fromBytes(final ByteBuffer value, final Schema schema, final L
 
       try {
         BigInteger bg = null;
-        ByteBuffer buffer = decoder.readBytes(null);
-        byte[] array = buffer.array();
+        byte[] array = decoder.readBytes();
         if (array.length > 0) {
           bg = new BigInteger(array);
         }

lang/java/avro/src/main/java/org/apache/avro/file/DataFileReader12.java

@@ -75,9 +75,7 @@ public DataFileReader12(SeekableInput sin, DatumReader<D> reader) throws IOExcep
       do {
         for (long i = 0; i < l; i++) {
           String key = vin.readString(null).toString();
-          ByteBuffer value = vin.readBytes(null);
-          byte[] bb = new byte[value.remaining()];
-          value.get(bb);
+          byte[] bb = vin.readBytes();
           meta.put(key, bb);
         }
       } while ((l = vin.mapNext()) != 0);

lang/java/avro/src/main/java/org/apache/avro/file/DataFileStream.java

@@ -128,9 +128,7 @@ void initialize(InputStream in, byte[] magic) throws IOException {
       do {
         for (long i = 0; i < l; i++) {
           String key = vin.readString(null).toString();
-          ByteBuffer value = vin.readBytes(null);
-          byte[] bb = new byte[value.remaining()];
-          value.get(bb);
+          byte[] bb = vin.readBytes();
           header.meta.put(key, bb);
           header.metaKeyList.add(key);
         }

lang/java/avro/src/main/java/org/apache/avro/io/Decoder.java

@@ -20,6 +20,7 @@
 import java.io.IOException;
 import java.nio.ByteBuffer;
 
+import org.apache.avro.SystemLimitException;
 import org.apache.avro.util.Utf8;
 
 /**
@@ -41,6 +42,8 @@
 
 public abstract class Decoder {
 
+  private static final byte[] EMPTY_BYTES = new byte[0];
+
   /**
    * "Reads" a null value. (Doesn't actually read anything, but advances the state
    * of the parser if the implementation is stateful.)
@@ -126,6 +129,28 @@ public abstract class Decoder {
    */
   public abstract ByteBuffer readBytes(ByteBuffer old) throws IOException;
 
+  /**
+   * Reads a byte-string written by {@link Encoder#writeBytes}.
+   * <p>
+   * This is useful when you want to avoid the creation of a ByteBuffer, and only want the byte[], e.g.:
+   * <pre>
+   *     ByteBuffer buffer = decoder.readBytes(null);
+   *     byte[] array = buffer.array();
+   * </pre>
+   *
+   * @throws AvroTypeException If this is a stateful reader and byte-string is not the type of the next value to be
+   *                           read
+   */
+  public byte[] readBytes() throws IOException {
+    int length = SystemLimitException.checkMaxBytesLength(readLong());
+    if (length == 0) {
+      return EMPTY_BYTES;
+    }
+    byte[] result = new byte[length];
+    readFixed(result);
+    return result;
+  }
+
   /**
    * Discards a byte-string written by {@link Encoder#writeBytes}.
    *

lang/java/avro/src/main/java/org/apache/avro/io/FastReaderBuilder.java

@@ -385,18 +385,14 @@ private FieldReader createSimpleStringReader(Schema readerSchema) {
   private FieldReader createBytesPromotingToStringReader(Schema readerSchema) {
     String stringProperty = readerSchema.getProp(GenericData.STRING_PROP);
     if (GenericData.StringType.String.name().equals(stringProperty)) {
-      return (old, decoder) -> getStringFromByteBuffer(decoder.readBytes(null));
+      return (old, decoder) -> new String(decoder.readBytes(), StandardCharsets.UTF_8);
     } else {
-      return (old, decoder) -> getUtf8FromByteBuffer(old, decoder.readBytes(null));
+      return (old, decoder) -> getUtf8FromByteArray(old, decoder.readBytes());
     }
   }
 
-  private String getStringFromByteBuffer(ByteBuffer buffer) {
-    return new String(buffer.array(), buffer.position(), buffer.remaining(), StandardCharsets.UTF_8);
-  }
-
-  private Utf8 getUtf8FromByteBuffer(Object old, ByteBuffer buffer) {
-    return (old instanceof Utf8) ? ((Utf8) old).set(new Utf8(buffer.array())) : new Utf8(buffer.array());
+  private Utf8 getUtf8FromByteArray(Object old, byte[] bytes) {
+    return (old instanceof Utf8) ? ((Utf8) old).set(bytes) : new Utf8(bytes);
   }
 
   private FieldReader createUnionReader(WriterUnion action) throws IOException {

lang/java/avro/src/main/java/org/apache/avro/io/ResolvingDecoder.java

@@ -202,7 +202,7 @@ public double readDouble() throws IOException {
   public Utf8 readString(Utf8 old) throws IOException {
     Symbol actual = parser.advance(Symbol.STRING);
     if (actual == Symbol.BYTES) {
-      return new Utf8(in.readBytes(null).array());
+      return old == null? new Utf8(in.readBytes()) : old.set(in.readBytes());
     } else {
       assert actual == Symbol.STRING;
       return in.readString(old);
@@ -213,7 +213,7 @@ public Utf8 readString(Utf8 old) throws IOException {
   public String readString() throws IOException {
     Symbol actual = parser.advance(Symbol.STRING);
     if (actual == Symbol.BYTES) {
-      return new String(in.readBytes(null).array(), StandardCharsets.UTF_8);
+      return new String(in.readBytes(), StandardCharsets.UTF_8);
     } else {
       assert actual == Symbol.STRING;
       return in.readString();

lang/java/avro/src/main/java/org/apache/avro/reflect/ReflectDatumReader.java

@@ -244,14 +244,11 @@ protected Object createString(String value) {
 
   @Override
   protected Object readBytes(Object old, Schema s, Decoder in) throws IOException {
-    ByteBuffer bytes = in.readBytes(null);
     Class<?> c = ReflectData.getClassProp(s, SpecificData.CLASS_PROP);
     if (c != null && c.isArray()) {
-      byte[] result = new byte[bytes.remaining()];
-      bytes.get(result);
-      return result;
+      return in.readBytes();
     } else {
-      return bytes;
+      return in.readBytes(null);
     }
   }
 

lang/java/avro/src/main/java/org/apache/avro/util/Utf8.java

@@ -132,6 +132,13 @@ public Utf8 set(String string) {
     return this;
   }
 
+  public Utf8 set(byte[] bytes) {
+    this.bytes = bytes;
+    this.length = SystemLimitException.checkMaxStringLength(length);;
+    this.hash = 0;
+    return this;
+  }
+
   public Utf8 set(Utf8 other) {
     if (this.bytes.length < other.length) {
       this.bytes = new byte[other.length];

lang/java/avro/src/test/java/org/apache/avro/io/TestBinaryDecoder.java

@@ -147,6 +147,12 @@ void eofBytes(boolean useDirect) {
     Assertions.assertThrows(EOFException.class, () -> newDecoderWithNoData(useDirect).readBytes(null));
   }
 
+  @ParameterizedTest
+  @ValueSource(booleans = { true, false })
+  void eofBytesRaw(boolean useDirect) {
+    Assertions.assertThrows(EOFException.class, () -> newDecoderWithNoData(useDirect).readBytes());
+  }
+
   @ParameterizedTest
   @ValueSource(booleans = { true, false })
   void eofString(boolean useDirect) {
@@ -424,6 +430,13 @@ public void testBytesNegativeLength(boolean useDirect) throws IOException {
         () -> this.newDecoder(useDirect, -1).readBytes(null));
     Assertions.assertEquals(ERROR_NEGATIVE, ex.getMessage());
   }
+  @ParameterizedTest
+  @ValueSource(booleans = { true, false })
+  public void testBytesNegativeLengthRaw(boolean useDirect) throws IOException {
+    Exception ex = Assertions.assertThrows(AvroRuntimeException.class,
+        () -> this.newDecoder(useDirect, -1).readBytes());
+    Assertions.assertEquals(ERROR_NEGATIVE, ex.getMessage());
+  }
 
   @ParameterizedTest
   @ValueSource(booleans = { true, false })
@@ -433,6 +446,14 @@ public void testBytesVmMaxSize(boolean useDirect) throws IOException {
     Assertions.assertEquals(ERROR_VM_LIMIT_BYTES, ex.getMessage());
   }
 
+  @ParameterizedTest
+  @ValueSource(booleans = { true, false })
+  public void testBytesVmMaxSizeRaw(boolean useDirect) throws IOException {
+    Exception ex = Assertions.assertThrows(UnsupportedOperationException.class,
+        () -> this.newDecoder(useDirect, MAX_ARRAY_VM_LIMIT + 1).readBytes());
+    Assertions.assertEquals(ERROR_VM_LIMIT_BYTES, ex.getMessage());
+  }
+
   @ParameterizedTest
   @ValueSource(booleans = { true, false })
   public void testBytesMaxCustom(boolean useDirect) throws IOException {

lang/java/avro/src/test/java/org/apache/avro/util/TestUtf8.java

@@ -27,6 +27,7 @@
 import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
 import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
 
 import org.apache.avro.SystemLimitException;
 import org.apache.avro.TestSystemLimitException;
@@ -97,6 +98,20 @@ void hashCodeReused() {
     assertEquals(127791473, u.hashCode());
     u.setByteLength(4);
     assertEquals(4122302, u.hashCode());
+
+    u.set(getTrimmedBytes(new Utf8("zz")));
+    assertEquals(4865, u.hashCode());
+    u.setByteLength(1);
+    assertEquals(153, u.hashCode());
+
+    u.set(getTrimmedBytes(new Utf8("hello")));
+    assertEquals(127791473, u.hashCode());
+    u.setByteLength(4);
+    assertEquals(4122302, u.hashCode());
+  }
+
+  private byte[] getTrimmedBytes(Utf8 utf8) {
+    return Arrays.copyOf(utf8.getBytes(), utf8.getByteLength());
   }
 
   /**