ch

README

ch

TCP ClickHouse client in Go. Designed for very fast data block streaming with low network, cpu and memory overhead.

Work in progress, please leave feedback on package API or features. Also, see benchmarks.

ClickHouse is an open-source, high performance columnar OLAP database management system for real-time analytics using SQL.

go get github.com/go-faster/ch

Example

package main

import (
  "context"
  "fmt"

  "github.com/go-faster/ch"
  "github.com/go-faster/ch/proto"
)

func main() {
  ctx := context.Background()
  c, err := ch.Dial(ctx, "localhost:9000", ch.Options{})
  if err != nil {
    panic(err)
  }
  var (
    numbers int
    data    proto.ColUInt64
  )
  if err := c.Do(ctx, ch.Query{
    Body: "SELECT number FROM system.numbers LIMIT 500000000",
    // OnResult will be called on next received data block.
    OnResult: func(ctx context.Context, b proto.Block) error {
      numbers += len(data)
      return nil
    },
    Result: proto.Results{
      {Name: "number", Data: &data},
    },
  }); err != nil {
    panic(err)
  }
  fmt.Println("numbers:", numbers)
}

393ms 0.5B rows  4GB  10GB/s 1 job
874ms 2.0B rows 16GB  18GB/s 4 jobs

Results

To stream query results, set Result and OnResult fields of Query. The OnResult will be called after Result is filled with received data block.

The OnResult is optional, but query will fail if more than single block is received, so it is ok to solely set the Result if only one row is expected.

Automatic result inference

var result proto.Results
q := ch.Query{
  Body:   "SELECT * FROM table",
  Result: result.Auto(),
}

Single result with column name inference

var res proto.ColBool
q := ch.Query{
  Body:   "SELECT v FROM test_table",
  Result: proto.ResultColumn{Data: &res},
}

Features

• OpenTelemetry support
• No reflection or interface{}
• Generics (go1.18) for ArrayOf[T], LowCardinaliyOf[T], EnumOf[T]
• Column-oriented design that operates with blocks
  • Dramatically more efficient
  • Up to 100x faster than row-first design around sql
  • Up to 700x faster than HTTP API
  • Low memory overhead (data blocks are slices, i.e. continuous memory)
  • Highly efficient input and output block streaming
  • As close to ClickHouse as possible
• Structured query execution telemetry streaming
  • Query progress
  • Profiles
  • Logs
  • Profile events
• LZ4, ZSTD or None (just checksums) compression
• External data support
• Rigorously tested
  • ARM64, Windows, Mac, Linux (also x86)
  • Unit tests for encoding and decoding
    • ClickHouse Server in Go for faster tests
    • Golden files for all packets, columns
    • Both server and client structures
    • Ensuring that partial read leads to failure
  • End-to-end tests
    • 21.8.11.4-lts
    • 21.9.6.24-stable
    • 21.10.4.26-stable
    • 21.11.4.14-stable
    • 21.11.7.9-stable
    • 21.12.2.17-stable
  • Fuzzing

Supported types

• UInt8, UInt16, UInt32, UInt64, UInt128, UInt256
• Int8, Int16, Int32, Int64, Int128, Int256
• Date, Date32, DateTime, DateTime64
• Decimal32, Decimal64, Decimal128, Decimal256
• IPv4, IPv6
• String, FixedString(N)
• UUID
• Array(T)
• Enum8, Enum16
• LowCardinality(T)
• Map(K, V)
• Bool
• Tuple(T1, T2, ..., Tn)
• Nullable(T)

TODO

• Connection pools
• TLS
• API UX Improvements (with 1.18 generics?)
  • Enum
  • LowCardinality
  • Array(T)
  • FixedString(N)
  • Map(K, V)
  • Decimal(P, S)
  • Nullable(T)
  • Tuple?
• Code generation from DDL
  • Parser
  • Code generator for SELECT/INSERT
  • Query builder
• DSL for DDL
• database/sql integration
• Reading and writing Native format dumps

Reference

• clickhouse-cpp
• clickhouse-go
• python driver

License

Apache License 2.0

Documentation

Overview

Package ch implements ClickHouse client.

Index

• func CompressionStrings() []string
• func IsErr(err error, codes ...proto.Error) bool
• func IsException(err error) bool
• func ProfileEventTypeStrings() []string
• type Client
  • func Connect(ctx context.Context, conn net.Conn, opt Options) (*Client, error)
  • func Dial(ctx context.Context, addr string, opt Options) (*Client, error)
  • func (c *Client) Close() error
  • func (c *Client) Do(ctx context.Context, q Query) error
  • func (c *Client) Location() *time.Location
  • func (c *Client) Ping(ctx context.Context) error
• type Compression
  • func CompressionString(s string) (Compression, error)
  • func CompressionValues() []Compression
  • func (i Compression) IsACompression() bool
  • func (i Compression) String() string
• type CorruptedDataErr
  • func (c *CorruptedDataErr) Error() string
• type Exception
  • func AsException(err error) (*Exception, bool)
  • func (e *Exception) Error() string
  • func (e *Exception) IsCode(codes ...proto.Error) bool
• type Log
• type Options
• type ProfileEvent
• type ProfileEventType
  • func ProfileEventTypeString(s string) (ProfileEventType, error)
  • func ProfileEventTypeValues() []ProfileEventType
  • func (i ProfileEventType) IsAProfileEventType() bool
  • func (i ProfileEventType) String() string
• type Query
• type Server
  • func NewServer(opt ServerOptions) *Server
  • func (s *Server) Serve(ln net.Listener) error
• type ServerConn
  • func (c *ServerConn) Handle() error
• type ServerOptions
• type Setting
  • func SettingInt(k string, v int) Setting

Examples

• Query (MultipleInputColumns)

Functions

func CompressionStrings

func CompressionStrings() []string

CompressionStrings returns a slice of all String values of the enum

func IsErr

func IsErr(err error, codes ...proto.Error) bool

IsErr reports whether err is error with provided exception codes.

func IsException

func IsException(err error) bool

IsException reports whether err is Exception.

func ProfileEventTypeStrings

func ProfileEventTypeStrings() []string

ProfileEventTypeStrings returns a slice of all String values of the enum

Types

type Client

type Client struct {
	// contains filtered or unexported fields
}

Client implements ClickHouse binary protocol client on top of single TCP connection.

func Connect

func Connect(ctx context.Context, conn net.Conn, opt Options) (*Client, error)

Connect performs handshake with ClickHouse server and initializes application level connection.

func Dial

func Dial(ctx context.Context, addr string, opt Options) (*Client, error)

Dial dials requested address and establishes TCP connection to ClickHouse server, performing handshake.

func (*Client) Close

func (c *Client) Close() error

Close closes underlying connection and frees all resources, rendering Client to unusable state.

func (*Client) Do

func (c *Client) Do(ctx context.Context, q Query) error

Do performs Query on ClickHouse server.

func (*Client) Location

func (c *Client) Location() *time.Location

Location returns current server timezone.

func (*Client) Ping

func (c *Client) Ping(ctx context.Context) error

type Compression

type Compression byte

Compression setting.

Trade bandwidth for CPU.

const (
	// CompressionDisabled disables compression. Lowest CPU overhead.
	CompressionDisabled Compression = iota
	// CompressionLZ4 enables LZ4 compression for data. Medium CPU overhead.
	CompressionLZ4
	// CompressionZSTD enables ZStandard compression. High CPU overhead.
	CompressionZSTD
	// CompressionNone uses no compression but data has checksums.
	CompressionNone
)

func CompressionString

func CompressionString(s string) (Compression, error)

CompressionString retrieves an enum value from the enum constants string name. Throws an error if the param is not part of the enum.

func CompressionValues

func CompressionValues() []Compression

CompressionValues returns all values of the enum

func (Compression) IsACompression

func (i Compression) IsACompression() bool

IsACompression returns "true" if the value is listed in the enum definition. "false" otherwise

func (Compression) String

func (i Compression) String() string

type CorruptedDataErr

type CorruptedDataErr struct {
	Actual    city.U128
	Reference city.U128
	RawSize   int
	DataSize  int
}

CorruptedDataErr means that provided hash mismatch with calculated.

func (*CorruptedDataErr) Error

func (c *CorruptedDataErr) Error() string

type Exception

type Exception struct {
	Code    proto.Error
	Name    string
	Message string
	Stack   string
	Next    []Exception // non-nil only for top exception
}

Exception is server-side error.

func AsException

func AsException(err error) (*Exception, bool)

AsException finds first *Exception in err chain.

func (*Exception) Error

func (e *Exception) Error() string

func (*Exception) IsCode

func (e *Exception) IsCode(codes ...proto.Error) bool

type Log

type Log struct {
	Time     time.Time
	Host     string
	QueryID  string
	ThreadID uint64
	Priority int8
	Source   string
	Text     string
}

Log from server.

type Options

type Options struct {
	Logger      *zap.Logger
	Database    string
	User        string
	Password    string
	Settings    []Setting
	Compression Compression
}

Options for Client.

type ProfileEvent

type ProfileEvent struct {
	ThreadID uint64
	Host     string
	Time     time.Time
	Type     ProfileEventType
	Name     string
	Value    int64
}

ProfileEvent is detailed profiling event from Server.

type ProfileEventType

type ProfileEventType byte

const (
	ProfileIncrement ProfileEventType = 1
	ProfileGauge     ProfileEventType = 2
)

func ProfileEventTypeString

func ProfileEventTypeString(s string) (ProfileEventType, error)

ProfileEventTypeString retrieves an enum value from the enum constants string name. Throws an error if the param is not part of the enum.

func ProfileEventTypeValues

func ProfileEventTypeValues() []ProfileEventType

ProfileEventTypeValues returns all values of the enum

func (ProfileEventType) IsAProfileEventType

func (i ProfileEventType) IsAProfileEventType() bool

IsAProfileEventType returns "true" if the value is listed in the enum definition. "false" otherwise

func (ProfileEventType) String

func (i ProfileEventType) String() string

type Query

type Query struct {
	// Body of query, like "SELECT 1".
	Body string
	// QueryID is ID of query, defaults to new UUIDv4.
	QueryID string
	// QuotaKey of query, optional.
	QuotaKey string

	// Input columns for INSERT operations.
	Input proto.Input
	// OnInput is called to allow ingesting more data to Input.
	//
	// The io.EOF reports that no more input should be ingested.
	//
	// Optional, single block is ingested from Input if not provided,
	// but query will fail if Input is set but has zero rows.
	OnInput func(ctx context.Context) error

	// Result columns for SELECT operations.
	Result proto.Result
	// OnResult is called when Result is filled with result block.
	//
	// Optional, but query will fail of more than one block is received
	// and no OnResult is provided.
	OnResult func(ctx context.Context, block proto.Block) error

	// OnProgress is optional progress handler. The progress value contain
	// difference, so progress should be accumulated if needed.
	OnProgress func(ctx context.Context, p proto.Progress) error
	// OnProfile is optional handler for profiling data.
	OnProfile func(ctx context.Context, p proto.Profile) error
	// OnProfileEvent is optional handler for profiling event stream data.
	OnProfileEvent func(ctx context.Context, e ProfileEvent) error
	// OnLog is optional handler for server log entry.
	OnLog func(ctx context.Context, l Log) error

	// Settings are optional query-scoped settings. Can override client settings.
	Settings []Setting

	// ExternalData is optional data for server to load.
	//
	// https://clickhouse.com/docs/en/engines/table-engines/special/external-data/
	ExternalData []proto.InputColumn
	// ExternalTable name. Defaults to _data.
	ExternalTable string
}

Query to ClickHouse.

Example (MultipleInputColumns)

var (
	body      proto.ColStr
	timestamp proto.ColDateTime64
	name      proto.ColStr
	sevText   proto.ColEnum8Auto
	sevNumber proto.ColUInt8
	arrValues proto.ColStr

	arr = proto.ColArr{Data: &arrValues} // Array(String)
	now = time.Date(2010, 1, 1, 10, 22, 33, 345678, time.UTC)
)
// Append 10 rows.
for i := 0; i < 10; i++ {
	body.AppendBytes([]byte("Hello"))
	timestamp = append(timestamp, proto.ToDateTime64(now, proto.PrecisionNano))
	name.Append("name")
	sevText.Values = append(sevText.Values, "INFO")
	sevNumber = append(sevNumber, 10)
	arrValues.ArrAppend(&arr, []string{"foo", "bar", "baz"})
}
input := proto.Input{
	{Name: "timestamp", Data: timestamp.Wrap(proto.PrecisionNano)},
	{Name: "severity_text", Data: &sevText},
	{Name: "severity_number", Data: sevNumber},
	{Name: "body", Data: body},
	{Name: "name", Data: name},
	{Name: "arr", Data: arr},
}
fmt.Println(input.Into("logs"))

Output:

INSERT INTO logs (timestamp, severity_text, severity_number, body, name, arr) VALUES

type Server

type Server struct {
	// contains filtered or unexported fields
}

Server is basic ClickHouse server.

func NewServer

func NewServer(opt ServerOptions) *Server

NewServer returns new ClickHouse Server.

func (*Server) Serve

func (s *Server) Serve(ln net.Listener) error

Serve connections on net.Listener.

type ServerConn

type ServerConn struct {
	// contains filtered or unexported fields
}

ServerConn wraps Server connection.

func (*ServerConn) Handle

func (c *ServerConn) Handle() error

Handle connection.

type ServerOptions

type ServerOptions struct {
	Logger   *zap.Logger
	Timezone *time.Location
	OnError  func(err error)
}

ServerOptions wraps possible Server configuration.

type Setting

type Setting struct {
	Key, Value string
	Important  bool
}

Setting to send to server.

func SettingInt

func SettingInt(k string, v int) Setting

SettingInt returns Setting with integer value v.