icydb-core 0.285.0

//! Module: executor::stream::access::physical
//! Responsibility: lower executable access-path payloads into physical key streams.
//! Does not own: planner eligibility decisions or post-access semantics.
//! Boundary: physical key resolution through primary/index scan adapters.

use crate::{
    db::{
        access::{ExecutionPathPayload, IndexShapeDetails},
        cursor::{CursorBoundary, CursorBoundarySlot, IndexScanContinuationInput},
        data::{
            DecodedDataStoreKey, RawDataStoreKey, RawRow, StoreVisit,
            primary_key_value_from_structural_values,
        },
        direction::Direction,
        executor::{
            ACCESS_SCAN_CHUNK_ENTRIES, AccessStreamExecutionPolicy, IndexLeafOrderPolicy,
            IndexScan, LoweredIndexPrefixSpec, LoweredIndexRangeSpec, LoweredKey, OrderedKeyStream,
            OrderedKeyStreamBox, PrefixSetExecutionShape, PrefixSetMergeSafety, PrimaryScan,
            active_lowered_index_prefix_specs, apply_index_scan_chunk_progress,
            branch_stream_chunk_entries, expand_index_prefix_family_with_exact_child_prefixes,
            index_predicate_rejects_prefix_components, index_stream_chunk_entries_for_remaining,
            index_stream_output_limit_for_chunk, lowered_index_prefix_liveness,
            ordered_key_stream_from_materialized_keys,
            pipeline::contracts::AccessScanContinuationInput,
            production_scalar_page_access_entry_limit,
            route::{
                IndexPrefixChildExpansionBudget, IndexPrefixChildExpansionHint,
                primary_scan_fetch_hint_shape_supported,
            },
            stream::key::KeyOrderComparator,
            traversal::IndexRangeTraversalContract,
        },
        index::{
            IndexId, IndexKey, IndexKeyKind, RawIndexStoreKey, predicate::IndexPredicateExecution,
            resume_prefix_bounds_for_continuation,
        },
        key_taxonomy::RawDataStoreKeyRange,
        registry::StoreHandle,
        resource::budget::{
            ExecutionConstructionBudget, charge_current_execution_budget, charge_sort_work,
        },
    },
    error::InternalError,
    types::EntityTag,
    value::Value,
};
use icydb_diagnostic_code::DiagnosticExecutionBudgetResource;
use std::{cell::Cell, mem::size_of, ops::Bound, vec::IntoIter};

///
/// KeyOrderState
///
/// Explicit ordering state for key vectors produced by one access-path resolver.
/// This keeps normalization behavior local and avoids implicit path-shape proxies.
///

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum KeyOrderState {
    FinalOrder,
    AscendingSorted,
    Unordered,
}

enum PhysicalKeyResolution {
    Stream(Box<OrderedKeyStreamBox>),
    Materialized {
        candidates: Vec<DecodedDataStoreKey>,
        key_order_state: KeyOrderState,
    },
}

#[derive(Clone, Copy)]
enum PrefixMergeResumePolicy {
    None,
    PrimaryKeySuffix,
}

impl PrefixMergeResumePolicy {
    const fn from_index_leaf_order_policy(policy: IndexLeafOrderPolicy) -> Self {
        match policy {
            IndexLeafOrderPolicy::PreservePhysicalLeaf => Self::PrimaryKeySuffix,
            IndexLeafOrderPolicy::CanonicalKey | IndexLeafOrderPolicy::PreservePrefixBranch => {
                Self::None
            }
        }
    }
}

///
/// MergedIndexPrefixStreamSpec
///
/// Runtime-local contract for one family of exact secondary-index prefix
/// streams that may be merged by decoded primary key.
///

#[derive(Clone, Copy)]
struct MergedIndexPrefixStreamSpec<'a> {
    index: &'a IndexShapeDetails,
    index_prefix_specs: &'a [LoweredIndexPrefixSpec],
    continuation: AccessScanContinuationInput<'a>,
    index_fetch_hint: Option<usize>,
    resume_policy: PrefixMergeResumePolicy,
}

impl<'a> MergedIndexPrefixStreamSpec<'a> {
    const fn new(
        index: &'a IndexShapeDetails,
        index_prefix_specs: &'a [LoweredIndexPrefixSpec],
        continuation: AccessScanContinuationInput<'a>,
        index_fetch_hint: Option<usize>,
        resume_policy: PrefixMergeResumePolicy,
    ) -> Self {
        Self {
            index,
            index_prefix_specs,
            continuation,
            index_fetch_hint,
            resume_policy,
        }
    }

    fn resume_anchor_for(
        self,
        entity_tag: EntityTag,
        spec: &LoweredIndexPrefixSpec,
    ) -> Result<Option<RawIndexStoreKey>, InternalError> {
        if let Some(anchor) = self.continuation.index_scan_continuation().anchor() {
            return Ok(Some(anchor.clone()));
        }
        match self.resume_policy {
            PrefixMergeResumePolicy::None => Ok(None),
            PrefixMergeResumePolicy::PrimaryKeySuffix => self
                .continuation
                .primary_key_boundary()
                .map(|boundary| {
                    primary_key_suffix_resume_anchor_for_prefix(
                        entity_tag, self.index, spec, boundary,
                    )
                })
                .transpose(),
        }
    }
}

///
/// StructuralPhysicalStreamRequest
///
/// StructuralPhysicalStreamRequest is the generic-free physical access request
/// used by structural traversal and erased runtime execution.
/// It carries direct store/index authority plus one entity tag so physical scan
/// leaves do not need typed `Context<'_, E>` recovery.
///

pub(super) struct StructuralPhysicalStreamRequest<'a> {
    pub(super) store: StoreHandle,
    pub(super) entity_tag: EntityTag,
    pub(super) index_prefix_specs: &'a [LoweredIndexPrefixSpec],
    pub(super) index_range_spec: Option<&'a LoweredIndexRangeSpec>,
    pub(super) continuation: AccessScanContinuationInput<'a>,
    pub(super) execution_policy: AccessStreamExecutionPolicy,
    pub(super) index_predicate_execution: Option<IndexPredicateExecution<'a>>,
    pub(super) index_prefix_child_expansion: Option<IndexPrefixChildExpansionHint>,
}

///
/// PhysicalStreamBindings
///
/// Structural physical-resolution inputs shared by all entity-specific
/// resolvers.
/// This excludes the typed executor context so the outer dispatch body can
/// collapse to one key-shape-specific implementation.
///

#[derive(Clone, Copy)]
struct PhysicalStreamBindings<'a> {
    index_prefix_specs: &'a [LoweredIndexPrefixSpec],
    index_range_spec: Option<&'a LoweredIndexRangeSpec>,
    continuation: AccessScanContinuationInput<'a>,
    execution_policy: AccessStreamExecutionPolicy,
    index_predicate_execution: Option<IndexPredicateExecution<'a>>,
    index_prefix_child_expansion: Option<IndexPrefixChildExpansionHint>,
}

///
/// KeyAccessRuntime
///
/// KeyAccessRuntime binds one recovered typed context to the
/// structural planner-key boundary used by structural fast-path traversal.
/// It recovers typed primary-key values only inside physical leaf resolution.
///

struct KeyAccessRuntime {
    store: StoreHandle,
    entity_tag: EntityTag,
}

impl KeyAccessRuntime {
    const fn new(store: StoreHandle, entity_tag: EntityTag) -> Self {
        Self { store, entity_tag }
    }

    // Resolve one direct primary-key lookup into its canonical ordered output.
    fn resolve_by_key(
        &self,
        key: &Value,
    ) -> Result<(Vec<DecodedDataStoreKey>, KeyOrderState), InternalError> {
        charge_current_execution_budget(
            DiagnosticExecutionBudgetResource::KeyIndexEntriesVisited,
            1,
        )?;
        Ok((
            vec![DecodedDataStoreKey::try_from_structural_key(
                self.entity_tag,
                key,
            )?],
            KeyOrderState::FinalOrder,
        ))
    }

    // Resolve one multi-key primary lookup into canonical ascending key order.
    fn resolve_by_keys(
        &self,
        keys: &[Value],
    ) -> Result<(Vec<DecodedDataStoreKey>, KeyOrderState), InternalError> {
        let mut data_keys = Vec::with_capacity(keys.len());
        for key in keys {
            data_keys.push(DecodedDataStoreKey::try_from_structural_key(
                self.entity_tag,
                key,
            )?);
        }
        data_keys.sort_unstable();
        data_keys.dedup();
        charge_current_execution_budget(
            DiagnosticExecutionBudgetResource::KeyIndexEntriesVisited,
            u64::try_from(data_keys.len()).unwrap_or(u64::MAX),
        )?;

        Ok((data_keys, KeyOrderState::AscendingSorted))
    }

    // Resolve one primary-key range scan as a dynamic ordered stream.
    fn resolve_key_range_stream(
        &self,
        start: Option<Value>,
        end: Option<Value>,
        continuation: AccessScanContinuationInput<'_>,
        primary_scan_fetch_hint: Option<usize>,
    ) -> Result<OrderedKeyStreamBox, InternalError> {
        let start = start
            .as_ref()
            .map(|start| DecodedDataStoreKey::try_from_structural_key(self.entity_tag, start))
            .transpose()?;
        let end = end
            .as_ref()
            .map(|end| DecodedDataStoreKey::try_from_structural_key(self.entity_tag, end))
            .transpose()?;
        let mut stream = PrimaryRangeKeyStream::new_bounds(
            self.store,
            self.entity_tag,
            start,
            end,
            continuation.direction(),
            primary_scan_fetch_hint,
        )?;
        if let Some(boundary) = continuation.primary_key_boundary() {
            stream
                .resume_strictly_after(primary_key_boundary_data_key(self.entity_tag, boundary)?)?;
        }

        Ok(OrderedKeyStreamBox::primary_range(stream))
    }

    // Resolve one full primary-key scan as a dynamic ordered stream.
    fn resolve_full_scan_stream(
        &self,
        continuation: AccessScanContinuationInput<'_>,
        primary_scan_fetch_hint: Option<usize>,
    ) -> Result<OrderedKeyStreamBox, InternalError> {
        let mut stream = PrimaryRangeKeyStream::new_full_scan(
            self.store,
            self.entity_tag,
            continuation.direction(),
            primary_scan_fetch_hint,
        )?;
        if let Some(boundary) = continuation.primary_key_boundary() {
            stream
                .resume_strictly_after(primary_key_boundary_data_key(self.entity_tag, boundary)?)?;
        }

        Ok(OrderedKeyStreamBox::primary_range(stream))
    }

    // Resolve one single-prefix secondary-index scan.
    fn resolve_index_prefix(
        &self,
        index_prefix_specs: &[LoweredIndexPrefixSpec],
        continuation: IndexScanContinuationInput<'_>,
        index_fetch_hint: Option<usize>,
        index_predicate_execution: Option<IndexPredicateExecution<'_>>,
    ) -> Result<(Vec<DecodedDataStoreKey>, KeyOrderState), InternalError> {
        let [spec] = index_prefix_specs else {
            return Err(InternalError::query_executor_invariant());
        };
        let key_order_state = if index_fetch_hint.is_some() {
            KeyOrderState::FinalOrder
        } else {
            KeyOrderState::Unordered
        };
        if index_predicate_rejects_prefix_components(
            spec.prefix_components(),
            index_predicate_execution,
        ) {
            return Ok((Vec::new(), key_order_state));
        }
        if !lowered_index_prefix_liveness(self.store, spec).should_scan() {
            return Ok((Vec::new(), key_order_state));
        }

        let keys = IndexScan::prefix_structural(
            self.store,
            self.entity_tag,
            spec,
            continuation,
            index_fetch_hint.unwrap_or(usize::MAX),
            index_predicate_execution,
        )?;
        charge_materialized_secondary_index_keys(keys.as_slice(), keys.capacity())?;

        Ok((keys, key_order_state))
    }

    // Resolve one single-prefix secondary-index scan as a dynamic ordered stream.
    fn resolve_index_prefix_stream(
        &self,
        index: &IndexShapeDetails,
        index_prefix_specs: &[LoweredIndexPrefixSpec],
        continuation: AccessScanContinuationInput<'_>,
        index_fetch_hint: Option<usize>,
        index_leaf_order_policy: IndexLeafOrderPolicy,
    ) -> Result<OrderedKeyStreamBox, InternalError> {
        self.resolve_index_prefix_family_stream(
            index,
            index_prefix_specs,
            1,
            continuation,
            index_fetch_hint,
            PrefixMergeResumePolicy::from_index_leaf_order_policy(index_leaf_order_policy),
        )
    }

    // Resolve a branch-aware composite prefix scan as lazily merged dynamic
    // prefix streams. Each branch is internally ordered by the primary-key
    // suffix after the fixed index prefix, and the merge stream suppresses
    // duplicate decoded primary keys defensively.
    fn resolve_index_branch_set_stream(
        &self,
        index: &IndexShapeDetails,
        index_prefix_specs: &[LoweredIndexPrefixSpec],
        branch_count: usize,
        continuation: AccessScanContinuationInput<'_>,
        index_fetch_hint: Option<usize>,
    ) -> Result<OrderedKeyStreamBox, InternalError> {
        self.resolve_index_prefix_family_stream(
            index,
            index_prefix_specs,
            branch_count,
            continuation,
            index_fetch_hint,
            PrefixMergeResumePolicy::PrimaryKeySuffix,
        )
    }

    // Resolve one multi-lookup secondary-index scan and normalize duplicates.
    fn resolve_index_multi_lookup(
        &self,
        index_prefix_specs: &[LoweredIndexPrefixSpec],
        value_count: usize,
        continuation: IndexScanContinuationInput<'_>,
        index_leaf_order_policy: IndexLeafOrderPolicy,
        index_fetch_hint: Option<usize>,
        index_predicate_execution: Option<IndexPredicateExecution<'_>>,
    ) -> Result<(Vec<DecodedDataStoreKey>, KeyOrderState), InternalError> {
        validate_index_prefix_count(index_prefix_specs, value_count)?;

        let per_prefix_limit = index_fetch_hint.unwrap_or(usize::MAX);
        let mut active_specs = active_lowered_index_prefix_specs(
            Some(self.store),
            index_prefix_specs,
            index_predicate_execution,
        );
        let preserve_branch_order = index_leaf_order_policy.preserves_prefix_branch_order();
        if preserve_branch_order {
            sort_lowered_index_prefix_specs_by_raw_lower_key(&mut active_specs)?;
            if matches!(continuation.direction(), Direction::Desc) {
                active_specs.reverse();
            }
        }
        let key_capacity = index_fetch_hint.map_or(0, |hint| {
            hint.saturating_mul(active_specs.len())
                .min(ACCESS_SCAN_CHUNK_ENTRIES)
        });
        let mut keys = new_materialized_secondary_index_key_vector(key_capacity)?;
        for spec in active_specs {
            let child = IndexScan::prefix_structural(
                self.store,
                self.entity_tag,
                spec,
                continuation,
                per_prefix_limit,
                index_predicate_execution,
            )?;
            charge_materialized_secondary_index_keys(child.as_slice(), child.capacity())?;
            reserve_materialized_secondary_index_key_capacity(&mut keys, child.len())?;
            keys.extend(child);
        }
        // Distinct canonical IN prefixes are disjoint. Preserve their index
        // order for secondary paging even when a predicate requires materialization.
        if preserve_branch_order {
            return Ok((keys, KeyOrderState::FinalOrder));
        }
        charge_sort_work::<DecodedDataStoreKey>(keys.len())?;
        keys.sort_unstable();
        charge_materialized_key_dedup_comparisons(keys.len())?;
        keys.dedup();

        Ok((keys, KeyOrderState::AscendingSorted))
    }

    // Resolve one multi-lookup secondary-index scan as lazily merged prefix streams.
    fn resolve_index_multi_lookup_stream(
        &self,
        index: &IndexShapeDetails,
        index_prefix_specs: &[LoweredIndexPrefixSpec],
        value_count: usize,
        continuation: AccessScanContinuationInput<'_>,
        index_fetch_hint: Option<usize>,
        index_leaf_order_policy: IndexLeafOrderPolicy,
    ) -> Result<OrderedKeyStreamBox, InternalError> {
        validate_index_prefix_count(index_prefix_specs, value_count)?;
        if index_leaf_order_policy.preserves_prefix_branch_order() {
            return self.resolve_branch_ordered_index_prefix_streams(
                MergedIndexPrefixStreamSpec::new(
                    index,
                    index_prefix_specs,
                    continuation,
                    index_fetch_hint,
                    PrefixMergeResumePolicy::None,
                ),
            );
        }

        self.resolve_index_prefix_family_stream(
            index,
            index_prefix_specs,
            value_count,
            continuation,
            index_fetch_hint,
            PrefixMergeResumePolicy::from_index_leaf_order_policy(index_leaf_order_policy),
        )
    }

    fn resolve_index_prefix_family_stream(
        &self,
        index: &IndexShapeDetails,
        index_prefix_specs: &[LoweredIndexPrefixSpec],
        expected_prefix_count: usize,
        continuation: AccessScanContinuationInput<'_>,
        index_fetch_hint: Option<usize>,
        resume_policy: PrefixMergeResumePolicy,
    ) -> Result<OrderedKeyStreamBox, InternalError> {
        validate_index_prefix_count(index_prefix_specs, expected_prefix_count)?;
        let spec = MergedIndexPrefixStreamSpec::new(
            index,
            index_prefix_specs,
            continuation,
            index_fetch_hint,
            resume_policy,
        );

        self.resolve_merged_index_prefix_streams(spec)
    }

    fn expanded_index_multi_lookup_stream(
        &self,
        index: &IndexShapeDetails,
        index_prefix_specs: &[LoweredIndexPrefixSpec],
        value_count: usize,
        continuation: AccessScanContinuationInput<'_>,
        index_fetch_hint: Option<usize>,
        expansion: IndexPrefixChildExpansionHint,
    ) -> Result<Option<OrderedKeyStreamBox>, InternalError> {
        validate_index_prefix_count(index_prefix_specs, value_count)?;

        let Some(expanded_family) = expand_index_prefix_family_with_exact_child_prefixes(
            self.store,
            self.entity_tag,
            index,
            index_prefix_specs,
            expansion,
        )?
        else {
            return Ok(None);
        };
        if expanded_family.specs().is_empty() {
            return Ok(Some(ordered_key_stream_from_materialized_keys(Vec::new())));
        }

        let request = MergedIndexPrefixStreamSpec::new(
            expanded_family.index(),
            expanded_family.specs(),
            continuation,
            index_fetch_hint,
            PrefixMergeResumePolicy::PrimaryKeySuffix,
        );

        self.resolve_merged_index_prefix_streams(request).map(Some)
    }

    fn resolve_merged_index_prefix_streams(
        &self,
        request: MergedIndexPrefixStreamSpec<'_>,
    ) -> Result<OrderedKeyStreamBox, InternalError> {
        if request.index_prefix_specs.is_empty() {
            return Ok(ordered_key_stream_from_materialized_keys(Vec::new()));
        }

        let active_specs =
            active_lowered_index_prefix_specs(Some(self.store), request.index_prefix_specs, None);
        charge_prefix_family_construction(active_specs.as_slice())?;
        match PrefixSetExecutionShape::from_active_prefixes(
            active_specs,
            PrefixSetMergeSafety::OrderedMergeSafe,
        ) {
            PrefixSetExecutionShape::Empty => {
                Ok(ordered_key_stream_from_materialized_keys(Vec::new()))
            }
            PrefixSetExecutionShape::Single(spec) => self.index_prefix_stream(request, spec, 1),
            PrefixSetExecutionShape::OrderedMerge(active_specs) => {
                let branch_count = active_specs.len();
                let mut streams = Vec::new();
                streams
                    .try_reserve_exact(branch_count)
                    .map_err(|_| InternalError::executor_internal())?;
                for spec in active_specs {
                    streams.push(self.index_prefix_stream(request, spec, branch_count)?);
                }

                OrderedKeyStreamBox::merge_all(
                    streams,
                    KeyOrderComparator::from_direction(request.continuation.direction()),
                )
            }
            PrefixSetExecutionShape::OrderedConcat(_) | PrefixSetExecutionShape::Fallback => {
                Err(InternalError::query_executor_invariant())
            }
        }
    }

    fn resolve_branch_ordered_index_prefix_streams(
        &self,
        request: MergedIndexPrefixStreamSpec<'_>,
    ) -> Result<OrderedKeyStreamBox, InternalError> {
        if request.index_prefix_specs.is_empty() {
            return Ok(ordered_key_stream_from_materialized_keys(Vec::new()));
        }

        let mut active_specs =
            active_lowered_index_prefix_specs(Some(self.store), request.index_prefix_specs, None);
        charge_prefix_family_construction(active_specs.as_slice())?;
        sort_lowered_index_prefix_specs_by_raw_lower_key(&mut active_specs)?;
        if matches!(request.continuation.direction(), Direction::Desc) {
            active_specs.reverse();
        }

        match PrefixSetExecutionShape::from_active_prefixes(
            active_specs,
            PrefixSetMergeSafety::OrderedConcatSafe,
        ) {
            PrefixSetExecutionShape::Empty => {
                Ok(ordered_key_stream_from_materialized_keys(Vec::new()))
            }
            PrefixSetExecutionShape::Single(spec) => self.index_prefix_stream(request, spec, 1),
            PrefixSetExecutionShape::OrderedConcat(active_specs) => {
                let branch_count = active_specs.len();
                let mut streams = Vec::new();
                streams
                    .try_reserve_exact(branch_count)
                    .map_err(|_| InternalError::executor_internal())?;
                for spec in active_specs {
                    streams.push(self.index_prefix_stream(request, spec, branch_count)?);
                }

                Ok(OrderedKeyStreamBox::concat_all(streams))
            }
            PrefixSetExecutionShape::OrderedMerge(_) | PrefixSetExecutionShape::Fallback => {
                Err(InternalError::query_executor_invariant())
            }
        }
    }

    fn index_prefix_stream(
        &self,
        request: MergedIndexPrefixStreamSpec<'_>,
        spec: &LoweredIndexPrefixSpec,
        active_branch_count: usize,
    ) -> Result<OrderedKeyStreamBox, InternalError> {
        let branch_chunk_entries = index_leaf_chunk_entries_for_active_page(
            branch_stream_chunk_entries(request.index_fetch_hint, active_branch_count),
            active_branch_count,
        )?;
        let resume_anchor = request.resume_anchor_for(self.entity_tag, spec)?;

        let stream = IndexRangeKeyStream::from_prefix(
            self.store,
            self.entity_tag,
            spec,
            request.continuation.direction(),
            resume_anchor,
            request.index_fetch_hint,
            branch_chunk_entries,
        )?;

        Ok(OrderedKeyStreamBox::index_range(stream))
    }

    // Resolve one secondary-index range scan.
    fn resolve_index_range(
        &self,
        index_range_spec: Option<&LoweredIndexRangeSpec>,
        continuation: IndexScanContinuationInput<'_>,
        index_fetch_hint: Option<usize>,
        index_predicate_execution: Option<IndexPredicateExecution<'_>>,
    ) -> Result<(Vec<DecodedDataStoreKey>, KeyOrderState), InternalError> {
        let spec = IndexRangeTraversalContract::require_spec(index_range_spec)?;
        let fetch_limit = index_fetch_hint.unwrap_or(usize::MAX);
        let key_order_state = if index_fetch_hint.is_some() {
            KeyOrderState::FinalOrder
        } else {
            KeyOrderState::Unordered
        };

        let keys = IndexScan::range_structural(
            self.store,
            self.entity_tag,
            spec,
            continuation,
            fetch_limit,
            index_predicate_execution,
        )?;
        charge_materialized_secondary_index_keys(keys.as_slice(), keys.capacity())?;

        Ok((keys, key_order_state))
    }

    // Resolve one secondary-index range scan as a dynamic ordered stream.
    fn resolve_index_range_stream(
        &self,
        index_range_spec: Option<&LoweredIndexRangeSpec>,
        continuation: IndexScanContinuationInput<'_>,
        index_fetch_hint: Option<usize>,
    ) -> Result<OrderedKeyStreamBox, InternalError> {
        let spec = IndexRangeTraversalContract::require_spec(index_range_spec)?;
        let chunk_entries = index_leaf_chunk_entries_for_active_page(ACCESS_SCAN_CHUNK_ENTRIES, 1)?;

        Ok(OrderedKeyStreamBox::index_range(
            IndexRangeKeyStream::from_range(
                self.store,
                self.entity_tag,
                spec,
                continuation,
                index_fetch_hint,
                chunk_entries,
            ),
        ))
    }
}

fn charge_prefix_family_construction(
    specs: &[&LoweredIndexPrefixSpec],
) -> Result<(), InternalError> {
    IndexPrefixChildExpansionBudget::validate_child_count(specs.len())?;

    let mut descriptor_bytes = specs
        .len()
        .checked_mul(size_of::<OrderedKeyStreamBox>())
        .ok_or_else(InternalError::executor_invariant)?;
    for spec in specs {
        for component in spec.prefix_components() {
            descriptor_bytes = descriptor_bytes
                .checked_add(component.len())
                .ok_or_else(InternalError::executor_invariant)?;
        }
        let (lower, upper) = spec.raw_bounds(&ExecutionConstructionBudget)?;
        descriptor_bytes = descriptor_bytes
            .checked_add(RawIndexStoreKey::bound_backing_bytes(lower))
            .and_then(|bytes| bytes.checked_add(RawIndexStoreKey::bound_backing_bytes(upper)))
            .ok_or_else(InternalError::executor_invariant)?;
    }

    charge_current_execution_budget(
        DiagnosticExecutionBudgetResource::CursorSteps,
        u64::try_from(specs.len()).unwrap_or(u64::MAX),
    )?;
    charge_current_execution_budget(
        DiagnosticExecutionBudgetResource::TemporaryBytes,
        u64::try_from(descriptor_bytes).unwrap_or(u64::MAX),
    )
}

fn primary_key_suffix_resume_anchor_for_prefix(
    entity_tag: EntityTag,
    index: &IndexShapeDetails,
    spec: &LoweredIndexPrefixSpec,
    primary_key_boundary: &CursorBoundary,
) -> Result<RawIndexStoreKey, InternalError> {
    let prefix_len = index.slot_arity();
    let key_arity = index.key_arity();
    if prefix_len > key_arity || spec.prefix_components().len() > key_arity {
        return Err(InternalError::query_executor_invariant());
    }

    let prefix = spec
        .prefix_components()
        .get(..prefix_len)
        .ok_or_else(InternalError::query_executor_invariant)?;
    let suffix_len = if prefix_len == key_arity {
        primary_key_boundary.slots.len()
    } else {
        key_arity - prefix_len
    };
    let primary_key = primary_key_boundary_value(primary_key_boundary, suffix_len)?;
    // The boundary has proved every slot present. Only an uncovered index suffix
    // needs owned values; a full prefix retains the primary-key identity alone.
    let suffix_values = if prefix_len == key_arity {
        Vec::new()
    } else {
        primary_key_boundary
            .slots
            .iter()
            .filter_map(|slot| match slot {
                CursorBoundarySlot::Present(value) => Some(value.clone()),
                CursorBoundarySlot::Missing => None,
            })
            .collect()
    };

    // Prefix-family continuation is valid only when route planning has proven
    // that the remaining index suffix is exactly the primary key. Fill that
    // suffix from the cursor boundary so each prefix stream resumes at the
    // same global primary-key position.
    IndexKey::raw_from_prefix_and_suffix_values_with_primary_key_value(
        IndexId::new_with_generation(entity_tag, index.ordinal(), index.physical_generation()),
        IndexKeyKind::User,
        prefix,
        suffix_values.as_slice(),
        &primary_key,
    )
}

fn sort_lowered_index_prefix_specs_by_raw_lower_key(
    specs: &mut [&LoweredIndexPrefixSpec],
) -> Result<(), InternalError> {
    // Bounds stay owned by their specifications. Admit deferred bounds before
    // sorting, then charge only the extra borrowed descriptors, not key copies.
    let sort_bytes = specs
        .len()
        .checked_mul(size_of::<(&RawIndexStoreKey, &LoweredIndexPrefixSpec)>())
        .ok_or_else(InternalError::executor_invariant)?;
    for spec in specs.iter() {
        spec.raw_bounds(&ExecutionConstructionBudget)?;
    }
    charge_current_execution_budget(
        DiagnosticExecutionBudgetResource::TemporaryBytes,
        u64::try_from(sort_bytes).unwrap_or(u64::MAX),
    )?;
    let mut keyed_specs = Vec::new();
    keyed_specs
        .try_reserve_exact(specs.len())
        .map_err(|_| InternalError::executor_internal())?;
    for &spec in specs.iter() {
        let (lower, _upper) = spec.raw_bounds(&ExecutionConstructionBudget)?;
        let Bound::Included(raw_key) = lower else {
            return Err(InternalError::query_executor_invariant());
        };
        keyed_specs.push((raw_key, spec));
    }

    keyed_specs.sort_by(|left, right| left.0.cmp(right.0));
    for (destination, (_raw_key, spec)) in specs.iter_mut().zip(keyed_specs) {
        *destination = spec;
    }

    Ok(())
}

fn primary_key_boundary_value(
    boundary: &CursorBoundary,
    suffix_len: usize,
) -> Result<crate::db::PrimaryKeyValue, InternalError> {
    // Absence rejects before key arity/type decoding, as it did when collecting
    // owned values. The immutable borrowed projection below cannot omit a slot.
    if boundary.slots.len() != suffix_len
        || boundary
            .slots
            .iter()
            .any(|slot| matches!(slot, CursorBoundarySlot::Missing))
    {
        return Err(InternalError::query_executor_invariant());
    }

    primary_key_value_from_structural_values(boundary.slots.iter().filter_map(|slot| match slot {
        CursorBoundarySlot::Present(value) => Some(value),
        CursorBoundarySlot::Missing => None,
    }))
}

fn primary_key_boundary_data_key(
    entity_tag: EntityTag,
    boundary: &CursorBoundary,
) -> Result<DecodedDataStoreKey, InternalError> {
    let primary_key = primary_key_boundary_value(boundary, boundary.slots.len())?;

    Ok(DecodedDataStoreKey::new_primary_key_value(
        entity_tag,
        &primary_key,
    ))
}

// Both physical leaves expose the same polling contract. Storage-specific
// refill and cursor binding remain on the concrete backend; page owners admit
// the next indivisible refill using its entry bound before polling.
macro_rules! impl_ordered_physical_stream {
    ($stream:ident) => {
        impl OrderedKeyStream for $stream {
            fn next_key(&mut self) -> Result<Option<DecodedDataStoreKey>, InternalError> {
                self.pull_next_key()
            }

            fn page_access_entry_bound(&self) -> Option<usize> {
                Some(self.next_pull_entry_bound())
            }
        }
    };
}

///
/// PrimaryRangeKeyStream
///
/// PrimaryRangeKeyStream incrementally resolves one primary-key data-store
/// range.
/// It owns only raw range bounds and a small decoded-key buffer so callers can
/// consume primary scans without materializing every candidate key up front.
///

pub(in crate::db::executor) struct PrimaryRangeKeyStream {
    store: StoreHandle,
    entity_tag: EntityTag,
    lower_bound: Bound<RawDataStoreKey>,
    upper_bound: Bound<RawDataStoreKey>,
    direction: Direction,
    remaining: Option<usize>,
    chunk_entries: usize,
    buffer: IntoIter<DecodedDataStoreKey>,
    // Retain first-pull admission for direct visits and continuation binding.
    buffer_pos: usize,
    exhausted: bool,
}

impl PrimaryRangeKeyStream {
    // Build one primary stream from validated structural data keys.
    #[cfg(test)]
    pub(in crate::db::executor) fn new(
        store: StoreHandle,
        start: DecodedDataStoreKey,
        end: DecodedDataStoreKey,
        direction: Direction,
        limit: Option<usize>,
    ) -> Result<Self, InternalError> {
        Self::new_bounds(
            store,
            start.entity_tag(),
            Some(start),
            Some(end),
            direction,
            limit,
        )
    }

    // Reuse the entity-prefix limits for whichever authored endpoint is absent.
    pub(in crate::db::executor) fn new_bounds(
        store: StoreHandle,
        entity: EntityTag,
        start: Option<DecodedDataStoreKey>,
        end: Option<DecodedDataStoreKey>,
        direction: Direction,
        limit: Option<usize>,
    ) -> Result<Self, InternalError> {
        let range = RawDataStoreKeyRange::entity_prefix(entity);
        let lower_bound = match start {
            Some(start) => Bound::Included(start.to_raw()?),
            None => Bound::Included(RawDataStoreKey::store_range_lower_key(&range)),
        };
        let upper_bound = match end {
            Some(end) => Bound::Included(end.to_raw()?),
            None => range
                .upper_exclusive()
                .map(RawDataStoreKey::from_store_range_bound)
                .map_or(Bound::Unbounded, Bound::Excluded),
        };
        Ok(Self {
            store,
            entity_tag: entity,
            lower_bound,
            upper_bound,
            direction,
            remaining: limit,
            chunk_entries: primary_range_chunk_entries_for_active_page()?,
            buffer: Vec::new().into_iter(),
            buffer_pos: 0,
            exhausted: false,
        })
    }

    // Build one primary stream over all rows for one entity using compact
    // raw-prefix bounds rather than synthetic primary-key sentinels.
    pub(in crate::db::executor) fn new_full_scan(
        store: StoreHandle,
        entity: EntityTag,
        direction: Direction,
        limit: Option<usize>,
    ) -> Result<Self, InternalError> {
        let range = RawDataStoreKeyRange::entity_prefix(entity);
        let lower_bound = Bound::Included(RawDataStoreKey::store_range_lower_key(&range));
        let upper_bound = range
            .upper_exclusive()
            .map(RawDataStoreKey::from_store_range_bound)
            .map_or(Bound::Unbounded, Bound::Excluded);

        Ok(Self {
            store,
            entity_tag: entity,
            lower_bound,
            upper_bound,
            direction,
            remaining: limit,
            chunk_entries: primary_range_chunk_entries_for_active_page()?,
            buffer: Vec::new().into_iter(),
            buffer_pos: 0,
            exhausted: false,
        })
    }

    /// Visit primary rows in either direction through one open range when the store
    /// has a single visible backing and this leaf has not otherwise advanced.
    pub(in crate::db::executor) fn try_visit_rows_direct(
        &mut self,
        begin_row: &mut dyn FnMut() -> Result<bool, InternalError>,
        visit_row: &mut dyn for<'row> FnMut(
            DecodedDataStoreKey,
            &'row RawRow,
        ) -> Result<StoreVisit, InternalError>,
    ) -> Result<Option<()>, InternalError> {
        if self.buffer_pos != 0 || !self.buffer.as_slice().is_empty() {
            return Ok(None);
        }
        if self.exhausted || matches!(self.remaining, Some(0)) {
            self.exhausted = true;
            return Ok(Some(()));
        }

        let row_limit = self.remaining;
        let visited = Cell::new(0usize);
        let mut last_primary_key = None;
        let outcome = self.store.with_data(|store| {
            store.try_visit_range_with_row_preflight(
                (self.lower_bound.as_ref(), self.upper_bound.as_ref()),
                self.direction,
                |_raw_key| {
                    if row_limit.is_some_and(|limit| visited.get() >= limit) || !begin_row()? {
                        return Ok::<StoreVisit, InternalError>(StoreVisit::Stop);
                    }

                    Ok::<StoreVisit, InternalError>(StoreVisit::Continue)
                },
                |raw_key, row| {
                    charge_current_execution_budget(
                        DiagnosticExecutionBudgetResource::KeyIndexEntriesVisited,
                        1,
                    )?;
                    let key = PrimaryScan::decode_data_key(raw_key)?;
                    let primary_key = key.primary_key_value();
                    let visit = visit_row(key, row)?;
                    visited.set(visited.get().saturating_add(1));
                    last_primary_key = Some(primary_key);

                    Ok(visit)
                },
            )
        })?;
        let Some(naturally_exhausted) = outcome else {
            return Ok(None);
        };

        // The visitor consumes decoded keys. Retain their compact identity
        // and encode only the final resume boundary after traversal.
        if let Some(primary_key) = last_primary_key {
            let boundary = Bound::Excluded(
                DecodedDataStoreKey::new_primary_key_value(self.entity_tag, &primary_key)
                    .to_raw()?,
            );
            match self.direction {
                Direction::Asc => self.lower_bound = boundary,
                Direction::Desc => self.upper_bound = boundary,
            }
        }
        if let Some(remaining) = self.remaining.as_mut() {
            *remaining = remaining.saturating_sub(visited.get());
        }
        self.exhausted = naturally_exhausted || matches!(self.remaining, Some(0));

        Ok(Some(()))
    }

    // Bind an authenticated primary-key continuation directly into this
    // stream's raw range before its first pull. The route fetch hint is then
    // page-local rather than being consumed while replaying earlier pages.
    fn resume_strictly_after(
        &mut self,
        boundary: DecodedDataStoreKey,
    ) -> Result<(), InternalError> {
        if boundary.entity_tag() != self.entity_tag
            || self.buffer_pos != 0
            || !self.buffer.as_slice().is_empty()
        {
            return Err(InternalError::query_executor_invariant());
        }
        let raw_boundary = boundary.to_raw()?;
        let bound_bytes = u64::try_from(raw_boundary.as_bytes().len()).unwrap_or(u64::MAX);
        charge_current_execution_budget(DiagnosticExecutionBudgetResource::CursorSteps, 1)?;
        charge_current_execution_budget(
            DiagnosticExecutionBudgetResource::TemporaryBytes,
            bound_bytes,
        )?;

        match self.direction {
            Direction::Asc => {
                if lower_bound_precedes_raw(&self.lower_bound, &raw_boundary) {
                    self.lower_bound = Bound::Excluded(raw_boundary);
                }
            }
            Direction::Desc => {
                if upper_bound_follows_raw(&self.upper_bound, &raw_boundary) {
                    self.upper_bound = Bound::Excluded(raw_boundary);
                }
            }
        }
        self.exhausted = !raw_bounds_may_contain_key(&self.lower_bound, &self.upper_bound);

        Ok(())
    }

    // Return the maximum number of keys to read during the next store borrow.
    const fn next_chunk_limit(&self) -> usize {
        match self.remaining {
            Some(remaining) if remaining < self.chunk_entries => remaining,
            Some(_) | None => self.chunk_entries,
        }
    }

    // Return the complete physical-entry bound for the next pull. Buffered
    // keys require no new storage traversal; an empty buffer may refill
    // one already-bounded chunk. The page owner reserves this amount before
    // polling, so batching does not advance unadmitted storage work.
    fn next_pull_entry_bound(&self) -> usize {
        if !self.buffer.as_slice().is_empty() || self.exhausted || matches!(self.remaining, Some(0))
        {
            0
        } else {
            self.next_chunk_limit()
        }
    }

    // Re-enter the data store for one bounded range chunk.
    fn load_next_chunk(&mut self) -> Result<(), InternalError> {
        let chunk_limit = self.next_chunk_limit();
        if self.exhausted || chunk_limit == 0 {
            self.exhausted = true;
            return Ok(());
        }

        let (keys, last_raw_key) = self.store.with_data(|store| {
            let mut keys = Vec::with_capacity(chunk_limit);
            match self.direction {
                Direction::Asc => {
                    store.visit_key_range(
                        (self.lower_bound.as_ref(), self.upper_bound.as_ref()),
                        |raw_key| {
                            charge_current_execution_budget(
                                DiagnosticExecutionBudgetResource::KeyIndexEntriesVisited,
                                1,
                            )?;
                            keys.push(PrimaryScan::decode_data_key(raw_key)?);
                            Ok::<StoreVisit, InternalError>(if keys.len() == chunk_limit {
                                StoreVisit::Stop
                            } else {
                                StoreVisit::Continue
                            })
                        },
                    )?;
                }
                Direction::Desc => {
                    store.visit_key_range_rev(
                        (self.lower_bound.as_ref(), self.upper_bound.as_ref()),
                        |raw_key| {
                            charge_current_execution_budget(
                                DiagnosticExecutionBudgetResource::KeyIndexEntriesVisited,
                                1,
                            )?;
                            keys.push(PrimaryScan::decode_data_key(raw_key)?);
                            Ok::<StoreVisit, InternalError>(if keys.len() == chunk_limit {
                                StoreVisit::Stop
                            } else {
                                StoreVisit::Continue
                            })
                        },
                    )?;
                }
            }

            // Decoding already freezes each raw key. Copy only the final one
            // for the next chunk boundary instead of every visited key.
            let last_raw_key = keys.last().map(DecodedDataStoreKey::to_raw).transpose()?;
            Ok::<_, InternalError>((keys, last_raw_key))
        })?;

        let emitted = keys.len();
        self.buffer = keys.into_iter();
        self.buffer_pos = 0;

        if let Some(raw_key) = last_raw_key {
            match self.direction {
                Direction::Asc => self.lower_bound = Bound::Excluded(raw_key),
                Direction::Desc => self.upper_bound = Bound::Excluded(raw_key),
            }
        } else {
            self.exhausted = true;
        }

        if let Some(remaining) = self.remaining.as_mut() {
            *remaining = remaining.saturating_sub(emitted);
            if *remaining == 0 {
                self.exhausted = true;
            }
        }

        if emitted < chunk_limit {
            self.exhausted = true;
        }

        Ok(())
    }

    fn pull_next_key(&mut self) -> Result<Option<DecodedDataStoreKey>, InternalError> {
        if self.buffer.as_slice().is_empty() {
            self.load_next_chunk()?;
        }
        let key = self.buffer.next();
        if key.is_some() {
            self.buffer_pos += 1;
        }
        Ok(key)
    }
}

fn primary_range_chunk_entries_for_active_page() -> Result<usize, InternalError> {
    let Some(page_entry_limit) = production_scalar_page_access_entry_limit()? else {
        return Ok(ACCESS_SCAN_CHUNK_ENTRIES);
    };

    Ok(ACCESS_SCAN_CHUNK_ENTRIES.min(page_entry_limit.max(1)))
}

// Keep one index leaf's next refill inside its share of the active page-entry
// envelope. A merged prefix family may poll every child to establish its next
// winner, so each child owns at most an equal conservative share.
fn index_leaf_chunk_entries_for_active_page(
    requested_entries: usize,
    active_branch_count: usize,
) -> Result<usize, InternalError> {
    let Some(page_entry_limit) = production_scalar_page_access_entry_limit()? else {
        return Ok(requested_entries);
    };
    let branch_count = active_branch_count.max(1);
    let per_branch_limit = page_entry_limit / branch_count;

    Ok(requested_entries.min(per_branch_limit.max(1)))
}

fn lower_bound_precedes_raw(lower: &Bound<RawDataStoreKey>, raw: &RawDataStoreKey) -> bool {
    match lower {
        Bound::Included(value) => value <= raw,
        Bound::Excluded(value) => value < raw,
        Bound::Unbounded => true,
    }
}

fn upper_bound_follows_raw(upper: &Bound<RawDataStoreKey>, raw: &RawDataStoreKey) -> bool {
    match upper {
        Bound::Included(value) => value >= raw,
        Bound::Excluded(value) => value > raw,
        Bound::Unbounded => true,
    }
}

fn raw_bounds_may_contain_key(
    lower: &Bound<RawDataStoreKey>,
    upper: &Bound<RawDataStoreKey>,
) -> bool {
    match (lower, upper) {
        (Bound::Unbounded, _) | (_, Bound::Unbounded) => true,
        (Bound::Included(lower), Bound::Included(upper)) => lower <= upper,
        (Bound::Included(lower) | Bound::Excluded(lower), Bound::Excluded(upper))
        | (Bound::Excluded(lower), Bound::Included(upper)) => lower < upper,
    }
}

impl_ordered_physical_stream!(PrimaryRangeKeyStream);

///
/// IndexRangeKeyStream
///
/// IndexRangeKeyStream incrementally resolves one lowered secondary-index
/// range when physical index order is already the final caller-visible order.
/// Cases that still require `DecodedDataStoreKey` sorting, deduplication, or residual
/// index-predicate filtering intentionally stay on the materialized fallback.
///

pub(in crate::db::executor) struct IndexRangeKeyStream {
    store: StoreHandle,
    entity_tag: EntityTag,
    lower: Bound<LoweredKey>,
    upper: Bound<LoweredKey>,
    direction: Direction,
    anchor: Option<RawIndexStoreKey>,
    remaining: Option<usize>,
    chunk_entries: usize,
    buffer: IntoIter<DecodedDataStoreKey>,
    exhausted: bool,
}

impl IndexRangeKeyStream {
    // Build one index stream from a lowered prefix envelope.
    fn from_prefix(
        store: StoreHandle,
        entity_tag: EntityTag,
        spec: &LoweredIndexPrefixSpec,
        direction: Direction,
        anchor: Option<RawIndexStoreKey>,
        limit: Option<usize>,
        chunk_entries: usize,
    ) -> Result<Self, InternalError> {
        let (lower, upper) = spec.raw_bounds(&ExecutionConstructionBudget)?;
        let bounds =
            resume_prefix_bounds_for_continuation(direction, anchor.as_ref(), lower, upper)?;
        Ok(Self::new(
            store,
            entity_tag,
            bounds,
            direction,
            None,
            limit,
            chunk_entries,
        ))
    }

    // Build one index stream from a lowered range envelope and continuation.
    fn from_range(
        store: StoreHandle,
        entity_tag: EntityTag,
        spec: &LoweredIndexRangeSpec,
        continuation: IndexScanContinuationInput<'_>,
        limit: Option<usize>,
        chunk_entries: usize,
    ) -> Self {
        Self::new(
            store,
            entity_tag,
            (spec.lower().clone(), spec.upper().clone()),
            continuation.direction(),
            continuation.anchor().cloned(),
            limit,
            chunk_entries,
        )
    }

    fn new(
        store: StoreHandle,
        entity_tag: EntityTag,
        bounds: (Bound<LoweredKey>, Bound<LoweredKey>),
        direction: Direction,
        anchor: Option<RawIndexStoreKey>,
        limit: Option<usize>,
        chunk_entries: usize,
    ) -> Self {
        let (lower, upper) = bounds;
        Self {
            store,
            entity_tag,
            lower,
            upper,
            direction,
            anchor,
            remaining: limit,
            chunk_entries,
            buffer: Vec::new().into_iter(),
            exhausted: false,
        }
    }

    // Return the maximum raw index entries one empty-buffer pull may visit.
    const fn next_chunk_limit(&self) -> usize {
        index_stream_chunk_entries_for_remaining(self.chunk_entries, self.remaining)
    }

    // Return the complete physical-entry bound for the next pull. The bound
    // falls to zero while decoded keys remain buffered, allowing page-local
    // admission to preserve chunked store traversal without over-reserving
    // every candidate.
    fn next_pull_entry_bound(&self) -> usize {
        if !self.buffer.as_slice().is_empty() || self.exhausted || matches!(self.remaining, Some(0))
        {
            0
        } else {
            self.next_chunk_limit()
        }
    }

    // Re-enter the index store for one bounded raw-index chunk.
    fn load_next_chunk(&mut self) -> Result<(), InternalError> {
        if self.exhausted || matches!(self.remaining, Some(0)) {
            self.exhausted = true;
            return Ok(());
        }

        let chunk_entries = self.next_chunk_limit();
        let continuation = IndexScanContinuationInput::new(self.anchor.as_ref(), self.direction);
        let chunk = IndexScan::chunk_structural(
            self.store,
            self.entity_tag,
            &self.lower,
            &self.upper,
            continuation,
            chunk_entries,
            index_stream_output_limit_for_chunk(self.remaining, chunk_entries),
        )?;
        let (keys, last_raw_key) = chunk.into_decoded_keys_and_resume_anchor();
        let emitted = keys.len();
        self.buffer = keys.into_iter();

        apply_index_scan_chunk_progress(
            &mut self.anchor,
            &mut self.remaining,
            &mut self.exhausted,
            emitted,
            last_raw_key,
        );

        Ok(())
    }

    fn pull_next_key(&mut self) -> Result<Option<DecodedDataStoreKey>, InternalError> {
        while self.buffer.as_slice().is_empty() && !self.exhausted {
            self.load_next_chunk()?;
        }
        Ok(self.buffer.next())
    }
}

impl_ordered_physical_stream!(IndexRangeKeyStream);

// Normalize key ordering according to explicit resolver output state.
fn normalize_ordered_keys(
    keys: &mut [DecodedDataStoreKey],
    direction: Direction,
    key_order_state: KeyOrderState,
) {
    match key_order_state {
        KeyOrderState::FinalOrder => {}
        KeyOrderState::AscendingSorted => {
            if matches!(direction, Direction::Desc) {
                keys.reverse();
            }
        }
        KeyOrderState::Unordered => {
            keys.sort_unstable();
            if matches!(direction, Direction::Desc) {
                keys.reverse();
            }
        }
    }
}

/// Finish a blocking ordered stream using the existing materialized-key budget.
/// Retain input order and duplicates; physical pulls keep their existing budgets.
pub(in crate::db::executor) fn materialize_ordered_key_stream(
    mut stream: OrderedKeyStreamBox,
) -> Result<OrderedKeyStreamBox, InternalError> {
    let mut keys = Vec::new();
    while let Some(key) = stream.next_key()? {
        if keys.len() == keys.capacity() {
            let growth = keys.capacity().max(1);
            reserve_materialized_secondary_index_key_capacity(&mut keys, growth)?;
        }
        charge_materialized_secondary_index_keys(std::slice::from_ref(&key), 0)?;
        keys.push(key);
    }

    Ok(ordered_key_stream_from_materialized_keys(keys))
}

// Charge the complete retained owner of one materialized secondary-index key
// vector. `DecodedDataStoreKey` keeps its raw persisted key in an owned cache,
// so both vector capacity and every live raw-key allocation belong to this
// blocking fallback rather than only the logical key count.
pub(in crate::db::executor) fn charge_materialized_secondary_index_keys(
    keys: &[DecodedDataStoreKey],
    capacity: usize,
) -> Result<(), InternalError> {
    charge_materialized_secondary_index_key_capacity(capacity)?;

    let mut raw_key_bytes = 0usize;
    for key in keys {
        raw_key_bytes = raw_key_bytes
            .checked_add(key.raw_key()?.as_bytes().len())
            .ok_or_else(InternalError::executor_invariant)?;
    }

    charge_current_execution_budget(
        DiagnosticExecutionBudgetResource::TemporaryBytes,
        u64::try_from(raw_key_bytes).unwrap_or(u64::MAX),
    )
}

fn charge_materialized_secondary_index_key_capacity(capacity: usize) -> Result<(), InternalError> {
    let bytes = capacity
        .checked_mul(size_of::<DecodedDataStoreKey>())
        .ok_or_else(InternalError::executor_invariant)?;
    charge_current_execution_budget(
        DiagnosticExecutionBudgetResource::TemporaryBytes,
        u64::try_from(bytes).unwrap_or(u64::MAX),
    )
}

fn new_materialized_secondary_index_key_vector(
    capacity: usize,
) -> Result<Vec<DecodedDataStoreKey>, InternalError> {
    charge_materialized_secondary_index_key_capacity(capacity)?;
    let mut keys = Vec::new();
    keys.try_reserve_exact(capacity)
        .map_err(|_| InternalError::executor_internal())?;
    let uncharged_capacity = keys.capacity().saturating_sub(capacity);
    charge_materialized_secondary_index_key_capacity(uncharged_capacity)?;

    Ok(keys)
}

fn reserve_materialized_secondary_index_key_capacity(
    keys: &mut Vec<DecodedDataStoreKey>,
    additional: usize,
) -> Result<(), InternalError> {
    let required = keys
        .len()
        .checked_add(additional)
        .ok_or_else(InternalError::executor_invariant)?;
    if required <= keys.capacity() {
        return Ok(());
    }

    let prior_capacity = keys.capacity();
    let minimum_growth = required.saturating_sub(prior_capacity);
    charge_materialized_secondary_index_key_capacity(minimum_growth)?;
    keys.try_reserve_exact(additional)
        .map_err(|_| InternalError::executor_internal())?;
    let actual_growth = keys.capacity().saturating_sub(prior_capacity);
    charge_materialized_secondary_index_key_capacity(actual_growth.saturating_sub(minimum_growth))
}

fn charge_materialized_key_dedup_comparisons(entries: usize) -> Result<(), InternalError> {
    charge_current_execution_budget(
        DiagnosticExecutionBudgetResource::SortComparisons,
        u64::try_from(entries.saturating_sub(1)).unwrap_or(u64::MAX),
    )
}

// Return whether one secondary-index path can preserve raw index traversal
// order directly instead of materializing to sort or deduplicate `DecodedDataStoreKey`s.
const fn index_path_can_stream_in_final_order(request: PhysicalStreamBindings<'_>) -> bool {
    request.index_predicate_execution.is_none()
        && (request
            .execution_policy
            .index_leaf_order_policy()
            .preserves_leaf_index_order()
            || request.execution_policy.physical_fetch_hint().is_some())
}

fn validate_index_prefix_count(
    index_prefix_specs: &[LoweredIndexPrefixSpec],
    expected_prefix_count: usize,
) -> Result<(), InternalError> {
    if index_prefix_specs.len() != expected_prefix_count {
        return Err(InternalError::query_executor_invariant());
    }

    Ok(())
}

fn resolve_index_multi_lookup_physical_key_stream(
    index: &IndexShapeDetails,
    value_count: usize,
    request: PhysicalStreamBindings<'_>,
    runtime: &KeyAccessRuntime,
) -> Result<PhysicalKeyResolution, InternalError> {
    if let Some(expansion) = request.index_prefix_child_expansion {
        if let Some(stream) = runtime.expanded_index_multi_lookup_stream(
            index,
            request.index_prefix_specs,
            value_count,
            request.continuation,
            request.execution_policy.physical_fetch_hint(),
            expansion,
        )? {
            return Ok(PhysicalKeyResolution::Stream(Box::new(stream)));
        }

        let (candidates, key_order_state) = runtime.resolve_index_multi_lookup(
            request.index_prefix_specs,
            value_count,
            request.continuation.index_scan_continuation(),
            request.execution_policy.index_leaf_order_policy(),
            request.execution_policy.physical_fetch_hint(),
            request.index_predicate_execution,
        )?;

        return Ok(PhysicalKeyResolution::Materialized {
            candidates,
            key_order_state,
        });
    }

    if index_path_can_stream_in_final_order(request) {
        return Ok(PhysicalKeyResolution::Stream(Box::new(
            runtime.resolve_index_multi_lookup_stream(
                index,
                request.index_prefix_specs,
                value_count,
                request.continuation,
                request.execution_policy.physical_fetch_hint(),
                request.execution_policy.index_leaf_order_policy(),
            )?,
        )));
    }

    let (candidates, key_order_state) = runtime.resolve_index_multi_lookup(
        request.index_prefix_specs,
        value_count,
        request.continuation.index_scan_continuation(),
        request.execution_policy.index_leaf_order_policy(),
        request.execution_policy.physical_fetch_hint(),
        request.index_predicate_execution,
    )?;

    Ok(PhysicalKeyResolution::Materialized {
        candidates,
        key_order_state,
    })
}

fn resolve_index_physical_key_stream(
    path: &ExecutionPathPayload<'_, Value>,
    request: PhysicalStreamBindings<'_>,
    runtime: &KeyAccessRuntime,
) -> Result<PhysicalKeyResolution, InternalError> {
    let (candidates, key_order_state) = match path {
        ExecutionPathPayload::IndexPrefix { index } => {
            if index_path_can_stream_in_final_order(request) {
                return Ok(PhysicalKeyResolution::Stream(Box::new(
                    runtime.resolve_index_prefix_stream(
                        index,
                        request.index_prefix_specs,
                        request.continuation,
                        request.execution_policy.physical_fetch_hint(),
                        request.execution_policy.index_leaf_order_policy(),
                    )?,
                )));
            }

            runtime.resolve_index_prefix(
                request.index_prefix_specs,
                request.continuation.index_scan_continuation(),
                request.execution_policy.physical_fetch_hint(),
                request.index_predicate_execution,
            )?
        }
        ExecutionPathPayload::IndexMultiLookup { index, value_count } => {
            match resolve_index_multi_lookup_physical_key_stream(
                index,
                *value_count,
                request,
                runtime,
            )? {
                PhysicalKeyResolution::Stream(stream) => {
                    return Ok(PhysicalKeyResolution::Stream(stream));
                }
                PhysicalKeyResolution::Materialized {
                    candidates,
                    key_order_state,
                } => (candidates, key_order_state),
            }
        }
        ExecutionPathPayload::IndexBranchSet {
            index,
            branch_count,
        } => {
            if index_path_can_stream_in_final_order(request) {
                return Ok(PhysicalKeyResolution::Stream(Box::new(
                    runtime.resolve_index_branch_set_stream(
                        index,
                        request.index_prefix_specs,
                        *branch_count,
                        request.continuation,
                        request.execution_policy.physical_fetch_hint(),
                    )?,
                )));
            }

            runtime.resolve_index_multi_lookup(
                request.index_prefix_specs,
                *branch_count,
                request.continuation.index_scan_continuation(),
                request.execution_policy.index_leaf_order_policy(),
                request.execution_policy.physical_fetch_hint(),
                request.index_predicate_execution,
            )?
        }
        ExecutionPathPayload::IndexRange { .. } => {
            if index_path_can_stream_in_final_order(request) {
                return Ok(PhysicalKeyResolution::Stream(Box::new(
                    runtime.resolve_index_range_stream(
                        request.index_range_spec,
                        request.continuation.index_scan_continuation(),
                        request.execution_policy.physical_fetch_hint(),
                    )?,
                )));
            }

            runtime.resolve_index_range(
                request.index_range_spec,
                request.continuation.index_scan_continuation(),
                request.execution_policy.physical_fetch_hint(),
                request.index_predicate_execution,
            )?
        }
        ExecutionPathPayload::ByKey(_)
        | ExecutionPathPayload::ByKeys(_)
        | ExecutionPathPayload::KeyRange { .. }
        | ExecutionPathPayload::FullScan => return Err(InternalError::query_executor_invariant()),
    };

    Ok(PhysicalKeyResolution::Materialized {
        candidates,
        key_order_state,
    })
}

// Resolve one physical access path by dispatching only the coarse path shape
// through the runtime leaf boundary.
fn resolve_physical_key_stream(
    path: &ExecutionPathPayload<'_, Value>,
    request: PhysicalStreamBindings<'_>,
    runtime: &KeyAccessRuntime,
) -> Result<OrderedKeyStreamBox, InternalError> {
    let path_facts = path.shape_facts();
    let primary_scan_fetch_hint = if primary_scan_fetch_hint_shape_supported(&path_facts) {
        request.execution_policy.physical_fetch_hint()
    } else {
        None
    };

    let (mut candidates, mut key_order_state) = match path {
        ExecutionPathPayload::ByKey(key) => runtime.resolve_by_key(key)?,
        ExecutionPathPayload::ByKeys(keys) => runtime.resolve_by_keys(keys)?,
        ExecutionPathPayload::KeyRange { start, end } => {
            return runtime.resolve_key_range_stream(
                start.cloned(),
                end.cloned(),
                request.continuation,
                primary_scan_fetch_hint,
            );
        }
        ExecutionPathPayload::FullScan => {
            return runtime.resolve_full_scan_stream(request.continuation, primary_scan_fetch_hint);
        }
        ExecutionPathPayload::IndexPrefix { .. }
        | ExecutionPathPayload::IndexMultiLookup { .. }
        | ExecutionPathPayload::IndexBranchSet { .. }
        | ExecutionPathPayload::IndexRange { .. } => {
            match resolve_index_physical_key_stream(path, request, runtime)? {
                PhysicalKeyResolution::Stream(stream) => return Ok(*stream),
                PhysicalKeyResolution::Materialized {
                    candidates,
                    key_order_state,
                } => (candidates, key_order_state),
            }
        }
    };

    // Top-level single-path secondary-index scans must preserve physical index
    // traversal order so route-owned secondary ORDER BY contracts can drive
    // paging without an extra materialized reorder. Composite child streams
    // still disable this flag so merge/intersection reducers continue to
    // consume canonical `DecodedDataStoreKey` order.
    if request
        .execution_policy
        .index_leaf_order_policy()
        .preserves_leaf_index_order()
        && matches!(
            path,
            ExecutionPathPayload::IndexPrefix { .. } | ExecutionPathPayload::IndexRange { .. }
        )
        && matches!(key_order_state, KeyOrderState::Unordered)
    {
        key_order_state = KeyOrderState::FinalOrder;
    }

    if matches!(key_order_state, KeyOrderState::Unordered) {
        charge_sort_work::<DecodedDataStoreKey>(candidates.len())?;
    }
    normalize_ordered_keys(
        &mut candidates,
        request.continuation.direction(),
        key_order_state,
    );

    // Apply authenticated primary-key progress before bounded page traversal.
    // Otherwise materialized leaves repeatedly spend the page on earlier keys,
    // and an empty page can lose logical progress without advancing physically.
    if let Some(boundary) = request.continuation.primary_key_boundary() {
        let boundary = primary_key_boundary_data_key(runtime.entity_tag, boundary)?;
        let comparator = KeyOrderComparator::from_direction(request.continuation.direction());
        charge_current_execution_budget(
            DiagnosticExecutionBudgetResource::CursorSteps,
            u64::try_from(candidates.len()).unwrap_or(u64::MAX),
        )?;
        candidates.retain(|key| comparator.compare_data_keys(key, &boundary).is_gt());
    }

    Ok(ordered_key_stream_from_materialized_keys(candidates))
}

impl ExecutionPathPayload<'_, Value> {
    // Physical access lowering for one structural executable access path.
    // Typed key recovery is deferred to the concrete path leaves in the
    // structural runtime adapter.
    /// Build an ordered key stream for one structural access path.
    pub(super) fn resolve_structural_physical_key_stream(
        &self,
        request: StructuralPhysicalStreamRequest<'_>,
    ) -> Result<OrderedKeyStreamBox, InternalError> {
        let runtime = KeyAccessRuntime::new(request.store, request.entity_tag);
        let bindings = PhysicalStreamBindings {
            index_prefix_specs: request.index_prefix_specs,
            index_range_spec: request.index_range_spec,
            continuation: request.continuation,
            execution_policy: request.execution_policy,
            index_predicate_execution: request.index_predicate_execution,
            index_prefix_child_expansion: request.index_prefix_child_expansion,
        };

        resolve_physical_key_stream(self, bindings, &runtime)
    }
}

#[cfg(test)]
mod physical_polling_tests {
    use super::*;
    use crate::{
        db::{
            QueryError,
            data::{DataStore, RawRow},
            index::{IndexEntryValue, IndexId, IndexStore},
            key_taxonomy::{CompositePrimaryKeyValue, PrimaryKeyComponent, PrimaryKeyValue},
            registry::{StoreAllocationIdentities, StoreRuntimeStorageCapabilities},
            resource::budget::{
                HardExecutionBudget, HardExecutionContext, HardExecutionFailureHeadroom,
                with_query_execution_budget_for_tests,
            },
            schema::SchemaStore,
            test_support::index::nat64_index_key,
        },
        types::EntityTag,
    };
    use ic_storage::adapter::Storable;
    use icydb_diagnostic_code::{
        DiagnosticDetail, DiagnosticExecutionBudgetScope, DiagnosticExecutionLane,
        DiagnosticFactTag, RuntimeBoundaryCode,
    };
    use std::{borrow::Cow, cell::RefCell};

    const ENTITY: EntityTag = EntityTag::new(0x222);

    thread_local! {
        static DATA: RefCell<DataStore> = const { RefCell::new(DataStore::init_heap()) };
        static INDEX: RefCell<IndexStore> = const { RefCell::new(IndexStore::init_heap()) };
        static SCHEMA: RefCell<SchemaStore> = const { RefCell::new(SchemaStore::init_heap()) };
    }

    const STORE: StoreHandle = StoreHandle::new(
        &DATA,
        &INDEX,
        &SCHEMA,
        StoreAllocationIdentities::absent(),
        StoreRuntimeStorageCapabilities::heap(),
    );

    fn data_key(value: u64) -> DecodedDataStoreKey {
        DecodedDataStoreKey::new(
            ENTITY,
            &PrimaryKeyValue::from(PrimaryKeyComponent::Nat64(value)),
        )
    }

    fn index_key(index_id: &IndexId, component: &[u8], value: u64) -> IndexKey {
        nat64_index_key(index_id, component, value)
    }

    fn reset_heap_stores() {
        DATA.with_borrow_mut(|store| *store = DataStore::init_heap());
        INDEX.with_borrow_mut(|store| *store = IndexStore::init_heap());
    }

    fn load_primary_keys() {
        DATA.with_borrow_mut(|store| {
            for value in 1..=100 {
                store.insert_raw_for_test(
                    data_key(value)
                        .to_raw()
                        .expect("test data key should encode"),
                    RawRow::try_new(vec![0]).expect("test row should be bounded"),
                );
            }
        });
    }

    fn load_index_keys(index_id: &IndexId, component: &[u8]) {
        INDEX.with_borrow_mut(|store| {
            for value in 1..=100 {
                store.insert(
                    index_key(index_id, component, value)
                        .to_raw()
                        .expect("test index key should encode"),
                    IndexEntryValue::presence(),
                );
            }
        });
    }

    fn ordered_index_stream(
        direction: Direction,
        limit: Option<usize>,
        anchor: Option<RawIndexStoreKey>,
    ) -> IndexRangeKeyStream {
        let index_id = IndexId::new(ENTITY, 1);
        IndexRangeKeyStream::new(
            STORE,
            ENTITY,
            (
                Bound::Included(index_key(&index_id, b"lane", 0).to_raw().unwrap()),
                Bound::Included(index_key(&index_id, b"lane", 101).to_raw().unwrap()),
            ),
            direction,
            anchor,
            limit,
            ACCESS_SCAN_CHUNK_ENTRIES,
        )
    }

    fn collect_keys(stream: &mut impl OrderedKeyStream) -> Vec<DecodedDataStoreKey> {
        let mut keys = Vec::new();
        while let Some(key) = stream.next_key().unwrap() {
            keys.push(key);
        }
        assert_eq!(stream.next_key().unwrap(), None);
        keys
    }

    #[test]
    fn primary_boundaries_preserve_key_bytes_and_typed_refusals() {
        use crate::types::{Account, Principal, Subaccount};

        let account = Value::Account(Account::from_owner_and_subaccount(
            Principal::MAX,
            Some(Subaccount::MAX),
        ));
        let cases = [
            vec![],
            vec![Value::Unit],
            vec![account.clone()],
            vec![Value::Nat64(7), account.clone()],
            vec![account; 4],
            vec![Value::List(vec![Value::Nat64(1), Value::Nat64(2)])],
            vec![Value::Null],
            vec![Value::Text("unsupported".into())],
            vec![Value::Nat64(1), Value::Unit],
            vec![Value::List(Vec::new())],
            vec![Value::Null; 5],
        ];
        for values in cases {
            let boundary = CursorBoundary {
                slots: values
                    .iter()
                    .cloned()
                    .map(CursorBoundarySlot::Present)
                    .collect(),
            };
            let structural = match values.as_slice() {
                [value] => value.clone(),
                _ => Value::List(values),
            };
            let expected = DecodedDataStoreKey::try_from_structural_key(ENTITY, &structural);
            let actual = primary_key_boundary_data_key(ENTITY, &boundary);
            match (expected, actual) {
                (Ok(expected), Ok(actual)) => {
                    assert_eq!(actual, expected);
                    assert_eq!(actual.to_raw().unwrap(), expected.to_raw().unwrap());
                }
                (Err(expected), Err(actual)) => {
                    assert_eq!(actual.diagnostic(), expected.diagnostic());
                    assert_eq!(actual.diagnostic_facts(), expected.diagnostic_facts());
                }
                results => panic!("boundary and structural key results differ: {results:?}"),
            }
            for position in 0..=boundary.slots.len() {
                let mut missing = boundary.clone();
                missing.slots.insert(position, CursorBoundarySlot::Missing);
                let error = primary_key_boundary_data_key(ENTITY, &missing).unwrap_err();
                assert_eq!(
                    error.diagnostic(),
                    InternalError::query_executor_invariant().diagnostic()
                );
            }
            let error =
                primary_key_boundary_value(&boundary, boundary.slots.len() + 1).unwrap_err();
            assert_eq!(
                error.diagnostic(),
                InternalError::query_executor_invariant().diagnostic()
            );
        }
    }

    #[test]
    fn physical_polling_preserves_limits_across_chunk_boundaries() {
        reset_heap_stores();
        load_primary_keys();
        load_index_keys(&IndexId::new(ENTITY, 1), b"lane");
        for direction in [Direction::Asc, Direction::Desc] {
            for limit in [
                Some(0),
                Some(1),
                Some(63),
                Some(64),
                Some(65),
                Some(100),
                None,
            ] {
                let mut expected = (1..=100).map(data_key).collect::<Vec<_>>();
                if direction == Direction::Desc {
                    expected.reverse();
                }
                if let Some(limit) = limit {
                    expected.truncate(limit);
                }
                let mut primary =
                    PrimaryRangeKeyStream::new_full_scan(STORE, ENTITY, direction, limit).unwrap();
                let mut index = ordered_index_stream(direction, limit, None);
                assert_eq!(collect_keys(&mut primary), expected);
                assert_eq!(collect_keys(&mut index), expected);
            }
        }
    }

    #[test]
    fn direct_primary_visits_preserve_bounded_progress_after_early_stop() {
        for (composite, direction) in [
            (false, Direction::Asc),
            (false, Direction::Desc),
            (true, Direction::Asc),
            (true, Direction::Desc),
        ] {
            reset_heap_stores();
            let keys = (1..=100)
                .map(|value| {
                    if composite {
                        DecodedDataStoreKey::new_primary_key_value(
                            ENTITY,
                            &PrimaryKeyValue::Composite(
                                CompositePrimaryKeyValue::try_from_components(&[
                                    PrimaryKeyComponent::Nat64(value),
                                    PrimaryKeyComponent::Nat64(500 + value),
                                ])
                                .unwrap(),
                            ),
                        )
                    } else {
                        data_key(value)
                    }
                })
                .collect::<Vec<_>>();
            DATA.with_borrow_mut(|store| {
                for key in &keys {
                    store.insert_raw_for_test(
                        key.to_raw().unwrap(),
                        RawRow::try_new(vec![0]).unwrap(),
                    );
                }
            });
            for limit in [Some(0), Some(1), Some(64), Some(65), None] {
                for stop_after in [0, 1, 63, 64, 65, 100] {
                    for stop_in_visitor in [false, true] {
                        let mut stream = PrimaryRangeKeyStream::new(
                            STORE,
                            keys[9].clone(),
                            keys[89].clone(),
                            direction,
                            limit,
                        )
                        .unwrap();
                        let visited = Cell::new(0);
                        let mut rows = Vec::new();
                        assert_eq!(
                            stream
                                .try_visit_rows_direct(
                                    &mut || Ok(visited.get() < stop_after),
                                    &mut |key, _| {
                                        visited.set(visited.get() + 1);
                                        rows.push(key);
                                        Ok(if stop_in_visitor && visited.get() == stop_after {
                                            StoreVisit::Stop
                                        } else {
                                            StoreVisit::Continue
                                        })
                                    },
                                )
                                .unwrap(),
                            Some(())
                        );
                        let mut expected = keys[9..90].to_vec();
                        if matches!(direction, Direction::Desc) {
                            expected.reverse();
                        }
                        if let Some(limit) = limit {
                            expected.truncate(limit);
                        }
                        assert_eq!(
                            rows,
                            expected
                                .iter()
                                .take(stop_after)
                                .cloned()
                                .collect::<Vec<_>>()
                        );
                        rows.extend(collect_keys(&mut stream));
                        assert_eq!(rows, expected);
                    }
                }
            }
        }
    }

    #[test]
    fn drained_primary_chunk_does_not_reopen_first_pull_admission() {
        reset_heap_stores();
        load_primary_keys();
        let mut stream =
            PrimaryRangeKeyStream::new_full_scan(STORE, ENTITY, Direction::Asc, None).unwrap();
        stream.chunk_entries = 1;
        assert_eq!(stream.next_key().unwrap(), Some(data_key(1)));
        assert!(stream.resume_strictly_after(data_key(50)).is_err());
        let visits = Cell::new(0);
        assert_eq!(
            stream
                .try_visit_rows_direct(
                    &mut || {
                        visits.set(visits.get() + 1);
                        Ok(true)
                    },
                    &mut |_, _| {
                        visits.set(visits.get() + 1);
                        Ok(StoreVisit::Continue)
                    },
                )
                .unwrap(),
            None
        );
        assert_eq!(visits.get(), 0);
        assert_eq!(
            collect_keys(&mut stream),
            (2..=100).map(data_key).collect::<Vec<_>>()
        );
    }

    #[test]
    fn primary_polling_preserves_malformed_key_rejection() {
        reset_heap_stores();
        let mut bytes = data_key(50).to_raw().unwrap().as_bytes().to_vec();
        bytes.pop();
        DATA.with_borrow_mut(|store| {
            store.insert_raw_for_test(
                RawDataStoreKey::from_persisted_bytes(bytes),
                RawRow::try_new(vec![0]).unwrap(),
            );
        });
        for direction in [Direction::Asc, Direction::Desc] {
            let mut stream =
                PrimaryRangeKeyStream::new_full_scan(STORE, ENTITY, direction, None).unwrap();
            let error = stream
                .next_key()
                .expect_err("malformed primary keys must fail before emission");
            assert_eq!(error.class(), crate::error::ErrorClass::Corruption);
        }
    }

    #[test]
    fn physical_polling_preserves_sparse_cursor_suffixes_and_limits() {
        // The oracle orders physical membership independently of chunk/refill
        // state, including cursor targets absent from the stored population.
        for seed in 1_u64..=8 {
            reset_heap_stores();
            let mut state = seed;
            let mut values = (0..48)
                .map(|_| {
                    state = state
                        .wrapping_mul(6_364_136_223_846_793_005)
                        .wrapping_add(1);
                    (state >> 32) % 100 + 1
                })
                .collect::<Vec<_>>();
            values.sort_unstable();
            values.dedup();
            let index_id = IndexId::new(ENTITY, 1);
            for &value in &values {
                DATA.with_borrow_mut(|store| {
                    store.insert_raw_for_test(
                        data_key(value).to_raw().unwrap(),
                        RawRow::try_new(vec![0]).unwrap(),
                    );
                });
                INDEX.with_borrow_mut(|store| {
                    store.insert(
                        index_key(&index_id, b"lane", value).to_raw().unwrap(),
                        IndexEntryValue::presence(),
                    );
                });
            }
            for direction in [Direction::Asc, Direction::Desc] {
                for limit in [None, Some(3)] {
                    for boundary in [0, seed.wrapping_mul(17) % 100, 101] {
                        let mut expected = values
                            .iter()
                            .copied()
                            .filter(|value| match direction {
                                Direction::Asc => *value > boundary,
                                Direction::Desc => *value < boundary,
                            })
                            .map(data_key)
                            .collect::<Vec<_>>();
                        if direction == Direction::Desc {
                            expected.reverse();
                        }
                        if let Some(limit) = limit {
                            expected.truncate(limit);
                        }
                        let mut primary = PrimaryRangeKeyStream::new(
                            STORE,
                            data_key(1),
                            data_key(100),
                            direction,
                            limit,
                        )
                        .unwrap();
                        primary.resume_strictly_after(data_key(boundary)).unwrap();
                        let anchor = index_key(&index_id, b"lane", boundary).to_raw().unwrap();
                        let mut index = ordered_index_stream(direction, limit, Some(anchor));
                        assert_eq!(collect_keys(&mut primary), expected);
                        assert_eq!(collect_keys(&mut index), expected);
                    }
                }
            }
        }
    }

    #[test]
    fn index_polling_preserves_duplicate_physical_occurrences() {
        reset_heap_stores();
        let index_id = IndexId::new(ENTITY, 1);
        for (component, value) in [(b"a", 1), (b"b", 1), (b"c", 2)] {
            INDEX.with_borrow_mut(|store| {
                store.insert(
                    index_key(&index_id, component, value).to_raw().unwrap(),
                    IndexEntryValue::presence(),
                );
            });
        }
        for direction in [Direction::Asc, Direction::Desc] {
            let mut stream = IndexRangeKeyStream::new(
                STORE,
                ENTITY,
                (
                    Bound::Included(index_key(&index_id, b"a", 1).to_raw().unwrap()),
                    Bound::Included(index_key(&index_id, b"c", 2).to_raw().unwrap()),
                ),
                direction,
                None,
                None,
                1,
            );
            let mut expected = vec![data_key(1), data_key(1), data_key(2)];
            if direction == Direction::Desc {
                expected.reverse();
            }
            assert_eq!(collect_keys(&mut stream), expected);
        }
    }

    #[test]
    fn primary_polling_preserves_one_sided_bounds() {
        reset_heap_stores();
        load_primary_keys();
        for (start, end, direction, range) in [
            (Some(data_key(40)), None, Direction::Asc, 40..=100),
            (Some(data_key(40)), None, Direction::Desc, 40..=100),
            (None, Some(data_key(60)), Direction::Asc, 1..=60),
            (None, Some(data_key(60)), Direction::Desc, 1..=60),
        ] {
            let mut stream =
                PrimaryRangeKeyStream::new_bounds(STORE, ENTITY, start, end, direction, None)
                    .unwrap();
            let mut expected = range.map(data_key).collect::<Vec<_>>();
            if direction == Direction::Desc {
                expected.reverse();
            }
            assert_eq!(collect_keys(&mut stream), expected);
        }
    }

    #[test]
    fn physical_polling_hard_budget_rejects_refill_before_progress() {
        reset_heap_stores();
        load_primary_keys();
        load_index_keys(&IndexId::new(ENTITY, 1), b"lane");
        for direction in [Direction::Asc, Direction::Desc] {
            let primary =
                PrimaryRangeKeyStream::new(STORE, data_key(1), data_key(100), direction, None)
                    .unwrap();
            let index = ordered_index_stream(direction, None, None);
            for mut stream in [
                OrderedKeyStreamBox::primary_range(primary),
                OrderedKeyStreamBox::index_range(index),
            ] {
                let budget = HardExecutionBudget::uniform_for_tests(
                    u64::MAX,
                    HardExecutionFailureHeadroom::new(500, 256),
                )
                .with_limit_for_tests(DiagnosticExecutionBudgetResource::KeyIndexEntriesVisited, 0);
                let context = HardExecutionContext::new(
                    DiagnosticExecutionBudgetScope::Execution,
                    DiagnosticExecutionLane::TrustedRead,
                    0x706f_6c6c,
                );
                let error = with_query_execution_budget_for_tests(budget, context, || {
                    stream.next_key().map_err(QueryError::execute)
                })
                .expect_err("physical refill must obey the hard budget");
                assert!(matches!(
                    error.diagnostic().detail(),
                    Some(DiagnosticDetail::RuntimeBoundary {
                        boundary: RuntimeBoundaryCode::ExecutionBudgetExceeded
                    })
                ));
                let first = if direction == Direction::Asc { 1 } else { 100 };
                assert_eq!(stream.next_key().unwrap(), Some(data_key(first)));
            }
        }
    }

    #[test]
    fn materialized_primary_keys_resume_strictly_in_both_directions() {
        let runtime = KeyAccessRuntime::new(STORE, ENTITY);
        for composite in [false, true] {
            let key_value = |key| {
                if composite {
                    Value::List(vec![Value::Nat64(7), Value::Nat64(key)])
                } else {
                    Value::Nat64(key)
                }
            };
            let keys = [3, 1, 2, 3].map(key_value);
            for path in [
                ExecutionPathPayload::ByKey(&keys[0]),
                ExecutionPathPayload::ByKeys(&keys),
            ] {
                for direction in [Direction::Asc, Direction::Desc] {
                    for boundary_value in [0, 2, 3, 4] {
                        let boundary = CursorBoundary {
                            slots: if composite {
                                vec![
                                    CursorBoundarySlot::Present(Value::Nat64(7)),
                                    CursorBoundarySlot::Present(Value::Nat64(boundary_value)),
                                ]
                            } else {
                                vec![CursorBoundarySlot::Present(Value::Nat64(boundary_value))]
                            },
                        };
                        let mut stream = resolve_physical_key_stream(
                            &path,
                            PhysicalStreamBindings {
                                index_prefix_specs: &[],
                                index_range_spec: None,
                                continuation:
                                    AccessScanContinuationInput::with_primary_key_boundary(
                                        None,
                                        direction,
                                        Some(&boundary),
                                    ),
                                execution_policy: AccessStreamExecutionPolicy::canonical_key_order(
                                    None,
                                ),
                                index_predicate_execution: None,
                                index_prefix_child_expansion: None,
                            },
                            &runtime,
                        )
                        .unwrap();
                        let mut expected = (1..=3)
                            .filter(|key| {
                                !matches!(path, ExecutionPathPayload::ByKey(_)) || *key == 3
                            })
                            .filter(|key| match direction {
                                Direction::Asc => *key > boundary_value,
                                Direction::Desc => *key < boundary_value,
                            })
                            .map(|key| {
                                DecodedDataStoreKey::try_from_structural_key(
                                    ENTITY,
                                    &key_value(key),
                                )
                                .unwrap()
                            })
                            .collect::<Vec<_>>();
                        if direction == Direction::Desc {
                            expected.reverse();
                        }
                        for key in expected {
                            assert_eq!(stream.next_key().unwrap(), Some(key));
                        }
                        assert_eq!(stream.next_key().unwrap(), None);
                    }
                }
            }
        }
    }

    #[test]
    fn limited_primary_leaf_applies_page_fetch_hint_after_resume_boundary() {
        reset_heap_stores();
        load_primary_keys();

        for (direction, expected) in [
            (Direction::Asc, [51, 52, 53]),
            (Direction::Desc, [49, 48, 47]),
        ] {
            let mut stream =
                PrimaryRangeKeyStream::new(STORE, data_key(1), data_key(100), direction, Some(3))
                    .expect("limited primary stream should build");
            stream
                .resume_strictly_after(data_key(50))
                .expect("primary continuation boundary should apply");

            for expected_key in expected {
                assert_eq!(
                    stream.next_key().expect("resumed key should load"),
                    Some(data_key(expected_key)),
                );
            }
            assert_eq!(
                stream.next_key().expect("fetch hint should end the page"),
                None,
            );
        }
    }

    #[test]
    fn primary_leaf_caps_refills_to_the_active_page_entry_limit() {
        for (page_limit, expected_refill) in [(4, 4), (100, 64), (0, 1)] {
            let envelope = crate::db::executor::PageWorkEnvelope::default_scalar()
                .with_limit_for_tests(
                    DiagnosticExecutionBudgetResource::KeyIndexEntriesVisited,
                    page_limit,
                );
            let bounded = crate::db::executor::with_production_scalar_page_work(envelope, || {
                PrimaryRangeKeyStream::new(STORE, data_key(1), data_key(100), Direction::Asc, None)
            })
            .expect("primary leaf should inherit the active page envelope");

            assert_eq!(bounded.value.next_pull_entry_bound(), expected_refill);
        }
    }

    #[test]
    fn index_leaf_caps_refills_to_its_active_page_branch_share() {
        for (page_limit, branch_count, expected_refill) in
            [(4, 1, 4), (4, 2, 2), (100, 1, 64), (0, 1, 1)]
        {
            let envelope = crate::db::executor::PageWorkEnvelope::default_scalar()
                .with_limit_for_tests(
                    DiagnosticExecutionBudgetResource::KeyIndexEntriesVisited,
                    page_limit,
                );
            let bounded = crate::db::executor::with_production_scalar_page_work(envelope, || {
                index_leaf_chunk_entries_for_active_page(ACCESS_SCAN_CHUNK_ENTRIES, branch_count)
            })
            .expect("index leaf should inherit its share of the active page envelope");

            assert_eq!(bounded.value, expected_refill);
        }
    }

    #[test]
    fn materialized_secondary_index_keys_charge_complete_temporary_backing() {
        let budget = HardExecutionBudget::uniform_for_tests(
            u64::MAX,
            HardExecutionFailureHeadroom::new(500, 256),
        )
        .with_limit_for_tests(DiagnosticExecutionBudgetResource::TemporaryBytes, 0);
        let context = HardExecutionContext::new(
            DiagnosticExecutionBudgetScope::Execution,
            DiagnosticExecutionLane::TrustedRead,
            0x6d61_7465_7269_616c,
        );
        let keys = vec![data_key(1)];
        let error = with_query_execution_budget_for_tests(budget, context, || {
            charge_materialized_secondary_index_keys(keys.as_slice(), keys.capacity())
                .map_err(QueryError::execute)
        })
        .expect_err("materialized key-vector capacity must consume the temporary-byte budget");

        assert!(matches!(
            error.diagnostic().detail(),
            Some(DiagnosticDetail::RuntimeBoundary {
                boundary: RuntimeBoundaryCode::ExecutionBudgetExceeded,
            })
        ));
        assert_eq!(
            error.diagnostic_facts()[0],
            (
                DiagnosticFactTag::BudgetResource,
                DiagnosticExecutionBudgetResource::TemporaryBytes.raw(),
            ),
        );
    }

    #[test]
    fn general_intersection_materialization_preserves_order_and_finite_page_bound() {
        for direction in [Direction::Asc, Direction::Desc] {
            let comparator = KeyOrderComparator::from_direction(direction);
            let streams = [vec![1, 2, 4], vec![2, 3, 4]]
                .into_iter()
                .map(|mut ids| {
                    if direction == Direction::Desc {
                        ids.reverse();
                    }
                    OrderedKeyStreamBox::materialized(ids.into_iter().map(data_key).collect())
                })
                .collect();
            let stream = OrderedKeyStreamBox::intersect_all(streams, comparator).unwrap();
            assert_eq!(stream.page_access_entry_bound(), None);
            let mut materialized = materialize_ordered_key_stream(stream).unwrap();
            assert_eq!(materialized.page_access_entry_bound(), Some(0));
            let expected = if direction == Direction::Asc {
                [2, 4]
            } else {
                [4, 2]
            };
            assert_eq!(collect_keys(&mut materialized), expected.map(data_key));
        }
    }

    #[test]
    fn general_intersection_materialization_retains_key_and_memory_budget_failures() {
        load_primary_keys();
        for resource in [
            DiagnosticExecutionBudgetResource::KeyIndexEntriesVisited,
            DiagnosticExecutionBudgetResource::TemporaryBytes,
        ] {
            let stream = if resource == DiagnosticExecutionBudgetResource::TemporaryBytes {
                // Prebuilt input isolates the collector's retained-memory charge.
                OrderedKeyStreamBox::materialized(vec![data_key(1)])
            } else {
                OrderedKeyStreamBox::primary_range(
                    PrimaryRangeKeyStream::new(
                        STORE,
                        data_key(1),
                        data_key(100),
                        Direction::Asc,
                        None,
                    )
                    .unwrap(),
                )
            };
            let budget = HardExecutionBudget::uniform_for_tests(
                u64::MAX,
                HardExecutionFailureHeadroom::new(500, 256),
            )
            .with_limit_for_tests(resource, 0);
            let context = HardExecutionContext::new(
                DiagnosticExecutionBudgetScope::Execution,
                DiagnosticExecutionLane::TrustedRead,
                0x696e_7465_7273_6563,
            );
            let error = with_query_execution_budget_for_tests(budget, context, || {
                materialize_ordered_key_stream(stream)
                    .map(|_| ())
                    .map_err(QueryError::execute)
            })
            .expect_err("blocking materialization must retain hard budget refusal");
            assert!(matches!(
                error.diagnostic().detail(),
                Some(DiagnosticDetail::RuntimeBoundary {
                    boundary: RuntimeBoundaryCode::ExecutionBudgetExceeded,
                })
            ));
            assert!(
                error
                    .diagnostic_facts()
                    .contains(&(DiagnosticFactTag::BudgetResource, resource.raw(),))
            );
        }
    }

    #[test]
    fn index_leaf_propagates_malformed_entry_corruption() {
        reset_heap_stores();
        let index_id = IndexId::new(ENTITY, 1);
        let component = b"lane";
        let raw_key = index_key(&index_id, component, 50)
            .to_raw()
            .expect("index key should encode");
        INDEX.with_borrow_mut(|store| {
            store.insert(
                raw_key.clone(),
                <IndexEntryValue as Storable>::from_bytes(Cow::Owned(vec![0xff])),
            );
        });
        let mut stream = IndexRangeKeyStream::new(
            STORE,
            ENTITY,
            (Bound::Included(raw_key.clone()), Bound::Included(raw_key)),
            Direction::Asc,
            None,
            None,
            ACCESS_SCAN_CHUNK_ENTRIES,
        );
        let error = stream
            .next_key()
            .expect_err("malformed index entries must fail");
        assert_eq!(error.class(), crate::error::ErrorClass::Corruption);
    }
}