feat(physical-expr-common): add list/required expr helpers to proto ctx

Expressions migrating to the `try_to_proto`/`try_from_proto` hooks (see
#21835, #22418) repeatedly hand-roll the same boilerplate the central
`datafusion-proto` match already factors out via `parse_physical_exprs`,
`parse_required_physical_expr`, and `serialize_physical_exprs`. Those
free functions can't be reused by expression authors — they take
`PhysicalProtoConverterExtension`/`PhysicalPlanDecodeContext`, which the
ctx structs deliberately hide.

Add the same shapes as thin methods on the ctx structs, built on the
existing `encode_child`/`decode` primitives:

- `PhysicalExprEncodeCtx::encode_children_expressions`
- `PhysicalExprDecodeCtx::decode_required_expression`
- `PhysicalExprDecodeCtx::decode_children_expressions`

`decode_required_expression` takes the expression name and proto field
so its "missing required field" error still names where the error is
(e.g. `InListExpr is missing required field 'expr'`), while keeping the
format in one place.

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

Author: adriangb

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

@@ -556,6 +556,24 @@ pub mod proto_encode {
         ) -> Result<PhysicalExprNode> {
             self.encoder.encode(expr)
         }
+
+        /// Encode a sequence of child expressions, preserving order.
+        ///
+        /// Convenience wrapper over [`Self::encode_child`] for expressions
+        /// holding a `repeated` proto field (e.g. the `list` of an `InList`).
+        /// The first encode error short-circuits.
+        pub fn encode_children_expressions<'b, I>(
+            &self,
+            exprs: I,
+        ) -> Result<Vec<PhysicalExprNode>>
+        where
+            I: IntoIterator<Item = &'b Arc<dyn PhysicalExpr>>,
+        {
+            exprs
+                .into_iter()
+                .map(|expr| self.encode_child(expr))
+                .collect()
+        }
     }
 
     /// Internal dispatch trait. Implementors live in `datafusion-proto` and
@@ -636,6 +654,43 @@ pub mod proto_decode {
         pub fn decode(&self, node: &PhysicalExprNode) -> Result<Arc<dyn PhysicalExpr>> {
             self.decoder.decode(node, self.schema)
         }
+
+        /// Decode a required child node, erroring if it is absent.
+        ///
+        /// Proto child expressions are encoded as `Option<Box<PhysicalExprNode>>`;
+        /// pass the field directly (e.g. `node.expr.as_deref()`). `expr_name`
+        /// is the expression being decoded (e.g. `"InListExpr"`) and `field`
+        /// the proto field (e.g. `"expr"`); both are woven into the error so
+        /// it names *where* the missing field is, without each author
+        /// hand-rolling the string.
+        pub fn decode_required_expression(
+            &self,
+            node: Option<&PhysicalExprNode>,
+            expr_name: &str,
+            field: &str,
+        ) -> Result<Arc<dyn PhysicalExpr>> {
+            let node = node.ok_or_else(|| {
+                datafusion_common::internal_datafusion_err!(
+                    "{expr_name} is missing required field '{field}'"
+                )
+            })?;
+            self.decode(node)
+        }
+
+        /// Decode a sequence of child nodes, preserving order.
+        ///
+        /// Convenience wrapper over [`Self::decode`] for expressions holding a
+        /// `repeated` proto field (e.g. the `list` of an `InList`). The first
+        /// decode error short-circuits.
+        pub fn decode_children_expressions<'b, I>(
+            &self,
+            nodes: I,
+        ) -> Result<Vec<Arc<dyn PhysicalExpr>>>
+        where
+            I: IntoIterator<Item = &'b PhysicalExprNode>,
+        {
+            nodes.into_iter().map(|node| self.decode(node)).collect()
+        }
     }
 
     /// Internal dispatch trait. Implementors live in `datafusion-proto`.