gustavodemorais commented on code in PR #29653:
URL: https://github.com/apache/flink/pull/29653#discussion_r4523969632
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/UpsertKeyUtil.java:
##########
@@ -63,4 +78,97 @@ public static Optional<int[]> smallestKey(@Nullable
Set<ImmutableBitSet> upsertK
upsertKeys.stream().map(ImmutableBitSet::toArray).collect(Collectors.toList());
return Optional.of(UpsertKeyUtils.smallestKey(asArrays));
}
+
+ /**
+ * Returns the upsert keys of the given node, plus each key whose columns
outside {@code
+ * columns} can all be replaced by columns inside that are equal to them
on every row, e.g.
+ * through an inner equi-join condition. For example, for {@code t1 JOIN
t2 ON a1 = a2} and
+ * columns {a1, b2}, the upsert key {a2, b2} adds {a1, b2}.
+ *
+ * @return the upsert keys and the aligned ones, or null if the upsert
keys are unknown
+ */
+ public static @Nullable Set<ImmutableBitSet> getUpsertKeysAlignedTo(
+ RelNode rel, ImmutableBitSet columns, RelMetadataQuery mq) {
+ final FlinkRelMetadataQuery fmq =
FlinkRelMetadataQuery.reuseOrCreate(mq);
+ final Set<ImmutableBitSet> upsertKeys = fmq.getUpsertKeys(rel);
+ if (upsertKeys == null ||
upsertKeys.stream().allMatch(columns::contains)) {
+ return upsertKeys;
+ }
+
+ final Map<Integer, Integer> parents = getEqualColumnParents(rel, fmq);
+ final Map<Integer, Integer> replacements = new HashMap<>();
+ for (int column : columns) {
+ replacements.putIfAbsent(root(parents, column), column);
+ }
+ final Set<ImmutableBitSet> alignedKeys = new HashSet<>(upsertKeys);
+ for (ImmutableBitSet key : upsertKeys) {
+ final ImmutableBitSet.Builder alignedKey =
ImmutableBitSet.builder();
+ for (int column : key) {
+ alignedKey.set(
+ columns.get(column)
+ ? column
+ : replacements.getOrDefault(root(parents,
column), column));
+ }
+ final ImmutableBitSet aligned = alignedKey.build();
+ if (columns.contains(aligned)) {
+ alignedKeys.add(aligned);
+ }
+ }
+ return alignedKeys;
+ }
+
+ /**
+ * Returns {@code columns} and every column of the given node that is
equal to one of them on
+ * every row.
+ */
+ public static ImmutableBitSet getEqualColumns(
+ RelNode rel, ImmutableBitSet columns, RelMetadataQuery mq) {
+ final Map<Integer, Integer> parents = getEqualColumnParents(rel, mq);
+ final Set<Integer> roots =
+ columns.asList().stream()
+ .map(column -> root(parents, column))
+ .collect(Collectors.toSet());
+ final ImmutableBitSet.Builder equalColumns = columns.rebuild();
+ for (int column : parents.keySet()) {
+ final int root = root(parents, column);
+ if (roots.contains(root)) {
+ equalColumns.set(column).set(root);
+ }
+ }
+ return equalColumns.build();
+ }
+
+ /** Groups the columns that are equal on every row, as a union-find forest
of parents. */
+ private static Map<Integer, Integer> getEqualColumnParents(RelNode rel,
RelMetadataQuery mq) {
+ final Map<Integer, Integer> parents = new HashMap<>();
+ for (RexNode predicate :
mq.getPulledUpPredicates(rel).pulledUpPredicates) {
+ if (!predicate.isA(EQUALITY_KINDS)) {
+ continue;
+ }
+ final RexNode left = ((RexCall) predicate).getOperands().get(0);
+ final RexNode right = ((RexCall) predicate).getOperands().get(1);
+ // comparing different types may implicitly cast one side, which
needn't be injective,
+ // and FLOAT/DOUBLE compare -0.0 = 0.0 although they are different
keys
+ if (left instanceof RexInputRef
+ && right instanceof RexInputRef
+ && SqlTypeUtil.equalSansNullability(left.getType(),
right.getType())
+ && !SqlTypeUtil.isApproximateNumeric(left.getType())) {
Review Comment:
`!isApproximateNumeric` only looks at the top-level type, but `=` on
MAP/MULTISET ignores entry order and ROW compares nested DOUBLE fields with
`==`, while upsert keys compare bytes. Example: `r` with `PRIMARY KEY (id, k)`,
sink `PRIMARY KEY (id, v)`:
```sql
INSERT INTO sink SELECT l.id, r.k, r.v, r.name FROM l JOIN r ON l.id = r.id
WHERE r.k = r.v
```
With `ROW<d DOUBLE>` columns and rows `+I(1, k1=ROW(0.0), v=ROW(-0.0), r1)`,
`+I(1, k2=ROW(-0.0), v, r2)`, `-D(1, k1, v, r1)`, both rows pass the filter and
share the sink key, so the delete wipes `r2` and the sink ends up empty (same
with MAP `{1:a, 2:b}` vs `{2:b, 1:a}`).
```suggestion
// comparing different types may implicitly cast one side, which
needn't be injective
if (left instanceof RexInputRef
&& right instanceof RexInputRef
&& SqlTypeUtil.equalSansNullability(left.getType(),
right.getType())
&& isEqualityByBytes(left.getType())) {
```
with
```java
// types whose "=" is byte equality, unlike FLOAT/DOUBLE (-0.0 = 0.0),
MAP, MULTISET or ROW
private static boolean isEqualityByBytes(RelDataType type) {
return SqlTypeUtil.isCharacter(type)
|| SqlTypeUtil.isBinary(type)
|| SqlTypeUtil.isExactNumeric(type)
|| SqlTypeUtil.isBoolean(type)
|| SqlTypeUtil.isDatetime(type)
|| SqlTypeUtil.isInterval(type);
}
```
--
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]
