reddb_server/runtime/
result_cache_runtime.rs

1use std::collections::hash_map::Entry;
2use std::collections::{HashMap, HashSet, VecDeque};
3
4use super::{RedDBRuntime, RuntimeQueryResult, RuntimeResultCacheEntry};
5
6const RESULT_CACHE_BACKEND_KEY: &str = "runtime.result_cache.backend";
7const RESULT_CACHE_DEFAULT_BACKEND: &str = "legacy";
8const RESULT_CACHE_BLOB_NAMESPACE: &str = "runtime.result_cache";
9const RESULT_CACHE_EMBEDDED_COLLECTION: &str = "red_internal_result_cache";
10const RESULT_CACHE_TTL_SECS: u64 = 30;
11const RESULT_CACHE_MAX_ENTRIES: usize = 1000;
12const RESULT_CACHE_ENABLED_KEY: &str = "runtime.result_cache.enabled";
13const RESULT_CACHE_TTL_KEY: &str = "runtime.result_cache.ttl_seconds";
14const RESULT_CACHE_CAPACITY_KEY: &str = "runtime.result_cache.capacity_entries";
15const RESULT_CACHE_PAYLOAD_MAGIC: &[u8; 8] = b"RDRC0001";
16
17#[derive(Clone, Copy, Debug, PartialEq, Eq)]
18enum RuntimeResultCacheBackend {
19    Legacy,
20    BlobCache,
21    Shadow,
22}
23
24fn trim_result_cache(
25    map: &mut HashMap<String, RuntimeResultCacheEntry>,
26    order: &mut VecDeque<String>,
27    max_entries: usize,
28) -> u64 {
29    let mut evicted = 0u64;
30    while map.len() > max_entries {
31        if let Some(oldest) = order.pop_front() {
32            if map.remove(&oldest).is_some() {
33                evicted += 1;
34            }
35        } else {
36            break;
37        }
38    }
39    evicted
40}
41
42fn result_cache_fingerprint(result: &RuntimeQueryResult) -> String {
43    format!(
44        "{:?}|{}|{}|{}|{}|{:?}",
45        result.result,
46        result.query,
47        result.statement,
48        result.engine,
49        result.affected_rows,
50        result.statement_type
51    )
52}
53
54fn mode_to_byte(mode: crate::storage::query::modes::QueryMode) -> u8 {
55    match mode {
56        crate::storage::query::modes::QueryMode::Sql => 0,
57        crate::storage::query::modes::QueryMode::Gremlin => 1,
58        crate::storage::query::modes::QueryMode::Cypher => 2,
59        crate::storage::query::modes::QueryMode::Sparql => 3,
60        crate::storage::query::modes::QueryMode::Path => 4,
61        crate::storage::query::modes::QueryMode::Natural => 5,
62        crate::storage::query::modes::QueryMode::Unknown => 255,
63    }
64}
65
66fn mode_from_byte(byte: u8) -> Option<crate::storage::query::modes::QueryMode> {
67    match byte {
68        0 => Some(crate::storage::query::modes::QueryMode::Sql),
69        1 => Some(crate::storage::query::modes::QueryMode::Gremlin),
70        2 => Some(crate::storage::query::modes::QueryMode::Cypher),
71        3 => Some(crate::storage::query::modes::QueryMode::Sparql),
72        4 => Some(crate::storage::query::modes::QueryMode::Path),
73        5 => Some(crate::storage::query::modes::QueryMode::Natural),
74        255 => Some(crate::storage::query::modes::QueryMode::Unknown),
75        _ => None,
76    }
77}
78
79fn result_cache_static_str(value: &str) -> Option<&'static str> {
80    match value {
81        "select" => Some("select"),
82        "materialized-graph" => Some("materialized-graph"),
83        "runtime-red-schema" => Some("runtime-red-schema"),
84        "runtime-fdw" => Some("runtime-fdw"),
85        "runtime-table-rls" => Some("runtime-table-rls"),
86        "runtime-table" => Some("runtime-table"),
87        "runtime-join-rls" => Some("runtime-join-rls"),
88        "runtime-join" => Some("runtime-join"),
89        "runtime-vector" => Some("runtime-vector"),
90        "runtime-hybrid" => Some("runtime-hybrid"),
91        "runtime-secret" => Some("runtime-secret"),
92        "runtime-config" => Some("runtime-config"),
93        "runtime-tenant" => Some("runtime-tenant"),
94        "runtime-explain" => Some("runtime-explain"),
95        "runtime-tree" => Some("runtime-tree"),
96        "runtime-kv" => Some("runtime-kv"),
97        "runtime-queue" => Some("runtime-queue"),
98        _ => None,
99    }
100}
101
102fn write_u32(out: &mut Vec<u8>, value: usize) -> Option<()> {
103    let value = u32::try_from(value).ok()?;
104    out.extend_from_slice(&value.to_le_bytes());
105    Some(())
106}
107
108fn write_string(out: &mut Vec<u8>, value: &str) -> Option<()> {
109    write_u32(out, value.len())?;
110    out.extend_from_slice(value.as_bytes());
111    Some(())
112}
113
114fn write_bytes(out: &mut Vec<u8>, value: &[u8]) -> Option<()> {
115    write_u32(out, value.len())?;
116    out.extend_from_slice(value);
117    Some(())
118}
119
120fn read_u8(input: &mut &[u8]) -> Option<u8> {
121    let (&value, rest) = input.split_first()?;
122    *input = rest;
123    Some(value)
124}
125
126fn read_u32(input: &mut &[u8]) -> Option<usize> {
127    if input.len() < 4 {
128        return None;
129    }
130    let value = u32::from_le_bytes(input[..4].try_into().ok()?) as usize;
131    *input = &input[4..];
132    Some(value)
133}
134
135fn read_u64(input: &mut &[u8]) -> Option<u64> {
136    if input.len() < 8 {
137        return None;
138    }
139    let value = u64::from_le_bytes(input[..8].try_into().ok()?);
140    *input = &input[8..];
141    Some(value)
142}
143
144fn read_string(input: &mut &[u8]) -> Option<String> {
145    let len = read_u32(input)?;
146    if input.len() < len {
147        return None;
148    }
149    let value = String::from_utf8(input[..len].to_vec()).ok()?;
150    *input = &input[len..];
151    Some(value)
152}
153
154fn read_bytes<'a>(input: &mut &'a [u8]) -> Option<&'a [u8]> {
155    let len = read_u32(input)?;
156    if input.len() < len {
157        return None;
158    }
159    let value = &input[..len];
160    *input = &input[len..];
161    Some(value)
162}
163
164fn encode_result_cache_payload(entry: &RuntimeResultCacheEntry) -> Option<Vec<u8>> {
165    let result = &entry.result;
166    if result.result.pre_serialized_json.is_some()
167        || result_cache_static_str(result.statement).is_none()
168        || result_cache_static_str(result.engine).is_none()
169        || result_cache_static_str(result.statement_type).is_none()
170        || result.result.records.iter().any(|record| {
171            !record.nodes.is_empty()
172                || !record.edges.is_empty()
173                || !record.paths.is_empty()
174                || !record.vector_results.is_empty()
175        })
176    {
177        return None;
178    }
179
180    let mut out = Vec::new();
181    out.extend_from_slice(RESULT_CACHE_PAYLOAD_MAGIC);
182    write_string(&mut out, &result.query)?;
183    out.push(mode_to_byte(result.mode));
184    write_string(&mut out, result.statement)?;
185    write_string(&mut out, result.engine)?;
186    out.extend_from_slice(&result.affected_rows.to_le_bytes());
187    write_string(&mut out, result.statement_type)?;
188
189    write_u32(&mut out, result.result.columns.len())?;
190    for column in &result.result.columns {
191        write_string(&mut out, column)?;
192    }
193    out.extend_from_slice(&result.result.stats.nodes_scanned.to_le_bytes());
194    out.extend_from_slice(&result.result.stats.edges_scanned.to_le_bytes());
195    out.extend_from_slice(&result.result.stats.rows_scanned.to_le_bytes());
196    out.extend_from_slice(&result.result.stats.exec_time_us.to_le_bytes());
197
198    write_u32(&mut out, result.result.records.len())?;
199    for record in &result.result.records {
200        let fields = record.iter_fields().collect::<Vec<_>>();
201        write_u32(&mut out, fields.len())?;
202        for (name, value) in fields {
203            write_string(&mut out, name)?;
204            let mut encoded = Vec::new();
205            crate::storage::schema::value_codec::encode(value, &mut encoded);
206            write_bytes(&mut out, &encoded)?;
207        }
208    }
209
210    write_u32(&mut out, entry.scopes.len())?;
211    for scope in &entry.scopes {
212        write_string(&mut out, scope)?;
213    }
214    Some(out)
215}
216
217fn decode_result_cache_payload(mut input: &[u8]) -> Option<(RuntimeQueryResult, HashSet<String>)> {
218    if input.len() < RESULT_CACHE_PAYLOAD_MAGIC.len()
219        || &input[..RESULT_CACHE_PAYLOAD_MAGIC.len()] != RESULT_CACHE_PAYLOAD_MAGIC
220    {
221        return None;
222    }
223    input = &input[RESULT_CACHE_PAYLOAD_MAGIC.len()..];
224
225    let query = read_string(&mut input)?;
226    let mode = mode_from_byte(read_u8(&mut input)?)?;
227    let statement = result_cache_static_str(&read_string(&mut input)?)?;
228    let engine = result_cache_static_str(&read_string(&mut input)?)?;
229    let affected_rows = read_u64(&mut input)?;
230    let statement_type = result_cache_static_str(&read_string(&mut input)?)?;
231
232    let mut columns = Vec::new();
233    for _ in 0..read_u32(&mut input)? {
234        columns.push(read_string(&mut input)?);
235    }
236    let stats = crate::storage::query::unified::QueryStats {
237        nodes_scanned: read_u64(&mut input)?,
238        edges_scanned: read_u64(&mut input)?,
239        rows_scanned: read_u64(&mut input)?,
240        exec_time_us: read_u64(&mut input)?,
241    };
242
243    let mut records = Vec::new();
244    for _ in 0..read_u32(&mut input)? {
245        let mut record = crate::storage::query::unified::UnifiedRecord::new();
246        for _ in 0..read_u32(&mut input)? {
247            let name = read_string(&mut input)?;
248            let bytes = read_bytes(&mut input)?;
249            let (value, used) = crate::storage::schema::value_codec::decode(bytes).ok()?;
250            if used != bytes.len() {
251                return None;
252            }
253            record.set_owned(name, value);
254        }
255        records.push(record);
256    }
257
258    let mut scopes = HashSet::new();
259    for _ in 0..read_u32(&mut input)? {
260        scopes.insert(read_string(&mut input)?);
261    }
262    if !input.is_empty() {
263        return None;
264    }
265
266    Some((
267        RuntimeQueryResult {
268            query,
269            mode,
270            statement,
271            engine,
272            result: crate::storage::query::unified::UnifiedResult {
273                columns,
274                records,
275                stats,
276                pre_serialized_json: None,
277            },
278            affected_rows,
279            statement_type,
280            bookmark: None,
281            notice: None,
282        },
283        scopes,
284    ))
285}
286
287impl RedDBRuntime {
288    fn result_cache_backend(&self) -> RuntimeResultCacheBackend {
289        match self
290            .config_string(RESULT_CACHE_BACKEND_KEY, RESULT_CACHE_DEFAULT_BACKEND)
291            .as_str()
292        {
293            "blob_cache" => RuntimeResultCacheBackend::BlobCache,
294            "shadow" => RuntimeResultCacheBackend::Shadow,
295            _ => RuntimeResultCacheBackend::Legacy,
296        }
297    }
298
299    fn result_cache_enabled(&self) -> bool {
300        self.config_bool(RESULT_CACHE_ENABLED_KEY, true)
301    }
302
303    fn result_cache_ttl_secs(&self) -> u64 {
304        self.config_u64(RESULT_CACHE_TTL_KEY, RESULT_CACHE_TTL_SECS)
305    }
306
307    fn result_cache_capacity(&self) -> usize {
308        self.config_u64(RESULT_CACHE_CAPACITY_KEY, RESULT_CACHE_MAX_ENTRIES as u64) as usize
309    }
310
311    pub fn result_cache_metrics(&self) -> (u64, u64, u64) {
312        use std::sync::atomic::Ordering::Relaxed;
313        (
314            self.inner.result_cache_hits.load(Relaxed),
315            self.inner.result_cache_misses.load(Relaxed),
316            self.inner.result_cache_evictions.load(Relaxed),
317        )
318    }
319
320    fn record_result_cache_evictions(&self, evicted: u64) {
321        if evicted > 0 {
322            self.inner
323                .result_cache_evictions
324                .fetch_add(evicted, std::sync::atomic::Ordering::Relaxed);
325        }
326    }
327
328    pub(super) fn get_result_cache_entry(&self, key: &str) -> Option<RuntimeQueryResult> {
329        if !self.result_cache_enabled() {
330            return None;
331        }
332        let hit = self.get_result_cache_entry_inner(key);
333        let counter = if hit.is_some() {
334            &self.inner.result_cache_hits
335        } else {
336            &self.inner.result_cache_misses
337        };
338        counter.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
339        hit
340    }
341
342    fn get_result_cache_entry_inner(&self, key: &str) -> Option<RuntimeQueryResult> {
343        match self.result_cache_backend() {
344            RuntimeResultCacheBackend::Legacy => self.get_legacy_result_cache_entry(key),
345            RuntimeResultCacheBackend::BlobCache => self.get_blob_result_cache_entry(key),
346            RuntimeResultCacheBackend::Shadow => {
347                let legacy = self.get_legacy_result_cache_entry(key);
348                let blob = self.get_blob_result_cache_entry(key);
349                if let (Some(ref legacy), Some(ref blob)) = (&legacy, &blob) {
350                    if result_cache_fingerprint(legacy) != result_cache_fingerprint(blob) {
351                        self.inner
352                            .result_cache_shadow_divergences
353                            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
354                        tracing::warn!(
355                            key,
356                            metric = crate::runtime::METRIC_CACHE_SHADOW_DIVERGENCE_TOTAL,
357                            "result cache shadow backend diverged from legacy"
358                        );
359                    }
360                }
361                legacy
362            }
363        }
364    }
365
366    fn get_legacy_result_cache_entry(&self, key: &str) -> Option<RuntimeQueryResult> {
367        let ttl = self.result_cache_ttl_secs();
368        let cache = self.inner.result_cache.read();
369        cache.0.get(key).and_then(|entry| {
370            if entry.cached_at.elapsed().as_secs() < ttl {
371                Some(entry.result.clone())
372            } else {
373                None
374            }
375        })
376    }
377
378    fn get_blob_result_cache_entry(&self, key: &str) -> Option<RuntimeQueryResult> {
379        if self.inner.embedded_single_file {
380            if let Some(bytes) = self.get_embedded_result_cache_payload(key) {
381                let policy = crate::storage::cache::BlobCachePolicy::default()
382                    .ttl_ms(self.result_cache_ttl_secs() * 1000)
383                    .priority(200);
384                let put = crate::storage::cache::BlobCachePut::new(bytes).with_policy(policy);
385                if self
386                    .inner
387                    .result_blob_cache
388                    .put(RESULT_CACHE_BLOB_NAMESPACE, key, put)
389                    .is_err()
390                {
391                    return None;
392                }
393            }
394        }
395        self.get_blob_result_cache_entry_from_blob_cache(key)
396    }
397
398    fn get_blob_result_cache_entry_from_blob_cache(&self, key: &str) -> Option<RuntimeQueryResult> {
399        let hit = self
400            .inner
401            .result_blob_cache
402            .get(RESULT_CACHE_BLOB_NAMESPACE, key)?;
403        {
404            let cache = self.inner.result_blob_entries.read();
405            if let Some(entry) = cache.0.get(key) {
406                return Some(entry.result.clone());
407            }
408        }
409
410        let (result, scopes) = decode_result_cache_payload(hit.value())?;
411        let mut cache = self.inner.result_blob_entries.write();
412        let (ref mut map, ref mut order) = *cache;
413        let new_entry = RuntimeResultCacheEntry {
414            result: result.clone(),
415            cached_at: std::time::Instant::now(),
416            scopes,
417        };
418        // Single hash lookup: drive the LRU-order push_back only on Vacant.
419        match map.entry(key.to_string()) {
420            Entry::Occupied(mut slot) => {
421                slot.insert(new_entry);
422            }
423            Entry::Vacant(slot) => {
424                order.push_back(slot.key().clone());
425                slot.insert(new_entry);
426            }
427        }
428        let evicted = trim_result_cache(map, order, self.result_cache_capacity());
429        drop(cache);
430        self.record_result_cache_evictions(evicted);
431        Some(result)
432    }
433
434    fn get_embedded_result_cache_payload(&self, key: &str) -> Option<Vec<u8>> {
435        let manager = self
436            .inner
437            .db
438            .store()
439            .get_collection(RESULT_CACHE_EMBEDDED_COLLECTION)?;
440        let mut latest: Option<(u64, Vec<u8>)> = None;
441        manager.for_each_entity(|entity| {
442            let Some(row) = entity.data.as_row() else {
443                return true;
444            };
445            let namespace_matches = row.get_field("namespace").and_then(|value| match value {
446                crate::storage::schema::Value::Text(value) => Some(value.as_ref()),
447                _ => None,
448            }) == Some(RESULT_CACHE_BLOB_NAMESPACE);
449            let key_matches = row.get_field("key").and_then(|value| match value {
450                crate::storage::schema::Value::Text(value) => Some(value.as_ref()),
451                _ => None,
452            }) == Some(key);
453            if namespace_matches && key_matches {
454                if let Some(crate::storage::schema::Value::Blob(payload)) = row.get_field("payload")
455                {
456                    let id = entity.id.raw();
457                    if latest
458                        .as_ref()
459                        .is_none_or(|(latest_id, _)| id >= *latest_id)
460                    {
461                        latest = Some((id, payload.clone()));
462                    }
463                }
464            }
465            true
466        });
467        latest.map(|(_, payload)| payload)
468    }
469
470    pub(super) fn put_result_cache_entry(&self, key: &str, entry: RuntimeResultCacheEntry) {
471        if !self.result_cache_enabled() {
472            return;
473        }
474        match self.result_cache_backend() {
475            RuntimeResultCacheBackend::Legacy => self.put_legacy_result_cache_entry(key, entry),
476            RuntimeResultCacheBackend::BlobCache => self.put_blob_result_cache_entry(key, entry),
477            RuntimeResultCacheBackend::Shadow => {
478                self.put_legacy_result_cache_entry(key, entry.clone());
479                self.put_blob_result_cache_entry(key, entry);
480            }
481        }
482    }
483
484    fn put_legacy_result_cache_entry(&self, key: &str, entry: RuntimeResultCacheEntry) {
485        let capacity = self.result_cache_capacity();
486        let mut cache = self.inner.result_cache.write();
487        let (ref mut map, ref mut order) = *cache;
488        // Single hash lookup: drive the LRU-order push_back only on Vacant.
489        match map.entry(key.to_string()) {
490            Entry::Occupied(mut slot) => {
491                slot.insert(entry);
492            }
493            Entry::Vacant(slot) => {
494                order.push_back(slot.key().clone());
495                slot.insert(entry);
496            }
497        }
498        let evicted = trim_result_cache(map, order, capacity);
499        drop(cache);
500        self.record_result_cache_evictions(evicted);
501    }
502
503    fn put_blob_result_cache_entry(&self, key: &str, entry: RuntimeResultCacheEntry) {
504        let policy = crate::storage::cache::BlobCachePolicy::default()
505            .ttl_ms(self.result_cache_ttl_secs() * 1000)
506            .priority(200);
507        let dependencies = entry.scopes.iter().cloned().collect::<Vec<_>>();
508        let bytes = encode_result_cache_payload(&entry)
509            .unwrap_or_else(|| result_cache_fingerprint(&entry.result).into_bytes());
510        if self.inner.embedded_single_file {
511            self.put_embedded_result_cache_payload(key, &bytes, &dependencies);
512        }
513        let put = crate::storage::cache::BlobCachePut::new(bytes)
514            .with_dependencies(dependencies)
515            .with_policy(policy);
516        if self
517            .inner
518            .result_blob_cache
519            .put(RESULT_CACHE_BLOB_NAMESPACE, key, put)
520            .is_err()
521        {
522            return;
523        }
524
525        let capacity = self.result_cache_capacity();
526        let mut cache = self.inner.result_blob_entries.write();
527        let (ref mut map, ref mut order) = *cache;
528        // Single hash lookup: drive the LRU-order push_back only on Vacant.
529        match map.entry(key.to_string()) {
530            Entry::Occupied(mut slot) => {
531                slot.insert(entry);
532            }
533            Entry::Vacant(slot) => {
534                order.push_back(slot.key().clone());
535                slot.insert(entry);
536            }
537        }
538        let evicted = trim_result_cache(map, order, capacity);
539        drop(cache);
540        self.record_result_cache_evictions(evicted);
541    }
542
543    fn put_embedded_result_cache_payload(&self, key: &str, bytes: &[u8], scopes: &[String]) {
544        let store = self.inner.db.store();
545        let _ = store.get_or_create_collection(RESULT_CACHE_EMBEDDED_COLLECTION);
546        let entity = crate::storage::UnifiedEntity::new(
547            crate::storage::EntityId::new(0),
548            crate::storage::EntityKind::TableRow {
549                table: std::sync::Arc::from(RESULT_CACHE_EMBEDDED_COLLECTION),
550                row_id: 0,
551            },
552            crate::storage::EntityData::Row(crate::storage::RowData {
553                columns: Vec::new(),
554                named: Some(HashMap::from([
555                    (
556                        "namespace".to_string(),
557                        crate::storage::schema::Value::text(RESULT_CACHE_BLOB_NAMESPACE),
558                    ),
559                    (
560                        "key".to_string(),
561                        crate::storage::schema::Value::text(key.to_string()),
562                    ),
563                    (
564                        "payload".to_string(),
565                        crate::storage::schema::Value::Blob(bytes.to_vec()),
566                    ),
567                    (
568                        "scopes".to_string(),
569                        crate::storage::schema::Value::text(scopes.join("\n")),
570                    ),
571                ])),
572                schema: None,
573            }),
574        );
575        let _ = store.insert_auto(RESULT_CACHE_EMBEDDED_COLLECTION, entity);
576    }
577
578    fn invalidate_embedded_result_cache(&self) {
579        let Some(manager) = self
580            .inner
581            .db
582            .store()
583            .get_collection(RESULT_CACHE_EMBEDDED_COLLECTION)
584        else {
585            return;
586        };
587        let ids = manager
588            .query_all(|_| true)
589            .into_iter()
590            .map(|entity| entity.id)
591            .collect::<Vec<_>>();
592        if !ids.is_empty() {
593            let _ = self
594                .inner
595                .db
596                .store()
597                .delete_batch(RESULT_CACHE_EMBEDDED_COLLECTION, &ids);
598        }
599    }
600
601    fn invalidate_embedded_result_cache_for_scope(&self, scope: &str) {
602        let Some(manager) = self
603            .inner
604            .db
605            .store()
606            .get_collection(RESULT_CACHE_EMBEDDED_COLLECTION)
607        else {
608            return;
609        };
610        let ids = manager
611            .query_all(|entity| {
612                entity
613                    .data
614                    .as_row()
615                    .and_then(|row| row.get_field("scopes"))
616                    .and_then(|value| match value {
617                        crate::storage::schema::Value::Text(value) => Some(value.as_ref()),
618                        _ => None,
619                    })
620                    .is_some_and(|scopes| scopes.lines().any(|entry| entry == scope))
621            })
622            .into_iter()
623            .map(|entity| entity.id)
624            .collect::<Vec<_>>();
625        if !ids.is_empty() {
626            let _ = self
627                .inner
628                .db
629                .store()
630                .delete_batch(RESULT_CACHE_EMBEDDED_COLLECTION, &ids);
631        }
632    }
633
634    pub fn result_cache_shadow_divergences(&self) -> u64 {
635        self.inner
636            .result_cache_shadow_divergences
637            .load(std::sync::atomic::Ordering::Relaxed)
638    }
639
640    pub fn invalidate_result_cache(&self) {
641        self.invalidate_result_cache_process_only();
642        if self.inner.embedded_single_file {
643            self.invalidate_embedded_result_cache();
644        }
645    }
646
647    pub(crate) fn invalidate_result_cache_process_only(&self) {
648        let mut cache = self.inner.result_cache.write();
649        cache.0.clear();
650        cache.1.clear();
651        let mut blob_entries = self.inner.result_blob_entries.write();
652        blob_entries.0.clear();
653        blob_entries.1.clear();
654        self.inner
655            .result_blob_cache
656            .invalidate_namespace(RESULT_CACHE_BLOB_NAMESPACE);
657        let mut ask_entries = self.inner.ask_answer_cache_entries.write();
658        ask_entries.0.clear();
659        ask_entries.1.clear();
660        self.inner
661            .result_blob_cache
662            .invalidate_namespace(super::ASK_ANSWER_CACHE_NAMESPACE);
663    }
664
665    pub(crate) fn invalidate_result_cache_for_table(&self, table: &str) {
666        let legacy_has_match = {
667            let cache = self.inner.result_cache.read();
668            let (ref map, _) = *cache;
669            !map.is_empty() && map.values().any(|entry| entry.scopes.contains(table))
670        };
671        let blob_has_match = {
672            let cache = self.inner.result_blob_entries.read();
673            let (ref map, _) = *cache;
674            !map.is_empty() && map.values().any(|entry| entry.scopes.contains(table))
675        };
676        if legacy_has_match {
677            let mut cache = self.inner.result_cache.write();
678            let (ref mut map, ref mut order) = *cache;
679            map.retain(|_, entry| !entry.scopes.contains(table));
680            order.retain(|key| map.contains_key(key));
681        }
682
683        if matches!(
684            self.result_cache_backend(),
685            RuntimeResultCacheBackend::BlobCache | RuntimeResultCacheBackend::Shadow
686        ) {
687            let mut blob_entries = self.inner.result_blob_entries.write();
688            let (ref mut blob_map, ref mut blob_order) = *blob_entries;
689            blob_map.clear();
690            blob_order.clear();
691            self.inner
692                .result_blob_cache
693                .invalidate_namespace(RESULT_CACHE_BLOB_NAMESPACE);
694            if self.inner.embedded_single_file {
695                self.invalidate_embedded_result_cache();
696            }
697        } else if blob_has_match {
698            let mut blob_entries = self.inner.result_blob_entries.write();
699            let (ref mut blob_map, ref mut blob_order) = *blob_entries;
700            blob_map.retain(|_, entry| !entry.scopes.contains(table));
701            blob_order.retain(|key| blob_map.contains_key(key));
702            if self.inner.embedded_single_file {
703                self.invalidate_embedded_result_cache_for_scope(table);
704            }
705        }
706        let mut ask_entries = self.inner.ask_answer_cache_entries.write();
707        ask_entries.0.clear();
708        ask_entries.1.clear();
709        self.inner
710            .result_blob_cache
711            .invalidate_namespace(super::ASK_ANSWER_CACHE_NAMESPACE);
712    }
713}
reddb_server/runtime/result_cache_runtime.rs

reddb_server/runtime/
result_cache_runtime.rs
