diff --git a/docs/content.zh/docs/concepts/sql-table-concepts/determinism.md b/docs/content.zh/docs/concepts/sql-table-concepts/determinism.md index 7f2b2ed79c3a53..aa145d5805f9e2 100644 --- a/docs/content.zh/docs/concepts/sql-table-concepts/determinism.md +++ b/docs/content.zh/docs/concepts/sql-table-concepts/determinism.md @@ -229,7 +229,7 @@ join dim_with_pk for system_time as of t1.proctime as t2 -- 执行计划：声明了 pk 后的维表，通过 pk 连接时左流的 upsertKey 属性得以保留，从而节省了高开销的物化节点 Sink(table=[default_catalog.default_database.sink_with_pk], fields=[a, b, c]) +- Calc(select=[a, b, c]) - +- LookupJoin(table=[default_catalog.default_database.dim_with_pk], joinType=[InnerJoin], lookup=[a=a], select=[a, b, a, c]) + +- LookupJoin(table=[default_catalog.default_database.dim_with_pk, project=[a, c], metadata=[]], joinType=[InnerJoin], lookup=[a=a], select=[a, b, a, c]) +- DropUpdateBefore +- TableSourceScan(table=[[default_catalog, default_database, cdc, project=[a, b], metadata=[]]], fields=[a, b]) ``` @@ -246,7 +246,7 @@ join dim_without_pk for system_time as of t1.proctime as t2 -- 不启用 `TRY_RESOLVE` 模式时在运行时可能产生错误，当启用 `TRY_RESOLVE` 时的执行计划 Sink(table=[default_catalog.default_database.sink_with_pk], fields=[a, b, c], upsertMaterialize=[true]) +- Calc(select=[a, b, c]) - +- LookupJoin(table=[default_catalog.default_database.dim_without_pk], joinType=[InnerJoin], lookup=[a=a], select=[a, b, a, c], upsertMaterialize=[true]) + +- LookupJoin(table=[default_catalog.default_database.dim_without_pk, project=[a, c], metadata=[]], joinType=[InnerJoin], lookup=[a=a], select=[a, b, a, c], upsertMaterialize=[true]) +- TableSourceScan(table=[[default_catalog, default_database, cdc, project=[a, b], metadata=[]]], fields=[a, b]) ``` 尽管第二个查询可以通过启用 `TRY_RESOLVE` 选项（增加物化）来解决正确性问题，但成本高昂，与声明了主键的第一个查询相比，会多出两个更昂贵的物化操作。 diff --git a/docs/content/docs/concepts/sql-table-concepts/determinism.md b/docs/content/docs/concepts/sql-table-concepts/determinism.md index f096f32d1e68b0..3b3afbabf1c8a6 100644 --- a/docs/content/docs/concepts/sql-table-concepts/determinism.md +++ b/docs/content/docs/concepts/sql-table-concepts/determinism.md @@ -240,7 +240,7 @@ join dim_with_pk for system_time as of t1.proctime as t2 -- plan: the upsertKey of left stream is reserved when lookup table with a pk definition and use it as lookup key, so that the high cost materialization can be omitted. Sink(table=[default_catalog.default_database.sink_with_pk], fields=[a, b, c]) +- Calc(select=[a, b, c]) - +- LookupJoin(table=[default_catalog.default_database.dim_with_pk], joinType=[InnerJoin], lookup=[a=a], select=[a, b, a, c]) + +- LookupJoin(table=[default_catalog.default_database.dim_with_pk, project=[a, c], metadata=[]], joinType=[InnerJoin], lookup=[a=a], select=[a, b, a, c]) +- DropUpdateBefore +- TableSourceScan(table=[[default_catalog, default_database, cdc, project=[a, b], metadata=[]]], fields=[a, b]) ``` @@ -257,7 +257,7 @@ join dim_without_pk for system_time as of t1.proctime as t2 -- execution plan when `TRY_RESOLVE` is enabled(may encounter errors at runtime when `TRY_RESOLVE` mode is not enabled): Sink(table=[default_catalog.default_database.sink_with_pk], fields=[a, b, c], upsertMaterialize=[true]) +- Calc(select=[a, b, c]) - +- LookupJoin(table=[default_catalog.default_database.dim_without_pk], joinType=[InnerJoin], lookup=[a=a], select=[a, b, a, c], upsertMaterialize=[true]) + +- LookupJoin(table=[default_catalog.default_database.dim_without_pk, project=[a, c], metadata=[]], joinType=[InnerJoin], lookup=[a=a], select=[a, b, a, c], upsertMaterialize=[true]) +- TableSourceScan(table=[[default_catalog, default_database, cdc, project=[a, b], metadata=[]]], fields=[a, b]) ``` Though the second case can be solved by adding materialization if `TRY_RESOLVE` is enabled, but the cost is very high, there will be two more costly materialization compared to the one with primary key. diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/common/CommonPhysicalLookupJoin.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/common/CommonPhysicalLookupJoin.scala index df0cf6702a0844..b1173eba72ae00 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/common/CommonPhysicalLookupJoin.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/common/CommonPhysicalLookupJoin.scala @@ -18,7 +18,7 @@ package org.apache.flink.table.planner.plan.nodes.physical.common import org.apache.flink.table.api.{TableConfig, TableException} -import org.apache.flink.table.catalog.{ObjectIdentifier, UniqueConstraint} +import org.apache.flink.table.catalog.UniqueConstraint import org.apache.flink.table.connector.ChangelogMode import org.apache.flink.table.planner.calcite.FlinkTypeFactory import org.apache.flink.table.planner.plan.nodes.FlinkRelNode @@ -187,14 +187,18 @@ abstract class CommonPhysicalLookupJoin( case None => resultFieldNames.mkString(", ") } - val tableIdentifier: ObjectIdentifier = temporalTable match { - case t: TableSourceTable => t.contextResolvedTable.getIdentifier - case t: LegacyTableSourceTable[_] => t.tableIdentifier + // Two lookup joins on the same table with different push-downs are different operators. The + // scan gets the spec digests from RelOptTable#getQualifiedName; the lookup join has to add them. + val tableDigest: String = temporalTable match { + case t: TableSourceTable => + val specDigests = t.getSpecDigests.asScala.toSeq + (t.contextResolvedTable.getIdentifier.asSummaryString +: specDigests).mkString(", ") + case t: LegacyTableSourceTable[_] => t.tableIdentifier.asSummaryString } super .explainTerms(pw) - .item("table", tableIdentifier.asSummaryString()) + .item("table", tableDigest) .item("joinType", JoinTypeUtil.getFlinkJoinType(joinType)) .item("lookup", lookupKeys) .itemIf("where", whereString, whereString.nonEmpty) diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/LookupJoinSemanticTestPrograms.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/LookupJoinSemanticTestPrograms.java new file mode 100644 index 00000000000000..03a68885727fc9 --- /dev/null +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/LookupJoinSemanticTestPrograms.java @@ -0,0 +1,109 @@ +/* + * 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.common; + +import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecLookupJoin; +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 static org.apache.flink.table.planner.plan.nodes.exec.common.LookupJoinTestPrograms.CUSTOMERS_BEFORE_DATA; +import static org.apache.flink.table.planner.plan.nodes.exec.common.LookupJoinTestPrograms.CUSTOMERS_SCHEMA; +import static org.apache.flink.table.planner.plan.nodes.exec.common.LookupJoinTestPrograms.ORDERS_BEFORE_DATA; +import static org.apache.flink.table.planner.plan.nodes.exec.common.LookupJoinTestPrograms.ORDERS_SCHEMA; +import static org.apache.flink.table.planner.plan.nodes.exec.common.LookupJoinTestPrograms.SINK_SCHEMA; + +/** {@link TableTestProgram} definitions for semantic testing {@link StreamExecLookupJoin}. */ +public class LookupJoinSemanticTestPrograms { + + /** A semantic test cannot take a source that carries after-restore data. */ + static final SourceTestStep CUSTOMERS = + SourceTestStep.newBuilder("customers_t") + .addOption("disable-lookup", "false") // static/lookup table + .addOption("filterable-fields", "age") + .addSchema(CUSTOMERS_SCHEMA) + .producedValues(CUSTOMERS_BEFORE_DATA) + .build(); + + static final SourceTestStep ORDERS = + SourceTestStep.newBuilder("orders_t") + .addOption("filterable-fields", "customer_id") + .addSchema(ORDERS_SCHEMA) + .producedValues(ORDERS_BEFORE_DATA) + .build(); + + private static String filteredLookupJoin(String filter) { + return "SELECT " + + "O.order_id, " + + "O.total, " + + "C.id, " + + "C.name, " + + "C.age, " + + "C.city, " + + "C.state, " + + "C.zipcode " + + "FROM orders_t as O " + + "JOIN customers_t FOR SYSTEM_TIME AS OF O.proc_time AS C " + + "ON O.customer_id = C.id AND " + + filter; + } + + /** + * Both branches select the same columns from the same dim table and differ only in the filter + * pushed into it, so the two lookup joins are indistinguishable unless the pushed-down filter + * is part of the lookup join's digest. See FLINK-36808. + */ + private static String unionOfTwoFilteredLookupJoins(String firstFilter, String secondFilter) { + return "INSERT INTO sink_t " + + filteredLookupJoin(firstFilter) + + " UNION ALL " + + filteredLookupJoin(secondFilter); + } + + private static SinkTestStep sink() { + return SinkTestStep.newBuilder("sink_t") + .addSchema(SINK_SCHEMA) + .consumedValues( + "+I[1, 44.44, 3, Claire, 37, Austin, Texas, 73301]", + "+I[2, 100.02, 5, Jake, 42, New York City, New York, 10001]", + "+I[4, 92.61, 2, Alice, 32, San Francisco, California, 95016]", + "+I[5, 12.78, 2, Alice, 32, San Francisco, California, 95016]") + .build(); + } + + public static final TableTestProgram LOOKUP_JOIN_UNION_DIFFERENT_FILTERS = + TableTestProgram.of( + "lookup-join-union-different-filters", + "validates two lookup joins on the same table with different pushed-down filters are not merged, with the matching filter first") + .setupTableSource(CUSTOMERS) + .setupTableSource(ORDERS) + .setupTableSink(sink()) + .runSql(unionOfTwoFilteredLookupJoins("C.age > 30", "C.age > 100")) + .build(); + + public static final TableTestProgram LOOKUP_JOIN_UNION_DIFFERENT_FILTERS_REVERSED = + TableTestProgram.of( + "lookup-join-union-different-filters-reversed", + "validates the same with the non-matching filter first, where a wrong merge returns nothing at all") + .setupTableSource(CUSTOMERS) + .setupTableSource(ORDERS) + .setupTableSink(sink()) + .runSql(unionOfTwoFilteredLookupJoins("C.age > 100", "C.age > 30")) + .build(); +} diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/LookupJoinTestPrograms.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/LookupJoinTestPrograms.java index 64f594a5d4d81f..fc6dc64625c76a 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/LookupJoinTestPrograms.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/LookupJoinTestPrograms.java @@ -84,22 +84,30 @@ public class LookupJoinTestPrograms { .producedAfterRestore(CUSTOMERS_AFTER_DATA) .build(); + static final String[] ORDERS_SCHEMA = + new String[] { + "order_id INT", + "customer_id INT", + "total DOUBLE", + "order_time STRING", + "proc_time AS PROCTIME()" + }; + + static final Row[] ORDERS_BEFORE_DATA = + new Row[] { + Row.of(1, 3, 44.44, "2020-10-10 00:00:01"), + Row.of(2, 5, 100.02, "2020-10-10 00:00:02"), + Row.of(4, 2, 92.61, "2020-10-10 00:00:04"), + Row.of(3, 1, 23.89, "2020-10-10 00:00:03"), + Row.of(6, 4, 7.65, "2020-10-10 00:00:06"), + Row.of(5, 2, 12.78, "2020-10-10 00:00:05") + }; + static final SourceTestStep ORDERS = SourceTestStep.newBuilder("orders_t") .addOption("filterable-fields", "customer_id") - .addSchema( - "order_id INT", - "customer_id INT", - "total DOUBLE", - "order_time STRING", - "proc_time AS PROCTIME()") - .producedBeforeRestore( - Row.of(1, 3, 44.44, "2020-10-10 00:00:01"), - Row.of(2, 5, 100.02, "2020-10-10 00:00:02"), - Row.of(4, 2, 92.61, "2020-10-10 00:00:04"), - Row.of(3, 1, 23.89, "2020-10-10 00:00:03"), - Row.of(6, 4, 7.65, "2020-10-10 00:00:06"), - Row.of(5, 2, 12.78, "2020-10-10 00:00:05")) + .addSchema(ORDERS_SCHEMA) + .producedBeforeRestore(ORDERS_BEFORE_DATA) .producedAfterRestore( Row.of(7, 6, 17.58, "2020-10-10 00:00:07"), // new customer Row.of(9, 1, 143.21, "2020-10-10 00:00:08") // updated zip code @@ -110,20 +118,9 @@ public class LookupJoinTestPrograms { SourceTestStep.newBuilder("orders_cdc_t") .addOption("filterable-fields", "customer_id") .addOption("changelog-mode", "I,UA,UB,D") - .addSchema( - "order_id INT", - "customer_id INT", - "total DOUBLE", - "order_time STRING", - "proc_time AS PROCTIME()") + .addSchema(ORDERS_SCHEMA) .addSchema("PRIMARY KEY (order_id) NOT ENFORCED") - .producedBeforeRestore( - Row.of(1, 3, 44.44, "2020-10-10 00:00:01"), - Row.of(2, 5, 100.02, "2020-10-10 00:00:02"), - Row.of(4, 2, 92.61, "2020-10-10 00:00:04"), - Row.of(3, 1, 23.89, "2020-10-10 00:00:03"), - Row.of(6, 4, 7.65, "2020-10-10 00:00:06"), - Row.of(5, 2, 12.78, "2020-10-10 00:00:05")) + .producedBeforeRestore(ORDERS_BEFORE_DATA) .producedAfterRestore( Row.ofKind(RowKind.DELETE, 3, 1, 23.89, "2020-10-10 00:00:03"), Row.ofKind(RowKind.INSERT, 3, 1, 33.01, "2020-10-10 01:01:06"), diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinSemanticTests.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinSemanticTests.java new file mode 100644 index 00000000000000..7b45ab870a4e99 --- /dev/null +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LookupJoinSemanticTests.java @@ -0,0 +1,36 @@ +/* + * 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.common.LookupJoinSemanticTestPrograms; +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 {@link StreamExecLookupJoin}. */ +public class LookupJoinSemanticTests extends SemanticTestBase { + + @Override + public List programs() { + return List.of( + LookupJoinSemanticTestPrograms.LOOKUP_JOIN_UNION_DIFFERENT_FILTERS, + LookupJoinSemanticTestPrograms.LOOKUP_JOIN_UNION_DIFFERENT_FILTERS_REVERSED); + } +} diff --git a/flink-table/flink-table-planner/src/test/resources/explain/stream/join/lookup/testAggAndAllConstantLookupKeyWithTryResolveMode.out b/flink-table/flink-table-planner/src/test/resources/explain/stream/join/lookup/testAggAndAllConstantLookupKeyWithTryResolveMode.out index 6cd5bf8dc23af3..ab4324720817a1 100644 --- a/flink-table/flink-table-planner/src/test/resources/explain/stream/join/lookup/testAggAndAllConstantLookupKeyWithTryResolveMode.out +++ b/flink-table/flink-table-planner/src/test/resources/explain/stream/join/lookup/testAggAndAllConstantLookupKeyWithTryResolveMode.out @@ -12,7 +12,7 @@ LogicalSink(table=[default_catalog.default_database.Sink1], fields=[a, name, age == Optimized Physical Plan == Sink(table=[default_catalog.default_database.Sink1], fields=[a, name, age]) -+- LookupJoin(table=[default_catalog.default_database.LookupTable], joinType=[LeftOuterJoin], lookup=[id=100], where=[=(id, 100)], select=[a, name, age], upsertMaterialize=[true]) ++- LookupJoin(table=[default_catalog.default_database.LookupTable, filter=[]], joinType=[LeftOuterJoin], lookup=[id=100], where=[=(id, 100)], select=[a, name, age], upsertMaterialize=[true]) +- Calc(select=[a]) +- GroupAggregate(groupBy=[b], select=[b, MAX(a) AS a]) +- Exchange(distribution=[hash[b]]) @@ -21,7 +21,7 @@ Sink(table=[default_catalog.default_database.Sink1], fields=[a, name, age]) == Optimized Execution Plan == Sink(table=[default_catalog.default_database.Sink1], fields=[a, name, age]) -+- LookupJoin(table=[default_catalog.default_database.LookupTable], joinType=[LeftOuterJoin], lookup=[id=100], where=[(id = 100)], select=[a, name, age], upsertMaterialize=[true]) ++- LookupJoin(table=[default_catalog.default_database.LookupTable, filter=[]], joinType=[LeftOuterJoin], lookup=[id=100], where=[(id = 100)], select=[a, name, age], upsertMaterialize=[true]) +- Calc(select=[a]) +- GroupAggregate(groupBy=[b], select=[b, MAX(a) AS a]) +- Exchange(distribution=[hash[b]]) @@ -84,7 +84,7 @@ Sink(table=[default_catalog.default_database.Sink1], fields=[a, name, age]) "id" : 10, "type" : "LookupJoin[6]", "pact" : "Operator", - "contents" : "[6]:LookupJoin(table=[default_catalog.default_database.LookupTable], joinType=[LeftOuterJoin], lookup=[id=100], where=[(id = 100)], select=[a, name, age], upsertMaterialize=[true])", + "contents" : "[6]:LookupJoin(table=[default_catalog.default_database.LookupTable, filter=[]], joinType=[LeftOuterJoin], lookup=[id=100], where=[(id = 100)], select=[a, name, age], upsertMaterialize=[true])", "parallelism" : 4, "predecessors" : [ { "id" : 8, diff --git a/flink-table/flink-table-planner/src/test/resources/explain/stream/join/lookup/testAggAndAllConstantLookupKeyWithTryResolveMode_newSource.out b/flink-table/flink-table-planner/src/test/resources/explain/stream/join/lookup/testAggAndAllConstantLookupKeyWithTryResolveMode_newSource.out index 73ba8a95ee8b2f..b01b38024cdc52 100644 --- a/flink-table/flink-table-planner/src/test/resources/explain/stream/join/lookup/testAggAndAllConstantLookupKeyWithTryResolveMode_newSource.out +++ b/flink-table/flink-table-planner/src/test/resources/explain/stream/join/lookup/testAggAndAllConstantLookupKeyWithTryResolveMode_newSource.out @@ -12,7 +12,7 @@ LogicalSink(table=[default_catalog.default_database.Sink1], fields=[a, name, age == Optimized Physical Plan == Sink(table=[default_catalog.default_database.Sink1], fields=[a, name, age]) -+- LookupJoin(table=[default_catalog.default_database.LookupTable], joinType=[LeftOuterJoin], lookup=[id=100], where=[=(id, 100)], select=[a, name, age], upsertMaterialize=[true]) ++- LookupJoin(table=[default_catalog.default_database.LookupTable, filter=[]], joinType=[LeftOuterJoin], lookup=[id=100], where=[=(id, 100)], select=[a, name, age], upsertMaterialize=[true]) +- Calc(select=[a]) +- GroupAggregate(groupBy=[b], select=[b, MAX(a) AS a]) +- Exchange(distribution=[hash[b]]) @@ -21,7 +21,7 @@ Sink(table=[default_catalog.default_database.Sink1], fields=[a, name, age]) == Optimized Execution Plan == Sink(table=[default_catalog.default_database.Sink1], fields=[a, name, age]) -+- LookupJoin(table=[default_catalog.default_database.LookupTable], joinType=[LeftOuterJoin], lookup=[id=100], where=[(id = 100)], select=[a, name, age], upsertMaterialize=[true]) ++- LookupJoin(table=[default_catalog.default_database.LookupTable, filter=[]], joinType=[LeftOuterJoin], lookup=[id=100], where=[(id = 100)], select=[a, name, age], upsertMaterialize=[true]) +- Calc(select=[a]) +- GroupAggregate(groupBy=[b], select=[b, MAX(a) AS a]) +- Exchange(distribution=[hash[b]]) @@ -84,7 +84,7 @@ Sink(table=[default_catalog.default_database.Sink1], fields=[a, name, age]) "id" : , "type" : "LookupJoin[]", "pact" : "Operator", - "contents" : "[]:LookupJoin(table=[default_catalog.default_database.LookupTable], joinType=[LeftOuterJoin], lookup=[id=100], where=[(id = 100)], select=[a, name, age], upsertMaterialize=[true])", + "contents" : "[]:LookupJoin(table=[default_catalog.default_database.LookupTable, filter=[]], joinType=[LeftOuterJoin], lookup=[id=100], where=[(id = 100)], select=[a, name, age], upsertMaterialize=[true])", "parallelism" : 4, "predecessors" : [ { "id" : , diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/analyze/NonDeterministicUpdateAnalyzerTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/analyze/NonDeterministicUpdateAnalyzerTest.xml index 72fc5bbff3248d..5800cf232e462e 100644 --- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/analyze/NonDeterministicUpdateAnalyzerTest.xml +++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/analyze/NonDeterministicUpdateAnalyzerTest.xml @@ -29,12 +29,12 @@ on t1.a = t2.a and ndFunc(t2.b) > 100]]> (ndFunc(b), 100)], select=[a, b, c, a], upsertKey=[[0]]) + +- LookupJoin(table=[default_catalog.default_database.dim_with_pk, project=[a, b], metadata=[], filter=[]], joinType=[InnerJoin], lookup=[a=a], where=[>(ndFunc(b), 100)], select=[a, b, c, a], upsertKey=[[0]]) +- TableSourceScan(table=[[default_catalog, default_database, cdc, project=[a, b, c], metadata=[]]], fields=[a, b, c]) advice[1]: [WARNING] There exists non deterministic function: 'ndFunc' in condition: '>(ndFunc($1), 100)' which may cause wrong result in update pipeline. related rel plan: -LookupJoin(table=[default_catalog.default_database.dim_with_pk], joinType=[InnerJoin], lookup=[a=a], where=[>(ndFunc(b), 100)], select=[a, b, c, a], upsertKey=[[0]], changelogMode=[I,UB,UA,D]) +LookupJoin(table=[default_catalog.default_database.dim_with_pk, project=[a, b], metadata=[], filter=[]], joinType=[InnerJoin], lookup=[a=a], where=[>(ndFunc(b), 100)], select=[a, b, c, a], upsertKey=[[0]], changelogMode=[I,UB,UA,D]) +- TableSourceScan(table=[[default_catalog, default_database, cdc, project=[a, b, c], metadata=[]]], fields=[a, b, c], changelogMode=[I,UB,UA,D], upsertKeys=[[a]]) @@ -54,7 +54,7 @@ LookupJoin(table=[default_catalog.default_database.dim_with_pk], joinType=[Inner diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/LookupJoinTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/LookupJoinTest.xml index 9195b47ebee37e..9916356cdff8b4 100644 --- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/LookupJoinTest.xml +++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/LookupJoinTest.xml @@ -54,7 +54,7 @@ HashAggregate(isMerge=[true], groupBy=[b], select=[b, Final_COUNT(count$0) AS EX +- Exchange(distribution=[hash[b]]) +- LocalHashAggregate(groupBy=[b], select=[b, Partial_COUNT(a) AS count$0, Partial_SUM(c) AS sum$1, Partial_SUM(d) AS sum$2]) +- Calc(select=[b, a, c, d]) - +- LookupJoin(table=[default_catalog.default_database.LookupTable], joinType=[InnerJoin], lookup=[id=a], where=[(age > 10)], select=[b, a, c, d, id]) + +- LookupJoin(table=[default_catalog.default_database.LookupTable, project=[id, age], metadata=[]], joinType=[InnerJoin], lookup=[id=a], where=[(age > 10)], select=[b, a, c, d, id]) +- Calc(select=[b, a, c, d]) +- HashAggregate(isMerge=[true], groupBy=[a, b], select=[a, b, Final_SUM(sum$0) AS c, Final_SUM(sum$1) AS d]) +- Exchange(distribution=[hash[a, b]]) @@ -409,7 +409,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2], proctime=[$3], id=[$4]) @@ -708,7 +708,7 @@ Calc(select=[EXPR$0, EXPR$1, EXPR$2]) +- LocalHashAggregate(groupBy=[b, b0], select=[b, b0, Partial_COUNT(a) AS count$0, Partial_COUNT(id) AS count$1, Partial_SUM(a0) AS sum$2]) +- HashJoin(joinType=[InnerJoin], where=[(a = a0)], select=[b, a, id, a0, b0], build=[right]) :- Exchange(distribution=[hash[a]]) - : +- LookupJoin(table=[default_catalog.default_database.LookupTable], joinType=[InnerJoin], lookup=[id=a], where=[(age > 10)], select=[b, a, id])(reuse_id=[1]) + : +- LookupJoin(table=[default_catalog.default_database.LookupTable, project=[id, age], metadata=[]], joinType=[InnerJoin], lookup=[id=a], where=[(age > 10)], select=[b, a, id])(reuse_id=[1]) : +- Calc(select=[b, a]) : +- HashAggregate(isMerge=[true], groupBy=[a, b], select=[a, b]) : +- Exchange(distribution=[hash[a, b]]) diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/stream/DuplicateChangesInferRuleTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/stream/DuplicateChangesInferRuleTest.xml index b7b793aec3243b..3159279ebf1a07 100644 --- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/stream/DuplicateChangesInferRuleTest.xml +++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/stream/DuplicateChangesInferRuleTest.xml @@ -514,7 +514,7 @@ LogicalSink(table=[default_catalog.default_database.no_pk_snk], fields=[a, b, c] diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/NonDeterministicDagTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/NonDeterministicDagTest.xml index 568139a28ece48..a079ae342303fb 100644 --- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/NonDeterministicDagTest.xml +++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/NonDeterministicDagTest.xml @@ -570,7 +570,7 @@ LogicalSink(table=[default_catalog.default_database.sink_without_pk], fields=[a, @@ -605,7 +605,7 @@ LogicalSink(table=[default_catalog.default_database.sink_with_pk], fields=[a, ve (UNIX_TIMESTAMP() - 300))], select=[a, a, b, c], upsertKey=[[0]]) + +- LookupJoin(table=[default_catalog.default_database.dim_with_pk, filter=[]], joinType=[InnerJoin], lookup=[a=a], where=[(b > (UNIX_TIMESTAMP() - 300))], select=[a, a, b, c], upsertKey=[[0]]) +- TableSourceScan(table=[[default_catalog, default_database, cdc, project=[a], metadata=[]]], fields=[a]) ]]> @@ -637,7 +637,7 @@ LogicalSink(table=[default_catalog.default_database.sink_without_pk], fields=[a, 100)], select=[a, b, c, a], upsertKey=[[0]]) + +- LookupJoin(table=[default_catalog.default_database.dim_with_pk, project=[a, b], metadata=[], filter=[]], joinType=[InnerJoin], lookup=[a=a], where=[(ndFunc(b) > 100)], select=[a, b, c, a], upsertKey=[[0]]) +- TableSourceScan(table=[[default_catalog, default_database, cdc, project=[a, b, c], metadata=[]]], fields=[a, b, c]) ]]> @@ -669,7 +669,7 @@ LogicalSink(table=[default_catalog.default_database.sink_without_pk], fields=[a, @@ -701,7 +701,7 @@ LogicalSink(table=[default_catalog.default_database.sink_without_pk], fields=[a, @@ -733,7 +733,7 @@ LogicalSink(table=[default_catalog.default_database.sink_without_pk], fields=[a, @@ -765,7 +765,7 @@ LogicalSink(table=[default_catalog.default_database.sink_without_pk], fields=[a, @@ -797,7 +797,7 @@ LogicalSink(table=[default_catalog.default_database.sink_with_pk], fields=[a, b, @@ -830,7 +830,7 @@ LogicalSink(table=[default_catalog.default_database.sink_with_pk], fields=[a, b, @@ -863,7 +863,7 @@ LogicalSink(table=[default_catalog.default_database.sink_without_pk], fields=[a, @@ -895,7 +895,7 @@ LogicalSink(table=[default_catalog.default_database.sink_without_pk], fields=[a, @@ -928,7 +928,7 @@ LogicalSink(table=[default_catalog.default_database.sink_with_pk], fields=[a, b, @@ -961,7 +961,7 @@ LogicalSink(table=[default_catalog.default_database.sink_with_pk], fields=[a, b, @@ -993,7 +993,7 @@ LogicalSink(table=[default_catalog.default_database.sink_without_pk], fields=[a, @@ -1025,7 +1025,7 @@ LogicalSink(table=[default_catalog.default_database.sink_without_pk], fields=[a, @@ -1165,7 +1165,7 @@ LogicalSink(table=[default_catalog.default_database.sink_with_pk], fields=[a, b, @@ -1198,7 +1198,7 @@ LogicalSink(table=[default_catalog.default_database.sink_with_pk], fields=[a, b, @@ -1231,7 +1231,7 @@ LogicalSink(table=[default_catalog.default_database.sink_without_pk], fields=[a, @@ -1263,7 +1263,7 @@ LogicalSink(table=[default_catalog.default_database.sink_without_pk], fields=[a, diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/RankTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/RankTest.xml index f27964e04e95d8..42e22947921aad 100644 --- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/RankTest.xml +++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/RankTest.xml @@ -1284,7 +1284,7 @@ Rank(strategy=[UpdateFastStrategy[0]], rankType=[ROW_NUMBER], rankRange=[rankSta +- GroupAggregate(groupBy=[name], select=[name, SUM(id) FILTER $f2 AS ids]) +- Exchange(distribution=[hash[name]]) +- Calc(select=[name, id, IS TRUE(>(id, 0)) AS $f2]) - +- LookupJoin(table=[default_catalog.default_database.LookupTable], joinType=[InnerJoin], lookup=[id=a], select=[a, id, name]) + +- LookupJoin(table=[default_catalog.default_database.LookupTable, project=[id, name], metadata=[]], joinType=[InnerJoin], lookup=[id=a], select=[a, id, name]) +- Calc(select=[a]) +- DataStreamScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, proctime, rowtime]) ]]> diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.xml index ba5669b8459127..ba3a0667556f4a 100644 --- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.xml +++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.xml @@ -268,7 +268,7 @@ LogicalAggregate(group=[{0}], EXPR$1=[COUNT($1)], EXPR$2=[SUM($2)], EXPR$3=[SUM( GroupAggregate(groupBy=[b], select=[b, COUNT_RETRACT(a) AS EXPR$1, SUM_RETRACT(c) AS EXPR$2, SUM_RETRACT(d) AS EXPR$3]) +- Exchange(distribution=[hash[b]]) +- Calc(select=[b, a, c, d]) - +- LookupJoin(table=[default_catalog.default_database.LookupTable], joinType=[InnerJoin], lookup=[id=a], where=[(age > 10)], select=[b, a, c, d, id], upsertKey=[[0, 1]]) + +- LookupJoin(table=[default_catalog.default_database.LookupTable, project=[id, age], metadata=[], filter=[]], joinType=[InnerJoin], lookup=[id=a], where=[(age > 10)], select=[b, a, c, d, id], upsertKey=[[0, 1]]) +- Calc(select=[b, a, c, d]) +- GroupAggregate(groupBy=[a, b], select=[a, b, SUM(c) AS c, SUM(d) AS d]) +- Exchange(distribution=[hash[a, b]]) @@ -368,6 +368,102 @@ LogicalProject(a=[$0], b=[$1], c=[$2], proctime=[$3], rowtime=[$4], id=[$5], nam Calc(select=[a, b, c, PROCTIME_MATERIALIZE(proctime) AS proctime, rowtime, id, name, age]) +- LookupJoin(table=[default_catalog.default_database.AsyncLookupTable], joinType=[InnerJoin], lookup=[id=a], select=[a, b, c, proctime, rowtime, id, name, age], async=[ORDERED, KEY_ORDERED: false, 180000ms, 100]) +- DataStreamScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, proctime, rowtime]) +]]> + + + + + + + + + + + + + + + + + + + + + + @@ -403,9 +499,9 @@ LogicalProject(a=[$0]) @@ -524,7 +620,7 @@ LogicalProject(b=[$1]) @@ -553,7 +649,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2], proctime=[$3], rowtime=[$4], id=[$5], nam 1000))], select=[a, b, c, proctime, rowtime, id, name]) ++- LookupJoin(table=[default_catalog.default_database.LookupTable, filter=[]], joinType=[InnerJoin], lookup=[age=10, id=a], where=[((age = 10) AND (CAST(name AS BIGINT) > 1000))], select=[a, b, c, proctime, rowtime, id, name]) +- DataStreamScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, proctime, rowtime]) ]]> @@ -612,7 +708,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2], name=[$6], age=[$7], nominal_age=[$8]) 12)], select=[a, b, c, id, name, age, (age + 1) AS nominal_age]) ++- LookupJoin(table=[default_catalog.default_database.LookupTableWithComputedColumn, filter=[]], joinType=[InnerJoin], lookup=[id=a], where=[((age + 1) > 12)], select=[a, b, c, id, name, age, (age + 1) AS nominal_age]) +- Calc(select=[a, b, c]) +- DataStreamScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, proctime, rowtime]) ]]> @@ -641,7 +737,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2], proctime=[$3], rowtime=[$4], id=[$5], nam 1000)]) +- DataStreamScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, proctime, rowtime]) ]]> @@ -668,7 +764,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2], proctime=[$3], rowtime=[$4], id=[$5], nam @@ -696,7 +792,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2], proctime=[$3], rowtime=[$4], id=[$5], nam @@ -745,7 +841,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2], proctime=[$3], rowtime=[$4], id=[$5], nam @@ -773,7 +869,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2], proctime=[$3], rowtime=[$4], id=[$5], nam 1000)]) +- DataStreamScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c, proctime, rowtime]) ]]> @@ -1037,7 +1133,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2], proctime=[$3], rowtime=[$4], id=[$5]) @@ -1196,7 +1292,7 @@ LogicalProject(a=[$0], b=[$1], c=[$2], name=[$6]) @@ -1412,7 +1508,7 @@ LogicalProject(a=[$0]) diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.scala index b143542a3cb5b6..f398eb03365171 100644 --- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.scala +++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.scala @@ -89,6 +89,19 @@ class LookupJoinTest extends TableTestBase with Serializable { | 'connector' = 'values' |) |""".stripMargin) + // 'status' must be listed in 'filterable-fields': the bug under test only appears when the + // dim-side predicate is *fully* consumed by the source. If any of it remained, the two lookup + // joins would differ in their 'where' item and would never have been merged in the first place. + util.addTable(""" + |CREATE TABLE LookupTableWithFilterableFields ( + | `id` INT, + | `name` STRING, + | `status` STRING + |) WITH ( + | 'connector' = 'values', + | 'filterable-fields' = 'id;status' + |) + |""".stripMargin) util.addTable(""" |CREATE TABLE AsyncLookupTable ( | `id` INT, @@ -366,6 +379,52 @@ class LookupJoinTest extends TableTestBase with Serializable { util.verifyExecPlan(sql) } + @Test + def testJoinFilterableTemporalTableWithUnion(): Unit = { + // FLINK-36808: both branches push a different filter into the same dim table. The two lookup + // joins must stay separate operators; if they are merged, one branch's rows are emitted under + // the other branch's literal. + val sql = + """ + |SELECT s.a, s.b, s.proctime, d.status + |FROM MyTable AS s + |JOIN LookupTableWithFilterableFields FOR SYSTEM_TIME AS OF s.proctime AS d + |ON s.a = d.id + |WHERE d.status = 'OK' + |UNION ALL + |SELECT s.a, s.b, s.proctime, d.status + |FROM MyTable AS s + |JOIN LookupTableWithFilterableFields FOR SYSTEM_TIME AS OF s.proctime AS d + |ON s.a = d.id + |WHERE d.status = 'KO' + """.stripMargin + + util.verifyExecPlan(sql) + } + + @Test + def testJoinFilterableTemporalTableWithUnionSameFilter(): Unit = { + // Counterpart to testJoinFilterableTemporalTableWithUnion: with the same filter on both sides + // the two lookup joins really are equivalent and must still be reused. Guards against a fix + // that simply makes every lookup join digest unique. + val sql = + """ + |SELECT s.a, s.b, s.proctime, d.status + |FROM MyTable AS s + |JOIN LookupTableWithFilterableFields FOR SYSTEM_TIME AS OF s.proctime AS d + |ON s.a = d.id + |WHERE d.status = 'OK' + |UNION ALL + |SELECT s.a, s.b, s.proctime, d.status + |FROM MyTable AS s + |JOIN LookupTableWithFilterableFields FOR SYSTEM_TIME AS OF s.proctime AS d + |ON s.a = d.id + |WHERE d.status = 'OK' + """.stripMargin + + util.verifyExecPlan(sql) + } + @Test def testJoinTemporalTableWithFilterPushDown(): Unit = { val sql =