fresh-borzoni commented on code in PR #29653: URL: https://github.com/apache/flink/pull/29653#discussion_r4423500412
########## flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/UpsertKeyEquivalenceUtil.scala: ########## @@ -0,0 +1,381 @@ +/* + * 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.utils + +import org.apache.flink.table.planner.plan.metadata.{EquiJoinColumnPair, FlinkRelMetadataQuery} +import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalMultiJoin + +import org.apache.calcite.plan.hep.HepRelVertex +import org.apache.calcite.plan.volcano.RelSubset +import org.apache.calcite.rel.`type`.RelDataType +import org.apache.calcite.rel.RelNode +import org.apache.calcite.rel.core.{Calc, Exchange, Join, JoinRelType, Project} +import org.apache.calcite.rex.{RexInputRef, RexNode} +import org.apache.calcite.util.{ImmutableBitSet, Util} + +import java.util +import java.util.{List => JList, Map => JMap, Set => JSet} + +import scala.collection.JavaConverters._ + +/** + * Finds the upsert keys of a subtree that fall within a target column set, treating the columns of + * [[FlinkRelMetadataQuery#getEquiJoinColumnPairs]] as interchangeable where the substitution is + * null-safe (see `directionalSubstitutions`), also through chains of such pairs. The target is + * fixed, so this stays bounded -- it never materializes equivalent key variants the way eager key + * enrichment would. For a Calc that forwards deletes by key, [[upsertKeyColumns]] gives the columns + * such keys can use, as one column set, so it does not enumerate keys either. + */ +object UpsertKeyEquivalenceUtil { + + /** + * Upsert keys of `rel` (in its output columns) that are a subset of `target`, both those reported + * by [[FlinkRelMetadataQuery#getUpsertKeys]] and those recovered by null-safe equi-join + * substitution. Null if the upsert keys are undetermined. + */ + def upsertKeysWithin( + rel: RelNode, + target: ImmutableBitSet, + mq: FlinkRelMetadataQuery): JSet[ImmutableBitSet] = { + val upsertKeys = mq.getUpsertKeys(rel) + if (upsertKeys == null) { + return null + } + val result = new util.LinkedHashSet[ImmutableBitSet]() + upsertKeys.asScala.foreach(key => if (target.contains(key)) result.add(key)) + result.addAll(recoverWithin(rel, target, mq)) + result + } + + /** + * Every output column of `rel` that can be part of one of its upsert keys, reported or recovered + * by equi-join substitution, for any target: a superset of the columns of every key + * [[upsertKeysWithin]] can return. Recovered columns are pass-through join columns, which every + * row a join emits carries: a join accepts deletes by key only from an input whose unique key + * contains its join key, and a row an all-inner multi-join emits matched every join condition. + */ + def upsertKeyColumns(rel: RelNode, mq: FlinkRelMetadataQuery): ImmutableBitSet = { + val upsertKeys = mq.getUpsertKeys(rel) + if (upsertKeys == null) { + return ImmutableBitSet.of() + } + keyColumns(upsertKeys).union(recoverableColumns(rel, mq)) + } + + /** The columns [[recoverWithin]] can use, over all targets. */ + private def recoverableColumns(rel: RelNode, mq: FlinkRelMetadataQuery): ImmutableBitSet = Review Comment: They answer different questions: upsertKeysWithin finds keys that fit one target, while upsertKeyColumns needs every column any such key could use, for the Calc's delete-by-key. One walk would have to enumerate key variants. The part where they could diverge, which join inputs keep their keys, is already shared in keyPreservingInputs. -- 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]
