[FLINK-38916][table-planner] MultiJoin produces incorrect results for OR join conditions when parallelism is greater than 1

This closes #27498.

Author: SteveStevenpoor

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/resources/org/apache/flink/table/planner/plan/stream/sql/MultiJoinTest.xml

@@ -2794,15 +2794,20 @@ LogicalProject(user_id=[$0], name=[$1], order_id=[$3], payment_id=[$6], location
     <Resource name="optimized rel plan">
       <![CDATA[
 Calc(select=[user_id, name, order_id, payment_id, location])
-+- MultiJoin(commonJoinKey=[user_id], joinTypes=[LEFT, LEFT, LEFT], inputUniqueKeys=[(user_id), (order_id), (payment_id), noUniqueKey], joinConditions=[=(user_id0, user_id), OR(=(user_id, user_id1), =(name, payment_id)), =(user_id1, user_id2)], select=[user_id,name,order_id,user_id0,payment_id,user_id1,location,user_id2], rowType=[RecordType(VARCHAR(2147483647) user_id, VARCHAR(2147483647) name, VARCHAR(2147483647) order_id, VARCHAR(2147483647) user_id0, VARCHAR(2147483647) payment_id, VARCHAR(2147483647) user_id1, VARCHAR(2147483647) location, VARCHAR(2147483647) user_id2)])
-   :- Exchange(distribution=[hash[user_id]])
-   :  +- ChangelogNormalize(key=[user_id])
-   :     +- Exchange(distribution=[hash[user_id]])
-   :        +- TableSourceScan(table=[[default_catalog, default_database, Users, project=[user_id, name], metadata=[]]], fields=[user_id, name])
-   :- Exchange(distribution=[hash[user_id]])
-   :  +- TableSourceScan(table=[[default_catalog, default_database, Orders, project=[order_id, user_id], metadata=[]]], fields=[order_id, user_id])
-   :- Exchange(distribution=[hash[user_id]])
-   :  +- TableSourceScan(table=[[default_catalog, default_database, Payments, project=[payment_id, user_id], metadata=[]]], fields=[payment_id, user_id])
++- MultiJoin(commonJoinKey=[user_id1], joinTypes=[LEFT], inputUniqueKeys=[noUniqueKey, noUniqueKey], joinConditions=[=(user_id1, user_id2)], select=[user_id,name,order_id,payment_id,user_id1,location,user_id2], rowType=[RecordType(VARCHAR(2147483647) user_id, VARCHAR(2147483647) name, VARCHAR(2147483647) order_id, VARCHAR(2147483647) payment_id, VARCHAR(2147483647) user_id1, VARCHAR(2147483647) location, VARCHAR(2147483647) user_id2)])
+   :- Exchange(distribution=[hash[user_id1]])
+   :  +- MultiJoin(commonJoinKey=[noCommonJoinKey], joinTypes=[LEFT], inputUniqueKeys=[noUniqueKey, (payment_id)], joinConditions=[OR(=(user_id, user_id1), =(name, payment_id))], select=[user_id,name,order_id,payment_id,user_id1], rowType=[RecordType(VARCHAR(2147483647) user_id, VARCHAR(2147483647) name, VARCHAR(2147483647) order_id, VARCHAR(2147483647) payment_id, VARCHAR(2147483647) user_id1)])
+   :     :- Exchange(distribution=[single])
+   :     :  +- Calc(select=[user_id, name, order_id])
+   :     :     +- MultiJoin(commonJoinKey=[user_id], joinTypes=[LEFT], inputUniqueKeys=[(user_id), (order_id)], joinConditions=[=(user_id0, user_id)], select=[user_id,name,order_id,user_id0], rowType=[RecordType(VARCHAR(2147483647) user_id, VARCHAR(2147483647) name, VARCHAR(2147483647) order_id, VARCHAR(2147483647) user_id0)])
+   :     :        :- Exchange(distribution=[hash[user_id]])
+   :     :        :  +- ChangelogNormalize(key=[user_id])
+   :     :        :     +- Exchange(distribution=[hash[user_id]])
+   :     :        :        +- TableSourceScan(table=[[default_catalog, default_database, Users, project=[user_id, name], metadata=[]]], fields=[user_id, name])
+   :     :        +- Exchange(distribution=[hash[user_id]])
+   :     :           +- TableSourceScan(table=[[default_catalog, default_database, Orders, project=[order_id, user_id], metadata=[]]], fields=[order_id, user_id])
+   :     +- Exchange(distribution=[single])
+   :        +- TableSourceScan(table=[[default_catalog, default_database, Payments, project=[payment_id, user_id], metadata=[]]], fields=[payment_id, user_id])
    +- Exchange(distribution=[hash[user_id]])
       +- TableSourceScan(table=[[default_catalog, default_database, Shipments]], fields=[location, user_id])
 ]]>