path: root/docs/clickhouse-streaming-plan.md

# ClickHouse Integration Plan for IOR

This document outlines the implementation plan for integrating ClickHouse database output into IOR, supporting both real-time streaming and batch file export.

## Overview

| Mode | Format | Use Case | Expected Throughput |
|------|--------|----------|---------------------|
| **Streaming** | Native TCP protocol | Real-time ingestion to ClickHouse | 100K-1M events/sec |
| **File Dump** | Parquet | Batch export for later import | Same (offline processing) |

## Architecture

### Current Data Flow
```
BPF → Ring Buffer → eventLoop → event.Pair → [flamegraph workers | console output]
```

### Proposed Data Flow
```
BPF → Ring Buffer → eventLoop → event.Pair → [clickhouse-stream | parquet-writer | flamegraph | console]
```

---

## Part 1: ClickHouse Streaming (Native TCP Protocol)

### 1.1 Dependencies

Add to `go.mod`:
```go
require (
    github.com/ClickHouse/clickhouse-go/v2 v2.23.0
)
```

### 1.2 New Package Structure

```
internal/
├── clickhouse/
│   ├── client.go        # Connection management, connection pooling
│   ├── schema.go        # Table schema definitions and DDL
│   ├── writer.go        # Batch writer with buffering
│   ├── config.go        # Configuration (host, port, database, table)
│   └── client_test.go   # Unit tests
```

### 1.3 ClickHouse Table Schema

```sql
CREATE TABLE ior_events (
    timestamp_ns    UInt64,           -- Event timestamp (nanoseconds)
    pid             UInt32,           -- Process ID (high cardinality, no LowCardinality)
    tid             UInt32,           -- Thread ID (high cardinality, no LowCardinality)
    comm            LowCardinality(String),
    syscall_name    LowCardinality(String),
    trace_id        UInt32,
    event_type      UInt8,            -- ENTER_OPEN_EVENT, EXIT_OPEN_EVENT, etc.
    
    -- Result
    ret_value       Int64,            -- Syscall return value
    ret_type        UInt32,           -- Return type classification
    
    -- File information
    fd              Int32,
    filename        String,
    pathname        String,
    oldname         String,
    newname         String,
    
    -- Flags and metadata
    flags           Int32,
    
    -- Calculated fields (from event.Pair)
    duration_ns     UInt64,           -- Duration of syscall
    duration_to_prev_ns UInt64,       -- Time since previous syscall
    
    -- Additional context
    hostname        LowCardinality(String),
    collection_id   UUID,             -- Groups events from same collection run
    ingested_at     DateTime64(3) DEFAULT now64(3),

    -- Secondary indices for high-cardinality fields
    INDEX idx_pid pid TYPE bloom_filter(0.01) GRANULARITY 4,
    INDEX idx_tid tid TYPE bloom_filter(0.01) GRANULARITY 4
)
ENGINE = MergeTree()
PARTITION BY toYYYYMMDD(fromUnixTimestamp64Nano(timestamp_ns))
ORDER BY (timestamp_ns, pid, tid)
SETTINGS index_granularity = 8192;
```

**Cardinality Considerations:**
| Field | Cardinality | Encoding | Reason |
|-------|-------------|----------|--------|
| `pid` | High (thousands) | Plain UInt32 | PIDs can range into thousands per server; bloom filter index for point lookups |
| `tid` | Very High (tens of thousands) | Plain UInt32 | TIDs are numerous in threaded workloads; bloom filter index for point lookups |
| `comm` | Low (hundreds) | LowCardinality | Limited number of unique process names |
| `syscall_name` | Very Low (~100) | LowCardinality | Fixed set of syscalls |
| `hostname` | Very Low | LowCardinality | Usually single host per collection |
| `filename` | Medium-High | Plain String | Depends on workload; could use token bloom filter |

**Optimization Notes:**
- **Bloom filter indices** on `pid` and `tid` enable efficient point lookups on these high-cardinality fields without bloating storage
- `LowCardinality` only for truly low-cardinality fields (`comm`, `syscall_name`, `hostname`)
- Partitioning by day for efficient time-based queries and TTL
- Ordering by `(timestamp_ns, pid, tid)` for time-range queries and per-process/thread analysis
- `collection_id` UUID to group events from the same tracing session

### 1.4 Implementation Details

#### 1.4.1 Configuration (`internal/clickhouse/config.go`)

```go
package clickhouse

type Config struct {
    Host         string        // ClickHouse host (default: localhost)
    Port         int           // ClickHouse port (default: 9000)
    Database     string        // Database name (default: ior)
    Table        string        // Table name (default: ior_events)
    User         string        // Username
    Password     string        // Password
    BatchSize    int           // Events per batch (default: 10000)
    FlushTimeout time.Duration // Max time before flush (default: 1s)
    MaxOpenConns int           // Connection pool size (default: 4)
    Async        bool          // Enable async inserts (default: true for high throughput)
}

func DefaultConfig() Config {
    return Config{
        Host:         "localhost",
        Port:         9000,
        Database:     "ior",
        Table:        "ior_events",
        BatchSize:    10000,
        FlushTimeout: time.Second,
        MaxOpenConns: 4,
        Async:        true,
    }
}

func ConfigFromFlags() Config {
    // Read from command-line flags
}
```

#### 1.4.2 Client (`internal/clickhouse/client.go`)

```go
package clickhouse

import (
    "context"
    "github.com/ClickHouse/clickhouse-go/v2"
    "github.com/ClickHouse/clickhouse-go/v2/lib/driver"
)

type Client struct {
    conn   driver.Conn
    config Config
}

func NewClient(ctx context.Context, config Config) (*Client, error) {
    conn, err := clickhouse.Open(&clickhouse.Options{
        Addr: []string{fmt.Sprintf("%s:%d", config.Host, config.Port)},
        Auth: clickhouse.Auth{
            Database: config.Database,
            Username: config.User,
            Password: config.Password,
        },
        MaxOpenConns:   config.MaxOpenConns,
        MaxIdleConns:   config.MaxOpenConns,
        ConnMaxLifetime: time.Hour,
        DialTimeout:    time.Second * 10,
        Settings: clickhouse.Settings{
            "max_execution_time": 60,
        },
        Compressio