fix: keep the output of a materialized CTE per run, not in the plan

[adriangb]

Which issue does this PR close?

• Closes MaterializedCteExec::reset_state clears its shared buffer in place #26229.

Rationale for this change

MaterializedCteExec::reset_state cleared its shared buffer in place and returned self. ExecutionPlan::reset_state must return a node with new state and leave the old node unchanged. Because the buffered output was in the plan, two runs of one physical plan also shared it.

Example with one physical plan for WITH c AS MATERIALIZED (SELECT random() AS x) SELECT * FROM c:

	Before	After
Run 1 and run 2 of the same plan, each with its own TaskContext	Run 2 replays the buffer of run 1.	Each run computes the body.
reset_plan_states(plan) while run 1 is open	The buffer that run 1 uses is replaced.	Run 1 is not changed.

No SQL query is known to give a wrong result on #25580.

What changes are included in this PR?

• TaskContext has a new RunState. It holds state that operators keep for one run of a plan.
• The output of a materialized CTE is now in the RunState, keyed by the CTE id. MaterializedCteBuffer holds plan-time data only.
• MaterializedCteExec uses the default reset_state.
• RecursiveQueryExec gives each iteration a child RunState, in which the CTEs that the recursive term declares are computed again. A CTE that is declared outside the recursive term keeps its output for the whole query.

Alternative that was not used: rebind the scans to a new buffer in reset_plan_states. That function cannot reach a scan in a scalar subquery, and it does not separate two runs when nothing calls reset.

Notes for reviewers:

• A run is identified by its TaskContext. If each partition is executed with a different TaskContext, the body is computed one time for each context.
• The output now lives until the TaskContext of the run and the replay streams are dropped. Release the buffer of a materialized CTE when the last scan finishes #26233 releases it earlier.
• Open point: the body future holds an Arc<TaskContext>, the TaskContext owns the RunState, and the RunState owns the future. While the body is not complete this is a reference cycle. A run that is abandoned before the body completes can keep its memory. This is not confirmed by a test and is not corrected yet.
• ScalarSubqueryExec::reset_state has the same in-place reset and is not changed here.

Comparison with PostgreSQL and DuckDB. This is what PostgreSQL does: the CTE output is a tuplestore in the executor state of the first CteScan, not in the plan, so each execution has its own (nodeCtescan.c).

Are these changes tested?

Yes. Unit tests: two overlapping runs of one plan have separate outputs, reset_plan_states leaves the old plan usable, a renewed context computes only the given CTEs again, and a test for RunState. The recursive cases in cte_materialized.slt pass with no change. There is no slt for nested recursion.

Are there any user-facing changes?

New public API: datafusion_execution::RunState, TaskContext::run_state, TaskContext::with_run_state, and TaskContext: Clone. No existing API changes.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 82.81938% with 39 lines in your changes missing coverage. Please review.
✅ Project coverage is 82.88%. Comparing base (5009f42) to head (07a4fbb).
⚠️ Report is 10 commits behind head on main.

Files with missing lines	Patch %	Lines
datafusion/physical-plan/src/materialized_cte.rs	79.16%	12 Missing and 23 partials ⚠️
datafusion/execution/src/task.rs	94.11%	3 Missing ⚠️
datafusion/physical-plan/src/recursive_query.rs	87.50%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #26270      +/-   ##
==========================================
+ Coverage   82.87%   82.88%   +0.01%     
==========================================
  Files        1153     1153              
  Lines      457101   457702     +601     
  Branches   457101   457702     +601     
==========================================
+ Hits       378818   379381     +563     
- Misses      55268    55286      +18     
- Partials    23015    23035      +20     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.