refactor: replace datafusion::arrow with arrow crate, delete dead UDFs

- Add arrow 58 as direct dependency, replace all datafusion::arrow
  re-exports in 10 non-UDF files (response_codec, cold storage, etc.)
- Delete inline_rewrite.rs and user_function.rs (dead — planning
  no longer uses DataFusion UDF inlining)
- Remove register_udfs_on from context.rs (no callers)
- Replace validate.rs DataFusion-based validation with sqlparser-based
  syntax check
- eval.rs: replace datafusion::arrow with arrow crate

DataFusion remains only for:
- Procedural executor expression evaluation (eval_sql_expr)
- UDF trait implementations (still needed by eval.rs indirectly)
- Temp table MemTable storage

Author: farhan-syah

Cargo.toml

@@ -130,7 +130,10 @@ snap = "1"
 flate2 = "1"
 tonic = "0.14"
 
-# Query engine
+# Arrow (columnar format)
+arrow = { version = "58", default-features = false, features = ["ipc"] }
+
+# Query engine (retained for procedural executor + function body validation)
 datafusion = { version = "53", default-features = false, features = ["sql"] }
 datafusion-common = "53"
 datafusion-execution = "53"

nodedb/Cargo.toml

@@ -36,6 +36,7 @@ nodedb-spatial = { workspace = true }
 nodedb-graph = { workspace = true }
 nodedb-vector = { workspace = true, features = ["collection"] }
 nodedb-sql = { workspace = true }
+arrow = { workspace = true }
 
 # Async runtime (Control Plane)
 tokio = { workspace = true }

nodedb/src/control/arrow_convert.rs

@@ -13,9 +13,9 @@
 
 use std::sync::Arc;
 
-use datafusion::arrow::array::{ArrayRef, Float64Array, Int64Array, StringArray};
-use datafusion::arrow::datatypes::{DataType, Field, Schema, SchemaRef};
-use datafusion::arrow::record_batch::RecordBatch;
+use arrow::array::{ArrayRef, Float64Array, Int64Array, StringArray};
+use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
+use arrow::record_batch::RecordBatch;
 
 /// Convert a JSON array of document rows into an Arrow RecordBatch.
 ///
@@ -163,9 +163,9 @@ pub fn arrow_sum(batch: &RecordBatch, column_name: &str) -> Option<f64> {
     let array = batch.column(idx);
 
     if let Some(f64_arr) = array.as_any().downcast_ref::<Float64Array>() {
-        Some(datafusion::arrow::compute::kernels::aggregate::sum(f64_arr).unwrap_or(0.0))
+        Some(arrow::compute::kernels::aggregate::sum(f64_arr).unwrap_or(0.0))
     } else if let Some(i64_arr) = array.as_any().downcast_ref::<Int64Array>() {
-        let sum = datafusion::arrow::compute::kernels::aggregate::sum(i64_arr).unwrap_or(0);
+        let sum = arrow::compute::kernels::aggregate::sum(i64_arr).unwrap_or(0);
         Some(sum as f64)
     } else {
         None
@@ -178,9 +178,9 @@ pub fn arrow_min(batch: &RecordBatch, column_name: &str) -> Option<f64> {
     let array = batch.column(idx);
 
     if let Some(f64_arr) = array.as_any().downcast_ref::<Float64Array>() {
-        datafusion::arrow::compute::kernels::aggregate::min(f64_arr)
+        arrow::compute::kernels::aggregate::min(f64_arr)
     } else if let Some(i64_arr) = array.as_any().downcast_ref::<Int64Array>() {
-        datafusion::arrow::compute::kernels::aggregate::min(i64_arr).map(|v| v as f64)
+        arrow::compute::kernels::aggregate::min(i64_arr).map(|v| v as f64)
     } else {
         None
     }
@@ -192,9 +192,9 @@ pub fn arrow_max(batch: &RecordBatch, column_name: &str) -> Option<f64> {
     let array = batch.column(idx);
 
     if let Some(f64_arr) = array.as_any().downcast_ref::<Float64Array>() {
-        datafusion::arrow::compute::kernels::aggregate::max(f64_arr)
+        arrow::compute::kernels::aggregate::max(f64_arr)
     } else if let Some(i64_arr) = array.as_any().downcast_ref::<Int64Array>() {
-        datafusion::arrow::compute::kernels::aggregate::max(i64_arr).map(|v| v as f64)
+        arrow::compute::kernels::aggregate::max(i64_arr).map(|v| v as f64)
     } else {
         None
     }
@@ -223,7 +223,7 @@ pub fn arrow_avg(batch: &RecordBatch, column_name: &str) -> Option<f64> {
 /// Used by the Control Plane to receive Arrow data from the Data Plane
 /// across the SPSC bridge.
 pub fn decode_arrow_ipc(bytes: &[u8]) -> Option<RecordBatch> {
-    use datafusion::arrow::ipc::reader::StreamReader;
+    use arrow::ipc::reader::StreamReader;
     let cursor = std::io::Cursor::new(bytes);
     let mut reader = StreamReader::try_new(cursor, None).ok()?;
     reader.next()?.ok()

nodedb/src/control/planner/context.rs

@@ -207,42 +207,3 @@ pub const SYSTEM_FUNCTION_NAMES: &[&str] = &[
     "setval",
     "next_preview",
 ];
-
-/// Register all system UDFs on a DataFusion `SessionContext`.
-///
-/// Used only by the DDL body validator (CREATE FUNCTION) which needs
-/// a temporary session with all system UDFs registered for validation.
-pub fn register_udfs_on(session: &datafusion::execution::context::SessionContext) {
-    use super::udf::spatial::{
-        GeoDistance, StContains, StDistance, StDwithin, StIntersects, StWithin,
-    };
-    use super::udf::{
-        Allocate, Bm25Score, ChunkText, ConvertCurrency, Distribute, DocArrayContains, DocExists,
-        DocGet, MultiVectorScore, MultiVectorSearch, RoundDecimal, RrfScore, SparseScore,
-        TextMatch, VectorDistance, VectorMetadata,
-    };
-    use datafusion::logical_expr::ScalarUDF;
-    session.register_udf(ScalarUDF::new_from_impl(ChunkText::new()));
-    session.register_udf(ScalarUDF::new_from_impl(DocGet::new()));
-    session.register_udf(ScalarUDF::new_from_impl(DocExists::new()));
-    session.register_udf(ScalarUDF::new_from_impl(DocArrayContains::new()));
-    session.register_udf(ScalarUDF::new_from_impl(VectorDistance::new()));
-    session.register_udf(ScalarUDF::new_from_impl(VectorMetadata::new()));
-    session.register_udf(ScalarUDF::new_from_impl(MultiVectorScore::new()));
-    session.register_udf(ScalarUDF::new_from_impl(MultiVectorSearch::new()));
-    session.register_udf(ScalarUDF::new_from_impl(RrfScore::new()));
-    session.register_udf(ScalarUDF::new_from_impl(SparseScore::new()));
-    session.register_udf(ScalarUDF::new_from_impl(Bm25Score::new()));
-    session.register_udf(ScalarUDF::new_from_impl(TextMatch::new()));
-    session.register_udf(ScalarUDF::new_from_impl(StDwithin::new()));
-    session.register_udf(ScalarUDF::new_from_impl(StContains::new()));
-    session.register_udf(ScalarUDF::new_from_impl(StIntersects::new()));
-    session.register_udf(ScalarUDF::new_from_impl(StWithin::new()));
-    session.register_udf(ScalarUDF::new_from_impl(StDistance::new()));
-    session.register_udf(ScalarUDF::new_from_impl(GeoDistance::new()));
-    session.register_udf(ScalarUDF::new_from_impl(RoundDecimal::new("HALF_EVEN")));
-    session.register_udf(ScalarUDF::new_from_impl(Distribute::new()));
-    session.register_udf(ScalarUDF::new_from_impl(Allocate::new()));
-    session.register_udf(ScalarUDF::new_from_impl(ConvertCurrency::new()));
-    nodedb_query::ts_udfs::register_timeseries_udfs(session);
-}

nodedb/src/control/planner/procedural/executor/arrow_conv.rs

@@ -6,8 +6,8 @@
 
 use std::sync::Arc;
 
-use datafusion::arrow::array::*;
-use datafusion::arrow::datatypes::DataType;
+use arrow::array::*;
+use arrow::datatypes::DataType;
 use nodedb_types::Value;
 
 /// Extract a single scalar value from an Arrow array at the given row index.
@@ -87,10 +87,9 @@ pub fn arrow_scalar_to_value(col: &Arc<dyn Array>, row: usize) -> Value {
             .map(|a| Value::String(a.value(row).to_string()))
             .unwrap_or(Value::Null),
         _ => {
-            // Fallback: format as string via ScalarValue.
-            let scalar = datafusion::common::ScalarValue::try_from_array(col, row);
-            match scalar {
-                Ok(s) => Value::String(s.to_string()),
+            // Fallback: format as display string.
+            match arrow::util::display::array_value_to_string(col, row) {
+                Ok(s) => Value::String(s),
                 Err(_) => Value::Null,
             }
         }

nodedb/src/control/planner/procedural/executor/eval.rs

@@ -1,8 +1,11 @@
 //! Expression evaluation helpers for the statement executor.
 //!
-//! Evaluates SQL expressions via DataFusion and converts results to
-//! Rust types (bool, i64, nodedb_types::Value). Used by ASSIGN, IF/WHILE
-//! conditions, FOR bounds, and OUT parameter capture.
+//! Evaluates SQL expressions and converts results to Rust types (bool, i64,
+//! nodedb_types::Value). Used by ASSIGN, IF/WHILE conditions, FOR bounds,
+//! and OUT parameter capture.
+//!
+//! Uses DataFusion for complex expression evaluation (arithmetic, CASE, etc.).
+//! Literal fast paths avoid DataFusion for simple constants.
 
 use crate::control::state::SharedState;
 use crate::types::TenantId;
@@ -29,13 +32,13 @@ pub async fn evaluate_condition(
             let col = batch.column(0);
             if let Some(bool_arr) = col
                 .as_any()
-                .downcast_ref::<datafusion::arrow::array::BooleanArray>()
+                .downcast_ref::<arrow::array::BooleanArray>()
             {
                 return Ok(bool_arr.value(0));
             }
             if let Some(int_arr) = col
                 .as_any()
-                .downcast_ref::<datafusion::arrow::array::Int32Array>()
+                .downcast_ref::<arrow::array::Int32Array>()
             {
                 return Ok(int_arr.value(0) != 0);
             }
@@ -64,13 +67,13 @@ pub async fn evaluate_int(
             let col = batch.column(0);
             if let Some(arr) = col
                 .as_any()
-                .downcast_ref::<datafusion::arrow::array::Int64Array>()
+                .downcast_ref::<arrow::array::Int64Array>()
             {
                 return Ok(arr.value(0));
             }
             if let Some(arr) = col
                 .as_any()
-                .downcast_ref::<datafusion::arrow::array::Int32Array>()
+                .downcast_ref::<arrow::array::Int32Array>()
             {
                 return Ok(arr.value(0) as i64);
             }
@@ -128,13 +131,13 @@ pub async fn evaluate_to_value(
 /// Execute a SQL expression via DataFusion and return the result batches.
 ///
 /// Wraps the expression in `SELECT (<expr>) as __result` and collects.
-/// Uses a standalone DataFusion session (not QueryContext) for scalar evaluation.
+/// Uses a standalone DataFusion session for scalar evaluation only.
 pub async fn eval_sql_expr(
     _state: &SharedState,
     _tenant_id: TenantId,
     expr: &str,
     context: &str,
-) -> crate::Result<Vec<datafusion::arrow::record_batch::RecordBatch>> {
+) -> crate::Result<Vec<arrow::record_batch::RecordBatch>> {
     let session = datafusion::execution::context::SessionContext::new();
     let select_sql = format!("SELECT ({expr}) as __result");
     let df = session