The streaming engine that starts embedded

# pip install laminardb

import laminardb

conn = laminardb.open(":memory:")

conn.execute("""
    CREATE SOURCE events (key VARCHAR, value DOUBLE, ts TIMESTAMP)
""")

conn.execute("""
    CREATE STREAM summary AS
    SELECT key, COUNT(*) AS total, AVG(value) AS avg
    FROM events
    GROUP BY key, TUMBLE(ts, INTERVAL '1' MINUTE)
    EMIT ON WINDOW CLOSE
""")

# Push events in, then query results back
conn.insert("events", [
    {"key": "sensor-1", "value": 42.0, "ts": 1700000000000},
    {"key": "sensor-1", "value": 44.0, "ts": 1700000001000},
])

conn.sql("SELECT * FROM summary").show()
conn.close()

-- Tumbling window: 1-minute OHLC bars
CREATE STREAM ohlc_1m AS
SELECT symbol,
       first_value(price) AS open,
       MAX(price) AS high, MIN(price) AS low,
       last_value(price) AS close,
       SUM(volume) AS volume
FROM trades
GROUP BY symbol, tumble(ts, INTERVAL '1' MINUTE)
EMIT ON WINDOW CLOSE;

-- Session window: detect activity bursts
CREATE STREAM sessions AS
SELECT user_id, COUNT(*) AS clicks,
       MAX(ts) - MIN(ts) AS duration_ms
FROM clickstream
GROUP BY user_id, session(ts, INTERVAL '30' SECOND)
EMIT ON WINDOW CLOSE;

-- ASOF join: enrich trades with closest preceding quote
SELECT t.symbol, t.price, q.bid, q.ask,
       t.price - q.bid AS spread
FROM trades t
ASOF JOIN quotes q
    MATCH_CONDITION(t.ts >= q.ts)
    ON t.symbol = q.symbol;

-- Interval join: match orders to payments within 1 hour
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;

-- Temporal join: point-in-time lookup
SELECT o.*, r.rate
FROM orders o
JOIN rates FOR SYSTEM_TIME AS OF o.order_time AS r
    ON o.currency = r.currency;

# Option 1: Download pre-compiled binary (latest release)
VERSION=$(curl -s https://api.github.com/repos/laminardb/laminardb/releases/latest | grep tag_name | cut -d '"' -f4)

# Linux x86_64
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

# macOS Apple Silicon
curl -LO "https://github.com/laminardb/laminardb/releases/download/${VERSION}/laminardb-server-aarch64-apple-darwin-${VERSION}.tar.gz"

# All platforms: github.com/laminardb/laminardb/releases

# Option 2: Build from source
cargo install laminar-server

# Option 3: Docker (multi-arch, on Docker Hub and GHCR)
docker run -p 8080:8080 laminardb/laminardb-server:latest
# or: docker run -p 8080:8080 ghcr.io/laminardb/laminardb-server:latest
# full stack (server + Redpanda + Prometheus + Grafana): docker compose up

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

-- Kafka source with Avro and Schema Registry
CREATE SOURCE trades (
    symbol VARCHAR, price DOUBLE, volume BIGINT, ts BIGINT
) FROM KAFKA (
    brokers = 'localhost:9092',
    topic = 'market-trades',
    group_id = 'laminar-analytics',
    format = 'avro',
    schema.registry.url = 'http://schema-registry:8081'
);

-- Real-time aggregation
CREATE STREAM trade_summary AS
SELECT symbol, COUNT(*) AS trades,
       AVG(price) AS avg_price,
       SUM(volume) AS total_volume
FROM trades
GROUP BY symbol, tumble(ts, INTERVAL '1' MINUTE)
EMIT ON WINDOW CLOSE;

-- Kafka sink with exactly-once delivery
CREATE SINK output INTO KAFKA (
    brokers = 'localhost:9092',
    topic = 'trade-summaries',
    format = 'json',
    delivery.guarantee = 'exactly-once'
) AS SELECT * FROM trade_summary;

-- Delta Lake sink for historical analytics
CREATE SINK lake INTO DELTA_LAKE (
    path = 's3://my-bucket/trade_summary',
    write_mode = 'append',
    delivery.guarantee = 'exactly-once'
) AS SELECT * FROM trade_summary;