perf(store): bound a compaction pass by the bytes it admits

internal/telemetry/parquet.go

@@ -49,6 +49,10 @@ type BatchMetadata struct {
 	Spans             int    `json:"spans"`
 	Logs              int    `json:"logs"`
 	Metrics           int    `json:"metrics"`
+	// Bytes is the batch's on-disk size, derived rather than persisted: it is
+	// measured when a batch is loaded, from stats the load already performs.
+	// Zero means not measured.
+	Bytes int64 `json:"-"`
 }
 
 type TraceQuery struct {
@@ -1119,6 +1123,10 @@ func loadStoredBatch(dir string) (*storedBatch, error) {
 		if !info.Mode().IsRegular() {
 			return nil, fmt.Errorf("%s Parquet is not a regular file", signal.name)
 		}
+		// Size the batch from the stat this validation already performs.
+		// Compaction prices a merge by the bytes it admits, and measuring here
+		// costs nothing over the estimate it would otherwise fall back to.
+		metadata.Bytes += info.Size()
 		file, err := os.Open(path)
 		if err != nil {
 			return nil, err

internal/telemetry/store/compaction.go

@@ -16,8 +16,17 @@ import (
 )
 
 const (
-	maxCompactionRows   = 25_000_000
-	minCompactionInputs = 8
+	maxCompactionRows  = 25_000_000
+	maxCompactionBytes = 256 << 20
+
+	// assumedCompactionBytesPerRow prices a batch whose size was never
+	// measured. Measured batches run well under this -- span rows with JSON
+	// attributes compress to roughly 75 bytes on disk -- so the estimate is
+	// deliberately several times pessimistic: over-charging costs smaller
+	// groups and more passes, while under-charging costs the merge that the
+	// byte ceiling exists to prevent.
+	assumedCompactionBytesPerRow = 256
+	minCompactionInputs          = 8
 )
 
 var parquetSignals = [...]string{"spans", "logs", "metrics"}
@@ -198,14 +207,15 @@ func selectBoundedCompactionGroup(group []telemetry.BatchMetadata, maxBatches in
 		return ordered[i].ID < ordered[j].ID
 	})
 	candidate := make([]telemetry.BatchMetadata, 0, min(maxBatches, len(ordered)))
-	var rows int64
+	var rows, bytes int64
 	saturated := false
 	var smallest int64
 	for _, batch := range ordered {
 		batchRows := compactionBatchRows(batch)
 		if batchRows <= 0 || batchRows > maxCompactionRows {
 			continue
 		}
+		batchBytes := compactionBatchBytes(batch)
 		if smallest == 0 {
 			smallest = batchRows
 		}
@@ -217,18 +227,47 @@ func selectBoundedCompactionGroup(group []telemetry.BatchMetadata, maxBatches in
 			saturated = true
 			break
 		}
+		// A merge opens a cursor per row group across every input and holds
+		// each one's dictionaries at once, so the live cost of a pass tracks
+		// the bytes it admits. Rows cannot stand in for that: wide spans and
+		// bare log lines differ by an order of magnitude at equal row counts.
+		// A pair is always admitted, however large: a merge of two inputs is
+		// bounded by those inputs, and refusing it would strand files that are
+		// individually over budget instead of ever shrinking them. Past a pair
+		// the ceiling binds.
+		if len(candidate) >= 2 && bytes > maxCompactionBytes-batchBytes {
+			saturated = true
+			break
+		}
 		candidate = append(candidate, batch)
 		rows += batchRows
+		bytes += batchBytes
 	}
 	if rows == maxCompactionRows || smallest > 0 && smallest > maxCompactionRows-rows {
 		saturated = true
 	}
+	if bytes == maxCompactionBytes {
+		saturated = true
+	}
 	if len(candidate) == maxBatches || saturated && len(candidate) >= 2 {
 		return candidate
 	}
 	return nil
 }
 
+// compactionBatchBytes prices a batch for the byte ceiling, falling back to a
+// pessimistic per-row estimate when the batch was never measured.
+func compactionBatchBytes(batch telemetry.BatchMetadata) int64 {
+	if batch.Bytes > 0 {
+		return batch.Bytes
+	}
+	rows := compactionBatchRows(batch)
+	if rows <= 0 || rows > math.MaxInt64/assumedCompactionBytesPerRow {
+		return math.MaxInt64
+	}
+	return rows * assumedCompactionBytesPerRow
+}
+
 func compactionBatchRows(batch telemetry.BatchMetadata) int64 {
 	if batch.Spans < 0 {
 		return math.MaxInt64

internal/telemetry/store/compaction_bytes_test.go

@@ -0,0 +1,73 @@
+package store
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/labstack/fanout/internal/telemetry"
+)
+
+// A merge holds every input's dictionaries and cursors at once, so the live
+// cost of a pass tracks the bytes it admits, not the number of files or the
+// number of rows. Row counts cannot stand in for that: a batch of wide spans
+// carrying large attribute payloads and a batch of bare log lines can hold the
+// same row count and differ by an order of magnitude on disk.
+func TestSelectBoundedCompactionGroupStopsAtTheByteBudget(t *testing.T) {
+	const oneMiB = 1 << 20
+	group := make([]telemetry.BatchMetadata, 0, 16)
+	for i := range 16 {
+		group = append(group, telemetry.BatchMetadata{
+			ID:               string(rune('a' + i)),
+			Spans:            1_000,
+			MinIngestedNanos: int64(i),
+			Bytes:            64 * oneMiB,
+		})
+	}
+
+	candidate := selectBoundedCompactionGroup(group, 128)
+
+	if len(candidate) == 0 {
+		t.Fatal("no group selected; the byte budget must still admit a mergeable group")
+	}
+	var admitted int64
+	for _, batch := range candidate {
+		admitted += batch.Bytes
+	}
+	if admitted > maxCompactionBytes {
+		t.Errorf("admitted %d bytes, budget %d: selection ignored the byte ceiling", admitted, maxCompactionBytes)
+	}
+	if len(candidate) == len(group) {
+		t.Errorf("admitted all %d inputs at %d bytes each; the budget never bound", len(group), int64(64*oneMiB))
+	}
+}
+
+// Batches whose size is unknown must not be treated as free. Nothing persists
+// the byte count -- it is measured when a batch is loaded -- so a zero means
+// "not measured". Charging those an estimate keeps the ceiling meaningful;
+// treating them as weightless would reopen the hole the budget exists to
+// close, and refusing them outright would stop compaction dead if any path
+// ever failed to measure.
+func TestSelectBoundedCompactionGroupChargesUnsizedBatchesAnEstimate(t *testing.T) {
+	group := make([]telemetry.BatchMetadata, 0, 128)
+	for i := range 128 {
+		group = append(group, telemetry.BatchMetadata{
+			ID:               fmt.Sprintf("batch-%03d", i),
+			Spans:            50_000,
+			MinIngestedNanos: int64(i),
+			// Bytes deliberately left zero: never measured.
+		})
+	}
+
+	candidate := selectBoundedCompactionGroup(group, 128)
+
+	if len(candidate) == 0 {
+		t.Fatal("no group selected; unsized batches must still be compactable")
+	}
+	if len(candidate) == len(group) {
+		t.Fatalf("admitted all %d unsized inputs; an unmeasured batch must not count as zero bytes", len(group))
+	}
+	estimated := int64(len(candidate)) * 50_000 * assumedCompactionBytesPerRow
+	if estimated > maxCompactionBytes {
+		t.Errorf("estimated %d bytes across %d inputs, budget %d", estimated, len(candidate), int64(maxCompactionBytes))
+	}
+}

internal/telemetry/store/repository_test.go

@@ -511,6 +511,9 @@ func TestSelectCompactionBatchesBuildsRowBoundedGroup(t *testing.T) {
 		batches[i] = telemetry.BatchMetadata{
 			ID: fmt.Sprintf("batch-%d", i), MaxIngestedNanos: 1, Generation: 2,
 			Spans: 2_000_000,
+			// Sized well under the byte ceiling so the row ceiling stays the
+			// constraint under test here; the byte ceiling has its own tests.
+			Bytes: 1 << 20,
 		}
 	}
 	selected := selectCompactionBatches(batches, maxBatches)
@@ -533,6 +536,9 @@ func TestSelectCompactionBatchesCombinesSaturatedLargeFiles(t *testing.T) {
 		batches[i] = telemetry.BatchMetadata{
 			ID: fmt.Sprintf("large-%d", i), MaxIngestedNanos: 1, Generation: 3,
 			Spans: 6_000_000,
+			// Sized well under the byte ceiling so saturation here is the row
+			// ceiling, which is what this test is about.
+			Bytes: 1 << 20,
 		}
 	}
 	selected := selectCompactionBatches(batches, maxBatches)