icydb-core 0.165.9

//! Module: query::plan::order_contract
//! Responsibility: planner-owned execution ordering contracts and direction normalization.
//! Does not own: runtime order application mechanics or cursor wire token encoding.
//! Boundary: exposes immutable order contracts consumed across planner/executor boundaries.

use crate::db::{
    access::{AccessShapeFacts, SemanticIndexKeyItemsRef},
    direction::Direction,
    query::plan::{OrderDirection, OrderSpec, order_term::index_key_item_order_terms},
};
#[cfg(test)]
use crate::{db::query::plan::order_term::index_order_terms, model::index::IndexModel};

///
/// DeterministicSecondaryOrderContract
///
/// Planner-owned shared `..., primary_key_fields` order contract with one
/// uniform direction. The non-primary-key term list may be empty, which
/// represents the primary-key-only order shape under the same normalized
/// contract.
///

///
/// DeterministicSecondaryIndexOrderMatch
///
/// Planner-owned match classification between one normalized secondary ORDER BY
/// contract and one canonical index key order.
/// This exists so covering, access-contract pushdown, and planner ranking all
/// consume the same full-vs-suffix match decision instead of re-deriving it in
/// each caller.
///

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(in crate::db) enum DeterministicSecondaryIndexOrderMatch {
    Full,
    Suffix,
    None,
}

///
/// DeterministicSecondaryIndexOrderCompatibility
///
/// Shared compatibility fact between one deterministic scalar ORDER BY
/// contract and one index-key order after a known equality-bound prefix.
/// Planner ranking and executor route pushdown both consume this value so the
/// match decision cannot drift across layers.
///

#[derive(Clone, Debug, Eq, PartialEq)]
pub(in crate::db) struct DeterministicSecondaryIndexOrderCompatibility {
    index_terms: Vec<String>,
    match_kind: DeterministicSecondaryIndexOrderMatch,
}

impl DeterministicSecondaryIndexOrderCompatibility {
    /// Build one compatibility fact from the shared order contract classifier.
    #[must_use]
    fn new(
        order_contract: &DeterministicSecondaryOrderContract,
        key_items: SemanticIndexKeyItemsRef<'_>,
        prefix_len: usize,
    ) -> Self {
        let index_terms = index_key_item_order_terms(key_items);
        let match_kind = order_contract.classify_index_match(&index_terms, prefix_len);

        Self {
            index_terms,
            match_kind,
        }
    }

    /// Return the full canonical index-order terms used for the match.
    #[must_use]
    pub(in crate::db) const fn index_terms(&self) -> &[String] {
        self.index_terms.as_slice()
    }

    /// Return the suffix terms remaining after the equality-bound prefix.
    #[must_use]
    pub(in crate::db) fn index_suffix_terms(&self, prefix_len: usize) -> Vec<String> {
        self.index_terms.iter().skip(prefix_len).cloned().collect()
    }

    /// Return the shared full-vs-suffix-vs-none match classification.
    #[must_use]
    #[cfg(test)]
    pub(in crate::db) const fn match_kind(&self) -> DeterministicSecondaryIndexOrderMatch {
        self.match_kind
    }

    /// Return whether this index traversal can satisfy the ORDER BY contract.
    #[must_use]
    pub(in crate::db) const fn is_satisfied(&self) -> bool {
        !matches!(self.match_kind, DeterministicSecondaryIndexOrderMatch::None)
    }
}

///
/// GroupedIndexOrderContract
///
/// Planner-owned grouped `ORDER BY` contract without the scalar
/// `..., primary_key` tie-break normalization.
/// This exists so grouped ranking and grouped order-only fallback share one
/// full-vs-suffix index-order classifier instead of rebuilding grouped order
/// labels and uniform-direction checks in parallel.
///

#[derive(Clone, Debug, Eq, PartialEq)]
pub(in crate::db) struct GroupedIndexOrderContract {
    terms: Vec<String>,
    direction: OrderDirection,
}

///
/// GroupedIndexOrderMatch
///
/// Planner-owned grouped-order match classification against one canonical
/// index key order.
/// This keeps grouped full-index and prefix-consumed suffix matching under one
/// owner instead of open-coding the same comparisons across planner helpers.
///

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(in crate::db) enum GroupedIndexOrderMatch {
    Full,
    Suffix,
    None,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub(in crate::db) struct DeterministicSecondaryOrderContract {
    non_primary_key_terms: Vec<String>,
    direction: OrderDirection,
}

impl DeterministicSecondaryOrderContract {
    /// Build one normalized deterministic `..., primary_key_fields` order
    /// contract from one executor-facing ORDER BY spec.
    #[must_use]
    #[cfg(test)]
    pub(in crate::db) fn from_order_spec(
        order: &OrderSpec,
        primary_key_name: &str,
    ) -> Option<Self> {
        Self::from_order_spec_fields(order, &[primary_key_name])
    }

    /// Build one normalized deterministic order contract with an ordered
    /// primary-key field suffix.
    #[must_use]
    pub(in crate::db) fn from_order_spec_fields(
        order: &OrderSpec,
        primary_key_names: &[&str],
    ) -> Option<Self> {
        let direction = order.fields.last()?.direction();
        has_exact_ordered_primary_key_tie_break_fields(order.fields.as_slice(), primary_key_names)
            .then_some(())?;
        if order
            .fields
            .iter()
            .any(|term| term.direction() != direction)
        {
            return None;
        }

        Some(Self {
            non_primary_key_terms: order
                .fields
                .iter()
                .take(order.fields.len().saturating_sub(primary_key_names.len()))
                .map(crate::db::query::plan::OrderTerm::rendered_label)
                .collect(),
            direction,
        })
    }

    /// Return the shared direction across the full deterministic order shape.
    #[must_use]
    pub(in crate::db) const fn direction(&self) -> OrderDirection {
        self.direction
    }

    /// Borrow the normalized non-primary-key ORDER BY terms.
    #[must_use]
    pub(in crate::db) const fn non_primary_key_terms(&self) -> &[String] {
        self.non_primary_key_terms.as_slice()
    }

    /// Return true when the normalized non-primary-key terms match one expected
    /// canonical term sequence.
    #[must_use]
    pub(in crate::db) fn matches_expected_non_primary_key_terms<'a, I>(&self, expected: I) -> bool
    where
        I: IntoIterator<Item = &'a str>,
    {
        self.non_primary_key_terms
            .iter()
            .map(String::as_str)
            .eq(expected)
    }

    /// Return true when this normalized contract matches one index suffix.
    #[must_use]
    pub(in crate::db) fn matches_index_suffix<S>(
        &self,
        index_fields: &[S],
        prefix_len: usize,
    ) -> bool
    where
        S: AsRef<str>,
    {
        if prefix_len > index_fields.len() {
            return false;
        }

        self.matches_expected_non_primary_key_terms(
            index_fields[prefix_len..].iter().map(AsRef::as_ref),
        )
    }

    /// Return true when this normalized contract matches one full index order.
    #[must_use]
    pub(in crate::db) fn matches_index_full<S>(&self, index_fields: &[S]) -> bool
    where
        S: AsRef<str>,
    {
        self.matches_expected_non_primary_key_terms(index_fields.iter().map(AsRef::as_ref))
    }

    /// Classify how this normalized contract matches one canonical index key
    /// order after one equality-bound prefix.
    #[must_use]
    pub(in crate::db) fn classify_index_match<S>(
        &self,
        index_fields: &[S],
        prefix_len: usize,
    ) -> DeterministicSecondaryIndexOrderMatch
    where
        S: AsRef<str>,
    {
        if self.matches_index_suffix(index_fields, prefix_len) {
            return DeterministicSecondaryIndexOrderMatch::Suffix;
        }
        if self.matches_index_full(index_fields) {
            return DeterministicSecondaryIndexOrderMatch::Full;
        }

        DeterministicSecondaryIndexOrderMatch::None
    }
}

/// Return the shared scalar secondary-index order compatibility fact.
#[must_use]
#[cfg(test)]
pub(in crate::db) fn deterministic_secondary_index_order_compatibility(
    order_contract: &DeterministicSecondaryOrderContract,
    index: &IndexModel,
    prefix_len: usize,
) -> DeterministicSecondaryIndexOrderCompatibility {
    deterministic_secondary_index_key_items_order_compatibility(
        order_contract,
        SemanticIndexKeyItemsRef::Static(index.key_items()),
        prefix_len,
    )
}

/// Return the shared scalar secondary-index order compatibility fact from
/// reduced key-item facts.
#[must_use]
pub(in crate::db) fn deterministic_secondary_index_key_items_order_compatibility(
    order_contract: &DeterministicSecondaryOrderContract,
    key_items: SemanticIndexKeyItemsRef<'_>,
    prefix_len: usize,
) -> DeterministicSecondaryIndexOrderCompatibility {
    DeterministicSecondaryIndexOrderCompatibility::new(order_contract, key_items, prefix_len)
}

/// Return whether one deterministic scalar ORDER BY contract is satisfied by
/// one secondary-index traversal after the equality-bound prefix.
#[must_use]
#[cfg(test)]
pub(in crate::db) fn deterministic_secondary_index_order_satisfied(
    order_contract: &DeterministicSecondaryOrderContract,
    index: &IndexModel,
    prefix_len: usize,
) -> bool {
    deterministic_secondary_index_order_compatibility(order_contract, index, prefix_len)
        .is_satisfied()
}

/// Return whether reduced key-item facts satisfy one deterministic scalar
/// ORDER BY contract after the equality-bound prefix.
#[must_use]
pub(in crate::db) fn deterministic_secondary_index_key_items_order_satisfied(
    order_contract: &DeterministicSecondaryOrderContract,
    key_items: SemanticIndexKeyItemsRef<'_>,
    prefix_len: usize,
) -> bool {
    deterministic_secondary_index_key_items_order_compatibility(
        order_contract,
        key_items,
        prefix_len,
    )
    .is_satisfied()
}

/// Return whether accepted field-path index order terms satisfy one
/// deterministic scalar ORDER BY contract after the equality-bound prefix.
#[must_use]
pub(in crate::db) fn deterministic_secondary_index_order_terms_satisfied(
    order_contract: &DeterministicSecondaryOrderContract,
    index_terms: &[String],
    prefix_len: usize,
) -> bool {
    !matches!(
        order_contract.classify_index_match(index_terms, prefix_len),
        DeterministicSecondaryIndexOrderMatch::None
    )
}

// Empty non-unique prefix scans still interleave several leading-key groups, so
// their traversal order cannot satisfy arbitrary suffix ordering on its own.
fn prefix_order_contract_safe(access_shape_facts: &AccessShapeFacts) -> bool {
    let Some(details) = access_shape_facts.single_path_index_prefix_details() else {
        return false;
    };

    details.is_unique() || details.slot_arity() > 0
}

/// Return whether one deterministic scalar ORDER BY contract is satisfied by
/// the final stream order of one access-capability shape.
#[must_use]
pub(in crate::db) fn access_satisfies_deterministic_secondary_order_contract(
    access_shape_facts: &AccessShapeFacts,
    order_contract: &DeterministicSecondaryOrderContract,
) -> bool {
    if !access_shape_facts.is_single_path() {
        return false;
    }

    if let Some(details) = access_shape_facts.single_path_index_prefix_details() {
        return prefix_order_contract_safe(access_shape_facts)
            && deterministic_secondary_index_key_items_order_satisfied(
                order_contract,
                details.key_items(),
                details.slot_arity(),
            );
    }

    access_shape_facts
        .single_path_index_range_details()
        .is_some_and(|details| {
            deterministic_secondary_index_key_items_order_satisfied(
                order_contract,
                details.key_items(),
                details.slot_arity(),
            )
        })
}

impl GroupedIndexOrderContract {
    /// Build one grouped ORDER BY contract from one uniform-direction grouped
    /// order spec.
    #[must_use]
    pub(in crate::db) fn from_order_spec(order: &OrderSpec) -> Option<Self> {
        let direction = order
            .fields
            .first()
            .map(crate::db::query::plan::OrderTerm::direction)?;
        if order
            .fields
            .iter()
            .any(|term| term.direction() != direction)
        {
            return None;
        }

        Some(Self {
            terms: order
                .fields
                .iter()
                .map(crate::db::query::plan::OrderTerm::rendered_label)
                .collect(),
            direction,
        })
    }

    /// Return true when this grouped order matches one full canonical index
    /// order.
    #[must_use]
    pub(in crate::db) fn matches_index_full<S>(&self, index_fields: &[S]) -> bool
    where
        S: AsRef<str>,
    {
        self.terms
            .iter()
            .map(String::as_str)
            .eq(index_fields.iter().map(AsRef::as_ref))
    }

    /// Return true when this grouped order matches one canonical index suffix
    /// after one equality-bound prefix.
    #[must_use]
    pub(in crate::db) fn matches_index_suffix<S>(
        &self,
        index_fields: &[S],
        prefix_len: usize,
    ) -> bool
    where
        S: AsRef<str>,
    {
        if prefix_len > index_fields.len() {
            return false;
        }

        self.terms
            .iter()
            .map(String::as_str)
            .eq(index_fields[prefix_len..].iter().map(AsRef::as_ref))
    }

    /// Classify how this grouped order matches one canonical index key order
    /// after one equality-bound prefix.
    #[must_use]
    pub(in crate::db) fn classify_index_match<S>(
        &self,
        index_fields: &[S],
        prefix_len: usize,
    ) -> GroupedIndexOrderMatch
    where
        S: AsRef<str>,
    {
        if prefix_len > 0 && self.matches_index_suffix(index_fields, prefix_len) {
            return GroupedIndexOrderMatch::Suffix;
        }
        if self.matches_index_full(index_fields) {
            return GroupedIndexOrderMatch::Full;
        }

        GroupedIndexOrderMatch::None
    }
}

/// Return the shared grouped secondary-index order match classification.
#[must_use]
#[cfg(test)]
pub(in crate::db) fn grouped_index_order_match(
    order_contract: &GroupedIndexOrderContract,
    index: &IndexModel,
    prefix_len: usize,
) -> GroupedIndexOrderMatch {
    let index_terms = index_order_terms(index);

    order_contract.classify_index_match(&index_terms, prefix_len)
}

/// Return whether accepted field-path index order terms satisfy one grouped
/// ORDER BY contract after the equality-bound prefix.
#[must_use]
pub(in crate::db) fn grouped_index_order_terms_satisfied(
    order_contract: &GroupedIndexOrderContract,
    index_terms: &[String],
    prefix_len: usize,
) -> bool {
    !matches!(
        order_contract.classify_index_match(index_terms, prefix_len),
        GroupedIndexOrderMatch::None
    )
}

impl OrderSpec {
    /// Return ordering direction when `ORDER BY` is exactly the ordered
    /// primary-key field list and every term has the same direction.
    #[must_use]
    pub(in crate::db) fn primary_key_only_direction_fields(
        &self,
        primary_key_names: &[&str],
    ) -> Option<OrderDirection> {
        if primary_key_names.is_empty() || self.fields.len() != primary_key_names.len() {
            return None;
        }

        let direction = self.fields.first()?.direction();
        self.fields
            .iter()
            .zip(primary_key_names.iter())
            .all(|(term, primary_key_name)| {
                term.direct_field() == Some(*primary_key_name) && term.direction() == direction
            })
            .then_some(direction)
    }

    /// Return true when ORDER BY has exactly the ordered primary-key fields as
    /// the terminal deterministic tie-break.
    #[must_use]
    pub(in crate::db) fn has_exact_primary_key_tie_break_fields(
        &self,
        primary_key_names: &[&str],
    ) -> bool {
        has_exact_ordered_primary_key_tie_break_fields(self.fields.as_slice(), primary_key_names)
    }

    /// Return the normalized deterministic `..., primary_key` order contract,
    /// if one exists for this ORDER BY shape.
    #[must_use]
    #[cfg(test)]
    pub(in crate::db) fn deterministic_secondary_order_contract(
        &self,
        primary_key_name: &str,
    ) -> Option<DeterministicSecondaryOrderContract> {
        DeterministicSecondaryOrderContract::from_order_spec(self, primary_key_name)
    }

    /// Return the normalized deterministic `..., primary_key_fields` order
    /// contract, if one exists for this ORDER BY shape.
    #[must_use]
    pub(in crate::db) fn deterministic_secondary_order_contract_fields(
        &self,
        primary_key_names: &[&str],
    ) -> Option<DeterministicSecondaryOrderContract> {
        DeterministicSecondaryOrderContract::from_order_spec_fields(self, primary_key_names)
    }

    /// Return the grouped order contract when grouped ORDER BY stays on one
    /// uniform direction.
    #[must_use]
    pub(in crate::db) fn grouped_index_order_contract(&self) -> Option<GroupedIndexOrderContract> {
        GroupedIndexOrderContract::from_order_spec(self)
    }
}

///
/// ExecutionOrdering
///
/// Planner-owned execution ordering selection.
/// Keeps scalar and grouped ordering contracts explicit at one boundary.
///
#[derive(Clone, Debug, Eq, PartialEq)]
pub(in crate::db) enum ExecutionOrdering {
    PrimaryKey,
    Explicit(OrderSpec),
    Grouped(Option<OrderSpec>),
}

///
/// ExecutionOrderContract
///
/// Immutable planner-projected execution ordering contract.
/// Encodes ordering shape, canonical traversal direction, and cursor support.
///
#[derive(Clone, Debug, Eq, PartialEq)]
pub(in crate::db) struct ExecutionOrderContract {
    ordering: ExecutionOrdering,
    direction: Direction,
    supports_cursor: bool,
}

impl ExecutionOrderContract {
    /// Construct one immutable planner-projected execution order contract.
    #[must_use]
    const fn new(ordering: ExecutionOrdering, direction: Direction, supports_cursor: bool) -> Self {
        Self {
            ordering,
            direction,
            supports_cursor,
        }
    }

    /// Build one execution ordering contract from grouped/order plan shape.
    #[must_use]
    pub(in crate::db) fn from_plan(is_grouped: bool, order: Option<&OrderSpec>) -> Self {
        let direction = primary_scan_direction(order);
        let ordering = if is_grouped {
            ExecutionOrdering::Grouped(order.cloned())
        } else {
            match order.cloned() {
                Some(order) => ExecutionOrdering::Explicit(order),
                None => ExecutionOrdering::PrimaryKey,
            }
        };
        let supports_cursor = is_grouped || order.is_some();

        Self::new(ordering, direction, supports_cursor)
    }

    #[must_use]
    pub(in crate::db) const fn ordering(&self) -> &ExecutionOrdering {
        &self.ordering
    }

    #[must_use]
    pub(in crate::db) const fn direction(&self) -> Direction {
        self.direction
    }

    /// Return canonical primary scan direction for this execution contract.
    #[must_use]
    pub(in crate::db) const fn primary_scan_direction(&self) -> Direction {
        self.direction
    }

    #[must_use]
    pub(in crate::db) const fn is_grouped(&self) -> bool {
        matches!(&self.ordering, ExecutionOrdering::Grouped(_))
    }

    #[must_use]
    pub(in crate::db) const fn order_spec(&self) -> Option<&OrderSpec> {
        match &self.ordering {
            ExecutionOrdering::PrimaryKey => None,
            ExecutionOrdering::Explicit(order) => Some(order),
            ExecutionOrdering::Grouped(order) => order.as_ref(),
        }
    }
}

fn primary_scan_direction(order: Option<&OrderSpec>) -> Direction {
    let Some(order) = order else {
        return Direction::Asc;
    };
    let Some(term) = order.fields.first() else {
        return Direction::Asc;
    };

    match term.direction() {
        OrderDirection::Asc => Direction::Asc,
        OrderDirection::Desc => Direction::Desc,
    }
}

fn has_exact_ordered_primary_key_tie_break_fields(
    fields: &[crate::db::query::plan::OrderTerm],
    primary_key_names: &[&str],
) -> bool {
    if primary_key_names.is_empty() || fields.len() < primary_key_names.len() {
        return false;
    }

    let split = fields.len() - primary_key_names.len();
    let (prefix, suffix) = fields.split_at(split);
    if !suffix
        .iter()
        .zip(primary_key_names.iter())
        .all(|(term, primary_key_name)| term.direct_field() == Some(*primary_key_name))
    {
        return false;
    }

    !prefix.iter().any(|term| {
        term.direct_field()
            .is_some_and(|field| primary_key_names.contains(&field))
    })
}