hawdb-storage 0.16.0

// Copyright 2026 Nowledge
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! Read, scan, seek, and scan-pruning methods for [`GraphStore`].

use super::*;
use crate::read_view::{
    AdmittedNodeRead, AdmittedNodeRecord, AdmittedRelationshipRead, AdmittedRelationshipRecord,
    ControlledGraphReadAllocator, GraphReadAdmission, GraphReadAllocation,
};
use hawdb_core::ids::{
    node_allocation_bytes, project_node_record_ref, projected_node_allocation_bytes,
};
#[path = "graph_read/admission.rs"]
mod admission;
#[path = "graph_read/ordered_composite.rs"]
mod ordered_composite;
#[path = "graph_read/resident_pruning.rs"]
mod resident_pruning;
use admission::{AdmittedProjection, ProjectionAdmission, UnaccountedReadAllocation};

struct PropertyRangeBounds<'a> {
    lower: Option<&'a (Value, bool)>,
    upper: Option<&'a (Value, bool)>,
}

enum RangeRecord<'a> {
    Live(&'a NodeRecord),
    Canonical(&'a CanonicalSegmentReader, NodeId),
}

#[cfg(test)]
#[path = "graph_read/ordered_range_tests.rs"]
mod ordered_range_tests;

#[cfg(test)]
#[path = "graph_read/retained_scan_tests.rs"]
mod retained_scan_tests;

/// Root-internal immutable materialized-node source. Capturing this source
/// shares only the node directory/pages and fail-closed poison signals, without
/// keeping relationships, indexes, statistics, file leases or a query cache.
/// It does not borrow a mutable store and does not decode canonical data.
#[doc(hidden)]
#[derive(Debug)]
pub struct MaterializedNodeReadSource {
    nodes: crate::cow::CowNodeMap,
    integrity_poisoned: Arc<AtomicBool>,
    post_wal_apply_poisoned: Arc<AtomicBool>,
}

/// A root-internal completion barrier, without retaining rows or directories.
#[doc(hidden)]
#[derive(Debug)]
pub struct MaterializedReadHealth {
    integrity_poisoned: Arc<AtomicBool>,
    post_wal_apply_poisoned: Arc<AtomicBool>,
}

impl MaterializedReadHealth {
    pub fn validate_usable(&self) -> Result<()> {
        ensure_graph_read_flags_usable(&self.integrity_poisoned, &self.post_wal_apply_poisoned)
    }

    /// Conservative shared-atomic/header allowance on supported 32/64-bit ABIs.
    pub const fn retained_capacity_bytes() -> usize {
        2 * (3 * std::mem::size_of::<usize>() + std::mem::size_of::<AtomicBool>())
    }
}

impl MaterializedNodeReadSource {
    pub fn iter_after(&self, after: Option<NodeId>) -> Result<impl Iterator<Item = &NodeRecord>> {
        ensure_graph_read_flags_usable(&self.integrity_poisoned, &self.post_wal_apply_poisoned)?;
        Ok(self.nodes.iter_after(after).map(|(_, node)| node))
    }

    pub fn row_count(&self) -> usize {
        self.nodes.len()
    }

    pub fn page_count(&self) -> usize {
        self.nodes.segment_count()
    }

    /// Capacity of the shared page-directory allocation only. This excludes
    /// node-page/record payload capacities and is not a full source memory bound.
    pub fn directory_capacity_bytes(&self) -> usize {
        self.nodes.directory_capacity_bytes()
    }
}

/// Root-internal point source. Unsupported storage must use a separate path;
/// it must never be reported as a missing node in the live overlay.
#[doc(hidden)]
pub enum MaterializedNodePointRead<A = Box<dyn GraphReadAllocation>> {
    Unsupported,
    Missing,
    Node(MaterializedNodePointSource<A>),
}

/// Own one immutable node row, without retaining its page or directory.
/// The admission permit outlives the row and all borrowed record access.
#[doc(hidden)]
pub struct MaterializedNodePointSource<A = Box<dyn GraphReadAllocation>> {
    row: crate::cow::CowSegment<NodeRecord>,
    integrity_poisoned: Arc<AtomicBool>,
    post_wal_apply_poisoned: Arc<AtomicBool>,
    _allocation: A,
}

impl<A> MaterializedNodePointSource<A> {
    pub fn node(&self) -> Result<&NodeRecord> {
        ensure_graph_read_flags_usable(&self.integrity_poisoned, &self.post_wal_apply_poisoned)?;
        Ok(&self.row)
    }
}

fn retain_point_allocation(allocation: &mut dyn GraphReadAllocation, bytes: usize) -> Result<()> {
    if allocation.bytes() < bytes {
        return Err(HawDBError::Execution(
            "retained point admission returned an insufficient allocation permit".into(),
        ));
    }
    allocation.retain_state()?;
    if allocation.bytes() < bytes {
        return Err(HawDBError::Execution(
            "retained point admission lost its allocation permit".into(),
        ));
    }
    Ok(())
}

#[cfg(test)]
#[path = "graph_read/relationship_scan_tests.rs"]
mod relationship_scan_tests;

#[cfg(test)]
#[path = "graph_read/relationship_stream_tests.rs"]
mod relationship_stream_tests;

impl GraphStore {
    #[doc(hidden)]
    pub fn materialized_read_health(&self) -> Result<MaterializedReadHealth> {
        self.ensure_usable()?;
        Ok(MaterializedReadHealth {
            integrity_poisoned: Arc::clone(&self.integrity_poisoned),
            post_wal_apply_poisoned: Arc::clone(&self.post_wal_apply_poisoned),
        })
    }

    /// Admit the selected row before pinning it, including nested and spare
    /// capacities. This does not construct Arrow arrays or
    /// establish a whole-process memory bound. Callers must supply their policy.
    #[doc(hidden)]
    pub fn materialized_node_point_read(
        &self,
        id: NodeId,
        admission: &GraphReadAdmission<'_>,
    ) -> Result<MaterializedNodePointRead> {
        self.materialized_node_point_read_with_policy(
            id,
            |bytes| admission.reserve(bytes),
            |allocation, bytes| retain_point_allocation(allocation.as_mut(), bytes),
        )
    }

    /// Preserve the concrete permit's auto traits for typed host ownership.
    /// No Send/Sync promise is imposed on standalone allocation policies.
    #[doc(hidden)]
    pub fn materialized_node_point_read_typed<A: GraphReadAllocation>(
        &self,
        id: NodeId,
        allocate: impl FnOnce(usize) -> Result<A>,
    ) -> Result<MaterializedNodePointRead<A>> {
        self.materialized_node_point_read_with_admission(id, allocate)
    }

    /// Preserve host admission errors without converting backpressure or a
    /// working-unit refusal into a generic storage error.
    #[doc(hidden)]
    pub fn materialized_node_point_read_with_admission<A: GraphReadAllocation, E>(
        &self,
        id: NodeId,
        allocate: impl FnOnce(usize) -> std::result::Result<A, E>,
    ) -> std::result::Result<MaterializedNodePointRead<A>, E>
    where
        E: From<HawDBError>,
    {
        self.materialized_node_point_read_with_policy(id, allocate, |allocation, bytes| {
            retain_point_allocation(allocation, bytes)
        })
    }

    fn materialized_node_point_read_with_policy<A, E>(
        &self,
        id: NodeId,
        allocate: impl FnOnce(usize) -> std::result::Result<A, E>,
        retain: impl FnOnce(&mut A, usize) -> Result<()>,
    ) -> std::result::Result<MaterializedNodePointRead<A>, E>
    where
        E: From<HawDBError>,
    {
        self.ensure_usable()?;
        if self.is_out_of_core() || self.canonical_base.is_some() {
            return Ok(MaterializedNodePointRead::Unsupported);
        }
        if self.node_tombstones.contains(&id) {
            return Ok(MaterializedNodePointRead::Missing);
        }
        let Some((row, capacity)) = self.nodes.retained_point_row_preflight(id) else {
            return Ok(MaterializedNodePointRead::Missing);
        };
        let bytes = capacity
            .and_then(|bytes| bytes.checked_add(64))
            .ok_or_else(|| {
                HawDBError::Execution(
                    "retained point capacity overflow or excessive value nesting".into(),
                )
            })?;
        let mut allocation = allocate(bytes)?;
        retain(&mut allocation, bytes)?;
        Ok(MaterializedNodePointRead::Node(
            MaterializedNodePointSource {
                row: row.clone(),
                integrity_poisoned: Arc::clone(&self.integrity_poisoned),
                post_wal_apply_poisoned: Arc::clone(&self.post_wal_apply_poisoned),
                _allocation: allocation,
            },
        ))
    }

    /// Bound every heap allocation pinned by the same materialized directory
    /// before a caller adopts it. This preflight borrows all source records.
    #[doc(hidden)]
    pub fn materialized_node_read_source_capacity_bytes(&self) -> Result<Option<usize>> {
        Ok(self
            .materialized_node_read_source_capacity_preflight()?
            .map(|(bytes, _)| bytes))
    }

    /// Return capacity and the records inspected on a cold snapshot preflight.
    #[doc(hidden)]
    pub fn materialized_node_read_source_capacity_preflight(
        &self,
    ) -> Result<Option<(usize, usize)>> {
        self.ensure_usable()?;
        if self.is_out_of_core() || self.canonical_base.is_some() {
            return Ok(None);
        }
        let (capacity, inspected) = self.nodes.retained_capacity_preflight();
        capacity
            .and_then(|bytes| bytes.checked_add(64))
            .map(|bytes| Some((bytes, inspected)))
            .ok_or_else(|| {
                HawDBError::Execution(
                    "retained source capacity overflow or excessive value nesting".into(),
                )
            })
    }

    /// Capture a heap-only materialized source without cloning the whole store.
    /// Caller admission of source capacities remains a separate requirement.
    #[doc(hidden)]
    pub fn try_materialized_node_read_source(&self) -> Result<Option<MaterializedNodeReadSource>> {
        self.ensure_usable()?;
        if self.is_out_of_core() || self.canonical_base.is_some() {
            return Ok(None);
        }
        Ok(Some(MaterializedNodeReadSource {
            nodes: self.nodes.clone(),
            integrity_poisoned: Arc::clone(&self.integrity_poisoned),
            post_wal_apply_poisoned: Arc::clone(&self.post_wal_apply_poisoned),
        }))
    }

    pub fn canonical_node_from_segments(&self, id: NodeId) -> Result<Option<NodeRecord>> {
        self.durable
            .as_ref()
            .and_then(|durable| durable.canonical_segments.as_ref())
            .map(|reader| reader.get_node(id).map_err(canonical_segment_error))
            .transpose()
            .map(Option::flatten)
    }

    pub fn canonical_relationship_from_segments(&self, id: RelId) -> Result<Option<RelRecord>> {
        self.durable
            .as_ref()
            .and_then(|durable| durable.canonical_segments.as_ref())
            .map(|reader| reader.get_relationship(id).map_err(canonical_segment_error))
            .transpose()
            .map(Option::flatten)
    }

    pub fn node_records_owned(&self) -> GraphNodeIterator {
        hawdb_storage::graph_overlay::node_records(
            self.canonical_base
                .as_ref()
                .map(CanonicalSegmentReader::node_records),
            self.nodes.values().cloned().collect::<Vec<_>>(),
            self.node_tombstones.clone(),
        )
    }

    pub fn relationship_records_owned(&self) -> GraphRelationshipIterator {
        hawdb_storage::graph_overlay::relationship_records(
            self.canonical_base
                .as_ref()
                .map(CanonicalSegmentReader::relationship_records),
            self.relationships.values().cloned().collect::<Vec<_>>(),
            self.relationship_tombstones.clone(),
        )
    }

    pub(super) fn checkpoint_node_records_owned<'a>(
        &'a self,
        work: &crate::background::CheckpointWorkContext,
    ) -> Result<
        crate::graph_overlay::CheckpointOverlayIterator<
            'a,
            NodeRecord,
            impl Iterator<Item = &'a NodeRecord>,
        >,
    > {
        // Preserve the capture cancellation boundaries without copying values.
        for _ in self.nodes.values() {
            let unit = work.start_unit().map_err(HawDBError::from_storage_error)?;
            unit.finish();
        }
        work.checkpoint().map_err(HawDBError::from_storage_error)?;
        Ok(crate::graph_overlay::CheckpointOverlayIterator::new(
            self.canonical_base.as_ref(),
            self.nodes.values(),
            &self.node_tombstones,
            work,
        ))
    }

    pub(super) fn checkpoint_relationship_records_owned<'a>(
        &'a self,
        work: &crate::background::CheckpointWorkContext,
    ) -> Result<
        crate::graph_overlay::CheckpointOverlayIterator<
            'a,
            RelRecord,
            impl Iterator<Item = &'a RelRecord>,
        >,
    > {
        for _ in self.relationships.values() {
            let unit = work.start_unit().map_err(HawDBError::from_storage_error)?;
            unit.finish();
        }
        work.checkpoint().map_err(HawDBError::from_storage_error)?;
        Ok(crate::graph_overlay::CheckpointOverlayIterator::new(
            self.canonical_base.as_ref(),
            self.relationships.values(),
            &self.relationship_tombstones,
            work,
        ))
    }

    #[cfg(test)]
    pub(in crate::store) fn checkpoint_node_owned(
        &self,
        id: NodeId,
        work: &crate::background::CheckpointWorkContext,
    ) -> Result<Option<NodeRecord>> {
        work.checkpoint().map_err(HawDBError::from_storage_error)?;
        if self.node_tombstones.contains(&id) {
            return Ok(None);
        }
        if let Some(node) = self.nodes.get(&id) {
            return Ok(Some(node.clone()));
        }
        Ok(self
            .checkpoint_mounted_node(id, work)?
            .map(|record| record.into_record()))
    }

    #[cfg(test)]
    pub(in crate::store) fn checkpoint_relationship_owned(
        &self,
        id: RelId,
        work: &crate::background::CheckpointWorkContext,
    ) -> Result<Option<RelRecord>> {
        work.checkpoint().map_err(HawDBError::from_storage_error)?;
        if self.relationship_tombstones.contains(&id) {
            return Ok(None);
        }
        if let Some(relationship) = self.relationships.get(&id) {
            return Ok(Some(relationship.clone()));
        }
        Ok(self
            .checkpoint_mounted_relationship(id, work)?
            .map(|record| record.into_record()))
    }

    // Delta estimation calls these only for records absent from the overlay.
    // Keep the allocation owner with the temporary decoded record through the
    // estimator's final borrow; a plain owned-record return would release early.
    pub(in crate::store) fn checkpoint_mounted_node(
        &self,
        id: NodeId,
        work: &crate::background::CheckpointWorkContext,
    ) -> Result<Option<crate::canonical::CheckpointRecord<NodeRecord>>> {
        work.checkpoint().map_err(HawDBError::from_storage_error)?;
        if self.node_tombstones.contains(&id) {
            return Ok(None);
        }
        self.canonical_base
            .as_ref()
            .map(|reader| {
                reader
                    .checkpoint_node(id, work)
                    .map_err(|error| match error {
                        CanonicalSegmentError::Work(error) => HawDBError::from_storage_error(error),
                        error => canonical_segment_error(error),
                    })
            })
            .transpose()
            .map(Option::flatten)
    }

    pub(in crate::store) fn checkpoint_mounted_relationship(
        &self,
        id: RelId,
        work: &crate::background::CheckpointWorkContext,
    ) -> Result<Option<crate::canonical::CheckpointRecord<RelRecord>>> {
        work.checkpoint().map_err(HawDBError::from_storage_error)?;
        if self.relationship_tombstones.contains(&id) {
            return Ok(None);
        }
        self.canonical_base
            .as_ref()
            .map(|reader| {
                reader
                    .checkpoint_relationship(id, work)
                    .map_err(|error| match error {
                        CanonicalSegmentError::Work(error) => HawDBError::from_storage_error(error),
                        error => canonical_segment_error(error),
                    })
            })
            .transpose()
            .map(Option::flatten)
    }

    pub fn node_owned(&self, id: NodeId) -> Result<Option<NodeRecord>> {
        if self.node_tombstones.contains(&id) {
            return Ok(None);
        }
        if let Some(node) = self.nodes.get(&id) {
            return Ok(Some(node.clone()));
        }
        self.canonical_base
            .as_ref()
            .map(|reader| reader.get_node(id).map_err(canonical_segment_error))
            .transpose()
            .map(Option::flatten)
    }

    /// Admit full point ownership before copying Values. The same immutable
    /// reader pins canonical preflight and decode; caller refusal never poisons it.
    pub fn node_with_allocation(
        &self,
        id: NodeId,
        label_ids: Option<&[LabelId]>,
        admit: &mut ControlledGraphReadAllocator<'_>,
    ) -> Result<AdmittedNodeRead> {
        let work = admit.work();
        work.charge_work(1)?;
        if self.node_tombstones.contains(&id) {
            return Ok(AdmittedNodeRead::Missing);
        }
        if let Some(node) = self.nodes.get(&id) {
            if label_ids
                .is_some_and(|labels| !labels.iter().any(|label| node.labels.contains(label)))
            {
                return Ok(AdmittedNodeRead::Missing);
            }
            let Some(allocation) = admit.reserve(node_allocation_bytes(node))? else {
                return Ok(AdmittedNodeRead::Stopped);
            };
            return AdmittedNodeRecord::clone_admitted(node, allocation)
                .map(AdmittedNodeRead::Node);
        }
        let Some(reader) = self.canonical_base.as_ref() else {
            return Ok(AdmittedNodeRead::Missing);
        };
        let Some(bytes) = reader
            .node_allocation_bytes(id, label_ids)
            .map_err(canonical_segment_error)?
        else {
            return Ok(AdmittedNodeRead::Missing);
        };
        let Some(allocation) = admit.reserve(bytes)? else {
            return Ok(AdmittedNodeRead::Stopped);
        };
        if allocation.bytes() < bytes {
            return Err(HawDBError::Execution(
                "full node admission returned an insufficient allocation permit".into(),
            ));
        }
        reader
            .get_node(id)
            .map_err(canonical_segment_error)
            .map(|node| {
                node.map_or(AdmittedNodeRead::Missing, |node| {
                    AdmittedNodeRead::Node(AdmittedNodeRecord::new(node, allocation))
                })
            })
    }

    /// Admit the selected node allocation before cloning or decoding values.
    /// The immutable reader pins the same record across preflight and decoding.
    pub fn projected_node_owned_admitted(
        &self,
        id: NodeId,
        required_properties: &BTreeSet<String>,
        admit: &mut dyn FnMut(usize) -> Result<()>,
    ) -> Result<Option<ProjectedNodeRecord>> {
        if self.node_tombstones.contains(&id) {
            return Ok(None);
        }
        if let Some(node) = self.nodes.get(&id) {
            admit(projected_node_allocation_bytes(node, required_properties))?;
            return Ok(Some(project_node_record_ref(node, required_properties)));
        }
        let Some(reader) = self.canonical_base.as_ref() else {
            return Ok(None);
        };
        let mut admission_error = None;
        let result = reader.with_node_input(id, |input| {
            input
                .decode_selected(required_properties, |bytes| {
                    admit(bytes).map_err(|error| {
                        admission_error = Some(error);
                        CanonicalSegmentError::Source("graph read admission refused".into())
                    })
                })
                .map(|(node, ())| node)
        });
        if let Some(error) = admission_error {
            return Err(error);
        }
        result.map_err(canonical_segment_error)
    }

    pub fn relationship_owned(&self, id: RelId) -> Result<Option<RelRecord>> {
        if self.relationship_tombstones.contains(&id) {
            return Ok(None);
        }
        if let Some(relationship) = self.relationships.get(&id) {
            return Ok(Some(relationship.clone()));
        }
        self.canonical_base
            .as_ref()
            .map(|reader| reader.get_relationship(id).map_err(canonical_segment_error))
            .transpose()
            .map(Option::flatten)
    }

    /// Full point ownership is admitted before resident clone or canonical decode.
    pub fn relationship_with_allocation(
        &self,
        id: RelId,
        admit: &mut ControlledGraphReadAllocator<'_>,
    ) -> Result<AdmittedRelationshipRead> {
        let work = admit.work();
        work.charge_work(1)?;
        if self.relationship_tombstones.contains(&id) {
            return Ok(AdmittedRelationshipRead::Missing);
        }
        if let Some(relationship) = self.relationships.get(&id) {
            let Some(allocation) =
                admit.reserve(hawdb_core::ids::relationship_allocation_bytes(relationship))?
            else {
                return Ok(AdmittedRelationshipRead::Stopped);
            };
            return AdmittedRelationshipRecord::clone_admitted(relationship, allocation)
                .map(AdmittedRelationshipRead::Relationship);
        }
        let Some(reader) = self.canonical_base.as_ref() else {
            return Ok(AdmittedRelationshipRead::Missing);
        };
        let mut result = Ok(AdmittedRelationshipRead::Missing);
        reader
            .visit_relationship_payload(id, |layout, payload| {
                let allocation = match admit.reserve(layout.allocation_bytes) {
                    Ok(Some(allocation)) => allocation,
                    Ok(None) => {
                        result = Ok(AdmittedRelationshipRead::Stopped);
                        return Ok(());
                    }
                    Err(error) => {
                        result = Err(error);
                        return Ok(());
                    }
                };
                if allocation.bytes() < layout.allocation_bytes {
                    result = Err(HawDBError::Execution(
                        "relationship admission returned an insufficient allocation permit".into(),
                    ));
                    return Ok(());
                }
                let relationship = reader.decode_scanned_relationship(id, payload)?;
                result = Ok(AdmittedRelationshipRead::Relationship(
                    AdmittedRelationshipRecord::new(relationship, allocation),
                ));
                Ok(())
            })
            .map_err(canonical_segment_error)?;
        result
    }

    pub fn visit_nodes_owned(
        &self,
        label_id: Option<LabelId>,
        mut consumer: impl FnMut(NodeRecord) -> GraphScanControl,
    ) -> Result<GraphScanControl> {
        self.visit_selected_nodes_owned(label_id, None, |node| {
            consumer(NodeRecord {
                id: node.id,
                labels: node.labels,
                properties: node.properties,
            })
        })
    }

    pub fn visit_projected_nodes_owned(
        &self,
        label_id: Option<LabelId>,
        required_properties: &BTreeSet<String>,
        mut consumer: impl FnMut(ProjectedNodeRecord) -> GraphScanControl,
    ) -> Result<GraphScanControl> {
        self.visit_projected_nodes_admitted(
            label_id,
            required_properties,
            &mut crate::read_view::GraphReadAllocator::new(
                &mut |bytes| Ok(Box::new(UnaccountedReadAllocation(bytes))),
                None,
            ),
            &mut |node| Ok(consumer(node)),
        )
    }

    pub fn visit_projected_nodes_by_access_owned(
        &self,
        label_id: LabelId,
        access: &hawdb_core::access::NodeProjectionAccess,
        required_properties: &BTreeSet<String>,
        mut consumer: impl FnMut(ProjectedNodeRecord) -> GraphScanControl,
    ) -> Result<GraphScanControl> {
        self.visit_projected_nodes_by_access_admitted(
            label_id,
            access,
            required_properties,
            &mut crate::read_view::GraphReadAllocator::new(
                &mut |bytes| Ok(Box::new(UnaccountedReadAllocation(bytes))),
                None,
            ),
            &mut |node| Ok(consumer(node)),
        )
    }

    fn visit_projected_nodes_by_access_with_admission(
        &self,
        label_id: LabelId,
        access: &hawdb_core::access::NodeProjectionAccess,
        required_properties: Option<&BTreeSet<String>>,
        admission: &ProjectionAdmission<'_>,
        consumer: impl FnMut(AdmittedProjection) -> GraphScanControl,
    ) -> Result<GraphScanControl> {
        match access {
            hawdb_core::access::NodeProjectionAccess::LabelScan => {
                self.visit_selected_nodes_with_admission(Some(label_id), required_properties, admission, consumer)
            }
            hawdb_core::access::NodeProjectionAccess::PropertyValues { property, values } => self
                .visit_projected_nodes_by_property_with_admission(
                    label_id,
                    property,
                    values,
                    required_properties,
                    admission,
                    consumer,
                ),
            hawdb_core::access::NodeProjectionAccess::PropertyUnion { .. } => Err(HawDBError::Execution(
                "property-union projection access requires executor-owned deduplication admission"
                    .to_string(),
            )),
            hawdb_core::access::NodeProjectionAccess::CompositeEquality { predicates } => self
                .visit_projected_nodes_by_composite_property_with_admission(
                    label_id,
                    predicates,
                    required_properties,
                    admission,
                    consumer,
                ),
            hawdb_core::access::NodeProjectionAccess::CompositeRange { seek } => self
                .visit_projected_nodes_by_composite_range_with_admission(
                    label_id,
                    seek,
                    required_properties,
                    admission,
                    consumer,
                ),
            hawdb_core::access::NodeProjectionAccess::CompositeOrdered { seek } => self
                .visit_projected_nodes_by_composite_order_with_admission(
                    label_id,
                    seek,
                    required_properties,
                    admission,
                    consumer,
                ),
            hawdb_core::access::NodeProjectionAccess::PropertyRange {
                property,
                lower,
                upper,
            } => self.visit_projected_nodes_by_property_range_with_admission(
                label_id,
                property,
                PropertyRangeBounds { lower: lower.as_ref(), upper: upper.as_ref() },
                required_properties,
                admission,
                consumer,
            ),
            hawdb_core::access::NodeProjectionAccess::FullText { property, query } => self
                .visit_projected_nodes_by_full_text_property_with_admission(
                    label_id,
                    property,
                    query,
                    required_properties,
                    admission,
                    consumer,
                ),
        }
    }

    fn visit_selected_nodes_owned(
        &self,
        label_id: Option<LabelId>,
        required_properties: Option<&BTreeSet<String>>,
        mut consumer: impl FnMut(ProjectedNodeRecord) -> GraphScanControl,
    ) -> Result<GraphScanControl> {
        let project_delta = |node: &NodeRecord| match required_properties {
            Some(required) => project_node_record_ref(node, required),
            None => ProjectedNodeRecord {
                id: node.id,
                labels: node.labels.clone(),
                properties: node.properties.clone(),
            },
        };
        let Some(reader) = &self.canonical_base else {
            for node in self.nodes.values() {
                if self.node_matches_label(node, label_id)
                    && consumer(project_delta(node)) == GraphScanControl::Stop
                {
                    return Ok(GraphScanControl::Stop);
                }
            }
            return Ok(GraphScanControl::Continue);
        };

        let mut delta = self.nodes.iter().peekable();
        let mut graph_control = GraphScanControl::Continue;
        let mut consume_base = |base: ProjectedNodeRecord| {
            while delta.peek().is_some_and(|(id, _)| **id < base.id) {
                let (id, node) = delta.next().expect("peeked delta node exists");
                if !self.node_tombstones.contains(id)
                    && self.node_matches_label(node, label_id)
                    && consumer(project_delta(node)) == GraphScanControl::Stop
                {
                    graph_control = GraphScanControl::Stop;
                    return Ok(CanonicalScanControl::Stop);
                }
            }
            if delta.peek().is_some_and(|(id, _)| **id == base.id) {
                let (id, node) = delta.next().expect("matching delta node exists");
                if !self.node_tombstones.contains(id)
                    && self.node_matches_label(node, label_id)
                    && consumer(project_delta(node)) == GraphScanControl::Stop
                {
                    graph_control = GraphScanControl::Stop;
                    return Ok(CanonicalScanControl::Stop);
                }
                return Ok(CanonicalScanControl::Continue);
            }
            if !self.node_tombstones.contains(&base.id)
                && label_id.is_none_or(|label_id| base.labels.contains(&label_id))
                && consumer(base) == GraphScanControl::Stop
            {
                graph_control = GraphScanControl::Stop;
                return Ok(CanonicalScanControl::Stop);
            }
            Ok(CanonicalScanControl::Continue)
        };
        let (_, canonical_control) = match required_properties {
            Some(required_properties) => {
                reader.scan_projected_nodes_control(required_properties, &mut consume_base)
            }
            None => reader.scan_nodes_control(|node| {
                consume_base(ProjectedNodeRecord {
                    id: node.id,
                    labels: node.labels,
                    properties: node.properties,
                })
            }),
        }
        .map_err(canonical_segment_error)?;
        if canonical_control == CanonicalScanControl::Stop {
            return Ok(graph_control);
        }
        for (id, node) in delta {
            if !self.node_tombstones.contains(id)
                && self.node_matches_label(node, label_id)
                && consumer(project_delta(node)) == GraphScanControl::Stop
            {
                return Ok(GraphScanControl::Stop);
            }
        }
        Ok(GraphScanControl::Continue)
    }

    pub fn try_visit_nodes_owned(
        &self,
        label_id: Option<LabelId>,
        mut consumer: impl FnMut(NodeRecord) -> Result<GraphScanControl>,
    ) -> Result<GraphScanControl> {
        let mut consumer_error = None;
        let control = self.visit_nodes_owned(label_id, |node| match consumer(node) {
            Ok(control) => control,
            Err(error) => {
                consumer_error = Some(error);
                GraphScanControl::Stop
            }
        })?;
        match consumer_error {
            Some(error) => Err(error),
            None => Ok(control),
        }
    }

    /// Stream live relationships in ID order, admitting each full record before
    /// cloning or decoding. Callback errors and normal Stop keep the reader healthy.
    pub fn visit_relationships_with_allocation(
        &self,
        rel_type: Option<RelTypeId>,
        admit: &mut ControlledGraphReadAllocator<'_>,
        consumer: &mut dyn FnMut(AdmittedRelationshipRecord) -> Result<GraphScanControl>,
    ) -> Result<GraphScanControl> {
        let work = admit.work();
        enum BaseVisit {
            Continue,
            Stop,
            Decode(Box<dyn crate::read_view::GraphReadAllocation>),
        }
        let mut delta = self.relationships.iter().peekable();
        if let Some(reader) = &self.canonical_base {
            let mut stopped = false;
            let mut callback_error = None;
            reader
                .scan_relationship_inputs_control(|input| {
                    let id = input.id;
                    let result = (|| {
                        work.charge_work(1)?;
                        let mut visit = |id| match self.relationship_with_allocation(id, admit)? {
                            AdmittedRelationshipRead::Missing => Ok(GraphScanControl::Continue),
                            AdmittedRelationshipRead::Stopped => Ok(GraphScanControl::Stop),
                            AdmittedRelationshipRead::Relationship(record) => consumer(record),
                        };
                        while delta.peek().is_some_and(|(next, _)| **next < id) {
                            let (next, record) = delta.next().expect("peeked delta exists");
                            work.charge_work(1)?;
                            if !self.relationship_tombstones.contains(next)
                                && self.relationship_matches_type(record, rel_type)
                                && visit(*next)? == GraphScanControl::Stop
                            {
                                return Ok(BaseVisit::Stop);
                            }
                        }
                        if delta.peek().is_some_and(|(next, _)| **next == id) {
                            let (next, record) = delta.next().expect("matching delta exists");
                            work.charge_work(1)?;
                            if !self.relationship_tombstones.contains(next)
                                && self.relationship_matches_type(record, rel_type)
                                && visit(*next)? == GraphScanControl::Stop
                            {
                                return Ok(BaseVisit::Stop);
                            }
                        } else if !self.relationship_tombstones.contains(&id)
                            && rel_type.is_none_or(|selected| selected == input.layout.rel_type)
                        {
                            let bytes = input.layout.allocation_bytes;
                            let Some(allocation) = admit.reserve(bytes)? else {
                                return Ok(BaseVisit::Stop);
                            };
                            if allocation.bytes() < bytes {
                                return Err(HawDBError::Execution(
                                    "relationship admission returned an insufficient allocation permit".into(),
                                ));
                            }
                            return Ok(BaseVisit::Decode(allocation));
                        }
                        Ok(BaseVisit::Continue)
                    })();
                    let result = match result {
                        Ok(BaseVisit::Continue) => Ok(GraphScanControl::Continue),
                        Ok(BaseVisit::Stop) => Ok(GraphScanControl::Stop),
                        Ok(BaseVisit::Decode(allocation)) => {
                            // Decode failures reach the canonical reader's poison
                            // boundary; caller errors below remain recoverable.
                            let relationship = input.decode()?;
                            consumer(AdmittedRelationshipRecord::new(relationship, allocation))
                        }
                        Err(error) => Err(error),
                    };
                    match result {
                        Ok(GraphScanControl::Continue) => Ok(CanonicalScanControl::Continue),
                        Ok(GraphScanControl::Stop) => {
                            stopped = true;
                            Ok(CanonicalScanControl::Stop)
                        }
                        Err(error) => {
                            callback_error = Some(error);
                            Ok(CanonicalScanControl::Stop)
                        }
                    }
                })
                .map_err(canonical_segment_error)?;
            if let Some(error) = callback_error {
                return Err(error);
            }
            if stopped {
                return Ok(GraphScanControl::Stop);
            }
        }
        let mut visit = |id| match self.relationship_with_allocation(id, admit)? {
            AdmittedRelationshipRead::Missing => Ok(GraphScanControl::Continue),
            AdmittedRelationshipRead::Stopped => Ok(GraphScanControl::Stop),
            AdmittedRelationshipRead::Relationship(record) => consumer(record),
        };
        for (id, record) in delta {
            work.charge_work(1)?;
            if !self.relationship_tombstones.contains(id)
                && self.relationship_matches_type(record, rel_type)
                && visit(*id)? == GraphScanControl::Stop
            {
                return Ok(GraphScanControl::Stop);
            }
        }
        Ok(GraphScanControl::Continue)
    }

    pub fn visit_relationships_owned(
        &self,
        rel_type: Option<RelTypeId>,
        mut consumer: impl FnMut(RelRecord) -> GraphScanControl,
    ) -> Result<GraphScanControl> {
        let Some(reader) = &self.canonical_base else {
            for relationship in self.relationships.values() {
                if self.relationship_matches_type(relationship, rel_type)
                    && consumer(relationship.clone()) == GraphScanControl::Stop
                {
                    return Ok(GraphScanControl::Stop);
                }
            }
            return Ok(GraphScanControl::Continue);
        };

        let mut delta = self.relationships.iter().peekable();
        let mut graph_control = GraphScanControl::Continue;
        let (_, canonical_control) = reader
            .scan_relationships_control(|base| {
                while delta.peek().is_some_and(|(id, _)| **id < base.id) {
                    let (id, relationship) =
                        delta.next().expect("peeked delta relationship exists");
                    if !self.relationship_tombstones.contains(id)
                        && self.relationship_matches_type(relationship, rel_type)
                        && consumer(relationship.clone()) == GraphScanControl::Stop
                    {
                        graph_control = GraphScanControl::Stop;
                        return Ok(CanonicalScanControl::Stop);
                    }
                }
                if delta.peek().is_some_and(|(id, _)| **id == base.id) {
                    let (id, relationship) =
                        delta.next().expect("matching delta relationship exists");
                    if !self.relationship_tombstones.contains(id)
                        && self.relationship_matches_type(relationship, rel_type)
                        && consumer(relationship.clone()) == GraphScanControl::Stop
                    {
                        graph_control = GraphScanControl::Stop;
                        return Ok(CanonicalScanControl::Stop);
                    }
                    return Ok(CanonicalScanControl::Continue);
                }
                if !self.relationship_tombstones.contains(&base.id)
                    && self.relationship_matches_type(&base, rel_type)
                    && consumer(base) == GraphScanControl::Stop
                {
                    graph_control = GraphScanControl::Stop;
                    return Ok(CanonicalScanControl::Stop);
                }
                Ok(CanonicalScanControl::Continue)
            })
            .map_err(canonical_segment_error)?;
        if canonical_control == CanonicalScanControl::Stop {
            return Ok(graph_control);
        }
        for (id, relationship) in delta {
            if !self.relationship_tombstones.contains(id)
                && self.relationship_matches_type(relationship, rel_type)
                && consumer(relationship.clone()) == GraphScanControl::Stop
            {
                return Ok(GraphScanControl::Stop);
            }
        }
        Ok(GraphScanControl::Continue)
    }

    pub fn try_visit_relationships_owned(
        &self,
        rel_type: Option<RelTypeId>,
        mut consumer: impl FnMut(RelRecord) -> Result<GraphScanControl>,
    ) -> Result<GraphScanControl> {
        let mut consumer_error = None;
        let control = self.visit_relationships_owned(rel_type, |relationship| {
            match consumer(relationship) {
                Ok(control) => control,
                Err(error) => {
                    consumer_error = Some(error);
                    GraphScanControl::Stop
                }
            }
        })?;
        match consumer_error {
            Some(error) => Err(error),
            None => Ok(control),
        }
    }

    pub fn statistics(&self, catalog: &Catalog) -> GraphStatistics {
        if !self.canonical_base_out_of_core {
            let mut statistics = compute_statistics_with_basic(
                &self.nodes,
                &self.relationships,
                Some(catalog),
                self.basic_statistics(),
            );
            statistics.index_samples = compute_index_statistics_samples(
                catalog,
                &self.property_index,
                &self.composite_property_index,
            );
            return statistics;
        }
        let mut statistics = self.checkpoint_statistics.materialize();
        retain_supported_property_statistics(&mut statistics, Some(catalog));
        retain_valid_index_statistics_samples(&mut statistics, catalog);
        let basic = self.basic_statistics();
        statistics.node_count = basic.node_count;
        statistics.relationship_count = basic.relationship_count;
        statistics.label_counts = basic.label_counts;
        statistics.rel_type_counts = basic.rel_type_counts;
        statistics
    }

    pub(super) fn checkpoint_statistics_with_work_context(
        &self,
        catalog: &Catalog,
        work: &crate::background::CheckpointWorkContext,
    ) -> Result<GraphStatistics> {
        use hawdb_storage::statistics::checkpoint::{
            clone_retained_with_index_samples_and_work_context, compute_with_work_context,
            index_samples_with_work_context,
        };
        let basic = self
            .basic_statistics
            .materialize_with_work_context(self.commit_epoch, work)
            .map_err(HawDBError::from_storage_error)?;
        if self.canonical_base_out_of_core {
            return clone_retained_with_index_samples_and_work_context(
                &self.checkpoint_statistics,
                self.checkpoint_statistics.index_samples.iter(),
                catalog,
                basic,
                work,
            )
            .map_err(HawDBError::from_storage_error);
        }
        let mut statistics =
            compute_with_work_context(&self.nodes, &self.relationships, Some(catalog), basic, work)
                .map_err(HawDBError::from_storage_error)?;
        statistics.index_samples = index_samples_with_work_context(
            catalog,
            &self.property_index,
            &self.composite_property_index,
            work,
        )
        .map_err(HawDBError::from_storage_error)?;
        Ok(statistics)
    }

    pub fn basic_statistics(&self) -> BasicGraphStatistics {
        self.basic_statistics.materialize(self.commit_epoch)
    }

    pub fn basic_statistics_consistency_report(&self) -> BasicStatisticsConsistencyReport {
        BasicStatisticsConsistencyReport::new(
            self.basic_statistics(),
            compute_basic_statistics(&self.nodes, &self.relationships, self.commit_epoch),
        )
    }

    pub fn adjacency_consistency_report(&self) -> AdjacencyConsistencyReport {
        AdjacencyConsistencyReport::new(
            self.commit_epoch,
            self.relationships.len(),
            maintained_adjacency_groups(&self.outgoing, &self.incoming),
            recompute_adjacency_groups(&self.relationships),
            &self.relationships,
        )
    }

    pub fn degree_statistics_consistency_report(&self) -> DegreeStatisticsConsistencyReport {
        DegreeStatisticsConsistencyReport::new(
            self.commit_epoch,
            compute_degree_statistics_from_adjacency(&self.nodes, &self.outgoing, &self.incoming),
            compute_degree_statistics_from_relationships(&self.nodes, &self.relationships),
        )
    }

    pub fn distinct_value_statistics_consistency_report(
        &self,
        catalog: &Catalog,
    ) -> DistinctValueStatisticsConsistencyReport {
        let recomputed = compute_statistics_for_catalog(
            &self.nodes,
            &self.relationships,
            catalog,
            self.basic_statistics(),
        );
        // The index-derived side can only speak for declared properties, so
        // the recomputed side is narrowed to the same keys. Comparing against
        // every property would flag the undeclared ones forever.
        let recomputed_property_distinct_counts = recomputed
            .property_distinct_counts
            .into_iter()
            .filter(|((label_id, property), _)| {
                catalog.has_scalar_property_index(*label_id, property)
            })
            .collect();
        DistinctValueStatisticsConsistencyReport::new(
            self.commit_epoch,
            compute_node_property_distinct_counts_from_index(&self.property_index, catalog),
            recomputed_property_distinct_counts,
            compute_relationship_property_distinct_counts_from_index(
                &self.relationship_property_index,
                catalog,
            ),
            recomputed.rel_property_distinct_counts,
        )
    }

    pub fn scan_nodes<'a>(
        &'a self,
        label_id: Option<LabelId>,
    ) -> impl Iterator<Item = &'a NodeRecord> + 'a {
        self.nodes
            .values()
            .filter(move |node| label_id.map(|id| node.labels.contains(&id)).unwrap_or(true))
    }

    /// A resumable borrowed source for a demanded retained numeric batch.
    /// Refuses sources requiring canonical decoding instead of exposing just
    /// their in-memory overlay. The caller bounds each demanded iterator step.
    pub fn try_scan_materialized_nodes_after(
        &self,
        label_id: Option<LabelId>,
        after: Option<NodeId>,
    ) -> Result<Option<impl Iterator<Item = &NodeRecord>>> {
        self.ensure_usable()?;
        if self.is_out_of_core() || self.canonical_base.is_some() {
            return Ok(None);
        }
        Ok(Some(
            self.nodes
                .iter_after(after)
                .map(|(_, node)| node)
                .filter(move |node| label_id.is_none_or(|id| node.labels.contains(&id))),
        ))
    }

    pub fn scan_nodes_with_filter_pruning<'a>(
        &'a self,
        catalog: &Catalog,
        label_id: Option<LabelId>,
        filter: Option<&PropertyFilter>,
    ) -> ScanPrunedNodeScan<'a> {
        self.scan_nodes_with_filter_pruning_admitted(catalog, label_id, filter, &mut |bytes| {
            Ok(Box::new(admission::UnaccountedReadAllocation(bytes)))
        })
        .expect("standalone graph-read allocation policy cannot refuse")
        .0
    }

    /// Admit every concurrently live candidate set and the returned reference
    /// vector before allocation. The returned permit must outlive that vector.
    pub fn scan_nodes_with_filter_pruning_admitted<'a>(
        &'a self,
        catalog: &Catalog,
        label_id: Option<LabelId>,
        filter: Option<&PropertyFilter>,
        admit: &mut dyn FnMut(usize) -> Result<Box<dyn crate::read_view::GraphReadAllocation>>,
    ) -> Result<(
        ScanPrunedNodeScan<'a>,
        Box<dyn crate::read_view::GraphReadAllocation>,
    )> {
        let admission = crate::read_view::GraphReadAdmission::new(admit);
        let candidate = filter
            .map(|filter| self.prune_node_candidates(catalog, label_id, filter, &admission))
            .transpose()?
            .flatten();
        let candidate_count_before_pruning = self.node_count_for_label(label_id);
        let mut nodes = crate::read_view::AdmittedVec::new(&admission)?;
        let (strategy, pruned, exact_empty, candidate_count_before_filter) =
            if let Some(candidate) = candidate {
                for id in candidate.node_ids.iter() {
                    if let Some(node) = self.nodes.get(id)
                        && self.node_matches_label(node, label_id)
                        && filter.is_none_or(|filter| {
                            property_filter_matches(filter, node.id.0, &node.properties)
                        })
                    {
                        nodes.try_push(node)?;
                    }
                }
                (
                    candidate.strategy,
                    true,
                    candidate.exact_empty,
                    candidate.node_ids.len(),
                )
            } else {
                for node in self.scan_nodes(label_id) {
                    if filter.is_none_or(|filter| {
                        property_filter_matches(filter, node.id.0, &node.properties)
                    }) {
                        nodes.try_push(node)?;
                    }
                }
                (
                    ScanPruningStrategy::FullLabelScan,
                    false,
                    false,
                    candidate_count_before_pruning,
                )
            };
        let (nodes, allocation) = nodes.into_parts();
        let output_count = nodes.len();
        Ok((
            ScanPrunedNodeScan {
                nodes,
                report: ScanPruningReport {
                    target_kind: ScanPruningTargetKind::Node,
                    label_id,
                    rel_type_id: None,
                    strategy,
                    pruned,
                    exact_empty,
                    candidate_count_before_pruning,
                    pruned_candidate_count: if pruned {
                        candidate_count_before_pruning.saturating_sub(candidate_count_before_filter)
                    } else {
                        0
                    },
                    candidate_count_before_filter,
                    output_count,
                    filtered_out_count: candidate_count_before_filter.saturating_sub(output_count),
                },
            },
            allocation,
        ))
    }

    pub fn node_count_for_label(&self, label_id: Option<LabelId>) -> usize {
        let count = label_id.map_or(self.basic_statistics.node_count, |label_id| {
            self.basic_statistics
                .label_counts
                .get(&label_id)
                .copied()
                .unwrap_or_default()
        });
        usize::try_from(count).unwrap_or(usize::MAX)
    }

    fn node_matches_label(&self, node: &NodeRecord, label_id: Option<LabelId>) -> bool {
        label_id.is_none_or(|label_id| node.labels.contains(&label_id))
    }

    fn prune_node_candidates(
        &self,
        catalog: &Catalog,
        label_id: Option<LabelId>,
        filter: &PropertyFilter,
        admission: &crate::read_view::GraphReadAdmission<'_>,
    ) -> Result<Option<ScanPruningCandidate>> {
        let (strategy, node_ids) = match filter {
            PropertyFilter::And(filters) => {
                return self.prune_and_node_candidates(catalog, label_id, filters, admission);
            }
            PropertyFilter::Or(filters) => {
                return self.prune_or_node_candidates(catalog, label_id, filters, admission);
            }
            PropertyFilter::Not(_)
            | PropertyFilter::IdNotEq { .. }
            | PropertyFilter::ListContains { .. }
            | PropertyFilter::ListContainsLower { .. }
            | PropertyFilter::Contains { .. }
            | PropertyFilter::StartsWith { .. }
            | PropertyFilter::EndsWith { .. }
            | PropertyFilter::RegexMatch { .. } => return Ok(None),
            PropertyFilter::IdEq { value } => (
                ScanPruningStrategy::IdEq,
                self.node_ids_for_id_values(label_id, std::slice::from_ref(value), admission)?,
            ),
            PropertyFilter::IdIn { values } => (
                if values.is_empty() {
                    ScanPruningStrategy::Empty
                } else {
                    ScanPruningStrategy::IdIn
                },
                self.node_ids_for_id_values(label_id, values, admission)?,
            ),
            PropertyFilter::IdRange { lower, upper } => {
                if lower.is_none() && upper.is_none() {
                    return Ok(None);
                }
                (
                    ScanPruningStrategy::IdRange,
                    self.collect_node_ids(
                        self.nodes
                            .values()
                            .filter(|node| self.node_matches_label(node, label_id))
                            .filter(|node| {
                                range_bounds_match(
                                    &Value::Int(node.id.0 as i64),
                                    lower.as_ref(),
                                    upper.as_ref(),
                                )
                            })
                            .map(|node| node.id),
                        admission,
                    )?,
                )
            }
            PropertyFilter::Eq { property, value } | PropertyFilter::NotEq { property, value } => {
                if !self.indexes_property(catalog, label_id, property) {
                    return Ok(None);
                }
                let negated = matches!(filter, PropertyFilter::NotEq { .. });
                let strategy = if negated {
                    ScanPruningStrategy::PropertyNotEq {
                        property: property.clone(),
                    }
                } else {
                    ScanPruningStrategy::PropertyEq {
                        property: property.clone(),
                    }
                };
                (
                    strategy,
                    self.node_ids_for_property_values(
                        label_id,
                        property,
                        std::iter::once(value),
                        negated,
                        admission,
                    )?,
                )
            }
            PropertyFilter::IsNull { property } | PropertyFilter::IsNotNull { property } => {
                if !self.indexes_property(catalog, label_id, property) {
                    return Ok(None);
                }
                if matches!(filter, PropertyFilter::IsNull { .. }) {
                    (
                        ScanPruningStrategy::PropertyMissingOrNull {
                            property: property.clone(),
                        },
                        self.node_ids_for_property_missing_or_null(label_id, property, admission)?,
                    )
                } else {
                    (
                        ScanPruningStrategy::PropertyExists {
                            property: property.clone(),
                        },
                        self.node_ids_for_indexed_property(
                            label_id,
                            property,
                            |value| value != &Value::Null,
                            admission,
                        )?,
                    )
                }
            }
            PropertyFilter::In { property, values } => {
                if !self.indexes_property(catalog, label_id, property) {
                    return Ok(None);
                }
                (
                    if values.is_empty() {
                        ScanPruningStrategy::Empty
                    } else {
                        ScanPruningStrategy::PropertyIn {
                            property: property.clone(),
                        }
                    },
                    self.node_ids_for_property_values(
                        label_id,
                        property,
                        values.iter(),
                        false,
                        admission,
                    )?,
                )
            }
            PropertyFilter::Range {
                property,
                lower,
                upper,
            } => {
                if (lower.is_none() && upper.is_none())
                    || !self.indexes_property(catalog, label_id, property)
                {
                    return Ok(None);
                }
                (
                    ScanPruningStrategy::PropertyRange {
                        property: property.clone(),
                    },
                    self.node_ids_for_indexed_property(
                        label_id,
                        property,
                        |value| range_bounds_match(value, lower.as_ref(), upper.as_ref()),
                        admission,
                    )?,
                )
            }
            PropertyFilter::DefaultIfNullOrEq {
                property,
                empty,
                default,
                value,
                negated,
            } => {
                if !self.indexes_property(catalog, label_id, property) {
                    return Ok(None);
                }
                let equal = self.node_ids_for_default_if_null_eq(
                    label_id,
                    property,
                    (empty, default, value),
                    admission,
                )?;
                if *negated {
                    let ids = self.collect_node_ids(
                        self.nodes
                            .values()
                            .filter(|node| self.node_matches_label(node, label_id))
                            .filter(|node| !equal.contains(&node.id))
                            .map(|node| node.id),
                        admission,
                    )?;
                    (
                        ScanPruningStrategy::PropertyDefaultIfNullNotEq {
                            property: property.clone(),
                        },
                        ids,
                    )
                } else {
                    (
                        ScanPruningStrategy::PropertyDefaultIfNullEq {
                            property: property.clone(),
                        },
                        equal,
                    )
                }
            }
        };
        Ok(Some(ScanPruningCandidate::exact(strategy, node_ids)))
    }

    fn prune_and_node_candidates(
        &self,
        catalog: &Catalog,
        label_id: Option<LabelId>,
        filters: &[PropertyFilter],
        admission: &crate::read_view::GraphReadAdmission<'_>,
    ) -> Result<Option<ScanPruningCandidate>> {
        let mut best: Option<ScanPruningCandidate> = None;
        for filter in filters {
            let Some(candidate) =
                self.prune_node_candidates(catalog, label_id, filter, admission)?
            else {
                continue;
            };
            if candidate.exact_empty {
                return Ok(Some(candidate));
            }
            if best
                .as_ref()
                .is_none_or(|best| candidate.node_ids.len() < best.node_ids.len())
            {
                best = Some(candidate);
            }
        }
        Ok(best)
    }

    fn prune_or_node_candidates(
        &self,
        catalog: &Catalog,
        label_id: Option<LabelId>,
        filters: &[PropertyFilter],
        admission: &crate::read_view::GraphReadAdmission<'_>,
    ) -> Result<Option<ScanPruningCandidate>> {
        let mut ids = crate::read_view::AdmittedKeySet::new(admission)?;
        for filter in filters {
            let Some(candidate) =
                self.prune_node_candidates(catalog, label_id, filter, admission)?
            else {
                return Ok(None);
            };
            ids.try_extend(candidate.node_ids.iter().copied())?;
        }
        Ok(Some(ScanPruningCandidate::exact(
            if filters.is_empty() {
                ScanPruningStrategy::Empty
            } else {
                ScanPruningStrategy::OrUnion
            },
            ids,
        )))
    }

    fn collect_node_ids(
        &self,
        ids: impl IntoIterator<Item = NodeId>,
        admission: &crate::read_view::GraphReadAdmission<'_>,
    ) -> Result<crate::read_view::AdmittedKeySet<NodeId>> {
        let mut result = crate::read_view::AdmittedKeySet::new(admission)?;
        result.try_extend(ids)?;
        Ok(result)
    }

    fn node_ids_for_id_values(
        &self,
        label_id: Option<LabelId>,
        values: &[Value],
        admission: &crate::read_view::GraphReadAdmission<'_>,
    ) -> Result<crate::read_view::AdmittedKeySet<NodeId>> {
        self.collect_node_ids(
            values
                .iter()
                .filter_map(|value| match value {
                    Value::Int(value) => u64::try_from(*value).ok().map(NodeId),
                    _ => None,
                })
                .filter(|id| {
                    self.nodes
                        .get(id)
                        .is_some_and(|node| self.node_matches_label(node, label_id))
                }),
            admission,
        )
    }

    fn node_ids_for_indexed_property(
        &self,
        label_id: Option<LabelId>,
        property: &str,
        mut matches: impl FnMut(&Value) -> bool,
        admission: &crate::read_view::GraphReadAdmission<'_>,
    ) -> Result<crate::read_view::AdmittedKeySet<NodeId>> {
        self.collect_node_ids(
            self.property_index
                .iter()
                .filter(|((candidate_label, candidate_property, value), _)| {
                    label_id.is_none_or(|label| *candidate_label == label)
                        && candidate_property == property
                        && matches(value)
                })
                .flat_map(|(_, ids)| ids.iter().copied()),
            admission,
        )
    }

    fn node_ids_for_property_values<'value>(
        &self,
        label_id: Option<LabelId>,
        property: &str,
        values: impl IntoIterator<Item = &'value Value>,
        negated: bool,
        admission: &crate::read_view::GraphReadAdmission<'_>,
    ) -> Result<crate::read_view::AdmittedKeySet<NodeId>> {
        let mut keys = crate::read_view::AdmittedKeySet::new(admission)?;
        keys.try_extend(values)?;
        self.node_ids_for_indexed_property(
            label_id,
            property,
            |value| keys.contains(value) != negated,
            admission,
        )
    }

    fn node_ids_for_property_missing_or_null(
        &self,
        label_id: Option<LabelId>,
        property: &str,
        admission: &crate::read_view::GraphReadAdmission<'_>,
    ) -> Result<crate::read_view::AdmittedKeySet<NodeId>> {
        let non_null = self.node_ids_for_indexed_property(
            label_id,
            property,
            |value| value != &Value::Null,
            admission,
        )?;
        self.collect_node_ids(
            self.nodes
                .values()
                .filter(|node| self.node_matches_label(node, label_id))
                .filter(|node| !non_null.contains(&node.id))
                .map(|node| node.id),
            admission,
        )
    }

    fn node_ids_for_default_if_null_eq(
        &self,
        label_id: Option<LabelId>,
        property: &str,
        values: (&Value, &Value, &Value),
        admission: &crate::read_view::GraphReadAdmission<'_>,
    ) -> Result<crate::read_view::AdmittedKeySet<NodeId>> {
        let (empty, default, value) = values;
        if value == default {
            let mut ids =
                self.node_ids_for_property_missing_or_null(label_id, property, admission)?;
            // Borrow the two lookup values; the old temporary cloned payloads.
            let matched = self.node_ids_for_property_values(
                label_id,
                property,
                [empty, value],
                false,
                admission,
            )?;
            ids.try_extend(matched.iter().copied())?;
            return Ok(ids);
        }
        if value == empty || value == &Value::Null {
            return crate::read_view::AdmittedKeySet::new(admission);
        }
        self.node_ids_for_property_values(
            label_id,
            property,
            std::iter::once(value),
            false,
            admission,
        )
    }

    pub fn seek_nodes_by_property<'a>(
        &'a self,
        label_id: LabelId,
        property: &str,
        value: &Value,
    ) -> impl Iterator<Item = &'a NodeRecord> + 'a {
        self.property_index
            .get(&(label_id, property.to_string(), value.clone()))
            .into_iter()
            .flat_map(|node_ids| node_ids.iter())
            .filter_map(|node_id| self.nodes.get(node_id))
    }

    pub fn visit_nodes_by_property_owned(
        &self,
        label_id: LabelId,
        property: &str,
        values: &[Value],
        mut consumer: impl FnMut(NodeRecord) -> GraphScanControl,
    ) -> Result<GraphScanControl> {
        let Some(reader) = &self.canonical_base else {
            return self.visit_nodes_owned(Some(label_id), |node| {
                if node
                    .properties
                    .get(property)
                    .is_some_and(|candidate| values.iter().any(|value| candidate == value))
                {
                    consumer(node)
                } else {
                    GraphScanControl::Continue
                }
            });
        };
        let mut seen = BTreeSet::new();
        let equality_projection =
            self.persistent_property_projection
                .as_ref()
                .filter(|projection| {
                    projection.manifest().supports(
                        label_id,
                        property,
                        PersistentPropertyProjectionKind::Equality,
                    )
                });
        for value in values {
            let mut graph_control = GraphScanControl::Continue;
            let canonical_control = if let Some(projection) = equality_projection {
                let (report, control) = projection
                    .scan_equality_candidates(label_id, property, value, |node_id| {
                        if self.node_tombstones.contains(&node_id)
                            || self.nodes.contains_key(&node_id)
                            || !seen.insert(node_id)
                        {
                            return Ok(CanonicalScanControl::Continue);
                        }
                        let node = reader.get_node(node_id)?.ok_or_else(|| {
                            PersistentPropertyProjectionError::Corrupt(format!(
                                "property projection references missing canonical node {}",
                                node_id.0
                            ))
                        })?;
                        if !node.labels.contains(&label_id)
                            || node.properties.get(property) != Some(value)
                        {
                            return Err(PersistentPropertyProjectionError::Corrupt(format!(
                                "property projection candidate {} fails its canonical equality predicate",
                                node_id.0
                            )));
                        }
                        if consumer(node) == GraphScanControl::Stop {
                            graph_control = GraphScanControl::Stop;
                            return Ok(CanonicalScanControl::Stop);
                        }
                        Ok(CanonicalScanControl::Continue)
                    })
                    .map_err(|error| HawDBError::StorageIntegrity(error.to_string()))?;
                self.graph_index_read_metrics
                    .record_property(PersistentGraphIndexClass::NodeEquality, report);
                control
            } else {
                let (_, control) = reader
                    .scan_nodes_by_property_control(label_id, property, value, |node| {
                        if self.node_tombstones.contains(&node.id)
                            || self.nodes.contains_key(&node.id)
                            || !seen.insert(node.id)
                        {
                            return Ok(CanonicalScanControl::Continue);
                        }
                        if consumer(node) == GraphScanControl::Stop {
                            graph_control = GraphScanControl::Stop;
                            return Ok(CanonicalScanControl::Stop);
                        }
                        Ok(CanonicalScanControl::Continue)
                    })
                    .map_err(canonical_segment_error)?;
                control
            };
            if canonical_control == CanonicalScanControl::Stop {
                return Ok(graph_control);
            }
        }
        for node in self.nodes.values() {
            if node.labels.contains(&label_id)
                && node
                    .properties
                    .get(property)
                    .is_some_and(|candidate| values.iter().any(|value| candidate == value))
                && consumer(node.clone()) == GraphScanControl::Stop
            {
                return Ok(GraphScanControl::Stop);
            }
        }
        Ok(GraphScanControl::Continue)
    }

    #[doc(hidden)]
    pub fn visit_projected_nodes_by_property_owned(
        &self,
        label_id: LabelId,
        property: &str,
        values: &[Value],
        required_properties: &BTreeSet<String>,
        mut consumer: impl FnMut(ProjectedNodeRecord) -> GraphScanControl,
    ) -> Result<GraphScanControl> {
        self.visit_projected_nodes_by_property_admitted(
            label_id,
            property,
            values,
            required_properties,
            &mut crate::read_view::GraphReadAllocator::new(
                &mut |bytes| Ok(Box::new(UnaccountedReadAllocation(bytes))),
                None,
            ),
            &mut |node| Ok(consumer(node)),
        )
    }

    fn visit_projected_nodes_by_property_with_admission(
        &self,
        label_id: LabelId,
        property: &str,
        values: &[Value],
        required_properties: Option<&BTreeSet<String>>,
        admission: &ProjectionAdmission<'_>,
        mut consumer: impl FnMut(AdmittedProjection) -> GraphScanControl,
    ) -> Result<GraphScanControl> {
        let bounded_work = admission.has_work_limit();
        let Some(reader) = &self.canonical_base else {
            let mut decode_properties = required_properties.cloned();
            if let Some(properties) = &mut decode_properties {
                properties.insert(property.to_string());
            }
            return self.visit_projected_property_fallback_with_admission(
                label_id,
                property,
                values,
                decode_properties.as_ref(),
                admission,
                consumer,
            );
        };
        let Some(projection) = self
            .persistent_property_projection
            .as_ref()
            .filter(|_| !bounded_work)
            .filter(|projection| {
                projection.manifest().supports(
                    label_id,
                    property,
                    PersistentPropertyProjectionKind::Equality,
                )
            })
        else {
            let mut decode_properties = required_properties.cloned();
            if let Some(properties) = &mut decode_properties {
                properties.insert(property.to_string());
            }
            return self.visit_projected_property_fallback_with_admission(
                label_id,
                property,
                values,
                decode_properties.as_ref(),
                admission,
                consumer,
            );
        };

        let mut decode_properties = required_properties.cloned();
        if let Some(properties) = &mut decode_properties {
            properties.insert(property.to_string());
        }
        let mut seen = admission.key_set()?;
        for value in values {
            let mut graph_control = GraphScanControl::Continue;
            let (report, projection_control) = projection
                .scan_equality_candidates(label_id, property, value, |node_id| {
                    if self.node_tombstones.contains(&node_id)
                        || self.nodes.contains_key(&node_id)
                        || !admission.insert_key(&mut seen, node_id)?
                    {
                        return Ok(CanonicalScanControl::Continue);
                    }
                    let node = admission.canonical_selection(reader, node_id, decode_properties.as_ref())?
                        .ok_or_else(|| {
                            PersistentPropertyProjectionError::Corrupt(format!(
                                "property projection references missing canonical node {}",
                                node_id.0
                            ))
                        })?;
                    if !node.labels.contains(&label_id)
                        || node.properties.get(property) != Some(value)
                    {
                        return Err(PersistentPropertyProjectionError::Corrupt(format!(
                            "property projection candidate {} fails its canonical equality predicate",
                            node_id.0
                        )));
                    }
                    if consumer(node) == GraphScanControl::Stop {
                        graph_control = GraphScanControl::Stop;
                        return Ok(CanonicalScanControl::Stop);
                    }
                    Ok(CanonicalScanControl::Continue)
                })
                .map_err(|error| HawDBError::StorageIntegrity(error.to_string()))?;
            self.graph_index_read_metrics
                .record_property(PersistentGraphIndexClass::NodeEquality, report);
            if projection_control == CanonicalScanControl::Stop {
                return Ok(graph_control);
            }
        }
        for node in self.nodes.values() {
            if node.labels.contains(&label_id)
                && node
                    .properties
                    .get(property)
                    .is_some_and(|candidate| values.iter().any(|value| candidate == value))
                && admission.visit_live_selection(node, decode_properties.as_ref(), &mut consumer)
                    == GraphScanControl::Stop
            {
                return Ok(GraphScanControl::Stop);
            }
        }
        Ok(GraphScanControl::Continue)
    }

    pub fn seek_nodes_by_composite_property<'a>(
        &'a self,
        label_id: LabelId,
        predicates: &[(String, Value)],
    ) -> Vec<&'a NodeRecord> {
        self.composite_property_index
            .get(&(label_id, predicates.to_vec()))
            .into_iter()
            .flat_map(|node_ids| node_ids.iter())
            .filter_map(|node_id| self.nodes.get(node_id))
            .collect()
    }

    pub fn visit_nodes_by_composite_property_owned(
        &self,
        label_id: LabelId,
        predicates: &[(String, Value)],
        mut consumer: impl FnMut(NodeRecord) -> GraphScanControl,
    ) -> Result<GraphScanControl> {
        let Some((first_property, first_value)) = predicates.first() else {
            return self.visit_nodes_owned(Some(label_id), consumer);
        };
        if let (Some(reader), Some(projection)) = (
            self.canonical_base.as_ref(),
            self.persistent_property_projection.as_ref(),
        ) {
            let properties = predicates
                .iter()
                .map(|(property, _)| property.clone())
                .collect::<Vec<_>>();
            if projection
                .manifest()
                .supports_composite_equality(label_id, &properties)
            {
                let values = predicates
                    .iter()
                    .map(|(_, value)| value)
                    .collect::<Vec<_>>();
                let mut graph_control = GraphScanControl::Continue;
                let (report, projection_control) = projection
                    .scan_composite_equality_candidates(
                        label_id,
                        &properties,
                        &values,
                        |node_id| {
                            if self.node_tombstones.contains(&node_id)
                                || self.nodes.contains_key(&node_id)
                            {
                                return Ok(CanonicalScanControl::Continue);
                            }
                            let node = reader.get_node(node_id)?.ok_or_else(|| {
                                PersistentPropertyProjectionError::Corrupt(format!(
                                    "composite property projection references missing canonical node {}",
                                    node_id.0
                                ))
                            })?;
                            if !node.labels.contains(&label_id)
                                || !predicates.iter().all(|(property, value)| {
                                    node.properties.get(property) == Some(value)
                                })
                            {
                                return Err(PersistentPropertyProjectionError::Corrupt(format!(
                                    "composite property projection candidate {} fails its canonical predicate",
                                    node_id.0
                                )));
                            }
                            if consumer(node) == GraphScanControl::Stop {
                                graph_control = GraphScanControl::Stop;
                                return Ok(CanonicalScanControl::Stop);
                            }
                            Ok(CanonicalScanControl::Continue)
                        },
                    )
                    .map_err(|error| HawDBError::StorageIntegrity(error.to_string()))?;
                self.graph_index_read_metrics
                    .record_property(PersistentGraphIndexClass::NodeCompositeEquality, report);
                if projection_control == CanonicalScanControl::Stop {
                    return Ok(graph_control);
                }
                for node in self.nodes.values() {
                    if node.labels.contains(&label_id)
                        && predicates
                            .iter()
                            .all(|(property, value)| node.properties.get(property) == Some(value))
                        && consumer(node.clone()) == GraphScanControl::Stop
                    {
                        return Ok(GraphScanControl::Stop);
                    }
                }
                return Ok(GraphScanControl::Continue);
            }
        }
        self.visit_nodes_by_property_owned(
            label_id,
            first_property,
            std::slice::from_ref(first_value),
            |node| {
                if predicates
                    .iter()
                    .all(|(property, value)| node.properties.get(property) == Some(value))
                {
                    consumer(node)
                } else {
                    GraphScanControl::Continue
                }
            },
        )
    }

    fn visit_projected_nodes_by_composite_property_with_admission(
        &self,
        label_id: LabelId,
        predicates: &[(String, Value)],
        required_properties: Option<&BTreeSet<String>>,
        admission: &ProjectionAdmission<'_>,
        mut consumer: impl FnMut(AdmittedProjection) -> GraphScanControl,
    ) -> Result<GraphScanControl> {
        let bounded_work = admission.has_work_limit();
        let properties = predicates
            .iter()
            .map(|(property, _)| property.clone())
            .collect::<Vec<_>>();
        let Some((reader, projection)) = self
            .canonical_base
            .as_ref()
            .zip(self.persistent_property_projection.as_ref())
            .filter(|_| !bounded_work)
            .filter(|(_, projection)| {
                projection
                    .manifest()
                    .supports_composite_equality(label_id, &properties)
            })
        else {
            let mut decode_properties = required_properties.cloned();
            if let Some(selected) = &mut decode_properties {
                selected.extend(properties.iter().cloned());
            }
            let mut consume = |node: AdmittedProjection| {
                if predicates
                    .iter()
                    .all(|(property, value)| node.properties.get(property) == Some(value))
                {
                    consumer(node)
                } else {
                    GraphScanControl::Continue
                }
            };
            return if let Some((property, value)) = predicates.first() {
                self.visit_projected_nodes_by_property_with_admission(
                    label_id,
                    property,
                    std::slice::from_ref(value),
                    decode_properties.as_ref(),
                    admission,
                    consume,
                )
            } else {
                self.visit_selected_nodes_with_admission(
                    Some(label_id),
                    decode_properties.as_ref(),
                    admission,
                    &mut consume,
                )
            };
        };

        let mut decode_properties = required_properties.cloned();
        if let Some(selected) = &mut decode_properties {
            selected.extend(properties.iter().cloned());
        }
        let values = predicates
            .iter()
            .map(|(_, value)| value)
            .collect::<Vec<_>>();
        let mut graph_control = GraphScanControl::Continue;
        let (report, projection_control) = projection
            .scan_composite_equality_candidates(label_id, &properties, &values, |node_id| {
                if self.node_tombstones.contains(&node_id) || self.nodes.contains_key(&node_id) {
                    return Ok(CanonicalScanControl::Continue);
                }
                let node = admission
                    .canonical_selection(reader, node_id, decode_properties.as_ref())?
                    .ok_or_else(|| {
                        PersistentPropertyProjectionError::Corrupt(format!(
                            "composite property projection references missing canonical node {}",
                            node_id.0
                        ))
                    })?;
                if !node.labels.contains(&label_id)
                    || !predicates
                        .iter()
                        .all(|(property, value)| node.properties.get(property) == Some(value))
                {
                    return Err(PersistentPropertyProjectionError::Corrupt(format!(
                        "composite property projection candidate {} fails its canonical predicate",
                        node_id.0
                    )));
                }
                if consumer(node) == GraphScanControl::Stop {
                    graph_control = GraphScanControl::Stop;
                    return Ok(CanonicalScanControl::Stop);
                }
                Ok(CanonicalScanControl::Continue)
            })
            .map_err(|error| HawDBError::StorageIntegrity(error.to_string()))?;
        self.graph_index_read_metrics
            .record_property(PersistentGraphIndexClass::NodeCompositeEquality, report);
        if projection_control == CanonicalScanControl::Stop {
            return Ok(graph_control);
        }
        for node in self.nodes.values() {
            if node.labels.contains(&label_id)
                && predicates
                    .iter()
                    .all(|(property, value)| node.properties.get(property) == Some(value))
                && admission.visit_live_selection(node, decode_properties.as_ref(), &mut consumer)
                    == GraphScanControl::Stop
            {
                return Ok(GraphScanControl::Stop);
            }
        }
        Ok(GraphScanControl::Continue)
    }

    pub fn visit_nodes_by_composite_range_owned(
        &self,
        label_id: LabelId,
        seek: &hawdb_core::access::CompositeRangeSeek,
        mut consumer: impl FnMut(NodeRecord) -> GraphScanControl,
    ) -> Result<GraphScanControl> {
        validate_composite_range_seek(seek)?;
        if let (Some(reader), Some(projection)) = (
            self.canonical_base.as_ref(),
            self.persistent_property_projection.as_ref(),
        ) && composite_range_requires_full_key(seek)
            && projection
                .manifest()
                .supports_composite_equality(label_id, &seek.index_properties)
        {
            let equality_values = seek
                .equality_prefix
                .iter()
                .map(|(_, value)| value)
                .collect::<Vec<_>>();
            let mut graph_control = GraphScanControl::Continue;
            let (report, projection_control) = projection
                .scan_composite_range_candidates(
                    label_id,
                    &seek.index_properties,
                    &equality_values,
                    seek.lower.as_ref(),
                    seek.upper.as_ref(),
                    |node_id| {
                        if self.node_tombstones.contains(&node_id)
                            || self.nodes.contains_key(&node_id)
                        {
                            return Ok(CanonicalScanControl::Continue);
                        }
                        let node = reader.get_node(node_id)?.ok_or_else(|| {
                            PersistentPropertyProjectionError::Corrupt(format!(
                                "composite range projection references missing canonical node {}",
                                node_id.0
                            ))
                        })?;
                        if !node.labels.contains(&label_id)
                            || !node_matches_composite_range(&node, seek)
                        {
                            return Err(PersistentPropertyProjectionError::Corrupt(format!(
                                "composite range projection candidate {} fails its canonical predicate",
                                node_id.0
                            )));
                        }
                        if consumer(node) == GraphScanControl::Stop {
                            graph_control = GraphScanControl::Stop;
                            return Ok(CanonicalScanControl::Stop);
                        }
                        Ok(CanonicalScanControl::Continue)
                    },
                )
                .map_err(|error| HawDBError::StorageIntegrity(error.to_string()))?;
            self.graph_index_read_metrics
                .record_property(PersistentGraphIndexClass::NodeCompositeEquality, report);
            if projection_control == CanonicalScanControl::Stop {
                return Ok(graph_control);
            }
            for node in self.nodes.values() {
                if node.labels.contains(&label_id)
                    && node_matches_composite_range(node, seek)
                    && consumer(node.clone()) == GraphScanControl::Stop
                {
                    return Ok(GraphScanControl::Stop);
                }
            }
            return Ok(GraphScanControl::Continue);
        }

        let (first_property, first_value) = seek
            .equality_prefix
            .first()
            .expect("validated composite range equality prefix");
        self.visit_nodes_by_property_owned(
            label_id,
            first_property,
            std::slice::from_ref(first_value),
            |node| {
                if node_matches_composite_range(&node, seek) {
                    consumer(node)
                } else {
                    GraphScanControl::Continue
                }
            },
        )
    }

    fn visit_projected_nodes_by_composite_range_with_admission(
        &self,
        label_id: LabelId,
        seek: &hawdb_core::access::CompositeRangeSeek,
        required_properties: Option<&BTreeSet<String>>,
        admission: &ProjectionAdmission<'_>,
        mut consumer: impl FnMut(AdmittedProjection) -> GraphScanControl,
    ) -> Result<GraphScanControl> {
        let bounded_work = admission.has_work_limit();
        validate_composite_range_seek(seek)?;
        let Some((reader, projection)) = self
            .canonical_base
            .as_ref()
            .zip(self.persistent_property_projection.as_ref())
            .filter(|_| !bounded_work && composite_range_requires_full_key(seek))
            .filter(|(_, projection)| {
                projection
                    .manifest()
                    .supports_composite_equality(label_id, &seek.index_properties)
            })
        else {
            let mut decode_properties = required_properties.cloned();
            if let Some(properties) = &mut decode_properties {
                properties.extend(
                    seek.equality_prefix
                        .iter()
                        .map(|(property, _)| property.clone()),
                );
                properties.insert(seek.range_property.clone());
            }
            let (property, value) = seek
                .equality_prefix
                .first()
                .expect("validated composite range equality prefix");
            return self.visit_projected_nodes_by_property_with_admission(
                label_id,
                property,
                std::slice::from_ref(value),
                decode_properties.as_ref(),
                admission,
                |node| {
                    if projected_node_matches_composite_range(&node, seek) {
                        consumer(node)
                    } else {
                        GraphScanControl::Continue
                    }
                },
            );
        };

        let mut decode_properties = required_properties.cloned();
        if let Some(properties) = &mut decode_properties {
            properties.extend(
                seek.equality_prefix
                    .iter()
                    .map(|(property, _)| property.clone()),
            );
            properties.insert(seek.range_property.clone());
        }
        let equality_values = seek
            .equality_prefix
            .iter()
            .map(|(_, value)| value)
            .collect::<Vec<_>>();
        let mut graph_control = GraphScanControl::Continue;
        let (report, projection_control) = projection
            .scan_composite_range_candidates(
                label_id,
                &seek.index_properties,
                &equality_values,
                seek.lower.as_ref(),
                seek.upper.as_ref(),
                |node_id| {
                    if self.node_tombstones.contains(&node_id) || self.nodes.contains_key(&node_id)
                    {
                        return Ok(CanonicalScanControl::Continue);
                    }
                    let node = admission
                        .canonical_selection(reader, node_id, decode_properties.as_ref())?
                        .ok_or_else(|| {
                            PersistentPropertyProjectionError::Corrupt(format!(
                                "composite range projection references missing canonical node {}",
                                node_id.0
                            ))
                        })?;
                    if !node.labels.contains(&label_id)
                        || !projected_node_matches_composite_range(&node, seek)
                    {
                        return Err(PersistentPropertyProjectionError::Corrupt(format!(
                            "composite range projection candidate {} fails its canonical predicate",
                            node_id.0
                        )));
                    }
                    if consumer(node) == GraphScanControl::Stop {
                        graph_control = GraphScanControl::Stop;
                        return Ok(CanonicalScanControl::Stop);
                    }
                    Ok(CanonicalScanControl::Continue)
                },
            )
            .map_err(|error| HawDBError::StorageIntegrity(error.to_string()))?;
        self.graph_index_read_metrics
            .record_property(PersistentGraphIndexClass::NodeCompositeEquality, report);
        if projection_control == CanonicalScanControl::Stop {
            return Ok(graph_control);
        }
        for node in self.nodes.values() {
            if node.labels.contains(&label_id) && node_matches_composite_range(node, seek) {
                let node = admission
                    .live_selection(node, decode_properties.as_ref())
                    .map_err(canonical_segment_error)?;
                if consumer(node) == GraphScanControl::Stop {
                    return Ok(GraphScanControl::Stop);
                }
            }
        }
        Ok(GraphScanControl::Continue)
    }

    pub fn seek_nodes_by_property_range<'a>(
        &'a self,
        label_id: LabelId,
        property: &str,
        lower: Option<&(Value, bool)>,
        upper: Option<&(Value, bool)>,
    ) -> Vec<&'a NodeRecord> {
        self.property_index
            .iter()
            .filter_map(
                |((candidate_label_id, candidate_property, value), node_ids)| {
                    if *candidate_label_id != label_id || candidate_property != property {
                        return None;
                    }
                    if range_bounds_match(value, lower, upper) {
                        Some(node_ids)
                    } else {
                        None
                    }
                },
            )
            .flat_map(|node_ids| node_ids.iter())
            .filter_map(|node_id| self.nodes.get(node_id))
            .collect()
    }

    /// Uses bounded ordered index delivery when backing data is complete.
    /// Without an index/projection the ordinary unordered scan remains available.
    pub fn visit_nodes_by_property_range_owned(
        &self,
        label_id: LabelId,
        property: &str,
        lower: Option<&(Value, bool)>,
        upper: Option<&(Value, bool)>,
        mut consumer: impl FnMut(NodeRecord) -> GraphScanControl,
    ) -> Result<GraphScanControl> {
        if let Some(control) =
            self.try_visit_ordered_range(label_id, property, lower, upper, |value, source| {
                let node = match source {
                    RangeRecord::Live(node) => node.clone(),
                    RangeRecord::Canonical(reader, id) => {
                        reader.get_node(id)?.ok_or_else(|| {
                            PersistentPropertyProjectionError::Corrupt(format!(
                                "range projection references missing canonical node {}",
                                id.0
                            ))
                        })?
                    }
                };
                if !node.labels.contains(&label_id) || node.properties.get(property) != Some(value)
                {
                    return Err(PersistentPropertyProjectionError::Corrupt(format!(
                        "range index key disagrees with node {}",
                        node.id.0
                    )));
                }
                Ok(consumer(node))
            })?
        {
            return Ok(control);
        }
        self.visit_nodes_owned(Some(label_id), |node| {
            if node
                .properties
                .get(property)
                .is_some_and(|value| range_bounds_match(value, lower, upper))
            {
                consumer(node)
            } else {
                GraphScanControl::Continue
            }
        })
    }

    fn visit_projected_nodes_by_property_range_with_admission(
        &self,
        label_id: LabelId,
        property: &str,
        bounds: PropertyRangeBounds<'_>,
        required_properties: Option<&BTreeSet<String>>,
        admission: &ProjectionAdmission<'_>,
        mut consumer: impl FnMut(AdmittedProjection) -> GraphScanControl,
    ) -> Result<GraphScanControl> {
        let PropertyRangeBounds { lower, upper } = bounds;
        let mut decode_properties = required_properties.cloned();
        if let Some(properties) = &mut decode_properties {
            properties.insert(property.to_string());
        }
        if !admission.has_work_limit()
            && let Some(control) =
                self.try_visit_ordered_range(label_id, property, lower, upper, |value, source| {
                    let node = match source {
                        RangeRecord::Live(node) => {
                            admission.live_selection(node, decode_properties.as_ref())?
                        }
                        RangeRecord::Canonical(reader, id) => admission
                            .canonical_selection(reader, id, decode_properties.as_ref())?
                            .ok_or_else(|| {
                                PersistentPropertyProjectionError::Corrupt(format!(
                                    "range projection references missing canonical node {}",
                                    id.0
                                ))
                            })?,
                    };
                    if !node.labels.contains(&label_id)
                        || node.properties.get(property) != Some(value)
                    {
                        return Err(PersistentPropertyProjectionError::Corrupt(format!(
                            "range index key disagrees with projected node {}",
                            node.id.0
                        )));
                    }
                    Ok(consumer(node))
                })?
        {
            return Ok(control);
        }
        self.visit_projected_nodes_filtered_with_admission(
            label_id,
            decode_properties.as_ref(),
            admission,
            |properties| {
                properties
                    .get(property)
                    .is_some_and(|value| range_bounds_match(value, lower, upper))
            },
            consumer,
        )
    }

    fn try_visit_ordered_range(
        &self,
        label_id: LabelId,
        property: &str,
        lower: Option<&(Value, bool)>,
        upper: Option<&(Value, bool)>,
        mut consumer: impl FnMut(
            &Value,
            RangeRecord<'_>,
        ) -> std::result::Result<
            GraphScanControl,
            PersistentPropertyProjectionError,
        >,
    ) -> Result<Option<GraphScanControl>> {
        let projection = self
            .persistent_property_projection
            .as_ref()
            .filter(|projection| {
                projection.manifest().supports(
                    label_id,
                    property,
                    PersistentPropertyProjectionKind::Range,
                )
            });
        if self.canonical_base.is_some() && projection.is_none() {
            return Ok(None);
        }
        // The public visitor also accepts undeclared indexes. With no posting
        // keys, retain its scan fallback; a declared but empty index emits no rows.
        if self.canonical_base.is_none()
            && !self
                .property_index
                .keys()
                .any(|(label, key, _)| *label == label_id && key == property)
        {
            return Ok(None);
        }
        let mut live = self
            .property_index
            .iter()
            .filter(|((label, key, value), _)| {
                *label == label_id && key == property && range_bounds_match(value, lower, upper)
            })
            .flat_map(|((_, _, value), ids)| ids.iter().map(move |id| (value, id)))
            .filter(|(_, id)| !self.node_tombstones.contains(id))
            .filter_map(|(value, id)| self.nodes.get(id).map(|node| (value, node)))
            .peekable();
        let integrity_error = |error: PersistentPropertyProjectionError| {
            HawDBError::StorageIntegrity(error.to_string())
        };
        if let (Some(reader), Some(projection)) = (&self.canonical_base, projection) {
            // Both inputs use Value::cmp, not the mixed-numeric predicate
            // comparator. Keep only the projection's bounded block and one
            // borrowed delta entry; never materialize or sort result nodes.
            let (report, control) = projection
                .scan_range_entries(label_id, property, lower, upper, |value, id| {
                    if self.node_tombstones.contains(&id) || self.nodes.contains_key(&id) {
                        return Ok(CanonicalScanControl::Continue);
                    }
                    while live
                        .peek()
                        .is_some_and(|(key, node)| (*key, node.id) <= (value, id))
                    {
                        let (key, node) = live.next().expect("peeked live range entry");
                        if consumer(key, RangeRecord::Live(node))? == GraphScanControl::Stop {
                            return Ok(CanonicalScanControl::Stop);
                        }
                    }
                    Ok(
                        if consumer(value, RangeRecord::Canonical(reader, id))?
                            == GraphScanControl::Stop
                        {
                            CanonicalScanControl::Stop
                        } else {
                            CanonicalScanControl::Continue
                        },
                    )
                })
                .map_err(integrity_error)?;
            self.graph_index_read_metrics
                .record_property(PersistentGraphIndexClass::NodeRange, report);
            if control == CanonicalScanControl::Stop {
                return Ok(Some(GraphScanControl::Stop));
            }
        }
        for (value, node) in live {
            if consumer(value, RangeRecord::Live(node)).map_err(integrity_error)?
                == GraphScanControl::Stop
            {
                return Ok(Some(GraphScanControl::Stop));
            }
        }
        Ok(Some(GraphScanControl::Continue))
    }

    pub fn seek_nodes_by_full_text_property<'a>(
        &'a self,
        label_id: LabelId,
        property: &str,
        query: &str,
    ) -> Vec<&'a NodeRecord> {
        let tokens = full_text_query_tokens(query);
        let Some((first, rest)) = tokens.split_first() else {
            return Vec::new();
        };
        let mut candidates = self
            .full_text_property_index
            .get(&(label_id, property.to_string(), first.clone()))
            .map(|node_ids| node_ids.iter().copied().collect::<BTreeSet<_>>())
            .unwrap_or_default();
        for token in rest {
            let Some(ids) =
                self.full_text_property_index
                    .get(&(label_id, property.to_string(), token.clone()))
            else {
                return Vec::new();
            };
            candidates = candidates.intersection(ids).copied().collect();
            if candidates.is_empty() {
                return Vec::new();
            }
        }
        candidates
            .into_iter()
            .filter_map(|node_id| self.nodes.get(&node_id))
            .collect()
    }

    pub fn visit_nodes_by_full_text_property_owned(
        &self,
        label_id: LabelId,
        property: &str,
        query: &str,
        mut consumer: impl FnMut(NodeRecord) -> GraphScanControl,
    ) -> Result<GraphScanControl> {
        let query_tokens = full_text_query_tokens(query);
        if query_tokens.is_empty() {
            return Ok(GraphScanControl::Continue);
        }
        let matches_query = |node: &NodeRecord| match node.properties.get(property) {
            Some(Value::String(value)) => {
                let tokens = full_text_index_tokens(value);
                query_tokens.iter().all(|token| tokens.contains(token))
            }
            _ => false,
        };
        let Some(reader) = &self.canonical_base else {
            return self.visit_nodes_owned(Some(label_id), |node| {
                if matches_query(&node) {
                    consumer(node)
                } else {
                    GraphScanControl::Continue
                }
            });
        };
        let Some(projection) = self
            .persistent_property_projection
            .as_ref()
            .filter(|projection| {
                projection.manifest().supports(
                    label_id,
                    property,
                    PersistentPropertyProjectionKind::FullText,
                )
            })
        else {
            return self.visit_nodes_owned(Some(label_id), |node| {
                if matches_query(&node) {
                    consumer(node)
                } else {
                    GraphScanControl::Continue
                }
            });
        };
        let mut seed_token = None;
        let mut estimate_report =
            hawdb_storage::property_projection::PersistentPropertyProjectionReadReport::default();
        for token in &query_tokens {
            let (estimated_entries, report) = projection
                .estimate_full_text_token_entries(label_id, property, token)
                .map_err(|error| HawDBError::StorageIntegrity(error.to_string()))?;
            accumulate_property_projection_report(&mut estimate_report, report);
            if seed_token
                .as_ref()
                .is_none_or(|(_, current_entries)| estimated_entries < *current_entries)
            {
                seed_token = Some((token, estimated_entries));
            }
        }
        let seed_token = seed_token
            .map(|(token, _)| token)
            .expect("non-empty full-text query has a seed token");
        let mut graph_control = GraphScanControl::Continue;
        let (report, projection_control) = projection
            .scan_full_text_token_candidates(label_id, property, seed_token, |node_id| {
                if self.node_tombstones.contains(&node_id) || self.nodes.contains_key(&node_id) {
                    return Ok(CanonicalScanControl::Continue);
                }
                let node = reader.get_node(node_id)?.ok_or_else(|| {
                    PersistentPropertyProjectionError::Corrupt(format!(
                        "property projection references missing canonical node {}",
                        node_id.0
                    ))
                })?;
                if !node.labels.contains(&label_id) || !matches_query(&node) {
                    return Ok(CanonicalScanControl::Continue);
                }
                if consumer(node) == GraphScanControl::Stop {
                    graph_control = GraphScanControl::Stop;
                    return Ok(CanonicalScanControl::Stop);
                }
                Ok(CanonicalScanControl::Continue)
            })
            .map_err(|error| HawDBError::StorageIntegrity(error.to_string()))?;
        accumulate_property_projection_report(&mut estimate_report, report);
        self.graph_index_read_metrics
            .record_property(PersistentGraphIndexClass::NodeFullText, estimate_report);
        if projection_control == CanonicalScanControl::Stop {
            return Ok(graph_control);
        }
        for node in self.nodes.values() {
            if self.node_tombstones.contains(&node.id) {
                continue;
            }
            if node.labels.contains(&label_id)
                && matches_query(node)
                && consumer(node.clone()) == GraphScanControl::Stop
            {
                return Ok(GraphScanControl::Stop);
            }
        }
        Ok(GraphScanControl::Continue)
    }

    fn visit_projected_nodes_by_full_text_property_with_admission(
        &self,
        label_id: LabelId,
        property: &str,
        query: &str,
        required_properties: Option<&BTreeSet<String>>,
        admission: &ProjectionAdmission<'_>,
        mut consumer: impl FnMut(AdmittedProjection) -> GraphScanControl,
    ) -> Result<GraphScanControl> {
        let bounded_work = admission.has_work_limit();
        let query_tokens = full_text_query_tokens(query);
        if query_tokens.is_empty() {
            return Ok(GraphScanControl::Continue);
        }
        let Some((reader, projection)) = self
            .canonical_base
            .as_ref()
            .zip(self.persistent_property_projection.as_ref())
            .filter(|_| !bounded_work)
            .filter(|(_, projection)| {
                projection.manifest().supports(
                    label_id,
                    property,
                    PersistentPropertyProjectionKind::FullText,
                )
            })
        else {
            let mut decode_properties = required_properties.cloned();
            if let Some(properties) = &mut decode_properties {
                properties.insert(property.to_string());
            }
            return self.visit_projected_nodes_filtered_with_admission(
                label_id,
                decode_properties.as_ref(),
                admission,
                |properties| match properties.get(property) {
                    Some(Value::String(value)) => {
                        let tokens = full_text_index_tokens(value);
                        query_tokens.iter().all(|token| tokens.contains(token))
                    }
                    _ => false,
                },
                consumer,
            );
        };

        let mut decode_properties = required_properties.cloned();
        if let Some(properties) = &mut decode_properties {
            properties.insert(property.to_string());
        }
        let matches_query = |properties: &BTreeMap<String, Value>| match properties.get(property) {
            Some(Value::String(value)) => {
                let tokens = full_text_index_tokens(value);
                query_tokens.iter().all(|token| tokens.contains(token))
            }
            _ => false,
        };
        let mut seed_token = None;
        let mut estimate_report =
            hawdb_storage::property_projection::PersistentPropertyProjectionReadReport::default();
        for token in &query_tokens {
            let (estimated_entries, report) = projection
                .estimate_full_text_token_entries(label_id, property, token)
                .map_err(|error| HawDBError::StorageIntegrity(error.to_string()))?;
            accumulate_property_projection_report(&mut estimate_report, report);
            if seed_token
                .as_ref()
                .is_none_or(|(_, current_entries)| estimated_entries < *current_entries)
            {
                seed_token = Some((token, estimated_entries));
            }
        }
        let seed_token = seed_token
            .map(|(token, _)| token)
            .expect("non-empty full-text query has a seed token");
        let mut graph_control = GraphScanControl::Continue;
        let (report, projection_control) = projection
            .scan_full_text_token_candidates(label_id, property, seed_token, |node_id| {
                if self.node_tombstones.contains(&node_id) || self.nodes.contains_key(&node_id) {
                    return Ok(CanonicalScanControl::Continue);
                }
                let node = admission
                    .canonical_selection(reader, node_id, decode_properties.as_ref())?
                    .ok_or_else(|| {
                        PersistentPropertyProjectionError::Corrupt(format!(
                            "property projection references missing canonical node {}",
                            node_id.0
                        ))
                    })?;
                if !node.labels.contains(&label_id) || !matches_query(&node.properties) {
                    return Ok(CanonicalScanControl::Continue);
                }
                if consumer(node) == GraphScanControl::Stop {
                    graph_control = GraphScanControl::Stop;
                    return Ok(CanonicalScanControl::Stop);
                }
                Ok(CanonicalScanControl::Continue)
            })
            .map_err(|error| HawDBError::StorageIntegrity(error.to_string()))?;
        accumulate_property_projection_report(&mut estimate_report, report);
        self.graph_index_read_metrics
            .record_property(PersistentGraphIndexClass::NodeFullText, estimate_report);
        if projection_control == CanonicalScanControl::Stop {
            return Ok(graph_control);
        }
        for node in self.nodes.values() {
            if self.node_tombstones.contains(&node.id) {
                continue;
            }
            if node.labels.contains(&label_id)
                && matches_query(&node.properties)
                && admission.visit_live_selection(node, decode_properties.as_ref(), &mut consumer)
                    == GraphScanControl::Stop
            {
                return Ok(GraphScanControl::Stop);
            }
        }
        Ok(GraphScanControl::Continue)
    }

    #[inline]
    pub fn outgoing_relationships<'a>(
        &'a self,
        source: NodeId,
        rel_type: RelTypeId,
    ) -> impl Iterator<Item = &'a RelRecord> + 'a {
        self.outgoing
            .get(&(source, rel_type))
            .into_iter()
            .flat_map(AdjacencyPostingList::iter_copied)
            .filter_map(|entry| self.relationships.get(&entry.relationship_id))
    }

    #[inline]
    pub fn incoming_relationships<'a>(
        &'a self,
        target: NodeId,
        rel_type: RelTypeId,
    ) -> impl Iterator<Item = &'a RelRecord> + 'a {
        self.incoming
            .get(&(target, rel_type))
            .into_iter()
            .flat_map(AdjacencyPostingList::iter_copied)
            .filter_map(|entry| self.relationships.get(&entry.relationship_id))
    }

    pub fn visit_adjacent_relationships_owned(
        &self,
        node_id: NodeId,
        rel_type: Option<RelTypeId>,
        direction: AdjacencyDirection,
        mut consumer: impl FnMut(RelRecord) -> GraphScanControl,
    ) -> Result<GraphScanControl> {
        let Some(reader) = &self.canonical_base else {
            if let Some(rel_type) = rel_type {
                return self.visit_live_adjacent_relationships_owned(
                    node_id, rel_type, direction, consumer,
                );
            }
            return self.visit_relationships_owned(rel_type, |relationship| {
                let adjacent = match direction {
                    AdjacencyDirection::Outgoing => relationship.source == node_id,
                    AdjacencyDirection::Incoming => relationship.target == node_id,
                };
                if adjacent {
                    consumer(relationship)
                } else {
                    GraphScanControl::Continue
                }
            });
        };
        let mut graph_control = GraphScanControl::Continue;
        let canonical_control = {
            let mut consume_canonical = |relationship: RelRecord| {
                if self.relationship_tombstones.contains(&relationship.id)
                    || self.relationships.contains_key(&relationship.id)
                {
                    return CanonicalScanControl::Continue;
                }
                if consumer(relationship) == GraphScanControl::Stop {
                    graph_control = GraphScanControl::Stop;
                    CanonicalScanControl::Stop
                } else {
                    CanonicalScanControl::Continue
                }
            };
            if let Some(adjacency) = &self.canonical_adjacency {
                let (report, control) = adjacency
                    .scan_endpoint_entries_control(node_id, direction, rel_type, |entry| {
                        let relationship = match entry {
                            CanonicalAdjacencyEntry::Inline(relationship) => relationship,
                            CanonicalAdjacencyEntry::CanonicalReference { relationship_id } => {
                                reader
                                    .get_relationship(relationship_id)
                                    .map_err(|error| {
                                        hawdb_storage::canonical_adjacency::CanonicalAdjacencyError::Source(
                                            error.to_string(),
                                        )
                                    })?
                                    .ok_or_else(|| {
                                        hawdb_storage::canonical_adjacency::CanonicalAdjacencyError::Corrupt(format!(
                                            "canonical adjacency references missing relationship {}",
                                            relationship_id.0
                                        ))
                                    })?
                            }
                        };
                        Ok(consume_canonical(relationship))
                    })
                    .map_err(|error| HawDBError::StorageIntegrity(error.to_string()))?;
                let class = match direction {
                    AdjacencyDirection::Outgoing => PersistentGraphIndexClass::ForwardAdjacency,
                    AdjacencyDirection::Incoming => PersistentGraphIndexClass::ReverseAdjacency,
                };
                self.graph_index_read_metrics
                    .record_adjacency(class, report);
                control
            } else {
                let endpoint_direction = match direction {
                    AdjacencyDirection::Outgoing => CanonicalEndpointDirection::Source,
                    AdjacencyDirection::Incoming => CanonicalEndpointDirection::Target,
                };
                reader
                    .scan_relationships_for_endpoint_control(
                        node_id,
                        endpoint_direction,
                        rel_type,
                        |relationship| Ok(consume_canonical(relationship)),
                    )
                    .map_err(canonical_segment_error)?
                    .1
            }
        };
        if canonical_control == CanonicalScanControl::Stop {
            return Ok(graph_control);
        }
        if let Some(rel_type) = rel_type {
            return self
                .visit_live_adjacent_relationships_owned(node_id, rel_type, direction, consumer);
        }
        for relationship in self.relationships.values() {
            if rel_type.is_some_and(|rel_type| relationship.rel_type != rel_type) {
                continue;
            }
            let adjacent = match direction {
                AdjacencyDirection::Outgoing => relationship.source == node_id,
                AdjacencyDirection::Incoming => relationship.target == node_id,
            };
            if adjacent && consumer(relationship.clone()) == GraphScanControl::Stop {
                return Ok(GraphScanControl::Stop);
            }
        }
        Ok(GraphScanControl::Continue)
    }

    fn visit_live_adjacent_relationships_owned(
        &self,
        node_id: NodeId,
        rel_type: RelTypeId,
        direction: AdjacencyDirection,
        mut consumer: impl FnMut(RelRecord) -> GraphScanControl,
    ) -> Result<GraphScanControl> {
        let Some(entries) = self.adjacency_relationship_ids(node_id, rel_type, direction) else {
            return Ok(GraphScanControl::Continue);
        };
        for entry in entries.iter_copied() {
            let Some(relationship) = self.relationships.get(&entry.relationship_id) else {
                return Err(HawDBError::StorageIntegrity(format!(
                    "live adjacency references missing relationship {}",
                    entry.relationship_id.0
                )));
            };
            if consumer(relationship.clone()) == GraphScanControl::Stop {
                return Ok(GraphScanControl::Stop);
            }
        }
        Ok(GraphScanControl::Continue)
    }

    /// Ordered keys remain separately bounded; each selected payload transfers
    /// its own before-copy permit. No property is owned while sorting keys.
    pub fn visit_ordered_adjacent_relationships_with_allocation(
        &self,
        node_id: NodeId,
        rel_type: Option<RelTypeId>,
        direction: AdjacencyDirection,
        memory_budget_bytes: usize,
        admit: &mut ControlledGraphReadAllocator<'_>,
        mut consumer: impl FnMut(AdmittedRelationshipRecord) -> Result<GraphScanControl>,
    ) -> Result<GraphScanControl> {
        let work = admit.work();
        if let Some(rel_type) = rel_type
            && (self.canonical_base.is_none() || self.canonical_adjacency.is_some())
        {
            return self.visit_typed_adjacent_relationships_with_allocation(
                node_id,
                rel_type,
                direction,
                admit,
                &mut consumer,
            );
        }
        let Some(mut key_allocation) = admit.reserve(0)? else {
            return Ok(GraphScanControl::Stop);
        };
        let mut charge_key_bytes =
            |bytes| grow_relationship_key_allocation(key_allocation.as_mut(), bytes);
        let mut entries = Vec::new();
        let adjacency = match direction {
            AdjacencyDirection::Outgoing => &self.outgoing,
            AdjacencyDirection::Incoming => &self.incoming,
        };
        for ((_, group_type), postings) in
            adjacency.range(&(node_id, RelTypeId(0)), &(node_id, RelTypeId(u32::MAX)))
        {
            if rel_type.is_none_or(|expected| expected == *group_type) {
                for entry in postings.iter_copied() {
                    work.charge_work(1)?;
                    push_admitted_adjacency_key(
                        &mut entries,
                        entry,
                        memory_budget_bytes,
                        &mut charge_key_bytes,
                    )?;
                }
            }
        }
        if let Some(reader) = self.canonical_base.as_ref() {
            let mut collect_error = None;
            if let Some(adjacency) = self.canonical_adjacency.as_ref() {
                let (report, _) = adjacency
                    .scan_endpoint_inputs_control(node_id, direction, rel_type, |input| {
                        if let Err(error) = work.charge_work(1) {
                            collect_error = Some(error);
                            return Ok(CanonicalScanControl::Stop);
                        }
                        if !self
                            .relationship_tombstones
                            .contains(&input.relationship_id)
                            && !self.relationships.contains_key(&input.relationship_id)
                            && let Err(error) = push_admitted_adjacency_key(
                                &mut entries,
                                OrderedAdjacencyEntry {
                                    neighbor_id: input.neighbor_id,
                                    relationship_id: input.relationship_id,
                                },
                                memory_budget_bytes,
                                &mut charge_key_bytes,
                            )
                        {
                            collect_error = Some(error);
                            return Ok(CanonicalScanControl::Stop);
                        }
                        Ok(CanonicalScanControl::Continue)
                    })
                    .map_err(|error| HawDBError::StorageIntegrity(error.to_string()))?;
                self.graph_index_read_metrics.record_adjacency(
                    match direction {
                        AdjacencyDirection::Outgoing => PersistentGraphIndexClass::ForwardAdjacency,
                        AdjacencyDirection::Incoming => PersistentGraphIndexClass::ReverseAdjacency,
                    },
                    report,
                );
            } else {
                // Older supported layouts lack the adjacency artifact. Keep the
                // existing canonical scan fallback, with properties encoded.
                reader
                    .scan_relationship_layouts_control(|id, layout| {
                        if let Err(error) = work.charge_work(1) {
                            collect_error = Some(error);
                            return Ok(CanonicalScanControl::Stop);
                        }
                        let (endpoint, neighbor) = match direction {
                            AdjacencyDirection::Outgoing => (layout.source, layout.target),
                            AdjacencyDirection::Incoming => (layout.target, layout.source),
                        };
                        if endpoint == node_id
                            && rel_type.is_none_or(|expected| expected == layout.rel_type)
                            && !self.relationship_tombstones.contains(&id)
                            && !self.relationships.contains_key(&id)
                            && let Err(error) = push_admitted_adjacency_key(
                                &mut entries,
                                OrderedAdjacencyEntry {
                                    neighbor_id: neighbor,
                                    relationship_id: id,
                                },
                                memory_budget_bytes,
                                &mut charge_key_bytes,
                            )
                        {
                            collect_error = Some(error);
                            return Ok(CanonicalScanControl::Stop);
                        }
                        Ok(CanonicalScanControl::Continue)
                    })
                    .map_err(canonical_segment_error)?;
            }
            if let Some(error) = collect_error {
                return Err(error);
            }
        }
        entries.sort_unstable();
        entries.dedup();
        for entry in entries {
            if emit_admitted_relationship(
                self,
                (node_id, rel_type, direction),
                entry,
                admit,
                &mut consumer,
            )? == GraphScanControl::Stop
            {
                return Ok(GraphScanControl::Stop);
            }
        }
        Ok(GraphScanControl::Continue)
    }

    fn visit_typed_adjacent_relationships_with_allocation(
        &self,
        node_id: NodeId,
        rel_type: RelTypeId,
        direction: AdjacencyDirection,
        admit: &mut ControlledGraphReadAllocator<'_>,
        consumer: &mut impl FnMut(AdmittedRelationshipRecord) -> Result<GraphScanControl>,
    ) -> Result<GraphScanControl> {
        let work = admit.work();
        let selection = (node_id, Some(rel_type), direction);
        let mut live_entries = self
            .adjacency_relationship_ids(node_id, rel_type, direction)
            .into_iter()
            .flat_map(AdjacencyPostingList::iter_copied)
            .peekable();
        let Some(adjacency) = self.canonical_adjacency.as_ref() else {
            return emit_admitted_live_before(
                self,
                selection,
                &mut live_entries,
                None,
                admit,
                consumer,
            );
        };
        let mut consumer_error = None;
        let (report, control) = adjacency
            .scan_endpoint_inputs_control(node_id, direction, Some(rel_type), |input| {
                if let Err(error) = work.charge_work(1) {
                    consumer_error = Some(error);
                    return Ok(CanonicalScanControl::Stop);
                }
                if self
                    .relationship_tombstones
                    .contains(&input.relationship_id)
                    || self.relationships.contains_key(&input.relationship_id)
                {
                    return Ok(CanonicalScanControl::Continue);
                }
                let entry = OrderedAdjacencyEntry {
                    neighbor_id: input.neighbor_id,
                    relationship_id: input.relationship_id,
                };
                let result = emit_admitted_live_before(
                    self,
                    selection,
                    &mut live_entries,
                    Some(entry),
                    admit,
                    consumer,
                )
                .and_then(|control| match control {
                    GraphScanControl::Continue => {
                        emit_admitted_relationship(self, selection, entry, admit, consumer)
                    }
                    GraphScanControl::Stop => Ok(GraphScanControl::Stop),
                });
                match result {
                    Ok(GraphScanControl::Continue) => Ok(CanonicalScanControl::Continue),
                    Ok(GraphScanControl::Stop) => Ok(CanonicalScanControl::Stop),
                    Err(error) => {
                        // Query admission/consumer failure is not storage corruption.
                        consumer_error = Some(error);
                        Ok(CanonicalScanControl::Stop)
                    }
                }
            })
            .map_err(|error| HawDBError::StorageIntegrity(error.to_string()))?;
        self.graph_index_read_metrics.record_adjacency(
            match direction {
                AdjacencyDirection::Outgoing => PersistentGraphIndexClass::ForwardAdjacency,
                AdjacencyDirection::Incoming => PersistentGraphIndexClass::ReverseAdjacency,
            },
            report,
        );
        if let Some(error) = consumer_error {
            return Err(error);
        }
        if control == CanonicalScanControl::Stop {
            return Ok(GraphScanControl::Stop);
        }
        emit_admitted_live_before(self, selection, &mut live_entries, None, admit, consumer)
    }

    /// Filtered ordering retains projection pruning while admitting every
    /// selected source payload and every retained sort key before ownership.
    pub fn visit_filtered_ordered_relationships_with_allocation(
        &self,
        selection: (NodeId, Option<RelTypeId>, AdjacencyDirection),
        filter: &PropertyFilter,
        memory_budget_bytes: usize,
        admit: &mut ControlledGraphReadAllocator<'_>,
        mut consumer: impl FnMut(AdmittedRelationshipRecord) -> Result<GraphScanControl>,
    ) -> Result<(GraphScanControl, Option<ScanPruningReport>)> {
        let (node_id, rel_type, direction) = selection;
        let work = admit.work();
        if !work.has_work_limit()
            && self.canonical_base.is_none()
            && let Some(rel_type) = rel_type
        {
            let adjacency_entries = self
                .adjacency_relationship_ids(node_id, rel_type, direction)
                .map(AdjacencyPostingList::len)
                .unwrap_or_default();
            let stopped = std::cell::Cell::new(false);
            let candidate = self.prune_resident_relationships_admitted(
                rel_type,
                filter,
                adjacency_entries,
                &mut |bytes| {
                    if stopped.get() {
                        return Ok(None);
                    }
                    let allocation = admit.reserve(bytes)?;
                    stopped.set(allocation.is_none());
                    Ok(allocation)
                },
            )?;
            if stopped.get() {
                return Ok((GraphScanControl::Stop, None));
            }
            if let Some(candidate) = candidate {
                let count = candidate.ids.len();
                let Some(mut keys) = admit.reserve(0)? else {
                    return Ok((GraphScanControl::Stop, None));
                };
                let mut charge = |bytes| grow_relationship_key_allocation(keys.as_mut(), bytes);
                let mut entries = Vec::new();
                let mut output_count = 0usize;
                for id in candidate.ids.iter() {
                    let record = self.relationships.get(id).ok_or_else(|| {
                        HawDBError::StorageIntegrity(format!(
                            "relationship property index references missing relationship {}",
                            id.0
                        ))
                    })?;
                    if property_filter_matches(filter, id.0, &record.properties) {
                        output_count = output_count.saturating_add(1);
                        if relationship_matches_endpoint(record, node_id, direction) {
                            push_admitted_adjacency_key(
                                &mut entries,
                                ordered_relationship_key(record, direction),
                                memory_budget_bytes,
                                &mut charge,
                            )?;
                        }
                    }
                }
                entries.sort_unstable();
                entries.dedup();
                for entry in entries {
                    if emit_admitted_relationship(self, selection, entry, admit, &mut consumer)?
                        == GraphScanControl::Stop
                    {
                        return Ok((GraphScanControl::Stop, None));
                    }
                }
                let before = self.relationship_count_for_type(Some(rel_type));
                return Ok((
                    GraphScanControl::Continue,
                    Some(ScanPruningReport {
                        target_kind: ScanPruningTargetKind::Relationship,
                        label_id: None,
                        rel_type_id: Some(rel_type),
                        strategy: candidate.strategy,
                        pruned: true,
                        exact_empty: count == 0,
                        candidate_count_before_pruning: before,
                        pruned_candidate_count: before.saturating_sub(count),
                        candidate_count_before_filter: count,
                        output_count,
                        filtered_out_count: count.saturating_sub(output_count),
                    }),
                ));
            }
        }
        let fallback = |admit: &mut ControlledGraphReadAllocator<'_>,
                        consumer: &mut dyn FnMut(
            AdmittedRelationshipRecord,
        ) -> Result<GraphScanControl>| {
            self.visit_ordered_adjacent_relationships_with_allocation(
                node_id,
                rel_type,
                direction,
                memory_budget_bytes,
                admit,
                |row| {
                    let (record, allocation) = row.into_parts();
                    if property_filter_matches(filter, record.id.0, &record.properties) {
                        consumer(AdmittedRelationshipRecord::new(record, allocation))
                    } else {
                        Ok(GraphScanControl::Continue)
                    }
                },
            )
            .map(|control| (control, None))
        };
        if work.has_work_limit() {
            return fallback(admit, &mut consumer);
        }
        let (Some(rel_type), Some(adjacency), Some(projection)) = (
            rel_type,
            self.canonical_adjacency.as_ref(),
            self.persistent_property_projection.as_ref(),
        ) else {
            return fallback(admit, &mut consumer);
        };
        // Probe alternatives contain borrowed filter values. Admit their small
        // descriptor/vector scratch before constructing the index probe.
        let probe_bytes = relationship_probe_allocation_bytes(filter);
        let Some(_probe_allocation) = admit.reserve(probe_bytes)? else {
            return Ok((GraphScanControl::Stop, None));
        };
        if _probe_allocation.bytes() < probe_bytes {
            return Err(HawDBError::Execution(
                "relationship probe admission returned an insufficient allocation permit".into(),
            ));
        }
        let Some(probe) = relationship_projection_probe(projection, rel_type, filter)
            .map_err(|error| HawDBError::StorageIntegrity(error.to_string()))?
        else {
            return fallback(admit, &mut consumer);
        };
        let adjacency_entries = adjacency
            .estimate_endpoint_entries(node_id, direction, Some(rel_type))
            .map_err(|error| HawDBError::StorageIntegrity(error.to_string()))?;
        if probe.estimated_entries() > adjacency_entries {
            self.graph_index_read_metrics
                .record_property(probe.index_class(), probe.estimate_report());
            return fallback(admit, &mut consumer);
        }
        let Some(mut key_allocation) = admit.reserve(0)? else {
            return Ok((GraphScanControl::Stop, None));
        };
        let mut charge = |bytes| grow_relationship_key_allocation(key_allocation.as_mut(), bytes);
        let mut entries = Vec::new();
        let mut read_error = None;
        let mut candidate_count = 0usize;
        let mut output_count = 0usize;
        let (report, control) = probe.scan(projection, |id| {
            if self.relationship_tombstones.contains(&id) || self.relationships.contains_key(&id) {
                return Ok(CanonicalScanControl::Continue);
            }
            candidate_count = candidate_count.saturating_add(1);
            let row = match self.relationship_with_allocation(id, admit) {
                Ok(AdmittedRelationshipRead::Relationship(row)) => row,
                Ok(AdmittedRelationshipRead::Stopped) => return Ok(CanonicalScanControl::Stop),
                Ok(AdmittedRelationshipRead::Missing) => return Err(PersistentPropertyProjectionError::Corrupt(
                    format!("relationship property projection references missing canonical relationship {}", id.0),
                )),
                Err(error) => { read_error = Some(error); return Ok(CanonicalScanControl::Stop); }
            };
            let (record, _allocation) = row.into_parts();
            if record.rel_type != rel_type || !probe.matches(&record) {
                return Err(PersistentPropertyProjectionError::Corrupt(format!(
                    "relationship property projection candidate {} fails its canonical seek predicate", id.0,
                )));
            }
            if relationship_matches_endpoint(&record, node_id, direction)
                && property_filter_matches(filter, id.0, &record.properties)
            {
                output_count = output_count.saturating_add(1);
                if let Err(error) = push_admitted_adjacency_key(
                    &mut entries, ordered_relationship_key(&record, direction), memory_budget_bytes, &mut charge,
                ) {
                    read_error = Some(error);
                    return Ok(CanonicalScanControl::Stop);
                }
            }
            Ok(CanonicalScanControl::Continue)
        }).map_err(|error| HawDBError::StorageIntegrity(error.to_string()))?;
        self.graph_index_read_metrics
            .record_property(probe.index_class(), report);
        if let Some(error) = read_error {
            return Err(error);
        }
        if control == CanonicalScanControl::Stop {
            return Ok((GraphScanControl::Stop, None));
        }
        for record in self.relationships.values() {
            if record.rel_type != rel_type || !probe.matches(record) {
                continue;
            }
            candidate_count = candidate_count.saturating_add(1);
            if relationship_matches_endpoint(record, node_id, direction)
                && property_filter_matches(filter, record.id.0, &record.properties)
            {
                output_count = output_count.saturating_add(1);
                push_admitted_adjacency_key(
                    &mut entries,
                    ordered_relationship_key(record, direction),
                    memory_budget_bytes,
                    &mut charge,
                )?;
            }
        }
        entries.sort_unstable();
        entries.dedup();
        for entry in entries {
            if emit_admitted_relationship(self, selection, entry, admit, &mut consumer)?
                == GraphScanControl::Stop
            {
                return Ok((GraphScanControl::Stop, None));
            }
        }
        let before = self.relationship_count_for_type(Some(rel_type));
        Ok((
            GraphScanControl::Continue,
            Some(ScanPruningReport {
                target_kind: ScanPruningTargetKind::Relationship,
                label_id: None,
                rel_type_id: Some(rel_type),
                strategy: probe.strategy(),
                pruned: true,
                exact_empty: candidate_count == 0,
                candidate_count_before_pruning: before,
                pruned_candidate_count: before.saturating_sub(candidate_count),
                candidate_count_before_filter: candidate_count,
                output_count,
                filtered_out_count: candidate_count.saturating_sub(output_count),
            }),
        ))
    }

    pub fn try_visit_ordered_adjacent_relationships_owned(
        &self,
        node_id: NodeId,
        rel_type: Option<RelTypeId>,
        direction: AdjacencyDirection,
        memory_budget_bytes: usize,
        consumer: impl FnMut(RelRecord) -> Result<GraphScanControl>,
    ) -> Result<GraphScanControl> {
        self.try_visit_ordered_adjacent_relationships_accounted(
            node_id,
            rel_type,
            direction,
            memory_budget_bytes,
            |_| Ok(()),
            consumer,
        )
    }

    #[doc(hidden)]
    pub fn try_visit_ordered_adjacent_relationships_accounted(
        &self,
        node_id: NodeId,
        rel_type: Option<RelTypeId>,
        direction: AdjacencyDirection,
        memory_budget_bytes: usize,
        charge_key_bytes: impl FnMut(usize) -> Result<()>,
        mut consumer: impl FnMut(RelRecord) -> Result<GraphScanControl>,
    ) -> Result<GraphScanControl> {
        let Some(rel_type) = rel_type else {
            return self.try_visit_compact_sorted_adjacency(
                node_id,
                None,
                direction,
                memory_budget_bytes,
                charge_key_bytes,
                consumer,
            );
        };
        let mut live_entries = self
            .adjacency_relationship_ids(node_id, rel_type, direction)
            .into_iter()
            .flat_map(AdjacencyPostingList::iter_copied)
            .peekable();
        let (Some(reader), Some(adjacency)) = (
            self.canonical_base.as_ref(),
            self.canonical_adjacency.as_ref(),
        ) else {
            return emit_live_adjacency_before(self, &mut live_entries, None, &mut consumer);
        };

        let mut graph_control = GraphScanControl::Continue;
        let mut consumer_error = None;
        let (report, canonical_control) = adjacency
            .scan_endpoint_entries_control(node_id, direction, Some(rel_type), |entry| {
                let relationship = match entry {
                    CanonicalAdjacencyEntry::Inline(relationship) => relationship,
                    CanonicalAdjacencyEntry::CanonicalReference { relationship_id } => reader
                        .get_relationship(relationship_id)
                        .map_err(|error| {
                            hawdb_storage::canonical_adjacency::CanonicalAdjacencyError::Source(
                                error.to_string(),
                            )
                        })?
                        .ok_or_else(|| {
                            hawdb_storage::canonical_adjacency::CanonicalAdjacencyError::Corrupt(
                                format!(
                                    "canonical adjacency references missing relationship {}",
                                    relationship_id.0
                                ),
                            )
                        })?,
                };
                if self.relationship_tombstones.contains(&relationship.id)
                    || self.relationships.contains_key(&relationship.id)
                {
                    return Ok(CanonicalScanControl::Continue);
                }
                let canonical_key = ordered_relationship_key(&relationship, direction);
                match emit_live_adjacency_before(
                    self,
                    &mut live_entries,
                    Some(canonical_key),
                    &mut consumer,
                ) {
                    Ok(GraphScanControl::Continue) => {}
                    Ok(GraphScanControl::Stop) => {
                        graph_control = GraphScanControl::Stop;
                        return Ok(CanonicalScanControl::Stop);
                    }
                    Err(error) => {
                        consumer_error = Some(error);
                        graph_control = GraphScanControl::Stop;
                        return Ok(CanonicalScanControl::Stop);
                    }
                }
                match consumer(relationship) {
                    Ok(GraphScanControl::Continue) => Ok(CanonicalScanControl::Continue),
                    Ok(GraphScanControl::Stop) => {
                        graph_control = GraphScanControl::Stop;
                        Ok(CanonicalScanControl::Stop)
                    }
                    Err(error) => {
                        consumer_error = Some(error);
                        graph_control = GraphScanControl::Stop;
                        Ok(CanonicalScanControl::Stop)
                    }
                }
            })
            .map_err(|error| HawDBError::StorageIntegrity(error.to_string()))?;
        self.graph_index_read_metrics.record_adjacency(
            match direction {
                AdjacencyDirection::Outgoing => PersistentGraphIndexClass::ForwardAdjacency,
                AdjacencyDirection::Incoming => PersistentGraphIndexClass::ReverseAdjacency,
            },
            report,
        );
        if let Some(error) = consumer_error {
            return Err(error);
        }
        if canonical_control == CanonicalScanControl::Stop {
            return Ok(graph_control);
        }
        emit_live_adjacency_before(self, &mut live_entries, None, &mut consumer)
    }

    fn try_visit_compact_sorted_adjacency(
        &self,
        node_id: NodeId,
        rel_type: Option<RelTypeId>,
        direction: AdjacencyDirection,
        memory_budget_bytes: usize,
        mut charge_key_bytes: impl FnMut(usize) -> Result<()>,
        mut consumer: impl FnMut(RelRecord) -> Result<GraphScanControl>,
    ) -> Result<GraphScanControl> {
        let mut entries = Vec::new();
        self.try_visit_adjacent_relationships_owned(
            node_id,
            rel_type,
            direction,
            |relationship| {
                push_compact_adjacency_key(
                    &mut entries,
                    ordered_relationship_key(&relationship, direction),
                    memory_budget_bytes,
                    &mut charge_key_bytes,
                )?;
                Ok(GraphScanControl::Continue)
            },
        )?;
        entries.sort_unstable();
        let mut index = 0usize;
        emit_compact_adjacency_before(self, &entries, &mut index, None, &mut consumer)
    }

    pub fn visit_adjacent_relationships_with_filter_owned(
        &self,
        node_id: NodeId,
        rel_type: Option<RelTypeId>,
        direction: AdjacencyDirection,
        filter: &PropertyFilter,
        mut consumer: impl FnMut(RelRecord) -> GraphScanControl,
    ) -> Result<(GraphScanControl, Option<ScanPruningReport>)> {
        let Some(rel_type) = rel_type else {
            return self.visit_adjacent_relationships_filter_fallback(
                node_id, None, direction, filter, consumer,
            );
        };
        let (Some(reader), Some(adjacency), Some(projection)) = (
            self.canonical_base.as_ref(),
            self.canonical_adjacency.as_ref(),
            self.persistent_property_projection.as_ref(),
        ) else {
            return self.visit_adjacent_relationships_filter_fallback(
                node_id,
                Some(rel_type),
                direction,
                filter,
                consumer,
            );
        };
        let Some(probe) = relationship_projection_probe(projection, rel_type, filter)
            .map_err(|error| HawDBError::StorageIntegrity(error.to_string()))?
        else {
            return self.visit_adjacent_relationships_filter_fallback(
                node_id,
                Some(rel_type),
                direction,
                filter,
                consumer,
            );
        };
        let adjacency_entries = adjacency
            .estimate_endpoint_entries(node_id, direction, Some(rel_type))
            .map_err(|error| HawDBError::StorageIntegrity(error.to_string()))?;
        if probe.estimated_entries() > adjacency_entries {
            self.graph_index_read_metrics
                .record_property(probe.index_class(), probe.estimate_report());
            return self.visit_adjacent_relationships_filter_fallback(
                node_id,
                Some(rel_type),
                direction,
                filter,
                consumer,
            );
        }

        let mut graph_control = GraphScanControl::Continue;
        let mut candidate_count = 0usize;
        let mut output_count = 0usize;
        let (projection_read_report, projection_control) = probe
            .scan(projection, |relationship_id| {
                if self.relationship_tombstones.contains(&relationship_id)
                    || self.relationships.contains_key(&relationship_id)
                {
                    return Ok(CanonicalScanControl::Continue);
                }
                candidate_count = candidate_count.saturating_add(1);
                let relationship = reader
                    .get_relationship(relationship_id)?
                    .ok_or_else(|| {
                        PersistentPropertyProjectionError::Corrupt(format!(
                            "relationship property projection references missing canonical relationship {}",
                            relationship_id.0
                        ))
                    })?;
                if relationship.rel_type != rel_type || !probe.matches(&relationship) {
                    return Err(PersistentPropertyProjectionError::Corrupt(format!(
                        "relationship property projection candidate {} fails its canonical seek predicate",
                        relationship_id.0
                    )));
                }
                if relationship_matches_endpoint(&relationship, node_id, direction)
                    && property_filter_matches(
                        filter,
                        relationship.id.0,
                        &relationship.properties,
                    )
                {
                    output_count = output_count.saturating_add(1);
                    if consumer(relationship) == GraphScanControl::Stop {
                        graph_control = GraphScanControl::Stop;
                        return Ok(CanonicalScanControl::Stop);
                    }
                }
                Ok(CanonicalScanControl::Continue)
            })
            .map_err(|error| HawDBError::StorageIntegrity(error.to_string()))?;
        self.graph_index_read_metrics
            .record_property(probe.index_class(), projection_read_report);
        if projection_control == CanonicalScanControl::Stop {
            return Ok((graph_control, None));
        }
        for relationship in self.relationships.values() {
            if relationship.rel_type != rel_type || !probe.matches(relationship) {
                continue;
            }
            candidate_count = candidate_count.saturating_add(1);
            if relationship_matches_endpoint(relationship, node_id, direction)
                && property_filter_matches(filter, relationship.id.0, &relationship.properties)
            {
                output_count = output_count.saturating_add(1);
                if consumer(relationship.clone()) == GraphScanControl::Stop {
                    return Ok((GraphScanControl::Stop, None));
                }
            }
        }
        let candidate_count_before_pruning = self.relationship_count_for_type(Some(rel_type));
        Ok((
            GraphScanControl::Continue,
            Some(ScanPruningReport {
                target_kind: ScanPruningTargetKind::Relationship,
                label_id: None,
                rel_type_id: Some(rel_type),
                strategy: probe.strategy(),
                pruned: true,
                exact_empty: candidate_count == 0,
                candidate_count_before_pruning,
                pruned_candidate_count: candidate_count_before_pruning
                    .saturating_sub(candidate_count),
                candidate_count_before_filter: candidate_count,
                output_count,
                filtered_out_count: candidate_count.saturating_sub(output_count),
            }),
        ))
    }

    fn visit_adjacent_relationships_filter_fallback(
        &self,
        node_id: NodeId,
        rel_type: Option<RelTypeId>,
        direction: AdjacencyDirection,
        filter: &PropertyFilter,
        mut consumer: impl FnMut(RelRecord) -> GraphScanControl,
    ) -> Result<(GraphScanControl, Option<ScanPruningReport>)> {
        if self.canonical_base.is_none()
            && let Some(rel_type) = rel_type
            && let Some(candidate) = self.prune_relationship_candidates(Some(rel_type), filter)
        {
            let adjacency_entries = self
                .adjacency_relationship_ids(node_id, rel_type, direction)
                .map(AdjacencyPostingList::len)
                .unwrap_or_default();
            if candidate.rel_ids.len() <= adjacency_entries {
                let candidate_count_before_filter = candidate.rel_ids.len();
                let mut output_count = 0usize;
                for relationship_id in &candidate.rel_ids {
                    let relationship =
                        self.relationships.get(relationship_id).ok_or_else(|| {
                            HawDBError::StorageIntegrity(format!(
                                "relationship property index references missing relationship {}",
                                relationship_id.0
                            ))
                        })?;
                    if !property_filter_matches(filter, relationship.id.0, &relationship.properties)
                    {
                        continue;
                    }
                    output_count = output_count.saturating_add(1);
                    if relationship_matches_endpoint(relationship, node_id, direction)
                        && consumer(relationship.clone()) == GraphScanControl::Stop
                    {
                        return Ok((GraphScanControl::Stop, None));
                    }
                }
                let candidate_count_before_pruning =
                    self.relationship_count_for_type(Some(rel_type));
                return Ok((
                    GraphScanControl::Continue,
                    Some(ScanPruningReport {
                        target_kind: ScanPruningTargetKind::Relationship,
                        label_id: None,
                        rel_type_id: Some(rel_type),
                        strategy: candidate.strategy,
                        pruned: true,
                        exact_empty: candidate.exact_empty,
                        candidate_count_before_pruning,
                        pruned_candidate_count: candidate_count_before_pruning
                            .saturating_sub(candidate_count_before_filter),
                        candidate_count_before_filter,
                        output_count,
                        filtered_out_count: candidate_count_before_filter
                            .saturating_sub(output_count),
                    }),
                ));
            }
        }
        self.visit_adjacent_relationships_owned(node_id, rel_type, direction, |relationship| {
            if property_filter_matches(filter, relationship.id.0, &relationship.properties) {
                consumer(relationship)
            } else {
                GraphScanControl::Continue
            }
        })
        .map(|control| (control, None))
    }

    pub fn try_visit_adjacent_relationships_owned(
        &self,
        node_id: NodeId,
        rel_type: Option<RelTypeId>,
        direction: AdjacencyDirection,
        mut consumer: impl FnMut(RelRecord) -> Result<GraphScanControl>,
    ) -> Result<GraphScanControl> {
        let mut consumer_error = None;
        let control = self.visit_adjacent_relationships_owned(
            node_id,
            rel_type,
            direction,
            |relationship| match consumer(relationship) {
                Ok(control) => control,
                Err(error) => {
                    consumer_error = Some(error);
                    GraphScanControl::Stop
                }
            },
        )?;
        match consumer_error {
            Some(error) => Err(error),
            None => Ok(control),
        }
    }

    pub fn adjacency_group_stats(
        &self,
        node_id: NodeId,
        rel_type: RelTypeId,
        direction: AdjacencyDirection,
    ) -> AdjacencyGroupStats {
        let degree = self
            .adjacency_relationship_ids(node_id, rel_type, direction)
            .map(AdjacencyPostingList::len)
            .unwrap_or_default();
        AdjacencyGroupStats {
            node_id,
            rel_type,
            direction,
            degree,
            layout: adjacency_layout_for_degree(degree),
        }
    }

    pub fn adjacency_group_stats_for_node(
        &self,
        node_id: NodeId,
        direction: AdjacencyDirection,
    ) -> Vec<AdjacencyGroupStats> {
        let adjacency = match direction {
            AdjacencyDirection::Outgoing => &self.outgoing,
            AdjacencyDirection::Incoming => &self.incoming,
        };
        let mut stats = adjacency
            .iter()
            .filter_map(|((group_node, rel_type), rel_ids)| {
                (*group_node == node_id).then_some(AdjacencyGroupStats {
                    node_id,
                    rel_type: *rel_type,
                    direction,
                    degree: rel_ids.len(),
                    layout: adjacency_layout_for_degree(rel_ids.len()),
                })
            })
            .collect::<Vec<_>>();
        stats.sort_by_key(|stats| {
            (
                stats.rel_type,
                adjacency_direction_sort_key(stats.direction),
            )
        });
        stats
    }

    pub fn ordered_adjacency_entries(
        &self,
        node_id: NodeId,
        rel_type: RelTypeId,
        direction: AdjacencyDirection,
    ) -> Vec<OrderedAdjacencyEntry> {
        self.adjacency_relationship_ids(node_id, rel_type, direction)
            .into_iter()
            .flat_map(AdjacencyPostingList::iter_copied)
            .collect()
    }

    pub fn ordered_adjacency_entries_for_node(
        &self,
        node_id: NodeId,
        direction: AdjacencyDirection,
    ) -> Vec<OrderedAdjacencyEntry> {
        let adjacency = match direction {
            AdjacencyDirection::Outgoing => &self.outgoing,
            AdjacencyDirection::Incoming => &self.incoming,
        };
        let mut entries = adjacency
            .iter()
            .filter(|((group_node, _), _)| *group_node == node_id)
            .flat_map(|(_, rel_ids)| rel_ids.iter_copied())
            .collect::<Vec<_>>();
        entries.sort_by_key(|entry| (entry.neighbor_id, entry.relationship_id));
        entries
    }

    pub fn scan_relationships<'a>(
        &'a self,
        rel_type: Option<RelTypeId>,
    ) -> impl Iterator<Item = &'a RelRecord> + 'a {
        self.relationships.values().filter(move |relationship| {
            rel_type
                .map(|rel_type| relationship.rel_type == rel_type)
                .unwrap_or(true)
        })
    }

    pub fn scan_relationships_with_filter_pruning<'a>(
        &'a self,
        rel_type: Option<RelTypeId>,
        filter: Option<&PropertyFilter>,
    ) -> ScanPrunedRelationshipScan<'a> {
        let candidate =
            filter.and_then(|filter| self.prune_relationship_candidates(rel_type, filter));
        let Some(candidate) = candidate else {
            let candidate_count_before_filter = self.relationship_count_for_type(rel_type);
            let relationships = self
                .scan_relationships(rel_type)
                .filter(|relationship| {
                    filter
                        .map(|filter| {
                            property_filter_matches(
                                filter,
                                relationship.id.0,
                                &relationship.properties,
                            )
                        })
                        .unwrap_or(true)
                })
                .collect::<Vec<_>>();
            let output_count = relationships.len();
            return ScanPrunedRelationshipScan {
                relationships,
                report: ScanPruningReport {
                    target_kind: ScanPruningTargetKind::Relationship,
                    label_id: None,
                    rel_type_id: rel_type,
                    strategy: ScanPruningStrategy::FullLabelScan,
                    pruned: false,
                    exact_empty: false,
                    candidate_count_before_pruning: candidate_count_before_filter,
                    pruned_candidate_count: 0,
                    candidate_count_before_filter,
                    output_count,
                    filtered_out_count: candidate_count_before_filter.saturating_sub(output_count),
                },
            };
        };

        let candidate_count_before_pruning = self.relationship_count_for_type(rel_type);
        let candidate_count_before_filter = candidate.rel_ids.len();
        let relationships = candidate
            .rel_ids
            .iter()
            .filter_map(|rel_id| self.relationships.get(rel_id))
            .filter(|relationship| self.relationship_matches_type(relationship, rel_type))
            .filter(|relationship| {
                filter
                    .map(|filter| {
                        property_filter_matches(filter, relationship.id.0, &relationship.properties)
                    })
                    .unwrap_or(true)
            })
            .collect::<Vec<_>>();
        let output_count = relationships.len();
        ScanPrunedRelationshipScan {
            relationships,
            report: ScanPruningReport {
                target_kind: ScanPruningTargetKind::Relationship,
                label_id: None,
                rel_type_id: rel_type,
                strategy: candidate.strategy,
                pruned: true,
                exact_empty: candidate.exact_empty,
                candidate_count_before_pruning,
                pruned_candidate_count: candidate_count_before_pruning
                    .saturating_sub(candidate_count_before_filter),
                candidate_count_before_filter,
                output_count,
                filtered_out_count: candidate_count_before_filter.saturating_sub(output_count),
            },
        }
    }

    pub fn relationship_count_for_type(&self, rel_type: Option<RelTypeId>) -> usize {
        let count = rel_type.map_or(self.basic_statistics.relationship_count, |rel_type| {
            self.basic_statistics
                .rel_type_counts
                .get(&rel_type)
                .copied()
                .unwrap_or_default()
        });
        usize::try_from(count).unwrap_or(usize::MAX)
    }

    fn relationship_matches_type(
        &self,
        relationship: &RelRecord,
        rel_type: Option<RelTypeId>,
    ) -> bool {
        rel_type
            .map(|rel_type| relationship.rel_type == rel_type)
            .unwrap_or(true)
    }

    fn prune_relationship_candidates(
        &self,
        rel_type: Option<RelTypeId>,
        filter: &PropertyFilter,
    ) -> Option<RelationshipScanPruningCandidate> {
        match filter {
            PropertyFilter::And(filters) => {
                self.prune_and_relationship_candidates(rel_type, filters)
            }
            PropertyFilter::Or(filters) => self.prune_or_relationship_candidates(rel_type, filters),
            PropertyFilter::Not(_) => None,
            PropertyFilter::IdEq { value } => Some(RelationshipScanPruningCandidate::exact(
                ScanPruningStrategy::IdEq,
                self.rel_ids_for_id_values(rel_type, std::slice::from_ref(value)),
            )),
            PropertyFilter::IdNotEq { .. } => None,
            PropertyFilter::IdRange { lower, upper } => {
                if lower.is_none() && upper.is_none() {
                    return None;
                }
                Some(RelationshipScanPruningCandidate::exact(
                    ScanPruningStrategy::IdRange,
                    self.rel_ids_for_id_range(rel_type, lower.as_ref(), upper.as_ref()),
                ))
            }
            PropertyFilter::IdIn { values } => Some(RelationshipScanPruningCandidate::exact(
                if values.is_empty() {
                    ScanPruningStrategy::Empty
                } else {
                    ScanPruningStrategy::IdIn
                },
                self.rel_ids_for_id_values(rel_type, values),
            )),
            PropertyFilter::Eq { property, value } => {
                Some(RelationshipScanPruningCandidate::exact(
                    ScanPruningStrategy::PropertyEq {
                        property: property.clone(),
                    },
                    self.rel_ids_for_property_values(
                        rel_type,
                        property,
                        std::slice::from_ref(value),
                    ),
                ))
            }
            PropertyFilter::NotEq { property, value } => {
                Some(RelationshipScanPruningCandidate::exact(
                    ScanPruningStrategy::PropertyNotEq {
                        property: property.clone(),
                    },
                    self.rel_ids_for_property_not_in_values(
                        rel_type,
                        property,
                        std::slice::from_ref(value),
                    ),
                ))
            }
            PropertyFilter::IsNull { property } => Some(RelationshipScanPruningCandidate::exact(
                ScanPruningStrategy::PropertyMissingOrNull {
                    property: property.clone(),
                },
                self.rel_ids_for_property_missing_or_null(rel_type, property),
            )),
            PropertyFilter::IsNotNull { property } => {
                Some(RelationshipScanPruningCandidate::exact(
                    ScanPruningStrategy::PropertyExists {
                        property: property.clone(),
                    },
                    self.rel_ids_for_property_exists(rel_type, property),
                ))
            }
            PropertyFilter::ListContains { .. }
            | PropertyFilter::ListContainsLower { .. }
            | PropertyFilter::Contains { .. }
            | PropertyFilter::StartsWith { .. }
            | PropertyFilter::EndsWith { .. }
            | PropertyFilter::RegexMatch { .. } => None,
            PropertyFilter::DefaultIfNullOrEq {
                property,
                empty,
                default,
                value,
                negated,
            } => {
                let strategy = if *negated {
                    ScanPruningStrategy::PropertyDefaultIfNullNotEq {
                        property: property.clone(),
                    }
                } else {
                    ScanPruningStrategy::PropertyDefaultIfNullEq {
                        property: property.clone(),
                    }
                };
                let rel_ids = if *negated {
                    self.rel_ids_for_default_if_null_not_eq(
                        rel_type, property, empty, default, value,
                    )
                } else {
                    self.rel_ids_for_default_if_null_eq(rel_type, property, empty, default, value)
                };
                Some(RelationshipScanPruningCandidate::exact(strategy, rel_ids))
            }
            PropertyFilter::In { property, values } => {
                Some(RelationshipScanPruningCandidate::exact(
                    if values.is_empty() {
                        ScanPruningStrategy::Empty
                    } else {
                        ScanPruningStrategy::PropertyIn {
                            property: property.clone(),
                        }
                    },
                    self.rel_ids_for_property_values(rel_type, property, values),
                ))
            }
            PropertyFilter::Range {
                property,
                lower,
                upper,
            } => {
                if lower.is_none() && upper.is_none() {
                    return None;
                }
                Some(RelationshipScanPruningCandidate::exact(
                    ScanPruningStrategy::PropertyRange {
                        property: property.clone(),
                    },
                    self.rel_ids_for_property_range(
                        rel_type,
                        property,
                        lower.as_ref(),
                        upper.as_ref(),
                    ),
                ))
            }
        }
    }

    fn prune_and_relationship_candidates(
        &self,
        rel_type: Option<RelTypeId>,
        filters: &[PropertyFilter],
    ) -> Option<RelationshipScanPruningCandidate> {
        let mut best: Option<RelationshipScanPruningCandidate> = None;
        for filter in filters {
            let Some(candidate) = self.prune_relationship_candidates(rel_type, filter) else {
                continue;
            };
            if candidate.exact_empty {
                return Some(candidate);
            }
            if best
                .as_ref()
                .map(|best| candidate.rel_ids.len() < best.rel_ids.len())
                .unwrap_or(true)
            {
                best = Some(candidate);
            }
        }
        best
    }

    fn prune_or_relationship_candidates(
        &self,
        rel_type: Option<RelTypeId>,
        filters: &[PropertyFilter],
    ) -> Option<RelationshipScanPruningCandidate> {
        if filters.is_empty() {
            return Some(RelationshipScanPruningCandidate {
                strategy: ScanPruningStrategy::Empty,
                rel_ids: BTreeSet::new(),
                exact_empty: true,
            });
        }

        let mut rel_ids = BTreeSet::new();
        for filter in filters {
            let candidate = self.prune_relationship_candidates(rel_type, filter)?;
            rel_ids.extend(candidate.rel_ids);
        }
        Some(RelationshipScanPruningCandidate::exact(
            ScanPruningStrategy::OrUnion,
            rel_ids,
        ))
    }

    fn rel_ids_for_id_values(
        &self,
        rel_type: Option<RelTypeId>,
        values: &[Value],
    ) -> BTreeSet<RelId> {
        values
            .iter()
            .filter_map(|value| match value {
                Value::Int(value) => u64::try_from(*value).ok().map(RelId),
                _ => None,
            })
            .filter(|rel_id| {
                self.relationships
                    .get(rel_id)
                    .map(|relationship| self.relationship_matches_type(relationship, rel_type))
                    .unwrap_or(false)
            })
            .collect()
    }

    fn rel_ids_for_type(&self, rel_type: Option<RelTypeId>) -> BTreeSet<RelId> {
        self.relationships
            .values()
            .filter(|relationship| self.relationship_matches_type(relationship, rel_type))
            .map(|relationship| relationship.id)
            .collect()
    }

    fn rel_ids_for_id_range(
        &self,
        rel_type: Option<RelTypeId>,
        lower: Option<&(Value, bool)>,
        upper: Option<&(Value, bool)>,
    ) -> BTreeSet<RelId> {
        self.relationships
            .keys()
            .copied()
            .filter(|rel_id| range_bounds_match(&Value::Int(rel_id.0 as i64), lower, upper))
            .filter(|rel_id| {
                self.relationships
                    .get(rel_id)
                    .map(|relationship| self.relationship_matches_type(relationship, rel_type))
                    .unwrap_or(false)
            })
            .collect()
    }

    fn rel_ids_for_property_values(
        &self,
        rel_type: Option<RelTypeId>,
        property: &str,
        values: &[Value],
    ) -> BTreeSet<RelId> {
        if values.is_empty() {
            return BTreeSet::new();
        }
        let values = values.iter().collect::<BTreeSet<_>>();
        self.relationship_property_index
            .iter()
            .filter(|((candidate_rel_type, candidate_property, value), _)| {
                rel_type
                    .map(|rel_type| *candidate_rel_type == rel_type)
                    .unwrap_or(true)
                    && candidate_property == property
                    && values.contains(value)
            })
            .flat_map(|(_, rel_ids)| rel_ids.iter().copied())
            .collect()
    }

    fn rel_ids_for_property_not_in_values(
        &self,
        rel_type: Option<RelTypeId>,
        property: &str,
        values: &[Value],
    ) -> BTreeSet<RelId> {
        let values = values.iter().collect::<BTreeSet<_>>();
        self.relationship_property_index
            .iter()
            .filter(|((candidate_rel_type, candidate_property, value), _)| {
                rel_type
                    .map(|rel_type| *candidate_rel_type == rel_type)
                    .unwrap_or(true)
                    && candidate_property == property
                    && !values.contains(value)
            })
            .flat_map(|(_, rel_ids)| rel_ids.iter().copied())
            .collect()
    }

    fn rel_ids_for_property_exists(
        &self,
        rel_type: Option<RelTypeId>,
        property: &str,
    ) -> BTreeSet<RelId> {
        self.relationship_property_index
            .iter()
            .filter(|((candidate_rel_type, candidate_property, value), _)| {
                rel_type
                    .map(|rel_type| *candidate_rel_type == rel_type)
                    .unwrap_or(true)
                    && candidate_property == property
                    && value != &Value::Null
            })
            .flat_map(|(_, rel_ids)| rel_ids.iter().copied())
            .collect()
    }

    fn rel_ids_for_property_missing_or_null(
        &self,
        rel_type: Option<RelTypeId>,
        property: &str,
    ) -> BTreeSet<RelId> {
        let non_null = self.rel_ids_for_property_exists(rel_type, property);
        self.relationships
            .values()
            .filter(|relationship| self.relationship_matches_type(relationship, rel_type))
            .filter(|relationship| !non_null.contains(&relationship.id))
            .map(|relationship| relationship.id)
            .collect()
    }

    fn rel_ids_for_default_if_null_eq(
        &self,
        rel_type: Option<RelTypeId>,
        property: &str,
        empty: &Value,
        default: &Value,
        value: &Value,
    ) -> BTreeSet<RelId> {
        if value == default {
            let mut rel_ids = self.rel_ids_for_property_missing_or_null(rel_type, property);
            let mut values = vec![empty.clone()];
            if value != empty {
                values.push(value.clone());
            }
            rel_ids.extend(self.rel_ids_for_property_values(rel_type, property, &values));
            return rel_ids;
        }

        if value == empty || value == &Value::Null {
            return BTreeSet::new();
        }
        self.rel_ids_for_property_values(rel_type, property, std::slice::from_ref(value))
    }

    fn rel_ids_for_default_if_null_not_eq(
        &self,
        rel_type: Option<RelTypeId>,
        property: &str,
        empty: &Value,
        default: &Value,
        value: &Value,
    ) -> BTreeSet<RelId> {
        let equal_rel_ids =
            self.rel_ids_for_default_if_null_eq(rel_type, property, empty, default, value);
        self.rel_ids_for_type(rel_type)
            .difference(&equal_rel_ids)
            .copied()
            .collect()
    }

    fn rel_ids_for_property_range(
        &self,
        rel_type: Option<RelTypeId>,
        property: &str,
        lower: Option<&(Value, bool)>,
        upper: Option<&(Value, bool)>,
    ) -> BTreeSet<RelId> {
        self.relationship_property_index
            .iter()
            .filter(|((candidate_rel_type, candidate_property, value), _)| {
                rel_type
                    .map(|rel_type| *candidate_rel_type == rel_type)
                    .unwrap_or(true)
                    && candidate_property == property
                    && range_bounds_match(value, lower, upper)
            })
            .flat_map(|(_, rel_ids)| rel_ids.iter().copied())
            .collect()
    }

    pub fn relationship(&self, id: RelId) -> Option<&RelRecord> {
        self.relationships.get(&id)
    }

    pub fn node(&self, id: NodeId) -> Option<&NodeRecord> {
        self.nodes.get(&id)
    }

    /// Selects a checkpoint-published segment manifest for the current graph
    /// snapshot. A missing or stale manifest is an explicit graph-scan fallback,
    /// never permission to use an older physical projection.
    pub fn plan_checkpoint_segment_scan(
        &self,
        manifest: Option<&ScanSegmentManifest>,
        predicate: &ScanPredicate,
    ) -> ScanSegmentAccessPlan {
        manifest.map_or_else(
            || ScanSegmentAccessPlan::fallback(ScanSegmentFallback::NoManifest),
            |manifest| manifest.plan_scan(self.commit_epoch, predicate),
        )
    }

    /// Plans the checkpoint-published Source sidecar for the current graph
    /// snapshot. A missing or stale sidecar is represented as an explicit
    /// fallback before payload consumption starts. Once a streaming read has
    /// started, corruption or canonical disagreement fails closed because a
    /// consumer may already have observed rows.
    pub fn plan_published_source_scan(&self, predicate: &ScanPredicate) -> ScanSegmentAccessPlan {
        self.plan_checkpoint_segment_scan(self.source_scan_manifest.as_ref(), predicate)
    }

    /// Reads only the persisted Source ranges selected by the current
    /// checkpoint-published manifest. This is a physical candidate operator,
    /// not a substitute for query residual evaluation.
    pub fn read_published_source_scan_candidates(
        &self,
        predicate: &ScanPredicate,
        io_depth: NonZeroUsize,
        max_coalesced_bytes: NonZeroU64,
        max_wave_bytes: NonZeroU64,
    ) -> Result<SourceScanCandidateRead> {
        self.collect_published_source_scan_candidates(
            predicate,
            SourceScanCandidateLimits::unbounded(io_depth, max_coalesced_bytes, max_wave_bytes),
            None,
        )
    }

    pub fn read_published_source_scan_candidates_with_context(
        &self,
        predicate: &ScanPredicate,
        io_depth: NonZeroUsize,
        max_coalesced_bytes: NonZeroU64,
        max_wave_bytes: NonZeroU64,
        task_context: &RuntimeTaskContext,
    ) -> Result<SourceScanCandidateRead> {
        self.collect_published_source_scan_candidates(
            predicate,
            SourceScanCandidateLimits::unbounded(io_depth, max_coalesced_bytes, max_wave_bytes),
            Some(task_context),
        )
    }

    #[doc(hidden)]
    pub fn visit_published_source_scan_candidates_bounded(
        &self,
        predicate: &ScanPredicate,
        limits: SourceScanCandidateLimits,
        task_context: Option<&RuntimeTaskContext>,
        consumer: &mut dyn FnMut(SourceScanRow) -> Result<GraphScanControl>,
    ) -> Result<SourceScanCandidateVisit> {
        self.visit_published_source_scan_candidates_internal(
            predicate,
            limits,
            task_context,
            consumer,
        )
    }

    fn collect_published_source_scan_candidates(
        &self,
        predicate: &ScanPredicate,
        limits: SourceScanCandidateLimits,
        task_context: Option<&RuntimeTaskContext>,
    ) -> Result<SourceScanCandidateRead> {
        let mut rows = Vec::new();
        let visit = self.visit_published_source_scan_candidates_internal(
            predicate,
            limits,
            task_context,
            &mut |row| {
                rows.push(row);
                Ok(GraphScanControl::Continue)
            },
        )?;
        match visit {
            SourceScanCandidateVisit::Rows {
                graph_epoch,
                skipped_segment_count,
                report,
                ..
            } => Ok(SourceScanCandidateRead::Rows {
                graph_epoch,
                skipped_segment_count,
                report,
                rows,
            }),
            SourceScanCandidateVisit::Fallback(reason) => {
                Ok(SourceScanCandidateRead::Fallback(reason))
            }
        }
    }

    fn visit_published_source_scan_candidates_internal(
        &self,
        predicate: &ScanPredicate,
        limits: SourceScanCandidateLimits,
        task_context: Option<&RuntimeTaskContext>,
        consumer: &mut dyn FnMut(SourceScanRow) -> Result<GraphScanControl>,
    ) -> Result<SourceScanCandidateVisit> {
        let SourceScanCandidateLimits {
            io_depth,
            max_coalesced_bytes,
            max_wave_bytes,
            max_live_candidate_bytes,
        } = limits;
        let plan = self.plan_published_source_scan(predicate);
        let ScanSegmentAccessPlan::Read(plan) = plan else {
            let ScanSegmentAccessPlan::Fallback(reason) = plan else {
                unreachable!("source scan plan is read or fallback")
            };
            return Ok(SourceScanCandidateVisit::Fallback(reason));
        };
        let Some(durable) = &self.durable else {
            return Ok(SourceScanCandidateVisit::Fallback(
                ScanSegmentFallback::NoManifest,
            ));
        };

        let reader = &durable.source_scan_reader;
        let ranges = plan
            .segments
            .iter()
            .map(|segment| (segment.segment_id, segment.payload_range.clone()))
            .collect::<BTreeMap<_, _>>();
        let manifest = self
            .source_scan_manifest
            .as_ref()
            .expect("read source scan must have a manifest");
        let checksums = manifest
            .segments()
            .iter()
            .map(|segment| (segment.summary.segment_id, segment.payload_range.checksum))
            .collect::<BTreeMap<_, _>>();
        let segment_row_counts = manifest
            .segments()
            .iter()
            .map(|segment| (segment.summary.segment_id, segment.summary.row_count))
            .collect::<BTreeMap<_, _>>();
        let candidate_count = plan.segments.iter().fold(0usize, |count, segment| {
            let segment_count = segment.candidates.as_ref().map_or_else(
                || segment_row_counts[&segment.segment_id],
                hawdb_storage::scan::CandidateCursor::remaining,
            );
            count.saturating_add(usize::try_from(segment_count).unwrap_or(usize::MAX))
        });
        let graph_epoch = plan.graph_epoch;
        let skipped_segment_count = plan.skipped_segment_count;
        let mut candidates = plan
            .segments
            .into_iter()
            .map(|segment| (segment.segment_id, segment.candidates))
            .collect::<BTreeMap<_, _>>();
        let schedule = SegmentReadScheduler::new(io_depth, max_coalesced_bytes)
            .schedule_with_wave_budget(ranges.values().cloned(), max_wave_bytes);
        let mut consume = |payload: SegmentReadPayload| {
            for segment_id in &payload.range.segment_ids {
                let range = ranges.get(segment_id).ok_or_else(|| {
                    HawDBError::StorageIntegrity(format!(
                        "source scan reader returned unknown segment {segment_id}"
                    ))
                })?;
                let start = usize::try_from(range.offset.saturating_sub(payload.range.offset))
                    .map_err(|_| {
                        HawDBError::StorageIntegrity(
                            "source scan payload offset exceeds address space".to_string(),
                        )
                    })?;
                let end = start
                    .checked_add(usize::try_from(range.length.get()).map_err(|_| {
                        HawDBError::StorageIntegrity(
                            "source scan payload length exceeds address space".to_string(),
                        )
                    })?)
                    .ok_or_else(|| {
                        HawDBError::StorageIntegrity(
                            "source scan payload slice overflows".to_string(),
                        )
                    })?;
                let bytes = payload.bytes.get(start..end).ok_or_else(|| {
                    HawDBError::StorageIntegrity(
                        "source scan coalesced payload does not cover a segment".to_string(),
                    )
                })?;
                if checksum_bytes(bytes) != checksums[segment_id] {
                    return Err(HawDBError::StorageIntegrity(format!(
                            "source scan segment {segment_id} checksum changed after manifest validation"
                        )));
                }
                let segment_rows = source_scan::decode_payload(bytes)
                    .map_err(|error| HawDBError::StorageIntegrity(error.to_string()))?;
                let candidate_positions = candidates
                    .remove(segment_id)
                    .flatten()
                    .map(|mut cursor| cursor.next_batch(usize::MAX));
                let live_candidate_bytes = segment_rows
                    .iter()
                    .fold(0usize, |total, row| {
                        total.saturating_add(
                            std::mem::size_of::<SourceScanRow>().saturating_add(
                                usize::try_from(estimated_properties_bytes(&row.properties))
                                    .unwrap_or(usize::MAX),
                            ),
                        )
                    })
                    .saturating_add(candidate_positions.as_ref().map_or(0, |positions| {
                        positions.len().saturating_mul(std::mem::size_of::<u64>())
                    }));
                if max_live_candidate_bytes.is_some_and(|limit| live_candidate_bytes > limit) {
                    return Err(HawDBError::Execution(format!(
                        "SourceSegmentScan decoded segment uses {live_candidate_bytes} bytes, exceeding blocking_operator_bytes {}",
                        max_live_candidate_bytes.unwrap_or_default()
                    )));
                }
                for (row_id, row) in segment_rows.into_iter().enumerate() {
                    if candidate_positions
                        .as_ref()
                        .is_some_and(|positions| positions.binary_search(&(row_id as u64)).is_err())
                    {
                        continue;
                    }
                    if consumer(row)? == GraphScanControl::Stop {
                        return Ok(hawdb_storage::scan::SegmentReadControl::Stop);
                    }
                }
            }
            Ok::<_, HawDBError>(hawdb_storage::scan::SegmentReadControl::Continue)
        };
        let executor = SegmentReadExecutor::new(max_wave_bytes);
        let report = match task_context {
            Some(task_context) => {
                executor.execute_with_context_control(reader, &schedule, task_context, &mut consume)
            }
            None => executor.execute_control(reader, &schedule, &mut consume),
        }
        .map_err(|error| match error {
            SegmentReadExecutionError::Stopped(reason) => {
                HawDBError::Execution(format!("runtime task stopped: {reason}"))
            }
            error => HawDBError::StorageIntegrity(error.to_string()),
        })?;
        Ok(SourceScanCandidateVisit::Rows {
            graph_epoch,
            skipped_segment_count,
            report,
            candidate_count,
        })
    }

    fn adjacency_relationship_ids(
        &self,
        node_id: NodeId,
        rel_type: RelTypeId,
        direction: AdjacencyDirection,
    ) -> Option<&AdjacencyPostingList> {
        match direction {
            AdjacencyDirection::Outgoing => self.outgoing.get(&(node_id, rel_type)),
            AdjacencyDirection::Incoming => self.incoming.get(&(node_id, rel_type)),
        }
    }
}

fn composite_range_requires_full_key(seek: &hawdb_core::access::CompositeRangeSeek) -> bool {
    // Missing suffix properties are absent from the projection, but may still
    // satisfy a prefix range. Keep those requests on a complete fallback.
    seek.equality_prefix.len().checked_add(1) == Some(seek.index_properties.len())
}

fn validate_composite_range_seek(seek: &hawdb_core::access::CompositeRangeSeek) -> Result<()> {
    let prefix_len = seek.equality_prefix.len();
    if prefix_len == 0
        || prefix_len >= seek.index_properties.len()
        || seek
            .equality_prefix
            .iter()
            .map(|(property, _)| property)
            .ne(seek.index_properties.iter().take(prefix_len))
        || seek.index_properties[prefix_len] != seek.range_property
        || (seek.lower.is_none() && seek.upper.is_none())
    {
        return Err(HawDBError::Execution(
            "invalid composite range seek shape".to_string(),
        ));
    }
    Ok(())
}

fn node_matches_composite_range(
    node: &NodeRecord,
    seek: &hawdb_core::access::CompositeRangeSeek,
) -> bool {
    properties_match_composite_range(&node.properties, seek)
}

fn projected_node_matches_composite_range(
    node: &ProjectedNodeRecord,
    seek: &hawdb_core::access::CompositeRangeSeek,
) -> bool {
    properties_match_composite_range(&node.properties, seek)
}

fn properties_match_composite_range(
    properties: &BTreeMap<String, Value>,
    seek: &hawdb_core::access::CompositeRangeSeek,
) -> bool {
    seek.equality_prefix
        .iter()
        .all(|(property, value)| properties.get(property) == Some(value))
        && properties.get(&seek.range_property).is_some_and(|value| {
            range_bounds_match(value, seek.lower.as_ref(), seek.upper.as_ref())
        })
}

fn ordered_relationship_key(
    relationship: &RelRecord,
    direction: AdjacencyDirection,
) -> OrderedAdjacencyEntry {
    let neighbor = match direction {
        AdjacencyDirection::Outgoing => relationship.target,
        AdjacencyDirection::Incoming => relationship.source,
    };
    OrderedAdjacencyEntry {
        neighbor_id: neighbor,
        relationship_id: relationship.id,
    }
}

fn grow_relationship_key_allocation(
    allocation: &mut dyn crate::read_view::GraphReadAllocation,
    bytes: usize,
) -> Result<()> {
    let required = allocation
        .bytes()
        .checked_add(bytes)
        .ok_or_else(|| HawDBError::Execution("relationship key allocation size overflow".into()))?;
    allocation.retain_state()?;
    allocation.grow(bytes)?;
    if allocation.bytes() < required {
        return Err(HawDBError::Execution(
            "relationship key admission returned an insufficient allocation permit".into(),
        ));
    }
    Ok(())
}

fn relationship_probe_allocation_bytes(filter: &PropertyFilter) -> usize {
    let count = match filter {
        PropertyFilter::In { values, .. } => values.len(),
        PropertyFilter::And(filters) | PropertyFilter::Or(filters) => {
            return filters.iter().fold(128usize, |total, child| {
                total.saturating_add(relationship_probe_allocation_bytes(child))
            });
        }
        _ => 1,
    };
    128usize.saturating_add(count.saturating_mul(std::mem::size_of::<&Value>()))
}

fn emit_admitted_relationship(
    store: &GraphStore,
    selection: (NodeId, Option<RelTypeId>, AdjacencyDirection),
    entry: OrderedAdjacencyEntry,
    admit: &mut ControlledGraphReadAllocator<'_>,
    consumer: &mut impl FnMut(AdmittedRelationshipRecord) -> Result<GraphScanControl>,
) -> Result<GraphScanControl> {
    let relationship = match store.relationship_with_allocation(entry.relationship_id, admit)? {
        AdmittedRelationshipRead::Missing => {
            return Err(HawDBError::StorageIntegrity(format!(
                "ordered adjacency references missing relationship {}",
                entry.relationship_id.0,
            )))
        }
        AdmittedRelationshipRead::Stopped => return Ok(GraphScanControl::Stop),
        AdmittedRelationshipRead::Relationship(relationship) => relationship,
    };
    let (record, allocation) = relationship.into_parts();
    let (node_id, rel_type, direction) = selection;
    let (endpoint, neighbor) = match direction {
        AdjacencyDirection::Outgoing => (record.source, record.target),
        AdjacencyDirection::Incoming => (record.target, record.source),
    };
    if endpoint != node_id
        || neighbor != entry.neighbor_id
        || rel_type.is_some_and(|expected| expected != record.rel_type)
    {
        return Err(HawDBError::StorageIntegrity(
            "ordered relationship does not match its adjacency key".into(),
        ));
    }
    consumer(AdmittedRelationshipRecord::new(record, allocation))
}

fn emit_admitted_live_before(
    store: &GraphStore,
    selection: (NodeId, Option<RelTypeId>, AdjacencyDirection),
    entries: &mut std::iter::Peekable<impl Iterator<Item = OrderedAdjacencyEntry>>,
    before: Option<OrderedAdjacencyEntry>,
    admit: &mut ControlledGraphReadAllocator<'_>,
    consumer: &mut impl FnMut(AdmittedRelationshipRecord) -> Result<GraphScanControl>,
) -> Result<GraphScanControl> {
    while let Some(entry) = entries.peek().copied() {
        if before.is_some_and(|before| entry >= before) {
            break;
        }
        entries.next();
        if emit_admitted_relationship(store, selection, entry, admit, consumer)?
            == GraphScanControl::Stop
        {
            return Ok(GraphScanControl::Stop);
        }
    }
    Ok(GraphScanControl::Continue)
}

fn push_admitted_adjacency_key(
    entries: &mut Vec<OrderedAdjacencyEntry>,
    entry: OrderedAdjacencyEntry,
    memory_budget_bytes: usize,
    charge: &mut impl FnMut(usize) -> Result<()>,
) -> Result<()> {
    if entries.len() == entries.capacity() {
        let size = std::mem::size_of::<OrderedAdjacencyEntry>();
        let maximum = memory_budget_bytes / size;
        let capacity = entries.capacity().saturating_mul(2).max(4).min(maximum);
        if capacity <= entries.len() {
            return Err(HawDBError::Execution(
                "ordered adjacency keys exceed blocking_operator_bytes".into(),
            ));
        }
        charge((capacity - entries.capacity()).saturating_mul(size))?;
        entries.reserve_exact(capacity - entries.len());
    }
    entries.push(entry);
    Ok(())
}

fn push_compact_adjacency_key(
    entries: &mut Vec<OrderedAdjacencyEntry>,
    entry: OrderedAdjacencyEntry,
    memory_budget_bytes: usize,
    charge_key_bytes: &mut impl FnMut(usize) -> Result<()>,
) -> Result<()> {
    let entry_bytes = std::mem::size_of::<OrderedAdjacencyEntry>();
    let required_bytes = entries.len().saturating_add(1).saturating_mul(entry_bytes);
    if required_bytes > memory_budget_bytes {
        return Err(HawDBError::Execution(format!(
            "ordered adjacency keys use {required_bytes} bytes, exceeding blocking_operator_bytes {memory_budget_bytes}"
        )));
    }
    charge_key_bytes(entry_bytes)?;
    entries.push(entry);
    Ok(())
}

fn emit_compact_adjacency_before(
    store: &GraphStore,
    entries: &[OrderedAdjacencyEntry],
    index: &mut usize,
    before: Option<OrderedAdjacencyEntry>,
    consumer: &mut impl FnMut(RelRecord) -> Result<GraphScanControl>,
) -> Result<GraphScanControl> {
    while let Some(entry) = entries.get(*index).copied() {
        if before.is_some_and(|before| entry >= before) {
            break;
        }
        *index = index.saturating_add(1);
        let Some(relationship) = store.relationship_owned(entry.relationship_id)? else {
            return Err(HawDBError::StorageIntegrity(format!(
                "ordered adjacency references missing relationship {}",
                entry.relationship_id.0
            )));
        };
        if consumer(relationship)? == GraphScanControl::Stop {
            return Ok(GraphScanControl::Stop);
        }
    }
    Ok(GraphScanControl::Continue)
}

fn emit_live_adjacency_before(
    store: &GraphStore,
    entries: &mut std::iter::Peekable<impl Iterator<Item = OrderedAdjacencyEntry>>,
    before: Option<OrderedAdjacencyEntry>,
    consumer: &mut impl FnMut(RelRecord) -> Result<GraphScanControl>,
) -> Result<GraphScanControl> {
    while let Some(entry) = entries.peek().copied() {
        if before.is_some_and(|before| entry >= before) {
            break;
        }
        entries.next();
        let Some(relationship) = store.relationships.get(&entry.relationship_id) else {
            return Err(HawDBError::StorageIntegrity(format!(
                "live ordered adjacency references missing relationship {}",
                entry.relationship_id.0
            )));
        };
        if consumer(relationship.clone())? == GraphScanControl::Stop {
            return Ok(GraphScanControl::Stop);
        }
    }
    Ok(GraphScanControl::Continue)
}

enum RelationshipProjectionProbe<'a> {
    Equality {
        rel_type: RelTypeId,
        property: &'a str,
        values: Vec<&'a Value>,
        estimated_entries: u64,
        estimate_report: hawdb_storage::property_projection::PersistentPropertyProjectionReadReport,
    },
    Range {
        rel_type: RelTypeId,
        property: &'a str,
        lower: Option<&'a (Value, bool)>,
        upper: Option<&'a (Value, bool)>,
        estimated_entries: u64,
        estimate_report: hawdb_storage::property_projection::PersistentPropertyProjectionReadReport,
    },
}

impl RelationshipProjectionProbe<'_> {
    fn estimated_entries(&self) -> u64 {
        match self {
            Self::Equality {
                estimated_entries, ..
            }
            | Self::Range {
                estimated_entries, ..
            } => *estimated_entries,
        }
    }

    fn estimate_report(
        &self,
    ) -> hawdb_storage::property_projection::PersistentPropertyProjectionReadReport {
        match self {
            Self::Equality {
                estimate_report, ..
            }
            | Self::Range {
                estimate_report, ..
            } => *estimate_report,
        }
    }

    fn replace_estimate_report(
        &mut self,
        report: hawdb_storage::property_projection::PersistentPropertyProjectionReadReport,
    ) {
        match self {
            Self::Equality {
                estimate_report, ..
            }
            | Self::Range {
                estimate_report, ..
            } => *estimate_report = report,
        }
    }

    fn strategy(&self) -> ScanPruningStrategy {
        match self {
            Self::Equality {
                property, values, ..
            } if values.len() == 1 => ScanPruningStrategy::PropertyEq {
                property: (*property).to_string(),
            },
            Self::Equality { property, .. } => ScanPruningStrategy::PropertyIn {
                property: (*property).to_string(),
            },
            Self::Range { property, .. } => ScanPruningStrategy::PropertyRange {
                property: (*property).to_string(),
            },
        }
    }

    fn index_class(&self) -> PersistentGraphIndexClass {
        match self {
            Self::Equality { .. } => PersistentGraphIndexClass::RelationshipEquality,
            Self::Range { .. } => PersistentGraphIndexClass::RelationshipRange,
        }
    }

    fn matches(&self, relationship: &RelRecord) -> bool {
        match self {
            Self::Equality {
                property, values, ..
            } => relationship
                .properties
                .get(*property)
                .is_some_and(|actual| values.contains(&actual)),
            Self::Range {
                property,
                lower,
                upper,
                ..
            } => relationship
                .properties
                .get(*property)
                .is_some_and(|value| range_bounds_match(value, *lower, *upper)),
        }
    }

    fn scan(
        &self,
        projection: &PersistentPropertyProjectionReader,
        mut consumer: impl FnMut(
            RelId,
        ) -> std::result::Result<
            CanonicalScanControl,
            PersistentPropertyProjectionError,
        >,
    ) -> std::result::Result<
        (
            hawdb_storage::property_projection::PersistentPropertyProjectionReadReport,
            CanonicalScanControl,
        ),
        PersistentPropertyProjectionError,
    > {
        match self {
            Self::Equality {
                rel_type,
                property,
                values,
                ..
            } => {
                let mut total = self.estimate_report();
                for value in values {
                    let (report, control) = projection.scan_relationship_equality_candidates(
                        *rel_type,
                        property,
                        value,
                        &mut consumer,
                    )?;
                    accumulate_property_projection_report(&mut total, report);
                    if control == CanonicalScanControl::Stop {
                        return Ok((total, control));
                    }
                }
                Ok((total, CanonicalScanControl::Continue))
            }
            Self::Range {
                rel_type,
                property,
                lower,
                upper,
                ..
            } => {
                let mut total = self.estimate_report();
                let (report, control) = projection.scan_relationship_range_candidates(
                    *rel_type, property, *lower, *upper, consumer,
                )?;
                accumulate_property_projection_report(&mut total, report);
                Ok((total, control))
            }
        }
    }
}

fn accumulate_property_projection_report(
    total: &mut hawdb_storage::property_projection::PersistentPropertyProjectionReadReport,
    report: hawdb_storage::property_projection::PersistentPropertyProjectionReadReport,
) {
    total.descriptor_pages_visited = total
        .descriptor_pages_visited
        .saturating_add(report.descriptor_pages_visited);
    total.descriptor_page_bytes_decoded = total
        .descriptor_page_bytes_decoded
        .saturating_add(report.descriptor_page_bytes_decoded);
    total.descriptor_storage_bytes_read = total
        .descriptor_storage_bytes_read
        .saturating_add(report.descriptor_storage_bytes_read);
    total.descriptors_examined = total
        .descriptors_examined
        .saturating_add(report.descriptors_examined);
    total.descriptor_cache_hits = total
        .descriptor_cache_hits
        .saturating_add(report.descriptor_cache_hits);
    total.descriptor_cache_misses = total
        .descriptor_cache_misses
        .saturating_add(report.descriptor_cache_misses);
    total.descriptor_cache_admission_rejections = total
        .descriptor_cache_admission_rejections
        .saturating_add(report.descriptor_cache_admission_rejections);
    total.blocks_considered = total
        .blocks_considered
        .saturating_add(report.blocks_considered);
    total.blocks_pruned = total.blocks_pruned.saturating_add(report.blocks_pruned);
    total.blocks_read = total.blocks_read.saturating_add(report.blocks_read);
    total.bytes_read = total.bytes_read.saturating_add(report.bytes_read);
    total.cache_hits = total.cache_hits.saturating_add(report.cache_hits);
    total.cache_misses = total.cache_misses.saturating_add(report.cache_misses);
    total.entries_decoded = total.entries_decoded.saturating_add(report.entries_decoded);
    total.candidates_returned = total
        .candidates_returned
        .saturating_add(report.candidates_returned);
}

fn relationship_projection_probe<'a>(
    projection: &PersistentPropertyProjectionReader,
    rel_type: RelTypeId,
    filter: &'a PropertyFilter,
) -> std::result::Result<Option<RelationshipProjectionProbe<'a>>, PersistentPropertyProjectionError>
{
    match filter {
        PropertyFilter::And(filters) => {
            let mut best = None;
            let mut selection_report =
                hawdb_storage::property_projection::PersistentPropertyProjectionReadReport::default(
                );
            for filter in filters {
                let Some(candidate) = relationship_projection_probe(projection, rel_type, filter)?
                else {
                    continue;
                };
                accumulate_property_projection_report(
                    &mut selection_report,
                    candidate.estimate_report(),
                );
                if best
                    .as_ref()
                    .is_none_or(|current: &RelationshipProjectionProbe<'_>| {
                        candidate.estimated_entries() < current.estimated_entries()
                    })
                {
                    best = Some(candidate);
                }
            }
            if let Some(best) = &mut best {
                best.replace_estimate_report(selection_report);
            }
            Ok(best)
        }
        PropertyFilter::Eq { property, value } => {
            if !projection.manifest().supports_relationship(
                rel_type,
                property,
                PersistentPropertyProjectionKind::RelationshipEquality,
            ) {
                return Ok(None);
            }
            let (estimated_entries, estimate_report) =
                projection.estimate_relationship_equality_entries(rel_type, property, value)?;
            Ok(Some(RelationshipProjectionProbe::Equality {
                rel_type,
                property,
                values: vec![value],
                estimated_entries,
                estimate_report,
            }))
        }
        PropertyFilter::In { property, values } => {
            if !projection.manifest().supports_relationship(
                rel_type,
                property,
                PersistentPropertyProjectionKind::RelationshipEquality,
            ) {
                return Ok(None);
            }
            let values = values
                .iter()
                .collect::<BTreeSet<_>>()
                .into_iter()
                .collect::<Vec<_>>();
            let mut estimated_entries = 0u64;
            let mut estimate_report =
                hawdb_storage::property_projection::PersistentPropertyProjectionReadReport::default(
                );
            for value in &values {
                let (entries, report) =
                    projection.estimate_relationship_equality_entries(rel_type, property, value)?;
                estimated_entries = estimated_entries.saturating_add(entries);
                accumulate_property_projection_report(&mut estimate_report, report);
            }
            Ok(Some(RelationshipProjectionProbe::Equality {
                rel_type,
                property,
                values,
                estimated_entries,
                estimate_report,
            }))
        }
        PropertyFilter::Range {
            property,
            lower,
            upper,
        } if lower.is_some() || upper.is_some() => {
            if !projection.manifest().supports_relationship(
                rel_type,
                property,
                PersistentPropertyProjectionKind::RelationshipRange,
            ) {
                return Ok(None);
            }
            let (estimated_entries, estimate_report) = projection
                .estimate_relationship_range_entries(
                    rel_type,
                    property,
                    lower.as_ref(),
                    upper.as_ref(),
                )?;
            Ok(Some(RelationshipProjectionProbe::Range {
                rel_type,
                property,
                lower: lower.as_ref(),
                upper: upper.as_ref(),
                estimated_entries,
                estimate_report,
            }))
        }
        _ => Ok(None),
    }
}

fn relationship_matches_endpoint(
    relationship: &RelRecord,
    node_id: NodeId,
    direction: AdjacencyDirection,
) -> bool {
    match direction {
        AdjacencyDirection::Outgoing => relationship.source == node_id,
        AdjacencyDirection::Incoming => relationship.target == node_id,
    }
}

#[cfg(all(test, not(target_arch = "wasm32")))]
#[path = "graph_read/checkpoint_scan_memory_tests.rs"]
mod checkpoint_scan_memory_tests;