mod dispatch;
pub mod frontier;
mod lifecycle;
mod process;
pub mod register;
mod tick;
mod timers;
use std::{
collections::{BTreeMap, BTreeSet},
sync::Arc,
};
use reifydb_catalog::catalog::Catalog;
use reifydb_core::interface::catalog::{
flow::{FlowId, OperatorId},
id::{TableId, ViewId},
object::ObjectId,
};
use reifydb_routine_abi::registry::Routines;
use reifydb_rql::flow::{
analyzer::{FlowDependencyGraph, FlowGraphAnalyzer},
flow::FlowDag,
};
use reifydb_runtime::context::{RuntimeContext, clock::Clock};
use tracing::instrument;
use crate::{
operator::{
BoxedHostOperator, HostOperator, metrics::OperatorSampleRegistry, provider::OperatorProvider,
sink::BoxedDurableSink,
},
timer::registry::TimerRegistry,
transaction::substrate::FlowSubstrate,
};
pub const COMPLETENESS_OBJECT: ObjectId = ObjectId::Table(TableId::SOURCE_COMPLETENESS);
pub struct FlowEngineInner {
pub(crate) catalog: Catalog,
pub(crate) routines: Routines,
pub(crate) operators: BTreeMap<(FlowId, OperatorId), BoxedHostOperator>,
pub(crate) durable_sinks: BTreeMap<(FlowId, OperatorId), BoxedDurableSink>,
pub(crate) flows: BTreeMap<FlowId, FlowDag>,
pub(crate) sources: BTreeMap<ObjectId, Vec<(FlowId, OperatorId)>>,
pub(crate) sinks: BTreeMap<ObjectId, Vec<(FlowId, OperatorId)>>,
pub(crate) analyzer: FlowGraphAnalyzer,
pub(crate) runtime_context: RuntimeContext,
pub(crate) operator_provider: Arc<dyn OperatorProvider>,
pub(crate) substrate: FlowSubstrate,
pub(crate) operator_samples: OperatorSampleRegistry,
pub(crate) timers: TimerRegistry,
}
impl FlowEngineInner {
#[instrument(
name = "flow::engine::new",
level = "debug",
skip(catalog, routines, runtime_context, operator_provider, substrate, operator_samples)
)]
#[allow(clippy::too_many_arguments)]
pub fn new(
catalog: Catalog,
routines: Routines,
runtime_context: RuntimeContext,
operator_provider: Arc<dyn OperatorProvider>,
substrate: FlowSubstrate,
operator_samples: OperatorSampleRegistry,
) -> Self {
Self {
catalog,
routines,
operators: BTreeMap::new(),
durable_sinks: BTreeMap::new(),
flows: BTreeMap::new(),
sources: BTreeMap::new(),
sinks: BTreeMap::new(),
analyzer: FlowGraphAnalyzer::new(),
runtime_context,
operator_provider,
substrate,
operator_samples,
timers: TimerRegistry::default(),
}
}
#[instrument(name = "flow::engine::sample", level = "debug", skip_all)]
pub fn sample_operators(&self) {
for ((_, operator_id), operator) in &self.operators {
if let Some(sample) = operator.sample() {
self.operator_samples.record(*operator_id, sample);
}
}
}
pub fn forget_operator_samples(&self) {
for (_, operator) in self.operators.keys() {
self.operator_samples.forget(*operator);
}
}
pub fn clock(&self) -> &Clock {
&self.runtime_context.clock
}
pub fn substrate(&self) -> &FlowSubstrate {
&self.substrate
}
pub fn operator(&self, flow_id: FlowId, operator_id: OperatorId) -> Option<&dyn HostOperator> {
self.operators.get(&(flow_id, operator_id)).map(|operator| &**operator)
}
pub fn insert_operator(&mut self, flow_id: FlowId, operator_id: OperatorId, operator: BoxedHostOperator) {
self.operators.insert((flow_id, operator_id), operator);
}
pub fn flow_by_id(&self, flow_id: FlowId) -> Option<FlowDag> {
self.flows.get(&flow_id).cloned()
}
pub fn has_sources(&self) -> bool {
!self.sources.is_empty()
}
pub fn flows_for_source_object(&self, object: ObjectId) -> Option<Vec<(FlowId, OperatorId)>> {
self.sources.get(&object).cloned()
}
pub fn flow_ids(&self) -> BTreeSet<FlowId> {
self.flows.keys().copied().collect()
}
pub fn get_dependency_graph(&self) -> FlowDependencyGraph {
self.analyzer.get_dependency_graph().clone()
}
pub fn get_flow_producing_view(&self, view_id: ViewId) -> Option<FlowId> {
let dependency_graph = self.analyzer.get_dependency_graph();
self.analyzer.get_flow_producing_view(dependency_graph, view_id)
}
}