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 06:22:37 UTC
(datafusion) branch gh-readonly-queue/main/pr-22550-7bf54dbb5b047e63dc0ce8ca7bde23d5104f1f72 created (now a786471abc)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-22550-7bf54dbb5b047e63dc0ce8ca7bde23d5104f1f72
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at a786471abc refactor: Port TryCastExpr proto serialization hooks (#22550)
This branch includes the following new commits:
new a786471abc refactor: Port TryCastExpr proto serialization hooks (#22550)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: refactor: Port TryCastExpr proto serialization hooks (#22550)
Posted by gi...@apache.org.
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 a786471abc04b7271bf4d764054473f9317fb90c
Author: chakkk309 <ch...@gmail.com>
AuthorDate: Wed Jun 3 14:22:16 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