Configuration Reference

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

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

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

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;

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;

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;

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

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

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

[state]
backend = "memory"

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