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/10/16 09:30:54 UTC
(datafusion) 01/01: refactor: use group clustering terminology in aggregate execution (#24697)
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-24697-b6f6990ed5f926fd04fe04c46410f8c6ddfb7674
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit d93699ce388af4153dd6e9f1a6a2c74855b42a5e
Author: xavlee <52...@users.noreply.github.com>
AuthorDate: Fri Oct 16 09:30:27 2026 +0000
refactor: use group clustering terminology in aggregate execution (#24697)
## Which issue does this PR relate to?
- Part of #24438.
- Builds on the baseline test in #24737 (merged).
## Rationale for this change
Aggregate execution currently uses `InputOrderMode` to select execution
paths and recognize completed groups.
A group can be completed incrementally when its rows form one contiguous
run: once the run ends, that group cannot appear again in the same input
partition. Sorted input supplies this guarantee, and unsorted input can
supply it too:
```text
AAABBBCCCC -> sorted; groups can be completed incrementally
CCCAAABBB -> unsorted; groups can still be completed incrementally
```
This PR describes the input's _**clustering**_ guarantee with
`GroupClusteringMode` and uses clustering terminology throughout the
aggregate execution machinery. Aggregate construction derives the mode
directly from the existing ordering analysis. The follow-up in #24497
will also derive it from grouped equivalence properties.
## What changes are included in this PR?
- Replace the aggregate's `InputOrderMode` field with
`GroupClusteringMode::{None, Partial, Full}`:
- `None`: Groups have no apparent clustering guarantee.
- `Partial(indices)`: rows are contiguous on a subset of grouping
expressions; a change in that tuple completes all groups in the
preceding run.
- `Full`: rows are contiguous on the complete grouping tuple; a change
in that tuple completes the preceding group.
- Rename the stateful `GroupOrdering` tracker and its `Partial` and
`Full` implementations to `GroupClustering`, `GroupClusteringPartial`,
and `GroupClusteringFull`.
- Rename the ordered aggregate streams, shared table, and adjacent-key
group-values implementation to use clustered terminology,
- Preserve actual sort expressions and options through equivalence
properties and input-ordering requirements.
- Display `group_clustering_mode` in aggregate plans and document the
public API migration in the DataFusion 56 upgrade guide.
## Stack
1. [#24737 — test: cover unsorted contiguous groups in one
partition](https://github.com/apache/datafusion/pull/24737) (merged)
2. [#24697 — refactor: use group clustering terminology in aggregate
execution](https://github.com/apache/datafusion/pull/24697) ← **this
PR**
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 there any user-facing changes?
Yes. This PR changes public Rust APIs:
- `AggregateExec::input_order_mode()` is replaced by
`AggregateExec::group_clustering_mode()`, returning the public
`GroupClusteringMode` enum.
- `AggregateExec::compute_properties` accepts `&GroupClusteringMode` in
place of `&InputOrderMode`.
- The public `GroupOrdering`, `GroupOrderingPartial`, and
`GroupOrderingFull` types in `aggregates::order` are renamed to
`GroupClustering`, `GroupClusteringPartial`, and `GroupClusteringFull`.
- `GroupClustering::try_new` takes `&GroupClusteringMode`;
`new_group_values` takes `&GroupClustering`.
Aggregate plan output uses `group_clustering_mode=Full` and
`group_clustering_mode=Partial(indices)` in place of
`ordering_mode=Sorted` and `ordering_mode=PartiallySorted(indices)`. SQL
results and aggregate execution-path selection retain their existing
behavior.
## Review this layer
[View this PR's
diff](https://github.com/apache/datafusion/pull/24697/files)
---
datafusion/core/tests/dataframe/mod.rs | 8 +-
datafusion/core/tests/fuzz_cases/aggregate_fuzz.rs | 19 +-
.../fuzz_cases/aggregation_fuzzer/query_builder.rs | 4 +-
.../physical_optimizer/enforce_distribution.rs | 8 +-
.../limited_distinct_aggregation.rs | 2 +-
.../core/tests/physical_optimizer/pushdown_sort.rs | 4 +-
.../benches/dictionary_group_values.rs | 14 +-
.../physical-plan/benches/ordered_group_values.rs | 22 +-
.../physical-plan/benches/partial_ordering.rs | 16 +-
...red_final_table.rs => clustered_final_table.rs} | 30 +-
...partial_table.rs => clustered_partial_table.rs} | 28 +-
...d_single_table.rs => clustered_single_table.rs} | 22 +-
.../src/aggregates/aggregate_hash_table/common.rs | 4 +-
.../{common_ordered.rs => common_clustered.rs} | 92 +++---
.../src/aggregates/aggregate_hash_table/mod.rs | 12 +-
.../aggregate_hash_table/partial_table.rs | 4 +-
...d_final_stream.rs => clustered_final_stream.rs} | 95 +++---
...rtial_stream.rs => clustered_partial_stream.rs} | 95 +++---
...single_stream.rs => clustered_single_stream.rs} | 228 ++++++-------
.../src/aggregates/group_values/mod.rs | 24 +-
.../multi_group_by/{ordered.rs => clustered.rs} | 82 +++--
.../aggregates/group_values/multi_group_by/list.rs | 12 +-
.../aggregates/group_values/multi_group_by/mod.rs | 4 +-
.../src/aggregates/grouped_hash_stream.rs | 60 ++--
.../physical-plan/src/aggregates/hash_stream.rs | 29 +-
datafusion/physical-plan/src/aggregates/mod.rs | 368 ++++++++++++++-------
.../physical-plan/src/aggregates/order/full.rs | 16 +-
.../physical-plan/src/aggregates/order/mod.rs | 202 ++++++-----
.../physical-plan/src/aggregates/order/partial.rs | 194 +++++------
.../src/aggregates/partial_reduce_stream.rs | 5 +-
.../physical-plan/src/aggregates/single_stream.rs | 13 +-
datafusion/physical-plan/src/aggregates/spill.rs | 49 +--
datafusion/physical-plan/src/recursive_query.rs | 4 +-
.../test_files/agg_func_substitute.slt | 12 +-
datafusion/sqllogictest/test_files/aggregate.slt | 12 +-
datafusion/sqllogictest/test_files/group_by.slt | 58 ++--
datafusion/sqllogictest/test_files/joins.slt | 8 +-
datafusion/sqllogictest/test_files/order.slt | 2 +-
.../test_files/ordered_aggregate_spill.slt | 24 +-
.../test_files/preserve_file_partitioning.slt | 12 +-
.../test_files/range_sorted_time_bin_agg.slt | 14 +-
.../test_files/repartition_subset_satisfaction.slt | 12 +-
.../sqllogictest/test_files/sort_pushdown.slt | 10 +-
datafusion/sqllogictest/test_files/unnest.slt | 4 +-
datafusion/sqllogictest/test_files/window.slt | 2 +-
docs/source/library-user-guide/upgrading/56.0.0.md | 21 ++
46 files changed, 1099 insertions(+), 861 deletions(-)
diff --git a/datafusion/core/tests/dataframe/mod.rs b/datafusion/core/tests/dataframe/mod.rs
index 43ceee3444..d011a25273 100644
--- a/datafusion/core/tests/dataframe/mod.rs
+++ b/datafusion/core/tests/dataframe/mod.rs
@@ -3422,9 +3422,9 @@ async fn union_with_mix_of_presorted_and_explicitly_resorted_inputs_with_reparti
assert_snapshot!(
union_with_mix_of_presorted_and_explicitly_resorted_inputs_impl(false).await?,
@r"
- AggregateExec: mode=Final, gby=[id@0 as id], aggr=[], ordering_mode=Sorted
+ AggregateExec: mode=Final, gby=[id@0 as id], aggr=[], group_clustering_mode=Full
SortPreservingMergeExec: [id@0 ASC NULLS LAST]
- AggregateExec: mode=Partial, gby=[id@0 as id], aggr=[], ordering_mode=Sorted
+ AggregateExec: mode=Partial, gby=[id@0 as id], aggr=[], group_clustering_mode=Full
UnionExec
DataSourceExec: file_groups={1 group: [[{testdata}/alltypes_tiny_pages.parquet]]}, projection=[id], output_ordering=[id@0 ASC NULLS LAST], file_type=parquet
SortExec: expr=[id@0 ASC NULLS LAST], preserve_partitioning=[false]
@@ -3440,9 +3440,9 @@ async fn union_with_mix_of_presorted_and_explicitly_resorted_inputs_with_reparti
assert_snapshot!(
union_with_mix_of_presorted_and_explicitly_resorted_inputs_impl(true).await?,
@r"
- AggregateExec: mode=Final, gby=[id@0 as id], aggr=[], ordering_mode=Sorted
+ AggregateExec: mode=Final, gby=[id@0 as id], aggr=[], group_clustering_mode=Full
SortPreservingMergeExec: [id@0 ASC NULLS LAST]
- AggregateExec: mode=Partial, gby=[id@0 as id], aggr=[], ordering_mode=Sorted
+ AggregateExec: mode=Partial, gby=[id@0 as id], aggr=[], group_clustering_mode=Full
UnionExec
DataSourceExec: file_groups={1 group: [[{testdata}/alltypes_tiny_pages.parquet]]}, projection=[id], output_ordering=[id@0 ASC NULLS LAST], file_type=parquet
SortExec: expr=[id@0 ASC NULLS LAST], preserve_partitioning=[false]
diff --git a/datafusion/core/tests/fuzz_cases/aggregate_fuzz.rs b/datafusion/core/tests/fuzz_cases/aggregate_fuzz.rs
index afbb1b21d0..74cc881c22 100644
--- a/datafusion/core/tests/fuzz_cases/aggregate_fuzz.rs
+++ b/datafusion/core/tests/fuzz_cases/aggregate_fuzz.rs
@@ -41,7 +41,6 @@ use datafusion_common_runtime::JoinSet;
use datafusion_functions_aggregate::sum::sum_udaf;
use datafusion_physical_expr::PhysicalSortExpr;
use datafusion_physical_expr::expressions::{Column, col, lit};
-use datafusion_physical_plan::InputOrderMode;
use test_utils::{StringBatchGenerator, add_empty_batches};
use datafusion_execution::TaskContext;
@@ -49,7 +48,7 @@ use datafusion_execution::memory_pool::FairSpillPool;
use datafusion_execution::runtime_env::RuntimeEnvBuilder;
use datafusion_physical_expr::aggregate::AggregateExprBuilder;
use datafusion_physical_plan::aggregates::{
- AggregateExec, AggregateMode, PhysicalGroupBy,
+ AggregateExec, AggregateMode, GroupClusteringMode, PhysicalGroupBy,
};
use datafusion_physical_plan::metrics::MetricValue;
use datafusion_physical_plan::{ExecutionPlan, collect, displayable};
@@ -303,7 +302,7 @@ async fn streaming_aggregate_test() {
/// two `AggregateExec` variants produce the same result: the pipeline breaking
/// one over unordered input (`PartialHashAggregateStream`) and the
/// non-pipeline breaking one over ordered input
-/// (`OrderedPartialAggregateStream`).
+/// (`ClusteredPartialAggregateStream`).
async fn run_aggregate_test(input1: Vec<RecordBatch>, group_by_columns: Vec<&str>) {
let schema = input1[0].schema();
let session_config = SessionConfig::new().with_batch_size(50);
@@ -354,8 +353,8 @@ async fn run_aggregate_test(input1: Vec<RecordBatch>, group_by_columns: Vec<&str
.unwrap(),
);
assert_ne!(
- aggregate_exec_running.input_order_mode(),
- &InputOrderMode::Linear,
+ aggregate_exec_running.group_clustering_mode(),
+ &GroupClusteringMode::None,
"running aggregate should observe ordered input for group_by: {group_by:?}"
);
@@ -555,13 +554,17 @@ async fn verify_ordered_aggregate(frame: &DataFrame, expected_sort: bool) {
fn f_down(&mut self, node: &'n Self::Node) -> Result<TreeNodeRecursion> {
if let Some(exec) = node.downcast_ref::<AggregateExec>() {
+ assert_eq!(
+ exec.properties().output_ordering().is_some(),
+ self.expected_sort
+ );
if self.expected_sort {
assert!(matches!(
- exec.input_order_mode(),
- InputOrderMode::PartiallySorted(_) | InputOrderMode::Sorted
+ exec.group_clustering_mode(),
+ GroupClusteringMode::Partial(_) | GroupClusteringMode::Full
));
} else {
- assert_eq!(*exec.input_order_mode(), InputOrderMode::Linear);
+ assert_eq!(*exec.group_clustering_mode(), GroupClusteringMode::None);
}
}
Ok(TreeNodeRecursion::Continue)
diff --git a/datafusion/core/tests/fuzz_cases/aggregation_fuzzer/query_builder.rs b/datafusion/core/tests/fuzz_cases/aggregation_fuzzer/query_builder.rs
index 2b2d2483d7..b4df25a006 100644
--- a/datafusion/core/tests/fuzz_cases/aggregation_fuzzer/query_builder.rs
+++ b/datafusion/core/tests/fuzz_cases/aggregation_fuzzer/query_builder.rs
@@ -93,9 +93,9 @@ pub struct QueryBuilder {
/// ...
/// ```
///
- /// More details can see [`GroupOrdering`].
+ /// More details can see [`GroupClustering`].
///
- /// [`GroupOrdering`]: datafusion_physical_plan::aggregates::order::GroupOrdering
+ /// [`GroupClustering`]: datafusion_physical_plan::aggregates::order::GroupClustering
dataset_sort_keys: Vec<Vec<String>>,
/// If we will also test the no grouping case like:
diff --git a/datafusion/core/tests/physical_optimizer/enforce_distribution.rs b/datafusion/core/tests/physical_optimizer/enforce_distribution.rs
index 6c00c0345d..283d22767f 100644
--- a/datafusion/core/tests/physical_optimizer/enforce_distribution.rs
+++ b/datafusion/core/tests/physical_optimizer/enforce_distribution.rs
@@ -5471,9 +5471,9 @@ fn preserve_ordering_for_streaming_sorted_aggregate() -> Result<()> {
let plan_distrib = test_config.to_plan(physical_plan.clone(), &DISTRIB_DISTRIB_SORT);
assert_plan!(plan_distrib, @r"
- AggregateExec: mode=FinalPartitioned, gby=[a@0 as a], aggr=[COUNT(b)], ordering_mode=Sorted
+ AggregateExec: mode=FinalPartitioned, gby=[a@0 as a], aggr=[COUNT(b)], group_clustering_mode=Full
RepartitionExec: partitioning=Hash([a@0], 2), input_partitions=2, preserve_order=true, sort_exprs=a@0 ASC
- AggregateExec: mode=Partial, gby=[a@0 as a], aggr=[COUNT(b)], ordering_mode=Sorted
+ AggregateExec: mode=Partial, gby=[a@0 as a], aggr=[COUNT(b)], group_clustering_mode=Full
DataSourceExec: file_groups={2 groups: [[x], [y]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC], file_type=parquet
");
@@ -5505,9 +5505,9 @@ fn preserve_ordering_for_streaming_partially_sorted_aggregate() -> Result<()> {
let plan_distrib = test_config.to_plan(physical_plan.clone(), &DISTRIB_DISTRIB_SORT);
assert_plan!(plan_distrib, @r"
- AggregateExec: mode=FinalPartitioned, gby=[a@0 as a, b@1 as b], aggr=[COUNT(c)], ordering_mode=PartiallySorted([0])
+ AggregateExec: mode=FinalPartitioned, gby=[a@0 as a, b@1 as b], aggr=[COUNT(c)], group_clustering_mode=Partial([0])
RepartitionExec: partitioning=Hash([a@0, b@1], 2), input_partitions=2, preserve_order=true, sort_exprs=a@0 ASC
- AggregateExec: mode=Partial, gby=[a@0 as a, b@1 as b], aggr=[COUNT(c)], ordering_mode=PartiallySorted([0])
+ AggregateExec: mode=Partial, gby=[a@0 as a, b@1 as b], aggr=[COUNT(c)], group_clustering_mode=Partial([0])
DataSourceExec: file_groups={2 groups: [[x], [y]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC], file_type=parquet
");
diff --git a/datafusion/core/tests/physical_optimizer/limited_distinct_aggregation.rs b/datafusion/core/tests/physical_optimizer/limited_distinct_aggregation.rs
index fae73af425..5ca6beccdb 100644
--- a/datafusion/core/tests/physical_optimizer/limited_distinct_aggregation.rs
+++ b/datafusion/core/tests/physical_optimizer/limited_distinct_aggregation.rs
@@ -549,7 +549,7 @@ fn test_has_order_by() -> Result<()> {
actual,
@r"
LocalLimitExec: fetch=10
- AggregateExec: mode=Single, gby=[a@0 as a], aggr=[], ordering_mode=Sorted
+ AggregateExec: mode=Single, gby=[a@0 as a], aggr=[], group_clustering_mode=Full
DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC], file_type=parquet
"
);
diff --git a/datafusion/core/tests/physical_optimizer/pushdown_sort.rs b/datafusion/core/tests/physical_optimizer/pushdown_sort.rs
index b72563a942..6cf05ecfd8 100644
--- a/datafusion/core/tests/physical_optimizer/pushdown_sort.rs
+++ b/datafusion/core/tests/physical_optimizer/pushdown_sort.rs
@@ -731,13 +731,13 @@ fn test_pushdown_through_blocking_node() {
OptimizationTest:
input:
- SortExec: expr=[a@0 ASC], preserve_partitioning=[false]
- - AggregateExec: mode=Final, gby=[a@0 as a], aggr=[COUNT(b)], ordering_mode=Sorted
+ - AggregateExec: mode=Final, gby=[a@0 as a], aggr=[COUNT(b)], group_clustering_mode=Full
- SortExec: expr=[a@0 DESC NULLS LAST], preserve_partitioning=[false]
- DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC], file_type=parquet
output:
Ok:
- SortExec: expr=[a@0 ASC], preserve_partitioning=[false]
- - AggregateExec: mode=Final, gby=[a@0 as a], aggr=[COUNT(b)], ordering_mode=Sorted
+ - AggregateExec: mode=Final, gby=[a@0 as a], aggr=[COUNT(b)], group_clustering_mode=Full
- SortExec: expr=[a@0 DESC NULLS LAST], preserve_partitioning=[false]
- DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], file_type=parquet, sort_order_for_reorder=[a@0 DESC NULLS LAST], reverse_row_groups=true
"
diff --git a/datafusion/physical-plan/benches/dictionary_group_values.rs b/datafusion/physical-plan/benches/dictionary_group_values.rs
index 502869787e..51363c0bdf 100644
--- a/datafusion/physical-plan/benches/dictionary_group_values.rs
+++ b/datafusion/physical-plan/benches/dictionary_group_values.rs
@@ -29,7 +29,7 @@ use criterion::{
};
use datafusion_expr::EmitTo;
use datafusion_physical_plan::aggregates::group_values::new_group_values;
-use datafusion_physical_plan::aggregates::order::{GroupOrdering, GroupOrderingFull};
+use datafusion_physical_plan::aggregates::order::{GroupClustering, GroupClusteringFull};
use rand::rngs::StdRng;
use rand::seq::SliceRandom;
use rand::{Rng, SeedableRng};
@@ -106,7 +106,7 @@ fn bench_intern_emit(c: &mut Criterion) {
b.iter_batched_ref(
|| {
(
- new_group_values(schema.clone(), &GroupOrdering::None)
+ new_group_values(schema.clone(), &GroupClustering::None)
.unwrap(),
Vec::<usize>::with_capacity(size),
)
@@ -151,7 +151,7 @@ fn bench_repeated_intern_emit(c: &mut Criterion) {
b.iter_batched_ref(
|| {
(
- new_group_values(schema.clone(), &GroupOrdering::None)
+ new_group_values(schema.clone(), &GroupClustering::None)
.unwrap(),
Vec::<usize>::with_capacity(size),
)
@@ -172,7 +172,7 @@ fn bench_repeated_intern_emit(c: &mut Criterion) {
group.finish();
}
-// GroupOrdering::Full -> GroupValuesColumn::<true>: scalar append_val/equal_to path.
+// GroupClustering::Full -> GroupValuesColumn::<true>: scalar append_val/equal_to path.
fn bench_scalar_append_equal(c: &mut Criterion) {
let mut group = c.benchmark_group("dict_scalar_append_equal");
let schema = dict_schema();
@@ -192,7 +192,7 @@ fn bench_scalar_append_equal(c: &mut Criterion) {
(
new_group_values(
schema.clone(),
- &GroupOrdering::Full(GroupOrderingFull::new()),
+ &GroupClustering::Full(GroupClusteringFull::new()),
)
.unwrap(),
Vec::<usize>::with_capacity(size),
@@ -227,7 +227,7 @@ fn bench_take_n(c: &mut Criterion) {
b.iter_batched_ref(
|| {
(
- new_group_values(schema.clone(), &GroupOrdering::None).unwrap(),
+ new_group_values(schema.clone(), &GroupClustering::None).unwrap(),
Vec::<usize>::with_capacity(size),
)
},
@@ -298,7 +298,7 @@ fn bench_shared_values_arc(c: &mut Criterion) {
b.iter_batched_ref(
|| {
(
- new_group_values(schema.clone(), &GroupOrdering::None)
+ new_group_values(schema.clone(), &GroupClustering::None)
.unwrap(),
Vec::<usize>::with_capacity(size),
)
diff --git a/datafusion/physical-plan/benches/ordered_group_values.rs b/datafusion/physical-plan/benches/ordered_group_values.rs
index 30af725cdd..fc60423c8b 100644
--- a/datafusion/physical-plan/benches/ordered_group_values.rs
+++ b/datafusion/physical-plan/benches/ordered_group_values.rs
@@ -38,12 +38,12 @@ use datafusion_physical_expr::expressions::col;
use datafusion_physical_expr::{LexOrdering, PhysicalSortExpr};
use datafusion_physical_plan::aggregates::group_values::multi_group_by::GroupValuesColumn;
use datafusion_physical_plan::aggregates::group_values::{GroupValues, new_group_values};
-use datafusion_physical_plan::aggregates::order::GroupOrdering;
+use datafusion_physical_plan::aggregates::order::GroupClustering;
use datafusion_physical_plan::aggregates::{
- AggregateExec, AggregateMode, PhysicalGroupBy,
+ AggregateExec, AggregateMode, GroupClusteringMode, PhysicalGroupBy,
};
use datafusion_physical_plan::test::TestMemoryExec;
-use datafusion_physical_plan::{ExecutionPlan, InputOrderMode, collect};
+use datafusion_physical_plan::{ExecutionPlan, collect};
use tokio::runtime::Runtime;
const ROWS: usize = 131_072;
@@ -112,8 +112,10 @@ fn grouping(c: &mut Criterion) {
let values: Box<dyn GroupValues> = if selected {
new_group_values(
Arc::clone(&schema),
- &GroupOrdering::try_new(&InputOrderMode::Sorted)
- .unwrap(),
+ &GroupClustering::try_new(
+ &GroupClusteringMode::Full,
+ )
+ .unwrap(),
)
.unwrap()
} else {
@@ -158,7 +160,7 @@ fn check_case(schema: &SchemaRef, batches: &[Vec<ArrayRef>]) -> (usize, usize) {
let mut hashed = GroupValuesColumn::<true>::try_new(Arc::clone(schema)).unwrap();
let mut selected = new_group_values(
Arc::clone(schema),
- &GroupOrdering::try_new(&InputOrderMode::Sorted).unwrap(),
+ &GroupClustering::try_new(&GroupClusteringMode::Full).unwrap(),
)
.unwrap();
let mut expected = Vec::new();
@@ -188,7 +190,7 @@ fn aggregate_plan(
schema: &SchemaRef,
keys: Vec<Vec<ArrayRef>>,
sort_columns: &[&str],
- input_order_mode: &InputOrderMode,
+ group_clustering_mode: &GroupClusteringMode,
) -> Arc<dyn ExecutionPlan> {
let mut fields = schema.fields().to_vec();
fields.push(Arc::new(Field::new("v", DataType::Int64, false)));
@@ -232,7 +234,7 @@ fn aggregate_plan(
schema,
)
.unwrap();
- assert_eq!(plan.input_order_mode(), input_order_mode);
+ assert_eq!(plan.group_clustering_mode(), group_clustering_mode);
Arc::new(plan)
}
@@ -246,7 +248,7 @@ fn aggregation(c: &mut Criterion) {
for run_length in [1, 8, 128, 8192] {
let (schema, keys) = inputs(run_length, 8192, strings);
let plan =
- aggregate_plan(&schema, keys, &["a", "b"], &InputOrderMode::Sorted);
+ aggregate_plan(&schema, keys, &["a", "b"], &GroupClusteringMode::Full);
let name =
format!("{}_run{run_length}", if strings { "string" } else { "int" });
group.bench_function(name, |b| {
@@ -291,7 +293,7 @@ fn partially_ordered_aggregation(c: &mut Criterion) {
&schema,
keys,
&["a"],
- &InputOrderMode::PartiallySorted(vec![0]),
+ &GroupClusteringMode::Partial(vec![0]),
);
let runtime = Runtime::new().unwrap();
let mut group = c.benchmark_group("partially_ordered_aggregate_exec");
diff --git a/datafusion/physical-plan/benches/partial_ordering.rs b/datafusion/physical-plan/benches/partial_ordering.rs
index bdadd6274b..967e84d78a 100644
--- a/datafusion/physical-plan/benches/partial_ordering.rs
+++ b/datafusion/physical-plan/benches/partial_ordering.rs
@@ -18,7 +18,7 @@
use std::sync::Arc;
use arrow::array::{ArrayRef, Int32Array};
-use datafusion_physical_plan::aggregates::order::GroupOrderingPartial;
+use datafusion_physical_plan::aggregates::order::GroupClusteringPartial;
use criterion::{Criterion, criterion_group, criterion_main};
@@ -34,20 +34,20 @@ fn create_test_arrays(num_columns: usize) -> Vec<ArrayRef> {
.collect()
}
fn bench_new_groups(c: &mut Criterion) {
- let mut group = c.benchmark_group("group_ordering_partial");
+ let mut group = c.benchmark_group("group_clustering_partial");
- // Test with 1, 2, 4, and 8 order indices
+ // Test with 1, 2, 4, and 8 grouping indices
for num_columns in [1, 2, 4, 8] {
- let order_indices: Vec<usize> = (0..num_columns).collect();
+ let grouping_indices: Vec<usize> = (0..num_columns).collect();
- group.bench_function(format!("order_indices_{num_columns}"), |b| {
+ group.bench_function(format!("grouping_indices_{num_columns}"), |b| {
let batch_group_values = create_test_arrays(num_columns);
let group_indices: Vec<usize> = (0..BATCH_SIZE).collect();
b.iter(|| {
- let mut ordering =
- GroupOrderingPartial::try_new(order_indices.clone()).unwrap();
- ordering
+ let mut clustering =
+ GroupClusteringPartial::try_new(grouping_indices.clone()).unwrap();
+ clustering
.new_groups(&batch_group_values, &group_indices, BATCH_SIZE)
.unwrap();
});
diff --git a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/ordered_final_table.rs b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/clustered_final_table.rs
similarity index 76%
rename from datafusion/physical-plan/src/aggregates/aggregate_hash_table/ordered_final_table.rs
rename to datafusion/physical-plan/src/aggregates/aggregate_hash_table/clustered_final_table.rs
index 9cf6497e48..e2bad901f0 100644
--- a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/ordered_final_table.rs
+++ b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/clustered_final_table.rs
@@ -15,9 +15,9 @@
// specific language governing permissions and limitations
// under the License.
-//! Aggregate table for final aggregation when partial-state input is ordered.
+//! Aggregate table for final aggregation when partial-state input is clustered.
//!
-//! See comments in [`super::ordered_partial_table`] for details.
+//! See comments in [`super::clustered_partial_table`] for details.
use std::sync::Arc;
@@ -25,12 +25,12 @@ use arrow::datatypes::SchemaRef;
use arrow::record_batch::RecordBatch;
use datafusion_common::Result;
-use crate::InputOrderMode;
use crate::aggregates::aggregate_hash_table::FinalMarker;
+use crate::aggregates::order::GroupClusteringMode;
use crate::aggregates::{AggregateExec, AggregateMode, group_values::AccumulatorPhase};
use super::common::HashAggregateAccumulator;
-use super::common_ordered::{OrderedAggregateTable, OrderedAggregateTableMetrics};
+use super::common_clustered::{ClusteredAggregateTable, ClusteredAggregateTableMetrics};
/// Implementation specific to final aggregation, where the table stores partial
/// aggregate states and the input rows are also partial states.
@@ -40,28 +40,28 @@ use super::common_ordered::{OrderedAggregateTable, OrderedAggregateTableMetrics}
/// - Aggregate table stores: `k, sum(x), count(x)`
/// - Input rows: `k, sum(x), count(x)`
///
-/// See comments at [`OrderedAggregateTable`] for details.
-impl OrderedAggregateTable<FinalMarker> {
- pub(in crate::aggregates) fn new_with_input_order(
+/// See comments at [`ClusteredAggregateTable`] for details.
+impl ClusteredAggregateTable<FinalMarker> {
+ pub(in crate::aggregates) fn new_with_group_clustering(
agg: &AggregateExec,
input_schema: &SchemaRef,
output_schema: SchemaRef,
- input_order_mode: &InputOrderMode,
- metrics: OrderedAggregateTableMetrics,
+ group_clustering_mode: &GroupClusteringMode,
+ metrics: ClusteredAggregateTableMetrics,
) -> Result<Self> {
Self::new_for_mode(
agg,
input_schema,
output_schema,
Arc::clone(input_schema),
- input_order_mode,
+ group_clustering_mode,
&AggregateMode::Final,
vec![None; agg.aggr_expr().len()],
metrics,
)
}
- /// Merges one partial-state input batch and updates ordering information for
+ /// Merges one partial-state input batch and updates completion state for
/// any newly observed groups.
pub(in crate::aggregates) fn aggregate_batch(
&mut self,
@@ -69,7 +69,7 @@ impl OrderedAggregateTable<FinalMarker> {
) -> Result<()> {
let evaluated_batch = self.evaluate_batch(batch)?;
// `PhysicalGroupBy::as_final()` removes grouping sets while planning
- // final aggregation, so final ordered aggregation sees one grouping.
+ // final aggregation, so clustered final aggregation sees one grouping.
debug_assert_eq!(evaluated_batch.grouping_set_args.len(), 1);
self.aggregate_evaluated_batch(
&evaluated_batch,
@@ -78,8 +78,8 @@ impl OrderedAggregateTable<FinalMarker> {
)
}
- /// Materializes final results for all groups proven complete by the input
- /// ordering, leaving the active ordered-key range in the table.
+ /// Materializes final results for all completed groups, leaving
+ /// the active contiguous-key range in the table.
///
/// Returns None if there are no completed groups.
pub(in crate::aggregates) fn take_completed_result_batch(
@@ -88,7 +88,7 @@ impl OrderedAggregateTable<FinalMarker> {
if self.is_empty() {
return Ok(None);
}
- let Some(emit_to) = self.group_ordering().emit_to() else {
+ let Some(emit_to) = self.group_clustering().emit_to() else {
return Ok(None);
};
self.materialize_groups(
diff --git a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/ordered_partial_table.rs b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/clustered_partial_table.rs
similarity index 76%
rename from datafusion/physical-plan/src/aggregates/aggregate_hash_table/ordered_partial_table.rs
rename to datafusion/physical-plan/src/aggregates/aggregate_hash_table/clustered_partial_table.rs
index 3756c4b8e7..fef9659fc2 100644
--- a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/ordered_partial_table.rs
+++ b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/clustered_partial_table.rs
@@ -15,13 +15,13 @@
// specific language governing permissions and limitations
// under the License.
-//! Aggregate table for partial aggregation when input is ordered by group keys.
+//! Aggregate table for partial aggregation when input is clustered by group keys.
//!
-//! See the [`super::common_ordered`] comments for the high-level ideas.
+//! See the [`super::common_clustered`] comments for the high-level ideas.
//!
-//! This operator handles input that is ordered by group keys:
-//! - Fully ordered: `GROUP BY a, b`, input is `ORDER BY a, b`
-//! - Partially ordered: `GROUP BY a, b`, input is `ORDER BY a`
+//! Ordering can establish either group-clustering mode:
+//! - Full: `GROUP BY a, b`, input is `ORDER BY a, b`
+//! - Partial: `GROUP BY a, b`, input is `ORDER BY a`
//!
//! When a group key combination is exhausted, this table eagerly flushes the
//! completed groups to improve memory efficiency.
@@ -41,7 +41,7 @@ use crate::aggregates::{
};
use super::common::HashAggregateAccumulator;
-use super::common_ordered::{OrderedAggregateTable, OrderedAggregateTableMetrics};
+use super::common_clustered::{ClusteredAggregateTable, ClusteredAggregateTableMetrics};
/// Implementation specific to partial aggregation, where the table stores
/// partial aggregate states and the input rows are raw rows.
@@ -51,8 +51,8 @@ use super::common_ordered::{OrderedAggregateTable, OrderedAggregateTableMetrics}
/// - Aggregate table stores: `k, sum(x), count(x)`
/// - Input rows: `k, x`
///
-/// See comments at [`OrderedAggregateTable`] for details.
-impl OrderedAggregateTable<PartialMarker> {
+/// See comments at [`ClusteredAggregateTable`] for details.
+impl ClusteredAggregateTable<PartialMarker> {
pub(in crate::aggregates) fn new(
agg: &AggregateExec,
partition: usize,
@@ -60,20 +60,20 @@ impl OrderedAggregateTable<PartialMarker> {
) -> Result<Self> {
let input_schema = agg.input().schema();
let state_schema = Arc::clone(&output_schema);
- let metrics = OrderedAggregateTableMetrics::new(agg, partition);
+ let metrics = ClusteredAggregateTableMetrics::new(agg, partition);
Self::new_for_mode(
agg,
&input_schema,
output_schema,
state_schema,
- &agg.input_order_mode,
+ &agg.group_clustering_mode,
&AggregateMode::Partial,
agg.filter_expr().to_vec(),
metrics,
)
}
- /// Aggregates one raw input batch and updates ordering information for any
+ /// Aggregates one raw input batch and updates completion state for any
/// newly observed groups.
pub(in crate::aggregates) fn aggregate_batch(
&mut self,
@@ -87,15 +87,15 @@ impl OrderedAggregateTable<PartialMarker> {
)
}
- /// Materializes all groups proven complete by the input ordering, leaving
- /// the active ordered-key range in the table.
+ /// Materializes all completed groups, leaving
+ /// the active contiguous-key range in the table.
pub(in crate::aggregates) fn take_completed_state_batch(
&mut self,
) -> Result<Option<RecordBatch>> {
if self.is_empty() {
return Ok(None);
}
- let Some(emit_to) = self.group_ordering().emit_to() else {
+ let Some(emit_to) = self.group_clustering().emit_to() else {
return Ok(None);
};
self.materialize_groups(
diff --git a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/ordered_single_table.rs b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/clustered_single_table.rs
similarity index 79%
rename from datafusion/physical-plan/src/aggregates/aggregate_hash_table/ordered_single_table.rs
rename to datafusion/physical-plan/src/aggregates/aggregate_hash_table/clustered_single_table.rs
index 89859bc036..c8ecbf4f7f 100644
--- a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/ordered_single_table.rs
+++ b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/clustered_single_table.rs
@@ -15,9 +15,9 @@
// specific language governing permissions and limitations
// under the License.
-//! Aggregate table for single aggregation when raw input is ordered.
+//! Aggregate table for single aggregation when raw input is clustered.
//!
-//! See comments in [`super::ordered_partial_table`] for details.
+//! See comments in [`super::clustered_partial_table`] for details.
use arrow::datatypes::SchemaRef;
use arrow::record_batch::RecordBatch;
@@ -27,7 +27,7 @@ use crate::aggregates::aggregate_hash_table::SingleMarker;
use crate::aggregates::{AggregateExec, AggregateMode, group_values::AccumulatorPhase};
use super::common::HashAggregateAccumulator;
-use super::common_ordered::{OrderedAggregateTable, OrderedAggregateTableMetrics};
+use super::common_clustered::{ClusteredAggregateTable, ClusteredAggregateTableMetrics};
/// Implementation specific to single aggregation, where the table stores final
/// aggregate values and the input rows are raw rows.
@@ -37,8 +37,8 @@ use super::common_ordered::{OrderedAggregateTable, OrderedAggregateTableMetrics}
/// - Aggregate table stores: `k, avg(x)`
/// - Input rows: `k, x`
///
-/// See comments at [`OrderedAggregateTable`] for details.
-impl OrderedAggregateTable<SingleMarker> {
+/// See comments at [`ClusteredAggregateTable`] for details.
+impl ClusteredAggregateTable<SingleMarker> {
pub(in crate::aggregates) fn new(
agg: &AggregateExec,
partition: usize,
@@ -51,20 +51,20 @@ impl OrderedAggregateTable<SingleMarker> {
));
let input_schema = agg.input().schema();
- let metrics = OrderedAggregateTableMetrics::new(agg, partition);
+ let metrics = ClusteredAggregateTableMetrics::new(agg, partition);
Self::new_for_mode(
agg,
&input_schema,
output_schema,
state_schema,
- &agg.input_order_mode,
+ &agg.group_clustering_mode,
&agg.mode,
agg.filter_expr().to_vec(),
metrics,
)
}
- /// Aggregates one raw input batch and updates ordering information for any
+ /// Aggregates one raw input batch and updates completion state for any
/// newly observed groups.
pub(in crate::aggregates) fn aggregate_batch(
&mut self,
@@ -78,8 +78,8 @@ impl OrderedAggregateTable<SingleMarker> {
)
}
- /// Materializes final results for all groups proven complete by the input
- /// ordering, leaving the active ordered-key range in the table.
+ /// Materializes final results for all completed groups, leaving
+ /// the active contiguous-key range in the table.
///
/// Returns None if there are no completed groups.
pub(in crate::aggregates) fn take_completed_result_batch(
@@ -88,7 +88,7 @@ impl OrderedAggregateTable<SingleMarker> {
if self.is_empty() {
return Ok(None);
}
- let Some(emit_to) = self.group_ordering().emit_to() else {
+ let Some(emit_to) = self.group_clustering().emit_to() else {
return Ok(None);
};
self.materialize_groups(
diff --git a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common.rs b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common.rs
index 729932eb04..698a3bddff 100644
--- a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common.rs
+++ b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common.rs
@@ -39,7 +39,7 @@ use crate::aggregates::group_values::{
AccumulatorPhase, AggregateAccumulatorMetrics, AggregateArgumentMetrics,
GroupByMetrics, GroupValues, new_group_values,
};
-use crate::aggregates::order::GroupOrdering;
+use crate::aggregates::order::GroupClustering;
use crate::aggregates::{
AggregateExec, PhysicalGroupBy, aggregate_expressions, evaluate_group_by,
group_id_array, max_duplicate_ordinal,
@@ -181,7 +181,7 @@ impl<AggrMode> AggregateHashTable<AggrMode> {
.collect::<Result<_>>()?;
let group_schema = agg.group_by().group_schema(&input_schema)?;
- let group_values = new_group_values(group_schema, &GroupOrdering::None)?;
+ let group_values = new_group_values(group_schema, &GroupClustering::None)?;
Ok(Self {
group_by_metrics: metrics.group_by,
diff --git a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common_ordered.rs b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common_clustered.rs
similarity index 84%
rename from datafusion/physical-plan/src/aggregates/aggregate_hash_table/common_ordered.rs
rename to datafusion/physical-plan/src/aggregates/aggregate_hash_table/common_clustered.rs
index c25b7485b8..e6c572677b 100644
--- a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common_ordered.rs
+++ b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common_clustered.rs
@@ -15,8 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-//! Common utilities for aggregate tables used in aggregations that inputs are ordered
-//! by the groups.
+//! Common utilities for aggregate tables with input clustered by group keys.
use std::marker::PhantomData;
use std::sync::Arc;
@@ -27,13 +26,12 @@ use datafusion_common::Result;
use datafusion_execution::memory_pool::proxy::VecAllocExt;
use datafusion_expr::{AggregateMetrics, EmitTo};
-use crate::InputOrderMode;
use crate::PhysicalExpr;
use crate::aggregates::group_values::{
AccumulatorPhase, AggregateAccumulatorMetrics, AggregateArgumentMetrics,
GroupByMetrics, GroupValues, new_group_values,
};
-use crate::aggregates::order::GroupOrdering;
+use crate::aggregates::order::{GroupClustering, GroupClusteringMode};
use crate::aggregates::{
AggregateExec, AggregateMode, PhysicalGroupBy, aggregate_expressions,
evaluate_group_by,
@@ -46,14 +44,14 @@ use super::common::{
};
#[derive(Clone)]
-pub(in crate::aggregates) struct OrderedAggregateTableMetrics {
+pub(in crate::aggregates) struct ClusteredAggregateTableMetrics {
pub(super) group_by: GroupByMetrics,
pub(super) aggregate_arguments: AggregateArgumentMetrics,
pub(super) accumulator: Arc<AggregateAccumulatorMetrics>,
pub(super) submetrics: Vec<Arc<dyn AggregateMetrics>>,
}
-impl OrderedAggregateTableMetrics {
+impl ClusteredAggregateTableMetrics {
pub(in crate::aggregates) fn new(agg: &AggregateExec, partition: usize) -> Self {
let metrics = AggregateTableMetrics::new(agg, partition);
Self {
@@ -76,20 +74,20 @@ impl OrderedAggregateTableMetrics {
}
}
-/// Aggregate table shared by the ordered single, partial and final paths.
+/// Aggregate table shared by the clustered single, partial and final paths.
///
-/// # Ordering optimization
+/// # Group clustering optimization
///
-/// The table consumes input batches while `GroupOrdering` tracks which groups
+/// The table consumes input batches while [`GroupClustering`] tracks which groups
/// are proven complete. Completed groups can be emitted before the input stream
-/// ends, which keeps memory bounded by the active ordered key range.
+/// ends, so completed groups no longer occupy the table.
///
/// # Single, partial and final variant difference
///
/// The partial and final aggregate tables implement the two stages of grouped
/// aggregation, while the single aggregate table implements both stages in one
/// table. See
-/// [`OrderedPartialAggregateStream`](crate::aggregates::ordered_partial_stream::OrderedPartialAggregateStream)
+/// [`ClusteredPartialAggregateStream`](crate::aggregates::clustered_partial_stream::ClusteredPartialAggregateStream)
/// for the high-level plan shape.
///
/// Example: `AVG(v) FILTER (WHERE v>0) GROUP BY k`
@@ -111,15 +109,15 @@ impl OrderedAggregateTableMetrics {
///
/// # Marker Type
///
-/// `OrderedAggrMode` selects the aggregate semantics. For example,
-/// `OrderedAggregateTable::<PartialMarker>::new(...)` consumes raw rows
+/// `AggrMode` selects the aggregate semantics. For example,
+/// `ClusteredAggregateTable::<PartialMarker>::new(...)` consumes raw rows
/// and emits partial states, while
-/// `OrderedAggregateTable::<FinalMarker>::new_with_input_order(...)`
+/// `ClusteredAggregateTable::<FinalMarker>::new_with_group_clustering(...)`
/// consumes partial states and emits final values.
///
/// Shared methods live on `impl<T>`; single/partial/final behavior lives on
/// marker-specific impls.
-pub(in crate::aggregates) struct OrderedAggregateTable<OrderedAggrMode> {
+pub(in crate::aggregates) struct ClusteredAggregateTable<AggrMode> {
/// Output schema: group columns followed by aggregate state or final values.
pub(super) output_schema: SchemaRef,
@@ -139,26 +137,26 @@ pub(in crate::aggregates) struct OrderedAggregateTable<OrderedAggrMode> {
/// Optional internal metrics owned by each aggregate expression.
pub(super) aggregate_submetrics: Vec<Arc<dyn AggregateMetrics>>,
- /// Group keys, ordering state, and accumulator states.
- pub(super) buffer: OrderedAggregateTableBuffer,
+ /// Group keys, completion state, and accumulator states.
+ pub(super) buffer: ClusteredAggregateTableBuffer,
- _mode: PhantomData<OrderedAggrMode>,
+ _mode: PhantomData<AggrMode>,
}
-/// Buffer for the ordered aggregate table's group keys and accumulator states.
+/// Buffer for the clustered aggregate table's group keys and accumulator states.
///
-/// It accumulates input during aggregation and emits output rows as soon as the
-/// input ordering proves those groups are complete.
+/// It accumulates input during aggregation and emits output rows as soon as
+/// groups are known to be complete.
///
-/// [`GroupOrdering`] tracks when and how to do early emit.
+/// [`GroupClustering`] tracks when and how to do early emit.
/// [`GroupValues`] stores the physical group-key layout, while
/// [`datafusion_expr::GroupsAccumulator`] stores per-group aggregate state.
-pub(super) struct OrderedAggregateTableBuffer {
+pub(super) struct ClusteredAggregateTableBuffer {
/// GROUP BY expressions evaluated against input batches.
pub(super) group_by: Arc<PhysicalGroupBy>,
- /// Tracks how far ordered input allows this table to drain safely.
- pub(super) group_ordering: GroupOrdering,
+ /// Tracks which groups are complete and can be emitted safely.
+ pub(super) group_clustering: GroupClustering,
/// Interned group keys, in the same group-id order used by accumulators.
pub(super) group_values: Box<dyn GroupValues>,
@@ -174,24 +172,24 @@ pub(super) struct OrderedAggregateTableBuffer {
}
/// Methods shared by all aggregate modes
-impl<AggrMode> OrderedAggregateTable<AggrMode> {
+impl<AggrMode> ClusteredAggregateTable<AggrMode> {
#[expect(
clippy::too_many_arguments,
- reason = "keeps ordered single, partial and final table construction explicit"
+ reason = "keeps clustered single, partial and final table construction explicit"
)]
pub(super) fn new_for_mode(
agg: &AggregateExec,
input_schema: &SchemaRef,
output_schema: SchemaRef,
state_schema: SchemaRef,
- input_order_mode: &InputOrderMode,
+ group_clustering_mode: &GroupClusteringMode,
aggregate_mode: &AggregateMode,
filters: Vec<Option<Arc<dyn PhysicalExpr>>>,
- metrics: OrderedAggregateTableMetrics,
+ metrics: ClusteredAggregateTableMetrics,
) -> Result<Self> {
- let group_ordering = GroupOrdering::try_new(input_order_mode)?;
+ let group_clustering = GroupClustering::try_new(group_clustering_mode)?;
let group_schema = agg.group_by().group_schema(input_schema)?;
- let group_values = new_group_values(group_schema, &group_ordering)?;
+ let group_values = new_group_values(group_schema, &group_clustering)?;
let aggregate_arguments = aggregate_expressions(
agg.aggr_expr(),
aggregate_mode,
@@ -223,9 +221,9 @@ impl<AggrMode> OrderedAggregateTable<AggrMode> {
aggregate_argument_metrics: metrics.aggregate_arguments,
aggregate_accumulator_metrics: metrics.accumulator,
aggregate_submetrics: metrics.submetrics,
- buffer: OrderedAggregateTableBuffer {
+ buffer: ClusteredAggregateTableBuffer {
group_by: Arc::clone(agg.group_by()),
- group_ordering,
+ group_clustering,
group_values,
group_indices: vec![],
accumulators,
@@ -268,15 +266,15 @@ impl<AggrMode> OrderedAggregateTable<AggrMode> {
/// Called after the input stream is exhausted and the last batch has been
/// aggregated.
///
- /// Updates the internal `GroupOrdering` so it can continue emitting until
+ /// Updates the internal [`GroupClustering`] so it can continue emitting until
/// the buffer is empty.
pub(in crate::aggregates) fn input_done(&mut self) {
- self.buffer.group_ordering.input_done();
+ self.buffer.group_clustering.input_done();
}
- /// Returns the ordering state used to decide how memory pressure is handled.
- pub(in crate::aggregates) fn group_ordering(&self) -> &GroupOrdering {
- &self.buffer.group_ordering
+ /// Returns the completion state used to decide how memory pressure is handled.
+ pub(in crate::aggregates) fn group_clustering(&self) -> &GroupClustering {
+ &self.buffer.group_clustering
}
/// Number of groups currently buffered.
@@ -297,12 +295,12 @@ impl<AggrMode> OrderedAggregateTable<AggrMode> {
.map(|acc| acc.size())
.sum::<usize>()
+ self.buffer.group_values.size()
- + self.buffer.group_ordering.size()
+ + self.buffer.group_clustering.size()
+ self.buffer.group_indices.allocated_size()
}
- pub(in crate::aggregates) fn metrics(&self) -> OrderedAggregateTableMetrics {
- OrderedAggregateTableMetrics {
+ pub(in crate::aggregates) fn metrics(&self) -> ClusteredAggregateTableMetrics {
+ ClusteredAggregateTableMetrics {
group_by: self.group_by_metrics.clone(),
aggregate_arguments: self.aggregate_argument_metrics.clone(),
accumulator: Arc::clone(&self.aggregate_accumulator_metrics),
@@ -311,9 +309,9 @@ impl<AggrMode> OrderedAggregateTable<AggrMode> {
}
/// Takes every intermediate aggregate state and resets the table so it can
- /// continue with a new ordered input segment.
+ /// continue with a new clustered input segment.
///
- /// Unlike normal ordered emission, this operation is allowed to take the
+ /// Unlike normal group-completion emission, this operation is allowed to take the
/// active (incomplete) groups. Partial aggregation can pass those states to
/// its final stage, while single and final aggregation sort and spill them
/// before replay.
@@ -346,7 +344,7 @@ impl<AggrMode> OrderedAggregateTable<AggrMode> {
self.buffer.group_values.clear_shrink(0);
self.buffer.group_indices.clear();
self.buffer.group_indices.shrink_to_fit();
- self.buffer.group_ordering.reset();
+ self.buffer.group_clustering.reset();
Ok(Some(batch))
}
@@ -374,7 +372,7 @@ impl<AggrMode> OrderedAggregateTable<AggrMode> {
.intern(group_values, &mut self.buffer.group_indices)?;
let total_num_groups = self.buffer.group_values.len();
if total_num_groups > starting_num_groups {
- self.buffer.group_ordering.new_groups(
+ self.buffer.group_clustering.new_groups(
group_values,
&self.buffer.group_indices,
total_num_groups,
@@ -418,10 +416,10 @@ impl<AggrMode> OrderedAggregateTable<AggrMode> {
let accumulator_metrics = Arc::clone(&self.aggregate_accumulator_metrics);
let output = self.group_by_metrics.time_emitting(|| {
let mut output = self.buffer.group_values.emit(emit_to)?;
- // `EmitTo::All` is only used after `input_done`, when the ordering
+ // `EmitTo::All` is only used after `input_done`, when the completion
// state no longer tracks group indexes.
if let EmitTo::First(n) = emit_to {
- self.buffer.group_ordering.remove_groups(n);
+ self.buffer.group_clustering.remove_groups(n);
}
for (idx, acc) in self.buffer.accumulators.iter_mut().enumerate() {
diff --git a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/mod.rs b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/mod.rs
index 960b7498f1..a356ea5c79 100644
--- a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/mod.rs
+++ b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/mod.rs
@@ -15,12 +15,12 @@
// specific language governing permissions and limitations
// under the License.
+mod clustered_final_table;
+mod clustered_partial_table;
+mod clustered_single_table;
mod common;
-mod common_ordered;
+mod common_clustered;
mod final_table;
-mod ordered_final_table;
-mod ordered_partial_table;
-mod ordered_single_table;
mod partial_reduce_table;
mod partial_table;
mod single_table;
@@ -101,7 +101,9 @@ pub(super) use common::{
AggregateHashTable, FinalMarker, PartialMarker, PartialReduceMarker,
PartialSkipMarker, SingleMarker, create_group_accumulator,
};
-pub(super) use common_ordered::{OrderedAggregateTable, OrderedAggregateTableMetrics};
+pub(super) use common_clustered::{
+ ClusteredAggregateTable, ClusteredAggregateTableMetrics,
+};
#[cfg(test)]
mod tests {
diff --git a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/partial_table.rs b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/partial_table.rs
index 144c8cd2fe..cba1c73c6b 100644
--- a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/partial_table.rs
+++ b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/partial_table.rs
@@ -19,7 +19,7 @@ use std::marker::PhantomData;
use std::sync::Arc;
use crate::aggregates::group_values::{AccumulatorPhase, new_group_values};
-use crate::aggregates::order::GroupOrdering;
+use crate::aggregates::order::GroupClustering;
use crate::aggregates::{AggregateExec, evaluate_group_by};
use arrow::datatypes::SchemaRef;
use arrow::record_batch::RecordBatch;
@@ -77,7 +77,7 @@ impl AggregateHashTable<PartialMarker> {
) -> Result<AggregateHashTable<PartialSkipMarker>> {
let state = self.state.building();
let group_schema = state.group_by.group_schema(&self.input_schema)?;
- let group_values = new_group_values(group_schema, &GroupOrdering::None)?;
+ let group_values = new_group_values(group_schema, &GroupClustering::None)?;
let accumulators = state
.accumulators
.iter()
diff --git a/datafusion/physical-plan/src/aggregates/ordered_final_stream.rs b/datafusion/physical-plan/src/aggregates/clustered_final_stream.rs
similarity index 89%
rename from datafusion/physical-plan/src/aggregates/ordered_final_stream.rs
rename to datafusion/physical-plan/src/aggregates/clustered_final_stream.rs
index daa5a85323..9c71c5ee13 100644
--- a/datafusion/physical-plan/src/aggregates/ordered_final_stream.rs
+++ b/datafusion/physical-plan/src/aggregates/clustered_final_stream.rs
@@ -15,7 +15,8 @@
// specific language governing permissions and limitations
// under the License.
-//! Final aggregate stream for ordered partial-state input.
+//! Final aggregate stream for partial-state input with group-clustering
+//! guarantees.
use std::sync::Arc;
@@ -28,24 +29,25 @@ use futures::stream::StreamExt;
use super::AggregateExec;
use super::aggregate_hash_table::{
- FinalMarker, OrderedAggregateTable, OrderedAggregateTableMetrics,
+ ClusteredAggregateTable, ClusteredAggregateTableMetrics, FinalMarker,
};
+use super::order::GroupClusteringMode;
use super::spill::AggregateSpill;
+use crate::SendableRecordBatchStream;
use crate::aggregates::AggregateMode;
use crate::metrics::{BaselineMetrics, SpillMetrics};
use crate::stream::{ObservedStream, RecordBatchStreamAdapter};
-use crate::{InputOrderMode, SendableRecordBatchStream};
-/// Final aggregate stream for `InputOrderMode::Sorted` and
-/// `InputOrderMode::PartiallySorted`.
+/// Final aggregate stream for [`GroupClusteringMode::Partial`] and
+/// [`GroupClusteringMode::Full`].
///
-/// See comments at [`super::ordered_partial_stream::OrderedPartialAggregateStream`] for details.
+/// See comments at [`super::clustered_partial_stream::ClusteredPartialAggregateStream`] for details.
///
/// # Spilling
///
-/// This section is only for implementation notes, for background, see [`super::ordered_partial_stream::OrderedPartialAggregateStream`]
+/// This section is only for implementation notes, for background, see [`super::clustered_partial_stream::ClusteredPartialAggregateStream`]
///
-/// For partially sorted input, spilling works as follows:
+/// For partial group clustering, spilling works as follows:
///
/// - Reserve the table footprint plus one `u32` sort index per buffered group. The
/// extra index array is used in later sorting before spilling.
@@ -55,13 +57,13 @@ use crate::{InputOrderMode, SendableRecordBatchStream};
/// batch and full index remain live until the run is written.
/// - After input ends, merge the sorted runs and replay them through a fully
/// ordered final aggregate stream.
-pub(crate) struct OrderedFinalAggregateStream {
+pub(crate) struct ClusteredFinalAggregateStream {
reservation: MemoryReservation,
- context: OrderedFinalAggregateContext,
+ context: ClusteredFinalAggregateContext,
stage: ExecutionStage,
}
-/// Execution stages described in [`OrderedFinalAggregateStream::into_stream`].
+/// Execution stages described in [`ClusteredFinalAggregateStream::into_stream`].
enum ExecutionStage {
Aggregating(Aggregating),
Outputting(Outputting),
@@ -70,8 +72,8 @@ enum ExecutionStage {
struct Aggregating {
input: SendableRecordBatchStream,
- table: OrderedAggregateTable<FinalMarker>,
- /// None when temporary files are disabled or all group keys are ordered.
+ table: ClusteredAggregateTable<FinalMarker>,
+ /// None when temporary files are disabled or group clustering is full.
spill_context: Option<Box<AggregateSpill>>,
}
@@ -83,13 +85,13 @@ struct Outputting {
}
/// Immutable execution context shared by aggregation and output emission.
-struct OrderedFinalAggregateContext {
+struct ClusteredFinalAggregateContext {
schema: SchemaRef,
batch_size: usize,
baseline_metrics: BaselineMetrics,
}
-impl OrderedFinalAggregateStream {
+impl ClusteredFinalAggregateStream {
pub fn new(
agg: &AggregateExec,
context: &Arc<TaskContext>,
@@ -99,10 +101,10 @@ impl OrderedFinalAggregateStream {
agg.mode,
AggregateMode::Final | AggregateMode::FinalPartitioned
));
- debug_assert_ne!(agg.input_order_mode, InputOrderMode::Linear);
+ debug_assert_ne!(agg.group_clustering_mode, GroupClusteringMode::None);
let input = agg.input.execute(partition, Arc::clone(context))?;
- Self::new_with_input(agg, context, partition, input, &agg.input_order_mode)
+ Self::new_with_input(agg, context, partition, input, &agg.group_clustering_mode)
}
pub(in crate::aggregates) fn new_with_input(
@@ -110,15 +112,15 @@ impl OrderedFinalAggregateStream {
context: &Arc<TaskContext>,
partition: usize,
input: SendableRecordBatchStream,
- input_order_mode: &InputOrderMode,
+ group_clustering_mode: &GroupClusteringMode,
) -> Result<Self> {
let baseline_metrics = BaselineMetrics::new(&agg.metrics, partition);
- let metrics = OrderedAggregateTableMetrics::new(agg, partition);
+ let metrics = ClusteredAggregateTableMetrics::new(agg, partition);
let spill_metrics = SpillMetrics::new(&agg.metrics, partition);
let reservation =
- MemoryConsumer::new(format!("OrderedFinalAggregateStream[{partition}]"))
- // HACK: Technically, fully ordered aggregate is a non-spillable
- // consumer, since it uses bounded memory. There is a known race
+ MemoryConsumer::new(format!("ClusteredFinalAggregateStream[{partition}]"))
+ // HACK: Full group clustering uses a non-spilling execution
+ // path. There is a known race
// condition bug, and we set it to spillable to let it have larger
// memory budget to suppress the bug.
// Bug issue: https://github.com/apache/datafusion/issues/17334
@@ -129,7 +131,7 @@ impl OrderedFinalAggregateStream {
context,
partition,
input,
- input_order_mode,
+ group_clustering_mode,
baseline_metrics,
metrics,
Some(spill_metrics),
@@ -149,9 +151,9 @@ impl OrderedFinalAggregateStream {
context: &Arc<TaskContext>,
partition: usize,
input: SendableRecordBatchStream,
- input_order_mode: &InputOrderMode,
+ group_clustering_mode: &GroupClusteringMode,
baseline_metrics: BaselineMetrics,
- metrics: OrderedAggregateTableMetrics,
+ metrics: ClusteredAggregateTableMetrics,
spill_metrics: Option<SpillMetrics>,
reservation: MemoryReservation,
) -> Result<Self> {
@@ -159,25 +161,27 @@ impl OrderedFinalAggregateStream {
agg.mode,
AggregateMode::Final | AggregateMode::FinalPartitioned
));
- debug_assert_ne!(*input_order_mode, InputOrderMode::Linear);
+ debug_assert_ne!(*group_clustering_mode, GroupClusteringMode::None);
let schema = Arc::clone(&agg.schema);
let input_schema = input.schema();
let batch_size = context.session_config().batch_size();
- let can_spill = matches!(input_order_mode, InputOrderMode::PartiallySorted(_))
+ let can_spill = matches!(group_clustering_mode, GroupClusteringMode::Partial(_))
&& context.runtime_env().disk_manager.tmp_files_enabled();
let spill_context = if can_spill {
let Some(spill_metrics) = spill_metrics else {
- return internal_err!("Spillable ordered final stream requires metrics");
+ return internal_err!(
+ "Spillable clustered final stream requires metrics"
+ );
};
Some(Box::new(AggregateSpill::try_new(
- "OrderedFinalAggregateSpill",
+ "ClusteredFinalAggregateSpill",
agg,
context,
partition,
batch_size,
- input_order_mode,
+ group_clustering_mode,
&input_schema,
spill_metrics,
)?))
@@ -185,11 +189,11 @@ impl OrderedFinalAggregateStream {
None
};
- let table = OrderedAggregateTable::<FinalMarker>::new_with_input_order(
+ let table = ClusteredAggregateTable::<FinalMarker>::new_with_group_clustering(
agg,
&input_schema,
Arc::clone(&schema),
- input_order_mode,
+ group_clustering_mode,
metrics,
)?;
@@ -198,7 +202,7 @@ impl OrderedFinalAggregateStream {
Ok(Self {
reservation,
- context: OrderedFinalAggregateContext {
+ context: ClusteredFinalAggregateContext {
schema,
batch_size,
baseline_metrics,
@@ -211,9 +215,9 @@ impl OrderedFinalAggregateStream {
})
}
- /// Entry point for the ordered final aggregate execution stages.
+ /// Entry point for the clustered final aggregate execution stages.
///
- /// See [`OrderedFinalAggregateStream`] for high-level ideas.
+ /// See [`ClusteredFinalAggregateStream`] for high-level ideas.
///
/// # Stage transition graph:
///
@@ -248,9 +252,9 @@ impl OrderedFinalAggregateStream {
///
/// ### Incremental output
///
- /// See the [ordered partial aggregate notes] for details.
+ /// See the [clustered partial aggregate notes] for details.
///
- /// [ordered partial aggregate notes]: super::ordered_partial_stream::OrderedPartialAggregateStream::into_stream
+ /// [clustered partial aggregate notes]: super::clustered_partial_stream::ClusteredPartialAggregateStream::into_stream
///
/// ## Transition Edges
///
@@ -259,7 +263,7 @@ impl OrderedFinalAggregateStream {
/// - If memory fits and no groups are complete, continue reading input.
/// - If OOM, spill.
/// 3. Prepare output:
- /// - Before any spill, ordering proves a prefix complete: materialize the
+ /// - Before any spill, a prefix of groups is complete: materialize the
/// entire prefix once, retaining the input and active groups to resume
/// aggregation.
/// - At EOF without spills, materialize all remaining results and prepare
@@ -312,7 +316,7 @@ impl Aggregating {
fn reservation_size(&self) -> usize {
let table_size = self.table.memory_size();
if self.spill_context.is_some() {
- // See `OrderedFinalAggregateStream` for the spill memory estimate.
+ // See `ClusteredFinalAggregateStream` for the spill memory estimate.
table_size
.saturating_add(self.table.num_groups().saturating_mul(size_of::<u32>()))
} else {
@@ -323,7 +327,7 @@ impl Aggregating {
/// Merges partial states until final results are ready or spill replay begins.
async fn handle_stage(
mut self,
- context: &OrderedFinalAggregateContext,
+ context: &ClusteredFinalAggregateContext,
reservation: &MemoryReservation,
) -> Result<Option<ExecutionStage>> {
let elapsed_compute = context.baseline_metrics.elapsed_compute();
@@ -406,7 +410,7 @@ impl Outputting {
/// Emits slices of one materialized batch without touching the aggregate table.
async fn handle_stage(
self,
- context: &OrderedFinalAggregateContext,
+ context: &ClusteredFinalAggregateContext,
reservation: &MemoryReservation,
emitter: &mut TryEmitter<RecordBatch, DataFusionError>,
) -> Result<Option<ExecutionStage>> {
@@ -454,6 +458,7 @@ impl Outputting {
mod tests {
use super::*;
use crate::ExecutionPlan;
+ use crate::aggregates::GroupClusteringMode;
use crate::aggregates::PhysicalGroupBy;
use crate::common::collect;
use crate::test::TestMemoryExec;
@@ -556,8 +561,8 @@ mod tests {
schema,
)?;
assert_eq!(
- aggregate.input_order_mode(),
- &InputOrderMode::PartiallySorted(vec![0])
+ aggregate.group_clustering_mode(),
+ &GroupClusteringMode::Partial(vec![0])
);
let pool: Arc<dyn MemoryPool> = Arc::new(GreedyMemoryPool::new(limit));
@@ -582,12 +587,12 @@ mod tests {
Arc::clone(&partial_schema),
receiver,
));
- let stream = OrderedFinalAggregateStream::new_with_input(
+ let stream = ClusteredFinalAggregateStream::new_with_input(
&aggregate,
&context,
partition,
input,
- aggregate.input_order_mode(),
+ &aggregate.group_clustering_mode,
)?;
Ok((sender, stream.into_stream()))
};
diff --git a/datafusion/physical-plan/src/aggregates/ordered_partial_stream.rs b/datafusion/physical-plan/src/aggregates/clustered_partial_stream.rs
similarity index 81%
rename from datafusion/physical-plan/src/aggregates/ordered_partial_stream.rs
rename to datafusion/physical-plan/src/aggregates/clustered_partial_stream.rs
index 83a25620cd..f7f7e2d903 100644
--- a/datafusion/physical-plan/src/aggregates/ordered_partial_stream.rs
+++ b/datafusion/physical-plan/src/aggregates/clustered_partial_stream.rs
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-//! Partial aggregate stream for ordered group input.
+//! Partial aggregate stream for input with group-clustering guarantees.
use std::sync::Arc;
@@ -27,27 +27,27 @@ use datafusion_execution::{TaskContext, TryEmitter, async_try_stream};
use futures::stream::StreamExt;
use super::AggregateExec;
-use super::aggregate_hash_table::{OrderedAggregateTable, PartialMarker};
+use super::aggregate_hash_table::{ClusteredAggregateTable, PartialMarker};
use crate::aggregates::AggregateMode;
-use crate::aggregates::order::GroupOrdering;
+use crate::aggregates::order::{GroupClustering, GroupClusteringMode};
use crate::metrics::{BaselineMetrics, MetricBuilder, SpillMetrics};
use crate::stream::{ObservedStream, RecordBatchStreamAdapter};
-use crate::{InputOrderMode, SendableRecordBatchStream, metrics};
+use crate::{SendableRecordBatchStream, metrics};
-/// Partial aggregate stream for `InputOrderMode::Sorted` and
-/// `InputOrderMode::PartiallySorted`.
+/// Partial aggregate stream for [`GroupClusteringMode::Partial`] and
+/// [`GroupClusteringMode::Full`].
///
/// # Example
///
/// SELECT k, AVG(v) FROM t GROUP BY k;
///
-/// If the input is ordered by `k`, the aggregate can use ordered partial and
+/// If the input is ordered by `k`, the aggregate can use clustered partial and
/// final stages:
///
/// ## Plan
-/// AggregateExec(stage=final, ordered)
+/// AggregateExec(stage=final, clustered)
/// -- RepartitionExec(hash(k), preserves_order=true)
-/// ---- AggregateExec(stage=partial, ordered)
+/// ---- AggregateExec(stage=partial, clustered)
///
/// ## Partial Stage Behavior
/// Input: raw rows
@@ -59,22 +59,24 @@ use crate::{InputOrderMode, SendableRecordBatchStream, metrics};
/// Output: results for all groups (for example, `AVG(x)` calculated from the
/// state)
///
-/// # Order-based Optimization
+/// # Group Clustering Optimization
///
/// For the aggregation work, the hash aggregation implementation is reused.
///
-/// After each input batch, check whether any groups can be emitted eagerly to
-/// improve memory efficiency. For example, if the last group key seen is
-/// `k = 100`, it is safe to emit all groups with keys less than 100 because the
-/// input is ordered. Materialize that entire completed prefix once, then emit
-/// slices of it before reading more input. This avoids repeatedly removing small
-/// batches of groups and shifting the remaining hash table and accumulator state.
+/// After each input batch, the group-clustering mode determines whether any
+/// groups can be emitted eagerly to improve memory efficiency. For example, if
+/// the input is ordered by `k` and the last group key seen is `k = 100`, all
+/// groups with keys less than 100 are complete. Materialize that entire completed
+/// prefix once, then emit slices of it before reading more input. This avoids
+/// repeatedly removing small batches of groups and shifting the remaining hash
+/// table and accumulator state.
///
/// # Memory Pressure and Spilling
///
-/// ## Fully ordered case
+/// ## Full group clustering
///
-/// If the input is ordered by every group key, for example:
+/// Every complete grouping tuple is contiguous. Ordering by every group key is
+/// one way to establish this mode, for example:
///
/// - Input order: `a, b`
/// - `GROUP BY`: `a, b`
@@ -86,9 +88,10 @@ use crate::{InputOrderMode, SendableRecordBatchStream, metrics};
/// If a memory reservation nevertheless fails, the stream returns the error
/// directly, indicating an unexpected behavior.
///
-/// ## Partially ordered case
+/// ## Partial group clustering
///
-/// If the input is ordered by only a subset of the group keys, for example:
+/// Rows are contiguous for a subset of the group keys. Ordering by that subset
+/// is one way to establish this mode, for example:
///
/// - Input order: `a`
/// - `GROUP BY`: `a, b`
@@ -96,21 +99,21 @@ use crate::{InputOrderMode, SendableRecordBatchStream, metrics};
/// If one `a` value contains many distinct `b` values, the table may accumulate
/// enough groups to exceed the memory limit.
///
-/// - `OrderedPartialAggregateStream`: On reservation failure, it emits all current
+/// - `ClusteredPartialAggregateStream`: On reservation failure, it emits all current
/// intermediate states downstream and resets the table. The final stage can
/// merge repeated `(a, b)` state rows, so no disk spill is required.
-/// - `OrderedFinalAggregateStream`: It cannot emit incomplete final results. On
+/// - `ClusteredFinalAggregateStream`: It cannot emit incomplete final results. On
/// reservation failure, it sorts the current intermediate states by the complete
/// group key and spills them as one run. After the input ends, it spills any
/// remaining states, performs a sort-preserving merge of all runs, and feeds the
-/// merged input into a fully ordered final aggregate stream.
-pub(crate) struct OrderedPartialAggregateStream {
+/// merged input into a fully clustered final aggregate stream.
+pub(crate) struct ClusteredPartialAggregateStream {
reservation: MemoryReservation,
- context: OrderedPartialAggregateContext,
+ context: ClusteredPartialAggregateContext,
stage: ExecutionStage,
}
-/// Execution stages described in [`OrderedPartialAggregateStream::into_stream`].
+/// Execution stages described in [`ClusteredPartialAggregateStream::into_stream`].
enum ExecutionStage {
Aggregating(Aggregating),
Outputting(Outputting),
@@ -118,7 +121,7 @@ enum ExecutionStage {
struct Aggregating {
input: SendableRecordBatchStream,
- table: OrderedAggregateTable<PartialMarker>,
+ table: ClusteredAggregateTable<PartialMarker>,
}
struct Outputting {
@@ -130,21 +133,21 @@ struct Outputting {
}
/// Immutable execution context shared by aggregation and output emission.
-struct OrderedPartialAggregateContext {
+struct ClusteredPartialAggregateContext {
schema: SchemaRef,
batch_size: usize,
baseline_metrics: BaselineMetrics,
reduction_factor: metrics::RatioMetrics,
}
-impl OrderedPartialAggregateStream {
+impl ClusteredPartialAggregateStream {
pub fn new(
agg: &AggregateExec,
context: &Arc<TaskContext>,
partition: usize,
) -> Result<Self> {
debug_assert_eq!(agg.mode, AggregateMode::Partial);
- debug_assert_ne!(agg.input_order_mode, InputOrderMode::Linear);
+ debug_assert_ne!(agg.group_clustering_mode, GroupClusteringMode::None);
let schema = Arc::clone(&agg.schema);
let input = agg.input.execute(partition, Arc::clone(context))?;
@@ -157,16 +160,16 @@ impl OrderedPartialAggregateStream {
.with_type(metrics::MetricType::Summary)
.ratio_metrics("reduction_factor", partition);
- let table = OrderedAggregateTable::<PartialMarker>::new(
+ let table = ClusteredAggregateTable::<PartialMarker>::new(
agg,
partition,
Arc::clone(&schema),
)?;
let reservation =
- MemoryConsumer::new(format!("OrderedPartialAggregateStream[{partition}]"))
+ MemoryConsumer::new(format!("ClusteredPartialAggregateStream[{partition}]"))
.with_can_spill(matches!(
- table.group_ordering(),
- GroupOrdering::Partial(_)
+ table.group_clustering(),
+ GroupClustering::Partial(_)
))
.register(context.memory_pool());
@@ -175,7 +178,7 @@ impl OrderedPartialAggregateStream {
Ok(Self {
reservation,
- context: OrderedPartialAggregateContext {
+ context: ClusteredPartialAggregateContext {
schema,
batch_size,
baseline_metrics,
@@ -185,9 +188,9 @@ impl OrderedPartialAggregateStream {
})
}
- /// Entry point for the ordered partial aggregate execution stages.
+ /// Entry point for the clustered partial aggregate execution stages.
///
- /// See [`OrderedPartialAggregateStream`] for high-level ideas.
+ /// See [`ClusteredPartialAggregateStream`] for high-level ideas.
///
/// # Stage transition graph:
///
@@ -256,10 +259,10 @@ impl OrderedPartialAggregateStream {
/// 2. Aggregate one input batch. If memory fits and no groups are complete,
/// continue reading input.
/// 3. Prepare output:
- /// - Ordering proves a prefix complete: materialize the entire prefix once,
+ /// - A prefix of groups is complete: materialize the entire prefix once,
/// retaining the input and active groups to resume aggregation.
- /// - On memory pressure with partial ordering, materialize all current
- /// states instead, including incomplete groups, and reset the table.
+ /// - On memory pressure with partial group clustering, materialize all
+ /// current states, including incomplete groups, and reset the table.
/// - At EOF, materialize all remaining states and prepare to output.
/// 4. Input was exhausted with no remaining groups, directly end.
/// 5. Yield one slice without materializing the table again. Keep the shared
@@ -299,10 +302,10 @@ impl OrderedPartialAggregateStream {
impl Aggregating {
/// Aggregates raw input and materializes one batch of partial states.
///
- /// See [`OrderedPartialAggregateStream::into_stream`] for stage transitions.
+ /// See [`ClusteredPartialAggregateStream::into_stream`] for stage transitions.
async fn handle_stage(
mut self,
- context: &OrderedPartialAggregateContext,
+ context: &ClusteredPartialAggregateContext,
reservation: &MemoryReservation,
) -> Result<Option<ExecutionStage>> {
let elapsed_compute = context.baseline_metrics.elapsed_compute();
@@ -315,9 +318,9 @@ impl Aggregating {
let output = match reservation.try_resize(self.table.memory_size()) {
Ok(()) => self.table.take_completed_state_batch()?,
Err(oom @ DataFusionError::ResourcesExhausted(_)) => {
- // Partial ordering may have an unbounded active key range.
+ // Partial group clustering may have an unbounded active key range.
// The final stage can merge incomplete states emitted here.
- if matches!(self.table.group_ordering(), GroupOrdering::Full(_)) {
+ if matches!(self.table.group_clustering(), GroupClustering::Full(_)) {
return Err(oom);
}
let Some(batch) = self.table.take_state_batch()? else {
@@ -363,11 +366,11 @@ impl Aggregating {
impl Outputting {
/// Emits slices of one materialized batch without touching the hash table.
///
- /// See [`OrderedPartialAggregateStream::into_stream`] for stage transitions
+ /// See [`ClusteredPartialAggregateStream::into_stream`] for stage transitions
/// and output memory accounting.
async fn handle_stage(
self,
- context: &OrderedPartialAggregateContext,
+ context: &ClusteredPartialAggregateContext,
reservation: &MemoryReservation,
emitter: &mut TryEmitter<RecordBatch, DataFusionError>,
) -> Result<Option<ExecutionStage>> {
diff --git a/datafusion/physical-plan/src/aggregates/ordered_single_stream.rs b/datafusion/physical-plan/src/aggregates/clustered_single_stream.rs
similarity index 77%
rename from datafusion/physical-plan/src/aggregates/ordered_single_stream.rs
rename to datafusion/physical-plan/src/aggregates/clustered_single_stream.rs
index 57744dec57..7095921ac7 100644
--- a/datafusion/physical-plan/src/aggregates/ordered_single_stream.rs
+++ b/datafusion/physical-plan/src/aggregates/clustered_single_stream.rs
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-//! Single-stage aggregate stream for ordered raw input.
+//! Single-stage aggregate stream for raw input with group-clustering guarantees.
use std::ops::ControlFlow;
use std::sync::Arc;
@@ -28,51 +28,52 @@ use datafusion_execution::TaskContext;
use datafusion_execution::memory_pool::{MemoryConsumer, MemoryReservation};
use futures::stream::{Stream, StreamExt};
-use super::aggregate_hash_table::{OrderedAggregateTable, SingleMarker};
+use super::aggregate_hash_table::{ClusteredAggregateTable, SingleMarker};
+use super::order::GroupClusteringMode;
use super::spill::AggregateSpill;
use super::{AggregateExec, create_schema};
use crate::aggregates::AggregateMode;
use crate::metrics::{BaselineMetrics, RecordOutput, SpillMetrics};
use crate::stream::EmptyRecordBatchStream;
-use crate::{InputOrderMode, RecordBatchStream, SendableRecordBatchStream};
+use crate::{RecordBatchStream, SendableRecordBatchStream};
-/// Single aggregate stream for `InputOrderMode::Sorted` and
-/// `InputOrderMode::PartiallySorted`.
+/// Single aggregate stream for [`GroupClusteringMode::Partial`] and
+/// [`GroupClusteringMode::Full`].
///
/// # Example
///
/// SELECT k, AVG(v) FROM t GROUP BY k;
///
-/// If the input is ordered by `k`, and there are existing key partitioning on group
-/// by keys, the single mode aggregation with ordering optimization can be used:
+/// If the input is ordered by `k` and already key-partitioned on the group-by
+/// keys, clustered single aggregation can be used:
///
/// ## Plan
-/// AggregateExec(stage=single, ordered)
+/// AggregateExec(stage=single, clustered)
/// -- DataSourceExec(t)
///
/// ## Single Stage Behavior
/// Input: raw rows
/// Output: final results for all groups (for example, `AVG(x)`)
///
-/// # Order-based Optimization
+/// # Group Clustering Optimization
///
/// For the aggregation work, the hash aggregation implementation is reused.
///
-/// After each input batch, check whether any groups can be emitted eagerly to
-/// improve memory efficiency. For example, if the last group key seen is
-/// `k = 100`, it is safe to emit all groups with keys less than 100 because the
-/// input is ordered. Materialize that entire completed prefix once, then emit
-/// slices of it before reading more input. See
-/// [`OrderedPartialAggregateStream::into_stream`] for why this avoids
+/// After each input batch, the group-clustering mode determines whether any
+/// groups can be emitted eagerly to improve memory efficiency. Materialize
+/// that entire completed prefix once, then emit slices of it before reading
+/// more input. See
+/// [`ClusteredPartialAggregateStream::into_stream`] for why this avoids
/// repeatedly removing small batches of groups from the table.
///
-/// [`OrderedPartialAggregateStream::into_stream`]: super::ordered_partial_stream::OrderedPartialAggregateStream::into_stream
+/// [`ClusteredPartialAggregateStream::into_stream`]: super::clustered_partial_stream::ClusteredPartialAggregateStream::into_stream
///
/// # Memory Pressure and Spilling
///
-/// ## Fully ordered case
+/// ## Full group clustering
///
-/// If the input is ordered by every group key, for example:
+/// Every complete grouping tuple is contiguous. Ordering by every group key is
+/// one way to establish this mode, for example:
///
/// - Input order: `a, b`
/// - `GROUP BY`: `a, b`
@@ -84,9 +85,10 @@ use crate::{InputOrderMode, RecordBatchStream, SendableRecordBatchStream};
/// If a memory reservation nevertheless fails, the stream returns the error
/// directly, indicating an unexpected behavior.
///
-/// ## Partially ordered case
+/// ## Partial group clustering
///
-/// If the input is ordered by only a subset of the group keys, for example:
+/// Rows are contiguous for a subset of the group keys. Ordering by that subset
+/// is one way to establish this mode, for example:
///
/// - Input order: `a`
/// - `GROUP BY`: `a, b`
@@ -97,27 +99,27 @@ use crate::{InputOrderMode, RecordBatchStream, SendableRecordBatchStream};
/// On reservation failure, the stream sorts the current intermediate states by
/// the complete group key and spills them as one run. After the input ends, it
/// spills any remaining states, performs a sort-preserving merge of all runs,
-/// and feeds the merged input into a fully ordered final aggregate stream.
-pub(crate) struct OrderedSingleAggregateStream {
+/// and feeds the merged input into a fully clustered final aggregate stream.
+pub(crate) struct ClusteredSingleAggregateStream {
schema: SchemaRef,
input: SendableRecordBatchStream,
reservation: MemoryReservation,
baseline_metrics: BaselineMetrics,
batch_size: usize,
- state: Option<OrderedSingleAggregateState>,
+ state: Option<ClusteredSingleAggregateState>,
}
/// See comments at `poll_next()` for details.
-enum OrderedSingleAggregateState {
+enum ClusteredSingleAggregateState {
ReadingInput {
- table: OrderedAggregateTable<SingleMarker>,
+ table: ClusteredAggregateTable<SingleMarker>,
/// None if either
/// - Disk Manager doesn't enable temporary file creation
- /// - The group keys are fully ordered, it's expected to use bounded memory
+ /// - Full group clustering is used, so completed groups can be released
spill_context: Option<Box<AggregateSpill>>,
},
Spilling {
- table: OrderedAggregateTable<SingleMarker>,
+ table: ClusteredAggregateTable<SingleMarker>,
spill_context: Box<AggregateSpill>,
},
/// Emits one materialized batch in `batch_size` slices, then continues
@@ -126,10 +128,10 @@ enum OrderedSingleAggregateState {
batch: RecordBatch,
/// Reserved memory of `batch`, released when handing off the last slice.
batch_memory: usize,
- next_state: Box<OrderedSingleAggregateState>,
+ next_state: Box<ClusteredSingleAggregateState>,
},
PreparingMergeInput {
- table: OrderedAggregateTable<SingleMarker>,
+ table: ClusteredAggregateTable<SingleMarker>,
spill_context: Box<AggregateSpill>,
},
MergingSpills {
@@ -142,13 +144,13 @@ enum OrderedSingleAggregateState {
Error,
}
-type OrderedSingleAggregatePoll = Poll<Option<Result<RecordBatch>>>;
-type OrderedSingleAggregateStateTransition = ControlFlow<
- (OrderedSingleAggregatePoll, OrderedSingleAggregateState),
- OrderedSingleAggregateState,
+type ClusteredSingleAggregatePoll = Poll<Option<Result<RecordBatch>>>;
+type ClusteredSingleAggregateStateTransition = ControlFlow<
+ (ClusteredSingleAggregatePoll, ClusteredSingleAggregateState),
+ ClusteredSingleAggregateState,
>;
-impl OrderedSingleAggregateStream {
+impl ClusteredSingleAggregateStream {
pub fn new(
agg: &AggregateExec,
context: &Arc<TaskContext>,
@@ -158,7 +160,7 @@ impl OrderedSingleAggregateStream {
agg.mode,
AggregateMode::Single | AggregateMode::SinglePartitioned
));
- debug_assert_ne!(agg.input_order_mode, InputOrderMode::Linear);
+ debug_assert_ne!(agg.group_clustering_mode, GroupClusteringMode::None);
let schema = Arc::clone(&agg.schema);
let input = agg.input.execute(partition, Arc::clone(context))?;
@@ -173,7 +175,7 @@ impl OrderedSingleAggregateStream {
AggregateMode::Partial,
)?);
- let table = OrderedAggregateTable::<SingleMarker>::new(
+ let table = ClusteredAggregateTable::<SingleMarker>::new(
agg,
partition,
Arc::clone(&schema),
@@ -181,16 +183,16 @@ impl OrderedSingleAggregateStream {
)?;
let can_spill =
- matches!(agg.input_order_mode, InputOrderMode::PartiallySorted(_))
+ matches!(agg.group_clustering_mode, GroupClusteringMode::Partial(_))
&& context.runtime_env().disk_manager.tmp_files_enabled();
let spill_context = if can_spill {
Some(Box::new(AggregateSpill::try_new(
- "OrderedSingleAggregateSpill",
+ "ClusteredSingleAggregateSpill",
agg,
context,
partition,
batch_size,
- &agg.input_order_mode,
+ &agg.group_clustering_mode,
&state_schema,
spill_metrics,
)?))
@@ -199,7 +201,7 @@ impl OrderedSingleAggregateStream {
};
let reservation =
- MemoryConsumer::new(format!("OrderedSingleAggregateStream[{partition}]"))
+ MemoryConsumer::new(format!("ClusteredSingleAggregateStream[{partition}]"))
.with_can_spill(can_spill)
.register(context.memory_pool());
@@ -212,7 +214,7 @@ impl OrderedSingleAggregateStream {
reservation,
baseline_metrics,
batch_size,
- state: Some(OrderedSingleAggregateState::ReadingInput {
+ state: Some(ClusteredSingleAggregateState::ReadingInput {
table,
spill_context,
}),
@@ -224,25 +226,25 @@ impl OrderedSingleAggregateStream {
self.input = Box::pin(EmptyRecordBatchStream::new(input_schema));
}
- fn break_with_err(error: DataFusionError) -> OrderedSingleAggregateStateTransition {
+ fn break_with_err(error: DataFusionError) -> ClusteredSingleAggregateStateTransition {
ControlFlow::Break((
Poll::Ready(Some(Err(error))),
- OrderedSingleAggregateState::Error,
+ ClusteredSingleAggregateState::Error,
))
}
- fn break_with_internal_err(message: &str) -> OrderedSingleAggregateStateTransition {
+ fn break_with_internal_err(message: &str) -> ClusteredSingleAggregateStateTransition {
Self::break_with_err(internal_datafusion_err!("{message}"))
}
/// Reserve memory for the current aggregate table.
fn reservation_size_for_table(
- table: &OrderedAggregateTable<SingleMarker>,
+ table: &ClusteredAggregateTable<SingleMarker>,
spill_context: Option<&AggregateSpill>,
) -> usize {
let table_size = table.memory_size();
if spill_context.is_some() {
- // See `OrderedSingleAggregateStream` comments for how is it estimated
+ // See `ClusteredSingleAggregateStream` comments for how is it estimated
table_size.saturating_add(table.num_groups().saturating_mul(size_of::<u32>()))
} else {
table_size
@@ -256,11 +258,11 @@ impl OrderedSingleAggregateStream {
&mut self,
batch: RecordBatch,
table_memory: usize,
- next_state: OrderedSingleAggregateState,
- ) -> OrderedSingleAggregateStateTransition {
+ next_state: ClusteredSingleAggregateState,
+ ) -> ClusteredSingleAggregateStateTransition {
let batch_memory = batch.get_array_memory_size();
match self.reservation.try_resize(table_memory + batch_memory) {
- Ok(()) => ControlFlow::Continue(OrderedSingleAggregateState::Outputting {
+ Ok(()) => ControlFlow::Continue(ClusteredSingleAggregateState::Outputting {
batch,
batch_memory,
next_state: Box::new(next_state),
@@ -279,8 +281,8 @@ impl OrderedSingleAggregateStream {
}
}
- /// Consumes one ordered raw input batch, then materializes all finalized
- /// groups if the ordering proves any group is ready.
+ /// Consumes one clustered raw input batch, then materializes all finalized
+ /// groups if any are complete.
///
/// See comments at `poll_next()` for details.
///
@@ -288,22 +290,22 @@ impl OrderedSingleAggregateStream {
fn handle_reading_input(
&mut self,
cx: &mut Context<'_>,
- original_state: OrderedSingleAggregateState,
- ) -> OrderedSingleAggregateStateTransition {
- let OrderedSingleAggregateState::ReadingInput {
+ original_state: ClusteredSingleAggregateState,
+ ) -> ClusteredSingleAggregateStateTransition {
+ let ClusteredSingleAggregateState::ReadingInput {
mut table,
spill_context,
} = original_state
else {
return Self::break_with_internal_err(
- "Ordered single aggregate stream expected ReadingInput state",
+ "Clustered single aggregate stream expected ReadingInput state",
);
};
match self.input.poll_next_unpin(cx) {
Poll::Pending => ControlFlow::Break((
Poll::Pending,
- OrderedSingleAggregateState::ReadingInput {
+ ClusteredSingleAggregateState::ReadingInput {
table,
spill_context,
},
@@ -332,16 +334,16 @@ impl OrderedSingleAggregateStream {
Err(e @ DataFusionError::ResourcesExhausted(_)) => {
let Some(spill_context) = spill_context else {
// `None` means spilling is not supported, see comments
- // at `OrderedSingleAggregateState` for details.
+ // at `ClusteredSingleAggregateState` for details.
return Self::break_with_err(e);
};
if table.is_empty() {
return Self::break_with_internal_err(
- "Ordered single aggregate ran out of memory with no aggregated groups",
+ "Clustered single aggregate ran out of memory with no aggregated groups",
);
}
return ControlFlow::Continue(
- OrderedSingleAggregateState::Spilling {
+ ClusteredSingleAggregateState::Spilling {
table,
spill_context,
},
@@ -377,19 +379,19 @@ impl OrderedSingleAggregateStream {
self.start_outputting(
batch,
table_memory,
- OrderedSingleAggregateState::ReadingInput {
+ ClusteredSingleAggregateState::ReadingInput {
table,
spill_context,
},
)
}
// Can't do early emit, continue aggregating.
- Ok(None) => {
- ControlFlow::Continue(OrderedSingleAggregateState::ReadingInput {
+ Ok(None) => ControlFlow::Continue(
+ ClusteredSingleAggregateState::ReadingInput {
table,
spill_context,
- })
- }
+ },
+ ),
Err(e) => Self::break_with_err(e),
}
}
@@ -399,7 +401,7 @@ impl OrderedSingleAggregateStream {
match spill_context {
Some(spill_context) if spill_context.has_spills() => {
ControlFlow::Continue(
- OrderedSingleAggregateState::PreparingMergeInput {
+ ClusteredSingleAggregateState::PreparingMergeInput {
table,
spill_context,
},
@@ -417,10 +419,10 @@ impl OrderedSingleAggregateStream {
Ok(Some(batch)) => self.start_outputting(
batch,
0,
- OrderedSingleAggregateState::Done,
+ ClusteredSingleAggregateState::Done,
),
Ok(None) => {
- ControlFlow::Continue(OrderedSingleAggregateState::Done)
+ ControlFlow::Continue(ClusteredSingleAggregateState::Done)
}
Err(e) => Self::break_with_err(e),
}
@@ -437,22 +439,22 @@ impl OrderedSingleAggregateStream {
/// Returns the next operator state with control flow decision.
fn handle_spilling(
&mut self,
- original_state: OrderedSingleAggregateState,
- ) -> OrderedSingleAggregateStateTransition {
- let OrderedSingleAggregateState::Spilling {
+ original_state: ClusteredSingleAggregateState,
+ ) -> ClusteredSingleAggregateStateTransition {
+ let ClusteredSingleAggregateState::Spilling {
mut table,
mut spill_context,
} = original_state
else {
return Self::break_with_internal_err(
- "Ordered single aggregate stream expected Spilling state",
+ "Clustered single aggregate stream expected Spilling state",
);
};
// Sanity check: it's impossible to OOM when the table is empty
if table.is_empty() {
return Self::break_with_internal_err(
- "Ordered single aggregation entered Spilling with an empty table",
+ "Clustered single aggregation entered Spilling with an empty table",
);
}
@@ -472,10 +474,12 @@ impl OrderedSingleAggregateStream {
match result {
// Finished spilling the aggregate table, continue aggregating from input
- Ok(()) => ControlFlow::Continue(OrderedSingleAggregateState::ReadingInput {
- table,
- spill_context: Some(spill_context),
- }),
+ Ok(()) => {
+ ControlFlow::Continue(ClusteredSingleAggregateState::ReadingInput {
+ table,
+ spill_context: Some(spill_context),
+ })
+ }
Err(e) => Self::break_with_err(e),
}
}
@@ -483,7 +487,7 @@ impl OrderedSingleAggregateStream {
/// 1. Spills the last in-memory run.
/// 2. Constructs a globally ordered input stream by applying a sort-preserving
/// merge to all spills.
- /// 3. Constructs a replay stream: an ordered aggregate stream over the fully
+ /// 3. Constructs a replay stream: a clustered aggregate stream over the fully
/// ordered input constructed from the spills.
///
/// See comments at `poll_next()` for details.
@@ -491,15 +495,15 @@ impl OrderedSingleAggregateStream {
/// Returns the next operator state with control flow decision.
fn handle_preparing_merge_input(
&mut self,
- original_state: OrderedSingleAggregateState,
- ) -> OrderedSingleAggregateStateTransition {
- let OrderedSingleAggregateState::PreparingMergeInput {
+ original_state: ClusteredSingleAggregateState,
+ ) -> ClusteredSingleAggregateStateTransition {
+ let ClusteredSingleAggregateState::PreparingMergeInput {
mut table,
mut spill_context,
} = original_state
else {
return Self::break_with_internal_err(
- "Ordered single aggregate stream expected PreparingMergeInput state",
+ "Clustered single aggregate stream expected PreparingMergeInput state",
);
};
@@ -527,7 +531,7 @@ impl OrderedSingleAggregateStream {
match replay {
Ok(stream) => {
- ControlFlow::Continue(OrderedSingleAggregateState::MergingSpills {
+ ControlFlow::Continue(ClusteredSingleAggregateState::MergingSpills {
stream,
})
}
@@ -544,26 +548,28 @@ impl OrderedSingleAggregateStream {
fn handle_merging_spills(
&mut self,
cx: &mut Context<'_>,
- original_state: OrderedSingleAggregateState,
- ) -> OrderedSingleAggregateStateTransition {
- let OrderedSingleAggregateState::MergingSpills { mut stream } = original_state
+ original_state: ClusteredSingleAggregateState,
+ ) -> ClusteredSingleAggregateStateTransition {
+ let ClusteredSingleAggregateState::MergingSpills { mut stream } = original_state
else {
return Self::break_with_internal_err(
- "Ordered single aggregate stream expected MergingSpills state",
+ "Clustered single aggregate stream expected MergingSpills state",
);
};
match stream.poll_next_unpin(cx) {
Poll::Pending => ControlFlow::Break((
Poll::Pending,
- OrderedSingleAggregateState::MergingSpills { stream },
+ ClusteredSingleAggregateState::MergingSpills { stream },
)),
Poll::Ready(Some(Ok(batch))) => ControlFlow::Break((
Poll::Ready(Some(Ok(batch))),
- OrderedSingleAggregateState::MergingSpills { stream },
+ ClusteredSingleAggregateState::MergingSpills { stream },
)),
Poll::Ready(Some(Err(e))) => Self::break_with_err(e),
- Poll::Ready(None) => ControlFlow::Continue(OrderedSingleAggregateState::Done),
+ Poll::Ready(None) => {
+ ControlFlow::Continue(ClusteredSingleAggregateState::Done)
+ }
}
}
@@ -574,16 +580,16 @@ impl OrderedSingleAggregateStream {
/// Returns the next operator state with control flow decision.
fn handle_outputting(
&mut self,
- original_state: OrderedSingleAggregateState,
- ) -> OrderedSingleAggregateStateTransition {
- let OrderedSingleAggregateState::Outputting {
+ original_state: ClusteredSingleAggregateState,
+ ) -> ClusteredSingleAggregateStateTransition {
+ let ClusteredSingleAggregateState::Outputting {
batch,
batch_memory,
next_state,
} = original_state
else {
return Self::break_with_internal_err(
- "Ordered single aggregate stream expected Outputting state",
+ "Clustered single aggregate stream expected Outputting state",
);
};
@@ -592,7 +598,7 @@ impl OrderedSingleAggregateStream {
let batch = batch.slice(self.batch_size, batch.num_rows() - self.batch_size);
return ControlFlow::Break((
Poll::Ready(Some(Ok(output.record_output(&self.baseline_metrics)))),
- OrderedSingleAggregateState::Outputting {
+ ClusteredSingleAggregateState::Outputting {
batch,
batch_memory,
next_state,
@@ -611,20 +617,20 @@ impl OrderedSingleAggregateStream {
}
}
-impl Stream for OrderedSingleAggregateStream {
+impl Stream for ClusteredSingleAggregateStream {
type Item = Result<RecordBatch>;
- /// Entry point for the ordered single aggregate state machine.
+ /// Entry point for the clustered single aggregate state machine.
///
- /// See comments in [`OrderedSingleAggregateStream`] for high-level ideas.
+ /// See comments in [`ClusteredSingleAggregateStream`] for high-level ideas.
///
/// State transition graph:
///
/// ```text
/// (start)
/// -> ReadingInput
- /// The stream starts by polling ordered raw input and updating the
- /// ordered single aggregate table.
+ /// The stream starts by polling clustered raw input and updating the
+ /// clustered single aggregate table.
///
/// ReadingInput
/// -> ReadingInput
@@ -634,7 +640,7 @@ impl Stream for OrderedSingleAggregateStream {
/// The table cannot reserve enough memory. Move all current states into
/// one fully group-key-sorted spill run.
/// -> Outputting
- /// Either the input ordering proves some groups complete, or input was
+ /// Either some groups are complete, or input was
/// exhausted without spilling and every remaining group is complete.
/// Materialize all of them once into one batch. If the batch cannot be
/// reserved, yield it whole and go to the state after `Outputting`.
@@ -689,31 +695,31 @@ impl Stream for OrderedSingleAggregateStream {
let cur_state = self
.state
.take()
- .expect("OrderedSingleAggregateStream state should not be None");
+ .expect("ClusteredSingleAggregateStream state should not be None");
let next_state = match cur_state {
- state @ OrderedSingleAggregateState::ReadingInput { .. } => {
+ state @ ClusteredSingleAggregateState::ReadingInput { .. } => {
self.handle_reading_input(cx, state)
}
- state @ OrderedSingleAggregateState::Spilling { .. } => {
+ state @ ClusteredSingleAggregateState::Spilling { .. } => {
self.handle_spilling(state)
}
- state @ OrderedSingleAggregateState::PreparingMergeInput { .. } => {
+ state @ ClusteredSingleAggregateState::PreparingMergeInput { .. } => {
self.handle_preparing_merge_input(state)
}
- state @ OrderedSingleAggregateState::MergingSpills { .. } => {
+ state @ ClusteredSingleAggregateState::MergingSpills { .. } => {
self.handle_merging_spills(cx, state)
}
- state @ OrderedSingleAggregateState::Outputting { .. } => {
+ state @ ClusteredSingleAggregateState::Outputting { .. } => {
self.handle_outputting(state)
}
- state @ OrderedSingleAggregateState::Error => {
+ state @ ClusteredSingleAggregateState::Error => {
self.close_input();
self.reservation.free();
self.state = Some(state);
return Poll::Ready(None);
}
- state @ OrderedSingleAggregateState::Done => {
+ state @ ClusteredSingleAggregateState::Done => {
let _ = self.reservation.try_resize(0);
self.state = Some(state);
return Poll::Ready(None);
@@ -727,14 +733,14 @@ impl Stream for OrderedSingleAggregateStream {
ControlFlow::Break((Poll::Ready(Some(Err(e))), next_state)) => {
debug_assert!(matches!(
next_state,
- OrderedSingleAggregateState::Error
+ ClusteredSingleAggregateState::Error
));
// The handler has already discarded its state-owned resources.
// Release the remaining stream-owned resources before returning.
self.close_input();
self.reservation.free();
- self.state = Some(OrderedSingleAggregateState::Error);
+ self.state = Some(ClusteredSingleAggregateState::Error);
return Poll::Ready(Some(Err(e)));
}
ControlFlow::Break((poll, next_state)) => {
@@ -746,7 +752,7 @@ impl Stream for OrderedSingleAggregateStream {
}
}
-impl RecordBatchStream for OrderedSingleAggregateStream {
+impl RecordBatchStream for ClusteredSingleAggregateStream {
fn schema(&self) -> SchemaRef {
Arc::clone(&self.schema)
}
diff --git a/datafusion/physical-plan/src/aggregates/group_values/mod.rs b/datafusion/physical-plan/src/aggregates/group_values/mod.rs
index 19d3b94f8a..989b8b928b 100644
--- a/datafusion/physical-plan/src/aggregates/group_values/mod.rs
+++ b/datafusion/physical-plan/src/aggregates/group_values/mod.rs
@@ -29,7 +29,7 @@ mod row;
pub use row::GroupValuesRows;
mod single_group_by;
use datafusion_physical_expr::binary_map::OutputType;
-use multi_group_by::{GroupValuesColumn, GroupValuesOrdered};
+use multi_group_by::{GroupValuesClustered, GroupValuesColumn};
pub(crate) use single_group_by::primitive::{GroupValuesPrimitive, HashValue};
@@ -38,7 +38,7 @@ use crate::aggregates::{
boolean::GroupValuesBoolean, bytes::GroupValuesBytes,
bytes_view::GroupValuesBytesView,
},
- order::GroupOrdering,
+ order::GroupClustering,
};
mod metrics;
@@ -139,7 +139,7 @@ pub trait GroupValues: Send {
///
/// [`GroupValues`] implementations choosing logic:
///
-/// - Fully ordered multi-column keys with supported scalar types use adjacent
+/// - Fully clustered multi-column keys with supported scalar types use adjacent
/// comparisons and column builders, without a hash table.
///
/// - If group by single column, and type of this column has
@@ -156,12 +156,12 @@ pub trait GroupValues: Send {
/// `GroupValuesRows`: crate::aggregates::group_values::GroupValuesRows
pub fn new_group_values(
schema: SchemaRef,
- group_ordering: &GroupOrdering,
+ group_clustering: &GroupClustering,
) -> Result<Box<dyn GroupValues>> {
- if matches!(group_ordering, GroupOrdering::Full(_))
- && GroupValuesOrdered::supports_schema(&schema)
+ if matches!(group_clustering, GroupClustering::Full(_))
+ && GroupValuesClustered::supports_schema(&schema)
{
- return Ok(Box::new(GroupValuesOrdered::try_new(schema)?));
+ return Ok(Box::new(GroupValuesClustered::try_new(schema)?));
}
if schema.fields.len() == 1 {
let d = schema.fields[0].data_type();
@@ -200,7 +200,7 @@ pub fn new_group_values(
}
if multi_group_by::supported_schema(schema.as_ref()) {
- if matches!(group_ordering, GroupOrdering::None) {
+ if matches!(group_clustering, GroupClustering::None) {
Ok(Box::new(GroupValuesColumn::<false>::try_new(schema)?))
} else {
Ok(Box::new(GroupValuesColumn::<true>::try_new(schema)?))
@@ -221,7 +221,7 @@ mod tests {
use datafusion_expr::{EmitTo, GroupSelection};
use super::new_group_values;
- use crate::aggregates::order::GroupOrdering;
+ use crate::aggregates::order::GroupClustering;
#[test]
fn preserving_values_keep_group_indices_valid() {
@@ -230,7 +230,7 @@ mod tests {
DataType::Int32,
true,
)]));
- let mut group_values = new_group_values(schema, &GroupOrdering::None).unwrap();
+ let mut group_values = new_group_values(schema, &GroupClustering::None).unwrap();
assert!(group_values.supports_values_preserving());
let input = Arc::new(Int32Array::from(vec![
@@ -287,7 +287,7 @@ mod tests {
Field::new("primitive", DataType::Int32, false),
Field::new("boolean", DataType::Boolean, false),
]));
- let mut group_values = new_group_values(schema, &GroupOrdering::None).unwrap();
+ let mut group_values = new_group_values(schema, &GroupClustering::None).unwrap();
let input = vec![
Arc::new(Int32Array::from(vec![10, 20, 10])) as ArrayRef,
Arc::new(BooleanArray::from(vec![true, false, true])) as ArrayRef,
@@ -318,7 +318,7 @@ mod tests {
true,
)]));
let mut group_values =
- new_group_values(schema, &GroupOrdering::None).unwrap();
+ new_group_values(schema, &GroupClustering::None).unwrap();
let input: ArrayRef = match data_type {
DataType::Utf8 => Arc::new(StringArray::from(vec![
Some("a"),
diff --git a/datafusion/physical-plan/src/aggregates/group_values/multi_group_by/ordered.rs b/datafusion/physical-plan/src/aggregates/group_values/multi_group_by/clustered.rs
similarity index 87%
rename from datafusion/physical-plan/src/aggregates/group_values/multi_group_by/ordered.rs
rename to datafusion/physical-plan/src/aggregates/group_values/multi_group_by/clustered.rs
index 9e34ac7aea..1fcd1fa0bf 100644
--- a/datafusion/physical-plan/src/aggregates/group_values/multi_group_by/ordered.rs
+++ b/datafusion/physical-plan/src/aggregates/group_values/multi_group_by/clustered.rs
@@ -27,7 +27,7 @@ use datafusion_expr::{EmitTo, GroupSelection};
use super::{GroupColumn, GroupValuesColumn};
use crate::aggregates::group_values::GroupValues;
-/// Columnar group keys for input fully ordered by all grouping expressions.
+/// Columnar group keys for input clustered by the complete grouping tuple.
///
/// Equal keys must be contiguous across input batches. Within a batch Arrow's
/// partition kernel finds those runs. Only the first run can continue the last
@@ -36,14 +36,15 @@ use crate::aggregates::group_values::GroupValues;
///
/// This removes hashing and hash-table storage, but does not change the dense
/// group-id or emission contracts. In particular, `First(n)` shifts the remaining
-/// keys and their ids together. Partially ordered inputs must use a hash table.
-pub(crate) struct GroupValuesOrdered {
+/// keys and their ids together. Inputs clustered by only a subset of the keys
+/// must use a hash table.
+pub(crate) struct GroupValuesClustered {
schema: SchemaRef,
columns: Vec<Box<dyn GroupColumn>>,
new_groups: Vec<usize>,
}
-impl GroupValuesOrdered {
+impl GroupValuesClustered {
/// Types for which adjacent Arrow equality agrees with GROUP BY equality.
/// Keep single-column specializations and floats/nested/encoded keys on the
/// established path until their semantics and performance are validated.
@@ -74,7 +75,7 @@ impl GroupValuesOrdered {
pub(crate) fn try_new(schema: SchemaRef) -> Result<Self> {
if !Self::supports_schema(&schema) {
return not_impl_err!(
- "Unsupported schema for fully ordered group values: {schema}"
+ "Unsupported schema for fully clustered group values: {schema}"
);
}
let columns = GroupValuesColumn::<true>::build_group_columns(&schema)?;
@@ -86,7 +87,7 @@ impl GroupValuesOrdered {
}
}
-impl GroupValues for GroupValuesOrdered {
+impl GroupValues for GroupValuesClustered {
fn intern(&mut self, cols: &[ArrayRef], groups: &mut Vec<usize>) -> Result<()> {
groups.clear();
let ranges = partition(cols)?.ranges();
@@ -170,7 +171,7 @@ impl GroupValues for GroupValuesOrdered {
fn clear_shrink(&mut self, num_rows: usize) {
self.columns = GroupValuesColumn::<true>::build_group_columns(&self.schema)
- .expect("schema validated by GroupValuesOrdered::try_new");
+ .expect("schema validated by GroupValuesClustered::try_new");
self.new_groups.clear();
self.new_groups.shrink_to(num_rows);
}
@@ -191,8 +192,12 @@ mod tests {
// Compare with the existing hash implementation, changing actual input
// boundaries and removing completed groups between batches. This checks the
// semantic contract rather than duplicating the adjacent-run algorithm.
- #[test]
- fn ordered_keys_match_hash_grouping_across_batches_and_emits() -> Result<()> {
+ #[rstest::rstest]
+ #[case::sorted(false)]
+ #[case::unsorted(true)]
+ fn clustered_keys_match_hash_grouping_across_batches_and_emits(
+ #[case] unsorted: bool,
+ ) -> Result<()> {
for key_type in [
DataType::Boolean,
DataType::Int8,
@@ -213,7 +218,9 @@ mod tests {
let rows = 1025;
let first: ArrayRef = Arc::new(Int32Array::from_iter((0..rows).map(|row| {
let group = row / 7;
- (group >= 4).then_some(group / 4)
+ let key = group / 4;
+ // Swap neighboring key values while keeping each run contiguous.
+ (group >= 4).then_some(if unsorted { key ^ 1 } else { key })
})));
let second: ArrayRef = match key_type {
DataType::Boolean => {
@@ -262,8 +269,8 @@ mod tests {
]));
for batch_size in [1, 2, 7, 8, 63, 1024] {
for emit_limit in [0, 1, 17, usize::MAX] {
- let mut ordered =
- GroupValuesOrdered::try_new(Arc::clone(&schema))?;
+ let mut clustered =
+ GroupValuesClustered::try_new(Arc::clone(&schema))?;
let mut hashed =
GroupValuesColumn::<true>::try_new(Arc::clone(&schema))?;
let mut actual = Vec::new();
@@ -274,16 +281,16 @@ mod tests {
.iter()
.map(|array| array.slice(offset, length))
.collect::<Vec<_>>();
- ordered.intern(&batch, &mut actual)?;
+ clustered.intern(&batch, &mut actual)?;
hashed.intern(&batch, &mut expected)?;
assert_eq!(
actual, expected,
"{key_type:?}, batch={batch_size}, offset={offset}, descending={descending}"
);
- assert_eq!(ordered.len(), hashed.len());
- let selection = GroupSelection::all(ordered.len());
+ assert_eq!(clustered.len(), hashed.len());
+ let selection = GroupSelection::all(clustered.len());
assert_eq!(
- ordered.values_preserving(selection)?,
+ clustered.values_preserving(selection)?,
hashed.values_preserving(selection)?
);
// A zero-row batch must neither forget the boundary
@@ -292,23 +299,29 @@ mod tests {
.iter()
.map(|array| array.slice(0, 0))
.collect::<Vec<_>>();
- ordered.intern(&empty, &mut actual)?;
+ clustered.intern(&empty, &mut actual)?;
assert!(actual.is_empty());
- let emit = emit_limit.min(ordered.len().saturating_sub(1));
+ let emit = emit_limit.min(clustered.len().saturating_sub(1));
assert_eq!(
- ordered.emit(EmitTo::First(emit))?,
+ clustered.emit(EmitTo::First(emit))?,
hashed.emit(EmitTo::First(emit))?
);
}
- assert_eq!(ordered.emit(EmitTo::All)?, hashed.emit(EmitTo::All)?);
- assert!(ordered.is_empty());
+ assert_eq!(
+ clustered.emit(EmitTo::All)?,
+ hashed.emit(EmitTo::All)?
+ );
+ assert!(clustered.is_empty());
// All and clear_shrink reset the cross-batch state.
- ordered.clear_shrink(0);
+ clustered.clear_shrink(0);
hashed.clear_shrink(0);
- ordered.intern(&input, &mut actual)?;
+ clustered.intern(&input, &mut actual)?;
hashed.intern(&input, &mut expected)?;
assert_eq!(actual, expected);
- assert_eq!(ordered.emit(EmitTo::All)?, hashed.emit(EmitTo::All)?);
+ assert_eq!(
+ clustered.emit(EmitTo::All)?,
+ hashed.emit(EmitTo::All)?
+ );
}
}
}
@@ -317,7 +330,7 @@ mod tests {
}
#[test]
- fn ordered_schema_gate_keeps_unvalidated_types_on_existing_paths() {
+ fn clustered_schema_gate_keeps_unvalidated_types_on_existing_paths() {
for data_type in [
DataType::Float32,
DataType::Float64,
@@ -329,18 +342,20 @@ mod tests {
Field::new("a", DataType::Int32, false),
Field::new("b", data_type, true),
]);
- assert!(!GroupValuesOrdered::supports_schema(&schema));
+ assert!(!GroupValuesClustered::supports_schema(&schema));
}
- assert!(!GroupValuesOrdered::supports_schema(&Schema::new(vec![
+ assert!(!GroupValuesClustered::supports_schema(&Schema::new(vec![
Field::new("a", DataType::Int32, false)
])));
}
#[tokio::test]
async fn ordered_single_and_partial_final_match_unordered_execution() -> Result<()> {
- use crate::aggregates::{AggregateExec, AggregateMode, PhysicalGroupBy};
+ use crate::aggregates::{
+ AggregateExec, AggregateMode, GroupClusteringMode, PhysicalGroupBy,
+ };
use crate::test::TestMemoryExec;
- use crate::{ExecutionPlan, InputOrderMode, collect};
+ use crate::{ExecutionPlan, collect};
use arrow::compute::{SortOptions, take_record_batch};
use arrow::record_batch::RecordBatch;
use arrow::row::{RowConverter, SortField};
@@ -437,7 +452,10 @@ mod tests {
Arc::clone(&schema),
)?;
if sorted {
- assert_eq!(plan.input_order_mode(), &InputOrderMode::Sorted);
+ assert_eq!(
+ plan.group_clustering_mode(),
+ &GroupClusteringMode::Full
+ );
}
let plan: Arc<dyn ExecutionPlan> = if two_stage {
let plan = AggregateExec::try_new(
@@ -450,8 +468,8 @@ mod tests {
)?;
if sorted {
assert_eq!(
- plan.input_order_mode(),
- &InputOrderMode::Sorted
+ plan.group_clustering_mode(),
+ &GroupClusteringMode::Full
);
}
Arc::new(plan)
diff --git a/datafusion/physical-plan/src/aggregates/group_values/multi_group_by/list.rs b/datafusion/physical-plan/src/aggregates/group_values/multi_group_by/list.rs
index b61e1987d7..b6e46684b5 100644
--- a/datafusion/physical-plan/src/aggregates/group_values/multi_group_by/list.rs
+++ b/datafusion/physical-plan/src/aggregates/group_values/multi_group_by/list.rs
@@ -447,7 +447,7 @@ mod tests {
// List/Struct together correctly and that intern/emit round-trips
// through `GroupValuesColumn` rather than `GroupValuesRows`.
use crate::aggregates::group_values::new_group_values;
- use crate::aggregates::order::GroupOrdering;
+ use crate::aggregates::order::GroupClustering;
use arrow::array::{
Int32Array, LargeListArray, StringArray, StructArray, builder::Int32Builder,
builder::LargeListBuilder, builder::StringBuilder, builder::StructBuilder,
@@ -509,7 +509,7 @@ mod tests {
Arc::new(list_builder.finish())
};
- let mut gv = new_group_values(schema, &GroupOrdering::None).unwrap();
+ let mut gv = new_group_values(schema, &GroupClustering::None).unwrap();
// Batch 1: a mix of duplicate / distinct / null lists.
let batch1 = notes_v(&[
@@ -586,7 +586,7 @@ mod tests {
#[test]
fn list_dispatcher_round_trip_through_new_group_values() {
use crate::aggregates::group_values::new_group_values;
- use crate::aggregates::order::GroupOrdering;
+ use crate::aggregates::order::GroupClustering;
use arrow::datatypes::Schema;
use datafusion_expr::EmitTo;
@@ -595,7 +595,7 @@ mod tests {
DataType::List(child_field()),
true,
)]));
- let mut gv = new_group_values(schema, &GroupOrdering::None).unwrap();
+ let mut gv = new_group_values(schema, &GroupClustering::None).unwrap();
// Batch 1.
let batch1: ArrayRef = list_array(&[
@@ -871,7 +871,7 @@ mod tests {
// ints. Exercises a recursive child GroupColumn built via the
// dispatcher.
use crate::aggregates::group_values::new_group_values;
- use crate::aggregates::order::GroupOrdering;
+ use crate::aggregates::order::GroupClustering;
use arrow::array::{ListArray, builder::Int32Builder, builder::ListBuilder};
use arrow::datatypes::Schema;
use datafusion_expr::EmitTo;
@@ -919,7 +919,7 @@ mod tests {
Arc::new(outer.finish())
};
- let mut gv = new_group_values(schema, &GroupOrdering::None).unwrap();
+ let mut gv = new_group_values(schema, &GroupClustering::None).unwrap();
// Three groups: [[1,2],[3]], its duplicate, a distinct value, and a null.
let batch = mk(&[
diff --git a/datafusion/physical-plan/src/aggregates/group_values/multi_group_by/mod.rs b/datafusion/physical-plan/src/aggregates/group_values/multi_group_by/mod.rs
index 68d56f00a9..7ab70bf0c8 100644
--- a/datafusion/physical-plan/src/aggregates/group_values/multi_group_by/mod.rs
+++ b/datafusion/physical-plan/src/aggregates/group_values/multi_group_by/mod.rs
@@ -20,11 +20,11 @@
mod boolean;
mod bytes;
pub mod bytes_view;
+mod clustered;
mod dictionary;
mod fixed_size_binary;
mod list;
-mod ordered;
-pub(super) use ordered::GroupValuesOrdered;
+pub(super) use clustered::GroupValuesClustered;
pub mod primitive;
pub mod row_backed;
diff --git a/datafusion/physical-plan/src/aggregates/grouped_hash_stream.rs b/datafusion/physical-plan/src/aggregates/grouped_hash_stream.rs
index fd78ee721e..89131b38bc 100644
--- a/datafusion/physical-plan/src/aggregates/grouped_hash_stream.rs
+++ b/datafusion/physical-plan/src/aggregates/grouped_hash_stream.rs
@@ -35,14 +35,14 @@ use std::task::{Context, Poll};
use std::vec;
use super::aggregate_hash_table::{accumulator_phases, create_group_accumulator};
-use super::order::GroupOrdering;
+use super::order::GroupClustering;
use super::skip_partial::SkipAggregationProbe;
use super::{AggregateExec, format_human_display};
use crate::aggregates::group_values::{
AccumulatorPhase, AggregateAccumulatorMetrics, AggregateArgumentMetrics,
GroupByMetrics, GroupValues, aggregate_sub_metrics, new_group_values,
};
-use crate::aggregates::order::GroupOrderingFull;
+use crate::aggregates::order::GroupClusteringFull;
use crate::aggregates::{
AggregateInputMode, AggregateMode, AggregateOutputMode, PhysicalGroupBy,
aggregate_metric_label, create_schema, evaluate_group_by, evaluate_optional,
@@ -355,10 +355,8 @@ pub(crate) struct GroupedHashAggregateStream {
// TASK-SPECIFIC STATES:
// Inner states groups together properties, states for a specific task.
// ========================================================================
- /// Optional ordering information, that might allow groups to be
- /// emitted from the hash table prior to seeing the end of the
- /// input
- group_ordering: GroupOrdering,
+ /// Tracks groups that can be emitted from the hash table before the input ends.
+ group_clustering: GroupClustering,
/// The spill state object
spill_state: SpillState,
@@ -533,20 +531,20 @@ impl GroupedHashAggregateStream {
.collect::<Vec<_>>()
.join(", ");
let name = format!("GroupedHashAggregateStream[{partition}] ({agg_fn_names})");
- let group_ordering = GroupOrdering::try_new(&agg.input_order_mode)?;
- let oom_mode = match (agg.mode, &group_ordering) {
+ let group_clustering = GroupClustering::try_new(&agg.group_clustering_mode)?;
+ let oom_mode = match (agg.mode, &group_clustering) {
// In partial aggregation mode, always prefer to emit incomplete results early.
(AggregateMode::Partial, _) => OutOfMemoryMode::EmitEarly,
// For non-partial aggregation modes, emitting incomplete results is not an option.
// Instead, use disk spilling to store sorted, incomplete results, and merge them
// afterwards.
- (_, GroupOrdering::None | GroupOrdering::Partial(_))
+ (_, GroupClustering::None | GroupClustering::Partial(_))
if context.runtime_env().disk_manager.tmp_files_enabled() =>
{
OutOfMemoryMode::Spill
}
- // For `GroupOrdering::Full`, the incoming stream is already sorted. This ensures the
- // number of incomplete groups can be kept small at all times. If we still hit
+ // For `GroupClustering::Full`, each group is contiguous in the input. This keeps
+ // the number of incomplete groups small at all times. If we still hit
// an out-of-memory condition, spilling to disk would not be beneficial since the same
// situation is likely to reoccur when reading back the spilled data.
// Therefore, we fall back to simply reporting the error immediately.
@@ -555,7 +553,7 @@ impl GroupedHashAggregateStream {
_ => OutOfMemoryMode::ReportError,
};
- let group_values = new_group_values(group_schema, &group_ordering)?;
+ let group_values = new_group_values(group_schema, &group_clustering)?;
let merge_pool = Arc::new(MergeMemoryPool::new(
Arc::clone(context.memory_pool()),
MemoryConsumer::new(name)
@@ -595,7 +593,7 @@ impl GroupedHashAggregateStream {
// since Final mode expects unique group values as its input
// - there is only one GROUP BY expressions set
let skip_aggregation_probe = if agg.mode == AggregateMode::Partial
- && matches!(group_ordering, GroupOrdering::None)
+ && matches!(group_clustering, GroupClustering::None)
&& agg_group_by.is_single()
{
let options = &context.session_config().options().execution;
@@ -651,7 +649,7 @@ impl GroupedHashAggregateStream {
aggregate_argument_metrics,
aggregate_accumulator_metrics,
batch_size,
- group_ordering,
+ group_clustering,
input_done: false,
spill_state,
group_values_soft_limit: agg.limit_options().map(|config| config.limit()),
@@ -708,7 +706,7 @@ impl Stream for GroupedHashAggregateStream {
// this might lead to incorrect output ordering
if (self.spill_state.spills.is_empty()
|| self.spill_state.is_stream_merging)
- && let Some(to_emit) = self.group_ordering.emit_to()
+ && let Some(to_emit) = self.group_clustering.emit_to()
{
timer.done();
if let Some(batch) = self.emit(to_emit, false)? {
@@ -917,7 +915,7 @@ impl GroupedHashAggregateStream {
.collect::<Vec<_>>();
for group_values in &group_by_values {
- // Calculate group indices and update ordering information.
+ // Calculate group indices and update the completion tracker.
let total_num_groups =
self.group_by_metrics.time_group_key_preparation(|| {
let starting_num_groups = self.group_values.len();
@@ -926,7 +924,7 @@ impl GroupedHashAggregateStream {
let group_indices = &self.current_group_indices;
let total_num_groups = self.group_values.len();
if total_num_groups > starting_num_groups {
- self.group_ordering.new_groups(
+ self.group_clustering.new_groups(
group_values,
group_indices,
total_num_groups,
@@ -1008,7 +1006,7 @@ impl GroupedHashAggregateStream {
self.group_values.len()
};
- if let Some(emit_to) = self.group_ordering.oom_emit_to(n)
+ if let Some(emit_to) = self.group_clustering.oom_emit_to(n)
&& let Some(batch) = self.emit(emit_to, false)?
{
return Ok(Some(ExecutionState::ProducingOutput(batch)));
@@ -1025,7 +1023,7 @@ impl GroupedHashAggregateStream {
let acc = self.accumulators.iter().map(|x| x.size()).sum::<usize>();
let groups_and_acc_size = acc
+ self.group_values.size()
- + self.group_ordering.size()
+ + self.group_clustering.size()
+ self.current_group_indices.allocated_size();
// Reserve extra headroom for sorting during potential spill.
@@ -1071,7 +1069,7 @@ impl GroupedHashAggregateStream {
let output = group_by_metrics.time_emitting(|| {
let mut output = self.group_values.emit(emit_to)?;
if let EmitTo::First(n) = emit_to {
- self.group_ordering.remove_groups(n);
+ self.group_clustering.remove_groups(n);
}
// Next output each aggregate value.
@@ -1151,7 +1149,7 @@ impl GroupedHashAggregateStream {
.intern(&cols, &mut self.current_group_indices)?;
let total_groups = self.group_values.len();
if total_groups > starting_groups {
- self.group_ordering.new_groups(
+ self.group_clustering.new_groups(
&cols,
&self.current_group_indices,
total_groups,
@@ -1325,7 +1323,7 @@ impl GroupedHashAggregateStream {
/// in case of disk spilling, the SPM stream have been drained.
fn set_input_done_and_produce_output(&mut self) -> Result<()> {
self.input_done = true;
- self.group_ordering.input_done();
+ self.group_clustering.input_done();
// Release the original input pipeline's resources now that we're done
// reading from it. In the spill branch below, `self.input` is replaced
// again with a stream that merges spill files.
@@ -1371,12 +1369,12 @@ impl GroupedHashAggregateStream {
// Reset the group values collectors.
self.clear_all();
- // We can now use `GroupOrdering::Full` since the spill files are sorted
+ // We can now use `GroupClustering::Full` since the spill files are sorted
// on the grouping columns.
- self.group_ordering = GroupOrdering::Full(GroupOrderingFull::new());
+ self.group_clustering = GroupClustering::Full(GroupClusteringFull::new());
// Recreate `group_values` for streaming merge so group ids are assigned
- // in first-seen order, as required by `GroupOrderingFull`.
+ // in first-seen order, as required by `GroupClusteringFull`.
// The pre-spill collector may use `vectorized_intern`, which can assign
// new group ids out of input order under hash collisions. That is the
// multi-column collector, which also serves a single group column
@@ -1386,7 +1384,7 @@ impl GroupedHashAggregateStream {
.spill_state
.merging_group_by
.group_schema(&self.spill_state.spill_schema)?;
- self.group_values = new_group_values(group_schema, &self.group_ordering)?;
+ self.group_values = new_group_values(group_schema, &self.group_clustering)?;
// Use `OutOfMemoryMode::ReportError` from this point on
// to ensure we don't spill the spilled data to disk again.
@@ -1495,7 +1493,7 @@ impl GroupedHashAggregateStream {
mod tests {
use super::*;
use crate::ExecutionPlan;
- use crate::InputOrderMode;
+ use crate::aggregates::GroupClusteringMode;
use crate::test::TestMemoryExec;
use arrow::array::{Int32Array, Int64Array, UInt32Array};
use arrow::datatypes::{DataType, Field, Schema};
@@ -1782,7 +1780,7 @@ mod tests {
Ok(())
}
- // Migrated to OrderedPartialAggregateStream coverage in aggregates/mod.rs;
+ // Migrated to ClusteredPartialAggregateStream coverage in aggregates/mod.rs;
// kept here for the legacy GroupedHashAggregateStream implementation.
#[tokio::test]
async fn test_emit_early_with_partially_sorted() -> Result<()> {
@@ -1850,11 +1848,11 @@ mod tests {
Arc::clone(&schema),
)?;
assert!(matches!(
- aggregate_exec.input_order_mode(),
- InputOrderMode::PartiallySorted(_)
+ aggregate_exec.group_clustering_mode(),
+ GroupClusteringMode::Partial(_)
));
- // Must not panic with "assertion failed: *current_sort >= n"
+ // Must not panic with "assertion failed: *current_run_start >= n"
let mut stream = GroupedHashAggregateStream::new(&aggregate_exec, &task_ctx, 0)?;
while let Some(result) = stream.next().await {
if let Err(e) = result {
diff --git a/datafusion/physical-plan/src/aggregates/hash_stream.rs b/datafusion/physical-plan/src/aggregates/hash_stream.rs
index 12babfe59c..5ef4707cc1 100644
--- a/datafusion/physical-plan/src/aggregates/hash_stream.rs
+++ b/datafusion/physical-plan/src/aggregates/hash_stream.rs
@@ -34,16 +34,17 @@ use futures::stream::{Stream, StreamExt};
use super::AggregateExec;
use super::aggregate_hash_table::{
- AggregateHashTable, FinalMarker, OrderedAggregateTableMetrics, PartialMarker,
+ AggregateHashTable, ClusteredAggregateTableMetrics, FinalMarker, PartialMarker,
PartialSkipMarker,
};
+use super::order::GroupClusteringMode;
use super::skip_partial::SkipAggregationProbe;
use super::spill::AggregateSpill;
use crate::metrics::{
BaselineMetrics, MetricBuilder, MetricCategory, RecordOutput, SpillMetrics,
};
use crate::stream::{EmptyRecordBatchStream, RecordBatchStreamAdapter};
-use crate::{InputOrderMode, SendableRecordBatchStream, metrics};
+use crate::{SendableRecordBatchStream, metrics};
/// Hash aggregation is implemented in two stages: partial and final. This
/// stream implements the partial stage.
@@ -136,7 +137,7 @@ use crate::{InputOrderMode, SendableRecordBatchStream, metrics};
/// 3. Perform a sort-preserving merge of all spill files and feed the merged output
/// into an ordered streaming aggregation, which ensures bounded memory usage and
/// evaluates the final result.
-/// - [`OrderedFinalAggregateStream`](super::ordered_final_stream::OrderedFinalAggregateStream) is reused for the streaming aggregation.
+/// - [`ClusteredFinalAggregateStream`](super::clustered_final_stream::ClusteredFinalAggregateStream) is reused for the streaming aggregation.
pub(crate) struct PartialHashAggregateStream {
/// Output schema: group columns followed by partial aggregate state columns.
schema: SchemaRef,
@@ -217,7 +218,7 @@ impl PartialHashAggregateStream {
partition: usize,
) -> Result<Self> {
debug_assert_eq!(agg.mode, super::AggregateMode::Partial);
- debug_assert_eq!(agg.input_order_mode, InputOrderMode::Linear);
+ debug_assert_eq!(agg.group_clustering_mode, GroupClusteringMode::None);
let schema = Arc::clone(&agg.schema);
let input = agg.input.execute(partition, Arc::clone(context))?;
@@ -548,7 +549,7 @@ impl FinalHashAggregateStream {
agg.mode,
super::AggregateMode::Final | super::AggregateMode::FinalPartitioned
));
- debug_assert_eq!(agg.input_order_mode, InputOrderMode::Linear);
+ debug_assert_eq!(agg.group_clustering_mode, GroupClusteringMode::None);
let input = agg.input.execute(partition, Arc::clone(context))?;
Self::new_with_input(agg, context, partition, input)
@@ -582,7 +583,7 @@ impl FinalHashAggregateStream {
context,
partition,
batch_size,
- &InputOrderMode::Linear,
+ &GroupClusteringMode::None,
&input_schema,
spill_metrics,
)?))
@@ -758,7 +759,7 @@ impl FinalHashAggregateStream {
/// Produce output from spills
/// 1. Spill in progress in-memory hash table
- /// 2. Switch to ordered final stream
+ /// 2. Switch to clustered final stream
/// 3. passthrough stream output
async fn produce_output_from_spills(
&mut self,
@@ -776,7 +777,7 @@ impl FinalHashAggregateStream {
// Construct the ordered input used to merge all spill files.
let mut output_stream =
- self.switch_to_ordered_final_stream(hash_table, spill_context)?;
+ self.switch_to_clustered_final_stream(hash_table, spill_context)?;
timer.done();
@@ -794,16 +795,16 @@ impl FinalHashAggregateStream {
/// 1. Constructs a globally ordered input stream by applying a sort-preserving
/// merge to all spills.
- /// 2. Constructs a replay stream: an ordered final aggregate stream over the
+ /// 2. Constructs a replay stream: a clustered final aggregate stream over the
/// fully ordered input constructed from the spills.
///
/// Returns the replay stream
- fn switch_to_ordered_final_stream(
+ fn switch_to_clustered_final_stream(
&mut self,
hash_table: AggregateHashTable<FinalMarker>,
spill_context: Box<AggregateSpill>,
) -> Result<SendableRecordBatchStream> {
- let metrics = OrderedAggregateTableMetrics::from_hash_table(&hash_table);
+ let metrics = ClusteredAggregateTableMetrics::from_hash_table(&hash_table);
drop(hash_table);
self.reservation.try_resize(0)?;
spill_context.into_replay_stream(
@@ -854,6 +855,7 @@ mod tests {
use std::time::Duration;
use super::*;
+ use crate::aggregates::GroupClusteringMode;
use crate::aggregates::{AggregateMode, PhysicalGroupBy};
use crate::common::collect;
use crate::execution_plan::ExecutionPlan;
@@ -1458,7 +1460,10 @@ mod tests {
input,
schema,
)?;
- assert_eq!(aggregate.input_order_mode(), &InputOrderMode::Linear);
+ assert_eq!(
+ aggregate.group_clustering_mode(),
+ &GroupClusteringMode::None
+ );
let pool: Arc<dyn MemoryPool> = match limit {
Some(limit) => Arc::new(GreedyMemoryPool::new(limit)),
diff --git a/datafusion/physical-plan/src/aggregates/mod.rs b/datafusion/physical-plan/src/aggregates/mod.rs
index ed11a48fb8..7957c804d5 100644
--- a/datafusion/physical-plan/src/aggregates/mod.rs
+++ b/datafusion/physical-plan/src/aggregates/mod.rs
@@ -48,19 +48,19 @@
//!
//! See [`PartialHashAggregateStream`] and [`FinalHashAggregateStream`] for details.
//!
-//! ### Ordering optimization
+//! ### Group clustering optimization
//!
-//! When the input is ordered by the group key, an ordered fast path is used. It
-//! uses a similar two-stage hash aggregation with an early-emission optimization.
+//! When the input is ordered by group keys, rows are clustered by those keys.
+//! The clustered paths use this guarantee to emit completed groups early.
//!
//! ```text
-//! AggregateExec (final, ordered)
+//! AggregateExec (final, clustered)
//! RepartitionExec (hash by group keys, order-preserving)
-//! AggregateExec (partial, ordered)
+//! AggregateExec (partial, clustered)
//! ```
//!
-//! See [`OrderedPartialAggregateStream`], [`OrderedFinalAggregateStream`], and
-//! [`OrderedSingleAggregateStream`] for details.
+//! See [`ClusteredPartialAggregateStream`], [`ClusteredFinalAggregateStream`], and
+//! [`ClusteredSingleAggregateStream`] for details.
//!
//! Related configuration:
//!
@@ -81,7 +81,7 @@
//! input
//! ```
//!
-//! See [`SingleHashAggregateStream`] and [`OrderedSingleAggregateStream`] for
+//! See [`SingleHashAggregateStream`] and [`ClusteredSingleAggregateStream`] for
//! details.
//!
//! Related configuration:
@@ -153,12 +153,12 @@ use std::sync::Arc;
use super::{DisplayAs, ExecutionPlanProperties, PlanProperties};
use crate::aggregates::{
aggregate_stream::AggregateStream,
+ clustered_final_stream::ClusteredFinalAggregateStream,
+ clustered_partial_stream::ClusteredPartialAggregateStream,
+ clustered_single_stream::ClusteredSingleAggregateStream,
grouped_hash_stream::GroupedHashAggregateStream,
grouped_topk_stream::GroupedTopKAggregateStream,
hash_stream::{FinalHashAggregateStream, PartialHashAggregateStream},
- ordered_final_stream::OrderedFinalAggregateStream,
- ordered_partial_stream::OrderedPartialAggregateStream,
- ordered_single_stream::OrderedSingleAggregateStream,
partial_reduce_stream::PartialReduceHashAggregateStream,
single_stream::SingleHashAggregateStream,
};
@@ -174,7 +174,7 @@ use crate::statistics::{ChildStats, StatisticsArgs};
use crate::{ChildrenPropertiesMode, ReplaceChildrenOptions, validate_child_count};
use crate::{
DisplayFormatType, Distribution, ExecutionPlan, InputDistributionRequirements,
- InputOrderMode, Partitioning, SendableRecordBatchStream, Statistics,
+ Partitioning, SendableRecordBatchStream, Statistics,
};
use datafusion_common::config::ConfigOptions;
use parking_lot::Mutex;
@@ -207,20 +207,21 @@ use datafusion_physical_expr_common::sort_expr::{
use datafusion_expr::utils::AggregateOrderSensitivity;
use datafusion_physical_expr_common::utils::evaluate_expressions_to_arrays;
use itertools::Itertools;
+pub use order::GroupClusteringMode;
use topk::hash_table::is_supported_hash_key_type;
use topk::heap::is_supported_heap_type;
mod aggregate_argument;
mod aggregate_hash_table;
mod aggregate_stream;
+mod clustered_final_stream;
+mod clustered_partial_stream;
+mod clustered_single_stream;
pub mod group_values;
mod grouped_hash_stream;
mod grouped_topk_stream;
mod hash_stream;
pub mod order;
-mod ordered_final_stream;
-mod ordered_partial_stream;
-mod ordered_single_stream;
mod partial_reduce_stream;
mod single_stream;
mod skip_partial;
@@ -692,12 +693,12 @@ enum StreamType {
/// Single stage of the hash aggregation
/// Input output scheme: initial input -> final result
SingleHash(SingleHashAggregateStream),
- /// Partial stage of aggregation for ordered input.
- OrderedPartialAggregate(OrderedPartialAggregateStream),
- /// Final stage of aggregation for ordered input.
- OrderedFinalAggregate(OrderedFinalAggregateStream),
- /// Single stage of aggregation for ordered input.
- OrderedSingleAggregate(OrderedSingleAggregateStream),
+ /// Partial stage of aggregation for clustered input.
+ ClusteredPartialAggregate(ClusteredPartialAggregateStream),
+ /// Final stage of aggregation for clustered input.
+ ClusteredFinalAggregate(ClusteredFinalAggregateStream),
+ /// Single stage of aggregation for clustered input.
+ ClusteredSingleAggregate(ClusteredSingleAggregateStream),
/// Legacy hash aggregation reused for multiple stages
///
/// Every path it handles now has a dedicated stream, so this variant is only
@@ -720,9 +721,9 @@ impl From<StreamType> for SendableRecordBatchStream {
StreamType::PartialReduceHash(stream) => stream.into_stream(),
StreamType::FinalHash(stream) => stream.into_stream(),
StreamType::SingleHash(stream) => stream.into_stream(),
- StreamType::OrderedPartialAggregate(stream) => stream.into_stream(),
- StreamType::OrderedFinalAggregate(stream) => stream.into_stream(),
- StreamType::OrderedSingleAggregate(stream) => Box::pin(stream),
+ StreamType::ClusteredPartialAggregate(stream) => stream.into_stream(),
+ StreamType::ClusteredFinalAggregate(stream) => stream.into_stream(),
+ StreamType::ClusteredSingleAggregate(stream) => Box::pin(stream),
StreamType::GroupedHash(stream) => Box::pin(stream),
StreamType::GroupedPriorityQueue(stream) => Box::pin(stream),
}
@@ -911,12 +912,12 @@ pub struct AggregateExec {
/// Execution metrics
metrics: ExecutionPlanMetricsSet,
required_input_ordering: Option<OrderingRequirements>,
- /// Describes how the input is ordered relative to the group by columns
+ /// Describes how input rows are clustered by grouping expressions.
///
- /// This field is also overloaded to mean "the output MUST preserve this
- /// input order". When that is not possible, the constructor overwrites it
- /// with the unordered variant [`InputOrderMode::Linear`].
- input_order_mode: InputOrderMode,
+ /// Input ordering describes a subset of the cases in which groups can be
+ /// safely emitted before the input ends. Full group clustering requires only
+ /// that rows for each complete grouping tuple are contiguous.
+ group_clustering_mode: GroupClusteringMode,
cache: Arc<PlanProperties>,
/// During initialization, if the plan supports dynamic filtering (see [`AggrDynFilter`]),
/// it is set to `Some(..)` regardless of whether it can be pushed down to a child node.
@@ -1108,7 +1109,7 @@ impl AggregateExec {
// Commit the kind and properties together: heap output is unordered and final.
self.kind = kind;
- self.input_order_mode = InputOrderMode::Linear;
+ self.group_clustering_mode = GroupClusteringMode::None;
self.required_input_ordering = None;
// Keep unchanged properties so parent aggregates do not need rebuilding.
if !self.cache.eq_properties.oeq_class().is_empty()
@@ -1347,21 +1348,21 @@ impl AggregateExec {
.iter()
.filter(|expr| input_eq_properties.is_expr_constant(expr).is_none())
.count();
- let mut input_order_mode = if indices.len() == num_non_constant_groupby_exprs
+ let mut group_clustering_mode = if indices.len() == num_non_constant_groupby_exprs
&& !indices.is_empty()
&& group_by.groups.len() == 1
{
- InputOrderMode::Sorted
+ GroupClusteringMode::Full
} else if !indices.is_empty() {
- InputOrderMode::PartiallySorted(indices)
+ GroupClusteringMode::Partial(indices)
} else {
- InputOrderMode::Linear
+ GroupClusteringMode::None
};
- // Input order mode is also used to advertise plan output ordering, grouping
- // sets handling, and partial reduce aggregation can't promise that.
+ // Grouping sets can change group keys. PartialReduce combines intermediate
+ // states without using group boundaries to recognize completed groups.
if group_by.has_grouping_set() || mode == AggregateMode::PartialReduce {
- input_order_mode = InputOrderMode::Linear;
+ group_clustering_mode = GroupClusteringMode::None;
}
// construct a map from the input expression to the output expression of the Aggregation group by
@@ -1377,7 +1378,7 @@ impl AggregateExec {
&group_expr_mapping,
group_by.is_true_no_grouping(),
&mode,
- &input_order_mode,
+ &group_clustering_mode,
aggr_expr.as_ref(),
)?
};
@@ -1394,7 +1395,7 @@ impl AggregateExec {
input_schema,
metrics: ExecutionPlanMetricsSet::new(),
required_input_ordering,
- input_order_mode,
+ group_clustering_mode,
cache: Arc::new(cache),
dynamic_filter: None,
};
@@ -1716,47 +1717,51 @@ impl AggregateExec {
);
}
- // Choose the execution path based on (aggregation mode, ordering).
- //
- // Note that `self.input_order_mode` represents both input ordering and output
- // order promise. See its comment for details.
+ // Choose the execution path based on aggregation mode and when groups
+ // are known to be complete.
use AggregateMode::*;
- use InputOrderMode::*;
- let stream = match (self.mode, &self.input_order_mode) {
- (Partial, Linear) => StreamType::PartialHash(
+ let stream = match (self.mode, &self.group_clustering_mode) {
+ (Partial, GroupClusteringMode::None) => StreamType::PartialHash(
PartialHashAggregateStream::new(self, context, partition)?,
),
- (Partial, Sorted | PartiallySorted(_)) => {
- StreamType::OrderedPartialAggregate(OrderedPartialAggregateStream::new(
- self, context, partition,
- )?)
+ (Partial, GroupClusteringMode::Partial(_) | GroupClusteringMode::Full) => {
+ StreamType::ClusteredPartialAggregate(
+ ClusteredPartialAggregateStream::new(self, context, partition)?,
+ )
}
- (PartialReduce, Linear) => StreamType::PartialReduceHash(
+ (PartialReduce, GroupClusteringMode::None) => StreamType::PartialReduceHash(
PartialReduceHashAggregateStream::new(self, context, partition)?,
),
- (PartialReduce, Sorted | PartiallySorted(_)) => {
- // See the comment above: the builder enforces `Linear` order for
- // `PartialReduce` mode.
+ (
+ PartialReduce,
+ GroupClusteringMode::Partial(_) | GroupClusteringMode::Full,
+ ) => {
return internal_err!(
- "PartialReduce aggregation must use InputOrderMode::Linear"
+ "PartialReduce aggregation must use GroupClusteringMode::None"
);
}
- (Final | FinalPartitioned, Linear) => StreamType::FinalHash(
- FinalHashAggregateStream::new(self, context, partition)?,
- ),
- (Final | FinalPartitioned, Sorted | PartiallySorted(_)) => {
- StreamType::OrderedFinalAggregate(OrderedFinalAggregateStream::new(
+ (Final | FinalPartitioned, GroupClusteringMode::None) => {
+ StreamType::FinalHash(FinalHashAggregateStream::new(
self, context, partition,
)?)
}
- (Single | SinglePartitioned, Linear) => StreamType::SingleHash(
- SingleHashAggregateStream::new(self, context, partition)?,
- ),
- (Single | SinglePartitioned, Sorted | PartiallySorted(_)) => {
- StreamType::OrderedSingleAggregate(OrderedSingleAggregateStream::new(
+ (
+ Final | FinalPartitioned,
+ GroupClusteringMode::Partial(_) | GroupClusteringMode::Full,
+ ) => StreamType::ClusteredFinalAggregate(ClusteredFinalAggregateStream::new(
+ self, context, partition,
+ )?),
+ (Single | SinglePartitioned, GroupClusteringMode::None) => {
+ StreamType::SingleHash(SingleHashAggregateStream::new(
self, context, partition,
)?)
}
+ (
+ Single | SinglePartitioned,
+ GroupClusteringMode::Partial(_) | GroupClusteringMode::Full,
+ ) => StreamType::ClusteredSingleAggregate(
+ ClusteredSingleAggregateStream::new(self, context, partition)?,
+ ),
};
Ok(stream)
}
@@ -1814,7 +1819,7 @@ impl AggregateExec {
group_expr_mapping: &ProjectionMapping,
is_true_no_grouping: bool,
mode: &AggregateMode,
- input_order_mode: &InputOrderMode,
+ group_clustering_mode: &GroupClusteringMode,
aggr_exprs: &[Arc<AggregateFunctionExpr>],
) -> Result<PlanProperties> {
// Construct equivalence properties:
@@ -1822,9 +1827,11 @@ 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 {
+ // Only the clustered paths preserve existing ordering on group keys.
+ // Project the input's actual sort expressions; clustering alone does
+ // not establish an output ordering. Keep this consistent with
+ // `maintains_input_order`.
+ if *group_clustering_mode == GroupClusteringMode::None {
eq_properties.clear_orderings();
}
@@ -1872,7 +1879,7 @@ impl AggregateExec {
};
// TODO: Emission type and boundedness information can be enhanced here
- let emission_type = if *input_order_mode == InputOrderMode::Linear {
+ let emission_type = if *group_clustering_mode == GroupClusteringMode::None {
EmissionType::Final
} else {
input.pipeline_behavior()
@@ -1902,8 +1909,12 @@ impl AggregateExec {
)
}
- pub fn input_order_mode(&self) -> &InputOrderMode {
- &self.input_order_mode
+ /// Describes how input rows are clustered by grouping expressions.
+ ///
+ /// This does not imply a sort order. See [`ExecutionPlanProperties::output_ordering`]
+ /// for the ordering of the aggregate's output.
+ pub fn group_clustering_mode(&self) -> &GroupClusteringMode {
+ &self.group_clustering_mode
}
/// Estimates output statistics for this aggregate node.
@@ -2387,8 +2398,12 @@ impl DisplayAs for AggregateExec {
write!(f, ", lim=[{}]", config.limit)?;
}
- if self.input_order_mode != InputOrderMode::Linear {
- write!(f, ", ordering_mode={:?}", self.input_order_mode)?;
+ if self.group_clustering_mode != GroupClusteringMode::None {
+ write!(
+ f,
+ ", group_clustering_mode={:?}",
+ self.group_clustering_mode
+ )?;
}
}
DisplayFormatType::TreeRender => {
@@ -2514,17 +2529,10 @@ impl ExecutionPlan for AggregateExec {
vec![self.required_input_ordering.clone()]
}
- /// The output ordering of [`AggregateExec`] is determined by its `group_by`
- /// columns. Although this method is not explicitly used by any optimizer
- /// rules yet, overriding the default implementation ensures that it
- /// accurately reflects the actual behavior.
- ///
- /// If the [`InputOrderMode`] is `Linear`, the `group_by` columns don't have
- /// an ordering, which means the results do not either. However, in the
- /// `Ordered` and `PartiallyOrdered` cases, the `group_by` columns do have
- /// an ordering, which is preserved in the output.
+ /// Clustered aggregation preserves existing ordering on the group-by
+ /// columns. Aggregate result columns do not inherit input ordering.
fn maintains_input_order(&self) -> Vec<bool> {
- vec![self.input_order_mode != InputOrderMode::Linear]
+ vec![self.group_clustering_mode != GroupClusteringMode::None]
}
fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
@@ -2813,7 +2821,7 @@ impl ExecutionPlan for AggregateExec {
// Derived at construction from the input ordering and `group_by`.
required_input_ordering: _,
// Derived at construction from the input ordering and `group_by`.
- input_order_mode: _,
+ group_clustering_mode: _,
// Derived at construction by `Self::compute_properties`.
cache: _,
dynamic_filter,
@@ -4239,11 +4247,11 @@ mod tests {
match (mode, ordered, &stream) {
(AggregateMode::Final, false, StreamType::FinalHash(_))
| (AggregateMode::Single, false, StreamType::SingleHash(_)) => {}
- (AggregateMode::Final, true, StreamType::OrderedFinalAggregate(_))
- | (AggregateMode::Single, true, StreamType::OrderedSingleAggregate(_)) => {
+ (AggregateMode::Final, true, StreamType::ClusteredFinalAggregate(_))
+ | (AggregateMode::Single, true, StreamType::ClusteredSingleAggregate(_)) => {
assert_eq!(
- aggregate.input_order_mode(),
- &InputOrderMode::PartiallySorted(vec![0])
+ aggregate.group_clustering_mode(),
+ &GroupClusteringMode::Partial(vec![0])
);
}
_ => panic!("unexpected stream for {mode:?}, ordered={ordered}"),
@@ -5193,12 +5201,12 @@ mod tests {
let stream = aggregate.execute_typed(0, &fits)?;
match (mode, ordered, &stream) {
(AggregateMode::Partial, false, StreamType::PartialHash(_))
- | (AggregateMode::Partial, true, StreamType::OrderedPartialAggregate(_))
+ | (AggregateMode::Partial, true, StreamType::ClusteredPartialAggregate(_))
| (AggregateMode::PartialReduce, false, StreamType::PartialReduceHash(_))
| (AggregateMode::Final, false, StreamType::FinalHash(_))
- | (AggregateMode::Final, true, StreamType::OrderedFinalAggregate(_))
+ | (AggregateMode::Final, true, StreamType::ClusteredFinalAggregate(_))
| (AggregateMode::Single, false, StreamType::SingleHash(_))
- | (AggregateMode::Single, true, StreamType::OrderedSingleAggregate(_)) => {}
+ | (AggregateMode::Single, true, StreamType::ClusteredSingleAggregate(_)) => {}
_ => panic!("unexpected stream for {mode:?}, ordered={ordered}"),
}
let reserved = fits.memory_pool().reserved();
@@ -5686,7 +5694,116 @@ mod tests {
Ok(())
}
- /// Ensures `OrderedSingleAggregateStream` is used for ordered raw input.
+ #[rstest::rstest]
+ #[case::full(true)]
+ #[case::partial(false)]
+ fn group_clustering_preserves_sort_options(
+ #[case] full: bool,
+ #[values(AggregateMode::Partial, AggregateMode::Single)] mode: AggregateMode,
+ #[values(false, true)] descending: bool,
+ #[values(false, true)] nulls_first: bool,
+ ) -> Result<()> {
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("a", DataType::Int32, true),
+ Field::new("b", DataType::Int32, true),
+ Field::new("value", DataType::Int64, true),
+ ]));
+ let options = SortOptions::new(descending, nulls_first);
+ let mut input_ordering = vec![PhysicalSortExpr::new(col("a", &schema)?, options)];
+ let mut output_ordering = vec![PhysicalSortExpr::new(
+ Arc::new(Column::new("group_a", 1)),
+ options,
+ )];
+ if full {
+ input_ordering.push(PhysicalSortExpr::new(col("b", &schema)?, options));
+ output_ordering.push(PhysicalSortExpr::new(
+ Arc::new(Column::new("group_b", 0)),
+ options,
+ ));
+ }
+ let input_ordering = LexOrdering::new(input_ordering).unwrap();
+ let input = TestMemoryExec::try_new(&[vec![]], Arc::clone(&schema), None)?
+ .try_with_sort_information(vec![input_ordering.clone()])?;
+ let aggregate = AggregateExec::try_new(
+ mode,
+ PhysicalGroupBy::new_single(vec![
+ (col("b", &schema)?, "group_b".to_string()),
+ (col("a", &schema)?, "group_a".to_string()),
+ ]),
+ vec![Arc::new(
+ AggregateExprBuilder::new(sum_udaf(), vec![col("value", &schema)?])
+ .schema(Arc::clone(&schema))
+ .alias("SUM(value)")
+ .build()?,
+ )],
+ vec![None],
+ Arc::new(TestMemoryExec::update_cache(&Arc::new(input))),
+ schema,
+ )?;
+
+ let expected_mode = if full {
+ GroupClusteringMode::Full
+ } else {
+ GroupClusteringMode::Partial(vec![1])
+ };
+ assert_eq!(aggregate.group_clustering_mode(), &expected_mode);
+ assert_eq!(aggregate.maintains_input_order(), vec![true]);
+ assert_eq!(
+ aggregate.properties().emission_type,
+ EmissionType::Incremental
+ );
+ assert_eq!(
+ aggregate.properties().output_ordering(),
+ LexOrdering::new(output_ordering).as_ref()
+ );
+ assert_eq!(
+ aggregate.required_input_ordering(),
+ vec![Some(OrderingRequirements::new_soft(input_ordering.into()))]
+ );
+ Ok(())
+ }
+
+ #[test]
+ fn group_clustering_is_recomputed_with_new_children() -> Result<()> {
+ let aggregate = Arc::new(single_test_aggregate()?);
+ let original_input = Arc::clone(aggregate.input());
+ let schema = original_input.schema();
+ let ordering =
+ LexOrdering::new([PhysicalSortExpr::new_default(col("a", &schema)?)])
+ .unwrap();
+ let sorted_input = TestMemoryExec::try_new(&[vec![]], schema, None)?
+ .try_with_sort_information(vec![ordering.clone()])?;
+ let sorted_input =
+ Arc::new(TestMemoryExec::update_cache(&Arc::new(sorted_input)));
+
+ let sorted = aggregate.replace_children(
+ vec![sorted_input],
+ ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
+ )?;
+ let sorted_aggregate = sorted.downcast_ref::<AggregateExec>().unwrap();
+ assert_eq!(
+ sorted_aggregate.group_clustering_mode(),
+ &GroupClusteringMode::Full
+ );
+ assert_eq!(sorted.output_ordering(), Some(&ordering));
+ assert_eq!(sorted.pipeline_behavior(), EmissionType::Incremental);
+
+ let unordered = sorted.replace_children(
+ vec![original_input],
+ ReplaceChildrenOptions::new(ChildrenPropertiesMode::Recompute),
+ )?;
+ let unordered_aggregate = unordered.downcast_ref::<AggregateExec>().unwrap();
+ assert_eq!(
+ unordered_aggregate.group_clustering_mode(),
+ &GroupClusteringMode::None
+ );
+ assert_eq!(unordered.maintains_input_order(), vec![false]);
+ assert!(unordered.output_ordering().is_none());
+ assert_eq!(unordered.pipeline_behavior(), EmissionType::Final);
+ Ok(())
+ }
+
+ /// Ensures `ClusteredSingleAggregateStream` is used for ordered raw input.
#[tokio::test]
async fn ordered_single_aggregate_planning() -> Result<()> {
let schema = Arc::new(Schema::new(vec![
@@ -5735,14 +5852,14 @@ mod tests {
input,
Arc::clone(&schema),
)?;
- assert!(matches!(
- aggregate.input_order_mode(),
- InputOrderMode::PartiallySorted(_)
- ));
+ assert_eq!(
+ aggregate.group_clustering_mode(),
+ &GroupClusteringMode::Partial(vec![0])
+ );
let task_ctx = new_migrated_hash_ctx(2);
let stream = aggregate.execute_typed(0, &task_ctx)?;
- assert!(matches!(stream, StreamType::OrderedSingleAggregate(_)));
+ assert!(matches!(stream, StreamType::ClusteredSingleAggregate(_)));
let stream: SendableRecordBatchStream = stream.into();
let output = collect(stream).await?;
assert_snapshot!(batches_to_sort_string(&output), @r"
@@ -5758,7 +5875,7 @@ mod tests {
let finite_memory_task_ctx = new_finite_memory_migrated_hash_ctx(2, 1024 * 1024)?;
let stream = aggregate.execute_typed(0, &finite_memory_task_ctx)?;
- assert!(matches!(stream, StreamType::OrderedSingleAggregate(_)));
+ assert!(matches!(stream, StreamType::ClusteredSingleAggregate(_)));
Ok(())
}
@@ -5795,7 +5912,10 @@ mod tests {
Arc::clone(&schema),
)?;
- assert_eq!(partial_reduce.input_order_mode(), &InputOrderMode::Linear);
+ assert_eq!(
+ partial_reduce.group_clustering_mode(),
+ &GroupClusteringMode::None
+ );
assert_eq!(partial_reduce.maintains_input_order(), vec![false]);
assert!(
partial_reduce.properties().output_ordering().is_none(),
@@ -5850,7 +5970,10 @@ mod tests {
None,
)?;
let aggregate = build_aggregate(unordered_input)?;
- assert_eq!(aggregate.input_order_mode(), &InputOrderMode::Linear);
+ assert_eq!(
+ aggregate.group_clustering_mode(),
+ &GroupClusteringMode::None
+ );
assert_eq!(
aggregate.schema().as_ref(),
&Schema::new(vec![
@@ -5882,7 +6005,10 @@ mod tests {
let ordered_input =
Arc::new(TestMemoryExec::update_cache(&Arc::new(ordered_input)));
let aggregate = build_aggregate(ordered_input)?;
- assert_eq!(aggregate.input_order_mode(), &InputOrderMode::Sorted);
+ assert_eq!(
+ aggregate.group_clustering_mode(),
+ &GroupClusteringMode::Full
+ );
Ok(())
}
@@ -6202,7 +6328,10 @@ mod tests {
schema,
)?;
- assert_eq!(aggregate.input_order_mode(), &InputOrderMode::Linear);
+ assert_eq!(
+ aggregate.group_clustering_mode(),
+ &GroupClusteringMode::None
+ );
// This captures the behavior before #24438. When the source can declare
// `(key, time_bin)` group-contiguous, the corresponding case can use
// `EmissionType::Incremental`.
@@ -6227,7 +6356,7 @@ mod tests {
Ok(())
}
- /// Ensures for ordered input, `OrderedPartialAggregateStream` is used.
+ /// Ensures for ordered input, `ClusteredPartialAggregateStream` is used.
#[tokio::test]
async fn ordered_partial_aggregate_planning() -> Result<()> {
let schema = Arc::new(Schema::new(vec![
@@ -6281,13 +6410,13 @@ mod tests {
Arc::clone(&schema),
)?;
assert!(matches!(
- aggregate.input_order_mode(),
- InputOrderMode::PartiallySorted(_)
+ aggregate.group_clustering_mode(),
+ GroupClusteringMode::Partial(_)
));
let task_ctx = new_migrated_hash_ctx(2);
let stream = aggregate.execute_typed(0, &task_ctx)?;
- assert!(matches!(stream, StreamType::OrderedPartialAggregate(_)));
+ assert!(matches!(stream, StreamType::ClusteredPartialAggregate(_)));
let stream: SendableRecordBatchStream = stream.into();
let output = collect(stream).await?;
@@ -6303,15 +6432,15 @@ mod tests {
+----------+-----------+-------------------------+
");
- // Ordered partial aggregation supports finite memory.
+ // Clustered partial aggregation supports finite memory.
let finite_memory_task_ctx = new_finite_memory_migrated_hash_ctx(2, 1024 * 1024)?;
let stream = aggregate.execute_typed(0, &finite_memory_task_ctx)?;
- assert!(matches!(stream, StreamType::OrderedPartialAggregate(_)));
+ assert!(matches!(stream, StreamType::ClusteredPartialAggregate(_)));
Ok(())
}
- /// Ensures for ordered input, `OrderedFinalAggregateStream` is used.
+ /// Ensures for ordered input, `ClusteredFinalAggregateStream` is used.
#[tokio::test]
async fn ordered_final_aggregate_planning() -> Result<()> {
let schema = Arc::new(Schema::new(vec![
@@ -6362,11 +6491,14 @@ mod tests {
final_input,
Arc::clone(&schema),
)?;
- assert_eq!(final_aggregate.input_order_mode(), &InputOrderMode::Sorted);
+ assert_eq!(
+ final_aggregate.group_clustering_mode(),
+ &GroupClusteringMode::Full
+ );
let task_ctx = new_migrated_hash_ctx(2);
let stream = final_aggregate.execute_typed(0, &task_ctx)?;
- assert!(matches!(stream, StreamType::OrderedFinalAggregate(_)));
+ assert!(matches!(stream, StreamType::ClusteredFinalAggregate(_)));
let stream: SendableRecordBatchStream = stream.into();
let output = collect(stream).await?;
@@ -6381,10 +6513,10 @@ mod tests {
+-----+--------------+
");
- // Ordered final aggregation supports finite memory.
+ // Clustered final aggregation supports finite memory.
let finite_memory_task_ctx = new_finite_memory_migrated_hash_ctx(2, 1024 * 1024)?;
let stream = final_aggregate.execute_typed(0, &finite_memory_task_ctx)?;
- assert!(matches!(stream, StreamType::OrderedFinalAggregate(_)));
+ assert!(matches!(stream, StreamType::ClusteredFinalAggregate(_)));
Ok(())
}
@@ -6436,8 +6568,8 @@ mod tests {
Arc::clone(&schema),
)?;
assert!(matches!(
- aggregate.input_order_mode(),
- InputOrderMode::PartiallySorted(_)
+ aggregate.group_clustering_mode(),
+ GroupClusteringMode::Partial(_)
));
let runtime = RuntimeEnvBuilder::default()
@@ -6454,7 +6586,7 @@ mod tests {
);
let mut stream: SendableRecordBatchStream =
- OrderedPartialAggregateStream::new(&aggregate, &task_ctx, 0)?.into_stream();
+ ClusteredPartialAggregateStream::new(&aggregate, &task_ctx, 0)?.into_stream();
while let Some(result) = stream.next().await {
if let Err(e) = result {
@@ -6735,7 +6867,7 @@ mod tests {
//
// "AggregateExec: mode=Final, gby=[a@0 as a], aggr=[FIRST_VALUE(b)]",
// " CoalescePartitionsExec",
- // " AggregateExec: mode=Partial, gby=[a@0 as a], aggr=[FIRST_VALUE(b)], ordering_mode=None",
+ // " AggregateExec: mode=Partial, gby=[a@0 as a], aggr=[FIRST_VALUE(b)], group_clustering_mode=None",
// " DataSourceExec: partitions=4, partition_sizes=[1, 1, 1, 1]",
//
// and checks whether the function `merge_batch` works correctly for
diff --git a/datafusion/physical-plan/src/aggregates/order/full.rs b/datafusion/physical-plan/src/aggregates/order/full.rs
index 5b17d527e5..7da50f6fa0 100644
--- a/datafusion/physical-plan/src/aggregates/order/full.rs
+++ b/datafusion/physical-plan/src/aggregates/order/full.rs
@@ -17,10 +17,10 @@
use datafusion_expr::EmitTo;
-/// Tracks grouping state when the data is ordered entirely by its
-/// group keys
+/// Tracks group completion when rows are contiguous for the complete
+/// grouping tuple.
///
-/// When the group values are sorted, as soon as we see group `n+1` we
+/// When groups are contiguous, as soon as we see group `n+1` we
/// know we will never see any rows for group `n` again and thus they
/// can be emitted.
///
@@ -54,7 +54,7 @@ use datafusion_expr::EmitTo;
/// `0..12` can be emitted. Note that `13` can not yet be emitted as
/// there may be more values in the next batch with the same group_id.
#[derive(Debug)]
-pub struct GroupOrderingFull {
+pub struct GroupClusteringFull {
state: State,
}
@@ -71,7 +71,7 @@ enum State {
Complete,
}
-impl GroupOrderingFull {
+impl GroupClusteringFull {
pub fn new() -> Self {
Self {
state: State::Start,
@@ -114,13 +114,13 @@ impl GroupOrderingFull {
self.state = State::Complete;
}
- /// Starts tracking a new fully ordered input segment.
+ /// Starts tracking a new input segment with contiguous groups.
pub fn reset(&mut self) {
self.state = State::Start;
}
/// Called when new groups are added in a batch. See documentation
- /// on [`super::GroupOrdering::new_groups`]
+ /// on [`super::GroupClustering::new_groups`]
pub fn new_groups(&mut self, total_num_groups: usize) {
assert_ne!(total_num_groups, 0);
@@ -144,7 +144,7 @@ impl GroupOrderingFull {
}
}
-impl Default for GroupOrderingFull {
+impl Default for GroupClusteringFull {
fn default() -> Self {
Self::new()
}
diff --git a/datafusion/physical-plan/src/aggregates/order/mod.rs b/datafusion/physical-plan/src/aggregates/order/mod.rs
index 88ee2af5f2..9b59376a4b 100644
--- a/datafusion/physical-plan/src/aggregates/order/mod.rs
+++ b/datafusion/physical-plan/src/aggregates/order/mod.rs
@@ -24,48 +24,86 @@ use datafusion_expr::EmitTo;
mod full;
mod partial;
-use crate::InputOrderMode;
-pub use full::GroupOrderingFull;
-pub use partial::GroupOrderingPartial;
+pub use full::GroupClusteringFull;
+pub use partial::GroupClusteringPartial;
+
+/// Describes how rows are clustered by grouping expressions within each input
+/// partition.
+///
+/// Input ordering is one way to establish clustering. Aggregate execution uses
+/// these guarantees to determine when groups are complete and can be emitted.
+/// This mode does not describe the sort order of the input or output.
+///
+/// For example, when grouping by `key`, both inputs have fully contiguous
+/// groups within the input partition:
+///
+/// ```text
+/// sorted: A A B B C C
+/// not sorted: C C A A B B
+/// ```
+///
+/// In both cases, once the key changes, the previous key will not appear again,
+/// so its group is complete and can be emitted.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum GroupClusteringMode {
+ /// No clustering guarantee is available to complete groups before the input ends.
+ None,
+ /// Rows with the same values at these grouping-expression indices form one
+ /// contiguous range. When those values change, every group in the previous
+ /// range is complete and can be emitted.
+ ///
+ /// For example, with `GROUP BY (a, b)`, `Partial(vec![0])` means all rows
+ /// for each value of `a` are contiguous, while an `(a, b)` tuple may recur
+ /// within that range.
+ Partial(Vec<usize>),
+ /// Rows with the same complete grouping tuple form one contiguous range.
+ /// When the tuple changes, the previous group can be emitted.
+ Full,
+}
-/// Ordering information for each group in the hash table
+/// Tracks when groups in the hash table are complete and can be emitted.
#[derive(Debug)]
-pub enum GroupOrdering {
- /// Groups are not ordered
+pub enum GroupClustering {
+ /// No group can be known complete before the input ends.
None,
- /// Groups are ordered by some pre-set of the group keys
- Partial(GroupOrderingPartial),
- /// Groups are entirely contiguous,
- Full(GroupOrderingFull),
+ /// Rows are contiguous for a subset of the grouping keys.
+ /// When those key values change, all groups in the previous run
+ /// are complete and can be emitted.
+ Partial(GroupClusteringPartial),
+ /// Rows are contiguous for the complete grouping tuple.
+ /// When the tuple changes, the previous group can be emitted.
+ Full(GroupClusteringFull),
}
-impl GroupOrdering {
- /// Create a `GroupOrdering` for the specified ordering
- pub fn try_new(mode: &InputOrderMode) -> Result<Self> {
+impl GroupClustering {
+ /// Create a `GroupClustering` for the specified group-clustering mode.
+ pub fn try_new(mode: &GroupClusteringMode) -> Result<Self> {
match mode {
- InputOrderMode::Linear => Ok(GroupOrdering::None),
- InputOrderMode::PartiallySorted(order_indices) => {
- GroupOrderingPartial::try_new(order_indices.clone())
- .map(GroupOrdering::Partial)
+ GroupClusteringMode::None => Ok(GroupClustering::None),
+ GroupClusteringMode::Partial(grouping_indices) => {
+ GroupClusteringPartial::try_new(grouping_indices.clone())
+ .map(GroupClustering::Partial)
+ }
+ GroupClusteringMode::Full => {
+ Ok(GroupClustering::Full(GroupClusteringFull::new()))
}
- InputOrderMode::Sorted => Ok(GroupOrdering::Full(GroupOrderingFull::new())),
}
}
- /// Returns how many groups can be emitted while respecting the current
- /// ordering guarantees, or `None` if no data can be emitted.
+ /// Returns how many completed groups can be emitted, or `None` if no data
+ /// can be emitted.
pub fn emit_to(&self) -> Option<EmitTo> {
match self {
- GroupOrdering::None => None,
- GroupOrdering::Partial(partial) => partial.emit_to(),
- GroupOrdering::Full(full) => full.emit_to(),
+ GroupClustering::None => None,
+ GroupClustering::Partial(partial) => partial.emit_to(),
+ GroupClustering::Full(full) => full.emit_to(),
}
}
/// Returns the emit strategy to use under memory pressure (OOM).
///
/// Returns the strategy that must be used when emitting up to `n` groups
- /// while respecting the current ordering guarantees.
+ /// while respecting the configured group-clustering mode.
///
/// Returns `None` if no data can be emitted.
pub fn oom_emit_to(&self, n: usize) -> Option<EmitTo> {
@@ -74,8 +112,8 @@ impl GroupOrdering {
}
match self {
- GroupOrdering::None => Some(EmitTo::First(n)),
- GroupOrdering::Partial(_) | GroupOrdering::Full(_) => {
+ GroupClustering::None => Some(EmitTo::First(n)),
+ GroupClustering::Partial(_) | GroupClustering::Full(_) => {
self.emit_to().map(|emit_to| match emit_to {
EmitTo::First(max) => EmitTo::First(n.min(max)),
EmitTo::All => EmitTo::First(n),
@@ -87,23 +125,23 @@ impl GroupOrdering {
/// Updates the state to indicate that the input is complete.
pub fn input_done(&mut self) {
match self {
- GroupOrdering::None => {}
- GroupOrdering::Partial(partial) => partial.input_done(),
- GroupOrdering::Full(full) => full.input_done(),
+ GroupClustering::None => {}
+ GroupClustering::Partial(partial) => partial.input_done(),
+ GroupClustering::Full(full) => full.input_done(),
}
}
- /// Resets the ordering state while preserving the configured ordering mode.
+ /// Resets the completion state while preserving the configured mode.
///
- /// Ordered partial aggregation uses this after passing intermediate states
- /// downstream, and ordered final aggregation uses it after spilling a run.
+ /// Clustered partial aggregation uses this after passing intermediate states
+ /// downstream, and clustered final aggregation uses it after spilling a run.
/// In both cases the hash table is empty and can start tracking the next
- /// input batch from a fresh ordering state.
+ /// input batch from a fresh completion state.
pub fn reset(&mut self) {
match self {
- GroupOrdering::None => {}
- GroupOrdering::Partial(partial) => partial.reset(),
- GroupOrdering::Full(full) => full.reset(),
+ GroupClustering::None => {}
+ GroupClustering::Partial(partial) => partial.reset(),
+ GroupClustering::Full(full) => full.reset(),
}
}
@@ -111,9 +149,9 @@ impl GroupOrdering {
/// existing indexes down by `n`.
pub fn remove_groups(&mut self, n: usize) {
match self {
- GroupOrdering::None => {}
- GroupOrdering::Partial(partial) => partial.remove_groups(n),
- GroupOrdering::Full(full) => full.remove_groups(n),
+ GroupClustering::None => {}
+ GroupClustering::Partial(partial) => partial.remove_groups(n),
+ GroupClustering::Full(full) => full.remove_groups(n),
}
}
@@ -132,22 +170,22 @@ impl GroupOrdering {
total_num_groups: usize,
) -> Result<()> {
match self {
- GroupOrdering::None => {}
- GroupOrdering::Partial(partial) => {
+ GroupClustering::None => {}
+ GroupClustering::Partial(partial) => {
partial.new_groups(
batch_group_values,
group_indices,
total_num_groups,
)?;
}
- GroupOrdering::Full(full) => {
+ GroupClustering::Full(full) => {
full.new_groups(total_num_groups);
}
}
Ok(())
}
- /// Returns the size of memory used by the ordering state, in bytes.
+ /// Returns the size of memory used by the completion state, in bytes.
///
/// Includes the enum descriptor once (covering the inline active variant)
/// and any heap allocations retained by that variant.
@@ -155,8 +193,8 @@ impl GroupOrdering {
size_of::<Self>()
+ match self {
// These variants hold only inline state, already counted above.
- GroupOrdering::None | GroupOrdering::Full(_) => 0,
- GroupOrdering::Partial(partial) => partial.heap_size(),
+ GroupClustering::None | GroupClustering::Full(_) => 0,
+ GroupClustering::Partial(partial) => partial.heap_size(),
}
}
}
@@ -172,85 +210,85 @@ mod tests {
#[test]
fn test_size_inline_only() -> Result<()> {
- let expected = size_of::<GroupOrdering>();
- assert_eq!(GroupOrdering::None.size(), expected);
- let ordering = GroupOrdering::try_new(&InputOrderMode::Sorted)?;
- assert_eq!(ordering.size(), expected);
+ let expected = size_of::<GroupClustering>();
+ assert_eq!(GroupClustering::None.size(), expected);
+ let group_clustering = GroupClustering::try_new(&GroupClusteringMode::Full)?;
+ assert_eq!(group_clustering.size(), expected);
Ok(())
}
#[test]
fn test_size_partial_retained_allocations() -> Result<()> {
- // The direct partial constructor retains spare order-index capacity.
- let mut order_indices = Vec::with_capacity(32);
- order_indices.push(0);
- let expected =
- size_of::<GroupOrdering>() + order_indices.capacity() * size_of::<usize>();
- let mut ordering =
- GroupOrdering::Partial(GroupOrderingPartial::try_new(order_indices)?);
- assert_eq!(ordering.size(), expected);
+ // The direct partial constructor retains spare grouping-index capacity.
+ let mut grouping_indices = Vec::with_capacity(32);
+ grouping_indices.push(0);
+ let expected = size_of::<GroupClustering>()
+ + grouping_indices.capacity() * size_of::<usize>();
+ let mut group_clustering =
+ GroupClustering::Partial(GroupClusteringPartial::try_new(grouping_indices)?);
+ assert_eq!(group_clustering.size(), expected);
// A variable-width key checks both the scalar descriptor and its payload.
- let key = "retained sort key";
+ let key = "retained group key";
let batch_group_values: Vec<ArrayRef> =
vec![Arc::new(StringArray::from(vec!["a", key]))];
- ordering.new_groups(&batch_group_values, &[0, 1], 2)?;
+ group_clustering.new_groups(&batch_group_values, &[0, 1], 2)?;
let in_progress = expected + ScalarValue::Utf8(Some(key.to_owned())).size();
- assert_eq!(ordering.size(), in_progress);
+ assert_eq!(group_clustering.size(), in_progress);
- // Completing drops the key, but retains order-index capacity.
- ordering.input_done();
- assert_eq!(ordering.size(), expected);
+ // Completing drops the key, but retains grouping-index capacity.
+ group_clustering.input_done();
+ assert_eq!(group_clustering.size(), expected);
Ok(())
}
#[test]
- fn test_oom_emit_to_none_ordering() {
- let group_ordering = GroupOrdering::None;
+ fn test_oom_emit_to_none_clustering() {
+ let group_clustering = GroupClustering::None;
- assert_eq!(group_ordering.oom_emit_to(0), None);
- assert_eq!(group_ordering.oom_emit_to(5), Some(EmitTo::First(5)));
+ assert_eq!(group_clustering.oom_emit_to(0), None);
+ assert_eq!(group_clustering.oom_emit_to(5), Some(EmitTo::First(5)));
}
- /// Creates a partially ordered grouping state with three groups.
+ /// Creates a partial group-clustering tracker with three groups.
///
- /// `sort_key_values` controls whether a sort boundary exists in the batch:
+ /// `group_key_values` controls whether a run boundary exists in the batch:
/// distinct values such as `[1, 2, 3]` create boundaries, while repeated
/// values such as `[1, 1, 1]` do not.
- fn partial_ordering(sort_key_values: Vec<i32>) -> Result<GroupOrdering> {
- let mut group_ordering =
- GroupOrdering::Partial(GroupOrderingPartial::try_new(vec![0])?);
+ fn partial_clustering(group_key_values: Vec<i32>) -> Result<GroupClustering> {
+ let mut group_clustering =
+ GroupClustering::Partial(GroupClusteringPartial::try_new(vec![0])?);
let batch_group_values: Vec<ArrayRef> = vec![
- Arc::new(Int32Array::from(sort_key_values)),
+ Arc::new(Int32Array::from(group_key_values)),
Arc::new(Int32Array::from(vec![10, 20, 30])),
];
let group_indices = vec![0, 1, 2];
- group_ordering.new_groups(&batch_group_values, &group_indices, 3)?;
+ group_clustering.new_groups(&batch_group_values, &group_indices, 3)?;
- Ok(group_ordering)
+ Ok(group_clustering)
}
#[test]
fn test_oom_emit_to_partial_clamps_to_boundary() -> Result<()> {
- let group_ordering = partial_ordering(vec![1, 2, 3])?;
+ let group_clustering = partial_clustering(vec![1, 2, 3])?;
// Can emit both `1` and `2` groups because we have seen `3`
- assert_eq!(group_ordering.emit_to(), Some(EmitTo::First(2)));
- assert_eq!(group_ordering.oom_emit_to(1), Some(EmitTo::First(1)));
- assert_eq!(group_ordering.oom_emit_to(3), Some(EmitTo::First(2)));
+ assert_eq!(group_clustering.emit_to(), Some(EmitTo::First(2)));
+ assert_eq!(group_clustering.oom_emit_to(1), Some(EmitTo::First(1)));
+ assert_eq!(group_clustering.oom_emit_to(3), Some(EmitTo::First(2)));
Ok(())
}
#[test]
fn test_oom_emit_to_partial_without_boundary() -> Result<()> {
- let group_ordering = partial_ordering(vec![1, 1, 1])?;
+ let group_clustering = partial_clustering(vec![1, 1, 1])?;
// Can't emit the last `1` group as it may have more values
- assert_eq!(group_ordering.emit_to(), None);
- assert_eq!(group_ordering.oom_emit_to(3), None);
+ assert_eq!(group_clustering.emit_to(), None);
+ assert_eq!(group_clustering.oom_emit_to(3), None);
Ok(())
}
diff --git a/datafusion/physical-plan/src/aggregates/order/partial.rs b/datafusion/physical-plan/src/aggregates/order/partial.rs
index 2488c24913..3864eaa7a1 100644
--- a/datafusion/physical-plan/src/aggregates/order/partial.rs
+++ b/datafusion/physical-plan/src/aggregates/order/partial.rs
@@ -26,12 +26,11 @@ use datafusion_common::{Result, ScalarValue};
use datafusion_execution::memory_pool::proxy::VecAllocExt;
use datafusion_expr::EmitTo;
-/// Tracks grouping state when the data is ordered by some subset of
+/// Tracks group completion when rows are contiguous for a subset of
/// the group keys.
///
-/// Once the next *sort key* value is seen, never see groups with that
-/// sort key again, so we can emit all groups with the previous sort
-/// key and earlier.
+/// Once those key values change, they will not appear again, so all groups
+/// in the previous run are complete and can be emitted.
///
/// For example, given `SUM(amt) GROUP BY id, state` if the input is
/// sorted by `state`, when a new value of `state` is seen, all groups
@@ -43,11 +42,11 @@ use datafusion_expr::EmitTo;
/// ┏━━━━━━━━━━━━━━━━━┓ ┏━━━━━━━┓
/// ┌─────┐ ┌───────────────────┐ ┌─────┃ 9 ┃ ┃ "MD" ┃
/// │┌───┐│ │ ┌──────────────┐ │ │ ┗━━━━━━━━━━━━━━━━━┛ ┗━━━━━━━┛
-/// ││ 0 ││ │ │ 123, "MA" │ │ │ current_sort sort_key
+/// ││ 0 ││ │ │ 123, "MA" │ │ │ current_run_start group_key
/// │└───┘│ │ └──────────────┘ │ │
-/// │ ... │ │ ... │ │ current_sort tracks the
+/// │ ... │ │ ... │ │ current_run_start tracks the
/// │┌───┐│ │ ┌──────────────┐ │ │ smallest group index that had
-/// ││ 8 ││ │ │ 765, "MA" │ │ │ the same sort_key as current
+/// ││ 8 ││ │ │ 765, "MA" │ │ │ the same group_key as current
/// │├───┤│ │ ├──────────────┤ │ │
/// ││ 9 ││ │ │ 923, "MD" │◀─┼─┘
/// │├───┤│ │ ├──────────────┤ │ ┏━━━━━━━━━━━━━━┓
@@ -62,22 +61,22 @@ use datafusion_expr::EmitTo;
/// order) recent group index
/// ```
#[derive(Debug)]
-pub struct GroupOrderingPartial {
+pub struct GroupClusteringPartial {
/// State machine
state: State,
- /// The indexes of the group by columns that form the sort key.
- /// For example if grouping by `id, state` and ordered by `state`
+ /// The indexes of the group by columns whose values form contiguous runs.
+ /// For example if grouping by `id, state` and contiguous on `state`
/// this would be `[1]`.
- order_indices: Vec<usize>,
+ grouping_indices: Vec<usize>,
}
#[derive(Debug, Default, PartialEq)]
enum State {
- /// The ordering was temporarily taken. `Self::Taken` is left
+ /// The state was temporarily taken. `Self::Taken` is left
/// when state must be temporarily taken to satisfy the borrow
/// checker. If an error happens before the state can be restored,
- /// the ordering information is lost and execution can not
+ /// the completion information is lost and execution can not
/// proceed, but there is no undefined behavior.
#[default]
Taken,
@@ -87,10 +86,10 @@ enum State {
/// Data is in progress.
InProgress {
- /// Smallest group index with the sort_key
- current_sort: usize,
- /// The sort key of group_index `current_sort`
- sort_key: Vec<ScalarValue>,
+ /// Smallest group index in the current run.
+ current_run_start: usize,
+ /// The key values of the current run.
+ group_key: Vec<ScalarValue>,
/// index of the current group for which values are being
/// generated
current: usize,
@@ -105,7 +104,7 @@ impl State {
match self {
State::Taken => 0,
State::Start => 0,
- State::InProgress { sort_key, .. } => sort_key
+ State::InProgress { group_key, .. } => group_key
.iter()
.map(|scalar_value| scalar_value.size())
.sum(),
@@ -114,24 +113,23 @@ impl State {
}
}
-impl GroupOrderingPartial {
- /// TODO: Remove unnecessary `input_schema` parameter.
- pub fn try_new(order_indices: Vec<usize>) -> Result<Self> {
- debug_assert!(!order_indices.is_empty());
+impl GroupClusteringPartial {
+ /// Creates a tracker for runs defined by the specified grouping columns.
+ pub fn try_new(grouping_indices: Vec<usize>) -> Result<Self> {
+ debug_assert!(!grouping_indices.is_empty());
Ok(Self {
state: State::Start,
- order_indices,
+ grouping_indices,
})
}
- /// Select sort keys from the group values
+ /// Select the keys that define contiguous runs from the group values.
///
- /// For example, if group_values had `A, B, C` but the input was
- /// only sorted on `B` and `C` this should return rows for (`B`,
- /// `C`)
- fn compute_sort_keys(&mut self, group_values: &[ArrayRef]) -> Vec<ArrayRef> {
- // Take only the columns that are in the sort key
- self.order_indices
+ /// For example, if `group_values` contains `A, B, C` but the input is
+ /// contiguous on `(B, C)`, this returns the arrays for `B` and `C`.
+ fn compute_group_keys(&mut self, group_values: &[ArrayRef]) -> Vec<ArrayRef> {
+ // Take only the columns that define contiguous runs.
+ self.grouping_indices
.iter()
.map(|&idx| Arc::clone(&group_values[idx]))
.collect()
@@ -142,14 +140,14 @@ impl GroupOrderingPartial {
match &self.state {
State::Taken => unreachable!("State previously taken"),
State::Start => None,
- State::InProgress { current_sort, .. } => {
- // Can not emit if we are still on the first row sort
- // row otherwise we can emit all groups that had earlier sort keys
- //
- if *current_sort == 0 {
+ State::InProgress {
+ current_run_start, ..
+ } => {
+ // The current run is incomplete; only groups from earlier runs can be emitted.
+ if *current_run_start == 0 {
None
} else {
- Some(EmitTo::First(*current_sort))
+ Some(EmitTo::First(*current_run_start))
}
}
State::Complete => Some(EmitTo::All),
@@ -163,15 +161,15 @@ impl GroupOrderingPartial {
State::Taken => unreachable!("State previously taken"),
State::Start => panic!("invalid state: start"),
State::InProgress {
- current_sort,
+ current_run_start,
current,
- sort_key: _,
+ group_key: _,
} => {
// shift indexes down by n
assert!(*current >= n);
*current -= n;
- assert!(*current_sort >= n);
- *current_sort -= n;
+ assert!(*current_run_start >= n);
+ *current_run_start -= n;
}
State::Complete => panic!("invalid state: complete"),
}
@@ -185,31 +183,31 @@ impl GroupOrderingPartial {
};
}
- /// Starts tracking a new ordered input segment with the same sort-key
+ /// Starts tracking a new input segment with the same contiguous-key
/// columns.
pub fn reset(&mut self) {
self.state = State::Start;
}
- fn updated_sort_key(
- current_sort: usize,
- sort_key: Option<Vec<ScalarValue>>,
- range_current_sort: usize,
- range_sort_key: Vec<ScalarValue>,
+ fn updated_group_key(
+ current_run_start: usize,
+ group_key: Option<Vec<ScalarValue>>,
+ range_current_run_start: usize,
+ range_group_key: Vec<ScalarValue>,
) -> Result<(usize, Vec<ScalarValue>)> {
- if let Some(sort_key) = sort_key {
- let sort_options = vec![SortOptions::new(false, false); sort_key.len()];
- let ordering = compare_rows(&sort_key, &range_sort_key, &sort_options)?;
+ if let Some(group_key) = group_key {
+ let sort_options = vec![SortOptions::new(false, false); group_key.len()];
+ let ordering = compare_rows(&group_key, &range_group_key, &sort_options)?;
if ordering == Ordering::Equal {
- return Ok((current_sort, sort_key));
+ return Ok((current_run_start, group_key));
}
}
- Ok((range_current_sort, range_sort_key))
+ Ok((range_current_run_start, range_group_key))
}
/// Called when new groups are added in a batch. See documentation
- /// on [`super::GroupOrdering::new_groups`]
+ /// on [`super::GroupClustering::new_groups`]
pub fn new_groups(
&mut self,
batch_group_values: &[ArrayRef],
@@ -221,55 +219,55 @@ impl GroupOrderingPartial {
let max_group_index = total_num_groups - 1;
- let (current_sort, sort_key) = match std::mem::take(&mut self.state) {
+ let (current_run_start, group_key) = match std::mem::take(&mut self.state) {
State::Taken => unreachable!("State previously taken"),
State::Start => (0, None),
State::InProgress {
- current_sort,
- sort_key,
+ current_run_start,
+ group_key,
..
- } => (current_sort, Some(sort_key)),
+ } => (current_run_start, Some(group_key)),
State::Complete => {
panic!("Saw new group after the end of input");
}
};
- // Select the sort key columns
- let sort_keys = self.compute_sort_keys(batch_group_values);
+ // Select the columns that define contiguous runs.
+ let group_keys = self.compute_group_keys(batch_group_values);
- // Check if the sort keys indicate a boundary inside the batch
- let ranges = partition(&sort_keys)?.ranges();
+ // Check if the key values indicate a boundary inside the batch.
+ let ranges = partition(&group_keys)?.ranges();
let last_range = ranges.last().unwrap();
- let range_current_sort = group_indices[last_range.start];
- let range_sort_key = get_row_at_idx(&sort_keys, last_range.start)?;
+ let range_current_run_start = group_indices[last_range.start];
+ let range_group_key = get_row_at_idx(&group_keys, last_range.start)?;
- let (current_sort, sort_key) = if last_range.start == 0 {
- // There was no boundary in the batch. Compare with the previous sort_key (if present)
+ let (current_run_start, group_key) = if last_range.start == 0 {
+ // There was no boundary in the batch. Compare with the previous group_key (if present)
// to check if there was a boundary between the current batch and the previous one.
- Self::updated_sort_key(
- current_sort,
- sort_key,
- range_current_sort,
- range_sort_key,
+ Self::updated_group_key(
+ current_run_start,
+ group_key,
+ range_current_run_start,
+ range_group_key,
)?
} else {
- (range_current_sort, range_sort_key)
+ (range_current_run_start, range_group_key)
};
self.state = State::InProgress {
- current_sort,
+ current_run_start,
current: max_group_index,
- sort_key,
+ group_key,
};
Ok(())
}
/// Returns retained heap allocations, excluding the inline descriptor
- /// already counted by [`super::GroupOrdering::size`].
+ /// already counted by [`super::GroupClustering::size`].
pub(crate) fn heap_size(&self) -> usize {
- self.order_indices.allocated_size() + self.state.heap_size()
+ self.grouping_indices.allocated_size() + self.state.heap_size()
}
}
@@ -279,79 +277,85 @@ mod tests {
use arrow::array::Int32Array;
- #[test]
- fn test_group_ordering_partial() -> Result<()> {
- // Ordered on column a
- let order_indices = vec![0];
- let mut group_ordering = GroupOrderingPartial::try_new(order_indices)?;
+ #[rstest::rstest]
+ #[case::sorted([1, 2, 3, 4])]
+ #[case::clustered([3, 1, 4, 2])]
+ fn test_group_clustering_partial(#[case] keys: [i32; 4]) -> Result<()> {
+ let [first, second, third, fourth] = keys;
+ // Contiguous on column a.
+ let grouping_indices = vec![0];
+ let mut group_clustering = GroupClusteringPartial::try_new(grouping_indices)?;
let batch_group_values: Vec<ArrayRef> = vec![
- Arc::new(Int32Array::from(vec![1, 2, 3])),
+ Arc::new(Int32Array::from(vec![first, second, third])),
Arc::new(Int32Array::from(vec![2, 1, 3])),
];
let group_indices = vec![0, 1, 2];
let total_num_groups = 3;
- group_ordering.new_groups(
+ group_clustering.new_groups(
&batch_group_values,
&group_indices,
total_num_groups,
)?;
assert_eq!(
- group_ordering.state,
+ group_clustering.state,
State::InProgress {
- current_sort: 2,
- sort_key: vec![ScalarValue::Int32(Some(3))],
+ current_run_start: 2,
+ group_key: vec![ScalarValue::Int32(Some(third))],
current: 2
}
);
+ assert_eq!(group_clustering.emit_to(), Some(EmitTo::First(2)));
// push without a boundary
let batch_group_values: Vec<ArrayRef> = vec![
- Arc::new(Int32Array::from(vec![3, 3, 3])),
+ Arc::new(Int32Array::from(vec![third, third, third])),
Arc::new(Int32Array::from(vec![2, 1, 7])),
];
let group_indices = vec![3, 4, 5];
let total_num_groups = 6;
- group_ordering.new_groups(
+ group_clustering.new_groups(
&batch_group_values,
&group_indices,
total_num_groups,
)?;
assert_eq!(
- group_ordering.state,
+ group_clustering.state,
State::InProgress {
- current_sort: 2,
- sort_key: vec![ScalarValue::Int32(Some(3))],
+ current_run_start: 2,
+ group_key: vec![ScalarValue::Int32(Some(third))],
current: 5
}
);
+ assert_eq!(group_clustering.emit_to(), Some(EmitTo::First(2)));
// push with only a boundary to previous batch
let batch_group_values: Vec<ArrayRef> = vec![
- Arc::new(Int32Array::from(vec![4, 4, 4])),
- Arc::new(Int32Array::from(vec![1, 1, 1])),
+ Arc::new(Int32Array::from(vec![fourth, fourth, fourth])),
+ Arc::new(Int32Array::from(vec![1, 2, 3])),
];
let group_indices = vec![6, 7, 8];
let total_num_groups = 9;
- group_ordering.new_groups(
+ group_clustering.new_groups(
&batch_group_values,
&group_indices,
total_num_groups,
)?;
assert_eq!(
- group_ordering.state,
+ group_clustering.state,
State::InProgress {
- current_sort: 6,
- sort_key: vec![ScalarValue::Int32(Some(4))],
+ current_run_start: 6,
+ group_key: vec![ScalarValue::Int32(Some(fourth))],
current: 8
}
);
+ assert_eq!(group_clustering.emit_to(), Some(EmitTo::First(6)));
Ok(())
}
diff --git a/datafusion/physical-plan/src/aggregates/partial_reduce_stream.rs b/datafusion/physical-plan/src/aggregates/partial_reduce_stream.rs
index 54479865f1..759dd9c7a3 100644
--- a/datafusion/physical-plan/src/aggregates/partial_reduce_stream.rs
+++ b/datafusion/physical-plan/src/aggregates/partial_reduce_stream.rs
@@ -27,10 +27,11 @@ use datafusion_execution::{TaskContext, TryEmitter, async_try_stream};
use futures::stream::{Stream, StreamExt};
use super::AggregateExec;
+use super::GroupClusteringMode;
use super::aggregate_hash_table::{AggregateHashTable, PartialReduceMarker};
+use crate::SendableRecordBatchStream;
use crate::metrics::{BaselineMetrics, Count, MetricBuilder, RecordOutput, SpillMetrics};
use crate::stream::{EmptyRecordBatchStream, RecordBatchStreamAdapter};
-use crate::{InputOrderMode, SendableRecordBatchStream};
/// Hash aggregation can combine multiple partial stages before final
/// evaluation. This stream implements the partial-reduce stage.
@@ -109,7 +110,7 @@ impl PartialReduceHashAggregateStream {
partition: usize,
) -> Result<Self> {
debug_assert_eq!(agg.mode, super::AggregateMode::PartialReduce);
- debug_assert_eq!(agg.input_order_mode, InputOrderMode::Linear);
+ debug_assert_eq!(agg.group_clustering_mode, GroupClusteringMode::None);
let schema = Arc::clone(&agg.schema);
let input = agg.input.execute(partition, Arc::clone(context))?;
diff --git a/datafusion/physical-plan/src/aggregates/single_stream.rs b/datafusion/physical-plan/src/aggregates/single_stream.rs
index f27017784b..8d2edb7407 100644
--- a/datafusion/physical-plan/src/aggregates/single_stream.rs
+++ b/datafusion/physical-plan/src/aggregates/single_stream.rs
@@ -30,14 +30,15 @@ use datafusion_execution::{TaskContext, TryEmitter, async_try_stream};
use futures::stream::StreamExt;
use super::aggregate_hash_table::{
- AggregateHashTable, OrderedAggregateTableMetrics, SingleMarker,
+ AggregateHashTable, ClusteredAggregateTableMetrics, SingleMarker,
};
+use super::order::GroupClusteringMode;
use super::spill::AggregateSpill;
use super::{AggregateExec, create_schema};
+use crate::SendableRecordBatchStream;
use crate::aggregates::AggregateMode;
use crate::metrics::{BaselineMetrics, SpillMetrics};
use crate::stream::{ObservedStream, RecordBatchStreamAdapter};
-use crate::{InputOrderMode, SendableRecordBatchStream};
/// Hash aggregation can run the full logical aggregation in one operator. This
/// stream implements the single stage for grouped hash aggregation.
@@ -82,7 +83,7 @@ use crate::{InputOrderMode, SendableRecordBatchStream};
/// 3. Perform a sort-preserving merge of all spill files and feed the merged output
/// into an ordered streaming aggregation, which ensures bounded memory usage and
/// evaluates the final result.
-/// - [`OrderedFinalAggregateStream`](super::ordered_final_stream::OrderedFinalAggregateStream) is reused for the streaming aggregation.
+/// - [`ClusteredFinalAggregateStream`](super::clustered_final_stream::ClusteredFinalAggregateStream) is reused for the streaming aggregation.
///
/// # Optimization: DISTINCT LIMIT Soft Limit
///
@@ -162,7 +163,7 @@ impl SingleHashAggregateStream {
agg.mode,
AggregateMode::Single | AggregateMode::SinglePartitioned
));
- debug_assert_eq!(agg.input_order_mode, InputOrderMode::Linear);
+ debug_assert_eq!(agg.group_clustering_mode, GroupClusteringMode::None);
let schema = Arc::clone(&agg.schema);
let input = agg.input.execute(partition, Arc::clone(context))?;
@@ -193,7 +194,7 @@ impl SingleHashAggregateStream {
context,
partition,
batch_size,
- &InputOrderMode::Linear,
+ &GroupClusteringMode::None,
&state_schema,
spill_metrics,
)?))
@@ -398,7 +399,7 @@ impl Aggregating {
// Construct the replay stream: an ordered final aggregate stream
// over the sort-preserving merge of all spill runs.
- let metrics = OrderedAggregateTableMetrics::from_hash_table(&hash_table);
+ let metrics = ClusteredAggregateTableMetrics::from_hash_table(&hash_table);
drop(hash_table);
reservation.try_resize(0)?;
// The outer ObservedStream counts output; replay only shares compute time.
diff --git a/datafusion/physical-plan/src/aggregates/spill.rs b/datafusion/physical-plan/src/aggregates/spill.rs
index 887be41855..d2f763cfbd 100644
--- a/datafusion/physical-plan/src/aggregates/spill.rs
+++ b/datafusion/physical-plan/src/aggregates/spill.rs
@@ -28,14 +28,15 @@ use datafusion_physical_expr::PhysicalSortExpr;
use datafusion_physical_expr::expressions::Column;
use datafusion_physical_expr_common::sort_expr::LexOrdering;
-use super::aggregate_hash_table::OrderedAggregateTableMetrics;
-use super::ordered_final_stream::OrderedFinalAggregateStream;
+use super::aggregate_hash_table::ClusteredAggregateTableMetrics;
+use super::clustered_final_stream::ClusteredFinalAggregateStream;
+use super::order::GroupClusteringMode;
use super::{AggregateExec, AggregateMode};
+use crate::SendableRecordBatchStream;
use crate::metrics::{BaselineMetrics, SpillMetrics};
use crate::sorts::IncrementalSortIterator;
use crate::sorts::streaming_merge::{SortedSpillFile, StreamingMergeBuilder};
use crate::spill::spill_manager::SpillManager;
-use crate::{InputOrderMode, SendableRecordBatchStream};
/// Spill configuration and accumulated runs of one grouped aggregation stream.
///
@@ -43,7 +44,7 @@ use crate::{InputOrderMode, SendableRecordBatchStream};
/// drains all currently buffered groups as intermediate state (see
/// `take_state_batch` on the aggregate tables), sorts them by the full group
/// key, and writes them to one spill file. After the original input ends, all
-/// files are merged and replayed through an [`OrderedFinalAggregateStream`],
+/// files are merged and replayed through a [`ClusteredFinalAggregateStream`],
/// which merges the states and evaluates the final aggregate values.
pub(super) struct AggregateSpill {
/// Aggregate configuration used to construct the replay stream.
@@ -92,7 +93,7 @@ pub(super) struct AggregateSpill {
/// using the two previously sorted spill files.
/// 2. Build a final aggregation stream:
/// - The input is the SPM stream.
- /// - It reuses `OrderedFinalAggregateStream` for processing.
+ /// - It reuses `ClusteredFinalAggregateStream` for processing.
/// - It returns the final aggregation result directly.
///
/// SPM output Final aggregate output
@@ -126,10 +127,11 @@ impl AggregateSpill {
/// Creates the spill context of a stream, whose spill requests are described
/// as `label`.
///
- /// `input_order_mode` is the order of the stream's input: spill files are
- /// sorted by the already ordered group columns first, followed by the
- /// remaining ones, so that replay keeps the ordering the stream promised.
- /// Fully sorted input aggregates in bounded memory and never spills.
+ /// `group_clustering_mode` determines which group columns are already
+ /// contiguous. Spill files are sorted by those columns first, followed by
+ /// the remaining ones. Existing output sort options are retained so replay
+ /// preserves any advertised ordering as well as group clustering.
+ /// Full group clustering aggregates in bounded memory and never spills.
///
/// `spill_schema` is the schema of the intermediate state batches.
#[expect(clippy::too_many_arguments)]
@@ -139,12 +141,12 @@ impl AggregateSpill {
context: &Arc<TaskContext>,
partition: usize,
batch_size: usize,
- input_order_mode: &InputOrderMode,
+ group_clustering_mode: &GroupClusteringMode,
spill_schema: &SchemaRef,
spill_metrics: SpillMetrics,
) -> Result<Self> {
let mut replay_agg = agg.clone();
- replay_agg.input_order_mode = InputOrderMode::Sorted;
+ replay_agg.group_clustering_mode = GroupClusteringMode::Full;
let group_schema = match agg.mode {
AggregateMode::Final | AggregateMode::FinalPartitioned => {
agg.group_by().group_schema(spill_schema)?
@@ -164,17 +166,16 @@ impl AggregateSpill {
};
let num_group_columns = group_schema.fields().len();
- let ordered_indices: &[usize] = match input_order_mode {
- InputOrderMode::Linear => &[],
- InputOrderMode::PartiallySorted(ordered_indices) => ordered_indices,
- InputOrderMode::Sorted => {
- return internal_err!("{label}: fully ordered input does not spill");
+ let contiguous_indices: &[usize] = match group_clustering_mode {
+ GroupClusteringMode::None => &[],
+ GroupClusteringMode::Partial(contiguous_indices) => contiguous_indices,
+ GroupClusteringMode::Full => {
+ return internal_err!("{label}: fully contiguous groups do not spill");
}
};
- let spill_indices = ordered_indices
- .iter()
- .copied()
- .chain((0..num_group_columns).filter(|idx| !ordered_indices.contains(idx)));
+ let spill_indices = contiguous_indices.iter().copied().chain(
+ (0..num_group_columns).filter(|idx| !contiguous_indices.contains(idx)),
+ );
let output_ordering = agg.cache.output_ordering();
let spill_sort_exprs = spill_indices.map(|idx| {
let output_expr = Column::new(group_schema.field(idx).name(), idx);
@@ -249,11 +250,11 @@ impl AggregateSpill {
}
/// Merges every sorted run, and does the aggregate evaluation with
- /// [`OrderedFinalAggregateStream`].
+ /// [`ClusteredFinalAggregateStream`].
pub(super) fn into_replay_stream(
self,
baseline_metrics: &BaselineMetrics,
- metrics: OrderedAggregateTableMetrics,
+ metrics: ClusteredAggregateTableMetrics,
reservation: MemoryReservation,
) -> Result<SendableRecordBatchStream> {
let Self {
@@ -284,12 +285,12 @@ impl AggregateSpill {
.with_replay_headroom()
.with_intermediate_merge_sizing(Some(min_spill_batch_rows))
.build()?;
- let replay = OrderedFinalAggregateStream::new_with_input_and_metrics(
+ let replay = ClusteredFinalAggregateStream::new_with_input_and_metrics(
&replay_agg,
&context,
partition,
merged,
- &InputOrderMode::Sorted,
+ &GroupClusteringMode::Full,
baseline_metrics.clone(),
metrics,
None,
diff --git a/datafusion/physical-plan/src/recursive_query.rs b/datafusion/physical-plan/src/recursive_query.rs
index 0a56488de8..494887f906 100644
--- a/datafusion/physical-plan/src/recursive_query.rs
+++ b/datafusion/physical-plan/src/recursive_query.rs
@@ -23,7 +23,7 @@ use std::task::{Context, Poll};
use super::work_table::{ReservedBatches, WorkTable};
use crate::aggregates::group_values::{GroupValues, new_group_values};
-use crate::aggregates::order::GroupOrdering;
+use crate::aggregates::order::GroupClustering;
use crate::common::project_plan_to_schema;
use crate::execution_plan::{Boundedness, EmissionType, reset_plan_states};
use crate::metrics::{
@@ -464,7 +464,7 @@ struct DistinctDeduplicator {
impl DistinctDeduplicator {
fn new(schema: SchemaRef, task_context: &TaskContext) -> Result<Self> {
- let group_values = new_group_values(schema, &GroupOrdering::None)?;
+ let group_values = new_group_values(schema, &GroupClustering::None)?;
let reservation = MemoryConsumer::new("RecursiveQueryHashTable")
.register(task_context.memory_pool());
Ok(Self {
diff --git a/datafusion/sqllogictest/test_files/agg_func_substitute.slt b/datafusion/sqllogictest/test_files/agg_func_substitute.slt
index be9749bc95..f926b8d72d 100644
--- a/datafusion/sqllogictest/test_files/agg_func_substitute.slt
+++ b/datafusion/sqllogictest/test_files/agg_func_substitute.slt
@@ -44,9 +44,9 @@ logical_plan
03)----TableScan: multiple_ordered_table projection=[a, c]
physical_plan
01)ProjectionExec: expr=[a@0 as a, nth_value(multiple_ordered_table.c,Int64(1)) ORDER BY [multiple_ordered_table.c ASC NULLS LAST]@1 as result]
-02)--AggregateExec: mode=FinalPartitioned, gby=[a@0 as a], aggr=[nth_value(multiple_ordered_table.c,Int64(1)) ORDER BY [multiple_ordered_table.c ASC NULLS LAST]], ordering_mode=Sorted
+02)--AggregateExec: mode=FinalPartitioned, gby=[a@0 as a], aggr=[nth_value(multiple_ordered_table.c,Int64(1)) ORDER BY [multiple_ordered_table.c ASC NULLS LAST]], group_clustering_mode=Full
03)----RepartitionExec: partitioning=Hash([a@0], 4), input_partitions=4, preserve_order=true, sort_exprs=a@0 ASC NULLS LAST
-04)------AggregateExec: mode=Partial, gby=[a@0 as a], aggr=[nth_value(multiple_ordered_table.c,Int64(1)) ORDER BY [multiple_ordered_table.c ASC NULLS LAST]], ordering_mode=Sorted
+04)------AggregateExec: mode=Partial, gby=[a@0 as a], aggr=[nth_value(multiple_ordered_table.c,Int64(1)) ORDER BY [multiple_ordered_table.c ASC NULLS LAST]], group_clustering_mode=Full
05)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1, maintains_sort_order=true
06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, c], output_orderings=[[a@0 ASC NULLS LAST], [c@1 ASC NULLS LAST]], file_type=csv, has_header=true
@@ -62,9 +62,9 @@ logical_plan
03)----TableScan: multiple_ordered_table projection=[a, c]
physical_plan
01)ProjectionExec: expr=[a@0 as a, nth_value(multiple_ordered_table.c,Int64(1)) ORDER BY [multiple_ordered_table.c ASC NULLS LAST]@1 as result]
-02)--AggregateExec: mode=FinalPartitioned, gby=[a@0 as a], aggr=[nth_value(multiple_ordered_table.c,Int64(1)) ORDER BY [multiple_ordered_table.c ASC NULLS LAST]], ordering_mode=Sorted
+02)--AggregateExec: mode=FinalPartitioned, gby=[a@0 as a], aggr=[nth_value(multiple_ordered_table.c,Int64(1)) ORDER BY [multiple_ordered_table.c ASC NULLS LAST]], group_clustering_mode=Full
03)----RepartitionExec: partitioning=Hash([a@0], 4), input_partitions=4, preserve_order=true, sort_exprs=a@0 ASC NULLS LAST
-04)------AggregateExec: mode=Partial, gby=[a@0 as a], aggr=[nth_value(multiple_ordered_table.c,Int64(1)) ORDER BY [multiple_ordered_table.c ASC NULLS LAST]], ordering_mode=Sorted
+04)------AggregateExec: mode=Partial, gby=[a@0 as a], aggr=[nth_value(multiple_ordered_table.c,Int64(1)) ORDER BY [multiple_ordered_table.c ASC NULLS LAST]], group_clustering_mode=Full
05)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1, maintains_sort_order=true
06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, c], output_orderings=[[a@0 ASC NULLS LAST], [c@1 ASC NULLS LAST]], file_type=csv, has_header=true
@@ -79,9 +79,9 @@ logical_plan
03)----TableScan: multiple_ordered_table projection=[a, c]
physical_plan
01)ProjectionExec: expr=[a@0 as a, nth_value(multiple_ordered_table.c,Int64(1) + Int64(100)) ORDER BY [multiple_ordered_table.c ASC NULLS LAST]@1 as result]
-02)--AggregateExec: mode=FinalPartitioned, gby=[a@0 as a], aggr=[nth_value(multiple_ordered_table.c, 101) ORDER BY [multiple_ordered_table.c ASC NULLS LAST] as nth_value(multiple_ordered_table.c,Int64(1) + Int64(100)) ORDER BY [multiple_ordered_table.c ASC NULLS LAST]], ordering_mode=Sorted
+02)--AggregateExec: mode=FinalPartitioned, gby=[a@0 as a], aggr=[nth_value(multiple_ordered_table.c, 101) ORDER BY [multiple_ordered_table.c ASC NULLS LAST] as nth_value(multiple_ordered_table.c,Int64(1) + Int64(100)) ORDER BY [multiple_ordered_table.c ASC NULLS LAST]], group_clustering_mode=Full
03)----RepartitionExec: partitioning=Hash([a@0], 4), input_partitions=4, preserve_order=true, sort_exprs=a@0 ASC NULLS LAST
-04)------AggregateExec: mode=Partial, gby=[a@0 as a], aggr=[nth_value(multiple_ordered_table.c, 101) ORDER BY [multiple_ordered_table.c ASC NULLS LAST] as nth_value(multiple_ordered_table.c,Int64(1) + Int64(100)) ORDER BY [multiple_ordered_table.c ASC NULLS LAST]], ordering_mode=Sorted
+04)------AggregateExec: mode=Partial, gby=[a@0 as a], aggr=[nth_value(multiple_ordered_table.c, 101) ORDER BY [multiple_ordered_table.c ASC NULLS LAST] as nth_value(multiple_ordered_table.c,Int64(1) + Int64(100)) ORDER BY [multiple_ordered_table.c ASC NULLS LAST]], group_clustering_mode=Full
05)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1, maintains_sort_order=true
06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, c], output_orderings=[[a@0 ASC NULLS LAST], [c@1 ASC NULLS LAST]], file_type=csv, has_header=true
diff --git a/datafusion/sqllogictest/test_files/aggregate.slt b/datafusion/sqllogictest/test_files/aggregate.slt
index fd7a6256a1..0665b63cc3 100644
--- a/datafusion/sqllogictest/test_files/aggregate.slt
+++ b/datafusion/sqllogictest/test_files/aggregate.slt
@@ -9671,7 +9671,7 @@ CREATE TABLE stream_test (
(3, 1.0, 1.0, 7, false, 'e'), (3, 2.0, 2.0, 8, false, 'f');
# Test comprehensive aggregates with streaming
-# This verifies that CORR and other aggregates work together in a streaming plan (ordering_mode=Sorted)
+# This verifies that CORR and other aggregates work together in a streaming plan (group_clustering_mode=Full)
# Basic Aggregates
query TT
@@ -9710,7 +9710,7 @@ logical_plan
05)--------TableScan: stream_test projection=[g, x, y, i, b]
physical_plan
01)ProjectionExec: expr=[g@0 as g, count(Int64(1))@1 as count(*), sum(stream_test.x)@2 as sum(stream_test.x), avg(stream_test.x)@3 as avg(stream_test.x), avg(stream_test.x)@3 as mean(stream_test.x), min(stream_test.x)@4 as min(stream_test.x), max(stream_test.y)@5 as max(stream_test.y), bit_and(stream_test.i)@6 as bit_and(stream_test.i), bit_or(stream_test.i)@7 as bit_or(stream_test.i), bit_xor(stream_test.i)@8 as bit_xor(stream_test.i), bool_and(stream_test.b)@9 as bool_and(stream_test.b [...]
-02)--AggregateExec: mode=Single, gby=[g@0 as g], aggr=[count(Int64(1)), sum(stream_test.x), avg(stream_test.x), min(stream_test.x), max(stream_test.y), bit_and(stream_test.i), bit_or(stream_test.i), bit_xor(stream_test.i), bool_and(stream_test.b), bool_or(stream_test.b), median(stream_test.x), var(stream_test.x), var_pop(stream_test.x), stddev(stream_test.x), stddev_pop(stream_test.x)], ordering_mode=Sorted
+02)--AggregateExec: mode=Single, gby=[g@0 as g], aggr=[count(Int64(1)), sum(stream_test.x), avg(stream_test.x), min(stream_test.x), max(stream_test.y), bit_and(stream_test.i), bit_or(stream_test.i), bit_xor(stream_test.i), bool_and(stream_test.b), bool_or(stream_test.b), median(stream_test.x), var(stream_test.x), var_pop(stream_test.x), stddev(stream_test.x), stddev_pop(stream_test.x)], group_clustering_mode=Full
03)----SortExec: TopK(fetch=10000), expr=[g@0 ASC NULLS LAST], preserve_partitioning=[false]
04)------DataSourceExec: partitions=1, partition_sizes=[1]
@@ -9765,7 +9765,7 @@ logical_plan
03)----Sort: stream_test.g ASC NULLS LAST, fetch=10000
04)------TableScan: stream_test projection=[g, x]
physical_plan
-01)AggregateExec: mode=Single, gby=[g@0 as g], aggr=[array_agg(stream_test.x) ORDER BY [stream_test.x ASC NULLS LAST], array_agg(DISTINCT stream_test.x) ORDER BY [stream_test.x ASC NULLS LAST], first_value(stream_test.x) ORDER BY [stream_test.x ASC NULLS LAST], last_value(stream_test.x) ORDER BY [stream_test.x ASC NULLS LAST], nth_value(stream_test.x,Int64(1)) ORDER BY [stream_test.x ASC NULLS LAST]], ordering_mode=Sorted
+01)AggregateExec: mode=Single, gby=[g@0 as g], aggr=[array_agg(stream_test.x) ORDER BY [stream_test.x ASC NULLS LAST], array_agg(DISTINCT stream_test.x) ORDER BY [stream_test.x ASC NULLS LAST], first_value(stream_test.x) ORDER BY [stream_test.x ASC NULLS LAST], last_value(stream_test.x) ORDER BY [stream_test.x ASC NULLS LAST], nth_value(stream_test.x,Int64(1)) ORDER BY [stream_test.x ASC NULLS LAST]], group_clustering_mode=Full
02)--SortExec: TopK(fetch=10000), expr=[g@0 ASC NULLS LAST, x@1 ASC NULLS LAST], preserve_partitioning=[false]
03)----DataSourceExec: partitions=1, partition_sizes=[1]
@@ -9802,7 +9802,7 @@ logical_plan
03)----Sort: stream_test.g ASC NULLS LAST, fetch=10000
04)------TableScan: stream_test projection=[g, s]
physical_plan
-01)AggregateExec: mode=Single, gby=[g@0 as g], aggr=[array_agg(stream_test.s) ORDER BY [stream_test.s ASC NULLS LAST], string_agg(stream_test.s,Utf8("|")) ORDER BY [stream_test.s ASC NULLS LAST], string_agg(DISTINCT stream_test.s,Utf8("|")) ORDER BY [stream_test.s ASC NULLS LAST]], ordering_mode=Sorted
+01)AggregateExec: mode=Single, gby=[g@0 as g], aggr=[array_agg(stream_test.s) ORDER BY [stream_test.s ASC NULLS LAST], string_agg(stream_test.s,Utf8("|")) ORDER BY [stream_test.s ASC NULLS LAST], string_agg(DISTINCT stream_test.s,Utf8("|")) ORDER BY [stream_test.s ASC NULLS LAST]], group_clustering_mode=Full
02)--SortExec: TopK(fetch=10000), expr=[g@0 ASC NULLS LAST, s@1 ASC NULLS LAST], preserve_partitioning=[false]
03)----DataSourceExec: partitions=1, partition_sizes=[1]
@@ -9849,7 +9849,7 @@ logical_plan
05)--------TableScan: stream_test projection=[g, x, y]
physical_plan
01)ProjectionExec: expr=[g@0 as g, corr(stream_test.x,stream_test.y)@1 as corr(stream_test.x,stream_test.y), covar_samp(stream_test.x,stream_test.y)@2 as covar(stream_test.x,stream_test.y), covar_samp(stream_test.x,stream_test.y)@2 as covar_samp(stream_test.x,stream_test.y), covar_pop(stream_test.x,stream_test.y)@3 as covar_pop(stream_test.x,stream_test.y), regr_sxx(stream_test.x,stream_test.y)@4 as regr_sxx(stream_test.x,stream_test.y), regr_sxy(stream_test.x,stream_test.y)@5 as regr_sx [...]
-02)--AggregateExec: mode=Single, gby=[g@0 as g], aggr=[corr(stream_test.x,stream_test.y), covar_samp(stream_test.x,stream_test.y), covar_pop(stream_test.x,stream_test.y), regr_sxx(stream_test.x,stream_test.y), regr_sxy(stream_test.x,stream_test.y), regr_syy(stream_test.x,stream_test.y), regr_avgx(stream_test.x,stream_test.y), regr_avgy(stream_test.x,stream_test.y), regr_count(stream_test.x,stream_test.y), regr_slope(stream_test.x,stream_test.y), regr_intercept(stream_test.x,stream_test.y [...]
+02)--AggregateExec: mode=Single, gby=[g@0 as g], aggr=[corr(stream_test.x,stream_test.y), covar_samp(stream_test.x,stream_test.y), covar_pop(stream_test.x,stream_test.y), regr_sxx(stream_test.x,stream_test.y), regr_sxy(stream_test.x,stream_test.y), regr_syy(stream_test.x,stream_test.y), regr_avgx(stream_test.x,stream_test.y), regr_avgy(stream_test.x,stream_test.y), regr_count(stream_test.x,stream_test.y), regr_slope(stream_test.x,stream_test.y), regr_intercept(stream_test.x,stream_test.y [...]
03)----SortExec: TopK(fetch=10000), expr=[g@0 ASC NULLS LAST], preserve_partitioning=[false]
04)------DataSourceExec: partitions=1, partition_sizes=[1]
@@ -9902,7 +9902,7 @@ logical_plan
05)--------TableScan: stream_test projection=[g, x, i]
physical_plan
01)ProjectionExec: expr=[g@0 as g, approx_distinct(stream_test.i)@1 as approx_distinct(stream_test.i), approx_median(stream_test.x)@2 as approx_median(stream_test.x), percentile_cont(Float64(0.5)) WITHIN GROUP [stream_test.x ASC NULLS LAST]@3 as percentile_cont(Float64(0.5)) WITHIN GROUP [stream_test.x ASC NULLS LAST], percentile_cont(Float64(0.5)) WITHIN GROUP [stream_test.x ASC NULLS LAST]@3 as quantile_cont(stream_test.x,Float64(0.5)), approx_percentile_cont(Float64(0.5)) WITHIN GROUP [...]
-02)--AggregateExec: mode=Single, gby=[g@0 as g], aggr=[approx_distinct(stream_test.i), approx_median(stream_test.x), percentile_cont(Float64(0.5)) WITHIN GROUP [stream_test.x ASC NULLS LAST], approx_percentile_cont(Float64(0.5)) WITHIN GROUP [stream_test.x ASC NULLS LAST], approx_percentile_cont_with_weight(Float64(1),Float64(0.5)) WITHIN GROUP [stream_test.x ASC NULLS LAST], percentile_cont(stream_test.x,Float64(0.5)), approx_percentile_cont(stream_test.x,Float64(0.5)), approx_percentil [...]
+02)--AggregateExec: mode=Single, gby=[g@0 as g], aggr=[approx_distinct(stream_test.i), approx_median(stream_test.x), percentile_cont(Float64(0.5)) WITHIN GROUP [stream_test.x ASC NULLS LAST], approx_percentile_cont(Float64(0.5)) WITHIN GROUP [stream_test.x ASC NULLS LAST], approx_percentile_cont_with_weight(Float64(1),Float64(0.5)) WITHIN GROUP [stream_test.x ASC NULLS LAST], percentile_cont(stream_test.x,Float64(0.5)), approx_percentile_cont(stream_test.x,Float64(0.5)), approx_percentil [...]
03)----SortExec: TopK(fetch=10000), expr=[g@0 ASC NULLS LAST], preserve_partitioning=[false]
04)------DataSourceExec: partitions=1, partition_sizes=[1]
diff --git a/datafusion/sqllogictest/test_files/group_by.slt b/datafusion/sqllogictest/test_files/group_by.slt
index 9eb1865ecf..281038f62e 100644
--- a/datafusion/sqllogictest/test_files/group_by.slt
+++ b/datafusion/sqllogictest/test_files/group_by.slt
@@ -2112,7 +2112,7 @@ logical_plan
03)----TableScan: annotated_data_infinite2 projection=[a, b, c]
physical_plan
01)ProjectionExec: expr=[a@1 as a, b@0 as b, sum(annotated_data_infinite2.c)@2 as summation1]
-02)--AggregateExec: mode=Single, gby=[b@1 as b, a@0 as a], aggr=[sum(annotated_data_infinite2.c)], ordering_mode=Sorted
+02)--AggregateExec: mode=Single, gby=[b@1 as b, a@0 as a], aggr=[sum(annotated_data_infinite2.c)], group_clustering_mode=Full
03)----StreamingTableExec: partition_sizes=1, projection=[a, b, c], infinite_source=true, output_ordering=[a@0 ASC NULLS LAST, b@1 ASC NULLS LAST, c@2 ASC NULLS LAST]
@@ -2143,7 +2143,7 @@ logical_plan
03)----TableScan: annotated_data_infinite2 projection=[a, c, d]
physical_plan
01)ProjectionExec: expr=[a@1 as a, d@0 as d, sum(annotated_data_infinite2.c) ORDER BY [annotated_data_infinite2.a DESC NULLS FIRST]@2 as summation1]
-02)--AggregateExec: mode=Single, gby=[d@2 as d, a@0 as a], aggr=[sum(annotated_data_infinite2.c) ORDER BY [annotated_data_infinite2.a DESC NULLS FIRST]], ordering_mode=PartiallySorted([1])
+02)--AggregateExec: mode=Single, gby=[d@2 as d, a@0 as a], aggr=[sum(annotated_data_infinite2.c) ORDER BY [annotated_data_infinite2.a DESC NULLS FIRST]], group_clustering_mode=Partial([1])
03)----StreamingTableExec: partition_sizes=1, projection=[a, c, d], infinite_source=true, output_ordering=[a@0 ASC NULLS LAST]
query III
@@ -2176,7 +2176,7 @@ logical_plan
03)----TableScan: annotated_data_infinite2 projection=[a, b, c]
physical_plan
01)ProjectionExec: expr=[a@0 as a, b@1 as b, first_value(annotated_data_infinite2.c) ORDER BY [annotated_data_infinite2.a DESC NULLS FIRST]@2 as first_c]
-02)--AggregateExec: mode=Single, gby=[a@0 as a, b@1 as b], aggr=[first_value(annotated_data_infinite2.c) ORDER BY [annotated_data_infinite2.a DESC NULLS FIRST]], ordering_mode=Sorted
+02)--AggregateExec: mode=Single, gby=[a@0 as a, b@1 as b], aggr=[first_value(annotated_data_infinite2.c) ORDER BY [annotated_data_infinite2.a DESC NULLS FIRST]], group_clustering_mode=Full
03)----StreamingTableExec: partition_sizes=1, projection=[a, b, c], infinite_source=true, output_ordering=[a@0 ASC NULLS LAST, b@1 ASC NULLS LAST, c@2 ASC NULLS LAST]
query III
@@ -2202,7 +2202,7 @@ logical_plan
03)----TableScan: annotated_data_infinite2 projection=[a, b, c]
physical_plan
01)ProjectionExec: expr=[a@0 as a, b@1 as b, last_value(annotated_data_infinite2.c) ORDER BY [annotated_data_infinite2.a DESC NULLS FIRST]@2 as last_c]
-02)--AggregateExec: mode=Single, gby=[a@0 as a, b@1 as b], aggr=[last_value(annotated_data_infinite2.c) ORDER BY [annotated_data_infinite2.a DESC NULLS FIRST]], ordering_mode=Sorted
+02)--AggregateExec: mode=Single, gby=[a@0 as a, b@1 as b], aggr=[last_value(annotated_data_infinite2.c) ORDER BY [annotated_data_infinite2.a DESC NULLS FIRST]], group_clustering_mode=Full
03)----StreamingTableExec: partition_sizes=1, projection=[a, b, c], infinite_source=true, output_ordering=[a@0 ASC NULLS LAST, b@1 ASC NULLS LAST, c@2 ASC NULLS LAST]
query III
@@ -2229,7 +2229,7 @@ logical_plan
03)----TableScan: annotated_data_infinite2 projection=[a, b, c]
physical_plan
01)ProjectionExec: expr=[a@0 as a, b@1 as b, last_value(annotated_data_infinite2.c)@2 as last_c]
-02)--AggregateExec: mode=Single, gby=[a@0 as a, b@1 as b], aggr=[last_value(annotated_data_infinite2.c)], ordering_mode=Sorted
+02)--AggregateExec: mode=Single, gby=[a@0 as a, b@1 as b], aggr=[last_value(annotated_data_infinite2.c)], group_clustering_mode=Full
03)----StreamingTableExec: partition_sizes=1, projection=[a, b, c], infinite_source=true, output_ordering=[a@0 ASC NULLS LAST, b@1 ASC NULLS LAST, c@2 ASC NULLS LAST]
query III
@@ -2288,7 +2288,7 @@ logical_plan
01)Aggregate: groupBy=[[annotated_data_infinite2.a, annotated_data_infinite2.b]], aggr=[[array_agg(annotated_data_infinite2.d) ORDER BY [annotated_data_infinite2.d ASC NULLS LAST]]]
02)--TableScan: annotated_data_infinite2 projection=[a, b, d]
physical_plan
-01)AggregateExec: mode=Single, gby=[a@0 as a, b@1 as b], aggr=[array_agg(annotated_data_infinite2.d) ORDER BY [annotated_data_infinite2.d ASC NULLS LAST]], ordering_mode=Sorted
+01)AggregateExec: mode=Single, gby=[a@0 as a, b@1 as b], aggr=[array_agg(annotated_data_infinite2.d) ORDER BY [annotated_data_infinite2.d ASC NULLS LAST]], group_clustering_mode=Full
02)--PartialSortExec: expr=[a@0 ASC NULLS LAST, b@1 ASC NULLS LAST, d@2 ASC NULLS LAST], common_prefix_length=[2]
03)----StreamingTableExec: partition_sizes=1, projection=[a, b, d], infinite_source=true, output_ordering=[a@0 ASC NULLS LAST, b@1 ASC NULLS LAST]
@@ -2612,7 +2612,7 @@ logical_plan
05)--------TableScan: sales_global projection=[country, amount]
physical_plan
01)ProjectionExec: expr=[country@0 as country, array_agg(s.amount) ORDER BY [s.amount DESC NULLS FIRST]@1 as amounts, sum(s.amount)@2 as sum1]
-02)--AggregateExec: mode=Single, gby=[country@0 as country], aggr=[array_agg(s.amount) ORDER BY [s.amount DESC NULLS FIRST], sum(s.amount)], ordering_mode=Sorted
+02)--AggregateExec: mode=Single, gby=[country@0 as country], aggr=[array_agg(s.amount) ORDER BY [s.amount DESC NULLS FIRST], sum(s.amount)], group_clustering_mode=Full
03)----SortExec: TopK(fetch=10), expr=[country@0 ASC NULLS LAST, amount@1 DESC], preserve_partitioning=[false]
04)------DataSourceExec: partitions=1, partition_sizes=[1]
@@ -2650,7 +2650,7 @@ logical_plan
05)--------TableScan: sales_global projection=[zip_code, country, amount]
physical_plan
01)ProjectionExec: expr=[country@0 as country, zip_code@1 as zip_code, array_agg(s.amount) ORDER BY [s.amount DESC NULLS FIRST]@2 as amounts, sum(s.amount)@3 as sum1]
-02)--AggregateExec: mode=Single, gby=[country@1 as country, zip_code@0 as zip_code], aggr=[array_agg(s.amount) ORDER BY [s.amount DESC NULLS FIRST], sum(s.amount)], ordering_mode=PartiallySorted([0])
+02)--AggregateExec: mode=Single, gby=[country@1 as country, zip_code@0 as zip_code], aggr=[array_agg(s.amount) ORDER BY [s.amount DESC NULLS FIRST], sum(s.amount)], group_clustering_mode=Partial([0])
03)----SortExec: TopK(fetch=10), expr=[country@1 ASC NULLS LAST, amount@2 DESC], preserve_partitioning=[false]
04)------DataSourceExec: partitions=1, partition_sizes=[1]
@@ -2687,7 +2687,7 @@ logical_plan
05)--------TableScan: sales_global projection=[country, amount]
physical_plan
01)ProjectionExec: expr=[country@0 as country, array_agg(s.amount) ORDER BY [s.country DESC NULLS FIRST]@1 as amounts, sum(s.amount)@2 as sum1]
-02)--AggregateExec: mode=Single, gby=[country@0 as country], aggr=[array_agg(s.amount) ORDER BY [s.country DESC NULLS FIRST], sum(s.amount)], ordering_mode=Sorted
+02)--AggregateExec: mode=Single, gby=[country@0 as country], aggr=[array_agg(s.amount) ORDER BY [s.country DESC NULLS FIRST], sum(s.amount)], group_clustering_mode=Full
03)----SortExec: TopK(fetch=10), expr=[country@0 ASC NULLS LAST], preserve_partitioning=[false]
04)------DataSourceExec: partitions=1, partition_sizes=[1]
@@ -2723,7 +2723,7 @@ logical_plan
05)--------TableScan: sales_global projection=[country, amount]
physical_plan
01)ProjectionExec: expr=[country@0 as country, array_agg(s.amount) ORDER BY [s.country DESC NULLS FIRST, s.amount DESC NULLS FIRST]@1 as amounts, sum(s.amount)@2 as sum1]
-02)--AggregateExec: mode=Single, gby=[country@0 as country], aggr=[array_agg(s.amount) ORDER BY [s.country DESC NULLS FIRST, s.amount DESC NULLS FIRST], sum(s.amount)], ordering_mode=Sorted
+02)--AggregateExec: mode=Single, gby=[country@0 as country], aggr=[array_agg(s.amount) ORDER BY [s.country DESC NULLS FIRST, s.amount DESC NULLS FIRST], sum(s.amount)], group_clustering_mode=Full
03)----SortExec: TopK(fetch=10), expr=[country@0 ASC NULLS LAST, amount@1 DESC], preserve_partitioning=[false]
04)------DataSourceExec: partitions=1, partition_sizes=[1]
@@ -4027,7 +4027,7 @@ logical_plan
12)------------------TableScan: multiple_ordered_table projection=[a, d]
physical_plan
01)ProjectionExec: expr=[last_value(l.d) ORDER BY [l.a ASC NULLS LAST]@1 as amount_usd]
-02)--AggregateExec: mode=Single, gby=[row_n@2 as row_n], aggr=[last_value(l.d) ORDER BY [l.a ASC NULLS LAST]], ordering_mode=Sorted
+02)--AggregateExec: mode=Single, gby=[row_n@2 as row_n], aggr=[last_value(l.d) ORDER BY [l.a ASC NULLS LAST]], group_clustering_mode=Full
03)----HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(d@1, d@1)], filter=CAST(a@0 AS Int64) >= CAST(a@1 AS Int64) - 10, projection=[a@0, d@1, row_n@4]
04)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, d], output_ordering=[a@0 ASC NULLS LAST], file_type=csv, has_header=true
05)------ProjectionExec: expr=[a@0 as a, d@1 as d, row_number() ORDER BY [r.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@2 as row_n]
@@ -4067,9 +4067,9 @@ logical_plan
01)Aggregate: groupBy=[[multiple_ordered_table_with_pk.c, multiple_ordered_table_with_pk.b]], aggr=[[sum(CAST(multiple_ordered_table_with_pk.d AS Int64))]]
02)--TableScan: multiple_ordered_table_with_pk projection=[b, c, d]
physical_plan
-01)AggregateExec: mode=FinalPartitioned, gby=[c@0 as c, b@1 as b], aggr=[sum(multiple_ordered_table_with_pk.d)], ordering_mode=PartiallySorted([0])
+01)AggregateExec: mode=FinalPartitioned, gby=[c@0 as c, b@1 as b], aggr=[sum(multiple_ordered_table_with_pk.d)], group_clustering_mode=Partial([0])
02)--RepartitionExec: partitioning=Hash([c@0, b@1], 8), input_partitions=8, preserve_order=true, sort_exprs=c@0 ASC NULLS LAST
-03)----AggregateExec: mode=Partial, gby=[c@1 as c, b@0 as b], aggr=[sum(multiple_ordered_table_with_pk.d)], ordering_mode=PartiallySorted([0])
+03)----AggregateExec: mode=Partial, gby=[c@1 as c, b@0 as b], aggr=[sum(multiple_ordered_table_with_pk.d)], group_clustering_mode=Partial([0])
04)------RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1, maintains_sort_order=true
05)--------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[b, c, d], output_ordering=[c@1 ASC NULLS LAST], constraints=[PrimaryKey([3])], file_type=csv, has_header=true
@@ -4106,9 +4106,9 @@ logical_plan
01)Aggregate: groupBy=[[multiple_ordered_table_with_pk.c, multiple_ordered_table_with_pk.b]], aggr=[[sum(CAST(multiple_ordered_table_with_pk.d AS Int64))]]
02)--TableScan: multiple_ordered_table_with_pk projection=[b, c, d]
physical_plan
-01)AggregateExec: mode=FinalPartitioned, gby=[c@0 as c, b@1 as b], aggr=[sum(multiple_ordered_table_with_pk.d)], ordering_mode=PartiallySorted([0])
+01)AggregateExec: mode=FinalPartitioned, gby=[c@0 as c, b@1 as b], aggr=[sum(multiple_ordered_table_with_pk.d)], group_clustering_mode=Partial([0])
02)--RepartitionExec: partitioning=Hash([c@0, b@1], 8), input_partitions=8, preserve_order=true, sort_exprs=c@0 ASC NULLS LAST
-03)----AggregateExec: mode=Partial, gby=[c@1 as c, b@0 as b], aggr=[sum(multiple_ordered_table_with_pk.d)], ordering_mode=PartiallySorted([0])
+03)----AggregateExec: mode=Partial, gby=[c@1 as c, b@0 as b], aggr=[sum(multiple_ordered_table_with_pk.d)], group_clustering_mode=Partial([0])
04)------RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1, maintains_sort_order=true
05)--------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[b, c, d], output_ordering=[c@1 ASC NULLS LAST], constraints=[PrimaryKey([3])], file_type=csv, has_header=true
@@ -4129,9 +4129,9 @@ logical_plan
03)----Aggregate: groupBy=[[multiple_ordered_table_with_pk.c]], aggr=[[sum(CAST(multiple_ordered_table_with_pk.d AS Int64))]]
04)------TableScan: multiple_ordered_table_with_pk projection=[c, d]
physical_plan
-01)AggregateExec: mode=Single, gby=[c@0 as c, sum1@1 as sum1], aggr=[], ordering_mode=PartiallySorted([0])
+01)AggregateExec: mode=Single, gby=[c@0 as c, sum1@1 as sum1], aggr=[], group_clustering_mode=Partial([0])
02)--ProjectionExec: expr=[c@0 as c, sum(multiple_ordered_table_with_pk.d)@1 as sum1]
-03)----AggregateExec: mode=Single, gby=[c@0 as c], aggr=[sum(multiple_ordered_table_with_pk.d)], ordering_mode=Sorted
+03)----AggregateExec: mode=Single, gby=[c@0 as c], aggr=[sum(multiple_ordered_table_with_pk.d)], group_clustering_mode=Full
04)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[c, d], output_ordering=[c@0 ASC NULLS LAST], constraints=[PrimaryKey([3])], file_type=csv, has_header=true
query TT
@@ -4151,7 +4151,7 @@ physical_plan
01)ProjectionExec: expr=[c@0 as c, sum1@2 as sum1, sum(multiple_ordered_table_with_pk.b) ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING@3 as sumb]
02)--WindowAggExec: wdw=[sum(multiple_ordered_table_with_pk.b) ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING: Ok(Field { name: "sum(multiple_ordered_table_with_pk.b) ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING", data_type: Int64, nullable: true }), frame: WindowFrame { units: Rows, start_bound: Preceding(UInt64(NULL)), end_bound: Following(UInt64(NULL)), is_causal: false }]
03)----ProjectionExec: expr=[c@0 as c, b@1 as b, sum(multiple_ordered_table_with_pk.d)@2 as sum1]
-04)------AggregateExec: mode=Single, gby=[c@1 as c, b@0 as b], aggr=[sum(multiple_ordered_table_with_pk.d)], ordering_mode=PartiallySorted([0])
+04)------AggregateExec: mode=Single, gby=[c@1 as c, b@0 as b], aggr=[sum(multiple_ordered_table_with_pk.d)], group_clustering_mode=Partial([0])
05)--------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[b, c, d], output_ordering=[c@1 ASC NULLS LAST], constraints=[PrimaryKey([3])], file_type=csv, has_header=true
query TT
@@ -4180,10 +4180,10 @@ logical_plan
physical_plan
01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(b@1, b@1)], projection=[c@0, c@3, sum1@2, sum1@5]
02)--ProjectionExec: expr=[c@0 as c, b@1 as b, sum(multiple_ordered_table_with_pk.d)@2 as sum1]
-03)----AggregateExec: mode=Single, gby=[c@1 as c, b@0 as b], aggr=[sum(multiple_ordered_table_with_pk.d)], ordering_mode=PartiallySorted([0])
+03)----AggregateExec: mode=Single, gby=[c@1 as c, b@0 as b], aggr=[sum(multiple_ordered_table_with_pk.d)], group_clustering_mode=Partial([0])
04)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[b, c, d], output_ordering=[c@1 ASC NULLS LAST], constraints=[PrimaryKey([3])], file_type=csv, has_header=true
05)--ProjectionExec: expr=[c@0 as c, b@1 as b, sum(multiple_ordered_table_with_pk.d)@2 as sum1]
-06)----AggregateExec: mode=Single, gby=[c@1 as c, b@0 as b], aggr=[sum(multiple_ordered_table_with_pk.d)], ordering_mode=PartiallySorted([0])
+06)----AggregateExec: mode=Single, gby=[c@1 as c, b@0 as b], aggr=[sum(multiple_ordered_table_with_pk.d)], group_clustering_mode=Partial([0])
07)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[b, c, d], output_ordering=[c@1 ASC NULLS LAST], constraints=[PrimaryKey([3])], file_type=csv, has_header=true
query TT
@@ -4212,10 +4212,10 @@ physical_plan
01)ProjectionExec: expr=[c@0 as c, c@2 as c, sum1@1 as sum1, sum1@3 as sum1]
02)--CrossJoinExec
03)----ProjectionExec: expr=[c@0 as c, sum(multiple_ordered_table_with_pk.d)@1 as sum1]
-04)------AggregateExec: mode=Single, gby=[c@0 as c], aggr=[sum(multiple_ordered_table_with_pk.d)], ordering_mode=Sorted
+04)------AggregateExec: mode=Single, gby=[c@0 as c], aggr=[sum(multiple_ordered_table_with_pk.d)], group_clustering_mode=Full
05)--------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[c, d], output_ordering=[c@0 ASC NULLS LAST], constraints=[PrimaryKey([3])], file_type=csv, has_header=true
06)----ProjectionExec: expr=[c@0 as c, sum(multiple_ordered_table_with_pk.d)@1 as sum1]
-07)------AggregateExec: mode=Single, gby=[c@0 as c], aggr=[sum(multiple_ordered_table_with_pk.d)], ordering_mode=Sorted
+07)------AggregateExec: mode=Single, gby=[c@0 as c], aggr=[sum(multiple_ordered_table_with_pk.d)], group_clustering_mode=Full
08)--------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[c, d], output_ordering=[c@0 ASC NULLS LAST], constraints=[PrimaryKey([3])], file_type=csv, has_header=true
# we do not generate physical plan for Repartition yet (e.g Distribute By queries).
@@ -4254,10 +4254,10 @@ logical_plan
physical_plan
01)UnionExec
02)--ProjectionExec: expr=[c@0 as c, a@1 as a, sum(multiple_ordered_table_with_pk.d)@2 as sum1]
-03)----AggregateExec: mode=Single, gby=[c@1 as c, a@0 as a], aggr=[sum(multiple_ordered_table_with_pk.d)], ordering_mode=Sorted
+03)----AggregateExec: mode=Single, gby=[c@1 as c, a@0 as a], aggr=[sum(multiple_ordered_table_with_pk.d)], group_clustering_mode=Full
04)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, c, d], output_orderings=[[a@0 ASC NULLS LAST], [c@1 ASC NULLS LAST]], constraints=[PrimaryKey([3])], file_type=csv, has_header=true
05)--ProjectionExec: expr=[c@0 as c, a@1 as a, sum(multiple_ordered_table_with_pk.d)@2 as sum1]
-06)----AggregateExec: mode=Single, gby=[c@1 as c, a@0 as a], aggr=[sum(multiple_ordered_table_with_pk.d)], ordering_mode=Sorted
+06)----AggregateExec: mode=Single, gby=[c@1 as c, a@0 as a], aggr=[sum(multiple_ordered_table_with_pk.d)], group_clustering_mode=Full
07)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, c, d], output_orderings=[[a@0 ASC NULLS LAST], [c@1 ASC NULLS LAST]], constraints=[PrimaryKey([3])], file_type=csv, has_header=true
# table scan should be simplified.
@@ -4272,7 +4272,7 @@ logical_plan
03)----TableScan: multiple_ordered_table_with_pk projection=[a, c, d]
physical_plan
01)ProjectionExec: expr=[c@0 as c, a@1 as a, sum(multiple_ordered_table_with_pk.d)@2 as sum1]
-02)--AggregateExec: mode=Single, gby=[c@1 as c, a@0 as a], aggr=[sum(multiple_ordered_table_with_pk.d)], ordering_mode=Sorted
+02)--AggregateExec: mode=Single, gby=[c@1 as c, a@0 as a], aggr=[sum(multiple_ordered_table_with_pk.d)], group_clustering_mode=Full
03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, c, d], output_orderings=[[a@0 ASC NULLS LAST], [c@1 ASC NULLS LAST]], constraints=[PrimaryKey([3])], file_type=csv, has_header=true
# limit should be simplified
@@ -4291,7 +4291,7 @@ logical_plan
physical_plan
01)ProjectionExec: expr=[c@0 as c, a@1 as a, sum(multiple_ordered_table_with_pk.d)@2 as sum1]
02)--GlobalLimitExec: skip=0, fetch=5
-03)----AggregateExec: mode=Single, gby=[c@1 as c, a@0 as a], aggr=[sum(multiple_ordered_table_with_pk.d)], ordering_mode=Sorted
+03)----AggregateExec: mode=Single, gby=[c@1 as c, a@0 as a], aggr=[sum(multiple_ordered_table_with_pk.d)], group_clustering_mode=Full
04)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, c, d], output_orderings=[[a@0 ASC NULLS LAST], [c@1 ASC NULLS LAST]], constraints=[PrimaryKey([3])], file_type=csv, has_header=true
statement ok
@@ -4374,9 +4374,9 @@ logical_plan
physical_plan
01)SortPreservingMergeExec: [time_chunks@0 DESC], fetch=5
02)--ProjectionExec: expr=[date_bin(Utf8("15 minutes"),unbounded_csv_with_timestamps.ts)@0 as time_chunks]
-03)----AggregateExec: mode=FinalPartitioned, gby=[date_bin(Utf8("15 minutes"),unbounded_csv_with_timestamps.ts)@0 as date_bin(Utf8("15 minutes"),unbounded_csv_with_timestamps.ts)], aggr=[], ordering_mode=Sorted
+03)----AggregateExec: mode=FinalPartitioned, gby=[date_bin(Utf8("15 minutes"),unbounded_csv_with_timestamps.ts)@0 as date_bin(Utf8("15 minutes"),unbounded_csv_with_timestamps.ts)], aggr=[], group_clustering_mode=Full
04)------RepartitionExec: partitioning=Hash([date_bin(Utf8("15 minutes"),unbounded_csv_with_timestamps.ts)@0], 8), input_partitions=8, preserve_order=true, sort_exprs=date_bin(Utf8("15 minutes"),unbounded_csv_with_timestamps.ts)@0 DESC
-05)--------AggregateExec: mode=Partial, gby=[date_bin(IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 900000000000 }, ts@0) as date_bin(Utf8("15 minutes"),unbounded_csv_with_timestamps.ts)], aggr=[], ordering_mode=Sorted
+05)--------AggregateExec: mode=Partial, gby=[date_bin(IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 900000000000 }, ts@0) as date_bin(Utf8("15 minutes"),unbounded_csv_with_timestamps.ts)], aggr=[], group_clustering_mode=Full
06)----------RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1, maintains_sort_order=true
07)------------StreamingTableExec: partition_sizes=1, projection=[ts], infinite_source=true, output_ordering=[ts@0 DESC]
@@ -5099,7 +5099,7 @@ logical_plan
02)--Aggregate: groupBy=[[multiple_ordered_table.a, multiple_ordered_table.b]], aggr=[[array_agg(multiple_ordered_table.c) ORDER BY [multiple_ordered_table.c DESC NULLS FIRST]]]
03)----TableScan: multiple_ordered_table projection=[a, b, c]
physical_plan
-01)AggregateExec: mode=Single, gby=[a@0 as a, b@1 as b], aggr=[array_agg(multiple_ordered_table.c) ORDER BY [multiple_ordered_table.c DESC NULLS FIRST]], ordering_mode=Sorted
+01)AggregateExec: mode=Single, gby=[a@0 as a, b@1 as b], aggr=[array_agg(multiple_ordered_table.c) ORDER BY [multiple_ordered_table.c DESC NULLS FIRST]], group_clustering_mode=Full
02)--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
query II?
diff --git a/datafusion/sqllogictest/test_files/joins.slt b/datafusion/sqllogictest/test_files/joins.slt
index e922233fd8..6f498a9925 100644
--- a/datafusion/sqllogictest/test_files/joins.slt
+++ b/datafusion/sqllogictest/test_files/joins.slt
@@ -3510,7 +3510,7 @@ logical_plan
08)----------TableScan: annotated_data projection=[a, b]
physical_plan
01)ProjectionExec: expr=[a@0 as a, last_value(r.b) ORDER BY [r.a ASC NULLS FIRST]@3 as last_col1]
-02)--AggregateExec: mode=Single, gby=[a@0 as a, b@1 as b, c@2 as c], aggr=[last_value(r.b) ORDER BY [r.a ASC NULLS FIRST]], ordering_mode=PartiallySorted([0])
+02)--AggregateExec: mode=Single, gby=[a@0 as a, b@1 as b, c@2 as c], aggr=[last_value(r.b) ORDER BY [r.a ASC NULLS FIRST]], group_clustering_mode=Partial([0])
03)----HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, a@0)]
04)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, b, c], output_ordering=[a@0 ASC, b@1 ASC NULLS LAST, c@2 ASC NULLS LAST], file_type=csv, has_header=true
05)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, b], output_ordering=[a@0 ASC, b@1 ASC NULLS LAST], file_type=csv, has_header=true
@@ -3557,7 +3557,7 @@ logical_plan
12)------------------TableScan: multiple_ordered_table projection=[a, d]
physical_plan
01)ProjectionExec: expr=[last_value(l.d) ORDER BY [l.a ASC NULLS LAST]@1 as amount_usd]
-02)--AggregateExec: mode=Single, gby=[row_n@2 as row_n], aggr=[last_value(l.d) ORDER BY [l.a ASC NULLS LAST]], ordering_mode=Sorted
+02)--AggregateExec: mode=Single, gby=[row_n@2 as row_n], aggr=[last_value(l.d) ORDER BY [l.a ASC NULLS LAST]], group_clustering_mode=Full
03)----HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(d@1, d@1)], filter=CAST(a@0 AS Int64) >= CAST(a@1 AS Int64) - 10, projection=[a@0, d@1, row_n@4]
04)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, d], output_ordering=[a@0 ASC NULLS LAST], file_type=csv, has_header=true
05)------ProjectionExec: expr=[a@0 as a, d@1 as d, row_number() ORDER BY [r.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@2 as row_n]
@@ -3592,9 +3592,9 @@ logical_plan
physical_plan
01)SortPreservingMergeExec: [a@0 ASC]
02)--ProjectionExec: expr=[a@0 as a, last_value(r.b) ORDER BY [r.a ASC NULLS FIRST]@3 as last_col1]
-03)----AggregateExec: mode=FinalPartitioned, gby=[a@0 as a, b@1 as b, c@2 as c], aggr=[last_value(r.b) ORDER BY [r.a ASC NULLS FIRST]], ordering_mode=PartiallySorted([0])
+03)----AggregateExec: mode=FinalPartitioned, gby=[a@0 as a, b@1 as b, c@2 as c], aggr=[last_value(r.b) ORDER BY [r.a ASC NULLS FIRST]], group_clustering_mode=Partial([0])
04)------RepartitionExec: partitioning=Hash([a@0, b@1, c@2], 2), input_partitions=2, preserve_order=true, sort_exprs=a@0 ASC
-05)--------AggregateExec: mode=Partial, gby=[a@0 as a, b@1 as b, c@2 as c], aggr=[last_value(r.b) ORDER BY [r.a ASC NULLS FIRST]], ordering_mode=PartiallySorted([0])
+05)--------AggregateExec: mode=Partial, gby=[a@0 as a, b@1 as b, c@2 as c], aggr=[last_value(r.b) ORDER BY [r.a ASC NULLS FIRST]], group_clustering_mode=Partial([0])
06)----------HashJoinExec: mode=Partitioned, join_type=Inner, on=[(a@0, a@0)]
07)------------RepartitionExec: partitioning=Hash([a@0], 2), input_partitions=1, maintains_sort_order=true
08)--------------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, b, c], output_ordering=[a@0 ASC, b@1 ASC NULLS LAST, c@2 ASC NULLS LAST], file_type=csv, has_header=true
diff --git a/datafusion/sqllogictest/test_files/order.slt b/datafusion/sqllogictest/test_files/order.slt
index c8c1cd2210..3a2c7ccb84 100644
--- a/datafusion/sqllogictest/test_files/order.slt
+++ b/datafusion/sqllogictest/test_files/order.slt
@@ -1953,7 +1953,7 @@ EXPLAIN SELECT c1, SUM(c2) as sum_c2 FROM table_with_ordered_not_null GROUP BY c
----
physical_plan
01)ProjectionExec: expr=[c1@0 as c1, sum(table_with_ordered_not_null.c2)@1 as sum_c2]
-02)--AggregateExec: mode=Single, gby=[c1@0 as c1], aggr=[sum(table_with_ordered_not_null.c2)], ordering_mode=Sorted
+02)--AggregateExec: mode=Single, gby=[c1@0 as c1], aggr=[sum(table_with_ordered_not_null.c2)], group_clustering_mode=Full
03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/aggregate_agg_multi_order.csv]]}, projection=[c1, c2], output_ordering=[c1@0 ASC NULLS LAST], file_type=csv, has_header=true
statement ok
diff --git a/datafusion/sqllogictest/test_files/ordered_aggregate_spill.slt b/datafusion/sqllogictest/test_files/ordered_aggregate_spill.slt
index 548dfc69b1..6e0c72e58e 100644
--- a/datafusion/sqllogictest/test_files/ordered_aggregate_spill.slt
+++ b/datafusion/sqllogictest/test_files/ordered_aggregate_spill.slt
@@ -55,9 +55,9 @@ FROM generate_series(20000) AS t1(v1)
GROUP BY v1
----
Plan with Metrics
-01)AggregateExec: mode=FinalPartitioned,<slt:ignore>ordering_mode=Sorted, metrics=[<slt:ignore>spill_count=0,<slt:ignore>]
+01)AggregateExec: mode=FinalPartitioned,<slt:ignore>group_clustering_mode=Full, metrics=[<slt:ignore>spill_count=0,<slt:ignore>]
02)--RepartitionExec:<slt:ignore>preserve_order=true<slt:ignore>
-03)----AggregateExec: mode=Partial,<slt:ignore>ordering_mode=Sorted, metrics=[<slt:ignore>spill_count=0,<slt:ignore>]
+03)----AggregateExec: mode=Partial,<slt:ignore>group_clustering_mode=Full, metrics=[<slt:ignore>spill_count=0,<slt:ignore>]
<slt:ignore>
query II rowsort
@@ -100,9 +100,9 @@ FROM generate_series(20000) AS t1(v1)
GROUP BY round(v1, -4), v1 % 5000
----
Plan with Metrics
-01)AggregateExec: mode=FinalPartitioned,<slt:ignore>aggr=[sum(t1.v1 * Int64(2))], ordering_mode=PartiallySorted([0]), metrics=[<slt:ignore>spill_count=0,<slt:ignore>]
+01)AggregateExec: mode=FinalPartitioned,<slt:ignore>aggr=[sum(t1.v1 * Int64(2))], group_clustering_mode=Partial([0]), metrics=[<slt:ignore>spill_count=0,<slt:ignore>]
02)--RepartitionExec:<slt:ignore>input_partitions=1, maintains_sort_order=true<slt:ignore>
-03)----AggregateExec: mode=Partial,<slt:ignore>ordering_mode=PartiallySorted([0]), metrics=[<slt:ignore>spill_count=0,<slt:ignore>]
+03)----AggregateExec: mode=Partial,<slt:ignore>group_clustering_mode=Partial([0]), metrics=[<slt:ignore>spill_count=0,<slt:ignore>]
<slt:ignore>
# All rounds should have the same result hash
@@ -124,9 +124,9 @@ FROM generate_series(20000) AS t1(v1)
GROUP BY round(v1, -4), v1 % 5000
----
Plan with Metrics
-01)AggregateExec: mode=FinalPartitioned,<slt:ignore>aggr=[sum(t1.v1 * Int64(2))], ordering_mode=PartiallySorted([0]), metrics=[<slt:ignore>spilled_bytes=<slt:ignore> KB,<slt:ignore>]
+01)AggregateExec: mode=FinalPartitioned,<slt:ignore>aggr=[sum(t1.v1 * Int64(2))], group_clustering_mode=Partial([0]), metrics=[<slt:ignore>spilled_bytes=<slt:ignore> KB,<slt:ignore>]
02)--RepartitionExec:<slt:ignore>input_partitions=1, maintains_sort_order=true<slt:ignore>
-03)----AggregateExec: mode=Partial,<slt:ignore>ordering_mode=PartiallySorted([0]), metrics=[<slt:ignore>spill_count=0,<slt:ignore>]
+03)----AggregateExec: mode=Partial,<slt:ignore>group_clustering_mode=Partial([0]), metrics=[<slt:ignore>spill_count=0,<slt:ignore>]
<slt:ignore>
# All rounds should have the same result hash
@@ -148,9 +148,9 @@ FROM generate_series(20000) AS t1(v1)
GROUP BY round(v1, -4), v1 % 5000
----
Plan with Metrics
-01)AggregateExec: mode=FinalPartitioned,<slt:ignore>aggr=[sum(t1.v1 * Int64(2))], ordering_mode=PartiallySorted([0]), metrics=[<slt:ignore>spilled_bytes=<slt:ignore> KB,<slt:ignore>]
+01)AggregateExec: mode=FinalPartitioned,<slt:ignore>aggr=[sum(t1.v1 * Int64(2))], group_clustering_mode=Partial([0]), metrics=[<slt:ignore>spilled_bytes=<slt:ignore> KB,<slt:ignore>]
02)--RepartitionExec:<slt:ignore>input_partitions=1, maintains_sort_order=true<slt:ignore>
-03)----AggregateExec: mode=Partial,<slt:ignore>ordering_mode=PartiallySorted([0]), metrics=[<slt:ignore>spill_count=0,<slt:ignore>]
+03)----AggregateExec: mode=Partial,<slt:ignore>group_clustering_mode=Partial([0]), metrics=[<slt:ignore>spill_count=0,<slt:ignore>]
<slt:ignore>
# All rounds should have the same result hash
@@ -176,9 +176,9 @@ FROM generate_series(20000) AS t1(v1)
GROUP BY round(v1, -4), v1 % 5000
----
Plan with Metrics
-01)AggregateExec: mode=FinalPartitioned,<slt:ignore>aggr=[sum(t1.v1 * Int64(2)), min(t1.v1 % Int64(2))], ordering_mode=PartiallySorted([0]), metrics=[<slt:ignore>spilled_rows=<slt:ignore> K,<slt:ignore>]
+01)AggregateExec: mode=FinalPartitioned,<slt:ignore>aggr=[sum(t1.v1 * Int64(2)), min(t1.v1 % Int64(2))], group_clustering_mode=Partial([0]), metrics=[<slt:ignore>spilled_rows=<slt:ignore> K,<slt:ignore>]
02)--RepartitionExec:<slt:ignore>input_partitions=1, maintains_sort_order=true<slt:ignore>
-03)----AggregateExec: mode=Partial,<slt:ignore>aggr=[sum(t1.v1 * Int64(2)), min(t1.v1 % Int64(2))], ordering_mode=PartiallySorted([0]), metrics=[<slt:ignore>spill_count=0,<slt:ignore>]
+03)----AggregateExec: mode=Partial,<slt:ignore>aggr=[sum(t1.v1 * Int64(2)), min(t1.v1 % Int64(2))], group_clustering_mode=Partial([0]), metrics=[<slt:ignore>spill_count=0,<slt:ignore>]
<slt:ignore>
# ==================================================================================
@@ -201,7 +201,7 @@ FROM generate_series(20000) AS t1(v1)
GROUP BY round(v1, -4), v1 % 5000
----
Plan with Metrics
-01)AggregateExec: mode=Single,<slt:ignore>aggr=[min(t1.v1 * Int64(2)), avg(t1.v1)], ordering_mode=PartiallySorted([0]), metrics=[<slt:ignore>spill_count=0,<slt:ignore>]
+01)AggregateExec: mode=Single,<slt:ignore>aggr=[min(t1.v1 * Int64(2)), avg(t1.v1)], group_clustering_mode=Partial([0]), metrics=[<slt:ignore>spill_count=0,<slt:ignore>]
<slt:ignore>
query IIIR rowsort
@@ -222,7 +222,7 @@ FROM generate_series(20000) AS t1(v1)
GROUP BY round(v1, -4), v1 % 5000
----
Plan with Metrics
-01)AggregateExec: mode=Single,<slt:ignore>aggr=[min(t1.v1 * Int64(2)), avg(t1.v1)], ordering_mode=PartiallySorted([0]), metrics=[<slt:ignore>spilled_bytes=<slt:ignore> KB,<slt:ignore>]
+01)AggregateExec: mode=Single,<slt:ignore>aggr=[min(t1.v1 * Int64(2)), avg(t1.v1)], group_clustering_mode=Partial([0]), metrics=[<slt:ignore>spilled_bytes=<slt:ignore> KB,<slt:ignore>]
<slt:ignore>
# Same result hash as the no-spill round above
diff --git a/datafusion/sqllogictest/test_files/preserve_file_partitioning.slt b/datafusion/sqllogictest/test_files/preserve_file_partitioning.slt
index e2dd22cc82..f4aab94ea6 100644
--- a/datafusion/sqllogictest/test_files/preserve_file_partitioning.slt
+++ b/datafusion/sqllogictest/test_files/preserve_file_partitioning.slt
@@ -287,9 +287,9 @@ logical_plan
physical_plan
01)SortPreservingMergeExec: [f_dkey@0 ASC NULLS LAST]
02)--ProjectionExec: expr=[f_dkey@0 as f_dkey, count(Int64(1))@1 as count(*), avg(fact_table_ordered.value)@2 as avg(fact_table_ordered.value)]
-03)----AggregateExec: mode=FinalPartitioned, gby=[f_dkey@0 as f_dkey], aggr=[count(Int64(1)), avg(fact_table_ordered.value)], ordering_mode=Sorted
+03)----AggregateExec: mode=FinalPartitioned, gby=[f_dkey@0 as f_dkey], aggr=[count(Int64(1)), avg(fact_table_ordered.value)], group_clustering_mode=Full
04)------RepartitionExec: partitioning=Hash([f_dkey@0], 3), input_partitions=3, preserve_order=true, sort_exprs=f_dkey@0 ASC NULLS LAST
-05)--------AggregateExec: mode=Partial, gby=[f_dkey@1 as f_dkey], aggr=[count(Int64(1)), avg(fact_table_ordered.value)], ordering_mode=Sorted
+05)--------AggregateExec: mode=Partial, gby=[f_dkey@1 as f_dkey], aggr=[count(Int64(1)), avg(fact_table_ordered.value)], group_clustering_mode=Full
06)----------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=C/data.parquet]]}, projection=[value, f_dkey], output_ordering=[f_dkey@1 ASC NULLS LAST], file_type=parquet
# Verify results without optimization
@@ -319,7 +319,7 @@ logical_plan
physical_plan
01)SortPreservingMergeExec: [f_dkey@0 ASC NULLS LAST]
02)--ProjectionExec: expr=[f_dkey@0 as f_dkey, count(Int64(1))@1 as count(*), avg(fact_table_ordered.value)@2 as avg(fact_table_ordered.value)]
-03)----AggregateExec: mode=SinglePartitioned, gby=[f_dkey@1 as f_dkey], aggr=[count(Int64(1)), avg(fact_table_ordered.value)], ordering_mode=Sorted
+03)----AggregateExec: mode=SinglePartitioned, gby=[f_dkey@1 as f_dkey], aggr=[count(Int64(1)), avg(fact_table_ordered.value)], group_clustering_mode=Full
04)------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=C/data.parquet]]}, projection=[value, f_dkey], output_ordering=[f_dkey@1 ASC NULLS LAST], output_partitioning=Hash( [...]
query TIR
@@ -359,9 +359,9 @@ logical_plan
physical_plan
01)SortPreservingMergeExec: [f_dkey@0 ASC NULLS LAST]
02)--ProjectionExec: expr=[f_dkey@0 as f_dkey, max(d.env)@1 as max(d.env), max(d.service)@2 as max(d.service), count(Int64(1))@3 as count(*), sum(f.value)@4 as sum(f.value)]
-03)----AggregateExec: mode=FinalPartitioned, gby=[f_dkey@0 as f_dkey], aggr=[max(d.env), max(d.service), count(Int64(1)), sum(f.value)], ordering_mode=Sorted
+03)----AggregateExec: mode=FinalPartitioned, gby=[f_dkey@0 as f_dkey], aggr=[max(d.env), max(d.service), count(Int64(1)), sum(f.value)], group_clustering_mode=Full
04)------RepartitionExec: partitioning=Hash([f_dkey@0], 3), input_partitions=3, preserve_order=true, sort_exprs=f_dkey@0 ASC NULLS LAST
-05)--------AggregateExec: mode=Partial, gby=[f_dkey@1 as f_dkey], aggr=[max(d.env), max(d.service), count(Int64(1)), sum(f.value)], ordering_mode=Sorted
+05)--------AggregateExec: mode=Partial, gby=[f_dkey@1 as f_dkey], aggr=[max(d.env), max(d.service), count(Int64(1)), sum(f.value)], group_clustering_mode=Full
06)----------HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(d_dkey@0, f_dkey@1)], projection=[value@3, f_dkey@4, env@1, service@2]
07)------------CoalescePartitionsExec
08)--------------FilterExec: service@2 = log
@@ -412,7 +412,7 @@ logical_plan
physical_plan
01)SortPreservingMergeExec: [f_dkey@0 ASC NULLS LAST]
02)--ProjectionExec: expr=[f_dkey@0 as f_dkey, max(d.env)@1 as max(d.env), max(d.service)@2 as max(d.service), count(Int64(1))@3 as count(*), sum(f.value)@4 as sum(f.value)]
-03)----AggregateExec: mode=SinglePartitioned, gby=[f_dkey@1 as f_dkey], aggr=[max(d.env), max(d.service), count(Int64(1)), sum(f.value)], ordering_mode=Sorted
+03)----AggregateExec: mode=SinglePartitioned, gby=[f_dkey@1 as f_dkey], aggr=[max(d.env), max(d.service), count(Int64(1)), sum(f.value)], group_clustering_mode=Full
04)------HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(d_dkey@0, f_dkey@1)], projection=[value@3, f_dkey@4, env@1, service@2]
05)--------CoalescePartitionsExec
06)----------FilterExec: service@2 = log
diff --git a/datafusion/sqllogictest/test_files/range_sorted_time_bin_agg.slt b/datafusion/sqllogictest/test_files/range_sorted_time_bin_agg.slt
index 18123a492d..400bbab1a3 100644
--- a/datafusion/sqllogictest/test_files/range_sorted_time_bin_agg.slt
+++ b/datafusion/sqllogictest/test_files/range_sorted_time_bin_agg.slt
@@ -35,9 +35,9 @@
# single streaming SinglePartitioned step with no hash shuffle.
#
# Today's plan still hash-repartitions:
-# Partial AggregateExec (ordering_mode=Sorted)
+# Partial AggregateExec (group_clustering_mode=Full)
# -> RepartitionExec Hash([key, date_bin(...)])
-# -> FinalPartitioned AggregateExec (ordering_mode=Sorted)
+# -> FinalPartitioned AggregateExec (group_clustering_mode=Full)
statement ok
set datafusion.explain.physical_plan_only = true;
@@ -86,7 +86,7 @@ physical_plan DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion
# overlap across the two 60-minute streams.
#
# Today this is still Partial + hash RepartitionExec + Final, even though
-# ordering_mode=Sorted is already recognized.
+# group_clustering_mode=Full is already recognized.
##########
query TT
@@ -97,9 +97,9 @@ GROUP BY key, time_bin;
----
physical_plan
01)ProjectionExec: expr=[key@0 as key, date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 60000000000 }"),range_sorted_time_bin.timestamp)@1 as time_bin, sum(range_sorted_time_bin.value)@2 as sum(range_sorted_time_bin.value)]
-02)--AggregateExec: mode=FinalPartitioned, gby=[key@0 as key, date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 60000000000 }"),range_sorted_time_bin.timestamp)@1 as date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 60000000000 }"),range_sorted_time_bin.timestamp)], aggr=[sum(range_sorted_time_bin.value)], ordering_mode=Sorted
+02)--AggregateExec: mode=FinalPartitioned, gby=[key@0 as key, date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 60000000000 }"),range_sorted_time_bin.timestamp)@1 as date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 60000000000 }"),range_sorted_time_bin.timestamp)], aggr=[sum(range_sorted_time_bin.value)], group_clustering_mode=Full
03)----RepartitionExec: partitioning=Hash([key@0, date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 60000000000 }"),range_sorted_time_bin.timestamp)@1], 2), input_partitions=2, preserve_order=true, sort_exprs=key@0 ASC, date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 60000000000 }"),range_sorted_time_bin.timestamp)@1 ASC
-04)------AggregateExec: mode=Partial, gby=[key@0 as key, date_bin(IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 60000000000 }, timestamp@1) as date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 60000000000 }"),range_sorted_time_bin.timestamp)], aggr=[sum(range_sorted_time_bin.value)], ordering_mode=Sorted
+04)------AggregateExec: mode=Partial, gby=[key@0 as key, date_bin(IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 60000000000 }, timestamp@1) as date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 60000000000 }"),range_sorted_time_bin.timestamp)], aggr=[sum(range_sorted_time_bin.value)], group_clustering_mode=Full
05)--------FilterExec: col4@1 = a, projection=[key@0, timestamp@2, value@3]
06)----------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_sorted_time_bin/part-0.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_sorted_time_bin/part-1.parquet]]}, projection=[key, col4, timestamp, value], output_ordering=[key@0 ASC, timestamp@2 ASC], output_partitioning=Range([timestamp@2 ASC], [(1704070800000000000)], 2), file_type=parquet, predicate=col4@4 = a, [...]
@@ -128,9 +128,9 @@ GROUP BY key, time_bin;
----
physical_plan
01)ProjectionExec: expr=[key@0 as key, date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 60000000000 }"),range_sorted_time_bin.timestamp)@1 as time_bin, sum(range_sorted_time_bin.value)@2 as sum(range_sorted_time_bin.value)]
-02)--AggregateExec: mode=FinalPartitioned, gby=[key@0 as key, date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 60000000000 }"),range_sorted_time_bin.timestamp)@1 as date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 60000000000 }"),range_sorted_time_bin.timestamp)], aggr=[sum(range_sorted_time_bin.value)], ordering_mode=Sorted
+02)--AggregateExec: mode=FinalPartitioned, gby=[key@0 as key, date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 60000000000 }"),range_sorted_time_bin.timestamp)@1 as date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 60000000000 }"),range_sorted_time_bin.timestamp)], aggr=[sum(range_sorted_time_bin.value)], group_clustering_mode=Full
03)----RepartitionExec: partitioning=Hash([key@0, date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 60000000000 }"),range_sorted_time_bin.timestamp)@1], 2), input_partitions=2, preserve_order=true, sort_exprs=key@0 ASC, date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 60000000000 }"),range_sorted_time_bin.timestamp)@1 ASC
-04)------AggregateExec: mode=Partial, gby=[key@0 as key, date_bin(IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 60000000000 }, timestamp@1) as date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 60000000000 }"),range_sorted_time_bin.timestamp)], aggr=[sum(range_sorted_time_bin.value)], ordering_mode=Sorted
+04)------AggregateExec: mode=Partial, gby=[key@0 as key, date_bin(IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 60000000000 }, timestamp@1) as date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 60000000000 }"),range_sorted_time_bin.timestamp)], aggr=[sum(range_sorted_time_bin.value)], group_clustering_mode=Full
05)--------DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_sorted_time_bin/part-0.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch_range_partitioning/range_sorted_time_bin/part-1.parquet]]}, projection=[key, timestamp, value], output_ordering=[key@0 ASC, timestamp@1 ASC], output_partitioning=Range([timestamp@1 ASC], [(1704070800000000000)], 2), file_type=parquet
query TPI
diff --git a/datafusion/sqllogictest/test_files/repartition_subset_satisfaction.slt b/datafusion/sqllogictest/test_files/repartition_subset_satisfaction.slt
index 5371ca59be..09bf131e5e 100644
--- a/datafusion/sqllogictest/test_files/repartition_subset_satisfaction.slt
+++ b/datafusion/sqllogictest/test_files/repartition_subset_satisfaction.slt
@@ -161,9 +161,9 @@ logical_plan
physical_plan
01)SortPreservingMergeExec: [f_dkey@0 ASC NULLS LAST, time_bin@1 ASC NULLS LAST]
02)--ProjectionExec: expr=[f_dkey@0 as f_dkey, date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),fact_table_ordered.timestamp)@1 as time_bin, count(Int64(1))@2 as count(*), avg(fact_table_ordered.value)@3 as avg(fact_table_ordered.value)]
-03)----AggregateExec: mode=FinalPartitioned, gby=[f_dkey@0 as f_dkey, date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),fact_table_ordered.timestamp)@1 as date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),fact_table_ordered.timestamp)], aggr=[count(Int64(1)), avg(fact_table_ordered.value)], ordering_mode=Sorted
+03)----AggregateExec: mode=FinalPartitioned, gby=[f_dkey@0 as f_dkey, date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),fact_table_ordered.timestamp)@1 as date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),fact_table_ordered.timestamp)], aggr=[count(Int64(1)), avg(fact_table_ordered.value)], group_clustering_mode=Full
04)------RepartitionExec: partitioning=Hash([f_dkey@0, date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),fact_table_ordered.timestamp)@1], 3), input_partitions=3, preserve_order=true, sort_exprs=f_dkey@0 ASC NULLS LAST, date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),fact_table_ordered.timestamp)@1 ASC NULLS LAST
-05)--------AggregateExec: mode=Partial, gby=[f_dkey@2 as f_dkey, date_bin(IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }, timestamp@0) as date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),fact_table_ordered.timestamp)], aggr=[count(Int64(1)), avg(fact_table_ordered.value)], ordering_mode=Sorted
+05)--------AggregateExec: mode=Partial, gby=[f_dkey@2 as f_dkey, date_bin(IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }, timestamp@0) as date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),fact_table_ordered.timestamp)], aggr=[count(Int64(1)), avg(fact_table_ordered.value)], group_clustering_mode=Full
06)----------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=C/data.parquet]]}, projection=[timestamp, value, f_dkey], output_ordering=[f_dkey@2 ASC NULLS LA [...]
# Verify results without subset satisfaction
@@ -203,7 +203,7 @@ logical_plan
physical_plan
01)SortPreservingMergeExec: [f_dkey@0 ASC NULLS LAST, time_bin@1 ASC NULLS LAST]
02)--ProjectionExec: expr=[f_dkey@0 as f_dkey, date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),fact_table_ordered.timestamp)@1 as time_bin, count(Int64(1))@2 as count(*), avg(fact_table_ordered.value)@3 as avg(fact_table_ordered.value)]
-03)----AggregateExec: mode=SinglePartitioned, gby=[f_dkey@2 as f_dkey, date_bin(IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }, timestamp@0) as date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),fact_table_ordered.timestamp)], aggr=[count(Int64(1)), avg(fact_table_ordered.value)], ordering_mode=Sorted
+03)----AggregateExec: mode=SinglePartitioned, gby=[f_dkey@2 as f_dkey, date_bin(IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }, timestamp@0) as date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),fact_table_ordered.timestamp)], aggr=[count(Int64(1)), avg(fact_table_ordered.value)], group_clustering_mode=Full
04)------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_subset_satisfaction/fact/f_dkey=C/data.parquet]]}, projection=[timestamp, value, f_dkey], output_ordering=[f_dkey@2 ASC NULLS LAST, [...]
# Verify results match with subset satisfaction
@@ -373,9 +373,9 @@ physical_plan
05)--------RepartitionExec: partitioning=Hash([env@0, time_bin@1], 3), input_partitions=3
06)----------AggregateExec: mode=Partial, gby=[env@1 as env, time_bin@0 as time_bin], aggr=[avg(a.max_bin_value)]
07)------------ProjectionExec: expr=[date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),j.timestamp)@1 as time_bin, env@2 as env, max(j.value)@3 as max_bin_value]
-08)--------------AggregateExec: mode=FinalPartitioned, gby=[f_dkey@0 as f_dkey, date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),j.timestamp)@1 as date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),j.timestamp), env@2 as env], aggr=[max(j.value)], ordering_mode=PartiallySorted([0, 1])
+08)--------------AggregateExec: mode=FinalPartitioned, gby=[f_dkey@0 as f_dkey, date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),j.timestamp)@1 as date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),j.timestamp), env@2 as env], aggr=[max(j.value)], group_clustering_mode=Partial([0, 1])
09)----------------RepartitionExec: partitioning=Hash([f_dkey@0, date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),j.timestamp)@1, env@2], 3), input_partitions=3, preserve_order=true, sort_exprs=f_dkey@0 ASC NULLS LAST, date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),j.timestamp)@1 ASC NULLS LAST
-10)------------------AggregateExec: mode=Partial, gby=[f_dkey@0 as f_dkey, date_bin(IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }, timestamp@2) as date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),j.timestamp), env@1 as env], aggr=[max(j.value)], ordering_mode=PartiallySorted([0, 1])
+10)------------------AggregateExec: mode=Partial, gby=[f_dkey@0 as f_dkey, date_bin(IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }, timestamp@2) as date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),j.timestamp), env@1 as env], aggr=[max(j.value)], group_clustering_mode=Partial([0, 1])
11)--------------------HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(d_dkey@1, f_dkey@2)], projection=[f_dkey@4, env@0, timestamp@2, value@3]
12)----------------------CoalescePartitionsExec
13)------------------------FilterExec: service@1 = log, projection=[env@0, d_dkey@2]
@@ -470,7 +470,7 @@ physical_plan
05)--------RepartitionExec: partitioning=Hash([env@0, time_bin@1], 3), input_partitions=3
06)----------AggregateExec: mode=Partial, gby=[env@1 as env, time_bin@0 as time_bin], aggr=[avg(a.max_bin_value)]
07)------------ProjectionExec: expr=[date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),j.timestamp)@1 as time_bin, env@2 as env, max(j.value)@3 as max_bin_value]
-08)--------------AggregateExec: mode=SinglePartitioned, gby=[f_dkey@0 as f_dkey, date_bin(IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }, timestamp@2) as date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),j.timestamp), env@1 as env], aggr=[max(j.value)], ordering_mode=PartiallySorted([0, 1])
+08)--------------AggregateExec: mode=SinglePartitioned, gby=[f_dkey@0 as f_dkey, date_bin(IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }, timestamp@2) as date_bin(IntervalMonthDayNano("IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 30000000000 }"),j.timestamp), env@1 as env], aggr=[max(j.value)], group_clustering_mode=Partial([0, 1])
09)----------------HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(d_dkey@1, f_dkey@2)], projection=[f_dkey@4, env@0, timestamp@2, value@3]
10)------------------CoalescePartitionsExec
11)--------------------FilterExec: service@1 = log, projection=[env@0, d_dkey@2]
diff --git a/datafusion/sqllogictest/test_files/sort_pushdown.slt b/datafusion/sqllogictest/test_files/sort_pushdown.slt
index a173e76d6c..da583f127c 100644
--- a/datafusion/sqllogictest/test_files/sort_pushdown.slt
+++ b/datafusion/sqllogictest/test_files/sort_pushdown.slt
@@ -912,7 +912,7 @@ logical_plan
03)----TableScan: agg_expr_parquet projection=[x, y, v]
physical_plan
01)SortExec: expr=[x@0 ASC NULLS LAST, agg_expr_parquet.y % Int64(2)@1 ASC NULLS LAST], preserve_partitioning=[false]
-02)--AggregateExec: mode=Single, gby=[x@0 as x, CAST(y@1 AS Int64) % 2 as agg_expr_parquet.y % Int64(2)], aggr=[sum(agg_expr_parquet.v)], ordering_mode=PartiallySorted([0])
+02)--AggregateExec: mode=Single, gby=[x@0 as x, CAST(y@1 AS Int64) % 2 as agg_expr_parquet.y % Int64(2)], aggr=[sum(agg_expr_parquet.v)], group_clustering_mode=Partial([0])
03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/agg_expr_sorted.parquet]]}, projection=[x, y, v], output_ordering=[x@0 ASC NULLS LAST, y@1 ASC NULLS LAST], file_type=parquet
# Expected output pattern from ORDER BY [x, bucket]:
@@ -946,7 +946,7 @@ logical_plan
02)--Aggregate: groupBy=[[agg_expr_parquet.x, CAST(agg_expr_parquet.y AS Int64)]], aggr=[[sum(CAST(agg_expr_parquet.v AS Int64))]]
03)----TableScan: agg_expr_parquet projection=[x, y, v]
physical_plan
-01)AggregateExec: mode=Single, gby=[x@0 as x, CAST(y@1 AS Int64) as agg_expr_parquet.y], aggr=[sum(agg_expr_parquet.v)], ordering_mode=Sorted
+01)AggregateExec: mode=Single, gby=[x@0 as x, CAST(y@1 AS Int64) as agg_expr_parquet.y], aggr=[sum(agg_expr_parquet.v)], group_clustering_mode=Full
02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/agg_expr_sorted.parquet]]}, projection=[x, y, v], output_ordering=[x@0 ASC NULLS LAST, y@1 ASC NULLS LAST], file_type=parquet
query III
@@ -978,7 +978,7 @@ logical_plan
03)----TableScan: agg_expr_parquet projection=[x, v]
physical_plan
01)SortExec: expr=[sum(agg_expr_parquet.v)@1 ASC NULLS LAST], preserve_partitioning=[false]
-02)--AggregateExec: mode=Single, gby=[x@0 as x], aggr=[sum(agg_expr_parquet.v)], ordering_mode=Sorted
+02)--AggregateExec: mode=Single, gby=[x@0 as x], aggr=[sum(agg_expr_parquet.v)], group_clustering_mode=Full
03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/agg_expr_sorted.parquet]]}, projection=[x, v], output_ordering=[x@0 ASC NULLS LAST], file_type=parquet
query II
@@ -1034,7 +1034,7 @@ logical_plan
03)----TableScan: agg_expr_parquet projection=[x, v]
physical_plan
01)SortExec: expr=[CAST(x@0 AS Int64) + 1 DESC], preserve_partitioning=[false]
-02)--AggregateExec: mode=Single, gby=[x@0 as x], aggr=[sum(agg_expr_parquet.v)], ordering_mode=Sorted
+02)--AggregateExec: mode=Single, gby=[x@0 as x], aggr=[sum(agg_expr_parquet.v)], group_clustering_mode=Full
03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/agg_expr_sorted.parquet]]}, projection=[x, v], output_ordering=[x@0 ASC NULLS LAST], file_type=parquet
query II
@@ -1060,7 +1060,7 @@ logical_plan
03)----TableScan: agg_expr_parquet projection=[x, v]
physical_plan
01)SortExec: expr=[2 * CAST(x@0 AS Int64) ASC NULLS LAST], preserve_partitioning=[false]
-02)--AggregateExec: mode=Single, gby=[x@0 as x], aggr=[sum(agg_expr_parquet.v)], ordering_mode=Sorted
+02)--AggregateExec: mode=Single, gby=[x@0 as x], aggr=[sum(agg_expr_parquet.v)], group_clustering_mode=Full
03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/sort_pushdown/agg_expr_sorted.parquet]]}, projection=[x, v], output_ordering=[x@0 ASC NULLS LAST], file_type=parquet
query II
diff --git a/datafusion/sqllogictest/test_files/unnest.slt b/datafusion/sqllogictest/test_files/unnest.slt
index 8e6013328f..39ed64a526 100644
--- a/datafusion/sqllogictest/test_files/unnest.slt
+++ b/datafusion/sqllogictest/test_files/unnest.slt
@@ -988,9 +988,9 @@ logical_plan
08)--------------TableScan: range() projection=[value]
physical_plan
01)ProjectionExec: expr=[array_agg(unnested.ar)@1 as array_agg(unnested.ar)]
-02)--AggregateExec: mode=FinalPartitioned, gby=[generated_id@0 as generated_id], aggr=[array_agg(unnested.ar)], ordering_mode=Sorted
+02)--AggregateExec: mode=FinalPartitioned, gby=[generated_id@0 as generated_id], aggr=[array_agg(unnested.ar)], group_clustering_mode=Full
03)----RepartitionExec: partitioning=Hash([generated_id@0], 4), input_partitions=4, preserve_order=true, sort_exprs=generated_id@0 ASC NULLS LAST
-04)------AggregateExec: mode=Partial, gby=[generated_id@0 as generated_id], aggr=[array_agg(unnested.ar)], ordering_mode=Sorted
+04)------AggregateExec: mode=Partial, gby=[generated_id@0 as generated_id], aggr=[array_agg(unnested.ar)], group_clustering_mode=Full
05)--------ProjectionExec: expr=[generated_id@0 as generated_id, __unnest_placeholder(make_array(range().value),depth=1)@1 as ar]
06)----------UnnestExec
07)------------ProjectionExec: expr=[row_number() ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING@1 as generated_id, make_array(value@0) as __unnest_placeholder(make_array(range().value))]
diff --git a/datafusion/sqllogictest/test_files/window.slt b/datafusion/sqllogictest/test_files/window.slt
index a6b64b98de..254fdf20e5 100644
--- a/datafusion/sqllogictest/test_files/window.slt
+++ b/datafusion/sqllogictest/test_files/window.slt
@@ -357,7 +357,7 @@ logical_plan
physical_plan
01)SortPreservingMergeExec: [b@0 ASC NULLS LAST]
02)--ProjectionExec: expr=[b@0 as b, max(d.a)@1 as max_a, max(d.seq)@2 as max(d.seq)]
-03)----AggregateExec: mode=SinglePartitioned, gby=[b@2 as b], aggr=[max(d.a), max(d.seq)], ordering_mode=Sorted
+03)----AggregateExec: mode=SinglePartitioned, gby=[b@2 as b], aggr=[max(d.a), max(d.seq)], group_clustering_mode=Full
04)------ProjectionExec: expr=[row_number() PARTITION BY [s.b] ORDER BY [s.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@2 as seq, a@0 as a, b@1 as b]
05)--------BoundedWindowAggExec: wdw=[row_number() PARTITION BY [s.b] ORDER BY [s.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field { "row_number() PARTITION BY [s.b] ORDER BY [s.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW": UInt64 }, frame: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted]
06)----------SortExec: expr=[b@1 ASC NULLS LAST, a@0 ASC NULLS LAST], preserve_partitioning=[true]
diff --git a/docs/source/library-user-guide/upgrading/56.0.0.md b/docs/source/library-user-guide/upgrading/56.0.0.md
index cb373fdc09..e0f1aff1a7 100644
--- a/docs/source/library-user-guide/upgrading/56.0.0.md
+++ b/docs/source/library-user-guide/upgrading/56.0.0.md
@@ -75,6 +75,27 @@ The Minimum Supported Rust Version (MSRV) has been updated to [`1.95.0`].
[`1.95.0`]: https://releases.rs/docs/1.95.0/
+### Aggregate group-clustering APIs
+
+`AggregateExec::input_order_mode()` has been replaced by
+`AggregateExec::group_clustering_mode()`. It returns a `GroupClusteringMode`:
+`None`, `Partial(indices)`, or `Full`, corresponding to the previous aggregate
+modes `Linear`, `PartiallySorted(indices)`, and `Sorted`.
+`AggregateExec::compute_properties` now accepts `&GroupClusteringMode` in place of
+`&InputOrderMode`.
+
+`GroupClusteringMode` describes the input's group-clustering guarantees, which
+determine when groups can be emitted during aggregation.
+
+In `datafusion_physical_plan::aggregates::order`, `GroupOrdering`,
+`GroupOrderingPartial`, and `GroupOrderingFull` have been renamed to
+`GroupClustering`, `GroupClusteringPartial`, and `GroupClusteringFull`.
+`GroupClustering::try_new` accepts `&GroupClusteringMode`.
+
+Aggregate plans now display `group_clustering_mode=Full` or
+`group_clustering_mode=Partial(indices)` instead of `ordering_mode=Sorted` or
+`ordering_mode=PartiallySorted(indices)`.
+
### Upgrade arrow/parquet to 60.0.0 and object_store to 0.14.2
DataFusion 56.0.0 uses `arrow` and `parquet` 60.0.0, and `object_store` 0.14.2.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org