Skip to content

Commit c286a59

Browse files
kevinhongzlkevinhong
andauthored
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 <kevinhong@cw.com.tw>
1 parent 77240f9 commit c286a59

3 files changed

Lines changed: 147 additions & 21 deletions

File tree

datafusion/physical-expr/src/expressions/negative.rs

Lines changed: 145 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -174,6 +174,43 @@ impl PhysicalExpr for NegativeExpr {
174174
self.arg.fmt_sql(f)?;
175175
write!(f, ")")
176176
}
177+
178+
#[cfg(feature = "proto")]
179+
fn try_to_proto(
180+
&self,
181+
ctx: &datafusion_physical_expr_common::physical_expr::proto_encode::PhysicalExprEncodeCtx<'_>,
182+
) -> Result<Option<datafusion_proto_models::protobuf::PhysicalExprNode>> {
183+
use datafusion_proto_models::protobuf;
184+
185+
Ok(Some(protobuf::PhysicalExprNode {
186+
expr_id: None,
187+
expr_type: Some(protobuf::physical_expr_node::ExprType::Negative(Box::new(
188+
protobuf::PhysicalNegativeNode {
189+
expr: Some(Box::new(ctx.encode_child(&self.arg)?)),
190+
},
191+
))),
192+
}))
193+
}
194+
}
195+
196+
#[cfg(feature = "proto")]
197+
impl NegativeExpr {
198+
/// Reconstruct a [`NegativeExpr`] from its protobuf representation.
199+
pub fn try_from_proto(
200+
node: &datafusion_proto_models::protobuf::PhysicalExprNode,
201+
ctx: &datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx<'_>,
202+
) -> Result<Arc<dyn PhysicalExpr>> {
203+
use datafusion_proto_models::protobuf;
204+
205+
let expr = match &node.expr_type {
206+
Some(protobuf::physical_expr_node::ExprType::Negative(n)) => {
207+
ctx.decode_required_expression(n.expr.as_deref(), "NegativeExpr", "expr")?
208+
}
209+
_ => return internal_err!("PhysicalExprNode is not a Negative"),
210+
};
211+
212+
Ok(Arc::new(NegativeExpr::new(expr)))
213+
}
177214
}
178215

179216
/// Creates a unary expression NEGATIVE
@@ -402,3 +439,111 @@ mod tests {
402439
Ok(())
403440
}
404441
}
442+
443+
#[cfg(all(test, feature = "proto"))]
444+
mod proto_tests {
445+
use super::*;
446+
use crate::expressions::{Column, col};
447+
use crate::proto_test_util::{
448+
StubDecoder, StubEncoder, UnreachableDecoder, column_node,
449+
};
450+
use arrow::datatypes::Field;
451+
use datafusion_common::DataFusionError;
452+
use datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx;
453+
use datafusion_physical_expr_common::physical_expr::proto_encode::PhysicalExprEncodeCtx;
454+
use datafusion_proto_models::protobuf::{
455+
PhysicalExprNode, PhysicalNegativeNode, physical_expr_node,
456+
};
457+
458+
/// Build a `NegativeExpr` proto node with the given children.
459+
fn negative_node(expr: Option<Box<PhysicalExprNode>>) -> PhysicalExprNode {
460+
PhysicalExprNode {
461+
expr_id: None,
462+
expr_type: Some(physical_expr_node::ExprType::Negative(Box::new(
463+
PhysicalNegativeNode { expr },
464+
))),
465+
}
466+
}
467+
468+
/// A `NegativeExpr` over a column of type Int32.
469+
fn negative_fixture() -> NegativeExpr {
470+
let schema = Schema::new(vec![Field::new("a", DataType::Int32, true)]);
471+
NegativeExpr::new(col("a", &schema).unwrap())
472+
}
473+
474+
#[test]
475+
fn try_to_proto_encodes_negative_expr() {
476+
let negative = negative_fixture();
477+
let encoder = StubEncoder::ok();
478+
let ctx = PhysicalExprEncodeCtx::new(&encoder);
479+
480+
let node = negative
481+
.try_to_proto(&ctx)
482+
.unwrap()
483+
.expect("NegativeExpr should encode to Some(node)");
484+
485+
assert!(node.expr_id.is_none());
486+
let negative_node = match node.expr_type {
487+
Some(physical_expr_node::ExprType::Negative(boxed)) => *boxed,
488+
other => panic!("expected a NegativeExpr node, got {other:?}"),
489+
};
490+
assert!(negative_node.expr.is_some());
491+
}
492+
493+
#[test]
494+
fn try_to_proto_propagates_expr_encode_error() {
495+
let negative = negative_fixture();
496+
let encoder = StubEncoder::failing_on(1);
497+
let ctx = PhysicalExprEncodeCtx::new(&encoder);
498+
let err = negative.try_to_proto(&ctx).unwrap_err();
499+
assert!(matches!(err, DataFusionError::Internal(msg) if msg.contains("call 1")));
500+
}
501+
502+
#[test]
503+
fn try_from_proto_decodes_negative_expr() {
504+
let node = negative_node(Some(Box::new(column_node("a"))));
505+
let schema = Schema::empty();
506+
let decoder = StubDecoder::ok();
507+
let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
508+
509+
let decoded = NegativeExpr::try_from_proto(&node, &ctx).unwrap();
510+
let negative = decoded
511+
.downcast_ref::<NegativeExpr>()
512+
.expect("decoded expr should be a NegativeExpr");
513+
assert!(negative.arg().downcast_ref::<Column>().is_some());
514+
}
515+
516+
#[test]
517+
fn try_from_proto_rejects_non_negative_node() {
518+
let node = column_node("a");
519+
let schema = Schema::empty();
520+
let decoder = UnreachableDecoder;
521+
let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
522+
let err = NegativeExpr::try_from_proto(&node, &ctx).unwrap_err();
523+
assert!(
524+
matches!(err, DataFusionError::Internal(msg) if msg.contains("PhysicalExprNode is not a Negative"))
525+
);
526+
}
527+
528+
#[test]
529+
fn try_from_proto_rejects_missing_expr() {
530+
let node = negative_node(None);
531+
let schema = Schema::empty();
532+
let decoder = UnreachableDecoder;
533+
let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
534+
let err = NegativeExpr::try_from_proto(&node, &ctx).unwrap_err();
535+
assert!(
536+
matches!(err, DataFusionError::Internal(msg) if msg.contains("NegativeExpr is missing required field 'expr'"))
537+
);
538+
}
539+
540+
#[test]
541+
fn try_from_proto_propagates_expr_decode_error() {
542+
let node = negative_node(Some(Box::new(column_node("a"))));
543+
let schema = Schema::empty();
544+
let decoder = StubDecoder::failing_on(1);
545+
let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
546+
let err = NegativeExpr::try_from_proto(&node, &ctx).unwrap_err();
547+
assert!(matches!(err, DataFusionError::Internal(msg) if msg.contains("call 1")));
548+
}
549+
}

datafusion/proto/src/physical_plan/from_proto.rs

Lines changed: 1 addition & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -319,15 +319,7 @@ pub fn parse_physical_expr_with_converter(
319319
input_schema,
320320
proto_converter,
321321
)?)),
322-
ExprType::Negative(e) => {
323-
Arc::new(NegativeExpr::new(parse_required_physical_expr(
324-
e.expr.as_deref(),
325-
ctx,
326-
"expr",
327-
input_schema,
328-
proto_converter,
329-
)?))
330-
}
322+
ExprType::Negative(_) => NegativeExpr::try_from_proto(proto, &decode_ctx)?,
331323
ExprType::InList(_) => InListExpr::try_from_proto(proto, &decode_ctx)?,
332324
ExprType::Case(e) => Arc::new(CaseExpr::try_new(
333325
e.expr

datafusion/proto/src/physical_plan/to_proto.rs

Lines changed: 1 addition & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,7 @@ use datafusion_physical_expr::window::{SlidingAggregateWindowExpr, StandardWindo
3737
use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr;
3838
use datafusion_physical_plan::expressions::{
3939
CaseExpr, CastExpr, DynamicFilterPhysicalExpr, IsNotNullExpr, IsNullExpr, Literal,
40-
NegativeExpr, NotExpr, TryCastExpr, UnKnownColumn,
40+
NotExpr, TryCastExpr, UnKnownColumn,
4141
};
4242
use datafusion_physical_plan::joins::HashExpr;
4343
use datafusion_physical_plan::udaf::AggregateFunctionExpr;
@@ -387,17 +387,6 @@ pub fn serialize_physical_expr_with_converter(
387387
}),
388388
)),
389389
})
390-
} else if let Some(expr) = expr.downcast_ref::<NegativeExpr>() {
391-
Ok(protobuf::PhysicalExprNode {
392-
expr_id,
393-
expr_type: Some(protobuf::physical_expr_node::ExprType::Negative(Box::new(
394-
protobuf::PhysicalNegativeNode {
395-
expr: Some(Box::new(
396-
proto_converter.physical_expr_to_proto(expr.arg(), codec)?,
397-
)),
398-
},
399-
))),
400-
})
401390
} else if let Some(lit) = expr.downcast_ref::<Literal>() {
402391
Ok(protobuf::PhysicalExprNode {
403392
expr_id,

0 commit comments

Comments
 (0)