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 f3e462b7de [format] Fix reading Avro nullable unions when null value
is in second branch (#9418)
f3e462b7de is described below
commit f3e462b7def2d2d48af158b055212c586bd1349c
Author: Arnav Balyan <[email protected]>
AuthorDate: Fri Aug 28 13:51:27 2026 +0530
[format] Fix reading Avro nullable unions when null value is in second
branch (#9418)
---
.../paimon/format/avro/AvroRowDatumReader.java | 12 ++---
.../paimon/format/avro/FieldReaderFactory.java | 27 +++++++++--
.../paimon/format/avro/AvroFileFormatTest.java | 54 ++++++++++++++++++++++
3 files changed, 83 insertions(+), 10 deletions(-)
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroRowDatumReader.java
b/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroRowDatumReader.java
index c62777bf80..4fd5423a9a 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroRowDatumReader.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroRowDatumReader.java
@@ -36,7 +36,7 @@ public class AvroRowDatumReader implements
DatumReader<InternalRow> {
private final UriReader uriReader;
private RowReader reader;
- private boolean isUnion;
+ private int nullIndex;
public AvroRowDatumReader(RowType projectedRowType) {
this(projectedRowType, null);
@@ -49,10 +49,10 @@ public class AvroRowDatumReader implements
DatumReader<InternalRow> {
@Override
public void setSchema(Schema schema) {
- this.isUnion = false;
+ this.nullIndex = -1;
if (schema.isUnion()) {
- this.isUnion = true;
- schema = schema.getTypes().get(1);
+ this.nullIndex = FieldReaderFactory.nullableUnionNullIndex(schema);
+ schema = schema.getTypes().get(1 - nullIndex);
}
this.reader =
new FieldReaderFactory(uriReader)
@@ -61,9 +61,9 @@ public class AvroRowDatumReader implements
DatumReader<InternalRow> {
@Override
public InternalRow read(InternalRow reuse, Decoder in) throws IOException {
- if (isUnion) {
+ if (nullIndex >= 0) {
int index = in.readIndex();
- if (index == 0) {
+ if (index == nullIndex) {
throw new RuntimeException("Cannot read a null row.");
}
}
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/avro/FieldReaderFactory.java
b/paimon-format/src/main/java/org/apache/paimon/format/avro/FieldReaderFactory.java
index 6c5a4d018e..c4303ce371 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/avro/FieldReaderFactory.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/avro/FieldReaderFactory.java
@@ -118,7 +118,24 @@ public class FieldReaderFactory implements
AvroSchemaVisitor<FieldReader> {
@Override
public FieldReader visitUnion(Schema schema, @Nullable DataType type) {
- return new NullableReader(visit(schema.getTypes().get(1), type));
+ int nullIndex = nullableUnionNullIndex(schema);
+ return new NullableReader(visit(schema.getTypes().get(1 - nullIndex),
type), nullIndex);
+ }
+
+ static int nullableUnionNullIndex(Schema schema) {
+ List<Schema> types = schema.getTypes();
+ if (types.size() != 2) {
+ throw new IllegalArgumentException(
+ "Only nullable Avro unions are supported: " + schema);
+ }
+
+ if (types.get(0).getType() == Schema.Type.NULL) {
+ return 0;
+ } else if (types.get(1).getType() == Schema.Type.NULL) {
+ return 1;
+ }
+
+ throw new IllegalArgumentException("Only nullable Avro unions are
supported: " + schema);
}
@Override
@@ -227,21 +244,23 @@ public class FieldReaderFactory implements
AvroSchemaVisitor<FieldReader> {
private static class NullableReader implements FieldReader {
private final FieldReader reader;
+ private final int nullIndex;
- public NullableReader(FieldReader reader) {
+ public NullableReader(FieldReader reader, int nullIndex) {
this.reader = reader;
+ this.nullIndex = nullIndex;
}
@Override
public Object read(Decoder decoder, Object reuse) throws IOException {
int index = decoder.readIndex();
- return index == 0 ? null : reader.read(decoder, reuse);
+ return index == nullIndex ? null : reader.read(decoder, reuse);
}
@Override
public void skip(Decoder decoder) throws IOException {
int index = decoder.readIndex();
- if (index == 1) {
+ if (index != nullIndex) {
reader.skip(decoder);
}
}
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 0bcc799b3b..89605046b1 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
@@ -323,6 +323,60 @@ public class AvroFileFormatTest {
assertThat(values).containsExactly(42L, 21.0d, 10.5d);
}
+ @Test
+ void testReadNullableUnionWithNullSecond() throws IOException {
+ Schema nullableInt =
+ Schema.createUnion(
+ Arrays.asList(
+ Schema.create(Schema.Type.INT),
Schema.create(Schema.Type.NULL)));
+ Schema recordSchema =
+ SchemaBuilder.record("record")
+ .fields()
+ .name("skipped")
+ .type(nullableInt)
+ .noDefault()
+ .name("value")
+ .type(nullableInt)
+ .noDefault()
+ .endRecord();
+ Schema writerSchema =
+ Schema.createUnion(Arrays.asList(recordSchema,
Schema.create(Schema.Type.NULL)));
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path file = new Path(new Path(tempPath.toUri()),
UUID.randomUUID().toString());
+
+ try (PositionOutputStream out = fileIO.newOutputStream(file, false);
+ DataFileWriter<GenericRecord> writer =
+ new DataFileWriter<>(new
GenericDatumWriter<>(writerSchema))) {
+ writer.create(writerSchema, out);
+ GenericRecord value = new GenericData.Record(recordSchema);
+ value.put("skipped", 100);
+ value.put("value", 42);
+ writer.append(value);
+ GenericRecord nullValue = new GenericData.Record(recordSchema);
+ nullValue.put("skipped", null);
+ nullValue.put("value", null);
+ writer.append(nullValue);
+ }
+
+ RowType tableType =
+ RowType.builder()
+ .field("skipped", DataTypes.INT())
+ .field("value", DataTypes.INT())
+ .build();
+ RowType projectedType = RowType.builder().field("value",
DataTypes.INT()).build();
+ List<Integer> values = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
+ fileFormat
+ .createReaderFactory(tableType, projectedType, new
ArrayList<>())
+ .createReader(
+ new FormatReaderContext(
+ fileIO, file,
fileIO.getFileSize(file), null, null))) {
+ reader.forEachRemaining(row -> values.add(row.isNullAt(0) ? null :
row.getInt(0)));
+ }
+
+ assertThat(values).containsExactly(42, null);
+ }
+
@Test
void testReadsLargeZstdBlock() throws IOException {
RowType rowType =
