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-25072-62e30aaa2ee99dca5a0a4e7a7af65d631a34a9c0 in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit cc29a1a55610ab491af99593f88d440419b6dffe Author: Raz Luvaton <[email protected]> AuthorDate: Wed Sep 9 09:42:27 2026 +0000 fix: partial reduce should not propegate input ordering (#25072) ## Which issue does this PR close? N/A ## Rationale for this change Hash Aggregate Partial reduce propagate the ordering of the child even though it does not maintain child ordering ## What changes are included in this PR? clear ordering and test ## What is the testing strategy for this PR? unit test ## Are there any user-facing changes? not really --- datafusion/physical-plan/src/aggregates/mod.rs | 49 ++++++++++++++++++++++++++ 1 file changed, 49 insertions(+) diff --git a/datafusion/physical-plan/src/aggregates/mod.rs b/datafusion/physical-plan/src/aggregates/mod.rs index 3252cad956..91cd6b1b50 100644 --- a/datafusion/physical-plan/src/aggregates/mod.rs +++ b/datafusion/physical-plan/src/aggregates/mod.rs @@ -1423,6 +1423,12 @@ impl AggregateExec { .equivalence_properties() .project(group_expr_mapping, schema); + // An aggregation that does not maintain its input order must not + // propegrate the input's ordering either, match `maintains_input_order` value + if *input_order_mode == InputOrderMode::Linear { + eq_properties.clear_orderings(); + } + // True no-group aggregates produce only one row in each output // partition, so aggregate outputs are constants within the partition. // Grouping sets with empty grouping expressions are not covered here: @@ -4673,6 +4679,49 @@ mod tests { Ok(()) } + #[test] + fn partial_reduce_does_not_advertise_input_ordering() -> Result<()> { + let schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::UInt32, false), + Field::new("b", DataType::Float64, false), + ])); + let ordering = LexOrdering::new([PhysicalSortExpr::new_default(Arc::new( + Column::new("a", 0), + ))]) + .unwrap(); + let input = TestMemoryExec::try_new(&[vec![]], Arc::clone(&schema), None)? + .try_with_sort_information(vec![ordering])?; + let input = Arc::new(TestMemoryExec::update_cache(&Arc::new(input))); + assert!( + input.properties().output_ordering().is_some(), + "test setup: the input is ordered by the group key" + ); + + let partial_reduce = AggregateExec::try_new( + AggregateMode::PartialReduce, + PhysicalGroupBy::new_single(vec![(col("a", &schema)?, "a".to_string())]), + vec![Arc::new( + AggregateExprBuilder::new(sum_udaf(), vec![col("b", &schema)?]) + .schema(Arc::clone(&schema)) + .alias("SUM(b)") + .build()?, + )], + vec![None], + input, + Arc::clone(&schema), + )?; + + assert_eq!(partial_reduce.input_order_mode(), &InputOrderMode::Linear); + assert_eq!(partial_reduce.maintains_input_order(), vec![false]); + assert!( + partial_reduce.properties().output_ordering().is_none(), + "partial reduce advertised an ordering it does not maintain: {:?}", + partial_reduce.properties().output_ordering() + ); + + Ok(()) + } + fn partial_reduce_test_aggregate() -> Result<AggregateExec> { partial_reduce_test_aggregate_with_batches(1) } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
