[FLINK-40173][table-planner] Add restore coverage for early-fire interval join - #29045
Draft
weiqingy wants to merge 2 commits into
Draft
[FLINK-40173][table-planner] Add restore coverage for early-fire interval join#29045weiqingy wants to merge 2 commits into
weiqingy wants to merge 2 commits into
Conversation
…row-time interval join
Add the cross-domain timer combination the previous commit left out: an
event-time interval join with EARLY_FIRE('time_mode'='proctime') now fires its
speculative pads on the wall clock while keeping its event-time cleanup. The
temporary "not yet supported" rejection in the planner rule is removed; the
row-time-on-processing-time rejection is retained.
onTimer distinguishes the two timer kinds by OnTimerContext.timeDomain(): in
the cross-domain case early-fire timers are processing-time and cleanup timers
are event-time, so a processing-time firing runs early fire and returns while
an event-time firing runs cleanup only. The discrimination is gated on a new
cross-domain flag, so the natural pairings keep the previous timestamp - delay
recovery where early fire and cleanup share a domain.
A processing-time firing timestamp cannot be mapped back to an event-time cache
bucket arithmetically, so a per-side MapState<Long, List<Long>> keyed by firing
processing-time records the event-time bucket keys due to fire then. It is
allocated only in the cross-domain case and reuses the existing per-bucket emit
and positional fired bit, so the retract-and-correct path is shared. Every
scheduled firing time fires and removes its own entry, and a bucket already
cleaned by event-time expiry makes the firing a no-op, so nothing accumulates.
The schedule is value-typed and order-preserving and processing-time timers are
checkpointed, so a timer pending at snapshot fires after restore against the
restored schedule and fired bits and emits at most the not-yet-emitted pad.
Harness tests cover the wall-clock trigger without watermark advance, a snapshot
before the timer fires, and a snapshot after the pad is emitted.
…rval join Add an INTERVAL_JOIN_EARLY_FIRE restore test program with its plan and savepoint fixtures, and register it in IntervalJoinRestoreTest, exercising end-to-end plan and savepoint restore for the early-fire interval join. Cover both early-fire time modes. The row-time program exercises the fired bookkeeping restored from a savepoint; a second program with a processing-time delay covers the cross-domain schedule state, which only exists in that mode. The processing-time program restores with no further input, so its only output can come from the restored schedule.
Collaborator
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
2 participants
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.
Part of the FLIP-497 implementation stack under umbrella FLINK-36953. Landing order:
targetoption (#28827, merged)Opened as a draft because it is stacked on #28953, which is in review. Until that merges, the commit list and diff here also carry PR-5's commit. Once #28953 merges I will rebase onto master, leaving only this PR's change, and take it out of draft.
What is the purpose of the change
Adds end-to-end restore coverage for the early-fire interval join: a compiled plan plus a savepoint, restored and run. The two early-fire time modes keep different operator state, so there is one program for each.
Brief change log
INTERVAL_JOIN_EARLY_FIRE, a row-time program covering the fired bookkeeping restored from a savepoint. A row padded before the savepoint is matched after it, so the restored bit is what turns the match into a-U/+Ucorrection rather than a plain+I.INTERVAL_JOIN_PROC_TIME_EARLY_FIRE, a processing-time program covering the cross-domain schedule state, which only exists in that mode.IntervalJoinProcTimeEarlyFireRestoreTestwith anINFINITEafter-restore source and no after-restore input, followingGroupWindowAggregateProcTimeRestoreTest. With no further input there is noMAX_WATERMARKand so no cleanup path, which keeps the result deterministic: the single emitted row can only come from the restored schedule. A finite program would race the restored timer against the first record.RestoreTestBase.enablePerRecordWatermarksmatchesWATERMARK FORwhile the interval-join schema spells itWATERMARK for, so the automatic path has never applied to these programs.Verifying this change
This change added tests and can be verified as follows:
Does this pull request potentially affect one of the following parts:
@Public(Evolving): noDocumentation
Was generative AI tooling used to co-author this PR?
Generated-by: Claude Code (Anthropic)