Skip to main content

spg_engine/
plan_cache.rs

1//! v6.3.0 — Engine-level plan cache.
2//!
3//! Caches the post-`prepare()` `Statement` (clock-rewritten,
4//! ORDER-BY-position-resolved, JOIN-reordered) keyed on the raw SQL
5//! string. Hit path skips parse + clock rewrite + JOIN reorder — for
6//! a 5-table JOIN that's the dominant cost.
7//!
8//! `statistics_version` and `source_tables` are stored on the entry
9//! so v6.3.1 can invalidate selectively when ANALYZE bumps the stats
10//! version, or when DDL changes one of the source tables.
11//!
12//! `describe_columns` is reserved for v6.3.3 Describe pre-Execute —
13//! v6.3.0 leaves it empty.
14//!
15//! Cache is bounded by `PLAN_CACHE_MAX_ENTRIES` (256). Eviction is
16//! LRU via a `VecDeque<String>` move-to-back on get. Both `get` and
17//! `insert` are sub-microsecond at 256 entries.
18
19use alloc::collections::{BTreeMap, VecDeque};
20use alloc::string::String;
21use alloc::vec::Vec;
22
23use spg_sql::ast::{Expr, FromClause, FromJoin, SelectItem, SelectStatement, Statement, TableRef};
24use spg_storage::ColumnSchema;
25
26/// Hard cap on plan-cache entries. At 256 the cap holds the typical
27/// app's reusable statement set without unbounded growth; at average
28/// 4 KiB per cached AST the worst-case footprint is 1 MiB per
29/// Engine. NOT a frozen surface — v6.3.x can re-tune.
30pub(crate) const PLAN_CACHE_MAX_ENTRIES: usize = 256;
31
32/// One cached plan. The cached `stmt` is the same one
33/// `Engine::prepare()` would return — parse + clock rewrite +
34/// ORDER-BY position resolution + JOIN reorder all already applied.
35#[derive(Debug, Clone)]
36pub struct PreparedPlan {
37    pub stmt: Statement,
38    /// Statistics version snapshot at prepare time. v6.3.1 compares
39    /// this against the live statistics version and evicts on
40    /// mismatch. v6.3.0 stores it but doesn't consult on lookup.
41    pub statistics_version: u64,
42    /// Tables referenced by `stmt` (deduplicated, lexical order).
43    /// v6.3.1 uses this for selective DDL/ANALYZE invalidation.
44    pub source_tables: Vec<String>,
45    /// Column shape v6.3.3 will populate for `Describe statement`.
46    /// v6.3.0 leaves this empty.
47    pub describe_columns: Vec<ColumnSchema>,
48}
49
50#[derive(Debug, Clone)]
51pub struct PlanCache {
52    /// SQL string → cached plan. `BTreeMap` for deterministic
53    /// iteration (test stability); ordering of LRU is tracked
54    /// separately in `lru`.
55    entries: BTreeMap<String, PreparedPlan>,
56    /// LRU queue. Newest entry at the back. `get` moves the
57    /// referenced key to the back; `insert` pushes to the back and
58    /// evicts the front when at cap.
59    lru: VecDeque<String>,
60    /// v6.5.6 — runtime-configurable cap. Defaults to
61    /// `PLAN_CACHE_MAX_ENTRIES` (256); spg-server reads
62    /// `SPG_PLAN_CACHE_MAX` env at startup and overrides via
63    /// `PlanCache::with_max_entries`.
64    max_entries: usize,
65}
66
67impl Default for PlanCache {
68    fn default() -> Self {
69        Self {
70            entries: BTreeMap::new(),
71            lru: VecDeque::new(),
72            max_entries: PLAN_CACHE_MAX_ENTRIES,
73        }
74    }
75}
76
77impl PlanCache {
78    pub fn new() -> Self {
79        Self::default()
80    }
81
82    /// v6.5.6 — runtime cap override. Operator-tunable via
83    /// `SPG_PLAN_CACHE_MAX` env at startup. Minimum 1; values
84    /// above the compile-time `PLAN_CACHE_MAX_ENTRIES` are
85    /// clamped down to it (defensive backstop against runaway
86    /// configs).
87    pub fn set_max_entries(&mut self, n: usize) {
88        self.max_entries = n.max(1).min(PLAN_CACHE_MAX_ENTRIES);
89    }
90
91    pub fn max_entries(&self) -> usize {
92        self.max_entries
93    }
94
95    pub fn len(&self) -> usize {
96        self.entries.len()
97    }
98
99    pub fn is_empty(&self) -> bool {
100        self.entries.is_empty()
101    }
102
103    /// Read-only peek without LRU promotion. Used by introspection
104    /// and v6.3.1 tests that want to inspect a cached plan's
105    /// metadata without mutating the cache.
106    pub fn get_snapshot(&self, sql: &str) -> Option<&PreparedPlan> {
107        self.entries.get(sql)
108    }
109
110    /// Returns the cached plan if present and promotes it to most-
111    /// recently-used. Returns `None` on miss.
112    pub fn get(&mut self, sql: &str) -> Option<&PreparedPlan> {
113        if !self.entries.contains_key(sql) {
114            return None;
115        }
116        if let Some(idx) = self.lru.iter().position(|k| k == sql) {
117            let key = self.lru.remove(idx).expect("idx came from position()");
118            self.lru.push_back(key);
119        }
120        self.entries.get(sql)
121    }
122
123    /// Inserts (or replaces) the plan. Evicts the oldest entry if
124    /// we'd exceed `PLAN_CACHE_MAX_ENTRIES`.
125    pub fn insert(&mut self, sql: String, plan: PreparedPlan) {
126        if self.entries.contains_key(&sql) {
127            if let Some(idx) = self.lru.iter().position(|k| k == &sql) {
128                let key = self.lru.remove(idx).expect("idx came from position()");
129                self.lru.push_back(key);
130            }
131            self.entries.insert(sql, plan);
132            return;
133        }
134        if self.entries.len() >= self.max_entries {
135            if let Some(oldest) = self.lru.pop_front() {
136                self.entries.remove(&oldest);
137            }
138        }
139        self.lru.push_back(sql.clone());
140        self.entries.insert(sql, plan);
141    }
142
143    pub fn clear(&mut self) {
144        self.entries.clear();
145        self.lru.clear();
146    }
147
148    /// v6.3.1 will use this for explicit invalidation. v6.3.0
149    /// exposes it for tests + future use.
150    pub fn evict(&mut self, sql: &str) -> Option<PreparedPlan> {
151        let plan = self.entries.remove(sql)?;
152        if let Some(idx) = self.lru.iter().position(|k| k == sql) {
153            self.lru.remove(idx);
154        }
155        Some(plan)
156    }
157
158    /// v6.3.1 will use this to evict every plan that references a
159    /// specific table.
160    pub fn evict_referencing(&mut self, table: &str) -> usize {
161        let to_evict: Vec<String> = self
162            .entries
163            .iter()
164            .filter_map(|(k, p)| {
165                if p.source_tables.iter().any(|t| t == table) {
166                    Some(k.clone())
167                } else {
168                    None
169                }
170            })
171            .collect();
172        let n = to_evict.len();
173        for k in to_evict {
174            self.entries.remove(&k);
175            if let Some(idx) = self.lru.iter().position(|x| x == &k) {
176                self.lru.remove(idx);
177            }
178        }
179        n
180    }
181}
182
183/// Walk a `Statement` and collect every distinct table name referenced
184/// by its FROM clauses (including JOIN tables and subquery FROMs).
185/// Used by `PreparedPlan::source_tables` for v6.3.1 selective
186/// invalidation.
187pub fn collect_source_tables(stmt: &Statement) -> Vec<String> {
188    let mut out: Vec<String> = Vec::new();
189    match stmt {
190        Statement::Select(s) => collect_from_select(s, &mut out),
191        Statement::Insert(s) => push_unique(&mut out, &s.table),
192        Statement::Update(s) => {
193            push_unique(&mut out, &s.table);
194            if let Some(w) = &s.where_ {
195                collect_expr(w, &mut out);
196            }
197        }
198        Statement::Delete(s) => {
199            push_unique(&mut out, &s.table);
200            if let Some(w) = &s.where_ {
201                collect_expr(w, &mut out);
202            }
203        }
204        // v7.39 (round 225) — the body is a whole Statement (SELECT or DML).
205        Statement::Explain(inner) => {
206            if let Statement::Select(sel) = &*inner.inner {
207                collect_from_select(sel, &mut out);
208            }
209        }
210        _ => {}
211    }
212    out.sort();
213    out.dedup();
214    out
215}
216
217fn collect_from_select(s: &SelectStatement, out: &mut Vec<String>) {
218    if let Some(from) = &s.from {
219        collect_from_clause(from, out);
220    }
221    if let Some(w) = &s.where_ {
222        collect_expr(w, out);
223    }
224    if let Some(h) = &s.having {
225        collect_expr(h, out);
226    }
227    for item in &s.items {
228        if let SelectItem::Expr { expr, .. } = item {
229            collect_expr(expr, out);
230        }
231    }
232    for (_, peer) in &s.unions {
233        collect_from_select(peer, out);
234    }
235}
236
237fn collect_from_clause(from: &FromClause, out: &mut Vec<String>) {
238    collect_table_ref(&from.primary, out);
239    for j in &from.joins {
240        collect_from_join(j, out);
241    }
242}
243
244fn collect_from_join(j: &FromJoin, out: &mut Vec<String>) {
245    collect_table_ref(&j.table, out);
246    if let Some(on) = &j.on {
247        collect_expr(on, out);
248    }
249}
250
251fn collect_table_ref(t: &TableRef, out: &mut Vec<String>) {
252    push_unique(out, &t.name);
253}
254
255fn collect_expr(e: &Expr, out: &mut Vec<String>) {
256    match e {
257        Expr::NamedArg { expr, .. } => collect_expr(expr, out),
258        Expr::Variadic(expr) => collect_expr(expr, out),
259        Expr::AggregateOrdered {
260            call,
261            order_by,
262            filter,
263            ..
264        } => {
265            collect_expr(call, out);
266            for o in order_by {
267                collect_expr(&o.expr, out);
268            }
269            // A `FILTER (WHERE …)` predicate can carry a subquery that
270            // names a table; track it so writes to that table still
271            // invalidate this cached plan.
272            if let Some(f) = filter {
273                collect_expr(f, out);
274            }
275        }
276        Expr::ScalarSubquery(inner) => collect_from_select(inner, out),
277        Expr::Exists { subquery, .. } => collect_from_select(subquery, out),
278        Expr::InSubquery { expr, subquery, .. } => {
279            collect_expr(expr, out);
280            collect_from_select(subquery, out);
281        }
282        Expr::RowInSubquery { row, subquery, .. } => {
283            for el in row {
284                collect_expr(el, out);
285            }
286            collect_from_select(subquery, out);
287        }
288        Expr::RowCmpSubquery { row, subquery, .. } => {
289            for el in row {
290                collect_expr(el, out);
291            }
292            collect_from_select(subquery, out);
293        }
294        Expr::Binary { lhs, rhs, .. } => {
295            collect_expr(lhs, out);
296            collect_expr(rhs, out);
297        }
298        Expr::Unary { expr, .. } => collect_expr(expr, out),
299        Expr::Cast { expr, .. } | Expr::FieldAccess { base: expr, .. } => collect_expr(expr, out),
300        Expr::IsNull { expr, .. } | Expr::BoolTest { expr, .. } => collect_expr(expr, out),
301        Expr::Like { expr, pattern, .. } => {
302            collect_expr(expr, out);
303            collect_expr(pattern, out);
304        }
305        Expr::FunctionCall { args, .. } => {
306            for a in args {
307                collect_expr(a, out);
308            }
309        }
310        Expr::WindowFunction {
311            args,
312            partition_by,
313            order_by,
314            ..
315        } => {
316            for a in args {
317                collect_expr(a, out);
318            }
319            for p in partition_by {
320                collect_expr(p, out);
321            }
322            for (o, _, _) in order_by {
323                collect_expr(o, out);
324            }
325        }
326        Expr::Extract { source, .. } => collect_expr(source, out),
327        Expr::Array(items) => {
328            for elem in items {
329                collect_expr(elem, out);
330            }
331        }
332        Expr::ArraySubscript { target, index } => {
333            collect_expr(target, out);
334            collect_expr(index, out);
335        }
336        Expr::ArraySlice { target, lo, hi } => {
337            collect_expr(target, out);
338            if let Some(l) = lo {
339                collect_expr(l, out);
340            }
341            if let Some(h) = hi {
342                collect_expr(h, out);
343            }
344        }
345        Expr::AnyAll { expr, array, .. } => {
346            collect_expr(expr, out);
347            collect_expr(array, out);
348        }
349        Expr::InList { expr, list, .. } => {
350            collect_expr(expr, out);
351            for item in list {
352                collect_expr(item, out);
353            }
354        }
355        Expr::Case {
356            operand,
357            branches,
358            else_branch,
359        } => {
360            if let Some(o) = operand {
361                collect_expr(o, out);
362            }
363            for (w, t) in branches {
364                collect_expr(w, out);
365                collect_expr(t, out);
366            }
367            if let Some(e) = else_branch {
368                collect_expr(e, out);
369            }
370        }
371        Expr::Literal(_) | Expr::Column(_) | Expr::Placeholder(_) => {}
372    }
373}
374
375fn push_unique(out: &mut Vec<String>, s: &str) {
376    if !out.iter().any(|x| x == s) {
377        out.push(String::from(s));
378    }
379}
380
381// ── unit tests ────────────────────────────────────────────────────
382
383#[cfg(test)]
384mod tests {
385    use super::*;
386    use alloc::string::ToString;
387    use spg_sql::parser::parse_statement;
388
389    fn dummy_plan(version: u64, tables: &[&str]) -> PreparedPlan {
390        let stmt = parse_statement("SELECT 1").expect("trivial SELECT parses");
391        PreparedPlan {
392            stmt,
393            statistics_version: version,
394            source_tables: tables.iter().map(|s| s.to_string()).collect(),
395            describe_columns: Vec::new(),
396        }
397    }
398
399    #[test]
400    fn new_cache_is_empty() {
401        let cache = PlanCache::new();
402        assert!(cache.is_empty());
403        assert_eq!(cache.len(), 0);
404    }
405
406    #[test]
407    fn insert_then_get_returns_the_plan() {
408        let mut cache = PlanCache::new();
409        cache.insert("SELECT 1".into(), dummy_plan(0, &["t"]));
410        assert_eq!(cache.len(), 1);
411        let plan = cache.get("SELECT 1").expect("hit");
412        assert_eq!(plan.source_tables, alloc::vec!["t".to_string()]);
413    }
414
415    #[test]
416    fn miss_returns_none() {
417        let mut cache = PlanCache::new();
418        cache.insert("SELECT 1".into(), dummy_plan(0, &[]));
419        assert!(cache.get("SELECT 2").is_none());
420    }
421
422    #[test]
423    fn replace_overwrites_existing_entry() {
424        let mut cache = PlanCache::new();
425        cache.insert("SELECT 1".into(), dummy_plan(1, &["a"]));
426        cache.insert("SELECT 1".into(), dummy_plan(2, &["b"]));
427        assert_eq!(cache.len(), 1);
428        let plan = cache.get("SELECT 1").expect("hit");
429        assert_eq!(plan.statistics_version, 2);
430    }
431
432    #[test]
433    fn lru_evicts_oldest_at_cap() {
434        let mut cache = PlanCache::new();
435        for i in 0..PLAN_CACHE_MAX_ENTRIES {
436            cache.insert(alloc::format!("SELECT {i}"), dummy_plan(i as u64, &[]));
437        }
438        assert_eq!(cache.len(), PLAN_CACHE_MAX_ENTRIES);
439        cache.insert("SELECT new".into(), dummy_plan(999, &[]));
440        assert_eq!(cache.len(), PLAN_CACHE_MAX_ENTRIES);
441        assert!(cache.get("SELECT 0").is_none());
442        assert!(cache.get("SELECT new").is_some());
443    }
444
445    #[test]
446    fn get_promotes_lru_position() {
447        let mut cache = PlanCache::new();
448        cache.insert("a".into(), dummy_plan(0, &[]));
449        cache.insert("b".into(), dummy_plan(0, &[]));
450        cache.insert("c".into(), dummy_plan(0, &[]));
451        // Touch "a" to make it MRU.
452        let _ = cache.get("a");
453        // Fill to cap so the next insert evicts. After we touched "a",
454        // "b" should be the oldest now.
455        for i in 0..(PLAN_CACHE_MAX_ENTRIES - 3) {
456            cache.insert(alloc::format!("filler{i}"), dummy_plan(0, &[]));
457        }
458        cache.insert("trigger".into(), dummy_plan(0, &[]));
459        assert!(
460            cache.get("a").is_some(),
461            "a was MRU after get(); should survive"
462        );
463        assert!(cache.get("b").is_none(), "b should be evicted");
464    }
465
466    #[test]
467    fn clear_drops_everything() {
468        let mut cache = PlanCache::new();
469        cache.insert("a".into(), dummy_plan(0, &[]));
470        cache.insert("b".into(), dummy_plan(0, &[]));
471        cache.clear();
472        assert!(cache.is_empty());
473        assert!(cache.get("a").is_none());
474    }
475
476    #[test]
477    fn evict_referencing_drops_only_matching_plans() {
478        let mut cache = PlanCache::new();
479        cache.insert("a".into(), dummy_plan(0, &["users"]));
480        cache.insert("b".into(), dummy_plan(0, &["orders"]));
481        cache.insert("c".into(), dummy_plan(0, &["users", "orders"]));
482        let n = cache.evict_referencing("users");
483        assert_eq!(n, 2);
484        assert!(cache.get("a").is_none());
485        assert!(cache.get("b").is_some());
486        assert!(cache.get("c").is_none());
487    }
488
489    #[test]
490    fn collect_source_tables_from_simple_select() {
491        let stmt = parse_statement("SELECT a, b FROM t1 WHERE x = 1").expect("parses");
492        let tables = collect_source_tables(&stmt);
493        assert_eq!(tables, alloc::vec!["t1".to_string()]);
494    }
495
496    #[test]
497    fn collect_source_tables_from_join() {
498        let stmt =
499            parse_statement("SELECT * FROM t1 JOIN t2 ON t1.a = t2.b JOIN t3 ON t2.c = t3.d")
500                .expect("parses");
501        let tables = collect_source_tables(&stmt);
502        assert_eq!(
503            tables,
504            alloc::vec!["t1".to_string(), "t2".to_string(), "t3".to_string()]
505        );
506    }
507
508    #[test]
509    fn collect_source_tables_dedupes_self_join() {
510        let stmt = parse_statement("SELECT * FROM t1 a JOIN t1 b ON a.x = b.y").expect("parses");
511        let tables = collect_source_tables(&stmt);
512        assert_eq!(tables, alloc::vec!["t1".to_string()]);
513    }
514}