fix: don't push down volatile predicates in projection

[haohuaijin]

Which issue does this PR close?

related to #7876

Rationale for this change

As @jonahgao said,

Pushing random through the projection can also cause problems.
If it can be fixed, then random will not be pushed further through the aggregatation in this case.

Previously, projection pushed down all predicates. Now, projection pushes down predicates that do not contain volatile functions or contain volatile functions but are not included in projection.

For example,

SELECT t.c1, t.r FROM (SELECT c1, random() AS r FROM test GROUP BY c1) as t WHERE t.r > 0.5 AND random() > 0.8 AND t.c1 > 5;

initial logical plan

+------------------------------------------------------------+---------------------------------------------------------------------------------------------------------------------------+
| initial_logical_plan                                       | Projection: t.c1, t.r                                                                                                     |
|                                                            |   Filter: t.r > Float64(0.5) AND random() > Float64(0.8) AND t.c1 > Int64(5)                                              |
|                                                            |     SubqueryAlias: t                                                                                                      |
|                                                            |       Projection: test.csv.c1, random() AS r                                                                              |
|                                                            |         Aggregate: groupBy=[[test.csv.c1]], aggr=[[]]                                                                     |
|                                                            |           TableScan: test.csv       

before this pr

| logical_plan  | SubqueryAlias: t                                                                                                                                                 |
|               |   Projection: aggregate_test_100.c1, random() AS r                                                                                                               |
|               |     Aggregate: groupBy=[[aggregate_test_100.c1]], aggr=[[]]                                                                                                      |
|               |       Projection: aggregate_test_100.c1                                                                                                                          |
|               |         Filter: random() > Float64(0.5) AND random() > Float64(0.8) AND aggregate_test_100.c1 > Utf8("5")                                                        |
|               |           Projection: random(), aggregate_test_100.c1                                                                                                            |
|               |             TableScan: aggregate_test_100 projection=[c1], partial_filters=[random() > Float64(0.5), random() > Float64(0.8), aggregate_test_100.c1 > Utf8("5")] |

after this pr

| logical_plan  | SubqueryAlias: t                                                                                                                                      |
|               |   Filter: r > Float64(0.5)                                                                                                                            |
|               |     Projection: aggregate_test_100.c1, random() AS r                                                                                                  |
|               |       Aggregate: groupBy=[[aggregate_test_100.c1]], aggr=[[]]                                                                                         |
|               |         Filter: random() > Float64(0.8) AND aggregate_test_100.c1 > Utf8("5")                                                                         |
|               |           TableScan: aggregate_test_100 projection=[c1], partial_filters=[random() > Float64(0.8), aggregate_test_100.c1 > Utf8("5")] 

What changes are included in this PR?

modify push_down_filter to support only push down predicates that do not contain volatile functions or contain volatile functions but are not included in projection.

Are these changes tested?

Yes, if need, I can add more.

Are there any user-facing changes?

[haohuaijin]

cc @jonahgao @alamb

[alamb]

Thank you very much @haohuaijin -- I think this is very close. The only thing I see that is possibly incorrect is handling expressions like random() + 1 -- I also left some suggestions that I think could make the code cleaner but are not required in my opinion.

Thanks again

[alamb]

I don't think this will work for expressions that have random

It will find expressions like random() but not expressions like random() + 1,

To do so I think you need to write a function (perhaps via expr.apply() that looks for volatile expressions recursively

[haohuaijin]

you are right, already fix, thanks!

[jonahgao]

Thank you @haohuaijin. I have suggested some changes about clone. Everything else looks good to me.

[alamb]

Thank you @haohuaijin and @jonahgao