icydb_core/db/query/plan/semantics/

logical.rs

//! Module: query::plan::semantics::logical
//! Responsibility: logical-plan semantic lowering from planner contracts to access-planned queries.
//! Does not own: access-path index selection internals or runtime execution behavior.
//! Boundary: derives planner-owned execution semantics, shape signatures, and continuation policy.

use crate::db::predicate::MissingRowPolicy;
use crate::{
    db::{
        access::{AccessPlan, ExecutableAccessPlan, SemanticIndexKeyItemRef},
        predicate::{IndexCompileTarget, IndexCompileTargetKind, Predicate, PredicateProgram},
        query::plan::{
            AccessPlannedQuery, ContinuationPolicy, DistinctExecutionStrategy,
            EffectiveRuntimeFilterProgram, ExecutionShapeSignature, FieldSlot, GroupPlan,
            GroupedAggregateExecutionSpec, GroupedDistinctExecutionStrategy, GroupedPlanStrategy,
            LogicalPlan, PlannerRouteProfile, PredicatePushdownDiagnostics, QueryMode,
            ResidualFilterContract, ResidualFilterShape, ResolvedOrder, ResolvedOrderField,
            ResolvedOrderValueSource, ScalarPlan, StaticExecutionPlanningContract,
            derive_logical_pushdown_eligibility,
            expr::{
                CompiledExpr, Expr, ProjectionSpec, compile_scalar_projection_expr_with_schema,
                compile_scalar_projection_plan_with_schema,
            },
            extend_unique_grouped_aggregate_specs_from_expr, grouped_aggregate_execution_specs,
            grouped_aggregate_specs_from_projection_spec, grouped_cursor_policy_violation,
            grouped_plan_strategy, lower_data_row_direct_projection_slots_with_schema,
            lower_direct_projection_slots_with_schema, lower_projection_identity,
            lower_projection_intent_with_schema, residual_query_predicate_after_access_path_bounds,
            residual_query_predicate_after_filtered_access_contract,
            resolved_grouped_distinct_execution_strategy_with_schema_info,
        },
        schema::SchemaInfo,
    },
    error::InternalError,
};

impl QueryMode {
    /// True if this mode represents a load intent.
    #[must_use]
    pub const fn is_load(&self) -> bool {
        match self {
            Self::Load(_) => true,
            Self::Delete(_) => false,
        }
    }

    /// True if this mode represents a delete intent.
    #[must_use]
    pub const fn is_delete(&self) -> bool {
        match self {
            Self::Delete(_) => true,
            Self::Load(_) => false,
        }
    }
}

impl LogicalPlan {
    /// Borrow scalar semantic fields shared by scalar/grouped logical variants.
    #[must_use]
    pub(in crate::db) const fn scalar_semantics(&self) -> &ScalarPlan {
        match self {
            Self::Scalar(plan) => plan,
            Self::Grouped(plan) => &plan.scalar,
        }
    }
}

impl AccessPlannedQuery {
    /// Borrow scalar semantic fields shared by scalar/grouped logical variants.
    #[must_use]
    pub(in crate::db) const fn scalar_plan(&self) -> &ScalarPlan {
        self.logical.scalar_semantics()
    }

    /// Borrow scalar missing-row consistency without exposing the full scalar
    /// plan to executor owners that only need row-presence policy.
    #[must_use]
    pub(in crate::db) fn scalar_consistency(&self) -> MissingRowPolicy {
        if self.access.has_selected_index_access_path() {
            // An accepted secondary index is a persisted claim that every
            // emitted key identifies an authoritative row. Ignoring a missing
            // row would turn accepted-index corruption into an incomplete
            // successful result.
            MissingRowPolicy::Error
        } else {
            self.scalar_plan().consistency
        }
    }

    /// Borrow grouped semantic fields when this plan is grouped.
    #[must_use]
    pub(in crate::db) const fn grouped_plan(&self) -> Option<&GroupPlan> {
        match &self.logical {
            LogicalPlan::Scalar(_) => None,
            LogicalPlan::Grouped(plan) => Some(plan),
        }
    }

    /// Lower this plan through accepted schema projection authority.
    #[must_use]
    pub(in crate::db) fn projection_spec_with_schema(&self, schema: &SchemaInfo) -> ProjectionSpec {
        if let Some(static_contract) = &self.static_execution_planning_contract {
            return static_contract.projection_spec.clone();
        }

        lower_projection_intent_with_schema(schema, &self.logical, &self.projection_selection)
    }

    /// Lower this plan into one projection semantic shape for identity hashing.
    #[must_use]
    pub(in crate::db::query) fn projection_spec_for_identity(&self) -> ProjectionSpec {
        lower_projection_identity(&self.logical, &self.projection_selection)
    }

    /// Return the executor-facing predicate after removing only filtered-index
    /// guard clauses the chosen access path already proves.
    ///
    /// This conservative form is used by preparation/explain surfaces that
    /// still need to see access-bound equalities as index-predicate input.
    #[must_use]
    pub(in crate::db) fn execution_preparation_predicate(&self) -> Option<Predicate> {
        if let Some(static_contract) = self.static_execution_planning_contract.as_ref() {
            return static_contract.execution_preparation_predicate.clone();
        }

        derive_execution_preparation_predicate(self)
    }

    /// Return the executor-facing residual predicate after removing any
    /// filtered-index guard clauses and fixed access-bound equalities already
    /// guaranteed by the chosen path.
    #[must_use]
    pub(in crate::db) fn effective_execution_predicate(&self) -> Option<Predicate> {
        if let Some(static_contract) = self.static_execution_planning_contract.as_ref() {
            return static_contract
                .residual_filter_contract
                .residual_filter_predicate()
                .cloned();
        }

        derive_residual_filter_predicate(self)
    }

    /// Return whether one explicit residual predicate survives access
    /// planning and still participates in residual execution.
    #[must_use]
    pub(in crate::db) fn has_residual_filter_predicate(&self) -> bool {
        self.effective_execution_predicate().is_some()
    }

    /// Borrow the planner-owned residual scalar filter expression when one
    /// surviving semantic remainder still requires runtime evaluation.
    #[must_use]
    pub(in crate::db) fn residual_filter_expr(&self) -> Option<&Expr> {
        if let Some(static_contract) = self.static_execution_planning_contract.as_ref() {
            return static_contract
                .residual_filter_contract
                .residual_filter_expr();
        }

        if !derive_has_residual_filter(self) {
            return None;
        }

        self.scalar_plan().filter_expr.as_ref()
    }

    /// Return whether one explicit residual scalar filter expression survives
    /// access planning and still requires runtime evaluation.
    #[must_use]
    pub(in crate::db) fn has_residual_filter_expr(&self) -> bool {
        self.residual_filter_expr().is_some()
    }

    /// Return the planner-owned residual-filter shape used by diagnostics.
    #[must_use]
    pub(in crate::db) fn residual_filter_shape(&self) -> ResidualFilterShape {
        if let Some(static_contract) = self.static_execution_planning_contract.as_ref() {
            return static_contract.residual_filter_contract.shape();
        }

        ResidualFilterShape::from_presence(
            self.residual_filter_expr().is_some(),
            self.effective_execution_predicate().is_some(),
        )
    }

    /// Return the planner-owned predicate pushdown label consumed by verbose
    /// execution diagnostics.
    #[must_use]
    #[cfg(feature = "sql")]
    pub(in crate::db) fn predicate_pushdown_label(&self) -> String {
        self.predicate_pushdown_diagnostics().label()
    }

    /// Return planner-owned predicate-pushdown diagnostics.
    #[must_use]
    #[cfg(feature = "sql")]
    pub(in crate::db) fn predicate_pushdown_diagnostics(&self) -> PredicatePushdownDiagnostics {
        if let Some(static_contract) = self.static_execution_planning_contract.as_ref() {
            return static_contract.predicate_pushdown_diagnostics;
        }

        derive_predicate_pushdown_diagnostics(self, self.residual_filter_shape())
    }

    /// Return the planner-owned predicate-pushdown outcome label.
    #[must_use]
    #[cfg(feature = "sql")]
    pub(in crate::db) fn predicate_pushdown_outcome_label(&self) -> &'static str {
        self.predicate_pushdown_diagnostics().outcome_label()
    }

    /// Return the planner-owned predicate-pushdown reason label.
    #[must_use]
    #[cfg(feature = "sql")]
    pub(in crate::db) fn predicate_pushdown_reason_label(&self) -> &'static str {
        self.predicate_pushdown_diagnostics().reason_label()
    }

    /// Borrow the planner-compiled execution-preparation predicate program.
    #[must_use]
    pub(in crate::db) fn execution_preparation_compiled_predicate(
        &self,
    ) -> Option<&PredicateProgram> {
        self.static_execution_planning_contract()?
            .execution_preparation_compiled_predicate
            .as_ref()
    }

    /// Borrow the planner-compiled effective runtime predicate program.
    #[must_use]
    pub(in crate::db) fn effective_runtime_compiled_predicate(&self) -> Option<&PredicateProgram> {
        match self
            .static_execution_planning_contract()?
            .residual_filter_contract
            .effective_runtime_filter_program()
        {
            Some(program) => program.predicate_program(),
            None => None,
        }
    }

    /// Borrow the planner-frozen effective runtime scalar filter program.
    #[must_use]
    pub(in crate::db) fn effective_runtime_filter_program(
        &self,
    ) -> Option<&EffectiveRuntimeFilterProgram> {
        self.static_execution_planning_contract()?
            .residual_filter_contract
            .effective_runtime_filter_program()
    }

    /// Lower scalar DISTINCT semantics into one executor-facing execution strategy.
    #[must_use]
    pub(in crate::db) fn distinct_execution_strategy(&self) -> DistinctExecutionStrategy {
        if !self.scalar_plan().distinct {
            return DistinctExecutionStrategy::None;
        }

        // DISTINCT on duplicate-safe single-path access shapes is a planner
        // no-op for runtime dedup mechanics. Composite shapes can surface
        // duplicate keys and therefore retain explicit dedup execution.
        match distinct_runtime_dedup_strategy(&self.access) {
            Some(strategy) => strategy,
            None => DistinctExecutionStrategy::None,
        }
    }

    /// Freeze one planner-owned route profile from accepted schema authority.
    pub(in crate::db) fn finalize_planner_route_profile_for_model_with_schema(
        &mut self,
        schema_info: &SchemaInfo,
    ) {
        self.set_planner_route_profile(project_planner_route_profile_for_schema(schema_info, self));
    }

    /// Freeze planner-owned executor metadata with explicit schema authority.
    pub(in crate::db) fn finalize_static_execution_planning_contract_with_schema(
        &mut self,
        schema_info: &SchemaInfo,
    ) -> Result<(), InternalError> {
        self.bind_group_field_slots_to_schema(schema_info)?;
        self.static_execution_planning_contract = Some(
            project_static_execution_planning_contract_with_schema(schema_info, self)?,
        );

        Ok(())
    }

    // Resolve authoring-time group field names onto accepted slots before the
    // plan becomes executable.
    fn bind_group_field_slots_to_schema(
        &mut self,
        schema_info: &SchemaInfo,
    ) -> Result<(), InternalError> {
        let LogicalPlan::Grouped(grouped) = &mut self.logical else {
            return Ok(());
        };

        let accepted_slots = grouped
            .group
            .group_fields
            .iter()
            .map(|field_slot| FieldSlot::resolve_with_schema(schema_info, field_slot.field()))
            .collect::<Option<Vec<_>>>()
            .ok_or_else(InternalError::planner_executor_invariant)?;
        grouped.group.group_fields = accepted_slots;

        Ok(())
    }

    /// Build one immutable execution-shape signature contract for runtime layers.
    #[must_use]
    pub(in crate::db) fn execution_shape_signature(
        &self,
        entity_path: &str,
    ) -> ExecutionShapeSignature {
        ExecutionShapeSignature::new(self.continuation_signature(entity_path))
    }

    /// Return whether the chosen access contract fully satisfies the current
    /// scalar query predicate without any additional runtime residual filtering.
    #[must_use]
    pub(in crate::db) fn predicate_fully_satisfied_by_access_contract(&self) -> bool {
        if let Some(static_contract) = self.static_execution_planning_contract.as_ref() {
            return self.scalar_plan().predicate.is_some()
                && !static_contract
                    .residual_filter_contract
                    .has_residual_filter();
        }

        derive_predicate_fully_satisfied_by_access_contract(self)
    }

    /// Borrow the planner-frozen compiled scalar projection program.
    #[must_use]
    pub(in crate::db) fn scalar_projection_plan(&self) -> Option<&[CompiledExpr]> {
        self.static_execution_planning_contract()?
            .scalar_projection_plan
            .as_deref()
    }

    /// Return whether planner-owned static execution metadata has already been frozen.
    #[must_use]
    pub(in crate::db) const fn has_static_execution_planning_contract(&self) -> bool {
        self.static_execution_planning_contract.is_some()
    }

    /// Borrow the planner-frozen ordered primary-key field names.
    pub(in crate::db) fn primary_key_names(&self) -> Result<Vec<&str>, InternalError> {
        Ok(self
            .require_static_execution_planning_contract()?
            .primary_key_names
            .iter()
            .map(String::as_str)
            .collect())
    }

    /// Borrow the planner-frozen projection slot reachability set.
    pub(in crate::db) fn projection_referenced_slots(&self) -> Result<&[usize], InternalError> {
        Ok(self
            .require_static_execution_planning_contract()?
            .projection_referenced_slots
            .as_slice())
    }

    /// Borrow the planner-frozen mask for direct projected output slots.
    #[cfg(any(test, feature = "diagnostics"))]
    pub(in crate::db) fn projected_slot_mask(&self) -> Result<&[bool], InternalError> {
        Ok(self
            .require_static_execution_planning_contract()?
            .projected_slot_mask
            .as_slice())
    }

    /// Return whether projection remains the full model-identity field list.
    pub(in crate::db) fn projection_is_model_identity(&self) -> Result<bool, InternalError> {
        Ok(self
            .require_static_execution_planning_contract()?
            .projection_is_model_identity)
    }

    /// Borrow the planner-frozen ORDER BY slot reachability set, if any.
    #[must_use]
    pub(in crate::db) fn order_referenced_slots(&self) -> Option<&[usize]> {
        self.static_execution_planning_contract()?
            .order_referenced_slots
            .as_deref()
    }

    /// Borrow the planner-frozen resolved ORDER BY program, if one exists.
    #[must_use]
    pub(in crate::db) fn resolved_order(&self) -> Option<&ResolvedOrder> {
        self.static_execution_planning_contract()?
            .resolved_order
            .as_ref()
    }

    /// Borrow the planner-frozen access slot map used by index predicate compilation.
    #[must_use]
    pub(in crate::db) fn slot_map(&self) -> Option<&[usize]> {
        self.static_execution_planning_contract()?
            .slot_map
            .as_deref()
    }

    /// Borrow grouped aggregate execution specs already resolved during static planning.
    #[must_use]
    pub(in crate::db) fn grouped_aggregate_execution_specs(
        &self,
    ) -> Option<&[GroupedAggregateExecutionSpec]> {
        self.static_execution_planning_contract()?
            .grouped_aggregate_execution_specs
            .as_deref()
    }

    /// Borrow the planner-resolved grouped DISTINCT execution strategy when present.
    #[must_use]
    pub(in crate::db) fn grouped_distinct_execution_strategy(
        &self,
    ) -> Option<&GroupedDistinctExecutionStrategy> {
        self.static_execution_planning_contract()?
            .grouped_distinct_execution_strategy
            .as_ref()
    }

    /// Borrow the frozen projection semantic shape without reopening model ownership.
    pub(in crate::db) fn frozen_projection_spec(&self) -> Result<&ProjectionSpec, InternalError> {
        Ok(&self
            .require_static_execution_planning_contract()?
            .projection_spec)
    }

    /// Borrow the frozen direct projection slots without reopening model ownership.
    #[must_use]
    pub(in crate::db) fn frozen_direct_projection_slots(&self) -> Option<&[usize]> {
        self.static_execution_planning_contract()?
            .projection_direct_slots
            .as_deref()
    }

    /// Borrow duplicate-preserving direct projection slots for raw data-row readers.
    #[must_use]
    pub(in crate::db) fn frozen_data_row_direct_projection_slots(&self) -> Option<&[usize]> {
        self.static_execution_planning_contract()?
            .projection_data_row_direct_slots
            .as_deref()
    }

    /// Borrow the planner-frozen key-item-aware compile targets for the chosen access path.
    #[must_use]
    pub(in crate::db) fn index_compile_targets(&self) -> Option<&[IndexCompileTarget]> {
        self.static_execution_planning_contract()?
            .index_compile_targets
            .as_deref()
    }

    const fn static_execution_planning_contract(&self) -> Option<&StaticExecutionPlanningContract> {
        self.static_execution_planning_contract.as_ref()
    }

    fn require_static_execution_planning_contract(
        &self,
    ) -> Result<&StaticExecutionPlanningContract, InternalError> {
        self.static_execution_planning_contract
            .as_ref()
            .ok_or_else(InternalError::query_executor_invariant)
    }
}

fn distinct_runtime_dedup_strategy<K>(access: &AccessPlan<K>) -> Option<DistinctExecutionStrategy> {
    match access {
        AccessPlan::Union(_) | AccessPlan::Intersection(_) => {
            Some(DistinctExecutionStrategy::PreOrdered)
        }
        AccessPlan::Path(path) if path.as_ref().is_index_multi_lookup() => {
            Some(DistinctExecutionStrategy::HashMaterialize)
        }
        AccessPlan::Path(_) => None,
    }
}

fn derive_continuation_policy_validated(plan: &AccessPlannedQuery) -> ContinuationPolicy {
    let is_grouped_safe = plan
        .grouped_plan()
        .is_none_or(|grouped| grouped_cursor_policy_violation(grouped, true).is_none());

    ContinuationPolicy::new(
        true, // Continuation resume windows require anchor semantics for pushdown-safe replay.
        true, // Continuation resumes must advance strictly to prevent replay/regression loops.
        is_grouped_safe,
    )
}

/// Project one planner-owned route profile from accepted schema authority.
#[must_use]
pub(in crate::db) fn project_planner_route_profile_for_schema(
    schema_info: &SchemaInfo,
    plan: &AccessPlannedQuery,
) -> PlannerRouteProfile {
    let primary_key_names = primary_key_names_from_schema(schema_info);
    let secondary_order_contract = plan.scalar_plan().order.as_ref().and_then(|order| {
        order.deterministic_secondary_order_contract_fields(primary_key_names.as_slice())
    });

    PlannerRouteProfile::new(
        derive_continuation_policy_validated(plan),
        derive_logical_pushdown_eligibility(plan, secondary_order_contract.as_ref()),
        secondary_order_contract,
    )
}

fn project_static_execution_planning_contract_with_schema(
    schema_info: &SchemaInfo,
    plan: &AccessPlannedQuery,
) -> Result<StaticExecutionPlanningContract, InternalError> {
    let projection_spec =
        lower_projection_intent_with_schema(schema_info, &plan.logical, &plan.projection_selection);
    let execution_preparation_predicate = plan.execution_preparation_predicate();
    let residual_filter_predicate = derive_residual_filter_predicate_from_preparation(
        plan,
        execution_preparation_predicate.as_ref(),
    );
    let residual_filter_expr = derive_residual_filter_expr(plan);
    let effective_runtime_filter_program = compile_effective_runtime_filter_program(
        schema_info,
        residual_filter_expr.as_ref(),
        residual_filter_predicate.as_ref(),
    )?;
    let residual_filter_contract = ResidualFilterContract::new(
        residual_filter_expr,
        residual_filter_predicate,
        effective_runtime_filter_program,
    );
    let residual_filter_shape = residual_filter_contract.shape();
    let execution_preparation_compiled_predicate =
        should_compile_execution_preparation_predicate(residual_filter_shape)
            .then(|| {
                compile_optional_predicate(schema_info, execution_preparation_predicate.as_ref())
            })
            .flatten();
    let predicate_pushdown_diagnostics =
        derive_predicate_pushdown_diagnostics(plan, residual_filter_shape);
    let scalar_projection_plan = if plan.grouped_plan().is_none() {
        Some(
            compile_scalar_projection_plan_with_schema(schema_info, &projection_spec)
                .ok_or_else(InternalError::query_executor_invariant)?
                .iter()
                .map(CompiledExpr::compile)
                .collect(),
        )
    } else {
        None
    };
    let (grouped_aggregate_execution_specs, grouped_distinct_execution_strategy) =
        resolve_grouped_static_planning_semantics(schema_info, plan, &projection_spec)?;
    let projection_direct_slots = lower_direct_projection_slots_with_schema(
        schema_info,
        &plan.logical,
        &plan.projection_selection,
    );
    let projection_data_row_direct_slots = lower_data_row_direct_projection_slots_with_schema(
        schema_info,
        &plan.logical,
        &plan.projection_selection,
    );
    let projection_referenced_slots = projection_spec.referenced_slots_for_schema(schema_info)?;
    let projected_slot_mask =
        projected_slot_mask_for_spec(schema_info, projection_direct_slots.as_deref());
    let projection_is_model_identity = projection_spec.is_schema_identity_for(schema_info);
    let resolved_order = resolved_order_for_plan(schema_info, plan)?;
    let order_referenced_slots = order_referenced_slots_for_resolved_order(resolved_order.as_ref());
    let slot_map = slot_map_for_schema_plan(schema_info, plan);
    let index_compile_targets = index_compile_targets_for_schema_plan(schema_info, plan);

    Ok(StaticExecutionPlanningContract {
        primary_key_names: schema_info.primary_key_names().to_vec(),
        projection_spec,
        execution_preparation_predicate,
        execution_preparation_compiled_predicate,
        residual_filter_contract,
        predicate_pushdown_diagnostics,
        scalar_projection_plan,
        grouped_aggregate_execution_specs,
        grouped_distinct_execution_strategy,
        projection_direct_slots,
        projection_data_row_direct_slots,
        projection_referenced_slots,
        projected_slot_mask,
        projection_is_model_identity,
        resolved_order,
        order_referenced_slots,
        slot_map,
        index_compile_targets,
    })
}

fn primary_key_names_from_schema(schema_info: &SchemaInfo) -> Vec<&str> {
    schema_info
        .primary_key_names()
        .iter()
        .map(String::as_str)
        .collect()
}

// Compile the executor-owned residual scalar filter contract once from the
// planner-derived residual artifacts so runtime never has to rediscover
// residual presence or shape from semantic/filter/pushdown state.
fn compile_effective_runtime_filter_program(
    schema_info: &SchemaInfo,
    residual_filter_expr: Option<&Expr>,
    residual_filter_predicate: Option<&Predicate>,
) -> Result<Option<EffectiveRuntimeFilterProgram>, InternalError> {
    // Keep the existing predicate fast path when the residual semantics still
    // fit the derived predicate contract. The expression-owned lane is only
    // needed once pushdown loses semantic coverage and a residual predicate no
    // longer exists.
    if let Some(predicate) = residual_filter_predicate {
        return Ok(Some(EffectiveRuntimeFilterProgram::predicate(
            PredicateProgram::compile_with_schema_info(schema_info, predicate),
        )));
    }

    if let Some(filter_expr) = residual_filter_expr {
        let compiled = compile_scalar_projection_expr_with_schema(schema_info, filter_expr)
            .ok_or_else(InternalError::query_invalid_logical_plan)?;

        return Ok(Some(EffectiveRuntimeFilterProgram::expression(
            CompiledExpr::compile(&compiled),
        )));
    }

    Ok(None)
}

// Derive the executor-preparation predicate once from the selected access path.
// This strips only filtered-index guard clauses while preserving access-bound
// equalities that still matter to preparation/explain consumers.
fn derive_execution_preparation_predicate(plan: &AccessPlannedQuery) -> Option<Predicate> {
    let query_predicate = plan.scalar_plan().predicate.as_ref()?;

    match plan.access.selected_index_contract() {
        Some(index) => {
            residual_query_predicate_after_filtered_access_contract(index, query_predicate)
        }
        None => Some(query_predicate.clone()),
    }
}

// Derive the final residual predicate once from the already-filtered
// preparation predicate plus any equality bounds guaranteed by the concrete
// access path.
fn derive_residual_filter_predicate(plan: &AccessPlannedQuery) -> Option<Predicate> {
    let filtered_residual = derive_execution_preparation_predicate(plan);

    derive_residual_filter_predicate_from_preparation(plan, filtered_residual.as_ref())
}

fn derive_residual_filter_predicate_from_preparation(
    plan: &AccessPlannedQuery,
    execution_preparation_predicate: Option<&Predicate>,
) -> Option<Predicate> {
    let execution_preparation_predicate = execution_preparation_predicate?;

    residual_query_predicate_after_access_path_bounds(
        plan.access.as_path(),
        execution_preparation_predicate,
    )
}

// Derive the explicit residual semantic expression once for finalized plans.
// The residual expression remains the planner-owned semantic filter when any
// runtime filtering still survives access satisfaction.
fn derive_residual_filter_expr(plan: &AccessPlannedQuery) -> Option<Expr> {
    let filter_expr = plan.scalar_plan().filter_expr.as_ref()?;
    if derive_semantic_filter_fully_satisfied_by_access_contract(plan) {
        return None;
    }

    Some(filter_expr.clone())
}

// Return whether any residual filtering survives after access planning. This
// helper exists only for pre-finalization assembly; finalized plans must read
// the explicit residual artifacts frozen in `StaticExecutionPlanningContract`.
fn derive_has_residual_filter(plan: &AccessPlannedQuery) -> bool {
    match (
        plan.scalar_plan().filter_expr.as_ref(),
        plan.scalar_plan().predicate.as_ref(),
    ) {
        (None, None) => false,
        (Some(_), None) => true,
        (Some(_) | None, Some(_)) => !plan.predicate_fully_satisfied_by_access_contract(),
    }
}

// Freeze predicate-pushdown diagnostics from one logical plan shape. This keeps
// lazy plan accessors and finalized static planning on the same argument
// contract while leaving route selection and residual filtering unchanged.
fn derive_predicate_pushdown_diagnostics(
    plan: &AccessPlannedQuery,
    residual_filter_shape: ResidualFilterShape,
) -> PredicatePushdownDiagnostics {
    PredicatePushdownDiagnostics::from_plan(
        plan.scalar_plan().filter_expr.is_some(),
        plan.scalar_plan().predicate_covers_filter_expr,
        plan.scalar_plan().predicate.as_ref(),
        &plan.access,
        residual_filter_shape,
    )
}

// Return true when the planner-owned predicate contract is fully satisfied by
// access planning and no semantic residual filter expression survives.
fn derive_predicate_fully_satisfied_by_access_contract(plan: &AccessPlannedQuery) -> bool {
    plan.scalar_plan().predicate.is_some()
        && derive_residual_filter_predicate(plan).is_none()
        && derive_residual_filter_expr(plan).is_none()
}

// Return true when the semantic filter expression is entirely represented by
// the planner-owned predicate contract and the chosen access path satisfies
// that predicate without any runtime remainder.
const fn derive_semantic_filter_fully_satisfied_by_access_contract(
    plan: &AccessPlannedQuery,
) -> bool {
    plan.scalar_plan().filter_expr.is_some()
        && plan.scalar_plan().predicate.is_some()
        && plan.scalar_plan().predicate_covers_filter_expr
}

// Compile one optional planner-frozen predicate program while keeping the
// static planning assembly path free of repeated `Option` mapping boilerplate.
fn compile_optional_predicate(
    schema_info: &SchemaInfo,
    predicate: Option<&Predicate>,
) -> Option<PredicateProgram> {
    predicate.map(|predicate| PredicateProgram::compile_with_schema_info(schema_info, predicate))
}

// Avoid compiling large access-proven predicates into executor preparation.
// When no residual filter survives, the chosen access route already enforces
// the predicate and route/explain consumers can use the explicit residual
// contract instead of recompiling access-bound literals.
const fn should_compile_execution_preparation_predicate(
    residual_filter_shape: ResidualFilterShape,
) -> bool {
    !residual_filter_shape.is_absent()
}

// Resolve the grouped-only static planning semantics bundle once so grouped
// aggregate execution specs and grouped DISTINCT strategy stay derived under
// one shared grouped-plan branch.
fn resolve_grouped_static_planning_semantics(
    schema_info: &SchemaInfo,
    plan: &AccessPlannedQuery,
    projection_spec: &ProjectionSpec,
) -> Result<
    (
        Option<Vec<GroupedAggregateExecutionSpec>>,
        Option<GroupedDistinctExecutionStrategy>,
    ),
    InternalError,
> {
    let Some(grouped) = plan.grouped_plan() else {
        return Ok((None, None));
    };

    let mut aggregate_specs = grouped_aggregate_specs_from_projection_spec(
        projection_spec,
        grouped.group.group_fields.as_slice(),
        grouped.group.aggregates.as_slice(),
    )?;
    extend_grouped_having_aggregate_specs(&mut aggregate_specs, grouped)?;

    let grouped_aggregate_execution_specs = Some(grouped_aggregate_execution_specs(
        schema_info,
        aggregate_specs.as_slice(),
    )?);
    let grouped_distinct_execution_strategy = Some(
        resolved_grouped_distinct_execution_strategy_with_schema_info(
            schema_info,
            grouped.group.group_fields.as_slice(),
            grouped.group.aggregates.as_slice(),
            grouped.having_expr.as_ref(),
        )?,
    );

    Ok((
        grouped_aggregate_execution_specs,
        grouped_distinct_execution_strategy,
    ))
}

fn extend_grouped_having_aggregate_specs(
    aggregate_specs: &mut Vec<GroupedAggregateExecutionSpec>,
    grouped: &GroupPlan,
) -> Result<(), InternalError> {
    if let Some(having_expr) = grouped.having_expr.as_ref() {
        extend_unique_grouped_aggregate_specs_from_expr(aggregate_specs, having_expr)?;
    }

    Ok(())
}

fn projected_slot_mask_for_spec(
    schema_info: &SchemaInfo,
    direct_projection_slots: Option<&[usize]>,
) -> Vec<bool> {
    let schema_slot_len = direct_projection_slots
        .and_then(|slots| slots.iter().copied().max())
        .map_or(0, |slot| slot.saturating_add(1));
    let mut projected_slots = vec![
        false;
        schema_info
            .field_names_in_slot_order()
            .len()
            .max(schema_slot_len)
    ];

    let Some(direct_projection_slots) = direct_projection_slots else {
        return projected_slots;
    };

    for slot in direct_projection_slots.iter().copied() {
        if let Some(projected) = projected_slots.get_mut(slot) {
            *projected = true;
        }
    }

    projected_slots
}

fn resolved_order_for_plan(
    schema_info: &SchemaInfo,
    plan: &AccessPlannedQuery,
) -> Result<Option<ResolvedOrder>, InternalError> {
    if grouped_plan_strategy(plan).is_some_and(GroupedPlanStrategy::is_top_k_group) {
        return Ok(None);
    }

    let Some(order) = plan.scalar_plan().order.as_ref() else {
        return Ok(None);
    };

    let mut fields = Vec::with_capacity(order.fields.len());
    for term in &order.fields {
        fields.push(ResolvedOrderField::new(
            resolved_order_value_source_for_term(schema_info, term)?,
            term.direction(),
        ));
    }

    Ok(Some(ResolvedOrder::new(fields)))
}

fn resolved_order_value_source_for_term(
    schema_info: &SchemaInfo,
    term: &crate::db::query::plan::OrderTerm,
) -> Result<ResolvedOrderValueSource, InternalError> {
    if term.direct_field().is_none() {
        let rendered = term.rendered_label();
        validate_resolved_order_expr_fields(schema_info, term.expr(), rendered.as_str())?;
        let compiled = compile_scalar_projection_expr_with_schema(schema_info, term.expr())
            .ok_or_else(|| order_expression_scalar_seam_error(rendered.as_str()))?;

        return Ok(ResolvedOrderValueSource::expression(CompiledExpr::compile(
            &compiled,
        )));
    }

    let Some(field) = term.direct_field() else {
        return Err(InternalError::query_invalid_logical_plan());
    };
    let slot = resolve_required_schema_slot(
        schema_info,
        field,
        InternalError::query_invalid_logical_plan,
    )?;

    Ok(ResolvedOrderValueSource::direct_field(slot))
}

fn validate_resolved_order_expr_fields(
    schema_info: &SchemaInfo,
    expr: &Expr,
    rendered: &str,
) -> Result<(), InternalError> {
    expr.try_for_each_tree_expr(&mut |node| match node {
        Expr::Field(field_id) => resolve_required_schema_slot(
            schema_info,
            field_id.as_str(),
            InternalError::query_invalid_logical_plan,
        )
        .map(|_| ()),
        Expr::Aggregate(_) => Err(order_expression_scalar_seam_error(rendered)),
        #[cfg(test)]
        Expr::Alias { .. } => Err(order_expression_scalar_seam_error(rendered)),
        Expr::Unary { .. } => Err(order_expression_scalar_seam_error(rendered)),
        _ => Ok(()),
    })
}

// Resolve one schema-authoritative field slot while keeping planner
// invalid-logical-plan error construction at the callsite that owns the
// diagnostic wording.
fn resolve_required_schema_slot<F>(
    schema_info: &SchemaInfo,
    field: &str,
    invalid_plan_error: F,
) -> Result<usize, InternalError>
where
    F: FnOnce() -> InternalError,
{
    schema_info
        .field_slot_index(field)
        .ok_or_else(invalid_plan_error)
}

// Keep the scalar-order expression seam violation text under one helper so the
// parse validation and compile validation paths do not drift.
fn order_expression_scalar_seam_error(_rendered: &str) -> InternalError {
    InternalError::query_invalid_logical_plan()
}

// Keep one stable executor-facing slot list for grouped order terms after the
// planner has frozen the structural `ResolvedOrder`. The grouped Top-K route
// now consumes this same referenced-slot contract instead of re-deriving order
// sources from planner strategy at runtime.
fn order_referenced_slots_for_resolved_order(
    resolved_order: Option<&ResolvedOrder>,
) -> Option<Vec<usize>> {
    Some(resolved_order?.referenced_slots())
}

fn slot_map_for_schema_plan(
    schema_info: &SchemaInfo,
    plan: &AccessPlannedQuery,
) -> Option<Vec<usize>> {
    let executable = plan.access.executable_contract();

    resolved_index_slots_for_access_path(schema_info, &executable)
}

fn resolved_index_slots_for_access_path(
    schema_info: &SchemaInfo,
    access: &ExecutableAccessPlan<'_, crate::value::Value>,
) -> Option<Vec<usize>> {
    let path = access.as_path()?;
    let path_facts = path.shape_facts();
    let key_items = path_facts.index_key_items_for_slot_map()?;
    let mut slots = Vec::with_capacity(key_items.key_arity());
    for key_item in key_items.key_items() {
        let slot = schema_info.field_slot_index(key_item.as_ref().field())?;
        slots.push(slot);
    }

    Some(slots)
}

fn index_compile_targets_for_schema_plan(
    schema_info: &SchemaInfo,
    plan: &AccessPlannedQuery,
) -> Option<Vec<IndexCompileTarget>> {
    let executable = plan.access.executable_contract();
    let path = executable.as_path()?;
    let key_items = path.shape_facts().index_key_items_for_slot_map()?;
    let mut targets = Vec::new();

    for (component_index, key_item) in key_items.key_items().iter().enumerate() {
        let key_item = key_item.as_ref();
        let field_slot = schema_info.field_slot_index(key_item.field())?;
        targets.push(IndexCompileTarget {
            component_index,
            field_slot,
            kind: match key_item {
                SemanticIndexKeyItemRef::Field(_) => IndexCompileTargetKind::Field,
                SemanticIndexKeyItemRef::AcceptedExpression(expression) => {
                    IndexCompileTargetKind::Expression(expression.op())
                }
            },
        });
    }

    Some(targets)
}