#[cfg(feature = "secondary-indexes")]
use crate::storage::StableSlot;
use crate::storage::range_index::{RangeIndexInconsistency, RangeIndexKey};
mod rank;
mod ranked_slots;
pub(in crate::storage::warp_table) use rank::RangeIndexRankingCandidates;
pub use rank::RankingWorkspaceError;
#[cfg(test)]
use rank::{RangeIndexRanking, rank};
pub(in crate::storage::warp_table) use rank::{WorkspaceRanking, rank_with, ranking_storage_bytes};
#[cfg(test)]
use ranked_slots::RankedSlots;
const SLOTS_PER_BLOCK: u32 = u64::BITS;
const SLOT_OFFSET_MASK: u32 = 0b11_1111;
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum RangeIndexOrderingError {
Membership(crate::storage::query_workspace::QueryWorkspaceError),
Inconsistent {
field: &'static str,
reason: RangeIndexInconsistency,
},
MemberCountMismatch {
field: &'static str,
table_rows: usize,
index_members: usize,
},
}
#[doc(hidden)]
#[derive(Clone, Copy)]
pub enum RangeIndexOrderingDirection {
LargestFirst,
SmallestFirst,
}
impl RangeIndexOrderingDirection {
const fn preferred_rows(self, plane: u64) -> u64 {
match self {
Self::LargestFirst => plane,
Self::SmallestFirst => !plane,
}
}
#[cfg(test)]
const fn sortable_word(self, ordered_word: u64) -> u64 {
match self {
Self::LargestFirst => !ordered_word,
Self::SmallestFirst => ordered_word,
}
}
}
#[doc(hidden)]
pub struct RangeIndexOrderingLevel<'index> {
direction: RangeIndexOrderingDirection,
planes: Result<RangeIndexPlanes<'index>, RangeIndexOrderingError>,
}
impl<Key: RangeIndexKey, Storage: core::ops::DerefMut<Target = [u64]>>
crate::storage::range_index::RangeIndex<Key, Storage>
{
pub fn ordering_level(
&self,
field: &'static str,
direction: RangeIndexOrderingDirection,
) -> RangeIndexOrderingLevel<'_> {
RangeIndexOrderingLevel {
direction,
planes: self.planes(field),
}
}
fn planes(&self, field: &'static str) -> Result<RangeIndexPlanes<'_>, RangeIndexOrderingError> {
let words_per_block = self
.validated_words_per_block()
.map_err(|reason| RangeIndexOrderingError::Inconsistent { field, reason })?;
Ok(RangeIndexPlanes {
field,
maximum_rows: self.maximum_rows,
words_per_block,
blocks: &self.blocks,
})
}
}
impl<'index> RangeIndexOrderingLevel<'index> {
#[cfg(feature = "secondary-indexes")]
pub(in crate::storage::warp_table) fn first_member(
&self,
expected_members: usize,
) -> Result<Option<StableSlot>, RangeIndexOrderingError> {
let planes = self.planes()?;
let mut total = 0_usize;
let mut best: Option<StableSlot> = None;
let mut best_word = u64::MAX;
let selective_bound = 1_u64.wrapping_shl(planes.bits().saturating_sub(1) as u32);
'blocks: for (block_index, members) in planes.members().enumerate() {
let slot_base = planes.slot_base(block_index)?;
let block = planes.block(slot_base)?;
total = total.saturating_add(members.count_ones() as usize);
if members == 0 {
continue;
}
let mut candidates = members;
let mut sortable_word = 0_u64;
let bounded_bits = if best_word < selective_bound {
block.planes.len()
} else {
0
};
let (bounded, prefix) = block.planes.split_at(bounded_bits);
for &plane in prefix.iter().rev() {
let preferred = candidates & self.direction.preferred_rows(plane);
sortable_word = sortable_word.wrapping_shl(1);
if preferred == 0 {
sortable_word |= 1;
} else {
candidates = preferred;
}
}
sortable_word = sortable_word.checked_shl(bounded_bits as u32).unwrap_or(0);
for (bit, &plane) in bounded.iter().enumerate().rev() {
let preferred = candidates & self.direction.preferred_rows(plane);
if preferred == 0 {
sortable_word |= 1_u64.wrapping_shl(bit as u32);
if sortable_word > best_word {
continue 'blocks;
}
} else {
candidates = preferred;
}
}
let slot = StableSlot::new(slot_base | candidates.trailing_zeros());
if best.is_none() || sortable_word < best_word {
best_word = sortable_word;
best = Some(slot);
}
}
if total != expected_members {
return Err(RangeIndexOrderingError::MemberCountMismatch {
field: planes.field,
table_rows: expected_members,
index_members: total,
});
}
Ok(best)
}
#[cfg(test)]
pub(in crate::storage::warp_table) fn sortable_word_at(
&self,
slot: StableSlot,
) -> Result<u64, RangeIndexOrderingError> {
let (block, mask) = self.validated_block_at(slot)?;
let ordered_word = block.planes.iter().rev().fold(0_u64, |word, plane| {
word.wrapping_shl(1) | u64::from(plane & mask != 0)
});
Ok(self.direction.sortable_word(ordered_word))
}
#[cfg(feature = "secondary-indexes")]
pub(in crate::storage::warp_table) fn validate_slot(
&self,
slot: StableSlot,
) -> Result<(), RangeIndexOrderingError> {
self.validated_block_at(slot).map(|_| ())
}
#[cfg(feature = "secondary-indexes")]
fn validated_block_at(
&self,
slot: StableSlot,
) -> Result<(RangeIndexBlock<'index>, u64), RangeIndexOrderingError> {
let planes = self.planes()?;
let slot = slot.value();
let block = planes.block(slot & !SLOT_OFFSET_MASK)?;
let mask = 1_u64.wrapping_shl(slot & SLOT_OFFSET_MASK);
if block.members & mask == 0 {
return Err(planes.inconsistent(RangeIndexInconsistency::MemberAbsent));
}
Ok((block, mask))
}
fn planes(&self) -> Result<&RangeIndexPlanes<'index>, RangeIndexOrderingError> {
self.planes.as_ref().map_err(|failure| *failure)
}
}
struct RangeIndexPlanes<'index> {
field: &'static str,
maximum_rows: usize,
words_per_block: usize,
blocks: &'index [u64],
}
struct RangeIndexBlock<'index> {
members: u64,
planes: &'index [u64],
}
impl<'index> RangeIndexPlanes<'index> {
const fn inconsistent(&self, reason: RangeIndexInconsistency) -> RangeIndexOrderingError {
RangeIndexOrderingError::Inconsistent {
field: self.field,
reason,
}
}
#[cfg(feature = "secondary-indexes")]
const fn bits(&self) -> usize {
self.words_per_block.saturating_sub(1)
}
#[cfg(feature = "secondary-indexes")]
fn members(&self) -> impl Iterator<Item = u64> + 'index {
self.blocks.iter().step_by(self.words_per_block).copied()
}
fn block(&self, slot_base: u32) -> Result<RangeIndexBlock<'index>, RangeIndexOrderingError> {
usize::try_from(slot_base / SLOTS_PER_BLOCK)
.ok()
.and_then(|block_index| {
self.blocks
.chunks_exact(self.words_per_block)
.nth(block_index)
})
.and_then(<[u64]>::split_first)
.map(|(members, planes)| RangeIndexBlock {
members: *members,
planes,
})
.ok_or_else(|| {
self.inconsistent(RangeIndexInconsistency::SlotOutsideCapacity {
maximum_rows: self.maximum_rows,
})
})
}
#[cfg(feature = "secondary-indexes")]
fn slot_base(&self, block_index: usize) -> Result<u32, RangeIndexOrderingError> {
u32::try_from(block_index)
.ok()
.and_then(|block_index| block_index.checked_mul(SLOTS_PER_BLOCK))
.ok_or_else(|| {
self.inconsistent(RangeIndexInconsistency::SlotOutsideCapacity {
maximum_rows: self.maximum_rows,
})
})
}
fn plane_at(&self, block_planes: &[u64], bit: usize) -> Result<u64, RangeIndexOrderingError> {
block_planes.get(bit).copied().ok_or_else(|| {
self.inconsistent(RangeIndexInconsistency::BitPositionOutsideWord {
bit: u32::try_from(bit).unwrap_or(u32::MAX),
})
})
}
}
#[cfg(test)]
mod tests;