nodedb_physical/physical_plan/query.rs
1// SPDX-License-Identifier: Apache-2.0
2
3//! Query operations (joins, aggregates) dispatched to the Data Plane.
4
5/// Aggregate specification for Data Plane aggregate execution.
6#[derive(
7 Debug,
8 Clone,
9 PartialEq,
10 serde::Serialize,
11 serde::Deserialize,
12 zerompk::ToMessagePack,
13 zerompk::FromMessagePack,
14)]
15pub struct AggregateSpec {
16 pub function: String,
17 /// Internal aggregate key used by HAVING and downstream references.
18 pub alias: String,
19 /// Optional user-facing SQL alias for final output naming.
20 pub user_alias: Option<String>,
21 /// Field name for simple field-based aggregates. `"*"` is used for COUNT(*).
22 pub field: String,
23 /// Optional expression to evaluate per-document before aggregating.
24 pub expr: Option<nodedb_query::expr::SqlExpr>,
25}
26
27#[derive(
28 Debug,
29 Clone,
30 PartialEq,
31 serde::Serialize,
32 serde::Deserialize,
33 zerompk::ToMessagePack,
34 zerompk::FromMessagePack,
35)]
36pub struct JoinProjection {
37 pub source: String,
38 pub output: String,
39}
40
41/// Query-level physical operations (joins, aggregates).
42#[derive(
43 Debug,
44 Clone,
45 PartialEq,
46 serde::Serialize,
47 serde::Deserialize,
48 zerompk::ToMessagePack,
49 zerompk::FromMessagePack,
50)]
51pub enum QueryOp {
52 /// Aggregate: GROUP BY + aggregate functions.
53 Aggregate {
54 collection: String,
55 group_by: Vec<String>,
56 aggregates: Vec<AggregateSpec>,
57 filters: Vec<u8>,
58 /// HAVING predicates applied post-aggregation.
59 having: Vec<u8>,
60 limit: usize,
61 sub_group_by: Vec<String>,
62 sub_aggregates: Vec<AggregateSpec>,
63 /// ROLLUP / CUBE / GROUPING SETS expansion. Each inner `Vec<u32>` is
64 /// one grouping set — the indices into `group_by` that are *present*
65 /// (non-NULL) for rows in that set. Empty outer vec = plain single-set
66 /// GROUP BY (no null-filling needed).
67 grouping_sets: Vec<Vec<u32>>,
68 /// Post-aggregation sort keys: `(column_name, ascending)`.
69 /// Empty = preserve executor's natural order (hash-map iteration
70 /// for plain GROUP BY). The executor applies the sort after all
71 /// groups are finalized and HAVING is filtered.
72 #[serde(default)]
73 sort_keys: Vec<(String, bool)>,
74 },
75
76 /// Partial aggregate: each core computes locally, Control Plane merges.
77 PartialAggregate {
78 collection: String,
79 group_by: Vec<String>,
80 aggregates: Vec<AggregateSpec>,
81 filters: Vec<u8>,
82 },
83
84 /// Hash join: build hash map on right, probe with left.
85 HashJoin {
86 left_collection: String,
87 right_collection: String,
88 left_alias: Option<String>,
89 right_alias: Option<String>,
90 on: Vec<(String, String)>,
91 join_type: String,
92 limit: usize,
93 /// Post-join GROUP BY columns (empty = no aggregation).
94 post_group_by: Vec<String>,
95 /// Post-join aggregates: (op, field) pairs (empty = no aggregation).
96 post_aggregates: Vec<(String, String)>,
97 /// Post-join projection: column names to keep (empty = all).
98 projection: Vec<JoinProjection>,
99 /// Post-join WHERE filter predicates (MessagePack).
100 post_filters: Vec<u8>,
101 /// Inline left sub-plan for multi-way joins. When set, the executor
102 /// runs this sub-plan first and uses its result as the left side
103 /// instead of scanning `left_collection`.
104 inline_left: Option<Box<crate::physical_plan::PhysicalPlan>>,
105 /// Inline right sub-plan for scalar subqueries or other materialized
106 /// small-side inputs. The Control Plane executes this plan first,
107 /// merges it if needed, then embeds the result into `BroadcastJoin`.
108 inline_right: Option<Box<crate::physical_plan::PhysicalPlan>>,
109 /// Bitmap-producer sub-plan for the left side. When set, the executor
110 /// executes this plan first, collects surrogates from all returned rows,
111 /// and injects the resulting bitmap into the probe-side prefilter before
112 /// scanning. `None` = no bitmap pushdown for the left side.
113 inline_left_bitmap: Option<Box<crate::physical_plan::PhysicalPlan>>,
114 /// Bitmap-producer sub-plan for the right side. Same semantics as
115 /// `inline_left_bitmap` but applied to the right (probe) collection.
116 inline_right_bitmap: Option<Box<crate::physical_plan::PhysicalPlan>>,
117 },
118
119 /// Inline hash join: both sides are pre-gathered msgpack data.
120 /// Used for multi-way joins where the left side is the result of another join.
121 InlineHashJoin {
122 /// Left side: msgpack array of maps (from inner join result).
123 left_data: Vec<u8>,
124 /// Right side: raw broadcast scan data.
125 right_data: Vec<u8>,
126 right_alias: Option<String>,
127 on: Vec<(String, String)>,
128 join_type: String,
129 limit: usize,
130 projection: Vec<JoinProjection>,
131 post_filters: Vec<u8>,
132 },
133
134 /// Broadcast join: small side serialized in the plan.
135 BroadcastJoin {
136 large_collection: String,
137 small_collection: String,
138 large_alias: Option<String>,
139 small_alias: Option<String>,
140 broadcast_data: Vec<u8>,
141 on: Vec<(String, String)>,
142 join_type: String,
143 limit: usize,
144 /// Post-join GROUP BY columns (empty = no aggregation).
145 post_group_by: Vec<String>,
146 /// Post-join aggregates: (op, field) pairs (empty = no aggregation).
147 post_aggregates: Vec<(String, String)>,
148 /// Post-join projection: column names to keep (empty = all).
149 projection: Vec<JoinProjection>,
150 /// Post-join WHERE filter predicates (MessagePack).
151 post_filters: Vec<u8>,
152 },
153
154 /// Shuffle join: repartition by join key via SPSC.
155 ShuffleJoin {
156 left_collection: String,
157 right_collection: String,
158 on: Vec<(String, String)>,
159 join_type: String,
160 limit: usize,
161 target_core: usize,
162 },
163
164 /// Nested loop join: fallback for non-equi joins.
165 NestedLoopJoin {
166 left_collection: String,
167 right_collection: String,
168 /// Join condition as serialized `Vec<ScanFilter>`.
169 condition: Vec<u8>,
170 join_type: String,
171 limit: usize,
172 },
173
174 /// Sort-merge join: both sides pre-sorted by join key.
175 /// Optimal when both collections have index-ordered scans or
176 /// when the planner sorts both sides before joining.
177 SortMergeJoin {
178 left_collection: String,
179 right_collection: String,
180 on: Vec<(String, String)>,
181 join_type: String,
182 limit: usize,
183 /// If true, both sides are assumed pre-sorted by join key (skip sort phase).
184 pre_sorted: bool,
185 },
186
187 /// Multi-facet aggregation: compute facet counts for multiple fields
188 /// in a single query, sharing the filter evaluation across all facets.
189 FacetCounts {
190 collection: String,
191 /// Serialized `Vec<ScanFilter>` predicates (MessagePack).
192 filters: Vec<u8>,
193 /// Field names to facet on (each produces a `[{value, count}]` array).
194 fields: Vec<String>,
195 /// Maximum number of values to return per facet field (0 = unlimited).
196 limit_per_facet: usize,
197 },
198
199 /// Recursive CTE: iterative fixed-point execution.
200 ///
201 /// Executes the base query once, then repeatedly executes the recursive
202 /// query using the previous iteration's results as the working table,
203 /// until no new rows are produced (fixed point).
204 RecursiveScan {
205 /// Collection for the recursive scan.
206 collection: String,
207 /// Base query filters (seeded once).
208 base_filters: Vec<u8>,
209 /// Recursive step filters (applied to working table each iteration).
210 recursive_filters: Vec<u8>,
211 /// Equi-join link for tree-traversal recursion:
212 /// `(collection_field, working_table_field)`.
213 /// Each iteration finds rows where `collection_field` value
214 /// matches a `working_table_field` value from the previous iteration.
215 join_link: Option<(String, String)>,
216 /// Maximum iterations to prevent infinite loops. Default: 100.
217 max_iterations: usize,
218 /// Whether to deduplicate results (UNION vs UNION ALL).
219 distinct: bool,
220 limit: usize,
221 },
222
223 /// Value-generating recursive CTE: iterative expression evaluation.
224 ///
225 /// No collection is needed. The executor evaluates the anchor expressions
226 /// once to produce the first row, then repeatedly applies the step
227 /// expressions to the previous row until the condition becomes false,
228 /// a fixed point is reached, or `max_depth` is exceeded (typed error).
229 RecursiveValue {
230 /// CTE name (used in error messages).
231 cte_name: String,
232 /// Column names (length == `init_exprs.len()` == `step_exprs.len()`).
233 columns: Vec<String>,
234 /// Anchor SELECT expressions as raw SQL text.
235 init_exprs: Vec<String>,
236 /// Recursive step SELECT expressions as raw SQL text.
237 step_exprs: Vec<String>,
238 /// Optional WHERE condition as raw SQL text.
239 condition: Option<String>,
240 /// Maximum iterations before returning a depth-exceeded error.
241 max_depth: usize,
242 /// Whether to deduplicate (UNION vs UNION ALL).
243 distinct: bool,
244 },
245
246 /// LATERAL equi-correlated top-K: scan `inner_collection` once per outer
247 /// row, applying the equi-correlation as an equality filter, then return
248 /// the top `inner_limit` rows ordered by `inner_order_by`.
249 ///
250 /// The executor first runs `outer_plan` to materialise outer rows, then
251 /// for each outer row injects the equi-correlation as an equality filter
252 /// on `inner_collection`, applies `inner_order_by`, and keeps the top
253 /// `inner_limit` rows. Output rows are `(outer_row merged with inner_row)`.
254 ///
255 /// `correlation_keys` are `(outer_col, inner_col)` equi-join pairs.
256 LateralTopK {
257 /// Sub-plan that produces the outer (driving) rows.
258 outer_plan: Box<crate::physical_plan::PhysicalPlan>,
259 /// Alias qualifying the outer columns in output rows.
260 outer_alias: String,
261 /// Inner collection to scan per outer row.
262 inner_collection: String,
263 /// Non-correlated filters applied to every inner scan (msgpack bytes).
264 inner_filters: Vec<u8>,
265 /// Sort keys for the inner per-outer-row result.
266 /// Each entry is `(field_name, ascending)`.
267 inner_order_by: Vec<(String, bool)>,
268 /// Maximum inner rows per outer row.
269 inner_limit: usize,
270 /// Equi-join pairs `(outer_col, inner_col)`.
271 correlation_keys: Vec<(String, String)>,
272 /// Alias qualifying inner columns in output rows.
273 lateral_alias: String,
274 /// Output projection (empty = all columns).
275 projection: Vec<JoinProjection>,
276 /// LEFT join semantics: preserve outer rows when inner is empty.
277 left_join: bool,
278 },
279
280 /// LATERAL nested-loop: for each outer row, re-execute the inner plan
281 /// with correlated values injected as additional equality filters.
282 ///
283 /// The executor runs `outer_plan`, then for each outer row reads the
284 /// `outer_col` values from `correlation_predicates`, appends equality
285 /// filters on the corresponding `inner_col` fields, and scans
286 /// `inner_collection`.
287 LateralLoop {
288 /// Sub-plan that produces the outer (driving) rows.
289 outer_plan: Box<crate::physical_plan::PhysicalPlan>,
290 /// Alias qualifying the outer columns in output rows.
291 outer_alias: String,
292 /// Inner collection to scan per outer row.
293 inner_collection: String,
294 /// Base inner filters (non-correlated, msgpack bytes).
295 inner_filters: Vec<u8>,
296 /// Correlated predicates: `(inner_field, outer_field)`.
297 correlation_predicates: Vec<(String, String)>,
298 /// Alias qualifying inner columns in output rows.
299 lateral_alias: String,
300 /// Output projection (empty = all columns).
301 projection: Vec<JoinProjection>,
302 /// LEFT join semantics.
303 left_join: bool,
304 /// Hard cap on outer rows. Queries that exceed this cap return a
305 /// typed `LateralCapExceeded` error instead of silently truncating.
306 outer_row_cap: usize,
307 },
308}