MartijnVisser commented on code in PR #29200:
URL: https://github.com/apache/flink/pull/29200#discussion_r4024880682
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/common/CommonPhysicalLookupJoin.scala:
##########
@@ -187,14 +187,20 @@ abstract class CommonPhysicalLookupJoin(
case None =>
resultFieldNames.mkString(", ")
}
- val tableIdentifier: ObjectIdentifier = temporalTable match {
- case t: TableSourceTable => t.contextResolvedTable.getIdentifier
- case t: LegacyTableSourceTable[_] => t.tableIdentifier
+ // The abilities pushed into the temporal table, a filter above all, are
part of its identity:
+ // two lookup joins on the same table with different push-downs are
different operators. A
+ // TableSourceScan gets this for free because Calcite's TableScan digests
+ // RelOptTable#getQualifiedName, which TableSourceTable extends with its
spec digests.
Review Comment:
Cut to two lines.
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/LookupJoinTestPrograms.java:
##########
@@ -456,4 +453,97 @@ public class LookupJoinTestPrograms {
+ "JOIN customers_t FOR SYSTEM_TIME AS OF
O.proc_time AS C "
+ "ON O.customer_id = C.id")
.build();
+
+ /**
+ * {@link #CUSTOMERS} without the after-restore data, so that it is a
plain {@code
+ * SOURCE_WITH_DATA} step usable from a semantic test.
+ */
+ static final SourceTestStep CUSTOMERS_NO_RESTORE =
+ SourceTestStep.newBuilder("customers_t")
+ .addOption("disable-lookup", "false") // static/lookup
table
+ .addOption("filterable-fields", "age")
+ .addSchema(CUSTOMERS_SCHEMA)
+ .producedValues(CUSTOMERS_BEFORE_DATA)
+ .build();
+
+ /** {@link #ORDERS} without the after-restore data. */
+ static final SourceTestStep ORDERS_NO_RESTORE =
+ 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);
+ }
+
+ /**
+ * Only customers older than 30 match, so the second branch contributes
nothing. If the two
Review Comment:
Dropped it. The `TableTestProgram` description already says what the program
validates, so the javadoc was both duplicating that and restating the data.
##########
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])
Review Comment:
Done, empty spec digests are dropped now. This file is back to exactly what
it was before the PR.
--
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]
