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.

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 async fn adopt_assignment_snapshot( &self, snapshot: AssignmentSnapshot, ) -> SnapshotAdoption

Adopt a new vnode assignment snapshot atomically across the registry, state-backend fence, and coordinator, then rehydrate committed state for any vnodes this node newly acquired. Idempotent for versions ≤ the current registry version.

Rehydration (the durable read of newly-acquired vnodes’ partials) runs after the coordinator lock is released so a slow object store can’t stall the checkpoint cadence; the recovered bytes are staged in Self::rehydrated_vnode_state for operators to drain.

pub fn rehydrated_vnode_state(&self) -> HashMap<u32, RehydratedVnode>

Snapshot of committed per-vnode state staged for newly-acquired vnodes during the most recent rebalance adoptions. Keyed by vnode. Drained by operators that adopt the per-vnode state backend; exposed for inspection and tests.

pub fn session_context(&self) -> &SessionContext

The underlying DataFusion SessionContext.

pub fn builder() -> LaminarDbBuilder

Returns 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 set_session_property(&self, key: &str, value: &str)

Set a session property. Keys are lowercased.

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 subscribe_raw( &self, name: &str, ) -> Result<Subscription<ArrowRecord>, DbError>

Subscribe to a named stream’s output.

Errors

Returns DbError::StreamNotFound if the stream doesn’t exist.

pub fn lookup_subscription_schema( &self, name: &str, ) -> Option<(SchemaRef, bool)>

Schema a SUBSCRIBE against name would emit, plus a filterable flag that’s false only when the schema came from a StreamEntry sink placeholder (Schema::empty) — a WHERE clause can’t compile against that and must be rejected.

Lookup order: MV registry → start()-resolved stream output → StreamEntry sink (placeholder). A bare SOURCE is intentionally not resolved: only streams/MVs defined over it publish to the registry, so subscribing to the source directly would block forever — it falls through to None, which the caller surfaces as StreamNotFound.

pub async fn open_subscription( &self, name: &str, filter_sql: Option<&str>, start: SubscribeStart, ) -> Result<SubscriptionPortal, DbError>

Open a SUBSCRIBE portal against a named MV or resolved stream. A bare SOURCE is not subscribable (surfaced as StreamNotFound). filter_sql is rejected on streams (their schema is opaque).

Errors

StreamNotFound for unknown name; Pipeline for subscriber-cap or filter-compile failures; InvalidOperation when AsOfEpoch(n) is requested but n is no longer retained.

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. Returns DbError::SchemaMismatch if the Rust type’s schema does not match the source’s SQL schema.

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 materialized_views(&self) -> Vec<MaterializedViewInfo>

List all registered materialized views with their SQL and state.

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 uptime(&self) -> Duration

Time elapsed since the database was created.

pub fn set_engine_metrics(&self, metrics: Arc<EngineMetrics>)

Inject prometheus engine metrics. Called once at startup before start().

pub fn set_prometheus_registry(&self, registry: Arc<Registry>)

Inject a shared Prometheus registry for connector-level metrics.

Called once at startup, after the registry is constructed but before start(). Connectors created after this call will register their metrics on this registry so they appear in the scrape output.

pub fn engine_metrics(&self) -> Option<Arc<EngineMetrics>>

Get the engine metrics if set.

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 prometheus engine metrics 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 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 checkpoint_stats_nonblocking(&self) -> Option<CheckpointStats>

Returns checkpoint statistics if available (non-blocking).

Uses try_lock() on the coordinator mutex. Returns None if the coordinator is not initialized or the lock is contended.

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

Returns true if the database has been shut down.

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

Start the streaming pipeline. Idempotent if already running.

Activates registered connectors and spawns the background processing task. On failure the instance is unwound back to Created so the caller can retry after fixing the offending config.

Errors

Returns an error if the pipeline cannot be started.

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

Shut down the streaming pipeline gracefully. Idempotent.

Errors

Returns an error if shutdown encounters an error.

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

Stop the streaming pipeline so it can be restarted.

Errors

Returns DbError::InvalidOperation if the pipeline is still starting or the coordinator does not exit within the stop timeout.

impl LaminarDB

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

Build a SHOW CHECKPOINT STATUS metadata result. Public so the server’s GET /api/v1/cluster/checkpoints endpoint can surface it.

Errors

Returns DbError::Checkpoint if the metadata batch cannot be assembled from the latest checkpoint.

Trait Implementations

impl Debug for LaminarDB

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

Formats the value using the given formatter.