[FLINK-38916][FLINK-38417][FLINK-38445][FLINK-39015][table] Backport MultiJoin fixes to 2.1

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/MultiJoinUtil.java

@@ -71,9 +71,13 @@ private static void extractEqualityConditions(
         final SqlKind kind = call.getOperator().getKind();
 
         if (kind != SqlKind.EQUALS) {
-            for (final RexNode operand : call.getOperands()) {
-                extractEqualityConditions(
-                        operand, inputOffsets, inputFieldCounts, joinAttributeMap);
+            // Only conjunctions (AND) can contain equality conditions that are valid for multijoin.
+            // All other condition types are deferred to the postJoinFilter.
+            if (kind == SqlKind.AND) {
+                for (final RexNode operand : call.getOperands()) {
+                    extractEqualityConditions(
+                            operand, inputOffsets, inputFieldCounts, joinAttributeMap);
+                }
             }
             return;
         }

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

@@ -51,7 +51,9 @@ public List<TableTestProgram> programs() {
                 MultiJoinTestPrograms.MULTI_JOIN_LEFT_OUTER_WITH_NULL_KEYS,
                 MultiJoinTestPrograms.MULTI_JOIN_NULL_SAFE_JOIN_WITH_NULL_KEYS,
                 MultiJoinTestPrograms.MULTI_JOIN_MIXED_CHANGELOG_MODES,
-                MultiJoinTestPrograms
-                        .MULTI_JOIN_WITH_TIME_ATTRIBUTES_IN_CONDITIONS_MATERIALIZATION);
+                MultiJoinTestPrograms.MULTI_JOIN_WITH_TIME_ATTRIBUTES_IN_CONDITIONS_MATERIALIZATION,
+                MultiJoinTestPrograms.MULTI_JOIN_TWO_WAY_INNER_JOIN_WITH_WHERE_IN,
+                MultiJoinTestPrograms.MULTI_JOIN_THREE_WAY_INNER_JOIN_MULTI_KEY_TYPES,
+                MultiJoinTestPrograms.MULTI_JOIN_FOUR_WAY_MIXED_JOIN_MULTI_KEY_TYPES_SHUFFLED);
     }
 }

flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/JoinITCase.scala

@@ -20,6 +20,7 @@ package org.apache.flink.table.planner.runtime.stream.sql
 import org.apache.flink.table.api._
 import org.apache.flink.table.api.bridge.scala._
 import org.apache.flink.table.api.config.ExecutionConfigOptions
+import org.apache.flink.table.api.config.OptimizerConfigOptions
 import org.apache.flink.table.planner.expressions.utils.FuncWithOpen
 import org.apache.flink.table.planner.factories.TestValuesTableFactory
 import org.apache.flink.table.planner.runtime.utils._
@@ -623,6 +624,22 @@ class JoinITCase(miniBatch: MiniBatchMode, state: StateBackendMode, enableAsyncS
     assertThat(sink.getRetractResults.sorted).isEqualTo(expected.sorted)
   }
 
+  @TestTemplate
+  def testInnerMultiJoinWithEqualPk(): Unit = {
+    tEnv.getConfig.getConfiguration
+      .setString(OptimizerConfigOptions.TABLE_OPTIMIZER_MULTI_JOIN_ENABLED.key(), "true")
+    val query1 = "SELECT SUM(a2) AS a2, a1 FROM A group by a1"
+    val query2 = "SELECT SUM(b2) AS b2, b1 FROM B group by b1"
+    val query = s"SELECT a1, b1 FROM ($query1) JOIN ($query2) ON a1 = b1"
+
+    val sink = new TestingRetractSink
+    tEnv.sqlQuery(query).toRetractStream[Row].addSink(sink).setParallelism(1)
+    env.execute()
+
+    val expected = Seq("1,1", "2,2", "3,3")
+    assertThat(sink.getRetractResults.sorted).isEqualTo(expected.sorted)
+  }
+
   @TestTemplate
   def testInnerJoinWithPk(): Unit = {
     val query1 = "SELECT SUM(a2) AS a2, a1 FROM A group by a1"

flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/StreamingMultiJoinOperator.java

@@ -19,6 +19,7 @@
 package org.apache.flink.table.runtime.operators.join.stream;
 
 import org.apache.flink.api.common.functions.DefaultOpenContext;
+import org.apache.flink.runtime.state.heap.HeapKeyedStateBackend;
 import org.apache.flink.streaming.api.operators.AbstractInput;
 import org.apache.flink.streaming.api.operators.AbstractStreamOperatorV2;
 import org.apache.flink.streaming.api.operators.Input;
@@ -609,6 +610,10 @@ private Integer processRecords(
         }
     }
 
+    private boolean isHeapBackend() {
+        return getKeyedStateBackend() instanceof HeapKeyedStateBackend;
+    }
+
     private static RowData newJoinedRowData(int depth, RowData joinedRowData, RowData record) {
         RowData newJoinedRowData;
         if (depth == 0) {
@@ -808,6 +813,11 @@ private void initializeStateHandlers() {
                     "Keyed state store not found when initializing keyed state store handlers.");
         }
 
+        boolean prohibitReuseRow = isHeapBackend();
+        if (prohibitReuseRow) {
+            this.keyExtractor.requiresKeyDeepCopy();
+        }
+
         this.stateHandlers = new ArrayList<>(inputSpecs.size());
         for (int i = 0; i < inputSpecs.size(); i++) {
             MultiJoinStateView stateView;