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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
