refactor(nodedb): split value.rs, alter.rs, and point.rs into submodules

value.rs (509 lines) → value/ with assignments, convert, defaults,
msgpack_write, rows, and time_range submodules.

alter.rs (441 lines) → alter/ with add_column, alter_type, drop_column,
enforcement, materialized_sum, and rename_column submodules. DDL router
schema.rs gains explicit routing for ADD COLUMN, DROP COLUMN, and
RENAME COLUMN so requests reach the handlers in the new module layout.

point.rs (696 lines) → point/ with apply_put, delete, get, put, and
update submodules.

All files remain under 200 lines. No behaviour changes.

Author: farhan-syah

nodedb/src/control/planner/sql_plan_convert/value/assignments.rs

@@ -0,0 +1,28 @@
+//! UPDATE assignment serialization: `(field, SqlExpr)` pairs → wire-ready
+//! `UpdateValue` payloads.
+//!
+//! Literal RHS is pre-encoded as msgpack. Non-literal RHS (arithmetic,
+//! functions, CASE, concatenation, ...) is converted to the shared evaluator
+//! type `bridge::expr_eval::SqlExpr` and shipped to the Data Plane, where
+//! it is evaluated against the current row at apply time.
+
+use nodedb_sql::types::SqlExpr;
+
+use crate::bridge::physical_plan::UpdateValue;
+
+use super::super::expr::sql_expr_to_bridge_expr;
+use super::convert::sql_value_to_msgpack;
+
+pub(crate) fn assignments_to_update_values(
+    assignments: &[(String, SqlExpr)],
+) -> crate::Result<Vec<(String, UpdateValue)>> {
+    let mut result = Vec::with_capacity(assignments.len());
+    for (field, expr) in assignments {
+        let value = match expr {
+            SqlExpr::Literal(v) => UpdateValue::Literal(sql_value_to_msgpack(v)),
+            _ => UpdateValue::Expr(sql_expr_to_bridge_expr(expr)),
+        };
+        result.push((field.clone(), value));
+    }
+    Ok(result)
+}

nodedb/src/control/planner/sql_plan_convert/value/convert.rs

@@ -0,0 +1,45 @@
+//! Conversions from `nodedb_sql::types::SqlValue` into runtime / wire forms.
+
+use nodedb_sql::types::SqlValue;
+
+use super::msgpack_write::write_msgpack_value;
+
+pub(crate) fn sql_value_to_nodedb_value(v: &SqlValue) -> nodedb_types::Value {
+    match v {
+        SqlValue::Int(i) => nodedb_types::Value::Integer(*i),
+        SqlValue::Float(f) => nodedb_types::Value::Float(*f),
+        SqlValue::String(s) => nodedb_types::Value::String(s.clone()),
+        SqlValue::Bool(b) => nodedb_types::Value::Bool(*b),
+        SqlValue::Null => nodedb_types::Value::Null,
+        SqlValue::Array(arr) => {
+            nodedb_types::Value::Array(arr.iter().map(sql_value_to_nodedb_value).collect())
+        }
+        SqlValue::Bytes(b) => nodedb_types::Value::Bytes(b.clone()),
+    }
+}
+
+pub(crate) fn sql_value_to_string(v: &SqlValue) -> String {
+    match v {
+        SqlValue::String(s) => s.clone(),
+        SqlValue::Int(i) => i.to_string(),
+        SqlValue::Float(f) => f.to_string(),
+        SqlValue::Bool(b) => b.to_string(),
+        _ => String::new(),
+    }
+}
+
+pub(crate) fn sql_value_to_bytes(v: &SqlValue) -> Vec<u8> {
+    match v {
+        SqlValue::String(s) => s.as_bytes().to_vec(),
+        SqlValue::Bytes(b) => b.clone(),
+        SqlValue::Int(i) => i.to_string().as_bytes().to_vec(),
+        _ => sql_value_to_string(v).into_bytes(),
+    }
+}
+
+/// Encode a SQL value as standard msgpack for field-level updates.
+pub(crate) fn sql_value_to_msgpack(v: &SqlValue) -> Vec<u8> {
+    let mut buf = Vec::with_capacity(16);
+    write_msgpack_value(&mut buf, v);
+    buf
+}

nodedb/src/control/planner/sql_plan_convert/value/defaults.rs

@@ -0,0 +1,71 @@
+//! Column DEFAULT expression evaluation at insert time.
+//!
+//! Supports ID generation functions (UUIDv4/v7, ULID, CUID2, NANOID), `NOW()`,
+//! and literal values. More complex defaults (arbitrary expressions) go
+//! through the shared SqlExpr evaluator path.
+
+pub(crate) fn evaluate_default_expr(expr: &str) -> Option<nodedb_types::Value> {
+    let upper = expr.trim().to_uppercase();
+    match upper.as_str() {
+        "UUID_V7" | "UUIDV7" | "GEN_UUID_V7()" | "UUID_V7()" => {
+            Some(nodedb_types::Value::String(nodedb_types::id_gen::uuid_v7()))
+        }
+        "UUID_V4" | "UUIDV4" | "UUID" | "GEN_UUID_V4()" | "UUID_V4()" => {
+            Some(nodedb_types::Value::String(nodedb_types::id_gen::uuid_v4()))
+        }
+        "ULID" | "GEN_ULID()" | "ULID()" => {
+            Some(nodedb_types::Value::String(nodedb_types::id_gen::ulid()))
+        }
+        "CUID2" | "CUID2()" => Some(nodedb_types::Value::String(nodedb_types::id_gen::cuid2())),
+        "NANOID" | "NANOID()" => Some(nodedb_types::Value::String(nodedb_types::id_gen::nanoid())),
+        "NOW()" => {
+            let now = std::time::SystemTime::now()
+                .duration_since(std::time::UNIX_EPOCH)
+                .unwrap_or_default();
+            Some(nodedb_types::Value::String(
+                chrono::DateTime::from_timestamp_millis(now.as_millis() as i64)
+                    .map(|dt| dt.to_rfc3339())
+                    .unwrap_or_else(|| now.as_millis().to_string()),
+            ))
+        }
+        _ => parse_parametric_or_literal(expr, &upper),
+    }
+}
+
+fn parse_parametric_or_literal(expr: &str, upper: &str) -> Option<nodedb_types::Value> {
+    // NANOID(N) — custom length.
+    if upper.starts_with("NANOID(") && upper.ends_with(')') {
+        let len_str = &upper[7..upper.len() - 1];
+        if let Ok(len) = len_str.parse::<usize>() {
+            return Some(nodedb_types::Value::String(
+                nodedb_types::id_gen::nanoid_with_length(len),
+            ));
+        }
+    }
+    // CUID2(N) — custom length.
+    if upper.starts_with("CUID2(") && upper.ends_with(')') {
+        let len_str = &upper[6..upper.len() - 1];
+        if let Ok(len) = len_str.parse::<usize>() {
+            return Some(nodedb_types::Value::String(
+                nodedb_types::id_gen::cuid2_with_length(len),
+            ));
+        }
+    }
+    // Numeric literal.
+    if let Ok(i) = expr.trim().parse::<i64>() {
+        return Some(nodedb_types::Value::Integer(i));
+    }
+    if let Ok(f) = expr.trim().parse::<f64>() {
+        return Some(nodedb_types::Value::Float(f));
+    }
+    // Quoted string literal.
+    let trimmed = expr.trim();
+    if (trimmed.starts_with('\'') && trimmed.ends_with('\''))
+        || (trimmed.starts_with('"') && trimmed.ends_with('"'))
+    {
+        return Some(nodedb_types::Value::String(
+            trimmed[1..trimmed.len() - 1].to_string(),
+        ));
+    }
+    None
+}

nodedb/src/control/planner/sql_plan_convert/value/mod.rs

@@ -0,0 +1,21 @@
+//! Value conversion utilities: SqlValue ↔ nodedb_types::Value, msgpack encoding,
+//! column default evaluation, and WHERE-clause time-range extraction.
+
+pub(super) mod assignments;
+pub(super) mod convert;
+pub(super) mod defaults;
+pub(super) mod msgpack_write;
+pub(super) mod rows;
+pub(super) mod time_range;
+
+pub(super) use assignments::assignments_to_update_values;
+pub(super) use convert::{
+    sql_value_to_bytes, sql_value_to_msgpack, sql_value_to_nodedb_value, sql_value_to_string,
+};
+pub(super) use defaults::evaluate_default_expr;
+pub(super) use msgpack_write::{
+    row_to_msgpack, write_msgpack_array_header, write_msgpack_map_header, write_msgpack_str,
+    write_msgpack_value,
+};
+pub(super) use rows::rows_to_msgpack_array;
+pub(super) use time_range::extract_time_range;

nodedb/src/control/planner/sql_plan_convert/value/msgpack_write.rs

@@ -0,0 +1,152 @@
+//! Standard-msgpack writers for `SqlValue`.
+//!
+//! These are the *only* msgpack producers used by the DML path — no JSON or
+//! zerompk intermediary. Format matches the on-wire layout read by
+//! `json_from_msgpack` and the Data Plane row decoders.
+
+use nodedb_sql::types::SqlValue;
+
+pub(crate) fn row_to_msgpack(row: &[(String, SqlValue)]) -> crate::Result<Vec<u8>> {
+    let mut buf = Vec::with_capacity(row.len() * 32);
+    write_msgpack_map_header(&mut buf, row.len());
+    for (key, val) in row {
+        write_msgpack_str(&mut buf, key);
+        write_msgpack_value(&mut buf, val);
+    }
+    Ok(buf)
+}
+
+pub(crate) fn write_msgpack_map_header(buf: &mut Vec<u8>, len: usize) {
+    if len < 16 {
+        buf.push(0x80 | len as u8);
+    } else if len <= u16::MAX as usize {
+        buf.push(0xDE);
+        buf.extend_from_slice(&(len as u16).to_be_bytes());
+    } else {
+        buf.push(0xDF);
+        buf.extend_from_slice(&(len as u32).to_be_bytes());
+    }
+}
+
+pub(crate) fn write_msgpack_array_header(buf: &mut Vec<u8>, len: usize) {
+    if len < 16 {
+        buf.push(0x90 | len as u8);
+    } else if len <= u16::MAX as usize {
+        buf.push(0xDC);
+        buf.extend_from_slice(&(len as u16).to_be_bytes());
+    } else {
+        buf.push(0xDD);
+        buf.extend_from_slice(&(len as u32).to_be_bytes());
+    }
+}
+
+pub(crate) fn write_msgpack_str(buf: &mut Vec<u8>, s: &str) {
+    let bytes = s.as_bytes();
+    let len = bytes.len();
+    if len < 32 {
+        buf.push(0xA0 | len as u8);
+    } else if len <= u8::MAX as usize {
+        buf.push(0xD9);
+        buf.push(len as u8);
+    } else if len <= u16::MAX as usize {
+        buf.push(0xDA);
+        buf.extend_from_slice(&(len as u16).to_be_bytes());
+    } else {
+        buf.push(0xDB);
+        buf.extend_from_slice(&(len as u32).to_be_bytes());
+    }
+    buf.extend_from_slice(bytes);
+}
+
+pub(crate) fn write_msgpack_value(buf: &mut Vec<u8>, val: &SqlValue) {
+    match val {
+        SqlValue::Null => buf.push(0xC0),
+        SqlValue::Bool(true) => buf.push(0xC3),
+        SqlValue::Bool(false) => buf.push(0xC2),
+        SqlValue::Int(i) => write_msgpack_int(buf, *i),
+        SqlValue::Float(f) => {
+            buf.push(0xCB);
+            buf.extend_from_slice(&f.to_be_bytes());
+        }
+        SqlValue::String(s) => write_msgpack_str(buf, s),
+        SqlValue::Array(arr) => {
+            write_msgpack_array_header(buf, arr.len());
+            for item in arr {
+                write_msgpack_value(buf, item);
+            }
+        }
+        SqlValue::Bytes(b) => write_msgpack_bin(buf, b),
+    }
+}
+
+fn write_msgpack_int(buf: &mut Vec<u8>, i: i64) {
+    if (0..=127).contains(&i) {
+        buf.push(i as u8);
+    } else if (-32..0).contains(&i) {
+        buf.push(i as u8); // negative fixint
+    } else if i >= i8::MIN as i64 && i <= i8::MAX as i64 {
+        buf.push(0xD0);
+        buf.push(i as i8 as u8);
+    } else if i >= i16::MIN as i64 && i <= i16::MAX as i64 {
+        buf.push(0xD1);
+        buf.extend_from_slice(&(i as i16).to_be_bytes());
+    } else if i >= i32::MIN as i64 && i <= i32::MAX as i64 {
+        buf.push(0xD2);
+        buf.extend_from_slice(&(i as i32).to_be_bytes());
+    } else {
+        buf.push(0xD3);
+        buf.extend_from_slice(&i.to_be_bytes());
+    }
+}
+
+fn write_msgpack_bin(buf: &mut Vec<u8>, b: &[u8]) {
+    let len = b.len();
+    if len <= u8::MAX as usize {
+        buf.push(0xC4);
+        buf.push(len as u8);
+    } else if len <= u16::MAX as usize {
+        buf.push(0xC5);
+        buf.extend_from_slice(&(len as u16).to_be_bytes());
+    } else {
+        buf.push(0xC6);
+        buf.extend_from_slice(&(len as u32).to_be_bytes());
+    }
+    buf.extend_from_slice(b);
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn row_to_msgpack_produces_standard_format() {
+        let row = vec![
+            ("count".to_string(), SqlValue::Int(1)),
+            ("label".to_string(), SqlValue::String("homepage".into())),
+        ];
+        let bytes = row_to_msgpack(&row).unwrap();
+        assert_eq!(
+            bytes[0], 0x82,
+            "expected fixmap(2), got 0x{:02X}. bytes={bytes:?}",
+            bytes[0]
+        );
+        let json = nodedb_types::json_from_msgpack(&bytes).unwrap();
+        let obj = json.as_object().unwrap();
+        assert_eq!(obj["count"], 1);
+        assert_eq!(obj["label"], "homepage");
+    }
+
+    #[test]
+    fn write_msgpack_value_int() {
+        let mut buf = Vec::new();
+        write_msgpack_value(&mut buf, &SqlValue::Int(42));
+        assert_eq!(buf, vec![42]);
+    }
+
+    #[test]
+    fn write_msgpack_value_string() {
+        let mut buf = Vec::new();
+        write_msgpack_value(&mut buf, &SqlValue::String("hi".into()));
+        assert_eq!(buf, vec![0xA2, b'h', b'i']);
+    }
+}

nodedb/src/control/planner/sql_plan_convert/value/rows.rs

@@ -0,0 +1,34 @@
+//! Batch-of-rows msgpack encoding for columnar INSERT paths.
+
+use nodedb_sql::types::SqlValue;
+
+use super::convert::sql_value_to_nodedb_value;
+use super::defaults::evaluate_default_expr;
+
+pub(crate) fn rows_to_msgpack_array(
+    rows: &[&Vec<(String, SqlValue)>],
+    column_defaults: &[(String, String)],
+) -> crate::Result<Vec<u8>> {
+    let arr: Vec<nodedb_types::Value> = rows
+        .iter()
+        .map(|row| {
+            let mut map = std::collections::HashMap::new();
+            for (key, val) in row.iter() {
+                map.insert(key.clone(), sql_value_to_nodedb_value(val));
+            }
+            for (col_name, default_expr) in column_defaults {
+                if !map.contains_key(col_name)
+                    && let Some(val) = evaluate_default_expr(default_expr)
+                {
+                    map.insert(col_name.clone(), val);
+                }
+            }
+            nodedb_types::Value::Object(map)
+        })
+        .collect();
+    let val = nodedb_types::Value::Array(arr);
+    nodedb_types::value_to_msgpack(&val).map_err(|e| crate::Error::Serialization {
+        format: "msgpack".into(),
+        detail: format!("columnar row batch: {e}"),
+    })
+}