use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use nodedb_graph::CsrIndex;
use nodedb_types::SurrogateBitmap;
use nodedb_types::result::QueryResult;
use nodedb_types::value::Value;
use crate::engine::crdt::CrdtEngine;
use crate::error::LiteError;
use super::ast::MatchQuery;
use super::executor::{HydrationCtx, execute_query};
pub async fn graph_match(
csr_map: &Arc<Mutex<HashMap<String, CsrIndex>>>,
query_bytes: &[u8],
frontier_bitmap: Option<&SurrogateBitmap>,
crdt: Option<&Arc<Mutex<CrdtEngine>>>,
) -> Result<QueryResult, LiteError> {
let query: MatchQuery = zerompk::from_msgpack(query_bytes).map_err(|e| LiteError::Storage {
detail: format!("deserialize MatchQuery: {e}"),
})?;
let map = csr_map.lock().map_err(|_| LiteError::LockPoisoned)?;
let collection_name = query
.collection
.clone()
.or_else(|| map.keys().next().cloned());
let Some(ref col) = collection_name else {
return Ok(QueryResult::empty());
};
let Some(csr) = map.get(col) else {
return Ok(QueryResult::empty());
};
let hydration = crdt.map(|c| HydrationCtx {
crdt: c.as_ref(),
collection: col.as_str(),
});
let rows = execute_query(&query, csr, frontier_bitmap, hydration.as_ref())?;
let columns: Vec<String> = if query.return_columns.is_empty() {
query.bound_node_names()
} else {
query
.return_columns
.iter()
.map(|rc| rc.alias.clone().unwrap_or_else(|| rc.expr.clone()))
.collect()
};
let result_rows: Vec<Vec<Value>> = rows
.into_iter()
.map(|row| {
columns
.iter()
.map(|col_name| {
row.get(col_name)
.map(|v| Value::String(v.clone()))
.unwrap_or(Value::Null)
})
.collect()
})
.collect();
Ok(QueryResult {
columns,
rows: result_rows,
rows_affected: 0,
command: None,
})
}
#[cfg(test)]
mod tests {
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use nodedb_graph::CsrIndex;
use nodedb_types::value::Value;
use super::super::ast::*;
use super::graph_match;
fn make_csr() -> CsrIndex {
let mut csr = CsrIndex::new(crate::query::graph_ops::test_memory());
csr.add_edge("alice", "KNOWS", "bob").unwrap();
csr.add_edge("bob", "KNOWS", "carol").unwrap();
csr.add_edge("alice", "LIKES", "carol").unwrap();
csr.add_node_label("alice", "Person").unwrap();
csr.add_node_label("bob", "Person").unwrap();
csr.add_node_label("carol", "Person").unwrap();
csr
}
fn make_query_bytes(query: &MatchQuery) -> Vec<u8> {
zerompk::to_msgpack_vec(query).expect("serialize MatchQuery")
}
fn simple_query(
src_label: Option<&str>,
edge_type: Option<&str>,
dst_label: Option<&str>,
where_eq: Option<(&str, &str)>,
) -> MatchQuery {
let mut predicates = Vec::new();
if let Some((var, val)) = where_eq {
predicates.push(WherePredicate::Equals {
binding: var.to_string(),
field: var.to_string(),
value: val.to_string(),
});
}
MatchQuery {
clauses: vec![MatchClause {
patterns: vec![PatternChain {
triples: vec![PatternTriple {
src: NodeBinding {
name: Some("a".to_string()),
label: src_label.map(str::to_string),
},
edge: EdgeBinding {
name: None,
edge_type: edge_type.map(str::to_string),
direction: EdgeDirection::Right,
min_hops: 1,
max_hops: 1,
},
dst: NodeBinding {
name: Some("b".to_string()),
label: dst_label.map(str::to_string),
},
}],
}],
optional: false,
}],
where_predicates: predicates,
return_columns: vec![
ReturnColumn {
expr: "a".to_string(),
alias: None,
},
ReturnColumn {
expr: "b".to_string(),
alias: None,
},
],
distinct: false,
limit: None,
order_by: Vec::new(),
collection: Some("col".to_string()),
}
}
fn make_csr_map(csr: CsrIndex) -> Arc<Mutex<HashMap<String, CsrIndex>>> {
let mut map = HashMap::new();
map.insert("col".to_string(), csr);
Arc::new(Mutex::new(map))
}
#[tokio::test]
async fn match_node_label_and_edge_label() {
let csr_map = make_csr_map(make_csr());
let query = simple_query(Some("Person"), Some("KNOWS"), Some("Person"), None);
let bytes = make_query_bytes(&query);
let result = graph_match(&csr_map, &bytes, None, None)
.await
.expect("graph_match must not error");
assert_eq!(
result.rows.len(),
2,
"expected 2 KNOWS rows between Persons"
);
assert_eq!(result.columns, vec!["a", "b"]);
}
#[tokio::test]
async fn match_where_equals_filters_anchor() {
let csr_map = make_csr_map(make_csr());
let query = simple_query(Some("Person"), Some("KNOWS"), None, Some(("a", "alice")));
let bytes = make_query_bytes(&query);
let result = graph_match(&csr_map, &bytes, None, None)
.await
.expect("graph_match must not error");
assert_eq!(result.rows.len(), 1);
assert_eq!(result.rows[0][0], Value::String("alice".to_string()));
assert_eq!(result.rows[0][1], Value::String("bob".to_string()));
}
#[tokio::test]
async fn match_no_edge_label_returns_all_edges() {
let csr_map = make_csr_map(make_csr());
let query = simple_query(None, None, None, Some(("a", "alice")));
let bytes = make_query_bytes(&query);
let result = graph_match(&csr_map, &bytes, None, None)
.await
.expect("graph_match must not error");
assert_eq!(result.rows.len(), 2);
}
#[tokio::test]
async fn match_frontier_bitmap_restricts_anchors() {
use nodedb_types::{Surrogate, SurrogateBitmap};
let mut csr = make_csr();
csr.set_node_surrogate("alice", Surrogate::new(1));
csr.set_node_surrogate("bob", Surrogate::new(2));
csr.set_node_surrogate("carol", Surrogate::new(3));
let csr_map = make_csr_map(csr);
let bm = SurrogateBitmap::from_iter([Surrogate::new(1)]);
let query = simple_query(None, Some("KNOWS"), None, None);
let bytes = make_query_bytes(&query);
let result = graph_match(&csr_map, &bytes, Some(&bm), None)
.await
.expect("graph_match must not error");
assert_eq!(result.rows.len(), 1);
assert_eq!(result.rows[0][0], Value::String("alice".to_string()));
}
#[tokio::test]
async fn match_empty_csr_returns_empty() {
let csr_map = make_csr_map(CsrIndex::new(crate::query::graph_ops::test_memory()));
let query = simple_query(None, Some("KNOWS"), None, None);
let bytes = make_query_bytes(&query);
let result = graph_match(&csr_map, &bytes, None, None)
.await
.expect("empty CSR must not error");
assert!(result.rows.is_empty());
}
#[tokio::test]
async fn match_not_exists_anti_join() {
let mut csr = CsrIndex::new(crate::query::graph_ops::test_memory());
csr.add_edge("alice", "KNOWS", "bob").unwrap();
csr.add_edge("bob", "KNOWS", "carol").unwrap();
csr.add_edge("alice", "BLOCKED", "carol").unwrap();
let csr_map = make_csr_map(csr);
let anti = MatchClause {
patterns: vec![PatternChain {
triples: vec![PatternTriple {
src: NodeBinding {
name: Some("a".to_string()),
label: None,
},
edge: EdgeBinding {
name: None,
edge_type: Some("BLOCKED".to_string()),
direction: EdgeDirection::Right,
min_hops: 1,
max_hops: 1,
},
dst: NodeBinding {
name: Some("b".to_string()),
label: None,
},
}],
}],
optional: false,
};
let query = MatchQuery {
clauses: vec![MatchClause {
patterns: vec![PatternChain {
triples: vec![PatternTriple {
src: NodeBinding {
name: Some("a".to_string()),
label: None,
},
edge: EdgeBinding {
name: None,
edge_type: Some("KNOWS".to_string()),
direction: EdgeDirection::Right,
min_hops: 1,
max_hops: 1,
},
dst: NodeBinding {
name: Some("b".to_string()),
label: None,
},
}],
}],
optional: false,
}],
where_predicates: vec![WherePredicate::NotExists { sub_pattern: anti }],
return_columns: vec![
ReturnColumn {
expr: "a".to_string(),
alias: None,
},
ReturnColumn {
expr: "b".to_string(),
alias: None,
},
],
distinct: false,
limit: None,
order_by: Vec::new(),
collection: Some("col".to_string()),
};
let bytes = make_query_bytes(&query);
let result = graph_match(&csr_map, &bytes, None, None)
.await
.expect("graph_match must not error");
assert_eq!(
result.rows.len(),
2,
"expected 2 rows without blocked pairs"
);
}
#[tokio::test]
async fn match_where_subfield_crdt_hydration() {
use crate::engine::crdt::CrdtEngine;
use std::sync::Mutex;
let mut csr = CsrIndex::new(crate::query::graph_ops::test_memory());
csr.add_edge("alice", "KNOWS", "charlie").unwrap();
csr.add_edge("bob", "KNOWS", "charlie").unwrap();
csr.add_node_label("alice", "Person").unwrap();
csr.add_node_label("bob", "Person").unwrap();
csr.add_node_label("charlie", "Person").unwrap();
let mut crdt_engine = CrdtEngine::new(1).expect("CrdtEngine::new");
crdt_engine
.upsert(
"people",
"alice",
&[("name", loro::LoroValue::String("Alice".into()))],
)
.expect("upsert alice");
crdt_engine
.upsert(
"people",
"bob",
&[("name", loro::LoroValue::String("Bob".into()))],
)
.expect("upsert bob");
let crdt = Arc::new(Mutex::new(crdt_engine));
let mut csr_map_inner = HashMap::new();
csr_map_inner.insert("people".to_string(), csr);
let csr_map = Arc::new(Mutex::new(csr_map_inner));
let query = MatchQuery {
collection: Some("people".to_string()),
clauses: vec![MatchClause {
optional: false,
patterns: vec![PatternChain {
triples: vec![PatternTriple {
src: NodeBinding {
name: Some("a".to_string()),
label: Some("Person".to_string()),
},
edge: EdgeBinding {
name: None,
edge_type: Some("KNOWS".to_string()),
direction: EdgeDirection::Right,
min_hops: 1,
max_hops: 1,
},
dst: NodeBinding {
name: Some("b".to_string()),
label: None,
},
}],
}],
}],
where_predicates: vec![WherePredicate::Equals {
binding: "a".to_string(),
field: "name".to_string(),
value: "Alice".to_string(),
}],
return_columns: vec![ReturnColumn {
expr: "a".to_string(),
alias: None,
}],
distinct: true,
limit: None,
order_by: Vec::new(),
};
let bytes = make_query_bytes(&query);
let result = graph_match(&csr_map, &bytes, None, Some(&crdt))
.await
.expect("graph_match must not error");
assert_eq!(
result.rows.len(),
1,
"only Alice should pass the WHERE filter"
);
assert_eq!(result.rows[0][0], Value::String("alice".to_string()));
}
}