moving the plan

1 files changed, 1623 insertions, 0 deletions

diff --git a/docs/clickhouse-streaming-plan.md b/docs/clickhouse-streaming-plan.md
new file mode 100644
index 0000000..785d8f4
--- /dev/null
+++ b/docs/clickhouse-streaming-plan.md
@@ -0,0 +1,1623 @@
+# 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,
+        },
+        Compression: &clickhouse.Compression{
+            Method: clickhouse.CompressionLZ4,
+        },
+        BlockBufferSize: 10,
+    })
+    if err != nil {
+        return nil, err
+    }
+    return &Client{conn: conn, config: config}, nil
+}
+
+func (c *Client) Close() error {
+    return c.conn.Close()
+}
+```
+
+#### 1.4.3 Batch Writer (`internal/clickhouse/writer.go`)
+
+Key design for 100K-1M events/sec:
+- **Double buffering**: While one buffer is being sent, the other is being filled
+- **Sync.Pool for event rows**: Reduce GC pressure
+- **Async inserts**: ClickHouse async mode for lower latency
+- **Connection pooling**: Multiple connections for parallel inserts
+
+```go
+package clickhouse
+
+import (
+    "context"
+    "sync"
+    "time"
+    
+    "ior/internal/event"
+)
+
+type EventRow struct {
+    TimestampNs     uint64
+    Pid             uint32
+    Tid             uint32
+    Comm            string
+    SyscallName     string
+    TraceId         uint32
+    EventType       uint8
+    RetValue        int64
+    RetType         uint32
+    Fd              int32
+    Filename        string
+    Pathname        string
+    Oldname         string
+    Newname         string
+    Flags           int32
+    DurationNs      uint64
+    DurationToPrev  uint64
+    Hostname        string
+    CollectionId    string
+}
+
+type Writer struct {
+    client       *Client
+    config       Config
+    
+    bufferMu     sync.Mutex
+    buffer       []*EventRow
+    bufferSize   int
+    
+    flushTimer   *time.Timer
+    flushCh      chan struct{}
+    
+    ctx          context.Context
+    cancel       context.CancelFunc
+    wg           sync.WaitGroup
+    
+    pool         sync.Pool
+}
+
+func NewWriter(ctx context.Context, client *Client, config Config) *Writer {
+    ctx, cancel := context.WithCancel(ctx)
+    w := &Writer{
+        client:     client,
+        config:     config,
+        buffer:     make([]*EventRow, 0, config.BatchSize),
+        flushCh:    make(chan struct{}, 1),
+        ctx:        ctx,
+        cancel:     cancel,
+        pool: sync.Pool{
+            New: func() interface{} { return &EventRow{} },
+        },
+    }
+    w.flushTimer = time.AfterFunc(config.FlushTimeout, w.triggerFlush)
+    go w.flushLoop()
+    return w
+}
+
+func (w *Writer) Write(ep *event.Pair) error {
+    row := w.pool.Get().(*EventRow)
+    w.populateRow(row, ep)
+    
+    w.bufferMu.Lock()
+    w.buffer = append(w.buffer, row)
+    shouldFlush := len(w.buffer) >= w.config.BatchSize
+    w.bufferMu.Unlock()
+    
+    if shouldFlush {
+        w.triggerFlush()
+    }
+    return nil
+}
+
+func (w *Writer) triggerFlush() {
+    select {
+    case w.flushCh <- struct{}{}:
+    default:
+    }
+}
+
+func (w *Writer) flushLoop() {
+    for {
+        select {
+        case <-w.flushCh:
+            w.flush()
+        case <-w.ctx.Done():
+            w.flush()
+            return
+        }
+    }
+}
+
+func (w *Writer) flush() {
+    w.bufferMu.Lock()
+    if len(w.buffer) == 0 {
+        w.bufferMu.Unlock()
+        return
+    }
+    buffer := w.buffer
+    w.buffer = make([]*EventRow, 0, w.config.BatchSize)
+    w.bufferMu.Unlock()
+    
+    ctx := clickhouse.Context(w.ctx, clickhouse.WithSettings(clickhouse.Settings{
+        "async_insert":          1,
+        "wait_for_async_insert": 0,
+    }))
+    
+    batch, err := w.client.conn.PrepareBatch(ctx, 
+        "INSERT INTO ior_events (timestamp_ns, pid, tid, comm, syscall_name, ...)")
+    if err != nil {
+        return
+    }
+    
+    for _, row := range buffer {
+        batch.Append(
+            row.TimestampNs, row.Pid, row.Tid, row.Comm,
+            row.SyscallName, row.TraceId, row.EventType,
+            row.RetValue, row.RetType, row.Fd, row.Filename,
+            row.Pathname, row.Oldname, row.Newname, row.Flags,
+            row.DurationNs, row.DurationToPrev, row.Hostname,
+            row.CollectionId,
+        )
+        w.pool.Put(row)
+    }
+    
+    batch.Send()
+}
+
+func (w *Writer) Close() {
+    w.cancel()
+    w.wg.Wait()
+    w.flushTimer.Stop()
+}
+```
+
+#### 1.4.4 Schema Management (`internal/clickhouse/schema.go`)
+
+```go
+package clickhouse
+
+func (c *Client) CreateTableIfNotExists(ctx context.Context) error {
+    ddl := `CREATE TABLE IF NOT EXISTS ior_events (...)`
+    return c.conn.Exec(ctx, ddl)
+}
+```
+
+### 1.5 Integration with Event Loop
+
+Modify `internal/eventloop.go`:
+
+```go
+func (e *eventLoop) run(ctx context.Context, rawCh <-chan []byte) {
+    var chWriter *clickhouse.Writer
+    if flags.Get().ClickHouseEnable {
+        chClient, err := clickhouse.NewClient(ctx, clickhouse.ConfigFromFlags())
+        if err != nil {
+            panic(err)
+        }
+        defer chClient.Close()
+        chWriter = clickhouse.NewWriter(ctx, chClient, clickhouse.ConfigFromFlags())
+        defer chWriter.Close()
+    }
+    
+    for ep := range e.events(ctx, rawCh) {
+        switch {
+        case flags.Get().ClickHouseEnable:
+            chWriter.Write(ep)
+            ep.Recycle()
+        case flags.Get().FlamegraphEnable:
+            e.flamegraph.Ch <- ep
+        // ... rest
+        }
+    }
+}
+```
+
+### 1.6 New Command-Line Flags
+
+Add to `internal/flags/flags.go`:
+
+```go
+type Flags struct {
+    // ... existing fields ...
+    
+    // ClickHouse streaming
+    ClickHouseEnable   bool
+    ClickHouseHost     string
+    ClickHousePort     int
+    ClickHouseDatabase string
+    ClickHouseTable    string
+    ClickHouseUser     string
+    ClickHouseBatchSize int
+    ClickHouseAsync    bool
+}
+
+func parse() {
+    // ... existing flags ...
+    
+    flag.BoolVar(&singleton.ClickHouseEnable, "clickhouse", false, 
+        "Enable ClickHouse streaming output")
+    flag.StringVar(&singleton.ClickHouseHost, "ch-host", "localhost", 
+        "ClickHouse host")
+    flag.IntVar(&singleton.ClickHousePort, "ch-port", 9000, 
+        "ClickHouse native port")
+    flag.StringVar(&singleton.ClickHouseDatabase, "ch-db", "ior", 
+        "ClickHouse database")
+    flag.StringVar(&singleton.ClickHouseTable, "ch-table", "ior_events", 
+        "ClickHouse table")
+    flag.StringVar(&singleton.ClickHouseUser, "ch-user", "", 
+        "ClickHouse user")
+    flag.IntVar(&singleton.ClickHouseBatchSize, "ch-batch", 10000, 
+        "ClickHouse batch size")
+}
+```
+
+---
+
+## Part 2: Parquet File Export
+
+### 2.1 Dependencies
+
+Add to `go.mod`:
+```go
+require (
+    github.com/parquet-go/parquet-go v0.23.0
+)
+```
+
+### 2.2 New Package Structure
+
+```
+internal/
+├── parquet/
+│   ├── writer.go        # Parquet file writer
+│   ├── schema.go        # Parquet schema definition
+│   └── writer_test.go   # Unit tests
+```
+
+### 2.3 Parquet Schema
+
+```go
+package parquet
+
+import "github.com/parquet-go/parquet-go"
+
+type EventRow struct {
+    TimestampNs     uint64 `parquet:"timestamp_ns"`
+    Pid             uint32 `parquet:"pid"`
+    Tid             uint32 `parquet:"tid"`
+    Comm            string `parquet:"comm,dict"`
+    SyscallName     string `parquet:"syscall_name,dict"`
+    TraceId         uint32 `parquet:"trace_id"`
+    EventType       uint8  `parquet:"event_type"`
+    RetValue        int64  `parquet:"ret_value"`
+    RetType         uint32 `parquet:"ret_type"`
+    Fd              int32  `parquet:"fd"`
+    Filename        string `parquet:"filename"`
+    Pathname        string `parquet:"pathname"`
+    Oldname         string `parquet:"oldname"`
+    Newname         string `parquet:"newname"`
+    Flags           int32  `parquet:"flags"`
+    DurationNs      uint64 `parquet:"duration_ns"`
+    DurationToPrev  uint64 `parquet:"duration_to_prev_ns"`
+    Hostname        string `parquet:"hostname,dict"`
+    CollectionId    string `parquet:"collection_id"`
+}
+```
+
+**Parquet Optimizations:**
+- `dict` encoding for low-cardinality strings (`comm`, `syscall_name`, `hostname`)
+- Zstd compression (same as current GOB files)
+- Row groups of ~128MB for efficient querying
+- Column pruning - ClickHouse only reads needed columns
+
+### 2.4 Implementation
+
+#### 2.4.1 Writer (`internal/parquet/writer.go`)
+
+```go
+package parquet
+
+import (
+    "context"
+    "fmt"
+    "os"
+    "sync"
+    "time"
+    
+    "github.com/parquet-go/parquet-go"
+    "github.com/parquet-go/parquet-go/compress/zstd"
+    "ior/internal/event"
+)
+
+type Writer struct {
+    file       *os.File
+    writer     *parquet.GenericWriter[EventRow]
+    
+    bufferMu   sync.Mutex
+    buffer     []EventRow
+    bufferSize int
+    
+    rowGroupSize int
+    
+    ctx    context.Context
+    cancel context.CancelFunc
+}
+
+type WriterConfig struct {
+    Filename      string
+    BatchSize     int
+    RowGroupSize  int
+    Compression   parquet.Compression
+}
+
+func DefaultWriterConfig() WriterConfig {
+    hostname, _ := os.Hostname()
+    return WriterConfig{
+        Filename:     fmt.Sprintf("%s-%s.parquet", hostname, time.Now().Format("2006-01-02_15:04:05")),
+        BatchSize:    10000,
+        RowGroupSize: 100000,
+        Compression:  parquet.Zstd,
+    }
+}
+
+func NewWriter(ctx context.Context, config WriterConfig) (*Writer, error) {
+    file, err := os.Create(config.Filename)
+    if err != nil {
+        return nil, err
+    }
+    
+    writer := parquet.NewGenericWriter[EventRow](file,
+        parquet.Compression(&zstd.Codec{Level: zstd.DefaultLevel}),
+        parquet.RowGroupSize(config.RowGroupSize),
+    )
+    
+    ctx, cancel := context.WithCancel(ctx)
+    return &Writer{
+        file:         file,
+        writer:       writer,
+        buffer:       make([]EventRow, 0, config.BatchSize),
+        bufferSize:   config.BatchSize,
+        rowGroupSize: config.RowGroupSize,
+        ctx:          ctx,
+        cancel:       cancel,
+    }, nil
+}
+
+func (w *Writer) Write(ep *event.Pair) error {
+    row := EventRow{
+        TimestampNs:    ep.EnterEv.GetTime(),
+        Pid:           ep.EnterEv.GetPid(),
+        Tid:           ep.EnterEv.GetTid(),
+        Comm:          ep.Comm,
+        SyscallName:   ep.EnterEv.GetTraceId().Name(),
+        DurationNs:     ep.Duration,
+        DurationToPrev: ep.DurationToPrev,
+    }
+    
+    w.bufferMu.Lock()
+    w.buffer = append(w.buffer, row)
+    shouldFlush := len(w.buffer) >= w.bufferSize
+    w.bufferMu.Unlock()
+    
+    if shouldFlush {
+        return w.flush()
+    }
+    return nil
+}
+
+func (w *Writer) flush() error {
+    w.bufferMu.Lock()
+    defer w.bufferMu.Unlock()
+    
+    if len(w.buffer) == 0 {
+        return nil
+    }
+    
+    _, err := w.writer.Write(w.buffer)
+    w.buffer = w.buffer[:0]
+    return err
+}
+
+func (w *Writer) Close() error {
+    w.cancel()
+    if err := w.flush(); err != nil {
+        return err
+    }
+    if err := w.writer.Close(); err != nil {
+        return err
+    }
+    return w.file.Close()
+}
+```
+
+### 2.5 ClickHouse Import Command
+
+After generating a Parquet file:
+
+```bash
+# Local file import
+clickhouse-client --query "
+  INSERT INTO ior.ior_events 
+  FROM INFILE 'hostname-2024-01-15_10:30:00.parquet' 
+  FORMAT Parquet"
+
+# Or via HTTP (remote server)
+curl -X POST 'http://clickhouse:8123/?query=INSERT+INTO+ior.ior_events+FORMAT+Parquet' \
+  --data-binary @hostname-2024-01-15_10:30:00.parquet
+```
+
+### 2.6 New Command-Line Flags
+
+Add to `internal/flags/flags.go`:
+
+```go
+// Parquet file output
+ParquetEnable    bool
+ParquetFilename  string
+ParquetBatchSize int
+ParquetRowGroupSize int
+```
+
+---
+
+## Part 3: Shared Components
+
+### 3.1 Event-to-Row Converter
+
+Both ClickHouse and Parquet need to convert `event.Pair` to a row format:
+
+```go
+// internal/export/convert.go
+
+package export
+
+import (
+    "ior/internal/event"
+    "ior/internal/types"
+)
+
+type EventRow struct {
+    TimestampNs     uint64
+    Pid             uint32
+    Tid             uint32
+    Comm            string
+    SyscallName     string
+    TraceId         uint32
+    EventType       uint8
+    RetValue        int64
+    RetType         uint32
+    Fd              int32
+    Filename        string
+    Pathname        string
+    Oldname         string
+    Newname         string
+    Flags           int32
+    DurationNs      uint64
+    DurationToPrev  uint64
+    Hostname        string
+    CollectionId    string
+}
+
+func PairToRow(ep *event.Pair, hostname, collectionId string) EventRow {
+    row := EventRow{
+        TimestampNs:    ep.EnterEv.GetTime(),
+        Pid:           ep.EnterEv.GetPid(),
+        Tid:           ep.EnterEv.GetTid(),
+        Comm:          ep.Comm,
+        SyscallName:   ep.EnterEv.GetTraceId().Name(),
+        TraceId:       uint32(ep.EnterEv.GetTraceId()),
+        DurationNs:     ep.Duration,
+        DurationToPrev: ep.DurationToPrev,
+        Hostname:      hostname,
+        CollectionId:  collectionId,
+    }
+    
+    switch enter := ep.EnterEv.(type) {
+    case *types.OpenEvent:
+        row.EventType = types.ENTER_OPEN_EVENT
+        row.Filename = types.StringValue(enter.Filename[:])
+        row.Flags = enter.Flags
+    case *types.FdEvent:
+        row.EventType = types.ENTER_FD_EVENT
+        row.Fd = enter.Fd
+    }
+    
+    if ret, ok := ep.ExitEv.(*types.RetEvent); ok {
+        row.RetValue = ret.Ret
+        row.RetType = ret.RetType
+    }
+    
+    return row
+}
+```
+
+### 3.2 Output Mode Selector
+
+```go
+// internal/output/output.go
+
+package output
+
+type Output interface {
+    Write(ep *event.Pair) error
+    Close() error
+}
+
+func NewOutput(ctx context.Context, flags flags.Flags) (Output, error) {
+    switch {
+    case flags.ClickHouseEnable:
+        return clickhouse.NewWriter(ctx, ...)
+    case flags.ParquetEnable:
+        return parquet.NewWriter(ctx, ...)
+    case flags.FlamegraphEnable:
+        return flamegraph.NewCollector(ctx, ...)
+    default:
+        return &consoleOutput{}, nil
+    }
+}
+```
+
+---
+
+## Part 4: Performance Considerations
+
+### 4.1 Throughput Targets: 100K-1M events/sec
+
+| Component | Strategy |
+|-----------|----------|
+| **Memory** | `sync.Pool` for EventRow reuse, avoid allocations in hot path |
+| **Buffering** | Double buffering: fill one buffer while sending another |
+| **Batching** | Batch inserts: 10K-100K rows per batch |
+| **Compression** | LZ4 for streaming (fast), Zstd for files (compact) |
+| **Concurrency** | Multiple writer goroutines with separate connections |
+| **Backpressure** | Drop events if buffer full (configurable), report stats |
+
+### 4.2 Memory Budget
+
+For 1M events/sec with 10KB per event row:
+- Raw: 10GB/sec (too high)
+- With batching and buffering: ~100MB buffer is sufficient
+
+Buffer sizing:
+- 100K events × ~200 bytes per row = ~20MB per buffer
+- Double buffering = ~40MB total
+
+### 4.3 ClickHouse Server-Side Settings
+
+```sql
+SET max_insert_block_size = 1048576;
+SET max_block_size = 65536;
+SET async_insert_max_data_size = 10000000;
+SET async_insert_busy_timeout_ms = 1000;
+```
+
+---
+
+## Part 5: Testing Strategy
+
+### 5.1 Unit Tests
+
+```
+internal/clickhouse/
+├── client_test.go      # Mock server tests
+├── writer_test.go      # Buffer management, batch logic
+└── schema_test.go      # DDL generation
+
+internal/parquet/
+├── writer_test.go      # File writing, schema validation
+└── convert_test.go     # Event to row conversion
+```
+
+### 5.2 Integration Tests
+
+```bash
+docker run -d --name clickhouse -p 9000:9000 clickhouse/clickhouse-server
+make test-integration-clickhouse
+make test-integration-parquet
+```
+
+### 5.3 Benchmark Tests
+
+```go
+// internal/bench_test.go
+func BenchmarkClickHouseWriter(b *testing.B) {}
+
+func BenchmarkParquetWriter(b *testing.B) {}
+```
+
+---
+
+## Part 6: Implementation Order
+
+### Phase 1: Foundation
+1. Add dependencies to `go.mod`
+2. Create `internal/export/convert.go` - shared row conversion
+3. Create `internal/output/output.go` - output interface
+
+### Phase 2: Parquet Export
+1. Create `internal/parquet/` package
+2. Add parquet flags to `internal/flags/`
+3. Integrate with `internal/eventloop.go`
+4. Add unit tests
+5. Test ClickHouse import
+
+### Phase 3: ClickHouse Streaming
+1. Create `internal/clickhouse/` package
+2. Add ClickHouse flags to `internal/flags/`
+3. Implement double-buffering writer
+4. Integrate with `internal/eventloop.go`
+5. Add unit tests and integration tests
+
+### Phase 4: Polish
+1. Performance benchmarks and optimization
+2. Documentation
+3. Error handling and recovery
+4. Metrics/monitoring integration
+
+---
+
+## Part 7: Usage Examples
+
+### Parquet Export
+
+```bash
+# Capture to Parquet file
+ior -parquet -name my_trace -duration 60
+
+# Import to ClickHouse
+clickhouse-client --query "
+  INSERT INTO ior.ior_events 
+  FROM INFILE 'myhost-2024-01-15_10:30:00.parquet' 
+  FORMAT Parquet"
+```
+
+### Real-time Streaming
+
+```bash
+# Stream to ClickHouse
+ior -clickhouse -ch-host ch-server.example.com -ch-db ior -duration 300
+
+# With authentication
+ior -clickhouse -ch-host ch.example.com -ch-user app -ch-password secret
+```
+
+### Query Examples
+
+```sql
+-- Top 10 slowest syscalls by average duration
+SELECT 
+    syscall_name,
+    count() as total,
+    avg(duration_ns) as avg_duration,
+    quantile(0.99)(duration_ns) as p99_duration
+FROM ior_events
+WHERE timestamp_ns > now() - INTERVAL 1 HOUR
+GROUP BY syscall_name
+ORDER BY avg_duration DESC
+LIMIT 10;
+
+-- Events per process
+SELECT 
+    pid,
+    comm,
+    count() as total_events,
+    sum(duration_ns) / 1e9 as total_duration_sec
+FROM ior_events
+WHERE collection_id = 'uuid-here'
+GROUP BY pid, comm
+ORDER BY total_events DESC;
+
+-- Per-thread I/O activity (leverages bloom filter on tid)
+SELECT 
+    pid,
+    tid,
+    comm,
+    count() as total_events,
+    sum(duration_ns) / 1e9 as total_duration_sec,
+    uniqExact(syscall_name) as unique_syscalls
+FROM ior_events
+WHERE collection_id = 'uuid-here'
+GROUP BY pid, tid, comm
+ORDER BY total_events DESC
+LIMIT 50;
+
+-- Thread with most I/O latency (bloom filter helps for specific tid lookup)
+SELECT 
+    pid, tid, comm,
+    sum(duration_ns) / 1e6 as total_latency_ms,
+    avg(duration_ns) as avg_latency_ns,
+    count() as event_count
+FROM ior_events
+WHERE tid = 12345  -- bloom filter index used here
+  AND collection_id = 'uuid-here'
+GROUP BY pid, tid, comm;
+
+-- Most accessed files
+SELECT 
+    filename,
+    count() as access_count,
+    sum(duration_ns) / 1e9 as total_duration_sec
+FROM ior_events
+WHERE filename != ''
+GROUP BY filename
+ORDER BY access_count DESC
+LIMIT 20;
+```
+
+---
+
+## Appendix A: File Sizes Estimation
+
+For 1M events:
+| Format | Size (estimated) |
+|--------|------------------|
+| GOB + zstd (current) | ~50-80 MB |
+| Parquet + zstd | ~40-60 MB |
+| RowBinary | ~60-80 MB |
+| JSON (not recommended) | ~200-300 MB |
+
+---
+
+## Appendix C: ClickHouse Optimization Strategies
+
+### C.1 High Cardinality Optimization
+
+**Problem**: Fields like `tid` (thread ID) can have tens of thousands of unique values, making standard indexing inefficient.
+
+| Technique | Description | Best For |
+|-----------|-------------|----------|
+| **Bloom Filter Index** | Probabilistic index for point lookups (~1% storage overhead) | `WHERE tid = 12345` |
+| **Minmax Index** | Stores min/max values per granule | Range queries on numeric fields |
+| **Set Index** | Stores unique values per granule (limited to ~N values) | Low-medium cardinality |
+| **Token Bloom Filter** | Bloom filter on tokens in string | `WHERE hasToken(filename, 'log')` |
+
+```sql
+-- Bloom filter for point lookups on high-cardinality fields
+INDEX idx_tid tid TYPE bloom_filter(0.01) GRANULARITY 4,
+INDEX idx_pid pid TYPE bloom_filter(0.01) GRANULARITY 4,
+
+-- Minmax for range queries on timestamps (already default, but explicit)
+INDEX idx_duration duration_ns TYPE minmax GRANULARITY 1,
+
+-- Token bloom filter for filename substring searches
+INDEX idx_filename_tokens filename TYPE tokenbf_v1(512, 3, 0) GRANULARITY 4
+```
+
+**Bloom Filter Parameters**:
+- `0.01` = 1% false positive rate (lower = more accurate, larger index)
+- `GRANULARITY 4` = index covers 4 granules (8192 × 4 = 32768 rows)
+
+#### C.1.1 Are Bloom Filter Results Inexact?
+
+**No - query results are ALWAYS exact.** Bloom filters only affect performance, not correctness.
+
+```
+How Bloom Filter Indices Work:
++---------------------------------------------------------------+
+| QUERY: SELECT * FROM ior_events WHERE tid = 12345             |
++---------------------------------------------------------------+
+                              |
+                              v
++---------------------------------------------------------------+
+| STEP 1: Check bloom filter index for each granule             |
+|                                                               |
+| Granule 1 (rows 1-8192):     Bloom says "MAYBE" -> READ IT    |
+| Granule 2 (rows 8193-16384): Bloom says "DEFINITELY NOT"      |
+|                              -> SKIP                          |
+| Granule 3 (rows 16385-24576): Bloom says "MAYBE" -> READ IT   |
+| ...                                                           |
++---------------------------------------------------------------+
+                              |
+                              v
++---------------------------------------------------------------+
+| STEP 2: Read the "MAYBE" granules and filter EXACTLY          |
+|                                                               |
+| Result: ALL rows where tid = 12345 (no false positives)       |
++---------------------------------------------------------------+
+```
+
+**False Positive Impact**: The bloom filter might say "MAYBE contains tid=12345" for a granule that doesn't actually have it. This causes ClickHouse to read that granule unnecessarily - wasting I/O but NOT affecting result correctness.
+
+**False Negatives**: Impossible. Bloom filters never produce false negatives. If the value exists, the bloom filter will always say "MAYBE".
+
+**Trade-off**:
+| Bloom Filter Size | False Positive Rate | Storage Overhead | Unnecessary Reads |
+|-------------------|---------------------|------------------|-------------------|
+| Smaller | Higher (~5%) | ~0.5% | More wasted I/O |
+| Larger | Lower (~0.01%) | ~2% | Less wasted I/O |
+
+**Recommendation**: Use `bloom_filter(0.01)` for 1% false positive rate - good balance between index size and read efficiency.
+
+#### C.1.2 Projections vs Materialized Views
+
+Both optimize query performance, but work differently:
+
+| Aspect | Projection | Materialized View |
+|--------|------------|-------------------|
+| **Data storage** | Same table, different physical order | Separate table |
+| **Automatic sync** | Yes - part of the same table | Yes - triggered on INSERT |
+| **Query rewrite** | Automatic - ClickHouse picks best projection | Must query MV explicitly |
+| **Aggregation** | Can pre-aggregate | Often used for aggregation |
+| **Space overhead** | Copies ALL columns (unless aggregated) | Only stores aggregated result |
+| **Best for** | Different sort orders, same columns | Pre-computed aggregations |
+
+---
+
+### C.1.3 Projections Explained
+
+**What is a Projection?**
+
+A projection is an alternative physical representation of the SAME data within the SAME table. Think of it as "same data, sorted differently for different queries."
+
+**Example Problem**: Your main table is ordered by `(timestamp_ns, pid, tid)` for time-range queries. But you also need fast queries by thread:
+
+```sql
+-- This query is SLOW because tid is last in ORDER BY
+SELECT * FROM ior_events WHERE tid = 12345;
+-- Must scan almost all data!
+```
+
+**Solution - Add a Projection**:
+
+```sql
+CREATE TABLE ior_events (
+    timestamp_ns UInt64,
+    pid UInt32,
+    tid UInt32,
+    comm String,
+    syscall_name String,
+    duration_ns UInt64,
+    -- ... other columns ...
+)
+ENGINE = MergeTree()
+ORDER BY (timestamp_ns, pid, tid)  -- Primary order: time-first
+PARTITION BY toYYYYMMDD(fromUnixTimestamp64Nano(timestamp_ns));
+
+-- Add projection for thread-centric queries
+ALTER TABLE ior_events ADD PROJECTION proj_by_thread (
+    SELECT * ORDER BY (pid, tid, timestamp_ns)
+);
+
+-- Materialize the projection (build it for existing data)
+ALTER TABLE ior_events MATERIALIZE PROJECTION proj_by_thread;
+```
+
+**How it works**:
+
+```
+INSERT INTO ior_events VALUES (1000, 1, 100, 'app', 'read', 50);
+INSERT INTO ior_events VALUES (2000, 1, 100, 'app', 'write', 30);
+INSERT INTO ior_events VALUES (1500, 2, 200, 'db', 'read', 40);
+
+Main storage (ORDER BY timestamp_ns, pid, tid):
++--------------+-----+-----+------+-------------+-------------+
+| timestamp_ns | pid | tid | comm | syscall_name| duration_ns |
++--------------+-----+-----+------+-------------+-------------+
+| 1000         | 1   | 100 | app  | read        | 50          |
+| 1500         | 2   | 200 | db   | read        | 40          |
+| 2000         | 1   | 100 | app  | write       | 30          |
++--------------+-----+-----+------+-------------+-------------+
+
+Projection proj_by_thread (ORDER BY pid, tid, timestamp_ns):
++--------------+-----+-----+------+-------------+-------------+
+| timestamp_ns | pid | tid | comm | syscall_name| duration_ns |
++--------------+-----+-----+------+-------------+-------------+
+| 1000         | 1   | 100 | app  | read        | 50          |
+| 2000         | 1   | 100 | app  | write       | 30          |
+| 1500         | 2   | 200 | db   | read        | 40          |
++--------------+-----+-----+------+-------------+-------------+
+```
+
+**Query optimization**:
+
+```sql
+-- Query 1: Time range - uses main storage
+SELECT * FROM ior_events 
+WHERE timestamp_ns BETWEEN 1000 AND 1800;
+-- Reads rows in order: (1000,1,100), (1500,2,200) - efficient!
+
+-- Query 2: Thread lookup - uses projection AUTOMATICALLY
+SELECT * FROM ior_events WHERE tid = 100;
+-- ClickHouse sees projection has better ORDER BY for this query
+-- Uses proj_by_thread: rows (1,100,1000), (1,100,2000) are adjacent!
+```
+
+**Aggregating Projection** (smaller storage):
+
+```sql
+-- Pre-aggregated projection - stores only aggregated data
+ALTER TABLE ior_events ADD PROJECTION proj_hourly_stats (
+    SELECT 
+        toStartOfHour(fromUnixTimestamp64Nano(timestamp_ns)) as hour,
+        syscall_name,
+        count() as event_count,
+        sum(duration_ns) as total_duration,
+        avg(duration_ns) as avg_duration
+    GROUP BY hour, syscall_name
+);
+
+ALTER TABLE ior_events MATERIALIZE PROJECTION proj_hourly_stats;
+```
+
+**Storage comparison**:
+```
+Main table:        1,000,000,000 rows × ~200 bytes = ~200 GB
+proj_by_thread:    1,000,000,000 rows × ~200 bytes = ~200 GB (full copy)
+proj_hourly_stats: ~100,000 rows × ~50 bytes = ~5 MB (aggregated!)
+```
+
+**When to use projections**:
+- Different access patterns on the same raw data
+- Query patterns known upfront
+- Can afford storage overhead (projections copy data)
+
+---
+
+### C.1.4 Materialized Views Explained
+
+**What is a Materialized View?**
+
+A materialized view is a SEPARATE table that is automatically populated and maintained when data is inserted into the source table.
+
+**Example Problem**: You frequently query per-thread statistics:
+
+```sql
+-- This is slow - scans billions of rows every time
+SELECT pid, tid, comm, count(), sum(duration_ns)