ch-go

To write data, prepare a proto.Input containing columns that implement the necessary proto interfaces. You can then execute an INSERT query using c.Do.

var (
	body      proto.ColStr
	name      proto.ColStr
	sevText   proto.ColEnum
	sevNumber proto.ColUInt8

    serviceName = proto.NewLowCardinality(new(proto.ColStr))
    ts          = new(proto.ColDateTime64).WithPrecision(proto.PrecisionNano)
    arr         = new(proto.ColStr).Array()
    now         = time.Date(2010, 1, 1, 10, 22, 33, 345678, time.UTC)
)

// Append data to columns
for i := 0; i < 10; i++ {
	body.AppendBytes([]byte("Hello"))
	ts.Append(now)
	name.Append("name")
	sevText.Append("INFO")
	sevNumber.Append(10)
	arr.Append([]string{"foo", "bar", "baz"})
    serviceName.Append("service")
}

input := proto.Input{
	{Name: "ts", Data: ts},
	{Name: "severity_text", Data: &sevText},
	{Name: "severity_number", Data: sevNumber},
	{Name: "body", Data: body},
	{Name: "name", Data: name},
	{Name: "arr", Data: arr},
}

// Execute insertion
if err := conn.Do(ctx, ch.Query{
	Body: "INSERT INTO test_table_insert VALUES",
	Input: input,
}); err != nil {
	panic(err)
}

To stream large amounts of data in multiple blocks, use the OnInput field in ch.Query. The OnInput function is called before each block is encoded and sent.

Crucial: You must call input.Reset() inside OnInput to clear the columns before appending new data; otherwise, data will be duplicated in subsequent blocks. To stop streaming, return io.EOF from OnInput.

var blocks int
if err := conn.Do(ctx, ch.Query{
	Body:  input.Into("test_table_insert"),
	Input: input,
	OnInput: func(ctx context.Context) error {
		input.Reset() // Important: reset columns to avoid duplication

		if blocks >= 10 {
			return io.EOF
		}

		// Append new values for this block
		for i := 0; i < 10; i++ {
			body.AppendBytes([]byte("Hello"))
			ts.Append(now)
			name.Append("name")
			sevText.Append("DEBUG")
			sevNumber.Append(10)
			arr.Append([]string{"foo", "bar", "baz"})
            serviceName.Append("service")
		}

		blocks++
		return nil
	},
}); err != nil {
	panic(err)
}

The ch-go package is a low-level TCP client designed for high-performance data streaming. It is not goroutine-safe by default and does not provide connection pooling or automatic reconnects. For pooling, use the chpool package.

To execute a query and stream results, use ch.Dial to establish a connection and c.Do to execute a ch.Query.

package main

import (
  "context"
  "fmt"

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

func main() {
  ctx := context.Background()
  c, err := ch.Dial(ctx, ch.Options{Address: "localhost:9000"})
  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",
    Result: proto.Results{
      {Name: "number", Data: &data},
    },
    // OnResult will be called on next received data block.
    OnResult: func(ctx context.Context, b proto.Block) error {
      numbers += len(data)
      return nil
    },
  }); err != nil {
    panic(err)
  }
  fmt.Println("numbers:", numbers)
}

The Variant struct defines the characteristics of a ClickHouse column type used during code generation. Key properties include:

• Kind: The category of the type (e.g., KindInt, KindFloat, KindIP, KindDateTime, KindDate, KindTime32, KindTime64, KindEnum, KindDecimal, KindFixedStr).
• Signed: Whether the type is signed or unsigned.
• Bits: The bit width of the type.
• GenerateUnsafe: A flag indicating if an unsafe implementation should be generated.

Commonly used methods on Variant to determine type behavior:

• IsFloat(), IsInt(), IsIP(): Check the fundamental kind.
• Big(): Returns true if Bits > 64.
• FixedStr(): Returns true if the kind is KindFixedStr.
• Time(): Returns true if the kind is a time-related type (KindDate, KindDateTime, KindTime32, KindTime64).
• Byte(): Returns true if the type is a single-byte integer (Bits == 8 and !Signed).

When performing INSERT operations, use the OnInput callback in a Query to stream data blocks to the server. This prevents high memory consumption by allowing the client to flush data periodically.

Workflow:

1. Set Query.Input with the initial column definitions.
2. Implement OnInput(ctx context.Context) error.
3. Inside OnInput, provide the next block of data via the Input field.
4. Return io.EOF when all data has been sent. The client will then send a blank block to signal the end of the input stream.

Note: Atomicity is only guaranteed within a single block.

q := ch.Query{
    Body: "INSERT INTO my_table (col1, col2)",
    Input: proto.Input{
        // Initial column setup
    },
    OnInput: func(ctx context.Context) error {
        // 1. Load next block of data into q.Input
        // 2. If no more data, return io.EOF
        // 3. If error, return error
        return nil
    },
}