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 use callback::{BarrierOutcome, PipelineCallback, SkipReason, SourceRegistration};
12pub use config::PipelineConfig;
13pub use streaming_coordinator::StreamingCoordinator;
14
15use arrow::datatypes::SchemaRef;
16use laminar_sql::parser::EmitClause;
17use laminar_sql::translator::{JoinOperatorConfig, OrderOperatorConfig, WindowOperatorConfig};
18
19/// Control message sent from `LaminarDB` DDL handlers to the running
20/// `StreamingCoordinator` for live schema changes (add/drop streams).
21#[derive(Debug)]
22#[allow(clippy::large_enum_variant)]
23pub enum ControlMsg {
24    /// Add a new streaming query to the running pipeline.
25    AddStream {
26        /// Stream name.
27        name: String,
28        /// SQL query text.
29        sql: String,
30        /// EMIT clause (e.g., `OnWindowClose`).
31        emit_clause: Option<EmitClause>,
32        /// Window configuration (tumbling, hopping, session).
33        window_config: Option<WindowOperatorConfig>,
34        /// ORDER BY configuration.
35        order_config: Option<OrderOperatorConfig>,
36        /// Per-step join configs from the planner (left-deep).
37        join_config: Option<Vec<JoinOperatorConfig>>,
38    },
39    /// Remove a streaming query from the running pipeline.
40    DropStream {
41        /// Stream name to remove.
42        name: String,
43    },
44    /// Register a new source schema (for queries that depend on a source
45    /// created after `start()`).
46    AddSourceSchema {
47        /// Source name.
48        name: String,
49        /// Arrow schema.
50        schema: SchemaRef,
51    },
52}
laminar_db/pipeline/mod.rs

laminar_db/pipeline/
mod.rs
