feat(table): support scan.watermark batch time travel - #677
Open
u70b3 wants to merge 5 commits into
Open
Conversation
Add scan.watermark as a batch time-travel selector, mirroring Java's StaticFromWatermarkStartingScanner: resolve the earliest snapshot whose watermark is greater than or equal to the requested value and scan it in full. Snapshots without a watermark (None or the Flink Long.MIN_VALUE sentinel) are skipped. - SnapshotManager::later_or_equal_watermark: binary search over the actual snapshot id list (gap-tolerant), returning the earliest match. - CoreOptions: parse scan.watermark as a first-class TimeTravelSelector (mutual exclusion with other selectors, strict i64 parsing), drop it from the unsupported scan-option blocklist, and accept it for scan.mode=from-snapshot like Java's startupMode mapping. - travel_to_snapshot: resolve the selector, erroring at scan planning with Java's message when no snapshot matches. - Table::copy_with_options: changing scan.watermark invalidates the cached travel snapshot like the other selectors. - docs: sql.md Time Travel section gains a By Watermark subsection.
The C and Python read-builder gates enumerate the time-travel selectors to reject conflicts up front and to error when a set selector resolves to no snapshot. Without scan.watermark in those lists, an unresolvable watermark fell through the strict gate and silently read latest, and a watermark plus another selector went undetected (the core swallows the conflict via its Java-parity silent fallback).
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Purpose
Linked issue: close #676
Support watermark-based batch time travel, mirroring Java's
StaticFromWatermarkStartingScannerandTimeTravelUtil.adaptScanVersion: directscan.watermarkandVERSION AS OF 'watermark-<value>'resolve the earliest snapshot whose watermark is greater than or equal to the requested value and scan it in full. Todayscan.watermarkis on thevalidate_scan_optionsblocklist, 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 aslater_or_equal_time_millis). Snapshots without a watermark are skipped — bothNoneandSome(i64::MIN), since Flink writers useLong.MIN_VALUEas the no-watermark sentinel.CoreOptions:scan.watermarkbecomes a first-classTimeTravelSelector(mutual exclusion with the other selectors, strict i64 parsing); removed from the unsupported scan-option blocklist; accepted underscan.mode=from-snapshot(Java'sstartupMode()maps it toFROM_SNAPSHOT).scan.version: mirror Java's tag-first resolution order — existing tag →watermark-<value>→ snapshot id — enablingVERSION 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, whilecopy_with_time_travelkeeps Java's silent-fallback behavior.Table::copy_with_options: changingscan.watermarkinvalidates the cached resolved snapshot.unsupported_scan_option_is_rejectedtest now usesincremental-betweenas its example (scan.watermarkis supported now).docs/src/sql.md: new "By Watermark" subsection documenting bothVERSION AS OF 'watermark-<value>'and the dynamic option.Assumptions / deviations to be aware of (per the AI-assisted PR policy):
watermark >= requested, preserving the selector contract when watermark metadata is sparse. Documented in the method's doc comment.Snapshots with watermarks directly throughSnapshotManager; no write-path changes are included.Tests
cargo fmt --all -- --checkcargo clippy --locked --all-targets -p paimon -p paimon-datafusion -- -D warningscargo 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, Javawatermark-<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-endVERSION AS OF 'watermark-<value>'test matching Java's 1/9/10/no-match casescargo test --locked -p paimon-c unsupported_scan_option— 1 passedAPI and Format
No storage format changes. The public Rust API gains
SnapshotManager::later_or_equal_watermark, andSCAN_WATERMARK_OPTIONbecomes public alongside the other selector constants.Documentation
docs/src/sql.mdTime Travel section gains a "By Watermark" subsection documenting bothVERSION AS OF 'watermark-<value>'and the session-scopedSET 'paimon.scan.watermark'dynamic option.🤖 Generated with Claude Code