-
Notifications
You must be signed in to change notification settings - Fork 451
Expand file tree
/
Copy pathfile.go
More file actions
217 lines (202 loc) · 6.04 KB
/
file.go
File metadata and controls
217 lines (202 loc) · 6.04 KB
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
package migration
import (
"bytes"
"context"
"crypto/sha256"
"encoding/hex"
"fmt"
"io"
"io/fs"
"path/filepath"
"regexp"
"strings"
"github.com/go-errors/errors"
"github.com/jackc/pgconn"
"github.com/jackc/pgtype"
"github.com/jackc/pgx/v4"
"github.com/spf13/viper"
"github.com/supabase/cli/pkg/parser"
)
type MigrationFile struct {
Version string
Name string
Statements []string
}
var (
migrateFilePattern = regexp.MustCompile(`^([0-9]+)_(.*)\.sql$`)
typeNamePattern = regexp.MustCompile(`type "([^"]+)" does not exist`)
)
func NewMigrationFromFile(path string, fsys fs.FS) (*MigrationFile, error) {
lines, err := parseFile(path, fsys)
if err != nil {
return nil, err
}
file := MigrationFile{Statements: lines}
// Parse version from file name
filename := filepath.Base(path)
matches := migrateFilePattern.FindStringSubmatch(filename)
if len(matches) > 2 {
file.Version = matches[1]
file.Name = matches[2]
}
return &file, nil
}
func parseFile(path string, fsys fs.FS) ([]string, error) {
sql, err := fsys.Open(path)
if err != nil {
return nil, errors.Errorf("failed to open migration file: %w", err)
}
defer sql.Close()
// Unless explicitly specified, Use file length as max buffer size
if !viper.IsSet("SCANNER_BUFFER_SIZE") {
if fi, err := sql.Stat(); err == nil {
if size := int(fi.Size()); size > parser.MaxScannerCapacity {
parser.MaxScannerCapacity = size
}
}
}
return parser.SplitAndTrim(sql)
}
func NewMigrationFromReader(sql io.Reader) (*MigrationFile, error) {
lines, err := parser.SplitAndTrim(sql)
if err != nil {
return nil, err
}
return &MigrationFile{Statements: lines}, nil
}
func (m *MigrationFile) ExecBatch(ctx context.Context, conn *pgx.Conn) error {
// Batch migration commands, without using statement cache
batch := &pgconn.Batch{}
for _, line := range m.Statements {
batch.ExecParams(line, nil, nil, nil, nil)
}
// Insert into migration history
if len(m.Version) > 0 {
if err := m.insertVersionSQL(conn, batch); err != nil {
return err
}
}
// ExecBatch is implicitly transactional
if result, err := conn.PgConn().ExecBatch(ctx, batch).ReadAll(); err != nil {
// Defaults to printing the last statement on error
stat := INSERT_MIGRATION_VERSION
i := len(result)
if i < len(m.Statements) {
stat = m.Statements[i]
}
var msg []string
var pgErr *pgconn.PgError
if errors.As(err, &pgErr) {
stat = markError(stat, int(pgErr.Position))
if len(pgErr.Detail) > 0 {
msg = append(msg, pgErr.Detail)
}
// Provide helpful hint for extension type errors (SQLSTATE 42704: undefined_object)
if typeName := extractTypeName(pgErr.Message); len(typeName) > 0 && pgErr.Code == "42704" && !IsSchemaQualified(typeName) {
msg = append(msg, "")
msg = append(msg, "Hint: This type may be defined in a schema that's not in your search_path.")
msg = append(msg, " Use schema-qualified type references to avoid this error:")
msg = append(msg, fmt.Sprintf(" CREATE TABLE example (col extensions.%s);", typeName))
msg = append(msg, " Learn more: supabase migration new --help")
}
}
msg = append(msg, fmt.Sprintf("At statement: %d", i), stat)
return errors.Errorf("%w\n%s", err, strings.Join(msg, "\n"))
}
return nil
}
func markError(stat string, pos int) string {
lines := strings.Split(stat, "\n")
for j, r := range lines {
if c := len(r); pos > c {
pos -= c + 1
continue
}
// Show a caret below the error position
if pos > 0 {
caret := append(bytes.Repeat([]byte{' '}, pos-1), '^')
lines = append(lines[:j+1], string(caret))
}
break
}
return strings.Join(lines, "\n")
}
// extractTypeName extracts the type name from PostgreSQL error messages like:
// 'type "ltree" does not exist' -> "ltree"
func extractTypeName(errMsg string) string {
matches := typeNamePattern.FindStringSubmatch(errMsg)
if len(matches) > 1 {
return matches[1]
}
return ""
}
// IsSchemaQualified checks if a type name already contains a schema qualifier (e.g., "extensions.ltree")
func IsSchemaQualified(typeName string) bool {
return strings.Contains(typeName, ".")
}
func (m *MigrationFile) insertVersionSQL(conn *pgx.Conn, batch *pgconn.Batch) error {
value := pgtype.TextArray{}
if err := value.Set(m.Statements); err != nil {
return errors.Errorf("failed to set text array: %w", err)
}
ci := conn.ConnInfo()
var err error
var encoded []byte
var valueFormat int16
if conn.Config().PreferSimpleProtocol {
encoded, err = value.EncodeText(ci, encoded)
valueFormat = pgtype.TextFormatCode
} else {
encoded, err = value.EncodeBinary(ci, encoded)
valueFormat = pgtype.BinaryFormatCode
}
if err != nil {
return errors.Errorf("failed to encode binary: %w", err)
}
batch.ExecParams(
INSERT_MIGRATION_VERSION,
[][]byte{[]byte(m.Version), []byte(m.Name), encoded},
[]uint32{pgtype.TextOID, pgtype.TextOID, pgtype.TextArrayOID},
[]int16{pgtype.TextFormatCode, pgtype.TextFormatCode, valueFormat},
nil,
)
return nil
}
type SeedFile struct {
Path string
Hash string
Dirty bool `db:"-"`
}
func NewSeedFile(path string, fsys fs.FS) (*SeedFile, error) {
sql, err := fsys.Open(path)
if err != nil {
return nil, errors.Errorf("failed to open seed file: %w", err)
}
defer sql.Close()
hash := sha256.New()
if _, err := io.Copy(hash, sql); err != nil {
return nil, errors.Errorf("failed to hash file: %w", err)
}
digest := hex.EncodeToString(hash.Sum(nil))
return &SeedFile{Path: path, Hash: digest}, nil
}
func (m *SeedFile) ExecBatchWithCache(ctx context.Context, conn *pgx.Conn, fsys fs.FS) error {
// Parse each file individually to reduce memory usage
lines, err := parseFile(m.Path, fsys)
if err != nil {
return err
}
// Data statements don't mutate schemas, safe to use statement cache
batch := pgx.Batch{}
if !m.Dirty {
for _, line := range lines {
batch.Queue(line)
}
}
batch.Queue(UPSERT_SEED_FILE, m.Path, m.Hash)
// No need to track version here because there are no schema changes
if err := conn.SendBatch(ctx, &batch).Close(); err != nil {
return errors.Errorf("failed to send batch: %w", err)
}
return nil
}