This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-25034-b1467ea5c2b16cd68d78d2cf4c994e9cc55dd3b5 in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit f0eaf4b8f188884b45860c1849e7ff258e6255ac Author: Peter Lee <[email protected]> AuthorDate: Mon Sep 21 12:36:14 2026 +0000 fix: preserve overflow behavior and exhaustively destructure expression proto hooks (#25034) ## Which issue does this PR close? Closes #24614. ## Rationale for this change This PR fixes serialization losing the setting that makes arithmetic fail on overflow. For example, checked `Int32::MAX + 1` raises an error before serialization but returns `Int32::MIN` after decoding. Preserving this setting ensures that sending an expression through protobuf preserves its arithmetic behavior. ## What changes are included in this PR? The protobuf message now stores `BinaryExpr::fail_on_overflow`, and both supported decoding formats restore it. When the encoder combines nested expressions into a flat list of operands, it requires their operators and overflow settings to match. This preserves the behavior of expressions that mix checked and wrapping arithmetic. The generated Rust and JSON bindings include the new field. All six encoding and decoding hooks for `BinaryExpr`, `LikeExpr`, and `SqlSimilarToPattern` explicitly list every field without a rest pattern. Adding a field to an expression or its protobuf payload will cause a compile error until the corresponding hook handles it. The PR also changes the PostgreSQL SQLLogicTest decimal formatter to borrow its argument, resolving an existing Clippy error that blocked the required checks before committing. ## What is the testing strategy for this PR? The new tests serialize and decode nested additions, then check their evaluated results for all four combinations of checked and wrapping arithmetic. The regression test failed before the fix because an expression that should raise an overflow error returned `Int32(-2147483648)`. Additional tests cover older messages that omit the new field and verify that JSON preserves the overflow setting. All 17 focused expression tests passed. The extended workspace run passed 11,260 Rust tests, with 8 ignored, and completed all 511 SQLLogicTest files. Both conversion tests passed with the PostgreSQL feature enabled. Formatting, Clippy with all targets and features, and the complete `./dev/rust_lint.sh` suite also passed. ## Are there any user-facing changes? Expressions configured to fail on arithmetic overflow now raise the expected error after serialization and decoding, including nested expressions with different overflow settings. --- datafusion/physical-expr/src/expressions/binary.rs | 56 +++++++--- datafusion/physical-expr/src/expressions/like.rs | 38 ++++--- .../src/expressions/similar_to_pattern.rs | 8 +- datafusion/proto-models/proto/datafusion.proto | 3 +- datafusion/proto-models/src/generated/pbjson.rs | 18 +++ datafusion/proto-models/src/generated/prost.rs | 4 +- datafusion/proto/tests/cases/plans/exprs.rs | 124 +++++++++++++++++++++ 7 files changed, 213 insertions(+), 38 deletions(-) diff --git a/datafusion/physical-expr/src/expressions/binary.rs b/datafusion/physical-expr/src/expressions/binary.rs index 1117d3fdd5..a0a6548518 100644 --- a/datafusion/physical-expr/src/expressions/binary.rs +++ b/datafusion/physical-expr/src/expressions/binary.rs @@ -984,19 +984,27 @@ impl PhysicalExpr for BinaryExpr { ) -> 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; + let Self { + left, + op, + right, + fail_on_overflow, + } = self; + + // Linearize a nested binary expression tree with the same operator and + // overflow policy into flat operands to avoid deep recursion in proto. + let mut operand_refs: Vec<&Arc<dyn PhysicalExpr>> = vec![right]; + let mut current_left = left; loop { - match current_expr.left.downcast_ref::<BinaryExpr>() { - Some(bin) if bin.op == op => { + match current_left.downcast_ref::<BinaryExpr>() { + Some(bin) + if bin.op == *op && bin.fail_on_overflow == *fail_on_overflow => + { operand_refs.push(&bin.right); - current_expr = bin; + current_left = &bin.left; } _ => { - operand_refs.push(¤t_expr.left); + operand_refs.push(current_left); break; } } @@ -1014,6 +1022,7 @@ impl PhysicalExpr for BinaryExpr { r: None, op: format!("{op:?}"), operands, + fail_on_overflow: *fail_on_overflow, }), )), })) @@ -1045,17 +1054,23 @@ impl BinaryExpr { protobuf::physical_expr_node::ExprType::BinaryExpr, "BinaryExpr", ); - let op = Operator::from_proto_name(&node.op).ok_or_else(|| { + let protobuf::PhysicalBinaryExprNode { + l, + r, + op, + operands, + fail_on_overflow, + } = node.as_ref(); + let op = Operator::from_proto_name(op).ok_or_else(|| { datafusion_common::DataFusionError::Internal(format!( - "Unsupported binary operator '{}'", - node.op + "Unsupported binary operator '{op}'" )) })?; - if !node.operands.is_empty() { + if !operands.is_empty() { // New linearized format: reduce the flat operands list back into // a nested binary expression tree. - let operands = ctx.decode_children_expressions(&node.operands)?; + let operands = ctx.decode_children_expressions(operands)?; if operands.len() < 2 { return internal_err!( @@ -1066,16 +1081,21 @@ impl BinaryExpr { Ok(operands .into_iter() .reduce(|left, right| { - Arc::new(BinaryExpr::new(left, op, right)) as Arc<dyn PhysicalExpr> + Arc::new( + BinaryExpr::new(left, op, right) + .with_fail_on_overflow(*fail_on_overflow), + ) as Arc<dyn PhysicalExpr> }) .expect("Binary expression could not be reduced to a single expression.")) } else { // Legacy format with l/r fields. let left = - ctx.decode_required_expression(node.l.as_deref(), "BinaryExpr", "left")?; + ctx.decode_required_expression(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))) + ctx.decode_required_expression(r.as_deref(), "BinaryExpr", "right")?; + Ok(Arc::new( + BinaryExpr::new(left, op, right).with_fail_on_overflow(*fail_on_overflow), + )) } } } diff --git a/datafusion/physical-expr/src/expressions/like.rs b/datafusion/physical-expr/src/expressions/like.rs index 7535f109a0..8042eac412 100644 --- a/datafusion/physical-expr/src/expressions/like.rs +++ b/datafusion/physical-expr/src/expressions/like.rs @@ -153,14 +153,21 @@ impl PhysicalExpr for LikeExpr { ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalExprNode>> { use datafusion_proto_models::protobuf; + let Self { + negated, + case_insensitive, + expr, + pattern, + } = self; + Ok(Some(protobuf::PhysicalExprNode { expr_id: None, expr_type: Some(protobuf::physical_expr_node::ExprType::LikeExpr(Box::new( protobuf::PhysicalLikeExprNode { - negated: self.negated, - case_insensitive: self.case_insensitive, - expr: Some(Box::new(ctx.encode_child(&self.expr)?)), - pattern: Some(Box::new(ctx.encode_child(&self.pattern)?)), + negated: *negated, + case_insensitive: *case_insensitive, + expr: Some(Box::new(ctx.encode_child(expr)?)), + pattern: Some(Box::new(ctx.encode_child(pattern)?)), }, ))), })) @@ -189,19 +196,18 @@ impl LikeExpr { "LikeExpr", ); + let protobuf::PhysicalLikeExprNode { + negated, + case_insensitive, + expr, + pattern, + } = like_expr.as_ref(); + Ok(Arc::new(LikeExpr::new( - like_expr.negated, - like_expr.case_insensitive, - ctx.decode_required_expression( - like_expr.expr.as_deref(), - "LikeExpr", - "expr", - )?, - ctx.decode_required_expression( - like_expr.pattern.as_deref(), - "LikeExpr", - "pattern", - )?, + *negated, + *case_insensitive, + ctx.decode_required_expression(expr.as_deref(), "LikeExpr", "expr")?, + ctx.decode_required_expression(pattern.as_deref(), "LikeExpr", "pattern")?, ))) } } diff --git a/datafusion/physical-expr/src/expressions/similar_to_pattern.rs b/datafusion/physical-expr/src/expressions/similar_to_pattern.rs index 4078cb5256..0546158e35 100644 --- a/datafusion/physical-expr/src/expressions/similar_to_pattern.rs +++ b/datafusion/physical-expr/src/expressions/similar_to_pattern.rs @@ -136,11 +136,13 @@ impl PhysicalExpr for SqlSimilarToPattern { ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalExprNode>> { use datafusion_proto_models::protobuf; + let Self { expr } = self; + Ok(Some(protobuf::PhysicalExprNode { expr_id: None, expr_type: Some(protobuf::physical_expr_node::ExprType::SqlSimilarToPattern( Box::new(protobuf::PhysicalSqlSimilarToPatternNode { - expr: Some(Box::new(ctx.encode_child(&self.expr)?)), + expr: Some(Box::new(ctx.encode_child(expr)?)), }), )), })) @@ -169,9 +171,11 @@ impl SqlSimilarToPattern { "SqlSimilarToPattern", ); + let protobuf::PhysicalSqlSimilarToPatternNode { expr } = pattern.as_ref(); + Ok(Arc::new(SqlSimilarToPattern::new( ctx.decode_required_expression( - pattern.expr.as_deref(), + expr.as_deref(), "SqlSimilarToPattern", "expr", )?, diff --git a/datafusion/proto-models/proto/datafusion.proto b/datafusion/proto-models/proto/datafusion.proto index 15dc272fab..10acc53ccb 100644 --- a/datafusion/proto-models/proto/datafusion.proto +++ b/datafusion/proto-models/proto/datafusion.proto @@ -1183,9 +1183,10 @@ 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). + // Linearized operands for chains of the same operator and overflow policy. // When present, `l` and `r` are ignored and `operands` holds the flattened list. repeated PhysicalExprNode operands = 4; + bool fail_on_overflow = 5; } message PhysicalDateTimeIntervalExprNode { diff --git a/datafusion/proto-models/src/generated/pbjson.rs b/datafusion/proto-models/src/generated/pbjson.rs index cada03c9a6..e6e465cb0a 100644 --- a/datafusion/proto-models/src/generated/pbjson.rs +++ b/datafusion/proto-models/src/generated/pbjson.rs @@ -18672,6 +18672,9 @@ impl serde::Serialize for PhysicalBinaryExprNode { if !self.operands.is_empty() { len += 1; } + if self.fail_on_overflow { + len += 1; + } let mut struct_ser = serializer.serialize_struct("datafusion.PhysicalBinaryExprNode", len)?; if let Some(v) = self.l.as_ref() { struct_ser.serialize_field("l", v)?; @@ -18685,6 +18688,9 @@ impl serde::Serialize for PhysicalBinaryExprNode { if !self.operands.is_empty() { struct_ser.serialize_field("operands", &self.operands)?; } + if self.fail_on_overflow { + struct_ser.serialize_field("failOnOverflow", &self.fail_on_overflow)?; + } struct_ser.end() } } @@ -18699,6 +18705,8 @@ impl<'de> serde::Deserialize<'de> for PhysicalBinaryExprNode { "r", "op", "operands", + "fail_on_overflow", + "failOnOverflow", ]; #[allow(clippy::enum_variant_names)] @@ -18707,6 +18715,7 @@ impl<'de> serde::Deserialize<'de> for PhysicalBinaryExprNode { R, Op, Operands, + FailOnOverflow, } impl<'de> serde::Deserialize<'de> for GeneratedField { fn deserialize<D>(deserializer: D) -> std::result::Result<GeneratedField, D::Error> @@ -18732,6 +18741,7 @@ impl<'de> serde::Deserialize<'de> for PhysicalBinaryExprNode { "r" => Ok(GeneratedField::R), "op" => Ok(GeneratedField::Op), "operands" => Ok(GeneratedField::Operands), + "failOnOverflow" | "fail_on_overflow" => Ok(GeneratedField::FailOnOverflow), _ => Err(serde::de::Error::unknown_field(value, FIELDS)), } } @@ -18755,6 +18765,7 @@ impl<'de> serde::Deserialize<'de> for PhysicalBinaryExprNode { let mut r__ = None; let mut op__ = None; let mut operands__ = None; + let mut fail_on_overflow__ = None; while let Some(k) = map_.next_key()? { match k { GeneratedField::L => { @@ -18781,6 +18792,12 @@ impl<'de> serde::Deserialize<'de> for PhysicalBinaryExprNode { } operands__ = Some(map_.next_value()?); } + GeneratedField::FailOnOverflow => { + if fail_on_overflow__.is_some() { + return Err(serde::de::Error::duplicate_field("failOnOverflow")); + } + fail_on_overflow__ = Some(map_.next_value()?); + } } } Ok(PhysicalBinaryExprNode { @@ -18788,6 +18805,7 @@ impl<'de> serde::Deserialize<'de> for PhysicalBinaryExprNode { r: r__, op: op__.unwrap_or_default(), operands: operands__.unwrap_or_default(), + fail_on_overflow: fail_on_overflow__.unwrap_or_default(), }) } } diff --git a/datafusion/proto-models/src/generated/prost.rs b/datafusion/proto-models/src/generated/prost.rs index 4bb4af1e85..2480d26e46 100644 --- a/datafusion/proto-models/src/generated/prost.rs +++ b/datafusion/proto-models/src/generated/prost.rs @@ -1833,10 +1833,12 @@ pub struct PhysicalBinaryExprNode { pub r: ::core::option::Option<::prost::alloc::boxed::Box<PhysicalExprNode>>, #[prost(string, tag = "3")] pub op: ::prost::alloc::string::String, - /// Linearized operands for chains of the same operator (e.g. a AND b AND c). + /// Linearized operands for chains of the same operator and overflow policy. /// When present, `l` and `r` are ignored and `operands` holds the flattened list. #[prost(message, repeated, tag = "4")] pub operands: ::prost::alloc::vec::Vec<PhysicalExprNode>, + #[prost(bool, tag = "5")] + pub fail_on_overflow: bool, } #[derive(Clone, PartialEq, ::prost::Message)] pub struct PhysicalDateTimeIntervalExprNode { diff --git a/datafusion/proto/tests/cases/plans/exprs.rs b/datafusion/proto/tests/cases/plans/exprs.rs index f2b14b0439..ee6f9819a3 100644 --- a/datafusion/proto/tests/cases/plans/exprs.rs +++ b/datafusion/proto/tests/cases/plans/exprs.rs @@ -286,6 +286,130 @@ fn roundtrip_call_null_scalar_struct_dict() -> Result<()> { roundtrip_test(filter) } +#[test] +fn roundtrip_binary_expr_overflow() -> Result<()> { + use arrow::record_batch::RecordBatch; + use datafusion_proto::bytes::{physical_plan_from_bytes, physical_plan_to_bytes}; + + let schema = Arc::new(Schema::empty()); + let batch = RecordBatch::new_empty(Arc::clone(&schema)); + for inner_checked in [false, true] { + for outer_checked in [false, true] { + // Overflow occurs in the inner addition. A different outer policy + // must not overwrite it when the encoder linearizes the chain. + let inner = Arc::new( + BinaryExpr::new(lit(i32::MAX), Operator::Plus, lit(1i32)) + .with_fail_on_overflow(inner_checked), + ); + let expr: Arc<dyn PhysicalExpr> = Arc::new( + BinaryExpr::new(inner, Operator::Plus, lit(0i32)) + .with_fail_on_overflow(outer_checked), + ); + let plan = Arc::new(ProjectionExec::try_new( + vec![(Arc::clone(&expr), "result".to_string())], + Arc::new(EmptyExec::new(Arc::clone(&schema))), + )?); + let bytes = physical_plan_to_bytes(plan)?; + let decoded = + physical_plan_from_bytes(&bytes, &SessionContext::new().task_ctx())?; + let decoded = decoded.downcast_ref::<ProjectionExec>().unwrap(); + for expression in [&expr, &decoded.expr()[0].expr] { + let result = expression.evaluate(&batch); + if inner_checked { + assert!(result.unwrap_err().to_string().contains("overflow")); + } else { + assert_eq!( + result?.into_array(1)?.as_ref(), + ScalarValue::Int32(Some(i32::MIN)).to_array()?.as_ref() + ); + } + } + assert!(expr.eq(&decoded.expr()[0].expr)); + } + } + Ok(()) +} + +#[test] +#[cfg(feature = "json")] +fn roundtrip_binary_expr_overflow_legacy() -> Result<()> { + use arrow::record_batch::RecordBatch; + + let schema = Schema::empty(); + let batch = RecordBatch::new_empty(Arc::new(schema.clone())); + let codec = DefaultPhysicalExtensionCodec {}; + let converter = DefaultPhysicalProtoConverter {}; + let task_ctx = SessionContext::new().task_ctx(); + let decode_ctx = PhysicalPlanDecodeContext::new(&task_ctx, &codec); + // An older message has l/r operands and no overflow policy field. + let mut json = serde_json::json!({ + "binaryExpr": { + "l": converter.physical_expr_to_proto(&lit(i32::MAX), &codec)?, + "r": converter.physical_expr_to_proto(&lit(1i32), &codec)?, + "op": "Plus" + } + }); + for checked in [false, true] { + if checked { + json["binaryExpr"]["failOnOverflow"] = true.into(); + } + let proto: protobuf::PhysicalExprNode = + serde_json::from_value(json.clone()).unwrap(); + let decoded = converter.proto_to_physical_expr(&proto, &schema, &decode_ctx)?; + let result = decoded.evaluate(&batch); + if checked { + assert!(result.unwrap_err().to_string().contains("overflow")); + } else { + assert_eq!( + result?.into_array(1)?.as_ref(), + ScalarValue::Int32(Some(i32::MIN)).to_array()?.as_ref() + ); + } + let encoded = converter.physical_expr_to_proto(&decoded, &codec)?; + let json = serde_json::to_value(encoded).unwrap(); + assert_eq!( + json["binaryExpr"]["failOnOverflow"] + .as_bool() + .unwrap_or(false), + checked + ); + } + Ok(()) +} + +#[test] +#[cfg(feature = "json")] +fn roundtrip_binary_expr_overflow_json() -> Result<()> { + use arrow::record_batch::RecordBatch; + + let schema = Schema::empty(); + let codec = DefaultPhysicalExtensionCodec {}; + let converter = DefaultPhysicalProtoConverter {}; + let expr: Arc<dyn PhysicalExpr> = Arc::new( + BinaryExpr::new(lit(i32::MAX), Operator::Plus, lit(1i32)) + .with_fail_on_overflow(true), + ); + + let proto = converter.physical_expr_to_proto(&expr, &codec)?; + let json = serde_json::to_value(proto).unwrap(); + assert_eq!(json["binaryExpr"]["operands"].as_array().unwrap().len(), 2); + assert_eq!(json["binaryExpr"]["failOnOverflow"], true); + + let proto: protobuf::PhysicalExprNode = serde_json::from_value(json).unwrap(); + let task_ctx = SessionContext::new().task_ctx(); + let decode_ctx = PhysicalPlanDecodeContext::new(&task_ctx, &codec); + let decoded = converter.proto_to_physical_expr(&proto, &schema, &decode_ctx)?; + let batch = RecordBatch::new_empty(Arc::new(schema)); + assert!( + decoded + .evaluate(&batch) + .unwrap_err() + .to_string() + .contains("overflow") + ); + Ok(()) +} + /// Test that a chain of the same operator (a AND b AND c) is linearized /// and roundtrips correctly. #[test] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
