You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "weiqingy (via GitHub)" <gi...@apache.org> on 2026/08/02 22:46:15 UTC
[PR] [FLINK-40170][table-planner] Infer update-producing changelog mode for early-fire interval join [flink]
weiqingy opened a new pull request, #28877:
URL: https://github.com/apache/flink/pull/28877
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, in review) |
| **PR-3 (this PR)** | [FLINK-40170](https://issues.apache.org/jira/browse/FLINK-40170) | Update-producing changelog mode + insert-only guard |
| PR-4 | [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 |
Opened as a draft because it is stacked on #28827, which is still in review. Until that merges, the commit list and diff here also carry PR-2's commit. Once #28827 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
With the `EARLY_FIRE` hint, an outer interval join emits a speculative null-padded row after the delay and corrects it when a match later arrives, so its result is no longer insert-only. This makes the planner infer that update-producing changelog mode, and reject an insert-only downstream with an error that names the hint instead of the generic "doesn't support consuming update changes" message. Inner joins and negative-window joins only ever emit inserts, so the hint stays inert for them and they remain append-only.
Runtime behavior is unchanged. The operator still ignores the hint; emission lands in PR-4.
## Brief change log
- `FlinkChangelogModeInferenceProgram` gets its own `ModifyKindSet` arm for `StreamPhysicalIntervalJoin`: children still consume insert-only, but the node provides `INSERT` and, when the hint makes it update-producing, `UPDATE`. It keeps its place in the `UpdateKind` and `DeleteKind` arms.
- `StreamPhysicalIntervalJoin.produceEarlyFireUpdates` gates that on three conditions: the hint is set, the join is outer, and the window span is non-negative.
- A tailored `TableException` when such a join feeds an insert-only downstream.
- The compiled-plan round-trip test's sink now declares `sink-insert-only=false`. Its early-fire outer join produces updates under the new inference, so the previously insert-only sink no longer accepts that pipeline. This is the one existing behavior this PR changes.
## Verifying this change
This change added tests and can be verified as follows:
- `EarlyFireJoinHintTest`: an early-fire outer join is inferred as `changelogMode=[I,UA]`; an insert-only downstream fails with the tailored error; a negative-window join and an inner join both stay `[I]` while still carrying `earlyFireDelay` on the join node, so the hint is attached but inert rather than absent.
## 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): no
- Anything that affects deployment or recovery: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no (planner inference 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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40170][table-planner] Infer update-producing changelog mode for early-fire interval join [flink]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #28877:
URL: https://github.com/apache/flink/pull/28877#discussion_r3741240262
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala:
##########
@@ -362,9 +362,27 @@ class FlinkChangelogModeInferenceProgram extends FlinkOptimizeProgram[StreamOpti
val providedTrait = new ModifyKindSetTrait(builder.build())
createNewNode(over, children, providedTrait, requiredTrait, requester)
- case _: StreamPhysicalTemporalSort | _: StreamPhysicalIntervalJoin |
- _: StreamPhysicalPythonOverAggregate =>
- // TemporalSort, IntervalJoin only support consuming insert-only
+ case intervalJoin: StreamPhysicalIntervalJoin =>
+ // The interval join consumes insert-only input. Without the EARLY_FIRE hint it also only
+ // produces insert-only changes; an early-firing outer join additionally produces update
+ // changes, because it speculatively emits a padded row and later corrects it on a match.
+ val children = visitChildren(intervalJoin, ModifyKindSetTrait.INSERT_ONLY)
+ val builder = ModifyKindSet.newBuilder().addContainedKind(ModifyKind.INSERT)
+ if (intervalJoin.produceEarlyFireUpdates) {
+ builder.addContainedKind(ModifyKind.UPDATE)
+ }
+ val providedTrait = new ModifyKindSetTrait(builder.build())
+ if (intervalJoin.produceEarlyFireUpdates && !providedTrait.satisfies(requiredTrait)) {
+ throw new TableException(
+ s"$requester is insert-only, but the EARLY_FIRE hint makes this outer interval join " +
+ "produce update changes (a padded row is emitted speculatively and later corrected " +
+ "on a match). Remove the EARLY_FIRE hint, or write into a downstream/sink that " +
+ "accepts update changes.")
+ }
Review Comment:
Reworded. Went with "doesn't support consuming update changes" to match the existing phrasing in this file (line 609) rather than "requires insert-only", so it reads like the other messages while still naming the hint.
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalIntervalJoin.scala:
##########
@@ -64,6 +64,16 @@ class StreamPhysicalIntervalJoin(
override def requireWatermark: Boolean = windowBounds.isEventTime
+ /**
+ * Whether this interval join produces update changes because of the EARLY_FIRE hint. Only an
+ * outer join with a non-negative window can speculatively emit a padded row and later correct it;
+ * a negative-window join only ever emits inserts, so it must stay insert-only even with the hint
+ * set.
+ */
+ def produceEarlyFireUpdates: Boolean =
+ earlyFireDelay != null && getJoinType.isOuterJoin &&
+ (windowBounds.getLeftUpperBound - windowBounds.getLeftLowerBound) >= 0
Review Comment:
Changed to `upper >= lower`.
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40170][table-planner] Infer update-producing changelog mode for early-fire interval join [flink]
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal merged PR #28877:
URL: https://github.com/apache/flink/pull/28877
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40170][table-planner] Infer update-producing changelog mode for early-fire interval join [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #28877:
URL: https://github.com/apache/flink/pull/28877#issuecomment-5160723338
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "8f7448f7cb64bb68c5556bc1bd290b6a25482592",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "8f7448f7cb64bb68c5556bc1bd290b6a25482592",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8f7448f7cb64bb68c5556bc1bd290b6a25482592 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40170][table-planner] Infer update-producing changelog mode for early-fire interval join [flink]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #28877:
URL: https://github.com/apache/flink/pull/28877#discussion_r3740420425
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalIntervalJoin.scala:
##########
@@ -64,6 +64,16 @@ class StreamPhysicalIntervalJoin(
override def requireWatermark: Boolean = windowBounds.isEventTime
+ /**
+ * Whether this interval join produces update changes because of the EARLY_FIRE hint. Only an
+ * outer join with a non-negative window can speculatively emit a padded row and later correct it;
+ * a negative-window join only ever emits inserts, so it must stay insert-only even with the hint
+ * set.
+ */
+ def produceEarlyFireUpdates: Boolean =
+ earlyFireDelay != null && getJoinType.isOuterJoin &&
+ (windowBounds.getLeftUpperBound - windowBounds.getLeftLowerBound) >= 0
Review Comment:
`produceEarlyFireUpdates` checks non-negative window span via subtraction `(upper - lower) >= 0`. Since bounds are `long`s, this can overflow for extreme values and is harder to read than a direct comparison. Prefer `upper >= lower` to express “non-negative window” safely.
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala:
##########
@@ -362,9 +362,27 @@ class FlinkChangelogModeInferenceProgram extends FlinkOptimizeProgram[StreamOpti
val providedTrait = new ModifyKindSetTrait(builder.build())
createNewNode(over, children, providedTrait, requiredTrait, requester)
- case _: StreamPhysicalTemporalSort | _: StreamPhysicalIntervalJoin |
- _: StreamPhysicalPythonOverAggregate =>
- // TemporalSort, IntervalJoin only support consuming insert-only
+ case intervalJoin: StreamPhysicalIntervalJoin =>
+ // The interval join consumes insert-only input. Without the EARLY_FIRE hint it also only
+ // produces insert-only changes; an early-firing outer join additionally produces update
+ // changes, because it speculatively emits a padded row and later corrects it on a match.
+ val children = visitChildren(intervalJoin, ModifyKindSetTrait.INSERT_ONLY)
+ val builder = ModifyKindSet.newBuilder().addContainedKind(ModifyKind.INSERT)
+ if (intervalJoin.produceEarlyFireUpdates) {
+ builder.addContainedKind(ModifyKind.UPDATE)
+ }
+ val providedTrait = new ModifyKindSetTrait(builder.build())
+ if (intervalJoin.produceEarlyFireUpdates && !providedTrait.satisfies(requiredTrait)) {
+ throw new TableException(
+ s"$requester is insert-only, but the EARLY_FIRE hint makes this outer interval join " +
+ "produce update changes (a padded row is emitted speculatively and later corrected " +
+ "on a match). Remove the EARLY_FIRE hint, or write into a downstream/sink that " +
+ "accepts update changes.")
+ }
Review Comment:
The insert-only guard message says "$requester is insert-only", but `requester` is a label for the consumer that *requires* insert-only changes. Using “requires insert-only changes” is more accurate and avoids implying the node itself is inherently insert-only.
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40170][table-planner] Infer update-producing changelog mode for early-fire interval join [flink]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on PR #28877:
URL: https://github.com/apache/flink/pull/28877#issuecomment-5227526443
Copilot's two comments are resolved in 5e425a9783b: the insert-only error now uses the same "doesn't support consuming ... changes" wording as the rest of the file, and the window-span check compares the bounds directly instead of subtracting them.
cc @RocMarshal, this one is ready for review when you have time. Thanks!
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40170][table-planner] Infer update-producing changelog mode for early-fire interval join [flink]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on PR #28877:
URL: https://github.com/apache/flink/pull/28877#issuecomment-5224568908
Hi @RocMarshal, #28827 is merged, so this one is rebased onto master and out of draft. The diff is standalone now: 5 files, adding update-producing changelog inference for an early-fire outer interval join.
One thing worth flagging: this changes an existing test's sink. `MySink` in `EarlyFireJoinHintTest` was insert-only, and an early-fire outer join now produces updates, so `testEarlyFireJsonPlanRoundTrip` no longer plans against it. I declared `sink-insert-only=false` on that sink rather than weakening the test. That is the one existing behavior this PR changes.
PTAL when you have a moment. Thanks!
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40170][table-planner] Infer update-producing changelog mode for early-fire interval join [flink]
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on PR #28877:
URL: https://github.com/apache/flink/pull/28877#issuecomment-5241032626
Thanks @weiqingy for the hard work.
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org