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/10/13 09:06:36 UTC
(datafusion) 01/01: fix(proto): preserve physical cast expression state (#25056)
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-25056-70692f3627b84865535144cf5032cdc6988d9b7d
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 8f5da3a3d5aad96ff9559140a2e03bf2189b2c9b
Author: Burak Şen <bu...@gmail.com>
AuthorDate: Tue Oct 13 09:06:10 2026 +0000
fix(proto): preserve physical cast expression state (#25056)
## Which issue does this PR close?
- Closes #24615.
## Rationale for this change
Physical cast protobuf hooks did not serialize `CastExpr` options or
explicit target-field metadata. A round trip could therefore change cast
behavior, nullability, or metadata. The hooks also accessed fields
individually, so future fields could be omitted silently.
## What changes are included in this PR?
- Exhaustively destructure `CastExpr`, `TryCastExpr`, and their protobuf
messages.
- Preserve explicit target fields for `CAST` and `TRY_CAST` while
keeping legacy type-only payload behavior.
- Preserve all serializable Arrow cast/format options; reject custom
formatter factories instead of silently dropping them.
- Store decoded format strings in owned cast options.
- Regenerate protobuf models and document the `CastExpr::cast_options`
API adjustment.
The protobuf changes are additive and backward compatible.
## What is the testing strategy for this PR?
Added unit coverage for:
- target-field and cast-option round trips
- legacy payloads without the new fields
- explicit default-shaped targets
- malformed target-field types
- unsupported formatter factories
Validated with:
- `cargo test -p datafusion-physical-expr --features proto
expressions::cast::`
- `cargo test -p datafusion-physical-expr --features proto
expressions::try_cast::`
- `cargo clippy -p datafusion-physical-expr -p
datafusion-physical-expr-adapter --all-features --tests -- -D warnings`
- `cargo check -p datafusion-proto --all-features`
- `./ci/scripts/doc_prettier_check.sh --write --allow-dirty`
## Are there any user-facing changes?
Cast expressions now retain their options and explicit output-field
semantics across protobuf round trips. `CastExpr::cast_options()`
returns a borrowing `CastOptions<'_>` value rather than
`&CastOptions<'static>`; the upgrade guide documents migration for code
rebuilding casts.
---
datafusion/common/src/scalar/mod.rs | 2 +-
datafusion/expr-common/src/columnar_value.rs | 4 +-
.../physical-expr-adapter/src/schema_rewriter.rs | 11 +-
datafusion/physical-expr/src/expressions/cast.rs | 498 +++++++++++++++++++--
.../physical-expr/src/expressions/try_cast.rs | 168 ++++++-
datafusion/proto-models/proto/datafusion.proto | 31 ++
datafusion/proto-models/src/generated/pbjson.rs | 488 ++++++++++++++++++++
datafusion/proto-models/src/generated/prost.rs | 67 +++
datafusion/proto/tests/cases/plans/exprs.rs | 56 +++
docs/source/library-user-guide/upgrading/56.0.0.md | 10 +
10 files changed, 1281 insertions(+), 54 deletions(-)
diff --git a/datafusion/common/src/scalar/mod.rs b/datafusion/common/src/scalar/mod.rs
index ca9fc654b7..280aeaec16 100644
--- a/datafusion/common/src/scalar/mod.rs
+++ b/datafusion/common/src/scalar/mod.rs
@@ -4377,7 +4377,7 @@ impl ScalarValue {
pub fn cast_to_with_options(
&self,
target_type: &DataType,
- cast_options: &CastOptions<'static>,
+ cast_options: &CastOptions<'_>,
) -> Result<Self> {
let source_type = self.data_type();
diff --git a/datafusion/expr-common/src/columnar_value.rs b/datafusion/expr-common/src/columnar_value.rs
index ef9192c356..6bab07ed14 100644
--- a/datafusion/expr-common/src/columnar_value.rs
+++ b/datafusion/expr-common/src/columnar_value.rs
@@ -294,7 +294,7 @@ impl ColumnarValue {
pub fn cast_to(
&self,
cast_type: &DataType,
- cast_options: Option<&CastOptions<'static>>,
+ cast_options: Option<&CastOptions<'_>>,
) -> Result<ColumnarValue> {
let cast_options = cast_options.cloned().unwrap_or(DEFAULT_CAST_OPTIONS);
match self {
@@ -312,7 +312,7 @@ impl ColumnarValue {
fn cast_array_by_name(
array: &ArrayRef,
cast_type: &DataType,
- cast_options: &CastOptions<'static>,
+ cast_options: &CastOptions<'_>,
) -> Result<ArrayRef> {
// If types are already equal, no cast needed
if array.data_type() == cast_type {
diff --git a/datafusion/physical-expr-adapter/src/schema_rewriter.rs b/datafusion/physical-expr-adapter/src/schema_rewriter.rs
index 56be58f4dc..83278d751d 100644
--- a/datafusion/physical-expr-adapter/src/schema_rewriter.rs
+++ b/datafusion/physical-expr-adapter/src/schema_rewriter.rs
@@ -570,11 +570,9 @@ impl DefaultPhysicalExprAdapterRewriter {
return Ok(None);
};
let mut args = get_field_expr.args().to_vec();
- args[0] = Arc::new(CastExpr::new_with_target_field(
- Arc::clone(inner),
- target_field,
- Some(cast.cast_options().clone()),
- ));
+ args[0] = Arc::new(
+ cast.with_new_expr_and_target_field(Arc::clone(inner), target_field),
+ );
return Arc::clone(expr).with_new_children(args).map(Some);
}
@@ -598,10 +596,9 @@ impl DefaultPhysicalExprAdapterRewriter {
{
return Ok(Some(extracted));
}
- Ok(Some(Arc::new(CastExpr::new_with_target_field(
+ Ok(Some(Arc::new(cast.with_new_expr_and_target_field(
extracted,
logical_return_field,
- Some(cast.cast_options().clone()),
))))
}
diff --git a/datafusion/physical-expr/src/expressions/cast.rs b/datafusion/physical-expr/src/expressions/cast.rs
index 792687fe6a..3bbb18a22d 100644
--- a/datafusion/physical-expr/src/expressions/cast.rs
+++ b/datafusion/physical-expr/src/expressions/cast.rs
@@ -24,6 +24,7 @@ use crate::physical_expr::PhysicalExpr;
use arrow::compute::{CastOptions, can_cast_types};
use arrow::datatypes::{DataType, DataType::*, Field, FieldRef, Metadata, Schema};
use arrow::record_batch::RecordBatch;
+use arrow::util::display::{ArrayFormatterFactory, DurationFormat, FormatOptions};
use arrow_schema::extension::{EXTENSION_TYPE_METADATA_KEY, EXTENSION_TYPE_NAME_KEY};
use datafusion_common::datatype::DataTypeExt;
use datafusion_common::format::DEFAULT_FORMAT_OPTIONS;
@@ -45,6 +46,94 @@ const DEFAULT_SAFE_CAST_OPTIONS: CastOptions<'static> = CastOptions {
format_options: DEFAULT_FORMAT_OPTIONS,
};
+/// Owns Arrow's borrowed format strings so protobuf decoding does not leak them.
+#[derive(Debug, Clone)]
+struct OwnedCastOptions {
+ safe: bool,
+ format_safe: bool,
+ null: String,
+ date_format: Option<String>,
+ datetime_format: Option<String>,
+ timestamp_format: Option<String>,
+ timestamp_tz_format: Option<String>,
+ time_format: Option<String>,
+ duration_format: DurationFormat,
+ types_info: bool,
+ quoted_strings: bool,
+ formatter_factory: Option<&'static dyn ArrayFormatterFactory>,
+}
+
+impl From<CastOptions<'static>> for OwnedCastOptions {
+ fn from(options: CastOptions<'static>) -> Self {
+ let CastOptions {
+ safe,
+ format_options,
+ } = options;
+ Self {
+ safe,
+ format_safe: format_options.safe(),
+ null: format_options.null().to_owned(),
+ date_format: format_options.date_format().map(str::to_owned),
+ datetime_format: format_options.datetime_format().map(str::to_owned),
+ timestamp_format: format_options.timestamp_format().map(str::to_owned),
+ timestamp_tz_format: format_options.timestamp_tz_format().map(str::to_owned),
+ time_format: format_options.time_format().map(str::to_owned),
+ duration_format: format_options.duration_format(),
+ types_info: format_options.types_info(),
+ quoted_strings: format_options.quoted_strings(),
+ formatter_factory: format_options.formatter_factory(),
+ }
+ }
+}
+
+impl OwnedCastOptions {
+ fn as_arrow(&self) -> CastOptions<'_> {
+ let Self {
+ safe,
+ format_safe,
+ null,
+ date_format,
+ datetime_format,
+ timestamp_format,
+ timestamp_tz_format,
+ time_format,
+ duration_format,
+ types_info,
+ quoted_strings,
+ formatter_factory,
+ } = self;
+ CastOptions {
+ safe: *safe,
+ format_options: FormatOptions::new()
+ .with_display_error(*format_safe)
+ .with_null(null)
+ .with_date_format(date_format.as_deref())
+ .with_datetime_format(datetime_format.as_deref())
+ .with_timestamp_format(timestamp_format.as_deref())
+ .with_timestamp_tz_format(timestamp_tz_format.as_deref())
+ .with_time_format(time_format.as_deref())
+ .with_duration_format(*duration_format)
+ .with_types_info(*types_info)
+ .with_quoted_strings(*quoted_strings)
+ .with_formatter_factory(*formatter_factory),
+ }
+ }
+}
+
+impl PartialEq for OwnedCastOptions {
+ fn eq(&self, other: &Self) -> bool {
+ self.as_arrow() == other.as_arrow()
+ }
+}
+
+impl Eq for OwnedCastOptions {}
+
+impl Hash for OwnedCastOptions {
+ fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+ self.as_arrow().hash(state);
+ }
+}
+
/// Check if name-based struct casting is allowed by validating field compatibility.
///
/// This function applies the same validation rules as execution time to ensure
@@ -73,7 +162,7 @@ pub struct CastExpr {
/// nullability describe the output field exactly.
explicit_target: bool,
/// Cast options
- cast_options: CastOptions<'static>,
+ cast_options: OwnedCastOptions,
}
// Manually derive PartialEq and Hash to work around https://github.com/rust-lang/rust/issues/78808
@@ -126,7 +215,7 @@ impl CastExpr {
expr,
target_field: cast_type.into_nullable_field_ref(),
explicit_target: false,
- cast_options: cast_options.unwrap_or(DEFAULT_CAST_OPTIONS),
+ cast_options: cast_options.unwrap_or(DEFAULT_CAST_OPTIONS).into(),
}
}
@@ -152,7 +241,7 @@ impl CastExpr {
expr,
target_field,
explicit_target: true,
- cast_options: cast_options.unwrap_or(DEFAULT_CAST_OPTIONS),
+ cast_options: cast_options.unwrap_or(DEFAULT_CAST_OPTIONS).into(),
}
}
@@ -194,9 +283,23 @@ impl CastExpr {
&self.target_field
}
- /// The cast options
- pub fn cast_options(&self) -> &CastOptions<'static> {
- &self.cast_options
+ /// The cast options, borrowing custom format strings from this expression.
+ pub fn cast_options(&self) -> CastOptions<'_> {
+ self.cast_options.as_arrow()
+ }
+
+ /// Rebuild this cast with a new child and explicit target while preserving its options.
+ pub fn with_new_expr_and_target_field(
+ &self,
+ expr: Arc<dyn PhysicalExpr>,
+ target_field: FieldRef,
+ ) -> Self {
+ Self {
+ expr,
+ target_field,
+ explicit_target: true,
+ cast_options: self.cast_options.clone(),
+ }
}
/// Whether this cast has explicit metadata (vs pass-through from source).
@@ -352,6 +455,114 @@ pub(crate) fn cast_expr_properties(
}
}
+#[cfg(feature = "proto")]
+fn serialize_cast_options(
+ options: &OwnedCastOptions,
+) -> Result<Option<datafusion_proto_models::protobuf::PhysicalCastOptions>> {
+ use datafusion_proto_models::protobuf;
+
+ let options = options.as_arrow();
+ if options.format_options.formatter_factory().is_some() {
+ return not_impl_err!(
+ "CastExpr serialization does not support a custom formatter_factory"
+ );
+ }
+ if options == DEFAULT_CAST_OPTIONS {
+ return Ok(None);
+ }
+
+ let CastOptions {
+ safe,
+ format_options,
+ } = options;
+ let duration_format = match format_options.duration_format() {
+ DurationFormat::ISO8601 => protobuf::PhysicalDurationFormat::Iso8601,
+ DurationFormat::Pretty => protobuf::PhysicalDurationFormat::Pretty,
+ _ => {
+ return not_impl_err!(
+ "CastExpr serialization does not support this duration format"
+ );
+ }
+ };
+
+ Ok(Some(protobuf::PhysicalCastOptions {
+ safe,
+ format_options: Some(protobuf::PhysicalFormatOptions {
+ safe: format_options.safe(),
+ null: format_options.null().to_owned(),
+ date_format: format_options.date_format().map(str::to_owned),
+ datetime_format: format_options.datetime_format().map(str::to_owned),
+ timestamp_format: format_options.timestamp_format().map(str::to_owned),
+ timestamp_tz_format: format_options.timestamp_tz_format().map(str::to_owned),
+ time_format: format_options.time_format().map(str::to_owned),
+ duration_format: duration_format.into(),
+ types_info: format_options.types_info(),
+ quoted_strings: format_options.quoted_strings(),
+ }),
+ }))
+}
+
+#[cfg(feature = "proto")]
+fn deserialize_cast_options(
+ options: Option<&datafusion_proto_models::protobuf::PhysicalCastOptions>,
+) -> Result<OwnedCastOptions> {
+ use datafusion_common::internal_datafusion_err;
+ use datafusion_proto_models::protobuf;
+
+ let Some(options) = options else {
+ return Ok(DEFAULT_CAST_OPTIONS.into());
+ };
+ let protobuf::PhysicalCastOptions {
+ safe,
+ format_options,
+ } = options;
+ let format_options = format_options.as_ref().ok_or_else(|| {
+ internal_datafusion_err!(
+ "CastExpr cast_options is missing required field 'format_options'"
+ )
+ })?;
+ let protobuf::PhysicalFormatOptions {
+ safe: format_safe,
+ null,
+ date_format,
+ datetime_format,
+ timestamp_format,
+ timestamp_tz_format,
+ time_format,
+ duration_format,
+ types_info,
+ quoted_strings,
+ } = format_options;
+ let duration_format = match protobuf::PhysicalDurationFormat::try_from(
+ *duration_format,
+ )
+ .map_err(|_| {
+ internal_datafusion_err!(
+ "CastExpr has invalid duration format value {duration_format}"
+ )
+ })? {
+ protobuf::PhysicalDurationFormat::Iso8601 => DurationFormat::ISO8601,
+ protobuf::PhysicalDurationFormat::Pretty => DurationFormat::Pretty,
+ };
+
+ Ok(OwnedCastOptions {
+ safe: *safe,
+ format_safe: *format_safe,
+ null: null.clone(),
+ date_format: date_format.clone(),
+ datetime_format: datetime_format.clone(),
+ timestamp_format: timestamp_format.clone(),
+ timestamp_tz_format: timestamp_tz_format.clone(),
+ time_format: time_format.clone(),
+ duration_format,
+ types_info: *types_info,
+ quoted_strings: *quoted_strings,
+ // Runtime formatter factories have no protobuf representation; the
+ // encoder rejects them rather than silently dropping behavior.
+ formatter_factory: None,
+ })
+}
+
impl fmt::Display for CastExpr {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "CAST({} AS {})", self.expr, self.cast_type())
@@ -376,8 +587,9 @@ impl PhysicalExpr for CastExpr {
fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> {
let value = self.expr.evaluate(batch)?;
+ let cast_options = self.cast_options();
value
- .cast_to(self.cast_type(), Some(&self.cast_options))
+ .cast_to(self.cast_type(), Some(&cast_options))
.map_err(|error| {
let source = self
.expr
@@ -416,7 +628,7 @@ impl PhysicalExpr for CastExpr {
fn evaluate_bounds(&self, children: &[&Interval]) -> Result<Interval> {
// Cast current node's interval to the right type:
- children[0].cast_to(self.cast_type(), &self.cast_options)
+ children[0].cast_to(self.cast_type(), &self.cast_options())
}
fn propagate_constraints(
@@ -454,12 +666,28 @@ impl PhysicalExpr for CastExpr {
) -> Result<Option<datafusion_proto_models::protobuf::PhysicalExprNode>> {
use datafusion_proto_models::protobuf;
+ let Self {
+ expr,
+ target_field,
+ explicit_target,
+ cast_options,
+ } = self;
+ // Presence carries `explicit_target`, even for a default-shaped field.
+ let target_field_proto = if *explicit_target {
+ Some(target_field.as_ref().try_into()?)
+ } else {
+ None
+ };
+ let cast_options_proto = serialize_cast_options(cast_options)?;
+
Ok(Some(protobuf::PhysicalExprNode {
expr_id: None,
expr_type: Some(protobuf::physical_expr_node::ExprType::Cast(Box::new(
protobuf::PhysicalCastNode {
- expr: Some(Box::new(ctx.encode_child(self.expr())?)),
- arrow_type: Some(self.cast_type().try_into()?),
+ expr: Some(Box::new(ctx.encode_child(expr)?)),
+ arrow_type: Some(target_field.data_type().try_into()?),
+ target_field: target_field_proto,
+ cast_options: cast_options_proto,
},
))),
}))
@@ -492,27 +720,50 @@ impl CastExpr {
node: &datafusion_proto_models::protobuf::PhysicalExprNode,
ctx: &datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx<'_>,
) -> Result<Arc<dyn PhysicalExpr>> {
- use datafusion_common::internal_datafusion_err;
use datafusion_common::internal_err;
+ 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 cast_expr = match &node.expr_type {
- Some(protobuf::physical_expr_node::ExprType::Cast(cast_expr)) => {
- cast_expr.as_ref()
- }
- _ => return internal_err!("PhysicalExprNode is not a CastExpr"),
- };
-
- let expr = ctx.decode_required_expression(
- cast_expr.expr.as_deref(),
+ let cast_expr = expect_expr_variant!(
+ node,
+ protobuf::physical_expr_node::ExprType::Cast,
"CastExpr",
- "expr",
- )?;
- let arrow_type = cast_expr.arrow_type.as_ref().ok_or_else(|| {
- internal_datafusion_err!("CastExpr is missing required field 'arrow_type'")
- })?;
+ );
+ let protobuf::PhysicalCastNode {
+ expr,
+ arrow_type,
+ target_field,
+ cast_options,
+ } = &**cast_expr;
+ let expr = ctx.decode_required_expression(expr.as_deref(), "CastExpr", "expr")?;
+ let arrow_type =
+ require_proto_field(arrow_type.as_ref(), "CastExpr", "arrow_type")?;
+ let cast_type: DataType = arrow_type.try_into()?;
+ let target_field = target_field
+ .as_ref()
+ .map(|field| {
+ let field: Field = field.try_into()?;
+ if field.data_type() != &cast_type {
+ return internal_err!(
+ "CastExpr target_field type does not match arrow_type"
+ );
+ }
+ Ok(Arc::new(field))
+ })
+ .transpose()?;
+ let cast_options = deserialize_cast_options(cast_options.as_ref())?;
+ let (target_field, explicit_target) = match target_field {
+ Some(target_field) => (target_field, true),
+ None => (cast_type.into_nullable_field_ref(), false),
+ };
- Ok(Arc::new(CastExpr::new(expr, arrow_type.try_into()?, None)))
+ Ok(Arc::new(CastExpr {
+ expr,
+ target_field,
+ explicit_target,
+ cast_options,
+ }))
}
}
@@ -2113,6 +2364,25 @@ mod proto_tests {
PhysicalCastNode, PhysicalExprNode, physical_expr_node,
};
+ #[derive(Debug)]
+ struct TestFormatterFactory;
+
+ impl ArrayFormatterFactory for TestFormatterFactory {
+ fn create_array_formatter<'formatter>(
+ &self,
+ _array: &'formatter dyn arrow::array::Array,
+ _options: &FormatOptions<'formatter>,
+ _field: Option<&'formatter Field>,
+ ) -> std::result::Result<
+ Option<arrow::util::display::ArrayFormatter<'formatter>>,
+ arrow::error::ArrowError,
+ > {
+ Ok(None)
+ }
+ }
+
+ static TEST_FORMATTER_FACTORY: TestFormatterFactory = TestFormatterFactory;
+
/// A `CastExpr` over an `Int32` column, casting to `Int64`.
fn proto_cast_fixture() -> CastExpr {
let schema = Schema::new(vec![Field::new("a", Int32, false)]);
@@ -2123,7 +2393,7 @@ mod proto_tests {
(&Int64).try_into().unwrap()
}
- /// Build a `CastExpr` proto node with the given child and target type.
+ /// Build a legacy `CastExpr` proto node with the given child and target type.
fn proto_cast_node(
expr: Option<Box<PhysicalExprNode>>,
arrow_type: Option<ArrowType>,
@@ -2131,11 +2401,59 @@ mod proto_tests {
PhysicalExprNode {
expr_id: None,
expr_type: Some(physical_expr_node::ExprType::Cast(Box::new(
- PhysicalCastNode { expr, arrow_type },
+ PhysicalCastNode {
+ expr,
+ arrow_type,
+ target_field: None,
+ cast_options: None,
+ },
))),
}
}
+ fn encode_cast(cast: &CastExpr) -> PhysicalCastNode {
+ let encoder = StubEncoder::ok();
+ let node = cast
+ .try_to_proto(&PhysicalExprEncodeCtx::new(&encoder))
+ .unwrap()
+ .expect("CastExpr should encode to Some(node)");
+ match node.expr_type {
+ Some(physical_expr_node::ExprType::Cast(cast)) => *cast,
+ other => panic!("expected a Cast node, got {other:?}"),
+ }
+ }
+
+ fn round_trip_cast(cast: &CastExpr, schema: &Schema) -> CastExpr {
+ let encoder = StubEncoder::ok();
+ let node = cast
+ .try_to_proto(&PhysicalExprEncodeCtx::new(&encoder))
+ .unwrap()
+ .expect("CastExpr should encode to Some(node)");
+ let decoder = StubDecoder::ok();
+ CastExpr::try_from_proto(&node, &PhysicalExprDecodeCtx::new(schema, &decoder))
+ .unwrap()
+ .downcast_ref::<CastExpr>()
+ .expect("decoded expr should be a CastExpr")
+ .clone()
+ }
+
+ fn non_default_cast_options() -> CastOptions<'static> {
+ CastOptions {
+ safe: true,
+ format_options: FormatOptions::new()
+ .with_display_error(false)
+ .with_null("NULL")
+ .with_date_format(Some("%d/%m/%Y"))
+ .with_datetime_format(Some("%d/%m/%Y %H:%M:%S"))
+ .with_timestamp_format(Some("%s"))
+ .with_timestamp_tz_format(Some("%+"))
+ .with_time_format(Some("%H-%M-%S"))
+ .with_duration_format(DurationFormat::ISO8601)
+ .with_types_info(true)
+ .with_quoted_strings(true),
+ }
+ }
+
#[test]
fn try_to_proto_encodes_cast_expr() {
let cast = proto_cast_fixture();
@@ -2160,6 +2478,102 @@ mod proto_tests {
.expect("cast type should be encoded");
let data_type: DataType = arrow_type.try_into().unwrap();
assert_eq!(data_type, Int64);
+ assert!(cast_node.target_field.is_none());
+ assert!(cast_node.cast_options.is_none());
+ }
+
+ #[test]
+ fn cast_target_field_survives_proto_round_trip() {
+ let schema = Schema::new(vec![Field::new("a", Int32, true)]);
+ let target = Arc::new(Field::new("target", Int64, false).with_metadata(
+ Metadata::from([("extension".to_string(), "value".to_string())]),
+ ));
+ let cast = CastExpr::new_with_target_field(
+ col("a", &schema).unwrap(),
+ Arc::clone(&target),
+ None,
+ );
+
+ assert!(encode_cast(&cast).target_field.is_some());
+ let decoded = round_trip_cast(&cast, &schema);
+
+ assert_eq!(decoded.target_field(), &target);
+ assert_eq!(decoded.target_metadata(), Some(target.metadata()));
+ assert_eq!(decoded.target_nullable(), Some(false));
+ assert!(!decoded.return_field(&schema).unwrap().is_nullable());
+ }
+
+ #[test]
+ fn explicit_default_shaped_cast_target_is_encoded() {
+ let schema = Schema::new(vec![Field::new("a", Int32, false).with_metadata(
+ Metadata::from([("source".to_string(), "value".to_string())]),
+ )]);
+ let cast = CastExpr::new_with_target_field(
+ col("a", &schema).unwrap(),
+ Int64.into_nullable_field_ref(),
+ None,
+ );
+
+ assert!(encode_cast(&cast).target_field.is_some());
+ let decoded = round_trip_cast(&cast, &schema);
+ assert_eq!(decoded.target_metadata(), Some(&Metadata::default()));
+ assert_eq!(decoded.target_nullable(), Some(true));
+ let output = decoded.return_field(&schema).unwrap();
+ assert!(output.metadata().is_empty());
+ assert!(output.is_nullable());
+ }
+
+ #[test]
+ fn type_only_cast_target_is_omitted_and_remains_type_only() {
+ let metadata = Metadata::from([("source".to_string(), "value".to_string())]);
+ let schema = Schema::new(vec![
+ Field::new("a", Int32, false).with_metadata(metadata.clone()),
+ ]);
+ let cast = CastExpr::new(col("a", &schema).unwrap(), Int64, None);
+
+ assert!(encode_cast(&cast).target_field.is_none());
+ let decoded = round_trip_cast(&cast, &schema);
+
+ assert_eq!(decoded.target_metadata(), None);
+ assert_eq!(decoded.target_nullable(), None);
+ let output = decoded.return_field(&schema).unwrap();
+ assert_eq!(output.metadata(), &metadata);
+ assert!(!output.is_nullable());
+ }
+
+ #[test]
+ fn cast_options_survive_proto_round_trip() {
+ let schema = Schema::new(vec![Field::new("a", Int32, false)]);
+ let options = non_default_cast_options();
+ let cast =
+ CastExpr::new(col("a", &schema).unwrap(), Int64, Some(options.clone()));
+
+ assert!(encode_cast(&cast).cast_options.is_some());
+ assert_eq!(round_trip_cast(&cast, &schema).cast_options(), options);
+ }
+
+ #[test]
+ fn custom_formatter_factory_is_rejected() {
+ let schema = Schema::new(vec![Field::new("a", Int32, false)]);
+ let cast = CastExpr::new(
+ col("a", &schema).unwrap(),
+ Int64,
+ Some(CastOptions {
+ safe: false,
+ format_options: DEFAULT_FORMAT_OPTIONS
+ .with_formatter_factory(Some(&TEST_FORMATTER_FACTORY)),
+ }),
+ );
+ let encoder = StubEncoder::ok();
+
+ let err = cast
+ .try_to_proto(&PhysicalExprEncodeCtx::new(&encoder))
+ .unwrap_err();
+ assert!(matches!(
+ err,
+ DataFusionError::NotImplemented(msg)
+ if msg.contains("custom formatter_factory")
+ ));
}
#[test]
@@ -2176,7 +2590,7 @@ mod proto_tests {
}
#[test]
- fn try_from_proto_decodes_cast_expr() {
+ fn try_from_proto_decodes_legacy_cast_expr() {
let node = proto_cast_node(
Some(Box::new(column_node("a"))),
Some(proto_int64_arrow_type()),
@@ -2192,6 +2606,32 @@ mod proto_tests {
assert_eq!(cast.cast_type(), &Int64);
assert!(cast.expr().downcast_ref::<Column>().is_some());
+ assert_eq!(cast.target_metadata(), None);
+ assert_eq!(cast.target_nullable(), None);
+ assert_eq!(cast.cast_options(), DEFAULT_CAST_OPTIONS);
+ }
+
+ #[test]
+ fn try_from_proto_rejects_mismatched_target_field_type() {
+ let mut node = proto_cast_node(
+ Some(Box::new(column_node("a"))),
+ Some(proto_int64_arrow_type()),
+ );
+ let Some(physical_expr_node::ExprType::Cast(cast)) = node.expr_type.as_mut()
+ else {
+ unreachable!()
+ };
+ cast.target_field =
+ Some((&Field::new("target", Int32, true)).try_into().unwrap());
+ let schema = Schema::empty();
+ let decoder = StubDecoder::ok();
+
+ let err = CastExpr::try_from_proto(
+ &node,
+ &PhysicalExprDecodeCtx::new(&schema, &decoder),
+ )
+ .unwrap_err();
+ assert!(err.to_string().contains("target_field type"));
}
#[test]
diff --git a/datafusion/physical-expr/src/expressions/try_cast.rs b/datafusion/physical-expr/src/expressions/try_cast.rs
index 7e445a275d..fb6101affb 100644
--- a/datafusion/physical-expr/src/expressions/try_cast.rs
+++ b/datafusion/physical-expr/src/expressions/try_cast.rs
@@ -223,12 +223,25 @@ impl PhysicalExpr for TryCastExpr {
) -> Result<Option<datafusion_proto_models::protobuf::PhysicalExprNode>> {
use datafusion_proto_models::protobuf;
+ let Self {
+ expr,
+ target_field,
+ explicit_target,
+ } = self;
+ // Presence carries `explicit_target`, even for a default-shaped field.
+ let target_field_proto = if *explicit_target {
+ Some(target_field.as_ref().try_into()?)
+ } else {
+ None
+ };
+
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()?),
+ expr: Some(Box::new(ctx.encode_child(expr)?)),
+ arrow_type: Some(target_field.data_type().try_into()?),
+ target_field: target_field_proto,
},
))),
}))
@@ -242,6 +255,7 @@ impl TryCastExpr {
node: &datafusion_proto_models::protobuf::PhysicalExprNode,
ctx: &datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx<'_>,
) -> Result<Arc<dyn PhysicalExpr>> {
+ use datafusion_common::internal_err;
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;
@@ -251,19 +265,33 @@ impl TryCastExpr {
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 protobuf::PhysicalTryCastNode {
+ expr,
+ arrow_type,
+ target_field,
+ } = &**try_cast;
+ let expr =
+ ctx.decode_required_expression(expr.as_deref(), "TryCastExpr", "expr")?;
+ let arrow_type =
+ require_proto_field(arrow_type.as_ref(), "TryCastExpr", "arrow_type")?;
let cast_type: DataType = arrow_type.try_into()?;
+ let target_field = target_field
+ .as_ref()
+ .map(|field| {
+ let field: Field = field.try_into()?;
+ if field.data_type() != &cast_type {
+ return internal_err!(
+ "TryCastExpr target_field type does not match arrow_type"
+ );
+ }
+ Ok(Arc::new(field))
+ })
+ .transpose()?;
- Ok(Arc::new(TryCastExpr::new(expr, cast_type)))
+ Ok(Arc::new(match target_field {
+ Some(target_field) => TryCastExpr::new_with_target_field(expr, target_field),
+ None => TryCastExpr::new(expr, cast_type),
+ }))
}
}
@@ -989,11 +1017,41 @@ mod proto_tests {
PhysicalExprNode {
expr_id: None,
expr_type: Some(physical_expr_node::ExprType::TryCast(Box::new(
- PhysicalTryCastNode { expr, arrow_type },
+ PhysicalTryCastNode {
+ expr,
+ arrow_type,
+ target_field: None,
+ },
))),
}
}
+ fn encode_try_cast(try_cast: &TryCastExpr) -> PhysicalTryCastNode {
+ let encoder = StubEncoder::ok();
+ let node = try_cast
+ .try_to_proto(&PhysicalExprEncodeCtx::new(&encoder))
+ .unwrap()
+ .expect("TryCastExpr should encode to Some(node)");
+ match node.expr_type {
+ Some(physical_expr_node::ExprType::TryCast(try_cast)) => *try_cast,
+ other => panic!("expected a TryCastExpr node, got {other:?}"),
+ }
+ }
+
+ fn round_trip_try_cast(try_cast: &TryCastExpr, schema: &Schema) -> TryCastExpr {
+ let encoder = StubEncoder::ok();
+ let node = try_cast
+ .try_to_proto(&PhysicalExprEncodeCtx::new(&encoder))
+ .unwrap()
+ .expect("TryCastExpr should encode to Some(node)");
+ let decoder = StubDecoder::ok();
+ TryCastExpr::try_from_proto(&node, &PhysicalExprDecodeCtx::new(schema, &decoder))
+ .unwrap()
+ .downcast_ref::<TryCastExpr>()
+ .expect("decoded expr should be a TryCastExpr")
+ .clone()
+ }
+
#[test]
fn try_to_proto_encodes_try_cast_expr() {
let try_cast = try_cast_fixture();
@@ -1018,6 +1076,60 @@ mod proto_tests {
.expect("try cast type should be encoded");
let data_type: DataType = arrow_type.try_into().unwrap();
assert_eq!(data_type, DataType::Int32);
+ assert!(try_cast_node.target_field.is_none());
+ }
+
+ #[test]
+ fn try_cast_target_field_survives_proto_round_trip() {
+ let schema = Schema::new(vec![Field::new("a", DataType::Utf8, false)]);
+ let target =
+ Arc::new(Field::new("target", DataType::Int32, false).with_metadata(
+ Metadata::from([("extension".to_string(), "value".to_string())]),
+ ));
+ let try_cast = TryCastExpr::new_with_target_field(
+ col("a", &schema).unwrap(),
+ Arc::clone(&target),
+ );
+
+ assert!(encode_try_cast(&try_cast).target_field.is_some());
+ let decoded = round_trip_try_cast(&try_cast, &schema);
+
+ assert_eq!(decoded.target_field(), &target);
+ assert_eq!(decoded.target_metadata(), Some(target.metadata()));
+ }
+
+ #[test]
+ fn explicit_default_shaped_try_cast_target_is_encoded() {
+ let schema =
+ Schema::new(vec![Field::new("a", DataType::Utf8, false).with_metadata(
+ Metadata::from([("source".to_string(), "value".to_string())]),
+ )]);
+ let try_cast = TryCastExpr::new_with_target_field(
+ col("a", &schema).unwrap(),
+ DataType::Int32.into_nullable_field_ref(),
+ );
+
+ assert!(encode_try_cast(&try_cast).target_field.is_some());
+ let decoded = round_trip_try_cast(&try_cast, &schema);
+ assert_eq!(decoded.target_metadata(), Some(&Metadata::default()));
+ assert!(decoded.return_field(&schema).unwrap().metadata().is_empty());
+ }
+
+ #[test]
+ fn type_only_try_cast_target_is_omitted_and_remains_type_only() {
+ let metadata = Metadata::from([("source".to_string(), "value".to_string())]);
+ let schema = Schema::new(vec![
+ Field::new("a", DataType::Utf8, false).with_metadata(metadata.clone()),
+ ]);
+ let try_cast = TryCastExpr::new(col("a", &schema).unwrap(), DataType::Int32);
+
+ assert!(encode_try_cast(&try_cast).target_field.is_none());
+ let decoded = round_trip_try_cast(&try_cast, &schema);
+
+ assert_eq!(decoded.target_metadata(), None);
+ let output = decoded.return_field(&schema).unwrap();
+ assert_eq!(output.metadata(), &metadata);
+ assert!(output.is_nullable());
}
#[test]
@@ -1030,7 +1142,7 @@ mod proto_tests {
}
#[test]
- fn try_from_proto_decodes_try_cast_expr() {
+ fn try_from_proto_decodes_legacy_try_cast_expr() {
let node =
try_cast_node(Some(Box::new(column_node("a"))), Some(int32_arrow_type()));
let schema = Schema::empty();
@@ -1044,6 +1156,32 @@ mod proto_tests {
assert_eq!(try_cast.cast_type(), &DataType::Int32);
assert!(try_cast.expr().downcast_ref::<Column>().is_some());
+ assert_eq!(try_cast.target_metadata(), None);
+ }
+
+ #[test]
+ fn try_from_proto_rejects_mismatched_target_field_type() {
+ let mut node =
+ try_cast_node(Some(Box::new(column_node("a"))), Some(int32_arrow_type()));
+ let Some(physical_expr_node::ExprType::TryCast(try_cast)) =
+ node.expr_type.as_mut()
+ else {
+ unreachable!()
+ };
+ try_cast.target_field = Some(
+ (&Field::new("target", DataType::Int64, true))
+ .try_into()
+ .unwrap(),
+ );
+ let schema = Schema::empty();
+ let decoder = StubDecoder::ok();
+
+ let err = TryCastExpr::try_from_proto(
+ &node,
+ &PhysicalExprDecodeCtx::new(&schema, &decoder),
+ )
+ .unwrap_err();
+ assert!(err.to_string().contains("target_field type"));
}
#[test]
diff --git a/datafusion/proto-models/proto/datafusion.proto b/datafusion/proto-models/proto/datafusion.proto
index b82efd3b2b..ba5dda6e8a 100644
--- a/datafusion/proto-models/proto/datafusion.proto
+++ b/datafusion/proto-models/proto/datafusion.proto
@@ -1250,11 +1250,42 @@ message PhysicalCaseNode {
message PhysicalTryCastNode {
PhysicalExprNode expr = 1;
datafusion_common.ArrowType arrow_type = 2;
+ // Present only for an explicit target. Absent nodes retain the legacy
+ // type-only behavior, inheriting metadata from the input expression.
+ optional datafusion_common.Field target_field = 3;
}
message PhysicalCastNode {
PhysicalExprNode expr = 1;
datafusion_common.ArrowType arrow_type = 2;
+ // Present only for an explicit target. Absent nodes retain the legacy
+ // type-only behavior, inheriting metadata and nullability from the input.
+ optional datafusion_common.Field target_field = 3;
+ // Absent means DataFusion's default cast options.
+ optional PhysicalCastOptions cast_options = 4;
+}
+
+message PhysicalCastOptions {
+ bool safe = 1;
+ PhysicalFormatOptions format_options = 2;
+}
+
+enum PhysicalDurationFormat {
+ PHYSICAL_DURATION_FORMAT_ISO8601 = 0;
+ PHYSICAL_DURATION_FORMAT_PRETTY = 1;
+}
+
+message PhysicalFormatOptions {
+ bool safe = 1;
+ string null = 2;
+ optional string date_format = 3;
+ optional string datetime_format = 4;
+ optional string timestamp_format = 5;
+ optional string timestamp_tz_format = 6;
+ optional string time_format = 7;
+ PhysicalDurationFormat duration_format = 8;
+ bool types_info = 9;
+ bool quoted_strings = 10;
}
message PhysicalNegativeNode {
diff --git a/datafusion/proto-models/src/generated/pbjson.rs b/datafusion/proto-models/src/generated/pbjson.rs
index 01b7f02563..05e066e1e2 100644
--- a/datafusion/proto-models/src/generated/pbjson.rs
+++ b/datafusion/proto-models/src/generated/pbjson.rs
@@ -19077,6 +19077,12 @@ impl serde::Serialize for PhysicalCastNode {
if self.arrow_type.is_some() {
len += 1;
}
+ if self.target_field.is_some() {
+ len += 1;
+ }
+ if self.cast_options.is_some() {
+ len += 1;
+ }
let mut struct_ser = serializer.serialize_struct("datafusion.PhysicalCastNode", len)?;
if let Some(v) = self.expr.as_ref() {
struct_ser.serialize_field("expr", v)?;
@@ -19084,6 +19090,12 @@ impl serde::Serialize for PhysicalCastNode {
if let Some(v) = self.arrow_type.as_ref() {
struct_ser.serialize_field("arrowType", v)?;
}
+ if let Some(v) = self.target_field.as_ref() {
+ struct_ser.serialize_field("targetField", v)?;
+ }
+ if let Some(v) = self.cast_options.as_ref() {
+ struct_ser.serialize_field("castOptions", v)?;
+ }
struct_ser.end()
}
}
@@ -19097,12 +19109,18 @@ impl<'de> serde::Deserialize<'de> for PhysicalCastNode {
"expr",
"arrow_type",
"arrowType",
+ "target_field",
+ "targetField",
+ "cast_options",
+ "castOptions",
];
#[allow(clippy::enum_variant_names)]
enum GeneratedField {
Expr,
ArrowType,
+ TargetField,
+ CastOptions,
}
impl<'de> serde::Deserialize<'de> for GeneratedField {
fn deserialize<D>(deserializer: D) -> std::result::Result<GeneratedField, D::Error>
@@ -19126,6 +19144,8 @@ impl<'de> serde::Deserialize<'de> for PhysicalCastNode {
match value {
"expr" => Ok(GeneratedField::Expr),
"arrowType" | "arrow_type" => Ok(GeneratedField::ArrowType),
+ "targetField" | "target_field" => Ok(GeneratedField::TargetField),
+ "castOptions" | "cast_options" => Ok(GeneratedField::CastOptions),
_ => Err(serde::de::Error::unknown_field(value, FIELDS)),
}
}
@@ -19147,6 +19167,8 @@ impl<'de> serde::Deserialize<'de> for PhysicalCastNode {
{
let mut expr__ = None;
let mut arrow_type__ = None;
+ let mut target_field__ = None;
+ let mut cast_options__ = None;
while let Some(k) = map_.next_key()? {
match k {
GeneratedField::Expr => {
@@ -19161,17 +19183,140 @@ impl<'de> serde::Deserialize<'de> for PhysicalCastNode {
}
arrow_type__ = map_.next_value()?;
}
+ GeneratedField::TargetField => {
+ if target_field__.is_some() {
+ return Err(serde::de::Error::duplicate_field("targetField"));
+ }
+ target_field__ = map_.next_value()?;
+ }
+ GeneratedField::CastOptions => {
+ if cast_options__.is_some() {
+ return Err(serde::de::Error::duplicate_field("castOptions"));
+ }
+ cast_options__ = map_.next_value()?;
+ }
}
}
Ok(PhysicalCastNode {
expr: expr__,
arrow_type: arrow_type__,
+ target_field: target_field__,
+ cast_options: cast_options__,
})
}
}
deserializer.deserialize_struct("datafusion.PhysicalCastNode", FIELDS, GeneratedVisitor)
}
}
+impl serde::Serialize for PhysicalCastOptions {
+ #[allow(deprecated)]
+ fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
+ where
+ S: serde::Serializer,
+ {
+ use serde::ser::SerializeStruct;
+ let mut len = 0;
+ if self.safe {
+ len += 1;
+ }
+ if self.format_options.is_some() {
+ len += 1;
+ }
+ let mut struct_ser = serializer.serialize_struct("datafusion.PhysicalCastOptions", len)?;
+ if self.safe {
+ struct_ser.serialize_field("safe", &self.safe)?;
+ }
+ if let Some(v) = self.format_options.as_ref() {
+ struct_ser.serialize_field("formatOptions", v)?;
+ }
+ struct_ser.end()
+ }
+}
+impl<'de> serde::Deserialize<'de> for PhysicalCastOptions {
+ #[allow(deprecated)]
+ fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
+ where
+ D: serde::Deserializer<'de>,
+ {
+ const FIELDS: &[&str] = &[
+ "safe",
+ "format_options",
+ "formatOptions",
+ ];
+
+ #[allow(clippy::enum_variant_names)]
+ enum GeneratedField {
+ Safe,
+ FormatOptions,
+ }
+ impl<'de> serde::Deserialize<'de> for GeneratedField {
+ fn deserialize<D>(deserializer: D) -> std::result::Result<GeneratedField, D::Error>
+ where
+ D: serde::Deserializer<'de>,
+ {
+ struct GeneratedVisitor;
+
+ impl serde::de::Visitor<'_> for GeneratedVisitor {
+ type Value = GeneratedField;
+
+ fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ write!(formatter, "expected one of: {:?}", &FIELDS)
+ }
+
+ #[allow(unused_variables)]
+ fn visit_str<E>(self, value: &str) -> std::result::Result<GeneratedField, E>
+ where
+ E: serde::de::Error,
+ {
+ match value {
+ "safe" => Ok(GeneratedField::Safe),
+ "formatOptions" | "format_options" => Ok(GeneratedField::FormatOptions),
+ _ => Err(serde::de::Error::unknown_field(value, FIELDS)),
+ }
+ }
+ }
+ deserializer.deserialize_identifier(GeneratedVisitor)
+ }
+ }
+ struct GeneratedVisitor;
+ impl<'de> serde::de::Visitor<'de> for GeneratedVisitor {
+ type Value = PhysicalCastOptions;
+
+ fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ formatter.write_str("struct datafusion.PhysicalCastOptions")
+ }
+
+ fn visit_map<V>(self, mut map_: V) -> std::result::Result<PhysicalCastOptions, V::Error>
+ where
+ V: serde::de::MapAccess<'de>,
+ {
+ let mut safe__ = None;
+ let mut format_options__ = None;
+ while let Some(k) = map_.next_key()? {
+ match k {
+ GeneratedField::Safe => {
+ if safe__.is_some() {
+ return Err(serde::de::Error::duplicate_field("safe"));
+ }
+ safe__ = Some(map_.next_value()?);
+ }
+ GeneratedField::FormatOptions => {
+ if format_options__.is_some() {
+ return Err(serde::de::Error::duplicate_field("formatOptions"));
+ }
+ format_options__ = map_.next_value()?;
+ }
+ }
+ }
+ Ok(PhysicalCastOptions {
+ safe: safe__.unwrap_or_default(),
+ format_options: format_options__,
+ })
+ }
+ }
+ deserializer.deserialize_struct("datafusion.PhysicalCastOptions", FIELDS, GeneratedVisitor)
+ }
+}
impl serde::Serialize for PhysicalColumn {
#[allow(deprecated)]
fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
@@ -19407,6 +19552,77 @@ impl<'de> serde::Deserialize<'de> for PhysicalDateTimeIntervalExprNode {
deserializer.deserialize_struct("datafusion.PhysicalDateTimeIntervalExprNode", FIELDS, GeneratedVisitor)
}
}
+impl serde::Serialize for PhysicalDurationFormat {
+ #[allow(deprecated)]
+ fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
+ where
+ S: serde::Serializer,
+ {
+ let variant = match self {
+ Self::Iso8601 => "PHYSICAL_DURATION_FORMAT_ISO8601",
+ Self::Pretty => "PHYSICAL_DURATION_FORMAT_PRETTY",
+ };
+ serializer.serialize_str(variant)
+ }
+}
+impl<'de> serde::Deserialize<'de> for PhysicalDurationFormat {
+ #[allow(deprecated)]
+ fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
+ where
+ D: serde::Deserializer<'de>,
+ {
+ const FIELDS: &[&str] = &[
+ "PHYSICAL_DURATION_FORMAT_ISO8601",
+ "PHYSICAL_DURATION_FORMAT_PRETTY",
+ ];
+
+ struct GeneratedVisitor;
+
+ impl serde::de::Visitor<'_> for GeneratedVisitor {
+ type Value = PhysicalDurationFormat;
+
+ fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ write!(formatter, "expected one of: {:?}", &FIELDS)
+ }
+
+ fn visit_i64<E>(self, v: i64) -> std::result::Result<Self::Value, E>
+ where
+ E: serde::de::Error,
+ {
+ i32::try_from(v)
+ .ok()
+ .and_then(|x| x.try_into().ok())
+ .ok_or_else(|| {
+ serde::de::Error::invalid_value(serde::de::Unexpected::Signed(v), &self)
+ })
+ }
+
+ fn visit_u64<E>(self, v: u64) -> std::result::Result<Self::Value, E>
+ where
+ E: serde::de::Error,
+ {
+ i32::try_from(v)
+ .ok()
+ .and_then(|x| x.try_into().ok())
+ .ok_or_else(|| {
+ serde::de::Error::invalid_value(serde::de::Unexpected::Unsigned(v), &self)
+ })
+ }
+
+ fn visit_str<E>(self, value: &str) -> std::result::Result<Self::Value, E>
+ where
+ E: serde::de::Error,
+ {
+ match value {
+ "PHYSICAL_DURATION_FORMAT_ISO8601" => Ok(PhysicalDurationFormat::Iso8601),
+ "PHYSICAL_DURATION_FORMAT_PRETTY" => Ok(PhysicalDurationFormat::Pretty),
+ _ => Err(serde::de::Error::unknown_variant(value, FIELDS)),
+ }
+ }
+ }
+ deserializer.deserialize_any(GeneratedVisitor)
+ }
+}
impl serde::Serialize for PhysicalDynamicFilterNode {
#[allow(deprecated)]
fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
@@ -20273,6 +20489,260 @@ impl<'de> serde::Deserialize<'de> for PhysicalExtensionNode {
deserializer.deserialize_struct("datafusion.PhysicalExtensionNode", FIELDS, GeneratedVisitor)
}
}
+impl serde::Serialize for PhysicalFormatOptions {
+ #[allow(deprecated)]
+ fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
+ where
+ S: serde::Serializer,
+ {
+ use serde::ser::SerializeStruct;
+ let mut len = 0;
+ if self.safe {
+ len += 1;
+ }
+ if !self.null.is_empty() {
+ len += 1;
+ }
+ if self.date_format.is_some() {
+ len += 1;
+ }
+ if self.datetime_format.is_some() {
+ len += 1;
+ }
+ if self.timestamp_format.is_some() {
+ len += 1;
+ }
+ if self.timestamp_tz_format.is_some() {
+ len += 1;
+ }
+ if self.time_format.is_some() {
+ len += 1;
+ }
+ if self.duration_format != 0 {
+ len += 1;
+ }
+ if self.types_info {
+ len += 1;
+ }
+ if self.quoted_strings {
+ len += 1;
+ }
+ let mut struct_ser = serializer.serialize_struct("datafusion.PhysicalFormatOptions", len)?;
+ if self.safe {
+ struct_ser.serialize_field("safe", &self.safe)?;
+ }
+ if !self.null.is_empty() {
+ struct_ser.serialize_field("null", &self.null)?;
+ }
+ if let Some(v) = self.date_format.as_ref() {
+ struct_ser.serialize_field("dateFormat", v)?;
+ }
+ if let Some(v) = self.datetime_format.as_ref() {
+ struct_ser.serialize_field("datetimeFormat", v)?;
+ }
+ if let Some(v) = self.timestamp_format.as_ref() {
+ struct_ser.serialize_field("timestampFormat", v)?;
+ }
+ if let Some(v) = self.timestamp_tz_format.as_ref() {
+ struct_ser.serialize_field("timestampTzFormat", v)?;
+ }
+ if let Some(v) = self.time_format.as_ref() {
+ struct_ser.serialize_field("timeFormat", v)?;
+ }
+ if self.duration_format != 0 {
+ let v = PhysicalDurationFormat::try_from(self.duration_format)
+ .map_err(|_| serde::ser::Error::custom(format!("Invalid variant {}", self.duration_format)))?;
+ struct_ser.serialize_field("durationFormat", &v)?;
+ }
+ if self.types_info {
+ struct_ser.serialize_field("typesInfo", &self.types_info)?;
+ }
+ if self.quoted_strings {
+ struct_ser.serialize_field("quotedStrings", &self.quoted_strings)?;
+ }
+ struct_ser.end()
+ }
+}
+impl<'de> serde::Deserialize<'de> for PhysicalFormatOptions {
+ #[allow(deprecated)]
+ fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
+ where
+ D: serde::Deserializer<'de>,
+ {
+ const FIELDS: &[&str] = &[
+ "safe",
+ "null",
+ "date_format",
+ "dateFormat",
+ "datetime_format",
+ "datetimeFormat",
+ "timestamp_format",
+ "timestampFormat",
+ "timestamp_tz_format",
+ "timestampTzFormat",
+ "time_format",
+ "timeFormat",
+ "duration_format",
+ "durationFormat",
+ "types_info",
+ "typesInfo",
+ "quoted_strings",
+ "quotedStrings",
+ ];
+
+ #[allow(clippy::enum_variant_names)]
+ enum GeneratedField {
+ Safe,
+ Null,
+ DateFormat,
+ DatetimeFormat,
+ TimestampFormat,
+ TimestampTzFormat,
+ TimeFormat,
+ DurationFormat,
+ TypesInfo,
+ QuotedStrings,
+ }
+ impl<'de> serde::Deserialize<'de> for GeneratedField {
+ fn deserialize<D>(deserializer: D) -> std::result::Result<GeneratedField, D::Error>
+ where
+ D: serde::Deserializer<'de>,
+ {
+ struct GeneratedVisitor;
+
+ impl serde::de::Visitor<'_> for GeneratedVisitor {
+ type Value = GeneratedField;
+
+ fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ write!(formatter, "expected one of: {:?}", &FIELDS)
+ }
+
+ #[allow(unused_variables)]
+ fn visit_str<E>(self, value: &str) -> std::result::Result<GeneratedField, E>
+ where
+ E: serde::de::Error,
+ {
+ match value {
+ "safe" => Ok(GeneratedField::Safe),
+ "null" => Ok(GeneratedField::Null),
+ "dateFormat" | "date_format" => Ok(GeneratedField::DateFormat),
+ "datetimeFormat" | "datetime_format" => Ok(GeneratedField::DatetimeFormat),
+ "timestampFormat" | "timestamp_format" => Ok(GeneratedField::TimestampFormat),
+ "timestampTzFormat" | "timestamp_tz_format" => Ok(GeneratedField::TimestampTzFormat),
+ "timeFormat" | "time_format" => Ok(GeneratedField::TimeFormat),
+ "durationFormat" | "duration_format" => Ok(GeneratedField::DurationFormat),
+ "typesInfo" | "types_info" => Ok(GeneratedField::TypesInfo),
+ "quotedStrings" | "quoted_strings" => Ok(GeneratedField::QuotedStrings),
+ _ => Err(serde::de::Error::unknown_field(value, FIELDS)),
+ }
+ }
+ }
+ deserializer.deserialize_identifier(GeneratedVisitor)
+ }
+ }
+ struct GeneratedVisitor;
+ impl<'de> serde::de::Visitor<'de> for GeneratedVisitor {
+ type Value = PhysicalFormatOptions;
+
+ fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ formatter.write_str("struct datafusion.PhysicalFormatOptions")
+ }
+
+ fn visit_map<V>(self, mut map_: V) -> std::result::Result<PhysicalFormatOptions, V::Error>
+ where
+ V: serde::de::MapAccess<'de>,
+ {
+ let mut safe__ = None;
+ let mut null__ = None;
+ let mut date_format__ = None;
+ let mut datetime_format__ = None;
+ let mut timestamp_format__ = None;
+ let mut timestamp_tz_format__ = None;
+ let mut time_format__ = None;
+ let mut duration_format__ = None;
+ let mut types_info__ = None;
+ let mut quoted_strings__ = None;
+ while let Some(k) = map_.next_key()? {
+ match k {
+ GeneratedField::Safe => {
+ if safe__.is_some() {
+ return Err(serde::de::Error::duplicate_field("safe"));
+ }
+ safe__ = Some(map_.next_value()?);
+ }
+ GeneratedField::Null => {
+ if null__.is_some() {
+ return Err(serde::de::Error::duplicate_field("null"));
+ }
+ null__ = Some(map_.next_value()?);
+ }
+ GeneratedField::DateFormat => {
+ if date_format__.is_some() {
+ return Err(serde::de::Error::duplicate_field("dateFormat"));
+ }
+ date_format__ = map_.next_value()?;
+ }
+ GeneratedField::DatetimeFormat => {
+ if datetime_format__.is_some() {
+ return Err(serde::de::Error::duplicate_field("datetimeFormat"));
+ }
+ datetime_format__ = map_.next_value()?;
+ }
+ GeneratedField::TimestampFormat => {
+ if timestamp_format__.is_some() {
+ return Err(serde::de::Error::duplicate_field("timestampFormat"));
+ }
+ timestamp_format__ = map_.next_value()?;
+ }
+ GeneratedField::TimestampTzFormat => {
+ if timestamp_tz_format__.is_some() {
+ return Err(serde::de::Error::duplicate_field("timestampTzFormat"));
+ }
+ timestamp_tz_format__ = map_.next_value()?;
+ }
+ GeneratedField::TimeFormat => {
+ if time_format__.is_some() {
+ return Err(serde::de::Error::duplicate_field("timeFormat"));
+ }
+ time_format__ = map_.next_value()?;
+ }
+ GeneratedField::DurationFormat => {
+ if duration_format__.is_some() {
+ return Err(serde::de::Error::duplicate_field("durationFormat"));
+ }
+ duration_format__ = Some(map_.next_value::<PhysicalDurationFormat>()? as i32);
+ }
+ GeneratedField::TypesInfo => {
+ if types_info__.is_some() {
+ return Err(serde::de::Error::duplicate_field("typesInfo"));
+ }
+ types_info__ = Some(map_.next_value()?);
+ }
+ GeneratedField::QuotedStrings => {
+ if quoted_strings__.is_some() {
+ return Err(serde::de::Error::duplicate_field("quotedStrings"));
+ }
+ quoted_strings__ = Some(map_.next_value()?);
+ }
+ }
+ }
+ Ok(PhysicalFormatOptions {
+ safe: safe__.unwrap_or_default(),
+ null: null__.unwrap_or_default(),
+ date_format: date_format__,
+ datetime_format: datetime_format__,
+ timestamp_format: timestamp_format__,
+ timestamp_tz_format: timestamp_tz_format__,
+ time_format: time_format__,
+ duration_format: duration_format__.unwrap_or_default(),
+ types_info: types_info__.unwrap_or_default(),
+ quoted_strings: quoted_strings__.unwrap_or_default(),
+ })
+ }
+ }
+ deserializer.deserialize_struct("datafusion.PhysicalFormatOptions", FIELDS, GeneratedVisitor)
+ }
+}
impl serde::Serialize for PhysicalHashExprNode {
#[allow(deprecated)]
fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
@@ -23238,6 +23708,9 @@ impl serde::Serialize for PhysicalTryCastNode {
if self.arrow_type.is_some() {
len += 1;
}
+ if self.target_field.is_some() {
+ len += 1;
+ }
let mut struct_ser = serializer.serialize_struct("datafusion.PhysicalTryCastNode", len)?;
if let Some(v) = self.expr.as_ref() {
struct_ser.serialize_field("expr", v)?;
@@ -23245,6 +23718,9 @@ impl serde::Serialize for PhysicalTryCastNode {
if let Some(v) = self.arrow_type.as_ref() {
struct_ser.serialize_field("arrowType", v)?;
}
+ if let Some(v) = self.target_field.as_ref() {
+ struct_ser.serialize_field("targetField", v)?;
+ }
struct_ser.end()
}
}
@@ -23258,12 +23734,15 @@ impl<'de> serde::Deserialize<'de> for PhysicalTryCastNode {
"expr",
"arrow_type",
"arrowType",
+ "target_field",
+ "targetField",
];
#[allow(clippy::enum_variant_names)]
enum GeneratedField {
Expr,
ArrowType,
+ TargetField,
}
impl<'de> serde::Deserialize<'de> for GeneratedField {
fn deserialize<D>(deserializer: D) -> std::result::Result<GeneratedField, D::Error>
@@ -23287,6 +23766,7 @@ impl<'de> serde::Deserialize<'de> for PhysicalTryCastNode {
match value {
"expr" => Ok(GeneratedField::Expr),
"arrowType" | "arrow_type" => Ok(GeneratedField::ArrowType),
+ "targetField" | "target_field" => Ok(GeneratedField::TargetField),
_ => Err(serde::de::Error::unknown_field(value, FIELDS)),
}
}
@@ -23308,6 +23788,7 @@ impl<'de> serde::Deserialize<'de> for PhysicalTryCastNode {
{
let mut expr__ = None;
let mut arrow_type__ = None;
+ let mut target_field__ = None;
while let Some(k) = map_.next_key()? {
match k {
GeneratedField::Expr => {
@@ -23322,11 +23803,18 @@ impl<'de> serde::Deserialize<'de> for PhysicalTryCastNode {
}
arrow_type__ = map_.next_value()?;
}
+ GeneratedField::TargetField => {
+ if target_field__.is_some() {
+ return Err(serde::de::Error::duplicate_field("targetField"));
+ }
+ target_field__ = map_.next_value()?;
+ }
}
}
Ok(PhysicalTryCastNode {
expr: expr__,
arrow_type: arrow_type__,
+ target_field: target_field__,
})
}
}
diff --git a/datafusion/proto-models/src/generated/prost.rs b/datafusion/proto-models/src/generated/prost.rs
index 61ba256704..4bfa7c6fe8 100644
--- a/datafusion/proto-models/src/generated/prost.rs
+++ b/datafusion/proto-models/src/generated/prost.rs
@@ -1935,6 +1935,10 @@ pub struct PhysicalTryCastNode {
pub expr: ::core::option::Option<::prost::alloc::boxed::Box<PhysicalExprNode>>,
#[prost(message, optional, tag = "2")]
pub arrow_type: ::core::option::Option<super::datafusion_common::ArrowType>,
+ /// Present only for an explicit target. Absent nodes retain the legacy
+ /// type-only behavior, inheriting metadata from the input expression.
+ #[prost(message, optional, tag = "3")]
+ pub target_field: ::core::option::Option<super::datafusion_common::Field>,
}
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct PhysicalCastNode {
@@ -1942,6 +1946,43 @@ pub struct PhysicalCastNode {
pub expr: ::core::option::Option<::prost::alloc::boxed::Box<PhysicalExprNode>>,
#[prost(message, optional, tag = "2")]
pub arrow_type: ::core::option::Option<super::datafusion_common::ArrowType>,
+ /// Present only for an explicit target. Absent nodes retain the legacy
+ /// type-only behavior, inheriting metadata and nullability from the input.
+ #[prost(message, optional, tag = "3")]
+ pub target_field: ::core::option::Option<super::datafusion_common::Field>,
+ /// Absent means DataFusion's default cast options.
+ #[prost(message, optional, tag = "4")]
+ pub cast_options: ::core::option::Option<PhysicalCastOptions>,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct PhysicalCastOptions {
+ #[prost(bool, tag = "1")]
+ pub safe: bool,
+ #[prost(message, optional, tag = "2")]
+ pub format_options: ::core::option::Option<PhysicalFormatOptions>,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct PhysicalFormatOptions {
+ #[prost(bool, tag = "1")]
+ pub safe: bool,
+ #[prost(string, tag = "2")]
+ pub null: ::prost::alloc::string::String,
+ #[prost(string, optional, tag = "3")]
+ pub date_format: ::core::option::Option<::prost::alloc::string::String>,
+ #[prost(string, optional, tag = "4")]
+ pub datetime_format: ::core::option::Option<::prost::alloc::string::String>,
+ #[prost(string, optional, tag = "5")]
+ pub timestamp_format: ::core::option::Option<::prost::alloc::string::String>,
+ #[prost(string, optional, tag = "6")]
+ pub timestamp_tz_format: ::core::option::Option<::prost::alloc::string::String>,
+ #[prost(string, optional, tag = "7")]
+ pub time_format: ::core::option::Option<::prost::alloc::string::String>,
+ #[prost(enumeration = "PhysicalDurationFormat", tag = "8")]
+ pub duration_format: i32,
+ #[prost(bool, tag = "9")]
+ pub types_info: bool,
+ #[prost(bool, tag = "10")]
+ pub quoted_strings: bool,
}
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct PhysicalNegativeNode {
@@ -3027,6 +3068,32 @@ impl InsertOp {
}
}
}
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
+#[repr(i32)]
+pub enum PhysicalDurationFormat {
+ Iso8601 = 0,
+ Pretty = 1,
+}
+impl PhysicalDurationFormat {
+ /// String value of the enum field names used in the ProtoBuf definition.
+ ///
+ /// The values are not transformed in any way and thus are considered stable
+ /// (if the ProtoBuf definition does not change) and safe for programmatic use.
+ pub fn as_str_name(&self) -> &'static str {
+ match self {
+ Self::Iso8601 => "PHYSICAL_DURATION_FORMAT_ISO8601",
+ Self::Pretty => "PHYSICAL_DURATION_FORMAT_PRETTY",
+ }
+ }
+ /// Creates an enum from field names used in the ProtoBuf definition.
+ pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
+ match value {
+ "PHYSICAL_DURATION_FORMAT_ISO8601" => Some(Self::Iso8601),
+ "PHYSICAL_DURATION_FORMAT_PRETTY" => Some(Self::Pretty),
+ _ => None,
+ }
+ }
+}
/// Identifies which Arrow IPC format an ArrowScanExecNode reads.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
#[repr(i32)]
diff --git a/datafusion/proto/tests/cases/plans/exprs.rs b/datafusion/proto/tests/cases/plans/exprs.rs
index aea02f258d..84978ee3d7 100644
--- a/datafusion/proto/tests/cases/plans/exprs.rs
+++ b/datafusion/proto/tests/cases/plans/exprs.rs
@@ -755,6 +755,62 @@ fn roundtrip_binary_expr_overflow_json() -> Result<()> {
Ok(())
}
+#[tokio::test]
+async fn roundtrip_cast_safe_option_execution() -> Result<()> {
+ use arrow::array::{Array, Int32Array};
+ use arrow::compute::CastOptions;
+ use datafusion::physical_plan::collect;
+ use datafusion::physical_plan::expressions::CastExpr;
+ use datafusion::physical_plan::placeholder_row::PlaceholderRowExec;
+ use datafusion_common::assert_contains;
+ use datafusion_common::format::DEFAULT_CAST_OPTIONS;
+ use datafusion_proto::bytes::{physical_plan_from_bytes, physical_plan_to_bytes};
+
+ let ctx = SessionContext::new();
+ // `safe: false` matches DataFusion's default cast options, which the encoder
+ // omits from the wire; `safe: true` must be carried by the encoded options.
+ for safe in [true, false] {
+ let cast_options = CastOptions {
+ safe,
+ ..DEFAULT_CAST_OPTIONS
+ };
+ // A safe cast can turn a non-null input into NULL. Use an explicit
+ // nullable target: a type-only cast would inherit the literal's
+ // non-nullability, and the projection would reject the NULL.
+ let expr: Arc<dyn PhysicalExpr> = Arc::new(CastExpr::new_with_target_field(
+ lit("9.1"),
+ Arc::new(Field::new("result", DataType::Int32, true)),
+ Some(cast_options),
+ ));
+ let original: Arc<dyn ExecutionPlan> = Arc::new(ProjectionExec::try_new(
+ vec![(expr, "result".to_string())],
+ Arc::new(PlaceholderRowExec::new(Arc::new(Schema::empty()))),
+ )?);
+ let bytes = physical_plan_to_bytes(Arc::clone(&original))?;
+ let decoded = physical_plan_from_bytes(&bytes, &ctx.task_ctx())?;
+
+ for (label, plan) in [("original", original), ("decoded", decoded)] {
+ match (safe, collect(plan, ctx.task_ctx()).await) {
+ (true, Ok(batches)) => {
+ assert_eq!(batches.len(), 1, "{label} plan batch count");
+ assert_eq!(
+ batches[0].column(0).as_any().downcast_ref::<Int32Array>(),
+ Some(&Int32Array::new_null(1)),
+ "{label} plan should cast '9.1' to NULL"
+ );
+ }
+ (false, Err(e)) => {
+ assert_contains!(e.to_string(), "Cannot cast string '9.1'");
+ }
+ (_, other) => {
+ panic!("{label} plan with safe={safe} returned {other:?}")
+ }
+ }
+ }
+ }
+ Ok(())
+}
+
/// Test that a chain of the same operator (a AND b AND c) is linearized
/// and roundtrips correctly.
#[test]
diff --git a/docs/source/library-user-guide/upgrading/56.0.0.md b/docs/source/library-user-guide/upgrading/56.0.0.md
index 62af520611..cd60aa365c 100644
--- a/docs/source/library-user-guide/upgrading/56.0.0.md
+++ b/docs/source/library-user-guide/upgrading/56.0.0.md
@@ -296,6 +296,16 @@ let constraints = Constraints::try_from(proto_constraints)?;
See [issue #24170](https://github.com/apache/datafusion/issues/24170) for details.
+### `CastExpr::cast_options` now returns `CastOptions` by value
+
+`CastExpr` now owns format strings decoded from protobuf, so `cast_options()`
+returns a borrowing `CastOptions<'_>` value instead of `&CastOptions<'static>`.
+Code that only reads the options generally needs no change. To replace only the
+child, use `PhysicalExpr::with_new_children` to preserve the options and whether
+the target is type-only or explicit. Use `CastExpr::with_new_expr_and_target_field`
+when replacing both the child and an explicit target field. Neither method
+requires a `'static` borrow of the options.
+
### `ExecutionOptions` has a new `enable_nlj_coordinated_fallback` field
`ExecutionOptions` gained a public `enable_nlj_coordinated_fallback: bool` field
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org