use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use async_trait::async_trait;
use chrono::{DateTime, TimeZone, Utc};
use uuid::Uuid;
use khive_storage::error::StorageError;
use khive_storage::types::{
BatchWriteSummary, DeleteMode, DirectedNeighborHit, Direction, Edge, EdgeFilter, EdgeSeekPage,
EdgeSortField, GraphPath, GuardedBatchOutcome, GuardedBatchRefusal, GuardedWriteOutcome,
MissingEndpoints, NeighborHit, NeighborQuery, Page, PageRequest, PathNode, SortDirection,
SortOrder, SqlStatement, SqlValue, TraversalRequest,
};
use khive_storage::GraphStore;
use khive_storage::LinkId;
use khive_storage::StorageCapability;
use khive_types::EdgeRelation;
use crate::error::SqliteError;
use crate::pool::ConnectionPool;
use crate::sql_bridge::bind_params;
use crate::writer_task::WriterTaskHandle;
fn map_err(e: rusqlite::Error, op: &'static str) -> StorageError {
StorageError::driver(StorageCapability::Graph, op, e)
}
fn map_sqlite_err(e: SqliteError, op: &'static str) -> StorageError {
StorageError::driver(StorageCapability::Graph, op, e)
}
const EDGE_NATURAL_KEY_CONFLICT_SET: &str = "weight = excluded.weight, \
updated_at = excluded.updated_at, \
deleted_at = NULL, \
metadata = excluded.metadata, \
target_backend = excluded.target_backend";
fn endpoint_exists_clause(id_param: &str) -> String {
format!(
"EXISTS (SELECT 1 FROM entities WHERE id = {id_param} AND deleted_at IS NULL) \
OR EXISTS (SELECT 1 FROM notes WHERE id = {id_param} AND deleted_at IS NULL) \
OR EXISTS (SELECT 1 FROM events WHERE id = {id_param}) \
OR EXISTS (SELECT 1 FROM graph_edges WHERE id = {id_param} AND deleted_at IS NULL)"
)
}
pub fn edge_upsert_statement(edge: &Edge) -> SqlStatement {
let (source_id, target_id) =
canonical_edge_endpoints(edge.relation, edge.source_id, edge.target_id);
let metadata_str = edge
.metadata
.as_ref()
.map(|v| serde_json::to_string(v).unwrap_or_default());
SqlStatement {
sql: format!(
"INSERT INTO graph_edges \
(namespace, id, source_id, target_id, relation, weight, \
created_at, updated_at, deleted_at, metadata, target_backend) \
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11) \
ON CONFLICT(namespace, id) DO UPDATE SET \
source_id = excluded.source_id, \
target_id = excluded.target_id, \
relation = excluded.relation, \
{EDGE_NATURAL_KEY_CONFLICT_SET} \
ON CONFLICT(namespace, source_id, target_id, relation) DO UPDATE SET \
{EDGE_NATURAL_KEY_CONFLICT_SET}"
),
params: vec![
SqlValue::Text(edge.namespace.clone()),
SqlValue::Text(Uuid::from(edge.id).to_string()),
SqlValue::Text(source_id.to_string()),
SqlValue::Text(target_id.to_string()),
SqlValue::Text(edge.relation.to_string()),
SqlValue::Float(edge.weight),
SqlValue::Integer(edge.created_at.timestamp_micros()),
SqlValue::Integer(edge.updated_at.timestamp_micros()),
match edge.deleted_at {
Some(t) => SqlValue::Integer(t.timestamp_micros()),
None => SqlValue::Null,
},
match metadata_str {
Some(m) => SqlValue::Text(m),
None => SqlValue::Null,
},
match &edge.target_backend {
Some(b) => SqlValue::Text(b.clone()),
None => SqlValue::Null,
},
],
label: Some("edge-upsert".to_string()),
}
}
#[allow(clippy::too_many_arguments)]
pub fn edge_insert_guarded_by_endpoints_statement(
namespace: &str,
edge_id: Uuid,
source_id: Uuid,
target_id: Uuid,
relation: EdgeRelation,
weight: f64,
now: i64,
metadata: Option<&str>,
) -> SqlStatement {
let src_exists = endpoint_exists_clause("?3");
let tgt_exists = endpoint_exists_clause("?4");
SqlStatement {
sql: format!(
"INSERT INTO graph_edges \
(namespace, id, source_id, target_id, relation, weight, \
created_at, updated_at, metadata) \
SELECT ?1, ?2, ?3, ?4, ?5, ?6, ?7, ?7, ?8 \
WHERE ({src_exists}) AND ({tgt_exists}) \
ON CONFLICT(namespace, source_id, target_id, relation) DO UPDATE SET \
{EDGE_NATURAL_KEY_CONFLICT_SET}"
),
params: vec![
SqlValue::Text(namespace.to_string()),
SqlValue::Text(edge_id.to_string()),
SqlValue::Text(source_id.to_string()),
SqlValue::Text(target_id.to_string()),
SqlValue::Text(relation.as_str().to_string()),
SqlValue::Float(weight),
SqlValue::Integer(now),
match metadata {
Some(m) => SqlValue::Text(m.to_string()),
None => SqlValue::Null,
},
],
label: Some("atomic-link-insert-edge-where-exists".to_string()),
}
}
pub fn edge_soft_delete_statement(id: Uuid, now: i64) -> SqlStatement {
SqlStatement {
sql: "UPDATE graph_edges SET deleted_at = ?2, updated_at = ?2 \
WHERE id = ?1 AND deleted_at IS NULL"
.to_string(),
params: vec![SqlValue::Text(id.to_string()), SqlValue::Integer(now)],
label: Some("edge-delete-soft".to_string()),
}
}
pub fn edge_hard_delete_statement(id: Uuid) -> SqlStatement {
SqlStatement {
sql: "DELETE FROM graph_edges WHERE id = ?1".to_string(),
params: vec![SqlValue::Text(id.to_string())],
label: Some("edge-delete-hard".to_string()),
}
}
pub fn purge_incident_edges_statement(node_id: Uuid) -> SqlStatement {
SqlStatement {
sql: "DELETE FROM graph_edges WHERE source_id = ?1 OR target_id = ?1".to_string(),
params: vec![SqlValue::Text(node_id.to_string())],
label: Some("edge-purge-incident".to_string()),
}
}
pub const EDGE_SYMMETRIC_CONFLICT_PROBE_SQL: &str = "SELECT id FROM graph_edges \
WHERE namespace = ?1 AND source_id = ?2 AND target_id = ?3 \
AND relation = ?4 AND id != ?5";
pub const EDGE_SYMMETRIC_DELETE_NONCANONICAL_SQL: &str =
"DELETE FROM graph_edges WHERE namespace = ?1 AND id = ?2";
pub const EDGE_SYMMETRIC_REFRESH_CANONICAL_SQL: &str = "UPDATE graph_edges SET \
weight = ?1, updated_at = ?2, deleted_at = NULL, \
target_backend = ?3, metadata = ?4 \
WHERE namespace = ?5 AND id = ?6";
pub const EDGE_SYMMETRIC_UPDATE_INPLACE_SQL: &str = "UPDATE graph_edges SET \
source_id = ?1, target_id = ?2, relation = ?3, \
weight = ?4, updated_at = ?5, metadata = ?6 \
WHERE namespace = ?7 AND id = ?8";
pub fn edge_symmetric_conflict_probe_statement(
namespace: &str,
canon_src: Uuid,
canon_tgt: Uuid,
relation: EdgeRelation,
exclude_id: Uuid,
) -> SqlStatement {
SqlStatement {
sql: EDGE_SYMMETRIC_CONFLICT_PROBE_SQL.to_string(),
params: vec![
SqlValue::Text(namespace.to_string()),
SqlValue::Text(canon_src.to_string()),
SqlValue::Text(canon_tgt.to_string()),
SqlValue::Text(relation.to_string()),
SqlValue::Text(exclude_id.to_string()),
],
label: Some("edge-symmetric-conflict-probe".to_string()),
}
}
pub fn edge_symmetric_delete_noncanonical_statement(namespace: &str, id: Uuid) -> SqlStatement {
SqlStatement {
sql: EDGE_SYMMETRIC_DELETE_NONCANONICAL_SQL.to_string(),
params: vec![
SqlValue::Text(namespace.to_string()),
SqlValue::Text(id.to_string()),
],
label: Some("edge-symmetric-delete-noncanonical".to_string()),
}
}
#[allow(clippy::too_many_arguments)]
pub fn edge_symmetric_refresh_canonical_statement(
namespace: &str,
existing_id: Uuid,
weight: f64,
updated_at_micros: i64,
target_backend: Option<&str>,
metadata: Option<&str>,
) -> SqlStatement {
SqlStatement {
sql: EDGE_SYMMETRIC_REFRESH_CANONICAL_SQL.to_string(),
params: vec![
SqlValue::Float(weight),
SqlValue::Integer(updated_at_micros),
match target_backend {
Some(b) => SqlValue::Text(b.to_string()),
None => SqlValue::Null,
},
match metadata {
Some(m) => SqlValue::Text(m.to_string()),
None => SqlValue::Null,
},
SqlValue::Text(namespace.to_string()),
SqlValue::Text(existing_id.to_string()),
],
label: Some("edge-symmetric-refresh-canonical".to_string()),
}
}
#[allow(clippy::too_many_arguments)]
pub fn edge_symmetric_update_inplace_statement(
namespace: &str,
id: Uuid,
canon_src: Uuid,
canon_tgt: Uuid,
relation: EdgeRelation,
weight: f64,
updated_at_micros: i64,
metadata: Option<&str>,
) -> SqlStatement {
SqlStatement {
sql: EDGE_SYMMETRIC_UPDATE_INPLACE_SQL.to_string(),
params: vec![
SqlValue::Text(canon_src.to_string()),
SqlValue::Text(canon_tgt.to_string()),
SqlValue::Text(relation.to_string()),
SqlValue::Float(weight),
SqlValue::Integer(updated_at_micros),
match metadata {
Some(m) => SqlValue::Text(m.to_string()),
None => SqlValue::Null,
},
SqlValue::Text(namespace.to_string()),
SqlValue::Text(id.to_string()),
],
label: Some("edge-symmetric-update-inplace".to_string()),
}
}
pub fn edge_symmetric_delete_if_conflict_statement(
namespace: &str,
id: Uuid,
canon_src: Uuid,
canon_tgt: Uuid,
relation: EdgeRelation,
) -> SqlStatement {
SqlStatement {
sql: "DELETE FROM graph_edges \
WHERE namespace = ?1 AND id = ?2 \
AND EXISTS ( \
SELECT 1 FROM graph_edges \
WHERE namespace = ?1 AND source_id = ?3 AND target_id = ?4 \
AND relation = ?5 AND id != ?2 \
)"
.to_string(),
params: vec![
SqlValue::Text(namespace.to_string()),
SqlValue::Text(id.to_string()),
SqlValue::Text(canon_src.to_string()),
SqlValue::Text(canon_tgt.to_string()),
SqlValue::Text(relation.to_string()),
],
label: Some("edge-symmetric-delete-if-conflict".to_string()),
}
}
#[allow(clippy::too_many_arguments)]
pub fn edge_symmetric_refresh_or_update_inplace_statement(
namespace: &str,
id: Uuid,
canon_src: Uuid,
canon_tgt: Uuid,
relation: EdgeRelation,
weight: f64,
updated_at_micros: i64,
metadata: Option<&str>,
target_backend: Option<&str>,
) -> SqlStatement {
SqlStatement {
sql: "UPDATE graph_edges SET \
source_id = ?3, target_id = ?4, relation = ?5, \
weight = ?6, updated_at = ?7, deleted_at = NULL, metadata = ?8, \
target_backend = CASE WHEN changes() = 1 THEN ?9 ELSE target_backend END \
WHERE namespace = ?1 \
AND ( \
(id = ?2 AND changes() = 0) \
OR (source_id = ?3 AND target_id = ?4 AND relation = ?5 \
AND id != ?2 AND changes() = 1) \
)"
.to_string(),
params: vec![
SqlValue::Text(namespace.to_string()),
SqlValue::Text(id.to_string()),
SqlValue::Text(canon_src.to_string()),
SqlValue::Text(canon_tgt.to_string()),
SqlValue::Text(relation.to_string()),
SqlValue::Float(weight),
SqlValue::Integer(updated_at_micros),
match metadata {
Some(m) => SqlValue::Text(m.to_string()),
None => SqlValue::Null,
},
match target_backend {
Some(b) => SqlValue::Text(b.to_string()),
None => SqlValue::Null,
},
],
label: Some("edge-symmetric-refresh-or-update-inplace".to_string()),
}
}
pub struct SqlGraphStore {
pool: Arc<ConnectionPool>,
is_file_backed: bool,
namespace: String,
writer_task: Option<WriterTaskHandle>,
}
impl SqlGraphStore {
pub fn new_scoped(
pool: Arc<ConnectionPool>,
is_file_backed: bool,
namespace: impl Into<String>,
) -> Self {
let writer_task = pool.writer_task_handle().ok().flatten();
Self {
pool,
is_file_backed,
namespace: namespace.into(),
writer_task,
}
}
fn open_standalone_writer(&self) -> Result<rusqlite::Connection, StorageError> {
self.pool
.open_standalone_writer()
.map_err(|e| map_sqlite_err(e, "open_graph_writer"))
}
fn open_standalone_reader(&self) -> Result<rusqlite::Connection, StorageError> {
self.pool
.open_standalone_reader()
.map_err(|e| map_sqlite_err(e, "open_graph_reader"))
}
async fn with_writer<F, R>(&self, op: &'static str, f: F) -> Result<R, StorageError>
where
F: FnOnce(&rusqlite::Connection) -> Result<R, rusqlite::Error> + Send + 'static,
R: Send + 'static,
{
if let Some(writer_task) = &self.writer_task {
return writer_task
.send(move |conn| f(conn).map_err(|e| map_err(e, op)))
.await;
}
if self.is_file_backed {
let conn = self.open_standalone_writer()?;
tokio::task::spawn_blocking(move || f(&conn).map_err(|e| map_err(e, op)))
.await
.map_err(|e| StorageError::driver(StorageCapability::Graph, op, e))?
} else {
let pool = Arc::clone(&self.pool);
tokio::task::spawn_blocking(move || {
let guard = pool.try_writer().map_err(|e| map_sqlite_err(e, op))?;
f(guard.conn()).map_err(|e| map_err(e, op))
})
.await
.map_err(|e| StorageError::driver(StorageCapability::Graph, op, e))?
}
}
async fn with_reader<F, R>(&self, op: &'static str, f: F) -> Result<R, StorageError>
where
F: FnOnce(&rusqlite::Connection) -> Result<R, rusqlite::Error> + Send + 'static,
R: Send + 'static,
{
if self.is_file_backed {
let conn = self.open_standalone_reader()?;
tokio::task::spawn_blocking(move || f(&conn).map_err(|e| map_err(e, op)))
.await
.map_err(|e| StorageError::driver(StorageCapability::Graph, op, e))?
} else {
let pool = Arc::clone(&self.pool);
tokio::task::spawn_blocking(move || {
let guard = pool.reader().map_err(|e| map_sqlite_err(e, op))?;
f(guard.conn()).map_err(|e| map_err(e, op))
})
.await
.map_err(|e| StorageError::driver(StorageCapability::Graph, op, e))?
}
}
}
fn read_edge(row: &rusqlite::Row<'_>) -> Result<Edge, rusqlite::Error> {
let namespace: String = row.get(0)?;
let id_str: String = row.get(1)?;
let source_str: String = row.get(2)?;
let target_str: String = row.get(3)?;
let relation_str: String = row.get(4)?;
let weight: f64 = row.get(5)?;
let created_micros: i64 = row.get(6)?;
let updated_micros: i64 = row.get(7)?;
let deleted_micros: Option<i64> = row.get(8)?;
let metadata_str: Option<String> = row.get(9)?;
let target_backend: Option<String> = row.get(10)?;
let id = parse_uuid(&id_str)?;
let source_id = parse_uuid(&source_str)?;
let target_id = parse_uuid(&target_str)?;
let created_at = micros_to_datetime(created_micros);
let relation = relation_str.parse::<EdgeRelation>().map_err(|e| {
rusqlite::Error::FromSqlConversionFailure(4, rusqlite::types::Type::Text, Box::new(e))
})?;
let metadata = match metadata_str {
Some(s) => {
let v = serde_json::from_str(&s).map_err(|e| {
rusqlite::Error::FromSqlConversionFailure(
9,
rusqlite::types::Type::Text,
Box::new(e),
)
})?;
Some(v)
}
None => None,
};
Ok(Edge {
id: id.into(),
namespace,
source_id,
target_id,
relation,
weight,
created_at,
updated_at: micros_to_datetime(updated_micros),
deleted_at: deleted_micros.map(micros_to_datetime),
metadata,
target_backend,
})
}
fn parse_uuid(s: &str) -> Result<Uuid, rusqlite::Error> {
Uuid::parse_str(s).map_err(|e| {
rusqlite::Error::FromSqlConversionFailure(0, rusqlite::types::Type::Text, Box::new(e))
})
}
fn neighbor_extra_clause(
query: &NeighborQuery,
start_param_idx: usize,
) -> (String, String, Vec<Box<dyn rusqlite::types::ToSql>>) {
let mut conditions: Vec<String> = Vec::new();
let mut extra_params: Vec<Box<dyn rusqlite::types::ToSql>> = Vec::new();
let mut param_idx = start_param_idx;
if let Some(ref rels) = query.relations {
if !rels.is_empty() {
let placeholders: Vec<String> = rels
.iter()
.map(|r| {
extra_params.push(Box::new(r.to_string()));
let p = format!("?{}", param_idx);
param_idx += 1;
p
})
.collect();
conditions.push(format!("relation IN ({})", placeholders.join(",")));
}
}
if let Some(min_w) = query.min_weight {
extra_params.push(Box::new(min_w));
conditions.push(format!("weight >= ?{}", param_idx));
param_idx += 1;
}
let where_extra = if conditions.is_empty() {
String::new()
} else {
format!(" WHERE {}", conditions.join(" AND "))
};
let limit_clause = if let Some(lim) = query.limit {
extra_params.push(Box::new(lim as i64));
format!(" LIMIT ?{}", param_idx)
} else {
String::new()
};
(where_extra, limit_clause, extra_params)
}
#[cfg(test)]
static NEIGHBOR_SELECT_COUNT: std::sync::atomic::AtomicUsize =
std::sync::atomic::AtomicUsize::new(0);
#[cfg(test)]
fn count_neighbor_select() {
NEIGHBOR_SELECT_COUNT.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
}
#[cfg(not(test))]
fn count_neighbor_select() {}
#[cfg(test)]
pub(crate) fn reset_neighbor_select_count() {
NEIGHBOR_SELECT_COUNT.store(0, std::sync::atomic::Ordering::Relaxed);
}
#[cfg(test)]
pub(crate) fn neighbor_select_count() -> usize {
NEIGHBOR_SELECT_COUNT.load(std::sync::atomic::Ordering::Relaxed)
}
fn micros_to_datetime(micros: i64) -> DateTime<Utc> {
Utc.timestamp_micros(micros)
.single()
.unwrap_or_else(Utc::now)
}
fn build_edge_filter_sql(
namespace: &str,
filter: &EdgeFilter,
) -> (String, Vec<Box<dyn rusqlite::types::ToSql>>) {
let mut conditions: Vec<String> = vec![
"namespace = ?1".to_string(),
"deleted_at IS NULL".to_string(),
];
let mut params: Vec<Box<dyn rusqlite::types::ToSql>> = vec![Box::new(namespace.to_string())];
if !filter.ids.is_empty() {
let placeholders: Vec<String> = filter
.ids
.iter()
.map(|id| {
params.push(Box::new(id.to_string()));
format!("?{}", params.len())
})
.collect();
conditions.push(format!("id IN ({})", placeholders.join(",")));
}
if !filter.source_ids.is_empty() {
let placeholders: Vec<String> = filter
.source_ids
.iter()
.map(|id| {
params.push(Box::new(id.to_string()));
format!("?{}", params.len())
})
.collect();
conditions.push(format!("source_id IN ({})", placeholders.join(",")));
}
if !filter.target_ids.is_empty() {
let placeholders: Vec<String> = filter
.target_ids
.iter()
.map(|id| {
params.push(Box::new(id.to_string()));
format!("?{}", params.len())
})
.collect();
conditions.push(format!("target_id IN ({})", placeholders.join(",")));
}
if !filter.relations.is_empty() {
let placeholders: Vec<String> = filter
.relations
.iter()
.map(|r| {
params.push(Box::new(r.to_string()));
format!("?{}", params.len())
})
.collect();
conditions.push(format!("relation IN ({})", placeholders.join(",")));
}
if let Some(min_w) = filter.min_weight {
params.push(Box::new(min_w));
conditions.push(format!("weight >= ?{}", params.len()));
}
if let Some(max_w) = filter.max_weight {
params.push(Box::new(max_w));
conditions.push(format!("weight <= ?{}", params.len()));
}
if let Some(ref time_range) = filter.created_at {
if let Some(start) = time_range.start {
params.push(Box::new(start.timestamp_micros()));
conditions.push(format!("created_at >= ?{}", params.len()));
}
if let Some(end) = time_range.end {
params.push(Box::new(end.timestamp_micros()));
conditions.push(format!("created_at < ?{}", params.len()));
}
}
let clause = format!(" WHERE {}", conditions.join(" AND "));
(clause, params)
}
fn edge_sort_col(field: &EdgeSortField) -> &'static str {
match field {
EdgeSortField::CreatedAt => "created_at",
EdgeSortField::Weight => "weight",
EdgeSortField::Relation => "relation",
}
}
fn canonical_edge_endpoints(
relation: EdgeRelation,
source_id: Uuid,
target_id: Uuid,
) -> (Uuid, Uuid) {
if relation.is_symmetric() && target_id < source_id {
(target_id, source_id)
} else {
(source_id, target_id)
}
}
fn batch_upsert_edges(
conn: &rusqlite::Connection,
edges: &[Edge],
attempted: u64,
) -> Result<BatchWriteSummary, rusqlite::Error> {
let mut affected = 0u64;
for edge in edges {
let statement = edge_upsert_statement(edge);
let mut stmt = conn.prepare(&statement.sql)?;
bind_params(&mut stmt, &statement.params)?;
stmt.raw_execute()?;
affected += 1;
}
Ok(BatchWriteSummary {
attempted,
affected,
failed: 0,
first_error: String::new(),
})
}
fn edge_endpoints_exist(
conn: &rusqlite::Connection,
source_id: Uuid,
target_id: Uuid,
) -> Result<MissingEndpoints, rusqlite::Error> {
let src_exists = endpoint_exists_clause("?1");
let tgt_exists = endpoint_exists_clause("?2");
let sql = format!("SELECT ({src_exists}), ({tgt_exists})");
conn.query_row(
&sql,
rusqlite::params![source_id.to_string(), target_id.to_string()],
|row| {
let src_exists: bool = row.get(0)?;
let tgt_exists: bool = row.get(1)?;
Ok(MissingEndpoints {
source: !src_exists,
target: !tgt_exists,
})
},
)
}
fn edge_insert_guarded(
conn: &rusqlite::Connection,
statement: &SqlStatement,
source_id: Uuid,
target_id: Uuid,
) -> Result<GuardedWriteOutcome, rusqlite::Error> {
let mut stmt = conn.prepare(&statement.sql)?;
bind_params(&mut stmt, &statement.params)?;
if stmt.raw_execute()? > 0 {
return Ok(GuardedWriteOutcome::Written);
}
#[cfg(test)]
tests::insert_probe_seam::hook((source_id, target_id));
let missing = edge_endpoints_exist(conn, source_id, target_id)?;
Ok(GuardedWriteOutcome::Refused(missing))
}
fn batch_upsert_edges_guarded(
conn: &rusqlite::Connection,
edges: &[Edge],
attempted: u64,
) -> Result<GuardedBatchOutcome, rusqlite::Error> {
for (index, edge) in edges.iter().enumerate() {
let (source_id, target_id) =
canonical_edge_endpoints(edge.relation, edge.source_id, edge.target_id);
let missing = edge_endpoints_exist(conn, source_id, target_id)?;
if missing.any() {
return Ok(GuardedBatchOutcome {
summary: BatchWriteSummary {
attempted,
affected: 0,
failed: attempted,
first_error: format!(
"batch entry {index}: edge endpoint no longer exists at write time: source {source_id} or target {target_id}"
),
},
refused: Some(GuardedBatchRefusal {
entry_index: index,
missing,
}),
});
}
}
let mut affected = 0u64;
for edge in edges {
let statement = edge_upsert_statement(edge);
let mut stmt = conn.prepare(&statement.sql)?;
bind_params(&mut stmt, &statement.params)?;
stmt.raw_execute()?;
affected += 1;
}
Ok(GuardedBatchOutcome {
summary: BatchWriteSummary {
attempted,
affected,
failed: 0,
first_error: String::new(),
},
refused: None,
})
}
#[async_trait]
impl GraphStore for SqlGraphStore {
async fn upsert_edge(&self, edge: Edge) -> Result<(), StorageError> {
let statement = edge_upsert_statement(&edge);
self.with_writer("upsert_edge", move |conn| {
let mut stmt = conn.prepare(&statement.sql)?;
bind_params(&mut stmt, &statement.params)?;
stmt.raw_execute()?;
Ok(())
})
.await
}
async fn upsert_edges(&self, edges: Vec<Edge>) -> Result<BatchWriteSummary, StorageError> {
let attempted = edges.len() as u64;
if let Some(writer_task) = &self.writer_task {
return writer_task
.send(move |conn| {
batch_upsert_edges(conn, &edges, attempted)
.map_err(|e| map_err(e, "upsert_edges"))
})
.await;
}
self.with_writer("upsert_edges", move |conn| {
conn.execute_batch("BEGIN IMMEDIATE")?;
let _tx_handle =
khive_storage::tx_registry::register(Some("graph_upsert_edges".to_string()));
let summary = match batch_upsert_edges(conn, &edges, attempted) {
Ok(summary) => summary,
Err(e) => {
let _ = conn.execute_batch("ROLLBACK");
return Err(e);
}
};
if let Err(e) = conn.execute_batch("COMMIT") {
let _ = conn.execute_batch("ROLLBACK");
return Err(e);
}
Ok(summary)
})
.await
}
async fn upsert_edge_guarded(&self, edge: Edge) -> Result<GuardedWriteOutcome, StorageError> {
let (source_id, target_id) =
canonical_edge_endpoints(edge.relation, edge.source_id, edge.target_id);
let metadata_str = edge
.metadata
.as_ref()
.map(|v| serde_json::to_string(v).unwrap_or_default());
let statement = edge_insert_guarded_by_endpoints_statement(
&edge.namespace,
Uuid::from(edge.id),
source_id,
target_id,
edge.relation,
edge.weight,
edge.created_at.timestamp_micros(),
metadata_str.as_deref(),
);
if let Some(writer_task) = &self.writer_task {
return writer_task
.send(move |conn| {
edge_insert_guarded(conn, &statement, source_id, target_id)
.map_err(|e| map_err(e, "upsert_edge_guarded"))
})
.await;
}
self.with_writer("upsert_edge_guarded", move |conn| {
conn.execute_batch("BEGIN IMMEDIATE")?;
let _tx_handle =
khive_storage::tx_registry::register(Some("graph_upsert_edge_guarded".to_string()));
let outcome = match edge_insert_guarded(conn, &statement, source_id, target_id) {
Ok(outcome) => outcome,
Err(e) => {
let _ = conn.execute_batch("ROLLBACK");
return Err(e);
}
};
if let Err(e) = conn.execute_batch("COMMIT") {
let _ = conn.execute_batch("ROLLBACK");
return Err(e);
}
Ok(outcome)
})
.await
}
async fn upsert_edges_guarded(
&self,
edges: Vec<Edge>,
) -> Result<GuardedBatchOutcome, StorageError> {
let attempted = edges.len() as u64;
if let Some(writer_task) = &self.writer_task {
return writer_task
.send(move |conn| {
batch_upsert_edges_guarded(conn, &edges, attempted)
.map_err(|e| map_err(e, "upsert_edges_guarded"))
})
.await;
}
self.with_writer("upsert_edges_guarded", move |conn| {
conn.execute_batch("BEGIN IMMEDIATE")?;
let _tx_handle = khive_storage::tx_registry::register(Some(
"graph_upsert_edges_guarded".to_string(),
));
let summary = match batch_upsert_edges_guarded(conn, &edges, attempted) {
Ok(summary) => summary,
Err(e) => {
let _ = conn.execute_batch("ROLLBACK");
return Err(e);
}
};
if let Err(e) = conn.execute_batch("COMMIT") {
let _ = conn.execute_batch("ROLLBACK");
return Err(e);
}
Ok(summary)
})
.await
}
async fn get_edge(&self, id: LinkId) -> Result<Option<Edge>, StorageError> {
let id_str = Uuid::from(id).to_string();
self.with_reader("get_edge", move |conn| {
let mut stmt = conn.prepare(
"SELECT namespace, id, source_id, target_id, relation, weight, \
created_at, updated_at, deleted_at, metadata, target_backend \
FROM graph_edges WHERE id = ?1 AND deleted_at IS NULL",
)?;
let mut rows = stmt.query(rusqlite::params![id_str])?;
match rows.next()? {
Some(row) => Ok(Some(read_edge(row)?)),
None => Ok(None),
}
})
.await
}
async fn get_edge_including_deleted(&self, id: LinkId) -> Result<Option<Edge>, StorageError> {
let id_str = Uuid::from(id).to_string();
self.with_reader("get_edge_including_deleted", move |conn| {
let mut stmt = conn.prepare(
"SELECT namespace, id, source_id, target_id, relation, weight, \
created_at, updated_at, deleted_at, metadata, target_backend \
FROM graph_edges WHERE id = ?1",
)?;
let mut rows = stmt.query(rusqlite::params![id_str])?;
match rows.next()? {
Some(row) => Ok(Some(read_edge(row)?)),
None => Ok(None),
}
})
.await
}
async fn get_edges(&self, ids: &[LinkId]) -> Result<Vec<Edge>, StorageError> {
if ids.is_empty() {
return Ok(Vec::new());
}
const CHUNK: usize = 900;
let id_strs: Vec<String> = ids.iter().map(|id| Uuid::from(*id).to_string()).collect();
let mut result: Vec<Edge> = Vec::with_capacity(ids.len());
for chunk in id_strs.chunks(CHUNK) {
let chunk_owned: Vec<String> = chunk.to_vec();
let edges = self
.with_reader("get_edges", move |conn| {
let placeholders: Vec<String> =
(1..=chunk_owned.len()).map(|i| format!("?{}", i)).collect();
let sql = format!(
"SELECT namespace, id, source_id, target_id, relation, weight, \
created_at, updated_at, deleted_at, metadata, target_backend \
FROM graph_edges WHERE id IN ({}) AND deleted_at IS NULL",
placeholders.join(",")
);
let mut stmt = conn.prepare(&sql)?;
let params: Vec<&dyn rusqlite::types::ToSql> = chunk_owned
.iter()
.map(|s| s as &dyn rusqlite::types::ToSql)
.collect();
let rows = stmt.query_map(params.as_slice(), read_edge)?;
let mut edges = Vec::new();
for row in rows {
edges.push(row?);
}
Ok(edges)
})
.await?;
result.extend(edges);
}
Ok(result)
}
async fn batch_neighbors(
&self,
sources: &[Uuid],
query: NeighborQuery,
) -> Result<Vec<(Uuid, NeighborHit)>, StorageError> {
use khive_storage::types::Direction;
if sources.is_empty() {
return Ok(Vec::new());
}
let mut seen_sources = HashSet::with_capacity(sources.len());
let unique_sources: Vec<Uuid> = sources
.iter()
.copied()
.filter(|source| seen_sources.insert(*source))
.collect();
const CHUNK_SIZE: usize = 880;
let namespace = self.namespace.clone();
let mut result: Vec<(Uuid, NeighborHit)> = Vec::new();
for chunk in unique_sources.chunks(CHUNK_SIZE) {
let chunk_owned: Vec<Uuid> = chunk.to_vec();
let query_clone = query.clone();
let ns = namespace.clone();
let pairs = self
.with_reader("batch_neighbors", move |conn| {
let src_strs: Vec<String> = chunk_owned.iter().map(|u| u.to_string()).collect();
let sources_json = serde_json::to_string(&src_strs).map_err(|error| {
rusqlite::Error::ToSqlConversionFailure(Box::new(error))
})?;
let build_inner_sql =
|direction_out: bool,
q: &NeighborQuery|
-> (String, Vec<String>, Option<f64>) {
let (filter_col, node_col) = if direction_out {
("source_id", "target_id")
} else {
("target_id", "source_id")
};
let mut rel_params: Vec<String> = Vec::new();
let mut conditions: Vec<String> = Vec::new();
let mut param_idx = 3;
if let Some(ref rels) = q.relations {
if !rels.is_empty() {
let ps: Vec<String> = rels
.iter()
.map(|r| {
rel_params.push(r.to_string());
let p = format!("?{param_idx}");
param_idx += 1;
p
})
.collect();
conditions
.push(format!("edges.relation IN ({})", ps.join(",")));
}
}
let min_weight_val = if let Some(min_w) = q.min_weight {
conditions.push(format!("edges.weight >= ?{param_idx}"));
Some(min_w)
} else {
None
};
let where_extra = if conditions.is_empty() {
String::new()
} else {
format!(" AND {}", conditions.join(" AND "))
};
let sql = format!(
"SELECT requested.origin_id, edges.{node_col} AS node_id, \
edges.id AS edge_id, edges.relation, edges.weight \
FROM requested CROSS JOIN graph_edges AS edges \
ON edges.{filter_col} = requested.origin_id \
WHERE edges.namespace = ?1 \
AND edges.deleted_at IS NULL{where_extra}",
);
(sql, rel_params, min_weight_val)
};
let mut all_params: Vec<Box<dyn rusqlite::types::ToSql>> = Vec::new();
all_params.push(Box::new(ns.to_string()));
all_params.push(Box::new(sources_json));
let (combined_inner, rel_params, min_weight_val) = match query_clone.direction {
Direction::Out => build_inner_sql(true, &query_clone),
Direction::In => build_inner_sql(false, &query_clone),
Direction::Both => {
let (out_sql, rel_params, min_weight_val) =
build_inner_sql(true, &query_clone);
let (in_sql, _, _) = build_inner_sql(false, &query_clone);
(
format!("{out_sql} UNION ALL {in_sql}"),
rel_params,
min_weight_val,
)
}
};
for relation in rel_params {
all_params.push(Box::new(relation));
}
if let Some(min_weight) = min_weight_val {
all_params.push(Box::new(min_weight));
}
let limit_param_idx = all_params.len() + 1;
let full_sql = if let Some(lim) = query_clone.limit {
all_params.push(Box::new(lim as i64));
format!(
"WITH requested(origin_id) AS (\
SELECT value FROM json_each(?2)\
) SELECT origin_id, node_id, edge_id, relation, weight \
FROM (SELECT *, ROW_NUMBER() OVER (PARTITION BY origin_id \
ORDER BY weight DESC, node_id ASC) AS rn \
FROM ({combined_inner})) WHERE rn <= ?{limit_param_idx}",
)
} else {
format!(
"WITH requested(origin_id) AS (\
SELECT value FROM json_each(?2)\
) SELECT origin_id, node_id, edge_id, relation, weight \
FROM ({combined_inner})",
)
};
let param_refs: Vec<&dyn rusqlite::types::ToSql> =
all_params.iter().map(|p| p.as_ref()).collect();
let mut stmt = conn.prepare(&full_sql)?;
let rows = stmt.query_map(param_refs.as_slice(), |row| {
let origin_str: String = row.get(0)?;
let nid_str: String = row.get(1)?;
let eid_str: String = row.get(2)?;
let relation_str: String = row.get(3)?;
let weight: f64 = row.get(4)?;
Ok((origin_str, nid_str, eid_str, relation_str, weight))
})?;
let mut pairs = Vec::new();
for row in rows {
let (origin_str, nid_str, eid_str, relation_str, weight) = row?;
let origin = parse_uuid(&origin_str)?;
let node_id = parse_uuid(&nid_str)?;
let edge_id = parse_uuid(&eid_str)?;
let relation = relation_str.parse::<EdgeRelation>().map_err(|e| {
rusqlite::Error::FromSqlConversionFailure(
3,
rusqlite::types::Type::Text,
Box::new(e),
)
})?;
pairs.push((
origin,
NeighborHit {
node_id,
edge_id,
relation,
weight,
name: None,
kind: None,
entity_type: None,
},
));
}
Ok(pairs)
})
.await?;
result.extend(pairs);
}
let requested: HashSet<Uuid> = unique_sources.iter().copied().collect();
let mut grouped: HashMap<Uuid, Vec<NeighborHit>> =
HashMap::with_capacity(unique_sources.len());
for (origin, hit) in result {
if !requested.contains(&origin) {
return Err(StorageError::Internal(format!(
"batch_neighbors returned unrequested origin {origin}"
)));
}
grouped.entry(origin).or_default().push(hit);
}
for hits in grouped.values_mut() {
hits.sort_by(|a, b| {
b.weight
.partial_cmp(&a.weight)
.unwrap_or(std::cmp::Ordering::Equal)
.then(a.node_id.cmp(&b.node_id))
.then(a.edge_id.cmp(&b.edge_id))
});
}
let mut ordered = Vec::new();
for &source in sources {
if let Some(hits) = grouped.get(&source) {
ordered.extend(hits.iter().cloned().map(|hit| (source, hit)));
}
}
Ok(ordered)
}
async fn delete_edge(&self, id: LinkId, mode: DeleteMode) -> Result<bool, StorageError> {
let id = Uuid::from(id);
let statement = match mode {
DeleteMode::Soft => {
edge_soft_delete_statement(id, chrono::Utc::now().timestamp_micros())
}
DeleteMode::Hard => edge_hard_delete_statement(id),
};
self.with_writer("delete_edge", move |conn| {
let mut stmt = conn.prepare(&statement.sql)?;
bind_params(&mut stmt, &statement.params)?;
Ok(stmt.raw_execute()? > 0)
})
.await
}
async fn query_edges(
&self,
filter: EdgeFilter,
sort: Vec<SortOrder<EdgeSortField>>,
page: PageRequest,
) -> Result<Page<Edge>, StorageError> {
let namespace = self.namespace.clone();
let limit_i64 = i64::from(page.limit);
let offset_i64 = i64::try_from(page.offset).map_err(|_| StorageError::InvalidInput {
capability: StorageCapability::Graph,
operation: "query_edges".into(),
message: format!(
"PageRequest: offset must be <= i64::MAX, got {}",
page.offset
),
})?;
self.with_reader("query_edges", move |conn| {
let (where_clause, filter_params) = build_edge_filter_sql(&namespace, &filter);
let count_sql = format!("SELECT COUNT(*) FROM graph_edges{}", where_clause);
let total: i64 = {
let mut stmt = conn.prepare(&count_sql)?;
let param_refs: Vec<&dyn rusqlite::types::ToSql> =
filter_params.iter().map(|p| p.as_ref()).collect();
stmt.query_row(param_refs.as_slice(), |row| row.get(0))?
};
let order_clause = if sort.is_empty() {
" ORDER BY created_at DESC".to_string()
} else {
let parts: Vec<String> = sort
.iter()
.map(|s| {
let dir = match s.direction {
SortDirection::Asc => "ASC",
SortDirection::Desc => "DESC",
};
format!("{} {}", edge_sort_col(&s.field), dir)
})
.collect();
format!(" ORDER BY {}", parts.join(", "))
};
let (_, data_filter_params) = build_edge_filter_sql(&namespace, &filter);
let mut all_params: Vec<Box<dyn rusqlite::types::ToSql>> = data_filter_params;
all_params.push(Box::new(limit_i64));
all_params.push(Box::new(offset_i64));
let limit_idx = all_params.len() - 1;
let offset_idx = all_params.len();
let data_sql = format!(
"SELECT namespace, id, source_id, target_id, relation, weight, \
created_at, updated_at, deleted_at, metadata, target_backend \
FROM graph_edges{}{} LIMIT ?{} OFFSET ?{}",
where_clause, order_clause, limit_idx, offset_idx,
);
let mut stmt = conn.prepare(&data_sql)?;
let param_refs: Vec<&dyn rusqlite::types::ToSql> =
all_params.iter().map(|p| p.as_ref()).collect();
let rows = stmt.query_map(param_refs.as_slice(), read_edge)?;
let mut items = Vec::new();
for row in rows {
items.push(row?);
}
Ok(Page {
items,
total: Some(total as u64),
})
})
.await
}
async fn count_edges(&self, filter: EdgeFilter) -> Result<u64, StorageError> {
let namespace = self.namespace.clone();
self.with_reader("count_edges", move |conn| {
let (where_clause, params) = build_edge_filter_sql(&namespace, &filter);
let sql = format!("SELECT COUNT(*) FROM graph_edges{}", where_clause);
let mut stmt = conn.prepare(&sql)?;
let param_refs: Vec<&dyn rusqlite::types::ToSql> =
params.iter().map(|p| p.as_ref()).collect();
let count: i64 = stmt.query_row(param_refs.as_slice(), |row| row.get(0))?;
Ok(count as u64)
})
.await
}
async fn count_edges_by_relation(&self) -> Result<Vec<(EdgeRelation, u64)>, StorageError> {
let namespace = self.namespace.clone();
self.with_reader("count_edges_by_relation", move |conn| {
let sql = "SELECT relation, COUNT(*) FROM graph_edges \
WHERE namespace = ?1 AND deleted_at IS NULL \
GROUP BY relation";
let mut stmt = conn.prepare(sql)?;
let rows = stmt.query_map([&namespace], |row| {
let relation_str: String = row.get(0)?;
let count: i64 = row.get(1)?;
Ok((relation_str, count))
})?;
let mut out = Vec::new();
for row in rows {
let (relation_str, count) = row?;
let relation = relation_str.parse::<EdgeRelation>().map_err(|e| {
rusqlite::Error::FromSqlConversionFailure(
0,
rusqlite::types::Type::Text,
Box::new(e),
)
})?;
out.push((relation, count as u64));
}
Ok(out)
})
.await
}
async fn query_edges_after(
&self,
filter: EdgeFilter,
after: Option<Uuid>,
limit: u32,
) -> Result<EdgeSeekPage, StorageError> {
let namespace = self.namespace.clone();
let limit_usize = limit as usize;
let probe_limit_i64 = i64::from(limit) + 1;
self.with_reader("query_edges_after", move |conn| {
let (mut where_clause, mut params) = build_edge_filter_sql(&namespace, &filter);
if let Some(cursor) = after {
params.push(Box::new(cursor.to_string()));
where_clause.push_str(&format!(" AND id > ?{}", params.len()));
}
params.push(Box::new(probe_limit_i64));
let limit_idx = params.len();
let data_sql = format!(
"SELECT namespace, id, source_id, target_id, relation, weight, \
created_at, updated_at, deleted_at, metadata, target_backend \
FROM graph_edges{} ORDER BY id ASC LIMIT ?{}",
where_clause, limit_idx,
);
let mut stmt = conn.prepare(&data_sql)?;
let param_refs: Vec<&dyn rusqlite::types::ToSql> =
params.iter().map(|p| p.as_ref()).collect();
let rows = stmt.query_map(param_refs.as_slice(), read_edge)?;
let mut items = Vec::new();
for row in rows {
items.push(row?);
}
let has_more = items.len() > limit_usize;
if has_more {
items.truncate(limit_usize);
}
let next_after = if has_more {
items.last().map(|e| Uuid::from(e.id))
} else {
None
};
Ok(EdgeSeekPage { items, next_after })
})
.await
}
async fn neighbors(
&self,
node_id: Uuid,
query: NeighborQuery,
) -> Result<Vec<NeighborHit>, StorageError> {
count_neighbor_select();
let namespace = self.namespace.clone();
let node_str = node_id.to_string();
self.with_reader("neighbors", move |conn| {
let base_out = "SELECT target_id AS node_id, id AS edge_id, relation, weight \
FROM graph_edges \
WHERE namespace = ?1 AND source_id = ?2 AND deleted_at IS NULL";
let base_in = "SELECT source_id AS node_id, id AS edge_id, relation, weight \
FROM graph_edges \
WHERE namespace = ?1 AND target_id = ?2 AND deleted_at IS NULL";
let sql = match query.direction {
Direction::Out => base_out.to_string(),
Direction::In => base_in.to_string(),
Direction::Both => format!("{} UNION ALL {}", base_out, base_in),
};
let (where_extra, limit_clause, extra_params) = neighbor_extra_clause(&query, 3);
let full_sql = format!(
"SELECT node_id, edge_id, relation, weight FROM ({}){} \
ORDER BY weight DESC, node_id ASC{}",
sql, where_extra, limit_clause
);
let mut stmt = conn.prepare(&full_sql)?;
let mut all_params: Vec<Box<dyn rusqlite::types::ToSql>> = Vec::new();
all_params.push(Box::new(namespace.clone()));
all_params.push(Box::new(node_str.clone()));
all_params.extend(extra_params);
let param_refs: Vec<&dyn rusqlite::types::ToSql> =
all_params.iter().map(|p| p.as_ref()).collect();
let rows = stmt.query_map(param_refs.as_slice(), |row| {
let nid_str: String = row.get(0)?;
let eid_str: String = row.get(1)?;
let relation_str: String = row.get(2)?;
let weight: f64 = row.get(3)?;
Ok((nid_str, eid_str, relation_str, weight))
})?;
let mut hits = Vec::new();
for row in rows {
let (nid_str, eid_str, relation_str, weight) = row?;
let relation = relation_str.parse::<EdgeRelation>().map_err(|e| {
rusqlite::Error::FromSqlConversionFailure(
2,
rusqlite::types::Type::Text,
Box::new(e),
)
})?;
hits.push(NeighborHit {
node_id: parse_uuid(&nid_str)?,
edge_id: parse_uuid(&eid_str)?,
relation,
weight,
name: None,
kind: None,
entity_type: None,
});
}
Ok(hits)
})
.await
}
async fn neighbors_both_directions(
&self,
node_id: Uuid,
query: NeighborQuery,
) -> Result<Vec<DirectedNeighborHit>, StorageError> {
count_neighbor_select();
let namespace = self.namespace.clone();
let node_str = node_id.to_string();
self.with_reader("neighbors_both_directions", move |conn| {
let base_out = "SELECT target_id AS node_id, id AS edge_id, relation, weight, \
'out' AS dir \
FROM graph_edges \
WHERE namespace = ?1 AND source_id = ?2 AND deleted_at IS NULL";
let base_in = "SELECT source_id AS node_id, id AS edge_id, relation, weight, \
'in' AS dir \
FROM graph_edges \
WHERE namespace = ?1 AND target_id = ?2 AND deleted_at IS NULL";
let sql = format!("{} UNION ALL {}", base_out, base_in);
let (where_extra, limit_clause, extra_params) = neighbor_extra_clause(&query, 3);
let full_sql = format!(
"SELECT node_id, edge_id, relation, weight, dir FROM ({}){} \
ORDER BY weight DESC, node_id ASC, \
CASE dir WHEN 'out' THEN 0 ELSE 1 END ASC, edge_id ASC{}",
sql, where_extra, limit_clause
);
let mut stmt = conn.prepare(&full_sql)?;
let mut all_params: Vec<Box<dyn rusqlite::types::ToSql>> = Vec::new();
all_params.push(Box::new(namespace.clone()));
all_params.push(Box::new(node_str.clone()));
all_params.extend(extra_params);
let param_refs: Vec<&dyn rusqlite::types::ToSql> =
all_params.iter().map(|p| p.as_ref()).collect();
let rows = stmt.query_map(param_refs.as_slice(), |row| {
let nid_str: String = row.get(0)?;
let eid_str: String = row.get(1)?;
let relation_str: String = row.get(2)?;
let weight: f64 = row.get(3)?;
let dir_str: String = row.get(4)?;
Ok((nid_str, eid_str, relation_str, weight, dir_str))
})?;
let mut hits = Vec::new();
for row in rows {
let (nid_str, eid_str, relation_str, weight, dir_str) = row?;
let relation = relation_str.parse::<EdgeRelation>().map_err(|e| {
rusqlite::Error::FromSqlConversionFailure(
2,
rusqlite::types::Type::Text,
Box::new(e),
)
})?;
let direction = if dir_str == "out" {
Direction::Out
} else {
Direction::In
};
hits.push(DirectedNeighborHit {
hit: NeighborHit {
node_id: parse_uuid(&nid_str)?,
edge_id: parse_uuid(&eid_str)?,
relation,
weight,
name: None,
kind: None,
entity_type: None,
},
direction,
});
}
Ok(hits)
})
.await
}
async fn traverse(&self, request: TraversalRequest) -> Result<Vec<GraphPath>, StorageError> {
use std::collections::{HashMap, HashSet};
use khive_storage::types::Direction;
if request.roots.is_empty() {
return Ok(Vec::new());
}
let roots = request.roots.clone();
let opts = request.options.clone();
let include_roots = request.include_roots;
let namespace = self.namespace.clone();
let max_depth_i64 =
i64::try_from(opts.max_depth).map_err(|_| StorageError::InvalidInput {
capability: StorageCapability::Graph,
operation: "traverse".into(),
message: format!(
"TraversalOptions: max_depth must be <= i64::MAX, got {}",
opts.max_depth
),
})?;
self.with_reader("traverse", move |conn| {
const CHUNK_ROOTS: usize = 400;
let (join_condition, next_node) = match opts.direction {
Direction::Out => ("e.source_id = t.node_id", "e.target_id"),
Direction::In => ("e.target_id = t.node_id", "e.source_id"),
Direction::Both => (
"(e.source_id = t.node_id OR e.target_id = t.node_id)",
"CASE WHEN e.source_id = t.node_id THEN e.target_id ELSE e.source_id END",
),
};
let _tx_handle =
khive_storage::tx_registry::register(Some("graph_traverse_read".to_string()));
let tx = conn.unchecked_transaction()?;
let mut root_data: HashMap<Uuid, (Vec<(PathNode, f64)>, HashSet<Uuid>)> =
HashMap::with_capacity(roots.len());
for root_id in &roots {
let (nodes, seen) = root_data.entry(*root_id).or_default();
if include_roots {
seen.insert(*root_id);
nodes.push((
PathNode {
node_id: *root_id,
via_edge: None,
depth: 0,
name: None,
kind: None,
properties: None,
},
0.0,
));
}
}
for chunk in roots.chunks(CHUNK_ROOTS) {
let n_chunk = chunk.len();
let ns_param = n_chunk + 1;
let depth_param = n_chunk + 2;
let mut extra_param_idx = n_chunk + 3;
let mut relation_cond = String::new();
let mut extra_params: Vec<Box<dyn rusqlite::types::ToSql>> = Vec::new();
if let Some(ref rels) = opts.relations {
if !rels.is_empty() {
let placeholders: Vec<String> = rels
.iter()
.map(|r| {
extra_params.push(Box::new(r.to_string()));
let p = format!("?{extra_param_idx}");
extra_param_idx += 1;
p
})
.collect();
relation_cond = format!(" AND e.relation IN ({})", placeholders.join(","));
}
}
let mut weight_cond = String::new();
if let Some(min_w) = opts.min_weight {
extra_params.push(Box::new(min_w));
weight_cond = format!(" AND e.weight >= ?{extra_param_idx}");
}
let seed_rows: Vec<String> = (1..=n_chunk)
.map(|i| format!("(?{i}, ?{i}, NULL, 0, ?{i}, 0.0)"))
.collect();
let seeds = seed_rows.join(", ");
let cte_sql = format!(
"WITH RECURSIVE traversal(\
root_id, node_id, edge_id, depth, path, total_weight\
) AS (\
VALUES {seeds} \
UNION ALL \
SELECT t.root_id, {next_node}, e.id, t.depth + 1, \
t.path || ',' || {next_node}, \
t.total_weight + e.weight \
FROM traversal t CROSS JOIN graph_edges e \
ON {join_condition} \
WHERE e.namespace = ?{ns} \
AND e.deleted_at IS NULL \
AND t.depth < ?{depth} \
AND (',' || t.path || ',') NOT LIKE '%,' || {next_node} || ',%'\
{rel_cond}{wt_cond} \
) \
SELECT root_id, node_id, edge_id, depth, total_weight \
FROM traversal WHERE depth > 0 \
ORDER BY root_id, depth",
seeds = seeds,
next_node = next_node,
join_condition = join_condition,
ns = ns_param,
depth = depth_param,
rel_cond = relation_cond,
wt_cond = weight_cond,
);
let mut all_params: Vec<Box<dyn rusqlite::types::ToSql>> = Vec::new();
for root_id in chunk {
all_params.push(Box::new(root_id.to_string()));
}
all_params.push(Box::new(namespace.clone()));
all_params.push(Box::new(max_depth_i64));
all_params.extend(extra_params);
let param_refs: Vec<&dyn rusqlite::types::ToSql> =
all_params.iter().map(|p| p.as_ref()).collect();
let mut stmt = conn.prepare(&cte_sql)?;
let rows_iter = stmt.query_map(param_refs.as_slice(), |row| {
let root_str: String = row.get(0)?;
let node_str: String = row.get(1)?;
let edge_str: Option<String> = row.get(2)?;
let depth: i64 = row.get(3)?;
let total_weight: f64 = row.get(4)?;
Ok((root_str, node_str, edge_str, depth, total_weight))
})?;
for row in rows_iter {
let (root_str, node_str, edge_str, depth, total_weight) = row?;
let root_id = parse_uuid(&root_str)?;
let node_id = parse_uuid(&node_str)?;
let (nodes, seen) = root_data.entry(root_id).or_default();
if !seen.insert(node_id) {
continue;
}
let via_edge = edge_str.map(|s| parse_uuid(&s)).transpose()?;
nodes.push((
PathNode {
node_id,
via_edge,
depth: depth as usize,
name: None,
kind: None,
properties: None,
},
total_weight,
));
}
}
tx.commit()?;
let mut all_paths: Vec<GraphPath> = Vec::with_capacity(roots.len());
for root_id in &roots {
if let Some((mut nw, _)) = root_data.remove(root_id) {
if nw.is_empty() {
continue;
}
if let Some(lim) = opts.limit {
let root_count = usize::from(include_roots);
nw.truncate(root_count + lim as usize);
}
if nw.is_empty() {
continue;
}
let max_weight = nw.iter().map(|(_, w)| *w).fold(0.0_f64, f64::max);
let nodes: Vec<PathNode> = nw.into_iter().map(|(n, _)| n).collect();
all_paths.push(GraphPath {
root_id: *root_id,
nodes,
total_weight: max_weight,
});
}
}
Ok(all_paths)
})
.await
}
async fn purge_incident_edges(&self, node_id: Uuid) -> Result<u64, StorageError> {
let statement = purge_incident_edges_statement(node_id);
self.with_writer("purge_incident_edges", move |conn| {
let mut stmt = conn.prepare(&statement.sql)?;
bind_params(&mut stmt, &statement.params)?;
Ok(stmt.raw_execute()? as u64)
})
.await
}
}
const GRAPH_DDL: &str = include_str!("../../sql/graph-ddl.sql");
pub(crate) fn ensure_graph_schema(conn: &rusqlite::Connection) -> Result<(), rusqlite::Error> {
conn.execute_batch(GRAPH_DDL)
}
#[cfg(test)]
#[path = "graph_tests.rs"]
mod tests;