adriangb opened a new issue, #23494: URL: https://github.com/apache/datafusion/issues/23494
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`](https://github.com/apache/datafusion/blob/main/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, ride-along reference):** the `try_to_proto`/`try_from_proto` hooks + ctx + `ProjectionExec` — this must land first. _(link below)_ - **Full end-state (draft, for reference only):** the entire port across all plans + the `DataSource`/`DataSink` families. _(link below)_ ## Scope / checklist Each item is an independent follow-up PR. The two foundation items unblock their families. ### Foundations - [ ] DataSource/FileSource `try_to_proto` hook + `FileScanConfig` serde port - [ ] DataSink `try_to_proto` hook + `FileSinkConfig` serde port ### `datafusion-physical-plan` plans - [ ] FilterExec - [ ] Trivial single-child: CoalesceBatchesExec, CoalescePartitionsExec, CooperativeExec, BufferExec - [ ] Leaves: EmptyExec, PlaceholderRowExec - [ ] Limits: GlobalLimitExec, LocalLimitExec - [ ] UnionExec + InterleaveExec - [ ] RepartitionExec - [ ] SortExec + SortPreservingMergeExec - [ ] Simple joins: CrossJoinExec, NestedLoopJoinExec - [ ] HashJoinExec - [ ] SortMergeJoinExec - [ ] SymmetricHashJoinExec - [ ] UnnestExec - [ ] ExplainExec + AnalyzeExec - [ ] AggregateExec (function-codec path) - [ ] Window execs: WindowAggExec + BoundedWindowAggExec - [ ] AsyncFuncExec - [ ] ScalarSubqueryExec (adds a scalar-subquery-results scope to the decode ctx) ### DataSource / DataSink families (blocked by the foundations) - [ ] File scans: CsvSource, JsonSource, ArrowSource, AvroSource - [ ] ParquetSource - [ ] MemorySourceConfig - [ ] 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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
