From b3f045278bc489d45e58d486813e7c64a6bc4bc6 Mon Sep 17 00:00:00 2001 From: Vishal Rana Date: Sun, 20 Sep 2026 11:05:28 -0700 Subject: [PATCH] perf(telemetry): stop dictionary-encoding near-unique JSON columns Parquet dictionary encoding keeps every distinct value of a column chunk for the whole row group, and ingest flushes a row group only at parquetRowGroupRows (50,000). On a column whose values repeat -- service names, HTTP methods, scope identifiers, the resource block shared by every span in a ResourceSpans -- that is a saving. On attributes_json, events_json, links_json, exemplars_json and exception_message it is not: those are near-unique per row by construction, so the dictionary is a second full copy of the payload, held until the row group closes. Measured against a 23.7 MiB synthetic batch of 50k spans, peak heap for one CommitBatch: 4.09x-4.92x payload with dictionary encoding, 2.79x-3.49x without. Those columns are now written plain. This does not make one commit cheap, it makes it cheaper. What remains above 1x is the conversion copy in makeSpanParquetRow, roughly 24 MiB of row-struct headers per 50k spans, and the writer's own buffers -- each tracked separately in the ingest-memory issue. Adds TestCommitBatchPeakHeapStaysNearPayloadSize, which samples HeapInuse during a commit and fails above a ratio placed in the gap between the two measured distributions, plus BenchmarkCommitBatch reporting peak heap so a regression appears as a number rather than as a production OOM kill. There was no in-tree memory benchmark before this. No migration: DuckDB reads either encoding, and parquet-go compares schema by node rather than by encoding, so existing files and merges are unaffected. Ingest files grow modestly; compaction re-encodes them anyway. --- internal/telemetry/commit_memory_test.go | 171 +++++++++++++++++++++++ internal/telemetry/parquet_rows.go | 25 +++- 2 files changed, 189 insertions(+), 7 deletions(-) create mode 100644 internal/telemetry/commit_memory_test.go diff --git a/internal/telemetry/commit_memory_test.go b/internal/telemetry/commit_memory_test.go new file mode 100644 index 00000000..7e5b9824 --- /dev/null +++ b/internal/telemetry/commit_memory_test.go @@ -0,0 +1,171 @@ +package telemetry + +import ( + "context" + "fmt" + "runtime" + "runtime/debug" + "sync" + "sync/atomic" + "testing" + "time" +) + +// commitPeakRatioLimit is how much live heap one CommitBatch may hold relative +// to the payload it was handed. The number that matters is the multiplier, not +// an absolute byte count: peak scales with batch size, and batch size scales +// with ingest rate, so a process sized for a quiet hour is sized wrong for a +// busy one. +// +// This is a ratchet, not a target, and it is placed between two measured +// distributions rather than guessed. Under -race, dictionary-encoded +// near-unique JSON columns ran 4.09x-4.92x across repeated runs; writing them +// plain runs 2.79x-3.49x. The limit sits in the gap, so it passes the current +// encoding with headroom for GC timing noise and fails a regression to the one +// that was OOM-killing the process. +// +// What remains above 1x is a conversion copy of every JSON column +// (makeSpanParquetRow) plus ~24 MiB of row-struct headers per 50k spans and +// the writer's own buffers. Lower this constant as each of those is addressed; +// see the ordered plan in the ingest-memory issue. +const commitPeakRatioLimit = 3.75 + +// syntheticSpan builds a span whose JSON columns look like production: a +// resource block shared across the batch, and attributes/events/links that are +// near-unique per row. That distinction is the whole point — a dictionary over +// a near-unique column stores every value twice. +func syntheticSpan(i int, base time.Time) Span { + start := base.Add(time.Duration(i) * time.Microsecond) + return Span{ + Namespace: "prod", + TraceID: fmt.Sprintf("%032x", i), + SpanID: fmt.Sprintf("%016x", i), + ServiceName: "checkout", + Name: "POST /api/orders", + Kind: "SERVER", + StartUnixNanos: start.UnixNano(), + EndUnixNanos: start.Add(3 * time.Millisecond).UnixNano(), + DurationMS: 3, + StatusCode: "OK", + ResourceJSON: []byte(`{"service.name":"checkout","deployment.environment":"prod","host.name":"cube-10"}`), + AttributesJSON: []byte(fmt.Sprintf( + `{"http.method":"POST","http.route":"/api/orders","http.status_code":200,"order.id":"%d","user.id":"u-%d","session":"%032x"}`, i, i*7, i)), + EventsJSON: []byte(fmt.Sprintf( + `[{"name":"validated","time":%d},{"name":"charged","time":%d,"amount":%d}]`, start.UnixNano(), start.UnixNano()+1000, i%9999)), + LinksJSON: []byte(fmt.Sprintf(`[{"trace_id":"%032x","span_id":"%016x"}]`, i+1, i+1)), + } +} + +func payloadBytes(spans []Span) int64 { + var total int64 + for i := range spans { + total += int64(len(spans[i].ResourceJSON) + len(spans[i].AttributesJSON) + + len(spans[i].EventsJSON) + len(spans[i].LinksJSON) + + len(spans[i].TraceID) + len(spans[i].SpanID) + len(spans[i].ServiceName) + len(spans[i].Name)) + } + return total +} + +// samplePeakHeapInuse runs fn while polling HeapInuse, and reports the highest +// value observed above the pre-run baseline. GOGC is pinned low for the +// duration so the figure approximates live bytes rather than live bytes plus +// whatever slack the collector happened to be carrying; production GOGC roughly +// doubles it. +func samplePeakHeapInuse(fn func()) int64 { + previousGC := debug.SetGCPercent(5) + defer debug.SetGCPercent(previousGC) + + runtime.GC() + var stats runtime.MemStats + runtime.ReadMemStats(&stats) + baseline := int64(stats.HeapInuse) + + var peak atomic.Int64 + stop := make(chan struct{}) + var sampler sync.WaitGroup + sampler.Add(1) + go func() { + defer sampler.Done() + var sample runtime.MemStats + for { + select { + case <-stop: + return + default: + } + runtime.ReadMemStats(&sample) + if delta := int64(sample.HeapInuse) - baseline; delta > peak.Load() { + peak.Store(delta) + } + time.Sleep(2 * time.Millisecond) + } + }() + + fn() + close(stop) + sampler.Wait() + return peak.Load() +} + +// One commit must not cost several times the payload it was given. Four commit +// workers run concurrently, each queued behind more batches, so a multiplier +// here is a multiplier on the whole process. +func TestCommitBatchPeakHeapStaysNearPayloadSize(t *testing.T) { + if testing.Short() { + t.Skip("allocates ~50k spans") + } + store, err := OpenParquetStore(t.TempDir()) + if err != nil { + t.Fatalf("open parquet store: %v", err) + } + base := time.Date(2026, 9, 20, 12, 0, 0, 0, time.UTC) + spans := make([]Span, 50_000) + for i := range spans { + spans[i] = syntheticSpan(i, base) + } + raw := payloadBytes(spans) + + peak := samplePeakHeapInuse(func() { + if err := store.CommitBatch(context.Background(), BatchMetadata{ID: "peak-heap"}, spans, nil, nil); err != nil { + t.Errorf("commit batch: %v", err) + } + }) + runtime.KeepAlive(spans) + + ratio := float64(peak) / float64(raw) + t.Logf("payload %.1f MiB, peak heap %.1f MiB, ratio %.2fx", + float64(raw)/(1<<20), float64(peak)/(1<<20), ratio) + if ratio > commitPeakRatioLimit { + t.Errorf("CommitBatch peak heap %.2fx payload, limit %.2fx: one batch holds several copies of what it was handed", + ratio, commitPeakRatioLimit) + } +} + +// BenchmarkCommitBatch reports peak heap alongside time so a regression shows +// up as a number rather than as an out-of-memory kill in production. +func BenchmarkCommitBatch(b *testing.B) { + base := time.Date(2026, 9, 20, 12, 0, 0, 0, time.UTC) + spans := make([]Span, 50_000) + for i := range spans { + spans[i] = syntheticSpan(i, base) + } + raw := payloadBytes(spans) + + var peak int64 + for i := 0; b.Loop(); i++ { + store, err := OpenParquetStore(b.TempDir()) + if err != nil { + b.Fatalf("open parquet store: %v", err) + } + observed := samplePeakHeapInuse(func() { + if err := store.CommitBatch(context.Background(), BatchMetadata{ID: fmt.Sprintf("bench-%d", i)}, spans, nil, nil); err != nil { + b.Fatalf("commit batch: %v", err) + } + }) + if observed > peak { + peak = observed + } + } + b.ReportMetric(float64(peak)/(1<<20), "peak-heap-MiB") + b.ReportMetric(float64(peak)/float64(raw), "peak/payload") +} diff --git a/internal/telemetry/parquet_rows.go b/internal/telemetry/parquet_rows.go index f275c15b..e338c229 100644 --- a/internal/telemetry/parquet_rows.go +++ b/internal/telemetry/parquet_rows.go @@ -1,5 +1,16 @@ package telemetry +// Dictionary encoding is applied per column chunk and holds every distinct +// value for the whole row group, which ingest flushes only at +// parquetRowGroupRows. That pays for itself on a column whose values repeat — +// service names, HTTP methods, scope identifiers, the resource block shared by +// every span in a ResourceSpans — and costs a full second copy of the payload +// on a column whose values do not. The JSON columns carrying per-row +// attributes, events, links and exemplars are near-unique by construction, so +// they are written plain: dictionary-encoding them made one 50k-span commit +// hold several times the batch it was handed, which is what +// TestCommitBatchPeakHeapStaysNearPayloadSize pins. + type spanParquetRow struct { Namespace string `parquet:"namespace"` TraceID string `parquet:"trace_id"` @@ -17,9 +28,9 @@ type spanParquetRow struct { Status string `parquet:"status"` StatusMessage string `parquet:"status_message"` ResourceJSON string `parquet:"resource_json,dict"` - AttributesJSON string `parquet:"attributes_json,dict"` - EventsJSON string `parquet:"events_json,dict"` - LinksJSON string `parquet:"links_json,dict"` + AttributesJSON string `parquet:"attributes_json"` + EventsJSON string `parquet:"events_json"` + LinksJSON string `parquet:"links_json"` TraceState string `parquet:"trace_state,dict"` Flags int64 `parquet:"flags"` ScopeName string `parquet:"scope_name,dict"` @@ -36,7 +47,7 @@ type spanParquetRow struct { ServiceVersion string `parquet:"service_version,dict"` DeploymentEnv string `parquet:"deployment_env,dict"` ExceptionType string `parquet:"exception_type,dict"` - ExceptionMessage string `parquet:"exception_message,dict"` + ExceptionMessage string `parquet:"exception_message"` } func makeSpanParquetRow(r Span) spanParquetRow { @@ -90,7 +101,7 @@ type logParquetRow struct { SpanID string `parquet:"span_id"` Flags int64 `parquet:"flags"` ResourceJSON string `parquet:"resource_json,dict"` - AttributesJSON string `parquet:"attributes_json,dict"` + AttributesJSON string `parquet:"attributes_json"` ScopeName string `parquet:"scope_name,dict"` ScopeVersion string `parquet:"scope_version,dict"` IngestedAt int64 `parquet:"ingested_at,timestamp(nanosecond)"` @@ -123,8 +134,8 @@ type metricParquetRow struct { HistCountsJSON string `parquet:"hist_counts_json,dict"` HistCount int64 `parquet:"hist_count"` HistSum float64 `parquet:"hist_sum"` - ExemplarsJSON string `parquet:"exemplars_json,dict"` - AttributesJSON string `parquet:"attributes_json,dict"` + ExemplarsJSON string `parquet:"exemplars_json"` + AttributesJSON string `parquet:"attributes_json"` ResourceJSON string `parquet:"resource_json,dict"` ScopeName string `parquet:"scope_name,dict"` ScopeVersion string `parquet:"scope_version,dict"`