Struct LaminarDB 

pub struct LaminarDB { /* private fields */ }

The main LaminarDB database handle.

Provides a unified interface for SQL execution, data ingestion, and result consumption. All streaming infrastructure (sources, sinks, channels, subscriptions) is managed internally.

Example

use laminar_db::LaminarDB;

let db = LaminarDB::open()?;

db.execute("CREATE SOURCE trades (
    symbol VARCHAR, price DOUBLE, ts BIGINT,
    WATERMARK FOR ts AS ts - INTERVAL '1' SECOND
)").await?;

let query = db.execute("SELECT symbol, AVG(price) FROM trades
    GROUP BY symbol, TUMBLE(ts, INTERVAL '1' MINUTE)
").await?;

Implementations

impl LaminarDB

pub fn open() -> Result<Self, DbError>

Create an embedded in-memory database with default settings.

Errors

Returns DbError if DataFusion context creation fails.

pub fn open_with_config(config: LaminarConfig) -> Result<Self, DbError>

Create with custom configuration.

Errors

Returns DbError if DataFusion context creation fails.

pub fn builder() -> LaminarDbBuilder

Get a fluent builder for constructing a LaminarDB.

pub fn connector_registry(&self) -> &ConnectorRegistry

Returns the connector registry for registering custom connectors.

Use this to register user-defined source/sink connectors before calling start().

pub async fn register_delta_table( &self, name: &str, table_uri: &str, storage_options: HashMap<String, String>, ) -> Result<(), DbError>

Registers a Delta Lake table as a DataFusion TableProvider.

After registration, the table can be queried via SQL:

SELECT * FROM my_delta_table WHERE id > 100

Arguments

• name - SQL table name (e.g., "trades")
• table_uri - Path to the Delta Lake table (local, s3://, az://, gs://)
• storage_options - Storage credentials and configuration

Errors

Returns DbError if the table cannot be opened or registered.

pub async fn execute(&self, sql: &str) -> Result<ExecuteResult, DbError>

Execute a SQL statement.

Supports:

• CREATE SOURCE / CREATE SINK — registers sources and sinks
• DROP SOURCE / DROP SINK — removes sources and sinks
• SHOW SOURCES / SHOW SINKS / SHOW QUERIES — list registered objects
• DESCRIBE source_name — show source schema
• SELECT ... — execute a streaming query
• INSERT INTO source_name VALUES (...) — insert data
• CREATE MATERIALIZED VIEW — create a streaming materialized view
• EXPLAIN SELECT ... — show query plan

Errors

Returns DbError if SQL parsing, planning, or execution fails.

pub fn get_session_property(&self, key: &str) -> Option<String>

Get a session property value.

pub fn session_properties(&self) -> HashMap<String, String>

Get all session properties.

pub fn subscribe<T: FromBatch>( &self, name: &str, ) -> Result<TypedSubscription<T>, DbError>

Subscribe to a named stream or materialized view.

Errors

Returns DbError::StreamNotFound if the stream is not registered.

pub fn source<T: Record>(&self, name: &str) -> Result<SourceHandle<T>, DbError>

Get a typed source handle for pushing data.

The source must have been created via CREATE SOURCE.

Errors

Returns DbError::SourceNotFound if the source is not registered.

pub fn source_untyped(&self, name: &str) -> Result<UntypedSourceHandle, DbError>

Get an untyped source handle for pushing RecordBatch data.

Errors

Returns DbError::SourceNotFound if the source is not registered.

pub fn sources(&self) -> Vec<SourceInfo>

List all registered sources.

pub fn sinks(&self) -> Vec<SinkInfo>

List all registered sinks.

pub fn streams(&self) -> Vec<StreamInfo>

List all registered streams with their SQL definitions.

pub fn pipeline_topology(&self) -> PipelineTopology

Build the pipeline topology graph from registered sources, streams, and sinks.

Returns a PipelineTopology with nodes for every source, stream, and sink, plus edges derived from stream SQL FROM references and sink input fields.

pub fn queries(&self) -> Vec<QueryInfo>

List all active queries.

pub fn is_checkpoint_enabled(&self) -> bool

Returns whether streaming checkpointing is enabled.

pub fn checkpoint_store(&self) -> Option<Box<dyn CheckpointStore>>

Returns a checkpoint store instance, if checkpointing is configured.

Returns an ObjectStoreCheckpointStore when object_store_url is set, otherwise a FileSystemCheckpointStore.

pub async fn checkpoint(&self) -> Result<CheckpointResult, DbError>

Triggers a streaming checkpoint that persists source offsets, sink positions, and operator state to disk via the CheckpointCoordinator.

Returns the checkpoint result on success, including the checkpoint ID, epoch, and duration.

Errors

Returns DbError::Checkpoint if checkpointing is not enabled, the coordinator has not been initialized (call start() first), or the checkpoint operation fails.

pub async fn checkpoint_stats(&self) -> Option<CheckpointStats>

Returns checkpoint performance statistics.

Returns None if the checkpoint coordinator has not been initialized.

impl LaminarDB

pub fn pipeline_state(&self) -> &'static str

Get the current pipeline state as a string.

pub fn metrics(&self) -> PipelineMetrics

Get a pipeline-wide metrics snapshot.

Reads shared atomic counters and catalog sizes to produce a point-in-time view of pipeline health.

pub fn source_metrics(&self, name: &str) -> Option<SourceMetrics>

Get metrics for a single source by name.

pub fn all_source_metrics(&self) -> Vec<SourceMetrics>

Get metrics for all registered sources.

pub fn stream_metrics(&self, name: &str) -> Option<StreamMetrics>

Get metrics for a single stream by name.

pub fn all_stream_metrics(&self) -> Vec<StreamMetrics>

Get metrics for all registered streams.

pub fn total_events_processed(&self) -> u64

Get the total number of events processed (ingested + emitted).

pub fn counters(&self) -> &Arc<PipelineCounters>

Get a reference to the shared pipeline counters.

Useful for external code that needs to read counters directly (e.g. a TUI dashboard polling at high frequency).

pub fn pipeline_watermark(&self) -> i64

Returns the global pipeline watermark (minimum across all source watermarks).

Returns i64::MIN if no watermark-enabled sources exist or no events have been processed.

pub fn cancel_query(&self, query_id: u64) -> Result<(), DbError>

Cancel a running query by ID.

Marks the query as inactive in the catalog. Future subscription polls for this query will receive no more data.

Errors

Returns DbError if the query is not found.

pub fn source_count(&self) -> usize

Get the number of registered sources.

pub fn sink_count(&self) -> usize

Get the number of registered sinks.

pub fn active_query_count(&self) -> usize

Get the number of active queries.

impl LaminarDB

pub fn close(&self)

Shut down the database gracefully.

pub fn is_closed(&self) -> bool

Check if the database is shut down.

pub async fn start(&self) -> Result<(), DbError>

Start the streaming pipeline.

Activates all registered connectors and begins processing. This is a no-op if the pipeline is already running.

When the kafka feature is enabled and Kafka sources/sinks are registered, this builds KafkaSource/KafkaSink instances and spawns a background task that polls sources, executes stream queries via DataFusion, and writes results to sinks.

In embedded (in-memory) mode, this simply transitions to Running.

Errors

Returns an error if the pipeline cannot be started.

pub async fn shutdown(&self) -> Result<(), DbError>

Shut down the streaming pipeline gracefully.

Signals the processing loop to stop, waits for it to complete (with a timeout), then transitions to Stopped. This is idempotent – calling it multiple times is safe.

Errors

Returns an error if shutdown encounters an error.

Trait Implementations

impl Debug for LaminarDB

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.