Skip to content

feat: recognize more lossless casts for statistics and ordering #25500

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
haohuaijin wants to merge 3 commits into
base: main
Choose a base branch
from
+258 −7
Open

feat: recognize more lossless casts for statistics and ordering #25500

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
664bdb4
feat: recognize more lossless casts for statistics and ordering
haohuaijin Sep 19, 2026
28ede8a
fix: preserve existing CastExpr public API names
haohuaijin Sep 19, 2026
438d1a7
Merge upstream main and resolve cast conflicts
haohuaijin Sep 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
190 changes: 183 additions & 7 deletions datafusion/physical-expr/src/expressions/cast.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -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)
)
}

Expand Down Expand Up @@ -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::*;
Expand Down Expand Up @@ -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));
}
}

Expand Down
75 changes: 75 additions & 0 deletions datafusion/physical-expr/src/projection.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)]);
Expand Down
Loading