Add per-partition hash tables for partial aggregation

Introduce PartitionAggState to support multiple internal hash tables
in partial aggregation. When enabled via AggregateExec::with_num_agg_partitions(),
input rows are hashed by group keys (using the same hash as RepartitionExec)
and routed to separate smaller hash tables for better cache locality.

Defaults to 1 partition (no behavior change). The optimizer can set
higher values when a hash repartition follows the partial aggregate.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: Dandandan

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

@@ -660,6 +660,12 @@ pub struct AggregateExec {
     /// it remains `Some(..)` to enable dynamic filtering during aggregate execution;
     /// otherwise, it is cleared to `None`.
     dynamic_filter: Option<Arc<AggrDynFilter>>,
+
+    /// Number of internal hash table partitions for partial aggregation.
+    /// When > 1, input rows are hashed by group keys and routed to separate
+    /// smaller hash tables for better cache locality. Only used when mode is
+    /// Partial and input is unordered. Defaults to 1 (single hash table).
+    pub(crate) num_agg_partitions: usize,
 }
 
 impl AggregateExec {
@@ -685,6 +691,7 @@ impl AggregateExec {
             schema: Arc::clone(&self.schema),
             input_schema: Arc::clone(&self.input_schema),
             dynamic_filter: self.dynamic_filter.clone(),
+            num_agg_partitions: self.num_agg_partitions,
         }
     }
 
@@ -705,6 +712,7 @@ impl AggregateExec {
             schema: Arc::clone(&self.schema),
             input_schema: Arc::clone(&self.input_schema),
             dynamic_filter: self.dynamic_filter.clone(),
+            num_agg_partitions: self.num_agg_partitions,
         }
     }
 
@@ -839,6 +847,7 @@ impl AggregateExec {
             input_order_mode,
             cache: Arc::new(cache),
             dynamic_filter: None,
+            num_agg_partitions: 1,
         };
 
         exec.init_dynamic_filter();
@@ -851,6 +860,12 @@ impl AggregateExec {
         &self.mode
     }
 
+    /// Set the number of internal hash table partitions for partial aggregation.
+    pub fn with_num_agg_partitions(mut self, n: usize) -> Self {
+        self.num_agg_partitions = n;
+        self
+    }
+
     /// Set the limit options for this AggExec
     pub fn with_limit_options(mut self, limit_options: Option<LimitOptions>) -> Self {
         self.limit_options = limit_options;