use super::super::inconsistency::invariant_violation;
use super::super::{
IndexKey, WarpTable, WarpTableGetOutcome, WarpTableInsertWithIdOutcome, WarpTableRemoveOutcome,
WarpTableRow,
};
use super::change::{ProjectionChange, projection_change};
use super::delivery::{FirstChange, Previous, RetainedChanges};
use super::{ObservedTable, SourceId, ViewDeliveryBuffer, ViewDeliveryError};
use super::{ViewAdmission, ViewSubscription, ViewTransmitError, ViewTransmitOutcome};
use alloc::{collections::TryReserveError, vec::Vec};
use core::num::NonZeroU64;
crate::value_wrapper! {
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
pub struct ViewId(NonZeroU64);
}
pub trait ViewDefinition<Row: WarpTableRow> {
type Value: Clone + Eq;
fn project(&self, row: Row::Ref<'_>) -> Option<Self::Value>;
}
#[derive(
::core::clone::Clone,
::core::marker::Copy,
::core::fmt::Debug,
::core::cmp::Eq,
::core::cmp::PartialEq,
)]
#[cfg_attr(feature = "serde", derive(::serde::Serialize, ::serde::Deserialize))]
pub struct ViewRevision {
pub source: SourceId,
pub view: ViewId,
pub epoch: u64,
pub revision: u64,
}
#[derive(::core::clone::Clone, ::core::fmt::Debug, ::core::cmp::Eq, ::core::cmp::PartialEq)]
#[cfg_attr(feature = "serde", derive(::serde::Serialize, ::serde::Deserialize))]
pub enum ViewChange<Id, Value> {
Upsert {
id: Id,
incarnation: u64,
value: Value,
},
Remove {
id: Id,
incarnation: u64,
},
}
#[derive(
::core::clone::Clone,
::core::marker::Copy,
::core::fmt::Debug,
::core::cmp::Eq,
::core::cmp::PartialEq,
)]
pub enum ViewRefreshOutcome {
Unchanged,
Published { rows: usize, changes: usize },
Baseline { rows: usize },
}
#[derive(
::core::clone::Clone,
::core::marker::Copy,
::core::fmt::Debug,
::core::cmp::Eq,
::core::cmp::PartialEq,
)]
pub enum ViewRefreshError {
WrongSource,
RevisionExhausted,
}
#[derive(::core::fmt::Debug)]
pub enum ViewCreationError {
StorageUnavailable(TryReserveError),
TableUnavailable(
super::super::WarpTableCreationError<ViewColumnStorageError, core::convert::Infallible>,
),
CapacityOverflow,
}
#[derive(::core::fmt::Debug)]
pub struct ViewColumnStorageError {
pub column: &'static str,
pub source: TryReserveError,
}
pub(super) fn projected_storage<Id: IndexKey, Value>(
capacity: super::super::Capacity,
) -> Result<WarpTable<Projected<Id, Value>>, ViewCreationError> {
WarpTable::try_new(capacity).map_err(|error| {
creation_error(
error,
|ProjectedColumnsCreationError::StorageUnavailable { column, source }| {
ViewColumnStorageError { column, source }
},
|never| match never {},
)
})
}
pub(super) fn seen_storage<Id: IndexKey>(
capacity: super::super::Capacity,
) -> Result<WarpTable<Seen<Id>>, ViewCreationError> {
WarpTable::try_new(capacity).map_err(seen_creation_error)
}
fn seen_creation_error(
error: super::super::WarpTableCreationError<SeenColumnsCreationError, SeenIndexesCreationError>,
) -> ViewCreationError {
creation_error(
error,
|SeenColumnsCreationError::StorageUnavailable { column, source }| ViewColumnStorageError {
column,
source,
},
|never| match never {},
)
}
pub(super) fn creation_error<Columns, Indexes>(
error: super::super::WarpTableCreationError<Columns, Indexes>,
columns: impl FnOnce(Columns) -> ViewColumnStorageError,
indexes: impl FnOnce(Indexes) -> ViewCreationError,
) -> ViewCreationError {
use super::super::WarpTableCreationError as Error;
ViewCreationError::TableUnavailable(match error {
Error::CapacityUnrepresentable { capacity, source } => {
Error::CapacityUnrepresentable { capacity, source }
}
Error::ColumnsUnavailable { source } => Error::ColumnsUnavailable {
source: columns(source),
},
Error::IndexesUnavailable { source } => return indexes(source),
Error::IdStorageUnavailable { source } => Error::IdStorageUnavailable { source },
Error::PrimaryIndexCapacityOverflow { maximum_rows } => {
Error::PrimaryIndexCapacityOverflow { maximum_rows }
}
Error::PrimaryIndexStorageUnavailable { source } => {
Error::PrimaryIndexStorageUnavailable { source }
}
})
}
pub enum ViewPublication<'view, Id, Value> {
Baseline {
through: ViewRevision,
rows: &'view [ViewChange<Id, Value>],
},
Changes {
base: ViewRevision,
through: ViewRevision,
changes: &'view [ViewChange<Id, Value>],
},
}
pub struct ViewRow<'view, Id, Value> {
pub id: Id,
pub incarnation: u64,
pub value: &'view Value,
}
pub struct ViewRows<'view, Id: IndexKey, Value> {
inner: super::super::WarpTableRows<'view, Projected<Id, Value>>,
}
impl<'view, Id: IndexKey, Value> Iterator for ViewRows<'view, Id, Value> {
type Item = ViewRow<'view, Id, Value>;
fn next(&mut self) -> Option<Self::Item> {
self.inner.next().map(|row| ViewRow {
id: *row.id,
incarnation: *row.incarnation,
value: row.value,
})
}
fn size_hint(&self) -> (usize, Option<usize>) {
self.inner.size_hint()
}
}
impl<Id: IndexKey, Value> ExactSizeIterator for ViewRows<'_, Id, Value> {}
impl<Id: IndexKey, Value> core::iter::FusedIterator for ViewRows<'_, Id, Value> {}
pub enum ViewDelivery<'view, Id: IndexKey, Value> {
Baseline {
through: ViewRevision,
rows: ViewRows<'view, Id, Value>,
},
Changes {
base: ViewRevision,
through: ViewRevision,
changes: &'view [ViewChange<Id, Value>],
},
Current {
through: ViewRevision,
},
}
#[derive(::pipecircuit_macros::WarpTableRow)]
#[index(primary = Id)]
pub(super) struct Projected<Id: IndexKey, Value> {
pub(super) incarnation: u64,
pub(super) value: Value,
pub(super) marker: core::marker::PhantomData<Id>,
}
#[derive(::pipecircuit_macros::WarpTableRow)]
#[index(primary = Id)]
pub(super) struct Seen<Id: IndexKey> {
pub(super) marker: core::marker::PhantomData<Id>,
}
enum PublicationKind {
Baseline,
Changes { base: ViewRevision },
}
pub struct MaintainedView<Row: WarpTableRow, Definition: ViewDefinition<Row>, Brand = Row> {
source: SourceId,
id: ViewId,
definition: Definition,
epoch: u64,
revision: u64,
observed_through: u64,
rows: WarpTable<Projected<Row::Id, Definition::Value>>,
seen: WarpTable<Seen<Row::Id>>,
pending: Vec<ViewChange<Row::Id, Definition::Value>>,
publication: Vec<ViewChange<Row::Id, Definition::Value>>,
kind: PublicationKind,
history: RetainedChanges<Row::Id, Definition::Value>,
brand: core::marker::PhantomData<fn() -> Brand>,
}
impl<Row: WarpTableRow, Brand> ObservedTable<Row, Brand> {
pub fn maintain<Definition: ViewDefinition<Row>>(
&self,
id: ViewId,
definition: Definition,
) -> Result<MaintainedView<Row, Definition, Brand>, ViewCreationError> {
self.maintain_with_retention(id, definition, self.table.capacity())
}
pub fn maintain_with_retention<Definition: ViewDefinition<Row>>(
&self,
id: ViewId,
definition: Definition,
retained_changes: super::super::Capacity,
) -> Result<MaintainedView<Row, Definition, Brand>, ViewCreationError> {
let history = RetainedChanges::try_new(retained_changes)?;
let capacity = self.table.capacity();
let rows = projected_storage(capacity)?;
let seen = seen_storage(capacity)?;
let maximum_changes = self
.table
.maximum_rows
.checked_mul(2)
.ok_or(ViewCreationError::CapacityOverflow)?;
let mut pending = Vec::new();
pending
.try_reserve_exact(maximum_changes)
.map_err(ViewCreationError::StorageUnavailable)?;
let mut publication = Vec::new();
publication
.try_reserve_exact(maximum_changes)
.map_err(ViewCreationError::StorageUnavailable)?;
let mut view = MaintainedView {
source: self.source,
id,
definition,
epoch: 1,
revision: 0,
observed_through: self.sequence,
rows,
seen,
pending,
publication,
kind: PublicationKind::Baseline,
history,
brand: core::marker::PhantomData,
};
view.prepare_baseline(self);
view.commit(true, 0);
Ok(view)
}
}
impl<Row: WarpTableRow, Definition: ViewDefinition<Row>, Brand>
MaintainedView<Row, Definition, Brand>
{
pub const fn revision(&self) -> ViewRevision {
ViewRevision {
source: self.source,
view: self.id,
epoch: self.epoch,
revision: self.revision,
}
}
pub const fn len(&self) -> usize {
self.rows.len()
}
pub const fn is_empty(&self) -> bool {
self.rows.is_empty()
}
pub fn rows(&self) -> ViewRows<'_, Row::Id, Definition::Value> {
ViewRows {
inner: self.rows.rows(),
}
}
pub fn subscribe(&self) -> ViewSubscription {
ViewSubscription {
source: self.source,
view: self.id,
cursor: None,
}
}
pub fn transmit<Error>(
&self,
subscription: &mut ViewSubscription,
buffer: &mut ViewDeliveryBuffer<Row::Id, Definition::Value>,
send: impl for<'delivery> FnOnce(
ViewDelivery<'delivery, Row::Id, Definition::Value>,
) -> Result<ViewAdmission, Error>,
) -> Result<ViewTransmitOutcome, ViewTransmitError<Error>> {
if subscription.source != self.source || subscription.view != self.id {
return Err(ViewTransmitError::Preparation(ViewDeliveryError::WrongView));
}
let delivery = self
.delivery(subscription.cursor, buffer)
.map_err(ViewTransmitError::Preparation)?;
let through = self.revision();
if let ViewDelivery::Current { .. } = delivery {
return Ok(ViewTransmitOutcome::Current { through });
}
match send(delivery).map_err(ViewTransmitError::Transport)? {
ViewAdmission::Admitted => {
subscription.cursor = Some(through);
Ok(ViewTransmitOutcome::Admitted { through })
}
ViewAdmission::Backpressured => Ok(ViewTransmitOutcome::Backpressured { through }),
}
}
pub fn delivery_buffer(
&self,
) -> Result<ViewDeliveryBuffer<Row::Id, Definition::Value>, ViewCreationError> {
ViewDeliveryBuffer::try_new(self.rows.capacity(), self.history.capacity)
}
pub fn retained_changes(&self) -> usize {
self.history.entries.len()
}
pub fn delivery<'delivery>(
&'delivery self,
cursor: Option<ViewRevision>,
buffer: &'delivery mut ViewDeliveryBuffer<Row::Id, Definition::Value>,
) -> Result<ViewDelivery<'delivery, Row::Id, Definition::Value>, ViewDeliveryError> {
let through = self.revision();
let Some(base) = cursor else {
return Ok(ViewDelivery::Baseline {
through,
rows: self.rows(),
});
};
if base.source != through.source || base.view != through.view {
return Err(ViewDeliveryError::WrongView);
}
if (base.epoch, base.revision) > (through.epoch, through.revision) {
return Err(ViewDeliveryError::FutureCursor);
}
if base == through {
return Ok(ViewDelivery::Current { through });
}
if let PublicationKind::Changes { base: latest_base } = self.kind
&& base == latest_base
{
return Ok(ViewDelivery::Changes {
base,
through,
changes: &self.publication,
});
}
if base.epoch != through.epoch || base.revision < self.history.discarded_through {
return Ok(ViewDelivery::Baseline {
through,
rows: self.rows(),
});
}
if buffer.maximum_rows < self.rows.maximum_rows
|| buffer.maximum_history < self.history.maximum_changes
{
return Err(ViewDeliveryError::InsufficientBuffer);
}
clear(&mut buffer.first);
buffer.changes.clear();
for (index, change) in self
.history
.entries
.iter()
.enumerate()
.rev()
.take_while(|(_, change)| change.revision > base.revision)
{
match buffer.first.upsert(
change.id,
FirstChange {
index,
marker: core::marker::PhantomData,
},
) {
super::super::WarpTableUpsertOutcome::Inserted { .. }
| super::super::WarpTableUpsertOutcome::Replaced { .. } => {}
super::super::WarpTableUpsertOutcome::Full { .. } => {
invariant_violation("retained history exceeded delivery preparation bound")
}
super::super::WarpTableUpsertOutcome::UniqueConflict { conflict, .. } => {
match conflict {}
}
}
}
self.compose_delivery_changes(buffer);
Ok(ViewDelivery::Changes {
base,
through,
changes: &buffer.changes,
})
}
fn compose_delivery_changes(
&self,
buffer: &mut ViewDeliveryBuffer<Row::Id, Definition::Value>,
) {
for first in buffer.first.rows() {
let Some(previous) = self.history.entries.get(*first.index) else {
invariant_violation("retained history disappeared during delivery")
};
let current = match self.rows.get(*first.id) {
WarpTableGetOutcome::Present { row } => Some((*row.incarnation, row.value)),
WarpTableGetOutcome::Absent { .. } => None,
};
match projection_change(
previous
.before
.as_ref()
.map(|(incarnation, value)| (*incarnation, value)),
current,
) {
ProjectionChange::Unchanged => {}
ProjectionChange::Upsert { incarnation, value } => {
buffer.changes.push(ViewChange::Upsert {
id: *first.id,
incarnation,
value: value.clone(),
});
}
ProjectionChange::Remove { incarnation } => {
buffer.changes.push(ViewChange::Remove {
id: *first.id,
incarnation,
})
}
}
}
}
pub fn publication(&self) -> ViewPublication<'_, Row::Id, Definition::Value> {
let through = self.revision();
match self.kind {
PublicationKind::Baseline => ViewPublication::Baseline {
through,
rows: &self.publication,
},
PublicationKind::Changes { base } => ViewPublication::Changes {
base,
through,
changes: &self.publication,
},
}
}
pub fn refresh(
&mut self,
source: &ObservedTable<Row, Brand>,
) -> Result<ViewRefreshOutcome, ViewRefreshError> {
Ok(self.prepare_refresh(source)?.publish())
}
pub fn prepare_refresh(
&mut self,
source: &ObservedTable<Row, Brand>,
) -> Result<PreparedViewRefresh<'_, Row, Definition, Brand>, ViewRefreshError> {
if self.source != source.source {
return Err(ViewRefreshError::WrongSource);
}
let Some(next_revision) = self.revision.checked_add(1) else {
return Err(ViewRefreshError::RevisionExhausted);
};
self.pending.clear();
clear(&mut self.seen);
let rebuild = self.observed_through < source.discarded_through;
if rebuild {
self.prepare_full_refresh(source);
} else {
let observed_through = self.observed_through;
for change in source
.history
.iter()
.rev()
.take_while(|change| change.sequence > observed_through)
{
match self.seen.insert_with_id(
change.id,
Seen {
marker: core::marker::PhantomData,
},
) {
WarpTableInsertWithIdOutcome::Inserted { .. } => {
self.prepare_row(source, change.id)
}
WarpTableInsertWithIdOutcome::AlreadyPresent { .. } => {}
WarpTableInsertWithIdOutcome::Full { .. } => {
invariant_violation("bounded change history exceeded seen storage")
}
WarpTableInsertWithIdOutcome::UniqueConflict { conflict, .. } => {
match conflict {}
}
}
}
}
Ok(PreparedViewRefresh {
view: self,
observed_through: source.sequence,
next_revision,
})
}
pub fn replace_definition(
&mut self,
source: &ObservedTable<Row, Brand>,
definition: Definition,
) -> Result<ViewRefreshOutcome, (Definition, ViewRefreshError)> {
if self.source != source.source {
return Err((definition, ViewRefreshError::WrongSource));
}
let Some(epoch) = self.epoch.checked_add(1) else {
return Err((definition, ViewRefreshError::RevisionExhausted));
};
self.definition = definition;
self.pending.clear();
self.prepare_baseline(source);
self.commit(true, 0);
self.epoch = epoch;
self.revision = 0;
self.observed_through = source.sequence;
self.kind = PublicationKind::Baseline;
Ok(ViewRefreshOutcome::Baseline { rows: self.len() })
}
fn prepare_baseline(&mut self, source: &ObservedTable<Row, Brand>) {
for ((id, incarnation), row) in source
.table
.ids()
.iter()
.zip(&source.incarnations)
.zip(source.table.rows())
{
if let Some(value) = self.definition.project(row) {
self.pending.push(ViewChange::Upsert {
id: *id,
incarnation: *incarnation,
value,
});
}
}
}
fn prepare_full_refresh(&mut self, source: &ObservedTable<Row, Brand>) {
for row in self.rows.rows() {
if let WarpTableGetOutcome::Absent { .. } = source.table.get(*row.id) {
self.pending.push(ViewChange::Remove {
id: *row.id,
incarnation: *row.incarnation,
});
}
}
for id in source.table.ids() {
self.prepare_row(source, *id);
}
}
fn prepare_row(&mut self, source: &ObservedTable<Row, Brand>, id: Row::Id) {
let current = match source.table.primary_index.find(id, &source.table.ids) {
Ok(super::super::dense_key_index::FindOutcome::Present { location, key }) => {
let row = location.row;
let Some(incarnation) = source.incarnations.get(row) else {
invariant_violation("observed incarnation missing")
};
super::super::get::validate_id_count(source.table.ids.len(), source.table.len);
let row = super::super::get::row_ref::<Row>(key, &source.table.columns, row);
self.definition
.project(row)
.map(|value| (*incarnation, value))
}
Ok(super::super::dense_key_index::FindOutcome::Absent) => None,
Err(reason) => invariant_violation(reason),
};
let previous = match self.rows.get(id) {
WarpTableGetOutcome::Present { row } => Some((*row.incarnation, row.value)),
WarpTableGetOutcome::Absent { .. } => None,
};
match projection_change(previous, current) {
ProjectionChange::Unchanged => {}
ProjectionChange::Upsert { incarnation, value } => {
self.pending.push(ViewChange::Upsert {
id,
incarnation,
value,
});
}
ProjectionChange::Remove { incarnation } => {
self.pending.push(ViewChange::Remove { id, incarnation })
}
}
}
fn commit(&mut self, baseline: bool, revision: u64) {
self.publication.clear();
self.publication.extend(self.pending.iter().cloned());
if baseline {
clear(&mut self.rows);
self.history.reset(revision);
} else {
for change in &self.pending {
let id = match change {
ViewChange::Upsert { id, .. } | ViewChange::Remove { id, .. } => *id,
};
let before = match self.rows.get(id) {
WarpTableGetOutcome::Present { row } => {
Some((*row.incarnation, row.value.clone()))
}
WarpTableGetOutcome::Absent { .. } => None,
};
self.history.push(Previous {
revision,
id,
before,
});
}
}
for change in &self.pending {
if let ViewChange::Remove { id, .. } = change {
match self.rows.remove(*id) {
WarpTableRemoveOutcome::Removed { .. } => {}
WarpTableRemoveOutcome::Absent { .. } => {
invariant_violation("projected removal absent")
}
}
}
}
for change in self.pending.drain(..) {
if let ViewChange::Upsert {
id,
incarnation,
value,
} = change
{
match self.rows.upsert(
id,
Projected {
incarnation,
value,
marker: core::marker::PhantomData,
},
) {
super::super::WarpTableUpsertOutcome::Inserted { .. }
| super::super::WarpTableUpsertOutcome::Replaced { .. } => {}
super::super::WarpTableUpsertOutcome::Full { .. } => {
invariant_violation("projection exceeded source capacity")
}
super::super::WarpTableUpsertOutcome::UniqueConflict { conflict, .. } => {
match conflict {}
}
}
}
}
}
}
#[must_use = "publish the prepared cut or explicitly drop it to abandon it"]
pub struct PreparedViewRefresh<
'view,
Row: WarpTableRow,
Definition: ViewDefinition<Row>,
Brand = Row,
> {
view: &'view mut MaintainedView<Row, Definition, Brand>,
observed_through: u64,
next_revision: u64,
}
impl<Row: WarpTableRow, Definition: ViewDefinition<Row>, Brand>
PreparedViewRefresh<'_, Row, Definition, Brand>
{
pub fn publish(self) -> ViewRefreshOutcome {
let view = &mut *self.view;
if view.pending.is_empty() {
view.observed_through = self.observed_through;
return ViewRefreshOutcome::Unchanged;
}
let base = view.revision();
view.commit(false, self.next_revision);
view.revision = self.next_revision;
view.observed_through = self.observed_through;
view.kind = PublicationKind::Changes { base };
ViewRefreshOutcome::Published {
rows: view.len(),
changes: view.publication.len(),
}
}
}
impl<Row: WarpTableRow, Definition: ViewDefinition<Row>, Brand> Drop
for PreparedViewRefresh<'_, Row, Definition, Brand>
{
fn drop(&mut self) {
self.view.pending.clear();
}
}
pub(super) fn clear<Row: WarpTableRow>(table: &mut WarpTable<Row>) {
while let Some(id) = table.ids().last().copied() {
match table.remove(id) {
WarpTableRemoveOutcome::Removed { .. } => {}
WarpTableRemoveOutcome::Absent { .. } => {
invariant_violation("table identity unavailable during clear")
}
}
}
}
#[cfg(test)]
mod tests;