Add PhysicalExprDecodeCtx and migrate Column decode

Adds the decode-side counterpart to `PhysicalExprEncodeCtx`. The public
surface is a single `decode(node)` method on the context — the central
match (and, in future, third-party registry lookups) live behind it.

Per-expression contract:

```rust
impl Column {
    pub fn try_from_proto(
        node: &PhysicalExprNode,
        ctx: &PhysicalExprDecodeCtx,
    ) -> Result<Arc<dyn PhysicalExpr>>;
}
```

`try_from_proto` takes the whole `PhysicalExprNode` — the exact inverse of
what `try_to_proto` returns — and unwraps its own `ExprType` variant. That
keeps one signature for every expression (trait-ready) and gives decoders
access to outer-node fields such as `expr_id`. The central match in
`parse_physical_expr_with_converter` only routes to the right constructor.

`Column` is the first expression to migrate. Other variants stay inline;
they migrate in follow-ups, same shape, same trick (one expression per
commit, central match keeps shrinking).

Removes the temporary `FromProto<&PhysicalColumn> for Column` impl
introduced in the proto-models extraction — `Column` now owns both
directions of its own serialization.

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

Author: adriangb

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

@@ -566,6 +566,89 @@ pub mod proto_encode {
     }
 }
 
+/// 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.
+#[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(
     since = "50.0.0",
     note = "Use `datafusion_expr_common::dyn_eq` instead"

datafusion/physical-expr/src/expressions/column.rs

@@ -165,6 +165,32 @@ impl PhysicalExpr for Column {
     }
 }
 
+#[cfg(feature = "proto")]
+impl Column {
+    /// Reconstruct a [`Column`] from its protobuf representation.
+    ///
+    /// Takes the whole [`PhysicalExprNode`] — the exact inverse of what
+    /// [`PhysicalExpr::try_to_proto`] produces — so every expression's
+    /// `try_from_proto` shares one signature. The decode context is currently
+    /// unused, but is threaded through so that future expressions with child
+    /// sub-expressions can recurse via [`PhysicalExprDecodeCtx::decode`].
+    ///
+    /// [`PhysicalExprNode`]: datafusion_proto_models::protobuf::PhysicalExprNode
+    /// [`PhysicalExpr::try_to_proto`]: datafusion_physical_expr_common::physical_expr::PhysicalExpr::try_to_proto
+    /// [`PhysicalExprDecodeCtx::decode`]: datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx::decode
+    pub fn try_from_proto(
+        node: &datafusion_proto_models::protobuf::PhysicalExprNode,
+        _ctx: &datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx<'_>,
+    ) -> Result<Arc<dyn PhysicalExpr>> {
+        use datafusion_proto_models::protobuf;
+        let protobuf::PhysicalColumn { name, index } = match &node.expr_type {
+            Some(protobuf::physical_expr_node::ExprType::Column(c)) => c,
+            _ => return internal_err!("PhysicalExprNode is not a Column"),
+        };
+        Ok(Arc::new(Column::new(name, *index as usize)))
+    }
+}
+
 impl Column {
     fn bounds_check(&self, input_schema: &Schema) -> Result<()> {
         if self.index < input_schema.fields.len() {

datafusion/proto/src/physical_plan/from_proto.rs

@@ -57,20 +57,14 @@ use super::{
     DefaultPhysicalProtoConverter, PhysicalExtensionCodec, PhysicalPlanDecodeContext,
     PhysicalProtoConverterExtension,
 };
-use crate::convert::{FromProto, TryFromProto};
+use crate::convert::TryFromProto;
 use crate::logical_plan::{self};
 use crate::protobuf::physical_expr_node::ExprType;
 use crate::{convert_required, convert_required_proto, protobuf};
 use datafusion_physical_expr::expressions::{
     DynamicFilterInner, DynamicFilterPhysicalExpr,
 };
 
-impl FromProto<&protobuf::PhysicalColumn> for Column {
-    fn from_proto(c: &protobuf::PhysicalColumn) -> Column {
-        Column::new(&c.name, c.index as usize)
-    }
-}
-
 /// Parses a physical sort expression from a protobuf.
 ///
 /// # Arguments
@@ -267,11 +261,24 @@ pub fn parse_physical_expr_with_converter(
         .as_ref()
         .ok_or_else(|| proto_error("Unexpected empty physical expression"))?;
 
+    // Decoder context handed to per-expression `try_from_proto` constructors.
+    // This is the new shape the codebase is migrating toward (see #21835);
+    // for now only `Column` is migrated and the rest of the variants are still
+    // matched inline.
+    let decoder = ConverterDecoder {
+        ctx,
+        proto_converter,
+    };
+    let decode_ctx =
+        datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx::new(
+            input_schema,
+            &decoder,
+        );
+
     let pexpr: Arc<dyn PhysicalExpr> = match expr_type {
-        ExprType::Column(c) => {
-            let pcol = Column::from_proto(c);
-            Arc::new(pcol)
-        }
+        // Migrated expressions take the whole `PhysicalExprNode` and unwrap
+        // their own `ExprType` variant (see #21835); this match only routes.
+        ExprType::Column(_) => Column::try_from_proto(proto, &decode_ctx)?,
         ExprType::UnknownColumn(c) => Arc::new(UnKnownColumn::new(&c.name)),
         ExprType::Literal(scalar) => Arc::new(Literal::new(scalar.try_into()?)),
         ExprType::BinaryExpr(binary_expr) => {
@@ -905,6 +912,33 @@ impl TryFromProto<&protobuf::FileSinkConfig> for FileSinkConfig {
     }
 }
 
+/// Concrete [`PhysicalExprDecode`] driver that backs
+/// [`PhysicalExprDecodeCtx`] inside `parse_physical_expr_with_converter`.
+///
+/// Today this is a thin wrapper that re-enters the central match through
+/// `proto_to_physical_expr`; once more expressions migrate, the central match
+/// shrinks and a future builder-style decoder can take over.
+///
+/// [`PhysicalExprDecode`]: datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecode
+/// [`PhysicalExprDecodeCtx`]: datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx
+struct ConverterDecoder<'a, 'b> {
+    ctx: &'a PhysicalPlanDecodeContext<'b>,
+    proto_converter: &'a dyn PhysicalProtoConverterExtension,
+}
+
+impl datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecode
+    for ConverterDecoder<'_, '_>
+{
+    fn decode(
+        &self,
+        node: &protobuf::PhysicalExprNode,
+        schema: &Schema,
+    ) -> Result<Arc<dyn PhysicalExpr>> {
+        self.proto_converter
+            .proto_to_physical_expr(node, schema, self.ctx)
+    }
+}
+
 #[cfg(test)]
 mod tests {