[FLINK-40168][table] Thread the EARLY_FIRE hint into the interval join #28796
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -19,9 +19,13 @@ | |
| package org.apache.flink.table.planner.plan.rules.physical.stream; | ||
|
|
||
| import org.apache.flink.api.java.tuple.Tuple2; | ||
| import org.apache.flink.configuration.Configuration; | ||
| import org.apache.flink.table.api.TableException; | ||
| import org.apache.flink.table.api.ValidationException; | ||
| import org.apache.flink.table.api.config.EarlyFireJoinHintOptions; | ||
| import org.apache.flink.table.api.config.EarlyFireJoinHintOptions.TimeMode; | ||
| import org.apache.flink.table.planner.calcite.FlinkTypeFactory; | ||
| import org.apache.flink.table.planner.hint.JoinStrategy; | ||
| import org.apache.flink.table.planner.plan.nodes.FlinkRelNode; | ||
| import org.apache.flink.table.planner.plan.nodes.exec.spec.IntervalJoinSpec; | ||
| import org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalJoin; | ||
|
|
@@ -32,11 +36,16 @@ | |
| import org.apache.calcite.plan.RelOptRuleCall; | ||
| import org.apache.calcite.plan.RelTraitSet; | ||
| import org.apache.calcite.rel.RelNode; | ||
| import org.apache.calcite.rel.hint.RelHint; | ||
| import org.apache.calcite.rel.type.RelDataType; | ||
| import org.apache.calcite.rex.RexNode; | ||
| import org.immutables.value.Value; | ||
|
|
||
| import javax.annotation.Nullable; | ||
|
|
||
| import java.time.Duration; | ||
| import java.util.Collection; | ||
| import java.util.List; | ||
| import java.util.function.Function; | ||
| import java.util.stream.Collectors; | ||
|
|
||
|
|
@@ -133,6 +142,8 @@ public FlinkRelNode transform( | |
| RelTraitSet providedTraitSet) { | ||
| Tuple2<Option<IntervalJoinSpec.WindowBounds>, Option<RexNode>> tuple2 = | ||
| extractWindowBounds(join); | ||
| boolean isEventTime = tuple2.f0.get().isEventTime(); | ||
| EarlyFire earlyFire = extractEarlyFire(join.getHints(), isEventTime); | ||
| return new StreamPhysicalIntervalJoin( | ||
| join.getCluster(), | ||
| providedTraitSet, | ||
|
|
@@ -141,7 +152,53 @@ public FlinkRelNode transform( | |
| join.getJoinType(), | ||
| join.getCondition(), | ||
| tuple2.f1.getOrElse(() -> join.getCluster().getRexBuilder().makeLiteral(true)), | ||
| tuple2.f0.get(), | ||
| earlyFire.delay, | ||
| earlyFire.timeMode); | ||
|
Comment on lines
+156
to
+157
Contributor
There was a problem hiding this comment. At Anchor-A, the early fire parameters are now encapsulated within a dedicated class. That said, the call sites currently split them apart for invocation. Given the semantics of this link and its subsequent stages, are there any specific downsides to propagating these parameters strictly via the EarlyFire(or EarlyFireConfig) wrapper? Since these two parameters are almost always co-located, I’m merely questioning the trade-off. If using EarlyFire(or EarlyFireConfig) proves cumbersome, decoupling them is an acceptable compromise. And I'd like to hear more ideas about it
Contributor
Author
There was a problem hiding this comment. Thanks @RocMarshal for the review and the +1. Good observation. EarlyFire is a small planner-internal holder for the result of extractEarlyFire(), effectively Java's stand-in for a tuple, because that method resolves the effective time mode and performs the domain validation together. It is unpacked once at its single call site into the two values propagated downstream. My reason for continuing to propagate them separately is the eventual ExecNode boundary, which is also the compiled-plan serde boundary. earlyFireDelay and earlyFireTimeMode are persisted as two optional scalar properties on StreamExecIntervalJoin. Keeping them flat follows existing ExecNode patterns for small optional properties and avoids adding a nullable nested spec to the plan format for two values. IntervalJoinSpec is nested, but it represents an always-present group of interval-join properties, whereas early fire itself is optional. Regarding whether more configuration will join these values: the next PR adds a target option, but it remains a rule-level applicability gate. It determines whether the hint applies to this operator kind and is discarded after extraction, so the propagated state remains exactly delay plus timeMode. My preference is therefore to keep the two flat fields for now. If another parameter eventually needs to reach the operator, promoting them to a small shared spec would make more sense. If you prefer establishing the wrapper now, though, I'm happy to change it. |
||
| } | ||
|
|
||
| private static EarlyFire extractEarlyFire(List<RelHint> hints, boolean isEventTime) { | ||
| RelHint earlyFireHint = null; | ||
| for (RelHint hint : hints) { | ||
| if (JoinStrategy.isEarlyFireHint(hint.hintName)) { | ||
| earlyFireHint = hint; | ||
| break; | ||
| } | ||
| } | ||
| if (earlyFireHint == null) { | ||
| return new EarlyFire(null, null); | ||
| } | ||
|
|
||
| Configuration conf = Configuration.fromMap(earlyFireHint.kvOptions); | ||
| Duration delay = conf.get(EarlyFireJoinHintOptions.DELAY); | ||
| TimeMode timeMode = conf.get(EarlyFireJoinHintOptions.TIME_MODE); | ||
| if (timeMode == null) { | ||
| timeMode = isEventTime ? TimeMode.ROWTIME : TimeMode.PROCTIME; | ||
| } | ||
|
|
||
| if (!isEventTime && timeMode == TimeMode.ROWTIME) { | ||
| throw new ValidationException( | ||
| "EARLY_FIRE hint requested row-time triggering on a processing-time interval" | ||
| + " join. Row-time triggering requires a row-time interval join."); | ||
| } | ||
| if (isEventTime && timeMode == TimeMode.PROCTIME) { | ||
| // Processing-time triggering on an event-time interval join is not supported. | ||
| throw new TableException( | ||
| "EARLY_FIRE hint requested processing-time triggering on a row-time interval" | ||
| + " join, which is not yet supported."); | ||
| } | ||
|
|
||
| return new EarlyFire(delay == null ? null : delay.toMillis(), timeMode); | ||
| } | ||
|
|
||
| private static final class EarlyFire { | ||
| @Nullable private final Long delay; | ||
| @Nullable private final TimeMode timeMode; | ||
|
|
||
| EarlyFire(@Nullable Long delay, @Nullable TimeMode timeMode) { | ||
| this.delay = delay; | ||
| this.timeMode = timeMode; | ||
| } | ||
| } | ||
|
|
||
| /** Configuration for {@link StreamPhysicalIntervalJoinRule}. */ | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
anchor-A