[FLINK-40171][table-runtime] Emit and retract early-fire results in the interval join operator - #4
Draft
weiqingy wants to merge 7 commits into
Draft
Conversation
…dation Register the EARLY_FIRE join hint and its typed options (EarlyFireJoinHintOptions: delay, time_mode) with a key-value option checker, and wire the query-hint propagation touchpoints. Exclude EARLY_FIRE from the generic join-hint test coverage and config-docs generation, since it is a key-value hint validated on its own. The hint is recognized and validated here but not yet consumed by any rule; threading it into the interval join follows in a separate change.
weiqingy
force-pushed
the
FLINK-36953-pr3-changelog
branch
from
July 18, 2026 22:12
f6163da to
16ae7d2
Compare
weiqingy
force-pushed
the
FLINK-36953-pr4-runtime
branch
from
July 18, 2026 22:12
a56a787 to
fcb8e67
Compare
weiqingy
changed the title
…time-mode Flink option keys use hyphens (e.g. output-mode, fixed-delay); rename the EARLY_FIRE hint's time_mode option key to time-mode to match, and update the affected tests.
weiqingy
force-pushed
the
FLINK-36953-pr4-runtime
branch
from
July 21, 2026 02:17
fcb8e67 to
72e92e5
Compare
weiqingy
force-pushed
the
FLINK-36953-pr3-changelog
branch
from
July 21, 2026 02:17
16ae7d2 to
17d607b
Compare
…idation tests The runtime uses delay.toMillis(), so a sub-millisecond delay truncates to zero. Make the contract explicit: require at least 1 millisecond in the DELAY description and the checker error message. Add coverage for a sub-millisecond delay, list-style options (only key-value is supported), and case-insensitive hint-name capitalization that preserves the key-value options.
Read the EARLY_FIRE hint in StreamPhysicalIntervalJoinRule and thread its delay and time mode through StreamPhysicalIntervalJoin into StreamExecIntervalJoin as NON_NULL JSON fields. Resolve the effective time mode from the join's time domain, reject row-time triggering on a processing-time join, and reject (for now) processing-time triggering on an event-time join. The operator receives the parameters but ignores them; runtime behavior follows in a later change.
Add an optional `target` option to the EARLY_FIRE join hint. Only `interval_join` is accepted today; any other value fails planning. An omitted target is equivalent to `interval_join`, so existing hints keep their meaning. `target` scopes the hint to a single operator kind. The interval-join rule consumes the hint only when it targets the interval join and leaves a hint aimed at another operator kind untouched, so an untargeted hint never silently expands its scope.
…fire interval join With the EARLY_FIRE hint, an outer interval join speculatively emits a padded unmatched row after the delay and corrects it when a match later arrives, so it no longer produces insert-only changes. Teach FlinkChangelogModeInferenceProgram to reflect this. Split StreamPhysicalIntervalJoin into its own ModifyKindSet arm: its children still consume insert-only, but the node provides INSERT and, when the hint makes it update-producing, UPDATE. A new produceEarlyFireUpdates accessor gates that on the hint being set, the join being outer, and a non-negative window span, so the hint stays inert for inner joins and negative-window joins (which only ever emit inserts). The interval join keeps its place in the UpdateKind and DeleteKind arms. When such a join feeds an insert-only downstream, planning fails with a tailored error that names the hint, rather than the generic "doesn't support consuming update changes" message. Runtime behavior is unchanged; the operator still ignores the hint.
…val join operator Wire the EARLY_FIRE delay into the interval join operator so an outer join speculatively emits its padded unmatched row after the delay and corrects it when a real match arrives. Covers the natural timer pairings: a row-time join fires on event time, a processing-time join fires on processing time. Processing-time triggering on a row-time join stays rejected at planning. When an unmatched outer row is cached, the operator registers an early-fire timer at rowTime + delay. On that timer it emits the padded row as an INSERT and records that it fired. When the row later matches, it retracts the padded row as UPDATE_BEFORE and emits the matched row as UPDATE_AFTER, matching the update-producing changelog mode inferred for the node. The retraction is tied to the one-time matched-and-emitted flip, so a row that matches several times emits a single correction followed by ordinary inserts. The already-fired marker is a new per-side MapState<Long, List<Boolean>> kept positionally aligned with the existing row cache, rather than widening the cache tuple, so the cache serializer is unchanged and old savepoints restore the new state empty. The marker is the single gate that keeps a row padded exactly once when the delay is at or beyond the window span. All early-fire work is gated on the hint being set, an outer join, and a non-negative window, so a plain interval join is unchanged and allocates nothing new. EmitAwareCollector carries the changelog stamping so IntervalJoinFunction stays changelog-agnostic, and every padded or matched emit stamps its RowKind explicitly to avoid leaking a kind onto a reused row.
weiqingy
force-pushed
the
FLINK-36953-pr4-runtime
branch
from
July 21, 2026 04:43
72e92e5 to
ebe6e95
Compare
weiqingy
force-pushed
the
FLINK-36953-pr3-changelog
branch
from
July 21, 2026 04:43
17d607b to
9dbe90d
Compare
weiqingy
force-pushed
the
FLINK-36953-pr3-changelog
branch
from
August 2, 2026 22:42
9dbe90d to
8f7448f
Compare
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
1 participant
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 stack under umbrella FLINK-36953. Stacked on PR-3. Landing order: PR-1a → PR-1b → PR-2 → PR-3 → PR-4 (this) → PR-5 → PR-6 → PR-7.
What is the purpose of the change
Implements the runtime behavior for the
EARLY_FIREhint on an interval join. An unmatched outer row is emitted speculatively with a null-padded counterpart after the configured delay; if a real match later arrives within the window, the speculative row is retracted and corrected. This covers the natural time-domain pairings (event-time join with event-time delay, processing-time join with processing-time delay).Brief change log
earlyFireDelayand bookkeepingMapStatetracking whether a row has already early-fired.StreamExecIntervalJoinunboxes and passes the delay to the operator.Verifying this change
This change added tests and can be verified as follows:
RowTimeIntervalJoinTest/ProcTimeIntervalJoinTestasserting the speculative+Ithen-D/-U++Uon match, for left/right/full outer joins.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)