Skip to main content

mongreldb_kit/
query.rs

1//! Query execution for kit statements.
2//!
3//! The implementation is intentionally simple: it materializes every visible row
4//! from the target table(s), evaluates predicates in Rust, then sorts, groups,
5//! joins, and reduces in memory. This keeps the crate independent of MongrelDB
6//! core's native query primitives while remaining correct for the supported
7//! subset.
8//!
9//! ponytail: every operator here is computed in-memory after a full scan; there
10//! is no predicate/projection pushdown into MongrelDB native conditions on the
11//! Rust path. That is the deliberate non-pushdown ceiling — the async SQL engine
12//! path is out of scope for this crate.
13
14use crate::error::{KitError, Result};
15use crate::schema::{core_row_to_json, Row};
16use mongreldb_core::memtable::Row as CoreRow;
17use mongreldb_core::query::Condition;
18use mongreldb_kit_core::query::{
19    AggFunc, Aggregate, AggregateQuery, Direction, Expr, JoinKind, JoinQuery, Literal, OrderBy,
20    Query, Select,
21};
22use mongreldb_kit_core::schema::{Schema as KitSchema, Table as KitTable};
23use serde_json::{Map, Value};
24use std::cmp::Ordering;
25use std::collections::{HashMap, HashSet};
26
27/// A combined join row: a JSON object keyed by table alias whose values are the
28/// matched source rows (or JSON `null` for an unmatched right side of a `LEFT`
29/// join). See [`JoinQuery`] for the documented shape.
30pub type JoinRow = Map<String, Value>;
31
32/// Fetches the visible rows for a real (non-CTE) table at a read snapshot.
33/// When `Some(conditions)` is provided, the fetcher resolves them via native
34/// indexes and returns only matching rows (Kit Priority 1 pushdown).
35type TableFetch<'a> = Box<dyn Fn(&str, Option<&[Condition]>) -> Result<Vec<Row>> + 'a>;
36
37/// Resolves the base rows for a table name and carries materialized CTEs.
38///
39/// CTE materializations shadow real tables, so a later query can read a `with`
40/// result by name. Subqueries and joins evaluate against the same context, which
41/// is how cross-table reads stay consistent at the transaction's read snapshot.
42pub(crate) struct ExecCtx<'a> {
43    schema: Option<&'a KitSchema>,
44    fetch: TableFetch<'a>,
45    ctes: HashMap<String, Vec<Row>>,
46}
47
48impl<'a> ExecCtx<'a> {
49    pub(crate) fn new(
50        schema: Option<&'a KitSchema>,
51        fetch: impl Fn(&str, Option<&[Condition]>) -> Result<Vec<Row>> + 'a,
52    ) -> Self {
53        Self {
54            schema,
55            fetch: Box::new(fetch),
56            ctes: HashMap::new(),
57        }
58    }
59
60    pub(crate) fn add_cte(&mut self, name: String, rows: Vec<Row>) {
61        self.ctes.insert(name, rows);
62    }
63
64    fn table_rows(&self, name: &str) -> Result<Vec<Row>> {
65        if let Some(rows) = self.ctes.get(name) {
66            return Ok(rows.clone());
67        }
68        (self.fetch)(name, None)
69    }
70
71    /// Fetch rows for `name` with native `conditions` pushed to the engine.
72    /// Falls back to an unfiltered fetch for CTEs (which are already
73    /// materialized) or empty conditions.
74    fn table_rows_filtered(&self, name: &str, conditions: &[Condition]) -> Result<Vec<Row>> {
75        if conditions.is_empty() || self.ctes.contains_key(name) {
76            return self.table_rows(name);
77        }
78        (self.fetch)(name, Some(conditions))
79    }
80
81    fn table_def(&self, name: &str) -> Option<&KitTable> {
82        self.schema.and_then(|s| s.table(name))
83    }
84}
85
86/// A name-resolution scope for predicate/value evaluation.
87trait Scope {
88    fn get(&self, name: &str) -> Value;
89}
90
91/// Resolves bare column names against a single flat row.
92struct FlatScope<'a>(&'a Map<String, Value>);
93
94impl Scope for FlatScope<'_> {
95    fn get(&self, name: &str) -> Value {
96        self.0.get(name).cloned().unwrap_or(Value::Null)
97    }
98}
99
100/// Resolves `alias.column` (and, as a fallback, bare column) names against a
101/// combined join row keyed by table alias.
102struct JoinScope<'a>(&'a Map<String, Value>);
103
104impl Scope for JoinScope<'_> {
105    fn get(&self, name: &str) -> Value {
106        if let Some((alias, col)) = name.split_once('.') {
107            return self
108                .0
109                .get(alias)
110                .and_then(|t| t.as_object())
111                .and_then(|o| o.get(col))
112                .cloned()
113                .unwrap_or(Value::Null);
114        }
115        for value in self.0.values() {
116            if let Some(obj) = value.as_object() {
117                if let Some(v) = obj.get(name) {
118                    return v.clone();
119                }
120            }
121        }
122        Value::Null
123    }
124}
125
126// ── public entry points ─────────────────────────────────────────────────────
127
128/// Execute a kit [`Query::Select`] against the supplied visible rows.
129///
130/// `rows` must be the newest visible version of every non-deleted row in the
131/// target table at the transaction's read snapshot. This standalone form cannot
132/// resolve other tables, so subqueries/joins referencing a different table fail.
133pub fn execute_select(
134    table: &KitTable,
135    visible_rows: Vec<CoreRow>,
136    select: &Select,
137) -> Result<Vec<Row>> {
138    let rows: Vec<Row> = visible_rows
139        .into_iter()
140        .map(|r| core_row_to_json(&r, table))
141        .collect::<Result<Vec<_>>>()?;
142    let mut ctx = ExecCtx::new(None, |name: &str, _conds: Option<&[Condition]>| {
143        Err(KitError::Validation(format!(
144            "table {name} is not available outside a transaction context"
145        )))
146    });
147    ctx.add_cte(table.name.clone(), rows);
148    run_select(&ctx, select)
149}
150
151/// Execute any supported kit query statement against visible rows.
152pub fn execute_query(
153    table: &KitTable,
154    visible_rows: Vec<CoreRow>,
155    query: &Query,
156) -> Result<Vec<Row>> {
157    match query {
158        Query::Select(select) => execute_select(table, visible_rows, select),
159        _ => Err(KitError::Validation(
160            "only SELECT queries are supported by execute_query".into(),
161        )),
162    }
163}
164
165// ── select ──────────────────────────────────────────────────────────────────
166
167pub(crate) fn run_select(ctx: &ExecCtx, select: &Select) -> Result<Vec<Row>> {
168    // Kit Priority 1 pushdown: translate the filter into native Conditions so
169    // the engine resolves them via indexes (HOT/bitmap/range) instead of a
170    // full scan. Core conditions return a superset, so the original filter is
171    // re-applied in Rust when the translation was partial.
172    let (mut rows, residual_needed) = match &select.filter {
173        Some(filter) => {
174            let pushed = ctx
175                .table_def(&select.table)
176                .and_then(|t| crate::pushdown::translate_predicate(t, filter));
177            match pushed {
178                Some(plan) if plan.can_push() => {
179                    let fetched = ctx.table_rows_filtered(&select.table, &plan.conditions)?;
180                    (fetched, !plan.fully_translated)
181                }
182                _ => (ctx.table_rows(&select.table)?, true),
183            }
184        }
185        None => (ctx.table_rows(&select.table)?, false),
186    };
187
188    // Residual Rust evaluation for partially-translated predicates.
189    if residual_needed {
190        if let Some(filter) = &select.filter {
191            let mut kept = Vec::with_capacity(rows.len());
192            for r in rows {
193                if eval_pred(filter, &FlatScope(&r.values), ctx)? {
194                    kept.push(r);
195                }
196            }
197            rows = kept;
198        }
199    }
200
201    for order in &select.order_by {
202        sort_rows(ctx, &mut rows, &select.table, order)?;
203    }
204
205    apply_limit_offset(&mut rows, select.limit, select.offset);
206    Ok(rows)
207}
208
209/// Project to `select.columns` (when given) then drop duplicate rows, preserving
210/// first-seen order. Used by `select_distinct`.
211pub(crate) fn project_distinct(select: &Select, rows: Vec<Row>) -> Vec<Row> {
212    let cols: Vec<String> = select
213        .columns
214        .iter()
215        .filter_map(|e| match e {
216            Expr::Column(n) => Some(n.clone()),
217            _ => None,
218        })
219        .collect();
220    let mut seen = HashSet::new();
221    let mut out = Vec::new();
222    for r in rows {
223        let values = if cols.is_empty() {
224            r.values
225        } else {
226            let mut m = Map::new();
227            for c in &cols {
228                m.insert(c.clone(), r.values.get(c).cloned().unwrap_or(Value::Null));
229            }
230            m
231        };
232        let key = serde_json::to_string(&values).unwrap_or_default();
233        if seen.insert(key) {
234            out.push(Row { row_id: 0, values });
235        }
236    }
237    out
238}
239
240// ── aggregates / group by / having ──────────────────────────────────────────
241
242pub(crate) fn run_aggregate(ctx: &ExecCtx, query: &AggregateQuery) -> Result<Vec<Row>> {
243    // Kit Priority 1 pushdown: same strategy as run_select.
244    let (mut rows, residual_needed) = match &query.filter {
245        Some(filter) => {
246            let pushed = ctx
247                .table_def(&query.table)
248                .and_then(|t| crate::pushdown::translate_predicate(t, filter));
249            match pushed {
250                Some(plan) if plan.can_push() => {
251                    let fetched = ctx.table_rows_filtered(&query.table, &plan.conditions)?;
252                    (fetched, !plan.fully_translated)
253                }
254                _ => (ctx.table_rows(&query.table)?, true),
255            }
256        }
257        None => (ctx.table_rows(&query.table)?, false),
258    };
259
260    if residual_needed {
261        if let Some(filter) = &query.filter {
262            let mut kept = Vec::with_capacity(rows.len());
263            for r in rows {
264                if eval_pred(filter, &FlatScope(&r.values), ctx)? {
265                    kept.push(r);
266                }
267            }
268            rows = kept;
269        }
270    }
271
272    // Group rows by the key columns, preserving first-seen group order.
273    let mut groups: Vec<(Vec<Value>, Vec<Row>)> = Vec::new();
274    if query.group_by.is_empty() {
275        groups.push((Vec::new(), rows));
276    } else {
277        let mut index: HashMap<String, usize> = HashMap::new();
278        for r in rows {
279            let key_vals: Vec<Value> = query
280                .group_by
281                .iter()
282                .map(|c| r.values.get(c).cloned().unwrap_or(Value::Null))
283                .collect();
284            let key_str = serde_json::to_string(&key_vals).unwrap_or_default();
285            match index.get(&key_str) {
286                Some(&i) => groups[i].1.push(r),
287                None => {
288                    index.insert(key_str, groups.len());
289                    groups.push((key_vals, vec![r]));
290                }
291            }
292        }
293    }
294
295    let mut out = Vec::with_capacity(groups.len());
296    for (key_vals, group_rows) in groups {
297        let mut values = Map::new();
298        for (col, val) in query.group_by.iter().zip(key_vals.iter()) {
299            values.insert(col.clone(), val.clone());
300        }
301        for agg in &query.aggregates {
302            values.insert(agg.alias.clone(), compute_aggregate(agg, &group_rows)?);
303        }
304        if let Some(having) = &query.having {
305            if !eval_pred(having, &FlatScope(&values), ctx)? {
306                continue;
307            }
308        }
309        out.push(Row { row_id: 0, values });
310    }
311    Ok(out)
312}
313
314fn compute_aggregate(agg: &Aggregate, rows: &[Row]) -> Result<Value> {
315    match agg.func {
316        AggFunc::Count => {
317            let n = match (&agg.column, agg.distinct) {
318                // COUNT(*) — DISTINCT is meaningless without a column.
319                (None, _) => rows.len(),
320                // COUNT(DISTINCT col): unique non-null values.
321                (Some(col), true) => distinct_non_null(rows, col).len(),
322                // COUNT(col): non-null values.
323                (Some(col), false) => rows
324                    .iter()
325                    .filter(|r| !r.values.get(col).map(Value::is_null).unwrap_or(true))
326                    .count(),
327            };
328            Ok(Value::Number((n as i64).into()))
329        }
330        AggFunc::Sum | AggFunc::Avg => {
331            let col = require_agg_column(agg)?;
332            // DISTINCT ⇒ aggregate over the distinct non-null values.
333            let (nums, all_int): (Vec<f64>, bool) = if agg.distinct {
334                let vals = distinct_non_null(rows, col);
335                let all_int = vals.iter().all(|v| v.as_i64().is_some());
336                (vals.iter().filter_map(|&v| num_of(v)).collect(), all_int)
337            } else {
338                let all_int = rows
339                    .iter()
340                    .filter_map(|r| r.values.get(col))
341                    .filter(|v| !v.is_null())
342                    .all(|v| v.as_i64().is_some());
343                (numeric_values(rows, col), all_int)
344            };
345            if nums.is_empty() {
346                return Ok(Value::Null);
347            }
348            let sum: f64 = nums.iter().sum();
349            if matches!(agg.func, AggFunc::Avg) {
350                return Ok(number_value(sum / nums.len() as f64));
351            }
352            // Preserve integer-ness when every summand was an integer.
353            if all_int {
354                Ok(Value::Number((sum as i64).into()))
355            } else {
356                Ok(number_value(sum))
357            }
358        }
359        AggFunc::Min | AggFunc::Max => {
360            let col = require_agg_column(agg)?;
361            let mut best: Option<&Value> = None;
362            for r in rows {
363                let Some(v) = r.values.get(col) else { continue };
364                if v.is_null() {
365                    continue;
366                }
367                best = Some(match best {
368                    None => v,
369                    Some(cur) => {
370                        let take = matches!(
371                            (json_cmp(v, cur), agg.func),
372                            (Some(Ordering::Less), AggFunc::Min)
373                                | (Some(Ordering::Greater), AggFunc::Max)
374                        );
375                        if take {
376                            v
377                        } else {
378                            cur
379                        }
380                    }
381                });
382            }
383            Ok(best.cloned().unwrap_or(Value::Null))
384        }
385    }
386}
387
388fn require_agg_column(agg: &Aggregate) -> Result<&String> {
389    agg.column
390        .as_ref()
391        .ok_or_else(|| KitError::Validation(format!("aggregate {:?} requires a column", agg.func)))
392}
393
394fn numeric_values(rows: &[Row], col: &str) -> Vec<f64> {
395    rows.iter()
396        .filter_map(|r| r.values.get(col))
397        .filter_map(num_of)
398        .collect()
399}
400
401fn num_of(v: &Value) -> Option<f64> {
402    match v {
403        Value::Number(n) => n.as_f64(),
404        _ => None,
405    }
406}
407
408fn number_value(f: f64) -> Value {
409    serde_json::Number::from_f64(f)
410        .map(Value::Number)
411        .unwrap_or(Value::Null)
412}
413
414/// Distinct non-null values of `col` across `rows`, deduplicated by their
415/// canonical JSON form — the same value identity GROUP BY keys use. Backs the
416/// `DISTINCT` aggregates (`COUNT`/`SUM`/`AVG`).
417fn distinct_non_null<'a>(rows: &'a [Row], col: &str) -> Vec<&'a Value> {
418    let mut seen = std::collections::HashSet::new();
419    let mut out = Vec::new();
420    for r in rows {
421        if let Some(v) = r.values.get(col) {
422            if v.is_null() {
423                continue;
424            }
425            if seen.insert(serde_json::to_string(v).unwrap_or_default()) {
426                out.push(v);
427            }
428        }
429    }
430    out
431}
432
433// ── joins ───────────────────────────────────────────────────────────────────
434
435/// When a join `ON` is `right.col = left.col` (column = column) and the right
436/// column has a declared bitmap index, build a `BitmapIn` over the distinct
437/// left-side key values so the right table is fetched by an index probe instead
438/// of a full scan. `eval_pred` still re-checks the predicate for every combined
439/// row, so this only shrinks the candidate set — the result is identical.
440/// Returns `None` (→ full scan) for any non-FK-equality shape, a right column
441/// without a bitmap index, or an empty probe set.
442fn fk_join_condition(
443    right_table: &KitTable,
444    right_alias: &str,
445    on: &Expr,
446    left_rows: &[JoinRow],
447) -> Option<Condition> {
448    let Expr::Eq(a, b) = on else { return None };
449    let (Expr::Column(ca), Expr::Column(cb)) = (a.as_ref(), b.as_ref()) else {
450        return None;
451    };
452    let prefix = format!("{right_alias}.");
453    // Exactly one side must reference the right (joined) alias.
454    let (right_qual, left_qual) = match (ca.starts_with(&prefix), cb.starts_with(&prefix)) {
455        (true, false) => (ca.as_str(), cb.as_str()),
456        (false, true) => (cb.as_str(), ca.as_str()),
457        _ => return None,
458    };
459    let right_col = right_qual.strip_prefix(&prefix)?;
460    if !crate::pushdown::has_declared_bitmap_index(right_table, right_col) {
461        return None;
462    }
463    let col = right_table.column(right_col)?;
464    let (left_alias, left_col) = left_qual.split_once('.')?;
465
466    let mut seen = std::collections::HashSet::new();
467    let mut values = Vec::new();
468    for row in left_rows {
469        let key = row
470            .get(left_alias)
471            .and_then(|t| t.as_object())
472            .and_then(|o| o.get(left_col))
473            .and_then(|v| crate::pushdown::value_index_key(v, col.storage_type));
474        if let Some(key) = key {
475            if seen.insert(key.clone()) {
476                values.push(key);
477            }
478        }
479    }
480    if values.is_empty() {
481        return None;
482    }
483    Some(Condition::BitmapIn {
484        column_id: col.id as u16,
485        values,
486    })
487}
488
489pub(crate) fn run_join(ctx: &ExecCtx, query: &JoinQuery) -> Result<Vec<JoinRow>> {
490    let base_alias = query.alias.clone().unwrap_or_else(|| query.table.clone());
491    let mut acc: Vec<JoinRow> = ctx
492        .table_rows(&query.table)?
493        .into_iter()
494        .map(|r| {
495            let mut m = Map::new();
496            m.insert(base_alias.clone(), Value::Object(r.values));
497            m
498        })
499        .collect();
500
501    for join in &query.joins {
502        let right_alias = join.alias.clone().unwrap_or_else(|| join.table.clone());
503        // For an FK-equality ON with a bitmap-indexed right column, probe the
504        // right table by `BitmapIn` over the accumulated left keys instead of a
505        // full scan; otherwise fall back to the full scan.
506        let right_rows = match (join.kind, join.on.as_ref()) {
507            (JoinKind::Cross, _) | (_, None) => ctx.table_rows(&join.table)?,
508            (_, Some(on)) => match ctx
509                .table_def(&join.table)
510                .and_then(|t| fk_join_condition(t, &right_alias, on, &acc))
511            {
512                Some(cond) => ctx.table_rows_filtered(&join.table, &[cond])?,
513                None => ctx.table_rows(&join.table)?,
514            },
515        };
516        let mut next = Vec::new();
517        for left in acc {
518            let mut matched = false;
519            for rr in &right_rows {
520                let mut combined = left.clone();
521                combined.insert(right_alias.clone(), Value::Object(rr.values.clone()));
522                let keep = match join.kind {
523                    JoinKind::Cross => true,
524                    _ => match &join.on {
525                        Some(on) => eval_pred(on, &JoinScope(&combined), ctx)?,
526                        None => true,
527                    },
528                };
529                if keep {
530                    matched = true;
531                    next.push(combined);
532                }
533            }
534            if !matched && join.kind == JoinKind::Left {
535                let mut combined = left;
536                combined.insert(right_alias.clone(), Value::Null);
537                next.push(combined);
538            }
539        }
540        acc = next;
541    }
542
543    if let Some(filter) = &query.filter {
544        let mut kept = Vec::with_capacity(acc.len());
545        for row in acc {
546            if eval_pred(filter, &JoinScope(&row), ctx)? {
547                kept.push(row);
548            }
549        }
550        acc = kept;
551    }
552
553    for order in &query.order_by {
554        let key = match &order.expr {
555            Expr::Column(n) => n.clone(),
556            other => {
557                return Err(KitError::Validation(format!(
558                    "unsupported order by: {other:?}"
559                )))
560            }
561        };
562        acc.sort_by(|a, b| {
563            let av = JoinScope(a).get(&key);
564            let bv = JoinScope(b).get(&key);
565            let ord = json_cmp(&av, &bv).unwrap_or(Ordering::Equal);
566            match order.direction {
567                Direction::Asc => ord,
568                Direction::Desc => ord.reverse(),
569            }
570        });
571    }
572
573    apply_limit_offset(&mut acc, query.limit, query.offset);
574    Ok(acc)
575}
576
577// ── expression evaluation ───────────────────────────────────────────────────
578
579fn eval_pred<S: Scope>(expr: &Expr, scope: &S, ctx: &ExecCtx) -> Result<bool> {
580    Ok(match expr {
581        Expr::Column(name) => truthy(&scope.get(name)),
582        Expr::Literal(lit) => truthy(&literal_to_value(lit)),
583        Expr::And(parts) => {
584            for p in parts {
585                if !eval_pred(p, scope, ctx)? {
586                    return Ok(false);
587                }
588            }
589            true
590        }
591        Expr::Or(parts) => {
592            for p in parts {
593                if eval_pred(p, scope, ctx)? {
594                    return Ok(true);
595                }
596            }
597            false
598        }
599        Expr::Not(inner) => !eval_pred(inner, scope, ctx)?,
600        Expr::Eq(a, b) => cmp(a, b, scope, ctx)? == Some(Ordering::Equal),
601        Expr::Ne(a, b) => cmp(a, b, scope, ctx)? != Some(Ordering::Equal),
602        Expr::Gt(a, b) => cmp(a, b, scope, ctx)? == Some(Ordering::Greater),
603        Expr::Gte(a, b) => cmp(a, b, scope, ctx)?.is_some_and(|o| o != Ordering::Less),
604        Expr::Lt(a, b) => cmp(a, b, scope, ctx)? == Some(Ordering::Less),
605        Expr::Lte(a, b) => cmp(a, b, scope, ctx)?.is_some_and(|o| o != Ordering::Greater),
606        Expr::In(col, list) => {
607            let v = eval_val(col, scope, ctx)?;
608            list.iter().any(|lit| v == literal_to_value(lit))
609        }
610        Expr::NotIn(col, list) => {
611            let v = eval_val(col, scope, ctx)?;
612            list.iter().all(|lit| v != literal_to_value(lit))
613        }
614        Expr::IsNull(inner) => eval_val(inner, scope, ctx)?.is_null(),
615        Expr::IsNotNull(inner) => !eval_val(inner, scope, ctx)?.is_null(),
616        Expr::Like(col, pattern) => match eval_val(col, scope, ctx)? {
617            Value::String(s) => like(&s, pattern),
618            _ => false,
619        },
620        Expr::Contains(col, needle) => match eval_val(col, scope, ctx)? {
621            Value::String(s) => s.contains(needle.as_str()),
622            _ => false,
623        },
624        Expr::InSubquery(col, sub) => {
625            // ponytail: subqueries are uncorrelated — the sub-SELECT is evaluated
626            // once against the same snapshot/CTEs and cannot reference the outer
627            // row. That covers `id IN (SELECT ...)` and EXISTS-of-a-condition; a
628            // correlated executor is the deferred ceiling.
629            let v = eval_val(col, scope, ctx)?;
630            let key = subquery_column(sub);
631            run_select(ctx, sub)?
632                .iter()
633                .any(|r| subquery_value(r, key.as_deref()) == v)
634        }
635        Expr::Exists(sub) => !run_select(ctx, sub)?.is_empty(),
636        Expr::NotExists(sub) => run_select(ctx, sub)?.is_empty(),
637    })
638}
639
640fn eval_val<S: Scope>(expr: &Expr, scope: &S, _ctx: &ExecCtx) -> Result<Value> {
641    Ok(match expr {
642        Expr::Column(name) => scope.get(name),
643        Expr::Literal(lit) => literal_to_value(lit),
644        other => {
645            return Err(KitError::Validation(format!(
646                "expression {other:?} cannot be used as a scalar value"
647            )))
648        }
649    })
650}
651
652fn cmp<S: Scope>(a: &Expr, b: &Expr, scope: &S, ctx: &ExecCtx) -> Result<Option<Ordering>> {
653    Ok(json_cmp(
654        &eval_val(a, scope, ctx)?,
655        &eval_val(b, scope, ctx)?,
656    ))
657}
658
659/// The column name a subquery exposes for `IN`: the first projected `Column`, or
660/// `None` to fall back to the first value of each result row.
661fn subquery_column(select: &Select) -> Option<String> {
662    select.columns.iter().find_map(|e| match e {
663        Expr::Column(n) => Some(n.clone()),
664        _ => None,
665    })
666}
667
668fn subquery_value(row: &Row, key: Option<&str>) -> Value {
669    match key {
670        Some(k) => row.values.get(k).cloned().unwrap_or(Value::Null),
671        None => row.values.values().next().cloned().unwrap_or(Value::Null),
672    }
673}
674
675fn literal_to_value(lit: &Literal) -> Value {
676    match lit {
677        Literal::Null => Value::Null,
678        Literal::Bool(b) => Value::Bool(*b),
679        Literal::Int(i) => Value::Number((*i).into()),
680        Literal::Float(f) => serde_json::to_value(*f).unwrap_or(Value::Null),
681        Literal::Text(s) => Value::String(s.clone()),
682        Literal::Json(v) => v.clone(),
683    }
684}
685
686fn json_cmp(a: &Value, b: &Value) -> Option<Ordering> {
687    match (a, b) {
688        (Value::Null, Value::Null) => Some(Ordering::Equal),
689        (Value::Null, _) | (_, Value::Null) => None,
690        (Value::Bool(a), Value::Bool(b)) => Some(a.cmp(b)),
691        (Value::Number(a), Value::Number(b)) => {
692            if let (Some(ai), Some(bi)) = (a.as_i64(), b.as_i64()) {
693                Some(ai.cmp(&bi))
694            } else {
695                let af = a.as_f64().unwrap_or(f64::NAN);
696                let bf = b.as_f64().unwrap_or(f64::NAN);
697                af.partial_cmp(&bf)
698            }
699        }
700        (Value::String(a), Value::String(b)) => Some(a.cmp(b)),
701        (Value::Array(a), Value::Array(b)) => compare_arrays(a, b),
702        (Value::Object(a), Value::Object(b)) => compare_objects(a, b),
703        _ => None,
704    }
705}
706
707fn compare_arrays(a: &[Value], b: &[Value]) -> Option<Ordering> {
708    let len_cmp = a.len().partial_cmp(&b.len())?;
709    if len_cmp != Ordering::Equal {
710        return Some(len_cmp);
711    }
712    for (x, y) in a.iter().zip(b.iter()) {
713        match json_cmp(x, y) {
714            Some(Ordering::Equal) => {}
715            other => return other,
716        }
717    }
718    Some(Ordering::Equal)
719}
720
721fn compare_objects(a: &Map<String, Value>, b: &Map<String, Value>) -> Option<Ordering> {
722    let len_cmp = a.len().partial_cmp(&b.len())?;
723    if len_cmp != Ordering::Equal {
724        return Some(len_cmp);
725    }
726    let mut a_keys: Vec<&String> = a.keys().collect();
727    let mut b_keys: Vec<&String> = b.keys().collect();
728    a_keys.sort();
729    b_keys.sort();
730    for (ak, bk) in a_keys.iter().zip(b_keys.iter()) {
731        match ak.cmp(bk) {
732            Ordering::Equal => {}
733            other => return Some(other),
734        }
735        let av = a.get(*ak).unwrap();
736        let bv = b.get(*bk).unwrap();
737        match json_cmp(av, bv) {
738            Some(Ordering::Equal) => {}
739            other => return other,
740        }
741    }
742    Some(Ordering::Equal)
743}
744
745fn truthy(v: &Value) -> bool {
746    match v {
747        Value::Null => false,
748        Value::Bool(b) => *b,
749        Value::Number(n) => n.as_f64().map(|f| f != 0.0).unwrap_or(false),
750        Value::String(s) => !s.is_empty(),
751        Value::Array(a) => !a.is_empty(),
752        Value::Object(o) => !o.is_empty(),
753    }
754}
755
756fn sort_rows(ctx: &ExecCtx, rows: &mut [Row], table: &str, order: &OrderBy) -> Result<()> {
757    let col_name = match &order.expr {
758        Expr::Column(name) => name.clone(),
759        other => {
760            return Err(KitError::Validation(format!(
761                "unsupported order by: {other:?}"
762            )))
763        }
764    };
765    // Validate against the schema when the table is known; CTE/virtual tables are
766    // not in the schema and are sorted leniently (missing values sort as null).
767    if let Some(t) = ctx.table_def(table) {
768        if t.column(&col_name).is_none() {
769            return Err(KitError::Validation(format!(
770                "unknown order column {col_name}"
771            )));
772        }
773    }
774
775    rows.sort_by(|a, b| {
776        let av = a.values.get(&col_name).cloned().unwrap_or(Value::Null);
777        let bv = b.values.get(&col_name).cloned().unwrap_or(Value::Null);
778        let ord = json_cmp(&av, &bv).unwrap_or(Ordering::Equal);
779        match order.direction {
780            Direction::Asc => ord,
781            Direction::Desc => ord.reverse(),
782        }
783    });
784    Ok(())
785}
786
787fn apply_limit_offset<T>(rows: &mut Vec<T>, limit: Option<usize>, offset: Option<usize>) {
788    let offset = offset.unwrap_or(0);
789    if offset > 0 || limit.is_some() {
790        let start = offset.min(rows.len());
791        let end = limit
792            .map(|l| start + l)
793            .unwrap_or(rows.len())
794            .min(rows.len());
795        *rows = rows.drain(start..end).collect();
796    }
797}
798
799fn like(text: &str, pattern: &str) -> bool {
800    let regex = match regex_like(pattern) {
801        Ok(re) => re,
802        Err(_) => return false,
803    };
804    regex.is_match(text)
805}
806
807fn regex_like(pattern: &str) -> Result<regex::Regex> {
808    let mut out = String::with_capacity(pattern.len() * 2);
809    out.push('^');
810    for ch in pattern.chars() {
811        match ch {
812            '%' => out.push_str(".*"),
813            '_' => out.push('.'),
814            c => {
815                if regex::escape(&c.to_string()).len() > 1 {
816                    out.push_str(&regex::escape(&c.to_string()));
817                } else {
818                    out.push(c);
819                }
820            }
821        }
822    }
823    out.push('$');
824    regex::Regex::new(&out).map_err(|e| KitError::Validation(format!("invalid LIKE pattern: {e}")))
825}