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 = "embedded"
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 both the [discovery] and [coordination] sections to be defined.

[discovery]

[coordination]

node_id = "node-1"

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

[coordination]
strategy = "raft"
raft_port = 8888
election_timeout = "3s"
heartbeat_interval = "500ms"

[state]

The [state] section configures the engine's query state backend (used for aggregates, joins, and windowing). 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 (Durable, local filesystem)

Configure with backend = "local".

Object Store State (Durable, shared storage)

Configure with backend = "object_store". Essential for cluster mode deployments.

[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" }
  ]
}

Ad-hoc Query & Ephemeral Stream Lifecycle

To run an ad-hoc query and tail it in the browser:

1. Post the SQL query to POST /api/v1/queries:

   POST /api/v1/queries
   { "sql": "SELECT symbol, price FROM trades WHERE volume > 1000" }

2. The server registers an ephemeral stream named __console_<uuid> and returns:

   { "stream_id": "__console_<uuid>", "ws_url": "/ws/__console_<uuid>" }

3. The client connects to the WebSocket at ws_url to receive the data stream.
4. When the WebSocket connection closes or times out, the server automatically DROPs the ephemeral stream and releases resources.

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 assignment map mapping the 256 virtual nodes (vnodes) to active cluster nodes.
• GET /api/v1/cluster/leader — Returns node details for the active Raft 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 = "BIGINT"

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

[[pipeline]]

[[sink]]

[[lookup]]

Connectors

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

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

NATS Sink

PostgreSQL CDC Source

MySQL CDC Source

PostgreSQL Sink

MongoDB CDC Source

Consumes MongoDB change streams. Resume tokens provide at-least-once delivery across restarts. Requires MongoDB 4.0+ (replica set) or 4.2+ (sharded cluster).

Full document modes: delta sends the update description only (default). update_lookup fetches the current document on updates. required needs changeStreamPreAndPostImages enabled on the collection. when_available returns the post-image when present, otherwise null.

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 (Int64), _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 upsert.key.fields. replace rewrites the full document. cdc_replay routes by the _op column (I→insert, U→update, R→replace, D→delete). 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.

WebSocket Source

Note: WebSocket is non-replayable. At-most-once delivery on recovery.

WebSocket Sink

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}"

S3 Storage Class Tiering

Active checkpoints use the hot tier. Older checkpoints automatically tier down via S3 Lifecycle rules.

Delivery Guarantees

WebSocket sources are non-replayable. Under exactly-once mode, they degrade to at-most-once delivery with a warning.

Full Config Example

[server]
mode = "embedded"
bind = "0.0.0.0:8080"
workers = 4
log_level = "info"

[state]
backend = "in_process"

[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 = "BIGINT"

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

# --- Lookups ---

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

[lookup.properties]
hostname = "db.internal"
database = "reference_data"
username = "reader"
table.name = "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"
delivery = "exactly_once"

[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"
delivery = "exactly_once"

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