powdb_query/executor/plan_exec.rs
1//! The execute_plan method and associated helpers.
2
3use crate::ast::*;
4use crate::plan::*;
5use crate::result::{QueryError, QueryResult};
6use powdb_storage::catalog::Catalog;
7use powdb_storage::row::{decode_column, decode_row, patch_var_column_in_place, RowLayout};
8use powdb_storage::types::*;
9use std::cmp::Reverse;
10use std::collections::BinaryHeap;
11
12use super::compiled::*;
13use super::eval::*;
14use super::row_body_base;
15use super::{check_join_limit, Engine, MAX_SORT_ROWS};
16use powdb_storage::view::ViewDef;
17
18impl Engine {
19 pub fn execute_plan(&mut self, plan: &PlanNode) -> Result<QueryResult, QueryError> {
20 match plan {
21 PlanNode::SeqScan { table } => {
22 // Auto-refresh dirty materialized views on read.
23 if self.view_registry.is_dirty(table) {
24 self.refresh_view(table)?;
25 }
26 let schema = self
27 .catalog
28 .schema(table)
29 .ok_or_else(|| QueryError::TableNotFound(table.to_string()))?
30 .clone();
31 let columns: Vec<String> = schema.columns.iter().map(|c| c.name.clone()).collect();
32 let rows: Vec<Vec<Value>> = self
33 .catalog
34 .scan(table)
35 .map_err(|e| QueryError::StorageError(e.to_string()))?
36 .map(|(_, row)| row)
37 .collect();
38 Ok(QueryResult::Rows { columns, rows })
39 }
40
41 PlanNode::Filter { input, predicate } => {
42 // Materialize any IN-subqueries in the predicate before the
43 // scan loop — the closure can't call back into the engine.
44 // Correlated subqueries are left in place for per-row eval.
45 let materialized;
46 let predicate = if contains_subquery(predicate) {
47 materialized = self.materialize_subqueries(predicate)?;
48 &materialized
49 } else {
50 predicate
51 };
52
53 // Correlated subquery path: per-row materialisation.
54 if contains_subquery(predicate) {
55 let result = self.execute_plan(input)?;
56 return match result {
57 QueryResult::Rows { columns, rows } => {
58 let mut filtered = Vec::new();
59 for row in rows {
60 let row_pred =
61 self.materialize_correlated_for_row(predicate, &row, &columns)?;
62 if eval_predicate(&row_pred, &row, &columns) {
63 filtered.push(row);
64 }
65 }
66 Ok(QueryResult::Rows {
67 columns,
68 rows: filtered,
69 })
70 }
71 _ => Err("filter requires row input".into()),
72 };
73 }
74
75 // Fast path: fuse Filter + SeqScan into a zero-copy streaming
76 // loop. Uses decode_column() to evaluate the predicate on only
77 // the columns it references, avoiding heap allocations for
78 // String/Bytes columns that aren't part of the filter.
79 if let PlanNode::SeqScan { table } = input.as_ref() {
80 // Auto-refresh dirty materialized views.
81 if self.view_registry.is_dirty(table) {
82 self.refresh_view(table)?;
83 }
84 let schema = self
85 .catalog
86 .schema(table)
87 .ok_or_else(|| QueryError::TableNotFound(table.to_string()))?
88 .clone();
89 let columns: Vec<String> =
90 schema.columns.iter().map(|c| c.name.clone()).collect();
91 let fast = FastLayout::new(&schema);
92 let row_layout = RowLayout::new(&schema);
93 // Mission F: pre-size to skip the first 4 Vec doublings
94 // (4 → 8 → 16 → 32 → 64). On a 100K-row scan with 30%
95 // selectivity that's ~4 fewer reallocations + memcpys.
96 let mut rows: Vec<Vec<Value>> = Vec::with_capacity(64);
97
98 // Try compiled predicate for the filter check (handles
99 // int leaves, string-eq leaves, and And conjunctions).
100 if let Some(compiled) = compile_predicate(predicate, &columns, &fast, &schema) {
101 self.catalog
102 .for_each_row_raw(table, |_rid, data| {
103 if compiled(data) {
104 rows.push(decode_row(&schema, data));
105 }
106 })
107 .map_err(|e| QueryError::StorageError(e.to_string()))?;
108 } else {
109 let pred_cols = predicate_column_indices(predicate, &columns);
110 self.catalog
111 .for_each_row_raw(table, |_rid, data| {
112 let pred_row =
113 decode_selective(&schema, &row_layout, data, &pred_cols);
114 if eval_predicate(predicate, &pred_row, &columns) {
115 rows.push(decode_row(&schema, data));
116 }
117 })
118 .map_err(|e| QueryError::StorageError(e.to_string()))?;
119 }
120
121 return Ok(QueryResult::Rows { columns, rows });
122 }
123
124 // General path: materialise then filter.
125 let result = self.execute_plan(input)?;
126 match result {
127 QueryResult::Rows { columns, rows } => {
128 let filtered: Vec<Vec<Value>> = rows
129 .into_iter()
130 .filter(|row| eval_predicate(predicate, row, &columns))
131 .collect();
132 Ok(QueryResult::Rows {
133 columns,
134 rows: filtered,
135 })
136 }
137 _ => Err("filter requires row input".into()),
138 }
139 }
140
141 PlanNode::Project { input, fields } => {
142 // Fast path: Project over IndexScan — decode only projected
143 // columns from raw bytes instead of full decode_row.
144 if let PlanNode::IndexScan { table, column, key } = input.as_ref() {
145 let schema = self
146 .catalog
147 .schema(table)
148 .ok_or_else(|| QueryError::TableNotFound(table.to_string()))?
149 .clone();
150 let all_columns: Vec<String> =
151 schema.columns.iter().map(|c| c.name.clone()).collect();
152 let key_value = literal_to_value(key)?;
153 let tbl = self
154 .catalog
155 .get_table(table)
156 .ok_or_else(|| QueryError::TableNotFound(table.to_string()))?;
157
158 let proj_columns: Vec<String> = fields
159 .iter()
160 .map(|f| {
161 f.alias.clone().unwrap_or_else(|| match &f.expr {
162 Expr::Field(name) => name.clone(),
163 _ => "?".into(),
164 })
165 })
166 .collect();
167
168 // Determine which column indices the projection needs
169 let proj_indices: Vec<usize> = fields
170 .iter()
171 .filter_map(|f| {
172 if let Expr::Field(name) = &f.expr {
173 all_columns.iter().position(|c| c == name)
174 } else {
175 None
176 }
177 })
178 .collect();
179
180 if tbl.has_index(column) {
181 let layout = RowLayout::new(&schema);
182 let rids = tbl.index_lookup_all(column, &key_value);
183 let mut rows: Vec<Vec<Value>> = Vec::with_capacity(rids.len());
184 for rid in rids {
185 if let Some(data) = tbl.heap.get(rid) {
186 let row: Vec<Value> = proj_indices
187 .iter()
188 .map(|&ci| decode_column(&schema, &layout, &data, ci))
189 .collect();
190 rows.push(row);
191 }
192 }
193 return Ok(QueryResult::Rows {
194 columns: proj_columns,
195 rows,
196 });
197 }
198 }
199
200 // Fast path: Project(Limit(Sort(Filter(SeqScan)))) — bounded
201 // top-N heap. Decodes only the sort key + projected columns,
202 // keeps at most `limit` rows in a heap. Also handles the
203 // Project(Limit(Sort(SeqScan))) variant (no filter).
204 if let PlanNode::Limit {
205 input: inner,
206 count: limit_expr,
207 } = input.as_ref()
208 {
209 if let PlanNode::Sort {
210 input: sort_input,
211 keys,
212 } = inner.as_ref()
213 {
214 // Fast path only for single-key sorts
215 if keys.len() == 1 {
216 let sort_field = &keys[0].field;
217 let descending = keys[0].descending;
218 let limit = match limit_expr {
219 Expr::Literal(Literal::Int(v)) if *v >= 0 => *v as usize,
220 _ => usize::MAX,
221 };
222 let (table_opt, pred_opt): (Option<&str>, Option<&Expr>) =
223 match sort_input.as_ref() {
224 PlanNode::SeqScan { table } => (Some(table.as_str()), None),
225 PlanNode::Filter {
226 input: fi,
227 predicate,
228 } => {
229 if let PlanNode::SeqScan { table } = fi.as_ref() {
230 (Some(table.as_str()), Some(predicate))
231 } else {
232 (None, None)
233 }
234 }
235 _ => (None, None),
236 };
237 if let Some(table) = table_opt {
238 if let Some(result) = self.project_filter_sort_limit_fast(
239 table, fields, sort_field, descending, limit, pred_opt,
240 )? {
241 return Ok(result);
242 }
243 }
244 }
245 }
246 // Fast path: Project(Limit(Filter(SeqScan))) — stream,
247 // decode only projected columns, stop at limit.
248 if let PlanNode::Filter {
249 input: fi,
250 predicate,
251 } = inner.as_ref()
252 {
253 if let PlanNode::SeqScan { table } = fi.as_ref() {
254 let limit = match limit_expr {
255 Expr::Literal(Literal::Int(v)) if *v >= 0 => *v as usize,
256 _ => usize::MAX,
257 };
258 if let Some(result) = self.project_filter_limit_fast(
259 table,
260 fields,
261 limit,
262 Some(predicate),
263 )? {
264 return Ok(result);
265 }
266 }
267 }
268 // Fast path: Project(Limit(SeqScan)) — stream, no filter.
269 if let PlanNode::SeqScan { table } = inner.as_ref() {
270 let limit = match limit_expr {
271 Expr::Literal(Literal::Int(v)) if *v >= 0 => *v as usize,
272 _ => usize::MAX,
273 };
274 if let Some(result) =
275 self.project_filter_limit_fast(table, fields, limit, None)?
276 {
277 return Ok(result);
278 }
279 }
280 }
281
282 // Mission D4: Project(Filter(SeqScan)) without Limit. Reuses
283 // `project_filter_limit_fast` with limit = usize::MAX so the
284 // hot loop decodes only projected columns and uses the
285 // compiled predicate. Previously this fell through to the
286 // generic Filter branch which materialised every column via
287 // `decode_row` then re-projected — quadratic work.
288 //
289 // multi_col_and_filter (`U filter .age > 30 and .status =
290 // "active" { .name, .age }`) was 6.18ms (0.7x SQLite) and
291 // is the load-bearing workload for this fast path.
292 if let PlanNode::Filter {
293 input: fi,
294 predicate,
295 } = input.as_ref()
296 {
297 if let PlanNode::SeqScan { table } = fi.as_ref() {
298 if let Some(result) = self.project_filter_limit_fast(
299 table,
300 fields,
301 usize::MAX,
302 Some(predicate),
303 )? {
304 return Ok(result);
305 }
306 }
307 }
308
309 // Mission D4: Project(SeqScan) without Filter or Limit.
310 // Decode only projected columns; the previous fall-through
311 // built full Vec<Value> rows then re-projected.
312 if let PlanNode::SeqScan { table } = input.as_ref() {
313 if let Some(result) =
314 self.project_filter_limit_fast(table, fields, usize::MAX, None)?
315 {
316 return Ok(result);
317 }
318 }
319
320 let result = self.execute_plan(input)?;
321 match result {
322 QueryResult::Rows { columns, rows } => {
323 let proj_columns: Vec<String> = fields
324 .iter()
325 .map(|f| {
326 f.alias.clone().unwrap_or_else(|| match &f.expr {
327 Expr::Field(name) => name.clone(),
328 // Mission E1.2: `{ u.name }` projects as the
329 // qualified column name so callers can still
330 // disambiguate across the join output.
331 Expr::QualifiedField { qualifier, field } => {
332 format!("{qualifier}.{field}")
333 }
334 _ => "?".into(),
335 })
336 })
337 .collect();
338 let proj_rows: Vec<Vec<Value>> = rows
339 .iter()
340 .map(|row| {
341 fields
342 .iter()
343 .map(|f| eval_expr(&f.expr, row, &columns))
344 .collect()
345 })
346 .collect();
347 Ok(QueryResult::Rows {
348 columns: proj_columns,
349 rows: proj_rows,
350 })
351 }
352 _ => Err("project requires row input".into()),
353 }
354 }
355
356 PlanNode::Sort { input, keys } => {
357 let result = self.execute_plan(input)?;
358 match result {
359 QueryResult::Rows { columns, mut rows } => {
360 // WS2: row-count cap is a cheap secondary guard; the
361 // byte budget is the real OOM defense for the sort
362 // buffer (a few very large rows pass the row cap).
363 if rows.len() > MAX_SORT_ROWS {
364 return Err(QueryError::SortLimitExceeded);
365 }
366 self.charge_rows(&rows)?;
367 let key_indices: Vec<(usize, bool)> = keys
368 .iter()
369 .map(|k| {
370 columns
371 .iter()
372 .position(|c| c == &k.field)
373 .map(|idx| (idx, k.descending))
374 .ok_or_else(|| QueryError::ColumnNotFound {
375 table: String::new(),
376 column: k.field.clone(),
377 })
378 })
379 .collect::<Result<_, QueryError>>()?;
380 rows.sort_by(|a, b| {
381 for &(col_idx, descending) in &key_indices {
382 let cmp = a[col_idx].cmp(&b[col_idx]);
383 let cmp = if descending { cmp.reverse() } else { cmp };
384 if cmp != std::cmp::Ordering::Equal {
385 return cmp;
386 }
387 }
388 std::cmp::Ordering::Equal
389 });
390 Ok(QueryResult::Rows { columns, rows })
391 }
392 _ => Err("sort requires row input".into()),
393 }
394 }
395
396 PlanNode::Limit { input, count } => {
397 let result = self.execute_plan(input)?;
398 let n = match count {
399 Expr::Literal(Literal::Int(v)) => *v as usize,
400 _ => return Err("limit must be integer literal".into()),
401 };
402 match result {
403 QueryResult::Rows { columns, rows } => Ok(QueryResult::Rows {
404 columns,
405 rows: rows.into_iter().take(n).collect(),
406 }),
407 _ => Err("limit requires row input".into()),
408 }
409 }
410
411 PlanNode::Offset { input, count } => {
412 let result = self.execute_plan(input)?;
413 let n = match count {
414 Expr::Literal(Literal::Int(v)) => *v as usize,
415 _ => return Err("offset must be integer literal".into()),
416 };
417 match result {
418 QueryResult::Rows { columns, rows } => Ok(QueryResult::Rows {
419 columns,
420 rows: rows.into_iter().skip(n).collect(),
421 }),
422 _ => Err("offset requires row input".into()),
423 }
424 }
425
426 PlanNode::Aggregate {
427 input,
428 function,
429 field,
430 } => {
431 // Fast path: count() over SeqScan — count rows without any decode
432 if *function == AggFunc::Count {
433 if let PlanNode::SeqScan { table } = input.as_ref() {
434 // Auto-refresh a dirty materialized view before
435 // counting it — otherwise count(View) returns stale
436 // data after an underlying mutation (F3).
437 if self.view_registry.is_dirty(table) {
438 self.refresh_view(table)?;
439 }
440 let mut count: i64 = 0;
441 self.catalog
442 .for_each_row_raw(table, |_rid, _data| {
443 count += 1;
444 })
445 .map_err(|e| QueryError::StorageError(e.to_string()))?;
446 return Ok(QueryResult::Scalar(Value::Int(count)));
447 }
448 // Fast path: count() over Filter(SeqScan) — try compiled
449 // predicate first, fall back to decode_column path.
450 // Skip a predicate carrying a subquery: the raw-bytes
451 // evaluators here don't materialise subqueries, so
452 // `count(T filter .x in (...))` would silently count 0
453 // (F1). Falling through routes it to the generic path
454 // that resolves the subquery correctly.
455 if let PlanNode::Filter {
456 input: inner,
457 predicate,
458 } = input.as_ref()
459 {
460 if let PlanNode::SeqScan { table } = inner.as_ref() {
461 if self.view_registry.is_dirty(table) {
462 self.refresh_view(table)?;
463 }
464 }
465 if let (PlanNode::SeqScan { table }, false) =
466 (inner.as_ref(), contains_subquery(predicate))
467 {
468 let schema = self
469 .catalog
470 .schema(table)
471 .ok_or_else(|| QueryError::TableNotFound(table.to_string()))?
472 .clone();
473 let columns: Vec<String> =
474 schema.columns.iter().map(|c| c.name.clone()).collect();
475 let fast = FastLayout::new(&schema);
476 let row_layout = RowLayout::new(&schema);
477
478 // Try compiled predicate (zero-allocation hot path).
479 // Handles int leaves, string-eq leaves, AND conjunctions.
480 if let Some(compiled) =
481 compile_predicate(predicate, &columns, &fast, &schema)
482 {
483 let mut count: i64 = 0;
484 self.catalog
485 .for_each_row_raw(table, |_rid, data| {
486 if compiled(data) {
487 count += 1;
488 }
489 })
490 .map_err(|e| QueryError::StorageError(e.to_string()))?;
491 return Ok(QueryResult::Scalar(Value::Int(count)));
492 }
493
494 // Fallback: decode predicate columns
495 let pred_cols = predicate_column_indices(predicate, &columns);
496 let mut count: i64 = 0;
497 self.catalog
498 .for_each_row_raw(table, |_rid, data| {
499 let pred_row =
500 decode_selective(&schema, &row_layout, data, &pred_cols);
501 if eval_predicate(predicate, &pred_row, &columns) {
502 count += 1;
503 }
504 })
505 .map_err(|e| QueryError::StorageError(e.to_string()))?;
506
507 return Ok(QueryResult::Scalar(Value::Int(count)));
508 }
509 }
510 }
511
512 // Fast path: sum/avg/min/max over a single fixed-size int
513 // column with an optional compiled filter predicate. Walks
514 // raw row bytes, zero allocation per row.
515 if matches!(
516 function,
517 AggFunc::Sum
518 | AggFunc::Avg
519 | AggFunc::Min
520 | AggFunc::Max
521 | AggFunc::CountDistinct
522 ) {
523 if let Some(col) = field.as_ref() {
524 // Shape: Aggregate(SeqScan) or Aggregate(Filter(SeqScan))
525 let (table_opt, pred_opt): (Option<&str>, Option<&Expr>) =
526 match input.as_ref() {
527 PlanNode::SeqScan { table } => (Some(table.as_str()), None),
528 PlanNode::Filter {
529 input: inner,
530 predicate,
531 } => {
532 if let PlanNode::SeqScan { table } = inner.as_ref() {
533 (Some(table.as_str()), Some(predicate))
534 } else {
535 (None, None)
536 }
537 }
538 _ => (None, None),
539 };
540 if let Some(table) = table_opt {
541 if let Some(result) =
542 self.agg_single_col_fast(table, col, *function, pred_opt)?
543 {
544 return Ok(result);
545 }
546 }
547 }
548 }
549
550 // Fast path: Project(Limit(Filter(SeqScan))) — stream, decode
551 // only projected columns, stop once we hit the limit.
552 // (Handled in the Project branch; this branch only fires when
553 // the aggregate is the outer node.)
554 let result = self.execute_plan(input)?;
555 match result {
556 QueryResult::Rows { columns, rows } => {
557 match function {
558 AggFunc::Count => {
559 Ok(QueryResult::Scalar(Value::Int(rows.len() as i64)))
560 }
561 AggFunc::CountDistinct => {
562 let col = field.as_ref().ok_or("count distinct requires field")?;
563 let idx = columns
564 .iter()
565 .position(|c| c == col)
566 .ok_or("col not found")?;
567 let mut seen = std::collections::HashSet::new();
568 for row in &rows {
569 let v = &row[idx];
570 if !v.is_empty() {
571 seen.insert(v.clone());
572 }
573 }
574 Ok(QueryResult::Scalar(Value::Int(seen.len() as i64)))
575 }
576 AggFunc::Avg => {
577 let col = field.as_ref().ok_or("avg requires field")?;
578 let idx = columns
579 .iter()
580 .position(|c| c == col)
581 .ok_or("col not found")?;
582 let mut count: u64 = 0;
583 let sum: f64 = rows
584 .iter()
585 .filter_map(|r| match &r[idx] {
586 Value::Int(v) => Some(*v as f64),
587 Value::Float(v) => Some(*v),
588 _ => None,
589 })
590 .inspect(|_| count += 1)
591 .sum();
592 if count == 0 {
593 Ok(QueryResult::Scalar(Value::Empty))
594 } else {
595 Ok(QueryResult::Scalar(Value::Float(sum / count as f64)))
596 }
597 }
598 AggFunc::Sum => {
599 let col = field.as_ref().ok_or("sum requires field")?;
600 let idx = columns
601 .iter()
602 .position(|c| c == col)
603 .ok_or("col not found")?;
604 // Track int and float contributions separately so
605 // Float columns (and mixed Int/Float rows) don't get
606 // silently dropped as they did in the Int-only
607 // version. If any Float is present, the whole sum
608 // promotes to Float — matching Avg's semantics.
609 let mut int_sum: i64 = 0;
610 let mut float_sum: f64 = 0.0;
611 let mut saw_float = false;
612 for r in &rows {
613 match &r[idx] {
614 Value::Int(v) => int_sum += *v,
615 Value::Float(v) => {
616 float_sum += *v;
617 saw_float = true;
618 }
619 _ => {}
620 }
621 }
622 let result = if saw_float {
623 Value::Float(float_sum + int_sum as f64)
624 } else {
625 Value::Int(int_sum)
626 };
627 Ok(QueryResult::Scalar(result))
628 }
629 AggFunc::Min | AggFunc::Max => {
630 let col = field.as_ref().ok_or("min/max requires field")?;
631 let idx = columns
632 .iter()
633 .position(|c| c == col)
634 .ok_or("col not found")?;
635 let vals: Vec<&Value> = rows.iter().map(|r| &r[idx]).collect();
636 let result = if *function == AggFunc::Min {
637 vals.into_iter().min().cloned()
638 } else {
639 vals.into_iter().max().cloned()
640 };
641 Ok(QueryResult::Scalar(result.unwrap_or(Value::Empty)))
642 }
643 }
644 }
645 _ => Err("aggregate requires row input".into()),
646 }
647 }
648
649 PlanNode::Insert {
650 table,
651 rows,
652 returning,
653 } => {
654 // Build + validate EVERY row before inserting any, so a bad
655 // row (unknown/missing/uncoercible field) aborts the whole
656 // statement without a partial write. The WAL fsync happens
657 // once at statement end, so N rows = N appends + 1 fsync.
658 let mut returning_columns: Vec<String> = Vec::new();
659 let all_values: Vec<Vec<Value>> = {
660 let schema = self
661 .catalog
662 .schema(table)
663 .ok_or_else(|| QueryError::TableNotFound(table.to_string()))?;
664 if *returning {
665 returning_columns = schema.columns.iter().map(|c| c.name.clone()).collect();
666 }
667 let defaults = self.catalog.column_defaults(table).unwrap_or(&[]);
668 let auto = self.catalog.auto_columns(table).unwrap_or(&[]);
669 let mut all = Vec::with_capacity(rows.len());
670 for assignments in rows {
671 let mut values = vec![Value::Empty; schema.columns.len()];
672 for a in assignments {
673 let idx = schema.column_index(&a.field).ok_or_else(|| {
674 QueryError::ColumnNotFound {
675 table: String::new(),
676 column: a.field.clone(),
677 }
678 })?;
679 let raw = literal_to_value(&a.value)?;
680 values[idx] = coerce_value(raw, &schema.columns[idx])?;
681 }
682 // Fill any column left unset by this row from its
683 // declared default (applied before the required check,
684 // so a default satisfies a required column).
685 for (i, slot) in values.iter_mut().enumerate() {
686 if slot.is_empty() {
687 if let Some(Some(d)) = defaults.get(i) {
688 *slot = d.clone();
689 }
690 }
691 }
692 for col in &schema.columns {
693 let pos = col.position as usize;
694 // Auto columns are exempt from the required check —
695 // they are filled from the sequence just below.
696 let is_auto = auto.get(pos).copied().unwrap_or(false);
697 if col.required && !is_auto && matches!(values[pos], Value::Empty) {
698 return Err(QueryError::Execution(format!(
699 "column '{}' is required but no value was provided",
700 col.name
701 )));
702 }
703 }
704 all.push(values);
705 }
706 all
707 };
708 // Assign auto-increment columns now that the immutable
709 // schema/defaults/auto borrows are released. Done here (not in
710 // the build loop) so the assigned ids land in `all_values` and
711 // flow back through `returning`.
712 let mut all_values = all_values;
713 for values in all_values.iter_mut() {
714 self.catalog.assign_auto_columns(table, values);
715 }
716 // Charge the materialized batch against the per-query memory
717 // budget before inserting — keeps multi-row insert consistent
718 // with every other full-materialization point (sort/join/group)
719 // and bounds embedded callers (the server also caps the query
720 // string at 1 MB, but embedded callers have no such limit).
721 self.charge_rows(&all_values)?;
722 let n = all_values.len() as u64;
723 for values in &all_values {
724 self.catalog
725 .insert(table, values)
726 .map_err(|e| QueryError::StorageError(e.to_string()))?;
727 }
728 self.view_registry.mark_dependents_dirty(table);
729 if *returning {
730 Ok(QueryResult::Rows {
731 columns: returning_columns,
732 rows: all_values,
733 })
734 } else {
735 Ok(QueryResult::Modified(n))
736 }
737 }
738
739 PlanNode::Upsert {
740 table,
741 key_column,
742 assignments,
743 on_conflict,
744 } => {
745 let (values, key_idx) = {
746 let schema = self
747 .catalog
748 .schema(table)
749 .ok_or_else(|| QueryError::TableNotFound(table.to_string()))?;
750 let mut values = vec![Value::Empty; schema.columns.len()];
751 for a in assignments {
752 let idx = schema.column_index(&a.field).ok_or_else(|| {
753 QueryError::ColumnNotFound {
754 table: String::new(),
755 column: a.field.clone(),
756 }
757 })?;
758 let raw = literal_to_value(&a.value)?;
759 values[idx] = coerce_value(raw, &schema.columns[idx])?;
760 }
761 // Apply column defaults for the insert path, same as a plain
762 // insert (applied before the required-column check).
763 let defaults = self.catalog.column_defaults(table).unwrap_or(&[]);
764 for (i, slot) in values.iter_mut().enumerate() {
765 if slot.is_empty() {
766 if let Some(Some(d)) = defaults.get(i) {
767 *slot = d.clone();
768 }
769 }
770 }
771 for col in &schema.columns {
772 if col.required && matches!(values[col.position as usize], Value::Empty) {
773 return Err(QueryError::Execution(format!(
774 "column '{}' is required but no value was provided",
775 col.name
776 )));
777 }
778 }
779 let key_idx = schema
780 .column_index(key_column)
781 .ok_or_else(|| format!("key column '{key_column}' not found"))?;
782 (values, key_idx)
783 };
784
785 // Upsert requires the `on` column to be unique — otherwise
786 // there is no well-defined row to overwrite and a plain
787 // insert could silently create duplicate keys.
788 if self.catalog.is_index_unique(table, key_column) != Some(true) {
789 return Err(QueryError::Execution(format!(
790 "upsert on .{key_column} requires a unique column (declare it with \
791 `unique {key_column}: <type>` or `alter {table} add unique .{key_column}`)"
792 )));
793 }
794
795 let key_value = values[key_idx].clone();
796
797 // Probe the unique index for a conflict.
798 let existing = {
799 let tbl = self
800 .catalog
801 .get_table(table)
802 .ok_or_else(|| QueryError::TableNotFound(table.to_string()))?;
803 // The key column is guaranteed unique above, so this
804 // returns at most one matching row.
805 let rids = tbl.index_lookup_all(key_column, &key_value);
806 rids.into_iter().next().and_then(|rid| {
807 tbl.heap
808 .get(rid)
809 .map(|data| (rid, decode_row(&tbl.schema, &data)))
810 })
811 };
812
813 if let Some((rid, mut existing_row)) = existing {
814 // Conflict: apply on_conflict assignments (or all non-key if empty).
815 let update_assignments = if on_conflict.is_empty() {
816 assignments
817 } else {
818 on_conflict
819 };
820 let changed_cols: Vec<usize> = {
821 let schema = self
822 .catalog
823 .schema(table)
824 .ok_or_else(|| QueryError::TableNotFound(table.to_string()))?;
825 let mut indices = Vec::new();
826 for a in update_assignments {
827 let idx = schema.column_index(&a.field).ok_or_else(|| {
828 QueryError::ColumnNotFound {
829 table: String::new(),
830 column: a.field.clone(),
831 }
832 })?;
833 if idx != key_idx {
834 // Coerce to the target column type, same as the
835 // UPDATE and INSERT paths — an int→float literal
836 // here would otherwise persist as raw i64 bits
837 // (#118 corruption on the upsert conflict path).
838 existing_row[idx] =
839 coerce_value(literal_to_value(&a.value)?, &schema.columns[idx])
840 .map_err(QueryError::TypeError)?;
841 indices.push(idx);
842 }
843 }
844 indices
845 };
846 self.catalog
847 .update_hinted(table, rid, &existing_row, Some(&changed_cols))
848 .map_err(|e| QueryError::StorageError(e.to_string()))?;
849 self.view_registry.mark_dependents_dirty(table);
850 Ok(QueryResult::Modified(1))
851 } else {
852 // No conflict: insert.
853 self.catalog
854 .insert(table, &values)
855 .map_err(|e| QueryError::StorageError(e.to_string()))?;
856 self.view_registry.mark_dependents_dirty(table);
857 Ok(QueryResult::Modified(1))
858 }
859 }
860
861 PlanNode::Update {
862 input,
863 table,
864 assignments,
865 returning,
866 } => {
867 // Mission C Phase 3: resolve assignments against a borrowed
868 // schema, then drop the borrow before the mutation loop.
869 // Try literal-only path first; fall back to per-row expression
870 // evaluation if any assignment contains a non-literal expression
871 // (e.g., `age := .age + 1`).
872 let (col_indices, literal_vals, target_cols): (
873 Vec<usize>,
874 Option<Vec<Value>>,
875 Vec<ColumnDef>,
876 ) = {
877 let schema_ref = self
878 .catalog
879 .schema(table)
880 .ok_or_else(|| QueryError::TableNotFound(table.to_string()))?;
881 let indices: Vec<usize> = assignments
882 .iter()
883 .map(|a| {
884 schema_ref.column_index(&a.field).ok_or_else(|| {
885 QueryError::ColumnNotFound {
886 table: String::new(),
887 column: a.field.clone(),
888 }
889 })
890 })
891 .collect::<Result<_, _>>()?;
892 // The target column defs (aligned with `assignments`), owned
893 // so the per-row expression path can coerce without holding a
894 // catalog borrow across the mutation loop.
895 let target_cols: Vec<ColumnDef> = indices
896 .iter()
897 .map(|&idx| schema_ref.columns[idx].clone())
898 .collect();
899 // Resolve each assignment to a literal value. If any is a
900 // non-literal expression, fall back (None) to the per-row
901 // expression-eval path below.
902 let raw_vals: Result<Vec<Value>, _> = assignments
903 .iter()
904 .map(|a| literal_to_value(&a.value))
905 .collect();
906 // Coerce each literal to its target column's declared type
907 // before it can reach the byte-patch fast path (the same
908 // coercion the INSERT path applies). Without this, an int
909 // assigned to a float column is written as raw i64 bits
910 // (#118 silent corruption) and a str assigned to a
911 // fixed-size column reaches `unreachable!` and aborts the
912 // whole server (#117 remote DoS). A genuine type mismatch
913 // is a hard error to the client, not an expr-path fallback.
914 let coerced = match raw_vals {
915 Ok(raws) => {
916 let mut out = Vec::with_capacity(raws.len());
917 for (raw, &idx) in raws.into_iter().zip(indices.iter()) {
918 out.push(
919 coerce_value(raw, &schema_ref.columns[idx])
920 .map_err(QueryError::TypeError)?,
921 );
922 }
923 Some(out)
924 }
925 Err(_) => None,
926 };
927 (indices, coerced, target_cols)
928 };
929 let resolved_assignments: Option<Vec<(usize, Value)>> =
930 literal_vals.map(|vals| col_indices.iter().copied().zip(vals).collect());
931
932 // Mission C Phase 2: the hint Table::update_hinted needs to
933 // decide whether to read the old row for index diff.
934 let changed_cols: Vec<usize> = col_indices.clone();
935
936 // ── RETURNING path ──────────────────────────────────────
937 // `returning` materializes the post-update row image, so the
938 // byte-patch / fused fast paths (which never decode a row)
939 // can't serve it. Take the generic decode→mutate→collect
940 // route. Opt-in only: when `returning` is false every path
941 // below is byte-for-byte unchanged.
942 if *returning {
943 let columns: Vec<String> = {
944 let schema_ref = self
945 .catalog
946 .schema(table)
947 .ok_or_else(|| QueryError::TableNotFound(table.to_string()))?;
948 schema_ref.columns.iter().map(|c| c.name.clone()).collect()
949 };
950 let matching_rids = self.collect_rids_for_mutation(input, table)?;
951 let mut out_rows: Vec<Vec<Value>> = Vec::with_capacity(matching_rids.len());
952 for rid in matching_rids {
953 let mut row = match self.catalog.get(table, rid) {
954 Some(r) => r,
955 None => continue,
956 };
957 match &resolved_assignments {
958 // Literal path: apply the pre-coerced values.
959 Some(resolved) => {
960 for (idx, val) in resolved.iter() {
961 row[*idx] = val.clone();
962 }
963 }
964 // Expression path: evaluate each RHS against the
965 // (progressively mutated) row, then coerce to the
966 // target column type before writing — same guard the
967 // literal path gets, matching the non-returning expr
968 // path exactly (#117/#118 on computed assignments).
969 None => {
970 for (i, asgn) in assignments.iter().enumerate() {
971 let val = eval_expr(&asgn.value, &row, &columns);
972 row[col_indices[i]] = coerce_value(val, &target_cols[i])
973 .map_err(QueryError::TypeError)?;
974 }
975 }
976 }
977 self.catalog
978 .update_hinted(table, rid, &row, Some(&changed_cols))
979 .map_err(|e| QueryError::StorageError(e.to_string()))?;
980 out_rows.push(row);
981 }
982 self.view_registry.mark_dependents_dirty(table);
983 return Ok(QueryResult::Rows {
984 columns,
985 rows: out_rows,
986 });
987 }
988
989 // ── Fused scan+update for Update(Filter(SeqScan)) ────────
990 // Perf sprint: instead of the two-pass collect-RIDs-then-loop
991 // pattern (which pays one ensure_hot per matched row on the
992 // second pass), fuse the predicate evaluation and in-place
993 // byte-level mutation into a single heap walk. Same idea as
994 // the fused scan_delete_matching path for deletes.
995 if let Some(ref resolved_assignments) = resolved_assignments {
996 if let PlanNode::Filter {
997 input: inner,
998 predicate,
999 } = input.as_ref()
1000 {
1001 if let PlanNode::SeqScan { table: t } = inner.as_ref() {
1002 if t == table {
1003 let fused_result = self.try_fused_scan_update(
1004 table,
1005 predicate,
1006 resolved_assignments,
1007 &changed_cols,
1008 );
1009 if let Some(result) = fused_result {
1010 return result;
1011 }
1012 }
1013 }
1014 }
1015 }
1016
1017 // Collect matching RowIds in a single pass.
1018 let matching_rids = self.collect_rids_for_mutation(input, table)?;
1019
1020 // ── Literal-only fast paths ─────────────────────────────
1021 if let Some(ref resolved_assignments) = resolved_assignments {
1022 // Mission C Phase 4: in-place byte-patch fast path. If every
1023 // assignment targets a fixed-size non-null column AND none of
1024 // them is indexed, we can skip decode_row / Vec<Value> /
1025 // encode_row_into entirely and patch the row's raw bytes on
1026 // the hot page.
1027 let fast_patch: Option<Vec<FastPatch>> = {
1028 let tbl = self
1029 .catalog
1030 .get_table(table)
1031 .ok_or_else(|| QueryError::TableNotFound(table.to_string()))?;
1032 let schema = &tbl.schema;
1033 let all_fixed_nonnull = resolved_assignments.iter().all(|(idx, val)| {
1034 is_fixed_size(schema.columns[*idx].type_id) && !val.is_empty()
1035 });
1036 let no_indexed = !resolved_assignments
1037 .iter()
1038 .any(|(idx, _)| tbl.has_indexed_col(*idx));
1039
1040 if all_fixed_nonnull && no_indexed {
1041 let layout = RowLayout::new(schema);
1042 let bitmap_size = layout.bitmap_size();
1043 let patches: Vec<FastPatch> = resolved_assignments
1044 .iter()
1045 .map(|(idx, val)| {
1046 let fixed_off = layout
1047 .fixed_offset(*idx)
1048 .expect("is_fixed_size already checked");
1049 let field_off = 2 + bitmap_size + fixed_off;
1050 let bytes: FixedBytes = match val {
1051 Value::Int(v) => FixedBytes::I64(v.to_le_bytes()),
1052 Value::Float(v) => FixedBytes::F64(v.to_le_bytes()),
1053 Value::Bool(v) => FixedBytes::Bool(if *v { 1 } else { 0 }),
1054 Value::DateTime(v) => FixedBytes::I64(v.to_le_bytes()),
1055 Value::Uuid(v) => FixedBytes::Uuid(*v),
1056 _ => unreachable!("all_fixed_nonnull guard lied"),
1057 };
1058 FastPatch {
1059 field_off,
1060 bitmap_byte_off: 2 + idx / 8,
1061 bit_mask: 1u8 << (idx % 8),
1062 bytes,
1063 }
1064 })
1065 .collect();
1066 Some(patches)
1067 } else {
1068 None
1069 }
1070 };
1071
1072 if let Some(patches) = fast_patch {
1073 let mut count = 0u64;
1074 for rid in matching_rids {
1075 // Mission B2: WAL-log every patch so crash
1076 // recovery replays the update. Same mutation
1077 // closure as before — the wrapper just sandwiches
1078 // it between a hot-page read and a WAL append.
1079 let ok = self
1080 .catalog
1081 .update_row_bytes_logged(table, rid, |row| {
1082 let base = row_body_base(row);
1083 for p in &patches {
1084 row[base + p.bitmap_byte_off] &= !p.bit_mask;
1085 let field_bytes = p.bytes.as_slice();
1086 row[base + p.field_off
1087 ..base + p.field_off + field_bytes.len()]
1088 .copy_from_slice(field_bytes);
1089 }
1090 })
1091 .map_err(|e| QueryError::StorageError(e.to_string()))?;
1092 if ok {
1093 count += 1;
1094 }
1095 }
1096 self.view_registry.mark_dependents_dirty(table);
1097 return Ok(QueryResult::Modified(count));
1098 }
1099
1100 // Mission C Phase 10: var-column in-place shrink fast path.
1101 let var_fast: Option<(usize, Option<Vec<u8>>)> = {
1102 let tbl = self
1103 .catalog
1104 .get_table(table)
1105 .ok_or_else(|| QueryError::TableNotFound(table.to_string()))?;
1106 let schema = &tbl.schema;
1107 let is_single = resolved_assignments.len() == 1;
1108 let is_var_col = is_single
1109 && !is_fixed_size(schema.columns[resolved_assignments[0].0].type_id);
1110 let no_indexed = !resolved_assignments
1111 .iter()
1112 .any(|(idx, _)| tbl.has_indexed_col(*idx));
1113
1114 if is_single && is_var_col && no_indexed {
1115 let (idx, val) = &resolved_assignments[0];
1116 let bytes_opt: Option<Vec<u8>> = match val {
1117 Value::Str(s) => Some(s.as_bytes().to_vec()),
1118 Value::Bytes(b) => Some(b.clone()),
1119 Value::Empty => None,
1120 _ => {
1121 return Err(QueryError::TypeError(format!(
1122 "cannot assign non-var value to var column '{}'",
1123 schema.columns[*idx].name
1124 )))
1125 }
1126 };
1127 Some((*idx, bytes_opt))
1128 } else {
1129 None
1130 }
1131 };
1132
1133 if let Some((col_idx, new_bytes_opt)) = var_fast {
1134 let new_bytes_ref: Option<&[u8]> = new_bytes_opt.as_deref();
1135 let mut count = 0u64;
1136 let mut fallback_rids: Vec<RowId> = Vec::new();
1137 for rid in &matching_rids {
1138 // Mission B2: logged variant so crash recovery
1139 // replays the shrink. On a false return (row
1140 // would have to grow), the rid is pushed to
1141 // `fallback_rids` and the slower `update_hinted`
1142 // path — which is already WAL-logged — picks it up.
1143 let ok = self
1144 .catalog
1145 .patch_var_col_logged(table, *rid, col_idx, new_bytes_ref)
1146 .map_err(|e| QueryError::StorageError(e.to_string()))?;
1147 if ok {
1148 count += 1;
1149 } else {
1150 fallback_rids.push(*rid);
1151 }
1152 }
1153 for rid in fallback_rids {
1154 let mut row = match self.catalog.get(table, rid) {
1155 Some(r) => r,
1156 None => continue,
1157 };
1158 for (idx, val) in resolved_assignments.iter() {
1159 row[*idx] = val.clone();
1160 }
1161 self.catalog
1162 .update_hinted(table, rid, &row, Some(&changed_cols))
1163 .map_err(|e| QueryError::StorageError(e.to_string()))?;
1164 count += 1;
1165 }
1166 self.view_registry.mark_dependents_dirty(table);
1167 return Ok(QueryResult::Modified(count));
1168 }
1169
1170 // Generic literal path: decode row, apply literal values.
1171 let mut count = 0u64;
1172 for rid in matching_rids {
1173 let mut row = match self.catalog.get(table, rid) {
1174 Some(r) => r,
1175 None => continue,
1176 };
1177 for (idx, val) in resolved_assignments.iter() {
1178 row[*idx] = val.clone();
1179 }
1180 self.catalog
1181 .update_hinted(table, rid, &row, Some(&changed_cols))
1182 .map_err(|e| QueryError::StorageError(e.to_string()))?;
1183 count += 1;
1184 }
1185 self.view_registry.mark_dependents_dirty(table);
1186 return Ok(QueryResult::Modified(count));
1187 } // end if let Some(resolved_assignments)
1188
1189 // ── Expression-based update path ────────────────────────
1190 // At least one assignment contains a non-literal expression
1191 // (e.g., `age := .age + 1`). Evaluate per-row.
1192 let col_names: Vec<String> = {
1193 let schema_ref = self
1194 .catalog
1195 .schema(table)
1196 .ok_or_else(|| QueryError::TableNotFound(table.to_string()))?;
1197 schema_ref.columns.iter().map(|c| c.name.clone()).collect()
1198 };
1199 let mut count = 0u64;
1200 for rid in matching_rids {
1201 let mut row = match self.catalog.get(table, rid) {
1202 Some(r) => r,
1203 None => continue,
1204 };
1205 for (i, asgn) in assignments.iter().enumerate() {
1206 let val = eval_expr(&asgn.value, &row, &col_names);
1207 // Coerce to the target column type before writing, so a
1208 // computed int→float assignment stores f64 (not raw i64
1209 // bits, #118) and a str→fixed-col assignment returns a
1210 // typed error instead of hitting the encoder's
1211 // `unreachable!` and aborting the process (#117).
1212 row[col_indices[i]] =
1213 coerce_value(val, &target_cols[i]).map_err(QueryError::TypeError)?;
1214 }
1215 self.catalog
1216 .update_hinted(table, rid, &row, Some(&changed_cols))
1217 .map_err(|e| QueryError::StorageError(e.to_string()))?;
1218 count += 1;
1219 }
1220 self.view_registry.mark_dependents_dirty(table);
1221 Ok(QueryResult::Modified(count))
1222 }
1223
1224 PlanNode::Delete {
1225 input,
1226 table,
1227 returning,
1228 } => {
1229 // ── RETURNING path ──────────────────────────────────────
1230 // `returning` needs the pre-delete row image, so read each
1231 // matched row before removing it. The fused single-pass
1232 // delete primitives below never decode rows, so they can't
1233 // serve this. Opt-in only: when `returning` is false the
1234 // fast paths below are byte-for-byte unchanged.
1235 if *returning {
1236 let columns: Vec<String> = {
1237 let schema_ref = self
1238 .catalog
1239 .schema(table)
1240 .ok_or_else(|| QueryError::TableNotFound(table.to_string()))?;
1241 schema_ref.columns.iter().map(|c| c.name.clone()).collect()
1242 };
1243 let matching_rids = self.collect_rids_for_mutation(input, table)?;
1244 let mut out_rows: Vec<Vec<Value>> = Vec::with_capacity(matching_rids.len());
1245 for rid in &matching_rids {
1246 if let Some(row) = self.catalog.get(table, *rid) {
1247 out_rows.push(row);
1248 }
1249 }
1250 self.catalog
1251 .delete_many(table, &matching_rids)
1252 .map_err(|e| QueryError::StorageError(e.to_string()))?;
1253 self.view_registry.mark_dependents_dirty(table);
1254 return Ok(QueryResult::Rows {
1255 columns,
1256 rows: out_rows,
1257 });
1258 }
1259
1260 // Mission C Phase 3: no schema clone — collect_rids_for_mutation
1261 // looks up schema internally when it needs one, and the mutation
1262 // loop doesn't need the schema at all.
1263 //
1264 // Mission C Phase 12: route bulk deletes through
1265 // `Catalog::delete_many`, which batches the btree leaf
1266 // compaction and shares one `ensure_hot` per row between
1267 // the index-key extraction and the slot delete. On
1268 // `delete_by_filter` (100K fixture, ~20K matches) that
1269 // removes ~4ms of pure `Vec::remove` memmove from the btree
1270 // maintenance phase.
1271 //
1272 // Mission C Phase 16: for the common `delete where ...`
1273 // shape (Filter(SeqScan)) — and the rarer "delete
1274 // everything" shape (SeqScan) — skip the two-pass
1275 // `collect_rids_for_mutation` + `delete_many` flow entirely.
1276 // The fused `scan_delete_matching` primitive walks the
1277 // heap exactly once, paying one `ensure_hot` per page
1278 // instead of per-row. That closes the last major gap on
1279 // the bench's `delete_by_filter` workload.
1280 if let PlanNode::Filter {
1281 input: inner,
1282 predicate,
1283 } = input.as_ref()
1284 {
1285 if let PlanNode::SeqScan { table: t } = inner.as_ref() {
1286 if t == table {
1287 let schema = self
1288 .catalog
1289 .schema(table)
1290 .ok_or_else(|| QueryError::TableNotFound(table.to_string()))?;
1291 let columns: Vec<String> =
1292 schema.columns.iter().map(|c| c.name.clone()).collect();
1293 let fast = FastLayout::new(schema);
1294 if let Some(compiled) =
1295 compile_predicate(predicate, &columns, &fast, schema)
1296 {
1297 // Mission B2: logged variant so every
1298 // matched rid hits the WAL during the
1299 // single-pass scan. Structure of the
1300 // fused scan is unchanged — only the
1301 // hook closure now also appends.
1302 let count = self
1303 .catalog
1304 .scan_delete_matching_logged(table, |data| compiled(data))
1305 .map_err(|e| QueryError::StorageError(e.to_string()))?;
1306 self.view_registry.mark_dependents_dirty(table);
1307 return Ok(QueryResult::Modified(count));
1308 }
1309 }
1310 }
1311 } else if let PlanNode::SeqScan { table: t } = input.as_ref() {
1312 if t == table {
1313 // `delete from T` with no predicate — every live
1314 // row matches. One pass is still the right shape.
1315 // Mission B2: logged variant — see above.
1316 let count = self
1317 .catalog
1318 .scan_delete_matching_logged(table, |_| true)
1319 .map_err(|e| QueryError::StorageError(e.to_string()))?;
1320 self.view_registry.mark_dependents_dirty(table);
1321 return Ok(QueryResult::Modified(count));
1322 }
1323 }
1324
1325 let matching_rids = self.collect_rids_for_mutation(input, table)?;
1326 let count = self
1327 .catalog
1328 .delete_many(table, &matching_rids)
1329 .map_err(|e| QueryError::StorageError(e.to_string()))?;
1330 self.view_registry.mark_dependents_dirty(table);
1331 Ok(QueryResult::Modified(count))
1332 }
1333
1334 PlanNode::AliasScan { table, alias } => {
1335 // Mission E1.2: scan `table` and rename every output column
1336 // to `alias.field`. Used as a join leaf so downstream
1337 // NestedLoopJoin + Filter + Project nodes can resolve
1338 // `Expr::QualifiedField` lookups by direct column-name match.
1339 //
1340 // We don't bother with a fused zero-copy loop here yet — the
1341 // whole join path is nested-loop and correctness-first
1342 // (Phase E1.3 will introduce hash join and at that point we
1343 // can revisit whether to specialise AliasScan).
1344 let schema = self
1345 .catalog
1346 .schema(table)
1347 .ok_or_else(|| QueryError::TableNotFound(table.to_string()))?
1348 .clone();
1349 let columns: Vec<String> = schema
1350 .columns
1351 .iter()
1352 .map(|c| format!("{alias}.{}", c.name))
1353 .collect();
1354 let rows: Vec<Vec<Value>> = self
1355 .catalog
1356 .scan(table)
1357 .map_err(|e| QueryError::StorageError(e.to_string()))?
1358 .map(|(_, row)| row)
1359 .collect();
1360 Ok(QueryResult::Rows { columns, rows })
1361 }
1362
1363 PlanNode::NestedLoopJoin {
1364 left,
1365 right,
1366 on,
1367 kind,
1368 } => {
1369 // Materialise both sides. The executor ships two strategies:
1370 // 1. Hash join (E1.3) — when the `on` predicate is a
1371 // simple equi-predicate `left_col = right_col`, build a
1372 // FxHashMap<Value, Vec<row_idx>> over the right side
1373 // and probe with the left side. O(L + R) instead of
1374 // O(L × R). Handles Inner and LeftOuter.
1375 // 2. Nested loop (E1.2) — fallback for Cross, non-equi
1376 // predicates, or `on` expressions that reference
1377 // either side with something more complex than a
1378 // QualifiedField.
1379 let left_result = self.execute_plan(left)?;
1380 let right_result = self.execute_plan(right)?;
1381 let (left_columns, left_rows) = match left_result {
1382 QueryResult::Rows { columns, rows } => (columns, rows),
1383 _ => return Err("join left side must produce rows".into()),
1384 };
1385 let (right_columns, right_rows) = match right_result {
1386 QueryResult::Rows { columns, rows } => (columns, rows),
1387 _ => return Err("join right side must produce rows".into()),
1388 };
1389
1390 // WS2: byte-budget guard on the join build side. Charge both
1391 // materialized inputs before we build the hash table / probe;
1392 // the output is row-capped by check_join_limit below.
1393 self.charge_rows(&left_rows)?;
1394 self.charge_rows(&right_rows)?;
1395
1396 // Hash-join fast path.
1397 if !matches!(kind, JoinKind::Cross) {
1398 if let Some(pred) = on {
1399 if let Some((l_idx, r_idx)) =
1400 try_extract_equi_join_keys(pred, &left_columns, &right_columns)
1401 {
1402 let result = hash_join(
1403 left_columns,
1404 left_rows,
1405 right_columns,
1406 right_rows,
1407 l_idx,
1408 r_idx,
1409 *kind,
1410 );
1411 if let QueryResult::Rows { ref rows, .. } = result {
1412 check_join_limit(rows.len())?;
1413 }
1414 return Ok(result);
1415 }
1416 }
1417 }
1418
1419 // Nested-loop fallback.
1420 let n_left = left_columns.len();
1421 let n_right = right_columns.len();
1422 let mut columns = Vec::with_capacity(n_left + n_right);
1423 columns.extend(left_columns);
1424 columns.extend(right_columns);
1425
1426 let mut rows: Vec<Vec<Value>> = Vec::with_capacity(left_rows.len());
1427 let mut combined: Vec<Value> = Vec::with_capacity(n_left + n_right);
1428
1429 for left_row in &left_rows {
1430 let mut matched = false;
1431 for right_row in &right_rows {
1432 combined.clear();
1433 combined.extend_from_slice(left_row);
1434 combined.extend_from_slice(right_row);
1435 let keep = match kind {
1436 JoinKind::Cross => true,
1437 JoinKind::Inner | JoinKind::LeftOuter => match on {
1438 Some(pred) => eval_predicate(pred, &combined, &columns),
1439 // Missing `on` for non-cross joins is a
1440 // parser error, but if it slips through we
1441 // treat it as "match everything".
1442 None => true,
1443 },
1444 // RightOuter is rewritten to LeftOuter by the
1445 // planner, so we never see it here.
1446 JoinKind::RightOuter => {
1447 unreachable!("planner rewrites RightOuter to LeftOuter")
1448 }
1449 };
1450 if keep {
1451 rows.push(combined.clone());
1452 check_join_limit(rows.len())?;
1453 matched = true;
1454 }
1455 }
1456 if !matched && matches!(kind, JoinKind::LeftOuter) {
1457 let mut row = Vec::with_capacity(n_left + n_right);
1458 row.extend_from_slice(left_row);
1459 row.resize(n_left + n_right, Value::Empty);
1460 rows.push(row);
1461 check_join_limit(rows.len())?;
1462 }
1463 }
1464
1465 Ok(QueryResult::Rows { columns, rows })
1466 }
1467
1468 PlanNode::Distinct { input } => {
1469 let result = self.execute_plan(input)?;
1470 match result {
1471 QueryResult::Rows { columns, rows } => {
1472 let mut seen = std::collections::HashSet::new();
1473 let mut unique_rows = Vec::new();
1474 for row in rows {
1475 if seen.insert(row.clone()) {
1476 unique_rows.push(row);
1477 }
1478 }
1479 Ok(QueryResult::Rows {
1480 columns,
1481 rows: unique_rows,
1482 })
1483 }
1484 other => Ok(other),
1485 }
1486 }
1487
1488 PlanNode::GroupBy {
1489 input,
1490 keys,
1491 aggregates,
1492 having,
1493 } => {
1494 let result = self.execute_plan(input)?;
1495 match result {
1496 QueryResult::Rows { columns, rows } => {
1497 // WS2: byte-budget guard on the GROUP BY input buffer
1498 // (the hash table is bounded by the input it groups).
1499 self.charge_rows(&rows)?;
1500 // Resolve key column indices.
1501 let key_indices: Vec<usize> = keys
1502 .iter()
1503 .map(|k| {
1504 columns
1505 .iter()
1506 .position(|c| c == k)
1507 .ok_or_else(|| format!("group-by column '{k}' not found"))
1508 })
1509 .collect::<Result<Vec<_>, _>>()?;
1510
1511 // Resolve aggregate field indices. count(*) uses
1512 // sentinel usize::MAX — compute_group_aggregate
1513 // treats it as "count all rows in the group".
1514 let agg_field_indices: Vec<usize> = aggregates
1515 .iter()
1516 .map(|a| {
1517 if a.field == "*" {
1518 Ok(usize::MAX)
1519 } else {
1520 columns.iter().position(|c| c == &a.field).ok_or_else(|| {
1521 format!("aggregate column '{}' not found", a.field)
1522 })
1523 }
1524 })
1525 .collect::<Result<Vec<_>, _>>()?;
1526
1527 // Group rows by key values (preserving insertion order).
1528 let mut group_map: rustc_hash::FxHashMap<Vec<Value>, usize> =
1529 rustc_hash::FxHashMap::default();
1530 let mut groups: Vec<(Vec<Value>, Vec<usize>)> = Vec::new();
1531 for (ri, row) in rows.iter().enumerate() {
1532 let key: Vec<Value> =
1533 key_indices.iter().map(|&i| row[i].clone()).collect();
1534 match group_map.get(&key) {
1535 Some(&idx) => groups[idx].1.push(ri),
1536 None => {
1537 let idx = groups.len();
1538 group_map.insert(key.clone(), idx);
1539 groups.push((key, vec![ri]));
1540 }
1541 }
1542 }
1543
1544 // Build output column names: keys ++ aggregate output names.
1545 let mut out_columns: Vec<String> = keys.clone();
1546 for agg in aggregates.iter() {
1547 out_columns.push(agg.output_name.clone());
1548 }
1549
1550 // Compute aggregates per group.
1551 let mut out_rows: Vec<Vec<Value>> = Vec::with_capacity(groups.len());
1552 for (key_vals, row_indices) in &groups {
1553 let mut row = key_vals.clone();
1554 for (ai, agg) in aggregates.iter().enumerate() {
1555 let col_idx = agg_field_indices[ai];
1556 let val = compute_group_aggregate(
1557 agg.function,
1558 &rows,
1559 row_indices,
1560 col_idx,
1561 );
1562 row.push(val);
1563 }
1564 out_rows.push(row);
1565 }
1566
1567 // Apply HAVING filter.
1568 if let Some(having_expr) = having {
1569 out_rows.retain(|row| eval_predicate(having_expr, row, &out_columns));
1570 }
1571
1572 Ok(QueryResult::Rows {
1573 columns: out_columns,
1574 rows: out_rows,
1575 })
1576 }
1577 _ => Err("group by requires row input".into()),
1578 }
1579 }
1580
1581 PlanNode::CreateTable { name, fields } => {
1582 let columns: Vec<ColumnDef> = fields
1583 .iter()
1584 .enumerate()
1585 .map(|(i, f)| -> Result<ColumnDef, QueryError> {
1586 Ok(ColumnDef {
1587 name: f.name.clone(),
1588 type_id: type_name_to_id(&f.type_name)
1589 .map_err(QueryError::TypeError)?,
1590 required: f.required,
1591 position: i as u16,
1592 })
1593 })
1594 .collect::<Result<Vec<_>, _>>()?;
1595 // Coerce each literal default to its column's type now, so a
1596 // type mismatch (`count: int default "x"`) is rejected at DDL
1597 // time and the stored default is ready to drop into inserts.
1598 let mut defaults: Vec<Option<Value>> = vec![None; columns.len()];
1599 let mut auto_cols: Vec<bool> = vec![false; columns.len()];
1600 for (i, f) in fields.iter().enumerate() {
1601 if let Some(lit) = &f.default {
1602 let raw = literal_value_from(lit);
1603 defaults[i] = Some(coerce_value(raw, &columns[i])?);
1604 }
1605 if f.auto {
1606 // Auto-increment only makes sense on an integer column,
1607 // and combining it with a literal default is
1608 // contradictory (both want to supply the value).
1609 if columns[i].type_id != TypeId::Int {
1610 return Err(QueryError::TypeError(format!(
1611 "auto column '{}' must be of type int",
1612 f.name
1613 )));
1614 }
1615 if f.default.is_some() {
1616 return Err(QueryError::TypeError(format!(
1617 "auto column '{}' cannot also declare a default",
1618 f.name
1619 )));
1620 }
1621 auto_cols[i] = true;
1622 }
1623 }
1624 let schema = Schema {
1625 table_name: name.clone(),
1626 columns,
1627 };
1628 self.catalog
1629 .create_table_full(schema, defaults, auto_cols)
1630 .map_err(|e| QueryError::StorageError(e.to_string()))?;
1631 // Declaring a field `unique` auto-creates a unique B+tree
1632 // index, which is where uniqueness is enforced on writes.
1633 for f in fields.iter().filter(|f| f.unique) {
1634 self.catalog
1635 .create_index_unique(name, &f.name, true)
1636 .map_err(|e| QueryError::StorageError(e.to_string()))?;
1637 }
1638 Ok(QueryResult::Created(name.clone()))
1639 }
1640
1641 PlanNode::AlterTable { table, action } => match action {
1642 AlterAction::AddColumn {
1643 name,
1644 type_name,
1645 required,
1646 } => {
1647 let position = self
1648 .catalog
1649 .schema(table)
1650 .ok_or_else(|| QueryError::TableNotFound(table.to_string()))?
1651 .columns
1652 .len() as u16;
1653 let col = ColumnDef {
1654 name: name.clone(),
1655 type_id: type_name_to_id(type_name).map_err(QueryError::TypeError)?,
1656 required: *required,
1657 position,
1658 };
1659 self.catalog
1660 .alter_table_add_column(table, col)
1661 .map_err(|e| QueryError::StorageError(e.to_string()))?;
1662 Ok(QueryResult::Executed {
1663 message: format!("column '{name}' added to '{table}'"),
1664 })
1665 }
1666 AlterAction::DropColumn { name } => {
1667 self.catalog
1668 .alter_table_drop_column(table, name)
1669 .map_err(|e| QueryError::StorageError(e.to_string()))?;
1670 Ok(QueryResult::Executed {
1671 message: format!("column '{name}' dropped from '{table}'"),
1672 })
1673 }
1674 AlterAction::AddIndex { column } => {
1675 self.catalog
1676 .create_index(table, column)
1677 .map_err(|e| QueryError::StorageError(e.to_string()))?;
1678 Ok(QueryResult::Executed {
1679 message: format!("index on '{table}.{column}' created"),
1680 })
1681 }
1682 AlterAction::AddUnique { column } => {
1683 // No DropIndex exists, so we cannot upgrade an existing
1684 // non-unique index in place — reject it cleanly.
1685 if self.catalog.has_index(table, column) {
1686 return Err(QueryError::Execution(format!(
1687 "cannot add unique on {table}.{column}: column already indexed"
1688 )));
1689 }
1690 // Scan existing rows for duplicate (non-null) values
1691 // before creating the unique index.
1692 {
1693 let tbl = self
1694 .catalog
1695 .get_table(table)
1696 .ok_or_else(|| QueryError::TableNotFound(table.to_string()))?;
1697 let col_idx = tbl.schema.column_index(column).ok_or_else(|| {
1698 QueryError::ColumnNotFound {
1699 table: table.to_string(),
1700 column: column.clone(),
1701 }
1702 })?;
1703 let mut seen = std::collections::HashSet::new();
1704 for (_, row) in tbl.scan() {
1705 let v = &row[col_idx];
1706 if v.is_empty() {
1707 continue;
1708 }
1709 if !seen.insert(v.clone()) {
1710 return Err(QueryError::Execution(format!(
1711 "cannot add unique on {table}.{column}: \
1712 duplicate value {v:?} exists"
1713 )));
1714 }
1715 }
1716 }
1717 self.catalog
1718 .create_index_unique(table, column, true)
1719 .map_err(|e| QueryError::StorageError(e.to_string()))?;
1720 Ok(QueryResult::Executed {
1721 message: format!("unique index on '{table}.{column}' created"),
1722 })
1723 }
1724 },
1725
1726 PlanNode::DropTable { name } => {
1727 self.catalog
1728 .drop_table(name)
1729 .map_err(|e| QueryError::StorageError(e.to_string()))?;
1730 Ok(QueryResult::Executed {
1731 message: format!("table '{name}' dropped"),
1732 })
1733 }
1734
1735 PlanNode::CreateView { name, query_text } => {
1736 self.create_view(name, query_text)?;
1737 Ok(QueryResult::Executed {
1738 message: format!("materialized view '{name}' created"),
1739 })
1740 }
1741
1742 PlanNode::RefreshView { name } => {
1743 self.refresh_view(name)?;
1744 Ok(QueryResult::Executed {
1745 message: format!("materialized view '{name}' refreshed"),
1746 })
1747 }
1748
1749 PlanNode::DropView { name } => {
1750 self.drop_view(name)?;
1751 Ok(QueryResult::Executed {
1752 message: format!("materialized view '{name}' dropped"),
1753 })
1754 }
1755
1756 PlanNode::Window { input, windows } => {
1757 let result = self.execute_plan(input)?;
1758 execute_window(result, windows)
1759 }
1760
1761 PlanNode::Union { left, right, all } => {
1762 let left_result = self.execute_plan(left)?;
1763 let right_result = self.execute_plan(right)?;
1764 let (left_cols, left_rows) = match left_result {
1765 QueryResult::Rows { columns, rows } => (columns, rows),
1766 _ => return Err("UNION requires query results on left side".into()),
1767 };
1768 let (_, right_rows) = match right_result {
1769 QueryResult::Rows { columns, rows } => (columns, rows),
1770 _ => return Err("UNION requires query results on right side".into()),
1771 };
1772 let mut combined = left_rows;
1773 if *all {
1774 // UNION ALL — just concatenate.
1775 combined.extend(right_rows);
1776 } else {
1777 // UNION — deduplicate using the same HashSet approach
1778 // as DISTINCT. Value already implements Hash + Eq.
1779 let mut seen = std::collections::HashSet::new();
1780 for row in &combined {
1781 seen.insert(row.clone());
1782 }
1783 for row in right_rows {
1784 if seen.insert(row.clone()) {
1785 combined.push(row);
1786 }
1787 }
1788 }
1789 Ok(QueryResult::Rows {
1790 columns: left_cols,
1791 rows: combined,
1792 })
1793 }
1794
1795 PlanNode::Explain { input } => {
1796 let text = format_plan_tree(input, 0);
1797 Ok(QueryResult::Rows {
1798 columns: vec!["plan".to_string()],
1799 rows: text
1800 .lines()
1801 .map(|line| vec![Value::Str(line.to_string())])
1802 .collect(),
1803 })
1804 }
1805
1806 PlanNode::Begin => {
1807 if self.in_transaction {
1808 return Err(QueryError::Execution(
1809 "already in a transaction (nested transactions not supported)".into(),
1810 ));
1811 }
1812 self.catalog
1813 .begin_transaction()
1814 .map_err(|e| QueryError::StorageError(e.to_string()))?;
1815 self.in_transaction = true;
1816 Ok(QueryResult::Executed {
1817 message: "transaction started".to_string(),
1818 })
1819 }
1820
1821 PlanNode::Commit => {
1822 if !self.in_transaction {
1823 return Err(QueryError::Execution(
1824 "no active transaction to commit".into(),
1825 ));
1826 }
1827 self.catalog
1828 .commit_transaction()
1829 .map_err(|e| QueryError::StorageError(e.to_string()))?;
1830 self.in_transaction = false;
1831 Ok(QueryResult::Executed {
1832 message: "transaction committed".to_string(),
1833 })
1834 }
1835
1836 PlanNode::Rollback => {
1837 if !self.in_transaction {
1838 return Err(QueryError::Execution(
1839 "no active transaction to roll back".into(),
1840 ));
1841 }
1842 self.rollback_transaction_preserving_wal_archive()
1843 }
1844
1845 PlanNode::IndexScan { table, column, key } => {
1846 let key_value = literal_to_value(key)?;
1847 let tbl = self
1848 .catalog
1849 .get_table(table)
1850 .ok_or_else(|| QueryError::TableNotFound(table.to_string()))?;
1851 let columns: Vec<String> =
1852 tbl.schema.columns.iter().map(|c| c.name.clone()).collect();
1853
1854 // Fast path: the table has a B-tree on this column.
1855 // Uses index_lookup_all to return ALL matching rows for
1856 // both unique and non-unique indexes.
1857 if tbl.has_index(column) {
1858 let rids = tbl.index_lookup_all(column, &key_value);
1859 let mut rows: Vec<Vec<Value>> = Vec::with_capacity(rids.len());
1860 for rid in rids {
1861 if let Some(data) = tbl.heap.get(rid) {
1862 rows.push(decode_row(&tbl.schema, &data));
1863 }
1864 }
1865 return Ok(QueryResult::Rows { columns, rows });
1866 }
1867
1868 // Fallback: no index on this column. The planner emits IndexScan
1869 // eagerly (it has no visibility into which columns are indexed
1870 // at plan time), so here we must behave like SeqScan+Filter on
1871 // `.col = literal`: return *all* matching rows, not just the
1872 // first one. A non-indexed column isn't necessarily unique.
1873 // We compile the eq predicate once and stream without any
1874 // per-row decode for non-matching rows.
1875 let schema = &tbl.schema;
1876 let fast = FastLayout::new(schema);
1877 let synth_pred = Expr::BinaryOp(
1878 Box::new(Expr::Field(column.clone())),
1879 BinOp::Eq,
1880 Box::new(key.clone()),
1881 );
1882 if let Some(compiled) = compile_predicate(&synth_pred, &columns, &fast, schema) {
1883 // Mission F: skip the first 4 Vec doublings.
1884 let mut rows: Vec<Vec<Value>> = Vec::with_capacity(64);
1885 self.catalog
1886 .for_each_row_raw(table, |_rid, data| {
1887 if compiled(data) {
1888 rows.push(decode_row(schema, data));
1889 }
1890 })
1891 .map_err(|e| QueryError::StorageError(e.to_string()))?;
1892 return Ok(QueryResult::Rows { columns, rows });
1893 }
1894
1895 // Last resort: slow eq-check on materialised rows.
1896 let col_idx =
1897 schema
1898 .column_index(column)
1899 .ok_or_else(|| QueryError::ColumnNotFound {
1900 table: String::new(),
1901 column: column.clone(),
1902 })?;
1903 let rows: Vec<Vec<Value>> = tbl
1904 .scan()
1905 .filter_map(|(_, row)| {
1906 if row[col_idx] == key_value {
1907 Some(row)
1908 } else {
1909 None
1910 }
1911 })
1912 .collect();
1913 Ok(QueryResult::Rows { columns, rows })
1914 }
1915
1916 PlanNode::RangeScan {
1917 table,
1918 column,
1919 start,
1920 end,
1921 } => {
1922 let tbl = self
1923 .catalog
1924 .get_table(table)
1925 .ok_or_else(|| QueryError::TableNotFound(table.to_string()))?;
1926 let columns: Vec<String> =
1927 tbl.schema.columns.iter().map(|c| c.name.clone()).collect();
1928 let schema = &tbl.schema;
1929
1930 let start_val = match start {
1931 Some((expr, _)) => Some(literal_to_value(expr)?),
1932 None => None,
1933 };
1934 let end_val = match end {
1935 Some((expr, _)) => Some(literal_to_value(expr)?),
1936 None => None,
1937 };
1938 let start_inclusive = start.as_ref().map(|(_, inc)| *inc).unwrap_or(true);
1939 let end_inclusive = end.as_ref().map(|(_, inc)| *inc).unwrap_or(true);
1940
1941 // Non-unique index: walk the composite (value, rid) leaf
1942 // chain between prefix bounds, fetch each row from the heap,
1943 // and recheck. The recheck enforces exclusive bounds
1944 // (range_rids is inclusive) and defensively skips any decoded
1945 // null (nulls are never indexed, so they must not match).
1946 if tbl.is_index_unique(column) == Some(false) {
1947 if let Some(btree) = tbl.index(column) {
1948 if start_val.is_some() || end_val.is_some() {
1949 let col_idx = schema.column_index(column).ok_or_else(|| {
1950 QueryError::ColumnNotFound {
1951 table: String::new(),
1952 column: column.clone(),
1953 }
1954 })?;
1955 let rids = btree.range_rids(start_val.as_ref(), end_val.as_ref());
1956 let mut rows: Vec<Vec<Value>> = Vec::with_capacity(rids.len());
1957 for rid in rids {
1958 if let Some(data) = tbl.heap.get(rid) {
1959 let row = decode_row(schema, &data);
1960 if !row[col_idx].is_empty()
1961 && range_matches(
1962 &row[col_idx],
1963 &start_val,
1964 start_inclusive,
1965 &end_val,
1966 end_inclusive,
1967 )
1968 {
1969 rows.push(row);
1970 }
1971 }
1972 }
1973 return Ok(QueryResult::Rows { columns, rows });
1974 }
1975 }
1976 }
1977
1978 // Range scans use the btree fast path for unique indexes,
1979 // walking raw column-value keys directly.
1980 if tbl.is_index_unique(column) == Some(true) {
1981 if let Some(btree) = tbl.index(column) {
1982 let hits: Vec<(Value, RowId)> = match (&start_val, &end_val) {
1983 (Some(s), Some(e)) => btree.range(s, e).collect(),
1984 (Some(s), None) => btree.range_from(s),
1985 (None, Some(e)) => btree.range_to(e),
1986 (None, None) => {
1987 let rows: Vec<Vec<Value>> =
1988 tbl.scan().map(|(_, row)| row).collect();
1989 return Ok(QueryResult::Rows { columns, rows });
1990 }
1991 };
1992 let mut rows: Vec<Vec<Value>> = Vec::with_capacity(hits.len());
1993 for (key, rid) in hits {
1994 if !start_inclusive {
1995 if let Some(ref s) = start_val {
1996 if &key == s {
1997 continue;
1998 }
1999 }
2000 }
2001 if !end_inclusive {
2002 if let Some(ref e) = end_val {
2003 if &key == e {
2004 continue;
2005 }
2006 }
2007 }
2008 if let Some(data) = tbl.heap.get(rid) {
2009 rows.push(decode_row(schema, &data));
2010 }
2011 }
2012 return Ok(QueryResult::Rows { columns, rows });
2013 }
2014 }
2015
2016 // Fallback: no index — synthesize range predicate and scan.
2017 let fast = FastLayout::new(schema);
2018 let synth = synthesize_range_predicate(column, start, end);
2019 if let Some(compiled) = compile_predicate(&synth, &columns, &fast, schema) {
2020 let mut rows: Vec<Vec<Value>> = Vec::with_capacity(64);
2021 self.catalog
2022 .for_each_row_raw(table, |_rid, data| {
2023 if compiled(data) {
2024 rows.push(decode_row(schema, data));
2025 }
2026 })
2027 .map_err(|e| QueryError::StorageError(e.to_string()))?;
2028 return Ok(QueryResult::Rows { columns, rows });
2029 }
2030
2031 let col_idx =
2032 schema
2033 .column_index(column)
2034 .ok_or_else(|| QueryError::ColumnNotFound {
2035 table: String::new(),
2036 column: column.clone(),
2037 })?;
2038 let rows: Vec<Vec<Value>> = tbl
2039 .scan()
2040 .filter(|(_, row)| {
2041 range_matches(
2042 &row[col_idx],
2043 &start_val,
2044 start_inclusive,
2045 &end_val,
2046 end_inclusive,
2047 )
2048 })
2049 .map(|(_, row)| row)
2050 .collect();
2051 Ok(QueryResult::Rows { columns, rows })
2052 }
2053 }
2054 }
2055
2056 // ─── Materialized view operations ──────────────────────────────────────
2057
2058 /// Create a materialized view: execute the source query, store results
2059 /// in a new backing table, and register the view.
2060 fn create_view(&mut self, name: &str, query_text: &str) -> Result<(), QueryError> {
2061 if self.view_registry.is_view(name) {
2062 return Err(QueryError::ViewError(format!(
2063 "materialized view '{name}' already exists"
2064 )));
2065 }
2066 // Execute the source query to get the result set.
2067 let result = self.execute_powql(query_text)?;
2068 let (columns, rows) = match result {
2069 QueryResult::Rows { columns, rows } => (columns, rows),
2070 _ => return Err("view source query must be a SELECT".into()),
2071 };
2072 // Derive a schema for the backing table from the query result columns.
2073 let schema = self.derive_view_schema(name, &columns, &rows);
2074 // Create the backing table and insert the result rows.
2075 self.catalog
2076 .create_table(schema)
2077 .map_err(|e| QueryError::StorageError(e.to_string()))?;
2078 for row in &rows {
2079 self.catalog
2080 .insert(name, row)
2081 .map_err(|e| QueryError::StorageError(e.to_string()))?;
2082 }
2083 // Determine which base tables this view depends on by parsing the query.
2084 let depends_on = self.extract_view_deps(query_text);
2085 self.view_registry
2086 .register(ViewDef {
2087 name: name.to_string(),
2088 query: query_text.to_string(),
2089 depends_on,
2090 dirty: false,
2091 })
2092 .map_err(|e| QueryError::StorageError(e.to_string()))?;
2093 Ok(())
2094 }
2095
2096 /// Refresh a materialized view: re-execute its source query and replace
2097 /// the backing table's contents.
2098 fn refresh_view(&mut self, name: &str) -> Result<(), QueryError> {
2099 let def = self
2100 .view_registry
2101 .get(name)
2102 .ok_or_else(|| format!("materialized view '{name}' not found"))?;
2103 let query_text = def.query.clone();
2104 // Execute the source query.
2105 let result = self.execute_powql(&query_text)?;
2106 let (_columns, rows) = match result {
2107 QueryResult::Rows { columns, rows } => (columns, rows),
2108 _ => return Err("view source query must be a SELECT".into()),
2109 };
2110 // Clear old data and insert fresh results. Mission B2: logged
2111 // variant — view refreshes are a mutation and crash recovery
2112 // must see them.
2113 self.catalog
2114 .scan_delete_matching_logged(name, |_| true)
2115 .map_err(|e| QueryError::StorageError(e.to_string()))?;
2116 for row in &rows {
2117 self.catalog
2118 .insert(name, row)
2119 .map_err(|e| QueryError::StorageError(e.to_string()))?;
2120 }
2121 self.view_registry.mark_clean(name);
2122 Ok(())
2123 }
2124
2125 /// Drop a materialized view: remove the backing table and unregister.
2126 fn drop_view(&mut self, name: &str) -> Result<(), QueryError> {
2127 if !self.view_registry.is_view(name) {
2128 return Err(QueryError::ViewError(format!(
2129 "materialized view '{name}' not found"
2130 )));
2131 }
2132 self.view_registry
2133 .unregister(name)
2134 .map_err(|e| QueryError::StorageError(e.to_string()))?;
2135 self.catalog
2136 .drop_table(name)
2137 .map_err(|e| QueryError::StorageError(e.to_string()))?;
2138 Ok(())
2139 }
2140
2141 /// Derive a storage `Schema` for a view's backing table from query
2142 /// result column names and the first row's types.
2143 fn derive_view_schema(&self, name: &str, columns: &[String], rows: &[Vec<Value>]) -> Schema {
2144 use powdb_storage::types::{ColumnDef, TypeId};
2145 let cols: Vec<ColumnDef> = columns
2146 .iter()
2147 .enumerate()
2148 .map(|(i, col_name)| {
2149 let type_id = rows
2150 .first()
2151 .and_then(|row| row.get(i))
2152 .map(|v| v.type_id())
2153 .unwrap_or(TypeId::Str);
2154 ColumnDef {
2155 name: col_name.clone(),
2156 type_id,
2157 required: false,
2158 position: i as u16,
2159 }
2160 })
2161 .collect();
2162 Schema {
2163 table_name: name.to_string(),
2164 columns: cols,
2165 }
2166 }
2167
2168 /// Extract base table dependencies from a view's source query by
2169 /// parsing it and collecting the source table name.
2170 fn extract_view_deps(&self, query_text: &str) -> Vec<String> {
2171 use crate::parser::parse;
2172 match parse(query_text) {
2173 Ok(Statement::Query(q)) => {
2174 let mut deps = vec![q.source.clone()];
2175 for j in &q.joins {
2176 deps.push(j.source.clone());
2177 }
2178 deps
2179 }
2180 _ => Vec::new(),
2181 }
2182 }
2183
2184 // ─── Specialized fast paths ─────────────────────────────────────────────
2185 //
2186 // These methods are helpers for the `execute_plan` match arms above.
2187 // Each returns `Ok(Some(result))` when the fast path fires, `Ok(None)`
2188 // when the shape isn't supported (caller falls back to generic code).
2189
2190 /// Aggregate sum/avg/min/max over a single fixed-size i64 column, with
2191 /// an optional compiled filter predicate. Walks raw row bytes — zero
2192 /// per-row allocation. Uses i128 accumulator for sum/avg overflow safety.
2193 pub(super) fn agg_single_col_fast(
2194 &self,
2195 table: &str,
2196 col: &str,
2197 function: AggFunc,
2198 predicate: Option<&Expr>,
2199 ) -> Result<Option<QueryResult>, QueryError> {
2200 let schema = self
2201 .catalog
2202 .schema(table)
2203 .ok_or_else(|| QueryError::TableNotFound(table.to_string()))?
2204 .clone();
2205 let columns: Vec<String> = schema.columns.iter().map(|c| c.name.clone()).collect();
2206 let col_idx = match schema.column_index(col) {
2207 Some(i) => i,
2208 None => return Ok(None),
2209 };
2210 // Only fast-path fixed-size numeric columns (Int/Float) for
2211 // sum/avg/min/max/count. Mission D10: Float parity — prior version
2212 // bailed on Float columns, forcing them through the generic row-
2213 // decoding path that allocated a Vec<Value> per row and dispatched
2214 // on Value::cmp for every compare. f64 decode is structurally the
2215 // same as i64 (load 8 bytes, cast), so the fast path handles both.
2216 let col_type = schema.columns[col_idx].type_id;
2217 if col_type != TypeId::Int && col_type != TypeId::Float {
2218 return Ok(None);
2219 }
2220
2221 let fast = FastLayout::new(&schema);
2222 // Mission C Phase 20b: inline the numeric-column reader instead of
2223 // building a `Box<dyn Fn>`. Eliminates 100K vtable dispatches per
2224 // 100K-row agg scan — every reader call folds directly into the
2225 // hot loop below.
2226 let byte_offset = match fast.fixed_offsets[col_idx] {
2227 Some(o) => o,
2228 None => return Ok(None),
2229 };
2230 let bitmap_byte = col_idx / 8;
2231 let bitmap_bit = (col_idx % 8) as u32;
2232 let body_data_offset = 2 + fast.bitmap_size + byte_offset;
2233
2234 // Optional compiled filter.
2235 let compiled_pred: Option<CompiledPredicate> = match predicate {
2236 Some(pred) => match compile_predicate(pred, &columns, &fast, &schema) {
2237 Some(c) => Some(c),
2238 None => return Ok(None), // let generic path handle it
2239 },
2240 None => None,
2241 };
2242
2243 // Mission C Phase 20b: specialize the inner loop per aggregate
2244 // function. The previous version ran a `match function { ... }`
2245 // *inside* the closure, which kept LLVM from producing optimal
2246 // scalar code for each variant (agg_max regressed ~23% vs the
2247 // baseline Box<dyn Fn> version even though per-row vtable cost
2248 // should have been strictly lower). Pushing the match out of the
2249 // hot loop lets each specialized body fold cleanly into
2250 // `for_each_row_raw` and removes a captured `AggFunc` + match
2251 // dispatch per row.
2252 //
2253 // Mission D10: same specialisation applies to the Float branch.
2254 // For Min/Max we use `f64::total_cmp` so the result matches
2255 // `Value::Ord` — this is the same ordering ORDER BY and the
2256 // top-N sort fast path use, keeping semantics consistent across
2257 // read paths (NaN compares as greatest, -0.0 < +0.0 for
2258 // deterministic tie-breaking).
2259 //
2260 // Mission D11 Phase 1: each inner loop now splits on presence of
2261 // a predicate (`if let Some(pred) = &compiled_pred`) so the hot
2262 // body never re-tests `Option` per row, and reads column bytes
2263 // via `read_i64_unchecked` / `read_f64_unchecked` helpers that
2264 // drop two bounds checks per row (null bitmap byte + value
2265 // slice). Safety is carried by the `FastLayout` invariant that
2266 // `data_offset + 8 <= row_len` for any fixed-size column; see
2267 // the helper doc comments. Hot loops are macro-generated so the
2268 // with-pred / no-pred split can't drift between variants.
2269 let result = match col_type {
2270 TypeId::Int => match function {
2271 AggFunc::Sum | AggFunc::Avg => {
2272 let mut sum_i128: i128 = 0;
2273 let mut count: i64 = 0;
2274 agg_int_loop!(
2275 self,
2276 table,
2277 compiled_pred,
2278 bitmap_byte,
2279 bitmap_bit,
2280 body_data_offset,
2281 |v: i64| {
2282 count += 1;
2283 sum_i128 += v as i128;
2284 }
2285 );
2286 if matches!(function, AggFunc::Sum) {
2287 let clamped = sum_i128.clamp(i64::MIN as i128, i64::MAX as i128) as i64;
2288 QueryResult::Scalar(Value::Int(clamped))
2289 } else if count == 0 {
2290 QueryResult::Scalar(Value::Empty)
2291 } else {
2292 let avg = (sum_i128 as f64) / (count as f64);
2293 QueryResult::Scalar(Value::Float(avg))
2294 }
2295 }
2296 AggFunc::Min => {
2297 let mut min_v: Option<i64> = None;
2298 agg_int_loop!(
2299 self,
2300 table,
2301 compiled_pred,
2302 bitmap_byte,
2303 bitmap_bit,
2304 body_data_offset,
2305 |v: i64| {
2306 min_v = Some(match min_v {
2307 Some(m) => m.min(v),
2308 None => v,
2309 });
2310 }
2311 );
2312 QueryResult::Scalar(min_v.map(Value::Int).unwrap_or(Value::Empty))
2313 }
2314 AggFunc::Max => {
2315 let mut max_v: Option<i64> = None;
2316 agg_int_loop!(
2317 self,
2318 table,
2319 compiled_pred,
2320 bitmap_byte,
2321 bitmap_bit,
2322 body_data_offset,
2323 |v: i64| {
2324 max_v = Some(match max_v {
2325 Some(m) => m.max(v),
2326 None => v,
2327 });
2328 }
2329 );
2330 QueryResult::Scalar(max_v.map(Value::Int).unwrap_or(Value::Empty))
2331 }
2332 AggFunc::Count => {
2333 let mut count: i64 = 0;
2334 agg_int_loop!(
2335 self,
2336 table,
2337 compiled_pred,
2338 bitmap_byte,
2339 bitmap_bit,
2340 body_data_offset,
2341 |_v: i64| {
2342 count += 1;
2343 }
2344 );
2345 QueryResult::Scalar(Value::Int(count))
2346 }
2347 AggFunc::CountDistinct => {
2348 let mut seen = rustc_hash::FxHashSet::default();
2349 agg_int_loop!(
2350 self,
2351 table,
2352 compiled_pred,
2353 bitmap_byte,
2354 bitmap_bit,
2355 body_data_offset,
2356 |v: i64| {
2357 seen.insert(v);
2358 }
2359 );
2360 QueryResult::Scalar(Value::Int(seen.len() as i64))
2361 }
2362 },
2363 TypeId::Float => match function {
2364 AggFunc::Sum => {
2365 // Use a single f64 accumulator. Naive summation is
2366 // sufficient for MVP parity; if precision becomes an
2367 // issue on long scans we can upgrade to Kahan–Neumaier
2368 // compensated sum (~2x scalar cost, zero error growth).
2369 let mut sum: f64 = 0.0;
2370 agg_float_loop!(
2371 self,
2372 table,
2373 compiled_pred,
2374 bitmap_byte,
2375 bitmap_bit,
2376 body_data_offset,
2377 |v: f64| {
2378 sum += v;
2379 }
2380 );
2381 QueryResult::Scalar(Value::Float(sum))
2382 }
2383 AggFunc::Avg => {
2384 let mut sum: f64 = 0.0;
2385 let mut count: i64 = 0;
2386 agg_float_loop!(
2387 self,
2388 table,
2389 compiled_pred,
2390 bitmap_byte,
2391 bitmap_bit,
2392 body_data_offset,
2393 |v: f64| {
2394 sum += v;
2395 count += 1;
2396 }
2397 );
2398 if count == 0 {
2399 QueryResult::Scalar(Value::Empty)
2400 } else {
2401 QueryResult::Scalar(Value::Float(sum / count as f64))
2402 }
2403 }
2404 AggFunc::Min => {
2405 // `total_cmp` for deterministic NaN handling (matches
2406 // Value::Ord). NaN compares greatest, so Min will
2407 // correctly ignore it in favour of any finite value.
2408 let mut min_v: Option<f64> = None;
2409 agg_float_loop!(
2410 self,
2411 table,
2412 compiled_pred,
2413 bitmap_byte,
2414 bitmap_bit,
2415 body_data_offset,
2416 |v: f64| {
2417 min_v = Some(match min_v {
2418 Some(m) => {
2419 if v.total_cmp(&m).is_lt() {
2420 v
2421 } else {
2422 m
2423 }
2424 }
2425 None => v,
2426 });
2427 }
2428 );
2429 QueryResult::Scalar(min_v.map(Value::Float).unwrap_or(Value::Empty))
2430 }
2431 AggFunc::Max => {
2432 let mut max_v: Option<f64> = None;
2433 agg_float_loop!(
2434 self,
2435 table,
2436 compiled_pred,
2437 bitmap_byte,
2438 bitmap_bit,
2439 body_data_offset,
2440 |v: f64| {
2441 max_v = Some(match max_v {
2442 Some(m) => {
2443 if v.total_cmp(&m).is_gt() {
2444 v
2445 } else {
2446 m
2447 }
2448 }
2449 None => v,
2450 });
2451 }
2452 );
2453 QueryResult::Scalar(max_v.map(Value::Float).unwrap_or(Value::Empty))
2454 }
2455 AggFunc::Count => {
2456 let mut count: i64 = 0;
2457 agg_float_loop!(
2458 self,
2459 table,
2460 compiled_pred,
2461 bitmap_byte,
2462 bitmap_bit,
2463 body_data_offset,
2464 |_v: f64| {
2465 count += 1;
2466 }
2467 );
2468 QueryResult::Scalar(Value::Int(count))
2469 }
2470 AggFunc::CountDistinct => {
2471 // Hash on `f64::to_bits` — matches `Value::Hash`, so
2472 // distinct NaN bit patterns count as distinct and
2473 // -0.0/+0.0 count as distinct. Consistent with how
2474 // Float values are hashed in every other DISTINCT /
2475 // GROUP BY path.
2476 let mut seen = rustc_hash::FxHashSet::default();
2477 agg_float_loop!(
2478 self,
2479 table,
2480 compiled_pred,
2481 bitmap_byte,
2482 bitmap_bit,
2483 body_data_offset,
2484 |v: f64| {
2485 seen.insert(v.to_bits());
2486 }
2487 );
2488 QueryResult::Scalar(Value::Int(seen.len() as i64))
2489 }
2490 },
2491 _ => unreachable!("type guard above restricts to Int/Float"),
2492 };
2493 Ok(Some(result))
2494 }
2495
2496 /// `Project(Limit(Filter(SeqScan)))` and `Project(Limit(SeqScan))`.
2497 /// Streams rows, decodes only projected columns, stops at the limit.
2498 pub(super) fn project_filter_limit_fast(
2499 &self,
2500 table: &str,
2501 fields: &[ProjectField],
2502 limit: usize,
2503 predicate: Option<&Expr>,
2504 ) -> Result<Option<QueryResult>, QueryError> {
2505 let schema = self
2506 .catalog
2507 .schema(table)
2508 .ok_or_else(|| QueryError::TableNotFound(table.to_string()))?
2509 .clone();
2510 let all_columns: Vec<String> = schema.columns.iter().map(|c| c.name.clone()).collect();
2511
2512 // Each projection field must be a simple `.field` reference for this
2513 // fast path. Aliased or computed fields fall through.
2514 let mut proj_indices: Vec<usize> = Vec::with_capacity(fields.len());
2515 let mut proj_columns: Vec<String> = Vec::with_capacity(fields.len());
2516 for f in fields {
2517 let name = match &f.expr {
2518 Expr::Field(n) => n.clone(),
2519 _ => return Ok(None),
2520 };
2521 let idx = match all_columns.iter().position(|c| c == &name) {
2522 Some(i) => i,
2523 None => return Ok(None),
2524 };
2525 proj_indices.push(idx);
2526 proj_columns.push(f.alias.clone().unwrap_or(name));
2527 }
2528
2529 let fast = FastLayout::new(&schema);
2530 let row_layout = RowLayout::new(&schema);
2531
2532 let compiled_pred: Option<CompiledPredicate> = match predicate {
2533 Some(pred) => match compile_predicate(pred, &all_columns, &fast, &schema) {
2534 Some(c) => Some(c),
2535 None => return Ok(None),
2536 },
2537 None => None,
2538 };
2539
2540 let mut out: Vec<Vec<Value>> = Vec::with_capacity(limit.min(1024));
2541 // Mission D2: use try_for_each_row_raw to actually stop iterating
2542 // once the limit is reached. The previous `done` flag only short-
2543 // circuited the closure body, so a `limit 100` over 100K rows still
2544 // walked all 100K slots — burning ~30x SQLite on scan_filter_project_top100.
2545 self.catalog
2546 .try_for_each_row_raw(table, |_rid, data| {
2547 use std::ops::ControlFlow;
2548 if let Some(ref pred) = compiled_pred {
2549 if !pred(data) {
2550 return ControlFlow::Continue(());
2551 }
2552 }
2553 let row: Vec<Value> = proj_indices
2554 .iter()
2555 .map(|&ci| decode_column(&schema, &row_layout, data, ci))
2556 .collect();
2557 out.push(row);
2558 if out.len() >= limit {
2559 ControlFlow::Break(())
2560 } else {
2561 ControlFlow::Continue(())
2562 }
2563 })
2564 .map_err(|e| QueryError::StorageError(e.to_string()))?;
2565
2566 Ok(Some(QueryResult::Rows {
2567 columns: proj_columns,
2568 rows: out,
2569 }))
2570 }
2571
2572 /// `Project(Limit(Sort(Filter(SeqScan))))` and `Project(Limit(Sort(SeqScan)))`.
2573 /// Bounded top-N heap over the sort key. Only the sort key needs to be
2574 /// read per row; projected columns are decoded only for the final
2575 /// winning rows when the heap drains.
2576 pub(super) fn project_filter_sort_limit_fast(
2577 &self,
2578 table: &str,
2579 fields: &[ProjectField],
2580 sort_field: &str,
2581 descending: bool,
2582 limit: usize,
2583 predicate: Option<&Expr>,
2584 ) -> Result<Option<QueryResult>, QueryError> {
2585 if limit == 0 {
2586 // Degenerate case — empty result. Let the generic path handle it
2587 // for proper column naming.
2588 return Ok(None);
2589 }
2590 // The top-N heaps never hold more than `limit` rows, but `limit` is an
2591 // attacker-supplied literal (`order .x limit 99999999999`). Reserving
2592 // that capacity up front would allocate gigabytes and abort the
2593 // process before a single row is read. Cap the pre-allocation; the
2594 // heaps still grow on demand up to the true `limit`.
2595 const TOPN_PREALLOC_CAP: usize = 4096;
2596 let prealloc = limit.min(TOPN_PREALLOC_CAP);
2597 let schema = self
2598 .catalog
2599 .schema(table)
2600 .ok_or_else(|| QueryError::TableNotFound(table.to_string()))?
2601 .clone();
2602 let all_columns: Vec<String> = schema.columns.iter().map(|c| c.name.clone()).collect();
2603
2604 // Sort key must be a fixed-size numeric column (Int or Float).
2605 // Mission D10: extended from Int-only. Float sort keys use a
2606 // sortable-u64 transform (see `f64_to_sortable_u64`) so the heap
2607 // path stays keyed on `u64` and the whole branch shape is
2608 // identical to the Int case — no new heap types, no `total_cmp`
2609 // closures in the hot loop.
2610 let sort_idx = match schema.column_index(sort_field) {
2611 Some(i) => i,
2612 None => return Ok(None),
2613 };
2614 let sort_col_type = schema.columns[sort_idx].type_id;
2615 if sort_col_type != TypeId::Int && sort_col_type != TypeId::Float {
2616 return Ok(None);
2617 }
2618
2619 // Each projection field must be a simple `.field`.
2620 let mut proj_indices: Vec<usize> = Vec::with_capacity(fields.len());
2621 let mut proj_columns: Vec<String> = Vec::with_capacity(fields.len());
2622 for f in fields {
2623 let name = match &f.expr {
2624 Expr::Field(n) => n.clone(),
2625 _ => return Ok(None),
2626 };
2627 let idx = match all_columns.iter().position(|c| c == &name) {
2628 Some(i) => i,
2629 None => return Ok(None),
2630 };
2631 proj_indices.push(idx);
2632 proj_columns.push(f.alias.clone().unwrap_or(name));
2633 }
2634
2635 let fast = FastLayout::new(&schema);
2636 let row_layout = RowLayout::new(&schema);
2637 // Mission C Phase 20b: inline numeric-column reader (no Box<dyn Fn>).
2638 let sort_byte_offset = match fast.fixed_offsets[sort_idx] {
2639 Some(o) => o,
2640 None => return Ok(None),
2641 };
2642 let sort_bitmap_byte = sort_idx / 8;
2643 let sort_bitmap_bit = (sort_idx % 8) as u32;
2644 let sort_body_data_offset = 2 + fast.bitmap_size + sort_byte_offset;
2645
2646 let compiled_pred: Option<CompiledPredicate> = match predicate {
2647 Some(pred) => match compile_predicate(pred, &all_columns, &fast, &schema) {
2648 Some(c) => Some(c),
2649 None => return Ok(None),
2650 },
2651 None => None,
2652 };
2653
2654 // Bounded top-N heap. For `order .x desc limit N`, we want the N
2655 // largest values — use a min-heap so the smallest is at the top and
2656 // can be popped when a better candidate arrives. For ascending, use
2657 // a max-heap. We tie-break with a monotonic `seq` counter so the
2658 // result is deterministic and stable.
2659 //
2660 // To keep this simple we maintain two typed heaps and pick by
2661 // direction.
2662 let drained: Vec<Vec<u8>> = match sort_col_type {
2663 TypeId::Int => {
2664 let mut seq: u64 = 0;
2665 let mut heap_desc: BinaryHeap<Reverse<(i64, u64, Vec<u8>)>> =
2666 BinaryHeap::with_capacity(prealloc);
2667 let mut heap_asc: BinaryHeap<(i64, u64, Vec<u8>)> =
2668 BinaryHeap::with_capacity(prealloc);
2669
2670 self.catalog
2671 .for_each_row_raw(table, |_rid, data| {
2672 if let Some(ref pred) = compiled_pred {
2673 if !pred(data) {
2674 return;
2675 }
2676 }
2677 // Inlined int-column reader: null check + i64 decode.
2678 let base = row_body_base(data);
2679 let sort_data_offset = base + sort_body_data_offset;
2680 if data.len() < sort_data_offset + 8
2681 || data.len() <= base + 2 + sort_bitmap_byte
2682 {
2683 return;
2684 }
2685 let is_null =
2686 (data[base + 2 + sort_bitmap_byte] >> sort_bitmap_bit) & 1 == 1;
2687 if is_null {
2688 return;
2689 }
2690 let key = i64::from_le_bytes(
2691 data[sort_data_offset..sort_data_offset + 8]
2692 .try_into()
2693 .unwrap_or_else(|_| unreachable!()),
2694 );
2695 let id = seq;
2696 seq += 1;
2697
2698 if descending {
2699 if heap_desc.len() < limit {
2700 heap_desc.push(Reverse((key, id, data.to_vec())));
2701 } else if let Some(Reverse((top_key, _, _))) = heap_desc.peek() {
2702 if key > *top_key {
2703 heap_desc.pop();
2704 heap_desc.push(Reverse((key, id, data.to_vec())));
2705 }
2706 }
2707 } else if heap_asc.len() < limit {
2708 heap_asc.push((key, id, data.to_vec()));
2709 } else if let Some((top_key, _, _)) = heap_asc.peek() {
2710 if key < *top_key {
2711 heap_asc.pop();
2712 heap_asc.push((key, id, data.to_vec()));
2713 }
2714 }
2715 })
2716 .map_err(|e| QueryError::StorageError(e.to_string()))?;
2717
2718 let mut drained: Vec<(i64, u64, Vec<u8>)> = if descending {
2719 heap_desc.into_iter().map(|Reverse(t)| t).collect()
2720 } else {
2721 heap_asc.into_iter().collect()
2722 };
2723 if descending {
2724 drained.sort_unstable_by(|a, b| b.0.cmp(&a.0).then(a.1.cmp(&b.1)));
2725 } else {
2726 drained.sort_unstable_by(|a, b| a.0.cmp(&b.0).then(a.1.cmp(&b.1)));
2727 }
2728 drained.into_iter().map(|(_, _, d)| d).collect()
2729 }
2730 TypeId::Float => {
2731 // Novel angle: rather than introducing a `TotalF64` newtype
2732 // with `Ord via total_cmp`, transform the f64 bit pattern
2733 // into a sortable `u64` so `BinaryHeap<u64>` orders exactly
2734 // like `f64::total_cmp` would. Classic trick: flip the sign
2735 // bit on positives, flip all bits on negatives. Result:
2736 // - NaN (sign=0) stays greatest, matching total_cmp
2737 // - -0.0 sorts before +0.0, matching total_cmp
2738 // - Hot loop is branch-cheap (one compare + one xor)
2739 let mut seq: u64 = 0;
2740 let mut heap_desc: BinaryHeap<Reverse<(u64, u64, Vec<u8>)>> =
2741 BinaryHeap::with_capacity(prealloc);
2742 let mut heap_asc: BinaryHeap<(u64, u64, Vec<u8>)> =
2743 BinaryHeap::with_capacity(prealloc);
2744
2745 self.catalog
2746 .for_each_row_raw(table, |_rid, data| {
2747 if let Some(ref pred) = compiled_pred {
2748 if !pred(data) {
2749 return;
2750 }
2751 }
2752 let base = row_body_base(data);
2753 let sort_data_offset = base + sort_body_data_offset;
2754 if data.len() < sort_data_offset + 8
2755 || data.len() <= base + 2 + sort_bitmap_byte
2756 {
2757 return;
2758 }
2759 let is_null =
2760 (data[base + 2 + sort_bitmap_byte] >> sort_bitmap_bit) & 1 == 1;
2761 if is_null {
2762 return;
2763 }
2764 let bits = u64::from_le_bytes(
2765 data[sort_data_offset..sort_data_offset + 8]
2766 .try_into()
2767 .unwrap_or_else(|_| unreachable!()),
2768 );
2769 let key = f64_bits_to_sortable_u64(bits);
2770 let id = seq;
2771 seq += 1;
2772
2773 if descending {
2774 if heap_desc.len() < limit {
2775 heap_desc.push(Reverse((key, id, data.to_vec())));
2776 } else if let Some(Reverse((top_key, _, _))) = heap_desc.peek() {
2777 if key > *top_key {
2778 heap_desc.pop();
2779 heap_desc.push(Reverse((key, id, data.to_vec())));
2780 }
2781 }
2782 } else if heap_asc.len() < limit {
2783 heap_asc.push((key, id, data.to_vec()));
2784 } else if let Some((top_key, _, _)) = heap_asc.peek() {
2785 if key < *top_key {
2786 heap_asc.pop();
2787 heap_asc.push((key, id, data.to_vec()));
2788 }
2789 }
2790 })
2791 .map_err(|e| QueryError::StorageError(e.to_string()))?;
2792
2793 let mut drained: Vec<(u64, u64, Vec<u8>)> = if descending {
2794 heap_desc.into_iter().map(|Reverse(t)| t).collect()
2795 } else {
2796 heap_asc.into_iter().collect()
2797 };
2798 if descending {
2799 drained.sort_unstable_by(|a, b| b.0.cmp(&a.0).then(a.1.cmp(&b.1)));
2800 } else {
2801 drained.sort_unstable_by(|a, b| a.0.cmp(&b.0).then(a.1.cmp(&b.1)));
2802 }
2803 drained.into_iter().map(|(_, _, d)| d).collect()
2804 }
2805 _ => unreachable!("type guard above restricts to Int/Float"),
2806 };
2807
2808 let rows: Vec<Vec<Value>> = drained
2809 .into_iter()
2810 .map(|data| {
2811 proj_indices
2812 .iter()
2813 .map(|&ci| decode_column(&schema, &row_layout, &data, ci))
2814 .collect()
2815 })
2816 .collect();
2817
2818 Ok(Some(QueryResult::Rows {
2819 columns: proj_columns,
2820 rows,
2821 }))
2822 }
2823
2824 /// Gather the RowIds that a mutation should operate on, without
2825 /// materialising the full row set. Handles the shapes the planner emits
2826 /// for update/delete: SeqScan, IndexScan, and Filter(SeqScan). Other
2827 /// shapes fall back to `generic_rid_match`.
2828 ///
2829 /// Perf sprint: try to fuse the predicate evaluation and in-place
2830 /// byte-level mutation into a single heap walk. Returns `Some(result)`
2831 /// if the fused path fired, `None` to fall through to the generic
2832 /// two-pass code.
2833 ///
2834 /// Covers two shapes:
2835 /// 1. Fixed-width non-null literal assignments on non-indexed columns
2836 /// → byte-patch every matched row in place (row length unchanged).
2837 /// 2. Single var-col literal assignment on a non-indexed column
2838 /// → `patch_var_column_in_place` on every matched row (may shrink);
2839 /// rows that can't be patched in place are collected for fallback.
2840 fn try_fused_scan_update(
2841 &mut self,
2842 table: &str,
2843 predicate: &Expr,
2844 resolved: &[(usize, Value)],
2845 changed_cols: &[usize],
2846 ) -> Option<Result<QueryResult, QueryError>> {
2847 // Build compiled predicate. Requires a schema borrow that must be
2848 // dropped before we call scan_patch_matching_logged.
2849 let compiled = {
2850 let schema = self.catalog.schema(table)?;
2851 let columns: Vec<String> = schema.columns.iter().map(|c| c.name.clone()).collect();
2852 let fast = FastLayout::new(schema);
2853 compile_predicate(predicate, &columns, &fast, schema)?
2854 };
2855
2856 // ── Path 1: fixed-width fast patch ──────────────────────────
2857 let fixed_patches: Option<Vec<FastPatch>> = {
2858 let tbl = self.catalog.get_table(table)?;
2859 let schema = &tbl.schema;
2860 let all_fixed_nonnull = resolved
2861 .iter()
2862 .all(|(idx, val)| is_fixed_size(schema.columns[*idx].type_id) && !val.is_empty());
2863 let no_indexed = !resolved.iter().any(|(idx, _)| tbl.has_indexed_col(*idx));
2864 if all_fixed_nonnull && no_indexed {
2865 let layout = RowLayout::new(schema);
2866 let bitmap_size = layout.bitmap_size();
2867 Some(
2868 resolved
2869 .iter()
2870 .map(|(idx, val)| {
2871 let fixed_off = layout
2872 .fixed_offset(*idx)
2873 .expect("is_fixed_size already checked");
2874 let field_off = 2 + bitmap_size + fixed_off;
2875 let bytes: FixedBytes = match val {
2876 Value::Int(v) => FixedBytes::I64(v.to_le_bytes()),
2877 Value::Float(v) => FixedBytes::F64(v.to_le_bytes()),
2878 Value::Bool(v) => FixedBytes::Bool(if *v { 1 } else { 0 }),
2879 Value::DateTime(v) => FixedBytes::I64(v.to_le_bytes()),
2880 Value::Uuid(v) => FixedBytes::Uuid(*v),
2881 _ => unreachable!("all_fixed_nonnull guard"),
2882 };
2883 FastPatch {
2884 field_off,
2885 bitmap_byte_off: 2 + idx / 8,
2886 bit_mask: 1u8 << (idx % 8),
2887 bytes,
2888 }
2889 })
2890 .collect(),
2891 )
2892 } else {
2893 None
2894 }
2895 };
2896 if let Some(patches) = fixed_patches {
2897 let result = self
2898 .catalog
2899 .scan_patch_matching_logged(table, compiled, |row| {
2900 let base = row_body_base(row);
2901 for p in &patches {
2902 row[base + p.bitmap_byte_off] &= !p.bit_mask;
2903 let field_bytes = p.bytes.as_slice();
2904 row[base + p.field_off..base + p.field_off + field_bytes.len()]
2905 .copy_from_slice(field_bytes);
2906 }
2907 Some(row.len() as u16)
2908 })
2909 .map_err(|e| e.to_string());
2910 match result {
2911 Ok((count, _)) => {
2912 self.view_registry.mark_dependents_dirty(table);
2913 return Some(Ok(QueryResult::Modified(count)));
2914 }
2915 Err(e) => return Some(Err(QueryError::Execution(e))),
2916 }
2917 }
2918
2919 // ── Path 2: single var-col shrink fast patch ────────────────
2920 let var_patch: Option<(usize, Option<Vec<u8>>)> = {
2921 let tbl = self.catalog.get_table(table)?;
2922 let schema = &tbl.schema;
2923 let is_single = resolved.len() == 1;
2924 let is_var = is_single && !is_fixed_size(schema.columns[resolved[0].0].type_id);
2925 let no_indexed = !resolved.iter().any(|(idx, _)| tbl.has_indexed_col(*idx));
2926 if is_single && is_var && no_indexed {
2927 let (idx, val) = &resolved[0];
2928 let bytes_opt = match val {
2929 Value::Str(s) => Some(s.as_bytes().to_vec()),
2930 Value::Bytes(b) => Some(b.clone()),
2931 Value::Empty => None,
2932 _ => return None, // type mismatch, fall through
2933 };
2934 Some((*idx, bytes_opt))
2935 } else {
2936 None
2937 }
2938 };
2939 if let Some((col_idx, ref new_bytes_opt)) = var_patch {
2940 // Build a fresh RowLayout before the mutable borrow.
2941 let layout = {
2942 let schema = self.catalog.schema(table)?;
2943 RowLayout::new(schema)
2944 };
2945 let new_bytes_ref: Option<&[u8]> = new_bytes_opt.as_deref();
2946 let result = self
2947 .catalog
2948 .scan_patch_matching_logged(table, compiled, |row| {
2949 patch_var_column_in_place(row, &layout, col_idx, new_bytes_ref)
2950 })
2951 .map_err(|e| e.to_string());
2952 match result {
2953 Ok((mut count, fallback_rids)) => {
2954 // Handle rows where in-place patch failed (new > old).
2955 for rid in fallback_rids {
2956 let mut row = match self.catalog.get(table, rid) {
2957 Some(r) => r,
2958 None => continue,
2959 };
2960 for (idx, val) in resolved.iter() {
2961 row[*idx] = val.clone();
2962 }
2963 if let Err(e) =
2964 self.catalog
2965 .update_hinted(table, rid, &row, Some(changed_cols))
2966 {
2967 return Some(Err(QueryError::StorageError(e.to_string())));
2968 }
2969 count += 1;
2970 }
2971 self.view_registry.mark_dependents_dirty(table);
2972 return Some(Ok(QueryResult::Modified(count)));
2973 }
2974 Err(e) => return Some(Err(QueryError::Execution(e))),
2975 }
2976 }
2977
2978 None // no fused path applicable — fall through
2979 }
2980
2981 /// Mission C Phase 3: schema is looked up via `self.catalog.schema(table)`
2982 /// inside the branches that actually need it. Previously the caller had
2983 /// to clone the full Schema (6+ String allocs) before every mutation just
2984 /// so this function could borrow it — a cost the update/delete hot path
2985 /// did not need.
2986 fn collect_rids_for_mutation(
2987 &mut self,
2988 input: &PlanNode,
2989 table: &str,
2990 ) -> Result<Vec<RowId>, QueryError> {
2991 match input {
2992 PlanNode::SeqScan { table: t } if t == table => {
2993 // "Update/delete everything" — rare but legal.
2994 let rids: Vec<RowId> = self
2995 .catalog
2996 .scan(table)
2997 .map_err(|e| QueryError::StorageError(e.to_string()))?
2998 .map(|(rid, _)| rid)
2999 .collect();
3000 Ok(rids)
3001 }
3002 PlanNode::IndexScan {
3003 table: t,
3004 column,
3005 key,
3006 } if t == table => {
3007 let key_value = literal_to_value(key)?;
3008
3009 // Indexed case: single lookup, 0 or 1 rows.
3010 // Mission D7: int-specialized fast path on int-keyed indexes
3011 // (primary keys, created_at, etc.) — the common case for
3012 // `update_by_pk` / `delete where id = ?`.
3013 //
3014 // Scope the `tbl` borrow so it's released before we fall
3015 // through to the scan-based paths below (which reborrow
3016 // `self.catalog`).
3017 {
3018 let tbl = self
3019 .catalog
3020 .get_table(table)
3021 .ok_or_else(|| QueryError::TableNotFound(table.to_string()))?;
3022 if tbl.has_index(column) {
3023 let rids = tbl.index_lookup_all(column, &key_value);
3024 return Ok(rids);
3025 }
3026 }
3027
3028 // No index: the planner folds `.col = literal` to IndexScan
3029 // regardless of whether the column is actually unique. When
3030 // there's no index we must behave like Filter(SeqScan) and
3031 // return *all* matching RIDs — not just the first one.
3032 let schema = self
3033 .catalog
3034 .schema(table)
3035 .ok_or_else(|| QueryError::TableNotFound(table.to_string()))?;
3036 let columns: Vec<String> = schema.columns.iter().map(|c| c.name.clone()).collect();
3037 let fast = FastLayout::new(schema);
3038 let synth = Expr::BinaryOp(
3039 Box::new(Expr::Field(column.clone())),
3040 BinOp::Eq,
3041 Box::new(key.clone()),
3042 );
3043 if let Some(compiled) = compile_predicate(&synth, &columns, &fast, schema) {
3044 // Mission F: skip the first 4 Vec doublings.
3045 let mut rids: Vec<RowId> = Vec::with_capacity(64);
3046 self.catalog
3047 .for_each_row_raw(table, |rid, data| {
3048 if compiled(data) {
3049 rids.push(rid);
3050 }
3051 })
3052 .map_err(|e| QueryError::StorageError(e.to_string()))?;
3053 return Ok(rids);
3054 }
3055
3056 // Fallback: decode each row, compare values.
3057 let col_idx =
3058 schema
3059 .column_index(column)
3060 .ok_or_else(|| QueryError::ColumnNotFound {
3061 table: String::new(),
3062 column: column.clone(),
3063 })?;
3064 let rids: Vec<RowId> = self
3065 .catalog
3066 .scan(table)
3067 .map_err(|e| QueryError::StorageError(e.to_string()))?
3068 .filter_map(|(rid, row)| {
3069 if row[col_idx] == key_value {
3070 Some(rid)
3071 } else {
3072 None
3073 }
3074 })
3075 .collect();
3076 Ok(rids)
3077 }
3078 PlanNode::Filter {
3079 input: inner,
3080 predicate,
3081 } => {
3082 if let PlanNode::SeqScan { table: t } = inner.as_ref() {
3083 if t != table {
3084 return self.generic_rid_match(input, table);
3085 }
3086 let schema = self
3087 .catalog
3088 .schema(table)
3089 .ok_or_else(|| QueryError::TableNotFound(table.to_string()))?;
3090 let columns: Vec<String> =
3091 schema.columns.iter().map(|c| c.name.clone()).collect();
3092 let fast = FastLayout::new(schema);
3093 let row_layout = RowLayout::new(schema);
3094
3095 // Try compiled predicate first.
3096 if let Some(compiled) = compile_predicate(predicate, &columns, &fast, schema) {
3097 // Mission F: skip the first 4 Vec doublings.
3098 let mut rids: Vec<RowId> = Vec::with_capacity(64);
3099 self.catalog
3100 .for_each_row_raw(table, |rid, data| {
3101 if compiled(data) {
3102 rids.push(rid);
3103 }
3104 })
3105 .map_err(|e| QueryError::StorageError(e.to_string()))?;
3106 return Ok(rids);
3107 }
3108
3109 // Fallback: selective decode + eval.
3110 let pred_cols = predicate_column_indices(predicate, &columns);
3111 let mut rids: Vec<RowId> = Vec::with_capacity(64);
3112 self.catalog
3113 .for_each_row_raw(table, |rid, data| {
3114 let pred_row = decode_selective(schema, &row_layout, data, &pred_cols);
3115 if eval_predicate(predicate, &pred_row, &columns) {
3116 rids.push(rid);
3117 }
3118 })
3119 .map_err(|e| QueryError::StorageError(e.to_string()))?;
3120 return Ok(rids);
3121 }
3122 self.generic_rid_match(input, table)
3123 }
3124 _ => self.generic_rid_match(input, table),
3125 }
3126 }
3127
3128 /// Last-ditch generic match: execute the plan, collect matching rows,
3129 /// then find corresponding RowIds by value equality. This is the old
3130 /// O(N*M) code path; only used when the plan shape is something exotic.
3131 fn generic_rid_match(
3132 &mut self,
3133 input: &PlanNode,
3134 table: &str,
3135 ) -> Result<Vec<RowId>, QueryError> {
3136 let result = self.execute_plan(input)?;
3137 let rows = match result {
3138 QueryResult::Rows { rows, .. } => rows,
3139 _ => return Err("mutation source must be rows".into()),
3140 };
3141 let matching: Vec<RowId> = self
3142 .catalog
3143 .scan(table)
3144 .map_err(|e| QueryError::StorageError(e.to_string()))?
3145 .filter(|(_, row)| rows.iter().any(|r| r == row))
3146 .map(|(rid, _)| rid)
3147 .collect();
3148 Ok(matching)
3149 }
3150}
3151
3152pub(super) fn execute_window(
3153 result: QueryResult,
3154 windows: &[WindowDef],
3155) -> Result<QueryResult, QueryError> {
3156 let (mut columns, mut rows) = match result {
3157 QueryResult::Rows { columns, rows } => (columns, rows),
3158 _ => return Err("window function requires row input".into()),
3159 };
3160
3161 for wdef in windows {
3162 // Resolve partition/order column indices against current columns.
3163 let part_indices: Vec<usize> = wdef
3164 .partition_by
3165 .iter()
3166 .map(|name| {
3167 columns
3168 .iter()
3169 .position(|c| c == name)
3170 .ok_or_else(|| format!("window partition column '{name}' not found"))
3171 })
3172 .collect::<Result<Vec<_>, _>>()?;
3173
3174 let ord_indices: Vec<(usize, bool)> = wdef
3175 .order_by
3176 .iter()
3177 .map(|sk| {
3178 columns
3179 .iter()
3180 .position(|c| c == &sk.field)
3181 .map(|i| (i, sk.descending))
3182 .ok_or_else(|| format!("window order column '{}' not found", sk.field))
3183 })
3184 .collect::<Result<Vec<_>, _>>()?;
3185
3186 // Resolve the argument column index (for aggregate windows).
3187 let arg_col_idx: Option<usize> = if let Some(arg) = wdef.args.first() {
3188 match arg {
3189 Expr::Field(name) => {
3190 if name == "*" {
3191 None // count(*) style — no specific column
3192 } else {
3193 Some(
3194 columns
3195 .iter()
3196 .position(|c| c == name)
3197 .ok_or_else(|| format!("window arg column '{name}' not found"))?,
3198 )
3199 }
3200 }
3201 _ => None,
3202 }
3203 } else {
3204 None
3205 };
3206
3207 // Build a sort-index to sort rows by partition_by then order_by
3208 // without actually reordering the original Vec (we need original
3209 // order to write results back).
3210 let n = rows.len();
3211 let mut indices: Vec<usize> = (0..n).collect();
3212 indices.sort_by(|&a, &b| {
3213 // Compare partition keys first.
3214 for &pi in &part_indices {
3215 let cmp = rows[a][pi].cmp(&rows[b][pi]);
3216 if cmp != std::cmp::Ordering::Equal {
3217 return cmp;
3218 }
3219 }
3220 // Then order keys.
3221 for &(oi, desc) in &ord_indices {
3222 let cmp = rows[a][oi].cmp(&rows[b][oi]);
3223 if cmp != std::cmp::Ordering::Equal {
3224 return if desc { cmp.reverse() } else { cmp };
3225 }
3226 }
3227 std::cmp::Ordering::Equal
3228 });
3229
3230 // SQL window-frame semantics: with no `order` clause the frame for an
3231 // aggregate window is the ENTIRE partition, not the running prefix.
3232 // The loop below computes running values; for the no-order case we
3233 // back-fill every row of a partition with the partition's final
3234 // (i.e. complete) aggregate once its boundary is reached. Ranking
3235 // functions are untouched — row_number/rank/dense_rank are inherently
3236 // positional.
3237 let whole_partition_frame = wdef.order_by.is_empty()
3238 && matches!(
3239 wdef.function,
3240 WindowFunc::Sum
3241 | WindowFunc::Avg
3242 | WindowFunc::Count
3243 | WindowFunc::Min
3244 | WindowFunc::Max
3245 );
3246 // Original row indices of the partition currently being scanned
3247 // (only tracked when back-filling is needed).
3248 let mut partition_row_indices: Vec<usize> = Vec::new();
3249
3250 // Compute window values in sorted order, tracking partition boundaries.
3251 let mut win_values: Vec<Value> = vec![Value::Empty; n];
3252 let mut partition_start = 0usize;
3253 // Running state for aggregate windows:
3254 let mut running_count: i64 = 0;
3255 let mut running_int_sum: i64 = 0;
3256 let mut running_float_sum: f64 = 0.0;
3257 let mut running_saw_float = false;
3258 let mut running_min: Option<Value> = None;
3259 let mut running_max: Option<Value> = None;
3260 let mut rank_counter: i64 = 0;
3261 let mut dense_rank_counter: i64 = 0;
3262 let mut prev_order_key: Option<Vec<Value>> = None;
3263 let mut same_rank_count: i64 = 0;
3264
3265 for sorted_pos in 0..n {
3266 let row_idx = indices[sorted_pos];
3267
3268 // Detect partition boundary.
3269 let new_partition = if sorted_pos == 0 {
3270 true
3271 } else {
3272 let prev_row_idx = indices[sorted_pos - 1];
3273 part_indices
3274 .iter()
3275 .any(|&pi| rows[row_idx][pi] != rows[prev_row_idx][pi])
3276 };
3277
3278 if new_partition {
3279 // No-order aggregate frame: the partition that just ended is
3280 // complete, so its final running value IS the whole-partition
3281 // aggregate. Back-fill it onto every row of that partition.
3282 if whole_partition_frame && sorted_pos > 0 {
3283 let final_v = win_values[indices[sorted_pos - 1]].clone();
3284 for ri in partition_row_indices.drain(..) {
3285 win_values[ri] = final_v.clone();
3286 }
3287 }
3288 partition_start = sorted_pos;
3289 running_count = 0;
3290 running_int_sum = 0;
3291 running_float_sum = 0.0;
3292 running_saw_float = false;
3293 running_min = None;
3294 running_max = None;
3295 rank_counter = 0;
3296 dense_rank_counter = 0;
3297 prev_order_key = None;
3298 same_rank_count = 0;
3299 }
3300
3301 // Extract current order key for rank tracking.
3302 let current_order_key: Vec<Value> = ord_indices
3303 .iter()
3304 .map(|&(oi, _)| rows[row_idx][oi].clone())
3305 .collect();
3306 let same_as_prev = prev_order_key.as_ref() == Some(¤t_order_key);
3307
3308 let value = match wdef.function {
3309 WindowFunc::RowNumber => Value::Int((sorted_pos - partition_start + 1) as i64),
3310 WindowFunc::Rank => {
3311 if same_as_prev {
3312 same_rank_count += 1;
3313 } else {
3314 rank_counter += same_rank_count + 1;
3315 same_rank_count = 0;
3316 if rank_counter == 0 {
3317 rank_counter = 1;
3318 }
3319 }
3320 Value::Int(rank_counter)
3321 }
3322 WindowFunc::DenseRank => {
3323 if !same_as_prev {
3324 dense_rank_counter += 1;
3325 }
3326 Value::Int(dense_rank_counter)
3327 }
3328 WindowFunc::Sum => {
3329 if let Some(ci) = arg_col_idx {
3330 match &rows[row_idx][ci] {
3331 Value::Int(v) => running_int_sum += v,
3332 Value::Float(v) => {
3333 running_float_sum += v;
3334 running_saw_float = true;
3335 }
3336 _ => {}
3337 }
3338 }
3339 if running_saw_float {
3340 Value::Float(running_float_sum + running_int_sum as f64)
3341 } else {
3342 Value::Int(running_int_sum)
3343 }
3344 }
3345 WindowFunc::Avg => {
3346 if let Some(ci) = arg_col_idx {
3347 match &rows[row_idx][ci] {
3348 Value::Int(v) => {
3349 running_float_sum += *v as f64;
3350 running_count += 1;
3351 }
3352 Value::Float(v) => {
3353 running_float_sum += v;
3354 running_count += 1;
3355 }
3356 _ => {}
3357 }
3358 }
3359 if running_count == 0 {
3360 Value::Empty
3361 } else {
3362 Value::Float(running_float_sum / running_count as f64)
3363 }
3364 }
3365 WindowFunc::Count => {
3366 if let Some(ci) = arg_col_idx {
3367 if !rows[row_idx][ci].is_empty() {
3368 running_count += 1;
3369 }
3370 } else {
3371 // count(*) — count all rows
3372 running_count += 1;
3373 }
3374 Value::Int(running_count)
3375 }
3376 WindowFunc::Min => {
3377 if let Some(ci) = arg_col_idx {
3378 let v = &rows[row_idx][ci];
3379 if !v.is_empty() {
3380 running_min = Some(match &running_min {
3381 None => v.clone(),
3382 Some(cur) => {
3383 if v < cur {
3384 v.clone()
3385 } else {
3386 cur.clone()
3387 }
3388 }
3389 });
3390 }
3391 }
3392 running_min.clone().unwrap_or(Value::Empty)
3393 }
3394 WindowFunc::Max => {
3395 if let Some(ci) = arg_col_idx {
3396 let v = &rows[row_idx][ci];
3397 if !v.is_empty() {
3398 running_max = Some(match &running_max {
3399 None => v.clone(),
3400 Some(cur) => {
3401 if v > cur {
3402 v.clone()
3403 } else {
3404 cur.clone()
3405 }
3406 }
3407 });
3408 }
3409 }
3410 running_max.clone().unwrap_or(Value::Empty)
3411 }
3412 };
3413
3414 prev_order_key = Some(current_order_key);
3415 win_values[row_idx] = value;
3416 if whole_partition_frame {
3417 partition_row_indices.push(row_idx);
3418 }
3419 }
3420
3421 // Back-fill the final partition (the loop only flushes at boundaries).
3422 if whole_partition_frame && n > 0 {
3423 let final_v = win_values[indices[n - 1]].clone();
3424 for ri in partition_row_indices.drain(..) {
3425 win_values[ri] = final_v.clone();
3426 }
3427 }
3428
3429 // Append the computed window column to each row.
3430 for (ri, row) in rows.iter_mut().enumerate() {
3431 row.push(win_values[ri].clone());
3432 }
3433 columns.push(wdef.output_name.clone());
3434 }
3435
3436 Ok(QueryResult::Rows { columns, rows })
3437}
3438
3439/// Mission E2b: compute one aggregate over a set of rows in a group.
3440pub(super) fn compute_group_aggregate(
3441 func: AggFunc,
3442 all_rows: &[Vec<Value>],
3443 row_indices: &[usize],
3444 col_idx: usize,
3445) -> Value {
3446 match func {
3447 AggFunc::Count => {
3448 if col_idx == usize::MAX {
3449 // count(*) — count all rows in the group.
3450 return Value::Int(row_indices.len() as i64);
3451 }
3452 let count = row_indices
3453 .iter()
3454 .filter(|&&ri| !all_rows[ri][col_idx].is_empty())
3455 .count();
3456 Value::Int(count as i64)
3457 }
3458 AggFunc::CountDistinct => {
3459 let mut seen = std::collections::HashSet::new();
3460 for &ri in row_indices {
3461 let v = &all_rows[ri][col_idx];
3462 if !v.is_empty() {
3463 seen.insert(v.clone());
3464 }
3465 }
3466 Value::Int(seen.len() as i64)
3467 }
3468 AggFunc::Sum => {
3469 // Mirror the scalar Sum path: accumulate int and float
3470 // contributions separately and promote the final result to
3471 // Float if any Float row was observed. Prevents silent
3472 // drop of Float columns in GROUP BY aggregates.
3473 let mut int_sum: i64 = 0;
3474 let mut float_sum: f64 = 0.0;
3475 let mut saw_float = false;
3476 for &ri in row_indices {
3477 match &all_rows[ri][col_idx] {
3478 Value::Int(v) => int_sum += v,
3479 Value::Float(v) => {
3480 float_sum += *v;
3481 saw_float = true;
3482 }
3483 _ => {}
3484 }
3485 }
3486 if saw_float {
3487 Value::Float(float_sum + int_sum as f64)
3488 } else {
3489 Value::Int(int_sum)
3490 }
3491 }
3492 AggFunc::Avg => {
3493 let mut sum = 0.0f64;
3494 let mut count = 0usize;
3495 for &ri in row_indices {
3496 match &all_rows[ri][col_idx] {
3497 Value::Int(v) => {
3498 sum += *v as f64;
3499 count += 1;
3500 }
3501 Value::Float(v) => {
3502 sum += *v;
3503 count += 1;
3504 }
3505 _ => {}
3506 }
3507 }
3508 if count == 0 {
3509 Value::Empty
3510 } else {
3511 Value::Float(sum / count as f64)
3512 }
3513 }
3514 AggFunc::Min => row_indices
3515 .iter()
3516 .map(|&ri| &all_rows[ri][col_idx])
3517 .filter(|v| !v.is_empty())
3518 .min()
3519 .cloned()
3520 .unwrap_or(Value::Empty),
3521 AggFunc::Max => row_indices
3522 .iter()
3523 .map(|&ri| &all_rows[ri][col_idx])
3524 .filter(|v| !v.is_empty())
3525 .max()
3526 .cloned()
3527 .unwrap_or(Value::Empty),
3528 }
3529}
3530
3531/// Mission E1.3: try to extract equi-join key indices from a join `on`
3532/// predicate. Returns `Some((left_col_idx, right_col_idx))` when the
3533/// predicate is exactly `L = R` (or `R = L`) and both sides resolve
3534/// cleanly — `L` to the left subtree's column list and `R` to the right
3535/// subtree's column list.
3536///
3537/// This is deliberately narrow. We only recognise the two shapes:
3538/// * `QualifiedField = QualifiedField` (`u.id = o.user_id`)
3539/// * `Field = Field` (`.id = .user_id`, unqualified)
3540///
3541/// Anything else — conjunctions, constants, function calls, or predicates
3542/// that touch the same side on both halves — falls through to the
3543/// nested-loop path unchanged.
3544pub(super) fn try_extract_equi_join_keys(
3545 pred: &Expr,
3546 left_columns: &[String],
3547 right_columns: &[String],
3548) -> Option<(usize, usize)> {
3549 let (lhs, op, rhs) = match pred {
3550 Expr::BinaryOp(l, op, r) => (l.as_ref(), *op, r.as_ref()),
3551 _ => return None,
3552 };
3553 if op != BinOp::Eq {
3554 return None;
3555 }
3556 // Normal orientation: lhs in left, rhs in right.
3557 if let (Some(li), Some(ri)) = (
3558 resolve_side_column(lhs, left_columns),
3559 resolve_side_column(rhs, right_columns),
3560 ) {
3561 return Some((li, ri));
3562 }
3563 // Swapped: rhs in left, lhs in right. Both sides of `=` are
3564 // commutative so this is safe.
3565 if let (Some(li), Some(ri)) = (
3566 resolve_side_column(rhs, left_columns),
3567 resolve_side_column(lhs, right_columns),
3568 ) {
3569 return Some((li, ri));
3570 }
3571 None
3572}
3573
3574fn resolve_side_column(expr: &Expr, columns: &[String]) -> Option<usize> {
3575 match expr {
3576 Expr::QualifiedField { qualifier, field } => {
3577 // Byte-level match so we don't allocate a fresh `format!` on
3578 // every call — this runs once per plan, so allocation would be
3579 // cheap, but the match is trivial enough to keep inline with
3580 // the eval_expr version.
3581 let q = qualifier.as_bytes();
3582 let f = field.as_bytes();
3583 columns.iter().position(|c| {
3584 let b = c.as_bytes();
3585 b.len() == q.len() + 1 + f.len()
3586 && b[..q.len()] == *q
3587 && b[q.len()] == b'.'
3588 && b[q.len() + 1..] == *f
3589 })
3590 }
3591 Expr::Field(name) => columns.iter().position(|c| c == name),
3592 _ => None,
3593 }
3594}
3595
3596/// Mission E1.3: O(L + R) hash join. Builds a `FxHashMap<Value, Vec<usize>>`
3597/// over the right (inner) side's join keys, then streams the left (outer)
3598/// side and for each probe row emits every combined row whose right-side
3599/// key matches. For `JoinKind::LeftOuter`, unmatched left rows are emitted
3600/// padded with `Value::Empty` on the right side.
3601///
3602/// The right side is always the build side. That choice is forced for
3603/// LeftOuter (the left side must stream so we can detect orphans), and
3604/// for Inner it's a reasonable default — left-deep plans tend to grow the
3605/// left side with each join, so the un-joined right leaf is often the
3606/// smaller of the two at each level.
3607pub(super) fn hash_join(
3608 left_columns: Vec<String>,
3609 left_rows: Vec<Vec<Value>>,
3610 right_columns: Vec<String>,
3611 right_rows: Vec<Vec<Value>>,
3612 left_key_idx: usize,
3613 right_key_idx: usize,
3614 kind: JoinKind,
3615) -> QueryResult {
3616 use rustc_hash::FxHashMap;
3617
3618 let n_left = left_columns.len();
3619 let n_right = right_columns.len();
3620 let mut columns = Vec::with_capacity(n_left + n_right);
3621 columns.extend(left_columns);
3622 columns.extend(right_columns);
3623
3624 // Build: right_key -> list of right-row indices. Pre-size to the row
3625 // count so the map doesn't rehash mid-build.
3626 let mut build: FxHashMap<Value, Vec<usize>> =
3627 FxHashMap::with_capacity_and_hasher(right_rows.len(), Default::default());
3628 for (i, row) in right_rows.iter().enumerate() {
3629 // Skip Empty keys on the build side — they can never match under
3630 // SQL semantics (NULL ≠ NULL) and would collapse all nullables to
3631 // one bucket.
3632 if matches!(row[right_key_idx], Value::Empty) {
3633 continue;
3634 }
3635 build.entry(row[right_key_idx].clone()).or_default().push(i);
3636 }
3637
3638 // Reasonable starting capacity — inner joins produce ≥ left_rows.len()
3639 // rows in the common 1:1 case, left-outer always emits ≥ left_rows.len().
3640 let mut rows: Vec<Vec<Value>> = Vec::with_capacity(left_rows.len());
3641
3642 for left_row in &left_rows {
3643 let key = &left_row[left_key_idx];
3644 let matched = if matches!(key, Value::Empty) {
3645 None
3646 } else {
3647 build.get(key)
3648 };
3649 match matched {
3650 Some(matches) if !matches.is_empty() => {
3651 for &ri in matches {
3652 let right_row = &right_rows[ri];
3653 let mut combined = Vec::with_capacity(n_left + n_right);
3654 combined.extend_from_slice(left_row);
3655 combined.extend_from_slice(right_row);
3656 rows.push(combined);
3657 }
3658 }
3659 _ => {
3660 if matches!(kind, JoinKind::LeftOuter) {
3661 let mut row = Vec::with_capacity(n_left + n_right);
3662 row.extend_from_slice(left_row);
3663 row.resize(n_left + n_right, Value::Empty);
3664 rows.push(row);
3665 }
3666 }
3667 }
3668 }
3669
3670 QueryResult::Rows { columns, rows }
3671}
3672
3673/// Lower unindexed `RangeScan` and `IndexScan` nodes to `Filter(SeqScan)`
3674/// so that all downstream fast paths (count, project+limit, sort+limit,
3675/// agg, update, delete) continue to fire.
3676///
3677/// The planner emits `RangeScan` (for `.age > 30`) and `IndexScan` (for
3678/// `.email = lit`) speculatively because it has no catalog access. When
3679/// the column has a B-tree index, those plans are correct. When it
3680/// doesn't, the executor's fallbacks materialise every matching row with
3681/// full `decode_row` — bypassing the compiled-predicate fast paths that
3682/// `Filter(SeqScan)` would trigger. Lowering both speculative leaf kinds
3683/// also keeps EXPLAIN honest: it prints the plan that actually runs.
3684///
3685/// This pass runs once per query, before execution.
3686pub(super) fn lower_unindexed_scans(catalog: &Catalog, plan: &PlanNode) -> PlanNode {
3687 match plan {
3688 PlanNode::RangeScan {
3689 table,
3690 column,
3691 start,
3692 end,
3693 } => {
3694 if let Some(tbl) = catalog.get_table(table) {
3695 // Keep RangeScan whenever ANY index exists on the column:
3696 // unique indexes store raw column values, non-unique indexes
3697 // store composite (value, rid) keys that the executor walks
3698 // natively via BTree::range_rids. Only lower to Filter(SeqScan)
3699 // when the column is unindexed.
3700 if tbl.has_index(column) {
3701 return plan.clone();
3702 }
3703 }
3704 let pred = synthesize_range_predicate(column, start, end);
3705 PlanNode::Filter {
3706 input: Box::new(PlanNode::SeqScan {
3707 table: table.clone(),
3708 }),
3709 predicate: pred,
3710 }
3711 }
3712 PlanNode::Filter { input, predicate } => PlanNode::Filter {
3713 input: Box::new(lower_unindexed_scans(catalog, input)),
3714 predicate: predicate.clone(),
3715 },
3716 PlanNode::Project { input, fields } => PlanNode::Project {
3717 input: Box::new(lower_unindexed_scans(catalog, input)),
3718 fields: fields.clone(),
3719 },
3720 PlanNode::Sort { input, keys } => PlanNode::Sort {
3721 input: Box::new(lower_unindexed_scans(catalog, input)),
3722 keys: keys.clone(),
3723 },
3724 PlanNode::Limit { input, count } => PlanNode::Limit {
3725 input: Box::new(lower_unindexed_scans(catalog, input)),
3726 count: count.clone(),
3727 },
3728 PlanNode::Offset { input, count } => PlanNode::Offset {
3729 input: Box::new(lower_unindexed_scans(catalog, input)),
3730 count: count.clone(),
3731 },
3732 PlanNode::Aggregate {
3733 input,
3734 function,
3735 field,
3736 } => PlanNode::Aggregate {
3737 input: Box::new(lower_unindexed_scans(catalog, input)),
3738 function: *function,
3739 field: field.clone(),
3740 },
3741 PlanNode::Distinct { input } => PlanNode::Distinct {
3742 input: Box::new(lower_unindexed_scans(catalog, input)),
3743 },
3744 PlanNode::GroupBy {
3745 input,
3746 keys,
3747 aggregates,
3748 having,
3749 } => PlanNode::GroupBy {
3750 input: Box::new(lower_unindexed_scans(catalog, input)),
3751 keys: keys.clone(),
3752 aggregates: aggregates.clone(),
3753 having: having.clone(),
3754 },
3755 PlanNode::Update {
3756 input,
3757 table,
3758 assignments,
3759 returning,
3760 } => PlanNode::Update {
3761 input: Box::new(lower_unindexed_scans(catalog, input)),
3762 table: table.clone(),
3763 assignments: assignments.clone(),
3764 returning: *returning,
3765 },
3766 PlanNode::Delete {
3767 input,
3768 table,
3769 returning,
3770 } => PlanNode::Delete {
3771 input: Box::new(lower_unindexed_scans(catalog, input)),
3772 table: table.clone(),
3773 returning: *returning,
3774 },
3775 PlanNode::Window { input, windows } => PlanNode::Window {
3776 input: Box::new(lower_unindexed_scans(catalog, input)),
3777 windows: windows.clone(),
3778 },
3779 PlanNode::Union { left, right, all } => PlanNode::Union {
3780 left: Box::new(lower_unindexed_scans(catalog, left)),
3781 right: Box::new(lower_unindexed_scans(catalog, right)),
3782 all: *all,
3783 },
3784 PlanNode::Explain { input } => PlanNode::Explain {
3785 input: Box::new(lower_unindexed_scans(catalog, input)),
3786 },
3787 PlanNode::NestedLoopJoin {
3788 left,
3789 right,
3790 on,
3791 kind,
3792 } => PlanNode::NestedLoopJoin {
3793 left: Box::new(lower_unindexed_scans(catalog, left)),
3794 right: Box::new(lower_unindexed_scans(catalog, right)),
3795 on: on.clone(),
3796 kind: *kind,
3797 },
3798 PlanNode::IndexScan { table, column, key } => {
3799 if let Some(tbl) = catalog.get_table(table) {
3800 if tbl.has_index(column) {
3801 return plan.clone();
3802 }
3803 }
3804 PlanNode::Filter {
3805 input: Box::new(PlanNode::SeqScan {
3806 table: table.clone(),
3807 }),
3808 predicate: Expr::BinaryOp(
3809 Box::new(Expr::Field(column.clone())),
3810 BinOp::Eq,
3811 Box::new(key.clone()),
3812 ),
3813 }
3814 }
3815 // Leaf nodes: no children to recurse into.
3816 _ => plan.clone(),
3817 }
3818}
3819
3820/// Synthesize a range predicate from RangeScan bounds for the fallback path.
3821pub(super) fn synthesize_range_predicate(
3822 column: &str,
3823 start: &Option<(Expr, bool)>,
3824 end: &Option<(Expr, bool)>,
3825) -> Expr {
3826 let lower = start.as_ref().map(|(expr, inclusive)| {
3827 let op = if *inclusive { BinOp::Gte } else { BinOp::Gt };
3828 Expr::BinaryOp(
3829 Box::new(Expr::Field(column.to_string())),
3830 op,
3831 Box::new(expr.clone()),
3832 )
3833 });
3834 let upper = end.as_ref().map(|(expr, inclusive)| {
3835 let op = if *inclusive { BinOp::Lte } else { BinOp::Lt };
3836 Expr::BinaryOp(
3837 Box::new(Expr::Field(column.to_string())),
3838 op,
3839 Box::new(expr.clone()),
3840 )
3841 });
3842 match (lower, upper) {
3843 (Some(l), Some(u)) => Expr::BinaryOp(Box::new(l), BinOp::And, Box::new(u)),
3844 (Some(l), None) => l,
3845 (None, Some(u)) => u,
3846 (None, None) => Expr::Literal(Literal::Bool(true)),
3847 }
3848}
3849
3850/// Check if a value falls within a range (used in last-resort decoded-row eval).
3851pub(super) fn range_matches(
3852 val: &Value,
3853 start: &Option<Value>,
3854 start_inc: bool,
3855 end: &Option<Value>,
3856 end_inc: bool,
3857) -> bool {
3858 if let Some(ref s) = start {
3859 if start_inc {
3860 if val < s {
3861 return false;
3862 }
3863 } else if val <= s {
3864 return false;
3865 }
3866 }
3867 if let Some(ref e) = end {
3868 if end_inc {
3869 if val > e {
3870 return false;
3871 }
3872 } else if val >= e {
3873 return false;
3874 }
3875 }
3876 true
3877}
3878
3879/// Format a `PlanNode` tree as a human-readable, indented text
3880/// representation. Used by the `EXPLAIN` command.
3881pub(super) fn format_plan_tree(plan: &PlanNode, depth: usize) -> String {
3882 let indent = " ".repeat(depth);
3883 match plan {
3884 PlanNode::SeqScan { table } => format!("{indent}SeqScan table={table}"),
3885 PlanNode::AliasScan { table, alias } => {
3886 format!("{indent}AliasScan table={table} alias={alias}")
3887 }
3888 PlanNode::IndexScan { table, column, key } => {
3889 format!("{indent}IndexScan table={table} column={column} key={key:?}")
3890 }
3891 PlanNode::RangeScan {
3892 table,
3893 column,
3894 start,
3895 end,
3896 } => {
3897 let s = match start {
3898 Some((expr, inc)) => {
3899 let op = if *inc { ">=" } else { ">" };
3900 format!("{op}{expr:?}")
3901 }
3902 None => "unbounded".to_string(),
3903 };
3904 let e = match end {
3905 Some((expr, inc)) => {
3906 let op = if *inc { "<=" } else { "<" };
3907 format!("{op}{expr:?}")
3908 }
3909 None => "unbounded".to_string(),
3910 };
3911 format!("{indent}RangeScan table={table} column={column} [{s}, {e}]")
3912 }
3913 PlanNode::Filter { input, predicate } => {
3914 let child = format_plan_tree(input, depth + 1);
3915 format!("{indent}Filter predicate={predicate:?}\n{child}")
3916 }
3917 PlanNode::Project { input, fields } => {
3918 let names: Vec<String> = fields
3919 .iter()
3920 .map(|f| match &f.alias {
3921 Some(a) => format!("{a}: {:?}", f.expr),
3922 None => format!("{:?}", f.expr),
3923 })
3924 .collect();
3925 let child = format_plan_tree(input, depth + 1);
3926 format!("{indent}Project fields=[{}]\n{child}", names.join(", "))
3927 }
3928 PlanNode::Sort { input, keys } => {
3929 let ks: Vec<String> = keys
3930 .iter()
3931 .map(|k| {
3932 if k.descending {
3933 format!("{} desc", k.field)
3934 } else {
3935 k.field.clone()
3936 }
3937 })
3938 .collect();
3939 let child = format_plan_tree(input, depth + 1);
3940 format!("{indent}Sort keys=[{}]\n{child}", ks.join(", "))
3941 }
3942 PlanNode::Limit { input, count } => {
3943 let child = format_plan_tree(input, depth + 1);
3944 format!("{indent}Limit count={count:?}\n{child}")
3945 }
3946 PlanNode::Offset { input, count } => {
3947 let child = format_plan_tree(input, depth + 1);
3948 format!("{indent}Offset count={count:?}\n{child}")
3949 }
3950 PlanNode::Aggregate {
3951 input,
3952 function,
3953 field,
3954 } => {
3955 let f = field.as_deref().unwrap_or("*");
3956 let child = format_plan_tree(input, depth + 1);
3957 format!("{indent}Aggregate fn={function:?} field={f}\n{child}")
3958 }
3959 PlanNode::NestedLoopJoin {
3960 left,
3961 right,
3962 on,
3963 kind,
3964 } => {
3965 let left_child = format_plan_tree(left, depth + 1);
3966 let right_child = format_plan_tree(right, depth + 1);
3967 let on_str = match on {
3968 Some(pred) => format!("{pred:?}"),
3969 None => "none".to_string(),
3970 };
3971 format!("{indent}NestedLoopJoin kind={kind:?} on={on_str}\n{left_child}\n{right_child}")
3972 }
3973 PlanNode::Distinct { input } => {
3974 let child = format_plan_tree(input, depth + 1);
3975 format!("{indent}Distinct\n{child}")
3976 }
3977 PlanNode::GroupBy {
3978 input,
3979 keys,
3980 aggregates,
3981 having,
3982 } => {
3983 let agg_strs: Vec<String> = aggregates
3984 .iter()
3985 .map(|a| format!("{:?}({}) as {}", a.function, a.field, a.output_name))
3986 .collect();
3987 let having_str = match having {
3988 Some(h) => format!(" having={h:?}"),
3989 None => String::new(),
3990 };
3991 let child = format_plan_tree(input, depth + 1);
3992 format!(
3993 "{indent}GroupBy keys=[{}] aggs=[{}]{having_str}\n{child}",
3994 keys.join(", "),
3995 agg_strs.join(", "),
3996 )
3997 }
3998 PlanNode::Insert { table, rows, .. } => {
3999 let cols: Vec<&str> = rows
4000 .first()
4001 .map(|r| r.iter().map(|a| a.field.as_str()).collect())
4002 .unwrap_or_default();
4003 format!(
4004 "{indent}Insert table={table} rows={} cols=[{}]",
4005 rows.len(),
4006 cols.join(", ")
4007 )
4008 }
4009 PlanNode::Upsert {
4010 table,
4011 key_column,
4012 assignments,
4013 on_conflict,
4014 } => {
4015 let cols: Vec<&str> = assignments.iter().map(|a| a.field.as_str()).collect();
4016 let conflict_cols: Vec<&str> = on_conflict.iter().map(|a| a.field.as_str()).collect();
4017 if conflict_cols.is_empty() {
4018 format!(
4019 "{indent}Upsert table={table} key={key_column} cols=[{}]",
4020 cols.join(", ")
4021 )
4022 } else {
4023 format!(
4024 "{indent}Upsert table={table} key={key_column} cols=[{}] on_conflict=[{}]",
4025 cols.join(", "),
4026 conflict_cols.join(", ")
4027 )
4028 }
4029 }
4030 PlanNode::Update {
4031 input,
4032 table,
4033 assignments,
4034 returning,
4035 } => {
4036 let cols: Vec<&str> = assignments.iter().map(|a| a.field.as_str()).collect();
4037 let child = format_plan_tree(input, depth + 1);
4038 let ret = if *returning { " returning" } else { "" };
4039 format!(
4040 "{indent}Update table={table} set=[{}]{ret}\n{child}",
4041 cols.join(", ")
4042 )
4043 }
4044 PlanNode::Delete {
4045 input,
4046 table,
4047 returning,
4048 } => {
4049 let child = format_plan_tree(input, depth + 1);
4050 let ret = if *returning { " returning" } else { "" };
4051 format!("{indent}Delete table={table}{ret}\n{child}")
4052 }
4053 PlanNode::CreateTable { name, fields } => {
4054 let fs: Vec<String> = fields
4055 .iter()
4056 .map(|f| {
4057 let mut mods = String::new();
4058 if f.required {
4059 mods.push_str(" required");
4060 }
4061 if f.unique {
4062 mods.push_str(" unique");
4063 }
4064 format!("{}: {}{mods}", f.name, f.type_name)
4065 })
4066 .collect();
4067 format!("{indent}CreateTable name={name} fields=[{}]", fs.join(", "))
4068 }
4069 PlanNode::AlterTable { table, action } => {
4070 format!("{indent}AlterTable table={table} action={action:?}")
4071 }
4072 PlanNode::DropTable { name } => format!("{indent}DropTable name={name}"),
4073 PlanNode::CreateView { name, .. } => format!("{indent}CreateView name={name}"),
4074 PlanNode::RefreshView { name } => format!("{indent}RefreshView name={name}"),
4075 PlanNode::DropView { name } => format!("{indent}DropView name={name}"),
4076 PlanNode::Window { input, windows } => {
4077 let ws: Vec<String> = windows
4078 .iter()
4079 .map(|w| format!("{:?} as {}", w.function, w.output_name))
4080 .collect();
4081 let child = format_plan_tree(input, depth + 1);
4082 format!("{indent}Window fns=[{}]\n{child}", ws.join(", "))
4083 }
4084 PlanNode::Union { left, right, all } => {
4085 let kind = if *all { "UNION ALL" } else { "UNION" };
4086 let left_child = format_plan_tree(left, depth + 1);
4087 let right_child = format_plan_tree(right, depth + 1);
4088 format!("{indent}{kind}\n{left_child}\n{right_child}")
4089 }
4090 PlanNode::Explain { input } => {
4091 let child = format_plan_tree(input, depth + 1);
4092 format!("{indent}Explain\n{child}")
4093 }
4094 PlanNode::Begin => format!("{indent}Begin"),
4095 PlanNode::Commit => format!("{indent}Commit"),
4096 PlanNode::Rollback => format!("{indent}Rollback"),
4097 }
4098}