Flink SQL的灵异事件----查询语句中增加表中的某个字段时就没法正常查询了。

[jy l]

Hi：
FlinkSQL我在使用时发生一件很诡异的事件。具体如下：

我的DDL：
create table if not exists t_order(
id int PRIMARY KEY comment '订单id',
timestamps bigint comment '订单创建时间',
orderInformationId string comment '订单信息ID',
userId string comment '用户ID',
categoryId int comment '商品类别',
productId int comment '商品ID',
price decimal(10,2) comment '单价',
productCount int comment '购买数量',
priceSum decimal(10,2) comment '订单总价',
shipAddress string comment '商家地址',
receiverAddress string comment '收货地址',
ts AS TO_TIMESTAMP(FROM_UNIXTIME(timestamps/1000)),
WATERMARK FOR ts AS ts - INTERVAL '3' SECOND
)with(
'connector' = 'kafka',
'format' = 'debezium-avro-confluent',
'debezium-avro-confluent.schema-registry.url' = 'http://手动打码ip:8081
<<a href="http://xn--ip:8081-lm1l957ilcaw623a">http://xn--ip:8081-lm1l957ilcaw623a>',
'topic' = 'ods.userAnalysis.order',
'properties.bootstrap.servers' = '手动打码ip:9092',
'properties.group.id' = 'flink-analysis',
'scan.startup.mode' = 'latest-offset'
)

我在查询该表时，使用如下查询语句能够正常查询出来：

   - select * from t_order
   - select receiverAddress from t_order
   - select
   id,
   timestamps,
   orderInformationId,
   userId,
   categoryId,
   productId,
   price,
   productCount,
   priceSum,
   shipAddress
   from t_order

但是我在第三条语句中加上receiverAddress字段时，就查询不出来了，sql如下：
select
id,
timestamps,
orderInformationId,
userId,
categoryId,
productId,
price,
productCount,
priceSum,
shipAddress,
receiverAddress
from t_order,
报错信息如下：
Exception in thread "main" org.apache.flink.table.api.TableException: This
calc has no useful projection and no filter. It should be removed by
CalcRemoveRule.
at
org.apache.flink.table.planner.codegen.CalcCodeGenerator$.generateProcessCode(CalcCodeGenerator.scala:166)
at
org.apache.flink.table.planner.codegen.CalcCodeGenerator$.generateCalcOperator(CalcCodeGenerator.scala:59)
at
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecCalc.translateToPlanInternal(StreamExecCalc.scala:84)
at
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecCalc.translateToPlanInternal(StreamExecCalc.scala:39)
at
org.apache.flink.table.planner.plan.nodes.exec.ExecNode.translateToPlan(ExecNode.scala:59)
at
org.apache.flink.table.planner.plan.nodes.exec.ExecNode.translateToPlan$(ExecNode.scala:57)
at
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecCalcBase.translateToPlan(StreamExecCalcBase.scala:38)
at
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecLegacySink.translateToTransformation(StreamExecLegacySink.scala:158)
at
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecLegacySink.translateToPlanInternal(StreamExecLegacySink.scala:82)
at
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecLegacySink.translateToPlanInternal(StreamExecLegacySink.scala:48)
at
org.apache.flink.table.planner.plan.nodes.exec.ExecNode.translateToPlan(ExecNode.scala:59)
...

receiverAddress这个字段明明是在我的DDL中具体申明了的，且单独查询也能出来。
这具体是什么原因呢？望各位大佬告知。


祝好！

[Jark]

看报错像是一个 bug。 请问使用的是哪个版本呢？
可以去 JIRA issue 提个 issue。

Best,
Jark

On Tue, 24 Nov 2020 at 11:27, jy l <[hidden email]> wrote:

> Hi：
> FlinkSQL我在使用时发生一件很诡异的事件。具体如下：
>
> 我的DDL：
> create table if not exists t_order(
> id int PRIMARY KEY comment '订单id',
> timestamps bigint comment '订单创建时间',
> orderInformationId string comment '订单信息ID',
> userId string comment '用户ID',
> categoryId int comment '商品类别',
> productId int comment '商品ID',
> price decimal(10,2) comment '单价',
> productCount int comment '购买数量',
> priceSum decimal(10,2) comment '订单总价',
> shipAddress string comment '商家地址',
> receiverAddress string comment '收货地址',
> ts AS TO_TIMESTAMP(FROM_UNIXTIME(timestamps/1000)),
> WATERMARK FOR ts AS ts - INTERVAL '3' SECOND
> )with(
> 'connector' = 'kafka',
> 'format' = 'debezium-avro-confluent',
> 'debezium-avro-confluent.schema-registry.url' = 'http://手动打码ip:8081
> <<a href="http://xn--ip:8081-lm1l957ilcaw623a">http://xn--ip:8081-lm1l957ilcaw623a>
> <<a href="http://xn--ip:8081-lm1l957ilcaw623a">http://xn--ip:8081-lm1l957ilcaw623a>',
> 'topic' = 'ods.userAnalysis.order',
> 'properties.bootstrap.servers' = '手动打码ip:9092',
> 'properties.group.id' = 'flink-analysis',
> 'scan.startup.mode' = 'latest-offset'
> )
>
> 我在查询该表时，使用如下查询语句能够正常查询出来：
>
>    - select * from t_order
>    - select receiverAddress from t_order
>    - select
>    id,
>    timestamps,
>    orderInformationId,
>    userId,
>    categoryId,
>    productId,
>    price,
>    productCount,
>    priceSum,
>    shipAddress
>    from t_order
>
> 但是我在第三条语句中加上receiverAddress字段时，就查询不出来了，sql如下：
> select
> id,
> timestamps,
> orderInformationId,
> userId,
> categoryId,
> productId,
> price,
> productCount,
> priceSum,
> shipAddress,
> receiverAddress
> from t_order,
> 报错信息如下：
> Exception in thread "main" org.apache.flink.table.api.TableException: This
> calc has no useful projection and no filter. It should be removed by
> CalcRemoveRule.
> at
>
> org.apache.flink.table.planner.codegen.CalcCodeGenerator$.generateProcessCode(CalcCodeGenerator.scala:166)
> at
>
> org.apache.flink.table.planner.codegen.CalcCodeGenerator$.generateCalcOperator(CalcCodeGenerator.scala:59)
> at
>
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecCalc.translateToPlanInternal(StreamExecCalc.scala:84)
> at
>
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecCalc.translateToPlanInternal(StreamExecCalc.scala:39)
> at
>
> org.apache.flink.table.planner.plan.nodes.exec.ExecNode.translateToPlan(ExecNode.scala:59)
> at
>
> org.apache.flink.table.planner.plan.nodes.exec.ExecNode.translateToPlan$(ExecNode.scala:57)
> at
>
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecCalcBase.translateToPlan(StreamExecCalcBase.scala:38)
> at
>
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecLegacySink.translateToTransformation(StreamExecLegacySink.scala:158)
> at
>
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecLegacySink.translateToPlanInternal(StreamExecLegacySink.scala:82)
> at
>
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecLegacySink.translateToPlanInternal(StreamExecLegacySink.scala:48)
> at
>
> org.apache.flink.table.planner.plan.nodes.exec.ExecNode.translateToPlan(ExecNode.scala:59)
> ...
>
> receiverAddress这个字段明明是在我的DDL中具体申明了的，且单独查询也能出来。
> 这具体是什么原因呢？望各位大佬告知。
>
>
> 祝好！
>

[jy l]

我使用的是release-1.12.0-rc1

Best

Jark Wu <[hidden email]> 于2020年11月24日周二 上午11:42写道：

> 看报错像是一个 bug。 请问使用的是哪个版本呢？
> 可以去 JIRA issue 提个 issue。
>
> Best,
> Jark
>
> On Tue, 24 Nov 2020 at 11:27, jy l <[hidden email]> wrote:
>
> > Hi：
> > FlinkSQL我在使用时发生一件很诡异的事件。具体如下：
> >
> > 我的DDL：
> > create table if not exists t_order(
> > id int PRIMARY KEY comment '订单id',
> > timestamps bigint comment '订单创建时间',
> > orderInformationId string comment '订单信息ID',
> > userId string comment '用户ID',
> > categoryId int comment '商品类别',
> > productId int comment '商品ID',
> > price decimal(10,2) comment '单价',
> > productCount int comment '购买数量',
> > priceSum decimal(10,2) comment '订单总价',
> > shipAddress string comment '商家地址',
> > receiverAddress string comment '收货地址',
> > ts AS TO_TIMESTAMP(FROM_UNIXTIME(timestamps/1000)),
> > WATERMARK FOR ts AS ts - INTERVAL '3' SECOND
> > )with(
> > 'connector' = 'kafka',
> > 'format' = 'debezium-avro-confluent',
> > 'debezium-avro-confluent.schema-registry.url' = 'http://手动打码ip:8081
> <<a href="http://xn--ip:8081-lm1l957ilcaw623a">http://xn--ip:8081-lm1l957ilcaw623a>
> > <<a href="http://xn--ip:8081-lm1l957ilcaw623a">http://xn--ip:8081-lm1l957ilcaw623a>
> > <<a href="http://xn--ip:8081-lm1l957ilcaw623a">http://xn--ip:8081-lm1l957ilcaw623a>',
> > 'topic' = 'ods.userAnalysis.order',
> > 'properties.bootstrap.servers' = '手动打码ip:9092',
> > 'properties.group.id' = 'flink-analysis',
> > 'scan.startup.mode' = 'latest-offset'
> > )
> >
> > 我在查询该表时，使用如下查询语句能够正常查询出来：
> >
> >    - select * from t_order
> >    - select receiverAddress from t_order
> >    - select
> >    id,
> >    timestamps,
> >    orderInformationId,
> >    userId,
> >    categoryId,
> >    productId,
> >    price,
> >    productCount,
> >    priceSum,
> >    shipAddress
> >    from t_order
> >
> > 但是我在第三条语句中加上receiverAddress字段时，就查询不出来了，sql如下：
> > select
> > id,
> > timestamps,
> > orderInformationId,
> > userId,
> > categoryId,
> > productId,
> > price,
> > productCount,
> > priceSum,
> > shipAddress,
> > receiverAddress
> > from t_order,
> > 报错信息如下：
> > Exception in thread "main" org.apache.flink.table.api.TableException:
> This
> > calc has no useful projection and no filter. It should be removed by
> > CalcRemoveRule.
> > at
> >
> >
> org.apache.flink.table.planner.codegen.CalcCodeGenerator$.generateProcessCode(CalcCodeGenerator.scala:166)
> > at
> >
> >
> org.apache.flink.table.planner.codegen.CalcCodeGenerator$.generateCalcOperator(CalcCodeGenerator.scala:59)
> > at
> >
> >
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecCalc.translateToPlanInternal(StreamExecCalc.scala:84)
> > at
> >
> >
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecCalc.translateToPlanInternal(StreamExecCalc.scala:39)
> > at
> >
> >
> org.apache.flink.table.planner.plan.nodes.exec.ExecNode.translateToPlan(ExecNode.scala:59)
> > at
> >
> >
> org.apache.flink.table.planner.plan.nodes.exec.ExecNode.translateToPlan$(ExecNode.scala:57)
> > at
> >
> >
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecCalcBase.translateToPlan(StreamExecCalcBase.scala:38)
> > at
> >
> >
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecLegacySink.translateToTransformation(StreamExecLegacySink.scala:158)
> > at
> >
> >
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecLegacySink.translateToPlanInternal(StreamExecLegacySink.scala:82)
> > at
> >
> >
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecLegacySink.translateToPlanInternal(StreamExecLegacySink.scala:48)
> > at
> >
> >
> org.apache.flink.table.planner.plan.nodes.exec.ExecNode.translateToPlan(ExecNode.scala:59)
> > ...
> >
> > receiverAddress这个字段明明是在我的DDL中具体申明了的，且单独查询也能出来。
> > 这具体是什么原因呢？望各位大佬告知。
> >
> >
> > 祝好！
> >
>