Skip to content

Commit 04a4250

Browse files
committed
compiler: analyze queries concurrently on the analysis core
parseQueries now reads and parses every file first, then analyzes the statements, then assembles the result. Splitting those phases lets the middle one run concurrently, and keeps the output deterministic: results land in a slice indexed by source position, so queries and errors come out in source order however they were produced. Only the analysis core runs concurrently. It reads the catalog and nothing else, and the catalog it reads is an immutable file, so its connections are no longer pinned to one. Every other path holds state a second goroutine would race — a database connection, the legacy catalog — and stays serial. Analyzing 200 queries against a warm catalog: 88ms -> 76ms per run on four cores, with the analysis phase itself going from about 35ms to 23ms. Co-Authored-By: Claude <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011MnoUabwBWW9gaEn2Nj7eG
1 parent b804f36 commit 04a4250

2 files changed

Lines changed: 99 additions & 49 deletions

File tree

internal/compiler/compile.go

Lines changed: 89 additions & 44 deletions
Original file line numberDiff line numberDiff line change
@@ -7,8 +7,11 @@ import (
77
"io"
88
"os"
99
"path/filepath"
10+
"runtime"
1011
"strings"
1112

13+
"golang.org/x/sync/errgroup"
14+
1215
"github.com/sqlc-dev/sqlc/internal/core"
1316
coreschema "github.com/sqlc-dev/sqlc/internal/core/schema"
1417
"github.com/sqlc-dev/sqlc/internal/migrations"
@@ -150,6 +153,15 @@ func (c *Compiler) parseCatalogCore(files []schemaFile, merr *multierr.Error) er
150153
return nil
151154
}
152155

156+
// statement is a parsed query awaiting analysis, with what error reporting
157+
// needs to place it back in the file it came from.
158+
type statement struct {
159+
filename string
160+
src string
161+
pp *preprocess.Result
162+
raw *ast.RawStmt
163+
}
164+
153165
func (c *Compiler) parseQueries(o opts.Parser) (*Result, error) {
154166
ctx := context.Background()
155167

@@ -160,13 +172,13 @@ func (c *Compiler) parseQueries(o opts.Parser) (*Result, error) {
160172
}
161173
}
162174

163-
var q []*Query
164175
merr := multierr.New()
165-
set := map[string]struct{}{}
166176
files, err := sqlpath.Glob(c.conf.Queries)
167177
if err != nil {
168178
return nil, err
169179
}
180+
181+
var stmts []statement
170182
for _, filename := range files {
171183
blob, err := os.ReadFile(filename)
172184
if err != nil {
@@ -179,49 +191,58 @@ func (c *Compiler) parseQueries(o opts.Parser) (*Result, error) {
179191
// engine parser ever sees the query, so parsers only handle SQL.
180192
pp := preprocess.File(c.conf.Engine, src)
181193

182-
stmts, err := c.parser.Parse(strings.NewReader(pp.Text))
194+
parsed, err := c.parser.Parse(strings.NewReader(pp.Text))
183195
if err != nil {
184196
if reported := addSyntaxErrors(merr, filename, src, pp); !reported {
185197
merr.Add(filename, src, 0, err)
186198
}
187199
continue
188200
}
189-
for _, stmt := range stmts {
190-
query, err := c.parseQuery(stmt.Raw, pp, o)
191-
if err != nil {
192-
var e *sqlerr.Error
193-
loc := stmt.Raw.Pos()
194-
if errors.As(err, &e) && e.Location != 0 {
195-
loc = e.Location
196-
}
197-
// Locations are reported against the rewritten query; map them
198-
// back so errors point at what the user wrote.
199-
loc = pp.Origin(loc)
200-
if e != nil && e.Location != 0 {
201-
e.Location = loc
202-
}
203-
merr.Add(filename, src, loc, err)
204-
// If this rpc unauthenticated error bubbles up, then all future parsing/analysis will fail
205-
if errors.Is(err, rpc.ErrUnauthenticated) {
206-
return nil, merr
207-
}
208-
continue
201+
for _, stmt := range parsed {
202+
stmts = append(stmts, statement{filename: filename, src: src, pp: pp, raw: stmt.Raw})
203+
}
204+
}
205+
206+
queries, errs := c.analyzeStatements(stmts, o)
207+
208+
var q []*Query
209+
set := map[string]struct{}{}
210+
for i, stmt := range stmts {
211+
if err := errs[i]; err != nil {
212+
var e *sqlerr.Error
213+
loc := stmt.raw.Pos()
214+
if errors.As(err, &e) && e.Location != 0 {
215+
loc = e.Location
209216
}
210-
if query == nil {
211-
continue
217+
// Locations are reported against the rewritten query; map them
218+
// back so errors point at what the user wrote.
219+
loc = stmt.pp.Origin(loc)
220+
if e != nil && e.Location != 0 {
221+
e.Location = loc
212222
}
213-
query.Metadata.Filename = filepath.Base(filename)
214-
queryName := query.Metadata.Name
215-
if queryName != "" {
216-
if _, exists := set[queryName]; exists {
217-
merr.Add(filename, src, pp.Origin(stmt.Raw.Pos()), fmt.Errorf("duplicate query name: %s", queryName))
218-
continue
219-
}
220-
set[queryName] = struct{}{}
223+
merr.Add(stmt.filename, stmt.src, loc, err)
224+
// If this rpc unauthenticated error bubbles up, then all future parsing/analysis will fail
225+
if errors.Is(err, rpc.ErrUnauthenticated) {
226+
return nil, merr
221227
}
222-
q = append(q, query)
228+
continue
223229
}
230+
query := queries[i]
231+
if query == nil {
232+
continue
233+
}
234+
query.Metadata.Filename = filepath.Base(stmt.filename)
235+
queryName := query.Metadata.Name
236+
if queryName != "" {
237+
if _, exists := set[queryName]; exists {
238+
merr.Add(stmt.filename, stmt.src, stmt.pp.Origin(stmt.raw.Pos()), fmt.Errorf("duplicate query name: %s", queryName))
239+
continue
240+
}
241+
set[queryName] = struct{}{}
242+
}
243+
q = append(q, query)
224244
}
245+
225246
if len(merr.Errs()) > 0 {
226247
return nil, merr
227248
}
@@ -235,17 +256,41 @@ func (c *Compiler) parseQueries(o opts.Parser) (*Result, error) {
235256
}, nil
236257
}
237258

238-
// addSyntaxErrors reports every sqlc syntax error the preprocessor recorded
239-
// for a file, in source order, and says whether there were any. Locations come
240-
// back in the rewritten text's coordinates, so they are mapped through Origin
241-
// to point at what the user wrote.
259+
// analyzeStatements analyzes each statement, returning the results in the
260+
// order the statements were given so that queries and errors come out in
261+
// source order however they were produced.
242262
//
243-
// A statement whose sqlc syntax did not validate is copied through for the
244-
// engine to parse, which assumes the engine can parse it. SQLite cannot: it
245-
// has no schema-qualified function call, so a bad sqlc.arg() is a syntax error
246-
// there rather than a call the preprocessor's message can be attached to.
247-
// These messages name the cause, so they are reported in place of the failure
248-
// they produced; anything else wrong with the file surfaces on the next run.
263+
// The analysis core reads the catalog and nothing else, so its statements are
264+
// analyzed concurrently. Every other path holds state that a second goroutine
265+
// would race — a database connection, the legacy catalog — and stays serial.
266+
func (c *Compiler) analyzeStatements(stmts []statement, o opts.Parser) ([]*Query, []error) {
267+
queries := make([]*Query, len(stmts))
268+
errs := make([]error, len(stmts))
269+
270+
analyze := func(i int) {
271+
queries[i], errs[i] = c.parseQuery(stmts[i].raw, stmts[i].pp, o)
272+
}
273+
274+
workers := runtime.GOMAXPROCS(0)
275+
if !c.coreAnalysis || workers < 2 || len(stmts) < 2 {
276+
for i := range stmts {
277+
analyze(i)
278+
}
279+
return queries, errs
280+
}
281+
282+
var g errgroup.Group
283+
g.SetLimit(workers)
284+
for i := range stmts {
285+
g.Go(func() error {
286+
analyze(i)
287+
return nil
288+
})
289+
}
290+
g.Wait()
291+
return queries, errs
292+
}
293+
249294
func addSyntaxErrors(merr *multierr.Error, filename, src string, pp *preprocess.Result) bool {
250295
var found bool
251296
for _, stmt := range pp.Statements() {

internal/core/catalog.go

Lines changed: 10 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import (
44
"context"
55
"database/sql"
66
"fmt"
7+
"runtime"
78

89
"github.com/sqlc-dev/sqlc/internal/core/catalogdb"
910
"github.com/sqlc-dev/sqlc/internal/core/catalogdef"
@@ -73,14 +74,18 @@ func openFile(path string) (*sql.DB, error) {
7374
if err != nil {
7475
return nil, fmt.Errorf("core: open catalog %s: %w", path, err)
7576
}
76-
pinPool(db)
77+
// An immutable file has no locking to contend for, so queries analyzed
78+
// concurrently each get their own connection rather than queueing on one.
79+
db.SetMaxOpenConns(runtime.GOMAXPROCS(0))
80+
db.SetMaxIdleConns(runtime.GOMAXPROCS(0))
81+
db.SetConnMaxIdleTime(0)
82+
db.SetConnMaxLifetime(0)
7783
return db, nil
7884
}
7985

80-
// pinPool holds the catalog to a single connection that is never retired.
81-
// Every ":memory:" connection is its own empty database, so a second one would
82-
// see a catalog with no tables in it; a catalog read from a file has no such
83-
// constraint, but has no use for more connections either.
86+
// pinPool holds a catalog being built to a single connection that is never
87+
// retired: every ":memory:" connection is its own empty database, so a second
88+
// one would see a catalog with no tables in it.
8489
func pinPool(db *sql.DB) {
8590
db.SetMaxOpenConns(1)
8691
db.SetMaxIdleConns(1)

0 commit comments

Comments
 (0)