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

    Fields

    No fields configured for issues without a type.

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions