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 2d515d3bd47 [FLINK-39763][table] Allow inferring constants with `IS
NOT DISTINCT FROM`
2d515d3bd47 is described below
commit 2d515d3bd4788b3ce211a27e48b36cae545bb5dc
Author: Sergey Nuyanzin <[email protected]>
AuthorDate: Tue Jun 16 08:12:01 2026 +0200
[FLINK-39763][table] Allow inferring constants with `IS NOT DISTINCT FROM`
---
.../main/java/org/apache/calcite/rex/RexUtil.java | 12 +-
.../plan/batch/sql/DynamicFunctionPlanTest.xml | 9 +-
.../planner/plan/batch/sql/RemoveShuffleTest.xml | 54 ++++-----
.../plan/batch/sql/join/BroadcastHashJoinTest.xml | 85 ++++----------
.../plan/batch/sql/join/NestedLoopJoinTest.xml | 49 ++++----
.../plan/batch/sql/join/ShuffledHashJoinTest.xml | 127 ++++++++-------------
.../plan/batch/sql/join/SortMergeJoinTest.xml | 127 ++++++++-------------
.../table/planner/plan/optimize/ScanReuseTest.xml | 26 ++---
.../planner/plan/stream/sql/join/JoinTest.xml | 86 +++++++-------
.../batch/sql/join/BroadcastHashJoinTest.scala | 7 ++
.../plan/batch/sql/join/ShuffledHashJoinTest.scala | 7 ++
.../plan/batch/sql/join/SortMergeJoinTest.scala | 6 +
.../runtime/batch/table/DecimalITCase.scala | 20 ----
13 files changed, 260 insertions(+), 355 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 bddc6a15239..3072d3b9f38 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
@@ -76,13 +76,9 @@ import static
org.apache.calcite.rel.type.RelDataType.PRECISION_NOT_SPECIFIED;
/**
* Default implementation of {@link org.apache.calcite.rex.RexUtil}, the class
was copied over
- * because of current Calcite way of inferring constants from IS NOT DISTINCT
FROM clashes with
- * filter push down.
+ * because of Flink modifications.
*
- * <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 2488~2492
*/
public class RexUtil {
@@ -401,9 +397,7 @@ public class RexUtil {
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 c4442ef5dde..041111d0538 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 be2b0e58f29..f08d5635de7 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 897e52cf7f7..dc0d9c3e721 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 c37f2d9eb04..4f6c8e1e9a4 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 e3d85fe3193..1a1470c7146 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 249f1777090..b125f435441 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 0e508ef7691..aa0a43e59b2 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 eeebf3f02d8..bee054a0dfc 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 ec3ada5a2fb..fd3bc8c2b18 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 e7ad20d02c4..0370b2c4c0c 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 e10732166ee..dffbe8e8b82 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 d0e50a00da6..be8b8d791d0 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),
