icydb-core 0.261.17

//! Module: query::plan::group
//! Responsibility: grouped-plan handoff contract between query planning and executor.
//! Does not own: grouped runtime execution logic.
//! Boundary: explicit grouped query-to-executor transfer surface.

#[cfg(test)]
mod tests;

use crate::{
    db::{
        query::{
            builder::AggregateExpr,
            plan::{
                AccessPlannedQuery, AggregateIdentity, AggregateKind, AggregateSemanticKeyRef,
                FieldSlot, GlobalDistinctAggregateKind, GroupAggregateSpec,
                GroupDistinctAdmissibility, GroupDistinctPolicyReason, GroupedExecutionConfig,
                GroupedPlanStrategy,
                expr::{
                    CompiledExpr, Expr, ProjectionSpec, compile_scalar_projection_expr_with_schema,
                },
                grouped_distinct_admissibility, grouped_plan_strategy,
                resolve_global_distinct_field_aggregate, validate_grouped_projection_layout,
            },
        },
        schema::SchemaInfo,
    },
    error::InternalError,
};
use icydb_diagnostic_code::DiagnosticExecutionBudgetResource as Resource;

///
/// PlannedProjectionLayout
///
/// Planner-owned grouped projection position layout transferred to executor.
/// Positions are derived only from `ProjectionSpec` identity shape and preserve
/// projection declaration order for grouped fields and aggregates.
///

#[derive(Clone, Debug, Eq, PartialEq)]
pub(in crate::db) struct PlannedProjectionLayout {
    pub(in crate::db) group_field_positions: Vec<usize>,
    pub(in crate::db) aggregate_positions: Vec<usize>,
}

///
/// GroupedAggregateExecutionSpec
///
/// GroupedAggregateExecutionSpec is the canonical grouped aggregate carrier
/// shared across planner projection analysis and grouped runtime handoff.
/// Semantic identity stays present for projection/HAVING matching, while
/// planner-owned slot resolution and compiled input preparation are attached
/// once under model ownership before execution begins.
///

#[derive(Clone, Debug, Eq, PartialEq)]
pub(in crate::db) struct GroupedAggregateExecutionSpec {
    identity: AggregateIdentity,
    target_slot: Option<FieldSlot>,
    filter_expr: Option<Expr>,
    compiled_input_expr: Option<CompiledExpr>,
    compiled_filter_expr: Option<CompiledExpr>,
}

impl GroupedAggregateExecutionSpec {
    // Compile one grouped aggregate-attached scalar expression through the
    // shared scalar projection seam and preserve the planner/executor
    // error classification for aggregate inputs and aggregate-local filters.
    fn compile_attached_scalar_expr(
        schema_info: &SchemaInfo,
        expr: &Expr,
        budget: &dyn crate::db::query::construction::ConstructionBudget,
    ) -> Result<CompiledExpr, InternalError> {
        let scalar = compile_scalar_projection_expr_with_schema(schema_info, expr, budget)?
            .ok_or_else(InternalError::planner_executor_invariant)?;

        Ok(scalar)
    }

    /// Build one grouped aggregate spec from one aggregate identity expression.
    #[must_use]
    pub(in crate::db) fn from_aggregate_expr(aggregate_expr: &AggregateExpr) -> Self {
        Self {
            identity: AggregateIdentity::from_aggregate_expr(aggregate_expr),
            target_slot: None,
            filter_expr: aggregate_expr.filter_expr().cloned(),
            compiled_input_expr: None,
            compiled_filter_expr: None,
        }
    }

    /// Build one uncompiled grouped aggregate spec from already-canonical
    /// aggregate inputs.
    /// Global aggregate execution uses this to reuse the grouped
    /// post-aggregate compilation seam without rebuilding grouped planner
    /// state or pretending the implicit single aggregate row came from grouped
    /// route preparation.
    #[must_use]
    pub(in crate::db) fn from_uncompiled_inputs(
        kind: AggregateKind,
        target_slot: Option<FieldSlot>,
        input_expr: Option<Expr>,
        filter_expr: Option<Expr>,
        distinct: bool,
    ) -> Self {
        Self {
            identity: AggregateIdentity::from_kind_input_and_distinct(kind, input_expr, distinct),
            target_slot,
            filter_expr,
            compiled_input_expr: None,
            compiled_filter_expr: None,
        }
    }

    /// Resolve planner-owned grouped aggregate execution attachments from schema authority.
    pub(in crate::db) fn resolve_with_schema_info(
        &mut self,
        schema_info: &SchemaInfo,
        budget: &dyn crate::db::query::construction::ConstructionBudget,
    ) -> Result<(), InternalError> {
        budget.charge(Resource::PredicateExpressionSteps, 1)?;
        let compiled_input_expr = self
            .input_expr()
            .map(|expr| Self::compile_attached_scalar_expr(schema_info, expr, budget))
            .transpose()?;
        let compiled_filter_expr = self
            .filter_expr()
            .map(|expr| Self::compile_attached_scalar_expr(schema_info, expr, budget))
            .transpose()?;
        let target_slot = self
            .target_field()
            .map(|field| {
                resolve_aggregate_target_field_slot_from_schema(schema_info, field)
                    .ok_or_else(InternalError::planner_executor_invariant)
            })
            .transpose()?;

        // Compile every attachment before changing the owned carrier. Failure
        // preserves the original spec; success keeps its identity/filter buffers.
        self.target_slot = target_slot;
        self.compiled_input_expr = compiled_input_expr;
        self.compiled_filter_expr = compiled_filter_expr;
        Ok(())
    }

    /// Return the grouped aggregate kind.
    #[must_use]
    pub(in crate::db) const fn kind(&self) -> AggregateKind {
        self.identity.kind()
    }

    /// Borrow the optional grouped aggregate target field label.
    #[must_use]
    pub(in crate::db) const fn target_field(&self) -> Option<&str> {
        match self.input_expr() {
            Some(Expr::Field(field_id)) => Some(field_id.as_str()),
            _ => None,
        }
    }

    /// Borrow the optional planner-resolved grouped aggregate target slot.
    #[must_use]
    pub(in crate::db) const fn target_slot(&self) -> Option<&FieldSlot> {
        self.target_slot.as_ref()
    }

    /// Borrow the canonical grouped aggregate input expression, if any.
    #[must_use]
    pub(in crate::db) const fn input_expr(&self) -> Option<&Expr> {
        self.identity.input_expr()
    }

    /// Borrow the canonical grouped aggregate filter expression, if any.
    #[must_use]
    pub(in crate::db) const fn filter_expr(&self) -> Option<&Expr> {
        self.filter_expr.as_ref()
    }

    /// Borrow the filter-aware semantic key represented by this execution spec.
    #[must_use]
    pub(in crate::db) fn semantic_key(&self) -> AggregateSemanticKeyRef<'_> {
        AggregateSemanticKeyRef::new(
            self.kind(),
            self.input_expr(),
            self.filter_expr(),
            self.distinct(),
        )
    }

    /// Return whether the grouped aggregate uses DISTINCT semantics.
    #[must_use]
    pub(in crate::db) const fn distinct(&self) -> bool {
        self.identity.distinct()
    }

    /// Return whether grouped DISTINCT for this aggregate uses value-based
    /// input deduplication instead of key-only tracking.
    #[must_use]
    pub(in crate::db) const fn uses_grouped_distinct_value_dedup(&self) -> bool {
        self.identity.uses_grouped_distinct_value_dedup()
    }

    /// Borrow the compiled grouped aggregate input expression used by runtime, if any.
    #[must_use]
    pub(in crate::db) const fn compiled_input_expr(&self) -> Option<&CompiledExpr> {
        self.compiled_input_expr.as_ref()
    }

    /// Borrow the compiled grouped aggregate filter expression used by runtime, if any.
    #[must_use]
    pub(in crate::db) const fn compiled_filter_expr(&self) -> Option<&CompiledExpr> {
        self.compiled_filter_expr.as_ref()
    }

    /// Return whether this grouped aggregate stays eligible for the dedicated
    /// grouped `COUNT(*)` fold path.
    #[must_use]
    pub(in crate::db) const fn admits_count_rows_dedicated_fold(&self) -> bool {
        matches!(self.kind(), AggregateKind::Count)
            && self.input_expr().is_none()
            && self.filter_expr().is_none()
            && !self.distinct()
    }

    /// Build one grouped aggregate execution spec directly for tests that do
    /// not carry a model-owned grouped lowering context.
    #[cfg(test)]
    #[must_use]
    pub(in crate::db) fn from_test_inputs(
        kind: AggregateKind,
        target_slot: Option<FieldSlot>,
        target_field: Option<&str>,
        distinct: bool,
    ) -> Self {
        Self {
            identity: AggregateIdentity::from_kind_input_and_distinct(
                kind,
                target_field
                    .map(|field| Expr::Field(crate::db::query::plan::expr::FieldId::new(field))),
                distinct,
            ),
            target_slot,
            filter_expr: None,
            compiled_input_expr: None,
            compiled_filter_expr: None,
        }
    }
}

impl PlannedProjectionLayout {
    /// Borrow grouped field positions in projection declaration order.
    #[must_use]
    pub(in crate::db) const fn group_field_positions(&self) -> &[usize] {
        self.group_field_positions.as_slice()
    }

    /// Borrow aggregate positions in projection declaration order.
    #[must_use]
    pub(in crate::db) const fn aggregate_positions(&self) -> &[usize] {
        self.aggregate_positions.as_slice()
    }

    /// Construct one grouped layout invariant for non-monotonic grouped-field positions.
    pub(in crate::db) fn group_field_positions_not_strictly_increasing() -> InternalError {
        InternalError::planner_executor_invariant()
    }

    /// Construct one grouped layout invariant for non-monotonic aggregate positions.
    pub(in crate::db) fn aggregate_positions_not_strictly_increasing() -> InternalError {
        InternalError::planner_executor_invariant()
    }

    /// Construct one grouped layout invariant for mixed field/aggregate ordering.
    pub(in crate::db) fn group_fields_must_precede_aggregates() -> InternalError {
        InternalError::planner_executor_invariant()
    }
}

///
/// GroupedFoldPath
///
/// Planner-carried grouped fold-path contract for executor runtime.
/// This is execution-mechanical only: it tells grouped runtime whether the
/// planner admitted the dedicated grouped `COUNT(*)` fold path or the
/// canonical generic grouped reducer path. Runtime must consume this contract
/// instead of branching on grouped planner strategy directly.
///

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(in crate::db) enum GroupedFoldPath {
    CountRowsDedicated,
    GenericReducers,
}

impl GroupedFoldPath {
    /// Project one grouped fold path from the planner-owned grouped strategy
    /// plus the fully resolved grouped aggregate execution shapes.
    #[must_use]
    pub(in crate::db) fn from_plan_strategy(
        strategy: GroupedPlanStrategy,
        aggregate_specs: &[GroupedAggregateExecutionSpec],
    ) -> Self {
        if strategy.is_single_count_rows()
            && aggregate_specs
                .iter()
                .all(GroupedAggregateExecutionSpec::admits_count_rows_dedicated_fold)
        {
            Self::CountRowsDedicated
        } else {
            Self::GenericReducers
        }
    }
}

///
/// GroupedExecutionRoute
///
/// Planner-carried grouped execution-route contract consumed by grouped
/// executor runtime.
/// This combines grouped DISTINCT, Top-K selection, and fold-path admission
/// into one closed artifact so executor stages do not recombine partial
/// planner signals to rediscover runtime behavior.
///

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(in crate::db) enum GroupedExecutionRoute {
    GlobalDistinctTopK,
    GlobalDistinctFull,
    CountRowsDedicated,
    GenericTopK,
    GenericFull,
}

impl GroupedExecutionRoute {
    /// Project the grouped execution route from planner-owned strategy outputs.
    #[must_use]
    pub(in crate::db) const fn from_planner_strategy(
        strategy: GroupedPlanStrategy,
        fold_path: GroupedFoldPath,
        distinct_strategy: &GroupedDistinctExecutionStrategy,
    ) -> Self {
        let uses_global_distinct = distinct_strategy.global_distinct_target_slot().is_some();
        let uses_top_k = strategy.is_top_k_group();

        match (uses_global_distinct, uses_top_k, fold_path) {
            (true, true, _) => Self::GlobalDistinctTopK,
            (true, false, _) => Self::GlobalDistinctFull,
            (false, true, _) => Self::GenericTopK,
            (false, false, GroupedFoldPath::CountRowsDedicated) => Self::CountRowsDedicated,
            (false, false, GroupedFoldPath::GenericReducers) => Self::GenericFull,
        }
    }

    /// Return whether this route uses the dedicated global DISTINCT fold lane.
    #[must_use]
    pub(in crate::db) const fn uses_global_distinct_fold(self) -> bool {
        matches!(self, Self::GlobalDistinctTopK | Self::GlobalDistinctFull)
    }

    /// Return whether this route uses bounded Top-K grouped selection.
    #[must_use]
    pub(in crate::db) const fn uses_top_k_group_selection(self) -> bool {
        matches!(self, Self::GlobalDistinctTopK | Self::GenericTopK)
    }

    /// Return whether this route admits the dedicated grouped COUNT(*) fold.
    #[must_use]
    pub(in crate::db) const fn uses_count_rows_dedicated_fold(self) -> bool {
        matches!(self, Self::CountRowsDedicated)
    }
}

///
/// GroupedExecutorHandoff
///
/// Borrowed grouped planning handoff consumed at the query->executor boundary.
/// This contract keeps grouped execution routing input explicit while grouped
/// runtime entry remains explicit at query->executor boundaries.
///

#[derive(Clone)]
pub(in crate::db) struct GroupedExecutorHandoff<'a> {
    base: &'a AccessPlannedQuery,
    group_fields: &'a crate::db::query::plan::GroupFieldSet,
    grouped_aggregate_execution_specs: &'a [GroupedAggregateExecutionSpec],
    projection_layout: PlannedProjectionLayout,
    projection_is_identity: bool,
    grouped_plan_strategy: GroupedPlanStrategy,
    grouped_execution_route: GroupedExecutionRoute,
    grouped_distinct_policy_contract: GroupedDistinctPolicyContract<'a>,
    execution: GroupedExecutionConfig,
}

impl<'a> GroupedExecutorHandoff<'a> {
    /// Borrow the grouped query base plan.
    #[must_use]
    pub(in crate::db) const fn base(&self) -> &'a AccessPlannedQuery {
        self.base
    }

    /// Borrow declared grouped key fields.
    #[must_use]
    pub(in crate::db) const fn group_fields(&self) -> &'a crate::db::query::plan::GroupFieldSet {
        self.group_fields
    }

    /// Return whether planner already proved the grouped projection is row-identical.
    #[must_use]
    pub(in crate::db) const fn projection_is_identity(&self) -> bool {
        self.projection_is_identity
    }

    /// Borrow grouped execution strategy projected by planner semantics.
    #[must_use]
    pub(in crate::db) const fn grouped_plan_strategy(&self) -> GroupedPlanStrategy {
        self.grouped_plan_strategy
    }

    /// Return planner-carried grouped execution-route selection.
    #[must_use]
    pub(in crate::db) const fn grouped_execution_route(&self) -> GroupedExecutionRoute {
        self.grouped_execution_route
    }

    /// Materialize owned residents only for execution; diagnostic handoffs borrow them.
    #[must_use]
    pub(in crate::db) fn into_route_stage_residents(
        self,
    ) -> (
        Vec<GroupedAggregateExecutionSpec>,
        PlannedProjectionLayout,
        GroupedDistinctExecutionStrategy,
    ) {
        (
            self.grouped_aggregate_execution_specs.to_vec(),
            self.projection_layout,
            self.grouped_distinct_policy_contract
                .execution_strategy()
                .clone(),
        )
    }

    /// Borrow grouped DISTINCT policy violation reason for executor boundaries.
    #[must_use]
    pub(in crate::db) const fn distinct_policy_violation_for_executor(
        &self,
    ) -> Option<GroupDistinctPolicyReason> {
        self.grouped_distinct_policy_contract
            .violation_for_executor()
    }

    /// Borrow grouped execution hard-limit policy selected by planning.
    #[must_use]
    pub(in crate::db) const fn execution(&self) -> GroupedExecutionConfig {
        self.execution
    }
}

/// Build one grouped executor handoff from one grouped logical plan.
pub(in crate::db) fn grouped_executor_handoff(
    plan: &AccessPlannedQuery,
) -> Result<GroupedExecutorHandoff<'_>, InternalError> {
    // Grouped handoff is valid only for plans with grouped execution payload.
    let Some(grouped) = plan.grouped_plan() else {
        return Err(InternalError::planner_executor_invariant());
    };
    let projection_spec = plan.frozen_projection_spec()?;
    let grouped_aggregate_execution_specs = plan
        .grouped_aggregate_execution_specs()
        .ok_or_else(InternalError::planner_executor_invariant)?;
    let (projection_layout, projection_is_identity) = planned_projection_layout_from_spec(
        projection_spec,
        &grouped.group.group_fields,
        grouped.group.aggregates.as_slice(),
        grouped_aggregate_execution_specs,
    )?;
    validate_grouped_projection_layout(&projection_layout)?;
    let grouped_plan_strategy = grouped_plan_strategy(plan, || plan.residual_filter_shape())?
        .ok_or_else(InternalError::planner_executor_invariant)?;
    let grouped_fold_path = if grouped.group.group_fields.as_path_aware().is_some() {
        GroupedFoldPath::GenericReducers
    } else {
        GroupedFoldPath::from_plan_strategy(
            grouped_plan_strategy,
            grouped_aggregate_execution_specs,
        )
    };
    let grouped_distinct_policy_contract = GroupedDistinctPolicyContract::new(
        match grouped_distinct_admissibility(grouped.scalar.distinct, grouped.having_expr.is_some())
        {
            GroupDistinctAdmissibility::Allowed => None,
            GroupDistinctAdmissibility::Disallowed(reason) => Some(reason),
        },
        plan.grouped_distinct_execution_strategy()
            .ok_or_else(InternalError::planner_executor_invariant)?,
    );
    let grouped_execution_route = GroupedExecutionRoute::from_planner_strategy(
        grouped_plan_strategy,
        grouped_fold_path,
        grouped_distinct_policy_contract.execution_strategy(),
    );

    Ok(GroupedExecutorHandoff {
        base: plan,
        group_fields: &grouped.group.group_fields,
        grouped_aggregate_execution_specs,
        projection_layout,
        projection_is_identity,
        grouped_plan_strategy,
        grouped_execution_route,
        grouped_distinct_policy_contract,
        execution: grouped.group.execution,
    })
}

/// Build grouped aggregate execution specs from planner-owned aggregate
/// projection specs and explicit schema authority.
pub(in crate::db) fn grouped_aggregate_execution_specs(
    schema_info: &SchemaInfo,
    mut aggregate_specs: Vec<GroupedAggregateExecutionSpec>,
    budget: &dyn crate::db::query::construction::ConstructionBudget,
) -> Result<Vec<GroupedAggregateExecutionSpec>, InternalError> {
    // The caller transfers its unpublished list. Reuse its backing instead of
    // cloning syntax into a second list; no partial list escapes on failure.
    for aggregate_spec in &mut aggregate_specs {
        aggregate_spec.resolve_with_schema_info(schema_info, budget)?;
    }
    Ok(aggregate_specs)
}

/// Lower grouped aggregate specs directly from canonical grouped
/// projection semantics without requiring a frozen grouped executor handoff.
pub(in crate::db) fn grouped_aggregate_specs_from_projection_spec(
    projection_spec: &ProjectionSpec,
    group_fields: &crate::db::query::plan::GroupFieldSet,
) -> Result<Vec<GroupedAggregateExecutionSpec>, InternalError> {
    // Group-reference strictness is test-only; production admission already ran.
    #[cfg(not(test))]
    let _ = group_fields;
    #[cfg(test)]
    validate_grouped_projection_references(projection_spec, group_fields)?;
    let mut aggregate_specs = Vec::new();
    for field in projection_spec.fields() {
        extend_unique_grouped_aggregate_specs_from_expr(&mut aggregate_specs, field.expr())?;
    }

    Ok(aggregate_specs)
}

///
/// GroupedDistinctPolicyContract
///
/// Planner-projected grouped DISTINCT policy contract for executor boundaries.
/// Route/load layers consume this contract directly instead of re-deriving
/// grouped DISTINCT policy from logical plan internals.
///

#[derive(Clone, Debug, Eq, PartialEq)]
pub(in crate::db) struct GroupedDistinctPolicyContract<'a> {
    violation_for_executor: Option<GroupDistinctPolicyReason>,
    execution_strategy: &'a GroupedDistinctExecutionStrategy,
}

impl<'a> GroupedDistinctPolicyContract<'a> {
    /// Construct one grouped DISTINCT policy contract.
    #[must_use]
    const fn new(
        violation_for_executor: Option<GroupDistinctPolicyReason>,
        execution_strategy: &'a GroupedDistinctExecutionStrategy,
    ) -> Self {
        Self {
            violation_for_executor,
            execution_strategy,
        }
    }

    /// Borrow grouped DISTINCT execution strategy lowered by planner.
    #[must_use]
    pub(in crate::db) const fn execution_strategy(&self) -> &GroupedDistinctExecutionStrategy {
        self.execution_strategy
    }

    /// Borrow grouped DISTINCT policy violation reason for executor boundaries.
    #[must_use]
    pub(in crate::db) const fn violation_for_executor(&self) -> Option<GroupDistinctPolicyReason> {
        self.violation_for_executor
    }
}

///
/// GroupedDistinctExecutionStrategy
///
/// Planner-owned grouped DISTINCT execution strategy lowered for executor consumption.
/// This strategy is mechanical-only and must not be re-derived by executor policy checks.
///

#[derive(Clone, Debug, Eq, PartialEq)]
pub(in crate::db) enum GroupedDistinctExecutionStrategy {
    None,
    GlobalDistinctFieldCount { target_slot: FieldSlot },
    GlobalDistinctFieldSum { target_slot: FieldSlot },
    GlobalDistinctFieldAvg { target_slot: FieldSlot },
}

impl GroupedDistinctExecutionStrategy {
    /// Borrow the planner-resolved target field slot used by grouped DISTINCT runtime.
    #[must_use]
    pub(in crate::db) const fn global_distinct_target_slot(&self) -> Option<&FieldSlot> {
        match self {
            Self::None => None,
            Self::GlobalDistinctFieldCount { target_slot }
            | Self::GlobalDistinctFieldSum { target_slot }
            | Self::GlobalDistinctFieldAvg { target_slot } => Some(target_slot),
        }
    }

    /// Return the canonical aggregate kind used by the dedicated global
    /// DISTINCT field-target runtime path.
    #[must_use]
    pub(in crate::db) const fn global_distinct_aggregate_kind(&self) -> Option<AggregateKind> {
        match self {
            Self::None => None,
            Self::GlobalDistinctFieldCount { .. } => Some(AggregateKind::Count),
            Self::GlobalDistinctFieldSum { .. } => Some(AggregateKind::Sum),
            Self::GlobalDistinctFieldAvg { .. } => Some(AggregateKind::Avg),
        }
    }

    /// Lower one supported grouped global-DISTINCT aggregate family into the
    /// executor-facing grouped DISTINCT strategy.
    #[must_use]
    pub(in crate::db) const fn from_supported_global_distinct(
        kind: GlobalDistinctAggregateKind,
        target_slot: FieldSlot,
    ) -> Self {
        match kind {
            GlobalDistinctAggregateKind::Count => Self::GlobalDistinctFieldCount { target_slot },
            GlobalDistinctAggregateKind::Sum => Self::GlobalDistinctFieldSum { target_slot },
            GlobalDistinctAggregateKind::Avg => Self::GlobalDistinctFieldAvg { target_slot },
        }
    }
}

// Lower grouped DISTINCT execution strategy from validated grouped planner semantics
// while freezing the field-target slot under planner ownership.
pub(in crate::db) fn resolved_grouped_distinct_execution_strategy_with_schema_info(
    schema_info: &SchemaInfo,
    group_fields: &crate::db::query::plan::GroupFieldSet,
    aggregates: &[GroupAggregateSpec],
    having_expr: Option<&crate::db::query::plan::expr::Expr>,
) -> Result<GroupedDistinctExecutionStrategy, InternalError> {
    match resolve_global_distinct_field_aggregate(group_fields, aggregates, having_expr) {
        Ok(Some(aggregate)) => {
            let target_slot = resolve_aggregate_target_field_slot_from_schema(
                schema_info,
                aggregate.target_field(),
            )
            .ok_or_else(InternalError::planner_executor_invariant)?;

            let distinct_kind = aggregate
                .kind()
                .global_distinct_kind()
                .ok_or_else(InternalError::planner_executor_invariant)?;

            Ok(
                GroupedDistinctExecutionStrategy::from_supported_global_distinct(
                    distinct_kind,
                    target_slot,
                ),
            )
        }
        Ok(None) => Ok(GroupedDistinctExecutionStrategy::None),
        Err(reason) => Err(reason.into_planner_handoff_internal_error()),
    }
}

// Resolve one aggregate target slot through schema authority while retaining
// the field label and kind for diagnostics/explain surfaces.
fn resolve_aggregate_target_field_slot_from_schema(
    schema_info: &SchemaInfo,
    field: &str,
) -> Option<FieldSlot> {
    FieldSlot::resolve_with_schema(schema_info, field)
}

// Derive layout against frozen aggregate slots without rebuilding their syntax.
fn planned_projection_layout_from_spec(
    projection_spec: &ProjectionSpec,
    group_fields: &crate::db::query::plan::GroupFieldSet,
    aggregates: &[GroupAggregateSpec],
    aggregate_specs: &[GroupedAggregateExecutionSpec],
) -> Result<(PlannedProjectionLayout, bool), InternalError> {
    #[cfg(test)]
    validate_grouped_projection_references(projection_spec, group_fields)?;
    let mut group_field_positions = Vec::new();
    let mut aggregate_positions = Vec::new();
    // HAVING-only slots may follow projection slots. Only first encounters in
    // this projection advance its identity cursor; repeated leaves reuse a slot.
    let mut seen_aggregates = vec![false; aggregate_specs.len()];
    let mut projection_is_identity =
        projection_spec.len() == group_fields.len().saturating_add(aggregates.len());
    let mut next_group_field_index = 0usize;
    let mut next_aggregate_index = 0usize;

    for (index, field) in projection_spec.fields().enumerate() {
        let root_expr = expression_without_alias(field.expr());
        let mut contains_aggregate = false;
        let mut introduced_aggregate_count = 0usize;
        root_expr.try_for_each_tree_aggregate(&mut |aggregate| {
            contains_aggregate = true;
            let key = AggregateSemanticKeyRef::from_aggregate_expr(aggregate);
            let slot = aggregate_specs
                .iter()
                .position(|spec| spec.semantic_key() == key)
                .ok_or_else(InternalError::planner_executor_invariant)?;
            if !seen_aggregates[slot] {
                seen_aggregates[slot] = true;
                introduced_aggregate_count = introduced_aggregate_count.saturating_add(1);
            }
            Ok::<(), InternalError>(())
        })?;

        match root_expr {
            Expr::Field(_) | Expr::FieldPath(_) => {
                group_field_positions.push(index);
                projection_is_identity &= next_aggregate_index == 0
                    && group_fields
                        .get(next_group_field_index)
                        .is_some_and(|group_field| group_field.matches_expr(root_expr));
                next_group_field_index = next_group_field_index.saturating_add(1);
            }
            Expr::Aggregate(aggregate_expr) => {
                aggregate_positions.push(index);
                let aggregate_key = AggregateSemanticKeyRef::from_aggregate_expr(aggregate_expr);
                projection_is_identity &= next_group_field_index == group_fields.len()
                    && aggregates
                        .get(next_aggregate_index)
                        .is_some_and(|aggregate| aggregate_key == aggregate.semantic_key());
                next_aggregate_index =
                    next_aggregate_index.saturating_add(introduced_aggregate_count);
            }
            _ if contains_aggregate => {
                aggregate_positions.push(index);
                projection_is_identity = false;
                next_aggregate_index =
                    next_aggregate_index.saturating_add(introduced_aggregate_count);
            }
            _ => {
                // Computed scalar outputs occupy group positions but require
                // evaluation; only the field-leaf arm can preserve identity.
                group_field_positions.push(index);
                projection_is_identity = false;
                next_group_field_index = next_group_field_index.saturating_add(1);
            }
        }
    }
    projection_is_identity &=
        next_group_field_index == group_fields.len() && next_aggregate_index == aggregates.len();

    Ok((
        PlannedProjectionLayout {
            group_field_positions,
            aggregate_positions,
        },
        projection_is_identity,
    ))
}

#[cfg(test)]
fn validate_grouped_projection_references(
    projection_spec: &ProjectionSpec,
    group_fields: &crate::db::query::plan::GroupFieldSet,
) -> Result<(), InternalError> {
    // Test builds keep one extra strictness pass so grouped layout regressions
    // fail at the planner boundary instead of only in downstream assertions.
    crate::db::query::preparation::with_preparation_work(|work| {
        for field in projection_spec.fields() {
            let root_expr = expression_without_alias(field.expr());
            if !group_fields.try_contains_all_expr_references(root_expr, &mut |steps| {
                crate::db::query::construction::ConstructionBudget::charge(
                    work,
                    Resource::PredicateExpressionSteps,
                    steps,
                )
            })? {
                return Err(InternalError::planner_executor_invariant());
            }
        }
        Ok(())
    })
}

pub(in crate::db::query::plan) fn extend_unique_grouped_aggregate_specs_from_expr(
    aggregate_specs: &mut Vec<GroupedAggregateExecutionSpec>,
    expr: &Expr,
) -> Result<(), InternalError> {
    expr.try_for_each_tree_aggregate(&mut |aggregate_expr| {
        push_unique_grouped_aggregate_spec(aggregate_specs, aggregate_expr);

        Ok::<(), InternalError>(())
    })
}

// Keep grouped aggregate specs on one stable first-seen unique
// order so repeated aggregate leaves reuse the same grouped execution slot.
// This collection precedes schema resolution/compilation; compare borrowed
// semantic keys before retaining a new input/filter expression copy.
fn push_unique_grouped_aggregate_spec(
    aggregate_specs: &mut Vec<GroupedAggregateExecutionSpec>,
    aggregate_expr: &AggregateExpr,
) {
    let aggregate_key = AggregateSemanticKeyRef::from_aggregate_expr(aggregate_expr);
    if aggregate_specs
        .iter()
        .all(|current| current.semantic_key() != aggregate_key)
    {
        aggregate_specs.push(GroupedAggregateExecutionSpec::from_aggregate_expr(
            aggregate_expr,
        ));
    }
}

// Strip alias wrappers so layout classification uses semantic expression roots.
#[cfg_attr(
    not(test),
    expect(
        clippy::missing_const_for_fn,
        reason = "test-only alias stripping keeps the shared helper non-const across the full target matrix"
    )
)]
fn expression_without_alias(expr: &Expr) -> &Expr {
    #[cfg(test)]
    {
        let mut current = expr;
        while let Expr::Alias { expr: inner, .. } = current {
            current = inner.as_ref();
        }

        current
    }

    #[cfg(not(test))]
    expr
}

// Exhaustive cache-retention coverage; new owned fields require accounting.
crate::retained::retained_fields!(GroupedAggregateExecutionSpec {
Self{identity,target_slot,filter_expr,compiled_input_expr,compiled_filter_expr} => [identity,target_slot,filter_expr,compiled_input_expr,compiled_filter_expr],
});
crate::retained::retained_fields!(GroupedDistinctExecutionStrategy {
Self::None => [],
Self::GlobalDistinctFieldCount{target_slot} => [target_slot],
Self::GlobalDistinctFieldSum{target_slot} => [target_slot],
Self::GlobalDistinctFieldAvg{target_slot} => [target_slot],
});