Struct LaminarDbBuilder 

pub struct LaminarDbBuilder { /* private fields */ }

Fluent builder for constructing a LaminarDB instance.

Example

let db = LaminarDB::builder()
    .config_var("KAFKA_BROKERS", "localhost:9092")
    .buffer_size(131072)
    .build()
    .await?;

Implementations

impl LaminarDbBuilder

pub fn new() -> Self

Create a new builder with default settings.

pub fn target_partitions(self, n: usize) -> Self

Override target_partitions; requires a distributed-aware physical optimizer rule to replace RepartitionExec.

pub fn physical_optimizer_rule( self, rule: Arc<dyn PhysicalOptimizerRule + Send + Sync>, ) -> Self

Register an additional PhysicalOptimizerRule on the session state.

pub fn state_backend(self, backend: Arc<dyn StateBackend>) -> Self

Install a state backend. Must be paired with Self::vnode_registry — build() rejects a half-set configuration.

pub fn vnode_registry(self, registry: Arc<VnodeRegistry>) -> Self

Install a vnode registry. Must be paired with Self::state_backend — build() rejects a half-set configuration.

pub fn cluster_controller(self, controller: Arc<ClusterController>) -> Self

Install a cluster control facade. Activates cluster-mode checkpoint / shuffle semantics inside the engine. Called from laminar-server’s cluster startup path after discovery has converged.

pub fn shuffle_sender(self, sender: Arc<ShuffleSender>) -> Self

Install the outbound shuffle handle used by cluster-mode streaming aggregates. Rows whose group key hashes to a remote vnode are shipped through this sender. Pair with Self::shuffle_receiver; either alone is a no-op.

pub fn shuffle_receiver(self, receiver: Arc<ShuffleReceiver>) -> Self

Install the inbound shuffle handle used by cluster-mode streaming aggregates. Remote partial-aggregate rows arrive here and are drained into the local accumulator each cycle.

pub fn decision_store(self, store: Arc<CheckpointDecisionStore>) -> Self

Install the commit-marker store for cross-instance 2PC.

pub fn assignment_snapshot_store( self, store: Arc<AssignmentSnapshotStore>, ) -> Self

Install the assignment-snapshot store for dynamic rebalance.

pub fn config_var(self, key: &str, value: &str) -> Self

Set a config variable for ${VAR} substitution in SQL.

pub fn buffer_size(self, size: usize) -> Self

Set the default buffer size for streaming channels.

pub fn backpressure(self, strategy: BackpressureStrategy) -> Self

Set the default backpressure strategy.

pub fn storage_dir(self, path: impl Into<PathBuf>) -> Self

Set the storage directory for WAL and checkpoints.

pub fn checkpoint(self, config: StreamCheckpointConfig) -> Self

Set checkpoint configuration.

pub fn profile(self, profile: Profile) -> Self

Set the deployment profile.

See Profile for the available tiers.

pub fn object_store_url(self, url: impl Into<String>) -> Self

Set the object-store URL for durable checkpoints.

Required when using Profile::Durable or Profile::Cluster.

pub fn object_store_options(self, opts: HashMap<String, String>) -> Self

Set explicit credential/config overrides for the object store.

Keys are backend-specific (e.g., aws_access_key_id, aws_region). These supplement environment-variable-based credential resolution.

pub fn tiering(self, tiering: TieringConfig) -> Self

Set the S3 storage class tiering configuration.

pub fn delivery_guarantee(self, guarantee: DeliveryGuarantee) -> Self

Set the end-to-end delivery guarantee for the pipeline.

pub fn register_udf(self, udf: ScalarUDF) -> Self

Register a custom scalar UDF with the database.

The UDF will be available in SQL queries after build().

Example

use datafusion_expr::ScalarUDF;

let db = LaminarDB::builder()
    .register_udf(my_scalar_udf)
    .build()
    .await?;

pub fn register_udaf(self, udaf: AggregateUDF) -> Self

Register a custom aggregate UDF (UDAF) with the database.

The UDAF will be available in SQL queries after build().

Example

use datafusion_expr::AggregateUDF;

let db = LaminarDB::builder()
    .register_udaf(my_aggregate_udf)
    .build()
    .await?;

pub fn pipeline_channel_capacity(self, capacity: usize) -> Self

Source → coordinator channel capacity (default 64). Increase for burst absorption at the cost of memory.

pub fn pipeline_batch_window(self, window: Duration) -> Self

Micro-batch coalescing window (default 5ms for connectors, 0 for embedded). Larger values amortize per-cycle SQL overhead.

pub fn pipeline_drain_budget_ns(self, ns: u64) -> Self

Max time draining the source channel per cycle, in nanoseconds (default 1ms). Increase to process more messages per SQL execution.

pub fn pipeline_query_budget_ns(self, ns: u64) -> Self

Per-query execution budget in nanoseconds (default 8ms). When exceeded, remaining queries are deferred to the next cycle.

pub fn pipeline_max_input_buf_batches(self, batches: usize) -> Self

Per-port operator input-buffer cap in batches (default 256).

pub fn pipeline_max_input_buf_bytes(self, bytes: usize) -> Self

Per-port operator input-buffer cap in bytes.

pub fn pipeline_backpressure_policy(self, policy: BackpressurePolicy) -> Self

Backpressure policy (default Backpressure).

pub fn register_connector( self, f: impl FnOnce(&ConnectorRegistry) + Send + 'static, ) -> Self

Register custom connectors with the ConnectorRegistry.

The callback is invoked after the database is created and built-in connectors are registered. Use it to add user-defined source/sink implementations.

Example

let db = LaminarDB::builder()
    .register_connector(|registry| {
        registry.register_source("my-source", info, factory);
    })
    .build()
    .await?;

pub async fn build(self) -> Result<LaminarDB, DbError>

Build the LaminarDB instance.

Errors

Returns DbError if database creation fails.

Trait Implementations

impl Debug for LaminarDbBuilder

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

Formats the value using the given formatter.

impl Default for LaminarDbBuilder

fn default() -> Self

Returns the “default value” for a type.