This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12906-42bafc590cf2d9ed8c787e23143c58285d7953ab in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit e805677f731be93d254cd844355fd09c89225041 Author: hutiefang76 <[email protected]> AuthorDate: Fri Oct 30 05:02:21 2026 +0000 [Fix][Format] Preserve Avro byte buffer bounds and row ownership (#12906) --- .../seatunnel/format/avro/AvroToRowConverter.java | 21 +- .../seatunnel/format/avro/AvroConverterTest.java | 222 +++++++++++++++++++++ 2 files changed, 242 insertions(+), 1 deletion(-) diff --git a/seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/AvroToRowConverter.java b/seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/AvroToRowConverter.java index ff746d0486..36f7210cae 100644 --- a/seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/AvroToRowConverter.java +++ b/seatunnel-formats/seatunnel-format-avro/src/main/java/org/apache/seatunnel/format/avro/AvroToRowConverter.java @@ -106,6 +106,19 @@ public class AvroToRowConverter implements Serializable { return datumReader; } + /** + * Converts an Avro {@link GenericRecord} into a {@link SeaTunnelRow}. + * + * <p>{@code BYTES} field values are interpreted according to the {@link ByteBuffer#position()} + * and {@link ByteBuffer#limit()} of the supplied buffer, so only the remaining logical bytes + * are read, and they are returned in a newly allocated array. The supplied buffer's position + * and limit are left untouched. Because the returned array never aliases the supplied buffer, a + * converted row keeps its own bytes even if the reader reuses that buffer for a later record. + * + * @param record the Avro record to convert + * @param rowType the target row type + * @return the converted row + */ public SeaTunnelRow converter(GenericRecord record, SeaTunnelRowType rowType) { String[] fieldNames = rowType.getFieldNames(); @@ -138,7 +151,13 @@ public class AvroToRowConverter implements Serializable { case TIMESTAMP: return val; case BYTES: - return ((ByteBuffer) val).array(); + ByteBuffer buffer = (ByteBuffer) val; + // Always copy: a reader may reuse this buffer for the next value, which would + // silently rewrite the bytes of an already converted row. duplicate() keeps the + // supplied buffer's position and limit unchanged. + byte[] bytes = new byte[buffer.remaining()]; + buffer.duplicate().get(bytes); + return bytes; case SMALLINT: return ((Integer) val).shortValue(); case TINYINT: diff --git a/seatunnel-formats/seatunnel-format-avro/src/test/java/org/apache/seatunnel/format/avro/AvroConverterTest.java b/seatunnel-formats/seatunnel-format-avro/src/test/java/org/apache/seatunnel/format/avro/AvroConverterTest.java index 32e35d756c..7402bfb337 100644 --- a/seatunnel-formats/seatunnel-format-avro/src/test/java/org/apache/seatunnel/format/avro/AvroConverterTest.java +++ b/seatunnel-formats/seatunnel-format-avro/src/test/java/org/apache/seatunnel/format/avro/AvroConverterTest.java @@ -24,29 +24,54 @@ import org.apache.seatunnel.api.table.type.BasicType; import org.apache.seatunnel.api.table.type.DecimalType; import org.apache.seatunnel.api.table.type.LocalTimeType; import org.apache.seatunnel.api.table.type.MapType; +import org.apache.seatunnel.api.table.type.PrimitiveByteArrayType; import org.apache.seatunnel.api.table.type.SeaTunnelDataType; import org.apache.seatunnel.api.table.type.SeaTunnelRow; import org.apache.seatunnel.api.table.type.SeaTunnelRowType; import org.apache.avro.generic.GenericRecord; +import org.apache.avro.io.BinaryDecoder; import org.apache.avro.io.BinaryEncoder; +import org.apache.avro.io.DatumReader; +import org.apache.avro.io.DecoderFactory; import org.apache.avro.io.EncoderFactory; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; import org.junit.jupiter.params.provider.ValueSource; import java.io.ByteArrayOutputStream; import java.io.IOException; import java.math.BigDecimal; +import java.nio.ByteBuffer; import java.time.LocalDate; import java.time.LocalDateTime; +import java.util.Arrays; import java.util.HashMap; import java.util.Map; +import java.util.stream.Stream; class AvroConverterTest { + private static final byte[] BYTES_PAYLOAD = new byte[] {10, 11, 12, 13, 14, 15, 16, 17}; + + // Reader-reuse tests: the first value is 64 bytes long, the second value is either the same + // length or shorter, so Avro reuses the previous record's ByteBuffer in both cases. + private static final byte[] FIRST_PAYLOAD = sequentialPayload(64, 1); + private static final byte[] SAME_LENGTH_PAYLOAD = sequentialPayload(64, 101); + private static final byte[] SHORTER_PAYLOAD = new byte[] {101, 102, 103}; + + private static byte[] sequentialPayload(int length, int firstValue) { + byte[] payload = new byte[length]; + for (int i = 0; i < length; i++) { + payload[i] = (byte) (firstValue + i); + } + return payload; + } + private SeaTunnelRow buildSeaTunnelRow() { SeaTunnelRow subSeaTunnelRow = new SeaTunnelRow(14); Map<String, String> map = new HashMap<String, String>(); @@ -183,6 +208,203 @@ class AvroConverterTest { Assertions.assertEquals(converterRow, seaTunnelRow); } + /** + * Row type with one real {@code BYTES} ({@code SqlType.BYTES}) field plus a scalar field, so + * the byte conversion can be asserted independently of the rest of the row. + */ + private static SeaTunnelRowType buildBytesRowType() { + return new SeaTunnelRowType( + new String[] {"c_bytes", "c_id"}, + new SeaTunnelDataType<?>[] {PrimitiveByteArrayType.INSTANCE, BasicType.INT_TYPE}); + } + + private static Stream<Arguments> byteBufferCases() { + return Stream.of( + Arguments.of( + "heap buffer: position 0, limit == capacity", + ByteBuffer.wrap(BYTES_PAYLOAD.clone()), + BYTES_PAYLOAD), + Arguments.of( + "heap buffer: bounded region with sentinel prefix and suffix", + boundedHeapBuffer(), + BYTES_PAYLOAD), + Arguments.of( + "heap buffer: slice with non-zero arrayOffset", + slicedHeapBuffer(), + BYTES_PAYLOAD), + Arguments.of("direct buffer", directBuffer(), BYTES_PAYLOAD), + Arguments.of("read-only buffer", readOnlyBuffer(), BYTES_PAYLOAD), + Arguments.of("empty heap buffer", ByteBuffer.allocate(0), new byte[0]), + Arguments.of("null value", null, null)); + } + + private static ByteBuffer boundedHeapBuffer() { + byte[] backing = new byte[BYTES_PAYLOAD.length + 8]; + Arrays.fill(backing, (byte) 0x7F); + System.arraycopy(BYTES_PAYLOAD, 0, backing, 4, BYTES_PAYLOAD.length); + ByteBuffer buffer = ByteBuffer.wrap(backing); + buffer.position(4); + buffer.limit(4 + BYTES_PAYLOAD.length); + return buffer; + } + + private static ByteBuffer slicedHeapBuffer() { + byte[] backing = new byte[BYTES_PAYLOAD.length + 6]; + Arrays.fill(backing, (byte) 0x7F); + System.arraycopy(BYTES_PAYLOAD, 0, backing, 3, BYTES_PAYLOAD.length); + ByteBuffer buffer = ByteBuffer.wrap(backing); + buffer.position(3); + buffer.limit(3 + BYTES_PAYLOAD.length); + return buffer.slice(); + } + + private static ByteBuffer directBuffer() { + ByteBuffer buffer = ByteBuffer.allocateDirect(BYTES_PAYLOAD.length); + buffer.put(BYTES_PAYLOAD); + buffer.flip(); + return buffer; + } + + private static ByteBuffer readOnlyBuffer() { + return ByteBuffer.wrap(BYTES_PAYLOAD.clone()).asReadOnlyBuffer(); + } + + @ParameterizedTest(name = "{0}") + @MethodSource("byteBufferCases") + public void testConverterBytesRespectsByteBufferPositionAndLimit( + String caseName, ByteBuffer buffer, byte[] expectedBytes) { + SeaTunnelRowType rowType = buildBytesRowType(); + RowToAvroConverter rowToAvroConverter = new RowToAvroConverter(rowType); + SeaTunnelRow sourceRow = new SeaTunnelRow(new Object[] {BYTES_PAYLOAD.clone(), 42}); + GenericRecord record = rowToAvroConverter.convertRowToGenericRecord(sourceRow); + record.put("c_bytes", buffer); + int position = buffer == null ? -1 : buffer.position(); + int limit = buffer == null ? -1 : buffer.limit(); + + AvroToRowConverter avroToRowConverter = new AvroToRowConverter(rowType); + SeaTunnelRow convertedRow = avroToRowConverter.converter(record, rowType); + + Assertions.assertEquals(42, convertedRow.getField(1)); + if (expectedBytes == null) { + Assertions.assertNull(convertedRow.getField(0), caseName + ": bytes field"); + } else { + Assertions.assertArrayEquals( + expectedBytes, (byte[]) convertedRow.getField(0), caseName + ": bytes field"); + } + if (buffer != null) { + Assertions.assertEquals(position, buffer.position(), "source ByteBuffer position"); + Assertions.assertEquals(limit, buffer.limit(), "source ByteBuffer limit"); + } + } + + @Test + public void testConverterBytesMustNotAliasWholeHeapBuffer() { + SeaTunnelRowType rowType = buildBytesRowType(); + byte[] payload = BYTES_PAYLOAD.clone(); + ByteBuffer sourceBuffer = ByteBuffer.wrap(payload.clone()); + RowToAvroConverter rowToAvroConverter = new RowToAvroConverter(rowType); + SeaTunnelRow sourceRow = new SeaTunnelRow(new Object[] {payload.clone(), 42}); + GenericRecord record = rowToAvroConverter.convertRowToGenericRecord(sourceRow); + record.put("c_bytes", sourceBuffer); + + AvroToRowConverter avroToRowConverter = new AvroToRowConverter(rowType); + SeaTunnelRow convertedRow = avroToRowConverter.converter(record, rowType); + Assertions.assertEquals(42, convertedRow.getField(1)); + + // A reader may reuse the buffer it handed out earlier, overwriting it in place. + sourceBuffer.array()[0] = (byte) 0x7F; + + Assertions.assertArrayEquals( + payload, + (byte[]) convertedRow.getField(0), + "converted bytes must not alias the source ByteBuffer"); + } + + private static Stream<Arguments> readerReuseCases() { + return Stream.of( + Arguments.of("equal-length reuse: 64 bytes then 64 bytes", SAME_LENGTH_PAYLOAD), + Arguments.of("shorter reuse: 64 bytes then 3 bytes", SHORTER_PAYLOAD)); + } + + @ParameterizedTest(name = "{0}") + @MethodSource("readerReuseCases") + public void testConverterBytesAfterAvroReaderRecordReuse(String caseName, byte[] nextPayload) + throws IOException { + SeaTunnelRowType rowType = buildBytesRowType(); + AvroSerializationSchema serializationSchema = new AvroSerializationSchema(rowType); + SeaTunnelRow firstSourceRow = new SeaTunnelRow(new Object[] {FIRST_PAYLOAD, 1}); + SeaTunnelRow secondSourceRow = new SeaTunnelRow(new Object[] {nextPayload, 2}); + byte[] firstMessage = serializationSchema.serialize(firstSourceRow); + byte[] secondMessage = serializationSchema.serialize(secondSourceRow); + + AvroToRowConverter avroToRowConverter = new AvroToRowConverter(rowType); + DatumReader<GenericRecord> reader = avroToRowConverter.getReader(); + + BinaryDecoder firstDecoder = DecoderFactory.get().binaryDecoder(firstMessage, null); + GenericRecord firstRecord = reader.read(null, firstDecoder); + ByteBuffer firstBuffer = (ByteBuffer) firstRecord.get("c_bytes"); + SeaTunnelRow firstRow = avroToRowConverter.converter(firstRecord, rowType); + Assertions.assertArrayEquals( + FIRST_PAYLOAD, (byte[]) firstRow.getField(0), caseName + ": first row bytes"); + + // Avro reuses the previous record's ByteBuffer when the new value fits its capacity. + BinaryDecoder secondDecoder = DecoderFactory.get().binaryDecoder(secondMessage, null); + GenericRecord reusedRecord = reader.read(firstRecord, secondDecoder); + // Precondition, not a value assertion: without record reuse this test proves nothing. + Assertions.assertSame( + firstBuffer, + reusedRecord.get("c_bytes"), + caseName + ": Avro must reuse the record's ByteBuffer"); + SeaTunnelRow reusedRow = avroToRowConverter.converter(reusedRecord, rowType); + + Assertions.assertEquals(2, reusedRow.getField(1)); + Assertions.assertArrayEquals( + nextPayload, (byte[]) reusedRow.getField(0), caseName + ": reused row bytes"); + // The retained row was converted before the reuse read, so it must keep its own bytes. + Assertions.assertArrayEquals( + FIRST_PAYLOAD, + (byte[]) firstRow.getField(0), + caseName + ": retained first row bytes after reuse"); + } + + @Test + public void testDeserializeRetainsFirstRowBytesAcrossMessages() throws IOException { + SeaTunnelRowType rowType = buildBytesRowType(); + CatalogTable catalogTable = CatalogTableUtil.getCatalogTable("", "", "", "test", rowType); + AvroSerializationSchema serializationSchema = new AvroSerializationSchema(rowType); + // Both messages carry 64-byte values, the case where a reusing reader aliases buffers. + SeaTunnelRow firstSourceRow = new SeaTunnelRow(new Object[] {FIRST_PAYLOAD, 1}); + SeaTunnelRow secondSourceRow = new SeaTunnelRow(new Object[] {SAME_LENGTH_PAYLOAD, 2}); + byte[] firstMessage = serializationSchema.serialize(firstSourceRow); + byte[] secondMessage = serializationSchema.serialize(secondSourceRow); + + AvroDeserializationSchema deserializationSchema = + new AvroDeserializationSchema(catalogTable); + SeaTunnelRow firstRow = deserializationSchema.deserialize(firstMessage); + SeaTunnelRow secondRow = deserializationSchema.deserialize(secondMessage); + + Assertions.assertEquals(2, secondRow.getField(1)); + Assertions.assertArrayEquals(SAME_LENGTH_PAYLOAD, (byte[]) secondRow.getField(0)); + Assertions.assertArrayEquals( + FIRST_PAYLOAD, + (byte[]) firstRow.getField(0), + "retained first row bytes after the second message"); + } + + @Test + public void testConverterBytesRoundTripThroughSerializationAndDeserialization() + throws IOException { + SeaTunnelRowType rowType = buildBytesRowType(); + CatalogTable catalogTable = CatalogTableUtil.getCatalogTable("", "", "", "test", rowType); + SeaTunnelRow sourceRow = new SeaTunnelRow(new Object[] {BYTES_PAYLOAD.clone(), 7}); + + byte[] message = new AvroSerializationSchema(rowType).serialize(sourceRow); + SeaTunnelRow resultRow = new AvroDeserializationSchema(catalogTable).deserialize(message); + + Assertions.assertEquals(7, resultRow.getField(1)); + Assertions.assertArrayEquals(BYTES_PAYLOAD, (byte[]) resultRow.getField(0)); + } + // The default mode is the legacy mode, so the GenericRecord shape built for a nested ROW must // stay exactly the shape that existed before the nullable ROW option was introduced: the nested // record still carries the historic top level record name and the same values. The explicit
