Skip to content

Commit ef78b05

Browse files
committed
perf: use foldhash for BoundedWindowAggExec partition state maps
PartitionWindowAggStates and PartitionBatches were using IndexMap's default SipHash hasher. Switching to foldhash yields a small but measurable performance improvement. This improves performance in linear mode for 10k partitions by ~6%.
1 parent 5935aa7 commit ef78b05

2 files changed

Lines changed: 5 additions & 4 deletions

File tree

datafusion/physical-expr/src/window/window_expr.rs

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@ use arrow::compute::kernels::sort::SortColumn;
3030
use arrow::datatypes::FieldRef;
3131
use arrow::record_batch::RecordBatch;
3232
use datafusion_common::cast::as_boolean_array;
33+
use datafusion_common::hash_utils::RandomState;
3334
use datafusion_common::utils::compare_rows;
3435
use datafusion_common::{
3536
Result, ScalarValue, arrow_datafusion_err, exec_datafusion_err, internal_err,
@@ -609,10 +610,10 @@ pub struct WindowState {
609610
pub state: WindowAggState,
610611
pub window_fn: WindowFn,
611612
}
612-
pub type PartitionWindowAggStates = IndexMap<PartitionKey, WindowState>;
613+
pub type PartitionWindowAggStates = IndexMap<PartitionKey, WindowState, RandomState>;
613614

614615
/// The IndexMap (i.e. an ordered HashMap) where record batches are separated for each partition.
615-
pub type PartitionBatches = IndexMap<PartitionKey, PartitionBatchState>;
616+
pub type PartitionBatches = IndexMap<PartitionKey, PartitionBatchState, RandomState>;
616617

617618
#[cfg(test)]
618619
mod tests {

datafusion/physical-plan/src/windows/bounded_window_agg_exec.rs

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1054,13 +1054,13 @@ impl BoundedWindowAggStream {
10541054
baseline_metrics: BaselineMetrics,
10551055
search_mode: Box<dyn PartitionSearcher>,
10561056
) -> Result<Self> {
1057-
let state = window_expr.iter().map(|_| IndexMap::new()).collect();
1057+
let state = window_expr.iter().map(|_| IndexMap::default()).collect();
10581058
let empty_batch = RecordBatch::new_empty(Arc::clone(&schema));
10591059
Ok(Self {
10601060
schema,
10611061
input,
10621062
input_buffer: empty_batch,
1063-
partition_buffers: IndexMap::new(),
1063+
partition_buffers: IndexMap::default(),
10641064
window_agg_states: state,
10651065
finished: false,
10661066
window_expr,

0 commit comments

Comments
 (0)