Reduce duplicate Projection when PARQUET_PUSHDOWN_FILTERS is on.

[Ted-Jiang]

Is your feature request related to a problem or challenge? Please describe what you are trying to do.

❯ explain  select l_orderkey from lineitem where l_extendedprice='2618.76';
+---------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type     | plan                                                                                                                                                                                                                                                                      |
+---------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| logical_plan  | Projection: lineitem.l_orderkey                                                                                                                                                                                                                                           |
|               |   Filter: CAST(lineitem.l_extendedprice AS Utf8) = Utf8("2618.76")                                                                                                                                                                                                        |
|               |     TableScan: lineitem projection=[l_orderkey, l_extendedprice], partial_filters=[CAST(lineitem.l_extendedprice AS Utf8) = Utf8("2618.76")]                                                                                                                              |
| physical_plan | ProjectionExec: expr=[l_orderkey@0 as l_orderkey]                                                                                                                                                                                                                         |
|               |   CoalesceBatchesExec: target_batch_size=4096                                                                                                                                                                                                                             |
|               |     FilterExec: CAST(l_extendedprice@1 AS Utf8) = 2618.76                                                                                                                                                                                                                 |
|               |       RepartitionExec: partitioning=RoundRobinBatch(16)                                                                                                                                                                                                                   |
|               |         ParquetExec: limit=None, partitions=*.parquet], predicate=CAST(l_extendedprice AS Utf8) = Utf8("2618.76"), projection=[l_orderkey, l_extendedprice] |
|               |                                                                                                                                                                                                                                                                           |
+---------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
2 rows in set. Query took 0.017 seconds.

In this sql will have two related cols:
1. Selected col l_orderkey
2. Filter col l_extendedprice

IMO, this situation where Filter Col not exists in Selected col, is perfect for row_filter_push_down. We can only visit the Filter Col once, without using the selection do the decoding thing.

Describe the solution you'd like

Maybe we can change the physical plan to

| physical_plan | ProjectionExec: expr=[l_orderkey@0 as l_orderkey]                                                                                                                                                                                                                         |
|               |   CoalesceBatchesExec: target_batch_size=4096                                                                                                                                                                                                                             |                                                                                                                                                                                                             |
|               |       RepartitionExec: partitioning=RoundRobinBatch(16)                                                                                                                                                                                                                   |
|               |         ParquetExec: limit=None, partitions=[*.parquet], predicate=CAST(l_extendedprice AS Utf8) = Utf8("2618.76"), projection=[l_orderkey] |
|               |

Describe alternatives you've considered
A clear and concise description of any alternative solutions or features you've considered.

Additional context
Add any other context or screenshots about the feature request here.

[Ted-Jiang]

@alamb @tustvold @thinkharderdev Look forward to hearing your opinion😄

[tustvold]

I think this relates to #4020, which appears to have been closed?!

[thinkharderdev]

So if I understand correctly, you are saying we should push down the predicate so that we only used it to generate a row selection for the other rows (i.e. we do not buffer the predicate column in memory after we have generated the selection)?

[thinkharderdev]

Would this be accomplished by simply making the filter pushdown exact? Or is there something else required?

[tustvold]

Ah yes, I misread the ticket #4028 may resolve this, although it might need an additional optimisation pass to remove the now-redundant projection

[Ted-Jiang]

we do not buffer the predicate column in memory after we have generated the selection

Yes just like you mentioned it is an exact filter.

Would this be accomplished by simply making the filter pushdown exact?

Sounds like a reasonable solution，seems miss this part in row filter.

[alamb]

If the parquet pushdown filtering is indeed exact, returning exact sounds like a good idea to me. RowGroup pruning and page index pruning are definitely not exact.

[thinkharderdev]

Ah yes, I misread the ticket #4028 may resolve this, although it might need an additional optimisation pass to remove the now-redundant projection

I think it should work as is. As long and ProjectionPushdown runs after FilterPushdown then the projection should be eliminated entirely I think.