kglite 0.13.2

Pure-Rust knowledge graph engine — Cypher pipeline, snapshot/working CoW transactions, columnar/mmap/disk storage backends, optional dataset loaders (SEC EDGAR, Sodir, Wikidata). PyO3 wrappers live in the sibling kglite-py crate (the Python wheel); embeddable directly from any Rust binary without PyO3 in the dep tree.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

//! Lazy derived indexes for the default in-memory backend.

use super::{MemoryGraph, MemoryPeerCounts};
use crate::graph::schema::InternedKey;
use petgraph::visit::{EdgeRef, IntoEdgeReferences};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Instant;

impl MemoryGraph {
    /// Drop derived edge counts after any mutation that can change edge type,
    /// identity, or endpoints.
    pub(crate) fn invalidate_peer_counts(&mut self) {
        if let Ok(mut cache) = self.peer_counts.write() {
            cache.clear();
        }
    }

    /// Fetch or build source/target counts for one relationship type.
    pub(crate) fn ensure_peer_counts(&self, conn_type: InternedKey) -> Arc<MemoryPeerCounts> {
        self.ensure_peer_counts_with_deadline(conn_type, None)
            .expect("peer-count build without a deadline cannot time out")
    }

    /// Deadline-aware form used by query execution. A completed cache lookup
    /// is constant-time; only the initial edge scan needs periodic checks.
    pub(crate) fn ensure_peer_counts_with_deadline(
        &self,
        conn_type: InternedKey,
        deadline: Option<Instant>,
    ) -> Result<Arc<MemoryPeerCounts>, String> {
        let key = conn_type.as_u64();
        if let Ok(cache) = self.peer_counts.read() {
            if let Some(counts) = cache.get(&key) {
                return Ok(Arc::clone(counts));
            }
        }

        let mut by_target = HashMap::new();
        let mut by_source = HashMap::new();
        for (edge_idx, edge) in self.inner.edge_references().enumerate() {
            if edge_idx.is_multiple_of(1 << 20) && deadline.is_some_and(|dl| Instant::now() > dl) {
                return Err("Query timed out".to_string());
            }
            if edge.weight().connection_type != conn_type {
                continue;
            }
            *by_target.entry(edge.target().index() as u32).or_insert(0) += 1;
            *by_source.entry(edge.source().index() as u32).or_insert(0) += 1;
        }
        let built = Arc::new(MemoryPeerCounts {
            by_target: Arc::new(by_target),
            by_source: Arc::new(by_source),
        });
        let mut cache = match self.peer_counts.write() {
            Ok(cache) => cache,
            Err(_) => return Ok(built),
        };
        Ok(Arc::clone(cache.entry(key).or_insert(built)))
    }
}