Evaluate volatile column DEFAULTs per row instead of freezing them into cached plans

nodedb-cluster-tests/tests/common_suite/cases/gateway_execute.rs

@@ -20,7 +20,7 @@ use nodedb::control::gateway::core::QueryContext;
 use nodedb::control::gateway::plan_cache::PlanCacheKey;
 use nodedb::control::gateway::plan_cache::{hash_placeholder_types, hash_sql};
 use nodedb::control::gateway::version_set::GatewayVersionSet;
-use nodedb::control::gateway::{Gateway, PlanCache};
+use nodedb::control::gateway::{Gateway, LoweredPlan, PlanCache};
 use nodedb::types::TenantId;
 use nodedb_physical::physical_plan::{KvOp, PhysicalPlan};
 use nodedb_types::QualifiedCollection;
@@ -135,15 +135,15 @@ async fn gateway_execute_sql_plan_cache_populated() {
 
     let sql = "GET gw_cache_smoke smoke-key";
     let make_plan = || {
-        Ok(PhysicalPlan::Kv(KvOp::Get {
+        Ok(LoweredPlan::cacheable(PhysicalPlan::Kv(KvOp::Get {
             collection: QualifiedCollection::new(
                 nodedb_types::id::DatabaseId::DEFAULT,
                 "gw_cache_smoke",
             ),
             key: b"smoke-key".to_vec(),
             rls_filters: vec![],
             surrogate_ceiling: None,
-        }))
+        })))
     };
 
     // Cache starts empty.

nodedb-cluster-tests/tests/common_suite/cases/http_gateway_migration.rs

@@ -17,9 +17,9 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use nodedb::Error;
-use nodedb::control::gateway::Gateway;
 use nodedb::control::gateway::GatewayErrorMap;
 use nodedb::control::gateway::core::QueryContext;
+use nodedb::control::gateway::{Gateway, LoweredPlan};
 use nodedb::types::TenantId;
 use nodedb_physical::physical_plan::{KvOp, PhysicalPlan};
 use nodedb_types::QualifiedCollection;
@@ -184,15 +184,15 @@ async fn http_gateway_migration_cross_node_query() {
                 get_sql,
                 &[],
                 || {
-                    Ok(PhysicalPlan::Kv(KvOp::Get {
+                    Ok(LoweredPlan::cacheable(PhysicalPlan::Kv(KvOp::Get {
                         collection: QualifiedCollection::new(
                             nodedb_types::id::DatabaseId::DEFAULT,
                             "http_gw_cross_node",
                         ),
                         key: b"cross-key".to_vec(),
                         rls_filters: vec![],
                         surrogate_ceiling: None,
-                    }))
+                    })))
                 },
                 |plan| async {
                     Ok(common::authorize_gateway_plan(&follower.shared, &ctx, plan).await)

nodedb-cluster-tests/tests/common_suite/cases/pgwire_gateway_migration.rs

@@ -20,9 +20,9 @@ use crate::common;
 use std::sync::Arc;
 use std::time::Duration;
 
-use nodedb::control::gateway::Gateway;
 use nodedb::control::gateway::core::QueryContext;
 use nodedb::control::gateway::version_set::GatewayVersionSet;
+use nodedb::control::gateway::{Gateway, LoweredPlan};
 use nodedb::types::TenantId;
 use nodedb_physical::physical_plan::{KvOp, PhysicalPlan};
 use nodedb_types::QualifiedCollection;
@@ -200,15 +200,15 @@ async fn pgwire_gateway_migration_plan_cache_hits() {
 
         let sql = "GET pgwire_gw_cache cache-key";
         let make_plan = || {
-            Ok(PhysicalPlan::Kv(KvOp::Get {
+            Ok(LoweredPlan::cacheable(PhysicalPlan::Kv(KvOp::Get {
                 collection: QualifiedCollection::new(
                     nodedb_types::id::DatabaseId::DEFAULT,
                     "pgwire_gw_cache",
                 ),
                 key: b"cache-key".to_vec(),
                 rls_filters: vec![],
                 surrogate_ceiling: None,
-            }))
+            })))
         };
         let authorize_plan = |plan: PhysicalPlan| async {
             Ok(common::authorize_gateway_plan(&node.shared, &ctx, plan).await)

nodedb-sql/src/catalog.rs

@@ -192,6 +192,54 @@ pub trait SqlCatalog {
     fn resolve_regtype(&self, _name: &str) -> Option<i64> {
         None
     }
+
+    /// Advance a sequence and return the allocated value, recording it as this
+    /// session's `currval`. Scoping mirrors `resolve_regclass`.
+    ///
+    /// An unknown name returns [`SqlError::UndefinedObject`] (SQLSTATE
+    /// `42704`). The default returns "sequence access unavailable" so
+    /// implementors with no sequence state compile without change.
+    fn sequence_nextval(
+        &self,
+        _database_id: nodedb_types::DatabaseId,
+        _tenant_id: u64,
+        _name: &str,
+    ) -> Result<i64, crate::SqlError> {
+        Err(sequence_access_unavailable())
+    }
+
+    /// Return the last value this session obtained from `sequence_nextval`.
+    ///
+    /// A sequence this session never advanced returns
+    /// [`SqlError::ObjectNotInPrerequisiteState`] (SQLSTATE `55000`).
+    fn sequence_currval(
+        &self,
+        _database_id: nodedb_types::DatabaseId,
+        _tenant_id: u64,
+        _name: &str,
+    ) -> Result<i64, crate::SqlError> {
+        Err(sequence_access_unavailable())
+    }
+
+    /// Position a sequence so the next `sequence_nextval` returns
+    /// `value + increment`. Returns `value`.
+    fn sequence_setval(
+        &self,
+        _database_id: nodedb_types::DatabaseId,
+        _tenant_id: u64,
+        _name: &str,
+        _value: i64,
+    ) -> Result<i64, crate::SqlError> {
+        Err(sequence_access_unavailable())
+    }
+}
+
+/// The error a catalog with no sequence state returns for every accessor.
+fn sequence_access_unavailable() -> crate::SqlError {
+    crate::SqlError::ObjectNotInPrerequisiteState {
+        object: "sequence".into(),
+        detail: "sequence access unavailable: this catalog carries no sequence registry".into(),
+    }
 }
 
 /// View of a registered array, surfaced to the SQL planner. Decoded by

nodedb-sql/src/engine_rules/columnar.rs

@@ -13,6 +13,7 @@ impl EngineRules for ColumnarRules {
         Ok(vec![SqlPlan::Insert {
             collection: p.collection,
             engine: EngineType::Columnar,
+            route: WriteRoute::ColumnarFamily,
             rows: p.rows,
             column_defaults: p.column_defaults,
             if_absent: p.if_absent,
@@ -29,6 +30,7 @@ impl EngineRules for ColumnarRules {
         Ok(vec![SqlPlan::Upsert {
             collection: p.collection,
             engine: EngineType::Columnar,
+            route: WriteRoute::ColumnarFamily,
             rows: p.rows,
             column_defaults: p.column_defaults,
             on_conflict_updates: p.on_conflict_updates,

nodedb-sql/src/engine_rules/document_schemaless.rs

@@ -13,6 +13,7 @@ impl EngineRules for SchemalessRules {
         Ok(vec![SqlPlan::Insert {
             collection: p.collection,
             engine: EngineType::DocumentSchemaless,
+            route: WriteRoute::Document,
             rows: p.rows,
             column_defaults: p.column_defaults,
             if_absent: p.if_absent,
@@ -25,6 +26,7 @@ impl EngineRules for SchemalessRules {
         Ok(vec![SqlPlan::Upsert {
             collection: p.collection,
             engine: EngineType::DocumentSchemaless,
+            route: WriteRoute::Document,
             rows: p.rows,
             column_defaults: p.column_defaults,
             on_conflict_updates: p.on_conflict_updates,

nodedb-sql/src/engine_rules/document_strict.rs

@@ -13,6 +13,7 @@ impl EngineRules for StrictRules {
         Ok(vec![SqlPlan::Insert {
             collection: p.collection,
             engine: EngineType::DocumentStrict,
+            route: WriteRoute::Document,
             rows: p.rows,
             column_defaults: p.column_defaults,
             if_absent: p.if_absent,
@@ -25,6 +26,7 @@ impl EngineRules for StrictRules {
         Ok(vec![SqlPlan::Upsert {
             collection: p.collection,
             engine: EngineType::DocumentStrict,
+            route: WriteRoute::Document,
             rows: p.rows,
             column_defaults: p.column_defaults,
             on_conflict_updates: p.on_conflict_updates,

nodedb-sql/src/engine_rules/spatial.rs

@@ -13,6 +13,7 @@ impl EngineRules for SpatialRules {
         Ok(vec![SqlPlan::Insert {
             collection: p.collection,
             engine: EngineType::Spatial,
+            route: WriteRoute::ColumnarFamily,
             rows: p.rows,
             column_defaults: p.column_defaults,
             if_absent: p.if_absent,
@@ -28,6 +29,7 @@ impl EngineRules for SpatialRules {
         Ok(vec![SqlPlan::Upsert {
             collection: p.collection,
             engine: EngineType::Spatial,
+            route: WriteRoute::ColumnarFamily,
             rows: p.rows,
             column_defaults: p.column_defaults,
             on_conflict_updates: p.on_conflict_updates,

nodedb-sql/src/error.rs

@@ -19,6 +19,35 @@ pub enum SqlError {
     #[error("function {name}(...) does not exist")]
     UndefinedFunction { name: String },
 
+    /// A sequence accessor appeared where every output row needs its own
+    /// allocation, such as a SELECT list over a FROM clause.
+    ///
+    /// Rendered as SQLSTATE `0A000` (feature_not_supported). Constant
+    /// contexts evaluate the call for real: a FROM-less `SELECT`, a column
+    /// `DEFAULT`, a `VALUES` list. The refusal keeps a per-row call from
+    /// reaching the row evaluator, which has no sequence state and would
+    /// return `NULL` for every row.
+    #[error(
+        "{name}(...) is not supported in a per-row context; \
+         a SELECT list, WHERE clause, or SET clause over a FROM relation \
+         evaluates once per row. Call it in a FROM-less SELECT or a column \
+         DEFAULT instead"
+    )]
+    SequencePerRowUnsupported { name: String },
+
+    /// A statement names a database object that does not exist — a sequence,
+    /// most commonly. Distinct from [`SqlError::UndefinedFunction`]: the
+    /// function exists, the object it names does not. PostgreSQL rejects the
+    /// same input with SQLSTATE `42704` (`undefined_object`).
+    #[error("{kind} \"{name}\" does not exist")]
+    UndefinedObject { kind: &'static str, name: String },
+
+    /// An object exists but a prerequisite step has not run, such as `currval`
+    /// before this session called `nextval`. PostgreSQL rejects the same input
+    /// with SQLSTATE `55000` (`object_not_in_prerequisite_state`).
+    #[error("{detail}")]
+    ObjectNotInPrerequisiteState { object: String, detail: String },
+
     #[error("unknown column '{column}' in table '{table}'")]
     UnknownColumn { table: String, column: String },
 
@@ -85,6 +114,25 @@ pub enum SqlError {
     #[error("unsupported: {detail}")]
     Unsupported { detail: String },
 
+    /// A declared column DEFAULT the server cannot evaluate to a value.
+    ///
+    /// The column is never omitted instead. A DEFAULT that disappears stores
+    /// NULL where the declaration promised a value, and nothing reports it.
+    #[error("DEFAULT for column '{column}' cannot be evaluated: {expr}")]
+    UnevaluableDefault { column: String, expr: String },
+
+    /// `setval` appeared inside a column DEFAULT.
+    ///
+    /// A DEFAULT runs once per row, so evaluating `setval` there will move the
+    /// sequence's position on every inserted row. PostgreSQL reports a
+    /// function used in a context that forbids it as SQLSTATE `42601`
+    /// (`syntax_error`), and this refusal carries the same code.
+    #[error(
+        "setval() is not allowed in the DEFAULT for column '{column}'; \
+         a DEFAULT must not move a sequence's position"
+    )]
+    SetvalInColumnDefault { column: String },
+
     #[error("invalid function call: {detail}")]
     InvalidFunction { detail: String },
 

nodedb-sql/src/functions/arg_types.rs

@@ -391,3 +391,11 @@ pub static JSON_CONTAINS_ARGS: &[ArgTypeSpec] = &[any("container"), any("needle"
 
 /// `json_merge(base, overlay)` / `json_patch(base, overlay)`.
 pub static JSON_MERGE_ARGS: &[ArgTypeSpec] = &[any("base"), any("overlay")];
+
+// ── Sequence accessors ───────────────────────────────────────────────────────
+
+/// `nextval(name)` / `currval(name)` — one sequence name.
+pub static SEQUENCE_NAME_ARGS: &[ArgTypeSpec] = &[typed("sequence", TEXT)];
+
+/// `setval(name, value)` — a sequence name and the value it must take.
+pub static SETVAL_ARGS: &[ArgTypeSpec] = &[typed("sequence", TEXT), typed("value", INT64_ONLY)];

nodedb-sql/src/functions/builtins/helpers.rs

@@ -30,5 +30,6 @@ pub(super) fn m(
         return_type,
         arg_types,
         since: V0_1_0,
+        volatility: nodedb_types::Volatility::Immutable,
     }
 }

nodedb-sql/src/functions/builtins/scalars.rs

@@ -12,6 +12,7 @@ mod math;
 mod misc;
 mod pg_fts;
 mod pg_json;
+mod sequence_fn;
 mod spatial;
 mod string;
 mod vector;
@@ -32,6 +33,7 @@ pub(super) fn scalar_functions() -> Vec<FunctionMeta> {
     fns.extend(array_fn::array_fn_functions());
     fns.extend(array_elem::array_elem_functions());
     fns.extend(id_fn::id_fn_functions());
+    fns.extend(sequence_fn::sequence_fn_functions());
     fns.extend(misc::misc_functions());
     fns
 }

nodedb-sql/src/functions/builtins/scalars/datetime.rs

@@ -44,7 +44,8 @@ pub(super) fn datetime_functions() -> Vec<FunctionMeta> {
             no_trigger(),
             Some(ColumnType::Timestamptz),
             arg_types::NO_ARGS,
-        ),
+        )
+        .volatile(),
         m(
             "current_timestamp",
             Scalar,
@@ -53,7 +54,8 @@ pub(super) fn datetime_functions() -> Vec<FunctionMeta> {
             no_trigger(),
             Some(ColumnType::Timestamptz),
             arg_types::NO_ARGS,
-        ),
+        )
+        .volatile(),
         m(
             "datetime",
             Scalar,

nodedb-sql/src/functions/builtins/scalars/id_fn.rs

@@ -31,7 +31,8 @@ pub(super) fn id_fn_functions() -> Vec<FunctionMeta> {
             no_trigger(),
             Some(ColumnType::Uuid),
             arg_types::NO_ARGS,
-        ),
+        )
+        .volatile(),
         // `uuid_v4` and `gen_random_uuid` are aliases for `uuid` — same
         // `nodedb_query::functions::id::try_eval` match arm
         // (`"uuid" | "uuid_v4" | "gen_random_uuid"`).
@@ -43,7 +44,8 @@ pub(super) fn id_fn_functions() -> Vec<FunctionMeta> {
             no_trigger(),
             Some(ColumnType::Uuid),
             arg_types::NO_ARGS,
-        ),
+        )
+        .volatile(),
         m(
             "gen_random_uuid",
             Scalar,
@@ -52,7 +54,8 @@ pub(super) fn id_fn_functions() -> Vec<FunctionMeta> {
             no_trigger(),
             Some(ColumnType::Uuid),
             arg_types::NO_ARGS,
-        ),
+        )
+        .volatile(),
         m(
             "uuid_v7",
             Scalar,
@@ -61,7 +64,8 @@ pub(super) fn id_fn_functions() -> Vec<FunctionMeta> {
             no_trigger(),
             Some(ColumnType::Uuid),
             arg_types::NO_ARGS,
-        ),
+        )
+        .volatile(),
         m(
             "ulid",
             Scalar,
@@ -70,7 +74,8 @@ pub(super) fn id_fn_functions() -> Vec<FunctionMeta> {
             no_trigger(),
             Some(ColumnType::Ulid),
             arg_types::NO_ARGS,
-        ),
+        )
+        .volatile(),
         m(
             "cuid2",
             Scalar,
@@ -79,7 +84,8 @@ pub(super) fn id_fn_functions() -> Vec<FunctionMeta> {
             no_trigger(),
             Some(ColumnType::String),
             arg_types::NO_ARGS,
-        ),
+        )
+        .volatile(),
         // `nanoid(length?)` — the optional length argument means max_args is
         // 1 even though the common call form `nanoid()` takes none.
         m(
@@ -90,7 +96,8 @@ pub(super) fn id_fn_functions() -> Vec<FunctionMeta> {
             no_trigger(),
             Some(ColumnType::String),
             arg_types::NANOID_ARGS,
-        ),
+        )
+        .volatile(),
         m(
             "is_uuid",
             Scalar,