sz_orm_core/
query_cache.rs

1#![allow(missing_docs)]
2//! 查询缓存 + 时间戳缓存（Query Cache + Timestamp Cache）
3//!
4//! 对标 Hibernate `QueryCache` + `UpdateTimestampsCache`。
5//!
6//! 缓存查询结果，当相关表被修改时自动失效。
7//!
8//! # 工作原理
9//!
10//! 1. **Query Cache**：缓存 `(SQL, params) → results` 映射
11//! 2. **Timestamp Cache**：跟踪每个表的最后修改时间
12//! 3. 查询时：如果 Query Cache 命中且所有相关表的 timestamp 未变，返回缓存结果
13//! 4. 写入时：更新相关表的 timestamp，使依赖该表的 Query Cache 条目自动失效
14
15use std::collections::HashMap;
16use std::sync::{Arc, Mutex};
17use std::time::Instant;
18
19/// 查询缓存键
20#[derive(Debug, Clone, PartialEq, Eq, Hash)]
21pub struct QueryCacheKey {
22    pub sql: String,
23    pub params_hash: u64,
24}
25
26impl QueryCacheKey {
27    pub fn new(sql: &str, params_hash: u64) -> Self {
28        Self {
29            sql: sql.to_string(),
30            params_hash,
31        }
32    }
33}
34
35/// 缓存的查询结果
36#[derive(Debug, Clone)]
37pub struct CachedQueryResult {
38    pub rows: Vec<HashMap<String, crate::Value>>,
39    pub cached_at: Instant,
40    pub depends_on: Vec<String>,
41}
42
43/// 时间戳缓存
44///
45/// 跟踪每个表的最后修改时间。
46#[derive(Default)]
47pub struct TimestampCache {
48    table_timestamps: Arc<Mutex<HashMap<String, Instant>>>,
49}
50
51impl TimestampCache {
52    pub fn new() -> Self {
53        Self::default()
54    }
55
56    /// 获取表的最后修改时间
57    pub fn get(&self, table: &str) -> Option<Instant> {
58        self.table_timestamps.lock().unwrap().get(table).copied()
59    }
60
61    /// 更新表的修改时间
62    pub fn touch(&self, table: &str) {
63        self.table_timestamps
64            .lock()
65            .unwrap()
66            .insert(table.to_string(), Instant::now());
67    }
68
69    /// 批量更新表
70    pub fn touch_all(&self, tables: &[&str]) {
71        let mut ts = self.table_timestamps.lock().unwrap();
72        let now = Instant::now();
73        for table in tables {
74            ts.insert(table.to_string(), now);
75        }
76    }
77
78    /// 检查查询是否过期（依赖的任何表在缓存后被修改）
79    pub fn is_expired(&self, cached_at: Instant, depends_on: &[String]) -> bool {
80        let ts = self.table_timestamps.lock().unwrap();
81        for table in depends_on {
82            if let Some(t) = ts.get(table) {
83                if *t > cached_at {
84                    return true;
85                }
86            }
87        }
88        false
89    }
90
91    /// 清除所有时间戳
92    pub fn clear(&self) {
93        self.table_timestamps.lock().unwrap().clear();
94    }
95}
96
97/// 查询缓存
98///
99/// 缓存查询结果，配合 `TimestampCache` 实现自动失效。
100pub struct QueryCache {
101    cache: Arc<Mutex<HashMap<QueryCacheKey, CachedQueryResult>>>,
102    timestamps: TimestampCache,
103    max_entries: usize,
104}
105
106impl QueryCache {
107    pub fn new(max_entries: usize) -> Self {
108        Self {
109            cache: Arc::new(Mutex::new(HashMap::new())),
110            timestamps: TimestampCache::new(),
111            max_entries,
112        }
113    }
114
115    /// 尝试从缓存获取查询结果
116    ///
117    /// 如果缓存命中且未过期，返回 `Some(results)`。
118    /// 否则返回 `None`。
119    pub fn get(&self, key: &QueryCacheKey) -> Option<Vec<HashMap<String, crate::Value>>> {
120        let cache = self.cache.lock().unwrap();
121        if let Some(entry) = cache.get(key) {
122            if !self
123                .timestamps
124                .is_expired(entry.cached_at, &entry.depends_on)
125            {
126                return Some(entry.rows.clone());
127            }
128        }
129        None
130    }
131
132    /// 缓存查询结果
133    ///
134    /// `depends_on` 指定此查询依赖的表列表。
135    /// 当这些表被修改时，缓存条目自动失效。
136    pub fn put(
137        &self,
138        key: QueryCacheKey,
139        rows: Vec<HashMap<String, crate::Value>>,
140        depends_on: Vec<String>,
141    ) {
142        let mut cache = self.cache.lock().unwrap();
143        if cache.len() >= self.max_entries {
144            let oldest_key = cache
145                .iter()
146                .min_by_key(|(_, v)| v.cached_at)
147                .map(|(k, _)| k.clone());
148            if let Some(k) = oldest_key {
149                cache.remove(&k);
150            }
151        }
152        cache.insert(
153            key,
154            CachedQueryResult {
155                rows,
156                cached_at: Instant::now(),
157                depends_on,
158            },
159        );
160    }
161
162    /// 通知表被修改（使依赖该表的缓存条目失效）
163    pub fn invalidate_table(&self, table: &str) {
164        self.timestamps.touch(table);
165    }
166
167    /// 批量失效
168    pub fn invalidate_tables(&self, tables: &[&str]) {
169        self.timestamps.touch_all(tables);
170    }
171
172    /// 清除所有缓存
173    pub fn clear(&self) {
174        self.cache.lock().unwrap().clear();
175        self.timestamps.clear();
176    }
177
178    /// 当前缓存条目数
179    pub fn len(&self) -> usize {
180        self.cache.lock().unwrap().len()
181    }
182
183    /// 是否为空
184    pub fn is_empty(&self) -> bool {
185        self.len() == 0
186    }
187
188    /// 获取时间戳缓存引用
189    pub fn timestamps(&self) -> &TimestampCache {
190        &self.timestamps
191    }
192}
193
194impl Default for QueryCache {
195    fn default() -> Self {
196        Self::new(1000)
197    }
198}
199
200#[cfg(test)]
201mod tests {
202    use super::*;
203    use crate::Value;
204    use std::time::Duration;
205
206    fn make_row(name: &str, age: i64) -> HashMap<String, Value> {
207        let mut m = HashMap::new();
208        m.insert("name".to_string(), Value::String(name.to_string()));
209        m.insert("age".to_string(), Value::I64(age));
210        m
211    }
212
213    #[test]
214    fn test_query_cache_put_get() {
215        let cache = QueryCache::new(100);
216        let key = QueryCacheKey::new("SELECT * FROM users", 0);
217
218        cache.put(
219            key.clone(),
220            vec![make_row("alice", 25)],
221            vec!["users".to_string()],
222        );
223
224        let result = cache.get(&key).unwrap();
225        assert_eq!(result.len(), 1);
226        assert_eq!(result[0]["name"], Value::String("alice".to_string()));
227    }
228
229    #[test]
230    fn test_query_cache_miss() {
231        let cache = QueryCache::new(100);
232        let key = QueryCacheKey::new("SELECT * FROM users", 0);
233        assert!(cache.get(&key).is_none());
234    }
235
236    #[test]
237    fn test_query_cache_invalidate_on_table_modify() {
238        let cache = QueryCache::new(100);
239        let key = QueryCacheKey::new("SELECT * FROM users", 0);
240
241        cache.put(
242            key.clone(),
243            vec![make_row("alice", 25)],
244            vec!["users".to_string()],
245        );
246        assert!(cache.get(&key).is_some());
247
248        cache.invalidate_table("users");
249        assert!(cache.get(&key).is_none());
250    }
251
252    #[test]
253    fn test_query_cache_unaffected_table_no_invalidate() {
254        let cache = QueryCache::new(100);
255        let key = QueryCacheKey::new("SELECT * FROM users", 0);
256
257        cache.put(
258            key.clone(),
259            vec![make_row("alice", 25)],
260            vec!["users".to_string()],
261        );
262
263        cache.invalidate_table("orders");
264        assert!(cache.get(&key).is_some());
265    }
266
267    #[test]
268    fn test_query_cache_multi_table_dependency() {
269        let cache = QueryCache::new(100);
270        let key = QueryCacheKey::new(
271            "SELECT u.*, o.* FROM users u JOIN orders o ON u.id = o.user_id",
272            0,
273        );
274
275        cache.put(
276            key.clone(),
277            vec![make_row("alice", 25)],
278            vec!["users".to_string(), "orders".to_string()],
279        );
280        assert!(cache.get(&key).is_some());
281
282        cache.invalidate_table("orders");
283        assert!(cache.get(&key).is_none());
284    }
285
286    #[test]
287    fn test_query_cache_lru_eviction() {
288        let cache = QueryCache::new(2);
289
290        let key1 = QueryCacheKey::new("SELECT * FROM users WHERE id = 1", 0);
291        let key2 = QueryCacheKey::new("SELECT * FROM users WHERE id = 2", 0);
292        let key3 = QueryCacheKey::new("SELECT * FROM users WHERE id = 3", 0);
293
294        cache.put(key1.clone(), vec![], vec![]);
295        cache.put(key2.clone(), vec![], vec![]);
296        cache.put(key3.clone(), vec![], vec![]);
297
298        assert_eq!(cache.len(), 2);
299    }
300
301    #[test]
302    fn test_timestamp_cache_touch() {
303        let ts = TimestampCache::new();
304        assert!(ts.get("users").is_none());
305
306        ts.touch("users");
307        assert!(ts.get("users").is_some());
308    }
309
310    #[test]
311    fn test_timestamp_cache_is_expired() {
312        let ts = TimestampCache::new();
313        let cached_at = Instant::now();
314
315        std::thread::sleep(Duration::from_millis(1));
316        ts.touch("users");
317
318        assert!(ts.is_expired(cached_at, &["users".to_string()]));
319        assert!(!ts.is_expired(cached_at, &["orders".to_string()]));
320    }
321
322    #[test]
323    fn test_query_cache_clear() {
324        let cache = QueryCache::new(100);
325        let key = QueryCacheKey::new("SELECT * FROM users", 0);
326        cache.put(key, vec![], vec![]);
327        assert_eq!(cache.len(), 1);
328
329        cache.clear();
330        assert!(cache.is_empty());
331    }
332
333    #[test]
334    fn test_query_cache_different_params() {
335        let cache = QueryCache::new(100);
336        let key1 = QueryCacheKey::new("SELECT * FROM users WHERE age > ?", 1);
337        let key2 = QueryCacheKey::new("SELECT * FROM users WHERE age > ?", 2);
338
339        cache.put(key1.clone(), vec![make_row("alice", 25)], vec![]);
340        cache.put(key2.clone(), vec![make_row("bob", 30)], vec![]);
341
342        assert_eq!(
343            cache.get(&key1).unwrap()[0]["name"],
344            Value::String("alice".into())
345        );
346        assert_eq!(
347            cache.get(&key2).unwrap()[0]["name"],
348            Value::String("bob".into())
349        );
350    }
351
352    #[test]
353    fn test_invalidate_tables_batch() {
354        let cache = QueryCache::new(100);
355        let key = QueryCacheKey::new("SELECT * FROM users JOIN orders", 0);
356
357        cache.put(
358            key.clone(),
359            vec![],
360            vec!["users".to_string(), "orders".to_string()],
361        );
362        assert!(cache.get(&key).is_some());
363
364        cache.invalidate_tables(&["users", "orders"]);
365        assert!(cache.get(&key).is_none());
366    }
367
368    #[test]
369    fn test_e2e_query_cache_hit_miss_cycle() {
370        let cache = QueryCache::new(100);
371        let query_count = Arc::new(Mutex::new(0));
372
373        let execute_query = |_sql: &str, qc: Arc<Mutex<usize>>| -> Vec<HashMap<String, Value>> {
374            *qc.lock().unwrap() += 1;
375            vec![make_row("alice", 25)]
376        };
377
378        let key = QueryCacheKey::new("SELECT * FROM users WHERE id = ?", 1);
379        let sql = "SELECT * FROM users WHERE id = ?";
380
381        let result = execute_query(sql, Arc::clone(&query_count));
382        cache.put(key.clone(), result, vec!["users".to_string()]);
383        assert_eq!(*query_count.lock().unwrap(), 1);
384
385        let cached = cache.get(&key);
386        assert!(cached.is_some());
387        assert_eq!(*query_count.lock().unwrap(), 1);
388
389        cache.invalidate_table("users");
390
391        let cached_after = cache.get(&key);
392        assert!(cached_after.is_none());
393
394        let result2 = execute_query(sql, Arc::clone(&query_count));
395        cache.put(key.clone(), result2, vec!["users".to_string()]);
396        assert_eq!(*query_count.lock().unwrap(), 2);
397
398        assert!(cache.get(&key).is_some());
399        assert_eq!(*query_count.lock().unwrap(), 2);
400    }
401
402    #[test]
403    fn test_e2e_join_query_multi_table_invalidation() {
404        let cache = QueryCache::new(100);
405
406        let join_sql = "SELECT u.name, o.amount FROM users u JOIN orders o ON u.id = o.user_id";
407        let key = QueryCacheKey::new(join_sql, 0);
408
409        let mut row1 = HashMap::new();
410        row1.insert("name".to_string(), Value::String("alice".into()));
411        row1.insert("amount".to_string(), Value::F64(99.5));
412        let mut row2 = HashMap::new();
413        row2.insert("name".to_string(), Value::String("alice".into()));
414        row2.insert("amount".to_string(), Value::F64(200.0));
415
416        cache.put(
417            key.clone(),
418            vec![row1, row2],
419            vec!["users".to_string(), "orders".to_string()],
420        );
421
422        assert_eq!(cache.get(&key).unwrap().len(), 2);
423
424        cache.invalidate_table("orders");
425        assert!(cache.get(&key).is_none());
426
427        cache.put(
428            key.clone(),
429            vec![make_row("alice", 25)],
430            vec!["users".to_string(), "orders".to_string()],
431        );
432        assert!(cache.get(&key).is_some());
433
434        cache.invalidate_table("users");
435        assert!(cache.get(&key).is_none());
436    }
437
438    #[test]
439    fn test_e2e_cache_with_params_isolation() {
440        let cache = QueryCache::new(100);
441        let sql = "SELECT * FROM users WHERE age > ?";
442
443        let key_young = QueryCacheKey::new(sql, 18);
444        let key_old = QueryCacheKey::new(sql, 60);
445
446        cache.put(
447            key_young.clone(),
448            vec![make_row("alice", 25), make_row("bob", 30)],
449            vec!["users".to_string()],
450        );
451        cache.put(
452            key_old.clone(),
453            vec![make_row("charlie", 65)],
454            vec!["users".to_string()],
455        );
456
457        assert_eq!(cache.get(&key_young).unwrap().len(), 2);
458        assert_eq!(cache.get(&key_old).unwrap().len(), 1);
459
460        cache.invalidate_table("users");
461        assert!(cache.get(&key_young).is_none());
462        assert!(cache.get(&key_old).is_none());
463    }
464}
sz_orm_core/query_cache.rs

sz_orm_core/
query_cache.rs
