Hash join dynamic filters don't prune inside recursive CTEs #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.

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 #26642. Longer term, #21650 discusses moving this shared state out of the plan.