use crate::storage::warp_table::inconsistency::invariant_violation;
use crate::storage::warp_table::secondary_index::{
RangeIndexMembership, SlotSet, StableSlotResolveOutcome, StableSlotResolver,
};
use crate::storage::warp_table::{
RangeIndexBounds, WarpTable, WarpTableRemoveOutcome, WarpTableRow, WarpTableRowRelocation,
};
#[doc(hidden)]
pub struct DetachedSelection {
slots: Option<SlotSet>,
}
#[doc(hidden)]
#[must_use]
pub enum DetachedSecondaryIndexLookupOutcome<SecondaryKey> {
Present {
secondary_key: SecondaryKey,
selection: DetachedSelection,
},
Absent {
secondary_key: SecondaryKey,
selection: DetachedSelection,
},
}
#[doc(hidden)]
#[must_use]
pub enum DetachedRangeIndexLookupOutcome<Key> {
Present {
bounds: RangeIndexBounds<Key>,
selection: DetachedSelection,
},
Absent {
bounds: RangeIndexBounds<Key>,
selection: DetachedSelection,
},
}
macro_rules! impl_detached_lookup {
($outcome:ident) => {
impl<Key> $outcome<Key> {
fn into_selection(self) -> DetachedSelection {
match self {
Self::Present { selection, .. } | Self::Absent { selection, .. } => selection,
}
}
}
impl<Key> From<$outcome<Key>> for DetachedSelection {
fn from(outcome: $outcome<Key>) -> Self {
outcome.into_selection()
}
}
};
}
impl_detached_lookup!(DetachedSecondaryIndexLookupOutcome);
impl_detached_lookup!(DetachedRangeIndexLookupOutcome);
#[must_use]
pub struct MutableSelection<'table, Row: WarpTableRow, Brand = Row> {
table: &'table mut WarpTable<Row, Brand>,
slots: Option<SlotSet>,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[must_use]
pub enum MutableSelectionForEachOutcome {
Updated { rows: u64 },
}
#[derive(Debug, Eq, PartialEq)]
#[must_use]
pub enum MutableSelectionRemoveFirstOutcome<Id, Row> {
Removed {
id: Id,
removed: Row,
relocation: WarpTableRowRelocation<Id>,
},
Empty,
}
macro_rules! mutable_algebra_methods {
($receiver:ident, $selection:expr) => {
pub fn and<Operand>(
$receiver,
find: impl for<'indexes> FnOnce(Row::DetachedQuery<'indexes>) -> Operand,
) -> Operand::Output
where
Operand: MutableSelectionOperand<'table, Row, Brand>,
{
let selection = $selection;
find(Row::detached_query(selection.table.__indexes())).and_with(selection)
}
pub fn or<Operand>(
$receiver,
find: impl for<'indexes> FnOnce(Row::DetachedQuery<'indexes>) -> Operand,
) -> Operand::Output
where
Operand: MutableSelectionOperand<'table, Row, Brand>,
{
let selection = $selection;
find(Row::detached_query(selection.table.__indexes())).or_with(selection)
}
pub fn and_not<Operand>(
$receiver,
find: impl for<'indexes> FnOnce(Row::DetachedQuery<'indexes>) -> Operand,
) -> Operand::Output
where
Operand: MutableSelectionOperand<'table, Row, Brand>,
{
let selection = $selection;
find(Row::detached_query(selection.table.__indexes())).and_not_with(selection)
}
pub fn xor<Operand>(
$receiver,
find: impl for<'indexes> FnOnce(Row::DetachedQuery<'indexes>) -> Operand,
) -> Operand::Output
where
Operand: MutableSelectionOperand<'table, Row, Brand>,
{
let selection = $selection;
find(Row::detached_query(selection.table.__indexes())).xor_with(selection)
}
};
}
impl<'table, Row: WarpTableRow, Brand> MutableSelection<'table, Row, Brand> {
fn new(table: &'table mut WarpTable<Row, Brand>, slots: Option<SlotSet>) -> Self {
Self { table, slots }
}
pub fn len(&self) -> u64 {
self.slots.as_ref().map_or(0, SlotSet::len)
}
pub fn is_empty(&self) -> bool {
self.slots.as_ref().is_none_or(SlotSet::is_empty)
}
mutable_algebra_methods!(self, self);
pub fn for_each(self, mut update: impl FnMut(Row::Mut<'_>)) -> MutableSelectionForEachOutcome
where
Row::Indexes: super::rows::StableSlotResolver,
{
let Some(slots) = self.slots else {
return MutableSelectionForEachOutcome::Updated { rows: 0 };
};
let rows = slots.len();
for slot in slots.into_slots() {
let StableSlotResolveOutcome::Resolved { dense_row, .. } = self
.table
.__indexes()
.resolve_stable_slot(slot)
.unwrap_or_else(|reason| invariant_violation(reason));
self.table.update_at(dense_row, |row| update(row));
}
MutableSelectionForEachOutcome::Updated { rows }
}
pub fn remove_first(self) -> MutableSelectionRemoveFirstOutcome<Row::Id, Row>
where
Row::Indexes: StableSlotResolver,
{
let Some(slots) = self.slots else {
return MutableSelectionRemoveFirstOutcome::Empty;
};
let Some(slot) = slots.into_slots().next() else {
return MutableSelectionRemoveFirstOutcome::Empty;
};
let StableSlotResolveOutcome::Resolved { dense_row, .. } = self
.table
.__indexes()
.resolve_stable_slot(slot)
.unwrap_or_else(|reason| invariant_violation(reason));
let id = self.table.ids.get(dense_row).copied().unwrap_or_else(|| {
invariant_violation(
super::rows::SelectionMembershipInconsistency::RowUnavailable { row: dense_row },
)
});
match self.table.remove(id) {
WarpTableRemoveOutcome::Removed {
id,
removed,
relocation,
} => MutableSelectionRemoveFirstOutcome::Removed {
id,
removed,
relocation,
},
WarpTableRemoveOutcome::Absent { .. } => invariant_violation(
super::rows::SelectionMembershipInconsistency::RowUnavailable { row: dense_row },
),
}
}
fn union(self, right: DetachedSelection) -> Self {
let slots = match (self.slots, right.slots) {
(None, right) => right,
(left, None) => left,
(Some(mut left), Some(right)) => {
left.union_assign(&right);
Some(left)
}
};
Self { slots, ..self }
}
fn intersection(self, right: DetachedSelection) -> Self {
let slots = match (self.slots, right.slots) {
(Some(left), Some(right)) => Some(left.intersection_owned(right)),
_ => None,
};
Self { slots, ..self }
}
fn excluding(self, right: DetachedSelection) -> Self {
let slots = match (self.slots, right.slots) {
(left, None) => left,
(Some(mut left), Some(right)) => {
left.exclude_assign(&right);
Some(left)
}
(None, Some(_)) => None,
};
Self { slots, ..self }
}
fn union_excluding_overlap(self, right: DetachedSelection) -> Self {
let slots = match (self.slots, right.slots) {
(None, right) => right,
(left, None) => left,
(Some(mut left), Some(right)) => {
left.union_excluding_overlap_assign(&right);
Some(left)
}
};
Self { slots, ..self }
}
}
#[doc(hidden)]
pub trait MutableSelectionOperand<'table, Row: WarpTableRow, Brand> {
type Output;
fn and_with(self, left: MutableSelection<'table, Row, Brand>) -> Self::Output;
fn or_with(self, left: MutableSelection<'table, Row, Brand>) -> Self::Output;
fn and_not_with(self, left: MutableSelection<'table, Row, Brand>) -> Self::Output;
fn xor_with(self, left: MutableSelection<'table, Row, Brand>) -> Self::Output;
}
impl<'table, Row: WarpTableRow + 'table, Brand: 'table, Operand>
MutableSelectionOperand<'table, Row, Brand> for Operand
where
Operand: Into<DetachedSelection>,
{
type Output = MutableSelection<'table, Row, Brand>;
fn and_with(self, left: MutableSelection<'table, Row, Brand>) -> Self::Output {
left.intersection(self.into())
}
fn or_with(self, left: MutableSelection<'table, Row, Brand>) -> Self::Output {
left.union(self.into())
}
fn and_not_with(self, left: MutableSelection<'table, Row, Brand>) -> Self::Output {
left.excluding(self.into())
}
fn xor_with(self, left: MutableSelection<'table, Row, Brand>) -> Self::Output {
left.union_excluding_overlap(self.into())
}
}
impl<'table, Row, Brand, Operand, Error> MutableSelectionOperand<'table, Row, Brand>
for Result<Operand, Error>
where
Row: WarpTableRow + 'table,
Brand: 'table,
Operand: Into<DetachedSelection>,
{
type Output = Result<MutableSelection<'table, Row, Brand>, Error>;
fn and_with(self, left: MutableSelection<'table, Row, Brand>) -> Self::Output {
self.map(|right| left.intersection(right.into()))
}
fn or_with(self, left: MutableSelection<'table, Row, Brand>) -> Self::Output {
self.map(|right| left.union(right.into()))
}
fn and_not_with(self, left: MutableSelection<'table, Row, Brand>) -> Self::Output {
self.map(|right| left.excluding(right.into()))
}
fn xor_with(self, left: MutableSelection<'table, Row, Brand>) -> Self::Output {
self.map(|right| left.union_excluding_overlap(right.into()))
}
}
#[must_use]
pub enum MutableSecondaryIndexLookupOutcome<'table, Row: WarpTableRow, Brand, SecondaryKey> {
Present {
secondary_key: SecondaryKey,
selection: MutableSelection<'table, Row, Brand>,
},
Absent {
secondary_key: SecondaryKey,
selection: MutableSelection<'table, Row, Brand>,
},
}
#[must_use]
pub enum MutableRangeIndexLookupOutcome<'table, Row: WarpTableRow, Brand, Key> {
Present {
bounds: RangeIndexBounds<Key>,
selection: MutableSelection<'table, Row, Brand>,
},
Absent {
bounds: RangeIndexBounds<Key>,
selection: MutableSelection<'table, Row, Brand>,
},
}
macro_rules! impl_mutable_lookup {
($outcome:ident) => {
impl<'table, Row: WarpTableRow, Brand, Key> $outcome<'table, Row, Brand, Key> {
pub fn len(&self) -> u64 {
self.selection().len()
}
pub fn is_empty(&self) -> bool {
self.selection().is_empty()
}
mutable_algebra_methods!(self, self.into_selection());
pub fn for_each(
self,
update: impl FnMut(Row::Mut<'_>),
) -> MutableSelectionForEachOutcome
where
Row::Indexes: super::rows::StableSlotResolver,
{
self.into_selection().for_each(update)
}
pub fn remove_first(self) -> MutableSelectionRemoveFirstOutcome<Row::Id, Row>
where
Row::Indexes: StableSlotResolver,
{
self.into_selection().remove_first()
}
fn selection(&self) -> &MutableSelection<'table, Row, Brand> {
match self {
Self::Present { selection, .. } | Self::Absent { selection, .. } => selection,
}
}
fn into_selection(self) -> MutableSelection<'table, Row, Brand> {
match self {
Self::Present { selection, .. } | Self::Absent { selection, .. } => selection,
}
}
}
};
}
impl_mutable_lookup!(MutableSecondaryIndexLookupOutcome);
impl_mutable_lookup!(MutableRangeIndexLookupOutcome);
#[doc(hidden)]
pub fn bind_detached_secondary_lookup<SecondaryKey>(
membership: (SecondaryKey, Option<SlotSet>),
) -> DetachedSecondaryIndexLookupOutcome<SecondaryKey> {
let (secondary_key, slots) = membership;
let present = slots.is_some();
let selection = DetachedSelection { slots };
if present {
DetachedSecondaryIndexLookupOutcome::Present {
secondary_key,
selection,
}
} else {
DetachedSecondaryIndexLookupOutcome::Absent {
secondary_key,
selection,
}
}
}
#[doc(hidden)]
pub fn bind_detached_range_lookup<Key>(
membership: RangeIndexMembership<Key>,
) -> DetachedRangeIndexLookupOutcome<Key> {
let (bounds, slots) = membership.into_parts();
let present = slots.is_some();
let selection = DetachedSelection { slots };
if present {
DetachedRangeIndexLookupOutcome::Present { bounds, selection }
} else {
DetachedRangeIndexLookupOutcome::Absent { bounds, selection }
}
}
#[doc(hidden)]
pub fn bind_mutable_secondary_lookup<'table, Row, Brand, SecondaryKey>(
table: &'table mut WarpTable<Row, Brand>,
membership: (SecondaryKey, Option<SlotSet>),
) -> MutableSecondaryIndexLookupOutcome<'table, Row, Brand, SecondaryKey>
where
Row: WarpTableRow,
{
let (secondary_key, slots) = membership;
let present = slots.is_some();
let selection = MutableSelection::new(table, slots);
if present {
MutableSecondaryIndexLookupOutcome::Present {
secondary_key,
selection,
}
} else {
MutableSecondaryIndexLookupOutcome::Absent {
secondary_key,
selection,
}
}
}
#[doc(hidden)]
pub fn bind_mutable_range_lookup<'table, Row, Brand, Key>(
table: &'table mut WarpTable<Row, Brand>,
membership: RangeIndexMembership<Key>,
) -> MutableRangeIndexLookupOutcome<'table, Row, Brand, Key>
where
Row: WarpTableRow,
{
let (bounds, slots) = membership.into_parts();
let present = slots.is_some();
let selection = MutableSelection::new(table, slots);
if present {
MutableRangeIndexLookupOutcome::Present { bounds, selection }
} else {
MutableRangeIndexLookupOutcome::Absent { bounds, selection }
}
}