You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/06/03 05:46:49 UTC
(datafusion) 01/01: refactor: Port TryCastExpr proto serialization hooks (#22550)
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-22550-7bf54dbb5b047e63dc0ce8ca7bde23d5104f1f72
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 0c7673b8a92b4c8484eaf0d6580dea177a001ecf
Author: chakkk309 <ch...@gmail.com>
AuthorDate: Wed Jun 3 13:46:24 2026 +0800
refactor: Port TryCastExpr proto serialization hooks (#22550)
## Which issue does this PR close?
- Closes #22429.
## Rationale for this change
This is part of #22418, which migrates built-in `PhysicalExpr`
implementations away from the central protobuf serialization /
deserialization chains.
`TryCastExpr` can now own its protobuf serialization through
`PhysicalExpr::try_to_proto` and its deserialization through
`TryCastExpr::try_from_proto`, matching the pattern used by previously
migrated physical expressions.
## What changes are included in this PR?
- Adds `PhysicalExpr::try_to_proto` support for `TryCastExpr`.
- Adds `TryCastExpr::try_from_proto`.
- Wires `TryCastExpr` deserialization in `from_proto.rs` through the new
hook.
- Removes the old `TryCastExpr` serialization branch from the central
`to_proto.rs` downcast chain.
- Adds direct proto hook tests for successful encoding / decoding and
invalid protobuf inputs.
## Are these changes tested?
Yes.
I ran:
```bash
cargo fmt --all --check
git diff --check
cargo test -p datafusion-physical-expr --features proto try_cast
cargo test -p datafusion-proto --test proto_integration
cargo clippy -p datafusion-physical-expr --features proto --tests -- -D warnings
cargo clippy -p datafusion-proto --tests -- -D warnings
```
## Are there any user-facing changes?
No. This is an internal protobuf serialization refactor for
`TryCastExpr`; the wire format remains unchanged.
Co-authored-by: chakkk309 <ch...@users.noreply.github.com>
---
.../physical-expr/src/expressions/try_cast.rs | 190 +++++++++++++++++++++
datafusion/proto/src/physical_plan/from_proto.rs | 11 +-
datafusion/proto/src/physical_plan/to_proto.rs | 14 +-
3 files changed, 192 insertions(+), 23 deletions(-)
diff --git a/datafusion/physical-expr/src/expressions/try_cast.rs b/datafusion/physical-expr/src/expressions/try_cast.rs
index ba59d113ac..65b953fd18 100644
--- a/datafusion/physical-expr/src/expressions/try_cast.rs
+++ b/datafusion/physical-expr/src/expressions/try_cast.rs
@@ -119,6 +119,56 @@ impl PhysicalExpr for TryCastExpr {
self.expr.fmt_sql(f)?;
write!(f, " AS {:?})", self.cast_type)
}
+
+ #[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::TryCast(Box::new(
+ protobuf::PhysicalTryCastNode {
+ expr: Some(Box::new(ctx.encode_child(&self.expr)?)),
+ arrow_type: Some(self.cast_type().try_into()?),
+ },
+ ))),
+ }))
+ }
+}
+
+#[cfg(feature = "proto")]
+impl TryCastExpr {
+ /// Reconstruct a [`TryCastExpr`] 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_physical_expr_common::expect_expr_variant;
+ use datafusion_physical_expr_common::physical_expr::proto_decode::require_proto_field;
+ use datafusion_proto_models::protobuf;
+
+ let try_cast = expect_expr_variant!(
+ node,
+ protobuf::physical_expr_node::ExprType::TryCast,
+ "TryCastExpr",
+ );
+ let expr = ctx.decode_required_expression(
+ try_cast.expr.as_deref(),
+ "TryCastExpr",
+ "expr",
+ )?;
+ let arrow_type = require_proto_field(
+ try_cast.arrow_type.as_ref(),
+ "TryCastExpr",
+ "arrow_type",
+ )?;
+ let cast_type: DataType = arrow_type.try_into()?;
+
+ Ok(Arc::new(TryCastExpr::new(expr, cast_type)))
+ }
}
/// Return a PhysicalExpression representing `expr` casted to
@@ -593,3 +643,143 @@ 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::datafusion_common::ArrowType;
+ use datafusion_proto_models::protobuf::{
+ PhysicalExprNode, PhysicalTryCastNode, physical_expr_node,
+ };
+
+ fn try_cast_fixture() -> TryCastExpr {
+ let schema = Schema::new(vec![Field::new("a", DataType::Utf8, true)]);
+ TryCastExpr::new(col("a", &schema).unwrap(), DataType::Int32)
+ }
+
+ fn int32_arrow_type() -> ArrowType {
+ (&DataType::Int32).try_into().unwrap()
+ }
+
+ fn try_cast_node(
+ expr: Option<Box<PhysicalExprNode>>,
+ arrow_type: Option<ArrowType>,
+ ) -> PhysicalExprNode {
+ PhysicalExprNode {
+ expr_id: None,
+ expr_type: Some(physical_expr_node::ExprType::TryCast(Box::new(
+ PhysicalTryCastNode { expr, arrow_type },
+ ))),
+ }
+ }
+
+ #[test]
+ fn try_to_proto_encodes_try_cast_expr() {
+ let try_cast = try_cast_fixture();
+ let encoder = StubEncoder::ok();
+ let ctx = PhysicalExprEncodeCtx::new(&encoder);
+
+ let node = try_cast
+ .try_to_proto(&ctx)
+ .unwrap()
+ .expect("TryCastExpr should encode to Some(node)");
+
+ assert!(node.expr_id.is_none());
+ let try_cast_node = match node.expr_type {
+ Some(physical_expr_node::ExprType::TryCast(boxed)) => *boxed,
+ other => panic!("expected a TryCastExpr node, got {other:?}"),
+ };
+ assert!(try_cast_node.expr.is_some());
+
+ let arrow_type = try_cast_node
+ .arrow_type
+ .as_ref()
+ .expect("try cast type should be encoded");
+ let data_type: DataType = arrow_type.try_into().unwrap();
+ assert_eq!(data_type, DataType::Int32);
+ }
+
+ #[test]
+ fn try_to_proto_propagates_child_encode_error() {
+ let try_cast = try_cast_fixture();
+ let encoder = StubEncoder::failing_on(1);
+ let ctx = PhysicalExprEncodeCtx::new(&encoder);
+ let err = try_cast.try_to_proto(&ctx).unwrap_err();
+ assert!(matches!(err, DataFusionError::Internal(msg) if msg.contains("call 1")));
+ }
+
+ #[test]
+ fn try_from_proto_decodes_try_cast_expr() {
+ let node =
+ try_cast_node(Some(Box::new(column_node("a"))), Some(int32_arrow_type()));
+ let schema = Schema::empty();
+ let decoder = StubDecoder::ok();
+ let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
+
+ let decoded = TryCastExpr::try_from_proto(&node, &ctx).unwrap();
+ let try_cast = decoded
+ .downcast_ref::<TryCastExpr>()
+ .expect("decoded expr should be a TryCastExpr");
+
+ assert_eq!(try_cast.cast_type(), &DataType::Int32);
+ assert!(try_cast.expr().downcast_ref::<Column>().is_some());
+ }
+
+ #[test]
+ fn try_from_proto_rejects_non_try_cast_node() {
+ let node = column_node("a");
+ let schema = Schema::empty();
+ let decoder = UnreachableDecoder;
+ let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
+
+ let err = TryCastExpr::try_from_proto(&node, &ctx).unwrap_err();
+ assert!(
+ matches!(err, DataFusionError::Internal(msg) if msg.contains("PhysicalExprNode is not a TryCastExpr"))
+ );
+ }
+
+ #[test]
+ fn try_from_proto_rejects_missing_expr() {
+ let node = try_cast_node(None, Some(int32_arrow_type()));
+ let schema = Schema::empty();
+ let decoder = UnreachableDecoder;
+ let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
+
+ let err = TryCastExpr::try_from_proto(&node, &ctx).unwrap_err();
+ assert!(
+ matches!(err, DataFusionError::Internal(msg) if msg.contains("TryCastExpr is missing required field 'expr'"))
+ );
+ }
+
+ #[test]
+ fn try_from_proto_rejects_missing_arrow_type() {
+ let node = try_cast_node(Some(Box::new(column_node("a"))), None);
+ let schema = Schema::empty();
+ let decoder = StubDecoder::ok();
+ let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
+
+ let err = TryCastExpr::try_from_proto(&node, &ctx).unwrap_err();
+ assert!(
+ matches!(err, DataFusionError::Internal(msg) if msg.contains("TryCastExpr is missing required field 'arrow_type'"))
+ );
+ }
+
+ #[test]
+ fn try_from_proto_propagates_child_decode_error() {
+ let node =
+ try_cast_node(Some(Box::new(column_node("a"))), Some(int32_arrow_type()));
+ let schema = Schema::empty();
+ let decoder = StubDecoder::failing_on(1);
+ let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
+ let err = TryCastExpr::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 402f30caf7..21d700de89 100644
--- a/datafusion/proto/src/physical_plan/from_proto.rs
+++ b/datafusion/proto/src/physical_plan/from_proto.rs
@@ -339,16 +339,7 @@ pub fn parse_physical_expr_with_converter(
.transpose()?,
)?),
ExprType::Cast(_) => CastExpr::try_from_proto(proto, &decode_ctx)?,
- ExprType::TryCast(e) => Arc::new(TryCastExpr::new(
- parse_required_physical_expr(
- e.expr.as_deref(),
- ctx,
- "expr",
- input_schema,
- proto_converter,
- )?,
- convert_required!(e.arrow_type)?,
- )),
+ ExprType::TryCast(_) => TryCastExpr::try_from_proto(proto, &decode_ctx)?,
ExprType::ScalarUdf(e) => {
let udf = match &e.fun_definition {
Some(buf) => ctx.codec().try_decode_udf(&e.name, buf)?,
diff --git a/datafusion/proto/src/physical_plan/to_proto.rs b/datafusion/proto/src/physical_plan/to_proto.rs
index f8419c006b..096ed46935 100644
--- a/datafusion/proto/src/physical_plan/to_proto.rs
+++ b/datafusion/proto/src/physical_plan/to_proto.rs
@@ -36,7 +36,7 @@ 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::{
- CaseExpr, DynamicFilterPhysicalExpr, Literal, TryCastExpr,
+ CaseExpr, DynamicFilterPhysicalExpr, Literal,
};
use datafusion_physical_plan::udaf::AggregateFunctionExpr;
use datafusion_physical_plan::windows::{PlainAggregateWindowExpr, WindowUDFExpr};
@@ -352,18 +352,6 @@ pub fn serialize_physical_expr_with_converter(
lit.value().try_into()?,
)),
})
- } else if let Some(cast) = expr.downcast_ref::<TryCastExpr>() {
- Ok(protobuf::PhysicalExprNode {
- expr_id,
- expr_type: Some(protobuf::physical_expr_node::ExprType::TryCast(Box::new(
- protobuf::PhysicalTryCastNode {
- expr: Some(Box::new(
- proto_converter.physical_expr_to_proto(cast.expr(), codec)?,
- )),
- arrow_type: Some(cast.cast_type().try_into()?),
- },
- ))),
- })
} else if let Some(expr) = expr.downcast_ref::<ScalarFunctionExpr>() {
let mut buf = Vec::new();
codec.try_encode_udf(expr.fun(), &mut buf)?;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org