refactor(aggregate): bundle cache key inputs into a params struct

Replace aggregate_cache_key's nine positional arguments (and its
too_many_arguments allow) with an AggregateCacheKeyInputs struct,
naming the complete query shape that constitutes the cache identity.

Author: farhan-syah

nodedb/src/data/executor/handlers/aggregate/cache_key.rs

@@ -46,19 +46,37 @@ fn aggregate_specs_key(aggregates: &[AggregateSpec]) -> String {
         .join(",")
 }
 
-#[allow(clippy::too_many_arguments)] // complete query shape is the cache identity
+/// Complete query shape that constitutes an aggregate result's cache identity.
+/// Every field participates in the key — two aggregate queries that differ in
+/// any one of them (output aliases, `LIMIT`, `ORDER BY`, sub-aggregation) must
+/// not collide on a shared cache entry.
+pub(super) struct AggregateCacheKeyInputs<'a> {
+    pub database_id: u64,
+    pub tid: u64,
+    pub collection: &'a str,
+    pub group_by: &'a [GroupKeySpec],
+    pub aggregates: &'a [AggregateSpec],
+    pub sub_group_by: &'a [String],
+    pub sub_aggregates: &'a [AggregateSpec],
+    pub limit: usize,
+    pub sort_keys: &'a [(String, bool)],
+}
+
 pub(super) fn aggregate_cache_key(
-    database_id: u64,
-    tid: u64,
-    collection: &str,
-    group_by: &[GroupKeySpec],
-    aggregates: &[AggregateSpec],
-    sub_group_by: &[String],
-    sub_aggregates: &[AggregateSpec],
-    limit: usize,
-    sort_keys: &[(String, bool)],
+    inputs: AggregateCacheKeyInputs<'_>,
 ) -> (crate::types::DatabaseId, crate::types::TenantId, String) {
     use std::fmt::Write;
+    let AggregateCacheKeyInputs {
+        database_id,
+        tid,
+        collection,
+        group_by,
+        aggregates,
+        sub_group_by,
+        sub_aggregates,
+        limit,
+        sort_keys,
+    } = inputs;
     let mut rest = format!(
         "{collection}\0{}\0{}",
         group_specs_key(group_by),

nodedb/src/data/executor/handlers/aggregate/exec.rs

@@ -6,7 +6,7 @@
 
 use tracing::debug;
 
-use super::cache_key::{aggregate_cache_key, legacy_aggregate_pairs};
+use super::cache_key::{AggregateCacheKeyInputs, aggregate_cache_key, legacy_aggregate_pairs};
 use super::rows::{apply_user_aliases_to_rows, sort_aggregated_rows};
 use crate::bridge::envelope::{ErrorCode, Response};
 use crate::bridge::scan_filter::ScanFilter;
@@ -119,8 +119,8 @@ impl CoreLoop {
 
         // Fast path: incremental aggregate cache.
         if filters.is_empty() && having.is_empty() {
-            let cache_key = aggregate_cache_key(
-                task.request.database_id.as_u64(),
+            let cache_key = aggregate_cache_key(AggregateCacheKeyInputs {
+                database_id: task.request.database_id.as_u64(),
                 tid,
                 collection,
                 group_by,
@@ -129,7 +129,7 @@ impl CoreLoop {
                 sub_aggregates,
                 limit,
                 sort_keys,
-            );
+            });
             if let Some(cached) = self.aggregate_cache.get(&cache_key) {
                 debug!(core = self.core_id, %collection, "aggregate cache hit");
                 return self.response_with_payload(task, cached.clone());
@@ -273,8 +273,8 @@ impl CoreLoop {
                 ) {
                     Ok(payload) => {
                         if filters.is_empty() && having.is_empty() {
-                            let cache_key = aggregate_cache_key(
-                                task.request.database_id.as_u64(),
+                            let cache_key = aggregate_cache_key(AggregateCacheKeyInputs {
+                                database_id: task.request.database_id.as_u64(),
                                 tid,
                                 collection,
                                 group_by,
@@ -283,7 +283,7 @@ impl CoreLoop {
                                 sub_aggregates,
                                 limit,
                                 sort_keys,
-                            );
+                            });
                             if self.aggregate_cache.len() < 256 {
                                 self.aggregate_cache.insert(cache_key, payload.clone());
                             }

nodedb/src/data/executor/handlers/aggregate/streaming/over_docs.rs

@@ -3,7 +3,7 @@
 //! `aggregate_over_docs`: orchestrate accumulate + finalize over an
 //! already-materialized doc set, layering the per-shard result cache on top.
 
-use super::super::cache_key::aggregate_cache_key;
+use super::super::cache_key::{AggregateCacheKeyInputs, aggregate_cache_key};
 use crate::bridge::envelope::{ErrorCode, Response};
 use crate::data::executor::core_loop::CoreLoop;
 use crate::data::executor::task::ExecutionTask;
@@ -104,8 +104,8 @@ impl CoreLoop {
                     && filters.is_empty()
                     && having.is_empty()
                 {
-                    let cache_key = aggregate_cache_key(
-                        task.request.database_id.as_u64(),
+                    let cache_key = aggregate_cache_key(AggregateCacheKeyInputs {
+                        database_id: task.request.database_id.as_u64(),
                         tid,
                         collection,
                         group_by,
@@ -114,7 +114,7 @@ impl CoreLoop {
                         sub_aggregates,
                         limit,
                         sort_keys,
-                    );
+                    });
                     if self.aggregate_cache.len() < 256 {
                         self.aggregate_cache.insert(cache_key, payload.clone());
                     }