Skip to content

[EPIC] Port ExecutionPlan serialization to try_to_proto / try_from_proto hooks #23494

Description

@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). The plan's old PhysicalPlanNodeExt methods (try_into_<plan>_physical_plan / try_from_<plan>_exec) are not deleted: keep them marked #[deprecated(since = "55.0.0", note = ...)] per the API health policy, but reduce their bodies to thin shims that delegate to the new hook — build an ExecutionPlanEncodeCtx / ExecutionPlanDecodeCtx from the codec / proto_converter (via ConverterPlanEncoder / ConverterPlanDecoder) and call X::try_to_proto / X::try_from_proto, so the wire format stays single-sourced in the plan's own crate rather than duplicated in datafusion-proto (see the ProjectionExec reference in Add ExecutionPlan try_to_proto / try_from_proto hooks + ProjectionExec reference #23495, updated in refactor(proto): delegate deprecated ProjectionExec serde shims to new hooks #23731, and the FilterExec migration in refactor(proto): migrate FilterExec serde #23708).
  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

Scope / checklist

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

Foundations

datafusion-physical-plan plans

DataSource / DataSink families (blocked by the foundations)

Cleanup

  • Remove the #[deprecated(since = "55.0.0")] PhysicalPlanNodeExt scaffolding after the deprecation window (6 major versions or 6 months, whichever is longer) + Upgrade Guide entry

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 / deprecation plan

The per-plan PhysicalPlanNodeExt serialization methods and the TryFromProto<&protobuf::{Json,Csv,Parquet}Sink> impls are pub-for-proto scaffolding added in #21929 (after the 54.0.0 release, so none of it has shipped in a release yet). Rather than removing them as each plan migrates, each incremental PR keeps its plan's methods under #[deprecated(since = "55.0.0")] with their bodies reduced to thin shims that delegate to the plan's new try_to_proto / try_from_proto hook, per the API health policy — the dispatch no longer calls them, but downstream callers keep compiling with a warning. This keeps every follow-up PR semver-clean.

Actual removal happens in a single cleanup PR after the deprecation window (6 major versions or 6 months, whichever is longer), summarized in the Upgrade Guide (tracked in the checklist below). The load-bearing public API — AsExecutionPlan, PhysicalExtensionCodec, PhysicalProtoConverterExtension — is unchanged throughout.

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Type

    No type

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions