reddb_server/storage/unified/devx/reddb/

impl_access.rs

use super::*;

pub(crate) struct NativeArtifactBuild {
    pub records: Vec<(NativeVectorArtifactSummary, Vec<u8>)>,
    pub deferred_collections: std::collections::HashSet<String>,
}

impl RedDB {
    /// Access the quorum coordinator (Phase 2.6 PG parity).
    ///
    /// `None` when this instance is not a primary. Write paths call
    /// `quorum.wait_for_quorum(lsn)` after the primary WAL commit to
    /// block until the configured replica quorum has acked; `Async`
    /// mode (default) returns instantly so no behavioural change.
    pub fn quorum_coordinator(
        &self,
    ) -> Option<&Arc<crate::replication::quorum::QuorumCoordinator>> {
        self.quorum.as_ref()
    }

    /// Wait for the configured replica quorum on a given LSN.
    ///
    /// Convenience wrapper that returns `Ok(())` when:
    /// * there is no quorum coordinator (non-primary instance), OR
    /// * the coordinator is in `Async` mode (historical behaviour).
    ///
    /// Callers in the write path (INSERT/UPDATE/DELETE dispatch) invoke
    /// this after logging the commit record. Returning `Err` signals
    /// that replication did NOT reach quorum — the primary still has
    /// the record but the client ack should be deferred or rejected.
    pub fn wait_for_replication_quorum(
        &self,
        target_lsn: u64,
    ) -> Result<(), crate::replication::QuorumError> {
        match &self.quorum {
            Some(q) => q.wait_for_quorum(target_lsn),
            None => Ok(()),
        }
    }
}

impl RedDB {
    pub(crate) fn native_registry_summary_from_metadata(
        &self,
        metadata: &PhysicalMetadataFile,
        artifact_summaries: &[NativeVectorArtifactSummary],
        artifact_count: usize,
    ) -> NativeRegistrySummary {
        const SAMPLE_LIMIT: usize = 32;

        let collection_names: Vec<_> = metadata
            .catalog
            .stats_by_collection
            .keys()
            .take(SAMPLE_LIMIT)
            .cloned()
            .collect();
        let indexes: Vec<_> = metadata
            .indexes
            .iter()
            .take(SAMPLE_LIMIT)
            .map(|index| NativeRegistryIndexSummary {
                name: index.name.clone(),
                kind: index.kind.as_str().to_string(),
                collection: index.collection.clone(),
                enabled: index.enabled,
                entries: index.entries as u64,
                estimated_memory_bytes: index.estimated_memory_bytes,
                last_refresh_ms: index.last_refresh_ms,
                backend: index.backend.clone(),
            })
            .collect();
        let graph_projections: Vec<_> = metadata
            .graph_projections
            .iter()
            .take(SAMPLE_LIMIT)
            .map(|projection| NativeRegistryProjectionSummary {
                name: projection.name.clone(),
                source: projection.source.clone(),
                created_at_unix_ms: projection.created_at_unix_ms,
                updated_at_unix_ms: projection.updated_at_unix_ms,
                node_labels: projection.node_labels.clone(),
                node_types: projection.node_types.clone(),
                edge_labels: projection.edge_labels.clone(),
                last_materialized_sequence: projection.last_materialized_sequence,
            })
            .collect();
        let analytics_jobs = metadata
            .analytics_jobs
            .iter()
            .take(SAMPLE_LIMIT)
            .map(|job| NativeRegistryJobSummary {
                id: job.id.clone(),
                kind: job.kind.clone(),
                projection: job.projection.clone(),
                state: job.state.clone(),
                created_at_unix_ms: job.created_at_unix_ms,
                updated_at_unix_ms: job.updated_at_unix_ms,
                last_run_sequence: job.last_run_sequence,
                metadata: job.metadata.clone(),
            })
            .collect::<Vec<_>>();
        let vector_artifacts = artifact_summaries
            .iter()
            .take(SAMPLE_LIMIT)
            .cloned()
            .collect::<Vec<_>>();
        let vector_artifact_count = artifact_count as u32;

        NativeRegistrySummary {
            collection_count: metadata.catalog.total_collections as u32,
            index_count: metadata.indexes.len() as u32,
            graph_projection_count: metadata.graph_projections.len() as u32,
            analytics_job_count: metadata.analytics_jobs.len() as u32,
            vector_artifact_count,
            collections_complete: metadata.catalog.stats_by_collection.len() <= SAMPLE_LIMIT,
            indexes_complete: metadata.indexes.len() <= SAMPLE_LIMIT,
            graph_projections_complete: metadata.graph_projections.len() <= SAMPLE_LIMIT,
            analytics_jobs_complete: metadata.analytics_jobs.len() <= SAMPLE_LIMIT,
            vector_artifacts_complete: vector_artifact_count as usize <= SAMPLE_LIMIT,
            omitted_collection_count: metadata
                .catalog
                .stats_by_collection
                .len()
                .saturating_sub(collection_names.len())
                as u32,
            omitted_index_count: metadata.indexes.len().saturating_sub(indexes.len()) as u32,
            omitted_graph_projection_count: metadata
                .graph_projections
                .len()
                .saturating_sub(graph_projections.len())
                as u32,
            omitted_analytics_job_count: metadata
                .analytics_jobs
                .len()
                .saturating_sub(analytics_jobs.len())
                as u32,
            omitted_vector_artifact_count: vector_artifact_count
                .saturating_sub(vector_artifacts.len() as u32),
            collection_names,
            indexes,
            graph_projections,
            analytics_jobs,
            vector_artifacts,
        }
    }

    pub(crate) fn native_vector_artifact_records(
        &self,
    ) -> Result<NativeArtifactBuild, Box<dyn std::error::Error>> {
        let budget = crate::storage::memory_budget::resolve_for_boot(
            self.options.storage_profile.deploy_profile,
            self.options.memory_budget_bytes,
        )?
        .resolved_bytes;
        // These optional derived snapshots share a small, conservative slice
        // of the process budget. Their builders amplify source memory through
        // tokens, postings and encoded frames, so bound input before copying.
        let source_limit = usize::try_from((budget / 256).clamp(1024, 2 * 1024 * 1024))
            .expect("bounded source allowance");
        let output_limit = usize::try_from((budget / 128).min(8 * 1024 * 1024))
            .expect("bounded artifact allowance");
        let mut artifacts = Vec::new();
        let mut retained_bytes = 0usize;
        let mut deferred_collections = std::collections::HashSet::new();
        let mut collections = self.store.list_collections();
        collections.sort_unstable();
        'collections: for collection in collections {
            let Some(manager) = self.store.get_collection(&collection) else {
                continue;
            };
            if manager.count() == 0 {
                continue;
            }
            let mut entities = Vec::new();
            let mut source_bytes = 0usize;
            let mut exceeded = false;
            let column_schema = manager.column_schema();
            manager.for_each_entity(|entity| {
                let eligible = match &entity.data {
                    EntityData::Vector(vector) => {
                        self.options.has_capability(Capability::Vector) && !vector.dense.is_empty()
                    }
                    EntityData::Edge(_) => self.options.has_capability(Capability::Graph),
                    _ => self.options.has_capability(Capability::FullText),
                };
                if !eligible {
                    return true;
                }
                let bytes = reddb_engine::storage::memory_size::entity_bytes(entity);
                if bytes > source_limit.saturating_sub(source_bytes)
                    || retained_bytes >= output_limit
                {
                    exceeded = true;
                    return false;
                }
                source_bytes += bytes;
                let mut owned = entity.clone();
                if let EntityData::Row(row) = &mut owned.data {
                    if row.named.is_none() && row.schema.is_none() {
                        row.schema = column_schema.clone();
                    }
                }
                entities.push(owned);
                true
            });
            if exceeded {
                deferred_collections.insert(collection);
                continue;
            }
            if entities.is_empty() {
                continue;
            }
            let first_artifact = artifacts.len();
            let mut vectors = Vec::new();
            let mut graph_edges = Vec::new();
            let mut fulltext_documents = Vec::new();
            let mut document_records = Vec::new();
            let mut document_bytes = 0usize;
            for entity in entities {
                match entity.data {
                    EntityData::Vector(vector) => {
                        if !vector.dense.is_empty() {
                            vectors.push((entity.id, vector.dense));
                        }
                    }
                    EntityData::Edge(edge) => {
                        if let EntityKind::GraphEdge(edge_kind) = entity.kind {
                            graph_edges.push((
                                entity.id,
                                edge_kind.from_node,
                                edge_kind.to_node,
                                edge_kind.label,
                                edge.weight,
                            ));
                        }
                    }
                    data => {
                        let text = Self::native_fulltext_text_for_entity(&data);
                        if !text.trim().is_empty() {
                            fulltext_documents.push((entity.id, text));
                        }
                        let document = match Self::native_document_pathvalue_for_entity(
                            entity.id,
                            &data,
                            source_limit.saturating_sub(document_bytes),
                        ) {
                            Ok(document) => document,
                            Err(()) => {
                                artifacts.truncate(first_artifact);
                                deferred_collections.insert(collection);
                                continue 'collections;
                            }
                        };
                        if let Some(document) = document {
                            document_bytes += document
                                .1
                                .iter()
                                .map(|(path, value)| {
                                    path.len()
                                        + value.len()
                                        + std::mem::size_of::<(String, String)>()
                                })
                                .sum::<usize>();
                            document_records.push(document);
                        }
                    }
                }
            }
            if !vectors.is_empty() {
                let dimension = vectors[0].1.len();
                let mut hnsw = HnswIndex::with_dimension(dimension);
                for (id, vector) in vectors
                    .into_iter()
                    .filter(|(_, vector)| vector.len() == dimension)
                {
                    hnsw.insert_with_id(id.raw(), vector);
                }
                let stats = hnsw.stats();
                let bytes = hnsw.to_bytes();
                let summary = NativeVectorArtifactSummary {
                    collection: collection.clone(),
                    artifact_kind: "hnsw".to_string(),
                    vector_count: stats.node_count as u64,
                    dimension: stats.dimension as u32,
                    max_layer: stats.max_layer as u32,
                    serialized_bytes: bytes.len() as u64,
                    checksum: crate::storage::engine::crc32(&bytes) as u64,
                };
                artifacts.push((summary, bytes));

                let n_lists = ((stats.node_count as f64).sqrt().ceil() as usize).max(1);
                let mut ivf = IvfIndex::new(IvfConfig::new(dimension, n_lists));
                let training = manager
                    .query_all(|_| true)
                    .into_iter()
                    .filter_map(|entity| match entity.data {
                        EntityData::Vector(vector) if vector.dense.len() == dimension => {
                            Some(vector.dense)
                        }
                        _ => None,
                    })
                    .collect::<Vec<_>>();
                ivf.train(&training);
                let items = manager
                    .query_all(|_| true)
                    .into_iter()
                    .filter_map(|entity| match entity.data {
                        EntityData::Vector(vector) if vector.dense.len() == dimension => {
                            Some((entity.id.raw(), vector.dense))
                        }
                        _ => None,
                    })
                    .collect::<Vec<_>>();
                ivf.add_batch_with_ids(items);
                let ivf_stats = ivf.stats();
                let ivf_bytes = ivf.to_bytes();
                let ivf_summary = NativeVectorArtifactSummary {
                    collection: collection.clone(),
                    artifact_kind: "ivf".to_string(),
                    vector_count: ivf_stats.total_vectors as u64,
                    dimension: ivf_stats.dimension as u32,
                    max_layer: ivf_stats.n_lists as u32,
                    serialized_bytes: ivf_bytes.len() as u64,
                    checksum: crate::storage::engine::crc32(&ivf_bytes) as u64,
                };
                artifacts.push((ivf_summary, ivf_bytes));
            }

            if !graph_edges.is_empty() {
                let bytes = Self::serialize_native_graph_adjacency_artifact(&graph_edges);
                let (edge_count, node_count, label_count) =
                    Self::inspect_native_graph_adjacency_artifact(&bytes).unwrap_or((0, 0, 0));
                let summary = NativeVectorArtifactSummary {
                    collection: collection.clone(),
                    artifact_kind: "graph.adjacency".to_string(),
                    vector_count: edge_count,
                    dimension: node_count as u32,
                    max_layer: label_count,
                    serialized_bytes: bytes.len() as u64,
                    checksum: crate::storage::engine::crc32(&bytes) as u64,
                };
                artifacts.push((summary, bytes));
            }

            if !fulltext_documents.is_empty() {
                let bytes =
                    Self::serialize_native_fulltext_artifact(&collection, &fulltext_documents);
                let (doc_count, term_count, posting_count) =
                    Self::inspect_native_fulltext_artifact(&bytes).unwrap_or((0, 0, 0));
                let summary = NativeVectorArtifactSummary {
                    collection: collection.clone(),
                    artifact_kind: "text.fulltext".to_string(),
                    vector_count: posting_count,
                    dimension: doc_count as u32,
                    max_layer: term_count as u32,
                    serialized_bytes: bytes.len() as u64,
                    checksum: crate::storage::engine::crc32(&bytes) as u64,
                };
                artifacts.push((summary, bytes));
            }

            if !document_records.is_empty() {
                let bytes = Self::serialize_native_document_pathvalue_artifact(
                    &collection,
                    &document_records,
                );
                let (doc_count, path_count, value_count, unique_value_count) =
                    Self::inspect_native_document_pathvalue_artifact(&bytes)
                        .unwrap_or((0, 0, 0, 0));
                let _ = unique_value_count;
                let summary = NativeVectorArtifactSummary {
                    collection: collection.clone(),
                    artifact_kind: "document.pathvalue".to_string(),
                    vector_count: value_count,
                    dimension: doc_count as u32,
                    max_layer: path_count as u32,
                    serialized_bytes: bytes.len() as u64,
                    checksum: crate::storage::engine::crc32(&bytes) as u64,
                };
                artifacts.push((summary, bytes));
            }
            let bytes = artifacts[first_artifact..]
                .iter()
                .map(|(_, bytes)| bytes.len())
                .sum::<usize>();
            if bytes > output_limit.saturating_sub(retained_bytes) {
                artifacts.truncate(first_artifact);
                deferred_collections.insert(collection);
            } else {
                retained_bytes += bytes;
            }
        }
        for collection in &deferred_collections {
            tracing::warn!(collection, source_bytes_limit=source_limit, retained_bytes_limit=output_limit,
                "native search artifact deferred by memory budget; canonical records remain available");
        }
        Ok(NativeArtifactBuild {
            records: artifacts,
            deferred_collections,
        })
    }

    fn serialize_native_graph_adjacency_artifact(
        edges: &[(EntityId, String, String, String, f32)],
    ) -> Vec<u8> {
        let edges: Vec<reddb_file::NativeGraphEdge> = edges
            .iter()
            .map(
                |(edge_id, from_node, to_node, label, weight)| reddb_file::NativeGraphEdge {
                    edge_id: edge_id.raw(),
                    from_node: from_node.clone(),
                    to_node: to_node.clone(),
                    label: label.clone(),
                    weight: *weight,
                },
            )
            .collect();
        reddb_file::encode_native_graph_adjacency_frame(&reddb_file::NativeGraphAdjacencyFrame {
            edges,
        })
    }

    pub(crate) fn inspect_native_graph_adjacency_artifact(
        bytes: &[u8],
    ) -> Result<(u64, u64, u32), String> {
        let frame =
            reddb_file::decode_native_graph_adjacency_frame(bytes).map_err(|e| e.to_string())?;
        let edge_count = frame.edges.len() as u64;
        let mut nodes = BTreeSet::new();
        let mut labels = BTreeSet::new();
        for edge in frame.edges {
            nodes.insert(edge.from_node);
            nodes.insert(edge.to_node);
            labels.insert(edge.label);
        }
        Ok((edge_count, nodes.len() as u64, labels.len() as u32))
    }

    fn serialize_native_fulltext_artifact(
        collection: &str,
        documents: &[(EntityId, String)],
    ) -> Vec<u8> {
        let mut postings: BTreeMap<String, Vec<(u64, u32)>> = BTreeMap::new();
        for (entity_id, text) in documents {
            let mut frequencies: BTreeMap<String, u32> = BTreeMap::new();
            for token in Self::native_fulltext_tokenize(text) {
                *frequencies.entry(token).or_insert(0) += 1;
            }
            for (token, count) in frequencies {
                postings
                    .entry(token)
                    .or_default()
                    .push((entity_id.raw(), count));
            }
        }

        let terms = postings
            .into_iter()
            .map(|(term, postings)| reddb_file::NativeFulltextTerm {
                term,
                postings: postings
                    .into_iter()
                    .map(
                        |(entity_id, term_count)| reddb_file::NativeFulltextPosting {
                            entity_id,
                            term_count,
                        },
                    )
                    .collect(),
            })
            .collect();
        reddb_file::encode_native_fulltext_frame(&reddb_file::NativeFulltextFrame {
            collection: collection.to_string(),
            doc_count: documents.len() as u32,
            terms,
        })
    }

    pub(crate) fn inspect_native_fulltext_artifact(
        bytes: &[u8],
    ) -> Result<(u64, u64, u64), String> {
        let frame = reddb_file::decode_native_fulltext_frame(bytes).map_err(|e| e.to_string())?;
        let posting_count: u64 = frame
            .terms
            .iter()
            .map(|term| term.postings.len() as u64)
            .sum();
        Ok((
            frame.doc_count as u64,
            frame.terms.len() as u64,
            posting_count,
        ))
    }

    fn serialize_native_document_pathvalue_artifact(
        collection: &str,
        documents: &[(EntityId, Vec<(String, String)>)],
    ) -> Vec<u8> {
        let docs: Vec<reddb_file::NativeDocPathValue> = documents
            .iter()
            .map(|(entity_id, entries)| reddb_file::NativeDocPathValue {
                entity_id: entity_id.raw(),
                entries: entries
                    .iter()
                    .map(|(path, value)| reddb_file::NativeDocPathValueEntry {
                        path: path.clone(),
                        value: value.clone(),
                    })
                    .collect(),
            })
            .collect();
        reddb_file::encode_native_doc_pathvalue_frame(&reddb_file::NativeDocPathValueFrame {
            collection: collection.to_string(),
            docs,
        })
    }

    pub(crate) fn inspect_native_document_pathvalue_artifact(
        bytes: &[u8],
    ) -> Result<(u64, u64, u64, u64), String> {
        let frame =
            reddb_file::decode_native_doc_pathvalue_frame(bytes).map_err(|e| e.to_string())?;
        let doc_count = frame.docs.len() as u64;
        let mut paths = BTreeSet::new();
        let mut values = BTreeSet::new();
        let mut total_entries = 0u64;
        for doc in frame.docs {
            for entry in doc.entries {
                paths.insert(entry.path);
                values.insert(entry.value);
                total_entries += 1;
            }
        }
        Ok((
            doc_count,
            paths.len() as u64,
            total_entries,
            values.len() as u64,
        ))
    }

    fn native_document_pathvalue_for_entity(
        entity_id: EntityId,
        data: &EntityData,
        limit: usize,
    ) -> Result<Option<(EntityId, Vec<(String, String)>)>, ()> {
        let mut entries = Vec::new();
        let mut bytes = 0;
        match data {
            EntityData::Row(row) => {
                if row.named.is_some() || row.schema.is_some() {
                    for (key, value) in row.iter_fields() {
                        Self::collect_native_document_entries_from_value(
                            key,
                            value,
                            &mut entries,
                            &mut bytes,
                            limit,
                        )?;
                    }
                } else {
                    for (idx, value) in row.columns.iter().enumerate() {
                        let path = format!("columns[{idx}]");
                        Self::collect_native_document_entries_from_value(
                            &path,
                            value,
                            &mut entries,
                            &mut bytes,
                            limit,
                        )?;
                    }
                }
            }
            EntityData::Node(node) => {
                for (key, value) in &node.properties {
                    Self::collect_native_document_entries_from_value(
                        key,
                        value,
                        &mut entries,
                        &mut bytes,
                        limit,
                    )?;
                }
            }
            EntityData::Edge(edge) => {
                for (key, value) in &edge.properties {
                    Self::collect_native_document_entries_from_value(
                        key,
                        value,
                        &mut entries,
                        &mut bytes,
                        limit,
                    )?;
                }
            }
            EntityData::Vector(_) => {}
            EntityData::TimeSeries(_) => {}
            EntityData::QueueMessage(_) => {}
        }
        if entries.is_empty() {
            Ok(None)
        } else {
            Ok(Some((entity_id, entries)))
        }
    }

    fn collect_native_document_entries_from_value(
        path: &str,
        value: &Value,
        out: &mut Vec<(String, String)>,
        bytes: &mut usize,
        limit: usize,
    ) -> Result<(), ()> {
        match value {
            Value::Json(payload) | Value::Blob(payload) => {
                if let Ok(json) = crate::json::from_slice::<JsonValue>(payload) {
                    Self::collect_native_document_entries_from_json(
                        path, &json, out, bytes, limit,
                    )?;
                }
            }
            _ => {}
        }
        Ok(())
    }

    fn collect_native_document_entries_from_json(
        path: &str,
        value: &JsonValue,
        out: &mut Vec<(String, String)>,
        bytes: &mut usize,
        limit: usize,
    ) -> Result<(), ()> {
        let mut pending = vec![(path.to_string(), value)];
        let mut pending_bytes = path.len() + std::mem::size_of::<(String, &JsonValue)>();
        while let Some((path, value)) = pending.pop() {
            pending_bytes -= path.len() + std::mem::size_of::<(String, &JsonValue)>();
            match value {
                JsonValue::Object(entries) => {
                    for (key, value) in entries.iter().rev() {
                        let next = if path.is_empty() {
                            key.clone()
                        } else {
                            format!("{path}.{key}")
                        };
                        let cost = next.len() + std::mem::size_of::<(String, &JsonValue)>();
                        if cost > limit.saturating_sub(bytes.saturating_add(pending_bytes)) {
                            return Err(());
                        }
                        pending_bytes += cost;
                        pending.push((next, value));
                    }
                }
                JsonValue::Array(items) => {
                    for (idx, value) in items.iter().enumerate().rev() {
                        let next = format!("{path}[{idx}]");
                        let cost = next.len() + std::mem::size_of::<(String, &JsonValue)>();
                        if cost > limit.saturating_sub(bytes.saturating_add(pending_bytes)) {
                            return Err(());
                        }
                        pending_bytes += cost;
                        pending.push((next, value));
                    }
                }
                _ => {
                    if let Some(text) = Self::native_json_scalar_text(value) {
                        let cost =
                            path.len() + text.len() + std::mem::size_of::<(String, String)>();
                        if cost > limit.saturating_sub(bytes.saturating_add(pending_bytes)) {
                            return Err(());
                        }
                        *bytes += cost;
                        out.push((path, text));
                    }
                }
            }
        }
        Ok(())
    }

    fn native_json_scalar_text(value: &JsonValue) -> Option<String> {
        match value {
            JsonValue::Null => None,
            JsonValue::Bool(value) => Some(value.to_string()),
            JsonValue::Integer(value) => Some(value.to_string()),
            JsonValue::Number(value) => Some(value.to_string()),
            JsonValue::Decimal(value) => Some(value.clone()),
            JsonValue::String(value) => Some(value.clone()),
            JsonValue::Array(_) | JsonValue::Object(_) => None,
        }
    }

    fn native_fulltext_text_for_entity(data: &EntityData) -> String {
        match data {
            EntityData::Row(row) => {
                let mut parts = Vec::new();
                if let Some(named) = &row.named {
                    for value in named.values() {
                        if let Some(text) = Self::native_value_text(value) {
                            parts.push(text);
                        }
                    }
                }
                for value in &row.columns {
                    if let Some(text) = Self::native_value_text(value) {
                        parts.push(text);
                    }
                }
                parts.join(" ")
            }
            EntityData::Node(node) => node
                .properties
                .values()
                .filter_map(Self::native_value_text)
                .collect::<Vec<_>>()
                .join(" "),
            EntityData::Edge(edge) => edge
                .properties
                .values()
                .filter_map(Self::native_value_text)
                .collect::<Vec<_>>()
                .join(" "),
            EntityData::Vector(vector) => vector.content.clone().unwrap_or_default(),
            EntityData::TimeSeries(ts) => ts.metric.clone(),
            EntityData::QueueMessage(_) => String::new(),
        }
    }

    fn native_value_text(value: &Value) -> Option<String> {
        match value {
            Value::Text(value) => Some(value.to_string()),
            Value::Json(value) => String::from_utf8(value.clone()).ok(),
            Value::Blob(value) => String::from_utf8(value.clone()).ok(),
            Value::Integer(value) => Some(value.to_string()),
            Value::UnsignedInteger(value) => Some(value.to_string()),
            Value::Float(value) => Some(value.to_string()),
            Value::Boolean(value) => Some(value.to_string()),
            Value::IpAddr(value) => Some(value.to_string()),
            Value::NodeRef(value) => Some(value.clone()),
            Value::EdgeRef(value) => Some(value.clone()),
            Value::RowRef(table, row_id) => Some(format!("{table}:{row_id}")),
            Value::VectorRef(collection, vector_id) => Some(format!("{collection}:{vector_id}")),
            Value::Timestamp(value) => Some(value.to_string()),
            Value::Duration(value) => Some(value.to_string()),
            Value::Uuid(_) | Value::MacAddr(_) | Value::Vector(_) | Value::Null => None,
            Value::Color([r, g, b]) => Some(format!("#{:02X}{:02X}{:02X}", r, g, b)),
            Value::Email(s) => Some(s.clone()),
            Value::Url(s) => Some(s.clone()),
            Value::Phone(n) => Some(format!("+{}", n)),
            Value::Semver(packed) => Some(format!(
                "{}.{}.{}",
                packed / 1_000_000,
                (packed / 1_000) % 1_000,
                packed % 1_000
            )),
            Value::Cidr(ip, prefix) => Some(format!(
                "{}.{}.{}.{}/{}",
                (ip >> 24) & 0xFF,
                (ip >> 16) & 0xFF,
                (ip >> 8) & 0xFF,
                ip & 0xFF,
                prefix
            )),
            Value::Date(days) => Some(days.to_string()),
            Value::Time(ms) => {
                let total_secs = ms / 1000;
                Some(format!(
                    "{:02}:{:02}:{:02}",
                    total_secs / 3600,
                    (total_secs / 60) % 60,
                    total_secs % 60
                ))
            }
            Value::Decimal(v) => Some(Value::Decimal(*v).display_string()),
            Value::DecimalText(v) => Some(v.clone()),
            Value::EnumValue(i) => Some(format!("enum({})", i)),
            Value::Array(_) => None,
            Value::TimestampMs(ms) => Some(ms.to_string()),
            Value::Ipv4(ip) => Some(format!(
                "{}.{}.{}.{}",
                (ip >> 24) & 0xFF,
                (ip >> 16) & 0xFF,
                (ip >> 8) & 0xFF,
                ip & 0xFF
            )),
            Value::Ipv6(bytes) => Some(format!("{}", std::net::Ipv6Addr::from(*bytes))),
            Value::Subnet(ip, mask) => {
                let prefix = mask.leading_ones();
                Some(format!(
                    "{}.{}.{}.{}/{}",
                    (ip >> 24) & 0xFF,
                    (ip >> 16) & 0xFF,
                    (ip >> 8) & 0xFF,
                    ip & 0xFF,
                    prefix
                ))
            }
            Value::Port(p) => Some(p.to_string()),
            Value::Latitude(micro) => Some(format!("{:.6}", *micro as f64 / 1_000_000.0)),
            Value::Longitude(micro) => Some(format!("{:.6}", *micro as f64 / 1_000_000.0)),
            Value::GeoPoint(lat, lon) => Some(format!(
                "{:.6},{:.6}",
                *lat as f64 / 1_000_000.0,
                *lon as f64 / 1_000_000.0
            )),
            Value::Country2(c) => Some(String::from_utf8_lossy(c).to_string()),
            Value::Country3(c) => Some(String::from_utf8_lossy(c).to_string()),
            Value::Lang2(c) => Some(String::from_utf8_lossy(c).to_string()),
            Value::Lang5(c) => Some(String::from_utf8_lossy(c).to_string()),
            Value::Currency(c) => Some(String::from_utf8_lossy(c).to_string()),
            Value::AssetCode(code) => Some(code.clone()),
            Value::Money { .. } => Some(value.display_string()),
            Value::ColorAlpha([r, g, b, a]) => {
                Some(format!("#{:02X}{:02X}{:02X}{:02X}", r, g, b, a))
            }
            Value::BigInt(v) => Some(v.to_string()),
            Value::KeyRef(col, key) => Some(format!("{}:{}", col, key)),
            Value::DocRef(col, id) => Some(format!("{}#{}", col, id)),
            Value::TableRef(name) => Some(name.clone()),
            Value::PageRef(page_id) => Some(format!("page:{}", page_id)),
            Value::Secret(_) | Value::Password(_) => None,
        }
    }

    fn native_fulltext_tokenize(text: &str) -> Vec<String> {
        text.to_lowercase()
            .split(|c: char| !c.is_alphanumeric())
            .filter(|s| s.len() >= 2)
            .map(|s| s.to_string())
            .collect()
    }

    pub(crate) fn native_recovery_summary_from_metadata(
        metadata: &PhysicalMetadataFile,
    ) -> NativeRecoverySummary {
        const SAMPLE_LIMIT: usize = 16;

        let snapshots: Vec<_> = metadata
            .snapshots
            .iter()
            .rev()
            .take(SAMPLE_LIMIT)
            .map(|snapshot| NativeSnapshotSummary {
                snapshot_id: snapshot.snapshot_id,
                created_at_unix_ms: snapshot.created_at_unix_ms,
                superblock_sequence: snapshot.superblock_sequence,
                collection_count: snapshot.collection_count as u32,
                total_entities: snapshot.total_entities as u64,
            })
            .collect();
        let exports: Vec<_> = metadata
            .exports
            .iter()
            .rev()
            .take(SAMPLE_LIMIT)
            .map(|export| NativeExportSummary {
                name: export.name.clone(),
                created_at_unix_ms: export.created_at_unix_ms,
                snapshot_id: export.snapshot_id,
                superblock_sequence: export.superblock_sequence,
                collection_count: export.collection_count as u32,
                total_entities: export.total_entities as u64,
            })
            .collect();

        NativeRecoverySummary {
            snapshot_count: metadata.snapshots.len() as u32,
            export_count: metadata.exports.len() as u32,
            snapshots_complete: metadata.snapshots.len() <= SAMPLE_LIMIT,
            exports_complete: metadata.exports.len() <= SAMPLE_LIMIT,
            omitted_snapshot_count: metadata.snapshots.len().saturating_sub(snapshots.len()) as u32,
            omitted_export_count: metadata.exports.len().saturating_sub(exports.len()) as u32,
            snapshots,
            exports,
        }
    }

    pub(crate) fn native_catalog_summary_from_metadata(
        metadata: &PhysicalMetadataFile,
    ) -> NativeCatalogSummary {
        const SAMPLE_LIMIT: usize = 32;

        let collections: Vec<_> = metadata
            .catalog
            .stats_by_collection
            .iter()
            .take(SAMPLE_LIMIT)
            .map(|(name, stats)| NativeCatalogCollectionSummary {
                name: name.clone(),
                entities: stats.entities as u64,
                cross_refs: stats.cross_refs as u64,
                segments: stats.segments as u32,
            })
            .collect();

        NativeCatalogSummary {
            collection_count: metadata.catalog.total_collections as u32,
            total_entities: metadata.catalog.total_entities as u64,
            collections_complete: metadata.catalog.stats_by_collection.len() <= SAMPLE_LIMIT,
            omitted_collection_count: metadata
                .catalog
                .stats_by_collection
                .len()
                .saturating_sub(collections.len()) as u32,
            collections,
        }
    }

    pub(crate) fn native_metadata_state_summary_from_metadata(
        metadata: &PhysicalMetadataFile,
    ) -> NativeMetadataStateSummary {
        NativeMetadataStateSummary {
            protocol_version: metadata.protocol_version.clone(),
            generated_at_unix_ms: metadata.generated_at_unix_ms,
            last_loaded_from: metadata.last_loaded_from.clone(),
            last_healed_at_unix_ms: metadata.last_healed_at_unix_ms,
        }
    }

    pub(crate) fn inspect_native_header_against_metadata(
        native: PhysicalFileHeader,
        metadata: &PhysicalMetadataFile,
    ) -> NativeHeaderInspection {
        let expected = Self::native_header_from_metadata(metadata);
        let mut mismatches = Vec::new();

        if native.format_version != expected.format_version {
            mismatches.push(NativeHeaderMismatch {
                field: "format_version",
                native: native.format_version.to_string(),
                expected: expected.format_version.to_string(),
            });
        }
        if native.sequence != expected.sequence {
            mismatches.push(NativeHeaderMismatch {
                field: "sequence",
                native: native.sequence.to_string(),
                expected: expected.sequence.to_string(),
            });
        }
        if native.manifest_oldest_root != expected.manifest_oldest_root {
            mismatches.push(NativeHeaderMismatch {
                field: "manifest_oldest_root",
                native: native.manifest_oldest_root.to_string(),
                expected: expected.manifest_oldest_root.to_string(),
            });
        }
        if native.manifest_root != expected.manifest_root {
            mismatches.push(NativeHeaderMismatch {
                field: "manifest_root",
                native: native.manifest_root.to_string(),
                expected: expected.manifest_root.to_string(),
            });
        }
        if native.free_set_root != expected.free_set_root {
            mismatches.push(NativeHeaderMismatch {
                field: "free_set_root",
                native: native.free_set_root.to_string(),
                expected: expected.free_set_root.to_string(),
            });
        }
        if native.collection_root_count != expected.collection_root_count {
            mismatches.push(NativeHeaderMismatch {
                field: "collection_root_count",
                native: native.collection_root_count.to_string(),
                expected: expected.collection_root_count.to_string(),
            });
        }
        if native.snapshot_count != expected.snapshot_count {
            mismatches.push(NativeHeaderMismatch {
                field: "snapshot_count",
                native: native.snapshot_count.to_string(),
                expected: expected.snapshot_count.to_string(),
            });
        }
        if native.index_count != expected.index_count {
            mismatches.push(NativeHeaderMismatch {
                field: "index_count",
                native: native.index_count.to_string(),
                expected: expected.index_count.to_string(),
            });
        }
        if native.catalog_collection_count != expected.catalog_collection_count {
            mismatches.push(NativeHeaderMismatch {
                field: "catalog_collection_count",
                native: native.catalog_collection_count.to_string(),
                expected: expected.catalog_collection_count.to_string(),
            });
        }
        if native.catalog_total_entities != expected.catalog_total_entities {
            mismatches.push(NativeHeaderMismatch {
                field: "catalog_total_entities",
                native: native.catalog_total_entities.to_string(),
                expected: expected.catalog_total_entities.to_string(),
            });
        }
        if native.export_count != expected.export_count {
            mismatches.push(NativeHeaderMismatch {
                field: "export_count",
                native: native.export_count.to_string(),
                expected: expected.export_count.to_string(),
            });
        }
        if native.graph_projection_count != expected.graph_projection_count {
            mismatches.push(NativeHeaderMismatch {
                field: "graph_projection_count",
                native: native.graph_projection_count.to_string(),
                expected: expected.graph_projection_count.to_string(),
            });
        }
        if native.analytics_job_count != expected.analytics_job_count {
            mismatches.push(NativeHeaderMismatch {
                field: "analytics_job_count",
                native: native.analytics_job_count.to_string(),
                expected: expected.analytics_job_count.to_string(),
            });
        }
        if native.manifest_event_count != expected.manifest_event_count {
            mismatches.push(NativeHeaderMismatch {
                field: "manifest_event_count",
                native: native.manifest_event_count.to_string(),
                expected: expected.manifest_event_count.to_string(),
            });
        }

        NativeHeaderInspection {
            native,
            expected,
            consistent: mismatches.is_empty(),
            mismatches,
        }
    }

    pub(crate) fn repair_policy_for_inspection(
        inspection: &NativeHeaderInspection,
    ) -> NativeHeaderRepairPolicy {
        if inspection.consistent {
            return NativeHeaderRepairPolicy::InSync;
        }

        if inspection.expected.sequence >= inspection.native.sequence {
            NativeHeaderRepairPolicy::RepairNativeFromMetadata
        } else {
            NativeHeaderRepairPolicy::NativeAheadOfMetadata
        }
    }

    pub(crate) fn prune_export_registry(&self, exports: &mut Vec<ExportDescriptor>) {
        let retention = self.options.export_retention.max(1);
        if exports.len() <= retention {
            return;
        }

        exports.sort_by_key(|export| export.created_at_unix_ms);
        let removed: Vec<ExportDescriptor> =
            exports.drain(0..(exports.len() - retention)).collect();

        for export in removed {
            let _ = fs::remove_file(&export.data_path);
            let _ = fs::remove_file(&export.metadata_path);
            let binary_path = PhysicalMetadataFile::metadata_binary_path_for(std::path::Path::new(
                &export.data_path,
            ));
            let _ = fs::remove_file(binary_path);
        }
    }

    pub(crate) fn runtime_index_catalog(&self) -> IndexCatalog {
        let mut catalog = IndexCatalog::register_default_vector_graph(
            self.options.has_capability(Capability::Table),
            self.options.has_capability(Capability::Graph),
        );
        if self.options.has_capability(Capability::FullText) {
            catalog.register(RuntimeIndexConfig::new(
                "text-fulltext",
                IndexKind::FullText,
            ));
            catalog.register(RuntimeIndexConfig::new(
                "document-pathvalue",
                IndexKind::DocumentPathValue,
            ));
        }
        catalog.register(RuntimeIndexConfig::new(
            "search-hybrid",
            IndexKind::HybridSearch,
        ));
        catalog
    }

    pub(crate) fn physical_index_state(&self) -> Vec<PhysicalIndexState> {
        // Use a lightweight catalog snapshot that does NOT call physical_metadata()
        // to avoid infinite recursion: physical_metadata → metadata_from_native_state
        // → physical_index_state → catalog_model_snapshot → physical_metadata → ...
        let catalog = self.runtime_index_catalog();
        let snapshot = crate::catalog::snapshot_store_with_declarations(
            "reddb",
            self.store.as_ref(),
            Some(&catalog),
            None, // No declarations — breaks the recursive cycle
            None, // No contracts — avoids pulling physical metadata during bootstrap
        );
        let mut metrics_by_name = std::collections::BTreeMap::new();
        for metric in &snapshot.indices {
            metrics_by_name.insert(metric.name.clone(), metric.clone());
        }

        let mut states = Vec::new();
        for collection in snapshot.collections {
            // This infallible snapshot only observes resident entities. A cold
            // base, including one with a resident overlay, cannot supply an
            // inferred model here. The caller reconciles its durable index
            // descriptors from the native registry and physical metadata.
            // Treating an unobserved base as Mixed invents graph/search indexes
            // on ordinary tables each time metadata is republished.
            if self
                .store
                .get_collection(&collection.name)
                .is_some_and(|manager| manager.has_backing())
            {
                continue;
            }
            for index_name in &collection.indices {
                let metric = metrics_by_name.get(index_name);
                let kind = metric
                    .map(|metric| metric.kind)
                    .unwrap_or_else(|| infer_collection_index_kind(collection.model, index_name));
                let entries = estimate_index_entries(&collection, kind);
                states.push(PhysicalIndexState {
                    name: format!("{}::{}", collection.name, index_name),
                    kind,
                    collection: Some(collection.name.clone()),
                    enabled: metric.map(|metric| metric.enabled).unwrap_or(true),
                    entries,
                    estimated_memory_bytes: estimate_index_memory(entries, kind),
                    last_refresh_ms: metric.and_then(|metric| metric.last_refresh_ms),
                    backend: index_backend_name(kind).to_string(),
                    artifact_kind: None,
                    artifact_root_page: None,
                    artifact_checksum: None,
                    build_state: "catalog-derived".to_string(),
                });
            }
        }

        states
    }

    pub(crate) fn physical_collection_roots(&self) -> BTreeMap<String, u64> {
        let mut roots = BTreeMap::new();

        for name in self.store.list_collections() {
            let Some(manager) = self.store.get_collection(&name) else {
                continue;
            };

            let stats = manager.stats();
            let mut root = fnv1a_seed();
            fnv1a_hash_value(&mut root, &name);
            fnv1a_hash_value(&mut root, &stats.total_entities);
            fnv1a_hash_value(&mut root, &stats.growing_count);
            fnv1a_hash_value(&mut root, &stats.sealed_count);
            fnv1a_hash_value(&mut root, &stats.archived_count);
            fnv1a_hash_value(&mut root, &stats.total_memory_bytes);
            fnv1a_hash_value(&mut root, &stats.seal_ops);
            fnv1a_hash_value(&mut root, &stats.consolidation.runs_completed);

            let root = manager.cached_physical_root(root, || {
                manager.for_each_entity_in_id_order(|entity| {
                    record_physical_root_row_scan(&name);
                    fnv1a_hash_value(&mut root, &entity.id.raw());
                    fnv1a_hash_value(&mut root, &entity.kind);
                    fnv1a_hash_value(&mut root, &entity.created_at);
                    fnv1a_hash_value(&mut root, &entity.updated_at);
                    fnv1a_hash_value(&mut root, &entity.data);
                    fnv1a_hash_value(&mut root, &entity.sequence_id);
                    fnv1a_hash_value(&mut root, &entity.embeddings().len());
                    fnv1a_hash_value(&mut root, &entity.cross_refs().len());
                    true
                });
                root
            });

            roots.insert(name, root);
        }

        roots
    }

    // ========================================================================
    // Reference Helpers - For Metadata Linking
    // ========================================================================

    /// Create a reference to a table row
    pub fn table_ref(&self, table: impl Into<String>, row_id: u64) -> TableRef {
        TableRef::new(table, row_id)
    }

    /// Create a reference to a graph node
    pub fn node_ref(&self, collection: impl Into<String>, node_id: EntityId) -> NodeRef {
        NodeRef::new(collection, node_id)
    }

    /// Create a reference to a vector
    pub fn vector_ref(&self, collection: impl Into<String>, vector_id: EntityId) -> VectorRef {
        VectorRef::new(collection, vector_id)
    }

    // ========================================================================
    // Query API
    // ========================================================================

    /// Start building a query
    pub fn query(&self) -> QueryBuilder {
        QueryBuilder::new(self.store.clone())
    }

    /// Quick vector similarity search.
    ///
    /// For collections with >= 100 vectors a lazily-built HNSW index is used
    /// for fast approximate nearest-neighbor lookup.  Smaller collections fall
    /// back to an exact brute-force scan so that the overhead of building an
    /// index is avoided when it would not pay off.
    pub fn similar(&self, collection: &str, vector: &[f32], k: usize) -> Vec<SimilarResult> {
        if self.store.get_collection(collection).is_none() {
            return Vec::new();
        }

        // Try the HNSW fast path for collections with enough vectors.
        if let Some(index) = self.get_or_build_hnsw_index(collection, vector.len()) {
            let hnsw = index.read().unwrap_or_else(|e| e.into_inner());
            let results = hnsw.search(vector, k);
            let mapped = self.hnsw_results_to_similar(collection, &results);
            if !mapped.is_empty() {
                return mapped;
            }
        }

        // Fallback: brute-force scan (small / mixed-type collections).
        self.similar_brute_force(collection, vector, k)
    }

    /// Brute-force cosine similarity scan (exact results, O(n)).
    fn similar_brute_force(
        &self,
        collection: &str,
        vector: &[f32],
        k: usize,
    ) -> Vec<SimilarResult> {
        let manager = match self.store.get_collection(collection) {
            Some(m) => m,
            None => return Vec::new(),
        };

        let entities = manager.query_all(|_| true);
        let mut results: Vec<SimilarResult> = entities
            .iter()
            .filter_map(|e| {
                let score = match &e.data {
                    EntityData::Vector(v) => cosine_similarity(vector, &v.dense),
                    _ => e
                        .embeddings()
                        .iter()
                        .map(|emb| cosine_similarity(vector, &emb.vector))
                        .fold(0.0f32, f32::max),
                };
                let distance = (1.0 - score).max(0.0);
                if score > 0.0 {
                    Some(SimilarResult {
                        entity_id: e.id,
                        score,
                        distance,
                        entity: e.clone(),
                    })
                } else {
                    None
                }
            })
            .collect();

        results.sort_by(|a, b| {
            b.score
                .partial_cmp(&a.score)
                .unwrap_or(std::cmp::Ordering::Equal)
        });
        results.truncate(k);
        results
    }

    /// Return (or lazily build) a per-collection HNSW index.
    ///
    /// Returns `None` when the collection has fewer than 100 dense vectors
    /// (the brute-force path is cheaper for tiny collections) or when there
    /// is a dimension mismatch with the query vector.
    ///
    /// The cached index is automatically invalidated when the live entity
    /// count in the collection changes, so inserts and deletes are picked up
    /// transparently without requiring explicit invalidation calls.
    fn get_or_build_hnsw_index(
        &self,
        collection: &str,
        query_dim: usize,
    ) -> Option<Arc<RwLock<HnswIndex>>> {
        let manager = self.store.get_collection(collection)?;
        let live_count = manager.count();

        // Fast path: check if a fresh index already exists.
        {
            let indexes = self
                .vector_indexes
                .read()
                .unwrap_or_else(|e| e.into_inner());
            if let Some(cached) = indexes.get(collection) {
                if cached.entity_count == live_count {
                    return Some(Arc::clone(&cached.index));
                }
            }
        }

        // Either no cached index exists or it is stale -- (re)build it.
        let entities = manager.query_all(|_| true);

        let vectors: Vec<(u64, Vec<f32>)> = entities
            .iter()
            .filter_map(|e| match &e.data {
                EntityData::Vector(v) if !v.dense.is_empty() && v.dense.len() == query_dim => {
                    Some((e.id.raw(), v.dense.clone()))
                }
                _ => None,
            })
            .collect();

        // Only build the HNSW index when there are enough vectors to justify it.
        const MIN_VECTORS_FOR_HNSW: usize = 100;
        if vectors.len() < MIN_VECTORS_FOR_HNSW {
            return None;
        }

        // Build the HNSW index with cosine distance (matching the brute-force path).
        let config = crate::storage::engine::HnswConfig::with_m(16)
            .with_metric(crate::storage::engine::DistanceMetric::Cosine)
            .with_ef_construction(100)
            .with_ef_search(50);
        let mut hnsw = HnswIndex::new(query_dim, config);

        for (id, vec) in &vectors {
            hnsw.insert_with_id(*id, vec.clone());
        }

        let index = Arc::new(RwLock::new(hnsw));

        // Store in the cache (double-check pattern to avoid duplicate builds).
        let mut indexes = self
            .vector_indexes
            .write()
            .unwrap_or_else(|e| e.into_inner());
        // Re-check under write lock: another thread may have built in the meantime.
        if let Some(cached) = indexes.get(collection) {
            if cached.entity_count == live_count {
                return Some(Arc::clone(&cached.index));
            }
        }
        indexes.insert(
            collection.to_string(),
            CachedVectorIndex {
                index: Arc::clone(&index),
                entity_count: live_count,
            },
        );
        Some(index)
    }

    /// Convert HNSW `DistanceResult`s back to the DevX `SimilarResult` type.
    fn hnsw_results_to_similar(
        &self,
        collection: &str,
        results: &[crate::storage::engine::DistanceResult],
    ) -> Vec<SimilarResult> {
        results
            .iter()
            .filter_map(|dr| {
                let entity_id = EntityId::new(dr.id);
                let entity = self.store.get(collection, entity_id)?;
                // Cosine distance = 1 - similarity.
                let score = (1.0 - dr.distance).max(0.0);
                if score > 0.0 {
                    Some(SimilarResult {
                        entity_id,
                        score,
                        distance: dr.distance,
                        entity,
                    })
                } else {
                    None
                }
            })
            .collect()
    }

    /// Invalidate the cached HNSW index for a collection.
    ///
    /// Called after vector inserts / deletes so the next search lazily rebuilds
    /// a fresh index that includes the new data.
    pub(crate) fn invalidate_vector_index(&self, collection: &str) {
        let mut indexes = self
            .vector_indexes
            .write()
            .unwrap_or_else(|e| e.into_inner());
        indexes.remove(collection);
    }

    /// Get entity by ID from any collection
    pub fn get(&self, id: EntityId) -> Option<UnifiedEntity> {
        self.store.get_any(id).map(|(_, e)| e)
    }

    /// Get entity with its collection name
    pub fn get_with_collection(&self, id: EntityId) -> Option<(String, UnifiedEntity)> {
        self.store.get_any(id)
    }

    // ========================================================================
    // Batch Operations - Performance
    // ========================================================================

    /// Batch get multiple entities by ID. More efficient than N individual get() calls.
    pub fn batch_get(&self, ids: &[EntityId]) -> Vec<Option<UnifiedEntity>> {
        ids.iter().map(|id| self.get(*id)).collect()
    }

    /// Start a batch operation for bulk inserts
    pub fn batch(&self) -> BatchBuilder {
        BatchBuilder::new(self.store.clone(), self.preprocessors.clone())
    }

    // ========================================================================
    // Preprocessing
    // ========================================================================

    /// Add a preprocessor hook
    pub fn add_preprocessor(&mut self, preprocessor: Box<dyn Preprocessor>) {
        let mut preprocessors = self
            .preprocessors
            .write()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        preprocessors.push(Arc::from(preprocessor));
    }

    // ========================================================================
    // Cross-Reference Navigation
    // ========================================================================

    /// Get all entities linked FROM the given entity
    pub fn linked_from(&self, id: EntityId) -> Vec<LinkedEntity> {
        self.store
            .get_refs_from(id)
            .into_iter()
            .filter_map(|(target_id, ref_type, collection)| {
                self.store
                    .get(&collection, target_id)
                    .map(|entity| LinkedEntity {
                        entity,
                        ref_type,
                        collection,
                    })
            })
            .collect()
    }

    /// Get all entities linked TO the given entity
    pub fn linked_to(&self, id: EntityId) -> Vec<LinkedEntity> {
        self.store
            .get_refs_to(id)
            .into_iter()
            .filter_map(|(source_id, ref_type, collection)| {
                self.store
                    .get(&collection, source_id)
                    .map(|entity| LinkedEntity {
                        entity,
                        ref_type,
                        collection,
                    })
            })
            .collect()
    }

    /// Get the underlying store (for advanced operations)
    pub fn store(&self) -> Arc<UnifiedStore> {
        self.store.clone()
    }

    /// Per-process registry of `vector.turbo` collection state (issue
    /// #693). Lazily allocated on first access; the inner map holds an
    /// `Arc<TurboCollectionState>` per turbo collection so concurrent
    /// SEARCHers and INSERTers can share the in-memory index without
    /// holding the outer map lock for the whole operation.
    pub(crate) fn turbo_collections(
        &self,
    ) -> &Arc<
        parking_lot::Mutex<
            std::collections::HashMap<
                String,
                Arc<crate::runtime::vector_turbo_kind::TurboCollectionState>,
            >,
        >,
    > {
        self.turbo_collections
            .get_or_init(|| Arc::new(parking_lot::Mutex::new(std::collections::HashMap::new())))
    }

    /// Look up (or lazily create) the runtime state for a turbo
    /// collection. Returns `None` for collections that are not marked
    /// `turbo` in the catalog.
    ///
    /// Lazy creation matters for two reasons: (a) a fresh process
    /// reopens a persistent DB without any turbo state in memory; (b)
    /// the contract (dim + metric) is read here instead of being
    /// stored separately, which keeps the catalog as the single source
    /// of truth.
    pub(crate) fn turbo_state(
        &self,
        collection: &str,
    ) -> Option<Arc<crate::runtime::vector_turbo_kind::TurboCollectionState>> {
        if !crate::runtime::vector_turbo_kind::is_turbo(&self.store, collection) {
            return None;
        }
        let map = self.turbo_collections();
        {
            let guard = map.lock();
            if let Some(state) = guard.get(collection) {
                return Some(Arc::clone(state));
            }
        }
        let contract = self.collection_contract(collection)?;
        let dim = contract.vector_dimension?;
        let metric = contract
            .vector_metric
            .unwrap_or(crate::storage::engine::distance::DistanceMetric::Cosine);
        let state = Arc::new(
            crate::runtime::vector_turbo_kind::TurboCollectionState::new(
                dim,
                metric,
                self.store.pager(),
            ),
        );
        // Issue #674 — derive the `.tv` snapshot path from the
        // resolved tier layout if the active layout enables snapshot
        // files. `Minimal` (embedded) returns `None` and the state
        // never writes a `.tv` for this collection.
        if let Some((_, paths)) = self.options.resolve_tiered_layout() {
            state.set_snapshot_path(paths.turbo_snapshot_path(collection));
        }
        let mut guard = map.lock();
        let inserted_now = !guard.contains_key(collection);
        let entry = guard
            .entry(collection.to_string())
            .or_insert_with(|| Arc::clone(&state));
        let handle = Arc::clone(entry);
        drop(guard);
        // Issue #673 — first materialisation of this collection's
        // state kicks off a background rebuild worker. Non-vector
        // traffic is unaffected (it never reaches `turbo_state`);
        // vector SEARCH/INSERT observes the readiness flag via
        // `wait_until_ready` instead of blocking inline on the
        // synchronous lazy populate.
        if inserted_now {
            let join = crate::runtime::vector_turbo_kind::spawn_background_rebuild(
                Arc::clone(&self.store),
                collection.to_string(),
                Arc::clone(&handle),
            );
            self.turbo_rebuild_workers.lock().push(join);
        }
        Some(handle)
    }

    /// Lazily-initialised ML runtime. First caller wins; subsequent
    /// callers observe the same instance. The runtime is created
    /// with an in-memory persistence backend by default — production
    /// deployments that need durable model state will swap this for
    /// a store-backed backend when the persistence integration lands.
    pub fn ml_runtime(&self) -> &crate::storage::ml::MlRuntime {
        self.ml_runtime.get_or_init(|| {
            crate::storage::ml::MlRuntime::in_memory(std::sync::Arc::new(
                // No-op work fn — the SQL scalars call predict
                // synchronously; async training jobs come in a later
                // sprint alongside `CREATE MODEL`.
                |_handle| Ok(String::new()),
            ))
        })
    }

    /// Shared semantic cache for `SEMANTIC_CACHE_*` scalars. Uses
    /// default config until a `SET SEMANTIC_CACHE` admin surface
    /// lands.
    pub fn semantic_cache(&self) -> &Arc<crate::storage::ml::SemanticCache> {
        self.semantic_cache.get_or_init(|| {
            Arc::new(crate::storage::ml::SemanticCache::new(
                crate::storage::ml::SemanticCacheConfig::default(),
            ))
        })
    }

    /// Lazily-initialised hypertable registry. Populated by `CREATE
    /// HYPERTABLE` DDL; consumed by chunk routing + retention.
    pub fn hypertables(&self) -> &Arc<crate::storage::timeseries::HypertableRegistry> {
        self.hypertables
            .get_or_init(|| Arc::new(crate::storage::timeseries::HypertableRegistry::new()))
    }

    /// Lazily-initialised continuous-aggregate engine. Populated by
    /// `CA_REGISTER` and read by `CA_REFRESH` / `CA_STATE` scalars.
    pub fn continuous_aggregates(
        &self,
    ) -> &Arc<crate::storage::timeseries::continuous_aggregate::ContinuousAggregateEngine> {
        self.continuous_aggregates.get_or_init(|| {
            Arc::new(
                crate::storage::timeseries::continuous_aggregate::ContinuousAggregateEngine::new(),
            )
        })
    }

    pub(crate) fn is_binary_dump(path: &Path) -> Result<bool, std::io::Error> {
        let mut file = File::open(path)?;
        let mut magic = [0u8; 4];
        let read = file.read(&mut magic)?;
        Ok(read == 4 && reddb_file::native_store_magic_matches(&magic))
    }
}

#[cfg(not(test))]
fn record_physical_root_row_scan(_collection: &str) {}

#[cfg(test)]
std::thread_local! {
    static PHYSICAL_ROOT_ROW_SCANS: std::cell::RefCell<std::collections::BTreeMap<String, u64>> =
        const { std::cell::RefCell::new(std::collections::BTreeMap::new()) };
}

#[cfg(test)]
fn record_physical_root_row_scan(collection: &str) {
    PHYSICAL_ROOT_ROW_SCANS.with(|scans| {
        *scans
            .borrow_mut()
            .entry(collection.to_string())
            .or_default() += 1;
    });
}

#[cfg(test)]
impl RedDB {
    pub(crate) fn reset_physical_root_row_scans_for_test(&self) {
        PHYSICAL_ROOT_ROW_SCANS.with(|scans| scans.borrow_mut().clear());
    }

    pub(crate) fn physical_root_row_scans_for_test(
        &self,
    ) -> std::collections::BTreeMap<String, u64> {
        PHYSICAL_ROOT_ROW_SCANS.with(|scans| scans.borrow().clone())
    }
}

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

    #[test]
    fn physical_roots_hash_borrowed_payloads_without_rendered_copies() {
        fn legacy_root(db: &RedDB) -> u64 {
            fn hash<T: Debug>(root: &mut u64, value: &T) {
                fnv1a_hash_bytes(root, format!("{value:?}").as_bytes());
            }
            let manager = db.store().get_collection("root_source").expect("source");
            let stats = manager.stats();
            let mut root = fnv1a_seed();
            hash(&mut root, &"root_source".to_string());
            hash(&mut root, &stats.total_entities);
            hash(&mut root, &stats.growing_count);
            hash(&mut root, &stats.sealed_count);
            hash(&mut root, &stats.archived_count);
            hash(&mut root, &stats.total_memory_bytes);
            hash(&mut root, &stats.seal_ops);
            hash(&mut root, &stats.consolidation.runs_completed);
            manager.for_each_entity_in_id_order(|entity| {
                hash(&mut root, &entity.id.raw());
                hash(&mut root, &entity.kind);
                hash(&mut root, &entity.created_at);
                hash(&mut root, &entity.updated_at);
                hash(&mut root, &entity.data);
                hash(&mut root, &entity.sequence_id);
                hash(&mut root, &entity.embeddings().len());
                hash(&mut root, &entity.cross_refs().len());
                true
            });
            root
        }

        let runtime =
            crate::RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("runtime");
        runtime
            .execute_query("CREATE TABLE root_source (payload BLOB, label TEXT, body JSON)")
            .expect("table");
        let db = runtime.db();
        let manager = db.store().get_collection("root_source").expect("source");
        let mut entity = UnifiedEntity::table_row(
            EntityId::new(1001),
            "root_source",
            1,
            vec![
                Value::Blob(vec![255; 2 * 1024 * 1024]),
                Value::text("escaped\n\t\0\"\\ and Unicode ç🦀"),
                Value::Json(br#"{"nested":[null,true,17,"kept"]}"#.to_vec()),
            ],
        );
        manager.insert(entity.clone()).expect("source row");
        let expected = legacy_root(&db);
        let (roots, measured) = crate::storage::no_malloc_hot_paths::measure_allocations(|| {
            db.physical_collection_roots()
        });
        assert_eq!(roots["root_source"], expected, "retain legacy hash bytes");
        eprintln!(
            "physical root allocation bytes: {}",
            measured.allocated_bytes
        );
        assert!(
            measured.allocated_bytes < 64 * 1024,
            "fingerprinting must not render the borrowed payload: {measured:?}"
        );
        assert_eq!(db.physical_collection_roots(), roots, "unchanged cache");
        let EntityData::Row(row) = &mut entity.data else {
            unreachable!("fixture is a table row");
        };
        row.columns[0] = Value::Blob(vec![254; 2 * 1024 * 1024]);
        manager
            .update(entity)
            .expect("replace payload of equal size");
        let updated = db.physical_collection_roots();
        assert_ne!(updated["root_source"], roots["root_source"]);
        assert_eq!(updated["root_source"], legacy_root(&db));
    }

    #[test]
    fn document_artifact_paths_survive_shared_recovery_shapes() {
        let mut entity = UnifiedEntity::table_row(EntityId::new(1), "documents", 1, vec![]);
        entity.data = EntityData::Row(crate::storage::unified::entity::RowData::with_names(
            vec![Value::Json(br#"{"thing":"kept"}"#.to_vec())],
            vec!["body".into()],
        ));
        let before = RedDB::native_document_pathvalue_for_entity(entity.id, &entity.data, 4096)
            .expect("named document");
        let EntityData::Row(row) = &mut entity.data else {
            unreachable!("fixture is a row");
        };
        *row = crate::storage::unified::entity::RowData {
            columns: vec![row.get_field("body").expect("body field").clone()],
            named: None,
            schema: Some(Arc::new(vec!["body".into()])),
        };
        let after = RedDB::native_document_pathvalue_for_entity(entity.id, &entity.data, 4096)
            .expect("shared document");
        assert_eq!(
            before, after,
            "compaction must preserve named document paths"
        );
    }

    #[test]
    fn disabled_fulltext_does_not_copy_rows_or_publish_search_artifacts() {
        let runtime =
            crate::RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("runtime");
        runtime
            .execute_query("CREATE TABLE source (payload BLOB)")
            .expect("table");
        let db = runtime.db();
        let manager = db.store().get_collection("source").expect("manager");
        let (_, baseline) = crate::storage::no_malloc_hot_paths::measure_allocations(|| {
            db.native_vector_artifact_records().expect("empty build")
        });
        for id in 1..=32 {
            manager
                .insert(UnifiedEntity::table_row(
                    EntityId::new(1000 + id),
                    "source",
                    id,
                    vec![Value::Blob(vec![255; 64 * 1024])],
                ))
                .expect("source row");
        }
        let (build, measured) = crate::storage::no_malloc_hot_paths::measure_allocations(|| {
            db.native_vector_artifact_records().expect("disabled build")
        });
        assert!(build.records.is_empty());
        assert!(
            build.deferred_collections.is_empty(),
            "disabled capabilities do not consume artifact allowance"
        );
        assert!(
            measured.allocated_bytes <= baseline.allocated_bytes + 4096,
            "disabled fulltext copied source payloads: {baseline:?} -> {measured:?}"
        );
    }

    #[test]
    fn document_artifact_traversal_is_ordered_and_bounded() {
        let json = crate::json::from_slice::<JsonValue>(br#"{"a":{"b":1},"z":[2,3]}"#)
            .expect("JSON fixture");
        let mut entries = Vec::new();
        let mut bytes = 0;
        RedDB::collect_native_document_entries_from_json(
            "body",
            &json,
            &mut entries,
            &mut bytes,
            4096,
        )
        .expect("small document fits");
        assert_eq!(
            entries,
            vec![
                ("body.a.b".into(), "1".into()),
                ("body.z[0]".into(), "2".into()),
                ("body.z[1]".into(), "3".into())
            ]
        );
        let mut deep = JsonValue::Integer(1);
        for _ in 0..64 {
            deep = JsonValue::Object([("child".to_string(), deep)].into());
        }
        let mut entries = Vec::new();
        let mut bytes = 0;
        assert!(RedDB::collect_native_document_entries_from_json(
            "body",
            &deep,
            &mut entries,
            &mut bytes,
            128
        )
        .is_err());
        assert!(bytes <= 128);
    }
}