fix: adapt to updated core crate APIs across query, graph, and storage

Bump nodedb-* workspace dependencies to 0.4 and follow through on their
API changes: switch `PlanVisitor` methods to struct-style `*VisitArgs`,
adopt `BfsParams`/`ShortestPathParams` for graph traversal, support
computed GROUP BY keys via `GroupKeySpec` and MERGE source-expression
values via `UpdateValue::Expr`, handle new `MetaOp` transaction-overlay
and Calvin variants as unsupported on the single-node engine, propagate
the new fallible HNSW checkpoint encoding, and update wire-version and
formatting fallout in the affected tests.

Author: farhan-syah

Cargo.toml

@@ -14,20 +14,20 @@ homepage = "https://nodedb.dev"
 [workspace.dependencies]
 # Internal
 nodedb-lite = { path = "nodedb-lite", version = "0.1.0", default-features = false }
-nodedb-types = { version = "0.3" }
-nodedb-client = { version = "0.3" }
-nodedb-codec = { version = "0.3" }
-nodedb-crdt = { version = "0.3" }
-nodedb-query = { version = "0.3" }
-nodedb-spatial = { version = "0.3" }
-nodedb-graph = { version = "0.3" }
-nodedb-vector = { version = "0.3" }
-nodedb-fts = { version = "0.3" }
-nodedb-strict = { version = "0.3" }
-nodedb-columnar = { version = "0.3" }
-nodedb-sql = { version = "0.3" }
-nodedb-array = { version = "0.3" }
-nodedb-physical = { version = "0.3" }
+nodedb-types = { version = "0.4" }
+nodedb-client = { version = "0.4" }
+nodedb-codec = { version = "0.4" }
+nodedb-crdt = { version = "0.4" }
+nodedb-query = { version = "0.4" }
+nodedb-spatial = { version = "0.4" }
+nodedb-graph = { version = "0.4" }
+nodedb-vector = { version = "0.4" }
+nodedb-fts = { version = "0.4" }
+nodedb-strict = { version = "0.4" }
+nodedb-columnar = { version = "0.4" }
+nodedb-sql = { version = "0.4" }
+nodedb-array = { version = "0.4" }
+nodedb-physical = { version = "0.4" }
 
 # Async
 tokio = { version = "1" }

nodedb-lite/src/nodedb/batch.rs

@@ -176,12 +176,17 @@ impl<S: StorageEngine> NodeDbLite<S> {
                 match indices.get(&name) {
                     Some(idx) => {
                         if seg_ext.is_some() {
-                            let graph_bytes = idx.graph_checkpoint_to_bytes();
+                            let graph_bytes = idx.graph_checkpoint_to_bytes().map_err(|e| {
+                                NodeDbError::serialization("hnsw-graph-checkpoint", e)
+                            })?;
                             let (vectors, surrogates) = idx.extract_vectors_and_surrogates();
                             let dim = idx.dim();
                             (graph_bytes, Some((dim, vectors, surrogates)))
                         } else {
-                            (idx.checkpoint_to_bytes(), None)
+                            let blob = idx
+                                .checkpoint_to_bytes()
+                                .map_err(|e| NodeDbError::serialization("hnsw-checkpoint", e))?;
+                            (blob, None)
                         }
                     }
                     None => continue,
@@ -191,7 +196,9 @@ impl<S: StorageEngine> NodeDbLite<S> {
             let blob = {
                 let indices = self.vector_state.hnsw_indices.lock_or_recover();
                 match indices.get(&name) {
-                    Some(idx) => idx.checkpoint_to_bytes(),
+                    Some(idx) => idx
+                        .checkpoint_to_bytes()
+                        .map_err(|e| NodeDbError::serialization("hnsw-checkpoint", e))?,
                     None => continue,
                 }
             };

nodedb-lite/src/nodedb/core/flush.rs

@@ -197,7 +197,9 @@ impl<S: StorageEngine> NodeDbLite<S> {
                 {
                     if seg_ext.is_some() {
                         // Graph-only blob (vector bytes are empty placeholders).
-                        let graph_bytes = index.graph_checkpoint_to_bytes();
+                        let graph_bytes = index
+                            .graph_checkpoint_to_bytes()
+                            .map_err(|e| NodeDbError::serialization("hnsw-graph-checkpoint", e))?;
                         ops.push(WriteOp::Put {
                             ns: Namespace::Vector,
                             key: key.into_bytes(),
@@ -209,7 +211,9 @@ impl<S: StorageEngine> NodeDbLite<S> {
                         segment_data.push((name.clone(), dim, vectors, surrogates));
                     } else {
                         // Non-pagedb native backend: full checkpoint blob path.
-                        let checkpoint = index.checkpoint_to_bytes();
+                        let checkpoint = index
+                            .checkpoint_to_bytes()
+                            .map_err(|e| NodeDbError::serialization("hnsw-checkpoint", e))?;
                         ops.push(WriteOp::Put {
                             ns: Namespace::Vector,
                             key: key.into_bytes(),

nodedb-lite/src/nodedb/trait_impl/graph.rs

@@ -401,11 +401,14 @@ impl<S: StorageEngine> NodeDbLite<S> {
         let csr_map = self.csr.lock_or_recover();
         let path = match csr_map.get(collection) {
             Some(csr) => csr.shortest_path(
-                from.as_str(),
-                to.as_str(),
-                label_filter,
-                max_depth as usize,
-                DEFAULT_MAX_VISITED,
+                nodedb_graph::ShortestPathParams {
+                    src: from.as_str(),
+                    dst: to.as_str(),
+                    label_filter,
+                    max_depth: max_depth as usize,
+                    max_visited: DEFAULT_MAX_VISITED,
+                    frontier_bitmap: None,
+                },
                 None,
             ),
             None => None,

nodedb-lite/src/query/document_ops/sets.rs

@@ -366,7 +366,7 @@ pub async fn merge<S: StorageEngine>(
                 .get(source_join_col)
                 .map(value_to_string)
                 .unwrap_or_else(|| key.clone());
-            let map: HashMap<String, Value> = build_insert_map(columns, values)?;
+            let map: HashMap<String, Value> = build_insert_map(columns, values, source_val)?;
             let bytes = zerompk::to_msgpack_vec(&Value::Object(map)).map_err(|e| {
                 LiteError::Serialization {
                     detail: format!("merge insert serialize: {e}"),
@@ -525,18 +525,31 @@ fn qualify_updates_with_source(
     Ok(updates.to_vec())
 }
 
-/// Decode a parallel (columns, values) pair into a field map.
-fn build_insert_map(
+/// Resolve a parallel (columns, values) pair into a field map, evaluating each
+/// value against the source row.
+///
+/// `UpdateValue::Literal` arms decode the pre-encoded msgpack directly.
+/// `UpdateValue::Expr` arms (e.g. `s.new_embedding`, `s.qty * 2`) are evaluated
+/// against the source row — Lite's `convert_sql_expr` strips table qualifiers,
+/// so column references resolve against the source row's bare field names. The
+/// result is stored under the *target* column name.
+pub(in crate::query) fn build_insert_map(
     columns: &[String],
-    values: &[Vec<u8>],
+    values: &[UpdateValue],
+    source_val: &HashMap<String, Value>,
 ) -> Result<HashMap<String, Value>, LiteError> {
+    let source_ndb = Value::Object(source_val.clone());
     let mut map = HashMap::with_capacity(columns.len());
-    for (col, val_bytes) in columns.iter().zip(values.iter()) {
-        let val: Value =
-            zerompk::from_msgpack(val_bytes).map_err(|e| LiteError::Serialization {
-                detail: format!("merge insert decode column '{col}': {e}"),
-            })?;
-        map.insert(col.clone(), val);
+    for (col, val) in columns.iter().zip(values.iter()) {
+        let resolved: Value = match val {
+            UpdateValue::Literal(bytes) => {
+                zerompk::from_msgpack(bytes).map_err(|e| LiteError::Serialization {
+                    detail: format!("merge insert decode column '{col}': {e}"),
+                })?
+            }
+            UpdateValue::Expr(expr) => expr.eval(&source_ndb),
+        };
+        map.insert(col.clone(), resolved);
     }
     Ok(map)
 }
@@ -560,7 +573,7 @@ async fn apply_merge_action<S: StorageEngine>(
             point_delete(engine, collection, doc_id).await?;
         }
         MergeActionOp::Insert { columns, values } => {
-            let map = build_insert_map(columns, values)?;
+            let map = build_insert_map(columns, values, source_val)?;
             let bytes = zerompk::to_msgpack_vec(&Value::Object(map)).map_err(|e| {
                 LiteError::Serialization {
                     detail: format!("merge action insert serialize: {e}"),

nodedb-lite/src/query/graph_ops/fusion.rs

@@ -141,11 +141,14 @@ pub async fn rag_fusion<S: StorageEngine>(
                 .saturating_mul(vector_top_k)
                 .max(DEFAULT_MAX_VISITED);
             let expanded = csr.traverse_bfs(
-                &starts,
-                edge_label,
-                direction,
-                expansion_depth,
-                max_vis,
+                nodedb_graph::BfsParams {
+                    start_nodes: &starts,
+                    label_filter: edge_label,
+                    direction,
+                    max_depth: expansion_depth,
+                    max_visited: max_vis,
+                    frontier_bitmap: None,
+                },
                 None,
             );
 
@@ -302,6 +305,7 @@ mod tests {
             vector_state,
             array_state,
             fts_state,
+            Arc::new(crate::engine::sparse_vector::SparseVectorState::new()),
             spatial,
             Arc::new(Mutex::new(std::collections::HashMap::new())),
         )

nodedb-lite/src/query/graph_ops/match_engine/executor.rs

@@ -139,12 +139,15 @@ fn execute_triple(
     if triple.edge.is_variable_length() {
         for src_name in &src_nodes {
             let expanded = csr.traverse_bfs(
-                &[src_name.as_str()],
-                label_filter,
-                direction,
-                triple.edge.max_hops,
-                50_000,
-                frontier_bitmap,
+                nodedb_graph::BfsParams {
+                    start_nodes: &[src_name.as_str()],
+                    label_filter,
+                    direction,
+                    max_depth: triple.edge.max_hops,
+                    max_visited: 50_000,
+                    frontier_bitmap,
+                },
+                None,
             );
             for dst_name in expanded {
                 if dst_name == *src_name && triple.edge.min_hops > 0 {

nodedb-lite/src/query/graph_ops/traversal.rs

@@ -50,7 +50,17 @@ pub fn hop(
 
     let starts: Vec<&str> = start_nodes.iter().map(String::as_str).collect();
     let mv = max_visited(options);
-    let nodes = csr.traverse_bfs(&starts, edge_label, direction, depth, mv, frontier_bitmap);
+    let nodes = csr.traverse_bfs(
+        nodedb_graph::BfsParams {
+            start_nodes: &starts,
+            label_filter: edge_label,
+            direction,
+            max_depth: depth,
+            max_visited: mv,
+            frontier_bitmap,
+        },
+        None,
+    );
 
     let rows = nodes.iter().map(|n| node_row(n)).collect();
     Ok(QueryResult {
@@ -155,7 +165,17 @@ pub fn path(
     };
 
     let mv = max_visited(options);
-    let maybe_path = csr.shortest_path(src, dst, edge_label, max_depth, mv, frontier_bitmap);
+    let maybe_path = csr.shortest_path(
+        nodedb_graph::ShortestPathParams {
+            src,
+            dst,
+            label_filter: edge_label,
+            max_depth,
+            max_visited: mv,
+            frontier_bitmap,
+        },
+        None,
+    );
 
     let columns = vec!["path".to_string()];
     let rows = match maybe_path {
@@ -189,7 +209,7 @@ pub fn subgraph(
 
     let starts: Vec<&str> = start_nodes.iter().map(String::as_str).collect();
     let mv = max_visited(options);
-    let edges = csr.subgraph(&starts, edge_label, depth, mv);
+    let edges = csr.subgraph(&starts, edge_label, Direction::Out, depth, mv, None);
 
     let columns = vec!["src".to_string(), "label".to_string(), "dst".to_string()];
     let rows = edges

nodedb-lite/src/query/meta_ops/temporal.rs

@@ -196,6 +196,7 @@ mod tests {
             vector_state,
             array_state,
             fts_state,
+            Arc::new(crate::engine::sparse_vector::SparseVectorState::new()),
             spatial,
             Arc::new(Mutex::new(std::collections::HashMap::new())),
         )

nodedb-lite/src/query/physical_visitor/adapter/kv.rs

@@ -167,6 +167,7 @@ pub(super) fn dispatch<'a, S: StorageEngine + 'a>(
             collection,
             entries,
             ttl_ms,
+            surrogates: _,
         } => {
             let col = collection.clone();
             let ents = entries.clone();
@@ -209,6 +210,7 @@ pub(super) fn dispatch<'a, S: StorageEngine + 'a>(
             key,
             delta,
             ttl_ms,
+            surrogate: _,
         } => {
             let col = collection.clone();
             let k = key.clone();
@@ -223,6 +225,7 @@ pub(super) fn dispatch<'a, S: StorageEngine + 'a>(
             collection,
             key,
             delta,
+            surrogate: _,
         } => {
             let col = collection.clone();
             let k = key.clone();
@@ -237,6 +240,7 @@ pub(super) fn dispatch<'a, S: StorageEngine + 'a>(
             key,
             expected,
             new_value,
+            surrogate: _,
         } => {
             let col = collection.clone();
             let k = key.clone();
@@ -251,6 +255,7 @@ pub(super) fn dispatch<'a, S: StorageEngine + 'a>(
             collection,
             key,
             new_value,
+            surrogate: _,
         } => {
             let col = collection.clone();
             let k = key.clone();
@@ -264,6 +269,7 @@ pub(super) fn dispatch<'a, S: StorageEngine + 'a>(
             collection,
             key,
             updates,
+            surrogate: _,
         } => {
             let col = collection.clone();
             let k = key.clone();
@@ -279,6 +285,8 @@ pub(super) fn dispatch<'a, S: StorageEngine + 'a>(
             dest_key,
             field,
             amount,
+            debit_surrogate: _,
+            credit_surrogate: _,
         } => {
             let col = collection.clone();
             let src = source_key.clone();
@@ -295,6 +303,7 @@ pub(super) fn dispatch<'a, S: StorageEngine + 'a>(
             dest_collection,
             item_key,
             dest_key,
+            surrogate: _,
         } => {
             let src_col = source_collection.clone();
             let dst_col = dest_collection.clone();