Continuous aggregates
On this page
On TimescaleDB:
- the continuous aggregate's migration starts with -- sgl.migration(no_transaction: true), because CREATE MATERIALIZED VIEW ... WITH DATA cannot run in a transaction block; WITH NO DATA is the alternative
- CALL refresh_continuous_aggregate commits as it goes, so it fails inside RunInTx with SQLSTATE 25001 and is called on the pool instead
- timescaledb.materialized_only = true, so the view is empty until it is refreshed
- Seagull types the view's columns from its SELECT; stats_agg's statssummary1d would be any, so the queries read it through average, stddev, num_vals and rollup
- the refresh window is NULL to NULL, written into the SQL; as parameters, a CALL's bounds would be *time.Time fields named after the procedure's arguments
Summarising a long time series by the hour, every time someone opens a dashboard, means reading every raw row again. A continuous aggregate is TimescaleDB's answer: a materialised view over a hypertable that stores the summary, which a refresh brings up to date as new readings arrive. Queries read the small summary instead of the raw rows.
Seagull reads the view's SELECT from the migration, so queries against the view are typed like queries against a table. The example builds an hourly summary of the same buoy readings as hypertables and time buckets, refreshes it, and rolls it up over the whole night.
The hypertable
The first migration is the hypertable of readings, one chunk per day:
db/migrations/0001_readings.up.sql
CREATE EXTENSION IF NOT EXISTS timescaledb;
CREATE TABLE readings (
ts TIMESTAMPTZ NOT NULL,
buoy TEXT NOT NULL,
temperature DOUBLE PRECISION NOT NULL
);
SELECT create_hypertable('readings', 'ts', chunk_time_interval => INTERVAL '1 day');
The continuous aggregate
The second migration defines the view. timescaledb.continuous makes it a continuous aggregate, and for each buoy and hour it keeps a count and a stats_agg summary of the temperatures.
db/migrations/0002_hourly_readings.up.sql
-- sgl.migration(no_transaction: true)
CREATE MATERIALIZED VIEW hourly_readings
WITH (timescaledb.continuous, timescaledb.materialized_only = true) AS
SELECT
time_bucket(INTERVAL '1 hour', ts) AS bucket,
buoy,
count(*) AS samples,
stats_agg(temperature) AS stats
FROM readings
GROUP BY bucket, buoy;
The first line matters. TimescaleDB fills a new continuous aggregate as it creates it, and refuses to do that inside a transaction block. The migrator runs each migration in a transaction unless told otherwise, so the file starts with sgl.migration(no_transaction: true) and runs outside one. Keep such a statement in a file of its own, as here: sgl vet warns with Q043 when one shares a migration with other statements. Ending the view with WITH NO DATA would avoid the problem the other way, by creating it empty so that it can run in a transaction.
timescaledb.materialized_only = true makes the view return only what a refresh has written, which the example uses to show the refresh happening.
Refreshing
CALL refresh_continuous_aggregate materialises the view. Its window here is NULL to NULL, which means everything.
db/queries/readings.sql
-- sgl.query(RefreshHourly, exec)
CALL refresh_continuous_aggregate('hourly_readings', NULL, NULL);
The procedure commits as it goes, so it cannot run inside a transaction either. The program first tries it inside RunInTx, which fails with SQLSTATE 25001, and then calls it on the pool, outside any transaction, where it succeeds.
Reading the summary
Seagull types the view's columns from its SELECT: bucket is a time.Time and samples an int64. The stats column holds a TimescaleDB statssummary1d, which has no Go type, so the queries never select it directly. They read values out of it with hyperfunctions instead, and those are typed: average(stats) is the hour's mean.
db/queries/readings.sql
-- sgl.query(HourlyStats, many)
SELECT bucket, buoy, samples, average(stats) AS mean
FROM hourly_readings
ORDER BY bucket, buoy;
internal/store
type HourlyStatsRow struct {
Bucket time.Time `json:"bucket"`
Buoy string `json:"buoy"`
Samples int64 `json:"samples"`
Mean *float64 `json:"mean"`
}
func (queries *Queries) HourlyStats(ctx context.Context) ([]HourlyStatsRow, error) {
rows, err := queries.reader.QueryContext(ctx, hourlyStatsSQL)
if err != nil {
return nil, err
}
defer rows.Close()
var results []HourlyStatsRow
for rows.Next() {
var row HourlyStatsRow
if err := rows.Scan(&row.Bucket, &row.Buoy, &row.Samples, &row.Mean); err != nil {
return nil, err
}
results = append(results, row)
}
if err := rows.Err(); err != nil {
return nil, err
}
return results, nil
}
Rolling up
rollup combines several stats_agg summaries into one. A summary keeps enough to combine exactly, so average(rollup(stats)) is the mean of every reading in the night. Averaging the hourly means instead, as mean_of_hourly_means does, weights a quiet hour the same as a busy one.
db/queries/readings.sql
-- sgl.query(OverallStats, many)
SELECT
buoy,
num_vals(rollup(stats)) AS samples,
avg(average(stats)) AS mean_of_hourly_means,
average(rollup(stats)) AS mean,
stddev(rollup(stats)) AS spread
FROM hourly_readings
GROUP BY buoy
ORDER BY buoy;
Using it
example.go
package main
import (
"context"
"database/sql"
"errors"
"fmt"
"io"
"time"
"seagull.sh/seagull/examples/timescaledb/033_timescale_continuous_aggregates/internal/store"
)
// clock prints bucket times as hours and minutes in UTC.
const clock = "15:04"
// reading is one sea temperature sample from a buoy.
type reading struct {
// buoy names the buoy that took the sample.
buoy string
// minutes is how long after 22:00 UTC on 31 May the sample was taken.
minutes int
// temperature is the sea temperature in degrees Celsius.
temperature float64
}
// readings are two buoys' samples either side of midnight, the same as in
// 032_timescale_hypertables.
var readings = []reading{
{buoy: "Doom Bar", minutes: 0, temperature: 14.0},
{buoy: "Doom Bar", minutes: 20, temperature: 14.4},
{buoy: "Doom Bar", minutes: 40, temperature: 14.2},
{buoy: "Doom Bar", minutes: 70, temperature: 13.8},
{buoy: "Doom Bar", minutes: 110, temperature: 13.6},
{buoy: "Doom Bar", minutes: 195, temperature: 13.1},
{buoy: "Doom Bar", minutes: 225, temperature: 13.3},
{buoy: "Stepper Point", minutes: 5, temperature: 15.0},
{buoy: "Stepper Point", minutes: 90, temperature: 14.6},
{buoy: "Stepper Point", minutes: 210, temperature: 14.0},
}
// run records the buoys' readings, refreshes the hourly_readings continuous aggregate,
// and reads it back by the hour and rolled up over the whole night.
//
// Migration 0002 defines hourly_readings with CREATE MATERIALIZED VIEW ... WITH
// (timescaledb.continuous). TimescaleDB fills a new continuous aggregate as it creates
// it, which it refuses to do inside a transaction block, so the migration starts with
// sgl.migration(no_transaction: true) and runs outside one. Ending the view with WITH NO
// DATA would let it run in a transaction instead. Seagull reads the view's SELECT, so its
// columns arrive typed: bucket as time.Time, samples as int64.
//
// Takes database (*sql.DB) which is a migrated database.
// Takes output (io.Writer) which receives the summaries.
//
// Returns error when a query fails.
func run(ctx context.Context, database *sql.DB, output io.Writer) error {
queries := store.New(database)
if err := record(ctx, queries, database); err != nil {
return err
}
// timescaledb.materialized_only = true makes the view return only what a refresh has
// written, so the readings are not in it yet.
hours, err := queries.HourlyStats(ctx)
if err != nil {
return fmt.Errorf("reading the hourly summaries: %w", err)
}
fmt.Fprintf(output, "before the refresh, hourly_readings has %d rows\n", len(hours))
if err := refresh(ctx, queries, database, output); err != nil {
return err
}
if err := printHourly(ctx, queries, output); err != nil {
return err
}
return printOverall(ctx, queries, output)
}
// record writes the night's readings in one transaction, so they arrive together.
//
// Takes queries (*store.Queries) which runs the inserts.
// Takes database (*sql.DB) which begins the transaction.
//
// Returns error when an insert fails; none of the readings are kept then.
func record(ctx context.Context, queries *store.Queries, database *sql.DB) error {
start := time.Date(2026, time.May, 31, 22, 0, 0, 0, time.UTC)
return queries.RunInTx(ctx, database, func(transaction *store.Queries) error {
for _, sample := range readings {
err := transaction.RecordReading(ctx, store.RecordReadingParams{
Ts: start.Add(time.Duration(sample.minutes) * time.Minute),
Buoy: sample.buoy,
Temperature: sample.temperature,
})
if err != nil {
return fmt.Errorf("recording a reading: %w", err)
}
}
return nil
})
}
// refresh materialises the continuous aggregate with CALL refresh_continuous_aggregate.
//
// The procedure commits as it goes, so TimescaleDB refuses to run it inside a transaction
// block: the call in RunInTx fails, and the one on the pool, outside any transaction,
// succeeds. Its window is NULL to NULL, everything, written into the SQL. Parameters
// would work too: Seagull names a CALL's parameters after the procedure's arguments and
// types these as *time.Time, so a nil bound leaves that end open.
//
// Takes queries (*store.Queries) which runs the refresh.
// Takes database (*sql.DB) which begins the transaction.
// Takes output (io.Writer) which receives what each attempt did.
//
// Returns error when the refresh outside the transaction fails, or the one inside
// succeeds.
func refresh(ctx context.Context, queries *store.Queries, database *sql.DB, output io.Writer) error {
err := queries.RunInTx(ctx, database, func(transaction *store.Queries) error {
return transaction.RefreshHourly(ctx)
})
if err == nil {
return errors.New("refresh_continuous_aggregate ran inside a transaction")
}
fmt.Fprintf(output, "refreshing in RunInTx: %v\n", err)
if err := queries.RefreshHourly(ctx); err != nil {
return fmt.Errorf("refreshing hourly_readings: %w", err)
}
fmt.Fprintln(output, "refreshing outside a transaction: done")
return nil
}
// printHourly prints each buoy's hourly row from the continuous aggregate. The view keeps
// a stats_agg summary per hour, and average reads the mean out of it.
//
// Takes queries (*store.Queries) which runs the query.
// Takes output (io.Writer) which receives the rows.
//
// Returns error when the query fails.
func printHourly(ctx context.Context, queries *store.Queries, output io.Writer) error {
hours, err := queries.HourlyStats(ctx)
if err != nil {
return fmt.Errorf("reading the hourly summaries: %w", err)
}
fmt.Fprintf(output, "after the refresh, hourly_readings has %d rows:\n", len(hours))
for _, hour := range hours {
fmt.Fprintf(output, " %s %-13s readings %d mean %s\n",
hour.Bucket.UTC().Format(clock), hour.Buoy, hour.Samples, celsius(hour.Mean))
}
return nil
}
// printOverall prints each buoy's statistics for the whole night, combined from the
// hourly summaries with rollup. A stats_agg summary keeps enough to combine exactly, so
// the rolled-up mean is the mean of every reading; averaging the hourly means instead
// weights a quiet hour the same as a busy one, and gives a different answer.
//
// Takes queries (*store.Queries) which runs the query.
// Takes output (io.Writer) which receives the statistics.
//
// Returns error when the query fails.
func printOverall(ctx context.Context, queries *store.Queries, output io.Writer) error {
buoys, err := queries.OverallStats(ctx)
if err != nil {
return fmt.Errorf("rolling up the hourly summaries: %w", err)
}
fmt.Fprintln(output, "the whole night, rolled up from the hourly summaries:")
for _, buoy := range buoys {
samples := int64(0)
if buoy.Samples != nil {
samples = *buoy.Samples
}
fmt.Fprintf(output, " %-13s readings %d mean %s stddev %s mean of hourly means %s\n",
buoy.Buoy, samples, celsius(buoy.Mean), celsius(buoy.Spread), celsius(buoy.MeanOfHourlyMeans))
}
return nil
}
// celsius formats a temperature that may be NULL.
//
// Takes value (*float64) which is the temperature, or nil.
//
// Returns string which is the temperature to two decimal places, or a dash for nil.
func celsius(value *float64) string {
if value == nil {
return "-"
}
return fmt.Sprintf("%.2f", *value)
}
What it prints
The view is empty until the refresh, and the refresh fails inside the transaction and succeeds outside it. Doom Bar's overall mean differs from the mean of its hourly means because its hours hold three, two and two readings; Stepper Point took one reading an hour, so its two figures agree.
before the refresh, hourly_readings has 0 rows
refreshing in RunInTx: ERROR: refresh_continuous_aggregate() cannot run inside a transaction block (SQLSTATE 25001)
refreshing outside a transaction: done
after the refresh, hourly_readings has 6 rows:
22:00 Doom Bar readings 3 mean 14.20
22:00 Stepper Point readings 1 mean 15.00
23:00 Doom Bar readings 2 mean 13.70
23:00 Stepper Point readings 1 mean 14.60
01:00 Doom Bar readings 2 mean 13.20
01:00 Stepper Point readings 1 mean 14.00
the whole night, rolled up from the hourly summaries:
Doom Bar readings 7 mean 13.77 stddev 0.47 mean of hourly means 13.70
Stepper Point readings 3 mean 14.53 stddev 0.50 mean of hourly means 14.53
Try it
Add stats itself to the select list of HourlyStats and run sgl generate. The new field is any, because Seagull has no Go type for a statssummary1d, which is why the queries read it through average, stddev, num_vals and rollup.
Run it locally
From the directory you cloned the Seagull repository into:
cd seagull/examples/timescaledb && go run ./033_timescale_continuous_aggregates
Its test runs it too, and checks what it prints against output.txt:
cd seagull/examples/timescaledb && SGL_EXAMPLE_DSN="$DATABASE_URL" go test ./033_timescale_continuous_aggregates
Needs a running TimescaleDB 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: buoys
dialect: timescaledb
migrations: db/migrations
queries: db/queries
go:
package: store
out: internal/store
db/migrations/0001_readings.up.sql
CREATE EXTENSION IF NOT EXISTS timescaledb;
CREATE TABLE readings (
ts TIMESTAMPTZ NOT NULL,
buoy TEXT NOT NULL,
temperature DOUBLE PRECISION NOT NULL
);
SELECT create_hypertable('readings', 'ts', chunk_time_interval => INTERVAL '1 day');
db/migrations/0002_hourly_readings.up.sql
-- sgl.migration(no_transaction: true)
CREATE MATERIALIZED VIEW hourly_readings
WITH (timescaledb.continuous, timescaledb.materialized_only = true) AS
SELECT
time_bucket(INTERVAL '1 hour', ts) AS bucket,
buoy,
count(*) AS samples,
stats_agg(temperature) AS stats
FROM readings
GROUP BY bucket, buoy;
db/migrations/0001_readings.down.sql
DROP TABLE readings;
db/migrations/0002_hourly_readings.down.sql
DROP MATERIALIZED VIEW hourly_readings;
db/queries/readings.sql
-- sgl.query(RecordReading, exec)
INSERT INTO readings (ts, buoy, temperature) VALUES ($1, $2, $3);
-- sgl.query(RefreshHourly, exec)
CALL refresh_continuous_aggregate('hourly_readings', NULL, NULL);
-- sgl.query(HourlyStats, many)
SELECT bucket, buoy, samples, average(stats) AS mean
FROM hourly_readings
ORDER BY bucket, buoy;
-- sgl.query(OverallStats, many)
SELECT
buoy,
num_vals(rollup(stats)) AS samples,
avg(average(stats)) AS mean_of_hourly_means,
average(rollup(stats)) AS mean,
stddev(rollup(stats)) AS spread
FROM hourly_readings
GROUP BY buoy
ORDER BY buoy;
example.go
package main
import (
"context"
"database/sql"
"errors"
"fmt"
"io"
"time"
"seagull.sh/seagull/examples/timescaledb/033_timescale_continuous_aggregates/internal/store"
)
// clock prints bucket times as hours and minutes in UTC.
const clock = "15:04"
// reading is one sea temperature sample from a buoy.
type reading struct {
// buoy names the buoy that took the sample.
buoy string
// minutes is how long after 22:00 UTC on 31 May the sample was taken.
minutes int
// temperature is the sea temperature in degrees Celsius.
temperature float64
}
// readings are two buoys' samples either side of midnight, the same as in
// 032_timescale_hypertables.
var readings = []reading{
{buoy: "Doom Bar", minutes: 0, temperature: 14.0},
{buoy: "Doom Bar", minutes: 20, temperature: 14.4},
{buoy: "Doom Bar", minutes: 40, temperature: 14.2},
{buoy: "Doom Bar", minutes: 70, temperature: 13.8},
{buoy: "Doom Bar", minutes: 110, temperature: 13.6},
{buoy: "Doom Bar", minutes: 195, temperature: 13.1},
{buoy: "Doom Bar", minutes: 225, temperature: 13.3},
{buoy: "Stepper Point", minutes: 5, temperature: 15.0},
{buoy: "Stepper Point", minutes: 90, temperature: 14.6},
{buoy: "Stepper Point", minutes: 210, temperature: 14.0},
}
// run records the buoys' readings, refreshes the hourly_readings continuous aggregate,
// and reads it back by the hour and rolled up over the whole night.
//
// Migration 0002 defines hourly_readings with CREATE MATERIALIZED VIEW ... WITH
// (timescaledb.continuous). TimescaleDB fills a new continuous aggregate as it creates
// it, which it refuses to do inside a transaction block, so the migration starts with
// sgl.migration(no_transaction: true) and runs outside one. Ending the view with WITH NO
// DATA would let it run in a transaction instead. Seagull reads the view's SELECT, so its
// columns arrive typed: bucket as time.Time, samples as int64.
//
// Takes database (*sql.DB) which is a migrated database.
// Takes output (io.Writer) which receives the summaries.
//
// Returns error when a query fails.
func run(ctx context.Context, database *sql.DB, output io.Writer) error {
queries := store.New(database)
if err := record(ctx, queries, database); err != nil {
return err
}
// timescaledb.materialized_only = true makes the view return only what a refresh has
// written, so the readings are not in it yet.
hours, err := queries.HourlyStats(ctx)
if err != nil {
return fmt.Errorf("reading the hourly summaries: %w", err)
}
fmt.Fprintf(output, "before the refresh, hourly_readings has %d rows\n", len(hours))
if err := refresh(ctx, queries, database, output); err != nil {
return err
}
if err := printHourly(ctx, queries, output); err != nil {
return err
}
return printOverall(ctx, queries, output)
}
// record writes the night's readings in one transaction, so they arrive together.
//
// Takes queries (*store.Queries) which runs the inserts.
// Takes database (*sql.DB) which begins the transaction.
//
// Returns error when an insert fails; none of the readings are kept then.
func record(ctx context.Context, queries *store.Queries, database *sql.DB) error {
start := time.Date(2026, time.May, 31, 22, 0, 0, 0, time.UTC)
return queries.RunInTx(ctx, database, func(transaction *store.Queries) error {
for _, sample := range readings {
err := transaction.RecordReading(ctx, store.RecordReadingParams{
Ts: start.Add(time.Duration(sample.minutes) * time.Minute),
Buoy: sample.buoy,
Temperature: sample.temperature,
})
if err != nil {
return fmt.Errorf("recording a reading: %w", err)
}
}
return nil
})
}
// refresh materialises the continuous aggregate with CALL refresh_continuous_aggregate.
//
// The procedure commits as it goes, so TimescaleDB refuses to run it inside a transaction
// block: the call in RunInTx fails, and the one on the pool, outside any transaction,
// succeeds. Its window is NULL to NULL, everything, written into the SQL. Parameters
// would work too: Seagull names a CALL's parameters after the procedure's arguments and
// types these as *time.Time, so a nil bound leaves that end open.
//
// Takes queries (*store.Queries) which runs the refresh.
// Takes database (*sql.DB) which begins the transaction.
// Takes output (io.Writer) which receives what each attempt did.
//
// Returns error when the refresh outside the transaction fails, or the one inside
// succeeds.
func refresh(ctx context.Context, queries *store.Queries, database *sql.DB, output io.Writer) error {
err := queries.RunInTx(ctx, database, func(transaction *store.Queries) error {
return transaction.RefreshHourly(ctx)
})
if err == nil {
return errors.New("refresh_continuous_aggregate ran inside a transaction")
}
fmt.Fprintf(output, "refreshing in RunInTx: %v\n", err)
if err := queries.RefreshHourly(ctx); err != nil {
return fmt.Errorf("refreshing hourly_readings: %w", err)
}
fmt.Fprintln(output, "refreshing outside a transaction: done")
return nil
}
// printHourly prints each buoy's hourly row from the continuous aggregate. The view keeps
// a stats_agg summary per hour, and average reads the mean out of it.
//
// Takes queries (*store.Queries) which runs the query.
// Takes output (io.Writer) which receives the rows.
//
// Returns error when the query fails.
func printHourly(ctx context.Context, queries *store.Queries, output io.Writer) error {
hours, err := queries.HourlyStats(ctx)
if err != nil {
return fmt.Errorf("reading the hourly summaries: %w", err)
}
fmt.Fprintf(output, "after the refresh, hourly_readings has %d rows:\n", len(hours))
for _, hour := range hours {
fmt.Fprintf(output, " %s %-13s readings %d mean %s\n",
hour.Bucket.UTC().Format(clock), hour.Buoy, hour.Samples, celsius(hour.Mean))
}
return nil
}
// printOverall prints each buoy's statistics for the whole night, combined from the
// hourly summaries with rollup. A stats_agg summary keeps enough to combine exactly, so
// the rolled-up mean is the mean of every reading; averaging the hourly means instead
// weights a quiet hour the same as a busy one, and gives a different answer.
//
// Takes queries (*store.Queries) which runs the query.
// Takes output (io.Writer) which receives the statistics.
//
// Returns error when the query fails.
func printOverall(ctx context.Context, queries *store.Queries, output io.Writer) error {
buoys, err := queries.OverallStats(ctx)
if err != nil {
return fmt.Errorf("rolling up the hourly summaries: %w", err)
}
fmt.Fprintln(output, "the whole night, rolled up from the hourly summaries:")
for _, buoy := range buoys {
samples := int64(0)
if buoy.Samples != nil {
samples = *buoy.Samples
}
fmt.Fprintf(output, " %-13s readings %d mean %s stddev %s mean of hourly means %s\n",
buoy.Buoy, samples, celsius(buoy.Mean), celsius(buoy.Spread), celsius(buoy.MeanOfHourlyMeans))
}
return nil
}
// celsius formats a temperature that may be NULL.
//
// Takes value (*float64) which is the temperature, or nil.
//
// Returns string which is the temperature to two decimal places, or a dash for nil.
func celsius(value *float64) string {
if value == nil {
return "-"
}
return fmt.Sprintf("%.2f", *value)
}
main.go
// Command 033_timescale_continuous_aggregates applies its migrations to a TimescaleDB
// database, the second of which creates a continuous aggregate outside a transaction,
// then runs the example in example.go: two buoys' sea temperatures are recorded, the
// aggregate is refreshed, and its hourly summaries are read and rolled up.
//
// Run it with `go run .` against a local server, which `docker run -d -p 5432:5432 -e
// POSTGRES_PASSWORD=postgres timescale/timescaledb-ha:pg17` starts, or point it elsewhere
// with `go run . -dsn <connection string>` or DATABASE_URL. Use an empty database: the
// example's summaries assume only its own readings.
package main
import (
"cmp"
"context"
"database/sql"
"embed"
"flag"
"fmt"
"log"
"os"
_ "github.com/jackc/pgx/v5/stdlib"
"seagull.sh/seagull/sdk/migrate"
"seagull.sh/seagull/sdk/migrate/migrate_backend_postgres"
)
// localDSN reaches the server the docker command above starts, with the password that
// command gives it.
const localDSN = "postgresql://postgres:postgres@localhost:5432/postgres?sslmode=disable" //nolint:gosec // G101: the docker command's throwaway password, not a secret
// migrations holds the migration files, compiled into the binary.
//
//go:embed db/migrations
var migrations embed.FS
func main() {
dsn := flag.String("dsn", cmp.Or(os.Getenv("DATABASE_URL"), localDSN), "TimescaleDB connection string")
flag.Parse()
if err := execute(context.Background(), *dsn); err != nil {
log.Fatal(err)
}
}
// execute connects to the database at dsn, migrates it with the PostgreSQL backend, which
// TimescaleDB uses, and runs the example against it.
//
// Takes dsn (string) which is the connection string.
//
// Returns error when the database cannot be prepared or the example fails.
func execute(ctx context.Context, dsn string) error {
database, err := sql.Open("pgx", dsn)
if err != nil {
return err
}
defer func() { _ = database.Close() }()
migrator, err := migrate.NewMigrator(database, migrate_backend_postgres.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
import "time"
type Readings struct {
Ts time.Time `json:"ts"`
Buoy string `json:"buoy"`
Temperature float64 `json:"temperature"`
}
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"
"time"
)
const hourlyStatsSQL = `SELECT bucket, buoy, samples, average(stats) AS mean
FROM hourly_readings
ORDER BY bucket, buoy;`
type HourlyStatsRow struct {
Bucket time.Time `json:"bucket"`
Buoy string `json:"buoy"`
Samples int64 `json:"samples"`
Mean *float64 `json:"mean"`
}
func (queries *Queries) HourlyStats(ctx context.Context) ([]HourlyStatsRow, error) {
rows, err := queries.reader.QueryContext(ctx, hourlyStatsSQL)
if err != nil {
return nil, err
}
defer rows.Close()
var results []HourlyStatsRow
for rows.Next() {
var row HourlyStatsRow
if err := rows.Scan(&row.Bucket, &row.Buoy, &row.Samples, &row.Mean); err != nil {
return nil, err
}
results = append(results, row)
}
if err := rows.Err(); err != nil {
return nil, err
}
return results, nil
}
const overallStatsSQL = `SELECT
buoy,
num_vals(rollup(stats)) AS samples,
avg(average(stats)) AS mean_of_hourly_means,
average(rollup(stats)) AS mean,
stddev(rollup(stats)) AS spread
FROM hourly_readings
GROUP BY buoy
ORDER BY buoy;`
type OverallStatsRow struct {
Buoy string `json:"buoy"`
Samples *int64 `json:"samples"`
MeanOfHourlyMeans *float64 `json:"mean_of_hourly_means"`
Mean *float64 `json:"mean"`
Spread *float64 `json:"spread"`
}
func (queries *Queries) OverallStats(ctx context.Context) ([]OverallStatsRow, error) {
rows, err := queries.reader.QueryContext(ctx, overallStatsSQL)
if err != nil {
return nil, err
}
defer rows.Close()
var results []OverallStatsRow
for rows.Next() {
var row OverallStatsRow
if err := rows.Scan(&row.Buoy, &row.Samples, &row.MeanOfHourlyMeans, &row.Mean, &row.Spread); err != nil {
return nil, err
}
results = append(results, row)
}
if err := rows.Err(); err != nil {
return nil, err
}
return results, nil
}
const recordReadingSQL = `INSERT INTO readings (ts, buoy, temperature) VALUES ($1, $2, $3);`
type RecordReadingParams struct {
Ts time.Time
Buoy string
Temperature float64
}
func (queries *Queries) RecordReading(ctx context.Context, params RecordReadingParams) error {
_, err := queries.writer.ExecContext(ctx, recordReadingSQL, params.Ts, params.Buoy, params.Temperature)
return err
}
const refreshHourlySQL = `CALL refresh_continuous_aggregate('hourly_readings', NULL, NULL);`
func (queries *Queries) RefreshHourly(ctx context.Context) error {
_, err := queries.writer.ExecContext(ctx, refreshHourlySQL)
return err
}
output.txt
before the refresh, hourly_readings has 0 rows
refreshing in RunInTx: ERROR: refresh_continuous_aggregate() cannot run inside a transaction block (SQLSTATE 25001)
refreshing outside a transaction: done
after the refresh, hourly_readings has 6 rows:
22:00 Doom Bar readings 3 mean 14.20
22:00 Stepper Point readings 1 mean 15.00
23:00 Doom Bar readings 2 mean 13.70
23:00 Stepper Point readings 1 mean 14.60
01:00 Doom Bar readings 2 mean 13.20
01:00 Stepper Point readings 1 mean 14.00
the whole night, rolled up from the hourly summaries:
Doom Bar readings 7 mean 13.77 stddev 0.47 mean of hourly means 13.70
Stepper Point readings 3 mean 14.53 stddev 0.50 mean of hourly means 14.53