Skip to main content

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(&current_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}