write.go (customer)

write.go

// Single-row inserts: why writing 100 rows/second (one INSERT each) shows
// 1s+ latency when a lone insert takes ~5ms.
//
// Two stacked causes:
//
//  1. Server-side: concurrent single-row INSERTs to one table convoy -
//     ~200ms each at concurrency 10, nearly zero CPU. This affects every
//     client (clickhouse-client too), even in-memory Buffer tables.
//  2. Client-side amplification: 100 writes/s x 0.2s each needs 20
//     connections in flight, but the driver default is MaxOpenConns=10 -
//     capacity 50/s against 100/s arrivals, so the acquire queue grows
//     without bound (scenario A). Scenario B removes the queueing to expose
//     the server-side floor.
//
// The fix is fewer, fatter INSERT queries. Scenario D: async_insert makes the
// server do the batching; wait_for_async_insert=0 acks as soon as the row is
// in the server's buffer (same durability trade-off as a Buffer table).
// Fully-durable alternative: accumulate rows client-side and send ONE batch
// per flush interval.
//
// 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"
	"net"
	"sync"
	"sync/atomic"
	"time"

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

const (
	addr      = "127.0.0.1:9000"
	table     = "benchmark_test"
	numWrites = 300
	interval  = 10 * time.Millisecond // 100 writes/second
)

var dials atomic.Int64 // TCP connections opened by the driver

func connect(maxOpen int, settings clickhouse.Settings) driver.Conn {
	conn, err := clickhouse.Open(&clickhouse.Options{
		Addr:         []string{addr},
		MaxOpenConns: maxOpen, // 0 = driver default (10)
		MaxIdleConns: maxOpen,
		Settings:     settings,
		DialContext: func(ctx context.Context, addr string) (net.Conn, error) {
			dials.Add(1)
			var d net.Dialer
			return d.DialContext(ctx, "tcp", addr)
		},
	})
	if err != nil {
		log.Fatal(err)
	}
	return conn
}

// writeBench inserts numWrites single rows, one goroutine per tick, and
// reports the average/worst time from PrepareBatch to Send acknowledged.
func writeBench(label string, maxOpen int, settings clickhouse.Settings) {
	conn := connect(maxOpen, settings)
	defer conn.Close()
	before := dials.Load()

	var (
		mu           sync.Mutex
		total, worst time.Duration
		wg           sync.WaitGroup
	)
	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for i := 0; i < numWrites; i++ {
		<-ticker.C
		wg.Add(1)
		go func() {
			defer wg.Done()
			value := make([]float64, 200)
			for j := range value {
				value[j] = rand.Float64()
			}

			start := time.Now()
			batch, err := conn.PrepareBatch(context.Background(), "INSERT INTO "+table)
			if err != nil {
				log.Fatal(err)
			}
			if err := batch.Append(uint64(time.Now().UnixNano()), value); err != nil {
				log.Fatal(err)
			}
			if err := batch.Send(); err != nil {
				log.Fatal(err)
			}
			elapsed := time.Since(start)

			mu.Lock()
			total += elapsed
			if elapsed > worst {
				worst = elapsed
			}
			mu.Unlock()
		}()
	}
	wg.Wait()
	fmt.Printf("%-55s avg=%-11v worst=%-11v dials=%d\n",
		label, (total / numWrites).Round(time.Millisecond/10), worst.Round(time.Millisecond), dials.Load()-before)
}

func main() {
	writeBench("A: sync inserts, default pool (the problem)", 0, nil)
	writeBench("B: sync inserts, MaxOpenConns=32 (exposes server floor)", 32, nil)
	writeBench("C: async_insert, wait_for_async_insert=1 (no help)", 0,
		clickhouse.Settings{"async_insert": 1, "wait_for_async_insert": 1})
	writeBench("D: async_insert, wait_for_async_insert=0 (the fix)", 0,
		clickhouse.Settings{"async_insert": 1, "wait_for_async_insert": 0})
}

$ go run cmd/write/main.go
A: sync inserts, default pool (the problem)             avg=1.4506s     worst=2.734s      dials=11
B: sync inserts, MaxOpenConns=32 (exposes server floor) avg=115.9ms     worst=219ms       dials=22
C: async_insert, wait_for_async_insert=1 (no help)      avg=1.3847s     worst=2.624s      dials=11
D: async_insert, wait_for_async_insert=0 (the fix)      avg=3.1ms       worst=6ms         dials=1

Now how having increased the MaxOpenConns from default 10 -> 32 also removed the pool queuing bottleneck. The other optimizations is accumulate data either on server side or client side (async).