From 77b88d8bcddb36398a4b4d30f1963c471e5070b6 Mon Sep 17 00:00:00 2001 From: Vishal Rana Date: Thu, 15 Oct 2026 12:44:54 -0700 Subject: [PATCH 1/2] perf: merge Parquet by column, bound pressured merges, and free rollup gates An 8-vCPU benchmark of v2026.10.5-alpha failed its gates. A compaction merge of 128 log batches ran on one core at about 52k rows/s, against 250-380k logs/s of ingest. Generation-0 debt reached the hard limit, ingest stopped for about 100 s, and generators timed out. - Compaction merges each physical column with one shared row schedule. Span sort keys are merged once and drive the schedule. Up to eight columns decode, encode and compress concurrently, with no wide-row conversion. Locally, 6.4M-row merges went from 575k to 2.36M rows/s for logs and metrics, and from 233k to 794k for spans. Logical values, physical schema, row groups, footer bounds and the trace index are unchanged. - While reclaimable debt is at or above half the limit, a generation-0 merge takes at most 4M rows, so debt retires in steps of seconds. Inputs above half that ceiling no longer count toward blocking debt. - Service, edge and version passes no longer park the analytical gate: when background Parquet admission is unavailable they release it and retry on their cadence. - The service rollup materialises its affected keys once and binds exact native event-time bounds. A dense 15M-row minute went from 2.9 s to 0.5 s and from 954 MiB to 52 MiB peak. Claude-Session: https://claude.ai/code/session_01XA52nWyQ3R5f5kK8kjpAQW --- internal/query/annotations.go | 5 +- internal/query/background_admission_test.go | 9 +- internal/query/backlog_pressure_test.go | 4 +- internal/query/duck.go | 142 +++---- internal/query/duck_test.go | 30 +- internal/query/parquet_gate.go | 6 - internal/query/rollup_admission_test.go | 64 ++++ internal/query/service_bounds_test.go | 67 ++++ internal/query/version_budget_test.go | 46 +-- internal/query/version_gate_test.go | 212 ++++------- internal/telemetry/compaction_debt.go | 18 +- internal/telemetry/compaction_debt_test.go | 2 +- internal/telemetry/compaction_index_test.go | 4 +- internal/telemetry/parquet.go | 141 ------- internal/telemetry/parquet_merge.go | 345 ++++++++++++++++++ internal/telemetry/store/compaction.go | 14 +- .../telemetry/store/compaction_bytes_test.go | 4 +- .../store/compaction_pressure_test.go | 40 ++ .../store/compaction_starvation_test.go | 4 +- 19 files changed, 731 insertions(+), 426 deletions(-) create mode 100644 internal/query/rollup_admission_test.go create mode 100644 internal/query/service_bounds_test.go create mode 100644 internal/telemetry/parquet_merge.go create mode 100644 internal/telemetry/store/compaction_pressure_test.go diff --git a/internal/query/annotations.go b/internal/query/annotations.go index 771acef5..8885b446 100644 --- a/internal/query/annotations.go +++ b/internal/query/annotations.go @@ -144,9 +144,12 @@ func (d *Duck) refreshVersionRollup(ctx context.Context, waitBudget time.Duratio return 0, fmt.Errorf("version_rollup/write_gate: %w", err) } defer unlock() - if err := d.lockRollupParquetRead(waitCtx); err != nil { + if err := waitCtx.Err(); err != nil { return 0, err } + if !d.tryRollupParquetRead() { + return 0, nil + } defer d.parquetMu.RUnlock() cancelWait() ctx, cancel := context.WithTimeout(ctx, 10*time.Second) diff --git a/internal/query/background_admission_test.go b/internal/query/background_admission_test.go index 7c962663..9d5bbc15 100644 --- a/internal/query/background_admission_test.go +++ b/internal/query/background_admission_test.go @@ -56,8 +56,13 @@ func TestBackgroundPassesYieldToQueuedPublication(t *testing.T) { t.Run(name, func(t *testing.T) { waitCtx, stop := context.WithTimeout(t.Context(), 20*time.Millisecond) defer stop() - if err := pass(waitCtx); !errors.Is(err, ErrParquetReadWait) || !errors.Is(err, context.DeadlineExceeded) { - t.Fatalf("background pass did not wait behind publisher: %v", err) + err := pass(waitCtx) + if name == "stats" { + if !errors.Is(err, ErrParquetReadWait) || !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("stats did not wait behind publisher: %v", err) + } + } else if err != nil { + t.Fatalf("analytical pass did not skip queued publication: %v", err) } }) } diff --git a/internal/query/backlog_pressure_test.go b/internal/query/backlog_pressure_test.go index 4d683f16..7371a6d4 100644 --- a/internal/query/backlog_pressure_test.go +++ b/internal/query/backlog_pressure_test.go @@ -331,11 +331,13 @@ func TestBacklogServiceFailuresIdentifyStatementAndPreserveCause(t *testing.T) { mock.ExpectQuery("FROM rollup_state").WithArgs(key).WillReturnRows(sqlmock.NewRows([]string{"watermark"}).AddRow(1)) } mock.ExpectBegin() + mock.ExpectExec("CREATE OR REPLACE TEMP TABLE service_pass_affected").WillReturnResult(sqlmock.NewResult(0, 0)) + mock.ExpectQuery("SELECT epoch_ns").WillReturnRows(sqlmock.NewRows([]string{"lo", "hi"}).AddRow(0, 60_000_000_000)) if label == "service_rollup/delete" { mock.ExpectExec("DELETE FROM service_rollup").WillReturnError(cause) } else { mock.ExpectExec("DELETE FROM service_rollup").WillReturnResult(sqlmock.NewResult(0, 0)) - mock.ExpectExec("WITH affected").WillReturnError(cause) + mock.ExpectExec("WITH span_agg").WillReturnError(cause) } mock.ExpectRollback() d := &Duck{DB: db, writeDB: db, publicationBounds: func() (int64, int64, int64) { return 2, 2, 1 }} diff --git a/internal/query/duck.go b/internal/query/duck.go index 406d6f48..6820a46f 100644 --- a/internal/query/duck.go +++ b/internal/query/duck.go @@ -790,6 +790,14 @@ func (d *Duck) refreshServiceRollup(ctx context.Context) (int64, error) { // consistent and deadlock-free. unlock := d.writeGate.Lock(writegate.WriteRollupService) defer unlock() + if err := ctx.Err(); err != nil { + return 0, err + } + if !d.tryRollupParquetRead() { + result = metrics.RollupNoop + return 0, nil + } + defer d.parquetMu.RUnlock() horizon, published, minimum := d.publicationBounds() sourceMax = published // Failed attempts advance the horizon without publishing rows. Until the @@ -810,10 +818,6 @@ func (d *Duck) refreshServiceRollup(ctx context.Context) (int64, error) { updateRollupProgress(metrics.RollupService, true, lastWatermark, published) return 0, nil } - if err := d.lockRollupParquetRead(ctx); err != nil { - return 0, err - } - defer d.parquetMu.RUnlock() tx, err := d.writer().BeginTx(ctx, nil) if err != nil { return 0, err @@ -823,12 +827,32 @@ func (d *Duck) refreshServiceRollup(ctx context.Context) (int64, error) { newWatermark := windowEnd args := []any{windowStart, windowEnd, windowStart, windowEnd, windowStart, windowEnd} - if _, err := tx.ExecContext(ctx, serviceRollupDeleteSQL, args...); err != nil { - return 0, fmt.Errorf("service_rollup/delete: %w", err) + if _, err := tx.ExecContext(ctx, createServicePassAffectedSQL, args...); err != nil { + return 0, fmt.Errorf("service_rollup/affected: %w", err) + } + var lower, upper sql.NullInt64 + if err := tx.QueryRowContext(ctx, `SELECT epoch_ns(min(bucket)), epoch_ns(max(bucket) + INTERVAL 1 MINUTE) FROM service_pass_affected`).Scan(&lower, &upper); err != nil { + return 0, fmt.Errorf("service_rollup/bounds: %w", err) + } + var rows int64 + // An empty ingestion window still advances its watermark, but cannot + // delete or insert any buckets. Keep exact signed nanoseconds in Go. + if lower.Valid && upper.Valid { + lo, hi := time.Unix(0, lower.Int64).UTC(), time.Unix(0, upper.Int64).UTC() + if _, err := tx.ExecContext(ctx, serviceRollupDeleteSQL, lo, hi); err != nil { + return 0, fmt.Errorf("service_rollup/delete: %w", err) + } + res, err := tx.ExecContext(ctx, serviceRollupInsertSQL, lo, hi, lo, hi, lo, hi) + if err != nil { + return 0, fmt.Errorf("service_rollup/insert: %w", err) + } + rows, err = res.RowsAffected() + if err != nil { + slog.Warn("service rollup rows affected unavailable", "err", err) + } } - res, err := tx.ExecContext(ctx, serviceRollupInsertSQL, args...) - if err != nil { - return 0, fmt.Errorf("service_rollup/insert: %w", err) + if _, err := tx.ExecContext(ctx, `DROP TABLE service_pass_affected`); err != nil { + return 0, fmt.Errorf("service_rollup/drop_affected: %w", err) } if err := storeRollupWatermark(ctx, tx, serviceRollupStateKey, newWatermark); err != nil { @@ -841,14 +865,6 @@ func (d *Duck) refreshServiceRollup(ctx context.Context) (int64, error) { d.rollupWatermarks[serviceRollupStateKey] = newWatermark updateRollupProgress(metrics.RollupService, true, newWatermark, published) - rows, err := res.RowsAffected() - if err != nil { - // Warn, not Debug: recordedRows feeds fanout_rollup_component_rows_total, - // so a silent zero here reads in benchmark evidence as "the rollup ran and - // matched nothing" when it actually means "we do not know". - slog.Warn("service rollup rows affected unavailable", "err", err) - return 0, nil - } recordedRows = rows return rows, nil } @@ -887,6 +903,14 @@ func (d *Duck) refreshEdgeRollupPass(ctx context.Context) (int64, bool, error) { }() unlock := d.writeGate.Lock(writegate.WriteRollupEdge) defer unlock() + if err := ctx.Err(); err != nil { + return 0, false, err + } + if !d.tryRollupParquetRead() { + result = metrics.RollupNoop + return 0, false, nil + } + defer d.parquetMu.RUnlock() horizon, published, minimum := d.publicationBounds() sourceMax = published // Failed attempts advance the horizon without publishing rows. Until the @@ -907,10 +931,6 @@ func (d *Duck) refreshEdgeRollupPass(ctx context.Context) (int64, bool, error) { updateRollupProgress(metrics.RollupEdge, true, lastWatermark, published) return 0, false, nil } - if err := d.lockRollupParquetRead(ctx); err != nil { - return 0, false, err - } - defer d.parquetMu.RUnlock() tx, err := d.writer().BeginTx(ctx, nil) if err != nil { return 0, false, err @@ -1253,8 +1273,8 @@ func serviceRollupQuantileSQL(alias string, index int, column string) string { return fmt.Sprintf("COALESCE(%s[%d], %s[%d]) AS %s", served, index, all, index, column) } -const serviceRollupDeleteSQL = ` -WITH affected AS ( +const createServicePassAffectedSQL = ` +CREATE OR REPLACE TEMP TABLE service_pass_affected AS SELECT DISTINCT namespace, date_trunc('minute', start_time::TIMESTAMP_NS) AS bucket, service FROM spans WHERE ingested_unix_nano > ? @@ -1277,44 +1297,21 @@ WITH affected AS ( AND ingested_unix_nano <= ? AND time IS NOT NULL AND service IS NOT NULL - AND service != '' -) + AND service != '';` + +const serviceRollupDeleteSQL = ` DELETE FROM service_rollup -WHERE EXISTS ( - SELECT 1 - FROM affected - WHERE affected.namespace = service_rollup.namespace - AND affected.bucket = service_rollup.bucket - AND affected.service = service_rollup.service +WHERE bucket >= ?::TIMESTAMP_NS::TIMESTAMPTZ_NS + AND bucket < ?::TIMESTAMP_NS::TIMESTAMPTZ_NS + AND EXISTS ( + SELECT 1 FROM service_pass_affected a + WHERE a.namespace = service_rollup.namespace + AND a.bucket = service_rollup.bucket + AND a.service = service_rollup.service );` var serviceRollupInsertSQL = ` -WITH affected AS ( - SELECT DISTINCT namespace, date_trunc('minute', start_time::TIMESTAMP_NS) AS bucket, service - FROM spans - WHERE ingested_unix_nano > ? - AND ingested_unix_nano <= ? - AND start_time IS NOT NULL - AND service IS NOT NULL - AND service != '' - UNION - SELECT DISTINCT namespace, date_trunc('minute', time::TIMESTAMP_NS) AS bucket, service - FROM logs - WHERE ingested_unix_nano > ? - AND ingested_unix_nano <= ? - AND time IS NOT NULL - AND service IS NOT NULL - AND service != '' - UNION - SELECT DISTINCT namespace, date_trunc('minute', time::TIMESTAMP_NS) AS bucket, service - FROM metrics - WHERE ingested_unix_nano > ? - AND ingested_unix_nano <= ? - AND time IS NOT NULL - AND service IS NOT NULL - AND service != '' -), -span_agg AS ( +WITH span_agg AS ( SELECT s.namespace, date_trunc('minute', s.start_time::TIMESTAMP_NS) AS bucket, @@ -1336,7 +1333,7 @@ span_agg AS ( ` + serviceRollupP95SQL("s") + `, avg(CASE WHEN s.status IN ('STATUS_CODE_ERROR', 'ERROR') THEN 1.0 ELSE 0.0 END) AS error_rate FROM spans s - JOIN affected a + JOIN service_pass_affected a ON a.namespace = s.namespace AND a.bucket = date_trunc('minute', s.start_time::TIMESTAMP_NS) AND a.service = s.service @@ -1347,8 +1344,8 @@ span_agg AS ( -- [MIN(bucket), MAX(bucket)+1min) — so it drops no rows. A late span with an -- old start_time widens the range (and this scan) only for the pass that -- ingests it, same trade-off as the edge rollup's parent bound. - WHERE s.start_time >= ((SELECT MIN(bucket) FROM affected))::TIMESTAMP_NS::TIMESTAMPTZ_NS - AND s.start_time < ((SELECT MAX(bucket) FROM affected) + INTERVAL 1 MINUTE)::TIMESTAMP_NS::TIMESTAMPTZ_NS + WHERE s.start_time >= ?::TIMESTAMP_NS::TIMESTAMPTZ_NS + AND s.start_time < ?::TIMESTAMP_NS::TIMESTAMPTZ_NS GROUP BY s.namespace, date_trunc('minute', s.start_time::TIMESTAMP_NS), s.service ), log_agg AS ( @@ -1358,12 +1355,12 @@ log_agg AS ( l.service, COUNT(*) AS log_count FROM logs l - JOIN affected a + JOIN service_pass_affected a ON a.namespace = l.namespace AND a.bucket = date_trunc('minute', l.time::TIMESTAMP_NS) AND a.service = l.service - WHERE l.time >= ((SELECT MIN(bucket) FROM affected))::TIMESTAMP_NS::TIMESTAMPTZ_NS - AND l.time < ((SELECT MAX(bucket) FROM affected) + INTERVAL 1 MINUTE)::TIMESTAMP_NS::TIMESTAMPTZ_NS + WHERE l.time >= ?::TIMESTAMP_NS::TIMESTAMPTZ_NS + AND l.time < ?::TIMESTAMP_NS::TIMESTAMPTZ_NS GROUP BY l.namespace, date_trunc('minute', l.time::TIMESTAMP_NS), l.service ), metric_agg AS ( @@ -1373,12 +1370,12 @@ metric_agg AS ( m.service, COUNT(DISTINCT m.name) AS metric_count FROM metrics m - JOIN affected a + JOIN service_pass_affected a ON a.namespace = m.namespace AND a.bucket = date_trunc('minute', m.time::TIMESTAMP_NS) AND a.service = m.service - WHERE m.time >= ((SELECT MIN(bucket) FROM affected))::TIMESTAMP_NS::TIMESTAMPTZ_NS - AND m.time < ((SELECT MAX(bucket) FROM affected) + INTERVAL 1 MINUTE)::TIMESTAMP_NS::TIMESTAMPTZ_NS + WHERE m.time >= ?::TIMESTAMP_NS::TIMESTAMPTZ_NS + AND m.time < ?::TIMESTAMP_NS::TIMESTAMPTZ_NS GROUP BY m.namespace, date_trunc('minute', m.time::TIMESTAMP_NS), m.service ) INSERT INTO service_rollup ( @@ -1404,7 +1401,7 @@ SELECT COALESCE(s.error_rate, 0), COALESCE(l.log_count, 0), COALESCE(m.metric_count, 0) -FROM affected a +FROM service_pass_affected a LEFT JOIN span_agg s USING (namespace, bucket, service) LEFT JOIN log_agg l USING (namespace, bucket, service) LEFT JOIN metric_agg m USING (namespace, bucket, service);` @@ -1925,10 +1922,13 @@ func (d *Duck) lockParquetRead(ctx context.Context, reader string) error { return nil } -func (d *Duck) lockRollupParquetRead(ctx context.Context) error { - waitCtx, cancel := context.WithTimeout(ctx, rollupReaderLease) - defer cancel() - return d.lockParquetRead(waitCtx, readerRollup) +// Keep gate-first lock ordering, but never park the analytical writer gate +// behind a publication. The owning loop retries skipped work on its cadence. +func (d *Duck) tryRollupParquetRead() bool { + start := time.Now() + admitted := d.parquetMu.TryRLock(parquetReadBackground) + metrics.RecordParquetRead(readerRollup, time.Since(start).Seconds(), admitted) + return admitted } // ---- Queries for API ---- diff --git a/internal/query/duck_test.go b/internal/query/duck_test.go index 0bb84d2e..67b53a6f 100644 --- a/internal/query/duck_test.go +++ b/internal/query/duck_test.go @@ -423,38 +423,40 @@ func TestWaitingMaintenanceDoesNotBlockNewReaders(t *testing.T) { } } -func TestRollupReadLockHonorsContext(t *testing.T) { +func TestRollupReadAdmissionSkipsPublication(t *testing.T) { d := &Duck{publicationBounds: func() (int64, int64, int64) { return 2, 2, 1 }, rollupWatermarks: map[string]int64{serviceRollupStateKey: 1}} mustLock(&d.parquetMu) - ctx, cancel := context.WithTimeout(context.Background(), 25*time.Millisecond) - defer cancel() - _, err := d.refreshServiceRollup(ctx) - d.parquetMu.Unlock() - if !errors.Is(err, context.DeadlineExceeded) { - t.Fatalf("refreshServiceRollup error = %v, want deadline exceeded", err) + defer d.parquetMu.Unlock() + if rows, err := d.refreshServiceRollup(t.Context()); err != nil || rows != 0 { + t.Fatalf("service skip rows=%d error=%v", rows, err) + } + if rows, pending, err := d.refreshEdgeRollupPass(t.Context()); err != nil || rows != 0 || pending { + t.Fatalf("edge skip rows=%d pending=%v error=%v", rows, pending, err) + } + if d.rollupWatermarks[serviceRollupStateKey] != 1 { + t.Fatal("skipped pass moved watermark") } } -func TestRollupAdmissionLeaseDoesNotCancelAdmittedWork(t *testing.T) { +func TestRollupAdmissionDoesNotCancelAdmittedWork(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatal(err) } defer db.Close() - originalLease := rollupReaderLease - rollupReaderLease = 5 * time.Millisecond - t.Cleanup(func() { rollupReaderLease = originalLease }) - mock.ExpectQuery("FROM rollup_state").WithArgs(serviceRollupStateKey). WillReturnRows(sqlmock.NewRows([]string{"watermark"}).AddRow(100)) mock.ExpectBegin() + mock.ExpectExec("CREATE OR REPLACE TEMP TABLE service_pass_affected").WillReturnResult(sqlmock.NewResult(0, 0)) + mock.ExpectQuery("SELECT epoch_ns").WillReturnRows(sqlmock.NewRows([]string{"lo", "hi"}).AddRow(0, 60_000_000_000)) mock.ExpectExec("DELETE FROM service_rollup").WillDelayFor(20 * time.Millisecond).WillReturnResult(sqlmock.NewResult(0, 0)) - mock.ExpectExec("WITH affected").WillReturnResult(sqlmock.NewResult(0, 0)) + mock.ExpectExec("WITH span_agg").WillReturnResult(sqlmock.NewResult(0, 0)) + mock.ExpectExec("DROP TABLE service_pass_affected").WillReturnResult(sqlmock.NewResult(0, 0)) mock.ExpectExec("INSERT INTO rollup_state").WillReturnResult(sqlmock.NewResult(0, 1)) mock.ExpectCommit() d := &Duck{DB: db, writeDB: db, publicationBounds: func() (int64, int64, int64) { return 200, 200, 100 }} if _, err := d.refreshServiceRollup(context.Background()); err != nil { - t.Fatalf("admitted rollup was canceled by its admission lease: %v", err) + t.Fatal(err) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatal(err) diff --git a/internal/query/parquet_gate.go b/internal/query/parquet_gate.go index 9670160d..451fd881 100644 --- a/internal/query/parquet_gate.go +++ b/internal/query/parquet_gate.go @@ -12,12 +12,6 @@ import ( // behind maintenance, short enough that retention and compaction always run. const defaultWriterGrace = 30 * time.Second -// rollupReaderLease bounds only how long rebuildable rollup work waits to enter -// the Parquet snapshot. Once admitted, the bounded rollup chunk runs under its -// caller context so a slow but healthy pass can commit instead of retrying the -// same chunk forever. -var rollupReaderLease = defaultWriterGrace / 2 - // ErrParquetReadWait distinguishes publication contention from a query error. var ErrParquetReadWait = errors.New("wait for Parquet publication") diff --git a/internal/query/rollup_admission_test.go b/internal/query/rollup_admission_test.go new file mode 100644 index 00000000..e1e69603 --- /dev/null +++ b/internal/query/rollup_admission_test.go @@ -0,0 +1,64 @@ +package query + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/labstack/fanout/internal/annotations" +) + +func TestAnalyticalRollupsReleaseGateWhenPublicationBlocksAdmission(t *testing.T) { + for _, component := range []string{"service", "edge", "version"} { + t.Run(component, func(t *testing.T) { + d, repo := versionEngine(t) + commitVersionLogs(t, repo, "pending", time.Now().UTC().Add(-time.Hour), 2) + refresh := func() (int64, error) { + switch component { + case "service": + return d.refreshServiceRollup(t.Context()) + case "edge": + n, _, err := d.refreshEdgeRollupPass(t.Context()) + return n, err + default: + return d.RefreshVersionRollup(t.Context()) + } + } + if err := d.parquetMu.LockContext(t.Context()); err != nil { + t.Fatal(err) + } + locked := true + defer func() { + if locked { + d.parquetMu.Unlock() + } + }() + before := fmt.Sprint(d.rollupWatermarks) + for range 3 { + if n, err := refresh(); err != nil || n != 0 { + t.Fatalf("skipped pass rows=%d error=%v", n, err) + } + } + if fmt.Sprint(d.rollupWatermarks) != before || len(d.versionRollupFailures) != 0 { + t.Fatal("skipped admission moved progress or penalized a batch") + } + var marked int + if err := d.writer().QueryRow(`SELECT count(*) FROM version_rollup_batches`).Scan(&marked); err != nil || marked != 0 { + t.Fatalf("skipped version markers=%d error=%v", marked, err) + } + ctx, cancel := context.WithTimeout(t.Context(), 50*time.Millisecond) + defer cancel() + start := time.Now() + if err := d.RecordAnomalies(ctx, []annotations.Anomaly{}, time.Now()); err != nil { + t.Fatalf("anomaly parked behind %s admission: %v", component, err) + } + t.Logf("anomaly persisted with publication held: %s", time.Since(start)) + d.parquetMu.Unlock() + locked = false + if n, err := refresh(); err != nil || component != "edge" && n == 0 { + t.Fatalf("retry rows=%d error=%v", n, err) + } + }) + } +} diff --git a/internal/query/service_bounds_test.go b/internal/query/service_bounds_test.go new file mode 100644 index 00000000..6f777a7b --- /dev/null +++ b/internal/query/service_bounds_test.go @@ -0,0 +1,67 @@ +package query + +import ( + "testing" + "time" + + "github.com/DATA-DOG/go-sqlmock" +) + +func TestEmptyAffectedServiceSetAdvancesOnlyWatermark(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatal(err) + } + defer db.Close() + mock.ExpectQuery("FROM rollup_state").WithArgs(serviceRollupStateKey).WillReturnRows(sqlmock.NewRows([]string{"watermark"}).AddRow(100)) + mock.ExpectBegin() + mock.ExpectExec("CREATE OR REPLACE TEMP TABLE service_pass_affected").WithArgs(int64(100), int64(200), int64(100), int64(200), int64(100), int64(200)).WillReturnResult(sqlmock.NewResult(0, 0)) + mock.ExpectQuery("SELECT epoch_ns").WillReturnRows(sqlmock.NewRows([]string{"lo", "hi"}).AddRow(nil, nil)) + mock.ExpectExec("DROP TABLE service_pass_affected").WillReturnResult(sqlmock.NewResult(0, 0)) + mock.ExpectExec("INSERT INTO rollup_state").WithArgs(serviceRollupStateKey, int64(200)).WillReturnResult(sqlmock.NewResult(0, 1)) + mock.ExpectCommit() + d := &Duck{DB: db, writeDB: db, publicationBounds: func() (int64, int64, int64) { return 200, 200, 100 }} + if n, err := d.refreshServiceRollup(t.Context()); err != nil || n != 0 { + t.Fatalf("empty pass rows=%d error=%v", n, err) + } + if d.rollupWatermarks[serviceRollupStateKey] != 200 { + t.Fatal("empty pass did not advance watermark") + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatal(err) + } +} + +func TestServiceBoundsRebuildSignedBoundaryMinutes(t *testing.T) { + db := openTestDuck(t) + if err := CreateTables(db); err != nil { + t.Fatal(err) + } + if err := CreateViews(db); err != nil { + t.Fatal(err) + } + d := &Duck{DB: db, writeDB: db, publicationBounds: tablePublicationBounds(db)} + base := time.Unix(0, 0).UTC().Add(-time.Minute) + insert := func(id string, stamp time.Time, ingest int64) { + insertRollupTestSpan(t, db, rollupTestSpan{namespace: "ns", traceID: id, spanID: id, service: "api", start: stamp, duration: time.Millisecond, ingested: ingest}) + // Bind the signed fixture as an integer through make_timestamp_ns; + // the generic test insertion helper has no explicit parameter cast. + if _, err := db.Exec(`UPDATE telemetry.spans SET start_time=make_timestamp_ns(?) WHERE span_id=?`, stamp.UnixNano(), id); err != nil { + t.Fatal(err) + } + } + insert("signed-first", base, 100) + insert("signed-last", base.Add(time.Minute-time.Microsecond), 100) + insert("upper-edge", base.Add(time.Minute), 100) + if _, err := d.refreshServiceRollup(t.Context()); err != nil { + t.Fatal(err) + } + requireServiceRollupSpans(t, db, "ns", base, "api", 2) + requireServiceRollupSpans(t, db, "ns", base.Add(time.Minute), "api", 1) + insert("late", base.Add(time.Nanosecond), 200) + if _, err := d.refreshServiceRollup(t.Context()); err != nil { + t.Fatal(err) + } + requireServiceRollupSpans(t, db, "ns", base, "api", 3) + requireServiceRollupSpans(t, db, "ns", base.Add(time.Minute), "api", 1) +} diff --git a/internal/query/version_budget_test.go b/internal/query/version_budget_test.go index cd590db9..20bf4f8d 100644 --- a/internal/query/version_budget_test.go +++ b/internal/query/version_budget_test.go @@ -3,8 +3,6 @@ package query import ( "context" "fmt" - "runtime" - "strings" "testing" "time" @@ -93,8 +91,8 @@ func TestVersionMixedBacklogDrainsAndAllowsAnomalyPersistence(t *testing.T) { if err := d.DB.QueryRow(`SELECT count(*) FROM version_rollup`).Scan(&versions); err != nil || versions != 120 { t.Fatalf("versions=%d err=%v", versions, err) } - // Requeue the same real backlog and park the production version loop after - // it owns writeGate, so the anomaly writer demonstrably contends with it. + // Requeue the real backlog while publication blocks admission. Anomalies + // must persist before publication releases, and the loop must retry later. if _, err := d.writer().Exec(`DELETE FROM version_rollup_batches`); err != nil { t.Fatal(err) } @@ -111,15 +109,14 @@ func TestVersionMixedBacklogDrainsAndAllowsAnomalyPersistence(t *testing.T) { done := make(chan struct{}) go func() { defer close(done); d.runVersionRollupLoop(ctx) }() defer func() { cancel(); <-done }() - waitForVersionBudgetStack(t, "query.(*Duck).runVersionRollupLoop", "query.(*parquetReadGate).RLockContext") + if n, err := d.RefreshVersionRollup(ctx); err != nil || n != 0 { + t.Fatalf("publication skip=%d %v", n, err) + } now := time.Now().UTC() finding := annotations.Anomaly{Namespace: "namespace-0", Service: "service-0", Kind: "volume", From: now.Add(-time.Minute), To: now, Title: "Low volume", Severity: "warn"} persisted := make(chan error, 1) go func() { persisted <- d.RecordAnomalies(ctx, []annotations.Anomaly{finding}, now) }() - waitForVersionBudgetStack(t, "query.(*Duck).RecordAnomalies", "writegate.(*WriteGate).LockContext") start := time.Now() - d.parquetMu.Unlock() - locked = false select { case err := <-persisted: if err != nil { @@ -128,34 +125,25 @@ func TestVersionMixedBacklogDrainsAndAllowsAnomalyPersistence(t *testing.T) { case <-time.After(5 * time.Second): t.Fatal("anomaly persistence starved behind busy version loop") } - t.Logf("anomaly persisted behind productive version pass in %v", time.Since(start)) + d.parquetMu.Unlock() + locked = false + t.Logf("anomaly persisted during blocked Parquet admission in %v", time.Since(start)) var count int if err := d.DB.QueryRow(`SELECT count(*) FROM anomaly_log WHERE title='Low volume'`).Scan(&count); err != nil || count != 1 { t.Fatalf("persisted=%d err=%v", count, err) } - if err := d.DB.QueryRow(`SELECT count(*) FROM version_rollup_batches`).Scan(&marked); err != nil || marked == 0 { - t.Fatalf("version loop made no progress: %d %v", marked, err) - } -} - -func waitForVersionBudgetStack(t *testing.T, caller, frame string) { - t.Helper() - deadline := time.NewTimer(3 * time.Second) - defer deadline.Stop() - ticker := time.NewTicker(time.Millisecond) - defer ticker.Stop() - buffer := make([]byte, 1<<20) + deadline := time.Now().Add(10 * time.Second) for { - for stack := range strings.SplitSeq(string(buffer[:runtime.Stack(buffer, true)]), "\n\n") { - if strings.Contains(stack, "[select]") && strings.Contains(stack, caller) && strings.Contains(stack, frame) { - return - } + if err := d.DB.QueryRow(`SELECT count(*) FROM version_rollup_batches`).Scan(&marked); err != nil { + t.Fatal(err) + } + if marked > 0 { + break } - select { - case <-deadline.C: - t.Fatalf("never observed %s waiting in %s", caller, frame) - case <-ticker.C: + if time.Now().After(deadline) { + t.Fatal("version loop did not retry after publication") } + time.Sleep(10 * time.Millisecond) } } diff --git a/internal/query/version_gate_test.go b/internal/query/version_gate_test.go index 50abab4e..ece97cec 100644 --- a/internal/query/version_gate_test.go +++ b/internal/query/version_gate_test.go @@ -48,159 +48,85 @@ func TestGateTimeoutDoesNotBackOff(t *testing.T) { } func TestVersionRollupWaitIsBoundedWithoutCallerDeadline(t *testing.T) { - for _, gate := range []string{"write", "parquet"} { - t.Run(gate, func(t *testing.T) { - d, repo := versionEngine(t) - commitVersionLogs(t, repo, "healthy", time.Now().UTC().Add(-time.Hour), 1) - acquired, release, released := make(chan struct{}), make(chan struct{}), make(chan struct{}) - go func() { - if gate == "write" { - unlock := d.writeGate.Lock(writegate.WriteRollupService) - close(acquired) - <-release - unlock() - } else { - if err := d.parquetMu.LockContext(context.Background()); err != nil { - panic(err) - } - close(acquired) - <-release - d.parquetMu.Unlock() - } - close(released) - }() - <-acquired - // A watchdog releases the test fixture even if the implementation has no bound. - finished := make(chan struct{}) - go func() { - select { - case <-finished: - case <-time.After(time.Second): - } - close(release) - }() - counter := metrics.RollupComponentTotal.WithLabelValues("version", "error") - before := testutil.ToFloat64(counter) - var log bytes.Buffer - logger := slog.Default() - slog.SetDefault(slog.New(slog.NewTextHandler(&log, nil))) - defer slog.SetDefault(logger) - start := time.Now() - _, err := d.refreshVersionRollup(context.Background(), 20*time.Millisecond) - elapsed := time.Since(start) - close(finished) - <-released - if !errors.Is(err, context.DeadlineExceeded) || elapsed > 500*time.Millisecond { - t.Fatalf("unbounded %s wait: %v after %s", gate, err, elapsed) - } - if gate == "parquet" && !errors.Is(err, ErrParquetReadWait) { - t.Fatalf("timeout never reached parquet gate: %v", err) - } - if after := testutil.ToFloat64(counter); after != before+1 { - t.Fatalf("wait-budget error counter=%v want %v", after, before+1) - } - if output := log.String(); !strings.Contains(output, "level=WARN") || !strings.Contains(output, "version rollup pass timed out") { - t.Fatalf("missing wait-budget timeout warning: %s", output) - } - if len(d.versionRollupFailures) != 0 { - t.Fatalf("wait penalized batches: %+v", d.versionRollupFailures) - } - ctx, cancel := context.WithTimeout(t.Context(), time.Second) - defer cancel() - // Taking both locks after the failed second acquisition proves the first was released. - unlock, err := d.writeGate.LockContext(ctx, writegate.WriteRollupVersion) - if err != nil { - t.Fatalf("leaked write lock: %v", err) - } - defer unlock() - if err := d.parquetMu.LockContext(ctx); err != nil { - t.Fatalf("leaked parquet lock: %v", err) - } - d.parquetMu.Unlock() - }) + d, repo := versionEngine(t) + commitVersionLogs(t, repo, "healthy", time.Now().UTC().Add(-time.Hour), 1) + unlock := d.writeGate.Lock(writegate.WriteRollupService) + var releaseOnce sync.Once + release := func() { releaseOnce.Do(unlock) } + defer release() + watchdog := time.AfterFunc(time.Second, release) + defer watchdog.Stop() + counter := metrics.RollupComponentTotal.WithLabelValues("version", "error") + before := testutil.ToFloat64(counter) + var log bytes.Buffer + logger := slog.Default() + slog.SetDefault(slog.New(slog.NewTextHandler(&log, nil))) + defer slog.SetDefault(logger) + start := time.Now() + _, err := d.refreshVersionRollup(context.Background(), 20*time.Millisecond) + elapsed := time.Since(start) + release() + if !errors.Is(err, context.DeadlineExceeded) || elapsed > 500*time.Millisecond { + t.Fatalf("unbounded write wait: %v after %s", err, elapsed) + } + if after := testutil.ToFloat64(counter); after != before+1 { + t.Fatalf("error counter=%v want %v", after, before+1) + } + if output := log.String(); !strings.Contains(output, "level=WARN") || !strings.Contains(output, "version rollup pass timed out") { + t.Fatalf("missing timeout warning: %s", output) + } + if len(d.versionRollupFailures) != 0 { + t.Fatal("gate wait penalized batches") + } + if _, err := d.RefreshVersionRollup(t.Context()); err != nil { + t.Fatalf("gate leaked: %v", err) } } func TestVersionRollupWaitHonorsCallerCancellation(t *testing.T) { - for _, gate := range []string{"write", "parquet"} { - t.Run(gate, func(t *testing.T) { - d, _ := versionEngine(t) - acquired, release, released := make(chan struct{}), make(chan struct{}), make(chan struct{}) - go func() { - if gate == "write" { - unlock := d.writeGate.Lock(writegate.WriteRollupService) - close(acquired) - <-release - unlock() - } else { - if err := d.parquetMu.LockContext(context.Background()); err != nil { - panic(err) - } - close(acquired) - <-release - d.parquetMu.Unlock() - } - close(released) - }() - <-acquired - ctx, cancel := context.WithCancel(t.Context()) - defer cancel() - var releaseOnce sync.Once - releaseFixture := func() { releaseOnce.Do(func() { close(release) }) } - defer releaseFixture() - watchdog := time.AfterFunc(2*time.Second, func() { cancel(); releaseFixture() }) - defer watchdog.Stop() - finished := make(chan error, 1) - go func() { - _, err := d.refreshVersionRollup(ctx, time.Second) - finished <- err - }() - blocked := make(chan struct{}) - stop := make(chan struct{}) - defer close(stop) - go signalVersionGateWait(gate, blocked, stop) - select { - case <-blocked: - cancel() - case err := <-finished: - t.Fatalf("pass returned before observed %s wait: %v", gate, err) - case <-time.After(2 * time.Second): - t.Fatalf("watchdog: pass never blocked on %s gate", gate) - } - var err error - select { - case err = <-finished: - case <-time.After(2 * time.Second): - t.Fatalf("watchdog: %s gate ignored caller cancellation", gate) - } - releaseFixture() - <-released - if !errors.Is(err, context.Canceled) { - t.Fatalf("caller cancellation: %v", err) - } - if gate == "parquet" && !errors.Is(err, ErrParquetReadWait) { - t.Fatalf("cancellation never reached parquet gate: %v", err) - } - if len(d.versionRollupFailures) != 0 { - t.Fatal("cancellation penalized a batch") - } - ctx, cancel = context.WithTimeout(t.Context(), time.Second) - defer cancel() - if _, err := d.RefreshVersionRollup(ctx); err != nil { - t.Fatalf("lock leaked after cancellation: %v", err) - } - }) + d, repo := versionEngine(t) + commitVersionLogs(t, repo, "healthy", time.Now().UTC().Add(-time.Hour), 1) + unlock := d.writeGate.Lock(writegate.WriteRollupService) + var once sync.Once + release := func() { once.Do(unlock) } + defer release() + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + finished := make(chan error, 1) + go func() { _, err := d.refreshVersionRollup(ctx, time.Second); finished <- err }() + blocked, stop := make(chan struct{}), make(chan struct{}) + defer close(stop) + go signalVersionGateWait(blocked, stop) + select { + case <-blocked: + cancel() + case err := <-finished: + t.Fatalf("returned before gate wait: %v", err) + case <-time.After(2 * time.Second): + t.Fatal("never blocked on gate") + } + select { + case err := <-finished: + if !errors.Is(err, context.Canceled) { + t.Fatal(err) + } + case <-time.After(2 * time.Second): + t.Fatal("ignored cancellation") + } + release() + if len(d.versionRollupFailures) != 0 { + t.Fatal("cancellation penalized batch") + } + if _, err := d.RefreshVersionRollup(t.Context()); err != nil { + t.Fatalf("lock leaked: %v", err) } } // Observe a parked select, rather than guessing when a goroutine has reached -// a gate. The parquet stack also proves the free write gate was acquired. +// the write gate. // This test-only probe avoids adding instrumentation to the production gates. -func signalVersionGateWait(gate string, blocked chan<- struct{}, stop <-chan struct{}) { +func signalVersionGateWait(blocked chan<- struct{}, stop <-chan struct{}) { frame := "writegate.(*WriteGate).LockContext" - if gate == "parquet" { - frame = "query.(*parquetReadGate).RLockContext" - } ticker := time.NewTicker(time.Millisecond) defer ticker.Stop() buffer := make([]byte, 1<<20) diff --git a/internal/telemetry/compaction_debt.go b/internal/telemetry/compaction_debt.go index 163d7a64..a83be249 100644 --- a/internal/telemetry/compaction_debt.go +++ b/internal/telemetry/compaction_debt.go @@ -6,9 +6,13 @@ import ( ) const ( - CompactionMaxRows = 25_000_000 - CompactionBatchLimit = 128 - CompactionDebtLimit = 1024 + CompactionMaxRows = 25_000_000 + // The slowest measured merge with six busy cores sustained over 400k + // rows/s. Four million rows retire pressured debt in about ten seconds; + // unpressured and higher-generation merges retain the normal ceiling. + CompactionPressureMaxRows = 4_000_000 + CompactionBatchLimit = 128 + CompactionDebtLimit = 1024 ) // Generation0BatchCount is the exact live ingest-batch inventory, including @@ -30,15 +34,17 @@ func (p *ParquetStore) CompactionDebt() int { // changeGeneration0Locked also maintains conservative, reclaimable debt. A raw // generation-0 limit can deadlock on arbitrary oversized requests or on many // finished days with fewer than eight inputs. Exempt batches above half the -// merge row ceiling, and reserve one underfilled group per day. Positive debt -// then guarantees a selectable group without changing compaction policy. +// pressured merge row ceiling, and reserve one underfilled group per day. +// Positive debt then guarantees a selectable group within the pressured ceiling. Larger +// indivisible inputs can still compact below pressure, but cannot block ingest +// behind debt that the pressured policy is unable to retire. func (p *ParquetStore) changeGeneration0Locked(b BatchMetadata, delta int) { if b.Generation != 0 { return } p.generation0Batches += delta rows := int64(b.Spans) + int64(b.Logs) + int64(b.Metrics) - if b.MaxIngestedNanos <= 0 || rows <= 0 || rows > CompactionMaxRows/2 { + if b.MaxIngestedNanos <= 0 || rows <= 0 || rows > CompactionPressureMaxRows/2 { return } if p.generation0ByDay == nil { diff --git a/internal/telemetry/compaction_debt_test.go b/internal/telemetry/compaction_debt_test.go index 1996994e..d5c28ca8 100644 --- a/internal/telemetry/compaction_debt_test.go +++ b/internal/telemetry/compaction_debt_test.go @@ -108,7 +108,7 @@ func TestCompactionDebtExcludesUnreclaimableInventory(t *testing.T) { t.Fatalf("after install %d debt=%d want=%d", i, got, want) } } - p.installBatch(&storedBatch{metadata: BatchMetadata{ID: "oversized", Spans: CompactionMaxRows, MaxIngestedNanos: 1}}) + p.installBatch(&storedBatch{metadata: BatchMetadata{ID: "oversized", Spans: CompactionPressureMaxRows/2 + 1, MaxIngestedNanos: 1}}) p.installBatch(&storedBatch{metadata: BatchMetadata{ID: "another-day", Spans: 1, MaxIngestedNanos: int64(24 * time.Hour)}}) if got := p.CompactionDebt(); got != CompactionDebtLimit/2 { t.Fatalf("unreclaimable inputs raised debt: %d", got) diff --git a/internal/telemetry/compaction_index_test.go b/internal/telemetry/compaction_index_test.go index dc475d52..425ee3c5 100644 --- a/internal/telemetry/compaction_index_test.go +++ b/internal/telemetry/compaction_index_test.go @@ -25,7 +25,7 @@ func TestCompactionStreamingIndexMatchesReread(t *testing.T) { } var inputs []string for part := 0; part < 4; part++ { - spans := make([]Span, 5000) + spans := make([]Span, 15001) for i := range spans { // Shared traces cross inputs and read buffers. Tied and signed times // exercise the complete sort tuple and the existing metadata bounds. @@ -47,7 +47,7 @@ func TestCompactionStreamingIndexMatchesReread(t *testing.T) { if err != nil { t.Fatal(err) } - if stats.Rows != 20000 || stats.MinSpanStartNanos != 1 || stats.MaxSpanStartNanos != 5 { + if stats.Rows != 60004 || stats.MinSpanStartNanos != 1 || stats.MaxSpanStartNanos != 5 { t.Fatalf("merge stats=%+v", stats) } // This is the old narrow reread, retained only as an independent oracle. diff --git a/internal/telemetry/parquet.go b/internal/telemetry/parquet.go index c7fc1c3e..4ad62f44 100644 --- a/internal/telemetry/parquet.go +++ b/internal/telemetry/parquet.go @@ -1255,147 +1255,6 @@ func parquetWriterOptions(pageSize int, extra ...parquet.WriterOption) []parquet return append(options, extra...) } -type contextParquetRows struct { - context.Context - parquet.Rows - index *traceIndexWriter - stats *ParquetMergeStats - hashColumn int - startColumn int -} - -func (r contextParquetRows) ReadRows(rows []parquet.Row) (int, error) { - if err := r.Context.Err(); err != nil { - return 0, err - } - n, err := r.Rows.ReadRows(rows) - if r.index != nil { - for _, row := range rows[:n] { - var hash uint64 - var start int64 - found := 0 - for _, value := range row { - switch value.Column() { - case r.hashColumn: - hash = value.Uint64() - found++ - case r.startColumn: - start = value.Int64() - found++ - } - if found == 2 { - break - } - } - if found != 2 { - return 0, errors.New("compacted span is missing index columns") - } - if start > 0 && (r.stats.MinSpanStartNanos == 0 || start < r.stats.MinSpanStartNanos) { - r.stats.MinSpanStartNanos = start - } - r.stats.MaxSpanStartNanos = max(r.stats.MaxSpanStartNanos, start) - if indexErr := r.index.Append(hash); indexErr != nil { - return 0, indexErr - } - } - } - return n, err -} - -func mergeTypedParquet[T any](ctx context.Context, inputs []string, output string, sorted bool, stats *ParquetMergeStats) (err error) { - if err := ctx.Err(); err != nil { - return err - } - if len(inputs) == 0 { - return errors.New("merge Parquet requires at least one input") - } - files := make([]*os.File, 0, len(inputs)) - defer func() { - for _, file := range files { - err = errors.Join(err, file.Close()) - } - }() - groups := make([]parquet.RowGroup, 0, len(inputs)) - for _, input := range inputs { - file, openErr := os.Open(input) - if openErr != nil { - return openErr - } - files = append(files, file) - info, statErr := file.Stat() - if statErr != nil { - return statErr - } - parquetFile, parquetErr := parquet.OpenFile(file, info.Size()) - if parquetErr != nil { - return parquetErr - } - groups = append(groups, parquetFile.RowGroups()...) - } - var source parquet.RowGroup - var options []parquet.WriterOption - if sorted { - source, err = parquet.MergeRowGroups(groups, parquet.SortingRowGroupConfig(spanSortingColumns())) - options = append(options, parquet.SortingWriterConfig(spanSortingColumns())) - } else { - source = parquet.MultiRowGroup(groups...) - } - if err != nil { - return err - } - rows := source.Rows() - defer func() { err = errors.Join(err, rows.Close()) }() - out, err := os.OpenFile(output, os.O_CREATE|os.O_EXCL|os.O_RDWR, 0o644) - if err != nil { - return err - } - ok := false - defer func() { - _ = out.Close() - if !ok { - _ = os.Remove(output) - } - }() - options = append(options, telemetrySchema[T]()) - reader := contextParquetRows{Context: ctx, Rows: rows, stats: stats} - if sorted { - hash, found := source.Schema().Lookup("_trace_hash") - if !found { - return errors.New("compacted spans have no trace hash") - } - start, found := source.Schema().Lookup("start_unix_nano") - if !found { - return errors.New("compacted spans have no start time") - } - index, indexErr := newTraceIndexWriter(filepath.Join(filepath.Dir(output), "trace.fidx")) - if indexErr != nil { - return indexErr - } - defer func() { - if !ok { - index.Abort() - } - }() - reader.index, reader.hashColumn, reader.startColumn = index, hash.ColumnIndex, start.ColumnIndex - } - writer := parquet.NewGenericWriter[T](out, parquetWriterOptions(parquetPageSize, options...)...) - count, copyErr := parquet.CopyRows(writer, reader) - stats.Rows = count - if closeErr := writer.Close(); copyErr != nil || closeErr != nil { - return errors.Join(copyErr, closeErr) - } - if err := out.Close(); err != nil { - return err - } - if reader.index != nil { - if err := reader.index.Close(); err != nil { - return err - } - } - ok = true - return nil -} - func writeTypedParquet[T any](path string, rows []T, pageSize int, options ...parquet.WriterOption) error { f, err := os.OpenFile(path, os.O_CREATE|os.O_EXCL|os.O_RDWR, 0o644) if err != nil { diff --git a/internal/telemetry/parquet_merge.go b/internal/telemetry/parquet_merge.go new file mode 100644 index 00000000..98c48f63 --- /dev/null +++ b/internal/telemetry/parquet_merge.go @@ -0,0 +1,345 @@ +package telemetry + +import ( + "container/heap" + "context" + "errors" + "fmt" + "io" + "os" + "path/filepath" + "runtime" + + "github.com/parquet-go/parquet-go" + "golang.org/x/sync/errgroup" +) + +// Format 4 has one physical value per leaf column per row, including nulls: +// nested attributes live in VARIANT, not repeated Parquet lists. Merge the +// narrow span sort keys once, then apply the same input-row schedule to every +// physical column. No VARIANT reconstruction or wide-row conversion is needed. +// Columns own their input cursors and output buffers; only footer publication +// is serial. At most eight columns decode/encode at once, sharing the existing +// compression pool and preserving the ingest writer's row-group ceiling. +func mergeTypedParquet[T any](ctx context.Context, inputs []string, output string, sorted bool, stats *ParquetMergeStats) (err error) { + if err := ctx.Err(); err != nil { + return err + } + if len(inputs) == 0 { + return errors.New("merge Parquet requires at least one input") + } + var files []*os.File + defer func() { + for _, f := range files { + err = errors.Join(err, f.Close()) + } + }() + schema := telemetrySchema[T]() + var groups []parquet.RowGroup + for _, input := range inputs { + f, e := os.Open(input) + if e != nil { + return e + } + files = append(files, f) + info, e := f.Stat() + if e != nil { + return e + } + pf, e := parquet.OpenFile(f, info.Size(), parquet.ReadBufferSize(16<<10), parquet.OptimisticRead(true)) + if e != nil { + return e + } + if !parquet.EqualNodes(schema, pf.Schema()) { + return fmt.Errorf("%w: compaction input schema", ErrUnsupportedBatchFormat) + } + groups = append(groups, pf.RowGroups()...) + } + for _, path := range schema.Columns() { + leaf, _ := schema.Lookup(path...) + if leaf.MaxRepetitionLevel != 0 { + return errors.New("compaction requires non-repeated physical columns") + } + } + out, err := os.OpenFile(output, os.O_CREATE|os.O_EXCL|os.O_RDWR, 0644) + if err != nil { + return err + } + ok := false + defer func() { + _ = out.Close() + if !ok { + _ = os.Remove(output) + } + }() + options := []parquet.WriterOption{schema} + if sorted { + options = append(options, parquet.SortingWriterConfig(spanSortingColumns())) + } + writer := parquet.NewGenericWriter[T](out, parquetWriterOptions(parquetPageSize, options...)...) + // Close on failure too: it returns the column page buffers to their pool. + defer func() { + if !ok { + _ = writer.Close() + } + }() + var index *traceIndexWriter + var keys mergeKeyHeap + if sorted { + index, err = newTraceIndexWriter(filepath.Join(filepath.Dir(output), "trace.fidx")) + if err != nil { + return err + } + defer func() { + if !ok { + index.Abort() + } + }() + for i, group := range groups { + c := &mergeKeyCursor{input: i, reader: parquet.NewGenericRowGroupReader[mergeSpanKey](group)} + // Keep empty/exhausted cursors in a separate lifetime list below. + keys = append(keys, c) + } + allKeys := append(mergeKeyHeap(nil), keys...) + defer func() { + for _, c := range allKeys { + err = errors.Join(err, c.reader.Close()) + } + }() + keys = keys[:0] + for _, c := range allKeys { + has, e := c.next() + if e != nil { + return e + } + if has { + keys = append(keys, c) + } + } + heap.Init(&keys) + } + columns := make([]mergeColumn, len(schema.Columns())) + for i := range columns { + columns[i] = mergeColumn{column: i, groups: groups, readers: make([]parquet.ColumnChunkValueReader, len(groups)), remaining: make([]int64, len(groups))} + } + defer func() { + for i := range columns { + err = errors.Join(err, columns[i].close()) + } + }() + // Give workers a standard, concurrency-safe context even when a caller wraps + // Context.Err for instrumentation. Cancellation never leaves workers behind. + workerCtx, cancel := context.WithCancel(ctx) + defer cancel() + runs := make([]mergeRun, 0, parquetRowGroupRows) + input := 0 + var consumed int64 + for { + if err := ctx.Err(); err != nil { + return err + } + runs = runs[:0] + count := 0 + for count < parquetRowGroupRows { + if sorted { + if len(keys) == 0 { + break + } + c := keys[0] + key := c.current() + if key.Start > 0 && (stats.MinSpanStartNanos == 0 || key.Start < stats.MinSpanStartNanos) { + stats.MinSpanStartNanos = key.Start + } + stats.MaxSpanStartNanos = max(stats.MaxSpanStartNanos, key.Start) + if e := index.Append(key.Hash); e != nil { + return e + } + runs = appendMergeRun(runs, c.input, 1) + count++ + has, e := c.next() + if e != nil { + return e + } + if has { + heap.Fix(&keys, 0) + } else { + heap.Pop(&keys) + } + } else { + for input < len(groups) && consumed == groups[input].NumRows() { + input++ + consumed = 0 + } + if input == len(groups) { + break + } + n := int(min(int64(parquetRowGroupRows-count), groups[input].NumRows()-consumed)) + runs = appendMergeRun(runs, input, n) + count += n + consumed += int64(n) + } + } + if count == 0 { + break + } + writers := writer.ColumnWriters() + var work errgroup.Group + work.SetLimit(min(8, runtime.GOMAXPROCS(0))) + for i := range columns { + work.Go(func() error { return columns[i].write(workerCtx, writers[i], runs) }) + } + if e := work.Wait(); e != nil { + return e + } + if e := writer.Flush(); e != nil { + return e + } + stats.Rows += int64(count) + } + if err := writer.Close(); err != nil { + return err + } + if err := out.Close(); err != nil { + return err + } + if index != nil { + if err := index.Close(); err != nil { + return err + } + } + ok = true + return nil +} + +type mergeRun struct{ input, rows int } + +func appendMergeRun(runs []mergeRun, input, rows int) []mergeRun { + if len(runs) > 0 && runs[len(runs)-1].input == input { + runs[len(runs)-1].rows += rows + return runs + } + return append(runs, mergeRun{input, rows}) +} + +type mergeColumn struct { + column int + groups []parquet.RowGroup + readers []parquet.ColumnChunkValueReader + remaining []int64 + values [256]parquet.Value +} + +func (c *mergeColumn) close() (err error) { + for _, r := range c.readers { + if r != nil { + err = errors.Join(err, r.Close()) + } + } + return err +} +func (c *mergeColumn) write(ctx context.Context, writer *parquet.ColumnWriter, runs []mergeRun) error { + for _, run := range runs { + if err := ctx.Err(); err != nil { + return err + } + r := c.readers[run.input] + if r == nil { + r = parquet.NewColumnChunkValueReader(c.groups[run.input].ColumnChunks()[c.column]) + c.readers[run.input] = r + c.remaining[run.input] = c.groups[run.input].NumRows() + } + for remaining := run.rows; remaining > 0; { + n, e := r.ReadValues(c.values[:min(remaining, len(c.values))]) + if n > 0 { + written, err := writer.WriteRowValues(c.values[:n]) + clear(c.values[:n]) + if err != nil { + return err + } + if written != n { + return io.ErrShortWrite + } + remaining -= n + c.remaining[run.input] -= int64(n) + } + if e != nil { + if e == io.EOF { + return io.ErrUnexpectedEOF + } + return e + } + if n == 0 { + return io.ErrNoProgress + } + } + if c.remaining[run.input] == 0 { + if err := r.Close(); err != nil { + return err + } + c.readers[run.input] = nil + } + } + // Flush the final partial page concurrently too. Flush on the parent is + // called only after every column is finished, so footer statistics and row + // counts are observed without racing any column writer. + return writer.Flush() +} + +type mergeSpanKey struct { + Hash uint64 `parquet:"_trace_hash"` + Start int64 `parquet:"start_unix_nano"` + SpanID string `parquet:"span_id"` +} +type mergeKeyCursor struct { + input int + reader *parquet.GenericReader[mergeSpanKey] + buffer [256]mergeSpanKey + pos, n int + eof bool +} + +func (c *mergeKeyCursor) current() mergeSpanKey { return c.buffer[c.pos-1] } +func (c *mergeKeyCursor) next() (bool, error) { + if c.pos == c.n { + if c.eof { + return false, nil + } + clear(c.buffer[:]) + n, e := c.reader.Read(c.buffer[:]) + c.pos, c.n = 0, n + if e != nil { + if e != io.EOF { + return false, e + } + c.eof = true + } + if n == 0 { + if c.eof { + return false, nil + } + return false, io.ErrNoProgress + } + } + c.pos++ + return true, nil +} + +type mergeKeyHeap []*mergeKeyCursor + +func (h mergeKeyHeap) Len() int { return len(h) } +func (h mergeKeyHeap) Less(i, j int) bool { + a, b := h[i].current(), h[j].current() + if a.Hash != b.Hash { + return a.Hash < b.Hash + } + if a.Start != b.Start { + return a.Start < b.Start + } + if a.SpanID != b.SpanID { + return a.SpanID < b.SpanID + } + return h[i].input < h[j].input +} +func (h mergeKeyHeap) Swap(i, j int) { h[i], h[j] = h[j], h[i] } +func (h *mergeKeyHeap) Push(x any) { *h = append(*h, x.(*mergeKeyCursor)) } +func (h *mergeKeyHeap) Pop() any { old := *h; x := old[len(old)-1]; *h = old[:len(old)-1]; return x } diff --git a/internal/telemetry/store/compaction.go b/internal/telemetry/store/compaction.go index e1ecf979..201d93a3 100644 --- a/internal/telemetry/store/compaction.go +++ b/internal/telemetry/store/compaction.go @@ -292,7 +292,11 @@ func selectCompactionBatches(batches []telemetry.BatchMetadata, maxBatches, debt var selected []telemetry.BatchMetadata found := false for key, group := range groups { - candidate := selectBoundedCompactionGroup(group, maxBatches, key.day < newestDay) + rowLimit := int64(maxCompactionRows) + if pressure && key.generation == 0 { + rowLimit = telemetry.CompactionPressureMaxRows + } + candidate := selectBoundedCompactionGroup(group, maxBatches, rowLimit, key.day < newestDay) if len(candidate) < 2 { continue } @@ -315,7 +319,7 @@ func selectCompactionBatches(batches []telemetry.BatchMetadata, maxBatches, debt // small files, but admits a smaller group when the row ceiling fills first. // Without the latter, one successful generation can make every later group too // large for the ceiling and permanently strand those files. -func selectBoundedCompactionGroup(group []telemetry.BatchMetadata, maxBatches int, finished bool) []telemetry.BatchMetadata { +func selectBoundedCompactionGroup(group []telemetry.BatchMetadata, maxBatches int, rowLimit int64, finished bool) []telemetry.BatchMetadata { ordered := append([]telemetry.BatchMetadata(nil), group...) sort.Slice(ordered, func(i, j int) bool { left, right := compactionBatchRows(ordered[i]), compactionBatchRows(ordered[j]) @@ -333,7 +337,7 @@ func selectBoundedCompactionGroup(group []telemetry.BatchMetadata, maxBatches in var smallest int64 for _, batch := range ordered { batchRows := compactionBatchRows(batch) - if batchRows <= 0 || batchRows > maxCompactionRows { + if batchRows <= 0 || batchRows > rowLimit { continue } batchBytes := compactionBatchBytes(batch) @@ -344,7 +348,7 @@ func selectBoundedCompactionGroup(group []telemetry.BatchMetadata, maxBatches in saturated = true break } - if rows > maxCompactionRows-batchRows { + if rows > rowLimit-batchRows { saturated = true break } @@ -364,7 +368,7 @@ func selectBoundedCompactionGroup(group []telemetry.BatchMetadata, maxBatches in rows += batchRows bytes += batchBytes } - if rows == maxCompactionRows || smallest > 0 && smallest > maxCompactionRows-rows { + if rows == rowLimit || smallest > 0 && smallest > rowLimit-rows { saturated = true } if bytes == maxCompactionBytes { diff --git a/internal/telemetry/store/compaction_bytes_test.go b/internal/telemetry/store/compaction_bytes_test.go index 294bfa1c..37371d21 100644 --- a/internal/telemetry/store/compaction_bytes_test.go +++ b/internal/telemetry/store/compaction_bytes_test.go @@ -24,7 +24,7 @@ func TestSelectBoundedCompactionGroupStopsAtTheByteBudget(t *testing.T) { }) } - candidate := selectBoundedCompactionGroup(group, 128, false) + candidate := selectBoundedCompactionGroup(group, 128, maxCompactionRows, false) if len(candidate) == 0 { t.Fatal("no group selected; the byte budget must still admit a mergeable group") @@ -58,7 +58,7 @@ func TestSelectBoundedCompactionGroupChargesUnsizedBatchesAnEstimate(t *testing. }) } - candidate := selectBoundedCompactionGroup(group, 128, false) + candidate := selectBoundedCompactionGroup(group, 128, maxCompactionRows, false) if len(candidate) == 0 { t.Fatal("no group selected; unsized batches must still be compactable") diff --git a/internal/telemetry/store/compaction_pressure_test.go b/internal/telemetry/store/compaction_pressure_test.go new file mode 100644 index 00000000..d7dcdc6a --- /dev/null +++ b/internal/telemetry/store/compaction_pressure_test.go @@ -0,0 +1,40 @@ +package store + +import ( + "testing" + + "github.com/labstack/fanout/internal/telemetry" +) + +func TestPressuredCompactionBoundsRowsAndPreservesNormalCeilings(t *testing.T) { + for _, rows := range []int{50_000, telemetry.CompactionPressureMaxRows / 2, telemetry.CompactionPressureMaxRows/2 + 1} { + small := genBatches(2, 0, 1151, rows) + // Another day provides real reclaimable debt even when this day's inputs + // are too large for the pressure ceiling. Selection must still progress. + if rows > telemetry.CompactionPressureMaxRows/2 { + small = append(small, genBatches(3, 0, 1151, 50_000)...) + } + selected := selectCompactionBatches(small, 128, telemetry.CompactionDebtLimit) + var total int64 + for _, b := range selected { + total += compactionBatchRows(b) + } + want := telemetry.CompactionPressureMaxRows / min(rows, telemetry.CompactionPressureMaxRows/2) + if rows > telemetry.CompactionPressureMaxRows/2 { + want = telemetry.CompactionPressureMaxRows / 50_000 + } + if len(selected) != want || total > telemetry.CompactionPressureMaxRows { + t.Fatalf("rows/input=%d selection=%d total=%d", rows, len(selected), total) + } + } + group := genBatches(2, 0, 128, 50_000) + if selected := selectCompactionBatches(group, 128, telemetry.CompactionDebtLimit/2-1); len(selected) != 128 { + t.Fatalf("normal ceiling changed: %d", len(selected)) + } + for i := range group { + group[i].Generation = 1 + } + if selected := selectCompactionBatches(group, 128, telemetry.CompactionDebtLimit); len(selected) != 128 { + t.Fatalf("higher generation ceiling changed: %d", len(selected)) + } +} diff --git a/internal/telemetry/store/compaction_starvation_test.go b/internal/telemetry/store/compaction_starvation_test.go index dc287f13..d770436b 100644 --- a/internal/telemetry/store/compaction_starvation_test.go +++ b/internal/telemetry/store/compaction_starvation_test.go @@ -127,7 +127,7 @@ func TestSelectBoundedCompactionGroupCompactsAnUnderfilledDay(t *testing.T) { // A finished day: well under the count ceiling, nowhere near saturating. group := genBatches(3, 1, 56, 5_000) - candidate := selectBoundedCompactionGroup(group, 128, true) + candidate := selectBoundedCompactionGroup(group, 128, maxCompactionRows, true) if len(candidate) == 0 { t.Fatal("no candidate for a 56-batch day: a day that will never grow to 128 files can never consolidate") @@ -139,7 +139,7 @@ func TestSelectBoundedCompactionGroupCompactsAnUnderfilledDay(t *testing.T) { // Still not worth waking the merge for a couple of files. func TestSelectBoundedCompactionGroupIgnoresATrivialDay(t *testing.T) { - if got := selectBoundedCompactionGroup(genBatches(3, 1, 3, 5_000), 128, true); len(got) != 0 { + if got := selectBoundedCompactionGroup(genBatches(3, 1, 3, 5_000), 128, maxCompactionRows, true); len(got) != 0 { t.Errorf("candidate = %d for a 3-batch day, want none: below the effort floor", len(got)) } } From aabf0939e70bb214fa11b1a223a8856ca8328cce Mon Sep 17 00:00:00 2001 From: Vishal Rana Date: Thu, 15 Oct 2026 13:13:06 -0700 Subject: [PATCH 2/2] test(query): give admission tests headroom under the race detector TestAnalyticalRollupsReleaseGateWhenPublicationBlocksAdmission bounded RecordAnomalies by 50 ms. On a CI runner under -race the INSERT itself took longer and was interrupted, although the gate was not parked. The successful paths in both admission tests now allow 5 s, which still fails long before the former 15-second admission lease. The stats timeout case keeps its short deadline. Claude-Session: https://claude.ai/code/session_01XA52nWyQ3R5f5kK8kjpAQW --- internal/query/background_admission_test.go | 8 +++++++- internal/query/rollup_admission_test.go | 8 +++++--- 2 files changed, 12 insertions(+), 4 deletions(-) diff --git a/internal/query/background_admission_test.go b/internal/query/background_admission_test.go index 9d5bbc15..4123d5f5 100644 --- a/internal/query/background_admission_test.go +++ b/internal/query/background_admission_test.go @@ -54,7 +54,13 @@ func TestBackgroundPassesYieldToQueuedPublication(t *testing.T) { }, } { t.Run(name, func(t *testing.T) { - waitCtx, stop := context.WithTimeout(t.Context(), 20*time.Millisecond) + // Successful analytical passes need headroom under the race detector; + // only stats is expected to wait for its deadline. + waitBudget := 5 * time.Second + if name == "stats" { + waitBudget = 20 * time.Millisecond + } + waitCtx, stop := context.WithTimeout(t.Context(), waitBudget) defer stop() err := pass(waitCtx) if name == "stats" { diff --git a/internal/query/rollup_admission_test.go b/internal/query/rollup_admission_test.go index e1e69603..e2dc9fea 100644 --- a/internal/query/rollup_admission_test.go +++ b/internal/query/rollup_admission_test.go @@ -47,13 +47,15 @@ func TestAnalyticalRollupsReleaseGateWhenPublicationBlocksAdmission(t *testing.T if err := d.writer().QueryRow(`SELECT count(*) FROM version_rollup_batches`).Scan(&marked); err != nil || marked != 0 { t.Fatalf("skipped version markers=%d error=%v", marked, err) } - ctx, cancel := context.WithTimeout(t.Context(), 50*time.Millisecond) + // Allow slow race runs to finish the INSERT while still detecting a + // gate parked for the former 15-second admission lease. + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) defer cancel() start := time.Now() if err := d.RecordAnomalies(ctx, []annotations.Anomaly{}, time.Now()); err != nil { - t.Fatalf("anomaly parked behind %s admission: %v", component, err) + t.Fatalf("anomaly persistence failed with %s admission blocked: %v", component, err) } - t.Logf("anomaly persisted with publication held: %s", time.Since(start)) + t.Logf("anomaly persisted with publication blocking %s admission: %s", component, time.Since(start)) d.parquetMu.Unlock() locked = false if n, err := refresh(); err != nil || component != "edge" && n == 0 {