laminar_db/pipeline/
mod.rs

1//! Streaming connector pipeline. Each source connector runs as a tokio task
2//! pushing batches via crossfire mpsc to the `StreamingCoordinator`, which
3//! drives SQL execution cycles, routes results to sinks, and manages
4//! checkpoint barriers. See the `streaming_coordinator` submodule for the
5//! runtime topology.
6
7pub mod callback;
8pub mod config;
9pub mod streaming_coordinator;
10
11pub(crate) use callback::CheckpointCompletion;
12pub use callback::{
13    BarrierOutcome, CheckpointAssignmentAdmission, CheckpointControlOutcome, CycleError,
14    CycleOutcome, PipelineCallback, SkipReason, SourceRegistration,
15};
16pub use config::{CheckpointSchedule, PipelineConfig};
17pub use streaming_coordinator::{ExitReason, StreamingCoordinator, StreamingCoordinatorRuntime};
18
19use laminar_sql::parser::EmitClause;
20use laminar_sql::translator::{JoinOperatorConfig, OrderOperatorConfig, WindowOperatorConfig};
21
22const CONTROL_MUTATION_PENDING: u8 = 0;
23const CONTROL_MUTATION_APPLIED: u8 = 1;
24const CONTROL_MUTATION_CANCELLED: u8 = 2;
25
26#[derive(Clone, Copy, Debug, Eq, PartialEq)]
27pub(crate) enum ControlMutationState {
28    Pending,
29    Applied,
30    Cancelled,
31}
32
33#[derive(Debug)]
34pub(crate) struct ControlMutation {
35    state: std::sync::atomic::AtomicU8,
36}
37
38impl ControlMutation {
39    pub(crate) fn new() -> Self {
40        Self {
41            state: std::sync::atomic::AtomicU8::new(CONTROL_MUTATION_PENDING),
42        }
43    }
44
45    pub(crate) fn try_apply(&self) -> bool {
46        self.state
47            .compare_exchange(
48                CONTROL_MUTATION_PENDING,
49                CONTROL_MUTATION_APPLIED,
50                std::sync::atomic::Ordering::AcqRel,
51                std::sync::atomic::Ordering::Acquire,
52            )
53            .is_ok()
54    }
55
56    pub(crate) fn cancel(&self) -> ControlMutationState {
57        match self.state.compare_exchange(
58            CONTROL_MUTATION_PENDING,
59            CONTROL_MUTATION_CANCELLED,
60            std::sync::atomic::Ordering::AcqRel,
61            std::sync::atomic::Ordering::Acquire,
62        ) {
63            Ok(_) | Err(CONTROL_MUTATION_CANCELLED) => ControlMutationState::Cancelled,
64            Err(CONTROL_MUTATION_APPLIED) => ControlMutationState::Applied,
65            Err(_) => unreachable!("control mutation contains an invalid state"),
66        }
67    }
68
69    pub(crate) fn state(&self) -> ControlMutationState {
70        match self.state.load(std::sync::atomic::Ordering::Acquire) {
71            CONTROL_MUTATION_PENDING => ControlMutationState::Pending,
72            CONTROL_MUTATION_APPLIED => ControlMutationState::Applied,
73            CONTROL_MUTATION_CANCELLED => ControlMutationState::Cancelled,
74            _ => unreachable!("control mutation contains an invalid state"),
75        }
76    }
77}
78
79/// Opaque live-DDL message used by the streaming coordinator.
80#[derive(Debug)]
81pub struct ControlMsg(ControlMsgKind);
82
83#[derive(Debug)]
84#[allow(clippy::large_enum_variant)]
85pub(crate) enum ControlMsgKind {
86    AddStream {
87        name: String,
88        sql: String,
89        emit_clause: Option<EmitClause>,
90        window_config: Option<WindowOperatorConfig>,
91        order_config: Option<OrderOperatorConfig>,
92        join_config: Option<Vec<JoinOperatorConfig>>,
93        incremental: bool,
94        reply: tokio::sync::oneshot::Sender<Result<(), crate::error::DbError>>,
95        mutation: std::sync::Arc<ControlMutation>,
96    },
97    DropStreams {
98        names: Vec<String>,
99        reply: tokio::sync::oneshot::Sender<Result<(), crate::error::DbError>>,
100        mutation: std::sync::Arc<ControlMutation>,
101    },
102}
103
104impl ControlMsg {
105    pub(crate) fn add_stream(
106        name: String,
107        sql: String,
108        emit_clause: Option<EmitClause>,
109        window_config: Option<WindowOperatorConfig>,
110        order_config: Option<OrderOperatorConfig>,
111        join_config: Option<Vec<JoinOperatorConfig>>,
112        incremental: bool,
113        reply: tokio::sync::oneshot::Sender<Result<(), crate::error::DbError>>,
114        mutation: std::sync::Arc<ControlMutation>,
115    ) -> Self {
116        Self(ControlMsgKind::AddStream {
117            name,
118            sql,
119            emit_clause,
120            window_config,
121            order_config,
122            join_config,
123            incremental,
124            reply,
125            mutation,
126        })
127    }
128
129    pub(crate) fn drop_streams(
130        names: Vec<String>,
131        reply: tokio::sync::oneshot::Sender<Result<(), crate::error::DbError>>,
132        mutation: std::sync::Arc<ControlMutation>,
133    ) -> Self {
134        Self(ControlMsgKind::DropStreams {
135            names,
136            reply,
137            mutation,
138        })
139    }
140
141    pub(crate) fn into_kind(self) -> ControlMsgKind {
142        self.0
143    }
144}
laminar_db/pipeline/mod.rs

laminar_db/pipeline/
mod.rs
