Server configuration reference

The laminardb binary reads a TOML configuration file, constructs streaming pipelines, and serves a REST API with Prometheus metrics. Server and checkpoint defaults are checked against server config and engine config. This page describes the v0.33.0 release. Last updated: 19 October 2026.

Install

v0.33.0 is the current release. Release assets are named laminardb-server-<target>-v0.33.0.tar.gz (.zip for x86_64-pc-windows-msvc) for the targets x86_64-unknown-linux-gnu, aarch64-unknown-linux-gnu, x86_64-unknown-linux-musl, x86_64-apple-darwin, aarch64-apple-darwin and x86_64-pc-windows-msvc, with a SHA256SUMS file. The commands below link to the v0.33.0 tag so that they keep working after later releases.

BASE=https://github.com/laminardb/laminardb/releases/download/v0.33.0
curl -fLO "$BASE/SHA256SUMS"

# Linux x86_64
ASSET=laminardb-server-x86_64-unknown-linux-gnu-v0.33.0.tar.gz
curl -fLO "$BASE/$ASSET"
grep " $ASSET\$" SHA256SUMS | sha256sum -c -
tar xzf "$ASSET"

# macOS Apple silicon (instead of the Linux lines above)
ASSET=laminardb-server-aarch64-apple-darwin-v0.33.0.tar.gz
curl -fLO "$BASE/$ASSET"
grep " $ASSET\$" SHA256SUMS | shasum -a 256 -c -
tar xzf "$ASSET"

# Run the binary (the archive contains the laminardb executable)
./laminardb --config laminardb.toml

# Or build from a source checkout (laminar-server is not published to crates.io)
git clone https://github.com/laminardb/laminardb.git
cd laminardb && git checkout v0.33.0
cargo build --release -p laminar-server
./target/release/laminardb --config laminardb.toml

# Or Docker. The image's built-in config binds 0.0.0.0:8080 and reads
# console_token from LAMINAR_CONSOLE_TOKEN; the server exits without it.
export LAMINAR_CONSOLE_TOKEN="$(openssl rand -hex 32)"
docker run -d -p 8080:8080 -e LAMINAR_CONSOLE_TOKEN \
  -v laminardb-data:/var/lib/laminardb \
  laminardb/laminardb-server:0.33.0

The image runs laminardb --config /etc/laminardb/laminardb.toml. To add pipelines, mount your own file there (for example -v "$PWD/laminardb.toml:/etc/laminardb/laminardb.toml:ro"); keep a console_token, because a container needs a non-loopback bind. Its built-in checkpoint directory is file:///var/lib/laminardb/checkpoints. The image sets LAMINAR_DISABLE_FILE_WATCH=1, so call POST /api/v1/reload after editing a mounted config. The :latest and :main tags are rolling builds of the main branch, not releases.

CLI Options

  -c, --config <CONFIG>            [default: laminardb.toml]
      --log-level <LOG_LEVEL>      [default: info]
      --admin-bind <ADMIN_BIND>
      --pgwire-bind <PGWIRE_BIND>  Postgres wire bind address (e.g. `127.0.0.1:5433`). Wildcard binds rejected

[server]

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

Top-Level Config Keys:

Other top-level sections: [supervision] (see Supervision), [[process_function]] (supervised Python workers; see the server README), and [ai] / [models.<name>] (see AI functions). Unknown keys in [server], [checkpoint], [discovery], [[source]], [[sink]] and at the top level are rejected at startup.

[server]
mode = "single"
bind = "0.0.0.0:8080"
console_token = "${LAMINAR_CONSOLE_TOKEN}"   # required for a non-loopback bind
pgwire_bind = "127.0.0.1:5433"

Cluster Config

When mode = "cluster", the server requires a unique top-level node_id, a [discovery] section with seeds, and a [checkpoint] url on shared object storage (s3://, gs:// or az://). Cluster coordination uses that checkpoint store; there is no separate [coordination] section or embedded Raft service.

Cluster mode rejects best_effort delivery, local paths and file:// checkpoints (peers cannot recover them), and a checkpoint.interval below 100ms. At-least-once requires replayable input and a durable sink. Cluster exactly-once supports Kafka input with either an S3/S3A append-mode Delta Lake sink or a REST-catalog Iceberg append sink backed by S3/S3A storage.

[discovery]

The four cluster_tls_* keys must be set together and each path must exist. They do not protect the HTTP API; terminate HTTP TLS at a trusted proxy.

node_id = "node-1"   # unique per node

[server]
mode = "cluster"
bind = "0.0.0.0:8080"
console_token = "${LAMINAR_CONSOLE_TOKEN}"   # required for a non-loopback bind
delivery = "at_least_once"
key_groups = 256

[checkpoint]
url = "s3://my-bucket/laminardb/cluster-checkpoints"   # shared by every node

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

[checkpoint]

[checkpoint]
url = "file:///var/lib/laminardb/checkpoints"
interval = "30s"

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

A single-node server with replay-capable delivery and an s3://, gs:// or az:// checkpoint fails with [LDB-0014]: a shared cloud checkpoint namespace is not admitted for one node until its writer lease is term-fenced. A file:// checkpoint directory is locked by one process at a time.

Memory Limits & Production Tuning

The limits above are independent admission budgets. DataFusion reservations are shared per DB/node; source queue bytes are shared per DB/node; graph input limits apply to each port; reference-table and materialized-view limits apply to each local table or view. They do not sum to a process RSS cap. Connector decoding, direct Arrow allocations, managed state, query results, checkpoint encoding, restore staging and allocator overhead need separate headroom.

Managed operator working state has a default 256 MiB charged-byte budget. Set [server].pipeline_max_managed_state_bytes to change it (embedded users set the same field through LaminarConfig or the builder). Its charge is a lower bound, not total resident memory. Local reference tables and materialized views are unavailable in cluster plans that require distributed ownership.

Size a deployment

1. Run representative peak input, bursts and a fault/restart cycle with the intended SQL and connectors. Measure peak process or container RSS, source lag and checkpoint duration.
2. Observe /metrics: laminardb_cycles_backpressured_total, laminardb_input_buf_bytes, laminardb_managed_state_accounted_bytes, laminardb_checkpoint_size_bytes, laminardb_checkpoints_failed_total and laminardb_checkpoint_duration_seconds. Use OS or container RSS (on Linux, laminardb_process_resident_memory_bytes) for the actual memory envelope.
3. Set container memory requests and limits above the measured peak with room for Arrow and connector scratch, simultaneous old/new state on recovery, and checkpoint capture. The Helm chart defaults to resources: {}; supply workload-specific values.
4. If admission saturates, examine batch size and downstream throughput before raising queue or graph-port limits. A result too large for a port after execution faults the pipeline. Size checkpoint.max_node_data_bytes above the largest expected participant artifact and verify recovery at that setting.

checkpoint.interval defaults to 10s and trades replay work against checkpoint traffic. checkpoint.timeout defaults to 120s; keep it above observed attempt duration. Changes to [server] and [checkpoint] require a restart (see Hot Reload). The main I/O runtime uses TOKIO_WORKER_THREADS (logical CPU count by default), while streaming compute uses one dedicated thread. On the default non-MSVC build, jemalloc can be tuned with MALLOC_CONF at process start; see the allocator guidance.

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. When server.console_token is configured, every /api/v1/* route and /ws/{name} requires it. /health, /ready and /metrics are always public so probes and scrapers work without a token; restrict network access to them.

Security & Gating

• HTTP Authentication: Send the configured token in the header: Authorization: Bearer <token>. Duplicate or comma-joined bearer values are refused.
• WebSocket Authentication: Browsers cannot set headers on the upgrade request, so /ws/{name} also accepts ?token=<token>. The query parameter is accepted only on /ws/ routes.
• CORS Gating: Restrict client origins using the server.console_cors_allowed_origins array in TOML. Without it the CORS policy is permissive.
• TLS: The HTTP listener has no TLS settings; terminate TLS at a trusted proxy.

Lineage Dependency Graph

GET /api/v1/graph returns the pipeline lineage (sources → streams → sinks). node_type is "Source", "Stream" or "Sink"; sql is the node's SQL or null.

{
  "nodes": [
    { "name": "trades", "node_type": "Source", "sql": null },
    { "name": "vwap", "node_type": "Stream", "sql": "SELECT ..." },
    { "name": "trade_archive", "node_type": "Sink", "sql": null }
  ],
  "edges": [
    { "from": "trades", "to": "vwap" },
    { "from": "vwap", "to": "trade_archive" }
  ]
}

WebSocket Subscription Protocol

Connect to GET /ws/{name}, where name is a stream; add ?as_of_epoch=N to start from a committed epoch instead of the tail. Each text frame is a JSON object with a type. subscription_id is the stream name, and all counters are sent as decimal strings. Data frames carry at most 1 MiB, so a large batch is split across frames using row_offset/row_count:

{
  "type": "data",
  "subscription_id": "vwap",
  "data": [ { "symbol": "AAPL", "vwap": 178.5 } ],
  "sequence": "42",
  "log_sequence": "17",
  "row_offset": "0",
  "row_count": "1"
}

Other frame types: progress (epoch, checkpoint_id, log_sequence, through_log_sequence, sequence), gap (the subscriber lagged; code = subscription_lagged, skipped_messages) and error (code, message). In cluster mode data frames also carry stream_generation, partition, partition_sequence and committed_epoch. The server sends a WebSocket ping every 15 seconds and closes the connection after two unanswered pings. The server accepts at most 10,000 WebSocket connections.

Cluster Management Endpoints

• GET /api/v1/cluster/nodes — Cluster members: id, name, rpc_address, state (Joining, Active, Suspected, Draining, Left), metadata and last_heartbeat_ms.
• GET /api/v1/cluster/vnodes — The durable key-group (vnode) assignment snapshot.
• GET /api/v1/cluster/leader — The current leader's node record (if known) and whether this node is the leader.
• GET /api/v1/cluster/checkpoints — Latest checkpoint status; also available in single-node mode.
• GET /api/v1/cluster/topology and /api/v1/cluster/topology/* — Durable catalog version, topology validation, migration submission, preparation, status and legacy-catalog adoption (see the REST table below).

/api/v1/cluster, /nodes, /vnodes and /leader return 404 when the server is not in cluster mode.

REST API

The server exposes an HTTP API on the configured bind address. All responses are JSON except /metrics (Prometheus text format). The first three routes are public; the rest require the console token when one is configured. Until startup completes, and in cluster mode during coordinated recovery, routes other than the three probes return 503.

Probes and metrics (public)

Catalog, SQL and operations

POST /api/v1/sql takes {"sql": "..."} and returns at most 1000 rows, stopping after a 5 second collection budget; a capped result sets "truncated": true. Use a WebSocket or pgwire subscription for unbounded results.

Pipeline control

These routes act on the whole server, not on one named pipeline. In cluster mode, start and stop run locally first and are then forwarded to every peer; add ?local=true to act on this node only.

Cluster

The two local-* diagnostic routes accept the console token or the read-only server.diagnostic_read_token, allow one request at a time (at most 8 starts per second), and are outside the console CORS policy. See the server README for the topology migration workflow.

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; a second reload while one is running returns 409.

What reloads without restart: [[source]], [[lookup]], [[pipeline]] and [[sink]] entries (add, remove, or change).

What requires restart: [server], [checkpoint], [discovery], [supervision], [[process_function]], [ai], [models], the top-level sql string and node_id. The reload response carries a warning for each of these that changed.

Reload does not touch certificates. The pgwire listener watches its own pgwire_tls_cert, pgwire_tls_key and pgwire_tls_client_ca files and swaps them in when they change; a failed reload keeps the previous certificate.

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 (version 0.0.4) and needs no token. Every metric name carries the laminardb_ prefix and the constant labels instance (hostname) and pipeline (the first [[pipeline]] name, or default). A representative set:

On Linux the standard process collector is registered under the same prefix (for example laminardb_process_resident_memory_bytes). Cluster nodes add laminardb_cluster_subscription_* and placement metrics, and connectors register their own metrics when they start (for example laminardb_kafka_source_records_polled_total).

[[source]]

Each [[source]] becomes CREATE SOURCE name (columns, PRIMARY KEY, WATERMARK) FROM CONNECTOR (properties) [FORMAT fmt]. Unknown fields are rejected.

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

[source.properties]
"bootstrap.servers" = "localhost:9092"
topic = "market-trades"
"group.id" = "laminar-analytics"
"startup.mode" = "earliest"   # required under at_least_once / exactly_once

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

Each [[pipeline]] becomes CREATE STREAM name AS sql. It has two fields; other keys are ignored.

[[sink]]

Each [[sink]] becomes CREATE SINK name FROM pipeline INTO CONNECTOR (properties) [FORMAT fmt]. Unknown fields are rejected.

Delivery is configured once at [server].delivery. A per-sink delivery field is rejected, and a delivery.guarantee property is owned by the runtime and cannot be set, so a pipeline cannot silently mix checkpoint semantics.

[[lookup]]

Each [[lookup]] becomes CREATE LOOKUP TABLE name (columns, PRIMARY KEY (...)) WITH ('connector' = ..., 'strategy' = ..., properties).

With strategy = "replicated", a non-default [lookup.cache] value is rejected ("cache.memory and cache.ttl require strategy = 'on-demand'"). There is no cache.hybrid option, and cache.disk is rejected. Keys placed directly under [[lookup]] other than the fields above are ignored, so put pushdown under [lookup.properties].

[[lookup]]
name = "instruments"
connector = "postgres"
strategy = "replicated"
primary_key = ["symbol"]

[lookup.properties]
host = "db.internal"
database = "reference_data"
user = "reader"
password = "${PG_PASSWORD}"
table = "public.instruments"

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

[[lookup.schema]]
name = "exchange"
type = "VARCHAR"

Connectors

In Rust builds, include the connectors you need with Cargo features, for example cargo add laminar-db --features kafka,delta-lake. These are build options, not runtime approval gates. The v0.33.0 server binaries and Docker image include every connector below.

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.

Set the serialization format with the SQL FORMAT clause (... FROM KAFKA (...) FORMAT JSON) or the TOML format field. A 'format' connector option is rejected: "option 'format' is unsupported; declare the format with the FORMAT clause". Format-specific options go in FORMAT fmt WITH (...).

RabbitMQ, IBM MQ, FIX and FpML connectors are optional build features and are not in default builds or the Docker image. The v0.33.0 x86_64 Linux (glibc) binary includes all four, the x86_64 Linux (musl) binary none, and the other binaries FIX and FpML.

Kafka Source

Formats (FORMAT clause / TOML format): json (used when omitted), csv, raw, debezium, avro. Avro needs schema.registry.url; the registry supports Avro only.

Required

Startup position

Schema Registry

Security

Consumer Tuning

Properties prefixed with kafka. are passed to librdkafka with the prefix removed, except protected keys, which fail with "overrides a setting the Kafka connector owns; use the connector option for it instead". Protected keys are everything the connector sets itself (for example kafka.bootstrap.servers, kafka.group.id, kafka.enable.auto.commit, kafka.isolation.level), security and TLS settings, transaction and idempotence settings, kafka.partitioner, and sasl.kerberos.* / sasl.oauthbearer.*.

Kafka Sink

Formats: json (used when omitted), csv, raw, avro. debezium is source-only.

The sink always uses idempotent production with acks=all; acks is not configurable. Under exactly_once, output becomes visible to read_committed readers once per committed checkpoint.

NATS Source

Formats (FORMAT clause / TOML format): json (used when omitted), csv, raw. The source is ephemeral, so it needs best_effort delivery; it is rejected under at_least_once, the server default.

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

Formats: json (used when omitted), csv, raw. Core mode is ephemeral (best_effort only); JetStream mode supports at_least_once. Neither supports exactly_once.

PostgreSQL CDC Source

Reads one table through logical replication (pgoutput, PostgreSQL 17+) as typed keyed upserts or a retractable __weight changelog, with an optional initial snapshot handed off at the slot's consistent point. The table needs a primary key and REPLICA IDENTITY FULL, and the publication must be FOR TABLE exactly that table. The slot advances only after a checkpoint commits. Embedded and single-node server mode, at-least-once; cluster mode and exactly-once reject it. TRUNCATE and table or publication changes stop the source with reset guidance.

PostgreSQL Sink

Unknown properties are rejected. The sink is durable at-least-once in every mode; exactly_once rejects it.

MongoDB CDC Source

Reads one collection's change stream and delivers it straight to sinks, with no broker in between. Byte-bounded backpressure, classified bounded reconnects, and deployment- and collection-bound resume-token checkpoints. Embedded and single-node server mode, at-least-once; cluster mode and exactly-once reject it.

Full document modes: delta sends the update description only (default). required needs changeStreamPreAndPostImages enabled on the collection; a missing image stops the source and is never treated as a delete.

History schema (version 1): event_id, event_version, operation, database, collection, collection_uuid, document_key, full_document, update_description, event_details, resume_token, cluster_time_seconds, cluster_time_increment, wall_time, txn_number, lsid, snapshot_id. Document values are canonical Extended JSON v2. Operation names are data: no history column makes a sink delete rows.

Document mode: the PRIMARY KEY must be the document key (_id, plus shard-key fields); non-key columns are nullable and map exactly or fail. Sinks must be keyed on the same columns: PostgreSQL upsert with changelog.mode, or Delta write.mode=upsert. Filters, streams, joins and aggregates over a document source are rejected.

MongoDB Sink

Write modes: insert is append-only. upsert replaces documents by the fields listed in write.mode.key_fields. cdc_replay applies MongoDB CDC history records: insert, replace and snapshot replace by document key, update applies the post-image (or the update description), delete deletes by key, metadata events write nothing, and destructive lifecycle events stop the sink. Writes are ordered and majority-journaled; these correctness settings are not user-tunable. Time series collections only support insert.

Delta Lake Source

Replayable, so it works with at_least_once. It is a singleton source; cluster mode rejects it.

partition.filter and cdf.enabled are rejected.

Delta Lake Sink

Cloud storage options are passed with a storage. prefix, for example storage.aws_access_key_id, storage.aws_secret_access_key, storage.aws_region, storage.aws_endpoint; each falls back to the provider's environment variable. Delivery comes from the pipeline-wide setting. exactly_once is supported only for write.mode = append with the atomic-create DefaultLogStore; custom S3 endpoints must pass an atomic conditional-create probe at sink startup. Cluster exactly-once also requires an S3/S3A table URI.

Apache Iceberg Source & Sink

Connector name iceberg for both directions. Only the REST catalog is enabled in v0.33.0 builds; glue, hms, s3tables and sql are rejected at startup. Default builds include S3 and local filesystem storage; GCS and Azure need the iceberg-gcs / iceberg-azure build features.

Common

Other catalog and storage settings pass through as catalog.property.* and storage.property.*.

Source

Sink

Cluster exactly_once is admitted only for an append sink on a REST catalog with no or bearer auth and S3 storage with a direct S3 warehouse.

WebSocket Source

Note: the WebSocket source is a client and is ephemeral (non-replayable), so it requires pipeline-wide best_effort; it is rejected under at_least_once, the server default. Reconnects can create gaps or duplicates. Formats: json (used when omitted), csv (comma-delimited, no header) or binary. Unknown properties are rejected.

WebSocket Sink

Ephemeral, so best_effort only. The wire format is always JSON: any FORMAT clause (or TOML format) is rejected. Unknown properties are rejected.

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)

Watches a local directory and loads new files. Formats: csv, tsv, json (aliases jsonl, ndjson), text, parquet, arrow. Without a FORMAT clause the format is taken from the extension in path (e.g. /data/in/*.csv); a bare directory needs an explicit format. CSV and JSON sources need declared columns or schema.inference = true. Replayable; cluster mode rejects it.

Known limitations in v0.33.0:

• A json file must contain exactly one JSON object. Files with several JSON Lines records, or a top-level JSON array, fail to decode ("trailing characters" or "JSON value must be an object"), despite the jsonl/ndjson aliases.
• A file that fails to decode is retried continuously, with a warning in the log each time, and stays at the head of the queue, so later files are not loaded. Fixing or deleting the file does not clear it while the server runs (the source then reports that the file changed size or is missing): restart the server after fixing or removing it.
• The CSV reader cannot decode TIMESTAMP columns; declaring one fails with "CSV reader ... cannot decode". Use BIGINT, DOUBLE, BOOLEAN, DATE or VARCHAR columns, or another format such as Parquet.

File Sink

Writes immutable rolling files to a local directory. Durable at-least-once; exactly_once rejects it. The format (FORMAT clause / TOML format) is required: csv, json, text, parquet or arrow.

The mode option was removed and is rejected; files are always published as immutable rolling files.

OpenTelemetry Source

Receives OTLP/gRPC traces, metrics or logs. Each source handles one signal type on its own port and has a fixed schema for that signal, so there is no FORMAT clause. The source is ephemeral and node-local: it needs best_effort delivery and cannot run in cluster mode.

CREATE SOURCE traces FROM OTEL (port = '4317', signals = 'traces');

Generator Source

Emits a deterministic sequence with no external system, for demos and tests. It is always built in. Row i is (seq = i, ts_ms = i * 1000 / rows.per.second, value = "v{i}") with the fixed schema seq BIGINT NOT NULL, ts_ms BIGINT NOT NULL, value VARCHAR NOT NULL. Because output depends only on the offset, the source is replayable and admitted for single-node at_least_once and exactly_once. It is a singleton source, so cluster mode rejects it.

[[source]]
name = "gen"
connector = "generator"

[source.properties]
"rows.per.second" = "100"

Join Semantics

Bounded interval joins are the stateful stream-stream joins. Temporal joins look up the version of a keyed source that was valid at the probe row's event time. Lookup joins enrich a stream from a [[lookup]] table. Admission rejects unsupported join shapes before connector I/O. The rules below follow the SQL reference.

Interval Join (Stream-Stream)

Matches pairs of rows from two direct sources whose equality keys match and whose event times satisfy the directional predicate right.ts BETWEEN left.ts AND left.ts + bound, where the bound is a positive, finite interval. Both inputs need watermarks on TIMESTAMP NOT NULL event-time columns. Equality keys are one or more VARCHAR/BIGINT columns of matching types; NULL keys do not match.

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

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

• Every projected expression needs an explicit alias, and every projected or filtered column must use its left/right qualifier.
• Outer and anti unmatched rows are emitted only when the opposite input's watermark closes their possible match interval.
• Joining and aggregating in one statement is not supported: name the join output and aggregate it in a separate stream.
• Cross, unbounded, general non-equality, intermediate-input and multi-way joins are rejected on this path.

Temporal Join (FOR SYSTEM_TIME AS OF)

For each left row, looks up the right-side row that was valid at the left row's event time, using SQL:2011 FOR SYSTEM_TIME AS OF. Runs in single-node and cluster mode for one direct two-input join. ASOF JOIN is not an alias.

Supported types: INNER, LEFT

CREATE STREAM converted_orders AS
SELECT o.order_id AS order_id,
       o.amount AS amount,
       r.rate AS 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;

• The left source must be append-only. The right source must declare its equality keys as its PRIMARY KEY; it may be append-only or keyed-upsert.
• Both sides need watermarks on their event-time columns. A result becomes final only after the right watermark passes its probe time.
• Set [server].temporal_join_idle_history_retention; a pipeline containing a temporal join requires it. It bounds right-side version history while an input is idle.

Lookup Join

Enriches stream rows from a lookup table by equality on its single-column primary key, with no time bound. A replicated table is held in memory; an on-demand table fetches missing keys from the connector and keeps them in a bounded memory cache (cache.memory, cache.ttl). With pushdown set to auto or enabled, predicates are pushed to the lookup connector. Cluster mode rejects lookup joins.

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;

[supervision]

Single-node only. After a fatal pipeline fault, the server restarts the pipeline from the last checkpoint with exponential backoff, up to max_restarts times within the window; after that the pipeline stays faulted. Supervision is not enabled when [[process_function]] entries are configured. Changes require a restart.

AI Functions ([ai] and [models])

SQL functions such as ai_classify, ai_sentiment, ai_embed and ai_complete run a named model from a registry configured with [ai.providers.<name>], [models.<name>] and [ai.defaults]. See the server README AI section for the keys and an example.

• AI functions are single-node only; cluster mode rejects them ([LDB-4007]).
• Local (ONNX) models need ONNX Runtime 1.24 or newer installed on the host, on the library search path or via ORT_DYLIB_PATH. The Docker image does not include ONNX Runtime. Remote providers do not need it.
• Remote providers read their API key from the environment variable named in api_key_env; the key is not stored in the config.
• Changes to [ai] or [models] require a restart.

Environment Variables

Config file: before parsing, the server replaces ${VAR} and ${VAR:-default} anywhere in the TOML text with environment values, including inside [[pipeline]] SQL and the top-level sql string. A missing variable without a default is a startup error. Write $${VAR} to keep a literal ${VAR} in the loaded config so that it is resolved later, when the DDL runs.

Connector and format options in DDL: when a CREATE SOURCE, CREATE SINK or lookup-table statement runs, ${VAR} and ${VAR:-default} in its connector and format option values are resolved from the database's config variables (set with LaminarDbBuilder::config_var when embedded), then from the process environment. This also applies to DDL sent through POST /api/v1/sql or pgwire. An unresolved name without a default fails the statement. Other parts of SQL submitted at runtime are not substituted.

[source.properties]
"bootstrap.servers" = "${KAFKA_BROKERS:-localhost:9092}"
"sasl.username" = "${KAFKA_USER}"
"sasl.password" = "$${KAFKA_PASSWORD}"   # kept literal; resolved when the DDL runs

[checkpoint]
url = "s3://${S3_BUCKET}/checkpoints"   # cluster mode (single-node at_least_once needs file://)

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

Other variables read by the server: LAMINAR_DISABLE_FILE_WATCH=1 (or true) disables the config watcher, TOKIO_WORKER_THREADS sizes the I/O runtime, RUST_LOG overrides --log-level, and MALLOC_CONF tunes jemalloc.

Delivery Guarantees

Delivery is one pipeline-wide setting. Per-connector delivery fields are rejected, and incompatible connectors fail before external I/O. For example, the WebSocket, OpenTelemetry and NATS (Core and JetStream) sources are non-replayable and therefore require best_effort; they fail under the server default at_least_once. Cluster mode rejects best_effort. Cluster exactly_once admits Kafka input with an S3/S3A append-mode Delta Lake sink or a REST-catalog Iceberg append sink backed by S3/S3A storage.

Full Config Example

A single-node, at-least-once pipeline: a Kafka source read from the earliest offset, a one-minute VWAP window, and two sinks. Set LAMINAR_CONSOLE_TOKEN (at least 8 characters) and KAFKA_BROKERS before starting; a missing variable is a startup error. To add enrichment, define a [[lookup]] table and join it in a separate pipeline.

[server]
mode = "single"
bind = "0.0.0.0:8080"
console_token = "${LAMINAR_CONSOLE_TOKEN}"   # required for a non-loopback bind
delivery = "at_least_once"

[checkpoint]
# Single-node at_least_once / exactly_once need a local file:// directory.
url = "file:///var/lib/laminardb/checkpoints"
interval = "30s"

# --- Sources ---

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

[source.properties]
"bootstrap.servers" = "${KAFKA_BROKERS}"
topic = "market-trades"
"group.id" = "laminar-vwap"
"startup.mode" = "earliest"   # group offsets are rejected under at_least_once

[[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"
nullable = false

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

# --- Pipelines ---

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

# --- Sinks ---

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

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

[[sink]]
name = "archive"
pipeline = "vwap"
connector = "files"
format = "parquet"

[sink.properties]
path = "/var/lib/laminardb/archive/vwap"