[FLINK-38211][table-planner] Comma between inputs and add noUniqueKey

gustavodemorais · gustavodemorais · commit 96de70fd3c75 · 2025-08-19T10:39:24.000+02:00
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalMultiJoin.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalMultiJoin.java
@@ -164,9 +164,13 @@ public RelWriter explainTerms(final RelWriter pw) {
         }
 
         return pw.item("commonJoinKey", getCommonJoinKeyFieldNames())
-                .item("joinTypes", joinTypes)
-                .item("joinConditions", formatJoinConditionsWithFieldNames(pw))
+                .item(
+                        "joinTypes",
+                        joinTypes.stream()
+                                .map(JoinRelType::toString)
+                                .collect(Collectors.joining(", ")))
                 .item("inputUniqueKeys", formatInputUniqueKeysWithFieldNames())
+                .item("joinConditions", formatJoinConditionsWithFieldNames(pw))
                 .itemIf(
                         "joinFilter",
                         formatExpressionWithFieldNames(joinFilter, pw),
@@ -343,7 +347,7 @@ private String formatJoinConditionsWithFieldNames(final RelWriter pw) {
             }
         }
 
-        return String.join(" AND ", formattedConditions);
+        return String.join(", ", formattedConditions);
     }
 
     private String formatInputUniqueKeysWithFieldNames() {
@@ -367,13 +371,14 @@ private String formatInputUniqueKeysWithFieldNames() {
                     }
                 }
                 if (!uniqueKeyStrings.isEmpty()) {
-                    inputUniqueKeyStrings.add(
-                            "input#" + i + ": " + String.join(", ", uniqueKeyStrings));
+                    inputUniqueKeyStrings.add(String.join(", ", uniqueKeyStrings));
                 }
+            } else {
+                inputUniqueKeyStrings.add("noUniqueKey");
             }
         }
 
-        return String.join(" ", inputUniqueKeyStrings);
+        return String.join(", ", inputUniqueKeyStrings);
     }
 
     /**
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/MultiJoinTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/MultiJoinTest.xml
diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-multi-join_1/four-way-complex-updating-join-with-restore/plan/four-way-complex-updating-join-with-restore.json b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-multi-join_1/four-way-complex-updating-join-with-restore/plan/four-way-complex-updating-join-with-restore.json
@@ -447,7 +447,7 @@
       "priority" : 3
     } ],
     "outputType" : "ROW<`name` VARCHAR(2147483647), `user_id_0` VARCHAR(2147483647) NOT NULL, `cash` INT, `order_id` VARCHAR(2147483647), `user_id_1` VARCHAR(2147483647), `user_id_2` VARCHAR(2147483647), `payment_id` VARCHAR(2147483647) NOT NULL, `price` INT, `location` VARCHAR(2147483647), `user_id_3` VARCHAR(2147483647)>",
-    "description" : "MultiJoin(commonJoinKey=[user_id_0], joinTypes=[[INNER, LEFT, INNER, LEFT]], joinConditions=[true AND (user_id_0 = user_id_1) AND ((user_id_0 = user_id_2) AND ((cash >= price) OR (price < 0))) AND (user_id_2 = user_id_3)], inputUniqueKeys=[input#0: (user_id_0) input#1: (order_id) input#2: (payment_id)], joinFilter=[((user_id_0 = user_id_2) AND ((cash >= price) OR (price < 0)))], select=[name,user_id_0,cash,order_id,user_id_1,user_id_2,payment_id,price,location,user_id_3], outputRowType=[RecordType(VARCHAR(2147483647) name, VARCHAR(2147483647) user_id_0, INTEGER cash, VARCHAR(2147483647) order_id, VARCHAR(2147483647) user_id_1, VARCHAR(2147483647) user_id_2, VARCHAR(2147483647) payment_id, INTEGER price, VARCHAR(2147483647) location, VARCHAR(2147483647) user_id_3)])"
+    "description" : "MultiJoin(commonJoinKey=[user_id_0], joinTypes=[INNER, LEFT, INNER, LEFT], inputUniqueKeys=[(user_id_0), (order_id), (payment_id), noUniqueKey], joinConditions=[true, (user_id_0 = user_id_1), ((user_id_0 = user_id_2) AND ((cash >= price) OR (price < 0))), (user_id_2 = user_id_3)], joinFilter=[((user_id_0 = user_id_2) AND ((cash >= price) OR (price < 0)))], select=[name,user_id_0,cash,order_id,user_id_1,user_id_2,payment_id,price,location,user_id_3], outputRowType=[RecordType(VARCHAR(2147483647) name, VARCHAR(2147483647) user_id_0, INTEGER cash, VARCHAR(2147483647) order_id, VARCHAR(2147483647) user_id_1, VARCHAR(2147483647) user_id_2, VARCHAR(2147483647) payment_id, INTEGER price, VARCHAR(2147483647) location, VARCHAR(2147483647) user_id_3)])"
   }, {
     "id" : 30,
     "type" : "stream-exec-calc_1",
diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-multi-join_1/four-way-join-no-common-join-key-with-restore/plan/four-way-join-no-common-join-key-with-restore.json b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-multi-join_1/four-way-join-no-common-join-key-with-restore/plan/four-way-join-no-common-join-key-with-restore.json
@@ -312,7 +312,7 @@
       "priority" : 2
     } ],
     "outputType" : "ROW<`name` VARCHAR(2147483647), `user_id_0` VARCHAR(2147483647) NOT NULL, `order_id` VARCHAR(2147483647), `user_id_1` VARCHAR(2147483647), `user_id_2` VARCHAR(2147483647), `payment_id` VARCHAR(2147483647) NOT NULL>",
-    "description" : "MultiJoin(commonJoinKey=[user_id_0], joinTypes=[[INNER, LEFT, INNER]], joinConditions=[true AND (user_id_0 = user_id_1) AND (user_id_0 = user_id_2)], inputUniqueKeys=[input#0: (user_id_0) input#1: (order_id) input#2: (payment_id)], joinFilter=[(user_id_0 = user_id_2)], select=[name,user_id_0,order_id,user_id_1,user_id_2,payment_id], outputRowType=[RecordType(VARCHAR(2147483647) name, VARCHAR(2147483647) user_id_0, VARCHAR(2147483647) order_id, VARCHAR(2147483647) user_id_1, VARCHAR(2147483647) user_id_2, VARCHAR(2147483647) payment_id)])"
+    "description" : "MultiJoin(commonJoinKey=[user_id_0], joinTypes=[INNER, LEFT, INNER], inputUniqueKeys=[(user_id_0), (order_id), (payment_id)], joinConditions=[true, (user_id_0 = user_id_1), (user_id_0 = user_id_2)], joinFilter=[(user_id_0 = user_id_2)], select=[name,user_id_0,order_id,user_id_1,user_id_2,payment_id], outputRowType=[RecordType(VARCHAR(2147483647) name, VARCHAR(2147483647) user_id_0, VARCHAR(2147483647) order_id, VARCHAR(2147483647) user_id_1, VARCHAR(2147483647) user_id_2, VARCHAR(2147483647) payment_id)])"
   }, {
     "id" : 41,
     "type" : "stream-exec-calc_1",
@@ -462,7 +462,7 @@
       "priority" : 1
     } ],
     "outputType" : "ROW<`name` VARCHAR(2147483647), `user_id_0` VARCHAR(2147483647) NOT NULL, `order_id` VARCHAR(2147483647), `payment_id` VARCHAR(2147483647) NOT NULL, `location` VARCHAR(2147483647), `user_id_3` VARCHAR(2147483647)>",
-    "description" : "MultiJoin(commonJoinKey=[payment_id], joinTypes=[[INNER, LEFT]], joinConditions=[true AND (payment_id = user_id_3)], inputUniqueKeys=[], joinFilter=[true], select=[name,user_id_0,order_id,payment_id,location,user_id_3], outputRowType=[RecordType(VARCHAR(2147483647) name, VARCHAR(2147483647) user_id_0, VARCHAR(2147483647) order_id, VARCHAR(2147483647) payment_id, VARCHAR(2147483647) location, VARCHAR(2147483647) user_id_3)])"
+    "description" : "MultiJoin(commonJoinKey=[payment_id], joinTypes=[INNER, LEFT], inputUniqueKeys=[noUniqueKey, noUniqueKey], joinConditions=[true, (payment_id = user_id_3)], joinFilter=[true], select=[name,user_id_0,order_id,payment_id,location,user_id_3], outputRowType=[RecordType(VARCHAR(2147483647) name, VARCHAR(2147483647) user_id_0, VARCHAR(2147483647) order_id, VARCHAR(2147483647) payment_id, VARCHAR(2147483647) location, VARCHAR(2147483647) user_id_3)])"
   }, {
     "id" : 46,
     "type" : "stream-exec-calc_1",
diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-multi-join_1/three-way-inner-join-with-restore/plan/three-way-inner-join-with-restore.json b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-multi-join_1/three-way-inner-join-with-restore/plan/three-way-inner-join-with-restore.json
@@ -225,7 +225,7 @@
       "priority" : 2
     } ],
     "outputType" : "ROW<`user_id` VARCHAR(2147483647), `name` VARCHAR(2147483647), `user_id0` VARCHAR(2147483647), `order_id` VARCHAR(2147483647), `user_id1` VARCHAR(2147483647), `payment_id` VARCHAR(2147483647)>",
-    "description" : "MultiJoin(commonJoinKey=[user_id], joinTypes=[[INNER, INNER, INNER]], joinConditions=[true AND (user_id = user_id0) AND (user_id = user_id1)], inputUniqueKeys=[], joinFilter=[((user_id = user_id1) AND (user_id = user_id0))], select=[user_id,name,user_id0,order_id,user_id1,payment_id], outputRowType=[RecordType(VARCHAR(2147483647) user_id, VARCHAR(2147483647) name, VARCHAR(2147483647) user_id0, VARCHAR(2147483647) order_id, VARCHAR(2147483647) user_id1, VARCHAR(2147483647) payment_id)])"
+    "description" : "MultiJoin(commonJoinKey=[user_id], joinTypes=[INNER, INNER, INNER], inputUniqueKeys=[noUniqueKey, noUniqueKey, noUniqueKey], joinConditions=[true, (user_id = user_id0), (user_id = user_id1)], joinFilter=[((user_id = user_id1) AND (user_id = user_id0))], select=[user_id,name,user_id0,order_id,user_id1,payment_id], outputRowType=[RecordType(VARCHAR(2147483647) user_id, VARCHAR(2147483647) name, VARCHAR(2147483647) user_id0, VARCHAR(2147483647) order_id, VARCHAR(2147483647) user_id1, VARCHAR(2147483647) payment_id)])"
   }, {
     "id" : 8,
     "type" : "stream-exec-calc_1",
diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-multi-join_1/three-way-join-with-time-attributes-with-restore/plan/three-way-join-with-time-attributes-with-restore.json b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-multi-join_1/three-way-join-with-time-attributes-with-restore/plan/three-way-join-with-time-attributes-with-restore.json
@@ -357,7 +357,7 @@
       "priority" : 2
     } ],
     "outputType" : "ROW<`user_id_0` VARCHAR(2147483647) NOT NULL, `name` VARCHAR(2147483647), `order_id` VARCHAR(2147483647) NOT NULL, `user_id_1` VARCHAR(2147483647), `rowtime` TIMESTAMP(3), `price` INT, `user_id_2` VARCHAR(2147483647)>",
-    "description" : "MultiJoin(commonJoinKey=[user_id_0], joinTypes=[[INNER, INNER, INNER]], joinConditions=[true AND (user_id_0 = user_id_1) AND (user_id_0 = user_id_2)], inputUniqueKeys=[input#0: (user_id_0) input#1: (order_id)], joinFilter=[((user_id_0 = user_id_2) AND (user_id_0 = user_id_1))], select=[user_id_0,name,order_id,user_id_1,rowtime,price,user_id_2], outputRowType=[RecordType(VARCHAR(2147483647) user_id_0, VARCHAR(2147483647) name, VARCHAR(2147483647) order_id, VARCHAR(2147483647) user_id_1, TIMESTAMP(3) rowtime, INTEGER price, VARCHAR(2147483647) user_id_2)])"
+    "description" : "MultiJoin(commonJoinKey=[user_id_0], joinTypes=[INNER, INNER, INNER], inputUniqueKeys=[(user_id_0), (order_id), noUniqueKey], joinConditions=[true, (user_id_0 = user_id_1), (user_id_0 = user_id_2)], joinFilter=[((user_id_0 = user_id_2) AND (user_id_0 = user_id_1))], select=[user_id_0,name,order_id,user_id_1,rowtime,price,user_id_2], outputRowType=[RecordType(VARCHAR(2147483647) user_id_0, VARCHAR(2147483647) name, VARCHAR(2147483647) order_id, VARCHAR(2147483647) user_id_1, TIMESTAMP(3) rowtime, INTEGER price, VARCHAR(2147483647) user_id_2)])"
   }, {
     "id" : 57,
     "type" : "stream-exec-calc_1",
diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-multi-join_1/three-way-left-outer-join-with-restore/plan/three-way-left-outer-join-with-restore.json b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-multi-join_1/three-way-left-outer-join-with-restore/plan/three-way-left-outer-join-with-restore.json
@@ -225,7 +225,7 @@
       "priority" : 2
     } ],
     "outputType" : "ROW<`user_id` VARCHAR(2147483647), `name` VARCHAR(2147483647), `user_id0` VARCHAR(2147483647), `order_id` VARCHAR(2147483647), `user_id1` VARCHAR(2147483647), `payment_id` VARCHAR(2147483647)>",
-    "description" : "MultiJoin(commonJoinKey=[user_id], joinTypes=[[INNER, LEFT, LEFT]], joinConditions=[true AND (user_id = user_id0) AND (user_id = user_id1)], inputUniqueKeys=[], joinFilter=[true], select=[user_id,name,user_id0,order_id,user_id1,payment_id], outputRowType=[RecordType(VARCHAR(2147483647) user_id, VARCHAR(2147483647) name, VARCHAR(2147483647) user_id0, VARCHAR(2147483647) order_id, VARCHAR(2147483647) user_id1, VARCHAR(2147483647) payment_id)])"
+    "description" : "MultiJoin(commonJoinKey=[user_id], joinTypes=[INNER, LEFT, LEFT], inputUniqueKeys=[noUniqueKey, noUniqueKey, noUniqueKey], joinConditions=[true, (user_id = user_id0), (user_id = user_id1)], joinFilter=[true], select=[user_id,name,user_id0,order_id,user_id1,payment_id], outputRowType=[RecordType(VARCHAR(2147483647) user_id, VARCHAR(2147483647) name, VARCHAR(2147483647) user_id0, VARCHAR(2147483647) order_id, VARCHAR(2147483647) user_id1, VARCHAR(2147483647) payment_id)])"
   }, {
     "id" : 17,
     "type" : "stream-exec-calc_1",

Original file line number	Diff line number	Diff line change
`@@ -164,9 +164,13 @@ public RelWriter explainTerms(final RelWriter pw) {`
`164`	`164`	`}`
`165`	`165`
`166`	`166`	`return pw.item("commonJoinKey", getCommonJoinKeyFieldNames())`
`167`		`- .item("joinTypes", joinTypes)`
`168`		`- .item("joinConditions", formatJoinConditionsWithFieldNames(pw))`
	`167`	`+ .item(`
	`168`	`+ "joinTypes",`
	`169`	`+ joinTypes.stream()`
	`170`	`+ .map(JoinRelType::toString)`
	`171`	`+ .collect(Collectors.joining(", ")))`
`169`	`172`	`.item("inputUniqueKeys", formatInputUniqueKeysWithFieldNames())`
	`173`	`+ .item("joinConditions", formatJoinConditionsWithFieldNames(pw))`
`170`	`174`	`.itemIf(`
`171`	`175`	`"joinFilter",`
`172`	`176`	`formatExpressionWithFieldNames(joinFilter, pw),`
`@@ -343,7 +347,7 @@ private String formatJoinConditionsWithFieldNames(final RelWriter pw) {`
`343`	`347`	`}`
`344`	`348`	`}`
`345`	`349`
`346`		`- return String.join(" AND ", formattedConditions);`
	`350`	`+ return String.join(", ", formattedConditions);`
`347`	`351`	`}`
`348`	`352`
`349`	`353`	`private String formatInputUniqueKeysWithFieldNames() {`
`@@ -367,13 +371,14 @@ private String formatInputUniqueKeysWithFieldNames() {`
`367`	`371`	`}`
`368`	`372`	`}`
`369`	`373`	`if (!uniqueKeyStrings.isEmpty()) {`
`370`		`- inputUniqueKeyStrings.add(`
`371`		`- "input#" + i + ": " + String.join(", ", uniqueKeyStrings));`
	`374`	`+ inputUniqueKeyStrings.add(String.join(", ", uniqueKeyStrings));`
`372`	`375`	`}`
	`376`	`+ } else {`
	`377`	`+ inputUniqueKeyStrings.add("noUniqueKey");`
`373`	`378`	`}`
`374`	`379`	`}`
`375`	`380`
`376`		`- return String.join(" ", inputUniqueKeyStrings);`
	`381`	`+ return String.join(", ", inputUniqueKeyStrings);`
`377`	`382`	`}`
`378`	`383`
`379`	`384`	`/**`