refactor(query): split oversized engine.rs and graph.rs by concern

farhan-syah · farhan-syah · commit 563fadac1328 · 2026-07-21T03:34:40.000+08:00
Both files exceeded the 500-line non-test limit; split behavior-preserving:

- query/engine.rs (583 → 442): the INSERT/UPDATE/DELETE/TRUNCATE execution
  methods move into a second inherent `impl LiteQueryEngine` block in
  query/engine_dml.rs. `sql_value_to_string`/`sql_value_to_loro` widened to
  pub(super) so the new module can call them. Call sites (self.execute_*)
  are unchanged.
- physical_visitor/adapter/graph.rs (507 → ~486): the collection-resolution
  helper moves to a sibling graph_resolve.rs (it is CSR-traversal logic, not
  dispatch), imported back into the dispatcher.

Verified: compiles, clippy clean, 669/669 lite lib unit tests pass.
diff --git a/nodedb-lite/src/query/engine.rs b/nodedb-lite/src/query/engine.rs
@@ -334,150 +334,9 @@ impl<S: StorageEngine> LiteQueryEngine<S> {
         }
     }
 
-    pub(super) async fn execute_insert(
-        &self,
-        collection: &str,
-        engine: &EngineType,
-        rows: &[Vec<(String, SqlValue)>],
-        if_absent: bool,
-        primary_key: Option<&str>,
-    ) -> Result<QueryResult, LiteError> {
-        if *engine == EngineType::DocumentStrict {
-            return super::strict_dml::insert_strict(&self.strict, collection, rows, if_absent)
-                .await;
-        }
-        if *engine == EngineType::Columnar {
-            // `written` feeds outbound sync, which is compiled out on wasm32.
-            #[cfg_attr(target_arch = "wasm32", allow(unused_variables))]
-            let (result, written) =
-                super::columnar_dml::insert_columnar(&self.columnar, collection, rows)?;
-            // Durable outbound enqueue must run here (async) — the sync insert
-            // path cannot await. Covers the SQL-INSERT route to Origin sync.
-            #[cfg(not(target_arch = "wasm32"))]
-            crate::sync::reconcile_outbound_enqueue(
-                self.columnar.enqueue_outbound(collection, &written).await,
-                "columnar insert (sql)",
-                collection,
-                "",
-            )?;
-            return Ok(result);
-        }
-        // CRDT / schemaless path.
-        let mut crdt = self.crdt.lock().map_err(|_| LiteError::LockPoisoned)?;
-        let mut affected = 0;
-        for row in rows {
-            let id = row
-                .iter()
-                .find(|(k, _)| match primary_key {
-                    Some(pk) => k == pk,
-                    None => k == "id",
-                })
-                .map(|(_, v)| sql_value_to_string(v))
-                .unwrap_or_default();
-            if crdt.exists(collection, &id) {
-                if if_absent {
-                    continue;
-                }
-                return Err(LiteError::Query(format!(
-                    "duplicate key value violates unique constraint on '{collection}' (id = '{id}')"
-                )));
-            }
-            let fields: Vec<(&str, loro::LoroValue)> = row
-                .iter()
-                .map(|(k, v)| (k.as_str(), sql_value_to_loro(v)))
-                .collect();
-            crdt.upsert(collection, &id, &fields)
-                .map_err(|e| LiteError::Query(format!("insert: {e}")))?;
-            affected += 1;
-        }
-        Ok(QueryResult {
-            columns: Vec::new(),
-            rows: Vec::new(),
-            rows_affected: affected,
-        })
-    }
-
-    pub(super) async fn execute_update(
-        &self,
-        collection: &str,
-        engine: &EngineType,
-        assignments: &[(String, nodedb_sql::types::SqlExpr)],
-        target_keys: &[SqlValue],
-    ) -> Result<QueryResult, LiteError> {
-        if *engine == EngineType::DocumentStrict {
-            return super::strict_dml::update_strict(
-                &self.strict,
-                collection,
-                assignments,
-                target_keys,
-            )
-            .await;
-        }
-        // CRDT / schemaless path.
-        let mut crdt = self.crdt.lock().map_err(|_| LiteError::LockPoisoned)?;
-        let mut affected = 0;
-        for key in target_keys {
-            let key_str = sql_value_to_string(key);
-            let fields: Vec<(&str, loro::LoroValue)> = assignments
-                .iter()
-                .filter_map(|(field, expr)| {
-                    if let nodedb_sql::types::SqlExpr::Literal(val) = expr {
-                        Some((field.as_str(), sql_value_to_loro(val)))
-                    } else {
-                        None
-                    }
-                })
-                .collect();
-            crdt.upsert(collection, &key_str, &fields)
-                .map_err(|e| LiteError::Query(format!("update: {e}")))?;
-            affected += 1;
-        }
-        Ok(QueryResult {
-            columns: Vec::new(),
-            rows: Vec::new(),
-            rows_affected: affected,
-        })
-    }
-
-    pub(super) async fn execute_delete(
-        &self,
-        collection: &str,
-        engine: &EngineType,
-        target_keys: &[SqlValue],
-    ) -> Result<QueryResult, LiteError> {
-        if *engine == EngineType::DocumentStrict {
-            return super::strict_dml::delete_strict(&self.strict, collection, target_keys).await;
-        }
-        // CRDT / schemaless path.
-        let mut crdt = self.crdt.lock().map_err(|_| LiteError::LockPoisoned)?;
-        let mut affected = 0;
-        for key in target_keys {
-            let key_str = sql_value_to_string(key);
-            crdt.delete(collection, &key_str)
-                .map_err(|e| LiteError::Query(format!("delete: {e}")))?;
-            affected += 1;
-        }
-        Ok(QueryResult {
-            columns: Vec::new(),
-            rows: Vec::new(),
-            rows_affected: affected,
-        })
-    }
-
-    pub(super) async fn execute_truncate(
-        &self,
-        collection: &str,
-    ) -> Result<QueryResult, LiteError> {
-        self.crdt
-            .lock()
-            .map_err(|_| LiteError::LockPoisoned)?
-            .clear_collection(collection)
-            .map_err(|e| LiteError::Query(format!("truncate: {e}")))?;
-        Ok(QueryResult::empty())
-    }
 }
 
-fn sql_value_to_string(v: &SqlValue) -> String {
+pub(super) fn sql_value_to_string(v: &SqlValue) -> String {
     match v {
         SqlValue::String(s) => s.clone(),
         SqlValue::Int(i) => i.to_string(),
@@ -487,7 +346,7 @@ fn sql_value_to_string(v: &SqlValue) -> String {
     }
 }
 
-fn sql_value_to_loro(v: &SqlValue) -> loro::LoroValue {
+pub(super) fn sql_value_to_loro(v: &SqlValue) -> loro::LoroValue {
     match v {
         SqlValue::Int(i) => loro::LoroValue::I64(*i),
         SqlValue::Float(f) => loro::LoroValue::Double(*f),
diff --git a/nodedb-lite/src/query/engine_dml.rs b/nodedb-lite/src/query/engine_dml.rs
@@ -0,0 +1,154 @@
+//! DML execution methods for `LiteQueryEngine` (INSERT / UPDATE / DELETE /
+//! TRUNCATE). Split out of `engine.rs` as a second inherent `impl` block to
+//! keep that file under the size limit; behavior is unchanged.
+
+use nodedb_sql::types::{EngineType, SqlValue};
+use nodedb_types::result::QueryResult;
+
+use super::engine::{LiteQueryEngine, sql_value_to_loro, sql_value_to_string};
+use crate::error::LiteError;
+use crate::storage::engine::StorageEngine;
+
+impl<S: StorageEngine> LiteQueryEngine<S> {
+    pub(super) async fn execute_insert(
+        &self,
+        collection: &str,
+        engine: &EngineType,
+        rows: &[Vec<(String, SqlValue)>],
+        if_absent: bool,
+        primary_key: Option<&str>,
+    ) -> Result<QueryResult, LiteError> {
+        if *engine == EngineType::DocumentStrict {
+            return super::strict_dml::insert_strict(&self.strict, collection, rows, if_absent)
+                .await;
+        }
+        if *engine == EngineType::Columnar {
+            // `written` feeds outbound sync, which is compiled out on wasm32.
+            #[cfg_attr(target_arch = "wasm32", allow(unused_variables))]
+            let (result, written) =
+                super::columnar_dml::insert_columnar(&self.columnar, collection, rows)?;
+            // Durable outbound enqueue must run here (async) — the sync insert
+            // path cannot await. Covers the SQL-INSERT route to Origin sync.
+            #[cfg(not(target_arch = "wasm32"))]
+            crate::sync::reconcile_outbound_enqueue(
+                self.columnar.enqueue_outbound(collection, &written).await,
+                "columnar insert (sql)",
+                collection,
+                "",
+            )?;
+            return Ok(result);
+        }
+        // CRDT / schemaless path.
+        let mut crdt = self.crdt.lock().map_err(|_| LiteError::LockPoisoned)?;
+        let mut affected = 0;
+        for row in rows {
+            let id = row
+                .iter()
+                .find(|(k, _)| match primary_key {
+                    Some(pk) => k == pk,
+                    None => k == "id",
+                })
+                .map(|(_, v)| sql_value_to_string(v))
+                .unwrap_or_default();
+            if crdt.exists(collection, &id) {
+                if if_absent {
+                    continue;
+                }
+                return Err(LiteError::Query(format!(
+                    "duplicate key value violates unique constraint on '{collection}' (id = '{id}')"
+                )));
+            }
+            let fields: Vec<(&str, loro::LoroValue)> = row
+                .iter()
+                .map(|(k, v)| (k.as_str(), sql_value_to_loro(v)))
+                .collect();
+            crdt.upsert(collection, &id, &fields)
+                .map_err(|e| LiteError::Query(format!("insert: {e}")))?;
+            affected += 1;
+        }
+        Ok(QueryResult {
+            columns: Vec::new(),
+            rows: Vec::new(),
+            rows_affected: affected,
+        })
+    }
+
+    pub(super) async fn execute_update(
+        &self,
+        collection: &str,
+        engine: &EngineType,
+        assignments: &[(String, nodedb_sql::types::SqlExpr)],
+        target_keys: &[SqlValue],
+    ) -> Result<QueryResult, LiteError> {
+        if *engine == EngineType::DocumentStrict {
+            return super::strict_dml::update_strict(
+                &self.strict,
+                collection,
+                assignments,
+                target_keys,
+            )
+            .await;
+        }
+        // CRDT / schemaless path.
+        let mut crdt = self.crdt.lock().map_err(|_| LiteError::LockPoisoned)?;
+        let mut affected = 0;
+        for key in target_keys {
+            let key_str = sql_value_to_string(key);
+            let fields: Vec<(&str, loro::LoroValue)> = assignments
+                .iter()
+                .filter_map(|(field, expr)| {
+                    if let nodedb_sql::types::SqlExpr::Literal(val) = expr {
+                        Some((field.as_str(), sql_value_to_loro(val)))
+                    } else {
+                        None
+                    }
+                })
+                .collect();
+            crdt.upsert(collection, &key_str, &fields)
+                .map_err(|e| LiteError::Query(format!("update: {e}")))?;
+            affected += 1;
+        }
+        Ok(QueryResult {
+            columns: Vec::new(),
+            rows: Vec::new(),
+            rows_affected: affected,
+        })
+    }
+
+    pub(super) async fn execute_delete(
+        &self,
+        collection: &str,
+        engine: &EngineType,
+        target_keys: &[SqlValue],
+    ) -> Result<QueryResult, LiteError> {
+        if *engine == EngineType::DocumentStrict {
+            return super::strict_dml::delete_strict(&self.strict, collection, target_keys).await;
+        }
+        // CRDT / schemaless path.
+        let mut crdt = self.crdt.lock().map_err(|_| LiteError::LockPoisoned)?;
+        let mut affected = 0;
+        for key in target_keys {
+            let key_str = sql_value_to_string(key);
+            crdt.delete(collection, &key_str)
+                .map_err(|e| LiteError::Query(format!("delete: {e}")))?;
+            affected += 1;
+        }
+        Ok(QueryResult {
+            columns: Vec::new(),
+            rows: Vec::new(),
+            rows_affected: affected,
+        })
+    }
+
+    pub(super) async fn execute_truncate(
+        &self,
+        collection: &str,
+    ) -> Result<QueryResult, LiteError> {
+        self.crdt
+            .lock()
+            .map_err(|_| LiteError::LockPoisoned)?
+            .clear_collection(collection)
+            .map_err(|e| LiteError::Query(format!("truncate: {e}")))?;
+        Ok(QueryResult::empty())
+    }
+}
diff --git a/nodedb-lite/src/query/mod.rs b/nodedb-lite/src/query/mod.rs
@@ -6,6 +6,7 @@ pub mod crdt_ops;
 pub mod ddl;
 pub mod document_ops;
 pub mod engine;
+pub(crate) mod engine_dml;
 pub(crate) mod expr_convert;
 pub(crate) mod filter_convert;
 pub(crate) mod graph_ops;
diff --git a/nodedb-lite/src/query/physical_visitor/adapter/graph.rs b/nodedb-lite/src/query/physical_visitor/adapter/graph.rs
@@ -10,6 +10,7 @@
 
 use std::future::Future;
 use std::pin::Pin;
+use std::sync::Arc;
 
 use nodedb_physical::physical_plan::GraphOp;
 use nodedb_types::result::QueryResult;
@@ -21,6 +22,8 @@ use crate::query::graph_ops::{
 };
 use crate::storage::engine::StorageEngine;
 
+use super::graph_resolve::resolve_collection_for_nodes;
+
 #[cfg(not(target_arch = "wasm32"))]
 pub(crate) type GraphFut<'a> =
     Pin<Box<dyn Future<Output = Result<QueryResult, LiteError>> + Send + 'a>>;
@@ -478,30 +481,3 @@ pub(crate) fn dispatch<'a, S: StorageEngine + 'a>(
 
     Ok(fut)
 }
-
-/// Resolve which collection a set of node IDs belongs to by scanning the CSR map.
-///
-/// Returns the first collection that contains any of the given nodes, or an
-/// empty string when none is found (which will produce an empty result set
-/// rather than an error — correct for "no such graph" semantics).
-fn resolve_collection_for_nodes(
-    csr_map: &Arc<
-        std::sync::Mutex<std::collections::HashMap<String, crate::engine::graph::index::CsrIndex>>,
-    >,
-    node_ids: &[String],
-) -> String {
-    let Ok(map) = csr_map.lock() else {
-        return String::new();
-    };
-    for (coll, csr) in map.iter() {
-        for node in node_ids {
-            if csr.contains_node(node) {
-                return coll.clone();
-            }
-        }
-    }
-    // Fall back to the first collection in the map.
-    map.keys().next().cloned().unwrap_or_default()
-}
-
-use std::sync::Arc;
diff --git a/nodedb-lite/src/query/physical_visitor/adapter/graph_resolve.rs b/nodedb-lite/src/query/physical_visitor/adapter/graph_resolve.rs
diff --git a/nodedb-lite/src/query/physical_visitor/adapter/mod.rs b/nodedb-lite/src/query/physical_visitor/adapter/mod.rs
