gustavodemorais commented on code in PR #29583:
URL: https://github.com/apache/flink/pull/29583#discussion_r4483954147
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/state/MultiJoinStateViews.java:
##########
@@ -169,6 +187,25 @@ public Iterable<RowData> getRecords(RowData joinKey)
throws Exception {
}
return reusedList;
}
+
+ @Override
+ public @Nullable RowData getRecord(@Nullable RowData joinKey, RowData
record)
+ throws Exception {
+ // the join key contains the unique key, so a record under it has
the same unique key
+ final RowData storedRecord = recordState.get(joinKey);
+ if (storedRecord != null || !joinKeyMayChange) {
+ return storedRecord;
+ }
+ final BinaryRowData uniqueKey =
+
uniqueKeySerializer.toBinaryRow(uniqueKeySelector.getKey(record), true);
+ for (RowData candidate : recordState.values()) {
Review Comment:
This scan makes every new record of such an input read all records of its
common join key, e.g. with ON u.user_id = o.user_id AND u.region = o.region and
Orders keyed by (user_id, order_id), each new order scans all orders of the
user. Imagine if the user has 1k, 100k or 1 million orders. O(N) per record
isn't really a viable solution.
Alternative: could we use the same rule as the binary join instead: only let
an input drop UPDATE_BEFORE if its upsert key contains its whole join key, and
keep UPDATE_BEFORE otherwise (ChangelogNormalize for upsert sources)? Then the
old version is always under the same join key, the lookup stays a point read
and we can drop joinKeyMayChange and both scans. ChangelogNormalize is the
price to pay but it does one read and write for an update once in the pipeline
instead of a O(N) scan
--
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]
