icydb-core 0.231.0

//! Module: db::query::plan::planner::index_select
//! Selects and orders candidate indexes for predicate-backed access planning.

use crate::{
    db::{
        access::{AccessPath, SemanticIndexAccessContract},
        index::{TextPrefixBoundMode, starts_with_component_bounds},
        numeric::compare_numeric_or_strict_order,
        predicate::{CoercionId, CompareOp, ComparePredicate, Predicate},
        schema::{FieldType, SchemaInfo, literal_matches_type},
    },
    value::Value,
};
use std::{cmp::Ordering, ops::Bound};

pub(in crate::db::query) fn eligible_sorted_index_contracts(
    indexes: &[SemanticIndexAccessContract],
    query_predicate: &Predicate,
) -> Vec<SemanticIndexAccessContract> {
    debug_assert!(index_contracts_are_sorted(indexes));
    indexes
        .iter()
        .filter(|index| index_contract_predicate_implied_by_query(index, query_predicate))
        .cloned()
        .collect()
}

fn index_contracts_are_sorted(indexes: &[SemanticIndexAccessContract]) -> bool {
    indexes
        .windows(2)
        .all(|pair| pair[0].name() <= pair[1].name())
}

pub(in crate::db::query) fn index_literal_matches_schema(
    schema: &SchemaInfo,
    field: &str,
    value: &Value,
) -> bool {
    index_field_literal_matcher(schema, field).matches(value)
}

#[derive(Clone, Copy)]
pub(in crate::db::query) struct IndexFieldLiteralMatcher<'a> {
    field_type: Option<&'a FieldType>,
}

impl IndexFieldLiteralMatcher<'_> {
    #[must_use]
    pub(in crate::db::query) fn matches(self, value: &Value) -> bool {
        self.field_type
            .is_some_and(|field_type| literal_matches_type(value, field_type))
    }
}

#[must_use]
pub(in crate::db::query) fn index_field_literal_matcher<'a>(
    schema: &'a SchemaInfo,
    field: &str,
) -> IndexFieldLiteralMatcher<'a> {
    IndexFieldLiteralMatcher {
        field_type: schema.field(field),
    }
}

// Filtered indexes are eligible only when the full query predicate implies the
// index predicate. This check is intentionally conservative and fail-closed:
// unsupported predicate forms are treated as non-implying.
fn index_contract_predicate_implied_by_query(
    index: &SemanticIndexAccessContract,
    query_predicate: &Predicate,
) -> bool {
    let Some(index_predicate) = index.predicate_semantics() else {
        return true;
    };

    predicate_implies_predicate_for_planner(query_predicate, index_predicate)
}

pub(in crate::db) fn residual_query_predicate_after_filtered_access_contract(
    index: SemanticIndexAccessContract,
    query_predicate: &Predicate,
) -> Option<Predicate> {
    let Some(index_predicate) = index.predicate_semantics() else {
        return Some(query_predicate.clone());
    };

    if !predicate_implies_predicate_for_planner(query_predicate, index_predicate) {
        return Some(query_predicate.clone());
    }

    strip_query_clauses_satisfied_by_filtered_guard(query_predicate, index_predicate)
}

pub(in crate::db) fn residual_query_predicate_after_access_path_bounds(
    access_path: Option<&AccessPath<Value>>,
    query_predicate: &Predicate,
) -> Option<Predicate> {
    let Some(access_path) = access_path else {
        return Some(query_predicate.clone());
    };

    // Phase 1: derive only clauses that the concrete access path already
    // guarantees. Range paths may prove their own selected lower/upper bounds,
    // but stricter sibling predicates still remain residual.
    let implied_bounds = if let Some((index, values)) = access_path.as_index_prefix_contract() {
        AccessBoundClauses {
            equalities: access_bound_equalities(index, values),
            ranges: Vec::new(),
            branch_in: None,
        }
    } else if let Some((index, values)) = access_path.as_index_multi_lookup_contract() {
        AccessBoundClauses {
            equalities: Vec::new(),
            ranges: Vec::new(),
            branch_in: access_bound_branch_in(&index, 0, values),
        }
    } else if let Some(spec) = access_path.as_index_branch_set_spec() {
        AccessBoundClauses {
            equalities: access_bound_equalities(spec.index(), spec.fixed_values()),
            ranges: Vec::new(),
            branch_in: access_bound_branch_in(
                spec.index_ref(),
                spec.branch_slot(),
                spec.branch_values(),
            ),
        }
    } else if let Some(spec) = access_path.as_index_range() {
        let index = spec.index();
        AccessBoundClauses {
            equalities: access_bound_equalities(index.clone(), spec.prefix_values()),
            ranges: access_bound_range_clauses(
                index,
                spec.field_slots(),
                spec.lower(),
                spec.upper(),
            ),
            branch_in: None,
        }
    } else {
        AccessBoundClauses::default()
    };
    if implied_bounds.is_empty() {
        return Some(query_predicate.clone());
    }

    // Phase 2: strip only clauses already implied by those fixed equality
    // bounds so execution does not retain redundant runtime residual filtering.
    strip_query_clauses_satisfied_by_access_bounds(query_predicate, &implied_bounds)
}

pub(in crate::db::query::plan) fn predicate_implies_predicate_for_planner(
    implying: &Predicate,
    required: &Predicate,
) -> bool {
    let Some(required) = required_implication_clauses(required) else {
        return false;
    };
    let query = query_implication_clauses(implying);

    match query {
        QueryImplicationClauses::Unsatisfiable => true,
        QueryImplicationClauses::Unknown => false,
        QueryImplicationClauses::Clauses(query_clauses) => match required {
            RequiredImplicationClauses::Unsatisfiable => false,
            RequiredImplicationClauses::Clauses(required_clauses) => {
                required_clauses.compares.iter().all(|required_clause| {
                    query_clauses.compares.iter().any(|query_clause| {
                        query_clause_implies_required(query_clause, required_clause)
                    })
                }) && required_clauses
                    .non_null_fields
                    .iter()
                    .all(|required_field| {
                        query_clauses.non_null_fields.contains(required_field)
                            || query_clauses.compares.iter().any(|query_clause| {
                                comparison_proves_field_non_null(query_clause, required_field)
                            })
                    })
            }
        },
    }
}

fn strip_query_clauses_satisfied_by_filtered_guard(
    query_predicate: &Predicate,
    index_predicate: &Predicate,
) -> Option<Predicate> {
    strip_query_clauses(
        query_predicate,
        |cmp| {
            compare_clause_supported(cmp)
                && predicate_implies_predicate_for_planner(
                    index_predicate,
                    &Predicate::Compare(cmp.clone()),
                )
        },
        |field| {
            predicate_implies_predicate_for_planner(
                index_predicate,
                &Predicate::IsNotNull {
                    field: field.to_string(),
                },
            )
        },
    )
}

fn access_bound_equalities(
    index: SemanticIndexAccessContract,
    values: &[Value],
) -> Vec<ComparePredicate> {
    (0..values.len())
        .zip(values.iter())
        .filter_map(|(slot, value)| {
            let field = index.key_field_at(slot)?;

            Some(ComparePredicate::with_coercion(
                field,
                CompareOp::Eq,
                value.clone(),
                CoercionId::Strict,
            ))
        })
        .collect()
}

#[derive(Default)]
struct AccessBoundClauses<'a> {
    equalities: Vec<ComparePredicate>,
    ranges: Vec<ComparePredicate>,
    branch_in: Option<AccessBoundBranchIn<'a>>,
}

struct AccessBoundBranchIn<'a> {
    field: String,
    values: &'a [Value],
}

impl AccessBoundClauses<'_> {
    const fn is_empty(&self) -> bool {
        self.equalities.is_empty() && self.ranges.is_empty() && self.branch_in.is_none()
    }
}

fn access_bound_range_clauses(
    index: SemanticIndexAccessContract,
    field_slots: &[usize],
    lower: &Bound<Value>,
    upper: &Bound<Value>,
) -> Vec<ComparePredicate> {
    let Some(range_slot) = field_slots.last().copied() else {
        return Vec::new();
    };
    let Some(field) = index.key_field_at(range_slot) else {
        return Vec::new();
    };

    let mut ranges = Vec::with_capacity(2);
    if let Some(lower) = access_bound_lower_range_clause(field, lower) {
        ranges.push(lower);
    }
    if let Some(upper) = access_bound_upper_range_clause(field, upper) {
        ranges.push(upper);
    }
    ranges
}

fn access_bound_lower_range_clause(field: &str, bound: &Bound<Value>) -> Option<ComparePredicate> {
    match bound {
        Bound::Included(value) => Some(ComparePredicate::with_coercion(
            field,
            CompareOp::Gte,
            value.clone(),
            CoercionId::Strict,
        )),
        Bound::Excluded(value) => Some(ComparePredicate::with_coercion(
            field,
            CompareOp::Gt,
            value.clone(),
            CoercionId::Strict,
        )),
        Bound::Unbounded => None,
    }
}

fn access_bound_upper_range_clause(field: &str, bound: &Bound<Value>) -> Option<ComparePredicate> {
    match bound {
        Bound::Included(value) => Some(ComparePredicate::with_coercion(
            field,
            CompareOp::Lte,
            value.clone(),
            CoercionId::Strict,
        )),
        Bound::Excluded(value) => Some(ComparePredicate::with_coercion(
            field,
            CompareOp::Lt,
            value.clone(),
            CoercionId::Strict,
        )),
        Bound::Unbounded => None,
    }
}

fn access_bound_branch_in<'values>(
    index: &SemanticIndexAccessContract,
    branch_slot: usize,
    branch_values: &'values [Value],
) -> Option<AccessBoundBranchIn<'values>> {
    let field = index.key_field_at(branch_slot)?;

    Some(AccessBoundBranchIn {
        field: field.to_string(),
        values: branch_values,
    })
}

fn strip_query_clauses_satisfied_by_access_bounds(
    query_predicate: &Predicate,
    implied_bounds: &AccessBoundClauses,
) -> Option<Predicate> {
    strip_query_clauses(
        query_predicate,
        |cmp| access_bound_clauses_imply_required(implied_bounds, cmp),
        |_| false,
    )
}

fn access_bound_clauses_imply_required(
    implied_bounds: &AccessBoundClauses,
    cmp: &ComparePredicate,
) -> bool {
    access_bound_text_prefix_range_implies_required(implied_bounds, cmp)
        || branch_in_clause_implies_required(implied_bounds.branch_in.as_ref(), cmp)
        || implied_bounds
            .equalities
            .iter()
            .any(|bound| equality_bound_implies_required(bound, cmp))
        || implied_bounds
            .ranges
            .iter()
            .any(|bound| range_bound_implies_required(bound, cmp))
}

fn access_bound_text_prefix_range_implies_required(
    implied_bounds: &AccessBoundClauses,
    cmp: &ComparePredicate,
) -> bool {
    if cmp.op() != CompareOp::StartsWith || cmp.coercion().id != CoercionId::Strict {
        return false;
    }
    let Value::Text(prefix) = cmp.value() else {
        return false;
    };
    let Some((lower, upper)) = starts_with_component_bounds(prefix, TextPrefixBoundMode::Strict)
    else {
        return false;
    };

    access_bound_ranges_include_lower_bound(cmp.field(), &implied_bounds.ranges, &lower)
        && access_bound_ranges_include_upper_bound(cmp.field(), &implied_bounds.ranges, &upper)
}

fn access_bound_ranges_include_lower_bound(
    field: &str,
    ranges: &[ComparePredicate],
    required: &Bound<Value>,
) -> bool {
    let Some(required_clause) = access_bound_lower_range_clause(field, required) else {
        return true;
    };

    ranges
        .iter()
        .any(|bound| range_bound_implies_required(bound, &required_clause))
}

fn access_bound_ranges_include_upper_bound(
    field: &str,
    ranges: &[ComparePredicate],
    required: &Bound<Value>,
) -> bool {
    let Some(required_clause) = access_bound_upper_range_clause(field, required) else {
        return true;
    };

    ranges
        .iter()
        .any(|bound| range_bound_implies_required(bound, &required_clause))
}

fn equality_bound_implies_required(bound: &ComparePredicate, cmp: &ComparePredicate) -> bool {
    if bound.field() != cmp.field() || bound.op() != CompareOp::Eq {
        return false;
    }

    match cmp.op() {
        CompareOp::Eq | CompareOp::Gt | CompareOp::Gte | CompareOp::Lt | CompareOp::Lte => {
            compare_clause_supported(cmp) && query_clause_implies_required(bound, cmp)
        }
        CompareOp::Ne => !values_equal(bound.value(), cmp.value()),
        CompareOp::In => list_contains_value(cmp.value(), bound.value()),
        CompareOp::NotIn => !list_contains_value(cmp.value(), bound.value()),
        CompareOp::Contains | CompareOp::StartsWith | CompareOp::EndsWith => false,
    }
}

fn range_bound_implies_required(bound: &ComparePredicate, cmp: &ComparePredicate) -> bool {
    if bound.field() != cmp.field() {
        return false;
    }

    compare_clause_supported(cmp) && query_clause_implies_required(bound, cmp)
}

fn branch_in_clause_implies_required(
    branch_in: Option<&AccessBoundBranchIn<'_>>,
    cmp: &ComparePredicate,
) -> bool {
    let Some(branch_in) = branch_in else {
        return false;
    };
    if cmp.field() != branch_in.field.as_str() {
        return false;
    }

    match cmp.op() {
        CompareOp::Eq => branch_in
            .values
            .iter()
            .all(|branch_value| values_equal(branch_value, cmp.value())),
        CompareOp::Ne => branch_in
            .values
            .iter()
            .all(|branch_value| !values_equal(branch_value, cmp.value())),
        CompareOp::In => list_contains_all_values(cmp.value(), branch_in.values),
        CompareOp::NotIn => branch_in
            .values
            .iter()
            .all(|branch_value| !list_contains_value(cmp.value(), branch_value)),
        CompareOp::Gt
        | CompareOp::Gte
        | CompareOp::Lt
        | CompareOp::Lte
        | CompareOp::Contains
        | CompareOp::StartsWith
        | CompareOp::EndsWith => false,
    }
}

fn list_contains_value(list: &Value, value: &Value) -> bool {
    let Value::List(values) = list else {
        return false;
    };

    values
        .iter()
        .any(|candidate| values_equal(candidate, value))
}

fn list_contains_all_values(list: &Value, required_values: &[Value]) -> bool {
    let Value::List(values) = list else {
        return false;
    };
    if values == required_values {
        return true;
    }

    required_values.iter().all(|value| {
        values
            .iter()
            .any(|candidate| values_equal(candidate, value))
    })
}

fn values_equal(left: &Value, right: &Value) -> bool {
    compare_values(left, right).is_some_and(Ordering::is_eq)
}

// Both residual-stripping paths share the same recursive AND-collapse contract;
// they differ only in which comparison and exact non-null clauses are already
// guaranteed by the selected access contract.
fn strip_query_clauses<F, N>(
    query_predicate: &Predicate,
    compare_is_redundant: F,
    non_null_is_redundant: N,
) -> Option<Predicate>
where
    F: Fn(&ComparePredicate) -> bool + Copy,
    N: Fn(&str) -> bool + Copy,
{
    match query_predicate {
        Predicate::And(children) => {
            let mut residual_children = Vec::with_capacity(children.len());
            for child in children {
                if let Some(residual_child) =
                    strip_query_clauses(child, compare_is_redundant, non_null_is_redundant)
                {
                    residual_children.push(residual_child);
                }
            }

            match residual_children.len() {
                0 => None,
                1 => residual_children.pop(),
                _ => Some(Predicate::And(residual_children)),
            }
        }
        Predicate::Compare(cmp) if compare_is_redundant(cmp) => None,
        Predicate::IsNotNull { field } if non_null_is_redundant(field) => None,
        Predicate::True => None,
        Predicate::False
        | Predicate::Or(_)
        | Predicate::Not(_)
        | Predicate::CompareFields(_)
        | Predicate::Compare(_)
        | Predicate::IsNull { .. }
        | Predicate::IsNotNull { .. }
        | Predicate::IsMissing { .. }
        | Predicate::IsEmpty { .. }
        | Predicate::IsNotEmpty { .. }
        | Predicate::TextContains { .. }
        | Predicate::TextContainsCi { .. } => Some(query_predicate.clone()),
    }
}

///
/// QueryImplicationClauses
///
/// Supported clauses extracted from one query predicate for implication checks.
///

enum QueryImplicationClauses<'a> {
    Clauses(ImplicationClauses<'a>),
    Unsatisfiable,
    Unknown,
}

#[derive(Default)]
struct ImplicationClauses<'a> {
    compares: Vec<&'a ComparePredicate>,
    non_null_fields: Vec<&'a str>,
}

#[derive(Clone, Copy)]
enum CompareClauseMode {
    Query,
    Required,
}

enum CompareClauseCollect {
    Known,
    Unsatisfiable,
    Unknown,
}

///
/// RequiredImplicationClauses
///
/// Supported clauses extracted from one index predicate for implication checks.
///

enum RequiredImplicationClauses<'a> {
    Clauses(ImplicationClauses<'a>),
    Unsatisfiable,
}

fn query_implication_clauses(predicate: &Predicate) -> QueryImplicationClauses<'_> {
    match predicate {
        Predicate::False => QueryImplicationClauses::Unsatisfiable,
        Predicate::True => QueryImplicationClauses::Clauses(ImplicationClauses::default()),
        Predicate::Compare(cmp) => {
            if compare_clause_supported(cmp) {
                QueryImplicationClauses::Clauses(ImplicationClauses {
                    compares: vec![cmp],
                    non_null_fields: Vec::new(),
                })
            } else {
                QueryImplicationClauses::Unknown
            }
        }
        Predicate::IsNotNull { field } => QueryImplicationClauses::Clauses(ImplicationClauses {
            compares: Vec::new(),
            non_null_fields: vec![field],
        }),
        Predicate::And(children) => {
            let mut clauses = ImplicationClauses::default();
            for child in children {
                match collect_implication_clauses(child, &mut clauses, CompareClauseMode::Query) {
                    CompareClauseCollect::Unsatisfiable => {
                        return QueryImplicationClauses::Unsatisfiable;
                    }
                    CompareClauseCollect::Known | CompareClauseCollect::Unknown => {}
                }
            }

            QueryImplicationClauses::Clauses(clauses)
        }
        Predicate::Or(_)
        | Predicate::Not(_)
        | Predicate::CompareFields(_)
        | Predicate::IsNull { .. }
        | Predicate::IsMissing { .. }
        | Predicate::IsEmpty { .. }
        | Predicate::IsNotEmpty { .. }
        | Predicate::TextContains { .. }
        | Predicate::TextContainsCi { .. } => QueryImplicationClauses::Unknown,
    }
}

fn required_implication_clauses(predicate: &Predicate) -> Option<RequiredImplicationClauses<'_>> {
    match predicate {
        Predicate::True => Some(RequiredImplicationClauses::Clauses(
            ImplicationClauses::default(),
        )),
        Predicate::False => Some(RequiredImplicationClauses::Unsatisfiable),
        _ => {
            let mut clauses = ImplicationClauses::default();
            match collect_implication_clauses(predicate, &mut clauses, CompareClauseMode::Required)
            {
                CompareClauseCollect::Known => {}
                CompareClauseCollect::Unsatisfiable => {
                    return Some(RequiredImplicationClauses::Unsatisfiable);
                }
                CompareClauseCollect::Unknown => return None,
            }
            Some(RequiredImplicationClauses::Clauses(clauses))
        }
    }
}

fn collect_implication_clauses<'a>(
    predicate: &'a Predicate,
    out: &mut ImplicationClauses<'a>,
    mode: CompareClauseMode,
) -> CompareClauseCollect {
    match predicate {
        Predicate::And(children) => {
            for child in children {
                match collect_implication_clauses(child, out, mode) {
                    CompareClauseCollect::Known => {}
                    CompareClauseCollect::Unsatisfiable => {
                        return CompareClauseCollect::Unsatisfiable;
                    }
                    CompareClauseCollect::Unknown => {
                        if matches!(mode, CompareClauseMode::Required) {
                            return CompareClauseCollect::Unknown;
                        }
                    }
                }
            }

            CompareClauseCollect::Known
        }
        Predicate::Compare(cmp) => {
            if !compare_clause_supported(cmp) {
                return CompareClauseCollect::Unknown;
            }
            out.compares.push(cmp);
            CompareClauseCollect::Known
        }
        Predicate::IsNotNull { field } => {
            if !out.non_null_fields.contains(&field.as_str()) {
                out.non_null_fields.push(field);
            }
            CompareClauseCollect::Known
        }
        Predicate::True => CompareClauseCollect::Known,
        Predicate::False => match mode {
            CompareClauseMode::Query => CompareClauseCollect::Unsatisfiable,
            CompareClauseMode::Required => CompareClauseCollect::Unknown,
        },
        Predicate::CompareFields(_)
        | Predicate::Or(_)
        | Predicate::Not(_)
        | Predicate::IsNull { .. }
        | Predicate::IsMissing { .. }
        | Predicate::IsEmpty { .. }
        | Predicate::IsNotEmpty { .. }
        | Predicate::TextContains { .. }
        | Predicate::TextContainsCi { .. } => CompareClauseCollect::Unknown,
    }
}

fn comparison_proves_field_non_null(compare: &ComparePredicate, field: &str) -> bool {
    compare.field() == field
        && compare_clause_supported(compare)
        && !matches!(compare.value(), Value::Null | Value::Unit)
}

const fn compare_clause_supported(cmp: &ComparePredicate) -> bool {
    matches!(
        cmp.op(),
        CompareOp::Eq | CompareOp::Gt | CompareOp::Gte | CompareOp::Lt | CompareOp::Lte
    ) && matches!(
        cmp.coercion().id,
        CoercionId::Strict | CoercionId::NumericWiden
    )
}

fn query_clause_implies_required(query: &ComparePredicate, required: &ComparePredicate) -> bool {
    if query.field() != required.field() {
        return false;
    }
    if !compare_clause_supported(query) || !compare_clause_supported(required) {
        return false;
    }

    let query_value = query.value();
    let required_value = required.value();

    match required.op() {
        CompareOp::Eq => {
            query.op() == CompareOp::Eq
                && compare_values(query_value, required_value).is_some_and(Ordering::is_eq)
        }
        CompareOp::Gt => match query.op() {
            CompareOp::Eq | CompareOp::Gte => {
                compare_values(query_value, required_value).is_some_and(Ordering::is_gt)
            }
            CompareOp::Gt => compare_values(query_value, required_value)
                .is_some_and(|ordering| ordering.is_gt() || ordering.is_eq()),
            _ => false,
        },
        CompareOp::Gte => match query.op() {
            CompareOp::Eq => compare_values(query_value, required_value)
                .is_some_and(|ordering| ordering.is_gt() || ordering.is_eq()),
            CompareOp::Gt | CompareOp::Gte => compare_values(query_value, required_value)
                .is_some_and(|ordering| ordering.is_gt() || ordering.is_eq()),
            _ => false,
        },
        CompareOp::Lt => match query.op() {
            CompareOp::Eq | CompareOp::Lte => {
                compare_values(query_value, required_value).is_some_and(Ordering::is_lt)
            }
            CompareOp::Lt => compare_values(query_value, required_value)
                .is_some_and(|ordering| ordering.is_lt() || ordering.is_eq()),
            _ => false,
        },
        CompareOp::Lte => match query.op() {
            CompareOp::Eq => compare_values(query_value, required_value)
                .is_some_and(|ordering| ordering.is_lt() || ordering.is_eq()),
            CompareOp::Lt | CompareOp::Lte => compare_values(query_value, required_value)
                .is_some_and(|ordering| ordering.is_lt() || ordering.is_eq()),
            _ => false,
        },
        CompareOp::Ne
        | CompareOp::In
        | CompareOp::NotIn
        | CompareOp::Contains
        | CompareOp::StartsWith
        | CompareOp::EndsWith => false,
    }
}

fn compare_values(left: &Value, right: &Value) -> Option<Ordering> {
    compare_numeric_or_strict_order(left, right)
}

#[cfg(test)]
mod tests {
    use super::{
        predicate_implies_predicate_for_planner, strip_query_clauses_satisfied_by_filtered_guard,
    };
    use crate::{
        db::predicate::{CoercionId, CompareFieldsPredicate, CompareOp, Predicate},
        value::Value,
    };

    fn non_null(field: &str) -> Predicate {
        Predicate::is_not_null(field.to_string())
    }

    fn compare(field: &str, op: CompareOp, value: Value) -> Predicate {
        Predicate::Compare(crate::db::predicate::ComparePredicate::with_coercion(
            field,
            op,
            value,
            CoercionId::Strict,
        ))
    }

    #[test]
    fn nullable_guard_implication_accepts_exact_and_non_null_scalar_comparisons() {
        let required = non_null("email");
        assert!(predicate_implies_predicate_for_planner(
            &required, &required
        ));

        for op in [
            CompareOp::Eq,
            CompareOp::Gt,
            CompareOp::Gte,
            CompareOp::Lt,
            CompareOp::Lte,
        ] {
            assert!(predicate_implies_predicate_for_planner(
                &compare("email", op, Value::Text("a@example.com".to_string())),
                &required,
            ));
        }
    }

    #[test]
    fn nullable_guard_implication_requires_every_composite_guard_and_extra_filter() {
        let required = Predicate::and(vec![
            non_null("tenant"),
            non_null("email"),
            compare("active", CompareOp::Eq, Value::Bool(true)),
        ]);
        let complete = Predicate::and(vec![
            compare(
                "email",
                CompareOp::Eq,
                Value::Text("a@example.com".to_string()),
            ),
            non_null("tenant"),
            compare("active", CompareOp::Eq, Value::Bool(true)),
            Predicate::IsNotEmpty {
                field: "display_name".to_string(),
            },
        ]);
        assert!(predicate_implies_predicate_for_planner(
            &complete, &required
        ));

        let missing = Predicate::and(vec![
            non_null("email"),
            compare("active", CompareOp::Eq, Value::Bool(true)),
        ]);
        assert!(!predicate_implies_predicate_for_planner(
            &missing, &required
        ));
    }

    #[test]
    fn nullable_guard_implication_keeps_unsupported_shapes_conservative() {
        let required = non_null("email");
        let cross_field = Predicate::CompareFields(CompareFieldsPredicate::with_coercion(
            "email",
            CompareOp::Eq,
            "backup_email",
            CoercionId::Strict,
        ));
        for query in [
            Predicate::or(vec![required.clone(), non_null("tenant")]),
            Predicate::not(Predicate::IsNull {
                field: "email".to_string(),
            }),
            cross_field,
            compare("email", CompareOp::Eq, Value::Null),
            compare("email", CompareOp::Eq, Value::Unit),
            compare(
                "lower_email",
                CompareOp::Eq,
                Value::Text("a@example.com".to_string()),
            ),
        ] {
            assert!(!predicate_implies_predicate_for_planner(&query, &required));
        }
    }

    #[test]
    fn nullable_guard_implication_treats_unsatisfiable_query_as_vacuous_proof() {
        assert!(predicate_implies_predicate_for_planner(
            &Predicate::False,
            &non_null("email"),
        ));
    }

    #[test]
    fn filtered_guard_stripping_removes_only_guaranteed_non_null_clause() {
        let guard = non_null("email");
        assert_eq!(
            strip_query_clauses_satisfied_by_filtered_guard(&guard, &guard),
            None,
        );

        let query = Predicate::and(vec![guard.clone(), non_null("tenant")]);
        assert_eq!(
            strip_query_clauses_satisfied_by_filtered_guard(&query, &guard),
            Some(non_null("tenant")),
        );
    }
}