uni_plugin/traits/algorithm.rs
1//! Graph algorithm plugins.
2//!
3//! Two surfaces: [`AlgorithmProvider`] for black-box algorithms (the
4//! existing `uni-algo` library style), and [`GraphView`] — the stable,
5//! read-only topology API a provider obtains from its [`AlgorithmHost`]
6//! via [`AlgorithmHost::project`] to walk the graph without depending on
7//! `uni-store` / `uni-algo` types.
8
9use std::sync::Arc;
10
11use datafusion::execution::SendableRecordBatchStream;
12use futures::future::BoxFuture;
13use uni_common::core::id::Vid;
14
15use crate::errors::FnError;
16
17/// Static signature of an algorithm.
18///
19/// `args` and `slices` are additive, defaulted fields (construct with
20/// `..Default::default()`): existing providers that leave them empty keep the
21/// legacy untyped `config_json` contract unchanged. A provider that declares
22/// `args` opts into host-side arity/type validation before it runs (proposal
23/// §4.6 / decision D7); a provider that declares `slices` opts into load-time
24/// capability-slice version negotiation (proposal §4.3 / decision D6).
25#[derive(Clone, Debug, Default)]
26pub struct AlgorithmSignature {
27 /// Output column schema.
28 pub output_fields: Vec<arrow_schema::Field>,
29 /// Markdown docs.
30 pub docs: String,
31 /// Declared positional arguments, in call order.
32 ///
33 /// Empty (the default) preserves the legacy behavior: arguments arrive as a
34 /// raw positional `config_json` array the provider parses itself. When
35 /// non-empty, the host validates arity and coerces each positional argument
36 /// against the declared [`NamedArgType`](crate::traits::procedure::NamedArgType) before the provider runs, filling
37 /// omitted trailing arguments from their declared defaults.
38 pub args: Vec<crate::traits::procedure::NamedArgType>,
39 /// Required capability slices, checked at load time.
40 ///
41 /// Empty (the default) means the algorithm targets only the always-present
42 /// `graph-compute@1` surface. A declared [`SliceReq`] whose version the host
43 /// does not provide fails the load with a clear error (`0x86A`) rather than a
44 /// mysterious runtime "unknown kernel op" trap.
45 pub slices: Vec<SliceReq>,
46 /// Whether a `CALL` of this algorithm may be planned as a first-class
47 /// DataFusion `ExecutionPlan` node (the vectorized path), rather than only
48 /// through the row-based interpreter (proposal §6, DF-3).
49 ///
50 /// `false` (the default) preserves the row path. A provider sets this `true`
51 /// to *declare* it composes correctly as a leaf/source plan node — its `run`
52 /// returns a well-formed `RecordBatch` stream matching `output_fields`, and it
53 /// consumes MATCH-bound arguments via `outer_values` rather than a child plan.
54 /// This replaces the previous name-prefix allowlist (`uni.algo.*`): eligibility
55 /// is now **registration-driven**, so a third-party `myco.algo.*` provider that
56 /// declares `df_composable` is a first-class plan node like the first-party
57 /// ones, and a `uni.algo.`-named provider that does *not* declare it no longer
58 /// gets the DF path by prefix alone.
59 pub df_composable: bool,
60}
61
62/// A required capability-slice version an algorithm declares in its signature.
63///
64/// The host checks each requirement against the slices it actually implements
65/// (today only `graph-compute@1`) when the algorithm loads, refusing a mismatch
66/// up front (proposal §4.3 / decision D6). Adding a slice or bumping a version is
67/// a forward-compatible, additive change: an algorithm that declares no slices is
68/// grandfathered onto the base surface.
69#[derive(Clone, Debug, PartialEq, Eq)]
70pub struct SliceReq {
71 /// The capability-slice name, e.g. `"graph-compute"`.
72 pub slice: smol_str::SmolStr,
73 /// The minimum slice version the algorithm requires.
74 pub version: u16,
75}
76
77/// The capability slices this host implements, for load-time negotiation.
78///
79/// A guest algorithm's declared [`SliceReq`]s are checked against this table
80/// when it loads. The host implements `graph-compute@1` (the coarse read-only
81/// kernels over a projection) and `graph-arena@1` (mutable session-local
82/// structure, proposal §5.1). A future slice is added here in lockstep with its
83/// kernels so negotiation stays a pure lookup (proposal §4.3 / §10).
84pub const HOST_CAPABILITY_SLICES: &[(&str, u16)] = &[("graph-compute", 1), ("graph-arena", 1)];
85
86impl AlgorithmSignature {
87 /// Validates the declared capability slices against `host_slices`.
88 ///
89 /// Each requirement must be met by a host slice of the same name whose
90 /// version is at least the requested one. Pass [`HOST_CAPABILITY_SLICES`] for
91 /// the production surface.
92 ///
93 /// # Errors
94 /// Returns `0x86A` (`SliceVersionMismatch`) naming the first requirement the
95 /// host cannot satisfy (proposal §4.3 / §12, decision D6).
96 pub fn check_slices(&self, host_slices: &[(&str, u16)]) -> Result<(), FnError> {
97 for req in &self.slices {
98 let satisfied = host_slices
99 .iter()
100 .any(|(name, ver)| *name == req.slice.as_str() && *ver >= req.version);
101 if !satisfied {
102 return Err(FnError::new(
103 0x86A,
104 format!(
105 "algorithm requires capability slice `{}@{}` the host does not provide",
106 req.slice, req.version
107 ),
108 ));
109 }
110 }
111 Ok(())
112 }
113
114 /// Validates and normalizes a positional `config_json` array against `args`.
115 ///
116 /// When `args` is empty this is a no-op returning `config_json` unchanged, so
117 /// providers on the legacy untyped contract are unaffected. Otherwise it
118 /// parses the positional JSON array and, per declared argument: rejects a
119 /// present value whose JSON kind is incompatible with the declared
120 /// [`ArgType`](crate::traits::scalar::ArgType), errors on a missing argument
121 /// that has no default, and appends the declared default for an omitted
122 /// trailing argument. Extra positional arguments beyond the declared arity
123 /// are rejected. The returned JSON array is what the provider then parses, so
124 /// it observes defaults already filled in.
125 ///
126 /// # Errors
127 /// Returns `0x86E` (argument arity/type violation) with a message naming the
128 /// offending argument (proposal §4.6, decision D7).
129 pub fn coerce_config_json(&self, config_json: &str) -> Result<String, FnError> {
130 use crate::traits::scalar::ArgType;
131
132 if self.args.is_empty() {
133 return Ok(config_json.to_owned());
134 }
135 let mut provided: Vec<serde_json::Value> = if config_json.trim().is_empty() {
136 Vec::new()
137 } else {
138 serde_json::from_str(config_json)
139 .map_err(|e| FnError::new(0x86E, format!("bad positional config json: {e}")))?
140 };
141 if provided.len() > self.args.len() {
142 return Err(FnError::new(
143 0x86E,
144 format!(
145 "too many arguments: got {}, expected at most {}",
146 provided.len(),
147 self.args.len()
148 ),
149 ));
150 }
151 let mut out = Vec::with_capacity(self.args.len());
152 for (i, arg) in self.args.iter().enumerate() {
153 match provided.get_mut(i) {
154 Some(value) => {
155 let value = std::mem::replace(value, serde_json::Value::Null);
156 // A `CypherValue` argument is opaque and accepts any JSON.
157 if !matches!(arg.ty, ArgType::CypherValue)
158 && !json_matches_argtype(&value, &arg.ty)
159 {
160 return Err(FnError::new(
161 0x86E,
162 format!("argument `{}` (position {i}) has the wrong type", arg.name),
163 ));
164 }
165 out.push(value);
166 }
167 None => match &arg.default {
168 Some(default) => out.push(scalar_default_to_json(default)),
169 None => {
170 return Err(FnError::new(
171 0x86E,
172 format!("missing required argument `{}` (position {i})", arg.name),
173 ));
174 }
175 },
176 }
177 }
178 serde_json::to_string(&out)
179 .map_err(|e| FnError::new(0x86E, format!("re-encoding coerced config: {e}")))
180 }
181}
182
183/// Whether a JSON value is compatible with a declared primitive/vector arg type.
184fn json_matches_argtype(value: &serde_json::Value, ty: &crate::traits::scalar::ArgType) -> bool {
185 use arrow_schema::DataType;
186
187 use crate::traits::scalar::ArgType;
188 match ty {
189 ArgType::CypherValue => true,
190 ArgType::Vector { .. } => value.is_array(),
191 ArgType::Variadic(inner) => json_matches_argtype(value, inner),
192 ArgType::Primitive(dt) => match dt {
193 DataType::Boolean => value.is_boolean(),
194 DataType::Utf8 | DataType::LargeUtf8 => value.is_string(),
195 DataType::Float16 | DataType::Float32 | DataType::Float64 => value.is_number(),
196 d if d.is_integer() => value.is_i64() || value.is_u64(),
197 // Unknown/opaque primitive: don't reject, defer to the provider.
198 _ => true,
199 },
200 }
201}
202
203/// Renders a declared [`ScalarValue`](datafusion::scalar::ScalarValue) default as
204/// JSON to append for an omitted trailing argument.
205fn scalar_default_to_json(default: &datafusion::scalar::ScalarValue) -> serde_json::Value {
206 use datafusion::scalar::ScalarValue;
207
208 match default {
209 ScalarValue::Null => serde_json::Value::Null,
210 ScalarValue::Boolean(Some(b)) => serde_json::Value::Bool(*b),
211 ScalarValue::Float32(Some(x)) => serde_json::json!(*x),
212 ScalarValue::Float64(Some(x)) => serde_json::json!(*x),
213 ScalarValue::Int8(Some(x)) => serde_json::json!(*x),
214 ScalarValue::Int16(Some(x)) => serde_json::json!(*x),
215 ScalarValue::Int32(Some(x)) => serde_json::json!(*x),
216 ScalarValue::Int64(Some(x)) => serde_json::json!(*x),
217 ScalarValue::UInt8(Some(x)) => serde_json::json!(*x),
218 ScalarValue::UInt16(Some(x)) => serde_json::json!(*x),
219 ScalarValue::UInt32(Some(x)) => serde_json::json!(*x),
220 ScalarValue::UInt64(Some(x)) => serde_json::json!(*x),
221 ScalarValue::Utf8(Some(s)) | ScalarValue::LargeUtf8(Some(s)) => {
222 serde_json::Value::String(s.clone())
223 }
224 // Any other/None scalar defaults to JSON null.
225 _ => serde_json::Value::Null,
226 }
227}
228
229/// Per-invocation context passed to an [`AlgorithmProvider`].
230///
231/// `host` is an opaque [`AlgorithmHost`] callback the host populates
232/// when invoking the algorithm. Algorithms that need a concrete
233/// graph-projection / storage handle downcast through `host` rather
234/// than depend on `uni-store` / `uni-algo` types directly — this keeps
235/// `uni-plugin` free of upward dependencies.
236#[non_exhaustive]
237pub struct AlgorithmContext<'a> {
238 /// JSON-serialized algorithm configuration.
239 pub config_json: &'a str,
240 /// Optional opaque host handle. `None` when no host is bound — the
241 /// algorithm may fall back to a config-only path or surface an
242 /// `Unbound` error.
243 pub host: Option<&'a dyn AlgorithmHost>,
244}
245
246impl std::fmt::Debug for AlgorithmContext<'_> {
247 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
248 f.debug_struct("AlgorithmContext")
249 .field("config_json", &self.config_json)
250 .field("host_bound", &self.host.is_some())
251 .finish()
252 }
253}
254
255impl<'a> AlgorithmContext<'a> {
256 /// Construct an `AlgorithmContext` with no host bound.
257 #[must_use]
258 pub fn new(config_json: &'a str) -> Self {
259 Self {
260 config_json,
261 host: None,
262 }
263 }
264
265 /// Attach a host handle.
266 #[must_use]
267 pub fn with_host(mut self, host: &'a dyn AlgorithmHost) -> Self {
268 self.host = Some(host);
269 self
270 }
271}
272
273/// Host callback surfacing graph access to plugin algorithms.
274///
275/// A provider's [`AlgorithmProvider::run`] receives an [`AlgorithmHost`]
276/// through its [`AlgorithmContext`] and calls [`AlgorithmHost::project`]
277/// to materialize a [`GraphView`] over the requested subgraph. Hosts
278/// (e.g. `uni-plugin-builtin`) implement `project` by building a
279/// projection from their `StorageManager` / `L0Manager`; the
280/// [`AlgorithmHost::as_any`] downcast hook remains for hosts that expose
281/// additional concrete state. This keeps `uni-plugin` free of upward
282/// dependencies on `uni-store` / `uni-algo`.
283pub trait AlgorithmHost: Send + Sync {
284 /// Downcast hook — bridges implement this to expose the concrete
285 /// host type.
286 fn as_any(&self) -> &dyn std::any::Any;
287
288 /// Materialize a read-only [`GraphView`] over the subgraph named by
289 /// `spec`.
290 ///
291 /// The returned future is `'static` (owns its inputs) so a provider
292 /// can move it into the stream it returns from the synchronous
293 /// [`AlgorithmProvider::run`] and `.await` it there. The default
294 /// implementation reports that the host offers no graph access;
295 /// graph-capable hosts override it.
296 ///
297 /// # Errors
298 ///
299 /// Returns [`FnError`] if the host offers no graph access, the
300 /// caller lacks the required capability (e.g. `HostQuery`), or the
301 /// projection cannot be built.
302 fn project(
303 &self,
304 spec: &GraphProjectionSpec,
305 ) -> BoxFuture<'static, Result<Arc<dyn GraphView>, FnError>> {
306 let _ = spec;
307 Box::pin(async {
308 Err(FnError::new(
309 0x805,
310 "AlgorithmHost: project() is not supported by this host",
311 ))
312 })
313 }
314}
315
316/// Selects which subgraph an [`AlgorithmHost::project`] call materializes.
317///
318/// Naming `node_labels` / `edge_types` scopes the projection to exactly those.
319/// Leaving BOTH empty means "the whole graph", but that is only honored when
320/// [`Self::project_all`] is `true` — otherwise the projection fails loud (G9),
321/// so an unscoped projection can never *silently* pull in unrelated labels (e.g.
322/// a coexisting search tree) and corrupt an index-keyed kernel. `weight_property`
323/// names an edge property to expose through [`GraphView::out_weight`];
324/// `include_reverse` requests inbound adjacency ([`GraphView::in_neighbors`]).
325#[derive(Clone, Debug, Default)]
326pub struct GraphProjectionSpec {
327 /// Vertex labels to include; empty (with [`Self::project_all`]) selects every label.
328 pub node_labels: Vec<String>,
329 /// Edge types to include; empty (with [`Self::project_all`]) selects every type.
330 pub edge_types: Vec<String>,
331 /// Edge property surfaced as the traversal weight, if any.
332 pub weight_property: Option<String>,
333 /// Whether to also build inbound adjacency.
334 pub include_reverse: bool,
335 /// Vertex properties to materialize into per-vertex `[V]` tensors the guest
336 /// reads by name via `gc.node_property` (issue #151).
337 pub node_properties: Vec<String>,
338 /// Edge properties to materialize into per-edge `[E]` tensors the guest
339 /// reads by name via `gc.edge_property` (issue #151).
340 pub edge_properties: Vec<String>,
341 /// Deliberate opt-in to projecting the *whole* graph when neither
342 /// `node_labels` nor `edge_types` is named (G9). With both empty and this
343 /// `false`, [`AlgorithmHost::project`] fails loud instead of silently
344 /// pulling in every schema label/edge-type. First-party providers set this
345 /// to preserve the ergonomic whole-graph default; guest loaders must pass
346 /// `projectAll: true` in their config to project everything on purpose.
347 pub project_all: bool,
348}
349
350impl GraphProjectionSpec {
351 /// Parses the Native-mode projection knobs from a guest/procedure config
352 /// object. This is the single source of truth for `nodeLabels` /
353 /// `edgeTypes` (accepting the `relationshipTypes` alias) / `weightProperty`
354 /// / `includeReverse` — the native providers and all four guest loaders
355 /// funnel through it so the knob names cannot drift.
356 ///
357 /// `includeReverse` defaults to `true` (inbound adjacency is built unless
358 /// the caller opts out), matching the graphRef contract in `uni-algo` and
359 /// keeping In-direction kernels (WCC / k-core / HITS) working. Unknown keys
360 /// are ignored and malformed values fall back to the field default: this is
361 /// a best-effort projection hint, not a strict schema.
362 #[must_use]
363 pub fn from_config_object(cfg: &serde_json::Map<String, serde_json::Value>) -> Self {
364 fn string_array(v: &serde_json::Value) -> Vec<String> {
365 v.as_array()
366 .map(|arr| {
367 arr.iter()
368 .filter_map(|s| s.as_str().map(str::to_owned))
369 .collect()
370 })
371 .unwrap_or_default()
372 }
373
374 let node_labels = cfg.get("nodeLabels").map(string_array).unwrap_or_default();
375 // `relationshipTypes` is the openCypher-flavored alias the native
376 // procedures already accept as a synonym for `edgeTypes`.
377 let edge_types = cfg
378 .get("edgeTypes")
379 .or_else(|| cfg.get("relationshipTypes"))
380 .map(string_array)
381 .unwrap_or_default();
382 let weight_property = cfg
383 .get("weightProperty")
384 .and_then(serde_json::Value::as_str)
385 .map(str::to_owned);
386 let include_reverse = cfg
387 .get("includeReverse")
388 .and_then(serde_json::Value::as_bool)
389 .unwrap_or(true);
390 let node_properties = cfg
391 .get("nodeProperties")
392 .map(string_array)
393 .unwrap_or_default();
394 let edge_properties = cfg
395 .get("edgeProperties")
396 .map(string_array)
397 .unwrap_or_default();
398 // G9: explicit opt-in to a whole-graph projection. Absent/false means an
399 // unscoped projection is rejected fail-loud by the bridge.
400 let project_all = cfg
401 .get("projectAll")
402 .and_then(serde_json::Value::as_bool)
403 .unwrap_or(false);
404
405 Self {
406 node_labels,
407 edge_types,
408 weight_property,
409 include_reverse,
410 node_properties,
411 edge_properties,
412 project_all,
413 }
414 }
415
416 /// Keys that mark a trailing CALL argument as a projection-config object
417 /// (a "graphRef") rather than a guest algorithm argument. Covers the
418 /// Native knobs, the P2 property-tensor knobs, and the P3 Cypher/Named
419 /// graphRef knobs so the trailing object is stripped from the guest's
420 /// arguments consistently across every projection mode.
421 pub const CONFIG_KEYS: &'static [&'static str] = &[
422 "nodeLabels",
423 "edgeTypes",
424 "relationshipTypes",
425 "weightProperty",
426 "includeReverse",
427 "nodeProperties",
428 "edgeProperties",
429 "projectAll",
430 "nodeQuery",
431 "edgeQuery",
432 "weightColumn",
433 "name",
434 "scopes",
435 ];
436
437 /// Keys that mark a config object as a **Cypher/Named** `graphRef` rather
438 /// than a Native label/edge-type scoping object.
439 ///
440 /// Single-sourced so the query layer's graphRef sniffing and the per-scope
441 /// parsing below cannot drift: a scope routed to the Native storage scan
442 /// when it names a Cypher query would silently project the wrong graph.
443 pub const QUERY_CONFIG_KEYS: &'static [&'static str] = &["nodeQuery", "edgeQuery", "name"];
444
445 /// Whether a config object names a Cypher/Named projection.
446 #[must_use]
447 pub fn is_query_graph_ref(cfg: &serde_json::Map<String, serde_json::Value>) -> bool {
448 Self::QUERY_CONFIG_KEYS.iter().any(|k| cfg.contains_key(*k))
449 }
450
451 /// Parses the `scopes` map into pre-declared named projections.
452 ///
453 /// A guest that needs more than one view of the store declares them at the
454 /// CALL site rather than projecting on demand, because projection is the one
455 /// thing a guest must not be able to trigger in a loop:
456 ///
457 /// ```cypher
458 /// CALL myplugin.compare([], {
459 /// nodeLabels: ['Cell'], edgeTypes: ['ADJ'],
460 /// scopes: {
461 /// agg: {nodeLabels: ['Cell'], edgeTypes: ['AGGREGATES']},
462 /// flow: {nodeQuery: 'MATCH (c:Cell) RETURN id(c) AS id'}
463 /// }
464 /// })
465 /// ```
466 ///
467 /// The outer object stays the *primary* projection — the one `emit` keys its
468 /// `nodeId` column to. Each scope value is parsed by the same
469 /// [`Self::from_config_object`], so every Native knob works per scope; a
470 /// scope bearing a Cypher/Named key is carried through verbatim for the
471 /// resolver instead.
472 ///
473 /// # Errors
474 /// Returns a message naming the offending scope when `scopes` is not an
475 /// object, a scope name is empty, a scope value is not an object, or a scope
476 /// is named `graph` (which would shadow the primary handle's own accessor).
477 pub fn scopes_from_config_object(
478 cfg: &serde_json::Map<String, serde_json::Value>,
479 ) -> Result<Vec<GraphScopeSpec>, String> {
480 let Some(raw) = cfg.get("scopes") else {
481 return Ok(Vec::new());
482 };
483 let map = raw
484 .as_object()
485 .ok_or_else(|| "`scopes` must be an object of {name: projection-config}".to_string())?;
486 let mut out = Vec::with_capacity(map.len());
487 for (name, value) in map {
488 if name.is_empty() {
489 return Err("a scope name must not be empty".to_string());
490 }
491 if name == "graph" {
492 return Err(
493 "`graph` is not a valid scope name: it is the primary projection, \
494 reached with `gc.graph()` rather than `gc.graph_named(..)`"
495 .to_string(),
496 );
497 }
498 let obj = value.as_object().ok_or_else(|| {
499 format!("scope `{name}` must be a projection-config object, got {value}")
500 })?;
501 let graph_ref = Self::is_query_graph_ref(obj).then(|| value.clone());
502 out.push(GraphScopeSpec {
503 name: name.clone(),
504 spec: Self::from_config_object(obj),
505 graph_ref,
506 });
507 }
508 Ok(out)
509 }
510
511 /// Rejects a `scopes` map on an algorithm that does not consume one.
512 ///
513 /// `scopes` joined [`Self::CONFIG_KEYS`] so a scopes-only object is stripped
514 /// from the guest's positional arguments. That stripping applies to *every*
515 /// algorithm, but only the guest loader adapters build the declared
516 /// projections — so without this a first-party provider would accept a
517 /// `scopes` map, project nothing, and run as if it had never been asked.
518 /// Silently ignoring a projection the caller asked for is the same failure
519 /// the unscoped-projection change (G9) made loud; this keeps it loud.
520 ///
521 /// # Errors
522 /// Returns `0x86E` naming `algorithm` when `cfg` carries a non-empty `scopes`.
523 pub fn reject_scopes(
524 cfg: &serde_json::Map<String, serde_json::Value>,
525 algorithm: &str,
526 ) -> Result<(), FnError> {
527 let has_scopes = cfg
528 .get("scopes")
529 .and_then(serde_json::Value::as_object)
530 .is_some_and(|m| !m.is_empty());
531 if !has_scopes {
532 return Ok(());
533 }
534 Err(FnError::new(
535 0x86E,
536 format!(
537 "{algorithm} does not take named `scopes`: it runs a fixed algorithm over \
538 one projection. Named scopes are a guest-authored-algorithm feature -- \
539 the guest is what decides which scope to read."
540 ),
541 ))
542 }
543
544 /// Like [`Self::take_from_args`] but also returns the raw config object.
545 ///
546 /// [`Self::take_from_args`] discards the object after parsing, which is fine
547 /// for the Native knobs but loses `scopes` (whose values must be re-parsed
548 /// per scope, and whose Cypher entries must survive verbatim).
549 #[must_use]
550 pub fn take_config_from_args(
551 args: &mut Vec<serde_json::Value>,
552 ) -> Option<serde_json::Map<String, serde_json::Value>> {
553 let is_config = args
554 .last()
555 .and_then(serde_json::Value::as_object)
556 .is_some_and(|o| Self::CONFIG_KEYS.iter().any(|k| o.contains_key(*k)));
557 if !is_config {
558 return None;
559 }
560 match args.pop() {
561 Some(serde_json::Value::Object(cfg)) => Some(cfg),
562 _ => None,
563 }
564 }
565
566 /// If the last element of `args` is a JSON object bearing at least one
567 /// [`Self::CONFIG_KEYS`] key, removes it from `args` and returns the parsed
568 /// Native spec; otherwise leaves `args` untouched and returns `None`.
569 ///
570 /// The Native-spec half of the "trailing object is the projection config"
571 /// convention. The guest loaders now go through
572 /// [`ProjectionPlan`](../../../uni_plugin_builtin/algorithms/bridge/struct.ProjectionPlan.html),
573 /// which needs the raw object to parse `scopes`; this remains for callers
574 /// that want only the Native knobs. Both share
575 /// [`Self::take_config_from_args`], so the recognition rule cannot drift.
576 /// In P3 Cypher/Named
577 /// mode the returned Native spec is empty (the query/name keys are unknown
578 /// to [`Self::from_config_object`]) and is ignored by the bridge in favor of
579 /// the pre-built projection — but the object is still stripped here so it
580 /// never reaches the guest function.
581 #[must_use]
582 pub fn take_from_args(args: &mut Vec<serde_json::Value>) -> Option<Self> {
583 Self::take_config_from_args(args).map(|cfg| Self::from_config_object(&cfg))
584 }
585}
586
587/// One pre-declared named projection from a CALL-site `scopes` map.
588///
589/// Named scopes are how a guest algorithm reaches more than one view of the
590/// store. They are declared at the call site and built by the host *before* the
591/// guest runs, which is the point: a guest that could project on demand could
592/// project in a loop, and projection is `O(V+E)` storage work that the native
593/// work meter does not govern.
594#[derive(Clone, Debug)]
595pub struct GraphScopeSpec {
596 /// The name the guest passes to `graph_named`.
597 pub name: String,
598 /// Native knobs for this scope, parsed by [`GraphProjectionSpec::from_config_object`].
599 pub spec: GraphProjectionSpec,
600 /// The raw scope object when it names a Cypher/Named projection, to be
601 /// resolved through the host's injected resolver rather than scanned.
602 pub graph_ref: Option<serde_json::Value>,
603}
604
605/// Every graph-projection knob, enumerated so the guest/native surface contract
606/// is checked at compile time: adding a knob forces a classification in
607/// [`ProjectionKnob::reach`] and a key in [`ProjectionKnob::config_key`] (both
608/// wildcard-free `match`es), mirroring the capability
609/// `every_variant_classified_exactly_once` exhaustiveness test. This is the
610/// anti-drift guard for the guest/native gap issue #151 exposed. See
611/// `docs/proposals/graphcompute_projection_parity_2026-07-19.md` §4.
612#[derive(Clone, Copy, Debug, PartialEq, Eq)]
613pub enum ProjectionKnob {
614 /// Scope to these vertex labels (`nodeLabels`).
615 NodeLabels,
616 /// Scope to these edge types (`edgeTypes` / `relationshipTypes`).
617 EdgeTypes,
618 /// Bind an edge property as the traversal weight (`weightProperty`).
619 WeightProperty,
620 /// Also build inbound adjacency (`includeReverse`).
621 IncludeReverse,
622 /// Materialize per-vertex property tensors (`nodeProperties`).
623 NodeProperties,
624 /// Materialize per-edge property tensors (`edgeProperties`).
625 EdgeProperties,
626 /// Cypher-mode node selection query (`nodeQuery`).
627 CypherNodeQuery,
628 /// Cypher-mode edge selection query (`edgeQuery`).
629 CypherEdgeQuery,
630 /// Cypher-mode weight column (`weightColumn`).
631 CypherWeightColumn,
632 /// Named pre-registered projection (`name`).
633 NamedGraph,
634}
635
636/// How a projection knob is reachable from a guest algorithm.
637#[derive(Clone, Copy, Debug, PartialEq, Eq)]
638pub enum KnobReach {
639 /// Parsed by the shared [`GraphProjectionSpec::from_config_object`] — the
640 /// Native + property-tensor knobs.
641 GuestNative,
642 /// Resolved by the uni-query projection seam from a Cypher/Named graphRef.
643 GuestQuerySeam,
644 /// Host/native-only — not expressible by a guest (none today; reserved for
645 /// future knobs such as orientation or parallel-edge aggregation).
646 HostOnly,
647}
648
649impl ProjectionKnob {
650 /// Every knob, the single source of truth the exhaustiveness test iterates.
651 pub const ALL: &'static [ProjectionKnob] = &[
652 ProjectionKnob::NodeLabels,
653 ProjectionKnob::EdgeTypes,
654 ProjectionKnob::WeightProperty,
655 ProjectionKnob::IncludeReverse,
656 ProjectionKnob::NodeProperties,
657 ProjectionKnob::EdgeProperties,
658 ProjectionKnob::CypherNodeQuery,
659 ProjectionKnob::CypherEdgeQuery,
660 ProjectionKnob::CypherWeightColumn,
661 ProjectionKnob::NamedGraph,
662 ];
663
664 /// The graphRef config key for this knob. Wildcard-free: a new variant fails
665 /// to compile here until it is given a key.
666 #[must_use]
667 pub fn config_key(self) -> &'static str {
668 match self {
669 ProjectionKnob::NodeLabels => "nodeLabels",
670 ProjectionKnob::EdgeTypes => "edgeTypes",
671 ProjectionKnob::WeightProperty => "weightProperty",
672 ProjectionKnob::IncludeReverse => "includeReverse",
673 ProjectionKnob::NodeProperties => "nodeProperties",
674 ProjectionKnob::EdgeProperties => "edgeProperties",
675 ProjectionKnob::CypherNodeQuery => "nodeQuery",
676 ProjectionKnob::CypherEdgeQuery => "edgeQuery",
677 ProjectionKnob::CypherWeightColumn => "weightColumn",
678 ProjectionKnob::NamedGraph => "name",
679 }
680 }
681
682 /// How a guest reaches this knob. Wildcard-free: a new variant fails to
683 /// compile here until it is classified.
684 #[must_use]
685 pub fn reach(self) -> KnobReach {
686 match self {
687 ProjectionKnob::NodeLabels
688 | ProjectionKnob::EdgeTypes
689 | ProjectionKnob::WeightProperty
690 | ProjectionKnob::IncludeReverse
691 | ProjectionKnob::NodeProperties
692 | ProjectionKnob::EdgeProperties => KnobReach::GuestNative,
693 ProjectionKnob::CypherNodeQuery
694 | ProjectionKnob::CypherEdgeQuery
695 | ProjectionKnob::CypherWeightColumn
696 | ProjectionKnob::NamedGraph => KnobReach::GuestQuerySeam,
697 }
698 }
699}
700
701#[cfg(test)]
702mod projection_knob_contract {
703 use super::{GraphProjectionSpec, KnobReach, ProjectionKnob};
704
705 #[test]
706 fn every_projection_knob_is_classified_and_keyed() {
707 // A knob added to the enum makes `reach`/`config_key` non-exhaustive
708 // (compile error) and must be appended to `ALL`. Keys must be unique and
709 // recognized as graphRef markers so the adapters strip them from guest
710 // args.
711 let mut keys = std::collections::HashSet::new();
712 for knob in ProjectionKnob::ALL {
713 let key = knob.config_key();
714 assert!(keys.insert(key), "duplicate projection config key {key}");
715 assert!(
716 GraphProjectionSpec::CONFIG_KEYS.contains(&key),
717 "{key} missing from GraphProjectionSpec::CONFIG_KEYS"
718 );
719 let _ = knob.reach(); // total by construction
720 }
721 }
722
723 #[test]
724 fn guest_native_knobs_round_trip_through_the_shared_parser() {
725 // Every GuestNative knob must actually be honored by from_config_object,
726 // so a knob can't be declared guest-reachable yet silently unparsed —
727 // the precise failure that produced issue #151.
728 for &knob in ProjectionKnob::ALL {
729 if knob.reach() != KnobReach::GuestNative {
730 continue;
731 }
732 let key = knob.config_key();
733 let sample = match knob {
734 ProjectionKnob::IncludeReverse => serde_json::json!(false),
735 ProjectionKnob::WeightProperty => serde_json::json!("w"),
736 _ => serde_json::json!(["X"]),
737 };
738 let mut obj = serde_json::Map::new();
739 obj.insert(key.to_string(), sample);
740 let spec = GraphProjectionSpec::from_config_object(&obj);
741 let honored = match knob {
742 ProjectionKnob::NodeLabels => spec.node_labels == ["X"],
743 ProjectionKnob::EdgeTypes => spec.edge_types == ["X"],
744 ProjectionKnob::WeightProperty => spec.weight_property.as_deref() == Some("w"),
745 ProjectionKnob::IncludeReverse => !spec.include_reverse,
746 ProjectionKnob::NodeProperties => spec.node_properties == ["X"],
747 ProjectionKnob::EdgeProperties => spec.edge_properties == ["X"],
748 _ => unreachable!("only GuestNative knobs reach here"),
749 };
750 assert!(honored, "from_config_object did not honor `{key}`");
751 }
752 }
753}
754
755/// Stable, read-only topology view handed to a plugin algorithm.
756///
757/// Vertices are addressed by dense `u32` slots (`0..vertex_count`);
758/// [`GraphView::to_vid`] / [`GraphView::to_slot`] translate to and from
759/// external [`Vid`]s at the boundary. Neighbor accessors return neighbor
760/// *slots*, not vids. A `GraphView` reflects the subgraph named by the
761/// [`GraphProjectionSpec`] that produced it and does not observe later
762/// writes.
763///
764/// # Panics
765///
766/// [`GraphView::out_weight`] panics unless [`GraphView::has_weights`] is
767/// `true`, and [`GraphView::in_neighbors`] / [`GraphView::in_degree`]
768/// panic unless [`GraphView::has_reverse`] is `true`. Guard with those
769/// predicates before calling.
770pub trait GraphView: Send + Sync {
771 /// Number of vertices; valid slots are `0..vertex_count`.
772 fn vertex_count(&self) -> usize;
773
774 /// Total number of outbound edges.
775 fn edge_count(&self) -> usize;
776
777 /// Outbound neighbor slots of `slot`.
778 fn out_neighbors(&self, slot: u32) -> &[u32];
779
780 /// Number of outbound edges from `slot`.
781 fn out_degree(&self, slot: u32) -> u32;
782
783 /// Inbound neighbor slots of `slot`.
784 ///
785 /// # Panics
786 ///
787 /// Panics unless [`GraphView::has_reverse`] is `true`.
788 fn in_neighbors(&self, slot: u32) -> &[u32];
789
790 /// Number of inbound edges into `slot`.
791 ///
792 /// # Panics
793 ///
794 /// Panics unless [`GraphView::has_reverse`] is `true`.
795 fn in_degree(&self, slot: u32) -> u32;
796
797 /// Whether inbound adjacency is available.
798 fn has_reverse(&self) -> bool;
799
800 /// Weight of the `edge_idx`-th outbound edge of `slot`.
801 ///
802 /// `edge_idx` indexes into [`GraphView::out_neighbors`] of `slot`.
803 ///
804 /// # Panics
805 ///
806 /// Panics unless [`GraphView::has_weights`] is `true`.
807 fn out_weight(&self, slot: u32, edge_idx: usize) -> f64;
808
809 /// Whether edge weights are available.
810 fn has_weights(&self) -> bool;
811
812 /// Translate a dense slot to its external [`Vid`].
813 fn to_vid(&self, slot: u32) -> Vid;
814
815 /// Translate an external [`Vid`] to its dense slot, if present.
816 fn to_slot(&self, vid: Vid) -> Option<u32>;
817
818 /// Iterate over every `(slot, vid)` pair in the view.
819 fn vertices(&self) -> Box<dyn Iterator<Item = (u32, Vid)> + '_>;
820}
821
822/// A black-box graph algorithm.
823///
824/// The trait is intentionally minimal: a signature describing the output,
825/// plus a `run` method returning a streaming `RecordBatch` sequence. The
826/// algorithm is responsible for fetching graph data via host APIs (out of
827/// scope of this trait — `uni-algo` will provide a `GraphView` abstraction
828/// the host adapter passes via `AlgorithmContext` once those APIs are
829/// available).
830pub trait AlgorithmProvider: Send + Sync {
831 /// Static signature.
832 fn signature(&self) -> &AlgorithmSignature;
833
834 /// Execute the algorithm.
835 ///
836 /// # Errors
837 ///
838 /// Returns [`FnError`] if the algorithm cannot be started; per-batch
839 /// failures are signaled via `Err` items in the returned stream.
840 fn run(&self, ctx: AlgorithmContext<'_>) -> Result<SendableRecordBatchStream, FnError>;
841}
842
843#[cfg(test)]
844mod tests {
845 use arrow_schema::DataType;
846 use datafusion::scalar::ScalarValue;
847
848 use super::{AlgorithmSignature, HOST_CAPABILITY_SLICES, SliceReq};
849 use crate::traits::procedure::NamedArgType;
850 use crate::traits::scalar::ArgType;
851
852 fn sig_with(args: Vec<NamedArgType>, slices: Vec<SliceReq>) -> AlgorithmSignature {
853 AlgorithmSignature {
854 args,
855 slices,
856 ..Default::default()
857 }
858 }
859
860 fn arg(name: &str, ty: ArgType, default: Option<ScalarValue>) -> NamedArgType {
861 NamedArgType {
862 name: name.into(),
863 ty,
864 default,
865 doc: String::new(),
866 }
867 }
868
869 fn cfg(json: &str) -> serde_json::Map<String, serde_json::Value> {
870 match serde_json::from_str(json).expect("valid json") {
871 serde_json::Value::Object(o) => o,
872 other => panic!("expected an object, got {other}"),
873 }
874 }
875
876 /// A `scopes`-only object must still be recognised as the projection config.
877 ///
878 /// If `scopes` were missing from `CONFIG_KEYS`, this object would not be
879 /// stripped and would arrive at the guest as a positional argument — a
880 /// silent arity shift rather than an error.
881 #[test]
882 fn a_scopes_only_object_is_recognised_as_the_projection_config() {
883 let mut args: Vec<serde_json::Value> =
884 serde_json::from_str(r#"[1, {"scopes": {"agg": {"nodeLabels": ["N"]}}}]"#)
885 .expect("valid json");
886 let spec = super::GraphProjectionSpec::take_from_args(&mut args);
887 assert!(
888 spec.is_some(),
889 "the trailing object must be taken as config"
890 );
891 assert_eq!(args.len(), 1, "only the guest's own arg may remain");
892 }
893
894 /// Each scope is parsed by the same Native parser as the primary, and a
895 /// Cypher/Named scope is carried through verbatim for the resolver.
896 #[test]
897 fn scopes_parse_per_scope_and_keep_their_mode() {
898 let scopes = super::GraphProjectionSpec::scopes_from_config_object(&cfg(r#"{"scopes": {
899 "agg": {"nodeLabels": ["Cell"], "edgeTypes": ["AGG"], "weightProperty": "w"},
900 "flow": {"nodeQuery": "MATCH (c) RETURN id(c) AS id"}
901 }}"#))
902 .expect("well-formed scopes");
903 assert_eq!(scopes.len(), 2);
904
905 let agg = scopes.iter().find(|s| s.name == "agg").expect("agg");
906 assert_eq!(agg.spec.node_labels, vec!["Cell".to_string()]);
907 assert_eq!(agg.spec.weight_property.as_deref(), Some("w"));
908 assert!(
909 agg.graph_ref.is_none(),
910 "a Native scope must not be routed to the resolver"
911 );
912
913 let flow = scopes.iter().find(|s| s.name == "flow").expect("flow");
914 assert!(
915 flow.graph_ref.is_some(),
916 "a Cypher scope must reach the resolver verbatim"
917 );
918 }
919
920 /// No `scopes` key is not an error — it is the ordinary single-graph CALL.
921 #[test]
922 fn an_absent_scopes_key_yields_no_scopes() {
923 let scopes =
924 super::GraphProjectionSpec::scopes_from_config_object(&cfg(r#"{"nodeLabels": ["N"]}"#))
925 .expect("no scopes is fine");
926 assert!(scopes.is_empty());
927 }
928
929 /// Malformed scope maps are named, not silently dropped.
930 #[test]
931 fn a_malformed_scopes_map_is_rejected_with_the_offending_name() {
932 for (json, needle) in [
933 (r#"{"scopes": ["agg"]}"#, "must be an object"),
934 (r#"{"scopes": {"agg": 7}}"#, "agg"),
935 (r#"{"scopes": {"": {}}}"#, "must not be empty"),
936 // `graph` would shadow the primary handle's own accessor.
937 (r#"{"scopes": {"graph": {}}}"#, "primary projection"),
938 ] {
939 let err = super::GraphProjectionSpec::scopes_from_config_object(&cfg(json))
940 .expect_err("must be rejected");
941 assert!(
942 err.contains(needle),
943 "error for {json} must mention `{needle}`, got: {err}"
944 );
945 }
946 }
947
948 #[test]
949 fn check_slices_accepts_available_and_rejects_missing() {
950 // graph-compute@1 is the host surface; @1 passes, @2 and unknown fail 0x86A.
951 let ok = sig_with(
952 vec![],
953 vec![SliceReq {
954 slice: "graph-compute".into(),
955 version: 1,
956 }],
957 );
958 assert!(ok.check_slices(HOST_CAPABILITY_SLICES).is_ok());
959
960 let too_new = sig_with(
961 vec![],
962 vec![SliceReq {
963 slice: "graph-compute".into(),
964 version: 2,
965 }],
966 );
967 let err = too_new
968 .check_slices(HOST_CAPABILITY_SLICES)
969 .expect_err("graph-compute@2 must be refused");
970 assert_eq!(err.code, 0x86A, "slice mismatch is 0x86A");
971
972 let unknown = sig_with(
973 vec![],
974 vec![SliceReq {
975 slice: "tensor-compute".into(),
976 version: 1,
977 }],
978 );
979 assert_eq!(
980 unknown
981 .check_slices(HOST_CAPABILITY_SLICES)
982 .unwrap_err()
983 .code,
984 0x86A
985 );
986
987 // No declared slices is grandfathered onto the base surface.
988 assert!(
989 sig_with(vec![], vec![])
990 .check_slices(HOST_CAPABILITY_SLICES)
991 .is_ok()
992 );
993 }
994
995 #[test]
996 fn coerce_config_passes_through_when_untyped() {
997 // Empty `args` preserves the legacy raw contract byte-for-byte.
998 let s = sig_with(vec![], vec![]);
999 assert_eq!(s.coerce_config_json("[1, 2, 3]").unwrap(), "[1, 2, 3]");
1000 }
1001
1002 #[test]
1003 fn coerce_config_fills_defaults_and_validates() {
1004 let s = sig_with(
1005 vec![
1006 arg("src", ArgType::CypherValue, None),
1007 arg(
1008 "alpha",
1009 ArgType::Primitive(DataType::Float64),
1010 Some(ScalarValue::Float64(Some(0.85))),
1011 ),
1012 ],
1013 vec![],
1014 );
1015
1016 // A single provided arg fills the omitted `alpha` default.
1017 let out = s.coerce_config_json("[5]").unwrap();
1018 let arr: Vec<serde_json::Value> = serde_json::from_str(&out).unwrap();
1019 assert_eq!(arr.len(), 2, "the omitted default is appended");
1020 assert_eq!(arr[0], serde_json::json!(5));
1021 assert!((arr[1].as_f64().unwrap() - 0.85).abs() < 1e-12);
1022
1023 // A missing required arg is rejected.
1024 let err = s.coerce_config_json("[]").expect_err("src is required");
1025 assert_eq!(err.code, 0x86E);
1026
1027 // A wrong-typed alpha (string, not number) is rejected.
1028 let err = s
1029 .coerce_config_json(r#"[5, "not-a-number"]"#)
1030 .expect_err("alpha must be numeric");
1031 assert_eq!(err.code, 0x86E);
1032
1033 // Too many positional args is rejected.
1034 assert_eq!(s.coerce_config_json("[5, 0.9, 1]").unwrap_err().code, 0x86E);
1035
1036 // A CypherValue arg accepts an array (the `sourceVids` shape).
1037 let arr_src = s.coerce_config_json("[[1, 2, 3], 0.9]").unwrap();
1038 let parsed: Vec<serde_json::Value> = serde_json::from_str(&arr_src).unwrap();
1039 assert!(parsed[0].is_array(), "CypherValue accepts an array");
1040 }
1041}