Recursive CTE: MIN/MAX aggregate dynamic filter survives reset_state, later iterations return wrong results #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 (8248a57).

To Reproduce

With datafusion-cli (or any SessionContext):

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. [DISCUSS] Representing Shared State / ExecutionPlan::reset_state #21650 discusses the general problem of shared state across reset_state.
• Found through a downstream project (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.