icydb_core/db/query/intent/

mod.rs

#![expect(clippy::used_underscore_binding)]

//! Module: query::intent
//! Responsibility: query intent construction, coercion, and semantic-plan compilation.
//! Does not own: executor runtime behavior or index storage details.
//! Boundary: typed/fluent query inputs lowered into validated logical plans.

#[cfg(test)]
mod tests;

// Key-only access intent and helpers.
pub type DeleteSpec = crate::db::query::plan::DeleteSpec;
pub type LoadSpec = crate::db::query::plan::LoadSpec;
pub type QueryMode = crate::db::query::plan::QueryMode;

use crate::{
    db::{
        access::{AccessPath, AccessPlan, AccessPlanError, normalize_access_plan_value},
        cursor::CursorPlanError,
        predicate::{
            CompareOp, MissingRowPolicy, Predicate, SchemaInfo, ValidateError, normalize,
            normalize_enum_literals, reject_unsupported_query_features,
        },
        query::{
            builder::aggregate::AggregateExpr,
            explain::ExplainPlan,
            expr::{FilterExpr, SortExpr, SortLowerError},
            plan::{
                AccessPlannedQuery, CursorPagingPolicyError, DeleteLimitSpec,
                FluentLoadPolicyViolation, GroupAggregateSpec, GroupHavingClause, GroupHavingSpec,
                GroupHavingSymbol, GroupSpec, GroupedExecutionConfig,
                IntentKeyAccessKind as IntentValidationKeyAccessKind,
                IntentKeyAccessPolicyViolation, LogicalPlan, OrderDirection, OrderSpec, PageSpec,
                PlanError, PlannerError, PolicyPlanError, ScalarPlan, has_explicit_order,
                plan_access, resolve_group_field_slot, validate_group_query_semantics,
                validate_intent_key_access_policy, validate_intent_plan_shape,
                validate_order_shape, validate_query_semantics,
            },
        },
        response::ResponseError,
    },
    error::{ErrorClass, InternalError},
    model::entity::EntityModel,
    traits::{EntityKind, FieldValue, SingletonEntity},
    value::Value,
};
use std::marker::PhantomData;
use thiserror::Error as ThisError;

///
/// KeyAccess
/// Primary-key-only access hints for query planning.
///

#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) enum KeyAccess<K> {
    Single(K),
    Many(Vec<K>),
}

///
/// KeyAccessKind
/// Identifies which key-only builder set the access path.
///

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum KeyAccessKind {
    Single,
    Many,
    Only,
}

///
/// KeyAccessState
/// Tracks key-only access plus its origin for intent validation.
///

#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) struct KeyAccessState<K> {
    pub kind: KeyAccessKind,
    pub access: KeyAccess<K>,
}

// Build a model-level access plan for key-only intents.
pub(crate) fn access_plan_from_keys_value<K>(access: &KeyAccess<K>) -> AccessPlan<Value>
where
    K: FieldValue,
{
    // Phase 1: map typed keys into model-level Value access paths.
    let plan = match access {
        KeyAccess::Single(key) => AccessPlan::path(AccessPath::ByKey(key.to_value())),
        KeyAccess::Many(keys) => {
            let values = keys.iter().map(FieldValue::to_value).collect();
            AccessPlan::path(AccessPath::ByKeys(values))
        }
    };

    // Phase 2: canonicalize the access shape via the shared access boundary.
    normalize_access_plan_value(plan)
}

// Convert model-level access plans into entity-keyed access plans.
pub(crate) fn access_plan_to_entity_keys<E: EntityKind>(
    model: &EntityModel,
    access: AccessPlan<Value>,
) -> Result<AccessPlan<E::Key>, PlanError> {
    access.into_executable::<E>(model)
}

// Convert model-level key values into typed entity keys.
pub(crate) fn coerce_entity_key<E: EntityKind>(
    model: &EntityModel,
    key: &Value,
) -> Result<E::Key, PlanError> {
    E::Key::from_value(key).ok_or_else(|| {
        PlanError::from(AccessPlanError::PrimaryKeyMismatch {
            field: model.primary_key.name.to_string(),
            key: key.clone(),
        })
    })
}

impl AccessPlan<Value> {
    /// Convert model-level access plans into typed executable access plans.
    pub(crate) fn into_executable<E: EntityKind>(
        self,
        model: &EntityModel,
    ) -> Result<AccessPlan<E::Key>, PlanError> {
        match self {
            Self::Path(path) => Ok(AccessPlan::path(path.into_executable::<E>(model)?)),
            Self::Union(children) => {
                let mut out = Vec::with_capacity(children.len());
                for child in children {
                    out.push(child.into_executable::<E>(model)?);
                }

                Ok(AccessPlan::union(out))
            }
            Self::Intersection(children) => {
                let mut out = Vec::with_capacity(children.len());
                for child in children {
                    out.push(child.into_executable::<E>(model)?);
                }

                Ok(AccessPlan::intersection(out))
            }
        }
    }
}

impl AccessPath<Value> {
    /// Convert one model-level access path into a typed executable access path.
    pub(crate) fn into_executable<E: EntityKind>(
        self,
        model: &EntityModel,
    ) -> Result<AccessPath<E::Key>, PlanError> {
        match self {
            Self::ByKey(key) => Ok(AccessPath::ByKey(coerce_entity_key::<E>(model, &key)?)),
            Self::ByKeys(keys) => {
                let mut out = Vec::with_capacity(keys.len());
                for key in keys {
                    out.push(coerce_entity_key::<E>(model, &key)?);
                }

                Ok(AccessPath::ByKeys(out))
            }
            Self::KeyRange { start, end } => Ok(AccessPath::KeyRange {
                start: coerce_entity_key::<E>(model, &start)?,
                end: coerce_entity_key::<E>(model, &end)?,
            }),
            Self::IndexPrefix { index, values } => Ok(AccessPath::IndexPrefix { index, values }),
            Self::IndexRange { spec } => Ok(AccessPath::IndexRange { spec }),
            Self::FullScan => Ok(AccessPath::FullScan),
        }
    }
}

///
/// QueryModel
///
/// Model-level query intent and planning context.
/// Consumes an `EntityModel` derived from typed entity definitions.
///

#[derive(Debug)]
pub(crate) struct QueryModel<'m, K> {
    model: &'m EntityModel,
    mode: QueryMode,
    predicate: Option<Predicate>,
    key_access: Option<KeyAccessState<K>>,
    key_access_conflict: bool,
    group: Option<crate::db::query::plan::GroupSpec>,
    having: Option<GroupHavingSpec>,
    order: Option<OrderSpec>,
    distinct: bool,
    consistency: MissingRowPolicy,
}

impl<'m, K: FieldValue> QueryModel<'m, K> {
    #[must_use]
    pub(crate) const fn new(model: &'m EntityModel, consistency: MissingRowPolicy) -> Self {
        Self {
            model,
            mode: QueryMode::Load(LoadSpec::new()),
            predicate: None,
            key_access: None,
            key_access_conflict: false,
            group: None,
            having: None,
            order: None,
            distinct: false,
            consistency,
        }
    }

    /// Return the intent mode (load vs delete).
    #[must_use]
    pub(crate) const fn mode(&self) -> QueryMode {
        self.mode
    }

    #[must_use]
    fn has_explicit_order(&self) -> bool {
        has_explicit_order(self.order.as_ref())
    }

    #[must_use]
    const fn has_grouping(&self) -> bool {
        self.group.is_some()
    }

    #[must_use]
    const fn load_spec(&self) -> Option<LoadSpec> {
        match self.mode {
            QueryMode::Load(spec) => Some(spec),
            QueryMode::Delete(_) => None,
        }
    }

    /// Add a predicate, implicitly AND-ing with any existing predicate.
    #[must_use]
    pub(crate) fn filter(mut self, predicate: Predicate) -> Self {
        self.predicate = match self.predicate.take() {
            Some(existing) => Some(Predicate::And(vec![existing, predicate])),
            None => Some(predicate),
        };
        self
    }

    /// Apply a dynamic filter expression using the model schema.
    pub(crate) fn filter_expr(self, expr: FilterExpr) -> Result<Self, QueryError> {
        let schema = SchemaInfo::from_entity_model(self.model)?;
        let predicate = expr.lower_with(&schema).map_err(QueryError::Validate)?;

        Ok(self.filter(predicate))
    }

    /// Apply a dynamic sort expression using the model schema.
    pub(crate) fn sort_expr(self, expr: SortExpr) -> Result<Self, QueryError> {
        let schema = SchemaInfo::from_entity_model(self.model)?;
        let order = match expr.lower_with(&schema) {
            Ok(order) => order,
            Err(SortLowerError::Validate(err)) => return Err(QueryError::Validate(err)),
            Err(SortLowerError::Plan(err)) => return Err(QueryError::from(*err)),
        };

        validate_order_shape(Some(&order))
            .map_err(IntentError::from)
            .map_err(QueryError::from)?;

        Ok(self.order_spec(order))
    }

    /// Append an ascending sort key.
    #[must_use]
    pub(crate) fn order_by(mut self, field: impl AsRef<str>) -> Self {
        self.order = Some(push_order(self.order, field.as_ref(), OrderDirection::Asc));
        self
    }

    /// Append a descending sort key.
    #[must_use]
    pub(crate) fn order_by_desc(mut self, field: impl AsRef<str>) -> Self {
        self.order = Some(push_order(self.order, field.as_ref(), OrderDirection::Desc));
        self
    }

    /// Set a fully-specified order spec (validated before reaching this boundary).
    pub(crate) fn order_spec(mut self, order: OrderSpec) -> Self {
        self.order = Some(order);
        self
    }

    /// Enable DISTINCT semantics for this query intent.
    #[must_use]
    pub(crate) const fn distinct(mut self) -> Self {
        self.distinct = true;
        self
    }

    // Resolve one grouped field into one stable field slot and append it to the
    // grouped spec in declaration order.
    fn push_group_field(mut self, field: &str) -> Result<Self, QueryError> {
        let field_slot = resolve_group_field_slot(self.model, field).map_err(QueryError::from)?;
        let group = self.group.get_or_insert(GroupSpec {
            group_fields: Vec::new(),
            aggregates: Vec::new(),
            execution: GroupedExecutionConfig::unbounded(),
        });
        if !group
            .group_fields
            .iter()
            .any(|existing| existing.index() == field_slot.index())
        {
            group.group_fields.push(field_slot);
        }

        Ok(self)
    }

    // Append one grouped aggregate terminal to the grouped declarative spec.
    fn push_group_aggregate(mut self, aggregate: AggregateExpr) -> Self {
        let group = self.group.get_or_insert(GroupSpec {
            group_fields: Vec::new(),
            aggregates: Vec::new(),
            execution: GroupedExecutionConfig::unbounded(),
        });
        group
            .aggregates
            .push(GroupAggregateSpec::from_aggregate_expr(&aggregate));

        self
    }

    // Override grouped hard limits for this grouped query.
    fn grouped_limits(mut self, max_groups: u64, max_group_bytes: u64) -> Self {
        let group = self.group.get_or_insert(GroupSpec {
            group_fields: Vec::new(),
            aggregates: Vec::new(),
            execution: GroupedExecutionConfig::unbounded(),
        });
        group.execution = GroupedExecutionConfig::with_hard_limits(max_groups, max_group_bytes);

        self
    }

    // Append one grouped HAVING compare clause after GROUP BY terminal declaration.
    fn push_having_clause(mut self, clause: GroupHavingClause) -> Result<Self, QueryError> {
        if self.group.is_none() {
            return Err(QueryError::Intent(IntentError::HavingRequiresGroupBy));
        }

        let having = self.having.get_or_insert(GroupHavingSpec {
            clauses: Vec::new(),
        });
        having.clauses.push(clause);

        Ok(self)
    }

    // Append one grouped HAVING clause that references one grouped key field.
    fn push_having_group_clause(
        self,
        field: &str,
        op: CompareOp,
        value: Value,
    ) -> Result<Self, QueryError> {
        let field_slot = resolve_group_field_slot(self.model, field).map_err(QueryError::from)?;

        self.push_having_clause(GroupHavingClause {
            symbol: GroupHavingSymbol::GroupField(field_slot),
            op,
            value,
        })
    }

    // Append one grouped HAVING clause that references one grouped aggregate output.
    fn push_having_aggregate_clause(
        self,
        aggregate_index: usize,
        op: CompareOp,
        value: Value,
    ) -> Result<Self, QueryError> {
        self.push_having_clause(GroupHavingClause {
            symbol: GroupHavingSymbol::AggregateIndex(aggregate_index),
            op,
            value,
        })
    }

    /// Track key-only access paths and detect conflicting key intents.
    fn set_key_access(mut self, kind: KeyAccessKind, access: KeyAccess<K>) -> Self {
        if let Some(existing) = &self.key_access
            && existing.kind != kind
        {
            self.key_access_conflict = true;
        }

        self.key_access = Some(KeyAccessState { kind, access });

        self
    }

    /// Set the access path to a single primary key lookup.
    pub(crate) fn by_id(self, id: K) -> Self {
        self.set_key_access(KeyAccessKind::Single, KeyAccess::Single(id))
    }

    /// Set the access path to a primary key batch lookup.
    pub(crate) fn by_ids<I>(self, ids: I) -> Self
    where
        I: IntoIterator<Item = K>,
    {
        self.set_key_access(
            KeyAccessKind::Many,
            KeyAccess::Many(ids.into_iter().collect()),
        )
    }

    /// Set the access path to the singleton primary key.
    pub(crate) fn only(self, id: K) -> Self {
        self.set_key_access(KeyAccessKind::Only, KeyAccess::Single(id))
    }

    /// Mark this intent as a delete query.
    #[must_use]
    pub(crate) const fn delete(mut self) -> Self {
        if self.mode.is_load() {
            self.mode = QueryMode::Delete(DeleteSpec::new());
        }
        self
    }

    /// Apply a limit to the current mode.
    ///
    /// Load limits bound result size; delete limits bound mutation size.
    #[must_use]
    pub(crate) const fn limit(mut self, limit: u32) -> Self {
        match self.mode {
            QueryMode::Load(mut spec) => {
                spec.limit = Some(limit);
                self.mode = QueryMode::Load(spec);
            }
            QueryMode::Delete(mut spec) => {
                spec.limit = Some(limit);
                self.mode = QueryMode::Delete(spec);
            }
        }
        self
    }

    /// Apply an offset to a load intent.
    #[must_use]
    pub(crate) const fn offset(mut self, offset: u32) -> Self {
        if let QueryMode::Load(mut spec) = self.mode {
            spec.offset = offset;
            self.mode = QueryMode::Load(spec);
        }
        self
    }

    /// Build a model-level logical plan using Value-based access keys.
    fn build_plan_model(&self) -> Result<AccessPlannedQuery<Value>, QueryError> {
        // Phase 1: schema surface and intent validation.
        let schema_info = SchemaInfo::from_entity_model(self.model)?;
        self.validate_intent()?;

        // Phase 2: predicate normalization and access planning.
        let normalized_predicate = self
            .predicate
            .as_ref()
            .map(|predicate| {
                reject_unsupported_query_features(predicate).map_err(ValidateError::from)?;
                let predicate = normalize_enum_literals(&schema_info, predicate)?;
                Ok::<Predicate, ValidateError>(normalize(&predicate))
            })
            .transpose()?;
        let access_plan_value = match &self.key_access {
            Some(state) => access_plan_from_keys_value(&state.access),
            None => plan_access(self.model, &schema_info, normalized_predicate.as_ref())?,
        };

        // Phase 3: assemble the executor-ready plan.
        let scalar = ScalarPlan {
            mode: self.mode,
            predicate: normalized_predicate,
            // Canonicalize ORDER BY to include an explicit primary-key tie-break.
            // This ensures explain/fingerprint/execution share one deterministic order shape.
            order: canonicalize_order_spec(self.model, self.order.clone()),
            distinct: self.distinct,
            delete_limit: match self.mode {
                QueryMode::Delete(spec) => spec.limit.map(|max_rows| DeleteLimitSpec { max_rows }),
                QueryMode::Load(_) => None,
            },
            page: match self.mode {
                QueryMode::Load(spec) => {
                    if spec.limit.is_some() || spec.offset > 0 {
                        Some(PageSpec {
                            limit: spec.limit,
                            offset: spec.offset,
                        })
                    } else {
                        None
                    }
                }
                QueryMode::Delete(_) => None,
            },
            consistency: self.consistency,
        };
        let mut plan =
            AccessPlannedQuery::from_parts(LogicalPlan::Scalar(scalar), access_plan_value);
        if let Some(group) = self.group.clone() {
            plan = match self.having.clone() {
                Some(having) => plan.into_grouped_with_having(group, Some(having)),
                None => plan.into_grouped(group),
            };
        }

        if plan.grouped_plan().is_some() {
            validate_group_query_semantics(&schema_info, self.model, &plan)?;
        } else {
            validate_query_semantics(&schema_info, self.model, &plan)?;
        }

        Ok(plan)
    }

    // Validate pre-plan policy invariants and key-access rules before planning.
    fn validate_intent(&self) -> Result<(), IntentError> {
        validate_intent_plan_shape(self.mode, self.order.as_ref()).map_err(IntentError::from)?;

        let key_access_kind = self.key_access.as_ref().map(|state| match state.kind {
            KeyAccessKind::Single => IntentValidationKeyAccessKind::Single,
            KeyAccessKind::Many => IntentValidationKeyAccessKind::Many,
            KeyAccessKind::Only => IntentValidationKeyAccessKind::Only,
        });
        validate_intent_key_access_policy(
            self.key_access_conflict,
            key_access_kind,
            self.predicate.is_some(),
        )
        .map_err(IntentError::from)?;
        if self.having.is_some() && self.group.is_none() {
            return Err(IntentError::HavingRequiresGroupBy);
        }

        Ok(())
    }
}

///
/// Query
///
/// Typed, declarative query intent for a specific entity type.
///
/// This intent is:
/// - schema-agnostic at construction
/// - normalized and validated only during planning
/// - free of access-path decisions
///

///
/// PlannedQuery
///
/// Neutral query-owned planned contract produced by query planning.
/// Stores logical + access shape without executor compilation state.
///
#[derive(Debug)]
pub struct PlannedQuery<E: EntityKind> {
    plan: AccessPlannedQuery<E::Key>,
    _marker: PhantomData<E>,
}

impl<E: EntityKind> PlannedQuery<E> {
    #[must_use]
    pub(in crate::db) const fn new(plan: AccessPlannedQuery<E::Key>) -> Self {
        Self {
            plan,
            _marker: PhantomData,
        }
    }

    #[must_use]
    pub fn explain(&self) -> ExplainPlan {
        self.plan.explain_with_model(E::MODEL)
    }
}

///
/// CompiledQuery
///
/// Query-owned compiled handoff produced by `Query::plan()`.
/// This type intentionally carries only logical/access query semantics.
/// Executor runtime shape is derived explicitly at the executor boundary.
///
#[derive(Clone, Debug)]
pub struct CompiledQuery<E: EntityKind> {
    plan: AccessPlannedQuery<E::Key>,
    _marker: PhantomData<E>,
}

impl<E: EntityKind> CompiledQuery<E> {
    #[must_use]
    pub(in crate::db) const fn new(plan: AccessPlannedQuery<E::Key>) -> Self {
        Self {
            plan,
            _marker: PhantomData,
        }
    }

    #[must_use]
    pub fn explain(&self) -> ExplainPlan {
        self.plan.explain_with_model(E::MODEL)
    }

    /// Borrow planner-lowered projection semantics for this compiled query.
    #[must_use]
    #[cfg(test)]
    pub(crate) fn projection_spec(&self) -> crate::db::query::plan::expr::ProjectionSpec {
        self.plan.projection_spec(E::MODEL)
    }

    #[must_use]
    pub(in crate::db) fn into_inner(self) -> AccessPlannedQuery<E::Key> {
        self.plan
    }
}

#[derive(Debug)]
pub struct Query<E: EntityKind> {
    intent: QueryModel<'static, E::Key>,
    _marker: PhantomData<E>,
}

impl<E: EntityKind> Query<E> {
    /// Create a new intent with an explicit missing-row policy.
    /// Ignore favors idempotency and may mask index/data divergence on deletes.
    /// Use Error to surface missing rows during scan/delete execution.
    #[must_use]
    pub const fn new(consistency: MissingRowPolicy) -> Self {
        Self {
            intent: QueryModel::new(E::MODEL, consistency),
            _marker: PhantomData,
        }
    }

    /// Return the intent mode (load vs delete).
    #[must_use]
    pub const fn mode(&self) -> QueryMode {
        self.intent.mode()
    }

    #[must_use]
    pub(crate) fn has_explicit_order(&self) -> bool {
        self.intent.has_explicit_order()
    }

    #[must_use]
    pub(crate) const fn has_grouping(&self) -> bool {
        self.intent.has_grouping()
    }

    #[must_use]
    pub(crate) const fn load_spec(&self) -> Option<LoadSpec> {
        self.intent.load_spec()
    }

    /// Add a predicate, implicitly AND-ing with any existing predicate.
    #[must_use]
    pub fn filter(mut self, predicate: Predicate) -> Self {
        self.intent = self.intent.filter(predicate);
        self
    }

    /// Apply a dynamic filter expression.
    pub fn filter_expr(self, expr: FilterExpr) -> Result<Self, QueryError> {
        let Self { intent, _marker } = self;
        let intent = intent.filter_expr(expr)?;

        Ok(Self { intent, _marker })
    }

    /// Apply a dynamic sort expression.
    pub fn sort_expr(self, expr: SortExpr) -> Result<Self, QueryError> {
        let Self { intent, _marker } = self;
        let intent = intent.sort_expr(expr)?;

        Ok(Self { intent, _marker })
    }

    /// Append an ascending sort key.
    #[must_use]
    pub fn order_by(mut self, field: impl AsRef<str>) -> Self {
        self.intent = self.intent.order_by(field);
        self
    }

    /// Append a descending sort key.
    #[must_use]
    pub fn order_by_desc(mut self, field: impl AsRef<str>) -> Self {
        self.intent = self.intent.order_by_desc(field);
        self
    }

    /// Enable DISTINCT semantics for this query.
    #[must_use]
    pub fn distinct(mut self) -> Self {
        self.intent = self.intent.distinct();
        self
    }

    /// Add one GROUP BY field.
    pub fn group_by(self, field: impl AsRef<str>) -> Result<Self, QueryError> {
        let Self { intent, _marker } = self;
        let intent = intent.push_group_field(field.as_ref())?;

        Ok(Self { intent, _marker })
    }

    /// Add one aggregate terminal via composable aggregate expression.
    #[must_use]
    pub fn aggregate(mut self, aggregate: AggregateExpr) -> Self {
        self.intent = self.intent.push_group_aggregate(aggregate);
        self
    }

    /// Override grouped hard limits for grouped execution budget enforcement.
    #[must_use]
    pub fn grouped_limits(mut self, max_groups: u64, max_group_bytes: u64) -> Self {
        self.intent = self.intent.grouped_limits(max_groups, max_group_bytes);
        self
    }

    /// Add one grouped HAVING compare clause over one grouped key field.
    pub fn having_group(
        self,
        field: impl AsRef<str>,
        op: CompareOp,
        value: Value,
    ) -> Result<Self, QueryError> {
        let field = field.as_ref().to_owned();
        let Self { intent, _marker } = self;
        let intent = intent.push_having_group_clause(&field, op, value)?;

        Ok(Self { intent, _marker })
    }

    /// Add one grouped HAVING compare clause over one grouped aggregate output.
    pub fn having_aggregate(
        self,
        aggregate_index: usize,
        op: CompareOp,
        value: Value,
    ) -> Result<Self, QueryError> {
        let Self { intent, _marker } = self;
        let intent = intent.push_having_aggregate_clause(aggregate_index, op, value)?;

        Ok(Self { intent, _marker })
    }

    /// Set the access path to a single primary key lookup.
    pub(crate) fn by_id(self, id: E::Key) -> Self {
        let Self { intent, _marker } = self;
        Self {
            intent: intent.by_id(id),
            _marker,
        }
    }

    /// Set the access path to a primary key batch lookup.
    pub(crate) fn by_ids<I>(self, ids: I) -> Self
    where
        I: IntoIterator<Item = E::Key>,
    {
        let Self { intent, _marker } = self;
        Self {
            intent: intent.by_ids(ids),
            _marker,
        }
    }

    /// Mark this intent as a delete query.
    #[must_use]
    pub fn delete(mut self) -> Self {
        self.intent = self.intent.delete();
        self
    }

    /// Apply a limit to the current mode.
    ///
    /// Load limits bound result size; delete limits bound mutation size.
    /// For scalar load queries, any use of `limit` or `offset` requires an
    /// explicit `order_by(...)` so pagination is deterministic.
    /// GROUP BY queries use canonical grouped-key order by default.
    #[must_use]
    pub fn limit(mut self, limit: u32) -> Self {
        self.intent = self.intent.limit(limit);
        self
    }

    /// Apply an offset to a load intent.
    ///
    /// Scalar pagination requires an explicit `order_by(...)`.
    /// GROUP BY queries use canonical grouped-key order by default.
    #[must_use]
    pub fn offset(mut self, offset: u32) -> Self {
        self.intent = self.intent.offset(offset);
        self
    }

    /// Explain this intent without executing it.
    pub fn explain(&self) -> Result<ExplainPlan, QueryError> {
        let plan = self.planned()?;

        Ok(plan.explain())
    }

    /// Plan this intent into a neutral planned query contract.
    pub fn planned(&self) -> Result<PlannedQuery<E>, QueryError> {
        let plan = self.build_plan()?;
        let _projection = plan.projection_spec(E::MODEL);

        Ok(PlannedQuery::new(plan))
    }

    /// Compile this intent into query-owned handoff state.
    ///
    /// This boundary intentionally does not expose executor runtime shape.
    pub fn plan(&self) -> Result<CompiledQuery<E>, QueryError> {
        let plan = self.build_plan()?;
        let _projection = plan.projection_spec(E::MODEL);

        Ok(CompiledQuery::new(plan))
    }

    // Build a logical plan for the current intent.
    fn build_plan(&self) -> Result<AccessPlannedQuery<E::Key>, QueryError> {
        let plan_value = self.intent.build_plan_model()?;
        let (logical, access) = plan_value.into_parts();
        let access = access_plan_to_entity_keys::<E>(E::MODEL, access)?;
        let plan = AccessPlannedQuery::from_parts(logical, access);

        Ok(plan)
    }
}

impl<E> Query<E>
where
    E: EntityKind + SingletonEntity,
    E::Key: Default,
{
    /// Set the access path to the singleton primary key.
    pub(crate) fn only(self) -> Self {
        let Self { intent, _marker } = self;

        Self {
            intent: intent.only(E::Key::default()),
            _marker,
        }
    }
}

///
/// QueryError
///

#[derive(Debug, ThisError)]
pub enum QueryError {
    #[error("{0}")]
    Validate(#[from] ValidateError),

    #[error("{0}")]
    Plan(Box<PlanError>),

    #[error("{0}")]
    Intent(#[from] IntentError),

    #[error("{0}")]
    Response(#[from] ResponseError),

    #[error("{0}")]
    Execute(#[from] QueryExecuteError),
}

impl QueryError {
    /// Construct an execution-domain query error from one classified runtime error.
    pub(crate) fn execute(err: InternalError) -> Self {
        Self::Execute(QueryExecuteError::from(err))
    }
}

///
/// QueryExecuteError
///

#[derive(Debug, ThisError)]
pub enum QueryExecuteError {
    #[error("{0}")]
    Corruption(InternalError),

    #[error("{0}")]
    InvariantViolation(InternalError),

    #[error("{0}")]
    Conflict(InternalError),

    #[error("{0}")]
    NotFound(InternalError),

    #[error("{0}")]
    Unsupported(InternalError),

    #[error("{0}")]
    Internal(InternalError),
}

impl QueryExecuteError {
    #[must_use]
    /// Borrow the wrapped classified runtime error.
    pub const fn as_internal(&self) -> &InternalError {
        match self {
            Self::Corruption(err)
            | Self::InvariantViolation(err)
            | Self::Conflict(err)
            | Self::NotFound(err)
            | Self::Unsupported(err)
            | Self::Internal(err) => err,
        }
    }
}

impl From<InternalError> for QueryExecuteError {
    fn from(err: InternalError) -> Self {
        match err.class {
            ErrorClass::Corruption => Self::Corruption(err),
            ErrorClass::InvariantViolation => Self::InvariantViolation(err),
            ErrorClass::Conflict => Self::Conflict(err),
            ErrorClass::NotFound => Self::NotFound(err),
            ErrorClass::Unsupported => Self::Unsupported(err),
            ErrorClass::Internal => Self::Internal(err),
        }
    }
}

impl From<PlannerError> for QueryError {
    fn from(err: PlannerError) -> Self {
        match err {
            PlannerError::Plan(err) => Self::from(*err),
            PlannerError::Internal(err) => Self::execute(*err),
        }
    }
}

impl From<PlanError> for QueryError {
    fn from(err: PlanError) -> Self {
        Self::Plan(Box::new(err))
    }
}

///
/// IntentError
///

#[derive(Clone, Copy, Debug, ThisError)]
pub enum IntentError {
    #[error("{0}")]
    PlanShape(#[from] PolicyPlanError),

    #[error("by_ids() cannot be combined with predicates")]
    ByIdsWithPredicate,

    #[error("only() cannot be combined with predicates")]
    OnlyWithPredicate,

    #[error("multiple key access methods were used on the same query")]
    KeyAccessConflict,

    #[error(
        "{message}",
        message = CursorPlanError::cursor_requires_order_message()
    )]
    CursorRequiresOrder,

    #[error(
        "{message}",
        message = CursorPlanError::cursor_requires_limit_message()
    )]
    CursorRequiresLimit,

    #[error("cursor tokens can only be used with .page().execute()")]
    CursorRequiresPagedExecution,

    #[error("grouped queries require execute_grouped(...)")]
    GroupedRequiresExecuteGrouped,

    #[error("HAVING requires GROUP BY")]
    HavingRequiresGroupBy,
}

impl From<CursorPagingPolicyError> for IntentError {
    fn from(err: CursorPagingPolicyError) -> Self {
        match err {
            CursorPagingPolicyError::CursorRequiresOrder => Self::CursorRequiresOrder,
            CursorPagingPolicyError::CursorRequiresLimit => Self::CursorRequiresLimit,
        }
    }
}

impl From<IntentKeyAccessPolicyViolation> for IntentError {
    fn from(err: IntentKeyAccessPolicyViolation) -> Self {
        match err {
            IntentKeyAccessPolicyViolation::KeyAccessConflict => Self::KeyAccessConflict,
            IntentKeyAccessPolicyViolation::ByIdsWithPredicate => Self::ByIdsWithPredicate,
            IntentKeyAccessPolicyViolation::OnlyWithPredicate => Self::OnlyWithPredicate,
        }
    }
}

impl From<FluentLoadPolicyViolation> for IntentError {
    fn from(err: FluentLoadPolicyViolation) -> Self {
        match err {
            FluentLoadPolicyViolation::CursorRequiresPagedExecution => {
                Self::CursorRequiresPagedExecution
            }
            FluentLoadPolicyViolation::GroupedRequiresExecuteGrouped => {
                Self::GroupedRequiresExecuteGrouped
            }
            FluentLoadPolicyViolation::CursorRequiresOrder => Self::CursorRequiresOrder,
            FluentLoadPolicyViolation::CursorRequiresLimit => Self::CursorRequiresLimit,
        }
    }
}

/// Helper to append an ordering field while preserving existing order spec.
fn push_order(order: Option<OrderSpec>, field: &str, direction: OrderDirection) -> OrderSpec {
    match order {
        Some(mut spec) => {
            spec.fields.push((field.to_string(), direction));
            spec
        }
        None => OrderSpec {
            fields: vec![(field.to_string(), direction)],
        },
    }
}

// Normalize ORDER BY into a canonical, deterministic shape:
// - preserve user field order
// - remove explicit primary-key references from the user segment
// - append exactly one primary-key field as the terminal tie-break
fn canonicalize_order_spec(model: &EntityModel, order: Option<OrderSpec>) -> Option<OrderSpec> {
    let mut order = order?;
    if order.fields.is_empty() {
        return Some(order);
    }

    let pk_field = model.primary_key.name;
    let mut pk_direction = None;
    order.fields.retain(|(field, direction)| {
        if field == pk_field {
            if pk_direction.is_none() {
                pk_direction = Some(*direction);
            }
            false
        } else {
            true
        }
    });

    let pk_direction = pk_direction.unwrap_or(OrderDirection::Asc);
    order.fields.push((pk_field.to_string(), pk_direction));

    Some(order)
}