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..4123d5f5 100644
--- a/internal/query/background_admission_test.go
+++ b/internal/query/background_admission_test.go
@@ -54,10 +54,21 @@ 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()
-			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..e2dc9fea
--- /dev/null
+++ b/internal/query/rollup_admission_test.go
@@ -0,0 +1,66 @@
+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)
+			}
+			// 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 persistence failed with %s admission blocked: %v", component, err)
+			}
+			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 {
+				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))
 	}
 }