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 f97a68eb860f5aca53560c7db22f6ef671c428cf Author: Jingsong Lee <[email protected]> AuthorDate: Fri Sep 11 23:07:21 2026 +0800 [core][python] Add extra files to manifest metadata (#9746) --- docs/docs/concepts/spec/manifest.md | 4 + .../apache/paimon/manifest/ManifestFileMeta.java | 55 +++++++++- .../manifest/ManifestFileMetaSerializer.java | 9 +- .../apache/paimon/operation/FileDeletionBase.java | 3 + .../apache/paimon/operation/OrphanFilesClean.java | 5 + .../manifest/ManifestFileMetaSerializerTest.java | 44 ++++++++ .../apache/paimon/manifest/ManifestListTest.java | 30 +++++- .../paimon/operation/ExpireSnapshotsTest.java | 114 ++++++++++++++++++++- .../operation/LocalOrphanFilesCleanTest.java | 55 ++++++++++ .../pypaimon/manifest/manifest_list_manager.py | 2 + .../pypaimon/manifest/schema/manifest_file_meta.py | 4 +- .../tests/manifest/manifest_manager_test.py | 40 ++++++++ .../tests/manifest/manifest_schema_test.py | 10 +- 13 files changed, 364 insertions(+), 11 deletions(-) diff --git a/docs/docs/concepts/spec/manifest.md b/docs/docs/concepts/spec/manifest.md index ac216f753e..6ad271bb5a 100644 --- a/docs/docs/concepts/spec/manifest.md +++ b/docs/docs/concepts/spec/manifest.md @@ -58,6 +58,10 @@ skip manifests before opening them. | `_MIN_BUCKET`, `_MAX_BUCKET` | INT, nullable | Bucket bounds in the manifest. | | `_MIN_LEVEL`, `_MAX_LEVEL` | INT, nullable | Data-file level bounds in the manifest. | | `_MIN_ROW_ID`, `_MAX_ROW_ID` | BIGINT, nullable | Row-ID bounds when available. | +| `_EXTRA_FILES` | ARRAY of STRING, nullable | Names of additional files in the manifest directory; defaults to null. | + +Each extra file belongs exclusively to one manifest. It is retained and cleaned up together with +that manifest during snapshot, tag, or changelog deletion. ## Manifest diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMeta.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMeta.java index 4a66a3fb0d..2123a419a5 100644 --- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMeta.java +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMeta.java @@ -20,6 +20,7 @@ package org.apache.paimon.manifest; import org.apache.paimon.annotation.Public; import org.apache.paimon.stats.SimpleStats; +import org.apache.paimon.types.ArrayType; import org.apache.paimon.types.BigIntType; import org.apache.paimon.types.DataField; import org.apache.paimon.types.IntType; @@ -57,7 +58,12 @@ public class ManifestFileMeta { new DataField(8, "_MIN_LEVEL", new IntType(true)), new DataField(9, "_MAX_LEVEL", new IntType(true)), new DataField(10, "_MIN_ROW_ID", new BigIntType(true)), - new DataField(11, "_MAX_ROW_ID", new BigIntType(true)))); + new DataField(11, "_MAX_ROW_ID", new BigIntType(true)), + new DataField( + 12, + "_EXTRA_FILES", + new ArrayType( + true, new VarCharType(false, Integer.MAX_VALUE))))); private final String fileName; private final long fileSize; @@ -71,6 +77,7 @@ public class ManifestFileMeta { private final @Nullable Integer maxLevel; private final @Nullable Long minRowId; private final @Nullable Long maxRowId; + private final @Nullable List<String> extraFiles; public ManifestFileMeta( String fileName, @@ -85,6 +92,36 @@ public class ManifestFileMeta { @Nullable Integer maxLevel, @Nullable Long minRowId, @Nullable Long maxRowId) { + this( + fileName, + fileSize, + numAddedFiles, + numDeletedFiles, + partitionStats, + schemaId, + minBucket, + maxBucket, + minLevel, + maxLevel, + minRowId, + maxRowId, + null); + } + + public ManifestFileMeta( + String fileName, + long fileSize, + long numAddedFiles, + long numDeletedFiles, + SimpleStats partitionStats, + long schemaId, + @Nullable Integer minBucket, + @Nullable Integer maxBucket, + @Nullable Integer minLevel, + @Nullable Integer maxLevel, + @Nullable Long minRowId, + @Nullable Long maxRowId, + @Nullable List<String> extraFiles) { this.fileName = fileName; this.fileSize = fileSize; this.numAddedFiles = numAddedFiles; @@ -97,6 +134,7 @@ public class ManifestFileMeta { this.maxLevel = maxLevel; this.minRowId = minRowId; this.maxRowId = maxRowId; + this.extraFiles = extraFiles; } public String fileName() { @@ -147,6 +185,10 @@ public class ManifestFileMeta { return maxRowId; } + public @Nullable List<String> extraFiles() { + return extraFiles; + } + @Override public boolean equals(Object o) { if (!(o instanceof ManifestFileMeta)) { @@ -164,7 +206,8 @@ public class ManifestFileMeta { && Objects.equals(minLevel, that.minLevel) && Objects.equals(maxLevel, that.maxLevel) && Objects.equals(minRowId, that.minRowId) - && Objects.equals(maxRowId, that.maxRowId); + && Objects.equals(maxRowId, that.maxRowId) + && Objects.equals(extraFiles, that.extraFiles); } @Override @@ -181,13 +224,14 @@ public class ManifestFileMeta { minLevel, maxLevel, minRowId, - maxRowId); + maxRowId, + extraFiles); } @Override public String toString() { return String.format( - "{%s, %d, %d, %d, %s, %d, %s, %s, %s, %s, %s, %s}", + "{%s, %d, %d, %d, %s, %d, %s, %s, %s, %s, %s, %s, %s}", fileName, fileSize, numAddedFiles, @@ -199,7 +243,8 @@ public class ManifestFileMeta { minLevel, maxLevel, minRowId, - maxRowId); + maxRowId, + extraFiles); } // ----------------------- Serialization ----------------------------- diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMetaSerializer.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMetaSerializer.java index 4c07493242..4c2ab90c96 100644 --- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMetaSerializer.java +++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMetaSerializer.java @@ -25,6 +25,9 @@ import org.apache.paimon.stats.SimpleStats; import org.apache.paimon.utils.ObjectSerializer; import org.apache.paimon.utils.OffsetRow; +import static org.apache.paimon.utils.InternalRowUtils.fromStringArrayData; +import static org.apache.paimon.utils.InternalRowUtils.toStringArrayData; + /** Serializer for {@link ManifestFileMeta}. */ public class ManifestFileMetaSerializer extends ObjectSerializer<ManifestFileMeta> { @@ -56,7 +59,8 @@ public class ManifestFileMetaSerializer extends ObjectSerializer<ManifestFileMet meta.minLevel(), meta.maxLevel(), meta.minRowId(), - meta.maxRowId()); + meta.maxRowId(), + toStringArrayData(meta.extraFiles())); } @Override @@ -90,6 +94,7 @@ public class ManifestFileMetaSerializer extends ObjectSerializer<ManifestFileMet row.isNullAt(8) ? null : row.getInt(8), row.isNullAt(9) ? null : row.getInt(9), row.isNullAt(10) ? null : row.getLong(10), - row.isNullAt(11) ? null : row.getLong(11)); + row.isNullAt(11) ? null : row.getLong(11), + row.isNullAt(12) ? null : fromStringArrayData(row.getArray(12))); } } diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java index a0545c87e4..6494411a8c 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileDeletionBase.java @@ -326,6 +326,9 @@ public abstract class FileDeletionBase<T extends Snapshot> { String fileName = manifest.fileName(); if (skippingSet.add(fileName)) { manifests.add(fileName); + if (manifest.extraFiles() != null) { + manifests.addAll(manifest.extraFiles()); + } } } if (skippingSet.add(manifestName)) { diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/OrphanFilesClean.java b/paimon-core/src/main/java/org/apache/paimon/operation/OrphanFilesClean.java index 4245460225..04b4ae63ff 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/OrphanFilesClean.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/OrphanFilesClean.java @@ -309,6 +309,11 @@ public abstract class OrphanFilesClean implements Serializable { // collect manifests for (ManifestFileMeta manifest : manifestFileMetas) { usedFileWithFlagConsumer.accept(Pair.of(manifest.fileName(), true)); + if (manifest.extraFiles() != null) { + for (String extraFile : manifest.extraFiles()) { + usedFileWithFlagConsumer.accept(Pair.of(extraFile, false)); + } + } } // index files diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaSerializerTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaSerializerTest.java index 57b5a08ed0..2f4e32cd27 100644 --- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaSerializerTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaSerializerTest.java @@ -23,7 +23,10 @@ import org.apache.paimon.utils.ObjectSerializerTestBase; import org.junit.jupiter.api.Test; +import java.io.IOException; import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; import java.util.List; import static org.assertj.core.api.Assertions.assertThat; @@ -40,6 +43,47 @@ public class ManifestFileMetaSerializerTest extends ObjectSerializerTestBase<Man assertThat(new ManifestFileMetaSerializer().toRow(object()).getInt(0)).isEqualTo(2); } + @Test + void testExtraFiles() throws IOException { + ManifestFileMeta original = object(); + assertThat(original.extraFiles()).isNull(); + + ManifestFileMetaSerializer serializer = new ManifestFileMetaSerializer(); + for (List<String> extraFiles : + Arrays.asList( + null, + Collections.<String>emptyList(), + Arrays.asList("extra-1", "extra-2"))) { + ManifestFileMeta meta = + new ManifestFileMeta( + original.fileName(), + original.fileSize(), + original.numAddedFiles(), + original.numDeletedFiles(), + original.partitionStats(), + original.schemaId(), + original.minBucket(), + original.maxBucket(), + original.minLevel(), + original.maxLevel(), + original.minRowId(), + original.maxRowId(), + extraFiles); + + ManifestFileMeta fromRow = serializer.fromRow(serializer.toRow(meta)); + ManifestFileMeta fromBytes = serializer.deserializeFromBytes(meta.toBytes()); + assertThat(fromRow).isEqualTo(meta); + assertThat(fromBytes).isEqualTo(meta).hasSameHashCodeAs(meta); + assertThat(fromRow.extraFiles()).isEqualTo(extraFiles); + assertThat(fromBytes.extraFiles()).isEqualTo(extraFiles); + if (extraFiles == null) { + assertThat(meta).isEqualTo(original).hasSameHashCodeAs(original); + } else { + assertThat(meta).isNotEqualTo(original); + } + } + } + @Override protected ObjectSerializer<ManifestFileMeta> serializer() { return new ManifestFileMetaSerializer(); diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestListTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestListTest.java index 8442be28c6..4891a760b7 100644 --- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestListTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestListTest.java @@ -35,6 +35,8 @@ import org.junit.jupiter.api.io.TempDir; import java.io.IOException; import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; import java.util.List; import java.util.Random; import java.util.UUID; @@ -57,6 +59,9 @@ public class ManifestListTest { String manifestListName = manifestList.write(metas).getKey(); List<ManifestFileMeta> actualMetas = manifestList.read(manifestListName); assertThat(actualMetas).isEqualTo(metas); + for (int i = 0; i < metas.size(); i++) { + assertThat(actualMetas.get(i).extraFiles()).isEqualTo(metas.get(i).extraFiles()); + } } @RepeatedTest(10) @@ -96,6 +101,7 @@ public class ManifestListTest { ManifestList manifestList = createManifestList(tempDir.toString()); List<ManifestFileMeta> actualMetas = manifestList.read(manifestListName); assertThat(actualMetas).isEqualTo(getLegacyMetaPaimon10(metas)); + assertThat(actualMetas).allSatisfy(meta -> assertThat(meta.extraFiles()).isNull()); } @Test @@ -107,6 +113,7 @@ public class ManifestListTest { ManifestList legacyManifestList = createLegacyManifestListPaimon10(); List<ManifestFileMeta> actualMetas = legacyManifestList.read(manifestListName); assertThat(actualMetas).isEqualTo(getLegacyMetaPaimon10(metas)); + assertThat(actualMetas).allSatisfy(meta -> assertThat(meta.extraFiles()).isNull()); } private ManifestList createLegacyManifestListPaimon10() { @@ -156,7 +163,28 @@ public class ManifestListTest { for (int j = random.nextInt(10) + 1; j > 0; j--) { entries.add(gen.next()); } - metas.add(gen.createManifestFileMeta(entries)); + ManifestFileMeta meta = gen.createManifestFileMeta(entries); + List<String> extraFiles = + i % 3 == 0 + ? null + : i % 3 == 1 + ? Collections.emptyList() + : Arrays.asList("extra-" + i + "-1", "extra-" + i + "-2"); + metas.add( + new ManifestFileMeta( + meta.fileName(), + meta.fileSize(), + meta.numAddedFiles(), + meta.numDeletedFiles(), + meta.partitionStats(), + meta.schemaId(), + meta.minBucket(), + meta.maxBucket(), + meta.minLevel(), + meta.maxLevel(), + meta.minRowId(), + meta.maxRowId(), + extraFiles)); } return metas; } diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/ExpireSnapshotsTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/ExpireSnapshotsTest.java index c5a21e57c9..8dfcb1c21a 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/ExpireSnapshotsTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/operation/ExpireSnapshotsTest.java @@ -60,6 +60,8 @@ import org.junit.jupiter.api.BeforeEach; 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.ValueSource; import java.io.IOException; import java.nio.file.Files; @@ -404,10 +406,15 @@ public class ExpireSnapshotsTest { private Snapshot snapshotWithManifestLists( String deltaManifestList, String changelogManifestList) { + return snapshotWithManifestLists(null, deltaManifestList, changelogManifestList); + } + + private Snapshot snapshotWithManifestLists( + String baseManifestList, String deltaManifestList, String changelogManifestList) { return new Snapshot( 0, 0L, - null, + baseManifestList, null, deltaManifestList, null, @@ -429,6 +436,111 @@ public class ExpireSnapshotsTest { null); } + @Test + public void testCleanUnusedManifestExtraFiles() throws Exception { + ManifestFileMeta base = manifestWithExtraFiles("base", null); + ManifestFileMeta empty = manifestWithExtraFiles("empty", Collections.emptyList()); + ManifestFileMeta delta = + manifestWithExtraFiles("delta", Arrays.asList("delta-extra-1", "delta-extra-2")); + ManifestFileMeta changelog = + manifestWithExtraFiles("changelog", Collections.singletonList("changelog-extra")); + Snapshot snapshot = + snapshotWithManifestLists( + writeManifestList(base, empty), + writeManifestList(delta), + writeManifestList(changelog)); + + store.newSnapshotDeletion().cleanUnusedManifests(snapshot, new HashSet<>()); + + for (String name : + Arrays.asList( + "base", + "empty", + "delta", + "changelog", + "delta-extra-1", + "delta-extra-2", + "changelog-extra", + snapshot.baseManifestList(), + snapshot.deltaManifestList(), + snapshot.changelogManifestList())) { + assertThat(fileIO.exists(store.pathFactory().toManifestFilePath(name))) + .as(name) + .isFalse(); + } + } + + @ParameterizedTest + @ValueSource(strings = {"snapshot", "tag", "changelog"}) + public void testCleanManifestExtraFilesFollowsManifestRetention(String cleaner) + throws Exception { + ManifestFileMeta expired = + manifestWithExtraFiles("expired", Collections.singletonList("expired-extra")); + ManifestFileMeta retained = + manifestWithExtraFiles("retained", Collections.singletonList("retained-extra")); + Snapshot expiredSnapshot = + snapshotWithManifestLists( + writeManifestList(), writeManifestList(expired, retained), null); + Snapshot retainedSnapshot = + snapshotWithManifestLists(writeManifestList(), writeManifestList(retained), null); + + SnapshotDeletion snapshotDeletion = store.newSnapshotDeletion(); + List<Snapshot> skippingSnapshots = Collections.singletonList(retainedSnapshot); + if ("changelog".equals(cleaner)) { + ChangelogDeletion deletion = store.newChangelogDeletion(); + deletion.cleanUnusedManifestList( + expiredSnapshot.deltaManifestList(), + deletion.manifestSkippingSet(skippingSnapshots)); + } else { + FileDeletionBase<Snapshot> deletion = + "tag".equals(cleaner) ? store.newTagDeletion() : snapshotDeletion; + deletion.cleanUnusedManifests( + expiredSnapshot, deletion.manifestSkippingSet(skippingSnapshots)); + } + + assertThat(fileIO.exists(store.pathFactory().toManifestFilePath("expired"))).isFalse(); + assertThat(fileIO.exists(store.pathFactory().toManifestFilePath("expired-extra"))) + .isFalse(); + for (String name : Arrays.asList("retained", "retained-extra")) { + assertThat(fileIO.exists(store.pathFactory().toManifestFilePath(name))) + .as(name) + .isTrue(); + } + + snapshotDeletion.cleanUnusedManifests(retainedSnapshot, new HashSet<>()); + assertThat(fileIO.exists(store.pathFactory().toManifestFilePath("retained"))).isFalse(); + assertThat(fileIO.exists(store.pathFactory().toManifestFilePath("retained-extra"))) + .isFalse(); + } + + private ManifestFileMeta manifestWithExtraFiles(String fileName, List<String> extraFiles) + throws IOException { + fileIO.writeFile(store.pathFactory().toManifestFilePath(fileName), "manifest", true); + if (extraFiles != null) { + for (String extraFile : extraFiles) { + fileIO.writeFile(store.pathFactory().toManifestFilePath(extraFile), "extra", true); + } + } + return new ManifestFileMeta( + fileName, + 0L, + 0L, + 0L, + SimpleStats.EMPTY_STATS, + 0L, + null, + null, + null, + null, + null, + null, + extraFiles); + } + + private String writeManifestList(ManifestFileMeta... manifests) { + return store.manifestListFactory().create().write(Arrays.asList(manifests)).getKey(); + } + @Test public void testNoSnapshot() throws IOException { ExpireSnapshots expire = store.newExpire(1, 3, Long.MAX_VALUE); diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/LocalOrphanFilesCleanTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/LocalOrphanFilesCleanTest.java index bfe3f6d76c..36499bb4a6 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/LocalOrphanFilesCleanTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/operation/LocalOrphanFilesCleanTest.java @@ -31,6 +31,7 @@ import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.FileStatus; import org.apache.paimon.fs.Path; import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.manifest.ManifestFileMeta; import org.apache.paimon.manifest.ManifestList; import org.apache.paimon.mergetree.compact.ConcatRecordReader; import org.apache.paimon.options.Options; @@ -52,10 +53,14 @@ import org.apache.paimon.types.DataTypes; import org.apache.paimon.types.RowKind; import org.apache.paimon.types.RowType; import org.apache.paimon.utils.FileStorePathFactory; +import org.apache.paimon.utils.JsonSerdeUtil; +import org.apache.paimon.utils.Pair; import org.apache.paimon.utils.Preconditions; import org.apache.paimon.utils.SnapshotManager; import org.apache.paimon.utils.StringUtils; +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.node.ObjectNode; + import org.assertj.core.api.Assertions; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; @@ -541,6 +546,56 @@ public class LocalOrphanFilesCleanTest { validate(deleted, snapshotData, changelogData); } + @Test + public void testPreservesManifestExtraFiles() throws Exception { + commit(generateData()); + SnapshotManager snapshotManager = table.snapshotManager(); + Snapshot snapshot = snapshotManager.latestSnapshot(); + ManifestList manifestList = table.store().manifestListFactory().create(); + List<ManifestFileMeta> manifests = manifestList.read(snapshot.deltaManifestList()); + ManifestFileMeta meta = manifests.get(0); + String extraFile = "manifest-extra"; + manifests.set( + 0, + new ManifestFileMeta( + meta.fileName(), + meta.fileSize(), + meta.numAddedFiles(), + meta.numDeletedFiles(), + meta.partitionStats(), + meta.schemaId(), + meta.minBucket(), + meta.maxBucket(), + meta.minLevel(), + meta.maxLevel(), + meta.minRowId(), + meta.maxRowId(), + Collections.singletonList(extraFile))); + Pair<String, Long> newManifestList = manifestList.write(manifests); + ObjectNode node = + (ObjectNode) JsonSerdeUtil.OBJECT_MAPPER_INSTANCE.readTree(snapshot.toJson()); + node.put("deltaManifestList", newManifestList.getKey()); + node.put("deltaManifestListSize", newManifestList.getValue()); + fileIO.overwriteFileUtf8(snapshotManager.snapshotPath(snapshot.id()), node.toString()); + snapshotManager.invalidateCache(); + + Path extraPath = new Path(manifestDir, extraFile); + Path orphanPath = new Path(manifestDir, "orphan-extra"); + fileIO.writeFile(extraPath, "extra file, not an Avro manifest", true); + fileIO.writeFile(orphanPath, "orphan", true); + + LocalOrphanFilesClean cleaner = + new LocalOrphanFilesClean( + table, System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(2)); + List<Path> deleted = cleaner.clean().getDeletedFilesPath(); + assertThat(deleted) + .extracting(Path::getName) + .contains(orphanPath.getName()) + .doesNotContain(extraFile); + assertThat(fileIO.exists(extraPath)).isTrue(); + assertThat(fileIO.exists(orphanPath)).isFalse(); + } + /** Manually make a FileNotFoundException to simulate snapshot expire while clean. */ @Test public void testAbnormallyRemoving() throws Exception { diff --git a/paimon-python/pypaimon/manifest/manifest_list_manager.py b/paimon-python/pypaimon/manifest/manifest_list_manager.py index 3a0e606ef5..5b867e1ae2 100644 --- a/paimon-python/pypaimon/manifest/manifest_list_manager.py +++ b/paimon-python/pypaimon/manifest/manifest_list_manager.py @@ -98,6 +98,7 @@ class ManifestListManager: schema_id=record['_SCHEMA_ID'], min_row_id=record.get('_MIN_ROW_ID'), max_row_id=record.get('_MAX_ROW_ID'), + extra_files=record.get('_EXTRA_FILES'), ) manifest_files.append(manifest_file_meta) @@ -120,6 +121,7 @@ class ManifestListManager: "_SCHEMA_ID": meta.schema_id, "_MIN_ROW_ID": meta.min_row_id, "_MAX_ROW_ID": meta.max_row_id, + "_EXTRA_FILES": meta.extra_files, } avro_records.append(avro_record) diff --git a/paimon-python/pypaimon/manifest/schema/manifest_file_meta.py b/paimon-python/pypaimon/manifest/schema/manifest_file_meta.py index 3c45b71695..2681adacdd 100644 --- a/paimon-python/pypaimon/manifest/schema/manifest_file_meta.py +++ b/paimon-python/pypaimon/manifest/schema/manifest_file_meta.py @@ -17,7 +17,7 @@ from dataclasses import dataclass -from typing import Optional +from typing import List, Optional from pypaimon.manifest.schema.simple_stats import (PARTITION_STATS_SCHEMA, SimpleStats) @@ -33,6 +33,7 @@ class ManifestFileMeta: min_row_id: Optional[int] = None max_row_id: Optional[int] = None + extra_files: Optional[List[str]] = None MANIFEST_FILE_META_SCHEMA = { "type": "record", @@ -47,5 +48,6 @@ MANIFEST_FILE_META_SCHEMA = { {"name": "_SCHEMA_ID", "type": "long"}, {"name": "_MIN_ROW_ID", "type": ["null", "long"], "default": None}, {"name": "_MAX_ROW_ID", "type": ["null", "long"], "default": None}, + {"name": "_EXTRA_FILES", "type": ["null", {"type": "array", "items": "string"}], "default": None}, ] } diff --git a/paimon-python/pypaimon/tests/manifest/manifest_manager_test.py b/paimon-python/pypaimon/tests/manifest/manifest_manager_test.py index ad4754e057..e556c55cf8 100644 --- a/paimon-python/pypaimon/tests/manifest/manifest_manager_test.py +++ b/paimon-python/pypaimon/tests/manifest/manifest_manager_test.py @@ -22,7 +22,9 @@ import sys import tempfile import threading import unittest +from io import BytesIO +import fastavro import pyarrow as pa from pypaimon.catalog.filesystem_catalog import FileSystemCatalog @@ -465,6 +467,44 @@ class ManifestListManagerTest(_ManifestManagerSetup): ) manager.write(name, [meta]) + def test_extra_files_round_trip(self): + manager = self._make_manager() + expected_extra_files = [None, [], ["extra-1", "extra-2"]] + metas = [] + for i, extra_files in enumerate(expected_extra_files): + meta = ManifestFileMeta( + file_name=f"manifest-{i}.avro", file_size=1024, + num_added_files=1, num_deleted_files=0, + partition_stats=SimpleStats.empty_stats(), schema_id=0, + ) + self.assertIsNone(meta.extra_files) + meta.extra_files = extra_files + metas.append(meta) + + name = "manifest-list-extra-files" + manager.write(name, metas) + actual = manager.read(name) + self.assertEqual([meta.file_name for meta in actual], + [meta.file_name for meta in metas]) + self.assertEqual([meta.extra_files for meta in actual], expected_extra_files) + + with manager.file_io.new_input_stream(f"{manager.manifest_path}/{name}") as stream: + avro_bytes = stream.read() + reader = fastavro.reader(BytesIO(avro_bytes)) + records = list(reader) + self.assertEqual([record["_VERSION"] for record in records], [2, 2, 2]) + self.assertEqual([record["_EXTRA_FILES"] for record in records], expected_extra_files) + + legacy_schema = reader.writer_schema + legacy_schema["fields"] = [ + field for field in legacy_schema["fields"] if field["name"] != "_EXTRA_FILES" + ] + legacy_records = list(fastavro.reader(BytesIO(avro_bytes), reader_schema=legacy_schema)) + self.assertEqual([record["_FILE_NAME"] for record in legacy_records], + [meta.file_name for meta in metas]) + for record in legacy_records: + self.assertNotIn("_EXTRA_FILES", record) + def _make_snapshot(self, base_manifest_list, delta_manifest_list="delta-manifest-list"): from pypaimon.snapshot.snapshot import Snapshot return Snapshot( diff --git a/paimon-python/pypaimon/tests/manifest/manifest_schema_test.py b/paimon-python/pypaimon/tests/manifest/manifest_schema_test.py index f483d84ccc..1e7bfa3670 100644 --- a/paimon-python/pypaimon/tests/manifest/manifest_schema_test.py +++ b/paimon-python/pypaimon/tests/manifest/manifest_schema_test.py @@ -130,7 +130,7 @@ class ManifestSchemaTest(unittest.TestCase): expected_fields = [ "_VERSION", "_FILE_NAME", "_FILE_SIZE", "_NUM_ADDED_FILES", "_NUM_DELETED_FILES", "_PARTITION_STATS", "_SCHEMA_ID", - "_MIN_ROW_ID", "_MAX_ROW_ID", + "_MIN_ROW_ID", "_MAX_ROW_ID", "_EXTRA_FILES", ] for field_name in expected_fields: @@ -146,6 +146,9 @@ class ManifestSchemaTest(unittest.TestCase): self.assertEqual(field_map["_SCHEMA_ID"]["type"], "long") self.assertEqual(field_map["_MIN_ROW_ID"]["type"], ["null", "long"]) self.assertEqual(field_map["_MAX_ROW_ID"]["type"], ["null", "long"]) + self.assertEqual(field_map["_EXTRA_FILES"]["type"], + ["null", {"type": "array", "items": "string"}]) + self.assertIsNone(field_map["_EXTRA_FILES"]["default"]) self.assertIsNone( field_map["_MIN_ROW_ID"].get("default"), "_MIN_ROW_ID should have default None for backward compatibility", @@ -232,3 +235,8 @@ class ManifestSchemaTest(unittest.TestCase): self.assertEqual(meta.schema_id, 0) self.assertIsNone(meta.min_row_id) self.assertIsNone(meta.max_row_id) + self.assertIsNone(meta.extra_files) + + buffer.seek(0) + resolved_record = next(fastavro.reader(buffer, reader_schema=MANIFEST_FILE_META_SCHEMA)) + self.assertIsNone(resolved_record["_EXTRA_FILES"])
