add config to control if we enable blocked groups optimization.

Author: Rachelint

datafusion/common/src/config.rs

@@ -405,6 +405,18 @@ config_namespace! {
         /// in joins can reduce memory usage when joining large
         /// tables with a highly-selective join filter, but is also slightly slower.
         pub enforce_batch_size_in_joins: bool, default = false
+
+        /// Should DataFusion use the the blocked approach to manage the groups
+        /// values and their related states in accumulators.
+        /// By default, the blocked approach will be used. And the blocked approach
+        /// allocates capacity for the block based on a predefined block size firstly.
+        /// When the block reaches its limit, we allocate a new block (also with
+        /// the same predefined block size based capacity) instead of expanding
+        /// the current one and copying the data.
+        /// If setting this flag to `false`, will fall-back to use the single approach,
+        /// values are managed within a single large block(can think of it as a Vec).
+        /// As this block grows, it often triggers numerous copies, resulting in poor performance.
+        pub enable_aggregation_blocked_groups: bool, default = true
     }
 }
 

datafusion/physical-plan/src/aggregates/row_hash.rs

@@ -659,7 +659,7 @@ pub(crate) fn create_group_accumulator(
 
 /// Check if we can enable the blocked optimization for `GroupValues` and `GroupsAccumulator`s.
 /// The blocked optimization will be enabled when:
-///   - When `enable_aggregation_intermediate_states_blocked_approach` is true
+///   - When `enable_aggregation_blocked_groups` is true(default to true)
 ///   - It is not streaming aggregation(because blocked mode can't support Emit::first(exact n))
 ///   - The spilling is disabled(still need to consider more to support it efficiently)
 ///   - The accumulator is not empty(I am still not sure about logic in this case)
@@ -676,8 +676,12 @@ fn maybe_enable_blocked_groups(
     block_size: usize,
     group_ordering: &GroupOrdering,
 ) -> Result<bool> {
-    // if !context.session_config().options().execution
-    if !matches!(group_ordering, GroupOrdering::None)
+    if !context
+        .session_config()
+        .options()
+        .execution
+        .enable_aggregation_blocked_groups
+        || !matches!(group_ordering, GroupOrdering::None)
         || accumulators.is_empty()
         || context.runtime_env().disk_manager.tmp_files_enabled()
     {