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-25500-9951f0b351b751d26b66085190ac969ecd53b34d in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit b172f73f054d9dd6d51663ff713a9747d4f245a3 Author: Huaijin <[email protected]> AuthorDate: Thu Sep 24 12:05:17 2026 +0000 feat: recognize more lossless casts for statistics and ordering (#25500) ## Which issue does this PR close? Follow-up to #25227 and [its discussion](https://github.com/apache/datafusion/pull/25227#discussion_r3999343833). ## Rationale for this change Some lossless casts unnecessarily discard column statistics or fail to preserve strict ordering information. For example, UInt32 → Int64 currently requires both bounds to be exact to propagate statistics, even though every UInt32 value fits in Int64. Recognizing this conversion as globally safe preserves available statistics even when one bound is absent or bounds are inexact. ## What changes are included in this PR? - Recognize unsigned-to-signed widening casts: UInt8 → Int16 / Int32 / Int64, UInt16 → Int32 / Int64, and UInt32 → Int64. - Recognize Utf8 → Utf8View and Binary → LargeBinary / BinaryView. - Extend the existing `check_bigger_cast` / `is_bigger_cast` predicates to recognize these lossless conversions. ## What is the testing strategy for this PR? - Test values, nulls, integer extrema, and strict ordering properties. - Test empty values, Unicode, long strings, and non-UTF8 binary values. - Test statistics propagation with exact, inexact, and missing bounds. - Verify unsafe reverse conversions remain excluded. Validation on the latest revision: 25 relevant unit tests passed; one existing test is ignored. Formatting, all-target/all-feature Clippy, and the standard `rust_lint.sh` suite passed. ## Are there any user-facing changes? More lossless casts retain statistics and strict ordering information, enabling existing optimizer rules. --- datafusion/physical-expr/src/expressions/cast.rs | 190 ++++++++++++++++++++++- datafusion/physical-expr/src/projection.rs | 75 +++++++++ 2 files changed, 258 insertions(+), 7 deletions(-) diff --git a/datafusion/physical-expr/src/expressions/cast.rs b/datafusion/physical-expr/src/expressions/cast.rs index 60e68ca79d..792687fe6a 100644 --- a/datafusion/physical-expr/src/expressions/cast.rs +++ b/datafusion/physical-expr/src/expressions/cast.rs @@ -250,7 +250,7 @@ impl CastExpr { /// Check if casting from the source type to the target type is known to be /// lossless and strictly order-preserving for all source values, preserving nulls. - /// This includes widening casts (e.g. `Int8` to `Int16`) and representation + /// This includes widening casts (e.g. `Int8` or `UInt8` to `Int16`) and representation /// conversions such as `Int32` to `Date32`, which interprets the same integer /// as days since the epoch, or `Int64` to `Date64`, which interprets the same /// integer as milliseconds since the epoch. @@ -267,12 +267,13 @@ impl CastExpr { | (Date32, Int32) | (Int64, Date64) | (Date64, Int64) - | (UInt8, UInt16 | UInt32 | UInt64) - | (UInt16, UInt32 | UInt64) - | (UInt32, UInt64) + | (UInt8, UInt16 | UInt32 | UInt64 | Int16 | Int32 | Int64) + | (UInt16, UInt32 | UInt64 | Int32 | Int64) + | (UInt32, UInt64 | Int64) | (Int8 | Int16 | UInt8 | UInt16, Float32) | (Int8 | Int16 | Int32 | UInt8 | UInt16 | UInt32, Float64) - | (Utf8, LargeUtf8) + | (Utf8, LargeUtf8 | Utf8View) + | (Binary, LargeBinary | BinaryView) ) } @@ -1729,6 +1730,176 @@ mod tests { } } + #[test] + fn test_byte_representation_cast_preserves_values_and_ordering() -> Result<()> { + use arrow::array::{ + BinaryArray, BinaryViewArray, LargeBinaryArray, StringViewArray, + }; + use arrow::compute::SortOptions; + use datafusion_expr_common::sort_properties::SortProperties; + + // Cover nulls, empty values, inline/long views, Unicode, and non-UTF8 bytes. + let strings = vec![ + None, + Some(""), + Some("a"), + Some("a longer shared string"), + Some("a longer shared string"), + Some("🦀"), + ]; + let bytes: Vec<Option<&[u8]>> = vec![ + None, + Some(b""), + Some(b"\0"), + Some(b"a longer shared byte string"), + Some(b"a longer shared byte string"), + Some(b"\xff"), + ]; + let binary: ArrayRef = Arc::new(BinaryArray::from(bytes.clone())); + let cases: [(ArrayRef, ArrayRef); 3] = [ + ( + Arc::new(StringArray::from(strings.clone())), + Arc::new(StringViewArray::from(strings)), + ), + ( + Arc::clone(&binary), + Arc::new(LargeBinaryArray::from(bytes.clone())), + ), + (binary, Arc::new(BinaryViewArray::from(bytes))), + ]; + for (input, expected) in cases { + let schema = Arc::new(Schema::new(vec![Field::new( + "a", + input.data_type().clone(), + true, + )])); + let expr = + CastExpr::new(col("a", &schema)?, expected.data_type().clone(), None); + assert!(expr.is_bigger_cast(input.data_type())); + for descending in [false, true] { + for nulls_first in [false, true] { + let child = ExprProperties::new_unknown() + .with_range(Interval::make_unbounded(input.data_type())?) + .with_order(SortProperties::Ordered(SortOptions { + descending, + nulls_first, + })) + .with_strictly_order_preserving(true); + let properties = expr.get_properties(std::slice::from_ref(&child))?; + assert_eq!(properties.sort_properties, child.sort_properties); + assert!(properties.strictly_order_preserving); + assert_eq!(properties.range.data_type(), *expected.data_type()); + } + } + let batch = RecordBatch::try_new(schema, vec![input])?; + let actual = expr.evaluate(&batch)?.into_array(batch.num_rows())?; + assert_eq!(actual.as_ref(), expected.as_ref()); + } + for (source, target) in [ + (Utf8View, Utf8), + (LargeBinary, Binary), + (BinaryView, Binary), + ] { + assert!(!CastExpr::check_bigger_cast(&target, &source)); + } + Ok(()) + } + + #[test] + fn test_unsigned_to_signed_cast_preserves_values_and_ordering() { + use arrow::array::{UInt8Array, UInt16Array, UInt32Array}; + use arrow::compute::SortOptions; + use datafusion_expr_common::sort_properties::SortProperties; + + let inputs: [(ArrayRef, Vec<DataType>, i64); 3] = [ + ( + Arc::new(UInt8Array::from(vec![ + None, + Some(0), + Some(1), + Some(u8::MAX), + ])), + vec![Int16, Int32, Int64], + i64::from(u8::MAX), + ), + ( + Arc::new(UInt16Array::from(vec![ + None, + Some(0), + Some(1), + Some(u16::MAX), + ])), + vec![Int32, Int64], + i64::from(u16::MAX), + ), + ( + Arc::new(UInt32Array::from(vec![ + None, + Some(0), + Some(1), + Some(u32::MAX), + ])), + vec![Int64], + i64::from(u32::MAX), + ), + ]; + for (input, target_types, max) in inputs { + let schema = Arc::new(Schema::new(vec![Field::new( + "a", + input.data_type().clone(), + true, + )])); + let batch = + RecordBatch::try_new(Arc::clone(&schema), vec![Arc::clone(&input)]) + .expect("valid input batch"); + for target_type in target_types { + let expr = + CastExpr::new(col("a", &schema).unwrap(), target_type.clone(), None); + let actual = expr + .evaluate(&batch) + .unwrap() + .into_array(batch.num_rows()) + .unwrap(); + for (index, value) in + [None, Some(0), Some(1), Some(max)].into_iter().enumerate() + { + let expected = match target_type { + Int16 => { + ScalarValue::Int16(value.map(|v| i16::try_from(v).unwrap())) + } + Int32 => { + ScalarValue::Int32(value.map(|v| i32::try_from(v).unwrap())) + } + Int64 => ScalarValue::Int64(value), + _ => unreachable!(), + }; + assert_eq!( + ScalarValue::try_from_array(&actual, index).unwrap(), + expected + ); + } + for descending in [false, true] { + for nulls_first in [false, true] { + let child = ExprProperties::new_unknown() + .with_range( + Interval::make_unbounded(input.data_type()).unwrap(), + ) + .with_order(SortProperties::Ordered(SortOptions { + descending, + nulls_first, + })) + .with_strictly_order_preserving(true); + let properties = + expr.get_properties(std::slice::from_ref(&child)).unwrap(); + assert_eq!(properties.sort_properties, child.sort_properties); + assert!(properties.strictly_order_preserving); + assert_eq!(properties.range.data_type(), target_type); + } + } + } + } + } + #[test] fn test_temporal_cast_ordering() { use TimeUnit::*; @@ -1913,10 +2084,15 @@ mod tests { assert!(!CastExpr::check_bigger_cast(&Float64, &Int64)); assert!(!CastExpr::check_bigger_cast(&Float64, &UInt64)); - // Signed <-> Unsigned conversions should return false (not order-preserving due to negative values) + // Signed-to-unsigned and unsigned-to-signed casts whose target cannot + // represent the entire source range are not lossless for all values. assert!(!CastExpr::check_bigger_cast(&UInt16, &Int8)); assert!(!CastExpr::check_bigger_cast(&UInt32, &Int16)); - assert!(!CastExpr::check_bigger_cast(&Int16, &UInt8)); + assert!(!CastExpr::check_bigger_cast(&Int8, &UInt8)); + assert!(!CastExpr::check_bigger_cast(&Int16, &UInt16)); + assert!(!CastExpr::check_bigger_cast(&Int32, &UInt32)); + assert!(!CastExpr::check_bigger_cast(&Int64, &UInt64)); + assert!(!CastExpr::check_bigger_cast(&Int8, &UInt16)); } } diff --git a/datafusion/physical-expr/src/projection.rs b/datafusion/physical-expr/src/projection.rs index b7196e3841..7fe6789699 100644 --- a/datafusion/physical-expr/src/projection.rs +++ b/datafusion/physical-expr/src/projection.rs @@ -2493,6 +2493,81 @@ pub(crate) mod tests { } } + #[test] + fn test_project_statistics_lossless_cast() { + use Precision::{Absent, Exact, Inexact}; + + for (lower, upper, targets) in [ + ( + ScalarValue::UInt8(Some(0)), + ScalarValue::UInt8(Some(u8::MAX)), + vec![DataType::Int16, DataType::Int32, DataType::Int64], + ), + ( + ScalarValue::UInt16(Some(0)), + ScalarValue::UInt16(Some(u16::MAX)), + vec![DataType::Int32, DataType::Int64], + ), + ( + ScalarValue::UInt32(Some(0)), + ScalarValue::UInt32(Some(u32::MAX)), + vec![DataType::Int64], + ), + ( + ScalarValue::Utf8(Some(String::new())), + ScalarValue::Utf8(Some("🦀".to_string())), + vec![DataType::Utf8View], + ), + ( + ScalarValue::Binary(Some(vec![])), + ScalarValue::Binary(Some(vec![0xff])), + vec![DataType::LargeBinary, DataType::BinaryView], + ), + ] { + for target in targets { + // A globally safe cast does not need two exact extrema. Preserve + // whichever bounds are available without upgrading their precision. + for (min_value, max_value) in [ + (Exact(lower.clone()), Exact(upper.clone())), + (Exact(lower.clone()), Absent), + (Absent, Exact(upper.clone())), + (Inexact(lower.clone()), Inexact(upper.clone())), + (Inexact(lower.clone()), Absent), + (Absent, Inexact(upper.clone())), + ] { + let input = ColumnStatistics { + min_value: min_value.clone(), + max_value: max_value.clone(), + null_count: Exact(1), + distinct_count: Inexact(3), + sum_value: Absent, + byte_size: Absent, + }; + let expr = CastExpr::new( + Arc::new(Column::new("a", 0)), + target.clone(), + None, + ); + let output = + project_column_statistics_through_expr(&expr, &[input], None); + assert_eq!( + output, + ColumnStatistics { + min_value: min_value.cast_to(&target).unwrap(), + max_value: max_value.cast_to(&target).unwrap(), + null_count: Exact(1), + distinct_count: Inexact(3), + sum_value: Absent, + byte_size: Absent, + }, + "{} -> {target}, {min_value:?}..{max_value:?}", + lower.data_type() + ); + } + } + } + } + #[test] fn test_project_statistics_narrowing_cast_requires_safe_bounds() { let schema = Schema::new(vec![Field::new("a", DataType::Int32, true)]); --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
