fix(sql): enforce declared integer widths on write and keep them in sync across DDL

Narrowing a column's RowDescription OID to its declared width (issue
#217) is only honest if the stored value actually fits: nodedb keeps
every integer as a full i64, so nothing previously stopped a value
wider than SMALLINT/INTEGER from being written into a column that
advertises a 2- or 4-byte wire type, corrupting the value in transit.

Introduce IntWidth as the single source of truth mapping a declared
type string to its range, PostgreSQL OID, and type name, and route
every consumer through it: write-time range validation in the planner
(INSERT/UPSERT/UPDATE, including KV's separate insert path), a
defence-in-depth check on the read path for data that predates the
constraint, catalog introspection OIDs, and RowDescription narrowing
(replacing the old raw-string refinement). ALTER ADD/RENAME/DROP
COLUMN and ALTER COLUMN TYPE now keep the catalog's declared-type
record in sync so width fidelity survives schema changes instead of
silently reverting to BIGINT. Also fold the sign into negative integer
literals during AST conversion so i64::MIN parses exactly instead of
falling back to an approximate float.

Author: farhan-syah

nodedb-sql/src/error.rs

@@ -20,6 +20,21 @@ pub enum SqlError {
     #[error("type mismatch: {detail}")]
     TypeMismatch { detail: String },
 
+    /// A write supplied an integer wider than the column's declared type.
+    ///
+    /// nodedb stores every integer as an `i64`, so this is not a storage
+    /// limit — it is the constraint that makes the column's advertised wire
+    /// type honest. A column declared `INTEGER` reports OID 23, and a pgwire
+    /// client reading it in binary format decodes exactly four bytes; letting
+    /// a wider value in would mean either truncating it on read or lying about
+    /// the type. PostgreSQL rejects the same write with the same message.
+    #[error("value {value} is out of range for column '{column}' of type {declared_type}")]
+    IntegerOutOfRange {
+        column: String,
+        value: i64,
+        declared_type: &'static str,
+    },
+
     #[error("unsupported: {detail}")]
     Unsupported { detail: String },
 

nodedb-sql/src/planner/dml.rs

@@ -6,8 +6,8 @@ use nodedb_types::DatabaseId;
 use sqlparser::ast::{self};
 
 use super::dml_helpers::{
-    build_kv_insert_plan, build_vector_primary_insert_plan, convert_value_rows,
-    resolve_insert_columns,
+    build_kv_insert_plan, build_vector_primary_insert_plan, check_declared_int_ranges,
+    convert_value_rows, resolve_insert_columns,
 };
 use crate::engine_rules::{self, InsertParams};
 use crate::error::{Result, SqlError};
@@ -135,6 +135,7 @@ pub fn plan_insert(ins: &ast::Insert, catalog: &dyn SqlCatalog) -> Result<Vec<Sq
             intent,
             Vec::new(),
             info.primary_key.as_deref(),
+            &info.columns,
         );
     }
 
@@ -148,11 +149,13 @@ pub fn plan_insert(ins: &ast::Insert, catalog: &dyn SqlCatalog) -> Result<Vec<Sq
         && let Some(ref vpc) = info.vector_primary
     {
         let rows_parsed = convert_value_rows(&columns, rows_ast)?;
+        check_declared_int_ranges(&info.columns, &rows_parsed)?;
         return build_vector_primary_insert_plan(&table_name, vpc, &columns, rows_parsed);
     }
 
     // All other engines: delegate to engine rules.
     let rows = convert_value_rows(&columns, rows_ast)?;
+    check_declared_int_ranges(&info.columns, &rows)?;
     let column_defaults: Vec<(String, String)> = info
         .columns
         .iter()
@@ -218,6 +221,7 @@ pub fn plan_upsert(ins: &ast::Insert, catalog: &dyn SqlCatalog) -> Result<Vec<Sq
             KvInsertIntent::Put,
             Vec::new(),
             info.primary_key.as_deref(),
+            &info.columns,
         );
     }
 
@@ -226,6 +230,7 @@ pub fn plan_upsert(ins: &ast::Insert, catalog: &dyn SqlCatalog) -> Result<Vec<Sq
     let columns = resolve_insert_columns(columns, &info, rows_ast)?;
 
     let rows = convert_value_rows(&columns, rows_ast)?;
+    check_declared_int_ranges(&info.columns, &rows)?;
     let column_defaults: Vec<(String, String)> = info
         .columns
         .iter()
@@ -294,6 +299,7 @@ fn plan_upsert_with_on_conflict(
             KvInsertIntent::Put,
             on_conflict_updates,
             info.primary_key.as_deref(),
+            &info.columns,
         );
     }
 
@@ -302,6 +308,7 @@ fn plan_upsert_with_on_conflict(
     let columns = resolve_insert_columns(columns, &info, rows_ast)?;
 
     let rows = convert_value_rows(&columns, rows_ast)?;
+    check_declared_int_ranges(&info.columns, &rows)?;
     let column_defaults: Vec<(String, String)> = info
         .columns
         .iter()

nodedb-sql/src/planner/dml_helpers.rs

@@ -25,6 +25,85 @@ pub(super) fn convert_value_rows(
         .collect()
 }
 
+/// Reject any integer value that does not fit its column's declared width.
+///
+/// nodedb stores every integer as an `i64`, so this is not a storage limit.
+/// It is the constraint that makes the column's advertised wire type honest:
+/// a column declared `INTEGER` reports OID 23 in `RowDescription`, and a
+/// pgwire client reading it in binary format decodes exactly four bytes.
+/// Accepting a wider value would force a later choice between truncating it on
+/// read and lying about the column's type — so the value is refused at the
+/// point it enters, exactly as PostgreSQL refuses it.
+///
+/// This runs in the planner rather than in each engine because the declared
+/// width is engine-independent (the same `IntWidth` drives the wire type for
+/// schemaless, columnar, strict, and kv alike), and because parameters are
+/// bound into the AST before planning — so one check here covers both literal
+/// `VALUES` and `$1` placeholders, for every engine, on every DML path.
+///
+/// Non-integer values and columns with no declared width pass through: this
+/// checks range only, never type.
+pub(super) fn check_declared_int_ranges(
+    columns: &[ColumnInfo],
+    rows: &[Vec<(String, SqlValue)>],
+) -> Result<()> {
+    // Overwhelmingly the common case — skip the per-cell name lookup entirely
+    // when the collection declares no narrowed integer column.
+    if !columns.iter().any(|c| {
+        matches!(
+            c.int_width,
+            Some(nodedb_types::columnar::IntWidth::I16 | nodedb_types::columnar::IntWidth::I32)
+        )
+    }) {
+        return Ok(());
+    }
+
+    for row in rows {
+        for (name, value) in row {
+            let SqlValue::Int(v) = value else { continue };
+            let Some(width) = columns
+                .iter()
+                .find(|c| c.name.eq_ignore_ascii_case(name))
+                .and_then(|c| c.int_width)
+            else {
+                continue;
+            };
+            if !width.contains(*v) {
+                return Err(SqlError::IntegerOutOfRange {
+                    column: name.clone(),
+                    value: *v,
+                    declared_type: width.pg_type_name(),
+                });
+            }
+        }
+    }
+    Ok(())
+}
+
+/// [`check_declared_int_ranges`] for `UPDATE ... SET col = <literal>`.
+///
+/// Only literal assignments are checkable at plan time; a computed assignment
+/// (`SET n = n + 1`) has no value until the Data Plane evaluates it. Those are
+/// caught on the read path instead, where the encoder refuses to transmit a
+/// value that does not fit the column's advertised width — so an out-of-range
+/// value can never reach a client silently by either route.
+pub(super) fn check_declared_int_ranges_in_assignments(
+    columns: &[ColumnInfo],
+    assignments: &[(String, SqlExpr)],
+) -> Result<()> {
+    let literals: Vec<(String, SqlValue)> = assignments
+        .iter()
+        .filter_map(|(col, expr)| match expr {
+            SqlExpr::Literal(v @ SqlValue::Int(_)) => Some((col.clone(), v.clone())),
+            _ => None,
+        })
+        .collect();
+    if literals.is_empty() {
+        return Ok(());
+    }
+    check_declared_int_ranges(columns, std::slice::from_ref(&literals))
+}
+
 /// Resolve the effective column list for a `VALUES`-clause INSERT/UPSERT.
 ///
 /// A *positional* insert — `INSERT INTO t VALUES (...)` with no explicit
@@ -277,6 +356,7 @@ pub(super) fn build_kv_insert_plan(
     intent: KvInsertIntent,
     on_conflict_updates: Vec<(String, SqlExpr)>,
     pk_col: Option<&str>,
+    declared_columns: &[ColumnInfo],
 ) -> Result<Vec<SqlPlan>> {
     // Positional KV insert (no column list): the key/value split below is
     // driven entirely by matching column *names* against `key_col_name`/
@@ -288,6 +368,10 @@ pub(super) fn build_kv_insert_plan(
             collection: table_name,
         });
     }
+    // KV returns early from every INSERT/UPSERT entry point, so the declared
+    // width check lives here rather than at the call sites — otherwise a
+    // fourth KV entry point could be added without it.
+    check_declared_int_ranges(declared_columns, &convert_value_rows(columns, rows_ast)?)?;
     let key_col_name = pk_col.unwrap_or("key");
     let key_idx = columns.iter().position(|c| c == key_col_name);
     let ttl_idx = columns.iter().position(|c| c == "ttl");

nodedb-sql/src/planner/dml_update_delete.rs

@@ -8,7 +8,10 @@ use sqlparser::ast;
 use super::super::ast_helpers::{
     flatten_and_expr, qualified_ident_pair, strip_and_convert_filters,
 };
-use super::super::dml_helpers::{extract_point_keys, extract_table_name_from_table_with_joins};
+use super::super::dml_helpers::{
+    check_declared_int_ranges_in_assignments, extract_point_keys,
+    extract_table_name_from_table_with_joins,
+};
 use crate::engine_rules::{self, DeleteParams, UpdateFromParams, UpdateParams};
 use crate::error::{Result, SqlError};
 use crate::parser::normalize::{
@@ -41,6 +44,7 @@ pub fn plan_update(stmt: &ast::Statement, catalog: &dyn SqlCatalog) -> Result<Ve
         })?;
 
     let assigns = convert_assignments(&update.assignments)?;
+    check_declared_int_ranges_in_assignments(&info.columns, &assigns)?;
 
     let filters = match &update.selection {
         Some(expr) => super::super::select::convert_where_to_filters(expr)?,

nodedb-sql/src/resolver/columns.rs

@@ -285,6 +285,7 @@ fn array_columns(view: &ArrayCatalogView) -> Vec<ColumnInfo> {
             is_primary_key: false,
             default: None,
             raw_type: None,
+            int_width: None,
         });
     }
     for a in &view.attrs {
@@ -295,6 +296,7 @@ fn array_columns(view: &ArrayCatalogView) -> Vec<ColumnInfo> {
             is_primary_key: false,
             default: None,
             raw_type: None,
+            int_width: None,
         });
     }
     cols
@@ -356,6 +358,7 @@ mod tests {
                     is_primary_key: false,
                     default: None,
                     raw_type: None,
+                    int_width: None,
                 })
                 .collect(),
             primary_key: None,

nodedb-sql/src/resolver/expr/convert.rs

@@ -2,7 +2,7 @@
 
 //! Convert sqlparser AST expressions to our SqlExpr IR.
 
-use sqlparser::ast::{self, Expr, Value};
+use sqlparser::ast::{self, Expr, UnaryOperator, Value};
 
 use crate::error::{Result, SqlError};
 use crate::parser::normalize::{SCHEMA_QUALIFIED_MSG, normalize_ident};
@@ -130,6 +130,36 @@ fn convert_expr_inner(expr: &Expr, depth: &mut usize) -> Result<SqlExpr> {
                 right: Box::new(convert_expr_depth(right, depth)?),
             })
         }
+        // A negative integer literal reaches sqlparser as unary minus applied
+        // to a *positive* number, so the most negative `BIGINT` arrives as
+        // `-(9223372036854775808)` — and that operand does not fit an `i64`.
+        // Converting the operand on its own therefore falls back to `Float`
+        // and silently turns an exact integer into an approximate one. Folding
+        // the sign into the literal before parsing keeps the whole `i64` range
+        // exact; anything that still does not fit falls through to the general
+        // path below and is handled as before.
+        Expr::UnaryOp {
+            op: UnaryOperator::Minus,
+            expr: inner,
+        } if matches!(
+            inner.as_ref(),
+            Expr::Value(v) if matches!(&v.value, Value::Number(..))
+        ) =>
+        {
+            let Expr::Value(v) = inner.as_ref() else {
+                unreachable!("guarded by the `matches!` above")
+            };
+            let Value::Number(n, _) = &v.value else {
+                unreachable!("guarded by the `matches!` above")
+            };
+            match format!("-{n}").parse::<i64>() {
+                Ok(i) => Ok(SqlExpr::Literal(SqlValue::Int(i))),
+                Err(_) => Ok(SqlExpr::UnaryOp {
+                    op: UnaryOp::Neg,
+                    expr: Box::new(convert_expr_depth(inner, depth)?),
+                }),
+            }
+        }
         Expr::UnaryOp { op, expr } => Ok(SqlExpr::UnaryOp {
             op: convert_unary_op(op)?,
             expr: Box::new(convert_expr_depth(expr, depth)?),

nodedb-sql/src/types/collection.rs

@@ -76,4 +76,14 @@ pub struct ColumnInfo {
     /// Columnar INSERT converters use this to reconstruct the exact `ColumnType`
     /// so JSON / Geometry / UUID columns are not incorrectly inferred as String.
     pub raw_type: Option<String>,
+    /// Declared width of an integer column, resolved once from the catalog at
+    /// adapter-construction time. `None` for non-integer columns and for
+    /// integer columns whose declared type carried no width.
+    ///
+    /// This is deliberately a resolved [`IntWidth`] rather than another raw
+    /// string: it is read on both the write path (range validation) and the
+    /// read path (`RowDescription` OID and binary payload width), and those
+    /// two must agree exactly. Resolving once at the catalog boundary is what
+    /// makes disagreement unrepresentable.
+    pub int_width: Option<nodedb_types::columnar::IntWidth>,
 }

nodedb-sql/tests/positional_insert_column_binding.rs

@@ -40,6 +40,7 @@ impl SqlCatalog for Catalog {
                         is_primary_key: true,
                         default: None,
                         raw_type: Some("INT".into()),
+                        int_width: Some(nodedb_types::columnar::IntWidth::I32),
                     },
                     ColumnInfo {
                         name: "note".into(),
@@ -48,6 +49,7 @@ impl SqlCatalog for Catalog {
                         is_primary_key: false,
                         default: None,
                         raw_type: Some("TEXT".into()),
+                        int_width: None,
                     },
                 ],
                 primary_key: Some("id".into()),

nodedb-types/src/columnar/column_parse.rs

@@ -126,8 +126,9 @@ impl FromStr for ColumnType {
             // them as unknown types even though they're valid aliases). They
             // all collapse to the same `Int64` storage variant as
             // `BIGINT`/`INTEGER`/`INT` — nodedb always stores integers as a
-            // full i64; the declared width only narrows the advertised wire
-            // OID (see `response_shape::schema::sql_data_type_to_ddl_col_type_with_raw`).
+            // full i64. The declared width is carried separately as an
+            // [`super::IntWidth`], which is what bounds writes and narrows the
+            // advertised wire OID; it is deliberately not a storage variant.
             "BIGINT" | "INT64" | "INTEGER" | "INT" | "INT4" | "INT8" | "SMALLINT" | "INT2" => {
                 Ok(Self::Int64)
             }