Set aggregation hash seed (#16165)

Author: ctsk

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

@@ -270,7 +270,7 @@ impl<const STREAMING: bool> GroupValuesColumn<STREAMING> {
             map_size: 0,
             group_values: vec![],
             hashes_buffer: Default::default(),
-            random_state: Default::default(),
+            random_state: crate::aggregates::AGGREGATION_HASH_SEED,
         })
     }
 

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

@@ -106,7 +106,7 @@ impl GroupValuesRows {
             group_values: None,
             hashes_buffer: Default::default(),
             rows_buffer,
-            random_state: Default::default(),
+            random_state: crate::aggregates::AGGREGATION_HASH_SEED,
         })
     }
 }

datafusion/physical-plan/src/aggregates/group_values/single_group_by/primitive.rs

@@ -105,7 +105,7 @@ impl<T: ArrowPrimitiveType> GroupValuesPrimitive<T> {
             map: HashTable::with_capacity(128),
             values: Vec::with_capacity(128),
             null_group: None,
-            random_state: Default::default(),
+            random_state: crate::aggregates::AGGREGATION_HASH_SEED,
         }
     }
 }

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

@@ -58,6 +58,10 @@ mod row_hash;
 mod topk;
 mod topk_stream;
 
+/// Hard-coded seed for aggregations to ensure hash values differ from `RepartitionExec`, avoiding collisions.
+const AGGREGATION_HASH_SEED: ahash::RandomState =
+    ahash::RandomState::with_seeds('A' as u64, 'G' as u64, 'G' as u64, 'R' as u64);
+
 /// Aggregation modes
 ///
 /// See [`Accumulator::state`] for background information on multi-phase