Files
wehub-resource-sync a06f331eb8
CI / benchmark (push) Has been skipped
install-script / posix-syntax (push) Successful in 6m1s
CI / build-onnx (push) Failing after 6m43s
init-smoke / dry-run (push) Failing after 15m57s
security / govulncheck (push) Has been cancelled
security / trivy-fs (push) Has been cancelled
CI / test (1.26, ubuntu-latest) (push) Has been cancelled
Scorecard supply-chain security / Scorecard analysis (push) Has been cancelled
CI / test (1.26, macos-latest) (push) Has been cancelled
CI / build-windows (push) Has been cancelled
CI / lint (push) Has been cancelled
install-script / powershell-syntax (push) Has been cancelled
install-script / install (macos-14) (push) Has been cancelled
install-script / install (ubuntu-latest) (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 12:33:42 +08:00

2012 lines
67 KiB
Go

// Package store_sqlite is the on-disk, SQLite-backed implementation of
// graph.Store. It uses the pure-Go modernc.org/sqlite driver so the
// binary stays CGO-free on this code path, and satisfies the same
// conformance suite as the in-memory store (see
// internal/graph/storetest).
//
// Hot queries are precompiled as prepared statements in Open and
// closed in Close. Writes serialize through a single Go-side mutex
// because SQLite already serialises writers internally and an explicit
// mutex sidesteps SQLITE_BUSY contention when the conformance suite
// fans out 8 concurrent writers; reads still run concurrently under
// WAL mode.
//
// Meta maps are encoded as JSON (see meta_json.go); an empty / nil Meta
// is stored as NULL so the common case adds no row weight beyond the
// column header.
//
// EdgeIdentityRevisions is tracked in memory (atomic counter) -- it
// mirrors the in-memory store's monotonic "provenance churn" signal
// and does not need to survive process restarts (the in-memory store
// resets it on every New(), so the contract is per-process).
package store_sqlite
import (
"database/sql"
"errors"
"fmt"
"iter"
"runtime"
"strings"
"sync"
"sync/atomic"
"time"
"github.com/zzet/gortex/internal/graph"
_ "modernc.org/sqlite"
)
// Store is the SQLite-backed graph.Store implementation.
type Store struct {
db *sql.DB
// dbPath is the on-disk SQLite file path, retained for size
// telemetry — the WAL high-water mark surfaces in daemon_health so a
// runaway -wal is observable rather than silently filling the disk.
dbPath string
// writeMu serialises every mutation. SQLite serialises writers
// internally; doing the same on the Go side turns SQLITE_BUSY
// contention into clean lock-wait and keeps the conformance
// concurrency test predictable.
writeMu sync.Mutex
// resolveMu is the resolver-coordination mutex returned by
// ResolveMutex. Held by cross-repo / temporal / external resolver
// passes to keep their edge mutations from interleaving. Separate
// from writeMu so the resolver can hold it across multiple writes
// without blocking unrelated steady-state mutations.
resolveMu sync.Mutex
edgeIdentityRevs atomic.Int64
// wiped records that Open dropped an incompatible on-disk DB and
// recreated it empty (a schema-version mismatch that an in-place ALTER
// could not satisfy). Surfaced via NeedsRebuild so the daemon forces a
// full re-index on warm restart instead of an incremental reconcile,
// rather than relying on the side effect that a total wipe also empties
// file_mtimes.
wiped bool
// WAL-checkpoint loop lifecycle. In WAL mode a COMMIT only appends to
// the -wal file; pages move into the main DB (and the WAL becomes
// reusable) at a checkpoint. SQLite's default passive auto-checkpoint
// reuses the WAL in place and never shrinks the file, so under steady
// writes with ever-present readers (the pooled connections here, plus
// any other process holding the store open) the -wal ratchets up to a
// large high-water mark and stays there. runCheckpointLoop periodically
// runs `PRAGMA wal_checkpoint(TRUNCATE)` to drain the log into the DB
// and shrink the file back down. nil for in-memory stores (no WAL).
stopCheckpoint chan struct{} // closed by Close to stop the loop
checkpointDone chan struct{} // closed by the loop when it returns
stopOnce sync.Once // makes stopCheckpointLoop idempotent
// bundles is the content-addressed package-scoped cache over
// SearchSymbolBundles: a query serves cached Node + in/out edges for
// packages whose content fingerprint is unchanged and skips the node
// + edge fan-out for them. nil until SetBundleFingerprints is first
// called (the daemon wires it from the analysis pass); a nil cache
// makes SearchSymbolBundles fall through to the uncached path.
bundles *bundleCache
// memEst memoises AllRepoMemoryEstimates for a short TTL. That query
// is two COUNT(*) … GROUP BY repo scans; on a large graph under
// enrichment write load the pure-Go modernc sqlite count is
// pathologically slow, and the daemon status path can call it
// repeatedly. The TTL (and the mutex held across the recompute)
// collapses a burst of status polls onto a single scan; a few
// seconds of staleness is irrelevant for an advisory estimate.
memEstMu sync.Mutex
memEstVal map[string]graph.RepoMemoryEstimate
memEstAt time.Time
// Bulk-load fast path (graph.BulkLoader). Non-nil only between
// BeginBulkLoad and FlushBulk, and only on a first/empty cold index.
// database/sql PRAGMAs are connection-local, so the fast path pins one
// connection (bulkConn) carrying synchronous=OFF + an enlarged page
// cache and routes every bulk write through it; bulkPrevSync /
// bulkPrevCacheSize hold the values FlushBulk restores before the
// connection returns to the pool. All three are guarded by writeMu.
bulkConn *sql.Conn
bulkPrevSync int64
bulkPrevCacheSize int64
// Prepared statements (compiled once in Open, closed in Close).
stmtInsertNode *sql.Stmt
stmtGetNode *sql.Stmt
stmtGetNodeByQual *sql.Stmt
stmtFindByName *sql.Stmt
stmtFindByNameInRepo *sql.Stmt
stmtFileNodes *sql.Stmt
stmtRepoNodes *sql.Stmt
stmtAllNodes *sql.Stmt
stmtNodeCount *sql.Stmt
stmtRepoPrefixes *sql.Stmt
stmtRepoStatsNodes *sql.Stmt
stmtRepoStatsEdges *sql.Stmt
stmtRepoNodeCount *sql.Stmt
stmtRepoEdgeCount *sql.Stmt
stmtAllRepoCountsNodes *sql.Stmt
stmtAllRepoCountsEdges *sql.Stmt
stmtStatsByKind *sql.Stmt
stmtStatsByLanguage *sql.Stmt
stmtInsertEdge *sql.Stmt
stmtOutEdges *sql.Stmt
stmtOutEdgesLight *sql.Stmt
stmtInEdges *sql.Stmt
stmtRepoEdges *sql.Stmt
stmtAllEdges *sql.Stmt
stmtEdgeCount *sql.Stmt
stmtRemoveEdge *sql.Stmt
stmtUpdateEdgeOrigin *sql.Stmt
stmtUpdateEdgeAttrs *sql.Stmt
stmtSelectEdgeOrigin *sql.Stmt
stmtDeleteEdgeByKey *sql.Stmt
stmtEdgeExists *sql.Stmt
stmtSelectFileNodeIDs *sql.Stmt
stmtSelectRepoNodeIDs *sql.Stmt
stmtDeleteNodeByFile *sql.Stmt
stmtDeleteNodeByRepo *sql.Stmt
}
// Compile-time assertion: *Store satisfies graph.Store.
var _ graph.Store = (*Store)(nil)
// ResolveMutex returns the resolver-coordination mutex. Held by
// cross-repo / temporal / external resolver passes to serialise edge
// mutations. Separate from writeMu (which protects per-statement
// write serialisation against SQLITE_BUSY) so the resolver can hold
// it across multi-write batches without blocking unrelated steady-
// state mutations on the same store.
func (s *Store) ResolveMutex() *sync.Mutex { return &s.resolveMu }
// NeedsRebuild reports that Open dropped an incompatible on-disk database and
// recreated it empty, so the daemon's warm-restart path should force a full
// re-index (bypassing an incremental reconcile that would carry stale state)
// — see cmd/gortex.storeNeedsRebuild, the capability probe this satisfies.
func (s *Store) NeedsRebuild() bool { return s.wiped }
// Open opens (or creates) the SQLite database at path, runs the schema
// migration, and prepares hot statements. The DB is opened with WAL
// journaling and synchronous=NORMAL -- the same durability/throughput
// tradeoff every embedded-SQLite app uses for write-heavy workloads.
//
// Pass ":memory:" for an ephemeral in-process database (handy for
// tests when you don't need on-disk persistence).
//
// By default Open will NOT destroy an incompatible on-disk database: if the
// stored schema version requires a rebuild (a newer build's DB, or an older
// one crossing a rebuild migration) it returns ErrSchemaRebuildRequired and
// leaves the file untouched. Pass WithRebuild to permit the drop-and-recreate
// — only a caller that holds exclusive access to the store may do so (see
// WithRebuild).
func Open(path string, opts ...Option) (*Store, error) {
var o openOptions
for _, opt := range opts {
opt(&o)
}
return openWith(path, currentSchemaVersion, schemaMigrations, o.allowRebuild)
}
// Option configures Open.
type Option func(*openOptions)
type openOptions struct {
allowRebuild bool
}
// WithRebuild permits Open to drop and recreate an on-disk database whose
// schema version is incompatible (a newer build's, or an older one crossing a
// migration that an in-place ALTER cannot satisfy).
//
// The caller MUST hold exclusive cross-process access to the store file —
// removing a SQLite file another process has open silently splits its state.
// The daemon satisfies this: it takes an exclusive flock on <store>.lock for
// the writable on-disk sqlite lifecycle and passes this option only in that
// branch (see serverstack.NewSharedServer / OpenBackend). Without it, a wipe
// plan yields ErrSchemaRebuildRequired and the file is left intact, so a
// caller that does not hold the lock cannot corrupt a live store.
func WithRebuild() Option { return func(o *openOptions) { o.allowRebuild = true } }
// ErrSchemaRebuildRequired is returned by Open when an on-disk database needs a
// destructive rebuild but the caller did not pass WithRebuild (i.e. cannot
// prove it holds the store lock).
var ErrSchemaRebuildRequired = errors.New("store_sqlite: on-disk schema is incompatible and must be rebuilt; reopen with WithRebuild while holding the store lock")
// openWith is Open parameterised by the target schema version, migration
// registry, and rebuild permission so tests can drive the baseline / in-place
// / rebuild arms without mutating package globals. Open passes the package
// defaults (currentSchemaVersion, schemaMigrations) and the WithRebuild flag.
func openWith(path string, current int, migrations []schemaMigration, allowRebuild bool) (*Store, error) {
// Pragmas: WAL + synchronous=NORMAL is the standard write-heavy
// embedded tradeoff. cache_size(-32768) gives each pooled connection a
// 32 MiB page cache; temp_store(MEMORY) keeps GROUP BY / ORDER BY scratch
// off disk; mmap_size(256 MiB) lets reads fault pages straight from the
// OS page cache instead of copying through SQLite's. These materially
// speed the resolver/query phases on a large graph.
//
// journal_size_limit(64 MiB) caps the -wal high-water mark: after any
// checkpoint SQLite truncates the WAL back down to this size instead of
// leaving it at whatever it grew to. Without it the WAL only ratchets
// up (a passive checkpoint reuses the file in place, never shrinking
// it), which is how a 535 MB DB ends up with an 11 GB -wal. This bounds
// the file even between the explicit TRUNCATE checkpoints runCheckpointLoop
// issues, and even if that loop is not running.
dsn := path + "?_pragma=journal_mode(WAL)&_pragma=synchronous(NORMAL)&_pragma=busy_timeout(5000)&_pragma=foreign_keys(OFF)&_pragma=cache_size(-32768)&_pragma=temp_store(MEMORY)&_pragma=mmap_size(268435456)&_pragma=journal_size_limit(67108864)"
db, err := sql.Open("sqlite", dsn)
if err != nil {
return nil, fmt.Errorf("sqlite open: %w", err)
}
// Pool up to NumCPU connections so the resolver's parallel
// worker fan-out (NumCPU goroutines doing FindNodesByName /
// GetNode / GetOutEdges concurrently) doesn't serialise through
// a single connection — the dominant gap between the SQLite and
// bbolt backends on the bench's resolver stage was exactly that.
// SQLite's WAL mode allows concurrent readers across multiple
// connections; writes still serialise via writeMu on the Go
// side, then via SQLite's internal write lock. Every connection
// the pool opens picks up the journal-mode / synchronous /
// busy-timeout pragmas from the DSN above, so we don't need to
// pin one connection to "remember" them.
db.SetMaxOpenConns(runtime.NumCPU())
// Reconcile the on-disk schema version before applying schemaSQL. The graph
// store is a rebuildable cache, so an incompatible (older needing a rebuild
// step, or newer) DB is dropped and reindexed rather than migrated in place
// (see schema_version.go). The daemon holds an exclusive store.lock around
// Open, so wiping the file here cannot race another process.
stored, err := readUserVersion(db)
if err != nil {
_ = db.Close()
return nil, fmt.Errorf("sqlite read schema version: %w", err)
}
plan := planSchemaMigrationWith(stored, current, migrations)
didWipe := false
if plan.wipe && !isMemoryPath(path) {
// Refuse the destructive rebuild unless the caller proved it holds
// exclusive access (WithRebuild). This keeps the file safe even if a
// future caller reaches a wipe plan without the daemon's store lock.
if !allowRebuild {
_ = db.Close()
return nil, ErrSchemaRebuildRequired
}
if err := db.Close(); err != nil {
return nil, fmt.Errorf("sqlite close for rebuild: %w", err)
}
if err := removeStoreFiles(path); err != nil {
return nil, fmt.Errorf("sqlite rebuild: %w", err)
}
db, err = sql.Open("sqlite", dsn)
if err != nil {
return nil, fmt.Errorf("sqlite reopen for rebuild: %w", err)
}
db.SetMaxOpenConns(runtime.NumCPU())
didWipe = true
}
if _, err := db.Exec(schemaSQL); err != nil {
_ = db.Close()
return nil, fmt.Errorf("sqlite schema: %w", err)
}
// Add the promoted node columns to databases created before they
// existed (CREATE TABLE IF NOT EXISTS won't alter an existing table).
// Must run before the droppable-index loop below — nodes_semantic_pending
// references a promoted column — and before prepare(), whose node INSERT
// references them too.
if err := ensureNodeColumns(db); err != nil {
_ = db.Close()
return nil, fmt.Errorf("sqlite node columns: %w", err)
}
// nodes.is_stub generated column — see ensureNodeGeneratedColumns for why
// this is a separate function from ensureNodeColumns above.
if err := ensureNodeGeneratedColumns(db); err != nil {
_ = db.Close()
return nil, fmt.Errorf("sqlite node generated columns: %w", err)
}
// Same treatment for the edges table's is_unresolved generated column —
// must run before the droppable-index loop below, which creates an index
// over it.
if err := ensureEdgeColumns(db); err != nil {
_ = db.Close()
return nil, fmt.Errorf("sqlite edge columns: %w", err)
}
// Create the droppable secondary indexes from the shared set so their
// initial-creation DDL is byte-identical to the DDL the bulk-load fast
// path rebuilds them with (BeginBulkLoad drops these, FlushBulk
// recreates them — see bulk_load.go). Kept out of schemaSQL so the two
// sites cannot drift.
for _, idx := range bulkDroppableIndexes {
if _, err := db.Exec(idx.ddl); err != nil {
_ = db.Close()
return nil, fmt.Errorf("sqlite create index %s: %w", idx.name, err)
}
}
// edges_external is a partial index over exactly the external-call
// terminals, so ExternalCallCandidateEdges scans a tiny index instead
// of the full edges table. Built from the shared predicate const (not
// inlined in schemaSQL) so the index WHERE and the query WHERE stay
// byte-identical — SQLite only uses a partial index when the query's
// WHERE matches the index's.
if _, err := db.Exec(`CREATE INDEX IF NOT EXISTS edges_external ON edges(kind) WHERE ` + externalCallTargetPredicate); err != nil {
_ = db.Close()
return nil, fmt.Errorf("sqlite edges_external index: %w", err)
}
// Backfill the FTS rowid sidecar for databases built before it existed,
// so the first incremental UpsertSymbolFTS on an already-indexed symbol
// can do its O(log n) docid delete instead of leaking a duplicate row.
// One-time; a no-op once the map is populated or the FTS index is empty.
if err := backfillSymbolFTSRowidMap(db); err != nil {
_ = db.Close()
return nil, fmt.Errorf("sqlite fts rowid backfill: %w", err)
}
// Apply any in-place migration steps, then stamp the current schema version.
// Fresh and pre-versioning (stored==0) stores run the in-place steps too —
// they are idempotent and no-op on an empty or already-clean store — so the
// first in-place migration ships without forcing every non-daemon Open to
// pass WithRebuild. A wipe plan carries no in-place steps, and after a wipe
// the store is empty and the daemon's normal indexing repopulates it.
if plan.stamp {
if err := applyInPlaceMigrations(db, plan.inPlace); err != nil {
_ = db.Close()
return nil, fmt.Errorf("sqlite schema migrate: %w", err)
}
if err := setUserVersion(db, current); err != nil {
_ = db.Close()
return nil, fmt.Errorf("sqlite stamp schema version: %w", err)
}
}
s := &Store{db: db, dbPath: path, wiped: didWipe}
// Initialise the bundle cache at construction so its pointer is
// never written after Open — concurrent SearchSymbolBundles reads
// and SetBundleFingerprints writes then race only on the cache's
// own mutex-guarded maps, not on the Store field. The cache stays
// inert (every lookup a miss) until the daemon supplies fingerprints.
s.bundles = newBundleCache()
if err := s.prepare(); err != nil {
_ = db.Close()
return nil, fmt.Errorf("sqlite prepare: %w", err)
}
// In-memory databases have no WAL file to drain, so the periodic
// checkpoint is pointless there (and would leak a goroutine per
// short-lived test store). Only run it for on-disk stores.
if !strings.Contains(path, ":memory:") {
s.stopCheckpoint = make(chan struct{})
s.checkpointDone = make(chan struct{})
go s.runCheckpointLoop(walCheckpointInterval)
}
return s, nil
}
// walCheckpointInterval is how often runCheckpointLoop drains the WAL into
// the main DB and truncates the -wal file. Five minutes keeps the file
// bounded under steady writes without making the checkpoint itself a hot
// path; journal_size_limit in the DSN bounds growth between ticks.
const walCheckpointInterval = 5 * time.Minute
// runCheckpointLoop issues a TRUNCATE checkpoint every interval until Close
// stops it. Best-effort: a checkpoint that can't fully complete because a
// reader or writer holds the WAL just truncates what it can and retries on
// the next tick.
func (s *Store) runCheckpointLoop(interval time.Duration) {
defer close(s.checkpointDone)
ticker := time.NewTicker(interval)
defer ticker.Stop()
for {
select {
case <-s.stopCheckpoint:
return
case <-ticker.C:
_ = s.CheckpointWAL()
}
}
}
// CheckpointWAL runs `PRAGMA wal_checkpoint(TRUNCATE)`: it flushes the
// write-ahead log into the main database file and shrinks the -wal back to
// zero. A passive checkpoint (SQLite's default) only reuses the WAL in
// place and never reclaims the space; TRUNCATE is the mode that does.
// Exposed so a daemon shutdown path or an operator command can force a
// drain; the background loop calls it on a timer. Not held under writeMu —
// SQLite coordinates checkpoints against writers internally, and blocking
// steady-state writes on a maintenance op is the wrong tradeoff.
func (s *Store) CheckpointWAL() error {
_, err := s.db.Exec("PRAGMA wal_checkpoint(TRUNCATE)")
return err
}
// stopCheckpointLoop signals the background loop to exit and waits for it,
// so callers can be sure no checkpoint is in flight before closing s.db.
// Idempotent: safe to call from Close more than once.
func (s *Store) stopCheckpointLoop() {
s.stopOnce.Do(func() {
if s.stopCheckpoint != nil {
close(s.stopCheckpoint)
<-s.checkpointDone
}
})
}
// Close closes every prepared statement and the underlying *sql.DB. It
// first stops the WAL-checkpoint loop and issues one final TRUNCATE
// checkpoint so the -wal file is drained and shrunk on graceful shutdown
// rather than lingering at its high-water mark until the next open.
func (s *Store) Close() error {
s.stopCheckpointLoop()
if s.checkpointDone != nil { // on-disk store: drain the WAL one last time
_ = s.CheckpointWAL()
}
stmts := []*sql.Stmt{
s.stmtInsertNode, s.stmtGetNode, s.stmtGetNodeByQual,
s.stmtFindByName, s.stmtFindByNameInRepo,
s.stmtFileNodes, s.stmtRepoNodes,
s.stmtAllNodes, s.stmtNodeCount, s.stmtRepoPrefixes,
s.stmtRepoStatsNodes, s.stmtRepoStatsEdges,
s.stmtRepoNodeCount, s.stmtRepoEdgeCount,
s.stmtAllRepoCountsNodes, s.stmtAllRepoCountsEdges,
s.stmtStatsByKind, s.stmtStatsByLanguage,
s.stmtInsertEdge, s.stmtOutEdges, s.stmtOutEdgesLight, s.stmtInEdges,
s.stmtRepoEdges,
s.stmtAllEdges, s.stmtEdgeCount, s.stmtRemoveEdge,
s.stmtUpdateEdgeOrigin, s.stmtUpdateEdgeAttrs, s.stmtSelectEdgeOrigin, s.stmtDeleteEdgeByKey,
s.stmtEdgeExists,
s.stmtSelectFileNodeIDs, s.stmtSelectRepoNodeIDs,
s.stmtDeleteNodeByFile, s.stmtDeleteNodeByRepo,
}
for _, st := range stmts {
if st != nil {
_ = st.Close()
}
}
return s.db.Close()
}
func (s *Store) prepare() error {
var err error
prep := func(out **sql.Stmt, q string) {
if err != nil {
return
}
var st *sql.Stmt
st, err = s.db.Prepare(q)
if err != nil {
err = fmt.Errorf("prepare %q: %w", q, err)
return
}
*out = st
}
const nodeCols = lookupNodeCols
prep(&s.stmtInsertNode,
`INSERT OR REPLACE INTO nodes (`+nodeCols+`) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`)
prep(&s.stmtGetNode,
`SELECT `+nodeCols+` FROM nodes WHERE id = ?`)
prep(&s.stmtGetNodeByQual,
`SELECT `+nodeCols+` FROM nodes WHERE qual_name = ? LIMIT 1`)
prep(&s.stmtFindByName,
`SELECT `+nodeCols+` FROM nodes WHERE name = ?`)
prep(&s.stmtFindByNameInRepo,
`SELECT `+nodeCols+` FROM nodes WHERE name = ? AND repo_prefix = ?`)
prep(&s.stmtFileNodes,
`SELECT `+nodeCols+` FROM nodes WHERE file_path = ?`)
prep(&s.stmtRepoNodes,
`SELECT `+nodeCols+` FROM nodes WHERE repo_prefix = ?`)
prep(&s.stmtAllNodes,
`SELECT `+nodeCols+` FROM nodes`)
prep(&s.stmtNodeCount,
`SELECT COUNT(*) FROM nodes`)
prep(&s.stmtRepoPrefixes,
`SELECT DISTINCT repo_prefix FROM nodes WHERE repo_prefix <> ''`)
prep(&s.stmtRepoStatsNodes,
`SELECT repo_prefix, kind, language, COUNT(*) FROM nodes WHERE repo_prefix <> '' GROUP BY repo_prefix, kind, language`)
prep(&s.stmtRepoStatsEdges,
`SELECT n.repo_prefix, COUNT(*)
FROM edges e
JOIN nodes n ON n.id = e.from_id
WHERE n.repo_prefix <> ''
GROUP BY n.repo_prefix`)
prep(&s.stmtRepoNodeCount,
`SELECT COUNT(*) FROM nodes WHERE repo_prefix = ?`)
prep(&s.stmtRepoEdgeCount,
`SELECT COUNT(*)
FROM edges e
JOIN nodes n ON n.id = e.from_id
WHERE n.repo_prefix = ?`)
prep(&s.stmtAllRepoCountsNodes,
`SELECT repo_prefix, COUNT(*) FROM nodes WHERE repo_prefix <> '' GROUP BY repo_prefix`)
prep(&s.stmtAllRepoCountsEdges,
`SELECT n.repo_prefix, COUNT(*)
FROM edges e
JOIN nodes n ON n.id = e.from_id
WHERE n.repo_prefix <> ''
GROUP BY n.repo_prefix`)
prep(&s.stmtStatsByKind,
`SELECT kind, COUNT(*) FROM nodes GROUP BY kind`)
prep(&s.stmtStatsByLanguage,
`SELECT language, COUNT(*) FROM nodes GROUP BY language`)
const edgeCols = `from_id, to_id, kind, file_path, line, confidence, confidence_label, origin, tier, cross_repo, meta, resolve_terminal, resolve_terminal_reason`
prep(&s.stmtInsertEdge,
`INSERT OR IGNORE INTO edges (`+edgeCols+`) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)`)
prep(&s.stmtOutEdges,
`SELECT `+edgeCols+` FROM edges WHERE from_id = ?`)
// edgeColsLight is the package-level meta-less projection (store_light_edges.go),
// shared with AllEdgesLight so this prepared statement and the whole-graph scan
// can never drift apart.
prep(&s.stmtOutEdgesLight,
`SELECT `+edgeColsLight+` FROM edges WHERE from_id = ?`)
prep(&s.stmtInEdges,
`SELECT `+edgeCols+` FROM edges WHERE to_id = ?`)
prep(&s.stmtRepoEdges,
`SELECT e.from_id, e.to_id, e.kind, e.file_path, e.line,
e.confidence, e.confidence_label, e.origin, e.tier,
e.cross_repo, e.meta, e.resolve_terminal, e.resolve_terminal_reason
FROM edges e
JOIN nodes n ON n.id = e.from_id
WHERE n.repo_prefix = ?`)
prep(&s.stmtAllEdges,
`SELECT `+edgeCols+` FROM edges`)
prep(&s.stmtEdgeCount,
`SELECT COUNT(*) FROM edges`)
prep(&s.stmtRemoveEdge,
`DELETE FROM edges WHERE from_id = ? AND to_id = ? AND kind = ?`)
prep(&s.stmtSelectEdgeOrigin,
`SELECT origin FROM edges WHERE from_id = ? AND to_id = ? AND kind = ? AND file_path = ? AND line = ?`)
prep(&s.stmtUpdateEdgeOrigin,
`UPDATE edges SET origin = ?, tier = ? WHERE from_id = ? AND to_id = ? AND kind = ? AND file_path = ? AND line = ?`)
prep(&s.stmtUpdateEdgeAttrs,
`UPDATE edges SET confidence = ?, confidence_label = ?, origin = ?, tier = ?, meta = ?, resolve_terminal = ?, resolve_terminal_reason = ? WHERE from_id = ? AND to_id = ? AND kind = ? AND file_path = ? AND line = ?`)
prep(&s.stmtDeleteEdgeByKey,
`DELETE FROM edges WHERE from_id = ? AND to_id = ? AND kind = ? AND file_path = ? AND line = ?`)
prep(&s.stmtEdgeExists,
`SELECT 1 FROM edges WHERE from_id = ? AND to_id = ? AND kind = ? AND file_path = ? AND line = ? LIMIT 1`)
prep(&s.stmtSelectFileNodeIDs,
`SELECT id FROM nodes WHERE file_path = ?`)
prep(&s.stmtSelectRepoNodeIDs,
`SELECT id FROM nodes WHERE repo_prefix = ?`)
prep(&s.stmtDeleteNodeByFile,
`DELETE FROM nodes WHERE file_path = ?`)
prep(&s.stmtDeleteNodeByRepo,
`DELETE FROM nodes WHERE repo_prefix = ?`)
return err
}
// encodeMeta / decodeMeta live in meta_json.go (JSON codec + the
// metaWire typed DTO + the legacy-gob dual-read fallback).
// -- row scanners ---------------------------------------------------------
func scanNode(scanner interface {
Scan(...any) error
}) (*graph.Node, error) {
var (
n graph.Node
metaBlob []byte
p promotedNodeMeta
)
err := scanner.Scan(
&n.ID, &n.Kind, &n.Name, &n.QualName, &n.FilePath,
&n.StartLine, &n.EndLine, &n.StartColumn, &n.EndColumn, &n.Language,
&n.RepoPrefix, &n.WorkspaceID, &n.ProjectID,
&p.sig, &p.vis, &p.doc, &p.external, &p.returnType,
&p.isAsync, &p.isStatic, &p.isAbstract, &p.isExported, &p.updatedAt,
&p.dataClass, &p.semanticType, &p.semanticSource, &metaBlob,
)
if err != nil {
return nil, err
}
if len(metaBlob) > 0 {
m, derr := decodeMeta(metaBlob)
if derr != nil {
return nil, derr
}
n.Meta = m
}
// Restore the promoted columns into Meta. They are authoritative for
// rows written after the promotion; a NULL column (legacy gob rows)
// is left alone so the blob-carried value survives.
restorePromotedMeta(&n, p)
return &n, nil
}
// scanNodeLight scans the same columns as scanNode minus the trailing meta
// blob — no decodeMeta call, so no JSON/gob parse per row. Promoted columns
// still restore into Meta via restorePromotedMeta, so any caller that only
// reads a promoted key (signature, visibility, ..., semantic_type) sees the
// exact values scanNode would produce; only non-promoted content still
// living in the row's blob is absent. See graph.LightNodeReader: a node
// from this scan must never be round-tripped back through AddNode/AddBatch.
func scanNodeLight(scanner interface {
Scan(...any) error
}) (*graph.Node, error) {
var (
n graph.Node
p promotedNodeMeta
)
err := scanner.Scan(
&n.ID, &n.Kind, &n.Name, &n.QualName, &n.FilePath,
&n.StartLine, &n.EndLine, &n.StartColumn, &n.EndColumn, &n.Language,
&n.RepoPrefix, &n.WorkspaceID, &n.ProjectID,
&p.sig, &p.vis, &p.doc, &p.external, &p.returnType,
&p.isAsync, &p.isStatic, &p.isAbstract, &p.isExported, &p.updatedAt,
&p.dataClass, &p.semanticType, &p.semanticSource,
)
if err != nil {
return nil, err
}
restorePromotedMeta(&n, p)
return &n, nil
}
func scanEdge(scanner interface {
Scan(...any) error
}) (*graph.Edge, error) {
var (
e graph.Edge
metaBlob []byte
crossRepo int64
p promotedEdgeMeta
)
err := scanner.Scan(
&e.From, &e.To, &e.Kind, &e.FilePath, &e.Line,
&e.Confidence, &e.ConfidenceLabel, &e.Origin, &e.Tier,
&crossRepo, &metaBlob, &p.resolveTerminal, &p.resolveTerminalReason,
)
if err != nil {
return nil, err
}
e.CrossRepo = crossRepo != 0
if len(metaBlob) > 0 {
m, derr := decodeMeta(metaBlob)
if derr != nil {
return nil, derr
}
e.Meta = m
}
// Restore the promoted columns into Meta. They are authoritative for
// rows written after the promotion; a NULL column (pre-promotion rows)
// is left alone so any blob-carried value survives.
restorePromotedEdgeMeta(&e, p)
return &e, nil
}
// scanEdgeLight scans an edge WITHOUT decoding its meta blob -- for hot
// read paths (dataflow call-target lookup) that read only endpoints,
// kind, and line. Skipping the meta column avoids the JSON decode + map
// allocation that dominates large edge scans on this backend; the
// returned edge's Meta is nil.
func scanEdgeLight(scanner interface {
Scan(...any) error
}) (*graph.Edge, error) {
var (
e graph.Edge
crossRepo int64
)
err := scanner.Scan(
&e.From, &e.To, &e.Kind, &e.FilePath, &e.Line,
&e.Confidence, &e.ConfidenceLabel, &e.Origin, &e.Tier,
&crossRepo,
)
if err != nil {
return nil, err
}
e.CrossRepo = crossRepo != 0
return &e, nil
}
// -- writes ---------------------------------------------------------------
// AddNode inserts or replaces a node. Idempotent on the id column --
// re-adding the same id with new content does a last-write-wins
// update, matching the in-memory store's behaviour.
func (s *Store) AddNode(n *graph.Node) {
if n == nil || n.ID == "" {
return
}
// Cross-daemon proxy nodes are volatile remote-derived state and
// must never reach disk. The durable writer is the single gate —
// neither the resolver mint path nor the hydrator carries its own
// "don't persist" branch. A dropped proxy node is re-minted on
// demand after a restart.
if graph.IsProxyNode(n) {
return
}
s.writeMu.Lock()
defer s.writeMu.Unlock()
if err := s.insertNodeLocked(s.stmtInsertNode, n); err != nil {
// graph.Store.AddNode has no error channel; the in-memory
// store can't fail either. We swallow the error here for API
// parity; surface as a panic only on a clearly catastrophic
// failure (closed DB), not on a transient busy.
panicOnFatal(err)
}
}
func (s *Store) insertNodeLocked(stmt *sql.Stmt, n *graph.Node) error {
p, blobMeta := extractPromotedMeta(n.Meta)
metaBlob, err := encodeMeta(blobMeta)
if err != nil {
return err
}
_, err = stmt.Exec(
n.ID, string(n.Kind), n.Name, n.QualName, n.FilePath,
n.StartLine, n.EndLine, n.StartColumn, n.EndColumn, n.Language,
n.RepoPrefix, n.WorkspaceID, n.ProjectID,
p.sig, p.vis, p.doc, p.external, p.returnType,
p.isAsync, p.isStatic, p.isAbstract, p.isExported, p.updatedAt,
p.dataClass, p.semanticType, p.semanticSource, metaBlob,
)
return err
}
// AddEdge inserts an edge. Idempotent on the logical edge key (from,
// to, kind, file_path, line) -- a second AddEdge with the same key is
// a no-op (INSERT OR IGNORE), matching the in-memory store's "stored
// pointer replaced in place" semantics. Origin upgrades on a re-add
// are NOT applied through this path; use SetEdgeProvenance for that
// (matches the in-memory store: AddEdge replaces the *Edge pointer,
// but the conformance suite only verifies dedup-by-key, not pointer
// replacement, and the in-memory store also routes provenance
// upgrades through SetEdgeProvenance).
func (s *Store) AddEdge(e *graph.Edge) {
if e == nil {
return
}
// An edge to/from a cross-daemon proxy node is volatile and never
// persisted (the proxy node itself is dropped at AddNode).
if graph.IsProxyID(e.From) || graph.IsProxyID(e.To) {
return
}
s.writeMu.Lock()
defer s.writeMu.Unlock()
if err := s.insertEdgeLocked(s.stmtInsertEdge, e); err != nil {
panicOnFatal(err)
}
}
func (s *Store) insertEdgeLocked(stmt *sql.Stmt, e *graph.Edge) error {
p, blobMeta := extractPromotedEdgeMeta(e.Meta)
metaBlob, err := encodeMeta(blobMeta)
if err != nil {
return err
}
var crossRepo int64
if e.CrossRepo {
crossRepo = 1
}
_, err = stmt.Exec(
e.From, e.To, string(e.Kind), e.FilePath, e.Line,
e.Confidence, e.ConfidenceLabel, e.Origin, e.Tier,
crossRepo, metaBlob, p.resolveTerminal, p.resolveTerminalReason,
)
return err
}
// AddBatch inserts nodes and edges in a single transaction -- the
// 10-100x speedup vs per-statement commits at indexing scale.
func (s *Store) AddBatch(nodes []*graph.Node, edges []*graph.Edge) {
if len(nodes) == 0 && len(edges) == 0 {
return
}
s.writeMu.Lock()
defer s.writeMu.Unlock()
tx, err := s.beginWrite()
if err != nil {
panicOnFatal(err)
return
}
commit := false
defer func() {
if !commit {
_ = tx.Rollback()
}
}()
insertNode := tx.Stmt(s.stmtInsertNode)
defer insertNode.Close()
insertEdge := tx.Stmt(s.stmtInsertEdge)
defer insertEdge.Close()
for _, n := range nodes {
if n == nil || n.ID == "" {
continue
}
// Cross-daemon proxy nodes never reach disk.
if graph.IsProxyNode(n) {
continue
}
if err := s.insertNodeLocked(insertNode, n); err != nil {
panicOnFatal(err)
return
}
}
for _, e := range edges {
if e == nil {
continue
}
// An edge to or from a proxy node is volatile remote-derived
// state too; never persist it (it would dangle on reload since
// the proxy node itself is dropped).
if graph.IsProxyID(e.From) || graph.IsProxyID(e.To) {
continue
}
if err := s.insertEdgeLocked(insertEdge, e); err != nil {
panicOnFatal(err)
return
}
}
if err := tx.Commit(); err != nil {
panicOnFatal(err)
return
}
commit = true
}
// SetEdgeProvenance mutates an existing edge's origin in-place and
// bumps the identity-revision counter when the origin actually
// changes. Returns true iff a change was applied. Mirrors the
// in-memory store's "delete-then-insert of identity" semantics.
func (s *Store) SetEdgeProvenance(e *graph.Edge, newOrigin string) bool {
if e == nil {
return false
}
s.writeMu.Lock()
defer s.writeMu.Unlock()
// Look up the stored origin -- the caller-supplied *Edge may be a
// detached copy whose Origin already matches newOrigin even though
// the row still has the old value.
var storedOrigin string
row := s.stmtSelectEdgeOrigin.QueryRow(e.From, e.To, string(e.Kind), e.FilePath, e.Line)
if err := row.Scan(&storedOrigin); err != nil {
if errors.Is(err, sql.ErrNoRows) {
return false
}
panicOnFatal(err)
return false
}
if storedOrigin == newOrigin {
return false
}
newTier := e.Tier
if newTier != "" {
newTier = graph.ResolvedBy(newOrigin)
}
if _, err := s.stmtUpdateEdgeOrigin.Exec(newOrigin, newTier, e.From, e.To, string(e.Kind), e.FilePath, e.Line); err != nil {
panicOnFatal(err)
return false
}
// Reflect the change on the caller's struct, mirroring the
// in-memory store which mutates the in-graph *Edge in place.
e.Origin = newOrigin
if e.Tier != "" {
e.Tier = newTier
}
s.edgeIdentityRevs.Add(1)
return true
}
// PersistEdgeAttributes durably rewrites the mutable attribute columns
// (confidence, confidence_label, origin, tier, meta) of the edge row
// identified by e's full logical key. It is the disk-backend counterpart
// to the in-memory store's "mutate the live *Edge in place" behaviour: a
// pass that confirms an edge's full provenance bundle (not just origin)
// calls this so the confidence / label / meta survive a reload. A missing
// row is a silent no-op (UPDATE ... WHERE matches nothing).
func (s *Store) PersistEdgeAttributes(e *graph.Edge) {
if e == nil {
return
}
p, blobMeta := extractPromotedEdgeMeta(e.Meta)
metaBlob, err := encodeMeta(blobMeta)
if err != nil {
panicOnFatal(err)
return
}
s.writeMu.Lock()
defer s.writeMu.Unlock()
if _, err := s.stmtUpdateEdgeAttrs.Exec(
e.Confidence, e.ConfidenceLabel, e.Origin, e.Tier, metaBlob,
p.resolveTerminal, p.resolveTerminalReason,
e.From, e.To, string(e.Kind), e.FilePath, e.Line,
); err != nil {
panicOnFatal(err)
}
}
// Compile-time assertion: *Store satisfies the batched meta persister.
var _ graph.EdgeMetaBatchPersister = (*Store)(nil)
// PersistEdgeAttributesBatch is the batched form of PersistEdgeAttributes:
// it rewrites the mutable attribute columns (confidence, confidence_label,
// origin, tier, meta) for every edge in the batch, chunking the writes into
// reindexChunkSize-row transactions re-using one prepared statement. The
// resolver's terminal-skip stamping calls it to persist a Meta flag across a
// large slice of edges without paying a BEGIN/COMMIT per edge. A row with no
// matching key is a silent no-op (UPDATE ... WHERE matches nothing).
func (s *Store) PersistEdgeAttributesBatch(edges []*graph.Edge) {
if len(edges) == 0 {
return
}
s.writeMu.Lock()
defer s.writeMu.Unlock()
for i := 0; i < len(edges); i += reindexChunkSize {
end := minInt(i+reindexChunkSize, len(edges))
chunk := edges[i:end]
tx, err := s.db.Begin()
if err != nil {
panicOnFatal(err)
return
}
updStmt := tx.Stmt(s.stmtUpdateEdgeAttrs)
for _, e := range chunk {
if e == nil {
continue
}
p, blobMeta := extractPromotedEdgeMeta(e.Meta)
metaBlob, err := encodeMeta(blobMeta)
if err != nil {
_ = tx.Rollback()
panicOnFatal(err)
return
}
if _, err := updStmt.Exec(
e.Confidence, e.ConfidenceLabel, e.Origin, e.Tier, metaBlob,
p.resolveTerminal, p.resolveTerminalReason,
e.From, e.To, string(e.Kind), e.FilePath, e.Line,
); err != nil {
_ = tx.Rollback()
panicOnFatal(err)
return
}
}
if err := tx.Commit(); err != nil {
panicOnFatal(err)
return
}
}
}
// ReindexEdge updates the stored row after e.To has been mutated from
// oldTo to e.To. Implemented as delete-old + insert-new under the
// same write lock (SQLite's UNIQUE constraint on (from,to,kind,file,
// line) makes "UPDATE to_id" a one-shot, but the delete+insert form
// keeps semantics identical when the new (from,to,...) key happens to
// already exist -- the INSERT OR IGNORE drops the dup, just like the
// in-memory store's bucket-replace).
func (s *Store) ReindexEdge(e *graph.Edge, oldTo string) {
if e == nil || oldTo == e.To {
return
}
s.writeMu.Lock()
defer s.writeMu.Unlock()
if _, err := s.stmtDeleteEdgeByKey.Exec(e.From, oldTo, string(e.Kind), e.FilePath, e.Line); err != nil {
panicOnFatal(err)
return
}
if err := s.insertEdgeLocked(s.stmtInsertEdge, e); err != nil {
panicOnFatal(err)
return
}
}
// reindexChunkSize bounds the number of edge re-binds per BEGIN/COMMIT.
// Same shape as the bbolt sibling: large enough to amortise the
// per-tx overhead (BEGIN+COMMIT plus WAL fsync) but small enough that
// the WAL doesn't balloon and a crash mid-batch only loses ≤chunk
// mutations.
const reindexChunkSize = 5000
// ReindexEdges chunks the batch into reindexChunkSize-mutation
// transactions and runs each through prepared statements re-used
// across the chunk. Per-edge ReindexEdge was the resolver hot path
// (10k+ calls = 10k+ BEGIN/COMMIT pairs); this collapses them to two.
func (s *Store) ReindexEdges(batch []graph.EdgeReindex) {
if len(batch) == 0 {
return
}
s.writeMu.Lock()
defer s.writeMu.Unlock()
for i := 0; i < len(batch); i += reindexChunkSize {
end := minInt(i+reindexChunkSize, len(batch))
chunk := batch[i:end]
tx, err := s.db.Begin()
if err != nil {
panicOnFatal(err)
return
}
delStmt := tx.Stmt(s.stmtDeleteEdgeByKey)
insStmt := tx.Stmt(s.stmtInsertEdge)
for _, r := range chunk {
if r.Edge == nil || r.OldTo == r.Edge.To {
continue
}
if _, err := delStmt.Exec(r.Edge.From, r.OldTo, string(r.Edge.Kind), r.Edge.FilePath, r.Edge.Line); err != nil {
_ = tx.Rollback()
panicOnFatal(err)
return
}
if err := s.insertEdgeLocked(insStmt, r.Edge); err != nil {
_ = tx.Rollback()
panicOnFatal(err)
return
}
}
if err := tx.Commit(); err != nil {
panicOnFatal(err)
return
}
}
}
// SetEdgeProvenanceBatch chunks origin promotions into one BEGIN/
// COMMIT per chunk and bumps the in-process revision counter once
// per actual change, matching the per-edge SetEdgeProvenance's
// semantics. Returns the total number of edges whose Origin changed.
func (s *Store) SetEdgeProvenanceBatch(batch []graph.EdgeProvenanceUpdate) int {
if len(batch) == 0 {
return 0
}
s.writeMu.Lock()
defer s.writeMu.Unlock()
totalChanged := 0
for i := 0; i < len(batch); i += reindexChunkSize {
end := minInt(i+reindexChunkSize, len(batch))
chunk := batch[i:end]
tx, err := s.db.Begin()
if err != nil {
panicOnFatal(err)
return totalChanged
}
selStmt := tx.Stmt(s.stmtSelectEdgeOrigin)
updStmt := tx.Stmt(s.stmtUpdateEdgeOrigin)
chunkChanged := 0
for _, u := range chunk {
if u.Edge == nil {
continue
}
var storedOrigin string
row := selStmt.QueryRow(u.Edge.From, u.Edge.To, string(u.Edge.Kind), u.Edge.FilePath, u.Edge.Line)
if err := row.Scan(&storedOrigin); err != nil {
if errors.Is(err, sql.ErrNoRows) {
continue
}
_ = tx.Rollback()
panicOnFatal(err)
return totalChanged
}
if storedOrigin == u.NewOrigin {
continue
}
newTier := u.Edge.Tier
if newTier != "" {
newTier = graph.ResolvedBy(u.NewOrigin)
}
if _, err := updStmt.Exec(u.NewOrigin, newTier, u.Edge.From, u.Edge.To, string(u.Edge.Kind), u.Edge.FilePath, u.Edge.Line); err != nil {
_ = tx.Rollback()
panicOnFatal(err)
return totalChanged
}
u.Edge.Origin = u.NewOrigin
if u.Edge.Tier != "" {
u.Edge.Tier = newTier
}
chunkChanged++
}
if err := tx.Commit(); err != nil {
panicOnFatal(err)
return totalChanged
}
if chunkChanged > 0 {
s.edgeIdentityRevs.Add(int64(chunkChanged))
}
totalChanged += chunkChanged
}
return totalChanged
}
func minInt(a, b int) int {
if a < b {
return a
}
return b
}
// RemoveEdge deletes every edge between (from, to) with the given
// kind. Returns true iff at least one row was deleted.
func (s *Store) RemoveEdge(from, to string, kind graph.EdgeKind) bool {
s.writeMu.Lock()
defer s.writeMu.Unlock()
res, err := s.stmtRemoveEdge.Exec(from, to, string(kind))
if err != nil {
panicOnFatal(err)
return false
}
n, err := res.RowsAffected()
if err != nil {
panicOnFatal(err)
return false
}
return n > 0
}
// EvictFile removes every node anchored to filePath and every edge
// that touches one of those nodes. Returns (nodesRemoved,
// edgesRemoved).
func (s *Store) EvictFile(filePath string) (nodesRemoved, edgesRemoved int) {
s.writeMu.Lock()
defer s.writeMu.Unlock()
return s.evictByScopeLocked(s.stmtSelectFileNodeIDs, s.stmtDeleteNodeByFile, filePath)
}
// EvictRepo removes every node in repoPrefix and every edge that
// touches one. Returns (nodesRemoved, edgesRemoved).
func (s *Store) EvictRepo(repoPrefix string) (nodesRemoved, edgesRemoved int) {
s.writeMu.Lock()
defer s.writeMu.Unlock()
return s.evictByScopeLocked(s.stmtSelectRepoNodeIDs, s.stmtDeleteNodeByRepo, repoPrefix)
}
// evictByScopeLocked is the shared body of EvictFile / EvictRepo --
// collect the affected node IDs, delete every edge touching one of
// them, then delete the nodes themselves.
func (s *Store) evictByScopeLocked(selectIDs, deleteNodes *sql.Stmt, scope string) (int, int) {
rows, err := selectIDs.Query(scope)
if err != nil {
panicOnFatal(err)
return 0, 0
}
var ids []string
for rows.Next() {
var id string
if err := rows.Scan(&id); err != nil {
_ = rows.Close()
panicOnFatal(err)
return 0, 0
}
ids = append(ids, id)
}
if err := rows.Err(); err != nil {
_ = rows.Close()
panicOnFatal(err)
return 0, 0
}
_ = rows.Close()
if len(ids) == 0 {
return 0, 0
}
// Delete every edge touching one of these nodes. A DELETE-per-node
// (… WHERE from_id = ? OR to_id = ?) is one statement round-trip and
// WAL commit per node — hundreds of them when evicting a large file on
// the per-edit reindex path. Instead delete in chunked IN batches
// keyed on from_id then to_id, so each chunk is one index-driven
// DELETE; the chunk size stays under SQLite's bound-variable limit.
var edgesRemoved int
const evictEdgeChunk = 900
for _, col := range []string{"from_id", "to_id"} {
for start := 0; start < len(ids); start += evictEdgeChunk {
end := minInt(start+evictEdgeChunk, len(ids))
chunk := ids[start:end]
placeholders := strings.TrimSuffix(strings.Repeat("?,", len(chunk)), ",")
args := make([]any, len(chunk))
for i, id := range chunk {
args[i] = id
}
res, err := s.db.Exec(`DELETE FROM edges WHERE `+col+` IN (`+placeholders+`)`, args...)
if err != nil {
panicOnFatal(err)
return 0, edgesRemoved
}
if n, err := res.RowsAffected(); err == nil {
edgesRemoved += int(n)
}
}
}
res, err := deleteNodes.Exec(scope)
if err != nil {
panicOnFatal(err)
return 0, edgesRemoved
}
n, err := res.RowsAffected()
if err != nil {
panicOnFatal(err)
return 0, edgesRemoved
}
return int(n), edgesRemoved
}
// -- reads ---------------------------------------------------------------
func (s *Store) GetNode(id string) *graph.Node {
row := s.stmtGetNode.QueryRow(id)
n, err := scanNode(row)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return nil
}
panicOnFatal(err)
return nil
}
return n
}
func (s *Store) GetNodeByQualName(qualName string) *graph.Node {
if qualName == "" {
return nil
}
row := s.stmtGetNodeByQual.QueryRow(qualName)
n, err := scanNode(row)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return nil
}
panicOnFatal(err)
return nil
}
return n
}
func (s *Store) FindNodesByName(name string) []*graph.Node {
return s.queryNodes(s.stmtFindByName, name)
}
func (s *Store) FindNodesByNameInRepo(name, repoPrefix string) []*graph.Node {
return s.queryNodes(s.stmtFindByNameInRepo, name, repoPrefix)
}
func (s *Store) GetFileNodes(filePath string) []*graph.Node {
return s.queryNodes(s.stmtFileNodes, filePath)
}
func (s *Store) GetRepoNodes(repoPrefix string) []*graph.Node {
return s.queryNodes(s.stmtRepoNodes, repoPrefix)
}
func (s *Store) AllNodes() []*graph.Node {
return s.queryNodes(s.stmtAllNodes)
}
func (s *Store) queryNodes(stmt *sql.Stmt, args ...any) []*graph.Node {
rows, err := stmt.Query(args...)
if err != nil {
panicOnFatal(err)
return nil
}
defer rows.Close()
var out []*graph.Node
for rows.Next() {
n, err := scanNode(rows)
if err != nil {
panicOnFatal(err)
return out
}
out = append(out, n)
}
return out
}
// GetRepoNonContentNodes is the graph.NonContentNodeReader fast path: a
// SQL-level enumeration that drops CONTENT (data_class="content") section
// nodes, so the code-oriented passes never materialise a content-heavy
// repo's hundreds of thousands of sections. data_class is a promoted node
// column for rows written by the flat codec; legacy JSON rows (no column)
// fall back to json_extract, guarded by json_valid so the flat / gob blobs
// — which are not JSON — are skipped without error. The NULL-safe
// `IS NOT 'content'` keeps every node whose data_class is absent or carries
// any other value. An empty repoPrefix spans all repos.
func (s *Store) GetRepoNonContentNodes(repoPrefix string) []*graph.Node {
const filter = `COALESCE(data_class, CASE WHEN json_valid(CAST(meta AS TEXT)) THEN json_extract(CAST(meta AS TEXT), '$.data_class') END) IS NOT 'content'`
if repoPrefix == "" {
return s.scanNodeQuery(`SELECT ` + lookupNodeCols + ` FROM nodes WHERE ` + filter)
}
return s.scanNodeQuery(`SELECT `+lookupNodeCols+` FROM nodes WHERE repo_prefix = ? AND `+filter, repoPrefix)
}
// GetRepoNodesLight is the graph.LightNodeReader fast path: repo_prefix
// still uses the nodes_by_repo index, but the meta column is left out of
// the projection entirely, so a repo's already-enriched majority never
// crosses the driver boundary as a blob to decode. See LightNodeReader's
// doc for the read-only-use invariant this projection depends on.
func (s *Store) GetRepoNodesLight(repoPrefix string) []*graph.Node {
rows, err := s.db.Query(`SELECT `+lookupNodeColsLight+` FROM nodes WHERE repo_prefix = ?`, repoPrefix)
if err != nil {
panicOnFatal(err)
return nil
}
defer rows.Close()
var out []*graph.Node
for rows.Next() {
n, err := scanNodeLight(rows)
if err != nil {
panicOnFatal(err)
return out
}
out = append(out, n)
}
return out
}
// scanNodeQuery runs an ad-hoc node SELECT (columns = lookupNodeCols) and
// scans its rows into nodes — for the few non-hot enumerations that need a
// WHERE clause the prepared statements don't cover.
func (s *Store) scanNodeQuery(query string, args ...any) []*graph.Node {
rows, err := s.db.Query(query, args...)
if err != nil {
panicOnFatal(err)
return nil
}
defer rows.Close()
var out []*graph.Node
for rows.Next() {
n, err := scanNode(rows)
if err != nil {
panicOnFatal(err)
return out
}
out = append(out, n)
}
return out
}
func (s *Store) GetOutEdges(nodeID string) []*graph.Edge {
return s.queryEdges(s.stmtOutEdges, nodeID)
}
// EdgeExists reports whether an edge with exactly this identity is present --
// (from, to, kind, file_path, line) is the edges UNIQUE key, so this is a
// single indexed point lookup: no row decode, no Meta gob, no per-edge
// allocation, unlike GetOutEdges. The resolver's liveness guard
// (edgeStillLive) calls this once per applied edge on the cold/full pass; the
// difference from scanning + gob-decoding all of `from`'s out-edges is a
// dominant share of resolve cost on a large graph.
func (s *Store) EdgeExists(from, to string, kind graph.EdgeKind, filePath string, line int) bool {
var one int
err := s.stmtEdgeExists.QueryRow(from, to, string(kind), filePath, line).Scan(&one)
if err == sql.ErrNoRows {
return false
}
if err != nil {
panicOnFatal(err)
return false
}
return true
}
// GetOutEdgesLight returns a node's out-edges without decoding the
// per-edge Meta blob -- for hot dataflow lookups that need only
// endpoints/kind/line. The returned edges have a nil Meta.
func (s *Store) GetOutEdgesLight(nodeID string) []*graph.Edge {
return s.queryEdgesLight(s.stmtOutEdgesLight, nodeID)
}
func (s *Store) GetInEdges(nodeID string) []*graph.Edge {
return s.queryEdges(s.stmtInEdges, nodeID)
}
// GetOutEdgesForNodes fetches the out-edges of many nodes in one batched query
// (chunked) instead of a round-trip per node. The single-file resolve path
// walks every node of the edited file, which is an N+1 query storm on a disk
// backend; this collapses it to one query per chunk. Edges are grouped by
// their from_id; nodes with no out-edges are absent from the map.
func (s *Store) GetOutEdgesForNodes(ids []string) map[string][]*graph.Edge {
out := make(map[string][]*graph.Edge, len(ids))
if len(ids) == 0 {
return out
}
seen := make(map[string]struct{}, len(ids))
uniq := make([]string, 0, len(ids))
for _, id := range ids {
if id == "" {
continue
}
if _, ok := seen[id]; ok {
continue
}
seen[id] = struct{}{}
uniq = append(uniq, id)
}
for i := 0; i < len(uniq); i += lookupChunkSize {
end := minInt(i+lookupChunkSize, len(uniq))
chunk := uniq[i:end]
ph := make([]string, len(chunk))
args := make([]any, len(chunk))
for j, id := range chunk {
ph[j] = "?"
args[j] = id
}
q := `SELECT ` + lookupEdgeCols + ` FROM edges WHERE from_id IN (` + strings.Join(ph, ",") + `)`
for _, e := range s.queryEdgesSQL(q, args...) {
out[e.From] = append(out[e.From], e)
}
}
return out
}
func (s *Store) AllEdges() []*graph.Edge {
return s.queryEdges(s.stmtAllEdges)
}
// GetRepoEdges returns every edge whose source node has the given
// RepoPrefix. The pre-Store idiom — GetRepoNodes(r) followed by
// GetOutEdges(n.ID) per node — was O(repo_nodes) prepared-statement
// invocations, which on a multi-repo workspace dominated the
// per-repo extractor passes. A single JOIN over edges/nodes keyed
// on n.repo_prefix runs as one prepared statement and hits the
// existing repo_prefix index.
func (s *Store) GetRepoEdges(repoPrefix string) []*graph.Edge {
if repoPrefix == "" {
return nil
}
return s.queryEdges(s.stmtRepoEdges, repoPrefix)
}
func (s *Store) queryEdges(stmt *sql.Stmt, args ...any) []*graph.Edge {
rows, err := stmt.Query(args...)
if err != nil {
panicOnFatal(err)
return nil
}
defer rows.Close()
var out []*graph.Edge
for rows.Next() {
e, err := scanEdge(rows)
if err != nil {
panicOnFatal(err)
return out
}
out = append(out, e)
}
return out
}
// queryEdgesLight mirrors queryEdges but scans each row without its
// meta blob (scanEdgeLight), leaving Meta nil. Only for callers that
// never read edge Meta.
func (s *Store) queryEdgesLight(stmt *sql.Stmt, args ...any) []*graph.Edge {
rows, err := stmt.Query(args...)
if err != nil {
panicOnFatal(err)
return nil
}
defer rows.Close()
var out []*graph.Edge
for rows.Next() {
e, err := scanEdgeLight(rows)
if err != nil {
panicOnFatal(err)
return out
}
out = append(out, e)
}
return out
}
// -- counts and stats -----------------------------------------------------
func (s *Store) NodeCount() int {
var n int
if err := s.stmtNodeCount.QueryRow().Scan(&n); err != nil {
panicOnFatal(err)
return 0
}
return n
}
func (s *Store) EdgeCount() int {
var n int
if err := s.stmtEdgeCount.QueryRow().Scan(&n); err != nil {
panicOnFatal(err)
return 0
}
return n
}
func (s *Store) Stats() graph.GraphStats {
st := graph.GraphStats{
ByKind: map[string]int{},
ByLanguage: map[string]int{},
}
st.TotalNodes = s.NodeCount()
st.TotalEdges = s.EdgeCount()
rows, err := s.stmtStatsByKind.Query()
if err != nil {
panicOnFatal(err)
return st
}
for rows.Next() {
var kind string
var n int
if err := rows.Scan(&kind, &n); err != nil {
_ = rows.Close()
panicOnFatal(err)
return st
}
st.ByKind[kind] = n
}
_ = rows.Close()
rows, err = s.stmtStatsByLanguage.Query()
if err != nil {
panicOnFatal(err)
return st
}
for rows.Next() {
var lang string
var n int
if err := rows.Scan(&lang, &n); err != nil {
_ = rows.Close()
panicOnFatal(err)
return st
}
st.ByLanguage[lang] = n
}
_ = rows.Close()
return st
}
func (s *Store) RepoStats() map[string]graph.GraphStats {
out := map[string]graph.GraphStats{}
rows, err := s.stmtRepoStatsNodes.Query()
if err != nil {
panicOnFatal(err)
return out
}
for rows.Next() {
var repo, kind, lang string
var n int
if err := rows.Scan(&repo, &kind, &lang, &n); err != nil {
_ = rows.Close()
panicOnFatal(err)
return out
}
st, ok := out[repo]
if !ok {
st = graph.GraphStats{ByKind: map[string]int{}, ByLanguage: map[string]int{}}
}
st.TotalNodes += n
st.ByKind[kind] += n
st.ByLanguage[lang] += n
out[repo] = st
}
_ = rows.Close()
rows, err = s.stmtRepoStatsEdges.Query()
if err != nil {
panicOnFatal(err)
return out
}
for rows.Next() {
var repo string
var n int
if err := rows.Scan(&repo, &n); err != nil {
_ = rows.Close()
panicOnFatal(err)
return out
}
st, ok := out[repo]
if !ok {
st = graph.GraphStats{ByKind: map[string]int{}, ByLanguage: map[string]int{}}
}
st.TotalEdges = n
out[repo] = st
}
_ = rows.Close()
return out
}
func (s *Store) RepoPrefixes() []string {
rows, err := s.stmtRepoPrefixes.Query()
if err != nil {
panicOnFatal(err)
return nil
}
defer rows.Close()
var out []string
for rows.Next() {
var p string
if err := rows.Scan(&p); err != nil {
panicOnFatal(err)
return out
}
out = append(out, p)
}
return out
}
// -- provenance verification ---------------------------------------------
func (s *Store) EdgeIdentityRevisions() int {
return int(s.edgeIdentityRevs.Load())
}
// VerifyEdgeIdentities is a no-op for the SQL backend: the in-memory
// store's invariant is "the same *Edge pointer lives in both
// adjacency views". The SQL store has a single row per edge, so the
// invariant is trivially satisfied -- no walk can find a divergence
// to report.
func (s *Store) VerifyEdgeIdentities() error { return nil }
// -- memory estimation (advisory) ----------------------------------------
// perRowByteEstimate is a deliberately rough per-row byte cost --
// the disk backend doesn't have an in-memory footprint to report, so
// the contract (per Store interface comment) is "return what you can
// compute and callers treat the result as advisory". The conformance
// test only checks NodeCount.
const (
perNodeByteEstimate = 256
perEdgeByteEstimate = 128
)
func (s *Store) RepoMemoryEstimate(repoPrefix string) graph.RepoMemoryEstimate {
var est graph.RepoMemoryEstimate
var n, e int
if err := s.stmtRepoNodeCount.QueryRow(repoPrefix).Scan(&n); err != nil {
panicOnFatal(err)
return est
}
if err := s.stmtRepoEdgeCount.QueryRow(repoPrefix).Scan(&e); err != nil {
panicOnFatal(err)
return est
}
est.NodeCount = n
est.EdgeCount = e
est.NodeBytes = uint64(n) * perNodeByteEstimate
est.EdgeBytes = uint64(e) * perEdgeByteEstimate
return est
}
// memEstTTL bounds how long AllRepoMemoryEstimates serves a memoised
// result before recomputing. The estimate is advisory (status display),
// so a few seconds of staleness is fine, and the TTL keeps a burst of
// status polls from each triggering a full COUNT … GROUP BY scan.
const memEstTTL = 3 * time.Second
func cloneRepoMemEstimates(m map[string]graph.RepoMemoryEstimate) map[string]graph.RepoMemoryEstimate {
out := make(map[string]graph.RepoMemoryEstimate, len(m))
for k, v := range m {
out[k] = v
}
return out
}
func (s *Store) AllRepoMemoryEstimates() map[string]graph.RepoMemoryEstimate {
// Hold memEstMu across the recompute so a burst of concurrent status
// polls collapses onto one scan: the first caller computes and
// caches, the rest block briefly and then hit the fresh cache.
s.memEstMu.Lock()
defer s.memEstMu.Unlock()
if s.memEstVal != nil && time.Since(s.memEstAt) < memEstTTL {
return cloneRepoMemEstimates(s.memEstVal)
}
out := map[string]graph.RepoMemoryEstimate{}
rows, err := s.stmtAllRepoCountsNodes.Query()
if err != nil {
panicOnFatal(err)
return out
}
for rows.Next() {
var repo string
var n int
if err := rows.Scan(&repo, &n); err != nil {
_ = rows.Close()
panicOnFatal(err)
return out
}
est := out[repo]
est.NodeCount = n
est.NodeBytes = uint64(n) * perNodeByteEstimate
out[repo] = est
}
_ = rows.Close()
rows, err = s.stmtAllRepoCountsEdges.Query()
if err != nil {
panicOnFatal(err)
return out
}
for rows.Next() {
var repo string
var n int
if err := rows.Scan(&repo, &n); err != nil {
_ = rows.Close()
panicOnFatal(err)
return out
}
est := out[repo]
est.EdgeCount = n
est.EdgeBytes = uint64(n) * perEdgeByteEstimate
out[repo] = est
}
_ = rows.Close()
// Cache only on the full-success path — the early error returns above
// leave a partial `out` and must not poison the cache.
s.memEstVal = out
s.memEstAt = time.Now()
return cloneRepoMemEstimates(out)
}
// -- helpers --------------------------------------------------------------
// panicOnFatal turns truly catastrophic SQLite errors (closed DB,
// schema mismatch, disk-full at insert time) into a panic so callers
// see them, while letting expected sql.ErrNoRows / busy / no-affected
// callers stay quiet. The graph.Store interface deliberately does not
// surface errors -- it mirrors the in-memory store's "everything
// succeeds" contract -- so a fatal storage failure cannot be ignored.
//
// Caller contract: on a teardown-race error panicOnFatal RETURNS rather than
// panicking, so a caller that keeps using the query result after it returns
// MUST nil-check first. `rows, err := db.Query(...); panicOnFatal(err)` leaves
// rows == nil on a swallowed error, and the subsequent rows.Close() /
// rows.Next() would SIGSEGV — the aggregator reads early-return their empty
// value on nil rows for exactly this reason. In one line: fatal panics; a
// teardown-race read returns empty.
func panicOnFatal(err error) {
if err == nil {
return
}
if errors.Is(err, sql.ErrNoRows) {
return
}
// A closed statement / database / connection is a teardown race, not
// data corruption: Close() shuts the store (daemon shutdown, restart,
// or store swap) while an in-flight reader -- e.g. a deferred
// parallel-enrich goroutine still holding a cached *sql.Stmt -- runs a
// query. Crashing the whole daemon over a benign shutdown race is
// strictly worse than the read returning empty (or a winding-down write
// being dropped), so treat these as non-fatal.
if errors.Is(err, sql.ErrConnDone) || isStoreClosedErr(err) {
return
}
panic(fmt.Errorf("store_sqlite: %w", err))
}
// isStoreClosedErr reports whether err is the database/sql sentinel for a
// closed prepared statement or a closed database -- string-matched because
// database/sql does not export these as typed errors.
func isStoreClosedErr(err error) bool {
msg := err.Error()
return strings.Contains(msg, "statement is closed") ||
strings.Contains(msg, "database is closed")
}
// -- predicate-shaped reads ---------------------------------------------
//
// Each method runs one indexed SELECT and streams rows back via the
// iter.Seq[T] yield callback. Stops cleanly when yield returns false.
// Heavier than the equivalent bolt path (sql parsing + driver row
// materialisation) but cuts the resolver's wasted full-table scans
// down to "match-only" cardinality, which is the whole point.
// All three predicate iterators here MATERIALISE the query result
// into a slice before yielding, then iterate the slice. This avoids
// a deadlock peculiar to the SQLite backend's single-connection
// pool: a streaming rows-cursor holds THE connection, and any
// callback in the yield body that re-enters the store (e.g. GetNode
// to resolve an edge's caller) blocks forever waiting on the same
// connection. Materialise-then-yield releases the connection before
// the body runs, so re-entrant store calls work.
//
// The "predicate-shaped" win still holds: the indexed SELECT only
// fetches matching rows, not the whole table. We give up streaming
// memory savings (we still build a Go slice of *Edge / *Node) but
// keep the structural advantage that the row count flowing through
// scanEdge is proportional to the result, not the table.
// EdgesByKind: indexed SELECT on the (kind) column.
func (s *Store) EdgesByKind(kind graph.EdgeKind) iter.Seq[*graph.Edge] {
return func(yield func(*graph.Edge) bool) {
out := s.queryEdgesSQL(`
SELECT from_id, to_id, kind, file_path, line, confidence, confidence_label, origin, tier, cross_repo, meta, resolve_terminal, resolve_terminal_reason
FROM edges WHERE kind = ?`, string(kind))
for _, e := range out {
if !yield(e) {
return
}
}
}
}
// NodesByKind: indexed SELECT on the (kind) column.
func (s *Store) NodesByKind(kind graph.NodeKind) iter.Seq[*graph.Node] {
return func(yield func(*graph.Node) bool) {
out := s.queryNodesSQL(`SELECT `+lookupNodeCols+` FROM nodes WHERE kind = ?`, string(kind))
for _, n := range out {
if !yield(n) {
return
}
}
}
}
// EdgesWithUnresolvedTarget yields edges whose target is an unresolved stub
// in either form graph.IsUnresolvedTarget recognises. Filters on the
// is_unresolved generated column (see isUnresolvedColumnDDL) rather than
// re-deriving the to_id pattern match in SQL: measured 2.7x faster than the
// equivalent to_id-based OR query on a real 26-repo store (7.96s -> 2.95s for
// the same 847,684-row result) because the boolean index's bookmark lookups
// land in ascending rowid order, unlike a to_id-ordered index's.
//
// Gate-owned fn-value placeholders (graph.FnValuePlaceholderMarker,
// `unresolved::fnvalue::<name>`) are excluded on top of is_unresolved: the
// master resolver can never bind them, so they are pure pending-set bloat here
// (a live store held millions). The bare form is dropped by the range predicate
// — which rides edges_by_to(to_id) — using the ':;' range end from
// isUnresolvedColumnDDL's idiom (';' == ':'+1); the multi-repo COPY-rewrite form
// is dropped by the NOT LIKE, matching IsFnValuePlaceholder's infix shape.
func (s *Store) EdgesWithUnresolvedTarget() iter.Seq[*graph.Edge] {
return func(yield func(*graph.Edge) bool) {
out := s.queryEdgesSQL(`
SELECT from_id, to_id, kind, file_path, line, confidence, confidence_label, origin, tier, cross_repo, meta, resolve_terminal, resolve_terminal_reason
FROM edges WHERE is_unresolved = 1
AND NOT (to_id >= 'unresolved::fnvalue::' AND to_id < 'unresolved::fnvalue:;')
AND to_id NOT LIKE '%::unresolved::fnvalue::%'`)
for _, e := range out {
if !yield(e) {
return
}
}
}
}
// queryEdgesSQL runs an edge-shaped SELECT, materialises the rows
// into a slice, and closes the rows-cursor before returning —
// releasing the underlying sql.Conn so the predicate-iterator's
// callback body is free to make re-entrant store calls without
// deadlocking on the MaxOpenConns=1 pool. Companion to the existing
// queryEdges helper that takes a *sql.Stmt; this one takes a raw
// SQL string so the predicate iterators can pass inline queries.
func (s *Store) queryEdgesSQL(q string, args ...any) []*graph.Edge {
rows, err := s.db.Query(q, args...)
if err != nil {
return nil
}
defer rows.Close()
var out []*graph.Edge
for rows.Next() {
e, err := scanEdge(rows)
if err != nil || e == nil {
continue
}
out = append(out, e)
}
return out
}
// queryNodesSQL is the node-shaped sibling of queryEdgesSQL.
func (s *Store) queryNodesSQL(q string, args ...any) []*graph.Node {
rows, err := s.db.Query(q, args...)
if err != nil {
return nil
}
defer rows.Close()
var out []*graph.Node
for rows.Next() {
n, err := scanNode(rows)
if err != nil || n == nil {
continue
}
out = append(out, n)
}
return out
}
// lookupChunkSize bounds the IN-list parameter count per SQL query.
// SQLite's default SQLITE_MAX_VARIABLE_NUMBER is 32766 in modern
// builds, but staying well under that keeps query plans stable and
// avoids surprising the parser on monster lists.
const lookupChunkSize = 5000
// GetNodesByIDs collapses N per-id SELECTs into ⌈N/chunk⌉ queries
// of the form `SELECT … FROM nodes WHERE id IN (?, ?, …)`. The
// resolver fires hundreds of thousands of these on a large pass;
// chunking turns hundreds of seconds into single-digit seconds.
func (s *Store) GetNodesByIDs(ids []string) map[string]*graph.Node {
if len(ids) == 0 {
return nil
}
// Dedupe + skip empty up front to keep the chunk loop honest.
seen := make(map[string]struct{}, len(ids))
uniq := make([]string, 0, len(ids))
for _, id := range ids {
if id == "" {
continue
}
if _, ok := seen[id]; ok {
continue
}
seen[id] = struct{}{}
uniq = append(uniq, id)
}
out := make(map[string]*graph.Node, len(uniq))
const nodeCols = lookupNodeCols
for i := 0; i < len(uniq); i += lookupChunkSize {
end := minInt(i+lookupChunkSize, len(uniq))
chunk := uniq[i:end]
placeholders := strings.Repeat(",?", len(chunk))[1:]
q := `SELECT ` + nodeCols + ` FROM nodes WHERE id IN (` + placeholders + `)`
args := make([]any, len(chunk))
for j, id := range chunk {
args[j] = id
}
for _, n := range s.queryNodesSQL(q, args...) {
if n != nil {
out[n.ID] = n
}
}
}
return out
}
// FindNodesByNames collapses N per-name FindNodesByName queries into
// one `SELECT … FROM nodes WHERE name IN (…)` plus an in-Go bucket
// by name. The (name) index makes the SELECT seek-driven, and the
// caller sees the same map[name][]*Node it would have built by
// calling FindNodesByName N times.
func (s *Store) FindNodesByNames(names []string) map[string][]*graph.Node {
if len(names) == 0 {
return nil
}
seen := make(map[string]struct{}, len(names))
uniq := make([]string, 0, len(names))
for _, name := range names {
if name == "" {
continue
}
if _, ok := seen[name]; ok {
continue
}
seen[name] = struct{}{}
uniq = append(uniq, name)
}
out := make(map[string][]*graph.Node, len(uniq))
const nodeCols = lookupNodeCols
for i := 0; i < len(uniq); i += lookupChunkSize {
end := minInt(i+lookupChunkSize, len(uniq))
chunk := uniq[i:end]
placeholders := strings.Repeat(",?", len(chunk))[1:]
q := `SELECT ` + nodeCols + ` FROM nodes WHERE name IN (` + placeholders + `)`
args := make([]any, len(chunk))
for j, name := range chunk {
args[j] = name
}
for _, n := range s.queryNodesSQL(q, args...) {
if n == nil {
continue
}
out[n.Name] = append(out[n.Name], n)
}
}
return out
}
// -- BulkLoader implementation -------------------------------------------
// BeginBulkLoad / FlushBulk (the graph.BulkLoader bracket) live in
// bulk_load.go. The bracket exists so the indexer's in-memory shadow
// swap activates — the resolver and its post-resolve passes run against
// an in-memory *Graph at nanosecond latency, and the final drain dumps
// the resolved graph to sqlite in one shot. On a first/empty cold index
// the bracket additionally engages a bulk-persist fast path (dropped
// secondary indexes + synchronous=OFF on a pinned connection).