feat: make threshold for using scalar update in aggregate configurable

datafusion/common/src/config.rs

@@ -223,6 +223,9 @@ config_namespace! {
 
         /// Parquet options
         pub parquet: ParquetOptions, default = Default::default()
+
+        /// Aggregate options
+        pub aggregate: AggregateOptions, default = Default::default()
     }
 }
 
@@ -260,6 +263,23 @@ config_namespace! {
     }
 }
 
+config_namespace! {
+    /// Options related to aggregate execution
+    pub struct AggregateOptions {
+        /// Specifies the threshold for using `ScalarValue`s to update
+        /// accumulators during high-cardinality aggregations for each input batch.
+        ///
+        /// The aggregation is considered high-cardinality if the number of affected groups
+        /// is greater than or equal to `batch_size / scalar_update_factor`. In such cases,
+        /// `ScalarValue`s are utilized for updating accumulators, rather than the default
+        /// batch-slice approach. This can lead to performance improvements.
+        ///
+        /// By adjusting the `scalar_update_factor`, you can balance the trade-off between
+        /// more efficient accumulator updates and the number of groups affected.
+        pub scalar_update_factor: usize, default = 10
+    }
+}
+
 config_namespace! {
     /// Options related to query optimization
     pub struct OptimizerOptions {

datafusion/core/src/physical_plan/aggregates/bounded_aggregate_stream.rs

@@ -103,6 +103,9 @@ pub(crate) struct BoundedAggregateStream {
     random_state: RandomState,
     /// size to be used for resulting RecordBatches
     batch_size: usize,
+    /// threshold for using `ScalarValue`s to update
+    /// accumulators during high-cardinality aggregations for each input batch.
+    scalar_update_factor: usize,
     /// if the result is chunked into batches,
     /// last offset is preserved for continuation.
     row_group_skip_position: usize,
@@ -126,6 +129,7 @@ impl BoundedAggregateStream {
         input: SendableRecordBatchStream,
         baseline_metrics: BaselineMetrics,
         batch_size: usize,
+        scalar_update_factor: usize,
         context: Arc<TaskContext>,
         partition: usize,
         // Stores algorithm mode and output ordering
@@ -228,6 +232,7 @@ impl BoundedAggregateStream {
             baseline_metrics,
             random_state: Default::default(),
             batch_size,
+            scalar_update_factor,
             row_group_skip_position: 0,
             indices: [normal_agg_indices, row_agg_indices],
             is_end: false,
@@ -747,7 +752,7 @@ impl BoundedAggregateStream {
             if matches!(self.mode, AggregateMode::Partial | AggregateMode::Single)
                 && normal_aggr_input_values.is_empty()
                 && normal_filter_values.is_empty()
-                && groups_with_rows.len() >= batch.num_rows() / 10
+                && groups_with_rows.len() >= batch.num_rows() / self.scalar_update_factor
             {
                 self.update_accumulators_using_scalar(
                     &groups_with_rows,

datafusion/core/src/physical_plan/aggregates/mod.rs

@@ -424,6 +424,7 @@ impl AggregateExec {
         context: Arc<TaskContext>,
     ) -> Result<StreamType> {
         let batch_size = context.session_config().batch_size();
+        let scalar_update_factor = context.session_config().agg_scalar_update_factor();
         let input = self.input.execute(partition, Arc::clone(&context))?;
         let baseline_metrics = BaselineMetrics::new(&self.metrics, partition);
 
@@ -448,6 +449,7 @@ impl AggregateExec {
                 input,
                 baseline_metrics,
                 batch_size,
+                scalar_update_factor,
                 context,
                 partition,
                 aggregation_ordering.clone(),
@@ -463,6 +465,7 @@ impl AggregateExec {
                     input,
                     baseline_metrics,
                     batch_size,
+                    scalar_update_factor,
                     context,
                     partition,
                 )?,

datafusion/core/src/physical_plan/aggregates/row_hash.rs

@@ -98,6 +98,9 @@ pub(crate) struct GroupedHashAggregateStream {
     random_state: RandomState,
     /// size to be used for resulting RecordBatches
     batch_size: usize,
+    /// threshold for using `ScalarValue`s to update
+    /// accumulators during high-cardinality aggregations for each input batch.
+    scalar_update_factor: usize,
     /// if the result is chunked into batches,
     /// last offset is preserved for continuation.
     row_group_skip_position: usize,
@@ -119,6 +122,7 @@ impl GroupedHashAggregateStream {
         input: SendableRecordBatchStream,
         baseline_metrics: BaselineMetrics,
         batch_size: usize,
+        scalar_update_factor: usize,
         context: Arc<TaskContext>,
         partition: usize,
     ) -> Result<Self> {
@@ -219,6 +223,7 @@ impl GroupedHashAggregateStream {
             baseline_metrics,
             random_state: Default::default(),
             batch_size,
+            scalar_update_factor,
             row_group_skip_position: 0,
             indices: [normal_agg_indices, row_agg_indices],
         })
@@ -555,7 +560,7 @@ impl GroupedHashAggregateStream {
             if matches!(self.mode, AggregateMode::Partial | AggregateMode::Single)
                 && normal_aggr_input_values.is_empty()
                 && normal_filter_values.is_empty()
-                && groups_with_rows.len() >= batch.num_rows() / 10
+                && groups_with_rows.len() >= batch.num_rows() / self.scalar_update_factor
             {
                 self.update_accumulators_using_scalar(
                     &groups_with_rows,

datafusion/core/tests/sqllogictests/test_files/information_schema.slt

@@ -136,6 +136,7 @@ datafusion.catalog.format NULL
 datafusion.catalog.has_header false
 datafusion.catalog.information_schema true
 datafusion.catalog.location NULL
+datafusion.execution.aggregate.scalar_update_factor 10
 datafusion.execution.batch_size 8192
 datafusion.execution.coalesce_batches true
 datafusion.execution.collect_statistics false

datafusion/execution/src/config.rs

@@ -237,6 +237,19 @@ impl SessionConfig {
         self.options.execution.batch_size
     }
 
+    /// Get the currently configured scalar_update_factor for aggregate
+    pub fn agg_scalar_update_factor(&self) -> usize {
+        self.options.execution.aggregate.scalar_update_factor
+    }
+
+    /// Customize scalar_update_factor for aggregate
+    pub fn with_agg_scalar_update_factor(mut self, n: usize) -> Self {
+        // scalar update factor must be greater than zero
+        assert!(n > 0);
+        self.options.execution.aggregate.scalar_update_factor = n;
+        self
+    }
+
     /// Convert configuration options to name-value pairs with values
     /// converted to strings.
     ///