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 2 commits into
base: master
Choose a base branch
from
+105 −14
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
23da5fb
update
ArnavBalyan Aug 27, 2026
857949d
update
ArnavBalyan Aug 30, 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
37 changes: 23 additions & 14 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 @@ -360,16 +360,20 @@ 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) {
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) {
Expand All @@ -382,8 +386,8 @@ public long repairEarliestSnapshot(long snapshotId) {
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.

}
}
return finalSnapshot;
Expand All @@ -394,16 +398,21 @@ 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) {
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
Expand All @@ -415,8 +424,8 @@ public long repairEarliestSnapshot(long snapshotId) {
} 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.

}
}
return finalSnapshot;
Expand Down
82 changes: 82 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,64 @@ 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);
}

@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 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);
}

@ParameterizedTest
@ValueSource(booleans = {true, false})
public void testLaterOrEqualWatermark(boolean isRaceCondition) throws IOException {
Expand Down Expand Up @@ -955,4 +1013,28 @@ 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;
}
}
}
Loading