[FLINK-39420] Reject temporal joins in batch mode with a clear error

Batch mode only supports lookup joins. General temporal joins
(event-time or processing-time on non-lookup sources) previously
left an unresolved LogicalCorrelate in the plan, causing a confusing
"unexpected correlate variable" error from FlinkDecorrelateProgram.

Add RejectTemporalJoinInBatchRule to the batch EXPAND_PLAN_RULES.
The lookup join rules fire first and rewrite valid lookup joins;
any remaining Correlate+Snapshot pattern is unconditionally rejected
with an actionable message pointing users to lookup joins or
streaming mode.

Generated-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: jnh5y

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/RejectTemporalJoinInBatchRule.java

@@ -0,0 +1,82 @@
+/*
+ * 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.rules.logical;
+
+import org.apache.flink.table.api.TableException;
+
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelOptRuleOperand;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.logical.LogicalCorrelate;
+import org.apache.calcite.rel.logical.LogicalFilter;
+import org.apache.calcite.rel.logical.LogicalSnapshot;
+
+/**
+ * Rules that reject temporal table joins ({@code FOR SYSTEM_TIME AS OF}) in batch mode with a clear
+ * error message.
+ *
+ * <p>In the batch {@code EXPAND_PLAN_RULES}, lookup join rules run first and rewrite valid lookup
+ * joins (processing-time + {@code LookupTableSource}) into {@code TemporalJoin} nodes. Any
+ * remaining {@link LogicalCorrelate} + {@link LogicalSnapshot} pattern therefore represents an
+ * unsupported temporal join. These rules catch it and throw a {@link TableException} rather than
+ * letting the correlate survive into {@code FlinkDecorrelateProgram}, where it would cause a
+ * confusing "unexpected correlate variable" internal error.
+ */
+public class RejectTemporalJoinInBatchRule extends RelOptRule {
+
+    private static final String MESSAGE =
+            "Temporal joins (FOR SYSTEM_TIME AS OF) on regular tables are not supported in "
+                    + "batch mode. Use a lookup join or switch to streaming mode.";
+
+    /**
+     * Matches temporal joins where the right side of the Correlate is a Filter wrapping a Snapshot
+     * (non-trivial join condition).
+     */
+    public static final RelOptRule WITH_FILTER =
+            new RejectTemporalJoinInBatchRule(
+                    operand(
+                            LogicalCorrelate.class,
+                            operand(RelNode.class, any()),
+                            operand(
+                                    LogicalFilter.class,
+                                    operand(LogicalSnapshot.class, operand(RelNode.class, any())))),
+                    "RejectTemporalJoinInBatchRule_WithFilter");
+
+    /**
+     * Matches temporal joins where the right side of the Correlate is a Snapshot directly (trivial
+     * join condition).
+     */
+    public static final RelOptRule WITHOUT_FILTER =
+            new RejectTemporalJoinInBatchRule(
+                    operand(
+                            LogicalCorrelate.class,
+                            operand(RelNode.class, any()),
+                            operand(LogicalSnapshot.class, operand(RelNode.class, any()))),
+                    "RejectTemporalJoinInBatchRule_WithoutFilter");
+
+    private RejectTemporalJoinInBatchRule(RelOptRuleOperand operand, String description) {
+        super(operand, description);
+    }
+
+    @Override
+    public void onMatch(RelOptRuleCall call) {
+        throw new TableException(MESSAGE);
+    }
+}

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkBatchRuleSets.scala

@@ -50,10 +50,16 @@ object FlinkBatchRuleSets {
   /**
    * Expand plan by replacing references to tables into a proper plan sub trees. Those rules can
    * create new plan nodes.
+   *
+   * The lookup join rules run first and rewrite supported temporal joins. The rejection rules then
+   * catch any remaining temporal joins (unsupported in batch) with a clear error message.
    */
   val EXPAND_PLAN_RULES: RuleSet = RuleSets.ofList(
     LogicalCorrelateToJoinFromTemporalTableRule.LOOKUP_JOIN_WITH_FILTER,
-    LogicalCorrelateToJoinFromTemporalTableRule.LOOKUP_JOIN_WITHOUT_FILTER)
+    LogicalCorrelateToJoinFromTemporalTableRule.LOOKUP_JOIN_WITHOUT_FILTER,
+    RejectTemporalJoinInBatchRule.WITH_FILTER,
+    RejectTemporalJoinInBatchRule.WITHOUT_FILTER
+  )
 
   val POST_EXPAND_CLEAN_UP_RULES: RuleSet = RuleSets.ofList(EnumerableToLogicalTableScan.INSTANCE)
 

flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/join/TemporalJoinTest.scala

@@ -24,9 +24,11 @@ import org.assertj.core.api.Assertions.assertThatExceptionOfType
 import org.junit.jupiter.api.{BeforeEach, Test}
 
 /**
- * Test temporal join in batch mode.
+ * Tests that temporal joins on non-lookup tables are rejected in batch mode with a clear error.
  *
- * <p> Flink only supports lookup join in batch mode, the others Temporal join is not supported yet.
+ * <p>Batch mode only supports lookup joins. General temporal joins (event-time or processing-time
+ * on non-lookup sources) are caught by {@link RejectTemporalJoinInBatchRule} during the
+ * TEMPORAL_JOIN_REWRITE phase.
  */
 class TemporalJoinTest extends TableTestBase {
 
@@ -105,6 +107,10 @@ class TemporalJoinTest extends TableTestBase {
         "GROUP BY currency ")
   }
 
+  private val expectedMessage =
+    "Temporal joins (FOR SYSTEM_TIME AS OF) on regular tables are not supported in " +
+      "batch mode. Use a lookup join or switch to streaming mode."
+
   @Test
   def testSimpleJoin(): Unit = {
     val sqlQuery = "SELECT " +
@@ -115,6 +121,7 @@ class TemporalJoinTest extends TableTestBase {
 
     assertThatExceptionOfType(classOf[TableException])
       .isThrownBy(() => util.verifyExecPlan(sqlQuery))
+      .withMessage(expectedMessage)
   }
 
   @Test
@@ -128,6 +135,7 @@ class TemporalJoinTest extends TableTestBase {
 
     assertThatExceptionOfType(classOf[TableException])
       .isThrownBy(() => util.verifyExecPlan(sqlQuery))
+      .withMessage(expectedMessage)
   }
 
   @Test
@@ -141,11 +149,11 @@ class TemporalJoinTest extends TableTestBase {
 
     assertThatExceptionOfType(classOf[TableException])
       .isThrownBy(() => util.verifyExecPlan(sqlQuery))
+      .withMessage(expectedMessage)
   }
 
   @Test
   def testSimpleViewProcTimeJoin(): Unit = {
-
     val sqlQuery = "SELECT " +
       "o_amount * rate as rate " +
       "FROM Orders AS o JOIN " +
@@ -155,5 +163,6 @@ class TemporalJoinTest extends TableTestBase {
 
     assertThatExceptionOfType(classOf[TableException])
       .isThrownBy(() => util.verifyExecPlan(sqlQuery))
+      .withMessage(expectedMessage)
   }
 }