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)

Adopt a new assignment snapshot: update the registry, backend fence, and coordinator under the coordinator mutex so the change lands strictly between checkpoints. Idempotent for versions at or below the current registry version.

pub fn session_context(&self) -> &SessionContext

Underlying DataFusion SessionContext. Primarily for tests that need to compile SQL through the same session the engine uses.

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 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 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 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.

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. On failure, the instance is unwound back to STATE_CREATED so the caller can retry after fixing the offending config.

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.