Wrong results: with pushdown_filters = true, a filter on a volatile subquery column is evaluated again in the Parquet scan #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 2c568b4 with datafusion-cli.

To Reproduce

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