Skip to content

Commit 53afdb9

Browse files
farhan-syahfarhan-syah
committed
refactor(engine): replace long argument lists with params structs
Extends the aggregate/timeseries executor handlers and their dispatch call sites to take a single borrowed params struct instead of long positional argument lists, dropping the accompanying clippy::too_many_arguments allows.
1 parent cb9c91b commit 53afdb9

12 files changed

Lines changed: 339 additions & 176 deletions

File tree

‎nodedb/src/data/executor/dispatch/query.rs‎

Lines changed: 48 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -33,19 +33,21 @@ impl CoreLoop {
3333
grouping_sets,
3434
sort_keys,
3535
} => self.execute_aggregate(
36-
task,
37-
tid,
38-
collection,
39-
input.as_deref(),
40-
group_by,
41-
aggregates,
42-
filters,
43-
having,
44-
*limit,
45-
sub_group_by,
46-
sub_aggregates,
47-
grouping_sets,
48-
sort_keys,
36+
crate::data::executor::handlers::aggregate::exec::AggregateExecInputs {
37+
task,
38+
tid,
39+
collection,
40+
input: input.as_deref(),
41+
group_by,
42+
aggregates,
43+
filters,
44+
having,
45+
limit: *limit,
46+
sub_group_by,
47+
sub_aggregates,
48+
grouping_sets,
49+
sort_keys,
50+
},
4951
),
5052

5153
QueryOp::Exchange(_) => self.response_error(
@@ -235,19 +237,21 @@ impl CoreLoop {
235237
.map(|s| GroupKeySpec::column(s.as_str()))
236238
.collect();
237239
self.execute_aggregate(
238-
task,
239-
tid,
240-
collection,
241-
None,
242-
&group_specs,
243-
aggregates,
244-
filters,
245-
&[],
246-
usize::MAX,
247-
&[],
248-
&[],
249-
&[],
250-
&[],
240+
crate::data::executor::handlers::aggregate::exec::AggregateExecInputs {
241+
task,
242+
tid,
243+
collection,
244+
input: None,
245+
group_by: &group_specs,
246+
aggregates,
247+
filters,
248+
having: &[],
249+
limit: usize::MAX,
250+
sub_group_by: &[],
251+
sub_aggregates: &[],
252+
grouping_sets: &[],
253+
sort_keys: &[],
254+
},
251255
)
252256
}
253257

@@ -258,13 +262,15 @@ impl CoreLoop {
258262
aggregates,
259263
filters,
260264
} => self.execute_partial_aggregate_state(
261-
task,
262-
tid,
263-
collection,
264-
input.as_deref(),
265-
group_by,
266-
aggregates,
267-
filters,
265+
crate::data::executor::handlers::aggregate::state_emit::PartialAggregateStateParams {
266+
task,
267+
tid,
268+
collection,
269+
input: input.as_deref(),
270+
group_by,
271+
aggregates,
272+
filters,
273+
},
268274
),
269275

270276
QueryOp::ShuffleAggregateConsume {
@@ -282,13 +288,15 @@ impl CoreLoop {
282288
.map(|s| GroupKeySpec::column(s.as_str()))
283289
.collect();
284290
self.execute_shuffle_aggregate(
285-
task,
286-
state_path,
287-
&group_specs,
288-
aggregates,
289-
having,
290-
*limit,
291-
sort_keys,
291+
crate::data::executor::handlers::aggregate::shuffle_merge::ShuffleAggregateParams {
292+
task,
293+
state_path,
294+
group_by: &group_specs,
295+
aggregates,
296+
having,
297+
limit: *limit,
298+
sort_keys,
299+
},
292300
)
293301
}
294302

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

Lines changed: 64 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -16,24 +16,46 @@ use nodedb_physical::physical_plan::{AggregateSpec, GroupKeySpec};
1616
use nodedb_query::agg_key::canonical_agg_key;
1717
use nodedb_query::msgpack_scan;
1818

19+
/// Borrowed inputs to [`CoreLoop::execute_aggregate`]: the target collection
20+
/// (or input sub-plan), GROUP BY / aggregate / filter / HAVING / sub-group /
21+
/// grouping-set / sort specs for one aggregate dispatch.
22+
pub(in crate::data::executor) struct AggregateExecInputs<'a> {
23+
pub task: &'a ExecutionTask,
24+
pub tid: u64,
25+
pub collection: &'a str,
26+
pub input: Option<&'a nodedb_physical::physical_plan::PhysicalPlan>,
27+
pub group_by: &'a [GroupKeySpec],
28+
pub aggregates: &'a [AggregateSpec],
29+
pub filters: &'a [u8],
30+
pub having: &'a [u8],
31+
pub limit: usize,
32+
pub sub_group_by: &'a [String],
33+
pub sub_aggregates: &'a [AggregateSpec],
34+
pub grouping_sets: &'a [Vec<u32>],
35+
pub sort_keys: &'a [(String, bool)],
36+
}
37+
1938
impl CoreLoop {
20-
#[allow(clippy::too_many_arguments)]
2139
pub(in crate::data::executor) fn execute_aggregate(
2240
&mut self,
23-
task: &ExecutionTask,
24-
tid: u64,
25-
collection: &str,
26-
input: Option<&nodedb_physical::physical_plan::PhysicalPlan>,
27-
group_by: &[GroupKeySpec],
28-
aggregates: &[AggregateSpec],
29-
filters: &[u8],
30-
having: &[u8],
31-
limit: usize,
32-
sub_group_by: &[String],
33-
sub_aggregates: &[AggregateSpec],
34-
grouping_sets: &[Vec<u32>],
35-
sort_keys: &[(String, bool)],
41+
inputs: AggregateExecInputs<'_>,
3642
) -> Response {
43+
let AggregateExecInputs {
44+
task,
45+
tid,
46+
collection,
47+
input,
48+
group_by,
49+
aggregates,
50+
filters,
51+
having,
52+
limit,
53+
sub_group_by,
54+
sub_aggregates,
55+
grouping_sets,
56+
sort_keys,
57+
} = inputs;
58+
3759
debug!(core = self.core_id, %collection, has_input = input.is_some(), group_fields = group_by.len(), aggs = aggregates.len(), "aggregate");
3860

3961
// Input-sourced aggregate (catalog): the rows come from executing the
@@ -53,18 +75,20 @@ impl CoreLoop {
5375
crate::data::executor::response_codec::decode_response_to_docs(&sub_response)
5476
.unwrap_or_default();
5577
return self.aggregate_over_docs(
56-
task,
57-
collection,
58-
None,
59-
docs,
60-
group_by,
61-
aggregates,
62-
filters,
63-
having,
64-
limit,
65-
sub_group_by,
66-
sub_aggregates,
67-
sort_keys,
78+
super::streaming::over_docs::AggregateOverDocsParams {
79+
task,
80+
collection,
81+
cache_tid: None,
82+
docs,
83+
group_by,
84+
aggregates,
85+
filters,
86+
having,
87+
limit,
88+
sub_group_by,
89+
sub_aggregates,
90+
sort_keys,
91+
},
6892
);
6993
}
7094

@@ -79,15 +103,17 @@ impl CoreLoop {
79103
group_by.iter().filter_map(|s| s.field.clone()).collect();
80104
return super::super::grouping_sets_exec::execute_grouping_sets(
81105
self,
82-
task,
83-
tid,
84-
collection,
85-
&group_fields,
86-
aggregates,
87-
filters,
88-
having,
89-
limit,
90-
grouping_sets,
106+
super::super::grouping_sets_exec::GroupingSetsParams {
107+
task,
108+
tid,
109+
collection,
110+
group_by: &group_fields,
111+
aggregates,
112+
filters,
113+
having,
114+
limit,
115+
grouping_sets,
116+
},
91117
);
92118
}
93119

@@ -302,10 +328,10 @@ impl CoreLoop {
302328
);
303329
}
304330
};
305-
self.aggregate_over_docs(
331+
self.aggregate_over_docs(super::streaming::over_docs::AggregateOverDocsParams {
306332
task,
307333
collection,
308-
Some(tid),
334+
cache_tid: Some(tid),
309335
docs,
310336
group_by,
311337
aggregates,
@@ -315,6 +341,6 @@ impl CoreLoop {
315341
sub_group_by,
316342
sub_aggregates,
317343
sort_keys,
318-
)
344+
})
319345
}
320346
}

‎nodedb/src/data/executor/handlers/aggregate/mod.rs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -16,9 +16,9 @@
1616
//! partial-state producer), and `shuffle_merge` (the partial-state consumer).
1717
1818
mod cache_key;
19-
mod exec;
19+
pub(in crate::data::executor) mod exec;
2020
mod invalidate;
2121
mod rows;
22-
mod shuffle_merge;
23-
mod state_emit;
22+
pub(in crate::data::executor) mod shuffle_merge;
23+
pub(in crate::data::executor) mod state_emit;
2424
mod streaming;

‎nodedb/src/data/executor/handlers/aggregate/shuffle_merge.rs‎

Lines changed: 30 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -100,20 +100,36 @@ pub(in crate::data::executor) fn merge_state_frames(
100100
Ok(merged)
101101
}
102102

103+
/// Borrowed inputs to [`CoreLoop::execute_shuffle_aggregate`]: the staged
104+
/// partial-state frame path plus the GROUP BY / aggregate / HAVING / sort
105+
/// specs needed to merge and finalize it.
106+
pub(in crate::data::executor) struct ShuffleAggregateParams<'a> {
107+
pub task: &'a ExecutionTask,
108+
pub state_path: &'a str,
109+
pub group_by: &'a [GroupKeySpec],
110+
pub aggregates: &'a [AggregateSpec],
111+
pub having: &'a [u8],
112+
pub limit: usize,
113+
pub sort_keys: &'a [(String, bool)],
114+
}
115+
103116
impl CoreLoop {
104117
/// Execute a `ShuffleAggregateConsume`: merge the staged partial states,
105118
/// then finalize / HAVING / sort / LIMIT via the shared finalize tail.
106-
#[allow(clippy::too_many_arguments)]
107119
pub(in crate::data::executor) fn execute_shuffle_aggregate(
108120
&mut self,
109-
task: &ExecutionTask,
110-
state_path: &str,
111-
group_by: &[GroupKeySpec],
112-
aggregates: &[AggregateSpec],
113-
having: &[u8],
114-
limit: usize,
115-
sort_keys: &[(String, bool)],
121+
params: ShuffleAggregateParams<'_>,
116122
) -> Response {
123+
let ShuffleAggregateParams {
124+
task,
125+
state_path,
126+
group_by,
127+
aggregates,
128+
having,
129+
limit,
130+
sort_keys,
131+
} = params;
132+
117133
let merged = match merge_state_frames(Path::new(state_path), group_by, aggregates) {
118134
Ok(m) => m,
119135
Err(e) => {
@@ -127,17 +143,17 @@ impl CoreLoop {
127143
};
128144

129145
// Plain GROUP BY: no sub-groups in the shuffle path.
130-
match self.finalize_groups(
131-
merged,
132-
HashMap::new(),
146+
match self.finalize_groups(super::streaming::finalize::FinalizeGroupsParams {
147+
groups: merged,
148+
sub_groups: HashMap::new(),
133149
group_by,
134150
aggregates,
135151
having,
136152
limit,
137-
&[],
138-
&[],
153+
sub_group_by: &[],
154+
sub_aggregates: &[],
139155
sort_keys,
140-
) {
156+
}) {
141157
Ok(payload) => self.response_with_payload(task, payload),
142158
Err(e) => self.response_error(
143159
task,

0 commit comments

Comments
 (0)