[FLINK-40172][table-runtime] Support processing-time early fire on a row-time interval join

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

@@ -391,7 +391,8 @@ private TwoInputTransformation<RowData, RowData, RowData> createProcTimeJoin(
                         minCleanUpIntervalMillis,
                         leftTypeInfo,
                         rightTypeInfo,
-                        joinFunction);
+                        joinFunction,
+                        earlyFireDelay == null ? -1L : earlyFireDelay);
         // TODO: add async version procJoinFunc to use AsyncKeyedCoProcessOperator
         return ExecNodeUtil.createTwoInputTransformation(
                 leftInputTransform,
@@ -428,7 +429,12 @@ private TwoInputTransformation<RowData, RowData, RowData> createRowTimeJoin(
                         rightTypeInfo,
                         joinFunction,
                         windowBounds.getLeftTimeIdx(),
-                        windowBounds.getRightTimeIdx());
+                        windowBounds.getRightTimeIdx(),
+                        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/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>

flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/EmitAwareCollector.java

@@ -19,17 +19,29 @@
 package org.apache.flink.table.runtime.operators.join.interval;
 
 import org.apache.flink.table.data.RowData;
+import org.apache.flink.types.RowKind;
 import org.apache.flink.util.Collector;
 
 /**
  * Collector to wrap a [[org.apache.flink.table.dataformat.RowData]] and to track whether a row has
  * been emitted by the inner collector.
+ *
+ * <p>The collector can be armed with a correction before a single matched row is collected. When
+ * armed, the next collected row is treated as the corrected result of a previously emitted
+ * speculative outer-join pad: the pending pad is emitted first stamped {@link
+ * RowKind#UPDATE_BEFORE}, then the matched row is stamped {@link RowKind#UPDATE_AFTER}. This turns
+ * the join function's single {@code INSERT} emit into the {@code -U}/{@code +U} pair without the
+ * join function knowing about changelogs. When not armed, collected rows are forwarded with their
+ * existing {@link RowKind}.
  */
 class EmitAwareCollector implements Collector<RowData> {
 
     private boolean emitted = false;
     private Collector<RowData> innerCollector;
 
+    // The pad to retract before the next matched row, or null when no correction is armed.
+    private RowData pendingRetraction;
+
     void reset() {
         emitted = false;
     }
@@ -42,10 +54,35 @@ void setInnerCollector(Collector<RowData> innerCollector) {
         this.innerCollector = innerCollector;
     }
 
+    /**
+     * Arms the collector so the next collected matched row is corrected into a {@code -U}/{@code
+     * +U} pair against the given padded row.
+     */
+    void armRetraction(RowData retractionPad) {
+        retractionPad.setRowKind(RowKind.UPDATE_BEFORE);
+        this.pendingRetraction = retractionPad;
+    }
+
+    /** Clears an armed correction that was never consumed (the join condition did not match). */
+    void disarm() {
+        this.pendingRetraction = null;
+    }
+
     @Override
     public void collect(RowData record) {
         emitted = true;
-        innerCollector.collect(record);
+        if (pendingRetraction != null) {
+            innerCollector.collect(pendingRetraction);
+            pendingRetraction = null;
+            record.setRowKind(RowKind.UPDATE_AFTER);
+            innerCollector.collect(record);
+        } else {
+            // The matched row reuses a single instance whose kind may have been left as
+            // UPDATE_AFTER by a previous correction; force INSERT so a later ordinary match is not
+            // mis-emitted as an update.
+            record.setRowKind(RowKind.INSERT);
+            innerCollector.collect(record);
+        }
     }
 
     @Override

flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/ProcTimeIntervalJoin.java

@@ -34,7 +34,8 @@ public ProcTimeIntervalJoin(
             long minCleanUpInterval,
             InternalTypeInfo<RowData> leftType,
             InternalTypeInfo<RowData> rightType,
-            IntervalJoinFunction genJoinFunc) {
+            IntervalJoinFunction genJoinFunc,
+            long earlyFireDelay) {
         super(
                 joinType,
                 leftLowerBound,
@@ -43,7 +44,10 @@ public ProcTimeIntervalJoin(
                 minCleanUpInterval,
                 leftType,
                 rightType,
-                genJoinFunc);
+                genJoinFunc,
+                earlyFireDelay,
+                // A proctime join's early fire shares the cleanup domain; never cross-domain.
+                false);
     }
 
     @Override

flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/RowTimeIntervalJoin.java

@@ -40,7 +40,9 @@ public RowTimeIntervalJoin(
             InternalTypeInfo<RowData> rightType,
             IntervalJoinFunction joinFunc,
             int leftTimeIdx,
-            int rightTimeIdx) {
+            int rightTimeIdx,
+            long earlyFireDelay,
+            boolean earlyFireCrossDomain) {
         super(
                 joinType,
                 leftLowerBound,
@@ -49,7 +51,9 @@ public RowTimeIntervalJoin(
                 minCleanUpInterval,
                 leftType,
                 rightType,
-                joinFunc);
+                joinFunc,
+                earlyFireDelay,
+                earlyFireCrossDomain);
         this.leftTimeIdx = leftTimeIdx;
         this.rightTimeIdx = rightTimeIdx;
     }