You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "xavlee (via GitHub)" <gi...@apache.org> on 2026/08/27 18:54:03 UTC
[PR] test: cover unsorted contiguous groups in one partition [datafusion]
xavlee opened a new pull request, #24737:
URL: https://github.com/apache/datafusion/pull/24737
## Which issue does this PR relate to?
- Part of #24438.
## Rationale for this change
A source can concatenate sorted logical runs into one DataFusion output partition. The resulting stream can be globally unsorted while every distinct `(key, time_bin)` grouping tuple still occupies one contiguous range. This fixture records how aggregate execution handles that layout today.
## What changes are included in this PR?
- Add a one-partition test source that yields one batch and then remains pending.
- Concatenate two sorted logical runs whose `(key, time_bin)` order resets between runs.
- Group by the complete `(key, time_bin)` tuple.
- Assert `InputOrderMode::Linear`, `EmissionType::Final`, the ordinary single-hash stream, and pending output before EOF.
## Are these changes tested?
Yes. The new `group_contiguous_linear_aggregate_waits_for_input_end` test passes as part of:
- `cargo test -p datafusion-physical-plan --lib`
- workspace-wide clippy with all targets, all features, and warnings denied
- the extended workspace test suite with the CI profile and feature set
## Are there any user-facing changes?
This PR adds test coverage for existing aggregate behavior.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] test: cover unsorted contiguous groups in one partition [datafusion]
Posted by "xudong963 (via GitHub)" <gi...@apache.org>.
xudong963 commented on PR #24737:
URL: https://github.com/apache/datafusion/pull/24737#issuecomment-5862198573
Thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] test: cover unsorted contiguous groups in one partition [datafusion]
Posted by "xudong963 (via GitHub)" <gi...@apache.org>.
xudong963 merged PR #24737:
URL: https://github.com/apache/datafusion/pull/24737
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] test: cover unsorted contiguous groups in one partition [datafusion]
Posted by "gene-bordegaray (via GitHub)" <gi...@apache.org>.
gene-bordegaray commented on code in PR #24737:
URL: https://github.com/apache/datafusion/pull/24737#discussion_r3896780329
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -4539,6 +4539,60 @@ mod tests {
Ok(())
}
+ #[test]
+ 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 concatenated into one DataFusion partition.
+ // Every distinct grouping tuple occupies one contiguous range, but tuple
+ // order resets between runs, so (key, time_bin) is not globally sorted.
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![
+ Arc::new(Int32Array::from(vec![1, 1, 2, 2, 1, 1, 2, 2])),
+ Arc::new(Int64Array::from(vec![20, 20, 20, 20, 0, 0, 0, 0])),
+ Arc::new(Int64Array::from(vec![10, 20, 30, 40, 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(&[vec![batch]], 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)?;
Review Comment:
non blockiung: can we make this a tokiop test and colelct the stream that actaulla ssert the results
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -4539,6 +4539,60 @@ mod tests {
Ok(())
}
+ #[test]
+ fn unsorted_contiguous_groups_use_final_emission() -> Result<()> {
Review Comment:
done at a batch boundary
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -4539,6 +4539,60 @@ mod tests {
Ok(())
}
+ #[test]
+ fn unsorted_contiguous_groups_use_final_emission() -> Result<()> {
Review Comment:
for this use case i think it would also be good if there were two batches to show that this works over multiple polls 👍
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] test: cover unsorted contiguous groups in one partition [datafusion]
Posted by "xavlee (via GitHub)" <gi...@apache.org>.
xavlee commented on code in PR #24737:
URL: https://github.com/apache/datafusion/pull/24737#discussion_r3896922670
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -4539,6 +4539,60 @@ mod tests {
Ok(())
}
+ #[test]
+ fn unsorted_contiguous_groups_use_final_emission() -> Result<()> {
Review Comment:
Yeah that's a great point, thanks gene.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] test: cover unsorted contiguous groups in one partition [datafusion]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #24737:
URL: https://github.com/apache/datafusion/pull/24737#issuecomment-5444466567
## [Codecov](https://app.codecov.io/gh/apache/datafusion/pull/24737?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `76.31579%` with `9 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 81.42%. Comparing base ([`ee59f62`](https://app.codecov.io/gh/apache/datafusion/commit/ee59f628b44eb80e8a4f126288632ef51fda5dc2?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`060dd38`](https://app.codecov.io/gh/apache/datafusion/commit/060dd38dd02f843abbaa2133373614e9b76131a8?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
| [Files with missing lines](https://app.codecov.io/gh/apache/datafusion/pull/24737?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [datafusion/physical-plan/src/aggregates/mod.rs](https://app.codecov.io/gh/apache/datafusion/pull/24737?src=pr&el=tree&filepath=datafusion%2Fphysical-plan%2Fsrc%2Faggregates%2Fmod.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZGF0YWZ1c2lvbi9waHlzaWNhbC1wbGFuL3NyYy9hZ2dyZWdhdGVzL21vZC5ycw==) | 76.31% | [2 Missing and 7 partials :warning: ](https://app.codecov.io/gh/apache/datafusion/pull/24737?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #24737 +/- ##
==========================================
- Coverage 81.42% 81.42% -0.01%
==========================================
Files 1121 1121
Lines 402142 402180 +38
Branches 402142 402180 +38
==========================================
+ Hits 327460 327482 +22
- Misses 55484 55490 +6
- Partials 19198 19208 +10
```
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/datafusion/pull/24737?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] test: cover unsorted contiguous groups in one partition [datafusion]
Posted by "gene-bordegaray (via GitHub)" <gi...@apache.org>.
gene-bordegaray commented on PR #24737:
URL: https://github.com/apache/datafusion/pull/24737#issuecomment-5779141532
rebumping this, just went over again and looks good.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org