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 336d730943 [core][python] Add extra files to manifest metadata (#9746)
336d730943 is described below
commit 336d73094360eb34010adcc0544f12881b3e2231
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"])
