Skip to content

Commit 88d36d4

Browse files
committed
perf: bucket only aggregations whose group keys and state are fixed-width
1 parent 2868429 commit 88d36d4

6 files changed

Lines changed: 70 additions & 32 deletions

File tree

‎datafusion/common/src/config.rs‎

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1142,11 +1142,13 @@ config_namespace! {
11421142
/// one after another and can be spilled and released independently;
11431143
/// it does so at a quarter of this number when its input holds about
11441144
/// one row per group. Aggregations of millions of groups per partition
1145-
/// run faster and with less memory. Moving rows into buckets has a
1146-
/// cost of its own, so aggregations that end at a few times this
1147-
/// number of groups, string keys in particular, can run a few percent
1148-
/// slower, and input that repeats its groups can use more memory.
1149-
/// Set to 0 to disable.
1145+
/// run faster and with less memory. Only aggregations whose group keys
1146+
/// and aggregate state are all fixed-width are affected: a string or
1147+
/// other variable-length value costs about as much to move into its
1148+
/// bucket as the smaller table saves. Moving rows has a cost of its
1149+
/// own, so aggregations that end at a few times this number of groups
1150+
/// can run a few percent slower, and input that repeats its groups can
1151+
/// use more memory. Set to 0 to disable.
11501152
pub hash_aggregate_bucket_threshold: usize, default = 0
11511153

11521154
/// Should DataFusion use row number estimates at the input to decide

‎datafusion/physical-plan/src/aggregates/bucketed_aggregation.rs‎

Lines changed: 55 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@
2222
use std::sync::Arc;
2323

2424
use arrow::compute::BatchCoalescer;
25-
use arrow::datatypes::SchemaRef;
25+
use arrow::datatypes::{DataType, SchemaRef};
2626
use arrow::record_batch::RecordBatch;
2727
use datafusion_common::{DataFusionError, Result};
2828
use datafusion_execution::memory_pool::MemoryReservation;
@@ -107,20 +107,16 @@ impl BucketedAggregation {
107107

108108
/// True if a schema of partial state rows can be bucketed.
109109
///
110-
/// Bucketing turns the state of a table into rows and merges those rows
111-
/// again. That is cheap for fixed-width and string state, but nested
112-
/// state (the lists kept by `count(distinct)`, `array_agg` or `median`)
113-
/// is costly to rebuild and takes more memory as rows than inside the
114-
/// accumulator, so such aggregations keep their single table.
115-
pub(super) fn supports_state(
116-
state_schema: &SchemaRef,
117-
num_group_columns: usize,
118-
) -> bool {
119-
!state_schema
120-
.fields()
121-
.iter()
122-
.skip(num_group_columns)
123-
.any(|field| field.data_type().is_nested())
110+
/// Bucketing moves every row once more before it is aggregated, which only
111+
/// pays when moving a row is cheap compared to the cache miss it saves.
112+
/// That holds for fixed-width columns. Variable-length keys or state
113+
/// (strings, lists, ...) cost as much to move as the smaller tables save,
114+
/// so such aggregations keep their single table.
115+
pub(super) fn supports_state(state_schema: &SchemaRef) -> bool {
116+
state_schema.fields().iter().all(|field| {
117+
let data_type = field.data_type();
118+
data_type.is_primitive() || *data_type == DataType::Boolean
119+
})
124120
}
125121

126122
pub(super) fn threshold(&self) -> usize {
@@ -365,3 +361,47 @@ impl BucketedAggregation {
365361
})
366362
}
367363
}
364+
365+
#[cfg(test)]
366+
mod tests {
367+
use super::*;
368+
369+
use arrow::datatypes::{Field, Fields, Schema};
370+
371+
fn schema(types: Vec<DataType>) -> SchemaRef {
372+
let fields: Fields = types
373+
.into_iter()
374+
.enumerate()
375+
.map(|(i, data_type)| Field::new(format!("c{i}"), data_type, true))
376+
.collect();
377+
Arc::new(Schema::new(fields))
378+
}
379+
380+
#[test]
381+
fn only_fixed_width_state_is_bucketed() {
382+
assert!(BucketedAggregation::supports_state(&schema(vec![
383+
DataType::Int64,
384+
DataType::Boolean,
385+
DataType::Decimal128(38, 10),
386+
DataType::Float64,
387+
DataType::Date32,
388+
])));
389+
390+
// A variable-length group key or state column turns bucketing off
391+
for variable_length in [
392+
DataType::Utf8,
393+
DataType::Utf8View,
394+
DataType::LargeBinary,
395+
DataType::List(Arc::new(Field::new_list_field(DataType::Int64, true))),
396+
] {
397+
assert!(!BucketedAggregation::supports_state(&schema(vec![
398+
DataType::Int64,
399+
variable_length.clone(),
400+
])));
401+
assert!(!BucketedAggregation::supports_state(&schema(vec![
402+
variable_length,
403+
DataType::Int64,
404+
])));
405+
}
406+
}
407+
}

‎datafusion/physical-plan/src/aggregates/hash_stream.rs‎

Lines changed: 3 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -384,14 +384,10 @@ impl PartialHashAggregateStream {
384384
let num_group_columns = agg.group_by().num_group_exprs();
385385
// Same conditions as for bucketing in the final aggregation, which
386386
// receives what is flushed here: see `FinalHashAggregateStream::new`.
387-
let has_nested_state = schema
388-
.fields()
389-
.iter()
390-
.skip(num_group_columns)
391-
.any(|field| field.data_type().is_nested());
387+
let fixed_width_state = BucketedAggregation::supports_state(&schema);
392388
let table_flush = (bucket_threshold > 0
393389
&& group_values_soft_limit.is_none()
394-
&& !has_nested_state)
390+
&& fixed_width_state)
395391
.then(|| PartialTableFlush {
396392
threshold: bucket_threshold,
397393
num_group_columns,
@@ -833,10 +829,7 @@ impl FinalHashAggregateStream {
833829
// A soft limit stops reading input early, which bucketing cannot do.
834830
let bucketing = (bucket_threshold > 0
835831
&& group_values_soft_limit.is_none()
836-
&& BucketedAggregation::supports_state(
837-
&input_schema,
838-
agg.group_by().num_group_exprs(),
839-
))
832+
&& BucketedAggregation::supports_state(&input_schema))
840833
.then(|| {
841834
Arc::new(BucketedAggregation::new(
842835
bucket_threshold,

‎datafusion/sqllogictest/test_files/aggregate_bucketed.slt‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -128,6 +128,9 @@ select count(*), round(sum(a), 3) from (select v % 10007 as k, avg(v) a from t g
128128
----
129129
10007 1000705003.5
130130

131+
# The next three queries have a string key, so they keep a single table: only
132+
# fixed-width state is bucketed. Their results must not change.
133+
131134
# string key
132135
query II
133136
select count(*), sum(c) from (select concat('key-', cast(v % 9973 as varchar)) as k, count(*) c from t group by k);

‎datafusion/sqllogictest/test_files/information_schema.slt‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -385,7 +385,7 @@ datafusion.execution.enable_migration_aggregate true Whether aggregation uses th
385385
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 shares left-side state (the current left chunk, the visited bitmap and the probe-thread counter) across all right-side partitions, which assumes every partition runs in the same process. Distributed engines that execute each output partition as an independent task (e.g. Ballista, datafusion-distributed) give each task its own copy of this state and poll only one partition, so the cross-partition counter never reaches zero and the fallback would stall. Such engines should set this to `false`: the coordinated fallback is then disabled for left-emitting multi-partition joins, which instead fail with a resource-exhaustion error under memory pressure rather than deadlocking. Single-partition and non-left-emitting joins are unaffected and always keep the fallback.
386386
datafusion.execution.enable_recursive_ctes true Should DataFusion support recursive CTEs
387387
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. Note: this option currently only applies to the symmetric hash join.
388-
datafusion.execution.hash_aggregate_bucket_threshold 0 (experimental) Number of groups above which a hash aggregation stops growing a single hash table, so that its tables stay small enough to be cache friendly. A partial aggregation then emits the state of its table and starts over, as long as the emitted groups do not come back. A final aggregation splits the groups seen so far and all further input into hash buckets, which are aggregated one after another and can be spilled and released independently; it does so at a quarter of this number when its input holds about one row per group. Aggregations of millions of groups per partition run faster and with less memory. Moving rows into buckets has a cost of its own, so aggregations that end at a few times this number of groups, string keys in particular, can run a few percent slower, and input that repeats its groups can use more memory. Set to 0 to disable.
388+
datafusion.execution.hash_aggregate_bucket_threshold 0 (experimental) Number of groups above which a hash aggregation stops growing a single hash table, so that its tables stay small enough to be cache friendly. A partial aggregation then emits the state of its table and starts over, as long as the emitted groups do not come back. A final aggregation splits the groups seen so far and all further input into hash buckets, which are aggregated one after another and can be spilled and released independently; it does so at a quarter of this number when its input holds about one row per group. Aggregations of millions of groups per partition run faster and with less memory. Only aggregations whose group keys and aggregate state are all fixed-width are affected: a string or other variable-length value costs about as much to move into its bucket as the smaller table saves. Moving rows has a cost of its own, so aggregations that end at a few times this number of groups can run a few percent slower, and input that repeats its groups can use more memory. Set to 0 to disable.
389389
datafusion.execution.hash_join_buffering_capacity 0 How many bytes to buffer in the probe side of hash joins while the build side is concurrently being built. Without this, hash joins will wait until the full materialization of the build side before polling the probe side. This is useful in scenarios where the query is not completely CPU bounded, allowing to do some early work concurrently and reducing the latency of the query. Note that when hash join buffering is enabled, the probe side will start eagerly polling data, not giving time for the producer side of dynamic filters to produce any meaningful predicate. Queries with dynamic filters might see performance degradation. Disabled by default, set to a number greater than 0 for enabling it.
390390
datafusion.execution.keep_partition_by_columns false Should DataFusion keep the columns used for partition_by in the output RecordBatches
391391
datafusion.execution.listing_table_factory_infer_partitions true Should a `ListingTable` created through the `ListingTableFactory` infer table partitions from Hive compliant directories. Defaults to true (partition columns are inferred and will be represented in the table schema).

0 commit comments

Comments
 (0)