refactor(physical-expr): adopt new proto helpers in already-migrated expressions

Ports the existing `try_to_proto` / `try_from_proto` implementations
onto the helpers introduced in the previous commit
(`expect_expr_variant!`) and the wider use of
`decode_required_expression` / `decode_children_expressions` /
`encode_children_expressions` from #22513.

Covers every expression already migrated under #22418:

- `Column`, `BinaryExpr` (originally #21929)
- `LikeExpr` (#22471)
- `InListExpr` (#22503)
- `NegativeExpr` (#22483)

`BinaryExpr` additionally switches its `l`/`r` legacy-decode arms to
`decode_required_expression`, removing two more hand-rolled
"missing required field" strings, and runs its `operands` encode/
decode loops through `encode_children_expressions` /
`decode_children_expressions`.

One existing test changes assertion text — `InListExpr`'s rejected-
variant message was the only one using the article "an" instead of
the macro's article-free "a"; updated to match. No wire-format
change; `cargo test -p datafusion-proto --test proto_integration` is
green (173 / 173).

Author: adriangb

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

@@ -638,10 +638,7 @@ impl PhysicalExpr for BinaryExpr {
         // Reverse so operands are ordered from left innermost to right outermost.
         operand_refs.reverse();
 
-        let operands = operand_refs
-            .iter()
-            .map(|e| ctx.encode_child(e))
-            .collect::<Result<Vec<_>>>()?;
+        let operands = ctx.encode_children_expressions(operand_refs)?;
 
         Ok(Some(protobuf::PhysicalExprNode {
             expr_id: None,
@@ -675,11 +672,13 @@ impl BinaryExpr {
         node: &datafusion_proto_models::protobuf::PhysicalExprNode,
         ctx: &datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx<'_>,
     ) -> Result<Arc<dyn PhysicalExpr>> {
+        use datafusion_physical_expr_common::expect_expr_variant;
         use datafusion_proto_models::protobuf;
-        let node = match &node.expr_type {
-            Some(protobuf::physical_expr_node::ExprType::BinaryExpr(b)) => b.as_ref(),
-            _ => return internal_err!("PhysicalExprNode is not a BinaryExpr"),
-        };
+        let node = expect_expr_variant!(
+            node,
+            protobuf::physical_expr_node::ExprType::BinaryExpr,
+            "BinaryExpr",
+        );
         let op = Operator::from_proto_name(&node.op).ok_or_else(|| {
             datafusion_common::DataFusionError::Internal(format!(
                 "Unsupported binary operator '{}'",
@@ -690,17 +689,12 @@ impl BinaryExpr {
         if !node.operands.is_empty() {
             // New linearized format: reduce the flat operands list back into
             // a nested binary expression tree.
-            let operands = node
-                .operands
-                .iter()
-                .map(|e| ctx.decode(e))
-                .collect::<Result<Vec<_>>>()?;
+            let operands = ctx.decode_children_expressions(&node.operands)?;
 
             if operands.len() < 2 {
-                return Err(datafusion_common::DataFusionError::Internal(
+                return internal_err!(
                     "A binary expression must always have at least 2 operands"
-                        .to_string(),
-                ));
+                );
             }
 
             Ok(operands
@@ -711,21 +705,11 @@ impl BinaryExpr {
                 .expect("Binary expression could not be reduced to a single expression."))
         } else {
             // Legacy format with l/r fields.
-            let left = node.l.as_deref().ok_or_else(|| {
-                datafusion_common::DataFusionError::Internal(
-                    "BinaryExpr is missing required field 'left'".to_string(),
-                )
-            })?;
-            let right = node.r.as_deref().ok_or_else(|| {
-                datafusion_common::DataFusionError::Internal(
-                    "BinaryExpr is missing required field 'right'".to_string(),
-                )
-            })?;
-            Ok(Arc::new(BinaryExpr::new(
-                ctx.decode(left)?,
-                op,
-                ctx.decode(right)?,
-            )))
+            let left =
+                ctx.decode_required_expression(node.l.as_deref(), "BinaryExpr", "left")?;
+            let right =
+                ctx.decode_required_expression(node.r.as_deref(), "BinaryExpr", "right")?;
+            Ok(Arc::new(BinaryExpr::new(left, op, right)))
         }
     }
 }

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

@@ -182,11 +182,13 @@ impl Column {
         node: &datafusion_proto_models::protobuf::PhysicalExprNode,
         _ctx: &datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx<'_>,
     ) -> Result<Arc<dyn PhysicalExpr>> {
+        use datafusion_physical_expr_common::expect_expr_variant;
         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"),
-        };
+        let protobuf::PhysicalColumn { name, index } = expect_expr_variant!(
+            node,
+            protobuf::physical_expr_node::ExprType::Column,
+            "Column",
+        );
         Ok(Arc::new(Column::new(name, *index as usize)))
     }
 }

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

@@ -252,16 +252,14 @@ impl InListExpr {
         node: &datafusion_proto_models::protobuf::PhysicalExprNode,
         ctx: &datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx<'_>,
     ) -> Result<Arc<dyn PhysicalExpr>> {
+        use datafusion_physical_expr_common::expect_expr_variant;
         use datafusion_proto_models::protobuf;
 
-        let node = match &node.expr_type {
-            Some(protobuf::physical_expr_node::ExprType::InList(n)) => n,
-            _ => {
-                return datafusion_common::internal_err!(
-                    "PhysicalExprNode is not an InList"
-                );
-            }
-        };
+        let node = expect_expr_variant!(
+            node,
+            protobuf::physical_expr_node::ExprType::InList,
+            "InList",
+        );
 
         let expr =
             ctx.decode_required_expression(node.expr.as_deref(), "InListExpr", "expr")?;
@@ -3981,7 +3979,7 @@ mod proto_tests {
         let err = InListExpr::try_from_proto(&node, &ctx).unwrap_err();
         assert!(matches!(
             err,
-            DataFusionError::Internal(msg) if msg.contains("PhysicalExprNode is not an InList")
+            DataFusionError::Internal(msg) if msg.contains("PhysicalExprNode is not a InList")
         ));
     }
 

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

@@ -180,15 +180,14 @@ impl LikeExpr {
         node: &datafusion_proto_models::protobuf::PhysicalExprNode,
         ctx: &datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx<'_>,
     ) -> Result<Arc<dyn PhysicalExpr>> {
-        use datafusion_common::internal_err;
+        use datafusion_physical_expr_common::expect_expr_variant;
         use datafusion_proto_models::protobuf;
 
-        let like_expr = match &node.expr_type {
-            Some(protobuf::physical_expr_node::ExprType::LikeExpr(like_expr)) => {
-                like_expr.as_ref()
-            }
-            _ => return internal_err!("PhysicalExprNode is not a LikeExpr"),
-        };
+        let like_expr = expect_expr_variant!(
+            node,
+            protobuf::physical_expr_node::ExprType::LikeExpr,
+            "LikeExpr",
+        );
 
         Ok(Arc::new(LikeExpr::new(
             like_expr.negated,

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

@@ -200,14 +200,16 @@ impl NegativeExpr {
         node: &datafusion_proto_models::protobuf::PhysicalExprNode,
         ctx: &datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx<'_>,
     ) -> Result<Arc<dyn PhysicalExpr>> {
+        use datafusion_physical_expr_common::expect_expr_variant;
         use datafusion_proto_models::protobuf;
 
-        let expr = match &node.expr_type {
-            Some(protobuf::physical_expr_node::ExprType::Negative(n)) => {
-                ctx.decode_required_expression(n.expr.as_deref(), "NegativeExpr", "expr")?
-            }
-            _ => return internal_err!("PhysicalExprNode is not a Negative"),
-        };
+        let n = expect_expr_variant!(
+            node,
+            protobuf::physical_expr_node::ExprType::Negative,
+            "Negative",
+        );
+        let expr =
+            ctx.decode_required_expression(n.expr.as_deref(), "NegativeExpr", "expr")?;
 
         Ok(Arc::new(NegativeExpr::new(expr)))
     }