reddb_server/application/

ports_impls_entity.rs

use std::collections::HashMap;

use crate::application::collection_contract_enforcer::{
    normalize_row_update_assignment_with_plan, CollectionContractWriteEnforcer,
};
use crate::application::entity::{
    AppliedEntityMutation, CreateDocumentInput, CreateKvInput, CreateTimeSeriesPointInput,
    RowUpdateColumnRule, RowUpdateContractPlan,
};
use crate::application::ttl_payload::{
    normalize_ttl_patch_operations, parse_top_level_ttl_metadata_entries,
};
use crate::json::{to_vec as json_to_vec, Value as JsonValue};
use crate::storage::unified::MetadataValue;
use reddb_types::Value;

use super::*;

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

fn refresh_context_index(
    db: &crate::storage::unified::devx::RedDB,
    collection: &str,
    id: crate::storage::EntityId,
) -> RedDBResult<()> {
    let store = db.store();
    let Some(entity) = store.get(collection, id) else {
        return Ok(());
    };

    store.context_index().index_entity(collection, &entity);
    Ok(())
}

fn ensure_insert_entity_rls(
    runtime: &RedDBRuntime,
    collection: &str,
    entity: &crate::storage::UnifiedEntity,
) -> RedDBResult<()> {
    if crate::runtime::rls_injection::entity_passes_rls_action(
        runtime,
        collection,
        reddb_rql::ast::PolicyAction::Insert,
        entity,
    ) {
        Ok(())
    } else {
        Err(crate::RedDBError::Query(format!(
            "insert into collection '{collection}' denied by row-level security"
        )))
    }
}

/// Pull `(name, value)` pairs for every indexable field on an entity.
/// Returns empty if field names are not recoverable here, so secondary-index
/// maintenance has nothing to act on. Used by the delete + update paths.
pub(crate) fn entity_row_fields_snapshot(
    entity: &crate::storage::UnifiedEntity,
) -> Vec<(String, Value)> {
    let mut fields = match &entity.data {
        crate::storage::EntityData::Row(row) => {
            if let Some(named) = &row.named {
                named.iter().map(|(k, v)| (k.clone(), v.clone())).collect()
            } else if let Some(schema) = &row.schema {
                schema
                    .iter()
                    .zip(row.columns.iter())
                    .map(|(name, value)| (name.clone(), value.clone()))
                    .collect()
            } else {
                Vec::new()
            }
        }
        crate::storage::EntityData::Node(node) => node
            .properties
            .iter()
            .map(|(k, v)| (k.clone(), v.clone()))
            .collect(),
        crate::storage::EntityData::Edge(edge) => edge
            .properties
            .iter()
            .map(|(k, v)| (k.clone(), v.clone()))
            .collect(),
        _ => Vec::new(),
    };
    expand_binary_document_body_snapshot(&mut fields);
    fields
}

fn expand_binary_document_body_snapshot(fields: &mut Vec<(String, Value)>) {
    let Some(body_idx) = fields.iter().position(|(name, _)| name == "body") else {
        return;
    };
    let Value::Json(bytes) = &fields[body_idx].1 else {
        return;
    };
    let body_fields = crate::document_body::body_fields(bytes);
    if let Some(body_json) = crate::document_body::decode_container_to_json(bytes) {
        if let Ok(json_bytes) = crate::json::to_vec(&body_json) {
            fields[body_idx].1 = Value::Json(json_bytes);
        }
    }
    let Some(body_fields) = body_fields else {
        return;
    };
    for (name, value) in body_fields {
        if fields.iter().all(|(existing, _)| existing != &name) {
            fields.push((name, value));
        }
    }
}

fn set_row_field(row: &mut crate::storage::unified::entity::RowData, name: &str, value: Value) {
    if let Some(named) = row.named.as_mut() {
        named.insert(name.to_string(), value);
        return;
    }

    if let Some(schema) = row.schema.as_ref() {
        if let Some(index) = schema.iter().position(|column| column == name) {
            if let Some(slot) = row.columns.get_mut(index) {
                *slot = value;
                return;
            }
        }

        let mut named = HashMap::with_capacity(schema.len().saturating_add(1));
        for (column, current) in schema.iter().zip(row.columns.iter()) {
            named.insert(column.clone(), current.clone());
        }
        named.insert(name.to_string(), value);
        row.named = Some(named);
        return;
    }

    let mut named = HashMap::with_capacity(1);
    named.insert(name.to_string(), value);
    row.named = Some(named);
}

fn collect_row_fields(row: &crate::storage::unified::entity::RowData) -> Vec<(String, Value)> {
    if let Some(named) = row.named.as_ref() {
        named
            .iter()
            .map(|(key, value)| (key.clone(), value.clone()))
            .collect()
    } else if let Some(schema) = row.schema.as_ref() {
        schema
            .iter()
            .cloned()
            .zip(row.columns.iter().cloned())
            .collect()
    } else {
        Vec::new()
    }
}

fn apply_row_field_assignments_raw<I>(
    row: &mut crate::storage::unified::entity::RowData,
    field_assignments: I,
) where
    I: IntoIterator<Item = (String, Value)>,
{
    for (column, value) in field_assignments {
        set_row_field(row, &column, value);
    }
}

fn apply_row_field_assignments_incremental<I>(
    collection: &str,
    row: &mut crate::storage::unified::entity::RowData,
    field_assignments: I,
    row_contract_plan: Option<&RowUpdateContractPlan>,
) -> RedDBResult<()>
where
    I: IntoIterator<Item = (String, Value)>,
{
    for (column, value) in field_assignments {
        let value = normalize_row_update_assignment_with_plan(
            collection,
            &column,
            value,
            row_contract_plan,
        )?;

        set_row_field(row, &column, value);
    }

    Ok(())
}

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

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

fn reject_document_array_position_path(path: &[String]) -> RedDBResult<()> {
    if path.iter().any(|segment| segment.parse::<usize>().is_ok()) {
        return Err(crate::RedDBError::Query(
            "array positional document patch paths are unsupported; replace the array or full document body instead"
                .to_string(),
        ));
    }
    Ok(())
}

/// Reject a document body `set` whose path would traverse *through* an existing
/// non-object value (e.g. `SET a.b = …` where `a` is already a scalar). This
/// mirrors the HTTP JSON-patch contract: intermediate objects are created only
/// where a slot is absent, never by silently clobbering a scalar. `body` is the
/// pre-merge document body.
fn reject_document_patch_path_through_scalar(body: &JsonValue, path: &[String]) -> RedDBResult<()> {
    if path.len() < 2 {
        return Ok(());
    }
    let scalar_error = |prefix: &str| {
        crate::RedDBError::Query(format!(
            "cannot set nested path '{}': '{}' is not an object; setting a path through a non-object value is unsupported — replace that value or the full document body instead",
            path.join("."),
            prefix
        ))
    };
    let mut current = body;
    // Walk the parent path (every segment but the leaf). Each container we index
    // into must be an object; an absent slot is fine (the merge creates it).
    for (idx, segment) in path[..path.len() - 1].iter().enumerate() {
        let JsonValue::Object(map) = current else {
            return Err(scalar_error(&path[..idx].join(".")));
        };
        match map.get(segment) {
            Some(next) => current = next,
            None => return Ok(()),
        }
    }
    // `current` is now the leaf's parent container; it must be an object.
    if !matches!(current, JsonValue::Object(_)) {
        return Err(scalar_error(&path[..path.len() - 1].join(".")));
    }
    Ok(())
}

/// Apply document body `set`/`unset` operations, enforcing the shared document
/// patch contract first: array-positional paths and paths that traverse through
/// a scalar are rejected with clear errors before any mutation touches `body`.
/// Used by both the SQL nested-`SET` executor and the HTTP JSON-patch operation
/// path so the two surfaces stay consistent (ADR 0067).
fn apply_document_body_patch_operations(
    body: &mut JsonValue,
    operations: &[PatchEntityOperation],
) -> RedDBResult<()> {
    for op in operations {
        reject_document_array_position_path(&op.path)?;
        if matches!(op.op, PatchEntityOperationType::Set) {
            reject_document_patch_path_through_scalar(body, &op.path)?;
        }
    }
    apply_patch_operations_to_json(body, operations).map_err(crate::RedDBError::Query)
}

fn document_body_from_row(
    row: &crate::storage::unified::entity::RowData,
) -> RedDBResult<JsonValue> {
    match row.get_field("body") {
        Some(Value::Json(bytes)) => {
            // The stored body may be the native binary container (PRD-1398) or
            // plain JSON; decode either form back to JSON.
            if let Some(body) = crate::document_body::decode_container_to_json(bytes) {
                Ok(body)
            } else {
                crate::json::from_slice(bytes).map_err(|err| {
                    crate::RedDBError::Query(format!("failed to decode document body: {err}"))
                })
            }
        }
        Some(_) => Err(crate::RedDBError::Query(
            "document body field must contain JSON".to_string(),
        )),
        None => Ok(JsonValue::Object(Default::default())),
    }
}

fn document_body_from_assignment(value: &Value) -> RedDBResult<JsonValue> {
    match value {
        Value::Json(bytes) | Value::Blob(bytes) => crate::json::from_slice(bytes)
            .map_err(|err| crate::RedDBError::Query(format!("invalid JSON body: {err}"))),
        Value::Text(text) => crate::json::from_str(text.as_ref())
            .map_err(|err| crate::RedDBError::Query(format!("invalid JSON body: {err}"))),
        Value::Integer(value) => crate::json::from_str(&value.to_string())
            .map_err(|err| crate::RedDBError::Query(format!("invalid JSON body: {err}"))),
        Value::UnsignedInteger(value) => crate::json::from_str(&value.to_string())
            .map_err(|err| crate::RedDBError::Query(format!("invalid JSON body: {err}"))),
        Value::Float(value) => crate::json::from_str(&value.to_string())
            .map_err(|err| crate::RedDBError::Query(format!("invalid JSON body: {err}"))),
        other => Err(crate::RedDBError::Query(format!(
            "column 'body' expected JSON body, got {other:?}"
        ))),
    }
}

fn document_body_set_operation(column: &str, value: Value) -> PatchEntityOperation {
    PatchEntityOperation {
        op: PatchEntityOperationType::Set,
        path: column.split('.').map(str::to_string).collect(),
        value: Some(value.to_json()),
    }
}

fn replace_document_fields_body(
    fields: &mut HashMap<String, Value>,
    body: JsonValue,
    binary: bool,
    modified_columns: &mut Vec<String>,
) -> RedDBResult<()> {
    modified_columns.push("body".to_string());

    let old_keys: Vec<String> = fields
        .keys()
        .filter(|key| key.as_str() != "body")
        .cloned()
        .collect();
    for key in old_keys {
        fields.remove(&key);
        modified_columns.push(key);
    }

    let body_bytes = crate::document_body::serialize_document_body(&body, binary)?;
    fields.insert("body".to_string(), Value::Json(body_bytes));

    Ok(())
}

fn ensure_no_reserved_document_body_fields(body: &JsonValue, collection: &str) -> RedDBResult<()> {
    if let JsonValue::Object(map) = body {
        crate::reserved_fields::ensure_no_reserved_public_item_fields(
            map.keys().map(String::as_str),
            &format!("document '{collection}'"),
        )?;
    }
    Ok(())
}

fn replace_document_row_body(
    row: &mut crate::storage::unified::entity::RowData,
    body: JsonValue,
    collection: &str,
    binary: bool,
    modified_columns: &mut Vec<String>,
) -> RedDBResult<()> {
    ensure_no_reserved_document_body_fields(&body, collection)?;
    let fields = row.named.get_or_insert_with(Default::default);
    replace_document_fields_body(fields, body, binary, modified_columns)?;

    // Full body replacement rebuilds the document from the canonical `body`.
    // Stale positional/schema values would otherwise leak back through JSON
    // presentation even after the named map was pruned.
    row.columns.clear();
    row.schema = None;
    Ok(())
}

fn document_merge_patch_payload(payload: &JsonValue) -> &JsonValue {
    if let JsonValue::Object(object) = payload {
        if object.len() == 1 {
            if let Some(body) = object.get("body") {
                return body;
            }
        }
    }
    payload
}

fn apply_document_json_merge_patch(body: &mut JsonValue, patch: &JsonValue) -> RedDBResult<()> {
    if !matches!(patch, JsonValue::Object(_)) {
        *body = patch.clone();
        return Ok(());
    }

    if matches!(patch, JsonValue::Object(object) if object.is_empty()) {
        if !matches!(body, JsonValue::Object(_)) {
            *body = JsonValue::Object(Default::default());
        }
        return Ok(());
    }

    let mut operations = Vec::new();
    collect_document_json_merge_patch_operations(body, Vec::new(), patch, &mut operations);
    apply_patch_operations_to_json(body, &operations).map_err(crate::RedDBError::Query)
}

fn collect_document_json_merge_patch_operations(
    target: &JsonValue,
    path: Vec<String>,
    patch: &JsonValue,
    operations: &mut Vec<PatchEntityOperation>,
) {
    let JsonValue::Object(patch_object) = patch else {
        operations.push(PatchEntityOperation {
            op: PatchEntityOperationType::Set,
            path,
            value: Some(patch.clone()),
        });
        return;
    };

    for (key, value) in patch_object {
        let mut child_path = path.clone();
        child_path.push(key.clone());
        match value {
            JsonValue::Null => operations.push(PatchEntityOperation {
                op: PatchEntityOperationType::Unset,
                path: child_path,
                value: None,
            }),
            JsonValue::Object(object) if object.is_empty() => {
                let child_target = match target {
                    JsonValue::Object(target_object) => target_object.get(key),
                    _ => None,
                };
                if !matches!(child_target, Some(JsonValue::Object(_))) {
                    operations.push(PatchEntityOperation {
                        op: PatchEntityOperationType::Set,
                        path: child_path,
                        value: Some(JsonValue::Object(Default::default())),
                    });
                }
            }
            JsonValue::Object(_) => {
                let child_target = match target {
                    JsonValue::Object(target_object) => target_object.get(key),
                    _ => None,
                }
                .unwrap_or(&JsonValue::Null);
                collect_document_json_merge_patch_operations(
                    child_target,
                    child_path,
                    value,
                    operations,
                );
            }
            _ => operations.push(PatchEntityOperation {
                op: PatchEntityOperationType::Set,
                path: child_path,
                value: Some(value.clone()),
            }),
        }
    }
}

impl RedDBRuntime {
    pub(crate) fn apply_loaded_patch_entity_core(
        &self,
        collection: String,
        mut entity: crate::storage::UnifiedEntity,
        payload: JsonValue,
        operations: Vec<PatchEntityOperation>,
    ) -> RedDBResult<AppliedEntityMutation> {
        let id = entity.id;
        let previous_xmax = entity.xmax;
        let operations = normalize_ttl_patch_operations(operations)?;
        // Snapshot pre-patch row fields for the secondary-index hook —
        // empty for non-row entities, which is the desired no-op.
        let pre_mutation_fields = entity_row_fields_snapshot(&entity);

        // Versioned collections retain MVCC history: a PATCH/UPDATE must
        // produce a NEW full version (the merged document/row/node) with
        // the prior version tombstoned, mirroring the SQL UPDATE row
        // path. Gated STRICTLY on the collection `versioned` flag —
        // non-versioned collections keep last-writer-wins in-place
        // mutation. Row entities (documents / table rows / KV) carry
        // logical-id MVCC history (Phase 1/2); graph nodes & edges join
        // in Phase 3. The post-image re-stamp (new physical id, same
        // logical_id, fresh xmin) at the tail of this function is fully
        // kind-agnostic, so the same machinery installs a versioned
        // graph-node update.
        let patch_versions = matches!(
            entity.data,
            crate::storage::EntityData::Row(_)
                | crate::storage::EntityData::Node(_)
                | crate::storage::EntityData::Edge(_)
        ) && self.vcs_is_versioned(&collection).unwrap_or(false);
        let versioned_update_xid = if patch_versions {
            match self.current_xid() {
                Some(xid) => Some(xid),
                None => {
                    let snapshot_manager = self.snapshot_manager();
                    let xid = snapshot_manager.begin();
                    snapshot_manager.commit(xid);
                    Some(xid)
                }
            }
        } else {
            None
        };
        let mut replaced_entity = versioned_update_xid.map(|xid| {
            let mut old = entity.clone();
            old.set_xmax(xid);
            old
        });

        let db = self.db();
        let store = db.store();
        if store.get_collection(&collection).is_none() {
            return Err(crate::RedDBError::NotFound(format!(
                "collection not found: {collection}"
            )));
        }

        let mut patch_metadata: Option<crate::storage::unified::Metadata> = None;
        let mut metadata_changed = false;
        let mut modified_columns: Vec<String> = Vec::new();
        let mut context_index_dirty = false;
        let mut graph_node_type: Option<String> = None;
        let mut graph_edge_weight: Option<f32> = None;

        let row_contract_timestamps = db
            .collection_contract(&collection)
            .map(|c| c.timestamps_enabled)
            .unwrap_or(false);

        match &mut entity.data {
            crate::storage::EntityData::Row(row) => {
                // PATCH mutates a named map. A recovered row may instead own
                // its values under a shared schema; move those values into
                // the map before applying only the requested field changes.
                if row.named.is_none() {
                    if let Some(schema) = row.schema.as_ref() {
                        let fields = if schema.len() == row.columns.len() {
                            schema
                                .iter()
                                .cloned()
                                .zip(std::mem::take(&mut row.columns))
                                .collect()
                        } else {
                            collect_row_fields(row).into_iter().collect()
                        };
                        row.named = Some(fields);
                    }
                }
                let is_document_collection = db
                    .collection_contract(&collection)
                    .map(|contract| {
                        contract.declared_model == crate::catalog::CollectionModel::Document
                    })
                    .unwrap_or(false);
                let binary_body = self.binary_document_body_enabled();
                let mut field_ops = Vec::new();
                let mut metadata_ops = Vec::new();
                let mut document_body_ops = Vec::new();
                let mut document_body_replace: Option<JsonValue> = None;
                let has_patch_operations = !operations.is_empty();

                for mut op in operations {
                    let Some(root) = op.path.first().map(String::as_str) else {
                        return Err(crate::RedDBError::Query(
                            "patch path cannot be empty".to_string(),
                        ));
                    };

                    match root {
                        "body" if is_document_collection => {
                            if op.path.len() == 1 {
                                match op.op {
                                    PatchEntityOperationType::Set
                                    | PatchEntityOperationType::Replace => {
                                        let Some(value) = op.value.take() else {
                                            return Err(crate::RedDBError::Query(
                                                "document body replacement requires a value"
                                                    .to_string(),
                                            ));
                                        };
                                        document_body_replace = Some(value);
                                    }
                                    PatchEntityOperationType::Unset => {
                                        return Err(crate::RedDBError::Query(
                                            "document body cannot be unset; replace it instead"
                                                .to_string(),
                                        ));
                                    }
                                }
                                continue;
                            }
                            if op.path.len() < 2 {
                                return Err(crate::RedDBError::Query(
                                    "document body patch paths require a nested key; use payload.body for full replacement"
                                        .to_string(),
                                ));
                            }
                            op.path.remove(0);
                            reject_document_array_position_path(&op.path)?;
                            document_body_ops.push(op);
                        }
                        "fields" | "named" => {
                            if op.path.len() < 2 {
                                return Err(crate::RedDBError::Query(
                                    "patch path 'fields' requires a nested key".to_string(),
                                ));
                            }
                            if row_contract_timestamps {
                                let leaf = op.path.get(1).map(String::as_str);
                                if matches!(leaf, Some("created_at") | Some("updated_at")) {
                                    return Err(crate::RedDBError::Query(format!(
                                        "collection '{}' manages '{}' automatically — do not set it in UPDATE",
                                        collection,
                                        leaf.unwrap_or("")
                                    )));
                                }
                            }
                            op.path.remove(0);
                            field_ops.push(op);
                        }
                        "metadata" => {
                            if op.path.len() < 2 {
                                return Err(crate::RedDBError::Query(
                                    "patch path 'metadata' requires a nested key".to_string(),
                                ));
                            }
                            op.path.remove(0);
                            metadata_ops.push(op);
                        }
                        _ => {
                            return Err(crate::RedDBError::Query(format!(
                                "unsupported patch target '{root}' for table rows. Use fields/*, metadata/*, or weight"
                            )));
                        }
                    }
                }

                if document_body_replace.is_some() || !document_body_ops.is_empty() {
                    context_index_dirty = true;
                    let mut body = match document_body_replace {
                        Some(body) => body,
                        None => document_body_from_row(row)?,
                    };
                    apply_document_body_patch_operations(&mut body, &document_body_ops)?;
                    replace_document_row_body(
                        row,
                        body,
                        &collection,
                        binary_body,
                        &mut modified_columns,
                    )?;
                }

                if is_document_collection && !has_patch_operations {
                    let mut body = document_body_from_row(row)?;
                    let previous_body = body.clone();
                    apply_document_json_merge_patch(
                        &mut body,
                        document_merge_patch_payload(&payload),
                    )?;
                    if body != previous_body {
                        context_index_dirty = true;
                        replace_document_row_body(
                            row,
                            body,
                            &collection,
                            binary_body,
                            &mut modified_columns,
                        )?;
                    }
                }

                if !field_ops.is_empty() {
                    context_index_dirty = true;
                    if is_document_collection {
                        // Document fields are projections of the canonical
                        // `body` JSON. Apply the assignment to the body so the
                        // stored document remains single-source.
                        let mut body = document_body_from_row(row)?;
                        apply_patch_operations_to_json(&mut body, &field_ops)
                            .map_err(crate::RedDBError::Query)?;
                        replace_document_row_body(
                            row,
                            body,
                            &collection,
                            binary_body,
                            &mut modified_columns,
                        )?;
                    } else {
                        let named = row.named.get_or_insert_with(Default::default);
                        for op in &field_ops {
                            if let Some(col) = op.path.first() {
                                modified_columns.push(col.clone());
                            }
                        }
                        apply_patch_operations_to_storage_map(named, &field_ops)?;
                    }
                }

                if let Some(fields) = payload
                    .get("fields")
                    .and_then(crate::json::Value::as_object)
                {
                    if row_contract_timestamps {
                        for key in fields.keys() {
                            if key == "created_at" || key == "updated_at" {
                                return Err(crate::RedDBError::Query(format!(
                                    "collection '{}' manages '{}' automatically — do not set it in UPDATE",
                                    collection, key
                                )));
                            }
                        }
                    }
                    context_index_dirty = true;
                    if is_document_collection {
                        // Mirror the `field_ops` path above: keep the canonical
                        // `body` JSON as the single stored representation.
                        let mut body = document_body_from_row(row)?;
                        if let JsonValue::Object(map) = &mut body {
                            for (key, value) in fields {
                                map.insert(key.clone(), value.clone());
                            }
                        }
                        replace_document_row_body(
                            row,
                            body,
                            &collection,
                            binary_body,
                            &mut modified_columns,
                        )?;
                    } else {
                        let named = row.named.get_or_insert_with(Default::default);
                        for (key, value) in fields {
                            modified_columns.push(key.clone());
                            named.insert(key.clone(), json_to_storage_value(value)?);
                        }
                    }
                }

                if !metadata_ops.is_empty() {
                    ensure_non_tree_reserved_metadata_patch_paths(&metadata_ops)?;
                    let metadata = patch_metadata.get_or_insert_with(|| {
                        store.get_metadata(&collection, id).unwrap_or_default()
                    });
                    let mut metadata_json = metadata_to_json(metadata);
                    apply_patch_operations_to_json(&mut metadata_json, &metadata_ops)
                        .map_err(crate::RedDBError::Query)?;
                    *metadata = metadata_from_json(&metadata_json)?;
                    metadata_changed = true;
                }

                if !modified_columns.is_empty() || row_contract_timestamps {
                    let contract = CollectionContractWriteEnforcer::new(self, &db, &collection);
                    let current_fields = if let Some(named) = row.named.take() {
                        named.into_iter().collect::<Vec<_>>()
                    } else if let Some(schema) = row.schema.as_ref() {
                        schema
                            .iter()
                            .cloned()
                            .zip(row.columns.iter().cloned())
                            .collect::<Vec<_>>()
                    } else {
                        Vec::new()
                    };
                    let normalized_fields = contract.normalize_update_fields(current_fields)?;
                    if let Some(schema) = db.collection_contract_arc(&collection) {
                        for column in schema
                            .declared_columns
                            .iter()
                            .filter(|column| column.generated.is_some())
                        {
                            modified_columns.push(column.name.clone());
                            context_index_dirty = true;
                        }
                    }
                    if row_contract_timestamps {
                        modified_columns.push("updated_at".to_string());
                        context_index_dirty = true;
                    }
                    if contract.requires_uniqueness_check(&modified_columns) {
                        contract.enforce_row_uniqueness(&normalized_fields, Some(id))?;
                    }
                    if self
                        .index_store_ref()
                        .touches_unique_index(&collection, &modified_columns)
                    {
                        self.enforce_unique_updates(
                            &collection,
                            &[(id, normalized_fields.clone())],
                        )?;
                    }
                    row.named = Some(normalized_fields.into_iter().collect());
                }
            }
            crate::storage::EntityData::Node(node) => {
                let mut field_ops = Vec::new();
                let mut metadata_ops = Vec::new();
                let mut node_type_ops = Vec::new();

                for mut op in operations {
                    let Some(root) = op.path.first().map(String::as_str) else {
                        return Err(crate::RedDBError::Query(
                            "patch path cannot be empty".to_string(),
                        ));
                    };

                    match root {
                        "fields" | "properties" => {
                            if op.path.len() < 2 {
                                return Err(crate::RedDBError::Query(
                                    "patch path 'fields' requires a nested key".to_string(),
                                ));
                            }
                            op.path.remove(0);
                            field_ops.push(op);
                        }
                        "metadata" => {
                            if op.path.len() < 2 {
                                return Err(crate::RedDBError::Query(
                                    "patch path 'metadata' requires a nested key".to_string(),
                                ));
                            }
                            op.path.remove(0);
                            metadata_ops.push(op);
                        }
                        "node_type" => {
                            if op.path.len() != 1 {
                                return Err(crate::RedDBError::Query(
                                    "patch path 'node_type' does not allow nested keys".to_string(),
                                ));
                            }
                            op.path.clear();
                            node_type_ops.push(op);
                        }
                        _ => {
                            return Err(crate::RedDBError::Query(format!(
                                "unsupported patch target '{root}' for graph nodes. Use fields/*, properties/*, node_type, or metadata/*"
                            )));
                        }
                    }
                }

                for op in node_type_ops {
                    context_index_dirty = true;
                    let value = op.value.ok_or_else(|| {
                        crate::RedDBError::Query("node_type operations require a value".to_string())
                    })?;

                    match op.op {
                        PatchEntityOperationType::Unset => {
                            return Err(crate::RedDBError::Query(
                                "node_type cannot be unset through patch operations".to_string(),
                            ));
                        }
                        PatchEntityOperationType::Set | PatchEntityOperationType::Replace => {
                            let Some(node_type) = value.as_str() else {
                                return Err(crate::RedDBError::Query(
                                    "node_type operation requires a text value".to_string(),
                                ));
                            };
                            graph_node_type = Some(node_type.to_string());
                            modified_columns.push("node_type".to_string());
                        }
                    }
                }

                if !field_ops.is_empty() {
                    context_index_dirty = true;
                    apply_patch_operations_to_storage_map(&mut node.properties, &field_ops)?;
                }

                if let Some(fields) = payload
                    .get("fields")
                    .and_then(crate::json::Value::as_object)
                {
                    context_index_dirty = true;
                    for (key, value) in fields {
                        node.properties
                            .insert(key.clone(), json_to_storage_value(value)?);
                    }
                }

                if !metadata_ops.is_empty() {
                    ensure_non_tree_reserved_metadata_patch_paths(&metadata_ops)?;
                    let metadata = patch_metadata.get_or_insert_with(|| {
                        store.get_metadata(&collection, id).unwrap_or_default()
                    });
                    let mut metadata_json = metadata_to_json(metadata);
                    apply_patch_operations_to_json(&mut metadata_json, &metadata_ops)
                        .map_err(crate::RedDBError::Query)?;
                    *metadata = metadata_from_json(&metadata_json)?;
                    metadata_changed = true;
                }
            }
            crate::storage::EntityData::Edge(edge) => {
                let mut field_ops = Vec::new();
                let mut metadata_ops = Vec::new();
                let mut weight_ops = Vec::new();

                for mut op in operations {
                    let Some(root) = op.path.first().map(String::as_str) else {
                        return Err(crate::RedDBError::Query(
                            "patch path cannot be empty".to_string(),
                        ));
                    };

                    match root {
                        "fields" | "properties" => {
                            if op.path.len() < 2 {
                                return Err(crate::RedDBError::Query(
                                    "patch path 'fields' requires a nested key".to_string(),
                                ));
                            }
                            op.path.remove(0);
                            field_ops.push(op);
                        }
                        "weight" => {
                            if op.path.len() != 1 {
                                return Err(crate::RedDBError::Query(
                                    "patch path 'weight' does not allow nested keys".to_string(),
                                ));
                            }
                            op.path.clear();
                            weight_ops.push(op);
                        }
                        "metadata" => {
                            if op.path.len() < 2 {
                                return Err(crate::RedDBError::Query(
                                    "patch path 'metadata' requires a nested key".to_string(),
                                ));
                            }
                            op.path.remove(0);
                            metadata_ops.push(op);
                        }
                        _ => {
                            return Err(crate::RedDBError::Query(format!(
                                "unsupported patch target '{root}' for graph edges. Use fields/*, weight, metadata/*"
                            )));
                        }
                    }
                }

                if !field_ops.is_empty() {
                    context_index_dirty = true;
                    apply_patch_operations_to_storage_map(&mut edge.properties, &field_ops)?;
                }

                for op in weight_ops {
                    context_index_dirty = true;
                    let value = op.value.ok_or_else(|| {
                        crate::RedDBError::Query("weight operations require a value".to_string())
                    })?;

                    match op.op {
                        PatchEntityOperationType::Unset => {
                            return Err(crate::RedDBError::Query(
                                "weight cannot be unset through patch operations".to_string(),
                            ));
                        }
                        PatchEntityOperationType::Set | PatchEntityOperationType::Replace => {
                            let Some(weight) = value.as_f64() else {
                                return Err(crate::RedDBError::Query(
                                    "weight operation requires a numeric value".to_string(),
                                ));
                            };
                            graph_edge_weight = Some(weight as f32);
                            edge.weight = weight as f32;
                            modified_columns.push("weight".to_string());
                        }
                    }
                }

                if let Some(fields) = payload
                    .get("fields")
                    .and_then(crate::json::Value::as_object)
                {
                    context_index_dirty = true;
                    for (key, value) in fields {
                        edge.properties
                            .insert(key.clone(), json_to_storage_value(value)?);
                    }
                }

                if !metadata_ops.is_empty() {
                    ensure_non_tree_reserved_metadata_patch_paths(&metadata_ops)?;
                    let metadata = patch_metadata.get_or_insert_with(|| {
                        store.get_metadata(&collection, id).unwrap_or_default()
                    });
                    let mut metadata_json = metadata_to_json(metadata);
                    apply_patch_operations_to_json(&mut metadata_json, &metadata_ops)
                        .map_err(crate::RedDBError::Query)?;
                    *metadata = metadata_from_json(&metadata_json)?;
                    metadata_changed = true;
                }
            }
            crate::storage::EntityData::Vector(vector) => {
                let mut field_ops = Vec::new();
                let mut metadata_ops = Vec::new();

                for mut op in operations {
                    let Some(root) = op.path.first().map(String::as_str) else {
                        return Err(crate::RedDBError::Query(
                            "patch path cannot be empty".to_string(),
                        ));
                    };

                    match root {
                        "fields" => {
                            if op.path.len() < 2 {
                                return Err(crate::RedDBError::Query(
                                    "patch path 'fields' requires a nested key".to_string(),
                                ));
                            }
                            op.path.remove(0);
                            let Some(target) = op.path.first().map(String::as_str) else {
                                return Err(crate::RedDBError::Query(
                                    "patch path requires a target under fields".to_string(),
                                ));
                            };
                            if !matches!(target, "dense" | "content" | "sparse") {
                                return Err(crate::RedDBError::Query(format!(
                                    "unsupported vector patch target '{target}'"
                                )));
                            }
                            field_ops.push(op);
                        }
                        "metadata" => {
                            if op.path.len() < 2 {
                                return Err(crate::RedDBError::Query(
                                    "patch path 'metadata' requires a nested key".to_string(),
                                ));
                            }
                            op.path.remove(0);
                            metadata_ops.push(op);
                        }
                        _ => {
                            return Err(crate::RedDBError::Query(format!(
                                "unsupported patch target '{root}' for vectors. Use fields/* or metadata/*"
                            )));
                        }
                    }
                }

                if !field_ops.is_empty() {
                    context_index_dirty = true;
                    apply_patch_operations_to_vector_fields(vector, &field_ops)?;
                }

                if let Some(fields) = payload
                    .get("fields")
                    .and_then(crate::json::Value::as_object)
                {
                    context_index_dirty = true;
                    if let Some(content) =
                        fields.get("content").and_then(crate::json::Value::as_str)
                    {
                        vector.content = Some(content.to_string());
                    }
                    if let Some(dense) = fields.get("dense") {
                        vector.dense = dense
                            .as_array()
                            .ok_or_else(|| {
                                crate::RedDBError::Query(
                                    "field 'dense' must be an array".to_string(),
                                )
                            })?
                            .iter()
                            .map(|value| {
                                value.as_f64().map(|value| value as f32).ok_or_else(|| {
                                    crate::RedDBError::Query(
                                        "field 'dense' must contain only numbers".to_string(),
                                    )
                                })
                            })
                            .collect::<Result<Vec<_>, _>>()?;
                    }
                }

                if !metadata_ops.is_empty() {
                    ensure_non_tree_reserved_metadata_patch_paths(&metadata_ops)?;
                    let metadata = patch_metadata.get_or_insert_with(|| {
                        store.get_metadata(&collection, id).unwrap_or_default()
                    });
                    let mut metadata_json = metadata_to_json(metadata);
                    apply_patch_operations_to_json(&mut metadata_json, &metadata_ops)
                        .map_err(crate::RedDBError::Query)?;
                    *metadata = metadata_from_json(&metadata_json)?;
                    metadata_changed = true;
                }
            }
            crate::storage::EntityData::TimeSeries(_)
            | crate::storage::EntityData::QueueMessage(_) => {
                return Err(crate::RedDBError::Query(
                    "patch operations are not supported for TimeSeries or QueueMessage entities"
                        .to_string(),
                ));
            }
        }

        if let Some(node_type) = graph_node_type {
            if let crate::storage::EntityKind::GraphNode(node) = &mut entity.kind {
                node.node_type = node_type;
            }
        }
        if let Some(weight) = graph_edge_weight {
            if let crate::storage::EntityKind::GraphEdge(edge) = &mut entity.kind {
                edge.weight = (weight * 1000.0) as u32;
            }
        }

        if let Some(metadata) = payload
            .get("metadata")
            .and_then(crate::json::Value::as_object)
        {
            let patch_metadata = patch_metadata
                .get_or_insert_with(|| store.get_metadata(&collection, id).unwrap_or_default());
            for (key, value) in metadata {
                ensure_non_tree_reserved_metadata_key(key)?;
                patch_metadata.set(key.clone(), json_to_metadata_value(value)?);
            }
            metadata_changed = true;
        }

        for (key, value) in parse_top_level_ttl_metadata_entries(&payload)? {
            let patch_metadata = patch_metadata
                .get_or_insert_with(|| store.get_metadata(&collection, id).unwrap_or_default());
            if matches!(value, crate::storage::unified::MetadataValue::Null) {
                patch_metadata.remove(&key);
            } else {
                patch_metadata.set(key, value);
            }
            metadata_changed = true;
        }

        entity.field_bloom =
            crate::storage::unified::entity::compute_entity_field_bloom(&entity.data);
        entity.updated_at = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_secs();

        // Versioned PATCH: re-stamp the merged post-image as a fresh
        // physical version keyed on the same logical_id, and close the
        // prior version's xmax at the same xid. `install_versioned_*`
        // in persist installs the new row and tombstones the old one;
        // `record_pending_versioned_update` arms first-committer-wins.
        if let Some(xid) = versioned_update_xid {
            let logical_id = entity.logical_id();
            entity.id = store.next_entity_id();
            entity.set_logical_id(logical_id);
            entity.set_xmin(xid);
            entity.set_xmax(0);
            if let Some(old) = replaced_entity.as_mut() {
                old.set_xmax(xid);
            }
        }

        modified_columns = dedupe_modified_columns(modified_columns);

        Ok(AppliedEntityMutation {
            id: entity.id,
            collection,
            entity,
            metadata: patch_metadata,
            modified_columns,
            persist_metadata: metadata_changed,
            context_index_dirty,
            replaced_entity,
            replaced_entity_previous_xmax: previous_xmax,
            pre_mutation_fields,
        })
    }

    pub(crate) fn apply_loaded_sql_update_row_core(
        &self,
        collection: String,
        mut entity: crate::storage::UnifiedEntity,
        static_field_assignments: &[(String, Value)],
        dynamic_field_assignments: Vec<(String, Value)>,
        static_metadata_assignments: &[(String, MetadataValue)],
        dynamic_metadata_assignments: Vec<(String, MetadataValue)>,
        row_contract_plan: Option<&RowUpdateContractPlan>,
        row_modified_columns_template: &[String],
        row_touches_unique_columns: bool,
    ) -> RedDBResult<AppliedEntityMutation> {
        let id = entity.id;
        let previous_xmax = entity.xmax;
        let db = self.db();
        let store = db.store();
        let Some(_) = store.get_collection(&collection) else {
            return Err(crate::RedDBError::NotFound(format!(
                "collection not found: {collection}"
            )));
        };

        let versioned_update_xid = match self.current_xid() {
            Some(xid) => Some(xid),
            None => {
                let snapshot_manager = self.snapshot_manager();
                let xid = snapshot_manager.begin();
                snapshot_manager.commit(xid);
                Some(xid)
            }
        };
        let mut replaced_entity = versioned_update_xid.map(|xid| {
            let mut old = entity.clone();
            old.set_xmax(xid);
            old
        });

        let mut patch_metadata: Option<crate::storage::unified::Metadata> = None;
        let row_contract_timestamps = row_contract_plan
            .map(|plan| plan.timestamps_enabled)
            .unwrap_or(false);
        let mut metadata_changed = false;
        let mut modified_columns = row_modified_columns_template.to_vec();
        let mut context_index_dirty = !modified_columns.is_empty();

        // Snapshot OLD field values BEFORE applying the assignments —
        // the secondary-index maintenance hook needs both before/after to
        // delete-then-insert under changed indexed columns.
        let pre_mutation_fields = entity_row_fields_snapshot(&entity);

        let crate::storage::EntityData::Row(row) = &mut entity.data else {
            return Err(crate::RedDBError::Query(
                "SQL row update fast path requires a row entity".to_string(),
            ));
        };

        let _ = row_contract_plan;

        // Documents store the full JSON under the `body` column. Legacy rows
        // also promote top-level keys for filtering, while binary body mode
        // keeps `body` as the single source of truth. Reflect non-body SET
        // assignments into `body` as patch operations so dotted paths update
        // nested JSON instead of becoming literal top-level keys.
        let is_document_collection = db
            .collection_contract(&collection)
            .map(|contract| contract.declared_model == crate::catalog::CollectionModel::Document)
            .unwrap_or(false);
        let kv_key = if row.get_field("value").is_some() || row.columns.len() >= 2 {
            row.get_field("key")
                .cloned()
                .or_else(|| row.columns.first().cloned())
        } else {
            None
        };
        if is_document_collection {
            // #1366: when a SET assigns to `body`, treat it as a FULL REPLACE of
            // the stored document — drop other column assignments, the new body
            // is the single source of truth.
            let body_assignment = static_field_assignments
                .iter()
                .chain(dynamic_field_assignments.iter())
                .rev()
                .find(|(column, _)| column == "body")
                .cloned();

            match body_assignment {
                Some((_, value)) => {
                    let body = document_body_from_assignment(&value)?;
                    replace_document_row_body(
                        row,
                        body,
                        &collection,
                        self.binary_document_body_enabled(),
                        &mut modified_columns,
                    )?;
                    context_index_dirty = true;
                }
                None => {
                    // #1365: no `body` assignment — fall back to nested-path
                    // partial update via document_body_set_operation.
                    let mut static_row_assignments = Vec::new();
                    let mut dynamic_row_assignments = Vec::new();
                    let mut document_body_ops = Vec::new();
                    for (column, value) in static_field_assignments.iter().cloned() {
                        if column == "body" {
                            static_row_assignments.push((column, value));
                        } else {
                            document_body_ops.push(document_body_set_operation(&column, value));
                        }
                    }
                    for (column, value) in dynamic_field_assignments {
                        if column == "body" {
                            dynamic_row_assignments.push((column, value));
                        } else {
                            document_body_ops.push(document_body_set_operation(&column, value));
                        }
                    }
                    apply_row_field_assignments_raw(row, static_row_assignments);
                    apply_row_field_assignments_raw(row, dynamic_row_assignments);
                    if !document_body_ops.is_empty() {
                        let mut body = document_body_from_row(row)?;
                        apply_document_body_patch_operations(&mut body, &document_body_ops)?;
                        replace_document_row_body(
                            row,
                            body,
                            &collection,
                            self.binary_document_body_enabled(),
                            &mut modified_columns,
                        )?;
                        context_index_dirty = true;
                    }
                }
            }
        } else {
            // Non-document collection: every assignment is a regular row column.
            let mut all_assignments: Vec<(String, Value)> = static_field_assignments.to_vec();
            all_assignments.extend(dynamic_field_assignments);
            apply_row_field_assignments_raw(row, all_assignments);
            if let Some(key) = kv_key {
                set_row_field(row, "key", key);
            }
        }

        for (key, value) in static_metadata_assignments
            .iter()
            .cloned()
            .chain(dynamic_metadata_assignments)
        {
            ensure_non_tree_reserved_metadata_key(&key)?;
            patch_metadata
                .get_or_insert_with(|| store.get_metadata(&collection, id).unwrap_or_default())
                .set(key, value);
            metadata_changed = true;
        }

        if !modified_columns.is_empty() || row_contract_timestamps {
            let contract = CollectionContractWriteEnforcer::new(self, &db, &collection);
            if row_contract_timestamps {
                context_index_dirty = true;
                set_row_field(row, "updated_at", contract.managed_timestamp_value());
                modified_columns.push("updated_at".to_string());
            }
            let expression_contract = if row_contract_plan.is_some_and(|plan| plan.has_expressions)
            {
                db.collection_contract_arc(&collection)
            } else {
                None
            };
            if let Some(schema) = &expression_contract {
                let normalized = contract.normalize_update_fields(collect_row_fields(row))?;
                for column in schema
                    .declared_columns
                    .iter()
                    .filter(|column| column.generated.is_some())
                {
                    modified_columns.push(column.name.clone());
                }
                for (name, value) in normalized {
                    set_row_field(row, &name, value);
                }
                context_index_dirty = true;
            }
            if row_touches_unique_columns || expression_contract.is_some() {
                let current_fields = collect_row_fields(row);
                contract.enforce_row_uniqueness(&current_fields, Some(id))?;
            }
        }

        entity.field_bloom =
            crate::storage::unified::entity::compute_entity_field_bloom(&entity.data);
        entity.updated_at = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_secs();

        if let Some(xid) = versioned_update_xid {
            let logical_id = entity.logical_id();
            entity.id = store.next_entity_id();
            entity.set_logical_id(logical_id);
            entity.set_xmin(xid);
            entity.set_xmax(0);
            if let Some(old) = replaced_entity.as_mut() {
                old.set_xmax(xid);
            }
        }

        modified_columns = dedupe_modified_columns(modified_columns);

        Ok(AppliedEntityMutation {
            id: entity.id,
            collection,
            entity,
            metadata: patch_metadata,
            modified_columns,
            persist_metadata: metadata_changed,
            context_index_dirty,
            replaced_entity,
            replaced_entity_previous_xmax: previous_xmax,
            pre_mutation_fields,
        })
    }

    pub(crate) fn persist_applied_entity_mutations(
        &self,
        applied: &[AppliedEntityMutation],
    ) -> RedDBResult<()> {
        if applied.is_empty() {
            return Ok(());
        }

        let store = self.db().store();
        // Fence the in-memory change and its WAL append together. Otherwise a
        // checkpoint can copy a new transaction's rows or truncate a concurrent
        // autocommit's WAL before those rows belong to its frozen generation.
        let _publication = store.vault_publication_lock.lock();
        let collection = &applied[0].collection;
        let Some(manager) = store.get_collection(collection) else {
            return Err(crate::RedDBError::NotFound(format!(
                "collection not found: {collection}"
            )));
        };

        let mut ordinary = Vec::with_capacity(applied.len());
        for item in applied {
            if let Some(old_version) = item.replaced_entity.as_ref() {
                store
                    .install_versioned_table_row_update(
                        collection,
                        old_version.clone(),
                        item.entity.clone(),
                        if item.persist_metadata {
                            item.metadata.as_ref()
                        } else {
                            None
                        },
                    )
                    .map_err(|err| crate::RedDBError::Internal(err.to_string()))?;
                if self.current_xid().is_some() {
                    self.record_pending_versioned_update(
                        crate::runtime::impl_core::current_connection_id(),
                        collection,
                        old_version.id,
                        item.entity.id,
                        old_version.xmax,
                        item.replaced_entity_previous_xmax,
                    );
                }
            } else {
                ordinary.push(item);
            }
        }
        if ordinary.is_empty() {
            return Ok(());
        }

        manager
            .update_hot_batch_with_metadata(ordinary.iter().map(|item| {
                (
                    &item.entity,
                    item.modified_columns.as_slice(),
                    if item.persist_metadata {
                        item.metadata.as_ref()
                    } else {
                        None
                    },
                )
            }))
            .map_err(|err| crate::RedDBError::Query(err.to_string()))?;

        // PG-HOT-like fast path: segment in-place is done; when no
        // mutation touches a secondary-indexed column AND no metadata
        // payload needs to be folded into a B-tree record, skip the
        // in-line B-tree upsert. The WAL still records the update
        // (durability preserved; recovery replay rebuilds the B-tree),
        // and `manager.get()` prefers the live segment over the
        // B-tree for reads — so the short-circuit is invisible to
        // callers. See `persist_entities_to_pager_wal_only`.
        let indexed_cols = self
            .index_store_ref()
            .indexed_columns_set(collection.as_str());
        let all_hot = !indexed_cols.is_empty()
            && ordinary.iter().all(|item| {
                !item.persist_metadata
                    && !item
                        .modified_columns
                        .iter()
                        .any(|c| indexed_cols.contains(c))
            })
            || indexed_cols.is_empty() && ordinary.iter().all(|item| !item.persist_metadata);

        // Pass `&[&UnifiedEntity]` — no per-entity clone. The SQL UPDATE
        // inner loop hands us `applied` which already owns the post-image
        // entity; all we need for the persist path is a read borrow.
        let entity_refs: Vec<&crate::storage::UnifiedEntity> =
            ordinary.iter().map(|item| &item.entity).collect();
        let persist_fn = if all_hot {
            crate::storage::unified::UnifiedStore::persist_entity_refs_to_pager_wal_only
        } else {
            crate::storage::unified::UnifiedStore::persist_entity_refs_to_pager
        };
        persist_fn(store.as_ref(), collection, &entity_refs)
            .map_err(|err| crate::RedDBError::Internal(err.to_string()))
    }

    pub(crate) fn flush_applied_entity_mutation(
        &self,
        applied: &AppliedEntityMutation,
        topology_guard: parking_lot::RwLockReadGuard<'_, ()>,
    ) -> RedDBResult<()> {
        let store = self.db().store();
        if applied.context_index_dirty {
            store
                .context_index()
                .index_entity(&applied.collection, &applied.entity);
        }
        // Secondary-index maintenance for SQL UPDATE / JSON-Patch flows.
        // Skip when pre_mutation_fields is empty (entity wasn't a row, or
        // didn't carry recoverable column names) — there's nothing to
        // delete-then-insert in that case.
        //
        // Also build the CDC damage vector here so downstream consumers
        // see which columns changed without re-diffing.
        let mut changed_columns: Option<Vec<String>> = None;
        if !applied.pre_mutation_fields.is_empty() {
            let post = entity_row_fields_snapshot(&applied.entity);
            if !post.is_empty() {
                let damage = crate::application::entity::row_damage_vector(
                    &applied.pre_mutation_fields,
                    &post,
                );
                if !damage.is_empty() {
                    changed_columns = Some(
                        damage
                            .touched_columns()
                            .into_iter()
                            .map(str::to_string)
                            .collect(),
                    );
                }

                // HOT-like fast path (P3.T2/T3): when no modified
                // column is covered by a secondary index, skip the
                // `index_entity_update` call entirely. The function
                // would short-circuit internally, but the call still
                // reads the registry lock + walks the damage vector
                // — avoiding it saves a few microseconds per UPDATE.
                // Page-local replace + t_ctid chain support (true
                // HOT) lives in a follow-up storage spec.
                // Use the parent-expanded set so that dot-path indexes
                // (e.g. "body.service.tier") are triggered when the root
                // field ("body") appears in modified_columns.
                let indexed_cols: std::collections::HashSet<String> = self
                    .index_store_ref()
                    .indexed_columns_set_with_parents(applied.collection.as_str());
                // Single-source documents keep promoted index columns in the
                // body — surface them so the index sees the value move (a no-op
                // for ordinary stored columns).
                let pre_index_fields = self
                    .index_store_ref()
                    .augment_body_derived_index_fields(&applied.pre_mutation_fields, &indexed_cols);
                let post_index_fields = self
                    .index_store_ref()
                    .augment_body_derived_index_fields(&post, &indexed_cols);
                let modified_cols: std::collections::HashSet<String> =
                    crate::application::entity::row_damage_vector(
                        &pre_index_fields,
                        &post_index_fields,
                    )
                    .touched_columns()
                    .into_iter()
                    .map(str::to_string)
                    .collect();
                if let Some(old_version) = applied.replaced_entity.as_ref() {
                    let old_index_fields: Vec<(String, Value)> = pre_index_fields
                        .iter()
                        .filter(|(col, _)| indexed_cols.contains(col))
                        .cloned()
                        .collect();
                    let new_index_fields: Vec<(String, Value)> = post_index_fields
                        .iter()
                        .filter(|(col, _)| indexed_cols.contains(col))
                        .cloned()
                        .collect();
                    if !old_index_fields.is_empty() {
                        self.index_store_ref()
                            .index_entity_delete(
                                &applied.collection,
                                old_version.id,
                                &old_index_fields,
                            )
                            .map_err(crate::RedDBError::Internal)?;
                    }
                    if !new_index_fields.is_empty() {
                        self.index_store_ref()
                            .index_entity_insert(
                                &applied.collection,
                                applied.entity.id,
                                &new_index_fields,
                            )
                            .map_err(crate::RedDBError::Internal)?;
                    }
                } else {
                    let decision = crate::storage::engine::hot_update::decide(
                        &crate::storage::engine::hot_update::HotUpdateInputs {
                            collection: applied.collection.as_str(),
                            indexed_columns: &indexed_cols,
                            modified_columns: &modified_cols,
                            // The storage layer currently handles fit via
                            // the segment abstraction; we bypass the
                            // page-size check here.
                            new_tuple_size: 0,
                            page_free_space: usize::MAX,
                        },
                    );
                    if !decision.can_hot {
                        self.index_store_ref()
                            .index_entity_update(
                                &applied.collection,
                                applied.id,
                                &pre_index_fields,
                                &post_index_fields,
                            )
                            .map_err(crate::RedDBError::Internal)?;
                    } else {
                        // F-04: `applied.collection` is tenant-supplied;
                        // strip CR/LF/control bytes via the LogField
                        // escaper (ADR 0010).
                        tracing::debug!(
                            collection = %reddb_wire::audit_safe_log_field(&applied.collection),
                            "hot_update fast-path: skipped index_entity_update"
                        );
                    }
                }
            }
        }
        drop(topology_guard);
        #[cfg(test)]
        self.index_store_ref().mutation_test_hook(
            &applied.collection,
            crate::runtime::MutationTestPhase::BeforeEvents,
        );
        self.cdc_emit_prebuilt_with_columns(
            crate::replication::cdc::ChangeOperation::Update,
            &applied.collection,
            &applied.entity,
            "entity",
            applied.metadata.as_ref(),
            true,
            changed_columns,
        );
        Ok(())
    }

    pub(crate) fn apply_loaded_patch_entity(
        &self,
        collection: String,
        entity: crate::storage::UnifiedEntity,
        payload: JsonValue,
        operations: Vec<PatchEntityOperation>,
        topology_guard: parking_lot::RwLockReadGuard<'_, ()>,
    ) -> RedDBResult<CreateEntityOutput> {
        let applied =
            self.apply_loaded_patch_entity_core(collection, entity, payload, operations)?;
        if !crate::runtime::rls_injection::entity_passes_rls_action(
            self,
            &applied.collection,
            reddb_rql::ast::PolicyAction::Update,
            &applied.entity,
        ) {
            return Err(crate::RedDBError::NotFound("entity not found".to_string()));
        }
        let _reservation = self.admit_entity_mutation(&applied)?;
        self.persist_applied_entity_mutations(std::slice::from_ref(&applied))?;
        #[cfg(test)]
        self.index_store_ref().mutation_test_hook(
            &applied.collection,
            crate::runtime::MutationTestPhase::StoragePublished,
        );
        self.flush_applied_entity_mutation(&applied, topology_guard)?;
        Ok(CreateEntityOutput {
            id: applied.id,
            entity: Some(public_document_entity(applied.entity)),
        })
    }
}

/// Shape the canonical binary-body document into the public response entity.
/// The stored `body` is the single source of truth (RDOC binary container),
/// but callers read the returned entity two ways, so the response satisfies
/// both: top-level body fields are surfaced into `named` (mirroring the GET
/// presentation derive in `named_fields_json`), and `body` itself is decoded
/// back to plain JSON. Derivation runs against the binary container first,
/// then `body` is replaced with its JSON form. This only reshapes the
/// returned copy — persistence already happened against the binary body.
fn public_document_entity(
    mut entity: crate::storage::UnifiedEntity,
) -> crate::storage::UnifiedEntity {
    let crate::storage::EntityData::Row(row) = &mut entity.data else {
        return entity;
    };
    let Some(named) = row.named.as_mut() else {
        return entity;
    };
    let Some(Value::Json(bytes)) = named.get("body") else {
        return entity;
    };
    let bytes = bytes.clone();
    if let Some(fields) = crate::document_body::body_fields(&bytes) {
        for (name, value) in fields {
            named.entry(name).or_insert(value);
        }
    }
    if let Some(body_json) = crate::document_body::decode_container_to_json(&bytes) {
        if let Ok(json_bytes) = crate::json::to_vec(&body_json) {
            named.insert("body".to_string(), Value::Json(json_bytes));
        }
    }
    entity
}

fn ensure_non_tree_reserved_metadata_patch_paths(
    operations: &[PatchEntityOperation],
) -> RedDBResult<()> {
    for operation in operations {
        let Some(key) = operation.path.first().map(String::as_str) else {
            continue;
        };
        ensure_non_tree_reserved_metadata_key(key)?;
    }
    Ok(())
}

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

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

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

impl RedDBRuntime {
    pub(crate) fn create_node_unchecked(
        &self,
        mut input: CreateNodeInput,
    ) -> RedDBResult<CreateEntityOutput> {
        self.check_collection_privilege(reddb_rql::ast::PolicyAction::Insert, &input.collection)?;
        crate::runtime::mutation::enforce_tenant_fields(
            self,
            &input.collection,
            &mut input.properties,
        )?;
        let node_type = input
            .node_type
            .clone()
            .unwrap_or_else(|| input.label.clone());
        let candidate = crate::storage::UnifiedEntity::new(
            crate::storage::EntityId::new(0),
            crate::storage::EntityKind::GraphNode(Box::new(
                crate::storage::unified::GraphNodeKind {
                    label: input.label.clone(),
                    node_type,
                },
            )),
            crate::storage::EntityData::Node(crate::storage::NodeData::with_properties(
                input.properties.iter().cloned().collect(),
            )),
        );
        ensure_insert_entity_rls(self, &input.collection, &candidate)?;
        let db = self.db();
        let contract = CollectionContractWriteEnforcer::new(self, &db, &input.collection);
        contract.ensure_model(crate::catalog::CollectionModel::Graph)?;
        let mut metadata = input.metadata;
        contract.apply_default_ttl(&mut metadata);
        let mut builder = db.node(&input.collection, &input.label);

        if let Some(node_type) = input.node_type {
            builder = builder.node_type(node_type);
        }

        for (key, value) in input.properties {
            builder = builder.property(key, value);
        }

        for (key, value) in metadata {
            builder = builder.metadata(key, value);
        }

        for embedding in input.embeddings {
            if let Some(model) = embedding.model {
                builder = builder.embedding_with_model(embedding.name, embedding.vector, model);
            } else {
                builder = builder.embedding(embedding.name, embedding.vector);
            }
        }

        for link in input.table_links {
            builder = builder.link_to_table(link.key, link.table);
        }

        for link in input.node_links {
            builder = builder.link_to_weighted(link.target, link.edge_label, link.weight);
        }

        let id = builder.save()?;
        // Phase 1.1 MVCC universal: stamp xmin so concurrent snapshots
        // don't see this node until the transaction commits.
        self.stamp_xmin_if_in_txn(&input.collection, id);
        refresh_context_index(&db, &input.collection, id)?;
        self.cdc_emit(
            crate::replication::cdc::ChangeOperation::Insert,
            &input.collection,
            id.raw(),
            "graph_node",
        );
        Ok(CreateEntityOutput {
            id,
            entity: db.store().get(&input.collection, id),
        })
    }

    pub(crate) fn create_edge_unchecked(
        &self,
        mut input: CreateEdgeInput,
    ) -> RedDBResult<CreateEntityOutput> {
        self.check_collection_privilege(reddb_rql::ast::PolicyAction::Insert, &input.collection)?;
        crate::runtime::mutation::enforce_tenant_fields(
            self,
            &input.collection,
            &mut input.properties,
        )?;
        let weight = input.weight.unwrap_or(1.0);
        let candidate = crate::storage::UnifiedEntity::new(
            crate::storage::EntityId::new(0),
            crate::storage::EntityKind::GraphEdge(Box::new(
                crate::storage::unified::GraphEdgeKind {
                    label: input.label.clone(),
                    from_node: input.from.raw().to_string(),
                    to_node: input.to.raw().to_string(),
                    weight: (weight * 1000.0) as u32,
                },
            )),
            crate::storage::EntityData::Edge(crate::storage::EdgeData::with_properties(
                weight,
                input.properties.iter().cloned().collect(),
            )),
        );
        ensure_insert_entity_rls(self, &input.collection, &candidate)?;
        let db = self.db();
        let contract = CollectionContractWriteEnforcer::new(self, &db, &input.collection);
        contract.ensure_model(crate::catalog::CollectionModel::Graph)?;
        let mut metadata = input.metadata;
        contract.apply_default_ttl(&mut metadata);
        let mut builder = db
            .edge(&input.collection, &input.label)
            .from(input.from)
            .to(input.to);

        if let Some(weight) = input.weight {
            builder = builder.weight(weight);
        }

        for (key, value) in input.properties {
            builder = builder.property(key, value);
        }

        for (key, value) in metadata {
            builder = builder.metadata(key, value);
        }

        let id = builder.save()?;
        // Phase 1.1 MVCC universal: stamp xmin on the edge so other
        // sessions don't follow it until COMMIT.
        self.stamp_xmin_if_in_txn(&input.collection, id);
        refresh_context_index(&db, &input.collection, id)?;
        self.cdc_emit(
            crate::replication::cdc::ChangeOperation::Insert,
            &input.collection,
            id.raw(),
            "graph_edge",
        );
        Ok(CreateEntityOutput {
            id,
            entity: db.store().get(&input.collection, id),
        })
    }
}

fn create_rows_batch_prevalidated_columnar_with_outputs(
    runtime: &RedDBRuntime,
    collection: String,
    column_names: std::sync::Arc<Vec<String>>,
    rows: Vec<Vec<reddb_types::Value>>,
) -> RedDBResult<Vec<CreateEntityOutput>> {
    use crate::storage::{
        unified::{EntityData, EntityKind, RowData},
        EntityId, UnifiedEntity,
    };
    use std::sync::Arc;

    if rows.is_empty() {
        return Ok(Vec::new());
    }
    if crate::runtime::execution_context::current_auth_identity().is_some() {
        let inputs = rows
            .into_iter()
            .map(|values| CreateRowInput {
                collection: collection.clone(),
                fields: column_names.iter().cloned().zip(values).collect(),
                metadata: Vec::new(),
                node_links: Vec::new(),
                vector_links: Vec::new(),
            })
            .collect();
        return runtime.create_rows_batch_prevalidated_with_outputs(CreateRowsBatchInput {
            collection,
            rows: inputs,
            suppress_events: false,
        });
    }
    // Schema prevalidation does not admit standalone index keys. Route these
    // collections through the guarded row kernel before installing any row.
    if runtime.index_store_ref().has_unique_index(&collection) {
        return runtime.create_rows_batch_columnar_with_outputs(collection, column_names, rows);
    }
    runtime.check_write(crate::runtime::write_gate::WriteKind::Dml)?;
    runtime.check_batch_size(rows.len())?;
    runtime.check_db_size()?;
    let _indexed_mutation = runtime.index_store_ref().begin_indexed_mutation();

    let db = runtime.db();
    let contract = CollectionContractWriteEnforcer::new(runtime, &db, &collection);
    contract.ensure_model(crate::catalog::CollectionModel::Table)?;

    let store = db.store();
    let table_arc: Arc<str> = Arc::from(collection.as_str());

    let indexed_cols = runtime
        .index_store_ref()
        .indexed_columns_set(collection.as_str());
    let has_secondary_indexes = !indexed_cols.is_empty();
    let mut field_snapshots: Vec<Vec<(String, reddb_types::Value)>> = if has_secondary_indexes {
        Vec::with_capacity(rows.len())
    } else {
        Vec::new()
    };

    let entities: Vec<UnifiedEntity> = rows
        .into_iter()
        .map(|values| {
            if has_secondary_indexes {
                let mut snap: Vec<(String, reddb_types::Value)> =
                    Vec::with_capacity(indexed_cols.len());
                for (name, val) in column_names.iter().zip(values.iter()) {
                    if indexed_cols.contains(name) {
                        snap.push((name.clone(), val.clone()));
                    }
                }
                field_snapshots.push(snap);
            }
            let mut row = RowData::new(values);
            row.schema = Some(Arc::clone(&column_names));
            UnifiedEntity::new(
                EntityId::new(0),
                EntityKind::TableRow {
                    table: Arc::clone(&table_arc),
                    row_id: 0,
                },
                EntityData::Row(row),
            )
        })
        .collect();

    let ids = store
        .bulk_insert(&collection, entities)
        .map_err(|e| crate::RedDBError::Internal(format!("{e:?}")))?;

    if has_secondary_indexes {
        let index_rows: Vec<(EntityId, Vec<(String, reddb_types::Value)>)> = ids
            .iter()
            .zip(field_snapshots)
            .map(|(id, fields)| (*id, fields))
            .collect();
        runtime
            .index_store_ref()
            .index_entity_insert_batch(&collection, &index_rows)
            .map_err(crate::RedDBError::Internal)?;
    }

    runtime.invalidate_result_cache();
    runtime.cdc_emit_insert_batch_no_cache_invalidate(&collection, &ids, "table");

    Ok(ids
        .into_iter()
        .map(|id| CreateEntityOutput { id, entity: None })
        .collect())
}

impl RuntimeEntityPort for RedDBRuntime {
    fn create_row(&self, input: CreateRowInput) -> RedDBResult<CreateEntityOutput> {
        self.check_write(crate::runtime::write_gate::WriteKind::Dml)?;
        let constraint_lock = crate::application::collection_contract_enforcer::row_constraint_lock(
            self,
            &input.collection,
        );
        let _constraint_guard = constraint_lock.as_ref().map(|lock| lock.lock());
        let db = self.db();
        let CreateRowInput {
            collection,
            fields,
            metadata: input_metadata,
            node_links,
            vector_links,
        } = input;
        let contract = CollectionContractWriteEnforcer::new(self, &db, &collection);
        contract.ensure_model(crate::catalog::CollectionModel::Table)?;
        let mut metadata = input_metadata;
        contract.apply_default_ttl(&mut metadata);
        let fields = contract.normalize_insert_fields(fields)?;
        contract.enforce_row_uniqueness(&fields, None)?;
        // Route through MutationEngine for unified hot path.
        let engine = self.mutation_engine();
        let result = engine.apply(
            collection.clone(),
            vec![crate::runtime::mutation::MutationRow {
                fields,
                metadata,
                node_links,
                vector_links,
            }],
        )?;
        let id = result.ids[0];
        // Perf: `db.get(id)` does a *cross-collection* scan (get_any)
        // that also takes a write lock on the entity cache. We know
        // the collection — hit the manager directly. Cuts
        // create_row() p50 roughly in half on the hot path.
        Ok(CreateEntityOutput {
            id,
            entity: db.store().get(&collection, id),
        })
    }

    fn create_rows_batch(
        &self,
        input: CreateRowsBatchInput,
    ) -> RedDBResult<Vec<CreateEntityOutput>> {
        let constraint_lock = crate::application::collection_contract_enforcer::row_constraint_lock(
            self,
            &input.collection,
        );
        let _constraint_guard = constraint_lock.as_ref().map(|lock| lock.lock());
        if input.rows.is_empty() {
            return Ok(Vec::new());
        }
        self.check_batch_size(input.rows.len())?;
        self.check_db_size()?;

        let db = self.db();
        let collection = input.collection;
        let suppress_events = input.suppress_events;
        let contract = CollectionContractWriteEnforcer::new(self, &db, &collection);
        contract.ensure_model(crate::catalog::CollectionModel::Table)?;

        let mut prepared_rows = Vec::with_capacity(input.rows.len());
        let mut uniqueness_rows = Vec::with_capacity(input.rows.len());
        for row in input.rows {
            if row.collection != collection {
                return Err(crate::RedDBError::Query(format!(
                    "batch row collection mismatch: expected '{}', got '{}'",
                    collection, row.collection
                )));
            }

            let mut metadata = row.metadata;
            contract.apply_default_ttl(&mut metadata);
            let fields = contract.normalize_insert_fields(row.fields)?;
            contract.enforce_row_uniqueness(&fields, None)?;
            uniqueness_rows.push(fields.clone());
            prepared_rows.push((fields, metadata, row.node_links, row.vector_links));
        }

        contract.enforce_batch_uniqueness(&uniqueness_rows)?;

        // Route through MutationEngine: single bulk_insert + one CDC batch
        // instead of N separate cdc_emit() calls (each acquires a write lock).
        let engine = {
            let e = self.mutation_engine();
            if suppress_events {
                e.with_suppress_events()
            } else {
                e
            }
        };
        let mutation_rows: Vec<crate::runtime::mutation::MutationRow> = prepared_rows
            .into_iter()
            .map(|(fields, metadata, node_links, vector_links)| {
                crate::runtime::mutation::MutationRow {
                    fields,
                    metadata,
                    node_links,
                    vector_links,
                }
            })
            .collect();

        let result = engine.apply(collection.clone(), mutation_rows)?;

        let store = db.store();
        Ok(result
            .ids
            .into_iter()
            .map(|id| CreateEntityOutput {
                id,
                entity: store.get(&collection, id),
            })
            .collect())
    }

    fn create_rows_batch_prevalidated_columnar(
        &self,
        collection: String,
        column_names: std::sync::Arc<Vec<String>>,
        rows: Vec<Vec<reddb_types::Value>>,
    ) -> RedDBResult<usize> {
        create_rows_batch_prevalidated_columnar_with_outputs(self, collection, column_names, rows)
            .map(|outputs| outputs.len())
    }

    fn create_rows_batch_columnar(
        &self,
        collection: String,
        column_names: std::sync::Arc<Vec<String>>,
        rows: Vec<Vec<reddb_types::Value>>,
    ) -> RedDBResult<usize> {
        self.create_rows_batch_columnar_with_outputs(collection, column_names, rows)
            .map(|outputs| outputs.len())
    }

    fn create_rows_batch_columnar_with_outputs(
        &self,
        collection: String,
        column_names: std::sync::Arc<Vec<String>>,
        rows: Vec<Vec<reddb_types::Value>>,
    ) -> RedDBResult<Vec<CreateEntityOutput>> {
        if rows.is_empty() {
            return Ok(Vec::new());
        }
        self.check_batch_size(rows.len())?;
        self.check_db_size()?;

        let db = self.db();
        let contract = CollectionContractWriteEnforcer::new(self, &db, &collection);
        contract.ensure_model(crate::catalog::CollectionModel::Table)?;

        // Fast path: when the collection carries no contract (or the
        // contract has no declared columns) `normalize_insert_fields`
        // is a no-op and `enforce_row_uniqueness` finds no unique
        // columns to check. Skip the per-row `(String, Value)` tuple
        // materialisation entirely and route through the prevalidated
        // columnar kernel — same write, CDC, and index path, just
        // without the wasted (String, Value) clones. This is the
        // bench `bench_users` shape (no contract declared by the
        // adapter's `setup_schema`).
        let needs_normalisation = self.index_store_ref().has_unique_index(&collection)
            || match db.collection_contract(&collection) {
                Some(c) => {
                    c.declared_model == crate::catalog::CollectionModel::Table
                        && (!c.declared_columns.is_empty()
                            || c.table_def
                                .as_ref()
                                .map(|t| !t.columns.is_empty())
                                .unwrap_or(false))
                }
                None => false,
            };
        if !needs_normalisation {
            return create_rows_batch_prevalidated_columnar_with_outputs(
                self,
                collection,
                column_names,
                rows,
            );
        }

        // Slow path: contract requires per-row normalisation /
        // uniqueness checks. Materialise `Vec<(String, Value)>` from
        // the columnar layout (this still pays N×ncols `String::clone`
        // — but it's deferred out of the wire decoder hot loop and
        // happens behind a runtime API, not the per-row decode loop)
        // and fall through to the existing `create_rows_batch` path.
        let ncols = column_names.len();
        let tuple_rows: Vec<CreateRowInput> = rows
            .into_iter()
            .map(|values| {
                let mut fields: Vec<(String, reddb_types::Value)> = Vec::with_capacity(ncols);
                for (name, value) in column_names.iter().zip(values) {
                    fields.push((name.clone(), value));
                }
                CreateRowInput {
                    collection: collection.clone(),
                    fields,
                    metadata: Vec::new(),
                    node_links: Vec::new(),
                    vector_links: Vec::new(),
                }
            })
            .collect();
        self.create_rows_batch(CreateRowsBatchInput {
            collection,
            rows: tuple_rows,
            suppress_events: false,
        })
    }

    fn create_rows_batch_prevalidated(&self, input: CreateRowsBatchInput) -> RedDBResult<usize> {
        self.create_rows_batch_prevalidated_with_outputs(input)
            .map(|outputs| outputs.len())
    }

    fn create_rows_batch_prevalidated_with_outputs(
        &self,
        input: CreateRowsBatchInput,
    ) -> RedDBResult<Vec<CreateEntityOutput>> {
        if input.rows.is_empty() {
            return Ok(Vec::new());
        }
        self.check_write(crate::runtime::write_gate::WriteKind::Dml)?;
        self.check_batch_size(input.rows.len())?;
        self.check_db_size()?;

        let db = self.db();
        let collection = input.collection;
        let suppress_events = input.suppress_events;
        // Still verify the collection's declared model before we blast
        // rows at it — this one-off check is O(1), independent of
        // ncols, and catches schema-kind mismatches that the client
        // can't always see.
        let contract = CollectionContractWriteEnforcer::new(self, &db, &collection);
        contract.ensure_model(crate::catalog::CollectionModel::Table)?;

        // Hoist the per-collection default TTL lookup out of the
        // per-row loop — it only depends on the collection, not on
        // individual rows, and the old path did one HashMap read per
        // row (25k reads for a 25k typed_insert bulk). Fetch it once,
        // apply to any row whose metadata doesn't already carry a TTL.
        let default_ttl_plan = contract.default_ttl_plan();

        let mutation_rows: Vec<crate::runtime::mutation::MutationRow> =
            Vec::with_capacity(input.rows.len());
        let mut mutation_rows = mutation_rows;
        for row in input.rows {
            if row.collection != collection {
                return Err(crate::RedDBError::Query(format!(
                    "batch row collection mismatch: expected '{}', got '{}'",
                    collection, row.collection
                )));
            }
            let mut metadata = row.metadata;
            default_ttl_plan.apply(&mut metadata);
            mutation_rows.push(crate::runtime::mutation::MutationRow {
                fields: row.fields,
                metadata,
                node_links: row.node_links,
                vector_links: row.vector_links,
            });
        }

        let engine = {
            let engine = self.mutation_engine();
            if suppress_events {
                engine.with_suppress_events()
            } else {
                engine
            }
        };
        let result = engine.apply(collection.clone(), mutation_rows)?;
        let store = db.store();
        Ok(result
            .ids
            .into_iter()
            .map(|id| CreateEntityOutput {
                id,
                entity: store.get(&collection, id),
            })
            .collect())
    }

    fn create_node(&self, input: CreateNodeInput) -> RedDBResult<CreateEntityOutput> {
        self.check_write(crate::runtime::write_gate::WriteKind::Dml)?;
        crate::reserved_fields::ensure_no_reserved_public_item_fields(
            input.properties.iter().map(|(key, _)| key.as_str()),
            &format!("node '{}'", input.collection),
        )?;
        ensure_non_tree_reserved_metadata_entries(&input.metadata)?;
        self.create_node_unchecked(input)
    }

    fn create_edge(&self, input: CreateEdgeInput) -> RedDBResult<CreateEntityOutput> {
        self.check_write(crate::runtime::write_gate::WriteKind::Dml)?;
        crate::reserved_fields::ensure_no_reserved_public_item_fields(
            input.properties.iter().map(|(key, _)| key.as_str()),
            &format!("edge '{}'", input.collection),
        )?;
        ensure_non_tree_structural_edge_label(&input.label)?;
        ensure_non_tree_reserved_metadata_entries(&input.metadata)?;
        self.create_edge_unchecked(input)
    }

    fn create_vector(&self, input: CreateVectorInput) -> RedDBResult<CreateEntityOutput> {
        self.check_write(crate::runtime::write_gate::WriteKind::Dml)?;
        self.check_collection_privilege(reddb_rql::ast::PolicyAction::Insert, &input.collection)?;
        let mut vector_data = crate::storage::VectorData::new(input.dense.clone());
        vector_data.content = input.content.clone();
        let candidate = crate::storage::UnifiedEntity::new(
            crate::storage::EntityId::new(0),
            crate::storage::EntityKind::Vector {
                collection: input.collection.clone(),
            },
            crate::storage::EntityData::Vector(vector_data),
        );
        ensure_insert_entity_rls(self, &input.collection, &candidate)?;
        let db = self.db();
        let contract = CollectionContractWriteEnforcer::new(self, &db, &input.collection);
        contract.ensure_model(crate::catalog::CollectionModel::Vector)?;
        contract.ensure_vector_dimension(input.dense.len())?;
        let mut metadata = input.metadata;
        contract.apply_default_ttl(&mut metadata);
        let mut builder = db.vector(&input.collection).dense(input.dense);

        if let Some(content) = input.content {
            builder = builder.content(content);
        }

        for (key, value) in metadata {
            builder = builder.metadata(key, value);
        }

        if let Some(link_row) = input.link_row {
            builder = builder.link_to_table(link_row);
        }

        if let Some(link_node) = input.link_node {
            builder = builder.link_to_node(link_node);
        }

        // The transaction stamp must be present on the first published version.
        // Stamping after save exposed an xmin=0 vector to concurrent snapshots.
        let id = builder.save_with_xmin(self.current_xid())?;
        let dense_for_turbo = match db.store().get(&input.collection, id).map(|e| e.data) {
            Some(crate::storage::unified::EntityData::Vector(data)) => Some(data.dense.clone()),
            _ => None,
        };
        refresh_context_index(&db, &input.collection, id)?;
        // Issue #693 — vector.turbo write path. Order matters and
        // matches the write sequence the next slice (#673) will
        // attach kill-points to: WAL append → in-memory index update
        // → TurboExtent append. The legacy `vector` path is the
        // `None` branch and is untouched.
        if let (Some(state), Some(dense)) = (db.turbo_state(&input.collection), dense_for_turbo) {
            // Lazy populate so an INSERT after restart sees the
            // pre-restart vectors in the index before appending its
            // own.
            state.ensure_populated(&db.store(), &input.collection)?;
            // Issue #694 — named crash boundaries for #673.
            use crate::runtime::turbo_crash_inject::{fire, InjectionPoint};
            fire(InjectionPoint::BeforeWalFsync);
            let _ = db
                .store()
                .append_vector_insert_record(&input.collection, id.raw(), &dense);
            fire(InjectionPoint::BeforeIndexCommit);
            let mut index = state.index.lock();
            index.insert(id, dense.clone());
            // Mirror the encoded codes into the per-collection
            // TurboExtent when one is available. The bytes are the
            // 4-bit-packed code groups (`n_byte_groups`) plus the
            // little-endian L2 norm — same shape `BlockedCodeStorage`
            // appends in-memory. Failures are non-fatal in this
            // slice; durability of the extent is owned by #673.
            fire(InjectionPoint::BeforeExtentFsync);
            if let Some(extent) = state.extent.lock().as_mut() {
                let packed = index.encode_for_extent(&dense);
                let _ = extent.append(&packed);
            }
        }
        self.cdc_emit(
            crate::replication::cdc::ChangeOperation::Insert,
            &input.collection,
            id.raw(),
            "vector",
        );
        Ok(CreateEntityOutput {
            id,
            entity: db.store().get(&input.collection, id),
        })
    }

    fn create_document(&self, input: CreateDocumentInput) -> RedDBResult<CreateEntityOutput> {
        let _indexed_mutation = self.index_store_ref().begin_indexed_mutation();
        self.check_write(crate::runtime::write_gate::WriteKind::Dml)?;
        self.check_collection_privilege(reddb_rql::ast::PolicyAction::Insert, &input.collection)?;
        let db = self.db();
        let contract = CollectionContractWriteEnforcer::new(self, &db, &input.collection);
        contract.ensure_model(crate::catalog::CollectionModel::Document)?;

        if let JsonValue::Object(ref map) = input.body {
            crate::reserved_fields::ensure_no_reserved_public_item_fields(
                map.keys().map(String::as_str),
                &format!("document '{}'", input.collection),
            )?;
        }

        // Serialize the full body for the "body" field. Behind the
        // `storage.binary_document_body` flag (PRD-1398) this is the native binary
        // container; otherwise plain JSON. Reads decode either form to JSON.
        let mut fields: Vec<(String, reddb_types::Value)> = vec![(
            "body".to_string(),
            reddb_types::Value::Json(
                crate::json::to_vec(&input.body)
                    .map_err(|err| crate::RedDBError::Query(err.to_string()))?,
            ),
        )];
        crate::runtime::mutation::enforce_tenant_fields(self, &input.collection, &mut fields)?;
        let (field_names, field_values) =
            crate::runtime::impl_dml_support::pairwise_columns_values(&fields);
        let scoped_body =
            crate::runtime::impl_dml_support::find_document_body_json(&field_names, &field_values)?;
        let body_bytes = crate::document_body::serialize_document_body(
            &scoped_body,
            self.binary_document_body_enabled(),
        )?;
        if let Some((_, body)) = fields.iter_mut().find(|(name, _)| name == "body") {
            *body = reddb_types::Value::Json(body_bytes);
        }
        let mut row = crate::storage::RowData::new(Vec::new());
        row.named = Some(fields.iter().cloned().collect());
        let candidate = crate::storage::UnifiedEntity::new(
            crate::storage::EntityId::new(0),
            crate::storage::EntityKind::TableRow {
                table: std::sync::Arc::from(input.collection.as_str()),
                row_id: 0,
            },
            crate::storage::EntityData::Row(row),
        );
        ensure_insert_entity_rls(self, &input.collection, &candidate)?;

        let mut metadata = input.metadata;
        contract.apply_default_ttl(&mut metadata);
        let columns: Vec<(&str, reddb_types::Value)> = fields
            .iter()
            .map(|(key, value)| (key.as_str(), value.clone()))
            .collect();
        let mut builder = db.row(&input.collection, columns);

        for (key, value) in metadata {
            builder = builder.metadata(key, value);
        }

        for node in input.node_links {
            builder = builder.link_to_node(node);
        }

        for vector in input.vector_links {
            builder = builder.link_to_vector(vector);
        }

        let id = builder.save()?;
        // Phase 1.1 MVCC universal: stamp xmin on the document.
        self.stamp_xmin_if_in_txn(&input.collection, id);
        self.index_store_ref()
            .index_entity_insert(&input.collection, id, &fields)
            .map_err(crate::RedDBError::Internal)?;
        refresh_context_index(&db, &input.collection, id)?;
        self.cdc_emit(
            crate::replication::cdc::ChangeOperation::Insert,
            &input.collection,
            id.raw(),
            "document",
        );
        Ok(CreateEntityOutput {
            id,
            entity: db.store().get(&input.collection, id),
        })
    }

    fn create_kv(&self, input: CreateKvInput) -> RedDBResult<CreateEntityOutput> {
        crate::runtime::kv_write::create(self, input, None)
    }

    fn create_timeseries_point(
        &self,
        input: CreateTimeSeriesPointInput,
    ) -> RedDBResult<CreateEntityOutput> {
        self.check_write(crate::runtime::write_gate::WriteKind::Dml)?;
        self.check_collection_privilege(reddb_rql::ast::PolicyAction::Insert, &input.collection)?;
        let db = self.db();
        let contract = CollectionContractWriteEnforcer::new(self, &db, &input.collection);
        contract.ensure_model(crate::catalog::CollectionModel::TimeSeries)?;

        let mut fields = vec![
            ("metric".to_string(), reddb_types::Value::text(input.metric)),
            ("value".to_string(), reddb_types::Value::Float(input.value)),
        ];

        if let Some(timestamp_ns) = input.timestamp_ns {
            fields.push((
                "timestamp".to_string(),
                reddb_types::Value::UnsignedInteger(timestamp_ns),
            ));
        }

        if !input.tags.is_empty() {
            let tags_json = JsonValue::Object(
                input
                    .tags
                    .into_iter()
                    .map(|(key, value)| (key, JsonValue::String(value)))
                    .collect(),
            );
            let tags_bytes = json_to_vec(&tags_json).map_err(|err| {
                crate::RedDBError::Query(format!("failed to serialize timeseries tags: {err}"))
            })?;
            fields.push(("tags".to_string(), reddb_types::Value::Json(tags_bytes)));
        }

        let collection = input.collection;
        let id = self.insert_timeseries_point(&collection, fields, input.metadata)?;
        // Phase 1.1 MVCC universal: stamp xmin on the point so
        // concurrent range scans hide it until COMMIT.
        self.stamp_xmin_if_in_txn(&collection, id);
        refresh_context_index(&db, &collection, id)?;

        Ok(CreateEntityOutput {
            id,
            entity: db.store().get(&collection, id),
        })
    }

    fn get_kv(
        &self,
        collection: &str,
        key: &str,
    ) -> RedDBResult<Option<(reddb_types::Value, crate::storage::EntityId)>> {
        self.check_collection_privilege(reddb_rql::ast::PolicyAction::Select, collection)?;
        let resolver = crate::runtime::kv_read::KvReadResolver::for_rest_kv(self, collection)?;
        let Some(record) = resolver.record(key)? else {
            return Ok(None);
        };
        let visible = self
            .db()
            .store()
            .get(collection, record.id)
            .is_some_and(|entity| {
                crate::runtime::rls_injection::entity_passes_rls_action(
                    self,
                    collection,
                    reddb_rql::ast::PolicyAction::Select,
                    &entity,
                )
            });
        Ok(visible.then_some((record.value.unwrap_or(Value::Null), record.id)))
    }

    fn delete_kv(&self, collection: &str, key: &str) -> RedDBResult<bool> {
        self.check_write(crate::runtime::write_gate::WriteKind::Dml)?;
        crate::runtime::impl_kv::KvAtomicOps::new(self).delete(
            crate::catalog::CollectionModel::Kv,
            collection,
            key,
        )
    }

    fn patch_entity(&self, input: PatchEntityInput) -> RedDBResult<CreateEntityOutput> {
        self.check_write(crate::runtime::write_gate::WriteKind::Dml)?;
        let _indexed_mutation = self.index_store_ref().begin_indexed_mutation();
        self.check_collection_privilege(reddb_rql::ast::PolicyAction::Update, &input.collection)?;
        let constraint_lock = crate::application::collection_contract_enforcer::row_constraint_lock(
            self,
            &input.collection,
        );
        let _constraint_guard = constraint_lock.as_ref().map(|lock| lock.lock());
        let topology_lock = self
            .index_store_ref()
            .collection_topology_lock(&input.collection);
        let topology_guard = topology_lock.read();
        let PatchEntityInput {
            collection,
            id,
            payload,
            operations,
        } = input;
        let db = self.db();
        let store = db.store();
        let Some(manager) = store.get_collection(&collection) else {
            return Err(crate::RedDBError::NotFound(format!(
                "collection not found: {collection}"
            )));
        };
        let Some(entity) = manager.get(id) else {
            return Err(crate::RedDBError::NotFound(format!(
                "entity not found: {}",
                id.raw()
            )));
        };
        if !crate::runtime::rls_injection::entity_passes_rls_action(
            self,
            &collection,
            reddb_rql::ast::PolicyAction::Update,
            &entity,
        ) {
            return Err(crate::RedDBError::NotFound(format!(
                "entity not found: {}",
                id.raw()
            )));
        }
        self.apply_loaded_patch_entity(collection, entity, payload, operations, topology_guard)
    }

    fn delete_entity(&self, input: DeleteEntityInput) -> RedDBResult<DeleteEntityOutput> {
        self.check_write(crate::runtime::write_gate::WriteKind::Dml)?;
        let _indexed_mutation = self.index_store_ref().begin_indexed_mutation();
        self.check_collection_privilege(reddb_rql::ast::PolicyAction::Delete, &input.collection)?;
        let constraint_lock = crate::application::collection_contract_enforcer::row_constraint_lock(
            self,
            &input.collection,
        );
        let _constraint_guard = constraint_lock.as_ref().map(|lock| lock.lock());
        let topology_lock = self
            .index_store_ref()
            .collection_topology_lock(&input.collection);
        let topology_guard = topology_lock.read();
        let store = self.db().store();
        // Snapshot row fields before delete so we can mirror the removal
        // into every secondary index. The fetch is best-effort: if the
        // entity is already gone, the delete below is a no-op anyway.
        let Some(entity) = store.get(&input.collection, input.id) else {
            return Ok(DeleteEntityOutput {
                deleted: false,
                id: input.id,
            });
        };
        if !crate::runtime::rls_injection::entity_passes_rls_action(
            self,
            &input.collection,
            reddb_rql::ast::PolicyAction::Delete,
            &entity,
        ) {
            return Ok(DeleteEntityOutput {
                deleted: false,
                id: input.id,
            });
        }
        let pre_delete_fields = entity_row_fields_snapshot(&entity);
        // Store delete first (source of truth). Crash between here and index removal
        // leaves the entity invisible to most queries but recoverable; the reverse
        // (remove index first, then crash) leaves the entity permanently orphaned.
        let deleted = store
            .delete(&input.collection, input.id)
            .map_err(|err| crate::RedDBError::Internal(err.to_string()))?;
        if deleted {
            #[cfg(test)]
            self.index_store_ref().mutation_test_hook(
                &input.collection,
                crate::runtime::MutationTestPhase::StoragePublished,
            );
            store.context_index().remove_entity(input.id);
            // Secondary index maintenance — surface only registry-shape
            // errors; missing-index removals are tolerated inside the call.
            if !pre_delete_fields.is_empty() {
                self.index_store_ref()
                    .index_entity_delete(&input.collection, input.id, &pre_delete_fields)
                    .map_err(crate::RedDBError::Internal)?;
            }
            drop(topology_guard);
            #[cfg(test)]
            self.index_store_ref().mutation_test_hook(
                &input.collection,
                crate::runtime::MutationTestPhase::BeforeEvents,
            );
            self.cdc_emit(
                crate::replication::cdc::ChangeOperation::Delete,
                &input.collection,
                input.id.raw(),
                "entity",
            );
        }
        Ok(DeleteEntityOutput {
            deleted,
            id: input.id,
        })
    }
}