nodedb_lite/engine/htap/
routing.rs

1//! HTAP query routing: decides whether to read from strict (source) or
2//! columnar (materialized view) based on session settings.
3//!
4//! Default: `read_source = 'source'` — all queries go to strict document.
5//! Opt-in: `read_source = 'auto'` — planner routes analytical queries
6//! (GROUP BY, aggregations, full scans) to columnar, point lookups to strict.
7
8/// Where to read data from when a materialized view exists.
9#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
10pub enum ReadSource {
11    /// Always read from the source (strict document). Default.
12    /// No risk of stale reads.
13    #[default]
14    Source,
15    /// Planner decides: point lookups → source, analytical scans → materialized.
16    /// May return data that lags behind the source by the CDC interval.
17    Auto,
18}
19
20impl ReadSource {
21    pub fn as_str(&self) -> &'static str {
22        match self {
23            Self::Source => "source",
24            Self::Auto => "auto",
25        }
26    }
27}
28
29impl std::str::FromStr for ReadSource {
30    type Err = String;
31
32    fn from_str(s: &str) -> Result<Self, Self::Err> {
33        match s.to_lowercase().as_str() {
34            "source" => Ok(Self::Source),
35            "auto" => Ok(Self::Auto),
36            other => Err(format!(
37                "unknown read_source: '{other}' (use 'source' or 'auto')"
38            )),
39        }
40    }
41}
42
43/// Consistency level for reading materialized views.
44#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
45pub enum MaterializedConsistency {
46    /// Accept bounded lag — reads whatever is materialized. Default for `auto` mode.
47    #[default]
48    Eventual,
49    /// Force a CDC flush before reading. Higher latency, zero lag.
50    Strong,
51}
52
53impl std::str::FromStr for MaterializedConsistency {
54    type Err = String;
55
56    fn from_str(s: &str) -> Result<Self, Self::Err> {
57        match s.to_lowercase().as_str() {
58            "eventual" => Ok(Self::Eventual),
59            "strong" => Ok(Self::Strong),
60            other => Err(format!(
61                "unknown materialized_consistency: '{other}' (use 'eventual' or 'strong')"
62            )),
63        }
64    }
65}
66
67/// Per-session HTAP routing configuration.
68///
69/// Set via `SET read_source = 'auto'` and `SET materialized_consistency = 'strong'`.
70#[derive(Debug, Clone, Default)]
71pub struct HtapSession {
72    pub read_source: ReadSource,
73    pub materialized_consistency: MaterializedConsistency,
74}
75
76impl HtapSession {
77    /// Whether this session should use the materialized view for a given query.
78    ///
79    /// Returns `true` if `read_source = 'auto'` AND the query looks analytical
80    /// (indicated by `is_analytical` — determined by the caller from query shape).
81    pub fn should_use_materialized(&self, is_analytical: bool) -> bool {
82        self.read_source == ReadSource::Auto && is_analytical
83    }
84
85    /// Whether a CDC flush should be forced before reading the materialized view.
86    pub fn requires_strong_consistency(&self) -> bool {
87        self.materialized_consistency == MaterializedConsistency::Strong
88    }
89}
90
91/// Simple heuristic to detect analytical queries from SQL.
92///
93/// Returns `true` if the query contains GROUP BY, aggregate functions,
94/// or is a full scan without a PK filter.
95pub fn is_analytical_query(sql: &str) -> bool {
96    let upper = sql.to_uppercase();
97    upper.contains("GROUP BY")
98        || upper.contains("SUM(")
99        || upper.contains("AVG(")
100        || upper.contains("COUNT(")
101        || upper.contains("MIN(")
102        || upper.contains("MAX(")
103        || (upper.contains("SELECT") && !upper.contains("WHERE") && upper.contains("FROM"))
104}
105
106#[cfg(test)]
107mod tests {
108    use super::*;
109
110    #[test]
111    fn default_session() {
112        let session = HtapSession::default();
113        assert_eq!(session.read_source, ReadSource::Source);
114        assert!(!session.should_use_materialized(true));
115        assert!(!session.should_use_materialized(false));
116    }
117
118    #[test]
119    fn auto_routing() {
120        let session = HtapSession {
121            read_source: ReadSource::Auto,
122            materialized_consistency: MaterializedConsistency::Eventual,
123        };
124        assert!(session.should_use_materialized(true));
125        assert!(!session.should_use_materialized(false));
126    }
127
128    #[test]
129    fn strong_consistency() {
130        let session = HtapSession {
131            read_source: ReadSource::Auto,
132            materialized_consistency: MaterializedConsistency::Strong,
133        };
134        assert!(session.requires_strong_consistency());
135    }
136
137    #[test]
138    fn analytical_detection() {
139        assert!(is_analytical_query(
140            "SELECT SUM(balance) FROM customers GROUP BY status"
141        ));
142        assert!(is_analytical_query("SELECT COUNT(*) FROM orders"));
143        assert!(is_analytical_query("SELECT AVG(score) FROM metrics"));
144        assert!(!is_analytical_query(
145            "SELECT * FROM customers WHERE id = 42"
146        ));
147        assert!(!is_analytical_query(
148            "INSERT INTO customers VALUES (1, 'a')"
149        ));
150    }
151
152    #[test]
153    fn read_source_parse() {
154        assert_eq!("source".parse::<ReadSource>().unwrap(), ReadSource::Source);
155        assert_eq!("auto".parse::<ReadSource>().unwrap(), ReadSource::Auto);
156        assert!("bogus".parse::<ReadSource>().is_err());
157    }
158}
nodedb_lite/engine/htap/routing.rs

nodedb_lite/engine/htap/
routing.rs
