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.

#[cfg(any(test, feature = "sql"))]
use crate::db::predicate::MissingRowPolicy;
use crate::{
    db::{
        access::{AccessPlan, ExecutableAccessPlan, SemanticIndexKeyItemsRef},
        predicate::{IndexCompileTarget, Predicate, PredicateProgram},
        query::plan::{
            AccessPlannedQuery, ContinuationPolicy, DistinctExecutionStrategy,
            EffectiveRuntimeFilterProgram, ExecutionShapeSignature, 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,
            },
            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, 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,
    model::{
        entity::EntityModel,
        index::{IndexKeyItem, IndexKeyItemsRef},
    },
};

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,
        }
    }

    /// Borrow scalar semantic fields mutably across logical variants for tests.
    #[must_use]
    #[cfg(test)]
    pub(in crate::db) const fn scalar_semantics_mut(&mut self) -> &mut ScalarPlan {
        match self {
            Self::Scalar(plan) => plan,
            Self::Grouped(plan) => &mut plan.scalar,
        }
    }

    /// Test-only shorthand for explicit scalar semantic borrowing.
    #[must_use]
    #[cfg(test)]
    pub(in crate::db) const fn scalar(&self) -> &ScalarPlan {
        self.scalar_semantics()
    }

    /// Test-only shorthand for explicit mutable scalar semantic borrowing.
    #[must_use]
    #[cfg(test)]
    pub(in crate::db) const fn scalar_mut(&mut self) -> &mut ScalarPlan {
        self.scalar_semantics_mut()
    }
}

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]
    #[cfg(any(test, feature = "sql"))]
    pub(in crate::db) const fn scalar_consistency(&self) -> MissingRowPolicy {
        self.scalar_plan().consistency
    }

    /// Borrow scalar semantic fields mutably across logical variants for tests.
    #[must_use]
    #[cfg(test)]
    pub(in crate::db) const fn scalar_plan_mut(&mut self) -> &mut ScalarPlan {
        self.logical.scalar_semantics_mut()
    }

    /// Test-only shorthand for explicit scalar plan borrowing.
    #[must_use]
    #[cfg(test)]
    pub(in crate::db) const fn scalar(&self) -> &ScalarPlan {
        self.scalar_plan()
    }

    /// Test-only shorthand for explicit mutable scalar plan borrowing.
    #[must_use]
    #[cfg(test)]
    pub(in crate::db) const fn scalar_mut(&mut self) -> &mut ScalarPlan {
        self.scalar_plan_mut()
    }

    /// 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 into one canonical planner-owned projection semantic spec.
    #[must_use]
    pub(in crate::db) fn projection_spec(&self, model: &EntityModel) -> ProjectionSpec {
        if let Some(static_contract) = &self.static_execution_planning_contract {
            return static_contract.projection_spec.clone();
        }

        lower_projection_intent(model, &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]
    pub(in crate::db) fn predicate_pushdown_label(&self) -> String {
        self.predicate_pushdown_diagnostics().label()
    }

    /// Return planner-owned predicate-pushdown diagnostics.
    #[must_use]
    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;
        }

        PredicatePushdownDiagnostics::from_plan(
            self.scalar_plan().filter_expr.is_some(),
            self.scalar_plan().predicate_covers_filter_expr,
            self.scalar_plan().predicate.as_ref(),
            &self.access,
            self.residual_filter_shape(),
        )
    }

    /// Return the planner-owned predicate-pushdown outcome label.
    #[must_use]
    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]
    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) const 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) const 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-compiled effective runtime scalar filter expression.
    #[cfg(test)]
    #[must_use]
    pub(in crate::db) const fn effective_runtime_compiled_filter_expr(
        &self,
    ) -> Option<&CompiledExpr> {
        match self
            .static_execution_planning_contract()
            .residual_filter_contract
            .effective_runtime_filter_program()
        {
            Some(program) => program.expression_filter(),
            None => None,
        }
    }

    /// Borrow the planner-frozen effective runtime scalar filter program.
    #[must_use]
    pub(in crate::db) const 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 after model validation completes.
    #[cfg(test)]
    pub(in crate::db) fn finalize_planner_route_profile_for_model(&mut self, model: &EntityModel) {
        self.set_planner_route_profile(project_planner_route_profile_for_model(model, self));
    }

    /// 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 model-only executor metadata after logical/access planning completes.
    #[cfg(test)]
    pub(in crate::db) fn finalize_static_execution_planning_contract_for_model_only(
        &mut self,
        model: &EntityModel,
    ) -> Result<(), InternalError> {
        self.finalize_static_execution_planning_contract_for_model_with_schema(
            model,
            SchemaInfo::cached_for_generated_entity_model(model),
        )
    }

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

        Ok(())
    }

    /// Build one immutable execution-shape signature contract for runtime layers.
    #[must_use]
    pub(in crate::db) fn execution_shape_signature(
        &self,
        entity_path: &'static 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 post-access 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.
    #[must_use]
    pub(in crate::db) fn primary_key_names(&self) -> Vec<&str> {
        self.static_execution_planning_contract()
            .primary_key_names
            .iter()
            .map(String::as_str)
            .collect()
    }

    /// Borrow the planner-frozen projection slot reachability set.
    #[must_use]
    pub(in crate::db) const fn projection_referenced_slots(&self) -> &[usize] {
        self.static_execution_planning_contract()
            .projection_referenced_slots
            .as_slice()
    }

    /// Borrow the planner-frozen mask for direct projected output slots.
    #[must_use]
    #[cfg(any(test, all(feature = "sql", feature = "diagnostics")))]
    pub(in crate::db) const fn projected_slot_mask(&self) -> &[bool] {
        self.static_execution_planning_contract()
            .projected_slot_mask
            .as_slice()
    }

    /// Return whether projection remains the full model-identity field list.
    #[must_use]
    pub(in crate::db) const fn projection_is_model_identity(&self) -> bool {
        self.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) const 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) const 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.
    #[must_use]
    pub(in crate::db) const fn frozen_projection_spec(&self) -> &ProjectionSpec {
        &self.static_execution_planning_contract().projection_spec
    }

    /// Borrow the frozen direct projection slots without reopening model ownership.
    #[must_use]
    #[cfg(any(test, feature = "sql"))]
    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]
    #[cfg(any(test, feature = "sql"))]
    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) -> &StaticExecutionPlanningContract {
        self.static_execution_planning_contract
            .as_ref()
            .expect("query semantics 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 the finalized logical+access plan.
#[must_use]
#[cfg(test)]
pub(in crate::db) fn project_planner_route_profile_for_model(
    model: &EntityModel,
    plan: &AccessPlannedQuery,
) -> PlannerRouteProfile {
    let primary_key_names = ordered_primary_key_names(model);
    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,
    )
}

/// 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_for_model(
    model: &EntityModel,
    schema_info: &SchemaInfo,
    plan: &AccessPlannedQuery,
) -> Result<StaticExecutionPlanningContract, InternalError> {
    let projection_spec = lower_projection_intent(model, &plan.logical, &plan.projection_selection);
    let execution_preparation_predicate = plan.execution_preparation_predicate();
    let residual_filter_predicate = derive_residual_filter_predicate(plan);
    let residual_filter_expr = derive_residual_filter_expr_for_model(model, plan);
    let execution_preparation_compiled_predicate =
        compile_optional_predicate(schema_info, execution_preparation_predicate.as_ref());
    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 predicate_pushdown_diagnostics = 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_contract.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(
        model,
        schema_info,
        &plan.logical,
        &plan.projection_selection,
    );
    let projection_data_row_direct_slots = lower_data_row_direct_projection_slots_with_schema(
        model,
        schema_info,
        &plan.logical,
        &plan.projection_selection,
    );
    let projection_referenced_slots =
        projection_spec.referenced_slots_for_schema(model, schema_info)?;
    let projected_slot_mask =
        projected_slot_mask_for_spec(model, projection_direct_slots.as_deref());
    let projection_is_model_identity = projection_spec.is_model_identity_for(model);
    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,
    })
}

#[cfg(test)]
fn ordered_primary_key_names(model: &EntityModel) -> Vec<&'static str> {
    model.primary_key_names()
}

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);
    let filtered_residual = filtered_residual.as_ref()?;

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

// 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())
}

// Derive the explicit residual semantic expression during finalization using
// the trusted entity schema so compare-family literal normalization matches the
// planner-owned predicate contract before residual ownership is decided.
fn derive_residual_filter_expr_for_model(
    model: &EntityModel,
    plan: &AccessPlannedQuery,
) -> Option<Expr> {
    let filter_expr = plan.scalar_plan().filter_expr.as_ref()?;
    if derive_semantic_filter_fully_satisfied_by_access_contract_for_model(model, 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(),
    }
}

// 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
}

// Return true when finalized planning can prove that the semantic filter
// expression is completely represented by the planner-owned predicate contract
// after aligning compare literals through the trusted entity schema.
const fn derive_semantic_filter_fully_satisfied_by_access_contract_for_model(
    _model: &EntityModel,
    plan: &AccessPlannedQuery,
) -> bool {
    derive_semantic_filter_fully_satisfied_by_access_contract(plan)
}

// 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))
}

// 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() {
        collect_grouped_having_expr_aggregate_specs(aggregate_specs, having_expr)?;
    }

    Ok(())
}

fn collect_grouped_having_expr_aggregate_specs(
    aggregate_specs: &mut Vec<GroupedAggregateExecutionSpec>,
    expr: &Expr,
) -> Result<(), InternalError> {
    if !expr.contains_aggregate() {
        return Ok(());
    }

    expr.try_for_each_tree_aggregate(&mut |aggregate_expr| {
        let aggregate_spec = GroupedAggregateExecutionSpec::from_aggregate_expr(aggregate_expr);

        if aggregate_specs
            .iter()
            .all(|current| current != &aggregate_spec)
        {
            aggregate_specs.push(aggregate_spec);
        }

        Ok(())
    })
}

fn projected_slot_mask_for_spec(
    model: &EntityModel,
    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; model.fields().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 field = term.direct_field().expect("query semantics invariant");
    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::new();

    match key_items.key_items() {
        SemanticIndexKeyItemsRef::Fields(fields) => {
            slots.reserve(fields.len());
            for field_name in fields {
                let slot = schema_info.field_slot_index(field_name)?;
                slots.push(slot);
            }
        }
        SemanticIndexKeyItemsRef::Accepted(items) => {
            slots.reserve(items.len());
            for key_item in items {
                let slot = schema_info.field_slot_index(key_item.as_ref().field())?;
                slots.push(slot);
            }
        }
        SemanticIndexKeyItemsRef::Static(IndexKeyItemsRef::Fields(fields)) => {
            slots.reserve(fields.len());
            for &field_name in fields {
                let slot = schema_info.field_slot_index(field_name)?;
                slots.push(slot);
            }
        }
        SemanticIndexKeyItemsRef::Static(IndexKeyItemsRef::Items(items)) => {
            slots.reserve(items.len());
            for key_item in items {
                let slot = schema_info.field_slot_index(key_item.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();

    match key_items.key_items() {
        SemanticIndexKeyItemsRef::Fields(_) | SemanticIndexKeyItemsRef::Accepted(_) => {
            return None;
        }
        SemanticIndexKeyItemsRef::Static(IndexKeyItemsRef::Fields(fields)) => {
            for (component_index, &field_name) in fields.iter().enumerate() {
                let field_slot = schema_info.field_slot_index(field_name)?;
                targets.push(IndexCompileTarget {
                    component_index,
                    field_slot,
                    key_item: IndexKeyItem::Field(field_name),
                });
            }
        }
        SemanticIndexKeyItemsRef::Static(IndexKeyItemsRef::Items(items)) => {
            for (component_index, &key_item) in items.iter().enumerate() {
                let field_slot = schema_info.field_slot_index(key_item.field())?;
                targets.push(IndexCompileTarget {
                    component_index,
                    field_slot,
                    key_item,
                });
            }
        }
    }

    Some(targets)
}