icydb-core 0.222.0

//! Module: executor::stream::access::traversal
//! Responsibility: build and execute access-path traversal streams for runtime loading.
//! Does not own: access-plan construction or planner routing semantics.
//! Boundary: lowers executable access contracts into ordered key/data stream traversal.

use crate::{
    db::{
        data::{DecodedDataStoreKey, RawDataStoreKey},
        executor::{
            ExecutableAccessNode, ExecutableAccessPlan, ExecutionPathPayload,
            LoweredIndexPrefixSpec, LoweredIndexRangeSpec,
            budget::charge_current_execution_budget,
            lowered_index_prefix_exact_cardinality,
            pipeline::contracts::{AccessScanContinuationInput, AccessStreamBindings},
            route::IndexPrefixChildExpansionHint,
            stream::{
                access::{
                    bindings::{
                        AccessSpecCursor, AccessStreamExecutionPolicy, ExecutableAccess,
                        IndexLeafOrderPolicy, IndexStreamConstraints,
                    },
                    physical,
                },
                key::{
                    KeyOrderComparator, OrderedKeyStreamBox,
                    ordered_key_stream_from_materialized_keys,
                },
            },
            traversal::IndexRangeTraversalContract,
        },
        index::predicate::IndexPredicateExecution,
    },
    error::InternalError,
    value::Value,
};
use icydb_diagnostic_code::DiagnosticExecutionBudgetResource;
use std::mem::size_of;

///
/// TraversalInputs
///
/// TraversalInputs carries the structural traversal bindings needed by
/// access-plan stream resolution.
/// This deliberately excludes typed context so recursive traversal orchestration
/// can stay monomorphic while physical path resolution remains in typed leaves.
///

#[derive(Clone, Copy)]
struct TraversalInputs<'a> {
    index_prefix_specs: &'a [LoweredIndexPrefixSpec],
    index_range_specs: &'a [LoweredIndexRangeSpec],
    continuation: AccessScanContinuationInput<'a>,
    execution_policy: AccessStreamExecutionPolicy,
    index_predicate_execution: Option<crate::db::index::predicate::IndexPredicateExecution<'a>>,
    index_prefix_child_expansion: Option<IndexPrefixChildExpansionHint>,
}

#[cfg(test)]
mod exact_intersection_tests {
    use super::{AccessPlanStreamResolver, ExactIntersectionPreflight};

    fn preflight(child_cardinalities: &[u64]) -> ExactIntersectionPreflight {
        ExactIntersectionPreflight {
            child_cardinalities: child_cardinalities.to_vec(),
            total_cardinality: child_cardinalities.iter().sum(),
        }
    }

    #[test]
    fn worst_case_cost_gate_accepts_sparse_fixtures_and_rejects_dense_ties() {
        assert!(
            AccessPlanStreamResolver::exact_intersection_probe_can_beat_single(&preflight(&[
                21, 20
            ]),)
        );
        assert!(
            AccessPlanStreamResolver::exact_intersection_probe_can_beat_single(&preflight(&[
                120, 21, 20
            ]),)
        );
        assert!(
            !AccessPlanStreamResolver::exact_intersection_probe_can_beat_single(&preflight(&[
                20, 20
            ]),)
        );
    }

    #[test]
    fn overflowed_cost_authority_fails_closed() {
        assert!(
            !AccessPlanStreamResolver::exact_intersection_cost_beats_single(
                &ExactIntersectionPreflight {
                    child_cardinalities: vec![u64::MAX, 1],
                    total_cardinality: u64::MAX,
                },
                1,
            )
        );
    }
}

impl<'a> TraversalInputs<'a> {
    // Clone this traversal envelope with one overridden physical fetch hint.
    const fn with_physical_fetch_hint(self, physical_fetch_hint: Option<usize>) -> Self {
        Self {
            execution_policy: self
                .execution_policy
                .with_physical_fetch_hint(physical_fetch_hint),
            ..self
        }
    }

    // Composite child streams must stay canonicalized by `DecodedDataStoreKey` order so
    // merge/intersection reducers can consume them under one shared key comparator.
    const fn without_leaf_index_order_preservation(self) -> Self {
        Self {
            execution_policy: self
                .execution_policy
                .with_index_leaf_order_policy(IndexLeafOrderPolicy::CanonicalKey),
            ..self
        }
    }

    // Exact-prefix intersection leaves retain physical primary-key suffix order
    // so the bounded overlap probe can intersect them without reordering.
    const fn with_physical_leaf_order(self) -> Self {
        Self {
            execution_policy: self
                .execution_policy
                .with_index_leaf_order_policy(IndexLeafOrderPolicy::PreservePhysicalLeaf),
            ..self
        }
    }

    // Build one mutable spec-consumption cursor over prefix/range slices.
    const fn spec_cursor(&self) -> AccessSpecCursor<'a> {
        AccessSpecCursor::new(self.index_prefix_specs, self.index_range_specs)
    }
}

// Keep the historical traversal-layer invariant name stable for CI checks while
// routing the actual contract enforcement through the traversal owner.
fn validate_index_range_spec_alignment(
    path: &ExecutionPathPayload<'_, Value>,
    index_range_spec: Option<&LoweredIndexRangeSpec>,
) -> Result<(), InternalError> {
    IndexRangeTraversalContract::validate_spec_alignment(path, index_range_spec)
}

///
/// TraversalRuntime
///
/// TraversalRuntime carries the store/index authority
/// needed to resolve planner-key executable access paths without recovering
/// `Context<'_, E>` inside the execution hot path.
/// It is the fast-path runtime leaf used by erased execution
/// adapters and typed context shells alike.
///

#[derive(Clone, Copy)]
pub(in crate::db::executor) struct TraversalRuntime {
    pub(in crate::db::executor) store: crate::db::registry::StoreHandle,
    pub(in crate::db::executor) entity_tag: crate::types::EntityTag,
}

impl TraversalRuntime {
    /// Build one traversal runtime from canonical store authority.
    #[must_use]
    pub(in crate::db::executor) const fn new(
        store: crate::db::registry::StoreHandle,
        entity_tag: crate::types::EntityTag,
    ) -> Self {
        Self { store, entity_tag }
    }

    /// Resolve one executable access binding into an ordered key stream.
    pub(in crate::db::executor) fn ordered_key_stream_from_runtime_access(
        &self,
        request: ExecutableAccess<'_, Value>,
    ) -> Result<OrderedKeyStreamBox, InternalError> {
        self.ordered_key_stream_from_executable_plan(
            &request.plan,
            request.bindings,
            request.execution_policy,
            request.index_predicate_execution,
        )
    }

    /// Resolve one borrowed executable access plan plus bindings into an
    /// ordered key stream without cloning the access plan wrapper.
    pub(in crate::db::executor) fn ordered_key_stream_from_executable_plan<'input>(
        &self,
        plan: &ExecutableAccessPlan<'_, Value>,
        bindings: AccessStreamBindings<'input>,
        execution_policy: AccessStreamExecutionPolicy,
        index_predicate_execution: Option<IndexPredicateExecution<'input>>,
    ) -> Result<OrderedKeyStreamBox, InternalError> {
        let inputs = TraversalInputs {
            index_prefix_specs: bindings.index_prefix_specs,
            index_range_specs: bindings.index_range_specs,
            continuation: bindings.continuation,
            execution_policy,
            index_predicate_execution,
            index_prefix_child_expansion: bindings.index_prefix_child_expansion,
        };
        let mut spec_cursor = inputs.spec_cursor();
        let key_stream =
            AccessPlanStreamResolver::produce_key_stream(self, plan, inputs, &mut spec_cursor)?;
        spec_cursor.validate_consumed()?;

        Ok(key_stream)
    }

    // Resolve one executable path leaf through the structural physical access
    // boundary without re-erasing the traversal runtime behind a local trait.
    fn lower_path_access(
        &self,
        path: &ExecutionPathPayload<'_, Value>,
        inputs: TraversalInputs<'_>,
        index_prefix_specs: &[LoweredIndexPrefixSpec],
        index_range_spec: Option<&LoweredIndexRangeSpec>,
    ) -> Result<OrderedKeyStreamBox, InternalError> {
        let constraints = IndexStreamConstraints {
            prefixes: index_prefix_specs,
            range: index_range_spec,
        };
        path.resolve_structural_physical_key_stream(physical::StructuralPhysicalStreamRequest {
            store: self.store,
            entity_tag: self.entity_tag,
            index_prefix_specs: constraints.prefixes,
            index_range_spec: constraints.range,
            continuation: inputs.continuation,
            execution_policy: inputs.execution_policy,
            index_predicate_execution: inputs.index_predicate_execution,
            index_prefix_child_expansion: inputs.index_prefix_child_expansion,
        })
    }
}

///
/// AccessPlanStreamResolver
///
/// Executor-owned access-plan traversal and key-stream production.
/// This isolates physical stream wiring from `AccessPlan` so plan types remain
/// data-only while executor mechanics stay in executor modules.
///

struct AccessPlanStreamResolver;

const MAX_ATOMIC_EXACT_INTERSECTION_ENTRIES: u64 = 256;
const MAX_ATOMIC_EXACT_INTERSECTION_KEY_BYTES: u64 =
    MAX_ATOMIC_EXACT_INTERSECTION_ENTRIES * RawDataStoreKey::MAX_STORED_SIZE_BYTES;
const INTERSECTION_ROW_READ_COST_WEIGHT: u64 = 32;

struct ExactIntersectionPreflight {
    child_cardinalities: Vec<u64>,
    total_cardinality: u64,
}

enum ExactIntersectionAdmission {
    NotApplicable,
    ConservativeFallback,
    ProvenEmpty,
    Probe(ExactIntersectionPreflight),
}

impl AccessPlanStreamResolver {
    // Validate that a consumed prefix spec belongs to the same index path node.
    fn validate_index_prefix_spec_alignment(
        path: &ExecutionPathPayload<'_, Value>,
        index_prefix_specs: &[LoweredIndexPrefixSpec],
    ) -> Result<(), InternalError> {
        let path_facts = path.shape_facts();
        if let Some(details) = path_facts.index_prefix_details() {
            for spec in index_prefix_specs {
                if spec.scan_contract().name() != details.name() {
                    return Err(InternalError::query_executor_invariant());
                }
            }
        }

        Ok(())
    }

    // Collect one child key stream for each child access plan.
    fn collect_child_key_streams(
        runtime: &TraversalRuntime,
        children: &[ExecutableAccessPlan<'_, Value>],
        inputs: TraversalInputs<'_>,
        spec_cursor: &mut AccessSpecCursor<'_>,
    ) -> Result<Vec<OrderedKeyStreamBox>, InternalError> {
        let mut streams = Vec::with_capacity(children.len());
        for child in children {
            // Composite plans never need physical fetch-hint expansion on child lookups.
            let child_inputs = inputs
                .with_physical_fetch_hint(None)
                .without_leaf_index_order_preservation();
            streams.push(Self::produce_key_stream(
                runtime,
                child,
                child_inputs,
                spec_cursor,
            )?);
        }

        Ok(streams)
    }

    // Collect direct exact-prefix children while retaining their accepted
    // primary-key suffix order for one bounded overlap probe.
    fn collect_exact_intersection_child_streams(
        runtime: &TraversalRuntime,
        children: &[ExecutableAccessPlan<'_, Value>],
        inputs: TraversalInputs<'_>,
        spec_cursor: &mut AccessSpecCursor<'_>,
    ) -> Result<Vec<OrderedKeyStreamBox>, InternalError> {
        let mut streams = Vec::with_capacity(children.len());
        for child in children {
            let child_inputs = inputs
                .with_physical_fetch_hint(None)
                .with_physical_leaf_order();
            streams.push(Self::produce_key_stream(
                runtime,
                child,
                child_inputs,
                spec_cursor,
            )?);
        }

        Ok(streams)
    }

    fn exact_intersection_admission(
        runtime: &TraversalRuntime,
        children: &[ExecutableAccessPlan<'_, Value>],
        inputs: TraversalInputs<'_>,
        spec_cursor: AccessSpecCursor<'_>,
    ) -> ExactIntersectionAdmission {
        if !(2..=3).contains(&children.len()) || inputs.index_predicate_execution.is_some() {
            return ExactIntersectionAdmission::NotApplicable;
        }

        let mut metadata_cursor = spec_cursor;
        let mut child_cardinalities = Vec::with_capacity(children.len());
        let mut total_cardinality = 0u64;
        for child in children {
            let ExecutableAccessNode::Path(path) = child.node() else {
                return ExactIntersectionAdmission::NotApplicable;
            };
            let ExecutionPathPayload::IndexPrefix { .. } = path else {
                return ExactIntersectionAdmission::NotApplicable;
            };
            let path_facts = path.shape_facts();
            if path_facts.index_prefix_spec_count() != 1 || path_facts.consumes_index_range_spec() {
                return ExactIntersectionAdmission::NotApplicable;
            }
            let Some(spec) = metadata_cursor
                .next_index_prefix_specs(1)
                .and_then(|specs| specs.first())
            else {
                return ExactIntersectionAdmission::ConservativeFallback;
            };
            let Some(cardinality) = lowered_index_prefix_exact_cardinality(runtime.store, spec)
            else {
                return ExactIntersectionAdmission::ConservativeFallback;
            };
            if cardinality == 0 {
                return ExactIntersectionAdmission::ProvenEmpty;
            }
            let Some(next_total) = total_cardinality.checked_add(cardinality) else {
                return ExactIntersectionAdmission::ConservativeFallback;
            };
            if next_total > MAX_ATOMIC_EXACT_INTERSECTION_ENTRIES {
                return ExactIntersectionAdmission::ConservativeFallback;
            }
            total_cardinality = next_total;
            child_cardinalities.push(cardinality);
        }

        ExactIntersectionAdmission::Probe(ExactIntersectionPreflight {
            child_cardinalities,
            total_cardinality,
        })
    }

    fn exact_intersection_cost_beats_single(
        preflight: &ExactIntersectionPreflight,
        overlap_cardinality: u64,
    ) -> bool {
        let Some(single_cardinality) = preflight.child_cardinalities.first().copied() else {
            return false;
        };
        let Some(single_row_cost) = single_cardinality
            .checked_mul(INTERSECTION_ROW_READ_COST_WEIGHT)
            .and_then(|row_cost| row_cost.checked_add(single_cardinality))
        else {
            return false;
        };
        let Some(intersection_cost) = overlap_cardinality
            .checked_mul(INTERSECTION_ROW_READ_COST_WEIGHT)
            .and_then(|row_cost| row_cost.checked_add(preflight.total_cardinality))
        else {
            return false;
        };

        intersection_cost < single_row_cost
    }

    fn exact_intersection_probe_can_beat_single(preflight: &ExactIntersectionPreflight) -> bool {
        let Some(maximum_overlap) = preflight.child_cardinalities.iter().copied().min() else {
            return false;
        };

        Self::exact_intersection_cost_beats_single(preflight, maximum_overlap)
    }

    fn collect_exact_intersection_overlap(
        streams: Vec<OrderedKeyStreamBox>,
        comparator: KeyOrderComparator,
        preflight: &ExactIntersectionPreflight,
    ) -> Result<Vec<DecodedDataStoreKey>, InternalError> {
        let maximum_keys_u64 = preflight
            .child_cardinalities
            .iter()
            .copied()
            .min()
            .ok_or_else(InternalError::executor_invariant)?;
        let maximum_keys =
            usize::try_from(maximum_keys_u64).map_err(|_| InternalError::executor_invariant())?;
        let slot_bytes = maximum_keys
            .checked_mul(size_of::<DecodedDataStoreKey>())
            .ok_or_else(InternalError::executor_invariant)?;
        charge_current_execution_budget(
            DiagnosticExecutionBudgetResource::TemporaryBytes,
            u64::try_from(slot_bytes).unwrap_or(u64::MAX),
        )?;

        let mut intersection = OrderedKeyStreamBox::intersect_all(streams, comparator);
        let mut overlap = Vec::with_capacity(maximum_keys);
        let mut retained_key_bytes = 0u64;
        while let Some(key) = intersection.next_key()? {
            if overlap.len() >= maximum_keys {
                return Err(InternalError::executor_invariant());
            }
            let key_bytes = u64::try_from(key.raw_key()?.as_bytes().len()).unwrap_or(u64::MAX);
            retained_key_bytes = retained_key_bytes
                .checked_add(key_bytes)
                .ok_or_else(InternalError::executor_invariant)?;
            if retained_key_bytes > MAX_ATOMIC_EXACT_INTERSECTION_KEY_BYTES {
                return Err(InternalError::executor_invariant());
            }
            charge_current_execution_budget(
                DiagnosticExecutionBudgetResource::TemporaryBytes,
                key_bytes,
            )?;
            overlap.push(key);
        }

        Ok(overlap)
    }

    fn first_stream_or_empty(streams: Vec<OrderedKeyStreamBox>) -> OrderedKeyStreamBox {
        streams
            .into_iter()
            .next()
            .unwrap_or_else(OrderedKeyStreamBox::empty)
    }

    // Build an ordered key stream for this access plan.
    /// Produce one ordered key stream for an access plan while consuming lowered specs.
    fn produce_key_stream(
        runtime: &TraversalRuntime,
        access: &ExecutableAccessPlan<'_, Value>,
        inputs: TraversalInputs<'_>,
        spec_cursor: &mut AccessSpecCursor<'_>,
    ) -> Result<OrderedKeyStreamBox, InternalError> {
        match access.node() {
            ExecutableAccessNode::Path(path) => {
                let path_facts = path.shape_facts();
                let index_prefix_specs = if path_facts.index_prefix_spec_count() > 0 {
                    spec_cursor
                        .require_next_index_prefix_specs(path_facts.index_prefix_spec_count())?
                } else {
                    &[]
                };
                let index_range_spec = if path_facts.consumes_index_range_spec() {
                    Some(spec_cursor.require_next_index_range_spec()?)
                } else {
                    None
                };
                Self::validate_index_prefix_spec_alignment(path, index_prefix_specs)?;
                validate_index_range_spec_alignment(path, index_range_spec)?;

                runtime.lower_path_access(path, inputs, index_prefix_specs, index_range_spec)
            }
            ExecutableAccessNode::Union(children) => {
                Self::produce_union_key_stream(runtime, children, inputs, spec_cursor)
            }
            ExecutableAccessNode::Intersection(children) => {
                Self::produce_intersection_key_stream(runtime, children, inputs, spec_cursor)
            }
        }
    }

    // Build one canonical stream for a union by pairwise-merging child streams.
    fn produce_union_key_stream(
        runtime: &TraversalRuntime,
        children: &[ExecutableAccessPlan<'_, Value>],
        inputs: TraversalInputs<'_>,
        spec_cursor: &mut AccessSpecCursor<'_>,
    ) -> Result<OrderedKeyStreamBox, InternalError> {
        let streams = Self::collect_child_key_streams(runtime, children, inputs, spec_cursor)?;
        let key_comparator = KeyOrderComparator::from_direction(inputs.continuation.direction());

        Ok(OrderedKeyStreamBox::merge_all(streams, key_comparator))
    }

    // Build one canonical stream for an intersection by pairwise-intersecting child streams.
    fn produce_intersection_key_stream(
        runtime: &TraversalRuntime,
        children: &[ExecutableAccessPlan<'_, Value>],
        inputs: TraversalInputs<'_>,
        spec_cursor: &mut AccessSpecCursor<'_>,
    ) -> Result<OrderedKeyStreamBox, InternalError> {
        let key_comparator = KeyOrderComparator::from_direction(inputs.continuation.direction());
        let admission = Self::exact_intersection_admission(runtime, children, inputs, *spec_cursor);
        match admission {
            ExactIntersectionAdmission::NotApplicable => {
                let streams =
                    Self::collect_child_key_streams(runtime, children, inputs, spec_cursor)?;
                Ok(OrderedKeyStreamBox::intersect_all(streams, key_comparator))
            }
            ExactIntersectionAdmission::ConservativeFallback => {
                let streams = Self::collect_exact_intersection_child_streams(
                    runtime,
                    children,
                    inputs,
                    spec_cursor,
                )?;
                Ok(Self::first_stream_or_empty(streams))
            }
            ExactIntersectionAdmission::ProvenEmpty => {
                let _consumed_streams = Self::collect_exact_intersection_child_streams(
                    runtime,
                    children,
                    inputs,
                    spec_cursor,
                )?;
                Ok(OrderedKeyStreamBox::empty())
            }
            ExactIntersectionAdmission::Probe(preflight) => {
                if !Self::exact_intersection_probe_can_beat_single(&preflight) {
                    let streams = Self::collect_exact_intersection_child_streams(
                        runtime,
                        children,
                        inputs,
                        spec_cursor,
                    )?;
                    return Ok(Self::first_stream_or_empty(streams));
                }

                let mut probe_cursor = *spec_cursor;
                let probe_streams = Self::collect_exact_intersection_child_streams(
                    runtime,
                    children,
                    inputs,
                    &mut probe_cursor,
                )?;
                let overlap = Self::collect_exact_intersection_overlap(
                    probe_streams,
                    key_comparator,
                    &preflight,
                )?;
                *spec_cursor = probe_cursor;
                if !Self::exact_intersection_cost_beats_single(
                    &preflight,
                    u64::try_from(overlap.len()).unwrap_or(u64::MAX),
                ) {
                    return Err(InternalError::executor_invariant());
                }

                Ok(ordered_key_stream_from_materialized_keys(overlap))
            }
        }
    }
}