reddb_server/runtime/

impl_dml.rs

//! DML execution: INSERT, UPDATE, DELETE via SQL AST
//!
//! Implements `execute_insert`, `execute_update`, and `execute_delete` on
//! `RedDBRuntime`.  Each method translates the parsed AST into entity-level
//! operations through the existing `RuntimeEntityPort` trait so that all
//! cross-cutting concerns (WAL, indexing, replication) are automatically
//! applied.

use crate::application::entity::{
    metadata_from_json, AppliedEntityMutation, CreateDocumentInput, CreateEdgeInput,
    CreateEntityOutput, CreateKvInput, CreateNodeInput, CreateRowInput, CreateRowsBatchInput,
    CreateVectorInput, DeleteEntityInput, PatchEntityOperation, PatchEntityOperationType,
    RowUpdateColumnRule, RowUpdateContractPlan,
};
use crate::application::ports::{
    build_row_update_contract_plan, entity_row_fields_snapshot,
    normalize_row_update_assignment_with_plan, normalize_row_update_value_for_rule,
    RuntimeEntityPort,
};
use crate::application::ttl_payload::has_internal_ttl_metadata;
use crate::presentation::entity_json::storage_value_to_json;
use crate::runtime::mvcc::current_connection_id;
use crate::storage::query::ast::{BinOp, Expr, FieldRef, ReturningItem, UpdateTarget};
use crate::storage::query::sql_lowering::{
    effective_delete_filter, effective_insert_rows, effective_update_filter, fold_expr_to_value,
};
use crate::storage::query::unified::{
    sys_key_collection, sys_key_created_at, sys_key_kind, sys_key_rid, sys_key_tenant,
    sys_key_updated_at, UnifiedRecord, UnifiedResult,
};
use crate::storage::unified::MetadataValue;
use crate::storage::Metadata;
use std::collections::HashMap;
use std::sync::Arc;

use super::*;

const UPDATE_APPLY_CHUNK_SIZE: usize = 2048;
const TREE_CHILD_EDGE_LABEL: &str = "TREE_CHILD";
const TREE_METADATA_PREFIX: &str = "red.tree.";

#[derive(Clone)]
struct CompiledUpdateAssignment {
    column: String,
    expr: Expr,
    compound_op: Option<BinOp>,
    metadata_key: Option<&'static str>,
    row_rule: Option<RowUpdateColumnRule>,
}

struct CompiledUpdatePlan {
    static_field_assignments: Vec<(String, Value)>,
    static_metadata_assignments: Vec<(String, MetadataValue)>,
    dynamic_assignments: Vec<CompiledUpdateAssignment>,
    row_contract_plan: Option<RowUpdateContractPlan>,
    row_modified_columns: Vec<String>,
    row_touches_unique_columns: bool,
}

#[derive(Default)]
struct MaterializedUpdateAssignments {
    dynamic_field_assignments: Vec<(String, Value)>,
    dynamic_metadata_assignments: Vec<(String, MetadataValue)>,
}

impl RedDBRuntime {
    /// Issue #524 — public read of the in-memory chain tip. Returns `None`
    /// when the collection is not a chain or has no rows (pre-genesis). On a
    /// cold cache the first call falls back to a one-time scan so the HTTP
    /// `GET /collections/:name/chain-tip` handler stays consistent with the
    /// INSERT path after a restart.
    pub fn chain_tip_for_collection(
        &self,
        collection: &str,
    ) -> Option<crate::runtime::blockchain_kind::ChainTipFull> {
        let store = self.inner.db.store();
        if !crate::runtime::blockchain_kind::is_chain(&store, collection) {
            return None;
        }
        let mut cache = self.inner.chain_tip_cache.lock();
        if let Some(existing) = cache.get(collection) {
            return Some(existing.clone());
        }
        let scanned = crate::runtime::blockchain_kind::chain_tip_full(&store, collection)?;
        cache.insert(collection.to_string(), scanned.clone());
        Some(scanned)
    }

    /// Issue #525 — walks the chain end-to-end, recomputes each block's hash
    /// against the stored fields, and returns the verification outcome.  On
    /// `ok == false` the integrity flag is persisted and the in-memory cache
    /// is updated so subsequent INSERTs surface `ChainIntegrityBroken`.
    ///
    /// Returns `None` when the collection is absent or not a `KIND blockchain`.
    pub fn verify_chain_for_collection(
        &self,
        collection: &str,
    ) -> Option<crate::runtime::blockchain_kind::VerifyChainOutcome> {
        let store = self.inner.db.store();
        let outcome = crate::runtime::blockchain_kind::verify_chain_outcome(&store, collection)?;
        if !outcome.ok {
            crate::runtime::blockchain_kind::persist_integrity_flag(&store, collection, true);
            self.inner
                .chain_integrity_broken
                .lock()
                .insert(collection.to_string(), true);
        }
        Some(outcome)
    }

    /// Issue #525 — admin clears the `ChainIntegrityBroken` flag so the chain
    /// accepts INSERTs again.  Returns `false` when the collection is not a
    /// chain.
    pub fn clear_chain_integrity_flag(&self, collection: &str) -> bool {
        let store = self.inner.db.store();
        if !crate::runtime::blockchain_kind::is_chain(&store, collection) {
            return false;
        }
        crate::runtime::blockchain_kind::persist_integrity_flag(&store, collection, false);
        self.inner
            .chain_integrity_broken
            .lock()
            .insert(collection.to_string(), false);
        true
    }

    /// Issue #525 — INSERT-time check.  Combines in-memory cache (fast path)
    /// with a one-time scan of `red_config` on cold start so the flag survives
    /// restart.
    fn is_chain_integrity_broken(&self, collection: &str) -> bool {
        {
            let cache = self.inner.chain_integrity_broken.lock();
            if let Some(v) = cache.get(collection) {
                return *v;
            }
        }
        let store = self.inner.db.store();
        let persisted =
            crate::runtime::blockchain_kind::is_integrity_broken_persisted(&store, collection)
                .unwrap_or(false);
        self.inner
            .chain_integrity_broken
            .lock()
            .insert(collection.to_string(), persisted);
        persisted
    }

    /// Issue #765 / S6 — lazily hydrate the integrity-tombstone cache from
    /// `red_config` on first access. Returns `true` when at least one
    /// tombstone range is present. Subsequent calls observe the cached state
    /// flag (`1` empty / `2` present) and skip the store scan.
    fn ensure_integrity_tombstones_loaded(&self) -> bool {
        use std::sync::atomic::Ordering;
        match self
            .inner
            .integrity_tombstones_state
            .load(Ordering::Relaxed)
        {
            1 => return false,
            2 => return true,
            _ => {}
        }
        // Cold: load under the cache lock so a concurrent reader cannot
        // observe a half-populated vector.
        let mut guard = self.inner.integrity_tombstones.lock();
        if self
            .inner
            .integrity_tombstones_state
            .load(Ordering::Relaxed)
            == 0
        {
            let ranges = crate::runtime::integrity_tombstone::load_ranges(&self.inner.db.store());
            let present = !ranges.is_empty();
            *guard = ranges;
            self.inner
                .integrity_tombstones_state
                .store(if present { 2 } else { 1 }, Ordering::Relaxed);
        }
        self.inner
            .integrity_tombstones_state
            .load(Ordering::Relaxed)
            == 2
    }

    /// Issue #765 / S6 — durably record an integrity tombstone over the
    /// inclusive RID range `[lo, hi]` of `table` (the committed rows of an
    /// input stream whose end-to-end SHA-256 digest did not match). The range
    /// is persisted to `red_config` (survives restart) and folded into the
    /// in-memory cache so the same process filters it immediately.
    pub fn record_integrity_tombstone(&self, table: &str, lo: u64, hi: u64) {
        use std::sync::atomic::Ordering;
        self.ensure_integrity_tombstones_loaded();
        let mut guard = self.inner.integrity_tombstones.lock();
        guard.push(crate::runtime::integrity_tombstone::TombstoneRange::new(
            table.to_string(),
            lo,
            hi,
        ));
        crate::runtime::integrity_tombstone::persist_ranges(&self.inner.db.store(), &guard);
        self.inner
            .integrity_tombstones_state
            .store(2, Ordering::Relaxed);
    }

    /// Issue #765 / S6 — snapshot of the currently-cached tombstone ranges.
    /// Intended for tests and forensic surfaces; the read path uses
    /// [`Self::filter_integrity_tombstoned`] which avoids the clone.
    pub fn integrity_tombstone_ranges(
        &self,
    ) -> Vec<crate::runtime::integrity_tombstone::TombstoneRange> {
        self.ensure_integrity_tombstones_loaded();
        self.inner.integrity_tombstones.lock().clone()
    }

    /// Issue #765 / S6 — drop tombstoned rows from a SELECT result in place.
    /// Fast no-op (one relaxed atomic load) when no tombstone has ever been
    /// recorded. Clears `pre_serialized_json` when any row is removed so the
    /// fast-path JSON cannot leak a filtered row back onto the wire.
    pub fn filter_integrity_tombstoned(&self, result: &mut UnifiedResult) {
        if !self.ensure_integrity_tombstones_loaded() {
            return;
        }
        let guard = self.inner.integrity_tombstones.lock();
        if guard.is_empty() {
            return;
        }
        let before = result.records.len();
        result.records.retain(|record| {
            !crate::runtime::integrity_tombstone::record_tombstoned(&guard, record)
        });
        if result.records.len() != before {
            result.pre_serialized_json = None;
        }
    }

    /// Phase 2.5.4: inject `CURRENT_TENANT()` into an INSERT when the
    /// target table is tenant-scoped and the user's column list does
    /// not already name the tenant column.
    ///
    /// Returns:
    /// * `Ok(None)` — no injection needed (non-tenant table, or user
    ///   supplied the column explicitly). Caller uses the original
    ///   query unchanged.
    /// * `Ok(Some(augmented))` — a cloned query with the tenant column
    ///   + literal value appended to every row.
    /// * `Err(..)` — table is tenant-scoped but no tenant is bound to
    ///   the current session. Fails loudly so callers don't produce
    ///   rows that RLS would then hide on read.
    fn maybe_inject_tenant_column(&self, query: &InsertQuery) -> RedDBResult<Option<InsertQuery>> {
        let Some(tenant_col) = self.tenant_column(&query.table) else {
            return Ok(None);
        };
        // User already named the column (literal match) — trust them.
        if query
            .columns
            .iter()
            .any(|c| c.eq_ignore_ascii_case(&tenant_col))
        {
            return Ok(None);
        }

        // Phase 2 PG parity: dotted-path tenancy. When `tenant_col` is a
        // nested key like `headers.tenant` we operate on the root
        // column (`headers`) and set / add the nested path inside its
        // JSON value. If the user named the root column we mutate in
        // place; otherwise we create a fresh JSON column for every row.
        if let Some(dot_pos) = tenant_col.find('.') {
            let (root, tail) = tenant_col.split_at(dot_pos);
            let tail = &tail[1..]; // drop leading '.'
            return self.inject_dotted_tenant(query, root, tail);
        }

        let Some(tenant_id) = crate::runtime::impl_core::current_tenant() else {
            return Err(RedDBError::Query(format!(
                "INSERT into tenant-scoped table '{}' requires an active tenant — \
                 run SET TENANT '<id>' first or name column '{}' explicitly",
                query.table, tenant_col
            )));
        };

        let mut augmented = query.clone();
        augmented.columns.push(tenant_col);
        let lit = Value::text(tenant_id.clone());
        for row in augmented.values.iter_mut() {
            row.push(lit.clone());
        }
        for row in augmented.value_exprs.iter_mut() {
            row.push(crate::storage::query::ast::Expr::Literal {
                value: lit.clone(),
                span: crate::storage::query::ast::Span::synthetic(),
            });
        }
        Ok(Some(augmented))
    }

    /// Dotted-path auto-fill — set `root.tail` to `CURRENT_TENANT()` on
    /// every row. Mirrors `maybe_inject_tenant_column` but mutates
    /// nested JSON instead of appending a flat column.
    ///
    /// Cases:
    /// * Root column already in the INSERT list → mutate per-row JSON
    ///   (parse, set path, re-serialize).
    /// * Root column absent → create a fresh `{tail: tenant}` JSON
    ///   object and append the root column to the INSERT.
    fn inject_dotted_tenant(
        &self,
        query: &InsertQuery,
        root: &str,
        tail: &str,
    ) -> RedDBResult<Option<InsertQuery>> {
        let active_tenant = crate::runtime::impl_core::current_tenant();
        let mut augmented = query.clone();
        let root_idx = augmented
            .columns
            .iter()
            .position(|c| c.eq_ignore_ascii_case(root));

        if let Some(idx) = root_idx {
            // User supplied the root column. Per-row: if the dotted
            // tail is already present we trust the user (admin / bulk
            // loader scenario); otherwise fill from the active
            // tenant. An unbound tenant is only an error when some
            // row actually needs filling.
            for row in augmented.values.iter_mut() {
                let Some(slot) = row.get_mut(idx) else {
                    continue;
                };
                if dotted_tail_already_set(slot, tail) {
                    continue;
                }
                let Some(tenant_id) = &active_tenant else {
                    return Err(RedDBError::Query(format!(
                        "INSERT into tenant-scoped table '{}' requires an active tenant — \
                         run SET TENANT '<id>' first or set '{}.{}' explicitly in each row",
                        query.table, root, tail
                    )));
                };
                *slot = merge_dotted_tenant(slot.clone(), tail, tenant_id)?;
            }
            // Expression row is kept in sync by re-wrapping the
            // mutated literal; the canonical path will re-evaluate
            // against the same JSON shape.
            for (row_idx, row) in augmented.value_exprs.iter_mut().enumerate() {
                if let Some(slot) = row.get_mut(idx) {
                    let new_value = augmented
                        .values
                        .get(row_idx)
                        .and_then(|v| v.get(idx))
                        .cloned()
                        .unwrap_or(Value::Null);
                    *slot = crate::storage::query::ast::Expr::Literal {
                        value: new_value,
                        span: crate::storage::query::ast::Span::synthetic(),
                    };
                }
            }
        } else {
            // No root column in the INSERT list — auto-fill needs a
            // bound tenant to synthesise one. Error loud so we never
            // create a tenant-less row that RLS would then hide.
            let Some(tenant_id) = &active_tenant else {
                return Err(RedDBError::Query(format!(
                    "INSERT into tenant-scoped table '{}' requires an active tenant — \
                     run SET TENANT '<id>' first or name path '{}.{}' explicitly",
                    query.table, root, tail
                )));
            };
            // Create a fresh JSON column with only the tenant path set.
            augmented.columns.push(root.to_string());
            let fresh = merge_dotted_tenant(Value::Null, tail, tenant_id)?;
            for row in augmented.values.iter_mut() {
                row.push(fresh.clone());
            }
            for row in augmented.value_exprs.iter_mut() {
                row.push(crate::storage::query::ast::Expr::Literal {
                    value: fresh.clone(),
                    span: crate::storage::query::ast::Span::synthetic(),
                });
            }
        }

        Ok(Some(augmented))
    }

    /// Returns `(affected_count, lsns)`. For the txn (xmax-stamp) path,
    /// `lsns` is empty because events fire at commit time.
    fn delete_entities_batch(
        &self,
        collection: &str,
        ids: &[EntityId],
    ) -> RedDBResult<(u64, Vec<u64>)> {
        if ids.is_empty() {
            return Ok((0, vec![]));
        }

        let store = self.db().store();
        let Some(manager) = store.get_collection(collection) else {
            return Ok((0, vec![]));
        };

        let active_xid = self.current_xid();
        let conn_id = crate::runtime::impl_core::current_connection_id();
        let mut autocommit_xid = None;
        let mut tombstoned_ids = Vec::new();
        let mut tombstoned_entities = Vec::new();
        let mut physical_delete_ids = Vec::new();
        let table_row_resolver =
            crate::runtime::table_row_mvcc_resolver::TableRowMvccReadResolver::current_statement();
        // Phase 3: versioned graph collections tombstone node/edge
        // deletes (xmax stamp) instead of physically removing them, so
        // `AS OF` time-travel can still resolve the pre-delete version.
        // Non-versioned graph collections keep the legacy physical
        // delete (no history). Row entities (Phase 1/2) always tombstone
        // under universal MVCC and are unaffected by this flag.
        //
        // Vectors (Phase 3b): versioned vector deletes tombstone (xmax
        // stamp) instead of physically dropping, so history is retained.
        // The `VECTOR SEARCH` read path post-filters every candidate
        // through the current-snapshot visibility gate (which hides
        // `xmax != 0` even in autocommit), so a tombstoned versioned
        // vector never reaches live search. Non-versioned vectors keep
        // the legacy physical delete.
        let versioned_collection = self.vcs_is_versioned(collection).unwrap_or(false);

        for &id in ids {
            let Some(mut entity) = manager.get(id) else {
                continue;
            };
            let is_versioned_graph = versioned_collection
                && matches!(entity.data, EntityData::Node(_) | EntityData::Edge(_));
            let is_versioned_vector =
                versioned_collection && matches!(entity.data, EntityData::Vector(_));
            if matches!(entity.data, EntityData::Row(_))
                || is_versioned_graph
                || is_versioned_vector
            {
                let previous_xmax = entity.xmax;
                if matches!(entity.kind, crate::storage::EntityKind::TableRow { .. }) {
                    if table_row_resolver.resolve_candidate(&entity).is_none() {
                        continue;
                    }
                } else if is_versioned_vector {
                    // Versioned vectors gate the delete on the statement
                    // snapshot (not a crude `xmax != 0`) so a concurrent
                    // delete whose snapshot predates a still-uncommitted
                    // tombstone still targets the row, records its own
                    // pending tombstone, and conflicts at commit
                    // (first-committer-wins).
                    if table_row_resolver.resolve_read_candidate(&entity).is_none() {
                        continue;
                    }
                } else if entity.xmax != 0 {
                    continue;
                }

                let xid = match active_xid {
                    Some(xid) => xid,
                    None => match autocommit_xid {
                        Some(xid) => xid,
                        None => {
                            let mgr = self.snapshot_manager();
                            let xid = mgr.begin();
                            autocommit_xid = Some(xid);
                            xid
                        }
                    },
                };
                entity.set_xmax(xid);
                if manager.update(entity.clone()).is_ok() {
                    if active_xid.is_some() {
                        self.record_pending_tombstone(conn_id, collection, id, xid, previous_xmax);
                    }
                    tombstoned_entities.push(entity);
                    tombstoned_ids.push(id);
                }
            } else {
                physical_delete_ids.push(id);
            }
        }

        if let Some(xid) = autocommit_xid {
            self.snapshot_manager().commit(xid);
        }

        let mut affected = tombstoned_ids.len() as u64;
        let mut lsns = Vec::with_capacity(tombstoned_ids.len() + physical_delete_ids.len());
        if active_xid.is_some() {
            store
                .persist_entities_to_pager(collection, &tombstoned_entities)
                .map_err(|err| RedDBError::Internal(err.to_string()))?;
        } else {
            store
                .persist_entities_to_pager(collection, &tombstoned_entities)
                .map_err(|err| RedDBError::Internal(err.to_string()))?;
            for id in &tombstoned_ids {
                store.context_index().remove_entity(*id);
                let lsn = self.cdc_emit(
                    crate::replication::cdc::ChangeOperation::Delete,
                    collection,
                    id.raw(),
                    "entity",
                );
                lsns.push(lsn);
            }
        }

        let deleted_ids = store
            .delete_batch(collection, &physical_delete_ids)
            .map_err(|err| RedDBError::Internal(err.to_string()))?;
        affected += deleted_ids.len() as u64;
        for id in &deleted_ids {
            store.context_index().remove_entity(*id);
            let lsn = self.cdc_emit(
                crate::replication::cdc::ChangeOperation::Delete,
                collection,
                id.raw(),
                "entity",
            );
            lsns.push(lsn);
        }

        Ok((affected, lsns))
    }

    /// Flushes context-index updates and CDC for each applied mutation.
    /// Returns one LSN per entity in the same order as `applied`.
    fn flush_update_chunk(&self, applied: &[AppliedEntityMutation]) -> RedDBResult<Vec<u64>> {
        if applied.is_empty() {
            return Ok(Vec::new());
        }

        let store = self.db().store();
        if applied.iter().any(|item| item.context_index_dirty) {
            store.context_index().index_entities(
                &applied[0].collection,
                applied
                    .iter()
                    .filter(|item| item.context_index_dirty)
                    .map(|item| &item.entity),
            );
        }

        for item in applied {
            self.refresh_update_secondary_indexes(item)?;
        }

        let mut lsns = Vec::with_capacity(applied.len());
        for item in applied {
            let lsn = self.cdc_emit_prebuilt(
                crate::replication::cdc::ChangeOperation::Update,
                &item.collection,
                &item.entity,
                update_cdc_item_kind(self, &item.collection, &item.entity),
                item.metadata.as_ref(),
                false,
            );
            lsns.push(lsn);
        }
        Ok(lsns)
    }

    fn persist_update_chunk(&self, applied: &[AppliedEntityMutation]) -> RedDBResult<()> {
        self.persist_applied_entity_mutations(applied)
    }

    fn refresh_update_secondary_indexes(&self, applied: &AppliedEntityMutation) -> RedDBResult<()> {
        if applied.pre_mutation_fields.is_empty() {
            return Ok(());
        }
        let post = entity_row_fields_snapshot(&applied.entity);
        if post.is_empty() {
            return Ok(());
        }

        // Use the parent-expanded set so that dot-path indexes (e.g.
        // "body.service.tier") are triggered when the root field ("body")
        // is modified.
        let indexed_cols = self
            .index_store_ref()
            .indexed_columns_set_with_parents(&applied.collection);
        if indexed_cols.is_empty() {
            return Ok(());
        }

        // Single-source documents keep promoted index columns (`score`) in the
        // body, not as stored fields. Resolve each indexed column from the body
        // and fold it into the field snapshots so the diff below sees the value
        // move — for an ordinary stored column this adds nothing.
        let pre = self
            .index_store_ref()
            .augment_body_derived_index_fields(&applied.pre_mutation_fields, &indexed_cols);
        let post = self
            .index_store_ref()
            .augment_body_derived_index_fields(&post, &indexed_cols);

        if let Some(old_version) = applied.replaced_entity.as_ref() {
            let old_index_fields: Vec<(String, crate::storage::schema::Value)> = pre
                .iter()
                .filter(|(col, _)| indexed_cols.contains(col))
                .cloned()
                .collect();
            let new_index_fields: Vec<(String, crate::storage::schema::Value)> = post
                .iter()
                .filter(|(col, _)| indexed_cols.contains(col))
                .cloned()
                .collect();
            if !old_index_fields.is_empty() {
                self.index_store_ref()
                    .index_entity_delete(&applied.collection, old_version.id, &old_index_fields)
                    .map_err(crate::RedDBError::Internal)?;
            }
            if !new_index_fields.is_empty() {
                self.index_store_ref()
                    .index_entity_insert(&applied.collection, applied.entity.id, &new_index_fields)
                    .map_err(crate::RedDBError::Internal)?;
            }
            return Ok(());
        }

        let damage = crate::application::entity::row_damage_vector(&pre, &post);
        if damage
            .touched_columns()
            .into_iter()
            .any(|col| indexed_cols.contains(col))
        {
            self.index_store_ref()
                .index_entity_update(&applied.collection, applied.id, &pre, &post)
                .map_err(crate::RedDBError::Internal)?;
        }
        Ok(())
    }

    /// Execute INSERT INTO table [entity_type] (cols) VALUES (vals), ...
    ///
    /// Each row in `query.values` is zipped with `query.columns` to produce a
    /// set of named fields, which is then dispatched based on entity_type.
    pub fn execute_insert(
        &self,
        raw_query: &str,
        query: &InsertQuery,
    ) -> RedDBResult<RuntimeQueryResult> {
        self.check_write(crate::runtime::write_gate::WriteKind::Dml)?;
        // CollectionContract gate (#49): single entry point for the
        // operator's collection-level write rules. Today this is a
        // no-op for INSERT (APPEND ONLY permits insert); routing
        // through the gate now means future contract bits — versioned,
        // vault-only writes — plug in once instead of per verb.
        crate::runtime::collection_contract::CollectionContractGate::check(
            self,
            &query.table,
            crate::runtime::collection_contract::MutationKind::Insert,
        )?;
        // Phase 2.5.4 table-scoped tenancy: if the target table is
        // tenant-scoped and the user didn't name the tenant column,
        // auto-inject it with the thread-local `CURRENT_TENANT()`
        // value. When the column is named explicitly we trust the
        // caller (useful for admin tooling that writes on behalf of
        // specific tenants). An unbound tenant on an implicit-fill
        // path errors up front rather than producing a row the RLS
        // policy would silently hide.
        let augmented_owned;
        let query = match self.maybe_inject_tenant_column(query)? {
            Some(new_q) => {
                augmented_owned = new_q;
                &augmented_owned
            }
            None => query,
        };
        self.check_insert_column_policy(query)?;
        if let Some(ref embed_config) = query.auto_embed {
            let provider = crate::ai::parse_provider(&embed_config.provider)?;
            // S3 / #711: planner-level provider gate. Runs before the
            // local-model preflight and the API-key resolver so neither
            // side-effect fires when policy denies.
            crate::runtime::ai::provider_gate::enforce(self, &provider)?;
            if matches!(provider, crate::ai::AiProvider::Local) {
                crate::runtime::ai::local_embedding::ensure_local_embedding_available()?;
                // Issue #682 — pre-flight the local model registry before
                // any row write. Missing model, uninstalled artifacts,
                // wrong task, and disabled-feature failures surface as
                // deterministic errors that leave the target collection
                // untouched, satisfying the "no partial writes on
                // embedding failure" criterion for the failure modes
                // owned by the local provider.
                let model_name = embed_config.model.as_deref().map(str::trim).unwrap_or("");
                if model_name.is_empty() {
                    return Err(RedDBError::Query(
                        "AUTO EMBED with provider=local requires MODEL '<registered-model-name>'; \
                         the local provider does not have an implicit default model"
                            .to_string(),
                    ));
                }
                crate::runtime::ai::local_embedding::preflight_local_embedding(
                    &self.inner.db,
                    model_name,
                )?;
            }
        }

        let mut inserted_count: u64 = 0;
        let effective_rows =
            effective_insert_rows(query).map_err(|msg| RedDBError::Query(msg.to_string()))?;

        // Ensure the collection exists (auto-create on first insert).
        let store = self.inner.db.store();
        let _ = store.get_or_create_collection(&query.table);
        let declared_model = self
            .db()
            .collection_contract_arc(&query.table)
            .map(|contract| contract.declared_model);

        let mut returning_snapshots: Option<Vec<Vec<(String, Value)>>> =
            if query.returning.is_some() {
                Some(Vec::with_capacity(effective_rows.len()))
            } else {
                None
            };
        let mut returning_result: Option<UnifiedResult> = None;

        if matches!(query.entity_type, InsertEntityType::Row)
            && !matches!(
                declared_model,
                Some(crate::catalog::CollectionModel::TimeSeries)
            )
        {
            // Issue #523 + #524: blockchain collections seal each row into the
            // chain. When the caller omits the reserved columns, the engine
            // auto-fills (#523). When the caller supplies any reserved column,
            // the values are validated against the current tip and a mismatch
            // surfaces a `BlockchainConflict:` error mapped to HTTP 409 (#524).
            //
            // The whole batch runs under a per-collection chain lock so two
            // concurrent submitters can't both bind to the same prev_hash —
            // the loser observes the advanced tip and gets 409 with the new
            // tip so it can retry.
            let chain_mode = crate::runtime::blockchain_kind::is_chain(&store, &query.table);
            let _chain_lock_arc: Option<Arc<parking_lot::Mutex<()>>> = if chain_mode {
                Some(self.inner.rmw_locks.lock_for(&query.table, "__chain__"))
            } else {
                None
            };
            let _chain_guard = _chain_lock_arc.as_ref().map(|m| m.lock());

            // Issue #525 — refuse new blocks if the chain has been marked
            // `integrity = broken` until an admin clears the flag.
            if chain_mode && self.is_chain_integrity_broken(&query.table) {
                return Err(RedDBError::InvalidOperation(format!(
                    "ChainIntegrityBroken: collection '{}' is locked until \
                     POST /collections/{}/clear-integrity-flag is called by an admin",
                    query.table, query.table
                )));
            }

            // Pull the tip from the in-memory cache; fall back to a one-time
            // scan if the cache hasn't seen this collection yet (cold start
            // after restart). Cache is updated below as rows are sealed.
            let mut chain_tip_full: Option<crate::runtime::blockchain_kind::ChainTipFull> =
                if chain_mode {
                    let mut cache = self.inner.chain_tip_cache.lock();
                    if let Some(existing) = cache.get(&query.table) {
                        Some(existing.clone())
                    } else if let Some(scanned) =
                        crate::runtime::blockchain_kind::chain_tip_full(&store, &query.table)
                    {
                        cache.insert(query.table.clone(), scanned.clone());
                        Some(scanned)
                    } else {
                        None
                    }
                } else {
                    None
                };

            let mut rows = Vec::with_capacity(effective_rows.len());
            for row_values in &effective_rows {
                if row_values.len() != query.columns.len() {
                    return Err(RedDBError::Query(format!(
                        "INSERT column count ({}) does not match value count ({})",
                        query.columns.len(),
                        row_values.len()
                    )));
                }
                let (mut fields, mut metadata) =
                    split_insert_metadata(self, &query.columns, row_values)?;
                if chain_mode {
                    use crate::runtime::blockchain_kind::{
                        chain_conflict_error, COL_BLOCK_HEIGHT, COL_HASH, COL_PREV_HASH,
                        COL_TIMESTAMP, RESERVED_COLUMNS,
                    };
                    let supplied_height = fields
                        .iter()
                        .find(|(k, _)| k == COL_BLOCK_HEIGHT)
                        .map(|(_, v)| v.clone());
                    let supplied_prev = fields
                        .iter()
                        .find(|(k, _)| k == COL_PREV_HASH)
                        .map(|(_, v)| v.clone());
                    let supplied_ts = fields
                        .iter()
                        .find(|(k, _)| k == COL_TIMESTAMP)
                        .map(|(_, v)| v.clone());
                    let supplied_hash = fields.iter().any(|(k, _)| k == COL_HASH);
                    let user_supplied_any = supplied_height.is_some()
                        || supplied_prev.is_some()
                        || supplied_ts.is_some()
                        || supplied_hash;

                    fields.retain(|(k, _)| !RESERVED_COLUMNS.contains(&k.as_str()));
                    let payload = crate::runtime::blockchain_kind::canonical_payload(&fields);

                    let (tip_prev_hash, tip_next_height) = match &chain_tip_full {
                        Some(t) => (t.hash, t.height + 1),
                        None => (crate::storage::blockchain::GENESIS_PREV_HASH, 0u64),
                    };
                    let server_now = crate::runtime::blockchain_kind::now_ms();

                    let (use_prev, use_height, use_ts) = if user_supplied_any {
                        // Caller is participating in the chain protocol —
                        // every field must be supplied AND match the tip.
                        if supplied_hash {
                            return Err(chain_conflict_error(
                                tip_next_height.saturating_sub(1),
                                tip_prev_hash,
                                chain_tip_full.as_ref().map(|t| t.timestamp_ms).unwrap_or(0),
                                server_now,
                                "hash column is engine-computed and cannot be supplied",
                            ));
                        }
                        let caller_prev = match &supplied_prev {
                            Some(Value::Blob(b)) if b.len() == 32 => {
                                let mut a = [0u8; 32];
                                a.copy_from_slice(b);
                                a
                            }
                            Some(Value::Text(s)) if s.len() == 64 => {
                                // Accept hex-encoded prev_hash so JSON / SQL
                                // callers without literal-blob syntax can
                                // still participate in the chain protocol.
                                let mut a = [0u8; 32];
                                let mut ok = true;
                                for (i, slot) in a.iter_mut().enumerate() {
                                    let pair = &s.as_ref()[i * 2..i * 2 + 2];
                                    match u8::from_str_radix(pair, 16) {
                                        Ok(byte) => *slot = byte,
                                        Err(_) => {
                                            ok = false;
                                            break;
                                        }
                                    }
                                }
                                if !ok {
                                    return Err(chain_conflict_error(
                                        tip_next_height.saturating_sub(1),
                                        tip_prev_hash,
                                        chain_tip_full
                                            .as_ref()
                                            .map(|t| t.timestamp_ms)
                                            .unwrap_or(0),
                                        server_now,
                                        "prev_hash is not valid hex",
                                    ));
                                }
                                a
                            }
                            _ => {
                                return Err(chain_conflict_error(
                                    tip_next_height.saturating_sub(1),
                                    tip_prev_hash,
                                    chain_tip_full.as_ref().map(|t| t.timestamp_ms).unwrap_or(0),
                                    server_now,
                                    "prev_hash missing or not a 32-byte Blob",
                                ));
                            }
                        };
                        if caller_prev != tip_prev_hash {
                            return Err(chain_conflict_error(
                                tip_next_height.saturating_sub(1),
                                tip_prev_hash,
                                chain_tip_full.as_ref().map(|t| t.timestamp_ms).unwrap_or(0),
                                server_now,
                                "prev_hash does not match current tip",
                            ));
                        }
                        let caller_height = match &supplied_height {
                            Some(Value::UnsignedInteger(v)) => *v,
                            Some(Value::Integer(v)) if *v >= 0 => *v as u64,
                            _ => {
                                return Err(chain_conflict_error(
                                    tip_next_height.saturating_sub(1),
                                    tip_prev_hash,
                                    chain_tip_full.as_ref().map(|t| t.timestamp_ms).unwrap_or(0),
                                    server_now,
                                    "block_height missing or not an unsigned integer",
                                ));
                            }
                        };
                        if caller_height != tip_next_height {
                            return Err(chain_conflict_error(
                                tip_next_height.saturating_sub(1),
                                tip_prev_hash,
                                chain_tip_full.as_ref().map(|t| t.timestamp_ms).unwrap_or(0),
                                server_now,
                                "block_height does not match tip+1",
                            ));
                        }
                        let caller_ts = match &supplied_ts {
                            Some(Value::UnsignedInteger(v)) => *v,
                            Some(Value::Integer(v)) if *v >= 0 => *v as u64,
                            _ => {
                                return Err(chain_conflict_error(
                                    tip_next_height.saturating_sub(1),
                                    tip_prev_hash,
                                    chain_tip_full.as_ref().map(|t| t.timestamp_ms).unwrap_or(0),
                                    server_now,
                                    "timestamp missing or not an unsigned integer",
                                ));
                            }
                        };
                        let drift = (caller_ts as i128) - (server_now as i128);
                        if drift.abs() > 60_000 {
                            return Err(chain_conflict_error(
                                tip_next_height.saturating_sub(1),
                                tip_prev_hash,
                                chain_tip_full.as_ref().map(|t| t.timestamp_ms).unwrap_or(0),
                                server_now,
                                "timestamp outside ±60s of server_time",
                            ));
                        }
                        (caller_prev, caller_height, caller_ts)
                    } else {
                        (tip_prev_hash, tip_next_height, server_now)
                    };

                    let (reserved, new_hash) =
                        crate::runtime::blockchain_kind::make_block_reserved_fields(
                            use_prev, use_height, use_ts, &payload,
                        );
                    fields.extend(reserved);
                    chain_tip_full = Some(crate::runtime::blockchain_kind::ChainTipFull {
                        height: use_height,
                        hash: new_hash,
                        timestamp_ms: use_ts,
                    });
                }
                // Issue #522 — signed-writes verification. On collections
                // created with `SIGNED_BY (...)` the row must carry valid
                // `signer_pubkey` + `signature` reserved columns. Runs
                // after chain_mode so canonical payload covers user-supplied
                // fields only (blockchain reserved columns are filtered by
                // `canonical_payload`; the two signed-writes reserved
                // columns are split out before payload computation, then
                // re-attached for storage). The blockchain + SIGNED_BY
                // composition is owned by issue #526; we keep #522 to the
                // non-chain path and let chain_mode collections punt to that
                // slice rather than half-wire it here.
                if crate::runtime::signed_writes_kind::is_signed(&store, &query.table) {
                    let (pk_col, sig_col, residual) =
                        crate::runtime::signed_writes_kind::split_signature_fields(fields);
                    let payload = crate::runtime::blockchain_kind::canonical_payload(&residual);
                    let reg = crate::runtime::signed_writes_kind::registry(&store, &query.table);
                    crate::runtime::signed_writes_kind::verify_row(
                        &reg,
                        pk_col.as_ref().map(|c| c.bytes.as_slice()),
                        sig_col.as_ref().map(|c| c.bytes.as_slice()),
                        &payload,
                    )
                    .map_err(crate::runtime::signed_writes_kind::map_error)?;
                    fields = residual;
                    // Round-trip the reserved columns with the value
                    // type the caller supplied (Text/hex on the SQL path,
                    // Blob on the binary path). Keeps SELECT and WHERE
                    // predicates symmetric with the INSERT shape.
                    if let Some(col) = pk_col {
                        fields.push((
                            crate::storage::signed_writes::RESERVED_SIGNER_PUBKEY_COL.to_string(),
                            col.raw_value,
                        ));
                    }
                    if let Some(col) = sig_col {
                        fields.push((
                            crate::storage::signed_writes::RESERVED_SIGNATURE_COL.to_string(),
                            col.raw_value,
                        ));
                    }
                }
                merge_with_clauses(
                    &mut metadata,
                    query.ttl_ms,
                    query.expires_at_ms,
                    &query.with_metadata,
                );
                if let Some(snaps) = returning_snapshots.as_mut() {
                    snaps.push(fields.clone());
                }
                rows.push(CreateRowInput {
                    collection: query.table.clone(),
                    fields,
                    metadata,
                    node_links: Vec::new(),
                    vector_links: Vec::new(),
                });
            }
            let outputs = self.create_rows_batch(CreateRowsBatchInput {
                collection: query.table.clone(),
                rows,
                suppress_events: query.suppress_events,
            })?;
            inserted_count = outputs.len() as u64;

            // Chain mode: commit the new tip to the in-memory cache only after
            // the batch persisted successfully. If the batch threw mid-way the
            // cache stays on the previous tip and the chain lock releases.
            if chain_mode {
                if let Some(new_tip) = chain_tip_full.as_ref() {
                    self.inner
                        .chain_tip_cache
                        .lock()
                        .insert(query.table.clone(), new_tip.clone());
                }
            }

            // Hypertable chunk routing: if this table was declared via
            // CREATE HYPERTABLE, register each row's time-column value
            // with the registry so chunk metadata (bounds, row counts,
            // TTL eligibility) stays current. This is what lets
            // HYPERTABLE_PRUNE_CHUNKS answer real questions + lets the
            // retention daemon sweep expired chunks without scanning
            // every row.
            if let Some(spec) = self.inner.db.hypertables().get(&query.table) {
                let time_col = &spec.time_column;
                // Find the column's index in the INSERT column list.
                if let Some(idx) = query.columns.iter().position(|c| c == time_col) {
                    for row in &effective_rows {
                        if let Some(Value::Integer(n) | Value::BigInt(n)) = row.get(idx) {
                            if *n >= 0 {
                                let _ = self.inner.db.hypertables().route(&query.table, *n as u64);
                            }
                        } else if let Some(Value::UnsignedInteger(n)) = row.get(idx) {
                            let _ = self.inner.db.hypertables().route(&query.table, *n);
                        }
                    }
                }
            }

            if let (Some(items), Some(snaps)) =
                (query.returning.as_ref(), returning_snapshots.take())
            {
                let snaps = row_insert_returning_snapshots(&outputs, snaps);
                returning_result = Some(build_returning_result(items, &snaps, Some(&outputs)));
            }
        } else {
            // Issue #419: surface the inserted entity id on every INSERT path.
            // For Node/Edge/Vector/Document/Kv we now keep each CreateEntityOutput
            // so a RETURNING clause (and the unconditional inserted_ids list,
            // below) can expose the engine-assigned id. TimeSeries (the row
            // branch in this else) still returns the not-supported error
            // because create_timeseries_point isn't plumbed through this fn.
            let mut entity_outputs: Vec<crate::application::entity::CreateEntityOutput> =
                Vec::with_capacity(effective_rows.len());
            let mut returning_field_snaps: Vec<Vec<(String, Value)>> = if query.returning.is_some()
            {
                Vec::with_capacity(effective_rows.len())
            } else {
                Vec::new()
            };
            if matches!(
                query.entity_type,
                InsertEntityType::Node | InsertEntityType::Edge
            ) {
                enum PreparedGraphInsert {
                    Node {
                        fields: Vec<(String, Value)>,
                        input: CreateNodeInput,
                    },
                    Edge {
                        fields: Vec<(String, Value)>,
                        input: CreateEdgeInput,
                    },
                }

                let mut prepared = Vec::with_capacity(effective_rows.len());
                for row_values in &effective_rows {
                    if row_values.len() != query.columns.len() {
                        return Err(RedDBError::Query(format!(
                            "INSERT column count ({}) does not match value count ({})",
                            query.columns.len(),
                            row_values.len()
                        )));
                    }

                    match query.entity_type {
                        InsertEntityType::Node => {
                            let (node_values, mut metadata) =
                                split_insert_metadata(self, &query.columns, row_values)?;
                            merge_with_clauses(
                                &mut metadata,
                                query.ttl_ms,
                                query.expires_at_ms,
                                &query.with_metadata,
                            );
                            ensure_non_tree_reserved_metadata_entries(&metadata)?;
                            apply_collection_default_ttl_metadata(
                                self,
                                &query.table,
                                &mut metadata,
                            );
                            let (columns, values) = pairwise_columns_values(&node_values);
                            let label = find_column_value_string(&columns, &values, "label")?;
                            let node_type =
                                find_column_value_opt_string(&columns, &values, "node_type");
                            let properties = extract_remaining_properties(
                                &columns,
                                &values,
                                &["label", "node_type"],
                            );
                            crate::reserved_fields::ensure_no_reserved_public_item_fields(
                                properties.iter().map(|(key, _)| key.as_str()),
                                &format!("node '{}'", query.table),
                            )?;
                            prepared.push(PreparedGraphInsert::Node {
                                fields: node_values,
                                input: CreateNodeInput {
                                    collection: query.table.clone(),
                                    label,
                                    node_type,
                                    properties,
                                    metadata,
                                    embeddings: Vec::new(),
                                    table_links: Vec::new(),
                                    node_links: Vec::new(),
                                },
                            });
                        }
                        InsertEntityType::Edge => {
                            let (edge_values, mut metadata) =
                                split_insert_metadata(self, &query.columns, row_values)?;
                            merge_with_clauses(
                                &mut metadata,
                                query.ttl_ms,
                                query.expires_at_ms,
                                &query.with_metadata,
                            );
                            ensure_non_tree_reserved_metadata_entries(&metadata)?;
                            apply_collection_default_ttl_metadata(
                                self,
                                &query.table,
                                &mut metadata,
                            );
                            let (columns, values) = pairwise_columns_values(&edge_values);
                            let label = find_column_value_string(&columns, &values, "label")?;
                            ensure_non_tree_structural_edge_label(&label)?;
                            let from_id = resolve_edge_endpoint_any(
                                self.inner.db.store().as_ref(),
                                &query.table,
                                &columns,
                                &values,
                                &["from_rid", "from"],
                            )?;
                            let to_id = resolve_edge_endpoint_any(
                                self.inner.db.store().as_ref(),
                                &query.table,
                                &columns,
                                &values,
                                &["to_rid", "to"],
                            )?;
                            let weight = find_column_value_f32_opt(&columns, &values, "weight");
                            let properties = extract_remaining_properties(
                                &columns,
                                &values,
                                &["label", "from_rid", "to_rid", "from", "to", "weight"],
                            );
                            crate::reserved_fields::ensure_no_reserved_public_item_fields(
                                properties.iter().map(|(key, _)| key.as_str()),
                                &format!("edge '{}'", query.table),
                            )?;
                            prepared.push(PreparedGraphInsert::Edge {
                                fields: edge_values,
                                input: CreateEdgeInput {
                                    collection: query.table.clone(),
                                    label,
                                    from: EntityId::new(from_id),
                                    to: EntityId::new(to_id),
                                    weight,
                                    properties,
                                    metadata,
                                },
                            });
                        }
                        _ => unreachable!("prepared graph insert only handles NODE and EDGE"),
                    }
                }

                ensure_graph_insert_contract(self, &query.table)?;
                let mut batch = self.inner.db.batch();
                for item in prepared {
                    match item {
                        PreparedGraphInsert::Node { fields, input } => {
                            if query.returning.is_some() {
                                returning_field_snaps.push(fields);
                            }
                            let node_type = input.node_type.unwrap_or_else(|| input.label.clone());
                            batch = batch.add_node_with_type(
                                input.collection,
                                input.label,
                                node_type,
                                input.properties.into_iter().collect(),
                                input.metadata.into_iter().collect(),
                            );
                        }
                        PreparedGraphInsert::Edge { fields, input } => {
                            if query.returning.is_some() {
                                returning_field_snaps.push(fields);
                            }
                            batch = batch.add_edge(
                                input.collection,
                                input.label,
                                input.from,
                                input.to,
                                input.weight.unwrap_or(1.0),
                                input.properties.into_iter().collect(),
                                input.metadata.into_iter().collect(),
                            );
                        }
                    }
                }
                let batch_result = batch
                    .execute()
                    .map_err(|err| RedDBError::Internal(format!("{err:?}")))?;
                let (ids, entity_kind) = match query.entity_type {
                    InsertEntityType::Node => (batch_result.nodes, "graph_node"),
                    InsertEntityType::Edge => (batch_result.edges, "graph_edge"),
                    _ => unreachable!("prepared graph insert only handles NODE and EDGE"),
                };
                for id in &ids {
                    self.stamp_xmin_if_in_txn(&query.table, *id);
                }
                if query.returning.is_some() {
                    returning_field_snaps = graph_insert_returning_snapshots(
                        self.inner.db.store().as_ref(),
                        &query.table,
                        &ids,
                    );
                }
                self.cdc_emit_insert_batch_no_cache_invalidate(&query.table, &ids, entity_kind);
                let store = self.inner.db.store();
                entity_outputs.extend(ids.iter().map(|id| {
                    crate::application::entity::CreateEntityOutput {
                        id: *id,
                        entity: store.get(&query.table, *id),
                    }
                }));
                inserted_count = ids.len() as u64;
            } else {
                for row_values in &effective_rows {
                    if row_values.len() != query.columns.len() {
                        return Err(RedDBError::Query(format!(
                            "INSERT column count ({}) does not match value count ({})",
                            query.columns.len(),
                            row_values.len()
                        )));
                    }

                    match query.entity_type {
                        InsertEntityType::Row => {
                            if query.returning.is_some() {
                                return Err(RedDBError::Query(
                                "RETURNING is not yet supported for this INSERT path (TimeSeries)"
                                    .to_string(),
                            ));
                            }
                            let (fields, mut metadata) =
                                split_insert_metadata(self, &query.columns, row_values)?;
                            merge_with_clauses(
                                &mut metadata,
                                query.ttl_ms,
                                query.expires_at_ms,
                                &query.with_metadata,
                            );
                            self.insert_timeseries_point(&query.table, fields, metadata)?;
                        }
                        InsertEntityType::Node | InsertEntityType::Edge => {
                            unreachable!("NODE and EDGE are handled by the prepared graph path")
                        }
                        InsertEntityType::Vector => {
                            let (vector_values, mut metadata) =
                                split_insert_metadata(self, &query.columns, row_values)?;
                            merge_with_clauses(
                                &mut metadata,
                                query.ttl_ms,
                                query.expires_at_ms,
                                &query.with_metadata,
                            );
                            let (columns, values) = pairwise_columns_values(&vector_values);
                            let dense = find_column_value_vec_f32_any(
                                &columns,
                                &values,
                                &["dense", "embedding"],
                            )?;
                            merge_vector_metadata_column(&mut metadata, &columns, &values)?;
                            let content =
                                find_column_value_opt_string(&columns, &values, "content");
                            if query.returning.is_some() {
                                returning_field_snaps.push(vector_values.clone());
                            }
                            let input = CreateVectorInput {
                                collection: query.table.clone(),
                                dense,
                                content,
                                metadata,
                                link_row: None,
                                link_node: None,
                            };
                            entity_outputs.push(self.create_vector(input)?);
                        }
                        InsertEntityType::Document => {
                            let (document_values, mut metadata) =
                                split_insert_metadata(self, &query.columns, row_values)?;
                            merge_with_clauses(
                                &mut metadata,
                                query.ttl_ms,
                                query.expires_at_ms,
                                &query.with_metadata,
                            );
                            let (columns, values) = pairwise_columns_values(&document_values);
                            let body = find_document_body_json(&columns, &values)?;
                            let input = CreateDocumentInput {
                                collection: query.table.clone(),
                                body,
                                metadata,
                                node_links: Vec::new(),
                                vector_links: Vec::new(),
                            };
                            let output = self.create_document(input)?;
                            if query.returning.is_some() {
                                let fields = output
                                    .entity
                                    .as_ref()
                                    .map(entity_row_fields_snapshot)
                                    .filter(|fields| !fields.is_empty())
                                    .unwrap_or(document_values);
                                returning_field_snaps.push(fields);
                            }
                            entity_outputs.push(output);
                        }
                        InsertEntityType::Kv => {
                            let (kv_values, mut metadata) =
                                split_insert_metadata(self, &query.columns, row_values)?;
                            merge_with_clauses(
                                &mut metadata,
                                query.ttl_ms,
                                query.expires_at_ms,
                                &query.with_metadata,
                            );
                            let (columns, values) = pairwise_columns_values(&kv_values);
                            let key = find_column_value_string(&columns, &values, "key")?;
                            let value = find_column_value(&columns, &values, "value")?;
                            if query.returning.is_some() {
                                returning_field_snaps.push(kv_values.clone());
                            }
                            let input = CreateKvInput {
                                collection: query.table.clone(),
                                key,
                                value,
                                metadata,
                            };
                            entity_outputs.push(self.create_kv(input)?);
                        }
                    }

                    inserted_count += 1;
                }
            }

            if let Some(items) = query.returning.as_ref() {
                if !entity_outputs.is_empty() {
                    returning_result = Some(build_returning_result(
                        items,
                        &returning_field_snaps,
                        Some(&entity_outputs),
                    ));
                }
            }
        }

        // Auto-embed pipeline: batch-embed fields across all inserted rows via AiBatchClient.
        if let Some(ref embed_config) = query.auto_embed {
            let store = self.inner.db.store();
            let provider = crate::ai::parse_provider(&embed_config.provider)?;
            let is_local_provider = matches!(provider, crate::ai::AiProvider::Local);
            // Local provider runs in-process — no API key path applies.
            // The pre-flight above already required `MODEL '<name>'`
            // for the local case, so the unwrap_or default below only
            // ever fires for OpenAI-compatible providers.
            let api_key = if is_local_provider {
                String::new()
            } else {
                crate::ai::resolve_api_key_from_runtime(&provider, None, self)?
            };
            let model = embed_config.model.clone().unwrap_or_else(|| {
                std::env::var("REDDB_OPENAI_EMBEDDING_MODEL")
                    .ok()
                    .unwrap_or_else(|| crate::ai::DEFAULT_OPENAI_EMBEDDING_MODEL.to_string())
            });

            // Collect the just-inserted rows (most-recently appended, reversed back to insert order).
            let manager = store
                .get_collection(&query.table)
                .ok_or_else(|| RedDBError::NotFound(query.table.clone()))?;
            let entities = manager.query_all(|_| true);
            let recent: Vec<_> = entities
                .into_iter()
                .rev()
                .take(effective_rows.len())
                .collect();

            // Collector phase: (entity_index, combined_text) for rows that have non-empty fields.
            let entity_combos: Vec<(usize, String)> = recent
                .iter()
                .enumerate()
                .filter_map(|(i, entity)| {
                    if let EntityData::Row(ref row) = entity.data {
                        if let Some(ref named) = row.named {
                            let texts: Vec<String> = embed_config
                                .fields
                                .iter()
                                .filter_map(|field| match named.get(field) {
                                    Some(Value::Text(t)) if !t.is_empty() => Some(t.to_string()),
                                    _ => None,
                                })
                                .collect();
                            if !texts.is_empty() {
                                return Some((i, texts.join(" ")));
                            }
                        }
                    }
                    None
                })
                .collect();

            if !entity_combos.is_empty() {
                // Batch phase: single provider round-trip for all rows.
                let batch_texts: Vec<String> =
                    entity_combos.iter().map(|(_, t)| t.clone()).collect();

                // Issue #682 — when the provider is `local`, bypass
                // AiBatchClient (which is HTTP-only) and dispatch
                // directly through the in-process local embedding
                // backend. All texts go in one call, mirroring the
                // single-round-trip shape of the remote path. The
                // local backend does not perform intra-batch dedup —
                // each input position gets its own row in the output
                // — which keeps the per-row "create_vector" loop
                // below correct without additional fan-out logic.
                let embeddings = if is_local_provider {
                    let response = crate::runtime::ai::local_embedding::embed_local_with_db(
                        &self.inner.db,
                        &model,
                        batch_texts,
                    )?;
                    response.embeddings
                } else {
                    let batch_client =
                        crate::runtime::ai::batch_client::AiBatchClient::from_runtime(self);

                    match tokio::runtime::Handle::try_current() {
                        Ok(handle) => tokio::task::block_in_place(|| {
                            handle.block_on(batch_client.embed_batch(
                                &provider,
                                &model,
                                &api_key,
                                batch_texts,
                            ))
                        }),
                        Err(_) => {
                            return Err(RedDBError::Query(
                                "AUTO EMBED requires a Tokio runtime context".to_string(),
                            ));
                        }
                    }
                    .map_err(|e| RedDBError::Query(e.to_string()))?
                };

                // Distribute phase: persist one vector per non-empty embedding.
                for ((_, combined), dense) in entity_combos.iter().zip(embeddings) {
                    if dense.is_empty() {
                        continue;
                    }
                    self.create_vector(CreateVectorInput {
                        collection: query.table.clone(),
                        dense,
                        content: Some(combined.clone()),
                        metadata: Vec::new(),
                        link_row: None,
                        link_node: None,
                    })?;
                }
            }
        }

        if inserted_count > 0 {
            self.note_table_write(&query.table);
        }

        let mut result = RuntimeQueryResult::dml_result(
            raw_query.to_string(),
            inserted_count,
            "insert",
            "runtime-dml",
        );
        if let Some(returning) = returning_result {
            result.result = returning;
        }
        Ok(result)
    }

    fn check_insert_column_policy(&self, query: &InsertQuery) -> RedDBResult<()> {
        let Some(auth_store) = self.inner.auth_store.read().clone() else {
            return Ok(());
        };
        if !auth_store.iam_authorization_enabled() {
            return Ok(());
        }
        let Some((username, role)) = crate::runtime::impl_core::current_auth_identity() else {
            return Ok(());
        };

        let tenant = crate::runtime::impl_core::current_tenant();
        let principal = crate::auth::UserId::from_parts(tenant.as_deref(), &username);
        let request = crate::auth::ColumnAccessRequest {
            action: "insert".to_string(),
            schema: None,
            table: query.table.clone(),
            columns: query.columns.clone(),
        };
        let ctx = crate::auth::policies::EvalContext {
            principal_tenant: tenant.clone(),
            current_tenant: tenant,
            peer_ip: None,
            mfa_present: false,
            now_ms: crate::auth::now_ms(),
            principal_is_admin_role: role == crate::auth::Role::Admin,
            principal_is_platform_scoped: principal.tenant.is_none(),
        };

        let outcome = auth_store.check_column_projection_authz(&principal, &request, &ctx);
        let table_allowed = matches!(
            outcome.table_decision,
            crate::auth::policies::Decision::Allow { .. }
                | crate::auth::policies::Decision::AdminBypass
        );
        if !table_allowed {
            return Err(RedDBError::Query(format!(
                "principal=`{username}` action=`insert` resource=`{}:{}` denied by IAM policy",
                outcome.table_resource.kind, outcome.table_resource.name
            )));
        }
        if let Some(denied) = outcome.first_denied_column() {
            return Err(RedDBError::Query(format!(
                "principal=`{username}` action=`insert` resource=`{}:{}` denied by IAM policy",
                denied.resource.kind, denied.resource.name
            )));
        }

        Ok(())
    }

    pub(crate) fn insert_timeseries_point(
        &self,
        collection: &str,
        fields: Vec<(String, Value)>,
        mut metadata: Vec<(String, MetadataValue)>,
    ) -> RedDBResult<EntityId> {
        apply_collection_default_ttl_metadata(self, collection, &mut metadata);

        let (columns, values) = pairwise_columns_values(&fields);
        validate_timeseries_insert_columns(&columns)?;

        // Issue #577 — AnalyticsSchemaRegistry hook. If the row carries
        // an `event_name` whose schema is registered, validate the
        // `payload` JSON against it BEFORE any write side-effect. On
        // failure we return a typed error and the row is not
        // persisted. When no schema is registered for the event name
        // (or no `event_name` column is supplied at all) we fall
        // through to the normal write path for back-compat with
        // existing timeseries rows.
        let event_name_opt = find_column_value_opt_string(&columns, &values, "event_name");
        let payload_opt = find_column_value_opt_string(&columns, &values, "payload");
        if let Some(event_name) = event_name_opt.as_deref() {
            let store_for_schema = self.inner.db.store();
            if super::analytics_schema_registry::latest(store_for_schema.as_ref(), event_name)
                .is_some()
            {
                let payload_json = payload_opt.as_deref().unwrap_or("{}");
                super::analytics_schema_registry::validate(
                    store_for_schema.as_ref(),
                    event_name,
                    payload_json,
                )
                .map_err(super::analytics_schema_registry::validation_error_to_reddb)?;
            }
        }

        // `metric` is required by the existing timeseries write path;
        // when an analytics-style row supplies `event_name` but not
        // `metric`, fall back to the event name so the storage path
        // still has a non-empty metric tag.
        let metric = match find_column_value_opt_string(&columns, &values, "metric") {
            Some(m) => m,
            None => event_name_opt.clone().ok_or_else(|| {
                RedDBError::Query(
                    "timeseries INSERT requires either `metric` or `event_name`".to_string(),
                )
            })?,
        };
        // `value` is optional for analytics-event rows (which are
        // semantically counts of 1); default to 1.0 when missing so
        // analytics inserts don't have to fabricate a metric value.
        let value = match find_column_value_opt_string(&columns, &values, "value") {
            Some(s) => s.parse::<f64>().unwrap_or(1.0),
            None => columns
                .iter()
                .position(|c| c.eq_ignore_ascii_case("value"))
                .and_then(|i| match &values[i] {
                    Value::Float(f) => Some(*f),
                    Value::Integer(n) | Value::BigInt(n) => Some(*n as f64),
                    Value::UnsignedInteger(n) => Some(*n as f64),
                    _ => None,
                })
                .unwrap_or(1.0),
        };
        let timestamp_ns =
            find_timeseries_timestamp_ns(&columns, &values)?.unwrap_or_else(current_unix_ns);
        let mut tags = find_timeseries_tags(&columns, &values)?;
        if let Some(ref name) = event_name_opt {
            tags.entry("event_name".to_string())
                .or_insert_with(|| name.clone());
        }
        if let Some(ref payload) = payload_opt {
            tags.entry("payload".to_string())
                .or_insert_with(|| payload.clone());
        }

        let mut entity = UnifiedEntity::new(
            EntityId::new(0),
            EntityKind::TimeSeriesPoint(Box::new(crate::storage::TimeSeriesPointKind {
                series: collection.to_string(),
                metric: metric.clone(),
            })),
            EntityData::TimeSeries(crate::storage::TimeSeriesData {
                metric,
                timestamp_ns,
                value,
                tags,
            }),
        );
        // MVCC #30: stamp xmin with the active tx xid (inside a tx)
        // or an autocommit xid (allocated and committed up-front so
        // future snapshots see the row as soon as it lands).
        let writer_xid = match self.current_xid() {
            Some(xid) => xid,
            None => {
                let mgr = self.snapshot_manager();
                let xid = mgr.begin();
                mgr.commit(xid);
                xid
            }
        };
        entity.set_xmin(writer_xid);

        let store = self.inner.db.store();
        let id = store
            .insert_auto(collection, entity)
            .map_err(|err| RedDBError::Internal(err.to_string()))?;

        if !metadata.is_empty() {
            let _ = store.set_metadata(
                collection,
                id,
                Metadata::with_fields(metadata.into_iter().collect()),
            );
        }

        self.cdc_emit(
            crate::replication::cdc::ChangeOperation::Insert,
            collection,
            id.raw(),
            "timeseries",
        );

        Ok(id)
    }

    /// Execute UPDATE table SET col=val, ... WHERE filter
    ///
    /// Scans the target collection, evaluates the WHERE filter against each
    /// record, and patches every matching entity.
    pub fn execute_update(
        &self,
        raw_query: &str,
        query: &UpdateQuery,
    ) -> RedDBResult<RuntimeQueryResult> {
        self.check_write(crate::runtime::write_gate::WriteKind::Dml)?;
        // Issue #523 — blockchain collections are immutable. Reject before
        // RLS / RETURNING work so the operator sees a clean 409-mapped
        // error instead of a partially-applied mutation surface.
        if crate::runtime::blockchain_kind::is_chain(self.inner.db.store().as_ref(), &query.table) {
            return Err(RedDBError::InvalidOperation(format!(
                "BlockchainCollectionImmutable: UPDATE not allowed on '{}'",
                query.table
            )));
        }
        // Queue-shaped CLAIM (ADR 0020, #1609): a CLAIM on a queue collection
        // is a QueueLifecycle delivery acquisition, not a raw row UPDATE.
        // Route it through the lifecycle seam before the table-shaped
        // contract / RLS gates below (which would reject an UPDATE against a
        // queue's declared model) so QueueLifecycle stays the sole authority
        // for delivery state.
        if query.claim_limit.is_some() && self.is_queue_collection(&query.table) {
            return self.execute_queue_shaped_claim(raw_query, query);
        }
        // CollectionContract gate (#50): runs the APPEND ONLY guard
        // (and any future contract bits) before RLS / RETURNING work
        // so the operator's immutability declaration is honoured
        // uniformly and the error message points at the DDL rather
        // than at a downstream symptom.
        crate::runtime::collection_contract::CollectionContractGate::check(
            self,
            &query.table,
            crate::runtime::collection_contract::MutationKind::Update,
        )?;
        ensure_update_target_contract(self, &query.table, query.target)?;
        ensure_kv_key_update_target_allowed(query)?;
        ensure_graph_identity_update_target_allowed(query)?;

        // Apply RLS augmentation first so every downstream path — plain
        // UPDATE, UPDATE...RETURNING, the inner scan — observes the
        // same policy-filtered target set. This prevents RETURNING
        // from ever exposing rows the UPDATE policy would have
        // denied.
        let rls_gated = crate::runtime::impl_core::rls_is_enabled(self, &query.table);
        let augmented_query: UpdateQuery;
        let effective_query: &UpdateQuery = if rls_gated {
            let update_filter = crate::runtime::impl_core::rls_policy_filter(
                self,
                &query.table,
                crate::storage::query::ast::PolicyAction::Update,
            );
            let Some(mut policy) = update_filter else {
                // No admitting policy: zero rows affected, empty
                // RETURNING (never leak rows the caller can't touch).
                let mut response = RuntimeQueryResult::dml_result(
                    raw_query.to_string(),
                    0,
                    "update",
                    "runtime-dml-rls",
                );
                if let Some(items) = query.returning.clone() {
                    response.result = build_returning_result(&items, &[], None);
                }
                return Ok(response);
            };
            if query.claim_limit.is_some() {
                let read_filter = crate::runtime::impl_core::rls_policy_filter(
                    self,
                    &query.table,
                    crate::storage::query::ast::PolicyAction::Select,
                );
                let Some(read_policy) = read_filter else {
                    let mut response = RuntimeQueryResult::dml_result(
                        raw_query.to_string(),
                        0,
                        "update",
                        "runtime-dml-rls",
                    );
                    if let Some(items) = query.returning.clone() {
                        response.result = build_returning_result(&items, &[], None);
                    }
                    return Ok(response);
                };
                policy = crate::storage::query::ast::Filter::And(
                    Box::new(read_policy),
                    Box::new(policy),
                );
            }
            let mut augmented = query.clone();
            augmented.filter = Some(match augmented.filter.take() {
                Some(existing) => {
                    crate::storage::query::ast::Filter::And(Box::new(existing), Box::new(policy))
                }
                None => policy,
            });
            augmented_query = augmented;
            &augmented_query
        } else {
            query
        };

        // RETURNING wraps the inner executor and uses the touched-id
        // list the inner reports so the post-image reflects exactly
        // the rows the UPDATE actually mutated (not whatever a
        // separate SELECT might have observed).
        if let Some(items) = effective_query.returning.clone() {
            let mut inner_query = effective_query.clone();
            inner_query.returning = None;
            let (mut response, touched_ids) =
                self.execute_update_inner_tracked(raw_query, &inner_query)?;

            let mut snapshots = if matches!(
                effective_query.target,
                UpdateTarget::Nodes | UpdateTarget::Edges
            ) {
                graph_update_returning_snapshots(self, &effective_query.table, &touched_ids)
            } else {
                super::dml_target_scan::DmlTargetScan::new(self, &effective_query.table, None, None)
                    .row_snapshots(&touched_ids)
            };
            if matches!(effective_query.target, UpdateTarget::Kv) {
                restore_kv_returning_keys(
                    self,
                    &effective_query.table,
                    &touched_ids,
                    &mut snapshots,
                );
            }

            response.result = build_returning_result(&items, &snapshots, None);
            response.engine = "runtime-dml-returning";
            return Ok(response);
        }

        self.execute_update_inner(raw_query, effective_query)
    }

    /// Back-compat shim: the older entry point ignored touched ids.
    fn execute_update_inner(
        &self,
        raw_query: &str,
        query: &UpdateQuery,
    ) -> RedDBResult<RuntimeQueryResult> {
        self.execute_update_inner_tracked(raw_query, query)
            .map(|(res, _)| res)
    }

    /// Enforce the concurrent-claim ORDER BY index-gate (ADR 0063, #1607).
    ///
    /// A `CLAIM LIMIT n` / `CLAIM EXACT n` on a logical-identity model (tables
    /// and documents) must order its candidates through a compatible index; the
    /// planner rejects an ordering no index on the collection can serve rather
    /// than falling back to a broad write-path sort. KV (key identity) and graph
    /// nodes/edges are exempt — their claim identity is intrinsic, not a user
    /// ORDER BY column that needs a secondary index.
    fn enforce_claim_order_by_index_gate(&self, query: &UpdateQuery) -> RedDBResult<()> {
        if query.claim_limit.is_none()
            || !matches!(query.target, UpdateTarget::Rows | UpdateTarget::Documents)
        {
            return Ok(());
        }
        let available_indexes: Vec<Vec<String>> = self
            .index_store_ref()
            .list_indices(&query.table)
            .into_iter()
            .map(|index| index.columns)
            .collect();
        reddb_rql::planner::check_claim_order_by_index_gate(query, &available_indexes)
            .map_err(RedDBError::InvalidOperation)
    }

    fn execute_update_inner_tracked(
        &self,
        raw_query: &str,
        query: &UpdateQuery,
    ) -> RedDBResult<(RuntimeQueryResult, Vec<EntityId>)> {
        self.enforce_claim_order_by_index_gate(query)?;
        let store = self.inner.db.store();
        let effective_filter = effective_update_filter(query);
        let compiled_plan = self.compile_update_plan(query)?;
        let needs_rmw_lock = update_needs_rmw_lock(query);
        let claim_model = update_model_name(update_target_model(query.target));
        if query.claim_limit.is_some() {
            self.inner
                .claim_telemetry
                .record_attempt(&query.table, claim_model);
        }
        let claim_lock = query.claim_limit.map(|_| {
            self.inner
                .rmw_locks
                .lock_for(&query.table, "__table_claim_update__")
        });
        let _claim_guard = if let Some(lock) = claim_lock.as_ref() {
            let Some(guard) = lock.try_lock() else {
                let skipped_locked = query.claim_limit.unwrap_or(1);
                self.inner.claim_telemetry.record_skipped_locked(
                    &query.table,
                    claim_model,
                    skipped_locked,
                );
                tracing::debug!(
                    target: "reddb::claim",
                    collection = %query.table,
                    model = claim_model,
                    skipped_locked,
                    "concurrent claim skipped locked candidates"
                );
                return Ok((
                    RuntimeQueryResult::dml_result(
                        raw_query.to_string(),
                        0,
                        "update",
                        "runtime-dml",
                    ),
                    Vec::new(),
                ));
            };
            Some(guard)
        } else {
            None
        };
        let table_rmw_lock = if needs_rmw_lock {
            Some(
                self.inner
                    .rmw_locks
                    .lock_for(&query.table, "__table_rmw_update__"),
            )
        } else {
            None
        };
        let _table_rmw_guard = table_rmw_lock.as_ref().map(|lock| lock.lock());
        let mut touched_ids: Vec<EntityId> = Vec::new();
        let claim_cap = query.claim_limit.map(|limit| limit as usize);
        let limit_cap = claim_cap.or_else(|| query.limit.map(|limit| limit as usize));
        let manager = store
            .get_collection(&query.table)
            .ok_or_else(|| RedDBError::NotFound(query.table.clone()))?;
        let scan_limit = if query.order_by.is_empty() {
            limit_cap
        } else {
            None
        };
        let mut target_scan = super::dml_target_scan::DmlTargetScan::with_update_target(
            self,
            &query.table,
            effective_filter.as_ref(),
            scan_limit,
            query.target,
        );
        if needs_rmw_lock {
            target_scan = target_scan.with_live_table_rows();
        }
        let ids_to_update = target_scan.find_target_ids()?;
        let order_limit = if query.claim_limit.is_some() {
            None
        } else {
            limit_cap
        };
        let ids_to_update = if query.order_by.is_empty() {
            ids_to_update
        } else {
            ordered_update_target_ids(&manager, &ids_to_update, &query.order_by, order_limit)
        };
        let mut ids_to_update = if query.claim_limit.is_some() {
            self.filter_claim_locked_target_ids(&query.table, ids_to_update)
        } else {
            ids_to_update
        };
        if let Some(claim_cap) = claim_cap {
            ids_to_update.truncate(claim_cap);
        }
        if query.claim_exact
            && claim_cap.is_some_and(|claim_count| ids_to_update.len() < claim_count)
        {
            self.inner
                .claim_telemetry
                .record_miss(&query.table, claim_model);
            return Ok((
                RuntimeQueryResult::dml_result(raw_query.to_string(), 0, "update", "runtime-dml"),
                Vec::new(),
            ));
        }

        if needs_rmw_lock {
            let result = self.execute_update_inner_tracked_locked(
                raw_query,
                query,
                &compiled_plan,
                &ids_to_update,
                effective_filter.as_ref(),
            )?;
            if query.claim_limit.is_some() {
                self.record_pending_claim_locks_for_touched_ids(&query.table, &result.1);
            }
            record_claim_outcome(
                &self.inner.claim_telemetry,
                query.claim_limit,
                &query.table,
                claim_model,
                result.0.affected_rows,
            );
            return Ok(result);
        }

        let mut affected: u64 = 0;
        for chunk in ids_to_update.chunks(UPDATE_APPLY_CHUNK_SIZE) {
            let mut applied_chunk = Vec::with_capacity(chunk.len());
            for entity in manager.get_many(chunk).into_iter().flatten() {
                let assignments =
                    self.materialize_update_assignments_for_entity(query, &entity, &compiled_plan)?;
                let applied = self.apply_materialized_update_for_entity(
                    query,
                    entity,
                    &compiled_plan,
                    assignments,
                )?;
                touched_ids.push(applied.id);
                applied_chunk.push(applied);
            }
            self.persist_update_chunk(&applied_chunk)?;
            affected += applied_chunk.len() as u64;
            let lsns = self.flush_update_chunk(&applied_chunk)?;
            if !query.suppress_events {
                self.emit_update_events_for_collection(&query.table, &applied_chunk, &lsns)?;
            }
        }

        if affected > 0 {
            self.note_table_write(&query.table);
        }
        if query.claim_limit.is_some() {
            self.record_pending_claim_locks_for_touched_ids(&query.table, &touched_ids);
        }
        record_claim_outcome(
            &self.inner.claim_telemetry,
            query.claim_limit,
            &query.table,
            claim_model,
            affected,
        );

        Ok((
            RuntimeQueryResult::dml_result(
                raw_query.to_string(),
                affected,
                "update",
                "runtime-dml",
            ),
            touched_ids,
        ))
    }

    fn filter_claim_locked_target_ids(&self, table: &str, ids: Vec<EntityId>) -> Vec<EntityId> {
        let conn_id = current_connection_id();
        let locks = self.inner.pending_claim_locks.read();
        ids.into_iter()
            .filter(|id| {
                let Some(entity) = self.inner.db.store().get(table, *id) else {
                    return false;
                };
                let key = (table.to_string(), entity.logical_id());
                locks
                    .get(&key)
                    .is_none_or(|owner_conn_id| *owner_conn_id == conn_id)
            })
            .collect()
    }

    fn record_pending_claim_locks_for_touched_ids(&self, table: &str, ids: &[EntityId]) {
        if self.current_xid().is_none() || ids.is_empty() {
            return;
        }

        let conn_id = current_connection_id();
        let store = self.inner.db.store();
        let mut locks = self.inner.pending_claim_locks.write();
        for id in ids {
            let Some(entity) = store.get(table, *id) else {
                continue;
            };
            locks.insert((table.to_string(), entity.logical_id()), conn_id);
        }
    }

    fn execute_update_inner_tracked_locked(
        &self,
        raw_query: &str,
        query: &UpdateQuery,
        compiled_plan: &CompiledUpdatePlan,
        ids_to_update: &[EntityId],
        effective_filter: Option<&Filter>,
    ) -> RedDBResult<(RuntimeQueryResult, Vec<EntityId>)> {
        let store = self.inner.db.store();
        let mut touched_ids = Vec::new();
        let mut lock_entries = Vec::new();

        for id in ids_to_update {
            let Some(candidate) = store.get(&query.table, *id) else {
                continue;
            };
            let logical_id = candidate.logical_id();
            let lock_key = format!("row:{}", logical_id.raw());
            let rmw_lock = self.inner.rmw_locks.lock_for(&query.table, &lock_key);
            lock_entries.push((lock_key, logical_id, rmw_lock));
        }

        lock_entries.sort_by(|left, right| left.0.cmp(&right.0));
        lock_entries.dedup_by(|left, right| left.0 == right.0);
        let _rmw_guards: Vec<_> = lock_entries.iter().map(|entry| entry.2.lock()).collect();

        let mut applied_chunk = Vec::new();
        for (_, logical_id, _) in &lock_entries {
            let Some(entity) = resolve_update_entity_by_logical_id(self, &query.table, *logical_id)
            else {
                continue;
            };
            if let Some(filter) = effective_filter {
                if !crate::runtime::query_exec::evaluate_entity_filter_with_db(
                    Some(self.inner.db.as_ref()),
                    &entity,
                    filter,
                    &query.table,
                    &query.table,
                ) {
                    continue;
                }
            }

            let assignments =
                self.materialize_update_assignments_for_entity(query, &entity, compiled_plan)?;
            let applied = self.apply_materialized_update_for_entity(
                query,
                entity,
                compiled_plan,
                assignments,
            )?;
            touched_ids.push(applied.id);
            applied_chunk.push(applied);
        }

        let affected = applied_chunk.len() as u64;
        if !applied_chunk.is_empty() {
            self.persist_update_chunk(&applied_chunk)?;
            let lsns = self.flush_update_chunk(&applied_chunk)?;
            if !query.suppress_events {
                self.emit_update_events_for_collection(&query.table, &applied_chunk, &lsns)?;
            }
        }

        if affected > 0 {
            self.note_table_write(&query.table);
        }

        Ok((
            RuntimeQueryResult::dml_result(
                raw_query.to_string(),
                affected,
                "update",
                "runtime-dml",
            ),
            touched_ids,
        ))
    }

    fn compile_update_plan(&self, query: &UpdateQuery) -> RedDBResult<CompiledUpdatePlan> {
        let mut static_field_assignments = Vec::new();
        let mut static_metadata_assignments = Vec::new();
        let mut dynamic_assignments = Vec::new();
        let row_contract_plan = build_row_update_contract_plan(&self.db(), &query.table)?;
        let mut row_modified_columns = Vec::new();

        for (idx, (column, expr)) in query.assignment_exprs.iter().enumerate() {
            let compound_op = query.compound_assignment_ops.get(idx).copied().flatten();
            let metadata_key = resolve_sql_ttl_metadata_key(column);
            if compound_op.is_some() && metadata_key.is_some() {
                return Err(RedDBError::Query(format!(
                    "compound assignment is only supported for row fields: {column}"
                )));
            }
            if compound_op.is_none() {
                if let Ok(value) = fold_expr_to_value(expr.clone()) {
                    if let Some(metadata_key) = metadata_key {
                        let raw_value = sql_literal_to_metadata_value(metadata_key, &value)?;
                        let (canonical_key, canonical_value) =
                            canonicalize_sql_ttl_metadata(metadata_key, raw_value);
                        static_metadata_assignments
                            .push((canonical_key.to_string(), canonical_value));
                    } else {
                        let value = self.resolve_crypto_sentinel(value)?;
                        static_field_assignments.push((
                            column.clone(),
                            normalize_row_update_assignment_with_plan(
                                &query.table,
                                column,
                                value,
                                row_contract_plan.as_ref(),
                            )?,
                        ));
                        row_modified_columns.push(column.clone());
                    }
                    continue;
                }
            }

            dynamic_assignments.push(CompiledUpdateAssignment {
                column: column.clone(),
                expr: expr.clone(),
                compound_op,
                metadata_key,
                row_rule: if metadata_key.is_none() {
                    if let Some(plan) = row_contract_plan.as_ref() {
                        if plan.timestamps_enabled
                            && (column == "created_at" || column == "updated_at")
                        {
                            return Err(RedDBError::Query(format!(
                                "collection '{}' manages '{}' automatically — do not set it in UPDATE",
                                query.table, column
                            )));
                        }
                        if let Some(rule) = plan.declared_rules.get(column) {
                            Some(rule.clone())
                        } else if plan.strict_schema {
                            return Err(RedDBError::Query(format!(
                                "collection '{}' is strict and does not allow undeclared fields: {}",
                                query.table, column
                            )));
                        } else {
                            None
                        }
                    } else {
                        None
                    }
                } else {
                    None
                },
            });
            if metadata_key.is_none() {
                row_modified_columns.push(column.clone());
            }
        }

        let row_modified_columns = dedupe_update_columns(row_modified_columns);
        let row_touches_unique_columns = row_contract_plan.as_ref().is_some_and(|plan| {
            row_modified_columns.iter().any(|column| {
                plan.unique_columns
                    .keys()
                    .any(|unique| unique.eq_ignore_ascii_case(column))
            })
        });

        if let Some(ttl_ms) = query.ttl_ms {
            static_metadata_assignments
                .push(("_ttl_ms".to_string(), metadata_u64_to_value(ttl_ms)));
        }
        if let Some(expires_at_ms) = query.expires_at_ms {
            static_metadata_assignments.push((
                "_expires_at".to_string(),
                metadata_u64_to_value(expires_at_ms),
            ));
        }
        for (key, val) in &query.with_metadata {
            static_metadata_assignments.push((key.clone(), storage_value_to_metadata_value(val)));
        }

        Ok(CompiledUpdatePlan {
            static_field_assignments,
            static_metadata_assignments,
            dynamic_assignments,
            row_contract_plan,
            row_modified_columns,
            row_touches_unique_columns,
        })
    }

    fn materialize_update_assignments_for_entity(
        &self,
        query: &UpdateQuery,
        entity: &UnifiedEntity,
        compiled_plan: &CompiledUpdatePlan,
    ) -> RedDBResult<MaterializedUpdateAssignments> {
        let mut assignments = MaterializedUpdateAssignments::default();
        let mut record: Option<UnifiedRecord> = None;

        for assignment in &compiled_plan.dynamic_assignments {
            if assignment.compound_op.is_some()
                && !matches!(
                    entity.data,
                    EntityData::Row(_) | EntityData::Node(_) | EntityData::Edge(_)
                )
            {
                return Err(RedDBError::Query(format!(
                    "compound assignment is only supported for row or graph UPDATE column '{}'",
                    assignment.column
                )));
            }
            if record.is_none() {
                record = runtime_any_record_from_entity_ref(entity);
            }
            let Some(record) = record.as_ref() else {
                return Err(RedDBError::Query(format!(
                    "UPDATE could not materialize runtime record for entity {} in '{}'",
                    entity.id.raw(),
                    query.table
                )));
            };
            let rhs = super::expr_eval::evaluate_runtime_expr_with_db(
                Some(self.inner.db.as_ref()),
                &assignment.expr,
                record,
                Some(query.table.as_str()),
                Some(query.table.as_str()),
            )
            .ok_or_else(|| {
                RedDBError::Query(format!(
                    "failed to evaluate UPDATE expression for column '{}'",
                    assignment.column
                ))
            })?;
            let value = if let Some(op) = assignment.compound_op {
                evaluate_compound_update_assignment(&assignment.column, record, op, rhs)?
            } else {
                rhs
            };

            if let Some(metadata_key) = assignment.metadata_key {
                let raw_value = sql_literal_to_metadata_value(metadata_key, &value)?;
                let (canonical_key, canonical_value) =
                    canonicalize_sql_ttl_metadata(metadata_key, raw_value);
                assignments
                    .dynamic_metadata_assignments
                    .push((canonical_key.to_string(), canonical_value));
            } else {
                assignments.dynamic_field_assignments.push((
                    assignment.column.clone(),
                    normalize_row_update_value_for_rule(
                        &query.table,
                        self.resolve_crypto_sentinel(value)?,
                        assignment.row_rule.as_ref(),
                    )?,
                ));
            }
        }

        Ok(assignments)
    }

    fn apply_materialized_update_for_entity(
        &self,
        query: &UpdateQuery,
        entity: UnifiedEntity,
        compiled_plan: &CompiledUpdatePlan,
        mut assignments: MaterializedUpdateAssignments,
    ) -> RedDBResult<AppliedEntityMutation> {
        if matches!(query.target, UpdateTarget::Kv)
            && !compiled_plan
                .static_field_assignments
                .iter()
                .any(|(column, _)| column.eq_ignore_ascii_case("key"))
            && !assignments
                .dynamic_field_assignments
                .iter()
                .any(|(column, _)| column.eq_ignore_ascii_case("key"))
        {
            if let Some(key) = runtime_any_record_from_entity_ref(&entity)
                .and_then(|record| record.get("key").cloned())
            {
                assignments
                    .dynamic_field_assignments
                    .push(("key".to_string(), key));
            }
        }
        if matches!(entity.data, EntityData::Row(_)) {
            return self.apply_loaded_sql_update_row_core(
                query.table.clone(),
                entity,
                &compiled_plan.static_field_assignments,
                assignments.dynamic_field_assignments,
                &compiled_plan.static_metadata_assignments,
                assignments.dynamic_metadata_assignments,
                compiled_plan.row_contract_plan.as_ref(),
                &compiled_plan.row_modified_columns,
                compiled_plan.row_touches_unique_columns,
            );
        }

        ensure_graph_identity_update_allowed(&entity, compiled_plan, &assignments)?;

        let operations = build_patch_operations_from_materialized_assignments(
            &entity,
            compiled_plan,
            assignments,
        );
        self.apply_loaded_patch_entity_core(
            query.table.clone(),
            entity,
            crate::json::Value::Null,
            operations,
        )
    }

    /// Execute DELETE FROM table WHERE filter
    pub fn execute_delete(
        &self,
        raw_query: &str,
        query: &DeleteQuery,
    ) -> RedDBResult<RuntimeQueryResult> {
        self.check_write(crate::runtime::write_gate::WriteKind::Dml)?;
        // Issue #523 — blockchain collections are immutable; see
        // execute_update for the same gate.
        if crate::runtime::blockchain_kind::is_chain(self.inner.db.store().as_ref(), &query.table) {
            return Err(RedDBError::InvalidOperation(format!(
                "BlockchainCollectionImmutable: DELETE not allowed on '{}'",
                query.table
            )));
        }
        // CollectionContract gate (#50) — see execute_update for
        // rationale. The gate handles APPEND ONLY rejection and is
        // the single point where future contract bits land.
        crate::runtime::collection_contract::CollectionContractGate::check(
            self,
            &query.table,
            crate::runtime::collection_contract::MutationKind::Delete,
        )?;

        // RETURNING on DELETE: capture the pre-image via an internal
        // SELECT that reuses the same WHERE, then run the delete with
        // the RETURNING clause stripped, then project the captured
        // rows through the requested items. The extra SELECT is a
        // pragmatic MVP — a future pass can fuse the scan with the
        // delete to avoid the second pass over the heap.
        if let Some(items) = query.returning.clone() {
            let select_sql = delete_to_select_sql(raw_query).ok_or_else(|| {
                RedDBError::Query(
                    "DELETE ... RETURNING: cannot rewrite query for pre-image scan".to_string(),
                )
            })?;
            let captured = self.execute_query(&select_sql)?;

            let mut inner_query = query.clone();
            inner_query.returning = None;
            let _ = self.execute_delete(raw_query, &inner_query)?;

            let snapshots: Vec<Vec<(String, Value)>> = captured
                .result
                .records
                .iter()
                .map(|rec| {
                    rec.iter_fields()
                        .map(|(k, v)| (k.as_ref().to_string(), v.clone()))
                        .collect()
                })
                .collect();
            let affected = snapshots.len() as u64;
            let result = build_returning_result(&items, &snapshots, None);

            let mut response = RuntimeQueryResult::dml_result(
                raw_query.to_string(),
                affected,
                "delete",
                "runtime-dml-returning",
            );
            response.result = result;
            return Ok(response);
        }
        // Row-Level Security enforcement (Phase 2.5.2 PG parity).
        //
        // When the table has RLS enabled, gate the DELETE by the
        // per-role policy set: mutations only touch rows that *every*
        // matching `FOR DELETE` policy would accept. No policies =>
        // zero rows affected (PG restrictive-default).
        if crate::runtime::impl_core::rls_is_enabled(self, &query.table) {
            let rls_filter = crate::runtime::impl_core::rls_policy_filter(
                self,
                &query.table,
                crate::storage::query::ast::PolicyAction::Delete,
            );
            let Some(policy) = rls_filter else {
                return Ok(RuntimeQueryResult::dml_result(
                    raw_query.to_string(),
                    0,
                    "delete",
                    "runtime-dml-rls",
                ));
            };
            // Fold the policy predicate into the user's WHERE before
            // dispatching — the remainder of this function reads the
            // filter from `query` via `effective_delete_filter`, which
            // respects the updated value.
            let mut augmented = query.clone();
            augmented.filter = Some(match augmented.filter.take() {
                Some(existing) => {
                    crate::storage::query::ast::Filter::And(Box::new(existing), Box::new(policy))
                }
                None => policy,
            });
            return self.execute_delete_inner(raw_query, &augmented);
        }
        self.execute_delete_inner(raw_query, query)
    }

    fn execute_delete_inner(
        &self,
        raw_query: &str,
        query: &DeleteQuery,
    ) -> RedDBResult<RuntimeQueryResult> {
        let effective_filter = effective_delete_filter(query);

        // Find the rows that match the WHERE clause. The "find target
        // rows" loop lives in DmlTargetScan so UPDATE (#52) can reuse
        // the same scan strategy.
        let scan = super::dml_target_scan::DmlTargetScan::new(
            self,
            &query.table,
            effective_filter.as_ref(),
            None,
        );
        let ids_to_delete = scan.find_target_ids()?;

        // For event-enabled collections, snapshot the pre-delete state
        // before rows are physically removed.
        let needs_delete_events =
            !query.suppress_events && self.collection_has_delete_subscriptions(&query.table);
        let mut pre_images: HashMap<u64, crate::json::Value> = if needs_delete_events {
            scan.row_json_pre_images(&ids_to_delete)
        } else {
            HashMap::new()
        };

        let mut affected: u64 = 0;
        for chunk in ids_to_delete.chunks(UPDATE_APPLY_CHUNK_SIZE) {
            let (count, lsns) = self.delete_entities_batch(&query.table, chunk)?;
            affected += count;
            if needs_delete_events && !lsns.is_empty() {
                // lsns.len() == actually-deleted entities; align with chunk ids.
                // `delete_batch` may skip missing entities, so we correlate by
                // the number returned (they're emitted in chunk order).
                let deleted_chunk = &chunk[..lsns.len().min(chunk.len())];
                self.emit_delete_events_for_collection(
                    &query.table,
                    deleted_chunk,
                    &lsns,
                    &pre_images,
                )?;
            }
        }
        pre_images.clear();

        if affected > 0 {
            self.note_table_write(&query.table);
        }

        Ok(RuntimeQueryResult::dml_result(
            raw_query.to_string(),
            affected,
            "delete",
            "runtime-dml",
        ))
    }
}

/// Reject UPDATE … NODES/EDGES that assign to graph identity/topology
/// columns regardless of whether any row matches the WHERE clause. The
/// per-entity guard below covers only the matched-rows case, but ADR 0019
/// declares these columns immutable on the surface itself, so a zero-row
/// UPDATE should still surface the same error to operators and SDKs.
fn ensure_graph_identity_update_target_allowed(query: &UpdateQuery) -> RedDBResult<()> {
    if !matches!(query.target, UpdateTarget::Nodes | UpdateTarget::Edges) {
        return Ok(());
    }
    for (column, _) in &query.assignment_exprs {
        if is_immutable_graph_identity_field(column) {
            return Err(RedDBError::Query(format!(
                "immutable graph field '{column}' cannot be updated"
            )));
        }
    }
    Ok(())
}

fn ensure_kv_key_update_target_allowed(query: &UpdateQuery) -> RedDBResult<()> {
    if !matches!(query.target, UpdateTarget::Kv) {
        return Ok(());
    }
    for (column, _) in &query.assignment_exprs {
        if column.eq_ignore_ascii_case("key") {
            return Err(RedDBError::Query(
                "KV key cannot be updated; delete and insert a new key instead".to_string(),
            ));
        }
    }
    Ok(())
}

fn ensure_graph_identity_update_allowed(
    entity: &UnifiedEntity,
    compiled_plan: &CompiledUpdatePlan,
    assignments: &MaterializedUpdateAssignments,
) -> RedDBResult<()> {
    if !matches!(entity.data, EntityData::Node(_) | EntityData::Edge(_)) {
        return Ok(());
    }

    for (column, _) in compiled_plan
        .static_field_assignments
        .iter()
        .chain(assignments.dynamic_field_assignments.iter())
    {
        if is_immutable_graph_identity_field(column) {
            return Err(RedDBError::Query(format!(
                "immutable graph field '{column}' cannot be updated"
            )));
        }
    }

    Ok(())
}

fn is_immutable_graph_identity_field(column: &str) -> bool {
    ["rid", "label", "from_rid", "to_rid", "from", "to"]
        .iter()
        .any(|reserved| column.eq_ignore_ascii_case(reserved))
}

fn build_patch_operations_from_materialized_assignments(
    entity: &UnifiedEntity,
    compiled_plan: &CompiledUpdatePlan,
    assignments: MaterializedUpdateAssignments,
) -> Vec<PatchEntityOperation> {
    let mut operations = Vec::with_capacity(
        compiled_plan.static_field_assignments.len()
            + compiled_plan.static_metadata_assignments.len()
            + assignments.dynamic_field_assignments.len()
            + assignments.dynamic_metadata_assignments.len(),
    );

    for (column, value) in &compiled_plan.static_field_assignments {
        operations.push(PatchEntityOperation {
            op: PatchEntityOperationType::Set,
            path: update_patch_path_for_entity(entity, column),
            value: Some(storage_value_to_json(value)),
        });
    }

    for (column, value) in assignments.dynamic_field_assignments {
        operations.push(PatchEntityOperation {
            op: PatchEntityOperationType::Set,
            path: update_patch_path_for_entity(entity, &column),
            value: Some(storage_value_to_json(&value)),
        });
    }

    for (key, value) in &compiled_plan.static_metadata_assignments {
        operations.push(PatchEntityOperation {
            op: PatchEntityOperationType::Set,
            path: vec!["metadata".to_string(), key.clone()],
            value: Some(metadata_value_to_json(value)),
        });
    }

    for (key, value) in assignments.dynamic_metadata_assignments {
        operations.push(PatchEntityOperation {
            op: PatchEntityOperationType::Set,
            path: vec!["metadata".to_string(), key],
            value: Some(metadata_value_to_json(&value)),
        });
    }

    operations
}

fn update_patch_path_for_entity(entity: &UnifiedEntity, column: &str) -> Vec<String> {
    if matches!(
        (&entity.kind, &entity.data),
        (
            crate::storage::EntityKind::GraphNode(_),
            EntityData::Node(_)
        )
    ) && column.eq_ignore_ascii_case("node_type")
    {
        return vec!["node_type".to_string()];
    }
    if matches!(
        (&entity.kind, &entity.data),
        (
            crate::storage::EntityKind::GraphEdge(_),
            EntityData::Edge(_)
        )
    ) && column.eq_ignore_ascii_case("weight")
    {
        return vec!["weight".to_string()];
    }
    vec!["fields".to_string(), column.to_string()]
}

/// Rewrite `DELETE FROM <table> [WHERE …] [RETURNING …]` as
/// `SELECT * FROM <table> [WHERE …]` so the delete executor can
/// capture the pre-image before actually removing the rows. Returns
/// `None` when the input does not start with `DELETE`.
///
/// Case-insensitive on the keywords. Preserves everything between
/// the table name and the RETURNING clause, so WHERE / ORDER BY /
/// LIMIT survive untouched. The RETURNING tail — if present — is
/// truncated at the first top-level `RETURNING` token.
fn delete_to_select_sql(sql: &str) -> Option<String> {
    let trimmed = sql.trim_start();
    let lowered = trimmed.to_ascii_lowercase();
    if !lowered.starts_with("delete ") && !lowered.starts_with("delete\t") {
        return None;
    }
    // Find `FROM` after DELETE.
    let from_idx = lowered.find(" from ")?;
    let after_from = &trimmed[from_idx + " from ".len()..];
    let after_from_lc = &lowered[from_idx + " from ".len()..];

    // Cut off the RETURNING tail (a naive search — the RETURNING
    // clause only appears once per statement at top level in our
    // grammar). Matches whitespace-bounded tokens to avoid clipping
    // `RETURNING` inside a string literal.
    let mut body = after_from.to_string();
    if let Some(pos) = find_top_level_keyword(after_from_lc, "returning") {
        body.truncate(pos);
    }
    Some(format!("SELECT * FROM {}", body.trim_end()))
}

/// Find the byte offset of a whitespace-bounded keyword in a
/// lowercased haystack, skipping matches inside single-quoted
/// string literals. Naive — no escape handling — but enough for
/// the shapes the DML parser emits.
fn find_top_level_keyword(haystack: &str, needle: &str) -> Option<usize> {
    let bytes = haystack.as_bytes();
    let nlen = needle.len();
    let mut i = 0usize;
    let mut in_string = false;
    while i < bytes.len() {
        let c = bytes[i];
        if c == b'\'' {
            in_string = !in_string;
            i += 1;
            continue;
        }
        if !in_string
            && i + nlen <= bytes.len()
            && &bytes[i..i + nlen] == needle.as_bytes()
            && (i == 0 || bytes[i - 1].is_ascii_whitespace())
            && (i + nlen == bytes.len() || bytes[i + nlen].is_ascii_whitespace())
        {
            return Some(i);
        }
        i += 1;
    }
    None
}

/// Build a `UnifiedResult` from the rows affected by a DML statement plus
/// its `RETURNING` clause. Each snapshot is a list of (column, value) pairs
/// for one affected row; `outputs`, when provided, supplies the engine-
/// assigned entity id for the same row (INSERT path). Projection honours
/// the RETURNING items: `*` expands to every snapshot column plus
/// the public row envelope when available.
fn build_returning_result(
    items: &[ReturningItem],
    snapshots: &[Vec<(String, Value)>],
    outputs: Option<&[CreateEntityOutput]>,
) -> UnifiedResult {
    let project_all = items.iter().any(|it| matches!(it, ReturningItem::All));
    let public_item_outputs = outputs.is_some_and(|outs| {
        outs.first()
            .and_then(|out| out.entity.as_ref())
            .is_some_and(|entity| public_returning_item_kind(entity).is_some())
    });

    let mut columns: Vec<String> = if project_all {
        let mut cols: Vec<String> = Vec::new();
        if public_item_outputs {
            cols.extend(
                [
                    "rid",
                    "collection",
                    "kind",
                    "tenant",
                    "created_at",
                    "updated_at",
                ]
                .into_iter()
                .map(str::to_string),
            );
        } else if outputs.is_some() {
            cols.push("rid".to_string());
        }
        if let Some(first) = snapshots.first() {
            for (name, _) in first {
                cols.push(name.clone());
            }
        }
        cols
    } else {
        items
            .iter()
            .filter_map(|it| match it {
                ReturningItem::Column(c) => Some(c.clone()),
                ReturningItem::All => None,
            })
            .collect()
    };
    // Guarantee unique order-preserving column list.
    {
        let mut seen = std::collections::HashSet::new();
        columns.retain(|c| seen.insert(c.clone()));
    }

    let mut records: Vec<UnifiedRecord> = Vec::with_capacity(snapshots.len());
    for (idx, snap) in snapshots.iter().enumerate() {
        let mut values: HashMap<Arc<str>, Value> = HashMap::with_capacity(columns.len());
        if let Some(outs) = outputs {
            if let Some(out) = outs.get(idx) {
                if let Some(entity) = out.entity.as_ref() {
                    if let Some(kind) = public_returning_item_kind(entity) {
                        values.insert(
                            Arc::clone(&sys_key_rid()),
                            Value::UnsignedInteger(out.id.raw()),
                        );
                        values.insert(
                            Arc::clone(&sys_key_collection()),
                            Value::text(entity.kind.collection().to_string()),
                        );
                        values.insert(Arc::clone(&sys_key_kind()), Value::text(kind.to_string()));
                        values.insert(
                            Arc::clone(&sys_key_created_at()),
                            Value::UnsignedInteger(entity.created_at),
                        );
                        values.insert(
                            Arc::clone(&sys_key_updated_at()),
                            Value::UnsignedInteger(entity.updated_at),
                        );
                    } else {
                        values.insert(
                            Arc::clone(&sys_key_rid()),
                            Value::Integer(out.id.raw() as i64),
                        );
                    }
                } else {
                    values.insert(
                        Arc::clone(&sys_key_rid()),
                        Value::Integer(out.id.raw() as i64),
                    );
                }
            }
        }
        for (name, val) in snap {
            values.insert(Arc::from(name.as_str()), val.clone());
        }
        if !values.contains_key("tenant") {
            let tenant = values.get("tenant_id").cloned().unwrap_or(Value::Null);
            values.insert(Arc::clone(&sys_key_tenant()), tenant);
        }
        let mut rec = UnifiedRecord::default();
        // Only keep projected columns on the record.
        for col in &columns {
            if let Some(v) = values.get(col.as_str()) {
                rec.set_arc(Arc::from(col.as_str()), v.clone());
            }
        }
        records.push(rec);
    }

    UnifiedResult {
        columns,
        records,
        stats: Default::default(),
        pre_serialized_json: None,
    }
}

fn public_returning_item_kind(entity: &crate::storage::UnifiedEntity) -> Option<&'static str> {
    match (&entity.kind, &entity.data) {
        (crate::storage::EntityKind::GraphNode(_), crate::storage::EntityData::Node(_)) => {
            Some("node")
        }
        (crate::storage::EntityKind::GraphEdge(_), crate::storage::EntityData::Edge(_)) => {
            Some("edge")
        }
        (_, crate::storage::EntityData::Row(_)) => Some(public_returning_row_kind(entity)),
        // #1369 — every entity model must expose its `rid` in RETURNING *.
        // Vectors carry their payload in `EntityData::Vector`, not `Row`, so
        // they were falling through to a no-rid envelope
        // and `RETURNING *` never surfaced the entity-id.
        (_, crate::storage::EntityData::Vector(_)) => Some("vector"),
        _ => None,
    }
}

fn public_returning_row_kind(entity: &crate::storage::UnifiedEntity) -> &'static str {
    let Some(row) = entity.data.as_row() else {
        return "row";
    };

    let is_kv = row.named.as_ref().is_some_and(|named| {
        (named.len() == 2 && named.contains_key("key") && named.contains_key("value"))
            || (named.len() == 1 && (named.contains_key("key") || named.contains_key("value")))
    });
    if is_kv {
        return "kv";
    }

    let is_document = row
        .named
        .as_ref()
        .is_some_and(|named| named.values().any(runtime_returning_documentish_value))
        || row.columns.iter().any(runtime_returning_documentish_value);
    if is_document {
        "document"
    } else {
        "row"
    }
}

fn runtime_returning_documentish_value(value: &Value) -> bool {
    matches!(value, Value::Json(_) | Value::Blob(_))
}

fn row_insert_returning_snapshots(
    outputs: &[CreateEntityOutput],
    fallback: Vec<Vec<(String, Value)>>,
) -> Vec<Vec<(String, Value)>> {
    outputs
        .iter()
        .enumerate()
        .map(|(idx, out)| {
            out.entity
                .as_ref()
                .map(entity_row_fields_snapshot)
                .filter(|snap| !snap.is_empty())
                .unwrap_or_else(|| fallback.get(idx).cloned().unwrap_or_default())
        })
        .collect()
}

fn graph_insert_returning_snapshots(
    store: &crate::storage::unified::UnifiedStore,
    collection: &str,
    ids: &[EntityId],
) -> Vec<Vec<(String, Value)>> {
    let Some(manager) = store.get_collection(collection) else {
        return Vec::new();
    };

    ids.iter()
        .filter_map(|id| manager.get(*id))
        .filter_map(|entity| {
            let mut record = runtime_any_record_from_entity_ref(&entity)?;
            record.set_arc(sys_key_collection(), Value::text(collection.to_string()));
            Some(record)
        })
        .map(|record| {
            record
                .iter_fields()
                .map(|(key, value)| (key.as_ref().to_string(), value.clone()))
                .collect()
        })
        .collect()
}

fn graph_update_returning_snapshots(
    runtime: &RedDBRuntime,
    collection: &str,
    ids: &[EntityId],
) -> Vec<Vec<(String, Value)>> {
    let store = runtime.db().store();
    let Some(manager) = store.get_collection(collection) else {
        return Vec::new();
    };

    manager
        .get_many(ids)
        .into_iter()
        .flatten()
        .filter_map(|entity| runtime_any_record_from_entity_ref(&entity))
        .map(|record| {
            record
                .iter_fields()
                .map(|(key, value)| (key.as_ref().to_string(), value.clone()))
                .collect()
        })
        .collect()
}

fn restore_kv_returning_keys(
    runtime: &RedDBRuntime,
    collection: &str,
    ids: &[EntityId],
    snapshots: &mut [Vec<(String, Value)>],
) {
    if ids.is_empty() || snapshots.is_empty() {
        return;
    }

    let store = runtime.db().store();
    for (idx, id) in ids.iter().enumerate() {
        let Some(snapshot) = snapshots.get_mut(idx) else {
            continue;
        };
        if snapshot.iter().any(|(name, _)| name == "key") {
            continue;
        }
        let Some(entity) = store.get(collection, *id) else {
            continue;
        };
        let logical_id = entity.logical_id();
        let key = store
            .table_row_versions_by_logical_id(collection, logical_id)
            .into_iter()
            .find_map(kv_key_value_from_entity)
            .or_else(|| kv_key_value_from_entity(entity));
        if let Some(key) = key {
            snapshot.push(("key".to_string(), key));
        }
    }
}

fn kv_key_value_from_entity(entity: UnifiedEntity) -> Option<Value> {
    let row = entity.data.as_row()?;
    row.get_field("key")
        .cloned()
        .or_else(|| row.columns.first().cloned())
}

fn ensure_update_target_contract(
    runtime: &RedDBRuntime,
    collection: &str,
    target: UpdateTarget,
) -> RedDBResult<()> {
    let Some(contract) = runtime.db().collection_contract(collection) else {
        return Ok(());
    };
    if update_target_contract_is_advisory(&contract)
        || update_target_allows_model(contract.declared_model, update_target_model(target))
    {
        return Ok(());
    }
    Err(RedDBError::InvalidOperation(format!(
        "collection '{}' is declared as '{}' and does not allow '{}' updates",
        collection,
        update_model_name(contract.declared_model),
        update_model_name(update_target_model(target))
    )))
}

fn update_target_contract_is_advisory(contract: &crate::physical::CollectionContract) -> bool {
    matches!(
        (&contract.origin, &contract.schema_mode),
        (
            crate::physical::ContractOrigin::Implicit,
            crate::catalog::SchemaMode::Dynamic,
        )
    )
}

fn update_target_model(target: UpdateTarget) -> crate::catalog::CollectionModel {
    match target {
        UpdateTarget::Rows => crate::catalog::CollectionModel::Table,
        UpdateTarget::Documents => crate::catalog::CollectionModel::Document,
        UpdateTarget::Kv => crate::catalog::CollectionModel::Kv,
        UpdateTarget::Nodes | UpdateTarget::Edges => crate::catalog::CollectionModel::Graph,
    }
}

fn update_target_allows_model(
    declared_model: crate::catalog::CollectionModel,
    requested_model: crate::catalog::CollectionModel,
) -> bool {
    declared_model == requested_model || declared_model == crate::catalog::CollectionModel::Mixed
}

fn update_model_name(model: crate::catalog::CollectionModel) -> &'static str {
    match model {
        crate::catalog::CollectionModel::Table => "table",
        crate::catalog::CollectionModel::Document => "document",
        crate::catalog::CollectionModel::Graph => "graph",
        crate::catalog::CollectionModel::Vector => "vector",
        crate::catalog::CollectionModel::Hll => "hll",
        crate::catalog::CollectionModel::Sketch => "sketch",
        crate::catalog::CollectionModel::Filter => "filter",
        crate::catalog::CollectionModel::Kv => "kv",
        crate::catalog::CollectionModel::Config => "config",
        crate::catalog::CollectionModel::Vault => "vault",
        crate::catalog::CollectionModel::Mixed => "mixed",
        crate::catalog::CollectionModel::TimeSeries => "timeseries",
        crate::catalog::CollectionModel::Queue => "queue",
        crate::catalog::CollectionModel::Metrics => "metrics",
    }
}

fn ensure_graph_insert_contract(runtime: &RedDBRuntime, collection: &str) -> RedDBResult<()> {
    let db = runtime.db();
    if let Some(contract) = db.collection_contract(collection) {
        let advisory_implicit_dynamic = matches!(
            (&contract.origin, &contract.schema_mode),
            (
                crate::physical::ContractOrigin::Implicit,
                crate::catalog::SchemaMode::Dynamic,
            )
        );
        if advisory_implicit_dynamic
            || matches!(
                contract.declared_model,
                crate::catalog::CollectionModel::Graph | crate::catalog::CollectionModel::Mixed
            )
        {
            return Ok(());
        }
        return Err(RedDBError::InvalidOperation(format!(
            "collection '{}' is declared as '{:?}' and does not allow 'Graph' writes",
            collection, contract.declared_model
        )));
    }

    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis();
    db.save_collection_contract(crate::physical::CollectionContract {
        name: collection.to_string(),
        declared_model: crate::catalog::CollectionModel::Graph,
        schema_mode: crate::catalog::SchemaMode::Dynamic,
        origin: crate::physical::ContractOrigin::Implicit,
        version: 1,
        created_at_unix_ms: now,
        updated_at_unix_ms: now,
        default_ttl_ms: db.collection_default_ttl_ms(collection),
        vector_dimension: None,
        vector_metric: None,
        context_index_fields: Vec::new(),
        declared_columns: Vec::new(),
        table_def: None,
        timestamps_enabled: false,
        context_index_enabled: false,
        metrics_raw_retention_ms: None,
        metrics_rollup_policies: Vec::new(),
        metrics_tenant_identity: None,
        metrics_namespace: None,
        append_only: false,
        subscriptions: Vec::new(),
        analytics_config: Vec::new(),
        session_key: None,
        session_gap_ms: None,
        retention_duration_ms: None,
        analytical_storage: None,

        ai_policy: None,
    })
    .map(|_| ())
    .map_err(|err| RedDBError::Internal(err.to_string()))
}

fn update_needs_rmw_lock(query: &UpdateQuery) -> bool {
    query
        .assignment_exprs
        .iter()
        .enumerate()
        .any(|(idx, (column, expr))| {
            query
                .compound_assignment_ops
                .get(idx)
                .is_some_and(|op| op.is_some())
                || expr_references_update_column(expr, &query.table, column)
        })
}

fn record_claim_outcome(
    telemetry: &crate::runtime::claim_telemetry::ClaimTelemetryCounters,
    claim_limit: Option<u64>,
    table: &str,
    model: &str,
    affected: u64,
) {
    if claim_limit.is_none() {
        return;
    }
    if affected == 0 {
        telemetry.record_miss(table, model);
        tracing::debug!(
            target: "reddb::claim",
            collection = table,
            model,
            "concurrent claim missed"
        );
    } else {
        telemetry.record_successful(table, model, affected);
        tracing::debug!(
            target: "reddb::claim",
            collection = table,
            model,
            successful = affected,
            "concurrent claim succeeded"
        );
    }
}

fn evaluate_compound_update_assignment(
    column: &str,
    record: &UnifiedRecord,
    op: BinOp,
    rhs: Value,
) -> RedDBResult<Value> {
    let lhs = record.get(column).ok_or_else(|| {
        RedDBError::Query(format!(
            "compound assignment requires existing numeric field '{column}'"
        ))
    })?;
    if matches!(lhs, Value::Null) {
        return Err(RedDBError::Query(format!(
            "compound assignment requires non-null numeric field '{column}'"
        )));
    }
    apply_compound_numeric_op(column, op, lhs, &rhs)
}

fn apply_compound_numeric_op(
    column: &str,
    op: BinOp,
    lhs: &Value,
    rhs: &Value,
) -> RedDBResult<Value> {
    let Some(lhs_number) = CompoundNumber::from_value(lhs) else {
        return Err(RedDBError::Query(format!(
            "compound assignment requires numeric field '{column}'"
        )));
    };
    let Some(rhs_number) = CompoundNumber::from_value(rhs) else {
        return Err(RedDBError::Query(format!(
            "compound assignment requires numeric right-hand value for field '{column}'"
        )));
    };

    if lhs_number.is_float() || rhs_number.is_float() || matches!(op, BinOp::Div) {
        let a = lhs_number.as_f64();
        let b = rhs_number.as_f64();
        let out = match op {
            BinOp::Add => a + b,
            BinOp::Sub => a - b,
            BinOp::Mul => a * b,
            BinOp::Div => {
                if b == 0.0 {
                    return Err(RedDBError::Query(format!(
                        "division by zero in compound assignment for field '{column}'"
                    )));
                }
                a / b
            }
            BinOp::Mod => {
                if b == 0.0 {
                    return Err(RedDBError::Query(format!(
                        "modulo by zero in compound assignment for field '{column}'"
                    )));
                }
                a % b
            }
            _ => {
                return Err(RedDBError::Query(format!(
                    "unsupported compound assignment operator for field '{column}'"
                )));
            }
        };
        if !out.is_finite() {
            return Err(RedDBError::Query(format!(
                "numeric overflow in compound assignment for field '{column}'"
            )));
        }
        return Ok(Value::Float(out));
    }

    let a = lhs_number.as_i128();
    let b = rhs_number.as_i128();
    let out = match op {
        BinOp::Add => a.checked_add(b),
        BinOp::Sub => a.checked_sub(b),
        BinOp::Mul => a.checked_mul(b),
        BinOp::Mod => {
            if b == 0 {
                return Err(RedDBError::Query(format!(
                    "modulo by zero in compound assignment for field '{column}'"
                )));
            }
            a.checked_rem(b)
        }
        BinOp::Div => unreachable!("integer division is handled by the float branch"),
        _ => None,
    }
    .ok_or_else(|| {
        RedDBError::Query(format!(
            "numeric overflow in compound assignment for field '{column}'"
        ))
    })?;

    if matches!(lhs, Value::UnsignedInteger(_)) {
        let value = u64::try_from(out).map_err(|_| {
            RedDBError::Query(format!(
                "numeric overflow in compound assignment for field '{column}'"
            ))
        })?;
        Ok(Value::UnsignedInteger(value))
    } else {
        let value = i64::try_from(out).map_err(|_| {
            RedDBError::Query(format!(
                "numeric overflow in compound assignment for field '{column}'"
            ))
        })?;
        Ok(Value::Integer(value))
    }
}

#[derive(Clone, Copy)]
enum CompoundNumber {
    Integer(i128),
    Float(f64),
}

impl CompoundNumber {
    fn from_value(value: &Value) -> Option<Self> {
        match value {
            Value::Integer(value) | Value::BigInt(value) => Some(Self::Integer(*value as i128)),
            Value::UnsignedInteger(value) => Some(Self::Integer(*value as i128)),
            Value::Float(value) => value.is_finite().then_some(Self::Float(*value)),
            Value::Decimal(value) => Some(Self::Float(*value as f64 / 10_000.0)),
            _ => None,
        }
    }

    fn is_float(self) -> bool {
        matches!(self, Self::Float(_))
    }

    fn as_f64(self) -> f64 {
        match self {
            Self::Integer(value) => value as f64,
            Self::Float(value) => value,
        }
    }

    fn as_i128(self) -> i128 {
        match self {
            Self::Integer(value) => value,
            Self::Float(_) => unreachable!("float compound number used as integer"),
        }
    }
}

fn expr_references_update_column(expr: &Expr, table_name: &str, target_column: &str) -> bool {
    match expr {
        Expr::Literal { .. } | Expr::Parameter { .. } | Expr::Subquery { .. } => false,
        Expr::Column { field, .. } => {
            field_ref_matches_update_column(field, table_name, target_column)
        }
        Expr::BinaryOp { lhs, rhs, .. } => {
            expr_references_update_column(lhs, table_name, target_column)
                || expr_references_update_column(rhs, table_name, target_column)
        }
        Expr::UnaryOp { operand, .. } | Expr::Cast { inner: operand, .. } => {
            expr_references_update_column(operand, table_name, target_column)
        }
        Expr::FunctionCall { args, .. } => args
            .iter()
            .any(|arg| expr_references_update_column(arg, table_name, target_column)),
        Expr::Case {
            branches, else_, ..
        } => {
            branches.iter().any(|(cond, value)| {
                expr_references_update_column(cond, table_name, target_column)
                    || expr_references_update_column(value, table_name, target_column)
            }) || else_
                .as_deref()
                .is_some_and(|expr| expr_references_update_column(expr, table_name, target_column))
        }
        Expr::IsNull { operand, .. } => {
            expr_references_update_column(operand, table_name, target_column)
        }
        Expr::InList { target, values, .. } => {
            expr_references_update_column(target, table_name, target_column)
                || values
                    .iter()
                    .any(|value| expr_references_update_column(value, table_name, target_column))
        }
        Expr::Between {
            target, low, high, ..
        } => {
            expr_references_update_column(target, table_name, target_column)
                || expr_references_update_column(low, table_name, target_column)
                || expr_references_update_column(high, table_name, target_column)
        }
        Expr::WindowFunctionCall { args, window, .. } => {
            args.iter()
                .any(|arg| expr_references_update_column(arg, table_name, target_column))
                || window
                    .partition_by
                    .iter()
                    .any(|e| expr_references_update_column(e, table_name, target_column))
                || window
                    .order_by
                    .iter()
                    .any(|o| expr_references_update_column(&o.expr, table_name, target_column))
        }
    }
}

fn field_ref_matches_update_column(
    field: &FieldRef,
    table_name: &str,
    target_column: &str,
) -> bool {
    match field {
        FieldRef::TableColumn { table, column } => {
            column.eq_ignore_ascii_case(target_column)
                && (table.is_empty() || table.eq_ignore_ascii_case(table_name))
        }
        FieldRef::NodeProperty { .. } | FieldRef::EdgeProperty { .. } | FieldRef::NodeId { .. } => {
            false
        }
    }
}

fn resolve_update_entity_by_logical_id(
    runtime: &RedDBRuntime,
    table: &str,
    logical_id: EntityId,
) -> Option<UnifiedEntity> {
    let store = runtime.inner.db.store();
    if let Some(entity) = store.get_table_row_by_logical_id(table, logical_id) {
        return Some(entity);
    }
    // Fallback for non-table-row entities (graph nodes/edges, etc.) where
    // entity_id == logical_id and the MVCC table-row resolver doesn't apply.
    store.get(table, logical_id)
}

fn update_cdc_item_kind(
    runtime: &RedDBRuntime,
    collection: &str,
    entity: &UnifiedEntity,
) -> &'static str {
    match &entity.data {
        EntityData::Node(_) => return "node",
        EntityData::Edge(_) => return "edge",
        _ => {}
    }

    match runtime
        .db()
        .collection_contract(collection)
        .map(|contract| contract.declared_model)
    {
        Some(crate::catalog::CollectionModel::Document) => "document",
        Some(crate::catalog::CollectionModel::Kv)
        | Some(crate::catalog::CollectionModel::Vault) => "kv",
        _ => "row",
    }
}

fn ordered_update_target_ids(
    manager: &Arc<crate::storage::SegmentManager>,
    entity_ids: &[EntityId],
    order_by: &[OrderByClause],
    limit: Option<usize>,
) -> Vec<EntityId> {
    let mut entities: Vec<UnifiedEntity> =
        manager.get_many(entity_ids).into_iter().flatten().collect();
    entities.sort_by(|left, right| compare_update_order(left, right, order_by));
    if let Some(limit) = limit {
        entities.truncate(limit);
    }
    entities.into_iter().map(|entity| entity.id).collect()
}

fn compare_update_order(
    left: &UnifiedEntity,
    right: &UnifiedEntity,
    order_by: &[OrderByClause],
) -> Ordering {
    for clause in order_by {
        let left_value = update_order_value(left, &clause.field);
        let right_value = update_order_value(right, &clause.field);
        let ordering = compare_update_order_values(
            left_value.as_ref(),
            right_value.as_ref(),
            clause.nulls_first,
        );
        if ordering != Ordering::Equal {
            return if clause.ascending {
                ordering
            } else {
                ordering.reverse()
            };
        }
    }
    left.logical_id().raw().cmp(&right.logical_id().raw())
}

fn compare_update_order_values(
    left: Option<&Value>,
    right: Option<&Value>,
    nulls_first: bool,
) -> Ordering {
    match (left, right) {
        (None, None) => Ordering::Equal,
        (None, Some(_)) => {
            if nulls_first {
                Ordering::Less
            } else {
                Ordering::Greater
            }
        }
        (Some(_), None) => {
            if nulls_first {
                Ordering::Greater
            } else {
                Ordering::Less
            }
        }
        (Some(left), Some(right)) => {
            crate::storage::query::value_compare::total_compare_values(left, right)
        }
    }
}

fn update_order_value(entity: &UnifiedEntity, field: &FieldRef) -> Option<Value> {
    let FieldRef::TableColumn { table, column } = field else {
        return None;
    };
    if !table.is_empty() {
        return None;
    }
    if column.eq_ignore_ascii_case("rid") {
        return Some(Value::UnsignedInteger(entity.logical_id().raw()));
    }
    match &entity.data {
        // After the single-source binary-body cutover (ADR 0063) a DOCUMENT's
        // top-level fields live only inside the binary `body` container, not as
        // promoted row fields, so a direct `get_field` misses them and the
        // claim/UPDATE `ORDER BY <body-field>` would silently fall back to
        // insertion order. Mirror the filter read-seam: when the field isn't a
        // direct row field, offset-read it from the binary body.
        EntityData::Row(row) => {
            row.get_field(column)
                .cloned()
                .or_else(|| match row.get_field("body") {
                    Some(Value::Json(bytes)) => {
                        crate::document_body::read_body_field(bytes, column)
                    }
                    _ => None,
                })
        }
        EntityData::Node(_) | EntityData::Edge(_) => runtime_any_record_from_entity_ref(entity)
            .and_then(|record| record.get(column).cloned()),
        _ => None,
    }
}

fn dedupe_update_columns(mut columns: Vec<String>) -> Vec<String> {
    if columns.is_empty() {
        return columns;
    }

    let mut unique = Vec::with_capacity(columns.len());
    for column in columns.drain(..) {
        if !unique
            .iter()
            .any(|existing: &String| existing.eq_ignore_ascii_case(&column))
        {
            unique.push(column);
        }
    }
    unique
}

// =============================================================================
// Helper functions for extracting typed values from column/value pairs
// =============================================================================

const SQL_TTL_METADATA_COLUMNS: [&str; 3] = ["_ttl", "_ttl_ms", "_expires_at"];

fn resolve_sql_ttl_metadata_key(column: &str) -> Option<&'static str> {
    if column.eq_ignore_ascii_case("_ttl") {
        Some(SQL_TTL_METADATA_COLUMNS[0])
    } else if column.eq_ignore_ascii_case("_ttl_ms") {
        Some(SQL_TTL_METADATA_COLUMNS[1])
    } else if column.eq_ignore_ascii_case("_expires_at") {
        Some(SQL_TTL_METADATA_COLUMNS[2])
    } else {
        None
    }
}

/// Canonicalize a SQL TTL metadata `(key, value)` pair so the retention
/// sweeper sees a single key (`_ttl_ms`) regardless of which legacy form
/// the operator wrote. `_ttl` is scaled from seconds to milliseconds;
/// `_ttl_ms` and `_expires_at` are passed through.
fn canonicalize_sql_ttl_metadata(
    key: &'static str,
    value: MetadataValue,
) -> (&'static str, MetadataValue) {
    if key != "_ttl" {
        return (key, value);
    }
    let scaled = match value {
        MetadataValue::Int(s) => MetadataValue::Int(s.saturating_mul(1_000)),
        MetadataValue::Timestamp(ms_or_s) => {
            // Timestamp is already chosen for very large values; treat as
            // already-ms to avoid silent overflow.
            MetadataValue::Timestamp(ms_or_s)
        }
        MetadataValue::Float(f) => MetadataValue::Float(f * 1_000.0),
        other => other,
    };
    ("_ttl_ms", scaled)
}

/// Sentinel prefix produced by the parser for `PASSWORD('...')` and
/// `SECRET('...')` literals. The runtime strips this marker and
/// applies the actual crypto transform during INSERT execution.
pub(crate) const PLAINTEXT_SENTINEL: &str = "@@plain@@";

impl RedDBRuntime {
    /// Strip the plaintext sentinel from a `Value::Password` or
    /// `Value::Secret` produced by the parser and apply the real
    /// crypto transform. `Password` is always hashed with argon2id.
    /// `Secret` is encrypted with AES-256-GCM keyed by the vault
    /// when `red.config.secret.auto_encrypt = true` (default).
    pub(crate) fn resolve_crypto_sentinel(&self, value: Value) -> RedDBResult<Value> {
        match value {
            Value::Password(marked) => {
                if let Some(plain) = marked.strip_prefix(PLAINTEXT_SENTINEL) {
                    Ok(Value::Password(crate::auth::store::hash_password(plain)))
                } else {
                    Ok(Value::Password(marked))
                }
            }
            Value::Secret(bytes) => {
                if bytes.starts_with(PLAINTEXT_SENTINEL.as_bytes()) {
                    if !self.secret_auto_encrypt() {
                        return Err(RedDBError::Query(
                            "SECRET() literal rejected: red.config.secret.auto_encrypt \
                             is false. Insert pre-encrypted bytes directly instead."
                                .to_string(),
                        ));
                    }
                    let key = self.secret_aes_key().ok_or_else(|| {
                        RedDBError::Query(
                            "SECRET() column encryption requires a bootstrapped \
                             vault (red.secret.aes_key is missing). Start the server \
                             with --vault to enable."
                                .to_string(),
                        )
                    })?;
                    let plain = &bytes[PLAINTEXT_SENTINEL.len()..];
                    Ok(Value::Secret(encrypt_secret_payload(&key, plain)))
                } else {
                    Ok(Value::Secret(bytes))
                }
            }
            other => Ok(other),
        }
    }
}

/// Encode an AES-256-GCM ciphertext as `[12-byte nonce][ciphertext||tag]`.
/// This is the on-disk representation of `Value::Secret`.
fn encrypt_secret_payload(key: &[u8; 32], plaintext: &[u8]) -> Vec<u8> {
    let nonce_bytes = crate::auth::store::random_bytes(12);
    let mut nonce = [0u8; 12];
    nonce.copy_from_slice(&nonce_bytes[..12]);
    let ct = crate::crypto::aes_gcm::aes256_gcm_encrypt(key, &nonce, b"reddb.secret", plaintext);
    let mut out = Vec::with_capacity(12 + ct.len());
    out.extend_from_slice(&nonce);
    out.extend_from_slice(&ct);
    out
}

/// Decode a `Value::Secret` payload back to plaintext. Returns
/// `None` when the payload is too short or AES-GCM authentication
/// fails (tampered or wrong key).
pub(crate) fn decrypt_secret_payload(key: &[u8; 32], payload: &[u8]) -> Option<Vec<u8>> {
    if payload.len() < 12 {
        return None;
    }
    let mut nonce = [0u8; 12];
    nonce.copy_from_slice(&payload[..12]);
    crate::crypto::aes_gcm::aes256_gcm_decrypt(key, &nonce, b"reddb.secret", &payload[12..]).ok()
}

fn split_insert_metadata(
    runtime: &RedDBRuntime,
    columns: &[String],
    values: &[Value],
) -> RedDBResult<(Vec<(String, Value)>, Vec<(String, MetadataValue)>)> {
    let mut fields = Vec::new();
    let mut metadata = Vec::new();

    for (column, value) in columns.iter().zip(values.iter()) {
        // Still support legacy _ttl columns for backward compat
        if let Some(metadata_key) = resolve_sql_ttl_metadata_key(column) {
            let raw_value = sql_literal_to_metadata_value(metadata_key, value)?;
            let (canonical_key, canonical_value) =
                canonicalize_sql_ttl_metadata(metadata_key, raw_value);
            metadata.push((canonical_key.to_string(), canonical_value));
            continue;
        }
        fields.push((
            column.clone(),
            runtime.resolve_crypto_sentinel(value.clone())?,
        ));
    }

    Ok((fields, metadata))
}

/// Merge structured WITH TTL, WITH EXPIRES AT, and WITH METADATA clauses into metadata entries.
fn merge_with_clauses(
    metadata: &mut Vec<(String, MetadataValue)>,
    ttl_ms: Option<u64>,
    expires_at_ms: Option<u64>,
    with_metadata: &[(String, Value)],
) {
    if let Some(ms) = ttl_ms {
        metadata.push((
            "_ttl_ms".to_string(),
            if ms <= i64::MAX as u64 {
                MetadataValue::Int(ms as i64)
            } else {
                MetadataValue::Timestamp(ms)
            },
        ));
    }
    if let Some(ms) = expires_at_ms {
        metadata.push(("_expires_at".to_string(), MetadataValue::Timestamp(ms)));
    }
    for (key, value) in with_metadata {
        let meta_value = match value {
            Value::Text(s) => MetadataValue::String(s.to_string()),
            Value::Integer(n) => MetadataValue::Int(*n),
            Value::Float(n) => MetadataValue::Float(*n),
            Value::Boolean(b) => MetadataValue::Bool(*b),
            _ => MetadataValue::String(value.to_string()),
        };
        metadata.push((key.clone(), meta_value));
    }
}

fn merge_vector_metadata_column(
    metadata: &mut Vec<(String, MetadataValue)>,
    columns: &[String],
    values: &[Value],
) -> RedDBResult<()> {
    let Some(value) = columns
        .iter()
        .position(|column| column.eq_ignore_ascii_case("metadata"))
        .map(|index| &values[index])
    else {
        return Ok(());
    };
    let json = match value {
        Value::Null => return Ok(()),
        Value::Json(bytes) => crate::json::from_slice(bytes).map_err(|err| {
            RedDBError::Query(format!("column 'metadata' invalid JSON object: {err}"))
        })?,
        Value::Text(text) => crate::json::from_str(text).map_err(|err| {
            RedDBError::Query(format!("column 'metadata' invalid JSON object: {err}"))
        })?,
        other => {
            return Err(RedDBError::Query(format!(
                "column 'metadata' expected JSON object, got {other:?}"
            )))
        }
    };
    let parsed = metadata_from_json(&json)?;
    for (key, value) in parsed.iter() {
        metadata.push((key.clone(), value.clone()));
    }
    Ok(())
}

fn apply_collection_default_ttl_metadata(
    runtime: &RedDBRuntime,
    collection: &str,
    metadata: &mut Vec<(String, MetadataValue)>,
) {
    if has_internal_ttl_metadata(metadata) {
        return;
    }

    let Some(default_ttl_ms) = runtime.db().collection_default_ttl_ms(collection) else {
        return;
    };

    metadata.push((
        "_ttl_ms".to_string(),
        if default_ttl_ms <= i64::MAX as u64 {
            MetadataValue::Int(default_ttl_ms as i64)
        } else {
            MetadataValue::Timestamp(default_ttl_ms)
        },
    ));
}

fn ensure_non_tree_reserved_metadata_entries(
    metadata: &[(String, MetadataValue)],
) -> RedDBResult<()> {
    for (key, _) in metadata {
        ensure_non_tree_reserved_metadata_key(key)?;
    }
    Ok(())
}

fn ensure_non_tree_reserved_metadata_key(key: &str) -> RedDBResult<()> {
    if key.starts_with(TREE_METADATA_PREFIX) {
        return Err(RedDBError::Query(format!(
            "metadata key '{}' is reserved for managed trees",
            key
        )));
    }
    Ok(())
}

fn ensure_non_tree_structural_edge_label(label: &str) -> RedDBResult<()> {
    if label.eq_ignore_ascii_case(TREE_CHILD_EDGE_LABEL) {
        return Err(RedDBError::Query(format!(
            "edge label '{}' is reserved for managed trees",
            TREE_CHILD_EDGE_LABEL
        )));
    }
    Ok(())
}

fn pairwise_columns_values(pairs: &[(String, Value)]) -> (Vec<String>, Vec<Value>) {
    let mut columns = Vec::with_capacity(pairs.len());
    let mut values = Vec::with_capacity(pairs.len());

    for (column, value) in pairs {
        columns.push(column.clone());
        values.push(value.clone());
    }

    (columns, values)
}

/// Find a required column value and return it as-is.
fn find_column_value(columns: &[String], values: &[Value], name: &str) -> RedDBResult<Value> {
    for (i, col) in columns.iter().enumerate() {
        if col.eq_ignore_ascii_case(name) {
            return Ok(values[i].clone());
        }
    }
    Err(RedDBError::Query(format!(
        "required column '{name}' not found in INSERT"
    )))
}

/// Find a required column value and coerce to String.
fn find_column_value_string(
    columns: &[String],
    values: &[Value],
    name: &str,
) -> RedDBResult<String> {
    let val = find_column_value(columns, values, name)?;
    match val {
        Value::Text(s) => Ok(s.to_string()),
        Value::Integer(n) => Ok(n.to_string()),
        Value::Float(n) => Ok(n.to_string()),
        other => Err(RedDBError::Query(format!(
            "column '{name}' expected text, got {other:?}"
        ))),
    }
}

fn find_document_body_json(
    columns: &[String],
    values: &[Value],
) -> RedDBResult<crate::json::Value> {
    let val = find_column_value(columns, values, "body")?;
    match val {
        Value::Json(bytes) | Value::Blob(bytes) => {
            if let Some(body) = crate::document_body::decode_container_to_json(&bytes) {
                Ok(body)
            } else {
                crate::json::from_slice(&bytes)
                    .map_err(|err| RedDBError::Query(format!("invalid JSON body: {err}")))
            }
        }
        Value::Text(text) => crate::json::from_str(text.as_ref())
            .map_err(|err| RedDBError::Query(format!("invalid JSON body: {err}"))),
        Value::Integer(value) => crate::json::from_str(&value.to_string())
            .map_err(|err| RedDBError::Query(format!("invalid JSON body: {err}"))),
        Value::UnsignedInteger(value) => crate::json::from_str(&value.to_string())
            .map_err(|err| RedDBError::Query(format!("invalid JSON body: {err}"))),
        Value::Float(value) => crate::json::from_str(&value.to_string())
            .map_err(|err| RedDBError::Query(format!("invalid JSON body: {err}"))),
        other => Err(RedDBError::Query(format!(
            "column 'body' expected JSON body, got {other:?}"
        ))),
    }
}

fn find_column_value_f64(columns: &[String], values: &[Value], name: &str) -> RedDBResult<f64> {
    let val = find_column_value(columns, values, name)?;
    match val {
        Value::Float(n) => Ok(n),
        Value::Integer(n) => Ok(n as f64),
        Value::UnsignedInteger(n) => Ok(n as f64),
        Value::Text(s) => s
            .parse::<f64>()
            .map_err(|_| RedDBError::Query(format!("column '{name}' expected number, got '{s}'"))),
        other => Err(RedDBError::Query(format!(
            "column '{name}' expected number, got {other:?}"
        ))),
    }
}

/// Find an optional column value as String.
fn find_column_value_opt_string(
    columns: &[String],
    values: &[Value],
    name: &str,
) -> Option<String> {
    for (i, col) in columns.iter().enumerate() {
        if col.eq_ignore_ascii_case(name) {
            return match &values[i] {
                Value::Null => None,
                Value::Text(s) => Some(s.to_string()),
                Value::Integer(n) => Some(n.to_string()),
                Value::Float(n) => Some(n.to_string()),
                _ => None,
            };
        }
    }
    None
}

/// Resolve an EDGE endpoint (`from`/`to`) to a numeric entity id.
///
/// Accepts integer literals, decimal strings, and node labels resolved via
/// the per-collection graph label index (same source of truth that
/// `GRAPH NEIGHBORHOOD` / `GRAPH TRAVERSE` use at query time). Ambiguous
/// labels error so callers can fall back to the numeric id form.
fn resolve_edge_endpoint(
    store: &crate::storage::unified::UnifiedStore,
    collection: &str,
    columns: &[String],
    values: &[Value],
    name: &str,
) -> RedDBResult<u64> {
    let val = find_column_value(columns, values, name)?;
    match val {
        Value::Integer(n) => Ok(n as u64),
        Value::UnsignedInteger(n) => Ok(n),
        Value::Text(s) => {
            if let Ok(n) = s.parse::<u64>() {
                return Ok(n);
            }
            let matches = store.lookup_graph_nodes_by_label_in(collection, &s);
            match matches.len() {
                0 => Err(RedDBError::Query(format!(
                    "column '{name}': no graph node with label '{s}' in collection '{collection}'"
                ))),
                1 => Ok(matches[0].raw()),
                n => Err(RedDBError::Query(format!(
                    "column '{name}': ambiguous label '{s}' matches {n} nodes in collection '{collection}'; use the numeric id"
                ))),
            }
        }
        other => Err(RedDBError::Query(format!(
            "column '{name}' expected integer or node label, got {other:?}"
        ))),
    }
}

fn resolve_edge_endpoint_any(
    store: &crate::storage::unified::UnifiedStore,
    collection: &str,
    columns: &[String],
    values: &[Value],
    names: &[&str],
) -> RedDBResult<u64> {
    for name in names {
        if columns
            .iter()
            .any(|column| column.eq_ignore_ascii_case(name))
        {
            return resolve_edge_endpoint(store, collection, columns, values, name);
        }
    }

    Err(RedDBError::Query(format!(
        "required column '{}' not found in INSERT",
        names.first().copied().unwrap_or("from_rid")
    )))
}

/// Find a required column value and coerce to u64.
fn find_column_value_u64(columns: &[String], values: &[Value], name: &str) -> RedDBResult<u64> {
    let val = find_column_value(columns, values, name)?;
    match val {
        Value::Integer(n) => Ok(n as u64),
        Value::UnsignedInteger(n) => Ok(n),
        Value::Text(s) => s
            .parse::<u64>()
            .map_err(|_| RedDBError::Query(format!("column '{name}' expected integer, got '{s}'"))),
        other => Err(RedDBError::Query(format!(
            "column '{name}' expected integer, got {other:?}"
        ))),
    }
}

/// Find an optional column value as f32.
fn find_column_value_f32_opt(columns: &[String], values: &[Value], name: &str) -> Option<f32> {
    for (i, col) in columns.iter().enumerate() {
        if col.eq_ignore_ascii_case(name) {
            return match &values[i] {
                Value::Float(n) => Some(*n as f32),
                Value::Integer(n) => Some(*n as f32),
                Value::Null => None,
                _ => None,
            };
        }
    }
    None
}

/// Find a required column value and coerce to Vec<f32> (from Value::Vector).
fn find_column_value_vec_f32(
    columns: &[String],
    values: &[Value],
    name: &str,
) -> RedDBResult<Vec<f32>> {
    let val = find_column_value(columns, values, name)?;
    match val {
        Value::Vector(v) => Ok(v),
        Value::Json(bytes) => {
            // Try to parse as JSON array of numbers
            let s = std::str::from_utf8(&bytes).map_err(|_| {
                RedDBError::Query(format!("column '{name}' contains invalid UTF-8"))
            })?;
            let arr: Vec<f32> = crate::json::from_str(s).map_err(|e| {
                RedDBError::Query(format!("column '{name}' invalid vector JSON: {e}"))
            })?;
            Ok(arr)
        }
        other => Err(RedDBError::Query(format!(
            "column '{name}' expected vector, got {other:?}"
        ))),
    }
}

fn find_column_value_vec_f32_any(
    columns: &[String],
    values: &[Value],
    names: &[&str],
) -> RedDBResult<Vec<f32>> {
    for name in names {
        if columns
            .iter()
            .any(|column| column.eq_ignore_ascii_case(name))
        {
            return find_column_value_vec_f32(columns, values, name);
        }
    }
    Err(RedDBError::Query(format!(
        "required vector column '{}' not found in INSERT",
        names.join("' or '")
    )))
}

/// Extract remaining properties (all columns not in the exclusion list).
fn extract_remaining_properties(
    columns: &[String],
    values: &[Value],
    exclude: &[&str],
) -> Vec<(String, Value)> {
    columns
        .iter()
        .zip(values.iter())
        .filter(|(col, _)| !exclude.iter().any(|e| col.eq_ignore_ascii_case(e)))
        .map(|(col, val)| (col.clone(), val.clone()))
        .collect()
}

fn validate_timeseries_insert_columns(columns: &[String]) -> RedDBResult<()> {
    let mut invalid = Vec::new();
    for column in columns {
        if !is_timeseries_insert_column(column) && resolve_sql_ttl_metadata_key(column).is_none() {
            invalid.push(column.clone());
        }
    }

    if invalid.is_empty() {
        Ok(())
    } else {
        Err(RedDBError::Query(format!(
            "timeseries INSERT only accepts metric, value, tags, timestamp, timestamp_ns, or time columns; got {}",
            invalid.join(", ")
        )))
    }
}

fn is_timeseries_insert_column(column: &str) -> bool {
    matches!(
        column.to_ascii_lowercase().as_str(),
        "metric"
            | "value"
            | "tags"
            | "timestamp"
            | "timestamp_ns"
            | "time"
            // Analytics-event extension (#577): an analytics row carries
            // an `event_name` + JSON `payload`. The payload is validated
            // against the AnalyticsSchemaRegistry inside
            // `insert_timeseries_point` before the row lands.
            | "event_name"
            | "payload"
    )
}

fn find_timeseries_timestamp_ns(columns: &[String], values: &[Value]) -> RedDBResult<Option<u64>> {
    let mut found = None;

    for alias in ["timestamp_ns", "timestamp", "time"] {
        for (index, column) in columns.iter().enumerate() {
            if !column.eq_ignore_ascii_case(alias) {
                continue;
            }

            if found.is_some() {
                return Err(RedDBError::Query(
                    "timeseries INSERT accepts only one timestamp column".to_string(),
                ));
            }

            found = Some(coerce_value_to_non_negative_u64(&values[index], alias)?);
        }
    }

    Ok(found)
}

fn find_timeseries_tags(
    columns: &[String],
    values: &[Value],
) -> RedDBResult<std::collections::HashMap<String, String>> {
    for (index, column) in columns.iter().enumerate() {
        if column.eq_ignore_ascii_case("tags") {
            return parse_timeseries_tags(&values[index]);
        }
    }
    Ok(std::collections::HashMap::new())
}

fn parse_timeseries_tags(value: &Value) -> RedDBResult<std::collections::HashMap<String, String>> {
    match value {
        Value::Null => Ok(std::collections::HashMap::new()),
        Value::Json(bytes) => parse_timeseries_tags_json(bytes),
        Value::Text(text) => parse_timeseries_tags_json(text.as_bytes()),
        other => Err(RedDBError::Query(format!(
            "timeseries tags must be a JSON object or JSON text, got {other:?}"
        ))),
    }
}

fn parse_timeseries_tags_json(
    bytes: &[u8],
) -> RedDBResult<std::collections::HashMap<String, String>> {
    let json: crate::json::Value = crate::json::from_slice(bytes)
        .map_err(|err| RedDBError::Query(format!("timeseries tags must be valid JSON: {err}")))?;

    let object = match json {
        crate::json::Value::Object(object) => object,
        other => {
            return Err(RedDBError::Query(format!(
                "timeseries tags must be a JSON object, got {other:?}"
            )))
        }
    };

    let mut tags = std::collections::HashMap::with_capacity(object.len());
    for (key, value) in object {
        tags.insert(key, json_tag_value_to_string(&value));
    }
    Ok(tags)
}

/// Encode a tag value for storage so the original JSON type can be
/// recovered on read (issue #543).
///
/// Time-series tags are stored as `HashMap<String, String>` on the
/// physical record (see [`crate::storage::TimeSeriesData`]) so that
/// the segment codec, WAL and gRPC mirrors don't need a new value
/// variant. To preserve the original JSON type across that
/// string-only channel we prepend the
/// [`crate::runtime::query_exec::TIMESERIES_TAG_JSON_PREFIX`] marker
/// and serialize the value as compact JSON text. The read paths
/// (`timeseries_tags_json_value` / `timeseries_tags_value`) detect
/// the marker, parse the suffix, and recover a real JSON value.
/// Tags written through other channels (Prometheus remote write,
/// metrics handlers, legacy on-disk data) lack the marker and are
/// returned as `JsonValue::String(raw)` exactly as before.
fn json_tag_value_to_string(value: &crate::json::Value) -> String {
    let mut buf = String::with_capacity(value.to_string_compact().len() + 1);
    buf.push(crate::runtime::query_exec::TIMESERIES_TAG_JSON_PREFIX);
    buf.push_str(&value.to_string_compact());
    buf
}

fn coerce_value_to_non_negative_u64(value: &Value, column: &str) -> RedDBResult<u64> {
    match value {
        Value::UnsignedInteger(value) => Ok(*value),
        Value::Integer(value) if *value >= 0 => Ok(*value as u64),
        Value::Float(value) if *value >= 0.0 => Ok(*value as u64),
        Value::Text(value) => value.parse::<u64>().map_err(|_| {
            RedDBError::Query(format!(
                "column '{column}' expected a non-negative integer timestamp, got '{value}'"
            ))
        }),
        other => Err(RedDBError::Query(format!(
            "column '{column}' expected a non-negative integer timestamp, got {other:?}"
        ))),
    }
}

fn current_unix_ns() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_nanos()
        .min(u128::from(u64::MAX)) as u64
}

fn metadata_value_to_json(value: &MetadataValue) -> crate::json::Value {
    use crate::json::{Map, Value as JV};
    match value {
        MetadataValue::Null => JV::Null,
        MetadataValue::Bool(value) => JV::Bool(*value),
        MetadataValue::Int(value) => JV::Number(*value as f64),
        MetadataValue::Float(value) => JV::Number(*value),
        MetadataValue::String(value) => JV::String(value.clone()),
        MetadataValue::Bytes(value) => JV::Array(
            value
                .iter()
                .map(|value| JV::Number(*value as f64))
                .collect(),
        ),
        MetadataValue::Timestamp(value) => JV::Number(*value as f64),
        MetadataValue::Array(values) => {
            JV::Array(values.iter().map(metadata_value_to_json).collect())
        }
        MetadataValue::Object(object) => {
            let entries = object
                .iter()
                .map(|(key, value)| (key.clone(), metadata_value_to_json(value)))
                .collect();
            JV::Object(entries)
        }
        MetadataValue::Geo { lat, lon } => {
            let mut object = Map::new();
            object.insert("lat".to_string(), JV::Number(*lat));
            object.insert("lon".to_string(), JV::Number(*lon));
            JV::Object(object)
        }
        MetadataValue::Reference(target) => {
            let mut object = Map::new();
            object.insert(
                "collection".to_string(),
                JV::String(target.collection().to_string()),
            );
            object.insert(
                "entity_id".to_string(),
                JV::Number(target.entity_id().raw() as f64),
            );
            JV::Object(object)
        }
        MetadataValue::References(values) => {
            let refs = values
                .iter()
                .map(|target| {
                    let mut object = Map::new();
                    object.insert(
                        "collection".to_string(),
                        JV::String(target.collection().to_string()),
                    );
                    object.insert(
                        "entity_id".to_string(),
                        JV::Number(target.entity_id().raw() as f64),
                    );
                    JV::Object(object)
                })
                .collect();
            JV::Array(refs)
        }
    }
}

fn storage_value_to_metadata_value(value: &Value) -> MetadataValue {
    match value {
        Value::Null => MetadataValue::Null,
        Value::Boolean(value) => MetadataValue::Bool(*value),
        Value::Integer(value) => MetadataValue::Int(*value),
        Value::UnsignedInteger(value) => metadata_u64_to_value(*value),
        Value::Float(value) => MetadataValue::Float(*value),
        Value::Text(value) => MetadataValue::String(value.to_string()),
        Value::Blob(value) => MetadataValue::Bytes(value.clone()),
        Value::Timestamp(value) => {
            if *value >= 0 {
                metadata_u64_to_value(*value as u64)
            } else {
                MetadataValue::Int(*value)
            }
        }
        Value::TimestampMs(value) => {
            if *value >= 0 {
                metadata_u64_to_value(*value as u64)
            } else {
                MetadataValue::Int(*value)
            }
        }
        Value::Json(value) => MetadataValue::String(String::from_utf8_lossy(value).into_owned()),
        Value::Uuid(value) => MetadataValue::String(format!("{value:?}")),
        Value::Date(value) => MetadataValue::String(value.to_string()),
        Value::Time(value) => MetadataValue::String(value.to_string()),
        Value::Decimal(value) => MetadataValue::String(value.to_string()),
        Value::Ipv4(value) => MetadataValue::String(format!(
            "{}.{}.{}.{}",
            (value >> 24) & 0xFF,
            (value >> 16) & 0xFF,
            (value >> 8) & 0xFF,
            value & 0xFF
        )),
        Value::Port(value) => MetadataValue::Int(i64::from(*value)),
        Value::Latitude(value) => MetadataValue::Float(*value as f64 / 1_000_000.0),
        Value::Longitude(value) => MetadataValue::Float(*value as f64 / 1_000_000.0),
        Value::GeoPoint(lat, lon) => MetadataValue::Geo {
            lat: *lat as f64 / 1_000_000.0,
            lon: *lon as f64 / 1_000_000.0,
        },
        Value::BigInt(value) => MetadataValue::String(value.to_string()),
        Value::TableRef(value) => MetadataValue::String(value.clone()),
        Value::PageRef(value) => MetadataValue::Int(*value as i64),
        Value::Password(value) => MetadataValue::String(value.clone()),
        Value::Array(values) => {
            MetadataValue::Array(values.iter().map(storage_value_to_metadata_value).collect())
        }
        _ => MetadataValue::String(value.to_string()),
    }
}

fn sql_literal_to_metadata_value(field: &str, value: &Value) -> RedDBResult<MetadataValue> {
    match value {
        Value::Null => Ok(MetadataValue::Null),
        Value::Integer(value) if *value >= 0 => Ok(metadata_u64_to_value(*value as u64)),
        Value::Integer(_) => Err(RedDBError::Query(format!(
            "column '{field}' must be non-negative for TTL metadata"
        ))),
        Value::UnsignedInteger(value) => Ok(metadata_u64_to_value(*value)),
        Value::Float(value) if value.is_finite() => {
            if value.fract().abs() >= f64::EPSILON {
                return Err(RedDBError::Query(format!(
                    "column '{field}' must be an integer (TTL metadata must be an integer)"
                )));
            }
            if *value < 0.0 {
                return Err(RedDBError::Query(format!(
                    "column '{field}' must be non-negative for TTL metadata"
                )));
            }
            if *value > u64::MAX as f64 {
                return Err(RedDBError::Query(format!(
                    "column '{field}' value is too large"
                )));
            }
            Ok(metadata_u64_to_value(*value as u64))
        }
        Value::Float(_) => Err(RedDBError::Query(format!(
            "column '{field}' must be a finite number"
        ))),
        Value::Text(value) => {
            let value = value.trim();
            if let Ok(value) = value.parse::<u64>() {
                Ok(metadata_u64_to_value(value))
            } else if let Ok(value) = value.parse::<i64>() {
                if value < 0 {
                    return Err(RedDBError::Query(format!(
                        "column '{field}' must be non-negative for TTL metadata"
                    )));
                }
                Ok(metadata_u64_to_value(value as u64))
            } else if let Ok(value) = value.parse::<f64>() {
                if !value.is_finite() {
                    return Err(RedDBError::Query(format!(
                        "column '{field}' must be a finite number"
                    )));
                }
                if value.fract().abs() >= f64::EPSILON {
                    return Err(RedDBError::Query(format!(
                        "column '{field}' must be an integer (TTL metadata must be an integer)"
                    )));
                }
                if value < 0.0 {
                    return Err(RedDBError::Query(format!(
                        "column '{field}' must be non-negative for TTL metadata"
                    )));
                }
                if value > u64::MAX as f64 {
                    return Err(RedDBError::Query(format!(
                        "column '{field}' value is too large"
                    )));
                }
                Ok(metadata_u64_to_value(value as u64))
            } else {
                Err(RedDBError::Query(format!(
                    "column '{field}' expects a numeric value for TTL metadata"
                )))
            }
        }
        _ => Err(RedDBError::Query(format!(
            "column '{field}' expects a numeric value for TTL metadata"
        ))),
    }
}

fn metadata_u64_to_value(value: u64) -> MetadataValue {
    if value <= i64::MAX as u64 {
        MetadataValue::Int(value as i64)
    } else {
        MetadataValue::Timestamp(value)
    }
}

/// Phase 2 PG parity: inspect a column value and return `true` when
/// the dotted `tail` path is already present under it. Used by the
/// tenant auto-fill so rows that already carry an explicit value
/// (bulk import, admin insert on behalf of a tenant) are not
/// double-stamped with the session's current_tenant().
fn dotted_tail_already_set(value: &Value, tail: &str) -> bool {
    let json = match value {
        Value::Null => return false,
        Value::Json(bytes) | Value::Blob(bytes) => {
            match crate::json::from_slice::<crate::json::Value>(bytes) {
                Ok(v) => v,
                Err(_) => return false,
            }
        }
        Value::Text(s) => {
            let trimmed = s.trim_start();
            if !(trimmed.starts_with('{') || trimmed.starts_with('[')) {
                return false;
            }
            match crate::json::from_str::<crate::json::Value>(s) {
                Ok(v) => v,
                Err(_) => return false,
            }
        }
        _ => return false,
    };
    let mut cursor = &json;
    for seg in tail.split('.') {
        match cursor {
            crate::json::Value::Object(map) => match map.iter().find(|(k, _)| *k == seg) {
                Some((_, v)) => cursor = v,
                None => return false,
            },
            _ => return false,
        }
    }
    !matches!(cursor, crate::json::Value::Null)
}

/// Phase 2 PG parity: take a column value (possibly Null / Text /
/// Json) and return a `Value::Json` with the dotted `tail` path set
/// to `tenant_id`. Preserves every pre-existing key.
///
/// Accepts:
/// * `Value::Null`  → fresh `{tail: tenant_id}` object
/// * `Value::Json(bytes)` → parse, navigate / create path, re-serialize
/// * `Value::text(s)` if `s` is valid JSON → same as Json
/// * anything else → error (user supplied a scalar where we need
///   a JSON container)
fn merge_dotted_tenant(current: Value, tail: &str, tenant_id: &str) -> RedDBResult<Value> {
    let mut root = match current {
        Value::Null => crate::json::Value::Object(Default::default()),
        Value::Json(bytes) | Value::Blob(bytes) => {
            crate::json::from_slice(&bytes).map_err(|err| {
                RedDBError::Query(format!(
                    "tenant auto-fill: root column is not valid JSON ({err})"
                ))
            })?
        }
        Value::Text(s) => {
            if s.trim().is_empty() {
                crate::json::Value::Object(Default::default())
            } else {
                crate::json::from_str::<crate::json::Value>(&s).map_err(|err| {
                    RedDBError::Query(format!(
                        "tenant auto-fill: text root is not valid JSON ({err})"
                    ))
                })?
            }
        }
        other => {
            return Err(RedDBError::Query(format!(
                "tenant auto-fill: root column must be JSON / NULL, got {other:?}"
            )));
        }
    };

    // Navigate path segments, creating intermediate objects on demand.
    let segments: Vec<&str> = tail.split('.').collect();
    let mut cursor: &mut crate::json::Value = &mut root;
    for (i, seg) in segments.iter().enumerate() {
        let is_last = i + 1 == segments.len();
        let map = match cursor {
            crate::json::Value::Object(m) => m,
            _ => {
                return Err(RedDBError::Query(format!(
                    "tenant auto-fill: segment '{seg}' is not inside an object"
                )));
            }
        };
        if is_last {
            map.insert(
                seg.to_string(),
                crate::json::Value::String(tenant_id.to_string()),
            );
            break;
        }
        cursor = map
            .entry(seg.to_string())
            .or_insert_with(|| crate::json::Value::Object(Default::default()));
    }

    let bytes = crate::json::to_vec(&root).map_err(|err| {
        RedDBError::Query(format!(
            "tenant auto-fill: failed to re-serialize JSON ({err})"
        ))
    })?;
    Ok(Value::Json(bytes))
}

#[cfg(test)]
mod tests {
    use crate::storage::schema::Value;
    use crate::storage::wal::{WalReader, WalRecord};
    use crate::storage::{DeployProfile, StoragePackaging, StorageProfileSelection};
    use crate::{RedDBOptions, RedDBRuntime};
    use std::path::Path;

    fn persistent_operational_options(path: &Path) -> RedDBOptions {
        RedDBOptions::persistent(path)
            .with_storage_profile(StorageProfileSelection {
                deploy_profile: DeployProfile::Embedded,
                packaging: StoragePackaging::OperationalDirectory,
                replica_count: 0,
                managed_backup: false,
                wal_retention: false,
            })
            .unwrap()
    }

    fn store_commit_batches(wal_path: &Path) -> Vec<Vec<Vec<u8>>> {
        WalReader::open(wal_path)
            .expect("wal opens")
            .iter()
            .map(|record| record.expect("wal record decodes").1)
            .filter_map(|record| match record {
                WalRecord::TxCommitBatch { actions, .. } => Some(actions),
                _ => None,
            })
            .collect()
    }

    fn action_contains_text(action: &[u8], needle: &str) -> bool {
        action
            .windows(needle.len())
            .any(|window| window == needle.as_bytes())
    }

    fn claim_metric_count(
        snapshot: &crate::runtime::ClaimTelemetrySnapshot,
        metric: &str,
        collection: &str,
        model: &str,
    ) -> u64 {
        let rows = match metric {
            "attempts" => &snapshot.attempts,
            "successful" => &snapshot.successful,
            "misses" => &snapshot.misses,
            "skipped_locked" => &snapshot.skipped_locked,
            other => panic!("unknown claim metric {other}"),
        };
        rows.iter()
            .find(|((actual_collection, actual_model), _)| {
                actual_collection == collection && actual_model == model
            })
            .map(|(_, count)| *count)
            .unwrap_or(0)
    }

    fn assert_statement_writes_collections_in_one_new_wal_batch(
        rt: &RedDBRuntime,
        wal_path: &Path,
        statement: &str,
        source: &str,
        event_queue: &str,
    ) {
        let before_batches = store_commit_batches(wal_path).len();

        rt.execute_query(statement).unwrap();

        let batches = store_commit_batches(wal_path);
        let statement_batches = &batches[before_batches..];
        let source_batch = statement_batches
            .iter()
            .position(|actions| {
                actions.iter().any(|action| {
                    action_contains_text(action, source)
                        && !action_contains_text(action, event_queue)
                })
            })
            .expect("source collection write batch is present");
        let event_batch = statement_batches
            .iter()
            .position(|actions| {
                actions
                    .iter()
                    .any(|action| action_contains_text(action, event_queue))
            })
            .expect("event queue write batch is present");

        assert_eq!(
            source_batch, event_batch,
            "WITH EVENTS must persist the source write and queue event in the same WAL batch"
        );
    }

    #[test]
    fn with_events_autocommit_persists_mutation_and_event_in_one_wal_batch() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("events_dual_write.rdb");
        let wal_path = reddb_file::layout::unified_wal_path(&db_path);
        let rt = RedDBRuntime::with_options(persistent_operational_options(&db_path)).unwrap();

        rt.execute_query("CREATE TABLE users (id INT, email TEXT) WITH EVENTS")
            .unwrap();
        assert_statement_writes_collections_in_one_new_wal_batch(
            &rt,
            &wal_path,
            "INSERT INTO users (id, email) VALUES (1, 'a@example.test')",
            "users",
            "users_events",
        );
    }

    #[test]
    fn with_events_autocommit_update_persists_mutation_and_event_in_one_wal_batch() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("events_update_atomic.rdb");
        let wal_path = reddb_file::layout::unified_wal_path(&db_path);
        let rt = RedDBRuntime::with_options(persistent_operational_options(&db_path)).unwrap();

        rt.execute_query(
            "CREATE TABLE users (id INT, email TEXT) WITH EVENTS (UPDATE) TO user_updates",
        )
        .unwrap();
        rt.execute_query("INSERT INTO users (id, email) VALUES (1, 'a@example.test')")
            .unwrap();

        assert_statement_writes_collections_in_one_new_wal_batch(
            &rt,
            &wal_path,
            "UPDATE users SET email = 'b@example.test' WHERE id = 1",
            "users",
            "user_updates",
        );
    }

    #[test]
    fn with_events_autocommit_delete_persists_mutation_and_event_in_one_wal_batch() {
        let dir = tempfile::tempdir().unwrap();
        let db_path = dir.path().join("events_delete_atomic.rdb");
        let wal_path = reddb_file::layout::unified_wal_path(&db_path);
        let rt = RedDBRuntime::with_options(persistent_operational_options(&db_path)).unwrap();

        rt.execute_query(
            "CREATE TABLE users (id INT, email TEXT) WITH EVENTS (DELETE) TO user_deletes",
        )
        .unwrap();
        rt.execute_query("INSERT INTO users (id, email) VALUES (1, 'a@example.test')")
            .unwrap();

        assert_statement_writes_collections_in_one_new_wal_batch(
            &rt,
            &wal_path,
            "DELETE FROM users WHERE id = 1",
            "users",
            "user_deletes",
        );
    }

    #[test]
    fn update_where_id_in_with_hash_index_updates_expected_rows() {
        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
        rt.execute_query("CREATE TABLE users (id INT, score INT)")
            .unwrap();
        for id in 0..5 {
            rt.execute_query(&format!("INSERT INTO users (id, score) VALUES ({id}, 0)"))
                .unwrap();
        }
        rt.execute_query("CREATE INDEX idx_id ON users (id) USING HASH")
            .unwrap();

        let updated = rt
            .execute_query("UPDATE users SET score = 42 WHERE id IN (1,3,4)")
            .unwrap();
        assert_eq!(updated.affected_rows, 3);

        let selected = rt
            .execute_query("SELECT id, score FROM users ORDER BY id")
            .unwrap();
        let scores: Vec<(i64, i64)> = selected
            .result
            .records
            .iter()
            .map(|record| {
                let id = match record.get("id").unwrap() {
                    Value::Integer(value) => *value,
                    other => panic!("expected integer id, got {other:?}"),
                };
                let score = match record.get("score").unwrap() {
                    Value::Integer(value) => *value,
                    other => panic!("expected integer score, got {other:?}"),
                };
                (id, score)
            })
            .collect();
        assert_eq!(scores, vec![(0, 0), (1, 42), (2, 0), (3, 42), (4, 42)]);
    }

    /// Drives UPDATE through the shared `DmlTargetScan` module — the
    /// same code path DELETE uses (#51, #52). Exercises the indexed
    /// equality fast-path (WHERE id = N with a HASH index), the
    /// unindexed range scan (WHERE score > N), and the no-WHERE
    /// full-scan branch to confirm the extracted "find target rows"
    /// loop preserves affected-row counts and the resulting row state.
    #[test]
    fn update_routes_through_dml_target_scan_for_indexed_and_scan_paths() {
        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
        rt.execute_query("CREATE TABLE items (id INT, score INT)")
            .unwrap();
        for id in 0..5 {
            rt.execute_query(&format!(
                "INSERT INTO items (id, score) VALUES ({id}, {})",
                id * 10
            ))
            .unwrap();
        }
        rt.execute_query("CREATE INDEX idx_items_id ON items (id) USING HASH")
            .unwrap();

        // Indexed equality UPDATE — hits the hash fast-path inside
        // DmlTargetScan::find_target_ids. id=2 has score=20, drop it
        // below the score>25 cutoff so the next assertion stays clean.
        let updated_one = rt
            .execute_query("UPDATE items SET score = 5 WHERE id = 2")
            .unwrap();
        assert_eq!(updated_one.affected_rows, 1);

        // Unindexed scan UPDATE — bumps everyone with score > 25,
        // i.e. ids 3 and 4 (scores 30, 40). Goes through the
        // zoned/full-scan branch.
        let updated_many = rt
            .execute_query("UPDATE items SET score = 7 WHERE score > 25")
            .unwrap();
        assert_eq!(updated_many.affected_rows, 2);

        let snapshot = rt
            .execute_query("SELECT id, score FROM items ORDER BY id")
            .unwrap();
        let pairs: Vec<(i64, i64)> = snapshot
            .result
            .records
            .iter()
            .map(|record| {
                let id = match record.get("id").unwrap() {
                    Value::Integer(value) => *value,
                    other => panic!("expected integer id, got {other:?}"),
                };
                let score = match record.get("score").unwrap() {
                    Value::Integer(value) => *value,
                    other => panic!("expected integer score, got {other:?}"),
                };
                (id, score)
            })
            .collect();
        assert_eq!(pairs, vec![(0, 0), (1, 10), (2, 5), (3, 7), (4, 7)]);

        // Full-scan UPDATE with no WHERE rewrites every remaining row.
        let updated_all = rt.execute_query("UPDATE items SET score = 1").unwrap();
        assert_eq!(updated_all.affected_rows, 5);
        let after = rt
            .execute_query("SELECT score FROM items ORDER BY id")
            .unwrap();
        let scores: Vec<i64> = after
            .result
            .records
            .iter()
            .map(|record| match record.get("score").unwrap() {
                Value::Integer(value) => *value,
                other => panic!("expected integer score, got {other:?}"),
            })
            .collect();
        assert_eq!(scores, vec![1, 1, 1, 1, 1]);
    }

    /// Drives DELETE through the new `DmlTargetScan` module. Exercises
    /// both the index fast-path (WHERE id = N with a HASH index) and
    /// the unindexed scan path (WHERE score > N) to confirm the
    /// extracted "find target rows" loop preserves the affected-row
    /// count and which rows survive.
    #[test]
    fn delete_routes_through_dml_target_scan_for_indexed_and_scan_paths() {
        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
        rt.execute_query("CREATE TABLE items (id INT, score INT)")
            .unwrap();
        for id in 0..5 {
            rt.execute_query(&format!(
                "INSERT INTO items (id, score) VALUES ({id}, {})",
                id * 10
            ))
            .unwrap();
        }
        rt.execute_query("CREATE INDEX idx_items_id ON items (id) USING HASH")
            .unwrap();

        // Indexed equality DELETE — hits the hash fast-path inside
        // DmlTargetScan::find_target_ids.
        let deleted_one = rt.execute_query("DELETE FROM items WHERE id = 2").unwrap();
        assert_eq!(deleted_one.affected_rows, 1);

        // Unindexed scan DELETE — drops everyone with score > 25,
        // i.e. ids 3 and 4 (scores 30, 40). Goes through the
        // zoned/full-scan branch.
        let deleted_many = rt
            .execute_query("DELETE FROM items WHERE score > 25")
            .unwrap();
        assert_eq!(deleted_many.affected_rows, 2);

        let surviving = rt
            .execute_query("SELECT id FROM items ORDER BY id")
            .unwrap();
        let ids: Vec<i64> = surviving
            .result
            .records
            .iter()
            .map(|record| match record.get("id").unwrap() {
                Value::Integer(value) => *value,
                other => panic!("expected integer id, got {other:?}"),
            })
            .collect();
        assert_eq!(ids, vec![0, 1]);

        // Sanity: full-scan DELETE with no WHERE clears the rest.
        let deleted_rest = rt.execute_query("DELETE FROM items").unwrap();
        assert_eq!(deleted_rest.affected_rows, 2);
        let empty = rt.execute_query("SELECT id FROM items").unwrap();
        assert!(empty.result.records.is_empty());
    }

    /// CollectionContract gate (#49 + #50): APPEND ONLY tables accept
    /// INSERT but reject UPDATE and DELETE with the documented
    /// operator-facing error strings. Drives all three DML verbs so
    /// the centralized gate is exercised end-to-end.
    #[test]
    fn collection_contract_gate_blocks_update_and_delete_on_append_only() {
        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
        rt.execute_query("CREATE TABLE events (id INT, payload TEXT) APPEND ONLY")
            .unwrap();

        // INSERT must succeed — APPEND ONLY exists precisely to allow
        // appends. The gate should be a no-op for INSERT.
        let inserted = rt
            .execute_query("INSERT INTO events (id, payload) VALUES (1, 'hello')")
            .unwrap();
        assert_eq!(inserted.affected_rows, 1);

        // UPDATE is rejected with the gate's UPDATE-specific message.
        let update_err = rt
            .execute_query("UPDATE events SET payload = 'mut' WHERE id = 1")
            .unwrap_err();
        let msg = format!("{update_err}");
        assert!(
            msg.contains("APPEND ONLY") && msg.contains("UPDATE is rejected"),
            "expected UPDATE rejection message, got: {msg}"
        );

        // DELETE is rejected with the gate's DELETE-specific message.
        let delete_err = rt
            .execute_query("DELETE FROM events WHERE id = 1")
            .unwrap_err();
        let msg = format!("{delete_err}");
        assert!(
            msg.contains("APPEND ONLY") && msg.contains("DELETE is rejected"),
            "expected DELETE rejection message, got: {msg}"
        );

        // Row should still be present — neither rejected mutation
        // touched storage.
        let surviving = rt.execute_query("SELECT id FROM events").unwrap();
        assert_eq!(surviving.result.records.len(), 1);
    }

    /// CollectionContract gate: tables without an APPEND ONLY contract
    /// permit INSERT, UPDATE, and DELETE — the gate's default branch
    /// is a true pass-through, not an accidental block.
    #[test]
    fn collection_contract_gate_allows_all_verbs_on_unrestricted_table() {
        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
        rt.execute_query("CREATE TABLE notes (id INT, body TEXT)")
            .unwrap();

        rt.execute_query("INSERT INTO notes (id, body) VALUES (1, 'a')")
            .unwrap();
        let updated = rt
            .execute_query("UPDATE notes SET body = 'b' WHERE id = 1")
            .unwrap();
        assert_eq!(updated.affected_rows, 1);
        let deleted = rt.execute_query("DELETE FROM notes WHERE id = 1").unwrap();
        assert_eq!(deleted.affected_rows, 1);
    }

    #[test]
    fn insert_into_event_enabled_table_emits_event_to_configured_queue() {
        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
        rt.execute_query(
            "CREATE TABLE users (id INT, email TEXT) WITH EVENTS (INSERT) TO audit_log",
        )
        .unwrap();

        let inserted = rt
            .execute_query("INSERT INTO users (id, email) VALUES (7, 'a@example.com')")
            .unwrap();
        assert_eq!(inserted.affected_rows, 1);

        let events = queue_payloads(&rt, "audit_log");
        assert_eq!(events.len(), 1);
        let event = events[0].as_object().expect("event payload object");
        assert!(event
            .get("event_id")
            .and_then(crate::json::Value::as_str)
            .is_some_and(|value| !value.is_empty()));
        assert_eq!(
            event.get("op").and_then(crate::json::Value::as_str),
            Some("insert")
        );
        assert_eq!(
            event.get("collection").and_then(crate::json::Value::as_str),
            Some("users")
        );
        assert_eq!(
            event.get("id").and_then(crate::json::Value::as_u64),
            Some(7)
        );
        assert!(event
            .get("ts")
            .and_then(crate::json::Value::as_u64)
            .is_some());
        assert!(event
            .get("lsn")
            .and_then(crate::json::Value::as_u64)
            .is_some());
        assert!(matches!(
            event.get("tenant"),
            Some(crate::json::Value::Null)
        ));
        assert!(matches!(
            event.get("before"),
            Some(crate::json::Value::Null)
        ));
        let after = event
            .get("after")
            .and_then(crate::json::Value::as_object)
            .expect("after object");
        assert_eq!(
            after.get("id").and_then(crate::json::Value::as_u64),
            Some(7)
        );
        assert_eq!(
            after.get("email").and_then(crate::json::Value::as_str),
            Some("a@example.com")
        );
    }

    #[test]
    fn multi_row_insert_emits_one_insert_event_per_row_in_order() {
        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
        rt.execute_query("CREATE TABLE users (id INT, email TEXT) WITH EVENTS")
            .unwrap();

        rt.execute_query(
            "INSERT INTO users (id, email) VALUES (1, 'a@example.com'), (2, 'b@example.com')",
        )
        .unwrap();

        let events = queue_payloads(&rt, "users_events");
        assert_eq!(events.len(), 2);
        let mut previous_lsn = 0;
        for (event, expected_id) in events.iter().zip([1_u64, 2]) {
            let object = event.as_object().expect("event payload object");
            assert_eq!(
                object.get("op").and_then(crate::json::Value::as_str),
                Some("insert")
            );
            assert_eq!(
                object.get("id").and_then(crate::json::Value::as_u64),
                Some(expected_id)
            );
            let lsn = object
                .get("lsn")
                .and_then(crate::json::Value::as_u64)
                .expect("event lsn");
            assert!(
                lsn > previous_lsn,
                "event LSNs should increase in row order"
            );
            previous_lsn = lsn;
            let after = object
                .get("after")
                .and_then(crate::json::Value::as_object)
                .expect("after object");
            assert_eq!(
                after.get("id").and_then(crate::json::Value::as_u64),
                Some(expected_id)
            );
        }
    }

    #[test]
    fn claim_metrics_increment_skipped_locked_without_counting_miss() {
        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("runtime");
        rt.execute_query("CREATE TABLE claim_metric_locked (id INT, rank INT, status TEXT)")
            .expect("create table");
        // ADR 0063: index-backed claim ordering on `rank`.
        rt.execute_query("CREATE INDEX idx_claim_metric_locked_rank ON claim_metric_locked (rank)")
            .expect("create index");
        rt.execute_query(
            "INSERT INTO claim_metric_locked (id, rank, status) VALUES \
             (1, 10, 'ready'), (2, 20, 'ready')",
        )
        .expect("insert rows");

        let claim_lock = rt
            .inner
            .rmw_locks
            .lock_for("claim_metric_locked", "__table_claim_update__");
        let _guard = claim_lock.lock();
        let updated = rt
            .execute_query(
                "UPDATE claim_metric_locked SET status = 'claimed' WHERE status = 'ready' \
                 CLAIM LIMIT 2 ORDER BY rank ASC",
            )
            .expect("claim skips locked candidates");

        assert_eq!(updated.affected_rows, 0);
        let snapshot = rt.claim_telemetry_snapshot();
        assert_eq!(
            claim_metric_count(&snapshot, "attempts", "claim_metric_locked", "table"),
            1
        );
        assert_eq!(
            claim_metric_count(&snapshot, "skipped_locked", "claim_metric_locked", "table"),
            2
        );
        assert_eq!(
            claim_metric_count(&snapshot, "misses", "claim_metric_locked", "table"),
            0
        );
        assert_eq!(
            snapshot.skipped_locked.len(),
            1,
            "skipped-lock labels stay bounded to collection/model"
        );
    }

    fn queue_payloads(rt: &RedDBRuntime, queue: &str) -> Vec<crate::json::Value> {
        let result = rt
            .execute_query(&format!("QUEUE PEEK {queue} 10"))
            .expect("peek queue");
        result
            .result
            .records
            .iter()
            .map(
                |record| match record.get("payload").expect("payload column") {
                    Value::Json(bytes) => crate::json::from_slice(bytes).expect("json payload"),
                    other => panic!("expected JSON queue payload, got {other:?}"),
                },
            )
            .collect()
    }

    // ── #112: auto-index user `id` on first insert ─────────────────────

    /// First insert into a fresh collection that carries a column named
    /// `id` registers an implicit HASH index on `id`. Subsequent inserts
    /// populate it transparently, and `WHERE id = N` lookups exercise
    /// the hash-index fast path in `DmlTargetScan::find_target_ids`.
    ///
    /// This is the load-bearing acceptance test for #112 — without the
    /// hook, `find_index_for_column` returns `None` and DELETE/UPDATE
    /// fall through to a full segment scan (the 4× perf gap documented
    /// in `docs/perf/delete-sequential-2026-05-06.md`).
    #[test]
    fn auto_index_id_fires_on_first_insert() {
        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
        rt.execute_query("CREATE TABLE bench_users (id INT, score INT)")
            .unwrap();

        // Pre-condition: no index on `id` yet.
        assert!(
            rt.index_store_ref()
                .find_index_for_column("bench_users", "id")
                .is_none(),
            "freshly created collection should not have an `id` index"
        );

        // Single-row INSERT — drives `MutationEngine::append_one`.
        rt.execute_query("INSERT INTO bench_users (id, score) VALUES (1, 10)")
            .unwrap();

        // Post-condition: hash index registered on `id`.
        let registered = rt
            .index_store_ref()
            .find_index_for_column("bench_users", "id")
            .expect("auto-index hook should have registered idx_id on first insert");
        assert_eq!(registered.name, "idx_id");
        assert_eq!(registered.collection, "bench_users");
        assert_eq!(registered.columns, vec!["id".to_string()]);
        assert!(matches!(
            registered.method,
            super::super::index_store::IndexMethodKind::Hash
        ));

        // Subsequent inserts populate the index; `WHERE id = N` should
        // resolve via the hash fast path and round-trip every row.
        for id in 2..=5 {
            rt.execute_query(&format!(
                "INSERT INTO bench_users (id, score) VALUES ({id}, {})",
                id * 10
            ))
            .unwrap();
        }
        for id in 1..=5 {
            let result = rt
                .execute_query(&format!("SELECT score FROM bench_users WHERE id = {id}"))
                .unwrap();
            assert_eq!(
                result.result.records.len(),
                1,
                "id={id} should match one row"
            );
        }

        // Delete via the hash fast-path — exactly the bench scenario the
        // perf doc identified as the 4× regression. With the index
        // present, `find_target_ids` short-circuits before
        // `for_each_entity_zoned` runs.
        let deleted = rt
            .execute_query("DELETE FROM bench_users WHERE id = 3")
            .unwrap();
        assert_eq!(deleted.affected_rows, 1);
    }

    /// Bulk INSERT (the multi-row VALUES path) drives
    /// `MutationEngine::append_batch`. The hook must fire there too —
    /// otherwise the batch entry points (gRPC binary bulk, HTTP bulk,
    /// wire bulk INSERT) skip auto-indexing entirely.
    #[test]
    fn auto_index_id_fires_on_first_bulk_insert() {
        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
        rt.execute_query("CREATE TABLE bench_bulk (id INT, score INT)")
            .unwrap();

        rt.execute_query("INSERT INTO bench_bulk (id, score) VALUES (1, 10), (2, 20), (3, 30)")
            .unwrap();

        let registered = rt
            .index_store_ref()
            .find_index_for_column("bench_bulk", "id")
            .expect("auto-index hook should fire on first bulk insert");
        assert_eq!(registered.name, "idx_id");

        // Every row populated via `index_entity_insert_batch`.
        for id in 1..=3 {
            let result = rt
                .execute_query(&format!("SELECT score FROM bench_bulk WHERE id = {id}"))
                .unwrap();
            assert_eq!(result.result.records.len(), 1);
        }
    }

    /// Hook is a no-op when the row carries no `id` column. Conservative
    /// match (case-sensitive `id`) — `Id`, `ID`, and `rid`
    /// don't trigger it.
    #[test]
    fn auto_index_id_skips_when_no_id_column() {
        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
        rt.execute_query("CREATE TABLE plain (uid INT, label TEXT)")
            .unwrap();
        rt.execute_query("INSERT INTO plain (uid, label) VALUES (1, 'a')")
            .unwrap();

        assert!(rt
            .index_store_ref()
            .find_index_for_column("plain", "id")
            .is_none());
        assert!(rt
            .index_store_ref()
            .find_index_for_column("plain", "uid")
            .is_none());
    }

    /// Hook only fires once per collection. If an explicit
    /// `CREATE INDEX ... USING BTREE` already covers `id`, the hook
    /// detects it via `find_index_for_column` and does NOT clobber it
    /// with a HASH index on the next insert.
    #[test]
    fn auto_index_id_skips_when_index_already_exists() {
        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
        rt.execute_query("CREATE TABLE pre (id INT, score INT)")
            .unwrap();
        // User-declared BTREE index on `id` before any insert.
        rt.execute_query("CREATE INDEX user_idx ON pre (id) USING BTREE")
            .unwrap();
        rt.execute_query("INSERT INTO pre (id, score) VALUES (1, 10)")
            .unwrap();

        let registered = rt
            .index_store_ref()
            .find_index_for_column("pre", "id")
            .expect("user index should still be there");
        assert_eq!(
            registered.name, "user_idx",
            "auto-index hook must not overwrite an existing index"
        );
    }

    /// Implicit `idx_id` is reaped when the collection drops. The
    /// existing `execute_drop_table` walks `list_indices` and drops every
    /// entry — confirm the auto-created index participates.
    #[test]
    fn auto_index_id_dropped_with_collection() {
        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
        rt.execute_query("CREATE TABLE ephemeral (id INT, score INT)")
            .unwrap();
        rt.execute_query("INSERT INTO ephemeral (id, score) VALUES (1, 10)")
            .unwrap();
        assert!(rt
            .index_store_ref()
            .find_index_for_column("ephemeral", "id")
            .is_some());

        rt.execute_query("DROP TABLE ephemeral").unwrap();

        assert!(
            rt.index_store_ref()
                .find_index_for_column("ephemeral", "id")
                .is_none(),
            "implicit `idx_id` must be reaped when its collection drops"
        );
    }

    /// Opt-out via `RedDBOptions::with_auto_index_id(false)` (which
    /// forwards to `UnifiedStoreConfig::auto_index_id`). With the knob
    /// off, first insert leaves the collection without an `id` index —
    /// DELETE/UPDATE fall back to the scan path.
    #[test]
    fn auto_index_id_disabled_by_config() {
        let opts = RedDBOptions::in_memory().with_auto_index_id(false);
        let rt = RedDBRuntime::with_options(opts).unwrap();

        rt.execute_query("CREATE TABLE off (id INT, score INT)")
            .unwrap();
        rt.execute_query("INSERT INTO off (id, score) VALUES (1, 10)")
            .unwrap();

        assert!(
            rt.index_store_ref()
                .find_index_for_column("off", "id")
                .is_none(),
            "with auto_index_id=false, no implicit index should be created"
        );
    }

    // ── #293: UPDATE / DELETE events ─────────────────────────────────────

    #[test]
    fn update_single_row_emits_update_event() {
        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
        rt.execute_query(
            "CREATE TABLE users (id INT, name TEXT) WITH EVENTS (UPDATE) TO audit_log",
        )
        .unwrap();
        rt.execute_query("INSERT INTO users (id, name) VALUES (1, 'Alice')")
            .unwrap();

        rt.execute_query("UPDATE users SET name = 'Bob' WHERE id = 1")
            .unwrap();

        let events = queue_payloads(&rt, "audit_log");
        assert_eq!(events.len(), 1, "expected exactly 1 update event");
        let event = events[0].as_object().expect("event payload object");
        assert_eq!(
            event.get("op").and_then(crate::json::Value::as_str),
            Some("update")
        );
        assert_eq!(
            event.get("collection").and_then(crate::json::Value::as_str),
            Some("users")
        );
        assert!(event
            .get("event_id")
            .and_then(crate::json::Value::as_str)
            .is_some_and(|v| !v.is_empty()));
        let before = event
            .get("before")
            .and_then(crate::json::Value::as_object)
            .expect("before must be an object");
        let after = event
            .get("after")
            .and_then(crate::json::Value::as_object)
            .expect("after must be an object");
        assert_eq!(
            before.get("name").and_then(crate::json::Value::as_str),
            Some("Alice"),
            "before.name should be the old value"
        );
        assert_eq!(
            after.get("name").and_then(crate::json::Value::as_str),
            Some("Bob"),
            "after.name should be the new value"
        );
    }

    #[test]
    fn update_event_only_includes_changed_fields() {
        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
        rt.execute_query(
            "CREATE TABLE users (id INT, name TEXT, email TEXT) WITH EVENTS (UPDATE) TO evts",
        )
        .unwrap();
        rt.execute_query("INSERT INTO users (id, name, email) VALUES (1, 'Alice', 'a@x.com')")
            .unwrap();

        rt.execute_query("UPDATE users SET name = 'Bob' WHERE id = 1")
            .unwrap();

        let events = queue_payloads(&rt, "evts");
        assert_eq!(events.len(), 1);
        let event = events[0].as_object().unwrap();
        let before = event
            .get("before")
            .and_then(crate::json::Value::as_object)
            .unwrap();
        let after = event
            .get("after")
            .and_then(crate::json::Value::as_object)
            .unwrap();
        // Only changed field included.
        assert!(
            before.contains_key("name"),
            "before must include changed field"
        );
        assert!(
            after.contains_key("name"),
            "after must include changed field"
        );
        // Unchanged fields must not appear.
        assert!(
            !before.contains_key("email"),
            "before must not include unchanged email"
        );
        assert!(
            !after.contains_key("email"),
            "after must not include unchanged email"
        );
    }

    #[test]
    fn multi_row_update_emits_one_event_per_row() {
        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
        rt.execute_query("CREATE TABLE items (id INT, status TEXT) WITH EVENTS (UPDATE) TO evts")
            .unwrap();
        rt.execute_query(
            "INSERT INTO items (id, status) VALUES (1, 'new'), (2, 'new'), (3, 'new')",
        )
        .unwrap();

        rt.execute_query("UPDATE items SET status = 'done'")
            .unwrap();

        let events = queue_payloads(&rt, "evts");
        assert_eq!(events.len(), 3, "expected one update event per row");
        for event in &events {
            let obj = event.as_object().unwrap();
            assert_eq!(
                obj.get("op").and_then(crate::json::Value::as_str),
                Some("update")
            );
        }
    }

    #[test]
    fn delete_single_row_emits_delete_event() {
        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
        rt.execute_query("CREATE TABLE users (id INT, name TEXT) WITH EVENTS (DELETE) TO del_log")
            .unwrap();
        rt.execute_query("INSERT INTO users (id, name) VALUES (42, 'Alice')")
            .unwrap();

        rt.execute_query("DELETE FROM users WHERE id = 42").unwrap();

        let events = queue_payloads(&rt, "del_log");
        assert_eq!(events.len(), 1);
        let event = events[0].as_object().expect("event payload object");
        assert_eq!(
            event.get("op").and_then(crate::json::Value::as_str),
            Some("delete")
        );
        assert_eq!(
            event.get("collection").and_then(crate::json::Value::as_str),
            Some("users")
        );
        assert!(event
            .get("event_id")
            .and_then(crate::json::Value::as_str)
            .is_some_and(|v| !v.is_empty()));
        let before = event
            .get("before")
            .and_then(crate::json::Value::as_object)
            .expect("before must be an object for delete");
        assert_eq!(
            before.get("id").and_then(crate::json::Value::as_u64),
            Some(42)
        );
        assert_eq!(
            before.get("name").and_then(crate::json::Value::as_str),
            Some("Alice")
        );
        assert!(matches!(event.get("after"), Some(crate::json::Value::Null)));
    }

    #[test]
    fn multi_row_delete_emits_one_event_per_row() {
        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
        rt.execute_query("CREATE TABLE items (id INT, val INT) WITH EVENTS (DELETE) TO del_log")
            .unwrap();
        rt.execute_query("INSERT INTO items (id, val) VALUES (1, 10), (2, 20), (3, 30)")
            .unwrap();

        rt.execute_query("DELETE FROM items").unwrap();

        let events = queue_payloads(&rt, "del_log");
        assert_eq!(events.len(), 3, "expected one delete event per deleted row");
        for event in &events {
            let obj = event.as_object().unwrap();
            assert_eq!(
                obj.get("op").and_then(crate::json::Value::as_str),
                Some("delete")
            );
            assert!(matches!(obj.get("after"), Some(crate::json::Value::Null)));
        }
    }

    #[test]
    fn ops_filter_update_does_not_emit_on_insert_or_delete() {
        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
        rt.execute_query("CREATE TABLE users (id INT, name TEXT) WITH EVENTS (UPDATE) TO evts")
            .unwrap();

        rt.execute_query("INSERT INTO users (id, name) VALUES (1, 'Alice')")
            .unwrap();
        rt.execute_query("DELETE FROM users WHERE id = 1").unwrap();

        let events = queue_payloads(&rt, "evts");
        assert!(
            events.is_empty(),
            "UPDATE-only filter must not emit INSERT or DELETE events"
        );
    }

    // ── SUPPRESS EVENTS ────────────────────────────────────────────────────

    #[test]
    fn suppress_events_on_insert_emits_no_events() {
        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
        rt.execute_query("CREATE TABLE users (id INT, name TEXT) WITH EVENTS TO evts")
            .unwrap();

        rt.execute_query("INSERT INTO users (id, name) VALUES (1, 'Alice') SUPPRESS EVENTS")
            .unwrap();

        let events = queue_payloads(&rt, "evts");
        assert!(
            events.is_empty(),
            "SUPPRESS EVENTS must prevent INSERT events"
        );
    }

    #[test]
    fn suppress_events_on_update_emits_no_events() {
        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
        rt.execute_query("CREATE TABLE users (id INT, name TEXT) WITH EVENTS TO evts")
            .unwrap();
        rt.execute_query("INSERT INTO users (id, name) VALUES (1, 'Alice')")
            .unwrap();
        // drain the INSERT event
        let _ = queue_payloads(&rt, "evts");
        // Force pop to drain; simpler: just check new count after UPDATE
        rt.execute_query("QUEUE PURGE evts").unwrap();

        rt.execute_query("UPDATE users SET name = 'Bob' WHERE id = 1 SUPPRESS EVENTS")
            .unwrap();

        let events = queue_payloads(&rt, "evts");
        assert!(
            events.is_empty(),
            "SUPPRESS EVENTS must prevent UPDATE events"
        );
    }

    #[test]
    fn suppress_events_on_delete_emits_no_events() {
        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
        rt.execute_query(
            "CREATE TABLE users (id INT, name TEXT) WITH EVENTS (INSERT, DELETE) TO evts",
        )
        .unwrap();
        rt.execute_query("INSERT INTO users (id, name) VALUES (1, 'Alice') SUPPRESS EVENTS")
            .unwrap();

        rt.execute_query("DELETE FROM users WHERE id = 1 SUPPRESS EVENTS")
            .unwrap();

        let events = queue_payloads(&rt, "evts");
        assert!(
            events.is_empty(),
            "SUPPRESS EVENTS must prevent DELETE events"
        );
    }

    #[test]
    fn normal_insert_after_suppress_still_emits() {
        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
        rt.execute_query("CREATE TABLE users (id INT, name TEXT) WITH EVENTS TO evts")
            .unwrap();

        rt.execute_query("INSERT INTO users (id, name) VALUES (1, 'Alice') SUPPRESS EVENTS")
            .unwrap();
        rt.execute_query("INSERT INTO users (id, name) VALUES (2, 'Bob')")
            .unwrap();

        let events = queue_payloads(&rt, "evts");
        assert_eq!(
            events.len(),
            1,
            "only the non-suppressed INSERT should emit"
        );
        assert_eq!(
            events[0].get("id").and_then(crate::json::Value::as_u64),
            Some(2)
        );
    }
}