feat(physical-expr-common): add proto helpers for variant match, outer node, required scalar field

Expressions migrating to `try_to_proto` / `try_from_proto` (#22418)
keep hand-rolling three shapes that don't fit the existing
`encode_child` / `decode_required_expression` helpers from #22513:

- the outer `match &node.expr_type { ... }` that opens every
  `try_from_proto`,
- the `PhysicalExprNode { expr_id: None, expr_type: Some(_) }`
  wrapper every built-in `try_to_proto` returns, and
- the `ok_or_else(|| internal_datafusion_err!("X is missing required
  field 'Y'"))` for non-expression fields like `arrow_type` on
  `CastExpr` / `TryCastExpr`.

This commit adds three thin helpers, all gated on `feature = "proto"`:

- `expect_expr_variant!` macro — matches the outer `Option<ExprType>`
  and returns the inner payload (auto-derefs through `Box`), or
  returns an `internal_err!` naming the expected variant.
- `proto_encode::expr_node` — constructs a `PhysicalExprNode` with no
  `expr_id` set from an `ExprType`. `expr_id` is a dedup tag used
  only by `DynamicFilterPhysicalExpr` (#22434); every other
  expression's `try_to_proto` reduces to `expr_node(ExprType::Foo(_))`.
- `proto_decode::require_proto_field<T>` — mirrors
  `decode_required_expression` for non-`PhysicalExprNode` fields,
  keeping the "missing required field" message format in one place.

No existing call sites are migrated in this commit; the follow-up
ports the already-migrated expressions onto these helpers.

Author: adriangb

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

@@ -576,6 +576,31 @@ pub mod proto_encode {
         }
     }
 
+    /// Build a [`PhysicalExprNode`] with no `expr_id` set.
+    ///
+    /// `expr_id` is a dedup tag used by node-sharing encoders (today only
+    /// [`DynamicFilterPhysicalExpr`], see #22434); every other
+    /// `PhysicalExpr` emits a node with `expr_id: None`. So
+    /// `try_to_proto` for those reduces to:
+    ///
+    /// ```ignore
+    /// Ok(Some(expr_node(ExprType::Cast(Box::new(PhysicalCastNode { .. })))))
+    /// ```
+    ///
+    /// If your expression *does* participate in dedup, set
+    /// [`PhysicalExprNode::expr_id`] on the returned value or construct the
+    /// node literally.
+    ///
+    /// [`DynamicFilterPhysicalExpr`]: https://docs.rs/datafusion-physical-expr/latest/datafusion_physical_expr/expressions/struct.DynamicFilterPhysicalExpr.html
+    pub fn expr_node(
+        expr_type: datafusion_proto_models::protobuf::physical_expr_node::ExprType,
+    ) -> PhysicalExprNode {
+        PhysicalExprNode {
+            expr_id: None,
+            expr_type: Some(expr_type),
+        }
+    }
+
     /// Internal dispatch trait. Implementors live in `datafusion-proto` and
     /// wrap the existing `PhysicalExtensionCodec` +
     /// `PhysicalProtoConverterExtension` plumbing. Expression authors should
@@ -619,6 +644,48 @@ pub mod proto_decode {
 
     use super::PhysicalExpr;
 
+    /// Open the outer [`PhysicalExprNode`] and assert it carries the expected
+    /// `ExprType` variant, returning the inner payload (auto-derefs through
+    /// `Box`) or bailing with an `Internal` error.
+    ///
+    /// Every `try_from_proto` starts with the same six-line `match`:
+    ///
+    /// ```ignore
+    /// let try_cast = match &node.expr_type {
+    ///     Some(protobuf::physical_expr_node::ExprType::TryCast(x)) => x.as_ref(),
+    ///     _ => return internal_err!("PhysicalExprNode is not a TryCastExpr"),
+    /// };
+    /// ```
+    ///
+    /// With this macro that collapses to:
+    ///
+    /// ```ignore
+    /// let try_cast = expect_expr_variant!(
+    ///     node,
+    ///     protobuf::physical_expr_node::ExprType::TryCast,
+    ///     "TryCastExpr",
+    /// );
+    /// ```
+    ///
+    /// Pass the variant as a `::` path so the macro stays agnostic to how
+    /// the caller imports the proto types.
+    #[macro_export]
+    macro_rules! expect_expr_variant {
+        ($node:expr, $variant:path, $expr_name:literal $(,)?) => {{
+            match &$node.expr_type {
+                ::core::option::Option::Some($variant(inner)) => inner,
+                _ => {
+                    return ::datafusion_common::internal_err!(concat!(
+                        "PhysicalExprNode is not a ",
+                        $expr_name
+                    ));
+                }
+            }
+        }};
+    }
+    #[doc(inline)]
+    pub use expect_expr_variant;
+
     /// Decoder context handed to per-expression `try_from_proto` constructors.
     ///
     /// Wraps an internal [`PhysicalExprDecode`] trait object plus a borrowed
@@ -693,6 +760,33 @@ pub mod proto_decode {
         }
     }
 
+    /// Unwrap a required non-expression proto field.
+    ///
+    /// Mirrors [`PhysicalExprDecodeCtx::decode_required_expression`] for proto
+    /// fields that aren't [`PhysicalExprNode`]s — e.g. the `arrow_type` of a
+    /// `PhysicalCastNode` or the `scalar` of a `PhysicalLiteralNode`. Keeps
+    /// the "missing required field" message format identical across
+    /// expressions:
+    ///
+    /// ```ignore
+    /// let arrow_type = require_proto_field(
+    ///     cast_expr.arrow_type.as_ref(),
+    ///     "CastExpr",
+    ///     "arrow_type",
+    /// )?;
+    /// ```
+    pub fn require_proto_field<T>(
+        opt: Option<T>,
+        expr_name: &str,
+        field: &str,
+    ) -> Result<T> {
+        opt.ok_or_else(|| {
+            datafusion_common::internal_datafusion_err!(
+                "{expr_name} is missing required field '{field}'"
+            )
+        })
+    }
+
     /// Internal dispatch trait. Implementors live in `datafusion-proto`.
     /// Expression authors should use [`PhysicalExprDecodeCtx`] instead of
     /// calling this directly.
@@ -1143,3 +1237,97 @@ mod test {
         );
     }
 }
+
+#[cfg(all(test, feature = "proto"))]
+mod proto_helper_tests {
+    use datafusion_common::DataFusionError;
+    use datafusion_proto_models::protobuf::{
+        self, PhysicalColumn, PhysicalExprNode, physical_expr_node,
+    };
+
+    use crate::expect_expr_variant;
+    use crate::physical_expr::proto_decode::require_proto_field;
+    use crate::physical_expr::proto_encode::expr_node;
+
+    fn column_node() -> PhysicalExprNode {
+        PhysicalExprNode {
+            expr_id: None,
+            expr_type: Some(physical_expr_node::ExprType::Column(PhysicalColumn {
+                name: "a".to_string(),
+                index: 0,
+            })),
+        }
+    }
+
+    #[test]
+    fn expr_node_sets_no_expr_id() {
+        let node = expr_node(physical_expr_node::ExprType::Column(PhysicalColumn {
+            name: "a".to_string(),
+            index: 0,
+        }));
+        assert!(node.expr_id.is_none());
+        assert!(matches!(
+            node.expr_type,
+            Some(physical_expr_node::ExprType::Column(_))
+        ));
+    }
+
+    #[test]
+    fn require_proto_field_returns_inner() {
+        let v = require_proto_field(Some(7_u32), "FooExpr", "answer").unwrap();
+        assert_eq!(v, 7);
+    }
+
+    #[test]
+    fn require_proto_field_reports_missing() {
+        let err = require_proto_field::<u32>(None, "FooExpr", "answer").unwrap_err();
+        assert!(matches!(
+            err,
+            DataFusionError::Internal(msg)
+                if msg.contains("FooExpr is missing required field 'answer'")
+        ));
+    }
+
+    fn expect_column(
+        node: &PhysicalExprNode,
+    ) -> Result<&PhysicalColumn, DataFusionError> {
+        let inner =
+            expect_expr_variant!(node, physical_expr_node::ExprType::Column, "Column",);
+        Ok(inner)
+    }
+
+    #[test]
+    fn expect_expr_variant_returns_inner_payload() {
+        let node = column_node();
+        let col = expect_column(&node).unwrap();
+        assert_eq!(col.name, "a");
+    }
+
+    #[test]
+    fn expect_expr_variant_rejects_wrong_variant() {
+        let node = PhysicalExprNode {
+            expr_id: None,
+            expr_type: Some(physical_expr_node::ExprType::Negative(Box::new(
+                protobuf::PhysicalNegativeNode { expr: None },
+            ))),
+        };
+        let err = expect_column(&node).unwrap_err();
+        assert!(matches!(
+            err,
+            DataFusionError::Internal(msg) if msg.contains("PhysicalExprNode is not a Column")
+        ));
+    }
+
+    #[test]
+    fn expect_expr_variant_rejects_missing_expr_type() {
+        let node = PhysicalExprNode {
+            expr_id: None,
+            expr_type: None,
+        };
+        let err = expect_column(&node).unwrap_err();
+        assert!(matches!(
+            err,
+            DataFusionError::Internal(msg) if msg.contains("PhysicalExprNode is not a Column")
+        ));
+    }
+}