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 f7d95930e8 [core] Fix SnapshotManager watermark binary searches with
null watermarks (#9037)
f7d95930e8 is described below
commit f7d95930e86d5e9578883f962e77aca8b6b2932d
Author: kid <[email protected]>
AuthorDate: Fri Aug 7 13:33:14 2026 +0800
[core] Fix SnapshotManager watermark binary searches with null watermarks
(#9037)
---
.../org/apache/paimon/utils/SnapshotManager.java | 100 ++++----
.../source/snapshot/WatermarkTimeTravelTest.java | 279 +++++++++++++++++++++
.../apache/paimon/utils/SnapshotManagerTest.java | 199 ++++++++++++++-
3 files changed, 519 insertions(+), 59 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java
b/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java
index 984a98902e..23694934e8 100644
--- a/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java
@@ -411,9 +411,7 @@ public class SnapshotManager implements Serializable {
public @Nullable Snapshot earlierOrEqualWatermark(long watermark) {
Long latest = latestSnapshotId();
- // If latest == Long.MIN_VALUE don't need next binary search for
watermark
- // which can reduce IO cost with snapshot
- if (latest == null || snapshot(latest).watermark() == Long.MIN_VALUE) {
+ if (latest == null) {
return null;
}
@@ -423,45 +421,35 @@ public class SnapshotManager implements Serializable {
}
long earliest = earliestSnapShot.id();
- Long earliestWatermark = null;
- // find the first snapshot with watermark
- if ((earliestWatermark = earliestSnapShot.watermark()) == null) {
- while (earliest < latest) {
- earliest++;
- earliestWatermark = snapshot(earliest).watermark();
- if (earliestWatermark != null) {
- break;
- }
- }
+ // find the first snapshot with a real watermark
+ Long earliestWatermark = earliestSnapShot.watermark();
+ while (isMissingWatermark(earliestWatermark) && earliest < latest) {
+ earliest++;
+ earliestWatermark = snapshot(earliest).watermark();
}
- if (earliestWatermark == null) {
+ if (isMissingWatermark(earliestWatermark) || earliestWatermark >
watermark) {
return null;
}
-
- if (earliestWatermark >= watermark) {
- return snapshot(earliest);
- }
Snapshot finalSnapshot = null;
while (earliest <= latest) {
long mid = earliest + (latest - earliest) / 2; // Avoid overflow
- Snapshot snapshot = snapshot(mid);
+ // A snapshot without a watermark takes the ordering position of
the
+ // nearest earlier snapshot (within the search window) that
carries one
+ long pos = mid;
+ Snapshot snapshot = snapshot(pos);
Long commitWatermark = snapshot.watermark();
- if (commitWatermark == null) {
- // find the first snapshot with watermark
- while (mid >= earliest) {
- mid--;
- commitWatermark = snapshot(mid).watermark();
- if (commitWatermark != null) {
- break;
- }
- }
+ while (isMissingWatermark(commitWatermark) && pos > earliest) {
+ pos--;
+ snapshot = snapshot(pos);
+ commitWatermark = snapshot.watermark();
}
- if (commitWatermark == null) {
+ if (isMissingWatermark(commitWatermark)) {
+ // No snapshot with watermark in [earliest, mid]: skip the
range
earliest = mid + 1;
} else {
if (commitWatermark > watermark) {
- latest = mid - 1; // Search in the left half
+ latest = pos - 1; // Search in the left half
} else if (commitWatermark < watermark) {
earliest = mid + 1; // Search in the right half
finalSnapshot = snapshot;
@@ -476,9 +464,7 @@ public class SnapshotManager implements Serializable {
public @Nullable Snapshot laterOrEqualWatermark(long watermark) {
Long latest = latestSnapshotId();
- // If latest == Long.MIN_VALUE don't need next binary search for
watermark
- // which can reduce IO cost with snapshot
- if (latest == null || snapshot(latest).watermark() == Long.MIN_VALUE) {
+ if (latest == null) {
return null;
}
@@ -488,18 +474,13 @@ public class SnapshotManager implements Serializable {
}
long earliest = earliestSnapShot.id();
- Long earliestWatermark = null;
- // find the first snapshot with watermark
- if ((earliestWatermark = earliestSnapShot.watermark()) == null) {
- while (earliest < latest) {
- earliest++;
- earliestWatermark = snapshot(earliest).watermark();
- if (earliestWatermark != null) {
- break;
- }
- }
+ // find the first snapshot with a real watermark
+ Long earliestWatermark = earliestSnapShot.watermark();
+ while (isMissingWatermark(earliestWatermark) && earliest < latest) {
+ earliest++;
+ earliestWatermark = snapshot(earliest).watermark();
}
- if (earliestWatermark == null) {
+ if (isMissingWatermark(earliestWatermark)) {
return null;
}
@@ -510,23 +491,22 @@ public class SnapshotManager implements Serializable {
while (earliest <= latest) {
long mid = earliest + (latest - earliest) / 2; // Avoid overflow
- Snapshot snapshot = snapshot(mid);
+ // A snapshot without a watermark takes the ordering position of
the
+ // nearest earlier snapshot (within the search window) that
carries one
+ long pos = mid;
+ Snapshot snapshot = snapshot(pos);
Long commitWatermark = snapshot.watermark();
- if (commitWatermark == null) {
- // find the first snapshot with watermark
- while (mid >= earliest) {
- mid--;
- commitWatermark = snapshot(mid).watermark();
- if (commitWatermark != null) {
- break;
- }
- }
+ while (isMissingWatermark(commitWatermark) && pos > earliest) {
+ pos--;
+ snapshot = snapshot(pos);
+ commitWatermark = snapshot.watermark();
}
- if (commitWatermark == null) {
+ if (isMissingWatermark(commitWatermark)) {
+ // No snapshot with watermark in [earliest, mid]: skip the
range
earliest = mid + 1;
} else {
if (commitWatermark > watermark) {
- latest = mid - 1; // Search in the left half
+ latest = pos - 1; // Search in the left half
finalSnapshot = snapshot;
} else if (commitWatermark < watermark) {
earliest = mid + 1; // Search in the right half
@@ -539,6 +519,14 @@ public class SnapshotManager implements Serializable {
return finalSnapshot;
}
+ /**
+ * Both {@code null} and {@link Long#MIN_VALUE} mean that the snapshot
carries no watermark (the
+ * latter is written as a sentinel by engines without watermark semantics).
+ */
+ private static boolean isMissingWatermark(@Nullable Long watermark) {
+ return watermark == null || watermark == Long.MIN_VALUE;
+ }
+
public long snapshotCount() throws IOException {
return snapshotIdStream().count();
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/source/snapshot/WatermarkTimeTravelTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/source/snapshot/WatermarkTimeTravelTest.java
new file mode 100644
index 0000000000..569e2d5bd4
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/snapshot/WatermarkTimeTravelTest.java
@@ -0,0 +1,279 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.table.source.snapshot;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.manifest.ManifestCommittable;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.StreamTableCommit;
+import org.apache.paimon.table.sink.StreamTableWrite;
+import org.apache.paimon.table.sink.TableCommitImpl;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import javax.annotation.Nullable;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.paimon.CoreOptions.SCAN_WATERMARK;
+import static
org.apache.paimon.testutils.assertj.PaimonAssertions.anyCauseMatches;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * End-to-end tests for watermark time travel: snapshots are produced through
the real commit path
+ * ({@link TableCommitImpl} with {@link ManifestCommittable} watermarks, the
same entry the Flink
+ * committer uses) and reads go through the real batch scan link ({@code
scan.watermark} →
+ * {@link StaticFromWatermarkStartingScanner} → {@code
SnapshotManager.laterOrEqualWatermark}).
+ */
+public class WatermarkTimeTravelTest extends ScannerTestBase {
+
+ @Test
+ public void testScanWatermarkWithDenseWatermarks() throws Exception {
+ createAppendOnlyTableWithoutCompaction();
+ StreamTableWrite write = table.newWrite(commitUser);
+ StreamTableCommit commit = table.newCommit(commitUser);
+
+ // snapshots 1, 2, 3 carry watermarks 100, 200, 300
+ for (int i = 0; i < 3; i++) {
+ commitRow(write, commit, i, 100L * (i + 1), i);
+ }
+
+ assertThat(scanFromWatermark(50)).hasSameElementsAs(Arrays.asList("+I
1|0|0"));
+ assertThat(scanFromWatermark(150))
+ .hasSameElementsAs(Arrays.asList("+I 1|0|0", "+I 1|1|100"));
+ assertThat(scanFromWatermark(200))
+ .hasSameElementsAs(Arrays.asList("+I 1|0|0", "+I 1|1|100"));
+ assertThat(scanFromWatermark(250))
+ .hasSameElementsAs(Arrays.asList("+I 1|0|0", "+I 1|1|100", "+I
1|2|200"));
+ assertNoSnapshotForWatermark(301);
+
+ write.close();
+ commit.close();
+ }
+
+ @Test
+ public void testScanWatermarkOnTableWithoutWatermarks() throws Exception {
+ createAppendOnlyTableWithoutCompaction();
+ StreamTableWrite write = table.newWrite(commitUser);
+ StreamTableCommit commit = table.newCommit(commitUser);
+
+ // snapshots 1, 2, 3 carry no watermark field, like any pure
batch-written table
+ for (int i = 0; i < 3; i++) {
+ commitRow(write, commit, i, null, i);
+ }
+
+ // the defective guard unboxed the null watermark and threw a raw
NullPointerException;
+ // the fix makes it a clean, actionable error
+ assertNoSnapshotForWatermark(100);
+
+ write.close();
+ commit.close();
+ }
+
+ @Test
+ @Timeout(60) // the defective search loops forever on this layout; the fix
must terminate
+ public void testScanWatermarkWithInterleavedNullWatermarks() throws
Exception {
+ createAppendOnlyTableWithoutCompaction();
+ StreamTableWrite write = table.newWrite(commitUser);
+ StreamTableCommit commit = table.newCommit(commitUser);
+
+ // snapshots 1..10, then snapshots 1, 5, 10 are assigned watermarks
100, 200, 300
+ for (int i = 0; i < 10; i++) {
+ commitRow(write, commit, i, null, i);
+ }
+ patchWatermark(1, 100L);
+ patchWatermark(5, 200L);
+ patchWatermark(10, 300L);
+
+ assertThat(scanFromWatermark(50)).hasSameElementsAs(Arrays.asList("+I
1|0|0"));
+ assertThat(scanFromWatermark(150))
+ .hasSameElementsAs(
+ Arrays.asList(
+ "+I 1|0|0",
+ "+I 1|1|100",
+ "+I 1|2|200",
+ "+I 1|3|300",
+ "+I 1|4|400"));
+ assertThat(scanFromWatermark(250))
+ .hasSameElementsAs(
+ Arrays.asList(
+ "+I 1|0|0",
+ "+I 1|1|100",
+ "+I 1|2|200",
+ "+I 1|3|300",
+ "+I 1|4|400",
+ "+I 1|5|500",
+ "+I 1|6|600",
+ "+I 1|7|700",
+ "+I 1|8|800",
+ "+I 1|9|900"));
+ assertNoSnapshotForWatermark(301);
+
+ write.close();
+ commit.close();
+ }
+
+ @Test
+ public void testScanWatermarkExactMatchWithInterleavedNullWatermarks()
throws Exception {
+ createAppendOnlyTableWithoutCompaction();
+ StreamTableWrite write = table.newWrite(commitUser);
+ StreamTableCommit commit = table.newCommit(commitUser);
+
+ // snapshots 1..5, then snapshots 1, 2, 5 are assigned watermarks 100,
150, 300
+ for (int i = 0; i < 5; i++) {
+ commitRow(write, commit, i, null, i);
+ }
+ patchWatermark(1, 100L);
+ patchWatermark(2, 150L);
+ patchWatermark(5, 300L);
+
+ // the exact match is snapshot 2; the defective search returned the
null-watermark
+ // snapshot 3 instead, silently including commit 2's row
+ assertThat(scanFromWatermark(150))
+ .hasSameElementsAs(Arrays.asList("+I 1|0|0", "+I 1|1|100"));
+
+ write.close();
+ commit.close();
+ }
+
+ @Test
+ public void testRollbackToWatermarkBelowMinimum() throws Exception {
+ createAppendOnlyTableWithoutCompaction();
+ StreamTableWrite write = table.newWrite(commitUser);
+ StreamTableCommit commit = table.newCommit(commitUser);
+
+ // snapshots 1, 2, 3 carry watermarks 100, 200, 300
+ for (int i = 0; i < 3; i++) {
+ commitRow(write, commit, i, 100L * (i + 1), i);
+ }
+
+ // the rollback_to_watermark procedures do earlierOrEqualWatermark +
checkNotNull +
+ // rollbackTo. The defective inverted early-return handed them
snapshot 1 (watermark
+ // 100 > 50), so the procedure rolled back to it and deleted snapshots
2 and 3;
+ // the fix returns null so the procedure rejects the request and the
table stays intact
+ SnapshotManager snapshotManager = table.snapshotManager();
+ assertThat(snapshotManager.earlierOrEqualWatermark(50)).isNull();
+ assertThat(snapshotManager.latestSnapshotId()).isEqualTo(3);
+
+ // a request within range still rolls back correctly
+ Snapshot target = snapshotManager.earlierOrEqualWatermark(150);
+ assertThat(target.id()).isEqualTo(1);
+ table.rollbackTo(target.id());
+ assertThat(snapshotManager.latestSnapshotId()).isEqualTo(1);
+ assertThat(getResult(table.newRead(), table.newScan().plan().splits()))
+ .hasSameElementsAs(Arrays.asList("+I 1|0|0"));
+
+ write.close();
+ commit.close();
+ }
+
+ // ------------------------------------------------------------------
+ // helpers
+ // ------------------------------------------------------------------
+
+ /**
+ * Creates an append-only table that never compacts, so the tests control
the exact snapshot
+ * layout: every commit produces exactly one snapshot, with sequential ids.
+ */
+ private void createAppendOnlyTableWithoutCompaction() throws Exception {
+ Options conf = new Options();
+ conf.set(CoreOptions.WRITE_ONLY, true);
+ createAppendOnlyTable(conf);
+ }
+
+ /** Commits one row {@code (1, value, 100 * value)} as one snapshot with
the given watermark. */
+ private void commitRow(
+ StreamTableWrite write,
+ StreamTableCommit commit,
+ long identifier,
+ @Nullable Long watermark,
+ int value)
+ throws Exception {
+ write.write(rowData(1, value, 100L * value));
+ List<CommitMessage> messages = write.prepareCommit(true, identifier);
+ if (watermark == null) {
+ commit.commit(identifier, messages);
+ } else {
+ ManifestCommittable committable = new
ManifestCommittable(identifier, watermark);
+ messages.forEach(committable::addFileCommittable);
+ ((TableCommitImpl) commit).commit(committable);
+ }
+ }
+
+ private List<String> scanFromWatermark(long watermark) throws Exception {
+ Map<String, String> dynamicOptions = new HashMap<>();
+ dynamicOptions.put(SCAN_WATERMARK.key(), String.valueOf(watermark));
+ List<Split> splits =
table.copy(dynamicOptions).newScan().plan().splits();
+ return getResult(table.newRead(), splits);
+ }
+
+ private void assertNoSnapshotForWatermark(long watermark) {
+ assertThatThrownBy(() -> scanFromWatermark(watermark))
+ .satisfies(
+ anyCauseMatches(
+ RuntimeException.class,
+ "There is currently no snapshot later than or
equal to watermark"));
+ }
+
+ /**
+ * Rewrites a snapshot file with the given watermark field set. Java
commits carry the previous
+ * watermark forward ({@code FileStoreCommitImpl}), so a pure-Java history
can never hold
+ * interleaved null watermarks; this produces on disk exactly what a
mixed-engine history looks
+ * like, e.g. Flink streaming commits (watermark-bearing) interleaved with
paimon-rust /
+ * pypaimon appends (no watermark field).
+ */
+ private void patchWatermark(long snapshotId, long watermark) throws
Exception {
+ SnapshotManager snapshotManager = table.snapshotManager();
+ Snapshot s = snapshotManager.snapshot(snapshotId);
+ Snapshot patched =
+ new Snapshot(
+ s.id(),
+ s.schemaId(),
+ s.baseManifestList(),
+ s.baseManifestListSize(),
+ s.deltaManifestList(),
+ s.deltaManifestListSize(),
+ s.changelogManifestList(),
+ s.changelogManifestListSize(),
+ s.indexManifest(),
+ s.commitUser(),
+ s.commitIdentifier(),
+ s.commitKind(),
+ s.timeMillis(),
+ s.totalRecordCount(),
+ s.deltaRecordCount(),
+ s.changelogRecordCount(),
+ watermark,
+ s.statistics(),
+ s.properties(),
+ s.nextRowId(),
+ s.operation());
+ fileIO.delete(snapshotManager.snapshotPath(snapshotId), false);
+ fileIO.tryToWriteAtomic(snapshotManager.snapshotPath(snapshotId),
patched.toJson());
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java
b/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java
index e44af2cd90..6c09b9a327 100644
--- a/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java
@@ -27,6 +27,7 @@ import org.apache.paimon.table.source.snapshot.TimeTravelUtil;
import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
@@ -37,8 +38,10 @@ import javax.annotation.Nullable;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicReference;
@@ -241,8 +244,13 @@ public class SnapshotManagerTest {
localFileIO.tryToWriteAtomic(snapshotManager.snapshotPath(i),
snapshot.toJson());
}
- assertThat(snapshotManager.earlierOrEqualWatermark(millis + 999).id())
- .isEqualTo(isRaceCondition ? 1 : 0);
+ if (isRaceCondition) {
+ // The earliest snapshot has expired, so no remaining snapshot
carries a
+ // watermark earlier than or equal to the requested one
+ assertThat(snapshotManager.earlierOrEqualWatermark(millis +
999)).isNull();
+ } else {
+ assertThat(snapshotManager.earlierOrEqualWatermark(millis +
999).id()).isEqualTo(0);
+ }
}
@ParameterizedTest
@@ -419,7 +427,7 @@ public class SnapshotManagerTest {
null);
}
- private Snapshot createSnapshotWithMillis(long id, long millis, long
watermark) {
+ private Snapshot createSnapshotWithMillis(long id, long millis, @Nullable
Long watermark) {
return new Snapshot(
id,
0L,
@@ -444,6 +452,14 @@ public class SnapshotManagerTest {
null);
}
+ private static Map<Long, Long> watermarkMap(long... idsAndWatermarks) {
+ Map<Long, Long> watermarks = new HashMap<>();
+ for (int i = 0; i < idsAndWatermarks.length; i += 2) {
+ watermarks.put(idsAndWatermarks[i], idsAndWatermarks[i + 1]);
+ }
+ return watermarks;
+ }
+
private Changelog createChangelogWithMillis(long id, long millis) {
return new Changelog(
new Snapshot(
@@ -662,6 +678,183 @@ public class SnapshotManagerTest {
assertDoesNotThrow(() -> snapshotManager.commitChangelog(changelog,
id));
}
+ @Test
+ @Timeout(60) // the search must terminate; defective code loops forever
here
+ public void testLaterOrEqualWatermarkWithInterleavedNullWatermarks()
throws IOException {
+ FileIO localFileIO = LocalFileIO.create();
+ SnapshotManager snapshotManager =
+ newSnapshotManager(localFileIO, new Path(tempDir.toString()));
+ // create 10 snapshots, only snapshots 0, 4 and 9 carry a watermark
+ Map<Long, Long> watermarks = watermarkMap(0, 100, 4, 200, 9, 300);
+ for (long i = 0; i < 10; i++) {
+ Snapshot snapshot = createSnapshotWithMillis(i, 1000 + i * 1000,
watermarks.get(i));
+ localFileIO.tryToWriteAtomic(snapshotManager.snapshotPath(i),
snapshot.toJson());
+ }
+
+ // the null-watermark fallback must terminate and return snapshot 4
+
assertThat(snapshotManager.laterOrEqualWatermark(150).id()).isEqualTo(4);
+
assertThat(snapshotManager.laterOrEqualWatermark(250).id()).isEqualTo(9);
+ // larger than the largest watermark returns null
+ assertThat(snapshotManager.laterOrEqualWatermark(301)).isNull();
+ // smaller than the smallest watermark returns the first snapshot with
a watermark
+
assertThat(snapshotManager.laterOrEqualWatermark(50).id()).isEqualTo(0);
+ }
+
+ @Test
+ public void testLaterOrEqualWatermarkExactMatchWithNullWatermarks() throws
IOException {
+ FileIO localFileIO = LocalFileIO.create();
+ SnapshotManager snapshotManager =
+ newSnapshotManager(localFileIO, new Path(tempDir.toString()));
+ // create 5 snapshots, snapshots 2 and 3 do not carry a watermark
+ Map<Long, Long> watermarks = watermarkMap(0, 100, 1, 150, 4, 300);
+ for (long i = 0; i < 5; i++) {
+ Snapshot snapshot = createSnapshotWithMillis(i, 1000 + i * 1000,
watermarks.get(i));
+ localFileIO.tryToWriteAtomic(snapshotManager.snapshotPath(i),
snapshot.toJson());
+ }
+
+ // the exact match is snapshot 1, not one of the null-watermark
snapshots
+
assertThat(snapshotManager.laterOrEqualWatermark(150).id()).isEqualTo(1);
+ }
+
+ @Test
+ public void testLaterOrEqualWatermarkWithAllNullWatermarks() throws
IOException {
+ FileIO localFileIO = LocalFileIO.create();
+ SnapshotManager snapshotManager =
+ newSnapshotManager(localFileIO, new Path(tempDir.toString()));
+ // create 5 snapshots without watermark
+ for (long i = 0; i < 5; i++) {
+ Snapshot snapshot = createSnapshotWithMillis(i, 1000 + i * 1000);
+ localFileIO.tryToWriteAtomic(snapshotManager.snapshotPath(i),
snapshot.toJson());
+ }
+
+ assertThat(snapshotManager.laterOrEqualWatermark(100)).isNull();
+ }
+
+ @Test
+ public void testLaterOrEqualWatermarkWithNullWatermarkTail() throws
IOException {
+ FileIO localFileIO = LocalFileIO.create();
+ SnapshotManager snapshotManager =
+ newSnapshotManager(localFileIO, new Path(tempDir.toString()));
+ // create 4 snapshots, snapshots 2 and 3 do not carry a watermark
+ Map<Long, Long> watermarks = watermarkMap(0, 100, 1, 200);
+ for (long i = 0; i < 4; i++) {
+ Snapshot snapshot = createSnapshotWithMillis(i, 1000 + i * 1000,
watermarks.get(i));
+ localFileIO.tryToWriteAtomic(snapshotManager.snapshotPath(i),
snapshot.toJson());
+ }
+
+
assertThat(snapshotManager.laterOrEqualWatermark(150).id()).isEqualTo(1);
+ // the null-watermark tail does not extend the watermark range
+ assertThat(snapshotManager.laterOrEqualWatermark(250)).isNull();
+ }
+
+ @Test
+ public void testEarlierOrEqualWatermarkBelowMinimum() throws IOException {
+ FileIO localFileIO = LocalFileIO.create();
+ SnapshotManager snapshotManager =
+ newSnapshotManager(localFileIO, new Path(tempDir.toString()));
+ // create 3 snapshots with dense watermarks
+ for (long i = 0; i < 3; i++) {
+ Snapshot snapshot = createSnapshotWithMillis(i, 1000 + i * 1000,
100 + i * 100);
+ localFileIO.tryToWriteAtomic(snapshotManager.snapshotPath(i),
snapshot.toJson());
+ }
+
+ // smaller than the smallest watermark returns null
+ assertThat(snapshotManager.earlierOrEqualWatermark(50)).isNull();
+ }
+
+ @Test
+ @Timeout(60) // the search must terminate; defective code loops forever
here
+ public void testEarlierOrEqualWatermarkWithInterleavedNullWatermarks()
throws IOException {
+ FileIO localFileIO = LocalFileIO.create();
+ SnapshotManager snapshotManager =
+ newSnapshotManager(localFileIO, new Path(tempDir.toString()));
+ // create 10 snapshots, only snapshots 0, 4 and 9 carry a watermark
+ Map<Long, Long> watermarks = watermarkMap(0, 100, 4, 200, 9, 300);
+ for (long i = 0; i < 10; i++) {
+ Snapshot snapshot = createSnapshotWithMillis(i, 1000 + i * 1000,
watermarks.get(i));
+ localFileIO.tryToWriteAtomic(snapshotManager.snapshotPath(i),
snapshot.toJson());
+ }
+
+
assertThat(snapshotManager.earlierOrEqualWatermark(150).id()).isEqualTo(0);
+
assertThat(snapshotManager.earlierOrEqualWatermark(250).id()).isEqualTo(4);
+
assertThat(snapshotManager.earlierOrEqualWatermark(350).id()).isEqualTo(9);
+ }
+
+ @Test
+ public void testEarlierOrEqualWatermarkExactMatchWithNullWatermarks()
throws IOException {
+ FileIO localFileIO = LocalFileIO.create();
+ SnapshotManager snapshotManager =
+ newSnapshotManager(localFileIO, new Path(tempDir.toString()));
+ // create 5 snapshots, snapshots 2 and 3 do not carry a watermark
+ Map<Long, Long> watermarks = watermarkMap(0, 100, 1, 150, 4, 300);
+ for (long i = 0; i < 5; i++) {
+ Snapshot snapshot = createSnapshotWithMillis(i, 1000 + i * 1000,
watermarks.get(i));
+ localFileIO.tryToWriteAtomic(snapshotManager.snapshotPath(i),
snapshot.toJson());
+ }
+
+ // the exact match is snapshot 1, not one of the null-watermark
snapshots
+
assertThat(snapshotManager.earlierOrEqualWatermark(150).id()).isEqualTo(1);
+ // null-watermark snapshots fall back to the nearest earlier watermark
+
assertThat(snapshotManager.earlierOrEqualWatermark(200).id()).isEqualTo(1);
+ }
+
+ @Test
+ public void testEarlierOrEqualWatermarkWithAllNullWatermarks() throws
IOException {
+ FileIO localFileIO = LocalFileIO.create();
+ SnapshotManager snapshotManager =
+ newSnapshotManager(localFileIO, new Path(tempDir.toString()));
+ // create 5 snapshots without watermark
+ for (long i = 0; i < 5; i++) {
+ Snapshot snapshot = createSnapshotWithMillis(i, 1000 + i * 1000);
+ localFileIO.tryToWriteAtomic(snapshotManager.snapshotPath(i),
snapshot.toJson());
+ }
+
+ assertThat(snapshotManager.earlierOrEqualWatermark(100)).isNull();
+ }
+
+ @Test
+ public void testWatermarkSearchWithMinValueSentinelAndNullWatermarks()
throws IOException {
+ FileIO localFileIO = LocalFileIO.create();
+ SnapshotManager snapshotManager =
+ newSnapshotManager(localFileIO, new Path(tempDir.toString()));
+ // a mixed-engine history: snapshot 0 carries the Long.MIN_VALUE
sentinel written by
+ // engines without watermark semantics, snapshot 1 carries no
watermark at all
+ Map<Long, Long> watermarks = watermarkMap(0, Long.MIN_VALUE);
+ for (long i = 0; i < 2; i++) {
+ Snapshot snapshot = createSnapshotWithMillis(i, 1000 + i * 1000,
watermarks.get(i));
+ localFileIO.tryToWriteAtomic(snapshotManager.snapshotPath(i),
snapshot.toJson());
+ }
+
+ // the Long.MIN_VALUE sentinel means "no watermark" and must never
match a query;
+ // in particular it must not be returned for a rollback to watermark 0
+ assertThat(snapshotManager.earlierOrEqualWatermark(0)).isNull();
+ assertThat(snapshotManager.laterOrEqualWatermark(0)).isNull();
+ }
+
+ @Test
+ public void
testWatermarkSearchWithMinValueSentinelMixedWithRealWatermarks()
+ throws IOException {
+ FileIO localFileIO = LocalFileIO.create();
+ SnapshotManager snapshotManager =
+ newSnapshotManager(localFileIO, new Path(tempDir.toString()));
+ // snapshots 0 and 3 carry the Long.MIN_VALUE sentinel, snapshots 1
and 2 carry real
+ // watermarks
+ Map<Long, Long> watermarks =
+ watermarkMap(0, Long.MIN_VALUE, 1, 100, 2, 200, 3,
Long.MIN_VALUE);
+ for (long i = 0; i < 4; i++) {
+ Snapshot snapshot = createSnapshotWithMillis(i, 1000 + i * 1000,
watermarks.get(i));
+ localFileIO.tryToWriteAtomic(snapshotManager.snapshotPath(i),
snapshot.toJson());
+ }
+
+ // the sentinel on the latest snapshot must not short-circuit the
search
+
assertThat(snapshotManager.earlierOrEqualWatermark(100).id()).isEqualTo(1);
+
assertThat(snapshotManager.earlierOrEqualWatermark(250).id()).isEqualTo(2);
+
assertThat(snapshotManager.laterOrEqualWatermark(150).id()).isEqualTo(2);
+ // the leading sentinel must not shadow the real watermarks after it
+
assertThat(snapshotManager.laterOrEqualWatermark(50).id()).isEqualTo(1);
+ assertThat(snapshotManager.earlierOrEqualWatermark(50)).isNull();
+ }
+
/**
* Test {@link SnapshotManager} to mock situations when there is a race
condition, that the
* earliest snapshot is deleted by another thread in the middle of the
current thread's
