[FLINK-40171][table-runtime] Emit and retract early-fire results in the interval join operator by weiqingy · Pull Request #28952 · apache/flink

weiqingy · 2026-08-10T20:00:56Z

Part of the FLIP-497 implementation stack under umbrella FLINK-36953. Landing order:

Step	Sub-task	Scope
PR-1a	FLINK-40167	EARLY_FIRE hint surface + option validation (#28353, merged)
PR-1b	FLINK-40168	Thread the hint into the interval join (#28796, merged)
PR-2	FLINK-40169	`target` option (#28827, merged)
PR-3	FLINK-40170	Update-producing changelog mode + insert-only guard (#28877, merged)
PR-4 (this PR)	FLINK-40171	Runtime early-fire emit + retraction
PR-5	FLINK-40172	Processing-time early fire on an event-time join
PR-6	FLINK-40173	State restore coverage
PR-7	FLINK-40174	User-facing documentation

What is the purpose of the change

Implements the runtime behavior for the EARLY_FIRE hint 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: an event-time join with an event-time delay, and a processing-time join with a processing-time delay.

Brief change log

New operator constructor parameter earlyFireDelay, and a bookkeeping MapState tracking whether a row has already early-fired.
Schedule an early-fire timer for unmatched outer rows. On the timer, emit the speculative padded row and set the bit.
On a later match, retract the padded row and emit the corrected join result.
StreamExecIntervalJoin unboxes the delay and passes it to the operator.

Verifying this change

This change added tests and can be verified as follows:

Harness tests in RowTimeIntervalJoinTest and ProcTimeIntervalJoinTest pin the full +I then -U then +U sequence for left, right and full outer joins. The assertor compares positionally, so the ordering is enforced rather than incidental.
testRowTimeEarlyFireRowKindIsolation covers the negative direction: a pad emitted after a retraction must be a plain +I, not a leaked UPDATE_BEFORE.
testRowTimeLeftOuterEarlyFireMultiMatch covers that repeated matches of one early-fired row produce exactly one retraction.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): no
The public API, i.e., is any changed class annotated with @Public(Evolving): no
The serializers: no
The runtime per-record code paths (performance sensitive): yes, the interval join record path. The behavior is gated on the hint and off by default.
Anything that affects deployment or recovery: yes. The operator adds a bookkeeping MapState. A savepoint taken before this change restores it empty, which is safe: a row is then treated as not yet early-fired, so the only effect is a possible duplicate speculative row, never a swallowed retraction.
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? no (runtime for the FLIP-497 hint)
If yes, how is the feature documented? not applicable

Was generative AI tooling used to co-author this PR?

Yes (please specify the tool below)

Generated-by: Claude Code (Anthropic)

…he interval 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.

flinkbot · 2026-08-10T20:11:09Z

CI report:

ad2fdc1 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

weiqingy · 2026-08-10T20:27:02Z

Hi @RocMarshal, this is the runtime slice of the FLIP-497 stack, now that #28877 is merged. It makes the interval join emit the speculative padded row when the early-fire delay elapses, then retract and correct it if a real match arrives later.

Two things worth your eye:

It adds a bookkeeping MapState recording whether a row has already early-fired. A savepoint taken before this change restores it empty, which just means a row is treated as not yet early-fired, so the worst case is a duplicate speculative row, never a swallowed retraction.
Early-fire timers are not deleted when a row matches or gets cleaned up. They fire once, find nothing, and go away, so it is self-clearing rather than a leak. With a large delay they do outlive the state they refer to. Happy to delete them explicitly if you would rather not carry that.

PTAL when you have time. Thanks!

Copilot

Pull request overview

Implements runtime support for the EARLY_FIRE hint in stream interval joins by emitting speculative null-padded outer rows after a configured delay and retracting/replacing them when a later match arrives, with coverage for both row-time and processing-time joins.

Changes:

Added earlyFireDelay plumbing into interval join operators and enabled early-fire only for eligible outer joins.
Introduced fired-bit bookkeeping state and timer-driven early-fire emission + later-match retraction/correction logic.
Added harness tests asserting the exact +I -> -U -> +U changelog sequences across join types and time domains.

Reviewed changes

Copilot reviewed 7 out of 7 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/TimeIntervalJoin.java	Core early-fire implementation: fired bookkeeping state, timer-based speculative pads, and match-time retraction/correction gating.
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/EmitAwareCollector.java	Collector enhancements to turn a single match emit into `-U/+U` when retracting a previously emitted speculative pad.
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/RowTimeIntervalJoin.java	Threads `earlyFireDelay` into the row-time interval join operator constructor.
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/ProcTimeIntervalJoin.java	Threads `earlyFireDelay` into the proc-time interval join operator constructor.
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecIntervalJoin.java	Unboxes/passes the planned early-fire delay to the runtime operator (defaulting to disabled when absent).
flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/interval/RowTimeIntervalJoinTest.java	Adds row-time harness tests asserting speculative pad emission and `-U/+U` correction sequences and edge cases.
flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/interval/ProcTimeIntervalJoinTest.java	Adds proc-time harness tests for early-fire + correction and delay edge cases.

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

weiqingy · 2026-08-11T04:04:57Z

 /**
 * Collector to wrap a [[org.apache.flink.table.dataformat.RowData]] and to track whether a row has
 * been emitted by the inner collector.


Fixed in ad2fdc1. Now {@link RowData}. The old link also pointed at org.apache.flink.table.dataformat.RowData, which does not exist.

weiqingy · 2026-08-11T04:05:00Z

                    List<Tuple2<RowData, Boolean>> rightRows = rightEntry.getValue();
+                    List<Boolean> rightFired =
+                            earlyFireEnabled
+                                    ? firedBits(rightFiredState, rightTime, rightRows)
+                                    : null;


Gated in 7912a13. It is worse than an extra read: on a LEFT join nothing ever writes rightFiredState, so the list was always all-false.

Used isRightOuter() rather than an equality test, since that returns true for FULL as well. Adding the gate showed the suite could not tell the two apart, so I extended testRowTimeFullOuterEarlyFireOneMatches to drive the right-side retraction too. With the gate written as == FlinkJoinType.RIGHT that test now fails; before the change the whole suite stayed green.

weiqingy · 2026-08-11T04:05:01Z

                    List<Tuple2<RowData, Boolean>> leftRows = leftEntry.getValue();
+                    List<Boolean> leftFired =
+                            earlyFireEnabled ? firedBits(leftFiredState, leftTime, leftRows) : null;


Same fix in 7912a13, gated on isLeftOuter().

weiqingy · 2026-08-11T04:04:58Z

+ * join function knowing about changelogs. When not armed, collected rows are forwarded with their
+ * existing {@link RowKind}.


You are right, the doc was wrong and the code is deliberate. Reworded in ad2fdc1 to say the row is stamped INSERT, and why: the join function reuses one row instance whose kind may have been left at UPDATE_AFTER by an earlier correction. testRowTimeEarlyFireRowKindIsolation covers it.

… side The early-fire fired bits are only consulted when retracting a speculative pad, which can only happen on an outer side. Gate each per-side read on that side being outer, so a one-sided outer join stops reading state it can never use on the matching path. Extend the full-outer test to also drive the right-side retraction, which the suite did not exercise before.

…adoc The doc used a Scala-style link to a class that does not exist, and stated that an unarmed collector forwards the existing row kind. It stamps INSERT instead, so a reused row instance cannot leak an UPDATE_AFTER left by an earlier correction.

weiqingy · 2026-08-11T04:18:00Z

Copilot's four comments are resolved in 7912a13 and ad2fdc1: the per-side fired-state read is now gated on that side actually being outer, and the EmitAwareCollector class javadoc no longer carries a dead Scala-style link or describes a row kind the code does not use.

Gating that read turned up a gap: the suite could not tell the correct gate from an equality test that would break FULL joins, so testRowTimeFullOuterEarlyFireOneMatches now drives the right-side retraction as well.

cc @RocMarshal, this one is ready for review when you have time. Thanks!

weiqingy mentioned this pull request Aug 10, 2026

[FLINK-40172][table-runtime] Support processing-time early fire on a row-time interval join #28953

Draft

1 task

RocMarshal requested a lite review from Copilot August 11, 2026 01:36

Copilot started reviewing on behalf of RocMarshal August 11, 2026 01:37 View session

Copilot AI reviewed Aug 11, 2026

View reviewed changes

weiqingy added 2 commits August 10, 2026 21:01

github-actions Bot added the community-reviewed PR has been reviewed by the community. label Aug 11, 2026

Navigation Menu

Uh oh!

[FLINK-40171][table-runtime] Emit and retract early-fire results in the interval join operator - #28952

[FLINK-40171][table-runtime] Emit and retract early-fire results in the interval join operator#28952
weiqingy wants to merge 3 commits into
apache:masterfrom
weiqingy:FLINK-36953-pr4-runtime

weiqingy commented Aug 10, 2026

Uh oh!

flinkbot commented Aug 10, 2026 •

edited

Loading

Uh oh!

weiqingy commented Aug 10, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

weiqingy Aug 11, 2026

Uh oh!

weiqingy Aug 11, 2026

Uh oh!

weiqingy Aug 11, 2026

Uh oh!

weiqingy Aug 11, 2026

Uh oh!

weiqingy commented Aug 11, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

		* join function knowing about changelogs. When not armed, collected rows are forwarded with their
		* existing {@link RowKind}.

Uh oh!

Conversation

weiqingy commented Aug 10, 2026

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Was generative AI tooling used to co-author this PR?

Uh oh!

flinkbot commented Aug 10, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

weiqingy commented Aug 10, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

weiqingy Aug 11, 2026

Choose a reason for hiding this comment

Uh oh!

weiqingy Aug 11, 2026

Choose a reason for hiding this comment

Uh oh!

weiqingy Aug 11, 2026

Choose a reason for hiding this comment

Uh oh!

weiqingy Aug 11, 2026

Choose a reason for hiding this comment

Uh oh!

weiqingy commented Aug 11, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

flinkbot commented Aug 10, 2026 •

edited

Loading