jayzhan211 opened a new issue, #26641:
URL: https://github.com/apache/datafusion/issues/26641
### Is your feature request related to a problem or challenge?
Inside a recursive CTE, a hash join's dynamic filter never reaches the
probe-side scan, so the scan reads every row on every iteration.
`RecursiveQueryExec` calls `reset_plan_states` on the recursive term before
each iteration. `HashJoinExec::reset_state` sets `dynamic_filter = None`, but
the scan is reset first and keeps the original filter. Nothing updates that
filter, so it stays empty.
```sql
SET datafusion.execution.target_partitions = 1;
SET datafusion.execution.parquet.pushdown_filters = true;
COPY (SELECT value AS k FROM generate_series(1, 100)) TO '/tmp/t.parquet';
CREATE EXTERNAL TABLE t STORED AS PARQUET LOCATION '/tmp/t.parquet';
EXPLAIN ANALYZE WITH RECURSIVE w(n) AS (
SELECT 1 UNION ALL SELECT t.k FROM w JOIN t ON t.k = w.n + 1 WHERE w.n < 5
) SELECT * FROM w ORDER BY n;
-- scan of t: DynamicFilter [ empty ], output_rows=400,
pushdown_rows_pruned=0
EXPLAIN ANALYZE SELECT t.k FROM (VALUES (1), (2)) w(n) JOIN t ON t.k = w.n +
1;
-- scan of t: DynamicFilter [ k@0 >= 2 AND k@0 <= 3 AND ... ],
pushdown_rows_pruned=98
```
This matters for graph walks, the most common use of recursive CTEs. On a
20M-row edge table sorted by `src`, one join step takes 3 ms with the dynamic
filter and 24 ms without it.
### Describe the solution you'd like
Reset the filter in place instead of dropping it, as #26056 does for
`AggregateExec` (see the discussion there):
- `HashJoinExec::reset_state` keeps its `DynamicFilterPhysicalExpr`, calls
`reset()` on it, and starts with a new `build_accumulator`.
- Add a `cte.slt` case that checks the scan prunes rows in a recursive join.
The scan already holds this filter, so it sees each iteration's values
without any re-wiring.
### Describe alternatives you've considered
Creating a new filter and reconnecting it to the scan
(`PushedDynamicFilters` in #19905). `reset_plan_states` resets the scan before
the join, so that needs an extra pass over the plan.
### Additional context
Depends on #26056 for `DynamicFilterPhysicalExpr::reset()`. The `SortExec`
counterpart is #TBD. Longer term, #21650 discusses moving this shared state out
of the plan.
--
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]
