reddb-io-server 1.24.3

//! Unified Index Store
//!
//! Holds all user-created secondary indices (Hash, Bitmap, BTree, H3) and
//! provides a single point of access for the query executor.
//!
//! The executor calls `lookup()` with a collection, column, and value —
//! the IndexStore finds the right index and returns matching entity IDs.

use parking_lot::{RwLock, RwLockReadGuard, RwLockWriteGuard};
use std::borrow::Cow;
use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::ops::Bound::{Excluded, Included, Unbounded};

use crate::storage::unified::bitmap_index::BitmapIndexManager;
use crate::storage::unified::entity::EntityId;
use crate::storage::unified::hash_index::{HashIndexConfig, HashIndexManager};
use reddb_types::{value_to_canonical_key, CanonicalKey, CanonicalKeyFamily, Value};

enum CanonicalizedValue {
    Exact(CanonicalKey),
    Unsupported,
}

fn read_unpoisoned<'a, T>(lock: &'a RwLock<T>) -> RwLockReadGuard<'a, T> {
    lock.read()
}

fn write_unpoisoned<'a, T>(lock: &'a RwLock<T>) -> RwLockWriteGuard<'a, T> {
    lock.write()
}

/// Per-entry constant charged to a `BTreeMap` entry: node slot, the `Vec`
/// header of its posting list, and allocator overhead. Matches the constant
/// the hash and bitmap estimators already use so the `index_memory` pool sums
/// comparable numbers across families (ADR 0073 §2).
const BTREE_ENTRY_OVERHEAD_BYTES: usize = 48;

/// Heap bytes a canonical key owns beyond its inline enum footprint.
fn canonical_key_bytes(key: &CanonicalKey) -> usize {
    let heap = match key {
        CanonicalKey::Text(_, text) => text.len(),
        CanonicalKey::Bytes(_, bytes) => bytes.len(),
        CanonicalKey::PairTextU64(_, text, _) => text.len(),
        CanonicalKey::PairTextText(_, left, right) => left.len() + right.len(),
        CanonicalKey::Null
        | CanonicalKey::Boolean(_)
        | CanonicalKey::Signed(_, _)
        | CanonicalKey::Unsigned(_, _)
        | CanonicalKey::Float(_)
        | CanonicalKey::PairU32U8(_, _, _)
        | CanonicalKey::PairU32U32(_, _, _)
        | CanonicalKey::PairI32I32(_, _, _) => 0,
    };
    std::mem::size_of::<CanonicalKey>() + heap
}

/// Approximate bytes for one sorted-index entry: its key (single or composite)
/// plus the entity IDs posted under it.
fn btree_entry_bytes<'a>(
    key_parts: impl Iterator<Item = &'a CanonicalKey>,
    posted_ids: usize,
) -> usize {
    key_parts.map(canonical_key_bytes).sum::<usize>()
        + posted_ids * std::mem::size_of::<EntityId>()
        + BTREE_ENTRY_OVERHEAD_BYTES
}

/// In-memory sorted index for exact range scans over a single canonical family.
///
/// Point lookups remain safe even when the column contains mixed families,
/// because BTree seeks are exact on the canonical key. Range scans are only
/// enabled when every indexed value for the column belongs to the same family.
pub struct SortedColumnIndex {
    /// Sorted entries: canonical key → entity IDs
    entries: BTreeMap<CanonicalKey, Vec<EntityId>>,
    entry_memory_bytes: usize,
    /// Family seen in this index. Mixed families keep exact lookup safe but
    /// disable range pushdown.
    range_family: Option<CanonicalKeyFamily>,
    has_mixed_families: bool,
    families: BTreeSet<CanonicalKeyFamily>,
}

impl SortedColumnIndex {
    pub fn new() -> Self {
        Self {
            entries: BTreeMap::new(),
            entry_memory_bytes: 0,
            range_family: None,
            has_mixed_families: false,
            families: BTreeSet::new(),
        }
    }

    pub fn insert(&mut self, key: CanonicalKey, entity_id: EntityId) {
        self.families.insert(key.family());
        match self.range_family {
            Some(existing) if existing != key.family() => self.has_mixed_families = true,
            None => self.range_family = Some(key.family()),
            _ => {}
        }
        let entry = self.entries.entry(key);
        if let std::collections::btree_map::Entry::Vacant(ref entry) = entry {
            self.entry_memory_bytes += btree_entry_bytes(std::iter::once(entry.key()), 0);
        }
        self.entry_memory_bytes += std::mem::size_of::<EntityId>();
        entry.or_default().push(entity_id);
    }

    fn range_enabled(&self, family: CanonicalKeyFamily) -> bool {
        !self.has_mixed_families && self.range_family == Some(family)
    }

    pub fn supports_range_key(&self, key: &CanonicalKey) -> bool {
        self.range_enabled(key.family())
    }

    pub fn supports_mixed_integral_ranges(&self) -> bool {
        !self.families.is_empty()
            && self.families.iter().all(|family| {
                matches!(
                    family,
                    CanonicalKeyFamily::Integer | CanonicalKeyFamily::UnsignedInteger
                )
            })
    }

    /// Range scan: returns all entity IDs where key is in [low, high].
    pub fn range(&self, low: CanonicalKey, high: CanonicalKey) -> Option<Vec<EntityId>> {
        if !self.range_enabled(low.family()) || low.family() != high.family() {
            return None;
        }
        if low > high {
            return Some(Vec::new());
        }
        Some(self.collect_range(low..=high))
    }

    /// Greater than: returns all entity IDs where key > threshold.
    pub fn greater_than(&self, threshold: CanonicalKey) -> Option<Vec<EntityId>> {
        if !self.range_enabled(threshold.family()) {
            return None;
        }
        Some(self.collect_range((Excluded(threshold), Unbounded)))
    }

    pub fn greater_equal(&self, threshold: CanonicalKey) -> Option<Vec<EntityId>> {
        if !self.range_enabled(threshold.family()) {
            return None;
        }
        Some(self.collect_range((Included(threshold), Unbounded)))
    }

    pub fn less_than(&self, threshold: CanonicalKey) -> Option<Vec<EntityId>> {
        if !self.range_enabled(threshold.family()) {
            return None;
        }
        Some(self.collect_range((Unbounded, Excluded(threshold))))
    }

    pub fn less_equal(&self, threshold: CanonicalKey) -> Option<Vec<EntityId>> {
        if !self.range_enabled(threshold.family()) {
            return None;
        }
        Some(self.collect_range((Unbounded, Included(threshold))))
    }

    pub fn len(&self) -> usize {
        self.entries.values().map(|v| v.len()).sum()
    }

    /// Approximate resident bytes (ADR 0073 §2). Same shape as the hash and
    /// bitmap estimators: keys plus posting lists plus a per-node constant.
    pub fn memory_bytes(&self) -> usize {
        std::mem::size_of::<Self>() + self.entry_memory_bytes
    }

    /// Range scan with early stop at `limit` entity IDs.
    /// Iterates the BTree in key order — cheaper than `range()` for LIMIT-bounded
    /// queries because it stops as soon as enough IDs are collected.
    /// Returns None when float values make ordering unsafe.
    pub fn range_limited(
        &self,
        low: CanonicalKey,
        high: CanonicalKey,
        limit: usize,
    ) -> Option<Vec<EntityId>> {
        if !self.range_enabled(low.family()) || low.family() != high.family() {
            return None;
        }
        if low > high {
            return Some(Vec::new());
        }
        Some(self.collect_range_limited(low..=high, limit))
    }

    pub fn range_limited_same_family(
        &self,
        low: CanonicalKey,
        high: CanonicalKey,
        limit: usize,
    ) -> Option<Vec<EntityId>> {
        if low.family() != high.family() {
            return None;
        }
        if low > high {
            return Some(Vec::new());
        }
        if !self.families.contains(&low.family()) {
            return Some(Vec::new());
        }
        Some(self.collect_range_limited(low..=high, limit))
    }

    /// Greater-than scan with early stop at `limit`.
    pub fn greater_than_limited(
        &self,
        threshold: CanonicalKey,
        limit: usize,
    ) -> Option<Vec<EntityId>> {
        if !self.range_enabled(threshold.family()) {
            return None;
        }
        Some(self.collect_range_limited((Excluded(threshold), Unbounded), limit))
    }

    /// Greater-or-equal scan with early stop at `limit`.
    pub fn greater_equal_limited(
        &self,
        threshold: CanonicalKey,
        limit: usize,
    ) -> Option<Vec<EntityId>> {
        if !self.range_enabled(threshold.family()) {
            return None;
        }
        Some(self.collect_range_limited((Included(threshold), Unbounded), limit))
    }

    /// Less-than scan with early stop at `limit`.
    pub fn less_than_limited(
        &self,
        threshold: CanonicalKey,
        limit: usize,
    ) -> Option<Vec<EntityId>> {
        if !self.range_enabled(threshold.family()) {
            return None;
        }
        Some(self.collect_range_limited((Unbounded, Excluded(threshold)), limit))
    }

    /// Less-or-equal scan with early stop at `limit`.
    pub fn less_equal_limited(
        &self,
        threshold: CanonicalKey,
        limit: usize,
    ) -> Option<Vec<EntityId>> {
        if !self.range_enabled(threshold.family()) {
            return None;
        }
        Some(self.collect_range_limited((Unbounded, Included(threshold)), limit))
    }

    /// IN-list multi-point lookup: performs one BTree point-lookup per value
    /// (O(log n) each) instead of a range scan — matches MongoDB's
    /// `IndexBoundsChecker` MUST_ADVANCE / multi-interval seek behaviour.
    ///
    /// `values` need not be sorted; the method sorts internally.
    /// Stops after collecting `limit` entity IDs total.
    /// Returns `None` when inexact numeric values make the index unsafe.
    pub fn in_lookup_limited(
        &self,
        values: &[CanonicalKey],
        limit: usize,
    ) -> Option<Vec<EntityId>> {
        let mut sorted_vals = values.to_vec();
        sorted_vals.sort_unstable();
        sorted_vals.dedup();

        let mut result = Vec::with_capacity(limit.min(sorted_vals.len() * 4));
        'outer: for key in &sorted_vals {
            if let Some(ids) = self.entries.get(key) {
                for &id in ids {
                    result.push(id);
                    if result.len() >= limit {
                        break 'outer;
                    }
                }
            }
        }
        Some(result)
    }

    /// Like `in_lookup_limited` but only returns IDs also present in `filter_set`.
    /// Used for bitmap-AND of a hash-index predicate + an IN-list sorted-index predicate.
    pub fn in_lookup_limited_filtered_by_set(
        &self,
        values: &[CanonicalKey],
        filter_set: &std::collections::HashSet<u64>,
        limit: usize,
    ) -> Option<Vec<EntityId>> {
        let mut sorted_vals = values.to_vec();
        sorted_vals.sort_unstable();
        sorted_vals.dedup();

        let mut result = Vec::with_capacity(limit.min(sorted_vals.len() * 4));
        'outer: for key in &sorted_vals {
            if let Some(ids) = self.entries.get(key) {
                for &id in ids {
                    if filter_set.contains(&id.raw()) {
                        result.push(id);
                        if result.len() >= limit {
                            break 'outer;
                        }
                    }
                }
            }
        }
        Some(result)
    }

    /// Covered-query projection: return the BTree **keys** (= column values) for a range.
    /// The key IS the value — no entity fetch needed for `SELECT col FROM t WHERE col op x`.
    /// Stops after `limit` distinct keys (each key may have multiple entity IDs, but for
    /// covered queries we only need the value once per distinct key).
    pub fn range_lookup_values<R>(&self, range: R, limit: usize) -> Vec<CanonicalKey>
    where
        R: std::ops::RangeBounds<CanonicalKey>,
    {
        self.entries
            .range(range)
            .take(limit)
            .map(|(key, _)| key.clone())
            .collect()
    }

    /// Covered-query projection for IN-lists: return BTree keys for the given values.
    pub fn in_lookup_values(&self, values: &[CanonicalKey], limit: usize) -> Vec<CanonicalKey> {
        let mut sorted_vals = values.to_vec();
        sorted_vals.sort_unstable();
        sorted_vals.dedup();
        let mut result = Vec::with_capacity(sorted_vals.len().min(limit));
        for key in sorted_vals {
            if self.entries.contains_key(&key) {
                result.push(key);
                if result.len() >= limit {
                    break;
                }
            }
        }
        result
    }

    fn collect_range<R>(&self, range: R) -> Vec<EntityId>
    where
        R: std::ops::RangeBounds<CanonicalKey>,
    {
        let mut result = Vec::new();
        for ids in self.entries.range(range).map(|(_, ids)| ids) {
            result.extend_from_slice(ids);
        }
        result
    }

    fn collect_range_limited<R>(&self, range: R, limit: usize) -> Vec<EntityId>
    where
        R: std::ops::RangeBounds<CanonicalKey>,
    {
        let mut result = Vec::with_capacity(limit.min(512));
        'outer: for ids in self.entries.range(range).map(|(_, ids)| ids) {
            for &id in ids {
                result.push(id);
                if result.len() >= limit {
                    break 'outer;
                }
            }
        }
        result
    }

    /// Bitmap AND: iterate the sorted range and collect only IDs present in
    /// `filter_set` (a hash-index candidate set from an equality predicate).
    /// Stops after `limit` results — matching PG's bitmap heap scan behaviour
    /// where the intersection is fetched in physical order.
    ///
    /// This avoids fetching ALL hash-index candidates when the range predicate
    /// is highly selective (e.g. `city='X'` ∩ `age > 30` → ~1K not 50K).
    /// Returns None when float values make the ordering unsafe.
    pub fn collect_range_filtered_by_set<R>(
        &self,
        range: R,
        filter_set: &std::collections::HashSet<u64>,
        limit: usize,
    ) -> Option<Vec<EntityId>>
    where
        R: std::ops::RangeBounds<CanonicalKey>,
    {
        if self.has_mixed_families {
            return None;
        }
        let mut result = Vec::new();
        'outer: for ids in self.entries.range(range).map(|(_, ids)| ids) {
            for &id in ids {
                if filter_set.contains(&id.raw()) {
                    result.push(id);
                    if result.len() >= limit {
                        break 'outer;
                    }
                }
            }
        }
        Some(result)
    }
}

/// In-memory sorted index over a TUPLE of canonical keys — the composite
/// index used to accelerate `WHERE col_a = X AND col_b > Y LIMIT N` style
/// queries where a single-column index would force a post-filter scan.
///
/// Ordering is the lexicographic ordering of the underlying `Vec<CanonicalKey>`
/// (Rust's derived `Ord` on `Vec` compares element-by-element), which
/// matches how PostgreSQL's multi-column B-tree behaves for prefix-seek
/// + trailing-range queries.
pub struct SortedCompositeIndex {
    entries: BTreeMap<Vec<CanonicalKey>, Vec<EntityId>>,
    entry_memory_bytes: usize,
}

impl SortedCompositeIndex {
    pub fn new() -> Self {
        Self {
            entries: BTreeMap::new(),
            entry_memory_bytes: 0,
        }
    }

    pub fn insert(&mut self, key: Vec<CanonicalKey>, entity_id: EntityId) {
        let entry = self.entries.entry(key);
        if let std::collections::btree_map::Entry::Vacant(ref entry) = entry {
            self.entry_memory_bytes += btree_entry_bytes(entry.key().iter(), 0);
        }
        self.entry_memory_bytes += std::mem::size_of::<EntityId>();
        entry.or_default().push(entity_id);
    }

    pub fn len(&self) -> usize {
        self.entries.values().map(|v| v.len()).sum()
    }

    fn insert_fields(
        &mut self,
        columns: &[String],
        entity_id: EntityId,
        fields: &[(String, Value)],
    ) {
        let mut tuple = Vec::with_capacity(columns.len());
        for column in columns {
            let Some((_, value)) = fields.iter().find(|(name, _)| name == column) else {
                return;
            };
            let CanonicalizedValue::Exact(key) = classify_sorted_value(value) else {
                return;
            };
            tuple.push(key);
        }
        self.insert(tuple, entity_id);
    }

    /// Approximate resident bytes (ADR 0073 §2). Every column of the composite
    /// key is charged, so a wide index costs what it weighs.
    pub fn memory_bytes(&self) -> usize {
        std::mem::size_of::<Self>() + self.entry_memory_bytes
    }

    /// Range scan with an exact prefix on the first `prefix.len()` columns
    /// and a range on the column at index `prefix.len()`. Returns up to
    /// `limit` entity IDs in key order.
    ///
    /// Preconditions — caller must ensure:
    /// - `prefix.len()` is < composite arity (strictly a prefix)
    /// - `low.family() == high.family()` so range semantics are well-defined
    pub fn prefix_range(
        &self,
        prefix: &[CanonicalKey],
        low: CanonicalKey,
        high: CanonicalKey,
        limit: usize,
    ) -> Vec<EntityId> {
        if limit == 0 {
            return Vec::new();
        }
        let mut low_key = prefix.to_vec();
        low_key.push(low);
        let mut high_key = prefix.to_vec();
        high_key.push(high);
        let mut out = Vec::with_capacity(limit.min(128));
        for (_, ids) in self.entries.range(low_key..=high_key) {
            for id in ids {
                out.push(*id);
                if out.len() >= limit {
                    return out;
                }
            }
        }
        out
    }

    /// Exact-prefix equality scan: returns entity IDs for every key
    /// whose first `prefix.len()` components match `prefix` exactly.
    /// Used by `WHERE col_a = X AND col_b = Y` on a `(col_a, col_b)`
    /// composite — treated as a prefix-only seek with no trailing range.
    pub fn prefix_eq(&self, prefix: &[CanonicalKey], limit: usize) -> Vec<EntityId> {
        if limit == 0 || prefix.is_empty() {
            return Vec::new();
        }
        // Low = prefix itself; high = prefix with one extra component that
        // is guaranteed to be > than any "same-prefix" key. We use the
        // half-open upper bound trick with BTreeMap::range(low..upper).
        let low = prefix.to_vec();
        let mut out = Vec::with_capacity(limit.min(128));
        for (k, ids) in self.entries.range(low.clone()..) {
            if k.len() < prefix.len() || &k[..prefix.len()] != prefix {
                break;
            }
            for id in ids {
                out.push(*id);
                if out.len() >= limit {
                    return out;
                }
            }
        }
        out
    }
}

/// Manages sorted column indices per (collection, column).
pub struct SortedIndexManager {
    indices: RwLock<HashMap<(String, String), SortedColumnIndex>>,
    composite: RwLock<HashMap<(String, Vec<String>), SortedCompositeIndex>>,
}

impl SortedIndexManager {
    pub fn new() -> Self {
        Self {
            indices: RwLock::new(HashMap::new()),
            composite: RwLock::new(HashMap::new()),
        }
    }

    /// Approximate resident bytes across every sorted and composite index —
    /// this manager's contribution to the `index_memory` pool (ADR 0073 §2).
    pub fn memory_bytes(&self) -> u64 {
        let single: usize = read_unpoisoned(&self.indices)
            .values()
            .map(SortedColumnIndex::memory_bytes)
            .sum();
        let composite: usize = read_unpoisoned(&self.composite)
            .values()
            .map(SortedCompositeIndex::memory_bytes)
            .sum();
        (single + composite) as u64
    }

    /// Build a composite (multi-column) sorted index from existing entities.
    /// Every entity must expose a value for every listed column; entities
    /// missing any one column are skipped.
    pub fn build_composite(
        &self,
        collection: &str,
        columns: &[String],
        entities: &[(EntityId, Vec<(String, Value)>)],
    ) -> usize {
        let mut index = SortedCompositeIndex::new();
        let mut count = 0;
        'entity: for (eid, fields) in entities {
            let mut tuple: Vec<CanonicalKey> = Vec::with_capacity(columns.len());
            for col in columns {
                let found = fields.iter().find(|(name, _)| name == col);
                let key = match found {
                    Some((_, val)) => match classify_sorted_value(val) {
                        CanonicalizedValue::Exact(k) => k,
                        CanonicalizedValue::Unsupported => continue 'entity,
                    },
                    None => continue 'entity,
                };
                tuple.push(key);
            }
            index.insert(tuple, *eid);
            count += 1;
        }
        write_unpoisoned(&self.composite).insert((collection.to_string(), columns.to_vec()), index);
        count
    }

    pub fn has_composite_index(&self, collection: &str, columns: &[String]) -> bool {
        let key = (collection.to_string(), columns.to_vec());
        read_unpoisoned(&self.composite).contains_key(&key)
    }

    /// Composite prefix-eq + trailing range lookup.
    /// `columns` must match a previously-built composite index exactly.
    pub fn composite_prefix_range_lookup(
        &self,
        collection: &str,
        columns: &[String],
        prefix: &[CanonicalKey],
        low: CanonicalKey,
        high: CanonicalKey,
        limit: usize,
    ) -> Option<Vec<EntityId>> {
        let guard = read_unpoisoned(&self.composite);
        let idx = guard.get(&(collection.to_string(), columns.to_vec()))?;
        Some(idx.prefix_range(prefix, low, high, limit))
    }

    /// Composite exact prefix lookup.
    pub fn composite_prefix_eq_lookup(
        &self,
        collection: &str,
        columns: &[String],
        prefix: &[CanonicalKey],
        limit: usize,
    ) -> Option<Vec<EntityId>> {
        let guard = read_unpoisoned(&self.composite);
        let idx = guard.get(&(collection.to_string(), columns.to_vec()))?;
        Some(idx.prefix_eq(prefix, limit))
    }

    /// Update a composite index when an entity's fields change. Drops the
    /// old tuple and inserts the new one when any of the listed columns
    /// differs between `old_fields` and `new_fields`.
    pub fn composite_entity_update(
        &self,
        collection: &str,
        columns: &[String],
        entity_id: EntityId,
        old_fields: &[(String, Value)],
        new_fields: &[(String, Value)],
    ) {
        let build_tuple = |fields: &[(String, Value)]| -> Option<Vec<CanonicalKey>> {
            let mut tuple = Vec::with_capacity(columns.len());
            for col in columns {
                let val = fields
                    .iter()
                    .find(|(name, _)| name == col)
                    .map(|(_, v)| v)?;
                match classify_sorted_value(val) {
                    CanonicalizedValue::Exact(k) => tuple.push(k),
                    CanonicalizedValue::Unsupported => return None,
                }
            }
            Some(tuple)
        };
        let old_tuple = build_tuple(old_fields);
        let new_tuple = build_tuple(new_fields);
        if old_tuple == new_tuple {
            return;
        }
        let mut guard = write_unpoisoned(&self.composite);
        let idx = match guard.get_mut(&(collection.to_string(), columns.to_vec())) {
            Some(i) => i,
            None => return,
        };
        if let Some(old) = old_tuple {
            if let Some(ids) = idx.entries.get_mut(&old) {
                let before = ids.len();
                ids.retain(|id| *id != entity_id);
                idx.entry_memory_bytes -= (before - ids.len()) * std::mem::size_of::<EntityId>();
                if ids.is_empty() {
                    idx.entry_memory_bytes -= btree_entry_bytes(old.iter(), 0);
                    idx.entries.remove(&old);
                }
            }
        }
        if let Some(new) = new_tuple {
            idx.insert(new, entity_id);
        }
    }

    /// Insert a single entity into all composite indexes registered on
    /// `collection`. Called from the entity-insert hot path alongside
    /// the existing single-column index maintenance.
    pub fn composite_entity_insert(
        &self,
        collection: &str,
        entity_id: EntityId,
        fields: &[(String, Value)],
    ) {
        let mut guard = write_unpoisoned(&self.composite);
        for ((coll, cols), idx) in guard.iter_mut() {
            if coll != collection {
                continue;
            }
            idx.insert_fields(cols, entity_id, fields);
        }
    }

    /// Maintain only the composite selected by the registry entry. Updating
    /// every composite here once per entry duplicates postings and memory.
    fn composite_insert_one(
        &self,
        collection: &str,
        columns: &[String],
        entity_id: EntityId,
        fields: &[(String, Value)],
    ) {
        let mut guard = self.composite.write();
        // Borrow the existing key; do not allocate collection/column copies
        // just to address an index in the single-row path.
        if let Some((_, index)) = guard
            .iter_mut()
            .find(|((name, indexed_columns), _)| name == collection && indexed_columns == columns)
        {
            index.insert_fields(columns, entity_id, fields);
        }
    }

    /// Batch insert rows into one composite index. Keeps the registry-level
    /// batch path from falling through to the single-column BTree branch for
    /// `(a, b, ...)` indexes.
    pub fn composite_insert_batch(
        &self,
        collection: &str,
        columns: &[String],
        rows: &[(EntityId, Vec<(String, Value)>)],
    ) {
        let mut guard = write_unpoisoned(&self.composite);
        let Some(idx) = guard.get_mut(&(collection.to_string(), columns.to_vec())) else {
            return;
        };

        for (entity_id, fields) in rows {
            idx.insert_fields(columns, *entity_id, fields);
        }
    }

    /// Build a sorted index from existing entities.
    pub fn build_index(
        &self,
        collection: &str,
        column: &str,
        entities: &[(EntityId, Vec<(String, Value)>)],
    ) -> usize {
        let mut index = SortedColumnIndex::new();
        let mut count = 0;
        for (eid, fields) in entities {
            for (col, val) in fields {
                if col == column {
                    match classify_sorted_value(val) {
                        CanonicalizedValue::Exact(key) => {
                            index.insert(key, *eid);
                            count += 1;
                        }
                        CanonicalizedValue::Unsupported => {}
                    }
                }
            }
        }
        write_unpoisoned(&self.indices).insert((collection.to_string(), column.to_string()), index);
        count
    }

    pub fn entry_count(&self, collection: &str, columns: &[String]) -> Option<usize> {
        if columns.len() > 1 {
            let guard = read_unpoisoned(&self.composite);
            return guard
                .get(&(collection.to_string(), columns.to_vec()))
                .map(SortedCompositeIndex::len);
        }

        let column = columns.first()?;
        let guard = read_unpoisoned(&self.indices);
        guard
            .get(&(collection.to_string(), column.clone()))
            .map(SortedColumnIndex::len)
    }

    /// Range lookup.
    pub(crate) fn range_lookup(
        &self,
        collection: &str,
        column: &str,
        low: CanonicalKey,
        high: CanonicalKey,
    ) -> Option<Vec<EntityId>> {
        let indices = read_unpoisoned(&self.indices);
        let key = (collection.to_string(), column.to_string());
        match indices.get(&key) {
            Some(index) => index.range(low, high),
            None => None,
        }
    }

    /// Greater-than lookup.
    pub(crate) fn gt_lookup(
        &self,
        collection: &str,
        column: &str,
        threshold: CanonicalKey,
    ) -> Option<Vec<EntityId>> {
        let indices = read_unpoisoned(&self.indices);
        let key = (collection.to_string(), column.to_string());
        match indices.get(&key) {
            Some(index) => index.greater_than(threshold),
            None => None,
        }
    }

    pub(crate) fn ge_lookup(
        &self,
        collection: &str,
        column: &str,
        threshold: CanonicalKey,
    ) -> Option<Vec<EntityId>> {
        let indices = read_unpoisoned(&self.indices);
        let key = (collection.to_string(), column.to_string());
        match indices.get(&key) {
            Some(index) => index.greater_equal(threshold),
            None => None,
        }
    }

    pub(crate) fn lt_lookup(
        &self,
        collection: &str,
        column: &str,
        threshold: CanonicalKey,
    ) -> Option<Vec<EntityId>> {
        let indices = read_unpoisoned(&self.indices);
        let key = (collection.to_string(), column.to_string());
        match indices.get(&key) {
            Some(index) => index.less_than(threshold),
            None => None,
        }
    }

    pub(crate) fn le_lookup(
        &self,
        collection: &str,
        column: &str,
        threshold: CanonicalKey,
    ) -> Option<Vec<EntityId>> {
        let indices = read_unpoisoned(&self.indices);
        let key = (collection.to_string(), column.to_string());
        match indices.get(&key) {
            Some(index) => index.less_equal(threshold),
            None => None,
        }
    }

    /// Range lookup with early stop at `limit` — avoids collecting all IDs
    /// when only the first N results are needed (LIMIT-bounded queries).
    pub(crate) fn range_lookup_limited(
        &self,
        collection: &str,
        column: &str,
        low: CanonicalKey,
        high: CanonicalKey,
        limit: usize,
    ) -> Option<Vec<EntityId>> {
        let indices = read_unpoisoned(&self.indices);
        let key = (collection.to_string(), column.to_string());
        indices.get(&key)?.range_limited(low, high, limit)
    }

    pub(crate) fn range_lookup_limited_same_family(
        &self,
        collection: &str,
        column: &str,
        low: CanonicalKey,
        high: CanonicalKey,
        limit: usize,
    ) -> Option<Vec<EntityId>> {
        let indices = read_unpoisoned(&self.indices);
        let key = (collection.to_string(), column.to_string());
        indices
            .get(&key)?
            .range_limited_same_family(low, high, limit)
    }

    pub(crate) fn gt_lookup_limited(
        &self,
        collection: &str,
        column: &str,
        threshold: CanonicalKey,
        limit: usize,
    ) -> Option<Vec<EntityId>> {
        let indices = read_unpoisoned(&self.indices);
        let key = (collection.to_string(), column.to_string());
        indices.get(&key)?.greater_than_limited(threshold, limit)
    }

    pub(crate) fn ge_lookup_limited(
        &self,
        collection: &str,
        column: &str,
        threshold: CanonicalKey,
        limit: usize,
    ) -> Option<Vec<EntityId>> {
        let indices = read_unpoisoned(&self.indices);
        let key = (collection.to_string(), column.to_string());
        indices.get(&key)?.greater_equal_limited(threshold, limit)
    }

    pub(crate) fn lt_lookup_limited(
        &self,
        collection: &str,
        column: &str,
        threshold: CanonicalKey,
        limit: usize,
    ) -> Option<Vec<EntityId>> {
        let indices = read_unpoisoned(&self.indices);
        let key = (collection.to_string(), column.to_string());
        indices.get(&key)?.less_than_limited(threshold, limit)
    }

    pub(crate) fn le_lookup_limited(
        &self,
        collection: &str,
        column: &str,
        threshold: CanonicalKey,
        limit: usize,
    ) -> Option<Vec<EntityId>> {
        let indices = read_unpoisoned(&self.indices);
        let key = (collection.to_string(), column.to_string());
        indices.get(&key)?.less_equal_limited(threshold, limit)
    }

    /// Bitmap AND: range [low, high] filtered to IDs in `filter_set`.
    pub(crate) fn range_filtered_by_set(
        &self,
        collection: &str,
        column: &str,
        low: CanonicalKey,
        high: CanonicalKey,
        filter_set: &std::collections::HashSet<u64>,
        limit: usize,
    ) -> Option<Vec<EntityId>> {
        let indices = read_unpoisoned(&self.indices);
        let key = (collection.to_string(), column.to_string());
        if low > high {
            return Some(Vec::new());
        }
        let idx = indices.get(&key)?;
        if !idx.supports_range_key(&low) || low.family() != high.family() {
            return None;
        }
        idx.collect_range_filtered_by_set(low..=high, filter_set, limit)
    }

    /// Bitmap AND: gt filtered to IDs in `filter_set`.
    pub(crate) fn gt_filtered_by_set(
        &self,
        collection: &str,
        column: &str,
        threshold: CanonicalKey,
        filter_set: &std::collections::HashSet<u64>,
        limit: usize,
    ) -> Option<Vec<EntityId>> {
        let indices = read_unpoisoned(&self.indices);
        let key = (collection.to_string(), column.to_string());
        let idx = indices.get(&key)?;
        if !idx.supports_range_key(&threshold) {
            return None;
        }
        idx.collect_range_filtered_by_set((Excluded(threshold), Unbounded), filter_set, limit)
    }

    /// Bitmap AND: ge filtered to IDs in `filter_set`.
    pub(crate) fn ge_filtered_by_set(
        &self,
        collection: &str,
        column: &str,
        threshold: CanonicalKey,
        filter_set: &std::collections::HashSet<u64>,
        limit: usize,
    ) -> Option<Vec<EntityId>> {
        let indices = read_unpoisoned(&self.indices);
        let key = (collection.to_string(), column.to_string());
        let idx = indices.get(&key)?;
        if !idx.supports_range_key(&threshold) {
            return None;
        }
        idx.collect_range_filtered_by_set((Included(threshold), Unbounded), filter_set, limit)
    }

    /// Bitmap AND: lt filtered to IDs in `filter_set`.
    pub(crate) fn lt_filtered_by_set(
        &self,
        collection: &str,
        column: &str,
        threshold: CanonicalKey,
        filter_set: &std::collections::HashSet<u64>,
        limit: usize,
    ) -> Option<Vec<EntityId>> {
        let indices = read_unpoisoned(&self.indices);
        let key = (collection.to_string(), column.to_string());
        let idx = indices.get(&key)?;
        if !idx.supports_range_key(&threshold) {
            return None;
        }
        idx.collect_range_filtered_by_set((Unbounded, Excluded(threshold)), filter_set, limit)
    }

    /// Bitmap AND: le filtered to IDs in `filter_set`.
    pub(crate) fn le_filtered_by_set(
        &self,
        collection: &str,
        column: &str,
        threshold: CanonicalKey,
        filter_set: &std::collections::HashSet<u64>,
        limit: usize,
    ) -> Option<Vec<EntityId>> {
        let indices = read_unpoisoned(&self.indices);
        let key = (collection.to_string(), column.to_string());
        let idx = indices.get(&key)?;
        if !idx.supports_range_key(&threshold) {
            return None;
        }
        idx.collect_range_filtered_by_set((Unbounded, Included(threshold)), filter_set, limit)
    }

    /// IN-list multi-point lookup on a sorted index.
    /// Performs one BTree point-lookup per value — O(k log n) for k values
    /// instead of O(n) for a range scan covering all values.
    /// Stops after `limit` total entity IDs.
    pub(crate) fn in_lookup_limited(
        &self,
        collection: &str,
        column: &str,
        values: &[CanonicalKey],
        limit: usize,
    ) -> Option<Vec<EntityId>> {
        let indices = read_unpoisoned(&self.indices);
        let key = (collection.to_string(), column.to_string());
        indices.get(&key)?.in_lookup_limited(values, limit)
    }

    /// IN-list multi-point lookup filtered by a hash-index candidate set.
    /// Bitmap AND: sorted-index point lookups ∩ hash-index set.
    pub(crate) fn in_lookup_limited_filtered_by_set(
        &self,
        collection: &str,
        column: &str,
        values: &[CanonicalKey],
        filter_set: &std::collections::HashSet<u64>,
        limit: usize,
    ) -> Option<Vec<EntityId>> {
        let indices = read_unpoisoned(&self.indices);
        let key = (collection.to_string(), column.to_string());
        indices
            .get(&key)?
            .in_lookup_limited_filtered_by_set(values, filter_set, limit)
    }

    /// Covered-query range projection: return BTree keys (= column values) for a range.
    /// No entity fetch required — the BTree key IS the column value.
    pub(crate) fn range_lookup_values(
        &self,
        collection: &str,
        column: &str,
        low: CanonicalKey,
        high: CanonicalKey,
        limit: usize,
    ) -> Option<Vec<CanonicalKey>> {
        let indices = read_unpoisoned(&self.indices);
        let key = (collection.to_string(), column.to_string());
        let idx = indices.get(&key)?;
        if !idx.supports_range_key(&low) || low.family() != high.family() {
            return None;
        }
        Some(idx.range_lookup_values((Included(low), Included(high)), limit))
    }

    /// Covered-query gt/ge/lt/le projection.
    pub(crate) fn compare_lookup_values(
        &self,
        collection: &str,
        column: &str,
        threshold: CanonicalKey,
        op: &reddb_rql::ast::CompareOp,
        limit: usize,
    ) -> Option<Vec<CanonicalKey>> {
        use reddb_rql::ast::CompareOp;
        let indices = read_unpoisoned(&self.indices);
        let key = (collection.to_string(), column.to_string());
        let idx = indices.get(&key)?;
        if !idx.supports_range_key(&threshold) {
            return None;
        }
        let values = match op {
            CompareOp::Gt => idx.range_lookup_values((Excluded(threshold), Unbounded), limit),
            CompareOp::Ge => idx.range_lookup_values((Included(threshold), Unbounded), limit),
            CompareOp::Lt => idx.range_lookup_values((Unbounded, Excluded(threshold)), limit),
            CompareOp::Le => idx.range_lookup_values((Unbounded, Included(threshold)), limit),
            _ => return None,
        };
        Some(values)
    }

    /// Covered-query IN-list projection.
    pub(crate) fn in_lookup_values(
        &self,
        collection: &str,
        column: &str,
        values: &[CanonicalKey],
        limit: usize,
    ) -> Option<Vec<CanonicalKey>> {
        let indices = read_unpoisoned(&self.indices);
        let key = (collection.to_string(), column.to_string());
        Some(indices.get(&key)?.in_lookup_values(values, limit))
    }

    /// Check if a sorted index exists for a column.
    pub fn has_index(&self, collection: &str, column: &str) -> bool {
        let indices = read_unpoisoned(&self.indices);
        indices.contains_key(&(collection.to_string(), column.to_string()))
    }

    pub fn supports_mixed_integral_ranges(&self, collection: &str, column: &str) -> bool {
        let indices = read_unpoisoned(&self.indices);
        indices
            .get(&(collection.to_string(), column.to_string()))
            .is_some_and(SortedColumnIndex::supports_mixed_integral_ranges)
    }

    /// Insert one value into an existing sorted index.
    /// No-op if the index hasn't been created yet — the next
    /// `build_index` or `create_index` call will pick up the entity on
    /// its full scan.
    pub(crate) fn insert_one(
        &self,
        collection: &str,
        column: &str,
        value: &Value,
        entity_id: EntityId,
    ) {
        let mut indices = write_unpoisoned(&self.indices);
        let k = (collection.to_string(), column.to_string());
        if let Some(index) = indices.get_mut(&k) {
            match classify_sorted_value(value) {
                CanonicalizedValue::Exact(key) => index.insert(key, entity_id),
                CanonicalizedValue::Unsupported => {}
            }
        }
    }

    /// Remove a single `entity_id` from the index. Linear in the
    /// number of entries at that key — fine for the benchmark's low
    /// per-key cardinality (age has ~200 buckets, city ~50).
    pub(crate) fn delete_one(
        &self,
        collection: &str,
        column: &str,
        value: &Value,
        entity_id: EntityId,
    ) {
        let mut indices = write_unpoisoned(&self.indices);
        let k = (collection.to_string(), column.to_string());
        if let Some(index) = indices.get_mut(&k) {
            let Some(key) = value_to_sorted_key(value) else {
                return;
            };
            if let Some(bucket) = index.entries.get_mut(&key) {
                let before = bucket.len();
                bucket.retain(|id| *id != entity_id);
                index.entry_memory_bytes -=
                    (before - bucket.len()) * std::mem::size_of::<EntityId>();
                if bucket.is_empty() {
                    index.entry_memory_bytes -= btree_entry_bytes(std::iter::once(&key), 0);
                    index.entries.remove(&key);
                }
            }
        }
    }
}

fn classify_sorted_value(val: &Value) -> CanonicalizedValue {
    match value_to_canonical_key(val) {
        Some(key) => CanonicalizedValue::Exact(key),
        None => CanonicalizedValue::Unsupported,
    }
}

pub(crate) fn value_to_sorted_key(val: &Value) -> Option<CanonicalKey> {
    match classify_sorted_value(val) {
        CanonicalizedValue::Exact(key) => Some(key),
        CanonicalizedValue::Unsupported => None,
    }
}

/// Convert a canonical sorted key back to a `Value` for covered-query projection.
pub(crate) fn sorted_key_to_value(key: CanonicalKey) -> Value {
    key.into_value()
}

/// Metadata about a registered index
#[derive(Debug, Clone)]
pub struct RegisteredIndex {
    pub name: String,
    pub collection: String,
    pub columns: Vec<String>,
    pub method: IndexMethodKind,
    pub unique: bool,
}

impl RegisteredIndex {
    /// Name under which this index is registered in the hash store.
    /// Pure Hash indexes use their own name; single-col BTree indexes
    /// store an auxiliary hash side-pocket as `{name}_hash` (see
    /// `create_index` for BTree at lines 1199–1216) — callers must use
    /// that suffix or `hash_lookup` returns `MissingIndex`.
    pub fn hash_lookup_name(&self) -> std::borrow::Cow<'_, str> {
        match self.method {
            IndexMethodKind::Hash => std::borrow::Cow::Borrowed(self.name.as_str()),
            IndexMethodKind::BTree => std::borrow::Cow::Owned(format!("{}_hash", self.name)),
            IndexMethodKind::Bitmap | IndexMethodKind::H3 { .. } => {
                std::borrow::Cow::Borrowed(self.name.as_str())
            }
        }
    }

    /// Multi-column key pocket in the hash store: a composite HASH index, or a
    /// composite UNIQUE BTREE index (whose `sorted` half cannot be probed by
    /// the byte key admission uses). Plain composite BTREE indexes have none.
    fn composite_key_pocket(&self) -> Option<std::borrow::Cow<'_, str>> {
        let has_pocket = match self.method {
            IndexMethodKind::Hash => true,
            IndexMethodKind::BTree => self.unique,
            IndexMethodKind::Bitmap | IndexMethodKind::H3 { .. } => false,
        };
        (has_pocket && self.columns.len() > 1).then(|| self.hash_lookup_name())
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum IndexMethodKind {
    Hash,
    Bitmap,
    BTree,
    /// H3 spatial index: the geo column's `(lat, lon)` is encoded to a
    /// single H3 cell-id and stored in the sorted (disk-paged B-tree)
    /// index, exactly like a single-column `BTree` index but with an
    /// integer cell key in place of the raw column value. `resolution`
    /// is the H3 grid level (0..=15). RAM stays O(working set), not
    /// O(points) — no separate resident structure (PRD #1574, #1576).
    H3 {
        resolution: u8,
    },
}

impl IndexMethodKind {
    /// Resolve a *generic* spatial index request (RQL `USING SPATIAL`, or a
    /// bare spatial index) to the engine's default spatial backend.
    ///
    /// As of PRD #1574 slice 4 (#1578) the default is the disk-resident
    /// [`IndexMethodKind::H3`] index: `(lat, lon)` is encoded to a single
    /// H3 cell-id `u64` and stored in the existing paged B-tree, so RAM
    /// stays O(working set) rather than O(total points). The old in-RAM
    /// R-tree backend was retired; `USING RTREE` is now a didactic error.
    pub fn default_spatial() -> Self {
        IndexMethodKind::H3 {
            resolution: DEFAULT_H3_RESOLUTION,
        }
    }
}

/// Default H3 resolution used when reconstructing an H3 index kind from
/// the incremental-maintainer mirror, which does not carry the
/// resolution. Mirrors `reddb_rql::ast::DEFAULT_H3_RESOLUTION`; the
/// authoritative resolution is always persisted in the index registry,
/// so this default is only a structural placeholder.
pub(crate) const DEFAULT_H3_RESOLUTION: u8 = 9;

#[cfg(test)]
type MutationTestHook = std::sync::Arc<dyn Fn(&str, MutationTestPhase) + Send + Sync>;

#[cfg(test)]
#[derive(Clone, Copy, PartialEq, Eq)]
pub(crate) enum MutationTestPhase {
    StoragePublished,
    BeforeEvents,
}

/// Unified index store aggregating all secondary index managers.
pub struct IndexStore {
    pub hash: HashIndexManager,
    pub bitmap: BitmapIndexManager,
    pub sorted: SortedIndexManager,
    /// Registry of all created indices: (collection, index_name) → metadata
    registry: RwLock<HashMap<(String, String), RegisteredIndex>>,
    // Keep identities stable across DROP/recreate, including queued writers.
    // The memory sampler never takes a topology lock.
    topology: RwLock<HashMap<String, std::sync::Arc<RwLock<()>>>>,
    #[cfg(test)]
    pub(super) before_build: parking_lot::Mutex<Option<std::sync::Arc<dyn Fn() + Send + Sync>>>,
    #[cfg(test)]
    pub(super) mutation_hook: parking_lot::Mutex<Option<MutationTestHook>>,
}

impl IndexStore {
    pub fn new() -> Self {
        Self {
            hash: HashIndexManager::new(),
            bitmap: BitmapIndexManager::new(),
            sorted: SortedIndexManager::new(),
            registry: RwLock::new(HashMap::new()),
            topology: RwLock::new(HashMap::new()),
            #[cfg(test)]
            before_build: parking_lot::Mutex::new(None),
            #[cfg(test)]
            mutation_hook: parking_lot::Mutex::new(None),
        }
    }

    #[cfg(test)]
    pub(crate) fn mutation_test_hook(&self, collection: &str, phase: MutationTestPhase) {
        let hook = self.mutation_hook.lock().clone();
        if let Some(hook) = hook {
            hook(collection, phase);
        }
    }

    /// Shared across row admission and maintenance; exclusive across topology
    /// changes, including the snapshot used to backfill. Release before events.
    /// Lock order: row constraint -> topology -> reservation -> storage/index.
    pub(crate) fn collection_topology_lock(&self, collection: &str) -> std::sync::Arc<RwLock<()>> {
        if let Some(lock) = self.topology.read().get(collection) {
            return std::sync::Arc::clone(lock);
        }
        std::sync::Arc::clone(
            self.topology
                .write()
                .entry(collection.to_string())
                .or_insert_with(|| std::sync::Arc::new(RwLock::new(()))),
        )
    }

    pub(crate) fn needs_auto_id_index(
        &self,
        collection: &str,
        fields: &[(String, Value)],
        enabled: bool,
    ) -> bool {
        enabled
            && fields.iter().any(|(name, _)| name == "id")
            && !self.registry.read().values().any(|index| {
                index.collection == collection
                    && index.columns.len() == 1
                    && index.columns[0] == "id"
            })
    }

    /// Approximate resident bytes across every in-RAM secondary index — the
    /// `index_memory` pool's live usage (ADR 0073 §2).
    ///
    /// Includes the sorted backing of BTree/H3, composite BTree tuples and
    /// the auxiliary equality hash of single-column BTree indexes.
    pub fn memory_bytes(&self) -> u64 {
        self.hash
            .memory_bytes()
            .saturating_add(self.bitmap.memory_bytes())
            .saturating_add(self.sorted.memory_bytes())
    }

    /// Estimate resident index growth before row mutation, without cloning
    /// rows or the registry. Each potential new key receives entry overhead;
    /// existing postings can therefore make this deliberately conservative.
    /// The caller holds the collection topology guard through storage mutation;
    /// no backend or registry guard survives into memory admission.
    pub(crate) fn estimate_insert_growth<'a>(
        &self,
        collection: &str,
        rows: impl Iterator<Item = &'a [(String, Value)]> + Clone,
        auto_index_id: bool,
    ) -> u64 {
        let registry = self.registry.read();
        let indexes = registry
            .values()
            .filter(|index| index.collection == collection);
        // Match the existing first-row auto-index trigger and its single-column
        // coverage rule. A composite containing id does not suppress idx_id.
        let needs_auto_id = auto_index_id
            && rows
                .clone()
                .next()
                .is_some_and(|fields| fields.iter().any(|(name, _)| name == "id"))
            && !indexes
                .clone()
                .any(|index| index.columns.len() == 1 && index.columns[0] == "id");
        let mut growth = 0u64;
        if needs_auto_id {
            growth = std::mem::size_of::<crate::storage::unified::hash_index::HashIndex>() as u64;
            for fields in rows.clone() {
                if let Some(value) = index_field_value(fields, "id") {
                    growth = growth.saturating_add(estimate_index_entry_bytes(value.as_ref()));
                }
            }
        }
        for index in indexes {
            for fields in rows.clone() {
                growth = growth.saturating_add(estimate_registered_index_growth(index, fields));
            }
        }
        growth
    }

    /// Reserve a new physical version's entries, or changed in-place keys.
    /// Replaced keys are not credited before publication: other writers must
    /// not spend space whose removal has not completed yet.
    pub(crate) fn estimate_update_growth(
        &self,
        collection: &str,
        before: &[(String, Value)],
        entity: &crate::storage::UnifiedEntity,
        versioned: bool,
    ) -> u64 {
        let registry = self.registry.read();
        let mut indexes = registry
            .values()
            .filter(|index| index.collection == collection)
            .peekable();
        if indexes.peek().is_none() {
            return 0;
        }
        let crate::storage::EntityData::Row(row) = &entity.data else {
            return 0;
        };
        let after: Vec<_> = row
            .iter_fields()
            .map(|(name, value)| (name.to_string(), value.clone()))
            .collect();
        indexes
            .filter(|index| {
                versioned
                    || index.columns.iter().any(|column| {
                        index_field_value(before, column) != index_field_value(&after, column)
                    })
            })
            .map(|index| estimate_registered_index_growth(index, &after))
            .fold(0, u64::saturating_add)
    }

    /// Build physical backing. Runtime callers hold exclusive collection topology
    /// from before collecting entities until after registering metadata.
    pub fn create_index(
        &self,
        name: &str,
        collection: &str,
        columns: &[String],
        method: IndexMethodKind,
        unique: bool,
        entities: &[(EntityId, Vec<(String, Value)>)],
    ) -> Result<usize, String> {
        #[cfg(test)]
        if let Some(hook) = self.before_build.lock().clone() {
            hook();
        }
        let col = columns.first().map(|s| s.as_str()).unwrap_or("");
        if columns.len() > 1 && columns.iter().any(|column| column.contains('.')) {
            return Err(
                "document path indexes currently support a single indexed column".to_string(),
            );
        }

        match method {
            IndexMethodKind::Hash => {
                self.hash
                    .create_index(&HashIndexConfig {
                        name: name.to_string(),
                        collection: collection.to_string(),
                        columns: columns.to_vec(),
                        unique,
                    })
                    .map_err(|e| e.to_string())?;

                // Index existing entities
                let mut count = 0;
                for (entity_id, fields) in entities {
                    let key = if unique {
                        unique_index_key(columns, fields)
                    } else {
                        hash_index_key(columns, fields)
                    };
                    if let Some(key) = key {
                        self.hash
                            .insert(collection, name, key, *entity_id)
                            .map_err(|err| err.to_string())?;
                        count += 1;
                    }
                }
                Ok(count)
            }
            IndexMethodKind::Bitmap => {
                self.bitmap.create_index(collection, col);

                let mut count = 0;
                for (entity_id, fields) in entities {
                    if let Some(value) = index_field_value(fields, col) {
                        let key = value_to_bytes(value.as_ref());
                        self.bitmap
                            .insert(collection, col, *entity_id, &key)
                            .map_err(|err| err.to_string())?;
                        count += 1;
                    }
                }
                Ok(count)
            }
            IndexMethodKind::H3 { resolution } => {
                // Encode each existing row's geo column to its H3 cell-id
                // and build a sorted (disk-paged B-tree) index over those
                // u64 keys — identical machinery to a single-column BTree,
                // only the stored key is the cell-id rather than the raw
                // GeoPoint. No resident HashMap / R-tree is allocated.
                let cell_entities = derive_h3_cell_entities(entities, col, resolution);
                let count = self.sorted.build_index(collection, col, &cell_entities);
                Ok(count)
            }
            IndexMethodKind::BTree => {
                // Multi-column BTree → composite sorted index.
                // `CREATE INDEX idx_cc ON users (city, age) USING BTREE`
                // registers a `Vec<CanonicalKey>` → ids BTreeMap so
                // `WHERE city='NYC' AND age>30` can seek prefix + range
                // instead of intersecting two single-col results.
                if columns.len() > 1 {
                    let count = self.sorted.build_composite(collection, columns, entities);
                    if unique {
                        // Admission probes the byte key, not the sorted tuple
                        // index; keep a multi-valued pocket beside it.
                        let pocket = format!("{name}_hash");
                        self.hash
                            .create_index(&HashIndexConfig {
                                name: pocket.clone(),
                                collection: collection.to_string(),
                                columns: columns.to_vec(),
                                unique: false,
                            })
                            .map_err(|err| err.to_string())?;
                        for (entity_id, fields) in entities {
                            if let Some(key) = hash_index_key(columns, fields) {
                                self.hash
                                    .insert(collection, &pocket, key, *entity_id)
                                    .map_err(|err| err.to_string())?;
                            }
                        }
                    }
                    return Ok(count);
                }
                // Build sorted in-memory index for range scans (single col).
                // Derive the indexed value for every row via `index_field_value`,
                // which resolves a bare promoted field (or a dotted path) from
                // the document body when it is no longer a materialised column
                // (single source of truth). For an ordinary stored column it
                // returns the value directly, so this is a no-op there — but it
                // ensures the sorted index is actually populated for a document
                // field, instead of silently building an empty index.
                let derived_entities = derive_index_entities_for_column(entities, col);
                let count = self.sorted.build_index(collection, col, &derived_entities);
                // Also build hash index for equality lookups on same column
                self.hash
                    .create_index(&HashIndexConfig {
                        name: format!("{name}_hash"),
                        collection: collection.to_string(),
                        columns: columns.to_vec(),
                        unique: false,
                    })
                    .map_err(|err| err.to_string())?;
                for (entity_id, fields) in entities {
                    if let Some(value) = index_field_value(fields, col) {
                        let key = value_to_bytes(value.as_ref());
                        self.hash
                            .insert(collection, &format!("{name}_hash"), key, *entity_id)
                            .map_err(|err| err.to_string())?;
                    }
                }
                Ok(count)
            }
        }
    }

    /// Drop an index
    pub fn drop_index(&self, name: &str, collection: &str) -> bool {
        let mut registry = write_unpoisoned(&self.registry);
        let key = (collection.to_string(), name.to_string());
        if let Some(info) = registry.remove(&key) {
            match info.method {
                IndexMethodKind::Hash => self.hash.drop_index(collection, name),
                IndexMethodKind::Bitmap => {
                    let col = info.columns.first().map(|s| s.as_str()).unwrap_or("");
                    self.bitmap.drop_index(collection, col)
                }
                // Sorted-backed indexes (BTree / H3) are removed from the
                // registry above; the sorted manager has no per-index drop
                // (mirrors the pre-existing BTree behaviour). A BTree's hash
                // pocket is dropped so a same-named index can be rebuilt.
                IndexMethodKind::BTree => self.hash.drop_index(collection, &format!("{name}_hash")),
                IndexMethodKind::H3 { .. } => false,
            };
            true
        } else {
            false
        }
    }

    /// Register index metadata
    pub fn register(&self, info: RegisteredIndex) {
        let mut registry = write_unpoisoned(&self.registry);
        registry.insert((info.collection.clone(), info.name.clone()), info);
    }

    /// Lookup entity IDs via hash index for a collection.column = value
    pub fn hash_lookup(
        &self,
        collection: &str,
        index_name: &str,
        key: &[u8],
    ) -> Result<Vec<EntityId>, String> {
        self.hash
            .lookup(collection, index_name, key)
            .map_err(|err| err.to_string())
    }

    /// Lookup entity IDs via bitmap index for a collection.column = value
    pub fn bitmap_lookup(
        &self,
        collection: &str,
        column: &str,
        value: &[u8],
    ) -> Result<Vec<EntityId>, String> {
        self.bitmap
            .lookup(collection, column, value)
            .map_err(|err| err.to_string())
    }

    /// Count via bitmap (O(1))
    pub fn bitmap_count(
        &self,
        collection: &str,
        column: &str,
        value: &[u8],
    ) -> Result<u64, String> {
        self.bitmap
            .count(collection, column, value)
            .map_err(|err| err.to_string())
    }

    /// Find which index (if any) covers a collection + column
    pub fn find_index_for_column(&self, collection: &str, column: &str) -> Option<RegisteredIndex> {
        let registry = read_unpoisoned(&self.registry);
        // Callers (`try_hash_eq_lookup`, IN-list lookup) feed `idx.name`
        // straight to `hash_lookup`, which only works for indexes that
        // actually own a hash backing — Hash itself, or a single-col
        // BTree whose `create_index` also registers an `{name}_hash`
        // aux. Composite BTree indexes only live in `sorted`, so
        // returning one here causes a `MissingIndex` error at lookup
        // time. Restrict the search to single-column indexes so the
        // ID list the caller gets back is always servable.
        registry
            .values()
            .find(|idx| {
                idx.collection == collection
                    && idx.columns.len() == 1
                    && idx.columns.first().is_some_and(|c| c == column)
            })
            .cloned()
    }

    /// List all indices for a collection
    pub fn list_indices(&self, collection: &str) -> Vec<RegisteredIndex> {
        let registry = read_unpoisoned(&self.registry);
        registry
            .values()
            .filter(|idx| idx.collection == collection)
            .cloned()
            .collect()
    }

    /// Standalone UNIQUE indexes enforced at admission. Both are backed by a
    /// hash-store key (`hash_lookup_name`) that admission probes.
    pub(crate) fn unique_indexes(&self, collection: &str) -> Vec<RegisteredIndex> {
        read_unpoisoned(&self.registry)
            .values()
            .filter(|index| is_enforced_unique(collection, index))
            .cloned()
            .collect()
    }

    pub(crate) fn has_unique_index(&self, collection: &str) -> bool {
        read_unpoisoned(&self.registry)
            .values()
            .any(|index| is_enforced_unique(collection, index))
    }

    /// Does an update of `modified_columns` change a standalone UNIQUE key?
    pub(crate) fn touches_unique_index(
        &self,
        collection: &str,
        modified_columns: &[String],
    ) -> bool {
        read_unpoisoned(&self.registry).values().any(|index| {
            is_enforced_unique(collection, index)
                && index.columns.iter().any(|column| {
                    modified_columns
                        .iter()
                        .any(|modified| modified.eq_ignore_ascii_case(column))
                })
        })
    }

    pub fn entries_indexed(&self, index: &RegisteredIndex) -> u64 {
        match index.method {
            IndexMethodKind::Hash => self
                .hash
                .index_stats(&index.collection, &index.name)
                .map(|stats| stats.total_entries as u64)
                .unwrap_or(0),
            IndexMethodKind::BTree | IndexMethodKind::H3 { .. } => {
                self.sorted
                    .entry_count(&index.collection, &index.columns)
                    .unwrap_or(0) as u64
            }
            IndexMethodKind::Bitmap => index
                .columns
                .first()
                .and_then(|column| self.bitmap.index_stats(&index.collection, column).ok())
                .map(|stats| stats.entity_count as u64)
                .unwrap_or(0),
        }
    }

    /// Collect the column-name set covered by any index on `collection`
    /// without cloning the full `RegisteredIndex` values — the
    /// `persist_applied_entity_mutations` HOT gate only needs to
    /// intersect `modified_columns` against this set, so cloning the
    /// index name / method / unique-flag every UPDATE is wasted work.
    pub fn indexed_columns_set(&self, collection: &str) -> std::collections::HashSet<String> {
        let registry = read_unpoisoned(&self.registry);
        let mut out = std::collections::HashSet::new();
        for idx in registry.values() {
            if idx.collection == collection {
                for col in &idx.columns {
                    out.insert(col.clone());
                }
            }
        }
        out
    }

    /// Like `indexed_columns_set` but also includes every parent path segment
    /// of dot-path indexed columns so that entity-field change detection works
    /// for document body indexes (e.g. a `"body.service.tier"` index also adds
    /// `"body"` and `"body.service"` so that a changed `"body"` field triggers
    /// the index refresh path for that nested column).
    pub fn indexed_columns_set_with_parents(
        &self,
        collection: &str,
    ) -> std::collections::HashSet<String> {
        let registry = read_unpoisoned(&self.registry);
        let mut out = std::collections::HashSet::new();
        for idx in registry.values() {
            if idx.collection == collection {
                for col in &idx.columns {
                    out.insert(col.clone());
                    if col.contains('.') {
                        let mut prefix = String::new();
                        for segment in col.split('.') {
                            if !prefix.is_empty() {
                                out.insert(prefix.clone());
                                prefix.push('.');
                            }
                            prefix.push_str(segment);
                        }
                    }
                }
            }
        }
        out
    }

    /// Batched counterpart of `index_entity_insert`: takes a full
    /// slice of `(EntityId, Vec<(String, Value)>)` pairs and walks
    /// the index registry ONCE, then loops inside each registered
    /// index. For an N-row insert with K indexes, this turns the
    /// previous `N × K` registry-lock acquisitions into exactly one.
    /// Matches the PG `heap_multi_insert` +  `ExecInsertIndexTuples`
    /// fusion so bulk ingest doesn't pay per-row overhead.
    pub fn index_entity_insert_batch(
        &self,
        collection: &str,
        rows: &[(EntityId, Vec<(String, Value)>)],
    ) -> Result<(), String> {
        if rows.is_empty() {
            return Ok(());
        }
        let registry = self.registry.read();

        // Pre-collect the index list for this collection once so the
        // hot loop isn't re-scanning the whole registry per row.
        let relevant: Vec<&RegisteredIndex> = registry
            .values()
            .filter(|idx| idx.collection == collection)
            .collect();
        if relevant.is_empty() {
            return Ok(());
        }

        // For each index, walk rows × fields but `break` as soon as
        // the matching column is found — every field name appears
        // at most once per row, so we average O(ncols/2) iterations
        // per (row, index) instead of O(ncols). Beats the
        // HashMap-pre-build shape when indexes < cols (the common
        // OLTP schema), because the inner `break` keeps string
        // compares short and amortised.
        for idx in &relevant {
            let composite_btree =
                matches!(idx.method, IndexMethodKind::BTree) && idx.columns.len() > 1;
            if composite_btree {
                self.sorted
                    .composite_insert_batch(collection, &idx.columns, rows);
            }

            // Composite HASH / UNIQUE BTREE keys span several columns.
            if let Some(pocket) = idx.composite_key_pocket() {
                for (entity_id, fields) in rows {
                    if let Some(key) = hash_index_key(&idx.columns, fields) {
                        self.hash
                            .insert(collection, &pocket, key, *entity_id)
                            .map_err(|err| err.to_string())?;
                    }
                }
                continue;
            }
            if composite_btree {
                continue;
            }

            let col = idx.columns.first().map(|s| s.as_str()).unwrap_or("");
            // Hoist the "{name}_hash" auxiliary index name out of
            // the per-row inner loop for BTree indexes. Previously
            // every row paid a fresh `format!()` allocation.
            let btree_hash_name =
                matches!(idx.method, IndexMethodKind::BTree).then(|| format!("{}_hash", idx.name));
            for (entity_id, fields) in rows {
                if let Some(value) = index_field_value(fields, col) {
                    let key = value_to_bytes(value.as_ref());
                    match idx.method {
                        IndexMethodKind::Hash => {
                            // A NULL never equals another NULL: no unique key.
                            if idx.unique && matches!(value.as_ref(), Value::Null) {
                                continue;
                            }
                            self.hash
                                .insert(collection, &idx.name, key, *entity_id)
                                .map_err(|err| err.to_string())?;
                        }
                        IndexMethodKind::Bitmap => {
                            self.bitmap
                                .insert(collection, col, *entity_id, &key)
                                .map_err(|err| err.to_string())?;
                        }
                        IndexMethodKind::BTree => {
                            if !self.sorted.has_index(collection, col) {
                                return Err(format!(
                                    "sorted index for collection '{collection}' column '{col}' was not found"
                                ));
                            }
                            self.sorted
                                .insert_one(collection, col, value.as_ref(), *entity_id);
                            let hash_name = btree_hash_name.as_deref().unwrap_or("");
                            self.hash
                                .insert(collection, hash_name, key, *entity_id)
                                .map_err(|err| err.to_string())?;
                        }
                        IndexMethodKind::H3 { resolution } => {
                            if let Some(cell) = h3_cell_value(value.as_ref(), resolution) {
                                self.sorted.insert_one(collection, col, &cell, *entity_id);
                            }
                        }
                    }
                }
            }
        }
        Ok(())
    }

    /// Insert one entity's relevant column values into every index
    /// registered on its collection. Called from the entity insert
    /// pipeline so that CREATE INDEX can land before or after the
    /// data without losing new rows. Missing backing structures are
    /// surfaced as errors instead of being silently ignored.
    pub fn index_entity_insert(
        &self,
        collection: &str,
        entity_id: EntityId,
        fields: &[(String, Value)],
    ) -> Result<(), String> {
        let registry = self.registry.read();
        for idx in registry.values() {
            if idx.collection != collection {
                continue;
            }

            // Composite BTree (multi-column) — maintain the tuple index.
            let composite_btree =
                matches!(idx.method, IndexMethodKind::BTree) && idx.columns.len() > 1;
            if composite_btree {
                self.sorted
                    .composite_insert_one(collection, &idx.columns, entity_id, fields);
            }

            // Composite HASH / UNIQUE BTREE keys span several columns.
            if let Some(pocket) = idx.composite_key_pocket() {
                if let Some(key) = hash_index_key(&idx.columns, fields) {
                    self.hash
                        .insert(collection, &pocket, key, entity_id)
                        .map_err(|err| err.to_string())?;
                }
                continue;
            }
            if composite_btree {
                continue;
            }

            let col = idx.columns.first().map(|s| s.as_str()).unwrap_or("");
            if let Some(value) = index_field_value(fields, col) {
                let key = value_to_bytes(value.as_ref());
                match idx.method {
                    IndexMethodKind::Hash => {
                        // A NULL never equals another NULL: no unique key.
                        if idx.unique && matches!(value.as_ref(), Value::Null) {
                            continue;
                        }
                        self.hash
                            .insert(collection, &idx.name, key, entity_id)
                            .map_err(|err| err.to_string())?;
                    }
                    IndexMethodKind::Bitmap => {
                        self.bitmap
                            .insert(collection, col, entity_id, &key)
                            .map_err(|err| err.to_string())?;
                    }
                    IndexMethodKind::BTree => {
                        if !self.sorted.has_index(collection, col) {
                            return Err(format!(
                                "sorted index for collection '{collection}' column '{col}' was not found"
                            ));
                        }
                        self.sorted
                            .insert_one(collection, col, value.as_ref(), entity_id);
                        self.hash
                            .insert(collection, &format!("{}_hash", idx.name), key, entity_id)
                            .map_err(|err| err.to_string())?;
                    }
                    IndexMethodKind::H3 { resolution } => {
                        if !self.sorted.has_index(collection, col) {
                            return Err(format!(
                                "sorted index for collection '{collection}' column '{col}' was not found"
                            ));
                        }
                        if let Some(cell) = h3_cell_value(value.as_ref(), resolution) {
                            self.sorted.insert_one(collection, col, &cell, entity_id);
                        }
                    }
                }
            }
        }
        Ok(())
    }

    /// Mirror of `index_entity_insert` for the delete path. Removes the
    /// row from every secondary index registered on its collection. Index
    /// misses are tolerated (index may have been dropped between fetch
    /// and delete) — only structural errors propagate.
    pub fn index_entity_delete(
        &self,
        collection: &str,
        entity_id: EntityId,
        fields: &[(String, Value)],
    ) -> Result<(), String> {
        let registry = self.registry.read();
        for idx in registry.values() {
            if idx.collection != collection {
                continue;
            }

            // Composite BTree — drop the entry under the old tuple, if
            // present. Swap with None-fields on the update side flushes
            // it out cleanly.
            let composite_btree =
                matches!(idx.method, IndexMethodKind::BTree) && idx.columns.len() > 1;
            if composite_btree {
                self.sorted.composite_entity_update(
                    collection,
                    &idx.columns,
                    entity_id,
                    fields,
                    &[],
                );
            }

            // Composite HASH / UNIQUE BTREE: drop the entry under the full
            // tuple key. Missing index on delete is non-fatal.
            if let Some(pocket) = idx.composite_key_pocket() {
                if let Some(key) = hash_index_key(&idx.columns, fields) {
                    let _ = self.hash.remove(collection, &pocket, &key, entity_id);
                }
                continue;
            }
            if composite_btree {
                continue;
            }

            let col = idx.columns.first().map(|s| s.as_str()).unwrap_or("");
            if let Some(value) = index_field_value(fields, col) {
                let key = value_to_bytes(value.as_ref());
                match idx.method {
                    IndexMethodKind::Hash => {
                        // Missing index on delete is non-fatal.
                        let _ = self.hash.remove(collection, &idx.name, &key, entity_id);
                    }
                    IndexMethodKind::Bitmap => {
                        let _ = self.bitmap.remove(collection, col, entity_id);
                    }
                    IndexMethodKind::BTree => {
                        self.sorted
                            .delete_one(collection, col, value.as_ref(), entity_id);
                        let _ = self.hash.remove(
                            collection,
                            &format!("{}_hash", idx.name),
                            &key,
                            entity_id,
                        );
                    }
                    IndexMethodKind::H3 { resolution } => {
                        if let Some(cell) = h3_cell_value(value.as_ref(), resolution) {
                            self.sorted.delete_one(collection, col, &cell, entity_id);
                        }
                    }
                }
            }
        }
        Ok(())
    }

    /// Apply an update to every secondary index registered on the
    /// collection. For each indexed column, if the value changed, deletes
    /// the entry under the old key and inserts under the new key. Columns
    /// that are absent from `new_fields` (typed as set-to-NULL) are
    /// removed from the index. New indexed columns appearing only in
    /// `new_fields` get inserted.
    pub fn index_entity_update(
        &self,
        collection: &str,
        entity_id: EntityId,
        old_fields: &[(String, Value)],
        new_fields: &[(String, Value)],
    ) -> Result<(), String> {
        // Snapshot the indexed columns once to avoid holding the registry
        // lock across delete/insert sub-calls (those re-acquire it).
        // Also include parent path segments (e.g. "body" for "body.service.tier")
        // so that a changed top-level "body" field triggers re-indexing for
        // nested path indexes.
        let indexed_cols: std::collections::HashSet<String> = {
            let registry = self.registry.read();
            let mut cols = std::collections::HashSet::new();
            for idx in registry.values().filter(|idx| idx.collection == collection) {
                for col in &idx.columns {
                    cols.insert(col.clone());
                    if col.contains('.') {
                        let mut prefix = String::new();
                        for segment in col.split('.') {
                            if !prefix.is_empty() {
                                cols.insert(prefix.clone());
                                prefix.push('.');
                            }
                            prefix.push_str(segment);
                        }
                    }
                }
            }
            cols
        };
        if indexed_cols.is_empty() {
            return Ok(());
        }

        // Composite HASH / UNIQUE BTREE keys span several columns, which the
        // per-column delete/insert below cannot rebuild (each call carries a
        // single field), so re-key those indexes here from the full old/new rows.
        let composite_pockets: Vec<(RegisteredIndex, String)> = self
            .registry
            .read()
            .values()
            .filter(|idx| idx.collection == collection)
            .filter_map(|idx| {
                let pocket = idx.composite_key_pocket()?.into_owned();
                Some((idx.clone(), pocket))
            })
            .collect();
        for (idx, pocket) in &composite_pockets {
            let old_key = hash_index_key(&idx.columns, old_fields);
            let new_key = hash_index_key(&idx.columns, new_fields);
            if old_key == new_key {
                continue;
            }
            if let Some(key) = old_key {
                let _ = self.hash.remove(collection, pocket, &key, entity_id);
            }
            if let Some(key) = new_key {
                self.hash
                    .insert(collection, pocket, key, entity_id)
                    .map_err(|err| err.to_string())?;
            }
        }

        // Compute the full damage-vector once, then filter to the
        // indexed columns (or their parent path segments). Drops the
        // previous O(indexed × old.len) pairwise scan to a single pass.
        let damage = crate::application::entity::row_damage_vector(old_fields, new_fields);

        for (col, old_value, new_value) in &damage.changed {
            if !indexed_cols.contains(col) {
                continue;
            }
            self.index_entity_delete(collection, entity_id, &[(col.clone(), old_value.clone())])?;
            self.index_entity_insert(collection, entity_id, &[(col.clone(), new_value.clone())])?;
        }
        for (col, old_value) in &damage.removed {
            if !indexed_cols.contains(col) {
                continue;
            }
            self.index_entity_delete(collection, entity_id, &[(col.clone(), old_value.clone())])?;
        }
        for (col, new_value) in &damage.added {
            if !indexed_cols.contains(col) {
                continue;
            }
            self.index_entity_insert(collection, entity_id, &[(col.clone(), new_value.clone())])?;
        }

        // Single-source documents: an index on a promoted field (`score` or a
        // dotted `body.score`) is backed by the body, not a stored column, so
        // the field-name damage vector above never sees it change — it only
        // sees `body` change. Re-evaluate every such index column directly from
        // the old/new body. No-op for non-document collections, where these
        // columns resolve to nothing in both snapshots.
        for col in &indexed_cols {
            if old_fields.iter().any(|(f, _)| f == col) || new_fields.iter().any(|(f, _)| f == col)
            {
                // A real stored column — already handled by the damage loop.
                continue;
            }
            let old_value = index_field_value(old_fields, col);
            let new_value = index_field_value(new_fields, col);
            match (old_value, new_value) {
                (Some(old), Some(new)) => {
                    if old.as_ref() != new.as_ref() {
                        self.index_entity_delete(
                            collection,
                            entity_id,
                            &[(col.clone(), old.into_owned())],
                        )?;
                        self.index_entity_insert(
                            collection,
                            entity_id,
                            &[(col.clone(), new.into_owned())],
                        )?;
                    }
                }
                (Some(old), None) => {
                    self.index_entity_delete(
                        collection,
                        entity_id,
                        &[(col.clone(), old.into_owned())],
                    )?;
                }
                (None, Some(new)) => {
                    self.index_entity_insert(
                        collection,
                        entity_id,
                        &[(col.clone(), new.into_owned())],
                    )?;
                }
                (None, None) => {}
            }
        }
        Ok(())
    }
}

impl Default for IndexStore {
    fn default() -> Self {
        Self::new()
    }
}

// --------------------------------------------------------------------
// IncrementalIndexMaintainer adapter (issue #160)
//
// Routes the maintainer's `IndexDeltaOp` stream through the existing
// hash/bitmap manager methods — no parallel write paths. Composite
// BTree and the sorted half of single-column BTree still flow through
// `index_entity_insert`/`composite_entity_*` because those need the
// raw `Value`, which the op stream doesn't carry; the maintainer
// documents this restriction.
// --------------------------------------------------------------------

use crate::storage::unified::index::incremental::{
    IndexMethodKind as IncMethodKind, SecondaryIndexBackend, SecondaryIndexHandle,
};

impl From<IndexMethodKind> for IncMethodKind {
    fn from(value: IndexMethodKind) -> Self {
        match value {
            IndexMethodKind::Hash => IncMethodKind::Hash,
            IndexMethodKind::Bitmap => IncMethodKind::Bitmap,
            IndexMethodKind::BTree => IncMethodKind::BTree,
            // The maintainer mirror doesn't carry the H3 resolution; the
            // sorted half is maintained via `index_entity_insert`.
            IndexMethodKind::H3 { .. } => IncMethodKind::H3,
        }
    }
}

impl From<IncMethodKind> for IndexMethodKind {
    fn from(value: IncMethodKind) -> Self {
        match value {
            IncMethodKind::Hash => IndexMethodKind::Hash,
            IncMethodKind::Bitmap => IndexMethodKind::Bitmap,
            IncMethodKind::BTree => IndexMethodKind::BTree,
            // Resolution isn't carried by the mirror — placeholder only;
            // the authoritative resolution lives in the index registry.
            IncMethodKind::H3 => IndexMethodKind::H3 {
                resolution: DEFAULT_H3_RESOLUTION,
            },
        }
    }
}

impl SecondaryIndexBackend for IndexStore {
    fn insert(
        &self,
        idx: &SecondaryIndexHandle,
        key: &[u8],
        row_id: EntityId,
    ) -> Result<(), String> {
        let collection = idx.collection.as_ref();
        let Some(col) = idx.leading_column() else {
            return Ok(());
        };
        match idx.method {
            IncMethodKind::Hash => self
                .hash
                .insert(collection, idx.name.as_ref(), key.to_vec(), row_id)
                .map_err(|e| e.to_string()),
            IncMethodKind::Bitmap => self
                .bitmap
                .insert(collection, col, row_id, key)
                .map_err(|e| e.to_string()),
            IncMethodKind::BTree => {
                // Single-col BTree: maintain the auxiliary hash side
                // pocket. Sorted half is maintained by the caller via
                // the pre-existing `index_entity_insert` route, which
                // has access to the raw `Value`. Composite BTree:
                // skipped (composite_entity_insert covers it).
                if idx.columns.len() > 1 {
                    return Ok(());
                }
                let aux = format!("{}_hash", idx.name);
                self.hash
                    .insert(collection, &aux, key.to_vec(), row_id)
                    .map_err(|e| e.to_string())
            }
            // H3 has no auxiliary hash side-pocket; its sorted half is
            // maintained by `index_entity_insert` (which has the raw
            // GeoPoint Value + resolution).
            IncMethodKind::H3 => Ok(()),
        }
    }

    fn remove(&self, idx: &SecondaryIndexHandle, key: &[u8], row_id: EntityId) {
        let collection = idx.collection.as_ref();
        let Some(col) = idx.leading_column() else {
            return;
        };
        match idx.method {
            IncMethodKind::Hash => {
                let _ = self.hash.remove(collection, idx.name.as_ref(), key, row_id);
            }
            IncMethodKind::Bitmap => {
                let _ = self.bitmap.remove(collection, col, row_id);
            }
            IncMethodKind::BTree => {
                if idx.columns.len() > 1 {
                    return;
                }
                let aux = format!("{}_hash", idx.name);
                let _ = self.hash.remove(collection, &aux, key, row_id);
            }
            IncMethodKind::H3 => {}
        }
    }
}

impl IndexStore {
    /// Resolve the value an index on `column` would store for a row, including a
    /// bare/dotted document field offset-read from the `body` container when the
    /// column is no longer a materialised field (single source of truth).
    ///
    /// Used by the update path to rebuild old/new index keys for body-derived
    /// columns, which the field-name damage vector cannot see change on its own.
    pub fn resolve_index_field_value(
        &self,
        fields: &[(String, Value)],
        column: &str,
    ) -> Option<Value> {
        index_field_value(fields, column).map(Cow::into_owned)
    }

    /// Fold body-derived index columns into a row's field snapshot.
    ///
    /// Single-source documents keep promoted index columns (e.g. `score`) in
    /// the `body` container, not as stored fields, so secondary-index
    /// maintenance must surface their values as ordinary columns — otherwise
    /// the diff between the old and new snapshots never sees a body-only change.
    /// For each indexed column absent from `fields`, resolve it from the body
    /// and append it; ordinary stored columns are returned untouched.
    pub fn augment_body_derived_index_fields(
        &self,
        fields: &[(String, Value)],
        indexed_cols: &std::collections::HashSet<String>,
    ) -> Vec<(String, Value)> {
        let mut out = fields.to_vec();
        for col in indexed_cols {
            if out.iter().any(|(existing, _)| existing == col) {
                continue;
            }
            if let Some(value) = self.resolve_index_field_value(fields, col) {
                out.push((col.clone(), value));
            }
        }
        out
    }
}

pub(crate) fn index_field_value<'a>(
    fields: &'a [(String, Value)],
    column: &str,
) -> Option<Cow<'a, Value>> {
    if let Some((_, Value::Json(bytes))) = fields.iter().find(|(field, _)| field == "body") {
        // Single-source document: an index on a bare or dotted document field
        // (`score`, `location.gps`) is backed by the body, not a stored column.
        // The ROOT segment must be offset-read from the binary `body`
        // container (`read_body_field` is the binary-container-aware,
        // storage-typed read); the generic document-path resolver does not
        // decode binary containers, so resolving the whole path against the
        // raw `body` value silently yields nothing and builds an empty
        // index. Descend into nested structure only AFTER the offset-read.
        let segments = super::join_filter::parse_runtime_document_path(column);
        if let Some((root, tail)) = segments.split_first() {
            let root_value = crate::document_body::read_body_field(bytes, root).or_else(|| {
                crate::json::from_slice::<crate::json::Value>(bytes)
                    .ok()?
                    .as_object()?
                    .get(root)
                    .and_then(|value| crate::application::entity::json_to_storage_value(value).ok())
            });
            if let Some(root_value) = root_value {
                if tail.is_empty() {
                    return Some(Cow::Owned(root_value));
                }
                if let Some(value) =
                    super::join_filter::resolve_runtime_document_path_from_value(&root_value, tail)
                {
                    return Some(Cow::Owned(value));
                }
            }
        }
    }
    if let Some((_, value)) = fields.iter().find(|(field, _)| field == column) {
        return Some(Cow::Borrowed(value));
    }
    if !column.contains('.') {
        return None;
    }

    let segments = super::join_filter::parse_runtime_document_path(column);
    let (root, tail) = segments.split_first()?;
    if tail.is_empty() {
        return None;
    }
    let (_, root_value) = fields.iter().find(|(field, _)| field == root)?;
    super::join_filter::resolve_runtime_document_path_from_value(root_value, tail).map(Cow::Owned)
}

fn derive_index_entities_for_column(
    entities: &[(EntityId, Vec<(String, Value)>)],
    column: &str,
) -> Vec<(EntityId, Vec<(String, Value)>)> {
    entities
        .iter()
        .filter_map(|(entity_id, fields)| {
            index_field_value(fields, column)
                .map(|value| (*entity_id, vec![(column.to_string(), value.into_owned())]))
        })
        .collect()
}

/// Encode a geo column value to its H3 cell-id at `resolution`, wrapped
/// as a `Value::UnsignedInteger` so it stores and range-scans through the
/// ordinary sorted-index machinery. Mirrors the `H3_INDEX` RQL scalar
/// (slice 1), which also yields an `UnsignedInteger`, so the index key
/// family matches the query side.
///
/// Returns `None` when the value is not a recognized geo value, or when the
/// coordinate fails to encode (`lat_lng_to_cell` returns the `0`
/// sentinel) — those rows are simply absent from the index, exactly as
/// an `Unsupported` value is absent from a BTree index.
fn h3_cell_value(value: &Value, resolution: u8) -> Option<Value> {
    let (lat, lon) = crate::geo::recognize_geo_value(value)?;
    let cell = crate::geo::h3::lat_lng_to_cell(lat, lon, resolution);
    if cell == 0 {
        return None;
    }
    Some(Value::UnsignedInteger(cell))
}

/// Build the `(entity, [(column, cell-id)])` rows for an H3 index from a
/// collection's existing entities: resolve each row's geo column, encode
/// it to its cell-id, and skip rows whose column is absent or not a geo
/// point. The resulting rows feed `SortedIndexManager::build_index`
/// unchanged.
fn derive_h3_cell_entities(
    entities: &[(EntityId, Vec<(String, Value)>)],
    column: &str,
    resolution: u8,
) -> Vec<(EntityId, Vec<(String, Value)>)> {
    entities
        .iter()
        .filter_map(|(entity_id, fields)| {
            let value = index_field_value(fields, column)?;
            let cell = h3_cell_value(value.as_ref(), resolution)?;
            Some((*entity_id, vec![(column.to_string(), cell)]))
        })
        .collect()
}

// Conservative per-key allowance for hash/bitmap map slots, postings and
// bitmap forward/reverse IDs. Canonical sorted keys are added separately.
const INDEX_GROWTH_ENTRY_BYTES: u64 = 96;

fn estimate_index_entry_bytes(value: &Value) -> u64 {
    INDEX_GROWTH_ENTRY_BYTES.saturating_add(value_to_bytes_len(value))
}

fn estimate_registered_index_growth(index: &RegisteredIndex, fields: &[(String, Value)]) -> u64 {
    let canonical_bytes = std::mem::size_of::<CanonicalKey>() as u64;
    if matches!(index.method, IndexMethodKind::BTree | IndexMethodKind::Hash)
        && index.columns.len() > 1
    {
        return index
            .columns
            .iter()
            .try_fold(INDEX_GROWTH_ENTRY_BYTES, |growth, column| {
                let (_, value) = fields.iter().find(|(name, _)| name == column)?;
                Some(
                    growth
                        .saturating_add(canonical_bytes)
                        .saturating_add(value_to_bytes_len(value).max(16)),
                )
            })
            .unwrap_or(0);
    }
    let Some(column) = index.columns.first() else {
        return 0;
    };
    let Some(value) = index_field_value(fields, column) else {
        return 0;
    };
    match index.method {
        // H3 stores a fixed-width canonical cell, not the source geometry.
        IndexMethodKind::H3 { .. } => INDEX_GROWTH_ENTRY_BYTES
            .saturating_add(canonical_bytes)
            .saturating_add(16),
        IndexMethodKind::Hash | IndexMethodKind::Bitmap => {
            estimate_index_entry_bytes(value.as_ref())
        }
        IndexMethodKind::BTree => {
            // Single-column BTree maintains both a sorted key and equality hash.
            let key_bytes = value_to_bytes_len(value.as_ref());
            INDEX_GROWTH_ENTRY_BYTES
                .saturating_add(key_bytes)
                .saturating_add(INDEX_GROWTH_ENTRY_BYTES)
                .saturating_add(canonical_bytes)
                .saturating_add(key_bytes.max(16))
        }
    }
}

/// Size the existing hash-key encoding without materializing a second buffer.
/// Debug-encoded values can be much larger than their underlying payload.
fn value_to_bytes_len(value: &Value) -> u64 {
    match value {
        Value::Text(text) => text.len() as u64,
        Value::Integer(_) | Value::UnsignedInteger(_) | Value::Float(_) => 8,
        Value::Boolean(_) => 1,
        _ => {
            struct ByteCount(u64);
            impl std::fmt::Write for ByteCount {
                fn write_str(&mut self, text: &str) -> std::fmt::Result {
                    self.0 = self.0.saturating_add(text.len() as u64);
                    Ok(())
                }
            }
            let mut count = ByteCount(0);
            std::fmt::Write::write_fmt(&mut count, format_args!("{value:?}"))
                .expect("invariant: counting formatted bytes cannot fail");
            count.0
        }
    }
}

/// Physical HASH key of `fields` for an index over `columns`, or `None` when
/// the row has no entry in that index.
///
/// A single-column index keeps its historical key (`value_to_bytes` of the
/// column, an explicit NULL included), so existing single-column indexes and
/// lookups are unchanged. A composite index keys on **every** column: each
/// column's bytes are length-prefixed, so distinct tuples cannot collide
/// (`('ab','c')` vs `('a','bc')`), and a row with an absent or NULL column has
/// no key, because a NULL never equals anything and so cannot conflict.
pub(crate) fn hash_index_key(columns: &[String], fields: &[(String, Value)]) -> Option<Vec<u8>> {
    match columns {
        [] => None,
        [column] => index_field_value(fields, column).map(|value| value_to_bytes(value.as_ref())),
        _ => {
            let mut key = Vec::new();
            for column in columns {
                let value = index_field_value(fields, column)?;
                if matches!(value.as_ref(), Value::Null) {
                    return None;
                }
                let bytes = value_to_bytes(value.as_ref());
                key.extend_from_slice(&(bytes.len() as u32).to_le_bytes());
                key.extend_from_slice(&bytes);
            }
            Some(key)
        }
    }
}

/// Key a standalone UNIQUE index reserves for `fields`, or `None` when the row
/// reserves nothing. Same bytes as [`hash_index_key`], except that a NULL
/// single column reserves no key either: a NULL never equals another NULL, as
/// in a composite key and in a declared UNIQUE constraint.
pub(crate) fn unique_index_key(columns: &[String], fields: &[(String, Value)]) -> Option<Vec<u8>> {
    if let [column] = columns {
        if matches!(
            index_field_value(fields, column).as_deref(),
            Some(Value::Null)
        ) {
            return None;
        }
    }
    hash_index_key(columns, fields)
}

/// The first row among `rows` that reserves the same unique key as an earlier
/// one, as `(earlier, later)`. Used to refuse `CREATE UNIQUE INDEX` over
/// existing duplicates.
pub(crate) fn first_duplicate_unique_key(
    columns: &[String],
    rows: &[(EntityId, Vec<(String, Value)>)],
) -> Option<(EntityId, EntityId)> {
    let mut seen = std::collections::HashMap::with_capacity(rows.len());
    for (entity_id, fields) in rows {
        let Some(key) = unique_index_key(columns, fields) else {
            continue;
        };
        if let Some(earlier) = seen.insert(key, *entity_id) {
            return Some((earlier, *entity_id));
        }
    }
    None
}

fn is_enforced_unique(collection: &str, index: &RegisteredIndex) -> bool {
    index.collection == collection
        && index.unique
        && matches!(index.method, IndexMethodKind::Hash | IndexMethodKind::BTree)
}

/// Convert a Value to bytes for index key
pub(crate) fn value_to_bytes(value: &Value) -> Vec<u8> {
    match value {
        Value::Text(s) => s.as_bytes().to_vec(),
        Value::Integer(n) => n.to_le_bytes().to_vec(),
        Value::UnsignedInteger(n) => n.to_le_bytes().to_vec(),
        Value::Float(n) => n.to_le_bytes().to_vec(),
        Value::Boolean(b) => vec![*b as u8],
        _ => format!("{:?}", value).into_bytes(),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn row(pairs: &[(&str, Value)]) -> Vec<(String, Value)> {
        pairs
            .iter()
            .map(|(name, value)| ((*name).to_string(), value.clone()))
            .collect()
    }

    #[test]
    fn hash_index_key_single_column_keeps_the_historical_encoding() {
        let columns = vec!["id".to_string()];
        assert_eq!(
            hash_index_key(&columns, &row(&[("id", Value::Integer(7))])),
            Some(7i64.to_le_bytes().to_vec())
        );
        assert_eq!(
            hash_index_key(&columns, &row(&[("name", Value::text("x"))])),
            None
        );
        // An explicit NULL has always keyed a single-column index.
        assert!(hash_index_key(&columns, &row(&[("id", Value::Null)])).is_some());
    }

    #[test]
    fn hash_index_key_composite_covers_every_column_and_cannot_collide() {
        let columns = vec!["a".to_string(), "b".to_string()];
        let key = |a: &str, b: &str| {
            hash_index_key(
                &columns,
                &row(&[("a", Value::text(a)), ("b", Value::text(b))]),
            )
            .expect("both columns present")
        };
        assert_ne!(
            key("x", "1"),
            key("x", "2"),
            "the second column is part of the key"
        );
        assert_ne!(
            key("ab", "c"),
            key("a", "bc"),
            "length prefixes keep tuples apart"
        );
        assert_eq!(key("x", "1"), key("x", "1"));
    }

    #[test]
    fn hash_index_key_composite_has_no_key_for_null_or_absent_columns() {
        let columns = vec!["a".to_string(), "b".to_string()];
        let null_b = row(&[("a", Value::text("x")), ("b", Value::Null)]);
        assert_eq!(hash_index_key(&columns, &null_b), None);
        assert_eq!(
            hash_index_key(&columns, &row(&[("a", Value::text("x"))])),
            None
        );
    }

    #[test]
    fn unique_index_key_never_reserves_a_null_column() {
        let single = vec!["id".to_string()];
        assert_eq!(
            unique_index_key(&single, &row(&[("id", Value::Null)])),
            None,
            "a NULL never equals another NULL"
        );
        assert_eq!(
            unique_index_key(&single, &row(&[("id", Value::Integer(7))])),
            hash_index_key(&single, &row(&[("id", Value::Integer(7))])),
            "non-NULL keys keep the physical encoding"
        );
        let composite = vec!["a".to_string(), "b".to_string()];
        let full = row(&[("a", Value::text("x")), ("b", Value::text("1"))]);
        assert_eq!(
            unique_index_key(&composite, &full),
            hash_index_key(&composite, &full)
        );
    }

    #[test]
    fn first_duplicate_unique_key_reports_the_colliding_pair_and_skips_nulls() {
        let single = vec!["id".to_string()];
        let rows = vec![
            (EntityId::new(1), row(&[("id", Value::Null)])),
            (EntityId::new(2), row(&[("id", Value::Null)])),
            (EntityId::new(3), row(&[("id", Value::Integer(5))])),
            (EntityId::new(4), row(&[("id", Value::Integer(6))])),
            (EntityId::new(5), row(&[("id", Value::Integer(5))])),
        ];
        assert_eq!(
            first_duplicate_unique_key(&single, &rows),
            Some((EntityId::new(3), EntityId::new(5)))
        );
        assert_eq!(first_duplicate_unique_key(&single, &rows[..2]), None);

        let composite = vec!["a".to_string(), "b".to_string()];
        let tuples = vec![
            (
                EntityId::new(1),
                row(&[("a", Value::text("x")), ("b", Value::text("1"))]),
            ),
            (
                EntityId::new(2),
                row(&[("a", Value::text("x")), ("b", Value::text("2"))]),
            ),
        ];
        assert_eq!(first_duplicate_unique_key(&composite, &tuples), None);
    }

    #[test]
    fn rebuilding_a_unique_btree_index_over_existing_duplicates_never_refuses() {
        // Reopening a database rebuilds its indexes through `create_index`.
        // Data an older release let duplicate must not stop it from opening;
        // only `CREATE UNIQUE INDEX` (DDL) refuses duplicates.
        for columns in [
            vec!["id".to_string()],
            vec!["id".to_string(), "grp".to_string()],
        ] {
            let store = IndexStore::new();
            let rows = vec![
                (
                    EntityId::new(1),
                    row(&[("id", Value::Integer(1)), ("grp", Value::text("g"))]),
                ),
                (
                    EntityId::new(2),
                    row(&[("id", Value::Integer(1)), ("grp", Value::text("g"))]),
                ),
            ];
            store
                .create_index("dup", "rows", &columns, IndexMethodKind::BTree, true, &rows)
                .expect("rebuild keeps the duplicates instead of refusing");
            let key = hash_index_key(&columns, &rows[0].1).expect("key");
            assert_eq!(
                store
                    .hash_lookup("rows", "dup_hash", &key)
                    .expect("pocket")
                    .len(),
                2
            );
        }
    }

    #[test]
    fn composite_unique_btree_pocket_follows_insert_update_delete_and_drop() {
        let store = IndexStore::new();
        let columns = vec!["a".to_string(), "b".to_string()];
        store
            .create_index("ab", "rows", &columns, IndexMethodKind::BTree, true, &[])
            .expect("composite unique btree index");
        store.register(RegisteredIndex {
            name: "ab".to_string(),
            collection: "rows".to_string(),
            columns: columns.clone(),
            method: IndexMethodKind::BTree,
            unique: true,
        });
        let old = row(&[("a", Value::text("x")), ("b", Value::text("1"))]);
        let new = row(&[("a", Value::text("x")), ("b", Value::text("2"))]);
        let old_key = hash_index_key(&columns, &old).expect("old key");
        let new_key = hash_index_key(&columns, &new).expect("new key");
        let id = EntityId::new(7);
        let lookup = |key: &[u8]| store.hash_lookup("rows", "ab_hash", key).expect("pocket");

        store.index_entity_insert("rows", id, &old).expect("insert");
        assert_eq!(lookup(&old_key), vec![id]);

        store
            .index_entity_update("rows", id, &old, &new)
            .expect("update");
        assert!(lookup(&old_key).is_empty(), "the old tuple is released");
        assert_eq!(lookup(&new_key), vec![id]);

        store.index_entity_delete("rows", id, &new).expect("delete");
        assert!(lookup(&new_key).is_empty());

        assert!(store.drop_index("ab", "rows"));
        assert!(
            store.hash_lookup("rows", "ab_hash", &new_key).is_err(),
            "dropping the index drops its pocket so the name can be reused"
        );
    }

    #[test]
    fn composite_hash_index_is_rekeyed_from_the_full_rows_on_update_and_delete() {
        let store = IndexStore::new();
        let columns = vec!["a".to_string(), "b".to_string()];
        store
            .create_index("ab", "rows", &columns, IndexMethodKind::Hash, false, &[])
            .expect("composite hash index");
        store.register(RegisteredIndex {
            name: "ab".to_string(),
            collection: "rows".to_string(),
            columns: columns.clone(),
            method: IndexMethodKind::Hash,
            unique: false,
        });
        let old = row(&[("a", Value::text("x")), ("b", Value::text("1"))]);
        let new = row(&[("a", Value::text("x")), ("b", Value::text("2"))]);
        let key = |fields: &[(String, Value)]| hash_index_key(&columns, fields).expect("key");
        let id = EntityId::new(1);

        store.index_entity_insert("rows", id, &old).expect("insert");
        assert_eq!(
            store.hash_lookup("rows", "ab", &key(&old)).expect("lookup"),
            vec![id]
        );

        // The per-column delete/insert inside an update cannot rebuild a tuple;
        // the index must follow the change through the full old/new rows.
        store
            .index_entity_update("rows", id, &old, &new)
            .expect("update");
        assert!(store
            .hash_lookup("rows", "ab", &key(&old))
            .expect("lookup")
            .is_empty());
        assert_eq!(
            store.hash_lookup("rows", "ab", &key(&new)).expect("lookup"),
            vec![id]
        );

        store.index_entity_delete("rows", id, &new).expect("delete");
        assert!(store
            .hash_lookup("rows", "ab", &key(&new))
            .expect("lookup")
            .is_empty());
    }

    #[test]
    fn row_index_single_insert_updates_each_composite_once() {
        let store = IndexStore::new();
        for (name, columns) in [
            ("first", vec!["a".to_string(), "b".to_string()]),
            ("second", vec!["b".to_string(), "a".to_string()]),
        ] {
            store
                .create_index(name, "rows", &columns, IndexMethodKind::BTree, false, &[])
                .expect("composite index");
            store.register(RegisteredIndex {
                name: name.to_string(),
                collection: "rows".to_string(),
                columns,
                method: IndexMethodKind::BTree,
                unique: false,
            });
        }
        store
            .index_entity_insert(
                "rows",
                EntityId::new(1),
                &[
                    ("a".to_string(), Value::Integer(10)),
                    ("b".to_string(), Value::Integer(20)),
                ],
            )
            .expect("insert");
        for index in store.list_indices("rows") {
            assert_eq!(
                store.entries_indexed(&index),
                1,
                "one posting per composite index"
            );
        }
    }

    #[test]
    fn row_index_estimates_cover_resident_growth_for_single_and_batch_writes() {
        for batch_size in [1, 16] {
            for (method, columns) in [
                (IndexMethodKind::Hash, vec!["a".to_string()]),
                (IndexMethodKind::Bitmap, vec!["a".to_string()]),
                (IndexMethodKind::BTree, vec!["a".to_string()]),
                (
                    IndexMethodKind::BTree,
                    vec!["a".to_string(), "b".to_string()],
                ),
                (
                    IndexMethodKind::H3 { resolution: 9 },
                    vec!["geo".to_string()],
                ),
            ] {
                let store = IndexStore::new();
                for name in ["first", "second"] {
                    store
                        .create_index(name, "rows", &columns, method, false, &[])
                        .expect("index");
                    store.register(RegisteredIndex {
                        name: name.to_string(),
                        collection: "rows".to_string(),
                        columns: columns.clone(),
                        method,
                        unique: false,
                    });
                }
                let rows = (1..=batch_size)
                    .map(|id| {
                        (
                            EntityId::new(id),
                            vec![
                                (
                                    "a".to_string(),
                                    Value::text(format!("{id}-{}", "x".repeat(1024))),
                                ),
                                ("b".to_string(), Value::Integer(id as i64)),
                                ("geo".to_string(), Value::GeoPoint(23_000_000, -46_000_000)),
                            ],
                        )
                    })
                    .collect::<Vec<_>>();
                let estimate = store.estimate_insert_growth(
                    "rows",
                    rows.iter().map(|(_, fields)| fields.as_slice()),
                    false,
                );
                let before = store.memory_bytes();
                if batch_size == 1 {
                    store
                        .index_entity_insert("rows", rows[0].0, &rows[0].1)
                        .expect("single");
                } else {
                    store
                        .index_entity_insert_batch("rows", &rows)
                        .expect("batch");
                }
                let growth = store.memory_bytes() - before;
                assert!(growth > 0, "fixture must populate {method:?}");
                assert!(
                    estimate >= growth,
                    "{method:?}: estimated {estimate}, resident delta {growth}"
                );
            }
        }
    }

    #[test]
    fn row_index_auto_estimate_includes_header_and_stops_after_registration() {
        let store = IndexStore::new();
        let fields = vec![("id".to_string(), Value::Integer(1))];
        let rows = std::iter::once(fields.as_slice());
        assert_eq!(store.estimate_insert_growth("rows", rows.clone(), false), 0);
        let estimate = store.estimate_insert_growth("rows", rows.clone(), true);
        let columns = vec!["id".to_string()];
        store
            .create_index(
                "idx_id",
                "rows",
                &columns,
                IndexMethodKind::Hash,
                false,
                &[],
            )
            .expect("auto index");
        store.register(RegisteredIndex {
            name: "idx_id".to_string(),
            collection: "rows".to_string(),
            columns,
            method: IndexMethodKind::Hash,
            unique: false,
        });
        store
            .index_entity_insert("rows", EntityId::new(1), &fields)
            .expect("single");
        assert!(
            estimate >= store.memory_bytes(),
            "initial header and entry must fit"
        );
        assert_eq!(
            store.estimate_insert_growth("rows", rows.clone(), true),
            store.estimate_insert_growth("rows", rows, false),
            "registered id is not charged twice"
        );
        assert_eq!(
            store.estimate_insert_growth("rows", std::iter::once([].as_slice()), true),
            0,
            "missing fields do not create index entries"
        );
    }

    #[test]
    fn row_index_key_size_matches_the_existing_encoding() {
        for value in [
            Value::Null,
            Value::Integer(-7),
            Value::UnsignedInteger(u64::MAX),
            Value::Float(1.25),
            Value::Boolean(true),
            Value::text("ação\n\""),
            Value::Blob(vec![0, 127, 255]),
            Value::RowRef("rows".to_string(), 42),
            Value::Array(vec![Value::text("nested"), Value::Integer(3)]),
        ] {
            assert_eq!(
                value_to_bytes_len(&value),
                value_to_bytes(&value).len() as u64
            );
        }
    }

    fn ids(values: &[EntityId]) -> Vec<u64> {
        values.iter().map(|id| id.raw()).collect()
    }

    fn assert_memory_accounting(manager: &SortedIndexManager) {
        let single = manager
            .indices
            .read()
            .values()
            .map(|index| {
                std::mem::size_of::<SortedColumnIndex>()
                    + index
                        .entries
                        .iter()
                        .map(|(key, ids)| btree_entry_bytes(std::iter::once(key), ids.len()))
                        .sum::<usize>()
            })
            .sum::<usize>();
        let composite = manager
            .composite
            .read()
            .values()
            .map(|index| {
                std::mem::size_of::<SortedCompositeIndex>()
                    + index
                        .entries
                        .iter()
                        .map(|(key, ids)| btree_entry_bytes(key.iter(), ids.len()))
                        .sum::<usize>()
            })
            .sum::<usize>();
        assert_eq!(manager.memory_bytes(), (single + composite) as u64);
    }

    #[test]
    fn memory_accounting_tracks_sorted_updates_deletes_and_rebuilds() {
        let manager = SortedIndexManager::new();
        manager
            .indices
            .write()
            .insert(("items".into(), "key".into()), SortedColumnIndex::new());
        let columns = vec!["key".to_string(), "group".to_string()];
        let mut entities = Vec::new();
        for id in 0..128 {
            let value = Value::text(format!("key-{}", id % 13));
            manager.insert_one("items", "key", &value, EntityId::new(id));
            manager.insert_one("items", "key", &value, EntityId::new(id));
            entities.push((
                EntityId::new(id),
                vec![("key".into(), value), ("group".into(), Value::Integer(1))],
            ));
            assert_memory_accounting(&manager);
        }
        manager.build_composite("items", &columns, &entities);
        assert_memory_accounting(&manager);
        for (id, fields) in &entities {
            manager.delete_one("items", "key", &fields[0].1, *id);
            manager.delete_one("items", "key", &fields[0].1, *id);
            let updated = vec![
                ("key".into(), Value::text("updated")),
                ("group".into(), Value::Integer(2)),
            ];
            manager.composite_entity_update("items", &columns, *id, fields, &updated);
            assert_memory_accounting(&manager);
            manager.composite_entity_update("items", &columns, *id, &updated, &[]);
            assert_memory_accounting(&manager);
        }
        manager.build_composite("items", &columns, &entities);
        assert_memory_accounting(&manager);
        manager.composite.write().clear();
        manager.indices.write().clear();
        assert_eq!(manager.memory_bytes(), 0);
    }

    #[test]
    fn h3_cell_value_encodes_geopoint_to_unsigned_cell() {
        // Paris (Notre-Dame). i32 micro-degrees.
        let paris = Value::GeoPoint(48_856_600, 2_352_200);
        let cell = h3_cell_value(&paris, 9).expect("geopoint must encode to a cell");
        match cell {
            Value::UnsignedInteger(n) => {
                assert_ne!(n, 0, "valid coordinate must not yield the 0 sentinel");
                assert_eq!(
                    n,
                    crate::geo::h3::lat_lng_to_cell(48.8566, 2.3522, 9),
                    "index key must match the slice-1 encode of the same point"
                );
            }
            other => panic!("expected UnsignedInteger cell, got {other:?}"),
        }

        // Finer resolution yields a different (child) cell id.
        let fine = h3_cell_value(&paris, 12).unwrap();
        assert_ne!(
            fine, cell,
            "a finer resolution must produce a distinct cell"
        );
    }

    #[test]
    fn h3_cell_value_rejects_non_geopoint() {
        assert!(h3_cell_value(&Value::Integer(5), 9).is_none());
        assert!(h3_cell_value(&Value::text("48.8,2.3".to_string()), 9).is_none());
    }

    #[test]
    fn h3_cell_value_uses_shared_geo_recognition_for_json_objects() {
        let value = Value::Json(br#"{"latitude":48.8566,"longitude":2.3522}"#.to_vec());
        let cell = h3_cell_value(&value, 9).expect("json object coordinate must encode");
        assert_eq!(
            cell,
            Value::UnsignedInteger(crate::geo::h3::lat_lng_to_cell(48.8566, 2.3522, 9))
        );

        let geojson = Value::Json(br#"{"type":"Point","coordinates":[2.3522,48.8566]}"#.to_vec());
        let cell = h3_cell_value(&geojson, 9).expect("GeoJSON Point must encode");
        assert_eq!(
            cell,
            Value::UnsignedInteger(crate::geo::h3::lat_lng_to_cell(48.8566, 2.3522, 9))
        );
    }

    #[test]
    fn h3_cell_value_rejects_shared_geo_non_goals() {
        for value in [
            Value::Json(br#"{"lat":"48.8566","lon":"2.3522"}"#.to_vec()),
            Value::Json(br#"{"type":"Polygon","coordinates":[[2.3522,48.8566]]}"#.to_vec()),
            Value::Json(br#"{"type":"Point","coordinates":[2.3522]}"#.to_vec()),
            Value::Json(br#"{"type":"Point","coordinates":[2.3522,48.8566,0]}"#.to_vec()),
            Value::Json(br#"{"lat":48.8566}"#.to_vec()),
            Value::Json(br#"{"lat":91.0,"lon":2.3522}"#.to_vec()),
            Value::Json(br#"{"type":"Point","coordinates":[2.3522,91.0]}"#.to_vec()),
        ] {
            assert_eq!(h3_cell_value(&value, 9), None, "{value:?}");
        }
    }

    #[test]
    fn derive_h3_cell_entities_skips_missing_and_non_geo_rows() {
        let entities = vec![
            (
                EntityId::new(1),
                vec![("loc".to_string(), Value::GeoPoint(48_856_600, 2_352_200))],
            ),
            // No `loc` column — skipped.
            (
                EntityId::new(2),
                vec![("id".to_string(), Value::Integer(2))],
            ),
            // `loc` present but not a geo point — skipped.
            (
                EntityId::new(3),
                vec![("loc".to_string(), Value::Integer(0))],
            ),
        ];
        let derived = derive_h3_cell_entities(&entities, "loc", 9);
        assert_eq!(derived.len(), 1, "only the valid geo row is indexed");
        assert_eq!(derived[0].0, EntityId::new(1));
        assert!(matches!(
            derived[0].1.first(),
            Some((c, Value::UnsignedInteger(_))) if c == "loc"
        ));
    }

    #[test]
    fn test_sorted_key_supports_text_ranges() {
        let mut index = SortedColumnIndex::new();
        index.insert(
            value_to_sorted_key(&Value::text("alpha".to_string())).unwrap(),
            EntityId::new(1),
        );
        index.insert(
            value_to_sorted_key(&Value::text("bravo".to_string())).unwrap(),
            EntityId::new(2),
        );
        index.insert(
            value_to_sorted_key(&Value::text("charlie".to_string())).unwrap(),
            EntityId::new(3),
        );

        assert_eq!(
            ids(&index
                .greater_than(value_to_sorted_key(&Value::text("alpha".to_string())).unwrap())
                .unwrap()),
            vec![2, 3]
        );
        assert_eq!(
            ids(&index
                .less_equal(value_to_sorted_key(&Value::text("alpha".to_string())).unwrap())
                .unwrap()),
            vec![1]
        );
        assert_eq!(
            ids(&index
                .range(
                    value_to_sorted_key(&Value::text("bravo".to_string())).unwrap(),
                    value_to_sorted_key(&Value::text("charlie".to_string())).unwrap(),
                )
                .unwrap()),
            vec![2, 3]
        );
    }

    #[test]
    fn test_sorted_index_disables_range_lookup_when_mixed_families_are_present() {
        let manager = SortedIndexManager::new();
        let entities = vec![
            (
                EntityId::new(1),
                vec![("score".to_string(), Value::Integer(10))],
            ),
            (
                EntityId::new(2),
                vec![("score".to_string(), Value::Float(10.5))],
            ),
        ];

        manager.build_index("numbers", "score", &entities);

        assert_eq!(
            manager.range_lookup(
                "numbers",
                "score",
                value_to_sorted_key(&Value::Integer(0)).unwrap(),
                value_to_sorted_key(&Value::Integer(20)).unwrap(),
            ),
            None
        );
    }

    #[test]
    fn test_sorted_index_keeps_exact_in_lookup_when_mixed_families_are_present() {
        let manager = SortedIndexManager::new();
        let entities = vec![
            (
                EntityId::new(1),
                vec![("mixed".to_string(), Value::Integer(10))],
            ),
            (
                EntityId::new(2),
                vec![("mixed".to_string(), Value::text("ten".to_string()))],
            ),
        ];

        manager.build_index("mixed_table", "mixed", &entities);

        let matched = manager
            .in_lookup_limited(
                "mixed_table",
                "mixed",
                &[value_to_sorted_key(&Value::Integer(10)).unwrap()],
                10,
            )
            .unwrap();
        assert_eq!(ids(&matched), vec![1]);
    }

    #[test]
    fn test_index_entity_insert_errors_when_registered_hash_index_is_missing() {
        let store = IndexStore::new();
        store.register(RegisteredIndex {
            name: "idx_email".to_string(),
            collection: "users".to_string(),
            columns: vec!["email".to_string()],
            method: IndexMethodKind::Hash,
            unique: false,
        });

        let err = store
            .index_entity_insert(
                "users",
                EntityId::new(1),
                &[("email".to_string(), Value::text("a@b.com".to_string()))],
            )
            .expect_err("missing backing hash index should surface as an error");

        assert!(err.contains("idx_email"));
        assert!(err.contains("users"));
    }

    #[test]
    fn test_index_entity_insert_batch_maintains_composite_btree() {
        let store = IndexStore::new();
        let columns = vec!["city".to_string(), "age".to_string()];
        store
            .create_index(
                "idx_city_age",
                "users",
                &columns,
                IndexMethodKind::BTree,
                false,
                &[],
            )
            .unwrap();
        store.register(RegisteredIndex {
            name: "idx_city_age".to_string(),
            collection: "users".to_string(),
            columns: columns.clone(),
            method: IndexMethodKind::BTree,
            unique: false,
        });

        let rows = vec![
            (
                EntityId::new(1),
                vec![
                    ("city".to_string(), Value::text("NYC".to_string())),
                    ("age".to_string(), Value::Integer(30)),
                ],
            ),
            (
                EntityId::new(2),
                vec![
                    ("city".to_string(), Value::text("LA".to_string())),
                    ("age".to_string(), Value::Integer(30)),
                ],
            ),
            (
                EntityId::new(3),
                vec![
                    ("city".to_string(), Value::text("NYC".to_string())),
                    ("age".to_string(), Value::Integer(40)),
                ],
            ),
        ];

        store.index_entity_insert_batch("users", &rows).unwrap();

        let found = store
            .sorted
            .composite_prefix_range_lookup(
                "users",
                &columns,
                &[value_to_sorted_key(&Value::text("NYC".to_string())).unwrap()],
                value_to_sorted_key(&Value::Integer(20)).unwrap(),
                value_to_sorted_key(&Value::Integer(45)).unwrap(),
                10,
            )
            .unwrap();
        assert_eq!(ids(&found), vec![1, 3]);
    }

    /// Helper that fully provisions a hash index on `users.email` with
    /// the matching registry entry — the realistic state after a
    /// `CREATE INDEX` call.
    fn provision_hash_index(store: &IndexStore) {
        store
            .create_index(
                "idx_email",
                "users",
                &["email".to_string()],
                IndexMethodKind::Hash,
                false,
                &[],
            )
            .unwrap();
        store.register(RegisteredIndex {
            name: "idx_email".to_string(),
            collection: "users".to_string(),
            columns: vec!["email".to_string()],
            method: IndexMethodKind::Hash,
            unique: false,
        });
    }

    #[test]
    fn test_index_entity_delete_removes_row_from_hash_index() {
        let store = IndexStore::new();
        provision_hash_index(&store);
        let id = EntityId::new(7);
        let key = ("email".to_string(), Value::text("a@b.com".to_string()));

        store
            .index_entity_insert("users", id, &[key.clone()])
            .unwrap();
        assert_eq!(
            ids(&store.hash_lookup("users", "idx_email", b"a@b.com").unwrap()),
            vec![7]
        );

        store
            .index_entity_delete("users", id, &[key.clone()])
            .unwrap();
        assert!(store
            .hash_lookup("users", "idx_email", b"a@b.com")
            .unwrap()
            .is_empty());
    }

    #[test]
    fn test_index_entity_delete_tolerates_missing_index() {
        // No backing hash index exists — delete must still return Ok
        // (deleting from a dropped index is a non-fatal no-op).
        let store = IndexStore::new();
        store.register(RegisteredIndex {
            name: "idx_email".to_string(),
            collection: "users".to_string(),
            columns: vec!["email".to_string()],
            method: IndexMethodKind::Hash,
            unique: false,
        });
        store
            .index_entity_delete(
                "users",
                EntityId::new(1),
                &[("email".to_string(), Value::text("a@b.com".to_string()))],
            )
            .expect("delete must tolerate a missing backing index");
    }

    #[test]
    fn test_index_entity_update_moves_row_to_new_key() {
        let store = IndexStore::new();
        provision_hash_index(&store);
        let id = EntityId::new(11);
        let old = ("email".to_string(), Value::text("old@x.com".to_string()));
        let new = ("email".to_string(), Value::text("new@x.com".to_string()));

        store
            .index_entity_insert("users", id, &[old.clone()])
            .unwrap();

        store
            .index_entity_update("users", id, &[old.clone()], &[new.clone()])
            .unwrap();

        // Lookup under old key returns nothing, new key returns the id.
        assert!(store
            .hash_lookup("users", "idx_email", b"old@x.com")
            .unwrap()
            .is_empty());
        assert_eq!(
            ids(&store
                .hash_lookup("users", "idx_email", b"new@x.com")
                .unwrap()),
            vec![11]
        );
    }

    #[test]
    fn test_index_entity_update_skips_unchanged_columns() {
        // Update where the indexed column value didn't change must
        // leave the index untouched (no spurious delete + reinsert).
        let store = IndexStore::new();
        provision_hash_index(&store);
        let id = EntityId::new(13);
        let same = ("email".to_string(), Value::text("a@b.com".to_string()));

        store
            .index_entity_insert("users", id, &[same.clone()])
            .unwrap();
        store
            .index_entity_update("users", id, &[same.clone()], &[same.clone()])
            .unwrap();
        assert_eq!(
            ids(&store.hash_lookup("users", "idx_email", b"a@b.com").unwrap()),
            vec![13]
        );
    }

    #[test]
    fn test_index_entity_update_indexes_newly_added_column() {
        // Column present only in `new_fields` (was absent in `old`)
        // must be inserted into the index via the damage_vector.added
        // bucket.
        let store = IndexStore::new();
        provision_hash_index(&store);
        let id = EntityId::new(17);
        let old: Vec<(String, Value)> = vec![]; // email wasn't set before
        let new = vec![("email".to_string(), Value::text("fresh@x.com".to_string()))];

        store.index_entity_update("users", id, &old, &new).unwrap();

        assert_eq!(
            ids(&store
                .hash_lookup("users", "idx_email", b"fresh@x.com")
                .unwrap()),
            vec![17]
        );
    }

    #[test]
    fn test_index_entity_update_removes_dropped_column() {
        // Column present in `old_fields` but absent from `new_fields`
        // (SET email = NULL / DROP column) must be removed via the
        // damage_vector.removed bucket.
        let store = IndexStore::new();
        provision_hash_index(&store);
        let id = EntityId::new(19);
        let old = vec![("email".to_string(), Value::text("bye@x.com".to_string()))];
        let new: Vec<(String, Value)> = vec![]; // email dropped

        store.index_entity_insert("users", id, &old).unwrap();
        assert_eq!(
            ids(&store
                .hash_lookup("users", "idx_email", b"bye@x.com")
                .unwrap()),
            vec![19]
        );

        store.index_entity_update("users", id, &old, &new).unwrap();

        assert!(store
            .hash_lookup("users", "idx_email", b"bye@x.com")
            .unwrap()
            .is_empty());
    }

    #[test]
    fn test_index_entity_update_ignores_non_indexed_column_changes() {
        // Changing a column that isn't registered in the index
        // registry must be a no-op — touch the index zero times.
        let store = IndexStore::new();
        provision_hash_index(&store); // only idx_email is registered
        let id = EntityId::new(23);

        let insert = vec![("email".to_string(), Value::text("a@b.com".to_string()))];
        store.index_entity_insert("users", id, &insert).unwrap();

        // Update only changes `age` (not registered). email unchanged.
        let old = vec![
            ("email".to_string(), Value::text("a@b.com".to_string())),
            ("age".to_string(), Value::Integer(30)),
        ];
        let new = vec![
            ("email".to_string(), Value::text("a@b.com".to_string())),
            ("age".to_string(), Value::Integer(31)),
        ];
        store.index_entity_update("users", id, &old, &new).unwrap();

        // Index is still intact under the unchanged email key.
        assert_eq!(
            ids(&store.hash_lookup("users", "idx_email", b"a@b.com").unwrap()),
            vec![23]
        );
    }
}