perf(telemetry): stop copying JSON payloads when converting to Parquet rows

internal/ingest/attrs_test.go

@@ -71,10 +71,10 @@ func TestAttrsJSON_MatchesReflect(t *testing.T) {
 		fast := attrsJSON(attrs)
 		slow := attrsJSONReflect(attrs)
 		var mf, ms map[string]any
-		if err := json.Unmarshal(fast, &mf); err != nil {
+		if err := json.Unmarshal([]byte(fast), &mf); err != nil {
 			t.Fatalf("case %d: fast output invalid JSON: %v (%s)", i, err, fast)
 		}
-		if err := json.Unmarshal(slow, &ms); err != nil {
+		if err := json.Unmarshal([]byte(slow), &ms); err != nil {
 			t.Fatalf("case %d: reflect output invalid JSON: %v", i, err)
 		}
 		if !reflect.DeepEqual(mf, ms) {
@@ -84,14 +84,15 @@ func TestAttrsJSON_MatchesReflect(t *testing.T) {
 }
 
 // Non-finite floats route to the reflect path (json.Marshal errors on Inf/NaN),
-// so attrsJSON must never emit invalid JSON like {"k":Inf} — it returns nil for
-// the whole object (the reflect encoder's documented behavior).
+// so attrsJSON must never emit invalid JSON like {"k":Inf} — it returns an
+// empty string for the whole object (the reflect encoder's documented
+// behavior).
 func TestAttrsJSON_NonFiniteFloatIsSafe(t *testing.T) {
 	for _, v := range []float64{math.Inf(1), math.Inf(-1), math.NaN()} {
 		got := attrsJSON([]*common.KeyValue{kvDouble("k", v)})
-		if got != nil {
+		if got != "" {
 			var m map[string]any
-			if err := json.Unmarshal(got, &m); err != nil {
+			if err := json.Unmarshal([]byte(got), &m); err != nil {
 				t.Errorf("attrsJSON(%v) produced invalid JSON: %s", v, got)
 			}
 		}
@@ -110,7 +111,7 @@ func TestAttrsJSON_ConcurrentSafe(t *testing.T) {
 		go func() {
 			defer wg.Done()
 			for j := 0; j < 1000; j++ {
-				if got := attrsJSON(attrs); !bytes.Equal(got, want) {
+				if got := attrsJSON(attrs); got != want {
 					t.Errorf("concurrent attrsJSON = %s, want %s", got, want)
 					return
 				}
@@ -179,7 +180,7 @@ func TestAttrsJSON_FlatObject(t *testing.T) {
 	got := attrsJSON(attrs)
 
 	var m map[string]any
-	if err := json.Unmarshal(got, &m); err != nil {
+	if err := json.Unmarshal([]byte(got), &m); err != nil {
 		t.Fatalf("output is not a JSON object: %v (%s)", err, got)
 	}
 	if m["http.method"] != "GET" {
@@ -194,10 +195,10 @@ func TestAttrsJSON_FlatObject(t *testing.T) {
 }
 
 func TestAttrsJSON_EmptyIsNil(t *testing.T) {
-	if got := attrsJSON(nil); got != nil {
+	if got := attrsJSON(nil); got != "" {
 		t.Errorf("attrsJSON(nil) = %s, want nil", got)
 	}
-	if got := attrsJSON([]*common.KeyValue{}); got != nil {
+	if got := attrsJSON([]*common.KeyValue{}); got != "" {
 		t.Errorf("attrsJSON([]) = %s, want nil", got)
 	}
 }
@@ -209,7 +210,7 @@ func TestAttrsJSON_NestedKvlist(t *testing.T) {
 		}}},
 	}
 	var m map[string]any
-	if err := json.Unmarshal(attrsJSON(attrs), &m); err != nil {
+	if err := json.Unmarshal([]byte(attrsJSON(attrs)), &m); err != nil {
 		t.Fatalf("unmarshal: %v", err)
 	}
 	outer, ok := m["outer"].(map[string]any)

internal/ingest/server.go

@@ -347,16 +347,16 @@ func spanDurationMS(startNano, endNano uint64) float64 {
 	return float64(endNano-startNano) / 1e6
 }
 
-func toJSON(v interface{}) []byte {
+func toJSON(v interface{}) string {
 	if v == nil {
-		return nil
+		return ""
 	}
 	b, err := json.Marshal(v)
 	if err != nil {
 		slog.Error("json marshal failed", "err", err)
-		return []byte("null")
+		return "null"
 	}
-	return b
+	return string(b)
 }
 
 // attrsJSON flattens an OTLP attribute list into a flat JSON object keyed by the
@@ -378,9 +378,9 @@ var attrBufPool = sync.Pool{New: func() any { return new(bytes.Buffer) }}
 // whereas the reflect fallback (a map) sorts keys and keeps last-wins. This is
 // immaterial for fanout: OTLP attribute keys are unique by spec, and queries
 // read attributes_json by key via attr()/json_extract (order-independent).
-func attrsJSON(attrs []*common.KeyValue) []byte {
+func attrsJSON(attrs []*common.KeyValue) string {
 	if len(attrs) == 0 {
-		return nil
+		return ""
 	}
 	if attrsNeedReflect(attrs) {
 		return attrsJSONReflect(attrs)
@@ -402,10 +402,10 @@ func attrsJSON(attrs []*common.KeyValue) []byte {
 		appendScalarJSON(buf, kv.Value)
 		n++
 	}
-	var out []byte
+	var out string
 	if n > 0 {
 		buf.WriteByte('}')
-		out = append([]byte(nil), buf.Bytes()...) // copy out before returning buf to the pool
+		out = buf.String() // copies out of the pooled buffer, as []byte did
 	}
 	attrBufPool.Put(buf)
 	return out
@@ -536,7 +536,7 @@ func appendJSONString(buf *bytes.Buffer, s string) {
 }
 
 // attrsJSONReflect is the reflection-based encoder, retained for nested values.
-func attrsJSONReflect(attrs []*common.KeyValue) []byte {
+func attrsJSONReflect(attrs []*common.KeyValue) string {
 	m := make(map[string]any, len(attrs))
 	for _, kv := range attrs {
 		if kv == nil || kv.Key == "" {
@@ -545,21 +545,21 @@ func attrsJSONReflect(attrs []*common.KeyValue) []byte {
 		m[kv.Key] = attrValue(kv.Value)
 	}
 	if len(m) == 0 {
-		return nil
+		return ""
 	}
 	b, err := json.Marshal(m)
 	if err != nil {
 		slog.Error("attrs json marshal failed", "err", err)
-		return nil
+		return ""
 	}
-	return b
+	return string(b)
 }
 
 // resourceAttrsJSON flattens a resource's attributes into the same flat object
 // shape as attrsJSON, so attr(resource_json, 'key') resolves.
-func resourceAttrsJSON(r *resourcepb.Resource) []byte {
+func resourceAttrsJSON(r *resourcepb.Resource) string {
 	if r == nil {
-		return nil
+		return ""
 	}
 	return attrsJSON(r.Attributes)
 }
@@ -811,9 +811,9 @@ func hexOrEmpty(b []byte) string {
 	return fmt.Sprintf("%x", b)
 }
 
-func eventsToJSON(events []*tracepb.Span_Event) []byte {
+func eventsToJSON(events []*tracepb.Span_Event) string {
 	if len(events) == 0 {
-		return nil
+		return ""
 	}
 	type evt struct {
 		Time       int64             `json:"time_unix_nano"`
@@ -837,14 +837,14 @@ func eventsToJSON(events []*tracepb.Span_Event) []byte {
 	b, err := json.Marshal(out)
 	if err != nil {
 		slog.Error("json marshal events failed", "err", err)
-		return nil
+		return ""
 	}
-	return b
+	return string(b)
 }
 
-func linksToJSON(links []*tracepb.Span_Link) []byte {
+func linksToJSON(links []*tracepb.Span_Link) string {
 	if len(links) == 0 {
-		return nil
+		return ""
 	}
 	type link struct {
 		TraceID    string            `json:"trace_id"`
@@ -870,9 +870,9 @@ func linksToJSON(links []*tracepb.Span_Link) []byte {
 	b, err := json.Marshal(out)
 	if err != nil {
 		slog.Error("json marshal links failed", "err", err)
-		return nil
+		return ""
 	}
-	return b
+	return string(b)
 }
 
 func scopeInfo(scope *common.InstrumentationScope) (name, version string) {
@@ -882,9 +882,9 @@ func scopeInfo(scope *common.InstrumentationScope) (name, version string) {
 	return scope.Name, scope.Version
 }
 
-func exemplarsToJSON(exemplars []*metricspb.Exemplar) []byte {
+func exemplarsToJSON(exemplars []*metricspb.Exemplar) string {
 	if len(exemplars) == 0 {
-		return nil
+		return ""
 	}
 	type ex struct {
 		Time       int64             `json:"time_unix_nano"`
@@ -919,9 +919,9 @@ func exemplarsToJSON(exemplars []*metricspb.Exemplar) []byte {
 	b, err := json.Marshal(out)
 	if err != nil {
 		slog.Error("json marshal exemplars failed", "err", err)
-		return nil
+		return ""
 	}
-	return b
+	return string(b)
 }
 
 func expHistBuckets(dp *metricspb.ExponentialHistogramDataPoint) []float64 {

internal/ingest/server_test.go

@@ -32,10 +32,10 @@ func TestToJSON(t *testing.T) {
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
 			result := toJSON(tc.input)
-			if tc.nilOut && result != nil {
+			if tc.nilOut && result != "" {
 				t.Errorf("toJSON(%v) = %v, want nil", tc.input, result)
 			}
-			if !tc.nilOut && result == nil {
+			if !tc.nilOut && result == "" {
 				t.Errorf("toJSON(%v) = nil, want non-nil", tc.input)
 			}
 		})
@@ -209,10 +209,10 @@ func TestEventsToJSON(t *testing.T) {
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
 			result := eventsToJSON(tc.events)
-			if tc.nilOut && result != nil {
+			if tc.nilOut && result != "" {
 				t.Errorf("eventsToJSON() = %v, want nil", result)
 			}
-			if !tc.nilOut && result == nil {
+			if !tc.nilOut && result == "" {
 				t.Error("eventsToJSON() = nil, want non-nil")
 			}
 		})
@@ -247,10 +247,10 @@ func TestLinksToJSON(t *testing.T) {
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
 			result := linksToJSON(tc.links)
-			if tc.nilOut && result != nil {
+			if tc.nilOut && result != "" {
 				t.Errorf("linksToJSON() = %v, want nil", result)
 			}
-			if !tc.nilOut && result == nil {
+			if !tc.nilOut && result == "" {
 				t.Error("linksToJSON() = nil, want non-nil")
 			}
 		})
@@ -319,10 +319,10 @@ func TestExemplarsToJSON(t *testing.T) {
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
 			result := exemplarsToJSON(tc.exemplars)
-			if tc.nilOut && result != nil {
+			if tc.nilOut && result != "" {
 				t.Errorf("exemplarsToJSON() = %v, want nil", result)
 			}
-			if !tc.nilOut && result == nil {
+			if !tc.nilOut && result == "" {
 				t.Error("exemplarsToJSON() = nil, want non-nil")
 			}
 		})

internal/telemetry/commit_memory_test.go

@@ -62,12 +62,12 @@ func syntheticSpan(i int, base time.Time) Span {
 		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)),
+		ResourceJSON:   `{"service.name":"checkout","deployment.environment":"prod","host.name":"cube-10"}`,
+		AttributesJSON: 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: fmt.Sprintf(
+			`[{"name":"validated","time":%d},{"name":"charged","time":%d,"amount":%d}]`, start.UnixNano(), start.UnixNano()+1000, i%9999),
+		LinksJSON: fmt.Sprintf(`[{"trace_id":"%032x","span_id":"%016x"}]`, i+1, i+1),
 	}
 }
 

internal/telemetry/conversion_alloc_test.go

@@ -0,0 +1,60 @@
+//go:build !race
+
+package telemetry
+
+import (
+	"runtime"
+	"testing"
+	"time"
+	"unsafe"
+)
+
+// Converting a batch into Parquet rows must not duplicate the payload it was
+// handed. The JSON columns are the bulk of a span, the originals stay live
+// until the commit is durably acknowledged, and four commit workers run at
+// once -- so a copy here is four copies of the batch resident at the moment
+// the process is most likely to be killed.
+//
+// The end-to-end commit gate cannot see this on its own: parquet-go and zstd
+// pool their buffers, so its figure swings between 3.2x and 5.3x payload run
+// to run and a copy of this size hides inside that spread. Measuring the
+// conversion alone is quiet enough to assert on.
+func TestSpanConversionDoesNotCopyJSONPayloads(t *testing.T) {
+	base := time.Date(2026, 9, 20, 12, 0, 0, 0, time.UTC)
+	spans := make([]Span, 20_000)
+	for i := range spans {
+		spans[i] = syntheticSpan(i, base)
+	}
+	var jsonBytes int64
+	for i := range spans {
+		jsonBytes += int64(len(spans[i].ResourceJSON) + len(spans[i].AttributesJSON) +
+			len(spans[i].EventsJSON) + len(spans[i].LinksJSON))
+	}
+
+	var before, after runtime.MemStats
+	runtime.GC()
+	runtime.ReadMemStats(&before)
+
+	rows := make([]spanParquetRow, len(spans))
+	for i := range spans {
+		rows[i] = makeSpanParquetRow(spans[i])
+	}
+
+	runtime.ReadMemStats(&after)
+	runtime.KeepAlive(rows)
+	runtime.KeepAlive(spans)
+
+	allocated := int64(after.TotalAlloc - before.TotalAlloc)
+	headers := int64(len(spans)) * int64(spanParquetRowSize())
+	t.Logf("json payload %.1f MiB, row headers %.1f MiB, allocated %.1f MiB",
+		float64(jsonBytes)/(1<<20), float64(headers)/(1<<20), float64(allocated)/(1<<20))
+
+	// Allowing the row slice plus a margin, and nothing like the payload again.
+	limit := headers + jsonBytes/4
+	if allocated > limit {
+		t.Errorf("conversion allocated %.1f MiB against a %.1f MiB JSON payload; it is copying the payload rather than referencing it",
+			float64(allocated)/(1<<20), float64(jsonBytes)/(1<<20))
+	}
+}
+
+func spanParquetRowSize() uintptr { return unsafe.Sizeof(spanParquetRow{}) }

internal/telemetry/parquet_rows.go

@@ -56,7 +56,7 @@ func makeSpanParquetRow(r Span) spanParquetRow {
 		Service: r.ServiceName, Operation: r.Name, Kind: r.Kind, StartTime: r.StartUnixNanos,
 		EndTime: r.EndUnixNanos, StartUnixNano: r.StartUnixNanos, EndUnixNano: r.EndUnixNanos,
 		DurationMS: r.DurationMS, Status: r.StatusCode, StatusMessage: r.StatusMsg,
-		ResourceJSON: string(r.ResourceJSON), AttributesJSON: string(r.AttributesJSON), EventsJSON: string(r.EventsJSON), LinksJSON: string(r.LinksJSON),
+		ResourceJSON: r.ResourceJSON, AttributesJSON: r.AttributesJSON, EventsJSON: r.EventsJSON, LinksJSON: r.LinksJSON,
 		TraceState: r.TraceState, Flags: int64(r.Flags), ScopeName: r.ScopeName, ScopeVersion: r.ScopeVersion,
 		IngestedAt: r.IngestedAt, IngestedUnixNano: r.IngestedAt, HTTPMethod: r.HTTPMethod,
 		HTTPStatusCode: r.HTTPStatusCode, HTTPRoute: r.HTTPRoute, DBSystem: r.DBSystem, RPCMethod: r.RPCMethod,
@@ -115,7 +115,7 @@ func makeLogParquetRow(r Log) logParquetRow {
 		ObservedTime: FirstPositiveNanos(r.ObservedTimeNanos, r.EventUnixNanos, r.TimeUnixNanos, r.IngestedAt), TimeUnixNano: r.TimeUnixNanos,
 		ObservedTimeUnixNano: r.ObservedTimeNanos, Severity: r.Severity, SeverityNumber: int64(r.SeverityNumber),
 		Body: r.Body, Service: r.ServiceName, TraceID: r.TraceID, SpanID: r.SpanID, Flags: int64(r.Flags),
-		ResourceJSON: string(r.ResourceJSON), AttributesJSON: string(r.AttributesJSON), ScopeName: r.ScopeName,
+		ResourceJSON: r.ResourceJSON, AttributesJSON: r.AttributesJSON, ScopeName: r.ScopeName,
 		ScopeVersion: r.ScopeVersion, IngestedAt: r.IngestedAt, IngestedUnixNano: r.IngestedAt, BodyTemplate: r.BodyTemplate,
 	}
 }
@@ -147,9 +147,9 @@ func makeMetricParquetRow(r Metric) metricParquetRow {
 	return metricParquetRow{
 		Namespace: r.Namespace, MetricTime: FirstPositiveNanos(r.EventUnixNanos, r.TimeUnixNanos, r.IngestedAt), TimeUnixNano: r.TimeUnixNanos,
 		Name: r.Name, Description: r.Description, Unit: r.Unit, MetricType: r.Type, Service: r.ServiceName,
-		Value: r.Value, HistBoundsJSON: string(r.HistBoundsJSON), HistCountsJSON: string(r.HistCountsJSON),
-		HistCount: r.HistCount, HistSum: r.HistSum, ExemplarsJSON: string(r.ExemplarsJSON),
-		AttributesJSON: string(r.AttributesJSON), ResourceJSON: string(r.ResourceJSON), ScopeName: r.ScopeName,
+		Value: r.Value, HistBoundsJSON: r.HistBoundsJSON, HistCountsJSON: r.HistCountsJSON,
+		HistCount: r.HistCount, HistSum: r.HistSum, ExemplarsJSON: r.ExemplarsJSON,
+		AttributesJSON: r.AttributesJSON, ResourceJSON: r.ResourceJSON, ScopeName: r.ScopeName,
 		ScopeVersion: r.ScopeVersion, IngestedAt: r.IngestedAt, IngestedUnixNano: r.IngestedAt,
 	}
 }