This is an automated email from the ASF dual-hosted git repository.
leaves12138 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 b3b0b43c3a [core] Support reordered fields in manifest Avro files
(#9683)
b3b0b43c3a is described below
commit b3b0b43c3a8e5fa4f870956495260b7e881c8d00
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Sep 8 14:11:12 2026 +0800
[core] Support reordered fields in manifest Avro files (#9683)
---
.../apache/paimon/manifest/ManifestAvroReader.java | 115 +++++++++--
.../apache/paimon/manifest/ManifestFileTest.java | 224 ++++++++++++++++-----
2 files changed, 271 insertions(+), 68 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 a61a65f900..b186335877 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
@@ -37,6 +37,7 @@ import java.io.IOException;
import java.io.InputStream;
import java.io.UncheckedIOException;
import java.nio.ByteBuffer;
+import java.util.Arrays;
import java.util.Iterator;
import java.util.NoSuchElementException;
@@ -162,6 +163,15 @@ public final class ManifestAvroReader implements
AutoCloseable {
ManifestEntry.FILE
};
+ private static final FieldType[] TOP_LEVEL_FIELD_TYPES = {
+ FieldType.INT,
+ FieldType.INT,
+ FieldType.BYTES,
+ FieldType.INT,
+ FieldType.INT,
+ FieldType.RECORD
+ };
+
private final int projectedFieldCount;
private final int versionPosition;
private final int kindPosition;
@@ -169,6 +179,8 @@ public final class ManifestAvroReader implements
AutoCloseable {
private final int bucketPosition;
private final int totalBucketsPosition;
private final int filePosition;
+ private final int[] fieldOrder;
+ private final boolean standardFieldOrder;
private final FieldDecoder fileReader;
private ManifestRecordDecoder(AvroRecordDecoder decoder, RowType
projectedType) {
@@ -181,13 +193,22 @@ public final class ManifestAvroReader implements
AutoCloseable {
this.totalBucketsPosition =
projectedType.getFieldIndex(ManifestEntry.TOTAL_BUCKETS);
this.filePosition =
projectedType.getFieldIndex(ManifestEntry.FILE);
- validateTopLevelFields(decoder);
- // Manifest v2 has a fixed top-level layout, but DataFileMeta has
gained nullable
- // fields. Build this reader from the writer schema so legacy
files with fewer nested
- // fields still decode and expose the missing projected fields as
null.
+ this.fieldOrder = validateTopLevelFields(decoder);
+ boolean standardFieldOrder = true;
+ int writerFilePosition = -1;
+ for (int i = 0; i < fieldOrder.length; i++) {
+ standardFieldOrder &= fieldOrder[i] == i;
+ if (fieldOrder[i] == 5) {
+ writerFilePosition = i;
+ }
+ }
+ this.standardFieldOrder = standardFieldOrder;
+ // Resolve the file field from the writer schema, including its
nested field order.
+ // Legacy files with fewer nested fields expose missing projected
fields as null.
fileReader =
decoder.createFieldDecoder(
- 5, filePosition >= 0 ?
projectedType.getTypeAt(filePosition) : null);
+ writerFilePosition,
+ filePosition >= 0 ?
projectedType.getTypeAt(filePosition) : null);
}
private boolean read(
@@ -199,6 +220,9 @@ public final class ManifestAvroReader implements
AutoCloseable {
if (!decoder.readRecordStart()) {
throw new IOException("Unexpected null or non-record Manifest
Avro value.");
}
+ if (!standardFieldOrder) {
+ return readReordered(decoder, row, partitionFilter,
bucketFilter);
+ }
int version = decoder.readInt();
ManifestEntrySerializer.checkFormatIdentifier(version);
@@ -258,6 +282,64 @@ public final class ManifestAvroReader implements
AutoCloseable {
return true;
}
+ private boolean readReordered(
+ AvroRecordDecoder decoder,
+ GenericRow row,
+ @Nullable PartitionPredicate partitionFilter,
+ @Nullable BucketFilter bucketFilter)
+ throws IOException {
+ boolean partitionNeededForFilter = partitionFilter != null ||
bucketFilter != null;
+ byte[] partitionBytes = null;
+ int bucket = 0;
+ int totalBuckets = 0;
+ // Avro values follow writer field order. Filter fields may occur
after the file,
+ // so this compatibility path applies filters after consuming the
complete record.
+ for (int field : fieldOrder) {
+ switch (field) {
+ case 0:
+ int version = decoder.readInt();
+ ManifestEntrySerializer.checkFormatIdentifier(version);
+ setProjected(row, versionPosition, version);
+ break;
+ case 1:
+ setProjected(row, kindPosition, (byte)
decoder.readInt());
+ break;
+ case 2:
+ if (partitionPosition >= 0 ||
partitionNeededForFilter) {
+ partitionBytes = decoder.readBytes();
+ } else {
+ decoder.skipBytes();
+ }
+ setProjected(row, partitionPosition, partitionBytes);
+ break;
+ case 3:
+ bucket = decoder.readInt();
+ setProjected(row, bucketPosition, bucket);
+ break;
+ case 4:
+ totalBuckets = decoder.readInt();
+ setProjected(row, totalBucketsPosition, totalBuckets);
+ break;
+ case 5:
+ if (filePosition >= 0) {
+ row.setField(
+ filePosition,
+ fileReader.read(decoder,
row.getField(filePosition)));
+ } else {
+ fileReader.skip(decoder);
+ }
+ break;
+ default:
+ throw new IllegalStateException("Unexpected Manifest
field: " + field);
+ }
+ }
+
+ BinaryRow partition =
+ partitionNeededForFilter ?
deserializeBinaryRow(partitionBytes) : null;
+ return (partitionFilter == null || partitionFilter.test(partition))
+ && (bucketFilter == null || bucketFilter.test(partition,
bucket, totalBuckets));
+ }
+
private void skipBucketAndFile(AvroRecordDecoder decoder) throws
IOException {
decoder.readInt();
decoder.readInt();
@@ -275,30 +357,27 @@ public final class ManifestAvroReader implements
AutoCloseable {
}
}
- private static void validateTopLevelFields(AvroRecordDecoder decoder) {
+ private static int[] validateTopLevelFields(AvroRecordDecoder decoder)
{
if (decoder.fieldCount() != TOP_LEVEL_FIELDS.length) {
throw new IllegalArgumentException(
String.format(
"Manifest Avro schema has %s top-level fields,
expected %s.",
decoder.fieldCount(),
TOP_LEVEL_FIELDS.length));
}
- for (int i = 0; i < TOP_LEVEL_FIELDS.length; i++) {
+ int[] fieldOrder = new int[TOP_LEVEL_FIELDS.length];
+ for (int i = 0; i < fieldOrder.length; i++) {
String actual = decoder.fieldName(i);
- String expected = TOP_LEVEL_FIELDS[i];
- if (!expected.equals(actual)) {
+ int field = Arrays.asList(TOP_LEVEL_FIELDS).indexOf(actual);
+ if (field < 0) {
throw new IllegalArgumentException(
String.format(
- "Unexpected Manifest Avro field at
position %s: expected %s but found %s.",
- i, expected, actual));
+ "Unexpected Manifest Avro field at
position %s: %s.",
+ i, actual));
}
+ validateFieldType(decoder, i, TOP_LEVEL_FIELD_TYPES[field]);
+ fieldOrder[i] = field;
}
-
- validateFieldType(decoder, 0, FieldType.INT);
- validateFieldType(decoder, 1, FieldType.INT);
- validateFieldType(decoder, 2, FieldType.BYTES);
- validateFieldType(decoder, 3, FieldType.INT);
- validateFieldType(decoder, 4, FieldType.INT);
- validateFieldType(decoder, 5, FieldType.RECORD);
+ return fieldOrder;
}
private static void validateFieldType(
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 6587f1e273..38194fd2a8 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
@@ -44,6 +44,9 @@ import org.apache.paimon.utils.FileStorePathFactory;
import org.junit.jupiter.api.RepeatedTest;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
import java.io.IOException;
import java.io.UncheckedIOException;
@@ -56,6 +59,7 @@ import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ThreadLocalRandom;
import java.util.stream.Collectors;
+import java.util.stream.Stream;
import static org.apache.paimon.TestKeyValueGenerator.DEFAULT_PART_TYPE;
import static
org.apache.paimon.stats.StatsTestUtils.convertWithoutSchemaEvolution;
@@ -434,8 +438,10 @@ public class ManifestFileTest {
}
}
- @Test
- void testProjectedScanRejectsUnsupportedFormatIdentifier() throws
Exception {
+ @ParameterizedTest
+ @MethodSource("reorderedManifestFieldOrders")
+ void testProjectedScanRejectsUnsupportedFormatIdentifier(
+ int[] fieldOrder, boolean reorderNestedFields) throws Exception {
ManifestEntry entry = gen.next();
ManifestFile manifestFile = createManifestFile(tempDir.toString(),
Long.MAX_VALUE);
ManifestFileMeta manifest =
@@ -443,20 +449,13 @@ public class ManifestFileTest {
Path path = new Path(new Path(tempDir.toUri()), "manifest/" +
manifest.fileName());
LocalFileIO fileIO = LocalFileIO.create();
ManifestEntrySerializer serializer = new ManifestEntrySerializer();
- InternalRow valid = serializer.toRow(entry);
+ GenericRow invalid = (GenericRow) serializer.toRow(entry);
+ invalid.setField(0, 1);
+ RowType writerType = reorderedManifestType(fieldOrder,
reorderNestedFields);
try (PositionOutputStream out = fileIO.newOutputStream(path, true);
- FormatWriter writer =
-
avro.createWriterFactory(ManifestEntry.MANIFEST_ROW_TYPE)
- .create(out, "zstd")) {
- writer.addElement(
- GenericRow.of(
- 1,
- valid.getByte(1),
- valid.getBinary(2),
- valid.getInt(3),
- valid.getInt(4),
- valid.getRow(5,
DataFileMeta.SCHEMA.getFieldCount())));
+ FormatWriter writer =
avro.createWriterFactory(writerType).create(out, "zstd")) {
+ writer.addElement(reorderRow(invalid,
ManifestEntry.MANIFEST_ROW_TYPE, writerType));
}
try (CloseableIterator<ProjectedManifestEntry> entries =
@@ -467,6 +466,18 @@ public class ManifestFileTest {
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("not compatible");
}
+
+ // An unprojected version must still be checked even when every row is
filtered out.
+ try (ManifestAvroReader reader = new
ManifestAvroReader(fileIO.newInputStream(path));
+ CloseableIterator<InternalRow> rows =
+ reader.read(
+ new RowType(false, Collections.emptyList()),
+ null,
+ new BucketFilter(false, null, bucket -> false,
null))) {
+ assertThatThrownBy(rows::hasNext)
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("not compatible");
+ }
}
@Test
@@ -575,49 +586,162 @@ public class ManifestFileTest {
assertThat(actual.file().columnMaxSequenceNumbers()).isNull();
}
- @Test
- void testAvroReaderRejectsReorderedTopLevelFields() throws Exception {
- ManifestEntry entry = gen.next();
- List<DataField> fields = ManifestEntry.MANIFEST_ROW_TYPE.getFields();
- RowType reorderedType =
- new RowType(
- false,
- Arrays.asList(
- fields.get(0),
- fields.get(5),
- fields.get(1),
- fields.get(2),
- fields.get(3),
- fields.get(4)));
+ @ParameterizedTest
+ @MethodSource("reorderedManifestFieldOrders")
+ void testAvroReaderSupportsReorderedWriterFields(int[] fieldOrder, boolean
reorderNestedFields)
+ throws Exception {
+ List<ManifestEntry> entries = generateData();
+ RowType writerType = reorderedManifestType(fieldOrder,
reorderNestedFields);
Path path = new Path(new Path(tempDir.toUri()),
"reordered-manifest.avro");
LocalFileIO fileIO = LocalFileIO.create();
ManifestEntrySerializer serializer = new ManifestEntrySerializer();
-
try (PositionOutputStream out = fileIO.newOutputStream(path, false);
- FormatWriter writer =
avro.createWriterFactory(reorderedType).create(out, "zstd")) {
- InternalRow row = serializer.toRow(entry);
- writer.addElement(
- GenericRow.of(
- row.getInt(0),
- row.getRow(5, DataFileMeta.SCHEMA.getFieldCount()),
- row.getByte(1),
- row.getBinary(2),
- row.getInt(3),
- row.getInt(4)));
+ FormatWriter writer =
avro.createWriterFactory(writerType).create(out, "zstd")) {
+ for (ManifestEntry entry : entries) {
+ writer.addElement(
+ reorderRow(
+ serializer.toRow(entry),
+ ManifestEntry.MANIFEST_ROW_TYPE,
+ writerType));
+ }
}
- assertThatThrownBy(
- () -> {
- try (ManifestAvroReader reader =
- new
ManifestAvroReader(fileIO.newInputStream(path));
- CloseableIterator<InternalRow> rows =
- reader.read(
-
ManifestEntry.MANIFEST_ROW_TYPE, null, null)) {
- rows.hasNext();
+ boolean rawCopySupported =
+ !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 =
+ reader.read(ManifestEntry.MANIFEST_ROW_TYPE, null,
null)) {
+
assertThat(reader.rawBlockCopySupported()).isEqualTo(rawCopySupported);
+ while (rows.hasNext()) {
+ retained.add(rows.next());
+ }
+ }
+
assertThat(retained.stream().map(serializer::fromRow).collect(Collectors.toList()))
+ .containsExactlyElementsOf(entries);
+
+ // Raw blocks must also decode into canonical rows, including when
reusing a row.
+ List<ManifestEntry> decoded = new ArrayList<>();
+ try (ManifestAvroReader reader = new
ManifestAvroReader(fileIO.newInputStream(path))) {
+ while (reader.hasNext()) {
+ ManifestAvroReader.RawBlock block = reader.next().stableCopy();
+
assertThat(block.rawBlockCopySupported()).isEqualTo(rawCopySupported);
+ ManifestAvroReader.RowIterator rows =
block.toRows(ManifestEntry.MANIFEST_ROW_TYPE);
+ while (rows.hasNext()) {
+ decoded.add(serializer.fromRow(rows.next()));
+ }
+ }
+ }
+ assertThat(decoded).containsExactlyElementsOf(entries);
+
+ ManifestEntry selected = entries.get(0);
+ PartitionPredicate partitionFilter =
+ PartitionPredicate.fromMultiple(
+ DEFAULT_PART_TYPE,
Collections.singletonList(selected.partition()));
+ BucketFilter bucketFilter =
+ new BucketFilter(
+ false,
+ null,
+ null,
+ (partition, bucket, totalBuckets) ->
+ partition.equals(selected.partition())
+ && bucket == selected.bucket()
+ && totalBuckets ==
selected.totalBuckets());
+ List<DataField> fields = ManifestEntry.MANIFEST_ROW_TYPE.getFields();
+ RowType fileProjection =
+ new RowType(
+ false,
+ Collections.singletonList(
+ fields.get(5)
+ .newType(
+ DataFileMeta.SCHEMA.project(
+ DataFileMeta.ROW_COUNT,
+
DataFileMeta.FILE_NAME))));
+ RowType kindProjection = new RowType(false,
Collections.singletonList(fields.get(1)));
+ // Filter fields need not be projected, and unprojected file metadata
must be skipped.
+ for (RowType projectedType : Arrays.asList(fileProjection,
kindProjection)) {
+ for (PartitionPredicate filter : Arrays.asList(null,
partitionFilter)) {
+ for (BucketFilter buckets : Arrays.asList(null, bucketFilter))
{
+ List<ManifestEntry> expected =
+ entries.stream()
+ .filter(e -> filter == null ||
filter.test(e.partition()))
+ .filter(
+ e ->
+ buckets == null
+ || buckets.test(
+
e.partition(),
+ e.bucket(),
+
e.totalBuckets()))
+ .collect(Collectors.toList());
+ try (ManifestAvroReader reader =
+ new
ManifestAvroReader(fileIO.newInputStream(path));
+ CloseableIterator<InternalRow> rows =
+ reader.read(projectedType, filter,
buckets)) {
+ for (ManifestEntry entry : expected) {
+ assertThat(rows.hasNext()).isTrue();
+ InternalRow row = rows.next();
+ if (projectedType == kindProjection) {
+
assertThat(row.getByte(0)).isEqualTo(entry.kind().toByteValue());
+ } else {
+ InternalRow file = row.getRow(0, 2);
+
assertThat(file.getLong(0)).isEqualTo(entry.rowCount());
+ assertThat(file.getString(1).toString())
+ .isEqualTo(entry.fileName());
}
- })
- .isInstanceOf(IllegalArgumentException.class)
- .hasMessageContaining("expected _KIND but found _FILE");
+ }
+ assertThat(rows.hasNext()).isFalse();
+ }
+ }
+ }
+ }
+ }
+
+ 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[] {5, 4, 3, 2, 1, 0},
+ new int[] {3, 0, 5, 1, 4, 2},
+ new int[] {1, 0, 2, 4, 3, 5})
+ .flatMap(order -> Stream.of(Arguments.of(order, false),
Arguments.of(order, true)));
+ }
+
+ private static RowType reorderedManifestType(int[] fieldOrder, boolean
reorderNestedFields) {
+ List<DataField> fileFields = new
ArrayList<>(DataFileMeta.SCHEMA.getFields());
+ Collections.reverse(fileFields);
+ List<DataField> fields = ManifestEntry.MANIFEST_ROW_TYPE.getFields();
+ return new RowType(
+ false,
+ Arrays.stream(fieldOrder)
+ .mapToObj(fields::get)
+ .map(
+ field ->
+ reorderNestedFields
+ &&
ManifestEntry.FILE.equals(field.name())
+ ? field.newType(new
RowType(false, fileFields))
+ : field)
+ .collect(Collectors.toList()));
+ }
+
+ private static GenericRow reorderRow(InternalRow row, RowType sourceType,
RowType targetType) {
+ GenericRow result = new GenericRow(targetType.getFieldCount());
+ for (int i = 0; i < targetType.getFieldCount(); i++) {
+ DataField field = targetType.getFields().get(i);
+ int sourcePosition = sourceType.getFieldIndex(field.name());
+ Object value =
+ InternalRow.createFieldGetter(
+ sourceType.getTypeAt(sourcePosition),
sourcePosition)
+ .getFieldOrNull(row);
+ if (value != null && field.type() instanceof RowType) {
+ value =
+ reorderRow(
+ (InternalRow) value,
+ (RowType) sourceType.getTypeAt(sourcePosition),
+ (RowType) field.type());
+ }
+ result.setField(i, value);
+ }
+ return result;
}
@RepeatedTest(10)
