[FLINK-38220][table-planner] Add Values and TableFunctionScan as sources (#26887)

Author: gustavodemorais

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

@@ -58,6 +58,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
@@ -159,9 +160,9 @@ public StreamExecMultiJoin(
         validateInputs(inputProperties, joinTypes, joinConditions, inputUniqueKeys);
         this.joinTypes = checkNotNull(joinTypes);
         this.joinConditions = checkNotNull(joinConditions);
-        this.multiJoinCondition = multiJoinCondition;
-        this.joinAttributeMap = checkNotNull(joinAttributeMap);
         this.inputUniqueKeys = checkNotNull(inputUniqueKeys);
+        this.multiJoinCondition = multiJoinCondition;
+        this.joinAttributeMap = Objects.requireNonNullElseGet(joinAttributeMap, Map::of);
         this.stateMetadataList = stateMetadataList;
     }
 

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

@@ -34,9 +34,11 @@
 import org.apache.calcite.rel.core.Join;
 import org.apache.calcite.rel.core.JoinInfo;
 import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.core.TableFunctionScan;
+import org.apache.calcite.rel.core.TableScan;
+import org.apache.calcite.rel.core.Values;
 import org.apache.calcite.rel.logical.LogicalJoin;
 import org.apache.calcite.rel.logical.LogicalSnapshot;
-import org.apache.calcite.rel.logical.LogicalTableScan;
 import org.apache.calcite.rel.metadata.RelColumnOrigin;
 import org.apache.calcite.rel.metadata.RelMetadataQuery;
 import org.apache.calcite.rel.rules.CoreRules;
@@ -546,7 +548,10 @@ private Tuple2<RelNode, Integer> getTargetInputAndIdx(int inputRefIndex, List<Re
 
         assert targetInput != null;
 
-        if (targetInput instanceof LogicalTableScan) {
+        if (targetInput instanceof TableScan
+                || targetInput instanceof Values
+                || targetInput instanceof TableFunctionScan
+                || targetInput.getInputs().isEmpty()) {
             return new Tuple2<>(targetInput, idxInTargetInput);
         } else {
             return getTargetInputAndIdx(idxInTargetInput, targetInput.getInputs());

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

@@ -37,7 +37,9 @@ public List<TableTestProgram> programs() {
                 MultiJoinTestPrograms.MULTI_JOIN_WITH_TIME_ATTRIBUTES_MATERIALIZATION,
                 MultiJoinTestPrograms.MULTI_JOIN_THREE_WAY_INNER_JOIN_NO_JOIN_KEY,
                 MultiJoinTestPrograms.MULTI_JOIN_FOUR_WAY_NO_COMMON_JOIN_KEY,
+                MultiJoinTestPrograms.MULTI_JOIN_THREE_WAY_LEFT_OUTER_JOIN_WITH_CTE,
                 MultiJoinTestPrograms.MULTI_JOIN_MIXED_CHANGELOG_MODES,
-                MultiJoinTestPrograms.MULTI_JOIN_THREE_WAY_LEFT_OUTER_JOIN_WITH_CTE);
+                MultiJoinTestPrograms.MULTI_JOIN_LEFT_OUTER_WITH_NULL_KEYS,
+                MultiJoinTestPrograms.MULTI_JOIN_NULL_SAFE_JOIN_WITH_NULL_KEYS);
     }
 }

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

@@ -1031,4 +1031,75 @@ public class MultiJoinTestPrograms {
                                     + "LEFT JOIN Payments p ON ue.user_id = p.user_id "
                                     + "GROUP BY ue.user_id, us.name")
                     .build();
+
+    public static final TableTestProgram MULTI_JOIN_LEFT_OUTER_WITH_NULL_KEYS =
+            TableTestProgram.of(
+                            "three-way-left-outer-with-null-keys",
+                            "left outer join with NULL keys on multiple inputs")
+                    .setupConfig(OptimizerConfigOptions.TABLE_OPTIMIZER_MULTI_JOIN_ENABLED, true)
+                    .setupSql(
+                            "CREATE VIEW UsersNulls AS SELECT * FROM (VALUES "
+                                    + "('1','Gus'),"
+                                    + "(CAST(NULL AS STRING), 'NullUser')"
+                                    + ") AS T(user_id, name)")
+                    .setupSql(
+                            "CREATE VIEW OrdersNulls AS SELECT * FROM (VALUES "
+                                    + "('1','order1'),"
+                                    + "(CAST(NULL AS STRING), 'nullOrder')"
+                                    + ") AS T(user_id, order_id)")
+                    .setupSql(
+                            "CREATE VIEW PaymentsNulls AS SELECT * FROM (VALUES "
+                                    + "('1','payment1'),"
+                                    + "('1','payment3'),"
+                                    + "(CAST(NULL AS STRING), 'paymentNull')"
+                                    + ") AS T(user_id, payment_id)")
+                    .setupTableSink(
+                            SinkTestStep.newBuilder("sink")
+                                    .addSchema(
+                                            "user_id STRING",
+                                            "name STRING",
+                                            "order_id STRING",
+                                            "payment_id STRING")
+                                    .consumedValues(
+                                            "+I[1, Gus, order1, payment1]",
+                                            "+I[1, Gus, order1, payment3]",
+                                            "+I[null, NullUser, null, null]")
+                                    .testMaterializedData()
+                                    .build())
+                    .runSql(
+                            "INSERT INTO sink "
+                                    + "SELECT u.user_id, u.name, o.order_id, p.payment_id "
+                                    + "FROM UsersNulls u "
+                                    + "LEFT JOIN OrdersNulls o ON u.user_id = o.user_id "
+                                    + "LEFT JOIN PaymentsNulls p ON u.user_id = p.user_id")
+                    .build();
+
+    public static final TableTestProgram MULTI_JOIN_NULL_SAFE_JOIN_WITH_NULL_KEYS =
+            TableTestProgram.of(
+                            "null-safe-join-with-null-keys",
+                            "join with IS NOT DISTINCT FROM to match NULL keys")
+                    .setupConfig(OptimizerConfigOptions.TABLE_OPTIMIZER_MULTI_JOIN_ENABLED, true)
+                    .setupSql(
+                            "CREATE VIEW UsersNullSafe AS SELECT * FROM (VALUES "
+                                    + "('1','Gus'),"
+                                    + "(CAST(NULL AS STRING), 'NullUser')"
+                                    + ") AS T(user_id, name)")
+                    .setupSql(
+                            "CREATE VIEW OrdersNullSafe AS SELECT * FROM (VALUES "
+                                    + "('1','order1'),"
+                                    + "(CAST(NULL AS STRING), 'nullOrder')"
+                                    + ") AS T(user_id, order_id)")
+                    .setupTableSink(
+                            SinkTestStep.newBuilder("sink")
+                                    .addSchema("user_id STRING", "name STRING", "order_id STRING")
+                                    .consumedValues(
+                                            "+I[1, Gus, order1]", "+I[null, NullUser, nullOrder]")
+                                    .testMaterializedData()
+                                    .build())
+                    .runSql(
+                            "INSERT INTO sink "
+                                    + "SELECT u.user_id, u.name, o.order_id "
+                                    + "FROM UsersNullSafe u "
+                                    + "INNER JOIN OrdersNullSafe o ON u.user_id IS NOT DISTINCT FROM o.user_id")
+                    .build();
 }