[
https://issues.apache.org/jira/browse/FLINK-25447?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17655801#comment-17655801
]
Shengkai Fang commented on FLINK-25447:
---------------------------------------
After investigations, the main problem is that LogicalCalc doesn't generate
the correct {{RelCollation}} when converting the {{LogicalProject}} to
{{LogicalCalc}}. When the RelSubSet computes the cost, it finds the cancidate
can't satisfy the traits. (BTW, is there any better way we can print the
candiates in the RelSubSet...)
Just as the PR fixes, we only need to generate correct {{RelCollation}} in the
{{IntermediateTable}}. The problem only happens in the table api. When using
`CREATE TEMPORARY VIEW` to register the view, the problem disappers. The reason
behind is that we reuse the same calcite tree in the Table API, but in the SQL
we always generate a new calcite tree. Because the Table API, we use the
`QueryOperationCatalogView` to store the parsed view but in the SQL we use the
`ResolvedCatalogView` that only stores the origin SQL. Because of the
differences, the optimizer can reuse the view in the table api. For SQL
{code:java}
INSERT INTO sink1 SELECT a, b, d FROM sorted_view;
INSERT INTO sink2 SELECT a, b, c, d FROM sorted_view;
{code}
we have the following relblocks
{code:java}
relblock1:
sort
+- scan
relblock2:
sink1
+- project(fields=[a, b, d])
+-- scan(intermediate_table)
relblock3:
sink2
+- project(fields=[a, b, c, d])
+-- scan(intermediate_table)
{code}
But in the SQL, we have the following relblocks
{code:java}
relblock1:
sink1
+- project(fields=[a, b, d])
+-- sort
+--- scan
relblock2:
sink2
+- project(fields=[a, b, c, d])
+-- sort
+--- scan
{code}
Therefore, in the SQL we don't get the problem. (BTW, it seems we can reuse
more nodes in the Table API comparing to the SQL...)
The PR looks good in general, but I am not sure wheter we should also override
the {{getDistribution}} in the {{IntermediateTable}}.
> batch query cannot generate plan when a sorted view into multi sinks
> --------------------------------------------------------------------
>
> Key: FLINK-25447
> URL: https://issues.apache.org/jira/browse/FLINK-25447
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
> Affects Versions: 1.14.2
> Reporter: lincoln lee
> Assignee: Zheng yunhong
> Priority: Major
> Labels: pull-request-available
> Fix For: 1.17.0
>
>
> A batch query write a sorted view into multi sinks will get a cannot plan
> exception
> {code}
> @Test
> def testSortedResultIntoMultiSinks(): Unit = {
> util.tableEnv.executeSql(
> s"""
> |CREATE TABLE Src (
> | `a` INT,
> | `b` BIGINT,
> | `c` STRING,
> | `d` STRING,
> | `e` STRING
> |) WITH (
> | 'connector' = 'values',
> | 'bounded' = 'true'
> |)
> """.stripMargin)
> val query = "SELECT * FROM Src order by c"
> val table = util.tableEnv.sqlQuery(query)
> util.tableEnv.registerTable("sortedTable", table)
> util.tableEnv.executeSql(
> s"""
> |CREATE TABLE sink1 (
> | `a` INT,
> | `b` BIGINT,
> | `c` STRING
> |) WITH (
> | 'connector' = 'filesystem',
> | 'format' = 'testcsv',
> | 'path' = '/tmp/test'
> |)
> """.stripMargin)
> util.tableEnv.executeSql(
> s"""
> |CREATE TABLE sink2 (
> | `a` INT,
> | `b` BIGINT,
> | `c` STRING,
> | `d` STRING
> |) WITH (
> | 'connector' = 'filesystem',
> | 'format' = 'testcsv',
> | 'path' = '/tmp/test'
> |)
> """.stripMargin)
> val stmtSet= util.tableEnv.createStatementSet()
> stmtSet.addInsertSql(
> "insert into sink1 select a, b, listagg(d) from sortedTable group by a,
> b")
> stmtSet.addInsertSql(
> "insert into sink2 select a, b, c, d from sortedTable")
> util.verifyExecPlan(stmtSet)
> }
> {code}
> {code}
> org.apache.flink.table.api.TableException: Cannot generate a valid
> execution plan for the given query:
> LogicalSink(table=[default_catalog.default_database.sink2], fields=[a, b,
> c, d])
> +- LogicalProject(inputs=[0..3])
> +- LogicalTableScan(table=[[IntermediateRelTable_0]])
> This exception indicates that the query uses an unsupported SQL feature.
> Please check the documentation for the set of currently supported SQL
> features.
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkVolcanoProgram.optimize(FlinkVolcanoProgram.scala:76)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.$anonfun$optimize$1(FlinkChainedProgram.scala:62)
> at
> scala.collection.TraversableOnce.$anonfun$foldLeft$1(TraversableOnce.scala:156)
> at
> scala.collection.TraversableOnce.$anonfun$foldLeft$1$adapted(TraversableOnce.scala:156)
> at scala.collection.Iterator.foreach(Iterator.scala:937)
> at scala.collection.Iterator.foreach$(Iterator.scala:937)
> at scala.collection.AbstractIterator.foreach(Iterator.scala:1425)
> at scala.collection.IterableLike.foreach(IterableLike.scala:70)
> at scala.collection.IterableLike.foreach$(IterableLike.scala:69)
> at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
> at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:156)
> at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:154)
> at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:104)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.optimize(FlinkChainedProgram.scala:58)
> at
> org.apache.flink.table.planner.plan.optimize.BatchCommonSubGraphBasedOptimizer.optimizeTree(BatchCommonSubGraphBasedOptimizer.scala:88)
> at
> org.apache.flink.table.planner.plan.optimize.BatchCommonSubGraphBasedOptimizer.optimizeBlock(BatchCommonSubGraphBasedOptimizer.scala:59)
> at
> org.apache.flink.table.planner.plan.optimize.BatchCommonSubGraphBasedOptimizer.$anonfun$doOptimize$1(BatchCommonSubGraphBasedOptimizer.scala:47)
> at
> org.apache.flink.table.planner.plan.optimize.BatchCommonSubGraphBasedOptimizer.$anonfun$doOptimize$1$adapted(BatchCommonSubGraphBasedOptimizer.scala:47)
> at
> scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:58)
> at
> scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:51)
> at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
> at
> org.apache.flink.table.planner.plan.optimize.BatchCommonSubGraphBasedOptimizer.doOptimize(BatchCommonSubGraphBasedOptimizer.scala:47)
> at
> org.apache.flink.table.planner.plan.optimize.CommonSubGraphBasedOptimizer.optimize(CommonSubGraphBasedOptimizer.scala:77)
> at
> org.apache.flink.table.planner.delegation.PlannerBase.optimize(PlannerBase.scala:309)
> at
> org.apache.flink.table.planner.utils.TableTestUtilBase.assertPlanEquals(TableTestBase.scala:888)
> at
> org.apache.flink.table.planner.utils.TableTestUtilBase.doVerifyPlan(TableTestBase.scala:857)
> at
> org.apache.flink.table.planner.utils.TableTestUtilBase.verifyExecPlan(TableTestBase.scala:640)
> at
> org.apache.flink.table.planner.plan.batch.sql.TableSinkTest.testMultiSinksSplitOnSortedResult(TableSinkTest.scala:188)
> at
> java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> at
> java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> at
> java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> at java.base/java.lang.reflect.Method.invoke(Method.java:566)
> at
> org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:59)
> at
> org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
> at
> org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:56)
> at
> org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
> at
> org.junit.rules.ExpectedException$ExpectedExceptionStatement.evaluate(ExpectedException.java:258)
> at org.junit.rules.TestWatcher$1.evaluate(TestWatcher.java:61)
> at org.junit.rules.ExternalResource$1.evaluate(ExternalResource.java:54)
> at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306)
> at
> org.junit.runners.BlockJUnit4ClassRunner$1.evaluate(BlockJUnit4ClassRunner.java:100)
> at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:366)
> at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:103)
> at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:63)
> at org.junit.runners.ParentRunner$4.run(ParentRunner.java:331)
> at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:79)
> at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:329)
> at org.junit.runners.ParentRunner.access$100(ParentRunner.java:66)
> at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:293)
> at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306)
> at org.junit.runners.ParentRunner.run(ParentRunner.java:413)
> at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
> at
> com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:69)
> at
> com.intellij.rt.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:33)
> at
> com.intellij.rt.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:235)
> at com.intellij.rt.junit.JUnitStarter.main(JUnitStarter.java:54)
> Caused by: org.apache.calcite.plan.RelOptPlanner$CannotPlanException: There
> are not enough rules to produce a node with desired properties:
> convention=LOGICAL, FlinkRelDistributionTraitDef=any, sort=[].
> Missing conversion is LogicalProject[convention: NONE -> LOGICAL]
> There is 1 empty subset: rel#816:RelSubset#27.LOGICAL.any.[2
> ASC-nulls-first], the relevant part of the original plan is as follows
> 806:LogicalProject(a=[$0], b=[$1], c=[$2], d=[$3])
> 162:LogicalTableScan(subset=[rel#805:RelSubset#26.NONE.any.[]],
> table=[[IntermediateRelTable_0]])
> Root: rel#810:RelSubset#28.LOGICAL.any.[]
> Original rel:
> LogicalSort(subset=[rel#129:RelSubset#1.LOGICAL.any.[2 ASC-nulls-first]],
> sort0=[$2], dir0=[ASC-nulls-first]): rowcount = 1.0E8, cumulative cost =
> {1.0E8 rows, 3.684136148790473E10 cpu, 0.0 io, 0.0 network, 0.0 memory}, id =
> 127
> LogicalTableScan(subset=[rel#126:RelSubset#0.NONE.any.[]],
> table=[[default_catalog, default_database, SmallTable]]): rowcount = 1.0E8,
> cumulative cost = {1.0E8 rows, 1.00000001E8 cpu, 0.0 io, 0.0 network, 0.0
> memory}, id = 1
> Sets:
> Set#26, type: RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c,
> VARCHAR(2147483647) d, VARCHAR(2147483647) e)
> rel#805:RelSubset#26.NONE.any.[], best=null
>
> rel#162:LogicalTableScan.NONE.any.[](table=[IntermediateRelTable_0]),
> rowcount=1.0E8, cumulative cost={inf}
> rel#813:RelSubset#26.LOGICAL.any.[], best=rel#812
>
> rel#812:FlinkLogicalIntermediateTableScan.LOGICAL.any.[](table=[IntermediateRelTable_0],fields=a,
> b, c, d, e), rowcount=1.0E8, cumulative cost={1.0E8 rows, 1.0E8 cpu, 4.8E9
> io, 0.0 network, 0.0 memory}
> Set#27, type: RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c,
> VARCHAR(2147483647) d)
> rel#807:RelSubset#27.NONE.any.[2 ASC-nulls-first], best=null
> rel#806:LogicalProject.NONE.any.[2
> ASC-nulls-first](input=RelSubset#805,inputs=0..3), rowcount=1.0E8, cumulative
> cost={inf}
> rel#815:RelSubset#27.NONE.any.[], best=null
> rel#806:LogicalProject.NONE.any.[2
> ASC-nulls-first](input=RelSubset#805,inputs=0..3), rowcount=1.0E8, cumulative
> cost={inf}
>
> rel#814:LogicalCalc.NONE.any.[](input=RelSubset#805,expr#0..4={inputs},proj#0..3={exprs}),
> rowcount=1.0E8, cumulative cost={inf}
> rel#816:RelSubset#27.LOGICAL.any.[2 ASC-nulls-first], best=null
> rel#819:RelSubset#27.LOGICAL.any.[], best=rel#818
>
> rel#818:FlinkLogicalCalc.LOGICAL.any.[](input=RelSubset#813,select=a, b, c,
> d), rowcount=1.0E8, cumulative cost={2.0E8 rows, 1.0E8 cpu, 4.8E9 io, 0.0
> network, 0.0 memory}
> Set#28, type: RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c,
> VARCHAR(2147483647) d)
> rel#809:RelSubset#28.NONE.any.[], best=null
>
> rel#808:LogicalSink.NONE.any.[](input=RelSubset#807,table=default_catalog.default_database.sink2,fields=a,
> b, c, d), rowcount=1.0E8, cumulative cost={inf}
> rel#810:RelSubset#28.LOGICAL.any.[], best=null
>
> rel#811:AbstractConverter.LOGICAL.any.[](input=RelSubset#809,convention=LOGICAL,FlinkRelDistributionTraitDef=any,sort=[]),
> rowcount=1.0E8, cumulative cost={inf}
>
> rel#817:FlinkLogicalSink.LOGICAL.any.[](input=RelSubset#816,table=default_catalog.default_database.sink2,fields=a,
> b, c, d), rowcount=1.0E8, cumulative cost={inf}
> Graphviz:
> digraph G {
> root [style=filled,label="Root"];
> subgraph cluster26{
> label="Set 26 RecordType(INTEGER a, BIGINT b,
> VARCHAR(2147483647) c, VARCHAR(2147483647) d, VARCHAR(2147483647) e)";
> rel162
> [label="rel#162:LogicalTableScan\ntable=[IntermediateRelTable_0]\nrows=1.0E8,
> cost={inf}",shape=box]
> rel812
> [label="rel#812:FlinkLogicalIntermediateTableScan\ntable=[IntermediateRelTable_0],fields=a,
> b, c, d, e\nrows=1.0E8, cost={1.0E8 rows, 1.0E8 cpu, 4.8E9 io, 0.0 network,
> 0.0 memory}",color=blue,shape=box]
> subset805 [label="rel#805:RelSubset#26.NONE.any.[]"]
> subset813 [label="rel#813:RelSubset#26.LOGICAL.any.[]"]
> }
> subgraph cluster27{
> label="Set 27 RecordType(INTEGER a, BIGINT b,
> VARCHAR(2147483647) c, VARCHAR(2147483647) d)";
> rel806
> [label="rel#806:LogicalProject\ninput=RelSubset#805,inputs=0..3\nrows=1.0E8,
> cost={inf}",shape=box]
> rel814
> [label="rel#814:LogicalCalc\ninput=RelSubset#805,expr#0..4={inputs},proj#0..3={exprs}\nrows=1.0E8,
> cost={inf}",shape=box]
> rel818
> [label="rel#818:FlinkLogicalCalc\ninput=RelSubset#813,select=a, b, c,
> d\nrows=1.0E8, cost={2.0E8 rows, 1.0E8 cpu, 4.8E9 io, 0.0 network, 0.0
> memory}",color=blue,shape=box]
> subset807 [label="rel#807:RelSubset#27.NONE.any.[2
> ASC-nulls-first]"]
> subset815 [label="rel#815:RelSubset#27.NONE.any.[]"]
> subset816 [label="rel#816:RelSubset#27.LOGICAL.any.[2
> ASC-nulls-first]",color=red]
> subset819 [label="rel#819:RelSubset#27.LOGICAL.any.[]"]
> subset819 -> subset816; subset815 -> subset807; }
> subgraph cluster28{
> label="Set 28 RecordType(INTEGER a, BIGINT b,
> VARCHAR(2147483647) c, VARCHAR(2147483647) d)";
> rel808
> [label="rel#808:LogicalSink\ninput=RelSubset#807,table=default_catalog.default_database.sink2,fields=a,
> b, c, d\nrows=1.0E8, cost={inf}",shape=box]
> rel811
> [label="rel#811:AbstractConverter\ninput=RelSubset#809,convention=LOGICAL,FlinkRelDistributionTraitDef=any,sort=[]\nrows=1.0E8,
> cost={inf}",shape=box]
> rel817
> [label="rel#817:FlinkLogicalSink\ninput=RelSubset#816,table=default_catalog.default_database.sink2,fields=a,
> b, c, d\nrows=1.0E8, cost={inf}",shape=box]
> subset809 [label="rel#809:RelSubset#28.NONE.any.[]"]
> subset810 [label="rel#810:RelSubset#28.LOGICAL.any.[]"]
> }
> root -> subset810;
> subset805 -> rel162;
> subset813 -> rel812[color=blue];
> subset807 -> rel806; rel806 -> subset805;
> subset815 -> rel814; rel814 -> subset805;
> subset819 -> rel818[color=blue]; rel818 -> subset813[color=blue];
> subset809 -> rel808; rel808 -> subset807;
> subset810 -> rel811; rel811 -> subset809;
> subset810 -> rel817; rel817 -> subset816;
> }
> at
> org.apache.calcite.plan.volcano.RelSubset$CheapestPlanReplacer.visit(RelSubset.java:742)
> at
> org.apache.calcite.plan.volcano.RelSubset.buildCheapestPlan(RelSubset.java:365)
> at
> org.apache.calcite.plan.volcano.VolcanoPlanner.findBestExp(VolcanoPlanner.java:520)
> at
> org.apache.calcite.tools.Programs$RuleSetProgram.run(Programs.java:312)
> at
> org.apache.flink.table.planner.plan.optimize.program.FlinkVolcanoProgram.optimize(FlinkVolcanoProgram.scala:69)
> ... 55 more
> {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
