diff --git a/internal/query/duck.go b/internal/query/duck.go index 726869ec..406d6f48 100644 --- a/internal/query/duck.go +++ b/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 diff --git a/internal/query/edge_pass_test.go b/internal/query/edge_pass_test.go index 9e754cf3..4af10537 100644 --- a/internal/query/edge_pass_test.go +++ b/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) } } diff --git a/internal/query/edge_rollup_plan_test.go b/internal/query/edge_rollup_plan_test.go index 357842ff..1586fe82 100644 --- a/internal/query/edge_rollup_plan_test.go +++ b/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) + } +} diff --git a/internal/telemetry/store/backpressure_test.go b/internal/telemetry/store/backpressure_test.go index 2b3d68b5..e34c9231 100644 --- a/internal/telemetry/store/backpressure_test.go +++ b/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) + } + }) + }) + } +} diff --git a/internal/telemetry/store/writer.go b/internal/telemetry/store/writer.go index 9fd95802..296809fd 100644 --- a/internal/telemetry/store/writer.go +++ b/internal/telemetry/store/writer.go @@ -24,6 +24,7 @@ const ( // durable acknowledgement when the row target is not reached. groupAdmissionWindow = 10 * time.Millisecond pressureAdmissionWindow = 250 * time.Millisecond + pressureAdmissionIdle = 20 * time.Millisecond maxAdmissionRequests = 512 maxGroupBatchRows = 50_000 maxCommitWorkers = 4 @@ -227,8 +228,16 @@ func (w *Writer) enqueueSubmissions(ctx context.Context, request submission, out // Coalesce more exports before the hard debt stop is needed. Read // reclaimable debt once per group; low-load acknowledgements keep // their short window as soon as compaction relieves the pressure. + var idle <-chan time.Time + var idleTimer *time.Timer if w.compaction != nil && w.compaction.CompactionDebt() >= telemetry.CompactionDebtLimit/2 { window = pressureAdmissionWindow + // Closed-loop exporters cannot send again until this group commits. + // Close an idle group promptly, while a paced stream can continue + // coalescing up to the pressure window and the existing bounds. + idleTimer = time.NewTimer(pressureAdmissionIdle) + defer idleTimer.Stop() + idle = idleTimer.C } timer := time.NewTimer(window) defer timer.Stop() @@ -239,6 +248,11 @@ func (w *Writer) enqueueSubmissions(ctx context.Context, request submission, out requests = append(requests, next) rows += batchRows(next.batch) bytes += next.bytes + if idleTimer != nil { + idleTimer.Reset(pressureAdmissionIdle) + } + case <-idle: + break admit case <-timer.C: break admit case <-ctx.Done():