[FLINK-41446][table-planner] Ignore the EARLY_FIRE join hint in batch mode

docs/content.zh/docs/sql/reference/queries/joins.md

@@ -126,7 +126,7 @@ AND o.order_time BETWEEN s.ship_time - INTERVAL '1' HOUR AND s.ship_time
 | `delay` | 是 | 一个至少为 1 毫秒的时间长度（例如 `'5s'`）。在 `rowtime` 模式下从行的时间属性开始计算；在 `proctime` 模式下从该行被处理时的墙上时钟时间开始计算。窗口内没有匹配时，在该延迟过去后输出推测性的 null 补齐行。 |
 | `time-mode` | 否 | 取值为 `rowtime` 或 `proctime`，用于控制延迟是相对水位线（`rowtime`）还是相对墙上时钟（`proctime`）来计算。事件时间 join 默认为 `rowtime`，处理时间 join 默认为 `proctime`。在事件时间 join 上设置 `'time-mode'='proctime'` 会改为按墙上时钟触发提前触发，适用于水位线稀疏或停滞的场景；匹配和状态清理仍按事件时间进行，并且 join 会在每一侧输入上额外维护状态，用于记录待触发的提前触发。在处理时间 join 上设置 `'time-mode'='rowtime'` 会报错。 |
 
-该提示仅对 outer join（`LEFT`、`RIGHT`、`FULL`）生效。在 inner join 以及窗口跨度为负的 join 上会被忽略，这些情况仍保持仅追加。
+该提示仅对 outer join（`LEFT`、`RIGHT`、`FULL`）生效。在 inner join 以及窗口跨度为负的 join 上会被忽略，这些情况仍保持仅追加。在批模式下，该提示在所有 join 上都会被忽略。
 
 **注意：** `EARLY_FIRE` 与 `table.exec.emit.early-fire.*` 配置无关，后者用于控制窗口聚合的提前触发，而非 interval join。
 

docs/content/docs/sql/reference/queries/joins.md

@@ -131,7 +131,7 @@ The hint accepts the following options:
 | `delay` | yes | A duration of at least 1 millisecond (for example `'5s'`). In `rowtime` mode it is measured from the row's time attribute; in `proctime` mode it is measured from the wall-clock time at which the row is processed. The speculative null-padded row is emitted once this delay elapses without a match. |
 | `time-mode` | no | Either `rowtime` or `proctime`. Controls whether the delay is measured against the watermark (`rowtime`) or wall-clock time (`proctime`). Defaults to `rowtime` for an event-time join and `proctime` for a processing-time join. Setting `'time-mode'='proctime'` on an event-time join triggers early fire on wall-clock time instead of the watermark, which helps when watermarks are sparse or stalled; matching and state cleanup still follow event time, and the join keeps extra state on each input to track the pending early fires. Setting `'time-mode'='rowtime'` on a processing-time join is an error. |
 
-The hint only affects outer joins (`LEFT`, `RIGHT`, `FULL`). It is ignored on inner joins and on joins whose window has a negative span, both of which remain append-only.
+The hint only affects outer joins (`LEFT`, `RIGHT`, `FULL`). It is ignored on inner joins and on joins whose window has a negative span, both of which remain append-only. In batch mode, the hint is ignored on all joins.
 
 **Note:** `EARLY_FIRE` is unrelated to the `table.exec.emit.early-fire.*` configuration, which controls early firing for windowed aggregations rather than interval joins.
 

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalJoinRuleBase.scala

@@ -143,7 +143,8 @@ trait BatchPhysicalJoinRuleBase {
       case JoinStrategy.NEST_LOOP =>
         checkNestLoopJoin(join, tableConfig, withHint)
 
-      case JoinStrategy.LOOKUP =>
+      // these hints do not select one of the batch join operators above
+      case JoinStrategy.LOOKUP | JoinStrategy.EARLY_FIRE =>
         (false, false)
 
       case _ =>

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/JoinUtil.scala

@@ -318,6 +318,10 @@ object JoinUtil {
   }
 
   def containsJoinStrategyHint(relHints: ImmutableList[RelHint]): Boolean = {
-    relHints.exists(relHint => JoinStrategy.isJoinStrategy(relHint.hintName))
+    // EARLY_FIRE does not choose a join strategy, so it must not mark the join as hinted
+    relHints.exists(
+      relHint =>
+        JoinStrategy.isJoinStrategy(relHint.hintName) &&
+          !JoinStrategy.isEarlyFireHint(relHint.hintName))
   }
 }

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

@@ -0,0 +1,104 @@
+/*
+ * 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.planner.plan.hints.batch;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.config.OptimizerConfigOptions;
+import org.apache.flink.table.planner.utils.BatchTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests that the batch planner ignores the EARLY_FIRE join hint. */
+class EarlyFireJoinHintTest extends TableTestBase {
+
+    private static final String INTERVAL_JOIN_QUERY =
+            "SELECT %s o.id, s.id FROM orders o %s shipments s\n"
+                    + "ON o.id = s.id AND o.ts BETWEEN s.ts - INTERVAL '1' MINUTE AND s.ts";
+
+    private static final String EARLY_FIRE_HINT = "/*+ EARLY_FIRE('delay'='5s') */";
+
+    private BatchTableTestUtil util;
+
+    @BeforeEach
+    void before() {
+        util = batchTestUtil(TableConfig.getDefault());
+        // the test util turns adaptive joins off; turn them back on as in the default config
+        util.tableEnv()
+                .getConfig()
+                .set(
+                        OptimizerConfigOptions.TABLE_OPTIMIZER_ADAPTIVE_BROADCAST_JOIN_STRATEGY,
+                        OptimizerConfigOptions.AdaptiveBroadcastJoinStrategy.AUTO);
+        for (String table : new String[] {"orders", "shipments"}) {
+            util.tableEnv()
+                    .executeSql(
+                            "CREATE TABLE "
+                                    + table
+                                    + " (\n"
+                                    + "  id BIGINT,\n"
+                                    + "  ts TIMESTAMP(3),\n"
+                                    + "  WATERMARK FOR ts AS ts\n"
+                                    + ") WITH (\n"
+                                    + "  'connector' = 'values',\n"
+                                    + "  'bounded' = 'true'\n"
+                                    + ")");
+        }
+    }
+
+    @Test
+    void testLeftJoinPlansAsWithoutHint() {
+        assertPlansAsWithoutHint("LEFT JOIN");
+    }
+
+    @Test
+    void testInnerJoinPlansAsWithoutHint() {
+        assertPlansAsWithoutHint("JOIN");
+    }
+
+    @Test
+    void testStrategyHintStillAppliesWithEarlyFire() {
+        String withEarlyFire =
+                optimizedPlan(
+                        String.format(
+                                INTERVAL_JOIN_QUERY,
+                                "/*+ EARLY_FIRE('delay'='5s'), SHUFFLE_MERGE(o) */",
+                                "JOIN"));
+        String strategyOnly =
+                optimizedPlan(
+                        String.format(INTERVAL_JOIN_QUERY, "/*+ SHUFFLE_MERGE(o) */", "JOIN"));
+        assertThat(withEarlyFire).isEqualTo(strategyOnly).doesNotContain("AdaptiveJoin(");
+    }
+
+    private void assertPlansAsWithoutHint(String join) {
+        String withHint = optimizedPlan(String.format(INTERVAL_JOIN_QUERY, EARLY_FIRE_HINT, join));
+        String withoutHint = optimizedPlan(String.format(INTERVAL_JOIN_QUERY, "", join));
+        // without an adaptive join here, the comparison would not cover the adaptive conversion
+        assertThat(withoutHint).contains("AdaptiveJoin(");
+        assertThat(withHint).isEqualTo(withoutHint);
+    }
+
+    /** Drops the abstract syntax tree, the only EXPLAIN section that prints the hint itself. */
+    private String optimizedPlan(String sql) {
+        String explain = util.tableEnv().explainSql(sql);
+        return explain.substring(explain.indexOf("== Optimized Physical Plan =="));
+    }
+}