nodedb_physical/physical_plan/graph.rs
1// SPDX-License-Identifier: Apache-2.0
2
3//! Graph engine operations dispatched to the Data Plane.
4
5use nodedb_graph::{AlgoParams, Direction, GraphAlgorithm, GraphTraversalOptions};
6use nodedb_types::{Surrogate, SurrogateBitmap};
7
8/// One edge in an `EdgePutBatch` / `EdgeDeleteBatch`.
9///
10/// `src_surrogate` / `dst_surrogate` carry the global row identity for the
11/// edge endpoints (resolved at construction time via the surrogate assigner).
12/// `Surrogate::ZERO` is used in test fixtures and on in-memory paths where
13/// no catalog is wired; production paths always populate real surrogates.
14#[derive(
15 Debug,
16 Clone,
17 PartialEq,
18 serde::Serialize,
19 serde::Deserialize,
20 zerompk::ToMessagePack,
21 zerompk::FromMessagePack,
22)]
23pub struct BatchEdge {
24 pub collection: String,
25 pub src_id: String,
26 pub label: String,
27 pub dst_id: String,
28 pub src_surrogate: Surrogate,
29 pub dst_surrogate: Surrogate,
30}
31
32/// Graph engine physical operations.
33#[derive(
34 Debug,
35 Clone,
36 PartialEq,
37 serde::Serialize,
38 serde::Deserialize,
39 zerompk::ToMessagePack,
40 zerompk::FromMessagePack,
41)]
42pub enum GraphOp {
43 /// Insert a graph edge with properties.
44 ///
45 /// `src_surrogate` / `dst_surrogate` carry the global row identity for
46 /// the two endpoints, resolved at construction time. The string `src_id`
47 /// / `dst_id` remain user-visible identifiers (used by the CSR partition
48 /// for label interning and by the edge store for keying), while the
49 /// surrogates are the cross-engine join currency.
50 EdgePut {
51 collection: String,
52 src_id: String,
53 label: String,
54 dst_id: String,
55 properties: Vec<u8>,
56 src_surrogate: Surrogate,
57 dst_surrogate: Surrogate,
58 },
59
60 /// Batched edge insert: many `(collection, src, label, dst)` tuples.
61 /// Every edge in the batch must target the same collection — the
62 /// batch is a unit of work, not a cross-collection scatter.
63 EdgePutBatch { edges: Vec<BatchEdge> },
64
65 /// Delete a graph edge.
66 EdgeDelete {
67 collection: String,
68 src_id: String,
69 label: String,
70 dst_id: String,
71 },
72
73 /// Batched edge delete: used to revert a partial `EdgePutBatch` on
74 /// failure so the DDL leaves no stranded edges.
75 EdgeDeleteBatch { edges: Vec<BatchEdge> },
76
77 /// Graph hop traversal: BFS from start nodes via label, bounded by depth.
78 Hop {
79 start_nodes: Vec<String>,
80 edge_label: Option<String>,
81 direction: Direction,
82 depth: usize,
83 options: GraphTraversalOptions,
84 /// RLS filters applied to traversed nodes before returning.
85 rls_filters: Vec<u8>,
86 /// Optional surrogate prefilter restricting which frontier nodes are
87 /// eligible as traversal targets. `None` = no restriction.
88 frontier_bitmap: Option<SurrogateBitmap>,
89 },
90
91 /// Immediate 1-hop neighbors lookup.
92 Neighbors {
93 node_id: String,
94 edge_label: Option<String>,
95 direction: Direction,
96 /// RLS filters applied to neighbor nodes before returning.
97 rls_filters: Vec<u8>,
98 },
99
100 /// Batched 1-hop neighbors lookup: one RPC per hop of a BFS frontier
101 /// instead of one RPC per frontier node. Returns
102 /// `[{ src, label, node }, ...]` so the caller can attribute each
103 /// neighbor to its origin (needed for shortest-path parent pointers).
104 ///
105 /// `max_results` is the per-RPC cap: the Data Plane handler stops
106 /// emitting entries once the batch reaches this size so a single
107 /// wide hop cannot allocate past the caller's budget. `0` means
108 /// unbounded (use with care).
109 NeighborsMulti {
110 node_ids: Vec<String>,
111 edge_label: Option<String>,
112 direction: Direction,
113 max_results: u32,
114 /// RLS filters applied to neighbor nodes before returning.
115 rls_filters: Vec<u8>,
116 },
117
118 /// Shortest path between two nodes.
119 Path {
120 src: String,
121 dst: String,
122 edge_label: Option<String>,
123 max_depth: usize,
124 options: GraphTraversalOptions,
125 /// RLS filters applied to path nodes before returning.
126 rls_filters: Vec<u8>,
127 /// Optional surrogate prefilter restricting which nodes may appear
128 /// on the path. `None` = no restriction.
129 frontier_bitmap: Option<SurrogateBitmap>,
130 },
131
132 /// Materialize a subgraph as edge tuples.
133 Subgraph {
134 start_nodes: Vec<String>,
135 edge_label: Option<String>,
136 depth: usize,
137 options: GraphTraversalOptions,
138 /// RLS filters applied to subgraph nodes/edges before returning.
139 rls_filters: Vec<u8>,
140 },
141
142 /// GraphRAG fusion: vector search → graph expansion → RRF ranking.
143 ///
144 /// Two-source form: vector + graph (backwards-compatible; `bm25_query` is `None`).
145 /// Three-source form: vector + BM25 text + graph; activated when `bm25_query` is set.
146 RagFusion {
147 collection: String,
148 query_vector: Vec<f32>,
149 vector_top_k: usize,
150 edge_label: Option<String>,
151 direction: Direction,
152 expansion_depth: usize,
153 final_top_k: usize,
154 /// Two-source RRF k constants: (vector_k, graph_k).
155 /// Used when `bm25_query` is absent (backwards-compatible two-source form).
156 rrf_k: (f64, f64),
157 /// Three-source RRF k constants: (vector_k, text_k, graph_k).
158 /// Set when the FUSION DSL carries a `BM25 '...' ON '...'` clause.
159 rrf_k_triple: Option<(f64, f64, f64)>,
160 /// Vector index field name. Empty string selects the raw (field-less)
161 /// index created via `VectorOp::Insert`; a non-empty value selects
162 /// the field-backed index created when documents are inserted with an
163 /// embedded vector column (e.g. `INSERT INTO col (id, embedding) VALUES …`).
164 vector_field: String,
165 options: GraphTraversalOptions,
166 /// BM25 query string for the text leg of three-source fusion. `None` = two-source.
167 bm25_query: Option<String>,
168 /// Document field on which BM25 scoring is applied. Required when `bm25_query` is set.
169 bm25_field: Option<String>,
170 },
171
172 /// Graph algorithm execution (PageRank, WCC, SSSP, etc.).
173 Algo {
174 algorithm: GraphAlgorithm,
175 params: AlgoParams,
176 },
177
178 /// Graph pattern matching (MATCH clause execution).
179 Match {
180 /// Serialized `MatchQuery` (MessagePack).
181 query: Vec<u8>,
182 /// Optional surrogate prefilter restricting which nodes are eligible
183 /// as pattern anchors. `None` = no restriction.
184 frontier_bitmap: Option<SurrogateBitmap>,
185 },
186
187 /// Set node labels (bitset-based, up to 64 distinct labels).
188 SetNodeLabels {
189 node_id: String,
190 labels: Vec<String>,
191 },
192
193 /// Remove node labels.
194 RemoveNodeLabels {
195 node_id: String,
196 labels: Vec<String>,
197 },
198
199 /// Bitemporal 1-hop neighbors lookup.
200 ///
201 /// Resolves edges whose latest version with `system_from <= system_as_of_ms`
202 /// (converted to HLC ordinal) is not a sentinel, optionally also filtering
203 /// by `valid_from_ms <= valid_at_ms < valid_until_ms`. The handler calls
204 /// `ceiling_resolve_edge` per candidate base edge.
205 TemporalNeighbors {
206 /// Edge store is collection-scoped; current-state `Neighbors` reads
207 /// the tenant-wide CSR, but the versioned key layout is
208 /// `{collection}\x00...`, so the bitemporal path must name the
209 /// collection explicitly.
210 collection: String,
211 node_id: String,
212 edge_label: Option<String>,
213 direction: Direction,
214 /// System-time cutoff in ms. `None` falls back to current-state
215 /// semantics identical to `Neighbors`.
216 system_as_of_ms: Option<i64>,
217 /// Optional valid-time point. Skipped when `None`.
218 valid_at_ms: Option<i64>,
219 rls_filters: Vec<u8>,
220 },
221
222 /// Bitemporal graph algorithm execution.
223 ///
224 /// Identical to `Algo` but builds its CSR snapshot via
225 /// `CsrSnapshot::from_edge_store_as_of` at the given system-time cutoff
226 /// before running the algorithm.
227 TemporalAlgorithm {
228 algorithm: GraphAlgorithm,
229 params: AlgoParams,
230 /// System-time cutoff in ms. `None` means current state (equivalent
231 /// to plain `Algo`).
232 system_as_of_ms: Option<i64>,
233 },
234
235 /// Read persistent graph-stats counters from the edge store.
236 ///
237 /// `collection = Some` returns stats for one `(tenant, collection)` pair.
238 /// `collection = None` returns stats for every collection that has
239 /// edges (or had any, per cold-start rebuild) for this tenant.
240 ///
241 /// `as_of = None` is the O(1) live-snapshot path (reads the cached
242 /// summary row). `as_of = Some(ms)` falls back to a historical scan.
243 Stats {
244 collection: Option<String>,
245 as_of: Option<i64>,
246 },
247}