> ## Documentation Index
> Fetch the complete documentation index at: https://docs.coreflux.org/llms.txt
> Use this file to discover all available pages before exploring further.

# CrateDB Route

> Store MQTT data in CrateDB for distributed time-series analytics and SQL queries

## CrateDB Overview

The `CRATEDB` route stores MQTT messages in CrateDB, a distributed SQL database optimized for time-series and machine data. It combines the familiarity of SQL with the scalability needed for IoT workloads.

> [!NOTE]
>
> CrateDB excels at high-volume time-series data with fast ingestion and SQL query capabilities. Use it when you need real-time analytics on sensor data at scale.

## Basic Syntax

```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
DEFINE ROUTE SensorDB WITH TYPE CRATEDB
    ADD CRATEDB_CONFIG
        WITH HOST "cratedb.example.com"
        WITH PORT '5432'
        WITH DATABASE "iot"
        WITH USERNAME "crate"
        WITH PASSWORD "secure_password"
    ADD EVENT StoreSensorReading
        WITH SOURCE_TOPIC "sensors/+/reading"
        WITH QUERY "INSERT INTO sensor_data (ts, sensor_id, value) VALUES (?, ?, ?)"
```

***

## Connection Configuration

### CRATEDB\_CONFIG Parameters

**HOST** `string` *required*

CrateDB server hostname or IP address.

**USERNAME** `string` *required*

Database username.

**PASSWORD** `string` *required*

Database password.

> **Optional Settings**
>
> **PORT** `integer`
>
> CrateDB PostgreSQL protocol port. Default: 5432.
>
> **DATABASE** `string`
>
> Target database/schema name.

> [!WARNING]
>
> **Never hardcode credentials in production.** Use environment variables and encrypted secrets to keep sensitive values out of your route definitions:
>
> ```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
> WITH HOST GET ENV "DB_HOST"
> WITH USERNAME GET ENV "DB_USER"
> WITH PASSWORD GET SECRET "DB_PASSWORD"
> ```
>
> See [Environment Variables & Secrets](/v2.14/mqtt-broker/secrets-and-env) for setup and usage.

***

## Writing Data

**Basic Time-Series**

Store sensor readings with timestamps:

```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
DEFINE ROUTE TimeSeriesDB WITH TYPE CRATEDB
    ADD CRATEDB_CONFIG
        WITH HOST "cratedb.example.com"
        WITH PORT '5432'
        WITH DATABASE "iot"
        WITH USERNAME "crate"
        WITH PASSWORD "secure_password"
    ADD EVENT StoreTimeSeries
        WITH SOURCE_TOPIC "sensors/+/reading"
        WITH DESTINATION_TOPIC "db/crate/status"
        WITH QUERY "INSERT INTO sensor_data (ts, sensor_id, value) VALUES (?, ?, ?)"
```

**High-Volume Metrics**

Optimized for high ingestion rates:

```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
DEFINE ROUTE MetricsDB WITH TYPE CRATEDB
    ADD CRATEDB_CONFIG
        WITH HOST "cratedb.example.com"
        WITH PORT '5432'
        WITH DATABASE "metrics"
        WITH USERNAME "metrics_writer"
        WITH PASSWORD "secure_password"
    ADD EVENT StoreMetrics
        WITH SOURCE_TOPIC "metrics/#"
        WITH QUERY "INSERT INTO metrics (ts, topic, data) VALUES (?, ?, ?)"
```

**Multiple Tables**

Route to different tables by topic:

```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
DEFINE ROUTE MultiTableDB WITH TYPE CRATEDB
    ADD CRATEDB_CONFIG
        WITH HOST "cratedb.example.com"
        WITH PORT '5432'
        WITH DATABASE "iot"
        WITH USERNAME "crate"
        WITH PASSWORD "secure_password"
    
    ADD EVENT StoreSensors
        WITH SOURCE_TOPIC "sensors/#"
        WITH QUERY "INSERT INTO sensors (ts, topic, data) VALUES (?, ?, ?)"
    
    ADD EVENT StoreAlerts
        WITH SOURCE_TOPIC "alerts/#"
        WITH QUERY "INSERT INTO alerts (ts, topic, data) VALUES (?, ?, ?)"
```

**UPSERT**

Insert or ignore on conflict. CrateDB supports `ON CONFLICT DO NOTHING` to skip duplicate inserts for idempotent ingestion:

```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
DEFINE ROUTE EventLog WITH TYPE CRATEDB
    ADD CRATEDB_CONFIG
        WITH HOST "cratedb.example.com"
        WITH PORT '5432'
        WITH DATABASE "iot"
        WITH USERNAME "crate"
        WITH PASSWORD "secure_password"
    ADD EVENT StoreUniqueEvent
        WITH SOURCE_TOPIC "events/+/occurred"
        WITH QUERY "INSERT INTO events (event_id, device_id, event_type, ts) VALUES ('{value.json.event_id}', '{value.json.device_id}', '{value.json.event_type}', '{value.json.timestamp}') ON CONFLICT (event_id) DO NOTHING"
```

`ON CONFLICT (event_id) DO NOTHING` ensures duplicate messages — common in IoT delivery — never create duplicate rows.

**UPDATE**

Modify existing records when a status change arrives:

```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
DEFINE ROUTE DeviceRegistry WITH TYPE CRATEDB
    ADD CRATEDB_CONFIG
        WITH HOST "cratedb.example.com"
        WITH PORT '5432'
        WITH DATABASE "iot"
        WITH USERNAME "crate"
        WITH PASSWORD "secure_password"
    ADD EVENT UpdateDeviceStatus
        WITH SOURCE_TOPIC "devices/+/status"
        WITH QUERY "UPDATE devices SET current_status = '{value.json.status}', last_seen = NOW() WHERE device_id = '{value.json.device_id}'"
```

**DELETE**

Remove records on a decommission or cleanup event:

```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
DEFINE ROUTE DeviceRegistry WITH TYPE CRATEDB
    ADD CRATEDB_CONFIG
        WITH HOST "cratedb.example.com"
        WITH PORT '5432'
        WITH DATABASE "iot"
        WITH USERNAME "crate"
        WITH PASSWORD "secure_password"
    ADD EVENT DecommissionDevice
        WITH SOURCE_TOPIC "devices/+/decommissioned"
        WITH QUERY "DELETE FROM devices WHERE device_id = '{value.json.device_id}'"
```

> [!NOTE]
>
> **Alternative: STORE IN with Models** — Instead of writing EVENT queries, you can bind a [model](/v2.14/lot-language/models/overview) directly to this route. Every `PUBLISH MODEL` call automatically inserts a row — no query needed:
>
> ```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
> DEFINE MODEL SensorReading
>     ADD STRING "sensor_id"
>     ADD DOUBLE "value"
>     STORE IN "TimeSeriesDB"
>         WITH TABLE "sensor_data"
> ```
>
> See [Data Storage Overview](./overview#with-models-store-in) for the full STORE IN workflow.

***

## Reading Data

EVENTs also support SELECT queries to read data back from the database. Publish a message to the event's `SOURCE_TOPIC`, and the query result is published to `DESTINATION_TOPIC`:

```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
DEFINE ROUTE SensorDB WITH TYPE CRATEDB
    ADD CRATEDB_CONFIG
        WITH HOST "cratedb.example.com"
        WITH PORT '5432'
        WITH DATABASE "iot"
        WITH USERNAME "crate"
        WITH PASSWORD "secure_password"
    ADD EVENT GetLatestReading
        WITH SOURCE_TOPIC "db/query/latest"
        WITH DESTINATION_TOPIC "db/result/latest"
        WITH QUERY "SELECT sensor_id, value FROM sensor_data WHERE sensor_id = '{payload}' ORDER BY ts DESC LIMIT 1"
```

To trigger this query, publish the sensor ID to the source topic:

```
Topic:   db/query/latest
Payload: temp001
```

The query result is published to `db/result/latest`, where your actions or external clients can consume it.

### Timed Polling

EVENTs can also run SELECT queries on a fixed schedule using `WITH EVERY`. This is especially powerful with CrateDB's fast aggregations:

```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
DEFINE ROUTE TimeSeriesDB WITH TYPE CRATEDB
    ADD CRATEDB_CONFIG
        WITH HOST "cratedb.example.com"
        WITH PORT '5432'
        WITH DATABASE "iot"
        WITH USERNAME "crate"
        WITH PASSWORD "secure_password"
    ADD EVENT PollLatestReadings
        WITH EVERY 5 SECONDS
        WITH DESTINATION_TOPIC "dashboard/sensors/latest"
        WITH QUERY "SELECT device_id, AVG(value) as avg_value, MAX(ts) as last_seen FROM sensor_data WHERE ts > NOW() - INTERVAL '1 minute' GROUP BY device_id ORDER BY device_id"
```

***

## Table Schema Examples

Create tables optimized for time-series:

```sql wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
-- Partitioned time-series table
CREATE TABLE sensor_data (
    ts TIMESTAMP WITH TIME ZONE,
    sensor_id TEXT,
    value DOUBLE PRECISION
) PARTITIONED BY (ts);

-- Metrics with JSON storage
CREATE TABLE metrics (
    ts TIMESTAMP WITH TIME ZONE,
    topic TEXT,
    data OBJECT(DYNAMIC)
) PARTITIONED BY (ts);
```

***

## Troubleshooting

> **Connection Failed**
>
> * Verify HOST and PORT are correct
> * Check CrateDB is running and accepting connections
> * Verify firewall allows connections on port 5432

> **Authentication Failed**
>
> * Verify USERNAME and PASSWORD are correct
> * Check user has permissions on the DATABASE

> **Query Errors**
>
> * Verify table exists and has correct schema
> * Check column types match inserted data

***

## Next Steps

> **[Data Storage Routes Overview](./overview)**
>
> Compare all storage options.

> **[Snowflake Route](./snowflake)**
>
> Configure Snowflake cloud data warehouse.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.