feat(table): support scan.watermark batch time travel

[u70b3]

Purpose

Linked issue: close #676

Support watermark-based batch time travel, mirroring Java's StaticFromWatermarkStartingScanner and TimeTravelUtil.adaptScanVersion: direct scan.watermark and VERSION AS OF 'watermark-<value>' resolve the earliest snapshot whose watermark is greater than or equal to the requested value and scan it in full. Today scan.watermark is on the validate_scan_options blocklist, so Java-written tables carrying it cannot be read from Rust at all.

Brief change log

• SnapshotManager::later_or_equal_watermark: binary search over the actual snapshot id list (gap-tolerant, same pattern as later_or_equal_time_millis). Snapshots without a watermark are skipped — both None and Some(i64::MIN), since Flink writers use Long.MIN_VALUE as the no-watermark sentinel.
• CoreOptions: scan.watermark becomes a first-class TimeTravelSelector (mutual exclusion with the other selectors, strict i64 parsing); removed from the unsupported scan-option blocklist; accepted under scan.mode=from-snapshot (Java's startupMode() maps it to FROM_SNAPSHOT).
• scan.version: mirror Java's tag-first resolution order — existing tag → watermark-<value> → snapshot id — enabling VERSION AS OF 'watermark-<value>' in DataFusion SQL.
• travel_to_snapshot: direct and version-prefixed watermark selectors share one resolver; no match fails at scan planning with Java's message, while copy_with_time_travel keeps Java's silent-fallback behavior.
• Table::copy_with_options: changing scan.watermark invalidates the cached resolved snapshot.
• C bindings: the unsupported_scan_option_is_rejected test now uses incremental-between as its example (scan.watermark is supported now).
• docs/src/sql.md: new "By Watermark" subsection documenting both VERSION AS OF 'watermark-<value>' and the dynamic option.

Assumptions / deviations to be aware of (per the AI-assisted PR policy):

• Deliberate difference from the current Java implementation: Java's binary search can retain the raw midpoint snapshot after walking backward over missing watermark metadata. This implementation only returns a snapshot whose own effective watermark satisfies watermark >= requested, preserving the selector contract when watermark metadata is sparse. Documented in the method's doc comment.
• Watermarks are assumed non-decreasing in snapshot order (guaranteed by Flink/Java writers); the binary search relies on this.
• Rust's own commit path never writes watermarks, so the new tests commit Snapshots with watermarks directly through SnapshotManager; no write-path changes are included.

Tests

• cargo fmt --all -- --check
• cargo clippy --locked --all-targets -p paimon -p paimon-datafusion -- -D warnings
• cargo test --locked -p paimon --lib — 2101 passed, 0 failed, 1 ignored (11 new watermark tests: sentinel/missing/duplicate watermarks, exact/between/out-of-range matches, snapshot-id gaps, selector mutual exclusion, Java watermark-<value> resolution with tag precedence, cache invalidation, and scan-time errors)
• cargo test --locked -p paimon-datafusion --test time_travel_schema_tests — 6 passed, including an end-to-end VERSION AS OF 'watermark-<value>' test matching Java's 1/9/10/no-match cases
• cargo test --locked -p paimon-c unsupported_scan_option — 1 passed

API and Format

No storage format changes. The public Rust API gains SnapshotManager::later_or_equal_watermark, and SCAN_WATERMARK_OPTION becomes public alongside the other selector constants.

Documentation

docs/src/sql.md Time Travel section gains a "By Watermark" subsection documenting both VERSION AS OF 'watermark-<value>' and the session-scoped SET 'paimon.scan.watermark' dynamic option.

🤖 Generated with Claude Code