[Bug] SnapshotManager watermark binary searches: infinite loop, wrong result and NPE with null watermarks

[u70b3]

Search before asking

• I searched in the issues and found nothing similar.

Paimon version

master @ 6602ec963 (verified line-by-line against GitHub master)

Compute Engine

Java API (paimon-core, SnapshotManager). Reachable from Flink batch scan (scan.watermark), Flink create_tag_from_watermark / rollback_to_watermark procedures, and Spark rollback_to_watermark procedure.

Context: how this was found

These defects were first noticed while implementing scan.watermark batch time travel for paimon-rust (https://github.com/apache/paimon-rust PR #677). Cross-validating the Rust implementation's behavior against the Java reference implementation surfaced that the Java SDK itself has these problems. The Rust implementation avoids them structurally (bounded fallback, id-list based binary search, None/i64::MIN unified as missing).

Minimal reproduce step

All defects are in the two watermark binary searches in SnapshotManager:

• laterOrEqualWatermark (SnapshotManager.java:427-490)
• earlierOrEqualWatermark (SnapshotManager.java:362-425) — a verbatim copy of the same code, so it shares the same defects, plus one of its own

Null watermark (snapshot JSON without the watermark field) is a documented live state (Snapshot.java:158-164). In pure-Java single-writer lineages, commit-time carry-over (FileStoreCommitImpl.java:1049-1059) keeps nulls as a prefix only, so the interleaved-null cases below need a mixed-engine table (e.g. Flink streaming writes with watermarks interleaved with paimon-rust / pypaimon appends, which never write the watermark field). The all-null case needs no interleaving at all — any table that never carried a watermark (pure batch/Spark/Rust-written) triggers it.

Defect 1 — infinite loop (both methods). Snapshots ids 1–10, watermarks {1:100, 5:200, 10:300}, rest null; request 150:

• round 1: window [1,10], mid=5 (w=200 > 150) → latest=4, correct answer (snapshot 5) already recorded
• round 2: window [1,4], mid=2 (null) → fallback walks to id 1 (w=100 < 150) → earliest=2
• round 3: window [2,4], mid=3 (null) → fallback at :467-473 decrements mid past the window's left edge (while (mid >= earliest) { mid--; ... } reads earliest-1) and finds the stale w=100 at id 1 → earliest = mid + 1 = 2, window unchanged
• round 4+: identical to round 3 — the scan thread hangs forever, re-reading snapshot files on every iteration

Root cause: the fallback traversal mutates mid and reads outside the search window; the window update earliest = mid + 1 then recomputes its previous value.

Defect 2 — exact-match returns a snapshot whose own watermark is null (both methods). Watermarks {1:100, 2:150, 3:null, 4:null, 5:300}, request 150: mid=3 (null) → fallback finds w=150 at id 2 → finalSnapshot = snapshot (:484) assigns snapshot 3 (own watermark null) instead of snapshot 2. Same on the > branch (:480). CreateTagFromWatermarkProcedure already defends against this quirk at its call site (snapshot.watermark() == null check), which suggests it has been hit in practice; StaticFromWatermarkStartingScanner and the rollback procedures are undefended.

Defect 3 — NPE in the guard (both methods). :431 / :366: snapshot(latest).watermark() == Long.MIN_VALUE unboxes a null Long when the latest snapshot has no watermark → raw NullPointerException instead of a clean "no match" null. Trigger: any table whose snapshots all lack the watermark field, queried with scan.watermark (e.g. a pure batch-written table, or a table written by paimon-rust / pypaimon and read by the Java SDK).

Defect 4 — inverted early-return in earlierOrEqualWatermark only. :391-392 was copied verbatim from laterOrEqualWatermark:456-458:

if (earliestWatermark >= watermark) {
    return snapshot(earliest);
}

For "earlier or equal" semantics this is inverted — compare earlierOrEqualTimeMills:306, which returns null when the earliest value is already greater than the request. Watermarks {1:100, 2:200, 3:300}, request 50: the method returns snapshot 1 (w=100 > 50), violating its own contract; it should return null. The rollback_to_watermark procedures then roll the table back to a snapshot newer than the requested watermark (silent under-rollback) instead of failing with "count not find snapshot". This one is reachable with plain dense watermarks — no nulls needed.

What doesn't meet your expectations?

• laterOrEqualWatermark / earlierOrEqualWatermark must terminate on any input, return only snapshots whose own watermark satisfies the predicate, and never throw NPE on tables without watermarks.
• earlierOrEqualWatermark must return null when the requested watermark is below every snapshot's watermark.

Test gap: SnapshotManagerTest.testLaterOrEqualWatermark (:259-273) only covers the all-MIN_VALUE guard early-exit; testEarlierOrEqualWatermark (:111-124) uses dense watermarks with a request above the minimum, so neither the binary-search fallback nor the inverted early-return has any coverage — which is why these survived.

Proposed fix direction

Same pattern for both methods:

1. Null-safe the guard (snapshot(latest).watermark() may be null; keep the MIN_VALUE short-circuit).
2. Bound the fallback traversal inside the [earliest, mid] search window (walk a separate pos, never mutate mid), so the window provably shrinks every iteration.
3. Record the snapshot the fallback actually landed on; compute window updates from the original mid (earliest = mid + 1 / latest = pos - 1).
4. earlierOrEqualWatermark:391: change to earliestWatermark > watermark → return null, mirroring earlierOrEqualTimeMills:306.

Fix #4 is a behavior change for the below-minimum request case (wrong snapshot → null, i.e. the rollback procedure starts failing loudly instead of silently under-rolling-back); worth calling out in review.

Anything else?

PR #9037 contains black-box reproductions of all four defects in WatermarkTimeTravelTest (paimon-core). Snapshots are produced through the real commit path (TableCommitImpl with ManifestCommittable watermarks, the same entry the Flink committer uses; the interleaved-null layout is built by rewriting snapshot files, simulating a mixed-engine table) and reads go through the real batch scan link (scan.watermark → StaticFromWatermarkStartingScanner).

• Defect 1 (non-terminating search) → testScanWatermarkWithInterleavedNullWatermarks (guarded by @Timeout)
• Defect 2 (exact match returns null-watermark snapshot) → testScanWatermarkExactMatchWithInterleavedNullWatermarks
• Defect 3 (NPE on watermark-less table) → testScanWatermarkOnTableWithoutWatermarks
• Defect 4 (inverted early-return → silent under-rollback) → testRollbackToWatermarkBelowMinimum

Unit-level coverage of the search internals is in SnapshotManagerTest in the same PR.

Are you willing to submit a PR?

• I'm willing to submit a PR!

[u70b3]

take — I'd like to fix this.

For context: I first ran into this while implementing scan.watermark batch time travel in paimon-rust (apache/paimon-rust#677). Cross-validating the Rust behavior against the Java reference implementation surfaced these defects in the Java SDK itself; the Rust side avoids them structurally (bounded in-window fallback, binary search over the actual snapshot id list, None/i64::MIN treated uniformly as missing).

I'll submit a PR fixing both laterOrEqualWatermark and earlierOrEqualWatermark (including the inverted early-return at :391) along the lines described in the issue, with tests covering the interleaved-null, all-null and below-minimum cases.