Support watermark-based batch time travel (scan.watermark)

[u70b3]

Search before asking

• I searched in the issues and found nothing similar.

Motivation

Java supports batch time travel by watermark via scan.watermark and VERSION AS OF 'watermark-<value>' (StaticFromWatermarkStartingScanner + SnapshotManager.laterOrEqualWatermark): it resolves the earliest snapshot whose watermark is greater than or equal to the requested value and scans it in full (ScanMode.ALL).

Rust already parses the watermark field on Snapshot (spec layer is complete), but scan.watermark is on the CoreOptions::validate_scan_options blocklist, so any table options carrying it are rejected with Error::Unsupported before any IO. This is also a Java-interop trap: Flink/Spark-written tables whose options include scan.watermark cannot be read from Rust at all.

Solution

Implement watermark-based batch time travel, mirroring Java semantics:

1. SnapshotManager::later_or_equal_watermark(watermark): return the earliest snapshot with watermark >= requested, or None. Snapshots without a watermark are skipped — both None and Some(i64::MIN), since Java writers (Flink) use Long.MIN_VALUE as the no-watermark sentinel. Binary search over the actual snapshot id list, tolerating id gaps from deleted snapshots (same pattern as the existing later_or_equal_time_millis).
2. CoreOptions: parse scan.watermark as a first-class TimeTravelSelector — mutually exclusive with scan.timestamp-millis / scan.version / scan.snapshot-id / scan.tag-name, strict i64 parsing. Remove it from the unsupported blocklist, and accept it under scan.mode=from-snapshot (Java's CoreOptions.startupMode() maps scan.watermark to FROM_SNAPSHOT).
3. travel_to_snapshot: resolve both direct scan.watermark and Java-compatible scan.version=watermark-<value> (tag first, then watermark prefix, then snapshot id); when no snapshot matches, fail at scan planning with Java's message (There is currently no snapshot later than or equal to watermark[...]), preserving Java tryTravelToSnapshot's silent-fallback behavior at copy_with_time_travel time.
4. Table::copy_with_options: changing scan.watermark invalidates the cached resolved snapshot, like the other selectors.
5. Docs: docs/src/sql.md Time Travel section documents both Java-compatible VERSION AS OF 'watermark-<value>' and session-scoped SET 'paimon.scan.watermark'.

One deliberate difference from the current Java implementation is documented in code: Java's binary search can retain the raw midpoint snapshot after walking backward over missing watermark metadata. The Rust implementation only returns a snapshot whose own effective watermark satisfies watermark >= requested, preserving the selector contract when watermark metadata is sparse.

Anything else?

No response

Willingness to contribute

• I'm willing to submit a PR!