forked from pressly/goose
-
Notifications
You must be signed in to change notification settings - Fork 0
/
migration_sql.go
240 lines (199 loc) · 6.01 KB
/
migration_sql.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
package goose
import (
"bufio"
"bytes"
"database/sql"
"fmt"
"io"
"os"
"regexp"
"strings"
"sync"
"github.com/pkg/errors"
)
const sqlCmdPrefix = "-- +goose "
const scanBufSize = 4 * 1024 * 1024
var bufferPool = sync.Pool{
New: func() interface{} {
return make([]byte, scanBufSize)
},
}
// Checks the line to see if the line has a statement-ending semicolon
// or if the line contains a double-dash comment.
func endsWithSemicolon(line string) bool {
scanBuf := bufferPool.Get().([]byte)
defer bufferPool.Put(scanBuf)
prev := ""
scanner := bufio.NewScanner(strings.NewReader(line))
scanner.Buffer(scanBuf, scanBufSize)
scanner.Split(bufio.ScanWords)
for scanner.Scan() {
word := scanner.Text()
if strings.HasPrefix(word, "--") {
break
}
prev = word
}
return strings.HasSuffix(prev, ";")
}
// Split the given sql script into individual statements.
//
// The base case is to simply split on semicolons, as these
// naturally terminate a statement.
//
// However, more complex cases like pl/pgsql can have semicolons
// within a statement. For these cases, we provide the explicit annotations
// 'StatementBegin' and 'StatementEnd' to allow the script to
// tell us to ignore semicolons.
func getSQLStatements(r io.Reader, direction bool) ([]string, bool, error) {
var buf bytes.Buffer
scanBuf := bufferPool.Get().([]byte)
defer bufferPool.Put(scanBuf)
scanner := bufio.NewScanner(r)
scanner.Buffer(scanBuf, scanBufSize)
// track the count of each section
// so we can diagnose scripts with no annotations
upSections := 0
downSections := 0
statementEnded := false
ignoreSemicolons := false
directionIsActive := false
tx := true
stmts := []string{}
for scanner.Scan() {
line := scanner.Text()
// handle any goose-specific commands
if strings.HasPrefix(line, sqlCmdPrefix) {
cmd := strings.TrimSpace(line[len(sqlCmdPrefix):])
switch cmd {
case "Up":
directionIsActive = (direction == true)
upSections++
break
case "Down":
directionIsActive = (direction == false)
downSections++
break
case "StatementBegin":
if directionIsActive {
ignoreSemicolons = true
}
break
case "StatementEnd":
if directionIsActive {
statementEnded = (ignoreSemicolons == true)
ignoreSemicolons = false
}
break
case "NO TRANSACTION":
tx = false
break
}
}
if !directionIsActive {
continue
}
if _, err := buf.WriteString(line + "\n"); err != nil {
return nil, false, fmt.Errorf("io err: %v", err)
}
// Wrap up the two supported cases: 1) basic with semicolon; 2) psql statement
// Lines that end with semicolon that are in a statement block
// do not conclude statement.
if (!ignoreSemicolons && endsWithSemicolon(line)) || statementEnded {
statementEnded = false
stmts = append(stmts, buf.String())
buf.Reset()
}
}
if err := scanner.Err(); err != nil {
return nil, false, fmt.Errorf("scanning migration: %v", err)
}
// diagnose likely migration script errors
if ignoreSemicolons {
return nil, false, fmt.Errorf("parsing migration: saw '-- +goose StatementBegin' with no matching '-- +goose StatementEnd'")
}
if bufferRemaining := strings.TrimSpace(buf.String()); len(bufferRemaining) > 0 {
return nil, false, fmt.Errorf("parsing migration: unexpected unfinished SQL query: %s. potential missing semicolon", bufferRemaining)
}
if upSections == 0 && downSections == 0 {
return nil, false, fmt.Errorf("parsing migration: no Up/Down annotations found, so no statements were executed. See https://bitbucket.org/liamstask/goose/overview for details")
}
return stmts, tx, nil
}
// Run a migration specified in raw SQL.
//
// Sections of the script can be annotated with a special comment,
// starting with "-- +goose" to specify whether the section should
// be applied during an Up or Down migration
//
// All statements following an Up or Down directive are grouped together
// until another direction directive is found.
func runSQLMigration(db *sql.DB, sqlFile string, v int64, direction bool) error {
f, err := os.Open(sqlFile)
if err != nil {
return errors.Wrap(err, "failed to open SQL migration file")
}
defer f.Close()
statements, useTx, err := getSQLStatements(f, direction)
if err != nil {
return err
}
if useTx {
// TRANSACTION.
printInfo("Begin transaction\n")
tx, err := db.Begin()
if err != nil {
errors.Wrap(err, "failed to begin transaction")
}
for _, query := range statements {
printInfo("Executing statement: %s\n", clearStatement(query))
if _, err = tx.Exec(query); err != nil {
printInfo("Rollback transaction\n")
tx.Rollback()
return errors.Wrapf(err, "failed to execute SQL query %q", clearStatement(query))
}
}
if direction {
if _, err := tx.Exec(GetDialect().insertVersionSQL(), v, direction); err != nil {
printInfo("Rollback transaction\n")
tx.Rollback()
return errors.Wrap(err, "failed to insert new goose version")
}
} else {
if _, err := tx.Exec(GetDialect().deleteVersionSQL(), v); err != nil {
printInfo("Rollback transaction\n")
tx.Rollback()
return errors.Wrap(err, "failed to delete goose version")
}
}
printInfo("Commit transaction\n")
if err := tx.Commit(); err != nil {
return errors.Wrap(err, "failed to commit transaction")
}
return nil
}
// NO TRANSACTION.
for _, query := range statements {
printInfo("Executing statement: %s\n", clearStatement(query))
if _, err := db.Exec(query); err != nil {
return errors.Wrapf(err, "failed to execute SQL query %q", clearStatement(query))
}
}
if _, err := db.Exec(GetDialect().insertVersionSQL(), v, direction); err != nil {
return errors.Wrap(err, "failed to insert new goose version")
}
return nil
}
func printInfo(s string, args ...interface{}) {
if verbose {
log.Printf(s, args...)
}
}
var (
matchSQLComments = regexp.MustCompile(`(?m)^--.*$[\r\n]*`)
matchEmptyLines = regexp.MustCompile(`(?m)^$[\r\n]*`)
)
func clearStatement(s string) string {
s = matchSQLComments.ReplaceAllString(s, ``)
return matchEmptyLines.ReplaceAllString(s, ``)
}