hologres-connector-flink-1.13 是不是有点问题，连接开源flink，可以使holo表插入数据，那边holo也可以查询新增数据，但是从holo那边查询就是一直报错:

[hecheng64]

Flink SQL> create TEMPORARY table holo_source(

id INTEGER, item INTEGER ) with ( 'connector'='hologres', 'dbname'='dps_test', 'tablename'='flink_test', 'username'='', 'password'='', 'endpoint'='hgpostcn-cn-2r42go4bp00c-cn-hangzhou.hologres.aliyuncs.com:80' ); [INFO] Execute statement succeed.

Flink SQL> select id,item from holo_source; [ERROR] Could not execute SQL statement. Reason: org.apache.calcite.plan.RelOptPlanner$CannotPlanException: There are not enough rules to produce a node with desired properties: convention=BATCH_PHYSICAL, FlinkRelDistributionTraitDef=any, sort=[]. Missing conversion is FlinkLogicalTableSourceScan[convention: LOGICAL -> BATCH_PHYSICAL] There is 1 empty subset: rel#133:RelSubset#2.BATCH_PHYSICAL.any.[], the relevant part of the original plan is as follows 121:FlinkLogicalTableSourceScan(table=[[default_catalog, default_database, holo_source]], fields=[id, item])

Root: rel#131:RelSubset#3.BATCH_PHYSICAL.any.[] Original rel: FlinkLogicalSink(subset=[rel#119:RelSubset#1.LOGICAL.any.[]], table=[default_catalog.default_database.Unregistered_Collect_Sink_1], fields=[id, item]): rowcount = 1.0E8, cumulative cost = {1.0E8 rows, 1.0E8 cpu, 0.0 io, 0.0 network, 0.0 memory}, id = 123 FlinkLogicalTableSourceScan(subset=[rel#122:RelSubset#0.LOGICAL.any.[]], table=[[default_catalog, default_database, holo_source]], fields=[id, item]): rowcount = 1.0E8, cumulative cost = {1.0E8 rows, 1.0E8 cpu, 8.0E8 io, 0.0 network, 0.0 memory}, id = 121

Sets: Set#2, type: RecordType(INTEGER id, INTEGER item) rel#128:RelSubset#2.LOGICAL.any.[], best=rel#121 rel#121:FlinkLogicalTableSourceScan.LOGICAL.any.[](table=[default_catalog, default_database, holo_source],fields=id, item), rowcount=1.0E8, cumulative cost={1.0E8 rows, 1.0E8 cpu, 8.0E8 io, 0.0 network, 0.0 memory} rel#133:RelSubset#2.BATCH_PHYSICAL.any.[], best=null Set#3, type: RecordType(INTEGER id, INTEGER item) rel#130:RelSubset#3.LOGICAL.any.[], best=rel#129 rel#129:FlinkLogicalSink.LOGICAL.any.[](input=RelSubset#128,table=default_catalog.default_database.Unregistered_Collect_Sink_1,fields=id, item), rowcount=1.0E8, cumulative cost={2.0E8 rows, 2.0E8 cpu, 8.0E8 io, 0.0 network, 0.0 memory} rel#131:RelSubset#3.BATCH_PHYSICAL.any.[], best=null rel#132:AbstractConverter.BATCH_PHYSICAL.any., rowcount=1.0E8, cumulative cost={inf} rel#134:BatchPhysicalSink.BATCH_PHYSICAL.any.[](input=RelSubset#133,table=default_catalog.default_database.Unregistered_Collect_Sink_1,fields=id, item), rowcount=1.0E8, cumulative cost={inf}

Graphviz: digraph G { root [style=filled,label="Root"]; subgraph cluster2{ label="Set 2 RecordType(INTEGER id, INTEGER item)"; rel121 [label="rel#121:FlinkLogicalTableSourceScan\ntable=[default_catalog, default_database, holo_source],fields=id, item\nrows=1.0E8, cost={1.0E8 rows, 1.0E8 cpu, 8.0E8 io, 0.0 network, 0.0 memory}",color=blue,shape=box] subset128 [label="rel#128:RelSubset#2.LOGICAL.any.[]"] subset133 [label="rel#133:RelSubset#2.BATCH_PHYSICAL.any.[]",color=red] } subgraph cluster3{ label="Set 3 RecordType(INTEGER id, INTEGER item)"; rel129 [label="rel#129:FlinkLogicalSink\ninput=RelSubset#128,table=default_catalog.default_database.Unregistered_Collect_Sink_1,fields=id, item\nrows=1.0E8, cost={2.0E8 rows, 2.0E8 cpu, 8.0E8 io, 0.0 network, 0.0 memory}",color=blue,shape=box] rel132 [label="rel#132:AbstractConverter\ninput=RelSubset#130,convention=BATCH_PHYSICAL,FlinkRelDistributionTraitDef=any,sort=[]\nrows=1.0E8, cost={inf}",shape=box] rel134 [label="rel#134:BatchPhysicalSink\ninput=RelSubset#133,table=default_catalog.default_database.Unregistered_Collect_Sink_1,fields=id, item\nrows=1.0E8, cost={inf}",shape=box] subset130 [label="rel#130:RelSubset#3.LOGICAL.any.[]"] subset131 [label="rel#131:RelSubset#3.BATCH_PHYSICAL.any.[]"] } root -> subset131; subset128 -> rel121[color=blue]; subset130 -> rel129[color=blue]; rel129 -> subset128[color=blue]; subset131 -> rel132; rel132 -> subset130; subset131 -> rel134; rel134 -> subset133; }

[huafengw]

@hecheng64 现在hologres-connector还不支持作为Flink的Source