You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sn...@apache.org on 2024/08/28 09:45:07 UTC
(flink) branch master updated (20e6d634663 -> 55dc9fa6922)
This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from 20e6d634663 [FLINK-35773][docs] Document s5cmd
new f01de3c4303 [FLINK-34896][table] Migrate CorrelateSortToRankRule to java
new 55dc9fa6922 [FLINK-35054][table] Migrate TemporalJoinRewriteWithUniqueKeyRule to java
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../rules/logical/CorrelateSortToRankRule.java | 256 +++++++++++++++++
.../TemporalJoinRewriteWithUniqueKeyRule.java | 304 +++++++++++++++++++++
.../rules/logical/CorrelateSortToRankRule.scala | 192 -------------
.../TemporalJoinRewriteWithUniqueKeyRule.scala | 201 --------------
4 files changed, 560 insertions(+), 393 deletions(-)
create mode 100644 flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/CorrelateSortToRankRule.java
create mode 100644 flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/TemporalJoinRewriteWithUniqueKeyRule.java
delete mode 100644 flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/CorrelateSortToRankRule.scala
delete mode 100644 flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/TemporalJoinRewriteWithUniqueKeyRule.scala
(flink) 01/02: [FLINK-34896][table] Migrate CorrelateSortToRankRule to java
Posted by sn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit f01de3c4303c4f9ec2b5781af41347c7ecb1630b
Author: Sergey Nuyanzin <sn...@gmail.com>
AuthorDate: Fri Mar 15 09:33:14 2024 +0100
[FLINK-34896][table] Migrate CorrelateSortToRankRule to java
This closes #24545
---
.../rules/logical/CorrelateSortToRankRule.java | 256 +++++++++++++++++++++
.../rules/logical/CorrelateSortToRankRule.scala | 192 ----------------
2 files changed, 256 insertions(+), 192 deletions(-)
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/CorrelateSortToRankRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/CorrelateSortToRankRule.java
new file mode 100644
index 00000000000..82c38712a39
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/CorrelateSortToRankRule.java
@@ -0,0 +1,256 @@
+/*
+ * 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.rules.logical;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.table.planner.calcite.FlinkRelBuilder;
+import org.apache.flink.table.planner.calcite.FlinkRelFactories;
+import org.apache.flink.table.runtime.operators.rank.ConstantRankRange;
+import org.apache.flink.table.runtime.operators.rank.RankType;
+
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.plan.hep.HepPlanner;
+import org.apache.calcite.rel.RelCollation;
+import org.apache.calcite.rel.RelCollations;
+import org.apache.calcite.rel.RelFieldCollation;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Aggregate;
+import org.apache.calcite.rel.core.Correlate;
+import org.apache.calcite.rel.core.Filter;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.core.Project;
+import org.apache.calcite.rel.core.Sort;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexCorrelVariable;
+import org.apache.calcite.rex.RexFieldAccess;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexLiteral;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.tools.RelBuilder;
+import org.apache.calcite.util.ImmutableBitSet;
+import org.immutables.value.Value;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * Planner rule that rewrites sort correlation to a Rank. Typically, the following plan
+ *
+ * <pre>{@code
+ * LogicalProject(state=[$0], name=[$1])
+ * +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{0}])
+ * :- LogicalAggregate(group=[{0}])
+ * : +- LogicalProject(state=[$1])
+ * : +- LogicalTableScan(table=[[default_catalog, default_database, cities]])
+ * +- LogicalSort(sort0=[$1], dir0=[DESC-nulls-last], fetch=[3])
+ * +- LogicalProject(name=[$0], pop=[$2])
+ * +- LogicalFilter(condition=[=($1, $cor0.state)])
+ * +- LogicalTableScan(table=[[default_catalog, default_database, cities]])
+ * }</pre>
+ *
+ * <p>would be transformed to
+ *
+ * <pre>{@code
+ * LogicalProject(state=[$0], name=[$1])
+ * +- LogicalProject(state=[$1], name=[$0], pop=[$2])
+ * +- LogicalRank(rankType=[ROW_NUMBER], rankRange=[rankStart=1, rankEnd=3],
+ * partitionBy=[$1], orderBy=[$2 DESC], select=[name=$0, state=$1, pop=$2])
+ * +- LogicalTableScan(table=[[default_catalog, default_database, cities]])
+ * }</pre>
+ *
+ * <p>To match the Correlate, the LHS needs to be a global Aggregate on a scan, the RHS should be a
+ * Sort with an equal Filter predicate whose keys are same with the LHS grouping keys.
+ *
+ * <p>This rule can only be used in {@link HepPlanner}.
+ */
+@Value.Enclosing
+public class CorrelateSortToRankRule
+ extends RelRule<CorrelateSortToRankRule.CorrelateSortToRankRuleConfig> {
+
+ public static final CorrelateSortToRankRule INSTANCE =
+ CorrelateSortToRankRule.CorrelateSortToRankRuleConfig.DEFAULT.toRule();
+
+ protected CorrelateSortToRankRule(CorrelateSortToRankRuleConfig config) {
+ super(config);
+ }
+
+ @Override
+ public boolean matches(RelOptRuleCall call) {
+ Correlate correlate = call.rel(0);
+ if (correlate.getJoinType() != JoinRelType.INNER) {
+ return false;
+ }
+ Aggregate agg = call.rel(1);
+ if (!agg.getAggCallList().isEmpty() || agg.getGroupSets().size() > 1) {
+ return false;
+ }
+ Project aggInput = call.rel(2);
+ if (!aggInput.isMapping()) {
+ return false;
+ }
+ Sort sort = call.rel(3);
+ if (sort.offset != null || sort.fetch == null) {
+ // 1. we can not describe the offset using rank
+ // 2. there is no need to transform to rank if no fetch limit
+ return false;
+ }
+ Project sortInput = call.rel(4);
+ if (!sortInput.isMapping()) {
+ return false;
+ }
+ Filter filter = call.rel(5);
+
+ List<RexNode> cnfCond = RelOptUtil.conjunctions(filter.getCondition());
+ if (cnfCond.stream().anyMatch(c -> !isValidCondition(c, correlate))) {
+ return false;
+ }
+
+ return aggInput.getInput().getDigest().equals(filter.getInput().getDigest());
+ }
+
+ private boolean isValidCondition(RexNode condition, Correlate correlate) {
+ // must be equiv condition
+ if (condition.getKind() != SqlKind.EQUALS) {
+ return false;
+ }
+ Tuple2<RexInputRef, RexFieldAccess> tuple = resolveFilterCondition(condition);
+ if (tuple.f0 == null) {
+ return false;
+ }
+ RexCorrelVariable variable = (RexCorrelVariable) tuple.f1.getReferenceExpr();
+ return variable.id.equals(correlate.getCorrelationId());
+ }
+
+ /**
+ * Resolves the filter condition with specific pattern: input ref and field access.
+ *
+ * @param condition The join condition
+ * @return tuple of operands (RexInputRef, RexFieldAccess), or null if the pattern does not
+ * match
+ */
+ private Tuple2<RexInputRef, RexFieldAccess> resolveFilterCondition(RexNode condition) {
+ RexCall condCall = (RexCall) condition;
+ RexNode operand0 = condCall.getOperands().get(0);
+ RexNode operand1 = condCall.getOperands().get(1);
+ if (operand0.isA(SqlKind.INPUT_REF) && operand1.isA(SqlKind.FIELD_ACCESS)) {
+ return Tuple2.of((RexInputRef) operand0, (RexFieldAccess) operand1);
+ } else if (operand0.isA(SqlKind.FIELD_ACCESS) && operand1.isA(SqlKind.INPUT_REF)) {
+ return Tuple2.of((RexInputRef) operand1, (RexFieldAccess) operand0);
+ } else {
+ return Tuple2.of(null, null);
+ }
+ }
+
+ @Override
+ public void onMatch(RelOptRuleCall call) {
+ RelBuilder builder = call.builder();
+
+ Sort sort = call.rel(3);
+ Project sortInput = call.rel(4);
+ Filter filter = call.rel(5);
+
+ List<RexNode> cnfCond = RelOptUtil.conjunctions(filter.getCondition());
+ ImmutableBitSet partitionKey =
+ ImmutableBitSet.of(
+ cnfCond.stream()
+ .map(c -> resolveFilterCondition(c).f0.getIndex())
+ .collect(Collectors.toList()));
+
+ RelDataType baseType = sortInput.getInput().getRowType();
+ List<RexNode> projects = new ArrayList<>();
+ partitionKey.asList().forEach(k -> projects.add(RexInputRef.of(k, baseType)));
+ projects.addAll(sortInput.getProjects());
+
+ RelCollation oriCollation = sort.getCollation();
+ List<RelFieldCollation> newFieldCollations =
+ oriCollation.getFieldCollations().stream()
+ .map(
+ fc -> {
+ int newFieldIdx =
+ ((RexInputRef)
+ sortInput
+ .getProjects()
+ .get(fc.getFieldIndex()))
+ .getIndex();
+ return fc.withFieldIndex(newFieldIdx);
+ })
+ .collect(Collectors.toList());
+ RelCollation newCollation = RelCollations.of(newFieldCollations);
+
+ RelNode newRel =
+ ((FlinkRelBuilder) (builder.push(filter.getInput())))
+ .rank(
+ partitionKey,
+ newCollation,
+ RankType.ROW_NUMBER,
+ new ConstantRankRange(
+ 1, ((RexLiteral) sort.fetch).getValueAs(Long.class)),
+ null,
+ false)
+ .project(projects)
+ .build();
+
+ call.transformTo(newRel);
+ }
+
+ /** Rule configuration. */
+ @Value.Immutable(singleton = false)
+ public interface CorrelateSortToRankRuleConfig extends RelRule.Config {
+ CorrelateSortToRankRule.CorrelateSortToRankRuleConfig DEFAULT =
+ ImmutableCorrelateSortToRankRule.CorrelateSortToRankRuleConfig.builder()
+ .operandSupplier(
+ b0 ->
+ b0.operand(Correlate.class)
+ .inputs(
+ b1 ->
+ b1.operand(Aggregate.class)
+ .oneInput(
+ b2 ->
+ b2.operand(
+ Project
+ .class)
+ .anyInputs()),
+ b2 ->
+ b2.operand(Sort.class)
+ .inputs(
+ b3 ->
+ b3.operand(
+ Project
+ .class)
+ .inputs(
+ b4 ->
+ b4.operand(
+ Filter
+ .class)
+ .anyInputs()))))
+ .relBuilderFactory(FlinkRelFactories.FLINK_REL_BUILDER())
+ .description("CorrelateSortToRankRule")
+ .build();
+
+ @Override
+ default CorrelateSortToRankRule toRule() {
+ return new CorrelateSortToRankRule(this);
+ }
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/CorrelateSortToRankRule.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/CorrelateSortToRankRule.scala
deleted file mode 100644
index 32efd087bbe..00000000000
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/CorrelateSortToRankRule.scala
+++ /dev/null
@@ -1,192 +0,0 @@
-/*
- * 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.rules.logical
-
-import org.apache.flink.table.planner.calcite.{FlinkRelBuilder, FlinkRelFactories}
-import org.apache.flink.table.runtime.operators.rank.{ConstantRankRange, RankType}
-
-import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall, RelOptUtil}
-import org.apache.calcite.plan.RelOptRule.{any, operand}
-import org.apache.calcite.rel.`type`.RelDataType
-import org.apache.calcite.rel.RelCollations
-import org.apache.calcite.rel.core.{Aggregate, Correlate, Filter, JoinRelType, Project, Sort}
-import org.apache.calcite.rex.{RexCall, RexCorrelVariable, RexFieldAccess, RexInputRef, RexLiteral, RexNode}
-import org.apache.calcite.sql.SqlKind
-import org.apache.calcite.util.ImmutableBitSet
-
-import java.util
-
-import scala.collection.JavaConversions._
-
-/**
- * Planner rule that rewrites sort correlation to a Rank. Typically, the following plan
- *
- * {{{
- * LogicalProject(state=[$0], name=[$1])
- * +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{0}])
- * :- LogicalAggregate(group=[{0}])
- * : +- LogicalProject(state=[$1])
- * : +- LogicalTableScan(table=[[default_catalog, default_database, cities]])
- * +- LogicalSort(sort0=[$1], dir0=[DESC-nulls-last], fetch=[3])
- * +- LogicalProject(name=[$0], pop=[$2])
- * +- LogicalFilter(condition=[=($1, $cor0.state)])
- * +- LogicalTableScan(table=[[default_catalog, default_database, cities]])
- * }}}
- *
- * <p>would be transformed to
- *
- * {{{
- * LogicalProject(state=[$0], name=[$1])
- * +- LogicalProject(state=[$1], name=[$0], pop=[$2])
- * +- LogicalRank(rankType=[ROW_NUMBER], rankRange=[rankStart=1, rankEnd=3],
- * partitionBy=[$1], orderBy=[$2 DESC], select=[name=$0, state=$1, pop=$2])
- * +- LogicalTableScan(table=[[default_catalog, default_database, cities]])
- * }}}
- *
- * <p>To match the Correlate, the LHS needs to be a global Aggregate on a scan, the RHS should be a
- * Sort with an equal Filter predicate whose keys are same with the LHS grouping keys.
- *
- * <p>This rule can only be used in HepPlanner.
- */
-class CorrelateSortToRankRule
- extends RelOptRule(
- operand(
- classOf[Correlate],
- operand(classOf[Aggregate], operand(classOf[Project], any())),
- operand(classOf[Sort], operand(classOf[Project], operand(classOf[Filter], any())))
- ),
- FlinkRelFactories.FLINK_REL_BUILDER,
- "CorrelateSortToRankRule") {
-
- override def matches(call: RelOptRuleCall): Boolean = {
- val correlate: Correlate = call.rel(0)
- if (correlate.getJoinType != JoinRelType.INNER) {
- return false
- }
- val agg: Aggregate = call.rel(1)
- if (agg.getAggCallList.size() > 0 || agg.getGroupSets.size() > 1) {
- return false
- }
- val aggInput: Project = call.rel(2)
- if (!aggInput.isMapping) {
- return false
- }
- val sort: Sort = call.rel(3)
- if (sort.offset != null || sort.fetch == null) {
- // 1. we can not describe the offset using rank
- // 2. there is no need to transform to rank if no fetch limit
- return false
- }
- val sortInput: Project = call.rel(4)
- if (!sortInput.isMapping) {
- return false
- }
- val filter: Filter = call.rel(5)
-
- val cnfCond = RelOptUtil.conjunctions(filter.getCondition)
- if (cnfCond.exists(c => !isValidCondition(c, correlate))) {
- return false
- }
-
- aggInput.getInput.getDigest.equals(filter.getInput.getDigest)
- }
-
- private def isValidCondition(condition: RexNode, correlate: Correlate): Boolean = {
- // must be equiv condition
- if (condition.getKind != SqlKind.EQUALS) {
- return false
- }
- val (inputRef, fieldAccess) = resolveFilterCondition(condition)
- if (inputRef == null) {
- return false
- }
- val variable = fieldAccess.getReferenceExpr.asInstanceOf[RexCorrelVariable]
- variable.id.equals(correlate.getCorrelationId)
- }
-
- /**
- * Resolves the filter condition with specific pattern: input ref and field access.
- *
- * @param condition
- * The join condition
- * @return
- * tuple of operands (RexInputRef, RexFieldAccess), or null if the pattern does not match
- */
- private def resolveFilterCondition(condition: RexNode): (RexInputRef, RexFieldAccess) = {
- val condCall = condition.asInstanceOf[RexCall]
- val operand0 = condCall.getOperands.get(0)
- val operand1 = condCall.getOperands.get(1)
- if (operand0.isA(SqlKind.INPUT_REF) && operand1.isA(SqlKind.FIELD_ACCESS)) {
- (operand0.asInstanceOf[RexInputRef], operand1.asInstanceOf[RexFieldAccess])
- } else if (operand0.isA(SqlKind.FIELD_ACCESS) && operand1.isA(SqlKind.INPUT_REF)) {
- (operand1.asInstanceOf[RexInputRef], operand0.asInstanceOf[RexFieldAccess])
- } else {
- (null, null)
- }
- }
-
- override def onMatch(call: RelOptRuleCall): Unit = {
- val builder = call.builder()
-
- val sort: Sort = call.rel(3)
- val sortInput: Project = call.rel(4)
- val filter: Filter = call.rel(5)
-
- val cnfCond = RelOptUtil.conjunctions(filter.getCondition)
- val partitionKey: ImmutableBitSet =
- ImmutableBitSet.of(cnfCond.map(c => resolveFilterCondition(c)._1.getIndex): _*)
-
- val baseType: RelDataType = sortInput.getInput().getRowType
- val projects = new util.ArrayList[RexNode]()
- partitionKey.asList().foreach(k => projects.add(RexInputRef.of(k, baseType)))
- projects.addAll(sortInput.getProjects)
-
- val oriCollation = sort.getCollation
- val newFieldCollations = oriCollation.getFieldCollations.map {
- fc =>
- val newFieldIdx = sortInput.getProjects
- .get(fc.getFieldIndex)
- .asInstanceOf[RexInputRef]
- .getIndex
- fc.withFieldIndex(newFieldIdx)
- }
- val newCollation = RelCollations.of(newFieldCollations)
-
- val newRel = builder
- .push(filter.getInput())
- .asInstanceOf[FlinkRelBuilder]
- .rank(
- partitionKey,
- newCollation,
- RankType.ROW_NUMBER,
- new ConstantRankRange(
- 1,
- sort.fetch.asInstanceOf[RexLiteral].getValueAs(classOf[java.lang.Long])),
- null,
- false
- )
- .project(projects)
- .build()
-
- call.transformTo(newRel)
- }
-}
-
-object CorrelateSortToRankRule {
- val INSTANCE = new CorrelateSortToRankRule
-}
(flink) 02/02: [FLINK-35054][table] Migrate TemporalJoinRewriteWithUniqueKeyRule to java
Posted by sn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 55dc9fa6922aacab3258159affe5867934771c6e
Author: liuyongvs <li...@gmail.com>
AuthorDate: Tue Apr 9 15:23:22 2024 +0800
[FLINK-35054][table] Migrate TemporalJoinRewriteWithUniqueKeyRule to java
This closes #24635
---
.../TemporalJoinRewriteWithUniqueKeyRule.java | 304 +++++++++++++++++++++
.../TemporalJoinRewriteWithUniqueKeyRule.scala | 201 --------------
2 files changed, 304 insertions(+), 201 deletions(-)
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/TemporalJoinRewriteWithUniqueKeyRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/TemporalJoinRewriteWithUniqueKeyRule.java
new file mode 100644
index 00000000000..342cafd2526
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/TemporalJoinRewriteWithUniqueKeyRule.java
@@ -0,0 +1,304 @@
+/*
+ * 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.rules.logical;
+
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.planner.plan.metadata.FlinkRelMetadataQuery;
+import org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalJoin;
+import org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalRel;
+import org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalSnapshot;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalDeduplicate;
+import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalRank;
+import org.apache.flink.table.planner.plan.rules.common.CommonTemporalTableJoinRule;
+import org.apache.flink.table.planner.plan.utils.TemporalJoinUtil;
+
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexShuttle;
+import org.apache.calcite.util.ImmutableBitSet;
+import org.immutables.value.Value;
+
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import scala.collection.JavaConverters;
+
+/**
+ * Planner rule that rewrites temporal join with extracted primary key, Event-time temporal table
+ * join requires primary key and row time attribute of versioned table. The versioned table could be
+ * a table source or a view only if it contains the unique key and time attribute.
+ *
+ * <p>Flink supports extract the primary key and row time attribute from the view if the view comes
+ * from {@link StreamPhysicalRank} node which can convert to a {@link StreamPhysicalDeduplicate}
+ * node.
+ */
+@Value.Enclosing
+public class TemporalJoinRewriteWithUniqueKeyRule
+ extends RelRule<
+ TemporalJoinRewriteWithUniqueKeyRule.TemporalJoinRewriteWithUniqueKeyRuleConfig>
+ implements CommonTemporalTableJoinRule {
+
+ public static final TemporalJoinRewriteWithUniqueKeyRule INSTANCE =
+ TemporalJoinRewriteWithUniqueKeyRule.TemporalJoinRewriteWithUniqueKeyRuleConfig.DEFAULT
+ .toRule();
+
+ private TemporalJoinRewriteWithUniqueKeyRule(
+ TemporalJoinRewriteWithUniqueKeyRuleConfig config) {
+ super(config);
+ }
+
+ @Override
+ public boolean matches(RelOptRuleCall call) {
+ FlinkLogicalJoin join = call.rel(0);
+ FlinkLogicalSnapshot snapshot = call.rel(2);
+ FlinkLogicalRel snapshotInput = call.rel(3);
+
+ boolean isTemporalJoin = matches(snapshot);
+ boolean canConvertToLookup = canConvertToLookupJoin(snapshot, snapshotInput);
+ List<JoinRelType> supportedJoinTypes = Arrays.asList(JoinRelType.INNER, JoinRelType.LEFT);
+
+ return isTemporalJoin
+ && !canConvertToLookup
+ && supportedJoinTypes.contains(join.getJoinType());
+ }
+
+ @Override
+ public void onMatch(RelOptRuleCall call) {
+ FlinkLogicalJoin join = call.rel(0);
+ FlinkLogicalRel leftInput = call.rel(1);
+ FlinkLogicalSnapshot snapshot = call.rel(2);
+ FlinkLogicalRel snapshotInput = call.rel(3);
+
+ RexNode joinCondition = join.getCondition();
+ RexNode newJoinCondition =
+ joinCondition.accept(
+ new RexShuttle() {
+ @Override
+ public RexNode visitCall(RexCall call) {
+ if (call.getOperator()
+ .equals(
+ TemporalJoinUtil
+ .INITIAL_TEMPORAL_JOIN_CONDITION())) {
+ RexNode snapshotTimeInputRef;
+ List<RexNode> leftJoinKey;
+ List<RexNode> rightJoinKey;
+
+ if (TemporalJoinUtil.isInitialRowTimeTemporalTableJoin(call)) {
+ snapshotTimeInputRef = call.operands.get(0);
+ leftJoinKey =
+ ((RexCall) call.operands.get(2)).getOperands();
+ rightJoinKey =
+ ((RexCall) call.operands.get(3)).getOperands();
+ } else {
+ snapshotTimeInputRef = call.operands.get(0);
+ leftJoinKey =
+ ((RexCall) call.operands.get(1)).getOperands();
+ rightJoinKey =
+ ((RexCall) call.operands.get(2)).getOperands();
+ }
+
+ RexBuilder rexBuilder = join.getCluster().getRexBuilder();
+ Optional<List<RexNode>> primaryKeyInputRefs =
+ extractPrimaryKeyInputRefs(
+ leftInput, snapshot, snapshotInput, rexBuilder);
+ validateRightPrimaryKey(
+ join, rightJoinKey, primaryKeyInputRefs);
+
+ if (TemporalJoinUtil.isInitialRowTimeTemporalTableJoin(call)) {
+ RexNode rightTimeInputRef = call.operands.get(1);
+ return TemporalJoinUtil.makeRowTimeTemporalTableJoinConCall(
+ rexBuilder,
+ snapshotTimeInputRef,
+ rightTimeInputRef,
+ JavaConverters.asScalaBufferConverter(
+ primaryKeyInputRefs.get())
+ .asScala(),
+ JavaConverters.asScalaBufferConverter(leftJoinKey)
+ .asScala(),
+ JavaConverters.asScalaBufferConverter(rightJoinKey)
+ .asScala());
+ } else {
+ return TemporalJoinUtil
+ .makeProcTimeTemporalTableJoinConCall(
+ rexBuilder,
+ snapshotTimeInputRef,
+ JavaConverters.asScalaBufferConverter(
+ primaryKeyInputRefs.get())
+ .asScala(),
+ JavaConverters.asScalaBufferConverter(
+ leftJoinKey)
+ .asScala(),
+ JavaConverters.asScalaBufferConverter(
+ rightJoinKey)
+ .asScala());
+ }
+ } else {
+ return super.visitCall(call);
+ }
+ }
+ });
+ RelNode rewriteJoin =
+ FlinkLogicalJoin.create(
+ leftInput, snapshot, newJoinCondition, join.getHints(), join.getJoinType());
+ call.transformTo(rewriteJoin);
+ }
+
+ private void validateRightPrimaryKey(
+ FlinkLogicalJoin join,
+ List<RexNode> rightJoinKeyExpressions,
+ Optional<List<RexNode>> rightPrimaryKeyInputRefs) {
+
+ if (!rightPrimaryKeyInputRefs.isPresent()) {
+ throw new ValidationException(
+ "Temporal Table Join requires primary key in versioned table, "
+ + "but no primary key can be found. "
+ + "The physical plan is:\n"
+ + RelOptUtil.toString(join)
+ + "\n");
+ }
+
+ List<Integer> rightJoinKeyRefIndices =
+ rightJoinKeyExpressions.stream()
+ .map(rex -> ((RexInputRef) rex).getIndex())
+ .collect(Collectors.toList());
+
+ List<Integer> rightPrimaryKeyRefIndices =
+ rightPrimaryKeyInputRefs.get().stream()
+ .map(rex -> ((RexInputRef) rex).getIndex())
+ .collect(Collectors.toList());
+
+ boolean primaryKeyContainedInJoinKey =
+ rightPrimaryKeyRefIndices.stream()
+ .allMatch(pk -> rightJoinKeyRefIndices.contains(pk));
+
+ if (!primaryKeyContainedInJoinKey) {
+ List<String> joinFieldNames = join.getRowType().getFieldNames();
+ List<String> joinLeftFieldNames = join.getLeft().getRowType().getFieldNames();
+ List<String> joinRightFieldNames = join.getRight().getRowType().getFieldNames();
+
+ String primaryKeyNames =
+ rightPrimaryKeyRefIndices.stream()
+ .map(i -> joinFieldNames.get(i))
+ .collect(Collectors.joining(","));
+
+ String joinEquiInfo =
+ join.analyzeCondition().pairs().stream()
+ .map(
+ pair ->
+ joinLeftFieldNames.get(pair.source)
+ + "="
+ + joinRightFieldNames.get(pair.target))
+ .collect(Collectors.joining(","));
+
+ throw new ValidationException(
+ "Temporal table's primary key ["
+ + primaryKeyNames
+ + "] must be included in the "
+ + "equivalence condition of temporal join, but current temporal join condition is ["
+ + joinEquiInfo
+ + "].");
+ }
+ }
+
+ private Optional<List<RexNode>> extractPrimaryKeyInputRefs(
+ RelNode leftInput,
+ FlinkLogicalSnapshot snapshot,
+ FlinkLogicalRel snapshotInput,
+ RexBuilder rexBuilder) {
+ List<RelDataTypeField> rightFields = snapshot.getRowType().getFieldList();
+ FlinkRelMetadataQuery fmq =
+ FlinkRelMetadataQuery.reuseOrCreate(snapshot.getCluster().getMetadataQuery());
+
+ Set<ImmutableBitSet> upsertKeySet = fmq.getUpsertKeys(snapshotInput);
+ List<RelDataTypeField> fields = snapshot.getRowType().getFieldList();
+
+ if (upsertKeySet != null && !upsertKeySet.isEmpty()) {
+ int leftFieldCnt = leftInput.getRowType().getFieldCount();
+ List<List<RexNode>> upsertKeySetInputRefs =
+ upsertKeySet.stream()
+ .filter(bitSet -> !bitSet.isEmpty())
+ .map(
+ bitSet ->
+ Arrays.stream(bitSet.toArray())
+ .mapToObj(index -> fields.get(index))
+ // build InputRef of upsert key in snapshot
+ .map(
+ f ->
+ (RexNode)
+ rexBuilder.makeInputRef(
+ f.getType(),
+ leftFieldCnt
+ + rightFields
+ .indexOf(
+ f)))
+ .collect(Collectors.toList()))
+ .collect(Collectors.toList());
+ // select shortest upsert key as primary key
+ return upsertKeySetInputRefs.stream()
+ .sorted(Comparator.comparingInt(List::size))
+ .findFirst();
+ } else {
+ return Optional.empty();
+ }
+ }
+
+ /** Rule configuration. */
+ @Value.Immutable(singleton = false)
+ public interface TemporalJoinRewriteWithUniqueKeyRuleConfig extends RelRule.Config {
+ TemporalJoinRewriteWithUniqueKeyRule.TemporalJoinRewriteWithUniqueKeyRuleConfig DEFAULT =
+ ImmutableTemporalJoinRewriteWithUniqueKeyRule
+ .TemporalJoinRewriteWithUniqueKeyRuleConfig.builder()
+ .build()
+ .withOperandSupplier(
+ b0 ->
+ b0.operand(FlinkLogicalJoin.class)
+ .inputs(
+ b1 ->
+ b1.operand(FlinkLogicalRel.class)
+ .anyInputs(),
+ b2 ->
+ b2.operand(
+ FlinkLogicalSnapshot
+ .class)
+ .oneInput(
+ b3 ->
+ b3.operand(
+ FlinkLogicalRel
+ .class)
+ .anyInputs())))
+ .withDescription("TemporalJoinRewriteWithUniqueKeyRule");
+
+ @Override
+ default TemporalJoinRewriteWithUniqueKeyRule toRule() {
+ return new TemporalJoinRewriteWithUniqueKeyRule(this);
+ }
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/TemporalJoinRewriteWithUniqueKeyRule.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/TemporalJoinRewriteWithUniqueKeyRule.scala
deleted file mode 100644
index 74a17e5f55e..00000000000
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/logical/TemporalJoinRewriteWithUniqueKeyRule.scala
+++ /dev/null
@@ -1,201 +0,0 @@
-/*
- * 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.rules.logical
-
-import org.apache.flink.table.api.ValidationException
-import org.apache.flink.table.planner.plan.metadata.FlinkRelMetadataQuery
-import org.apache.flink.table.planner.plan.nodes.logical.{FlinkLogicalJoin, FlinkLogicalRel, FlinkLogicalSnapshot}
-import org.apache.flink.table.planner.plan.rules.common.CommonTemporalTableJoinRule
-import org.apache.flink.table.planner.plan.utils.TemporalJoinUtil
-
-import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall, RelOptUtil}
-import org.apache.calcite.plan.RelOptRule.{any, operand}
-import org.apache.calcite.rel.RelNode
-import org.apache.calcite.rel.core.JoinRelType
-import org.apache.calcite.rex._
-
-import scala.collection.JavaConversions._
-
-/**
- * Planner rule that rewrites temporal join with extracted primary key, Event-time temporal table
- * join requires primary key and row time attribute of versioned table. The versioned table could be
- * a table source or a view only if it contains the unique key and time attribute.
- *
- * <p> Flink supports extract the primary key and row time attribute from the view if the view comes
- * from [[LogicalRank]] node which can convert to a [[Deduplicate]] node.
- */
-class TemporalJoinRewriteWithUniqueKeyRule
- extends RelOptRule(
- operand(
- classOf[FlinkLogicalJoin],
- operand(classOf[FlinkLogicalRel], any()),
- operand(classOf[FlinkLogicalSnapshot], operand(classOf[FlinkLogicalRel], any()))),
- "TemporalJoinRewriteWithUniqueKeyRule")
- with CommonTemporalTableJoinRule {
-
- override def matches(call: RelOptRuleCall): Boolean = {
- val join = call.rel[FlinkLogicalJoin](0)
- val snapshot = call.rel[FlinkLogicalSnapshot](2)
- val snapshotInput = call.rel[FlinkLogicalRel](3)
-
- val isTemporalJoin = matches(snapshot)
- val canConvertToLookup = canConvertToLookupJoin(snapshot, snapshotInput)
- val supportedJoinTypes = Seq(JoinRelType.INNER, JoinRelType.LEFT)
-
- isTemporalJoin && !canConvertToLookup && supportedJoinTypes.contains(join.getJoinType)
- }
-
- override def onMatch(call: RelOptRuleCall): Unit = {
- val join = call.rel[FlinkLogicalJoin](0)
- val leftInput = call.rel[FlinkLogicalRel](1)
- val snapshot = call.rel[FlinkLogicalSnapshot](2)
- val snapshotInput = call.rel[FlinkLogicalRel](3)
-
- val joinCondition = join.getCondition
-
- val newJoinCondition = joinCondition.accept(new RexShuttle {
- override def visitCall(call: RexCall): RexNode = {
- if (call.getOperator == TemporalJoinUtil.INITIAL_TEMPORAL_JOIN_CONDITION) {
- val (snapshotTimeInputRef, leftJoinKey, rightJoinKey) =
- if (TemporalJoinUtil.isInitialRowTimeTemporalTableJoin(call)) {
- val snapshotTimeInputRef = call.operands(0)
- val leftJoinKey = call.operands(2).asInstanceOf[RexCall].operands
- val rightJoinKey = call.operands(3).asInstanceOf[RexCall].operands
- (snapshotTimeInputRef, leftJoinKey, rightJoinKey)
- } else {
- val snapshotTimeInputRef = call.operands(0)
- val leftJoinKey = call.operands(1).asInstanceOf[RexCall].operands
- val rightJoinKey = call.operands(2).asInstanceOf[RexCall].operands
- (snapshotTimeInputRef, leftJoinKey, rightJoinKey)
- }
-
- val rexBuilder = join.getCluster.getRexBuilder
- val primaryKeyInputRefs =
- extractPrimaryKeyInputRefs(leftInput, snapshot, snapshotInput, rexBuilder)
- validateRightPrimaryKey(join, rightJoinKey, primaryKeyInputRefs)
-
- if (TemporalJoinUtil.isInitialRowTimeTemporalTableJoin(call)) {
- val rightTimeInputRef = call.operands(1)
- TemporalJoinUtil.makeRowTimeTemporalTableJoinConCall(
- rexBuilder,
- snapshotTimeInputRef,
- rightTimeInputRef,
- primaryKeyInputRefs.get,
- leftJoinKey,
- rightJoinKey)
- } else {
- TemporalJoinUtil.makeProcTimeTemporalTableJoinConCall(
- rexBuilder,
- snapshotTimeInputRef,
- primaryKeyInputRefs.get,
- leftJoinKey,
- rightJoinKey)
- }
- } else {
- super.visitCall(call)
- }
- }
- })
- val rewriteJoin =
- FlinkLogicalJoin.create(
- leftInput,
- snapshot,
- newJoinCondition,
- join.getHints,
- join.getJoinType)
- call.transformTo(rewriteJoin)
- }
-
- private def validateRightPrimaryKey(
- join: FlinkLogicalJoin,
- rightJoinKeyExpression: Seq[RexNode],
- rightPrimaryKeyInputRefs: Option[Seq[RexNode]]): Unit = {
-
- if (rightPrimaryKeyInputRefs.isEmpty) {
- throw new ValidationException(
- "Temporal Table Join requires primary key in versioned table, " +
- s"but no primary key can be found. " +
- s"The physical plan is:\n${RelOptUtil.toString(join)}\n")
- }
-
- val rightJoinKeyRefIndices = rightJoinKeyExpression
- .map(rex => rex.asInstanceOf[RexInputRef].getIndex)
- .toArray
-
- val rightPrimaryKeyRefIndices = rightPrimaryKeyInputRefs.get
- .map(rex => rex.asInstanceOf[RexInputRef].getIndex)
- .toArray
-
- val primaryKeyContainedInJoinKey = rightPrimaryKeyRefIndices
- .forall(pk => rightJoinKeyRefIndices.contains(pk))
-
- if (!primaryKeyContainedInJoinKey) {
- val joinFieldNames = join.getRowType.getFieldNames
- val joinLeftFieldNames = join.getLeft.getRowType.getFieldNames
- val joinRightFieldNamess = join.getRight.getRowType.getFieldNames
- val primaryKeyNames = rightPrimaryKeyRefIndices
- .map(i => joinFieldNames.get(i))
- .toList
- .mkString(",")
- val joinEquiInfo = join.analyzeCondition
- .pairs()
- .map {
- pair => joinLeftFieldNames.get(pair.source) + "=" + joinRightFieldNamess.get(pair.target)
- }
- .toList
- .mkString(",")
- throw new ValidationException(
- s"Temporal table's primary key [$primaryKeyNames] must be included in the equivalence " +
- s"condition of temporal join, but current temporal join condition is [$joinEquiInfo].")
- }
- }
-
- private def extractPrimaryKeyInputRefs(
- leftInput: RelNode,
- snapshot: FlinkLogicalSnapshot,
- snapshotInput: FlinkLogicalRel,
- rexBuilder: RexBuilder): Option[Seq[RexNode]] = {
- val rightFields = snapshot.getRowType.getFieldList
- val fmq = FlinkRelMetadataQuery.reuseOrCreate(snapshot.getCluster.getMetadataQuery)
-
- val upsertKeySet = fmq.getUpsertKeys(snapshotInput)
- val fields = snapshot.getRowType.getFieldList
-
- if (upsertKeySet != null && upsertKeySet.size() > 0) {
- val leftFieldCnt = leftInput.getRowType.getFieldCount
- val upsertKeySetInputRefs = upsertKeySet
- .filter(_.nonEmpty)
- .map(
- _.toArray
- .map(fields)
- // build InputRef of upsert key in snapshot
- .map(f => rexBuilder.makeInputRef(f.getType, leftFieldCnt + rightFields.indexOf(f)))
- .toSeq)
- // select shortest upsert key as primary key
- upsertKeySetInputRefs.toArray
- .sortBy(_.length)
- .headOption
- } else {
- None
- }
- }
-}
-
-object TemporalJoinRewriteWithUniqueKeyRule {
- val INSTANCE: TemporalJoinRewriteWithUniqueKeyRule = new TemporalJoinRewriteWithUniqueKeyRule
-}