[FLINK-40173][table-planner] Add restore coverage for early-fire interval join

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

@@ -430,7 +430,11 @@ private TwoInputTransformation<RowData, RowData, RowData> createRowTimeJoin(
                         joinFunction,
                         windowBounds.getLeftTimeIdx(),
                         windowBounds.getRightTimeIdx(),
-                        earlyFireDelay == null ? -1L : earlyFireDelay);
+                        earlyFireDelay == null ? -1L : earlyFireDelay,
+                        // Cross-domain flag: an event-time interval join early-fires on the wall
+                        // clock while keeping its event-time cleanup. The operator only acts on it
+                        // once early-firing is enabled (earlyFireDelay >= 0).
+                        earlyFireTimeMode == EarlyFireJoinHintOptions.TimeMode.PROCTIME);
         // TODO: add async version rowJoinFunc to use AsyncKeyedCoProcessOperator
         return ExecNodeUtil.createTwoInputTransformation(
                 leftInputTransform,

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

@@ -187,12 +187,6 @@ private static EarlyFire extractEarlyFire(List<RelHint> hints, boolean isEventTi
                     "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);
     }

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

@@ -202,7 +202,7 @@ void testEarlyFireProcTimeOnRowTimeJoin() {
                         + "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");
+        verify(sql);
     }
 
     @Test

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/IntervalJoinProcTimeEarlyFireRestoreTest.java

@@ -0,0 +1,67 @@
+/*
+ * 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.nodes.exec.stream;
+
+import org.apache.flink.table.planner.factories.TestValuesTableFactory;
+import org.apache.flink.table.planner.plan.nodes.exec.testutils.RestoreTestBase;
+import org.apache.flink.table.test.program.SourceTestStep;
+import org.apache.flink.table.test.program.TableTestProgram;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Restore tests for {@link StreamExecIntervalJoin} early-firing on processing time.
+ *
+ * <p>The early-fire timer lives on the wall clock while the join keeps its row-time cleanup, so the
+ * speculative pad is due at a processing-time instant recorded in the savepoint. The restored job
+ * ingests no data at all: the pad is emitted purely from restored state once that instant passes,
+ * which needs a source that stays open rather than one that ends input and closes the window.
+ */
+public class IntervalJoinProcTimeEarlyFireRestoreTest extends RestoreTestBase {
+
+    private static final long SAVEPOINT_READY_TIMEOUT_MILLIS = TimeUnit.MINUTES.toMillis(5);
+
+    public IntervalJoinProcTimeEarlyFireRestoreTest() {
+        super(StreamExecIntervalJoin.class, AfterRestoreSource.INFINITE);
+    }
+
+    @Override
+    public List<TableTestProgram> programs() {
+        return Collections.singletonList(
+                IntervalJoinTestPrograms.INTERVAL_JOIN_PROC_TIME_EARLY_FIRE);
+    }
+
+    @Override
+    protected void awaitSavepointReady(TableTestProgram program, List<CompletableFuture<?>> futures)
+            throws Exception {
+        // The join emits nothing before the savepoint, so the default sink-based trigger would fire
+        // immediately. Gate on the sources instead: stop-with-savepoint then drains their rows into
+        // keyed state, capturing the still-pending early-fire schedule.
+        for (SourceTestStep source : program.getSetupSourceTestSteps()) {
+            final int count = source.dataBeforeRestore.size();
+            if (count > 0) {
+                TestValuesTableFactory.awaitSourceEmitted(source.name, count)
+                        .get(SAVEPOINT_READY_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);
+            }
+        }
+    }
+}

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/IntervalJoinRestoreTest.java

@@ -36,6 +36,7 @@ public List<TableTestProgram> programs() {
         return Arrays.asList(
                 IntervalJoinTestPrograms.INTERVAL_JOIN_EVENT_TIME,
                 IntervalJoinTestPrograms.INTERVAL_JOIN_PROC_TIME,
-                IntervalJoinTestPrograms.INTERVAL_JOIN_NEGATIVE_INTERVAL);
+                IntervalJoinTestPrograms.INTERVAL_JOIN_NEGATIVE_INTERVAL,
+                IntervalJoinTestPrograms.INTERVAL_JOIN_EARLY_FIRE);
     }
 }

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/IntervalJoinTestPrograms.java

@@ -23,6 +23,8 @@
 import org.apache.flink.table.test.program.TableTestProgram;
 import org.apache.flink.types.Row;
 
+import java.util.Map;
+
 /** {@link TableTestProgram} definitions for testing {@link StreamExecIntervalJoin}. */
 public class IntervalJoinTestPrograms {
 
@@ -152,6 +154,125 @@ public class IntervalJoinTestPrograms {
                                     + " WHERE o.proc_time BETWEEN s.proc_time - INTERVAL '5' SECOND AND s.proc_time + INTERVAL '5' SECOND;")
                     .build();
 
+    static final Row[] EARLY_FIRE_ORDER_BEFORE_DATA = {
+        Row.of(1, "2020-04-15 08:00:01"), Row.of(9, "2020-04-15 08:00:06"),
+    };
+
+    static final Row[] EARLY_FIRE_SHIPMENT_BEFORE_DATA = {
+        Row.of(100, 9, "2020-04-15 08:00:06"),
+    };
+
+    static final Row[] EARLY_FIRE_ORDER_AFTER_DATA = {
+        Row.of(20, "2020-04-15 08:00:20"),
+    };
+
+    static final Row[] EARLY_FIRE_SHIPMENT_AFTER_DATA = {
+        Row.of(101, 1, "2020-04-15 08:00:03"), Row.of(102, 20, "2020-04-15 08:00:20"),
+    };
+
+    static final TableTestProgram INTERVAL_JOIN_EARLY_FIRE =
+            TableTestProgram.of(
+                            "interval-join-early-fire",
+                            "validates the EARLY_FIRE hint on an outer interval join: an unmatched"
+                                    + " left row is speculatively padded before the savepoint and"
+                                    + " retracted when its match arrives after restore")
+                    .setupTableSource(
+                            SourceTestStep.newBuilder("orders_t")
+                                    .addSchema(ORDERS_EVENT_TIME_SCHEMA)
+                                    .producedBeforeRestore(EARLY_FIRE_ORDER_BEFORE_DATA)
+                                    .producedAfterRestore(EARLY_FIRE_ORDER_AFTER_DATA)
+                                    .build())
+                    .setupTableSource(
+                            SourceTestStep.newBuilder("shipments_t")
+                                    .addSchema(SHIPMENTS_EVENT_TIME_SCHEMA)
+                                    .producedBeforeRestore(EARLY_FIRE_SHIPMENT_BEFORE_DATA)
+                                    .producedAfterRestore(EARLY_FIRE_SHIPMENT_AFTER_DATA)
+                                    .build())
+                    .setupTableSink(
+                            SinkTestStep.newBuilder("sink_t")
+                                    .addSchema(SINK_SCHEMA)
+                                    .consumedBeforeRestore(
+                                            "+I[1, 2020-04-15 08:00:01, null]",
+                                            "+I[9, 2020-04-15 08:00:06, 2020-04-15 08:00:06]")
+                                    .consumedAfterRestore(
+                                            "-U[1, 2020-04-15 08:00:01, null]",
+                                            "+U[1, 2020-04-15 08:00:01, 2020-04-15 08:00:03]",
+                                            "+I[20, 2020-04-15 08:00:20, 2020-04-15 08:00:20]")
+                                    .build())
+                    .runSql(
+                            "INSERT INTO sink_t SELECT /*+ EARLY_FIRE('delay'='2s') */\n"
+                                    + "     o.id AS order_id,\n"
+                                    + "     o.order_ts_str,\n"
+                                    + "     s.shipment_ts_str\n"
+                                    + " FROM orders_t o LEFT OUTER JOIN shipments_t s\n"
+                                    + " ON o.id = s.order_id\n"
+                                    + " AND o.order_ts BETWEEN s.shipment_ts - INTERVAL '5' SECOND AND s.shipment_ts + INTERVAL '5' SECOND;")
+                    .build();
+
+    // Selects the watermark-push-down source runtime, the only values-source runtime that reports
+    // how many rows it has emitted. The savepoint trigger gates on that count because the query
+    // below emits nothing before the savepoint.
+    static final Map<String, String> PER_RECORD_WATERMARK_SOURCE_OPTIONS =
+            Map.of(
+                    "disable-lookup", "true",
+                    "enable-watermark-push-down", "true",
+                    "scan.watermark.emit.strategy", "on-event");
+
+    // The unmatched left row whose speculative pad is scheduled on the wall clock.
+    static final Row[] PROC_TIME_EARLY_FIRE_ORDER_BEFORE_DATA = {
+        Row.of(1, "2020-04-15 08:00:01"),
+    };
+
+    // A shipment for an order that does not exist: a restore test source must produce at least one
+    // row, and this one neither joins the left row's key nor produces output on a left outer join.
+    static final Row[] PROC_TIME_EARLY_FIRE_SHIPMENT_BEFORE_DATA = {
+        Row.of(100, 99, "2020-04-15 08:00:02"),
+    };
+
+    // The 30s delay has to outlast the savepoint trigger so the schedule entry is still pending
+    // when the snapshot is taken. It also bounds how long a freshly generated savepoint makes the
+    // restored job wait for its early-fire timer; a savepoint older than the delay fires at once.
+    static final TableTestProgram INTERVAL_JOIN_PROC_TIME_EARLY_FIRE =
+            TableTestProgram.of(
+                            "interval-join-proc-time-early-fire",
+                            "validates the EARLY_FIRE hint driving a row-time interval join from"
+                                    + " processing time: the savepoint captures the pending"
+                                    + " early-fire schedule of an unmatched left row, and the pad"
+                                    + " is emitted only once that schedule is restored")
+                    .setupTableSource(
+                            SourceTestStep.newBuilder("orders_t")
+                                    .addSchema(ORDERS_EVENT_TIME_SCHEMA)
+                                    .addOptions(PER_RECORD_WATERMARK_SOURCE_OPTIONS)
+                                    .producedBeforeRestore(PROC_TIME_EARLY_FIRE_ORDER_BEFORE_DATA)
+                                    .build())
+                    .setupTableSource(
+                            SourceTestStep.newBuilder("shipments_t")
+                                    .addSchema(SHIPMENTS_EVENT_TIME_SCHEMA)
+                                    .addOptions(PER_RECORD_WATERMARK_SOURCE_OPTIONS)
+                                    .producedBeforeRestore(
+                                            PROC_TIME_EARLY_FIRE_SHIPMENT_BEFORE_DATA)
+                                    .build())
+                    .setupTableSink(
+                            SinkTestStep.newBuilder("sink_t")
+                                    .addSchema(SINK_SCHEMA)
+                                    // Nothing is emitted before the savepoint: the left row is
+                                    // still cached, its pad is still only scheduled, and the
+                                    // watermark has not closed its window.
+                                    .consumedBeforeRestore(new String[0])
+                                    // No data is ingested after restore, so this pad can only come
+                                    // from the restored schedule and cache.
+                                    .consumedAfterRestore("+I[1, 2020-04-15 08:00:01, null]")
+                                    .build())
+                    .runSql(
+                            "INSERT INTO sink_t SELECT /*+ EARLY_FIRE('delay'='30s', 'time-mode'='proctime') */\n"
+                                    + "     o.id AS order_id,\n"
+                                    + "     o.order_ts_str,\n"
+                                    + "     s.shipment_ts_str\n"
+                                    + " FROM orders_t o LEFT OUTER JOIN shipments_t s\n"
+                                    + " ON o.id = s.order_id\n"
+                                    + " AND o.order_ts BETWEEN s.shipment_ts - INTERVAL '5' SECOND AND s.shipment_ts + INTERVAL '5' SECOND;")
+                    .build();
+
     static final TableTestProgram INTERVAL_JOIN_NEGATIVE_INTERVAL =
             TableTestProgram.of(
                             "interval-join-negative-interval",

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

@@ -241,6 +241,38 @@ Calc(select=[a, b], changelogMode=[I,UA])
    +- Exchange(distribution=[hash[a]], changelogMode=[I])
       +- WatermarkAssigner(rowtime=[rowtime], watermark=[rowtime], changelogMode=[I])
          +- TableSourceScan(table=[[default_catalog, default_database, MyTable2, project=[a, b, rowtime], metadata=[]]], fields=[a, b, rowtime], changelogMode=[I])
+]]>
+    </Resource>
+  </TestCase>
+  <TestCase name="testEarlyFireProcTimeOnRowTimeJoin">
+    <Resource name="sql">
+      <![CDATA[SELECT /*+ EARLY_FIRE('delay'='5s', 'time-mode'='proctime') */ t1.a, t2.b
+FROM MyTable t1 LEFT OUTER JOIN MyTable2 t2 ON
+  t1.a = t2.a AND
+  t1.rowtime BETWEEN t2.rowtime - INTERVAL '10' SECOND AND t2.rowtime + INTERVAL '1' HOUR]]>
+    </Resource>
+    <Resource name="ast">
+      <![CDATA[
+LogicalProject(a=[$0], b=[$6])
++- LogicalJoin(condition=[AND(=($0, $5), >=($4, -($9, 10000:INTERVAL SECOND)), <=($4, +($9, 3600000:INTERVAL HOUR)))], joinType=[left], joinHints=[[[EARLY_FIRE inheritPath:[0] options:{delay=5s, time-mode=proctime}]]])
+   :- LogicalWatermarkAssigner(rowtime=[rowtime], watermark=[$4])
+   :  +- LogicalProject(a=[$0], b=[$1], c=[$2], proctime=[PROCTIME()], rowtime=[$3])
+   :     +- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+   +- LogicalWatermarkAssigner(rowtime=[rowtime], watermark=[$4])
+      +- LogicalProject(a=[$0], b=[$1], c=[$2], proctime=[PROCTIME()], rowtime=[$3])
+         +- LogicalTableScan(table=[[default_catalog, default_database, MyTable2]])
+]]>
+    </Resource>
+    <Resource name="optimized exec plan">
+      <![CDATA[
+Calc(select=[a, b])
++- IntervalJoin(joinType=[LeftOuterJoin], windowBounds=[isRowTime=true, leftLowerBound=-10000, leftUpperBound=3600000, leftTimeIndex=1, rightTimeIndex=2], where=[((a = a0) AND (rowtime >= (rowtime0 - 10000:INTERVAL SECOND)) AND (rowtime <= (rowtime0 + 3600000:INTERVAL HOUR)))], select=[a, rowtime, a0, b, rowtime0], earlyFireDelay=[5000], earlyFireTimeMode=[PROCTIME])
+   :- Exchange(distribution=[hash[a]])
+   :  +- WatermarkAssigner(rowtime=[rowtime], watermark=[rowtime])
+   :     +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, rowtime], metadata=[]]], fields=[a, rowtime])
+   +- Exchange(distribution=[hash[a]])
+      +- WatermarkAssigner(rowtime=[rowtime], watermark=[rowtime])
+         +- TableSourceScan(table=[[default_catalog, default_database, MyTable2, project=[a, b, rowtime], metadata=[]]], fields=[a, b, rowtime])
 ]]>
     </Resource>
   </TestCase>