You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by yu...@apache.org on 2026/05/22 12:10:55 UTC
(paimon) branch master updated: [core] Changelog expiration skip expired changes to avoid FileNotFoundException (#7923)
This is an automated email from the ASF dual-hosted git repository.
yuzelin 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 df830aac9d [core] Changelog expiration skip expired changes to avoid FileNotFoundException (#7923)
df830aac9d is described below
commit df830aac9dda6dfa82a43e110afcc70a56250bd9
Author: yuzelin <33...@users.noreply.github.com>
AuthorDate: Fri May 22 20:10:49 2026 +0800
[core] Changelog expiration skip expired changes to avoid FileNotFoundException (#7923)
---
.../apache/paimon/table/ExpireChangelogImpl.java | 52 +++++++++++++++-------
.../apache/paimon/table/ChangelogExpireTest.java | 50 +++++++++++++++++++++
2 files changed, 87 insertions(+), 15 deletions(-)
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 bed5749895..73f9c88b5f 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
@@ -130,10 +130,20 @@ public class ExpireChangelogImpl implements ExpireSnapshots {
// Only clean the snapshot in changelog dir
maxExclusive = Math.min(maxExclusive, latestChangelogId);
+ if (maxExclusive <= earliestChangelogId) {
+ // This happens when retainMin >= total changelog count
+ // (e.g. latestSnapshotId - retainMin + 1 <= earliestChangelogId)
+ return 0;
+ }
+
for (long id = min; id <= maxExclusive; id++) {
- if (changelogManager.longLivedChangelogExists(id)
- && olderThanMills <= changelogManager.longLivedChangelog(id).timeMillis()) {
- return expireUntil(earliestChangelogId, id);
+ try {
+ if (olderThanMills <= changelogManager.tryGetChangelog(id).timeMillis()) {
+ return expireUntil(earliestChangelogId, id);
+ }
+ } catch (FileNotFoundException e) {
+ // ignore
+ // snapshot may have been deleted by another process
}
}
return expireUntil(earliestChangelogId, maxExclusive);
@@ -141,29 +151,42 @@ public class ExpireChangelogImpl implements ExpireSnapshots {
public int expireUntil(long earliestId, long endExclusiveId) {
if (LOG.isDebugEnabled()) {
- LOG.debug("Changelog expire range is [" + earliestId + ", " + endExclusiveId + ")");
+ LOG.debug("Changelog expire range is [{}, {})", earliestId, endExclusiveId);
}
List<Snapshot> taggedSnapshots = tagManager.taggedSnapshots();
List<Snapshot> skippingSnapshots =
findSkippingTags(taggedSnapshots, earliestId, endExclusiveId);
- skippingSnapshots.add(changelogManager.changelog(endExclusiveId));
+ try {
+ skippingSnapshots.add(changelogManager.tryGetChangelog(endExclusiveId));
+ } catch (FileNotFoundException e) {
+ LOG.error(
+ "The endExclusive changelog #{} not found, skip expiration. Maybe you should use expire_changelogs to delete the separated changelogs.",
+ endExclusiveId,
+ e);
+ return 0;
+ }
skippingSnapshots.add(snapshotManager.earliestSnapshot());
Set<String> manifestSkippSet = changelogDeletion.manifestSkippingSet(skippingSnapshots);
for (long id = earliestId; id < endExclusiveId; id++) {
if (LOG.isDebugEnabled()) {
- LOG.debug("Ready to delete changelog files from changelog #" + id);
+ LOG.debug("Ready to delete changelog files from changelog #{}", id);
+ }
+ Changelog changelog;
+ try {
+ changelog = changelogManager.tryGetChangelog(id);
+ } catch (FileNotFoundException e) {
+ LOG.info("Changelog #{} not found, skip it.", id, e);
+ continue;
}
- Changelog changelog = changelogManager.longLivedChangelog(id);
Predicate<ExpireFileEntry> skipper;
try {
skipper = changelogDeletion.createDataFileSkipperForTags(taggedSnapshots, id);
} catch (Exception e) {
LOG.info(
- String.format(
- "Skip cleaning data files of changelog '%s' due to failed to build skipping set.",
- id),
+ "Skip cleaning data files of changelog #{} due to failed to build skipping set.",
+ id,
e);
continue;
}
@@ -219,13 +242,13 @@ public class ExpireChangelogImpl implements ExpireSnapshots {
Set<String> manifestSkippSet = changelogDeletion.manifestSkippingSet(skippingSnapshots);
for (long id = earliestChangelogId; id <= latestChangelogId; id++) {
- LOG.info("Ready to delete changelog files from changelog #" + id);
+ LOG.info("Ready to delete changelog files from changelog #{}", id);
Changelog changelog;
try {
changelog = changelogManager.tryGetChangelog(id);
} catch (FileNotFoundException e) {
- LOG.info("fail to get changelog #" + id);
+ LOG.info("fail to get changelog #{}", id);
continue;
}
Predicate<ExpireFileEntry> skipper;
@@ -233,9 +256,8 @@ public class ExpireChangelogImpl implements ExpireSnapshots {
skipper = changelogDeletion.createDataFileSkipperForTags(taggedSnapshots, id);
} catch (Exception e) {
LOG.info(
- String.format(
- "Skip cleaning data files of changelog '%s' due to failed to build skipping set.",
- id),
+ "Skip cleaning data files of changelog '{}' due to failed to build skipping set.",
+ id,
e);
continue;
}
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 1821d927c4..281dd0f1a6 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
@@ -34,6 +34,7 @@ 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.ChangelogManager;
import org.apache.paimon.utils.FileStorePathFactory;
import org.apache.paimon.utils.SnapshotManager;
import org.apache.paimon.utils.TraceableFileIO;
@@ -41,6 +42,7 @@ import org.apache.paimon.utils.TraceableFileIO;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
@@ -143,4 +145,52 @@ public class ChangelogExpireTest extends IndexFileExpireTableTest {
assertThat(fileIO.exists(pathFactory.toManifestFilePath(manifest.fileName()))).isTrue();
}
}
+
+ @Test
+ public void testExpireWithMiddleChangelogNotFound() throws Exception {
+ StreamWriteBuilder writeBuilder = table.newStreamWriteBuilder();
+ StreamTableWrite write = writeBuilder.newWrite();
+ StreamTableCommit commit = writeBuilder.newCommit();
+ for (int i = 1; i <= 10; i++) {
+ write(write, createRow(1, 0, i, i * 10));
+ commit.commit(i, write.prepareCommit(true, i));
+ }
+ write.close();
+ commit.close();
+
+ SnapshotManager snapshotManager = table.snapshotManager();
+ long latestSnapshotId = snapshotManager.latestSnapshotId();
+
+ // changelogRetainMax > snapshotRetainMax to ensure changelogDecoupled=true
+ ExpireConfig expireConfig =
+ ExpireConfig.builder()
+ .changelogRetainMax((int) latestSnapshotId)
+ .changelogRetainMin(1)
+ .changelogTimeRetain(Duration.ofMillis(0))
+ .snapshotRetainMax(1)
+ .snapshotRetainMin(1)
+ .build();
+ ExpireSnapshotsImpl expireSnapshots =
+ (ExpireSnapshotsImpl) table.newExpireSnapshots().config(expireConfig);
+ expireSnapshots.expire();
+
+ ChangelogManager changelogManager = table.changelogManager();
+ FileIO fileIO = table.fileIO();
+ long latestChangelogId = changelogManager.latestLongLivedChangelogId();
+ long earliestChangelogId = changelogManager.earliestLongLivedChangelogId();
+
+ // Delete a middle changelog to simulate concurrent deletion
+ long middleId = (earliestChangelogId + latestChangelogId) / 2;
+ assertThat(fileIO.exists(changelogManager.longLivedChangelogPath(middleId))).isTrue();
+ fileIO.deleteQuietly(changelogManager.longLivedChangelogPath(middleId));
+
+ ExpireChangelogImpl expire =
+ (ExpireChangelogImpl) table.newExpireChangelog().config(expireConfig);
+
+ // should not throw even though a middle changelog is missing
+ assertThatCode(expire::expire).doesNotThrowAnyException();
+
+ // earliest should be advanced past the deleted range
+ assertThat(changelogManager.earliestLongLivedChangelogId()).isEqualTo(latestChangelogId);
+ }
}