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.segments_total.to_le_bytes());
197    out.extend_from_slice(&result.result.stats.segments_scanned.to_le_bytes());
198    out.extend_from_slice(&result.result.stats.segments_pruned.to_le_bytes());
199    out.extend_from_slice(&result.result.stats.exec_time_us.to_le_bytes());
200
201    write_u32(&mut out, result.result.records.len())?;
202    for record in &result.result.records {
203        let fields = record.iter_fields().collect::<Vec<_>>();
204        write_u32(&mut out, fields.len())?;
205        for (name, value) in fields {
206            write_string(&mut out, name)?;
207            let mut encoded = Vec::new();
208            crate::storage::schema::value_codec::encode(value, &mut encoded);
209            write_bytes(&mut out, &encoded)?;
210        }
211    }
212
213    write_u32(&mut out, entry.scopes.len())?;
214    for scope in &entry.scopes {
215        write_string(&mut out, scope)?;
216    }
217    Some(out)
218}
219
220fn decode_result_cache_payload(mut input: &[u8]) -> Option<(RuntimeQueryResult, HashSet<String>)> {
221    if input.len() < RESULT_CACHE_PAYLOAD_MAGIC.len()
222        || &input[..RESULT_CACHE_PAYLOAD_MAGIC.len()] != RESULT_CACHE_PAYLOAD_MAGIC
223    {
224        return None;
225    }
226    input = &input[RESULT_CACHE_PAYLOAD_MAGIC.len()..];
227
228    let query = read_string(&mut input)?;
229    let mode = mode_from_byte(read_u8(&mut input)?)?;
230    let statement = result_cache_static_str(&read_string(&mut input)?)?;
231    let engine = result_cache_static_str(&read_string(&mut input)?)?;
232    let affected_rows = read_u64(&mut input)?;
233    let statement_type = result_cache_static_str(&read_string(&mut input)?)?;
234
235    let mut columns = Vec::new();
236    for _ in 0..read_u32(&mut input)? {
237        columns.push(read_string(&mut input)?);
238    }
239    let stats = crate::storage::query::unified::QueryStats {
240        nodes_scanned: read_u64(&mut input)?,
241        edges_scanned: read_u64(&mut input)?,
242        rows_scanned: read_u64(&mut input)?,
243        segments_total: read_u64(&mut input)?,
244        segments_scanned: read_u64(&mut input)?,
245        segments_pruned: read_u64(&mut input)?,
246        exec_time_us: read_u64(&mut input)?,
247    };
248
249    let mut records = Vec::new();
250    for _ in 0..read_u32(&mut input)? {
251        let mut record = crate::storage::query::unified::UnifiedRecord::new();
252        for _ in 0..read_u32(&mut input)? {
253            let name = read_string(&mut input)?;
254            let bytes = read_bytes(&mut input)?;
255            let (value, used) = crate::storage::schema::value_codec::decode(bytes).ok()?;
256            if used != bytes.len() {
257                return None;
258            }
259            record.set_owned(name, value);
260        }
261        records.push(record);
262    }
263
264    let mut scopes = HashSet::new();
265    for _ in 0..read_u32(&mut input)? {
266        scopes.insert(read_string(&mut input)?);
267    }
268    if !input.is_empty() {
269        return None;
270    }
271
272    Some((
273        RuntimeQueryResult {
274            query,
275            mode,
276            statement,
277            engine,
278            result: crate::storage::query::unified::UnifiedResult {
279                columns,
280                records,
281                stats,
282                pre_serialized_json: None,
283            },
284            affected_rows,
285            statement_type,
286            bookmark: None,
287            notice: None,
288        },
289        scopes,
290    ))
291}
292
293impl RedDBRuntime {
294    fn result_cache_backend(&self) -> RuntimeResultCacheBackend {
295        match self
296            .config_string(RESULT_CACHE_BACKEND_KEY, RESULT_CACHE_DEFAULT_BACKEND)
297            .as_str()
298        {
299            "blob_cache" => RuntimeResultCacheBackend::BlobCache,
300            "shadow" => RuntimeResultCacheBackend::Shadow,
301            _ => RuntimeResultCacheBackend::Legacy,
302        }
303    }
304
305    fn result_cache_enabled(&self) -> bool {
306        self.config_bool(RESULT_CACHE_ENABLED_KEY, true)
307    }
308
309    fn result_cache_ttl_secs(&self) -> u64 {
310        self.config_u64(RESULT_CACHE_TTL_KEY, RESULT_CACHE_TTL_SECS)
311    }
312
313    fn result_cache_capacity(&self) -> usize {
314        self.config_u64(RESULT_CACHE_CAPACITY_KEY, RESULT_CACHE_MAX_ENTRIES as u64) as usize
315    }
316
317    pub fn result_cache_metrics(&self) -> (u64, u64, u64) {
318        use std::sync::atomic::Ordering::Relaxed;
319        (
320            self.inner.result_cache_hits.load(Relaxed),
321            self.inner.result_cache_misses.load(Relaxed),
322            self.inner.result_cache_evictions.load(Relaxed),
323        )
324    }
325
326    fn record_result_cache_evictions(&self, evicted: u64) {
327        if evicted > 0 {
328            self.inner
329                .result_cache_evictions
330                .fetch_add(evicted, std::sync::atomic::Ordering::Relaxed);
331        }
332    }
333
334    pub(super) fn get_result_cache_entry(&self, key: &str) -> Option<RuntimeQueryResult> {
335        if !self.result_cache_enabled() {
336            return None;
337        }
338        let hit = self.get_result_cache_entry_inner(key);
339        let counter = if hit.is_some() {
340            &self.inner.result_cache_hits
341        } else {
342            &self.inner.result_cache_misses
343        };
344        counter.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
345        hit
346    }
347
348    fn get_result_cache_entry_inner(&self, key: &str) -> Option<RuntimeQueryResult> {
349        match self.result_cache_backend() {
350            RuntimeResultCacheBackend::Legacy => self.get_legacy_result_cache_entry(key),
351            RuntimeResultCacheBackend::BlobCache => self.get_blob_result_cache_entry(key),
352            RuntimeResultCacheBackend::Shadow => {
353                let legacy = self.get_legacy_result_cache_entry(key);
354                let blob = self.get_blob_result_cache_entry(key);
355                if let (Some(ref legacy), Some(ref blob)) = (&legacy, &blob) {
356                    if result_cache_fingerprint(legacy) != result_cache_fingerprint(blob) {
357                        self.inner
358                            .result_cache_shadow_divergences
359                            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
360                        tracing::warn!(
361                            key,
362                            metric = crate::runtime::METRIC_CACHE_SHADOW_DIVERGENCE_TOTAL,
363                            "result cache shadow backend diverged from legacy"
364                        );
365                    }
366                }
367                legacy
368            }
369        }
370    }
371
372    fn get_legacy_result_cache_entry(&self, key: &str) -> Option<RuntimeQueryResult> {
373        let ttl = self.result_cache_ttl_secs();
374        let cache = self.inner.result_cache.read();
375        cache.0.get(key).and_then(|entry| {
376            if entry.cached_at.elapsed().as_secs() < ttl {
377                Some(entry.result.clone())
378            } else {
379                None
380            }
381        })
382    }
383
384    fn get_blob_result_cache_entry(&self, key: &str) -> Option<RuntimeQueryResult> {
385        if self.inner.embedded_single_file {
386            if let Some(bytes) = self.get_embedded_result_cache_payload(key) {
387                let policy = crate::storage::cache::BlobCachePolicy::default()
388                    .ttl_ms(self.result_cache_ttl_secs() * 1000)
389                    .priority(200);
390                let put = crate::storage::cache::BlobCachePut::new(bytes).with_policy(policy);
391                if self
392                    .inner
393                    .result_blob_cache
394                    .put(RESULT_CACHE_BLOB_NAMESPACE, key, put)
395                    .is_err()
396                {
397                    return None;
398                }
399            }
400        }
401        self.get_blob_result_cache_entry_from_blob_cache(key)
402    }
403
404    fn get_blob_result_cache_entry_from_blob_cache(&self, key: &str) -> Option<RuntimeQueryResult> {
405        let hit = self
406            .inner
407            .result_blob_cache
408            .get(RESULT_CACHE_BLOB_NAMESPACE, key)?;
409        {
410            let cache = self.inner.result_blob_entries.read();
411            if let Some(entry) = cache.0.get(key) {
412                return Some(entry.result.clone());
413            }
414        }
415
416        let (result, scopes) = decode_result_cache_payload(hit.value())?;
417        let mut cache = self.inner.result_blob_entries.write();
418        let (ref mut map, ref mut order) = *cache;
419        let new_entry = RuntimeResultCacheEntry {
420            result: result.clone(),
421            cached_at: std::time::Instant::now(),
422            scopes,
423        };
424        // Single hash lookup: drive the LRU-order push_back only on Vacant.
425        match map.entry(key.to_string()) {
426            Entry::Occupied(mut slot) => {
427                slot.insert(new_entry);
428            }
429            Entry::Vacant(slot) => {
430                order.push_back(slot.key().clone());
431                slot.insert(new_entry);
432            }
433        }
434        let evicted = trim_result_cache(map, order, self.result_cache_capacity());
435        drop(cache);
436        self.record_result_cache_evictions(evicted);
437        Some(result)
438    }
439
440    fn get_embedded_result_cache_payload(&self, key: &str) -> Option<Vec<u8>> {
441        let manager = self
442            .inner
443            .db
444            .store()
445            .get_collection(RESULT_CACHE_EMBEDDED_COLLECTION)?;
446        let mut latest: Option<(u64, Vec<u8>)> = None;
447        manager.for_each_entity(|entity| {
448            let Some(row) = entity.data.as_row() else {
449                return true;
450            };
451            let namespace_matches = row.get_field("namespace").and_then(|value| match value {
452                crate::storage::schema::Value::Text(value) => Some(value.as_ref()),
453                _ => None,
454            }) == Some(RESULT_CACHE_BLOB_NAMESPACE);
455            let key_matches = row.get_field("key").and_then(|value| match value {
456                crate::storage::schema::Value::Text(value) => Some(value.as_ref()),
457                _ => None,
458            }) == Some(key);
459            if namespace_matches && key_matches {
460                if let Some(crate::storage::schema::Value::Blob(payload)) = row.get_field("payload")
461                {
462                    let id = entity.id.raw();
463                    if latest
464                        .as_ref()
465                        .is_none_or(|(latest_id, _)| id >= *latest_id)
466                    {
467                        latest = Some((id, payload.clone()));
468                    }
469                }
470            }
471            true
472        });
473        latest.map(|(_, payload)| payload)
474    }
475
476    pub(super) fn put_result_cache_entry(&self, key: &str, entry: RuntimeResultCacheEntry) {
477        if !self.result_cache_enabled() {
478            return;
479        }
480        match self.result_cache_backend() {
481            RuntimeResultCacheBackend::Legacy => self.put_legacy_result_cache_entry(key, entry),
482            RuntimeResultCacheBackend::BlobCache => self.put_blob_result_cache_entry(key, entry),
483            RuntimeResultCacheBackend::Shadow => {
484                self.put_legacy_result_cache_entry(key, entry.clone());
485                self.put_blob_result_cache_entry(key, entry);
486            }
487        }
488    }
489
490    fn put_legacy_result_cache_entry(&self, key: &str, entry: RuntimeResultCacheEntry) {
491        let capacity = self.result_cache_capacity();
492        let mut cache = self.inner.result_cache.write();
493        let (ref mut map, ref mut order) = *cache;
494        // Single hash lookup: drive the LRU-order push_back only on Vacant.
495        match map.entry(key.to_string()) {
496            Entry::Occupied(mut slot) => {
497                slot.insert(entry);
498            }
499            Entry::Vacant(slot) => {
500                order.push_back(slot.key().clone());
501                slot.insert(entry);
502            }
503        }
504        let evicted = trim_result_cache(map, order, capacity);
505        drop(cache);
506        self.record_result_cache_evictions(evicted);
507    }
508
509    fn put_blob_result_cache_entry(&self, key: &str, entry: RuntimeResultCacheEntry) {
510        let policy = crate::storage::cache::BlobCachePolicy::default()
511            .ttl_ms(self.result_cache_ttl_secs() * 1000)
512            .priority(200);
513        let dependencies = entry.scopes.iter().cloned().collect::<Vec<_>>();
514        let bytes = encode_result_cache_payload(&entry)
515            .unwrap_or_else(|| result_cache_fingerprint(&entry.result).into_bytes());
516        if self.inner.embedded_single_file {
517            self.put_embedded_result_cache_payload(key, &bytes, &dependencies);
518        }
519        let put = crate::storage::cache::BlobCachePut::new(bytes)
520            .with_dependencies(dependencies)
521            .with_policy(policy);
522        if self
523            .inner
524            .result_blob_cache
525            .put(RESULT_CACHE_BLOB_NAMESPACE, key, put)
526            .is_err()
527        {
528            return;
529        }
530
531        let capacity = self.result_cache_capacity();
532        let mut cache = self.inner.result_blob_entries.write();
533        let (ref mut map, ref mut order) = *cache;
534        // Single hash lookup: drive the LRU-order push_back only on Vacant.
535        match map.entry(key.to_string()) {
536            Entry::Occupied(mut slot) => {
537                slot.insert(entry);
538            }
539            Entry::Vacant(slot) => {
540                order.push_back(slot.key().clone());
541                slot.insert(entry);
542            }
543        }
544        let evicted = trim_result_cache(map, order, capacity);
545        drop(cache);
546        self.record_result_cache_evictions(evicted);
547    }
548
549    fn put_embedded_result_cache_payload(&self, key: &str, bytes: &[u8], scopes: &[String]) {
550        let store = self.inner.db.store();
551        let _ = store.get_or_create_collection(RESULT_CACHE_EMBEDDED_COLLECTION);
552        let entity = crate::storage::UnifiedEntity::new(
553            crate::storage::EntityId::new(0),
554            crate::storage::EntityKind::TableRow {
555                table: std::sync::Arc::from(RESULT_CACHE_EMBEDDED_COLLECTION),
556                row_id: 0,
557            },
558            crate::storage::EntityData::Row(crate::storage::RowData {
559                columns: Vec::new(),
560                named: Some(HashMap::from([
561                    (
562                        "namespace".to_string(),
563                        crate::storage::schema::Value::text(RESULT_CACHE_BLOB_NAMESPACE),
564                    ),
565                    (
566                        "key".to_string(),
567                        crate::storage::schema::Value::text(key.to_string()),
568                    ),
569                    (
570                        "payload".to_string(),
571                        crate::storage::schema::Value::Blob(bytes.to_vec()),
572                    ),
573                    (
574                        "scopes".to_string(),
575                        crate::storage::schema::Value::text(scopes.join("\n")),
576                    ),
577                ])),
578                schema: None,
579            }),
580        );
581        let _ = store.insert_auto(RESULT_CACHE_EMBEDDED_COLLECTION, entity);
582    }
583
584    fn invalidate_embedded_result_cache(&self) {
585        let Some(manager) = self
586            .inner
587            .db
588            .store()
589            .get_collection(RESULT_CACHE_EMBEDDED_COLLECTION)
590        else {
591            return;
592        };
593        let ids = manager
594            .query_all(|_| true)
595            .into_iter()
596            .map(|entity| entity.id)
597            .collect::<Vec<_>>();
598        if !ids.is_empty() {
599            let _ = self
600                .inner
601                .db
602                .store()
603                .delete_batch(RESULT_CACHE_EMBEDDED_COLLECTION, &ids);
604        }
605    }
606
607    fn invalidate_embedded_result_cache_for_scope(&self, scope: &str) {
608        let Some(manager) = self
609            .inner
610            .db
611            .store()
612            .get_collection(RESULT_CACHE_EMBEDDED_COLLECTION)
613        else {
614            return;
615        };
616        let ids = manager
617            .query_all(|entity| {
618                entity
619                    .data
620                    .as_row()
621                    .and_then(|row| row.get_field("scopes"))
622                    .and_then(|value| match value {
623                        crate::storage::schema::Value::Text(value) => Some(value.as_ref()),
624                        _ => None,
625                    })
626                    .is_some_and(|scopes| scopes.lines().any(|entry| entry == scope))
627            })
628            .into_iter()
629            .map(|entity| entity.id)
630            .collect::<Vec<_>>();
631        if !ids.is_empty() {
632            let _ = self
633                .inner
634                .db
635                .store()
636                .delete_batch(RESULT_CACHE_EMBEDDED_COLLECTION, &ids);
637        }
638    }
639
640    pub fn result_cache_shadow_divergences(&self) -> u64 {
641        self.inner
642            .result_cache_shadow_divergences
643            .load(std::sync::atomic::Ordering::Relaxed)
644    }
645
646    pub fn invalidate_result_cache(&self) {
647        self.invalidate_result_cache_process_only();
648        if self.inner.embedded_single_file {
649            self.invalidate_embedded_result_cache();
650        }
651    }
652
653    pub(crate) fn invalidate_result_cache_process_only(&self) {
654        let mut cache = self.inner.result_cache.write();
655        cache.0.clear();
656        cache.1.clear();
657        let mut blob_entries = self.inner.result_blob_entries.write();
658        blob_entries.0.clear();
659        blob_entries.1.clear();
660        self.inner
661            .result_blob_cache
662            .invalidate_namespace(RESULT_CACHE_BLOB_NAMESPACE);
663        let mut ask_entries = self.inner.ask_answer_cache_entries.write();
664        ask_entries.0.clear();
665        ask_entries.1.clear();
666        self.inner
667            .result_blob_cache
668            .invalidate_namespace(super::ASK_ANSWER_CACHE_NAMESPACE);
669    }
670
671    pub(crate) fn invalidate_result_cache_for_table(&self, table: &str) {
672        let legacy_has_match = {
673            let cache = self.inner.result_cache.read();
674            let (ref map, _) = *cache;
675            !map.is_empty() && map.values().any(|entry| entry.scopes.contains(table))
676        };
677        let blob_has_match = {
678            let cache = self.inner.result_blob_entries.read();
679            let (ref map, _) = *cache;
680            !map.is_empty() && map.values().any(|entry| entry.scopes.contains(table))
681        };
682        if legacy_has_match {
683            let mut cache = self.inner.result_cache.write();
684            let (ref mut map, ref mut order) = *cache;
685            map.retain(|_, entry| !entry.scopes.contains(table));
686            order.retain(|key| map.contains_key(key));
687        }
688
689        if matches!(
690            self.result_cache_backend(),
691            RuntimeResultCacheBackend::BlobCache | RuntimeResultCacheBackend::Shadow
692        ) {
693            let mut blob_entries = self.inner.result_blob_entries.write();
694            let (ref mut blob_map, ref mut blob_order) = *blob_entries;
695            blob_map.clear();
696            blob_order.clear();
697            self.inner
698                .result_blob_cache
699                .invalidate_namespace(RESULT_CACHE_BLOB_NAMESPACE);
700            if self.inner.embedded_single_file {
701                self.invalidate_embedded_result_cache();
702            }
703        } else if blob_has_match {
704            let mut blob_entries = self.inner.result_blob_entries.write();
705            let (ref mut blob_map, ref mut blob_order) = *blob_entries;
706            blob_map.retain(|_, entry| !entry.scopes.contains(table));
707            blob_order.retain(|key| blob_map.contains_key(key));
708            if self.inner.embedded_single_file {
709                self.invalidate_embedded_result_cache_for_scope(table);
710            }
711        }
712        let mut ask_entries = self.inner.ask_answer_cache_entries.write();
713        ask_entries.0.clear();
714        ask_entries.1.clear();
715        self.inner
716            .result_blob_cache
717            .invalidate_namespace(super::ASK_ANSWER_CACHE_NAMESPACE);
718    }
719}
reddb_server/runtime/result_cache_runtime.rs

reddb_server/runtime/
result_cache_runtime.rs
