icydb-core 0.138.7

mod terminal;

use crate::db::sql::lowering::aggregate::terminal::{AggregateInput, SqlGlobalAggregateTerminal};
use crate::db::sql::lowering::{
    LoweredBaseQueryShape, LoweredSqlCommand, LoweredSqlCommandInner, LoweredSqlFilter,
    PreparedSqlStatement, SqlLoweringError, analyze_lowered_expr,
    predicate::{
        lower_sql_pre_aggregate_bool_expr, lower_sql_where_bool_expr, lower_sql_where_expr,
    },
};
#[cfg(test)]
use crate::{db::query::intent::Query, traits::EntityKind};
use crate::{
    db::{
        executor::{StructuralAggregateTerminal, StructuralAggregateTerminalKind},
        predicate::MissingRowPolicy,
        query::{
            builder::{
                AggregateExpr,
                aggregate::{avg, count, count_by, max_by, min_by, sum},
            },
            intent::StructuralQuery,
            plan::{
                AggregateKind, FieldSlot,
                expr::{
                    Alias, Expr, ProjectionField, ProjectionSpec,
                    canonicalize_aggregate_input_expr, compile_scalar_projection_expr,
                },
                lower_global_aggregate_projection, resolve_aggregate_target_field_slot,
            },
        },
        sql::{
            lowering::expr::{SqlExprPhase, lower_sql_expr},
            lowering::select::{
                lower_global_aggregate_having_expr, lower_order_terms, lower_select_item_expr,
            },
            parser::{
                SqlAggregateCall, SqlAggregateKind, SqlExplainMode, SqlExpr, SqlProjection,
                SqlSelectItem, SqlSelectStatement, SqlStatement,
            },
        },
    },
    model::entity::EntityModel,
};

/// PreparedSqlScalarAggregateRuntimeDescriptor
///
/// Stable runtime-family projection for one prepared typed SQL scalar
/// aggregate strategy.
/// Session SQL aggregate execution consumes this descriptor instead of
/// rebuilding runtime boundary choice from raw SQL terminal variants or
/// parallel metadata tuple matches.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum PreparedSqlScalarAggregateRuntimeDescriptor {
    CountRows,
    CountField,
    NumericField { kind: AggregateKind },
    ExtremalWinnerField { kind: AggregateKind },
}

pub(crate) type PreparedSqlScalarAggregateDescriptorShape =
    PreparedSqlScalarAggregateRuntimeDescriptor;

impl PreparedSqlScalarAggregateRuntimeDescriptor {
    // Map supported SQL aggregate kinds onto the field/expression descriptor
    // family that will own runtime dispatch and EXPLAIN labeling.
    fn targeted(kind: AggregateKind) -> Self {
        match kind {
            AggregateKind::Count => Self::CountField,
            AggregateKind::Sum | AggregateKind::Avg => Self::NumericField { kind },
            AggregateKind::Min | AggregateKind::Max => Self::ExtremalWinnerField { kind },
            AggregateKind::Exists | AggregateKind::First | AggregateKind::Last => {
                unreachable!("unsupported SQL aggregate kind reached scalar aggregate descriptor")
            }
        }
    }

    /// Return the stable runtime-family projection for this descriptor shape.
    #[must_use]
    pub(crate) const fn runtime_descriptor(self) -> Self {
        self
    }

    /// Return the canonical aggregate kind for this descriptor shape.
    #[must_use]
    pub(crate) const fn aggregate_kind(self) -> AggregateKind {
        match self {
            Self::CountRows | Self::CountField => AggregateKind::Count,
            Self::NumericField { kind } | Self::ExtremalWinnerField { kind } => kind,
        }
    }
}

///
/// PreparedSqlScalarAggregateStrategy
///
/// PreparedSqlScalarAggregateStrategy is the single typed SQL scalar aggregate
/// behavior source for the first `0.71` slice.
/// It resolves aggregate domain, descriptor shape, target-slot ownership, and
/// runtime behavior once so runtime and EXPLAIN do not re-derive that
/// behavior from raw SQL terminal variants.
/// Explain-visible aggregate expressions are projected on demand from this
/// prepared strategy instead of being carried as owned execution metadata.
///
#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) struct PreparedSqlScalarAggregateStrategy {
    target_slot: Option<FieldSlot>,
    input_expr: Option<Expr>,
    filter_expr: Option<Expr>,
    distinct_input: bool,
    descriptor_shape: PreparedSqlScalarAggregateDescriptorShape,
}

impl PreparedSqlScalarAggregateStrategy {
    // Build one prepared aggregate strategy from the already-resolved target
    // slot and descriptor shape so higher entrypoints only own target
    // resolution, not the descriptor policy bundle.
    pub(in crate::db) const fn from_resolved_shape(
        target_slot: Option<FieldSlot>,
        input_expr: Option<Expr>,
        filter_expr: Option<Expr>,
        distinct_input: bool,
        descriptor_shape: PreparedSqlScalarAggregateDescriptorShape,
    ) -> Self {
        Self {
            target_slot,
            input_expr,
            filter_expr,
            distinct_input,
            descriptor_shape,
        }
    }

    const fn for_rows(filter_expr: Option<Expr>) -> Self {
        Self::from_resolved_shape(
            None,
            None,
            filter_expr,
            false,
            PreparedSqlScalarAggregateDescriptorShape::CountRows,
        )
    }

    fn for_field(
        kind: AggregateKind,
        target_slot: FieldSlot,
        filter_expr: Option<Expr>,
        distinct_input: bool,
    ) -> Self {
        Self::from_resolved_shape(
            Some(target_slot),
            None,
            filter_expr,
            distinct_input,
            PreparedSqlScalarAggregateDescriptorShape::targeted(kind),
        )
    }

    fn for_expr(
        kind: AggregateKind,
        input_expr: Expr,
        filter_expr: Option<Expr>,
        distinct_input: bool,
    ) -> Self {
        Self::from_resolved_shape(
            None,
            Some(input_expr),
            filter_expr,
            distinct_input,
            PreparedSqlScalarAggregateDescriptorShape::targeted(kind),
        )
    }

    // Keep terminal preparation on one owner-local seam so field-target and
    // expression-input aggregate shapes cannot drift apart across parallel
    // helpers.
    fn from_lowered_terminal(
        model: &'static EntityModel,
        terminal: SqlGlobalAggregateTerminal,
    ) -> Result<Self, SqlLoweringError> {
        let target_slot = resolve_target_slot_if_field(model, &terminal)?;
        if terminal.is_expr() {
            let Some(input_expr) = terminal.expr() else {
                return Err(SqlLoweringError::unsupported_global_aggregate_projection());
            };
            validate_expr_if_present(model, input_expr)?;
        }

        match terminal.input {
            AggregateInput::Rows => Ok(Self::for_rows(terminal.filter_expr)),
            AggregateInput::Field(_) => {
                let Some(target_slot) = target_slot else {
                    return Err(SqlLoweringError::unsupported_global_aggregate_projection());
                };
                Ok(Self::for_field(
                    terminal.kind,
                    target_slot,
                    terminal.filter_expr,
                    terminal.distinct,
                ))
            }
            AggregateInput::Expr(input_expr) => Ok(Self::for_expr(
                terminal.kind,
                input_expr,
                terminal.filter_expr,
                terminal.distinct,
            )),
        }
    }

    /// Borrow the resolved target slot when this prepared SQL scalar strategy is field-targeted.
    #[must_use]
    pub(crate) const fn target_slot(&self) -> Option<&FieldSlot> {
        self.target_slot.as_ref()
    }

    /// Borrow the aggregate input expression when this prepared SQL scalar strategy is expression-backed.
    #[cfg(test)]
    #[must_use]
    pub(crate) const fn input_expr(&self) -> Option<&Expr> {
        self.input_expr.as_ref()
    }

    /// Borrow the aggregate filter expression when this prepared SQL scalar strategy is filtered.
    #[must_use]
    pub(crate) const fn filter_expr(&self) -> Option<&Expr> {
        self.filter_expr.as_ref()
    }

    /// Return whether this prepared SQL scalar aggregate deduplicates field inputs.
    #[cfg(test)]
    #[must_use]
    pub(crate) const fn is_distinct(&self) -> bool {
        self.distinct_input
    }

    /// Return the stable descriptor/runtime shape label for this prepared strategy.
    #[cfg(test)]
    #[must_use]
    pub(crate) const fn descriptor_shape(&self) -> PreparedSqlScalarAggregateDescriptorShape {
        self.descriptor_shape
    }

    /// Return the stable runtime-family projection for this prepared SQL
    /// scalar aggregate strategy.
    #[cfg(test)]
    #[must_use]
    pub(crate) const fn runtime_descriptor(&self) -> PreparedSqlScalarAggregateRuntimeDescriptor {
        self.descriptor_shape.runtime_descriptor()
    }

    /// Return the canonical aggregate kind for this prepared SQL scalar strategy.
    #[must_use]
    pub(crate) const fn aggregate_kind(&self) -> AggregateKind {
        self.descriptor_shape.aggregate_kind()
    }

    /// Build the executor terminal consumed by structural aggregate execution.
    ///
    /// SQL lowering owns the aggregate strategy and therefore owns the final
    /// executor terminal construction. Session execution keeps only SQL result
    /// labels, fixed scales, cache attribution, and statement shaping.
    pub(in crate::db) fn into_executor_terminal(
        self,
    ) -> Result<StructuralAggregateTerminal, &'static str> {
        let Self {
            target_slot,
            input_expr,
            filter_expr,
            distinct_input,
            descriptor_shape,
        } = self;

        let kind = match descriptor_shape.runtime_descriptor() {
            PreparedSqlScalarAggregateRuntimeDescriptor::CountRows => {
                StructuralAggregateTerminalKind::CountRows
            }
            PreparedSqlScalarAggregateRuntimeDescriptor::CountField => {
                StructuralAggregateTerminalKind::CountValues
            }
            PreparedSqlScalarAggregateRuntimeDescriptor::NumericField {
                kind: AggregateKind::Sum,
            } => StructuralAggregateTerminalKind::Sum,
            PreparedSqlScalarAggregateRuntimeDescriptor::NumericField {
                kind: AggregateKind::Avg,
            } => StructuralAggregateTerminalKind::Avg,
            PreparedSqlScalarAggregateRuntimeDescriptor::ExtremalWinnerField {
                kind: AggregateKind::Min,
            } => StructuralAggregateTerminalKind::Min,
            PreparedSqlScalarAggregateRuntimeDescriptor::ExtremalWinnerField {
                kind: AggregateKind::Max,
            } => StructuralAggregateTerminalKind::Max,
            PreparedSqlScalarAggregateRuntimeDescriptor::NumericField { .. }
            | PreparedSqlScalarAggregateRuntimeDescriptor::ExtremalWinnerField { .. } => {
                return Err("prepared SQL scalar aggregate strategy drifted outside SQL support");
            }
        };

        Ok(StructuralAggregateTerminal::new(
            kind,
            target_slot,
            input_expr,
            filter_expr,
            distinct_input,
        ))
    }

    /// Return the projected field label for descriptor/explain projection when
    /// this prepared strategy is field-targeted.
    #[must_use]
    pub(crate) fn projected_field(&self) -> Option<&str> {
        self.target_slot().map(FieldSlot::field)
    }
}

fn resolve_target_slot_if_field(
    model: &'static EntityModel,
    terminal: &SqlGlobalAggregateTerminal,
) -> Result<Option<FieldSlot>, SqlLoweringError> {
    if !terminal.is_field() {
        return Ok(None);
    }

    terminal
        .field()
        .map(|field| resolve_aggregate_target_field_slot(model, field))
        .transpose()
        .map_err(SqlLoweringError::from)
}

fn validate_expr_if_present(
    model: &'static EntityModel,
    input_expr: &Expr,
) -> Result<(), SqlLoweringError> {
    if let Some(field) = analyze_lowered_expr(input_expr, Some(model)).first_unknown_field() {
        return Err(SqlLoweringError::unknown_field(field));
    }
    if compile_scalar_projection_expr(model, input_expr).is_none() {
        return Err(SqlLoweringError::unsupported_aggregate_input_expressions());
    }

    Ok(())
}

///
/// LoweredSqlGlobalAggregateCommand
///
/// Generic-free global aggregate command shape prepared before typed query
/// binding.
/// This keeps aggregate SQL lowering shared across entities until the final
/// execution boundary converts the base query shape into `Query<E>`.
///
#[derive(Clone, Debug)]
pub(crate) struct LoweredSqlGlobalAggregateCommand {
    pub(in crate::db::sql::lowering) query: LoweredBaseQueryShape,
    pub(in crate::db::sql::lowering) terminals: Vec<SqlGlobalAggregateTerminal>,
    pub(in crate::db::sql::lowering) projection: ProjectionSpec,
    pub(in crate::db::sql::lowering) having: Option<Expr>,
    #[cfg(test)]
    pub(in crate::db::sql::lowering) output_remap: Vec<usize>,
}

impl LoweredSqlGlobalAggregateCommand {
    /// Lower one constrained global aggregate select into the generic-free
    /// command shape shared by typed and structural aggregate binders.
    fn from_select_statement(statement: SqlSelectStatement) -> Result<Self, SqlLoweringError> {
        let SqlSelectStatement {
            projection,
            projection_aliases,
            predicate,
            distinct,
            group_by,
            having,
            order_by,
            limit,
            offset,
            entity: _,
            table_alias: _,
        } = statement;

        if distinct {
            return Err(SqlLoweringError::unsupported_select_distinct());
        }
        if !group_by.is_empty() {
            return Err(SqlLoweringError::global_aggregate_does_not_support_group_by());
        }
        let projection_for_having = projection.clone();
        let order_by = strip_inert_global_aggregate_output_order_terms(
            order_by,
            &projection_for_having,
            projection_aliases.as_slice(),
        )?;

        let mut lowered_terminals =
            LoweredSqlGlobalAggregateTerminals::from_projection(projection, &projection_aliases)?;
        let having =
            lower_global_aggregate_having_expr(having, &projection_for_having, |aggregate| {
                resolve_or_insert_global_aggregate_terminal_index_from_expr(
                    &mut lowered_terminals.terminals,
                    aggregate,
                )
            })?;

        Ok(Self {
            query: LoweredBaseQueryShape {
                filter: predicate
                    .as_ref()
                    .map(|expr| {
                        Ok::<_, SqlLoweringError>(
                            LoweredSqlFilter::from_visible_expr_and_predicate_subset(
                                lower_sql_where_bool_expr(expr)?,
                                lower_sql_where_expr(expr)?,
                            ),
                        )
                    })
                    .transpose()?,
                order_by: lower_order_terms(order_by)?,
                limit,
                offset,
            },
            terminals: lowered_terminals.terminals,
            projection: lowered_terminals.projection,
            having,
            #[cfg(test)]
            output_remap: lowered_terminals.output_remap,
        })
    }

    /// Bind this lowered aggregate command onto one entity-owned typed query.
    #[cfg(test)]
    fn into_typed<E: EntityKind>(
        self,
        consistency: MissingRowPolicy,
    ) -> Result<SqlGlobalAggregateCommand<E>, SqlLoweringError> {
        let Self {
            query,
            terminals,
            projection,
            having,
            output_remap,
        } = self;

        let terminals = terminals
            .into_iter()
            .map(|terminal| {
                PreparedSqlScalarAggregateStrategy::from_lowered_terminal(E::MODEL, terminal)
            })
            .collect::<Result<Vec<_>, _>>()?;

        Ok(SqlGlobalAggregateCommand {
            query: Query::from_inner(crate::db::sql::lowering::apply_lowered_base_query_shape(
                StructuralQuery::new(E::MODEL, consistency),
                query,
            )),
            terminals,
            projection,
            having,
            output_remap,
        })
    }

    /// Bind this lowered aggregate command onto the structural query surface
    /// used by aggregate explain and dynamic SQL execution.
    fn into_structural(
        self,
        model: &'static EntityModel,
        consistency: MissingRowPolicy,
    ) -> Result<SqlGlobalAggregateCommandCore, SqlLoweringError> {
        let Self {
            query,
            terminals,
            projection,
            having,
            #[cfg(test)]
                output_remap: _,
        } = self;

        let strategies = terminals
            .into_iter()
            .map(|terminal| {
                PreparedSqlScalarAggregateStrategy::from_lowered_terminal(model, terminal)
            })
            .collect::<Result<Vec<_>, _>>()?;

        Ok(SqlGlobalAggregateCommandCore {
            query: crate::db::sql::lowering::apply_lowered_base_query_shape(
                StructuralQuery::new(model, consistency),
                query,
            ),
            strategies,
            projection,
            having,
        })
    }
}

// Drop singleton-result ORDER BY terms that target the global aggregate output
// row itself, while preserving base-row ordering used to shape the aggregate input window.
fn strip_inert_global_aggregate_output_order_terms(
    order_by: Vec<crate::db::sql::parser::SqlOrderTerm>,
    projection: &SqlProjection,
    projection_aliases: &[Option<String>],
) -> Result<Vec<crate::db::sql::parser::SqlOrderTerm>, SqlLoweringError> {
    let inert_targets =
        collect_global_aggregate_output_order_targets(projection, projection_aliases)?;

    Ok(order_by
        .into_iter()
        .filter(|term| !inert_targets.iter().any(|target| target == &term.field))
        .collect())
}

// Collect the canonical ORDER BY spellings that refer to the singleton global
// aggregate output row so the dedicated aggregate lane can ignore them instead
// of re-deriving them as base-row ordering.
fn collect_global_aggregate_output_order_targets(
    projection: &SqlProjection,
    projection_aliases: &[Option<String>],
) -> Result<Vec<crate::db::sql::parser::SqlExpr>, SqlLoweringError> {
    let SqlProjection::Items(items) = projection else {
        return Ok(Vec::new());
    };

    let mut targets = Vec::with_capacity(items.len());
    for (item, alias) in items.iter().zip(projection_aliases.iter()) {
        let expr = lower_select_item_expr(item, SqlExprPhase::PostAggregate)?;
        let analysis = analyze_lowered_expr(&expr, None);
        if !analysis.contains_aggregate() || analysis.references_direct_fields() {
            continue;
        }

        targets.push(crate::db::sql::parser::SqlExpr::from_select_item(item));
        if let Some(alias) = alias {
            targets.push(crate::db::sql::parser::SqlExpr::Field(alias.clone()));
        }
    }

    Ok(targets)
}

///
/// LoweredSqlAggregateShape
///
/// Locally validated aggregate-call shape used by SQL lowering to avoid
/// duplicating `(SqlAggregateKind, field)` validation across lowering lanes.
///
enum LoweredSqlAggregateShape {
    CountRows {
        filter_expr: Option<Expr>,
    },
    CountField {
        field: String,
        filter_expr: Option<Expr>,
        distinct: bool,
    },
    FieldTarget {
        kind: SqlAggregateKind,
        field: String,
        filter_expr: Option<Expr>,
        distinct: bool,
    },
    ExpressionInput {
        kind: SqlAggregateKind,
        input_expr: Expr,
        filter_expr: Option<Expr>,
        distinct: bool,
    },
}

///
/// SqlGlobalAggregateCommand
///
/// Lowered global SQL aggregate command carrying base query shape plus terminal.
///
#[cfg(test)]
#[derive(Debug)]
pub(crate) struct SqlGlobalAggregateCommand<E: EntityKind> {
    query: Query<E>,
    terminals: Vec<PreparedSqlScalarAggregateStrategy>,
    projection: ProjectionSpec,
    having: Option<Expr>,
    output_remap: Vec<usize>,
}

#[cfg(test)]
impl<E: EntityKind> SqlGlobalAggregateCommand<E> {
    /// Borrow the lowered base query shape for aggregate execution.
    #[must_use]
    pub(crate) const fn query(&self) -> &Query<E> {
        &self.query
    }

    /// Borrow the lowered aggregate terminals.
    #[must_use]
    pub(crate) fn terminals(&self) -> &[PreparedSqlScalarAggregateStrategy] {
        self.terminals.as_slice()
    }

    /// Borrow the canonical output projection contract for this global aggregate command.
    #[must_use]
    #[cfg(test)]
    pub(crate) const fn projection(&self) -> &ProjectionSpec {
        &self.projection
    }

    /// Borrow the optional global aggregate HAVING expression.
    #[must_use]
    #[cfg(test)]
    pub(crate) const fn having(&self) -> Option<&Expr> {
        self.having.as_ref()
    }

    /// Borrow the output-to-unique-terminal remap preserved from original SQL projection order.
    #[cfg(test)]
    #[must_use]
    pub(crate) fn output_remap(&self) -> &[usize] {
        self.output_remap.as_slice()
    }

    /// Borrow the first lowered aggregate terminal for single-terminal callers.
    #[cfg(test)]
    #[must_use]
    pub(crate) fn terminal(&self) -> &PreparedSqlScalarAggregateStrategy {
        self.terminals
            .first()
            .expect("global aggregate command must contain at least one terminal")
    }
}

///
/// SqlGlobalAggregateCommandCore
///
/// Generic-free lowered global aggregate command bound onto the structural
/// query surface.
/// This keeps global aggregate EXPLAIN on the shared query/explain path until
/// a typed boundary is strictly required.
///
#[derive(Clone, Debug)]
pub(crate) struct SqlGlobalAggregateCommandCore {
    query: StructuralQuery,
    strategies: Vec<PreparedSqlScalarAggregateStrategy>,
    projection: ProjectionSpec,
    having: Option<Expr>,
}

impl SqlGlobalAggregateCommandCore {
    /// Borrow the structural query payload for aggregate explain/execution.
    #[must_use]
    pub(in crate::db) const fn query(&self) -> &StructuralQuery {
        &self.query
    }

    /// Borrow prepared structural SQL scalar aggregate strategies.
    #[must_use]
    pub(in crate::db) const fn strategies(&self) -> &[PreparedSqlScalarAggregateStrategy] {
        self.strategies.as_slice()
    }

    /// Move the structural aggregate execution parts out of this command.
    #[must_use]
    pub(in crate::db) fn into_execution_parts(
        self,
    ) -> (
        StructuralQuery,
        Vec<PreparedSqlScalarAggregateStrategy>,
        ProjectionSpec,
        Option<Expr>,
    ) {
        (self.query, self.strategies, self.projection, self.having)
    }
}

impl SqlStatement {
    /// Return whether this parsed SQL statement is an executable constrained
    /// global aggregate shape owned by the dedicated aggregate lane.
    #[must_use]
    pub(in crate::db) fn is_global_aggregate_lane_shape(&self) -> bool {
        let Self::Select(statement) = self else {
            return false;
        };

        statement.is_global_aggregate_lane_shape()
    }
}

impl SqlSelectStatement {
    /// Return whether this parsed SELECT shape can route onto the dedicated
    /// global aggregate lowering lane.
    #[must_use]
    fn is_global_aggregate_lane_shape(&self) -> bool {
        if self.distinct || !self.group_by.is_empty() {
            return false;
        }

        // Skip the heavier global-aggregate shape lowering when one plain scalar
        // SELECT cannot possibly route onto the dedicated aggregate lane.
        if !self.might_require_global_aggregate_lane() {
            return false;
        }

        LoweredSqlGlobalAggregateCommand::from_select_statement(self.clone()).is_ok()
    }

    // Use one cheap parsed-shape screen before the dedicated aggregate lane opens
    // the full lowering path. Plain scalar selects with no HAVING and no aggregate
    // projection items can never become executable global aggregates.
    fn might_require_global_aggregate_lane(&self) -> bool {
        if !self.having.is_empty() {
            return true;
        }

        match &self.projection {
            SqlProjection::Items(items) => items.iter().any(SqlSelectItem::contains_aggregate),
            SqlProjection::All => false,
        }
    }
}

/// Bind one lowered global aggregate EXPLAIN shape onto the structural query
/// surface when the explain command carries that specialized form.
pub(crate) fn bind_lowered_sql_explain_global_aggregate_structural(
    lowered: &LoweredSqlCommand,
    model: &'static EntityModel,
    consistency: MissingRowPolicy,
) -> Result<Option<(SqlExplainMode, bool, SqlGlobalAggregateCommandCore)>, SqlLoweringError> {
    let LoweredSqlCommandInner::ExplainGlobalAggregate {
        mode,
        verbose,
        command,
    } = &lowered.0
    else {
        return Ok(None);
    };

    Ok(Some((
        *mode,
        *verbose,
        bind_lowered_sql_global_aggregate_command_structural(model, command.clone(), consistency)?,
    )))
}

/// Parse and lower one SQL statement into global aggregate execution command for `E`.
#[cfg(test)]
pub(crate) fn compile_sql_global_aggregate_command<E: EntityKind>(
    sql: &str,
    consistency: MissingRowPolicy,
) -> Result<SqlGlobalAggregateCommand<E>, SqlLoweringError> {
    let statement = crate::db::sql::parser::parse_sql(sql)?;
    let prepared = crate::db::sql::lowering::prepare_sql_statement(statement, E::MODEL.name())?;

    compile_sql_global_aggregate_command_from_prepared::<E>(prepared, consistency)
}

// Lower one already-prepared SQL statement into the constrained global
// aggregate command envelope so callers that already parsed and routed the
// statement do not pay the parser again.
#[cfg(test)]
pub(crate) fn compile_sql_global_aggregate_command_from_prepared<E: EntityKind>(
    prepared: PreparedSqlStatement,
    consistency: MissingRowPolicy,
) -> Result<SqlGlobalAggregateCommand<E>, SqlLoweringError> {
    let SqlStatement::Select(statement) = prepared.statement else {
        return Err(SqlLoweringError::unsupported_select_projection());
    };

    bind_lowered_sql_global_aggregate_command::<E>(
        lower_global_aggregate_select_shape(statement)?,
        consistency,
    )
}

// Lower one already-prepared SQL statement into the generic-free global
// aggregate command envelope so dynamic SQL surfaces can share the same
// aggregate-shape authority before choosing their outward payload contract.
pub(in crate::db) fn compile_sql_global_aggregate_command_core_from_prepared(
    prepared: PreparedSqlStatement,
    model: &'static EntityModel,
    consistency: MissingRowPolicy,
) -> Result<SqlGlobalAggregateCommandCore, SqlLoweringError> {
    let SqlStatement::Select(statement) = prepared.statement else {
        return Err(SqlLoweringError::unsupported_select_projection());
    };

    bind_lowered_sql_global_aggregate_command_structural(
        model,
        lower_global_aggregate_select_shape(statement)?,
        consistency,
    )
}

pub(in crate::db::sql::lowering) fn lower_global_aggregate_select_shape(
    statement: SqlSelectStatement,
) -> Result<LoweredSqlGlobalAggregateCommand, SqlLoweringError> {
    LoweredSqlGlobalAggregateCommand::from_select_statement(statement)
}

#[cfg(test)]
pub(in crate::db::sql::lowering) fn bind_lowered_sql_global_aggregate_command<E: EntityKind>(
    lowered: LoweredSqlGlobalAggregateCommand,
    consistency: MissingRowPolicy,
) -> Result<SqlGlobalAggregateCommand<E>, SqlLoweringError> {
    lowered.into_typed::<E>(consistency)
}

fn bind_lowered_sql_global_aggregate_command_structural(
    model: &'static EntityModel,
    lowered: LoweredSqlGlobalAggregateCommand,
    consistency: MissingRowPolicy,
) -> Result<SqlGlobalAggregateCommandCore, SqlLoweringError> {
    lowered.into_structural(model, consistency)
}

fn resolve_or_insert_global_aggregate_terminal_index_from_expr(
    terminals: &mut Vec<SqlGlobalAggregateTerminal>,
    aggregate_expr: &AggregateExpr,
) -> Result<usize, SqlLoweringError> {
    let terminal = SqlGlobalAggregateTerminal::from_aggregate_expr(aggregate_expr)?;

    Ok(terminals
        .iter()
        .position(|current| current == &terminal)
        .unwrap_or_else(|| {
            let index = terminals.len();
            terminals.push(terminal);
            index
        }))
}

pub(in crate::db::sql::lowering) fn resolve_having_aggregate_expr_index(
    target: &AggregateExpr,
    grouped_projection_aggregates: &[SqlAggregateCall],
) -> Result<usize, SqlLoweringError> {
    let mut matched =
        grouped_projection_aggregates
            .iter()
            .enumerate()
            .filter_map(|(index, aggregate)| {
                lower_aggregate_call(aggregate.clone())
                    .ok()
                    .filter(|current| current == target)
                    .map(|_| index)
            });
    let Some(index) = matched.next() else {
        return Err(SqlLoweringError::unsupported_select_having());
    };
    if matched.next().is_some() {
        return Err(SqlLoweringError::unsupported_select_having());
    }

    Ok(index)
}

///
/// LoweredSqlGlobalAggregateTerminals
///
/// Canonical global aggregate lowering result that keeps only unique
/// executable terminals plus one remap back to original SQL projection order.
///
struct LoweredSqlGlobalAggregateTerminals {
    terminals: Vec<SqlGlobalAggregateTerminal>,
    projection: ProjectionSpec,
    #[cfg(test)]
    output_remap: Vec<usize>,
}

impl LoweredSqlGlobalAggregateTerminals {
    /// Lower one SQL projection into unique executable aggregate terminals plus
    /// the output remap needed to preserve original projection order.
    fn from_projection(
        projection: SqlProjection,
        projection_aliases: &[Option<String>],
    ) -> Result<Self, SqlLoweringError> {
        let SqlProjection::Items(items) = projection else {
            return Err(SqlLoweringError::unsupported_global_aggregate_projection());
        };
        if items.is_empty() {
            return Err(SqlLoweringError::unsupported_global_aggregate_projection());
        }

        let mut terminals = Vec::<SqlGlobalAggregateTerminal>::with_capacity(items.len());
        #[cfg(test)]
        let mut output_remap = Vec::<usize>::with_capacity(items.len());
        let mut fields = Vec::<ProjectionField>::with_capacity(items.len());
        #[cfg(test)]
        let mut saw_wrapped_projection = false;

        for (index, item) in items.into_iter().enumerate() {
            let expr = lower_select_item_expr(&item, SqlExprPhase::PostAggregate)?;
            let analysis = analyze_lowered_expr(&expr, None);
            if !analysis.contains_aggregate() || analysis.references_direct_fields() {
                return Err(SqlLoweringError::unsupported_global_aggregate_projection());
            }

            let direct_terminal_index =
                collect_unique_global_aggregate_terminals_from_expr(&expr, &mut terminals)?;
            #[cfg(test)]
            match direct_terminal_index {
                Some(unique_index) => output_remap.push(unique_index),
                None => {
                    saw_wrapped_projection = true;
                }
            }
            #[cfg(not(test))]
            let _ = direct_terminal_index;

            fields.push(ProjectionField::Scalar {
                expr,
                alias: projection_aliases
                    .get(index)
                    .and_then(Option::as_deref)
                    .map(Alias::new),
            });
        }

        Ok(Self {
            terminals,
            projection: lower_global_aggregate_projection(fields),
            #[cfg(test)]
            output_remap: if saw_wrapped_projection {
                Vec::new()
            } else {
                output_remap
            },
        })
    }
}

// Global post-aggregate projection expressions may compose aggregate leaves
// with literals/functions/arithmetic, but they may not reopen direct field
// access outside aggregate inputs.
pub(in crate::db::sql::lowering) fn expr_references_global_direct_fields(expr: &Expr) -> bool {
    analyze_lowered_expr(expr, None).references_direct_fields()
}

// Collect every aggregate leaf referenced by one global post-aggregate output
// expression while deduplicating onto the canonical executable terminal list.
// Direct aggregate terminals still report the first-seen terminal remap so the
// legacy terminal-remap tests keep their existing contract.
fn collect_unique_global_aggregate_terminals_from_expr(
    expr: &Expr,
    terminals: &mut Vec<SqlGlobalAggregateTerminal>,
) -> Result<Option<usize>, SqlLoweringError> {
    let mut direct_terminal_index = None;
    expr.try_for_each_tree_aggregate(&mut |aggregate_expr| {
        let terminal = SqlGlobalAggregateTerminal::from_aggregate_expr(aggregate_expr)?;
        let unique_index = terminals
            .iter()
            .position(|current| current == &terminal)
            .unwrap_or_else(|| {
                let index = terminals.len();
                terminals.push(terminal);
                index
            });
        if direct_terminal_index.is_none() && matches!(expr, Expr::Aggregate(_)) {
            direct_terminal_index = Some(unique_index);
        }

        Ok::<(), SqlLoweringError>(())
    })?;

    Ok(direct_terminal_index)
}

fn lower_sql_aggregate_shape(
    call: SqlAggregateCall,
) -> Result<LoweredSqlAggregateShape, SqlLoweringError> {
    let SqlAggregateCall {
        kind,
        input,
        filter_expr,
        distinct,
    } = call;
    let filter_expr = filter_expr
        .map(|expr| lower_sql_pre_aggregate_bool_expr(expr.as_ref()))
        .transpose()?;

    if distinct && filter_expr.is_some() {
        return Err(SqlLoweringError::unsupported_select_projection());
    }

    match input.map(|input| *input) {
        None if kind.supports_star_input() && !distinct => {
            Ok(LoweredSqlAggregateShape::CountRows { filter_expr })
        }
        Some(SqlExpr::Field(field)) if matches!(kind, SqlAggregateKind::Count) => {
            Ok(LoweredSqlAggregateShape::CountField {
                field,
                filter_expr,
                distinct,
            })
        }
        Some(SqlExpr::Field(field)) if kind.lowers_shared_field_target_shape() => {
            Ok(LoweredSqlAggregateShape::FieldTarget {
                kind,
                field,
                filter_expr,
                distinct,
            })
        }
        Some(input) => Ok(LoweredSqlAggregateShape::ExpressionInput {
            kind,
            input_expr: canonicalize_aggregate_input_expr(
                kind.aggregate_kind(),
                lower_sql_expr(&input, SqlExprPhase::PreAggregate)?,
            ),
            filter_expr,
            distinct,
        }),
        _ => Err(SqlLoweringError::unsupported_select_projection()),
    }
}

pub(in crate::db::sql::lowering) fn grouped_projection_aggregate_calls(
    projection: &SqlProjection,
    group_by_fields: &[String],
    model: &'static EntityModel,
) -> Result<Vec<SqlAggregateCall>, SqlLoweringError> {
    if group_by_fields.is_empty() {
        return Err(SqlLoweringError::unsupported_select_group_by());
    }

    let SqlProjection::Items(items) = projection else {
        return Err(SqlLoweringError::grouped_projection_requires_explicit_list());
    };

    GroupedProjectionAggregateCollector::new(group_by_fields, model)?.collect_from_items(items)
}

// Extend one unique aggregate-call list from one SQL expression while keeping
// first-seen SQL order stable for grouped reducer slot assignment.
pub(in crate::db::sql::lowering) fn extend_unique_sql_expr_aggregate_calls(
    aggregate_calls: &mut Vec<SqlAggregateCall>,
    expr: &SqlExpr,
) {
    expr.for_each_tree_aggregate(&mut |aggregate| {
        push_unique_sql_aggregate_call(aggregate_calls, aggregate.clone());
    });
}

// Extend one unique aggregate-call list from one SQL select item while keeping
// SQL item-order ownership local to shared aggregate collection helpers.
pub(in crate::db::sql::lowering) fn extend_unique_sql_select_item_aggregate_calls(
    aggregate_calls: &mut Vec<SqlAggregateCall>,
    item: &SqlSelectItem,
) {
    match item {
        SqlSelectItem::Field(_) => {}
        SqlSelectItem::Aggregate(aggregate) => {
            push_unique_sql_aggregate_call(aggregate_calls, aggregate.clone());
        }
        SqlSelectItem::Expr(expr) => {
            extend_unique_sql_expr_aggregate_calls(aggregate_calls, expr);
        }
    }
}

///
/// GroupedProjectionAggregateCollector
///
/// Local grouped-projection aggregate extraction owner. It validates grouped
/// field authority, preserves the first aggregate ordering rule, and keeps one
/// stable unique aggregate list so grouped reducer slots are derived once.
///

struct GroupedProjectionAggregateCollector<'a> {
    grouped_field_names: Vec<&'a str>,
    model: &'static EntityModel,
    aggregate_calls: Vec<SqlAggregateCall>,
    seen_aggregate: bool,
}

impl<'a> GroupedProjectionAggregateCollector<'a> {
    // Build the grouped projection collector once so field-authority and
    // aggregate-ordering policy stay on one local owner.
    fn new(
        group_by_fields: &'a [String],
        model: &'static EntityModel,
    ) -> Result<Self, SqlLoweringError> {
        if group_by_fields.is_empty() {
            return Err(SqlLoweringError::unsupported_select_group_by());
        }

        Ok(Self {
            grouped_field_names: group_by_fields.iter().map(String::as_str).collect(),
            model,
            aggregate_calls: Vec::new(),
            seen_aggregate: false,
        })
    }

    // Walk grouped projection items in SQL order so first-seen aggregate leaves
    // map onto one stable grouped reducer slot ordering.
    fn collect_from_items(
        mut self,
        items: &[SqlSelectItem],
    ) -> Result<Vec<SqlAggregateCall>, SqlLoweringError> {
        for (index, item) in items.iter().enumerate() {
            self.collect_item(index, item)?;
        }

        if self.aggregate_calls.is_empty() {
            return Err(SqlLoweringError::grouped_projection_requires_aggregate());
        }

        Ok(self.aggregate_calls)
    }

    // Validate one grouped projection item before collecting any aggregate
    // leaves so field-resolution and grouped-key diagnostics stay precise.
    fn collect_item(&mut self, index: usize, item: &SqlSelectItem) -> Result<(), SqlLoweringError> {
        let expr = crate::db::sql::lowering::select::lower_select_item_expr(
            item,
            SqlExprPhase::PostAggregate,
        )?;
        let analysis = analyze_lowered_expr(&expr, Some(self.model));
        let contains_aggregate = analysis.contains_aggregate();
        if self.seen_aggregate && !contains_aggregate {
            return Err(SqlLoweringError::grouped_projection_scalar_after_aggregate(
                index,
            ));
        }
        if let Some(field) = analysis.first_unknown_field() {
            return Err(SqlLoweringError::unknown_field(field));
        }
        if !expr.references_only_fields(self.grouped_field_names.as_slice()) {
            return Err(SqlLoweringError::grouped_projection_references_non_group_field(index));
        }
        if contains_aggregate {
            self.seen_aggregate = true;
            extend_unique_sql_select_item_aggregate_calls(&mut self.aggregate_calls, item);
        }

        Ok(())
    }
}

// Keep aggregate extraction on one stable first-seen unique terminal order so
// repeated SQL aggregate leaves reuse the same reducer slot.
fn push_unique_sql_aggregate_call(
    aggregate_calls: &mut Vec<SqlAggregateCall>,
    aggregate: SqlAggregateCall,
) {
    if aggregate_calls.iter().all(|current| current != &aggregate) {
        aggregate_calls.push(aggregate);
    }
}

pub(in crate::db::sql::lowering) fn lower_aggregate_call(
    call: SqlAggregateCall,
) -> Result<crate::db::query::builder::AggregateExpr, SqlLoweringError> {
    match lower_sql_aggregate_shape(call)? {
        LoweredSqlAggregateShape::CountRows { filter_expr } => {
            Ok(apply_aggregate_filter_expr(count(), filter_expr))
        }
        LoweredSqlAggregateShape::CountField {
            field,
            filter_expr,
            distinct,
        } => {
            let aggregate = if distinct {
                count_by(field).distinct()
            } else {
                count_by(field)
            };

            Ok(apply_aggregate_filter_expr(aggregate, filter_expr))
        }
        LoweredSqlAggregateShape::FieldTarget {
            kind,
            field,
            filter_expr,
            distinct,
        } => kind.lower_field_target_aggregate(field, filter_expr, distinct),
        LoweredSqlAggregateShape::ExpressionInput {
            kind,
            input_expr,
            filter_expr,
            distinct,
        } => Ok(apply_aggregate_filter_expr(
            kind.lower_expression_owned_aggregate(input_expr, distinct),
            filter_expr,
        )),
    }
}

// Lower one grouped aggregate call while validating its model-bound scalar
// subexpressions before grouped execution can compile them into reducer state.
pub(in crate::db::sql::lowering) fn lower_grouped_aggregate_call(
    model: &'static EntityModel,
    call: SqlAggregateCall,
) -> Result<crate::db::query::builder::AggregateExpr, SqlLoweringError> {
    let aggregate = lower_aggregate_call(call)?;

    validate_grouped_aggregate_scalar_subexpressions(model, &aggregate)?;

    Ok(aggregate)
}

// Attach one optional normalized planner-owned filter expression to an
// aggregate expression so parser/lowering support can stay on the aggregate
// identity boundary without reopening aggregate construction at callsites.
fn apply_aggregate_filter_expr(
    aggregate: AggregateExpr,
    filter_expr: Option<Expr>,
) -> AggregateExpr {
    match filter_expr {
        Some(filter_expr) => aggregate.with_filter_expr(filter_expr),
        None => aggregate,
    }
}

// Keep grouped aggregate scalar-subexpression validation on one lowering seam
// so alias leakage inside FILTER or aggregate inputs fails as a user-facing
// SQL error before grouped execution reaches its scalar compiler invariant.
fn validate_grouped_aggregate_scalar_subexpressions(
    model: &'static EntityModel,
    aggregate: &AggregateExpr,
) -> Result<(), SqlLoweringError> {
    if let Some(input_expr) = aggregate.input_expr() {
        validate_grouped_model_bound_scalar_expr(
            model,
            input_expr,
            SqlLoweringError::unsupported_aggregate_input_expressions,
        )?;
    }
    if let Some(filter_expr) = aggregate.filter_expr() {
        validate_grouped_model_bound_scalar_expr(
            model,
            filter_expr,
            SqlLoweringError::unsupported_where_expression,
        )?;
    }

    Ok(())
}

// Validate one grouped model-bound scalar expression while preserving the
// first unknown-field diagnostic before generic expression-family fallback.
fn validate_grouped_model_bound_scalar_expr(
    model: &'static EntityModel,
    expr: &Expr,
    unsupported: impl FnOnce() -> SqlLoweringError,
) -> Result<(), SqlLoweringError> {
    if let Some(field) = analyze_lowered_expr(expr, Some(model)).first_unknown_field() {
        return Err(SqlLoweringError::unknown_field(field));
    }
    if compile_scalar_projection_expr(model, expr).is_none() {
        return Err(unsupported());
    }

    Ok(())
}

impl SqlAggregateKind {
    // Lower one field-target aggregate call through the parser-owned aggregate
    // taxonomy so SQL lowering keeps the supported field-target family on the
    // enum instead of repeating the kind ladder at callsites.
    fn lower_field_target_aggregate(
        self,
        field: String,
        filter_expr: Option<Expr>,
        distinct: bool,
    ) -> Result<AggregateExpr, SqlLoweringError> {
        let aggregate = match self {
            Self::Count => return Err(SqlLoweringError::unsupported_select_projection()),
            Self::Sum => {
                if distinct {
                    sum(field).distinct()
                } else {
                    sum(field)
                }
            }
            Self::Avg => {
                if distinct {
                    avg(field).distinct()
                } else {
                    avg(field)
                }
            }
            Self::Min => min_by(field),
            Self::Max => max_by(field),
        };

        Ok(apply_aggregate_filter_expr(aggregate, filter_expr))
    }

    // Lower one expression-owned aggregate call through the parser-owned
    // aggregate taxonomy so SQL lowering reuses the enum's planner-kind
    // mapping instead of reopening it as a free function.
    fn lower_expression_owned_aggregate(self, input_expr: Expr, distinct: bool) -> AggregateExpr {
        let aggregate = AggregateExpr::from_expression_input(self.aggregate_kind(), input_expr);

        if distinct {
            aggregate.distinct()
        } else {
            aggregate
        }
    }
}