icydb-core 0.213.38

//! Module: index::plan::unique
//! Responsibility: preflight unique-constraint validation against the active
//! planner reader view (committed state or preflight overlay).
//! Does not own: commit-op encoding or apply-time writes.
//! Boundary: internal helper for index planning.

use crate::{
    db::{
        data::{DecodedDataStoreKey, StructuralRowContract, StructuralSlotReader},
        index::{
            IndexId, IndexKey, IndexPlanReadView, IndexReadContract,
            plan::{accepted_expression_key_item_label, error::IndexPlanError},
        },
        key_taxonomy::PrimaryKeyValue,
        schema::{SchemaExpressionIndexInfo, SchemaIndexFieldPathInfo, SchemaIndexInfo},
    },
    error::InternalError,
    types::EntityTag,
};
use std::ops::Bound;

enum UniqueKeyAuthority<'a> {
    AcceptedFieldPath(&'a SchemaIndexInfo),
    AcceptedExpression(&'a SchemaExpressionIndexInfo),
}

const fn batch_override_releases_unique_membership(
    batch_overrides_existing: bool,
    final_row_still_matches: bool,
) -> bool {
    batch_overrides_existing && !final_row_still_matches
}

impl UniqueKeyAuthority<'_> {
    const fn index_id(&self, entity_tag: EntityTag) -> IndexId {
        match self {
            Self::AcceptedFieldPath(index) => IndexId::new_with_generation(
                entity_tag,
                index.ordinal(),
                index.physical_generation(),
            ),
            Self::AcceptedExpression(index) => IndexId::new_with_generation(
                entity_tag,
                index.ordinal(),
                index.physical_generation(),
            ),
        }
    }

    fn build_index_key_from_row_slots(
        &self,
        entity_tag: EntityTag,
        primary_key: &PrimaryKeyValue,
        row_fields: &StructuralSlotReader<'_>,
    ) -> Result<Option<IndexKey>, InternalError> {
        match self {
            Self::AcceptedFieldPath(index) => {
                IndexKey::new_from_slots_with_accepted_field_path_index_primary_key_value(
                    entity_tag,
                    primary_key,
                    index,
                    row_fields,
                )
            }
            Self::AcceptedExpression(index) => {
                IndexKey::new_from_slots_with_accepted_expression_index_primary_key_value(
                    entity_tag,
                    primary_key,
                    index,
                    row_fields,
                )
            }
        }
    }

    fn unique_violation(
        &self,
        entity_path: &str,
        entity_tag: EntityTag,
        primary_key: &PrimaryKeyValue,
    ) -> Result<IndexPlanError, InternalError> {
        let (identity, fields) = match self {
            Self::AcceptedFieldPath(index) => (
                index.unique_constraint(),
                index
                    .fields()
                    .iter()
                    .map(SchemaIndexFieldPathInfo::field_name)
                    .map(str::to_string)
                    .collect::<Vec<_>>(),
            ),
            Self::AcceptedExpression(index) => (
                index.unique_constraint(),
                index
                    .key_items()
                    .iter()
                    .map(accepted_expression_key_item_label)
                    .collect::<Vec<_>>(),
            ),
        };
        let identity = identity.ok_or_else(InternalError::index_unique_validation_corruption)?;
        let data_key = DecodedDataStoreKey::new_primary_key_value(entity_tag, primary_key);
        let raw_data_key = data_key.to_raw()?;
        let primary_key = raw_data_key
            .encoded_primary_key_bytes()
            .ok_or_else(InternalError::index_unique_validation_corruption)?
            .to_vec();
        Ok(IndexPlanError::unique_violation(
            identity.id().get(),
            identity.name(),
            entity_path,
            primary_key,
            fields,
        ))
    }
}

/// Validate one accepted field-path unique index constraint.
#[expect(clippy::too_many_arguments)]
pub(super) fn validate_unique_constraint_accepted_field_path_structural(
    entity_path: &str,
    entity_tag: EntityTag,
    read_view: &dyn IndexPlanReadView,
    row_contract: &StructuralRowContract,
    accepted_index: &SchemaIndexInfo,
    read_contract: IndexReadContract<'_>,
    new_primary_key: Option<&PrimaryKeyValue>,
    new_index_key: Option<&IndexKey>,
) -> Result<(), IndexPlanError> {
    validate_unique_constraint_structural_impl(
        entity_path,
        entity_tag,
        read_view,
        row_contract,
        UniqueKeyAuthority::AcceptedFieldPath(accepted_index),
        read_contract,
        new_primary_key,
        new_index_key,
    )
}

/// Validate one accepted expression unique index constraint.
#[expect(clippy::too_many_arguments)]
pub(super) fn validate_unique_constraint_accepted_expression_structural(
    entity_path: &str,
    entity_tag: EntityTag,
    read_view: &dyn IndexPlanReadView,
    row_contract: &StructuralRowContract,
    accepted_index: &SchemaExpressionIndexInfo,
    read_contract: IndexReadContract<'_>,
    new_primary_key: Option<&PrimaryKeyValue>,
    new_index_key: Option<&IndexKey>,
) -> Result<(), IndexPlanError> {
    validate_unique_constraint_structural_impl(
        entity_path,
        entity_tag,
        read_view,
        row_contract,
        UniqueKeyAuthority::AcceptedExpression(accepted_index),
        read_contract,
        new_primary_key,
        new_index_key,
    )
}

#[expect(clippy::too_many_arguments)]
fn validate_unique_constraint_structural_impl(
    entity_path: &str,
    entity_tag: EntityTag,
    read_view: &dyn IndexPlanReadView,
    row_contract: &StructuralRowContract,
    key_authority: UniqueKeyAuthority<'_>,
    read_contract: IndexReadContract<'_>,
    new_primary_key: Option<&PrimaryKeyValue>,
    new_index_key: Option<&IndexKey>,
) -> Result<(), IndexPlanError> {
    // Phase 1: fast exits for non-unique or non-insert/update paths.
    if !read_contract.unique() {
        return Ok(());
    }

    let Some(new_index_key) = new_index_key else {
        // Delete/no-op paths do not need unique validation.
        return Ok(());
    };

    let Some(new_primary_key) = new_primary_key else {
        return Err(InternalError::index_unique_validation_entity_key_required().into());
    };

    let index_id = key_authority.index_id(entity_tag);
    if new_index_key.index_id() != &index_id {
        return Err(InternalError::index_unique_validation_corruption().into());
    }
    let (lower, upper) = new_index_key
        .raw_bounds_for_all_components()
        .map_err(InternalError::from)?;
    let lower = Bound::Included(lower);
    let upper = Bound::Included(upper);

    // Unique validation only needs to distinguish 0, 1, or "more than 1".
    // Capping this probe avoids scanning large corrupted buckets.
    let unique_probe_limit = 2usize;
    let matching_primary_keys = read_view.read_index_keys_in_raw_range(
        entity_path,
        entity_tag,
        read_contract,
        (&lower, &upper),
        unique_probe_limit,
    )?;

    if matching_primary_keys.is_empty() {
        return Ok(());
    }

    if matching_primary_keys.len() > 1 {
        return Err(InternalError::index_unique_validation_corruption().into());
    }

    let existing_key = matching_primary_keys[0];
    if &existing_key == new_primary_key {
        return Ok(());
    }

    // Phase 3: prove that the stored row still belongs to this key and value
    // through the structural persisted-row decode path only. A complete batch
    // overlay may intentionally move or delete that row; in that case the
    // committed index membership does not conflict with the final batch.
    let data_key = DecodedDataStoreKey::new_primary_key_value(entity_tag, &existing_key);
    let batch_overrides_existing = read_view.has_primary_row_override(&data_key)?;
    let Some(row) = read_view.read_primary_row(&data_key)? else {
        if batch_override_releases_unique_membership(batch_overrides_existing, false) {
            return Ok(());
        }
        return Err(InternalError::index_unique_validation_row_required().into());
    };
    let row_fields = decode_unique_row_slots(&data_key, &row, row_contract)?;

    let Some(current_index_key) = build_unique_index_key_from_row_slots(
        entity_tag,
        &existing_key,
        &row_fields,
        &key_authority,
    )?
    else {
        if batch_override_releases_unique_membership(batch_overrides_existing, false) {
            return Ok(());
        }
        return Err(InternalError::index_unique_validation_corruption().into());
    };
    if !current_index_key.has_same_components(new_index_key) {
        if batch_override_releases_unique_membership(batch_overrides_existing, false) {
            return Ok(());
        }
        return Err(InternalError::index_unique_validation_corruption().into());
    }

    Err(key_authority.unique_violation(entity_path, entity_tag, new_primary_key)?)
}

// Decode one stored row through the canonical structural persisted-row scanner
// and validate its authoritative primary-key slot for unique validation.
fn decode_unique_row_slots<'a>(
    data_key: &DecodedDataStoreKey,
    row: &'a crate::db::data::RawRow,
    row_contract: &StructuralRowContract,
) -> Result<StructuralSlotReader<'a>, InternalError> {
    let row_fields =
        StructuralSlotReader::from_raw_row_with_validated_contract(row, row_contract.clone())
            .map_err(|_| InternalError::index_unique_validation_row_deserialize_failed())?;
    row_fields
        .validate_primary_key(data_key)
        .map_err(|_| InternalError::index_unique_validation_primary_key_decode_failed())?;

    Ok(row_fields)
}

// Build the canonical stored unique index key from one structural row slot
// reader without reconstructing the full typed entity.
fn build_unique_index_key_from_row_slots(
    entity_tag: EntityTag,
    primary_key: &PrimaryKeyValue,
    row_fields: &StructuralSlotReader<'_>,
    key_authority: &UniqueKeyAuthority<'_>,
) -> Result<Option<IndexKey>, InternalError> {
    let key = key_authority.build_index_key_from_row_slots(entity_tag, primary_key, row_fields);

    key.map_err(|_| InternalError::index_unique_validation_key_rebuild_failed())
}

#[cfg(test)]
mod tests {
    use super::batch_override_releases_unique_membership;

    #[test]
    fn complete_batch_override_releases_moved_or_deleted_unique_membership() {
        assert!(batch_override_releases_unique_membership(true, false));
    }

    #[test]
    fn complete_batch_override_preserves_final_unique_collision() {
        assert!(!batch_override_releases_unique_membership(true, true));
    }

    #[test]
    fn committed_unique_membership_mismatch_is_not_silently_released() {
        assert!(!batch_override_releases_unique_membership(false, false));
    }
}