ipfrs_storage/
storage_query_planner.rs

1//! Storage Query Planner — optimizes storage access patterns for batch read/write workloads.
2//!
3//! [`StorageQueryPlanner`] analyses queries, selects between sequential and index scans,
4//! applies predicate pushdown, caches plans via FNV-1a fingerprinting, and exposes
5//! detailed explain output and planner statistics.
6
7use std::collections::{HashMap, VecDeque};
8
9// ─────────────────────────────────────────────────────────────────────────────
10// Type aliases
11// ─────────────────────────────────────────────────────────────────────────────
12
13/// Numeric identifier assigned to a registered index.
14pub type SqpIndexId = u32;
15
16/// Convenience alias for [`StorageQueryPlanner`].
17pub type SqpStorageQueryPlanner = StorageQueryPlanner;
18
19// ─────────────────────────────────────────────────────────────────────────────
20// Utility helpers
21// ─────────────────────────────────────────────────────────────────────────────
22
23/// FNV-1a 64-bit hash of arbitrary bytes.
24#[inline]
25fn fnv1a_64(data: &[u8]) -> u64 {
26    let mut h: u64 = 14_695_981_039_346_656_037;
27    for &b in data {
28        h ^= b as u64;
29        h = h.wrapping_mul(1_099_511_628_211);
30    }
31    h
32}
33
34/// Monotonic Unix-epoch timestamp (nanoseconds).
35fn now_ns() -> u64 {
36    use std::time::{SystemTime, UNIX_EPOCH};
37    SystemTime::now()
38        .duration_since(UNIX_EPOCH)
39        .map(|d| d.as_nanos() as u64)
40        .unwrap_or(1)
41}
42
43// ─────────────────────────────────────────────────────────────────────────────
44// SqpValue
45// ─────────────────────────────────────────────────────────────────────────────
46
47/// A scalar value that can appear in a predicate.
48#[derive(Clone, Debug, PartialEq)]
49pub enum SqpValue {
50    Int(i64),
51    Float(f64),
52    Text(String),
53    Bool(bool),
54    Null,
55}
56
57impl SqpValue {
58    /// Returns a stable byte representation suitable for hashing.
59    fn to_hash_bytes(&self) -> Vec<u8> {
60        match self {
61            SqpValue::Int(i) => {
62                let mut v = vec![0u8];
63                v.extend_from_slice(&i.to_le_bytes());
64                v
65            }
66            SqpValue::Float(f) => {
67                let mut v = vec![1u8];
68                v.extend_from_slice(&f.to_bits().to_le_bytes());
69                v
70            }
71            SqpValue::Text(s) => {
72                let mut v = vec![2u8];
73                v.extend_from_slice(s.as_bytes());
74                v
75            }
76            SqpValue::Bool(b) => vec![3u8, if *b { 1 } else { 0 }],
77            SqpValue::Null => vec![4u8],
78        }
79    }
80}
81
82impl std::fmt::Display for SqpValue {
83    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
84        match self {
85            SqpValue::Int(i) => write!(f, "{i}"),
86            SqpValue::Float(v) => write!(f, "{v}"),
87            SqpValue::Text(s) => write!(f, "'{s}'"),
88            SqpValue::Bool(b) => write!(f, "{b}"),
89            SqpValue::Null => write!(f, "NULL"),
90        }
91    }
92}
93
94// ─────────────────────────────────────────────────────────────────────────────
95// SqpOp
96// ─────────────────────────────────────────────────────────────────────────────
97
98/// Comparison / membership operator for a [`SqpPredicate`].
99#[derive(Clone, Debug, PartialEq)]
100pub enum SqpOp {
101    Eq,
102    Ne,
103    Lt,
104    Le,
105    Gt,
106    Ge,
107    In(Vec<SqpValue>),
108    IsNull,
109    IsNotNull,
110}
111
112impl SqpOp {
113    fn tag(&self) -> u8 {
114        match self {
115            SqpOp::Eq => 0,
116            SqpOp::Ne => 1,
117            SqpOp::Lt => 2,
118            SqpOp::Le => 3,
119            SqpOp::Gt => 4,
120            SqpOp::Ge => 5,
121            SqpOp::In(_) => 6,
122            SqpOp::IsNull => 7,
123            SqpOp::IsNotNull => 8,
124        }
125    }
126}
127
128impl std::fmt::Display for SqpOp {
129    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
130        match self {
131            SqpOp::Eq => write!(f, "="),
132            SqpOp::Ne => write!(f, "!="),
133            SqpOp::Lt => write!(f, "<"),
134            SqpOp::Le => write!(f, "<="),
135            SqpOp::Gt => write!(f, ">"),
136            SqpOp::Ge => write!(f, ">="),
137            SqpOp::In(vals) => {
138                write!(f, "IN (")?;
139                for (i, v) in vals.iter().enumerate() {
140                    if i > 0 {
141                        write!(f, ", ")?;
142                    }
143                    write!(f, "{v}")?;
144                }
145                write!(f, ")")
146            }
147            SqpOp::IsNull => write!(f, "IS NULL"),
148            SqpOp::IsNotNull => write!(f, "IS NOT NULL"),
149        }
150    }
151}
152
153// ─────────────────────────────────────────────────────────────────────────────
154// SqpPredicate
155// ─────────────────────────────────────────────────────────────────────────────
156
157/// A single filter condition applied to a field.
158#[derive(Clone, Debug, PartialEq)]
159pub struct SqpPredicate {
160    pub field: String,
161    pub op: SqpOp,
162    pub value: SqpValue,
163}
164
165impl SqpPredicate {
166    pub fn new(field: impl Into<String>, op: SqpOp, value: SqpValue) -> Self {
167        Self {
168            field: field.into(),
169            op,
170            value,
171        }
172    }
173
174    fn hash_bytes(&self) -> Vec<u8> {
175        let mut v = Vec::new();
176        v.extend_from_slice(self.field.as_bytes());
177        v.push(self.op.tag());
178        if let SqpOp::In(vals) = &self.op {
179            for sv in vals {
180                v.extend(sv.to_hash_bytes());
181            }
182        }
183        v.extend(self.value.to_hash_bytes());
184        v
185    }
186}
187
188impl std::fmt::Display for SqpPredicate {
189    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
190        write!(f, "{} {} {}", self.field, self.op, self.value)
191    }
192}
193
194// ─────────────────────────────────────────────────────────────────────────────
195// SqpHint
196// ─────────────────────────────────────────────────────────────────────────────
197
198/// Planner hint supplied by the caller to guide plan selection.
199#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
200pub enum SqpHint {
201    /// Force a sequential (full-table) scan regardless of available indexes.
202    ForceSeqScan,
203    /// Force an index scan if any matching index exists.
204    ForceIndexScan,
205    /// Enable parallel scan execution.
206    Parallel,
207    /// Prefer cache-friendly access order (e.g. sequential prefetch).
208    CacheFriendly,
209}
210
211// ─────────────────────────────────────────────────────────────────────────────
212// SqpCostModel
213// ─────────────────────────────────────────────────────────────────────────────
214
215/// Cost model used to estimate query execution cost.
216#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)]
217pub enum SqpCostModel {
218    /// Cost is proportional to the estimated row count touched.
219    SimpleRowCount,
220    /// Cost includes I/O page reads based on block counts.
221    IOCost,
222    /// Cost based on working-set memory pressure.
223    MemoryCost,
224    /// Weighted combination of row-count, I/O, and memory.
225    #[default]
226    HybridCost,
227}
228
229// ─────────────────────────────────────────────────────────────────────────────
230// SqpPlannerConfig
231// ─────────────────────────────────────────────────────────────────────────────
232
233/// Configuration that drives planner behaviour.
234#[derive(Clone, Debug)]
235pub struct SqpPlannerConfig {
236    /// Maximum number of plans kept in the plan cache.
237    pub max_cache_size: usize,
238    /// Cost model used when estimating plan costs.
239    pub cost_model: SqpCostModel,
240    /// Whether predicates are pushed down into scans before projection.
241    pub enable_predicate_pushdown: bool,
242    /// Number of parallel scan threads to assume when estimating cost.
243    pub parallel_scans: u32,
244    /// Index selectivity threshold below which an index scan is preferred.
245    /// (0.0 = never use index, 1.0 = always use index when available)
246    pub index_selectivity_threshold: f64,
247    /// Row count assumed for a full sequential scan of a table.
248    pub default_table_rows: u64,
249    /// Average bytes per row (used for I/O cost estimation).
250    pub bytes_per_row: u64,
251}
252
253impl Default for SqpPlannerConfig {
254    fn default() -> Self {
255        Self {
256            max_cache_size: 200,
257            cost_model: SqpCostModel::HybridCost,
258            enable_predicate_pushdown: true,
259            parallel_scans: 4,
260            index_selectivity_threshold: 0.15,
261            default_table_rows: 1_000_000,
262            bytes_per_row: 256,
263        }
264    }
265}
266
267// ─────────────────────────────────────────────────────────────────────────────
268// SqpQuery
269// ─────────────────────────────────────────────────────────────────────────────
270
271/// A storage query presented to the planner.
272#[derive(Clone, Debug)]
273pub struct SqpQuery {
274    /// Caller-assigned query identifier.
275    pub id: u64,
276    /// Filter conditions (ANDed together).
277    pub predicates: Vec<SqpPredicate>,
278    /// Column names to project (empty = all columns).
279    pub projections: Vec<String>,
280    /// Maximum rows to return.
281    pub limit: Option<usize>,
282    /// `(field, ascending)` ordering specification.
283    pub order_by: Option<(String, bool)>,
284    /// Optional planner hint overriding automatic decisions.
285    pub hint: Option<SqpHint>,
286}
287
288impl SqpQuery {
289    pub fn new(id: u64) -> Self {
290        Self {
291            id,
292            predicates: Vec::new(),
293            projections: Vec::new(),
294            limit: None,
295            order_by: None,
296            hint: None,
297        }
298    }
299
300    /// Compute a deterministic fingerprint for plan-cache lookups.
301    pub fn fingerprint(&self) -> u64 {
302        let mut buf: Vec<u8> = Vec::new();
303        buf.extend_from_slice(&self.id.to_le_bytes());
304        for p in &self.predicates {
305            buf.extend(p.hash_bytes());
306        }
307        for proj in &self.projections {
308            buf.extend_from_slice(proj.as_bytes());
309            buf.push(b'|');
310        }
311        if let Some(lim) = self.limit {
312            buf.extend_from_slice(&lim.to_le_bytes());
313        }
314        if let Some((ref field, asc)) = self.order_by {
315            buf.extend_from_slice(field.as_bytes());
316            buf.push(if asc { 1 } else { 0 });
317        }
318        if let Some(hint) = self.hint {
319            buf.push(hint as u8);
320        }
321        fnv1a_64(&buf)
322    }
323}
324
325// ─────────────────────────────────────────────────────────────────────────────
326// SqpPlanStep
327// ─────────────────────────────────────────────────────────────────────────────
328
329/// A single step in a [`SqpQueryPlan`].
330#[derive(Clone, Debug)]
331pub enum SqpPlanStep {
332    /// Full sequential scan of the given table.
333    SeqScan {
334        table: String,
335        /// Number of predicate filters applied during the scan.
336        filter_count: usize,
337    },
338    /// Index-based point or range lookup.
339    IndexScan {
340        index_id: SqpIndexId,
341        /// Fraction of rows the index is expected to match (0.0–1.0).
342        selectivity: f64,
343    },
344    /// Post-scan predicate filter.
345    Filter(SqpPredicate),
346    /// Sort the result stream.
347    Sort { field: String, asc: bool },
348    /// Truncate the result to at most N rows.
349    Limit(usize),
350    /// Project (select) specific columns.
351    Project(Vec<String>),
352}
353
354impl std::fmt::Display for SqpPlanStep {
355    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
356        match self {
357            SqpPlanStep::SeqScan {
358                table,
359                filter_count,
360            } => {
361                write!(f, "SeqScan(table={table}, filters={filter_count})")
362            }
363            SqpPlanStep::IndexScan {
364                index_id,
365                selectivity,
366            } => {
367                write!(f, "IndexScan(id={index_id}, selectivity={selectivity:.4})")
368            }
369            SqpPlanStep::Filter(pred) => write!(f, "Filter({pred})"),
370            SqpPlanStep::Sort { field, asc } => {
371                write!(f, "Sort({field} {})", if *asc { "ASC" } else { "DESC" })
372            }
373            SqpPlanStep::Limit(n) => write!(f, "Limit({n})"),
374            SqpPlanStep::Project(cols) => write!(f, "Project([{}])", cols.join(", ")),
375        }
376    }
377}
378
379// ─────────────────────────────────────────────────────────────────────────────
380// SqpQueryPlan
381// ─────────────────────────────────────────────────────────────────────────────
382
383/// A fully resolved execution plan for a [`SqpQuery`].
384#[derive(Clone, Debug)]
385pub struct SqpQueryPlan {
386    /// Identifier of the originating query.
387    pub query_id: u64,
388    /// Ordered list of execution steps.
389    pub steps: Vec<SqpPlanStep>,
390    /// Estimated cost (model-specific unit).
391    pub estimated_cost: f64,
392    /// Estimated number of rows returned after all filters.
393    pub estimated_rows: u64,
394    /// Whether at least one index scan is present in the plan.
395    pub uses_index: bool,
396    /// Timestamp when the plan was generated (nanoseconds).
397    pub planned_at_ns: u64,
398}
399
400impl SqpQueryPlan {
401    fn new(query_id: u64) -> Self {
402        Self {
403            query_id,
404            steps: Vec::new(),
405            estimated_cost: 0.0,
406            estimated_rows: 0,
407            uses_index: false,
408            planned_at_ns: now_ns(),
409        }
410    }
411}
412
413// ─────────────────────────────────────────────────────────────────────────────
414// SqpIndexStat
415// ─────────────────────────────────────────────────────────────────────────────
416
417/// Runtime statistics about a registered index.
418#[derive(Clone, Debug)]
419pub struct SqpIndexStat {
420    /// Human-readable name of the index.
421    pub name: String,
422    /// Number of distinct values in the indexed column.
423    pub cardinality: u64,
424    /// Fraction of rows matched by a typical equality predicate (0.0–1.0).
425    pub selectivity: f64,
426    /// How many times this index has been used in a chosen plan.
427    pub usage_count: u64,
428    /// Timestamp of registration (nanoseconds).
429    pub registered_at_ns: u64,
430    /// Which query fingerprints reference this index.
431    pub referencing_queries: Vec<u64>,
432}
433
434impl SqpIndexStat {
435    fn new(name: String, cardinality: u64, selectivity: f64) -> Self {
436        Self {
437            name,
438            cardinality,
439            selectivity: selectivity.clamp(0.0, 1.0),
440            usage_count: 0,
441            registered_at_ns: now_ns(),
442            referencing_queries: Vec::new(),
443        }
444    }
445}
446
447// ─────────────────────────────────────────────────────────────────────────────
448// SqpQueryRecord
449// ─────────────────────────────────────────────────────────────────────────────
450
451/// A historical record of a query planning event.
452#[derive(Clone, Debug)]
453pub struct SqpQueryRecord {
454    /// Fingerprint of the query.
455    pub fingerprint: u64,
456    /// Whether the plan was served from cache.
457    pub cache_hit: bool,
458    /// Estimated cost of the chosen plan.
459    pub estimated_cost: f64,
460    /// Whether the chosen plan uses an index.
461    pub uses_index: bool,
462    /// Timestamp of the planning event (nanoseconds).
463    pub planned_at_ns: u64,
464}
465
466// ─────────────────────────────────────────────────────────────────────────────
467// SqpPlannerStats
468// ─────────────────────────────────────────────────────────────────────────────
469
470/// Aggregate statistics for the planner since creation.
471#[derive(Clone, Debug)]
472pub struct SqpPlannerStats {
473    /// Total number of plans generated (including cache hits).
474    pub total_plans: u64,
475    /// Fraction of planning calls that were served from cache (0.0–1.0).
476    pub cache_hit_rate: f64,
477    /// Average estimated cost across all plans.
478    pub avg_cost: f64,
479    /// Fraction of plans that use at least one index (0.0–1.0).
480    pub index_usage_rate: f64,
481    /// Number of entries currently in the plan cache.
482    pub cache_size: usize,
483    /// Number of registered indexes.
484    pub registered_indexes: usize,
485}
486
487// ─────────────────────────────────────────────────────────────────────────────
488// StorageQueryPlanner
489// ─────────────────────────────────────────────────────────────────────────────
490
491/// Production-grade query planner that optimizes storage access patterns.
492///
493/// # Examples
494/// ```
495/// use ipfrs_storage::storage_query_planner::{
496///     StorageQueryPlanner, SqpPlannerConfig, SqpQuery, SqpPredicate, SqpOp, SqpValue,
497/// };
498///
499/// let planner = StorageQueryPlanner::new(SqpPlannerConfig::default());
500/// let mut query = SqpQuery::new(1);
501/// query.predicates.push(SqpPredicate::new("age", SqpOp::Gt, SqpValue::Int(18)));
502/// let plan = planner.plan(&query);
503/// assert!(plan.estimated_rows > 0);
504/// ```
505pub struct StorageQueryPlanner {
506    config: SqpPlannerConfig,
507    /// Bounded ring-buffer of planning history (max 500).
508    history: parking_lot::Mutex<VecDeque<SqpQueryRecord>>,
509    /// Registered index metadata keyed by [`SqpIndexId`].
510    index_stats: parking_lot::RwLock<HashMap<SqpIndexId, SqpIndexStat>>,
511    /// Plan cache: FNV-1a fingerprint → cached plan (max `config.max_cache_size`).
512    plan_cache: parking_lot::Mutex<HashMap<u64, SqpQueryPlan>>,
513    /// Monotonically increasing counter for assigning index IDs.
514    next_index_id: std::sync::atomic::AtomicU32,
515    /// Cumulative totals for stats reporting.
516    total_plans: std::sync::atomic::AtomicU64,
517    total_cache_hits: std::sync::atomic::AtomicU64,
518    total_index_plans: std::sync::atomic::AtomicU64,
519    /// Sum of estimated costs, used for avg_cost computation.
520    total_cost_bits: std::sync::atomic::AtomicU64,
521}
522
523impl Default for StorageQueryPlanner {
524    fn default() -> Self {
525        Self::new(SqpPlannerConfig::default())
526    }
527}
528
529impl StorageQueryPlanner {
530    /// Create a new planner with the given configuration.
531    pub fn new(config: SqpPlannerConfig) -> Self {
532        Self {
533            config,
534            history: parking_lot::Mutex::new(VecDeque::with_capacity(500)),
535            index_stats: parking_lot::RwLock::new(HashMap::new()),
536            plan_cache: parking_lot::Mutex::new(HashMap::new()),
537            next_index_id: std::sync::atomic::AtomicU32::new(1),
538            total_plans: std::sync::atomic::AtomicU64::new(0),
539            total_cache_hits: std::sync::atomic::AtomicU64::new(0),
540            total_index_plans: std::sync::atomic::AtomicU64::new(0),
541            total_cost_bits: std::sync::atomic::AtomicU64::new(0),
542        }
543    }
544
545    // ─── Index registration ────────────────────────────────────────────────
546
547    /// Register a new index and return its assigned [`SqpIndexId`].
548    ///
549    /// If an index with the same name already exists its statistics are updated
550    /// in place and the existing ID is returned.
551    pub fn register_index(
552        &self,
553        name: impl Into<String>,
554        cardinality: u64,
555        selectivity: f64,
556    ) -> SqpIndexId {
557        let name = name.into();
558        let mut stats = self.index_stats.write();
559
560        // Check for existing index with this name.
561        if let Some((&id, existing)) = stats.iter_mut().find(|(_, s)| s.name == name) {
562            existing.cardinality = cardinality;
563            existing.selectivity = selectivity.clamp(0.0, 1.0);
564            return id;
565        }
566
567        let id = self
568            .next_index_id
569            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
570        stats.insert(id, SqpIndexStat::new(name, cardinality, selectivity));
571        id
572    }
573
574    /// Look up the stat record for a registered index.
575    pub fn index_stat(&self, id: SqpIndexId) -> Option<SqpIndexStat> {
576        self.index_stats.read().get(&id).cloned()
577    }
578
579    // ─── Planning ──────────────────────────────────────────────────────────
580
581    /// Generate an optimised plan for `query` (no caching).
582    pub fn plan(&self, query: &SqpQuery) -> SqpQueryPlan {
583        let plan = self.build_plan(query);
584        self.record_planning(
585            query.fingerprint(),
586            false,
587            plan.estimated_cost,
588            plan.uses_index,
589        );
590        plan
591    }
592
593    /// Generate or retrieve a cached plan for `query`.
594    pub fn plan_cached(&self, query: &SqpQuery) -> SqpQueryPlan {
595        let fp = query.fingerprint();
596
597        // Cache lookup.
598        {
599            let cache = self.plan_cache.lock();
600            if let Some(cached) = cache.get(&fp) {
601                let plan = cached.clone();
602                drop(cache);
603                self.record_planning(fp, true, plan.estimated_cost, plan.uses_index);
604                return plan;
605            }
606        }
607
608        // Cache miss — build and store.
609        let plan = self.build_plan(query);
610        {
611            let mut cache = self.plan_cache.lock();
612            if cache.len() >= self.config.max_cache_size {
613                // Evict the oldest entry (first inserted, hash-map order is arbitrary;
614                // we remove an arbitrary key for bounded-size enforcement).
615                if let Some(evict_key) = cache.keys().copied().next() {
616                    cache.remove(&evict_key);
617                }
618            }
619            cache.insert(fp, plan.clone());
620        }
621        self.record_planning(fp, false, plan.estimated_cost, plan.uses_index);
622        plan
623    }
624
625    /// Invalidate the entire plan cache.
626    pub fn invalidate_cache(&self) {
627        self.plan_cache.lock().clear();
628    }
629
630    /// Remove all cached plans that reference the given index.
631    pub fn invalidate_for_index(&self, id: SqpIndexId) {
632        let referencing: Vec<u64> = {
633            let stats = self.index_stats.read();
634            stats
635                .get(&id)
636                .map(|s| s.referencing_queries.clone())
637                .unwrap_or_default()
638        };
639        let mut cache = self.plan_cache.lock();
640        for fp in referencing {
641            cache.remove(&fp);
642        }
643        // Also clear the referencing list on the index stat.
644        if let Some(stat) = self.index_stats.write().get_mut(&id) {
645            stat.referencing_queries.clear();
646        }
647    }
648
649    // ─── Cost estimation ──────────────────────────────────────────────────
650
651    /// Re-estimate the cost of an already-built plan.
652    ///
653    /// This applies the configured cost model independently of plan generation,
654    /// making it useful for comparing alternative plans.
655    pub fn estimate_cost(&self, plan: &SqpQueryPlan) -> f64 {
656        self.compute_cost(plan.estimated_rows, plan.uses_index, plan.steps.len())
657    }
658
659    // ─── Explain output ───────────────────────────────────────────────────
660
661    /// Produce a human-readable description of a plan.
662    pub fn explain(&self, plan: &SqpQueryPlan) -> String {
663        let mut out = String::new();
664        out.push_str(&format!(
665            "QueryPlan(id={}, cost={:.2}, rows={}, uses_index={})\n",
666            plan.query_id, plan.estimated_cost, plan.estimated_rows, plan.uses_index
667        ));
668        for (i, step) in plan.steps.iter().enumerate() {
669            out.push_str(&format!("  [{i:02}] {step}\n"));
670        }
671        // Append index details when relevant.
672        if plan.uses_index {
673            let stats = self.index_stats.read();
674            for step in &plan.steps {
675                if let SqpPlanStep::IndexScan {
676                    index_id,
677                    selectivity,
678                } = step
679                {
680                    if let Some(stat) = stats.get(index_id) {
681                        out.push_str(&format!(
682                            "       -> index '{}': cardinality={}, selectivity={:.4}, usages={}\n",
683                            stat.name, stat.cardinality, selectivity, stat.usage_count
684                        ));
685                    }
686                }
687            }
688        }
689        out
690    }
691
692    // ─── Statistics ───────────────────────────────────────────────────────
693
694    /// Return aggregate planner statistics.
695    pub fn planner_stats(&self) -> SqpPlannerStats {
696        use std::sync::atomic::Ordering::Relaxed;
697        let total = self.total_plans.load(Relaxed);
698        let hits = self.total_cache_hits.load(Relaxed);
699        let index_plans = self.total_index_plans.load(Relaxed);
700        let cost_bits = self.total_cost_bits.load(Relaxed);
701
702        let cache_hit_rate = if total == 0 {
703            0.0
704        } else {
705            hits as f64 / total as f64
706        };
707        let avg_cost = if total == 0 {
708            0.0
709        } else {
710            f64::from_bits(cost_bits) / total as f64
711        };
712        let index_usage_rate = if total == 0 {
713            0.0
714        } else {
715            index_plans as f64 / total as f64
716        };
717
718        SqpPlannerStats {
719            total_plans: total,
720            cache_hit_rate,
721            avg_cost,
722            index_usage_rate,
723            cache_size: self.plan_cache.lock().len(),
724            registered_indexes: self.index_stats.read().len(),
725        }
726    }
727
728    /// Return the most recent N planning records from history.
729    pub fn recent_history(&self, n: usize) -> Vec<SqpQueryRecord> {
730        let hist = self.history.lock();
731        hist.iter().rev().take(n).cloned().collect()
732    }
733
734    // ─── Private helpers ──────────────────────────────────────────────────
735
736    /// Core plan-construction logic.
737    fn build_plan(&self, query: &SqpQuery) -> SqpQueryPlan {
738        let mut plan = SqpQueryPlan::new(query.id);
739        let fp = query.fingerprint();
740
741        // Decide scan type.
742        let (scan_step, uses_index, chosen_index_id) = self.choose_scan(query, fp);
743
744        // If using an index, bump its usage counter and record the fingerprint.
745        if let Some(iid) = chosen_index_id {
746            if let Some(stat) = self.index_stats.write().get_mut(&iid) {
747                stat.usage_count += 1;
748                if !stat.referencing_queries.contains(&fp) {
749                    stat.referencing_queries.push(fp);
750                }
751            }
752        }
753
754        plan.uses_index = uses_index;
755        plan.steps.push(scan_step);
756
757        // Predicate pushdown (remaining predicates not absorbed by index scan).
758        if self.config.enable_predicate_pushdown && query.hint != Some(SqpHint::ForceSeqScan) {
759            for pred in &query.predicates {
760                // Skip predicates that were already handled by the index scan.
761                if uses_index {
762                    // Only the first predicate was used in the index scan.
763                    // Skip it here; the remaining ones become Filter steps.
764                    if query.predicates.first().map(|p| p == pred) == Some(true) {
765                        continue;
766                    }
767                }
768                plan.steps.push(SqpPlanStep::Filter(pred.clone()));
769            }
770        } else {
771            // No pushdown: emit explicit Filter steps for all predicates.
772            for pred in &query.predicates {
773                plan.steps.push(SqpPlanStep::Filter(pred.clone()));
774            }
775        }
776
777        // Order-by.
778        if let Some((ref field, asc)) = query.order_by {
779            plan.steps.push(SqpPlanStep::Sort {
780                field: field.clone(),
781                asc,
782            });
783        }
784
785        // Limit.
786        if let Some(lim) = query.limit {
787            plan.steps.push(SqpPlanStep::Limit(lim));
788        }
789
790        // Projection (only if non-empty).
791        if !query.projections.is_empty() {
792            plan.steps
793                .push(SqpPlanStep::Project(query.projections.clone()));
794        }
795
796        // Estimate rows and cost.
797        plan.estimated_rows = self.estimate_rows(query, uses_index, chosen_index_id);
798        plan.estimated_cost = self.compute_cost(plan.estimated_rows, uses_index, plan.steps.len());
799
800        plan
801    }
802
803    /// Choose between a sequential scan and an index scan.
804    ///
805    /// Returns `(step, uses_index, Option<index_id>)`.
806    fn choose_scan(&self, query: &SqpQuery, _fp: u64) -> (SqpPlanStep, bool, Option<SqpIndexId>) {
807        // Honour explicit hints first.
808        match query.hint {
809            Some(SqpHint::ForceSeqScan) => {
810                return (
811                    SqpPlanStep::SeqScan {
812                        table: "default".to_string(),
813                        filter_count: query.predicates.len(),
814                    },
815                    false,
816                    None,
817                );
818            }
819            Some(SqpHint::ForceIndexScan) => {
820                if let Some((id, stat)) = self.best_index_for_query(query) {
821                    return (
822                        SqpPlanStep::IndexScan {
823                            index_id: id,
824                            selectivity: stat.selectivity,
825                        },
826                        true,
827                        Some(id),
828                    );
829                }
830                // Fall through to automatic selection if no index is available.
831            }
832            _ => {}
833        }
834
835        // Automatic selection: use index if selectivity is below threshold.
836        if let Some((id, stat)) = self.best_index_for_query(query) {
837            if stat.selectivity <= self.config.index_selectivity_threshold {
838                return (
839                    SqpPlanStep::IndexScan {
840                        index_id: id,
841                        selectivity: stat.selectivity,
842                    },
843                    true,
844                    Some(id),
845                );
846            }
847        }
848
849        // Fall back to sequential scan.
850        (
851            SqpPlanStep::SeqScan {
852                table: "default".to_string(),
853                filter_count: query.predicates.len(),
854            },
855            false,
856            None,
857        )
858    }
859
860    /// Find the most selective index that matches any equality predicate.
861    fn best_index_for_query(&self, query: &SqpQuery) -> Option<(SqpIndexId, SqpIndexStat)> {
862        let stats = self.index_stats.read();
863        if stats.is_empty() || query.predicates.is_empty() {
864            return None;
865        }
866
867        // Prefer indexes whose name matches a predicate field.
868        let mut best: Option<(SqpIndexId, SqpIndexStat)> = None;
869        for pred in &query.predicates {
870            for (&id, stat) in stats.iter() {
871                if stat.name == pred.field {
872                    match best {
873                        None => best = Some((id, stat.clone())),
874                        Some((_, ref b)) if stat.selectivity < b.selectivity => {
875                            best = Some((id, stat.clone()))
876                        }
877                        _ => {}
878                    }
879                }
880            }
881        }
882
883        // If no field match, return the globally most selective index.
884        if best.is_none() {
885            best = stats
886                .iter()
887                .min_by(|(_, a), (_, b)| {
888                    a.selectivity
889                        .partial_cmp(&b.selectivity)
890                        .unwrap_or(std::cmp::Ordering::Equal)
891                })
892                .map(|(&id, stat)| (id, stat.clone()));
893        }
894
895        best
896    }
897
898    /// Estimate the number of rows returned after all filters.
899    fn estimate_rows(
900        &self,
901        query: &SqpQuery,
902        uses_index: bool,
903        index_id: Option<SqpIndexId>,
904    ) -> u64 {
905        let base = self.config.default_table_rows;
906
907        let selectivity: f64 = if uses_index {
908            if let Some(id) = index_id {
909                let stats = self.index_stats.read();
910                stats.get(&id).map(|s| s.selectivity).unwrap_or(0.5)
911            } else {
912                0.5
913            }
914        } else {
915            // Estimate per-predicate selectivity.
916            let per_pred = query.predicates.iter().fold(1.0f64, |acc, pred| {
917                acc * Self::predicate_selectivity_estimate(&pred.op)
918            });
919            per_pred.clamp(0.0, 1.0)
920        };
921
922        let mut rows = ((base as f64) * selectivity).ceil() as u64;
923        rows = rows.max(1);
924
925        // Apply limit.
926        if let Some(lim) = query.limit {
927            rows = rows.min(lim as u64);
928        }
929        rows
930    }
931
932    /// Heuristic per-operator selectivity.
933    fn predicate_selectivity_estimate(op: &SqpOp) -> f64 {
934        match op {
935            SqpOp::Eq => 0.05,
936            SqpOp::Ne => 0.95,
937            SqpOp::Lt | SqpOp::Le => 0.33,
938            SqpOp::Gt | SqpOp::Ge => 0.33,
939            SqpOp::In(vals) => {
940                // Each IN value contributes ~5 % selectivity, capped at 50 %.
941                (vals.len() as f64 * 0.05).min(0.50)
942            }
943            SqpOp::IsNull => 0.01,
944            SqpOp::IsNotNull => 0.99,
945        }
946    }
947
948    /// Compute estimated cost according to the configured cost model.
949    fn compute_cost(&self, rows: u64, uses_index: bool, step_count: usize) -> f64 {
950        let rows_f = rows as f64;
951        match self.config.cost_model {
952            SqpCostModel::SimpleRowCount => rows_f,
953            SqpCostModel::IOCost => {
954                let pages = (rows_f * self.config.bytes_per_row as f64 / 8192.0).ceil();
955                if uses_index {
956                    // Index scan: log(pages) seek + pages/selectivity read.
957                    pages.ln().max(1.0) + pages * 1.5
958                } else {
959                    // Full scan.
960                    let total_pages = (self.config.default_table_rows as f64
961                        * self.config.bytes_per_row as f64
962                        / 8192.0)
963                        .ceil();
964                    total_pages + step_count as f64 * 0.1
965                }
966            }
967            SqpCostModel::MemoryCost => {
968                let working_set_mb =
969                    (rows_f * self.config.bytes_per_row as f64) / (1024.0 * 1024.0);
970                working_set_mb * if uses_index { 1.0 } else { 2.5 }
971            }
972            SqpCostModel::HybridCost => {
973                let io = {
974                    let pages = (rows_f * self.config.bytes_per_row as f64 / 8192.0).ceil();
975                    if uses_index {
976                        pages.ln().max(1.0) + pages * 1.5
977                    } else {
978                        let total_pages = (self.config.default_table_rows as f64
979                            * self.config.bytes_per_row as f64
980                            / 8192.0)
981                            .ceil();
982                        total_pages + step_count as f64 * 0.1
983                    }
984                };
985                let mem = {
986                    let wsm = (rows_f * self.config.bytes_per_row as f64) / (1024.0 * 1024.0);
987                    wsm * if uses_index { 1.0 } else { 2.5 }
988                };
989                // Weighted blend: 60 % I/O, 30 % row count, 10 % memory.
990                0.6 * io + 0.3 * rows_f + 0.1 * mem
991            }
992        }
993    }
994
995    /// Record a planning event into the history ring-buffer and update counters.
996    fn record_planning(&self, fp: u64, cache_hit: bool, cost: f64, uses_index: bool) {
997        use std::sync::atomic::Ordering::Relaxed;
998
999        self.total_plans.fetch_add(1, Relaxed);
1000        if cache_hit {
1001            self.total_cache_hits.fetch_add(1, Relaxed);
1002        }
1003        if uses_index {
1004            self.total_index_plans.fetch_add(1, Relaxed);
1005        }
1006
1007        // Accumulate cost as f64 bits (single-threaded addition via CAS would be
1008        // racy but acceptable for statistics; we use a simple fetch_add on bits).
1009        let prev_bits = self.total_cost_bits.load(Relaxed);
1010        let prev_cost = f64::from_bits(prev_bits);
1011        let new_cost_bits = (prev_cost + cost).to_bits();
1012        // Best-effort CAS loop (stats do not need strict consistency).
1013        let _ = self
1014            .total_cost_bits
1015            .compare_exchange(prev_bits, new_cost_bits, Relaxed, Relaxed);
1016
1017        let record = SqpQueryRecord {
1018            fingerprint: fp,
1019            cache_hit,
1020            estimated_cost: cost,
1021            uses_index,
1022            planned_at_ns: now_ns(),
1023        };
1024
1025        let mut hist = self.history.lock();
1026        if hist.len() >= 500 {
1027            hist.pop_front();
1028        }
1029        hist.push_back(record);
1030    }
1031}
1032
1033// ─────────────────────────────────────────────────────────────────────────────
1034// Tests
1035// ─────────────────────────────────────────────────────────────────────────────
1036
1037#[cfg(test)]
1038mod tests {
1039    use super::*;
1040
1041    // ── Helpers ────────────────────────────────────────────────────────────
1042
1043    fn make_planner() -> StorageQueryPlanner {
1044        StorageQueryPlanner::new(SqpPlannerConfig::default())
1045    }
1046
1047    fn simple_query(id: u64) -> SqpQuery {
1048        SqpQuery::new(id)
1049    }
1050
1051    fn predicate_eq(field: &str, v: i64) -> SqpPredicate {
1052        SqpPredicate::new(field, SqpOp::Eq, SqpValue::Int(v))
1053    }
1054
1055    // ── fnv1a helper ───────────────────────────────────────────────────────
1056
1057    #[test]
1058    fn test_fnv1a_empty() {
1059        assert_eq!(fnv1a_64(&[]), 14_695_981_039_346_656_037u64);
1060    }
1061
1062    #[test]
1063    fn test_fnv1a_deterministic() {
1064        assert_eq!(fnv1a_64(b"hello"), fnv1a_64(b"hello"));
1065    }
1066
1067    #[test]
1068    fn test_fnv1a_different_inputs() {
1069        assert_ne!(fnv1a_64(b"foo"), fnv1a_64(b"bar"));
1070    }
1071
1072    // ── SqpValue ───────────────────────────────────────────────────────────
1073
1074    #[test]
1075    fn test_sqpvalue_display_int() {
1076        assert_eq!(SqpValue::Int(42).to_string(), "42");
1077    }
1078
1079    #[test]
1080    fn test_sqpvalue_display_float() {
1081        let s = SqpValue::Float(2.71).to_string();
1082        assert!(s.starts_with("2.71"));
1083    }
1084
1085    #[test]
1086    fn test_sqpvalue_display_text() {
1087        assert_eq!(SqpValue::Text("hi".into()).to_string(), "'hi'");
1088    }
1089
1090    #[test]
1091    fn test_sqpvalue_display_bool_true() {
1092        assert_eq!(SqpValue::Bool(true).to_string(), "true");
1093    }
1094
1095    #[test]
1096    fn test_sqpvalue_display_bool_false() {
1097        assert_eq!(SqpValue::Bool(false).to_string(), "false");
1098    }
1099
1100    #[test]
1101    fn test_sqpvalue_display_null() {
1102        assert_eq!(SqpValue::Null.to_string(), "NULL");
1103    }
1104
1105    #[test]
1106    fn test_sqpvalue_hash_bytes_distinct() {
1107        assert_ne!(
1108            SqpValue::Int(1).to_hash_bytes(),
1109            SqpValue::Float(1.0).to_hash_bytes()
1110        );
1111    }
1112
1113    // ── SqpOp ──────────────────────────────────────────────────────────────
1114
1115    #[test]
1116    fn test_sqpop_display_eq() {
1117        assert_eq!(SqpOp::Eq.to_string(), "=");
1118    }
1119
1120    #[test]
1121    fn test_sqpop_display_ne() {
1122        assert_eq!(SqpOp::Ne.to_string(), "!=");
1123    }
1124
1125    #[test]
1126    fn test_sqpop_display_lt() {
1127        assert_eq!(SqpOp::Lt.to_string(), "<");
1128    }
1129
1130    #[test]
1131    fn test_sqpop_display_le() {
1132        assert_eq!(SqpOp::Le.to_string(), "<=");
1133    }
1134
1135    #[test]
1136    fn test_sqpop_display_gt() {
1137        assert_eq!(SqpOp::Gt.to_string(), ">");
1138    }
1139
1140    #[test]
1141    fn test_sqpop_display_ge() {
1142        assert_eq!(SqpOp::Ge.to_string(), ">=");
1143    }
1144
1145    #[test]
1146    fn test_sqpop_display_in() {
1147        let op = SqpOp::In(vec![SqpValue::Int(1), SqpValue::Int(2)]);
1148        assert!(op.to_string().contains("IN"));
1149    }
1150
1151    #[test]
1152    fn test_sqpop_display_is_null() {
1153        assert_eq!(SqpOp::IsNull.to_string(), "IS NULL");
1154    }
1155
1156    #[test]
1157    fn test_sqpop_display_is_not_null() {
1158        assert_eq!(SqpOp::IsNotNull.to_string(), "IS NOT NULL");
1159    }
1160
1161    #[test]
1162    fn test_sqpop_tags_unique() {
1163        let ops: Vec<u8> = vec![
1164            SqpOp::Eq.tag(),
1165            SqpOp::Ne.tag(),
1166            SqpOp::Lt.tag(),
1167            SqpOp::Le.tag(),
1168            SqpOp::Gt.tag(),
1169            SqpOp::Ge.tag(),
1170            SqpOp::In(vec![]).tag(),
1171            SqpOp::IsNull.tag(),
1172            SqpOp::IsNotNull.tag(),
1173        ];
1174        let unique: std::collections::HashSet<u8> = ops.iter().copied().collect();
1175        assert_eq!(unique.len(), ops.len());
1176    }
1177
1178    // ── SqpPredicate ───────────────────────────────────────────────────────
1179
1180    #[test]
1181    fn test_predicate_display() {
1182        let p = predicate_eq("age", 30);
1183        assert!(p.to_string().contains("age"));
1184        assert!(p.to_string().contains("30"));
1185    }
1186
1187    #[test]
1188    fn test_predicate_hash_bytes_non_empty() {
1189        let p = predicate_eq("field", 1);
1190        assert!(!p.hash_bytes().is_empty());
1191    }
1192
1193    #[test]
1194    fn test_predicates_different_fields_different_hash() {
1195        let p1 = predicate_eq("field_a", 1);
1196        let p2 = predicate_eq("field_b", 1);
1197        assert_ne!(p1.hash_bytes(), p2.hash_bytes());
1198    }
1199
1200    // ── SqpQuery fingerprint ───────────────────────────────────────────────
1201
1202    #[test]
1203    fn test_query_fingerprint_deterministic() {
1204        let mut q = simple_query(42);
1205        q.predicates.push(predicate_eq("x", 1));
1206        assert_eq!(q.fingerprint(), q.fingerprint());
1207    }
1208
1209    #[test]
1210    fn test_query_fingerprint_changes_with_predicate() {
1211        let mut q1 = simple_query(1);
1212        let mut q2 = simple_query(1);
1213        q1.predicates.push(predicate_eq("a", 1));
1214        q2.predicates.push(predicate_eq("b", 1));
1215        assert_ne!(q1.fingerprint(), q2.fingerprint());
1216    }
1217
1218    #[test]
1219    fn test_query_fingerprint_changes_with_limit() {
1220        let mut q1 = simple_query(1);
1221        let mut q2 = simple_query(1);
1222        q1.limit = Some(10);
1223        q2.limit = Some(20);
1224        assert_ne!(q1.fingerprint(), q2.fingerprint());
1225    }
1226
1227    // ── Index registration ─────────────────────────────────────────────────
1228
1229    #[test]
1230    fn test_register_index_returns_id() {
1231        let planner = make_planner();
1232        let id = planner.register_index("age_idx", 10_000, 0.05);
1233        assert!(id > 0);
1234    }
1235
1236    #[test]
1237    fn test_register_index_same_name_returns_same_id() {
1238        let planner = make_planner();
1239        let id1 = planner.register_index("idx", 100, 0.1);
1240        let id2 = planner.register_index("idx", 200, 0.2);
1241        assert_eq!(id1, id2);
1242    }
1243
1244    #[test]
1245    fn test_register_index_updates_stats() {
1246        let planner = make_planner();
1247        let id = planner.register_index("idx", 100, 0.1);
1248        planner.register_index("idx", 999, 0.5);
1249        let stat = planner.index_stat(id).expect("stat must exist");
1250        assert_eq!(stat.cardinality, 999);
1251    }
1252
1253    #[test]
1254    fn test_register_two_indexes() {
1255        let planner = make_planner();
1256        let id1 = planner.register_index("a", 100, 0.1);
1257        let id2 = planner.register_index("b", 200, 0.2);
1258        assert_ne!(id1, id2);
1259    }
1260
1261    #[test]
1262    fn test_index_stat_unknown_id_returns_none() {
1263        let planner = make_planner();
1264        assert!(planner.index_stat(9999).is_none());
1265    }
1266
1267    #[test]
1268    fn test_selectivity_clamped_above_one() {
1269        let planner = make_planner();
1270        let id = planner.register_index("x", 100, 5.0);
1271        let stat = planner.index_stat(id).unwrap();
1272        assert!(stat.selectivity <= 1.0);
1273    }
1274
1275    #[test]
1276    fn test_selectivity_clamped_below_zero() {
1277        let planner = make_planner();
1278        let id = planner.register_index("x", 100, -1.0);
1279        let stat = planner.index_stat(id).unwrap();
1280        assert!(stat.selectivity >= 0.0);
1281    }
1282
1283    // ── plan() ────────────────────────────────────────────────────────────
1284
1285    #[test]
1286    fn test_plan_returns_non_zero_rows() {
1287        let planner = make_planner();
1288        let q = simple_query(1);
1289        let plan = planner.plan(&q);
1290        assert!(plan.estimated_rows > 0);
1291    }
1292
1293    #[test]
1294    fn test_plan_has_at_least_one_step() {
1295        let planner = make_planner();
1296        let q = simple_query(1);
1297        let plan = planner.plan(&q);
1298        assert!(!plan.steps.is_empty());
1299    }
1300
1301    #[test]
1302    fn test_plan_positive_cost() {
1303        let planner = make_planner();
1304        let q = simple_query(1);
1305        let plan = planner.plan(&q);
1306        assert!(plan.estimated_cost > 0.0);
1307    }
1308
1309    #[test]
1310    fn test_plan_seq_scan_when_no_index() {
1311        let planner = make_planner();
1312        let q = simple_query(1);
1313        let plan = planner.plan(&q);
1314        assert!(!plan.uses_index);
1315    }
1316
1317    #[test]
1318    fn test_plan_index_scan_when_low_selectivity() {
1319        let planner = make_planner();
1320        planner.register_index("age", 100_000, 0.01);
1321        let mut q = SqpQuery::new(1);
1322        q.predicates.push(predicate_eq("age", 25));
1323        let plan = planner.plan(&q);
1324        assert!(plan.uses_index);
1325    }
1326
1327    #[test]
1328    fn test_plan_seq_scan_when_high_selectivity() {
1329        let planner = make_planner();
1330        planner.register_index("status", 2, 0.5);
1331        let mut q = SqpQuery::new(1);
1332        q.predicates.push(predicate_eq("status", 1));
1333        let plan = planner.plan(&q);
1334        assert!(!plan.uses_index);
1335    }
1336
1337    #[test]
1338    fn test_plan_with_limit_reduces_rows() {
1339        let planner = make_planner();
1340        let mut q = simple_query(1);
1341        q.limit = Some(5);
1342        let plan = planner.plan(&q);
1343        assert!(plan.estimated_rows <= 5);
1344    }
1345
1346    #[test]
1347    fn test_plan_with_projection_adds_project_step() {
1348        let planner = make_planner();
1349        let mut q = simple_query(1);
1350        q.projections = vec!["col_a".into(), "col_b".into()];
1351        let plan = planner.plan(&q);
1352        let has_project = plan
1353            .steps
1354            .iter()
1355            .any(|s| matches!(s, SqpPlanStep::Project(_)));
1356        assert!(has_project);
1357    }
1358
1359    #[test]
1360    fn test_plan_with_order_by_adds_sort_step() {
1361        let planner = make_planner();
1362        let mut q = simple_query(1);
1363        q.order_by = Some(("ts".into(), true));
1364        let plan = planner.plan(&q);
1365        let has_sort = plan
1366            .steps
1367            .iter()
1368            .any(|s| matches!(s, SqpPlanStep::Sort { .. }));
1369        assert!(has_sort);
1370    }
1371
1372    // ── Hints ─────────────────────────────────────────────────────────────
1373
1374    #[test]
1375    fn test_hint_force_seq_scan() {
1376        let planner = make_planner();
1377        planner.register_index("x", 10_000, 0.001); // would normally trigger index scan
1378        let mut q = SqpQuery::new(1);
1379        q.predicates.push(predicate_eq("x", 1));
1380        q.hint = Some(SqpHint::ForceSeqScan);
1381        let plan = planner.plan(&q);
1382        assert!(!plan.uses_index);
1383    }
1384
1385    #[test]
1386    fn test_hint_force_index_scan() {
1387        let planner = make_planner();
1388        planner.register_index("x", 10_000, 0.9); // high selectivity, normally seq scan
1389        let mut q = SqpQuery::new(1);
1390        q.predicates.push(predicate_eq("x", 1));
1391        q.hint = Some(SqpHint::ForceIndexScan);
1392        let plan = planner.plan(&q);
1393        assert!(plan.uses_index);
1394    }
1395
1396    #[test]
1397    fn test_hint_force_index_scan_no_index_falls_back_to_seq() {
1398        let planner = make_planner();
1399        let mut q = SqpQuery::new(1);
1400        q.predicates.push(predicate_eq("x", 1));
1401        q.hint = Some(SqpHint::ForceIndexScan);
1402        let plan = planner.plan(&q);
1403        // No index registered — should fall back to seq scan.
1404        assert!(!plan.uses_index);
1405    }
1406
1407    // ── plan_cached() ─────────────────────────────────────────────────────
1408
1409    #[test]
1410    fn test_plan_cached_second_call_is_cache_hit() {
1411        let planner = make_planner();
1412        let q = simple_query(77);
1413        planner.plan_cached(&q);
1414        planner.plan_cached(&q);
1415        let stats = planner.planner_stats();
1416        assert!(stats.cache_hit_rate > 0.0);
1417    }
1418
1419    #[test]
1420    fn test_plan_cached_returns_same_plan() {
1421        let planner = make_planner();
1422        let q = simple_query(77);
1423        let p1 = planner.plan_cached(&q);
1424        let p2 = planner.plan_cached(&q);
1425        assert_eq!(p1.estimated_cost.to_bits(), p2.estimated_cost.to_bits());
1426        assert_eq!(p1.estimated_rows, p2.estimated_rows);
1427    }
1428
1429    #[test]
1430    fn test_plan_cache_respects_max_size() {
1431        let config = SqpPlannerConfig {
1432            max_cache_size: 3,
1433            ..Default::default()
1434        };
1435        let planner = StorageQueryPlanner::new(config);
1436        for i in 0..10u64 {
1437            planner.plan_cached(&simple_query(i));
1438        }
1439        let stats = planner.planner_stats();
1440        assert!(stats.cache_size <= 3);
1441    }
1442
1443    // ── invalidate_cache ──────────────────────────────────────────────────
1444
1445    #[test]
1446    fn test_invalidate_cache_clears_cache() {
1447        let planner = make_planner();
1448        planner.plan_cached(&simple_query(1));
1449        planner.plan_cached(&simple_query(2));
1450        planner.invalidate_cache();
1451        assert_eq!(planner.planner_stats().cache_size, 0);
1452    }
1453
1454    #[test]
1455    fn test_invalidate_for_index_removes_plan() {
1456        let planner = make_planner();
1457        let idx = planner.register_index("field", 1000, 0.05);
1458        let mut q = SqpQuery::new(1);
1459        q.predicates.push(predicate_eq("field", 42));
1460        planner.plan_cached(&q);
1461        planner.invalidate_for_index(idx);
1462        // After invalidation a new plan_cached call must miss the cache.
1463        // The cache_size should now be 0 (or the plan was evicted).
1464        let before_total = planner.planner_stats().total_plans;
1465        planner.plan_cached(&q);
1466        // A new planning call increments total_plans.
1467        assert!(planner.planner_stats().total_plans > before_total);
1468    }
1469
1470    // ── estimate_cost ─────────────────────────────────────────────────────
1471
1472    #[test]
1473    fn test_estimate_cost_positive() {
1474        let planner = make_planner();
1475        let q = simple_query(1);
1476        let plan = planner.plan(&q);
1477        assert!(planner.estimate_cost(&plan) > 0.0);
1478    }
1479
1480    #[test]
1481    fn test_estimate_cost_index_less_than_seq_for_low_rows() {
1482        let config = SqpPlannerConfig {
1483            cost_model: SqpCostModel::IOCost,
1484            ..Default::default()
1485        };
1486        let planner = StorageQueryPlanner::new(config);
1487        planner.register_index("x", 100_000, 0.01);
1488        let mut q_idx = SqpQuery::new(1);
1489        q_idx.predicates.push(predicate_eq("x", 1));
1490        q_idx.hint = Some(SqpHint::ForceIndexScan);
1491
1492        let mut q_seq = SqpQuery::new(2);
1493        q_seq.predicates.push(predicate_eq("x", 1));
1494        q_seq.hint = Some(SqpHint::ForceSeqScan);
1495
1496        let idx_plan = planner.plan(&q_idx);
1497        let seq_plan = planner.plan(&q_seq);
1498        // Index scan on 1 % of rows should be cheaper than full scan.
1499        assert!(
1500            idx_plan.estimated_cost < seq_plan.estimated_cost,
1501            "idx={} seq={}",
1502            idx_plan.estimated_cost,
1503            seq_plan.estimated_cost
1504        );
1505    }
1506
1507    // ── Cost models ───────────────────────────────────────────────────────
1508
1509    #[test]
1510    fn test_cost_model_simple_row_count() {
1511        let config = SqpPlannerConfig {
1512            cost_model: SqpCostModel::SimpleRowCount,
1513            ..Default::default()
1514        };
1515        let planner = StorageQueryPlanner::new(config);
1516        let q = simple_query(1);
1517        let plan = planner.plan(&q);
1518        assert_eq!(plan.estimated_cost, plan.estimated_rows as f64);
1519    }
1520
1521    #[test]
1522    fn test_cost_model_memory_cost() {
1523        let config = SqpPlannerConfig {
1524            cost_model: SqpCostModel::MemoryCost,
1525            ..Default::default()
1526        };
1527        let planner = StorageQueryPlanner::new(config);
1528        let plan = planner.plan(&simple_query(1));
1529        assert!(plan.estimated_cost > 0.0);
1530    }
1531
1532    #[test]
1533    fn test_cost_model_io_cost() {
1534        let config = SqpPlannerConfig {
1535            cost_model: SqpCostModel::IOCost,
1536            ..Default::default()
1537        };
1538        let planner = StorageQueryPlanner::new(config);
1539        let plan = planner.plan(&simple_query(1));
1540        assert!(plan.estimated_cost > 0.0);
1541    }
1542
1543    #[test]
1544    fn test_cost_model_hybrid() {
1545        let config = SqpPlannerConfig {
1546            cost_model: SqpCostModel::HybridCost,
1547            ..Default::default()
1548        };
1549        let planner = StorageQueryPlanner::new(config);
1550        let plan = planner.plan(&simple_query(1));
1551        assert!(plan.estimated_cost > 0.0);
1552    }
1553
1554    // ── explain() ─────────────────────────────────────────────────────────
1555
1556    #[test]
1557    fn test_explain_contains_query_id() {
1558        let planner = make_planner();
1559        let plan = planner.plan(&simple_query(99));
1560        let text = planner.explain(&plan);
1561        assert!(text.contains("id=99"));
1562    }
1563
1564    #[test]
1565    fn test_explain_contains_steps() {
1566        let planner = make_planner();
1567        let plan = planner.plan(&simple_query(1));
1568        let text = planner.explain(&plan);
1569        assert!(text.contains("Scan"));
1570    }
1571
1572    #[test]
1573    fn test_explain_contains_index_info_when_used() {
1574        let planner = make_planner();
1575        planner.register_index("age", 100_000, 0.01);
1576        let mut q = SqpQuery::new(5);
1577        q.predicates.push(predicate_eq("age", 30));
1578        let plan = planner.plan(&q);
1579        let text = planner.explain(&plan);
1580        assert!(text.contains("age"));
1581    }
1582
1583    // ── planner_stats() ───────────────────────────────────────────────────
1584
1585    #[test]
1586    fn test_stats_total_plans_increments() {
1587        let planner = make_planner();
1588        planner.plan(&simple_query(1));
1589        planner.plan(&simple_query(2));
1590        assert_eq!(planner.planner_stats().total_plans, 2);
1591    }
1592
1593    #[test]
1594    fn test_stats_registered_indexes() {
1595        let planner = make_planner();
1596        planner.register_index("a", 100, 0.1);
1597        planner.register_index("b", 200, 0.2);
1598        assert_eq!(planner.planner_stats().registered_indexes, 2);
1599    }
1600
1601    #[test]
1602    fn test_stats_index_usage_rate_non_zero() {
1603        let planner = make_planner();
1604        planner.register_index("x", 1_000_000, 0.001);
1605        let mut q = SqpQuery::new(1);
1606        q.predicates.push(predicate_eq("x", 1));
1607        planner.plan(&q);
1608        let stats = planner.planner_stats();
1609        assert!(stats.index_usage_rate > 0.0);
1610    }
1611
1612    #[test]
1613    fn test_stats_avg_cost_positive_after_plan() {
1614        let planner = make_planner();
1615        planner.plan(&simple_query(1));
1616        let stats = planner.planner_stats();
1617        assert!(stats.avg_cost > 0.0);
1618    }
1619
1620    #[test]
1621    fn test_stats_cache_hit_rate_zero_before_cached() {
1622        let planner = make_planner();
1623        planner.plan(&simple_query(1));
1624        let stats = planner.planner_stats();
1625        assert_eq!(stats.cache_hit_rate, 0.0);
1626    }
1627
1628    // ── History ───────────────────────────────────────────────────────────
1629
1630    #[test]
1631    fn test_history_grows_with_plans() {
1632        let planner = make_planner();
1633        planner.plan(&simple_query(1));
1634        planner.plan(&simple_query(2));
1635        assert_eq!(planner.recent_history(10).len(), 2);
1636    }
1637
1638    #[test]
1639    fn test_history_bounded_at_500() {
1640        let planner = make_planner();
1641        for i in 0..600u64 {
1642            planner.plan(&simple_query(i));
1643        }
1644        let hist = planner.recent_history(1000);
1645        assert!(hist.len() <= 500);
1646    }
1647
1648    #[test]
1649    fn test_history_cache_hit_recorded() {
1650        let planner = make_planner();
1651        let q = simple_query(1);
1652        planner.plan_cached(&q);
1653        planner.plan_cached(&q);
1654        let hist = planner.recent_history(2);
1655        assert!(hist.iter().any(|r| r.cache_hit));
1656    }
1657
1658    // ── SqpPlanStep display ───────────────────────────────────────────────
1659
1660    #[test]
1661    fn test_step_display_seq_scan() {
1662        let s = SqpPlanStep::SeqScan {
1663            table: "blocks".into(),
1664            filter_count: 2,
1665        };
1666        assert!(s.to_string().contains("blocks"));
1667    }
1668
1669    #[test]
1670    fn test_step_display_index_scan() {
1671        let s = SqpPlanStep::IndexScan {
1672            index_id: 3,
1673            selectivity: 0.05,
1674        };
1675        assert!(s.to_string().contains("3"));
1676    }
1677
1678    #[test]
1679    fn test_step_display_filter() {
1680        let s = SqpPlanStep::Filter(predicate_eq("col", 1));
1681        assert!(s.to_string().contains("col"));
1682    }
1683
1684    #[test]
1685    fn test_step_display_sort_asc() {
1686        let s = SqpPlanStep::Sort {
1687            field: "ts".into(),
1688            asc: true,
1689        };
1690        assert!(s.to_string().contains("ASC"));
1691    }
1692
1693    #[test]
1694    fn test_step_display_sort_desc() {
1695        let s = SqpPlanStep::Sort {
1696            field: "ts".into(),
1697            asc: false,
1698        };
1699        assert!(s.to_string().contains("DESC"));
1700    }
1701
1702    #[test]
1703    fn test_step_display_limit() {
1704        let s = SqpPlanStep::Limit(100);
1705        assert!(s.to_string().contains("100"));
1706    }
1707
1708    #[test]
1709    fn test_step_display_project() {
1710        let s = SqpPlanStep::Project(vec!["a".into(), "b".into()]);
1711        assert!(s.to_string().contains("a"));
1712    }
1713
1714    // ── Predicate pushdown ────────────────────────────────────────────────
1715
1716    #[test]
1717    fn test_predicate_pushdown_disabled_still_produces_filter_steps() {
1718        let config = SqpPlannerConfig {
1719            enable_predicate_pushdown: false,
1720            ..Default::default()
1721        };
1722        let planner = StorageQueryPlanner::new(config);
1723        let mut q = SqpQuery::new(1);
1724        q.predicates.push(predicate_eq("x", 1));
1725        q.predicates.push(predicate_eq("y", 2));
1726        let plan = planner.plan(&q);
1727        let filter_count = plan
1728            .steps
1729            .iter()
1730            .filter(|s| matches!(s, SqpPlanStep::Filter(_)))
1731            .count();
1732        assert_eq!(filter_count, 2);
1733    }
1734
1735    // ── Multiple predicates ───────────────────────────────────────────────
1736
1737    #[test]
1738    fn test_multiple_predicates_reduce_estimated_rows() {
1739        let planner = make_planner();
1740        let mut q_one = SqpQuery::new(1);
1741        q_one.predicates.push(predicate_eq("a", 1));
1742
1743        let mut q_two = SqpQuery::new(2);
1744        q_two.predicates.push(predicate_eq("a", 1));
1745        q_two.predicates.push(predicate_eq("b", 2));
1746
1747        let p1 = planner.plan(&q_one);
1748        let p2 = planner.plan(&q_two);
1749        assert!(p2.estimated_rows <= p1.estimated_rows);
1750    }
1751
1752    // ── SqpCostModel default ──────────────────────────────────────────────
1753
1754    #[test]
1755    fn test_cost_model_default_is_hybrid() {
1756        assert_eq!(SqpCostModel::default(), SqpCostModel::HybridCost);
1757    }
1758
1759    // ── SqpPlannerConfig default ──────────────────────────────────────────
1760
1761    #[test]
1762    fn test_planner_config_default_max_cache() {
1763        assert_eq!(SqpPlannerConfig::default().max_cache_size, 200);
1764    }
1765
1766    #[test]
1767    fn test_planner_config_default_pushdown_enabled() {
1768        assert!(SqpPlannerConfig::default().enable_predicate_pushdown);
1769    }
1770
1771    // ── SqpQuery::new ─────────────────────────────────────────────────────
1772
1773    #[test]
1774    fn test_query_new_has_no_predicates() {
1775        let q = SqpQuery::new(1);
1776        assert!(q.predicates.is_empty());
1777    }
1778
1779    #[test]
1780    fn test_query_new_has_no_limit() {
1781        let q = SqpQuery::new(1);
1782        assert!(q.limit.is_none());
1783    }
1784
1785    // ── SqpQueryPlan ──────────────────────────────────────────────────────
1786
1787    #[test]
1788    fn test_plan_query_id_matches() {
1789        let planner = make_planner();
1790        let plan = planner.plan(&simple_query(42));
1791        assert_eq!(plan.query_id, 42);
1792    }
1793
1794    #[test]
1795    fn test_plan_planned_at_ns_positive() {
1796        let planner = make_planner();
1797        let plan = planner.plan(&simple_query(1));
1798        assert!(plan.planned_at_ns > 0);
1799    }
1800
1801    // ── In-predicate selectivity ──────────────────────────────────────────
1802
1803    #[test]
1804    fn test_in_predicate_selectivity_scales_with_count() {
1805        let planner = make_planner();
1806        let mut q_small = SqpQuery::new(1);
1807        q_small.predicates.push(SqpPredicate::new(
1808            "x",
1809            SqpOp::In(vec![SqpValue::Int(1)]),
1810            SqpValue::Null,
1811        ));
1812        let mut q_large = SqpQuery::new(2);
1813        q_large.predicates.push(SqpPredicate::new(
1814            "x",
1815            SqpOp::In(vec![
1816                SqpValue::Int(1),
1817                SqpValue::Int(2),
1818                SqpValue::Int(3),
1819                SqpValue::Int(4),
1820            ]),
1821            SqpValue::Null,
1822        ));
1823        let p_small = planner.plan(&q_small);
1824        let p_large = planner.plan(&q_large);
1825        assert!(p_large.estimated_rows >= p_small.estimated_rows);
1826    }
1827
1828    // ── Type alias ────────────────────────────────────────────────────────
1829
1830    #[test]
1831    fn test_type_alias_usable() {
1832        let _: SqpStorageQueryPlanner = StorageQueryPlanner::default();
1833    }
1834
1835    // ── Thread safety (smoke test) ────────────────────────────────────────
1836
1837    #[test]
1838    fn test_concurrent_plans_do_not_panic() {
1839        use std::sync::Arc;
1840        let planner = Arc::new(make_planner());
1841        let handles: Vec<_> = (0..8)
1842            .map(|i| {
1843                let p = Arc::clone(&planner);
1844                std::thread::spawn(move || {
1845                    let mut q = SqpQuery::new(i);
1846                    q.predicates.push(predicate_eq("col", i as i64));
1847                    p.plan_cached(&q);
1848                })
1849            })
1850            .collect();
1851        for h in handles {
1852            h.join().expect("thread must not panic");
1853        }
1854        assert!(planner.planner_stats().total_plans >= 8);
1855    }
1856}
ipfrs_storage/storage_query_planner.rs

ipfrs_storage/
storage_query_planner.rs
