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 c286a59039 port `NegativeExpr` to use the `try_to_proto` /
`try_from_proto` hooks (#22483)
c286a59039 is described below
commit c286a5903936594357e647b0669ffc3f8f333fcc
Author: Zhen-Lun (Kevin) Hong <[email protected]>
AuthorDate: Wed May 27 22:51:11 2026 +0800
port `NegativeExpr` to use the `try_to_proto` / `try_from_proto` hooks
(#22483)
## Which issue does this PR close?
- Closes #22426
## Rationale for this change
This change is part of the per-expression proto hooks migration #22418.
I moved the serialization and deserialization of `NegativeExpr` into its
proto hooks, keeping it aligned with the new pattern used by migrated
physical expressions and reducing special-case branching in the shared
conversion code.
## What changes are included in this PR?
- Added `try_to_proto` and `try_from_proto` to `NegativeExpr`
- Removed the central `NegativeExpr` serialization branch
- Updated `physical_plan/from_proto.rs` to route physical proto decode
through `try_from_proto`
## Are these changes tested?
Yes. This PR is verified by running
- `cargo fmt --all -- --check`
- `cargo check -p datafusion-physical-expr --features proto`
- `cargo check -p datafusion-proto`
- `cargo test -p datafusion-proto --test proto_integration
roundtrip_physical_plan`
- `cargo test -p datafusion-proto --test proto_integration
roundtrip_physical_expr`
- `git diff --check`
## Are there any user-facing changes?
No user-facing changes are intended.
---------
Co-authored-by: kevinhong <[email protected]>
---
.../physical-expr/src/expressions/negative.rs | 145 +++++++++++++++++++++
datafusion/proto/src/physical_plan/from_proto.rs | 10 +-
datafusion/proto/src/physical_plan/to_proto.rs | 13 +-
3 files changed, 147 insertions(+), 21 deletions(-)
diff --git a/datafusion/physical-expr/src/expressions/negative.rs
b/datafusion/physical-expr/src/expressions/negative.rs
index e2bda4c8aa..b3ede9f1e9 100644
--- a/datafusion/physical-expr/src/expressions/negative.rs
+++ b/datafusion/physical-expr/src/expressions/negative.rs
@@ -174,6 +174,43 @@ impl PhysicalExpr for NegativeExpr {
self.arg.fmt_sql(f)?;
write!(f, ")")
}
+
+ #[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;
+
+ Ok(Some(protobuf::PhysicalExprNode {
+ expr_id: None,
+ expr_type:
Some(protobuf::physical_expr_node::ExprType::Negative(Box::new(
+ protobuf::PhysicalNegativeNode {
+ expr: Some(Box::new(ctx.encode_child(&self.arg)?)),
+ },
+ ))),
+ }))
+ }
+}
+
+#[cfg(feature = "proto")]
+impl NegativeExpr {
+ /// Reconstruct a [`NegativeExpr`] from its protobuf representation.
+ 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 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"),
+ };
+
+ Ok(Arc::new(NegativeExpr::new(expr)))
+ }
}
/// Creates a unary expression NEGATIVE
@@ -402,3 +439,111 @@ mod tests {
Ok(())
}
}
+
+#[cfg(all(test, feature = "proto"))]
+mod proto_tests {
+ use super::*;
+ use crate::expressions::{Column, col};
+ use crate::proto_test_util::{
+ StubDecoder, StubEncoder, UnreachableDecoder, column_node,
+ };
+ use arrow::datatypes::Field;
+ use datafusion_common::DataFusionError;
+ use
datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx;
+ use
datafusion_physical_expr_common::physical_expr::proto_encode::PhysicalExprEncodeCtx;
+ use datafusion_proto_models::protobuf::{
+ PhysicalExprNode, PhysicalNegativeNode, physical_expr_node,
+ };
+
+ /// Build a `NegativeExpr` proto node with the given children.
+ fn negative_node(expr: Option<Box<PhysicalExprNode>>) -> PhysicalExprNode {
+ PhysicalExprNode {
+ expr_id: None,
+ expr_type: Some(physical_expr_node::ExprType::Negative(Box::new(
+ PhysicalNegativeNode { expr },
+ ))),
+ }
+ }
+
+ /// A `NegativeExpr` over a column of type Int32.
+ fn negative_fixture() -> NegativeExpr {
+ let schema = Schema::new(vec![Field::new("a", DataType::Int32, true)]);
+ NegativeExpr::new(col("a", &schema).unwrap())
+ }
+
+ #[test]
+ fn try_to_proto_encodes_negative_expr() {
+ let negative = negative_fixture();
+ let encoder = StubEncoder::ok();
+ let ctx = PhysicalExprEncodeCtx::new(&encoder);
+
+ let node = negative
+ .try_to_proto(&ctx)
+ .unwrap()
+ .expect("NegativeExpr should encode to Some(node)");
+
+ assert!(node.expr_id.is_none());
+ let negative_node = match node.expr_type {
+ Some(physical_expr_node::ExprType::Negative(boxed)) => *boxed,
+ other => panic!("expected a NegativeExpr node, got {other:?}"),
+ };
+ assert!(negative_node.expr.is_some());
+ }
+
+ #[test]
+ fn try_to_proto_propagates_expr_encode_error() {
+ let negative = negative_fixture();
+ let encoder = StubEncoder::failing_on(1);
+ let ctx = PhysicalExprEncodeCtx::new(&encoder);
+ let err = negative.try_to_proto(&ctx).unwrap_err();
+ assert!(matches!(err, DataFusionError::Internal(msg) if
msg.contains("call 1")));
+ }
+
+ #[test]
+ fn try_from_proto_decodes_negative_expr() {
+ let node = negative_node(Some(Box::new(column_node("a"))));
+ let schema = Schema::empty();
+ let decoder = StubDecoder::ok();
+ let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
+
+ let decoded = NegativeExpr::try_from_proto(&node, &ctx).unwrap();
+ let negative = decoded
+ .downcast_ref::<NegativeExpr>()
+ .expect("decoded expr should be a NegativeExpr");
+ assert!(negative.arg().downcast_ref::<Column>().is_some());
+ }
+
+ #[test]
+ fn try_from_proto_rejects_non_negative_node() {
+ let node = column_node("a");
+ let schema = Schema::empty();
+ let decoder = UnreachableDecoder;
+ let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
+ let err = NegativeExpr::try_from_proto(&node, &ctx).unwrap_err();
+ assert!(
+ matches!(err, DataFusionError::Internal(msg) if
msg.contains("PhysicalExprNode is not a Negative"))
+ );
+ }
+
+ #[test]
+ fn try_from_proto_rejects_missing_expr() {
+ let node = negative_node(None);
+ let schema = Schema::empty();
+ let decoder = UnreachableDecoder;
+ let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
+ let err = NegativeExpr::try_from_proto(&node, &ctx).unwrap_err();
+ assert!(
+ matches!(err, DataFusionError::Internal(msg) if
msg.contains("NegativeExpr is missing required field 'expr'"))
+ );
+ }
+
+ #[test]
+ fn try_from_proto_propagates_expr_decode_error() {
+ let node = negative_node(Some(Box::new(column_node("a"))));
+ let schema = Schema::empty();
+ let decoder = StubDecoder::failing_on(1);
+ let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
+ let err = NegativeExpr::try_from_proto(&node, &ctx).unwrap_err();
+ assert!(matches!(err, DataFusionError::Internal(msg) if
msg.contains("call 1")));
+ }
+}
diff --git a/datafusion/proto/src/physical_plan/from_proto.rs
b/datafusion/proto/src/physical_plan/from_proto.rs
index 55022608e5..d2a48aa457 100644
--- a/datafusion/proto/src/physical_plan/from_proto.rs
+++ b/datafusion/proto/src/physical_plan/from_proto.rs
@@ -319,15 +319,7 @@ pub fn parse_physical_expr_with_converter(
input_schema,
proto_converter,
)?)),
- ExprType::Negative(e) => {
- Arc::new(NegativeExpr::new(parse_required_physical_expr(
- e.expr.as_deref(),
- ctx,
- "expr",
- input_schema,
- proto_converter,
- )?))
- }
+ ExprType::Negative(_) => NegativeExpr::try_from_proto(proto,
&decode_ctx)?,
ExprType::InList(_) => InListExpr::try_from_proto(proto, &decode_ctx)?,
ExprType::Case(e) => Arc::new(CaseExpr::try_new(
e.expr
diff --git a/datafusion/proto/src/physical_plan/to_proto.rs
b/datafusion/proto/src/physical_plan/to_proto.rs
index c359f651c0..28c0a57e94 100644
--- a/datafusion/proto/src/physical_plan/to_proto.rs
+++ b/datafusion/proto/src/physical_plan/to_proto.rs
@@ -37,7 +37,7 @@ use
datafusion_physical_expr::window::{SlidingAggregateWindowExpr, StandardWindo
use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr;
use datafusion_physical_plan::expressions::{
CaseExpr, CastExpr, DynamicFilterPhysicalExpr, IsNotNullExpr, IsNullExpr,
Literal,
- NegativeExpr, NotExpr, TryCastExpr, UnKnownColumn,
+ NotExpr, TryCastExpr, UnKnownColumn,
};
use datafusion_physical_plan::joins::HashExpr;
use datafusion_physical_plan::udaf::AggregateFunctionExpr;
@@ -387,17 +387,6 @@ pub fn serialize_physical_expr_with_converter(
}),
)),
})
- } else if let Some(expr) = expr.downcast_ref::<NegativeExpr>() {
- Ok(protobuf::PhysicalExprNode {
- expr_id,
- expr_type:
Some(protobuf::physical_expr_node::ExprType::Negative(Box::new(
- protobuf::PhysicalNegativeNode {
- expr: Some(Box::new(
- proto_converter.physical_expr_to_proto(expr.arg(),
codec)?,
- )),
- },
- ))),
- })
} else if let Some(lit) = expr.downcast_ref::<Literal>() {
Ok(protobuf::PhysicalExprNode {
expr_id,
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
