reddb_server/runtime/
result_cache_runtime.rs

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

reddb_server/runtime/
result_cache_runtime.rs
