use super::*;
use crate::{
db::{
data::{CanonicalSlotReader, ScalarSlotValueRef, SlotReader},
index::{
IndexEntryValue, IndexId, IndexKey, IndexKeyKind, IndexState, IndexStore,
RawIndexStoreKey,
},
key_taxonomy::{PrimaryKeyComponent, PrimaryKeyValue},
schema::{
AcceptedFieldKind, AcceptedSchemaMutationError, FieldId, MutationPlan,
PersistedFieldSnapshot, PersistedIndexExpressionOp, PersistedIndexExpressionSnapshot,
PersistedIndexFieldPathSnapshot, PersistedIndexKeyItemSnapshot,
PersistedIndexKeySnapshot, PersistedIndexSnapshot, PersistedSchemaSnapshot,
SchemaFieldDefault, SchemaFieldSlot, SchemaMutationDelta, SchemaMutationRequest,
SchemaRowLayout, SchemaVersion, classify_schema_mutation_delta,
schema_mutation_request_for_snapshots,
},
},
error::InternalError,
model::field::{FieldStorageDecode, LeafCodec, ScalarCodec},
testing::test_memory,
types::EntityTag,
value::Value,
};
use ic_stable_structures::Storable;
use std::{borrow::Cow, collections::BTreeMap};
struct RebuildSlotReader {
values: Vec<Option<Value>>,
}
#[derive(Default)]
struct RecordingStagedStoreWriter {
writes: Vec<(String, RawIndexStoreKey, IndexEntryValue)>,
}
#[derive(Default)]
struct RecordingStagedStoreRollbackWriter {
actions: Vec<(String, RawIndexStoreKey, Option<IndexEntryValue>)>,
}
#[derive(Default)]
struct RecordingStagedStoreReadView {
entries: BTreeMap<(String, RawIndexStoreKey), IndexEntryValue>,
}
impl RecordingStagedStoreReadView {
fn insert(&mut self, store: &str, key: RawIndexStoreKey, entry: IndexEntryValue) {
self.entries.insert((store.to_string(), key), entry);
}
}
impl super::SchemaFieldPathIndexStagedStoreReadView for RecordingStagedStoreReadView {
fn read_staged_entry(&self, store: &str, key: &RawIndexStoreKey) -> Option<IndexEntryValue> {
self.entries.get(&(store.to_string(), key.clone())).cloned()
}
}
impl super::SchemaFieldPathIndexStagedStoreWriter for RecordingStagedStoreWriter {
fn write_staged_entry(&mut self, store: &str, key: &RawIndexStoreKey, entry: &IndexEntryValue) {
self.writes
.push((store.to_string(), key.clone(), entry.clone()));
}
}
impl super::SchemaFieldPathIndexStagedStoreRollbackWriter for RecordingStagedStoreRollbackWriter {
fn restore_staged_entry(
&mut self,
store: &str,
key: &RawIndexStoreKey,
entry: &IndexEntryValue,
) {
self.actions
.push((store.to_string(), key.clone(), Some(entry.clone())));
}
fn remove_staged_entry(&mut self, store: &str, key: &RawIndexStoreKey) {
self.actions.push((store.to_string(), key.clone(), None));
}
}
impl SlotReader for RebuildSlotReader {
fn has(&self, slot: usize) -> bool {
self.values.get(slot).is_some_and(Option::is_some)
}
fn get_bytes(&self, _slot: usize) -> Option<&[u8]> {
panic!("rebuild key test reader should not decode raw bytes")
}
fn get_scalar(&self, _slot: usize) -> Result<Option<ScalarSlotValueRef<'_>>, InternalError> {
panic!("rebuild key test reader should not route through scalar fast paths")
}
fn get_value(&mut self, _slot: usize) -> Result<Option<Value>, InternalError> {
panic!("rebuild key test reader should not route through generated get_value")
}
}
impl CanonicalSlotReader for RebuildSlotReader {
fn field_name(&self, _slot: usize) -> Result<&str, InternalError> {
Ok("test")
}
fn field_leaf_codec(&self, _slot: usize) -> Result<LeafCodec, InternalError> {
panic!("rebuild key test reader should not decode through field contracts")
}
fn required_value_by_contract(&self, slot: usize) -> Result<Value, InternalError> {
self.values
.get(slot)
.and_then(Option::as_ref)
.cloned()
.ok_or_else(|| InternalError::persisted_row_declared_field_missing("test"))
}
fn required_value_by_contract_cow(&self, slot: usize) -> Result<Cow<'_, Value>, InternalError> {
self.values
.get(slot)
.and_then(Option::as_ref)
.map(Cow::Borrowed)
.ok_or_else(|| InternalError::persisted_row_declared_field_missing("test"))
}
}
fn nullable_text_field(name: &str, id: u32, slot: u16) -> PersistedFieldSnapshot {
PersistedFieldSnapshot::new(
FieldId::new(id),
name.to_string(),
SchemaFieldSlot::new(slot),
AcceptedFieldKind::Text { max_len: None },
Vec::new(),
true,
SchemaFieldDefault::None,
FieldStorageDecode::ByKind,
LeafCodec::Scalar(ScalarCodec::Text),
)
}
fn non_unique_name_index() -> PersistedIndexSnapshot {
PersistedIndexSnapshot::new(
1,
"by_name".to_string(),
"test::mutation::by_name".to_string(),
false,
PersistedIndexKeySnapshot::FieldPath(vec![PersistedIndexFieldPathSnapshot::new(
FieldId::new(2),
SchemaFieldSlot::new(1),
vec!["name".to_string()],
AcceptedFieldKind::Text { max_len: None },
false,
)]),
Some("name IS NOT NULL".to_string()),
)
}
fn unique_name_rebuild_target() -> SchemaFieldPathIndexRebuildTarget {
SchemaFieldPathIndexRebuildTarget {
ordinal: 3,
name: "uniq_name".to_string(),
store: "test::mutation::uniq_name".to_string(),
unique: true,
predicate_sql: Some("name IS NOT NULL".to_string()),
key_paths: vec![SchemaFieldPathIndexRebuildKey {
field_id: FieldId::new(2),
slot: SchemaFieldSlot::new(1),
path: vec!["name".to_string()],
kind: AcceptedFieldKind::Text { max_len: None },
nullable: false,
}],
}
}
fn name_key_path() -> PersistedIndexFieldPathSnapshot {
PersistedIndexFieldPathSnapshot::new(
FieldId::new(2),
SchemaFieldSlot::new(1),
vec!["name".to_string()],
AcceptedFieldKind::Text { max_len: None },
false,
)
}
fn expression_name_index() -> PersistedIndexSnapshot {
PersistedIndexSnapshot::new(
2,
"by_lower_name".to_string(),
"test::mutation::by_lower_name".to_string(),
false,
PersistedIndexKeySnapshot::Items(vec![PersistedIndexKeyItemSnapshot::Expression(
Box::new(PersistedIndexExpressionSnapshot::new(
PersistedIndexExpressionOp::Lower,
name_key_path(),
AcceptedFieldKind::Text { max_len: None },
AcceptedFieldKind::Text { max_len: None },
"expr:v1:LOWER(name)".to_string(),
)),
)]),
Some("LOWER(name) IS NOT NULL".to_string()),
)
}
fn accepted_name_field_path_target() -> super::SchemaFieldPathIndexRebuildTarget {
let request = SchemaMutationRequest::from_accepted_field_path_index(&non_unique_name_index())
.expect("non-unique field-path index should lower to a rebuild target");
let SchemaMutationRequest::AddFieldPathIndex { target } = request else {
panic!("field-path index request should preserve rebuild target");
};
target
}
fn accepted_lower_name_expression_target() -> super::SchemaExpressionIndexRebuildTarget {
let request = SchemaMutationRequest::from_accepted_expression_index(&expression_name_index())
.expect("accepted expression index should lower to a rebuild target");
let SchemaMutationRequest::AddExpressionIndex { target } = request else {
panic!("expression index request should preserve rebuild target");
};
target
}
fn unique_lower_name_expression_target() -> super::SchemaExpressionIndexRebuildTarget {
let unique = PersistedIndexSnapshot::new(
3,
"unique_lower_name".to_string(),
"test::mutation::unique_lower_name".to_string(),
true,
expression_name_index().key().clone(),
Some("LOWER(name) IS NOT NULL".to_string()),
);
let request = SchemaMutationRequest::from_accepted_expression_index(&unique)
.expect("unique expression index should lower to a rebuild target");
let SchemaMutationRequest::AddExpressionIndex { target } = request else {
panic!("unique expression index request should preserve rebuild target");
};
target
}
fn staged_name_index_store() -> super::SchemaFieldPathIndexStagedStore {
let first = RebuildSlotReader {
values: vec![None, Some(Value::Text("Ada".to_string()))],
};
let second = RebuildSlotReader {
values: vec![None, Some(Value::Text("Grace".to_string()))],
};
let staged = super::SchemaFieldPathIndexStagedRebuild::from_rows(
"test::mutation::entity",
EntityTag::new(7),
accepted_name_field_path_target(),
None,
[
super::SchemaFieldPathIndexRebuildRow::new(PrimaryKeyComponent::Nat64(2), &second),
super::SchemaFieldPathIndexRebuildRow::new(PrimaryKeyComponent::Nat64(1), &first),
],
)
.expect("field-path rebuild rows should stage into raw index entries");
super::SchemaFieldPathIndexStagedStore::from_rebuild(&staged)
.expect("valid staged rebuild should write into an in-memory staged store buffer")
}
fn extra_staged_name_index_entry() -> super::SchemaFieldPathIndexStagedEntry {
let extra = RebuildSlotReader {
values: vec![None, Some(Value::Text("Margaret".to_string()))],
};
let staged = super::SchemaFieldPathIndexStagedRebuild::from_rows(
"test::mutation::entity",
EntityTag::new(7),
accepted_name_field_path_target(),
None,
[super::SchemaFieldPathIndexRebuildRow::new(
PrimaryKeyComponent::Nat64(3),
&extra,
)],
)
.expect("extra field-path rebuild row should stage into a raw index entry");
staged.entries()[0].clone()
}
fn initialized_index_store(memory_id: u8) -> IndexStore {
let mut store = IndexStore::init_journaled(test_memory(memory_id));
store.clear();
store
}
fn field_path_index_runner_context() -> (
PersistedSchemaSnapshot,
PersistedSchemaSnapshot,
MutationPlan,
) {
let before = base_snapshot();
let after = snapshot_with_indexes(&before, vec![non_unique_name_index()]);
let plan: MutationPlan =
SchemaMutationRequest::from_accepted_field_path_index(&non_unique_name_index())
.expect("non-unique field-path index should lower")
.into();
(before, after, plan)
}
fn base_snapshot() -> PersistedSchemaSnapshot {
PersistedSchemaSnapshot::new(
SchemaVersion::initial(),
"test::MutationEntity".to_string(),
"MutationEntity".to_string(),
FieldId::new(1),
SchemaRowLayout::new(
SchemaVersion::initial(),
vec![
(FieldId::new(1), SchemaFieldSlot::new(0)),
(FieldId::new(2), SchemaFieldSlot::new(1)),
],
),
vec![
PersistedFieldSnapshot::new(
FieldId::new(1),
"id".to_string(),
SchemaFieldSlot::new(0),
AcceptedFieldKind::Ulid,
Vec::new(),
false,
SchemaFieldDefault::None,
FieldStorageDecode::ByKind,
LeafCodec::Scalar(ScalarCodec::Ulid),
),
PersistedFieldSnapshot::new(
FieldId::new(2),
"name".to_string(),
SchemaFieldSlot::new(1),
AcceptedFieldKind::Text { max_len: None },
Vec::new(),
false,
SchemaFieldDefault::None,
FieldStorageDecode::ByKind,
LeafCodec::Scalar(ScalarCodec::Text),
),
],
)
}
fn append_fields_snapshot(
snapshot: &PersistedSchemaSnapshot,
fields: &[PersistedFieldSnapshot],
) -> PersistedSchemaSnapshot {
let mut next_fields = snapshot.fields().to_vec();
next_fields.extend_from_slice(fields);
let mut next_layout_entries = snapshot.row_layout().field_to_slot().to_vec();
next_layout_entries.extend(fields.iter().map(|field| (field.id(), field.slot())));
PersistedSchemaSnapshot::new(
SchemaVersion::new(snapshot.version().get() + 1),
snapshot.entity_path().to_string(),
snapshot.entity_name().to_string(),
snapshot.first_primary_key_field_id(),
SchemaRowLayout::new(
SchemaVersion::new(snapshot.row_layout().version().get() + 1),
next_layout_entries,
),
next_fields,
)
}
fn snapshot_with_indexes(
snapshot: &PersistedSchemaSnapshot,
indexes: Vec<PersistedIndexSnapshot>,
) -> PersistedSchemaSnapshot {
PersistedSchemaSnapshot::new_with_indexes(
SchemaVersion::new(snapshot.version().get() + 1),
snapshot.entity_path().to_string(),
snapshot.entity_name().to_string(),
snapshot.first_primary_key_field_id(),
SchemaRowLayout::new(
SchemaVersion::new(snapshot.row_layout().version().get() + 1),
snapshot.row_layout().field_to_slot().to_vec(),
),
snapshot.fields().to_vec(),
indexes,
)
}
mod expression_staging;
mod field_path_runner;
mod field_path_staging;
mod field_path_store;
mod planning;