Repository navigation
feat: allow extension PhysicalExprs to decode via a per-type registry - #24628
Conversation
|
Thank you for opening this pull request! Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch). Details |
1684c28 to
7b19bb3
Compare
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #24628 +/- ##
==========================================
- Coverage 82.92% 82.92% -0.01%
==========================================
Files 1155 1156 +1
Lines 458204 458302 +98
Branches 458204 458302 +98
==========================================
+ Hits 379965 380034 +69
- Misses 55205 55229 +24
- Partials 23034 23039 +5 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
11c4ae5 to
615feb3
Compare
615feb3 to
1133780
Compare
31a7449 to
569a62e
Compare
569a62e to
a58c8e1
Compare
86b779f to
9e43f5d
Compare
|
I put this in draft. Lets sequence behind #24631 to avoid duplication. I will keep it updated though. |
5064ec8 to
f11996d
Compare
f11996d to
7f9361a
Compare
7f9361a to
f3d8556
Compare
…try instead of `PhysicalExtensionCodec` (apache#24631) ## Which issue does this PR close? - Closes apache#24625. ## Rationale for this change ### The user-visible problem A library that ships its own `ExecutionPlan` (datafusion-distributed ships six) has to write a `PhysicalExtensionCodec` to serialize it: a central `downcast_ref` chain for encode and a matching `match` for decode, in a file separate from the plan. Two such libraries on one session also have to compose their codecs by hand through `ComposedPhysicalExtensionCodec`, and their users have to remember to install the composed codec on every process that decodes a plan. That composition is fragile in a way that is easy to miss. `PhysicalExtensionNode` carries no type discriminator, so the codec *is* the discriminator. `ComposedPhysicalExtensionCodec` copes by writing the **position** of the encoding codec into the payload bytes. The decoding side must therefore register the same codecs in the same order, a plan that two codecs both accept resolves to whichever was registered first, and a name collision between two independent crates cannot be detected. datafusion-python and datafusion-distributed have each built their own workaround for this (a codec-id layer and a `user_codec.rs` composition file respectively). ### Where this sits in the bigger picture Physical-plan serialization is moving from central `downcast_ref` chains in `datafusion-proto` to per-type hooks that live next to each type: `try_to_proto` (a trait method, dispatched through the vtable) plus an inherent `try_from_proto` (a constructor, so it cannot be a trait method). apache#23494 did this for every built-in `ExecutionPlan`; apache#22418 did it for `PhysicalExpr`. Third-party types could use the encode half of that but not the decode half, because decoding an extension node always routed through `PhysicalExtensionCodec::try_decode`. This PR closes the loop for `ExecutionPlan`; apache#24628 does the same for `PhysicalExpr` with the same shape and the same names (`Extension*FromProto` pairing the wire name with the constructor, `register_*` / `decode_*` facades over one shared registry, attached through `SessionConfig::with_extension`). **apache#24628 is stacked on this PR**, which adds the shared registry, and must land second. ### How this relates to `PhysicalExtensionCodec` The codec is **not deprecated** and this PR does not touch `ComposedPhysicalExtensionCodec`. The decode rule is: - A `PhysicalExtensionNode` **with a `plan_name` that the decoding session has registered** decodes through the plan's own `try_from_proto`. A failure there is fatal: falling through to the codec would let it wrongly decode a payload it never wrote, which is the bug the name exists to prevent. - **Anything else** (no name, because a codec wrote it, or a name this session does not know) takes the codec chain exactly as before. If the codec also fails, the missing registration is added as *context* on the codec's own error, and lists what the session does know. The codec's error kind is kept, because a codec can fail for a reason that has nothing to do with a missing registration. The codec keeps three jobs the registry does not take: UDF/UDAF/UDWF payloads (a function is an *instance*, not a type, so a type-keyed registry does not fit), anything not yet migrated, and the released `FFI_PhysicalExtensionCodec` in `datafusion-ffi`, which datafusion-python exposes as public API. What becomes deprecable once plans and expressions are registry-backed is `ComposedPhysicalExtensionCodec`, whose reason to exist was composition; that is a later, separate decision. Making the composed codec itself name-keyed instead of position-keyed (as suggested on the issue) is also orthogonal to this PR and can be done independently. ### How FFI will work Today `FFI_ExecutionPlan` has no serialization slot, so a foreign plan always goes through the codec. This PR does not change that. It is designed so FFI can plug in later with no breaking change: - The registry stores decoders as a **private** `fn` pointer. The public surface is `ExtensionPlanFromProto` and the three facade functions. The storage can grow to carry an FFI vtable and private data, and a lower-level entry point that takes a name and a decoder object can be added, both without breaking anything shipped here. - `PhysicalPlanNode` is a prost message, so it crosses the ABI as bytes, the same way `FFI_ExecutionPlan::partition_statistics` already ships a prost-encoded `Statistics`. Protobuf's field-skipping gives version-skew tolerance that a `#[repr(C)]` struct cannot. - One limit FFI inherits from the codec rather than from this PR: `ForeignPhysicalExtensionCodec::try_decode` discards the caller's `TaskContext` and rebuilds one from the provider's side, so a decoder that reads the *decoding* session's state (the second test below) would not see it across FFI. ## What changes are included in this PR? Three stacked commits, each green on its own. Built-in plan serialization is untouched: they keep the inherent `try_from_proto` they have had since 55.0.0, dispatched from their own `PhysicalPlanType` variant. 1. **`feat: add an optional plan_name discriminator to PhysicalExtensionNode`.** `optional string plan_name = 3;` — additive on the binary wire format: old writers omit it and old readers skip it. The generated JSON codec is stricter, see "user-facing changes" below. Nothing reads it yet. 2. **`feat: add ProtoDecoderRegistry, one decoder store for every extension kind`.** A `(trait, name)`-keyed store of type-erased decoders in `datafusion-proto-models`, with no knowledge of any decode context. Nothing uses it yet. 3. **`feat: decode extension ExecutionPlans through a session-scoped registry`.** - An extension plan implements one trait, `ExtensionPlanFromProto`, which pairs the name it is dispatched by with the constructor that rebuilds it. Built-ins do not implement it — they have a wire variant of their own — so `register_execution_plan::<FilterExec>(..)` does not compile, and **no built-in code or caller changes at all**. Its `try_to_proto` writes itself with `ctx.extension_node::<Self>(payload, self.children())`, which builds the `PhysicalExtensionNode` and stamps `Self::NAME`, so the name on the wire and the registry key cannot drift apart. Its `try_from_proto` receives the whole `PhysicalPlanNode`, matches the `Extension` variant, and decodes `inputs` with `ctx.decode_children`. - `ProtoDecoderRegistry` (new, in `datafusion-proto-models`): build one, `register_execution_plan::<T>(&mut registry)` for every extension plan the session must decode, and attach it with the existing `SessionConfig::with_extension`. Registering the same type twice is a no-op (keyed by `TypeId`); a *different* type under a taken name is an error, so collisions surface at registration. `decode_execution_plan` reads the name off the node rather than taking it as an argument, so a caller cannot pair a node with a name it does not carry. - The registry is **keyed by the trait as well as the name**, and is shared by every extension kind DataFusion can decode. This crate contributes only the `dyn ExecutionPlan` face of it. A session therefore carries **one** registry rather than one per kind, a plan and an expression may use the same name without colliding, and apache#24628 registers into the same object. It lives in `datafusion-proto-models` because that crate already sits below every crate that owns one of these traits, behind the same `proto` feature; it stores each decoder type-erased and never names a decode context. - A session carries one registry and `with_extension` replaces it, so the **application** that owns the session builds it. A library exposes `pub fn register(registry: &mut ProtoDecoderRegistry) -> Result<()>` — which can register its plans *and* its expressions — and the application composes the libraries it uses. The libraries need to know nothing about each other and the call order does not matter. There is no `SessionConfig` extension trait and no merge-on-register helper: public API is added when a use needs it, not ahead of one. - It is attached through the `SessionConfig` extension map because `datafusion-execution` sits *below* the crates that own these traits and cannot name them; the extension map is the one type-erased hole through that layering, and it reaches every `TaskContext` derived from the session with no new plumbing. - Decode dispatch in `datafusion-proto`, per the rule above, plus the upgrade-guide entry (in 56.0.0, the guide for the next release). ## What is the testing strategy for this PR? 12 unit tests across `datafusion-proto-models` and `datafusion-physical-plan`, 10 integration tests in `datafusion/proto/tests/cases/plans/extensions.rs`, and doctests on the public traits: - `ProtoDecoderRegistry` itself: 6 unit tests, including the same name under two different traits resolving to two different decoders. - Registration: idempotent for the same type, an error for a different type under the same name, empty names rejected, a name registered for another kind invisible to the `ExecutionPlan` face, and a `compile_fail` doctest for `register_execution_plan::<FilterExec>(..)` (verified to fail on the `ExtensionPlanFromProto` bound, not on anything incidental). - A runnable example, `datafusion-examples/examples/proto/extension_plan_registry.rs`: the same two-crate tree `composed_extension_codec.rs` builds, decoded by name with no composed codec, plus the error a session that forgot to register gets. - A plan round-tripping with `DefaultPhysicalExtensionCodec`, which decodes nothing, so whatever survives came from the registry. - A **session-dependent** plan modeled on datafusion-distributed's `NetworkShuffleExec`: its worker pool is rebuilt from the *decoding* session through `ExecutionPlanDecodeCtx::task_ctx()`, and never rides the wire. - Two independent crates' plans nested inside each other, decoding with no `ComposedPhysicalExtensionCodec`. - The registry taking precedence over a codec that claims every payload, and a registered decoder's failure *not* falling back to that codec. - A codec failing for an unrelated reason: the missing-registration hint is added, and the codec's own error kind still reaches `find_root()`. - Back-compat both ways: codec-encoded plans stay anonymous on the wire and decode as before; a name the session does not know (on a session that *does* have a registry with other plans in it) still decodes if a codec understands the payload; a node re-encoded without the field reads back as unnamed. - A bytes-level `encode_to_vec` / `decode` hop, so tag 3 is exercised as wire bytes and not just as a struct field. Verified locally on the final tree and with `cargo check` on every commit of the stack: the `datafusion-physical-expr`, `datafusion-physical-plan` and `datafusion-proto` test suites, the examples crate, CI's clippy script, rustdoc with private items under `-D warnings`, prettier, typos and fmt. ## Are there any user-facing changes? New public API, all opt-in; no behavior change for existing code. One mechanical break, in the 56.0.0 upgrade guide: `PhysicalExtensionNode` gained a field, so exhaustive struct literals must set `plan_name` (`None` reproduces the previous behavior). Built-in plans are untouched — no signature, trait or import change for anyone calling `FilterExec::try_from_proto(..)`. Two things worth a reviewer's attention: - **JSON is not as forgiving as the binary format.** The generated `pbjson` deserializer rejects a field it does not know, so a *named* node written by a new writer cannot be read as JSON by a reader on an older DataFusion. The binary prost path skips the unknown field as usual, and a plan that does not opt in writes no name and is unaffected. If you serialize plans as JSON across versions, upgrade the readers before you opt a plan in. - **Opting a plan in changes what the codec fallback means for that plan.** Once its `try_to_proto` writes its own payload, a reader that fails to register the type hands that payload to the codec it removed, and a codec that still recognizes the plan may decode the new payload as the old message rather than fail. The upgrade guide says to register migrated plans everywhere their plans are read. 🤖 Generated with [Claude Code](https://claude.com/claude-code) https://claude.ai/code/session_017UzVkTrKmRRc2fKCdMkqBp --------- Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
f3d8556 to
6e88226
Compare
|
@gabotechs this is ready for review if you are interested |
…ExprNode` Additive and proto3-compatible: on the binary wire format old writers omit the field and old readers skip it. The generated JSON codec is stricter — `pbjson` rejects a field it does not know — so a named node written by a new writer cannot be read as JSON by an older reader. An unnamed node is unaffected, and nothing writes a name yet. Nothing reads it either; the codec path keeps writing `None`, since there the codec remains the discriminator. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
6e88226 to
f8b165c
Compare
`PhysicalExtensionExprNode` carried no type discriminator, so the `PhysicalExtensionCodec` *was* the discriminator, with the same problems the plan side has: composition is by registration order, and a name collision between two independent crates is undetectable. An extension expression now implements `ExtensionExprFromProto`, which pairs the name it is dispatched by with the constructor that rebuilds it. Its `try_to_proto` writes itself with `extension_expr_node`, which stamps `Self::NAME`, so the encoded name and the registry key cannot drift apart. `register_physical_expr` puts its decoder in the `ProtoDecoderRegistry` the decoding session carries — the same registry extension `ExecutionPlan`s use, keyed on the trait as well as the name. Only extension expressions implement the trait. A built-in has an `ExprType` variant of its own, keeps its inherent `try_from_proto`, and has no wire name to be registered under — so `register_physical_expr::<Column>(..)` does not compile. Nothing about built-in serialization changes, and no caller needs a new import. `extension_expr_node` is a free function rather than a method on `PhysicalExprEncodeCtx` because that context lives in `datafusion-physical-expr-common`, below the crate that can name `ExtensionExprFromProto` — the same layering that makes the decode context generic over its session. Decode rule: a named node the session has a decoder for goes to that decoder, and a failure there is fatal (falling through would let a codec decode a payload it never wrote); anything else takes the `PhysicalExtensionCodec` chain exactly as before. If the codec fails too, the missing registration is added as *context* on the codec's own error, so the kind the codec chose still reaches a caller that matches on it. `decode_physical_expr` reads the name off the node rather than taking it as an argument, so a caller cannot pair a node with a name it does not carry. Documented in the 56.0.0 upgrade guide; tests in `datafusion/proto/tests/cases/plans/expr_registry.rs`. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
f8b165c to
40473b0
Compare
kosiew
left a comment
There was a problem hiding this comment.
Thanks for working on this. The registry-based decoding approach looks good overall, and I appreciate the backward compatibility with the existing codec path.
I found one issue with preserving expression identity when migrating to the new helper. I've also left two minor suggestions for test coverage and documentation.
| expr_name: Some(T::NAME.to_string()), | ||
| }, | ||
| )), | ||
| ..Default::default() |
There was a problem hiding this comment.
The helper leaves expr_id as None, and the serializer returns hook-generated nodes without adding the expression's ID. This means an extension with expression_id() == Some(42) loses its identity when switching to registry encoding, preventing the deduplicating decoder from restoring shared state. Could we stamp the ID at the common serialization boundary when the hook hasn't supplied one, and add a regression test with two references to an identity-bearing extension that verifies shared state and correctly decoded children?
There was a problem hiding this comment.
Good catch. Fixed in 5dc260a. The serialized now sets expression_id() on any node from a hook that leaves expr_id empty. I added a test for deduping by id.
| /// A whole plan, through `physical_plan_to_bytes` and | ||
| /// `physical_plan_from_bytes`. | ||
| #[test] | ||
| fn registry_expr_survives_a_whole_plan_roundtrip() -> Result<()> { |
There was a problem hiding this comment.
Could we add a follow-up roundtrip test containing both a registered extension plan and expression with the same wire name? The existing registry tests cover name isolation, but this would also exercise both registration paths together in a real plan.
| repeated PhysicalExprNode inputs = 2; | ||
| // Optional type discriminator naming the extension expression, e.g. | ||
| // "my_crate.MyExpr". When present and registered in the session's | ||
| // PhysicalExprRegistry, decoding is routed to that registered |
There was a problem hiding this comment.
This comment refers to PhysicalExprRegistry, but the implementation uses ProtoDecoderRegistry. Could we update the name here and regenerate the corresponding comment in prost.rs?
…dtrip tests - serialize_physical_expr_with_converter now sets the node's expr_id from PhysicalExpr::expression_id when try_to_proto left it unset, so extension expressions built with extension_expr_node keep dedup identity. - Add a dedup regression test and a plan+expr same-wire-name roundtrip test. - Fix stale PhysicalExprRegistry reference in datafusion.proto / prost.rs. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
There was a problem hiding this comment.
Thanks for addressing the earlier review comments. I've rechecked the follow-up changes, and the expression ID preservation, shared-state deduplication, same-name plan/expression roundtrip, and registry documentation corrections all look good. I don't see any remaining correctness blockers.
One non-blocking suggestion: the PR description still references the old generic PhysicalExprDecodeCtx<'a, S>, ExprDecodeSession, a third constructor argument, and a required PhysicalExprFromProto import for built-ins. Could you update the API examples, migration guidance, re-export list, and test count to match the current implementation?
The prerequisite on #24631 and the major-release compatibility considerations still apply. Otherwise, this looks good to me.
|
PR description updated, thanks for pointing it out @kosiew! |
Which issue does this PR close?
PhysicalExprs to decode via a per-type registry instead ofPhysicalExtensionCodec#24626.Rationale for this change
The user-visible problem
A library that ships its own
PhysicalExprhas to write aPhysicalExtensionCodecto serialize it, in a file separate from the expression, and two such libraries on one session have to compose their codecs by hand.PhysicalExtensionExprNodecarries an opaque payload and the encoded children but no type discriminator, so the codec is the discriminator:ComposedPhysicalExtensionCodecwrites the position of the encoding codec into the payload bytes, the decoding side must register the same codecs in the same order, and a name collision between two independent crates cannot be detected.A decoder also had no way to reach session state.
PhysicalExprDecodeCtxhadschema()anddecode()but nothing likeExecutionPlanDecodeCtx::task_ctx(), so an extension expression could not resolve a UDF from the function registry or read session configuration while decoding.Where this sits in the bigger picture
Physical-plan serialization is moving from central
downcast_refchains indatafusion-prototo per-type hooks that live next to each type:try_to_proto(a trait method) plus atry_from_protoconstructor. #22418 did this for built-inPhysicalExprs and #23494 forExecutionPlans. Third-party types could use the encode half but not the decode half. This PR closes the loop forPhysicalExpr; #24631 does the same forExecutionPlanwith the same shape and names (Extension*FromProtopairing the wire name with the constructor,register_*/decode_*facades over one shared registry, attached throughSessionConfig::with_extension).#24631 (merged) added the shared
ProtoDecoderRegistrythat both kinds register into. #26353 (open) addsSessionConfig::update_extension, so that several libraries can add to the session's registry without replacing each other's decoders. This PR does not depend on it.How this relates to
PhysicalExtensionCodecThe codec is not deprecated and this PR does not touch
ComposedPhysicalExtensionCodec. The decode rule is:PhysicalExtensionExprNodewith anexpr_namethe decoding session has registered decodes through the expression's ownExtensionExprFromProto::try_from_proto. A failure there is fatal: falling through to the codec would let it wrongly decode a payload it never wrote.PhysicalExtensionCodec::try_decode_exprexactly as before. If the codec also fails, the error names the missing registration and lists what the session does know.The codec keeps UDF/UDAF/UDWF payloads (a function is an instance, not a type), anything not yet migrated, and the released
FFI_PhysicalExtensionCodecthat datafusion-python exposes.How FFI will work
The registry stores decoders as a private
fnpointer; the public surface isExtensionExprFromProto,extension_expr_nodeand the facade functions. The storage can later grow to carry an FFI vtable and private data without breaking anything shipped here.PhysicalExprNodecrosses an ABI as prost bytes, the patterndatafusion-ffialready uses forStatistics.What changes are included in this PR?
Built-in expression serialization is untouched: built-ins keep their inherent
try_from_proto, dispatched from their ownExprTypevariant, and no built-in code or caller changes.PhysicalExtensionExprNodegainsoptional string expr_name = 3;— additive and proto3-compatible. The codec encode path writesNone.PhysicalExprDecodeCtx::session::<T>() -> Option<&T>returns the session the decode runs under, type-erased becauseTaskContextlives in a crate that depends on this one. It is backed by a new defaultedPhysicalExprDecode::session()method (returnsNone);datafusion-protoandExecutionPlanDecodeCtxsupply theirTaskContext, so a decoder callsctx.session::<TaskContext>().PhysicalExprDecodeCtxis not generic and its constructor is unchanged.expr_idstamping.serialize_physical_expr_with_converternow sets the expression'sexpression_id()on a node returned bytry_to_protowhen the hook leftexpr_idempty (a hook that sets its own keeps it). Without this, an identity-bearing extension written withextension_expr_nodewould lose its identity and defeatDeduplicatingProtoConverter.ExtensionExprFromProto(indatafusion_physical_expr::proto). An extension expression implements this one trait, pairingconst NAMEwithfn try_from_proto(node: &PhysicalExprNode, ctx: &PhysicalExprDecodeCtx<'_>). Itstry_to_protois one call —extension_expr_node::<Self>(ctx, payload, self.children())— which builds the node and stampsSelf::NAME, so the encoded name and the registry key cannot drift apart. Built-ins do not implement it, soregister_physical_expr::<Column>(..)does not compile (acompile_faildoctest checks this).extension_expr_nodeis a free function becausePhysicalExprEncodeCtxlives indatafusion-physical-expr-common, below the crate that can name the trait.ProtoDecoderRegistryfrom feat: allow extensionExecutionPlans to decode via a per-type registry instead ofPhysicalExtensionCodec#24631: callregister_physical_expr::<T>(&mut registry)for every extension expression the session must decode, and attach it withSessionConfig::with_extension. Registering the same type twice is a no-op; a different type under a taken name, or an empty name, is an error. Names are scoped per kind, so a plan and an expression may share a name in one registry. This crate addsregister_physical_expr,physical_expr_names, anddecode_physical_expr(#[doc(hidden)], internal todatafusion-proto; it reads the name off the node so a caller cannot pair a node with a name it does not carry).datafusion::physical_plan::protore-exportsExtensionExprFromProto,extension_expr_node,register_physical_expr,physical_expr_names,decode_physical_expr,PhysicalExprDecodeCtxandPhysicalExprEncodeCtx, alongside theExecutionPlanfacade andProtoDecoderRegistry.datafusion-protoper the rule above (when the codec fallback also fails, the missing registration is added as context on the codec's own error so its kind still reaches callers), a doc pointer onPhysicalExtensionCodec::try_decode_expr, and an upgrade-guide entry in 56.0.0.Why a separate trait rather than a method on
PhysicalExprE0038), which breaks everyArc<dyn PhysicalExpr>, so the wire name cannot live onPhysicalExpr.proto-gated trait isE0046in a downstream crate that never enabled the feature.NotImplementedinstead of a compile error.Decoding is a constructor with no
selfand can never be dispatched through&dyn PhysicalExpr, so the encode/decode asymmetry is structural.Session scoping goes through
SessionConfigextensions rather than a new field onTaskContext/SessionState, which would forcedatafusion-physical-expr-common/protoon for everyone.What is the testing strategy for this PR?
datafusion/proto/tests/cases/plans/expr_registry.rs, 10 tests. ARefusingCodecerrors on every method, so any codec involvement is fatal; a codec that can decode the same payload makes the fallback paths distinguishable from mere failure.ctx.session::<TaskContext>().physical_plan_to_bytes_with_extension_codec/physical_plan_from_bytes_with_extension_codec.expression_id()underDeduplicatingProtoConverter: shared state is preserved and children decode correctly.Are there any user-facing changes?
Additive. Nobody is affected unless they opt in; there are no breaking API changes. Documented in the 56.0.0 upgrade guide.
Two things worth a reviewer's attention:
exprNameneeds a reader built after this change.try_to_protowrites its own named node, a reader that fails to register the type hands that payload totry_decode_expr, and a codec that still recognizes the expression may decode the new payload as the old message. The upgrade guide says to register migrated expressions on every reader before writers start using them.🤖 Generated with Claude Code