perf: keep low-concurrency exporters fast and edge parent lookups near

internal/query/duck.go

@@ -989,7 +989,7 @@ FROM edge_pass_buckets`).Scan(&minStartT, &maxStartT)
 			if _, err := tx.ExecContext(ctx, edgeRollupDeleteSQL, subLo, subHi); err != nil {
 				return 0, false, err
 			}
-			res, err := tx.ExecContext(ctx, edgeRollupInsertSQL, subLo, subHi)
+			res, err := insertEdgeRollup(ctx, tx, subLo, subHi)
 			if err != nil {
 				return 0, false, err
 			}
@@ -1480,17 +1480,14 @@ WHERE EXISTS (
     AND affected.bucket = edge_rollup.bucket
 );`
 
-const edgeRollupInsertSQL = edgeAffectedBucketsSQL + `,
--- The affected bucket range, computed once. As four scalar subqueries
--- repeated across the predicates below, these read as correlated to the
--- planner and cost a re-derivation each.
+const edgeChildTracesSQL = edgeAffectedBucketsSQL + `,
+-- Compute the affected bucket range once for the child and parent filters.
 bounds AS (
   SELECT MIN(bucket) AS lo, MAX(bucket) AS hi FROM affected
 ),
 -- Only traces with children in affected buckets can contribute call edges.
 -- Bound this distinct relation to the same child window as call_edges, before
--- building the parent side; unrelated traces in the two-hour range must not
--- occupy the parent hash table. Namespace is part of the trace identity.
+-- building the parent side. Namespace is part of the trace identity.
 child_traces AS (
   SELECT DISTINCT child.namespace, child.trace_id
   FROM spans child
@@ -1502,20 +1499,113 @@ child_traces AS (
     AND child.service != ''
     AND child.start_time >= (bounds.lo)::TIMESTAMP_NS::TIMESTAMPTZ_NS
     AND child.start_time < (bounds.hi + INTERVAL 1 MINUTE)::TIMESTAMP_NS::TIMESTAMPTZ_NS
+)`
+
+// Parent lookup is connection-local and transactional, like edge_pass_buckets.
+// The wide scan is issued only after an explicit unmatched-trace check.
+// The result reports whether that scan ran.
+func prepareEdgeParents(ctx context.Context, tx *sql.Tx, lo, hi time.Time) (bool, error) {
+	if _, err := tx.ExecContext(ctx, createEdgeParentsSQL, lo, hi); err != nil {
+		return false, err
+	}
+	if _, err := tx.ExecContext(ctx, createEdgeUnmatchedSQL, lo, hi); err != nil {
+		return false, err
+	}
+	var unmatched bool
+	if err := tx.QueryRowContext(ctx, "SELECT EXISTS (SELECT 1 FROM edge_pass_unmatched)").Scan(&unmatched); err != nil {
+		return false, err
+	}
+	if unmatched {
+		// Replace these traces rather than unioning overlapping parent windows.
+		// This retains duplicates without a disjoint time filter on the wide scan.
+		if _, err := tx.ExecContext(ctx, deleteEdgeUnmatchedParentsSQL); err != nil {
+			return false, err
+		}
+		if _, err := tx.ExecContext(ctx, insertEdgeDistantParentsSQL, lo, hi); err != nil {
+			return false, err
+		}
+	}
+	_, err := tx.ExecContext(ctx, "DROP TABLE edge_pass_unmatched")
+	return unmatched, err
+}
+
+func insertEdgeRollup(ctx context.Context, tx *sql.Tx, lo, hi time.Time) (sql.Result, error) {
+	if _, err := prepareEdgeParents(ctx, tx, lo, hi); err != nil {
+		return nil, err
+	}
+	res, err := tx.ExecContext(ctx, edgeRollupInsertSQL, lo, hi)
+	if err != nil {
+		return nil, err
+	}
+	if _, err := tx.ExecContext(ctx, "DROP TABLE edge_pass_parents"); err != nil {
+		return nil, err
+	}
+	return res, nil
+}
+
+const createEdgeParentsSQL = `CREATE TEMP TABLE edge_pass_parents AS ` + edgeChildTracesSQL + `
+SELECT parent.namespace, parent.span_id, parent.trace_id, parent.service
+FROM spans parent
+SEMI JOIN child_traces child
+  ON parent.namespace = child.namespace
+ AND parent.trace_id = child.trace_id
+CROSS JOIN bounds
+WHERE parent.start_time >= (bounds.lo - INTERVAL 5 MINUTE)::TIMESTAMP_NS::TIMESTAMPTZ_NS
+  AND parent.start_time <= (bounds.hi + INTERVAL 5 MINUTE)::TIMESTAMP_NS::TIMESTAMPTZ_NS
+  AND parent.service IS NOT NULL
+  AND parent.service != ''`
+
+const createEdgeUnmatchedSQL = `CREATE TEMP TABLE edge_pass_unmatched AS ` + edgeAffectedBucketsSQL + `,
+bounds AS (
+  SELECT MIN(bucket) AS lo, MAX(bucket) AS hi FROM affected
+)
+SELECT DISTINCT child.namespace, child.trace_id
+FROM spans child
+JOIN affected a
+  ON a.namespace = child.namespace
+ AND a.bucket = date_trunc('minute', child.start_time::TIMESTAMP_NS)
+ANTI JOIN edge_pass_parents parent
+  ON parent.namespace = child.namespace
+ AND parent.trace_id = child.trace_id
+ AND parent.span_id = child.parent_span_id
+CROSS JOIN bounds
+WHERE child.service IS NOT NULL
+  AND child.service != ''
+  AND child.parent_span_id IS NOT NULL
+  AND child.parent_span_id != ''
+  AND child.start_time >= bounds.lo::TIMESTAMP_NS::TIMESTAMPTZ_NS
+  AND child.start_time < (bounds.hi + INTERVAL 1 MINUTE)::TIMESTAMP_NS::TIMESTAMPTZ_NS`
+
+const deleteEdgeUnmatchedParentsSQL = `
+DELETE FROM edge_pass_parents
+WHERE EXISTS (
+  SELECT 1 FROM edge_pass_unmatched child
+  WHERE child.namespace = edge_pass_parents.namespace
+    AND child.trace_id = edge_pass_parents.trace_id
+)`
+
+const insertEdgeDistantParentsSQL = edgeAffectedBucketsSQL + `,
+bounds AS (
+  SELECT MIN(bucket) AS lo, MAX(bucket) AS hi FROM affected
+)
+INSERT INTO edge_pass_parents
+SELECT parent.namespace, parent.span_id, parent.trace_id, parent.service
+FROM spans parent
+SEMI JOIN edge_pass_unmatched child
+  ON parent.namespace = child.namespace
+ AND parent.trace_id = child.trace_id
+CROSS JOIN bounds
+WHERE parent.start_time >= (bounds.lo - INTERVAL 1 HOUR)::TIMESTAMP_NS::TIMESTAMPTZ_NS
+  AND parent.start_time <= (bounds.hi + INTERVAL 1 HOUR)::TIMESTAMP_NS::TIMESTAMPTZ_NS
+  AND parent.service IS NOT NULL
+  AND parent.service != ''`
+
+const edgeRollupInsertSQL = edgeAffectedBucketsSQL + `,
+bounds AS (
+  SELECT MIN(bucket) AS lo, MAX(bucket) AS hi FROM affected
 ),
--- Retain the existing +/-1h parent range, including parents outside the child
--- minute. Late arrivals can still widen the affected bucket range.
 parent_scope AS (
-  SELECT parent.namespace, parent.span_id, parent.trace_id, parent.service
-  FROM spans parent
-  SEMI JOIN child_traces child
-    ON parent.namespace = child.namespace
-   AND parent.trace_id = child.trace_id
-  CROSS JOIN bounds
-  WHERE parent.start_time >= (bounds.lo - INTERVAL 1 HOUR)::TIMESTAMP_NS::TIMESTAMPTZ_NS
-    AND parent.start_time <= (bounds.hi + INTERVAL 1 HOUR)::TIMESTAMP_NS::TIMESTAMPTZ_NS
-    AND parent.service IS NOT NULL
-    AND parent.service != ''
+  SELECT namespace, span_id, trace_id, service FROM edge_pass_parents
 ),
 call_edges AS (
   SELECT

internal/query/edge_pass_test.go

@@ -110,23 +110,23 @@ 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='edge_pass_buckets'").Scan(&count); err != nil || count != 0 {
+				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 {
 					t.Fatalf("pass relation survived: count=%d err=%v", count, err)
 				}
 			}
 			if !empty {
 				// Fail after materialization, then retry on the same SQL connection.
-				if _, err := db.Exec("ALTER TABLE edge_rollup RENAME TO saved_edge_rollup"); err != nil {
+				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("missing rollup table did not fail the pass")
+					t.Fatal("incompatible rollup schema did not fail the pass")
 				}
 				checkAbsent()
 				if watermark, cursor := readEdgeRollupState(t, d); watermark != 0 || cursor != 0 {
 					t.Fatalf("failed pass advanced state: watermark=%d cursor=%d", watermark, cursor)
 				}
-				if _, err := db.Exec("ALTER TABLE saved_edge_rollup RENAME TO edge_rollup"); err != nil {
+				if _, err := db.Exec("ALTER TABLE edge_rollup RENAME COLUMN saved_avg_ms TO avg_ms"); err != nil {
 					t.Fatal(err)
 				}
 			}

internal/query/edge_rollup_plan_test.go

@@ -39,6 +39,17 @@ func TestEdgeRollupPlanUsesHashJoins(t *testing.T) {
 	if _, err := db.Exec(createEdgePassBucketsSQL, windowLo, windowHi); err != nil {
 		t.Fatal(err)
 	}
+	tx, err := db.BeginTx(t.Context(), nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := prepareEdgeParents(t.Context(), tx, spanLo, spanHi); err != nil {
+		t.Fatal(err)
+	}
+	if err := tx.Commit(); err != nil {
+		t.Fatal(err)
+	}
+	parentPlan := explain(t, db, strings.TrimPrefix(createEdgeParentsSQL, "CREATE TEMP TABLE edge_pass_parents AS "), spanLo, spanHi)
 	shippedPlan := explain(t, db, edgeRollupInsertSQL, spanLo, spanHi)
 
 	// DuckDB 2 optimizes the historical comma form too. Guard the shipped plan
@@ -51,8 +62,8 @@ func TestEdgeRollupPlanUsesHashJoins(t *testing.T) {
 			t.Errorf("edge rollup plans a %s; the FROM clause has regressed to the comma form:\n%s", bad, shippedPlan)
 		}
 	}
-	if !strings.Contains(shippedPlan, "SEMI") {
-		t.Errorf("edge rollup lost the child-trace semi-join:\n%s", shippedPlan)
+	if !strings.Contains(strings.ToUpper(parentPlan), "SEMI") {
+		t.Errorf("edge rollup lost the child-trace semi-join:\n%s", parentPlan)
 	}
 	if !strings.Contains(shippedPlan, "Hash Join") {
 		t.Errorf("edge rollup plans no HASH_JOIN at all:\n%s", shippedPlan)
@@ -261,7 +272,22 @@ func TestEdgeRollupParentTraceScope(t *testing.T) {
 	if _, err := db.Exec(createEdgePassBucketsSQL, 50, 200); err != nil {
 		t.Fatal(err)
 	}
-	if _, err := db.Exec(edgeRollupInsertSQL, lo, lo.Add(time.Minute)); err != nil {
+	tx, err := db.BeginTx(t.Context(), nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer tx.Rollback()
+	wide, err := prepareEdgeParents(t.Context(), tx, lo, lo.Add(time.Minute))
+	if err != nil || !wide {
+		t.Fatalf("30-minute parent did not run tier 2: wide=%v err=%v", wide, err)
+	}
+	if _, err := tx.ExecContext(t.Context(), edgeRollupInsertSQL, lo, lo.Add(time.Minute)); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := tx.Exec("DROP TABLE edge_pass_parents"); err != nil {
+		t.Fatal(err)
+	}
+	if err := tx.Commit(); err != nil {
 		t.Fatal(err)
 	}
 	var namespace, caller, callee, edgeType string
@@ -289,3 +315,63 @@ func TestEdgeRollupParentTraceScope(t *testing.T) {
 		t.Fatal(err)
 	}
 }
+
+func TestEdgeRollupSkipsDistantParentsWhenChildrenMatch(t *testing.T) {
+	db := openTestDuck(t)
+	seedEdgeRollupFixture(t, db)
+	if _, err := db.Exec(createEdgePassBucketsSQL, 50, 200); err != nil {
+		t.Fatal(err)
+	}
+	tx, err := db.BeginTx(t.Context(), nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer tx.Rollback()
+	lo := time.Date(2026, 9, 20, 12, 0, 0, 0, time.UTC)
+	wide, err := prepareEdgeParents(t.Context(), tx, lo, lo.Add(time.Minute))
+	if err != nil || wide {
+		t.Fatalf("matched children ran tier 2: wide=%v err=%v", wide, err)
+	}
+	if _, err := tx.ExecContext(t.Context(), edgeRollupInsertSQL, lo, lo.Add(time.Minute)); err != nil {
+		t.Fatal(err)
+	}
+	var calls int
+	if err := tx.QueryRow("SELECT sum(calls)::BIGINT FROM edge_rollup").Scan(&calls); err != nil || calls != 1003 {
+		t.Fatalf("calls=%d err=%v", calls, err)
+	}
+}
+
+func TestEdgeRollupDistantParentPreservesNearParentMultiplicity(t *testing.T) {
+	db := openTestDuck(t)
+	if _, err := db.Exec(createEdgeRollupTable); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := db.Exec(`CREATE TABLE spans AS
+ SELECT namespace,trace_id,span_id,parent_span_id,service,kind,status,duration_ms,
+ start_time::TIMESTAMPTZ_NS start_time,ingested_unix_nano,NULL::VARCHAR messaging_system,NULL::VARCHAR messaging_destination
+ FROM (VALUES
+ ('ns','trace','far','','upstream','SPAN_KIND_SERVER','OK',1.0,TIMESTAMP '2026-09-20 11:30:00',1),
+ ('ns','trace','near','','upstream','SPAN_KIND_SERVER','OK',1.0,TIMESTAMP '2026-09-20 11:55:00',1),
+ ('ns','trace','child-far','far','downstream','SPAN_KIND_CLIENT','ERROR',30.0,TIMESTAMP '2026-09-20 12:00:00',100),
+ ('ns','trace','child-near','near','downstream','SPAN_KIND_CLIENT','OK',10.0,TIMESTAMP '2026-09-20 12:00:01',100)
+ ) t(namespace,trace_id,span_id,parent_span_id,service,kind,status,duration_ms,start_time,ingested_unix_nano)`); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := db.Exec(createEdgePassBucketsSQL, 50, 200); err != nil {
+		t.Fatal(err)
+	}
+	tx, err := db.BeginTx(t.Context(), nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer tx.Rollback()
+	lo := time.Date(2026, 9, 20, 12, 0, 0, 0, time.UTC)
+	if _, err := insertEdgeRollup(t.Context(), tx, lo, lo.Add(time.Minute)); err != nil {
+		t.Fatal(err)
+	}
+	var calls int
+	var avg, errorRate float64
+	if err := tx.QueryRow("SELECT calls,avg_ms,error_rate FROM edge_rollup WHERE namespace='ns' AND caller='upstream' AND callee='downstream'").Scan(&calls, &avg, &errorRate); err != nil || calls != 2 || avg != 20 || errorRate != 0.5 {
+		t.Fatalf("calls=%d avg=%g error_rate=%g err=%v", calls, avg, errorRate, err)
+	}
+}

internal/telemetry/store/backpressure_test.go

@@ -208,7 +208,7 @@ func TestWriterAdaptsAdmissionToReclaimableDebt(t *testing.T) {
 			}
 			go submit(0)
 			synctest.Wait()
-			time.Sleep(20 * time.Millisecond)
+			time.Sleep(15 * time.Millisecond)
 			go submit(1)
 			synctest.Wait()
 			time.Sleep(300 * time.Millisecond)
@@ -222,7 +222,7 @@ func TestWriterAdaptsAdmissionToReclaimableDebt(t *testing.T) {
 			wantLatency := [2]time.Duration{10 * time.Millisecond, 10 * time.Millisecond}
 			if debt >= telemetry.CompactionDebtLimit/2 {
 				want = 1
-				wantLatency = [2]time.Duration{250 * time.Millisecond, 230 * time.Millisecond}
+				wantLatency = [2]time.Duration{35 * time.Millisecond, 20 * time.Millisecond}
 			}
 			if latencies != wantLatency {
 				t.Fatalf("debt=%d acknowledgement latencies=%v want=%v", debt, latencies, wantLatency)
@@ -281,3 +281,49 @@ func TestWriterPressureAdmissionClosesAtEveryBound(t *testing.T) {
 		})
 	}
 }
+
+func TestWriterPressureAdmissionClosesOnIdleOrDeadline(t *testing.T) {
+	for _, interval := range []time.Duration{15 * time.Millisecond, 25 * time.Millisecond} {
+		t.Run(interval.String(), func(t *testing.T) {
+			synctest.Test(t, func(t *testing.T) {
+				c := &pressureRecordingCommitter{recordingCommitter: &recordingCommitter{}}
+				c.debt.Store(telemetry.CompactionDebtLimit / 2)
+				w := testWriter(c, maxGroupBatchRows)
+				w.compaction = c
+				request := submission{batch: testBatch(), ack: make(chan error, 1)}
+				out := make(chan commitJob, 1)
+				done := make(chan error, 1)
+				at := time.Now()
+				var elapsed time.Duration
+				go func() {
+					err := w.enqueueSubmissions(t.Context(), request, out)
+					elapsed = time.Since(at)
+					done <- err
+				}()
+				synctest.Wait()
+				// Keep arrivals below the idle cutoff, or let the first group idle.
+				for i := 1; i <= 16; i++ {
+					time.Sleep(interval)
+					w.submissions <- request
+					synctest.Wait()
+				}
+				if err := <-done; err != nil {
+					t.Fatal(err)
+				}
+				job := <-out
+				wantRequests := 17
+				wantElapsed := pressureAdmissionWindow
+				if interval > pressureAdmissionIdle {
+					wantRequests = 1
+					wantElapsed = pressureAdmissionIdle
+				}
+				if elapsed != wantElapsed {
+					t.Fatalf("dispatch after %v want=%v", elapsed, wantElapsed)
+				}
+				if len(job.acks) != wantRequests {
+					t.Fatalf("requests=%d want=%d", len(job.acks), wantRequests)
+				}
+			})
+		})
+	}
+}