jayzhan211 commented on issue #19905:
URL: https://github.com/apache/datafusion/issues/19905#issuecomment-6226151447
@askalt are you still working on this? I looked at the same area and would
like to suggest a simpler model, based on what main does today.
**What happens today in a recursive CTE** (the recursive term is reset with
`reset_state` before each iteration). The three dynamic-filter producers each
do something different:
| Producer | `reset_state` | Effect inside a recursive term |
|---|---|---|
| `AggregateExec` (MIN/MAX) | none, keeps the filter | wrong results from
iteration 2 (#26054) |
| `HashJoinExec` | drops its filter (`hash_join/exec.rs`, `dynamic_filter =
None`) | the scan keeps the old filter, which is never updated: no pruning |
| `SortExec` (TopK) | creates a new filter (`sort.rs`, `create_filter()`) |
the scan keeps the old filter, which is never updated: no pruning |
Repro for the hash join (main, 81824e8230):
```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 ], 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 common use of recursive CTEs. On a 20M-row
edge table sorted by `src`, one join step takes 3 ms with the dynamic filter
(19 of 20 row groups pruned) and 24 ms without it; a 10-step recursive walk
takes 0.90 s today.
**Suggested model: a producer owns one filter and resets its value, never
its identity.** The scan already holds a reference to that filter from
planning. If `reset_state` clears the value in place instead of dropping or
replacing the filter, the scan sees the next iteration's values with no
re-wiring, so the `dynamic_filters()` / `PushedDynamicFilters` plumbing
proposed above would not be needed.
#26056 already does this for `AggregateExec` and adds
`DynamicFilterPhysicalExpr::reset()`. After it lands, the remaining steps are
small:
1. `HashJoinExec::reset_state` keeps its filter and calls `reset()` instead
of setting it to `None`, plus a `cte.slt` case that checks pruning in a
recursive join.
2. `SortExec::reset_state` does the same instead of `create_filter()`, plus
a `cte.slt` case for TopK.
If you are no longer working on this, would it be OK to mark it help wanted
with these steps?
--
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]
