perf(query): resume edge drains without rediscovering the whole window

internal/query/batch_cache.go

@@ -17,7 +17,7 @@ import (
 
 // readCacheVersion covers both schema and aggregation semantics. These tables
 // are disposable; a mismatch rebuilds them together from immutable Parquet.
-const readCacheVersion = 4
+const readCacheVersion = 5
 
 const (
 	readCacheBatchLimit = 64
@@ -62,6 +62,12 @@ func createBatchCaches(db *sql.DB) error {
 				return err
 			}
 		}
+		// Bucket counts and the unfinished edge window are one disposable
+		// unit. Restart that window from its unchanged ingestion watermark
+		// when upgrading from a cache without these counts (or new semantics).
+		if _, err := tx.Exec(`DELETE FROM rollup_state WHERE cache_key IN (?, ?)`, edgeRollupSubCursorKey, edgeRollupSubWindowKey); err != nil {
+			return err
+		}
 		if _, err := tx.Exec(`DELETE FROM read_cache_version`); err != nil {
 			return err
 		}
@@ -70,6 +76,7 @@ func createBatchCaches(db *sql.DB) error {
 		}
 	}
 	for _, stmt := range []string{
+		`CREATE TABLE IF NOT EXISTS read_edge_buckets (namespace VARCHAR, bucket TIMESTAMPTZ_NS, spans BIGINT)`,
 		`CREATE TABLE IF NOT EXISTS read_batches (batch_id VARCHAR PRIMARY KEY,replacement_id VARCHAR)`,
 		`CREATE TABLE IF NOT EXISTS read_trace_candidates (trace_id VARCHAR PRIMARY KEY,namespace VARCHAR,service VARCHAR,min_start BIGINT,max_start BIGINT,max_end BIGINT,has_error INTEGER)`,
 		`CREATE TABLE IF NOT EXISTS read_trace_parts (batch_id VARCHAR, namespace VARCHAR, service VARCHAR, trace_id VARCHAR, min_start BIGINT, max_start BIGINT, max_end BIGINT, has_error INTEGER)`,

internal/query/duck.go

@@ -342,16 +342,14 @@ func (d *Duck) skipRollupToLatest(ctx context.Context) error {
 	}{
 		{serviceRollupStateKey, horizon},
 		{edgeRollupStateKey, horizon},
-		// Skipping discards any half-aggregated window, so its resume point
-		// must go too or the next pass would reopen a window behind the
-		// watermark it just advanced.
-		{edgeRollupSubCursorKey, 0},
-		{edgeRollupSubWindowKey, 0},
 	} {
 		if err := storeRollupWatermark(ctx, tx, w.key, w.watermark); err != nil {
 			return err
 		}
 	}
+	if err := clearEdgeRollupCursor(ctx, tx); err != nil {
+		return err
+	}
 	if err := tx.Commit(); err != nil {
 		return err
 	}
@@ -960,12 +958,19 @@ func (d *Duck) refreshEdgeRollupPass(ctx context.Context) (int64, bool, error) {
 	}
 	newWatermark := windowEnd
 
-	// Build the incoming/recent-arrival repair relation once in this snapshot.
-	// Count every affected span per minute so adaptive counting preserves repair
-	// duplicates without retaining all of their individual rows.
-	// TEMP is connection-local; creation rolls back on failure, and the table
-	// is explicitly dropped before commit so no later pass can observe it.
-	if _, err := tx.ExecContext(ctx, createEdgePassBucketsSQL, windowStart, windowEnd); err != nil {
+	// Discover affected buckets once per frozen ingestion window, including
+	// recent-arrival repairs. Persist only their counts, atomically with the
+	// cursor. Resumes read these small summaries rather than the original spans.
+	if !resuming {
+		if _, err := tx.ExecContext(ctx, "DELETE FROM read_edge_buckets"); err != nil {
+			return 0, false, err
+		}
+		if _, err := tx.ExecContext(ctx, "INSERT INTO read_edge_buckets "+edgePassBucketsSelectSQL, windowStart, windowEnd); err != nil {
+			return 0, false, err
+		}
+	}
+	// Keep the per-pass name connection-local. It never holds individual spans.
+	if _, err := tx.ExecContext(ctx, "CREATE TEMP VIEW edge_pass_buckets AS SELECT * FROM read_edge_buckets"); err != nil {
 		return 0, false, err
 	}
 
@@ -1037,14 +1042,17 @@ FROM edge_pass_buckets`).Scan(&minStartT, &maxStartT)
 		}
 	} else {
 		storedWatermark = lastWatermark
+		if _, err := tx.ExecContext(ctx, `DELETE FROM read_edge_buckets WHERE bucket < ?::TIMESTAMP_NS::TIMESTAMPTZ_NS`, time.Unix(0, nextCursor)); err != nil {
+			return 0, false, err
+		}
 		if err := storeRollupWatermark(ctx, tx, edgeRollupSubCursorKey, nextCursor); err != nil {
 			return 0, false, err
 		}
 		if err := storeRollupWatermark(ctx, tx, edgeRollupSubWindowKey, windowEnd); err != nil {
 			return 0, false, err
 		}
 	}
-	if _, err := tx.ExecContext(ctx, "DROP TABLE edge_pass_buckets"); err != nil {
+	if _, err := tx.ExecContext(ctx, "DROP VIEW edge_pass_buckets"); err != nil {
 		return 0, false, err
 	}
 	if err := tx.Commit(); err != nil {
@@ -1058,9 +1066,12 @@ FROM edge_pass_buckets`).Scan(&minStartT, &maxStartT)
 	return totalAffected, !completed || chunked, nil
 }
 
-// clearEdgeRollupCursor drops the resume point once its window is fully
-// aggregated, so the next pass opens a fresh ingested window.
+// clearEdgeRollupCursor discards the resume point and its bucket summaries
+// together, so the next pass opens a fresh ingested window.
 func clearEdgeRollupCursor(ctx context.Context, tx *sql.Tx) error {
+	if _, err := tx.ExecContext(ctx, "DELETE FROM read_edge_buckets"); err != nil {
+		return err
+	}
 	if err := storeRollupWatermark(ctx, tx, edgeRollupSubCursorKey, 0); err != nil {
 		return err
 	}
@@ -1413,8 +1424,9 @@ const edgeArrivalGap = 5 * time.Minute
 // Only relationship keys are read from the recent ingestion range. Its stamp
 // predicates allow Parquet footer pruning independently of event-time history.
 // Messaging uses the aggregate's exact keys, including the shared minute.
-var createEdgePassBucketsSQL = fmt.Sprintf(`
-CREATE TEMP TABLE edge_pass_buckets AS
+var createEdgePassBucketsSQL = "CREATE TEMP TABLE edge_pass_buckets AS " + edgePassBucketsSelectSQL
+
+var edgePassBucketsSelectSQL = fmt.Sprintf(`
 WITH incoming AS MATERIALIZED (
   SELECT namespace, trace_id, span_id, start_time, kind,
          messaging_destination, messaging_system

internal/query/edge_pass_test.go

@@ -110,7 +110,7 @@ func TestEdgePassDropsRelationOnEveryExit(t *testing.T) {
 			d := &Duck{DB: db, writeDB: db, publicationBounds: func() (int64, int64, int64) { return stamp, stamp, stamp }}
 			checkAbsent := func() {
 				var count int
-				if err := db.QueryRow("SELECT count(*) FROM duckdb_tables() WHERE table_name IN ('edge_pass_buckets','edge_pass_parents','edge_pass_unmatched')").Scan(&count); err != nil || count != 0 {
+				if err := db.QueryRow("SELECT count(*) FROM (SELECT table_name AS name FROM duckdb_tables() UNION ALL SELECT view_name AS name FROM duckdb_views()) WHERE name IN ('edge_pass_buckets','edge_pass_parents','edge_pass_unmatched')").Scan(&count); err != nil || count != 0 {
 					t.Fatalf("pass relation survived: count=%d err=%v", count, err)
 				}
 			}
@@ -126,6 +126,10 @@ func TestEdgePassDropsRelationOnEveryExit(t *testing.T) {
 				if watermark, cursor := readEdgeRollupState(t, d); watermark != 0 || cursor != 0 {
 					t.Fatalf("failed pass advanced state: watermark=%d cursor=%d", watermark, cursor)
 				}
+				var buckets int
+				if err := db.QueryRow("SELECT count(*) FROM read_edge_buckets").Scan(&buckets); err != nil || buckets != 0 {
+					t.Fatalf("failed discovery retained buckets=%d err=%v", buckets, err)
+				}
 				if _, err := db.Exec("ALTER TABLE edge_rollup RENAME COLUMN saved_avg_ms TO avg_ms"); err != nil {
 					t.Fatal(err)
 				}
@@ -135,6 +139,25 @@ func TestEdgePassDropsRelationOnEveryExit(t *testing.T) {
 			}
 			checkAbsent()
 			if !empty {
+				watermark, cursor := readEdgeRollupState(t, d)
+				// A failed resume must retain the previous cursor AND summaries.
+				if _, err := db.Exec("ALTER TABLE edge_rollup RENAME COLUMN avg_ms TO saved_avg_ms"); err != nil {
+					t.Fatal(err)
+				}
+				if _, _, err := d.refreshEdgeRollupPass(t.Context()); err == nil {
+					t.Fatal("incompatible rollup schema did not fail the resume")
+				}
+				checkAbsent()
+				if w, c := readEdgeRollupState(t, d); w != watermark || c != cursor {
+					t.Fatalf("failed resume moved state: watermark=%d cursor=%d", w, c)
+				}
+				var buckets int
+				if err := db.QueryRow("SELECT count(*) FROM read_edge_buckets").Scan(&buckets); err != nil || buckets != 1 {
+					t.Fatalf("failed resume lost summaries: buckets=%d err=%v", buckets, err)
+				}
+				if _, err := db.Exec("ALTER TABLE edge_rollup RENAME COLUMN saved_avg_ms TO avg_ms"); err != nil {
+					t.Fatal(err)
+				}
 				if _, pending, err := d.refreshEdgeRollupPass(t.Context()); err != nil || pending {
 					t.Fatalf("resumed pass pending=%v err=%v", pending, err)
 				}

internal/query/edge_resume_test.go

@@ -0,0 +1,140 @@
+package query
+
+import (
+	"database/sql"
+	"fmt"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestEdgeResumeSetupIsLinearAcrossRestart(t *testing.T) {
+	for _, minutes := range []int{5, 15, 30} {
+		t.Run(fmt.Sprintf("%d_minutes", minutes), func(t *testing.T) {
+			path := filepath.Join(t.TempDir(), "cache.duckdb")
+			var db *sql.DB
+			open := func() {
+				var err error
+				db, err = sql.Open("duckdb", path)
+				if err != nil {
+					t.Fatal(err)
+				}
+				db.SetMaxOpenConns(1)
+				if _, err := db.Exec("SET TimeZone='UTC'"); err != nil {
+					t.Fatal(err)
+				}
+				if err := CreateTables(db); err != nil {
+					t.Fatal(err)
+				}
+			}
+			open()
+			defer func() { _ = db.Close() }()
+			base := time.Date(2026, 10, 1, 12, 0, 0, 0, time.UTC)
+			stamp := base.UnixNano()
+			// N occupied event minutes, separated to force N resumable passes
+			// without needing 250k spans per minute in the regular test suite.
+			if _, err := db.Exec(`INSERT INTO telemetry.spans
+ (namespace,trace_id,span_id,parent_span_id,service,kind,start_time,duration_ms,status,ingested_unix_nano)
+ SELECT 'ns',(i//2)::VARCHAR,i::VARCHAR,CASE WHEN i%2=1 THEN (i-1)::VARCHAR END,
+ CASE WHEN i%2=0 THEN 'gateway' ELSE 'checkout' END,'SPAN_KIND_SERVER',
+ ?::TIMESTAMP_NS + (i//2)*INTERVAL 31 MINUTE,5,'OK',? FROM range(?) t(i)`, base, stamp, minutes*2); err != nil {
+				t.Fatal(err)
+			}
+			if _, err := db.Exec("CREATE SEQUENCE edge_setup_reads"); err != nil {
+				t.Fatal(err)
+			}
+			// Only discovery uses ingestion stamps. Count actual source-row
+			// evaluations there, rather than timing the drain or inspecting SQL.
+			view := strings.Replace(viewSpans, "  ingested_unix_nano,", "  CASE WHEN nextval('edge_setup_reads')>0 THEN ingested_unix_nano END AS ingested_unix_nano,", 1)
+			if _, err := db.Exec(view); err != nil {
+				t.Fatal(err)
+			}
+			var setupReads int64
+			for pass := 0; pass < minutes; pass++ {
+				d := &Duck{DB: db, writeDB: db, publicationBounds: func() (int64, int64, int64) { return stamp, stamp, stamp }}
+				_, pending, err := d.refreshEdgeRollupPass(t.Context())
+				if err != nil || pending != (pass < minutes-1) {
+					t.Fatalf("pass=%d pending=%v err=%v", pass, pending, err)
+				}
+				var reads int64
+				if err := db.QueryRow("SELECT last_value FROM duckdb_sequences() WHERE sequence_name='edge_setup_reads'").Scan(&reads); err != nil {
+					t.Fatal(err)
+				}
+				if pass == 0 {
+					setupReads = reads
+					if reads < int64(minutes*2) || reads > int64(minutes*8) {
+						t.Fatalf("setup reads=%d, want a constant number per input row", reads)
+					}
+				} else if reads != setupReads {
+					t.Fatalf("resume repeated discovery: reads=%d first pass=%d", reads, setupReads)
+				}
+				var remaining int
+				if err := db.QueryRow("SELECT count(*) FROM read_edge_buckets").Scan(&remaining); err != nil || remaining != minutes-pass-1 {
+					t.Fatalf("remaining=%d pass=%d err=%v", remaining, pass, err)
+				}
+				if pass == 0 {
+					// Close the native database: summaries and cursor must survive,
+					// while the connection-local pass view must have disappeared.
+					if err := db.Close(); err != nil {
+						t.Fatal(err)
+					}
+					open()
+				}
+			}
+			var calls, buckets int
+			if err := db.QueryRow("SELECT sum(calls),count(DISTINCT bucket) FROM edge_rollup").Scan(&calls, &buckets); err != nil || calls != minutes || buckets != minutes {
+				t.Fatalf("calls=%d buckets=%d err=%v", calls, buckets, err)
+			}
+			t.Logf("occupied_minutes=%d setup_row_evaluations=%d passes=%d", minutes, setupReads, minutes)
+		})
+	}
+}
+
+func TestEdgeResumeKeepsFrozenWindowAndRepairsNewArrivals(t *testing.T) {
+	db := openTestDuck(t)
+	if err := CreateTables(db); err != nil {
+		t.Fatal(err)
+	}
+	if err := CreateViews(db); err != nil {
+		t.Fatal(err)
+	}
+	base := time.Date(2026, 10, 1, 12, 0, 0, 0, time.UTC)
+	stamp := base.Add(time.Hour).UnixNano()
+	horizon := stamp
+	d := &Duck{DB: db, writeDB: db, publicationBounds: func() (int64, int64, int64) { return horizon, horizon, stamp }}
+	for _, span := range []rollupTestSpan{
+		{namespace: "ns", traceID: "late", spanID: "child", parentSpanID: "parent", service: "late-child", start: base, ingested: stamp},
+		{namespace: "ns", traceID: "existing", spanID: "parent", service: "gateway", start: base, ingested: stamp},
+		{namespace: "ns", traceID: "existing", spanID: "child", parentSpanID: "parent", service: "existing-child", start: base, ingested: stamp},
+		{namespace: "ns", traceID: "tail", spanID: "parent", service: "gateway", start: base.Add(31 * time.Minute), ingested: stamp},
+		{namespace: "ns", traceID: "tail", spanID: "child", parentSpanID: "parent", service: "tail-child", start: base.Add(31 * time.Minute), ingested: stamp},
+	} {
+		insertRollupTestSpan(t, db, span)
+	}
+	if _, pending, err := d.refreshEdgeRollupPass(t.Context()); err != nil || !pending {
+		t.Fatalf("initial pass pending=%v err=%v", pending, err)
+	}
+	// Publish a parent for an already-completed event bucket while the old
+	// ingestion window is paused. It belongs to the NEXT discovery window.
+	horizon++
+	insertRollupTestSpan(t, db, rollupTestSpan{namespace: "ns", traceID: "late", spanID: "parent", service: "late-gateway", start: base, ingested: horizon})
+	if _, pending, err := d.refreshEdgeRollupPass(t.Context()); err != nil || !pending {
+		t.Fatalf("resumed pass pending=%v err=%v", pending, err)
+	}
+	if watermark, cursor := readEdgeRollupState(t, d); watermark != stamp || cursor != 0 {
+		t.Fatalf("resume changed frozen ingestion end: watermark=%d cursor=%d", watermark, cursor)
+	}
+	requireRowCount(t, db, "SELECT count(*) FROM edge_rollup WHERE callee=?", "late-child", 0)
+	if _, pending, err := d.refreshEdgeRollupPass(t.Context()); err != nil || pending {
+		t.Fatalf("repair pass pending=%v err=%v", pending, err)
+	}
+	// Repair replaces the complete bucket: both the new relationship and the
+	// unrelated edge that was already aggregated must appear exactly once.
+	for _, child := range []string{"late-child", "existing-child", "tail-child"} {
+		requireRowCount(t, db, "SELECT count(*) FROM edge_rollup WHERE calls=1 AND callee=?", child, 1)
+	}
+	if watermark, cursor := readEdgeRollupState(t, d); watermark != horizon || cursor != 0 {
+		t.Fatalf("repair did not reach publication horizon: watermark=%d cursor=%d", watermark, cursor)
+	}
+}

internal/query/file_snapshot_test.go

@@ -321,20 +321,27 @@ func TestReadCacheSemanticVersionRebuildsAllContributions(t *testing.T) {
 	if _, err := d.RefreshReadCaches(t.Context()); err != nil {
 		t.Fatal(err)
 	}
-	for _, stmt := range []string{`UPDATE read_trace_candidates SET min_start=0`, `UPDATE read_cache_version SET version=3`, `CREATE TABLE read_endpoints (value INTEGER)`, `CREATE TABLE read_logs (value INTEGER)`, `CREATE TABLE read_obsolete (value INTEGER)`, `CREATE TABLE endpoint_rollup (value INTEGER)`, `INSERT INTO rollup_state VALUES ('endpoint_rollup:prod',1,now())`} {
+	for _, stmt := range []string{`UPDATE read_trace_candidates SET min_start=0`, `UPDATE read_cache_version SET version=4`, `INSERT INTO read_edge_buckets VALUES ('ns',now(),10)`, `INSERT INTO rollup_state VALUES ('edge_rollup_v2_substart',1,now()),('edge_rollup_v2_subwindow_end',2,now())`, `CREATE TABLE read_endpoints (value INTEGER)`, `CREATE TABLE read_logs (value INTEGER)`, `CREATE TABLE read_obsolete (value INTEGER)`, `CREATE TABLE endpoint_rollup (value INTEGER)`, `INSERT INTO rollup_state VALUES ('endpoint_rollup:prod',1,now())`} {
 		if _, err := d.DB.Exec(stmt); err != nil {
 			t.Fatal(err)
 		}
 	}
 	if err := CreateCacheTables(d.DB); err != nil {
 		t.Fatal(err)
 	}
-	for _, table := range []string{"read_batches", "read_trace_parts", "read_trace_candidates"} {
+	for _, table := range []string{"read_batches", "read_trace_parts", "read_trace_candidates", "read_edge_buckets"} {
 		var count int
 		if err := d.DB.QueryRow("SELECT count(*) FROM " + table).Scan(&count); err != nil || count != 0 {
 			t.Fatalf("%s survived semantic change: %d %v", table, count, err)
 		}
 	}
+	if watermark, cursor := readEdgeRollupState(t, d); watermark != 0 || cursor != 0 {
+		t.Fatalf("obsolete edge cursor survived: watermark=%d cursor=%d", watermark, cursor)
+	}
+	var resumeEnd int64
+	if err := d.DB.QueryRow(`SELECT coalesce(max(last_ingested_unix_nano),0) FROM rollup_state WHERE cache_key=?`, edgeRollupSubWindowKey).Scan(&resumeEnd); err != nil || resumeEnd != 0 {
+		t.Fatalf("obsolete edge window survived: end=%d err=%v", resumeEnd, err)
+	}
 	var obsolete int
 	if err := d.DB.QueryRow(`SELECT count(*) FROM duckdb_tables() WHERE table_name IN ('read_obsolete','endpoint_rollup','read_endpoints','read_logs')`).Scan(&obsolete); err != nil || obsolete != 0 {
 		t.Fatalf("obsolete tables=%d: %v", obsolete, err)
@@ -350,7 +357,7 @@ func TestReadCacheSemanticVersionRebuildsAllContributions(t *testing.T) {
 		t.Fatalf("rebuilt bound=%d: %v", minStart, err)
 	}
 	var version int
-	if err := d.DB.QueryRow(`SELECT version FROM read_cache_version`).Scan(&version); err != nil || version != 4 {
+	if err := d.DB.QueryRow(`SELECT version FROM read_cache_version`).Scan(&version); err != nil || version != 5 {
 		t.Fatalf("version=%d: %v", version, err)
 	}
 

internal/query/sql_boundary_test.go

@@ -33,6 +33,7 @@ func TestSQLBoundaryRejectsHiddenReadsAndMutations(t *testing.T) {
 		"SELECT * FROM read_traces",
 		"SELECT * FROM read_trace_candidates",
 		"SELECT * FROM read_cache_version",
+		"SELECT * FROM read_edge_buckets",
 		"SELECT * FROM endpoint_rollup",
 		"SELECT current_setting('allowed_directories')",
 		"SELECT getvariable('operator_secret')",
@@ -115,6 +116,7 @@ func TestExecuteSQLRejectsEngineMetadataReads(t *testing.T) {
 		"SELECT * FROM read_traces",
 		"SELECT * FROM read_trace_candidates",
 		"SELECT * FROM read_cache_version",
+		"SELECT * FROM read_edge_buckets",
 		"SELECT * FROM endpoint_rollup",
 		"SELECT current_setting('allowed_directories')",
 		"SELECT getvariable('operator_secret')",