fix: linearized operands in physical binaryexpr protobuf to avoid recursion limit (#21031)

## Which issue does this PR close?

- part of #18602.

## Rationale for this change

When a SQL query contains many filter conditions (e.g., 40+ `AND`/`OR`
clauses in a `WHERE`), serializing the physical plan to protobuf and
deserializing it fails with `DecodeError: recursion limit reached`.
[This is because prost has a default recursion limit of
100](https://docs.rs/prost/latest/src/prost/lib.rs.html#30), and each
`BinaryExpr` nesting consumes ~2 levels of protobuf recursion depth, so
a chain of ~50 AND conditions exceeds the limit.

## What changes are included in this PR?

Applied the same **linearization** approach that [logical expressions
already
use](https://github.com/apache/datafusion/blob/b6b542e87b84f4744096106bea0de755b2e70cc5/datafusion/proto/src/logical_plan/to_proto.rs#L228-L256)
that convert a left-deep tree to linearization list. Instead of encoding
a chain of same-operator binary expressions as a deeply nested tree, we
flatten it into a flat `operands` list:

**Before (nested, O(n) recursion depth):**
```
BinaryExpr(AND) { l: BinaryExpr(AND) { l: BinaryExpr(AND) { l: a, r: b }, r: c }, r: d }
```

**After (flat, O(1) recursion depth for the chain):**
```
BinaryExpr(AND) { operands: [a, b, c, d] }
```
## Are these changes tested?

yes, add some test case

## Are there any user-facing changes?

Author: haohuaijin

datafusion/proto/proto/datafusion.proto

@@ -981,6 +981,9 @@ message PhysicalBinaryExprNode {
   PhysicalExprNode l = 1;
   PhysicalExprNode r = 2;
   string op = 3;
+  // Linearized operands for chains of the same operator (e.g. a AND b AND c).
+  // When present, `l` and `r` are ignored and `operands` holds the flattened list.
+  repeated PhysicalExprNode operands = 4;
 }
 
 message PhysicalDateTimeIntervalExprNode {

datafusion/proto/src/generated/pbjson.rs

@@ -273,25 +273,59 @@ pub fn parse_physical_expr_with_converter(
         }
         ExprType::UnknownColumn(c) => Arc::new(UnKnownColumn::new(&c.name)),
         ExprType::Literal(scalar) => Arc::new(Literal::new(scalar.try_into()?)),
-        ExprType::BinaryExpr(binary_expr) => Arc::new(BinaryExpr::new(
-            parse_required_physical_expr(
-                binary_expr.l.as_deref(),
-                ctx,
-                "left",
-                input_schema,
-                codec,
-                proto_converter,
-            )?,
-            logical_plan::from_proto::from_proto_binary_op(&binary_expr.op)?,
-            parse_required_physical_expr(
-                binary_expr.r.as_deref(),
-                ctx,
-                "right",
-                input_schema,
-                codec,
-                proto_converter,
-            )?,
-        )),
+        ExprType::BinaryExpr(binary_expr) => {
+            let op = logical_plan::from_proto::from_proto_binary_op(&binary_expr.op)?;
+            if !binary_expr.operands.is_empty() {
+                // New linearized format: reduce the flat operands list back into
+                // a nested binary expression tree.
+                let operands: Vec<Arc<dyn PhysicalExpr>> = binary_expr
+                    .operands
+                    .iter()
+                    .map(|e| {
+                        proto_converter.proto_to_physical_expr(
+                            e,
+                            ctx,
+                            input_schema,
+                            codec,
+                        )
+                    })
+                    .collect::<Result<Vec<_>>>()?;
+
+                if operands.len() < 2 {
+                    return Err(proto_error(
+                        "A binary expression must always have at least 2 operands",
+                    ));
+                }
+
+                operands
+                    .into_iter()
+                    .reduce(|left, right| Arc::new(BinaryExpr::new(left, op, right)))
+                    .expect(
+                        "Binary expression could not be reduced to a single expression.",
+                    )
+            } else {
+                // Legacy format with l/r fields
+                Arc::new(BinaryExpr::new(
+                    parse_required_physical_expr(
+                        binary_expr.l.as_deref(),
+                        ctx,
+                        "left",
+                        input_schema,
+                        codec,
+                        proto_converter,
+                    )?,
+                    op,
+                    parse_required_physical_expr(
+                        binary_expr.r.as_deref(),
+                        ctx,
+                        "right",
+                        input_schema,
+                        codec,
+                        proto_converter,
+                    )?,
+                ))
+            }
+        }
         ExprType::AggregateExpr(_) => {
             return not_impl_err!(
                 "Cannot convert aggregate expr node to physical expression"

datafusion/proto/src/generated/prost.rs

@@ -306,14 +306,37 @@ pub fn serialize_physical_expr_with_converter(
             )),
         })
     } else if let Some(expr) = expr.downcast_ref::<BinaryExpr>() {
+        // Linearize a nested binary expression tree of the same operator
+        // into a flat vector of operands to avoid deep recursion in proto.
+        let op = expr.op();
+        let mut operand_refs: Vec<&Arc<dyn PhysicalExpr>> = vec![expr.right()];
+        let mut current_expr: &BinaryExpr = expr;
+        loop {
+            match current_expr.left().as_any().downcast_ref::<BinaryExpr>() {
+                Some(bin) if bin.op() == op => {
+                    operand_refs.push(bin.right());
+                    current_expr = bin;
+                }
+                _ => {
+                    operand_refs.push(current_expr.left());
+                    break;
+                }
+            }
+        }
+
+        // Reverse so operands are ordered from left innermost to right outermost
+        operand_refs.reverse();
+
+        let operands = operand_refs
+            .iter()
+            .map(|e| proto_converter.physical_expr_to_proto(e, codec))
+            .collect::<Result<Vec<_>>>()?;
+
         let binary_expr = Box::new(protobuf::PhysicalBinaryExprNode {
-            l: Some(Box::new(
-                proto_converter.physical_expr_to_proto(expr.left(), codec)?,
-            )),
-            r: Some(Box::new(
-                proto_converter.physical_expr_to_proto(expr.right(), codec)?,
-            )),
-            op: format!("{:?}", expr.op()),
+            l: None,
+            r: None,
+            op: format!("{:?}", op),
+            operands,
         });
 
         Ok(protobuf::PhysicalExprNode {