> ## 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.
</Note>

## Basic Syntax

```lot theme={null}
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

<ParamField path="HOST" type="string" required>
  CrateDB server hostname or IP address.
</ParamField>

<ParamField path="USERNAME" type="string" required>
  Database username.
</ParamField>

<ParamField path="PASSWORD" type="string" required>
  Database password.
</ParamField>

<AccordionGroup>
  <Accordion title="Optional Settings">
    <ParamField path="PORT" type="integer">
      CrateDB PostgreSQL protocol port. Default: 5432.
    </ParamField>

    <ParamField path="DATABASE" type="string">
      Target database/schema name.
    </ParamField>
  </Accordion>
</AccordionGroup>

<Warning>
  **Never hardcode credentials in production.** Use environment variables and encrypted secrets to keep sensitive values out of your route definitions:

  ```lot theme={null}
  WITH HOST GET ENV "DB_HOST"
  WITH USERNAME GET ENV "DB_USER"
  WITH PASSWORD GET SECRET "DB_PASSWORD"
  ```

  See [Environment Variables & Secrets](/mqtt-broker/secrets-and-env) for setup and usage.
</Warning>

***

## Writing Data

<Tabs>
  <Tab title="Basic Time-Series">
    Store sensor readings with timestamps:

    ```lot theme={null}
    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 (?, ?, ?)"
    ```
  </Tab>

  <Tab title="High-Volume Metrics">
    Optimized for high ingestion rates:

    ```lot theme={null}
    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 (?, ?, ?)"
    ```
  </Tab>

  <Tab title="Multiple Tables">
    Route to different tables by topic:

    ```lot theme={null}
    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 (?, ?, ?)"
    ```
  </Tab>

  <Tab title="UPSERT">
    Insert or ignore on conflict. CrateDB supports `ON CONFLICT DO NOTHING` to skip duplicate inserts for idempotent ingestion:

    ```lot theme={null}
    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.
  </Tab>

  <Tab title="UPDATE">
    Modify existing records when a status change arrives:

    ```lot theme={null}
    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}'"
    ```
  </Tab>

  <Tab title="DELETE">
    Remove records on a decommission or cleanup event:

    ```lot theme={null}
    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}'"
    ```
  </Tab>
</Tabs>

<Info>
  **Alternative: STORE IN with Models** — Instead of writing EVENT queries, you can bind a [model](/lot-language/models/overview) directly to this route. Every `PUBLISH MODEL` call automatically inserts a row — no query needed:

  ```lot theme={null}
  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.
</Info>

***

## 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 theme={null}
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 theme={null}
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 theme={null}
-- 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

<AccordionGroup>
  <Accordion title="Connection Failed">
    * Verify HOST and PORT are correct
    * Check CrateDB is running and accepting connections
    * Verify firewall allows connections on port 5432
  </Accordion>

  <Accordion title="Authentication Failed">
    * Verify USERNAME and PASSWORD are correct
    * Check user has permissions on the DATABASE
  </Accordion>

  <Accordion title="Query Errors">
    * Verify table exists and has correct schema
    * Check column types match inserted data
  </Accordion>
</AccordionGroup>

***

## Next Steps

<CardGroup cols={2}>
  <Card title="Data Storage Routes Overview" icon="database" href="./overview">
    Compare all storage options.
  </Card>

  <Card title="Snowflake Route" icon="snowflake" href="./snowflake">
    Configure Snowflake cloud data warehouse.
  </Card>
</CardGroup>