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