You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/09/09 03:48:40 UTC
(datafusion) branch main updated: chore: Clean up out-of-date comments for aggregate refactor (#25007)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new bf01e288ee chore: Clean up out-of-date comments for aggregate refactor (#25007)
bf01e288ee is described below
commit bf01e288eeb5cbfb2c8a4cf77f82f97f81c109ba
Author: Yongting You <20...@gmail.com>
AuthorDate: Wed Sep 9 03:38:10 2026 +0000
chore: Clean up out-of-date comments for aggregate refactor (#25007)
## Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax. For example
`Closes #123` indicates that this PR will close issue #123.
-->
Part of https://github.com/apache/datafusion/issues/25006
## Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly
in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand
your changes and offer better suggestions for fixes.
Please explain the problem you are trying to solve in terms of the
user-visible
behavior, rather than the implementation.
For example, "The code in `foo.rs` doesn't handle nulls" is a symptom of
the
implementation. "COUNT(DISTINCT) returns wrong results when the column
contains
nulls" is the user-visible problem.
-->
The refactor has been completed functionally, this PR updates the
comments to keep things in sync.
## What changes are included in this PR?
<!--
There is no need to duplicate the description in the issue here, but it
is sometimes worth providing a summary of the individual changes in this
PR.
-->
## What is the testing strategy for this PR?
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code
Briefly describe how this PR is tested, and point to the specific tests
you added. For example: 'This new feature is covered by the
`sqllogictest` cases added in `foo.slt`'.
If this PR does not add tests, explain why. For example, if the change
is already covered by existing tests, please mention it.
You should also check the `codecov` bot reply on this PR to confirm the
changed code is exercised.
-->
## Are there any user-facing changes?
<!--
If there are user-facing changes then we may require documentation to be
updated before approving the PR.
If there are any breaking changes to public APIs, please add the `api
change` label.
-->
---
datafusion/common/src/config.rs | 12 ++++----
datafusion/core/tests/fuzz_cases/aggregate_fuzz.rs | 8 +++--
.../src/aggregates/group_values/metrics.rs | 6 ++--
.../src/aggregates/grouped_hash_stream.rs | 26 ++++++++++-------
.../physical-plan/src/aggregates/hash_stream.rs | 5 ----
datafusion/physical-plan/src/aggregates/mod.rs | 34 ++++++++++------------
.../src/aggregates/ordered_partial_stream.rs | 8 -----
.../src/aggregates/partial_reduce_stream.rs | 5 ----
.../physical-plan/src/aggregates/single_stream.rs | 5 ----
.../physical-plan/src/aggregates/skip_partial.rs | 5 ++--
.../sqllogictest/test_files/information_schema.slt | 2 +-
docs/source/user-guide/configs.md | 2 +-
12 files changed, 50 insertions(+), 68 deletions(-)
diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs
index 042d3c330c..f770cebf88 100644
--- a/datafusion/common/src/config.rs
+++ b/datafusion/common/src/config.rs
@@ -964,13 +964,13 @@ config_namespace! {
/// the new schema verification step.
pub skip_physical_aggregate_schema_check: bool, default = false
- /// Temporary switch for aggregate stream implementations that are being
- /// migrated from `GroupedHashAggregateStream`.
+ /// Whether aggregation uses the implementation from the major refactor
+ /// completed in the 56.0.0 release. When set to `false`, aggregation
+ /// falls back to the implementation used before 55.0.0.
///
- /// When set to true, DataFusion tries the migrated implementations when
- /// their preconditions are satisfied. When set to false, grouped
- /// aggregation falls back to `GroupedHashAggregateStream`. This option
- /// will be removed after the migration is finished.
+ /// The fallback exists only as a workaround for bugs in the new
+ /// implementation and will be removed, together with this option, after
+ /// the 56.0.0 release.
///
/// See <https://github.com/apache/datafusion/issues/22710> for details.
pub enable_migration_aggregate: bool, default = true
diff --git a/datafusion/core/tests/fuzz_cases/aggregate_fuzz.rs b/datafusion/core/tests/fuzz_cases/aggregate_fuzz.rs
index 5efdfd0d45..afbb1b21d0 100644
--- a/datafusion/core/tests/fuzz_cases/aggregate_fuzz.rs
+++ b/datafusion/core/tests/fuzz_cases/aggregate_fuzz.rs
@@ -299,9 +299,11 @@ async fn streaming_aggregate_test() {
}
}
-/// Perform batch and streaming aggregation with same input
-/// and verify outputs of `AggregateExec` with pipeline breaking stream `GroupedHashAggregateStream`
-/// and non-pipeline breaking stream `BoundedAggregateStream` produces same result.
+/// Perform batch and streaming aggregation with same input and verify that the
+/// 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`).
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);
diff --git a/datafusion/physical-plan/src/aggregates/group_values/metrics.rs b/datafusion/physical-plan/src/aggregates/group_values/metrics.rs
index d6405bb403..5e66fe496e 100644
--- a/datafusion/physical-plan/src/aggregates/group_values/metrics.rs
+++ b/datafusion/physical-plan/src/aggregates/group_values/metrics.rs
@@ -401,9 +401,9 @@ mod tests {
schema,
)?);
- // This test is for `GroupByMetrics`, which are maintained by
- // `GroupedHashAggregateStream`. Use a finite memory pool so the partial
- // aggregate does not take the initial-partial stream path.
+ // This test is for `GroupByMetrics`, which every grouped aggregation
+ // stream records. The memory limit is large enough that the partial
+ // aggregate stays on the in-memory path.
let runtime = RuntimeEnvBuilder::new()
.with_memory_limit(10 * 1024 * 1024, 1.0)
.build_arc()?;
diff --git a/datafusion/physical-plan/src/aggregates/grouped_hash_stream.rs b/datafusion/physical-plan/src/aggregates/grouped_hash_stream.rs
index 24bb4c16d8..dd77abfeb6 100644
--- a/datafusion/physical-plan/src/aggregates/grouped_hash_stream.rs
+++ b/datafusion/physical-plan/src/aggregates/grouped_hash_stream.rs
@@ -15,7 +15,20 @@
// specific language governing permissions and limitations
// under the License.
-//! Hash aggregation
+//! Legacy hash aggregation.
+//!
+//! # Deprecation
+//!
+//! [`GroupedHashAggregateStream`] handled every grouped execution path before
+//! they were split into dedicated streams. It is no longer planned by default:
+//! it is only reachable by setting
+//! `datafusion.execution.enable_migration_aggregate` to `false`. It is kept as
+//! a fallback in case of major bugs in the new streams, and will be deleted
+//! after the 56.0.0 release together with that option.
+//!
+//! New features and improvements should go into the dedicated streams instead.
+//!
+//! See issue for details: <https://github.com/apache/datafusion/issues/22710>
use std::sync::Arc;
use std::task::{Context, Poll};
@@ -136,15 +149,8 @@ enum OutOfMemoryMode {
/// HashTable based Grouping Aggregator
///
-/// # Development Note
-///
-/// This implementation is being incrementally refactored. See the tracking issue
-/// for details.
-///
-/// New features and improvements should go directly into the new implementation.
-/// Please coordinate through the tracking issue.
-///
-/// Issue: <https://github.com/apache/datafusion/issues/22710>
+/// This is the legacy implementation. See the [module documentation](self) for
+/// the deprecation schedule.
///
/// # Design Goals
///
diff --git a/datafusion/physical-plan/src/aggregates/hash_stream.rs b/datafusion/physical-plan/src/aggregates/hash_stream.rs
index 7ebf32c3ed..2195b5eb60 100644
--- a/datafusion/physical-plan/src/aggregates/hash_stream.rs
+++ b/datafusion/physical-plan/src/aggregates/hash_stream.rs
@@ -19,11 +19,6 @@
//!
//! See comments in [`PartialHashAggregateStream`] and [`FinalHashAggregateStream`]
//! for details.
-//!
-//! Note these streams are an incremental migration of the existing
-//! [`crate::aggregates::grouped_hash_stream::GroupedHashAggregateStream`].
-//!
-//! See issue for details: <https://github.com/apache/datafusion/issues/22710>
use std::mem::size_of;
use std::sync::Arc;
diff --git a/datafusion/physical-plan/src/aggregates/mod.rs b/datafusion/physical-plan/src/aggregates/mod.rs
index 774c08535c..9cab9fa276 100644
--- a/datafusion/physical-plan/src/aggregates/mod.rs
+++ b/datafusion/physical-plan/src/aggregates/mod.rs
@@ -102,7 +102,7 @@
//! ```
//!
//! Every stage without grouping expressions uses [`AggregateStream`]. This path
-//! is selected before the grouped-stream migration setting is considered.
+//! is selected before any of the grouped streams are considered.
//!
//! ## 4. Grouped TopK aggregation
//!
@@ -135,10 +135,14 @@
//!
//! See [`PartialReduceHashAggregateStream`] for details.
//!
-//! ## 6. Fallback grouped hash aggregation
+//! ## 6. Legacy grouped hash aggregation
//!
-//! [`GroupedHashAggregateStream`] is the legacy implementation for several of the
-//! stream types above. It is being incrementally migrated to separate streams.
+//! [`GroupedHashAggregateStream`] is the legacy implementation that all of the
+//! grouped streams above were split out of. The split is complete, so it is no
+//! longer planned: it is only reachable by setting
+//! [`datafusion.execution.enable_migration_aggregate`](datafusion_common::config::ExecutionOptions::enable_migration_aggregate)
+//! to `false`. That fallback is kept for one release and will then be removed
+//! together with this stream.
//!
//! See the issue for details: <https://github.com/apache/datafusion/issues/22710>
#![expect(rustdoc::private_intra_doc_links)]
@@ -691,15 +695,11 @@ enum StreamType {
OrderedFinalAggregate(OrderedFinalAggregateStream),
/// Single stage of aggregation for ordered input.
OrderedSingleAggregate(OrderedSingleAggregateStream),
- /// Hash aggregation reused for multiple stages
+ /// Legacy hash aggregation reused for multiple stages
///
- /// Note this is being incrementally migrated to dedicated streams like
- /// [`StreamType::PartialHash`], [`StreamType::FinalHash`],
- /// [`StreamType::OrderedPartialAggregate`],
- /// [`StreamType::OrderedFinalAggregate`], and
- /// [`StreamType::OrderedSingleAggregate`]
- ///
- /// See issue for details: <https://github.com/apache/datafusion/issues/22710>
+ /// Every path it handles now has a dedicated stream, so this variant is only
+ /// produced when `datafusion.execution.enable_migration_aggregate` is set to
+ /// `false`. See [`grouped_hash_stream`] for the deprecation schedule.
GroupedHash(GroupedHashAggregateStream),
/// Grouped TopK aggregate stream.
/// Input output scheme: initial input -> final result
@@ -1234,11 +1234,8 @@ impl AggregateExec {
// `enable_migration_aggregate` config option selects between the new
// streams and the legacy implementation.
//
- // The legacy implementation is deprecated. It is kept for one more
- // release as a fallback for potential bugs in the new streams, and
- // will be removed after that.
- //
- // Issue: <https://github.com/apache/datafusion/issues/22710>
+ // See the `grouped_hash_stream` module documentation for the
+ // deprecation schedule.
if context
.session_config()
.options()
@@ -1292,7 +1289,8 @@ impl AggregateExec {
);
}
- // Execution paths that have not been migrated use the fallback implementation
+ // `enable_migration_aggregate` is disabled: fall back to the legacy
+ // implementation, which handles every grouped execution path itself.
Ok(StreamType::GroupedHash(GroupedHashAggregateStream::new(
self, context, partition,
)?))
diff --git a/datafusion/physical-plan/src/aggregates/ordered_partial_stream.rs b/datafusion/physical-plan/src/aggregates/ordered_partial_stream.rs
index 9e93a111a6..8c2315588a 100644
--- a/datafusion/physical-plan/src/aggregates/ordered_partial_stream.rs
+++ b/datafusion/physical-plan/src/aggregates/ordered_partial_stream.rs
@@ -102,14 +102,6 @@ use crate::{InputOrderMode, SendableRecordBatchStream, metrics};
/// 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.
-///
-/// ## Implementation Note
-///
-/// This is intentionally kept simple and closely maps to
-/// `GroupedHashAggregateStream` to finish the refactor sooner.
-///
-/// See issue for details: <https://github.com/apache/datafusion/issues/22710>
-///
pub(crate) struct OrderedPartialAggregateStream {
schema: SchemaRef,
input: SendableRecordBatchStream,
diff --git a/datafusion/physical-plan/src/aggregates/partial_reduce_stream.rs b/datafusion/physical-plan/src/aggregates/partial_reduce_stream.rs
index d8f1447bc9..c9a6cb675d 100644
--- a/datafusion/physical-plan/src/aggregates/partial_reduce_stream.rs
+++ b/datafusion/physical-plan/src/aggregates/partial_reduce_stream.rs
@@ -16,11 +16,6 @@
// under the License.
//! Partial-reduce hash aggregation stream implementation.
-//!
-//! This stream is part of the incremental migration from
-//! [`crate::aggregates::grouped_hash_stream::GroupedHashAggregateStream`].
-//!
-//! See issue for details: <https://github.com/apache/datafusion/issues/22710>
use std::ops::ControlFlow;
use std::sync::Arc;
diff --git a/datafusion/physical-plan/src/aggregates/single_stream.rs b/datafusion/physical-plan/src/aggregates/single_stream.rs
index 40412385ef..6d46b90c26 100644
--- a/datafusion/physical-plan/src/aggregates/single_stream.rs
+++ b/datafusion/physical-plan/src/aggregates/single_stream.rs
@@ -16,11 +16,6 @@
// under the License.
//! Single-stage hash aggregation stream implementation.
-//!
-//! This stream is part of the incremental migration from
-//! [`crate::aggregates::grouped_hash_stream::GroupedHashAggregateStream`].
-//!
-//! See issue for details: <https://github.com/apache/datafusion/issues/22710>
use std::ops::ControlFlow;
use std::sync::Arc;
diff --git a/datafusion/physical-plan/src/aggregates/skip_partial.rs b/datafusion/physical-plan/src/aggregates/skip_partial.rs
index 20e17d2b27..40e35dd7d8 100644
--- a/datafusion/physical-plan/src/aggregates/skip_partial.rs
+++ b/datafusion/physical-plan/src/aggregates/skip_partial.rs
@@ -21,8 +21,7 @@ use crate::metrics;
/// Tracks if the aggregate should skip partial aggregations
///
-/// See "partial aggregation" discussion on
-/// [`crate::aggregates::grouped_hash_stream::GroupedHashAggregateStream`].
+/// Used by [`crate::aggregates::hash_stream::PartialHashAggregateStream`].
pub(super) struct SkipAggregationProbe {
// ========================================================================
// PROPERTIES:
@@ -40,7 +39,7 @@ pub(super) struct SkipAggregationProbe {
// ========================================================================
// STATES:
// Fields changes during execution. Can be buffer, or state flags that
- // influence the execution in parent `GroupedHashAggregateStream`
+ // influence the execution in the parent aggregate stream
// ========================================================================
/// Number of processed input rows (updated during probing)
input_rows: usize,
diff --git a/datafusion/sqllogictest/test_files/information_schema.slt b/datafusion/sqllogictest/test_files/information_schema.slt
index 4ef57e4ae2..4551f21f21 100644
--- a/datafusion/sqllogictest/test_files/information_schema.slt
+++ b/datafusion/sqllogictest/test_files/information_schema.slt
@@ -380,7 +380,7 @@ datafusion.execution.coalesce_batches true When set to true, record batches will
datafusion.execution.collect_statistics true Should DataFusion collect statistics when first creating a table. Has no effect after the table is created. Defaults to true.
datafusion.execution.enable_ansi_mode false Whether to enable ANSI SQL mode. The flag is experimental and relevant only for DataFusion Spark built-in functions When `enable_ansi_mode` is set to `true`, the query engine follows ANSI SQL semantics for expressions, casting, and error handling. This means: - **Strict type coercion rules:** implicit casts between incompatible types are disallowed. - **Standard SQL arithmetic behavior:** operations such as division by zero, numeric overflow, [...]
datafusion.execution.enable_file_stream_work_stealing true When `true` (the default), DataFusion's built-in file scans dynamically rebalance files across partitions at query execution time: a partition that goes idle reads files (or byte-range morsels) originally assigned to a sibling partition, which keeps all partitions busy in a single process. Executors that depend on the plan-time partition assignment — such as Ballista and datafusion-distributed, which run each partition as an isol [...]
-datafusion.execution.enable_migration_aggregate true Temporary switch for aggregate stream implementations that are being migrated from `GroupedHashAggregateStream`. When set to true, DataFusion tries the migrated implementations when their preconditions are satisfied. When set to false, grouped aggregation falls back to `GroupedHashAggregateStream`. This option will be removed after the migration is finished. See <https://github.com/apache/datafusion/issues/22710> for details.
+datafusion.execution.enable_migration_aggregate true Whether aggregation uses the implementation from the major refactor completed in the 56.0.0 release. When set to `false`, aggregation falls back to the implementation used before 55.0.0. The fallback exists only as a workaround for bugs in the new implementation and will be removed, together with this option, after the 56.0.0 release. See <https://github.com/apache/datafusion/issues/22710> for details.
datafusion.execution.enable_nlj_coordinated_fallback true Enables the memory-limited fallback for `NestedLoopJoinExec` join types that emit unmatched left rows in the final output (LEFT, LEFT SEMI, LEFT ANTI, LEFT MARK, FULL) when the right side has multiple partitions. This fallback coordinates per-chunk left state (visited bitmap and probe-thread counter) across all right-side partitions, which assumes every partition runs in the same process. Distributed engines that execute each outp [...]
datafusion.execution.enable_recursive_ctes true Should DataFusion support recursive CTEs
datafusion.execution.enforce_batch_size_in_joins false Should DataFusion enforce batch size in joins or not. By default, DataFusion will not enforce batch size in joins. Enforcing batch size in joins can reduce memory usage when joining large tables with a highly-selective join filter, but is also slightly slower.
diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md
index f7e47ade4f..fbeb38f1ce 100644
--- a/docs/source/user-guide/configs.md
+++ b/docs/source/user-guide/configs.md
@@ -121,7 +121,7 @@ The following configuration settings are available:
| datafusion.execution.parquet.content_defined_chunking.norm_level | 0 | Normalization level. Increasing this improves deduplication ratio but increases fragmentation. Recommended range is [-3, 3], default is 0. [...]
| datafusion.execution.planning_concurrency | 0 | Fan-out during initial physical planning. This is mostly use to plan `UNION` children in parallel. Defaults to the number of CPU cores on the system [...]
| datafusion.execution.skip_physical_aggregate_schema_check | false | When set to true, skips verifying that the schema produced by planning the input of `LogicalPlan::Aggregate` exactly matches the schema of the input plan. When set to false, if the schema does not match exactly (including nullability and metadata), a planning error will be raised. This is used to workaround bugs in the planner that are now caught by the new schema verification step. [...]
-| datafusion.execution.enable_migration_aggregate | true | Temporary switch for aggregate stream implementations that are being migrated from `GroupedHashAggregateStream`. When set to true, DataFusion tries the migrated implementations when their preconditions are satisfied. When set to false, grouped aggregation falls back to `GroupedHashAggregateStream`. This option will be removed after the migration is finished. See <https://github.com/apa [...]
+| datafusion.execution.enable_migration_aggregate | true | Whether aggregation uses the implementation from the major refactor completed in the 56.0.0 release. When set to `false`, aggregation falls back to the implementation used before 55.0.0. The fallback exists only as a workaround for bugs in the new implementation and will be removed, together with this option, after the 56.0.0 release. See <https://github.com/apache/datafusion/issues/22 [...]
| datafusion.execution.spill_compression | uncompressed | Sets the compression codec used when spilling data to disk. Since datafusion writes spill files using the Arrow IPC Stream format, only codecs supported by the Arrow IPC Stream Writer are allowed. Valid values are: uncompressed, lz4_frame, zstd. Note: lz4_frame offers faster (de)compression, but typically results in larger spill files. In contrast, zstd achieves higher compression rati [...]
| datafusion.execution.sort_spill_reservation_bytes | 10485760 | Specifies the reserved memory for each spillable sort operation to facilitate an in-memory merge. When a sort operation spills to disk, the in-memory data must be sorted and merged before being written to a file. This setting reserves a specific amount of memory for that in-memory sort/merge process. Note: This setting is irrelevant if the sort operation cannot spill (i.e., if there's [...]
| datafusion.execution.sort_in_place_threshold_bytes | 1048576 | When sorting, below what size should data be concatenated and sorted in a single RecordBatch rather than sorted in batches and merged. [...]
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org