MaterializedCteExec::reset_state clears its shared buffer in place #26229

Is your feature request related to a problem or challenge?

Follow-up to #25580 (WITH ... AS MATERIALIZED), part of #17737. Raised by @nathanb9 in #25580 (comment).

ExecutionPlan::reset_state is expected to return a node with new state and to leave the old node unchanged. MaterializedCteExec::reset_state does not do this. It clears the shared MaterializedCteBuffer in place and returns self.

No SQL query is known to give a wrong result because of this. A scan that has already called execute() holds the old output and continues to work. A result can change only if two runs of the same plan overlap, and one run starts a scan after the other run resets the buffer. Nothing in DataFusion does that today.

The in-place reset is there for two reasons:

• RecursiveQueryExec calls reset_plan_states on the recursive term before each iteration. A MATERIALIZED CTE that is declared in the recursive term must be computed again in each iteration. If the buffer is not cleared, each iteration replays the first result and the query does not terminate.
• A new node with a new buffer also needs new scans. reset_state sees only its own node. A scan in a scalar subquery is not below MaterializedCteExec, because the planner puts ScalarSubqueryExec above the root, so the producer cannot rebind that scan.

ScalarSubqueryExec::reset_state clears its shared results in place for the same reason.

Describe the solution you'd like

Keep the materialized output per run, not on the plan node. For example, a registry in TaskContext, keyed by the CTE id, that RecursiveQueryExec renews for each iteration. MaterializedCteExec::reset_state can then follow the contract, and two overlapping runs of one plan do not share a buffer.

The same mechanism could be used for ScalarSubqueryExec.

Describe alternatives you've considered

Keep the in-place reset and document it as an exception to the contract.

Additional context

No response