icydb-core 0.276.0

//! Module: db::schema::publication::user_index_domain
//! Responsibility: stage one SQL-DDL-owned accepted user-index domain.
//! Does not own: index derivation, physical apply, or accepted publication.
//! Boundary: accepted catalog plus stored rows to one zero-write staged domain.

use crate::{
    db::{
        data::{
            AcceptedStructuralRowAuthority, DecodedDataStoreKey, StoreVisit, StructuralRowContract,
            StructuralSlotReader,
        },
        registry::StoreHandle,
        resource::budget::MaintenanceConstructionBudget,
        schema::{
            AcceptedCatalogIdentity, AcceptedSchemaSnapshot, PersistedSchemaSnapshot,
            SchemaUserIndexDomainRow, StagedUserIndexDomainError, StagedUserIndexDomainReplacement,
            StagedUserIndexDomainReplacementBuilder,
        },
    },
    error::InternalError,
};

/// Stage one complete accepted-after user-index domain for SQL index DDL
/// without changing schema or physical index state.
pub(super) fn stage_sql_ddl_user_index_domain_replacement(
    store: StoreHandle,
    accepted_before_identity: &AcceptedCatalogIdentity,
    accepted_before: &PersistedSchemaSnapshot,
    accepted_after: &PersistedSchemaSnapshot,
) -> Result<StagedUserIndexDomainReplacement, InternalError> {
    // Both row contracts use the same accepted catalogue selection. Validate the
    // before snapshot before deriving candidate decode authority from its catalogues.
    let selection = store
        .with_schema(|schema_store| {
            schema_store.current_accepted_catalog_selection(
                accepted_before_identity.entity_tag(),
                accepted_before_identity.entity_path(),
                accepted_before_identity.store_path(),
            )
        })?
        .ok_or_else(InternalError::store_corruption)?;
    let authority = AcceptedStructuralRowAuthority::from_catalog_selection(
        accepted_before_identity.entity_path(),
        &selection,
    )?;
    if authority.accepted_schema().persisted_snapshot() != accepted_before {
        return Err(InternalError::store_unsupported());
    }
    let accepted_before_row_contract = authority.into_row_contract();

    let accepted_after_snapshot = AcceptedSchemaSnapshot::try_new(accepted_after.clone())?;
    let accepted_after_row_contract = AcceptedStructuralRowAuthority::from_candidate_snapshot(
        accepted_before_identity.entity_path(),
        accepted_after_snapshot.into(),
        selection.value_catalog_handle().clone(),
    )?
    .into_row_contract();

    stage_user_index_domain_replacement(
        store,
        accepted_before_identity.clone(),
        accepted_before,
        accepted_after,
        accepted_before_row_contract,
        accepted_after_row_contract,
    )
}

fn stage_user_index_domain_replacement(
    store: StoreHandle,
    accepted_before_identity: AcceptedCatalogIdentity,
    accepted_before: &PersistedSchemaSnapshot,
    accepted_after: &PersistedSchemaSnapshot,
    accepted_before_row_contract: StructuralRowContract,
    accepted_after_row_contract: StructuralRowContract,
) -> Result<StagedUserIndexDomainReplacement, InternalError> {
    let entity_tag = accepted_before_identity.entity_tag();
    let mut builder = store.with_index(|index_store| {
        StagedUserIndexDomainReplacementBuilder::new(
            accepted_before_identity,
            accepted_before,
            accepted_after,
            Some(&accepted_before_row_contract),
            Some(&accepted_after_row_contract),
            index_store,
            MaintenanceConstructionBudget::new(),
        )
        .map_err(StagedUserIndexDomainError::into_internal_error)
    })?;
    store.with_data(|data_store| {
        data_store.visit_entries(|raw_key, raw_row| {
            let data_key = DecodedDataStoreKey::try_from_raw(raw_key)
                .map_err(|_| InternalError::store_corruption())?;
            if data_key.entity_tag() != entity_tag {
                return Ok::<StoreVisit, InternalError>(StoreVisit::Continue);
            }
            let accepted_before_slots =
                StructuralSlotReader::from_raw_row_with_validated_borrowed_contract(
                    raw_row,
                    &accepted_before_row_contract,
                )?;
            accepted_before_slots.validate_primary_key(&data_key)?;
            if accepted_before_row_contract.has_same_decode_authority(&accepted_after_row_contract)
            {
                let row = SchemaUserIndexDomainRow::new(
                    data_key.primary_key_value(),
                    &accepted_before_slots,
                    &accepted_before_slots,
                    raw_row.len(),
                );
                builder
                    .observe_row(&row)
                    .map_err(StagedUserIndexDomainError::into_internal_error)?;
            } else {
                let accepted_after_slots =
                    StructuralSlotReader::from_raw_row_with_validated_borrowed_contract(
                        raw_row,
                        &accepted_after_row_contract,
                    )?;
                accepted_after_slots.validate_primary_key(&data_key)?;
                let row = SchemaUserIndexDomainRow::new(
                    data_key.primary_key_value(),
                    &accepted_before_slots,
                    &accepted_after_slots,
                    raw_row.len(),
                );
                builder
                    .observe_row(&row)
                    .map_err(StagedUserIndexDomainError::into_internal_error)?;
            }
            Ok::<StoreVisit, InternalError>(StoreVisit::Continue)
        })
    })?;
    store.with_index(|index_store| {
        builder
            .finish(index_store)
            .map_err(StagedUserIndexDomainError::into_internal_error)
    })
}