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 60846c0cdbf [FLINK-40797][table-planner] Exclude `NestedLoopJoin` from
CBO search space for equi-joins
60846c0cdbf is described below
commit 60846c0cdbf57d5297c44f218025b189d18a2765
Author: Bonnie Varghese <[email protected]>
AuthorDate: Fri Oct 9 09:08:12 2026 -0700
[FLINK-40797][table-planner] Exclude `NestedLoopJoin` from CBO search space
for equi-joins
---
.../table/planner/codegen/GenerateUtils.scala | 1 +
.../physical/batch/BatchPhysicalJoinRuleBase.scala | 76 +++++-
.../sql/join/NestedLoopJoinExclusionTest.java | 220 ++++++++++++++++++
.../program/FlinkRuntimeFilterProgramTest.java | 6 +-
.../planner/plan/batch/sql/SubplanReuseTest.xml | 6 +-
.../batch/sql/join/NestedLoopJoinExclusionTest.xml | 255 +++++++++++++++++++++
.../table/planner/plan/optimize/ScanReuseTest.xml | 8 +-
.../program/DynamicPartitionPruningProgramTest.xml | 10 +-
.../program/FlinkRuntimeFilterProgramTest.xml | 8 +-
9 files changed, 565 insertions(+), 25 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala
index 04ea9a9c8c7..b4874396122 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala
@@ -638,6 +638,7 @@ object GenerateUtils {
case TINYINT | SMALLINT | INTEGER | BIGINT | FLOAT | DOUBLE | DATE |
TIME_WITHOUT_TIME_ZONE |
INTERVAL_YEAR_MONTH | INTERVAL_DAY_TIME =>
s"($leftTerm > $rightTerm ? 1 : $leftTerm < $rightTerm ? -1 : 0)"
+ // keep in sync with BatchPhysicalJoinRuleBase#hasOrderableJoinKeys
case TIMESTAMP_WITH_TIME_ZONE | MULTISET | MAP | VARIANT | BITMAP =>
throw new ValidationException(
s"Type '$t' cannot be ordered, so it cannot be used as a key for
sorting, grouping, " +
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalJoinRuleBase.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalJoinRuleBase.scala
index d895c33ff06..7b1481f04a8 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalJoinRuleBase.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalJoinRuleBase.scala
@@ -19,12 +19,15 @@ package
org.apache.flink.table.planner.plan.rules.physical.batch
import org.apache.flink.table.api.{TableConfig, TableException,
ValidationException}
import org.apache.flink.table.api.config.OptimizerConfigOptions
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory
import org.apache.flink.table.planner.hint.{FlinkHints, JoinStrategy}
import org.apache.flink.table.planner.plan.nodes.FlinkConventions
import
org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalLocalHashAggregate
import
org.apache.flink.table.planner.plan.rules.physical.batch.BatchPhysicalJoinRuleBase.SEMI_JOIN_BUILD_DISTINCT_NDV_RATIO
import org.apache.flink.table.planner.plan.utils.{JoinUtil, OperatorType}
import org.apache.flink.table.planner.utils.TableConfigUtils.isOperatorDisabled
+import org.apache.flink.table.types.logical.LogicalTypeRoot._
+import org.apache.flink.table.types.logical.utils.LogicalTypeChecks.hasNested
import org.apache.calcite.plan.RelOptRule
import org.apache.calcite.rel.RelNode
@@ -153,6 +156,32 @@ trait BatchPhysicalJoinRuleBase {
!joinInfo.pairs().isEmpty
}
+ /**
+ * Whether all equi-join key columns, including nested fields, can be
ordered. Every equi-capable
+ * strategy needs this, as the hash join also builds a sort merge fallback.
Mirrors
+ * `GenerateUtils.generateCompare`; RAW is conservatively treated as
unorderable.
+ *
+ * Only used to decide whether NestedLoopJoin can be excluded; the
equi-capable strategies still
+ * do not check key types themselves.
+ */
+ private def hasOrderableJoinKeys(join: Join): Boolean = {
+ val joinInfo = join.analyzeCondition
+ val leftType = FlinkTypeFactory.toLogicalRowType(join.getLeft.getRowType)
+ val rightType = FlinkTypeFactory.toLogicalRowType(join.getRight.getRowType)
+ val keyTypes = joinInfo.leftKeys.map(i => leftType.getTypeAt(i)) ++
+ joinInfo.rightKeys.map(i => rightType.getTypeAt(i))
+ !keyTypes.exists(
+ hasNested(_, _.isAnyOf(TIMESTAMP_WITH_TIME_ZONE, MULTISET, MAP, VARIANT,
BITMAP, RAW)))
+ }
+
+ private def hasApplicableEquivJoinStrategy(join: Join, tableConfig:
TableConfig): Boolean = {
+ isEquivJoin(join) && hasOrderableJoinKeys(join) && (
+ checkSortMergeJoinApplicable(tableConfig) ||
+ checkShuffleHashApplicable(join, tableConfig, withShuffleHashHint =
false)._1 ||
+ checkBroadcastApplicable(join, tableConfig, withBroadcastHint =
false)._1
+ )
+ }
+
/**
* Decides whether the join can convert to BroadcastHashJoin.
*
@@ -168,7 +197,18 @@ trait BatchPhysicalJoinRuleBase {
tableConfig: TableConfig,
withBroadcastHint: Boolean): (Boolean, Boolean) = {
- if (!isEquivJoin(join) || isOperatorDisabled(tableConfig,
OperatorType.BroadcastHashJoin)) {
+ if (!isEquivJoin(join)) {
+ return (false, false)
+ }
+ checkBroadcastApplicable(join, tableConfig, withBroadcastHint)
+ }
+
+ private def checkBroadcastApplicable(
+ join: Join,
+ tableConfig: TableConfig,
+ withBroadcastHint: Boolean): (Boolean, Boolean) = {
+
+ if (isOperatorDisabled(tableConfig, OperatorType.BroadcastHashJoin)) {
return (false, false)
}
@@ -235,7 +275,17 @@ trait BatchPhysicalJoinRuleBase {
join: Join,
tableConfig: TableConfig,
withShuffleHashHint: Boolean): (Boolean, Boolean) = {
- if (!isEquivJoin(join) || isOperatorDisabled(tableConfig,
OperatorType.ShuffleHashJoin)) {
+ if (!isEquivJoin(join)) {
+ return (false, false)
+ }
+ checkShuffleHashApplicable(join, tableConfig, withShuffleHashHint)
+ }
+
+ private def checkShuffleHashApplicable(
+ join: Join,
+ tableConfig: TableConfig,
+ withShuffleHashHint: Boolean): (Boolean, Boolean) = {
+ if (isOperatorDisabled(tableConfig, OperatorType.ShuffleHashJoin)) {
return (false, false)
}
@@ -260,11 +310,11 @@ trait BatchPhysicalJoinRuleBase {
// the sort merge join doesn't distinct the build side
protected def checkSortMergeJoin(join: Join, tableConfig: TableConfig):
Boolean = {
- if (!isEquivJoin(join) || isOperatorDisabled(tableConfig,
OperatorType.SortMergeJoin)) {
- false
- } else {
- true
- }
+ isEquivJoin(join) && checkSortMergeJoinApplicable(tableConfig)
+ }
+
+ private def checkSortMergeJoinApplicable(tableConfig: TableConfig): Boolean
= {
+ !isOperatorDisabled(tableConfig, OperatorType.SortMergeJoin)
}
protected def checkNestLoopJoin(
@@ -276,6 +326,10 @@ trait BatchPhysicalJoinRuleBase {
return (false, false)
}
+ if (!withNestLoopHint && hasApplicableEquivJoinStrategy(join,
tableConfig)) {
+ return (false, false)
+ }
+
val isLeftToBuild = if (withNestLoopHint) {
getFirstArgInJoinHint(join, JoinStrategy.NEST_LOOP.getJoinHintName)
.equals(FlinkHints.LEFT_INPUT)
@@ -297,7 +351,13 @@ trait BatchPhysicalJoinRuleBase {
}
- // all join can use NEST LOOP JOIN
+ // reached only for:
+ // 1. non-equi joins
+ // 2. equi-joins with an explicit NEST_LOOP hint
+ // 3. equi-joins where no equi-capable strategy is applicable (e.g.
ShuffleHashJoin/SortMergeJoin
+ // disabled and BroadcastHashJoin inapplicable, due to unknown or
oversized inputs)
+ // 4. equi-joins on keys that cannot be ordered (e.g. MAP), which no
equi-capable strategy
+ // supports
(true, isLeftToBuild)
}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/batch/sql/join/NestedLoopJoinExclusionTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/batch/sql/join/NestedLoopJoinExclusionTest.java
new file mode 100644
index 00000000000..14e1f391f48
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/batch/sql/join/NestedLoopJoinExclusionTest.java
@@ -0,0 +1,220 @@
+/*
+ * 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.batch.sql.join;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.MapTypeInfo;
+import org.apache.flink.api.java.typeutils.MultisetTypeInfo;
+import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.config.ExecutionConfigOptions;
+import org.apache.flink.table.plan.stats.TableStats;
+import org.apache.flink.table.planner.plan.stats.FlinkStatistic;
+import org.apache.flink.table.planner.utils.BatchTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Plan tests for excluding {@code NestedLoopJoin} from the CBO's candidate
set for equi-joins,
+ * added by {@link
+ *
org.apache.flink.table.planner.plan.rules.physical.batch.BatchPhysicalJoinRuleBase#checkNestLoopJoin}.
+ *
+ * <p>For an equi-join, nested-loop join must never be picked over an
equi-capable strategy, even
+ * when the cost-based cardinality estimate makes it look cheap.
+ */
+public class NestedLoopJoinExclusionTest extends TableTestBase {
+
+ private BatchTableTestUtil util;
+
+ private static final TypeInformation<?> INT_INFO =
BasicTypeInfo.INT_TYPE_INFO;
+ private static final TypeInformation<?> MAP_INFO = new
MapTypeInfo<>(INT_INFO, INT_INFO);
+
+ @BeforeEach
+ void setup() {
+ util = batchTestUtil(TableConfig.getDefault());
+ }
+
+ @Test
+ void testEquiJoinWithSkewedCardinalityDoesNotUseNestedLoopJoin() {
+ // Left side's estimated row count is tiny relative to the right side.
+ // NestedLoopJoin's L*R cost (~2M) looks cheaper than HashJoin's
8*(L+R) cost (~8M),
+ // even though the join condition is a plain equality.
+ util.addTableSource(
+ "TinySide",
+ new TypeInformation<?>[] {INT_INFO, INT_INFO},
+ new String[] {"a", "b"},
+ FlinkStatistic.builder().tableStats(new
TableStats(2L)).build());
+ util.addTableSource(
+ "HugeSide",
+ new TypeInformation<?>[] {INT_INFO, INT_INFO},
+ new String[] {"d", "e"},
+ FlinkStatistic.builder().tableStats(new
TableStats(1_000_000L)).build());
+
+ util.verifyRelPlan("SELECT a, d FROM TinySide, HugeSide WHERE a = d");
+ }
+
+ @Test
+ void testEquiJoinOnComplexKeyStillProducesValidPlan() {
+ // An equi-join on a complex (ARRAY) key still yields a valid plan via
an
+ // equi-capable strategy rather than being stranded with no candidate
at all.
+ TypeInformation<?> arrayInfo =
PrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO;
+ util.addTableSource(
+ "ArrLeft",
+ new TypeInformation<?>[] {INT_INFO, arrayInfo},
+ new String[] {"a", "r"},
+ FlinkStatistic.UNKNOWN());
+ util.addTableSource(
+ "ArrRight",
+ new TypeInformation<?>[] {INT_INFO, arrayInfo},
+ new String[] {"d", "s"},
+ FlinkStatistic.UNKNOWN());
+
+ util.verifyRelPlan("SELECT a, d FROM ArrLeft, ArrRight WHERE r = s");
+ }
+
+ @Test
+ void
testEquiJoinFallsBackToNestedLoopJoinWhenOnlyInapplicableBroadcastIsEnabled() {
+ // ShuffleHashJoin/SortMergeJoin disabled; BroadcastHashJoin stays
enabled but cannot
+ // apply since both sides have unknown size stats. NestedLoopJoin must
remain available
+ // as a fallback here, otherwise the equi-join is left with no valid
physical strategy.
+ util.tableEnv()
+ .getConfig()
+ .set(
+ ExecutionConfigOptions.TABLE_EXEC_DISABLED_OPERATORS,
+ "ShuffleHashJoin, SortMergeJoin");
+ util.addTableSource(
+ "LeftUnknown",
+ new TypeInformation<?>[] {INT_INFO, INT_INFO},
+ new String[] {"a", "b"},
+ FlinkStatistic.UNKNOWN());
+ util.addTableSource(
+ "RightUnknown",
+ new TypeInformation<?>[] {INT_INFO, INT_INFO},
+ new String[] {"d", "e"},
+ FlinkStatistic.UNKNOWN());
+
+ util.verifyRelPlan("SELECT a, d FROM LeftUnknown, RightUnknown WHERE a
= d");
+ }
+
+ @Test
+ void testEquiJoinStillAvoidsNestedLoopJoinWhenBroadcastIsApplicable() {
+ // Same skewed-cardinality setup as
+ // testEquiJoinWithSkewedCardinalityDoesNotUseNestedLoopJoin, but with
+ // ShuffleHashJoin/SortMergeJoin disabled so only BroadcastHashJoin is
applicable
+ util.tableEnv()
+ .getConfig()
+ .set(
+ ExecutionConfigOptions.TABLE_EXEC_DISABLED_OPERATORS,
+ "ShuffleHashJoin, SortMergeJoin");
+ util.addTableSource(
+ "TinySide",
+ new TypeInformation<?>[] {INT_INFO, INT_INFO},
+ new String[] {"a", "b"},
+ FlinkStatistic.builder().tableStats(new
TableStats(2L)).build());
+ util.addTableSource(
+ "HugeSide",
+ new TypeInformation<?>[] {INT_INFO, INT_INFO},
+ new String[] {"d", "e"},
+ FlinkStatistic.builder().tableStats(new
TableStats(1_000_000L)).build());
+
+ util.verifyRelPlan("SELECT a, d FROM TinySide, HugeSide WHERE a = d");
+ }
+
+ @Test
+ void testEquiJoinOnMapKeyKeepsNestedLoopJoinWhenHashJoinsAreDisabled() {
+ // MAP cannot be ordered, so SortMergeJoin cannot handle it either.
NestedLoopJoin is the
+ // only strategy that can run, so it must not be excluded.
+ util.tableEnv()
+ .getConfig()
+ .set(
+ ExecutionConfigOptions.TABLE_EXEC_DISABLED_OPERATORS,
+ "ShuffleHashJoin, BroadcastHashJoin");
+ addSkewedTables(MAP_INFO);
+
+ util.verifyRelPlan("SELECT a, d FROM TinySide, HugeSide WHERE m = n");
+ }
+
+ @Test
+ void testEquiJoinOnMapKeyKeepsNestedLoopJoinWhenAllStrategiesAreEnabled() {
+ // The hash joins need an orderable key too (they build a sort merge
fallback).
+ addSkewedTables(MAP_INFO);
+
+ util.verifyRelPlan("SELECT a, d FROM TinySide, HugeSide WHERE m = n");
+ }
+
+ @Test
+ void testEquiJoinOnMultisetKeyKeepsNestedLoopJoin() {
+ addSkewedTables(new MultisetTypeInfo<>(INT_INFO));
+
+ util.verifyRelPlan("SELECT a, d FROM TinySide, HugeSide WHERE m = n");
+ }
+
+ @Test
+ void testEquiJoinOnNestedMapKeyKeepsNestedLoopJoin() {
+ // The unorderable type is nested inside ARRAY and ROW.
+ addSkewedTables(
+ ObjectArrayTypeInfo.getInfoFor(
+ new RowTypeInfo(new TypeInformation<?>[] {MAP_INFO},
new String[] {"f"})));
+
+ util.verifyRelPlan("SELECT a, d FROM TinySide, HugeSide WHERE m = n");
+ }
+
+ @Test
+ void testEquiJoinOnCompositeKeyWithMapKeepsNestedLoopJoin() {
+ // One unorderable key column is enough: the orderable column does not
rescue the join.
+ util.addTableSource(
+ "TinySide",
+ new TypeInformation<?>[] {INT_INFO, INT_INFO, MAP_INFO},
+ new String[] {"a", "b", "m"},
+ FlinkStatistic.builder().tableStats(new
TableStats(2L)).build());
+ util.addTableSource(
+ "HugeSide",
+ new TypeInformation<?>[] {INT_INFO, INT_INFO, MAP_INFO},
+ new String[] {"d", "e", "n"},
+ FlinkStatistic.builder().tableStats(new
TableStats(1_000_000L)).build());
+
+ util.verifyRelPlan("SELECT a, d FROM TinySide, HugeSide WHERE a = d
AND m = n");
+ }
+
+ @Test
+ void testEquiJoinWithMapNonKeyColumnStillAvoidsNestedLoopJoin() {
+ // Only the equi-key columns matter; a MAP payload column must not
block a hash join.
+ addSkewedTables(MAP_INFO);
+
+ util.verifyRelPlan("SELECT a, d, m, n FROM TinySide, HugeSide WHERE a
= d");
+ }
+
+ private void addSkewedTables(TypeInformation<?> mapColumnType) {
+ util.addTableSource(
+ "TinySide",
+ new TypeInformation<?>[] {INT_INFO, mapColumnType},
+ new String[] {"a", "m"},
+ FlinkStatistic.builder().tableStats(new
TableStats(2L)).build());
+ util.addTableSource(
+ "HugeSide",
+ new TypeInformation<?>[] {INT_INFO, mapColumnType},
+ new String[] {"d", "n"},
+ FlinkStatistic.builder().tableStats(new
TableStats(1_000_000L)).build());
+ }
+}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/optimize/program/FlinkRuntimeFilterProgramTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/optimize/program/FlinkRuntimeFilterProgramTest.java
index 0031885a858..dec87fcbe35 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/optimize/program/FlinkRuntimeFilterProgramTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/optimize/program/FlinkRuntimeFilterProgramTest.java
@@ -152,7 +152,11 @@ public class FlinkRuntimeFilterProgramTest extends
TableTestBase {
// runtime filter will not succeed
setupTableRowCount("dim", 1L);
setupTableRowCount("fact", SUITABLE_FACT_ROW_COUNT);
- String query = "select * from fact, dim where fact.amount = dim.amount
and dim.price < 500";
+ // force nested loop join explicitly using a hint
+ // with an equi-condition, the CBO now prefers a hash join over nested
loop join
+ String query =
+ "SELECT /*+ NEST_LOOP(dim) */ * FROM fact, dim"
+ + " WHERE fact.amount = dim.amount AND dim.price <
500";
util.verifyPlan(query);
}
diff --git
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/SubplanReuseTest.xml
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/SubplanReuseTest.xml
index 2120efbd43b..411e16c1095 100644
---
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/SubplanReuseTest.xml
+++
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/SubplanReuseTest.xml
@@ -1174,7 +1174,7 @@ LogicalProject(c=[$0], a=[$1], b=[$2], c0=[$3], a0=[$4],
b0=[$5])
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-NestedLoopJoin(joinType=[InnerJoin], where=[(a = a0)], select=[c, a, b, c0,
a0, b0], build=[right])
+HashJoin(joinType=[InnerJoin], where=[(a = a0)], select=[c, a, b, c0, a0, b0],
isBroadcast=[true], build=[right])
:- Exchange(distribution=[any], shuffle_mode=[BATCH])
: +- Calc(select=[c, a, b], where=[(a > 1)])
: +- SortLimit(orderBy=[a ASC, b DESC], offset=[0], fetch=[10],
global=[true])(reuse_id=[1])
@@ -1362,7 +1362,7 @@ LogicalIntersect(all=[false])
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-MultipleInput(readOrder=[0,1],
members=[\nNestedLoopJoin(joinType=[LeftSemiJoin], where=[IS NOT DISTINCT
FROM(random, random0)], select=[random], build=[right])\n:-
SortAggregate(isMerge=[false], groupBy=[random], select=[random])\n: +-
Sort(orderBy=[random ASC])\n: +- [#2]
Exchange(distribution=[hash[random]])\n+- [#1]
Exchange(distribution=[broadcast])\n])
+MultipleInput(readOrder=[0,1], members=[\nHashJoin(joinType=[LeftSemiJoin],
where=[IS NOT DISTINCT FROM(random, random0)], select=[random],
isBroadcast=[true], build=[right])\n:- SortAggregate(isMerge=[false],
groupBy=[random], select=[random])\n: +- Sort(orderBy=[random ASC])\n: +-
[#2] Exchange(distribution=[hash[random]])\n+- [#1]
Exchange(distribution=[broadcast])\n])
:- Exchange(distribution=[broadcast])(reuse_id=[2])
: +- Calc(select=[random])(reuse_id=[1])
: +- SortLimit(orderBy=[EXPR$1 ASC], offset=[0], fetch=[1], global=[true])
@@ -1371,7 +1371,7 @@ MultipleInput(readOrder=[0,1],
members=[\nNestedLoopJoin(joinType=[LeftSemiJoin]
: +- Calc(select=[a AS random, RAND() AS EXPR$1])
: +- TableSourceScan(table=[[default_catalog,
default_database, x]], fields=[a, b, c])
+- Exchange(distribution=[hash[random]])
- +- NestedLoopJoin(joinType=[LeftSemiJoin], where=[IS NOT DISTINCT
FROM(random, random0)], select=[random], build=[right])
+ +- HashJoin(joinType=[LeftSemiJoin], where=[IS NOT DISTINCT FROM(random,
random0)], select=[random], isBroadcast=[true], build=[right])
:- Exchange(distribution=[any], shuffle_mode=[BATCH])
: +- Reused(reference_id=[1])
+- Reused(reference_id=[2])
diff --git
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/NestedLoopJoinExclusionTest.xml
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/NestedLoopJoinExclusionTest.xml
new file mode 100644
index 00000000000..d1986272b94
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/NestedLoopJoinExclusionTest.xml
@@ -0,0 +1,255 @@
+<?xml version="1.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.
+-->
+<Root>
+ <TestCase
name="testEquiJoinFallsBackToNestedLoopJoinWhenOnlyInapplicableBroadcastIsEnabled">
+ <Resource name="sql">
+ <![CDATA[SELECT a, d FROM LeftUnknown, RightUnknown WHERE a = d]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(a=[$0], d=[$2])
++- LogicalFilter(condition=[=($0, $2)])
+ +- LogicalJoin(condition=[true], joinType=[inner])
+ :- LogicalTableScan(table=[[default_catalog, default_database,
LeftUnknown, source: [TestTableSource(a, b)]]])
+ +- LogicalTableScan(table=[[default_catalog, default_database,
RightUnknown, source: [TestTableSource(d, e)]]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+NestedLoopJoin(joinType=[InnerJoin], where=[=(a, d)], select=[a, d],
build=[left])
+:- Exchange(distribution=[broadcast])
+: +- Calc(select=[a])
+: +- LegacyTableSourceScan(table=[[default_catalog, default_database,
LeftUnknown, source: [TestTableSource(a, b)]]], fields=[a, b])
++- Calc(select=[d])
+ +- LegacyTableSourceScan(table=[[default_catalog, default_database,
RightUnknown, source: [TestTableSource(d, e)]]], fields=[d, e])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testEquiJoinOnComplexKeyStillProducesValidPlan">
+ <Resource name="sql">
+ <![CDATA[SELECT a, d FROM ArrLeft, ArrRight WHERE r = s]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(a=[$0], d=[$2])
++- LogicalFilter(condition=[=($1, $3)])
+ +- LogicalJoin(condition=[true], joinType=[inner])
+ :- LogicalTableScan(table=[[default_catalog, default_database, ArrLeft,
source: [TestTableSource(a, r)]]])
+ +- LogicalTableScan(table=[[default_catalog, default_database, ArrRight,
source: [TestTableSource(d, s)]]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Calc(select=[a, d])
++- HashJoin(joinType=[InnerJoin], where=[=(r, s)], select=[a, r, d, s],
build=[left])
+ :- Exchange(distribution=[hash[r]])
+ : +- LegacyTableSourceScan(table=[[default_catalog, default_database,
ArrLeft, source: [TestTableSource(a, r)]]], fields=[a, r])
+ +- Exchange(distribution=[hash[s]])
+ +- LegacyTableSourceScan(table=[[default_catalog, default_database,
ArrRight, source: [TestTableSource(d, s)]]], fields=[d, s])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testEquiJoinOnCompositeKeyWithMapKeepsNestedLoopJoin">
+ <Resource name="sql">
+ <![CDATA[SELECT a, d FROM TinySide, HugeSide WHERE a = d AND m = n]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(a=[$0], d=[$3])
++- LogicalFilter(condition=[AND(=($0, $3), =($2, $5))])
+ +- LogicalJoin(condition=[true], joinType=[inner])
+ :- LogicalTableScan(table=[[default_catalog, default_database, TinySide,
source: [TestTableSource(a, b, m)]]])
+ +- LogicalTableScan(table=[[default_catalog, default_database, HugeSide,
source: [TestTableSource(d, e, n)]]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Calc(select=[a, d])
++- NestedLoopJoin(joinType=[InnerJoin], where=[AND(=(a, d), =(m, n))],
select=[a, m, d, n], build=[left])
+ :- Exchange(distribution=[broadcast])
+ : +- Calc(select=[a, m])
+ : +- LegacyTableSourceScan(table=[[default_catalog, default_database,
TinySide, source: [TestTableSource(a, b, m)]]], fields=[a, b, m])
+ +- Calc(select=[d, n])
+ +- LegacyTableSourceScan(table=[[default_catalog, default_database,
HugeSide, source: [TestTableSource(d, e, n)]]], fields=[d, e, n])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase
name="testEquiJoinOnMapKeyKeepsNestedLoopJoinWhenAllStrategiesAreEnabled">
+ <Resource name="sql">
+ <![CDATA[SELECT a, d FROM TinySide, HugeSide WHERE m = n]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(a=[$0], d=[$2])
++- LogicalFilter(condition=[=($1, $3)])
+ +- LogicalJoin(condition=[true], joinType=[inner])
+ :- LogicalTableScan(table=[[default_catalog, default_database, TinySide,
source: [TestTableSource(a, m)]]])
+ +- LogicalTableScan(table=[[default_catalog, default_database, HugeSide,
source: [TestTableSource(d, n)]]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Calc(select=[a, d])
++- NestedLoopJoin(joinType=[InnerJoin], where=[=(m, n)], select=[a, m, d, n],
build=[left])
+ :- Exchange(distribution=[broadcast])
+ : +- LegacyTableSourceScan(table=[[default_catalog, default_database,
TinySide, source: [TestTableSource(a, m)]]], fields=[a, m])
+ +- LegacyTableSourceScan(table=[[default_catalog, default_database,
HugeSide, source: [TestTableSource(d, n)]]], fields=[d, n])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase
name="testEquiJoinOnMapKeyKeepsNestedLoopJoinWhenHashJoinsAreDisabled">
+ <Resource name="sql">
+ <![CDATA[SELECT a, d FROM TinySide, HugeSide WHERE m = n]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(a=[$0], d=[$2])
++- LogicalFilter(condition=[=($1, $3)])
+ +- LogicalJoin(condition=[true], joinType=[inner])
+ :- LogicalTableScan(table=[[default_catalog, default_database, TinySide,
source: [TestTableSource(a, m)]]])
+ +- LogicalTableScan(table=[[default_catalog, default_database, HugeSide,
source: [TestTableSource(d, n)]]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Calc(select=[a, d])
++- NestedLoopJoin(joinType=[InnerJoin], where=[=(m, n)], select=[a, m, d, n],
build=[left])
+ :- Exchange(distribution=[broadcast])
+ : +- LegacyTableSourceScan(table=[[default_catalog, default_database,
TinySide, source: [TestTableSource(a, m)]]], fields=[a, m])
+ +- LegacyTableSourceScan(table=[[default_catalog, default_database,
HugeSide, source: [TestTableSource(d, n)]]], fields=[d, n])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testEquiJoinOnMultisetKeyKeepsNestedLoopJoin">
+ <Resource name="sql">
+ <![CDATA[SELECT a, d FROM TinySide, HugeSide WHERE m = n]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(a=[$0], d=[$2])
++- LogicalFilter(condition=[=($1, $3)])
+ +- LogicalJoin(condition=[true], joinType=[inner])
+ :- LogicalTableScan(table=[[default_catalog, default_database, TinySide,
source: [TestTableSource(a, m)]]])
+ +- LogicalTableScan(table=[[default_catalog, default_database, HugeSide,
source: [TestTableSource(d, n)]]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Calc(select=[a, d])
++- NestedLoopJoin(joinType=[InnerJoin], where=[=(m, n)], select=[a, m, d, n],
build=[left])
+ :- Exchange(distribution=[broadcast])
+ : +- LegacyTableSourceScan(table=[[default_catalog, default_database,
TinySide, source: [TestTableSource(a, m)]]], fields=[a, m])
+ +- LegacyTableSourceScan(table=[[default_catalog, default_database,
HugeSide, source: [TestTableSource(d, n)]]], fields=[d, n])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testEquiJoinOnNestedMapKeyKeepsNestedLoopJoin">
+ <Resource name="sql">
+ <![CDATA[SELECT a, d FROM TinySide, HugeSide WHERE m = n]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(a=[$0], d=[$2])
++- LogicalFilter(condition=[=($1, $3)])
+ +- LogicalJoin(condition=[true], joinType=[inner])
+ :- LogicalTableScan(table=[[default_catalog, default_database, TinySide,
source: [TestTableSource(a, m)]]])
+ +- LogicalTableScan(table=[[default_catalog, default_database, HugeSide,
source: [TestTableSource(d, n)]]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Calc(select=[a, d])
++- NestedLoopJoin(joinType=[InnerJoin], where=[=(m, n)], select=[a, m, d, n],
build=[left])
+ :- Exchange(distribution=[broadcast])
+ : +- LegacyTableSourceScan(table=[[default_catalog, default_database,
TinySide, source: [TestTableSource(a, m)]]], fields=[a, m])
+ +- LegacyTableSourceScan(table=[[default_catalog, default_database,
HugeSide, source: [TestTableSource(d, n)]]], fields=[d, n])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase
name="testEquiJoinStillAvoidsNestedLoopJoinWhenBroadcastIsApplicable">
+ <Resource name="sql">
+ <![CDATA[SELECT a, d FROM TinySide, HugeSide WHERE a = d]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(a=[$0], d=[$2])
++- LogicalFilter(condition=[=($0, $2)])
+ +- LogicalJoin(condition=[true], joinType=[inner])
+ :- LogicalTableScan(table=[[default_catalog, default_database, TinySide,
source: [TestTableSource(a, b)]]])
+ +- LogicalTableScan(table=[[default_catalog, default_database, HugeSide,
source: [TestTableSource(d, e)]]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+HashJoin(joinType=[InnerJoin], where=[=(a, d)], select=[a, d],
isBroadcast=[true], build=[left])
+:- Exchange(distribution=[broadcast])
+: +- Calc(select=[a])
+: +- LegacyTableSourceScan(table=[[default_catalog, default_database,
TinySide, source: [TestTableSource(a, b)]]], fields=[a, b])
++- Calc(select=[d])
+ +- LegacyTableSourceScan(table=[[default_catalog, default_database,
HugeSide, source: [TestTableSource(d, e)]]], fields=[d, e])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testEquiJoinWithMapNonKeyColumnStillAvoidsNestedLoopJoin">
+ <Resource name="sql">
+ <![CDATA[SELECT a, d, m, n FROM TinySide, HugeSide WHERE a = d]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(a=[$0], d=[$2], m=[$1], n=[$3])
++- LogicalFilter(condition=[=($0, $2)])
+ +- LogicalJoin(condition=[true], joinType=[inner])
+ :- LogicalTableScan(table=[[default_catalog, default_database, TinySide,
source: [TestTableSource(a, m)]]])
+ +- LogicalTableScan(table=[[default_catalog, default_database, HugeSide,
source: [TestTableSource(d, n)]]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Calc(select=[a, d, m, n])
++- HashJoin(joinType=[InnerJoin], where=[=(a, d)], select=[a, m, d, n],
isBroadcast=[true], build=[left])
+ :- Exchange(distribution=[broadcast])
+ : +- LegacyTableSourceScan(table=[[default_catalog, default_database,
TinySide, source: [TestTableSource(a, m)]]], fields=[a, m])
+ +- LegacyTableSourceScan(table=[[default_catalog, default_database,
HugeSide, source: [TestTableSource(d, n)]]], fields=[d, n])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testEquiJoinWithSkewedCardinalityDoesNotUseNestedLoopJoin">
+ <Resource name="sql">
+ <![CDATA[SELECT a, d FROM TinySide, HugeSide WHERE a = d]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(a=[$0], d=[$2])
++- LogicalFilter(condition=[=($0, $2)])
+ +- LogicalJoin(condition=[true], joinType=[inner])
+ :- LogicalTableScan(table=[[default_catalog, default_database, TinySide,
source: [TestTableSource(a, b)]]])
+ +- LogicalTableScan(table=[[default_catalog, default_database, HugeSide,
source: [TestTableSource(d, e)]]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+HashJoin(joinType=[InnerJoin], where=[=(a, d)], select=[a, d],
isBroadcast=[true], build=[left])
+:- Exchange(distribution=[broadcast])
+: +- Calc(select=[a])
+: +- LegacyTableSourceScan(table=[[default_catalog, default_database,
TinySide, source: [TestTableSource(a, b)]]], fields=[a, b])
++- Calc(select=[d])
+ +- LegacyTableSourceScan(table=[[default_catalog, default_database,
HugeSide, source: [TestTableSource(d, e)]]], fields=[d, e])
+]]>
+ </Resource>
+ </TestCase>
+</Root>
diff --git
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/optimize/ScanReuseTest.xml
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/optimize/ScanReuseTest.xml
index 8126119b1a6..155081bd8bf 100644
---
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/optimize/ScanReuseTest.xml
+++
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/optimize/ScanReuseTest.xml
@@ -449,14 +449,14 @@ LogicalProject(a=[$0], c=[$2], c0=[$9])
<Resource name="optimized exec plan">
<![CDATA[
Calc(select=[a, c, c0])
-+- NestedLoopJoin(joinType=[InnerJoin], where=[(a = a0)], select=[a, c, a0,
c0], build=[left])
- :- Exchange(distribution=[broadcast])
++- HashJoin(joinType=[InnerJoin], where=[(a = a0)], select=[a, c, a0, c0],
isBroadcast=[true], build=[right])
+ :- Exchange(distribution=[any], shuffle_mode=[BATCH])
: +- Calc(select=[a, c])(reuse_id=[1])
: +- Limit(offset=[0], fetch=[10], global=[true])
: +- Exchange(distribution=[single])
: +- Limit(offset=[0], fetch=[10], global=[false])
: +- TableSourceScan(table=[[default_catalog,
default_database, MyTable, limit=[10]]], fields=[a, b, c, nested, metadata_1,
metadata_2])
- +- Exchange(distribution=[any], shuffle_mode=[BATCH])
+ +- Exchange(distribution=[broadcast])
+- Reused(reference_id=[1])
]]>
</Resource>
@@ -888,7 +888,7 @@ LogicalProject(a=[$0], c=[$2], c0=[$9])
<Resource name="optimized exec plan">
<![CDATA[
Calc(select=[a, c, c0])
-+- NestedLoopJoin(joinType=[InnerJoin], where=[(a = a0)], select=[a, c, a0,
c0], build=[right])
++- HashJoin(joinType=[InnerJoin], where=[(a = a0)], select=[a, c, a0, c0],
isBroadcast=[true], build=[right])
:- Calc(select=[a, c])
: +- Limit(offset=[0], fetch=[11], global=[true])
: +- Exchange(distribution=[single])
diff --git
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/optimize/program/DynamicPartitionPruningProgramTest.xml
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/optimize/program/DynamicPartitionPruningProgramTest.xml
index c41077ca0dc..9e288d37bf5 100644
---
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/optimize/program/DynamicPartitionPruningProgramTest.xml
+++
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/optimize/program/DynamicPartitionPruningProgramTest.xml
@@ -716,7 +716,7 @@ LogicalProject(id=[$0], name=[$1], amount=[$2], price=[$3],
fact_date_sk=[$4], i
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-NestedLoopJoin(joinType=[InnerJoin], where=[=(fact_date_sk, dim_date_sk)],
select=[id, name, amount, price, fact_date_sk, id0, male, amount0, price0,
dim_date_sk], build=[right])
+HashJoin(joinType=[InnerJoin], where=[=(fact_date_sk, dim_date_sk)],
select=[id, name, amount, price, fact_date_sk, id0, male, amount0, price0,
dim_date_sk], isBroadcast=[true], build=[right])
:- DynamicFilteringTableSourceScan(table=[[testCatalog, test_database,
fact_part]], fields=[id, name, amount, price, fact_date_sk])
: +- DynamicFilteringDataCollector(fields=[dim_date_sk])
: +- Calc(select=[id, male, amount, price, dim_date_sk], where=[<(price,
500)])
@@ -742,7 +742,7 @@ LogicalProject(id=[$0], name=[$1], amount=[$2], price=[$3],
fact_date_sk=[$4], i
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-NestedLoopJoin(joinType=[InnerJoin], where=[=(fact_date_sk, dim_date_sk)],
select=[id, name, amount, price, fact_date_sk, id0, male, amount0, price0,
dim_date_sk], build=[right])
+HashJoin(joinType=[InnerJoin], where=[=(fact_date_sk, dim_date_sk)],
select=[id, name, amount, price, fact_date_sk, id0, male, amount0, price0,
dim_date_sk], isBroadcast=[true], build=[right])
:- Calc(select=[id, name, amount, price, fact_date_sk], where=[>(price, 200)])
: +- DynamicFilteringTableSourceScan(table=[[testCatalog, test_database,
fact_part, filter=[]]], fields=[id, name, amount, price, fact_date_sk])
: +- DynamicFilteringDataCollector(fields=[dim_date_sk])
@@ -824,7 +824,7 @@ LogicalProject(id=[$0], male=[$1], amount=[$2], price=[$3],
dim_date_sk=[$4], id
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-NestedLoopJoin(joinType=[InnerJoin], where=[=(fact_date_sk, dim_date_sk)],
select=[id, male, amount, price, dim_date_sk, id0, name, amount0, price0,
fact_date_sk], build=[left])
+HashJoin(joinType=[InnerJoin], where=[=(fact_date_sk, dim_date_sk)],
select=[id, male, amount, price, dim_date_sk, id0, name, amount0, price0,
fact_date_sk], isBroadcast=[true], build=[left])
:- Exchange(distribution=[broadcast])
: +- Calc(select=[id, male, amount, price, dim_date_sk], where=[<(price,
500)])
: +- TableSourceScan(table=[[testCatalog, test_database, dim, filter=[]]],
fields=[id, male, amount, price, dim_date_sk])
@@ -850,7 +850,7 @@ LogicalProject(id=[$0], male=[$1], amount=[$2], price=[$3],
dim_date_sk=[$4], id
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-NestedLoopJoin(joinType=[InnerJoin], where=[=(fact_date_sk, dim_date_sk)],
select=[id, male, amount, price, dim_date_sk, id0, name, amount0, price0,
fact_date_sk], build=[left])
+HashJoin(joinType=[InnerJoin], where=[=(fact_date_sk, dim_date_sk)],
select=[id, male, amount, price, dim_date_sk, id0, name, amount0, price0,
fact_date_sk], isBroadcast=[true], build=[left])
:- Exchange(distribution=[broadcast])
: +- Calc(select=[id, male, amount, price, dim_date_sk], where=[<(price,
500)])
: +- TableSourceScan(table=[[testCatalog, test_database, dim, filter=[]]],
fields=[id, male, amount, price, dim_date_sk])
@@ -933,7 +933,7 @@ LogicalProject(id=[$0], male=[$1], amount=[$2], price=[$3],
dim_date_sk=[$4], fa
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-NestedLoopJoin(joinType=[InnerJoin], where=[AND(=(fact_date_sk, dim_date_sk),
=(amount0, amount))], select=[id, male, amount, price, dim_date_sk,
fact_date_sk, amount0], build=[left])
+HashJoin(joinType=[InnerJoin], where=[AND(=(fact_date_sk, dim_date_sk),
=(amount0, amount))], select=[id, male, amount, price, dim_date_sk,
fact_date_sk, amount0], isBroadcast=[true], build=[left])
:- Exchange(distribution=[broadcast])
: +- Calc(select=[id, male, amount, price, dim_date_sk], where=[<(price,
500)])
: +- TableSourceScan(table=[[testCatalog, test_database, dim, filter=[]]],
fields=[id, male, amount, price, dim_date_sk])
diff --git
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/optimize/program/FlinkRuntimeFilterProgramTest.xml
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/optimize/program/FlinkRuntimeFilterProgramTest.xml
index 22749949dc4..148836f6870 100644
---
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/optimize/program/FlinkRuntimeFilterProgramTest.xml
+++
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/optimize/program/FlinkRuntimeFilterProgramTest.xml
@@ -757,15 +757,15 @@ HashJoin(joinType=[LeftOuterJoin], where=[(amount =
amount0)], select=[id, name,
</TestCase>
<TestCase name="testNestedLoopJoin">
<Resource name="sql">
- <![CDATA[select * from fact, dim where fact.amount = dim.amount and
dim.price < 500]]>
+ <![CDATA[SELECT /*+ NEST_LOOP(dim) */ * FROM fact, dim WHERE fact.amount
= dim.amount AND dim.price < 500]]>
</Resource>
<Resource name="ast">
<![CDATA[
LogicalProject(id=[$0], name=[$1], amount=[$2], price=[$3], fact_date_sk=[$4],
id0=[$5], male=[$6], amount0=[$7], price0=[$8], dim_date_sk=[$9])
+- LogicalFilter(condition=[AND(=($2, $7), <($8, 500))])
- +- LogicalJoin(condition=[true], joinType=[inner])
- :- LogicalTableScan(table=[[testCatalog, test_database, fact]])
- +- LogicalTableScan(table=[[testCatalog, test_database, dim]])
+ +- LogicalJoin(condition=[true], joinType=[inner], joinHints=[[[NEST_LOOP
inheritPath:[0, 0] options:[dim]]]])
+ :- LogicalTableScan(table=[[testCatalog, test_database, fact]],
hints=[[[ALIAS inheritPath:[] options:[fact]]]])
+ +- LogicalTableScan(table=[[testCatalog, test_database, dim]],
hints=[[[ALIAS inheritPath:[] options:[dim]]]])
]]>
</Resource>
<Resource name="optimized rel plan">
