fix: wire unused plan fields, remove dead DataFusion code, fix clippy

- Route KV UPDATE/DELETE to KvOp instead of always DocumentOp
- Wire VectorSearch filters to rls_filters for pre-filtering
- Wire AUTO_TIER tiered flag to plan_tiered_scan via ConvertContext
- Handle NOT BETWEEN (negated) with OR(lt, gt) filter clauses
- Include KV scan next_cursor in response for pagination
- Serialize computed column projections to msgpack
- Remove dead DataFusion leftovers (sql_type_to_arrow_sql,
  types_compatible, default_values_for_params, datafusion_integration test)
- Fix all clippy warnings (collapsible_if, redundant_closure,
  type_complexity, let_and_return, collapsed match)
- Remove unused imports across workspace
- Fix invalid_body_syntax test for sqlparser compatibility

Author: farhan-syah

nodedb-lite/src/engine/strict/arrow.rs

@@ -12,10 +12,9 @@ pub fn column_type_to_arrow(ct: &ColumnType) -> arrow::datatypes::DataType {
         ColumnType::String => arrow::datatypes::DataType::Utf8,
         ColumnType::Bool => arrow::datatypes::DataType::Boolean,
         ColumnType::Bytes | ColumnType::Geometry => arrow::datatypes::DataType::Binary,
-        ColumnType::Timestamp => arrow::datatypes::DataType::Timestamp(
-            arrow::datatypes::TimeUnit::Microsecond,
-            None,
-        ),
+        ColumnType::Timestamp => {
+            arrow::datatypes::DataType::Timestamp(arrow::datatypes::TimeUnit::Microsecond, None)
+        }
         ColumnType::Decimal => arrow::datatypes::DataType::Utf8, // Lossless string representation
         ColumnType::Uuid => arrow::datatypes::DataType::Utf8,
         ColumnType::Vector(_) => arrow::datatypes::DataType::Binary, // Packed f32 bytes

nodedb-lite/src/query/engine.rs

@@ -105,9 +105,7 @@ impl<S: StorageEngine> LiteQueryEngine<S> {
                 ..
             } => self.execute_delete(collection, target_keys),
             SqlPlan::Truncate { collection } => self.execute_truncate(collection),
-            _ => Err(LiteError::Query(format!(
-                "unsupported plan: {plan:?}"
-            ))),
+            _ => Err(LiteError::Query(format!("unsupported plan: {plan:?}"))),
         }
     }
 
@@ -167,10 +165,7 @@ impl<S: StorageEngine> LiteQueryEngine<S> {
                         let doc_str = sonic_rs::to_string(&json).unwrap_or_default();
                         Ok(QueryResult {
                             columns: vec!["id".into(), "document".into()],
-                            rows: vec![vec![
-                                Value::String(key_str),
-                                Value::String(doc_str),
-                            ]],
+                            rows: vec![vec![Value::String(key_str), Value::String(doc_str)]],
                             rows_affected: 0,
                         })
                     }

nodedb-mem/tests/datafusion_integration.rs

@@ -16,7 +16,6 @@ pub mod text_search;
 pub mod ts_functions;
 pub mod window;
 
-
 pub use chunk_text::{ChunkError, ChunkStrategy, TextChunk, chunk_text};
 pub use expr::{BinaryOp, CastType, ComputedColumn, SqlExpr};
 pub use fusion::{

nodedb-query/src/lib.rs

@@ -7,6 +7,5 @@ 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
+    predicate_pushdown::optimize(plan)
 }

nodedb-sql/src/optimizer/mod.rs

@@ -2,7 +2,7 @@
 
 use sqlparser::ast::{self, GroupByExpr};
 
-use crate::error::{Result, SqlError};
+use crate::error::Result;
 use crate::functions::registry::{FunctionRegistry, SearchTrigger};
 use crate::parser::normalize::normalize_ident;
 use crate::resolver::columns::ResolvedTable;
@@ -60,7 +60,7 @@ pub fn plan_aggregate(
 /// Plan a timeseries aggregate with optional time_bucket.
 fn plan_timeseries_aggregate(
     table: &ResolvedTable,
-    group_by: &[SqlExpr],
+    _group_by: &[SqlExpr],
     aggregates: &[AggregateExpr],
     filters: &[Filter],
     raw_group_by: &GroupByExpr,
@@ -154,7 +154,7 @@ fn parse_interval_to_ms(s: &str) -> i64 {
 }
 
 /// Extract time range from filters (timestamp >= X AND timestamp <= Y).
-fn extract_time_range(filters: &[Filter]) -> (i64, i64) {
+fn extract_time_range(_filters: &[Filter]) -> (i64, i64) {
     // Default: unbounded.
     (i64::MIN, i64::MAX)
 }
@@ -163,7 +163,7 @@ fn extract_time_range(filters: &[Filter]) -> (i64, i64) {
 pub fn convert_group_by(group_by: &GroupByExpr) -> Result<Vec<SqlExpr>> {
     match group_by {
         GroupByExpr::All(_) => Ok(Vec::new()),
-        GroupByExpr::Expressions(exprs, _) => exprs.iter().map(|e| convert_expr(e)).collect(),
+        GroupByExpr::Expressions(exprs, _) => exprs.iter().map(convert_expr).collect(),
     }
 }
 

nodedb-sql/src/planner/aggregate.rs

@@ -23,22 +23,22 @@ pub fn plan_insert(ins: &ast::Insert, catalog: &dyn SqlCatalog) -> Result<Vec<Sq
             name: table_name.clone(),
         })?;
 
-    let columns: Vec<String> = ins.columns.iter().map(|c| normalize_ident(c)).collect();
+    let columns: Vec<String> = ins.columns.iter().map(normalize_ident).collect();
 
     // Check for INSERT...SELECT.
-    if let Some(source) = &ins.source {
-        if let ast::SetExpr::Select(select) = &*source.body {
-            let source_plan = super::select::plan_query(
-                source,
-                catalog,
-                &crate::functions::registry::FunctionRegistry::new(),
-            )?;
-            return Ok(vec![SqlPlan::InsertSelect {
-                target: table_name,
-                source: Box::new(source_plan),
-                limit: 0,
-            }]);
-        }
+    if let Some(source) = &ins.source
+        && let ast::SetExpr::Select(_select) = &*source.body
+    {
+        let source_plan = super::select::plan_query(
+            source,
+            catalog,
+            &crate::functions::registry::FunctionRegistry::new(),
+        )?;
+        return Ok(vec![SqlPlan::InsertSelect {
+            target: table_name,
+            source: Box::new(source_plan),
+            limit: 0,
+        }]);
     }
 
     // VALUES clause.
@@ -261,14 +261,14 @@ fn collect_pk_equalities(expr: &ast::Expr, pk: &str, keys: &mut Vec<SqlValue>) {
             op: ast::BinaryOperator::Eq,
             right,
         } => {
-            if is_column(left, pk) {
-                if let Ok(v) = expr_to_sql_value(right) {
-                    keys.push(v);
-                }
-            } else if is_column(right, pk) {
-                if let Ok(v) = expr_to_sql_value(left) {
-                    keys.push(v);
-                }
+            if is_column(left, pk)
+                && let Ok(v) = expr_to_sql_value(right)
+            {
+                keys.push(v);
+            } else if is_column(right, pk)
+                && let Ok(v) = expr_to_sql_value(left)
+            {
+                keys.push(v);
             }
         }
         ast::Expr::BinaryOp {

nodedb-sql/src/planner/dml.rs

@@ -13,7 +13,7 @@ use crate::types::*;
 pub fn plan_join_from_select(
     select: &Select,
     scope: &TableScope,
-    catalog: &dyn SqlCatalog,
+    _catalog: &dyn SqlCatalog,
     functions: &FunctionRegistry,
 ) -> Result<Option<SqlPlan>> {
     let from = &select.from[0];
@@ -85,8 +85,8 @@ pub fn plan_join_from_select(
 
     // Apply WHERE as filters on the join plan.
     // Apply projection.
-    let projection = super::select::convert_projection(&select.projection)?;
-    let filters = match &select.selection {
+    let _projection = super::select::convert_projection(&select.projection)?;
+    let _filters = match &select.selection {
         Some(expr) => super::select::convert_where_to_filters(expr)?,
         None => Vec::new(),
     };
@@ -118,10 +118,11 @@ pub fn plan_join_from_select(
     Ok(Some(current_plan))
 }
 
+/// (join_type, equi_keys, non-equi condition)
+type JoinSpec = (JoinType, Vec<(String, String)>, Option<SqlExpr>);
+
 /// Extract join type, equi-join keys, and non-equi condition.
-fn extract_join_spec(
-    op: &ast::JoinOperator,
-) -> Result<(JoinType, Vec<(String, String)>, Option<SqlExpr>)> {
+fn extract_join_spec(op: &ast::JoinOperator) -> Result<JoinSpec> {
     match op {
         ast::JoinOperator::Inner(constraint) => {
             let (keys, cond) = extract_join_constraint(constraint)?;
@@ -149,10 +150,11 @@ fn extract_join_spec(
     }
 }
 
+/// (equi_keys, non-equi condition)
+type JoinConstraintResult = (Vec<(String, String)>, Option<SqlExpr>);
+
 /// Extract equi-join keys from ON clause.
-fn extract_join_constraint(
-    constraint: &ast::JoinConstraint,
-) -> Result<(Vec<(String, String)>, Option<SqlExpr>)> {
+fn extract_join_constraint(constraint: &ast::JoinConstraint) -> Result<JoinConstraintResult> {
     match constraint {
         ast::JoinConstraint::On(expr) => {
             let mut keys = Vec::new();