Standalone Server & Configuration Reference

The laminardb binary reads a TOML configuration file, constructs streaming pipelines, and serves a REST API with Prometheus metrics. Every field on this page is verified against config.rs. Last updated: 25 May 2026.

Install

# Pre-built binary (Linux x86_64) — resolve the latest release tag
VERSION=$(curl -s https://api.github.com/repos/laminardb/laminardb/releases/latest | grep tag_name | cut -d '"' -f4)
curl -LO "https://github.com/laminardb/laminardb/releases/download/${VERSION}/laminardb-server-x86_64-unknown-linux-gnu-${VERSION}.tar.gz"
tar xzf laminardb-server-*.tar.gz

# Or build from source
cargo install laminar-server

# Or Docker (multi-arch — Docker Hub or GHCR)
docker run -d -p 8080:8080 laminardb/laminardb-server:latest
# docker run -d -p 8080:8080 ghcr.io/laminardb/laminardb-server:latest

# Run the binary
./laminardb --config laminardb.toml

CLI Options

[server]

The [server] section configures standalone server behaviors, control plane security, and pgwire network parameters.

Top-Level Config Keys:

[server]
mode = "single"
bind = "0.0.0.0:8080"
console_token = "supersecret"
pgwire_bind = "0.0.0.0:5433"
pgwire_allow_remote = true
log_level = "info"

Cluster Config

When mode = "cluster", the server requires a unique node_id and a [discovery] section. Cluster coordination uses the configured shared state store; there is no separate [coordination] section or embedded Raft service.

Cluster mode requires delivery = "at_least_once" and cluster-shared state on S3, GCS, or Azure. Cluster exactly-once fails closed with [LDB-0013]: checkpoint decisions are term-fenced, but supported connectors do not yet have certified term-fenced source handoff and external sink cursors. Local paths and file:// state are rejected because peers cannot recover them.

[discovery]

node_id = "node-1"

[server]
mode = "cluster"
key_groups = 256

[discovery]
strategy = "gossip"
gossip_port = 7946
advertise_host = "10.0.0.1"
seeds = ["10.0.0.1:7946", "10.0.0.2:7946"]

[state]

The [state] section configures storage for per-vnode checkpoint artifacts and exact-attempt durability seals. Hot aggregate, join, and window state remains in the operator runtime; this setting does not provide an incremental disk-backed keyed-state engine. It is a tagged union using the backend key.

In-Process State (Non-durable, in-memory)

Configure with backend = "in_process" (the default).

Local State (Node-durable filesystem)

Configure with backend = "local". This survives a same-node process restart and is valid for embedded/single-node exactly-once, but it is not visible to another cluster node.

Object Store State (Failure scope determined by URL)

Configure with backend = "object_store". S3, GCS, and Azure URLs are cluster-shared; a file:// URL remains node-durable only. Cluster mode requires one of the shared cloud schemes.

[checkpoint]

[checkpoint]
url = "s3://my-bucket/laminardb/checkpoints"
interval = "30s"

[checkpoint.storage]
aws_region = "us-east-1"
aws_access_key_id = "${AWS_ACCESS_KEY_ID}"
aws_secret_access_key = "${AWS_SECRET_ACCESS_KEY}"

Console UI & Control API

The Console UI is a web client that lets you inspect your streaming SQL nodes, view lineage dependency graphs, and author queries. You can access the hosted console at laminardb.github.io/laminardb-console-ui.

The console connects to the control-plane APIs exposed by the server. All endpoints are protected by token authentication if server.console_token is configured.

Security & Gating

• HTTP Authentication: Send the configured token in the header: Authorization: Bearer <token>.
• WebSocket Authentication: Pass the token as a query parameter: /ws/{name}?token=<token>.
• CORS Gating: Restrict client origins using the server.console_cors_allowed_origins array in TOML.

Lineage Dependency Graph

GET /api/v1/graph parses SQL statements and queries the engine schema to return a directed acyclic graph (DAG) of the data pipeline.

{
  "nodes": [
    { "id": "trades", "type": "source", "label": "trades" },
    { "id": "vwap", "type": "stream", "label": "vwap" },
    { "id": "trade_archive", "type": "sink", "label": "trade_archive" }
  ],
  "edges": [
    { "source": "trades", "target": "vwap" },
    { "source": "vwap", "target": "trade_archive" }
  ]
}

Realtime Telemetry & WebSocket Protocol

WebSocket client connects to GET /ws/{name}. Messages are serialized using Arrow JSON format:

{
  "type": "data",
  "subscription_id": 1,
  "data": [ { "symbol": "AAPL", "price": 178.5 } ],
  "sequence": 42
}

A heartbeat ping frame is sent by the server every 15 seconds; the client must respond to prevent connection termination.

Cluster Management Endpoints

• GET /api/v1/cluster/nodes — Returns list of cluster members, including their node IDs, network addresses, and states (Active, Draining, Suspected, Left).
• GET /api/v1/cluster/vnodes — Returns the current key-group assignment map for active cluster nodes.
• GET /api/v1/cluster/leader — Returns node details for the active durable leader-lease holder.
• GET /api/v1/cluster/checkpoints — Returns history and completion metadata of recent coordinator checkpoints.

REST API

The server exposes an HTTP API on the configured bind address. All responses are JSON except /metrics (Prometheus text format).

Hot Reload

Edit the TOML config file while the server is running. A file watcher detects changes (500ms debounce, content-hash comparison), diffs the old and new configs, and applies incremental DDL. Or trigger manually via POST /api/v1/reload.

What reloads without restart: sources, pipelines, sinks, lookups (add, remove, or change).

What requires restart: [server], [state], [checkpoint] sections. A warning is returned if these sections change.

Reload order: remove changed/deleted sinks → pipelines → lookups → sources, then recreate sources → lookups → pipelines → sinks. Disable file watcher with LAMINAR_DISABLE_FILE_WATCH=1.

Prometheus Metrics

GET /metrics returns Prometheus text format. Key metrics:

[[source]]

[[source]]
name = "trades"
connector = "kafka"
format = "json"

[source.properties]
bootstrap.servers = "localhost:9092"
topic = "market-trades"
group.id = "laminar-analytics"

[[source.schema]]
name = "symbol"
type = "VARCHAR"
nullable = false

[[source.schema]]
name = "price"
type = "DOUBLE"

[[source.schema]]
name = "ts"
type = "TIMESTAMP"

[source.watermark]
column = "ts"
max_out_of_orderness = "5s"

[[pipeline]]

[[sink]]

Delivery is configured once at [server].delivery. A per-sink delivery field is rejected so a pipeline cannot silently mix checkpoint semantics.

[[lookup]]

Connectors

All connectors are feature-gated. Enable via cargo add laminar-db --features kafka,delta-lake.

Set delivery once at [server].delivery (or LaminarDbBuilder::delivery_guarantee when embedded). The runtime injects the selected protocol and owns checkpoint/committer identities; connector properties do not require public writer IDs.

Kafka Source

Required

Format & Schema

Security

Consumer Tuning

Any property prefixed with kafka. is passed through to the underlying rdkafka consumer.

Kafka Sink

NATS Source

JetStream messages receive confirmed acknowledgements asynchronously after successful deserialization; acknowledgement failures and timeouts leave them eligible for broker redelivery. NATS source progress is not checkpoint-coupled, and both JetStream and Core modes are admitted only as ephemeral/best-effort sources.

NATS Sink

PostgreSQL CDC Source

Delivery boundary: the source resumes from the last fully emitted committed transaction and is admitted for at-least-once processing. It does not create an initial snapshot, so fresh startup is rejected before network I/O. Both the control and replication connections use the same verified TLS policy.

PostgreSQL Sink

MongoDB CDC Source

Consumes fixed-collection MongoDB change streams with byte-bounded backpressure, reconnect, and deployment- and collection-bound resume-token checkpoints. Fresh sources establish an exact empty bootstrap cut. Delivery is replayable at-least-once.

Full document modes: delta sends the update description only (default). required needs changeStreamPreAndPostImages enabled on the collection.

CDC envelope schema: _namespace (Utf8), _op (Utf8: I/U/R/D/DROP), _document_key (Utf8), _cluster_time_s (UInt32), _cluster_time_i (UInt32), _wall_time_ms (Timestamp(ms)), _full_document (Utf8, nullable), _update_desc (Utf8, nullable), _resume_token (Utf8).

MongoDB Sink

Write modes: insert is append-only. upsert replaces documents by the fields listed in write.mode.key_fields. cdc_replay routes by the _op column (I→insert, U→update, R→replace, D→delete). Writes are ordered and majority-journaled; these correctness settings are not user-tunable. Time series collections only support insert.

Delta Lake Source

Delta Lake Sink

Cloud storage credentials: use storage.aws_access_key_id, storage.aws_secret_access_key, storage.aws_region, etc. Delivery comes from the pipeline-wide setting. Delta's checkpoint-committable protocol is limited to append mode in embedded/single-node runtime; production exactly-once certification remains a release fault/soak gate. Alternate log stores and custom object-store endpoints fail closed. Its coordinated transaction namespace is internal.

WebSocket Source

Note: WebSocket is non-replayable and therefore requires pipeline-wide best_effort. Reconnects can create gaps or duplicates; stronger guarantees fail admission.

WebSocket Sink

Client mode sends one schema-encoded JSON row per text frame. Server-mode clients must first send {"action":"subscribe"}; the server replies with a subscription ID and then emits {"type":"data","data":[...],"sequence":N} text frames. Server fan-out uses a shared 32-frame ring with 256 KiB frames; an oversized row or burst is rejected before publication, and a lagging subscriber is disconnected so sequence gaps remain visible. The built-in server does not terminate TLS or authenticate clients; expose it only on a trusted network or behind an authenticated TLS reverse proxy.

File Source (Auto-Loader)

File Sink

Join Semantics

LaminarDB supports five join types. All are stateful and operate on streaming data. Join types are detected from SQL syntax automatically.

Interval Join (Stream-Stream)

Matches all pairs of rows from two streams where timestamps fall within a time bound: |left_ts - right_ts| <= time_bound. Both sides are buffered in state. Expired rows are evicted when the watermark advances past the bound.

Supported types: INNER, LEFT, RIGHT, FULL, LEFT SEMI, LEFT ANTI, RIGHT SEMI, RIGHT ANTI

CREATE STREAM matched AS
SELECT o.order_id, o.amount, p.status
FROM orders o
INNER JOIN payments p
    ON o.order_id = p.order_id
    AND p.ts BETWEEN o.ts AND o.ts + 3600000;

SELECT o.order_id, o.amount
FROM orders o
LEFT SEMI JOIN payments p
    ON o.order_id = p.order_id
    AND p.ts BETWEEN o.ts AND o.ts + 3600000;

State: Both sides buffered as Arrow batches with per-key BTreeMap<timestamp> indices. Time columns are extracted automatically from the BETWEEN clause. New left rows probe all right state; new right rows probe only old left state (avoids double-emit). Checkpointed via Arrow IPC serialization.

ASOF Join

For each left row, finds the one closest right row by timestamp within the same key partition. Uses Snowflake-compatible MATCH_CONDITION syntax.

Supported types: INNER, LEFT

Three directions

SELECT t.symbol, t.price, q.bid, q.ask
FROM trades t
ASOF JOIN quotes q
    MATCH_CONDITION(t.ts >= q.ts)
    ON t.symbol = q.symbol;

SELECT t.symbol, t.price, q.bid, q.ask
FROM trades t
ASOF JOIN quotes q
    MATCH_CONDITION(t.ts >= q.ts AND t.ts - q.ts <= INTERVAL '5' SECOND)
    ON t.symbol = q.symbol;

SELECT s.sensor_id, s.reading, c.calibration
FROM sensor_data s
ASOF JOIN calibrations c
    MATCH_CONDITION(NEAREST(s.ts, c.ts))
    ON s.sensor_id = c.sensor_id;

State: Right-side events stored in per-key BTreeMap<timestamp, Vec<row_index>> for O(log n) temporal lookups. Multiple rows at the same timestamp are handled correctly. Key columns support VARCHAR and BIGINT types.

Temporal Probe Join

For each left event, produces N output rows by probing the right stream at a fixed list of time offsets. Each probe returns the ASOF-matched right value at left.event_time + offset. Useful for lagged feature lookups and impact curves.

Offset specification

Directions: Backward, Forward, and Nearest. Each probed offset is resolved using the same semantics as a standalone ASOF join in that direction.

Temporal Join (FOR SYSTEM_TIME AS OF)

Point-in-time lookup against a versioned table. For each stream event, looks up the table row valid at the event's timestamp. Uses SQL:2011 FOR SYSTEM_TIME AS OF syntax.

Supported types: INNER, LEFT

SELECT o.order_id, o.amount, r.rate,
       o.amount * r.rate AS converted
FROM orders o
JOIN currency_rates FOR SYSTEM_TIME AS OF o.order_time AS r
    ON o.currency = r.currency;

Versioned tables: The right-side table must have a timestamp column marking version validity. Event-time semantics are deterministic; process-time (PROCTIME()) is non-deterministic.

Lookup Join

Enriches stream events against a static or slowly-changing reference table. Matches by key equality with no time bound. Results are cached with a configurable TTL.

Supported types: INNER, LEFT

SELECT o.order_id, o.amount, p.name, p.category
FROM orders o
LEFT JOIN products p
    ON o.product_id = p.id;

Cache: quick_cache-backed in-memory cache. Configure via [[lookup]] section: cache.size_bytes, cache.ttl, cache.hybrid. Supports predicate pushdown to the lookup connector.

Environment Variables

All string values in the TOML configuration file support ${VAR} substitution. Use ${VAR:-default} for optional variables with fallbacks. Missing variables without defaults cause a startup error.

SQL Queries: Because substitution is performed on the raw configuration file contents before parsing, environment variables can be embedded directly within SQL statements defined in the TOML file (such as in [[pipeline]] SQL queries or the top-level sql string).

Dynamic Queries: Environment variables are not expanded in queries submitted dynamically after startup (e.g., via the pgwire protocol or the POST /api/v1/sql HTTP endpoint).

[source.properties]
bootstrap.servers = "${KAFKA_BROKERS:-localhost:9092}"
sasl.username = "${KAFKA_USER}"
sasl.password = "${KAFKA_PASSWORD}"

[checkpoint]
url = "s3://${S3_BUCKET}/checkpoints"

[checkpoint.storage]
aws_access_key_id = "${AWS_ACCESS_KEY_ID}"
aws_secret_access_key = "${AWS_SECRET_ACCESS_KEY}"

Delivery Guarantees

Delivery is one pipeline-wide setting. Per-connector delivery fields are rejected, and incompatible connectors fail before external I/O. For example, WebSocket, OTLP, and NATS Core sources are non-replayable and therefore require best_effort. Cluster exactly_once fails closed with [LDB-0013] until connector handoff and external sink cursor publication consume the leader term end to end.

Full Config Example

[server]
mode = "single"
bind = "0.0.0.0:8080"
delivery = "at_least_once"

[state]
backend = "object_store"
url = "s3://my-bucket/laminardb/state"

[state.storage]
aws_region = "us-east-1"

[checkpoint]
url = "s3://my-bucket/laminardb/checkpoints"
interval = "30s"

[checkpoint.storage]
aws_region = "us-east-1"

# --- Sources ---

[[source]]
name = "trades"
connector = "kafka"
format = "avro"

[source.properties]
bootstrap.servers = "${KAFKA_BROKERS}"
topic = "market-trades"
group.id = "laminar-vwap"
schema.registry.url = "${SCHEMA_REGISTRY_URL}"
startup.mode = "earliest"

[[source.schema]]
name = "symbol"
type = "VARCHAR"
nullable = false

[[source.schema]]
name = "price"
type = "DOUBLE"

[[source.schema]]
name = "volume"
type = "BIGINT"

[[source.schema]]
name = "ts"
type = "TIMESTAMP"

[source.watermark]
column = "ts"
max_out_of_orderness = "5s"

# --- Lookups ---

[[lookup]]
name = "instruments"
connector = "postgres"
strategy = "poll"
pushdown = true

[lookup.properties]
host = "db.internal"
database = "reference_data"
username = "reader"
table = "instruments"

[lookup.cache]
size_bytes = 104857600
ttl = "5m"

# --- Pipelines ---

[[pipeline]]
name = "vwap"
sql = """
SELECT t.symbol,
       SUM(t.price * CAST(t.volume AS DOUBLE))
         / SUM(CAST(t.volume AS DOUBLE)) AS vwap,
       i.exchange
FROM trades t
JOIN instruments i ON t.symbol = i.symbol
GROUP BY t.symbol, i.exchange, tumble(t.ts, INTERVAL '1' MINUTE)
EMIT ON WINDOW CLOSE
"""

# --- Sinks ---

[[sink]]
name = "kafka_output"
pipeline = "vwap"
connector = "kafka"

[sink.properties]
bootstrap.servers = "${KAFKA_BROKERS}"
topic = "vwap-1m"
format = "json"
key.column = "symbol"
compression.type = "lz4"

[[sink]]
name = "lake"
pipeline = "vwap"
connector = "delta-lake"

[sink.properties]
table.path = "s3://my-bucket/delta/vwap"
partition.columns = "exchange"