fraiseql-core 2.19.1

//! Aggregate and window query execution runner.

use std::sync::Arc;

use super::super::context::ExecutorContext;
use crate::{
    backend::WhereClause,
    error::{FraiseQLError, Result},
    graphql::FieldSelection,
    runtime::suggestions_for,
    schema::FactInjectOrigin,
    security::{RlsWhereClause, SecurityContext, rls_policy::RlsTarget},
};

/// Give every `ORDER BY` key that names a **text** group-by output `collation` (#1512): a
/// JSONB or calendar dimension (both `->>` extractions), or a native dimension whose cast
/// is text. Temporal buckets, tree levels and aggregates sort by their own type.
fn collate_text_group_keys(
    request: &mut crate::compiler::aggregation::AggregationRequest,
    collation: &str,
) {
    use crate::compiler::aggregation::GroupBySelection;

    let text_aliases: Vec<&str> = request
        .group_by
        .iter()
        .filter_map(|selection| match selection {
            GroupBySelection::Dimension { alias, .. }
            | GroupBySelection::CalendarDimension { alias, .. } => Some(alias.as_str()),
            GroupBySelection::NativeDimension { alias, pg_cast, .. }
                if super::query_projection::sorts_as_text(pg_cast) =>
            {
                Some(alias.as_str())
            },
            _ => None,
        })
        .collect();
    for clause in &mut request.order_by {
        if text_aliases.contains(&clause.field.as_str()) {
            clause.collation = Some(collation.to_string());
        }
    }
}

/// The stored keys of the localized fields of `metadata`'s type, as dimension keys (#1524).
fn localized_dimension_keys(
    schema: &crate::schema::CompiledSchema,
    metadata: &crate::compiler::fact_table::FactTableMetadata,
) -> Vec<String> {
    use crate::compiler::fact_table::dimension_key;
    metadata
        .type_name
        .as_deref()
        .and_then(|name| schema.find_type(name))
        .map_or_else(Vec::new, |t| {
            t.fields
                .iter()
                .filter(|f| f.localized)
                .map(|f| dimension_key(f.name.as_str()))
                .collect()
        })
}

/// Whether a dimension `path` reads one of `keys` (a localized field), whole.
fn reads_localized(path: &[String], keys: &[String]) -> bool {
    matches!(path, [key] if keys.contains(&crate::compiler::fact_table::dimension_key(key)))
}

/// Group a localized dimension by its label through `chain` (#1524): the stored locale map
/// would make every distinct map a group, and report the map as the key.
fn localize_group_by(
    plan: &mut crate::compiler::aggregation::AggregationPlan,
    keys: &[String],
    chain: &[String],
) {
    use crate::compiler::aggregation::GroupByExpression;
    for expr in &mut plan.group_by_expressions {
        if let GroupByExpression::JsonbPath {
            path, localized, ..
        } = expr
        {
            if reads_localized(path, keys) {
                *localized = Some(chain.to_vec());
            }
        }
    }
}

/// Read a filter on a localized dimension through `chain`, under `collation` (#1524): the
/// label is compared, as on any read of the field, never the stored map.
fn localize_where(
    clause: WhereClause,
    keys: &[String],
    chain: &[String],
    collation: Option<&str>,
) -> WhereClause {
    match clause {
        WhereClause::Field { ref path, .. } if reads_localized(path, keys) => {
            WhereClause::Localized {
                chain:     chain.to_vec(),
                collation: collation.map(str::to_string),
                inner:     Box::new(clause),
            }
        },
        WhereClause::And(clauses) => WhereClause::And(
            clauses.into_iter().map(|c| localize_where(c, keys, chain, collation)).collect(),
        ),
        WhereClause::Or(clauses) => WhereClause::Or(
            clauses.into_iter().map(|c| localize_where(c, keys, chain, collation)).collect(),
        ),
        WhereClause::Not(inner) => {
            WhereClause::Not(Box::new(localize_where(*inner, keys, chain, collation)))
        },
        other => other,
    }
}

/// Runner for aggregate and window analytics queries.
/// The request an aggregate or window root field makes (#1609): its arguments written inline,
/// resolved against the request's variables (nested `$var`s substituted, an object literal's
/// written key order kept), with a variable named like an argument winning over the argument
/// written inline, as for every root field (`QueryMatcher`). An inline argument the planner
/// does not read is refused, naming the ones it does, rather than ignored.
fn root_request(
    query_name: &str,
    parsed: Option<&crate::graphql::ParsedQuery>,
    variables: Option<&serde_json::Value>,
    fact_table: &str,
    accepted: &[&str],
) -> Result<serde_json::Value> {
    let mut request = match variables {
        Some(serde_json::Value::Object(map)) => map.clone(),
        _ => serde_json::Map::new(),
    };
    let Some(root) = parsed.and_then(|p| p.selections.first()) else {
        return Ok(serde_json::Value::Object(request));
    };
    let variable_map = crate::runtime::matcher::QueryMatcher::extract_arguments(variables);
    for argument in &root.arguments {
        if argument.name == "table" {
            return Err(crate::runtime::aggregate_parser::table_argument_refusal(fact_table));
        }
        if !accepted.contains(&argument.name.as_str()) {
            return Err(FraiseQLError::Validation {
                message: format!(
                    "`{query_name}` takes no argument `{}`; it reads {}",
                    argument.name,
                    accepted.iter().map(|a| format!("`{a}`")).collect::<Vec<_>>().join(", ")
                ),
                path:    Some(format!("{query_name}.{}", argument.name)),
            });
        }
        if request.contains_key(&argument.name) {
            continue;
        }
        if let Some(value) =
            crate::runtime::matcher::QueryMatcher::resolve_inline_arg(argument, &variable_map)?
        {
            request.insert(argument.name.clone(), value);
        }
    }
    Ok(serde_json::Value::Object(request))
}

pub(in super::super) struct AggregateRunner {
    ctx: Arc<ExecutorContext>,
}

impl AggregateRunner {
    pub(in super::super) const fn new(ctx: Arc<ExecutorContext>) -> Self {
        Self { ctx }
    }

    /// Resolve configured session variables for `security_context` into owned
    /// `(name, value)` pairs, for passing to the connection-affine
    /// `*_with_session` adapter methods so `current_setting()`-backed RLS on
    /// aggregate views is effective (#329).
    fn resolve_session_vars(
        &self,
        security_context: Option<&SecurityContext>,
    ) -> Result<Vec<(String, String)>> {
        crate::runtime::executor::support::security::read_session_variables(
            &self.ctx.schema,
            security_context,
        )
    }

    /// The collation a text key sorts under in this request's locale, when the schema
    /// declares `[locale]` (#1512).
    fn request_collation(&self) -> Option<String> {
        let schema = &self.ctx.schema;
        schema
            .locale
            .as_ref()
            .zip(crate::runtime::request_locale(schema))
            .and_then(|(config, locale)| config.collation(&locale))
    }

    /// Refuse a read with no principal when a row policy is configured (ruling AB 3).
    ///
    /// The policy cannot be evaluated without a principal, and the composition below applies
    /// it only when there is one: an anonymous aggregate or window read every row. The regular
    /// read has refused this since #784; this is the same refusal, worded the same way (the
    /// read is not advertised to a caller who cannot make it). It sits in the two executors,
    /// so the GraphQL dispatch and the public embedder entries — which pass no principal —
    /// both meet it.
    fn refuse_anonymous_under_a_row_policy(
        &self,
        query_name: &str,
        security_context: Option<&SecurityContext>,
    ) -> Result<()> {
        if security_context.is_none() && self.ctx.config.rls_policy.is_some() {
            return Err(FraiseQLError::Validation {
                message: format!("Query '{query_name}' not found in schema"),
                path:    None,
            });
        }
        Ok(())
    }

    /// Refuse a read of a tenant-scoped fact table that nothing scopes (#1799).
    ///
    /// An aggregate or a window has no `inject_params` of its own, so the only tenant clause
    /// composed below is the row policy's. In a schema that scopes the fact table's rows by
    /// tenant ([`CompiledSchema::unenforced_tenant_scope`]) and installs no policy, the read
    /// would cover every tenant: it is refused instead, whoever the caller is. In the two
    /// executors, so the GraphQL dispatch and the public embedder entries all meet it.
    ///
    /// [`CompiledSchema::unenforced_tenant_scope`]: crate::schema::CompiledSchema::unenforced_tenant_scope
    fn refuse_an_unscoped_tenant_read(
        &self,
        query_name: &str,
        metadata: &crate::compiler::fact_table::FactTableMetadata,
    ) -> Result<()> {
        if self.ctx.config.rls_policy.is_some() {
            return Ok(());
        }
        match self.ctx.schema.unenforced_tenant_scope(metadata) {
            Some(reason) => Err(FraiseQLError::Validation {
                message: format!(
                    "Query '{query_name}' reads fact table '{}', which this schema scopes by \
                     tenant ({reason}), but no row policy scopes its aggregate and window \
                     roots: refused rather than read across tenants",
                    metadata.table_name
                ),
                path:    None,
            }),
            None => Ok(()),
        }
    }

    /// The filters a fact table's `inject_params` put on its roots for `principal` (#1800), as
    /// a list root's query puts its own: one `<column> = <value>` per param, the value resolved
    /// from the caller.
    ///
    /// Read off the schema's registration of the table, which metadata handed in cannot
    /// remove. An inherited scope stands in for #1799's refusal only: under a row policy the
    /// policy scopes the roots, as before. A caller the filters cannot be resolved for is
    /// refused: one with no security context, and one whose token lacks a claim. In the two
    /// executors, so the GraphQL dispatch and the embedder entries, which pass no principal,
    /// both meet it.
    fn inject_conditions(
        &self,
        query_name: &str,
        metadata: &crate::compiler::fact_table::FactTableMetadata,
        security_context: Option<&SecurityContext>,
    ) -> Result<Vec<WhereClause>> {
        let schema = &self.ctx.schema;
        let table = schema.get_fact_table(&metadata.table_name).unwrap_or(metadata);
        let Some(scope) = schema.fact_table_inject_scope(table) else {
            return Ok(Vec::new());
        };
        if matches!(scope.origin, FactInjectOrigin::Inherited(_))
            && self.ctx.config.rls_policy.is_some()
        {
            return Ok(Vec::new());
        }
        let Some(principal) = security_context else {
            return Err(FraiseQLError::Validation {
                message: format!(
                    "Query '{query_name}' reads fact table '{}', which is scoped by {}, which a \
                     request without a security context cannot resolve",
                    table.table_name, scope.origin
                ),
                path:    None,
            });
        };
        let native_columns = crate::runtime::native_columns::filter_columns_to_native_map(
            &table.denormalized_filters,
        );
        scope
            .params
            .iter()
            .map(|(column, source)| {
                let value = super::super::resolve_inject_value(
                    column,
                    source,
                    principal,
                    schema.tenant_claim(),
                )?;
                // Each column is one of the table's denormalized filters
                // (`fact_table_inject_scope`), compared as the parser compares a
                // caller's filter on it.
                Ok(WhereClause::NativeField {
                    column: column.clone(),
                    pg_cast: native_columns.get(column).cloned().unwrap_or_default(),
                    operator: crate::backend::where_clause::WhereOperator::Eq,
                    value,
                })
            })
            .collect()
    }

    /// The `WHERE` an aggregate or a window over `metadata` runs under: the row policy's
    /// predicate for the principal, then the fact table's `inject_params` filters, then the
    /// caller's own filter, AND-ed in that order so the caller's can only narrow them.
    ///
    /// SECURITY (#795): the policy is looked up by the fact table the root field resolved,
    /// never the client's `table` key. This runs *before* the planner's reconciliation check,
    /// so it must be correct on its own: an unpolicied name returns `None`, which composed no
    /// WHERE clause at all and silently dropped the tenant filter.
    fn scoped_where(
        &self,
        query_name: &str,
        metadata: &crate::compiler::fact_table::FactTableMetadata,
        security_context: Option<&SecurityContext>,
        inject: Vec<WhereClause>,
        user: Option<WhereClause>,
    ) -> Result<Option<WhereClause>> {
        let policy = match (security_context, self.ctx.config.rls_policy.as_ref()) {
            (Some(ctx), Some(policy)) => policy
                .evaluate(ctx, &RlsTarget::fact_table(query_name, &metadata.table_name))?
                .map(RlsWhereClause::into_where_clause),
            _ => None,
        };
        let mut conditions: Vec<WhereClause> =
            policy.into_iter().chain(inject).chain(user).collect();
        Ok(match conditions.len() {
            0 => None,
            1 => conditions.pop(),
            _ => Some(WhereClause::And(conditions)),
        })
    }

    /// Execute an aggregate query dispatch.
    ///
    /// # Errors
    ///
    /// * [`FraiseQLError::Validation`] — the query name does not end with `_aggregate`, the derived
    ///   fact table is not found in the compiled schema, or the root field is written with an
    ///   argument the planner does not read.
    /// * Propagates errors from [`execute_aggregate_query`](Self::execute_aggregate_query).
    pub(in super::super) async fn execute_aggregate_dispatch(
        &self,
        query_name: &str,
        parsed: Option<&crate::graphql::ParsedQuery>,
        selection: Option<&FieldSelection>,
        variables: Option<&serde_json::Value>,
        security_context: Option<&SecurityContext>,
    ) -> Result<serde_json::Value> {
        // Extract table name from query name (e.g., "sales_aggregate" -> "tf_sales")
        let table_name =
            query_name.strip_suffix("_aggregate").ok_or_else(|| FraiseQLError::Validation {
                message: format!("Invalid aggregate query name: {}", query_name),
                path:    None,
            })?;

        let fact_table_name = format!("tf_{}", table_name);

        let metadata = self.visible_fact_table(&fact_table_name, security_context)?;

        let query_json = root_request(
            query_name,
            parsed,
            variables,
            &metadata.table_name,
            &[
                "where",
                "groupBy",
                "aggregates",
                "having",
                "orderBy",
                "limit",
                "offset",
            ],
        )?;
        let query_json = &query_json;

        // Execute aggregate query
        // The response answers under the root's response key: its alias when it has one.
        let response_key = parsed
            .and_then(|p| p.selections.first())
            .map_or(query_name, |root| root.response_key());
        let projection = match parsed.and_then(|p| p.selections.first()) {
            Some(_) => self.row_selection(query_name, response_key, selection, variables)?,
            None => None,
        };
        let mut response = self
            .execute_aggregate_query(
                query_json,
                query_name,
                response_key,
                metadata,
                security_context,
            )
            .await?;
        if let Some((row_type, root)) = projection {
            self.project_rows(&mut response, row_type, root, variables, response_key);
        }
        Ok(response)
    }

    /// The row type an aggregate root publishes and the selection made on it (#1800), once the
    /// schema's derived surface is built; `None` before, when the rows are served as computed.
    ///
    /// # Errors
    ///
    /// [`FraiseQLError::Validation`] — the root selects no field (§ 5.3.3), or selects one the
    /// row type does not publish (§ 5.3.1), or gives a field a selection set, or its fragment
    /// spreads cannot be expanded. Refused before the database is read.
    fn row_selection<'s>(
        &'s self,
        query_name: &str,
        response_key: &str,
        selection: Option<&'s FieldSelection>,
        variables: Option<&serde_json::Value>,
    ) -> Result<Option<(&'s str, &'s FieldSelection)>> {
        let Some(row_type) = self
            .ctx
            .schema
            .fact_root(query_name)
            .and_then(|root| root.row_type.as_deref())
            .filter(|row_type| self.ctx.output_types.publishes_object(row_type))
        else {
            return Ok(None);
        };
        let refusal = |message: String| FraiseQLError::Validation {
            message,
            path: Some(response_key.to_string()),
        };
        let root = selection.ok_or_else(|| {
            refusal(format!("the fragment spreads selected on `{response_key}` cannot be expanded"))
        })?;
        if root.nested_fields.is_empty() {
            return Err(refusal(format!(
                "`{response_key}` returns `[{row_type}!]!` and must select its fields"
            )));
        }
        if let Some(message) =
            self.ctx
                .output_types
                .row_selection_violation(row_type, &root.nested_fields, variables)
        {
            return Err(refusal(message));
        }
        Ok(Some((row_type, root)))
    }

    /// Replace the computed rows under `response_key` with their projection onto `root`'s
    /// selection set, adding the field errors it raises to `errors`.
    fn project_rows(
        &self,
        response: &mut serde_json::Value,
        row_type: &str,
        root: &FieldSelection,
        variables: Option<&serde_json::Value>,
        response_key: &str,
    ) {
        let Some(rows) = response
            .get_mut("data")
            .and_then(|data| data.get_mut(response_key))
            .map(serde_json::Value::take)
        else {
            return;
        };
        let rows = match rows {
            serde_json::Value::Array(rows) => rows,
            other => vec![other],
        };
        let (rows, errors) = self.ctx.output_types.project_rows(
            rows,
            row_type,
            &root.nested_fields,
            variables,
            response_key,
        );
        response["data"][response_key] = rows;
        if errors.is_empty() {
            return;
        }
        match response.get_mut("errors").and_then(serde_json::Value::as_array_mut) {
            Some(existing) => existing.extend(errors),
            None => response["errors"] = serde_json::Value::Array(errors),
        }
    }

    /// The fact table a root reads, as `security_context` may see it: one read as a type whose
    /// role the caller lacks is absent, answered exactly as a table that does not exist and
    /// never suggested, as a role-gated query is (#1122).
    ///
    /// # Errors
    ///
    /// [`FraiseQLError::Validation`] — no fact table of that name is visible to the caller.
    fn visible_fact_table(
        &self,
        fact_table_name: &str,
        security_context: Option<&SecurityContext>,
    ) -> Result<&crate::compiler::fact_table::FactTableMetadata> {
        let schema = &self.ctx.schema;
        let visible = |table: &crate::compiler::fact_table::FactTableMetadata| {
            let role = schema
                .fact_table_type(table)
                .and_then(|type_def| type_def.requires_role.as_deref());
            crate::security::role_gate::caller_holds(role, security_context)
        };
        schema
            .get_fact_table(fact_table_name)
            .filter(|table| visible(table))
            .ok_or_else(|| {
                let known: Vec<&str> = schema
                    .list_fact_tables()
                    .into_iter()
                    .filter(|name| schema.get_fact_table(name).is_some_and(visible))
                    .collect();
                let suggestion =
                    suggestions_for(self.ctx.config.suggestions, fact_table_name, &known);
                let base = format!("Fact table '{fact_table_name}' not found in schema");
                let message = match suggestion.as_slice() {
                    [s] => format!("{base}. Did you mean '{s}'?"),
                    _ => base,
                };
                FraiseQLError::Validation {
                    message,
                    path: Some(format!("fact_tables.{fact_table_name}")),
                }
            })
    }

    /// Execute a window query dispatch.
    ///
    /// # Errors
    ///
    /// * [`FraiseQLError::Validation`] — the query name does not end with `_window`, the derived
    ///   fact table is not found in the compiled schema, or the root field is written with an
    ///   argument the planner does not read.
    /// * Propagates errors from [`execute_window_query`](Self::execute_window_query).
    pub(in super::super) async fn execute_window_dispatch(
        &self,
        query_name: &str,
        parsed: Option<&crate::graphql::ParsedQuery>,
        variables: Option<&serde_json::Value>,
        security_context: Option<&SecurityContext>,
    ) -> Result<serde_json::Value> {
        // Extract table name from query name (e.g., "sales_window" -> "tf_sales")
        let table_name =
            query_name.strip_suffix("_window").ok_or_else(|| FraiseQLError::Validation {
                message: format!("Invalid window query name: {}", query_name),
                path:    None,
            })?;

        let fact_table_name = format!("tf_{}", table_name);

        let metadata = self.visible_fact_table(&fact_table_name, security_context)?;

        let query_json = root_request(
            query_name,
            parsed,
            variables,
            &metadata.table_name,
            &["select", "windows", "where", "orderBy", "limit", "offset"],
        )?;
        let query_json = &query_json;

        // Execute window query
        // The response answers under the root's response key: its alias when it has one.
        let root = parsed.and_then(|p| p.selections.first());
        let response_key = root.map_or(query_name, |root| root.response_key());
        // Its rows are `JSON`, keyed by the `select` and `windows` aliases: a leaf (§ 5.3.3).
        if root.is_some_and(|root| !root.nested_fields.is_empty()) {
            return Err(FraiseQLError::Validation {
                message: format!(
                    "`{response_key}` returns `[JSON!]!`, a leaf, and takes no selection set: \
                     its rows hold the `select` and `windows` aliases"
                ),
                path:    Some(response_key.to_string()),
            });
        }
        self.execute_window_query(query_json, query_name, response_key, metadata, security_context)
            .await
    }

    /// Execute an aggregate query.
    ///
    /// # Arguments
    ///
    /// * `query_json` - JSON representation of the aggregate query
    /// * `query_name` - GraphQL field name (e.g., "`sales_aggregate`")
    /// * `response_key` - the key the result answers under: the root's alias, or its name
    /// * `metadata` - Fact table metadata
    ///
    /// # Returns
    ///
    /// GraphQL response as JSON string
    ///
    /// When `security_context` is `Some`, evaluates the configured RLS policy and
    /// AND-composes the resulting WHERE clause with the user-supplied WHERE before
    /// planning. RLS conditions are always placed first so they cannot be bypassed.
    ///
    /// # Errors
    ///
    /// Returns error if:
    /// - RLS policy evaluation fails
    /// - Query parsing fails
    /// - Execution plan generation fails
    /// - SQL generation fails
    /// - Database execution fails
    /// - Result projection fails
    ///
    /// # Example
    ///
    /// ```no_run
    /// // Requires: a live database adapter and compiled fact table metadata.
    /// // See: tests/integration/ for runnable examples.
    /// # use serde_json::json;
    /// let query_json = json!({
    ///     "groupBy": { "category": true },
    ///     "aggregates": [{"count": true}]
    /// });
    /// // let result = executor.execute_aggregate_query(&query_json, "sales_aggregate", &metadata).await?;
    /// ```
    /// This aggregate's compiled read routing (#957), or the server's policy when
    /// no compiled query carries the name.
    ///
    /// Aggregates dispatch by *name* against fact-table metadata rather than
    /// through a `QueryDefinition`, so the annotation has to be looked back up.
    /// Falling back to `Any` is the same answer every query gave before the field
    /// existed — an aggregate nobody annotated keeps following server policy.
    fn aggregate_read_routing(&self, query_name: &str) -> crate::backend::types::ReadRouting {
        self.ctx
            .schema
            .queries
            .iter()
            .find(|q| q.name == query_name)
            .map_or(crate::backend::types::ReadRouting::Any, |q| q.read_routing)
    }

    pub(in super::super) async fn execute_aggregate_query(
        &self,
        query_json: &serde_json::Value,
        query_name: &str,
        response_key: &str,
        metadata: &crate::compiler::fact_table::FactTableMetadata,
        security_context: Option<&SecurityContext>,
    ) -> Result<serde_json::Value> {
        self.refuse_anonymous_under_a_row_policy(query_name, security_context)?;
        self.refuse_an_unscoped_tenant_read(query_name, metadata)?;
        let inject = self.inject_conditions(query_name, metadata, security_context)?;

        // 1. Parse JSON query into AggregationRequest. Build native_columns from
        //    denormalized_filters so the parser can emit direct column references instead of JSONB
        //    extraction for native columns.
        let native_columns = crate::runtime::native_columns::filter_columns_to_native_map(
            &metadata.denormalized_filters,
        );
        let mut request =
            crate::runtime::AggregateQueryParser::parse(query_json, metadata, &native_columns)?;
        // #1532: a page of groups is held to `[validation] max_page_size` as a list's is.
        request.limit = super::query_params::enforce_max_page_size(
            request.limit,
            self.ctx.config.max_page_size,
            "limit",
        )?;
        // #1306: an offset over groups is held to `[validation] max_offset` as a list's is.
        request.offset = super::query_params::enforce_max_offset(
            request.offset,
            self.ctx.config.max_offset,
            "offset",
        )?;
        // #1512: an ORDER BY on a text group-by key sorts under the request locale's collation.
        if let Some(collation) = self.request_collation() {
            collate_text_group_keys(&mut request, &collation);
        }

        // 1a. A linked fact table is read as its type (ruling AB 2): every name the request
        //     references must be a field the caller may read. Before the policy is composed,
        //     so its own predicate is never classified.
        super::aggregate_gates::refuse_unreadable_aggregate(
            &self.ctx.schema,
            metadata,
            &request,
            security_context,
        )?;

        // 1a''. A localized dimension is read as its label in the request locale (#1524): in the
        //       caller's filter here, in the grouping once planned. The policy's own clause is
        //       composed below and never names one.
        let localized_keys = localized_dimension_keys(&self.ctx.schema, metadata);
        let chain = crate::runtime::localization_chain(&self.ctx.schema).unwrap_or_default();
        if !localized_keys.is_empty() {
            if let Some(clause) = request.where_clause.take() {
                let collation = self.request_collation();
                request.where_clause =
                    Some(localize_where(clause, &localized_keys, &chain, collation.as_deref()));
            }
        }

        // 1a'. Node-id filters resolve through the hierarchy their column declares (#1498).
        //      The caller's clause only: the policy below never carries one.
        if let Some(clause) = request.where_clause.take() {
            request.where_clause = Some(super::aggregate_hierarchy::attach_hierarchies(
                clause,
                metadata,
                self.ctx.schema.hierarchies_config.as_ref(),
            )?);
        }

        // 1b. Compose the row policy and the fact table's inject filters before the caller's own
        //     filter, so it cannot bypass them.
        let user = request.where_clause.take();
        request.where_clause =
            self.scoped_where(query_name, metadata, security_context, inject, user)?;

        // 2. Generate the execution plan
        let mut plan =
            crate::compiler::aggregation::AggregationPlanner::plan(request, metadata.clone())?;
        localize_group_by(&mut plan, &localized_keys, &chain);

        // 3. Generate parameterized SQL
        let sql_generator =
            crate::runtime::AggregationSqlGenerator::new(self.ctx.adapter.database_type());
        let parameterized = sql_generator.generate_parameterized(&plan)?;

        // 4. Execute with bind parameters (eliminates escape-based injection risk), pinning session
        //    variables to the connection for current_setting() RLS (#329).
        let resolved_session_vars = self.resolve_session_vars(security_context)?;
        let session_pairs: Vec<(&str, &str)> =
            resolved_session_vars.iter().map(|(k, v)| (k.as_str(), v.as_str())).collect();
        let routing = self.aggregate_read_routing(query_name);

        // 4a. #1459: a carried-forward measure reads one row per (bucket, entity) cell. Count
        //     the cells first, reading no measure, and refuse past the bound rather than run.
        let bound = self.ctx.config.max_semi_additive_cells;
        if let Some(count) = sql_generator.generate_cell_count(&plan, bound)? {
            let counted = self
                .ctx
                .adapter
                .execute_parameterized_aggregate_with_session(
                    &count.sql,
                    &count.params,
                    &session_pairs,
                    routing,
                )
                .await?;
            let read = |key: &str| {
                counted
                    .first()
                    .and_then(|row| row.get(key))
                    .and_then(serde_json::Value::as_u64)
                    .unwrap_or(0)
            };
            let (buckets, entities) = (read("buckets"), read("entities"));
            // At least one entity: a range no entity falls in still generates its buckets,
            // and a product of zero would let any number of them through.
            if buckets.saturating_mul(entities.max(1)) > bound {
                let measures = plan
                    .time_reduction
                    .as_ref()
                    .map(|r| r.measures.join("`, `"))
                    .unwrap_or_default();
                return Err(FraiseQLError::Validation {
                    message: format!(
                        "`{measures}` is carried forward per entity and bucket: this request \
                         spans more than {bound} cells ({buckets} buckets by {entities} \
                         entities, each counted to {}). Narrow the range or the entities, or \
                         coarsen the time grouping ([validation] max_semi_additive_cells)",
                        bound.saturating_add(1)
                    ),
                    path:    None,
                });
            }
        }

        let rows = self
            .ctx
            .adapter
            .execute_parameterized_aggregate_exact(
                &parameterized.sql,
                &parameterized.params,
                &session_pairs,
                routing,
                &plan.decimal_aliases(),
            )
            .await?;

        // 5. Project results
        let projected = crate::runtime::AggregationProjector::project(rows, &plan)?;

        // 6. Wrap in GraphQL data envelope
        let response =
            crate::runtime::AggregationProjector::wrap_in_data_envelope(projected, response_key);

        // 7. Serialize to JSON string
        Ok(response)
    }

    /// Execute a window query.
    ///
    /// # Arguments
    ///
    /// * `query_json` - JSON representation of the window query
    /// * `query_name` - GraphQL field name (e.g., "`sales_window`")
    /// * `response_key` - the key the result answers under: the root's alias, or its name
    /// * `metadata` - Fact table metadata
    ///
    /// # Returns
    ///
    /// GraphQL response as JSON string
    ///
    /// When `security_context` is `Some`, evaluates the configured RLS policy and
    /// AND-composes the resulting WHERE clause with the user-supplied WHERE before
    /// planning. RLS conditions are always placed first so they cannot be bypassed.
    ///
    /// # Errors
    ///
    /// Returns error if:
    /// - RLS policy evaluation fails
    /// - Query parsing fails
    /// - Execution plan generation fails
    /// - SQL generation fails
    /// - Database execution fails
    /// - Result projection fails
    ///
    /// # Example
    ///
    /// ```no_run
    /// // Requires: a live database adapter and compiled fact table metadata.
    /// // See: tests/integration/ for runnable examples.
    /// # use serde_json::json;
    /// let query_json = json!({
    ///     "select": [{"type": "measure", "name": "revenue", "alias": "revenue"}],
    ///     "windows": [{
    ///         "function": {"type": "row_number"},
    ///         "alias": "rank",
    ///         "partitionBy": [{"type": "dimension", "path": "category"}],
    ///         "orderBy": [{"field": "revenue", "direction": "DESC"}]
    ///     }]
    /// });
    /// // let result = executor.execute_window_query(&query_json, "sales_window", &metadata).await?;
    /// ```
    pub(in super::super) async fn execute_window_query(
        &self,
        query_json: &serde_json::Value,
        query_name: &str,
        response_key: &str,
        metadata: &crate::compiler::fact_table::FactTableMetadata,
        security_context: Option<&SecurityContext>,
    ) -> Result<serde_json::Value> {
        self.refuse_anonymous_under_a_row_policy(query_name, security_context)?;
        self.refuse_an_unscoped_tenant_read(query_name, metadata)?;
        let inject = self.inject_conditions(query_name, metadata, security_context)?;

        // 1. Parse JSON query into WindowRequest
        let mut request = crate::runtime::WindowQueryParser::parse(query_json, metadata)?;
        // #1459: a window aggregate runs over fact rows; over a measure not additive over time
        // it would combine one entity's value across them.
        crate::compiler::time_reduction::refuse_window_over_non_additive(&request, metadata)?;
        // #1532: as an aggregate's.
        request.limit = super::query_params::enforce_max_page_size(
            request.limit,
            self.ctx.config.max_page_size,
            "limit",
        )?;
        // #1306: as an aggregate's.
        request.offset = super::query_params::enforce_max_offset(
            request.offset,
            self.ctx.config.max_offset,
            "offset",
        )?;

        // 1a. A linked fact table is read as its type (ruling AB 2), as for an aggregate.
        super::aggregate_gates::refuse_unreadable_window(
            &self.ctx.schema,
            metadata,
            &request,
            security_context,
        )?;

        // 1a'. A localized dimension is read as its label (#1524): in the caller's filter here,
        //      and wherever the plan selects, partitions or orders by it.
        let localized_keys = localized_dimension_keys(&self.ctx.schema, metadata);
        let chain = crate::runtime::localization_chain(&self.ctx.schema).unwrap_or_default();
        let collation = self.request_collation();
        if !localized_keys.is_empty() {
            if let Some(clause) = request.where_clause.take() {
                request.where_clause =
                    Some(localize_where(clause, &localized_keys, &chain, collation.as_deref()));
            }
        }

        // 1b. Compose the row policy and the fact table's inject filters before the caller's own
        //     filter, so it cannot bypass them.
        let user = request.where_clause.take();
        request.where_clause =
            self.scoped_where(query_name, metadata, security_context, inject, user)?;

        // 2. Generate execution plan (validates semantic names against metadata) Text keys sort
        //    under the request locale's collation (#1512).
        let decimals = request.decimal_aliases(metadata);
        let plan = crate::compiler::window_functions::WindowPlanner::plan_in_locale(
            request,
            metadata,
            crate::compiler::window_functions::WindowLocale {
                collation: collation.as_deref(),
                localized: &localized_keys,
                chain:     &chain,
            },
        )?;

        // 3. Generate SQL
        let sql_generator =
            crate::runtime::WindowSqlGenerator::new(self.ctx.adapter.database_type());
        let sql = sql_generator.generate(&plan)?;

        // 4. Execute SQL — bind parameters via execute_parameterized_aggregate so WHERE clause
        //    values are passed as prepared-statement parameters, not inlined. Session variables are
        //    pinned to the connection for current_setting() RLS (#329).
        let resolved_session_vars = self.resolve_session_vars(security_context)?;
        let session_pairs: Vec<(&str, &str)> =
            resolved_session_vars.iter().map(|(k, v)| (k.as_str(), v.as_str())).collect();
        let routing = self.aggregate_read_routing(query_name);
        let rows = self
            .ctx
            .adapter
            .execute_parameterized_aggregate_exact(
                &sql.raw_sql,
                &sql.parameters,
                &session_pairs,
                routing,
                &decimals.iter().map(String::as_str).collect::<Vec<_>>(),
            )
            .await?;

        // 5. Project results
        let projected = crate::runtime::WindowProjector::project(rows, &plan)?;

        // 6. Wrap in GraphQL data envelope
        let response =
            crate::runtime::WindowProjector::wrap_in_data_envelope(projected, response_key);

        // 7. Serialize to JSON string
        Ok(response)
    }
}

#[cfg(test)]
#[path = "aggregate_tests.rs"]
mod aggregate_rls_tests;

#[cfg(test)]
#[path = "aggregate_inject_tests.rs"]
mod aggregate_inject_tests;