[FLINK-40169][table] Add target option to the EARLY_FIRE hint

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/EarlyFireJoinHintOptions.java

@@ -36,6 +36,18 @@
 @PublicEvolving
 public class EarlyFireJoinHintOptions {
 
+    /** The only operator kind the EARLY_FIRE hint applies to. */
+    public static final String INTERVAL_JOIN = "interval_join";
+
+    public static final ConfigOption<String> TARGET =
+            key("target")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "The operator kind that the EARLY_FIRE hint applies to. Currently only"
+                                    + " 'interval_join' is supported. When omitted, the hint applies"
+                                    + " to the interval join.");
+
     public static final ConfigOption<Duration> DELAY =
             key("delay")
                     .durationType()
@@ -60,6 +72,7 @@ public class EarlyFireJoinHintOptions {
     static {
         requiredKeys.add(DELAY);
 
+        supportedKeys.add(TARGET);
         supportedKeys.add(DELAY);
         supportedKeys.add(TIME_MODE);
     }

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/hint/FlinkHintStrategies.java

@@ -308,6 +308,14 @@ private static HintOptionChecker fixedSizeListOptionChecker(int size) {
                         "Invalid EARLY_FIRE hint option: {} value should be at least 1 millisecond but was {}",
                         EarlyFireJoinHintOptions.DELAY.key(),
                         delay);
+
+                String target = conf.get(EarlyFireJoinHintOptions.TARGET);
+                litmus.check(
+                        null == target || EarlyFireJoinHintOptions.INTERVAL_JOIN.equals(target),
+                        "Invalid EARLY_FIRE hint option: {} value '{}' is not supported, only '{}' is supported currently",
+                        EarlyFireJoinHintOptions.TARGET.key(),
+                        target,
+                        EarlyFireJoinHintOptions.INTERVAL_JOIN);
                 return true;
             };
 

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecIntervalJoin.java

@@ -28,6 +28,7 @@
 import org.apache.flink.streaming.api.transformations.TwoInputTransformation;
 import org.apache.flink.streaming.api.transformations.UnionTransformation;
 import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.api.config.EarlyFireJoinHintOptions;
 import org.apache.flink.table.api.config.ExecutionConfigOptions;
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.planner.delegation.PlannerBase;
@@ -59,11 +60,14 @@
 
 import org.apache.flink.shaded.guava33.com.google.common.collect.Lists;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.annotation.Nullable;
+
 import java.util.List;
 
 /** {@link StreamExecNode} for a time interval stream join. */
@@ -91,13 +95,27 @@ public class StreamExecIntervalJoin extends ExecNodeBase<RowData>
     public static final String INTERVAL_JOIN_TRANSFORMATION = "interval-join";
 
     public static final String FIELD_NAME_INTERVAL_JOIN_SPEC = "intervalJoinSpec";
+    public static final String FIELD_NAME_EARLY_FIRE_DELAY = "earlyFireDelay";
+    public static final String FIELD_NAME_EARLY_FIRE_TIME_MODE = "earlyFireTimeMode";
 
     @JsonProperty(FIELD_NAME_INTERVAL_JOIN_SPEC)
     private final IntervalJoinSpec intervalJoinSpec;
 
+    @Nullable
+    @JsonProperty(FIELD_NAME_EARLY_FIRE_DELAY)
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    private final Long earlyFireDelay;
+
+    @Nullable
+    @JsonProperty(FIELD_NAME_EARLY_FIRE_TIME_MODE)
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    private final EarlyFireJoinHintOptions.TimeMode earlyFireTimeMode;
+
     public StreamExecIntervalJoin(
             ReadableConfig tableConfig,
             IntervalJoinSpec intervalJoinSpec,
+            @Nullable Long earlyFireDelay,
+            @Nullable EarlyFireJoinHintOptions.TimeMode earlyFireTimeMode,
             InputProperty leftInputProperty,
             InputProperty rightInputProperty,
             RowType outputType,
@@ -107,6 +125,8 @@ public StreamExecIntervalJoin(
                 ExecNodeContext.newContext(StreamExecIntervalJoin.class),
                 ExecNodeContext.newPersistedConfig(StreamExecIntervalJoin.class, tableConfig),
                 intervalJoinSpec,
+                earlyFireDelay,
+                earlyFireTimeMode,
                 Lists.newArrayList(leftInputProperty, rightInputProperty),
                 outputType,
                 description);
@@ -118,12 +138,17 @@ public StreamExecIntervalJoin(
             @JsonProperty(FIELD_NAME_TYPE) ExecNodeContext context,
             @JsonProperty(FIELD_NAME_CONFIGURATION) ReadableConfig persistedConfig,
             @JsonProperty(FIELD_NAME_INTERVAL_JOIN_SPEC) IntervalJoinSpec intervalJoinSpec,
+            @Nullable @JsonProperty(FIELD_NAME_EARLY_FIRE_DELAY) Long earlyFireDelay,
+            @Nullable @JsonProperty(FIELD_NAME_EARLY_FIRE_TIME_MODE)
+                    EarlyFireJoinHintOptions.TimeMode earlyFireTimeMode,
             @JsonProperty(FIELD_NAME_INPUT_PROPERTIES) List<InputProperty> inputProperties,
             @JsonProperty(FIELD_NAME_OUTPUT_TYPE) RowType outputType,
             @JsonProperty(FIELD_NAME_DESCRIPTION) String description) {
         super(id, context, persistedConfig, inputProperties, outputType, description);
         Preconditions.checkArgument(inputProperties.size() == 2);
         this.intervalJoinSpec = Preconditions.checkNotNull(intervalJoinSpec);
+        this.earlyFireDelay = earlyFireDelay;
+        this.earlyFireTimeMode = earlyFireTimeMode;
     }
 
     @Override

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalIntervalJoinRule.java

@@ -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,59 @@ public FlinkRelNode transform(
                 join.getJoinType(),
                 join.getCondition(),
                 tuple2.f1.getOrElse(() -> join.getCluster().getRexBuilder().makeLiteral(true)),
-                tuple2.f0.get());
+                tuple2.f0.get(),
+                earlyFire.delay,
+                earlyFire.timeMode);
+    }
+
+    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);
+        // target scopes the hint to one operator kind: this rule applies it only when it targets
+        // the interval join, and leaves a hint aimed at any other operator kind untouched.
+        String target = conf.get(EarlyFireJoinHintOptions.TARGET);
+        if (target != null && !EarlyFireJoinHintOptions.INTERVAL_JOIN.equals(target)) {
+            return new EarlyFire(null, null);
+        }
+        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}. */

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalIntervalJoin.scala

@@ -18,6 +18,7 @@
 package org.apache.flink.table.planner.plan.nodes.physical.stream
 
 import org.apache.flink.table.api.TableException
+import org.apache.flink.table.api.config.EarlyFireJoinHintOptions.TimeMode
 import org.apache.flink.table.planner.calcite.FlinkTypeFactory
 import org.apache.flink.table.planner.plan.nodes.exec.{ExecNode, InputProperty}
 import org.apache.flink.table.planner.plan.nodes.exec.spec.IntervalJoinSpec
@@ -45,7 +46,9 @@ class StreamPhysicalIntervalJoin(
     val originalCondition: RexNode,
     // remaining join condition contains all of join condition except window bounds
     remainingCondition: RexNode,
-    windowBounds: WindowBounds)
+    windowBounds: WindowBounds,
+    earlyFireDelay: java.lang.Long,
+    earlyFireTimeMode: TimeMode)
   extends CommonPhysicalJoin(cluster, traitSet, leftRel, rightRel, remainingCondition, joinType)
   with StreamPhysicalRel {
 
@@ -76,7 +79,9 @@ class StreamPhysicalIntervalJoin(
       joinType,
       originalCondition,
       conditionExpr,
-      windowBounds)
+      windowBounds,
+      earlyFireDelay,
+      earlyFireTimeMode)
   }
 
   override def explainTerms(pw: RelWriter): RelWriter = {
@@ -98,12 +103,16 @@ class StreamPhysicalIntervalJoin(
           preferExpressionFormat(pw),
           pw.getDetailLevel))
       .item("select", getRowType.getFieldNames.mkString(", "))
+      .itemIf("earlyFireDelay", earlyFireDelay, earlyFireDelay != null)
+      .itemIf("earlyFireTimeMode", earlyFireTimeMode, earlyFireTimeMode != null)
   }
 
   override def translateToExecNode(): ExecNode[_] = {
     new StreamExecIntervalJoin(
       unwrapTableConfig(this),
       new IntervalJoinSpec(joinSpec, windowBounds),
+      earlyFireDelay,
+      earlyFireTimeMode,
       InputProperty.DEFAULT,
       InputProperty.DEFAULT,
       FlinkTypeFactory.toLogicalRowType(getRowType),

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/hints/stream/EarlyFireJoinHintTest.java

@@ -65,6 +65,14 @@ void before() {
                                 + "  'connector' = 'values',\n"
                                 + "  'bounded' = 'false'\n"
                                 + ")");
+        util.tableEnv()
+                .executeSql(
+                        "CREATE TABLE MySink (\n"
+                                + "  a INT,\n"
+                                + "  b VARCHAR\n"
+                                + ") WITH (\n"
+                                + "  'connector' = 'values'\n"
+                                + ")");
     }
 
     @Test
@@ -132,6 +140,17 @@ void testEarlyFireListOptionsRejected() {
                 .hasMessageContaining("only support key-value options");
     }
 
+    @Test
+    void testEarlyFireUnsupportedTarget() {
+        String sql =
+                "SELECT /*+ EARLY_FIRE('target'='window_join', 'delay'='5s') */ t1.a, t2.b\n"
+                        + "FROM MyTable t1 LEFT OUTER JOIN MyTable2 t2 ON\n"
+                        + "  t1.a = t2.a AND\n"
+                        + "  t1.rowtime BETWEEN t2.rowtime - INTERVAL '10' SECOND AND t2.rowtime + INTERVAL '1' HOUR";
+        assertThatThrownBy(() -> verify(sql))
+                .hasMessageContaining("target value 'window_join' is not supported");
+    }
+
     @Test
     void testEarlyFireLowerCaseHintNamePreservesOptions() {
         String sql =
@@ -142,6 +161,68 @@ void testEarlyFireLowerCaseHintNamePreservesOptions() {
         verify(sql);
     }
 
+    @Test
+    void testEarlyFireOnRowTimeLeftOuterJoin() {
+        String sql =
+                "SELECT /*+ EARLY_FIRE('delay'='5s') */ t1.a, t2.b\n"
+                        + "FROM MyTable t1 LEFT OUTER JOIN MyTable2 t2 ON\n"
+                        + "  t1.a = t2.a AND\n"
+                        + "  t1.rowtime BETWEEN t2.rowtime - INTERVAL '10' SECOND AND t2.rowtime + INTERVAL '1' HOUR";
+        verify(sql);
+    }
+
+    @Test
+    void testEarlyFireExplicitTargetIntervalJoin() {
+        String sql =
+                "SELECT /*+ EARLY_FIRE('target'='interval_join', 'delay'='5s') */ t1.a, t2.b\n"
+                        + "FROM MyTable t1 LEFT OUTER JOIN MyTable2 t2 ON\n"
+                        + "  t1.a = t2.a AND\n"
+                        + "  t1.rowtime BETWEEN t2.rowtime - INTERVAL '10' SECOND AND t2.rowtime + INTERVAL '1' HOUR";
+        verify(sql);
+    }
+
+    @Test
+    void testEarlyFireRowTimeOnProcTimeJoin() {
+        String sql =
+                "SELECT /*+ EARLY_FIRE('delay'='5s', 'time-mode'='rowtime') */ t1.a, t2.b\n"
+                        + "FROM MyTable t1 LEFT OUTER JOIN MyTable2 t2 ON\n"
+                        + "  t1.a = t2.a AND\n"
+                        + "  t1.proctime BETWEEN t2.proctime - INTERVAL '1' HOUR AND t2.proctime + INTERVAL '1' HOUR";
+        assertThatThrownBy(() -> verify(sql))
+                .hasStackTraceContaining("requires a row-time interval join");
+    }
+
+    @Test
+    void testEarlyFireProcTimeOnRowTimeJoin() {
+        String sql =
+                "SELECT /*+ EARLY_FIRE('delay'='5s', 'time-mode'='proctime') */ t1.a, t2.b\n"
+                        + "FROM MyTable t1 LEFT OUTER JOIN MyTable2 t2 ON\n"
+                        + "  t1.a = t2.a AND\n"
+                        + "  t1.rowtime BETWEEN t2.rowtime - INTERVAL '10' SECOND AND t2.rowtime + INTERVAL '1' HOUR";
+        assertThatThrownBy(() -> verify(sql)).hasStackTraceContaining("not yet supported");
+    }
+
+    @Test
+    void testEarlyFireOnProcTimeLeftOuterJoin() {
+        String sql =
+                "SELECT /*+ EARLY_FIRE('delay'='5s') */ t1.a, t2.b\n"
+                        + "FROM MyTable t1 LEFT OUTER JOIN MyTable2 t2 ON\n"
+                        + "  t1.a = t2.a AND\n"
+                        + "  t1.proctime BETWEEN t2.proctime - INTERVAL '1' HOUR AND t2.proctime + INTERVAL '1' HOUR";
+        verify(sql);
+    }
+
+    @Test
+    void testEarlyFireJsonPlanRoundTrip() {
+        String insert =
+                "INSERT INTO MySink\n"
+                        + "SELECT /*+ EARLY_FIRE('delay'='5s') */ t1.a, t2.b\n"
+                        + "FROM MyTable t1 LEFT OUTER JOIN MyTable2 t2 ON\n"
+                        + "  t1.a = t2.a AND\n"
+                        + "  t1.rowtime BETWEEN t2.rowtime - INTERVAL '10' SECOND AND t2.rowtime + INTERVAL '1' HOUR";
+        util.verifyJsonPlan(insert);
+    }
+
     private void verify(String sql) {
         util.doVerifyPlan(
                 sql,