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::runtime::mvcc::current_connection_id;
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 reddb_rql::ast::{
    BinOp, CompareOp, Expr, FieldRef, Filter, OnConflictAction, ReturningItem, UpdateQuery,
    UpdateTarget,
};
use reddb_rql::sql_lowering::{
    effective_delete_filter, effective_insert_rows, effective_update_filter, fold_expr_to_value,
};
use std::collections::HashMap;
use std::sync::Arc;

use super::*;
// Insert-support and value-conversion helpers were extracted to
// `impl_dml_support` (issue #1632); import them so existing call sites keep
// using their bare names unchanged.
use super::impl_dml_support::*;
// RETURNING / update-analysis / update-target / claim helpers were extracted
// to sibling modules (issue #1633); import them so existing call sites keep
// using their bare names unchanged.
use super::impl_dml_claim::*;
use super::impl_dml_returning::*;
use super::impl_dml_update_analysis::*;
use super::impl_dml_update_target::*;
// Chain-integrity / tenant-injection / batch-flush / crypto method families
// were extracted to sibling modules (issue #1634). Their methods dispatch
// through `self`, so no glob import is needed here. The SQL TTL free functions
// are re-exported so both the execution paths below and
// `impl_dml_support`'s `use super::impl_dml::{...}` keep resolving unchanged.
pub(super) use super::impl_dml_ttl::{canonicalize_sql_ttl_metadata, resolve_sql_ttl_metadata_key};

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

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

pub(super) struct CompiledUpdatePlan {
    pub(super) static_field_assignments: Vec<(String, Value)>,
    pub(super) 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,
    /// The SET list names a column of a standalone UNIQUE index.
    touches_unique_index: bool,
}

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

fn resolve_excluded_assignment(
    expr: &Expr,
    excluded_fields: &[(String, Value)],
) -> RedDBResult<Expr> {
    match expr {
        Expr::Column {
            field: FieldRef::TableColumn { table, column },
            span,
        } if table.eq_ignore_ascii_case("excluded") => {
            let value = excluded_fields
                .iter()
                .find(|(name, _)| name.eq_ignore_ascii_case(column))
                .map(|(_, value)| value.clone())
                .ok_or_else(|| {
                    RedDBError::Query(format!(
                        "ON CONFLICT EXCLUDED references unknown insert column '{column}'"
                    ))
                })?;
            Ok(Expr::Literal { value, span: *span })
        }
        Expr::BinaryOp { op, lhs, rhs, span } => Ok(Expr::BinaryOp {
            op: *op,
            lhs: Box::new(resolve_excluded_assignment(lhs, excluded_fields)?),
            rhs: Box::new(resolve_excluded_assignment(rhs, excluded_fields)?),
            span: *span,
        }),
        _ => Ok(expr.clone()),
    }
}

fn conflict_columns_match(index_columns: &[String], target_columns: &[String]) -> bool {
    index_columns.len() == target_columns.len()
        && index_columns.iter().all(|index_column| {
            target_columns
                .iter()
                .any(|target_column| target_column.eq_ignore_ascii_case(index_column))
        })
}

fn conflict_index_key(fields: &[(String, Value)], columns: &[String]) -> Option<Vec<Value>> {
    columns
        .iter()
        .map(|column| {
            super::index_store::index_field_value(fields, column)
                .filter(|value| !matches!(value.as_ref(), Value::Null))
                .map(|value| value.into_owned())
        })
        .collect()
}

fn document_on_conflict_fields(
    stored_fields: &[(String, Value)],
    body: &crate::json::Value,
) -> RedDBResult<Vec<(String, Value)>> {
    let mut fields = stored_fields.to_vec();
    if let crate::json::Value::Object(object) = body {
        for (name, value) in object {
            fields.push((
                name.clone(),
                crate::application::entity::json_to_storage_value(value)?,
            ));
        }
    }
    Ok(fields)
}

impl RedDBRuntime {
    fn conflict_unique_indexes(
        &self,
        collection: &str,
        target: Option<&[String]>,
    ) -> RedDBResult<Vec<super::index_store::RegisteredIndex>> {
        let mut indexes = self
            .index_store_ref()
            .list_indices(collection)
            .into_iter()
            .filter(|index| index.unique)
            .collect::<Vec<_>>();
        if let Some(target) = target {
            indexes.retain(|index| conflict_columns_match(&index.columns, target));
            if indexes.is_empty() {
                return Err(RedDBError::Query(format!(
                    "no unique index on collection '{}' matches ON CONFLICT ({})",
                    collection,
                    target.join(", ")
                )));
            }
        }
        Ok(indexes)
    }

    fn unique_index_conflict_id(
        &self,
        collection: &str,
        fields: &[(String, Value)],
        target: Option<&[String]>,
    ) -> RedDBResult<Option<EntityId>> {
        let indexes = self.conflict_unique_indexes(collection, target)?;
        let Some(manager) = self.inner.db.store().get_collection(collection) else {
            return Ok(None);
        };
        for index in indexes {
            let Some(proposed_key) = conflict_index_key(fields, &index.columns) else {
                continue;
            };
            for entity in manager.query_all(|_| true) {
                let Some(record) = runtime_any_record_from_entity_ref(&entity) else {
                    continue;
                };
                let existing_fields = record
                    .iter_fields()
                    .map(|(name, value)| (name.as_ref().to_string(), value.clone()))
                    .collect::<Vec<_>>();
                if self.snapshot_manager().is_aborted(entity.xmin) {
                    self.index_store_ref()
                        .index_entity_delete(collection, entity.id, &existing_fields)
                        .map_err(RedDBError::Internal)?;
                    continue;
                }
                if conflict_index_key(&existing_fields, &index.columns)
                    .is_some_and(|existing_key| existing_key == proposed_key)
                {
                    let conflicting_xid = entity.xmin;
                    if self.snapshot_manager().is_active(conflicting_xid)
                        && !self.own_transaction_xids().contains(&conflicting_xid)
                    {
                        return Err(RedDBError::Query(format!(
                            "serialization conflict: ON CONFLICT key in '{collection}' is owned by active transaction {conflicting_xid}; retry the statement after that transaction resolves"
                        )));
                    }
                    return Ok(Some(entity.id));
                }
            }
        }
        Ok(None)
    }

    fn on_conflict_unique_index_conflict_id(
        &self,
        collection: &str,
        fields: &[(String, Value)],
        target: Option<&[String]>,
    ) -> RedDBResult<Option<EntityId>> {
        let selected = self.unique_index_conflict_id(collection, fields, target)?;
        if selected.is_some() || target.is_none() {
            return Ok(selected);
        }
        if self
            .unique_index_conflict_id(collection, fields, None)?
            .is_some()
        {
            return Err(RedDBError::Query(format!(
                "unique index conflict on collection '{collection}' is not covered by ON CONFLICT ({})",
                target.unwrap_or_default().join(", ")
            )));
        }
        Ok(None)
    }

    fn has_unique_index_conflict_with_fields(
        &self,
        collection: &str,
        fields: &[(String, Value)],
        existing_fields: &[Vec<(String, Value)>],
        target: Option<&[String]>,
    ) -> RedDBResult<bool> {
        for index in self.conflict_unique_indexes(collection, target)? {
            let Some(proposed_key) = conflict_index_key(fields, &index.columns) else {
                continue;
            };
            if existing_fields.iter().any(|existing| {
                conflict_index_key(existing, &index.columns)
                    .is_some_and(|existing_key| existing_key == proposed_key)
            }) {
                return Ok(true);
            }
        }
        Ok(false)
    }

    fn has_on_conflict_unique_index_conflict_with_fields(
        &self,
        collection: &str,
        fields: &[(String, Value)],
        existing_fields: &[Vec<(String, Value)>],
        target: Option<&[String]>,
    ) -> RedDBResult<bool> {
        let selected = self.has_unique_index_conflict_with_fields(
            collection,
            fields,
            existing_fields,
            target,
        )?;
        if selected || target.is_none() {
            return Ok(selected);
        }
        if self.has_unique_index_conflict_with_fields(collection, fields, existing_fields, None)? {
            return Err(RedDBError::Query(format!(
                "unique index conflict on collection '{collection}' is not covered by ON CONFLICT ({})",
                target.unwrap_or_default().join(", ")
            )));
        }
        Ok(false)
    }

    fn execute_unique_index_conflict_update(
        &self,
        raw_query: &str,
        insert: &InsertQuery,
        conflict_id: EntityId,
        target: UpdateTarget,
        excluded_fields: &[(String, Value)],
        assignments: &[(String, Expr)],
    ) -> RedDBResult<RuntimeQueryResult> {
        let assignment_exprs = assignments
            .iter()
            .map(|(column, expr)| {
                Ok((
                    column.clone(),
                    resolve_excluded_assignment(expr, excluded_fields)?,
                ))
            })
            .collect::<RedDBResult<Vec<_>>>()?;
        let literal_assignments = assignment_exprs
            .iter()
            .filter_map(|(column, expr)| {
                fold_expr_to_value(expr.clone())
                    .ok()
                    .map(|value| (column.clone(), value))
            })
            .collect();
        let update = UpdateQuery {
            table: insert.table.clone(),
            target,
            compound_assignment_ops: vec![None; assignment_exprs.len()],
            assignment_exprs,
            assignments: literal_assignments,
            where_expr: None,
            filter: Some(Filter::Compare {
                field: FieldRef::column("", "rid"),
                op: CompareOp::Eq,
                value: Value::UnsignedInteger(conflict_id.raw()),
            }),
            ttl_ms: None,
            expires_at_ms: None,
            with_metadata: Vec::new(),
            returning: insert.returning.clone(),
            claim_limit: None,
            claim_exact: false,
            order_by: Vec::new(),
            limit: None,
            suppress_events: insert.suppress_events,
        };
        self.execute_update(raw_query, &update)
    }

    /// ADR 0067 (#1710): resolve the model of an unmarked bare-VALUES
    /// INSERT from the catalog, making the marker rule real — *a model
    /// marker exists only where it disambiguates what the catalog cannot
    /// know.*
    ///
    /// `INSERT INTO c VALUES ({…})` has no column list and no model marker,
    /// so the parser leaves `entity_type = Row` with an empty column list.
    /// Only that exact shape is a candidate for inference — every other
    /// INSERT (an explicit column list, or an explicit `DOCUMENT` / `NODE`
    /// / `VECTOR` / … marker) is returned untouched.
    ///
    /// * existing **document** collection → rewritten to a `DOCUMENT`
    ///   insert so the body routes through document creation, exactly the
    ///   path the explicit marker takes;
    /// * existing **non-document** collection → the bare form does not
    ///   apply; reported with a model-oriented error;
    /// * unknown collection → didactic error naming both recourses (the
    ///   `DOCUMENT` assertion form or `CREATE DOCUMENT` first).
    fn infer_unmarked_document_insert(
        &self,
        query: &InsertQuery,
    ) -> RedDBResult<Option<InsertQuery>> {
        if !matches!(query.entity_type, InsertEntityType::Row) || !query.columns.is_empty() {
            return Ok(None);
        }
        match self
            .db()
            .collection_contract_arc(&query.table)
            .map(|contract| contract.declared_model)
        {
            Some(crate::catalog::CollectionModel::Document) => {
                let mut rewritten = query.clone();
                rewritten.entity_type = InsertEntityType::Document;
                rewritten.columns = vec!["body".to_string()];
                Ok(Some(rewritten))
            }
            Some(other) => Err(RedDBError::InvalidOperation(format!(
                "collection '{table}' is declared as '{model}'; the bare \
                 `INSERT INTO {table} VALUES (…)` form is the document body shorthand — \
                 write an explicit column list (`INSERT INTO {table} (col, …) VALUES (…)`) \
                 to insert into a {model} collection",
                table = query.table,
                model = crate::runtime::ddl::polymorphic_resolver::model_name(other),
            ))),
            None => Err(RedDBError::InvalidOperation(format!(
                "collection '{table}' does not exist and the INSERT carries no model marker; \
                 write `INSERT INTO {table} DOCUMENT VALUES ({{…}})` to create it as a document \
                 (idempotent), or run `CREATE DOCUMENT {table}` first",
                table = query.table,
            ))),
        }
    }

    /// Resolve the model of an unmarked UPDATE from the catalog (ADR 0067,
    /// #1711). The `DOCUMENTS` / `ROWS` / `KV` UPDATE markers were removed, so
    /// an unmarked UPDATE parses to the `Rows` target; the catalog is the
    /// single source of truth for the collection's model. An existing document
    /// collection is rewritten to the `Documents` target (the runtime routes
    /// body / patch semantics off the target the explicit marker used to set);
    /// a KV collection to `Kv` so the key-immutability guard still fires. Table
    /// and unknown collections keep the `Rows` default.
    ///
    /// Dotted assignment targets (`SET a.b.c = …`) parse for every target — the
    /// model is not a parse-time fact — but they are legal only on a document
    /// collection and rejected here with a clear model-oriented error off it.
    fn resolve_unmarked_update_target(
        &self,
        query: &UpdateQuery,
    ) -> RedDBResult<Option<UpdateQuery>> {
        // NODES / EDGES are explicit, user-declared graph targets — never
        // inferred. Dotted paths off a graph target are still off-model.
        if matches!(query.target, UpdateTarget::Nodes | UpdateTarget::Edges) {
            ensure_update_dotted_targets_allowed(query, false)?;
            return Ok(None);
        }
        let declared_model = self
            .db()
            .collection_contract_arc(&query.table)
            .map(|contract| contract.declared_model);
        let is_document = matches!(
            declared_model,
            Some(crate::catalog::CollectionModel::Document)
        );
        ensure_update_dotted_targets_allowed(query, is_document)?;
        let inferred = match declared_model {
            Some(crate::catalog::CollectionModel::Document) => UpdateTarget::Documents,
            Some(crate::catalog::CollectionModel::Kv) => UpdateTarget::Kv,
            _ => return Ok(None),
        };
        let mut rewritten = query.clone();
        rewritten.target = inferred;
        Ok(Some(rewritten))
    }

    /// 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)?;
        let constraint_lock = crate::application::collection_contract_enforcer::row_constraint_lock(
            self,
            &query.table,
        );
        let _constraint_guard = constraint_lock.as_ref().map(|lock| lock.lock());
        // 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,
        )?;
        // ADR 0067 (#1710): catalog model inference for the unmarked
        // bare-VALUES INSERT. `INSERT INTO c VALUES ({…})` carries no
        // column list and no model marker (parsed as a Row insert with an
        // empty column list); resolve the model from the catalog so an
        // existing document collection routes to document creation and an
        // unknown collection surfaces a didactic error. Runs before tenant
        // injection so a rewritten document insert is tenant-scoped like
        // the explicit `DOCUMENT` marker path.
        let inferred_owned;
        let query = match self.infer_unmarked_document_insert(query)? {
            Some(rewritten) => {
                inferred_owned = rewritten;
                &inferred_owned
            }
            None => query,
        };
        // 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. Explicit values must match the active transport/session
        // tenant; an unbound platform caller may still write on behalf of a
        // tenant. 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,
        };
        if query.on_conflict.is_some()
            && !matches!(
                query.entity_type,
                InsertEntityType::Row
                    | InsertEntityType::Document
                    | InsertEntityType::Node
                    | InsertEntityType::Edge
            )
        {
            return Err(RedDBError::InvalidOperation(
                "ON CONFLICT is supported for table rows, documents, graph nodes, and graph edges"
                    .to_string(),
            ));
        }
        self.check_insert_column_policy(query)?;
        let resolved_auto_embed = if let Some(ref embed_config) = query.auto_embed {
            // Empty provider → resolve via the embeddings task pointer
            // (ADR-0068 §5); an explicit `USING` overrides it. A modality-
            // incapable provider fails didactically here.
            let provider =
                crate::ai::resolve_embeddings_provider_from_runtime(self, &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 connection = if matches!(provider, crate::ai::AiProvider::Local) {
                None
            } else {
                Some(crate::ai::resolve_connection_from_runtime(
                    &provider, None, self,
                )?)
            };
            let model = crate::ai::resolve_embeddings_model_from_runtime(
                self,
                &provider,
                embed_config.model.as_deref(),
            );
            Some((provider, model, connection, &embed_config.fields))
        } else {
            None
        };

        let mut inserted_count: u64 = 0;
        let effective_rows =
            effective_insert_rows(query).map_err(|msg| RedDBError::Query(msg.to_string()))?;
        if effective_rows.len() > 1
            && matches!(
                query.on_conflict.as_ref().map(|clause| &clause.action),
                Some(OnConflictAction::DoUpdate { .. })
            )
        {
            return Err(RedDBError::InvalidOperation(
                "multi-row ON CONFLICT DO UPDATE is not supported atomically; execute one proposed row per statement"
                    .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 contract = self.db().collection_contract_arc(&query.table);
        let declared_model = contract.as_ref().map(|contract| contract.declared_model);
        let secret_columns: Vec<bool> = query
            .columns
            .iter()
            .map(|column| {
                contract.as_ref().is_some_and(|contract| {
                    contract.declared_columns.iter().any(|declared| {
                        declared.name == *column
                            && declared.data_type.eq_ignore_ascii_case("secret")
                    })
                })
            })
            .collect();
        if query.on_conflict.is_some()
            && matches!(
                declared_model,
                Some(crate::catalog::CollectionModel::TimeSeries)
            )
        {
            return Err(RedDBError::InvalidOperation(
                "ON CONFLICT is not supported for time-series collections".to_string(),
            ));
        }

        // One collection-scoped acquisition per upsert statement. Conflict
        // discovery currently scans the collection, so the lock is not the
        // dominant cost; it closes the check-then-insert race for every model
        // until unique indexes become the atomic arbiter themselves.
        let conflict_rmw_lock = query.on_conflict.as_ref().map(|_| {
            self.inner
                .rmw_locks
                .lock_for(&query.table, "__collection_on_conflict_insert__")
        });
        let _conflict_rmw_guard = conflict_rmw_lock.as_ref().map(|lock| lock.lock());

        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;
        let mut conflict_returning_records: Vec<UnifiedRecord> = Vec::new();
        let mut auto_embed_ids = Vec::new();

        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::try_is_chain(&store, &query.table)?;
            let _chain_lock_arc: Option<Arc<super::RmwRowLock>> = 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::try_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());
            let mut accepted_row_fields: Vec<Vec<(String, Value)>> = Vec::new();
            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, &secret_columns, &query.columns, row_values)?;
                if query
                    .on_conflict
                    .as_ref()
                    .is_some_and(|clause| clause.target.is_none())
                {
                    fields = crate::application::collection_contract_enforcer::CollectionContractWriteEnforcer::new(
                        self,
                        &self.db(),
                        &query.table,
                    )
                    .normalize_insert_fields(fields)?;
                }
                if let Some(clause) = &query.on_conflict {
                    let conflict_id = crate::application::collection_contract_enforcer::CollectionContractWriteEnforcer::new(
                        self,
                        &self.db(),
                        &query.table,
                    )
                    .row_uniqueness_conflict_id(&fields, clause.target.as_deref())?
                    .filter(|conflict_id| {
                        store.get(&query.table, *conflict_id).is_some_and(|entity| {
                            !self.snapshot_manager().is_aborted(entity.xmin)
                        })
                    });
                    if let Some(conflict_id) = conflict_id {
                        if let Some(conflicting) = store.get(&query.table, conflict_id) {
                            let conflicting_xid = conflicting.xmin;
                            if self.snapshot_manager().is_active(conflicting_xid)
                                && !self.own_transaction_xids().contains(&conflicting_xid)
                            {
                                return Err(RedDBError::Query(format!(
                                    "serialization conflict: ON CONFLICT key in '{}' is owned by active transaction {conflicting_xid}; retry the statement after that transaction resolves",
                                    query.table
                                )));
                            }
                        }
                        match &clause.action {
                            OnConflictAction::DoNothing => continue,
                            OnConflictAction::DoUpdate { assignments } => {
                                let assignment_exprs = assignments
                                    .iter()
                                    .map(|(column, expr)| {
                                        Ok((
                                            column.clone(),
                                            resolve_excluded_assignment(expr, &fields)?,
                                        ))
                                    })
                                    .collect::<RedDBResult<Vec<_>>>()?;
                                let literal_assignments = assignment_exprs
                                    .iter()
                                    .filter_map(|(column, expr)| {
                                        fold_expr_to_value(expr.clone())
                                            .ok()
                                            .map(|value| (column.clone(), value))
                                    })
                                    .collect();
                                let update = UpdateQuery {
                                    table: query.table.clone(),
                                    target: UpdateTarget::Rows,
                                    compound_assignment_ops: vec![None; assignment_exprs.len()],
                                    assignment_exprs,
                                    assignments: literal_assignments,
                                    where_expr: None,
                                    filter: Some(Filter::Compare {
                                        field: FieldRef::column("", "rid"),
                                        op: CompareOp::Eq,
                                        // Uniqueness locates a physical version; RQL rid is
                                        // the stable logical identity across updates.
                                        value: Value::UnsignedInteger(
                                            store
                                                .get(&query.table, conflict_id)
                                                .map(|entity| entity.logical_id().raw())
                                                .unwrap_or(conflict_id.raw()),
                                        ),
                                    }),
                                    ttl_ms: None,
                                    expires_at_ms: None,
                                    with_metadata: Vec::new(),
                                    returning: query.returning.clone(),
                                    claim_limit: None,
                                    claim_exact: false,
                                    order_by: Vec::new(),
                                    limit: None,
                                    suppress_events: query.suppress_events,
                                };
                                let updated = self.execute_update(raw_query, &update)?;
                                inserted_count += updated.affected_rows;
                                conflict_returning_records.extend(updated.result.records);
                                continue;
                            }
                        }
                    }
                    if matches!(&clause.action, OnConflictAction::DoNothing)
                        && crate::application::collection_contract_enforcer::CollectionContractWriteEnforcer::new(
                            self,
                            &self.db(),
                            &query.table,
                        )
                        .has_row_uniqueness_conflict_with_rows(
                            &fields,
                            &accepted_row_fields,
                            clause.target.as_deref(),
                        )?
                    {
                        continue;
                    }
                }
                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::try_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::try_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());
                }
                accepted_row_fields.push(fields.clone());
                rows.push(CreateRowInput {
                    collection: query.table.clone(),
                    fields,
                    metadata,
                    node_links: Vec::new(),
                    vector_links: Vec::new(),
                });
            }
            let batch = CreateRowsBatchInput {
                collection: query.table.clone(),
                rows,
                suppress_events: query.suppress_events,
            };
            let outputs = if query
                .on_conflict
                .as_ref()
                .is_some_and(|clause| clause.target.is_none())
            {
                self.create_rows_batch_prevalidated_with_outputs(batch)?
            } else {
                self.create_rows_batch(batch)?
            };
            inserted_count += outputs.len() as u64;
            if query.auto_embed.is_some() {
                auto_embed_ids.extend(outputs.iter().map(|output| output.id));
            }

            // 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);
                let mut result = build_returning_result(items, &snaps, Some(&outputs));
                result
                    .records
                    .splice(0..0, conflict_returning_records.drain(..));
                returning_result = Some(result);
            }
        } 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());
                let mut accepted_graph_fields: Vec<Vec<(String, Value)>> = Vec::new();
                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,
                                &secret_columns,
                                &query.columns,
                                row_values,
                            )?;
                            if let Some(clause) = &query.on_conflict {
                                if let Some(conflict_id) = self
                                    .on_conflict_unique_index_conflict_id(
                                        &query.table,
                                        &node_values,
                                        clause.target.as_deref(),
                                    )?
                                {
                                    match &clause.action {
                                        OnConflictAction::DoNothing => continue,
                                        OnConflictAction::DoUpdate { assignments } => {
                                            let updated = self
                                                .execute_unique_index_conflict_update(
                                                    raw_query,
                                                    query,
                                                    conflict_id,
                                                    UpdateTarget::Nodes,
                                                    &node_values,
                                                    assignments,
                                                )?;
                                            inserted_count += updated.affected_rows;
                                            conflict_returning_records
                                                .extend(updated.result.records);
                                            continue;
                                        }
                                    }
                                }
                                if matches!(&clause.action, OnConflictAction::DoNothing)
                                    && self.has_on_conflict_unique_index_conflict_with_fields(
                                        &query.table,
                                        &node_values,
                                        &accepted_graph_fields,
                                        clause.target.as_deref(),
                                    )?
                                {
                                    continue;
                                }
                            }
                            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),
                            )?;
                            accepted_graph_fields.push(node_values.clone());
                            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,
                                &secret_columns,
                                &query.columns,
                                row_values,
                            )?;
                            if let Some(clause) = &query.on_conflict {
                                if let Some(conflict_id) = self
                                    .on_conflict_unique_index_conflict_id(
                                        &query.table,
                                        &edge_values,
                                        clause.target.as_deref(),
                                    )?
                                {
                                    match &clause.action {
                                        OnConflictAction::DoNothing => continue,
                                        OnConflictAction::DoUpdate { assignments } => {
                                            let updated = self
                                                .execute_unique_index_conflict_update(
                                                    raw_query,
                                                    query,
                                                    conflict_id,
                                                    UpdateTarget::Edges,
                                                    &edge_values,
                                                    assignments,
                                                )?;
                                            inserted_count += updated.affected_rows;
                                            conflict_returning_records
                                                .extend(updated.result.records);
                                            continue;
                                        }
                                    }
                                }
                                if matches!(&clause.action, OnConflictAction::DoNothing)
                                    && self.has_on_conflict_unique_index_conflict_with_fields(
                                        &query.table,
                                        &edge_values,
                                        &accepted_graph_fields,
                                        clause.target.as_deref(),
                                    )?
                                {
                                    continue;
                                }
                            }
                            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),
                            )?;
                            accepted_graph_fields.push(edge_values.clone());
                            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();
                let mut graph_index_fields: Vec<Vec<(String, Value)>> =
                    Vec::with_capacity(prepared.len());
                for item in prepared {
                    match item {
                        PreparedGraphInsert::Node { fields, input } => {
                            if query.returning.is_some() {
                                returning_field_snaps.push(fields.clone());
                            }
                            graph_index_fields.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.clone());
                            }
                            graph_index_fields.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 !graph_index_fields.is_empty() {
                    let index_rows: Vec<_> = ids
                        .iter()
                        .zip(graph_index_fields)
                        .map(|(id, fields)| (*id, fields))
                        .collect();
                    self.inner
                        .index_store
                        .index_entity_insert_batch(&query.table, &index_rows)
                        .map_err(RedDBError::Internal)?;
                }
                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,
                                &secret_columns,
                                &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,
                                &secret_columns,
                                &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,
                                &secret_columns,
                                &query.columns,
                                row_values,
                            )?;
                            let (columns, values) = pairwise_columns_values(&document_values);
                            let body = find_document_body_json(&columns, &values)?;
                            let conflict_fields =
                                document_on_conflict_fields(&document_values, &body)?;
                            if let Some(clause) = &query.on_conflict {
                                if let Some(conflict_id) = self
                                    .on_conflict_unique_index_conflict_id(
                                        &query.table,
                                        &conflict_fields,
                                        clause.target.as_deref(),
                                    )?
                                {
                                    match &clause.action {
                                        OnConflictAction::DoNothing => continue,
                                        OnConflictAction::DoUpdate { assignments } => {
                                            let updated = self
                                                .execute_unique_index_conflict_update(
                                                    raw_query,
                                                    query,
                                                    conflict_id,
                                                    UpdateTarget::Documents,
                                                    &conflict_fields,
                                                    assignments,
                                                )?;
                                            inserted_count += updated.affected_rows;
                                            conflict_returning_records
                                                .extend(updated.result.records);
                                            continue;
                                        }
                                    }
                                }
                            }
                            merge_with_clauses(
                                &mut metadata,
                                query.ttl_ms,
                                query.expires_at_ms,
                                &query.with_metadata,
                            );
                            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,
                                &secret_columns,
                                &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 query.auto_embed.is_some() {
                auto_embed_ids.extend(entity_outputs.iter().map(|output| output.id));
            }

            if let Some(items) = query.returning.as_ref() {
                let mut result =
                    build_returning_result(items, &returning_field_snaps, Some(&entity_outputs));
                result
                    .records
                    .splice(0..0, conflict_returning_records.drain(..));
                returning_result = Some(result);
            }
        }

        // Auto-embed pipeline: batch-embed fields across all inserted rows via AiBatchClient.
        if let Some((provider, model, connection, fields)) = resolved_auto_embed.as_ref() {
            let store = self.inner.db.store();
            let is_local_provider = matches!(provider, crate::ai::AiProvider::Local);
            let (api_key, api_base) = connection
                .as_ref()
                .map(|connection| (connection.api_key.as_str(), connection.api_base.as_str()))
                .unwrap_or(("", ""));

            // Read only this INSERT's returned IDs. A pre-write statement
            // snapshot correctly excludes newly committed rows; a fresh/global
            // scan could instead pick another writer's rows or conflict skips.
            let recent: Vec<_> = auto_embed_ids
                .iter()
                .filter_map(|id| store.get(&query.table, *id))
                .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 {
                        let texts: Vec<String> = fields
                            .iter()
                            .filter_map(|field| match row.get_field(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_at(
                                provider,
                                model,
                                api_key,
                                api_base,
                                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,
        mut fields: Vec<(String, Value)>,
        mut metadata: Vec<(String, MetadataValue)>,
    ) -> RedDBResult<EntityId> {
        self.check_collection_privilege(reddb_rql::ast::PolicyAction::Insert, collection)?;
        crate::runtime::mutation::enforce_tenant_fields(self, collection, &mut fields)?;
        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 fields = extract_remaining_properties(
            &columns,
            &values,
            &[
                "metric",
                "value",
                "tags",
                "timestamp",
                "timestamp_ns",
                "time",
                "event_name",
                "payload",
            ],
        );
        let index_fields = fields.clone();
        let _segment_reservation = self.admit_non_evictable_growth(
            crate::storage::memory_pools::MemoryPool::SegmentArena,
            &format!("insert into {collection}"),
            crate::runtime::memory_admission::estimate_timeseries_point_growth(
                &metric, &tags, &fields,
            ),
        )?;
        let indexed_columns = self
            .inner
            .index_store
            .list_indices(collection)
            .into_iter()
            .flat_map(|index| index.columns)
            .collect::<Vec<_>>();
        let _index_reservation = if indexed_columns.is_empty() {
            None
        } else {
            Some(self.admit_non_evictable_growth(
                crate::storage::memory_pools::MemoryPool::IndexMemory,
                &format!("index insert into {collection}"),
                crate::runtime::memory_admission::estimate_index_growth(
                    std::slice::from_ref(&index_fields),
                    &indexed_columns,
                ),
            )?)
        };
        let series_id = super::impl_timeseries::intern_timeseries_series(
            self.inner.db.store().as_ref(),
            collection,
            &metric,
            &tags,
        )?;

        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,
                series_id: Some(series_id),
                timestamp_ns,
                value,
                tags: HashMap::new(),
                fields: fields.into_iter().collect(),
            }),
        );
        if !crate::runtime::rls_injection::entity_passes_rls_action(
            self,
            collection,
            reddb_rql::ast::PolicyAction::Insert,
            &entity,
        ) {
            return Err(RedDBError::Query(format!(
                "insert into collection '{collection}' denied by row-level security"
            )));
        }
        // 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()))?;
        self.inner
            .index_store
            .index_entity_insert(collection, id, &index_fields)
            .map_err(RedDBError::Internal)?;

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

        let _ = self.inner.db.hypertables().route(collection, timestamp_ns);

        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)?;
        let constraint_lock = crate::application::collection_contract_enforcer::row_constraint_lock(
            self,
            &query.table,
        );
        let _constraint_guard = constraint_lock.as_ref().map(|lock| lock.lock());
        // 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::try_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);
        }
        // ADR 0067 (#1711): the DOCUMENTS / ROWS / KV UPDATE markers were
        // removed. Resolve the model of an unmarked UPDATE from the catalog so
        // a document collection routes to document semantics and a KV
        // collection keeps its key-immutability guard, and gate dotted SET
        // targets that only a document body can satisfy. Runs before the
        // contract / RLS gates so every downstream path — the inner scan,
        // RETURNING — observes the resolved target.
        let inferred_owned;
        let query = match self.resolve_unmarked_update_target(query)? {
            Some(rewritten) => {
                inferred_owned = rewritten;
                &inferred_owned
            }
            None => 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,
                reddb_rql::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,
                    reddb_rql::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 = reddb_rql::ast::Filter::And(Box::new(read_policy), Box::new(policy));
            }
            let mut augmented = query.clone();
            augmented.filter = Some(match augmented.filter.take() {
                Some(existing) => reddb_rql::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 topology_lock = self
            .index_store_ref()
            .collection_topology_lock(&query.table);
        let effective_filter = effective_update_filter(query);
        let compiled_plan = self.compile_update_plan(query)?;
        // Autocommit table-row UPDATE replaces an MVCC version even when its
        // assignments are constants. Resolve the current pre-image under the
        // RMW lock. Explicit transactions keep their statement snapshot and
        // first-committer-wins conflict detection.
        let autocommit_table_update =
            matches!(query.target, UpdateTarget::Rows) && self.current_xid().is_none();
        let needs_rmw_lock = autocommit_table_update || 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 && !autocommit_table_update {
            target_scan = target_scan.with_live_table_rows();
        }
        let ids_to_update = {
            let _topology_guard = topology_lock.read();
            // The statement snapshot predates the table lock. Refresh it
            // after acquiring that lock so earlier autocommits are visible,
            // but never select an uncommitted physical tip. Historical
            // versions can be absent from current indexes while a writer is
            // active, so retain the snapshot-aware index fallback then.
            let _fresh_snapshot = if autocommit_table_update {
                let mut ctx =
                    crate::runtime::impl_core::capture_current_snapshot().unwrap_or_else(|| {
                        let manager = self.snapshot_manager();
                        crate::runtime::impl_core::SnapshotContext {
                            snapshot: manager.fresh_read_snapshot(),
                            manager,
                            own_xids: std::collections::HashSet::new(),
                            requires_index_fallback: false,
                            serializable_reader: None,
                        }
                    });
                ctx.snapshot = ctx.manager.fresh_read_snapshot();
                ctx.requires_index_fallback |= !ctx.snapshot.in_progress.is_empty();
                Some(crate::runtime::impl_core::CurrentSnapshotGuard::install(
                    ctx,
                ))
            } else {
                None
            };
            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() || autocommit_table_update {
            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);
        }

        // Keep topology fixed while reserving every post-image. No chunk is
        // published until all targets fit; a later denial cannot leave earlier
        // autocommit chunks visible. Prepared payload is bounded by admission.
        let topology_guard = topology_lock.read();
        let mut prepared = Vec::new();
        let mut reservations = super::memory_admission::MutationMemoryReservations::new(self);
        for chunk in ids_to_update.chunks(UPDATE_APPLY_CHUNK_SIZE) {
            for id in chunk {
                #[cfg(test)]
                self.index_store_ref().mutation_test_hook(
                    &query.table,
                    super::index_store::MutationTestPhase::DmlPostImageCandidate,
                );
                super::query_budget::charge(1)?;
                let Some(entity) = manager.try_get(*id, usize::MAX)? else {
                    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,
                )?;
                reservations.admit(&applied)?;
                touched_ids.push(applied.id);
                prepared.push(applied);
            }
        }
        let affected = prepared.len() as u64;
        let mut published = 0;
        let mut _durable_publication = None;
        let persistence = (|| -> RedDBResult<()> {
            super::query_budget::charge(0)?;
            // Standalone UNIQUE keys are admitted for the whole statement
            // before any chunk is published.
            self.enforce_unique_update_post_images(&query.table, &compiled_plan, &prepared)?;
            // This is the cancellation boundary. Once durable publication
            // starts, finish the prepared statement rather than exposing a
            // timeout with only a prefix of its chunks committed.
            _durable_publication = Some(super::query_budget::begin_durable_publication()?);
            for chunk in prepared.chunks(UPDATE_APPLY_CHUNK_SIZE) {
                if compiled_plan.row_touches_unique_columns
                    || compiled_plan
                        .row_contract_plan
                        .as_ref()
                        .is_some_and(|plan| plan.has_expressions)
                {
                    let db = self.db();
                    let contract = crate::application::collection_contract_enforcer::CollectionContractWriteEnforcer::new(self, &db, &query.table);
                    for item in chunk {
                        let fields =
                            crate::application::ports::entity_row_fields_snapshot(&item.entity);
                        let previous_id =
                            item.replaced_entity.as_ref().map_or(item.id, |old| old.id);
                        contract.enforce_row_uniqueness(&fields, Some(previous_id))?;
                    }
                }
                self.persist_update_chunk(chunk)?;
                published += chunk.len();
            }
            Ok(())
        })();
        // A later storage/constraint error must still finish index and CDC
        // maintenance for earlier published chunks, as the chunked path did.
        let published_mutations = &prepared[..published];
        let lsns = self.flush_update_chunk(published_mutations, topology_guard)?;
        if !query.suppress_events {
            self.emit_update_events_for_collection(&query.table, published_mutations, &lsns)?;
        }
        persistence?;
        drop(reservations);

        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>,
    ) -> RedDBResult<Vec<EntityId>> {
        let conn_id = current_connection_id();
        let locks = self.inner.pending_claim_locks.read();
        if locks.is_empty() {
            return Ok(ids);
        }
        let mut admitted = Vec::with_capacity(ids.len());
        for id in ids {
            super::query_budget::charge(1)?;
            if let Some(entity) = self.inner.db.store().try_get(table, id, usize::MAX)? {
                let key = (table.to_string(), entity.logical_id());
                if locks.get(&key).is_none_or(|owner| *owner == conn_id) {
                    admitted.push(id);
                }
            }
        }
        Ok(admitted)
    }

    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>)> {
        // Existing table/claim locks precede topology; per-row RMW locks
        // below follow it. Keep the same index set through persistence/flush.
        let topology_lock = self
            .index_store_ref()
            .collection_topology_lock(&query.table);
        let topology_guard = topology_lock.read();
        let store = self.inner.db.store();
        let mut touched_ids = Vec::new();
        let mut lock_entries = Vec::new();

        for id in ids_to_update {
            super::query_budget::charge(1)?;
            let Some(candidate) = store.try_get(&query.table, *id, usize::MAX)? else {
                continue;
            };
            let logical_id = candidate.logical_id();
            let lock_key = super::RmwLockTable::row_key(logical_id);
            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();
        let mut reservations = super::memory_admission::MutationMemoryReservations::new(self);
        for (_, logical_id, _) in &lock_entries {
            #[cfg(test)]
            self.index_store_ref().mutation_test_hook(
                &query.table,
                super::index_store::MutationTestPhase::DmlPostImageCandidate,
            );
            super::query_budget::charge(1)?;
            let Some(entity) =
                try_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_result_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,
            )?;
            reservations.admit(&applied)?;
            touched_ids.push(applied.id);
            applied_chunk.push(applied);
        }

        let affected = applied_chunk.len() as u64;
        let mut published = 0;
        let mut _durable_publication = None;
        let persistence = (|| -> RedDBResult<()> {
            super::query_budget::charge(0)?;
            self.enforce_unique_update_post_images(&query.table, compiled_plan, &applied_chunk)?;
            // Final cooperative checkpoint before the first durable chunk.
            // Cancellation after this point cannot safely abort a prefix.
            _durable_publication = Some(super::query_budget::begin_durable_publication()?);
            for chunk in applied_chunk.chunks(UPDATE_APPLY_CHUNK_SIZE) {
                if compiled_plan.row_touches_unique_columns
                    || compiled_plan
                        .row_contract_plan
                        .as_ref()
                        .is_some_and(|plan| plan.has_expressions)
                {
                    let db = self.db();
                    let contract = crate::application::collection_contract_enforcer::CollectionContractWriteEnforcer::new(self, &db, &query.table);
                    for item in chunk {
                        let fields =
                            crate::application::ports::entity_row_fields_snapshot(&item.entity);
                        let previous_id =
                            item.replaced_entity.as_ref().map_or(item.id, |old| old.id);
                        contract.enforce_row_uniqueness(&fields, Some(previous_id))?;
                    }
                }
                self.persist_update_chunk(chunk)?;
                published += chunk.len();
            }
            Ok(())
        })();
        // Earlier chunks may have been published when a later constraint
        // check fails. Finish their indexes and events before returning it.
        let published_mutations = &applied_chunk[..published];
        let lsns = self.flush_update_chunk(published_mutations, topology_guard)?;
        if !query.suppress_events {
            self.emit_update_events_for_collection(&query.table, published_mutations, &lsns)?;
        }
        persistence?;

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

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

    fn enforce_unique_update_post_images(
        &self,
        collection: &str,
        compiled_plan: &CompiledUpdatePlan,
        applied: &[AppliedEntityMutation],
    ) -> RedDBResult<()> {
        if !compiled_plan.touches_unique_index {
            return Ok(());
        }
        let post_images: Vec<_> = applied
            .iter()
            .map(|item| {
                (
                    item.replaced_entity.as_ref().map_or(item.id, |old| old.id),
                    crate::application::ports::entity_row_fields_snapshot(&item.entity),
                )
            })
            .collect();
        self.enforce_unique_updates(collection, &post_images)
    }

    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()) {
                    // Without a declared type, a text constant may overwrite an
                    // existing Secret. Resolve it per row to preserve encryption.
                    let unknown_text_target = metadata_key.is_none()
                        && matches!(value, Value::Text(_))
                        && row_contract_plan
                            .as_ref()
                            .and_then(|plan| plan.declared_rules.get(column))
                            .is_none();
                    if !unknown_text_target {
                        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 is_secret_column = row_contract_plan
                                .as_ref()
                                .and_then(|plan| plan.declared_rules.get(column))
                                .is_some_and(|rule| {
                                    rule.data_type == reddb_types::DataType::Secret
                                });
                            let value =
                                self.resolve_secret_column_value(value, is_secret_column)?;
                            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))
            })
        });

        let touches_unique_index = self
            .index_store_ref()
            .touches_unique_index(&query.table, &row_modified_columns);

        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,
            touches_unique_index,
        })
    }

    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::try_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
            };

            let sensitive = super::join_filter::expression_uses_secret(
                &assignment.expr,
                record,
                Some(&query.table),
                Some(&query.table),
            );
            let secret_target = assignment
                .row_rule
                .as_ref()
                .is_some_and(|rule| rule.data_type == reddb_types::DataType::Secret)
                || record
                    .get(&assignment.column)
                    .is_some_and(|value| matches!(value, Value::Secret(_)));
            if sensitive && !secret_target {
                return Err(RedDBError::Query(
                    "secret-derived UPDATE values require a SECRET target column".into(),
                ));
            }
            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_secret_column_value(
                            value,
                            assignment.row_rule.as_ref().is_some_and(|rule| {
                                rule.data_type == reddb_types::DataType::Secret
                            }) || record
                                .get(&assignment.column)
                                .is_some_and(|value| matches!(value, Value::Secret(_))),
                        )?,
                        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)?;
        let constraint_lock = crate::application::collection_contract_enforcer::row_constraint_lock(
            self,
            &query.table,
        );
        let _constraint_guard = constraint_lock.as_ref().map(|lock| lock.lock());
        // Issue #523 — blockchain collections are immutable; see
        // execute_update for the same gate.
        if crate::runtime::blockchain_kind::try_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,
                reddb_rql::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) => reddb_rql::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 maintenance_limit = super::retention_sweeper::take_delete_limit();
        let topology_lock = self
            .index_store_ref()
            .collection_topology_lock(&query.table);
        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(),
            maintenance_limit,
        );
        let ids_to_delete = {
            let _topology_guard = topology_lock.read();
            let ids = scan.find_target_ids()?;
            if ids.len() > UPDATE_APPLY_CHUNK_SIZE
                && self.current_xid().is_none()
                && UnifiedStore::deferred_store_wal_capture_active()
            {
                super::delete_publication::order_target_ids(self, &query.table, ids)?
            } else {
                ids
            }
        };

        let needs_delete_events =
            !query.suppress_events && self.collection_has_delete_subscriptions(&query.table);
        let mut affected: u64 = 0;
        // Final cooperative checkpoint before DELETE begins publishing its
        // prepared target set. A later timeout must not strand a partial
        // statement merely to stop sooner.
        let _durable_publication = super::query_budget::begin_durable_publication()?;
        for chunk in ids_to_delete.chunks(UPDATE_APPLY_CHUNK_SIZE) {
            let deleted = self.delete_entities_batch(
                &query.table,
                chunk,
                topology_lock.read(),
                effective_filter.as_ref(),
                needs_delete_events,
            )?;
            #[cfg(test)]
            self.index_store_ref().mutation_test_hook(
                &query.table,
                super::index_store::MutationTestPhase::BeforeEvents,
            );
            affected += deleted.affected;
            if needs_delete_events && !deleted.changes.is_empty() {
                let ids: Vec<_> = deleted.changes.iter().map(|(id, _)| *id).collect();
                let lsns: Vec<_> = deleted.changes.iter().map(|(_, lsn)| *lsn).collect();
                self.emit_delete_events_for_collection(
                    &query.table,
                    &ids,
                    &lsns,
                    &deleted.pre_images,
                )?;
            }
        }

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

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

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

    #[test]
    fn captured_bulk_delete_keeps_global_row_lock_order_across_chunks() {
        use std::sync::{
            atomic::{AtomicUsize, Ordering},
            Arc,
        };
        let runtime = Arc::new(
            RedDBRuntime::with_options(
                RedDBOptions::in_memory().with_memory_budget(128 * 1024 * 1024),
            )
            .expect("bulk lock fixture"),
        );
        runtime
            .execute_query("CREATE TABLE bulk_lock_order (id INTEGER)")
            .expect("table");
        let count = super::UPDATE_APPLY_CHUNK_SIZE + 16;
        for batch in (0..count).collect::<Vec<_>>().chunks(512) {
            let values = batch
                .iter()
                .map(|id| format!("({id})"))
                .collect::<Vec<_>>()
                .join(",");
            runtime
                .execute_query(&format!("INSERT INTO bulk_lock_order (id) VALUES {values}"))
                .expect("seed chunk boundary");
        }
        let manager = runtime
            .db()
            .store()
            .get_collection("bulk_lock_order")
            .expect("rows");
        // Imported/versioned rows can have logical identities independent
        // of their physical positions. Cross decimal-width boundaries too;
        // fresh time-based physical IDs alone keep the same string order.
        let mut rows = manager.query_all(|_| true);
        rows.sort_unstable_by_key(|entity| entity.id);
        for (position, mut entity) in rows.into_iter().enumerate() {
            entity.set_logical_id(crate::storage::EntityId::new(
                u64::try_from(position + 1).expect("small identity"),
            ));
            manager
                .update(entity)
                .expect("independent logical identity");
        }
        let mut expected: Vec<_> = manager
            .query_all(|_| true)
            .iter()
            .map(|entity| format!("row:{}", entity.logical_id().raw()))
            .collect();
        expected.sort_unstable();
        let calls = Arc::new(AtomicUsize::new(0));
        let hook_calls = Arc::clone(&calls);
        let weak_runtime = Arc::downgrade(&runtime);
        *runtime.index_store_ref().mutation_hook.lock() =
            Some(Arc::new(move |collection, phase| {
                if collection == "bulk_lock_order"
                    && phase == super::super::index_store::MutationTestPhase::BeforeEvents
                {
                    let runtime = weak_runtime.upgrade().expect("writer still alive");
                    let held =
                        super::super::delete_publication::pending_lock_keys(&runtime, collection);
                    assert!(!held.is_empty());
                    assert!(held.len() <= expected.len());
                    let mismatch = held
                        .iter()
                        .zip(&expected)
                        .position(|(actual, expected)| actual != expected);
                    assert!(
                        mismatch.is_none(),
                        "captured DELETE violated global lock order at position {mismatch:?}"
                    );
                    hook_calls.fetch_add(1, Ordering::SeqCst);
                }
            }));
        let result = runtime
            .with_deferred_store_wal_for_dml(true, || {
                runtime.execute_query("DELETE FROM bulk_lock_order")
            })
            .expect("captured bulk delete");
        assert_eq!(
            result.affected_rows,
            u64::try_from(count).expect("small fixture")
        );
        assert!(
            calls.load(Ordering::SeqCst) >= 2,
            "fixture must cross an apply chunk"
        );
        assert!(runtime
            .execute_query("SELECT id FROM bulk_lock_order")
            .expect("deleted rows")
            .result
            .records
            .is_empty());
        assert_eq!(runtime.snapshot_manager().oldest_active_xid(), None);
    }

    #[test]
    fn locked_update_recheck_read_failure_preserves_preimages_and_retry() {
        use crate::storage::engine::BTree;
        use reddb_engine::storage::collection_read::PagedCollection;
        use std::sync::Arc;

        let directory = tempfile::tempdir().expect("locked update fixture");
        let options = RedDBOptions::persistent(directory.path().join("locked-update.rdb"))
            .with_memory_budget(128 * 1024 * 1024)
            .with_storage_profile(
                crate::storage::profile::StorageDeployPreset::PrimaryReplicaDev.selection(),
            )
            .expect("locked update profile");
        let runtime = RedDBRuntime::with_options(options).expect("locked update runtime");
        for sql in [
            "CREATE TABLE recheck_probe (id INTEGER)",
            "INSERT INTO recheck_probe (id) VALUES (1), (2)",
            "SET CONFIG fixture.flag = 'ready'",
        ] {
            runtime.execute_query(sql).expect(sql);
        }
        let sql = "UPDATE recheck_probe SET id = id + 10 WHERE id = 1 OR CONFIG('fixture.flag') = 'ready'";
        let reddb_rql::ast::QueryExpr::Update(query) =
            reddb_rql::parse(sql).expect("parse update").query
        else {
            panic!("expected UPDATE");
        };
        let compiled = runtime.compile_update_plan(&query).expect("compile update");
        let filter = reddb_rql::sql_lowering::effective_update_filter(&query);
        let ids = super::super::dml_target_scan::DmlTargetScan::new(
            &runtime,
            "recheck_probe",
            filter.as_ref(),
            None,
        )
        .find_target_ids()
        .expect("discover targets before failure");
        assert_eq!(ids.len(), 2);
        let store = runtime.db().store();
        store.persist().expect("publish recheck configuration");
        let root = store
            .collection_root_page("red_config")
            .expect("configuration root");
        let manager = store
            .get_collection("red_config")
            .expect("configuration collection");
        let install = |limit| {
            let tree = BTree::with_root(Arc::clone(store.pager().expect("pager")), root);
            manager.install_backing(
                Arc::new(PagedCollection::new(Arc::new(tree))),
                root,
                store.format_version(),
                limit,
                manager.count() as u64,
            );
        };
        let assert_ids = |expected: [i64; 2]| {
            runtime.invalidate_result_cache();
            let rows = runtime
                .execute_query("SELECT id FROM recheck_probe ORDER BY id")
                .expect("read preimages");
            let actual: Vec<_> = rows
                .result
                .records
                .iter()
                .map(|row| row.get("id").cloned().expect("row value"))
                .collect();
            assert_eq!(actual, expected.map(Value::Integer));
        };
        // Candidate discovery has succeeded. Fail only the locked recheck,
        // after the first row's postimage can already have been prepared.
        install(1);
        let error = runtime
            .execute_update_inner_tracked_locked(sql, &query, &compiled, &ids, filter.as_ref())
            .err()
            .expect("locked recheck must propagate its read failure");
        assert!(error.to_string().contains("read memory limit"), "{error}");
        install(4096);
        assert_ids([1, 2]);
        let retried = runtime
            .execute_update_inner_tracked_locked(sql, &query, &compiled, &ids, filter.as_ref())
            .expect("retry locked recheck");
        assert_eq!(retried.0.affected_rows, 2);
        assert_ids([11, 12]);
    }

    #[test]
    fn concurrent_constant_updates_leave_one_visible_row_per_logical_id() {
        use std::sync::{Arc, Barrier};

        const WRITERS: usize = 3;
        const ROUNDS: usize = 8;
        let rt = Arc::new(RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap());
        rt.execute_query("CREATE TABLE capabilities (id TEXT, observed_at TEXT)")
            .unwrap();
        rt.execute_query("INSERT INTO capabilities (id, observed_at) VALUES ('A', 'initial')")
            .unwrap();
        let barrier = Arc::new(Barrier::new(WRITERS));
        let handles: Vec<_> = (0..WRITERS)
            .map(|writer| {
                let rt = Arc::clone(&rt);
                let barrier = Arc::clone(&barrier);
                std::thread::spawn(move || {
                    for round in 0..ROUNDS {
                        barrier.wait();
                        rt.execute_query(&format!(
                            "UPDATE capabilities SET observed_at = '{writer}-{round}' WHERE id = 'A'"
                        ))
                        .unwrap();
                    }
                })
            })
            .collect();
        for handle in handles {
            handle.join().unwrap();
        }
        let selected = rt
            .execute_query("SELECT id, observed_at FROM capabilities WHERE id = 'A'")
            .unwrap();
        assert_eq!(
            selected.result.records.len(),
            1,
            "concurrent updates published multiple visible versions"
        );
    }

    #[test]
    fn locked_constant_updates_preserve_unique_constraints() {
        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
        rt.execute_query("CREATE TABLE accounts (id INT PRIMARY KEY, email TEXT UNIQUE)")
            .unwrap();
        rt.execute_query("INSERT INTO accounts (id, email) VALUES (1, 'a'), (2, 'b')")
            .unwrap();
        assert!(rt
            .execute_query("UPDATE accounts SET email = 'b' WHERE id = 1")
            .is_err());
        let selected = rt
            .execute_query("SELECT id FROM accounts WHERE email = 'a'")
            .unwrap();
        assert_eq!(selected.result.records.len(), 1);
    }

    #[test]
    fn concurrent_transactions_do_not_both_commit_the_same_row_update() {
        use crate::runtime::mvcc::{clear_current_connection_id, set_current_connection_id};

        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
        rt.execute_query("CREATE TABLE capabilities (id TEXT, observed_at TEXT)")
            .unwrap();
        rt.execute_query("INSERT INTO capabilities (id, observed_at) VALUES ('A', 'initial')")
            .unwrap();
        for connection in [41, 42] {
            set_current_connection_id(connection);
            rt.execute_query("BEGIN").unwrap();
        }
        set_current_connection_id(41);
        rt.execute_query("UPDATE capabilities SET observed_at = 'first' WHERE id = 'A'")
            .unwrap();
        set_current_connection_id(42);
        let competing =
            rt.execute_query("UPDATE capabilities SET observed_at = 'second' WHERE id = 'A'");
        set_current_connection_id(41);
        rt.execute_query("COMMIT").unwrap();
        set_current_connection_id(42);
        if competing.is_ok() {
            assert!(
                rt.execute_query("COMMIT").is_err(),
                "both writers committed"
            );
        } else {
            rt.execute_query("ROLLBACK").unwrap();
        }
        clear_current_connection_id();
        let selected = rt
            .execute_query("SELECT id FROM capabilities WHERE id = 'A'")
            .unwrap();
        assert_eq!(selected.result.records.len(), 1);
    }

    #[test]
    fn transaction_and_autocommit_update_do_not_publish_two_live_versions() {
        use crate::runtime::mvcc::{clear_current_connection_id, set_current_connection_id};

        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
        rt.execute_query("CREATE TABLE capabilities (id TEXT, observed_at TEXT)")
            .unwrap();
        rt.execute_query("INSERT INTO capabilities (id, observed_at) VALUES ('A', 'initial')")
            .unwrap();
        set_current_connection_id(41);
        rt.execute_query("BEGIN").unwrap();
        rt.execute_query("UPDATE capabilities SET observed_at = 'transaction' WHERE id = 'A'")
            .unwrap();
        clear_current_connection_id();
        let autocommit =
            rt.execute_query("UPDATE capabilities SET observed_at = 'auto' WHERE id = 'A'");
        set_current_connection_id(41);
        let commit = rt.execute_query("COMMIT");
        clear_current_connection_id();
        assert!(
            autocommit.is_err() || commit.is_err(),
            "conflicting writers both committed"
        );
        let selected = rt
            .execute_query("SELECT id FROM capabilities WHERE id = 'A'")
            .unwrap();
        assert_eq!(selected.result.records.len(), 1);
    }

    #[test]
    fn transaction_update_does_not_target_rows_created_after_snapshot() {
        use crate::runtime::mvcc::{clear_current_connection_id, set_current_connection_id};

        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
        rt.execute_query("CREATE TABLE capabilities (id TEXT, observed_at TEXT)")
            .unwrap();
        set_current_connection_id(41);
        rt.execute_query("BEGIN").unwrap();
        rt.execute_query("SELECT id FROM capabilities").unwrap();
        clear_current_connection_id();
        rt.execute_query("INSERT INTO capabilities (id, observed_at) VALUES ('A', 'after')")
            .unwrap();
        set_current_connection_id(41);
        let updated = rt
            .execute_query("UPDATE capabilities SET observed_at = 'changed' WHERE id = 'A'")
            .unwrap();
        assert_eq!(
            updated.affected_rows, 0,
            "updated a row outside the transaction snapshot"
        );
        rt.execute_query("COMMIT").unwrap();
        clear_current_connection_id();
        let selected = rt
            .execute_query("SELECT observed_at FROM capabilities WHERE id = 'A'")
            .unwrap();
        assert_eq!(selected.result.records.len(), 1);
        assert_eq!(
            selected.result.records[0].get("observed_at"),
            Some(&Value::text("after"))
        );
    }

    #[test]
    fn autocommit_update_does_not_filter_on_uncommitted_row_image() {
        use crate::runtime::mvcc::{clear_current_connection_id, set_current_connection_id};

        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
        rt.execute_query("CREATE TABLE capabilities (id TEXT, status TEXT, payload TEXT)")
            .unwrap();
        rt.execute_query(
            "INSERT INTO capabilities (id, status, payload) VALUES ('A', 'ready', 'old')",
        )
        .unwrap();
        set_current_connection_id(41);
        rt.execute_query("BEGIN").unwrap();
        rt.execute_query("UPDATE capabilities SET status = 'busy' WHERE id = 'A'")
            .unwrap();
        clear_current_connection_id();
        let autocommit =
            rt.execute_query("UPDATE capabilities SET payload = 'new' WHERE status = 'ready'");
        if let Ok(result) = &autocommit {
            assert_eq!(
                result.affected_rows, 1,
                "uncommitted status hid the committed target"
            );
        }
        set_current_connection_id(41);
        rt.execute_query("ROLLBACK").unwrap();
        clear_current_connection_id();
        let selected = rt
            .execute_query("SELECT status, payload FROM capabilities WHERE id = 'A'")
            .unwrap();
        assert_eq!(selected.result.records.len(), 1);
        assert_eq!(
            selected.result.records[0].get("status"),
            Some(&Value::text("ready"))
        );
        if autocommit.is_ok() {
            assert_eq!(
                selected.result.records[0].get("payload"),
                Some(&Value::text("new"))
            );
        }
    }

    #[test]
    fn autocommit_update_skips_row_claimed_by_another_transaction() {
        use crate::runtime::mvcc::{clear_current_connection_id, set_current_connection_id};

        let rt = RedDBRuntime::with_options(RedDBOptions::in_memory()).unwrap();
        rt.execute_query("CREATE TABLE capabilities (id INT, rank INT, status TEXT, note TEXT)")
            .unwrap();
        rt.execute_query("CREATE INDEX capabilities_rank ON capabilities (rank)")
            .unwrap();
        rt.execute_query(
            "INSERT INTO capabilities (id, rank, status, note) VALUES (1, 10, 'ready', 'old')",
        )
        .unwrap();
        set_current_connection_id(41);
        rt.execute_query("BEGIN").unwrap();
        let claim = rt.execute_query("UPDATE capabilities SET status = 'claimed' WHERE status = 'ready' CLAIM LIMIT 1 ORDER BY rank ASC").unwrap();
        assert_eq!(claim.affected_rows, 1);
        set_current_connection_id(42);
        let other = rt
            .execute_query("UPDATE capabilities SET note = 'new' WHERE id = 1")
            .unwrap();
        assert_eq!(other.affected_rows, 0);
        set_current_connection_id(41);
        rt.execute_query("ROLLBACK").unwrap();
        clear_current_connection_id();
        let rows = rt
            .execute_query("SELECT status, note FROM capabilities WHERE id = 1")
            .unwrap();
        assert_eq!(rows.result.records.len(), 1);
        assert_eq!(
            rows.result.records[0].get("status"),
            Some(&Value::text("ready"))
        );
        assert_eq!(
            rows.result.records[0].get("note"),
            Some(&Value::text("old"))
        );
    }

    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)
        );
    }
}