lora_store/memory/

graph.rs

//! The [`InMemoryGraph`] data structure: slot-indexed node/relationship
//! storage, adjacency lists, label/type indexes, and the inherent
//! helpers that the trait impls in `super::impls` delegate to.

use std::collections::BTreeMap;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, RwLock, RwLockWriteGuard};

use lora_ast::Direction;

use crate::{
    DeletedRecordSink, Labels, LoraPoint, MutationEvent, MutationRecorder, NodeId, NodeRecord,
    Properties, PropertyValue, RelationshipId, RelationshipRecord,
};

use super::adjacency::{self, AdjEntry, AdjSide, Adjacency, Side, TypeFilter};
use super::chunked_vec::ChunkedVec;
use super::constraint_catalog::{
    ConstraintBacking, ConstraintCatalog, ConstraintRequest, CreateConstraintError,
    CreateConstraintOutcome, DropConstraintError, DropConstraintOutcome,
};
use super::distinct_stats::DistinctStatsRegistry;
use super::entity_index_store::{
    read_shared, share, write_shared, IndexBundle, IndexRead, IndexWrite,
};
use super::fulltext_index::FulltextRegistry;
use super::hnsw::HnswParams;
use super::id_bitmap::IdBitmap;
use super::index_catalog::{
    CreateIndexError, CreateIndexOutcome, DropIndexError, DropIndexOutcome, IndexCatalog,
    IndexConfigValue, IndexDefinition, IndexRequest, StoredIndexEntity, StoredIndexKind,
    StoredIndexState,
};
use super::point_index::PointRegistry;
use super::property_index::{
    EntityValues, PropertyIndexKey, PropertyIndexRegistry, PropertyIndexState, UNLABELLED,
};
use super::record_slab::{BuiltLeaves, RecordSlab};
use super::secondary_index_maintenance::SecondaryIndexMutation;
use super::sorted_property_index::SortedPropertyIndex;
use super::stats::GraphStats;
use super::text_index::TrigramRegistry;
use super::vector_index::{VectorIndexProvider, VectorIndexRegistry, VectorSimilarity};
use crate::dict::Dicts;
use crate::encoded::{self, Scratch, StoredNode, StoredRel};
use crate::{NodeRef, RelRef};

/// The largest node or relationship id the store accepts. Slot storage
/// grows with the highest id ever used (about 16 bytes per 512 ids even
/// when the ids are deleted), so a corrupt snapshot or WAL claiming an id
/// near `u64::MAX` would otherwise allocate until the process aborts. At
/// 2^34 the worst case is about 512 MiB of slot tables, and no real
/// database allocates 17 billion ids.
pub(crate) const MAX_ENTITY_ID: u64 = (1 << 34) - 1;

/// Adjacency entries a batch delete still has to remove: node, side,
/// relationship. See `InMemoryGraph::flush_adjacency_removals`.
pub(super) type PendingAdjRemovals = Vec<(NodeId, Side, RelationshipId)>;

/// Lists up to this many bytes (about 32 entries) lose an entry at once
/// even in a batch delete: scanning them costs less than deferring.
const DEFER_REMOVAL_MIN_BYTES: usize = 256;

#[derive(Default)]
pub struct InMemoryGraph {
    pub(super) next_node_id: NodeId,
    pub(super) next_rel_id: RelationshipId,

    /// Slot-indexed node storage: slot `id` holds the node `id`, encoded
    /// (see [`crate::encoded`]). Empty slots are tombstones from deletes
    /// (we don't compact). Because `next_node_id` is monotonic the slot at
    /// `id` exists exactly when `id < next_node_id`.
    ///
    /// [`Self::clone`] (called on every staged write to build a working
    /// copy, and for every reader snapshot) shares this storage: a
    /// [`RecordSlab`] clone is one refcount bump, so the whole-graph
    /// clone is O(#labels + #relationship types), not O(N). A write
    /// copies only the leaf path it touches: a leaf's offsets, and at
    /// most one buffer of 64 records. A record is immutable: changing it
    /// stores a new encoding in its slot (see [`Self::update_node`]).
    pub(super) nodes: RecordSlab,
    pub(super) relationships: RecordSlab,
    /// Numbers the labels, relationship types and property keys the
    /// records and adjacency entries refer to. Shared with clones until
    /// one of them sees a new name.
    pub(super) dicts: Dicts,
    /// Live (non-tombstoned) counts kept in sync with `put_*`/`take_*` so
    /// `node_count` / `relationship_count` stay O(1) — without a counter
    /// they'd have to scan the slab.
    pub(super) live_node_count: usize,
    pub(super) live_rel_count: usize,

    /// Adjacency keyed by NodeId. `adjacency[id]` lists the relationships
    /// that leave `id` as `(type, target, relationship id)` and those that
    /// enter it with the source as the neighbour, in one allocation. A hop
    /// reads only the list: it never touches a relationship record. See
    /// [`super::adjacency`].
    pub(super) adjacency: ChunkedVec<Adjacency>,

    // secondary indexes
    /// Label -> the ids of the nodes that carry it, ascending. An
    /// [`IdBitmap`] holds an id in two bytes or less (an eighth of a byte
    /// for a label most nodes carry), where a list held eight, and
    /// removes one by binary search.
    ///
    /// Keyed by `Arc<str>`: every staged write clones the graph, and with
    /// `String` keys that cloned (allocated) every label and type name.
    pub(super) nodes_by_label: BTreeMap<Arc<str>, IdBitmap>,
    pub(super) relationships_by_type: BTreeMap<Arc<str>, IdBitmap>,

    /// All index machinery — the declared-index catalog, hash-bucket
    /// property registry, and the per-entity-kind secondary index
    /// registries (text, sorted, point, fulltext) plus their active
    /// counters — collapsed into one bundle. See [`IndexBundle`] for
    /// the rationale. The bundle is a packaging-only abstraction:
    /// every field accessed through `self.indexes.<x>` lives at the
    /// same address it would have as a top-level field.
    pub(super) indexes: IndexBundle,

    /// Per-(label / type, property key) distinct-value sketches for the
    /// planner (see [`super::distinct_stats`]). Kept for every key, not
    /// only indexed ones, and a function of the live data alone, so
    /// plans don't depend on index activation, lookup history or
    /// restarts. Cloning is four refcount bumps.
    pub(super) distinct_stats: DistinctStatsRegistry,

    /// Catalog of explicitly-created constraints (CREATE CONSTRAINT).
    /// Deliberately not part of [`IndexBundle`] — constraints describe
    /// data invariants, not indexed access. The fact that uniqueness /
    /// key constraints back range indexes is handled in the
    /// constraint code path, not by the bundle's layout.
    pub(super) constraint_catalog: RwLock<Arc<ConstraintCatalog>>,
    /// Fast-path counter for mutation-time constraint checks. Most
    /// workloads have no constraints installed; this lets the executor
    /// skip taking the catalog lock in that case.
    pub(super) active_constraints: AtomicUsize,

    /// Optional mutation observer. When `Some`, every committed mutation
    /// fans out to this recorder *after* the in-memory state has been
    /// updated. The recorder is not part of the graph's identity, so Clone
    /// and snapshot restore both reset it to `None`.
    pub(super) recorder: Option<Arc<dyn MutationRecorder>>,

    /// Optional sink that sees each node / relationship record just before
    /// a delete drops it. Change feeds use it to report deleted entities
    /// without copying the graph. Like the recorder, it is not part of the
    /// graph's identity and is dropped on clone.
    pub(super) deleted_sink: Option<Arc<dyn DeletedRecordSink>>,
}

impl std::fmt::Debug for InMemoryGraph {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("InMemoryGraph")
            .field("next_node_id", &self.next_node_id)
            .field("next_rel_id", &self.next_rel_id)
            .field(
                "nodes",
                &self.iter_nodes().map(|(_, n)| n).collect::<Vec<_>>(),
            )
            .field(
                "relationships",
                &self.iter_rels().map(|(_, r)| r).collect::<Vec<_>>(),
            )
            .field("adjacency", &self.adjacency)
            .field("nodes_by_label", &self.nodes_by_label)
            .field("relationships_by_type", &self.relationships_by_type)
            .field("indexes", &self.indexes)
            .field(
                "active_node_property_indexes",
                &self.active_node_property_index_count(),
            )
            .field(
                "active_relationship_property_indexes",
                &self.active_relationship_property_index_count(),
            )
            .field(
                "index_catalog_entries",
                &self
                    .indexes
                    .catalog
                    .read()
                    .map(|c| c.list().len())
                    .unwrap_or(0),
            )
            .field("active_constraints", &self.active_constraint_count())
            .field(
                "active_fulltext_indexes",
                &self.active_fulltext_index_count(),
            )
            .field("recorder", &self.recorder.as_ref().map(|_| "installed"))
            .finish()
    }
}

impl Clone for InMemoryGraph {
    fn clone(&self) -> Self {
        // Deliberately drop the recorder on clone: a cloned store is a
        // separate identity; it should not silently share the observer.
        Self {
            next_node_id: self.next_node_id,
            next_rel_id: self.next_rel_id,
            nodes: self.nodes.clone(),
            relationships: self.relationships.clone(),
            live_node_count: self.live_node_count,
            live_rel_count: self.live_rel_count,
            adjacency: self.adjacency.clone(),
            dicts: self.dicts.clone(),
            nodes_by_label: self.nodes_by_label.clone(),
            relationships_by_type: self.relationships_by_type.clone(),
            // Shares every registry and catalog; each is copied on its
            // first write (see `IndexBundle`).
            indexes: self.indexes.clone(),
            distinct_stats: self.distinct_stats.clone(),
            constraint_catalog: share(&self.constraint_catalog),
            active_constraints: AtomicUsize::new(self.active_constraint_count()),
            recorder: None,
            deleted_sink: None,
        }
    }
}

impl InMemoryGraph {
    pub fn new() -> Self {
        Self::default()
    }

    /// Kept for API compatibility. The chunked storage allocates full
    /// chunks as the graph grows, so there is no repeated doubling for a
    /// capacity hint to avoid.
    pub fn with_capacity_hint(_nodes: usize, _relationships: usize) -> Self {
        Self::default()
    }

    pub fn contains_node(&self, node_id: NodeId) -> bool {
        self.has_node_at(node_id)
    }

    pub fn contains_relationship(&self, rel_id: RelationshipId) -> bool {
        self.has_rel_at(rel_id)
    }

    /// Install (or clear) the mutation recorder. Passing `None` detaches any
    /// currently-installed recorder. The recorder observes every committed
    /// mutation *after* it has been applied.
    pub fn set_mutation_recorder(&mut self, recorder: Option<Arc<dyn MutationRecorder>>) {
        self.recorder = recorder;
    }

    /// Install (or clear) the [`DeletedRecordSink`].
    pub fn set_deleted_record_sink(&mut self, sink: Option<Arc<dyn DeletedRecordSink>>) {
        self.deleted_sink = sink;
    }

    /// Handle to the currently-installed recorder, if any.
    pub fn mutation_recorder(&self) -> Option<&Arc<dyn MutationRecorder>> {
        self.recorder.as_ref()
    }

    /// Emit a mutation event only if a recorder is installed. The event is
    /// built lazily — callers pass a closure, so when no recorder is
    /// attached we pay only a `None` check and the cost of constructing the
    /// event (labels/properties clones) is avoided.
    #[inline]
    pub(super) fn emit<F: FnOnce() -> MutationEvent>(&self, build: F) {
        if let Some(rec) = &self.recorder {
            rec.record(build());
        }
    }

    fn bump_next_node_id_past(&mut self, id: NodeId) -> Result<(), String> {
        let next = id
            .checked_add(1)
            .ok_or_else(|| format!("node id {id} leaves no valid next node id"))?;
        self.next_node_id = self.next_node_id.max(next);
        Ok(())
    }

    fn bump_next_rel_id_past(&mut self, id: RelationshipId) -> Result<(), String> {
        let next = id
            .checked_add(1)
            .ok_or_else(|| format!("relationship id {id} leaves no valid next relationship id"))?;
        self.next_rel_id = self.next_rel_id.max(next);
        Ok(())
    }

    pub(super) fn try_reserve_next_node_slot(&mut self) -> Option<(NodeId, usize)> {
        let id = self.next_node_id;
        let idx = self.ensure_node_slot_checked(id).ok()?;
        self.bump_next_node_id_past(id).ok()?;
        Some((id, idx))
    }

    pub(super) fn try_reserve_next_rel_slot(&mut self) -> Option<(RelationshipId, usize)> {
        let id = self.next_rel_id;
        let idx = self.ensure_rel_slot_checked(id).ok()?;
        self.bump_next_rel_id_past(id).ok()?;
        Some((id, idx))
    }

    // ---------- Slab access helpers ----------
    //
    // Slots are indexed by id. Readers get views over the encoded
    // records; nothing here hands out the bytes.

    #[inline]
    pub(super) fn node_at(&self, id: NodeId) -> Option<NodeRef<'_>> {
        let bytes = self.nodes.get(Self::slot_index(id)?)?;
        Some(NodeRef::stored(StoredNode::new(id, bytes, &self.dicts)))
    }

    #[inline]
    pub(super) fn has_node_at(&self, id: NodeId) -> bool {
        Self::slot_index(id).is_some_and(|idx| self.nodes.contains(idx))
    }

    #[inline]
    pub(super) fn rel_at(&self, id: RelationshipId) -> Option<RelRef<'_>> {
        let bytes = self.relationships.get(Self::slot_index(id)?)?;
        Some(RelRef::stored(StoredRel::new(id, bytes, &self.dicts)))
    }

    #[inline]
    pub(super) fn has_rel_at(&self, id: RelationshipId) -> bool {
        Self::slot_index(id).is_some_and(|idx| self.relationships.contains(idx))
    }

    /// `(source, target)` of a relationship, reading only the start of
    /// its record.
    #[inline]
    pub(super) fn rel_endpoints_at(&self, id: RelationshipId) -> Option<(NodeId, NodeId)> {
        let bytes = self.relationships.get(Self::slot_index(id)?)?;
        Some(encoded::rel_endpoints(bytes))
    }

    /// Change a node: decode it, let `change` edit the record, and store
    /// the new encoding. Records are immutable, so snapshots that share
    /// the old one keep reading it.
    pub(super) fn update_node<R>(
        &mut self,
        id: NodeId,
        change: impl FnOnce(&mut NodeRecord) -> R,
    ) -> Option<R> {
        let idx = Self::slot_index(id)?;
        let mut record = self.node_at(id)?.to_record();
        let result = change(&mut record);
        self.store_node(idx, &record);
        Some(result)
    }

    /// Set or remove one property of a node, rewriting only that entry of
    /// its record. Returns the property's previous value; `None` when the
    /// node does not exist or there was nothing to remove.
    pub(super) fn edit_node_property(
        &mut self,
        id: NodeId,
        key: &str,
        edit: encoded::PropEdit<'_>,
    ) -> Option<Option<PropertyValue>> {
        let idx = Self::slot_index(id)?;
        edit_record_property(
            &mut self.nodes,
            idx,
            encoded::RecordKind::Node,
            &mut self.dicts,
            key,
            edit,
        )
    }

    pub(super) fn edit_rel_property(
        &mut self,
        id: RelationshipId,
        key: &str,
        edit: encoded::PropEdit<'_>,
    ) -> Option<Option<PropertyValue>> {
        let idx = Self::slot_index(id)?;
        edit_record_property(
            &mut self.relationships,
            idx,
            encoded::RecordKind::Relationship,
            &mut self.dicts,
            key,
            edit,
        )
    }

    /// Resize the node-keyed Vecs so `id as usize` is in range. Adjacency
    /// lists are kept in lockstep with `nodes`, so a freshly-grown slot has
    /// empty outgoing/incoming Vecs ready to receive edges.
    fn slot_len_for_id(id: u64, kind: &str) -> Result<usize, String> {
        if id > MAX_ENTITY_ID {
            return Err(format!(
                "{kind} id {id} is above the largest supported id {MAX_ENTITY_ID}"
            ));
        }
        let idx = usize::try_from(id)
            .map_err(|_| format!("{kind} id {id} does not fit in usize on this platform"))?;
        idx.checked_add(1)
            .ok_or_else(|| format!("{kind} id {id} leaves no valid slab slot"))
    }

    #[inline]
    fn slot_index(id: u64) -> Option<usize> {
        usize::try_from(id).ok()
    }

    fn ensure_node_slot_checked(&mut self, id: NodeId) -> Result<usize, String> {
        let target = Self::slot_len_for_id(id, "node")?;
        if self.nodes.len() < target {
            let additional = target - self.nodes.len();
            self.nodes.try_reserve(additional).map_err(|e| {
                format!("node id {id} requires {target} slots, but allocation failed: {e}")
            })?;
            self.adjacency.try_reserve_exact(additional).map_err(|e| {
                format!(
                    "node id {id} requires {target} adjacency slots, but allocation failed: {e}"
                )
            })?;
            self.nodes.grow_to(target);
            self.adjacency.grow_shared(target, adjacency::empty_leaf());
        }
        Ok(target - 1)
    }

    fn ensure_rel_slot_checked(&mut self, id: RelationshipId) -> Result<usize, String> {
        let target = Self::slot_len_for_id(id, "relationship")?;
        if self.relationships.len() < target {
            self.relationships
                .try_reserve(target - self.relationships.len())
                .map_err(|e| {
                    format!(
                        "relationship id {id} requires {target} slots, but allocation failed: {e}"
                    )
                })?;
            self.relationships.grow_to(target);
        }
        Ok(target - 1)
    }

    pub(super) fn put_node_checked(&mut self, id: NodeId, node: &NodeRecord) -> Result<(), String> {
        let idx = self.ensure_node_slot_checked(id)?;
        self.put_node_at_slot(idx, node);
        Ok(())
    }

    pub(super) fn put_rel_checked(
        &mut self,
        id: RelationshipId,
        rel: &RelationshipRecord,
    ) -> Result<(), String> {
        let idx = self.ensure_rel_slot_checked(id)?;
        self.put_rel_at_slot(idx, rel);
        Ok(())
    }

    /// Store already encoded node records (labels and keys numbered by
    /// this graph's dictionaries), built in leaves. Fails on a taken id.
    pub(super) fn put_encoded_node_leaves(&mut self, built: BuiltLeaves) -> Result<(), String> {
        let Some(last) = built.last_slot() else {
            return Ok(());
        };
        self.ensure_node_slot_checked(last as NodeId)?;
        self.live_node_count += self
            .nodes
            .put_leaves(built)
            .map_err(|slot| format!("duplicate node id {slot} in snapshot payload"))?;
        Ok(())
    }

    /// Store already encoded relationship records, built in leaves; `links`
    /// holds each one's `(id, source, target, type number)`, whose
    /// endpoints must exist. Fails on a taken id or a missing endpoint.
    /// Linking them into the adjacency is [`Self::link_relationships`].
    pub(super) fn put_encoded_relationship_leaves(
        &mut self,
        built: BuiltLeaves,
        links: &[(RelationshipId, NodeId, NodeId, u32)],
    ) -> Result<(), String> {
        let Some(last) = built.last_slot() else {
            return Ok(());
        };
        for &(id, src, dst, _) in links {
            if !self.has_node_at(src) {
                return Err(format!(
                    "relationship {id} references missing source node {src}"
                ));
            }
            if !self.has_node_at(dst) {
                return Err(format!(
                    "relationship {id} references missing target node {dst}"
                ));
            }
        }
        self.ensure_rel_slot_checked(last as RelationshipId)?;
        self.live_rel_count += self
            .relationships
            .put_leaves(built)
            .map_err(|slot| format!("duplicate relationship id {slot} in snapshot payload"))?;
        Ok(())
    }

    /// Link each `(id, source, target, type number)` of `links` (in id
    /// order, endpoints present) into its endpoints' adjacency, as one
    /// [`Self::adjacency_push`] per end would. On every core when the build
    /// allows: the ends are sorted by adjacency leaf (stably, so each
    /// node's list gets them in `links` order) and each leaf is filled on
    /// its own thread.
    pub(super) fn link_relationships(&mut self, links: &[(RelationshipId, NodeId, NodeId, u32)]) {
        #[cfg(all(feature = "parallel", not(target_arch = "wasm32")))]
        {
            use rayon::prelude::*;
            let mut ends: Vec<(NodeId, Side, AdjEntry)> = Vec::with_capacity(2 * links.len());
            for &(id, src, dst, type_id) in links {
                let out = AdjEntry {
                    type_id,
                    neighbour: dst,
                    rel: id,
                };
                ends.push((src, Side::Out, out));
                let into = AdjEntry {
                    type_id,
                    neighbour: src,
                    rel: id,
                };
                ends.push((dst, Side::In, into));
            }
            const LEAF_SHIFT: u32 = super::chunked_vec::CHUNK.trailing_zeros();
            const SLOT_MASK: usize = super::chunked_vec::CHUNK - 1;
            let leaf_of = |node: NodeId| (node as usize) >> LEAF_SHIFT;
            ends.par_sort_by_key(|(node, _, _)| leaf_of(*node));
            let groups: Vec<&[(NodeId, Side, AdjEntry)]> =
                ends.chunk_by(|a, b| leaf_of(a.0) == leaf_of(b.0)).collect();
            let touched: Vec<usize> = groups.iter().map(|group| leaf_of(group[0].0)).collect();
            self.adjacency
                .leaves_mut(&touched)
                .into_par_iter()
                .zip(groups)
                .for_each(|((_, leaf), group)| {
                    for &(node, side, entry) in group {
                        leaf[node as usize & SLOT_MASK].push(side, entry);
                    }
                });
        }
        #[cfg(not(all(feature = "parallel", not(target_arch = "wasm32"))))]
        for &(id, src, dst, type_id) in links {
            let out = AdjEntry {
                type_id,
                neighbour: dst,
                rel: id,
            };
            self.adjacency_push(src, Side::Out, out);
            let into = AdjEntry {
                type_id,
                neighbour: src,
                rel: id,
            };
            self.adjacency_push(dst, Side::In, into);
        }
    }

    /// List `ids` under the label numbered `label_id`, as
    /// [`Self::insert_node_label_index`] would one by one.
    pub(super) fn extend_node_label_index(&mut self, label_id: u32, ids: &[NodeId]) {
        let label = self.dicts.labels.name(label_id).as_arc();
        let bucket = match self.nodes_by_label.get_mut(&**label) {
            Some(bucket) => bucket,
            None => self.nodes_by_label.entry(Arc::clone(label)).or_default(),
        };
        for &id in ids {
            bucket.insert(id);
        }
    }

    /// List `ids` under the relationship type numbered `type_id`.
    pub(super) fn extend_relationship_type_index(&mut self, type_id: u32, ids: &[RelationshipId]) {
        let rel_type = self.dicts.types.name(type_id).as_arc();
        let bucket = match self.relationships_by_type.get_mut(&**rel_type) {
            Some(bucket) => bucket,
            None => self
                .relationships_by_type
                .entry(Arc::clone(rel_type))
                .or_default(),
        };
        for &id in ids {
            bucket.insert(id);
        }
    }

    pub(super) fn put_node_at_slot(&mut self, idx: usize, node: &NodeRecord) {
        if self.store_node(idx, node) {
            self.live_node_count += 1;
        }
    }

    pub(super) fn put_rel_at_slot(&mut self, idx: usize, rel: &RelationshipRecord) {
        let mut out = Scratch::take();
        encoded::encode_rel(rel, &mut self.dicts, &mut out);
        if self.relationships.set(idx, &out) {
            self.live_rel_count += 1;
        }
    }

    /// Encode `node` into slot `idx`; returns whether the slot was empty.
    #[inline]
    fn store_node(&mut self, idx: usize, node: &NodeRecord) -> bool {
        let mut out = Scratch::take();
        encoded::encode_node(node, &mut self.dicts, &mut out);
        self.nodes.set(idx, &out)
    }

    pub(super) fn take_node(&mut self, id: NodeId) -> Option<NodeRecord> {
        let idx = Self::slot_index(id)?;
        let removed = StoredNode::new(id, self.nodes.get(idx)?, &self.dicts).to_record();
        self.nodes.remove(idx);
        self.live_node_count -= 1;
        // Also clear the per-id adjacency entries so the memory is reclaimed
        // on the typical "delete every node" pattern. We deliberately do not
        // shrink the slots — leaving the slot lets new ids reuse the same
        // index without growth churn (and `next_node_id` is monotonic
        // anyway, so no immediate reuse).
        self.clear_adjacency(id);
        Some(removed)
    }

    pub(super) fn take_rel(&mut self, id: RelationshipId) -> Option<RelationshipRecord> {
        let idx = Self::slot_index(id)?;
        let removed = StoredRel::new(id, self.relationships.get(idx)?, &self.dicts).to_record();
        self.relationships.remove(idx);
        self.live_rel_count -= 1;
        Some(removed)
    }

    #[inline]
    pub(super) fn outgoing_at(&self, id: NodeId) -> Option<AdjSide<'_>> {
        Some(self.adjacency.get(Self::slot_index(id)?)?.outgoing())
    }

    #[inline]
    pub(super) fn incoming_at(&self, id: NodeId) -> Option<AdjSide<'_>> {
        Some(self.adjacency.get(Self::slot_index(id)?)?.incoming())
    }

    /// Visit the entries of one adjacency list that pass `filter`.
    /// `skip_self_loops` is set for the incoming half of an undirected
    /// walk, whose outgoing half already reported the node's self-loops.
    #[inline]
    fn try_for_each_adjacent_entry<F, E>(
        node_id: NodeId,
        filter: &TypeFilter,
        adj: AdjSide<'_>,
        skip_self_loops: bool,
        visit: &mut F,
    ) -> Result<(), E>
    where
        F: FnMut(RelationshipId, NodeId) -> Result<(), E>,
    {
        for entry in adj.iter() {
            if skip_self_loops && entry.neighbour == node_id {
                continue;
            }
            if !filter.matches(entry.type_id) {
                continue;
            }
            visit(entry.rel, entry.neighbour)?;
        }
        Ok(())
    }

    #[inline]
    pub(super) fn try_for_each_adjacent_id_unchecked<F, E>(
        &self,
        node_id: NodeId,
        direction: Direction,
        types: &[String],
        mut visit: F,
    ) -> Result<(), E>
    where
        F: FnMut(RelationshipId, NodeId) -> Result<(), E>,
    {
        let filter = TypeFilter::resolve(&self.dicts.types, types);
        if matches!(filter, TypeFilter::Nothing) {
            return Ok(());
        }
        if !matches!(direction, Direction::Left) {
            if let Some(adj) = self.outgoing_at(node_id) {
                Self::try_for_each_adjacent_entry(node_id, &filter, adj, false, &mut visit)?;
            }
        }
        if !matches!(direction, Direction::Right) {
            if let Some(adj) = self.incoming_at(node_id) {
                let skip_self_loops = matches!(direction, Direction::Undirected);
                Self::try_for_each_adjacent_entry(
                    node_id,
                    &filter,
                    adj,
                    skip_self_loops,
                    &mut visit,
                )?;
            }
        }

        Ok(())
    }

    #[inline]
    pub(super) fn try_for_each_adjacent_id<F, E>(
        &self,
        node_id: NodeId,
        direction: Direction,
        types: &[String],
        visit: F,
    ) -> Result<(), E>
    where
        F: FnMut(RelationshipId, NodeId) -> Result<(), E>,
    {
        if !self.has_node_at(node_id) {
            return Ok(());
        }
        self.try_for_each_adjacent_id_unchecked(node_id, direction, types, visit)
    }

    pub(super) fn iter_node_ids(&self) -> impl Iterator<Item = NodeId> + '_ {
        self.nodes.slots_from(0).map(|i| i as NodeId)
    }

    pub(super) fn iter_node_refs(&self) -> impl Iterator<Item = NodeRef<'_>> + '_ {
        self.iter_nodes().map(|(_, node)| node)
    }

    pub(super) fn iter_rel_ids(&self) -> impl Iterator<Item = RelationshipId> + '_ {
        self.relationships
            .slots_from(0)
            .map(|i| i as RelationshipId)
    }

    pub(super) fn iter_rel_refs(&self) -> impl Iterator<Item = RelRef<'_>> + '_ {
        self.iter_rels().map(|(_, rel)| rel)
    }

    pub(super) fn iter_nodes(&self) -> impl Iterator<Item = (NodeId, NodeRef<'_>)> + '_ {
        let dicts = &self.dicts;
        self.nodes.iter().map(move |(i, bytes)| {
            let id = i as NodeId;
            (id, NodeRef::stored(StoredNode::new(id, bytes, dicts)))
        })
    }

    pub(super) fn iter_rels(&self) -> impl Iterator<Item = (RelationshipId, RelRef<'_>)> + '_ {
        let dicts = &self.dicts;
        self.relationships.iter().map(move |(i, bytes)| {
            let id = i as RelationshipId;
            (id, RelRef::stored(StoredRel::new(id, bytes, dicts)))
        })
    }

    /// Add an entry to `node_id`'s list on `side`. Relies on
    /// the monotonic-id invariant: relationship ids are allocated once and
    /// never re-used, so the list can never see a duplicate.
    fn adjacency_push(&mut self, node_id: NodeId, side: Side, entry: AdjEntry) {
        if let Ok(idx) = self.ensure_node_slot_checked(node_id) {
            self.adjacency[idx].push(side, entry);
        }
    }

    /// Drop both of `node_id`'s lists at once. A detach delete does this
    /// first, so removing each relationship touches only the far end's list
    /// instead of scanning and compacting the node's own list once per
    /// relationship (quadratic in its degree).
    pub(super) fn clear_adjacency(&mut self, node_id: NodeId) {
        let Some(idx) = Self::slot_index(node_id) else {
            return;
        };
        // An empty list needs no clearing, and reaching it mutably would
        // copy its shared leaf for nothing.
        if self.adjacency.get(idx).is_some_and(|adj| !adj.is_empty()) {
            if let Some(adj) = self.adjacency.get_mut(idx) {
                adj.clear();
            }
        }
    }

    /// Remove `rel_id` from `node_id`'s list on `side`. With `pending`, a
    /// list longer than [`DEFER_REMOVAL_MIN_BYTES`] is left for
    /// [`Self::flush_adjacency_removals`], which takes all of a list's
    /// pending entries in one pass: one at a time, each removal scans and
    /// compacts the list, which is quadratic when a statement deletes
    /// many relationships of one hub. A short list is edited at once.
    fn adjacency_remove(
        &mut self,
        node_id: NodeId,
        side: Side,
        rel_id: RelationshipId,
        pending: Option<&mut PendingAdjRemovals>,
    ) {
        let Some(idx) = Self::slot_index(node_id) else {
            return;
        };
        let len = self
            .adjacency
            .get(idx)
            .map_or(0, |adj| adj.side_bytes(side));
        // An empty list (a node detach delete already cleared) needs no
        // edit, and reaching it mutably would copy its shared leaf.
        if len == 0 {
            return;
        }
        match pending {
            Some(pending) if len > DEFER_REMOVAL_MIN_BYTES => {
                pending.push((node_id, side, rel_id));
            }
            _ => {
                if let Some(adj) = self.adjacency.get_mut(idx) {
                    adj.remove(side, rel_id);
                }
            }
        }
    }

    /// Apply the removals [`Self::adjacency_remove`] deferred: each list
    /// once, keeping the order of its other entries. Leaves `pending` empty.
    pub(super) fn flush_adjacency_removals(&mut self, pending: &mut PendingAdjRemovals) {
        pending.sort_unstable_by_key(|&(node, side, rel)| (node, side == Side::In, rel));
        let mut rels = Vec::new();
        for group in pending.chunk_by(|a, b| a.0 == b.0 && a.1 == b.1) {
            let (node_id, side, _) = group[0];
            let Some(idx) = Self::slot_index(node_id) else {
                continue;
            };
            if self
                .adjacency
                .get(idx)
                .is_none_or(|adj| adj.side_bytes(side) == 0)
            {
                continue;
            }
            rels.clear();
            rels.extend(group.iter().map(|&(_, _, rel)| rel));
            if let Some(adj) = self.adjacency.get_mut(idx) {
                adj.remove_all(side, &rels);
            }
        }
        pending.clear();
    }

    pub(super) fn normalize_labels(labels: Vec<String>) -> Labels {
        let mut out = Labels::new();
        for label in &labels {
            let label = label.trim();
            if !label.is_empty() && !out.has(label) {
                out.push(label);
            }
        }
        out
    }

    pub(super) fn insert_node_label_index(&mut self, node_id: NodeId, label: &str) {
        // Hot path: skip the `String` alloc when the label bucket already
        // exists. The monotonic-id invariant on the create path guarantees
        // `node_id` is unique, so we push unconditionally; the previous
        // `contains` guard turned bulk CREATE into O(n²).
        if let Some(bucket) = self.nodes_by_label.get_mut(label) {
            bucket.insert(node_id);
        } else {
            self.nodes_by_label
                .insert(Arc::from(label), std::iter::once(node_id).collect());
        }
    }

    fn remove_node_label_index(&mut self, node_id: NodeId, label: &str) {
        if let Some(ids) = self.nodes_by_label.get_mut(label) {
            ids.remove(node_id);
            if ids.is_empty() {
                self.nodes_by_label.remove(label);
            }
        }
    }

    fn insert_relationship_type_index(&mut self, rel_id: RelationshipId, rel_type: &str) {
        // See `insert_node_label_index` for the same hot-path rationale.
        if let Some(bucket) = self.relationships_by_type.get_mut(rel_type) {
            bucket.insert(rel_id);
        } else {
            self.relationships_by_type
                .insert(Arc::from(rel_type), std::iter::once(rel_id).collect());
        }
    }

    fn remove_relationship_type_index(&mut self, rel_id: RelationshipId, rel_type: &str) {
        if let Some(ids) = self.relationships_by_type.get_mut(rel_type) {
            ids.remove(rel_id);
            if ids.is_empty() {
                self.relationships_by_type.remove(rel_type);
            }
        }
    }

    /// Ids of `label` nodes whose `key` equals `value`, from the scoped
    /// hash index. `None` when that index is not active for `key` or
    /// `value` has no index image; the caller must then scan.
    pub(super) fn indexed_node_ids(
        &self,
        label: &str,
        key: &str,
        value: &PropertyValue,
    ) -> Option<Vec<NodeId>> {
        PropertyIndexKey::from_value(value)?;
        let indexes = self.indexes_read();
        if !indexes.node_properties.is_active(key) {
            return None;
        }
        Some(
            indexes
                .node_properties
                .scoped_ids_for(label, key, value, &self.node_values())
                .map(|ids| ids.to_vec())
                .unwrap_or_default(),
        )
    }

    /// Relationship counterpart of [`Self::indexed_node_ids`].
    pub(super) fn indexed_rel_ids(
        &self,
        rel_type: &str,
        key: &str,
        value: &PropertyValue,
    ) -> Option<Vec<RelationshipId>> {
        PropertyIndexKey::from_value(value)?;
        let indexes = self.indexes_read();
        if !indexes.relationship_properties.is_active(key) {
            return None;
        }
        Some(
            indexes
                .relationship_properties
                .scoped_ids_for(rel_type, key, value, &self.relationship_values())
                .map(|ids| ids.to_vec())
                .unwrap_or_default(),
        )
    }

    pub(super) fn indexes_read(&self) -> std::sync::RwLockReadGuard<'_, PropertyIndexRegistry> {
        self.indexes
            .properties
            .read()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
    }

    pub(super) fn indexes_write(&self) -> RwLockWriteGuard<'_, PropertyIndexRegistry> {
        self.indexes
            .properties
            .write()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
    }

    /// Reads node property values from their records, for the hash
    /// property index (see [`RecordValues`]).
    #[inline]
    pub(super) fn node_values(&self) -> RecordValues<'_> {
        RecordValues::nodes(&self.nodes, &self.dicts)
    }

    /// Relationship counterpart of [`Self::node_values`].
    #[inline]
    pub(super) fn relationship_values(&self) -> RecordValues<'_> {
        RecordValues::relationships(&self.relationships, &self.dicts)
    }

    /// [`Self::indexes_mut`] with [`Self::node_values`], borrowed together.
    fn node_index_parts(&mut self) -> (&mut PropertyIndexRegistry, RecordValues<'_>) {
        let registry = self
            .indexes
            .properties
            .get_mut()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        (registry, RecordValues::nodes(&self.nodes, &self.dicts))
    }

    /// [`Self::indexes_mut`] with [`Self::relationship_values`].
    fn relationship_index_parts(&mut self) -> (&mut PropertyIndexRegistry, RecordValues<'_>) {
        let registry = self
            .indexes
            .properties
            .get_mut()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        (
            registry,
            RecordValues::relationships(&self.relationships, &self.dicts),
        )
    }

    pub(super) fn indexes_mut(&mut self) -> &mut PropertyIndexRegistry {
        self.indexes
            .properties
            .get_mut()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
    }

    #[inline]
    pub(super) fn active_node_property_index_count(&self) -> usize {
        self.indexes
            .active_node_property_indexes
            .load(Ordering::Relaxed)
    }

    #[inline]
    pub(super) fn active_relationship_property_index_count(&self) -> usize {
        self.indexes
            .active_relationship_property_indexes
            .load(Ordering::Relaxed)
    }

    #[inline]
    pub(super) fn active_constraint_count(&self) -> usize {
        self.active_constraints.load(Ordering::Relaxed)
    }

    #[inline]
    pub(super) fn has_active_constraints(&self) -> bool {
        self.active_constraint_count() != 0
    }

    #[inline]
    pub(super) fn active_fulltext_index_count(&self) -> usize {
        self.indexes.active_fulltext_indexes.load(Ordering::Relaxed)
    }

    #[inline]
    pub(super) fn has_active_fulltext_indexes(&self) -> bool {
        self.active_fulltext_index_count() != 0
    }

    pub(super) fn node_property_index_is_active(&mut self, key: &str) -> bool {
        self.active_node_property_index_count() != 0
            && self.indexes_mut().node_properties.is_active(key)
    }

    pub(super) fn relationship_property_index_is_active(&mut self, key: &str) -> bool {
        self.active_relationship_property_index_count() != 0
            && self.indexes_mut().relationship_properties.is_active(key)
    }

    /// Give `key` its across-labels map, for lookups that name no label.
    /// Call after [`Self::ensure_node_property_index`].
    pub(super) fn ensure_any_scope_node_property_index(&self, key: &str) {
        if self.indexes_read().node_properties.any_scope_is_active(key) {
            return;
        }
        self.indexes_write()
            .node_properties
            .activate_any_scope(key, &self.node_values());
    }

    pub(super) fn ensure_any_scope_relationship_property_index(&self, key: &str) {
        if self
            .indexes_read()
            .relationship_properties
            .any_scope_is_active(key)
        {
            return;
        }
        self.indexes_write()
            .relationship_properties
            .activate_any_scope(key, &self.relationship_values());
    }

    /// For a lookup: activate the node property index on `key` unless
    /// another lookup is building it, in which case this one should scan.
    /// Returns whether the index is active.
    pub(super) fn try_ensure_node_property_index(&self, key: &str) -> bool {
        if self.indexes_read().node_properties.is_active(key) {
            return true;
        }
        let Some(_build) = self.indexes.property_builds.claim(true, key) else {
            return false;
        };
        self.ensure_node_property_index(key);
        true
    }

    /// As [`Self::try_ensure_node_property_index`], for relationships.
    pub(super) fn try_ensure_relationship_property_index(&self, key: &str) -> bool {
        if self.indexes_read().relationship_properties.is_active(key) {
            return true;
        }
        let Some(_build) = self.indexes.property_builds.claim(false, key) else {
            return false;
        };
        self.ensure_relationship_property_index(key);
        true
    }

    /// Built without the property index lock, which is then held only to
    /// file the result: other indexes build meanwhile (a restart builds
    /// them all at once), and lookups of other keys are not held up.
    /// Lookups go through [`Self::try_ensure_node_property_index`], so two
    /// of them do not build the same index at once.
    pub(super) fn ensure_node_property_index(&self, key: &str) {
        if self.indexes_read().node_properties.is_active(key) {
            return;
        }

        let built = if self.live_node_count >= PARALLEL_INDEX_MIN {
            self.build_node_property_index_parallel(key)
        } else {
            let mut built = PropertyIndexState::default();
            built.insert_bulk(
                key,
                || {
                    self.iter_nodes().filter_map(|(id, node)| {
                        let value = node.properties().get(key)?;
                        Some((id, node.labels().strs(), value))
                    })
                },
                &self.node_values(),
            );
            built
        };

        let mut indexes = self.indexes_write();
        if indexes.node_properties.is_active(key) {
            return;
        }
        indexes.node_properties.absorb(key, built);
        if indexes.node_properties.activate(key) {
            self.indexes
                .active_node_property_indexes
                .fetch_add(1, Ordering::Relaxed);
        }
    }

    /// [`PropertyIndexState::build_parallel`] over the nodes.
    #[cfg(all(feature = "parallel", not(target_arch = "wasm32")))]
    pub(super) fn build_node_property_index_parallel(&self, key: &str) -> PropertyIndexState {
        let Some(key_id) = self.dicts.keys.id_of(key) else {
            return PropertyIndexState::default();
        };
        PropertyIndexState::build_parallel(
            key,
            self.nodes.len() as u64,
            |ids| self.node_slots_in(ids),
            |id| {
                let node = StoredNode::new(id, self.nodes.get(id as usize)?, &self.dicts);
                let value = node.properties().get_id(key_id)?;
                Some((node.labels().iter().map(crate::Name::as_str), value))
            },
            &self.node_values(),
        )
    }

    #[cfg(not(all(feature = "parallel", not(target_arch = "wasm32"))))]
    pub(super) fn build_node_property_index_parallel(&self, key: &str) -> PropertyIndexState {
        let mut built = PropertyIndexState::default();
        built.insert_bulk(
            key,
            || {
                self.iter_nodes().filter_map(|(id, node)| {
                    let value = node.properties().get(key)?;
                    Some((id, node.labels().strs(), value))
                })
            },
            &self.node_values(),
        );
        built
    }

    #[cfg(all(feature = "parallel", not(target_arch = "wasm32")))]
    fn build_relationship_property_index_parallel(&self, key: &str) -> PropertyIndexState {
        let Some(key_id) = self.dicts.keys.id_of(key) else {
            return PropertyIndexState::default();
        };
        PropertyIndexState::build_parallel(
            key,
            self.relationships.len() as u64,
            |ids| self.rel_slots_in(ids),
            |id| {
                let rel = StoredRel::new(id, self.relationships.get(id as usize)?, &self.dicts);
                let value = rel.properties().get_id(key_id)?;
                Some(([rel.rel_type().as_str()], value))
            },
            &self.relationship_values(),
        )
    }

    #[cfg(not(all(feature = "parallel", not(target_arch = "wasm32"))))]
    fn build_relationship_property_index_parallel(&self, key: &str) -> PropertyIndexState {
        let mut built = PropertyIndexState::default();
        built.insert_bulk(
            key,
            || {
                self.iter_rels().filter_map(|(id, rel)| {
                    let value = rel.properties().get(key)?;
                    Some((id, [rel.rel_type()], value))
                })
            },
            &self.relationship_values(),
        );
        built
    }

    /// The node slots holding a record, among `ids`.
    #[cfg(all(feature = "parallel", not(target_arch = "wasm32")))]
    fn node_slots_in(&self, ids: std::ops::Range<u64>) -> impl Iterator<Item = NodeId> + '_ {
        let end = ids.end;
        self.nodes
            .slots_from(usize::try_from(ids.start).unwrap_or(usize::MAX))
            .map(|i| i as NodeId)
            .take_while(move |id| *id < end)
    }

    /// The relationship slots holding a record, among `ids`.
    #[cfg(all(feature = "parallel", not(target_arch = "wasm32")))]
    fn rel_slots_in(&self, ids: std::ops::Range<u64>) -> impl Iterator<Item = RelationshipId> + '_ {
        let end = ids.end;
        self.relationships
            .slots_from(usize::try_from(ids.start).unwrap_or(usize::MAX))
            .map(|i| i as RelationshipId)
            .take_while(move |id| *id < end)
    }

    /// As [`Self::ensure_node_property_index`].
    pub(super) fn ensure_relationship_property_index(&self, key: &str) {
        if self.indexes_read().relationship_properties.is_active(key) {
            return;
        }

        let built = if self.live_rel_count >= PARALLEL_INDEX_MIN {
            self.build_relationship_property_index_parallel(key)
        } else {
            let mut built = PropertyIndexState::default();
            built.insert_bulk(
                key,
                || {
                    self.iter_rels().filter_map(|(id, rel)| {
                        let value = rel.properties().get(key)?;
                        Some((id, [rel.rel_type()], value))
                    })
                },
                &self.relationship_values(),
            );
            built
        };

        let mut indexes = self.indexes_write();
        if indexes.relationship_properties.is_active(key) {
            return;
        }
        indexes.relationship_properties.absorb(key, built);
        if indexes.relationship_properties.activate(key) {
            self.indexes
                .active_relationship_property_indexes
                .fetch_add(1, Ordering::Relaxed);
        }
    }

    pub(super) fn index_catalog_read(&self) -> IndexRead<'_, IndexCatalog> {
        read_shared(&self.indexes.catalog)
    }

    /// Mutable access copies the catalog first if a clone still shares it.
    pub(super) fn index_catalog_write(&self) -> IndexWrite<'_, IndexCatalog> {
        write_shared(&self.indexes.catalog)
    }

    pub(super) fn constraint_catalog_read(&self) -> IndexRead<'_, ConstraintCatalog> {
        read_shared(&self.constraint_catalog)
    }

    /// Mutable access copies the catalog first if a clone still shares it.
    pub(super) fn constraint_catalog_write(&self) -> IndexWrite<'_, ConstraintCatalog> {
        write_shared(&self.constraint_catalog)
    }

    /// Register an explicitly-declared index in the catalog and, when
    /// applicable, force the underlying property-index buckets to be
    /// populated so equality lookups can use them immediately.
    ///
    /// Named with a `register_` prefix to avoid colliding with the
    /// trait method `GraphStorageMut::create_index` — the trait impl
    /// in `impls.rs` delegates here.
    #[allow(clippy::result_large_err)]
    pub(super) fn register_index(
        &self,
        request: IndexRequest,
        if_not_exists: bool,
    ) -> Result<CreateIndexOutcome, CreateIndexError> {
        self.register_index_with_recording(request, if_not_exists, true)
    }

    #[allow(clippy::result_large_err)]
    fn register_index_with_recording(
        &self,
        request: IndexRequest,
        if_not_exists: bool,
        record_event: bool,
    ) -> Result<CreateIndexOutcome, CreateIndexError> {
        self.register_index_deferring(request, if_not_exists, record_event, None)
    }

    /// [`Self::register_index_with_recording`]; with `deferred`, a new
    /// index's data is left in it for [`Self::run_deferred`] instead of
    /// populated now. Vector indexes populate now: a restore overlays
    /// their saved graph on them right after.
    #[allow(clippy::result_large_err)]
    pub(super) fn register_index_deferring(
        &self,
        request: IndexRequest,
        if_not_exists: bool,
        record_event: bool,
        deferred: Option<&mut DeferredData>,
    ) -> Result<CreateIndexOutcome, CreateIndexError> {
        let request_for_event = record_event.then(|| request.clone());
        let outcome = {
            let mut catalog = self.index_catalog_write();
            catalog.try_create(request, if_not_exists)?
        };

        if let CreateIndexOutcome::Created(def) = &outcome {
            match deferred {
                Some(deferred) if def.kind != StoredIndexKind::Vector => {
                    deferred.populate.push(def.clone());
                }
                _ => self.populate_index_data(def),
            }
        }

        // Both Created and NoOpExists are committed catalog states; we
        // log only Created because NoOpExists implies a redundant DDL
        // that adds nothing to durable state.
        if matches!(outcome, CreateIndexOutcome::Created(_)) {
            if let Some(request_for_event) = request_for_event {
                self.emit(|| crate::MutationEvent::CreateIndex {
                    request: request_for_event,
                    if_not_exists,
                });
            }
        }

        Ok(outcome)
    }

    /// Replay a CreateIndex event against an empty graph. Mirrors the
    /// `replay_create_node` shape: callers must invoke before installing
    /// a recorder so we don't re-emit during recovery.
    #[doc(hidden)]
    pub fn replay_create_index(
        &mut self,
        request: IndexRequest,
        if_not_exists: bool,
    ) -> Result<(), String> {
        if self.recorder.is_some() {
            return Err("cannot replay create_index while a mutation recorder is installed".into());
        }
        self.register_index(request, if_not_exists)
            .map(|_| ())
            .map_err(|e| e.to_string())
    }

    /// Replay a DropIndex event.
    #[doc(hidden)]
    pub fn replay_drop_index(&mut self, name: &str, if_exists: bool) -> Result<(), String> {
        if self.recorder.is_some() {
            return Err("cannot replay drop_index while a mutation recorder is installed".into());
        }
        self.drop_named_index(name, if_exists)
            .map(|_| ())
            .map_err(|e| e.to_string())
    }

    /// Register a constraint. For uniqueness/key kinds this also
    /// registers a backing RANGE index in the index catalog under the
    /// same name. Validation of existing data is the caller's
    /// responsibility (the enforcement layer runs a pre-create scan
    /// before this method commits).
    pub(super) fn register_constraint(
        &self,
        request: ConstraintRequest,
        if_not_exists: bool,
    ) -> Result<CreateConstraintOutcome, CreateConstraintError> {
        self.register_constraint_deferring(request, if_not_exists, None)
    }

    /// [`Self::register_constraint`]; with `deferred`, the scan of the
    /// existing data and the backing index's data are left in it for
    /// [`Self::run_deferred`].
    pub(super) fn register_constraint_deferring(
        &self,
        request: ConstraintRequest,
        if_not_exists: bool,
        mut deferred: Option<&mut DeferredData>,
    ) -> Result<CreateConstraintOutcome, CreateConstraintError> {
        // Constraint-level conflicts (22N65/66/67) take precedence over
        // index-catalog conflicts: if the request collides with an
        // existing *constraint* shape or name, we never get to the
        // backing-index step.
        {
            let constraint_catalog = self.constraint_catalog_read();
            if let Some(existing) = constraint_catalog.find_equivalent(&request) {
                let cloned = existing.clone();
                drop(constraint_catalog);
                if if_not_exists {
                    return Ok(CreateConstraintOutcome::NoOpExists(cloned));
                }
                return Err(CreateConstraintError::EquivalentConstraintExists(
                    cloned.name,
                ));
            }
            if let Some(existing) = constraint_catalog.get(&request.name) {
                let cloned = existing.clone();
                drop(constraint_catalog);
                if if_not_exists {
                    return Ok(CreateConstraintOutcome::NoOpExists(cloned));
                }
                return Err(CreateConstraintError::DuplicateName(cloned.name));
            }
            if let Some(existing) = constraint_catalog.find_same_schema(&request) {
                let cloned = existing.clone();
                drop(constraint_catalog);
                if super::constraint_catalog::kinds_conflict_for_validation(
                    &cloned.kind,
                    &request.kind,
                ) {
                    if if_not_exists {
                        return Ok(CreateConstraintOutcome::NoOpExists(cloned));
                    }
                    return Err(CreateConstraintError::ConflictingConstraint(cloned.name));
                }
            }
        }

        // Index-catalog conflicts only matter for constraints that need
        // a backing range index. The catalog won't yet own one for this
        // request — that registration happens below — so any existing
        // entry under the same name or schema is from a foreign index.
        if request.kind.requires_backing_index() {
            let idx_catalog = self.index_catalog_read();
            if idx_catalog.get(&request.name).is_some() {
                return Err(CreateConstraintError::DuplicateIndexName(
                    request.name.clone(),
                ));
            }
            let conflict = idx_catalog.list().into_iter().find(|def| {
                def.kind == StoredIndexKind::Range
                    && def.entity == request.entity
                    && def.label.as_deref() == Some(request.label.as_str())
                    && def.properties == request.properties
                    && def.name != request.name
            });
            drop(idx_catalog);
            if let Some(def) = conflict {
                return Err(CreateConstraintError::BackingIndexConflict(format!(
                    "(:{} {{{}}}) already covered by index `{}`",
                    request.label,
                    request.properties.join(", "),
                    def.name,
                )));
            }
        }

        let owns_backing = request.kind.requires_backing_index();
        let request_for_event = request.clone();
        let outcome = {
            let mut catalog = self.constraint_catalog_write();
            catalog.try_create(request, if_not_exists)?
        };

        if let CreateConstraintOutcome::Created(def) = &outcome {
            // Pre-create data scan: if the live graph already violates
            // the constraint, fail and roll back the catalog write.
            // Creating constraints when conflicting data exists fails
            // before the catalog change is retained (22N77/79/80).
            if let Some(deferred) = deferred.as_deref_mut() {
                deferred.validate.push(def.clone());
            } else if let Err(violation) = self.validate_existing_data_for_constraint(def) {
                let mut catalog = self.constraint_catalog_write();
                let _ = catalog.try_drop(&def.name, true);
                return Err(CreateConstraintError::DataViolation(violation.to_string()));
            }
        }

        if let CreateConstraintOutcome::Created(def) = &outcome {
            if owns_backing {
                // Register a backing RANGE index under the same name. This
                // is an implementation detail of the constraint, so WAL and
                // snapshot replay record only the constraint mutation.
                // A hash-backed constraint keeps the hash index only.
                let mut options = BTreeMap::new();
                if def.backing == ConstraintBacking::Hash {
                    options.insert(SORTED_OPTION.to_string(), IndexConfigValue::Bool(false));
                }
                let idx_request = IndexRequest {
                    explicit_name: Some(def.name.clone()),
                    kind: StoredIndexKind::Range,
                    entity: def.entity,
                    label: Some(def.label.clone()),
                    additional_labels: Vec::new(),
                    properties: def.properties.clone(),
                    options,
                };
                // Errors from the backing-index registration unwind the
                // constraint registration to keep the two catalogs in
                // step.
                if let Err(err) = self.register_index_deferring(idx_request, true, false, deferred)
                {
                    let mut catalog = self.constraint_catalog_write();
                    let _ = catalog.try_drop(&def.name, true);
                    return Err(CreateConstraintError::BackingIndexConflict(err.to_string()));
                }
            }
            self.emit(|| crate::MutationEvent::CreateConstraint {
                request: request_for_event,
                if_not_exists,
            });
            self.active_constraints.fetch_add(1, Ordering::Relaxed);
        }

        Ok(outcome)
    }

    /// Do the data work [`Self::register_index_deferring`] and
    /// [`Self::register_constraint_deferring`] left in `deferred`: every
    /// index population and constraint scan at once, on every core when
    /// the build allows. They only read the records, and each index files
    /// its result under its own lock. Fails with the first constraint (in
    /// registration order) the data violates.
    pub(super) fn run_deferred(&self, deferred: DeferredData) -> Result<(), String> {
        enum Work {
            Populate(IndexDefinition),
            /// A RANGE index's two parts, built apart: the hash index of a
            /// key, and the sorted index of a label's key.
            Hash(StoredIndexEntity, String),
            Sorted(StoredIndexEntity, String, String),
            Validate(super::ConstraintDefinition),
        }
        // Scans first: a constraint scan is as long as an index build, and
        // the last task to start bounds the wall time.
        let mut work: Vec<Work> = deferred.validate.into_iter().map(Work::Validate).collect();
        let mut hashed = std::collections::BTreeSet::new();
        for def in deferred.populate {
            if def.kind != StoredIndexKind::Range {
                work.push(Work::Populate(def));
                continue;
            }
            for key in &def.properties {
                // Indexes on one key under several labels share its hash index.
                if hashed.insert((def.entity == StoredIndexEntity::Node, key.clone())) {
                    work.push(Work::Hash(def.entity, key.clone()));
                }
                if let Some(label) = def.label.as_deref().filter(|_| is_sorted(&def)) {
                    work.push(Work::Sorted(def.entity, label.to_owned(), key.clone()));
                }
            }
        }
        let run = |work: Work| match work {
            Work::Populate(def) => {
                self.populate_index_data(&def);
                Ok(())
            }
            Work::Hash(entity, key) => {
                match entity {
                    StoredIndexEntity::Node => self.ensure_node_property_index(&key),
                    StoredIndexEntity::Relationship => {
                        self.ensure_relationship_property_index(&key)
                    }
                }
                Ok(())
            }
            Work::Sorted(entity, label, key) => {
                self.activate_sorted_scope(entity, &label, &key);
                Ok(())
            }
            Work::Validate(def) => {
                self.validate_existing_data_for_constraint(&def)
                    .map_err(|violation| {
                        let err = CreateConstraintError::DataViolation(violation.to_string());
                        format!("constraint `{}`: {err}", def.name)
                    })
            }
        };
        #[cfg(all(feature = "parallel", not(target_arch = "wasm32")))]
        let results: Vec<Result<(), String>> = {
            use rayon::prelude::*;
            work.into_par_iter().with_max_len(1).map(run).collect()
        };
        #[cfg(not(all(feature = "parallel", not(target_arch = "wasm32"))))]
        let results: Vec<Result<(), String>> = work.into_iter().map(run).collect();
        results.into_iter().collect()
    }

    /// Replay a CreateConstraint event against a recorder-detached graph.
    #[doc(hidden)]
    pub fn replay_create_constraint(
        &mut self,
        request: ConstraintRequest,
        if_not_exists: bool,
    ) -> Result<(), String> {
        if self.recorder.is_some() {
            return Err(
                "cannot replay create_constraint while a mutation recorder is installed".into(),
            );
        }
        self.register_constraint(request, if_not_exists)
            .map(|_| ())
            .map_err(|e| e.to_string())
    }

    /// Replay a DropConstraint event.
    #[doc(hidden)]
    pub fn replay_drop_constraint(&mut self, name: &str, if_exists: bool) -> Result<(), String> {
        if self.recorder.is_some() {
            return Err(
                "cannot replay drop_constraint while a mutation recorder is installed".into(),
            );
        }
        self.drop_named_constraint(name, if_exists)
            .map(|_| ())
            .map_err(|e| e.to_string())
    }

    /// Inverse of [`Self::register_constraint`]. Cascades to the backing
    /// range index when one is owned.
    pub(super) fn drop_named_constraint(
        &self,
        name: &str,
        if_exists: bool,
    ) -> Result<DropConstraintOutcome, DropConstraintError> {
        let outcome = {
            let mut catalog = self.constraint_catalog_write();
            catalog.try_drop(name, if_exists)?
        };
        if let DropConstraintOutcome::Dropped(def) = &outcome {
            if let Some(index_name) = def.owned_index.as_deref() {
                // The backing index is owned exclusively by the
                // constraint, so dropping it is unconditional.
                let _ = self.drop_named_index_inner(index_name, true, false);
            }
            self.active_constraints.fetch_sub(1, Ordering::Relaxed);
            self.emit(|| crate::MutationEvent::DropConstraint {
                name: name.to_string(),
                if_exists,
            });
        }
        Ok(outcome)
    }

    /// Inverse of [`Self::register_index`]. Removes the catalog entry
    /// and (for RANGE) releases the hash property index of each property
    /// no other RANGE index still declares; a later equality lookup
    /// activates it again.
    pub(super) fn drop_named_index(
        &self,
        name: &str,
        if_exists: bool,
    ) -> Result<DropIndexOutcome, DropIndexError> {
        self.drop_named_index_inner(name, if_exists, true)
    }

    fn drop_named_index_inner(
        &self,
        name: &str,
        if_exists: bool,
        emit_event: bool,
    ) -> Result<DropIndexOutcome, DropIndexError> {
        if let Some(owner) = self
            .constraint_catalog_read()
            .constraint_owning_index(name)
            .cloned()
        {
            return Err(DropIndexError::ConstraintOwned {
                index: name.to_string(),
                constraint: owner.name,
            });
        }

        let outcome = {
            let mut catalog = self.index_catalog_write();
            catalog.try_drop(name, if_exists)?
        };
        if let DropIndexOutcome::Dropped(def) = &outcome {
            // Release backing structures keyed off the dropped def.
            match def.kind {
                StoredIndexKind::Text => {
                    if let Some(label) = def.label.as_deref() {
                        for prop in &def.properties {
                            self.deactivate_text_scope(def.entity, label, prop);
                        }
                    }
                }
                StoredIndexKind::Range => {
                    if let Some(label) = def.label.as_deref().filter(|_| is_sorted(def)) {
                        for prop in &def.properties {
                            self.deactivate_sorted_scope(def.entity, label, prop);
                        }
                    }
                    // Writes stop paying for a hash index nothing declares
                    // any more; an equality lookup rebuilds it on demand.
                    let declared = self.declared_property_index_keys();
                    let mut indexes = self.indexes_write();
                    for prop in &def.properties {
                        if declared.contains(&(def.entity, prop.clone())) {
                            continue;
                        }
                        match def.entity {
                            StoredIndexEntity::Node => indexes.node_properties.deactivate(prop),
                            StoredIndexEntity::Relationship => {
                                indexes.relationship_properties.deactivate(prop)
                            }
                        };
                    }
                }
                StoredIndexKind::Point => {
                    if let Some(label) = def.label.as_deref() {
                        for prop in &def.properties {
                            self.deactivate_point_scope(def.entity, label, prop);
                        }
                    }
                }
                StoredIndexKind::Lookup => {
                    // Lookup rides on the eagerly-maintained label/type
                    // indexes; nothing to release.
                }
                StoredIndexKind::Vector => {
                    self.deactivate_vector_index(def.entity, &def.name);
                }
                StoredIndexKind::Fulltext => {
                    self.deactivate_fulltext_index(def.entity, &def.name);
                }
            }
            if emit_event {
                self.emit(|| crate::MutationEvent::DropIndex {
                    name: name.to_string(),
                    if_exists,
                });
            }
        }
        Ok(outcome)
    }

    /// `(entity, key)` pairs whose hash property index a declaration keeps
    /// active: every property of a RANGE index in the catalog, which
    /// includes the backing indexes of uniqueness / key constraints. These
    /// are the only hash indexes a restart rebuilds; any other active key
    /// was activated implicitly by an equality lookup.
    pub(super) fn declared_property_index_keys(
        &self,
    ) -> std::collections::BTreeSet<(StoredIndexEntity, String)> {
        self.index_catalog_read()
            .list()
            .into_iter()
            .filter(|def| def.kind == StoredIndexKind::Range)
            .flat_map(|def| {
                let entity = def.entity;
                def.properties.into_iter().map(move |p| (entity, p))
            })
            .collect()
    }

    fn populate_index_data(&self, def: &IndexDefinition) {
        // RANGE: piggy-back on the existing lazy property-index buckets.
        // TEXT: build a trigram inverted index over the existing entity
        //       data for the (label, property) tuple.
        // POINT: build a grid-bucket spatial index over the existing
        //        entity data.
        // LOOKUP: catalog-only; existing label/type indexes already
        //         answer the predicates.
        match def.kind {
            StoredIndexKind::Range => {
                for key in &def.properties {
                    match def.entity {
                        StoredIndexEntity::Node => self.ensure_node_property_index(key),
                        StoredIndexEntity::Relationship => {
                            self.ensure_relationship_property_index(key)
                        }
                    }
                    if let Some(label) = def.label.as_deref().filter(|_| is_sorted(def)) {
                        self.activate_sorted_scope(def.entity, label, key);
                    }
                }
            }
            StoredIndexKind::Text => {
                let label = match def.label.as_deref() {
                    Some(l) => l,
                    None => return,
                };
                for property in &def.properties {
                    self.activate_text_scope(def.entity, label, property);
                }
            }
            StoredIndexKind::Point => {
                let label = match def.label.as_deref() {
                    Some(l) => l,
                    None => return,
                };
                let cell_size = PointRegistry::cell_size_from_options(&def.options);
                for property in &def.properties {
                    self.activate_point_scope(def.entity, label, property, cell_size);
                }
            }
            StoredIndexKind::Fulltext => {
                let labels: Vec<String> = def.all_labels().map(String::from).collect();
                if labels.is_empty() {
                    return;
                }
                self.activate_fulltext_index(def.entity, &def.name, &labels, &def.properties);
            }
            StoredIndexKind::Vector => {
                let label = match def.label.as_deref() {
                    Some(l) => l,
                    None => return,
                };
                let property = match def.properties.first() {
                    Some(p) => p.as_str(),
                    None => return,
                };
                let similarity = VectorSimilarity::from_options(&def.options)
                    .unwrap_or(VectorSimilarity::Cosine);
                let provider = VectorIndexProvider::from_options(&def.options)
                    .unwrap_or(VectorIndexProvider::Flat);
                let hnsw = HnswParams::from_options(&def.options);
                let lazy = matches!(
                    def.options.get("vector.populate.async"),
                    Some(super::index_catalog::IndexConfigValue::Bool(true))
                );
                self.activate_vector_index(
                    def.entity, &def.name, label, property, similarity, provider, hnsw, lazy,
                );
                if lazy {
                    self.index_catalog_write()
                        .set_state(&def.name, StoredIndexState::Populating);
                }
            }
            // LOOKUP rides on the label/type indexes maintained eagerly.
            StoredIndexKind::Lookup => {}
        }
    }

    pub(super) fn text_indexes_read(
        &self,
        entity: StoredIndexEntity,
    ) -> IndexRead<'_, TrigramRegistry> {
        self.indexes.text.read(entity)
    }

    pub(super) fn text_indexes_write(
        &self,
        entity: StoredIndexEntity,
    ) -> IndexWrite<'_, TrigramRegistry> {
        self.indexes.text.write(entity)
    }

    pub(super) fn fulltext_indexes_read(
        &self,
        entity: StoredIndexEntity,
    ) -> IndexRead<'_, FulltextRegistry> {
        self.indexes.fulltext.read(entity)
    }

    #[allow(dead_code)]
    pub(super) fn fulltext_indexes_write(
        &self,
        entity: StoredIndexEntity,
    ) -> IndexWrite<'_, FulltextRegistry> {
        self.indexes.fulltext.write(entity)
    }

    fn activate_text_scope(&self, entity: StoredIndexEntity, label: &str, property: &str) {
        if !self.text_indexes_write(entity).add_scope(label, property) {
            return;
        }

        let backfill: Vec<(u64, String)> = match entity {
            StoredIndexEntity::Node => self
                .iter_nodes()
                .filter(|(_, node)| node.labels().strs().any(|l| l == label))
                .filter_map(|(id, node)| match node.properties().get(property) {
                    Some(crate::ValueRef::String(value)) => Some((id, value.to_owned())),
                    _ => None,
                })
                .collect(),
            StoredIndexEntity::Relationship => self
                .iter_rels()
                .filter(|(_, rel)| rel.rel_type() == label)
                .filter_map(|(id, rel)| match rel.properties().get(property) {
                    Some(crate::ValueRef::String(value)) => Some((id, value.to_owned())),
                    _ => None,
                })
                .collect(),
        };

        let mut registry = self.text_indexes_write(entity);
        for (id, value) in backfill {
            registry.insert(label, property, id, &value);
        }
    }

    /// Drop a (label, property) text scope, decrementing the refcount.
    pub(super) fn deactivate_text_scope(
        &self,
        entity: StoredIndexEntity,
        label: &str,
        property: &str,
    ) {
        self.text_indexes_write(entity)
            .remove_scope(label, property);
    }

    fn activate_fulltext_index(
        &self,
        entity: StoredIndexEntity,
        name: &str,
        labels: &[String],
        properties: &[String],
    ) {
        use super::fulltext_index::{term_counts_for_properties, TermCounts};

        {
            let mut registry = self.fulltext_indexes_write(entity);
            registry.register(name.to_string(), labels.to_vec(), properties.to_vec());
        }
        self.indexes
            .active_fulltext_indexes
            .fetch_add(1, Ordering::Relaxed);

        // Backfill: walk every entity matching any label, tokenise covered
        // string properties, install one posting batch per entity.
        let backfill: Vec<(u64, TermCounts)> = match entity {
            StoredIndexEntity::Node => self
                .iter_nodes()
                .filter(|(_, node)| {
                    labels
                        .iter()
                        .any(|wanted| node.labels().strs().any(|l| l == wanted))
                })
                .map(|(id, node)| {
                    let counts = term_counts_for_properties(node.properties(), properties);
                    (id, counts)
                })
                .filter(|(_, c)| !c.is_empty())
                .collect(),
            StoredIndexEntity::Relationship => self
                .iter_rels()
                .filter(|(_, rel)| labels.iter().any(|wanted| wanted == rel.rel_type()))
                .map(|(id, rel)| {
                    let counts = term_counts_for_properties(rel.properties(), properties);
                    (id, counts)
                })
                .filter(|(_, c)| !c.is_empty())
                .collect(),
        };

        let mut registry = self.fulltext_indexes_write(entity);
        if let Some(index) = registry.get_mut(name) {
            for (id, counts) in backfill {
                index.reindex_entity(id, counts);
            }
        }
    }

    pub(super) fn deactivate_fulltext_index(&self, entity: StoredIndexEntity, name: &str) {
        self.fulltext_indexes_write(entity).deregister(name);
        self.indexes
            .active_fulltext_indexes
            .fetch_sub(1, Ordering::Relaxed);
    }

    pub(super) fn sorted_indexes_read(
        &self,
        entity: StoredIndexEntity,
    ) -> IndexRead<'_, SortedPropertyIndex> {
        self.indexes.sorted.read(entity)
    }

    pub(super) fn sorted_indexes_write(
        &self,
        entity: StoredIndexEntity,
    ) -> IndexWrite<'_, SortedPropertyIndex> {
        self.indexes.sorted.write(entity)
    }

    pub(super) fn activate_sorted_scope(
        &self,
        entity: StoredIndexEntity,
        label: &str,
        property: &str,
    ) {
        let members = match entity {
            StoredIndexEntity::Node => self.nodes_by_label.get(label),
            StoredIndexEntity::Relationship => self.relationships_by_type.get(label),
        }
        .map_or(0, |ids| ids.len());
        if cfg!(all(feature = "parallel", not(target_arch = "wasm32")))
            && members >= PARALLEL_INDEX_MIN
        {
            return self.activate_sorted_scope_parallel(entity, label, property);
        }
        if !self.sorted_indexes_write(entity).add_scope(label, property) {
            return;
        }

        // Inserted while scanning: collecting every value first held a
        // second copy of the whole column until the index was built.
        let mut registry = self.sorted_indexes_write(entity);
        match entity {
            StoredIndexEntity::Node => {
                for (id, node) in self.iter_nodes() {
                    if !node.labels().strs().any(|l| l == label) {
                        continue;
                    }
                    match node.properties().get(property) {
                        // A list is indexed by its elements too.
                        Some(crate::ValueRef::Other(value)) => {
                            registry.insert(label, property, id, &value.get())
                        }
                        Some(value) => {
                            if let Some(key) = PropertyIndexKey::from_value_ref(value) {
                                registry.insert_key(label, property, id, key);
                            }
                        }
                        None => {}
                    }
                }
            }
            StoredIndexEntity::Relationship => {
                for (id, rel) in self.iter_rels() {
                    if rel.rel_type() != label {
                        continue;
                    }
                    match rel.properties().get(property) {
                        // A list is indexed by its elements too.
                        Some(crate::ValueRef::Other(value)) => {
                            registry.insert(label, property, id, &value.get())
                        }
                        Some(value) => {
                            if let Some(key) = PropertyIndexKey::from_value_ref(value) {
                                registry.insert_key(label, property, id, key);
                            }
                        }
                        None => {}
                    }
                }
            }
        }
    }

    /// [`Self::activate_sorted_scope`] for a large label: the entries are
    /// gathered over id ranges and sorted on every core, then filed in key
    /// order without the index lock, which is held only to install them.
    fn activate_sorted_scope_parallel(
        &self,
        entity: StoredIndexEntity,
        label: &str,
        property: &str,
    ) {
        use super::sorted_property_index::{build_scope, element_keys};
        if self.sorted_indexes_read(entity).has_scope(label, property) {
            self.sorted_indexes_write(entity).add_scope(label, property);
            return;
        }
        let (ids, slots) = match entity {
            StoredIndexEntity::Node => (self.nodes_by_label.get(label), self.nodes.len()),
            StoredIndexEntity::Relationship => (
                self.relationships_by_type.get(label),
                self.relationships.len(),
            ),
        };
        let Some(ids) = ids else {
            self.sorted_indexes_write(entity).add_scope(label, property);
            return;
        };
        let Some(key_id) = self.dicts.keys.id_of(property) else {
            self.sorted_indexes_write(entity).add_scope(label, property);
            return;
        };
        const RANGE: u64 = 1 << 16;
        let gather = |start: u64| {
            let mut values = Vec::new();
            let mut elements = Vec::new();
            for id in ids.iter_from(start).take_while(|&id| id < start + RANGE) {
                let value = match entity {
                    StoredIndexEntity::Node => self.nodes.get(id as usize).and_then(|bytes| {
                        StoredNode::new(id, bytes, &self.dicts)
                            .properties()
                            .get_id(key_id)
                    }),
                    StoredIndexEntity::Relationship => {
                        self.relationships.get(id as usize).and_then(|bytes| {
                            StoredRel::new(id, bytes, &self.dicts)
                                .properties()
                                .get_id(key_id)
                        })
                    }
                };
                match value {
                    // A list is indexed by its elements too.
                    Some(crate::ValueRef::Other(value)) => {
                        let value = value.get();
                        if let Some(key) = PropertyIndexKey::from_value(&value) {
                            values.push((key, id));
                        }
                        elements.extend(element_keys(&value).into_iter().map(|key| (key, id)));
                    }
                    Some(value) => {
                        if let Some(key) = PropertyIndexKey::from_value_ref(value) {
                            values.push((key, id));
                        }
                    }
                    None => {}
                }
            }
            (values, elements)
        };
        let starts: Vec<u64> = (0..(slots as u64).div_ceil(RANGE))
            .map(|i| i * RANGE)
            .collect();
        #[cfg(all(feature = "parallel", not(target_arch = "wasm32")))]
        let gathered: Vec<_> = {
            use rayon::prelude::*;
            starts.into_par_iter().map(gather).collect()
        };
        #[cfg(not(all(feature = "parallel", not(target_arch = "wasm32"))))]
        let gathered: Vec<_> = starts.into_iter().map(gather).collect();
        let mut values = Vec::new();
        let mut elements = Vec::new();
        for (v, e) in gathered {
            values.extend(v);
            elements.extend(e);
        }
        let built = build_scope(values, elements);
        self.sorted_indexes_write(entity)
            .add_built_scope(label, property, built);
    }

    pub(super) fn deactivate_sorted_scope(
        &self,
        entity: StoredIndexEntity,
        label: &str,
        property: &str,
    ) {
        self.sorted_indexes_write(entity)
            .remove_scope(label, property);
    }

    pub(super) fn point_indexes_read(
        &self,
        entity: StoredIndexEntity,
    ) -> IndexRead<'_, PointRegistry> {
        self.indexes.point.read(entity)
    }

    pub(super) fn point_indexes_write(
        &self,
        entity: StoredIndexEntity,
    ) -> IndexWrite<'_, PointRegistry> {
        self.indexes.point.write(entity)
    }

    fn activate_point_scope(
        &self,
        entity: StoredIndexEntity,
        label: &str,
        property: &str,
        cell_size: Option<f64>,
    ) {
        if !self
            .point_indexes_write(entity)
            .add_scope(label, property, cell_size)
        {
            return;
        }

        let backfill: Vec<(u64, LoraPoint)> = match entity {
            StoredIndexEntity::Node => self
                .iter_nodes()
                .filter(|(_, node)| node.labels().strs().any(|l| l == label))
                .filter_map(|(id, node)| {
                    match node.properties().get(property).map(|v| v.to_owned()) {
                        Some(PropertyValue::Point(point)) => Some((id, point)),
                        _ => None,
                    }
                })
                .collect(),
            StoredIndexEntity::Relationship => self
                .iter_rels()
                .filter(|(_, rel)| rel.rel_type() == label)
                .filter_map(|(id, rel)| {
                    match rel.properties().get(property).map(|v| v.to_owned()) {
                        Some(PropertyValue::Point(point)) => Some((id, point)),
                        _ => None,
                    }
                })
                .collect(),
        };

        let mut registry = self.point_indexes_write(entity);
        for (id, point) in backfill {
            registry.insert(label, property, id, point);
        }
    }

    pub(super) fn deactivate_point_scope(
        &self,
        entity: StoredIndexEntity,
        label: &str,
        property: &str,
    ) {
        self.point_indexes_write(entity)
            .remove_scope(label, property);
    }

    pub(super) fn vector_indexes_read(
        &self,
        entity: StoredIndexEntity,
    ) -> IndexRead<'_, VectorIndexRegistry> {
        self.indexes.vector.read(entity)
    }

    pub(super) fn vector_indexes_write(
        &self,
        entity: StoredIndexEntity,
    ) -> IndexWrite<'_, VectorIndexRegistry> {
        self.indexes.vector.write(entity)
    }

    #[allow(clippy::too_many_arguments)]
    fn activate_vector_index(
        &self,
        entity: StoredIndexEntity,
        name: &str,
        label: &str,
        property: &str,
        similarity: VectorSimilarity,
        provider: VectorIndexProvider,
        hnsw: HnswParams,
        lazy: bool,
    ) {
        {
            let mut registry = self.vector_indexes_write(entity);
            registry.register(
                name.to_string(),
                label.to_string(),
                property.to_string(),
                similarity,
                provider,
                hnsw,
            );
        }

        if lazy {
            // Skip the initial backfill — the catalog state is flipped
            // to Populating by the caller and the first query routed
            // to this index triggers `lazy_populate_vector_index`.
            // Mutations between CREATE and first query still feed the
            // registry via the maintenance hook, so the lazy phase
            // only handles vectors that existed before CREATE.
            return;
        }

        self.backfill_vector_index(entity, label, property);
    }

    /// Walk the property store for vectors matching this index's
    /// `(label, property)` scope and replay them into the registry.
    /// Shared by sync CREATE and lazy-populate flows.
    fn backfill_vector_index(&self, entity: StoredIndexEntity, label: &str, property: &str) {
        let backfill: Vec<(u64, crate::LoraVector)> = match entity {
            StoredIndexEntity::Node => self
                .iter_nodes()
                .filter(|(_, node)| node.labels().strs().any(|l| l == label))
                .filter_map(|(id, node)| {
                    match node.properties().get(property).map(|v| v.to_owned()) {
                        Some(PropertyValue::Vector(v)) => Some((id, v)),
                        _ => None,
                    }
                })
                .collect(),
            StoredIndexEntity::Relationship => self
                .iter_rels()
                .filter(|(_, rel)| rel.rel_type() == label)
                .filter_map(|(id, rel)| {
                    match rel.properties().get(property).map(|v| v.to_owned()) {
                        Some(PropertyValue::Vector(v)) => Some((id, v)),
                        _ => None,
                    }
                })
                .collect(),
        };

        let mut registry = self.vector_indexes_write(entity);
        for (id, vector) in backfill {
            registry.insert_for(label, property, id, &vector);
        }
    }

    /// Lazy-populate a `Populating` vector index: backfill from the
    /// property store, then flip catalog state to `Online`. Called
    /// from `GraphStorage::vector_search` on the first request that
    /// hits a still-populating index. Idempotent: a second concurrent
    /// caller finds the state already `Online` and does no work.
    pub(super) fn lazy_populate_vector_index(&self, name: &str) {
        let def = match self.index_catalog_read().get(name).cloned() {
            Some(d) => d,
            None => return,
        };
        if def.state != StoredIndexState::Populating || def.kind != StoredIndexKind::Vector {
            return;
        }
        let label = match def.label.as_deref() {
            Some(l) => l,
            None => return,
        };
        let property = match def.properties.first() {
            Some(p) => p.as_str(),
            None => return,
        };
        self.backfill_vector_index(def.entity, label, property);
        self.index_catalog_write()
            .set_state(name, StoredIndexState::Online);
    }

    pub(super) fn deactivate_vector_index(&self, entity: StoredIndexEntity, name: &str) {
        self.vector_indexes_write(entity).deregister(name);
    }

    /// Snapshot of cardinality stats. Cheap: derived from already-tracked
    /// `nodes_by_label` / `relationships_by_type` lengths and the
    /// distinct-value sketches (a cached estimate per (scope, key), and
    /// the previous map reused while those don't change). The cost model
    /// uses this to populate `estimated_rows` on plan-tree nodes.
    pub fn graph_stats(&self) -> GraphStats {
        let mut stats = GraphStats {
            node_count: self.live_node_count,
            relationship_count: self.live_rel_count,
            ..Default::default()
        };
        // Collected from ascending keys, the maps are built in one pass
        // instead of searched and grown per insert.
        stats.nodes_by_label = self
            .nodes_by_label
            .iter()
            .map(|(label, ids)| (label.to_string(), ids.len()))
            .collect();
        stats.relationships_by_type = self
            .relationships_by_type
            .iter()
            .map(|(rel_type, ids)| (rel_type.to_string(), ids.len()))
            .collect();
        // Distinct values per (label / type, property): estimated from
        // the distinct-value sketches, which exist for every key and are
        // a function of the live data alone. Deliberately *not* the
        // exact bucket counts of active hash indexes: which keys are
        // active depends on lookup history and on whether the process
        // restarted, and plans must not.
        stats.node_distinct_values = self.distinct_stats.nodes.distinct_values();
        stats.relationship_distinct_values = self.distinct_stats.relationships.distinct_values();

        for def in self.index_catalog_read().list() {
            if def.state != StoredIndexState::Online {
                continue;
            }
            // A hash-backed constraint's index answers equality only, which
            // the planner assumes of every key.
            if def.kind == StoredIndexKind::Range && !is_sorted(&def) {
                continue;
            }
            let Some(label) = def.label else {
                continue;
            };
            for property in def.properties {
                let scope = (label.clone(), property);
                match (def.entity, def.kind) {
                    (StoredIndexEntity::Node, StoredIndexKind::Range) => {
                        stats.node_range_indexes.insert(scope);
                    }
                    (StoredIndexEntity::Node, StoredIndexKind::Text) => {
                        stats.node_text_indexes.insert(scope);
                    }
                    (StoredIndexEntity::Node, StoredIndexKind::Point) => {
                        stats.node_point_indexes.insert(scope);
                    }
                    (StoredIndexEntity::Relationship, StoredIndexKind::Range) => {
                        stats.relationship_range_indexes.insert(scope);
                    }
                    (StoredIndexEntity::Relationship, StoredIndexKind::Text) => {
                        stats.relationship_text_indexes.insert(scope);
                    }
                    (StoredIndexEntity::Relationship, StoredIndexKind::Point) => {
                        stats.relationship_point_indexes.insert(scope);
                    }
                    (StoredIndexEntity::Node, StoredIndexKind::Vector) => {
                        stats.node_vector_indexes.insert(scope);
                    }
                    (StoredIndexEntity::Relationship, StoredIndexKind::Vector) => {
                        stats.relationship_vector_indexes.insert(scope);
                    }
                    (_, StoredIndexKind::Lookup | StoredIndexKind::Fulltext) => {}
                }
            }
        }
        for (entity, elements) in [
            (StoredIndexEntity::Node, &mut stats.node_list_elements),
            (
                StoredIndexEntity::Relationship,
                &mut stats.relationship_list_elements,
            ),
        ] {
            let sorted = self.sorted_indexes_read(entity);
            for (label, property, distinct, postings) in sorted.element_stats() {
                elements.insert(
                    (label.to_string(), property.to_string()),
                    super::ListElementStats { distinct, postings },
                );
            }
        }
        stats
    }

    /// [`GraphStats::plan_signature`] of [`Self::graph_stats`], computed
    /// without building the stats: one pass over the label / type maps,
    /// the distinct-value estimates and the index catalog. Plan caches
    /// call it after every write, so it must stay allocation-free.
    pub fn plan_signature(&self) -> u64 {
        let mut sig = super::PlanSignature::new();
        sig.totals(self.live_node_count, self.live_rel_count);
        for (label, ids) in &self.nodes_by_label {
            sig.label(label, ids.len());
        }
        for (rel_type, ids) in &self.relationships_by_type {
            sig.relationship_type(rel_type, ids.len());
        }
        for (label, property, distinct) in self.distinct_stats.nodes.estimates() {
            sig.node_distinct(label, property, distinct);
        }
        for (rel_type, property, distinct) in self.distinct_stats.relationships.estimates() {
            sig.relationship_distinct(rel_type, property, distinct);
        }
        for def in self.index_catalog_read().iter() {
            if def.state != StoredIndexState::Online
                || matches!(
                    def.kind,
                    StoredIndexKind::Lookup | StoredIndexKind::Fulltext
                )
            {
                continue;
            }
            let Some(label) = &def.label else {
                continue;
            };
            for property in &def.properties {
                sig.index(def.entity, def.kind, label, property);
            }
        }
        for entity in [StoredIndexEntity::Node, StoredIndexEntity::Relationship] {
            let sorted = self.sorted_indexes_read(entity);
            for (label, property, distinct, postings) in sorted.element_stats() {
                sig.list_elements(
                    entity,
                    label,
                    property,
                    super::ListElementStats { distinct, postings },
                );
            }
        }
        sig.finish()
    }

    /// Approximate retained-heap breakdown of this graph. See
    /// [`super::MemoryReport`] for the methodology and per-component
    /// fields. Intended for benches and the `mem_probe*` examples;
    /// not on a hot path.
    pub fn memory_estimate(&self) -> super::MemoryReport {
        super::mem_report::estimate(self)
    }

    /// Count (or uncount) every property of an entity under each of its
    /// scopes in the distinct-value sketches.
    pub(super) fn count_distinct_values<'a>(
        stats: &mut super::distinct_stats::DistinctStats,
        scopes: impl IntoIterator<Item = &'a str>,
        properties: &Properties,
        add: bool,
    ) {
        if properties.is_empty() {
            return;
        }
        for scope in scopes {
            if add {
                stats.add_all(scope, properties);
            } else {
                for (key, value) in properties {
                    stats.remove(scope, key, value);
                }
            }
        }
    }

    pub(super) fn on_node_created(&mut self, node: &NodeRecord) {
        for label in &node.labels {
            self.insert_node_label_index(node.id, label);
        }
        Self::count_distinct_values(
            &mut self.distinct_stats.nodes,
            node.labels.strs(),
            &node.properties,
            true,
        );
        self.index_node_properties_if_active(node.id, node.labels.strs(), &node.properties);
        self.maintain_node_secondary_indexes(node, SecondaryIndexMutation::Insert);
    }

    pub(super) fn on_node_property_set(
        &mut self,
        node_id: NodeId,
        key: &str,
        old: Option<&PropertyValue>,
        new: &PropertyValue,
    ) {
        let Some(labels) = self.node_at(node_id).map(|node| node.labels().to_owned()) else {
            return;
        };
        for label in &labels {
            self.distinct_stats.nodes.replace(label, key, old, new);
        }

        if self.node_property_index_is_active(key) {
            if let Some(old) = old {
                self.unindex_node_property_if_active(node_id, labels.strs(), key, old);
            }
            self.index_node_property_if_active(node_id, labels.strs(), key, new);
        }

        self.update_secondary_property(
            StoredIndexEntity::Node,
            labels.strs(),
            node_id,
            key,
            old,
            Some(new),
        );
    }

    pub(super) fn on_node_property_removed(
        &mut self,
        node_id: NodeId,
        key: &str,
        old: &PropertyValue,
    ) {
        let Some(labels) = self.node_at(node_id).map(|node| node.labels().to_owned()) else {
            return;
        };
        for label in &labels {
            self.distinct_stats.nodes.remove(label, key, old);
        }
        if self.node_property_index_is_active(key) {
            self.unindex_node_property_if_active(node_id, labels.strs(), key, old);
        }
        self.update_secondary_property(
            StoredIndexEntity::Node,
            labels.strs(),
            node_id,
            key,
            Some(old),
            None,
        );
    }

    pub(super) fn on_node_label_added(&mut self, node_id: NodeId, label: &str) {
        self.insert_node_label_index(node_id, label);

        let Some(properties) = self
            .node_at(node_id)
            .map(|node| node.properties().to_owned())
        else {
            return;
        };
        Self::count_distinct_values(&mut self.distinct_stats.nodes, [label], &properties, true);
        if self.active_node_property_index_count() != 0 {
            self.index_node_scope_properties_if_active(node_id, label, &properties);
            // Its first label: it was indexed as a node without any.
            if self.node_at(node_id).is_some_and(|n| n.labels().len() == 1) {
                self.unindex_node_scope_properties_if_active(node_id, UNLABELLED, &properties);
            }
        }
        for (key, value) in &properties {
            self.update_secondary_property(
                StoredIndexEntity::Node,
                [label],
                node_id,
                key,
                None,
                Some(value),
            );
        }
    }

    pub(super) fn on_node_label_removed(&mut self, node_id: NodeId, label: &str) {
        self.remove_node_label_index(node_id, label);

        let Some(properties) = self
            .node_at(node_id)
            .map(|node| node.properties().to_owned())
        else {
            return;
        };
        Self::count_distinct_values(&mut self.distinct_stats.nodes, [label], &properties, false);
        if self.active_node_property_index_count() != 0 {
            self.unindex_node_scope_properties_if_active(node_id, label, &properties);
            // Its last label: index it as a node without any.
            if self.node_at(node_id).is_some_and(|n| n.labels().is_empty()) {
                self.index_node_scope_properties_if_active(node_id, UNLABELLED, &properties);
            }
        }
        for (key, value) in &properties {
            self.update_secondary_property(
                StoredIndexEntity::Node,
                [label],
                node_id,
                key,
                Some(value),
                None,
            );
        }
    }

    pub(super) fn on_node_deleted(&mut self, node: &NodeRecord) {
        for label in &node.labels {
            self.remove_node_label_index(node.id, label);
        }
        Self::count_distinct_values(
            &mut self.distinct_stats.nodes,
            node.labels.strs(),
            &node.properties,
            false,
        );
        self.unindex_active_node_properties(node.id, node.labels.strs(), &node.properties);
        self.maintain_node_secondary_indexes(node, SecondaryIndexMutation::Remove);
    }

    pub(super) fn on_relationship_created(&mut self, rel: &RelationshipRecord) {
        self.attach_relationship(rel);
        Self::count_distinct_values(
            &mut self.distinct_stats.relationships,
            [rel.rel_type.as_str()],
            &rel.properties,
            true,
        );
        self.index_relationship_properties_if_active(
            rel.id,
            [rel.rel_type.as_str()],
            &rel.properties,
        );
        self.maintain_relationship_secondary_indexes(rel, SecondaryIndexMutation::Insert);
    }

    pub(super) fn on_relationship_property_set(
        &mut self,
        rel_id: RelationshipId,
        key: &str,
        old: Option<&PropertyValue>,
        new: &PropertyValue,
    ) {
        let Some(rel_type) = self.rel_at(rel_id).map(|rel| rel.type_name().clone()) else {
            return;
        };
        self.distinct_stats
            .relationships
            .replace(&rel_type, key, old, new);

        if self.relationship_property_index_is_active(key) {
            if let Some(old) = old {
                self.unindex_relationship_property_if_active(rel_id, [rel_type.as_str()], key, old);
            }
            self.index_relationship_property_if_active(rel_id, [rel_type.as_str()], key, new);
        }

        self.update_secondary_property(
            StoredIndexEntity::Relationship,
            [rel_type.as_str()],
            rel_id,
            key,
            old,
            Some(new),
        );
    }

    pub(super) fn on_relationship_property_removed(
        &mut self,
        rel_id: RelationshipId,
        key: &str,
        old: &PropertyValue,
    ) {
        let Some(rel_type) = self.rel_at(rel_id).map(|rel| rel.type_name().clone()) else {
            return;
        };
        self.distinct_stats
            .relationships
            .remove(&rel_type, key, old);
        if self.relationship_property_index_is_active(key) {
            self.unindex_relationship_property_if_active(rel_id, [rel_type.as_str()], key, old);
        }
        self.update_secondary_property(
            StoredIndexEntity::Relationship,
            [rel_type.as_str()],
            rel_id,
            key,
            Some(old),
            None,
        );
    }

    /// Index upkeep for a deleted relationship. `pending` defers its
    /// removal from long adjacency lists: see [`Self::adjacency_remove`].
    pub(super) fn on_relationship_deleted(
        &mut self,
        rel: &RelationshipRecord,
        pending: Option<&mut PendingAdjRemovals>,
    ) {
        self.detach_relationship_indexes(rel, pending);
        Self::count_distinct_values(
            &mut self.distinct_stats.relationships,
            [rel.rel_type.as_str()],
            &rel.properties,
            false,
        );
        self.unindex_active_relationship_properties(
            rel.id,
            [rel.rel_type.as_str()],
            &rel.properties,
        );
        self.maintain_relationship_secondary_indexes(rel, SecondaryIndexMutation::Remove);
    }

    fn index_node_property_if_active<'a>(
        &mut self,
        node_id: NodeId,
        labels: impl IntoIterator<Item = &'a str>,
        key: &str,
        value: &PropertyValue,
    ) {
        if self.active_node_property_index_count() == 0 {
            return;
        }
        let (indexes, records) = self.node_index_parts();
        if indexes.node_properties.is_active(key) {
            indexes
                .node_properties
                .insert_with_scopes(node_id, labels, key, value, &records);
        }
    }

    fn index_node_properties_if_active<'a>(
        &mut self,
        node_id: NodeId,
        labels: impl IntoIterator<Item = &'a str> + Clone,
        properties: &Properties,
    ) {
        if self.active_node_property_index_count() == 0 {
            return;
        }
        let (indexes, records) = self.node_index_parts();
        for (key, value) in properties {
            if indexes.node_properties.is_active(key) {
                indexes.node_properties.insert_with_scopes(
                    node_id,
                    labels.clone(),
                    key,
                    value,
                    &records,
                );
            }
        }
    }

    fn unindex_node_property_if_active<'a>(
        &mut self,
        node_id: NodeId,
        labels: impl IntoIterator<Item = &'a str>,
        key: &str,
        value: &PropertyValue,
    ) {
        if self.active_node_property_index_count() == 0 {
            return;
        }
        let indexes = self.indexes_mut();
        if indexes.node_properties.is_active(key) {
            indexes
                .node_properties
                .remove_with_scopes(node_id, labels, key, value);
        }
    }

    fn index_node_scope_properties_if_active(
        &mut self,
        node_id: NodeId,
        scope: &str,
        properties: &Properties,
    ) {
        if self.active_node_property_index_count() == 0 {
            return;
        }
        let (indexes, records) = self.node_index_parts();
        for (key, value) in properties {
            if indexes.node_properties.is_active(key) {
                indexes
                    .node_properties
                    .insert_scoped(node_id, scope, key, value, &records);
            }
        }
    }

    fn unindex_node_scope_properties_if_active(
        &mut self,
        node_id: NodeId,
        scope: &str,
        properties: &Properties,
    ) {
        if self.active_node_property_index_count() == 0 {
            return;
        }
        let indexes = self.indexes_mut();
        for (key, value) in properties {
            if indexes.node_properties.is_active(key) {
                indexes
                    .node_properties
                    .remove_scoped(node_id, scope, key, value);
            }
        }
    }

    fn unindex_active_node_properties<'a>(
        &mut self,
        node_id: NodeId,
        labels: impl IntoIterator<Item = &'a str> + Clone,
        properties: &Properties,
    ) {
        if self.active_node_property_index_count() == 0 {
            return;
        }
        let indexes = self.indexes_mut();
        for (key, value) in properties {
            if indexes.node_properties.is_active(key) {
                indexes
                    .node_properties
                    .remove_with_scopes(node_id, labels.clone(), key, value);
            }
        }
    }

    fn index_relationship_property_if_active<'a>(
        &mut self,
        rel_id: RelationshipId,
        scopes: impl IntoIterator<Item = &'a str>,
        key: &str,
        value: &PropertyValue,
    ) {
        if self.active_relationship_property_index_count() == 0 {
            return;
        }
        let (indexes, records) = self.relationship_index_parts();
        if indexes.relationship_properties.is_active(key) {
            indexes
                .relationship_properties
                .insert_with_scopes(rel_id, scopes, key, value, &records);
        }
    }

    fn index_relationship_properties_if_active<'a>(
        &mut self,
        rel_id: RelationshipId,
        scopes: impl IntoIterator<Item = &'a str> + Clone,
        properties: &Properties,
    ) {
        if self.active_relationship_property_index_count() == 0 {
            return;
        }
        let (indexes, records) = self.relationship_index_parts();
        for (key, value) in properties {
            if indexes.relationship_properties.is_active(key) {
                indexes.relationship_properties.insert_with_scopes(
                    rel_id,
                    scopes.clone(),
                    key,
                    value,
                    &records,
                );
            }
        }
    }

    fn unindex_relationship_property_if_active<'a>(
        &mut self,
        rel_id: RelationshipId,
        scopes: impl IntoIterator<Item = &'a str>,
        key: &str,
        value: &PropertyValue,
    ) {
        if self.active_relationship_property_index_count() == 0 {
            return;
        }
        let indexes = self.indexes_mut();
        if indexes.relationship_properties.is_active(key) {
            indexes
                .relationship_properties
                .remove_with_scopes(rel_id, scopes, key, value);
        }
    }

    fn unindex_active_relationship_properties<'a>(
        &mut self,
        rel_id: RelationshipId,
        scopes: impl IntoIterator<Item = &'a str> + Clone,
        properties: &Properties,
    ) {
        if self.active_relationship_property_index_count() == 0 {
            return;
        }
        let indexes = self.indexes_mut();
        for (key, value) in properties {
            if indexes.relationship_properties.is_active(key) {
                indexes.relationship_properties.remove_with_scopes(
                    rel_id,
                    scopes.clone(),
                    key,
                    value,
                );
            }
        }
    }

    pub(super) fn scan_nodes_by_property(
        &self,
        label: Option<&str>,
        key: &str,
        value: &PropertyValue,
    ) -> Vec<NodeRecord> {
        match label {
            Some(label) => self
                .nodes_by_label
                .get(label)
                .into_iter()
                .flat_map(|ids| ids.iter())
                .filter_map(|id| self.node_at(id))
                .filter(|node| node.properties().get(key).is_some_and(|v| v == *value))
                .map(|node| node.to_record())
                .collect(),
            None => self
                .iter_node_refs()
                .filter(|node| node.properties().get(key).is_some_and(|v| v == *value))
                .map(|node| node.to_record())
                .collect(),
        }
    }

    pub(super) fn scan_node_ids_by_property(
        &self,
        label: Option<&str>,
        key: &str,
        value: &PropertyValue,
    ) -> Vec<NodeId> {
        match label {
            Some(label) => self
                .nodes_by_label
                .get(label)
                .into_iter()
                .flat_map(|ids| ids.iter())
                .filter_map(|id| {
                    (self
                        .node_at(id)?
                        .properties()
                        .get(key)
                        .is_some_and(|v| v == *value))
                    .then_some(id)
                })
                .collect(),
            None => self
                .iter_nodes()
                .filter_map(|(id, node)| {
                    (node.properties().get(key).is_some_and(|v| v == *value)).then_some(id)
                })
                .collect(),
        }
    }

    pub(super) fn any_node_by_property(
        &self,
        label: &str,
        key: &str,
        value: &PropertyValue,
    ) -> bool {
        self.nodes_by_label
            .get(label)
            .into_iter()
            .flat_map(|ids| ids.iter())
            .filter_map(|id| self.node_at(id))
            .any(|node| node.properties().get(key).is_some_and(|v| v == *value))
    }

    pub(super) fn scan_relationships_by_property(
        &self,
        rel_type: Option<&str>,
        key: &str,
        value: &PropertyValue,
    ) -> Vec<RelationshipRecord> {
        match rel_type {
            Some(rel_type) => self
                .relationships_by_type
                .get(rel_type)
                .into_iter()
                .flat_map(|ids| ids.iter())
                .filter_map(|id| self.rel_at(id))
                .filter(|rel| rel.properties().get(key).is_some_and(|v| v == *value))
                .map(|rel| rel.to_record())
                .collect(),
            None => self
                .iter_rel_refs()
                .filter(|rel| rel.properties().get(key).is_some_and(|v| v == *value))
                .map(|rel| rel.to_record())
                .collect(),
        }
    }

    pub(super) fn scan_relationship_ids_by_property(
        &self,
        rel_type: Option<&str>,
        key: &str,
        value: &PropertyValue,
    ) -> Vec<RelationshipId> {
        match rel_type {
            Some(rel_type) => self
                .relationships_by_type
                .get(rel_type)
                .into_iter()
                .flat_map(|ids| ids.iter())
                .filter_map(|id| {
                    (self
                        .rel_at(id)?
                        .properties()
                        .get(key)
                        .is_some_and(|v| v == *value))
                    .then_some(id)
                })
                .collect(),
            None => self
                .iter_rels()
                .filter_map(|(id, rel)| {
                    (rel.properties().get(key).is_some_and(|v| v == *value)).then_some(id)
                })
                .collect(),
        }
    }

    pub(super) fn any_relationship_by_property(
        &self,
        rel_type: &str,
        key: &str,
        value: &PropertyValue,
    ) -> bool {
        self.relationships_by_type
            .get(rel_type)
            .into_iter()
            .flat_map(|ids| ids.iter())
            .filter_map(|id| self.rel_at(id))
            .any(|rel| rel.properties().get(key).is_some_and(|v| v == *value))
    }

    pub(super) fn attach_relationship(&mut self, rel: &RelationshipRecord) {
        let type_id = self.dicts.types.id_or_insert(&rel.rel_type);
        self.adjacency_push(
            rel.src,
            Side::Out,
            AdjEntry {
                type_id,
                neighbour: rel.dst,
                rel: rel.id,
            },
        );
        self.adjacency_push(
            rel.dst,
            Side::In,
            AdjEntry {
                type_id,
                neighbour: rel.src,
                rel: rel.id,
            },
        );
        self.insert_relationship_type_index(rel.id, &rel.rel_type);
    }

    fn detach_relationship_indexes(
        &mut self,
        rel: &RelationshipRecord,
        mut pending: Option<&mut PendingAdjRemovals>,
    ) {
        self.adjacency_remove(rel.src, Side::Out, rel.id, pending.as_deref_mut());
        self.adjacency_remove(rel.dst, Side::In, rel.id, pending);

        self.remove_relationship_type_index(rel.id, &rel.rel_type);
    }

    pub(super) fn relationship_ids_for_direction(
        &self,
        node_id: NodeId,
        direction: Direction,
    ) -> Vec<RelationshipId> {
        let mut ids = Vec::new();
        let _ = self.try_for_each_adjacent_id_unchecked(node_id, direction, &[], |rel_id, _| {
            ids.push(rel_id);
            Ok::<(), ()>(())
        });
        ids
    }

    pub(super) fn has_incident_relationships(&self, node_id: NodeId) -> bool {
        self.outgoing_at(node_id).is_some_and(|adj| !adj.is_empty())
            || self.incoming_at(node_id).is_some_and(|adj| !adj.is_empty())
    }

    pub(super) fn incident_relationship_ids(&self, node_id: NodeId) -> Vec<RelationshipId> {
        self.relationship_ids_for_direction(node_id, Direction::Undirected)
    }

    /// Replay a node creation using the id captured in a durable mutation
    /// event. This intentionally does not emit a new mutation event: callers
    /// must invoke it before installing a recorder on the graph.
    #[doc(hidden)]
    pub fn replay_create_node(
        &mut self,
        id: NodeId,
        labels: Vec<String>,
        properties: Properties,
    ) -> Result<NodeRecord, String> {
        if self.recorder.is_some() {
            return Err(
                "cannot replay node creation while a mutation recorder is installed".into(),
            );
        }
        if self.has_node_at(id) {
            return Err(format!("node id {id} already exists"));
        }
        let idx = self.ensure_node_slot_checked(id)?;
        self.bump_next_node_id_past(id)?;

        let labels = Self::normalize_labels(labels);
        let node = NodeRecord {
            id,
            labels,
            properties,
        };

        self.put_node_at_slot(idx, &node);
        // Same index maintenance as a live create: only hash indexes that
        // are already active (declared by a replayed CREATE INDEX /
        // CREATE CONSTRAINT, which backfills from the data replayed so
        // far) are kept current. Lookup-activated (implicit) indexes are
        // rebuilt lazily on first use, exactly as in a fresh process.
        self.on_node_created(&node);

        Ok(node)
    }

    /// Replay a relationship creation using the id captured in a durable
    /// mutation event. This intentionally does not emit a new mutation event:
    /// callers must invoke it before installing a recorder on the graph.
    #[doc(hidden)]
    pub fn replay_create_relationship(
        &mut self,
        id: RelationshipId,
        src: NodeId,
        dst: NodeId,
        rel_type: &str,
        properties: Properties,
    ) -> Result<RelationshipRecord, String> {
        if self.recorder.is_some() {
            return Err(
                "cannot replay relationship creation while a mutation recorder is installed".into(),
            );
        }
        if self.has_rel_at(id) {
            return Err(format!("relationship id {id} already exists"));
        }
        if !self.has_node_at(src) {
            return Err(format!(
                "relationship {id} references missing source node {src}"
            ));
        }
        if !self.has_node_at(dst) {
            return Err(format!(
                "relationship {id} references missing target node {dst}"
            ));
        }

        let trimmed = rel_type.trim();
        if trimmed.is_empty() {
            return Err(format!("relationship {id} has an empty type"));
        }
        let idx = self.ensure_rel_slot_checked(id)?;
        self.bump_next_rel_id_past(id)?;

        let rel = RelationshipRecord {
            id,
            src,
            dst,
            rel_type: trimmed.into(),
            properties,
        };

        self.put_rel_at_slot(idx, &rel);
        // See `replay_create_node`: active (declared) indexes only.
        self.on_relationship_created(&rel);

        Ok(rel)
    }

    #[cfg(test)]
    pub(super) fn assert_property_indexes_match_scan(&self) {
        let indexes = self.indexes_read();
        assert_eq!(
            indexes.node_properties.active_keys.len(),
            self.active_node_property_index_count(),
            "node property index counter diverged from active key set"
        );
        assert_eq!(
            indexes.relationship_properties.active_keys.len(),
            self.active_relationship_property_index_count(),
            "relationship property index counter diverged from active key set"
        );

        let mut expected_nodes = PropertyIndexState {
            active_keys: indexes.node_properties.active_keys.clone(),
            any_scope_keys: indexes.node_properties.any_scope_keys.clone(),
            ..PropertyIndexState::default()
        };
        for (id, node) in self.iter_nodes() {
            for (key, value) in node.properties() {
                if expected_nodes.is_active(key) {
                    expected_nodes.insert_with_scopes(
                        id,
                        node.labels().strs(),
                        key,
                        &value.to_owned(),
                        &self.node_values(),
                    );
                }
            }
        }
        assert_eq!(
            indexes.node_properties.scoped_values, expected_nodes.scoped_values,
            "node property scoped index values diverged from scan"
        );
        assert_eq!(
            indexes.node_properties.any_scope, expected_nodes.any_scope,
            "node property across-scopes index values diverged from scan"
        );

        let mut expected_relationships = PropertyIndexState {
            active_keys: indexes.relationship_properties.active_keys.clone(),
            any_scope_keys: indexes.relationship_properties.any_scope_keys.clone(),
            ..PropertyIndexState::default()
        };
        for (id, rel) in self.iter_rels() {
            for (key, value) in rel.properties() {
                if expected_relationships.is_active(key) {
                    expected_relationships.insert_with_scopes(
                        id,
                        [rel.rel_type()],
                        key,
                        &value.to_owned(),
                        &self.relationship_values(),
                    );
                }
            }
        }
        assert_eq!(
            indexes.relationship_properties.scoped_values, expected_relationships.scoped_values,
            "relationship property scoped index values diverged from scan"
        );
        assert_eq!(
            indexes.relationship_properties.any_scope, expected_relationships.any_scope,
            "relationship property across-scopes index values diverged from scan"
        );
    }
}

/// Property values read from a slab's records, for the hash property
/// index's digest entries (see `PropertyValueBuckets`).
pub(super) struct RecordValues<'g> {
    slab: &'g RecordSlab,
    dicts: &'g Dicts,
    nodes: bool,
}

impl<'g> RecordValues<'g> {
    pub(super) fn nodes(slab: &'g RecordSlab, dicts: &'g Dicts) -> Self {
        Self {
            slab,
            dicts,
            nodes: true,
        }
    }

    pub(super) fn relationships(slab: &'g RecordSlab, dicts: &'g Dicts) -> Self {
        Self {
            slab,
            dicts,
            nodes: false,
        }
    }
}

impl EntityValues for RecordValues<'_> {
    #[inline]
    fn get(&self, id: u64, key: &str) -> Option<crate::ValueRef<'_>> {
        let bytes = self.slab.get(usize::try_from(id).ok()?)?;
        if self.nodes {
            StoredNode::new(id, bytes, self.dicts).properties().get(key)
        } else {
            StoredRel::new(id, bytes, self.dicts).properties().get(key)
        }
    }
}

/// Set or remove one property of the record in `slab` slot `idx`,
/// rewriting only that entry. The previous value; `None` when the slot is
/// empty or there was nothing to remove.
#[inline]
fn edit_record_property(
    slab: &mut RecordSlab,
    idx: usize,
    kind: encoded::RecordKind,
    dicts: &mut Dicts,
    key: &str,
    edit: encoded::PropEdit<'_>,
) -> Option<Option<PropertyValue>> {
    let bytes = slab.get(idx)?;
    let mut out = Scratch::take();
    let old = encoded::edit_property(bytes, kind, dicts, key, edit, &mut out)?;
    slab.set(idx, &out);
    Some(old)
}

/// Option of a constraint's backing RANGE index: `false` when it keeps the
/// hash index only (see [`ConstraintBacking::Hash`]).
pub(super) const SORTED_OPTION: &str = "sorted";

/// Whether a RANGE index keeps a sorted copy of its values.
pub(super) fn is_sorted(def: &IndexDefinition) -> bool {
    !matches!(
        def.options.get(SORTED_OPTION),
        Some(IndexConfigValue::Bool(false))
    )
}

/// Entities from which a property index is built on every core: below
/// it, splitting the work costs more than it saves.
const PARALLEL_INDEX_MIN: usize = 1 << 17;

/// Data work a restore registers its catalogs without, to do it all at
/// once in [`InMemoryGraph::run_deferred`].
#[derive(Default)]
pub(super) struct DeferredData {
    /// Indexes to populate from the records.
    pub(super) populate: Vec<IndexDefinition>,
    /// Constraints to check the records against.
    pub(super) validate: Vec<super::ConstraintDefinition>,
}