-
Notifications
You must be signed in to change notification settings - Fork 0
/
schema.go
245 lines (209 loc) · 7.31 KB
/
schema.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
package update
import (
"context"
"database/sql"
"fmt"
"os"
"github.com/canonical/lxd/lxd/db/query"
"github.com/canonical/lxd/lxd/db/schema"
"github.com/canonical/lxd/shared"
)
// updateType represents whether the update is an internal or external schema update.
type updateType uint
const (
// updateInternal represents an internal schema update from microcluster.
updateInternal updateType = 0
// updateExternal represents a schema update external to microcluster, applied after all internal updates.
updateExternal updateType = 1
)
// SchemaUpdate holds the configuration for executing schema updates.
type SchemaUpdate struct {
updates map[updateType][]schema.Update // Ordered series of internal and external updates making up the schema
hook schema.Hook // Optional hook to execute whenever a update gets applied
fresh string // Optional SQL statement used to create schema from scratch
check schema.Check // Optional callback invoked before doing any update
path string // Optional path to a file containing extra queries to run
}
// Fresh sets a statement that will be used to create the schema from scratch
// when bootstraping an empty database. It should be a "flattening" of the
// available updates, generated using the Dump() method. If not given, all
// patches will be applied in order.
func (s *SchemaUpdate) Fresh(statement string) {
s.fresh = statement
}
// Check instructs the schema to invoke the given function whenever Ensure is
// invoked, before applying any due update. It can be used for aborting the
// operation.
func (s *SchemaUpdate) Check(check schema.Check) {
s.check = check
}
// Version returns the internal and external schema update versions, corresponding to the number of updates that have occurred.
func (s *SchemaUpdate) Version() (internalVersion uint64, externalVersion uint64) {
return uint64(len(s.updates[updateInternal])), uint64(len(s.updates[updateExternal]))
}
// Ensure makes sure that the actual schema in the given database matches the
// one defined by our updates.
//
// All updates are applied transactionally. In case any error occurs the
// transaction will be rolled back and the database will remain unchanged.
//
// A update will be applied only if it hasn't been before (currently applied
// updates are tracked in the a 'shema' table, which gets automatically
// created).
//
// If no error occurs, the integer returned by this method is the
// initial version that the schema has been upgraded from.
func (s *SchemaUpdate) Ensure(db *sql.DB) (int, error) {
var current int
aborted := false
err := query.Transaction(context.TODO(), db, func(ctx context.Context, tx *sql.Tx) error {
err := execFromFile(ctx, tx, s.path, s.hook)
if err != nil {
return fmt.Errorf("Failed to execute queries from %s: %w", s.path, err)
}
exists, err := doesSchemaTableExist(tx)
if err != nil {
return fmt.Errorf("Failed to check if schema table is there: %w", err)
}
var versions []int
if exists {
// updateFromV1 changes the schema table and needs to be run before we calculate the schema version.
err := updateFromV1(ctx, tx)
if err != nil {
return err
}
// maxVersionsStmt grabs the highest schema `version` column for each `type` (updateInternal/0) (updateExternal/1).
// The result is list of size 2, with index 0 corresponding to the max internal version and index 1 to the max external version, thanks to UNION ALL.
// The selected column must default to zero, otherwise query.SelectIntegers will fail to parse a null value as an integer.
maxVersionsStmt := "SELECT COALESCE(MAX(version), 0) FROM schemas WHERE type = 0 UNION ALL SELECT COALESCE(MAX(version), 0) FROM schemas WHERE type = 1"
versions, err = query.SelectIntegers(ctx, tx, maxVersionsStmt)
if err != nil {
return err
}
}
if s.check != nil {
err := s.check(ctx, current, tx)
if err == schema.ErrGracefulAbort {
// Abort the update gracefully, committing what
// we've done so far.
aborted = true
return nil
}
if err != nil {
return err
}
}
// When creating the schema from scratch, use the fresh dump if
// available. Otherwise just apply all relevant updates.
if versions == nil && s.fresh != "" {
_, err = tx.ExecContext(ctx, s.fresh)
if err != nil {
return fmt.Errorf("Cannot apply fresh schema: %w", err)
}
} else {
err = ensureUpdatesAreApplied(ctx, tx, versions, s.updates, s.hook)
if err != nil {
return err
}
}
return nil
})
if err != nil {
return -1, err
}
if aborted {
return current, schema.ErrGracefulAbort
}
return current, nil
}
// Apply any pending update that was not yet applied.
func ensureUpdatesAreApplied(ctx context.Context, tx *sql.Tx, versions []int, schemaUpdates map[updateType][]schema.Update, hook schema.Hook) error {
if versions == nil {
versions = []int{0, 0}
}
// Internal updates should be run before external ones.
updateOrder := []updateType{updateInternal, updateExternal}
for _, updateType := range updateOrder {
updates := schemaUpdates[updateType]
version := versions[updateType]
if version > len(updates) {
return fmt.Errorf("Schema version '%d' is more recent than expected '%d'", version, len(updates))
}
// If there are no updates, there's nothing to do.
if len(updates) == 0 {
return nil
}
// Apply missing updates.
for _, update := range updates[version:] {
if hook != nil {
err := hook(ctx, version, tx)
if err != nil {
return fmt.Errorf("Failed to execute hook (version %d): %w", version, err)
}
}
err := update(ctx, tx)
if err != nil {
return fmt.Errorf("Failed to apply update %d: %w", version, err)
}
if updateType == updateInternal && version == 0 {
err = updateFromV1(ctx, tx)
if err != nil {
return fmt.Errorf("Failed to apply special update 1: %w", err)
}
}
version++
statement := `INSERT INTO schemas (version, type, updated_at) VALUES (?, ?, strftime("%s"))`
_, err = tx.ExecContext(ctx, statement, version, updateType)
if err != nil {
return fmt.Errorf("Failed to insert version %d: %w", version, err)
}
}
}
return nil
}
// Read the given file (if it exists) and executes all queries it contains.
func execFromFile(ctx context.Context, tx *sql.Tx, path string, hook schema.Hook) error {
if !shared.PathExists(path) {
return nil
}
bytes, err := os.ReadFile(path)
if err != nil {
return fmt.Errorf("Failed to read file: %w", err)
}
if hook != nil {
err := hook(ctx, -1, tx)
if err != nil {
return fmt.Errorf("Failed to execute hook: %w", err)
}
}
_, err = tx.ExecContext(ctx, string(bytes))
if err != nil {
return err
}
err = os.Remove(path)
if err != nil {
return fmt.Errorf("Failed to remove file: %w", err)
}
return nil
}
// doesSchemaTableExist return whether the schema table is present in the
// database.
func doesSchemaTableExist(tx *sql.Tx) (bool, error) {
statement := `
SELECT COUNT(name) FROM sqlite_master WHERE type = 'table' AND name = 'schemas'
`
rows, err := tx.Query(statement)
if err != nil {
return false, err
}
defer func() { _ = rows.Close() }()
if !rows.Next() {
return false, fmt.Errorf("Schema table query returned no rows")
}
var count int
err = rows.Scan(&count)
if err != nil {
return false, err
}
return count == 1, nil
}