From 30e8f3cec1ff90e3fc792e02113e2c4ba0e96d5d Mon Sep 17 00:00:00 2001
From: Sergey Nuyanzin <snuyanzin@gmail.com>
Date: Thu, 1 Feb 2024 21:52:01 +0100
Subject: [PATCH] [FLINK-31362][table] plans

---
 .../batch/sql/join/BroadcastHashJoinTest.xml  | 19 +++++++++++++++++++
 .../batch/sql/join/NestedLoopJoinTest.xml     |  4 +---
 .../batch/sql/join/ShuffledHashJoinTest.xml   | 19 +++++++++++++++++++
 .../plan/batch/sql/join/SortMergeJoinTest.xml | 19 +++++++++++++++++++
 ...inDependentConditionDerivationRuleTest.xml |  2 +-
 .../logical/RemoveSingleAggregateRuleTest.xml |  2 +-
 ...bqueryCorrelateVariablesValidationTest.xml |  4 ++--
 .../sql/join/BroadcastHashJoinTest.scala      |  7 -------
 .../batch/sql/join/ShuffledHashJoinTest.scala |  7 -------
 .../batch/sql/join/SortMergeJoinTest.scala    |  7 -------
 10 files changed, 62 insertions(+), 28 deletions(-)

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 16537dc550afc..d5739c895240a 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
@@ -683,4 +683,23 @@ HashJoin(joinType=[LeftOuterJoin], where=[(k = k0)], select=[k, v, k0, v0], isBr
 ]]>
     </Resource>
   </TestCase>
+  <TestCase name="testRightOuterJoinOnFalse">
+		<Resource name="sql">
+			<![CDATA[SELECT * FROM MyTable2 RIGHT OUTER JOIN MyTable1 ON false]]>
+		</Resource>
+		<Resource name="ast">
+			<![CDATA[
+LogicalProject(d=[$0], e=[$1], f=[$2], g=[$3], h=[$4], a=[$5], b=[$6], c=[$7])
++- LogicalJoin(condition=[false], joinType=[right])
+   :- LogicalTableScan(table=[[default_catalog, default_database, MyTable2, source: [TestTableSource(d, e, f, g, h)]]])
+   +- LogicalTableScan(table=[[default_catalog, default_database, MyTable1, source: [TestTableSource(a, b, c)]]])
+]]>
+		</Resource>
+		<Resource name="optimized exec plan">
+			<![CDATA[
+Calc(select=[null:INTEGER AS d, null:BIGINT AS e, null:INTEGER AS f, null:VARCHAR(2147483647) AS g, null:BIGINT AS h, a, b, c])
++- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable1, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+]]>
+		</Resource>
+	</TestCase>
 </Root>
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 770853f2fbe99..9582daba656b4 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
@@ -917,9 +917,7 @@ LogicalProject(d=[$0], e=[$1], f=[$2], g=[$3], h=[$4], a=[$5], b=[$6], c=[$7])
     </Resource>
     <Resource name="optimized exec plan">
       <![CDATA[
-NestedLoopJoin(joinType=[RightOuterJoin], where=[true], select=[d, e, f, g, h, a, b, c], build=[left])
-:- Exchange(distribution=[broadcast])
-:  +- Values(tuples=[[]], values=[d, e, f, g, h])
+Calc(select=[null:INTEGER AS d, null:BIGINT AS e, null:INTEGER AS f, null:VARCHAR(2147483647) AS g, null:BIGINT AS h, a, b, c])
 +- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable1, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
 ]]>
     </Resource>
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 1b14167bace44..f88db3febe085 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
@@ -795,4 +795,23 @@ HashJoin(joinType=[LeftOuterJoin], where=[(k = k0)], select=[k, v, k0, v0], buil
 ]]>
     </Resource>
   </TestCase>
+	<TestCase name="testRightOuterJoinOnFalse">
+		<Resource name="sql">
+			<![CDATA[SELECT * FROM MyTable2 RIGHT OUTER JOIN MyTable1 ON false]]>
+		</Resource>
+		<Resource name="ast">
+			<![CDATA[
+LogicalProject(d=[$0], e=[$1], f=[$2], g=[$3], h=[$4], a=[$5], b=[$6], c=[$7])
++- LogicalJoin(condition=[false], joinType=[right])
+   :- LogicalTableScan(table=[[default_catalog, default_database, MyTable2, source: [TestTableSource(d, e, f, g, h)]]])
+   +- LogicalTableScan(table=[[default_catalog, default_database, MyTable1, source: [TestTableSource(a, b, c)]]])
+]]>
+		</Resource>
+		<Resource name="optimized exec plan">
+			<![CDATA[
+Calc(select=[null:INTEGER AS d, null:BIGINT AS e, null:INTEGER AS f, null:VARCHAR(2147483647) AS g, null:BIGINT AS h, a, b, c])
++- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable1, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+]]>
+		</Resource>
+	</TestCase>
 </Root>
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 855fd71c969fc..c6dfb533767db 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
@@ -795,4 +795,23 @@ SortMergeJoin(joinType=[LeftOuterJoin], where=[(k = k0)], select=[k, v, k0, v0])
 ]]>
     </Resource>
   </TestCase>
+	<TestCase name="testRightOuterJoinOnFalse">
+		<Resource name="sql">
+			<![CDATA[SELECT * FROM MyTable2 RIGHT OUTER JOIN MyTable1 ON false]]>
+		</Resource>
+		<Resource name="ast">
+			<![CDATA[
+LogicalProject(d=[$0], e=[$1], f=[$2], g=[$3], h=[$4], a=[$5], b=[$6], c=[$7])
++- LogicalJoin(condition=[false], joinType=[right])
+   :- LogicalTableScan(table=[[default_catalog, default_database, MyTable2, source: [TestTableSource(d, e, f, g, h)]]])
+   +- LogicalTableScan(table=[[default_catalog, default_database, MyTable1, source: [TestTableSource(a, b, c)]]])
+]]>
+		</Resource>
+		<Resource name="optimized exec plan">
+			<![CDATA[
+Calc(select=[null:INTEGER AS d, null:BIGINT AS e, null:INTEGER AS f, null:VARCHAR(2147483647) AS g, null:BIGINT AS h, a, b, c])
++- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable1, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+]]>
+		</Resource>
+	</TestCase>
 </Root>
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/JoinDependentConditionDerivationRuleTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/JoinDependentConditionDerivationRuleTest.xml
index 896723ea5148f..96572c6507649 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/JoinDependentConditionDerivationRuleTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/JoinDependentConditionDerivationRuleTest.xml
@@ -120,7 +120,7 @@ LogicalProject(a=[$0], d=[$3])
     <Resource name="optimized rel plan">
       <![CDATA[
 LogicalProject(a=[$0], d=[$3])
-+- LogicalJoin(condition=[AND(OR(AND(=($0, 1), =($3, 2)), AND(=($0, 2), =($3, 1))), OR(AND(=($0, 3), =($3, 4)), AND(=($0, 4), =($3, 3))), SEARCH($0, Sarg[1, 2]), SEARCH($3, Sarg[1, 2]), SEARCH($0, Sarg[3, 4]), SEARCH($3, Sarg[3, 4]))], joinType=[inner])
++- LogicalJoin(condition=[false], joinType=[inner])
    :- LogicalTableScan(table=[[default_catalog, default_database, MyTable1, source: [TestTableSource(a, b, c)]]])
    +- LogicalTableScan(table=[[default_catalog, default_database, MyTable2, source: [TestTableSource(d, e, f, g, h)]]])
 ]]>
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/RemoveSingleAggregateRuleTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/RemoveSingleAggregateRuleTest.xml
index d8afcc2b74aa7..a864eaaeef12a 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/RemoveSingleAggregateRuleTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/RemoveSingleAggregateRuleTest.xml
@@ -22,7 +22,7 @@ limitations under the License.
     </Resource>
     <Resource name="ast">
       <![CDATA[
-LogicalProject(EXPR$0=[$SCALAR_QUERY({
+LogicalProject(variablesSet=[[$cor0]], EXPR$0=[$SCALAR_QUERY({
 LogicalProject(EXPR$0=[-($0, 1)])
   LogicalAggregate(group=[{}], agg#0=[COUNT($0)])
     LogicalProject(x=[$0])
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/subquery/SubqueryCorrelateVariablesValidationTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/subquery/SubqueryCorrelateVariablesValidationTest.xml
index 6a8db57acf8dd..9c42ceef11f5d 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/subquery/SubqueryCorrelateVariablesValidationTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/subquery/SubqueryCorrelateVariablesValidationTest.xml
@@ -26,7 +26,7 @@ WHERE  t1a = 'test'
     </Resource>
     <Resource name="ast">
       <![CDATA[
-LogicalProject(min_t1d=[$SCALAR_QUERY({
+LogicalProject(variablesSet=[[$cor0]], min_t1d=[$SCALAR_QUERY({
 LogicalAggregate(group=[{}], EXPR$0=[MIN($0)])
   LogicalProject($f0=[$cor0.t1d])
     LogicalFilter(condition=[=($0, _UTF-16LE'test')])
@@ -65,7 +65,7 @@ FROM   t1
     </Resource>
     <Resource name="ast">
       <![CDATA[
-LogicalProject(min_t3d=[$SCALAR_QUERY({
+LogicalProject(variablesSet=[[$cor0]], min_t3d=[$SCALAR_QUERY({
 LogicalAggregate(group=[{}], EXPR$0=[MIN($0)])
   LogicalProject(t3d=[$3])
     LogicalFilter(condition=[=($0, $cor0.t1a)])
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 0f3c494ee5e6a..0d608b04b31e7 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
@@ -69,13 +69,6 @@ class BroadcastHashJoinTest extends JoinTestBase {
       .isInstanceOf[TableException]
   }
 
-  @Test
-  override def testRightOuterJoinOnFalse(): Unit = {
-    assertThatThrownBy(() => super.testRightOuterJoinOnFalse())
-      .hasMessageContaining("Cannot generate a valid execution plan for the given query")
-      .isInstanceOf[TableException]
-  }
-
   @Test
   override def testRightOuterJoinWithNonEquiPred(): Unit = {
     assertThatThrownBy(() => super.testRightOuterJoinWithNonEquiPred())
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 36687fb6fd91d..c9ad5909f9bc0 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
@@ -69,13 +69,6 @@ class ShuffledHashJoinTest extends JoinTestBase {
       .isInstanceOf[TableException]
   }
 
-  @Test
-  override def testRightOuterJoinOnFalse(): Unit = {
-    assertThatThrownBy(() => super.testRightOuterJoinOnFalse())
-      .hasMessageContaining("Cannot generate a valid execution plan for the given query")
-      .isInstanceOf[TableException]
-  }
-
   @Test
   override def testRightOuterJoinWithNonEquiPred(): Unit = {
     assertThatThrownBy(() => super.testRightOuterJoinWithNonEquiPred())
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 2d7cf24fab4d0..7e526138e263e 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
@@ -66,13 +66,6 @@ class SortMergeJoinTest extends JoinTestBase {
       .isInstanceOf[TableException]
   }
 
-  @Test
-  override def testRightOuterJoinOnFalse(): Unit = {
-    assertThatThrownBy(() => super.testRightOuterJoinOnFalse())
-      .hasMessageContaining("Cannot generate a valid execution plan for the given query")
-      .isInstanceOf[TableException]
-  }
-
   @Test
   override def testRightOuterJoinWithNonEquiPred(): Unit = {
     assertThatThrownBy(() => super.testRightOuterJoinWithNonEquiPred())