use std::{
collections::HashMap,
ops::Bound::{Excluded, Included, Unbounded},
};
use reifydb_catalog::catalog::Catalog;
use reifydb_codec::{
key::encoded::{EncodedKey, EncodedKeyRange},
row::{bytes::EncodedBytes, pod::EncodedPodRow, shape::RowShape},
};
use reifydb_core::{
actors::pending::PendingLayers,
common::CommitVersion,
interface::{catalog::flow::OperatorId, change::Change, store::MultiVersionRow},
key::any::TaggedKey,
};
use reifydb_runtime::context::clock::Clock;
use reifydb_store_operator::store::OperatorStore;
use reifydb_transaction::{
accumulator::ChangeAccumulator,
interceptor::{
WithInterceptors,
authentication::{AuthenticationPostCreateInterceptor, AuthenticationPreDeleteInterceptor},
chain::InterceptorChain as Chain,
dictionary::{
DictionaryPostCreateInterceptor, DictionaryPostUpdateInterceptor,
DictionaryPreDeleteInterceptor, DictionaryPreUpdateInterceptor,
},
dictionary_row::{
DictionaryRowPostDeleteInterceptor, DictionaryRowPostInsertInterceptor,
DictionaryRowPostUpdateInterceptor, DictionaryRowPreDeleteInterceptor,
DictionaryRowPreInsertInterceptor, DictionaryRowPreUpdateInterceptor,
},
granted_role::{GrantedRolePostCreateInterceptor, GrantedRolePreDeleteInterceptor},
identity::{IdentityPostCreateInterceptor, IdentityPreDeleteInterceptor},
identity_attribute::{IdentityAttributePostCreateInterceptor, IdentityAttributePreDeleteInterceptor},
identity_attribute_value::{
IdentityAttributeValuePostCreateInterceptor, IdentityAttributeValuePreDeleteInterceptor,
},
interceptors::Interceptors,
namespace::{
NamespacePostCreateInterceptor, NamespacePostUpdateInterceptor, NamespacePreDeleteInterceptor,
NamespacePreUpdateInterceptor,
},
ringbuffer::{
RingBufferPostCreateInterceptor, RingBufferPostUpdateInterceptor,
RingBufferPreDeleteInterceptor, RingBufferPreUpdateInterceptor,
},
ringbuffer_row::{
RingBufferRowPostDeleteInterceptor, RingBufferRowPostInsertInterceptor,
RingBufferRowPostUpdateInterceptor, RingBufferRowPreDeleteInterceptor,
RingBufferRowPreInsertInterceptor, RingBufferRowPreUpdateInterceptor,
},
role::{RolePostCreateInterceptor, RolePreDeleteInterceptor},
series::{
SeriesPostCreateInterceptor, SeriesPostUpdateInterceptor, SeriesPreDeleteInterceptor,
SeriesPreUpdateInterceptor,
},
series_row::{
SeriesRowPostDeleteInterceptor, SeriesRowPostInsertInterceptor, SeriesRowPostUpdateInterceptor,
SeriesRowPreDeleteInterceptor, SeriesRowPreInsertInterceptor, SeriesRowPreUpdateInterceptor,
},
table::{
TablePostCreateInterceptor, TablePostUpdateInterceptor, TablePreDeleteInterceptor,
TablePreUpdateInterceptor,
},
table_row::{
TableRowPostDeleteInterceptor, TableRowPostInsertInterceptor, TableRowPostUpdateInterceptor,
TableRowPreDeleteInterceptor, TableRowPreInsertInterceptor, TableRowPreUpdateInterceptor,
},
transaction::{PostCommitInterceptor, PreCommitInterceptor},
view::{
ViewPostCreateInterceptor, ViewPostUpdateInterceptor, ViewPreDeleteInterceptor,
ViewPreUpdateInterceptor,
},
},
multi::{RangeScope, transaction::read::MultiReadTransaction},
};
use reifydb_value::{Result, value::datetime::DateTime};
use tracing::instrument;
use crate::{
operator::sink::DurableSink,
timer::{Timer, TimerDue},
transaction::{
ChangeCoordinate, DeferredParams, FlowTransaction,
read::{OperatorStateRangeIter, ReadFrom, read_from},
scope::{OperatorRangeScope, OperatorScope, operator_state_coordinates, operator_state_scope},
substrate::FlowSubstrate,
},
};
pub struct DeferredTransaction {
pub version: CommitVersion,
pub pending: PendingLayers,
pub query: Option<MultiReadTransaction>,
pub state_query: Option<MultiReadTransaction>,
pub catalog: Catalog,
pub interceptors: Interceptors,
pub accumulator: ChangeAccumulator,
pub armed: Vec<TimerDue>,
pub clock: Clock,
pub change_coordinate: Option<ChangeCoordinate>,
pub flow_watermark: Option<DateTime>,
pub source_watermark_cache: HashMap<OperatorId, u64>,
pub row_shape_cache: HashMap<OperatorId, HashMap<EncodedKey, RowShape>>,
pub substrate: FlowSubstrate,
}
impl DeferredTransaction {
#[instrument(name = "flow::transaction::deferred", level = "debug", skip(params), fields(version = params.version.0))]
pub fn new(params: DeferredParams) -> Self {
let mut query = params.query;
if let Some(query) = query.as_mut() {
query.read_as_of_version_inclusive(params.version);
}
Self {
version: params.version,
pending: params.pending,
query,
state_query: params.state_query,
catalog: params.catalog,
interceptors: params.interceptors,
accumulator: ChangeAccumulator::new(),
armed: Vec::new(),
clock: params.clock,
change_coordinate: None,
flow_watermark: None,
source_watermark_cache: HashMap::new(),
row_shape_cache: HashMap::new(),
substrate: params.substrate,
}
}
}
const NO_OPERATOR_STORE: &str = "flow transaction was built without an operator store";
const NO_READ_TRANSACTION: &str = "flow transaction was built without a read transaction";
const UNDECODABLE_KEY: &str = "a key routed to the multi store must decode";
pub(crate) fn deferred_storage_get(
operators: Option<&OperatorStore>,
query: Option<&MultiReadTransaction>,
state_query: Option<&MultiReadTransaction>,
key: &EncodedKey,
) -> Result<Option<EncodedBytes>> {
let route = read_from(key);
if matches!(route, ReadFrom::OperatorState) {
let OperatorScope {
operator,
inner,
} = operator_state_coordinates(key).expect("an OperatorState-routed key must carry an operator id");
return Ok(operators.expect(NO_OPERATOR_STORE).get(operator, &inner).map(EncodedPodRow::into_bytes));
}
let query = match route {
ReadFrom::StateQuery | ReadFrom::OwnedRow => state_query,
ReadFrom::Query => query,
ReadFrom::OperatorState => unreachable!(),
};
let key = TaggedKey::decode(key).expect(UNDECODABLE_KEY);
Ok(query.expect(NO_READ_TRANSACTION).get(&key)?.map(|multi| multi.bytes().clone()))
}
pub(crate) fn deferred_storage_contains(
operators: Option<&OperatorStore>,
query: Option<&MultiReadTransaction>,
state_query: Option<&MultiReadTransaction>,
key: &EncodedKey,
) -> Result<bool> {
let query = match read_from(key) {
ReadFrom::OperatorState => {
let OperatorScope {
operator,
inner,
} = operator_state_coordinates(key)
.expect("an OperatorState-routed key must carry an operator id");
return Ok(operators.expect(NO_OPERATOR_STORE).contains(operator, &inner));
}
ReadFrom::StateQuery | ReadFrom::OwnedRow => state_query,
ReadFrom::Query => query,
};
query.expect(NO_READ_TRANSACTION).contains(&TaggedKey::decode(key).expect(UNDECODABLE_KEY))
}
pub(crate) fn deferred_storage_range<'a>(
operators: Option<&OperatorStore>,
query: Option<&'a MultiReadTransaction>,
state_query: Option<&'a MultiReadTransaction>,
version: CommitVersion,
range: EncodedKeyRange,
scope: RangeScope,
batch_size: usize,
) -> Box<dyn Iterator<Item = Result<MultiVersionRow<TaggedKey>>> + Send + 'a> {
if let Some(OperatorRangeScope {
operator,
inner,
}) = operator_state_scope(&range)
{
return Box::new(OperatorStateRangeIter::new(
operators.expect(NO_OPERATOR_STORE).clone(),
operator,
inner,
batch_size,
version,
));
}
let query = deferred_range_target(query, state_query, &range);
Box::new(query.range_encoded(range, scope, batch_size))
}
fn deferred_range_target<'a>(
query: Option<&'a MultiReadTransaction>,
state_query: Option<&'a MultiReadTransaction>,
range: &EncodedKeyRange,
) -> &'a MultiReadTransaction {
match range.start.as_ref() {
Included(start) | Excluded(start) => match read_from(start) {
ReadFrom::OperatorState => {
unreachable!("operator-state ranges take the operator-state path")
}
ReadFrom::StateQuery | ReadFrom::OwnedRow => state_query,
ReadFrom::Query => query,
},
Unbounded => query,
}
.expect(NO_READ_TRANSACTION)
}
pub(crate) fn deferred_fetch_state_external(
operators: Option<&OperatorStore>,
version: CommitVersion,
keys: Vec<EncodedKey>,
items: &mut Vec<MultiVersionRow<TaggedKey>>,
) {
if keys.is_empty() {
return;
}
let store = operators.expect(NO_OPERATOR_STORE);
let mut grouped: HashMap<OperatorId, Vec<(usize, EncodedKey)>> = HashMap::new();
for (index, encoded_key) in keys.iter().enumerate() {
let OperatorScope {
operator,
inner,
} = operator_state_coordinates(encoded_key).expect("state_get_many keys must carry an operator id");
grouped.entry(operator).or_default().push((index, inner));
}
let mut resolved: Vec<Option<EncodedPodRow>> = vec![None; keys.len()];
for (operator, entries) in grouped {
let inners: Vec<EncodedKey> = entries.iter().map(|(_, inner)| inner.clone()).collect();
for ((index, _), row) in entries.into_iter().zip(store.get_many(operator, &inners)) {
resolved[index] = row;
}
}
for (encoded_key, row) in keys.into_iter().zip(resolved) {
if let Some(row) = row {
items.push(MultiVersionRow {
key: TaggedKey::decode(&encoded_key).expect(UNDECODABLE_KEY),
bytes: row.into_bytes(),
version,
});
}
}
}
impl FlowTransaction for DeferredTransaction {
fn version(&self) -> CommitVersion {
self.version
}
fn clock(&self) -> &Clock {
&self.clock
}
fn catalog(&self) -> &Catalog {
&self.catalog
}
fn query(&self) -> MultiReadTransaction {
self.query.clone().expect(NO_READ_TRANSACTION)
}
fn substrate(&self) -> &FlowSubstrate {
&self.substrate
}
fn pending_layers(&self) -> &PendingLayers {
&self.pending
}
fn pending_layers_mut(&mut self) -> &mut PendingLayers {
&mut self.pending
}
fn accumulator_mut(&mut self) -> &mut ChangeAccumulator {
&mut self.accumulator
}
fn armed_mut(&mut self) -> &mut Vec<TimerDue> {
&mut self.armed
}
fn change_coordinate(&self) -> Option<ChangeCoordinate> {
self.change_coordinate
}
fn set_change_coordinate(&mut self, coordinate: ChangeCoordinate) {
self.change_coordinate = Some(coordinate);
}
fn flow_watermark(&self) -> Option<DateTime> {
self.flow_watermark
}
fn set_flow_watermark(&mut self, watermark: DateTime) {
self.flow_watermark = Some(watermark);
}
fn source_watermark_cache(&mut self) -> &mut HashMap<OperatorId, u64> {
&mut self.source_watermark_cache
}
fn row_shape_cache(&mut self, operator: OperatorId) -> &mut HashMap<EncodedKey, RowShape> {
self.row_shape_cache.entry(operator).or_default()
}
fn run_durable_sink(&mut self, sink: &mut dyn DurableSink, change: Change) -> Result<Change> {
sink.apply(self, change)
}
fn run_durable_sink_timer(&mut self, sink: &mut dyn DurableSink, timer: Timer) -> Result<Option<Change>> {
sink.on_timer(self, timer)
}
fn storage_get(&mut self, key: &EncodedKey) -> Result<Option<EncodedBytes>> {
deferred_storage_get(
self.substrate.operators.as_ref(),
self.query.as_ref(),
self.state_query.as_ref(),
key,
)
}
fn storage_contains(&mut self, key: &EncodedKey) -> Result<bool> {
deferred_storage_contains(
self.substrate.operators.as_ref(),
self.query.as_ref(),
self.state_query.as_ref(),
key,
)
}
fn storage_range(
&mut self,
range: EncodedKeyRange,
scope: RangeScope,
batch_size: usize,
) -> Box<dyn Iterator<Item = Result<MultiVersionRow<TaggedKey>>> + Send + '_> {
deferred_storage_range(
self.substrate.operators.as_ref(),
self.query.as_ref(),
self.state_query.as_ref(),
self.version,
range,
scope,
batch_size,
)
}
fn fetch_state_external(
&mut self,
keys: Vec<EncodedKey>,
items: &mut Vec<MultiVersionRow<TaggedKey>>,
) -> Result<()> {
deferred_fetch_state_external(self.substrate.operators.as_ref(), self.version, keys, items);
Ok(())
}
}
macro_rules! interceptor_method {
($method:ident, $field:ident, $trait_name:ident) => {
fn $method(&mut self) -> &mut Chain<dyn $trait_name + Send + Sync> {
&mut self.interceptors.$field
}
};
}
impl WithInterceptors for DeferredTransaction {
interceptor_method!(table_row_pre_insert_interceptors, table_row_pre_insert, TableRowPreInsertInterceptor);
interceptor_method!(table_row_post_insert_interceptors, table_row_post_insert, TableRowPostInsertInterceptor);
interceptor_method!(table_row_pre_update_interceptors, table_row_pre_update, TableRowPreUpdateInterceptor);
interceptor_method!(table_row_post_update_interceptors, table_row_post_update, TableRowPostUpdateInterceptor);
interceptor_method!(table_row_pre_delete_interceptors, table_row_pre_delete, TableRowPreDeleteInterceptor);
interceptor_method!(table_row_post_delete_interceptors, table_row_post_delete, TableRowPostDeleteInterceptor);
interceptor_method!(
ringbuffer_row_pre_insert_interceptors,
ringbuffer_row_pre_insert,
RingBufferRowPreInsertInterceptor
);
interceptor_method!(
ringbuffer_row_post_insert_interceptors,
ringbuffer_row_post_insert,
RingBufferRowPostInsertInterceptor
);
interceptor_method!(
ringbuffer_row_pre_update_interceptors,
ringbuffer_row_pre_update,
RingBufferRowPreUpdateInterceptor
);
interceptor_method!(
ringbuffer_row_post_update_interceptors,
ringbuffer_row_post_update,
RingBufferRowPostUpdateInterceptor
);
interceptor_method!(
ringbuffer_row_pre_delete_interceptors,
ringbuffer_row_pre_delete,
RingBufferRowPreDeleteInterceptor
);
interceptor_method!(
ringbuffer_row_post_delete_interceptors,
ringbuffer_row_post_delete,
RingBufferRowPostDeleteInterceptor
);
interceptor_method!(pre_commit_interceptors, pre_commit, PreCommitInterceptor);
interceptor_method!(post_commit_interceptors, post_commit, PostCommitInterceptor);
interceptor_method!(namespace_post_create_interceptors, namespace_post_create, NamespacePostCreateInterceptor);
interceptor_method!(namespace_pre_update_interceptors, namespace_pre_update, NamespacePreUpdateInterceptor);
interceptor_method!(namespace_post_update_interceptors, namespace_post_update, NamespacePostUpdateInterceptor);
interceptor_method!(namespace_pre_delete_interceptors, namespace_pre_delete, NamespacePreDeleteInterceptor);
interceptor_method!(table_post_create_interceptors, table_post_create, TablePostCreateInterceptor);
interceptor_method!(table_pre_update_interceptors, table_pre_update, TablePreUpdateInterceptor);
interceptor_method!(table_post_update_interceptors, table_post_update, TablePostUpdateInterceptor);
interceptor_method!(table_pre_delete_interceptors, table_pre_delete, TablePreDeleteInterceptor);
interceptor_method!(view_post_create_interceptors, view_post_create, ViewPostCreateInterceptor);
interceptor_method!(view_pre_update_interceptors, view_pre_update, ViewPreUpdateInterceptor);
interceptor_method!(view_post_update_interceptors, view_post_update, ViewPostUpdateInterceptor);
interceptor_method!(view_pre_delete_interceptors, view_pre_delete, ViewPreDeleteInterceptor);
interceptor_method!(
ringbuffer_post_create_interceptors,
ringbuffer_post_create,
RingBufferPostCreateInterceptor
);
interceptor_method!(ringbuffer_pre_update_interceptors, ringbuffer_pre_update, RingBufferPreUpdateInterceptor);
interceptor_method!(
ringbuffer_post_update_interceptors,
ringbuffer_post_update,
RingBufferPostUpdateInterceptor
);
interceptor_method!(ringbuffer_pre_delete_interceptors, ringbuffer_pre_delete, RingBufferPreDeleteInterceptor);
interceptor_method!(
dictionary_row_pre_insert_interceptors,
dictionary_row_pre_insert,
DictionaryRowPreInsertInterceptor
);
interceptor_method!(
dictionary_row_post_insert_interceptors,
dictionary_row_post_insert,
DictionaryRowPostInsertInterceptor
);
interceptor_method!(
dictionary_row_pre_update_interceptors,
dictionary_row_pre_update,
DictionaryRowPreUpdateInterceptor
);
interceptor_method!(
dictionary_row_post_update_interceptors,
dictionary_row_post_update,
DictionaryRowPostUpdateInterceptor
);
interceptor_method!(
dictionary_row_pre_delete_interceptors,
dictionary_row_pre_delete,
DictionaryRowPreDeleteInterceptor
);
interceptor_method!(
dictionary_row_post_delete_interceptors,
dictionary_row_post_delete,
DictionaryRowPostDeleteInterceptor
);
interceptor_method!(
dictionary_post_create_interceptors,
dictionary_post_create,
DictionaryPostCreateInterceptor
);
interceptor_method!(dictionary_pre_update_interceptors, dictionary_pre_update, DictionaryPreUpdateInterceptor);
interceptor_method!(
dictionary_post_update_interceptors,
dictionary_post_update,
DictionaryPostUpdateInterceptor
);
interceptor_method!(dictionary_pre_delete_interceptors, dictionary_pre_delete, DictionaryPreDeleteInterceptor);
interceptor_method!(series_row_pre_insert_interceptors, series_row_pre_insert, SeriesRowPreInsertInterceptor);
interceptor_method!(
series_row_post_insert_interceptors,
series_row_post_insert,
SeriesRowPostInsertInterceptor
);
interceptor_method!(series_row_pre_update_interceptors, series_row_pre_update, SeriesRowPreUpdateInterceptor);
interceptor_method!(
series_row_post_update_interceptors,
series_row_post_update,
SeriesRowPostUpdateInterceptor
);
interceptor_method!(series_row_pre_delete_interceptors, series_row_pre_delete, SeriesRowPreDeleteInterceptor);
interceptor_method!(
series_row_post_delete_interceptors,
series_row_post_delete,
SeriesRowPostDeleteInterceptor
);
interceptor_method!(series_post_create_interceptors, series_post_create, SeriesPostCreateInterceptor);
interceptor_method!(series_pre_update_interceptors, series_pre_update, SeriesPreUpdateInterceptor);
interceptor_method!(series_post_update_interceptors, series_post_update, SeriesPostUpdateInterceptor);
interceptor_method!(series_pre_delete_interceptors, series_pre_delete, SeriesPreDeleteInterceptor);
interceptor_method!(identity_post_create_interceptors, identity_post_create, IdentityPostCreateInterceptor);
interceptor_method!(identity_pre_delete_interceptors, identity_pre_delete, IdentityPreDeleteInterceptor);
interceptor_method!(
identity_attribute_post_create_interceptors,
identity_attribute_post_create,
IdentityAttributePostCreateInterceptor
);
interceptor_method!(
identity_attribute_pre_delete_interceptors,
identity_attribute_pre_delete,
IdentityAttributePreDeleteInterceptor
);
interceptor_method!(
identity_attribute_value_post_create_interceptors,
identity_attribute_value_post_create,
IdentityAttributeValuePostCreateInterceptor
);
interceptor_method!(
identity_attribute_value_pre_delete_interceptors,
identity_attribute_value_pre_delete,
IdentityAttributeValuePreDeleteInterceptor
);
interceptor_method!(role_post_create_interceptors, role_post_create, RolePostCreateInterceptor);
interceptor_method!(role_pre_delete_interceptors, role_pre_delete, RolePreDeleteInterceptor);
interceptor_method!(
granted_role_post_create_interceptors,
granted_role_post_create,
GrantedRolePostCreateInterceptor
);
interceptor_method!(
granted_role_pre_delete_interceptors,
granted_role_pre_delete,
GrantedRolePreDeleteInterceptor
);
interceptor_method!(
authentication_post_create_interceptors,
authentication_post_create,
AuthenticationPostCreateInterceptor
);
interceptor_method!(
authentication_pre_delete_interceptors,
authentication_pre_delete,
AuthenticationPreDeleteInterceptor
);
}