gustavodemorais commented on code in PR #26687: URL: https://github.com/apache/flink/pull/26687#discussion_r2154340934
########## 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: I've shuffled rows and the columns for the four way complex join for restore and semantic test. Also shuffled columns for the rel/exec/explain tests 👍 https://github.com/apache/flink/pull/26687/commits/570da7bb0ae69f416ef84bfd431dc4a853b3b061 -- 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: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
