feat(sql): replace DataFusion converter with nodedb-sql crate

Add nodedb-sql shared crate that parses SQL via sqlparser-rs and maps
directly to SqlPlan IR, eliminating the fragile 813-line converter.rs
that reverse-engineered DataFusion's LogicalPlan.

New crate (nodedb-sql/):
- parser/: sqlparser-rs wrapper, statement classification, normalization
- resolver/: column resolution, expression conversion (AST → SqlExpr)
- planner/: SELECT, DML, JOIN, GROUP BY, ORDER BY, window, UNION, CTE
- optimizer/: point-get detection, predicate pushdown stub
- functions/: 60+ built-in function registry with search triggers
- types.rs: SqlPlan IR with 20 plan variants, SqlCatalog trait

Origin integration:
- catalog_adapter.rs: SqlCatalog impl via CredentialStore
- sql_plan_convert.rs: SqlPlan → PhysicalPlan + vShard routing
- context.rs: plan_sql now uses nodedb-sql directly

Deleted (~4100 lines):
- converter.rs, converter_helpers.rs, search.rs, schemaless_dml.rs
- join.rs, dml.rs, dml_kv.rs, dml_columnar.rs, dml_timeseries.rs
- extract/ directory (filter.rs, dml.rs, spatial_expr.rs, etc.)

DataFusion dependency remains for UDFs, EXPLAIN, and Arrow types.

Author: farhan-syah

Cargo.toml

@@ -21,6 +21,7 @@ members = [
   "nodedb-fts",
   "nodedb-strict",
   "nodedb-columnar",
+  "nodedb-sql",
 ]
 resolver = "2"
 
@@ -53,6 +54,7 @@ nodedb-vector = { path = "nodedb-vector", version = "0.0.0-beta.1" }
 nodedb-fts = { path = "nodedb-fts", version = "0.0.0-beta.1" }
 nodedb-strict = { path = "nodedb-strict", version = "0.0.0-beta.1" }
 nodedb-columnar = { path = "nodedb-columnar", version = "0.0.0-beta.1" }
+nodedb-sql = { path = "nodedb-sql", version = "0.0.0-beta.1" }
 
 # Async runtimes
 tokio = { version = "1", features = ["full"] }

nodedb-sql/Cargo.toml

@@ -0,0 +1,11 @@
+[package]
+name = "nodedb-sql"
+version.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+license.workspace = true
+description = "SQL parser, planner, and optimizer for NodeDB"
+
+[dependencies]
+sqlparser = "0.61"
+thiserror = { workspace = true }

nodedb-sql/src/error.rs

@@ -0,0 +1,39 @@
+//! Error types for the nodedb-sql crate.
+
+/// Errors produced during SQL parsing, resolution, or planning.
+#[derive(Debug, thiserror::Error)]
+pub enum SqlError {
+    #[error("parse error: {detail}")]
+    Parse { detail: String },
+
+    #[error("unknown table: {name}")]
+    UnknownTable { name: String },
+
+    #[error("unknown column '{column}' in table '{table}'")]
+    UnknownColumn { table: String, column: String },
+
+    #[error("ambiguous column '{column}' — qualify with table name")]
+    AmbiguousColumn { column: String },
+
+    #[error("type mismatch: {detail}")]
+    TypeMismatch { detail: String },
+
+    #[error("unsupported: {detail}")]
+    Unsupported { detail: String },
+
+    #[error("invalid function call: {detail}")]
+    InvalidFunction { detail: String },
+
+    #[error("missing required field '{field}' for {context}")]
+    MissingField { field: String, context: String },
+}
+
+impl From<sqlparser::parser::ParserError> for SqlError {
+    fn from(e: sqlparser::parser::ParserError) -> Self {
+        Self::Parse {
+            detail: e.to_string(),
+        }
+    }
+}
+
+pub type Result<T> = std::result::Result<T, SqlError>;

nodedb-sql/src/functions/mod.rs

@@ -0,0 +1 @@
+pub mod registry;

nodedb-sql/src/functions/registry.rs

@@ -0,0 +1,211 @@
+//! Built-in function registry for SQL planning.
+//!
+//! Tracks known functions, their categories, and whether they trigger
+//! special engine routing (e.g., vector_distance → VectorSearch).
+
+/// Function category.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum FunctionCategory {
+    Scalar,
+    Aggregate,
+    Window,
+}
+
+/// Whether a function triggers special engine routing.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum SearchTrigger {
+    None,
+    VectorSearch,
+    MultiVectorSearch,
+    TextSearch,
+    HybridSearch,
+    TextMatch,
+    SpatialDWithin,
+    SpatialContains,
+    SpatialIntersects,
+    SpatialWithin,
+    TimeBucket,
+}
+
+/// Metadata about a known function.
+#[derive(Debug, Clone)]
+pub struct FunctionMeta {
+    pub name: &'static str,
+    pub category: FunctionCategory,
+    pub min_args: usize,
+    pub max_args: usize,
+    pub search_trigger: SearchTrigger,
+}
+
+/// The function registry.
+pub struct FunctionRegistry {
+    functions: Vec<FunctionMeta>,
+}
+
+impl FunctionRegistry {
+    /// Create the default registry with all built-in functions.
+    pub fn new() -> Self {
+        Self {
+            functions: builtin_functions(),
+        }
+    }
+
+    /// Look up a function by name (case-insensitive).
+    pub fn lookup(&self, name: &str) -> Option<&FunctionMeta> {
+        let lower = name.to_lowercase();
+        self.functions.iter().find(|f| f.name == lower)
+    }
+
+    /// Check if a function triggers special search routing.
+    pub fn search_trigger(&self, name: &str) -> SearchTrigger {
+        self.lookup(name)
+            .map(|f| f.search_trigger)
+            .unwrap_or(SearchTrigger::None)
+    }
+
+    /// Check if a function is an aggregate.
+    pub fn is_aggregate(&self, name: &str) -> bool {
+        self.lookup(name)
+            .is_some_and(|f| f.category == FunctionCategory::Aggregate)
+    }
+
+    /// Check if a function is a window function.
+    pub fn is_window(&self, name: &str) -> bool {
+        self.lookup(name)
+            .is_some_and(|f| f.category == FunctionCategory::Window)
+    }
+}
+
+impl Default for FunctionRegistry {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+fn s(
+    name: &'static str,
+    cat: FunctionCategory,
+    min: usize,
+    max: usize,
+    trigger: SearchTrigger,
+) -> FunctionMeta {
+    FunctionMeta {
+        name,
+        category: cat,
+        min_args: min,
+        max_args: max,
+        search_trigger: trigger,
+    }
+}
+
+fn builtin_functions() -> Vec<FunctionMeta> {
+    use FunctionCategory::*;
+    use SearchTrigger::*;
+
+    vec![
+        // ── Standard aggregates ──
+        s("count", Aggregate, 0, 1, None),
+        s("sum", Aggregate, 1, 1, None),
+        s("avg", Aggregate, 1, 1, None),
+        s("min", Aggregate, 1, 1, None),
+        s("max", Aggregate, 1, 1, None),
+        // ── Standard window ──
+        s("row_number", Window, 0, 0, None),
+        s("rank", Window, 0, 0, None),
+        s("dense_rank", Window, 0, 0, None),
+        s("lag", Window, 1, 3, None),
+        s("lead", Window, 1, 3, None),
+        s("first_value", Window, 1, 1, None),
+        s("last_value", Window, 1, 1, None),
+        s("nth_value", Window, 2, 2, None),
+        // ── Vector search ──
+        s("vector_distance", Scalar, 2, 3, VectorSearch),
+        s("multi_vector_search", Scalar, 1, 2, MultiVectorSearch),
+        s("multi_vector_score", Scalar, 3, 3, None),
+        s("sparse_score", Scalar, 3, 3, None),
+        // ── Text search ──
+        s("bm25_score", Scalar, 2, 2, TextSearch),
+        s("text_match", Scalar, 2, 2, TextMatch),
+        // ── Hybrid search ──
+        s("rrf_score", Scalar, 2, 4, HybridSearch),
+        // ── Spatial ──
+        s("st_dwithin", Scalar, 3, 3, SpatialDWithin),
+        s("st_contains", Scalar, 2, 2, SpatialContains),
+        s("st_intersects", Scalar, 2, 2, SpatialIntersects),
+        s("st_within", Scalar, 2, 2, SpatialWithin),
+        s("st_distance", Scalar, 2, 2, None),
+        s("st_point", Scalar, 2, 2, None),
+        // ── Timeseries ──
+        s("time_bucket", Scalar, 2, 2, TimeBucket),
+        // ── Timeseries aggregates ──
+        s("ts_percentile", Aggregate, 2, 2, None),
+        s("ts_stddev", Aggregate, 1, 1, None),
+        s("ts_correlate", Aggregate, 2, 2, None),
+        // ── Timeseries window ──
+        s("ts_rate", Window, 1, 1, None),
+        s("ts_derivative", Window, 1, 1, None),
+        s("ts_moving_avg", Window, 2, 2, None),
+        s("ts_ema", Window, 2, 2, None),
+        s("ts_delta", Window, 1, 1, None),
+        s("ts_interpolate", Window, 1, 1, None),
+        s("ts_lag", Window, 1, 3, None),
+        s("ts_lead", Window, 1, 3, None),
+        s("ts_rank", Window, 0, 0, None),
+        // ── Approximate aggregates ──
+        s("approx_count_distinct", Aggregate, 1, 1, None),
+        s("approx_percentile", Aggregate, 2, 2, None),
+        s("approx_topk", Aggregate, 2, 2, None),
+        s("approx_count", Aggregate, 1, 1, None),
+        // ── Document helpers ──
+        s("doc_get", Scalar, 2, 3, None),
+        s("doc_exists", Scalar, 2, 2, None),
+        s("doc_array_contains", Scalar, 3, 3, None),
+        s("nav", Scalar, 2, 2, None),
+        // ── Utility ──
+        s("chunk_text", Scalar, 2, 3, None),
+        s("currency", Scalar, 1, 2, None),
+        s("distribute", Scalar, 2, 3, None),
+        s("allocate", Scalar, 2, 3, None),
+        s("resolve_permission", Scalar, 2, 3, None),
+        // ── Standard scalar ──
+        s("coalesce", Scalar, 1, 255, None),
+        s("nullif", Scalar, 2, 2, None),
+        s("abs", Scalar, 1, 1, None),
+        s("ceil", Scalar, 1, 1, None),
+        s("floor", Scalar, 1, 1, None),
+        s("round", Scalar, 1, 2, None),
+        s("lower", Scalar, 1, 1, None),
+        s("upper", Scalar, 1, 1, None),
+        s("length", Scalar, 1, 1, None),
+        s("trim", Scalar, 1, 1, None),
+        s("substring", Scalar, 2, 3, None),
+        s("concat", Scalar, 1, 255, None),
+        s("replace", Scalar, 3, 3, None),
+        s("now", Scalar, 0, 0, None),
+        s("current_timestamp", Scalar, 0, 0, None),
+        s("make_array", Scalar, 0, 255, None),
+    ]
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn lookup_builtin() {
+        let reg = FunctionRegistry::new();
+        assert!(reg.is_aggregate("COUNT"));
+        assert!(reg.is_aggregate("sum"));
+        assert!(!reg.is_aggregate("vector_distance"));
+        assert!(reg.is_window("row_number"));
+        assert_eq!(
+            reg.search_trigger("vector_distance"),
+            SearchTrigger::VectorSearch
+        );
+        assert_eq!(
+            reg.search_trigger("st_dwithin"),
+            SearchTrigger::SpatialDWithin
+        );
+        assert_eq!(reg.search_trigger("time_bucket"), SearchTrigger::TimeBucket);
+    }
+}

nodedb-sql/src/lib.rs

@@ -0,0 +1,66 @@
+//! nodedb-sql: SQL parser, planner, and optimizer for NodeDB.
+//!
+//! Parses SQL via sqlparser-rs, resolves against a catalog, and produces
+//! `SqlPlan` — an intermediate representation that both Origin (server)
+//! and Lite (embedded) map to their own execution model.
+//!
+//! ```text
+//! SQL → parse → resolve → plan → optimize → SqlPlan
+//! ```
+
+pub mod error;
+pub mod functions;
+pub mod optimizer;
+pub mod parser;
+pub mod planner;
+pub mod resolver;
+pub mod types;
+
+pub use error::{Result, SqlError};
+pub use types::*;
+
+use functions::registry::FunctionRegistry;
+use parser::statement::{StatementKind, classify, parse_sql};
+
+/// Plan one or more SQL statements against the given catalog.
+///
+/// Returns a list of `SqlPlan` — one per statement (some statements
+/// like multi-row INSERT may produce multiple plans).
+pub fn plan_sql(sql: &str, catalog: &dyn SqlCatalog) -> Result<Vec<SqlPlan>> {
+    let functions = FunctionRegistry::new();
+    let statements = parse_sql(sql)?;
+    let mut plans = Vec::new();
+
+    for stmt in &statements {
+        match classify(stmt) {
+            StatementKind::Select(query) => {
+                let plan = planner::select::plan_query(query, catalog, &functions)?;
+                let plan = optimizer::optimize(plan);
+                plans.push(plan);
+            }
+            StatementKind::Insert(ins) => {
+                let mut insert_plans = planner::dml::plan_insert(ins, catalog)?;
+                plans.append(&mut insert_plans);
+            }
+            StatementKind::Update(stmt) => {
+                let mut update_plans = planner::dml::plan_update(stmt, catalog)?;
+                plans.append(&mut update_plans);
+            }
+            StatementKind::Delete(stmt) => {
+                let mut delete_plans = planner::dml::plan_delete(stmt, catalog)?;
+                plans.append(&mut delete_plans);
+            }
+            StatementKind::Truncate(stmt) => {
+                let mut trunc_plans = planner::dml::plan_truncate_stmt(stmt)?;
+                plans.append(&mut trunc_plans);
+            }
+            StatementKind::Other => {
+                return Err(SqlError::Unsupported {
+                    detail: format!("statement type: {stmt}"),
+                });
+            }
+        }
+    }
+
+    Ok(plans)
+}

nodedb-sql/src/optimizer/constant_fold.rs

@@ -0,0 +1,4 @@
+//! Constant folding: evaluate constant expressions at plan time.
+//!
+//! Examples: `1 + 2` → `3`, `WHERE 1 = 1` → remove filter.
+//! Currently a placeholder — the Data Plane handles expression evaluation.

nodedb-sql/src/optimizer/mod.rs

@@ -0,0 +1,12 @@
+pub mod constant_fold;
+pub mod point_get;
+pub mod predicate_pushdown;
+
+use crate::types::SqlPlan;
+
+/// Apply all optimization passes to a plan.
+pub fn optimize(plan: SqlPlan) -> SqlPlan {
+    let plan = point_get::optimize(plan);
+    let plan = predicate_pushdown::optimize(plan);
+    plan
+}