From bc22380a21286a68566b742c6268669bab3ed8ce Mon Sep 17 00:00:00 2001
From: Sergey Nuyanzin <snuyanzin@gmail.com>
Date: Wed, 13 May 2026 19:38:16 +0200
Subject: [PATCH] [FLINK-39763][table] Allow inferring constants with `IS NOT
 DISTINCT FROM`

---
 .../java/org/apache/calcite/rex/RexUtil.java  |   9 +-
 .../batch/sql/DynamicFunctionPlanTest.xml     |   9 +-
 .../plan/batch/sql/RemoveShuffleTest.xml      |  54 ++++----
 .../batch/sql/join/BroadcastHashJoinTest.xml  |  85 ++++--------
 .../batch/sql/join/NestedLoopJoinTest.xml     |  49 +++----
 .../batch/sql/join/ShuffledHashJoinTest.xml   | 127 +++++++-----------
 .../plan/batch/sql/join/SortMergeJoinTest.xml | 127 +++++++-----------
 .../planner/plan/optimize/ScanReuseTest.xml   |  26 ++--
 .../planner/plan/stream/sql/join/JoinTest.xml |  86 ++++++------
 .../sql/join/BroadcastHashJoinTest.scala      |   7 +
 .../batch/sql/join/ShuffledHashJoinTest.scala |   7 +
 .../batch/sql/join/SortMergeJoinTest.scala    |   6 +
 .../runtime/batch/table/DecimalITCase.scala   |  20 ---
 13 files changed, 259 insertions(+), 353 deletions(-)

diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rex/RexUtil.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rex/RexUtil.java
index bddc6a15239ad..a629af209bc6e 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rex/RexUtil.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rex/RexUtil.java
@@ -79,10 +79,7 @@
  * because of current Calcite way of inferring constants from IS NOT DISTINCT FROM clashes with
  * filter push down.
  *
- * <p>Lines 402 ~ 404, Use Calcite 1.32.0 behavior for {@link RexUtil#gatherConstraints(Class,
- * RexNode, Map, Set, RexBuilder)}.
- *
- * <p>FLINK modifications (backport of CALCITE-6764): Line 2481~2485
+ * <p>FLINK modifications (backport of CALCITE-6764): Line 2489~2494
  */
 public class RexUtil {
 
@@ -401,9 +398,7 @@ private static <C extends RexNode> void gatherConstraints(
         final RexNode right;
         switch (predicate.getKind()) {
             case EQUALS:
-                // FLINK BEGIN MODIFICATION
-                // case IS_NOT_DISTINCT_FROM:
-                // FLINK END MODIFICATION
+            case IS_NOT_DISTINCT_FROM:
                 left = ((RexCall) predicate).getOperands().get(0);
                 right = ((RexCall) predicate).getOperands().get(1);
                 break;
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/DynamicFunctionPlanTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/DynamicFunctionPlanTest.xml
index c4442ef5dde99..041111d053893 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/DynamicFunctionPlanTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/DynamicFunctionPlanTest.xml
@@ -60,10 +60,11 @@ LogicalAggregate(group=[{0, 1}], EXPR$2=[SUM($2)], EXPR$3=[COUNT()])
     </Resource>
     <Resource name="optimized exec plan">
       <![CDATA[
-HashAggregate(isMerge=[false], groupBy=[cat, hh], select=[cat, hh, SUM(cnt) AS EXPR$2, COUNT(*) AS EXPR$3])
-+- Exchange(distribution=[hash[cat, hh]])
-   +- Calc(select=[cat, SUBSTR(CAST(LOCALTIME() AS VARCHAR(2147483647)), 1, 2) AS hh, cnt], where=[(SUBSTR(CAST(ts AS VARCHAR(2147483647)), 1, 2) = SUBSTR(CAST(LOCALTIME() AS VARCHAR(2147483647)), 1, 2))])
-      +- TableSourceScan(table=[[default_catalog, default_database, src, filter=[], project=[cat, cnt, ts], metadata=[]]], fields=[cat, cnt, ts])
+Calc(select=[cat, SUBSTR(CAST(LOCALTIME() AS VARCHAR(2147483647)), 1, 2) AS hh, EXPR$2, EXPR$3])
++- HashAggregate(isMerge=[false], groupBy=[cat], select=[cat, SUM(cnt) AS EXPR$2, COUNT(*) AS EXPR$3])
+   +- Exchange(distribution=[hash[cat]])
+      +- Calc(select=[cat, cnt], where=[(SUBSTR(CAST(ts AS VARCHAR(2147483647)), 1, 2) = SUBSTR(CAST(LOCALTIME() AS VARCHAR(2147483647)), 1, 2))])
+         +- TableSourceScan(table=[[default_catalog, default_database, src, filter=[], project=[cat, cnt, ts], metadata=[]]], fields=[cat, cnt, ts])
 ]]>
     </Resource>
   </TestCase>
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/RemoveShuffleTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/RemoveShuffleTest.xml
index be2b0e58f2905..f08d5635de70e 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/RemoveShuffleTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/RemoveShuffleTest.xml
@@ -490,19 +490,20 @@ LogicalProject(a=[$0], b=[$1], c=[$2], d=[$3], e=[$4], f=[$5], a0=[$6], b0=[$7],
     </Resource>
     <Resource name="optimized exec plan">
       <![CDATA[
-MultipleInput(readOrder=[0,1,0], members=[\nHashJoin(joinType=[InnerJoin], where=[(a = d0)], select=[a, b, c, d, e, f, a0, b0, c0, d0, e0, f0], build=[left])\n:- [#1] Exchange(distribution=[hash[a]])\n+- HashJoin(joinType=[RightOuterJoin], where=[(a = d)], select=[a, b, c, d, e, f], build=[right])\n   :- [#2] Exchange(distribution=[hash[a]])\n   +- [#3] Exchange(distribution=[hash[d]])\n])
-:- Exchange(distribution=[hash[a]])
-:  +- HashJoin(joinType=[RightOuterJoin], where=[(a = d)], select=[a, b, c, d, e, f], build=[right])
-:     :- Exchange(distribution=[hash[a]])
-:     :  +- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
-:     +- Exchange(distribution=[hash[d]])
-:        +- Calc(select=[d, CAST(2 AS BIGINT) AS e, f], where=[(e = 2)])
-:           +- LegacyTableSourceScan(table=[[default_catalog, default_database, y, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])
-:- Exchange(distribution=[hash[a]])
-:  +- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
-+- Exchange(distribution=[hash[d]])
-   +- Calc(select=[d, CAST(2 AS BIGINT) AS e, f], where=[(e = 2)])
-      +- LegacyTableSourceScan(table=[[default_catalog, default_database, y, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])
+Calc(select=[a, b, c, d, CAST(2 AS BIGINT) AS e, f, a0, b0, c0, d0, CAST(2 AS BIGINT) AS e0, f0])
++- MultipleInput(readOrder=[0,1,0], members=[\nHashJoin(joinType=[InnerJoin], where=[(a = d0)], select=[a, b, c, d, f, a0, b0, c0, d0, f0], build=[left])\n:- [#1] Exchange(distribution=[hash[a]])\n+- HashJoin(joinType=[RightOuterJoin], where=[(a = d)], select=[a, b, c, d, f], build=[right])\n   :- [#2] Exchange(distribution=[hash[a]])\n   +- [#3] Exchange(distribution=[hash[d]])\n])
+   :- Exchange(distribution=[hash[a]])
+   :  +- HashJoin(joinType=[RightOuterJoin], where=[(a = d)], select=[a, b, c, d, f], build=[right])
+   :     :- Exchange(distribution=[hash[a]])
+   :     :  +- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+   :     +- Exchange(distribution=[hash[d]])
+   :        +- Calc(select=[d, f], where=[(e = 2)])
+   :           +- LegacyTableSourceScan(table=[[default_catalog, default_database, y, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])
+   :- Exchange(distribution=[hash[a]])
+   :  +- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+   +- Exchange(distribution=[hash[d]])
+      +- Calc(select=[d, f], where=[(e = 2)])
+         +- LegacyTableSourceScan(table=[[default_catalog, default_database, y, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])
 ]]>
     </Resource>
   </TestCase>
@@ -1158,19 +1159,20 @@ LogicalProject(a=[$0], b=[$1], c=[$2], d=[$3], e=[$4], f=[$5], a0=[$6], b0=[$7],
     </Resource>
     <Resource name="optimized exec plan">
       <![CDATA[
-MultipleInput(members=[\nSortMergeJoin(joinType=[InnerJoin], where=[(a = d0)], select=[a, b, c, d, e, f, a0, b0, c0, d0, e0, f0])\n:- [#1] Exchange(distribution=[hash[a]])\n+- SortMergeJoin(joinType=[RightOuterJoin], where=[(a = d)], select=[a, b, c, d, e, f])\n   :- [#2] Exchange(distribution=[hash[a]])\n   +- [#3] Exchange(distribution=[hash[d]])\n])
-:- Exchange(distribution=[hash[a]])
-:  +- SortMergeJoin(joinType=[RightOuterJoin], where=[(a = d)], select=[a, b, c, d, e, f])
-:     :- Exchange(distribution=[hash[a]])
-:     :  +- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
-:     +- Exchange(distribution=[hash[d]])
-:        +- Calc(select=[d, CAST(2 AS BIGINT) AS e, f], where=[(e = 2)])
-:           +- LegacyTableSourceScan(table=[[default_catalog, default_database, y, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])
-:- Exchange(distribution=[hash[a]])
-:  +- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
-+- Exchange(distribution=[hash[d]])
-   +- Calc(select=[d, CAST(2 AS BIGINT) AS e, f], where=[(e = 2)])
-      +- LegacyTableSourceScan(table=[[default_catalog, default_database, y, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])
+Calc(select=[a, b, c, d, CAST(2 AS BIGINT) AS e, f, a0, b0, c0, d0, CAST(2 AS BIGINT) AS e0, f0])
++- MultipleInput(members=[\nSortMergeJoin(joinType=[InnerJoin], where=[(a = d0)], select=[a, b, c, d, f, a0, b0, c0, d0, f0])\n:- [#1] Exchange(distribution=[hash[a]])\n+- SortMergeJoin(joinType=[RightOuterJoin], where=[(a = d)], select=[a, b, c, d, f])\n   :- [#2] Exchange(distribution=[hash[a]])\n   +- [#3] Exchange(distribution=[hash[d]])\n])
+   :- Exchange(distribution=[hash[a]])
+   :  +- SortMergeJoin(joinType=[RightOuterJoin], where=[(a = d)], select=[a, b, c, d, f])
+   :     :- Exchange(distribution=[hash[a]])
+   :     :  +- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+   :     +- Exchange(distribution=[hash[d]])
+   :        +- Calc(select=[d, f], where=[(e = 2)])
+   :           +- LegacyTableSourceScan(table=[[default_catalog, default_database, y, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])
+   :- Exchange(distribution=[hash[a]])
+   :  +- LegacyTableSourceScan(table=[[default_catalog, default_database, x, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+   +- Exchange(distribution=[hash[d]])
+      +- Calc(select=[d, f], where=[(e = 2)])
+         +- LegacyTableSourceScan(table=[[default_catalog, default_database, y, source: [TestTableSource(d, e, f)]]], fields=[d, e, f])
 ]]>
     </Resource>
   </TestCase>
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/BroadcastHashJoinTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/BroadcastHashJoinTest.xml
index 897e52cf7f783..dc0d9c3e721b3 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/BroadcastHashJoinTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/BroadcastHashJoinTest.xml
@@ -138,14 +138,15 @@ LogicalProject(a=[$0], b=[$1], d=[$2], e=[$3])
     </Resource>
     <Resource name="optimized exec plan">
       <![CDATA[
-Calc(select=[a, b, CAST(2 AS INTEGER) AS d, e])
-+- MultipleInput(readOrder=[0,1], members=[\nHashJoin(joinType=[InnerJoin], where=[((a = d) AND (b = e))], select=[a, b, d, e], isBroadcast=[true], build=[right])\n:- HashAggregate(isMerge=[true], groupBy=[a], select=[a, Final_COUNT(count$0) AS b])\n:  +- [#2] Exchange(distribution=[hash[a]])\n+- [#1] Exchange(distribution=[broadcast])\n])
+Calc(select=[CAST(2 AS INTEGER) AS a, b, CAST(2 AS INTEGER) AS d, e])
++- MultipleInput(readOrder=[0,1], members=[\nHashJoin(joinType=[InnerJoin], where=[(b = e)], select=[b, e], isBroadcast=[true], build=[right])\n:- Calc(select=[b])\n:  +- HashAggregate(isMerge=[true], groupBy=[a], select=[a, Final_COUNT(count$0) AS b])\n:     +- [#2] Exchange(distribution=[hash[a]])\n+- [#1] Exchange(distribution=[broadcast])\n])
    :- Exchange(distribution=[broadcast])
-   :  +- HashAggregate(isMerge=[true], groupBy=[d], select=[d, Final_COUNT(count$0) AS e])
-   :     +- Exchange(distribution=[hash[d]])
-   :        +- LocalHashAggregate(groupBy=[d], select=[d, Partial_COUNT(e) AS count$0])
-   :           +- Calc(select=[CAST(2 AS INTEGER) AS d, e], where=[(d = 2)])
-   :              +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[d, e, f, g, h])
+   :  +- Calc(select=[e])
+   :     +- HashAggregate(isMerge=[true], groupBy=[d], select=[d, Final_COUNT(count$0) AS e])
+   :        +- Exchange(distribution=[hash[d]])
+   :           +- LocalHashAggregate(groupBy=[d], select=[d, Partial_COUNT(e) AS count$0])
+   :              +- Calc(select=[CAST(2 AS INTEGER) AS d, e], where=[(d = 2)])
+   :                 +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[d, e, f, g, h])
    +- Exchange(distribution=[hash[a]])
       +- LocalHashAggregate(groupBy=[a], select=[a, Partial_COUNT(b) AS count$0])
          +- Calc(select=[CAST(2 AS INTEGER) AS a, b], where=[(a = 2)])
@@ -174,46 +175,6 @@ Calc(select=[c, g])
    +- Exchange(distribution=[broadcast])
       +- Calc(select=[d, e, g])
          +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[d, e, f, g, h])
-]]>
-    </Resource>
-  </TestCase>
-  <TestCase name="testInnerJoinWithJoinConditionPushDown">
-    <Resource name="sql">
-      <![CDATA[
-SELECT * FROM
-   (select a, count(b) as b from MyTable1 group by a)
-   join
-   (select d, count(e) as e from MyTable2 group by d)
-   on a = d and b = e and d = 2 and b = 1
-]]>
-    </Resource>
-    <Resource name="ast">
-      <![CDATA[
-LogicalProject(a=[$0], b=[$1], d=[$2], e=[$3])
-+- LogicalJoin(condition=[AND(=($0, $2), =($1, $3), =($2, 2), =($1, 1))], joinType=[inner])
-   :- LogicalAggregate(group=[{0}], b=[COUNT($1)])
-   :  +- LogicalProject(a=[$0], b=[$1])
-   :     +- LogicalTableScan(table=[[default_catalog, default_database, MyTable1]])
-   +- LogicalAggregate(group=[{0}], e=[COUNT($1)])
-      +- LogicalProject(d=[$0], e=[$1])
-         +- LogicalTableScan(table=[[default_catalog, default_database, MyTable2]])
-]]>
-    </Resource>
-    <Resource name="optimized exec plan">
-      <![CDATA[
-Calc(select=[a, 1 AS b, CAST(2 AS INTEGER) AS d, 1 AS e])
-+- MultipleInput(readOrder=[0,1], members=[\nHashJoin(joinType=[InnerJoin], where=[(a = d)], select=[a, d], isBroadcast=[true], build=[right])\n:- Calc(select=[a], where=[(b = 1)])\n:  +- HashAggregate(isMerge=[true], groupBy=[a], select=[a, Final_COUNT(count$0) AS b])\n:     +- [#2] Exchange(distribution=[hash[a]])\n+- [#1] Exchange(distribution=[broadcast])\n])
-   :- Exchange(distribution=[broadcast])
-   :  +- Calc(select=[d], where=[(e = 1)])
-   :     +- HashAggregate(isMerge=[true], groupBy=[d], select=[d, Final_COUNT(count$0) AS e])
-   :        +- Exchange(distribution=[hash[d]])
-   :           +- LocalHashAggregate(groupBy=[d], select=[d, Partial_COUNT(e) AS count$0])
-   :              +- Calc(select=[CAST(2 AS INTEGER) AS d, e], where=[(d = 2)])
-   :                 +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[d, e, f, g, h])
-   +- Exchange(distribution=[hash[a]])
-      +- LocalHashAggregate(groupBy=[a], select=[a, Partial_COUNT(b) AS count$0])
-         +- Calc(select=[CAST(2 AS INTEGER) AS a, b], where=[(a = 2)])
-            +- TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[a, b, c])
 ]]>
     </Resource>
   </TestCase>
@@ -311,14 +272,15 @@ LogicalProject(a=[$0], b=[$1], d=[$2], e=[$3])
     </Resource>
     <Resource name="optimized exec plan">
       <![CDATA[
-Calc(select=[CAST(2 AS INTEGER) AS a, b, d, CAST(e AS BIGINT) AS e])
-+- MultipleInput(readOrder=[0,1], members=[\nHashJoin(joinType=[InnerJoin], where=[((a = d) AND (b = e))], select=[a, b, d, e], isBroadcast=[true], build=[right])\n:- HashAggregate(isMerge=[true], groupBy=[a], select=[a, Final_COUNT(count$0) AS b])\n:  +- [#2] Exchange(distribution=[hash[a]])\n+- [#1] Exchange(distribution=[broadcast])\n])
+Calc(select=[CAST(2 AS INTEGER) AS a, b, CAST(2 AS INTEGER) AS d, CAST(e AS BIGINT) AS e])
++- MultipleInput(readOrder=[0,1], members=[\nHashJoin(joinType=[InnerJoin], where=[(b = e)], select=[b, e], isBroadcast=[true], build=[right])\n:- Calc(select=[b])\n:  +- HashAggregate(isMerge=[true], groupBy=[a], select=[a, Final_COUNT(count$0) AS b])\n:     +- [#2] Exchange(distribution=[hash[a]])\n+- [#1] Exchange(distribution=[broadcast])\n])
    :- Exchange(distribution=[broadcast])
-   :  +- HashAggregate(isMerge=[true], groupBy=[d], select=[d, Final_COUNT(count$0) AS e])
-   :     +- Exchange(distribution=[hash[d]])
-   :        +- LocalHashAggregate(groupBy=[d], select=[d, Partial_COUNT(e) AS count$0])
-   :           +- Calc(select=[CAST(2 AS INTEGER) AS d, e], where=[(d = 2)])
-   :              +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[d, e, f, g, h])
+   :  +- Calc(select=[e])
+   :     +- HashAggregate(isMerge=[true], groupBy=[d], select=[d, Final_COUNT(count$0) AS e])
+   :        +- Exchange(distribution=[hash[d]])
+   :           +- LocalHashAggregate(groupBy=[d], select=[d, Partial_COUNT(e) AS count$0])
+   :              +- Calc(select=[CAST(2 AS INTEGER) AS d, e], where=[(d = 2)])
+   :                 +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[d, e, f, g, h])
    +- Exchange(distribution=[hash[a]])
       +- LocalHashAggregate(groupBy=[a], select=[a, Partial_COUNT(b) AS count$0])
          +- Calc(select=[CAST(2 AS INTEGER) AS a, b], where=[(a = 2)])
@@ -547,14 +509,15 @@ LogicalProject(a=[$0], b=[$1], d=[$2], e=[$3])
     </Resource>
     <Resource name="optimized exec plan">
       <![CDATA[
-Calc(select=[a, CAST(b AS BIGINT) AS b, CAST(2 AS INTEGER) AS d, e])
-+- MultipleInput(readOrder=[0,1], members=[\nHashJoin(joinType=[InnerJoin], where=[((a = d) AND (b = e))], select=[a, b, d, e], isBroadcast=[true], build=[right])\n:- HashAggregate(isMerge=[true], groupBy=[a], select=[a, Final_COUNT(count$0) AS b])\n:  +- [#2] Exchange(distribution=[hash[a]])\n+- [#1] Exchange(distribution=[broadcast])\n])
+Calc(select=[CAST(2 AS INTEGER) AS a, CAST(b AS BIGINT) AS b, CAST(2 AS INTEGER) AS d, e])
++- MultipleInput(readOrder=[0,1], members=[\nHashJoin(joinType=[InnerJoin], where=[(b = e)], select=[b, e], isBroadcast=[true], build=[right])\n:- Calc(select=[b])\n:  +- HashAggregate(isMerge=[true], groupBy=[a], select=[a, Final_COUNT(count$0) AS b])\n:     +- [#2] Exchange(distribution=[hash[a]])\n+- [#1] Exchange(distribution=[broadcast])\n])
    :- Exchange(distribution=[broadcast])
-   :  +- HashAggregate(isMerge=[true], groupBy=[d], select=[d, Final_COUNT(count$0) AS e])
-   :     +- Exchange(distribution=[hash[d]])
-   :        +- LocalHashAggregate(groupBy=[d], select=[d, Partial_COUNT(e) AS count$0])
-   :           +- Calc(select=[CAST(2 AS INTEGER) AS d, e], where=[(d = 2)])
-   :              +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[d, e, f, g, h])
+   :  +- Calc(select=[e])
+   :     +- HashAggregate(isMerge=[true], groupBy=[d], select=[d, Final_COUNT(count$0) AS e])
+   :        +- Exchange(distribution=[hash[d]])
+   :           +- LocalHashAggregate(groupBy=[d], select=[d, Partial_COUNT(e) AS count$0])
+   :              +- Calc(select=[CAST(2 AS INTEGER) AS d, e], where=[(d = 2)])
+   :                 +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[d, e, f, g, h])
    +- Exchange(distribution=[hash[a]])
       +- LocalHashAggregate(groupBy=[a], select=[a, Partial_COUNT(b) AS count$0])
          +- Calc(select=[CAST(2 AS INTEGER) AS a, b], where=[(a = 2)])
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/NestedLoopJoinTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/NestedLoopJoinTest.xml
index c37f2d9eb04f6..4f6c8e1e9a445 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/NestedLoopJoinTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/NestedLoopJoinTest.xml
@@ -324,14 +324,15 @@ LogicalProject(a=[$0], b=[$1], d=[$2], e=[$3])
     </Resource>
     <Resource name="optimized exec plan">
       <![CDATA[
-Calc(select=[a, b, CAST(2 AS INTEGER) AS d, e])
-+- MultipleInput(readOrder=[0,1], members=[\nNestedLoopJoin(joinType=[InnerJoin], where=[((a = d) AND (b = e))], select=[a, b, d, e], build=[left])\n:- [#1] Exchange(distribution=[broadcast])\n+- HashAggregate(isMerge=[true], groupBy=[d], select=[d, Final_COUNT(count$0) AS e])\n   +- [#2] Exchange(distribution=[hash[d]])\n])
+Calc(select=[CAST(2 AS INTEGER) AS a, b, CAST(2 AS INTEGER) AS d, e])
++- MultipleInput(readOrder=[0,1], members=[\nNestedLoopJoin(joinType=[InnerJoin], where=[(b = e)], select=[b, e], build=[left])\n:- [#1] Exchange(distribution=[broadcast])\n+- Calc(select=[e])\n   +- HashAggregate(isMerge=[true], groupBy=[d], select=[d, Final_COUNT(count$0) AS e])\n      +- [#2] Exchange(distribution=[hash[d]])\n])
    :- Exchange(distribution=[broadcast])
-   :  +- HashAggregate(isMerge=[true], groupBy=[a], select=[a, Final_COUNT(count$0) AS b])
-   :     +- Exchange(distribution=[hash[a]])
-   :        +- LocalHashAggregate(groupBy=[a], select=[a, Partial_COUNT(b) AS count$0])
-   :           +- Calc(select=[CAST(2 AS INTEGER) AS a, b], where=[(a = 2)])
-   :              +- TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[a, b, c])
+   :  +- Calc(select=[b])
+   :     +- HashAggregate(isMerge=[true], groupBy=[a], select=[a, Final_COUNT(count$0) AS b])
+   :        +- Exchange(distribution=[hash[a]])
+   :           +- LocalHashAggregate(groupBy=[a], select=[a, Partial_COUNT(b) AS count$0])
+   :              +- Calc(select=[CAST(2 AS INTEGER) AS a, b], where=[(a = 2)])
+   :                 +- TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[a, b, c])
    +- Exchange(distribution=[hash[d]])
       +- LocalHashAggregate(groupBy=[d], select=[d, Partial_COUNT(e) AS count$0])
          +- Calc(select=[CAST(2 AS INTEGER) AS d, e], where=[(d = 2)])
@@ -387,8 +388,8 @@ LogicalProject(a=[$0], b=[$1], d=[$2], e=[$3])
     </Resource>
     <Resource name="optimized exec plan">
       <![CDATA[
-Calc(select=[a, 1 AS b, CAST(2 AS INTEGER) AS d, 1 AS e])
-+- MultipleInput(readOrder=[0,1], members=[\nNestedLoopJoin(joinType=[InnerJoin], where=[(a = d)], select=[a, d], build=[left])\n:- [#1] Exchange(distribution=[broadcast])\n+- Calc(select=[d], where=[(e = 1)])\n   +- HashAggregate(isMerge=[true], groupBy=[d], select=[d, Final_COUNT(count$0) AS e])\n      +- [#2] Exchange(distribution=[hash[d]])\n])
+Calc(select=[CAST(2 AS INTEGER) AS a, 1 AS b, CAST(2 AS INTEGER) AS d, 1 AS e])
++- MultipleInput(readOrder=[0,1], members=[\nNestedLoopJoin(joinType=[InnerJoin], where=[true], select=[a, d], build=[left])\n:- [#1] Exchange(distribution=[broadcast])\n+- Calc(select=[d], where=[(e = 1)])\n   +- HashAggregate(isMerge=[true], groupBy=[d], select=[d, Final_COUNT(count$0) AS e])\n      +- [#2] Exchange(distribution=[hash[d]])\n])
    :- Exchange(distribution=[broadcast])
    :  +- Calc(select=[a], where=[(b = 1)])
    :     +- HashAggregate(isMerge=[true], groupBy=[a], select=[a, Final_COUNT(count$0) AS b])
@@ -520,14 +521,15 @@ LogicalProject(a=[$0], b=[$1], d=[$2], e=[$3])
     </Resource>
     <Resource name="optimized exec plan">
       <![CDATA[
-Calc(select=[CAST(2 AS INTEGER) AS a, b, d, CAST(e AS BIGINT) AS e])
-+- MultipleInput(readOrder=[0,1], members=[\nNestedLoopJoin(joinType=[InnerJoin], where=[((a = d) AND (b = e))], select=[a, b, d, e], build=[left])\n:- [#1] Exchange(distribution=[broadcast])\n+- HashAggregate(isMerge=[true], groupBy=[d], select=[d, Final_COUNT(count$0) AS e])\n   +- [#2] Exchange(distribution=[hash[d]])\n])
+Calc(select=[CAST(2 AS INTEGER) AS a, b, CAST(2 AS INTEGER) AS d, CAST(e AS BIGINT) AS e])
++- MultipleInput(readOrder=[0,1], members=[\nNestedLoopJoin(joinType=[InnerJoin], where=[(b = e)], select=[b, e], build=[left])\n:- [#1] Exchange(distribution=[broadcast])\n+- Calc(select=[e])\n   +- HashAggregate(isMerge=[true], groupBy=[d], select=[d, Final_COUNT(count$0) AS e])\n      +- [#2] Exchange(distribution=[hash[d]])\n])
    :- Exchange(distribution=[broadcast])
-   :  +- HashAggregate(isMerge=[true], groupBy=[a], select=[a, Final_COUNT(count$0) AS b])
-   :     +- Exchange(distribution=[hash[a]])
-   :        +- LocalHashAggregate(groupBy=[a], select=[a, Partial_COUNT(b) AS count$0])
-   :           +- Calc(select=[CAST(2 AS INTEGER) AS a, b], where=[(a = 2)])
-   :              +- TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[a, b, c])
+   :  +- Calc(select=[b])
+   :     +- HashAggregate(isMerge=[true], groupBy=[a], select=[a, Final_COUNT(count$0) AS b])
+   :        +- Exchange(distribution=[hash[a]])
+   :           +- LocalHashAggregate(groupBy=[a], select=[a, Partial_COUNT(b) AS count$0])
+   :              +- Calc(select=[CAST(2 AS INTEGER) AS a, b], where=[(a = 2)])
+   :                 +- TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[a, b, c])
    +- Exchange(distribution=[hash[d]])
       +- LocalHashAggregate(groupBy=[d], select=[d, Partial_COUNT(e) AS count$0])
          +- Calc(select=[CAST(2 AS INTEGER) AS d, e], where=[(d = 2)])
@@ -823,14 +825,15 @@ LogicalProject(a=[$0], b=[$1], d=[$2], e=[$3])
     </Resource>
     <Resource name="optimized exec plan">
       <![CDATA[
-Calc(select=[a, CAST(b AS BIGINT) AS b, CAST(2 AS INTEGER) AS d, e])
-+- MultipleInput(readOrder=[0,1], members=[\nNestedLoopJoin(joinType=[InnerJoin], where=[((a = d) AND (b = e))], select=[a, b, d, e], build=[left])\n:- [#1] Exchange(distribution=[broadcast])\n+- HashAggregate(isMerge=[true], groupBy=[d], select=[d, Final_COUNT(count$0) AS e])\n   +- [#2] Exchange(distribution=[hash[d]])\n])
+Calc(select=[CAST(2 AS INTEGER) AS a, CAST(b AS BIGINT) AS b, CAST(2 AS INTEGER) AS d, e])
++- MultipleInput(readOrder=[0,1], members=[\nNestedLoopJoin(joinType=[InnerJoin], where=[(b = e)], select=[b, e], build=[left])\n:- [#1] Exchange(distribution=[broadcast])\n+- Calc(select=[e])\n   +- HashAggregate(isMerge=[true], groupBy=[d], select=[d, Final_COUNT(count$0) AS e])\n      +- [#2] Exchange(distribution=[hash[d]])\n])
    :- Exchange(distribution=[broadcast])
-   :  +- HashAggregate(isMerge=[true], groupBy=[a], select=[a, Final_COUNT(count$0) AS b])
-   :     +- Exchange(distribution=[hash[a]])
-   :        +- LocalHashAggregate(groupBy=[a], select=[a, Partial_COUNT(b) AS count$0])
-   :           +- Calc(select=[CAST(2 AS INTEGER) AS a, b], where=[(a = 2)])
-   :              +- TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[a, b, c])
+   :  +- Calc(select=[b])
+   :     +- HashAggregate(isMerge=[true], groupBy=[a], select=[a, Final_COUNT(count$0) AS b])
+   :        +- Exchange(distribution=[hash[a]])
+   :           +- LocalHashAggregate(groupBy=[a], select=[a, Partial_COUNT(b) AS count$0])
+   :              +- Calc(select=[CAST(2 AS INTEGER) AS a, b], where=[(a = 2)])
+   :                 +- TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[a, b, c])
    +- Exchange(distribution=[hash[d]])
       +- LocalHashAggregate(groupBy=[d], select=[d, Partial_COUNT(e) AS count$0])
          +- Calc(select=[CAST(2 AS INTEGER) AS d, e], where=[(d = 2)])
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/ShuffledHashJoinTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/ShuffledHashJoinTest.xml
index e3d85fe3193b3..1a1470c714680 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/ShuffledHashJoinTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/ShuffledHashJoinTest.xml
@@ -241,20 +241,22 @@ LogicalProject(a=[$0], b=[$1], d=[$2], e=[$3])
     </Resource>
     <Resource name="optimized exec plan">
       <![CDATA[
-Calc(select=[a, b, CAST(2 AS INTEGER) AS d, e])
-+- HashJoin(joinType=[InnerJoin], where=[((a = d) AND (b = e))], select=[a, b, d, e], build=[left])
-   :- Exchange(distribution=[hash[a, b]])
-   :  +- HashAggregate(isMerge=[true], groupBy=[a], select=[a, Final_COUNT(count$0) AS b])
-   :     +- Exchange(distribution=[hash[a]])
-   :        +- LocalHashAggregate(groupBy=[a], select=[a, Partial_COUNT(b) AS count$0])
-   :           +- Calc(select=[CAST(2 AS INTEGER) AS a, b], where=[(a = 2)])
-   :              +- TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[a, b, c])
-   +- Exchange(distribution=[hash[d, e]])
-      +- HashAggregate(isMerge=[true], groupBy=[d], select=[d, Final_COUNT(count$0) AS e])
-         +- Exchange(distribution=[hash[d]])
-            +- LocalHashAggregate(groupBy=[d], select=[d, Partial_COUNT(e) AS count$0])
-               +- Calc(select=[CAST(2 AS INTEGER) AS d, e], where=[(d = 2)])
-                  +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[d, e, f, g, h])
+Calc(select=[CAST(2 AS INTEGER) AS a, b, CAST(2 AS INTEGER) AS d, e])
++- HashJoin(joinType=[InnerJoin], where=[(b = e)], select=[b, e], build=[left])
+   :- Exchange(distribution=[hash[b]])
+   :  +- Calc(select=[b])
+   :     +- HashAggregate(isMerge=[true], groupBy=[a], select=[a, Final_COUNT(count$0) AS b])
+   :        +- Exchange(distribution=[hash[a]])
+   :           +- LocalHashAggregate(groupBy=[a], select=[a, Partial_COUNT(b) AS count$0])
+   :              +- Calc(select=[CAST(2 AS INTEGER) AS a, b], where=[(a = 2)])
+   :                 +- TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[a, b, c])
+   +- Exchange(distribution=[hash[e]])
+      +- Calc(select=[e])
+         +- HashAggregate(isMerge=[true], groupBy=[d], select=[d, Final_COUNT(count$0) AS e])
+            +- Exchange(distribution=[hash[d]])
+               +- LocalHashAggregate(groupBy=[d], select=[d, Partial_COUNT(e) AS count$0])
+                  +- Calc(select=[CAST(2 AS INTEGER) AS d, e], where=[(d = 2)])
+                     +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[d, e, f, g, h])
 ]]>
     </Resource>
   </TestCase>
@@ -280,43 +282,6 @@ Calc(select=[c, g])
    +- Exchange(distribution=[hash[e, d]])
       +- Calc(select=[d, e, g])
          +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[d, e, f, g, h])
-]]>
-    </Resource>
-  </TestCase>
-  <TestCase name="testInnerJoinWithJoinConditionPushDown">
-    <Resource name="sql">
-      <![CDATA[
-SELECT * FROM
-   (select a, count(b) as b from MyTable1 group by a)
-   join
-   (select d, count(e) as e from MyTable2 group by d)
-   on a = d and b = e and d = 2 and b = 1
-]]>
-    </Resource>
-    <Resource name="ast">
-      <![CDATA[
-LogicalProject(a=[$0], b=[$1], d=[$2], e=[$3])
-+- LogicalJoin(condition=[AND(=($0, $2), =($1, $3), =($2, 2), =($1, 1))], joinType=[inner])
-   :- LogicalAggregate(group=[{0}], b=[COUNT($1)])
-   :  +- LogicalProject(a=[$0], b=[$1])
-   :     +- LogicalTableScan(table=[[default_catalog, default_database, MyTable1]])
-   +- LogicalAggregate(group=[{0}], e=[COUNT($1)])
-      +- LogicalProject(d=[$0], e=[$1])
-         +- LogicalTableScan(table=[[default_catalog, default_database, MyTable2]])
-]]>
-    </Resource>
-    <Resource name="optimized exec plan">
-      <![CDATA[
-Calc(select=[a, 1 AS b, CAST(2 AS INTEGER) AS d, 1 AS e])
-+- MultipleInput(readOrder=[0,1], members=[\nHashJoin(joinType=[InnerJoin], where=[(a = d)], select=[a, d], build=[left])\n:- Calc(select=[a], where=[(b = 1)])\n:  +- HashAggregate(isMerge=[true], groupBy=[a], select=[a, Final_COUNT(count$0) AS b])\n:     +- [#1] Exchange(distribution=[hash[a]])\n+- Calc(select=[d], where=[(e = 1)])\n   +- HashAggregate(isMerge=[true], groupBy=[d], select=[d, Final_COUNT(count$0) AS e])\n      +- [#2] Exchange(distribution=[hash[d]])\n])
-   :- Exchange(distribution=[hash[a]])
-   :  +- LocalHashAggregate(groupBy=[a], select=[a, Partial_COUNT(b) AS count$0])
-   :     +- Calc(select=[CAST(2 AS INTEGER) AS a, b], where=[(a = 2)])
-   :        +- TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[a, b, c])
-   +- Exchange(distribution=[hash[d]])
-      +- LocalHashAggregate(groupBy=[d], select=[d, Partial_COUNT(e) AS count$0])
-         +- Calc(select=[CAST(2 AS INTEGER) AS d, e], where=[(d = 2)])
-            +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[d, e, f, g, h])
 ]]>
     </Resource>
   </TestCase>
@@ -416,20 +381,22 @@ LogicalProject(a=[$0], b=[$1], d=[$2], e=[$3])
     </Resource>
     <Resource name="optimized exec plan">
       <![CDATA[
-Calc(select=[CAST(2 AS INTEGER) AS a, b, d, CAST(e AS BIGINT) AS e])
-+- HashJoin(joinType=[InnerJoin], where=[((a = d) AND (b = e))], select=[a, b, d, e], build=[left])
-   :- Exchange(distribution=[hash[a, b]])
-   :  +- HashAggregate(isMerge=[true], groupBy=[a], select=[a, Final_COUNT(count$0) AS b])
-   :     +- Exchange(distribution=[hash[a]])
-   :        +- LocalHashAggregate(groupBy=[a], select=[a, Partial_COUNT(b) AS count$0])
-   :           +- Calc(select=[CAST(2 AS INTEGER) AS a, b], where=[(a = 2)])
-   :              +- TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[a, b, c])
-   +- Exchange(distribution=[hash[d, e]])
-      +- HashAggregate(isMerge=[true], groupBy=[d], select=[d, Final_COUNT(count$0) AS e])
-         +- Exchange(distribution=[hash[d]])
-            +- LocalHashAggregate(groupBy=[d], select=[d, Partial_COUNT(e) AS count$0])
-               +- Calc(select=[CAST(2 AS INTEGER) AS d, e], where=[(d = 2)])
-                  +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[d, e, f, g, h])
+Calc(select=[CAST(2 AS INTEGER) AS a, b, CAST(2 AS INTEGER) AS d, CAST(e AS BIGINT) AS e])
++- HashJoin(joinType=[InnerJoin], where=[(b = e)], select=[b, e], build=[left])
+   :- Exchange(distribution=[hash[b]])
+   :  +- Calc(select=[b])
+   :     +- HashAggregate(isMerge=[true], groupBy=[a], select=[a, Final_COUNT(count$0) AS b])
+   :        +- Exchange(distribution=[hash[a]])
+   :           +- LocalHashAggregate(groupBy=[a], select=[a, Partial_COUNT(b) AS count$0])
+   :              +- Calc(select=[CAST(2 AS INTEGER) AS a, b], where=[(a = 2)])
+   :                 +- TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[a, b, c])
+   +- Exchange(distribution=[hash[e]])
+      +- Calc(select=[e])
+         +- HashAggregate(isMerge=[true], groupBy=[d], select=[d, Final_COUNT(count$0) AS e])
+            +- Exchange(distribution=[hash[d]])
+               +- LocalHashAggregate(groupBy=[d], select=[d, Partial_COUNT(e) AS count$0])
+                  +- Calc(select=[CAST(2 AS INTEGER) AS d, e], where=[(d = 2)])
+                     +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[d, e, f, g, h])
 ]]>
     </Resource>
   </TestCase>
@@ -656,20 +623,22 @@ LogicalProject(a=[$0], b=[$1], d=[$2], e=[$3])
     </Resource>
     <Resource name="optimized exec plan">
       <![CDATA[
-Calc(select=[a, CAST(b AS BIGINT) AS b, CAST(2 AS INTEGER) AS d, e])
-+- HashJoin(joinType=[InnerJoin], where=[((a = d) AND (b = e))], select=[a, b, d, e], build=[left])
-   :- Exchange(distribution=[hash[a, b]])
-   :  +- HashAggregate(isMerge=[true], groupBy=[a], select=[a, Final_COUNT(count$0) AS b])
-   :     +- Exchange(distribution=[hash[a]])
-   :        +- LocalHashAggregate(groupBy=[a], select=[a, Partial_COUNT(b) AS count$0])
-   :           +- Calc(select=[CAST(2 AS INTEGER) AS a, b], where=[(a = 2)])
-   :              +- TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[a, b, c])
-   +- Exchange(distribution=[hash[d, e]])
-      +- HashAggregate(isMerge=[true], groupBy=[d], select=[d, Final_COUNT(count$0) AS e])
-         +- Exchange(distribution=[hash[d]])
-            +- LocalHashAggregate(groupBy=[d], select=[d, Partial_COUNT(e) AS count$0])
-               +- Calc(select=[CAST(2 AS INTEGER) AS d, e], where=[(d = 2)])
-                  +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[d, e, f, g, h])
+Calc(select=[CAST(2 AS INTEGER) AS a, CAST(b AS BIGINT) AS b, CAST(2 AS INTEGER) AS d, e])
++- HashJoin(joinType=[InnerJoin], where=[(b = e)], select=[b, e], build=[left])
+   :- Exchange(distribution=[hash[b]])
+   :  +- Calc(select=[b])
+   :     +- HashAggregate(isMerge=[true], groupBy=[a], select=[a, Final_COUNT(count$0) AS b])
+   :        +- Exchange(distribution=[hash[a]])
+   :           +- LocalHashAggregate(groupBy=[a], select=[a, Partial_COUNT(b) AS count$0])
+   :              +- Calc(select=[CAST(2 AS INTEGER) AS a, b], where=[(a = 2)])
+   :                 +- TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[a, b, c])
+   +- Exchange(distribution=[hash[e]])
+      +- Calc(select=[e])
+         +- HashAggregate(isMerge=[true], groupBy=[d], select=[d, Final_COUNT(count$0) AS e])
+            +- Exchange(distribution=[hash[d]])
+               +- LocalHashAggregate(groupBy=[d], select=[d, Partial_COUNT(e) AS count$0])
+                  +- Calc(select=[CAST(2 AS INTEGER) AS d, e], where=[(d = 2)])
+                     +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[d, e, f, g, h])
 ]]>
     </Resource>
   </TestCase>
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/SortMergeJoinTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/SortMergeJoinTest.xml
index 249f17770908a..b125f43544167 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/SortMergeJoinTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/SortMergeJoinTest.xml
@@ -241,20 +241,22 @@ LogicalProject(a=[$0], b=[$1], d=[$2], e=[$3])
     </Resource>
     <Resource name="optimized exec plan">
       <![CDATA[
-Calc(select=[a, b, CAST(2 AS INTEGER) AS d, e])
-+- SortMergeJoin(joinType=[InnerJoin], where=[((a = d) AND (b = e))], select=[a, b, d, e])
-   :- Exchange(distribution=[hash[a, b]])
-   :  +- HashAggregate(isMerge=[true], groupBy=[a], select=[a, Final_COUNT(count$0) AS b])
-   :     +- Exchange(distribution=[hash[a]])
-   :        +- LocalHashAggregate(groupBy=[a], select=[a, Partial_COUNT(b) AS count$0])
-   :           +- Calc(select=[CAST(2 AS INTEGER) AS a, b], where=[(a = 2)])
-   :              +- TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[a, b, c])
-   +- Exchange(distribution=[hash[d, e]])
-      +- HashAggregate(isMerge=[true], groupBy=[d], select=[d, Final_COUNT(count$0) AS e])
-         +- Exchange(distribution=[hash[d]])
-            +- LocalHashAggregate(groupBy=[d], select=[d, Partial_COUNT(e) AS count$0])
-               +- Calc(select=[CAST(2 AS INTEGER) AS d, e], where=[(d = 2)])
-                  +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[d, e, f, g, h])
+Calc(select=[CAST(2 AS INTEGER) AS a, b, CAST(2 AS INTEGER) AS d, e])
++- SortMergeJoin(joinType=[InnerJoin], where=[(b = e)], select=[b, e])
+   :- Exchange(distribution=[hash[b]])
+   :  +- Calc(select=[b])
+   :     +- HashAggregate(isMerge=[true], groupBy=[a], select=[a, Final_COUNT(count$0) AS b])
+   :        +- Exchange(distribution=[hash[a]])
+   :           +- LocalHashAggregate(groupBy=[a], select=[a, Partial_COUNT(b) AS count$0])
+   :              +- Calc(select=[CAST(2 AS INTEGER) AS a, b], where=[(a = 2)])
+   :                 +- TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[a, b, c])
+   +- Exchange(distribution=[hash[e]])
+      +- Calc(select=[e])
+         +- HashAggregate(isMerge=[true], groupBy=[d], select=[d, Final_COUNT(count$0) AS e])
+            +- Exchange(distribution=[hash[d]])
+               +- LocalHashAggregate(groupBy=[d], select=[d, Partial_COUNT(e) AS count$0])
+                  +- Calc(select=[CAST(2 AS INTEGER) AS d, e], where=[(d = 2)])
+                     +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[d, e, f, g, h])
 ]]>
     </Resource>
   </TestCase>
@@ -280,43 +282,6 @@ Calc(select=[c, g])
    +- Exchange(distribution=[hash[e, d]])
       +- Calc(select=[d, e, g])
          +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[d, e, f, g, h])
-]]>
-    </Resource>
-  </TestCase>
-  <TestCase name="testInnerJoinWithJoinConditionPushDown">
-    <Resource name="sql">
-      <![CDATA[
-SELECT * FROM
-   (select a, count(b) as b from MyTable1 group by a)
-   join
-   (select d, count(e) as e from MyTable2 group by d)
-   on a = d and b = e and d = 2 and b = 1
-]]>
-    </Resource>
-    <Resource name="ast">
-      <![CDATA[
-LogicalProject(a=[$0], b=[$1], d=[$2], e=[$3])
-+- LogicalJoin(condition=[AND(=($0, $2), =($1, $3), =($2, 2), =($1, 1))], joinType=[inner])
-   :- LogicalAggregate(group=[{0}], b=[COUNT($1)])
-   :  +- LogicalProject(a=[$0], b=[$1])
-   :     +- LogicalTableScan(table=[[default_catalog, default_database, MyTable1]])
-   +- LogicalAggregate(group=[{0}], e=[COUNT($1)])
-      +- LogicalProject(d=[$0], e=[$1])
-         +- LogicalTableScan(table=[[default_catalog, default_database, MyTable2]])
-]]>
-    </Resource>
-    <Resource name="optimized exec plan">
-      <![CDATA[
-Calc(select=[a, 1 AS b, CAST(2 AS INTEGER) AS d, 1 AS e])
-+- MultipleInput(members=[\nSortMergeJoin(joinType=[InnerJoin], where=[(a = d)], select=[a, d])\n:- Calc(select=[a], where=[(b = 1)])\n:  +- HashAggregate(isMerge=[true], groupBy=[a], select=[a, Final_COUNT(count$0) AS b])\n:     +- [#1] Exchange(distribution=[hash[a]])\n+- Calc(select=[d], where=[(e = 1)])\n   +- HashAggregate(isMerge=[true], groupBy=[d], select=[d, Final_COUNT(count$0) AS e])\n      +- [#2] Exchange(distribution=[hash[d]])\n])
-   :- Exchange(distribution=[hash[a]])
-   :  +- LocalHashAggregate(groupBy=[a], select=[a, Partial_COUNT(b) AS count$0])
-   :     +- Calc(select=[CAST(2 AS INTEGER) AS a, b], where=[(a = 2)])
-   :        +- TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[a, b, c])
-   +- Exchange(distribution=[hash[d]])
-      +- LocalHashAggregate(groupBy=[d], select=[d, Partial_COUNT(e) AS count$0])
-         +- Calc(select=[CAST(2 AS INTEGER) AS d, e], where=[(d = 2)])
-            +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[d, e, f, g, h])
 ]]>
     </Resource>
   </TestCase>
@@ -416,20 +381,22 @@ LogicalProject(a=[$0], b=[$1], d=[$2], e=[$3])
     </Resource>
     <Resource name="optimized exec plan">
       <![CDATA[
-Calc(select=[CAST(2 AS INTEGER) AS a, b, d, CAST(e AS BIGINT) AS e])
-+- SortMergeJoin(joinType=[InnerJoin], where=[((a = d) AND (b = e))], select=[a, b, d, e])
-   :- Exchange(distribution=[hash[a, b]])
-   :  +- HashAggregate(isMerge=[true], groupBy=[a], select=[a, Final_COUNT(count$0) AS b])
-   :     +- Exchange(distribution=[hash[a]])
-   :        +- LocalHashAggregate(groupBy=[a], select=[a, Partial_COUNT(b) AS count$0])
-   :           +- Calc(select=[CAST(2 AS INTEGER) AS a, b], where=[(a = 2)])
-   :              +- TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[a, b, c])
-   +- Exchange(distribution=[hash[d, e]])
-      +- HashAggregate(isMerge=[true], groupBy=[d], select=[d, Final_COUNT(count$0) AS e])
-         +- Exchange(distribution=[hash[d]])
-            +- LocalHashAggregate(groupBy=[d], select=[d, Partial_COUNT(e) AS count$0])
-               +- Calc(select=[CAST(2 AS INTEGER) AS d, e], where=[(d = 2)])
-                  +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[d, e, f, g, h])
+Calc(select=[CAST(2 AS INTEGER) AS a, b, CAST(2 AS INTEGER) AS d, CAST(e AS BIGINT) AS e])
++- SortMergeJoin(joinType=[InnerJoin], where=[(b = e)], select=[b, e])
+   :- Exchange(distribution=[hash[b]])
+   :  +- Calc(select=[b])
+   :     +- HashAggregate(isMerge=[true], groupBy=[a], select=[a, Final_COUNT(count$0) AS b])
+   :        +- Exchange(distribution=[hash[a]])
+   :           +- LocalHashAggregate(groupBy=[a], select=[a, Partial_COUNT(b) AS count$0])
+   :              +- Calc(select=[CAST(2 AS INTEGER) AS a, b], where=[(a = 2)])
+   :                 +- TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[a, b, c])
+   +- Exchange(distribution=[hash[e]])
+      +- Calc(select=[e])
+         +- HashAggregate(isMerge=[true], groupBy=[d], select=[d, Final_COUNT(count$0) AS e])
+            +- Exchange(distribution=[hash[d]])
+               +- LocalHashAggregate(groupBy=[d], select=[d, Partial_COUNT(e) AS count$0])
+                  +- Calc(select=[CAST(2 AS INTEGER) AS d, e], where=[(d = 2)])
+                     +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[d, e, f, g, h])
 ]]>
     </Resource>
   </TestCase>
@@ -656,20 +623,22 @@ LogicalProject(a=[$0], b=[$1], d=[$2], e=[$3])
     </Resource>
     <Resource name="optimized exec plan">
       <![CDATA[
-Calc(select=[a, CAST(b AS BIGINT) AS b, CAST(2 AS INTEGER) AS d, e])
-+- SortMergeJoin(joinType=[InnerJoin], where=[((a = d) AND (b = e))], select=[a, b, d, e])
-   :- Exchange(distribution=[hash[a, b]])
-   :  +- HashAggregate(isMerge=[true], groupBy=[a], select=[a, Final_COUNT(count$0) AS b])
-   :     +- Exchange(distribution=[hash[a]])
-   :        +- LocalHashAggregate(groupBy=[a], select=[a, Partial_COUNT(b) AS count$0])
-   :           +- Calc(select=[CAST(2 AS INTEGER) AS a, b], where=[(a = 2)])
-   :              +- TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[a, b, c])
-   +- Exchange(distribution=[hash[d, e]])
-      +- HashAggregate(isMerge=[true], groupBy=[d], select=[d, Final_COUNT(count$0) AS e])
-         +- Exchange(distribution=[hash[d]])
-            +- LocalHashAggregate(groupBy=[d], select=[d, Partial_COUNT(e) AS count$0])
-               +- Calc(select=[CAST(2 AS INTEGER) AS d, e], where=[(d = 2)])
-                  +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[d, e, f, g, h])
+Calc(select=[CAST(2 AS INTEGER) AS a, CAST(b AS BIGINT) AS b, CAST(2 AS INTEGER) AS d, e])
++- SortMergeJoin(joinType=[InnerJoin], where=[(b = e)], select=[b, e])
+   :- Exchange(distribution=[hash[b]])
+   :  +- Calc(select=[b])
+   :     +- HashAggregate(isMerge=[true], groupBy=[a], select=[a, Final_COUNT(count$0) AS b])
+   :        +- Exchange(distribution=[hash[a]])
+   :           +- LocalHashAggregate(groupBy=[a], select=[a, Partial_COUNT(b) AS count$0])
+   :              +- Calc(select=[CAST(2 AS INTEGER) AS a, b], where=[(a = 2)])
+   :                 +- TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[a, b, c])
+   +- Exchange(distribution=[hash[e]])
+      +- Calc(select=[e])
+         +- HashAggregate(isMerge=[true], groupBy=[d], select=[d, Final_COUNT(count$0) AS e])
+            +- Exchange(distribution=[hash[d]])
+               +- LocalHashAggregate(groupBy=[d], select=[d, Partial_COUNT(e) AS count$0])
+                  +- Calc(select=[CAST(2 AS INTEGER) AS d, e], where=[(d = 2)])
+                     +- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[d, e, f, g, h])
 ]]>
     </Resource>
   </TestCase>
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 0e508ef7691de..aa0a43e59b279 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
@@ -517,12 +517,10 @@ LogicalProject(a=[$0], c=[$2], c0=[$9])
     </Resource>
     <Resource name="optimized exec plan">
       <![CDATA[
-Calc(select=[a, c, c0])
-+- HashJoin(joinType=[InnerJoin], where=[(a = a0)], select=[a, c, a0, c0], build=[left])
-   :- Exchange(distribution=[keep_input_as_is[hash[a]]])
-   :  +- Calc(select=[a, CAST('1' AS VARCHAR(2147483647)) AS c])(reuse_id=[1])
-   :     +- Exchange(distribution=[hash[a]])
-   :        +- TableSourceScan(table=[[default_catalog, default_database, MyTable, partitions=[{c=1}], project=[a], metadata=[]]], fields=[a], hints=[[[OPTIONS options:{partition-list=c:1;c:2}]]])
+Calc(select=[a, CAST('1' AS VARCHAR(2147483647)) AS c, CAST('1' AS VARCHAR(2147483647)) AS c0])
++- HashJoin(joinType=[InnerJoin], where=[(a = a0)], select=[a, a0], build=[left])
+   :- Exchange(distribution=[hash[a]])
+   :  +- TableSourceScan(table=[[default_catalog, default_database, MyTable, partitions=[{c=1}], project=[a], metadata=[]]], fields=[a], hints=[[[OPTIONS options:{partition-list=c:1;c:2}]]])(reuse_id=[1])
    +- Exchange(distribution=[hash[a]], shuffle_mode=[BATCH])
       +- Reused(reference_id=[1])
 ]]>
@@ -767,10 +765,10 @@ LogicalProject(a=[$0], b=[$1], c=[$9])
     </Resource>
     <Resource name="optimized exec plan">
       <![CDATA[
-Calc(select=[a, b, c])
-+- HashJoin(joinType=[InnerJoin], where=[(a = a0)], select=[a, b, a0, c], build=[left])
+Calc(select=[a, CAST(2 AS BIGINT) AS b, c])
++- HashJoin(joinType=[InnerJoin], where=[(a = a0)], select=[a, a0, c], build=[left])
    :- Exchange(distribution=[hash[a]])
-   :  +- Calc(select=[a, CAST(2 AS BIGINT) AS b], where=[(b = 2)])
+   :  +- Calc(select=[a], where=[(b = 2)])
    :     +- TableSourceScan(table=[[default_catalog, default_database, MyTable, filter=[], project=[a, b, c], metadata=[]]], fields=[a, b, c])(reuse_id=[1])
    +- Exchange(distribution=[hash[a]])
       +- Calc(select=[a, c], where=[(b = 3)])
@@ -801,10 +799,10 @@ LogicalProject(a=[$0], b=[$1], c=[$10])
     </Resource>
     <Resource name="optimized exec plan">
       <![CDATA[
-Calc(select=[a, b, c])
-+- Join(joinType=[InnerJoin], where=[(a = a0)], select=[a, b, a0, c], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])
+Calc(select=[a, CAST(2 AS BIGINT) AS b, c])
++- Join(joinType=[InnerJoin], where=[(a = a0)], select=[a, a0, c], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])
    :- Exchange(distribution=[hash[a]])
-   :  +- Calc(select=[a, CAST(2 AS BIGINT) AS b], where=[(b = 2)])
+   :  +- Calc(select=[a], where=[(b = 2)])
    :     +- TableSourceScan(table=[[default_catalog, default_database, MyTable, project=[a, b, c, nested_s], metadata=[], watermark=[-(TO_TIMESTAMP(c, nested_s), 5000:INTERVAL SECOND)], watermarkEmitStrategy=[on-periodic], filter=[]]], fields=[a, b, c, nested_s])(reuse_id=[1])
    +- Exchange(distribution=[hash[a]])
       +- Calc(select=[a, c], where=[(b = 3)])
@@ -1196,8 +1194,8 @@ LogicalProject(a=[$0], c=[$2], c0=[$9])
     </Resource>
     <Resource name="optimized exec plan">
       <![CDATA[
-Calc(select=[a, c, c0])
-+- MultipleInput(readOrder=[0,1], members=[\nHashJoin(joinType=[InnerJoin], where=[(a = a0)], select=[a, c, a0, c0], build=[left])\n:- Calc(select=[a, CAST('1' AS VARCHAR(2147483647)) AS c])\n:  +- [#1] Exchange(distribution=[hash[a]])\n+- Calc(select=[a, CAST('2' AS VARCHAR(2147483647)) AS c])\n   +- [#2] Exchange(distribution=[hash[a]])\n])
+Calc(select=[a, CAST('1' AS VARCHAR(2147483647)) AS c, CAST('2' AS VARCHAR(2147483647)) AS c0])
++- HashJoin(joinType=[InnerJoin], where=[(a = a0)], select=[a, a0], build=[left])
    :- Exchange(distribution=[hash[a]])
    :  +- TableSourceScan(table=[[default_catalog, default_database, MyTable, partitions=[{c=1}], project=[a], metadata=[]]], fields=[a], hints=[[[OPTIONS options:{partition-list=c:1;c:2}]]])
    +- Exchange(distribution=[hash[a]])
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/JoinTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/JoinTest.xml
index eeebf3f02d893..bee054a0dfceb 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/JoinTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/JoinTest.xml
@@ -427,18 +427,20 @@ LogicalProject(a1=[$0], a2=[$1], b1=[$2], b2=[$3])
     </Resource>
     <Resource name="optimized exec plan">
       <![CDATA[
-Calc(select=[a1, a2, CAST(2 AS INTEGER) AS b1, b2])
-+- Join(joinType=[InnerJoin], where=[((a1 = b1) AND (a2 = b2))], select=[a1, a2, b1, b2], leftInputSpec=[JoinKeyContainsUniqueKey], rightInputSpec=[JoinKeyContainsUniqueKey])
-   :- Exchange(distribution=[hash[a1, a2]])
-   :  +- GroupAggregate(groupBy=[a1], select=[a1, COUNT(a2) AS a2])
-   :     +- Exchange(distribution=[hash[a1]])
-   :        +- Calc(select=[CAST(2 AS INTEGER) AS a1, a2], where=[(a1 = 2)])
-   :           +- TableSourceScan(table=[[default_catalog, default_database, A]], fields=[a1, a2, a3])
-   +- Exchange(distribution=[hash[b1, b2]])
-      +- GroupAggregate(groupBy=[b1], select=[b1, COUNT(b2) AS b2])
-         +- Exchange(distribution=[hash[b1]])
-            +- Calc(select=[CAST(2 AS INTEGER) AS b1, b2], where=[(b1 = 2)])
-               +- TableSourceScan(table=[[default_catalog, default_database, B]], fields=[b1, b2, b3])
+Calc(select=[CAST(2 AS INTEGER) AS a1, a2, CAST(2 AS INTEGER) AS b1, b2])
++- Join(joinType=[InnerJoin], where=[(a2 = b2)], select=[a2, b2], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])
+   :- Exchange(distribution=[hash[a2]])
+   :  +- Calc(select=[a2])
+   :     +- GroupAggregate(groupBy=[a1], select=[a1, COUNT(a2) AS a2])
+   :        +- Exchange(distribution=[hash[a1]])
+   :           +- Calc(select=[CAST(2 AS INTEGER) AS a1, a2], where=[(a1 = 2)])
+   :              +- TableSourceScan(table=[[default_catalog, default_database, A]], fields=[a1, a2, a3])
+   +- Exchange(distribution=[hash[b2]])
+      +- Calc(select=[b2])
+         +- GroupAggregate(groupBy=[b1], select=[b1, COUNT(b2) AS b2])
+            +- Exchange(distribution=[hash[b1]])
+               +- Calc(select=[CAST(2 AS INTEGER) AS b1, b2], where=[(b1 = 2)])
+                  +- TableSourceScan(table=[[default_catalog, default_database, B]], fields=[b1, b2, b3])
 ]]>
     </Resource>
   </TestCase>
@@ -467,15 +469,15 @@ LogicalProject(a1=[$0], a2=[$1], b1=[$2], b2=[$3])
     </Resource>
     <Resource name="optimized exec plan">
       <![CDATA[
-Calc(select=[a1, 1 AS a2, CAST(2 AS INTEGER) AS b1, 1 AS b2])
-+- Join(joinType=[InnerJoin], where=[(a1 = b1)], select=[a1, b1], leftInputSpec=[JoinKeyContainsUniqueKey], rightInputSpec=[JoinKeyContainsUniqueKey])
-   :- Exchange(distribution=[hash[a1]])
+Calc(select=[CAST(2 AS INTEGER) AS a1, 1 AS a2, CAST(2 AS INTEGER) AS b1, 1 AS b2])
++- Join(joinType=[InnerJoin], where=[true], select=[a1, b1], leftInputSpec=[HasUniqueKey], rightInputSpec=[HasUniqueKey])
+   :- Exchange(distribution=[single])
    :  +- Calc(select=[a1], where=[(a2 = 1)])
    :     +- GroupAggregate(groupBy=[a1], select=[a1, COUNT(a2) AS a2])
    :        +- Exchange(distribution=[hash[a1]])
    :           +- Calc(select=[CAST(2 AS INTEGER) AS a1, a2], where=[(a1 = 2)])
    :              +- TableSourceScan(table=[[default_catalog, default_database, A]], fields=[a1, a2, a3])
-   +- Exchange(distribution=[hash[b1]])
+   +- Exchange(distribution=[single])
       +- Calc(select=[b1], where=[(b2 = 1)])
          +- GroupAggregate(groupBy=[b1], select=[b1, COUNT(b2) AS b2])
             +- Exchange(distribution=[hash[b1]])
@@ -1016,18 +1018,20 @@ LogicalProject(a1=[$0], a2=[$1], b1=[$2], b2=[$3])
     </Resource>
     <Resource name="optimized exec plan">
       <![CDATA[
-Calc(select=[CAST(2 AS INTEGER) AS a1, a2, b1, CAST(b2 AS BIGINT) AS b2])
-+- Join(joinType=[InnerJoin], where=[((a1 = b1) AND (b2 = a2))], select=[a1, a2, b1, b2], leftInputSpec=[JoinKeyContainsUniqueKey], rightInputSpec=[JoinKeyContainsUniqueKey])
-   :- Exchange(distribution=[hash[a1, a2]])
-   :  +- GroupAggregate(groupBy=[a1], select=[a1, COUNT(a2) AS a2])
-   :     +- Exchange(distribution=[hash[a1]])
-   :        +- Calc(select=[CAST(2 AS INTEGER) AS a1, a2], where=[(a1 = 2)])
-   :           +- TableSourceScan(table=[[default_catalog, default_database, A]], fields=[a1, a2, a3])
-   +- Exchange(distribution=[hash[b1, b2]])
-      +- GroupAggregate(groupBy=[b1], select=[b1, COUNT(b2) AS b2])
-         +- Exchange(distribution=[hash[b1]])
-            +- Calc(select=[CAST(2 AS INTEGER) AS b1, b2], where=[(b1 = 2)])
-               +- TableSourceScan(table=[[default_catalog, default_database, B]], fields=[b1, b2, b3])
+Calc(select=[CAST(2 AS INTEGER) AS a1, a2, CAST(2 AS INTEGER) AS b1, CAST(b2 AS BIGINT) AS b2])
++- Join(joinType=[InnerJoin], where=[(b2 = a2)], select=[a2, b2], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])
+   :- Exchange(distribution=[hash[a2]])
+   :  +- Calc(select=[a2])
+   :     +- GroupAggregate(groupBy=[a1], select=[a1, COUNT(a2) AS a2])
+   :        +- Exchange(distribution=[hash[a1]])
+   :           +- Calc(select=[CAST(2 AS INTEGER) AS a1, a2], where=[(a1 = 2)])
+   :              +- TableSourceScan(table=[[default_catalog, default_database, A]], fields=[a1, a2, a3])
+   +- Exchange(distribution=[hash[b2]])
+      +- Calc(select=[b2])
+         +- GroupAggregate(groupBy=[b1], select=[b1, COUNT(b2) AS b2])
+            +- Exchange(distribution=[hash[b1]])
+               +- Calc(select=[CAST(2 AS INTEGER) AS b1, b2], where=[(b1 = 2)])
+                  +- TableSourceScan(table=[[default_catalog, default_database, B]], fields=[b1, b2, b3])
 ]]>
     </Resource>
   </TestCase>
@@ -1423,18 +1427,20 @@ LogicalProject(a1=[$0], a2=[$1], b1=[$2], b2=[$3])
     </Resource>
     <Resource name="optimized exec plan">
       <![CDATA[
-Calc(select=[a1, CAST(a2 AS BIGINT) AS a2, CAST(2 AS INTEGER) AS b1, b2])
-+- Join(joinType=[InnerJoin], where=[((a1 = b1) AND (a2 = b2))], select=[a1, a2, b1, b2], leftInputSpec=[JoinKeyContainsUniqueKey], rightInputSpec=[JoinKeyContainsUniqueKey])
-   :- Exchange(distribution=[hash[a1, a2]])
-   :  +- GroupAggregate(groupBy=[a1], select=[a1, COUNT(a2) AS a2])
-   :     +- Exchange(distribution=[hash[a1]])
-   :        +- Calc(select=[CAST(2 AS INTEGER) AS a1, a2], where=[(a1 = 2)])
-   :           +- TableSourceScan(table=[[default_catalog, default_database, A]], fields=[a1, a2, a3])
-   +- Exchange(distribution=[hash[b1, b2]])
-      +- GroupAggregate(groupBy=[b1], select=[b1, COUNT(b2) AS b2])
-         +- Exchange(distribution=[hash[b1]])
-            +- Calc(select=[CAST(2 AS INTEGER) AS b1, b2], where=[(b1 = 2)])
-               +- TableSourceScan(table=[[default_catalog, default_database, B]], fields=[b1, b2, b3])
+Calc(select=[CAST(2 AS INTEGER) AS a1, CAST(a2 AS BIGINT) AS a2, CAST(2 AS INTEGER) AS b1, b2])
++- Join(joinType=[InnerJoin], where=[(a2 = b2)], select=[a2, b2], leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey])
+   :- Exchange(distribution=[hash[a2]])
+   :  +- Calc(select=[a2])
+   :     +- GroupAggregate(groupBy=[a1], select=[a1, COUNT(a2) AS a2])
+   :        +- Exchange(distribution=[hash[a1]])
+   :           +- Calc(select=[CAST(2 AS INTEGER) AS a1, a2], where=[(a1 = 2)])
+   :              +- TableSourceScan(table=[[default_catalog, default_database, A]], fields=[a1, a2, a3])
+   +- Exchange(distribution=[hash[b2]])
+      +- Calc(select=[b2])
+         +- GroupAggregate(groupBy=[b1], select=[b1, COUNT(b2) AS b2])
+            +- Exchange(distribution=[hash[b1]])
+               +- Calc(select=[CAST(2 AS INTEGER) AS b1, b2], where=[(b1 = 2)])
+                  +- TableSourceScan(table=[[default_catalog, default_database, B]], fields=[b1, b2, b3])
 ]]>
     </Resource>
   </TestCase>
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/join/BroadcastHashJoinTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/join/BroadcastHashJoinTest.scala
index ec3ada5a2fb6f..fd3bc8c2b1816 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/join/BroadcastHashJoinTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/join/BroadcastHashJoinTest.scala
@@ -124,4 +124,11 @@ class BroadcastHashJoinTest extends JoinTestBase {
       .hasMessageContaining("Cannot generate a valid execution plan for the given query")
       .isInstanceOf[TableException]
   }
+
+  @Test
+  override def testInnerJoinWithJoinConditionPushDown(): Unit = {
+    assertThatThrownBy(() => super.testInnerJoinWithJoinConditionPushDown())
+      .hasMessageContaining("Cannot generate a valid execution plan for the given query")
+      .isInstanceOf[TableException]
+  }
 }
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/join/ShuffledHashJoinTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/join/ShuffledHashJoinTest.scala
index e7ad20d02c49b..0370b2c4c0cc0 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/join/ShuffledHashJoinTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/join/ShuffledHashJoinTest.scala
@@ -96,4 +96,11 @@ class ShuffledHashJoinTest extends JoinTestBase {
       .hasMessageContaining("Cannot generate a valid execution plan for the given query")
       .isInstanceOf[TableException]
   }
+
+  @Test
+  override def testInnerJoinWithJoinConditionPushDown(): Unit = {
+    assertThatThrownBy(() => super.testInnerJoinWithJoinConditionPushDown())
+      .hasMessageContaining("Cannot generate a valid execution plan for the given query")
+      .isInstanceOf[TableException]
+  }
 }
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/join/SortMergeJoinTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/join/SortMergeJoinTest.scala
index e10732166ee5e..dffbe8e8b8223 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/join/SortMergeJoinTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/join/SortMergeJoinTest.scala
@@ -94,4 +94,10 @@ class SortMergeJoinTest extends JoinTestBase {
       .isInstanceOf[TableException]
   }
 
+  @Test
+  override def testInnerJoinWithJoinConditionPushDown(): Unit = {
+    assertThatThrownBy(() => super.testInnerJoinWithJoinConditionPushDown())
+      .hasMessageContaining("Cannot generate a valid execution plan for the given query")
+      .isInstanceOf[TableException]
+  }
 }
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/table/DecimalITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/table/DecimalITCase.scala
index d0e50a00da6ec..be8b8d791d042 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/table/DecimalITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/table/DecimalITCase.scala
@@ -179,9 +179,6 @@ class DecimalITCase extends BatchTestBase {
 
   @Test
   def testJoin1(): Unit = {
-    tEnv.getConfig
-      .set(ExecutionConfigOptions.TABLE_EXEC_DISABLED_OPERATORS, "HashJoin, NestedLoopJoin")
-
     checkQuery(
       Seq(DECIMAL(8, 2), DECIMAL(8, 4), INT, DOUBLE),
       s1r(d"1", d"1", 1, 1.0),
@@ -193,9 +190,6 @@ class DecimalITCase extends BatchTestBase {
 
   @Test
   def testJoin2(): Unit = {
-    tEnv.getConfig
-      .set(ExecutionConfigOptions.TABLE_EXEC_DISABLED_OPERATORS, "HashJoin, NestedLoopJoin")
-
     checkQuery(
       Seq(DECIMAL(8, 2), DECIMAL(8, 4), INT, DOUBLE),
       s1r(d"1", d"1", 1, 1.0),
@@ -207,9 +201,6 @@ class DecimalITCase extends BatchTestBase {
 
   @Test
   def testJoin3(): Unit = {
-    tEnv.getConfig
-      .set(ExecutionConfigOptions.TABLE_EXEC_DISABLED_OPERATORS, "HashJoin, NestedLoopJoin")
-
     checkQuery(
       Seq(DECIMAL(8, 2), DECIMAL(8, 4), INT, DOUBLE),
       s1r(d"1", d"1", 1, 1.0),
@@ -222,9 +213,6 @@ class DecimalITCase extends BatchTestBase {
 
   @Test
   def testJoin4(): Unit = {
-    tEnv.getConfig
-      .set(ExecutionConfigOptions.TABLE_EXEC_DISABLED_OPERATORS, "HashJoin, NestedLoopJoin")
-
     checkQuery(
       Seq(DECIMAL(8, 2), DECIMAL(8, 4), INT, DOUBLE),
       s1r(d"1", d"1", 1, 1.0),
@@ -236,9 +224,6 @@ class DecimalITCase extends BatchTestBase {
 
   @Test
   def testJoin5(): Unit = {
-    tEnv.getConfig
-      .set(ExecutionConfigOptions.TABLE_EXEC_DISABLED_OPERATORS, "HashJoin, NestedLoopJoin")
-
     checkQuery(
       Seq(DECIMAL(8, 2), DECIMAL(8, 4), INT, DOUBLE),
       s1r(d"1", d"1", 1, 1.0),
@@ -250,9 +235,6 @@ class DecimalITCase extends BatchTestBase {
 
   @Test
   def testJoin6(): Unit = {
-    tEnv.getConfig
-      .set(ExecutionConfigOptions.TABLE_EXEC_DISABLED_OPERATORS, "HashJoin, NestedLoopJoin")
-
     checkQuery(
       Seq(DECIMAL(8, 2), DECIMAL(8, 4), INT, DOUBLE),
       s1r(d"1", d"1", 1, 1.0),
@@ -264,8 +246,6 @@ class DecimalITCase extends BatchTestBase {
 
   @Test
   def testJoin7(): Unit = {
-    tEnv.getConfig
-      .set(ExecutionConfigOptions.TABLE_EXEC_DISABLED_OPERATORS, "HashJoin, NestedLoopJoin")
     checkQuery(
       Seq(DECIMAL(8, 2), DECIMAL(8, 4), INT, DOUBLE),
       s1r(d"1", d"1", 1, 1.0),