Struct LaminarDB 

pub struct LaminarDB { /* private fields */ }

The main LaminarDB database handle.

Unified interface for SQL execution, data ingestion, and result consumption.

Implementations

impl LaminarDB

pub fn open() -> Result<Arc<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<Arc<Self>, DbError>

Create with custom configuration.

Errors

Returns DbError if DataFusion context creation fails.

pub async fn adopt_assignment_snapshot( &self, snapshot: AssignmentSnapshot, deadline: Instant, ) -> Result<SnapshotAdoption, DbError>

Adopt one authority-audited assignment, staging exact committed frames for live vnode acquisition. Startup recovery still restores the complete graph before launch.

Errors

Returns a checkpoint error when authority, state, or the deadline changes before atomic publication. The assignment is not published in that case.

pub async fn collect_local_table( &self, name: &str, ) -> Result<Vec<RecordBatch>, DbError>

Collect this node’s physical table slice without invoking distributed fan-out.

This is intended for cluster diagnostics and placement validation. It deliberately returns immutable batches rather than exposing the mutable DataFusion catalog.

Errors

Returns an error when the local table cannot be resolved, planned, or collected.

pub fn builder() -> LaminarDbBuilder

Returns a fluent builder for constructing a LaminarDB.

pub fn connector_registry(&self) -> &ConnectorRegistry

Returns the frozen connector registry for factory lookup and deployment introspection. Custom factories must be registered through LaminarDbBuilder::register_connector.

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

Execute a SQL statement.

Errors

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

pub async fn execute_cluster_bootstrap_batch( &self, sql: &[String], ) -> Result<Vec<ExecuteResult>, DbError>

Apply and durably seal the complete startup catalog as one immutable batch. Existing sealed catalogs accept exact replay/no-op definitions only.

Errors

Returns an error for a partial or divergent bootstrap, unsafe catalog mutation, lost leader authority, or manifest sealing failure. Local creates are rolled back on every error.

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

Apply one startup catalog definition and seal it as the complete inventory. Prefer Self::execute_cluster_bootstrap_batch for server configuration.

Errors

Returns an error when the definition is invalid or the catalog batch cannot be sealed.

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

Return a session property value.

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

Return all session properties.

pub fn set_session_property(&self, key: &str, value: &str)

Set a session property (keys are lowercased).

pub async 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 object or its output schema is unresolved.

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

Schema a SUBSCRIBE against name would emit. A stream is visible here only after its physical output schema has been resolved. Bare sources and unresolved streams are not subscribable.

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

Errors

StreamNotFound for unknown name; Subscription for unsupported cluster plans or durable replay failures; Pipeline for subscriber-cap or filter-compile failures; InvalidOperation when a local AsOfEpoch(n) is not committed or retained.

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

Get a typed handle for pushing data to a registered source.

Errors

Returns DbError::SourceNotFound if the source is not registered. Returns DbError::SchemaMismatch if the Rust type’s schema doesn’t match.

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

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

List registered sinks.

pub fn materialized_views(&self) -> Vec<MaterializedViewInfo>

List registered materialized views.

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

List registered streams.

pub fn pipeline_topology(&self) -> PipelineTopology

Build the pipeline topology graph (nodes + edges) from registered sources, streams, and sinks.

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

List active queries.

pub fn is_checkpoint_enabled(&self) -> bool

Returns whether streaming checkpointing is enabled.

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

Trigger a checkpoint that persists source offsets, sink positions, and operator state.

Errors

Returns DbError::Checkpoint if checkpointing is disabled, no live manual-checkpoint route exists, cluster leadership cannot be resolved, or the checkpoint fails.

pub async fn checkpoint_with_timeout( &self, timeout: Duration, ) -> Result<CheckpointResult, DbError>

Trigger one manual checkpoint within a caller-provided relative budget.

The budget is capped by the configured checkpoint timeout. It bounds admission and attempt work; once an exact attempt is reserved, this call waits for terminal cleanup even if that requires the coordinator’s private cleanup budget.

Errors

Returns DbError::Checkpoint under the same conditions as Self::checkpoint, or when the caller-provided budget cannot be represented as an absolute deadline.

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

Returns checkpoint performance statistics, or None if the coordinator is not 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>, ) -> Result<(), DbError>

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.

Errors

Returns an error after the database leaves Created or when a registry was already installed.

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

Get the engine metrics if set.

pub fn cluster_subscription_output_health( &self, ) -> Option<ClusterSubscriptionOutputHealth>

Read process-local committed subscription output-log health without storage I/O.

pub fn checkpoint_barrier_timing_snapshot( &self, expected_process: Option<LocalProcessAuthorityIdentity>, after_sequence: u64, limit: usize, ) -> Result<CheckpointBarrierTimingPage, CheckpointBarrierTimingReadError>

Read one bounded, process-bound page of local cluster checkpoint barrier timings.

This is an in-memory diagnostic read. It performs no checkpoint, state-backend, network, filesystem, or object-store operation. A continuation cursor is the pair of expected_process and after_sequence; callers must retain both from the prior page. Snapshot counters may advance immediately after this call, so certification must finish with a quiescent repeated read rather than treating one page as a stable cut.

Errors

Returns an error when a continuation omits or mismatches its process identity, live process authority cannot be sampled without waiting, authority changes during the read, or the bounded ledger snapshot rejects its cursor/page request.

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

Get the current pipeline state as a string.

pub fn last_fault(&self) -> Option<String>

The last runtime-fault or terminal resource-exhaustion reason. Populated when state is Faulted; cleared on a clean start.

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 is_closed(&self) -> bool

Returns true if the database has been shut down.

pub fn close(&self)

Fence new work and wake the runtime so it can shut down.

pub fn enable_supervision(self: &Arc<Self>)

Enable auto-restart from the last checkpoint on a fault. Without it, a fault parks in Faulted for manual restart (the embedded default).

pub fn fence_cluster_startup(&self)

Keep clustered source intake closed while startup restores state and certifies assignment. Call before Self::start; Self::finish_cluster_startup is the only startup path that opens the gate.

pub fn revoke_cluster_authority(&self)

Permanently withdraw this process’s clustered data-plane authority after lease loss.

pub fn cluster_intake_fenced(&self) -> bool

Whether clustered source and shuffle intake is still fenced.

pub fn coordinated_recovery_in_progress(&self) -> bool

Whether a clustered runtime is held by coordinated recovery lifecycle authority.

impl LaminarDB

pub async fn finish_cluster_startup( &self, deadline: Instant, ) -> Result<ClusterStartupDisposition, DbError>

Finish clustered startup after the exact assignment fence is available.

Fresh nodes open intake only when no durable recovery round exists. A process that restored local state, or one that observes an active/stale round, remains fenced and requests a full coordinated rewind.

Errors

Returns a checkpoint error when the local assignment has not been certified. Intake remains closed when recovery authority is unavailable so the monitor can retry without admitting records.

pub async fn prepare_cluster_startup_recovery_generation( &self, deadline: Instant, ) -> Result<(), DbError>

Import the last recovery generation that reached an irrevocable cluster Release before a fresh pipeline starts. The allocation high-watermark is deliberately not used: a driver can reserve a generation and fail before any data plane rewinds to it.

Errors

Returns an error when durable recovery authority is unavailable, the process lease is lost, or an already-active transport conflicts with the committed terminal.

pub fn enable_coordinated_recovery(self: &Arc<Self>) -> Result<(), DbError>

Start the database-owned per-node recovery supervisor once. Coordinated recovery is the only cluster fault path — a local-only restart rewinds one node while peers advance, an inconsistent cut.

Errors

Returns an error if the database-owned control runtime cannot be initialized.

impl LaminarDB

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

Shut down the streaming pipeline gracefully. Idempotent.

Errors

Returns Err if the watcher task panicked.

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 start(self: &Arc<Self>) -> Result<(), DbError>

Start the streaming pipeline. Idempotent if already running. On failure (or recovering from Faulted) it rebuilds from the surviving catalog.

Errors

Returns an error if the pipeline cannot be started.

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.

Trait Implementations

impl Debug for LaminarDB

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

Formats the value using the given formatter.

impl Drop for LaminarDB

fn drop(&mut self)

Executes the destructor for this type.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.