perf: add prefetching for aggregate multi group by [WIP]

Cargo.toml

@@ -156,7 +156,7 @@ flate2 = "1.1.5"
 futures = "0.3"
 glob = "0.3.0"
 half = { version = "2.7.0", default-features = false }
-hashbrown = { version = "0.14.5", features = ["raw"] }
+hashbrown = { features = ["raw"], git = "https://github.com/rluvaton/hashbrown", rev = "dbdac43f537223a469746f518929b542f3b8933d" }
 hex = { version = "0.4.3" }
 indexmap = "2.12.1"
 insta = { version = "1.45.0", features = ["glob", "filters"] }

datafusion/common/src/config.rs

@@ -649,6 +649,10 @@ config_namespace! {
         /// # Default
         /// `false` — ANSI SQL mode is disabled by default.
         pub enable_ansi_mode: bool, default = false
+
+        pub agg_prefetch_elements: usize, default = 1
+        pub agg_prefetch_locality: usize, default = 3
+        pub agg_prefetch_read: bool, default = false
     }
 }
 

datafusion/physical-plan/Cargo.toml

@@ -70,6 +70,7 @@ log = { workspace = true }
 parking_lot = { workspace = true }
 pin-project-lite = "^0.2.7"
 tokio = { workspace = true }
+cache-size = "0.7.0"
 
 [dev-dependencies]
 criterion = { workspace = true, features = ["async_futures"] }

datafusion/physical-plan/src/aggregates/group_values/mod.rs

@@ -25,7 +25,7 @@ use arrow::array::types::{
 use arrow::array::{ArrayRef, downcast_primitive};
 use arrow::datatypes::{DataType, SchemaRef, TimeUnit};
 use datafusion_common::Result;
-
+use datafusion_execution::TaskContext;
 use datafusion_expr::EmitTo;
 
 pub mod multi_group_by;
@@ -134,6 +134,7 @@ pub trait GroupValues: Send {
 pub fn new_group_values(
     schema: SchemaRef,
     group_ordering: &GroupOrdering,
+    ctx: Option<&TaskContext>
 ) -> Result<Box<dyn GroupValues>> {
     if schema.fields.len() == 1 {
         let d = schema.fields[0].data_type();
@@ -202,9 +203,9 @@ pub fn new_group_values(
 
     if multi_group_by::supported_schema(schema.as_ref()) {
         if matches!(group_ordering, GroupOrdering::None) {
-            Ok(Box::new(GroupValuesColumn::<false>::try_new(schema)?))
+            Ok(Box::new(GroupValuesColumn::<false>::try_new(schema, ctx)?))
         } else {
-            Ok(Box::new(GroupValuesColumn::<true>::try_new(schema)?))
+            Ok(Box::new(GroupValuesColumn::<true>::try_new(schema, ctx)?))
         }
     } else {
         Ok(Box::new(GroupValuesRows::try_new(schema)?))