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-23910-31ffab1418aae8966dd270b1042c386bc9646950 in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit e7304eec73607be8d82846fb279d64c301371238 Author: Burak Şen <[email protected]> AuthorDate: Wed Aug 5 09:57:00 2026 +0300 fix: do not derive ordering for arithmetic that can overflow (#23910) ## Which issue does this PR close? <!-- We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123. --> - Closes #23902. ## Rationale for this change Please check the issue for details but the main idea is that `a + b` over two sorted columns is not guaranteed to be sorted. ## What changes are included in this PR? Ordering is discarded when it overflows or wraps ## Are these changes tested? Yes, adjusted existing tests and added a regression test in `order.slt`. ## Are there any user-facing changes? no api changes --- .../core/tests/fuzz_cases/equivalence/ordering.rs | 33 ++- .../tests/fuzz_cases/equivalence/projection.rs | 32 ++- .../core/tests/fuzz_cases/equivalence/utils.rs | 19 +- .../physical-expr/src/equivalence/ordering.rs | 8 +- .../src/equivalence/properties/dependency.rs | 8 +- datafusion/physical-expr/src/expressions/binary.rs | 274 ++++++++++++++++++++- datafusion/physical-expr/src/projection.rs | 83 ++----- datafusion/sqllogictest/test_files/order.slt | 43 +++- datafusion/sqllogictest/test_files/topk.slt | 5 +- 9 files changed, 390 insertions(+), 115 deletions(-) diff --git a/datafusion/core/tests/fuzz_cases/equivalence/ordering.rs b/datafusion/core/tests/fuzz_cases/equivalence/ordering.rs index a57095066e..60b0997635 100644 --- a/datafusion/core/tests/fuzz_cases/equivalence/ordering.rs +++ b/datafusion/core/tests/fuzz_cases/equivalence/ordering.rs @@ -16,9 +16,9 @@ // under the License. use crate::fuzz_cases::equivalence::utils::{ - TestScalarUDF, create_random_schema, create_test_params, create_test_schema_2, - generate_table_for_eq_properties, generate_table_for_orderings, - is_table_same_after_sort, + TestScalarUDF, contains_overflowable_arithmetic, create_random_schema, + create_test_params, create_test_schema_2, generate_table_for_eq_properties, + generate_table_for_orderings, is_table_same_after_sort, }; use arrow::compute::SortOptions; use datafusion_common::Result; @@ -144,14 +144,27 @@ fn test_ordering_satisfy_with_equivalence_complex_random() -> Result<()> { let err_msg = format!( "Error in test case requirement:{ordering:?}, expected: {expected:?}, eq_properties: {eq_properties}", ); - // Check whether ordering_satisfy API result and - // experimental result matches. - - assert_eq!( - eq_properties.ordering_satisfy(ordering)?, - (expected | false), - "{err_msg}" + // A rejection turns inconclusive only from the first `+`/`-` + // key onwards, since possible overflow makes an ordering + // underivable even when the sample happens to be sorted. A + // table sorted by the full ordering is sorted by every prefix + // of it, so a rejected arithmetic-free prefix still proves + // the rejection is genuine. + let conclusive_prefix = LexOrdering::new( + ordering + .iter() + .take_while(|sort_expr| { + !contains_overflowable_arithmetic(&sort_expr.expr) + }) + .cloned(), ); + if eq_properties.ordering_satisfy(ordering)? { + assert!(expected, "{err_msg}"); + } else if let Some(prefix) = conclusive_prefix + && !eq_properties.ordering_satisfy(prefix)? + { + assert!(!expected, "{err_msg}"); + } } } } diff --git a/datafusion/core/tests/fuzz_cases/equivalence/projection.rs b/datafusion/core/tests/fuzz_cases/equivalence/projection.rs index 2f67e211ce..9593e1cf11 100644 --- a/datafusion/core/tests/fuzz_cases/equivalence/projection.rs +++ b/datafusion/core/tests/fuzz_cases/equivalence/projection.rs @@ -16,8 +16,8 @@ // under the License. use crate::fuzz_cases::equivalence::utils::{ - TestScalarUDF, apply_projection, create_random_schema, - generate_table_for_eq_properties, is_table_same_after_sort, + TestScalarUDF, apply_projection, contains_overflowable_arithmetic, + create_random_schema, generate_table_for_eq_properties, is_table_same_after_sort, }; use arrow::compute::SortOptions; use datafusion_common::Result; @@ -179,13 +179,29 @@ fn ordering_satisfy_after_projection_random() -> Result<()> { let err_msg = format!( "Error in test case requirement:{ordering:?}, expected: {expected:?}, eq_properties: {eq_properties}, projected_eq: {projected_eq}, projection_mapping: {projection_mapping:?}" ); - // Check whether ordering_satisfy API result and - // experimental result matches. - assert_eq!( - projected_eq.ordering_satisfy(ordering)?, - expected, - "{err_msg}" + // Same reasoning as in `ordering.rs`: only keys from + // the first `+`/`-` source onwards are inconclusive, + // so assert on the longest prefix without one. + let conclusive_prefix = LexOrdering::new( + ordering + .iter() + .take_while(|sort_expr| { + !projection_mapping.iter().any(|(source, targets)| { + targets + .iter() + .any(|(target, _)| target.eq(&sort_expr.expr)) + && contains_overflowable_arithmetic(source) + }) + }) + .cloned(), ); + if projected_eq.ordering_satisfy(ordering)? { + assert!(expected, "{err_msg}"); + } else if let Some(prefix) = conclusive_prefix + && !projected_eq.ordering_satisfy(prefix)? + { + assert!(!expected, "{err_msg}"); + } } } } diff --git a/datafusion/core/tests/fuzz_cases/equivalence/utils.rs b/datafusion/core/tests/fuzz_cases/equivalence/utils.rs index 8350cafb21..ca73db3ae9 100644 --- a/datafusion/core/tests/fuzz_cases/equivalence/utils.rs +++ b/datafusion/core/tests/fuzz_cases/equivalence/utils.rs @@ -21,11 +21,12 @@ use std::sync::Arc; use arrow::array::{ArrayRef, Float32Array, Float64Array, RecordBatch, UInt32Array}; use arrow::compute::{SortColumn, SortOptions, lexsort_to_indices, take_record_batch}; use arrow::datatypes::{DataType, Field, Schema, SchemaRef}; +use datafusion_common::tree_node::TreeNode; use datafusion_common::utils::{compare_rows, get_row_at_idx}; use datafusion_common::{Result, exec_err, internal_datafusion_err, plan_err}; use datafusion_expr::sort_properties::{ExprProperties, SortProperties}; use datafusion_expr::{ - ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility, + ColumnarValue, Operator, ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility, }; use datafusion_physical_expr::equivalence::{ EquivalenceClass, ProjectionMapping, convert_to_orderings, @@ -33,7 +34,7 @@ use datafusion_physical_expr::equivalence::{ use datafusion_physical_expr::{ConstExpr, EquivalenceProperties}; use datafusion_physical_expr_common::physical_expr::PhysicalExpr; use datafusion_physical_expr_common::sort_expr::{LexOrdering, PhysicalSortExpr}; -use datafusion_physical_plan::expressions::{Column, col}; +use datafusion_physical_plan::expressions::{BinaryExpr, Column, col}; use itertools::izip; use rand::prelude::*; @@ -209,6 +210,20 @@ fn add_equal_conditions_test() -> Result<()> { Ok(()) } +/// Returns `true` if `expr` contains a `+` or `-` anywhere in its tree. +/// +/// The equivalence framework conservatively discards orderings derived from +/// `+`/`-` expressions, because wrapping overflow can break them over the +/// type's full domain even when a finite batch happens to remain sorted. +pub fn contains_overflowable_arithmetic(expr: &Arc<dyn PhysicalExpr>) -> bool { + expr.exists(|e| { + Ok(e.downcast_ref::<BinaryExpr>().is_some_and(|binary| { + matches!(binary.op(), Operator::Plus | Operator::Minus) + })) + }) + .unwrap() +} + /// Checks if the table (RecordBatch) remains unchanged when sorted according to the provided `required_ordering`. /// /// The function works by adding a unique column of ascending integers to the original table. This column ensures diff --git a/datafusion/physical-expr/src/equivalence/ordering.rs b/datafusion/physical-expr/src/equivalence/ordering.rs index 15637d24e8..499187a603 100644 --- a/datafusion/physical-expr/src/equivalence/ordering.rs +++ b/datafusion/physical-expr/src/equivalence/ordering.rs @@ -525,8 +525,8 @@ mod tests { vec![col_e], // requirement [a ASC, c ASC, a+b ASC], vec![(col_a, options), (col_c, options), (&a_plus_b, options)], - // expected: requirement is satisfied. - true, + // expected: requirement is not satisfied because addition can wrap. + false, ), // ------------ TEST CASE 4 ------------ ( @@ -672,8 +672,8 @@ mod tests { vec![col_e], // requirement [c ASC, d ASC, a + b ASC], vec![(col_c, options), (col_d, options), (&a_plus_b, options)], - // expected: requirement is satisfied. - true, + // expected: requirement is not satisfied because addition can wrap. + false, ), ]; diff --git a/datafusion/physical-expr/src/equivalence/properties/dependency.rs b/datafusion/physical-expr/src/equivalence/properties/dependency.rs index d2a8c2f654..bd8bef84de 100644 --- a/datafusion/physical-expr/src/equivalence/properties/dependency.rs +++ b/datafusion/physical-expr/src/equivalence/properties/dependency.rs @@ -632,10 +632,10 @@ mod tests { ]); let test_cases = vec![ - // d + b + // d + b can wrap ( Arc::new(BinaryExpr::new(col_d, Operator::Plus, Arc::clone(&col_b))) as _, - SortProperties::Ordered(option_asc), + SortProperties::Unordered, ), // b (col_b, SortProperties::Ordered(option_asc)), @@ -717,8 +717,8 @@ mod tests { (vec![col_b], vec![]), // TEST CASE 5 (vec![col_d], vec![(col_d, option_asc)]), - // TEST CASE 5 - (vec![&a_plus_d], vec![(&a_plus_d, option_asc)]), + // TEST CASE 5: a + d is not ordered because addition can wrap. + (vec![&a_plus_d], vec![]), // TEST CASE 6 ( vec![col_b, col_d], diff --git a/datafusion/physical-expr/src/expressions/binary.rs b/datafusion/physical-expr/src/expressions/binary.rs index 170c1a4d02..1bd49696bb 100644 --- a/datafusion/physical-expr/src/expressions/binary.rs +++ b/datafusion/physical-expr/src/expressions/binary.rs @@ -19,6 +19,7 @@ mod kernels; use crate::PhysicalExpr; use crate::intervals::cp_solver::{propagate_arithmetic, propagate_comparison}; +use std::cmp::Ordering; use std::hash::Hash; use std::sync::Arc; @@ -33,7 +34,7 @@ use datafusion_common::{Result, ScalarValue, internal_err, not_impl_err}; use datafusion_expr::binary::BinaryTypeCoercer; use datafusion_expr::interval_arithmetic::{Interval, apply_operator}; -use datafusion_expr::sort_properties::ExprProperties; +use datafusion_expr::sort_properties::{ExprProperties, SortProperties}; #[expect(deprecated)] use datafusion_expr::statistics::Distribution::{Bernoulli, Gaussian}; #[expect(deprecated)] @@ -118,6 +119,68 @@ impl BinaryExpr { pub fn op(&self) -> &Operator { &self.op } + + /// Wrapping on overflow breaks monotonicity (e.g. the sum of two + /// ascending `UInt8` columns can wrap back to small values), so the + /// derived ordering is kept only when overflow is impossible. `time ± + /// interval` wraps around the 24-hour clock even in checked mode, so it + /// never preserves ordering. + fn arithmetic_sort_properties( + &self, + sort_properties: SortProperties, + l_range: &Interval, + r_range: &Interval, + range: &Interval, + ) -> SortProperties { + if sort_properties == SortProperties::Singleton { + return sort_properties; + } + let wraps_in_domain = match self.op { + Operator::Plus => { + is_time_plus_interval(&l_range.data_type(), &r_range.data_type()) + } + Operator::Minus => { + is_time_minus_interval(&l_range.data_type(), &r_range.data_type()) + } + _ => false, + }; + let cannot_overflow = !range.is_unbounded() + && !unsigned_subtraction_may_underflow(self.op, l_range, r_range, range); + if !wraps_in_domain && (self.fail_on_overflow || cannot_overflow) { + sort_properties + } else { + SortProperties::Unordered + } + } +} + +/// Returns `true` unless `l_range - r_range` provably stays within an unsigned +/// domain. +/// +/// [`Interval`] standardizes an underflowed (i.e. `null`) lower bound of an +/// unsigned type back to zero, so an apparently bounded result range is not +/// enough to rule out wrapping here -- e.g. `[0, 10] - [0, 10]` over `UInt32` +/// yields `[0, 10]` even though `0 - 10` wraps to `u32::MAX`. Compare the +/// endpoints that produce the smallest difference instead. +fn unsigned_subtraction_may_underflow( + op: Operator, + l_range: &Interval, + r_range: &Interval, + range: &Interval, +) -> bool { + if op != Operator::Minus || !range.data_type().is_unsigned_integer() { + return false; + } + let (smallest_lhs, largest_rhs) = (l_range.lower(), r_range.upper()); + if smallest_lhs.is_null() || largest_rhs.is_null() { + return true; + } + // Operands of differing types compare as incomparable, in which case we + // conservatively assume an underflow is possible. + !matches!( + smallest_lhs.partial_cmp(largest_rhs), + Some(Ordering::Greater | Ordering::Equal) + ) } impl std::fmt::Display for BinaryExpr { @@ -760,18 +823,34 @@ impl PhysicalExpr for BinaryExpr { let (l_order, l_range) = (children[0].sort_properties, &children[0].range); let (r_order, r_range) = (children[1].sort_properties, &children[1].range); match self.op() { - Operator::Plus => Ok(ExprProperties { - sort_properties: l_order.add(&r_order), - range: l_range.add(r_range)?, - preserves_lex_ordering: false, - strictly_order_preserving: false, - }), - Operator::Minus => Ok(ExprProperties { - sort_properties: l_order.sub(&r_order), - range: l_range.sub(r_range)?, - preserves_lex_ordering: false, - strictly_order_preserving: false, - }), + Operator::Plus => { + let range = l_range.add(r_range)?; + Ok(ExprProperties { + sort_properties: self.arithmetic_sort_properties( + l_order.add(&r_order), + l_range, + r_range, + &range, + ), + range, + preserves_lex_ordering: false, + strictly_order_preserving: false, + }) + } + Operator::Minus => { + let range = l_range.sub(r_range)?; + Ok(ExprProperties { + sort_properties: self.arithmetic_sort_properties( + l_order.sub(&r_order), + l_range, + r_range, + &range, + ), + range, + preserves_lex_ordering: false, + strictly_order_preserving: false, + }) + } Operator::Gt => Ok(ExprProperties { sort_properties: l_order.gt_or_gteq(&r_order), range: l_range.gt(r_range)?, @@ -1306,7 +1385,176 @@ mod tests { use crate::planner::logical2physical; use arrow::array::BooleanArray; + use arrow::compute::SortOptions; use datafusion_expr::col as logical_col; + + #[test] + fn test_arithmetic_ordering_overflow() -> Result<()> { + let asc = SortProperties::Ordered(Default::default()); + let ordered = |range: Interval| ExprProperties { + sort_properties: asc, + range, + preserves_lex_ordering: false, + strictly_order_preserving: false, + }; + + let schema = Schema::new(vec![ + Field::new("a", DataType::Int32, false), + Field::new("b", DataType::Int32, false), + ]); + let a_plus_b = + BinaryExpr::new(col("a", &schema)?, Operator::Plus, col("b", &schema)?); + let unbounded = [ + ordered(Interval::make_unbounded(&DataType::Int32)?), + ordered(Interval::make_unbounded(&DataType::Int32)?), + ]; + let bounded = [ + ordered(Interval::make(Some(0), Some(10))?), + ordered(Interval::make(Some(0), Some(10))?), + ]; + + // Unknown ranges: the sum may overflow and wrap, so it is unordered. + assert_eq!( + a_plus_b.get_properties(&unbounded)?.sort_properties, + SortProperties::Unordered + ); + // Bounded ranges that cannot overflow keep the ordering, as does + // checked arithmetic, which errors instead of wrapping. + assert_eq!(a_plus_b.get_properties(&bounded)?.sort_properties, asc); + let checked = a_plus_b.with_fail_on_overflow(true); + assert_eq!(checked.get_properties(&unbounded)?.sort_properties, asc); + + // `time + interval` wraps around the 24-hour clock even in checked + // mode, so it never preserves ordering. + let time = DataType::Time64(TimeUnit::Nanosecond); + let interval = DataType::Interval(IntervalUnit::MonthDayNano); + let schema = Schema::new(vec![ + Field::new("t", time.clone(), false), + Field::new("i", interval.clone(), false), + ]); + let time_plus_interval = + BinaryExpr::new(col("t", &schema)?, Operator::Plus, col("i", &schema)?) + .with_fail_on_overflow(true); + let time_props = [ + ordered(Interval::make_unbounded(&time)?), + ordered(Interval::make_unbounded(&interval)?), + ]; + assert_eq!( + time_plus_interval + .get_properties(&time_props)? + .sort_properties, + SortProperties::Unordered + ); + + Ok(()) + } + + /// `a - b` only derives an ordering when `a` and `b` are ordered in + /// opposite directions, so every case below pairs an ascending left-hand + /// side with a descending right-hand side. + #[test] + fn test_subtraction_ordering_overflow() -> Result<()> { + let asc = SortProperties::Ordered(SortOptions { + descending: false, + nulls_first: true, + }); + let desc = SortProperties::Ordered(SortOptions { + descending: true, + nulls_first: true, + }); + let props = |sort_properties, range| ExprProperties { + sort_properties, + range, + preserves_lex_ordering: false, + strictly_order_preserving: false, + }; + + let schema = Schema::new(vec![ + Field::new("a", DataType::Int32, false), + Field::new("b", DataType::Int32, false), + ]); + let a_minus_b = + BinaryExpr::new(col("a", &schema)?, Operator::Minus, col("b", &schema)?); + + // Signed minimum: the difference can underflow past `i32::MIN` and + // wrap around to large positive values. + let signed_underflow = [ + props(asc, Interval::make(Some(i32::MIN), Some(0))?), + props(desc, Interval::make(Some(0), Some(i32::MAX))?), + ]; + assert_eq!( + a_minus_b.get_properties(&signed_underflow)?.sort_properties, + SortProperties::Unordered + ); + // The very same ranges keep the ordering under checked arithmetic, + // which errors instead of wrapping. + let checked = a_minus_b.clone().with_fail_on_overflow(true); + assert_eq!( + checked.get_properties(&signed_underflow)?.sort_properties, + asc + ); + // Ranges whose difference stays inside `Int32` are safe. + let signed_safe = [ + props(asc, Interval::make(Some(0), Some(10))?), + props(desc, Interval::make(Some(0), Some(10))?), + ]; + assert_eq!(a_minus_b.get_properties(&signed_safe)?.sort_properties, asc); + + let schema = Schema::new(vec![ + Field::new("a", DataType::UInt32, false), + Field::new("b", DataType::UInt32, false), + ]); + let a_minus_b = + BinaryExpr::new(col("a", &schema)?, Operator::Minus, col("b", &schema)?); + + // Unsigned underflow: the ranges overlap, so `0 - 1` wraps to + // `u32::MAX` even though both operands are bounded. + let unsigned_underflow = [ + props(asc, Interval::make(Some(0_u32), Some(10_u32))?), + props(desc, Interval::make(Some(0_u32), Some(10_u32))?), + ]; + assert_eq!( + a_minus_b + .get_properties(&unsigned_underflow)? + .sort_properties, + SortProperties::Unordered + ); + // A left-hand range that always dominates the right-hand one cannot + // underflow. + let unsigned_safe = [ + props(asc, Interval::make(Some(10_u32), Some(20_u32))?), + props(desc, Interval::make(Some(0_u32), Some(5_u32))?), + ]; + assert_eq!( + a_minus_b.get_properties(&unsigned_safe)?.sort_properties, + asc + ); + + // `time - interval` wraps around the 24-hour clock even in checked + // mode, so it never preserves ordering. + let time = DataType::Time64(TimeUnit::Nanosecond); + let interval = DataType::Interval(IntervalUnit::MonthDayNano); + let schema = Schema::new(vec![ + Field::new("t", time.clone(), false), + Field::new("i", interval.clone(), false), + ]); + let time_minus_interval = + BinaryExpr::new(col("t", &schema)?, Operator::Minus, col("i", &schema)?) + .with_fail_on_overflow(true); + let time_props = [ + props(asc, Interval::make_unbounded(&time)?), + props(desc, Interval::make_unbounded(&interval)?), + ]; + assert_eq!( + time_minus_interval + .get_properties(&time_props)? + .sort_properties, + SortProperties::Unordered + ); + + Ok(()) + } + /// Performs a binary operation, applying any type coercion necessary fn binary_op( left: Arc<dyn PhysicalExpr>, diff --git a/datafusion/physical-expr/src/projection.rs b/datafusion/physical-expr/src/projection.rs index 1f6a6eb08f..bfa40c7838 100644 --- a/datafusion/physical-expr/src/projection.rs +++ b/datafusion/physical-expr/src/projection.rs @@ -1577,8 +1577,6 @@ pub(crate) mod tests { vec![("a_new", option_asc), ("b_new", option_asc)], // [a_new ASC, d_new ASC] vec![("a_new", option_asc), ("d_new", option_asc)], - // [a_new ASC, b+d ASC] - vec![("a_new", option_asc), ("b+d", option_asc)], ], ), // ------- TEST CASE 8 ---------- @@ -1660,12 +1658,6 @@ pub(crate) mod tests { ("b_new", option_asc), ("c_new", option_asc), ], - // [a_new ASC, b_new ASC, c+d ASC] - vec![ - ("a_new", option_asc), - ("b_new", option_asc), - ("c+d", option_asc), - ], ], ), // ------- TEST CASE 11 ---------- @@ -1687,8 +1679,6 @@ pub(crate) mod tests { vec![ // [a_new ASC, b_new ASC] vec![("a_new", option_asc), ("b_new", option_asc)], - // [a_new ASC, b + d ASC] - vec![("a_new", option_asc), ("b+d", option_asc)], ], ), // ------- TEST CASE 12 ---------- @@ -1770,30 +1760,12 @@ pub(crate) mod tests { ], // expected vec![ - // [a_new ASC, d_new ASC, b+e ASC] - vec![ - ("a_new", option_asc), - ("d_new", option_asc), - ("b+e", option_asc), - ], - // [d_new ASC, a_new ASC, b+e ASC] - vec![ - ("d_new", option_asc), - ("a_new", option_asc), - ("b+e", option_asc), - ], - // [c_new ASC, d_new ASC, b+e ASC] - vec![ - ("c_new", option_asc), - ("d_new", option_asc), - ("b+e", option_asc), - ], - // [d_new ASC, c_new ASC, b+e ASC] - vec![ - ("d_new", option_asc), - ("c_new", option_asc), - ("b+e", option_asc), - ], + // [a_new ASC] + vec![("a_new", option_asc)], + // [c_new ASC] + vec![("c_new", option_asc)], + // [d_new ASC] + vec![("d_new", option_asc)], ], ), // ------- TEST CASE 15 ---------- @@ -1815,12 +1787,8 @@ pub(crate) mod tests { ], // expected vec![ - // [a_new ASC, d_new ASC, b+e ASC] - vec![ - ("a_new", option_asc), - ("c_new", option_asc), - ("a+b", option_asc), - ], + // [a_new ASC, c_new ASC] + vec![("a_new", option_asc), ("c_new", option_asc)], ], ), // ------- TEST CASE 16 ---------- @@ -1845,8 +1813,6 @@ pub(crate) mod tests { vec![ // [a_new ASC, b_new ASC] vec![("a_new", option_asc), ("b_new", option_asc)], - // [a_new ASC, b_new ASC] - vec![("a_new", option_asc), ("b+e", option_asc)], // [c_new ASC, b_new DESC] vec![("c_new", option_asc), ("b_new", option_desc)], ], @@ -2119,7 +2085,6 @@ pub(crate) mod tests { let projection_mapping = ProjectionMapping::try_new(proj_exprs, &schema)?; let output_schema = output_schema(&projection_mapping, &schema)?; - let col_a_plus_b_new = &col("a+b", &output_schema)?; let col_c_new = &col("c_new", &output_schema)?; let col_d_new = &col("d_new", &output_schema)?; @@ -2137,18 +2102,10 @@ pub(crate) mod tests { vec![], // expected vec![ - // [d_new ASC, c_new ASC, a+b ASC] - vec![ - (col_d_new, option_asc), - (col_c_new, option_asc), - (col_a_plus_b_new, option_asc), - ], - // [c_new ASC, d_new ASC, a+b ASC] - vec![ - (col_c_new, option_asc), - (col_d_new, option_asc), - (col_a_plus_b_new, option_asc), - ], + // [c_new ASC] + vec![(col_c_new, option_asc)], + // [d_new ASC] + vec![(col_d_new, option_asc)], ], ), // ---------- TEST CASE 2 ------------ @@ -2164,18 +2121,10 @@ pub(crate) mod tests { vec![(col_e, col_a)], // expected vec![ - // [d_new ASC, c_new ASC, a+b ASC] - vec![ - (col_d_new, option_asc), - (col_c_new, option_asc), - (col_a_plus_b_new, option_asc), - ], - // [c_new ASC, d_new ASC, a+b ASC] - vec![ - (col_c_new, option_asc), - (col_d_new, option_asc), - (col_a_plus_b_new, option_asc), - ], + // [c_new ASC] + vec![(col_c_new, option_asc)], + // [d_new ASC] + vec![(col_d_new, option_asc)], ], ), // ---------- TEST CASE 3 ------------ diff --git a/datafusion/sqllogictest/test_files/order.slt b/datafusion/sqllogictest/test_files/order.slt index a267dddddd..4b136d24b0 100644 --- a/datafusion/sqllogictest/test_files/order.slt +++ b/datafusion/sqllogictest/test_files/order.slt @@ -674,8 +674,8 @@ SELECT DISTINCT time as "first_seen" FROM t ORDER BY 1; statement ok drop table t; -# Create a table having 3 columns which are ordering equivalent by the source. In the next step, -# we will expect to observe the removed SortExec by propagating the orders across projection. +# Create a table with three independently ordered columns. Their sum is not +# necessarily ordered because integer addition can wrap. statement ok CREATE EXTERNAL TABLE multiple_ordered_table ( a0 INTEGER, @@ -702,9 +702,10 @@ logical_plan 03)----TableScan: multiple_ordered_table projection=[a, b, c] physical_plan 01)SortPreservingMergeExec: [result@0 ASC NULLS LAST] -02)--ProjectionExec: expr=[b@1 + a@0 + c@2 as result] -03)----RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1, maintains_sort_order=true -04)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, b, c], output_orderings=[[a@0 ASC NULLS LAST], [b@1 ASC NULLS LAST], [c@2 ASC NULLS LAST]], file_type=csv, has_header=true +02)--SortExec: expr=[result@0 ASC NULLS LAST], preserve_partitioning=[true] +03)----ProjectionExec: expr=[b@1 + a@0 + c@2 as result] +04)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1, maintains_sort_order=true +05)--------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, b, c], output_orderings=[[a@0 ASC NULLS LAST], [b@1 ASC NULLS LAST], [c@2 ASC NULLS LAST]], file_type=csv, has_header=true statement ok drop table multiple_ordered_table; @@ -1845,6 +1846,38 @@ EXPLAIN SELECT a, named_struct('a', a, 'b', b) AS s FROM ordered_by_a ORDER BY s ---- physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/data/composite_order.csv]]}, projection=[a, named_struct(a, a@0, b, b@1) as s], output_ordering=[a@0 ASC NULLS LAST], file_type=csv, has_header=true +query I +COPY ( + SELECT * FROM (VALUES (1, 1), (2, 3), (200, 10), (255, 10)) AS t(a, b) + ORDER BY a +) +TO 'test_files/scratch/order/uint8_overflow.csv' +OPTIONS ('format.has_header' 'false'); +---- +4 + +statement ok +CREATE EXTERNAL TABLE ordered_u8 ( + a TINYINT UNSIGNED NOT NULL, + b TINYINT UNSIGNED NOT NULL +) +STORED AS CSV +LOCATION 'test_files/scratch/order/uint8_overflow.csv' +OPTIONS ('format.has_header' 'false') +WITH ORDER (a ASC) +WITH ORDER (b ASC); + +query I +SELECT (a + b) AS result FROM ordered_u8 ORDER BY result ASC; +---- +2 +5 +9 +210 + +statement ok +DROP TABLE ordered_u8; + # Config reset statement ok reset datafusion.catalog.information_schema; diff --git a/datafusion/sqllogictest/test_files/topk.slt b/datafusion/sqllogictest/test_files/topk.slt index e9c272889c..180350a735 100644 --- a/datafusion/sqllogictest/test_files/topk.slt +++ b/datafusion/sqllogictest/test_files/topk.slt @@ -374,14 +374,15 @@ physical_plan 02)--SortExec: TopK(fetch=3), expr=[number@0 DESC, letter@1 ASC NULLS LAST, age@2 DESC], preserve_partitioning=[false], sort_prefix=[number@0 DESC, letter@1 ASC NULLS LAST] 03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/topk/partial_sorted/1.parquet]]}, projection=[number, letter, age], output_ordering=[number@0 DESC, letter@1 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible -# Verify that the sort prefix is correctly computed over normalized, order-maintaining projections (number + 1, number, number + 1, age) +# `number + 1` is not order-maintaining (addition can overflow and wrap), so +# no sort prefix can be computed over the projected expression. query TT explain select number + 1 as number_plus, number, number + 1 as other_number_plus, age from partial_sorted order by number_plus desc, number desc, other_number_plus desc, age asc limit 3; ---- physical_plan 01)SortPreservingMergeExec: [number_plus@0 DESC, number@1 DESC, other_number_plus@2 DESC, age@3 ASC NULLS LAST], fetch=3 02)--ProjectionExec: expr=[__common_expr_1@0 as number_plus, number@1 as number, __common_expr_1@0 as other_number_plus, age@2 as age] -03)----SortExec: TopK(fetch=3), expr=[__common_expr_1@0 DESC, number@1 DESC, age@2 ASC NULLS LAST], preserve_partitioning=[true], sort_prefix=[__common_expr_1@0 DESC, number@1 DESC] +03)----SortExec: TopK(fetch=3), expr=[__common_expr_1@0 DESC, number@1 DESC, age@2 ASC NULLS LAST], preserve_partitioning=[true] 04)------ProjectionExec: expr=[CAST(number@0 AS Int64) + 1 as __common_expr_1, number@0 as number, age@1 as age] 05)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1, maintains_sort_order=true 06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/topk/partial_sorted/1.parquet]]}, projection=[number, age], output_ordering=[number@0 DESC], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
