This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 0da661506a [core] Fix nullable file records in the manifest Avro
reader (#9816)
0da661506a is described below
commit 0da661506aefb3b038a8493613c3ddc28dec012c
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 =
