Split proto serialization to encapsulate private state (#21835) (#21929)

## Which issue does this PR close?

- Closes #21835.

## Rationale for this change

`datafusion-proto` serializes every built-in `PhysicalExpr` through a
single
~300-line `downcast_ref` chain, with a mirror `match` on the decode
side. That
chain lives outside the crate where each expression is defined, so every
field
an expression wants to round-trip has to be made `pub`. #21807 is the
cautionary tale: it had to add five `pub` "proto-only, not stable" items
to
`DynamicFilterPhysicalExpr` just to serialize an `RwLock`-wrapped inner.

This PR adds the infrastructure so a `PhysicalExpr` can serialize itself
and
keep its state private.

## What changes are included in this PR?

A `PhysicalExpr` can now opt into serializing itself, in both
directions:

```rust
fn try_to_proto(&self, ctx: &PhysicalExprEncodeCtx) -> Result<Option<PhysicalExprNode>>
fn try_from_proto(node: &PhysicalExprNode, ctx: &PhysicalExprDecodeCtx) -> Result<Arc<dyn PhysicalExpr>>
```

`try_to_proto` returning `Ok(None)` (the default) means "fall through to
the
old downcast chain", so the change is purely additive — nothing is
forced to
migrate.

`Column` and `BinaryExpr` are migrated as working demos; everything
else stays on the old path and migrates later, one expression at a time,
with
no wire-format change.

Five stacked commits, each builds green on its own and is independently
reviewable (or splittable into its own PR):

1. **Extract `datafusion-proto-models` crate** — move the `.proto` file
and
   prost-generated types into a lightweight crate (mirrors the existing
   `datafusion-proto-common` split).
2. **Add the `try_to_proto` hook** — feature-gated, off by default.
3. **Migrate `Column` encode.**
4. **Add the decode side and migrate `Column` decode.**
5. **Migrate `BinaryExpr`** (both directions).

## A few design decisions worth flagging

- **`FromProto` / `TryFromProto` traits instead of plain `From` /
`TryFrom`.**
  Once the prost types move into their own crate they are *foreign* to
`datafusion-proto`, and the orphan rule forbids `impl From<&protobuf::X>
for Y`
when both `X` and `Y` are foreign. So those conversions become
`FromProto` /
`TryFromProto` traits in `datafusion_proto::convert`, and callers go
from
`(&x).into()` to `Y::from_proto(&x)`. This is a known workaround, not
the end
  state — see Future work.
- **The ctx is a concrete struct, not `&dyn`.** `PhysicalExprEncodeCtx`
/
`PhysicalExprDecodeCtx` wrap a sealed dispatch trait. Keeping them
concrete
keeps `&dyn` out of every expression's signature and gives a stable
place to
add helpers (UDF encoding, registry hooks) later without churning a
public
  trait.
- **`try_from_proto` takes the whole `PhysicalExprNode`**, not the
pre-unwrapped variant payload, so every expression's decoder has the
same
  signature and can still see outer-node fields like `expr_id`.

## Are these changes tested?

No new behavior, so no new tests. `Column` and `BinaryExpr` produce and
consume
the same wire format as before; the existing `roundtrip_physical_plan` /
`roundtrip_physical_expr` tests already cover both directions and now
exercise
the new path.

## Are there any user-facing changes?

Small API breaks in `datafusion-proto`:

- `try_from_physical_plan_with_converter` /
`try_into_physical_plan_with_converter`
  move to a `PhysicalPlanNodeExt` trait — callers add
  `use datafusion_proto::physical_plan::PhysicalPlanNodeExt;`.
- Foreign-foreign `From` / `TryFrom` conversions become `FromProto` /
  `TryFromProto` (see Design decisions above).
- `datafusion_proto::generated::*` is deprecated in favor of
  `datafusion_proto::protobuf`; it still works.

The new `proto` feature on `datafusion-physical-expr(-common)` is off by
default, so crates that don't serialize plans pay nothing.

## Future work

- Migrate the remaining built-in expressions — including
`DynamicFilterPhysicalExpr`, the original motivation — one per follow-up
PR.
- Apply the same pattern to `ExecutionPlan` serialization.
- Drop the `FromProto` / `TryFromProto` workaround: collapse
  `datafusion-proto-common` into `datafusion-proto-models` and push the
conversion impls down to the target-type crates so callers use plain
`From` /
`TryFrom` again. Full dep-graph analysis and a step-by-step plan are in
[#21835
(comment)](#21835 (comment)).

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: adriangb

.github/workflows/rust.yml

@@ -605,7 +605,6 @@ jobs:
           rust-version: stable
       - name: Run
         run: |
-          echo '' > datafusion/proto/src/generated/datafusion.rs
           ci/scripts/rust_fmt.sh
 
   # Coverage job disabled due to

Cargo.lock

@@ -47,9 +47,10 @@ members = [
     "datafusion/pruning",
     "datafusion/physical-plan",
     "datafusion/proto",
-    "datafusion/proto/gen",
     "datafusion/proto-common",
     "datafusion/proto-common/gen",
+    "datafusion/proto-models",
+    "datafusion/proto-models/gen",
     "datafusion/session",
     "datafusion/spark",
     "datafusion/sql",
@@ -152,6 +153,7 @@ datafusion-physical-optimizer = { path = "datafusion/physical-optimizer", versio
 datafusion-physical-plan = { path = "datafusion/physical-plan", version = "53.1.0" }
 datafusion-proto = { path = "datafusion/proto", version = "53.1.0" }
 datafusion-proto-common = { path = "datafusion/proto-common", version = "53.1.0" }
+datafusion-proto-models = { path = "datafusion/proto-models", version = "53.1.0" }
 datafusion-pruning = { path = "datafusion/pruning", version = "53.1.0" }
 datafusion-session = { path = "datafusion/session", version = "53.1.0" }
 datafusion-spark = { path = "datafusion/spark", version = "53.1.0" }

Cargo.toml

@@ -62,7 +62,8 @@ use datafusion_proto::bytes::{
 use datafusion_proto::physical_plan::from_proto::parse_physical_expr_with_converter;
 use datafusion_proto::physical_plan::to_proto::serialize_physical_expr_with_converter;
 use datafusion_proto::physical_plan::{
-    PhysicalExtensionCodec, PhysicalPlanDecodeContext, PhysicalProtoConverterExtension,
+    PhysicalExtensionCodec, PhysicalPlanDecodeContext, PhysicalPlanNodeExt,
+    PhysicalProtoConverterExtension,
 };
 use datafusion_proto::protobuf::physical_plan_node::PhysicalPlanType;
 use datafusion_proto::protobuf::{

datafusion-examples/examples/custom_data_source/adapter_serialization.rs

@@ -52,7 +52,7 @@ use datafusion_proto::physical_plan::from_proto::parse_physical_expr_with_conver
 use datafusion_proto::physical_plan::to_proto::serialize_physical_expr_with_converter;
 use datafusion_proto::physical_plan::{
     DefaultPhysicalExtensionCodec, PhysicalExtensionCodec, PhysicalPlanDecodeContext,
-    PhysicalProtoConverterExtension,
+    PhysicalPlanNodeExt, PhysicalProtoConverterExtension,
 };
 use datafusion_proto::protobuf::{PhysicalExprNode, PhysicalPlanNode};
 use prost::Message;

datafusion-examples/examples/proto/expression_deduplication.rs

@@ -390,6 +390,50 @@ impl Operator {
             | Operator::StringConcat => false,
         }
     }
+
+    /// Parse an `Operator` from the string name `datafusion-proto` uses on the
+    /// wire (the `Debug` name of the variant, e.g. `"Eq"`).
+    ///
+    /// Returns `None` for names with no binary-operator counterpart. This is
+    /// the canonical proto-string mapping, shared by `datafusion-proto`
+    /// (logical plans) and `PhysicalExpr` decoders such as `BinaryExpr`, so the
+    /// mapping is not duplicated across crates.
+    pub fn from_proto_name(name: &str) -> Option<Operator> {
+        Some(match name {
+            "And" => Operator::And,
+            "Or" => Operator::Or,
+            "Eq" => Operator::Eq,
+            "NotEq" => Operator::NotEq,
+            "LtEq" => Operator::LtEq,
+            "Lt" => Operator::Lt,
+            "Gt" => Operator::Gt,
+            "GtEq" => Operator::GtEq,
+            "Plus" => Operator::Plus,
+            "Minus" => Operator::Minus,
+            "Multiply" => Operator::Multiply,
+            "Divide" => Operator::Divide,
+            "Modulo" => Operator::Modulo,
+            "IsDistinctFrom" => Operator::IsDistinctFrom,
+            "IsNotDistinctFrom" => Operator::IsNotDistinctFrom,
+            "BitwiseAnd" => Operator::BitwiseAnd,
+            "BitwiseOr" => Operator::BitwiseOr,
+            "BitwiseXor" => Operator::BitwiseXor,
+            "BitwiseShiftLeft" => Operator::BitwiseShiftLeft,
+            "BitwiseShiftRight" => Operator::BitwiseShiftRight,
+            "RegexIMatch" => Operator::RegexIMatch,
+            "RegexMatch" => Operator::RegexMatch,
+            "RegexNotIMatch" => Operator::RegexNotIMatch,
+            "RegexNotMatch" => Operator::RegexNotMatch,
+            "LikeMatch" => Operator::LikeMatch,
+            "ILikeMatch" => Operator::ILikeMatch,
+            "NotLikeMatch" => Operator::NotLikeMatch,
+            "NotILikeMatch" => Operator::NotILikeMatch,
+            "StringConcat" => Operator::StringConcat,
+            "AtArrow" => Operator::AtArrow,
+            "ArrowAt" => Operator::ArrowAt,
+            _ => return None,
+        })
+    }
 }
 
 impl fmt::Display for Operator {

datafusion/expr-common/src/operator.rs

@@ -40,11 +40,18 @@ workspace = true
 [lib]
 name = "datafusion_physical_expr_common"
 
+[features]
+default = []
+# Enables the `PhysicalExpr::to_proto` hook used by `datafusion-proto`.
+# Off by default so crates that never serialize plans pay nothing.
+proto = ["dep:datafusion-proto-models"]
+
 [dependencies]
 arrow = { workspace = true }
 chrono = { workspace = true }
 datafusion-common = { workspace = true }
 datafusion-expr-common = { workspace = true }
+datafusion-proto-models = { workspace = true, optional = true }
 hashbrown = { workspace = true }
 indexmap = { workspace = true }
 itertools = { workspace = true }

datafusion/physical-expr-common/Cargo.toml

@@ -477,6 +477,180 @@ pub trait PhysicalExpr: Any + Send + Sync + Display + Debug + DynEq + DynHash {
     fn expression_id(&self) -> Option<u64> {
         None
     }
+
+    /// Serialize this expression to a [`PhysicalExprNode`] proto message.
+    ///
+    /// Returning `Ok(None)` means "this expression does not know how to
+    /// serialize itself"; the caller (typically `datafusion-proto`) will fall
+    /// back to its existing codec / extension paths. This matches today's
+    /// behavior for expressions that aren't built into `datafusion-proto`.
+    ///
+    /// Returning `Ok(Some(node))` means the expression has serialized itself
+    /// fully; the caller should not try any further fallback path.
+    ///
+    /// Returning `Err(_)` means a real serialization failure (e.g. the
+    /// expression knows it should serialize but a child failed).
+    ///
+    /// The motivating use case is letting expressions with private state
+    /// (e.g. `DynamicFilterPhysicalExpr`'s `RwLock`-protected inner fields)
+    /// reach into their own internals for `try_to_proto`/`try_from_proto`
+    /// without having to expose `pub` accessors to `datafusion-proto`. See
+    /// <https://github.com/apache/datafusion/issues/21835>.
+    ///
+    /// The `try_` prefix matches the fallible `try_from_proto` decode
+    /// constructors (and the `TryFromProto` trait in `datafusion-proto`);
+    /// both sides of the round-trip are fallible and named consistently.
+    ///
+    /// [`PhysicalExprNode`]: datafusion_proto_models::protobuf::PhysicalExprNode
+    #[cfg(feature = "proto")]
+    fn try_to_proto(
+        &self,
+        _ctx: &proto_encode::PhysicalExprEncodeCtx<'_>,
+    ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalExprNode>> {
+        Ok(None)
+    }
+}
+
+/// Encode-side context for [`PhysicalExpr::try_to_proto`].
+///
+/// Expression authors only ever see [`proto_encode::PhysicalExprEncodeCtx`]:
+/// a concrete struct with stable methods. Internally it dispatches to a
+/// [`proto_encode::PhysicalExprEncode`] implementor that lives in
+/// `datafusion-proto`, which is what lets `physical-expr-common` stay free
+/// of `datafusion-proto` as a dep.
+///
+/// More specialized helpers (e.g. encoding UDFs/UDAFs/UDWFs through the
+/// extension codec) can be added to the context as expressions migrate;
+/// today they're not required because the encoder forwards to the existing
+/// codec via the proto converter.
+#[cfg(feature = "proto")]
+pub mod proto_encode {
+    use std::sync::Arc;
+
+    use datafusion_common::Result;
+    use datafusion_proto_models::protobuf::PhysicalExprNode;
+
+    use super::PhysicalExpr;
+
+    /// Encoder context handed to [`super::PhysicalExpr::try_to_proto`].
+    ///
+    /// Wraps an internal [`PhysicalExprEncode`] trait object so callers see a
+    /// stable concrete type while implementations can evolve in
+    /// `datafusion-proto`.
+    pub struct PhysicalExprEncodeCtx<'a> {
+        encoder: &'a dyn PhysicalExprEncode,
+    }
+
+    impl<'a> PhysicalExprEncodeCtx<'a> {
+        /// Construct a new encode context. Typically called by
+        /// `datafusion-proto`; expression authors receive `&PhysicalExprEncodeCtx`.
+        pub fn new(encoder: &'a dyn PhysicalExprEncode) -> Self {
+            Self { encoder }
+        }
+
+        /// Encode a child expression. Routes through the configured encoder
+        /// so dedup-aware encoding is preserved.
+        pub fn encode_child(
+            &self,
+            expr: &Arc<dyn PhysicalExpr>,
+        ) -> Result<PhysicalExprNode> {
+            self.encoder.encode(expr)
+        }
+    }
+
+    /// Internal dispatch trait. Implementors live in `datafusion-proto` and
+    /// wrap the existing `PhysicalExtensionCodec` +
+    /// `PhysicalProtoConverterExtension` plumbing. Expression authors should
+    /// use [`PhysicalExprEncodeCtx`] instead of calling this directly.
+    pub trait PhysicalExprEncode {
+        /// Encode an expression to a protobuf node.
+        fn encode(&self, expr: &Arc<dyn PhysicalExpr>) -> Result<PhysicalExprNode>;
+    }
+}
+
+/// Decode-side counterpart to [`proto_encode`].
+///
+/// Expression authors implement an associated `try_from_proto` on their
+/// concrete type, with the signature
+///
+/// ```ignore
+/// fn try_from_proto(
+///     node: &PhysicalExprNode,
+///     ctx: &PhysicalExprDecodeCtx<'_>,
+/// ) -> Result<Arc<dyn PhysicalExpr>>
+/// ```
+///
+/// It takes the whole [`PhysicalExprNode`] — the exact inverse of what
+/// [`PhysicalExpr::try_to_proto`] returns — so the constructor can also see
+/// outer-node fields such as `expr_id`. The central match in
+/// `datafusion-proto` dispatches `ExprType` variants to these constructors.
+///
+/// As with the encode side, the public surface is a struct (not a `&dyn`
+/// trait) so future fields/helpers (registries for third-party expressions,
+/// schema-resolution caches, etc.) can be added without changing the
+/// signature every expression depends on.
+///
+/// [`PhysicalExprNode`]: datafusion_proto_models::protobuf::PhysicalExprNode
+#[cfg(feature = "proto")]
+pub mod proto_decode {
+    use std::sync::Arc;
+
+    use arrow::datatypes::Schema;
+    use datafusion_common::Result;
+    use datafusion_proto_models::protobuf::PhysicalExprNode;
+
+    use super::PhysicalExpr;
+
+    /// Decoder context handed to per-expression `try_from_proto` constructors.
+    ///
+    /// Wraps an internal [`PhysicalExprDecode`] trait object plus a borrowed
+    /// schema. The trait stays an implementation detail of `datafusion-proto`;
+    /// expression authors only see this struct.
+    pub struct PhysicalExprDecodeCtx<'a> {
+        schema: &'a Schema,
+        decoder: &'a dyn PhysicalExprDecode,
+    }
+
+    impl<'a> PhysicalExprDecodeCtx<'a> {
+        /// Construct a new decode context. Typically called by
+        /// `datafusion-proto`; expression authors receive
+        /// `&PhysicalExprDecodeCtx`.
+        pub fn new(schema: &'a Schema, decoder: &'a dyn PhysicalExprDecode) -> Self {
+            Self { schema, decoder }
+        }
+
+        /// The schema bound to this decode context. Use it for column lookups,
+        /// data-type resolution, etc.
+        pub fn schema(&self) -> &Schema {
+            self.schema
+        }
+
+        /// Decode an expression node, recursing into child sub-expressions.
+        ///
+        /// Routes built-in `ExprType` variants through `datafusion-proto`'s
+        /// central match and forwards extension nodes to the registered codec
+        /// (today via [`PhysicalExtensionCodec::try_decode_expr`]; later via
+        /// a per-type registry — see #21835).
+        ///
+        /// [`PhysicalExtensionCodec::try_decode_expr`]: https://docs.rs/datafusion-proto/latest/datafusion_proto/physical_plan/trait.PhysicalExtensionCodec.html#method.try_decode_expr
+        pub fn decode(&self, node: &PhysicalExprNode) -> Result<Arc<dyn PhysicalExpr>> {
+            self.decoder.decode(node, self.schema)
+        }
+    }
+
+    /// Internal dispatch trait. Implementors live in `datafusion-proto`.
+    /// Expression authors should use [`PhysicalExprDecodeCtx`] instead of
+    /// calling this directly.
+    pub trait PhysicalExprDecode {
+        /// Decode a proto node into a concrete `PhysicalExpr`. The schema is
+        /// passed alongside so implementations can support recursive children
+        /// and rebind the context per call (e.g. for nested plans).
+        fn decode(
+            &self,
+            node: &PhysicalExprNode,
+            schema: &Schema,
+        ) -> Result<Arc<dyn PhysicalExpr>>;
+    }
 }
 
 #[deprecated(

datafusion/physical-expr-common/src/physical_expr.rs

@@ -42,6 +42,12 @@ name = "datafusion_physical_expr"
 
 [features]
 recursive_protection = ["dep:recursive"]
+# Forwards the `proto` feature to `datafusion-physical-expr-common`, exposing
+# `PhysicalExpr::to_proto` and letting expressions in this crate implement it.
+proto = [
+    "dep:datafusion-proto-models",
+    "datafusion-physical-expr-common/proto",
+]
 
 [dependencies]
 arrow = { workspace = true }
@@ -50,6 +56,7 @@ datafusion-expr = { workspace = true }
 datafusion-expr-common = { workspace = true }
 datafusion-functions-aggregate-common = { workspace = true }
 datafusion-physical-expr-common = { workspace = true }
+datafusion-proto-models = { workspace = true, optional = true }
 hashbrown = { workspace = true }
 indexmap = { workspace = true }
 itertools = { workspace = true, features = ["use_std"] }