You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/03/18 11:25:07 UTC
(paimon) branch master updated: [core] Do not clean base and delta manifests in ChangelogDeletion (#7429)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 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 2201ac6bd7 [core] Do not clean base and delta manifests in ChangelogDeletion (#7429)
2201ac6bd7 is described below
commit 2201ac6bd7a0612fdb111ae86734d188ff42a40b
Author: tsreaper <ts...@gmail.com>
AuthorDate: Wed Mar 18 19:25:01 2026 +0800
[core] Do not clean base and delta manifests in ChangelogDeletion (#7429)
Currently, `ChangelogDeletion` will clean base and delta manifests.
However these manifests are also cleaned in `SnapshotDeletion`.
Consider the following scenario:
* `snapshot-1` is changed to `changelog-1`. However due to some issue,
its base and delta manifests are not cleaned.
* `snapshot-1` is changed to `changelog-2`. Its base and delta manifests
are cleaned.
* When expiring `changelog-1`, as `changelog-2` has no manifests, all
manifests in `changelog-1` will be cleaned, even if they're still in use
by later snapshots.
This PR fixes the bug by not cleaning base and delta manifests in
`ChangelogDeletion`.
---
.../apache/paimon/operation/ChangelogDeletion.java | 25 +++++----
.../apache/paimon/operation/FileDeletionBase.java | 7 +++
.../apache/paimon/table/ExpireChangelogImpl.java | 1 +
.../test/java/org/apache/paimon/TestFileStore.java | 3 +-
.../apache/paimon/table/ChangelogExpireTest.java | 65 ++++++++++++++++++++++
.../paimon/table/IndexFileExpireTableTest.java | 4 +-
6 files changed, 90 insertions(+), 15 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/ChangelogDeletion.java b/paimon-core/src/main/java/org/apache/paimon/operation/ChangelogDeletion.java
index 348d2da6bd..450f7fcb69 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/ChangelogDeletion.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/ChangelogDeletion.java
@@ -63,10 +63,10 @@ public class ChangelogDeletion extends FileDeletionBase<Changelog> {
public void cleanUnusedDataFiles(Changelog changelog, Predicate<ExpireFileEntry> skipper) {
if (changelog.changelogManifestList() != null) {
deleteAddedDataFiles(changelog.changelogManifestList());
- }
-
- if (manifestList.exists(changelog.deltaManifestList())) {
- cleanUnusedDataFiles(changelog.deltaManifestList(), skipper);
+ } else {
+ if (manifestList.exists(changelog.deltaManifestList())) {
+ cleanUnusedDataFiles(changelog.deltaManifestList(), skipper);
+ }
}
}
@@ -74,14 +74,15 @@ public class ChangelogDeletion extends FileDeletionBase<Changelog> {
public void cleanUnusedManifests(Changelog changelog, Set<String> skippingSet) {
if (changelog.changelogManifestList() != null) {
cleanUnusedManifestList(changelog.changelogManifestList(), skippingSet);
- }
-
- if (manifestList.exists(changelog.deltaManifestList())) {
- cleanUnusedManifestList(changelog.deltaManifestList(), skippingSet);
- }
-
- if (manifestList.exists(changelog.baseManifestList())) {
- cleanUnusedManifestList(changelog.baseManifestList(), skippingSet);
+ } else {
+ if (manifestList.exists(changelog.deltaManifestList())) {
+ cleanUnusedManifestList(changelog.deltaManifestList(), skippingSet);
+ }
+ // See FileDeletionBase#cleanUnusedManifests
+ // about why we need to clean base manifest
+ if (manifestList.exists(changelog.baseManifestList())) {
+ cleanUnusedManifestList(changelog.baseManifestList(), skippingSet);
+ }
}
// the index and statics manifest list should handle by snapshot deletion.
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 231636355a..4256a503c1 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
@@ -334,6 +334,13 @@ public abstract class FileDeletionBase<T extends Snapshot> {
boolean deleteDataManifestLists,
boolean deleteChangelog) {
if (deleteDataManifestLists) {
+ // deleteDataManifestLists will be false
+ // with changelog decouple + none changelog producer.
+ // Why don't we clean base manifest in this scenario?
+ // Because cleanUnusedManifestList is compared with the earliest snapshot.
+ // For none changelog producer, changelog files are the level 0 files.
+ // Even if these files are not used by the earliest snapshot,
+ // we have to keep them as changelog, and clean then in ChangelogDeletion.
cleanUnusedManifestList(snapshot.baseManifestList(), skippingSet);
cleanUnusedManifestList(snapshot.deltaManifestList(), skippingSet);
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/ExpireChangelogImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/ExpireChangelogImpl.java
index 1b42f2339b..bed5749895 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/ExpireChangelogImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/ExpireChangelogImpl.java
@@ -149,6 +149,7 @@ public class ExpireChangelogImpl implements ExpireSnapshots {
List<Snapshot> skippingSnapshots =
findSkippingTags(taggedSnapshots, earliestId, endExclusiveId);
skippingSnapshots.add(changelogManager.changelog(endExclusiveId));
+ skippingSnapshots.add(snapshotManager.earliestSnapshot());
Set<String> manifestSkippSet = changelogDeletion.manifestSkippingSet(skippingSnapshots);
for (long id = earliestId; id < endExclusiveId; id++) {
if (LOG.isDebugEnabled()) {
diff --git a/paimon-core/src/test/java/org/apache/paimon/TestFileStore.java b/paimon-core/src/test/java/org/apache/paimon/TestFileStore.java
index f904cde081..8142d44acf 100644
--- a/paimon-core/src/test/java/org/apache/paimon/TestFileStore.java
+++ b/paimon-core/src/test/java/org/apache/paimon/TestFileStore.java
@@ -717,7 +717,8 @@ public class TestFileStore extends KeyValueFileStore {
// delta file
if (options.changelogProducer() == CoreOptions.ChangelogProducer.NONE) {
- // TODO why we need to keep base manifests?
+ // See FileDeletionBase#cleanUnusedManifests
+ // about why we need to keep base manifest
result.add(pathFactory.toManifestListPath(changelog.baseManifestList()));
manifestList
.readDataManifests(changelog)
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/ChangelogExpireTest.java b/paimon-core/src/test/java/org/apache/paimon/table/ChangelogExpireTest.java
index c6eab6dd19..1821d927c4 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/ChangelogExpireTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/ChangelogExpireTest.java
@@ -18,22 +18,35 @@
package org.apache.paimon.table;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.catalog.CatalogFactory;
import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.manifest.ManifestList;
import org.apache.paimon.options.ExpireConfig;
import org.apache.paimon.schema.Schema;
+import org.apache.paimon.table.sink.StreamTableCommit;
+import org.apache.paimon.table.sink.StreamTableWrite;
+import org.apache.paimon.table.sink.StreamWriteBuilder;
import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.SnapshotManager;
import org.apache.paimon.utils.TraceableFileIO;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import java.util.HashMap;
+import java.util.Map;
import java.util.UUID;
import static org.apache.paimon.options.CatalogOptions.CACHE_ENABLED;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
/** Test for changelog expire. */
@@ -78,4 +91,56 @@ public class ChangelogExpireTest extends IndexFileExpireTableTest {
expireSnapshots.expireUntil(1, 7);
assertThatCode(() -> expire.expireUntil(1, 6)).doesNotThrowAnyException();
}
+
+ @Test
+ public void testChangelogExpireWithUncleanedManifestLists() throws Exception {
+ Map<String, String> dynamicOptions = new HashMap<>();
+ dynamicOptions.put(CoreOptions.SNAPSHOT_NUM_RETAINED_MIN.key(), "1");
+ dynamicOptions.put(CoreOptions.SNAPSHOT_NUM_RETAINED_MAX.key(), "1");
+ dynamicOptions.put(CoreOptions.CHANGELOG_NUM_RETAINED_MAX.key(), "3");
+ table = table.copy(dynamicOptions);
+
+ StreamWriteBuilder writeBuilder = table.newStreamWriteBuilder();
+ StreamTableWrite write = writeBuilder.newWrite();
+ StreamTableCommit commit = writeBuilder.newCommit();
+
+ write(write, createRow(1, 0, 1, 10));
+ commit.commit(1, write.prepareCommit(true, 1));
+
+ // Save snapshot-1's manifest list files before they get cleaned
+ SnapshotManager snapshotManager = table.snapshotManager();
+ FileStorePathFactory pathFactory = table.store().pathFactory();
+ FileIO fileIO = table.fileIO();
+ Snapshot snapshot1 = snapshotManager.snapshot(1);
+ Path manifestDir = pathFactory.manifestPath();
+ Path baseSaved = new Path(manifestDir, snapshot1.baseManifestList() + ".saved");
+ Path deltaSaved = new Path(manifestDir, snapshot1.deltaManifestList() + ".saved");
+ fileIO.copyFile(
+ pathFactory.toManifestListPath(snapshot1.baseManifestList()), baseSaved, false);
+ fileIO.copyFile(
+ pathFactory.toManifestListPath(snapshot1.deltaManifestList()), deltaSaved, false);
+
+ write(write, createRow(1, 0, 2, 20));
+ commit.commit(2, write.prepareCommit(true, 2));
+ write(write, createRow(1, 0, 3, 30));
+ commit.commit(3, write.prepareCommit(true, 3));
+
+ // Copy back changelog-1's manifest list files, simulating SnapshotDeletion failure
+ fileIO.copyFile(
+ baseSaved, pathFactory.toManifestListPath(snapshot1.baseManifestList()), true);
+ fileIO.copyFile(
+ deltaSaved, pathFactory.toManifestListPath(snapshot1.deltaManifestList()), true);
+
+ // expire changelog-1
+ write(write, createRow(1, 0, 4, 40));
+ commit.commit(4, write.prepareCommit(true, 4));
+
+ write.close();
+ commit.close();
+ Snapshot snapshot4 = snapshotManager.snapshot(4);
+ ManifestList manifestList = table.store().manifestListFactory().create();
+ for (ManifestFileMeta manifest : manifestList.readDataManifests(snapshot4)) {
+ assertThat(fileIO.exists(pathFactory.toManifestFilePath(manifest.fileName()))).isTrue();
+ }
+ }
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/IndexFileExpireTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/IndexFileExpireTableTest.java
index d9e58e245c..7293e8f8be 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/IndexFileExpireTableTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/IndexFileExpireTableTest.java
@@ -52,7 +52,7 @@ public class IndexFileExpireTableTest extends PrimaryKeyTableTestBase {
return options;
}
- private Pair<GenericRow, Integer> createRow(int partition, int bucket, int key, int value) {
+ protected Pair<GenericRow, Integer> createRow(int partition, int bucket, int key, int value) {
return Pair.of(GenericRow.of(partition, key, value), bucket);
}
@@ -239,7 +239,7 @@ public class IndexFileExpireTableTest extends PrimaryKeyTableTestBase {
.count();
}
- private void write(StreamTableWrite write, Pair<GenericRow, Integer> rowWithBucket)
+ protected void write(StreamTableWrite write, Pair<GenericRow, Integer> rowWithBucket)
throws Exception {
write.write(rowWithBucket.getKey(), rowWithBucket.getValue());
}