Range over rows

On DuckDB:

• StreamReadings returns func(func(StreamReadingsRow, error) bool), the shape of iter.Seq2

A many query reads every row into a slice before it returns, which is what you want for a page of results, and not what you want for a large export, a report over a whole table, or a scan that can stop at the first match. The stream command returns an iterator instead. Each row is scanned only when the loop asks for it, so the result is never held in memory at once, and breaking out of the loop stops reading.

This example records a morning of tide gauge readings and reads them back twice: once to the end, and once stopping at the first reading high enough for a flood warning.

The query

The SQL is an ordinary SELECT. Only the command changes:

db/queries/readings.sql

-- sgl.query(StreamReadings, stream)
SELECT taken_at, level_cm FROM readings ORDER BY taken_at;

What sgl generates

internal/store

type StreamReadingsRow struct {
	TakenAt string `json:"taken_at"`
	LevelCm int32  `json:"level_cm"`
}

func (queries *Queries) StreamReadings(ctx context.Context) iter.Seq2[StreamReadingsRow, error] {
	return func(yield func(StreamReadingsRow, error) bool) {
		rows, err := queries.reader.QueryContext(ctx, streamReadingsSQL)
		if err != nil {
			yield(StreamReadingsRow{}, err)
			return
		}
		defer rows.Close()
		if rows.Next() {
			var row StreamReadingsRow
			scanArgs := []any{&row.TakenAt, &row.LevelCm}
			for more := true; more; more = rows.Next() {
				row = StreamReadingsRow{}
				if err := rows.Scan(scanArgs...); err != nil {
					if !yield(StreamReadingsRow{}, err) {
						return
					}
					continue
				}
				if !yield(row, nil) {
					return
				}
			}
		}
		if err := rows.Err(); err != nil {
			yield(StreamReadingsRow{}, err)
		}
	}
}

StreamReadings returns a func(func(StreamReadingsRow, error) bool), the shape of iter.Seq2, so it works directly with for ... range:

• Calling the method does nothing yet. The query runs when the loop starts.
• Each step of the loop gets a row and an error. When a row fails to scan, the loop gets the error with an empty row, and the next step carries on with the next row, so return on the error yourself.
• Breaking out of the loop makes yield return false, and the iterator returns, which closes the rows and hands the connection back to the pool.

Using it

The first loop reads every reading to find the high water mark. The second stops as soon as a reading reaches the flood warning level, after five of the eight rows; the other three are never scanned. Both loops return on the first error.

example.go

package main

import (
	"context"
	"database/sql"
	"fmt"
	"io"

	"seagull.sh/seagull/examples/duckdb/017_streaming/internal/store"
)

// floodWarningCm is the tide level, in centimetres, that raises a flood warning.
const floodWarningCm = 450

// run records a morning of tide gauge readings and reads them back twice with
// StreamReadings. The stream command returns a range-over-func iterator instead of a
// slice, so each row is scanned only when the loop asks for it and a large result is
// never held in memory at once. Every step yields a row and an error; return on the
// error, as the loop goes on otherwise.
//
// Breaking out of the loop stops the query: the iterator closes its rows, which hands the
// connection back to the pool.
//
// Takes database (*sql.DB) which is a migrated database.
// Takes output (io.Writer) which receives the summary and the warning.
//
// Returns error when a query fails.
func run(ctx context.Context, database *sql.DB, output io.Writer) error {
	queries := store.New(database)

	readings := []store.RecordReadingParams{
		{TakenAt: "06:00", LevelCm: 210},
		{TakenAt: "07:00", LevelCm: 280},
		{TakenAt: "08:00", LevelCm: 350},
		{TakenAt: "09:00", LevelCm: 420},
		{TakenAt: "10:00", LevelCm: 470},
		{TakenAt: "11:00", LevelCm: 495},
		{TakenAt: "12:00", LevelCm: 460},
		{TakenAt: "13:00", LevelCm: 390},
	}
	for _, reading := range readings {
		if err := queries.RecordReading(ctx, reading); err != nil {
			return fmt.Errorf("recording the %s reading: %w", reading.TakenAt, err)
		}
	}

	count := 0
	var highest store.StreamReadingsRow
	for reading, err := range queries.StreamReadings(ctx) {
		if err != nil {
			return fmt.Errorf("streaming readings: %w", err)
		}
		count++
		if reading.LevelCm > highest.LevelCm {
			highest = reading
		}
	}
	fmt.Fprintf(output, "%d readings, high water %dcm at %s\n", count, highest.LevelCm, highest.TakenAt)

	read := 0
	for reading, err := range queries.StreamReadings(ctx) {
		if err != nil {
			return fmt.Errorf("streaming readings: %w", err)
		}
		read++
		if reading.LevelCm >= floodWarningCm {
			fmt.Fprintf(output, "flood warning at %s, %dcm, after reading %d of %d rows\n", reading.TakenAt, reading.LevelCm, read, count)
			break
		}
	}
	return nil
}

What it prints

8 readings, high water 495cm at 11:00
flood warning at 10:00, 470cm, after reading 5 of 8 rows

Stopping early saves the scanning, but the database may already have run the whole query. To read only the first few rows, put a LIMIT in the SQL, as in pagination.

Across engines

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

	DuckDB	SQLite	PostgreSQL	MySQL
level_cm column	int32	int64	int64	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 and MySQL.

Run it locally

From the directory you cloned the Seagull repository into:

cd seagull/examples/duckdb && go run ./017_streaming

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

cd seagull/examples/duckdb && go test ./017_streaming

Needs cgo and a C compiler (the DuckDB driver).

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: duckdb
    migrations: db/migrations
    queries: db/queries
    go:
      package: store
      out: internal/store

db/migrations/0001_readings.up.sql

CREATE SEQUENCE reading_ids;

CREATE TABLE readings (
    id       BIGINT  PRIMARY KEY DEFAULT nextval('reading_ids'),
    taken_at VARCHAR NOT NULL,
    level_cm INTEGER NOT NULL
);

db/migrations/0001_readings.down.sql

DROP TABLE readings;
DROP SEQUENCE reading_ids;

db/queries/readings.sql

-- sgl.query(RecordReading, exec)
INSERT INTO readings (taken_at, level_cm) VALUES ($1, $2);

-- sgl.query(StreamReadings, stream)
SELECT taken_at, level_cm FROM readings ORDER BY taken_at;

example.go

package main

import (
	"context"
	"database/sql"
	"fmt"
	"io"

	"seagull.sh/seagull/examples/duckdb/017_streaming/internal/store"
)

// floodWarningCm is the tide level, in centimetres, that raises a flood warning.
const floodWarningCm = 450

// run records a morning of tide gauge readings and reads them back twice with
// StreamReadings. The stream command returns a range-over-func iterator instead of a
// slice, so each row is scanned only when the loop asks for it and a large result is
// never held in memory at once. Every step yields a row and an error; return on the
// error, as the loop goes on otherwise.
//
// Breaking out of the loop stops the query: the iterator closes its rows, which hands the
// connection back to the pool.
//
// Takes database (*sql.DB) which is a migrated database.
// Takes output (io.Writer) which receives the summary and the warning.
//
// Returns error when a query fails.
func run(ctx context.Context, database *sql.DB, output io.Writer) error {
	queries := store.New(database)

	readings := []store.RecordReadingParams{
		{TakenAt: "06:00", LevelCm: 210},
		{TakenAt: "07:00", LevelCm: 280},
		{TakenAt: "08:00", LevelCm: 350},
		{TakenAt: "09:00", LevelCm: 420},
		{TakenAt: "10:00", LevelCm: 470},
		{TakenAt: "11:00", LevelCm: 495},
		{TakenAt: "12:00", LevelCm: 460},
		{TakenAt: "13:00", LevelCm: 390},
	}
	for _, reading := range readings {
		if err := queries.RecordReading(ctx, reading); err != nil {
			return fmt.Errorf("recording the %s reading: %w", reading.TakenAt, err)
		}
	}

	count := 0
	var highest store.StreamReadingsRow
	for reading, err := range queries.StreamReadings(ctx) {
		if err != nil {
			return fmt.Errorf("streaming readings: %w", err)
		}
		count++
		if reading.LevelCm > highest.LevelCm {
			highest = reading
		}
	}
	fmt.Fprintf(output, "%d readings, high water %dcm at %s\n", count, highest.LevelCm, highest.TakenAt)

	read := 0
	for reading, err := range queries.StreamReadings(ctx) {
		if err != nil {
			return fmt.Errorf("streaming readings: %w", err)
		}
		read++
		if reading.LevelCm >= floodWarningCm {
			fmt.Fprintf(output, "flood warning at %s, %dcm, after reading %d of %d rows\n", reading.TakenAt, reading.LevelCm, read, count)
			break
		}
	}
	return nil
}

main.go

// Command 017_streaming applies its migrations to an in-memory DuckDB database, then runs
// the example in example.go: tide gauge readings are ranged over one row at a time, once
// to the end and once stopping early.
//
// Run it with `go run .`, or `go run . -db example.duckdb` to keep the database in a
// file.
package main

import (
	"context"
	"database/sql"
	"embed"
	"flag"
	"fmt"
	"log"
	"os"

	_ "github.com/duckdb/duckdb-go/v2"

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

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

func main() {
	path := flag.String("db", "", "DuckDB database file; an in-memory database when empty")
	flag.Parse()
	if err := execute(context.Background(), *path); err != nil {
		log.Fatal(err)
	}
}

// execute opens and migrates the DuckDB database at path, or an in-memory one when path
// is empty, and runs the example against it.
//
// Takes path (string) which is the database file, or empty.
//
// Returns error when the database cannot be prepared or the example fails.
func execute(ctx context.Context, path string) error {
	database, err := sql.Open("duckdb", path)
	if err != nil {
		return err
	}
	defer func() { _ = database.Close() }()

	migrator, err := migrate.NewMigrator(database, migrate_backend_duckdb.New(), migrations, "db/migrations")
	if err != nil {
		return err
	}
	if _, err := migrator.Up(ctx); err != nil {
		return fmt.Errorf("migrating: %w", err)
	}
	return run(ctx, database, os.Stdout)
}

internal/store/models.go

// Code generated by sgl. DO NOT EDIT.

package store

type Readings struct {
	ID      int64  `json:"id"`
	TakenAt string `json:"taken_at"`
	LevelCm int32  `json:"level_cm"`
}

internal/store/querier.go

// Code generated by sgl. DO NOT EDIT.

package store

import (
	"context"
	"database/sql"
)

type DBTX interface {
	ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
	QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error)
	QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row
}

type dbtxWrapper interface {
	WrapDBTX(inner any) any
}

type Queries struct {
	reader  DBTX
	writer  DBTX
	wrapper dbtxWrapper
}

func New(db DBTX) *Queries {
	wrapper, _ := db.(dbtxWrapper)
	return &Queries{reader: db, writer: db, wrapper: wrapper}
}

func NewWithReplica(writer DBTX, reader DBTX) *Queries {
	wrapper, _ := writer.(dbtxWrapper)
	return &Queries{reader: reader, writer: writer, wrapper: wrapper}
}

func (queries *Queries) WithTx(transaction *sql.Tx) *Queries {
	if queries == nil {
		return &Queries{reader: transaction, writer: transaction, wrapper: nil}
	}
	db := queries.wrapDBTX(transaction)
	return &Queries{reader: db, writer: db, wrapper: queries.wrapper}
}

func (queries *Queries) wrapDBTX(db DBTX) DBTX {
	if queries.wrapper == nil {
		return db
	}
	wrapped, ok := queries.wrapper.WrapDBTX(db).(DBTX)
	if !ok {
		return db
	}
	return wrapped
}

func (queries *Queries) RunInTx(ctx context.Context, db *sql.DB, fn func(*Queries) error) error {
	transaction, err := db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer transaction.Rollback()
	if err = fn(queries.WithTx(transaction)); err != nil {
		return err
	}
	return transaction.Commit()
}

internal/store/readings.sql.go

// Code generated by sgl. DO NOT EDIT.

package store

import (
	"context"
	"iter"
)

const recordReadingSQL = `INSERT INTO readings (taken_at, level_cm) VALUES ($1, $2);`

type RecordReadingParams struct {
	TakenAt string
	LevelCm int32
}

func (queries *Queries) RecordReading(ctx context.Context, params RecordReadingParams) error {
	_, err := queries.writer.ExecContext(ctx, recordReadingSQL, params.TakenAt, params.LevelCm)
	return err
}

const streamReadingsSQL = `SELECT taken_at, level_cm FROM readings ORDER BY taken_at;`

type StreamReadingsRow struct {
	TakenAt string `json:"taken_at"`
	LevelCm int32  `json:"level_cm"`
}

func (queries *Queries) StreamReadings(ctx context.Context) iter.Seq2[StreamReadingsRow, error] {
	return func(yield func(StreamReadingsRow, error) bool) {
		rows, err := queries.reader.QueryContext(ctx, streamReadingsSQL)
		if err != nil {
			yield(StreamReadingsRow{}, err)
			return
		}
		defer rows.Close()
		if rows.Next() {
			var row StreamReadingsRow
			scanArgs := []any{&row.TakenAt, &row.LevelCm}
			for more := true; more; more = rows.Next() {
				row = StreamReadingsRow{}
				if err := rows.Scan(scanArgs...); err != nil {
					if !yield(StreamReadingsRow{}, err) {
						return
					}
					continue
				}
				if !yield(row, nil) {
					return
				}
			}
		}
		if err := rows.Err(); err != nil {
			yield(StreamReadingsRow{}, err)
		}
	}
}

output.txt

8 readings, high water 495cm at 11:00
flood warning at 10:00, 470cm, after reading 5 of 8 rows