use alloc::{format, vec::Vec};
use core::num::NonZeroUsize;
use proptest::{prelude::*, test_runner::TestCaseError};
use super::{
RangeIndexOrderingDirection, RangeIndexOrderingError, RangeIndexPlanes, RangeIndexRanking,
RangeIndexRankingCandidates, RankedSlots, rank,
};
use crate::storage::warp_table::secondary_index::{
RangeIndex, RangeIndexInconsistency, RangeIndexInsertOutcome, RangeIndexRemoveOutcome,
};
use crate::storage::warp_table::secondary_index::{SlotSet, SlotSetInsertOutcome, StableSlot};
#[derive(Debug, Eq, PartialEq)]
enum Ranked {
Exact(Vec<u32>),
Straddling {
leaders: Vec<u32>,
tied: Vec<u32>,
needed: usize,
},
}
#[test]
fn candidates_within_the_limit_are_exact_winners() -> Result<(), RangeIndexOrderingError> {
let scores = populated_index(&[4, 9, 4, 1])?;
let level = scores.ordering_level("score", RangeIndexOrderingDirection::LargestFirst);
assert_eq!(
ranked(rank(&level, &[], all_members(4), rows(4),)?),
Ranked::Exact(Vec::from([0, 1, 2, 3]))
);
Ok(())
}
#[test]
fn borrowed_refinement_preserves_later_index_failures() -> Result<(), RangeIndexOrderingError> {
let scores = populated_index(&[7, 7, 7])?;
let first = scores.ordering_level("score", RangeIndexOrderingDirection::SmallestFirst);
let expected = inconsistent(RangeIndexInconsistency::MemberAbsent);
let later = super::RangeIndexOrderingLevel {
direction: RangeIndexOrderingDirection::SmallestFirst,
planes: Err(expected),
};
let mut bytes = [core::mem::MaybeUninit::uninit(); 512];
assert!(
matches!(super::rank_with(&first, &[later], all_members(3), rows(1), crate::storage::warp_table::StorageLease::new(&mut bytes)),
Err(super::RankingWorkspaceError::Ordering(error)) if error == expected)
);
Ok(())
}
#[test]
fn leaders_filling_the_limit_are_exact_winners() -> Result<(), RangeIndexOrderingError> {
let scores = populated_index(&[4, 9, 4, 1, 7])?;
let largest = scores.ordering_level("score", RangeIndexOrderingDirection::LargestFirst);
let smallest = scores.ordering_level("score", RangeIndexOrderingDirection::SmallestFirst);
assert_eq!(
ranked(rank(&largest, &[], all_members(5), rows(2),)?),
Ranked::Exact(Vec::from([1, 4]))
);
assert_eq!(
ranked(rank(&smallest, &[], all_members(5), rows(3),)?),
Ranked::Exact(Vec::from([0, 2, 3]))
);
Ok(())
}
#[test]
fn ties_crossing_the_limit_straddle_it() -> Result<(), RangeIndexOrderingError> {
let scores = populated_index(&[4, 9, 4, 1, 4])?;
let levels = populated_index(&[2, 0, 3, 0, 3])?;
let largest_score = scores.ordering_level("score", RangeIndexOrderingDirection::LargestFirst);
assert_eq!(
ranked(rank(&largest_score, &[], all_members(5), rows(2),)?),
Ranked::Straddling {
leaders: Vec::from([1]),
tied: Vec::from([0, 2, 4]),
needed: 1,
}
);
assert_eq!(
ranked(rank(
&largest_score,
&[levels.ordering_level("level", RangeIndexOrderingDirection::LargestFirst)],
all_members(5),
rows(2),
)?),
Ranked::Straddling {
leaders: Vec::from([1]),
tied: Vec::from([2, 4]),
needed: 1,
}
);
assert_eq!(
ranked(rank(
&largest_score,
&[levels.ordering_level("level", RangeIndexOrderingDirection::SmallestFirst)],
all_members(5),
rows(2),
)?),
Ranked::Exact(Vec::from([0, 1]))
);
Ok(())
}
#[test]
fn every_member_skips_vacated_rows_and_empty_blocks() -> Result<(), RangeIndexOrderingError> {
let mut scores = populated_index(&[5; 130])?;
for slot in 0..128 {
let RangeIndexRemoveOutcome::Removed { .. } = scores
.try_prepare_remove(5, StableSlot::new(slot))
.map_err(|_| inconsistent(RangeIndexInconsistency::MemberAbsent))?
.commit();
}
let level = scores.ordering_level("score", RangeIndexOrderingDirection::LargestFirst);
assert_eq!(
ranked(rank(&level, &[], all_members(2), rows(1),)?),
Ranked::Straddling {
leaders: Vec::new(),
tied: Vec::from([128, 129]),
needed: 1,
}
);
Ok(())
}
#[test]
fn every_member_requires_the_table_and_index_to_cover_the_same_rows()
-> Result<(), RangeIndexOrderingError> {
let mut scores = populated_index(&[5, 6, 7])?;
let RangeIndexRemoveOutcome::Removed { .. } = scores
.try_prepare_remove(6, StableSlot::new(1))
.map_err(|_| inconsistent(RangeIndexInconsistency::MemberAbsent))?
.commit();
let level = scores.ordering_level("score", RangeIndexOrderingDirection::LargestFirst);
assert!(matches!(
rank(&level, &[], all_members(3), rows(1)),
Err(RangeIndexOrderingError::MemberCountMismatch {
field: "score",
table_rows: 3,
index_members: 2,
})
));
Ok(())
}
#[test]
fn corrupt_level_storage_is_reported_with_its_field() -> Result<(), RangeIndexOrderingError> {
let mut scores = populated_index(&[4, 9])?;
let _removed = scores.blocks.pop();
let level = scores.ordering_level("score", RangeIndexOrderingDirection::LargestFirst);
let corrupt = inconsistent(RangeIndexInconsistency::BlockStorageLengthMismatch {
expected: 9,
actual: 8,
});
assert!(matches!(
rank(&level, &[], all_members(2), rows(1)),
Err(failure) if failure == corrupt
));
assert_eq!(level.sortable_word_at(StableSlot::new(0)), Err(corrupt));
Ok(())
}
#[test]
fn candidates_outside_the_level_are_reported() -> Result<(), RangeIndexOrderingError> {
let mut scores = populated_index(&[4, 9, 4])?;
let RangeIndexRemoveOutcome::Removed { .. } = scores
.try_prepare_remove(9, StableSlot::new(1))
.map_err(|_| inconsistent(RangeIndexInconsistency::MemberAbsent))?
.commit();
let level = scores.ordering_level("score", RangeIndexOrderingDirection::LargestFirst);
let outside = inconsistent(RangeIndexInconsistency::SlotOutsideCapacity { maximum_rows: 3 });
let absent = inconsistent(RangeIndexInconsistency::MemberAbsent);
assert!(matches!(
rank(
&level,
&[],
RangeIndexRankingCandidates::Slots(slots(&[64]).read()),
rows(1)
),
Err(failure) if failure == outside
));
assert!(matches!(
rank(
&level,
&[],
RangeIndexRankingCandidates::Slots(slots(&[0, 1]).read()),
rows(1)
),
Err(failure) if failure == absent
));
assert!(matches!(
rank(
&level,
&[],
RangeIndexRankingCandidates::Slots(slots(&[5]).read()),
rows(1)
),
Err(failure) if failure == absent
));
assert_eq!(level.sortable_word_at(StableSlot::new(1)), Err(absent));
assert_eq!(level.sortable_word_at(StableSlot::new(64)), Err(outside));
Ok(())
}
#[test]
fn sortable_words_reverse_only_largest_first_levels() -> Result<(), RangeIndexOrderingError> {
let scores = populated_index(&[0b1010_0110])?;
let slot = StableSlot::new(0);
assert_eq!(
scores
.ordering_level("score", RangeIndexOrderingDirection::SmallestFirst)
.sortable_word_at(slot),
Ok(0b1010_0110)
);
assert_eq!(
scores
.ordering_level("score", RangeIndexOrderingDirection::LargestFirst)
.sortable_word_at(slot),
Ok(!0b1010_0110)
);
Ok(())
}
#[test]
fn unaddressable_planes_and_blocks_are_reported() {
let planes = RangeIndexPlanes {
field: "score",
maximum_rows: 1,
words_per_block: 9,
blocks: &[0; 9],
};
assert_eq!(
planes.plane_at(&[1, 2], 2),
Err(inconsistent(
RangeIndexInconsistency::BitPositionOutsideWord { bit: 2 }
))
);
assert_eq!(
planes.slot_base(usize::MAX),
Err(inconsistent(RangeIndexInconsistency::SlotOutsideCapacity {
maximum_rows: 1
}))
);
assert_eq!(planes.slot_base(2), Ok(128));
}
proptest! {
#![proptest_config(ProptestConfig::with_cases(256))]
#[test]
fn bit_sliced_ranking_matches_a_sorted_model(
rows_in_table in prop_oneof![
prop::collection::vec((0..12_u8, 0..4_u8, any::<bool>()), 1..160),
prop::collection::vec((0..12_u8, 0..4_u8, any::<bool>()), 1023..2114),
prop::collection::vec((any::<u8>(), any::<u8>(), any::<bool>()), 1..2114),
],
every_member in any::<bool>(),
second_level in any::<bool>(),
first_largest in any::<bool>(),
second_largest in any::<bool>(),
limit in 1..170_usize,
) {
let first_values = rows_in_table.iter().map(|(first, _, _)| *first).collect::<Vec<_>>();
let second_values = rows_in_table.iter().map(|(_, second, _)| *second).collect::<Vec<_>>();
let candidate_slots = rows_in_table
.iter()
.enumerate()
.filter(|(slot, (_, _, candidate))| every_member || *candidate || *slot == 0)
.map(|(slot, _)| u32::try_from(slot))
.collect::<Result<Vec<_>, _>>()
.map_err(|_| TestCaseError::fail("slot exceeded u32"))?;
let first_index = populated_index(&first_values)
.map_err(|_| TestCaseError::fail("first index was unavailable"))?;
let second_index = populated_index(&second_values)
.map_err(|_| TestCaseError::fail("second index was unavailable"))?;
let first_direction = direction(first_largest);
let second_direction = direction(second_largest);
let first_level = first_index.ordering_level("first", first_direction);
let later_levels = if second_level {
Vec::from([second_index.ordering_level("second", second_direction)])
} else {
Vec::new()
};
let candidate_set = slots(&candidate_slots);
let candidates = if every_member {
all_members(rows_in_table.len())
} else {
RangeIndexRankingCandidates::Slots(candidate_set.read())
};
let limit = NonZeroUsize::new(limit).ok_or(TestCaseError::fail("limit was zero"))?;
let actual = rank(&first_level, &later_levels, candidates, limit)
.map_err(|_| TestCaseError::fail("ranking was inconsistent"))?;
let model_key = |slot: u32| {
let index = slot as usize;
let first = first_values.get(index).copied().map(u64::from);
let second = second_values.get(index).copied().map(u64::from);
(
first.map(|word| first_direction.sortable_word(word)),
second.filter(|_| second_level).map(|word| second_direction.sortable_word(word)),
)
};
let expected = modeled(&candidate_slots, limit.get(), model_key);
prop_assert_eq!(&ranked(actual), &expected);
let bytes = super::ranking_storage_bytes(rows_in_table.len())
.map_err(|_| TestCaseError::fail("ranking storage overflowed"))?;
let mut storage = alloc::vec![core::mem::MaybeUninit::uninit(); bytes];
let candidates = if every_member {
all_members(rows_in_table.len())
} else {
RangeIndexRankingCandidates::Slots(candidate_set.read())
};
let (borrowed, _) = super::rank_with(&first_level, &later_levels, candidates, limit,
crate::storage::warp_table::StorageLease::new(&mut storage))
.map_err(|_| TestCaseError::fail("borrowed ranking failed"))?;
let leaders = borrowed.leaders().map(StableSlot::value).collect::<Vec<_>>();
let tied = borrowed.tied().map(StableSlot::value).collect::<Vec<_>>();
prop_assert_eq!(leaders.len(), borrowed.leader_count());
prop_assert_eq!(tied.len(), borrowed.tied_count());
let actual = match borrowed.needed {
None => Ranked::Exact(leaders),
Some(needed) => Ranked::Straddling { leaders, tied, needed: needed.get() },
};
prop_assert_eq!(actual, expected);
}
}
fn all_members(expected_members: usize) -> RangeIndexRankingCandidates<'static> {
RangeIndexRankingCandidates::EveryMember { expected_members }
}
fn modeled<Key: Ord>(candidates: &[u32], limit: usize, key: impl Fn(u32) -> Key) -> Ranked {
let mut sorted = candidates.to_vec();
sorted.sort_by_key(|slot| key(*slot));
let Some(cutoff) = sorted.get(limit.saturating_sub(1)).map(|slot| key(*slot)) else {
return Ranked::Exact(candidates.to_vec());
};
let leaders = candidates
.iter()
.copied()
.filter(|slot| key(*slot) < cutoff)
.collect::<Vec<_>>();
let tied = candidates
.iter()
.copied()
.filter(|slot| key(*slot) == cutoff)
.collect::<Vec<_>>();
let needed = limit.saturating_sub(leaders.len());
if tied.len() == needed {
let mut winners = [leaders, tied].concat();
winners.sort_unstable();
return Ranked::Exact(winners);
}
Ranked::Straddling {
leaders,
tied,
needed,
}
}
fn ranked(ranking: RangeIndexRanking) -> Ranked {
match ranking {
RangeIndexRanking::Exact { winners } => Ranked::Exact(slot_values(&winners)),
RangeIndexRanking::Straddling { leaders, ties } => Ranked::Straddling {
leaders: slot_values(&leaders),
tied: slot_values(ties.tied()),
needed: ties.needed().get(),
},
}
}
fn slot_values(slots: &RankedSlots) -> Vec<u32> {
let values = slots.slots().map(StableSlot::value).collect::<Vec<_>>();
assert_eq!(values.len(), slots.len());
values
}
const fn direction(largest_first: bool) -> RangeIndexOrderingDirection {
if largest_first {
RangeIndexOrderingDirection::LargestFirst
} else {
RangeIndexOrderingDirection::SmallestFirst
}
}
const fn inconsistent(reason: RangeIndexInconsistency) -> RangeIndexOrderingError {
RangeIndexOrderingError::Inconsistent {
field: "score",
reason,
}
}
fn rows(count: usize) -> NonZeroUsize {
NonZeroUsize::new(count).unwrap_or(NonZeroUsize::MIN)
}
fn slots(values: &[u32]) -> SlotSet {
let mut slots = SlotSet::empty();
for value in values.iter().copied() {
if let Ok(insertion) = slots.try_prepare_insert(StableSlot::new(value)) {
let SlotSetInsertOutcome::Inserted { .. } = insertion.commit();
}
}
slots
}
fn populated_index(values: &[u8]) -> Result<RangeIndex<u8>, RangeIndexOrderingError> {
let unavailable = inconsistent(RangeIndexInconsistency::MemberAbsent);
let mut index = RangeIndex::try_new(values.len()).map_err(|_| unavailable)?;
for (slot, key) in values.iter().copied().enumerate() {
let slot = StableSlot::new(u32::try_from(slot).map_err(|_| unavailable)?);
let RangeIndexInsertOutcome::Inserted { .. } = index
.try_prepare_insert(key, slot)
.map_err(|_| unavailable)?
.commit();
}
Ok(index)
}
proptest! {
#[test]
fn single_extremum_matches_stable_order_without_materializing_a_selection(
values in prop::collection::vec(any::<u8>(), 0..257),
largest in any::<bool>(),
) {
let index = populated_index(&values).map_err(|e| TestCaseError::fail(format!("{e:?}")))?;
let direction = if largest { RangeIndexOrderingDirection::LargestFirst } else { RangeIndexOrderingDirection::SmallestFirst };
let first = index.ordering_level("score", direction).first_member(values.len())
.map_err(|e| TestCaseError::fail(format!("{e:?}")))?;
let expected = values.iter().enumerate().min_by_key(|(slot, value)| {
(if largest { !**value } else { **value }, *slot)
}).map(|(slot, _)| slot);
prop_assert_eq!(first.map(|slot| slot.value() as usize), expected);
}
}
#[test]
fn single_extremum_preserves_empty_blocks_ties_and_exact_inconsistency()
-> Result<(), RangeIndexOrderingError> {
let mut index = populated_index(&[7_u8; 130])?;
for slot in 0..64 {
assert!(matches!(
index
.try_prepare_remove(7, StableSlot::new(slot))
.map_err(|_| inconsistent(RangeIndexInconsistency::MemberAbsent))?
.commit(),
RangeIndexRemoveOutcome::Removed { .. }
));
}
let level = index.ordering_level("score", RangeIndexOrderingDirection::SmallestFirst);
assert_eq!(level.first_member(66)?, Some(StableSlot::new(64)));
assert_eq!(
level.first_member(65),
Err(RangeIndexOrderingError::MemberCountMismatch {
field: "score",
table_rows: 65,
index_members: 66,
})
);
let broken = super::RangeIndexOrderingLevel {
direction: RangeIndexOrderingDirection::SmallestFirst,
planes: Err(RangeIndexOrderingError::MemberCountMismatch {
field: "broken",
table_rows: 1,
index_members: 0,
}),
};
assert_eq!(
broken.first_member(1),
Err(RangeIndexOrderingError::MemberCountMismatch {
field: "broken",
table_rows: 1,
index_members: 0,
})
);
Ok(())
}