From 664bdb41461ad330540c779672631e87fa94c3b6 Mon Sep 17 00:00:00 2001 From: Huaijin Date: Sat, 19 Sep 2026 20:14:43 +0800 Subject: [PATCH 1/2] feat: recognize more lossless casts for statistics and ordering --- .../custom_data_source/custom_file_casts.rs | 2 +- .../physical-expr/src/expressions/cast.rs | 238 +++++++++++++++--- datafusion/physical-expr/src/projection.rs | 76 +++++- 3 files changed, 283 insertions(+), 33 deletions(-) diff --git a/datafusion-examples/examples/custom_data_source/custom_file_casts.rs b/datafusion-examples/examples/custom_data_source/custom_file_casts.rs index 202c0a71257e9..ef15d524a0b46 100644 --- a/datafusion-examples/examples/custom_data_source/custom_file_casts.rs +++ b/datafusion-examples/examples/custom_data_source/custom_file_casts.rs @@ -189,7 +189,7 @@ impl PhysicalExprAdapter for CustomCastsPhysicalExprAdapter { let input_data_type = cast.expr().data_type(&self.physical_file_schema)?; let output_field = cast.target_field(); - if !cast.is_bigger_cast(&input_data_type) { + if !cast.is_lossless_cast(&input_data_type) { return not_impl_err!( "Unsupported CAST from {input_data_type} to {}", output_field.data_type() diff --git a/datafusion/physical-expr/src/expressions/cast.rs b/datafusion/physical-expr/src/expressions/cast.rs index 9a2e4ab648126..c6616f66e676d 100644 --- a/datafusion/physical-expr/src/expressions/cast.rs +++ b/datafusion/physical-expr/src/expressions/cast.rs @@ -250,11 +250,11 @@ 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. - pub fn check_bigger_cast(cast_type: &DataType, src: &DataType) -> bool { + pub fn check_lossless_cast(cast_type: &DataType, src: &DataType) -> bool { if cast_type.eq(src) { return true; } @@ -267,19 +267,20 @@ 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) ) } /// Check if the cast is lossless and strictly order-preserving for all source - /// values, preserving nulls. See [`Self::check_bigger_cast`]. - pub fn is_bigger_cast(&self, src: &DataType) -> bool { - Self::check_bigger_cast(self.cast_type(), src) + /// values, preserving nulls. See [`Self::check_lossless_cast`]. + pub fn is_lossless_cast(&self, src: &DataType) -> bool { + Self::check_lossless_cast(self.cast_type(), src) } } @@ -298,16 +299,16 @@ pub(crate) fn cast_expr_properties( ) -> Result { let unbounded = Interval::make_unbounded(target_type)?; let source_type = child.range.data_type(); - // A lossless cast recognized by check_bigger_cast is one-to-one, so it is + // A lossless cast recognized by check_lossless_cast is one-to-one, so it is // strictly order-preserving; a narrowing cast may collapse distinct values, // breaking the ordering of subsequent sort keys. - let bigger_cast = CastExpr::check_bigger_cast(target_type, &source_type); - if is_order_preserving_cast_family(&source_type, target_type) || bigger_cast { + let lossless_cast = CastExpr::check_lossless_cast(target_type, &source_type); + if is_order_preserving_cast_family(&source_type, target_type) || lossless_cast { Ok(child .clone() .with_range(unbounded) .with_strictly_order_preserving( - child.strictly_order_preserving && bigger_cast, + child.strictly_order_preserving && lossless_cast, )) } else { Ok(ExprProperties::new_unknown().with_range(unbounded)) @@ -1542,7 +1543,7 @@ mod tests { expected.data_type().clone(), None, ); - assert!(expr.is_bigger_cast(input.data_type())); + assert!(expr.is_lossless_cast(input.data_type())); let child = ExprProperties::new_unknown() .with_range( Interval::make_unbounded(input.data_type()) @@ -1569,28 +1570,203 @@ mod tests { } #[test] - fn test_check_bigger_cast_precision_loss() { + 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> = 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_lossless_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_lossless_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, 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_check_lossless_cast_precision_loss() { use DataType::*; // Exact conversions without precision loss - assert!(CastExpr::check_bigger_cast(&Int16, &Int8)); - assert!(CastExpr::check_bigger_cast(&Int64, &Int32)); - assert!(CastExpr::check_bigger_cast(&Float32, &Int16)); - assert!(CastExpr::check_bigger_cast(&Float32, &UInt16)); - assert!(CastExpr::check_bigger_cast(&Float64, &Int32)); - assert!(CastExpr::check_bigger_cast(&Float64, &UInt32)); - assert!(CastExpr::check_bigger_cast(&LargeUtf8, &Utf8)); + assert!(CastExpr::check_lossless_cast(&Int16, &Int8)); + assert!(CastExpr::check_lossless_cast(&Int64, &Int32)); + assert!(CastExpr::check_lossless_cast(&Float32, &Int16)); + assert!(CastExpr::check_lossless_cast(&Float32, &UInt16)); + assert!(CastExpr::check_lossless_cast(&Float64, &Int32)); + assert!(CastExpr::check_lossless_cast(&Float64, &UInt32)); + assert!(CastExpr::check_lossless_cast(&LargeUtf8, &Utf8)); // Precision-losing int-to-float conversions should return false - assert!(!CastExpr::check_bigger_cast(&Float32, &Int32)); - assert!(!CastExpr::check_bigger_cast(&Float32, &UInt32)); - 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) - assert!(!CastExpr::check_bigger_cast(&UInt16, &Int8)); - assert!(!CastExpr::check_bigger_cast(&UInt32, &Int16)); - assert!(!CastExpr::check_bigger_cast(&Int16, &UInt8)); + assert!(!CastExpr::check_lossless_cast(&Float32, &Int32)); + assert!(!CastExpr::check_lossless_cast(&Float32, &UInt32)); + assert!(!CastExpr::check_lossless_cast(&Float64, &Int64)); + assert!(!CastExpr::check_lossless_cast(&Float64, &UInt64)); + + // 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_lossless_cast(&UInt16, &Int8)); + assert!(!CastExpr::check_lossless_cast(&UInt32, &Int16)); + assert!(!CastExpr::check_lossless_cast(&Int8, &UInt8)); + assert!(!CastExpr::check_lossless_cast(&Int16, &UInt16)); + assert!(!CastExpr::check_lossless_cast(&Int32, &UInt32)); + assert!(!CastExpr::check_lossless_cast(&Int64, &UInt64)); + assert!(!CastExpr::check_lossless_cast(&Int8, &UInt16)); } } diff --git a/datafusion/physical-expr/src/projection.rs b/datafusion/physical-expr/src/projection.rs index 6a038b429b26c..169301f7497bd 100644 --- a/datafusion/physical-expr/src/projection.rs +++ b/datafusion/physical-expr/src/projection.rs @@ -890,7 +890,7 @@ fn project_column_statistics_through_expr( // domain, for example, does not bound the converted column. Merely casting // a failing endpoint to NULL also cannot establish the remaining extrema. let preserves_values = source_type.is_some_and(|source_type| { - CastExpr::check_bigger_cast(target_type, &source_type) + CastExpr::check_lossless_cast(target_type, &source_type) || is_within_extrema( &inner_stats.min_value, &inner_stats.max_value, @@ -2431,6 +2431,80 @@ 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]); + 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)]); From 28ede8a092a2b46125fa29d8c96c81ee7fe6a8ec Mon Sep 17 00:00:00 2001 From: Huaijin Date: Sat, 19 Sep 2026 20:31:54 +0800 Subject: [PATCH 2/2] fix: preserve existing CastExpr public API names --- .../custom_data_source/custom_file_casts.rs | 2 +- .../physical-expr/src/expressions/cast.rs | 56 +++++++++---------- datafusion/physical-expr/src/projection.rs | 2 +- 3 files changed, 30 insertions(+), 30 deletions(-) diff --git a/datafusion-examples/examples/custom_data_source/custom_file_casts.rs b/datafusion-examples/examples/custom_data_source/custom_file_casts.rs index ef15d524a0b46..202c0a71257e9 100644 --- a/datafusion-examples/examples/custom_data_source/custom_file_casts.rs +++ b/datafusion-examples/examples/custom_data_source/custom_file_casts.rs @@ -189,7 +189,7 @@ impl PhysicalExprAdapter for CustomCastsPhysicalExprAdapter { let input_data_type = cast.expr().data_type(&self.physical_file_schema)?; let output_field = cast.target_field(); - if !cast.is_lossless_cast(&input_data_type) { + if !cast.is_bigger_cast(&input_data_type) { return not_impl_err!( "Unsupported CAST from {input_data_type} to {}", output_field.data_type() diff --git a/datafusion/physical-expr/src/expressions/cast.rs b/datafusion/physical-expr/src/expressions/cast.rs index c6616f66e676d..be5e8449c9465 100644 --- a/datafusion/physical-expr/src/expressions/cast.rs +++ b/datafusion/physical-expr/src/expressions/cast.rs @@ -254,7 +254,7 @@ impl CastExpr { /// 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. - pub fn check_lossless_cast(cast_type: &DataType, src: &DataType) -> bool { + pub fn check_bigger_cast(cast_type: &DataType, src: &DataType) -> bool { if cast_type.eq(src) { return true; } @@ -278,9 +278,9 @@ impl CastExpr { } /// Check if the cast is lossless and strictly order-preserving for all source - /// values, preserving nulls. See [`Self::check_lossless_cast`]. - pub fn is_lossless_cast(&self, src: &DataType) -> bool { - Self::check_lossless_cast(self.cast_type(), src) + /// values, preserving nulls. See [`Self::check_bigger_cast`]. + pub fn is_bigger_cast(&self, src: &DataType) -> bool { + Self::check_bigger_cast(self.cast_type(), src) } } @@ -299,10 +299,10 @@ pub(crate) fn cast_expr_properties( ) -> Result { let unbounded = Interval::make_unbounded(target_type)?; let source_type = child.range.data_type(); - // A lossless cast recognized by check_lossless_cast is one-to-one, so it is + // A lossless cast recognized by check_bigger_cast is one-to-one, so it is // strictly order-preserving; a narrowing cast may collapse distinct values, // breaking the ordering of subsequent sort keys. - let lossless_cast = CastExpr::check_lossless_cast(target_type, &source_type); + let lossless_cast = CastExpr::check_bigger_cast(target_type, &source_type); if is_order_preserving_cast_family(&source_type, target_type) || lossless_cast { Ok(child .clone() @@ -1543,7 +1543,7 @@ mod tests { expected.data_type().clone(), None, ); - assert!(expr.is_lossless_cast(input.data_type())); + assert!(expr.is_bigger_cast(input.data_type())); let child = ExprProperties::new_unknown() .with_range( Interval::make_unbounded(input.data_type()) @@ -1614,7 +1614,7 @@ mod tests { )])); let expr = CastExpr::new(col("a", &schema)?, expected.data_type().clone(), None); - assert!(expr.is_lossless_cast(input.data_type())); + assert!(expr.is_bigger_cast(input.data_type())); for descending in [false, true] { for nulls_first in [false, true] { let child = ExprProperties::new_unknown() @@ -1639,7 +1639,7 @@ mod tests { (LargeBinary, Binary), (BinaryView, Binary), ] { - assert!(!CastExpr::check_lossless_cast(&target, &source)); + assert!(!CastExpr::check_bigger_cast(&target, &source)); } Ok(()) } @@ -1740,33 +1740,33 @@ mod tests { } #[test] - fn test_check_lossless_cast_precision_loss() { + fn test_check_bigger_cast_precision_loss() { use DataType::*; // Exact conversions without precision loss - assert!(CastExpr::check_lossless_cast(&Int16, &Int8)); - assert!(CastExpr::check_lossless_cast(&Int64, &Int32)); - assert!(CastExpr::check_lossless_cast(&Float32, &Int16)); - assert!(CastExpr::check_lossless_cast(&Float32, &UInt16)); - assert!(CastExpr::check_lossless_cast(&Float64, &Int32)); - assert!(CastExpr::check_lossless_cast(&Float64, &UInt32)); - assert!(CastExpr::check_lossless_cast(&LargeUtf8, &Utf8)); + assert!(CastExpr::check_bigger_cast(&Int16, &Int8)); + assert!(CastExpr::check_bigger_cast(&Int64, &Int32)); + assert!(CastExpr::check_bigger_cast(&Float32, &Int16)); + assert!(CastExpr::check_bigger_cast(&Float32, &UInt16)); + assert!(CastExpr::check_bigger_cast(&Float64, &Int32)); + assert!(CastExpr::check_bigger_cast(&Float64, &UInt32)); + assert!(CastExpr::check_bigger_cast(&LargeUtf8, &Utf8)); // Precision-losing int-to-float conversions should return false - assert!(!CastExpr::check_lossless_cast(&Float32, &Int32)); - assert!(!CastExpr::check_lossless_cast(&Float32, &UInt32)); - assert!(!CastExpr::check_lossless_cast(&Float64, &Int64)); - assert!(!CastExpr::check_lossless_cast(&Float64, &UInt64)); + assert!(!CastExpr::check_bigger_cast(&Float32, &Int32)); + assert!(!CastExpr::check_bigger_cast(&Float32, &UInt32)); + assert!(!CastExpr::check_bigger_cast(&Float64, &Int64)); + assert!(!CastExpr::check_bigger_cast(&Float64, &UInt64)); // 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_lossless_cast(&UInt16, &Int8)); - assert!(!CastExpr::check_lossless_cast(&UInt32, &Int16)); - assert!(!CastExpr::check_lossless_cast(&Int8, &UInt8)); - assert!(!CastExpr::check_lossless_cast(&Int16, &UInt16)); - assert!(!CastExpr::check_lossless_cast(&Int32, &UInt32)); - assert!(!CastExpr::check_lossless_cast(&Int64, &UInt64)); - assert!(!CastExpr::check_lossless_cast(&Int8, &UInt16)); + assert!(!CastExpr::check_bigger_cast(&UInt16, &Int8)); + assert!(!CastExpr::check_bigger_cast(&UInt32, &Int16)); + 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 169301f7497bd..bda106fbe7955 100644 --- a/datafusion/physical-expr/src/projection.rs +++ b/datafusion/physical-expr/src/projection.rs @@ -890,7 +890,7 @@ fn project_column_statistics_through_expr( // domain, for example, does not bound the converted column. Merely casting // a failing endpoint to NULL also cannot establish the remaining extrema. let preserves_values = source_type.is_some_and(|source_type| { - CastExpr::check_lossless_cast(target_type, &source_type) + CastExpr::check_bigger_cast(target_type, &source_type) || is_within_extrema( &inner_stats.min_value, &inner_stats.max_value,