You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/09/28 02:27:46 UTC
(datafusion) 01/01: test: cover unsorted contiguous groups in one partition (#24737)
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-24737-1bd2ad3efe70829c9524c5218aff8b4f2a93edbe
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 59d74cabfdcb358e93133b498d33bbc94ca9c934
Author: xavlee <52...@users.noreply.github.com>
AuthorDate: Mon Sep 28 02:27:21 2026 +0000
test: cover unsorted contiguous groups in one partition (#24737)
## Which issue does this PR relate to?
- Part of #24438.
## Rationale for this change
A source can concatenate several sorted logical runs into one DataFusion
output partition. The resulting stream may be globally unsorted while
every distinct `(key, time_bin)` tuple still occupies one contiguous
range.
This PR records how aggregate planning handles that layout before
grouped input properties are available. It provides the behavioral
baseline for the remaining PRs in the stack.
## What changes are included in this PR?
- Add a single-partition `TestMemoryExec` fixture containing two sorted
logical runs whose `(key, time_bin)` order resets at the record-batch
boundary.
- Aggregate by the complete `(key, time_bin)` tuple and verify the
result.
- Assert that planning selects `InputOrderMode::Linear`,
`EmissionType::Final`, and `SingleHashAggregateStream`.
## Stack
1. [#24737 — test: cover unsorted contiguous groups in one
partition](https://github.com/apache/datafusion/pull/24737) ← **this
PR**
2. [#24697 — refactor: separate aggregate group completion from input
ordering](https://github.com/apache/datafusion/pull/24697)
3. [#24698 — feat: add grouped equivalence
properties](https://github.com/apache/datafusion/pull/24698)
4. [#24497 — feat: stream aggregates over grouped
input](https://github.com/apache/datafusion/pull/24497)
## Are these changes tested?
The new aggregate test executes the single-partition input and
snapshot-checks all four grouped sums.
## Are there any user-facing changes?
No.
## Review this layer
[View only this PR
layer](https://github.com/apache/datafusion/pull/24737/commits/36969e72595162bbfd0c9a9f5fc07582ea386e2d)
---
datafusion/physical-plan/src/aggregates/mod.rs | 77 ++++++++++++++++++++++++++
1 file changed, 77 insertions(+)
diff --git a/datafusion/physical-plan/src/aggregates/mod.rs b/datafusion/physical-plan/src/aggregates/mod.rs
index 69568bcda3..d82d6de61c 100644
--- a/datafusion/physical-plan/src/aggregates/mod.rs
+++ b/datafusion/physical-plan/src/aggregates/mod.rs
@@ -5530,6 +5530,83 @@ mod tests {
Ok(())
}
+ #[tokio::test]
+ async fn unsorted_contiguous_groups_use_final_emission() -> Result<()> {
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("key", DataType::Int32, false),
+ Field::new("time_bin", DataType::Int64, false),
+ Field::new("value", DataType::Int64, false),
+ ]));
+ // Two sorted logical runs are emitted as batches in one DataFusion
+ // partition. Every distinct grouping tuple occupies one contiguous range,
+ // but tuple order resets at the batch boundary, so (key, time_bin) is not
+ // globally sorted.
+ let input_batches = vec![
+ RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![
+ Arc::new(Int32Array::from(vec![1, 1, 2, 2])),
+ Arc::new(Int64Array::from(vec![20, 20, 20, 20])),
+ Arc::new(Int64Array::from(vec![10, 20, 30, 40])),
+ ],
+ )?,
+ RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![
+ Arc::new(Int32Array::from(vec![1, 1, 2, 2])),
+ Arc::new(Int64Array::from(vec![0, 0, 0, 0])),
+ Arc::new(Int64Array::from(vec![50, 60, 70, 80])),
+ ],
+ )?,
+ ];
+ let group_by = PhysicalGroupBy::new_single(vec![
+ (col("key", &schema)?, "key".to_string()),
+ (col("time_bin", &schema)?, "time_bin".to_string()),
+ ]);
+ let aggr_expr = Arc::new(
+ AggregateExprBuilder::new(sum_udaf(), vec![col("value", &schema)?])
+ .schema(Arc::clone(&schema))
+ .alias("SUM(value)")
+ .build()?,
+ );
+ let input: Arc<dyn ExecutionPlan> =
+ TestMemoryExec::try_new_exec(&[input_batches], Arc::clone(&schema), None)?;
+ assert_eq!(input.output_partitioning().partition_count(), 1);
+
+ let aggregate = AggregateExec::try_new(
+ AggregateMode::Single,
+ group_by,
+ vec![aggr_expr],
+ vec![None],
+ input,
+ schema,
+ )?;
+
+ assert_eq!(aggregate.input_order_mode(), &InputOrderMode::Linear);
+ // This captures the behavior before #24438. When the source can declare
+ // `(key, time_bin)` group-contiguous, the corresponding case can use
+ // `EmissionType::Incremental`.
+ assert_eq!(aggregate.cache().emission_type, EmissionType::Final);
+
+ let task_ctx = new_migrated_hash_ctx(1024);
+ let stream = aggregate.execute_typed(0, &task_ctx)?;
+ assert!(matches!(stream, StreamType::SingleHash(_)));
+ let stream: SendableRecordBatchStream = stream.into();
+ let output = collect(stream).await?;
+ assert_snapshot!(batches_to_sort_string(&output), @r"
++-----+----------+------------+
+| key | time_bin | SUM(value) |
++-----+----------+------------+
+| 1 | 0 | 110 |
+| 1 | 20 | 30 |
+| 2 | 0 | 150 |
+| 2 | 20 | 70 |
++-----+----------+------------+
+");
+
+ Ok(())
+ }
+
/// Ensures for ordered input, `OrderedPartialAggregateStream` is used.
#[tokio::test]
async fn ordered_partial_aggregate_planning() -> Result<()> {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org