perf(db): port upstream bounded ANALYZE planner statistics (issue #100)

cmd/ingestor/config.go

@@ -203,6 +203,23 @@ func (c *Config) IncrementalVacuumPages() int {
 	return 1024
 }
 
+// AnalysisLimit returns the per-index row cap for the planner stats refresh
+// (#2058). A negative setting disables the refresh; zero means unset, matching
+// IncrementalVacuumPages above.
+//
+// 10000 rather than the 400 SQLite's documentation offers: measured on the
+// 9.4 GB staging database, 400 and 1000 leave the channel-query plan exactly as
+// it was, 10000 produces the same plan as an unbounded ANALYZE, and it costs 2.0s
+// against that ANALYZE's 242.9s, both timed warm. The full ladder, and the cold
+// figure that matters at startup, are in Store.RefreshPlannerStats and
+// Store.EnsurePlannerStats.
+func (c *Config) AnalysisLimit() int {
+	if c.DB != nil && c.DB.AnalysisLimit != 0 {
+		return c.DB.AnalysisLimit
+	}
+	return 10000
+}
+
 // ShouldValidateSignatures returns true (default) unless explicitly disabled.
 func (c *Config) ShouldValidateSignatures() bool {
 	if c.ValidateSignatures != nil {

cmd/ingestor/db.go

@@ -1707,6 +1707,126 @@ func (s *Store) RunIncrementalVacuum(pages int) {
 	}
 }
 
+// RefreshPlannerStats rebuilds the query planner's cardinality statistics
+// (#2058).
+//
+// Without a sqlite_stat1 table the planner works from built-in guesses, and on
+// the channel queries it guesses wrong: it drives from the plain
+// idx_transmissions_payload_type rather than idx_tx_channel_hash, the partial
+// index (WHERE payload_type = 5) this schema already carries for that exact
+// filter.
+//
+// Measured on the 9.4 GB staging database (1,250,489 transmissions, 14,169,329
+// observations), region-filtered GetChannels, counting page-cache misses
+// because wall time there is dominated by the OS page cache (56.7s cold, 0.80s
+// warm, for the same query and plan):
+//
+//	analysis_limit   ANALYZE    driving index                     page misses
+//	none (no stats)  -          idx_transmissions_payload_type     143,442
+//	400              171ms      idx_transmissions_payload_type     143,449
+//	1000             171ms      idx_transmissions_payload_type     143,450
+//	10000            2.0s*      idx_tx_channel_hash                107,429
+//	0 (unbounded)    242.9s     idx_tx_channel_hash                107,429
+//
+// (*) Every ANALYZE duration in that table was timed on a warm page cache, one
+// after another. The same statement at limit 10000 took 3m43.9s cold, on a
+// freshly started container. See EnsurePlannerStats.
+//
+// So 10000 buys the whole plan change, and the four-minute unbounded ANALYZE
+// buys nothing beyond it. 400, the value SQLite's documentation offers for the
+// bounded form, changes nothing at all on this data: it samples too few rows to
+// separate the 126,336-row partial index from the 920,700-row plain one.
+//
+// ANALYZE, not PRAGMA optimize. Measured on the same database: optimize is a
+// no-op here, because it only analyzes tables that the calling connection has
+// itself queried during the session, and a maintenance call has queried none.
+// PRAGMA optimize(0x03) returned no statements and sqlite_stat1 was not
+// created.
+//
+// Note that analysis_limit=0 means *no* limit to SQLite, not "use a default".
+// Config.AnalysisLimit maps an unset config to 10000 for that reason, and a
+// negative value here disables the refresh.
+//
+// Returns whether the statistics were refreshed. This function owns its
+// logging; callers need add nothing.
+func (s *Store) RefreshPlannerStats(analysisLimit int) bool {
+	if analysisLimit < 0 {
+		return false
+	}
+	first := !s.hasPlannerStats()
+	start := time.Now()
+	// Tagged for /api/perf writer-lock visibility (#1340).
+	if _, err := s.instrumentedExec("analyze", fmt.Sprintf("PRAGMA analysis_limit=%d", analysisLimit)); err != nil {
+		log.Printf("[analyze] could not set analysis_limit: %v", err)
+		return false
+	}
+	if _, err := s.instrumentedExec("analyze", "ANALYZE"); err != nil {
+		log.Printf("[analyze] ANALYZE failed: %v", err)
+		return false
+	}
+	elapsed := time.Since(start).Round(time.Millisecond)
+	if first {
+		log.Printf("[analyze] planner statistics built in %v (analysis_limit=%d, first run against this database)", elapsed, analysisLimit)
+	} else {
+		log.Printf("[analyze] planner statistics refreshed in %v (analysis_limit=%d)", elapsed, analysisLimit)
+	}
+	return true
+}
+
+// EnsurePlannerStats builds planner statistics when the database has none, and
+// reports whether it did (#2058).
+//
+// This exists to close the window the startup stagger opens. The refresh ticker
+// waits 2 minutes before its first run, and a query arriving in that window
+// against a database with no statistics at all gets the plan measured in
+// RefreshPlannerStats above: 143,442 pages read, which timed at 56.7s cold on
+// the 9.4 GB staging file.
+//
+// It fires once per database, not once per restart. sqlite_stat1 is an ordinary
+// table, so once written it stays in the file and a fresh read-only connection
+// reads it back (verified across connection close, and through a mode=ro
+// handle). Every later start therefore costs one query against sqlite_master and
+// leaves the work to the ticker.
+//
+// The trade is a one-time ANALYZE early in startup, and it is not cheap on a
+// cold page cache. Observed on staging at 9.4 GB: 3m43.9s, against the 2.0s the
+// same statement takes warm. For those 3m44s it holds the store's single write
+// connection (SetMaxOpenConns(1) in OpenStore), so ingest stalls and buffers: the
+// observations table took zero rows for four minutes and then 1027 in the minute
+// the ANALYZE finished, against about 130 a minute either side, with nothing
+// dropped. Hence the warning below, so an operator watching a first deploy can
+// tell this apart from a hang.
+//
+// That cost belongs to the first ANALYZE, not to running it here. The refresh
+// ticker would pay exactly the same 3m44s two minutes later; this only moves it
+// earlier, where it overlaps the startup burst the ingest buffer is already
+// sized for.
+//
+// A negative analysisLimit disables it without a word: the warning is only
+// true when an ANALYZE follows it.
+func (s *Store) EnsurePlannerStats(analysisLimit int) bool {
+	if analysisLimit < 0 || s.hasPlannerStats() {
+		return false
+	}
+	log.Printf("[analyze] this database has no planner statistics; building them now. " +
+		"ANALYZE holds the single write connection until it finishes (3m43.9s measured on 9.4 GB, cold), " +
+		"so ingest will buffer and catch up. Once per database, not once per restart.")
+	return s.RefreshPlannerStats(analysisLimit)
+}
+
+// hasPlannerStats reports whether ANALYZE has ever run against this database.
+// A query error reads as "no stats", which is the safe direction for both
+// callers: it costs the log line a wrong word, and costs EnsurePlannerStats one
+// ANALYZE that was not needed, rather than skipping one that was.
+func (s *Store) hasPlannerStats() bool {
+	var n int
+	if err := s.db.QueryRow(
+		`SELECT count(*) FROM sqlite_master WHERE type='table' AND name='sqlite_stat1'`).Scan(&n); err != nil {
+		return false
+	}
+	return n > 0
+}
+
 // Checkpoint runs a WAL checkpoint (TRUNCATE mode).
 // Returns the number of WAL frames checkpointed (0 if WAL was already empty).
 // TRUNCATE resets the WAL file to zero bytes when all frames are checkpointed;

cmd/ingestor/main.go

@@ -327,9 +327,16 @@ func main() {
 	// DB) is absorbed by IngestBuffer instead of delaying start-up. Cancelled
 	// on shutdown so store.Close() does not wait for it; it resumes on the
 	// next start.
+	//
+	// It shares its start with the one-off planner-statistics build (#2058,
+	// issue #100): both hold the single writer for one long statement, so
+	// they run one after the other, statistics first, each behind a drain of
+	// the ingest buffer, never back to back. Why that order, and where the
+	// other startup tickers fall: planner_stats_startup.go.
+	analysisLimit := cfg.AnalysisLimit()
 	routeMaskCtx, stopRouteMaskBackfill := context.WithCancel(context.Background())
 	defer stopRouteMaskBackfill()
-	store.StartRouteMaskBackfill(routeMaskCtx)
+	startupWritersDone := store.startPlannerStatsThenRouteMaskBackfill(routeMaskCtx, analysisLimit, ingestBuffer.Pending)
 
 	// Daily ticker for node retention
 	retentionTicker := time.NewTicker(1 * time.Hour)
@@ -412,6 +419,25 @@ func main() {
 	}()
 	log.Printf("[db] WAL checkpoint scheduled every 1h")
 
+	// Daily planner statistics refresh (#2058).
+	//
+	// The one-off build for a database that has never been analyzed runs in
+	// the startup sequence above (Store.EnsurePlannerStats carries what it
+	// costs). The routine refresh is staggered 2 minutes past the point that
+	// sequence hands over, so it never lands on the startup ANALYZE or the
+	// backfill's index build, and so it also sees the rows the startup burst
+	// added. On a first start it re-runs the build two minutes later, warm.
+	//
+	// Bounded by analysis_limit either way, so neither grows with the file the way
+	// an unbounded ANALYZE does: 2.0s against 242.9s on a 9.4 GB database, both
+	// timed warm. Cold, on a first start, it is 3m43.9s.
+	stopPlannerStatsRefresh, plannerStatsEnabled := store.schedulePlannerStatsRefresh(analysisLimit, startupWritersDone, 2*time.Minute, 24*time.Hour)
+	if plannerStatsEnabled {
+		log.Printf("[analyze] planner statistics refresh scheduled every 24h (analysis_limit=%d)", analysisLimit)
+	} else {
+		log.Printf("[analyze] planner statistics build and refresh disabled (db.analysisLimit=%d)", analysisLimit)
+	}
+
 	// Daily neighbor_edges retention (#1287 — moved from cmd/server).
 	{
 		nDays := cfg.NeighborEdgesDaysOrDefault()
@@ -509,6 +535,7 @@ func main() {
 	statsTicker.Stop()
 	pruneQueueTicker.Stop()
 	walCheckpointTicker.Stop()
+	stopPlannerStatsRefresh()
 	stopWatchdog()
 	store.LogStats() // final stats on shutdown
 	for _, c := range clients {

cmd/ingestor/planner_stats_startup.go

@@ -0,0 +1,124 @@
+package main
+
+import (
+	"context"
+	"log"
+	"sync"
+	"time"
+)
+
+// Startup ordering of the one-off single-writer holds (issue #100, the fork
+// adaptation of `Kpa-clawbot/CoreScope#2072`/`#2074`).
+//
+// Two startup steps hold the store's single write connection (writerMu,
+// SetMaxOpenConns(1)) for one long statement each:
+//
+//   - EnsurePlannerStats: ANALYZE on a database that has never been analyzed.
+//     Once per database; 3m43.9s cold on upstream's 9.4 GB staging file.
+//   - the #89 route_mask backfill: CREATE INDEX of its pending index (~20 s on
+//     a staging-sized DB), then short batches with a yield between them.
+//
+// While either runs, IngestBuffer's consumer is blocked and packets queue.
+// Started side by side, the two holds would run back to back (writerMu is
+// FIFO once contended) and the ingest pause would be their sum; started the
+// moment the buffer opens, either would also sit on top of the backlog the
+// buffer collected during the startup migrations. So they run in sequence on
+// one goroutine, each behind a drain of the buffer:
+//
+//	drain → EnsurePlannerStats → drain → StartRouteMaskBackfill
+//
+// ANALYZE goes first because it is what the read path is waiting for (the
+// channel queries run on the wrong index until sqlite_stat1 exists), while the
+// backfill is background repair whose batch phase alone runs for minutes on a
+// staging-sized table. On a restart EnsurePlannerStats is one sqlite_master
+// query, so the backfill starts after nothing more than the first drain.
+//
+// The routine refresh (schedulePlannerStatsRefresh) starts its stagger only
+// after this sequence hands over, so it cannot land on the startup ANALYZE.
+// The other startup tickers (WAL checkpoint +30s, observer retention +90s,
+// multibyte persist +2m, neighbor prune +4m) are left where they were: each is
+// a short write that queues behind whichever hold is running and then yields
+// writerMu back, so they add seconds, not a second long hold.
+
+// plannerStatsDrainTimeout bounds each drain wait: if ingest outpaces the
+// writer the buffer may never read empty, and the next step must still run.
+// Vars so tests can shrink them.
+var (
+	plannerStatsDrainTimeout = 2 * time.Minute
+	plannerStatsDrainPoll    = 250 * time.Millisecond
+)
+
+// startPlannerStatsThenRouteMaskBackfill runs the sequence above on its own
+// goroutine and returns a channel closed once the backfill has been scheduled
+// (or skipped because ctx was cancelled). pending reports the ingest backlog;
+// main passes IngestBuffer.Pending.
+func (s *Store) startPlannerStatsThenRouteMaskBackfill(ctx context.Context, analysisLimit int, pending func() int) <-chan struct{} {
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		waitForIngestDrain(ctx, pending, "planner statistics")
+		if ctx.Err() != nil {
+			return
+		}
+		s.EnsurePlannerStats(analysisLimit)
+		waitForIngestDrain(ctx, pending, "route_mask backfill")
+		if ctx.Err() != nil {
+			return
+		}
+		s.StartRouteMaskBackfill(ctx)
+	}()
+	return done
+}
+
+// waitForIngestDrain blocks until pending() reads zero, ctx is cancelled, or
+// plannerStatsDrainTimeout passes.
+func waitForIngestDrain(ctx context.Context, pending func() int, next string) {
+	deadline := time.Now().Add(plannerStatsDrainTimeout)
+	for pending() > 0 {
+		if time.Now().After(deadline) {
+			log.Printf("[startup] ingest buffer still has %d pending after %v; starting %s anyway", pending(), plannerStatsDrainTimeout, next)
+			return
+		}
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(plannerStatsDrainPoll):
+		}
+	}
+}
+
+// schedulePlannerStatsRefresh starts the routine refresh (#2058): once after
+// the stagger, then every interval. The stagger starts when after closes (the
+// startup sequence above has handed over). A negative analysisLimit disables
+// it and reports enabled=false. stop ends the ticker.
+func (s *Store) schedulePlannerStatsRefresh(analysisLimit int, after <-chan struct{}, stagger, interval time.Duration) (stop func(), enabled bool) {
+	if analysisLimit < 0 {
+		return func() {}, false
+	}
+	quit := make(chan struct{})
+	go func() {
+		select {
+		case <-after:
+		case <-quit:
+			return
+		}
+		select {
+		case <-time.After(stagger):
+		case <-quit:
+			return
+		}
+		s.RefreshPlannerStats(analysisLimit)
+		ticker := time.NewTicker(interval)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ticker.C:
+				s.RefreshPlannerStats(analysisLimit)
+			case <-quit:
+				return
+			}
+		}
+	}()
+	var once sync.Once
+	return func() { once.Do(func() { close(quit) }) }, true
+}