twalthr commented on code in PR #26687:
URL: https://github.com/apache/flink/pull/26687#discussion_r2152389094
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/JoinToMultiJoinForReorderRule.java:
##########
@@ -99,24 +99,24 @@
* @see CoreRules#JOIN_TO_MULTI_JOIN
*/
@Value.Enclosing
-public class FlinkJoinToMultiJoinRule extends
RelRule<FlinkJoinToMultiJoinRule.Config>
+public class JoinToMultiJoinForReorderRule extends
RelRule<JoinToMultiJoinForReorderRule.Config>
implements TransformationRule {
- public static final FlinkJoinToMultiJoinRule INSTANCE =
- FlinkJoinToMultiJoinRule.Config.DEFAULT.toRule();
+ public static final JoinToMultiJoinForReorderRule INSTANCE =
+ JoinToMultiJoinForReorderRule.Config.DEFAULT.toRule();
- /** Creates a JoinToMultiJoinRule. */
- public FlinkJoinToMultiJoinRule(Config config) {
+ /** Creates a JoinToMultiJoinForReorderRule. */
+ public JoinToMultiJoinForReorderRule(Config config) {
super(config);
}
@Deprecated // to be removed before 2.0
- public FlinkJoinToMultiJoinRule(Class<? extends Join> clazz) {
+ public JoinToMultiJoinForReorderRule(Class<? extends Join> clazz) {
Review Comment:
if this constructor is not used in our code base, feel free to remove unused
code.
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/logical/FlinkLogicalMultiJoin.java:
##########
@@ -0,0 +1,218 @@
+/*
+ * 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.nodes.logical;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.planner.plan.nodes.FlinkConventions;
+
+import org.apache.calcite.plan.Convention;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelOptCost;
+import org.apache.calcite.plan.RelOptPlanner;
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.AbstractRelNode;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.convert.ConverterRule;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.hint.RelHint;
+import org.apache.calcite.rel.metadata.RelMetadataQuery;
+import org.apache.calcite.rel.rules.MultiJoin;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexNode;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * A relational expression that represents a multi-input join in Flink.
+ *
+ * <p>This node is used to represent a join of multiple inputs, which can be
seen as a
+ * generalization of the binary {@link FlinkLogicalJoin}. It will be
translated into a {@link
+ *
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalMultiJoin}
by the
Review Comment:
very nit: it is possible to also import JavaDoc classes, makes the code a
bit more readable
```suggestion
* StreamPhysicalMultiJoin} by the
```
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/JoinToMultiJoinRule.java:
##########
@@ -0,0 +1,683 @@
+/*
+ * 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.RelOptRuleCall;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.rel.RelNode;
+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.logical.LogicalJoin;
+import org.apache.calcite.rel.rules.CoreRules;
+import org.apache.calcite.rel.rules.FilterMultiJoinMergeRule;
+import org.apache.calcite.rel.rules.MultiJoin;
+import org.apache.calcite.rel.rules.ProjectMultiJoinMergeRule;
+import org.apache.calcite.rel.rules.TransformationRule;
+import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.rex.RexVisitorImpl;
+import org.apache.calcite.tools.RelBuilderFactory;
+import org.apache.calcite.util.ImmutableBitSet;
+import org.apache.calcite.util.ImmutableIntList;
+import org.apache.calcite.util.Pair;
+import org.immutables.value.Value;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Flink Planner rule to flatten a tree of {@link Join}s into a single {@link
MultiJoin} with N
+ * inputs.
+ *
+ * <p>This rule is copied and adjusted from {@link
org.apache.calcite.rel.rules.JoinToMultiJoinRule}
+ * and {@link JoinToMultiJoinForReorderRule}. In this rule, we support a
broder sef of left and
Review Comment:
```suggestion
* and {@link JoinToMultiJoinForReorderRule}. In this rule, we support a
broder sef of left and
```
fix typos and make JavaDoc distinguishable from
`JoinToMultiJoinForReorderRule`
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/MultiJoinRestoreTest.java:
##########
@@ -0,0 +1,41 @@
+/*
+ * 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.nodes.exec.stream;
+
+import
org.apache.flink.table.planner.plan.nodes.exec.testutils.RestoreTestBase;
+import org.apache.flink.table.test.program.TableTestProgram;
+
+import java.util.Arrays;
+import java.util.List;
+
+/** Semantic tests for multi-join. */
Review Comment:
```suggestion
/** Restore tests for {@link StreamExecMultiJoin}. */
```
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ProcessTableFunctionTest.java:
##########
@@ -71,7 +72,12 @@ public class ProcessTableFunctionTest extends TableTestBase {
@BeforeEach
void setup() {
- util = streamTestUtil(TableConfig.getDefault());
+ util =
Review Comment:
undo?
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/MultiJoinTestPrograms.java:
##########
@@ -0,0 +1,461 @@
+/*
+ * 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.nodes.exec.stream;
+
+import org.apache.flink.table.api.config.OptimizerConfigOptions;
+import org.apache.flink.table.test.program.SinkTestStep;
+import org.apache.flink.table.test.program.SourceTestStep;
+import org.apache.flink.table.test.program.TableTestProgram;
+import org.apache.flink.types.Row;
+import org.apache.flink.types.RowKind;
+
+import static
org.apache.flink.table.planner.plan.nodes.exec.stream.MultiJoinTestUtils.ORDERS_SOURCE;
+import static
org.apache.flink.table.planner.plan.nodes.exec.stream.MultiJoinTestUtils.PAYMENTS_SOURCE;
+import static
org.apache.flink.table.planner.plan.nodes.exec.stream.MultiJoinTestUtils.USERS_SOURCE;
+
+/** {@link TableTestProgram} definitions for testing Multi-Join. */
+public class MultiJoinTestPrograms {
+ public static final TableTestProgram MULTI_JOIN_THREE_WAY_LEFT_OUTER_JOIN =
+ TableTestProgram.of("three-way-left-outer-join", "three way left
outer join")
+
.setupConfig(OptimizerConfigOptions.TABLE_OPTIMIZER_MULTI_JOIN_ENABLED, true)
+ .setupTableSource(USERS_SOURCE)
+ .setupTableSource(ORDERS_SOURCE)
+ .setupTableSource(PAYMENTS_SOURCE)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(
+ "user_id STRING",
+ "name STRING",
+ "order_id STRING",
+ "payment_id STRING")
+ .consumedValues(
+ "+I[1, Gus, order1, payment1]",
+ "+I[2, Bob, order2, payment2]",
+ "+I[2, Bob, order3, payment2]",
+ "+I[1, Gus, order1, payment3]",
+ "+I[3, Alice, null, null]")
+ .testMaterializedData()
+ .build())
+ .runSql(
+ "INSERT INTO sink "
+ + "SELECT u.user_id, u.name, o.order_id,
p.payment_id "
+ + "FROM Users u "
+ + "LEFT JOIN Orders o ON u.user_id =
o.user_id "
+ + "LEFT JOIN Payments p ON u.user_id =
p.user_id")
+ .build();
+
+ public static final TableTestProgram
MULTI_JOIN_THREE_WAY_LEFT_OUTER_JOIN_UPDATING =
+ TableTestProgram.of(
+ "three-way-left-outer-join-updating",
+ "three way left outer join updating")
+
.setupConfig(OptimizerConfigOptions.TABLE_OPTIMIZER_MULTI_JOIN_ENABLED, true)
+ .setupTableSource(
+ SourceTestStep.newBuilder("Users")
+ .addSchema(
+ "user_id STRING PRIMARY KEY NOT
ENFORCED, name STRING, cash INT")
+ .producedValues(
+ Row.of("1", "Gus", 100),
+ Row.of("2", "Bob", 200),
+ Row.of("3", "Alice", 300),
+ Row.ofKind(RowKind.UPDATE_BEFORE,
"1", "Gus", 100),
+ Row.ofKind(
+ RowKind.UPDATE_AFTER, "1",
"Gus Updated", 100),
+ Row.ofKind(
+ RowKind.UPDATE_BEFORE,
"1", "Gus Updated", 100),
+ Row.ofKind(RowKind.UPDATE_AFTER,
"1", "Gus Updated", 0),
+ Row.ofKind(RowKind.DELETE, "1",
"Gus Updated", 0))
+ .build())
+ .setupTableSource(ORDERS_SOURCE)
+ .setupTableSource(PAYMENTS_SOURCE)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(
+ "user_id STRING",
+ "name STRING",
+ "order_id STRING",
+ "payment_id STRING")
+ .consumedValues(
+ // After all updates and deletes,
Gus should not be here
+ // We only have Bob
+ "+I[2, Bob, order2, payment2]",
+ "+I[2, Bob, order3, payment2]")
+ .testMaterializedData()
+ .build())
+ .runSql(
+ "INSERT INTO sink "
+ + "SELECT u.user_id, u.name, o.order_id,
p.payment_id "
+ + "FROM Users u "
+ + "INNER JOIN Orders o ON u.user_id =
o.user_id "
+ + "LEFT JOIN Payments p ON u.user_id =
p.user_id AND u.cash >= p.price")
+ .build();
+
+ public static final TableTestProgram MULTI_JOIN_THREE_WAY_INNER_JOIN =
+ TableTestProgram.of("three-way-inner-join", "three way inner join")
+
.setupConfig(OptimizerConfigOptions.TABLE_OPTIMIZER_MULTI_JOIN_ENABLED, true)
+ .setupTableSource(USERS_SOURCE)
+ .setupTableSource(ORDERS_SOURCE)
+ .setupTableSource(PAYMENTS_SOURCE)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(
+ "user_id STRING",
+ "name STRING",
+ "order_id STRING",
+ "payment_id STRING")
+ .consumedValues(
+ "+I[1, Gus, order1, payment1]",
+ "+I[2, Bob, order2, payment2]",
+ "+I[2, Bob, order3, payment2]",
+ "+I[1, Gus, order1, payment3]")
+ .testMaterializedData()
+ .build())
+ .runSql(
+ "INSERT INTO sink "
+ + "SELECT u.user_id, u.name, o.order_id,
p.payment_id "
+ + "FROM Users u "
+ + "INNER JOIN Orders o ON u.user_id =
o.user_id "
+ + "INNER JOIN Payments p ON u.user_id =
p.user_id")
+ .build();
+
+ public static final TableTestProgram MULTI_JOIN_FOUR_WAY_COMPLEX =
+ TableTestProgram.of("four-way-complex-updating-join", "four way
complex updating join")
+
.setupConfig(OptimizerConfigOptions.TABLE_OPTIMIZER_MULTI_JOIN_ENABLED, true)
+ .setupTableSource(
+ SourceTestStep.newBuilder("Users")
+ .addSchema(
+ "user_id_0 STRING PRIMARY KEY NOT
ENFORCED",
+ "name STRING",
+ "cash INT")
+ .addOption("changelog-mode", "I,UA,D")
+ .producedValues(
+ Row.ofKind(RowKind.INSERT, "1",
"Gus", 100),
+ Row.ofKind(RowKind.INSERT, "2",
"Bob", 20),
+ Row.ofKind(RowKind.INSERT, "3",
"Nomad", 50),
+ Row.ofKind(RowKind.INSERT, "4",
"David", 5),
+ Row.ofKind(RowKind.INSERT, "5",
"Eve", 0),
+ Row.ofKind(RowKind.INSERT, "6",
"Frank", 70),
+ Row.ofKind(RowKind.INSERT, "7",
"Welcher", 100),
+ Row.ofKind(RowKind.INSERT, "8",
"Joe no order", 10),
+ Row.ofKind(RowKind.INSERT, "9",
"Charlie Smith", 50),
+ Row.ofKind(RowKind.DELETE, "2",
"Bob", 20),
+ Row.ofKind(
+ RowKind.UPDATE_AFTER,
+ "9",
+ "Charlie Taylor",
+ 50))
+ .build())
+ .setupTableSource(
+ SourceTestStep.newBuilder("Orders")
+ .addSchema(
+ "user_id_1 STRING",
+ "order_id STRING PRIMARY KEY NOT
ENFORCED",
+ "product STRING")
+ .addOption("changelog-mode", "I,D")
+ .producedValues(
+ Row.ofKind(RowKind.INSERT, "1",
"order0", "ProdB"),
+ Row.ofKind(RowKind.INSERT, "1",
"order1", "ProdA"),
+ Row.ofKind(RowKind.INSERT, "2",
"order2", "ProdB"),
+ Row.ofKind(RowKind.INSERT, "3",
"order3", "ProdC"),
+ Row.ofKind(RowKind.INSERT, "4",
"order4", "ProdD"),
+ Row.ofKind(RowKind.INSERT, "6",
"order6", "ProdF"),
+ Row.ofKind(RowKind.INSERT, "7",
"order7", "ProdG"),
+ Row.ofKind(RowKind.INSERT, "9",
"order9", "ProdA"))
+ .build())
+ .setupTableSource(
+ SourceTestStep.newBuilder("Payments")
+ .addSchema(
+ "user_id_2 STRING",
+ "payment_id STRING PRIMARY KEY NOT
ENFORCED",
+ "price INT")
+ .addOption("changelog-mode", "I")
+ .producedValues(
+ Row.ofKind(RowKind.INSERT, "1",
"payment1", 50),
+ Row.ofKind(RowKind.INSERT, "2",
"payment2", -5),
+ Row.ofKind(RowKind.INSERT, "3",
"payment3", 30),
+ Row.ofKind(RowKind.INSERT, "4",
"payment4", 40),
+ Row.ofKind(RowKind.INSERT, "5",
"payment5", -1),
+ Row.ofKind(RowKind.INSERT, "6",
"payment6", 60),
+ Row.ofKind(RowKind.INSERT, "9",
"payment9", 30))
+ .build())
+ .setupTableSource(
+ SourceTestStep.newBuilder("Shipments")
+ .addSchema("user_id_3 STRING", "location
STRING")
+ .addOption("changelog-mode", "I,UA,UB,D")
+ .producedValues(
+ Row.ofKind(RowKind.INSERT, "1",
"London"),
+ Row.ofKind(RowKind.INSERT, "2",
"Paris"),
+ Row.ofKind(RowKind.INSERT, "3",
"New York"),
+ Row.ofKind(RowKind.INSERT, "9",
"Melbourne"),
+ Row.ofKind(RowKind.INSERT, "10",
"Lost Shipment"),
+ Row.ofKind(RowKind.INSERT, "3",
"Brasília"))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(
+ "user_id STRING",
+ "name STRING",
+ "order_id STRING",
+ "payment_id STRING",
+ "location STRING")
+ .addOption("changelog-mode", "I,UA,UB,D")
+ .consumedValues(
+ "+I[1, Gus, order0, payment1,
London]",
+ "+I[1, Gus, order1, payment1,
London]",
+ // NO I[2, Bob..] here because Bob
was deleted
+ "+I[3, Nomad, order3, payment3,
New York]",
+ // NO I[4, David..] here because
David has only 5
+ // dollars, so this is not true
u.cash >= p.price
+ "+I[5, Eve, null, payment5,
null]", // Here because
+ // payment price as
+ // -1
+ // NO I[7, Welcher..] here because
Welcher has no
+ // payment
+ "+I[6, Frank, order6, payment6,
null]",
+ // NO I[8, Welcher..] here because
there is no order and
+ // no payment
+ "+I[9, Charlie Taylor, order9,
payment9, Melbourne]", // Charlie Taylor because the name was updated
+ "+I[3, Nomad, order3, payment3,
Brasília]" // New
+ // Location
+ // for user 3
+ )
+ .testMaterializedData()
+ .build())
+ .runSql(
+ "INSERT INTO sink "
+ + "SELECT u.user_id_0, u.name, o.order_id,
p.payment_id, s.location "
+ + "FROM Users u "
+ + "LEFT JOIN Orders o ON u.user_id_0 =
o.user_id_1 "
+ + "INNER JOIN Payments p ON u.user_id_0 =
p.user_id_2 AND (u.cash >= p.price OR p.price < 0) "
+ + "LEFT JOIN Shipments s ON p.user_id_2 =
s.user_id_3")
Review Comment:
Maybe we should also add a test with a WHERE clause (not ON condition). Just
to double check if the conditions are properly separated:
https://stackoverflow.com/questions/354070/sql-join-what-is-the-difference-between-where-clause-and-on-clause
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/JoinToMultiJoinRule.java:
##########
@@ -0,0 +1,683 @@
+/*
+ * 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.RelOptRuleCall;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.rel.RelNode;
+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.logical.LogicalJoin;
+import org.apache.calcite.rel.rules.CoreRules;
+import org.apache.calcite.rel.rules.FilterMultiJoinMergeRule;
+import org.apache.calcite.rel.rules.MultiJoin;
+import org.apache.calcite.rel.rules.ProjectMultiJoinMergeRule;
+import org.apache.calcite.rel.rules.TransformationRule;
+import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.rex.RexVisitorImpl;
+import org.apache.calcite.tools.RelBuilderFactory;
+import org.apache.calcite.util.ImmutableBitSet;
+import org.apache.calcite.util.ImmutableIntList;
+import org.apache.calcite.util.Pair;
+import org.immutables.value.Value;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Flink Planner rule to flatten a tree of {@link Join}s into a single {@link
MultiJoin} with N
+ * inputs.
+ *
+ * <p>This rule is copied and adjusted from {@link
org.apache.calcite.rel.rules.JoinToMultiJoinRule}
+ * and {@link JoinToMultiJoinForReorderRule}. In this rule, we support a
broder sef of left and
+ * inner joins by rewriting $canCombine() method. The multi join is not
expected to be used for
+ * reordering and will be turned into a {@link
+ *
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalMultiJoin}.
+ *
+ * <p>Join conditions are also pulled up from the inputs into the topmost
{@link MultiJoin}.
+ *
+ * <p>Join information is also stored in the {@link MultiJoin}. Join
conditions are stored in arrays
+ * in the {@link MultiJoin}. This outer join information is associated with
the null generating
+ * input in the outer join. So, in the case of a left outer join between A and
B, the information is
+ * associated with B, not A.
+ *
+ * <p>Here are examples of the {@link MultiJoin}s constructed after this rule
has been applied on
+ * following join trees.
+ *
+ * <ul>
+ * <li>A JOIN B → MJ(A, B)
+ * <li>A JOIN B JOIN C → MJ(A, B, C)
+ * <li>A LEFT JOIN B → MJ(A, B)
+ * <li>A RIGHT JOIN B → MJ(A, B)
+ * <li>A FULL JOIN B → MJ[full](A, B)
+ * <li>A LEFT JOIN (B JOIN C) → MJ(A, B, C)
+ * <li>(A JOIN B) LEFT JOIN C → MJ(A, B, C)
+ * <li>(A LEFT JOIN B) JOIN C → MJ(A, B, C)
+ * <li>(A LEFT JOIN B) LEFT JOIN C → MJ(A, B, C)
+ * <li>(A RIGHT JOIN B) RIGHT JOIN C → MJ(MJ(A, B), C)
+ * <li>(A LEFT JOIN B) RIGHT JOIN C → MJ(MJ(A, B), C)
+ * <li>(A RIGHT JOIN B) LEFT JOIN C → MJ(MJ(A, B), C)
+ * <li>A LEFT JOIN (B FULL JOIN C) → MJ(A, MJ[full](B, C))
+ * <li>(A LEFT JOIN B) FULL JOIN (C RIGHT JOIN D) → MJ[full](MJ(A, B),
MJ(C, D))
+ * <li>SEMI JOIN and ANTI JOIN not support now.
+ * </ul>
+ *
+ * <p>The constructor is parameterized to allow any sub-class of {@link Join},
not just {@link
+ * LogicalJoin}.
+ *
+ * @see FilterMultiJoinMergeRule
+ * @see ProjectMultiJoinMergeRule
+ * @see CoreRules#JOIN_TO_MULTI_JOIN
+ */
[email protected]
+public class JoinToMultiJoinRule extends RelRule<JoinToMultiJoinRule.Config>
+ implements TransformationRule {
+
+ public static final JoinToMultiJoinRule INSTANCE =
JoinToMultiJoinRule.Config.DEFAULT.toRule();
+
+ /** Creates a JoinToMultiJoinRule. */
+ public JoinToMultiJoinRule(Config config) {
+ super(config);
+ }
+
+ @Deprecated // to be removed before 2.0
+ public JoinToMultiJoinRule(Class<? extends Join> clazz) {
+ this(Config.DEFAULT.withOperandFor(clazz));
+ }
+
+ @Deprecated // to be removed before 2.0
+ public JoinToMultiJoinRule(
+ Class<? extends Join> joinClass, RelBuilderFactory
relBuilderFactory) {
+ this(
+ Config.DEFAULT
+ .withRelBuilderFactory(relBuilderFactory)
+ .as(Config.class)
+ .withOperandFor(joinClass));
+ }
+
+ // ~ Methods
----------------------------------------------------------------
+
+ @Override
+ public boolean matches(RelOptRuleCall call) {
+ final Join origJoin = call.rel(0);
+ if (origJoin.getJoinType() != JoinRelType.INNER
Review Comment:
I remember we had some issues with the multi way join rules and `FOR SYSTEM
TIME AS OF`, is it guaranteed that lookups joins are not merged into multijoin?
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/JoinToMultiJoinForReorderRule.java:
##########
@@ -99,24 +99,24 @@
* @see CoreRules#JOIN_TO_MULTI_JOIN
*/
@Value.Enclosing
-public class FlinkJoinToMultiJoinRule extends
RelRule<FlinkJoinToMultiJoinRule.Config>
+public class JoinToMultiJoinForReorderRule extends
RelRule<JoinToMultiJoinForReorderRule.Config>
Review Comment:
update JavaDoc?
```
This rule is copied from {@link
org.apache.calcite.rel.rules.JoinToMultiJoinRule}
```
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/MultiJoinSemanticTests.java:
##########
@@ -0,0 +1,37 @@
+/*
+ * 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.nodes.exec.stream;
+
+import
org.apache.flink.table.planner.plan.nodes.exec.testutils.SemanticTestBase;
+import org.apache.flink.table.test.program.TableTestProgram;
+
+import java.util.List;
+
+/** Semantic tests for multi-join. */
Review Comment:
```suggestion
/** Semantic tests for {@link StreamExecMultiJoin}. */
```
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/JoinToMultiJoinRule.java:
##########
@@ -0,0 +1,683 @@
+/*
+ * 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.RelOptRuleCall;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.rel.RelNode;
+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.logical.LogicalJoin;
+import org.apache.calcite.rel.rules.CoreRules;
+import org.apache.calcite.rel.rules.FilterMultiJoinMergeRule;
+import org.apache.calcite.rel.rules.MultiJoin;
+import org.apache.calcite.rel.rules.ProjectMultiJoinMergeRule;
+import org.apache.calcite.rel.rules.TransformationRule;
+import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.rex.RexVisitorImpl;
+import org.apache.calcite.tools.RelBuilderFactory;
+import org.apache.calcite.util.ImmutableBitSet;
+import org.apache.calcite.util.ImmutableIntList;
+import org.apache.calcite.util.Pair;
+import org.immutables.value.Value;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Flink Planner rule to flatten a tree of {@link Join}s into a single {@link
MultiJoin} with N
+ * inputs.
+ *
+ * <p>This rule is copied and adjusted from {@link
org.apache.calcite.rel.rules.JoinToMultiJoinRule}
+ * and {@link JoinToMultiJoinForReorderRule}. In this rule, we support a
broder sef of left and
+ * inner joins by rewriting $canCombine() method. The multi join is not
expected to be used for
+ * reordering and will be turned into a {@link
+ *
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalMultiJoin}.
+ *
+ * <p>Join conditions are also pulled up from the inputs into the topmost
{@link MultiJoin}.
+ *
+ * <p>Join information is also stored in the {@link MultiJoin}. Join
conditions are stored in arrays
+ * in the {@link MultiJoin}. This outer join information is associated with
the null generating
+ * input in the outer join. So, in the case of a left outer join between A and
B, the information is
+ * associated with B, not A.
+ *
+ * <p>Here are examples of the {@link MultiJoin}s constructed after this rule
has been applied on
+ * following join trees.
+ *
+ * <ul>
+ * <li>A JOIN B → MJ(A, B)
+ * <li>A JOIN B JOIN C → MJ(A, B, C)
+ * <li>A LEFT JOIN B → MJ(A, B)
+ * <li>A RIGHT JOIN B → MJ(A, B)
+ * <li>A FULL JOIN B → MJ[full](A, B)
+ * <li>A LEFT JOIN (B JOIN C) → MJ(A, B, C)
+ * <li>(A JOIN B) LEFT JOIN C → MJ(A, B, C)
+ * <li>(A LEFT JOIN B) JOIN C → MJ(A, B, C)
+ * <li>(A LEFT JOIN B) LEFT JOIN C → MJ(A, B, C)
+ * <li>(A RIGHT JOIN B) RIGHT JOIN C → MJ(MJ(A, B), C)
+ * <li>(A LEFT JOIN B) RIGHT JOIN C → MJ(MJ(A, B), C)
+ * <li>(A RIGHT JOIN B) LEFT JOIN C → MJ(MJ(A, B), C)
+ * <li>A LEFT JOIN (B FULL JOIN C) → MJ(A, MJ[full](B, C))
+ * <li>(A LEFT JOIN B) FULL JOIN (C RIGHT JOIN D) → MJ[full](MJ(A, B),
MJ(C, D))
+ * <li>SEMI JOIN and ANTI JOIN not support now.
+ * </ul>
+ *
+ * <p>The constructor is parameterized to allow any sub-class of {@link Join},
not just {@link
+ * LogicalJoin}.
+ *
+ * @see FilterMultiJoinMergeRule
+ * @see ProjectMultiJoinMergeRule
+ * @see CoreRules#JOIN_TO_MULTI_JOIN
+ */
[email protected]
+public class JoinToMultiJoinRule extends RelRule<JoinToMultiJoinRule.Config>
+ implements TransformationRule {
+
+ public static final JoinToMultiJoinRule INSTANCE =
JoinToMultiJoinRule.Config.DEFAULT.toRule();
+
+ /** Creates a JoinToMultiJoinRule. */
+ public JoinToMultiJoinRule(Config config) {
+ super(config);
+ }
+
+ @Deprecated // to be removed before 2.0
+ public JoinToMultiJoinRule(Class<? extends Join> clazz) {
+ this(Config.DEFAULT.withOperandFor(clazz));
+ }
+
+ @Deprecated // to be removed before 2.0
+ public JoinToMultiJoinRule(
+ Class<? extends Join> joinClass, RelBuilderFactory
relBuilderFactory) {
+ this(
+ Config.DEFAULT
+ .withRelBuilderFactory(relBuilderFactory)
+ .as(Config.class)
+ .withOperandFor(joinClass));
+ }
+
+ // ~ Methods
----------------------------------------------------------------
+
+ @Override
+ public boolean matches(RelOptRuleCall call) {
+ final Join origJoin = call.rel(0);
+ if (origJoin.getJoinType() != JoinRelType.INNER
Review Comment:
Potentially also interval joins should not be pulled into the multijoin.
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/MultiJoinTestPrograms.java:
##########
@@ -0,0 +1,461 @@
+/*
+ * 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.nodes.exec.stream;
+
+import org.apache.flink.table.api.config.OptimizerConfigOptions;
+import org.apache.flink.table.test.program.SinkTestStep;
+import org.apache.flink.table.test.program.SourceTestStep;
+import org.apache.flink.table.test.program.TableTestProgram;
+import org.apache.flink.types.Row;
+import org.apache.flink.types.RowKind;
+
+import static
org.apache.flink.table.planner.plan.nodes.exec.stream.MultiJoinTestUtils.ORDERS_SOURCE;
+import static
org.apache.flink.table.planner.plan.nodes.exec.stream.MultiJoinTestUtils.PAYMENTS_SOURCE;
+import static
org.apache.flink.table.planner.plan.nodes.exec.stream.MultiJoinTestUtils.USERS_SOURCE;
+
+/** {@link TableTestProgram} definitions for testing Multi-Join. */
+public class MultiJoinTestPrograms {
+ public static final TableTestProgram MULTI_JOIN_THREE_WAY_LEFT_OUTER_JOIN =
+ TableTestProgram.of("three-way-left-outer-join", "three way left
outer join")
+
.setupConfig(OptimizerConfigOptions.TABLE_OPTIMIZER_MULTI_JOIN_ENABLED, true)
+ .setupTableSource(USERS_SOURCE)
+ .setupTableSource(ORDERS_SOURCE)
+ .setupTableSource(PAYMENTS_SOURCE)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(
+ "user_id STRING",
+ "name STRING",
+ "order_id STRING",
+ "payment_id STRING")
+ .consumedValues(
+ "+I[1, Gus, order1, payment1]",
+ "+I[2, Bob, order2, payment2]",
+ "+I[2, Bob, order3, payment2]",
+ "+I[1, Gus, order1, payment3]",
+ "+I[3, Alice, null, null]")
+ .testMaterializedData()
+ .build())
+ .runSql(
+ "INSERT INTO sink "
+ + "SELECT u.user_id, u.name, o.order_id,
p.payment_id "
+ + "FROM Users u "
+ + "LEFT JOIN Orders o ON u.user_id =
o.user_id "
+ + "LEFT JOIN Payments p ON u.user_id =
p.user_id")
+ .build();
+
+ public static final TableTestProgram
MULTI_JOIN_THREE_WAY_LEFT_OUTER_JOIN_UPDATING =
+ TableTestProgram.of(
+ "three-way-left-outer-join-updating",
+ "three way left outer join updating")
+
.setupConfig(OptimizerConfigOptions.TABLE_OPTIMIZER_MULTI_JOIN_ENABLED, true)
+ .setupTableSource(
+ SourceTestStep.newBuilder("Users")
+ .addSchema(
+ "user_id STRING PRIMARY KEY NOT
ENFORCED, name STRING, cash INT")
+ .producedValues(
+ Row.of("1", "Gus", 100),
+ Row.of("2", "Bob", 200),
+ Row.of("3", "Alice", 300),
+ Row.ofKind(RowKind.UPDATE_BEFORE,
"1", "Gus", 100),
+ Row.ofKind(
+ RowKind.UPDATE_AFTER, "1",
"Gus Updated", 100),
+ Row.ofKind(
+ RowKind.UPDATE_BEFORE,
"1", "Gus Updated", 100),
+ Row.ofKind(RowKind.UPDATE_AFTER,
"1", "Gus Updated", 0),
+ Row.ofKind(RowKind.DELETE, "1",
"Gus Updated", 0))
+ .build())
+ .setupTableSource(ORDERS_SOURCE)
+ .setupTableSource(PAYMENTS_SOURCE)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(
+ "user_id STRING",
+ "name STRING",
+ "order_id STRING",
+ "payment_id STRING")
+ .consumedValues(
+ // After all updates and deletes,
Gus should not be here
+ // We only have Bob
+ "+I[2, Bob, order2, payment2]",
+ "+I[2, Bob, order3, payment2]")
+ .testMaterializedData()
+ .build())
+ .runSql(
+ "INSERT INTO sink "
+ + "SELECT u.user_id, u.name, o.order_id,
p.payment_id "
+ + "FROM Users u "
+ + "INNER JOIN Orders o ON u.user_id =
o.user_id "
+ + "LEFT JOIN Payments p ON u.user_id =
p.user_id AND u.cash >= p.price")
+ .build();
+
+ public static final TableTestProgram MULTI_JOIN_THREE_WAY_INNER_JOIN =
+ TableTestProgram.of("three-way-inner-join", "three way inner join")
+
.setupConfig(OptimizerConfigOptions.TABLE_OPTIMIZER_MULTI_JOIN_ENABLED, true)
+ .setupTableSource(USERS_SOURCE)
+ .setupTableSource(ORDERS_SOURCE)
+ .setupTableSource(PAYMENTS_SOURCE)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(
+ "user_id STRING",
+ "name STRING",
+ "order_id STRING",
+ "payment_id STRING")
+ .consumedValues(
+ "+I[1, Gus, order1, payment1]",
+ "+I[2, Bob, order2, payment2]",
+ "+I[2, Bob, order3, payment2]",
+ "+I[1, Gus, order1, payment3]")
+ .testMaterializedData()
+ .build())
+ .runSql(
+ "INSERT INTO sink "
+ + "SELECT u.user_id, u.name, o.order_id,
p.payment_id "
+ + "FROM Users u "
+ + "INNER JOIN Orders o ON u.user_id =
o.user_id "
+ + "INNER JOIN Payments p ON u.user_id =
p.user_id")
+ .build();
+
+ public static final TableTestProgram MULTI_JOIN_FOUR_WAY_COMPLEX =
+ TableTestProgram.of("four-way-complex-updating-join", "four way
complex updating join")
+
.setupConfig(OptimizerConfigOptions.TABLE_OPTIMIZER_MULTI_JOIN_ENABLED, true)
+ .setupTableSource(
+ SourceTestStep.newBuilder("Users")
+ .addSchema(
+ "user_id_0 STRING PRIMARY KEY NOT
ENFORCED",
+ "name STRING",
+ "cash INT")
+ .addOption("changelog-mode", "I,UA,D")
+ .producedValues(
+ Row.ofKind(RowKind.INSERT, "1",
"Gus", 100),
+ Row.ofKind(RowKind.INSERT, "2",
"Bob", 20),
+ Row.ofKind(RowKind.INSERT, "3",
"Nomad", 50),
+ Row.ofKind(RowKind.INSERT, "4",
"David", 5),
+ Row.ofKind(RowKind.INSERT, "5",
"Eve", 0),
+ Row.ofKind(RowKind.INSERT, "6",
"Frank", 70),
+ Row.ofKind(RowKind.INSERT, "7",
"Welcher", 100),
+ Row.ofKind(RowKind.INSERT, "8",
"Joe no order", 10),
+ Row.ofKind(RowKind.INSERT, "9",
"Charlie Smith", 50),
+ Row.ofKind(RowKind.DELETE, "2",
"Bob", 20),
+ Row.ofKind(
+ RowKind.UPDATE_AFTER,
+ "9",
+ "Charlie Taylor",
+ 50))
+ .build())
+ .setupTableSource(
+ SourceTestStep.newBuilder("Orders")
+ .addSchema(
+ "user_id_1 STRING",
+ "order_id STRING PRIMARY KEY NOT
ENFORCED",
+ "product STRING")
+ .addOption("changelog-mode", "I,D")
+ .producedValues(
+ Row.ofKind(RowKind.INSERT, "1",
"order0", "ProdB"),
+ Row.ofKind(RowKind.INSERT, "1",
"order1", "ProdA"),
+ Row.ofKind(RowKind.INSERT, "2",
"order2", "ProdB"),
+ Row.ofKind(RowKind.INSERT, "3",
"order3", "ProdC"),
+ Row.ofKind(RowKind.INSERT, "4",
"order4", "ProdD"),
+ Row.ofKind(RowKind.INSERT, "6",
"order6", "ProdF"),
+ Row.ofKind(RowKind.INSERT, "7",
"order7", "ProdG"),
+ Row.ofKind(RowKind.INSERT, "9",
"order9", "ProdA"))
Review Comment:
looks like the ids are mostly perfectly ordered and also always at column 0,
I suggest to reorder some and move columns also to the middle or end.
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/JoinToMultiJoinRule.java:
##########
@@ -0,0 +1,683 @@
+/*
+ * 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.RelOptRuleCall;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.rel.RelNode;
+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.logical.LogicalJoin;
+import org.apache.calcite.rel.rules.CoreRules;
+import org.apache.calcite.rel.rules.FilterMultiJoinMergeRule;
+import org.apache.calcite.rel.rules.MultiJoin;
+import org.apache.calcite.rel.rules.ProjectMultiJoinMergeRule;
+import org.apache.calcite.rel.rules.TransformationRule;
+import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.rex.RexVisitorImpl;
+import org.apache.calcite.tools.RelBuilderFactory;
+import org.apache.calcite.util.ImmutableBitSet;
+import org.apache.calcite.util.ImmutableIntList;
+import org.apache.calcite.util.Pair;
+import org.immutables.value.Value;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Flink Planner rule to flatten a tree of {@link Join}s into a single {@link
MultiJoin} with N
+ * inputs.
+ *
+ * <p>This rule is copied and adjusted from {@link
org.apache.calcite.rel.rules.JoinToMultiJoinRule}
+ * and {@link JoinToMultiJoinForReorderRule}. In this rule, we support a
broder sef of left and
+ * inner joins by rewriting $canCombine() method. The multi join is not
expected to be used for
+ * reordering and will be turned into a {@link
+ *
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalMultiJoin}.
+ *
+ * <p>Join conditions are also pulled up from the inputs into the topmost
{@link MultiJoin}.
+ *
+ * <p>Join information is also stored in the {@link MultiJoin}. Join
conditions are stored in arrays
+ * in the {@link MultiJoin}. This outer join information is associated with
the null generating
+ * input in the outer join. So, in the case of a left outer join between A and
B, the information is
+ * associated with B, not A.
+ *
+ * <p>Here are examples of the {@link MultiJoin}s constructed after this rule
has been applied on
+ * following join trees.
+ *
+ * <ul>
+ * <li>A JOIN B → MJ(A, B)
+ * <li>A JOIN B JOIN C → MJ(A, B, C)
+ * <li>A LEFT JOIN B → MJ(A, B)
+ * <li>A RIGHT JOIN B → MJ(A, B)
+ * <li>A FULL JOIN B → MJ[full](A, B)
+ * <li>A LEFT JOIN (B JOIN C) → MJ(A, B, C)
+ * <li>(A JOIN B) LEFT JOIN C → MJ(A, B, C)
+ * <li>(A LEFT JOIN B) JOIN C → MJ(A, B, C)
+ * <li>(A LEFT JOIN B) LEFT JOIN C → MJ(A, B, C)
+ * <li>(A RIGHT JOIN B) RIGHT JOIN C → MJ(MJ(A, B), C)
+ * <li>(A LEFT JOIN B) RIGHT JOIN C → MJ(MJ(A, B), C)
+ * <li>(A RIGHT JOIN B) LEFT JOIN C → MJ(MJ(A, B), C)
+ * <li>A LEFT JOIN (B FULL JOIN C) → MJ(A, MJ[full](B, C))
+ * <li>(A LEFT JOIN B) FULL JOIN (C RIGHT JOIN D) → MJ[full](MJ(A, B),
MJ(C, D))
+ * <li>SEMI JOIN and ANTI JOIN not support now.
+ * </ul>
+ *
+ * <p>The constructor is parameterized to allow any sub-class of {@link Join},
not just {@link
+ * LogicalJoin}.
+ *
+ * @see FilterMultiJoinMergeRule
+ * @see ProjectMultiJoinMergeRule
+ * @see CoreRules#JOIN_TO_MULTI_JOIN
+ */
[email protected]
+public class JoinToMultiJoinRule extends RelRule<JoinToMultiJoinRule.Config>
+ implements TransformationRule {
+
+ public static final JoinToMultiJoinRule INSTANCE =
JoinToMultiJoinRule.Config.DEFAULT.toRule();
+
+ /** Creates a JoinToMultiJoinRule. */
+ public JoinToMultiJoinRule(Config config) {
+ super(config);
+ }
+
+ @Deprecated // to be removed before 2.0
+ public JoinToMultiJoinRule(Class<? extends Join> clazz) {
+ this(Config.DEFAULT.withOperandFor(clazz));
+ }
+
+ @Deprecated // to be removed before 2.0
+ public JoinToMultiJoinRule(
+ Class<? extends Join> joinClass, RelBuilderFactory
relBuilderFactory) {
+ this(
+ Config.DEFAULT
+ .withRelBuilderFactory(relBuilderFactory)
+ .as(Config.class)
+ .withOperandFor(joinClass));
+ }
+
+ // ~ Methods
----------------------------------------------------------------
+
+ @Override
+ public boolean matches(RelOptRuleCall call) {
+ final Join origJoin = call.rel(0);
+ if (origJoin.getJoinType() != JoinRelType.INNER
+ && origJoin.getJoinType() != JoinRelType.LEFT) {
+ /* This rule expects only INNER and LEFT joins. Right joins are
expected to be
+ rewritten to left joins by the optimizer with {@link
FlinkRightJoinToLeftJoinRule} */
+ return false;
+ }
+
+ return origJoin.getJoinType().projectsRight();
+ }
+
+ @Override
+ public void onMatch(RelOptRuleCall call) {
+ final Join origJoin = call.rel(0);
+ final RelNode left = call.rel(1);
+ final RelNode right = call.rel(2);
+
+ // inputNullGenFieldList records whether the field in originJoin is
null generate field.
+ List<Boolean> inputNullGenFieldList = new ArrayList<>();
+ // Build null generate field list.
+ buildInputNullGenFieldList(left, right, origJoin.getJoinType(),
inputNullGenFieldList);
+
+ // Combine the children MultiJoin inputs into an array of inputs for
the new MultiJoin.
+ final List<ImmutableBitSet> projFieldsList = new ArrayList<>();
+ final List<int[]> joinFieldRefCountsList = new ArrayList<>();
+ final List<RelNode> newInputs =
+ combineInputs(
+ origJoin,
+ left,
+ right,
+ projFieldsList,
+ joinFieldRefCountsList,
+ inputNullGenFieldList);
+
+ // Combine the join information from the left and right inputs, and
include the
+ // join information from the current join.
+ final List<Pair<JoinRelType, RexNode>> joinSpecs = new ArrayList<>();
+ combineJoinInfo(origJoin, newInputs, left, right, joinSpecs,
inputNullGenFieldList);
+
+ // Pull up the join filters from the children MultiJoinRels and
combine them with the join
+ // filter associated with this LogicalJoin to form the join filter for
the new MultiJoin.
+ List<RexNode> newJoinFilters =
+ combineJoinFilters(origJoin, left, right,
inputNullGenFieldList);
+
+ // Add on the join field reference counts for the join condition
associated with this
+ // LogicalJoin.
+ final com.google.common.collect.ImmutableMap<Integer, ImmutableIntList>
+ newJoinFieldRefCountsMap =
+ addOnJoinFieldRefCounts(
+ newInputs,
+ origJoin.getRowType().getFieldCount(),
+ origJoin.getCondition(),
+ joinFieldRefCountsList);
+
+ List<RexNode> newPostJoinFilters = combinePostJoinFilters(origJoin,
left, right);
+
+ final RexBuilder rexBuilder = origJoin.getCluster().getRexBuilder();
+ RelNode multiJoin =
+ new MultiJoin(
+ origJoin.getCluster(),
+ newInputs,
+ RexUtil.composeConjunction(rexBuilder, newJoinFilters),
+ origJoin.getRowType(),
+ origJoin.getJoinType() == JoinRelType.FULL,
+ Pair.right(joinSpecs),
+ Pair.left(joinSpecs),
+ projFieldsList,
+ newJoinFieldRefCountsMap,
+ RexUtil.composeConjunction(rexBuilder,
newPostJoinFilters, true));
+
+ call.transformTo(multiJoin);
+ }
+
+ private void buildInputNullGenFieldList(
+ RelNode left, RelNode right, JoinRelType joinType, List<Boolean>
isNullGenFieldList) {
+ if (joinType == JoinRelType.INNER) {
+ buildNullGenFieldList(left, isNullGenFieldList);
+ buildNullGenFieldList(right, isNullGenFieldList);
+ } else if (joinType == JoinRelType.LEFT) {
+ // If origin joinType is left means join fields from right side
must be null generated
+ // fields, so we need only judge these join fields in left side
and set null generate
+ // field is true for all right fields.
+ buildNullGenFieldList(left, isNullGenFieldList);
+
+ for (int i = 0; i < right.getRowType().getFieldCount(); i++) {
+ isNullGenFieldList.add(true);
+ }
+ } else {
+ // Now, join to multi join rule only support Full outer join,
Inner join and Left/Right
+ // join.
+ throw new TableException(
+ "This is a bug. Now, join to multi join rule only support
Full outer "
+ + "join, Inner join and Left/Right join.");
+ }
+ }
+
+ private void buildNullGenFieldList(RelNode rel, List<Boolean>
isNullGenFieldList) {
+ MultiJoin multiJoin = rel instanceof MultiJoin ? (MultiJoin) rel :
null;
+ if (multiJoin == null) {
+ // other operators.
+ for (int i = 0; i < rel.getRowType().getFieldCount(); i++) {
+ isNullGenFieldList.add(false);
+ }
+ } else {
+ List<RelNode> inputs = multiJoin.getInputs();
+ List<JoinRelType> joinTypes = multiJoin.getJoinTypes();
+ for (int i = 0; i < inputs.size() - 1; i++) {
+ // In list joinTypes, right join node will be added as [RIGHT,
INNER], so we need to
+ // get the joinType from joinTypes in index i.
+ if (joinTypes.get(i) == JoinRelType.RIGHT) {
+ buildInputNullGenFieldList(
+ inputs.get(i), inputs.get(i + 1),
joinTypes.get(i), isNullGenFieldList);
+ } else {
+ // In list joinTypes, left join node and inner join node
will be added as
+ // [INNER, LEFT] and [INNER, INNER] respectively. so we
need to get the joinType
+ // from joinTypes in index i + 1.
+ buildInputNullGenFieldList(
+ inputs.get(i),
+ inputs.get(i + 1),
+ joinTypes.get(i + 1),
+ isNullGenFieldList);
+ }
+ }
+ }
+ }
+
+ /**
+ * Combines the inputs into a LogicalJoin into an array of inputs.
+ *
+ * @param join original join
+ * @param left left input into join
+ * @param right right input into join
+ * @param projFieldsList returns a list of the new combined projection
fields
+ * @param joinFieldRefCountsList returns a list of the new combined join
field reference counts
+ * @return combined left and right inputs in an array
+ */
+ private List<RelNode> combineInputs(
+ Join join,
+ RelNode left,
+ RelNode right,
+ List<ImmutableBitSet> projFieldsList,
+ List<int[]> joinFieldRefCountsList,
+ List<Boolean> inputNullGenFieldList) {
+ final List<RelNode> newInputs = new ArrayList<>();
+ // Leave the null generating sides of an outer join intact; don't pull
up those children
+ // inputs into the array we're constructing.
+ JoinInfo joinInfo = join.analyzeCondition();
+ ImmutableIntList leftKeys = joinInfo.leftKeys;
+ ImmutableIntList rightKeys = joinInfo.rightKeys;
+
+ if (canCombine(
+ left,
+ leftKeys,
+ join.getJoinType(),
+ join.getJoinType().generatesNullsOnLeft(),
+ true,
+ inputNullGenFieldList,
+ 0)) {
+ final MultiJoin leftMultiJoin = (MultiJoin) left;
+ for (int i = 0; i < leftMultiJoin.getInputs().size(); i++) {
+ newInputs.add(leftMultiJoin.getInput(i));
+ projFieldsList.add(leftMultiJoin.getProjFields().get(i));
+ joinFieldRefCountsList.add(
+
leftMultiJoin.getJoinFieldRefCountsMap().get(i).toIntArray());
+ }
+
+ } else {
+ newInputs.add(left);
+ projFieldsList.add(null);
+ joinFieldRefCountsList.add(new
int[left.getRowType().getFieldCount()]);
+ }
+
+ if (canCombine(
+ right,
+ rightKeys,
+ join.getJoinType(),
+ join.getJoinType().generatesNullsOnRight(),
+ false,
+ inputNullGenFieldList,
+ left.getRowType().getFieldCount())) {
+ final MultiJoin rightMultiJoin = (MultiJoin) right;
+ for (int i = 0; i < rightMultiJoin.getInputs().size(); i++) {
+ newInputs.add(rightMultiJoin.getInput(i));
+ projFieldsList.add(rightMultiJoin.getProjFields().get(i));
+ joinFieldRefCountsList.add(
+
rightMultiJoin.getJoinFieldRefCountsMap().get(i).toIntArray());
+ }
+ } else {
+ newInputs.add(right);
+ projFieldsList.add(null);
+ joinFieldRefCountsList.add(new
int[right.getRowType().getFieldCount()]);
+ }
+
+ return newInputs;
+ }
+
+ /**
+ * Combines the join conditions and join types from the left and right
join inputs. If the join
+ * itself is either a left or right outer join, then the join condition
corresponding to the
+ * join is also set in the position corresponding to the null-generating
input into the join.
+ * The join type is also set.
+ *
+ * @param joinRel join rel
+ * @param combinedInputs the combined inputs to the join
+ * @param left left child of the joinrel
+ * @param right right child of the joinrel
+ * @param joinSpecs the list where the join types and conditions will be
copied
+ */
+ private void combineJoinInfo(
+ Join joinRel,
+ List<RelNode> combinedInputs,
+ RelNode left,
+ RelNode right,
+ List<Pair<JoinRelType, RexNode>> joinSpecs,
+ List<Boolean> inputNullGenFieldList) {
+ JoinRelType joinType = joinRel.getJoinType();
+ JoinInfo joinInfo = joinRel.analyzeCondition();
+ ImmutableIntList leftKeys = joinInfo.leftKeys;
+ final RexBuilder rexBuilder = joinRel.getCluster().getRexBuilder();
+ boolean leftCombined =
+ canCombine(
+ left,
+ leftKeys,
+ joinType,
+ joinType.generatesNullsOnLeft(),
+ true,
+ inputNullGenFieldList,
+ 0);
+ switch (joinType) {
+ case LEFT:
+ if (leftCombined) {
+ copyJoinInfo((MultiJoin) left, joinSpecs, 0, null, null);
+ } else {
+ joinSpecs.add(Pair.of(JoinRelType.INNER,
rexBuilder.makeLiteral(true)));
+ }
+ joinSpecs.add(Pair.of(joinType, joinRel.getCondition()));
+ break;
+ case INNER:
+ if (leftCombined) {
+ copyJoinInfo((MultiJoin) left, joinSpecs, 0, null, null);
+ } else {
+ joinSpecs.add(Pair.of(JoinRelType.INNER,
rexBuilder.makeLiteral(true)));
+ }
+
+ joinSpecs.add(Pair.of(joinType, joinRel.getCondition()));
+ break;
+ default:
+ throw new TableException(
+ "This is a bug. This rule only supports left and inner
joins");
+ }
+ }
+
+ /**
+ * Copies join data from a source MultiJoin to a new set of arrays. Also
adjusts the conditions
+ * to reflect the new position of an input if that input ends up being
shifted to the right.
+ *
+ * @param multiJoin the source MultiJoin
+ * @param destJoinSpecs the list where the join types and conditions will
be copied
+ * @param adjustmentAmount if > 0, the amount the RexInputRefs in the
join conditions need to
+ * be adjusted by
+ * @param srcFields the source fields that the original join conditions
are referencing
+ * @param destFields the destination fields that the new join conditions
+ */
+ private void copyJoinInfo(
+ MultiJoin multiJoin,
+ List<Pair<JoinRelType, RexNode>> destJoinSpecs,
+ int adjustmentAmount,
+ @Nullable List<RelDataTypeField> srcFields,
+ @Nullable List<RelDataTypeField> destFields) {
+ // getOuterJoinConditions are return all join conditions since that's
how we use it
+ final List<Pair<JoinRelType, RexNode>> srcJoinSpecs =
+ Pair.zip(multiJoin.getJoinTypes(),
multiJoin.getOuterJoinConditions());
+
+ if (adjustmentAmount == 0) {
+ destJoinSpecs.addAll(srcJoinSpecs);
+ } else {
+ assert srcFields != null;
+ assert destFields != null;
+ int nFields = srcFields.size();
+ int[] adjustments = new int[nFields];
+ for (int idx = 0; idx < nFields; idx++) {
+ adjustments[idx] = adjustmentAmount;
+ }
+ for (Pair<JoinRelType, RexNode> src : srcJoinSpecs) {
+ destJoinSpecs.add(
+ Pair.of(
+ src.left,
+ src.right == null
+ ? null
+ : src.right.accept(
+ new
RelOptUtil.RexInputConverter(
+
multiJoin.getCluster().getRexBuilder(),
+ srcFields,
+ destFields,
+ adjustments))));
+ }
+ }
+ }
+
+ /**
+ * Combines the join filters from the left and right inputs (if they are
MultiJoinRels) with the
+ * join filter in the joinrel into a single AND'd join filter, unless the
inputs correspond to
+ * null generating inputs in an outer join.
+ *
+ * @param join Join
+ * @param left Left input of the join
+ * @param right Right input of the join
+ * @return combined join filters AND-ed together
+ */
+ private List<RexNode> combineJoinFilters(
+ Join join, RelNode left, RelNode right, List<Boolean>
inputNullGenFieldList) {
+ JoinRelType joinType = join.getJoinType();
+ JoinInfo joinInfo = join.analyzeCondition();
+ ImmutableIntList leftKeys = joinInfo.leftKeys;
+
+ if (joinType == JoinRelType.RIGHT) {
+ throw new TableException("This is a bug. This rule only supports
left and inner joins");
+ }
+ // AND the join condition if this isn't a left join; In those cases,
the
+ // outer join condition is already tracked separately.
+ final List<RexNode> filters = new ArrayList<>();
+ if ((joinType != JoinRelType.LEFT)) {
+ filters.add(join.getCondition());
+ }
+ if (canCombine(
+ left,
+ leftKeys,
+ joinType,
+ joinType.generatesNullsOnLeft(),
+ true,
+ inputNullGenFieldList,
+ 0)) {
+ filters.add(((MultiJoin) left).getJoinFilter());
+ }
+
+ return filters;
+ }
+
+ /**
+ * Returns whether an input can be merged into a given relational
expression without changing
+ * semantics.
+ *
+ * <p>This method should be extended to check for the common join key
restriction to support
+ * multiple multi joins. See <a
+ *
href="https://issues.apache.org/jira/browse/FLINK-37890";>FLINK-37890</a>.
+ *
+ * @param input input into a join
+ * @param nullGenerating true if the input is null generating
+ * @return true if the input can be combined into a parent MultiJoin
+ */
+ private boolean canCombine(
+ RelNode input,
+ ImmutableIntList joinKeys,
+ JoinRelType joinType,
+ boolean nullGenerating,
+ boolean isLeft,
+ List<Boolean> inputNullGenFieldList,
+ int beginIndex) {
+ if (input instanceof MultiJoin) {
+ MultiJoin join = (MultiJoin) input;
+ if (join.isFullOuterJoin() || nullGenerating) {
+ return false;
+ }
+
+ if (joinType == JoinRelType.LEFT) {
+ if (!isLeft) {
+ return false;
+ } else {
+ for (int joinKey : joinKeys) {
+ if (inputNullGenFieldList.get(joinKey + beginIndex)) {
+ return false;
+ }
+ }
+ }
+ } else if (joinType == JoinRelType.RIGHT) {
+ if (isLeft) {
+ return false;
+ } else {
+ for (int joinKey : joinKeys) {
+ if (inputNullGenFieldList.get(joinKey + beginIndex)) {
+ return false;
+ }
+ }
+ }
+ } else if (joinType == JoinRelType.INNER) {
+ for (int joinKey : joinKeys) {
+ if (inputNullGenFieldList.get(joinKey + beginIndex)) {
+ return false;
+ }
+ }
+ } else {
+ return false;
+ }
+ return true;
+ } else {
+ return false;
+ }
+ }
+
+ /**
+ * Shifts a filter originating from the right child of the LogicalJoin to
the right, to reflect
+ * the filter now being applied on the resulting MultiJoin.
+ *
+ * @param joinRel the original LogicalJoin
+ * @param left the left child of the LogicalJoin
+ * @param right the right child of the LogicalJoin
+ * @param rightFilter the filter originating from the right child
+ * @return the adjusted right filter
+ */
+ private RexNode shiftRightFilter(
+ Join joinRel, RelNode left, MultiJoin right, RexNode rightFilter) {
+ if (rightFilter == null) {
+ return null;
+ }
+
+ int nFieldsOnLeft = left.getRowType().getFieldList().size();
+ int nFieldsOnRight = right.getRowType().getFieldList().size();
+ int[] adjustments = new int[nFieldsOnRight];
+ for (int i = 0; i < nFieldsOnRight; i++) {
+ adjustments[i] = nFieldsOnLeft;
+ }
+ rightFilter =
+ rightFilter.accept(
+ new RelOptUtil.RexInputConverter(
+ joinRel.getCluster().getRexBuilder(),
+ right.getRowType().getFieldList(),
+ joinRel.getRowType().getFieldList(),
+ adjustments));
+ return rightFilter;
+ }
+
+ /**
+ * Adds on to the existing join condition reference counts the references
from the new join
+ * condition.
+ *
+ * @param multiJoinInputs inputs into the new MultiJoin
+ * @param nTotalFields total number of fields in the MultiJoin
+ * @param joinCondition the new join condition
+ * @param origJoinFieldRefCounts existing join condition reference counts
+ * @return Map containing the new join condition
+ */
+ private com.google.common.collect.ImmutableMap<Integer, ImmutableIntList>
Review Comment:
use the shaded guava. Or avoid it altogether and use a regular map that is
guarded with `Collections.immutableMap()`. We don't use Guava directly in Flink.
##########
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:
##########
@@ -0,0 +1,273 @@
+{
+ "flinkVersion" : "2.1",
+ "nodes" : [ {
+ "id" : 1,
+ "type" : "stream-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`Users`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "user_id",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "name",
+ "dataType" : "VARCHAR(2147483647)"
+ } ],
+ "watermarkSpecs" : [ ]
+ },
+ "partitionKeys" : [ ]
+ }
+ }
+ },
+ "outputType" : "ROW<`user_id` VARCHAR(2147483647), `name`
VARCHAR(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, Users]], fields=[user_id, name])",
+ "inputProperties" : [ ]
+ }, {
+ "id" : 2,
+ "type" : "stream-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`Orders`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "user_id",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "order_id",
+ "dataType" : "VARCHAR(2147483647)"
+ } ],
+ "watermarkSpecs" : [ ]
+ },
+ "partitionKeys" : [ ]
+ }
+ }
+ },
+ "outputType" : "ROW<`user_id` VARCHAR(2147483647), `order_id`
VARCHAR(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, Orders]], fields=[user_id, order_id])",
+ "inputProperties" : [ ]
+ }, {
+ "id" : 3,
+ "type" : "stream-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`Payments`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "user_id",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "payment_id",
+ "dataType" : "VARCHAR(2147483647)"
+ } ],
+ "watermarkSpecs" : [ ]
+ },
+ "partitionKeys" : [ ]
+ }
+ }
+ },
+ "outputType" : "ROW<`user_id` VARCHAR(2147483647), `payment_id`
VARCHAR(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, Payments]], fields=[user_id, payment_id])",
+ "inputProperties" : [ ]
+ }, {
+ "id" : 4,
+ "type" : "stream-exec-multi-join_1",
+ "joinTypes" : [ "INNER", "INNER", "INNER" ],
+ "joinConditions" : [ {
+ "kind" : "LITERAL",
+ "value" : true,
+ "type" : "BOOLEAN NOT NULL"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$=$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "type" : "BOOLEAN"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$=$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 4,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "type" : "BOOLEAN"
+ } ],
+ "multiJoinCondition" : null,
+ "joinAttributeMap" : {
+ "1" : [ {
+ "leftInputId" : 0,
+ "leftFieldIndex" : 0,
+ "rightInputId" : 1,
+ "rightFieldIndex" : 0
+ } ],
+ "2" : [ {
+ "leftInputId" : 0,
+ "leftFieldIndex" : 0,
+ "rightInputId" : 2,
+ "rightFieldIndex" : 0
+ } ]
+ },
+ "inputUpsertKeys" : [ [ ], [ ], [ ] ],
Review Comment:
some comment as able: https://github.com/apache/flink/pull/26683
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/keyselector/AttributeBasedJoinKeyExtractor.java:
##########
@@ -417,14 +417,19 @@ private boolean
isCommonConceptualAttributeSet(Set<AttributeRef> eqSet) {
private void processCommonAttributes(List<Set<AttributeRef>>
commonConceptualAttributeSets) {
for (int currentInputId = 0; currentInputId < inputTypes.size();
currentInputId++) {
- List<AttributeRef> commonAttrsForThisInput =
+ final List<AttributeRef> commonAttrsForThisInput =
findCommonAttributesForInput(currentInputId,
commonConceptualAttributeSets);
if (commonAttrsForThisInput.isEmpty()) {
+ // This indicates that there is no common join key among all
inputs.
+ // In this case, we cannot use a multi-join, so throw an
exception.
throw new IllegalStateException(
"No common attributes found for inputId "
+ currentInputId
- + ". This indicates a misconfiguration in
joinAttributeMap.");
+ + ". This indicates a misconfiguration in
joinAttributeMap,"
Review Comment:
Make the error more user friendly. User doesn't care about internals like
joinAttributeMap.
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/MultiJoinTestPrograms.java:
##########
@@ -0,0 +1,461 @@
+/*
+ * 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.nodes.exec.stream;
+
+import org.apache.flink.table.api.config.OptimizerConfigOptions;
+import org.apache.flink.table.test.program.SinkTestStep;
+import org.apache.flink.table.test.program.SourceTestStep;
+import org.apache.flink.table.test.program.TableTestProgram;
+import org.apache.flink.types.Row;
+import org.apache.flink.types.RowKind;
+
+import static
org.apache.flink.table.planner.plan.nodes.exec.stream.MultiJoinTestUtils.ORDERS_SOURCE;
+import static
org.apache.flink.table.planner.plan.nodes.exec.stream.MultiJoinTestUtils.PAYMENTS_SOURCE;
+import static
org.apache.flink.table.planner.plan.nodes.exec.stream.MultiJoinTestUtils.USERS_SOURCE;
+
+/** {@link TableTestProgram} definitions for testing Multi-Join. */
+public class MultiJoinTestPrograms {
+ public static final TableTestProgram MULTI_JOIN_THREE_WAY_LEFT_OUTER_JOIN =
+ TableTestProgram.of("three-way-left-outer-join", "three way left
outer join")
+
.setupConfig(OptimizerConfigOptions.TABLE_OPTIMIZER_MULTI_JOIN_ENABLED, true)
+ .setupTableSource(USERS_SOURCE)
+ .setupTableSource(ORDERS_SOURCE)
+ .setupTableSource(PAYMENTS_SOURCE)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(
+ "user_id STRING",
+ "name STRING",
+ "order_id STRING",
+ "payment_id STRING")
+ .consumedValues(
+ "+I[1, Gus, order1, payment1]",
+ "+I[2, Bob, order2, payment2]",
+ "+I[2, Bob, order3, payment2]",
+ "+I[1, Gus, order1, payment3]",
+ "+I[3, Alice, null, null]")
+ .testMaterializedData()
+ .build())
+ .runSql(
+ "INSERT INTO sink "
+ + "SELECT u.user_id, u.name, o.order_id,
p.payment_id "
+ + "FROM Users u "
+ + "LEFT JOIN Orders o ON u.user_id =
o.user_id "
+ + "LEFT JOIN Payments p ON u.user_id =
p.user_id")
+ .build();
+
+ public static final TableTestProgram
MULTI_JOIN_THREE_WAY_LEFT_OUTER_JOIN_UPDATING =
+ TableTestProgram.of(
+ "three-way-left-outer-join-updating",
+ "three way left outer join updating")
+
.setupConfig(OptimizerConfigOptions.TABLE_OPTIMIZER_MULTI_JOIN_ENABLED, true)
+ .setupTableSource(
+ SourceTestStep.newBuilder("Users")
+ .addSchema(
+ "user_id STRING PRIMARY KEY NOT
ENFORCED, name STRING, cash INT")
+ .producedValues(
+ Row.of("1", "Gus", 100),
+ Row.of("2", "Bob", 200),
+ Row.of("3", "Alice", 300),
+ Row.ofKind(RowKind.UPDATE_BEFORE,
"1", "Gus", 100),
+ Row.ofKind(
+ RowKind.UPDATE_AFTER, "1",
"Gus Updated", 100),
+ Row.ofKind(
+ RowKind.UPDATE_BEFORE,
"1", "Gus Updated", 100),
+ Row.ofKind(RowKind.UPDATE_AFTER,
"1", "Gus Updated", 0),
+ Row.ofKind(RowKind.DELETE, "1",
"Gus Updated", 0))
+ .build())
+ .setupTableSource(ORDERS_SOURCE)
+ .setupTableSource(PAYMENTS_SOURCE)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(
+ "user_id STRING",
+ "name STRING",
+ "order_id STRING",
+ "payment_id STRING")
+ .consumedValues(
+ // After all updates and deletes,
Gus should not be here
+ // We only have Bob
+ "+I[2, Bob, order2, payment2]",
+ "+I[2, Bob, order3, payment2]")
+ .testMaterializedData()
+ .build())
+ .runSql(
+ "INSERT INTO sink "
+ + "SELECT u.user_id, u.name, o.order_id,
p.payment_id "
+ + "FROM Users u "
+ + "INNER JOIN Orders o ON u.user_id =
o.user_id "
+ + "LEFT JOIN Payments p ON u.user_id =
p.user_id AND u.cash >= p.price")
+ .build();
+
+ public static final TableTestProgram MULTI_JOIN_THREE_WAY_INNER_JOIN =
+ TableTestProgram.of("three-way-inner-join", "three way inner join")
+
.setupConfig(OptimizerConfigOptions.TABLE_OPTIMIZER_MULTI_JOIN_ENABLED, true)
+ .setupTableSource(USERS_SOURCE)
+ .setupTableSource(ORDERS_SOURCE)
+ .setupTableSource(PAYMENTS_SOURCE)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(
+ "user_id STRING",
+ "name STRING",
+ "order_id STRING",
+ "payment_id STRING")
+ .consumedValues(
+ "+I[1, Gus, order1, payment1]",
+ "+I[2, Bob, order2, payment2]",
+ "+I[2, Bob, order3, payment2]",
+ "+I[1, Gus, order1, payment3]")
+ .testMaterializedData()
+ .build())
+ .runSql(
+ "INSERT INTO sink "
+ + "SELECT u.user_id, u.name, o.order_id,
p.payment_id "
+ + "FROM Users u "
+ + "INNER JOIN Orders o ON u.user_id =
o.user_id "
+ + "INNER JOIN Payments p ON u.user_id =
p.user_id")
+ .build();
+
+ public static final TableTestProgram MULTI_JOIN_FOUR_WAY_COMPLEX =
+ TableTestProgram.of("four-way-complex-updating-join", "four way
complex updating join")
+
.setupConfig(OptimizerConfigOptions.TABLE_OPTIMIZER_MULTI_JOIN_ENABLED, true)
+ .setupTableSource(
+ SourceTestStep.newBuilder("Users")
+ .addSchema(
+ "user_id_0 STRING PRIMARY KEY NOT
ENFORCED",
+ "name STRING",
+ "cash INT")
+ .addOption("changelog-mode", "I,UA,D")
+ .producedValues(
+ Row.ofKind(RowKind.INSERT, "1",
"Gus", 100),
+ Row.ofKind(RowKind.INSERT, "2",
"Bob", 20),
+ Row.ofKind(RowKind.INSERT, "3",
"Nomad", 50),
+ Row.ofKind(RowKind.INSERT, "4",
"David", 5),
+ Row.ofKind(RowKind.INSERT, "5",
"Eve", 0),
+ Row.ofKind(RowKind.INSERT, "6",
"Frank", 70),
+ Row.ofKind(RowKind.INSERT, "7",
"Welcher", 100),
+ Row.ofKind(RowKind.INSERT, "8",
"Joe no order", 10),
+ Row.ofKind(RowKind.INSERT, "9",
"Charlie Smith", 50),
+ Row.ofKind(RowKind.DELETE, "2",
"Bob", 20),
+ Row.ofKind(
+ RowKind.UPDATE_AFTER,
+ "9",
+ "Charlie Taylor",
+ 50))
+ .build())
+ .setupTableSource(
+ SourceTestStep.newBuilder("Orders")
+ .addSchema(
+ "user_id_1 STRING",
+ "order_id STRING PRIMARY KEY NOT
ENFORCED",
+ "product STRING")
+ .addOption("changelog-mode", "I,D")
+ .producedValues(
+ Row.ofKind(RowKind.INSERT, "1",
"order0", "ProdB"),
+ Row.ofKind(RowKind.INSERT, "1",
"order1", "ProdA"),
+ Row.ofKind(RowKind.INSERT, "2",
"order2", "ProdB"),
+ Row.ofKind(RowKind.INSERT, "3",
"order3", "ProdC"),
+ Row.ofKind(RowKind.INSERT, "4",
"order4", "ProdD"),
+ Row.ofKind(RowKind.INSERT, "6",
"order6", "ProdF"),
+ Row.ofKind(RowKind.INSERT, "7",
"order7", "ProdG"),
+ Row.ofKind(RowKind.INSERT, "9",
"order9", "ProdA"))
+ .build())
+ .setupTableSource(
+ SourceTestStep.newBuilder("Payments")
+ .addSchema(
+ "user_id_2 STRING",
+ "payment_id STRING PRIMARY KEY NOT
ENFORCED",
+ "price INT")
+ .addOption("changelog-mode", "I")
+ .producedValues(
+ Row.ofKind(RowKind.INSERT, "1",
"payment1", 50),
+ Row.ofKind(RowKind.INSERT, "2",
"payment2", -5),
+ Row.ofKind(RowKind.INSERT, "3",
"payment3", 30),
+ Row.ofKind(RowKind.INSERT, "4",
"payment4", 40),
+ Row.ofKind(RowKind.INSERT, "5",
"payment5", -1),
+ Row.ofKind(RowKind.INSERT, "6",
"payment6", 60),
+ Row.ofKind(RowKind.INSERT, "9",
"payment9", 30))
+ .build())
+ .setupTableSource(
+ SourceTestStep.newBuilder("Shipments")
+ .addSchema("user_id_3 STRING", "location
STRING")
+ .addOption("changelog-mode", "I,UA,UB,D")
+ .producedValues(
+ Row.ofKind(RowKind.INSERT, "1",
"London"),
+ Row.ofKind(RowKind.INSERT, "2",
"Paris"),
+ Row.ofKind(RowKind.INSERT, "3",
"New York"),
+ Row.ofKind(RowKind.INSERT, "9",
"Melbourne"),
+ Row.ofKind(RowKind.INSERT, "10",
"Lost Shipment"),
+ Row.ofKind(RowKind.INSERT, "3",
"Brasília"))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(
+ "user_id STRING",
+ "name STRING",
+ "order_id STRING",
+ "payment_id STRING",
+ "location STRING")
+ .addOption("changelog-mode", "I,UA,UB,D")
+ .consumedValues(
+ "+I[1, Gus, order0, payment1,
London]",
+ "+I[1, Gus, order1, payment1,
London]",
+ // NO I[2, Bob..] here because Bob
was deleted
+ "+I[3, Nomad, order3, payment3,
New York]",
+ // NO I[4, David..] here because
David has only 5
+ // dollars, so this is not true
u.cash >= p.price
+ "+I[5, Eve, null, payment5,
null]", // Here because
+ // payment price as
+ // -1
+ // NO I[7, Welcher..] here because
Welcher has no
+ // payment
+ "+I[6, Frank, order6, payment6,
null]",
+ // NO I[8, Welcher..] here because
there is no order and
+ // no payment
+ "+I[9, Charlie Taylor, order9,
payment9, Melbourne]", // Charlie Taylor because the name was updated
+ "+I[3, Nomad, order3, payment3,
Brasília]" // New
Review Comment:
nit: I would suggest to put comments always above lines otherwise they are
hard to read after spotless
##########
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:
##########
@@ -0,0 +1,455 @@
+{
+ "flinkVersion" : "2.1",
+ "nodes" : [ {
+ "id" : 1,
+ "type" : "stream-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`Users`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "user_id_0",
+ "dataType" : "VARCHAR(2147483647) NOT NULL"
+ }, {
+ "name" : "name",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "cash",
+ "dataType" : "INT"
+ } ],
+ "watermarkSpecs" : [ ],
+ "primaryKey" : {
+ "name" : "PK_user_id_0",
+ "type" : "PRIMARY_KEY",
+ "columns" : [ "user_id_0" ]
+ }
+ },
+ "partitionKeys" : [ ]
+ }
+ }
+ },
+ "outputType" : "ROW<`user_id_0` VARCHAR(2147483647) NOT NULL, `name`
VARCHAR(2147483647), `cash` INT>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, Users]], fields=[user_id_0, name, cash])",
+ "inputProperties" : [ ]
+ }, {
+ "id" : 2,
+ "type" : "stream-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`user_id_0` VARCHAR(2147483647) NOT NULL, `name`
VARCHAR(2147483647), `cash` INT>",
+ "description" : "Exchange(distribution=[hash[user_id_0]])"
+ }, {
+ "id" : 3,
+ "type" : "stream-exec-changelog-normalize_1",
+ "configuration" : {
+ "table.exec.mini-batch.enabled" : "false",
+ "table.exec.mini-batch.size" : "-1"
+ },
+ "uniqueKeys" : [ 0 ],
+ "generateUpdateBefore" : true,
+ "state" : [ {
+ "index" : 0,
+ "ttl" : "0 ms",
+ "name" : "changelogNormalizeState"
+ } ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`user_id_0` VARCHAR(2147483647) NOT NULL, `name`
VARCHAR(2147483647), `cash` INT>",
+ "description" : "ChangelogNormalize(key=[user_id_0])"
+ }, {
+ "id" : 4,
+ "type" : "stream-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`Orders`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "user_id_1",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "order_id",
+ "dataType" : "VARCHAR(2147483647) NOT NULL"
+ }, {
+ "name" : "product",
+ "dataType" : "VARCHAR(2147483647)"
+ } ],
+ "watermarkSpecs" : [ ],
+ "primaryKey" : {
+ "name" : "PK_order_id",
+ "type" : "PRIMARY_KEY",
+ "columns" : [ "order_id" ]
+ }
+ },
+ "partitionKeys" : [ ]
+ }
+ }
+ },
+ "outputType" : "ROW<`user_id_1` VARCHAR(2147483647), `order_id`
VARCHAR(2147483647) NOT NULL, `product` VARCHAR(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, Orders]], fields=[user_id_1, order_id, product])",
+ "inputProperties" : [ ]
+ }, {
+ "id" : 5,
+ "type" : "stream-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`Payments`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "user_id_2",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "payment_id",
+ "dataType" : "VARCHAR(2147483647) NOT NULL"
+ }, {
+ "name" : "price",
+ "dataType" : "INT"
+ } ],
+ "watermarkSpecs" : [ ],
+ "primaryKey" : {
+ "name" : "PK_payment_id",
+ "type" : "PRIMARY_KEY",
+ "columns" : [ "payment_id" ]
+ }
+ },
+ "partitionKeys" : [ ]
+ }
+ }
+ },
+ "outputType" : "ROW<`user_id_2` VARCHAR(2147483647), `payment_id`
VARCHAR(2147483647) NOT NULL, `price` INT>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, Payments]], fields=[user_id_2, payment_id, price])",
+ "inputProperties" : [ ]
+ }, {
+ "id" : 6,
+ "type" : "stream-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`Shipments`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "user_id_3",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "location",
+ "dataType" : "VARCHAR(2147483647)"
+ } ],
+ "watermarkSpecs" : [ ]
+ },
+ "partitionKeys" : [ ]
+ }
+ }
+ },
+ "outputType" : "ROW<`user_id_3` VARCHAR(2147483647), `location`
VARCHAR(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, Shipments]], fields=[user_id_3, location])",
+ "inputProperties" : [ ]
+ }, {
+ "id" : 7,
+ "type" : "stream-exec-multi-join_1",
+ "joinTypes" : [ "INNER", "LEFT", "INNER", "LEFT" ],
+ "joinConditions" : [ {
+ "kind" : "LITERAL",
+ "value" : true,
+ "type" : "BOOLEAN NOT NULL"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$=$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "VARCHAR(2147483647) NOT NULL"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 3,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "type" : "BOOLEAN"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$AND$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$=$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "VARCHAR(2147483647) NOT NULL"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 6,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "type" : "BOOLEAN"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$OR$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$>=$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "INT"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 8,
+ "type" : "INT"
+ } ],
+ "type" : "BOOLEAN"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$<$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 8,
+ "type" : "INT"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 0,
+ "type" : "INT NOT NULL"
+ } ],
+ "type" : "BOOLEAN"
+ } ],
+ "type" : "BOOLEAN"
+ } ],
+ "type" : "BOOLEAN"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$=$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 6,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 9,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "type" : "BOOLEAN"
+ } ],
+ "multiJoinCondition" : null,
Review Comment:
we could also drop it like in https://github.com/apache/flink/pull/26683
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/MultiJoinTestPrograms.java:
##########
@@ -0,0 +1,461 @@
+/*
+ * 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.nodes.exec.stream;
+
+import org.apache.flink.table.api.config.OptimizerConfigOptions;
+import org.apache.flink.table.test.program.SinkTestStep;
+import org.apache.flink.table.test.program.SourceTestStep;
+import org.apache.flink.table.test.program.TableTestProgram;
+import org.apache.flink.types.Row;
+import org.apache.flink.types.RowKind;
+
+import static
org.apache.flink.table.planner.plan.nodes.exec.stream.MultiJoinTestUtils.ORDERS_SOURCE;
+import static
org.apache.flink.table.planner.plan.nodes.exec.stream.MultiJoinTestUtils.PAYMENTS_SOURCE;
+import static
org.apache.flink.table.planner.plan.nodes.exec.stream.MultiJoinTestUtils.USERS_SOURCE;
+
+/** {@link TableTestProgram} definitions for testing Multi-Join. */
+public class MultiJoinTestPrograms {
+ public static final TableTestProgram MULTI_JOIN_THREE_WAY_LEFT_OUTER_JOIN =
+ TableTestProgram.of("three-way-left-outer-join", "three way left
outer join")
+
.setupConfig(OptimizerConfigOptions.TABLE_OPTIMIZER_MULTI_JOIN_ENABLED, true)
+ .setupTableSource(USERS_SOURCE)
+ .setupTableSource(ORDERS_SOURCE)
+ .setupTableSource(PAYMENTS_SOURCE)
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(
+ "user_id STRING",
+ "name STRING",
+ "order_id STRING",
+ "payment_id STRING")
+ .consumedValues(
+ "+I[1, Gus, order1, payment1]",
+ "+I[2, Bob, order2, payment2]",
+ "+I[2, Bob, order3, payment2]",
+ "+I[1, Gus, order1, payment3]",
+ "+I[3, Alice, null, null]")
+ .testMaterializedData()
+ .build())
+ .runSql(
+ "INSERT INTO sink "
+ + "SELECT u.user_id, u.name, o.order_id,
p.payment_id "
+ + "FROM Users u "
+ + "LEFT JOIN Orders o ON u.user_id =
o.user_id "
+ + "LEFT JOIN Payments p ON u.user_id =
p.user_id")
+ .build();
+
+ public static final TableTestProgram
MULTI_JOIN_THREE_WAY_LEFT_OUTER_JOIN_UPDATING =
+ TableTestProgram.of(
+ "three-way-left-outer-join-updating",
+ "three way left outer join updating")
+
.setupConfig(OptimizerConfigOptions.TABLE_OPTIMIZER_MULTI_JOIN_ENABLED, true)
+ .setupTableSource(
+ SourceTestStep.newBuilder("Users")
+ .addSchema(
+ "user_id STRING PRIMARY KEY NOT
ENFORCED, name STRING, cash INT")
+ .producedValues(
+ Row.of("1", "Gus", 100),
Review Comment:
nit: to make it explicit here
```suggestion
Row.ofKind(INSERT, "1", "Gus",
100),
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
