rionmonster commented on code in PR #29583:
URL: https://github.com/apache/flink/pull/29583#discussion_r4515665726
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/MultiJoinUtil.java:
##########
@@ -44,17 +51,89 @@ public class MultiJoinUtil {
.map(input -> input.getRowType().getFieldCount())
.collect(Collectors.toList());
+ final List<Integer> inputOffsets = inputOffsets(inputFieldCounts);
+
+ for (final RexNode condition : joinConditions) {
+ extractEqualityConditions(condition, inputOffsets,
inputFieldCounts, joinAttributeMap);
+ }
+ return joinAttributeMap;
+ }
+
+ /**
+ * Returns whether an input is stored under its upsert key, because it may
update a record
+ * without UPDATE_BEFORE or delete it by key. The planner only allows this
if an upsert key
+ * contains the common join key and, for all inputs but the first, the
join key of the input.
+ *
+ * @param inputChangelogMode the changelog mode of the input
+ * @param upsertKeys the upsert keys of the input that contain its common
join key and join key
+ */
+ public static boolean isStoredByUpsertKey(
+ final ChangelogMode inputChangelogMode, final List<int[]>
upsertKeys) {
+ final boolean isUpsert =
+ inputChangelogMode.contains(RowKind.UPDATE_AFTER)
+ && !inputChangelogMode.contains(RowKind.UPDATE_BEFORE);
+ return (isUpsert || inputChangelogMode.keyOnlyDeletes()) &&
!upsertKeys.isEmpty();
+ }
+
+ /**
+ * Returns, for each input, the indices of the fields that the join
conditions read and that are
+ * not part of every key of the input. Two versions of a record have the
same key, so only these
+ * fields can make the join conditions match them differently.
+ *
+ * @param inputRowTypes the row types of all inputs
+ * @param joinConditions the join condition of each input, over the fields
of all inputs up to
+ * and including it
+ * @param inputKeys the keys of each input
+ */
+ public static int[][] createConditionFields(
Review Comment:
Done and renamed! 👍
--
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]
