[EPIC] Port ExecutionPlan serialization to try_to_proto / try_from_proto hooks

[adriangb]

Follow-up to #22418 (the equivalent PhysicalExpr effort). This issue tracks porting the built-in ExecutionPlan implementations — and the DataSource/FileSource/DataSink families — to per-type try_to_proto / try_from_proto hooks, eliminating the central downcast_ref chain in datafusion/proto/src/physical_plan/mod.rs.

Why

Today every built-in ExecutionPlan is serialized through a large downcast_ref chain on the encode side and a symmetric match on PhysicalPlanType on the decode side. This is the same problem #21835 described for PhysicalExpr: it forces internal state to be exposed as pub for the encoder to reach, makes it easy to forget serialization when adding a plan, and splits built-in vs. third-party plans across different code paths.

Mirroring #21929 / #22418, each plan can move onto a try_to_proto hook (a trait method, default Ok(None)) plus a try_from_proto associated fn, one PR at a time, with no wire-format change.

The hook + context

Added by the base PR (see below), in datafusion-physical-plan:

• ExecutionPlan::try_to_proto(&self, ctx: &ExecutionPlanEncodeCtx) -> Result<Option<PhysicalPlanNode>> — trait method, default Ok(None), feature-gated #[cfg(feature = "proto")].
• <PlanType>::try_from_proto(node, ctx: &ExecutionPlanDecodeCtx) -> Result<Arc<dyn ExecutionPlan>> — inherent associated fn, wired into the matching PhysicalPlanType::… decode arm.
• ExecutionPlanEncodeCtx / ExecutionPlanDecodeCtx expose everything a plan needs without naming datafusion-proto: encode_child/decode_child(+ _children), encode_expr/decode_expr (against any schema), and — for function-carrying plans — typed bytes-only encode_udaf/decode_udaf (+ udf/udwf). The dependency inversion (internal dispatch traits defined in physical-plan, implemented in datafusion-proto) is identical to the PhysicalExpr ctx.

The encode dispatch resolves downcast_delegate() before calling the hook, so wrapper plans serialize as their delegate.

Note on function-carrying plans (AggregateExec, window execs, ScalarSubqueryExec)

Unlike the PhysicalExpr side — where the ctx lives below datafusion-expr and so ScalarFunctionExpr must stay special-cased — datafusion-physical-plan sits above datafusion-expr, so the ctx can carry typed function serde directly. This means aggregate/window/scalar-subquery plans ride the hook via ctx.encode_udaf / decode_udaf etc.; no PhysicalExtensionCodec is exposed and no proto type crosses the boundary. (This is why the ScalarUdfCodec proposal in #23421 is not needed for plans.)

Pattern (per plan)

1. try_to_proto inside impl ExecutionPlan for FooExec, #[cfg(feature = "proto")]. ⚠️ It must be the trait-method override — an inherent method with the same name is silently never called through &dyn ExecutionPlan.
2. FooExec::try_from_proto in a separate #[cfg(feature = "proto")] impl FooExec block, wired into the PhysicalPlanType::Foo(_) => FooExec::try_from_proto(self.node(), &decode_ctx) decode arm.
3. In the same PR, delete the plan's old encode downcast_ref arm and repoint its decode arm — this is the only proof the hook is actually reached (roundtrip tests pass through the new path).
4. Keep the wire format byte-for-byte identical. Enum conversions (JoinType, NullEquality, PartitionMode, JoinSide, AggregateMode, …) inline as by-name exhaustive matches — the proto and datafusion_common enums are numbered differently, so a numeric cast silently corrupts them.

The reference implementation is ProjectionExec in the base PR. HashJoinExec is the reference for joins (two children, on-columns, JoinFilter, projection sentinel, enum conversions); AggregateExec for the function-codec path.

Reference PRs

• Base (foundation): Add ExecutionPlan try_to_proto / try_from_proto hooks + ProjectionExec reference #23495 — the try_to_proto/try_from_proto hooks + ctx + ProjectionExec. Must land first.
• Full end-state (draft, reference only): [EPIC/reference] Port ExecutionPlan serialization to try_to_proto / try_from_proto hooks (full) #23496 — the entire port across all plans + the DataSource/DataSink families.

Scope / checklist

Each item is an independent follow-up PR. The two foundation items unblock their families.

Foundations

• Proto: add DataSource/FileSource try_to_proto hook + port FileScanConfig serde #23497 — DataSource/FileSource try_to_proto hook + FileScanConfig serde port
• Proto: add DataSink try_to_proto hook + port FileSinkConfig serde #23498 — DataSink try_to_proto hook + FileSinkConfig serde port

datafusion-physical-plan plans

• Proto: migrate FilterExec to the try_to_proto hook #23499 — FilterExec
• Proto: migrate trivial single-child plans (CoalesceBatches, CoalescePartitions, Cooperative, Buffer) #23500 — Trivial single-child: CoalesceBatches, CoalescePartitions, Cooperative, Buffer
• Proto: migrate leaf plans EmptyExec + PlaceholderRowExec #23501 — Leaves: EmptyExec, PlaceholderRowExec
• Proto: migrate GlobalLimitExec + LocalLimitExec #23502 — Limits: GlobalLimitExec, LocalLimitExec
• Proto: migrate UnionExec + InterleaveExec #23503 — UnionExec + InterleaveExec
• Proto: migrate RepartitionExec #23504 — RepartitionExec
• Proto: migrate SortExec + SortPreservingMergeExec #23505 — SortExec + SortPreservingMergeExec
• Proto: migrate CrossJoinExec + NestedLoopJoinExec #23506 — Simple joins: CrossJoinExec, NestedLoopJoinExec
• Proto: migrate HashJoinExec #23507 — HashJoinExec
• Proto: migrate SortMergeJoinExec #23508 — SortMergeJoinExec
• Proto: migrate SymmetricHashJoinExec #23509 — SymmetricHashJoinExec
• Proto: migrate UnnestExec #23510 — UnnestExec
• Proto: migrate ExplainExec + AnalyzeExec #23511 — ExplainExec + AnalyzeExec
• Proto: migrate AggregateExec (function-codec path) #23512 — AggregateExec (function-codec path)
• Proto: migrate WindowAggExec + BoundedWindowAggExec #23513 — Window execs: WindowAggExec + BoundedWindowAggExec
• Proto: migrate AsyncFuncExec #23514 — AsyncFuncExec
• Proto: migrate ScalarSubqueryExec (adds a subquery-results scope to the decode ctx) #23515 — ScalarSubqueryExec (adds a subquery-results scope to the decode ctx)

DataSource / DataSink families (blocked by the foundations)

• Proto: migrate the simple file scans (CsvSource, JsonSource, ArrowSource, AvroSource) #23516 — File scans: CsvSource, JsonSource, ArrowSource, AvroSource
• Proto: migrate ParquetSource #23517 — ParquetSource
• Proto: migrate MemorySourceConfig #23518 — MemorySourceConfig
• Proto: migrate the file sinks (CsvSink, JsonSink, ParquetSink) #23519 — File sinks: CsvSink, JsonSink, ParquetSink

Intentionally left as typed dispatch

• GenerateSeries / LazyMemoryExec — the generate-series generator types live in datafusion-functions-table, which sits above datafusion-physical-plan; an ExecutionPlan-keyed hook cannot name them without a crate cycle. Would require a generator-level hook (separate design).
• Extension — the terminal codec.try_encode fallback for arbitrary third-party plans. This is the intended endpoint and stays.

API change

Removing the per-plan PhysicalPlanNodeExt serialization methods and the TryFromProto<&protobuf::{Json,Csv,Parquet}Sink> impls is a deliberate breaking change (these were pub-for-proto scaffolding added in #21929 and released in 54.0.0). The full end-state PR removes them; each incremental PR removes its own arms. This will be summarized in the Upgrade Guide. The load-bearing public API — AsExecutionPlan, PhysicalExtensionCodec, PhysicalProtoConverterExtension — is unchanged.