Compare · 027

Migrations at start-up: CockroachDB and ClickHouse

Every file of the CockroachDB and ClickHouse versions side by side, with the lines that differ marked.

Compare CockroachDB withSQLitePostgreSQLTimescaleDBMySQLMariaDBDuckDBClickHouse

Configuration

sgl.yamldiffers
CockroachDB
ClickHouse
1version: 1
1version: 1
2
2
3databases:
3databases:
4 - name: harbour
4 - name: harbour
5 dialect: cockroachdb
5 dialect: clickhouse
6 migrations: db/migrations
6 migrations: db/migrations
7 queries: db/queries
7 queries: db/queries
8 go:
8 go:
9 package: store
9 package: store
10 out: internal/store
10 out: internal/store

Migrations

db/migrations/0001_boats.up.sqldiffers
CockroachDB
ClickHouse
1CREATE TABLE boats (
1CREATE TABLE boats (
2 id INT8 NOT NULL DEFAULT unique_rowid() PRIMARY KEY,
2 id UInt64,
3 name STRING NOT NULL
3 name String
4);
4) ENGINE = MergeTree() ORDER BY id;
db/migrations/0001_boats.down.sqlidentical
CockroachDB
ClickHouse
1DROP TABLE boats;
1DROP TABLE boats;
db/migrations/0002_boat_home_port.up.sqldiffers
CockroachDB
ClickHouse
1ALTER TABLE boats ADD COLUMN home_port STRING NOT NULL DEFAULT 'Padstow';
1ALTER TABLE boats ADD COLUMN home_port String DEFAULT 'Padstow';
db/migrations/0002_boat_home_port.down.sqlidentical
CockroachDB
ClickHouse
1ALTER TABLE boats DROP COLUMN home_port;
1ALTER TABLE boats DROP COLUMN home_port;

Queries

AddBoatdiffers
CockroachDB · db/queries/boats.sql
ClickHouse · db/queries/boats.sql
1-- sgl.query(AddBoat, exec)
1-- sgl.query(AddBoat, exec)
2INSERT INTO boats (name, home_port) VALUES ($1, $2);
2INSERT INTO boats (id, name, home_port) VALUES ({id:UInt64}, {name:String}, {home_port:String});
ListBoatsdiffers
CockroachDB · db/queries/boats.sql
ClickHouse · db/queries/boats.sql
1-- sgl.query(ListBoats, many)
1-- sgl.query(ListBoats, many)
2SELECT name, home_port FROM boats ORDER BY name;
2SELECT id, name, home_port FROM boats ORDER BY id;

What sgl generates

AddBoatdiffers
CockroachDB
ClickHouse
1func (queries *Queries) AddBoat(ctx context.Context, params AddBoatParams) error
1func (queries *Queries) AddBoat(ctx context.Context, params AddBoatParams) error
2
2
3type AddBoatParams struct {
3type AddBoatParams struct {
4 ID uint64
4 Name string
5 Name string
5 HomePort string
6 HomePort string
6}
7}
ListBoatsdiffers
CockroachDB
ClickHouse
1func (queries *Queries) ListBoats(ctx context.Context) ([]ListBoatsRow, error)
1func (queries *Queries) ListBoats(ctx context.Context) ([]ListBoatsRow, error)
2
2
3type ListBoatsRow struct {
3type ListBoatsRow struct {
4 ID uint64 `json:"id"`
4 Name string `json:"name"`
5 Name string `json:"name"`
5 HomePort string `json:"home_port"`
6 HomePort string `json:"home_port"`
6}
7}

The program

example.godiffers
CockroachDB
ClickHouse
1package main
1package main
2
2
3import (
3import (
4 "context"
4 "context"
5 "database/sql"
5 "database/sql"
6 "errors"
7 "fmt"
6 "fmt"
8 "io"
7 "io"
9 "io/fs"
8 "io/fs"
10 "sync"
11
9
12 "seagull.sh/seagull/examples/cockroachdb/027_migrations_at_startup/internal/store"
10 "seagull.sh/seagull/examples/clickhouse/027_migrations_at_startup/internal/store"
13 "seagull.sh/seagull/sdk/migrate"
11 "seagull.sh/seagull/sdk/migrate"
14 "seagull.sh/seagull/sdk/migrate/migrate_backend_cockroachdb"
12 "seagull.sh/seagull/sdk/migrate/migrate_backend_clickhouse"
15)
13)
16
14
17// run deploys the program against an empty database the way an orchestrator does,
15// run starts the program twice against the same database. The first start applies both
18// starting several replicas at the same moment, each of which migrates before it works.
16// migrations before the program uses the schema; the second finds nothing to do.
19// Then one replica restarts and finds nothing to migrate.
20//
17//
21// Takes database (*sql.DB) which is the database to migrate, empty or not.
18// Takes database (*sql.DB) which is the database to migrate, empty or not.
22// Takes migrations (fs.FS) which holds db/migrations, the embed.FS in main.go.
19// Takes migrations (fs.FS) which holds db/migrations, the embed.FS in main.go.
23// Takes output (io.Writer) which receives what each start did.
20// Takes output (io.Writer) which receives what each start did.
24//
21//
25// Returns error when migrating or a query fails.
22// Returns error when migrating or a query fails.
26func run(ctx context.Context, database *sql.DB, migrations fs.FS, output io.Writer) error {
23func run(ctx context.Context, database *sql.DB, migrations fs.FS, output io.Writer) error {
27 replicas := []string{"Kittiwake", "Puffin", "Gannet"}
24 boats := []store.AddBoatParams{
28 applied, err := deploy(ctx, database, migrations, replicas)
25 {ID: 1, Name: "Kittiwake", HomePort: "Padstow"},
29 if err != nil {
26 {ID: 2, Name: "Puffin", HomePort: "Padstow"},
30 return err
31 }
32 fmt.Fprintf(output, "%d replicas started at once and applied %d migrations between them\n", len(replicas), applied)
33
34 restarted, err := start(ctx, database, migrations, "Cormorant")
35 if err != nil {
36 return err
37 }
38 fmt.Fprintf(output, "a replica restarted and applied %d migrations\n", restarted)
39
40 if err := printBoats(ctx, database, output); err != nil {
41 return err
42 }
27 }
43 if err := refuseAdvisoryLock(database, migrations, output); err != nil {
28 for _, boat := range boats {
44 return err
29 if err := start(ctx, database, migrations, boat, output); err != nil {
30 return err
31 }
45 }
32 }
46 return printHistory(ctx, database, migrations, output)
33 return printHistory(ctx, database, migrations, output)
47}
34}
48
35
49// deploy starts one replica per boat at the same moment and waits for them all.
50//
51// The replicas here share this process's connection pool, where real replicas are
52// separate processes with pools of their own. The lock lives in the database, so it
53// serialises them the same way.
54//
55// Takes database (*sql.DB) which is the database.
56// Takes migrations (fs.FS) which holds the migration files.
57// Takes boats ([]string) which names the boat each replica adds.
58//
59// Returns int which is how many migrations the replicas applied between them.
60// Returns error which joins every replica's failure.
61func deploy(ctx context.Context, database *sql.DB, migrations fs.FS, boats []string) (int, error) {
62 applied := make([]int, len(boats))
63 failures := make([]error, len(boats))
64 var replicas sync.WaitGroup
65 for index, boat := range boats {
66 replicas.Go(func() {
67 applied[index], failures[index] = start(ctx, database, migrations, boat)
68 })
69 }
70 replicas.Wait()
71
72 total := 0
73 for _, count := range applied {
74 total += count
75 }
76 return total, errors.Join(failures...)
77}
78
79// start is what the program does each time it starts: migrate first, then work. Here the
36// start is what the program does each time it starts: migrate first, then work. Here the
80// work is adding a boat.
37// work is adding a boat and listing the harbour.
81//
38//
82// Takes database (*sql.DB) which is the database.
39// Takes database (*sql.DB) which is the database.
83// Takes migrations (fs.FS) which holds the migration files.
40// Takes migrations (fs.FS) which holds the migration files.
84// Takes boat (string) which names the boat to add.
41// Takes boat (store.AddBoatParams) which is the boat to add.
42// Takes output (io.Writer) which receives what happened.
85//
43//
86// Returns int which is how many migrations this start applied.
44// Returns error when migrating or a query fails.
87// Returns error when migrating or the query fails.
45func start(ctx context.Context, database *sql.DB, migrations fs.FS, boat store.AddBoatParams, output io.Writer) error {
88func start(ctx context.Context, database *sql.DB, migrations fs.FS, boat string) (int, error) {
46 // migrate_backend_clickhouse.New() is the ClickHouse backend. It runs each statement of
89 // migrate_backend_cockroachdb.New() is the PostgreSQL backend with a lock table in place
47 // a migration without a transaction and records the migration once every statement has
90 // of an advisory lock, which CockroachDB does not have. Up creates sgl_migration_lock if
48 // succeeded; it takes no lock, so only one process should migrate at a time. NewMigrator
91 // it is missing and holds its one row with SELECT ... FOR UPDATE in an open transaction,
49 // reads the numbered *.up.sql and *.down.sql files from the directory inside migrations.
92 // so a second replica waits on that row until the first commits.
50 migrator, err := migrate.NewMigrator(database, migrate_backend_clickhouse.New(), migrations, "db/migrations")
93 //
94 // By default CockroachDB keeps a SERIALIZABLE transaction's FOR UPDATE locks in memory
95 // and drops them when a range splits or moves, which a schema change such as 0002's ADD
96 // COLUMN ... DEFAULT can cause. The backend turns on
97 // enable_durable_locking_for_serializable for its lock transaction, so the lock holds
98 // until the migrations are done; that needs CockroachDB 23.2 or later.
99 //
100 // The migrations run on a second connection, one statement at a time and not in a
101 // transaction: CockroachDB cannot roll a schema change back as part of a larger
102 // transaction, so the backend records each finished statement instead. A file that fails
103 // partway is marked dirty and resumes from the statement that failed once the cause is
104 // fixed. Keeping one schema change per file keeps each step its own record.
105 migrator, err := migrate.NewMigrator(database, migrate_backend_cockroachdb.New(), migrations, "db/migrations")
106 if err != nil {
51 if err != nil {
107 return 0, err
52 return err
108 }
53 }
109
54
110 // Up applies every migration not yet in the history and returns how many it applied. It
55 // Up applies every migration not yet in the history, in version order, and returns how
111 // takes the lock only when something is pending, and reads the history again once it
56 // many it applied. On a database that is up to date it applies none.
112 // holds the lock, so a replica that waited finds the work done.
113 applied, err := migrator.Up(ctx)
57 applied, err := migrator.Up(ctx)
114 if err != nil {
58 if err != nil {
115 return 0, fmt.Errorf("migrating: %w", err)
59 return fmt.Errorf("migrating: %w", err)
116 }
60 }
61 fmt.Fprintf(output, "start-up: applied %d migrations\n", applied)
117
62
118 queries := store.New(database)
63 queries := store.New(database)
119 if err := queries.AddBoat(ctx, store.AddBoatParams{Name: boat, HomePort: "Padstow"}); err != nil {
64 if err := queries.AddBoat(ctx, boat); err != nil {
120 return 0, fmt.Errorf("adding %s: %w", boat, err)
65 return fmt.Errorf("adding %s: %w", boat.Name, err)
121 }
66 }
122 return applied, nil
67 listed, err := queries.ListBoats(ctx)
123}
124
125// printBoats lists the harbour, which every replica wrote to once the schema existed.
126//
127// Takes database (*sql.DB) which is the migrated database.
128// Takes output (io.Writer) which receives the list.
129//
130// Returns error when the query fails.
131func printBoats(ctx context.Context, database *sql.DB, output io.Writer) error {
132 boats, err := store.New(database).ListBoats(ctx)
133 if err != nil {
68 if err != nil {
134 return fmt.Errorf("listing boats: %w", err)
69 return fmt.Errorf("listing boats: %w", err)
135 }
70 }
136 for _, row := range boats {
71 for _, row := range listed {
137 fmt.Fprintf(output, " %-9s of %s\n", row.Name, row.HomePort)
72 fmt.Fprintf(output, " %d %-9s of %s\n", row.ID, row.Name, row.HomePort)
138 }
139 return nil
140}
141
142// refuseAdvisoryLock shows that the CockroachDB backend rejects an advisory lock, as a
143// configuration copied from PostgreSQL might ask for, before it touches the database.
144//
145// Takes database (*sql.DB) which is the database.
146// Takes migrations (fs.FS) which holds the migration files.
147// Takes output (io.Writer) which receives the refusal.
148//
149// Returns error when the backend accepts the advisory lock.
150func refuseAdvisoryLock(database *sql.DB, migrations fs.FS, output io.Writer) error {
151 backend := migrate_backend_cockroachdb.New(migrate.WithLock(migrate.LockAdvisory))
152 _, err := migrate.NewMigrator(database, backend, migrations, "db/migrations")
153 if err == nil {
154 return errors.New("an advisory lock on CockroachDB gave no error, want migrate.ErrUnsupportedLock")
155 }
156 if !errors.Is(err, migrate.ErrUnsupportedLock) {
157 return fmt.Errorf("an advisory lock on CockroachDB gave %w, want migrate.ErrUnsupportedLock", err)
158 }
73 }
159 fmt.Fprintf(output, "advisory lock refused: %v\n", err)
160 return nil
74 return nil
161}
75}
162
76
163// printHistory prints the migration history the migrator keeps in the database, and the
77// printHistory prints the migration history the migrator keeps in the database.
164// table its lock uses.
165//
78//
166// Takes database (*sql.DB) which is the migrated database.
79// Takes database (*sql.DB) which is the migrated database.
167// Takes migrations (fs.FS) which holds the migration files.
80// Takes migrations (fs.FS) which holds the migration files.
168// Takes output (io.Writer) which receives the history.
81// Takes output (io.Writer) which receives the history.
169//
82//
170// Returns error when the history cannot be read.
83// Returns error when the history cannot be read.
171func printHistory(ctx context.Context, database *sql.DB, migrations fs.FS, output io.Writer) error {
84func printHistory(ctx context.Context, database *sql.DB, migrations fs.FS, output io.Writer) error {
172 backend := migrate_backend_cockroachdb.New()
85 backend := migrate_backend_clickhouse.New()
173 migrator, err := migrate.NewMigrator(database, backend, migrations, "db/migrations")
86 migrator, err := migrate.NewMigrator(database, backend, migrations, "db/migrations")
174 if err != nil {
87 if err != nil {
175 return err
88 return err
176 }
89 }
177 statuses, err := migrator.Status(ctx)
90 statuses, err := migrator.Status(ctx)
178 if err != nil {
91 if err != nil {
179 return fmt.Errorf("reading migration status: %w", err)
92 return fmt.Errorf("reading migration status: %w", err)
180 }
93 }
181 names := backend.Names()
94 fmt.Fprintf(output, "history in %s:\n", backend.Names().History)
182 fmt.Fprintf(output, "history in %s, locked through %s:\n", names.History, names.MigrationLock)
183 for _, status := range statuses {
95 for _, status := range statuses {
184 fmt.Fprintf(output, " %04d %-15s applied=%t\n", status.Version, status.Name, status.Applied)
96 fmt.Fprintf(output, " %04d %-15s applied=%t\n", status.Version, status.Name, status.Applied)
185 }
97 }
186 return nil
98 return nil
187}
99}