Range over rows

On MySQL:

• StreamReadings returns func(func(StreamReadingsRow, error) bool), the shape of iter.Seq2
• go-sql-driver/mysql reads rows from the connection as the loop asks for them; breaking out closes the result set, and the driver discards the rows the server had still to send

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/mysql/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.

	MySQL	SQLite	PostgreSQL	DuckDB
level_cm column	int32	int64	int64	int32

MariaDB runs the MySQL version unchanged.

CockroachDB and TimescaleDB run the PostgreSQL version unchanged.

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

Run it locally

From the directory you cloned the Seagull repository into:

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

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

cd seagull/examples/mysql && SGL_EXAMPLE_DSN="$DATABASE_URL" go test ./017_streaming

Needs a running MySQL server: set DATABASE_URL, or pass -dsn, to 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: mysql
    migrations: db/migrations
    queries: db/queries
    go:
      package: store
      out: internal/store

db/migrations/0001_readings.up.sql

CREATE TABLE readings (
    id       BIGINT  NOT NULL AUTO_INCREMENT PRIMARY KEY,
    taken_at CHAR(5) NOT NULL,
    level_cm INT     NOT NULL
);

db/migrations/0001_readings.down.sql

DROP TABLE readings;

db/queries/readings.sql

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

-- 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/mysql/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 a MySQL 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 at the first high reading.
//
// Run it with `go run . -dsn 'user:password@tcp(localhost:3306)/tides'`, or set
// DATABASE_URL instead of passing -dsn. Give it an empty database of its own, as the
// migrations create its tables and the example adds rows each time it runs.
package main

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

	"github.com/go-sql-driver/mysql"

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

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

func main() {
	dsn := flag.String("dsn", os.Getenv("DATABASE_URL"), "MySQL data source name; defaults to $DATABASE_URL")
	flag.Parse()
	if err := execute(context.Background(), *dsn); err != nil {
		log.Fatal(err)
	}
}

// execute opens and migrates the database dsn names, and runs the example against it.
//
// Takes dsn (string) which is the database's data source name.
//
// Returns error when the database cannot be prepared or the example fails.
func execute(ctx context.Context, dsn string) error {
	if dsn == "" {
		return errors.New("set -dsn or DATABASE_URL to a MySQL data source name")
	}
	config, err := mysql.ParseDSN(dsn)
	if err != nil {
		return err
	}
	// Seagull maps DATETIME and TIMESTAMP columns to time.Time, which the driver returns
	// only when parseTime is set.
	config.ParseTime = true
	dsn = config.FormatDSN()

	database, err := sql.Open("mysql", dsn)
	if err != nil {
		return err
	}
	defer func() { _ = database.Close() }()

	migrator, err := migrate.NewMigrator(database, migrate_backend_mysql.New(migrate.WithDriverMultiStatements(config.MultiStatements)), 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 reads DATETIME, DATE and TIMESTAMP columns into time.Time, so the DSN it connects with must set parseTime=true.
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 (?, ?);`

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