urna_format/sections/graph/
adjacency.rs

1//! chunk-to-chunk csr adjacency codec for SECTION_GRAPH_ADJACENCY (0x0C).
2//!
3//! OPTIONAL and EXCLUDED from content_hash (not in CANONICAL_SECTIONS), so
4//! adding a graph never invalidates a urna:// citation. the wire encoding is
5//! `raw`: one self-describing payload that reuses the shared `intpack` codec
6//! (encoding id 4) for its integer columns, exactly like the hnsw section
7//! (0x07). nothing here touches the embedding hot path. all integers le.
8//!
9//! payload: [u32 version=1][u64 n_nodes][intpack offsets, len n_nodes+1]
10//! [intpack delta-gapped neighbor-id blob][edge-type column: u8 kind then
11//! either one iso scalar (falkordb iso, kind 0) or an intpack u8 run of
12//! len total_edges (kind 1)]. canonical edge order for byte-identical output
13//! is ascending (src, edge_type, dst); [`encode_graph_adjacency`] sorts, so
14//! two builds of the same edge set produce byte-identical bytes. typed
15//! errors on truncation/hostile claims, never panics.
16
17use crate::bytes::{le_u32, le_u64};
18use crate::encoding::pack_u64s;
19use crate::error::UrnaError;
20use crate::layout::SECTION_GRAPH_ADJACENCY;
21
22/// internal payload version for the csr layout. bumped on an internal
23/// layout change, NEVER `URNA_FORMAT_VERSION` (the same lane-c discipline
24/// as `HNSW_PAYLOAD_VERSION`).
25pub const GRAPH_ADJACENCY_PAYLOAD_VERSION: u32 = 1;
26
27/// edge type ids. a u8 column (iso single scalar when uniform).
28pub const EDGE_TYPE_NEXT_CHUNK: u8 = 0;
29pub const EDGE_TYPE_SEMANTIC: u8 = 1;
30pub const EDGE_TYPE_CITATION: u8 = 2;
31
32/// edge-type column kinds (the leading byte of the edge-type column).
33const EDGE_COL_ISO: u8 = 0;
34const EDGE_COL_RUN: u8 = 1;
35
36/// hostile-input cap on a single node's out-degree. a real chunk graph
37/// (NEXT_CHUNK + top-m semantic + citations) never approaches this, so it
38/// never rejects a valid file; it bounds an untrusted decoded degree before
39/// any allocation from a claim alone.
40pub const MAX_DEGREE: u64 = 1 << 20;
41
42/// one directed edge: `src -> dst` of a given `edge_type`.
43#[derive(Clone, Copy, Debug, PartialEq, Eq)]
44pub struct Edge {
45    pub src: u32,
46    pub dst: u32,
47    pub edge_type: u8,
48}
49
50fn malformed(reason: impl Into<String>) -> UrnaError {
51    UrnaError::MalformedSectionPayload {
52        section_id: SECTION_GRAPH_ADJACENCY,
53        reason: reason.into(),
54    }
55}
56
57/// encode `edges` over `n_nodes` chunk ordinals into the csr payload, byte-
58/// identical for the same edge set (canonical sort). every `src`/`dst` < n.
59pub fn encode_graph_adjacency(n_nodes: usize, edges: &[Edge]) -> Result<Vec<u8>, UrnaError> {
60    for e in edges {
61        if e.src as usize >= n_nodes || e.dst as usize >= n_nodes {
62            return Err(malformed(format!(
63                "edge ({},{}) out of range for n_nodes {}",
64                e.src, e.dst, n_nodes
65            )));
66        }
67    }
68    // canonical order: ascending (src, edge_type, dst). byte-identical.
69    let mut sorted = edges.to_vec();
70    sorted.sort_unstable_by(|a, b| {
71        a.src
72            .cmp(&b.src)
73            .then(a.edge_type.cmp(&b.edge_type))
74            .then(a.dst.cmp(&b.dst))
75    });
76
77    // csr offsets (len n_nodes+1) + delta-gapped dst column + edge-type col.
78    let mut offsets: Vec<u64> = Vec::with_capacity(n_nodes + 1);
79    let mut dst_gaps: Vec<u64> = Vec::with_capacity(sorted.len());
80    let mut types: Vec<u8> = Vec::with_capacity(sorted.len());
81    offsets.push(0);
82    let mut cursor = 0usize;
83    for node in 0..n_nodes {
84        // delta-gap dst within each (src, edge_type) run. the canonical sort
85        // makes dst ascending inside a run, so every gap is non-negative; the
86        // first dst of a run is absolute. resetting at each edge_type boundary
87        // keeps decode a plain prefix-sum (no backward jumps across types).
88        let mut prev_dst: Option<u32> = None;
89        let mut prev_type: Option<u8> = None;
90        while cursor < sorted.len() && sorted[cursor].src as usize == node {
91            let e = sorted[cursor];
92            if prev_type != Some(e.edge_type) {
93                prev_dst = None;
94                prev_type = Some(e.edge_type);
95            }
96            let gap = match prev_dst {
97                Some(p) => (e.dst - p) as u64,
98                None => e.dst as u64,
99            };
100            dst_gaps.push(gap);
101            types.push(e.edge_type);
102            prev_dst = Some(e.dst);
103            cursor += 1;
104        }
105        offsets.push(dst_gaps.len() as u64);
106    }
107
108    let mut out = Vec::new();
109    out.extend_from_slice(&GRAPH_ADJACENCY_PAYLOAD_VERSION.to_le_bytes());
110    out.extend_from_slice(&(n_nodes as u64).to_le_bytes());
111    push_intpack(&mut out, &offsets);
112    push_intpack(&mut out, &dst_gaps);
113    push_edge_types(&mut out, &types);
114    Ok(out)
115}
116
117/// decode the csr payload back to its canonical edge list (the exact order
118/// [`encode_graph_adjacency`] emitted: ascending src, edge_type, dst).
119/// typed errors on any truncation or hostile claim; never panics.
120pub fn decode_graph_adjacency(bytes: &[u8]) -> Result<(usize, Vec<Edge>), UrnaError> {
121    let parts = parse_csr_parts(bytes)?;
122    let mut edges = Vec::with_capacity(parts.neighbors.len());
123    for node in 0..parts.n_nodes {
124        let start = parts.offsets[node] as usize;
125        let end = parts.offsets[node + 1] as usize;
126        for i in start..end {
127            edges.push(Edge {
128                src: node as u32,
129                dst: parts.neighbors[i],
130                edge_type: parts.edge_types[i],
131            });
132        }
133    }
134    Ok((parts.n_nodes, edges))
135}
136
137fn push_intpack(out: &mut Vec<u8>, values: &[u64]) {
138    let blob = pack_u64s(values);
139    out.extend_from_slice(&(blob.len() as u32).to_le_bytes());
140    out.extend_from_slice(&blob);
141}
142
143fn push_edge_types(out: &mut Vec<u8>, types: &[u8]) {
144    // iso single scalar when every edge shares one type (falkordb iso). an
145    // empty edge set is iso with edge_type 0 (decode reads zero edges).
146    let iso = types.first().copied().unwrap_or(0);
147    if types.iter().all(|&t| t == iso) {
148        out.push(EDGE_COL_ISO);
149        out.push(iso);
150    } else {
151        out.push(EDGE_COL_RUN);
152        let col: Vec<u64> = types.iter().map(|&t| t as u64).collect();
153        push_intpack(out, &col);
154    }
155}
156
157fn read_edge_types(cur: &mut Cursor, total: usize) -> Result<Vec<u8>, UrnaError> {
158    let kind = cur.u8()?;
159    match kind {
160        EDGE_COL_ISO => {
161            let iso = cur.u8()?;
162            Ok(vec![iso; total])
163        }
164        EDGE_COL_RUN => {
165            let col = cur.intpack_column()?;
166            if col.len() != total {
167                return Err(malformed("graph_adjacency: edge-type run length mismatch"));
168            }
169            col.iter()
170                .map(|&v| {
171                    u8::try_from(v).map_err(|_| malformed("graph_adjacency: edge-type > 255"))
172                })
173                .collect()
174        }
175        other => Err(malformed(format!(
176            "graph_adjacency: unknown edge-type kind {}",
177            other
178        ))),
179    }
180}
181
182/// light cursor over the csr payload. every read is bounds-checked and
183/// returns a typed error, never a panic on a hostile mmap.
184struct Cursor<'a> {
185    buf: &'a [u8],
186    pos: usize,
187}
188
189impl<'a> Cursor<'a> {
190    fn new(buf: &'a [u8]) -> Self {
191        Self { buf, pos: 0 }
192    }
193    fn take(&mut self, n: usize) -> Result<&'a [u8], UrnaError> {
194        if n > self.buf.len() - self.pos {
195            return Err(malformed("graph_adjacency: unexpected EOF"));
196        }
197        let s = &self.buf[self.pos..self.pos + n];
198        self.pos += n;
199        Ok(s)
200    }
201    fn u8(&mut self) -> Result<u8, UrnaError> {
202        Ok(self.take(1)?[0])
203    }
204    fn u32(&mut self) -> Result<u32, UrnaError> {
205        le_u32(self.take(4)?)
206    }
207    fn u64(&mut self) -> Result<u64, UrnaError> {
208        le_u64(self.take(8)?)
209    }
210    /// read a length-prefixed intpack blob, decode it to u64s.
211    fn intpack_column(&mut self) -> Result<Vec<u64>, UrnaError> {
212        let len = self.u32()? as usize;
213        let blob = self.take(len)?;
214        crate::encoding::unpack_u64s(blob)
215    }
216}
217
218/// the validated csr columns the runtime `CsrIndex` indexes in O(1):
219/// `offsets` (len n_nodes+1) bounds each node's run in `neighbors` (absolute
220/// decoded dst ids) and `edge_types` (1:1 with `neighbors`).
221pub struct CsrParts {
222    pub n_nodes: usize,
223    pub offsets: Vec<u64>,
224    pub neighbors: Vec<u32>,
225    pub edge_types: Vec<u8>,
226}
227
228/// parse the csr payload into validated owned columns (the single parser):
229/// bounds-checks the header, monotone offsets, degree cap, column lengths,
230/// and dst range, then de-gaps the neighbor ids into absolute ids so
231/// `neighbors[start..end]` is a contiguous slice per node. never panics.
232pub fn parse_csr_parts(bytes: &[u8]) -> Result<CsrParts, UrnaError> {
233    let mut cur = Cursor::new(bytes);
234    let version = cur.u32()?;
235    if version != GRAPH_ADJACENCY_PAYLOAD_VERSION {
236        return Err(UrnaError::UnsupportedSectionVersion {
237            section_id: SECTION_GRAPH_ADJACENCY,
238            version,
239        });
240    }
241    let n_nodes = cur.u64()? as usize;
242    let offsets = cur.intpack_column()?;
243    // compare against offsets.len() (bounded by the physical payload) WITHOUT
244    // computing n_nodes+1, which would overflow on a hostile n_nodes claim.
245    if offsets.is_empty() || offsets.len() - 1 != n_nodes {
246        return Err(malformed("graph_adjacency: offsets length != n_nodes+1"));
247    }
248    // offsets must be monotone non-decreasing; the last is the edge count.
249    let mut total = 0u64;
250    for w in offsets.windows(2) {
251        if w[1] < w[0] {
252            return Err(malformed("graph_adjacency: offsets not monotone"));
253        }
254        let deg = w[1] - w[0];
255        if deg > MAX_DEGREE {
256            return Err(malformed("graph_adjacency: degree exceeds cap"));
257        }
258        total = total.saturating_add(deg);
259    }
260    if offsets.last().copied() != Some(total) {
261        return Err(malformed("graph_adjacency: final offset != edge count"));
262    }
263    let dst_gaps = cur.intpack_column()?;
264    if dst_gaps.len() as u64 != total {
265        return Err(malformed("graph_adjacency: dst column length mismatch"));
266    }
267    let edge_types = read_edge_types(&mut cur, total as usize)?;
268
269    // de-gap per (node, edge_type) run into absolute dst ids, mirroring the
270    // encoder's reset at each edge_type boundary, range-checking each.
271    let mut neighbors = Vec::with_capacity(total as usize);
272    for node in 0..n_nodes {
273        let start = offsets[node] as usize;
274        let end = offsets[node + 1] as usize;
275        let mut prev_dst: Option<u32> = None;
276        let mut prev_type: Option<u8> = None;
277        for i in start..end {
278            let t = edge_types[i];
279            if prev_type != Some(t) {
280                prev_dst = None;
281                prev_type = Some(t);
282            }
283            let dst = match prev_dst {
284                Some(p) => (p as u64).wrapping_add(dst_gaps[i]) as u32,
285                None => dst_gaps[i] as u32,
286            };
287            if dst as usize >= n_nodes {
288                return Err(malformed("graph_adjacency: dst out of range"));
289            }
290            neighbors.push(dst);
291            prev_dst = Some(dst);
292        }
293    }
294    Ok(CsrParts {
295        n_nodes,
296        offsets,
297        neighbors,
298        edge_types,
299    })
300}
urna_format/sections/graph/adjacency.rs

urna_format/sections/graph/
adjacency.rs
