[jira] [Updated] (FLINK-39804) Nested projection fields not pushed down to source

["ASF GitHub Bot (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/FLINK-39804?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

ASF GitHub Bot updated FLINK-39804:
-----------------------------------
    Labels: pull-request-available  (was: )

> Nested projection fields not pushed down to source
> --------------------------------------------------
>
>                 Key: FLINK-39804
>                 URL: https://issues.apache.org/jira/browse/FLINK-39804
>             Project: Flink
>          Issue Type: Bug
>          Components: Table SQL / Planner
>            Reporter: Daniel Rossos
>            Priority: Major
>              Labels: pull-request-available
>
> h1. Summary
> If you have a source that supports nested-projection-pushdown, if there is a non-pushed-down filter between the projection and source scan, only the top-level columns of your projection will be pushed down into the source.
> Highlevel e.g 
> {{SQL Query Example:}}
> {{SELECT a.aOne, b FROM t1 WHERE <unpushable Filter> }}
> Results in following Optimized Execution Plan:
> {{== Optimized Execution Plan ==}}
> {{Calc(select=[a.aOne, b], where=[<filter>])}}
> {{+- TableSourceScan(table=[[c, db, t1, filter=[], project=[a, b]]], fields=[a, b])}}
> h1. Detailed Trace
> As part of the [PROJECT_RULES|https://github.com/apache/flink/blob/master/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala#L212] in our query optimizer, we utilize the Calcite rule CoreRules.PROJECT_FILTER_TRANSPOSE. To [summarize|https://github.com/apache/calcite/blob/main/core/src/main/java/org/apache/calcite/rel/rules/ProjectFilterTransposeRule.java#L92], this rule will check to see if we have a PROJECTION above a FILTER node in our graph, if we do it will attempt to 'transpose' the projection node making it so the structure is that the Projection node now occurs "below" the filtering. This then sets up the structure to allow us for this projection to be pushed down into our source-scan. The problem is that in the default PROJECT_FILTER_TRANSPOSE mode of this transposition rule it will utilize a [PushProjector|https://github.com/apache/calcite/blob/main/core/src/main/java/org/apache/calcite/rel/rules/ProjectFilterTransposeRule.java#L176] to traverse the Projection that down the chain eventually uses visitInputRef(), which will only record the top-level index of the column in the projection. 
> h1. Potential Solution / Investigation
> In our testing, we found that by changing the projection rule to PROJECT_FILTER_TRANSPOSE_WHOLE_EXPRESSIONS or PROJECT_FILTER_TRANSPOSE_WHOLE_PROJECT_EXPRESSIONS in the [PROJECT_REWRITE|https://github.com/apache/flink/blob/master/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkStreamProgram.scala#L253] rule-set* we were able to avoid this issue because these variants preserve whole project/filter expressions instead.
>  
> *Note, I specify PROJECT_REWRITE specifically, because if we use the PROJECT_FILTER_TRANSPOSE_WHOLE_EXPRESSIONS transpose rule in the Volcano LOGICAL phase we get an infinite oscillation problem because PROJECT_FILTER_TRANSPOSE_WHOLE_EXPRESSIONS pushes Project below Filter, FILTER_PROJECT_TRANSPOSE can pull Filter back below/above that Project, 
>  and PROJECT_MERGE collapses the resulting projects back into the original shape, making the same rewrite applicable again. In our testing we left the transpose rule in LOGICAL phase as the current default, but still analyzing implications here.
> Still investigating further, but interested to hear the opinion / perspective from someone with more understanding on the Calcite planner / Flink table optimizations work. 



--
This message was sent by Atlassian Jira
(v8.20.10#820010)