diff --git a/datafusion/core/tests/fuzz_cases/equivalence/ordering.rs b/datafusion/core/tests/fuzz_cases/equivalence/ordering.rs index a57095066ee12..81b098d8c1093 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,17 @@ 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}" - ); + let may_overflow = ordering + .iter() + .any(|sort_expr| contains_overflowable_arithmetic(&sort_expr.expr)); + if eq_properties.ordering_satisfy(ordering)? { + assert!(expected, "{err_msg}"); + } else if !may_overflow { + // A rejection is only conclusive without `+`/`-`: the + // sample can be sorted even when possible overflow makes + // the ordering underivable. + 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..34653b513f60e 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,22 @@ 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}" - ); + let may_overflow = ordering.iter().any(|sort_expr| { + projection_mapping.iter().any(|(source, targets)| { + targets + .iter() + .any(|(target, _)| target.eq(&sort_expr.expr)) + && contains_overflowable_arithmetic(source) + }) + }); + if projected_eq.ordering_satisfy(ordering)? { + assert!(expected, "{err_msg}"); + } else if !may_overflow { + // A rejection is only conclusive without `+`/`-`: + // the sample can be sorted even when possible + // overflow makes the ordering underivable. + 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..4b543792d8e9f 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::*; @@ -216,6 +217,18 @@ fn add_equal_conditions_test() -> Result<()> { /// still be differentiated. When sorting the extended table, the unique column acts as a tie-breaker to produce /// deterministic sorting results. /// +/// 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() +} + /// If the table remains the same after sorting with the added unique column, it indicates that the table was /// already sorted according to `required_ordering` to begin with. pub fn is_table_same_after_sort( 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..fae528ae69f16 100644 --- a/datafusion/physical-expr/src/expressions/binary.rs +++ b/datafusion/physical-expr/src/expressions/binary.rs @@ -33,7 +33,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 +118,37 @@ 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, + }; + if !wraps_in_domain && (self.fail_on_overflow || !range.is_unbounded()) { + sort_properties + } else { + SortProperties::Unordered + } + } } impl std::fmt::Display for BinaryExpr { @@ -760,18 +791,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)?, @@ -1307,6 +1354,68 @@ mod tests { use crate::planner::logical2physical; use arrow::array::BooleanArray; 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(()) + } + /// 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