RTT-issue-reproduction

latency-issue.go

// Minimal native-protocol client to observe the end-of-query packet-train
// stall (the "2xRTT" issue).
//
// It runs N identical SELECTs on ONE reused connection and, for each query,
// splits the client-observed latency into:
//
//	first_row : query sent -> first result row decoded  (the query itself)
//	tail      : first row  -> EndOfStream               (the epilogue train)
//
// A healthy query has tail ~= 0. When the epilogue train (profile info,
// progress, profile events, empty block, end-of-stream - one small TCP packet
// each) exceeds the server's congestion window, tail == exactly 1xRTT.
//

package main

import (
	"context"
	"flag"
	"fmt"
	"log"
	"time"

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

func main() {
	addr := flag.String("addr", "10.99.0.2:9000", "server native-protocol address")
	n := flag.Int("n", 5, "number of queries")
	sleep := flag.Duration("sleep", 2*time.Second, "idle gap between queries (0 = back-to-back)")
	rows := flag.Int("rows", 0, "if >0, fetch this many ~108-byte rows instead of max(id) (tests window scaling)")
	noProfileEvents := flag.Bool("no-profile-events", false, "shorten the epilogue train (send_profile_events=0)")
	flag.Parse()

	opts := &clickhouse.Options{Addr: []string{*addr}}
	if *noProfileEvents {
		opts.Settings = clickhouse.Settings{"send_profile_events": 0}
	}
	conn, err := clickhouse.Open(opts)
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()

	ctx := context.Background()
	if err := conn.Ping(ctx); err != nil {
		log.Fatal(err)
	}

	query := "SELECT max(id) FROM benchmark_test"
	if *rows > 0 {
		query = fmt.Sprintf("SELECT number, randomString(100) FROM numbers(%d)", *rows)
	}
	fmt.Printf("addr=%s  sleep=%v  profile_events=%v\nquery: %s\n\n", *addr, *sleep, !*noProfileEvents, query)

	for i := 1; i <= *n; i++ {
		if *sleep > 0 && i > 1 {
			time.Sleep(*sleep)
		}

		start := time.Now()
		r, err := conn.Query(ctx, query)
		if err != nil {
			log.Fatal(err)
		}
		var firstRow time.Duration
		var count int
		for r.Next() {
			if firstRow == 0 {
				firstRow = time.Since(start)
			}
			count++
		}
		if err := r.Close(); err != nil { // Close drains to EndOfStream
			log.Fatal(err)
		}
		total := time.Since(start)

		fmt.Printf("query %2d: first_row=%7.2fms  eos=%7.2fms  tail=%7.2fms  (rows=%d)\n",
			i, ms(firstRow), ms(total), ms(total-firstRow), count)
	}
}

func ms(d time.Duration) float64 { return float64(d.Microseconds()) / 1000.0 }