[FLINK-22334] Fail to translate the hive-sql in STREAMING mode

• Type: Bug
• Status: Open
• Priority: Minor
• Resolution: Unresolved
• Affects Version/s: 1.13.0
• Fix Version/s: 1.20.0
• Component/s: Connectors / Hive
• Labels: auto-deprioritized-major
• Assignee: Unassigned
• Reporter: Shengkai Fang
• Created: 19/Apr/21 01:49
• Updated: 11/Mar/24 12:44

Description

Please run in the streaming mode.

The failed statement 

// Some comments here
insert into dest(y,x) select x,y from foo cluster by x

Exception stack:

org.apache.calcite.plan.RelOptPlanner$CannotPlanException: There are not enough rules to produce a node with desired properties: convention=LOGICAL, FlinkRelDistributionTraitDef=any, MiniBatchIntervalTraitDef=None: 0, ModifyKindSetTraitDef=[NONE], UpdateKindTraitDef=[NONE].
Missing conversion is LogicalDistribution[convention: NONE -> LOGICAL]
There is 1 empty subset: rel#5176:RelSubset#43.LOGICAL.any.None: 0.[NONE].[NONE], the relevant part of the original plan is as follows
5174:LogicalDistribution(collation=[[0 ASC-nulls-first]], dist=[[]])
  5172:LogicalProject(subset=[rel#5173:RelSubset#42.NONE.any.None: 0.[NONE].[NONE]], x=[$0])
    5106:LogicalTableScan(subset=[rel#5171:RelSubset#41.NONE.any.None: 0.[NONE].[NONE]], table=[[myhive, default, foo]])

Root: rel#5176:RelSubset#43.LOGICAL.any.None: 0.[NONE].[NONE]
Original rel:
FlinkLogicalLegacySink(subset=[rel#4254:RelSubset#8.LOGICAL.any.None: 0.[NONE].[NONE]], name=[collect], fields=[_o__c0]): rowcount = 1.0E8, cumulative cost = {1.0E8 rows, 1.0E8 cpu, 0.0 io, 0.0 network, 0.0 memory}, id = 4276
  FlinkLogicalCalc(subset=[rel#4275:RelSubset#7.LOGICAL.any.None: 0.[NONE].[NONE]], select=[CASE(IS NULL($f1), 0:BIGINT, $f1) AS _o__c0]): rowcount = 1.0E8, cumulative cost = {1.0E8 rows, 1.0E8 cpu, 0.0 io, 0.0 network, 0.0 memory}, id = 4288
    FlinkLogicalJoin(subset=[rel#4272:RelSubset#6.LOGICAL.any.None: 0.[NONE].[NONE]], condition=[=($0, $1)], joinType=[left]): rowcount = 1.0E8, cumulative cost = {1.0E8 rows, 1.0856463237676364E8 cpu, 4.0856463237676364E8 io, 0.0 network, 0.0 memory}, id = 4271
      FlinkLogicalTableSourceScan(subset=[rel#4270:RelSubset#1.LOGICAL.any.None: 0.[NONE].[NONE]], table=[[myhive, default, bar, project=[i]]], fields=[i]): rowcount = 1.0E8, cumulative cost = {1.0E8 rows, 1.0E8 cpu, 4.0E8 io, 0.0 network, 0.0 memory}, id = 4279
      FlinkLogicalAggregate(subset=[rel#4268:RelSubset#5.LOGICAL.any.None: 0.[NONE].[NONE]], group=[{1}], agg#0=[COUNT($0)]): rowcount = 8564632.376763644, cumulative cost = {9.0E7 rows, 1.89E8 cpu, 7.2E8 io, 0.0 network, 0.0 memory}, id = 4286
        FlinkLogicalCalc(subset=[rel#4283:RelSubset#3.LOGICAL.any.None: 0.[NONE].[NONE]], select=[x, y], where=[IS NOT NULL(y)]): rowcount = 9.0E7, cumulative cost = {9.0E7 rows, 0.0 cpu, 0.0 io, 0.0 network, 0.0 memory}, id = 4282
          FlinkLogicalTableSourceScan(subset=[rel#4262:RelSubset#2.LOGICAL.any.None: 0.[NONE].[NONE]], table=[[myhive, default, foo]], fields=[x, y]): rowcount = 1.0E8, cumulative cost = {1.0E8 rows, 1.0E8 cpu, 8.0E8 io, 0.0 network, 0.0 memory}, id = 4261

Sets:
Set#41, type: RecordType(INTEGER x, INTEGER y)
	rel#5171:RelSubset#41.NONE.any.None: 0.[NONE].[NONE], best=null
		rel#5106:LogicalTableScan.NONE.any.None: 0.[NONE].[NONE](table=[myhive, default, foo]), rowcount=1.0E8, cumulative cost={inf}
	rel#5179:RelSubset#41.LOGICAL.any.None: 0.[NONE].[NONE], best=rel#5178
		rel#5178:FlinkLogicalTableSourceScan.LOGICAL.any.None: 0.[NONE].[NONE](table=[myhive, default, foo],fields=x, y), rowcount=1.0E8, cumulative cost={1.0E8 rows, 1.0E8 cpu, 8.0E8 io, 0.0 network, 0.0 memory}
Set#42, type: RecordType(INTEGER x)
	rel#5173:RelSubset#42.NONE.any.None: 0.[NONE].[NONE], best=null
		rel#5172:LogicalProject.NONE.any.None: 0.[NONE].[NONE](input=RelSubset#5171,inputs=0), rowcount=1.0E8, cumulative cost={inf}
		rel#5180:LogicalTableScan.NONE.any.None: 0.[NONE].[NONE](table=[myhive, default, foo, project=[x]]), rowcount=1.0E8, cumulative cost={inf}
		rel#5182:LogicalCalc.NONE.any.None: 0.[NONE].[NONE](input=RelSubset#5171,expr#0..1={inputs},0=$t0), rowcount=1.0E8, cumulative cost={inf}
	rel#5184:RelSubset#42.LOGICAL.any.None: 0.[NONE].[NONE], best=rel#5183
		rel#5183:FlinkLogicalTableSourceScan.LOGICAL.any.None: 0.[NONE].[NONE](table=[myhive, default, foo, project=[x]],fields=x), rowcount=1.0E8, cumulative cost={1.0E8 rows, 1.0E8 cpu, 4.0E8 io, 0.0 network, 0.0 memory}
		rel#5185:FlinkLogicalCalc.LOGICAL.any.None: 0.[NONE].[NONE](input=RelSubset#5179,select=x), rowcount=1.0E8, cumulative cost={2.0E8 rows, 1.0E8 cpu, 8.0E8 io, 0.0 network, 0.0 memory}
Set#43, type: RecordType(INTEGER x)
	rel#5175:RelSubset#43.NONE.any.None: 0.[NONE].[NONE], best=null
		rel#5174:LogicalDistribution.NONE.any.None: 0.[NONE].[NONE](input=RelSubset#5173,collation=[0 ASC-nulls-first],dist=[]), rowcount=1.0E8, cumulative cost={inf}
	rel#5176:RelSubset#43.LOGICAL.any.None: 0.[NONE].[NONE], best=null
		rel#5177:AbstractConverter.LOGICAL.any.None: 0.[NONE].[NONE](input=RelSubset#5175,convention=LOGICAL,FlinkRelDistributionTraitDef=any,MiniBatchIntervalTraitDef=None: 0,ModifyKindSetTraitDef=[NONE],UpdateKindTraitDef=[NONE]), rowcount=1.0E8, cumulative cost={inf}

Graphviz:
digraph G {
	root [style=filled,label="Root"];
	subgraph cluster41{
		label="Set 41 RecordType(INTEGER x, INTEGER y)";
		rel5106 [label="rel#5106:LogicalTableScan\ntable=[myhive, default, foo]\nrows=1.0E8, cost={inf}",shape=box]
		rel5178 [label="rel#5178:FlinkLogicalTableSourceScan\ntable=[myhive, default, foo],fields=x, y\nrows=1.0E8, cost={1.0E8 rows, 1.0E8 cpu, 8.0E8 io, 0.0 network, 0.0 memory}",color=blue,shape=box]
		subset5171 [label="rel#5171:RelSubset#41.NONE.any.None: 0.[NONE].[NONE]"]
		subset5179 [label="rel#5179:RelSubset#41.LOGICAL.any.None: 0.[NONE].[NONE]"]
	}
	subgraph cluster42{
		label="Set 42 RecordType(INTEGER x)";
		rel5172 [label="rel#5172:LogicalProject\ninput=RelSubset#5171,inputs=0\nrows=1.0E8, cost={inf}",shape=box]
		rel5180 [label="rel#5180:LogicalTableScan\ntable=[myhive, default, foo, project=[x]]\nrows=1.0E8, cost={inf}",shape=box]
		rel5182 [label="rel#5182:LogicalCalc\ninput=RelSubset#5171,expr#0..1={inputs},0=$t0\nrows=1.0E8, cost={inf}",shape=box]
		rel5183 [label="rel#5183:FlinkLogicalTableSourceScan\ntable=[myhive, default, foo, project=[x]],fields=x\nrows=1.0E8, cost={1.0E8 rows, 1.0E8 cpu, 4.0E8 io, 0.0 network, 0.0 memory}",color=blue,shape=box]
		rel5185 [label="rel#5185:FlinkLogicalCalc\ninput=RelSubset#5179,select=x\nrows=1.0E8, cost={2.0E8 rows, 1.0E8 cpu, 8.0E8 io, 0.0 network, 0.0 memory}",shape=box]
		subset5173 [label="rel#5173:RelSubset#42.NONE.any.None: 0.[NONE].[NONE]"]
		subset5184 [label="rel#5184:RelSubset#42.LOGICAL.any.None: 0.[NONE].[NONE]"]
	}
	subgraph cluster43{
		label="Set 43 RecordType(INTEGER x)";
		rel5174 [label="rel#5174:LogicalDistribution\ninput=RelSubset#5173,collation=[0 ASC-nulls-first],dist=[]\nrows=1.0E8, cost={inf}",shape=box]
		rel5177 [label="rel#5177:AbstractConverter\ninput=RelSubset#5175,convention=LOGICAL,FlinkRelDistributionTraitDef=any,MiniBatchIntervalTraitDef=None: 0,ModifyKindSetTraitDef=[NONE],UpdateKindTraitDef=[NONE]\nrows=1.0E8, cost={inf}",shape=box]
		subset5175 [label="rel#5175:RelSubset#43.NONE.any.None: 0.[NONE].[NONE]"]
		subset5176 [label="rel#5176:RelSubset#43.LOGICAL.any.None: 0.[NONE].[NONE]",color=red]
	}
	root -> subset5176;
	subset5171 -> rel5106;
	subset5179 -> rel5178[color=blue];
	subset5173 -> rel5172; rel5172 -> subset5171;
	subset5173 -> rel5180;
	subset5173 -> rel5182; rel5182 -> subset5171;
	subset5184 -> rel5183[color=blue];
	subset5184 -> rel5185; rel5185 -> subset5179;
	subset5175 -> rel5174; rel5174 -> subset5173;
	subset5176 -> rel5177; rel5177 -> subset5175;
}