Skip to content

[core] Fix snapshot lookup for duplicate commit timestamps #9437

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ArnavBalyan wants to merge 4 commits into
base: master
Choose a base branch
from
Open

[core] Fix snapshot lookup for duplicate commit timestamps #9437

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
23da5fb
update
ArnavBalyan Aug 27, 2026
857949d
update
ArnavBalyan Aug 30, 2026
f1b8d3e
update
ArnavBalyan Sep 14, 2026
47c0039
update
ArnavBalyan Sep 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
117 changes: 95 additions & 22 deletions paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,8 @@
/** Manager for {@link Snapshot}, providing utility methods related to paths and snapshot hints. */
public class SnapshotManager implements Serializable {

private static final int TIME_TRAVEL_SNAPSHOT_LOOKUP_RETRY_NUM = 3;

private static final long serialVersionUID = 1L;

private static final Logger LOG = LoggerFactory.getLogger(SnapshotManager.class);
Expand All @@ -67,6 +69,9 @@ public class SnapshotManager implements Serializable {

public static final int EARLIEST_SNAPSHOT_DEFAULT_RETRY_NUM = 300;

private static final int SNAPSHOT_EXISTS_MAX_ATTEMPTS = 3;
private static final long SNAPSHOT_EXISTS_RETRY_INTERVAL_MILLIS = 1_000L;

private final FileIO fileIO;
private final Path tablePath;
private final String branch;
Expand Down Expand Up @@ -147,13 +152,44 @@ public Snapshot tryGetSnapshot(long snapshotId) throws FileNotFoundException {

public boolean snapshotExists(long snapshotId) {
Path path = snapshotPath(snapshotId);
try {
return fileIO.exists(path);
} catch (IOException e) {
throw new RuntimeException(
"Failed to determine if snapshot #" + snapshotId + " exists in path " + path,
e);
IOException failure = null;
for (int attempt = 1; attempt <= SNAPSHOT_EXISTS_MAX_ATTEMPTS; attempt++) {

@JingsongLi JingsongLi Sep 19, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Split the unrelated snapshotExists retry into its own PR

This change also rewrites snapshotExists to retry fileIO.exists three times with a 1s blocking sleep (SNAPSHOT_EXISTS_MAX_ATTEMPTS, SNAPSHOT_EXISTS_RETRY_INTERVAL_MILLIS). It is unrelated to the duplicate-timestamp lookup, is not mentioned in the PR description, and has no test. It applies to a method with a dozen-plus production call sites, including commit paths (StrictModeChecker, IcebergCommitCallback), streaming poll loops (NextSnapshotFetcher, AppendCompactCoordinator, TagAutoCreation) and branch/tag maintenance (FileSystemBranchManager, TagManager).

Trigger: any caller where fileIO.exists throws persistently (permission or short-read errors, an object store under throttling).

Impact: every snapshotExists call then blocks up to 2s, logs two WARNs and issues 3x the number of exists requests before failing with the same RuntimeException as before. So a focused race-fix PR silently changes failure latency and request volume for commit/streaming paths. Note that the lookup fix retries because it must re-resolve concurrent rollback/expiration; this retry only re-issues an identical exists, so its benefit is limited to transient FS errors, which one would not expect on the same path the PR is about.

Fix: keep this PR to the duplicate-timestamp lookup and move the retry to a separate PR with its own rationale and test (e.g. a FileIO that throws once and then succeeds). If it is kept here, please add a test and consider a shorter interval (the existing tryFromPath retry uses 200ms).

try {
return fileIO.exists(path);
} catch (IOException e) {
failure = e;
if (attempt == SNAPSHOT_EXISTS_MAX_ATTEMPTS) {
break;
}
LOG.warn(
"Failed to check whether snapshot #{} exists at {} (attempt {}/{}). Retrying.",
snapshotId,
path,
attempt,
SNAPSHOT_EXISTS_MAX_ATTEMPTS,
e);
try {
Thread.sleep(SNAPSHOT_EXISTS_RETRY_INTERVAL_MILLIS);
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
throw new RuntimeException(
"Interrupted while checking whether snapshot #"
+ snapshotId
+ " exists at "
+ path,
ie);
}
}
}
throw new RuntimeException(
"Failed to check whether snapshot #"
+ snapshotId
+ " exists at "
+ path
+ " after "
+ SNAPSHOT_EXISTS_MAX_ATTEMPTS
+ " attempts.",
failure);
}

public void deleteSnapshot(long snapshotId) {
Expand Down Expand Up @@ -360,30 +396,48 @@ public long repairEarliestSnapshot(long snapshotId) {
* mills. If there is no such a snapshot, returns null.
*/
public @Nullable Snapshot earlierOrEqualTimeMills(long timestampMills) {
Long latest = latestSnapshotId();
if (latest == null) {
FileNotFoundException exception = null;
for (int retry = 0; retry <= TIME_TRAVEL_SNAPSHOT_LOOKUP_RETRY_NUM; retry++) {
try {
return earlierOrEqualTimeMillsOnce(timestampMills);
} catch (FileNotFoundException e) {
exception = e;
}
}
throw new UncheckedIOException(
"Failed to find snapshot while searching by commit time.", exception);
}

private @Nullable Snapshot earlierOrEqualTimeMillsOnce(long timestampMills)
throws FileNotFoundException {
Snapshot latestSnapshot = latestSnapshot();
if (latestSnapshot == null) {
return null;
}
if (latestSnapshot.timeMillis() <= timestampMills) {
return latestSnapshot;
}

Snapshot earliestSnapShot = earliestSnapshot(latest);
if (earliestSnapShot == null || earliestSnapShot.timeMillis() > timestampMills) {
Snapshot earliestSnapshot = earliestSnapshot(latestSnapshot.id());
if (earliestSnapshot == null || earliestSnapshot.timeMillis() > timestampMills) {
return null;
}
long earliest = earliestSnapShot.id();
long earliest = earliestSnapshot.id();
long latest = latestSnapshot.id() - 1;

Snapshot finalSnapshot = null;
while (earliest <= latest) {
long mid = earliest + (latest - earliest) / 2; // Avoid overflow
Snapshot snapshot = snapshot(mid);
Snapshot snapshot = tryGetSnapshot(mid);
long commitTime = snapshot.timeMillis();
if (commitTime > timestampMills) {
latest = mid - 1; // Search in the left half
} else if (commitTime < timestampMills) {
earliest = mid + 1; // Search in the right half
finalSnapshot = snapshot;
} else {
finalSnapshot = snapshot; // Found the exact match
break;
finalSnapshot = snapshot;
earliest = mid + 1;

@JingsongLi JingsongLi Aug 28, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Refresh the upper bound before searching past an exact match

latest is only an ID captured before the search. A concurrent rollback can update the latest hint and delete that snapshot after latestSnapshotId() returns. With snapshots 0/1/2 sharing this timestamp, if rollback removes snapshot 2 after we capture it, the first probe hits snapshot 1; this new continuation then probes snapshot 2 and throws Snapshot file ... does not exist, although snapshot 1 is still the correct answer. The old break returned snapshot 1.

Please resolve the upper boundary through the existing live-snapshot path (for example, latestSnapshot() plus an early boundary return) and add a rollback-race regression test.

@JingsongLi JingsongLi Sep 2, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Retry if rollback moves the upper bound after the first equal probe

Resolving latestSnapshot only closes the race before the binary search starts. With 0@(t-1), 1@t, 2@t, 3@t, 4@(t+1), earlierOrEqualTimeMills(t) first records snapshot 1. If RollbackHelper retains snapshot 1 at that point, it commits LATEST=1 and deletes snapshots 4 through 2. This branch advances earliest to 2, so the next snapshot(2) throws even though snapshot 1 is still the correct result. I reproduced this deterministically by triggering rollback immediately after the first equal probe.

Please read probes through tryGetSnapshot and, when a post-match probe disappears, refresh the live upper bound and restart with a bounded retry. A regression test should trigger rollback after the first equal midpoint has been read.

}
}
return finalSnapshot;
Expand All @@ -394,29 +448,48 @@ public long repairEarliestSnapshot(long snapshotId) {
* If there is no such a snapshot, returns null.
*/
public @Nullable Snapshot laterOrEqualTimeMills(long timestampMills) {
Long earliest = earliestSnapshotId();
Long latest = latestSnapshotId();
if (earliest == null || latest == null) {
FileNotFoundException exception = null;
for (int retry = 0; retry <= TIME_TRAVEL_SNAPSHOT_LOOKUP_RETRY_NUM; retry++) {
try {
return laterOrEqualTimeMillsOnce(timestampMills);
} catch (FileNotFoundException e) {
exception = e;
}
}
throw new UncheckedIOException(
"Failed to find snapshot while searching by commit time.", exception);
}

private @Nullable Snapshot laterOrEqualTimeMillsOnce(long timestampMills)
throws FileNotFoundException {
Snapshot latestSnapshot = latestSnapshot();
if (latestSnapshot == null || latestSnapshot.timeMillis() < timestampMills) {
return null;
}

Snapshot latestSnapShot = snapshot(latest);
if (latestSnapShot.timeMillis() < timestampMills) {
Snapshot earliestSnapshot = earliestSnapshot(latestSnapshot.id());
if (earliestSnapshot == null) {
return null;
}
if (earliestSnapshot.timeMillis() >= timestampMills) {
return earliestSnapshot;
}

long earliest = earliestSnapshot.id() + 1;
long latest = latestSnapshot.id();
Snapshot finalSnapshot = null;
while (earliest <= latest) {
long mid = earliest + (latest - earliest) / 2; // Avoid overflow
Snapshot snapshot = snapshot(mid);
Snapshot snapshot = tryGetSnapshot(mid);
long commitTime = snapshot.timeMillis();
if (commitTime > timestampMills) {
latest = mid - 1; // Search in the left half
finalSnapshot = snapshot;
} else if (commitTime < timestampMills) {
earliest = mid + 1; // Search in the right half
} else {
finalSnapshot = snapshot; // Found the exact match
break;
finalSnapshot = snapshot;
latest = mid - 1;

@JingsongLi JingsongLi Aug 28, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Refresh the lower bound before searching past an exact match

earliest can become stale when expiration runs between earliestSnapshotId() and the binary search. With snapshots 0/1/2 sharing this timestamp, expiration can delete snapshot 0 after the ID is read; the search first hits snapshot 1, then this new continuation probes snapshot 0 and fails even though snapshot 1 is the earliest surviving match. The old break returned snapshot 1.

Please obtain the lower bound through earliestSnapshot(latest), which already retries concurrent earliest deletion, return it directly when it satisfies the query, and cover this race in the test.

@JingsongLi JingsongLi Sep 2, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Retry if expiration moves the lower bound after the first equal probe

Resolving the initial earliest snapshot does not cover expiration that starts during the binary search. With 0@(t-1), 1@t, 2@t, 3@t, 4@(t+1), laterOrEqualTimeMills(t) first records snapshot 2. If expiration then retains snapshot 2 and deletes snapshots 0 and 1, this branch moves latest to 1 and the next snapshot(1) throws, although snapshot 2 is the correct earliest surviving result. I reproduced this deterministically by expiring the prefix immediately after the first equal probe.

Please handle a missing post-match probe by refreshing the live lower bound and restarting with a bounded retry, and add a regression test where expiration runs after the first equal midpoint has been read.

}
}
return finalSnapshot;
Expand Down
190 changes: 190 additions & 0 deletions paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -423,6 +423,114 @@ public void testLaterOrEqualTimeMills() throws IOException {
assertThat(snapshotManager.laterOrEqualTimeMills(millis + 10001)).isNull();
}

@Test
public void testEarlierOrEqualTimeMillsWithDuplicateCommitTimes() throws IOException {
long millis = 1684726826L;
FileIO localFileIO = LocalFileIO.create();
SnapshotManager snapshotManager =
newSnapshotManager(localFileIO, new Path(tempDir.toString()));
for (long i = 0; i < 3; i++) {
Snapshot snapshot = createSnapshotWithMillis(i, millis);
localFileIO.tryToWriteAtomic(snapshotManager.snapshotPath(i), snapshot.toJson());
}

assertThat(snapshotManager.earlierOrEqualTimeMills(millis).id()).isEqualTo(2);

@JingsongLi JingsongLi Sep 2, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Exercise an interior duplicate-timestamp range

These duplicate-time tests give every snapshot the same timestamp, so earlierOrEqualTimeMills returns through the latest-snapshot shortcut and laterOrEqualTimeMills returns through the earliest-snapshot shortcut. Neither test executes the changed equality branches inside the binary searches. The two race tests added below also end at those shortcuts. I temporarily restored both equality branches to the old break, while keeping the new boundary logic, and all four tests introduced by this PR still passed.

Please add a history such as [t-1, t, t, t, t+1] and assert that earlier-or-equal returns the last t snapshot while later-or-equal returns the first one.

@ArnavBalyan ArnavBalyan Sep 14, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks @JingsongLi! updated both paths to restart with refreshed snapshot boundaries when concurrent rollback or expiration removes a snapshot during the search

}

@Test
public void testLaterOrEqualTimeMillsWithDuplicateCommitTimes() throws IOException {
long millis = 1684726826L;
FileIO localFileIO = LocalFileIO.create();
SnapshotManager snapshotManager =
newSnapshotManager(localFileIO, new Path(tempDir.toString()));
for (long i = 0; i < 3; i++) {
Snapshot snapshot = createSnapshotWithMillis(i, millis);
localFileIO.tryToWriteAtomic(snapshotManager.snapshotPath(i), snapshot.toJson());
}

assertThat(snapshotManager.laterOrEqualTimeMills(millis).id()).isEqualTo(0);
}

@Test
public void testTimeMillsWithInteriorDuplicateCommitTimes() throws IOException {
long millis = 1684726826L;
FileIO localFileIO = LocalFileIO.create();
SnapshotManager snapshotManager =
newSnapshotManager(localFileIO, new Path(tempDir.toString()));
long[] commitTimes = {millis - 1, millis, millis, millis, millis + 1};
for (int i = 0; i < commitTimes.length; i++) {
Snapshot snapshot = createSnapshotWithMillis(i, commitTimes[i]);
localFileIO.tryToWriteAtomic(snapshotManager.snapshotPath(i), snapshot.toJson());
}

assertThat(snapshotManager.earlierOrEqualTimeMills(millis).id()).isEqualTo(3);
assertThat(snapshotManager.laterOrEqualTimeMills(millis).id()).isEqualTo(1);
}

@Test
public void testEarlierOrEqualTimeMillsWithConcurrentRollback() throws IOException {
long millis = 1684726826L;
FileIO localFileIO = LocalFileIO.create();
SnapshotManager snapshotManager =
new LatestSnapshotRollbackRaceManager(localFileIO, new Path(tempDir.toString()));
for (long i = 0; i < 3; i++) {
Snapshot snapshot = createSnapshotWithMillis(i, millis);
localFileIO.tryToWriteAtomic(snapshotManager.snapshotPath(i), snapshot.toJson());
}
snapshotManager.commitLatestHint(2);

assertThat(snapshotManager.earlierOrEqualTimeMills(millis).id()).isEqualTo(1);
}

@Test
public void testLaterOrEqualTimeMillsWithConcurrentExpiration() throws IOException {
long millis = 1684726826L;
FileIO localFileIO = LocalFileIO.create();
SnapshotManager snapshotManager =
new TestSnapshotManager(localFileIO, new Path(tempDir.toString()), true);
for (long i = 0; i < 3; i++) {
Snapshot snapshot = createSnapshotWithMillis(i, millis);
localFileIO.tryToWriteAtomic(snapshotManager.snapshotPath(i), snapshot.toJson());
}
snapshotManager.commitEarliestHint(0);

assertThat(snapshotManager.laterOrEqualTimeMills(millis).id()).isEqualTo(1);
}

@Test
public void testEarlierOrEqualTimeMillsWithRollbackAfterEqualProbe() throws IOException {
long millis = 1684726826L;
FileIO localFileIO = LocalFileIO.create();
SnapshotManager snapshotManager =
new PostMatchRollbackSnapshotManager(
localFileIO, new Path(tempDir.toString()), 1, 4);
long[] commitTimes = {millis - 1, millis, millis, millis, millis + 1};
for (int i = 0; i < commitTimes.length; i++) {
Snapshot snapshot = createSnapshotWithMillis(i, commitTimes[i]);
localFileIO.tryToWriteAtomic(snapshotManager.snapshotPath(i), snapshot.toJson());
}
snapshotManager.commitLatestHint(4);

assertThat(snapshotManager.earlierOrEqualTimeMills(millis).id()).isEqualTo(1);
}

@Test
public void testLaterOrEqualTimeMillsWithExpirationAfterEqualProbe() throws IOException {
long millis = 1684726826L;
FileIO localFileIO = LocalFileIO.create();
SnapshotManager snapshotManager =
new PostMatchExpirationSnapshotManager(
localFileIO, new Path(tempDir.toString()), 2);
long[] commitTimes = {millis - 1, millis, millis, millis, millis + 1};
for (int i = 0; i < commitTimes.length; i++) {
Snapshot snapshot = createSnapshotWithMillis(i, commitTimes[i]);
localFileIO.tryToWriteAtomic(snapshotManager.snapshotPath(i), snapshot.toJson());
}
snapshotManager.commitEarliestHint(0);

assertThat(snapshotManager.laterOrEqualTimeMills(millis).id()).isEqualTo(2);
}

@ParameterizedTest
@ValueSource(booleans = {true, false})
public void testLaterOrEqualWatermark(boolean isRaceCondition) throws IOException {
Expand Down Expand Up @@ -955,4 +1063,86 @@ private LatestSnapshotRaceManager(FileIO fileIO, Path tablePath, long millis) {
return snapshotId;
}
}

/** Simulates a rollback after finding the latest snapshot ID. */
private static class LatestSnapshotRollbackRaceManager extends SnapshotManager {
private boolean rollbackLatestSnapshot = true;

private LatestSnapshotRollbackRaceManager(FileIO fileIO, Path tablePath) {
super(fileIO, tablePath, DEFAULT_MAIN_BRANCH, null, null);
}

@Override
public @Nullable Long latestSnapshotIdFromFileSystem() {
Long snapshotId = super.latestSnapshotIdFromFileSystem();
if (snapshotId != null && rollbackLatestSnapshot) {
try {
commitLatestHint(snapshotId - 1);
fileIO().delete(snapshotPath(snapshotId), true);
} catch (IOException e) {
throw new RuntimeException(e);
}
rollbackLatestSnapshot = false;
}
return snapshotId;
}
}

private static class PostMatchRollbackSnapshotManager extends SnapshotManager {
private final long retainedSnapshotId;
private final long latestSnapshotId;
private boolean rollback = true;

private PostMatchRollbackSnapshotManager(
FileIO fileIO, Path tablePath, long retainedSnapshotId, long latestSnapshotId) {
super(fileIO, tablePath, DEFAULT_MAIN_BRANCH, null, null);
this.retainedSnapshotId = retainedSnapshotId;
this.latestSnapshotId = latestSnapshotId;
}

@Override
public Snapshot tryGetSnapshot(long snapshotId) throws FileNotFoundException {
Snapshot snapshot = super.tryGetSnapshot(snapshotId);
if (rollback && snapshotId == retainedSnapshotId) {
try {
commitLatestHint(retainedSnapshotId);
for (long id = retainedSnapshotId + 1; id <= latestSnapshotId; id++) {
fileIO().delete(snapshotPath(id), true);
}
} catch (IOException e) {
throw new RuntimeException(e);
}
rollback = false;
}
return snapshot;
}
}

private static class PostMatchExpirationSnapshotManager extends SnapshotManager {
private final long retainedSnapshotId;
private boolean expire = true;

private PostMatchExpirationSnapshotManager(
FileIO fileIO, Path tablePath, long retainedSnapshotId) {
super(fileIO, tablePath, DEFAULT_MAIN_BRANCH, null, null);
this.retainedSnapshotId = retainedSnapshotId;
}

@Override
public Snapshot tryGetSnapshot(long snapshotId) throws FileNotFoundException {
Snapshot snapshot = super.tryGetSnapshot(snapshotId);
if (expire && snapshotId == retainedSnapshotId) {
try {
commitEarliestHint(retainedSnapshotId);
for (long id = 0; id < retainedSnapshotId; id++) {
fileIO().delete(snapshotPath(id), true);
}
} catch (IOException e) {
throw new RuntimeException(e);
}
expire = false;
}
return snapshot;
}
}
}
Loading