diff --git a/datafusion/core/tests/fuzz_cases/equivalence/ordering.rs b/datafusion/core/tests/fuzz_cases/equivalence/ordering.rs index a57095066ee12..60b09976355e9 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 2f67e211ce915..9593e1cf11565 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 8350cafb215cb..ca73db3ae99ec 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) -> bool { + expr.exists(|e| { + Ok(e.downcast_ref::().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 15637d24e8a4b..499187a603979 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 d2a8c2f654cf0..bd8bef84de2d8 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 8b71b3ec409c7..bafe73b046c16 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, diff --git a/datafusion/physical-expr/src/projection.rs b/datafusion/physical-expr/src/projection.rs index 1f6a6eb08fb78..bfa40c7838734 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 a267ddddddd54..4b136d24b0751 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 e9c272889cb4a..180350a735b46 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