Preserve and reuse shared physical subplans across distributed task specialization

Problem

Queries that reference the same CTE multiple times can expand that CTE into independent physical-plan branches. During distributed planning, each branch is task-specialized and serialized as part of a complete protobuf plan tree.

The current DeduplicatingProtoConverter preserves shared PhysicalExpr identity, but it does not preserve or reuse shared ExecutionPlan nodes.

For CTE-heavy queries such as TPC-DS Q4, Q11, and Q74, the same CTE can appear four to six times. Independent branches can concurrently repeat:

• Scan execution.
• Aggregation.
• Broadcast or shuffle work.
• Hash-table construction.
• Plan serialization and deserialization.

This multiplies transient memory and can require downstream consumers to detect reused CTEs and reduce scan concurrency or eager probe polling.

How to reproduce

1. Plan a query that references one CTE several times, such as TPC-DS Q4.
2. Inspect the distributed physical plan and per-task serialized plans.
3. Observe that each reference contains an independent copy of the CTE branch.
4. Compare serialized plan size, scan execution count, and peak memory as the number of references increases.

A focused test could construct an ExecutionPlan DAG where several consumers reference the same Arc<dyn ExecutionPlan>, then verify whether task specialization and protobuf round-tripping preserve that identity.

Expected behavior

Where semantics allow, distributed planning should preserve common-subplan identity and execute a shared producer once, with its output consumed by multiple downstream branches.

Possible directions include:

• Represent shared plan nodes using stable IDs and references during serialization.
• Preserve DAG identity during task specialization.
• Materialize a shared distributed stage once and fan its output out to multiple consumers.
• Expose a planner hook allowing engines to choose which common subplans should be materialized.

Any implementation must preserve independent consumer backpressure and cancellation behavior.

Additional context

• DeduplicatingProtoConverter addresses shared physical expressions, especially dynamic filters, but not shared execution-plan nodes.
• Limiting concurrency reduces peak memory but still repeats the underlying work.
• The file-partition amplification observed on an older revision has since been redesigned upstream and is intentionally outside this request.
• If common-subplan identity must first be exposed by Apache DataFusion, this request can track that dependency.

Start with the DeduplicatingProtoConverter and the distributed task-specialization and protobuf round-tripping paths. Build the proposed focused test with an ExecutionPlan DAG whose consumers share an Arc, then compare identity, serialized plans, and execution behavior. Done means shared producers are preserved and reused where semantics allow without breaking backpressure or cancellation.