Migrate BinaryExpr to PhysicalExpr::try_to_proto / try_from_proto

adriangb · claude · adriangb · commit 0ff65cc2bfcb · 2026-05-21T08:38:21.000-05:00
Second expression to use the new hooks, following the same pattern as
Column. The BinaryExpr downcast arm in
`serialize_physical_expr_with_converter` and the inline match arm in
`parse_physical_expr_with_converter` are removed; both directions now
live on `BinaryExpr` itself.

Linearization of nested same-op chains stays bit-for-bit identical with
the previous behavior, just expressed in terms of
`PhysicalExprEncodeCtx::encode_child` / `PhysicalExprDecodeCtx::decode`
instead of the proto converter.

`BinaryExpr::try_from_proto` takes the whole `PhysicalExprNode` and parses
the operator string itself. The proto-string-to-enum mapping moves from
`from_proto_binary_op` in `datafusion-proto` down to
`Operator::from_proto_name` in `datafusion-expr-common`, so the logical
plan path and the physical `BinaryExpr` decoder share one source of truth
without `physical-expr` depending on `datafusion-proto`; `from_proto_binary_op`
now delegates to it.

Co-Authored-By: Claude Opus 4.7 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/datafusion/expr-common/src/operator.rs b/datafusion/expr-common/src/operator.rs
@@ -390,6 +390,50 @@ impl Operator {
             | Operator::StringConcat => false,
         }
     }
+
+    /// Parse an `Operator` from the string name `datafusion-proto` uses on the
+    /// wire (the `Debug` name of the variant, e.g. `"Eq"`).
+    ///
+    /// Returns `None` for names with no binary-operator counterpart. This is
+    /// the canonical proto-string mapping, shared by `datafusion-proto`
+    /// (logical plans) and `PhysicalExpr` decoders such as `BinaryExpr`, so the
+    /// mapping is not duplicated across crates.
+    pub fn from_proto_name(name: &str) -> Option<Operator> {
+        Some(match name {
+            "And" => Operator::And,
+            "Or" => Operator::Or,
+            "Eq" => Operator::Eq,
+            "NotEq" => Operator::NotEq,
+            "LtEq" => Operator::LtEq,
+            "Lt" => Operator::Lt,
+            "Gt" => Operator::Gt,
+            "GtEq" => Operator::GtEq,
+            "Plus" => Operator::Plus,
+            "Minus" => Operator::Minus,
+            "Multiply" => Operator::Multiply,
+            "Divide" => Operator::Divide,
+            "Modulo" => Operator::Modulo,
+            "IsDistinctFrom" => Operator::IsDistinctFrom,
+            "IsNotDistinctFrom" => Operator::IsNotDistinctFrom,
+            "BitwiseAnd" => Operator::BitwiseAnd,
+            "BitwiseOr" => Operator::BitwiseOr,
+            "BitwiseXor" => Operator::BitwiseXor,
+            "BitwiseShiftLeft" => Operator::BitwiseShiftLeft,
+            "BitwiseShiftRight" => Operator::BitwiseShiftRight,
+            "RegexIMatch" => Operator::RegexIMatch,
+            "RegexMatch" => Operator::RegexMatch,
+            "RegexNotIMatch" => Operator::RegexNotIMatch,
+            "RegexNotMatch" => Operator::RegexNotMatch,
+            "LikeMatch" => Operator::LikeMatch,
+            "ILikeMatch" => Operator::ILikeMatch,
+            "NotLikeMatch" => Operator::NotLikeMatch,
+            "NotILikeMatch" => Operator::NotILikeMatch,
+            "StringConcat" => Operator::StringConcat,
+            "AtArrow" => Operator::AtArrow,
+            "ArrowAt" => Operator::ArrowAt,
+            _ => return None,
+        })
+    }
 }
 
 impl fmt::Display for Operator {
diff --git a/datafusion/physical-expr/src/expressions/binary.rs b/datafusion/physical-expr/src/expressions/binary.rs
@@ -610,6 +610,124 @@ impl PhysicalExpr for BinaryExpr {
         write!(f, " {} ", self.op)?;
         write_child(f, self.right.as_ref(), precedence)
     }
+
+    #[cfg(feature = "proto")]
+    fn try_to_proto(
+        &self,
+        ctx: &datafusion_physical_expr_common::physical_expr::proto_encode::PhysicalExprEncodeCtx<'_>,
+    ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalExprNode>> {
+        use datafusion_proto_models::protobuf;
+
+        // Linearize a nested binary expression tree of the same operator
+        // into a flat vector of operands to avoid deep recursion in proto.
+        let op = self.op;
+        let mut operand_refs: Vec<&Arc<dyn PhysicalExpr>> = vec![&self.right];
+        let mut current_expr: &BinaryExpr = self;
+        loop {
+            match current_expr.left.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| ctx.encode_child(e))
+            .collect::<Result<Vec<_>>>()?;
+
+        Ok(Some(protobuf::PhysicalExprNode {
+            expr_id: None,
+            expr_type: Some(protobuf::physical_expr_node::ExprType::BinaryExpr(
+                Box::new(protobuf::PhysicalBinaryExprNode {
+                    l: None,
+                    r: None,
+                    op: format!("{op:?}"),
+                    operands,
+                }),
+            )),
+        }))
+    }
+}
+
+#[cfg(feature = "proto")]
+impl BinaryExpr {
+    /// Reconstruct a [`BinaryExpr`] (or a left-deep tree of them when the proto
+    /// uses the linearized `operands` form) from its protobuf representation.
+    ///
+    /// Takes the whole [`PhysicalExprNode`] — the exact inverse of what
+    /// [`PhysicalExpr::try_to_proto`] produces — so every expression's
+    /// `try_from_proto` shares one signature. The operator string is parsed
+    /// via the canonical [`Operator::from_proto_name`] mapping, so no `op`
+    /// argument needs to be threaded in by the caller.
+    ///
+    /// [`PhysicalExprNode`]: datafusion_proto_models::protobuf::PhysicalExprNode
+    /// [`PhysicalExpr::try_to_proto`]: datafusion_physical_expr_common::physical_expr::PhysicalExpr::try_to_proto
+    /// [`PhysicalExprDecodeCtx::decode`]: datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx::decode
+    pub fn try_from_proto(
+        node: &datafusion_proto_models::protobuf::PhysicalExprNode,
+        ctx: &datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx<'_>,
+    ) -> Result<Arc<dyn PhysicalExpr>> {
+        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 op = Operator::from_proto_name(&node.op).ok_or_else(|| {
+            datafusion_common::DataFusionError::Internal(format!(
+                "Unsupported binary operator '{}'",
+                node.op
+            ))
+        })?;
+
+        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<_>>>()?;
+
+            if operands.len() < 2 {
+                return Err(datafusion_common::DataFusionError::Internal(
+                    "A binary expression must always have at least 2 operands"
+                        .to_string(),
+                ));
+            }
+
+            Ok(operands
+                .into_iter()
+                .reduce(|left, right| {
+                    Arc::new(BinaryExpr::new(left, op, right)) as Arc<dyn PhysicalExpr>
+                })
+                .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)?,
+            )))
+        }
+    }
 }
 
 /// Casts dictionary array to result type for binary numerical operators. Such operators
diff --git a/datafusion/proto/src/logical_plan/from_proto.rs b/datafusion/proto/src/logical_plan/from_proto.rs
@@ -753,42 +753,11 @@ fn parse_escape_char(s: &str) -> Result<Option<char>> {
 }
 
 pub fn from_proto_binary_op(op: &str) -> Result<Operator, Error> {
-    match op {
-        "And" => Ok(Operator::And),
-        "Or" => Ok(Operator::Or),
-        "Eq" => Ok(Operator::Eq),
-        "NotEq" => Ok(Operator::NotEq),
-        "LtEq" => Ok(Operator::LtEq),
-        "Lt" => Ok(Operator::Lt),
-        "Gt" => Ok(Operator::Gt),
-        "GtEq" => Ok(Operator::GtEq),
-        "Plus" => Ok(Operator::Plus),
-        "Minus" => Ok(Operator::Minus),
-        "Multiply" => Ok(Operator::Multiply),
-        "Divide" => Ok(Operator::Divide),
-        "Modulo" => Ok(Operator::Modulo),
-        "IsDistinctFrom" => Ok(Operator::IsDistinctFrom),
-        "IsNotDistinctFrom" => Ok(Operator::IsNotDistinctFrom),
-        "BitwiseAnd" => Ok(Operator::BitwiseAnd),
-        "BitwiseOr" => Ok(Operator::BitwiseOr),
-        "BitwiseXor" => Ok(Operator::BitwiseXor),
-        "BitwiseShiftLeft" => Ok(Operator::BitwiseShiftLeft),
-        "BitwiseShiftRight" => Ok(Operator::BitwiseShiftRight),
-        "RegexIMatch" => Ok(Operator::RegexIMatch),
-        "RegexMatch" => Ok(Operator::RegexMatch),
-        "RegexNotIMatch" => Ok(Operator::RegexNotIMatch),
-        "RegexNotMatch" => Ok(Operator::RegexNotMatch),
-        "LikeMatch" => Ok(Operator::LikeMatch),
-        "ILikeMatch" => Ok(Operator::ILikeMatch),
-        "NotLikeMatch" => Ok(Operator::NotLikeMatch),
-        "NotILikeMatch" => Ok(Operator::NotILikeMatch),
-        "StringConcat" => Ok(Operator::StringConcat),
-        "AtArrow" => Ok(Operator::AtArrow),
-        "ArrowAt" => Ok(Operator::ArrowAt),
-        other => Err(proto_error(format!(
-            "Unsupported binary operator '{other:?}'"
-        ))),
-    }
+    // The proto-string <-> `Operator` mapping is canonically owned by
+    // `datafusion-expr-common` so `datafusion-proto` (logical plans) and
+    // `PhysicalExpr` decoders (e.g. `BinaryExpr`) share one source of truth.
+    Operator::from_proto_name(op)
+        .ok_or_else(|| proto_error(format!("Unsupported binary operator '{op:?}'")))
 }
 
 fn parse_optional_expr(
diff --git a/datafusion/proto/src/physical_plan/from_proto.rs b/datafusion/proto/src/physical_plan/from_proto.rs
@@ -58,7 +58,6 @@ use super::{
     PhysicalProtoConverterExtension,
 };
 use crate::convert::TryFromProto;
-use crate::logical_plan::{self};
 use crate::protobuf::physical_expr_node::ExprType;
 use crate::{convert_required, convert_required_proto, protobuf};
 use datafusion_physical_expr::expressions::{
@@ -263,8 +262,7 @@ pub fn parse_physical_expr_with_converter(
 
     // Decoder context handed to per-expression `try_from_proto` constructors.
     // This is the new shape the codebase is migrating toward (see #21835);
-    // for now only `Column` is migrated and the rest of the variants are still
-    // matched inline.
+    // the remaining `ExprType` variants stay matched inline until they migrate.
     let decoder = ConverterDecoder {
         ctx,
         proto_converter,
@@ -277,54 +275,12 @@ pub fn parse_physical_expr_with_converter(
 
     let pexpr: Arc<dyn PhysicalExpr> = match expr_type {
         // Migrated expressions take the whole `PhysicalExprNode` and unwrap
-        // their own `ExprType` variant (see #21835); this match only routes.
+        // their own `ExprType` variant — see #21835. This match only routes
+        // to the right constructor.
         ExprType::Column(_) => Column::try_from_proto(proto, &decode_ctx)?,
         ExprType::UnknownColumn(c) => Arc::new(UnKnownColumn::new(&c.name)),
         ExprType::Literal(scalar) => Arc::new(Literal::new(scalar.try_into()?)),
-        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, input_schema, ctx))
-                    .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,
-                        proto_converter,
-                    )?,
-                    op,
-                    parse_required_physical_expr(
-                        binary_expr.r.as_deref(),
-                        ctx,
-                        "right",
-                        input_schema,
-                        proto_converter,
-                    )?,
-                ))
-            }
-        }
+        ExprType::BinaryExpr(_) => BinaryExpr::try_from_proto(proto, &decode_ctx)?,
         ExprType::AggregateExpr(_) => {
             return not_impl_err!(
                 "Cannot convert aggregate expr node to physical expression"
diff --git a/datafusion/proto/src/physical_plan/to_proto.rs b/datafusion/proto/src/physical_plan/to_proto.rs
@@ -36,8 +36,8 @@ use datafusion_physical_expr::scalar_subquery::ScalarSubqueryExpr;
 use datafusion_physical_expr::window::{SlidingAggregateWindowExpr, StandardWindowExpr};
 use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr;
 use datafusion_physical_plan::expressions::{
-    BinaryExpr, CaseExpr, CastExpr, DynamicFilterPhysicalExpr, InListExpr, IsNotNullExpr,
-    IsNullExpr, LikeExpr, Literal, NegativeExpr, NotExpr, TryCastExpr, UnKnownColumn,
+    CaseExpr, CastExpr, DynamicFilterPhysicalExpr, InListExpr, IsNotNullExpr, IsNullExpr,
+    LikeExpr, Literal, NegativeExpr, NotExpr, TryCastExpr, UnKnownColumn,
 };
 use datafusion_physical_plan::joins::{HashExpr, HashTableLookupExpr};
 use datafusion_physical_plan::udaf::AggregateFunctionExpr;
@@ -252,7 +252,7 @@ pub fn serialize_physical_expr(
 }
 
 /// Concrete [`PhysicalExprEncode`] driver used to back
-/// [`PhysicalExprEncodeCtx`] when expressions invoke `PhysicalExpr::try_to_proto`.
+/// [`PhysicalExprEncodeCtx`] when expressions invoke `PhysicalExpr::to_proto`.
 ///
 /// Wraps the existing extension codec + converter pair so individual
 /// expressions can recurse into children without depending on
@@ -336,46 +336,6 @@ 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().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: None,
-            r: None,
-            op: format!("{:?}", op),
-            operands,
-        });
-
-        Ok(protobuf::PhysicalExprNode {
-            expr_id,
-            expr_type: Some(protobuf::physical_expr_node::ExprType::BinaryExpr(
-                binary_expr,
-            )),
-        })
     } else if let Some(expr) = expr.downcast_ref::<CaseExpr>() {
         Ok(protobuf::PhysicalExprNode {
             expr_id,
