Add Logical filter on a query plan from Flink Table API

[Felipe Gutierrez <fe...@gmail.com>]

Hi,

I am a newbie in Apache Calcite. I am trying to use it with Apache Flink.
To start I am trying to create a HelloWorld which just add a logical filter
on my query.
1 - I have my Flink app using Table API [1].
2 - I have created my Calcite filter rule which is applied to my FLink
query if I use CalciteConfig cc = new
CalciteConfigBuilder().addLogicalOptRuleSet(RuleSets.ofList(MyFilterRule.INSTANCE)).build()
[2];
3 - The debug thread only goes to my rule if there is a filter on my query.

I would like to create a logical filter if there is no filter set on the
logical query. How should I implement it?
I see my LogicalFilter been created when I call "tableEnv.explain()"
method. I suppose that I can add some logical filters on the plan.

== Abstract Syntax Tree ==
LogicalFilter(condition=[>=($6, 50)])
  LogicalTableScan(table=[[TicketsStation01Plat01]])

== Optimized Logical Plan ==
DataStreamCalc(select=[sensorId, sensorType, platformId, platformType,
stationId, timestamp, value, trip, eventTime], where=[>=(value, 50)])
  StreamTableSourceScan(table=[[TicketsStation01Plat01]], fields=[sensorId,
sensorType, platformId, platformType, stationId, timestamp, value, trip,
eventTime], source=[SensorTuples])

== Physical Execution Plan ==
....

Thanks,
Felipe

[1]
https://github.com/felipegutierrez/explore-flink/blob/master/src/main/java/org/sense/flink/examples/stream/table/HelloWorldCalcitePlanTableAPI.java#L62
[2]
https://github.com/felipegutierrez/explore-flink/blob/master/src/main/java/org/sense/calcite/rules/MyFilterRule.java#L14
*--*
*-- Felipe Gutierrez*

*-- skype: felipe.o.gutierrez*
*--* *https://felipeogutierrez.blogspot.com
<https://felipeogutierrez.blogspot.com>*

Re: Add Logical filter on a query plan from Flink Table API

[Stamatis Zampetakis <za...@gmail.com>]

Hi Felipe,

If you want to build a plan (add a logical filter in the query) then
instead of using rules the most straightforward way would be to use the
RelBuilder [1] API of Calcite or the Table [2] API of Flink as you
mentioned.
On the other hand if you want to do things where you need information about
the whole plan (e.g., the plan must not contain a filter) then the visitor
approach seems more adequate [3, 4].
Creating rules which match anything (i.e., the whole plan) are not that
frequent.

Best,
Stamatis

[1]
https://github.com/apache/calcite/blob/master/core/src/main/java/org/apache/calcite/tools/RelBuilder.java
[2]
https://github.com/apache/flink/blob/master/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Table.java
[3]
https://github.com/apache/calcite/blob/e8d598a434e8dbadaf756f8c57c748f4d7e16fdf/core/src/main/java/org/apache/calcite/rel/RelNode.java#L433
[4]
https://github.com/apache/calcite/blob/e8d598a434e8dbadaf756f8c57c748f4d7e16fdf/core/src/main/java/org/apache/calcite/rel/RelNode.java#L247
On Mon, Jul 8, 2019 at 3:27 PM Felipe Gutierrez <
felipe.o.gutierrez@gmail.com> wrote:

> Hi,
>
> I am a newbie in Apache Calcite. I am trying to use it with Apache Flink.
> To start I am trying to create a HelloWorld which just add a logical filter
> on my query.
> 1 - I have my Flink app using Table API [1].
> 2 - I have created my Calcite filter rule which is applied to my FLink
> query if I use CalciteConfig cc = new
>
> CalciteConfigBuilder().addLogicalOptRuleSet(RuleSets.ofList(MyFilterRule.INSTANCE)).build()
> [2];
> 3 - The debug thread only goes to my rule if there is a filter on my query.
>
> I would like to create a logical filter if there is no filter set on the
> logical query. How should I implement it?
> I see my LogicalFilter been created when I call "tableEnv.explain()"
> method. I suppose that I can add some logical filters on the plan.
>
> == Abstract Syntax Tree ==
> LogicalFilter(condition=[>=($6, 50)])
>   LogicalTableScan(table=[[TicketsStation01Plat01]])
>
> == Optimized Logical Plan ==
> DataStreamCalc(select=[sensorId, sensorType, platformId, platformType,
> stationId, timestamp, value, trip, eventTime], where=[>=(value, 50)])
>   StreamTableSourceScan(table=[[TicketsStation01Plat01]], fields=[sensorId,
> sensorType, platformId, platformType, stationId, timestamp, value, trip,
> eventTime], source=[SensorTuples])
>
> == Physical Execution Plan ==
> ....
>
> Thanks,
> Felipe
>
> [1]
>
> https://github.com/felipegutierrez/explore-flink/blob/master/src/main/java/org/sense/flink/examples/stream/table/HelloWorldCalcitePlanTableAPI.java#L62
> [2]
>
> https://github.com/felipegutierrez/explore-flink/blob/master/src/main/java/org/sense/calcite/rules/MyFilterRule.java#L14
> *--*
> *-- Felipe Gutierrez*
>
> *-- skype: felipe.o.gutierrez*
> *--* *https://felipeogutierrez.blogspot.com
> <https://felipeogutierrez.blogspot.com>*
>