refactor(physical-expr): add proto ctx expr helpers and adopt in InList/Like (apache#22513)

## Which issue does this PR close?

- Part of apache#22418 (decentralizing `datafusion-proto` serialization onto
the expressions themselves; follows apache#21835, apache#22471, apache#22503).

## Rationale for this change

Expressions migrating to the `try_to_proto` / `try_from_proto` hooks
keep hand-rolling the same boilerplate that 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 `PhysicalExprEncodeCtx` / `PhysicalExprDecodeCtx` surfaces
deliberately hide.

This was raised in review on apache#22503 — rather than re-implement the list
maps and "missing required field" checks in every migrated expression,
expose the same shapes on the ctx structs.

## What changes are included in this PR?

- **`datafusion-physical-expr-common`**: three thin convenience methods,
built on the existing `encode_child` / `decode` primitives:
  - `PhysicalExprEncodeCtx::encode_children_expressions`
- `PhysicalExprDecodeCtx::decode_required_expression` (also standardizes
the `Missing required field "<name>"` error so each expression no longer
spells its own)
  - `PhysicalExprDecodeCtx::decode_children_expressions`
- **`datafusion-physical-expr`**: adopt them in `InListExpr` and
`LikeExpr`, removing the hand-rolled list maps and per-field
`ok_or_else` checks.

### Behavior note

`decode_required_expression` couples the presence check with the decode,
so `LikeExpr` now decodes children left-to-right rather than validating
both required fields up front. The end result is unchanged (a missing
required field still errors), but a present sibling is decoded before a
later missing field is reported. The
`try_from_proto_rejects_missing_pattern` unit test is updated to reflect
this.

## Are these changes tested?

Yes — covered by existing tests, no new ones needed:

- The isolated `proto_tests` modules in `in_list.rs` and `like.rs`
already exercise all three helpers (list encode/decode, required decode,
and the missing-field + child-error paths) through the migrated
`try_to_proto` / `try_from_proto`.
- The `datafusion-proto` round-trip integration tests
(`roundtrip_inlist`, `roundtrip_like`,
`roundtrip_filter_with_not_and_in_list`,
`test_tpch_part_in_list_query_with_real_parquet_data`, etc.) continue to
pass.
- `cargo clippy --all-targets --features proto -D warnings` is clean on
the touched crates; `cargo fmt --all` applied.

## Are there any user-facing changes?

No. The per-expression `missing required field` error wording is
preserved: `LikeExpr` is byte-for-byte identical, and `InList` only
changes its label from `InList` to `InListExpr` (the actual type name).
The format now lives in one place (`decode_required_expression`) instead
of being hand-spelled per expression.

---------

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`.

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

@@ -263,17 +263,9 @@ impl InListExpr {
             }
         };
 
-        let expr = ctx.decode(node.expr.as_deref().ok_or_else(|| {
-            datafusion_common::DataFusionError::Internal(
-                "InList is missing required field 'expr'".to_string(),
-            )
-        })?)?;
-
-        let list = node
-            .list
-            .iter()
-            .map(|e| ctx.decode(e))
-            .collect::<Result<Vec<_>>>()?;
+        let expr =
+            ctx.decode_required_expression(node.expr.as_deref(), "InListExpr", "expr")?;
+        let list = ctx.decode_children_expressions(&node.list)?;
 
         Ok(Arc::new(InListExpr::try_new(
             expr,
@@ -491,11 +483,7 @@ impl PhysicalExpr for InListExpr {
             expr_type: Some(protobuf::physical_expr_node::ExprType::InList(Box::new(
                 protobuf::PhysicalInListNode {
                     expr: Some(Box::new(ctx.encode_child(&self.expr)?)),
-                    list: self
-                        .list
-                        .iter()
-                        .map(|e| ctx.encode_child(e))
-                        .collect::<Result<Vec<_>>>()?,
+                    list: ctx.encode_children_expressions(&self.list)?,
                     negated: self.negated,
                 },
             ))),
@@ -4007,7 +3995,7 @@ mod proto_tests {
         let err = InListExpr::try_from_proto(&node, &ctx).unwrap_err();
         assert!(matches!(
             err,
-            DataFusionError::Internal(msg) if msg.contains("InList is missing required field 'expr'")
+            DataFusionError::Internal(msg) if msg.contains("InListExpr is missing required field 'expr'")
         ));
     }
 

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

@@ -190,22 +190,19 @@ impl LikeExpr {
             _ => return internal_err!("PhysicalExprNode is not a LikeExpr"),
         };
 
-        let expr = like_expr.expr.as_deref().ok_or_else(|| {
-            datafusion_common::DataFusionError::Internal(
-                "LikeExpr is missing required field 'expr'".to_string(),
-            )
-        })?;
-        let pattern = like_expr.pattern.as_deref().ok_or_else(|| {
-            datafusion_common::DataFusionError::Internal(
-                "LikeExpr is missing required field 'pattern'".to_string(),
-            )
-        })?;
-
         Ok(Arc::new(LikeExpr::new(
             like_expr.negated,
             like_expr.case_insensitive,
-            ctx.decode(expr)?,
-            ctx.decode(pattern)?,
+            ctx.decode_required_expression(
+                like_expr.expr.as_deref(),
+                "LikeExpr",
+                "expr",
+            )?,
+            ctx.decode_required_expression(
+                like_expr.pattern.as_deref(),
+                "LikeExpr",
+                "pattern",
+            )?,
         )))
     }
 }
@@ -491,7 +488,9 @@ mod proto_tests {
     fn try_from_proto_rejects_missing_pattern() {
         let node = like_node(false, false, Some(Box::new(column_node("a"))), None);
         let schema = Schema::empty();
-        let decoder = UnreachableDecoder;
+        // `expr` is present, so it is decoded before the missing-`pattern`
+        // check fires; use a decoder that succeeds for that first child.
+        let decoder = StubDecoder::ok();
         let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
         let err = LikeExpr::try_from_proto(&node, &ctx).unwrap_err();
         assert!(matches!(