e2e.go (customer)

e2e.go

// Read-your-own-write: why "write a row, then read it back" takes 200ms+
// when the target is <20ms.
//
// End-to-end latency = write ack + visibility delay + read. With sync
// single-row inserts the write path drowns everything (scenario A - see the
// write repro for why). With async_insert + wait_for_async_insert=0 writes
// are ~ms and e2e is bounded by the VISIBILITY WINDOW: a row only becomes
// readable when the server flushes its async buffer, every
// async_insert_busy_timeout_ms (default 200). A write lands at a random
// point inside the window, so median e2e ≈ window/2 + one read (scenario B).
// Shrink the window to hit the target (scenario C).
//
// Trade-off: a smaller window means more frequent flushes and more small
// parts to merge - tune it against your real write volume.
//
// Setup:
//
//	CREATE TABLE benchmark_test (id UInt64, value Array(Float64)) ENGINE = MergeTree ORDER BY id;
//
// Run:
//
//	go mod init repro && go get github.com/ClickHouse/clickhouse-go/v2 && go run .
package main

import (
	"context"
	"fmt"
	"log"
	"math/rand"
	"sync"
	"time"

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

const (
	addr       = "127.0.0.1:19000"
	table      = "benchmark_test"
	numWrites  = 300
	interval   = 10 * time.Millisecond // 100 writes/second
	probeEvery = 100                   // read-your-write after every Nth write
)

func connect(settings clickhouse.Settings) driver.Conn {
	conn, err := clickhouse.Open(&clickhouse.Options{
		Addr:     []string{addr},
		Settings: settings,
	})
	if err != nil {
		log.Fatal(err)
	}
	return conn
}

// e2eBench runs a writer at 100 rows/s; after every probeEvery-th write it
// polls SELECT max(id) until that write is visible and records the time from
// write start to first read that observes it.
func e2eBench(label string, settings clickhouse.Settings) {
	conn := connect(settings)
	defer conn.Close()
	ctx := context.Background()

	var (
		mu     sync.Mutex
		e2es   []time.Duration
		wg, pg sync.WaitGroup
	)

	probe := func(id uint64, writeStart time.Time) {
		defer pg.Done()
		for {
			var maxID uint64
			if err := conn.QueryRow(ctx, "SELECT max(id) FROM "+table).Scan(&maxID); err != nil {
				log.Fatal(err)
			}
			if maxID >= id { // our write is visible
				mu.Lock()
				e2es = append(e2es, time.Since(writeStart))
				mu.Unlock()
				return
			}
		}
	}

	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for i := 1; i <= numWrites; i++ {
		<-ticker.C
		wg.Add(1)
		go func(seq int) {
			defer wg.Done()
			id := uint64(time.Now().UnixNano())
			value := make([]float64, 200)
			for j := range value {
				value[j] = rand.Float64()
			}

			writeStart := time.Now()
			batch, err := conn.PrepareBatch(ctx, "INSERT INTO "+table)
			if err != nil {
				log.Fatal(err)
			}
			if err := batch.Append(id, value); err != nil {
				log.Fatal(err)
			}
			if err := batch.Send(); err != nil {
				log.Fatal(err)
			}
			if seq%probeEvery == 0 {
				pg.Add(1)
				go probe(id, writeStart)
			}
		}(i)
	}
	wg.Wait()
	pg.Wait()

	var total, worst time.Duration
	for _, d := range e2es {
		total += d
		if d > worst {
			worst = d
		}
	}
	fmt.Printf("%-45s e2e avg=%-11v worst=%-11v (n=%d)\n",
		label, (total / time.Duration(len(e2es))).Round(time.Millisecond/10), worst.Round(time.Millisecond), len(e2es))
}

func main() {
	e2eBench("A: sync inserts (the problem)", nil)
	e2eBench("B: async wait=0, default 200ms window",
		clickhouse.Settings{"async_insert": 1, "wait_for_async_insert": 0})
	e2eBench("C: async wait=0, 10ms window (the fix)",
		clickhouse.Settings{"async_insert": 1, "wait_for_async_insert": 0, "async_insert_busy_timeout_ms": 10})
}

$ go run cmd/e2e/main.go
A: sync inserts (the problem)                 e2e avg=2.7681s     worst=3.375s      (n=3)
B: async wait=0, default 200ms window         e2e avg=120.1ms     worst=168ms       (n=3)
C: async wait=0, 10ms window (the fix)        e2e avg=19ms        worst=22ms        (n=3)