chconn - ClickHouse Native Protocol Driver for Go

package main

import (
	"context"
	"fmt"

	"github.com/vahid-sohrabloo/chconn/v3/chpool"
	"github.com/vahid-sohrabloo/chconn/v3/column"
)

func main() {
	conn, err := chpool.New("clickhouse://default:@localhost:9000/default")
	if err != nil {
		panic(err)
	}
	defer conn.Close()

	// Create table
	err = conn.Exec(context.Background(), `CREATE TABLE IF NOT EXISTS example (
		id    UInt64,
		name  String
	) Engine=Memory`)
	if err != nil {
		panic(err)
	}

	// Insert
	colID := column.New[uint64]()
	colName := column.NewString()
	colID.Append(1)
	colName.Append("Alice")
	colID.Append(2)
	colName.Append("Bob")

	err = conn.Insert(context.Background(), "INSERT INTO example (id, name) VALUES", colID, colName)
	if err != nil {
		panic(err)
	}

	// Select
	colIDRead := column.New[uint64]()
	colNameRead := column.NewString()
	selectStmt, err := conn.Select(context.Background(), "SELECT id, name FROM example", colIDRead, colNameRead)
	if err != nil {
		panic(err)
	}
	defer selectStmt.Close()

	for selectStmt.Next() {
		for i := 0; i < selectStmt.RowsInBlock(); i++ {
			fmt.Printf("id=%d name=%s\n", colIDRead.Row(i), colNameRead.Row(i))
		}
	}
	if err := selectStmt.Err(); err != nil {
		panic(err)
	}
}

col1 := column.New[uint64]()
col2 := column.NewString()
col3 := column.New[uint64]().Nullable()

// Pre-allocate write buffers for large inserts
col1.SetWriteBufferSize(1_000_000)
col2.SetWriteBufferSize(1_000_000)
col3.SetWriteBufferSize(1_000_000)

for i := range 1_000_000 {
    col1.Append(uint64(i))
    col2.Append(fmt.Sprintf("row_%d", i))
    if i%2 == 0 {
        col3.Append(uint64(i))
    } else {
        col3.AppendNil()
    }
}

err := conn.Insert(ctx, "INSERT INTO table (id, name, value) VALUES", col1, col2, col3)

colID := column.New[uint64]()
colName := column.NewString()

selectStmt, err := conn.Select(ctx, "SELECT id, name FROM table", colID, colName)
if err != nil {
    return err
}
defer selectStmt.Close()

var ids []uint64
var names []string
for selectStmt.Next() {
    ids = colID.Read(ids)
    names = colName.Read(names)
}
if err := selectStmt.Err(); err != nil {
    return err
}

rows, err := conn.Query(ctx, "SELECT id, name FROM users WHERE age > {age: UInt32}",
    chconn.IntParameter("age", 18),
)
if err != nil {
    return err
}
defer rows.Close()

for rows.Next() {
    var id uint64
    var name string
    if err := rows.Scan(&id, &name); err != nil {
        return err
    }
    fmt.Printf("%d: %s\n", id, name)
}
return rows.Err()

type User struct {
    ID    uint64
    Name  string
    Email string
}

// Collect all rows into a slice
users, err := chconn.QueryAll[User](ctx, conn, "SELECT id, name, email FROM users")

// Get exactly one row
user, err := chconn.QueryExactlyOne[User](ctx, conn,
    "SELECT id, name, email FROM users WHERE id = {id: UInt64}",
    chconn.UintParameter("id", 42),
)

// Iterator pattern
for user, err := range chconn.QueryIter[User](ctx, conn, "SELECT id, name, email FROM users") {
    if err != nil {
        return err
    }
    fmt.Println(user.Name)
}

stmt, err := conn.InsertStream(ctx, "INSERT INTO table (id, name) VALUES")
if err != nil {
    return err
}

// Row-by-row append
for i := range 1000 {
    if err := stmt.Append(uint64(i), fmt.Sprintf("name_%d", i)); err != nil {
        return err
    }
}

// Or write column batches
col1 := column.New[uint64]()
col2 := column.NewString()
col1.Append(1001)
col2.Append("batch_row")
if err := stmt.Write(ctx, col1, col2); err != nil {
    return err
}

return stmt.Flush(ctx)

rows, err := conn.Query(ctx,
    "SELECT * FROM events WHERE ts > {start: DateTime} AND level = {level: String}",
    chconn.StringParameter("start", "2024-01-01 00:00:00"),
    chconn.StringParameter("level", "error"),
)

pool, err := chpool.New("clickhouse://user:password@localhost:9000/mydb")
// or with full config:
config, _ := chpool.ParseConfig("clickhouse://user:password@localhost:9000/mydb")
config.MaxConns = 10
config.MinConns = 2
config.MaxConnLifetime = time.Hour
config.MaxConnIdleTime = 30 * time.Minute
config.HealthCheckPeriod = time.Minute

pool, err := chpool.NewWithConfig(ctx, config)

// LZ4 (recommended)
config.Compress = chconn.CompressLZ4

// ZSTD
config.Compress = chconn.CompressZSTD

clickhouse://user:pass@host:9440/db?sslmode=verify-full&sslrootcert=/path/to/ca.pem

sb := sqlbuilder.NewSelectBuilder()
query, _ := sb.Select("id", "name", "email").
    From("users").
    Where("age > 18").
    PreWhere("active = 1").
    OrderBy("name").
    Limit(100).
    Build()

exporter := format.NewJSON(1000, func(data []byte, cols []column.ColumnCore) {
    os.Stdout.Write(data)
})
selectStmt, _ := conn.Select(ctx, "SELECT * FROM table", cols...)
exporter.Read(selectStmt)        // JSON objects
exporter.ReadCompact(selectStmt) // JSON arrays (compact)
exporter.ReadEachRow(selectStmt) // One JSON object per row

opts := &chconn.QueryOptions{
    QueryID: "my-query-123",
    OnProgress: func(p *chconn.Progress) {
        fmt.Printf("Read %d rows, %d bytes\n", p.ReadRows, p.ReadBytes)
    },
    OnProfile: func(p *chconn.Profile) {
        fmt.Printf("Result: %d rows in %d blocks\n", p.Rows, p.Blocks)
    },
}
selectStmt, err := conn.SelectWithOption(ctx, "SELECT ...", opts, cols...)

var chErr *chconn.ChError
if errors.As(err, &chErr) {
    fmt.Printf("ClickHouse error %d: %s\n", chErr.Code, chErr.Message)
    fmt.Println(chErr.StackTrace)
}

// Sentinel errors
if errors.Is(err, chconn.ErrNoRows) { ... }
if errors.Is(err, chconn.ErrTooManyRows) { ... }

ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()

err := conn.Insert(ctx, "INSERT INTO ...", cols...)