ClickHouse

Technical Preview

Overview

A client-mode connector for the ClickHouse HTTP interface that streams MQTT telemetry into ClickHouse tables and runs SQL from MQTT. It works with self-hosted ClickHouse and ClickHouse Cloud.

Catalog connector event	Endpoint	Notes
POST insert (JSONEachRow)	`POST /?query=INSERT INTO {table} FORMAT JSONEachRow&async_insert=1&wait_for_async_insert=1`	The MQTT payload is the row. Async inserts let the server batch many small writes.
POST SQL	`POST /?default_format=JSON`	Raw SQL in the body: `SELECT … FORMAT JSON`, `CREATE TABLE`, `INSERT … VALUES`
GET query	`GET /?query=&default_format=JSON`	Read-only SQL in the URL (ClickHouse treats every GET as read-only)
GET ping	`GET /ping`	Health check, `200 Ok.`

Default BASE_ADDRESS is https://YOUR_CLICKHOUSE_HOST:8443 (ClickHouse Cloud and TLS). For a self-hosted server on plain HTTP, use http://YOUR_CLICKHOUSE_HOST:8123 and set USE_SSL "false". Topics use data/clickhouse/<object>/<verb>.

GET is read-only. ClickHouse rejects CREATE, INSERT, or DROP sent with GET (Cannot execute query in readonly mode). Send those through the ExecSql event, which uses POST.

Auth

The connector sends HTTP Basic auth from WITH USERNAME and WITH PASSWORD in REST_API_CONFIG. Store the password as a broker secret named CLICKHOUSE_PASSWORD and read it with GET SECRET; never commit it. ClickHouse Cloud users are usually default with the password shown when the service is created. ClickHouse also accepts the X-ClickHouse-User and X-ClickHouse-Key headers if you prefer them.

Basic Syntax

DEFINE ROUTE ClickhouseRestApi WITH TYPE REST_API
    ADD REST_API_CONFIG
        WITH BASE_ADDRESS "https://YOUR_CLICKHOUSE_HOST:8443"
        WITH ENABLE_CLIENT "true"
        WITH USE_SSL "true"
        WITH REQUEST_TIMEOUT '60'
        WITH DEFAULT_CONTENT_TYPE "application/json"
        WITH USERNAME "default"
        WITH PASSWORD GET SECRET "CLICKHOUSE_PASSWORD"
    ADD EVENT Ping
        WITH SOURCE_TOPIC "data/clickhouse/ping"
        WITH DESTINATION_TOPIC "data/clickhouse/ping/response"
        WITH METHOD "GET"
        WITH ENDPOINT "/ping"
    ADD EVENT InsertJsonEachRow
        WITH SOURCE_TOPIC "data/clickhouse/rows/insert"
        WITH DESTINATION_TOPIC "data/clickhouse/rows/insert/response"
        WITH METHOD "POST"
        WITH ENDPOINT "/?query=INSERT%20INTO%20{value.json.table}%20FORMAT%20JSONEachRow&async_insert=1&wait_for_async_insert=1&input_format_skip_unknown_fields=1&date_time_input_format=best_effort"
        WITH HEADERS
            ADD "Content-Type" "application/json"
        WITH BODY "{payload}"
    ADD EVENT ExecSql
        WITH SOURCE_TOPIC "data/clickhouse/sql/exec"
        WITH DESTINATION_TOPIC "data/clickhouse/sql/exec/response"
        WITH METHOD "POST"
        WITH ENDPOINT "/?default_format=JSON"
        WITH HEADERS
            ADD "Content-Type" "text/plain; charset=utf-8"
        WITH BODY "{payload}"
    ADD EVENT QueryJson
        WITH SOURCE_TOPIC "data/clickhouse/query/json"
        WITH DESTINATION_TOPIC "data/clickhouse/query/json/response"
        WITH METHOD "GET"
        WITH ENDPOINT "/?query={value.json.query}&default_format=JSON"
        WITH HEADERS
            ADD "Accept" "application/json"

Example MQTT payloads

Insert a row (topic data/clickhouse/rows/insert). The payload is the JSON row. The table field picks the target table and is skipped as a column because the endpoint sets input_format_skip_unknown_fields=1:

{ "table": "sensor_readings", "ts": "2026-10-04 22:15:00.000", "site": "porto", "value": 21.5 }

Create the table first, for example through ExecSql:

CREATE TABLE IF NOT EXISTS sensor_readings (ts DateTime64(3), site String, value Float64) ENGINE = MergeTree ORDER BY ts

Run SQL (topic data/clickhouse/sql/exec). Publish the raw SQL text, not JSON:

SELECT site, avg(value) AS avg_value FROM sensor_readings GROUP BY site FORMAT JSON

The response topic receives ClickHouse’s JSON, for example {"meta":[…],"data":[{"site":"porto","avg_value":21.5}],"rows":1,…}. DDL and INSERT … VALUES return an empty body with HTTP 200. Read-only query (topic data/clickhouse/query/json). URL-encode the SQL before publishing:

{ "query": "SELECT%20count()%20FROM%20sensor_readings" }

Health (topic data/clickhouse/ping) takes any payload, such as {}.

Testing

The factory smoke ran against a local ClickHouse 26.10 server with a password-protected user. GET /ping returned 200, CREATE TABLE through a POST SQL body returned 200, a JSONEachRow async insert from a JSON payload with an extra table field returned 200, and the rows came back through POST … FORMAT JSON and the read-only GET query (both 200). A wrong password returned 403 and malformed JSON returned 400, and both error bodies reach the response topic.

Quick Start

Coreflux MQTT Broker

LoT Language

Coreflux HUB

VS Code - LoT Notebooks

Coreflux Portal

Using AI with Coreflux

Product Updates

Overview

Auth

Basic Syntax

Example MQTT payloads

Testing

​Overview

​Auth

​Basic Syntax

​Example MQTT payloads

​Testing

​Related

Overview

Auth

Basic Syntax

Example MQTT payloads

Testing

Related