# clickhouse

Buffer events until `buffer` count or `interval` seconds, then ship them to ClickHouse via the HTTP `INSERT FORMAT JSONEachRow` interface.

On startup the plugin automatically creates the target table (using `CREATE TABLE IF NOT EXISTS`) with the schema below. If ClickHouse is unreachable, the pipeline fails to start.

## Table Schema

```sql
CREATE TABLE IF NOT EXISTS <database>.<table>
(
    timestamp   DateTime64(3, 'UTC'),
    message     String,
    tags        Array(String),
    host        String,
    process     String,
    kind        String,
    severity    Int8,
    _id         String,
    raw         String
)
ENGINE = ReplacingMergeTree()
PARTITION BY toYYYYMM(timestamp)
ORDER BY (_id, timestamp)
```

Key fields are extracted into typed columns for fast filtering. The full event JSON is stored in `raw` for ad-hoc drill-down. `ReplacingMergeTree` deduplicates rows by `_id` on background merges — safe for journal replay.

!!!info
Dedup is eventual. Queries may see duplicate rows until ClickHouse background merges run. Use `SELECT ... FINAL` to force dedup at query time.
!!!

## Config

**Required:**
- `table`: Target table name

**Optional:**
- `endpoint`: ClickHouse HTTP endpoint (default: `http://localhost:8123`)
- `database`: Target database (default: `default`)
- `buffer`: Events to buffer before flushing (default: `1000`)
- `interval`: Seconds between flushes (default: `60`)
- `username`: Basic auth username
- `password`: Basic auth password

## Field Extraction

| Column | Source | Fallback |
|--------|--------|---------|
| `timestamp` | `event.timestamp` (epoch ms) | current wall clock |
| `message` | `event.message` | `""` |
| `tags` | `event.tags` (array) | `[]` |
| `host` | `event.host.name` | `""` |
| `process` | `event.process.name` | `""` |
| `kind` | `event.event.kind` | `""` |
| `severity` | `event.event.severity` | `7` (debug) |
| `_id` | `event._id` | `{host}:{kind}:{timestamp}:{message_prefix}` |
| `raw` | full event JSON | — |

## Example

```yaml
pipeline:
  clickhouse:
    module: clickhouse
    outputs: []
    config:
      endpoint: http://localhost:8123
      table: logbus
      database: default
      buffer: 1000
      interval: 10
```

With authentication (e.g. ClickHouse Cloud):

```yaml
pipeline:
  clickhouse:
    module: clickhouse
    outputs: []
    config:
      endpoint: https://abc123.clickhouse.cloud:8443
      table: logbus
      database: default
      username: default
      password: your-password
      buffer: 1000
      interval: 10
```

!!!warning
This plugin is a sink and does not emit events downstream. Declare the stage with `outputs: []` to avoid a dead-end pipeline warning.
!!!