This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 714956b3a8 fix: preserve overflow behavior and exhaustively
destructure expression proto hooks (#25034)
714956b3a8 is described below
commit 714956b3a8aed43e8abd27ed400710417d231982
Author: Peter Lee <[email protected]>
AuthorDate: Tue Sep 22 09:25:15 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.
---------
Co-authored-by: Andrew Lamb <[email protected]>
---
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]
