MartijnVisser commented on code in PR #29200:
URL: https://github.com/apache/flink/pull/29200#discussion_r4024882069
##########
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]]>
<![CDATA[
Sink(table=[default_catalog.default_database.sink_without_pk], fields=[a, b,
c])
+- Calc(select=[a, b, c])
- +- 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]])
+ +- 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]])
Review Comment:
Same change, `metadata=[]` and `filter=[]` are gone here too. It took 14
lines out of the diff that were carrying no information at all.
##########
flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/LookupJoinTest.scala:
##########
@@ -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
Review Comment:
The same-filter one cannot be a semantic test. Reuse or not, the UNION ALL
still has two inputs, so the sink sees each row twice either way and the output
is identical. Only the plan shows whether the two joins collapsed. It guards
the other direction of this fix, a digest that is too unique and silently kills
legitimate subplan reuse.
`ScanReuseTest` has the same pair on the scan side,
`testProjectWithFilterPushDown` and `testProjectReuseWithFilterPushDown`, both
plan tests.
##########
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.LookupJoinTestPrograms;
+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 {
Review Comment:
`JoinSemanticTests` is for `StreamExecJoin` only. Its eight programs are all
plain stream-stream joins, there is no `SYSTEM_TIME` or `LOOKUP` in any of
them. Lateral snapshot join and multi join are also joins and both got their
own class rather than going in there, so I kept this one.
You did make me look at where the programs live though. They were in
`LookupJoinTestPrograms`, which is the restore-test file, and the two sources I
added only exist because a semantic test cannot take restore data. Moved them
to `LookupJoinSemanticTestPrograms`, same split FLINK-38720 did for joins.
--
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]
