alexy opened a new issue, #26054:
URL: https://github.com/apache/datafusion/issues/26054
### Describe the bug
In a recursive CTE, an uncorrelated `MIN`/`MAX` aggregate over a table scan
in the recursive term returns the right value on the first iteration and NULL
(or a wrong value) on every later iteration.
`RecursiveQueryExec` runs the recursive term again for each iteration after
`reset_plan_states`. `AggregateExec` has no `reset_state` of its own, and its
`replace_children` copies `dynamic_filter` into the new node
(`me.dynamic_filter.clone_from(&self.dynamic_filter)`). So the rebuilt plan
shares the `DynamicFilterPhysicalExpr` that the first iteration tightened. The
scan in the second iteration starts with `r > 128`, finds nothing, and `MAX` is
NULL.
`HashJoinExec` and `SortExec` reset their dynamic filters in `reset_state`;
`AggregateExec` does not. Reproduced on 55.1.0, and the same code is on `main`
(8248a57969).
### To Reproduce
With datafusion-cli (or any `SessionContext`):
```sql
COPY (SELECT CAST(value AS DOUBLE) AS r FROM generate_series(0, 128)) TO
'/tmp/df_recursive_max/t.parquet';
CREATE EXTERNAL TABLE t STORED AS PARQUET LOCATION
'/tmp/df_recursive_max/t.parquet';
WITH RECURSIVE w(n, m) AS (
SELECT 0, CAST(NULL AS DOUBLE)
UNION ALL
SELECT w.n + 1, x.m
FROM w CROSS JOIN (SELECT MAX(r) AS m FROM t WHERE r % 2 = 0) x
WHERE w.n < 3
)
SELECT n, m FROM w ORDER BY n;
```
```
+---+-------+
| n | m |
+---+-------+
| 0 | |
| 1 | 128.0 |
| 2 | |
| 3 | |
+---+-------+
```
The `WHERE r % 2 = 0` matters only because without a filter, `MAX(r)` is
answered from the Parquet statistics and nothing is scanned.
### Expected behavior
```
+---+-------+
| n | m |
+---+-------+
| 0 | |
| 1 | 128.0 |
| 2 | 128.0 |
| 3 | 128.0 |
+---+-------+
```
which is what `SET
datafusion.optimizer.enable_aggregate_dynamic_filter_pushdown = false` gives.
### Additional context
- A possible fix is an `AggregateExec::reset_state` that creates a fresh
dynamic filter (as `init_dynamic_filter` does) and pushes it to the input
again, as `HashJoinExec::reset_state` and `SortExec::reset_state` do. #21650
discusses the general problem of shared state across `reset_state`.
- Found through a downstream project
([Sail](https://github.com/lakehq/sail)) running a long recursive query, where
a later iteration silently lost rows. Work around it by turning off
`enable_aggregate_dynamic_filter_pushdown` for plans that contain a
`RecursiveQueryExec`.
--
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]
