icydb-core 0.213.37

//! Module: access::validate
//! Responsibility: schema-aware access-plan shape and key compatibility validation.
//! Does not own: access-path lowering or runtime scan semantics.
//! Boundary: validation boundary before lowering/execution.

use crate::{
    db::{
        access::{
            AccessPath, AccessPlan, MAX_INDEX_BRANCH_SET_VALUES, SemanticIndexAccessContract,
            SemanticIndexRangeSpec,
        },
        schema::{SchemaInfo, literal_matches_type},
    },
    error::InternalError,
    value::Value,
};
use std::ops::Bound;

///
/// AccessPlanError
///
/// Access-path and key-shape validation failures.
///

#[derive(Debug)]
pub enum AccessPlanError {
    /// Access plan references an index not declared on the entity.
    IndexNotFound { index: String },

    /// Index prefix exceeds the number of indexed fields.
    IndexPrefixTooLong { prefix_len: usize, field_len: usize },

    /// Index prefix must include at least one value.
    IndexPrefixEmpty,

    /// Index branch set exceeds the conservative route cap.
    IndexBranchSetTooLarge { branch_count: usize, max: usize },

    /// Index branch set values must already be canonicalized.
    IndexBranchSetNotCanonical,

    /// Index prefix literal does not match indexed field type.
    IndexPrefixValueMismatch { field: String },

    /// Primary key field exists but is not key-compatible.
    PrimaryKeyNotKeyable { field: String },

    /// Supplied key does not match the primary key type.
    PrimaryKeyMismatch { field: String, key: Value },

    /// Key range has invalid ordering.
    InvalidKeyRange,

    /// Primary-key range scans are currently scalar-primary-key only.
    CompositePrimaryKeyRangeUnsupported { fields: String },
}

impl AccessPlanError {
    /// Map access-validation failures into query-boundary runtime invariants.
    pub(crate) fn into_internal_error(self) -> InternalError {
        let _ = self;
        InternalError::query_executor_invariant()
    }
}

/// Validate executor-owned access invariants against schema-index facts without
/// reopening generated entity model authority.
pub(in crate::db) fn validate_access_runtime_invariants_with_schema(
    schema: &SchemaInfo,
    access: &AccessPlan<Value>,
) -> Result<(), AccessPlanError> {
    access.validate_runtime_invariants(schema)
}

fn validate_composite_pk_literal(
    schema: &SchemaInfo,
    primary_key_names: &[String],
    key: &Value,
) -> Result<(), AccessPlanError> {
    let field = composite_primary_key_field_label(primary_key_names);
    let Value::List(values) = key else {
        return Err(AccessPlanError::PrimaryKeyMismatch {
            field,
            key: key.clone(),
        });
    };
    if values.len() != primary_key_names.len() {
        return Err(AccessPlanError::PrimaryKeyMismatch {
            field,
            key: key.clone(),
        });
    }

    for (field, value) in primary_key_names.iter().zip(values) {
        validate_pk_literal_component(schema, field, value, key)?;
    }

    Ok(())
}

fn validate_pk_literal_component(
    schema: &SchemaInfo,
    field: &str,
    value: &Value,
    reported_key: &Value,
) -> Result<(), AccessPlanError> {
    let field_type = schema
        .field(field)
        .ok_or_else(|| AccessPlanError::PrimaryKeyNotKeyable {
            field: field.to_string(),
        })?;

    if !field_type.is_keyable() {
        return Err(AccessPlanError::PrimaryKeyNotKeyable {
            field: field.to_string(),
        });
    }

    if !literal_matches_type(value, field_type) {
        return Err(AccessPlanError::PrimaryKeyMismatch {
            field: field.to_string(),
            key: reported_key.clone(),
        });
    }

    Ok(())
}

fn validate_pk_literal_runtime(schema: &SchemaInfo, key: &Value) -> Result<(), AccessPlanError> {
    let primary_key_names = schema.primary_key_names();
    if primary_key_names.len() > 1 {
        return validate_composite_pk_literal(schema, primary_key_names, key);
    }

    let Some(field) = primary_key_names.first().map(String::as_str) else {
        return Err(AccessPlanError::PrimaryKeyNotKeyable {
            field: "<missing>".to_string(),
        });
    };

    validate_pk_literal_component(schema, field, key, key)
}

fn composite_primary_key_field_label(primary_key_names: &[String]) -> String {
    primary_key_names.join(", ")
}

fn values_are_canonical_set(values: &[Value]) -> bool {
    values
        .windows(2)
        .all(|pair| Value::canonical_cmp(&pair[0], &pair[1]).is_lt())
}

impl AccessPlan<Value> {
    // Validate executor-runtime access invariants through accepted schema facts.
    fn validate_runtime_invariants(&self, schema: &SchemaInfo) -> Result<(), AccessPlanError> {
        match self {
            Self::Path(path) => path.validate_runtime_invariants(schema),
            Self::Union(children) | Self::Intersection(children) => {
                for child in children {
                    child.validate_runtime_invariants(schema)?;
                }

                Ok(())
            }
        }
    }
}

impl AccessPath<Value> {
    // Validate concrete path invariants through accepted schema facts.
    fn validate_runtime_invariants(&self, schema: &SchemaInfo) -> Result<(), AccessPlanError> {
        match self {
            Self::ByKey(key) => validate_pk_literal_runtime(schema, key),
            Self::ByKeys(keys) => {
                for key in keys {
                    validate_pk_literal_runtime(schema, key)?;
                }

                Ok(())
            }
            Self::FullScan => Ok(()),
            Self::KeyRange { start, end } => validate_pk_range_runtime(schema, start, end),
            Self::IndexPrefix { index, values } => {
                validate_index_reference_with_schema(schema, index)?;
                validate_index_prefix_shape(index, values.len())
            }
            Self::IndexMultiLookup { index, values } => {
                validate_index_reference_with_schema(schema, index)?;
                if values.is_empty() {
                    return Err(AccessPlanError::IndexPrefixEmpty);
                }
                validate_index_prefix_shape(index, 1)
            }
            Self::IndexBranchSet { spec } => {
                validate_index_reference_with_schema(schema, spec.index_ref())?;
                if spec.is_empty() {
                    return Err(AccessPlanError::IndexPrefixEmpty);
                }
                if spec.branch_count() > MAX_INDEX_BRANCH_SET_VALUES {
                    return Err(AccessPlanError::IndexBranchSetTooLarge {
                        branch_count: spec.branch_count(),
                        max: MAX_INDEX_BRANCH_SET_VALUES,
                    });
                }
                if !values_are_canonical_set(spec.branch_values()) {
                    return Err(AccessPlanError::IndexBranchSetNotCanonical);
                }
                validate_index_prefix_shape(spec.index_ref(), spec.branch_prefix_len())
            }
            Self::IndexRange { spec } => validate_index_range_runtime(schema, spec),
        }
    }
}

fn validate_pk_range_runtime(
    schema: &SchemaInfo,
    start: &Value,
    end: &Value,
) -> Result<(), AccessPlanError> {
    let primary_key_names = schema.primary_key_names();
    if primary_key_names.len() > 1 {
        return Err(AccessPlanError::CompositePrimaryKeyRangeUnsupported {
            fields: composite_primary_key_field_label(primary_key_names),
        });
    }

    let ordering = Value::canonical_cmp(start, end);
    if ordering == std::cmp::Ordering::Greater {
        return Err(AccessPlanError::InvalidKeyRange);
    }

    Ok(())
}

fn validate_index_reference_with_schema(
    schema: &SchemaInfo,
    index: &SemanticIndexAccessContract,
) -> Result<(), AccessPlanError> {
    if (0..index.key_arity()).all(|slot| {
        index.key_item_at(slot).is_some_and(|key_item| {
            let field = key_item.field();

            schema.field(field).is_some() && schema.field_is_indexed(field)
        })
    }) {
        return Ok(());
    }

    Err(AccessPlanError::IndexNotFound {
        index: index.name().to_string(),
    })
}

fn validate_index_prefix_shape(
    index: &SemanticIndexAccessContract,
    prefix_len: usize,
) -> Result<(), AccessPlanError> {
    if prefix_len == 0 {
        return Err(AccessPlanError::IndexPrefixEmpty);
    }

    let field_len = index.key_arity();
    if prefix_len > field_len {
        return Err(AccessPlanError::IndexPrefixTooLong {
            prefix_len,
            field_len,
        });
    }

    Ok(())
}

fn validate_index_range_runtime(
    schema: &SchemaInfo,
    spec: &SemanticIndexRangeSpec,
) -> Result<(), AccessPlanError> {
    let index = spec.index();
    let prefix = spec.prefix_values();
    let lower = spec.lower();
    let upper = spec.upper();

    validate_index_reference_with_schema(schema, &index)?;

    if prefix.len() >= index.key_arity() {
        return Err(AccessPlanError::IndexPrefixTooLong {
            prefix_len: prefix.len(),
            field_len: index.key_arity().saturating_sub(1),
        });
    }

    let range_slot = prefix.len();
    if spec.field_slots().len() != prefix.len().saturating_add(1) {
        return Err(AccessPlanError::InvalidKeyRange);
    }
    for (expected_slot, actual_slot) in (0..=range_slot).zip(spec.field_slots().iter().copied()) {
        if actual_slot != expected_slot {
            return Err(AccessPlanError::InvalidKeyRange);
        }
    }

    let (
        Bound::Included(lower_value) | Bound::Excluded(lower_value),
        Bound::Included(upper_value) | Bound::Excluded(upper_value),
    ) = (lower, upper)
    else {
        return Ok(());
    };

    if Value::canonical_cmp(lower_value, upper_value) == std::cmp::Ordering::Greater {
        return Err(AccessPlanError::InvalidKeyRange);
    }

    Ok(())
}