You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "jackylee-ch (via GitHub)" <gi...@apache.org> on 2026/09/26 09:17:36 UTC
[PR] [spark] Add create_tag_from_watermark procedure [paimon]
jackylee-ch opened a new pull request, #10198:
URL: https://github.com/apache/paimon/pull/10198
## Purpose
Flink has a `create_tag_from_watermark` procedure but Spark has none, so a
Spark user cannot tag the first snapshot whose watermark reaches a target —
even though Spark already exposes `rollback_to_watermark` and
`create_tag_from_timestamp`. It is useful when a Flink writer commits
event-time watermarks and a Spark job needs a consistent tag at a watermark
boundary.
## Change
Add `CreateTagFromWatermarkProcedure`, registered as
`create_tag_from_watermark`. It mirrors the existing
`CreateTagFromTimestampProcedure` structure and the Flink procedure's logic:
resolve `SnapshotManager.laterOrEqualWatermark`, fall back to the earliest tag
whose watermark already covers the target, else raise
`SnapshotNotExistException`.
## Tests
`CreateTagFromWatermarkProcedureTest` commits three snapshots with watermarks
1000/2000/3000 (through the stream write builder, since a Spark batch write
carries no watermark) and asserts the resolved snapshot for a below/equal
watermark and a `SnapshotNotExistException` past the last one.
Written with Claude Code; verification is mine.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Add create_tag_from_watermark procedure [paimon]
Posted by "jackylee-ch (via GitHub)" <gi...@apache.org>.
jackylee-ch commented on PR #10198:
URL: https://github.com/apache/paimon/pull/10198#issuecomment-5967280288
Good catch — `laterOrEqualWatermark` stops on whichever equal-watermark snapshot the binary search lands on, so a run sharing the target watermark can resolve to a later member.
Rather than change the shared `SnapshotManager` (other callers rely on its current contract), I fix it at this consumption site: after `laterOrEqualWatermark`, the procedure now walks back to the earliest snapshot whose watermark still covers the target, stopping at any gap or missing-watermark snapshot. The tag fallback then runs on that earliest retained id, so the earlier-id-on-tie rule stays consistent.
Added a retained repeated-watermark regression alongside the tie/expired-tag one: watermarks `1000,2000,2000,2000,3000`, `watermark => 2000` now tags snapshot 2 (binary search landed on 3). Pushed in 425bc4793.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Add create_tag_from_watermark procedure [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #10198:
URL: https://github.com/apache/paimon/pull/10198#issuecomment-5950888416
[P2] Prefer the earlier tagged snapshot when watermarks tie (`CreateTagFromWatermarkProcedure.java:98–101`)
The new procedure promises the first qualifying snapshot, including tagged snapshots, but the strict `tagSnapshot.watermark() < snapshot.watermark()` comparison ignores an older tagged snapshot with the same watermark as the retained candidate. Expiration can therefore change the selected dataset even though the earlier snapshot is still available through a tag.
I reproduced this with real stream commits, snapshot expiration, Spark SQL CALL, and tag time-travel reads:
- Snapshot 1: watermark 1000, row `(1, 'row1')`; preserve it as tag `historical`.
- Snapshot 2: watermark 1000, with the additional row `(2, 'row2')`.
- Expire snapshot 1 while keeping snapshot 2. The historical tag still reads only row 1.
- `CALL ...create_tag_from_watermark(table => 'test.T', tag => 'boundary', watermark => 500)` creates a tag for snapshot 2. Reading `T VERSION AS OF 'boundary'` returns rows 1 and 2, instead of the first qualifying snapshot's row 1.
`SnapshotManager` correctly returns the first *retained* candidate, snapshot 2. This procedure's new merge with tagged snapshots should select snapshot 1, but equal watermarks fail the strict comparison. Repeated watermarks are normal: commits retain the maximum/inherit the previous watermark, so a stalled Flink watermark or a Spark append after a Flink commit can produce this state.
When candidate watermarks are equal, choose the smaller snapshot ID. Simply changing `<` to `<=` would instead let a newer tag replace an earlier retained snapshot, so cover that direction as well. The Flink counterpart has the same condition, but this new Spark API should meet its documented first-snapshot contract rather than propagate that defect.
Validation: the wrong tagged dataset reproduces on both Spark 3.5 and Spark 4.1. On Spark 3.5, the same request selects snapshot 1 before expiration and snapshot 2 afterward; the control with only a newer equal-watermark tag correctly keeps earlier retained snapshot 1. The PR's original actual SQL suite passes on both versions with normal Maven checks, and the 1-day tag-retention argument is verified in the reproduction.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Add create_tag_from_watermark procedure [paimon]
Posted by "jackylee-ch (via GitHub)" <gi...@apache.org>.
jackylee-ch commented on PR #10198:
URL: https://github.com/apache/paimon/pull/10198#issuecomment-5954528957
Fixed in 2b9ebee70. The tag fallback now prefers the smaller watermark and, on a tie, the smaller snapshot id, so an older tagged snapshot sharing the retained candidate's watermark wins (and the reverse — a newer equal-watermark tag no longer displaces an earlier retained snapshot). This meets the documented first-qualifying-snapshot contract.
Added a regression mirroring your repro: snapshot 1 (watermark 1000) is tagged then expired, snapshot 2 (watermark 1000) is retained, and `create_tag_from_watermark(500)` now selects snapshot 1 via the tag; `VERSION AS OF 'boundary'` reads only row 1.
Verified locally: compiles, spotless and `git diff --check` clean. The test itself is CI-verified — my local run can't instantiate the codegen `Projection` for the primary-key write path (`ComponentClassLoader` NPE), which also prevents this file's existing test from running locally though it's green in CI; the new test runs on the same lane.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Add create_tag_from_watermark procedure [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #10198:
URL: https://github.com/apache/paimon/pull/10198#issuecomment-5966220554
[P2] Select the first retained snapshot when qualifying watermarks repeat (`CreateTagFromWatermarkProcedure.java:89`)
The expired-tag tie regression is resolved. There is another common tie case in the new procedure's documented “first snapshot” contract: the retained candidate can already be too late before the tag fallback runs.
At `f97d656738`, I committed five real stream snapshots with watermarks `1000, 2000, 2000, 2000, 3000`, distinct rows `(1, 'row1')` through `(5, 'row5')`, and no pre-existing tags. The fixture uses the supported `write-only=true` option to keep exactly five append snapshots. This call creates a tag for snapshot **3**:
```sql
CALL paimon.sys.create_tag_from_watermark(
table => 'test.T', tag => 'boundary', watermark => 2000, time_retained => '1 d');
SELECT * FROM T VERSION AS OF 'boundary';
```
The persisted tag has snapshot ID 3 and the read returns rows 1, 2, and 3. The first qualifying snapshot is **2**, whose dataset contains only rows 1 and 2. On a fresh table with the same snapshot chain, target `1999` correctly chooses snapshot 2; target `2000` takes the binary-search equality shortcut and chooses snapshot 3. Repeated watermarks occur during ordinary stream checkpoints when the watermark stalls.
`SnapshotManager.laterOrEqualWatermark` stops on an equal midpoint instead of continuing left. That helper behavior predates this PR, but this new Spark API consumes its result and promises the first qualifying snapshot; the new tag comparison cannot repair the result when there is no older qualifying tag. Please make the retained search choose the earliest qualifying snapshot as well, and add a retained repeated-watermark regression alongside the expired-tag regression.
Validation: the two committed Spark 3 tests pass locally with normal Maven checks, and both are also confirmed passing in the current Spark 4 CI log. Seven additional actual Spark 3 SQL cases give six passes and this one failure. An isolated control changing only the helper's equality branch to continue searching left passes all seven cases, including expired-tag ties, the reverse newer-tag tie, exact time-travel datasets, one-day tag metadata, and missing-watermark cases.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Add create_tag_from_watermark procedure [paimon]
Posted by "jackylee-ch (via GitHub)" <gi...@apache.org>.
jackylee-ch closed pull request #10198: [spark] Add create_tag_from_watermark procedure
URL: https://github.com/apache/paimon/pull/10198
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org