weiqingy opened a new pull request, #28952: URL: https://github.com/apache/flink/pull/28952
Part of the FLIP-497 implementation stack under umbrella [FLINK-36953](https://issues.apache.org/jira/browse/FLINK-36953). Landing order: | Step | Sub-task | Scope | | --- | --- | --- | | PR-1a | [FLINK-40167](https://issues.apache.org/jira/browse/FLINK-40167) | EARLY_FIRE hint surface + option validation (#28353, merged) | | PR-1b | [FLINK-40168](https://issues.apache.org/jira/browse/FLINK-40168) | Thread the hint into the interval join (#28796, merged) | | PR-2 | [FLINK-40169](https://issues.apache.org/jira/browse/FLINK-40169) | `target` option (#28827, merged) | | PR-3 | [FLINK-40170](https://issues.apache.org/jira/browse/FLINK-40170) | Update-producing changelog mode + insert-only guard (#28877, merged) | | **PR-4 (this PR)** | [FLINK-40171](https://issues.apache.org/jira/browse/FLINK-40171) | Runtime early-fire emit + retraction | | PR-5 | [FLINK-40172](https://issues.apache.org/jira/browse/FLINK-40172) | Processing-time early fire on an event-time join | | PR-6 | [FLINK-40173](https://issues.apache.org/jira/browse/FLINK-40173) | State restore coverage | | PR-7 | [FLINK-40174](https://issues.apache.org/jira/browse/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? - [X] Yes (please specify the tool below) Generated-by: Claude Code (Anthropic) -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
