This is an automated email from the ASF dual-hosted git repository. RyanSkraba pushed a commit to branch branch-1.11 in repository https://gitbox.apache.org/repos/asf/avro.git
commit 9edad7361427e9a691fff125bed74202dc71c6fc Author: Ismaël Mejía <[email protected]> AuthorDate: Fri May 1 12:49:38 2026 +0200 AVRO-4241: [Java] Validate remaining bytes before allocating collections to prevent OOM from malicious data --- .../apache/avro/generic/GenericDatumReader.java | 118 ++++++++- .../java/org/apache/avro/io/BinaryDecoder.java | 65 +++++ .../src/main/java/org/apache/avro/io/Decoder.java | 10 + .../java/org/apache/avro/io/ValidatingDecoder.java | 5 + .../apache/avro/util/ByteBufferInputStream.java | 12 + .../avro/generic/TestGenericDatumReader.java | 186 +++++++++++++ .../java/org/apache/avro/io/TestBinaryDecoder.java | 45 ++++ .../perf/test/generic/MinBytesPerElementTest.java | 287 +++++++++++++++++++++ 8 files changed, 726 insertions(+), 2 deletions(-) diff --git a/lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader.java b/lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader.java index b818a070c1..f8f092b399 100644 --- a/lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader.java +++ b/lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader.java @@ -17,12 +17,16 @@ */ package org.apache.avro.generic; +import java.io.EOFException; import java.io.IOException; import java.lang.reflect.Constructor; import java.nio.ByteBuffer; import java.util.Collection; +import java.util.Collections; import java.util.HashMap; +import java.util.IdentityHashMap; import java.util.Map; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.function.Function; @@ -285,6 +289,7 @@ public class GenericDatumReader<D> implements DatumReader<D> { long l = in.readArrayStart(); long base = 0; if (l > 0) { + ensureAvailableCollectionBytes(in, l, expectedType); LogicalType logicalType = expectedType.getLogicalType(); Conversion<?> conversion = getData().getConversionFor(logicalType); Object array = newArray(old, (int) l, expected); @@ -300,13 +305,23 @@ public class GenericDatumReader<D> implements DatumReader<D> { } } base += l; - } while ((l = in.arrayNext()) > 0); + } while ((l = arrayNext(in, expectedType)) > 0); return pruneArray(array); } else { return pruneArray(newArray(old, 0, expected)); } } + /** + * Reads the next array block count and validates remaining bytes before the + * caller allocates storage. + */ + private long arrayNext(ResolvingDecoder in, Schema elementType) throws IOException { + long l = in.arrayNext(); + ensureAvailableCollectionBytes(in, l, elementType); + return l; + } + private Object pruneArray(Object object) { if (object instanceof GenericArray<?>) { ((GenericArray<?>) object).prune(); @@ -342,6 +357,7 @@ public class GenericDatumReader<D> implements DatumReader<D> { long l = in.readMapStart(); LogicalType logicalType = eValue.getLogicalType(); Conversion<?> conversion = getData().getConversionFor(logicalType); + ensureAvailableMapBytes(in, l, eValue); Object map = newMap(old, (int) l); if (l > 0) { do { @@ -355,11 +371,39 @@ public class GenericDatumReader<D> implements DatumReader<D> { addToMap(map, readMapKey(null, expected, in), readWithoutConversion(null, eValue, in)); } } - } while ((l = in.mapNext()) > 0); + } while ((l = mapNext(in, eValue)) > 0); } return map; } + /** + * Reads the next map block count and validates remaining bytes before the + * caller allocates storage. + */ + private long mapNext(ResolvingDecoder in, Schema valueType) throws IOException { + long l = in.mapNext(); + ensureAvailableMapBytes(in, l, valueType); + return l; + } + + /** + * Validates remaining bytes for a map block. Each map entry has a string key + * (at least 1 byte for the length varint) plus a value, so the minimum bytes + * per entry is {@code 1 + minBytesPerElement(valueSchema)}. + */ + private static void ensureAvailableMapBytes(Decoder decoder, long count, Schema valueSchema) throws EOFException { + if (count <= 0) { + return; + } + // Map keys are always strings: at least 1 byte for the length varint + long minBytesPerEntry = 1L + minBytesPerElement(valueSchema); + int remaining = decoder.remainingBytes(); + if (remaining >= 0 && count * minBytesPerEntry > remaining) { + throw new EOFException("Map claims " + count + " entries with at least " + minBytesPerEntry + + " bytes each, but only " + remaining + " bytes are available"); + } + } + /** * Called by the default implementation of {@link #readMap} to read a key value. * The default implementation returns delegates to @@ -378,6 +422,76 @@ public class GenericDatumReader<D> implements DatumReader<D> { ((Map) map).put(key, value); } + /** + * Returns the minimum number of bytes required to encode a single value of the + * given schema in Avro binary format. Used to validate that the decoder has + * enough data remaining before allocating collection backing arrays. + * <p> + * Returns 0 for types whose binary encoding is empty ({@code null}, zero-length + * {@code fixed}, records with only zero-byte fields). Returns a positive value + * for all other types. + */ + static int minBytesPerElement(Schema schema) { + return minBytesPerElement(schema, Collections.newSetFromMap(new IdentityHashMap<>())); + } + + private static int minBytesPerElement(Schema schema, Set<Schema> visited) { + switch (schema.getType()) { + case NULL: + return 0; + case FIXED: + return schema.getFixedSize(); + case FLOAT: + return 4; + case DOUBLE: + return 8; + case RECORD: + if (!visited.add(schema)) { + return 0; // break recursion for self-referencing schemas + } + long sum = 0; + for (Schema.Field f : schema.getFields()) { + sum += minBytesPerElement(f.schema(), visited); + if (sum >= Integer.MAX_VALUE) { + sum = Integer.MAX_VALUE; + break; + } + } + visited.remove(schema); + return (int) sum; + case UNION: + // The branch index varint is always at least 1 byte + return 1; + default: + // BOOLEAN, INT, LONG, ENUM, STRING, BYTES, ARRAY, MAP are all >= 1 byte + return 1; + } + } + + /** + * Validates that the decoder has enough remaining bytes to hold {@code count} + * elements of the given schema, assuming each element requires at least + * {@link #minBytesPerElement} bytes. Throws {@link EOFException} if the decoder + * reports fewer remaining bytes than required. + * <p> + * This check prevents out-of-memory errors from pre-allocating huge backing + * arrays when the source data is truncated or malicious. + */ + private static void ensureAvailableCollectionBytes(Decoder decoder, long count, Schema elementSchema) + throws EOFException { + if (count <= 0) { + return; + } + int minBytes = minBytesPerElement(elementSchema); + if (minBytes > 0) { + int remaining = decoder.remainingBytes(); + if (remaining >= 0 && count * (long) minBytes > remaining) { + throw new EOFException("Collection claims " + count + " elements with at least " + minBytes + + " bytes each, but only " + remaining + " bytes are available"); + } + } + } + /** * Called to read a fixed value. May be overridden for alternate fixed * representations. By default, returns {@link GenericFixed}. diff --git a/lang/java/avro/src/main/java/org/apache/avro/io/BinaryDecoder.java b/lang/java/avro/src/main/java/org/apache/avro/io/BinaryDecoder.java index 4e45781b03..67ffa34f9c 100644 --- a/lang/java/avro/src/main/java/org/apache/avro/io/BinaryDecoder.java +++ b/lang/java/avro/src/main/java/org/apache/avro/io/BinaryDecoder.java @@ -17,6 +17,7 @@ */ package org.apache.avro.io; +import java.io.ByteArrayInputStream; import java.io.EOFException; import java.io.IOException; import java.io.InputStream; @@ -27,6 +28,7 @@ import java.util.Arrays; import org.apache.avro.AvroRuntimeException; import org.apache.avro.InvalidNumberEncodingException; import org.apache.avro.SystemLimitException; +import org.apache.avro.util.ByteBufferInputStream; import org.apache.avro.util.Utf8; /** @@ -283,6 +285,7 @@ public class BinaryDecoder extends Decoder { @Override public Utf8 readString(Utf8 old) throws IOException { int length = SystemLimitException.checkMaxStringLength(readLong()); + ensureAvailableBytes(length); Utf8 result = (old != null ? old : new Utf8()); result.setByteLength(length); if (0 != length) { @@ -306,6 +309,7 @@ public class BinaryDecoder extends Decoder { @Override public ByteBuffer readBytes(ByteBuffer old) throws IOException { int length = SystemLimitException.checkMaxBytesLength(readLong()); + ensureAvailableBytes(length); final ByteBuffer result; if (old != null && length <= old.capacity()) { result = old; @@ -493,6 +497,21 @@ public class BinaryDecoder extends Decoder { return (0 == read); } + /** + * Returns the total number of bytes remaining that can be read from this + * decoder (including any buffered bytes), or {@code -1} if the total is + * unknown. + * <p> + * Byte-array-backed decoders return an exact count. InputStream-backed decoders + * return an exact count only when the wrapped stream can report one. + * <p> + * {@link DirectBinaryDecoder} always returns {@code -1}. + */ + @Override + public int remainingBytes() { + return source != null ? source.remainingBytes() : -1; + } + /** * Ensures that buf[pos + num - 1] is not out of the buffer array bounds. * However, buf[pos + num -1] may be >= limit if there is not enough data left @@ -515,6 +534,27 @@ public class BinaryDecoder extends Decoder { } } + /** + * Validates that the source has at least {@code length} bytes remaining before + * proceeding. Throws early if the declared length is inconsistent with the + * available data. + * <p> + * This check is only applied when the decoder knows the exact remaining byte + * count. + * + * @param length the number of bytes expected to be available + * @throws EOFException if the source is known to have fewer bytes remaining + */ + private void ensureAvailableBytes(int length) throws EOFException { + if (source != null && length > 0) { + int remaining = source.remainingBytes(); + if (remaining >= 0 && length > remaining) { + throw new EOFException( + "Attempted to read " + length + " bytes, but only " + remaining + " bytes are available"); + } + } + } + /** * Returns an {@link java.io.InputStream} that is aware of any buffering that * may occur in this BinaryDecoder. Readers that need to interleave decoding @@ -649,6 +689,12 @@ public class BinaryDecoder extends Decoder { abstract boolean isEof(); + /** + * Returns the total number of bytes remaining that can be read from this source + * (including any buffered bytes), or {@code -1} if the total is unknown. + */ + protected abstract int remainingBytes(); + protected void attach(int bufferSize, BinaryDecoder decoder) { decoder.buf = new byte[bufferSize]; decoder.pos = 0; @@ -895,6 +941,20 @@ public class BinaryDecoder extends Decoder { return isEof; } + @Override + protected int remainingBytes() { + int buffered = ba.getLim() - ba.getPos(); + try { + if (in.getClass() == ByteArrayInputStream.class || in.getClass() == ByteBufferInputStream.class) { + long total = (long) buffered + in.available(); + return (int) Math.min(total, Integer.MAX_VALUE); + } + } catch (IOException e) { + return -1; + } + return -1; + } + @Override public void close() throws IOException { in.close(); @@ -1013,5 +1073,10 @@ public class BinaryDecoder extends Decoder { int remaining = ba.getLim() - ba.getPos(); return (remaining == 0); } + + @Override + protected int remainingBytes() { + return ba.getLim() - ba.getPos(); + } } } diff --git a/lang/java/avro/src/main/java/org/apache/avro/io/Decoder.java b/lang/java/avro/src/main/java/org/apache/avro/io/Decoder.java index 11fc28d762..80640a61aa 100644 --- a/lang/java/avro/src/main/java/org/apache/avro/io/Decoder.java +++ b/lang/java/avro/src/main/java/org/apache/avro/io/Decoder.java @@ -299,4 +299,14 @@ public abstract class Decoder { * type of the next value to be read */ public abstract int readIndex() throws IOException; + + /** + * Returns the total number of bytes remaining that can be read from this + * decoder, or {@code -1} if the total is unknown. Implementations that can + * determine remaining capacity (for example, byte-array-backed decoders) should + * override this method. The default returns {@code -1}. + */ + public int remainingBytes() { + return -1; + } } diff --git a/lang/java/avro/src/main/java/org/apache/avro/io/ValidatingDecoder.java b/lang/java/avro/src/main/java/org/apache/avro/io/ValidatingDecoder.java index dbee445857..26f79a16ff 100644 --- a/lang/java/avro/src/main/java/org/apache/avro/io/ValidatingDecoder.java +++ b/lang/java/avro/src/main/java/org/apache/avro/io/ValidatingDecoder.java @@ -246,4 +246,9 @@ public class ValidatingDecoder extends ParsingDecoder implements Parser.ActionHa public Symbol doAction(Symbol input, Symbol top) throws IOException { return null; } + + @Override + public int remainingBytes() { + return in != null ? in.remainingBytes() : -1; + } } diff --git a/lang/java/avro/src/main/java/org/apache/avro/util/ByteBufferInputStream.java b/lang/java/avro/src/main/java/org/apache/avro/util/ByteBufferInputStream.java index 6abb62015d..375abc23fb 100644 --- a/lang/java/avro/src/main/java/org/apache/avro/util/ByteBufferInputStream.java +++ b/lang/java/avro/src/main/java/org/apache/avro/util/ByteBufferInputStream.java @@ -65,6 +65,18 @@ public class ByteBufferInputStream extends InputStream { } } + @Override + public int available() throws IOException { + long remaining = 0; + for (int i = current; i < buffers.size(); i++) { + remaining += buffers.get(i).remaining(); + if (remaining >= Integer.MAX_VALUE) { + return Integer.MAX_VALUE; + } + } + return (int) remaining; + } + /** * Read a buffer from the input without copying, if possible. */ diff --git a/lang/java/avro/src/test/java/org/apache/avro/generic/TestGenericDatumReader.java b/lang/java/avro/src/test/java/org/apache/avro/generic/TestGenericDatumReader.java index 3ec67cd113..714d2e6081 100644 --- a/lang/java/avro/src/test/java/org/apache/avro/generic/TestGenericDatumReader.java +++ b/lang/java/avro/src/test/java/org/apache/avro/generic/TestGenericDatumReader.java @@ -17,17 +17,26 @@ */ package org.apache.avro.generic; +import java.io.ByteArrayOutputStream; +import java.io.EOFException; +import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.List; import java.util.Random; import java.util.stream.Collectors; import java.util.stream.IntStream; import org.apache.avro.Schema; +import org.apache.avro.io.BinaryDecoder; +import org.apache.avro.io.BinaryEncoder; +import org.apache.avro.io.DecoderFactory; +import org.apache.avro.io.EncoderFactory; import org.junit.Test; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThrows; public class TestGenericDatumReader { @@ -117,4 +126,181 @@ public class TestGenericDatumReader { } } } + + // --- minBytesPerElement tests --- + + @Test + public void testMinBytesPerElementPrimitives() { + assertEquals(0, GenericDatumReader.minBytesPerElement(Schema.create(Schema.Type.NULL))); + assertEquals(1, GenericDatumReader.minBytesPerElement(Schema.create(Schema.Type.BOOLEAN))); + assertEquals(1, GenericDatumReader.minBytesPerElement(Schema.create(Schema.Type.INT))); + assertEquals(1, GenericDatumReader.minBytesPerElement(Schema.create(Schema.Type.LONG))); + assertEquals(4, GenericDatumReader.minBytesPerElement(Schema.create(Schema.Type.FLOAT))); + assertEquals(8, GenericDatumReader.minBytesPerElement(Schema.create(Schema.Type.DOUBLE))); + assertEquals(1, GenericDatumReader.minBytesPerElement(Schema.create(Schema.Type.STRING))); + assertEquals(1, GenericDatumReader.minBytesPerElement(Schema.create(Schema.Type.BYTES))); + } + + @Test + public void testMinBytesPerElementFixed() { + assertEquals(0, GenericDatumReader.minBytesPerElement(Schema.createFixed("ZeroFixed", null, "test", 0))); + assertEquals(5, GenericDatumReader.minBytesPerElement(Schema.createFixed("FiveFixed", null, "test", 5))); + assertEquals(16, GenericDatumReader.minBytesPerElement(Schema.createFixed("SixteenFixed", null, "test", 16))); + } + + @Test + public void testMinBytesPerElementUnion() { + // Union always >= 1 byte (branch index varint) + Schema nullableInt = Schema.createUnion(Schema.create(Schema.Type.NULL), Schema.create(Schema.Type.INT)); + assertEquals(1, GenericDatumReader.minBytesPerElement(nullableInt)); + } + + @Test + public void testMinBytesPerElementRecord() { + // Empty record = 0 bytes + Schema emptyRecord = Schema.createRecord("Empty", null, "test", false); + emptyRecord.setFields(Collections.emptyList()); + assertEquals(0, GenericDatumReader.minBytesPerElement(emptyRecord)); + + // Record with a single non-null field >= 1 byte + Schema recWithInt = Schema.createRecord("WithInt", null, "test", false); + recWithInt.setFields(Collections.singletonList(new Schema.Field("x", Schema.create(Schema.Type.INT)))); + assertEquals(1, GenericDatumReader.minBytesPerElement(recWithInt)); + + // Record with only null fields = 0 bytes + Schema recWithNull = Schema.createRecord("WithNull", null, "test", false); + recWithNull.setFields(Collections.singletonList(new Schema.Field("n", Schema.create(Schema.Type.NULL)))); + assertEquals(0, GenericDatumReader.minBytesPerElement(recWithNull)); + + Schema recWithMultipleFields = Schema.createRecord("WithMultipleFields", null, "test", false); + recWithMultipleFields.setFields(Arrays.asList(new Schema.Field("f", Schema.create(Schema.Type.FLOAT)), + new Schema.Field("d", Schema.create(Schema.Type.DOUBLE)))); + assertEquals(12, GenericDatumReader.minBytesPerElement(recWithMultipleFields)); + } + + @Test + public void testMinBytesPerElementNestedCollections() { + // Array and map types are >= 1 byte (count varint) + assertEquals(1, GenericDatumReader.minBytesPerElement(Schema.createArray(Schema.create(Schema.Type.INT)))); + assertEquals(1, GenericDatumReader.minBytesPerElement(Schema.createMap(Schema.create(Schema.Type.INT)))); + } + + // --- Collection byte validation end-to-end tests --- + + /** + * Encodes the given longs as Avro varints into a byte array. + */ + private static byte[] encodeVarints(long... values) throws IOException { + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + BinaryEncoder enc = EncoderFactory.get().directBinaryEncoder(baos, null); + for (long v : values) { + enc.writeLong(v); + } + enc.flush(); + return baos.toByteArray(); + } + + /** + * Verify that reading an array of ints with a huge count but no element data + * throws EOFException from the schema-aware byte check. + */ + @Test + public void arrayOfIntsRejectsHugeCount() throws Exception { + Schema schema = Schema.createArray(Schema.create(Schema.Type.INT)); + GenericDatumReader<Object> reader = new GenericDatumReader<>(schema); + + // Binary: varint(10_000_000) for block count, varint(0) for terminator. + // No actual element data -- the reader should reject before allocating. + byte[] data = encodeVarints(10_000_000L, 0L); + BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(data, null); + assertThrows(EOFException.class, () -> reader.read(null, decoder)); + } + + /** + * Verify that reading an array of nulls with a large count SUCCEEDS because + * null elements are 0 bytes each, so the byte check is correctly skipped. + */ + @Test + public void arrayOfNullsAcceptsLargeCount() throws Exception { + Schema schema = Schema.createArray(Schema.create(Schema.Type.NULL)); + GenericDatumReader<Object> reader = new GenericDatumReader<>(schema); + + // Binary: varint(1000) for block count, varint(0) for terminator. + // 1000 null elements = 0 bytes of element data. + byte[] data = encodeVarints(1000L, 0L); + BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(data, null); + GenericData.Array<?> result = (GenericData.Array<?>) reader.read(null, decoder); + assertEquals(1000, result.size()); + } + + /** + * Verify that reading a map of string->int with a huge count throws + * EOFException. Each map entry needs at least 2 bytes (1 for key length varint + * + 1 for int value). + */ + @Test + public void mapOfStringToIntRejectsHugeCount() throws Exception { + Schema schema = Schema.createMap(Schema.create(Schema.Type.INT)); + GenericDatumReader<Object> reader = new GenericDatumReader<>(schema); + + byte[] data = encodeVarints(10_000_000L, 0L); + BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(data, null); + assertThrows(EOFException.class, () -> reader.read(null, decoder)); + } + + /** + * Verify that reading a map of string->null with a huge count also throws + * EOFException because map keys are always strings (at least 1 byte each). + */ + @Test + public void mapOfStringToNullRejectsHugeCount() throws Exception { + Schema schema = Schema.createMap(Schema.create(Schema.Type.NULL)); + GenericDatumReader<Object> reader = new GenericDatumReader<>(schema); + + byte[] data = encodeVarints(10_000_000L, 0L); + BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(data, null); + assertThrows(EOFException.class, () -> reader.read(null, decoder)); + } + + /** + * Verify that reading an array of zero-length fixed elements with a large count + * SUCCEEDS because zero-length fixed elements are 0 bytes each. + */ + @Test + public void arrayOfZeroLengthFixedAcceptsLargeCount() throws Exception { + Schema fixedSchema = Schema.createFixed("Empty", null, "test", 0); + Schema schema = Schema.createArray(fixedSchema); + GenericDatumReader<Object> reader = new GenericDatumReader<>(schema); + + byte[] data = encodeVarints(500L, 0L); + BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(data, null); + GenericData.Array<?> result = (GenericData.Array<?>) reader.read(null, decoder); + assertEquals(500, result.size()); + } + + @Test + public void arrayOfRecordsRejectsHugeCountUsingFullRecordSize() throws Exception { + Schema recordSchema = Schema.createRecord("Element", null, "test", false); + recordSchema.setFields(Arrays.asList(new Schema.Field("f", Schema.create(Schema.Type.FLOAT)), + new Schema.Field("d", Schema.create(Schema.Type.DOUBLE)))); + Schema schema = Schema.createArray(recordSchema); + GenericDatumReader<Object> reader = new GenericDatumReader<>(schema); + + byte[] data = encodeVarints(2L, 0L); + BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(data, null); + assertThrows(EOFException.class, () -> reader.read(null, decoder)); + } + + @Test + public void mapOfRecordsRejectsHugeCountUsingFullRecordSize() throws Exception { + Schema recordSchema = Schema.createRecord("MapValue", null, "test", false); + recordSchema.setFields(Arrays.asList(new Schema.Field("f", Schema.create(Schema.Type.FLOAT)), + new Schema.Field("d", Schema.create(Schema.Type.DOUBLE)))); + Schema schema = Schema.createMap(recordSchema); + GenericDatumReader<Object> reader = new GenericDatumReader<>(schema); + + byte[] data = encodeVarints(1L, 0L); + BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(data, null); + assertThrows(EOFException.class, () -> reader.read(null, decoder)); + } } diff --git a/lang/java/avro/src/test/java/org/apache/avro/io/TestBinaryDecoder.java b/lang/java/avro/src/test/java/org/apache/avro/io/TestBinaryDecoder.java index 6010fc9c69..f171366656 100644 --- a/lang/java/avro/src/test/java/org/apache/avro/io/TestBinaryDecoder.java +++ b/lang/java/avro/src/test/java/org/apache/avro/io/TestBinaryDecoder.java @@ -682,4 +682,49 @@ public class TestBinaryDecoder { Assertions.assertThrows(EOFException.class, () -> d.readInt()); } + /** + * Verify that a byte-array-backed decoder rejects a string whose varint length + * exceeds the remaining bytes, throwing {@link EOFException} <em>before</em> + * allocating the buffer. + */ + @Test + public void testStringLengthExceedsAvailableBytes() throws IOException { + // Encode a varint claiming 10_000_000 bytes of string data, but supply none. + // The byte-array-backed decoder knows it has only a few bytes left after + // the varint, so ensureAvailableBytes must throw EOFException. + BinaryDecoder bd = newDecoder(false, 10_000_000L); + Assertions.assertThrows(EOFException.class, () -> bd.readString(null)); + } + + /** + * Same as {@link #testStringLengthExceedsAvailableBytes()} but for + * {@link BinaryDecoder#readBytes(ByteBuffer)}. + */ + @Test + public void testBytesLengthExceedsAvailableBytes() throws IOException { + BinaryDecoder bd = newDecoder(false, 10_000_000L); + Assertions.assertThrows(EOFException.class, () -> bd.readBytes(null)); + } + + @Test + public void testStringLengthDoesNotTrustUnknownAvailable() throws IOException { + byte[] encoded; + try (ByteArrayOutputStream baos = new ByteArrayOutputStream()) { + BinaryEncoder encoder = EncoderFactory.get().binaryEncoder(baos, null); + encoder.writeString("hello"); + encoder.flush(); + encoded = baos.toByteArray(); + } + + InputStream in = new ByteArrayInputStream(encoded) { + @Override + public synchronized int available() { + return 0; + } + }; + + BinaryDecoder decoder = factory.binaryDecoder(in, null); + Assertions.assertEquals("hello", decoder.readString(null).toString()); + } + } diff --git a/lang/java/perf/src/main/java/org/apache/avro/perf/test/generic/MinBytesPerElementTest.java b/lang/java/perf/src/main/java/org/apache/avro/perf/test/generic/MinBytesPerElementTest.java new file mode 100644 index 0000000000..dc831c73d5 --- /dev/null +++ b/lang/java/perf/src/main/java/org/apache/avro/perf/test/generic/MinBytesPerElementTest.java @@ -0,0 +1,287 @@ +/* + * 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 + * + * https://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.avro.perf.test.generic; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Random; + +import org.apache.avro.Schema; +import org.apache.avro.generic.GenericData; +import org.apache.avro.generic.GenericDatumReader; +import org.apache.avro.generic.GenericDatumWriter; +import org.apache.avro.generic.GenericRecord; +import org.apache.avro.io.DatumReader; +import org.apache.avro.io.DatumWriter; +import org.apache.avro.io.Decoder; +import org.apache.avro.io.DecoderFactory; +import org.apache.avro.io.Encoder; +import org.apache.avro.io.EncoderFactory; +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Level; +import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Param; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.Warmup; +import org.openjdk.jmh.infra.Blackhole; + +import java.util.concurrent.TimeUnit; + +/** + * Benchmarks to measure the overhead of {@code minBytesPerElement} computation + * during array/map decoding via {@link GenericDatumReader}. Tests complex, + * wide, and recursive schema structures to verify that the per-block schema + * traversal cost is acceptable. + */ +@BenchmarkMode(Mode.Throughput) +@OutputTimeUnit(TimeUnit.SECONDS) +@Warmup(iterations = 3, time = 2) +@Measurement(iterations = 5, time = 3) +@Fork(1) +@State(Scope.Thread) +public class MinBytesPerElementTest { + + /** + * Schema complexity levels to benchmark: - WIDE: Record with 50 fields of + * various types - DEEP: Deeply nested records (10 levels) - RECURSIVE: + * Self-referencing record (linked-list style) + */ + @Param({ "WIDE", "DEEP", "RECURSIVE" }) + public String schemaType; + + private byte[] encodedArrayData; + private byte[] encodedMapData; + private DatumReader<GenericRecord> arrayReader; + private DatumReader<GenericRecord> mapReader; + private Schema arrayWrapperSchema; + private Schema mapWrapperSchema; + + @Setup(Level.Trial) + public void setup() throws IOException { + Schema elementSchema = buildElementSchema(schemaType); + + // Wrap in array and map schemas for testing collection decoding paths + arrayWrapperSchema = Schema.createRecord("ArrayWrapper", null, "test", false); + arrayWrapperSchema + .setFields(Arrays.asList(new Schema.Field("items", Schema.createArray(elementSchema), null, null))); + + mapWrapperSchema = Schema.createRecord("MapWrapper", null, "test", false); + mapWrapperSchema.setFields(Arrays.asList(new Schema.Field("entries", Schema.createMap(elementSchema), null, null))); + + arrayReader = new GenericDatumReader<>(arrayWrapperSchema); + mapReader = new GenericDatumReader<>(mapWrapperSchema); + + // Encode test data: array with 1000 elements + encodedArrayData = encodeArrayData(arrayWrapperSchema, elementSchema, 1000); + // Encode test data: map with 1000 entries + encodedMapData = encodeMapData(mapWrapperSchema, elementSchema, 1000); + } + + @Benchmark + public void decodeArrayOfComplexRecords(Blackhole bh) throws IOException { + Decoder decoder = DecoderFactory.get().binaryDecoder(encodedArrayData, null); + GenericRecord result = arrayReader.read(null, decoder); + bh.consume(result); + } + + @Benchmark + public void decodeMapOfComplexRecords(Blackhole bh) throws IOException { + Decoder decoder = DecoderFactory.get().binaryDecoder(encodedMapData, null); + GenericRecord result = mapReader.read(null, decoder); + bh.consume(result); + } + + private static Schema buildElementSchema(String type) { + switch (type) { + case "WIDE": + return buildWideSchema(); + case "DEEP": + return buildDeepSchema(); + case "RECURSIVE": + return buildRecursiveSchema(); + default: + throw new IllegalArgumentException("Unknown schema type: " + type); + } + } + + /** + * Wide record: 50 fields of mixed types (int, long, double, float, string, + * boolean, bytes, nested record). + */ + private static Schema buildWideSchema() { + Schema innerRecord = Schema.createRecord("Inner", null, "test", false); + innerRecord.setFields(Arrays.asList(new Schema.Field("x", Schema.create(Schema.Type.INT), null, null), + new Schema.Field("y", Schema.create(Schema.Type.DOUBLE), null, null))); + + List<Schema.Field> fields = new ArrayList<>(); + for (int i = 0; i < 10; i++) { + fields.add(new Schema.Field("int_" + i, Schema.create(Schema.Type.INT), null, null)); + } + for (int i = 0; i < 10; i++) { + fields.add(new Schema.Field("long_" + i, Schema.create(Schema.Type.LONG), null, null)); + } + for (int i = 0; i < 10; i++) { + fields.add(new Schema.Field("double_" + i, Schema.create(Schema.Type.DOUBLE), null, null)); + } + for (int i = 0; i < 10; i++) { + fields.add(new Schema.Field("str_" + i, Schema.create(Schema.Type.STRING), null, null)); + } + for (int i = 0; i < 5; i++) { + fields.add(new Schema.Field("bool_" + i, Schema.create(Schema.Type.BOOLEAN), null, null)); + } + for (int i = 0; i < 5; i++) { + fields.add(new Schema.Field("rec_" + i, innerRecord, null, null)); + } + + Schema wide = Schema.createRecord("WideRecord", null, "test", false); + wide.setFields(fields); + return wide; + } + + /** + * Deeply nested: 10 levels of records, each containing an int and the next + * level. + */ + private static Schema buildDeepSchema() { + Schema current = Schema.createRecord("Level10", null, "test", false); + current.setFields(Arrays.asList(new Schema.Field("value", Schema.create(Schema.Type.INT), null, null))); + + for (int i = 9; i >= 1; i--) { + Schema next = Schema.createRecord("Level" + i, null, "test", false); + next.setFields(Arrays.asList(new Schema.Field("value", Schema.create(Schema.Type.INT), null, null), + new Schema.Field("child", current, null, null))); + current = next; + } + return current; + } + + /** + * Recursive: self-referencing record (linked list). The "next" field is a union + * of null and the record itself. + */ + private static Schema buildRecursiveSchema() { + Schema recursive = Schema.createRecord("LinkedNode", null, "test", false); + Schema nullSchema = Schema.create(Schema.Type.NULL); + Schema union = Schema.createUnion(Arrays.asList(nullSchema, recursive)); + recursive.setFields(Arrays.asList(new Schema.Field("value", Schema.create(Schema.Type.INT), null, null), + new Schema.Field("next", union, null, null))); + return recursive; + } + + private static byte[] encodeArrayData(Schema wrapperSchema, Schema elementSchema, int count) throws IOException { + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + DatumWriter<GenericRecord> writer = new GenericDatumWriter<>(wrapperSchema); + Encoder encoder = EncoderFactory.get().binaryEncoder(baos, null); + + GenericRecord wrapper = new GenericData.Record(wrapperSchema); + List<GenericRecord> items = new ArrayList<>(count); + Random r = new Random(42); + for (int i = 0; i < count; i++) { + items.add(buildRecord(elementSchema, r, 0)); + } + wrapper.put("items", items); + writer.write(wrapper, encoder); + encoder.flush(); + return baos.toByteArray(); + } + + private static byte[] encodeMapData(Schema wrapperSchema, Schema elementSchema, int count) throws IOException { + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + DatumWriter<GenericRecord> writer = new GenericDatumWriter<>(wrapperSchema); + Encoder encoder = EncoderFactory.get().binaryEncoder(baos, null); + + GenericRecord wrapper = new GenericData.Record(wrapperSchema); + HashMap<String, GenericRecord> map = new HashMap<>(); + Random r = new Random(42); + for (int i = 0; i < count; i++) { + map.put("key_" + i, buildRecord(elementSchema, r, 0)); + } + wrapper.put("entries", map); + writer.write(wrapper, encoder); + encoder.flush(); + return baos.toByteArray(); + } + + private static GenericRecord buildRecord(Schema schema, Random r, int depth) { + GenericRecord rec = new GenericData.Record(schema); + for (Schema.Field field : schema.getFields()) { + rec.put(field.pos(), buildValue(field.schema(), r, depth)); + } + return rec; + } + + private static Object buildValue(Schema schema, Random r, int depth) { + switch (schema.getType()) { + case INT: + return r.nextInt(); + case LONG: + return r.nextLong(); + case FLOAT: + return r.nextFloat(); + case DOUBLE: + return r.nextDouble(); + case BOOLEAN: + return r.nextBoolean(); + case STRING: + return "s" + r.nextInt(1000); + case BYTES: + byte[] b = new byte[4]; + r.nextBytes(b); + return java.nio.ByteBuffer.wrap(b); + case RECORD: + return buildRecord(schema, r, depth + 1); + case UNION: + // For recursive schemas, limit depth + List<Schema> types = schema.getTypes(); + if (depth > 3) { + // Pick the null branch to terminate recursion + for (int i = 0; i < types.size(); i++) { + if (types.get(i).getType() == Schema.Type.NULL) { + return null; + } + } + } + // Pick non-null branch + for (int i = 0; i < types.size(); i++) { + if (types.get(i).getType() != Schema.Type.NULL) { + return buildValue(types.get(i), r, depth); + } + } + return null; + case ARRAY: + return new ArrayList<>(); + case MAP: + return new HashMap<>(); + case NULL: + return null; + default: + return null; + } + } +}
