use super::{Error, PersistentRow, TableProtocol};
use crate::storage::warp_table::*;
use alloc::vec::Vec;
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[cfg_attr(feature = "serde", derive(::serde::Serialize, ::serde::Deserialize))]
pub struct RecordLocation {
pub generation: u64,
pub offset: u64,
pub bytes: u32,
}
#[derive(Clone, Debug, Eq, PartialEq)]
#[cfg_attr(feature = "serde", derive(::serde::Serialize, ::serde::Deserialize))]
pub struct Located<Key> {
pub keys: Key,
pub location: RecordLocation,
}
pub trait IndexedRow: PersistentRow {
type Keys: PersistentRow<Id = Self::Id>;
fn keys(&self) -> Self::Keys;
}
pub trait IndexedProtocol<Row: IndexedRow>: TableProtocol<Row> {
fn query_plan(
query: &Self::Query,
indexes: IndexView<'_, Row::Keys>,
targets: &mut Targets<Row::Id>,
) -> Result<QueryPlan<Self::Value>, Self::Refusal>;
fn mutation_targets(
command: &Self::Mutation,
indexes: IndexView<'_, Row::Keys>,
targets: &mut Targets<Row::Id>,
) -> Result<(), Self::Refusal>;
}
pub enum QueryPlan<Value> {
Answer(Value),
Fetch,
}
pub struct Targets<Id> {
ids: Vec<Id>,
limit: usize,
refused: bool,
}
impl<Id: Copy + Eq> Targets<Id> {
pub fn new(limit: usize) -> Result<Self, Error> {
if limit == 0 {
return Err(Error::InvalidBounds);
}
let mut ids = Vec::new();
ids.try_reserve_exact(limit)
.map_err(|_| Error::Allocation)?;
Ok(Self {
ids,
limit,
refused: false,
})
}
pub fn push(&mut self, id: Id) -> Result<(), Error> {
if self.ids.contains(&id) {
return Ok(());
}
if self.ids.len() == self.limit {
self.refused = true;
return Err(Error::Full);
}
self.ids.push(id);
Ok(())
}
pub fn finish(self) -> Result<Vec<Id>, Error> {
if self.refused {
Err(Error::Full)
} else {
Ok(self.ids)
}
}
}
pub struct IndexView<'a, Key: PersistentRow, Brand = Located<Key>> {
table: &'a WarpTable<Located<Key>, Brand>,
}
impl<'a, Key: PersistentRow, Brand> IndexView<'a, Key, Brand> {
pub const fn new(table: &'a WarpTable<Located<Key>, Brand>) -> Self {
Self { table }
}
pub fn len(&self) -> usize {
self.table.len()
}
pub fn is_empty(&self) -> bool {
self.table.is_empty()
}
pub fn ids(&self) -> &'a [Key::Id] {
self.table.ids()
}
pub fn get(&self, id: Key::Id) -> Option<Key::Ref<'a>> {
match self.table.get(id) {
WarpTableGetOutcome::Present { row } => Some(row.keys),
WarpTableGetOutcome::Absent { .. } => None,
}
}
pub fn query(&self) -> Key::DetachedQuery<'a> {
Key::detached_query(self.table.__indexes())
}
#[cfg(feature = "secondary-indexes")]
pub fn select<Selection: Into<DetachedSelection>>(
&self,
select: impl FnOnce(Key::DetachedQuery<'a>) -> Selection,
) -> crate::storage::warp_table::Selection<'a, Located<Key>, Brand> {
super::super::query::bind_detached_selection(self.table, select(self.query()).into())
}
}
pub struct LocatedRef<'a, Key: PersistentRow + 'a> {
pub keys: Key::Ref<'a>,
pub location: &'a RecordLocation,
}
pub struct LocatedMut<'a, Key: PersistentRow + 'a> {
pub keys: Key::Mut<'a>,
pub location: &'a mut RecordLocation,
}
pub struct LocatedColumns<Key: PersistentRow> {
keys: Key::Columns,
locations: Vec<RecordLocation>,
}
#[derive(Debug)]
pub enum ColumnsError<E> {
Keys(E),
Locations(alloc::collections::TryReserveError),
}
impl<Key: PersistentRow> PersistentRow for Located<Key> {
fn own(row: Self::Ref<'_>) -> Self {
Self {
keys: Key::own(row.keys),
location: *row.location,
}
}
fn retained_bytes(&self) -> usize {
self.keys
.retained_bytes()
.saturating_add(core::mem::size_of::<RecordLocation>())
}
}
impl<Key: PersistentRow> WarpTableRow for Located<Key> {
type Id = Key::Id;
type Columns = LocatedColumns<Key>;
type ColumnsCreationError = ColumnsError<Key::ColumnsCreationError>;
type Indexes = Key::Indexes;
type IndexesCreationError = Key::IndexesCreationError;
type UniqueConflict = Key::UniqueConflict;
type ColumnsMut<'columns>
= ()
where
Self: 'columns;
fn columns_mut<'columns>(_: &'columns mut Self::Columns) -> Self::ColumnsMut<'columns>
where
Self: 'columns,
{
}
type IndexesInsertError = Key::IndexesInsertError;
type PreparedIndexesInsert<'a>
= Key::PreparedIndexesInsert<'a>
where
Self: 'a;
type IndexesRemoveError = Key::IndexesRemoveError;
type PreparedIndexesRemove<'a>
= Key::PreparedIndexesRemove<'a>
where
Self: 'a;
type IndexesReplaceError = Key::IndexesReplaceError;
type PreparedIndexesReplace<'a>
= Key::PreparedIndexesReplace<'a>
where
Self: 'a;
type Ref<'a>
= LocatedRef<'a, Key>
where
Self: 'a;
type Mut<'a>
= LocatedMut<'a, Key>
where
Self: 'a;
type Query<'a, Brand>
= IndexView<'a, Key, Brand>
where
Self: 'a,
Brand: 'a;
type MutQuery<'a, Brand>
= ()
where
Self: 'a,
Brand: 'a;
type DetachedQuery<'a>
= Key::DetachedQuery<'a>
where
Self: 'a;
fn try_new_columns(maximum_rows: usize) -> Result<Self::Columns, Self::ColumnsCreationError> {
let keys = Key::try_new_columns(maximum_rows).map_err(ColumnsError::Keys)?;
let mut locations = Vec::new();
locations
.try_reserve_exact(maximum_rows)
.map_err(ColumnsError::Locations)?;
Ok(LocatedColumns { keys, locations })
}
fn try_new_indexes(maximum_rows: usize) -> Result<Self::Indexes, Self::IndexesCreationError> {
Key::try_new_indexes(maximum_rows)
}
fn query<'a, Brand>(table: &'a WarpTable<Self, Brand>) -> Self::Query<'a, Brand>
where
Self: 'a,
Brand: 'a,
{
IndexView::new(table)
}
fn query_mut<'a, Brand>(_: &'a mut WarpTable<Self, Brand>) -> Self::MutQuery<'a, Brand>
where
Self: 'a,
Brand: 'a,
{
}
fn detached_query<'a>(indexes: &'a Self::Indexes) -> Self::DetachedQuery<'a>
where
Self: 'a,
{
Key::detached_query(indexes)
}
fn unique_insert_conflict(indexes: &Self::Indexes, row: &Self) -> Option<Self::UniqueConflict> {
Key::unique_insert_conflict(indexes, &row.keys)
}
fn unique_replace_conflict<'a>(
indexes: &Self::Indexes,
current: Self::Ref<'a>,
replacement: &Self,
) -> Option<Self::UniqueConflict>
where
Self: 'a,
{
Key::unique_replace_conflict(indexes, current.keys, &replacement.keys)
}
fn try_prepare_indexes_insert<'a>(
indexes: &'a mut Self::Indexes,
row: &Self,
) -> Result<Self::PreparedIndexesInsert<'a>, Self::IndexesInsertError>
where
Self: 'a,
{
Key::try_prepare_indexes_insert(indexes, &row.keys)
}
fn commit_indexes_insert<'a>(prepared: Self::PreparedIndexesInsert<'a>)
where
Self: 'a,
{
Key::commit_indexes_insert(prepared)
}
fn try_prepare_indexes_remove<'a, 'b>(
indexes: &'a mut Self::Indexes,
row: Self::Ref<'b>,
index: usize,
) -> Result<Self::PreparedIndexesRemove<'a>, Self::IndexesRemoveError>
where
Self: 'a + 'b,
{
Key::try_prepare_indexes_remove(indexes, row.keys, index)
}
fn commit_indexes_remove<'a>(prepared: Self::PreparedIndexesRemove<'a>)
where
Self: 'a,
{
Key::commit_indexes_remove(prepared)
}
fn try_prepare_indexes_replace<'a, 'b>(
indexes: &'a mut Self::Indexes,
current: Self::Ref<'b>,
replacement: &Self,
index: usize,
) -> Result<Self::PreparedIndexesReplace<'a>, Self::IndexesReplaceError>
where
Self: 'a + 'b,
{
Key::try_prepare_indexes_replace(indexes, current.keys, &replacement.keys, index)
}
fn commit_indexes_replace<'a>(prepared: Self::PreparedIndexesReplace<'a>)
where
Self: 'a,
{
Key::commit_indexes_replace(prepared)
}
fn append(columns: &mut Self::Columns, row: Self) {
Key::append(&mut columns.keys, row.keys);
columns.locations.push(row.location);
}
fn swap_remove(columns: &mut Self::Columns, index: usize) -> Self {
Self {
keys: Key::swap_remove(&mut columns.keys, index),
location: columns.locations.swap_remove(index),
}
}
fn get<'a>(
id: &'a Self::Id,
columns: &'a Self::Columns,
index: usize,
) -> Result<WarpTableRowGetOutcome<Self::Ref<'a>>, WarpTableRowGetError>
where
Self: 'a,
{
let location =
columns
.locations
.get(index)
.ok_or(WarpTableRowGetError::ColumnUnavailable {
column: 0,
row: index,
})?;
let WarpTableRowGetOutcome::Present { row: keys } = Key::get(id, &columns.keys, index)?;
Ok(WarpTableRowGetOutcome::Present {
row: LocatedRef { keys, location },
})
}
fn get_mut<'a>(
id: &'a Self::Id,
columns: &'a mut Self::Columns,
indexes: &'a mut Self::Indexes,
index: usize,
) -> Result<WarpTableRowGetOutcome<Self::Mut<'a>>, WarpTableRowGetError>
where
Self: 'a,
{
let location =
columns
.locations
.get_mut(index)
.ok_or(WarpTableRowGetError::ColumnUnavailable {
column: 0,
row: index,
})?;
let WarpTableRowGetOutcome::Present { row: keys } =
Key::get_mut(id, &mut columns.keys, indexes, index)?;
Ok(WarpTableRowGetOutcome::Present {
row: LocatedMut { keys, location },
})
}
fn replace(
columns: &mut Self::Columns,
index: usize,
replacement: Self,
) -> Result<WarpTableRowReplaceOutcome<Self>, WarpTableRowReplaceError<Self>> {
let Some(location) = columns.locations.get_mut(index) else {
return Err(WarpTableRowReplaceError::ColumnUnavailable {
column: 0,
row: index,
rejected: replacement,
});
};
match Key::replace(&mut columns.keys, index, replacement.keys) {
Ok(WarpTableRowReplaceOutcome::Replaced { previous }) => {
Ok(WarpTableRowReplaceOutcome::Replaced {
previous: Self {
keys: previous,
location: core::mem::replace(location, replacement.location),
},
})
}
Err(WarpTableRowReplaceError::ColumnUnavailable {
column,
row,
rejected,
}) => Err(WarpTableRowReplaceError::ColumnUnavailable {
column,
row,
rejected: Self {
keys: rejected,
location: replacement.location,
},
}),
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[derive(Clone, Eq, PartialEq, WarpTableRow)]
#[index(primary = u32)]
struct Key {
value: u8,
}
impl PersistentRow for Key {
fn own(row: KeyRef<'_>) -> Self {
Self { value: *row.value }
}
fn retained_bytes(&self) -> usize {
1
}
}
#[test]
fn missing_key_columns_preserve_the_rejected_row_and_stored_location() -> Result<(), Error> {
let location = RecordLocation {
generation: 1,
offset: 64,
bytes: 16,
};
let mut columns = LocatedColumns::<Key> {
keys: Key::try_new_columns(1).map_err(|_| Error::Allocation)?,
locations: alloc::vec![location],
};
let replacement = Located {
keys: Key { value: 9 },
location: RecordLocation {
offset: 128,
..location
},
};
match Located::<Key>::replace(&mut columns, 0, replacement.clone()) {
Err(WarpTableRowReplaceError::ColumnUnavailable {
column,
row,
rejected,
}) => {
assert_eq!(column, 0);
assert_eq!(row, 0);
assert!(rejected == replacement);
}
Ok(_) => return Err(Error::Corrupt),
}
assert_eq!(columns.locations, [location]);
Ok(())
}
}