icydb-core 0.285.0

//! 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::{
        commit::CommitSchemaFingerprint,
        data::{DecodedDataStoreKey, StructuralRowContract, StructuralSlotReader},
        index::{IndexKey, IndexPlanReadView, IndexReadContract, plan::AcceptedIndexKeyAuthority},
        key_taxonomy::PrimaryKeyValue,
    },
    error::{InternalError, MutationDiagnosticContext},
    types::EntityTag,
};
use std::ops::Bound;

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

#[expect(clippy::too_many_arguments)]
pub(super) fn validate_unique_constraint_structural_impl(
    accepted_schema_fingerprint: CommitSchemaFingerprint,
    mutation: Option<MutationDiagnosticContext>,
    entity_tag: EntityTag,
    read_view: &dyn IndexPlanReadView,
    row_contract: &StructuralRowContract,
    key_authority: AcceptedIndexKeyAuthority<'_>,
    read_contract: IndexReadContract<'_>,
    new_primary_key: Option<&PrimaryKeyValue>,
    new_index_key: Option<&IndexKey>,
) -> Result<(), InternalError> {
    // 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());
    };

    let index_id = key_authority.index_id(entity_tag);
    if new_index_key.index_id() != &index_id {
        return Err(InternalError::index_unique_validation_corruption());
    }
    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(
        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());
    }

    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 move/delete that row or change its predicate membership;
    // use the same accepted predicate and key builder as index mutation so
    // preflight, replay and fold agree on ownership of 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());
    };
    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());
    };
    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());
    }

    Err(key_authority.unique_violation(accepted_schema_fingerprint, mutation, entity_tag))
}

// 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: &AcceptedIndexKeyAuthority<'_>,
) -> 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));
    }
}