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
The following commit(s) were added to refs/heads/master by this push:
new ba95b2eb6f0 [FLINK-38981][table] Migrate
BatchPhysicalSortMergeJoinRule to java
ba95b2eb6f0 is described below
commit ba95b2eb6f0cb8b9f516cbc4b5e746abace903f6
Author: Jacky Lau <[email protected]>
AuthorDate: Fri Mar 6 05:46:43 2026 +0800
[FLINK-38981][table] Migrate BatchPhysicalSortMergeJoinRule to java
Co-authored-by: yongliu <[email protected]>
---
.../batch/BatchPhysicalSortMergeJoinRule.java | 127 +++++++++++++++++++++
.../batch/BatchPhysicalSortMergeJoinRule.scala | 94 ---------------
2 files changed, 127 insertions(+), 94 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalSortMergeJoinRule.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalSortMergeJoinRule.java
new file mode 100644
index 00000000000..516bd29445e
--- /dev/null
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalSortMergeJoinRule.java
@@ -0,0 +1,127 @@
+/*
+ * 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.physical.batch;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.planner.hint.JoinStrategy;
+import org.apache.flink.table.planner.plan.nodes.FlinkConventions;
+import org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalJoin;
+import
org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalSortMergeJoin;
+import org.apache.flink.table.planner.plan.trait.FlinkRelDistribution;
+import org.apache.flink.table.planner.plan.utils.JoinUtil;
+import org.apache.flink.table.planner.utils.ShortcutUtils;
+
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinInfo;
+import org.apache.calcite.util.ImmutableIntList;
+import org.immutables.value.Value;
+
+/**
+ * Rule that converts {@link FlinkLogicalJoin} to {@link
BatchPhysicalSortMergeJoin} if there exists
+ * at least one equal-join condition and SortMergeJoin is enabled.
+ */
[email protected]
+public class BatchPhysicalSortMergeJoinRule
+ extends
RelRule<BatchPhysicalSortMergeJoinRule.BatchPhysicalSortMergeJoinRuleConfig>
+ implements BatchPhysicalJoinRuleBase {
+
+ public static final BatchPhysicalSortMergeJoinRule INSTANCE =
+ BatchPhysicalSortMergeJoinRuleConfig.DEFAULT.toRule();
+
+ protected
BatchPhysicalSortMergeJoinRule(BatchPhysicalSortMergeJoinRuleConfig config) {
+ super(config);
+ }
+
+ @Override
+ public boolean matches(RelOptRuleCall call) {
+ final Join join = call.rel(0);
+ final TableConfig tableConfig = ShortcutUtils.unwrapTableConfig(join);
+ return canUseJoinStrategy(join, tableConfig,
JoinStrategy.SHUFFLE_MERGE);
+ }
+
+ @Override
+ public void onMatch(RelOptRuleCall call) {
+ final Join join = call.rel(0);
+ final JoinInfo joinInfo = join.analyzeCondition();
+
+ final RelNode left = join.getLeft();
+ final RelNode right = join.getRight();
+
+ final RelTraitSet leftRequiredTrait =
+ getTraitSetByShuffleKeys(call, joinInfo.leftKeys, true);
+ final RelTraitSet rightRequiredTrait =
+ getTraitSetByShuffleKeys(call, joinInfo.rightKeys, true);
+
+ final RelNode newLeft = RelRule.convert(left, leftRequiredTrait);
+ final RelNode newRight = RelRule.convert(right, rightRequiredTrait);
+
+ final RelTraitSet providedTraitSet =
+
call.getPlanner().emptyTraitSet().replace(FlinkConventions.BATCH_PHYSICAL());
+ final boolean withJobStrategyHint =
JoinUtil.containsJoinStrategyHint(join.getHints());
+ // do not try to remove redundant sort for shorter optimization time
+ final BatchPhysicalSortMergeJoin newJoin =
+ new BatchPhysicalSortMergeJoin(
+ join.getCluster(),
+ providedTraitSet,
+ newLeft,
+ newRight,
+ join.getCondition(),
+ join.getJoinType(),
+ false,
+ false,
+ withJobStrategyHint);
+ call.transformTo(newJoin);
+ }
+
+ private static RelTraitSet getTraitSetByShuffleKeys(
+ RelOptRuleCall call, ImmutableIntList shuffleKeys, boolean
requireStrict) {
+ return call.getPlanner()
+ .emptyTraitSet()
+ .replace(FlinkConventions.BATCH_PHYSICAL())
+ .replace(FlinkRelDistribution.hash(shuffleKeys,
requireStrict));
+ }
+
+ /** Configuration for {@link BatchPhysicalSortMergeJoinRule}. */
+ @Value.Immutable(singleton = false)
+ public interface BatchPhysicalSortMergeJoinRuleConfig extends
RelRule.Config {
+
+ BatchPhysicalSortMergeJoinRuleConfig DEFAULT =
+
ImmutableBatchPhysicalSortMergeJoinRule.BatchPhysicalSortMergeJoinRuleConfig
+ .builder()
+ .build()
+ .withOperandSupplier(
+ b0 ->
+ b0.operand(FlinkLogicalJoin.class)
+ .oneInput(
+ b1 ->
+
b1.operand(RelNode.class)
+
.anyInputs()))
+ .withDescription("BatchPhysicalSortMergeJoinRule")
+ .as(BatchPhysicalSortMergeJoinRuleConfig.class);
+
+ @Override
+ default BatchPhysicalSortMergeJoinRule toRule() {
+ return new BatchPhysicalSortMergeJoinRule(this);
+ }
+ }
+}
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalSortMergeJoinRule.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalSortMergeJoinRule.scala
deleted file mode 100644
index b6ca68fe87a..00000000000
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalSortMergeJoinRule.scala
+++ /dev/null
@@ -1,94 +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.physical.batch
-
-import org.apache.flink.table.planner.hint.JoinStrategy
-import org.apache.flink.table.planner.plan.`trait`.FlinkRelDistribution
-import org.apache.flink.table.planner.plan.nodes.FlinkConventions
-import org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalJoin
-import
org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalSortMergeJoin
-import org.apache.flink.table.planner.plan.utils.JoinUtil
-import org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConfig
-
-import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall, RelTraitSet}
-import org.apache.calcite.plan.RelOptRule.{any, operand}
-import org.apache.calcite.rel.RelNode
-import org.apache.calcite.rel.core.Join
-import org.apache.calcite.util.ImmutableIntList
-
-/**
- * Rule that converts [[FlinkLogicalJoin]] to [[BatchPhysicalSortMergeJoin]]
if there exists at
- * least one equal-join condition and SortMergeJoin is enabled.
- */
-class BatchPhysicalSortMergeJoinRule
- extends RelOptRule(
- operand(classOf[FlinkLogicalJoin], operand(classOf[RelNode], any)),
- "BatchPhysicalSortMergeJoinRule")
- with BatchPhysicalJoinRuleBase {
-
- override def matches(call: RelOptRuleCall): Boolean = {
- val join: Join = call.rel(0)
- val tableConfig = unwrapTableConfig(join)
- canUseJoinStrategy(join, tableConfig, JoinStrategy.SHUFFLE_MERGE)
- }
-
- override def onMatch(call: RelOptRuleCall): Unit = {
- val join: Join = call.rel(0)
- val joinInfo = join.analyzeCondition
- val left = join.getLeft
- val right = join.getRight
-
- def getTraitSetByShuffleKeys(
- shuffleKeys: ImmutableIntList,
- requireStrict: Boolean): RelTraitSet = {
- call.getPlanner
- .emptyTraitSet()
- .replace(FlinkConventions.BATCH_PHYSICAL)
- .replace(FlinkRelDistribution.hash(shuffleKeys, requireStrict))
- }
-
- val leftRequiredTrait =
- getTraitSetByShuffleKeys(joinInfo.leftKeys, requireStrict = true)
- val rightRequiredTrait =
- getTraitSetByShuffleKeys(joinInfo.rightKeys, requireStrict = true)
-
- val newLeft = RelOptRule.convert(left, leftRequiredTrait)
- val newRight = RelOptRule.convert(right, rightRequiredTrait)
-
- val providedTraitSet = call.getPlanner
- .emptyTraitSet()
- .replace(FlinkConventions.BATCH_PHYSICAL)
- val withJobStrategyHint = JoinUtil.containsJoinStrategyHint(join.getHints)
- // do not try to remove redundant sort for shorter optimization time
- val newJoin = new BatchPhysicalSortMergeJoin(
- join.getCluster,
- providedTraitSet,
- newLeft,
- newRight,
- join.getCondition,
- join.getJoinType,
- false,
- false,
- withJobStrategyHint)
- call.transformTo(newJoin)
- }
-}
-
-object BatchPhysicalSortMergeJoinRule {
- val INSTANCE: RelOptRule = new BatchPhysicalSortMergeJoinRule
-}
