[FLINK-40168][table] Thread the EARLY_FIRE hint into the interval join

flink-docs/src/main/java/org/apache/flink/docs/util/ConfigurationOptionLocator.java

@@ -105,7 +105,8 @@ public class ConfigurationOptionLocator {
                             "org.apache.flink.state.rocksdb.PredefinedOptions",
                             "org.apache.flink.python.PythonConfig",
                             "org.apache.flink.cep.configuration.SharedBufferCacheConfig",
-                            "org.apache.flink.table.api.config.LookupJoinHintOptions"));
+                            "org.apache.flink.table.api.config.LookupJoinHintOptions",
+                            "org.apache.flink.table.api.config.EarlyFireJoinHintOptions"));
 
     private static final String DEFAULT_PATH_PREFIX = "src/main/java";
 

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

@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.api.config;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+
+import org.apache.flink.shaded.guava33.com.google.common.collect.ImmutableSet;
+
+import java.time.Duration;
+import java.util.HashSet;
+import java.util.Set;
+
+import static org.apache.flink.configuration.ConfigOptions.key;
+
+/**
+ * This class holds hint option name definitions for EARLY_FIRE join hints based on {@link
+ * org.apache.flink.configuration.ConfigOption}.
+ */
+@PublicEvolving
+public class EarlyFireJoinHintOptions {
+
+    public static final ConfigOption<Duration> DELAY =
+            key("delay")
+                    .durationType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "The delay between the time an unmatched outer row becomes eligible to"
+                                    + " be emitted with null padding and the time it is actually"
+                                    + " emitted. Must be at least 1 millisecond.");
+
+    public static final ConfigOption<TimeMode> TIME_MODE =
+            key("time-mode")
+                    .enumType(TimeMode.class)
+                    .noDefaultValue()
+                    .withDescription(
+                            "The time domain that drives the early-fire delay, can be 'rowtime' or"
+                                    + " 'proctime'. If not set, it defaults to the time domain of"
+                                    + " the interval join.");
+
+    private static final Set<ConfigOption<?>> requiredKeys = new HashSet<>();
+    private static final Set<ConfigOption<?>> supportedKeys = new HashSet<>();
+
+    static {
+        requiredKeys.add(DELAY);
+
+        supportedKeys.add(DELAY);
+        supportedKeys.add(TIME_MODE);
+    }
+
+    public static ImmutableSet<ConfigOption> getRequiredOptions() {
+        return ImmutableSet.copyOf(requiredKeys);
+    }
+
+    public static ImmutableSet<ConfigOption> getSupportedOptions() {
+        return ImmutableSet.copyOf(supportedKeys);
+    }
+
+    /** The time domain that drives the early-fire delay. */
+    @PublicEvolving
+    public enum TimeMode {
+        ROWTIME,
+        PROCTIME
+    }
+}

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

@@ -46,7 +46,8 @@ protected RelNode doVisit(RelNode node) {
 
                                     changed.set(true);
                                     if (JoinStrategy.isJoinStrategy(capitalHintName)) {
-                                        if (JoinStrategy.isLookupHint(hint.hintName)) {
+                                        if (JoinStrategy.isLookupHint(hint.hintName)
+                                                || JoinStrategy.isEarlyFireHint(hint.hintName)) {
                                             return RelHint.builder(capitalHintName)
                                                     .hintOptions(hint.kvOptions)
                                                     .inheritPath(hint.inheritPath)

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

@@ -20,6 +20,7 @@
 
 import org.apache.flink.configuration.ConfigOption;
 import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.api.config.EarlyFireJoinHintOptions;
 import org.apache.flink.table.api.config.LookupJoinHintOptions;
 import org.apache.flink.table.factories.FactoryUtil;
 import org.apache.flink.table.planner.plan.rules.logical.WrapJsonAggFunctionArgumentsRule;
@@ -36,6 +37,9 @@
 import java.time.Duration;
 import java.util.Collections;
 import java.util.Optional;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.stream.Collectors;
 
 /**
  * A collection of Flink style {@link HintStrategy}s.
@@ -135,6 +139,11 @@ public static HintStrategyTable createHintStrategyTable() {
                                                 HintPredicates.JOIN, HintPredicates.AGGREGATE))
                                 .optionChecker(STATE_TTL_NON_EMPTY_KV_OPTION_CHECKER)
                                 .build())
+                .hintStrategy(
+                        JoinStrategy.EARLY_FIRE.getJoinHintName(),
+                        HintStrategy.builder(HintPredicates.JOIN)
+                                .optionChecker(EARLY_FIRE_KV_OPTION_CHECKER)
+                                .build())
                 .build();
     }
 
@@ -253,6 +262,55 @@ private static HintOptionChecker fixedSizeListOptionChecker(int size) {
                 return true;
             };
 
+    private static final HintOptionChecker EARLY_FIRE_KV_OPTION_CHECKER =
+            (earlyFireHint, litmus) -> {
+                litmus.check(
+                        earlyFireHint.listOptions.size() == 0,
+                        "Invalid list options in EARLY_FIRE hint, only support key-value options.");
+
+                Configuration conf = Configuration.fromMap(earlyFireHint.kvOptions);
+                ImmutableSet<ConfigOption> requiredKeys =
+                        EarlyFireJoinHintOptions.getRequiredOptions();
+                litmus.check(
+                        requiredKeys.stream().allMatch(conf::contains),
+                        "Invalid EARLY_FIRE hint: incomplete required option(s): {}",
+                        requiredKeys);
+
+                ImmutableSet<ConfigOption> supportedKeys =
+                        EarlyFireJoinHintOptions.getSupportedOptions();
+                Set<String> supportedKeyNames =
+                        supportedKeys.stream().map(ConfigOption::key).collect(Collectors.toSet());
+                Set<String> unknownKeys =
+                        earlyFireHint.kvOptions.keySet().stream()
+                                .filter(key -> !supportedKeyNames.contains(key))
+                                .collect(Collectors.toCollection(TreeSet::new));
+                litmus.check(
+                        unknownKeys.isEmpty(),
+                        "Unsupported EARLY_FIRE hint option(s) {}, supported options are {}.",
+                        unknownKeys,
+                        new TreeSet<>(supportedKeyNames));
+                litmus.check(
+                        earlyFireHint.kvOptions.size() <= supportedKeys.size(),
+                        "Too many EARLY_FIRE hint options {} beyond max number of supported options {}",
+                        earlyFireHint.kvOptions.size(),
+                        supportedKeys.size());
+
+                try {
+                    // try to validate all hint options by parsing them
+                    supportedKeys.forEach(conf::get);
+                } catch (IllegalArgumentException e) {
+                    litmus.fail("Invalid EARLY_FIRE hint options: {}", e.getMessage());
+                }
+
+                Duration delay = conf.get(EarlyFireJoinHintOptions.DELAY);
+                litmus.check(
+                        null != delay && delay.toMillis() > 0,
+                        "Invalid EARLY_FIRE hint option: {} value should be at least 1 millisecond but was {}",
+                        EarlyFireJoinHintOptions.DELAY.key(),
+                        delay);
+                return true;
+            };
+
     private static final HintOptionChecker STATE_TTL_NON_EMPTY_KV_OPTION_CHECKER =
             (ttlHint, litmus) -> {
                 litmus.check(

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

@@ -50,6 +50,12 @@ public enum JoinStrategy {
     /** Instructs the optimizer to use lookup join strategy. Only accept key-value hint options. */
     LOOKUP("LOOKUP"),
 
+    /**
+     * Instructs an outer interval join to emit unmatched outer rows with null padding after a
+     * configurable delay. Only accept key-value hint options.
+     */
+    EARLY_FIRE("EARLY_FIRE"),
+
     /**
      * Instructs the optimizer to use multi-way join strategy for streaming queries. This hint
      * allows specifying multiple tables to be joined together in a single {@link
@@ -89,6 +95,7 @@ public static boolean validOptions(String hintName, List<String> options) {
             case NEST_LOOP:
                 return options.size() > 0;
             case LOOKUP:
+            case EARLY_FIRE:
                 return null == options || options.size() == 0;
             case MULTI_JOIN:
                 return options.size() > 0;
@@ -101,4 +108,10 @@ public static boolean isLookupHint(String hintName) {
         return isJoinStrategy(formalizedHintName)
                 && JoinStrategy.valueOf(formalizedHintName) == LOOKUP;
     }
+
+    public static boolean isEarlyFireHint(String hintName) {
+        String formalizedHintName = hintName.toUpperCase(Locale.ROOT);
+        return isJoinStrategy(formalizedHintName)
+                && JoinStrategy.valueOf(formalizedHintName) == EARLY_FIRE;
+    }
 }

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/optimize/QueryHintsResolver.java

@@ -146,6 +146,13 @@ private List<RelHint> validateAndGetNewHints(
                     updateInfoForOptionCheck(hint.hintName, rightName);
                     newHints.add(hint);
                 }
+            } else if (JoinStrategy.isEarlyFireHint(hint.hintName)) {
+                // EARLY_FIRE carries only key-value options and is not bound to a specific input
+                // side, so it is passed through unchanged once its options are validated by the
+                // hint option checker.
+                allHints.add(trimInheritPath(hint));
+                validHints.add(trimInheritPath(hint));
+                newHints.add(hint);
             } else if (JoinStrategy.isJoinStrategy(hint.hintName)) {
                 allHints.add(trimInheritPath(hint));
                 // add options about this hint for finally checking