use super::FoundLocation;
use alloc::{vec, vec::Vec};
use super::bucket::reduce;
use super::insert::PreparedVacancy;
use super::{
CommitInconsistency, CommitOutcome, DenseKeyIndex, DenseKeyIndexCreationError, EMPTY_ROW,
FindOutcome, IndexInconsistency, PrepareInsertOutcome,
};
#[cfg(feature = "secondary-indexes")]
use super::{PrepareReplaceOutcome, ReplaceOutcome};
use crate::storage::warp_table::IndexKey;
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
struct Key(u64);
impl IndexKey for Key {
fn index_word(self) -> u64 {
self.0
}
}
#[test]
fn lemire_reduction_uses_the_complete_index_word() {
assert_eq!(reduce(0, 8), 0);
assert_eq!(reduce(u64::MAX, 8), 7);
}
#[test]
fn dense_key_index_preserves_capacity_overflow() {
assert!(matches!(
DenseKeyIndex::try_new(usize::MAX),
Err(DenseKeyIndexCreationError::CapacityOverflow { maximum_rows })
if maximum_rows == usize::MAX
));
}
#[test]
fn dense_key_index_resolves_collisions_and_rejects_duplicate_commits() -> Result<(), &'static str> {
let mut index = DenseKeyIndex::try_new(2).map_err(|_| "index storage unavailable")?;
let keys = [Key(1), Key(2)];
let Ok(PrepareInsertOutcome::Vacant(first)) = index.prepare_insert(keys[0], &[]) else {
return Err("first key was not vacant");
};
assert_eq!(
index.commit(first, 0),
Ok(CommitOutcome::Committed { row: 0 })
);
let Ok(PrepareInsertOutcome::Vacant(second)) = index.prepare_insert(keys[1], &keys[..1]) else {
return Err("collided key was not vacant");
};
assert_eq!(
index.commit(second, 1),
Ok(CommitOutcome::Committed { row: 1 })
);
assert_eq!(
index.find(keys[0], &keys),
Ok(FindOutcome::Present {
location: FoundLocation {
bucket: 0,
stored_row: 0,
row: 0
},
key: keys.first().ok_or("first stored key unavailable")?,
})
);
assert_eq!(
index.find(keys[1], &keys),
Ok(FindOutcome::Present {
location: FoundLocation {
bucket: 1,
stored_row: 1,
row: 1
},
key: keys.get(1).ok_or("second stored key unavailable")?,
})
);
assert_eq!(index.find(Key(3), &keys), Ok(FindOutcome::Absent));
assert_eq!(
index.prepare_insert(keys[0], &keys),
Ok(PrepareInsertOutcome::Present { row: 0 })
);
assert_eq!(
index.find(keys[1], &keys[..1]),
Err(IndexInconsistency::KeyUnavailable { row: 1 })
);
assert_eq!(
index.prepare_insert(keys[1], &keys[..1]),
Err(IndexInconsistency::KeyUnavailable { row: 1 })
);
assert_eq!(
index.commit(first, 2),
Err(CommitInconsistency::VacancyOccupied {
vacancy: first,
row: 2,
occupying_row: 0,
})
);
let unavailable = PreparedVacancy { bucket: usize::MAX };
assert_eq!(
index.commit(unavailable, 2),
Err(CommitInconsistency::VacancyUnavailable {
vacancy: unavailable,
row: 2,
})
);
assert_eq!(
index.commit(second, usize::MAX),
Err(CommitInconsistency::RowUnrepresentable {
vacancy: second,
row: usize::MAX,
})
);
if let Ok(empty_row) = usize::try_from(EMPTY_ROW) {
assert_eq!(
index.commit(second, empty_row),
Err(CommitInconsistency::RowReserved {
vacancy: second,
row: empty_row,
})
);
}
Ok(())
}
#[test]
fn dense_key_index_reports_broken_probe_invariants() {
let empty = DenseKeyIndex {
buckets: Vec::new(),
};
assert_eq!(
empty.find(Key(1), &[]),
Err(IndexInconsistency::ProbeLimitReached)
);
assert_eq!(
empty.prepare_insert(Key(1), &[]),
Err(IndexInconsistency::ProbeLimitReached)
);
let full = DenseKeyIndex {
buckets: vec![0, 1, 2, 3, 4, 5, 6, 7],
};
let keys = [
Key(0),
Key(1),
Key(2),
Key(3),
Key(4),
Key(5),
Key(6),
Key(7),
];
assert_eq!(
full.find(Key(8), &keys),
Err(IndexInconsistency::ProbeLimitReached)
);
assert_eq!(
full.prepare_insert(Key(8), &keys),
Err(IndexInconsistency::ProbeLimitReached)
);
assert_eq!(
full.prepare_remove(Key(0), &keys, 0),
Err(IndexInconsistency::ProbeLimitReached)
);
let broken_cluster = DenseKeyIndex {
buckets: vec![
1, EMPTY_ROW, EMPTY_ROW, EMPTY_ROW, EMPTY_ROW, EMPTY_ROW, EMPTY_ROW, 0,
],
};
assert_eq!(
broken_cluster.prepare_remove(Key(u64::MAX), &[Key(u64::MAX)], 0),
Err(IndexInconsistency::KeyUnavailable { row: 1 })
);
assert_eq!(
broken_cluster.prepare_remove(Key(u64::MAX), &[Key(u64::MAX), Key(u64::MAX)], 1,),
Err(IndexInconsistency::StoredRowMismatch {
expected: 1,
actual: 0,
})
);
}
#[test]
#[cfg(feature = "secondary-indexes")]
fn replacement_preserves_the_row_and_reuses_existing_storage() -> Result<(), &'static str> {
let mut index = DenseKeyIndex::try_new(3).map_err(|_| "index allocation failed")?;
let mut keys = [Key(1), Key(9), Key(17)];
for (row, key) in keys.iter().copied().enumerate() {
let Some(existing_keys) = keys.get(..row) else {
return Err("existing key prefix unavailable");
};
let PrepareInsertOutcome::Vacant(vacancy) = index
.prepare_insert(key, existing_keys)
.map_err(|_| "insert preparation failed")?
else {
return Err("key unexpectedly present");
};
let prepared = index
.prepare_commit(vacancy, row)
.map_err(|_| "insert commit preparation failed")?;
let CommitOutcome::Committed { .. } = index.commit_prepared(prepared);
}
assert_eq!(
index.prepare_replace(keys[1], keys[1], &keys),
Ok(PrepareReplaceOutcome::Unchanged { row: 1 })
);
assert_eq!(
index.prepare_replace(Key(33), Key(41), &keys),
Ok(PrepareReplaceOutcome::OriginalAbsent)
);
assert_eq!(
index.prepare_replace(keys[1], keys[2], &keys),
Ok(PrepareReplaceOutcome::ReplacementPresent { row: 2 })
);
let PrepareReplaceOutcome::Prepared(prepared) = index
.prepare_replace(keys[1], Key(u64::MAX), &keys)
.map_err(|_| "replacement preparation failed")?
else {
return Err("replacement was not prepared");
};
assert_eq!(
index.commit_replace(prepared, &keys),
ReplaceOutcome::Replaced { row: 1 }
);
keys[1] = Key(u64::MAX);
assert_eq!(index.find(Key(9), &keys), Ok(FindOutcome::Absent));
assert!(matches!(
index.find(Key(u64::MAX), &keys),
Ok(FindOutcome::Present {
location: FoundLocation { row: 1, .. },
..
})
));
assert!(matches!(
index.find(keys[2], &keys),
Ok(FindOutcome::Present {
location: FoundLocation { row: 2, .. },
..
})
));
Ok(())
}
#[test]
fn primary_index_failures_preserve_every_reason_and_coordinate() {
use super::super::WarpTableInconsistency;
for (reason, expected, description) in [
(
IndexInconsistency::BucketUnavailable { bucket: 7 },
WarpTableInconsistency::PrimaryBucketUnavailable { bucket: 7 },
"BucketUnavailable { bucket: 7 }",
),
(
IndexInconsistency::StoredRowUnrepresentable { stored_row: 11 },
WarpTableInconsistency::PrimaryStoredRowUnrepresentable { stored_row: 11 },
"StoredRowUnrepresentable { stored_row: 11 }",
),
(
IndexInconsistency::KeyUnavailable { row: 13 },
WarpTableInconsistency::IdUnavailable { row: 13 },
"KeyUnavailable { row: 13 }",
),
(
IndexInconsistency::StoredRowMismatch {
expected: 17,
actual: 19,
},
WarpTableInconsistency::PrimaryStoredRowMismatch {
expected: 17,
actual: 19,
},
"StoredRowMismatch { expected: 17, actual: 19 }",
),
(
IndexInconsistency::ProbeLimitReached,
WarpTableInconsistency::PrimaryProbeLimitReached,
"ProbeLimitReached",
),
] {
assert_eq!(alloc::format!("{reason:?}"), description);
assert_eq!(WarpTableInconsistency::from(reason), expected);
}
}
#[test]
fn primary_commit_failures_preserve_the_vacancy_and_rejected_row() {
use super::super::WarpTableInconsistency;
let vacancy = PreparedVacancy { bucket: 7 };
for (reason, expected) in [
(
CommitInconsistency::RowUnrepresentable { vacancy, row: 11 },
WarpTableInconsistency::PrimaryInsertRowUnrepresentable { bucket: 7, row: 11 },
),
(
CommitInconsistency::RowReserved { vacancy, row: 13 },
WarpTableInconsistency::PrimaryInsertRowReserved { bucket: 7, row: 13 },
),
(
CommitInconsistency::VacancyUnavailable { vacancy, row: 17 },
WarpTableInconsistency::PrimaryInsertVacancyUnavailable { bucket: 7, row: 17 },
),
(
CommitInconsistency::VacancyOccupied {
vacancy,
row: 19,
occupying_row: 23,
},
WarpTableInconsistency::PrimaryInsertVacancyOccupied {
bucket: 7,
row: 19,
occupying_row: 23,
},
),
] {
assert_eq!(WarpTableInconsistency::from(reason), expected);
}
}
#[test]
fn unavailable_columns_preserve_both_coordinates_and_the_diagnostic() {
use super::super::{WarpTableInconsistency, WarpTableRowGetError};
let reason = WarpTableRowGetError::ColumnUnavailable { column: 3, row: 29 };
assert_eq!(
alloc::format!("{reason:?}"),
"ColumnUnavailable { column: 3, row: 29 }"
);
assert_eq!(
WarpTableInconsistency::from(reason),
WarpTableInconsistency::ColumnUnavailable { column: 3, row: 29 }
);
}
#[test]
#[should_panic(
expected = "WarpTable invariant violated: PrimaryStoredRowMismatch { expected: 17, actual: 19 }"
)]
fn invariant_failure_reports_the_exact_broken_relationship() {
use super::super::{WarpTableInconsistency, inconsistency::invariant_violation};
invariant_violation(WarpTableInconsistency::PrimaryStoredRowMismatch {
expected: 17,
actual: 19,
});
}
#[test]
fn removal_rejects_a_missing_relocation_key_without_changing_buckets() {
let index = DenseKeyIndex {
buckets: vec![0, EMPTY_ROW, EMPTY_ROW, EMPTY_ROW],
};
for (keys, last_row) in [(&[Key(0)][..], 7), (&[Key(0), Key(1)][..], 1)] {
let before = index.buckets.clone();
assert_eq!(
index.prepare_remove(Key(0), keys, last_row),
Err(IndexInconsistency::KeyUnavailable { row: last_row })
);
assert_eq!(index.buckets, before);
}
}
#[test]
fn wrapped_clusters_preserve_borrowed_keys_and_every_dense_relocation() -> Result<(), &'static str>
{
for count in 1..=8_usize {
for start_at_end in [false, true] {
for remove_from_front in [false, true] {
let mut index = DenseKeyIndex::try_new(count).map_err(|_| "index unavailable")?;
let mut keys = Vec::with_capacity(count);
let bucket_capacity = index.buckets.capacity();
for ordinal in 0..count {
let ordinal = u64::try_from(ordinal).map_err(|_| "ordinal unrepresentable")?;
let word = if start_at_end {
u64::MAX.checked_sub(ordinal).ok_or("word underflow")?
} else {
ordinal
};
let key = Key(word);
let PrepareInsertOutcome::Vacant(vacancy) = index
.prepare_insert(key, &keys)
.map_err(|_| "insert preparation failed")?
else {
return Err("new key already present");
};
assert_eq!(
index.commit(vacancy, keys.len()),
Ok(CommitOutcome::Committed { row: keys.len() })
);
keys.push(key);
}
while let Some(last_row) = keys.len().checked_sub(1) {
for (expected_row, stored_key) in keys.iter().enumerate() {
let FindOutcome::Present { location, key } = index
.find(*stored_key, &keys)
.map_err(|_| "stored key lookup failed")?
else {
return Err("stored key disappeared");
};
assert_eq!(location.row, expected_row);
assert!(core::ptr::eq(key, stored_key));
assert_eq!(
index.prepare_insert(*stored_key, &keys),
Ok(PrepareInsertOutcome::Present { row: expected_row })
);
}
let removed_row = if remove_from_front { 0 } else { last_row };
let removed_key = *keys.get(removed_row).ok_or("removal key absent")?;
let super::PrepareRemoveOutcome::Prepared(removal) = index
.prepare_remove(removed_key, &keys, last_row)
.map_err(|_| "remove preparation failed")?
else {
return Err("removal key disappeared");
};
assert_eq!(removal.row, removed_row);
match removal.relocation {
super::RowRelocation::Stable => assert_eq!(removed_row, last_row),
super::RowRelocation::Relocated { key, from, to, .. } => {
assert_eq!(keys.get(last_row), Some(&key));
assert_eq!((from, to), (last_row, removed_row));
}
}
index.commit_remove(removal, &keys);
assert_eq!(keys.swap_remove(removed_row), removed_key);
assert_eq!(index.find(removed_key, &keys), Ok(FindOutcome::Absent));
assert_eq!(index.buckets.capacity(), bucket_capacity);
}
assert!(index.buckets.iter().all(|row| *row == EMPTY_ROW));
}
}
}
Ok(())
}
proptest::proptest! {
#[test]
fn lookup_and_insertion_agree_with_a_bounded_probe_model(
buckets in proptest::collection::vec(0..6_u32, 0..17),
keys in proptest::collection::vec(proptest::prelude::any::<u64>(), 0..5),
query in proptest::prelude::any::<u64>(),
use_existing in proptest::prelude::any::<bool>(),
) {
use proptest::prelude::*;
let index = DenseKeyIndex {
buckets: buckets.into_iter().map(|row| if row == 5 { EMPTY_ROW } else { row }).collect(),
};
let keys = keys.into_iter().map(Key).collect::<Vec<_>>();
let query = match (use_existing, keys.first()) {
(true, Some(key)) => *key,
(true, None) | (false, Some(_)) | (false, None) => Key(query),
};
let start = reduce(query.index_word(), index.buckets.len());
let mut order = index.buckets.iter().copied().enumerate().collect::<Vec<_>>();
order.rotate_left(start);
let mut expected_find = Err(IndexInconsistency::ProbeLimitReached);
let mut expected_insert = Err(IndexInconsistency::ProbeLimitReached);
for (bucket, stored_row) in order {
if stored_row == EMPTY_ROW {
expected_find = Ok(FindOutcome::Absent);
expected_insert = Ok(PrepareInsertOutcome::Vacant(PreparedVacancy { bucket }));
break;
}
let row = usize::try_from(stored_row).map_err(|_| TestCaseError::fail("model row unrepresentable"))?;
let Some(key) = keys.get(row) else {
expected_find = Err(IndexInconsistency::KeyUnavailable { row });
expected_insert = Err(IndexInconsistency::KeyUnavailable { row });
break;
};
if *key == query {
expected_find = Ok(FindOutcome::Present { location: FoundLocation { bucket, stored_row, row }, key });
expected_insert = Ok(PrepareInsertOutcome::Present { row });
break;
}
}
let before = index.buckets.clone();
prop_assert_eq!(index.find(query, &keys), expected_find);
prop_assert_eq!(index.prepare_insert(query, &keys), expected_insert);
prop_assert_eq!(index.buckets, before);
}
}
#[test]
fn stored_rows_preserve_the_exact_refusal_at_narrow_address_boundaries() {
for value in [0, 1, u32::from(u16::MAX)] {
assert_eq!(
super::find::represent_stored_row::<u16>(value).map(u32::from),
Ok(value)
);
assert_eq!(super::find::represent_stored_row::<u32>(value), Ok(value));
assert_eq!(
super::find::represent_stored_row::<u64>(value),
Ok(u64::from(value))
);
}
for value in [u32::from(u16::MAX).saturating_add(1), u32::MAX] {
assert_eq!(
super::find::represent_stored_row::<u16>(value),
Err(super::IndexInconsistency::StoredRowUnrepresentable { stored_row: value })
);
assert_eq!(super::find::represent_stored_row::<u32>(value), Ok(value));
}
}