This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit e84718606592d2d8108c7550a6081c6e0b73c63f Author: YeJunHao <[email protected]> AuthorDate: Mon Sep 14 17:53:31 2026 +0800 [core] Fix nullable file records in the manifest Avro reader (#9816) --- .../apache/paimon/manifest/ManifestAvroReader.java | 5 +- .../apache/paimon/manifest/ManifestFileTest.java | 115 ++++++++++++++++++++- .../paimon/format/avro/AvroRecordDecoder.java | 10 ++ .../paimon/format/avro/AvroFileFormatTest.java | 69 +++++++++++++ 4 files changed, 194 insertions(+), 5 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroReader.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroReader.java index b186335877..995834c37e 100644 --- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroReader.java +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroReader.java @@ -382,7 +382,10 @@ public final class ManifestAvroReader implements AutoCloseable { private static void validateFieldType( AvroRecordDecoder decoder, int position, FieldType expectedType) { - FieldType actualType = decoder.fieldType(position); + FieldType actualType = + expectedType == FieldType.RECORD + ? decoder.nonNullFieldType(position) + : decoder.fieldType(position); if (actualType != expectedType) { throw new IllegalArgumentException( String.format( diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java index c85d82bf9c..d8eab09df7 100644 --- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java @@ -37,6 +37,8 @@ import org.apache.paimon.partition.PartitionPredicate; import org.apache.paimon.schema.FileSystemSchemaManager; import org.apache.paimon.stats.StatsTestUtils; import org.apache.paimon.types.DataField; +import org.apache.paimon.types.DataType; +import org.apache.paimon.types.DataTypes; import org.apache.paimon.types.RowType; import org.apache.paimon.utils.CloseableIterator; import org.apache.paimon.utils.FailingFileIO; @@ -627,11 +629,14 @@ public class ManifestFileTest { } @ParameterizedTest - @MethodSource("reorderedManifestFieldOrders") - void testAvroReaderSupportsReorderedWriterFields(int[] fieldOrder, boolean reorderNestedFields) - throws Exception { + @MethodSource("manifestWriterSchemas") + void testAvroReaderSupportsReorderedWriterFields( + int[] fieldOrder, boolean reorderNestedFields, boolean nullableFile) throws Exception { List<ManifestEntry> entries = generateData(); RowType writerType = reorderedManifestType(fieldOrder, reorderNestedFields); + if (nullableFile) { + writerType = nullableFileType(writerType); + } Path path = new Path(new Path(tempDir.toUri()), "reordered-manifest.avro"); LocalFileIO fileIO = LocalFileIO.create(); ManifestEntrySerializer serializer = new ManifestEntrySerializer(); @@ -647,7 +652,9 @@ public class ManifestFileTest { } boolean rawCopySupported = - !reorderNestedFields && Arrays.equals(fieldOrder, new int[] {0, 1, 2, 3, 4, 5}); + !nullableFile + && !reorderNestedFields + && Arrays.equals(fieldOrder, new int[] {0, 1, 2, 3, 4, 5}); List<InternalRow> retained = new ArrayList<>(); try (ManifestAvroReader reader = new ManifestAvroReader(fileIO.newInputStream(path)); CloseableIterator<InternalRow> rows = @@ -736,10 +743,110 @@ public class ManifestFileTest { } } + @Test + void testNullableFileRecordWithManifestCache() throws Exception { + List<ManifestEntry> entries = generateData(); + LocalFileIO fileIO = LocalFileIO.create(); + Path path = new Path(new Path(tempDir.toString()), "manifest/nullable-manifest.avro"); + fileIO.mkdirs(path.getParent()); + ManifestEntrySerializer serializer = new ManifestEntrySerializer(); + try (PositionOutputStream out = fileIO.newOutputStream(path, false); + FormatWriter writer = + avro.createWriterFactory(nullableFileType(ManifestEntry.MANIFEST_ROW_TYPE)) + .create(out, "zstd")) { + for (ManifestEntry entry : entries) { + writer.addElement(serializer.toRow(entry)); + } + } + + SegmentsCache<Path> cache = + new SegmentsCache<>(32 * 1024, new MemorySize(4 * 1024 * 1024), 1024 * 1024); + ManifestFile manifestFile = createManifestFile(tempDir.toString(), Long.MAX_VALUE, cache); + assertThat(manifestFile.read(path.getName())).containsExactlyInAnyOrderElementsOf(entries); + assertThat(cache.getIfPresents(path)).isNotNull(); + assertThat(manifestFile.read(path.getName())).containsExactlyInAnyOrderElementsOf(entries); + } + + @ParameterizedTest + @MethodSource("invalidManifestFieldTypes") + void testRejectsInvalidManifestFieldTypes(String fieldName, DataType fieldType) + throws Exception { + RowType writerType = + new RowType( + false, + ManifestEntry.MANIFEST_ROW_TYPE.getFields().stream() + .map( + field -> + field.name().equals(fieldName) + ? field.newType(fieldType) + : field) + .collect(Collectors.toList())); + GenericRow row = (GenericRow) new ManifestEntrySerializer().toRow(gen.next()); + if (ManifestEntry.FILE.equals(fieldName)) { + row.setField(writerType.getFieldIndex(fieldName), 42); + } + LocalFileIO fileIO = LocalFileIO.create(); + Path path = new Path(new Path(tempDir.toUri()), "invalid-field.avro"); + try (PositionOutputStream out = fileIO.newOutputStream(path, false); + FormatWriter writer = avro.createWriterFactory(writerType).create(out, "zstd")) { + writer.addElement(row); + } + + for (RowType projectedType : + Arrays.asList( + ManifestEntry.MANIFEST_ROW_TYPE, + ManifestEntry.MANIFEST_ROW_TYPE.project(ManifestEntry.KIND))) { + try (ManifestAvroReader reader = new ManifestAvroReader(fileIO.newInputStream(path)); + CloseableIterator<InternalRow> rows = reader.read(projectedType, null, null)) { + assertThatThrownBy(rows::hasNext) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining( + "Unexpected Manifest Avro type for field " + fieldName); + } + } + } + + private static Stream<Arguments> invalidManifestFieldTypes() { + return Stream.of( + Arguments.of(ManifestEntry.FILE, DataTypes.INT().notNull()), + Arguments.of(ManifestEntry.FILE, DataTypes.INT()), + Arguments.of(ManifestEntry.BUCKET, DataTypes.INT()), + Arguments.of("_VERSION", DataTypes.INT()), + Arguments.of( + ManifestEntry.PARTITION, + ManifestEntry.MANIFEST_ROW_TYPE + .getField(ManifestEntry.PARTITION) + .type() + .copy(true))); + } + + private static RowType nullableFileType(RowType type) { + return new RowType( + type.isNullable(), + type.getFields().stream() + .map( + field -> + ManifestEntry.FILE.equals(field.name()) + ? field.newType(field.type().copy(true)) + : field) + .collect(Collectors.toList())); + } + + private static Stream<Arguments> manifestWriterSchemas() { + return reorderedManifestFieldOrders() + .flatMap( + arguments -> + Stream.of( + Arguments.of(arguments.get()[0], arguments.get()[1], false), + Arguments.of( + arguments.get()[0], arguments.get()[1], true))); + } + private static Stream<Arguments> reorderedManifestFieldOrders() { return Stream.of( new int[] {0, 1, 2, 3, 4, 5}, new int[] {0, 5, 1, 2, 3, 4}, + new int[] {1, 2, 3, 4, 5, 0}, new int[] {5, 4, 3, 2, 1, 0}, new int[] {3, 0, 5, 1, 4, 2}, new int[] {1, 0, 2, 4, 3, 5}) diff --git a/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroRecordDecoder.java b/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroRecordDecoder.java index 48788ad651..3fd6cb3d72 100644 --- a/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroRecordDecoder.java +++ b/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroRecordDecoder.java @@ -93,6 +93,16 @@ public final class AvroRecordDecoder { return FieldType.valueOf(recordSchema.getFields().get(position).schema().getType().name()); } + /** Returns the field type after resolving a nullable union, without changing its decoder. */ + public FieldType nonNullFieldType(int position) { + Schema schema = recordSchema.getFields().get(position).schema(); + if (schema.getType() == Schema.Type.UNION) { + int nullIndex = FieldReaderFactory.nullableUnionNullIndex(schema); + schema = schema.getTypes().get(1 - nullIndex); + } + return FieldType.valueOf(schema.getType().name()); + } + /** Creates a decoder for one writer field. */ public FieldDecoder createFieldDecoder(int position, @Nullable DataType readType) { FieldReader reader = diff --git a/paimon-format/src/test/java/org/apache/paimon/format/avro/AvroFileFormatTest.java b/paimon-format/src/test/java/org/apache/paimon/format/avro/AvroFileFormatTest.java index 89605046b1..c7ca45b88f 100644 --- a/paimon-format/src/test/java/org/apache/paimon/format/avro/AvroFileFormatTest.java +++ b/paimon-format/src/test/java/org/apache/paimon/format/avro/AvroFileFormatTest.java @@ -377,6 +377,75 @@ public class AvroFileFormatTest { assertThat(values).containsExactly(42, null); } + @Test + void testNonNullFieldTypePreservesNullableRecordDecoding() throws IOException { + Schema fileSchema = SchemaBuilder.record("file").fields().requiredInt("id").endRecord(); + Schema nullSchema = Schema.create(Schema.Type.NULL); + for (Schema fieldSchema : + Arrays.asList( + fileSchema, + Schema.createUnion(Arrays.asList(nullSchema, fileSchema)), + Schema.createUnion(Arrays.asList(fileSchema, nullSchema)))) { + Schema writerSchema = + SchemaBuilder.record("manifest") + .fields() + .name("_FILE") + .type(fieldSchema) + .noDefault() + .endRecord(); + AvroRecordDecoder decoder = new AvroRecordDecoder(writerSchema); + AvroRecordDecoder.FieldType rawType = decoder.fieldType(0); + assertThat(decoder.nonNullFieldType(0)).isEqualTo(AvroRecordDecoder.FieldType.RECORD); + assertThat(decoder.fieldType(0)).isEqualTo(rawType); + + GenericRecord file = new GenericData.Record(fileSchema); + file.put("id", 42); + GenericRecord manifest = new GenericData.Record(writerSchema); + manifest.put("_FILE", file); + ByteArrayOutputStream output = new ByteArrayOutputStream(); + BinaryEncoder encoder = EncoderFactory.get().binaryEncoder(output, null); + new GenericDatumWriter<GenericRecord>(writerSchema).write(manifest, encoder); + encoder.flush(); + byte[] bytes = output.toByteArray(); + AvroRecordDecoder.FieldDecoder field = + decoder.createFieldDecoder( + 0, RowType.builder().field("id", DataTypes.INT().notNull()).build()); + decoder.reset(bytes, 0, bytes.length); + assertThat(decoder.readRecordStart()).isTrue(); + assertThat(((InternalRow) field.read(decoder, null)).getInt(0)).isEqualTo(42); + assertThat(decoder.isEnd()).isTrue(); + + decoder.reset(bytes, 0, bytes.length); + assertThat(decoder.readRecordStart()).isTrue(); + field.skip(decoder); + assertThat(decoder.isEnd()).isTrue(); + } + } + + @Test + void testNonNullFieldTypeRejectsNonNullableUnions() { + Schema record = SchemaBuilder.record("file").fields().requiredInt("id").endRecord(); + Schema integer = Schema.create(Schema.Type.INT); + for (Schema union : + Arrays.asList( + Schema.createUnion(Arrays.asList(record, integer)), + Schema.createUnion( + Arrays.asList(Schema.create(Schema.Type.NULL), record, integer)))) { + Schema writerSchema = + SchemaBuilder.record("manifest") + .fields() + .name("_FILE") + .type(union) + .noDefault() + .endRecord(); + AvroRecordDecoder decoder = new AvroRecordDecoder(writerSchema); + assertThat(decoder.fieldType(0)).isEqualTo(AvroRecordDecoder.FieldType.UNION); + assertThatThrownBy(() -> decoder.nonNullFieldType(0)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Only nullable Avro unions are supported"); + } + } + @Test void testReadsLargeZstdBlock() throws IOException { RowType rowType =
