[FLINK-31362][table] plans

snuyanzin · Aug 22, 2024 · 61e50f1 · 61e50f1
1 parent 8a706db
commit 61e50f1
Show file tree

Hide file tree

Showing 10 changed files with 62 additions and 28 deletions.
diff --git a/...st/resources/org/apache/flink/table/planner/plan/batch/sql/join/BroadcastHashJoinTest.xml b/...st/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/.../test/resources/org/apache/flink/table/planner/plan/batch/sql/join/NestedLoopJoinTest.xml b/.../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/...est/resources/org/apache/flink/table/planner/plan/batch/sql/join/ShuffledHashJoinTest.xml b/...est/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/...c/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/SortMergeJoinTest.xml b/...c/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/...pache/flink/table/planner/plan/rules/logical/JoinDependentConditionDerivationRuleTest.xml b/...pache/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/...urces/org/apache/flink/table/planner/plan/rules/logical/RemoveSingleAggregateRuleTest.xml b/...urces/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/...nk/table/planner/plan/rules/logical/subquery/SubqueryCorrelateVariablesValidationTest.xml b/...nk/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/...test/scala/org/apache/flink/table/planner/plan/batch/sql/join/BroadcastHashJoinTest.scala b/...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/.../test/scala/org/apache/flink/table/planner/plan/batch/sql/join/ShuffledHashJoinTest.scala b/.../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/...src/test/scala/org/apache/flink/table/planner/plan/batch/sql/join/SortMergeJoinTest.scala b/...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())