adriangb opened a new issue, #26267:
URL: https://github.com/apache/datafusion/issues/26267
### Describe the bug
With `datafusion.execution.parquet.pushdown_filters = true`, a filter on a
subquery column that comes from a volatile function returns wrong results.
The query drops rows that match the filter and returns rows that do not
match the filter.
The physical plan has no `FilterExec`. The Parquet scan has the predicate
`random() < 0.1`, which calls `random()` again. The values that the scan tests
are not the values that the projection returns.
Tested on `main` at 2c568b478b3aff18de52609c7237f1c3f494d5b1 with
`datafusion-cli`.
### To Reproduce
```sql
COPY (SELECT value AS n FROM generate_series(1, 1000)) TO '/tmp/ev.parquet';
CREATE EXTERNAL TABLE ev STORED AS PARQUET LOCATION '/tmp/ev.parquet';
SET datafusion.explain.format = 'indent';
SET datafusion.execution.parquet.pushdown_filters = true;
-- Query 1: the predicate is true for every value of x, so the result must
be 1000.
SELECT count(*) FROM (SELECT random() AS x FROM ev) WHERE x < 0.5 OR x >=
0.5;
-- Query 2: every row in the result must have x < 0.1, so max(x) must be
less than 0.1.
SELECT count(*), max(x) FROM (SELECT random() AS x FROM ev) WHERE x < 0.1;
EXPLAIN SELECT count(*), max(x) FROM (SELECT random() AS x FROM ev) WHERE x
< 0.1;
```
Actual output of query 1 (the number changes on each run, it is near 750):
```
+----------+
| count(*) |
+----------+
| 729 |
+----------+
```
Actual output of query 2 (`max(x)` is larger than `0.1`):
```
+----------+--------------------+
| count(*) | max(x) |
+----------+--------------------+
| 110 | 0.9983553383621974 |
+----------+--------------------+
```
`EXPLAIN` output. The logical plan is correct: the `Filter` stays above the
`Projection`. The physical plan has no `FilterExec`, and the scan has
`predicate=random() < 0.1`:
```
logical_plan
Projection: count(Int64(1)) AS count(*), max(x)
Aggregate: groupBy=[[]], aggr=[[count(Int64(1)), max(x)]]
Filter: x < Float64(0.1)
Projection: random() AS x
TableScan: ev projection=[]
physical_plan
ProjectionExec: expr=[count(Int64(1))@0 as count(*), max(x)@1 as max(x)]
AggregateExec: mode=Final, gby=[], aggr=[count(Int64(1)), max(x)]
CoalescePartitionsExec
AggregateExec: mode=Partial, gby=[], aggr=[count(Int64(1)), max(x)]
ProjectionExec: expr=[random() as x]
RepartitionExec: partitioning=RoundRobinBatch(12),
input_partitions=1
DataSourceExec: file_groups={1 group: [[tmp/ev.parquet]]},
file_type=parquet, predicate=random() < 0.1
```
### Expected behavior
Query 1 returns `1000`.
Query 2 returns a `max(x)` that is less than `0.1`.
With `datafusion.execution.parquet.pushdown_filters = false` (the default)
the same session gives these correct results:
```
+----------+
| count(*) |
+----------+
| 1000 |
+----------+
+----------+---------------------+
| count(*) | max(x) |
+----------+---------------------+
| 93 | 0.09977364854487647 |
+----------+---------------------+
```
A filter must not move through a projection when that changes the filter
into a volatile expression.
### Additional context
Other observations at the same commit:
- `uuid()` shows the same bug: `SELECT count(*) FROM (SELECT n, uuid() AS x
FROM ev) WHERE x < '8' OR x >= '8'` returned `745`, not `1000`.
- The bug also occurs when the subquery selects other columns (`SELECT n,
random() AS x FROM ev`) and when the volatile function is a part of a larger
expression (`n + random() AS x` gives `predicate=CAST(n AS Float64) + random()
< 100` on the scan).
- With `pushdown_filters = false` the results are correct, because the
`FilterExec` stays in the plan. But the scan also has `predicate=random() <
0.1` in that configuration:
```
FilterExec: x@0 < 0.1
ProjectionExec: expr=[random() as x]
RepartitionExec: partitioning=RoundRobinBatch(12), input_partitions=1
DataSourceExec: file_groups={1 group: [[tmp/ev.parquet]]},
file_type=parquet, predicate=random() < 0.1
```
- CSV and in-memory tables give correct results. These sources do not accept
the filter, so the `FilterExec` stays in the plan.
- A volatile predicate written directly on the table (`SELECT count(*) FROM
ev WHERE random() < 0.1`) is not pushed down. The `FilterExec` stays in the
plan and the scan has no predicate.
- `EXPLAIN VERBOSE` shows that the plan changes in the physical
`FilterPushdown` rule. The logical `PushDownFilter` rule does not move the
filter.
Suspected cause, from reading the code (no fix was tested):
1. `ProjectionExec::gather_filters_for_pushdown` in
`datafusion/physical-plan/src/projection.rs` replaces each column of the parent
filter with the projection expression (`unproject_expr`) and marks the result
as supported. It does not check if the projection expression is volatile. Thus
`x@0 < 0.1` becomes `random() < 0.1`.
2. `push_down_filters` in
`datafusion/physical-optimizer/src/filter_pushdown.rs` removes volatile
predicates (`allow_pushdown_for_expr`) only from the list that it gives to
`gather_filters_for_pushdown`. It gives the full `parent_predicates` list, with
the volatile predicates, to `handle_child_pushdown_result`.
`DataSourceExec::handle_child_pushdown_result` in
`datafusion/datasource/src/source.rs` gives all of these filters to the data
source. The Parquet source accepts them, and then the `FilterExec` is removed.
This was found during work on #26239.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
