Batch and copyfrom

On ClickHouse:

• batch calls PrepareBatch, batch.Append for each row and batch.Send: one INSERT with the rows sent column by column, with no bind variables and no chunking
• The {name:Type} placeholders in VALUES only name and type the columns; the values come from Append
• copyfrom generates the same PrepareBatch code as batch
• Each INSERT writes one part that ClickHouse merges later, so batching keeps the part count down; the example counts parts with the _part virtual column
• An empty slice returns at once, without preparing an INSERT or contacting the server
• Q017 does not arise: no parameter is kind: slice, and an Array value is one column value

Inserting rows one call at a time costs a round trip for every row, which adds up quickly when you have thousands of them. A batch query is written as the INSERT for a single row, but its generated method takes a slice, one Params value per row, and writes them all in as few round trips as the driver allows. copyfrom asks for the driver's fastest bulk-load path in the same way.

How the rows travel depends on the driver, and Seagull generates the right code for each:

• With database/sql, which has no batch protocol, the rows become multi-row INSERT statements, split into chunks so that no statement binds more variables than the engine accepts.
• With the native ClickHouse driver, used on this page, both batch and copyfrom become one prepared batch: a single INSERT whose rows are sent column by column.

The example records tide readings: one a minute from two gauges, thousands of rows in a single call.

The query

The INSERT lists one row's values. The batch command is what makes the method take a slice.

db/queries/readings.sql

-- sgl.query(RecordReadings, batch)
INSERT INTO readings (station, minute, height_cm, note)
VALUES ({station:String}, {minute:UInt16}, {height_cm:UInt16}, {note:Nullable(String)});

The {name:Type} placeholders in the VALUES list only name and type the columns. Nothing is bound to them: the driver takes the values from each row the method appends.

What sgl generates

The method prepares the INSERT with PrepareBatch, calls batch.Append once per row and sends everything with batch.Send. There are no bind variables, so there is no limit to chunk under, and a copyfrom query generates the same code.

internal/store

type RecordReadingsParams struct {
	Station  string
	Minute   uint16
	HeightCm uint16
	Note     *string
}

func (queries *Queries) RecordReadings(ctx context.Context, params []RecordReadingsParams) error {
	if len(params) == 0 {
		return nil
	}
	batch, err := queries.db.PrepareBatch(ctx, recordReadingsSQL)
	if err != nil {
		return err
	}
	defer batch.Close()
	for _, item := range params {
		if err := batch.Append(item.Station, item.Minute, item.HeightCm, item.Note); err != nil {
			return err
		}
	}
	return batch.Send()
}

An empty slice still prepares the INSERT, then sends no rows, so nothing is written.

Sending the rows together matters more on ClickHouse than elsewhere. Every INSERT writes a new part, a sorted chunk of the table on disk, which ClickHouse merges with others in the background. Writing rows one at a time leaves many small parts for it to merge; one batch writes one part.

Using it

The program records an empty batch, then a day of readings from two gauges, and after each it counts the table's parts with CountParts, which reads ClickHouse's _part virtual column.

db/queries/readings.sql

-- sgl.query(CountParts, one)
SELECT uniqExact(_part) AS parts FROM readings;

example.go

package main

import (
	"context"
	"fmt"
	"io"

	"seagull.sh/seagull/examples/clickhouse/022_bulk_writes_native/internal/store"
)

const (
	// minutesPerDay is how many readings each tide gauge takes in a day.
	minutesPerDay = 24 * 60
)

// run records a day of readings from two tide gauges in one batch call, shows how many
// parts that wrote, and summarises what landed in the table.
//
// A batch query takes a slice of Params, one per row. On the native driver the generated
// method calls PrepareBatch, appends every row with batch.Append and sends them with
// batch.Send: one INSERT, its rows sent column by column in blocks, with no bind
// variables and so no chunking. The {name:Type} placeholders in the VALUES list only name
// and type the columns; the driver takes the values from Append. copyfrom generates the
// same code.
//
// Each INSERT writes a new part, which ClickHouse merges in the background, so writing
// rows one at a time leaves many small parts to merge. CountParts reads the _part virtual
// column to show that the whole day landed as one part.
//
// Takes connection (store.DBTX) which is a native connection to a migrated database.
// Takes output (io.Writer) which receives the part counts and the summary.
//
// Returns error when a query fails.
func run(ctx context.Context, connection store.DBTX, output io.Writer) error {
	queries := store.New(connection)

	// An empty batch returns at once, without preparing an INSERT, so it writes no part.
	if err := queries.RecordReadings(ctx, nil); err != nil {
		return fmt.Errorf("recording no readings: %w", err)
	}
	if err := printParts(ctx, queries, 0, output); err != nil {
		return err
	}

	readings := tideReadings("harbour", "pier")
	if err := queries.RecordReadings(ctx, readings); err != nil {
		return fmt.Errorf("recording readings: %w", err)
	}
	if err := printParts(ctx, queries, len(readings), output); err != nil {
		return err
	}

	summary, err := queries.SummariseReadings(ctx)
	if err != nil {
		return fmt.Errorf("summarising readings: %w", err)
	}
	for _, row := range summary {
		fmt.Fprintf(output, "%-8s %d readings, %d with a note\n", row.Station, row.Readings, row.Noted)
	}
	return nil
}

// printParts prints how many parts hold the table's rows.
//
// Takes queries (*store.Queries) which runs the query.
// Takes sent (int) which is how many readings were sent.
// Takes output (io.Writer) which receives the line.
//
// Returns error when the query fails.
func printParts(ctx context.Context, queries *store.Queries, sent int, output io.Writer) error {
	parts, err := queries.CountParts(ctx)
	if err != nil {
		return fmt.Errorf("counting parts: %w", err)
	}
	fmt.Fprintf(output, "%d readings sent, parts holding rows: %d\n", sent, parts.Parts)
	return nil
}

// tideReadings makes one reading a minute for a day at each station. The height rises and
// falls in a simple saw-tooth, and the reading on the hour carries a note.
//
// Takes stations (...string) which name the gauges.
//
// Returns []store.RecordReadingsParams which holds every reading, ready for the batch.
func tideReadings(stations ...string) []store.RecordReadingsParams {
	readings := make([]store.RecordReadingsParams, 0, len(stations)*minutesPerDay)
	for _, station := range stations {
		for minute := range uint16(minutesPerDay) {
			var note *string
			if minute%60 == 0 {
				note = new(fmt.Sprintf("hour %d check", minute/60))
			}
			readings = append(readings, store.RecordReadingsParams{
				Station:  station,
				Minute:   minute,
				HeightCm: 200 + minute%360,
				Note:     note,
			})
		}
	}
	return readings
}

What it prints

The empty batch wrote no part, and all 2880 readings landed in a single part.

0 readings sent, parts holding rows: 0
2880 readings sent, parts holding rows: 1
harbour  1440 readings, 24 with a note
pier     1440 readings, 24 with a note

Across engines

Only these Go types of RecordReadings differ between engines; the rest are the same everywhere.

	ClickHouse	SQLite	PostgreSQL	MySQL	DuckDB
minute parameter	uint16	int64	int64	int32	int32
height_cm parameter	uint16	int64	int64	int32	int32

CockroachDB and TimescaleDB run the PostgreSQL version unchanged.

MariaDB runs the MySQL version unchanged.

Compare the whole example side by side with SQLite, PostgreSQL, MySQL and DuckDB.

Run it locally

From the directory you cloned the Seagull repository into:

cd seagull/examples/clickhouse && go run ./022_bulk_writes_native

Its test runs it too, and checks what it prints against output.txt:

cd seagull/examples/clickhouse && SGL_EXAMPLE_DSN=clickhouse://localhost:9000/default go test ./022_bulk_writes_native

Needs a running ClickHouse server: it connects to clickhouse://localhost:9000/default, or pass -dsn with the address of an empty database.

The files

Every file of the example, as it is in the Seagull repository.

sgl.yaml

# Copyright 2026 PolitePixels Limited
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

# This project stands against fascism, authoritarianism, and all forms of
# oppression. We built this to empower people, not to enable those who would
# strip others of their rights and dignity.

version: 1

databases:
  - name: tides
    dialect: clickhouse
    migrations: db/migrations
    queries: db/queries
    go:
      package: store
      out: internal/store
      driver: clickhouse

db/migrations/0001_readings.up.sql

CREATE TABLE readings (
    station   String,
    minute    UInt16,
    height_cm UInt16,
    note      Nullable(String)
) ENGINE = MergeTree() ORDER BY (station, minute);

db/migrations/0001_readings.down.sql

DROP TABLE readings;

db/queries/readings.sql

-- sgl.query(RecordReadings, batch)
INSERT INTO readings (station, minute, height_cm, note)
VALUES ({station:String}, {minute:UInt16}, {height_cm:UInt16}, {note:Nullable(String)});

-- sgl.query(SummariseReadings, many)
SELECT station, count() AS readings, count(note) AS noted
FROM readings
GROUP BY station
ORDER BY station;

-- sgl.query(CountParts, one)
SELECT uniqExact(_part) AS parts FROM readings;

example.go

package main

import (
	"context"
	"fmt"
	"io"

	"seagull.sh/seagull/examples/clickhouse/022_bulk_writes_native/internal/store"
)

const (
	// minutesPerDay is how many readings each tide gauge takes in a day.
	minutesPerDay = 24 * 60
)

// run records a day of readings from two tide gauges in one batch call, shows how many
// parts that wrote, and summarises what landed in the table.
//
// A batch query takes a slice of Params, one per row. On the native driver the generated
// method calls PrepareBatch, appends every row with batch.Append and sends them with
// batch.Send: one INSERT, its rows sent column by column in blocks, with no bind
// variables and so no chunking. The {name:Type} placeholders in the VALUES list only name
// and type the columns; the driver takes the values from Append. copyfrom generates the
// same code.
//
// Each INSERT writes a new part, which ClickHouse merges in the background, so writing
// rows one at a time leaves many small parts to merge. CountParts reads the _part virtual
// column to show that the whole day landed as one part.
//
// Takes connection (store.DBTX) which is a native connection to a migrated database.
// Takes output (io.Writer) which receives the part counts and the summary.
//
// Returns error when a query fails.
func run(ctx context.Context, connection store.DBTX, output io.Writer) error {
	queries := store.New(connection)

	// An empty batch returns at once, without preparing an INSERT, so it writes no part.
	if err := queries.RecordReadings(ctx, nil); err != nil {
		return fmt.Errorf("recording no readings: %w", err)
	}
	if err := printParts(ctx, queries, 0, output); err != nil {
		return err
	}

	readings := tideReadings("harbour", "pier")
	if err := queries.RecordReadings(ctx, readings); err != nil {
		return fmt.Errorf("recording readings: %w", err)
	}
	if err := printParts(ctx, queries, len(readings), output); err != nil {
		return err
	}

	summary, err := queries.SummariseReadings(ctx)
	if err != nil {
		return fmt.Errorf("summarising readings: %w", err)
	}
	for _, row := range summary {
		fmt.Fprintf(output, "%-8s %d readings, %d with a note\n", row.Station, row.Readings, row.Noted)
	}
	return nil
}

// printParts prints how many parts hold the table's rows.
//
// Takes queries (*store.Queries) which runs the query.
// Takes sent (int) which is how many readings were sent.
// Takes output (io.Writer) which receives the line.
//
// Returns error when the query fails.
func printParts(ctx context.Context, queries *store.Queries, sent int, output io.Writer) error {
	parts, err := queries.CountParts(ctx)
	if err != nil {
		return fmt.Errorf("counting parts: %w", err)
	}
	fmt.Fprintf(output, "%d readings sent, parts holding rows: %d\n", sent, parts.Parts)
	return nil
}

// tideReadings makes one reading a minute for a day at each station. The height rises and
// falls in a simple saw-tooth, and the reading on the hour carries a note.
//
// Takes stations (...string) which name the gauges.
//
// Returns []store.RecordReadingsParams which holds every reading, ready for the batch.
func tideReadings(stations ...string) []store.RecordReadingsParams {
	readings := make([]store.RecordReadingsParams, 0, len(stations)*minutesPerDay)
	for _, station := range stations {
		for minute := range uint16(minutesPerDay) {
			var note *string
			if minute%60 == 0 {
				note = new(fmt.Sprintf("hour %d check", minute/60))
			}
			readings = append(readings, store.RecordReadingsParams{
				Station:  station,
				Minute:   minute,
				HeightCm: 200 + minute%360,
				Note:     note,
			})
		}
	}
	return readings
}

main.go

// Command 022_bulk_writes_native applies its migrations to a ClickHouse database, then
// runs the example in example.go over a native clickhouse-go connection: a day of tide
// readings from two gauges is written with one batch call, as one INSERT sent through
// PrepareBatch.
//
// Run it with `go run .` against a local server, or `go run . -dsn <dsn>` to choose the
// database; give it an empty one.
package main

import (
	"context"
	"embed"
	"flag"
	"fmt"
	"log"
	"os"

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

	"seagull.sh/seagull/sdk/migrate"
	"seagull.sh/seagull/sdk/migrate/migrate_backend_clickhouse"
)

// migrations holds the migration files, compiled into the binary.
//
//go:embed db/migrations
var migrations embed.FS

func main() {
	dsn := flag.String("dsn", "clickhouse://localhost:9000/default", "ClickHouse DSN of an empty database")
	flag.Parse()
	if err := execute(context.Background(), *dsn); err != nil {
		log.Fatal(err)
	}
}

// execute migrates the database the DSN names, then opens a native connection to it and
// runs the example.
//
// Takes dsn (string) which names the ClickHouse database.
//
// Returns error when the database cannot be prepared or the example fails.
func execute(ctx context.Context, dsn string) error {
	options, err := clickhouse.ParseDSN(dsn)
	if err != nil {
		return err
	}

	// The migrator runs on database/sql, so the migrations go through the driver's
	// database/sql adapter, opened from the same options as the native connection.
	database := clickhouse.OpenDB(options)
	defer func() { _ = database.Close() }()
	migrator, err := migrate.NewMigrator(database, migrate_backend_clickhouse.New(), migrations, "db/migrations")
	if err != nil {
		return err
	}
	if _, err := migrator.Up(ctx); err != nil {
		return fmt.Errorf("migrating: %w", err)
	}

	connection, err := clickhouse.Open(options)
	if err != nil {
		return err
	}
	defer func() { _ = connection.Close() }()
	return run(ctx, connection, os.Stdout)
}

internal/store/models.go

// Code generated by sgl. DO NOT EDIT.

package store

type Readings struct {
	Station  string  `json:"station"`
	Minute   uint16  `json:"minute"`
	HeightCm uint16  `json:"height_cm"`
	Note     *string `json:"note"`
}

internal/store/querier.go

// Code generated by sgl. DO NOT EDIT.

package store

import (
	"context"

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

type DBTX interface {
	Query(ctx context.Context, query string, args ...any) (driver.Rows, error)
	QueryRow(ctx context.Context, query string, args ...any) driver.Row
	Exec(ctx context.Context, query string, args ...any) error
	PrepareBatch(ctx context.Context, query string, opts ...driver.PrepareBatchOption) (driver.Batch, error)
}

type Queries struct {
	db DBTX
}

func New(db DBTX) *Queries {
	return &Queries{db: db}
}

internal/store/readings.sql.go

// Code generated by sgl. DO NOT EDIT.

package store

import "context"

const countPartsSQL = `SELECT uniqExact(_part) AS parts FROM readings;`

type CountPartsRow struct {
	Parts uint64 `json:"parts"`
}

func (queries *Queries) CountParts(ctx context.Context) (CountPartsRow, error) {
	var row CountPartsRow
	err := queries.db.QueryRow(ctx, countPartsSQL).Scan(&row.Parts)
	if err != nil {
		return CountPartsRow{}, err
	}
	return row, nil
}

const recordReadingsSQL = `INSERT INTO readings (station, minute, height_cm, note)
VALUES ({p_station:String}, {p_minute:UInt16}, {p_height_cm:UInt16}, {p_note:Nullable(String)});`

type RecordReadingsParams struct {
	Station  string
	Minute   uint16
	HeightCm uint16
	Note     *string
}

func (queries *Queries) RecordReadings(ctx context.Context, params []RecordReadingsParams) error {
	if len(params) == 0 {
		return nil
	}
	batch, err := queries.db.PrepareBatch(ctx, recordReadingsSQL)
	if err != nil {
		return err
	}
	defer batch.Close()
	for _, item := range params {
		if err := batch.Append(item.Station, item.Minute, item.HeightCm, item.Note); err != nil {
			return err
		}
	}
	return batch.Send()
}

const summariseReadingsSQL = `SELECT station, count() AS readings, count(note) AS noted
FROM readings
GROUP BY station
ORDER BY station;`

type SummariseReadingsRow struct {
	Station  string `json:"station"`
	Readings uint64 `json:"readings"`
	Noted    uint64 `json:"noted"`
}

func (queries *Queries) SummariseReadings(ctx context.Context) ([]SummariseReadingsRow, error) {
	rows, err := queries.db.Query(ctx, summariseReadingsSQL)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var results []SummariseReadingsRow
	if rows.Next() {
		var row SummariseReadingsRow
		scanArgs := []any{&row.Station, &row.Readings, &row.Noted}
		for more := true; more; more = rows.Next() {
			row = SummariseReadingsRow{}
			if err := rows.Scan(scanArgs...); err != nil {
				return nil, err
			}
			results = append(results, row)
		}
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return results, nil
}

output.txt

0 readings sent, parts holding rows: 0
2880 readings sent, parts holding rows: 1
harbour  1440 readings, 24 with a note
pier     1440 readings, 24 with a note