From 1a33cf046c9bac3d11eec567c147cfaa5a0249b4 Mon Sep 17 00:00:00 2001 From: Vishal Rana Date: Thu, 15 Oct 2026 01:03:38 -0700 Subject: [PATCH 1/2] perf(query): keep compaction, rollups and gates ahead of sustained ingest On a 32-vCPU host at 478k items/s, Parquet files grew from 3,038 to 17,251 in five minutes, DuckDB native memory swung up to 31 GiB with no read load, and version-rollup and anomaly writes timed out on the write gate. - The edge rollup built its parent hash table from every span within an hour of the affected minute. It now semi-joins parents against the distinct traces of the affected children, keeping the one-hour range and identical call edges, so the build side no longer grows with ingest rate. - Compaction starts its next cycle after 100 ms while cycles make progress and returns to the 10-second cadence when idle. - Span compaction builds trace.fidx from the ordered merge stream instead of rereading the output; the index bytes are identical. A 128-input compaction fell from 5.9-6.2 s to 5.1-5.3 s. - Version-marker transfer and retirement use set-based statements, cutting their time under the write gate from about 30 ms to 3 ms for 128 inputs. - OTLP conversion preallocates row slices and hoists per-resource work: 1,000 spans convert in 1.9 ms instead of 2.1 ms with 2.3 MB instead of 3.4 MB allocated. Locally under sustained mixed ingest, the end-of-run file count fell 37%, the service rollup's gate hold fell from 3.9 s to 2.3 s, and the anomaly writer's gate wait halved. Claude-Session: https://claude.ai/code/session_01XA52nWyQ3R5f5kK8kjpAQW --- internal/ingest/conversion_test.go | 164 ++++++ internal/ingest/server.go | 46 +- internal/ingest/testdata/conversion_rows.json | 548 ++++++++++++++++++ internal/query/annotation_maintenance_test.go | 35 ++ internal/query/annotations.go | 20 +- internal/query/duck.go | 86 +-- internal/query/edge_rollup_plan_test.go | 56 +- internal/query/maintenance_schedule_test.go | 64 ++ internal/telemetry/compaction_index_test.go | 152 +++++ internal/telemetry/metric_series_test.go | 2 +- internal/telemetry/parquet.go | 162 +++--- internal/telemetry/store/compaction.go | 10 +- internal/telemetry/store/repository_test.go | 7 +- internal/telemetry/variant_test.go | 2 +- 14 files changed, 1226 insertions(+), 128 deletions(-) create mode 100644 internal/ingest/conversion_test.go create mode 100644 internal/ingest/testdata/conversion_rows.json create mode 100644 internal/query/maintenance_schedule_test.go create mode 100644 internal/telemetry/compaction_index_test.go diff --git a/internal/ingest/conversion_test.go b/internal/ingest/conversion_test.go new file mode 100644 index 00000000..70c1cf43 --- /dev/null +++ b/internal/ingest/conversion_test.go @@ -0,0 +1,164 @@ +package ingest + +import ( + "bytes" + "context" + "encoding/json" + "os" + "testing" + + "github.com/labstack/fanout/internal/config" + telemetrystore "github.com/labstack/fanout/internal/telemetry/store" + collectorlogs "go.opentelemetry.io/proto/otlp/collector/logs/v1" + collectormetrics "go.opentelemetry.io/proto/otlp/collector/metrics/v1" + collectortrace "go.opentelemetry.io/proto/otlp/collector/trace/v1" + common "go.opentelemetry.io/proto/otlp/common/v1" + logspb "go.opentelemetry.io/proto/otlp/logs/v1" + metricspb "go.opentelemetry.io/proto/otlp/metrics/v1" + resourcepb "go.opentelemetry.io/proto/otlp/resource/v1" + tracepb "go.opentelemetry.io/proto/otlp/trace/v1" +) + +type conversionSubmitter func(context.Context, telemetrystore.Batch) error + +func (f conversionSubmitter) Submit(ctx context.Context, batch telemetrystore.Batch) error { + return f(ctx, batch) +} + +func conversionExports(n int) (*collectortrace.ExportTraceServiceRequest, *collectorlogs.ExportLogsServiceRequest, *collectormetrics.ExportMetricsServiceRequest) { + traces := &collectortrace.ExportTraceServiceRequest{} + logs := &collectorlogs.ExportLogsServiceRequest{} + metrics := &collectormetrics.ExportMetricsServiceRequest{} + text := func(k, v string) *common.KeyValue { + return &common.KeyValue{Key: k, Value: &common.AnyValue{Value: &common.AnyValue_StringValue{StringValue: v}}} + } + attrs := []*common.KeyValue{ + text("http.method", "POST"), + text("http.request.method", "GET"), + text("http.route", "/orders"), + {Key: "http.status_code", Value: &common.AnyValue{Value: &common.AnyValue_IntValue{IntValue: 503}}}, + {Key: "flag", Value: &common.AnyValue{Value: &common.AnyValue_BoolValue{BoolValue: true}}}, + {Key: "bytes", Value: &common.AnyValue{Value: &common.AnyValue_BytesValue{BytesValue: []byte{0, 255}}}}, + } + for r := 0; r < 2; r++ { + resource := &resourcepb.Resource{Attributes: []*common.KeyValue{text("service.name", []string{"api", "worker"}[r]), text("service.version", []string{"v1", "v2"}[r]), text("deployment.environment", []string{"prod", "test"}[r]), {Key: "nested", Value: &common.AnyValue{Value: &common.AnyValue_ArrayValue{ArrayValue: &common.ArrayValue{Values: []*common.AnyValue{{Value: &common.AnyValue_IntValue{IntValue: 7}}, {Value: &common.AnyValue_BoolValue{BoolValue: false}}}}}}}}} + if r == 0 { + resource.Attributes = append(resource.Attributes, text("service.namespace", "shop")) + } + rs := &tracepb.ResourceSpans{Resource: resource} + rl := &logspb.ResourceLogs{Resource: resource} + rm := &metricspb.ResourceMetrics{Resource: resource} + for scope := 0; scope < 2; scope++ { + info := &common.InstrumentationScope{Name: []string{"http", "jobs"}[scope], Version: []string{"1.0", "2.0"}[scope]} + ss := &tracepb.ScopeSpans{Scope: info} + sl := &logspb.ScopeLogs{Scope: info} + sm := &metricspb.ScopeMetrics{Scope: info} + for i := 0; i < n; i++ { + ss.Spans = append(ss.Spans, &tracepb.Span{ + TraceId: []byte{1, 2, byte(r)}, SpanId: []byte{3, byte(scope), byte(i)}, ParentSpanId: []byte{4}, + Name: "operation", Kind: tracepb.Span_SPAN_KIND_CLIENT, + StartTimeUnixNano: 1790856000000000017, EndTimeUnixNano: 1790856000001000019, + Status: &tracepb.Status{Code: tracepb.Status_STATUS_CODE_ERROR, Message: "failed"}, + Attributes: attrs, TraceState: "vendor=state", Flags: 257, + Events: []*tracepb.Span_Event{{ + Name: "exception", TimeUnixNano: 1790856000000000020, + Attributes: []*common.KeyValue{text("exception.type", "Timeout"), text("exception.message", "late")}, + }}, + Links: []*tracepb.Span_Link{{ + TraceId: []byte{5}, SpanId: []byte{6}, TraceState: "a=b", + Attributes: []*common.KeyValue{text("link", "value")}, + }}, + }) + sl.LogRecords = append(sl.LogRecords, &logspb.LogRecord{TimeUnixNano: 1790856000000000017, ObservedTimeUnixNano: 1790856000000000019, SeverityNumber: logspb.SeverityNumber_SEVERITY_NUMBER_ERROR, Body: &common.AnyValue{Value: &common.AnyValue_StringValue{StringValue: "order 123 failed"}}, TraceId: []byte{1, 2}, SpanId: []byte{3, 4}, Flags: 1, Attributes: attrs}) + } + point := &metricspb.NumberDataPoint{TimeUnixNano: 1790856000000000017, StartTimeUnixNano: 1790855940000000009, Value: &metricspb.NumberDataPoint_AsInt{AsInt: 12}, Attributes: attrs} + points := make([]*metricspb.NumberDataPoint, n) + for i := range points { + points[i] = point + } + switch r*2 + scope { + case 0: + sm.Metrics = []*metricspb.Metric{{Name: "gauge", Description: "description", Unit: "ms", Data: &metricspb.Metric_Gauge{Gauge: &metricspb.Gauge{DataPoints: points}}}, {Name: "sum", Data: &metricspb.Metric_Sum{Sum: &metricspb.Sum{IsMonotonic: true, AggregationTemporality: metricspb.AggregationTemporality_AGGREGATION_TEMPORALITY_DELTA, DataPoints: points}}}} + case 1: + sm.Metrics = []*metricspb.Metric{{Name: "histogram", Data: &metricspb.Metric_Histogram{Histogram: &metricspb.Histogram{AggregationTemporality: metricspb.AggregationTemporality_AGGREGATION_TEMPORALITY_CUMULATIVE, DataPoints: []*metricspb.HistogramDataPoint{{TimeUnixNano: 1790856000000000017, Count: 2, ExplicitBounds: []float64{1}, BucketCounts: []uint64{1, 1}, Attributes: attrs}}}}}} + case 2: + sm.Metrics = []*metricspb.Metric{{Name: "exponential", Data: &metricspb.Metric_ExponentialHistogram{ExponentialHistogram: &metricspb.ExponentialHistogram{AggregationTemporality: metricspb.AggregationTemporality_AGGREGATION_TEMPORALITY_DELTA, DataPoints: []*metricspb.ExponentialHistogramDataPoint{{TimeUnixNano: 1790856000000000017, Count: 1, Positive: &metricspb.ExponentialHistogramDataPoint_Buckets{BucketCounts: []uint64{1}}, Attributes: attrs}}}}}} + case 3: + sm.Metrics = []*metricspb.Metric{{Name: "summary", Data: &metricspb.Metric_Summary{Summary: &metricspb.Summary{DataPoints: []*metricspb.SummaryDataPoint{{TimeUnixNano: 1790856000000000017, Count: 1, Sum: 3, QuantileValues: []*metricspb.SummaryDataPoint_ValueAtQuantile{{Quantile: 0.5, Value: 3}}, Attributes: attrs}}}}}} + } + rs.ScopeSpans = append(rs.ScopeSpans, ss) + rl.ScopeLogs = append(rl.ScopeLogs, sl) + rm.ScopeMetrics = append(rm.ScopeMetrics, sm) + } + traces.ResourceSpans = append(traces.ResourceSpans, rs) + logs.ResourceLogs = append(logs.ResourceLogs, rl) + metrics.ResourceMetrics = append(metrics.ResourceMetrics, rm) + } + return traces, logs, metrics +} + +func TestExportConversionRowsMatchBaseline(t *testing.T) { + traces, logs, metrics := conversionExports(1) + var got telemetrystore.Batch + srv := NewServer(config.Config{DefaultNamespace: "fallback"}, conversionSubmitter(func(_ context.Context, b telemetrystore.Batch) error { + got.Spans = append(got.Spans, b.Spans...) + got.Logs = append(got.Logs, b.Logs...) + got.Metrics = append(got.Metrics, b.Metrics...) + return nil + })) + if _, err := srv.exportTraces(t.Context(), traces); err != nil { + t.Fatal(err) + } + if _, err := srv.exportLogs(t.Context(), logs); err != nil { + t.Fatal(err) + } + if _, err := srv.exportMetrics(t.Context(), metrics); err != nil { + t.Fatal(err) + } + for i := range got.Spans { + got.Spans[i].IngestedAt = 0 + } + for i := range got.Logs { + got.Logs[i].IngestedAt = 0 + } + for i := range got.Metrics { + got.Metrics[i].IngestedAt = 0 + } + encoded, err := json.MarshalIndent(got, "", " ") + if err != nil { + t.Fatal(err) + } + encoded = append(encoded, '\n') + path := "testdata/conversion_rows.json" + + want, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(encoded, want) { + t.Fatalf("converted rows changed:\n%s", encoded) + } +} +func BenchmarkExportConversion(b *testing.B) { + traces, logs, metrics := conversionExports(250) + srv := NewServer(config.Config{DefaultNamespace: "fallback"}, conversionSubmitter(func(context.Context, telemetrystore.Batch) error { return nil })) + for _, name := range []string{"spans", "logs", "metrics"} { + b.Run(name, func(b *testing.B) { + b.ReportAllocs() + for b.Loop() { + var err error + switch name { + case "spans": + _, err = srv.exportTraces(b.Context(), traces) + case "logs": + _, err = srv.exportLogs(b.Context(), logs) + case "metrics": + _, err = srv.exportMetrics(b.Context(), metrics) + } + if err != nil { + b.Fatal(err) + } + } + }) + } +} diff --git a/internal/ingest/server.go b/internal/ingest/server.go index c2740ccf..ece89a90 100644 --- a/internal/ingest/server.go +++ b/internal/ingest/server.go @@ -72,7 +72,16 @@ func (ts *traceService) Export(ctx context.Context, req *collectortrace.ExportTr func (s *Server) exportTraces(ctx context.Context, req *collectortrace.ExportTraceServiceRequest) (*collectortrace.ExportTraceServiceResponse, error) { cfg := s.cfg now := time.Now().UnixNano() + count := 0 + for _, rs := range req.ResourceSpans { + for _, ss := range rs.ScopeSpans { + count += len(ss.Spans) + } + } batch := telemetrystore.Batch{} + if count > 0 { + batch.Spans = make([]telemetry.Span, 0, count) + } for _, rs := range req.ResourceSpans { resources, err := resourceAttributes(rs.Resource) if err != nil { @@ -83,6 +92,8 @@ func (s *Server) exportTraces(ctx context.Context, req *collectortrace.ExportTra if namespace == "" { namespace = cfg.DefaultNamespace } + serviceVersion := getResourceAttr(rs.Resource, "service.version") + deploymentEnv := getResourceAttr(rs.Resource, "deployment.environment") for _, ss := range rs.ScopeSpans { scopeName, scopeVer := scopeInfo(ss.Scope) for _, sp := range ss.Spans { @@ -124,8 +135,8 @@ func (s *Server) exportTraces(ctx context.Context, req *collectortrace.ExportTra RPCMethod: spanAttr(sp.Attributes, "rpc.method"), RPCService: spanAttr(sp.Attributes, "rpc.service"), PeerService: spanAttr(sp.Attributes, "peer.service", "server.address"), - ServiceVersion: getResourceAttr(rs.Resource, "service.version"), - DeploymentEnv: getResourceAttr(rs.Resource, "deployment.environment"), + ServiceVersion: serviceVersion, + DeploymentEnv: deploymentEnv, } // Pre-extracted exception info from span events. excType, excMsg := extractException(sp.Events) @@ -161,7 +172,16 @@ func (ls *logsService) Export(ctx context.Context, req *collectorlogs.ExportLogs func (s *Server) exportLogs(ctx context.Context, req *collectorlogs.ExportLogsServiceRequest) (*collectorlogs.ExportLogsServiceResponse, error) { cfg := s.cfg now := time.Now().UnixNano() + count := 0 + for _, rl := range req.ResourceLogs { + for _, sl := range rl.ScopeLogs { + count += len(sl.LogRecords) + } + } batch := telemetrystore.Batch{} + if count > 0 { + batch.Logs = make([]telemetry.Log, 0, count) + } for _, rl := range req.ResourceLogs { resources, err := resourceAttributes(rl.Resource) if err != nil { @@ -218,7 +238,29 @@ func (ms *metricsService) Export(ctx context.Context, req *collectormetrics.Expo func (s *Server) exportMetrics(ctx context.Context, req *collectormetrics.ExportMetricsServiceRequest) (*collectormetrics.ExportMetricsServiceResponse, error) { cfg := s.cfg now := time.Now().UnixNano() + count := 0 + for _, rm := range req.ResourceMetrics { + for _, sm := range rm.ScopeMetrics { + for _, m := range sm.Metrics { + switch d := m.Data.(type) { + case *metricspb.Metric_Gauge: + count += len(d.Gauge.DataPoints) + case *metricspb.Metric_Sum: + count += len(d.Sum.DataPoints) + case *metricspb.Metric_Histogram: + count += len(d.Histogram.DataPoints) + case *metricspb.Metric_ExponentialHistogram: + count += len(d.ExponentialHistogram.DataPoints) + case *metricspb.Metric_Summary: + count += len(d.Summary.DataPoints) + } + } + } + } batch := telemetrystore.Batch{} + if count > 0 { + batch.Metrics = make([]telemetry.Metric, 0, count) + } for _, rm := range req.ResourceMetrics { resources, err := resourceAttributes(rm.Resource) if err != nil { diff --git a/internal/ingest/testdata/conversion_rows.json b/internal/ingest/testdata/conversion_rows.json new file mode 100644 index 00000000..f9b04d33 --- /dev/null +++ b/internal/ingest/testdata/conversion_rows.json @@ -0,0 +1,548 @@ +{ + "ID": "", + "Spans": [ + { + "Namespace": "shop", + "TraceID": "010200", + "SpanID": "030000", + "ParentSpanID": "04", + "ServiceName": "api", + "Name": "operation", + "Kind": "SPAN_KIND_CLIENT", + "StartUnixNanos": 1790856000000000017, + "EndUnixNanos": 1790856000001000019, + "DurationMS": 1.000002, + "StatusCode": "STATUS_CODE_ERROR", + "StatusMsg": "failed", + "Resource": { + "deployment.environment": "prod", + "nested": [ + 7, + false + ], + "service.name": "api", + "service.namespace": "shop", + "service.version": "v1" + }, + "Attributes": { + "bytes": "AP8=", + "flag": true, + "http.method": "POST", + "http.request.method": "GET", + "http.route": "/orders", + "http.status_code": 503 + }, + "EventsJSON": "[{\"time_unix_nano\":1790856000000000020,\"name\":\"exception\",\"attributes\":{\"exception.message\":\"late\",\"exception.type\":\"Timeout\"}}]", + "LinksJSON": "[{\"trace_id\":\"05\",\"span_id\":\"06\",\"trace_state\":\"a=b\",\"attributes\":{\"link\":\"value\"}}]", + "TraceState": "vendor=state", + "Flags": 257, + "ScopeName": "http", + "ScopeVersion": "1.0", + "IngestedAt": 0, + "HTTPMethod": "POST", + "HTTPStatusCode": "503", + "HTTPRoute": "/orders", + "DBSystem": "", + "RPCMethod": "", + "RPCService": "", + "PeerService": "", + "ServiceVersion": "v1", + "DeploymentEnv": "prod", + "ExceptionType": "Timeout", + "ExceptionMessage": "late" + }, + { + "Namespace": "shop", + "TraceID": "010200", + "SpanID": "030100", + "ParentSpanID": "04", + "ServiceName": "api", + "Name": "operation", + "Kind": "SPAN_KIND_CLIENT", + "StartUnixNanos": 1790856000000000017, + "EndUnixNanos": 1790856000001000019, + "DurationMS": 1.000002, + "StatusCode": "STATUS_CODE_ERROR", + "StatusMsg": "failed", + "Resource": { + "deployment.environment": "prod", + "nested": [ + 7, + false + ], + "service.name": "api", + "service.namespace": "shop", + "service.version": "v1" + }, + "Attributes": { + "bytes": "AP8=", + "flag": true, + "http.method": "POST", + "http.request.method": "GET", + "http.route": "/orders", + "http.status_code": 503 + }, + "EventsJSON": "[{\"time_unix_nano\":1790856000000000020,\"name\":\"exception\",\"attributes\":{\"exception.message\":\"late\",\"exception.type\":\"Timeout\"}}]", + "LinksJSON": "[{\"trace_id\":\"05\",\"span_id\":\"06\",\"trace_state\":\"a=b\",\"attributes\":{\"link\":\"value\"}}]", + "TraceState": "vendor=state", + "Flags": 257, + "ScopeName": "jobs", + "ScopeVersion": "2.0", + "IngestedAt": 0, + "HTTPMethod": "POST", + "HTTPStatusCode": "503", + "HTTPRoute": "/orders", + "DBSystem": "", + "RPCMethod": "", + "RPCService": "", + "PeerService": "", + "ServiceVersion": "v1", + "DeploymentEnv": "prod", + "ExceptionType": "Timeout", + "ExceptionMessage": "late" + }, + { + "Namespace": "fallback", + "TraceID": "010201", + "SpanID": "030000", + "ParentSpanID": "04", + "ServiceName": "worker", + "Name": "operation", + "Kind": "SPAN_KIND_CLIENT", + "StartUnixNanos": 1790856000000000017, + "EndUnixNanos": 1790856000001000019, + "DurationMS": 1.000002, + "StatusCode": "STATUS_CODE_ERROR", + "StatusMsg": "failed", + "Resource": { + "deployment.environment": "test", + "nested": [ + 7, + false + ], + "service.name": "worker", + "service.version": "v2" + }, + "Attributes": { + "bytes": "AP8=", + "flag": true, + "http.method": "POST", + "http.request.method": "GET", + "http.route": "/orders", + "http.status_code": 503 + }, + "EventsJSON": "[{\"time_unix_nano\":1790856000000000020,\"name\":\"exception\",\"attributes\":{\"exception.message\":\"late\",\"exception.type\":\"Timeout\"}}]", + "LinksJSON": "[{\"trace_id\":\"05\",\"span_id\":\"06\",\"trace_state\":\"a=b\",\"attributes\":{\"link\":\"value\"}}]", + "TraceState": "vendor=state", + "Flags": 257, + "ScopeName": "http", + "ScopeVersion": "1.0", + "IngestedAt": 0, + "HTTPMethod": "POST", + "HTTPStatusCode": "503", + "HTTPRoute": "/orders", + "DBSystem": "", + "RPCMethod": "", + "RPCService": "", + "PeerService": "", + "ServiceVersion": "v2", + "DeploymentEnv": "test", + "ExceptionType": "Timeout", + "ExceptionMessage": "late" + }, + { + "Namespace": "fallback", + "TraceID": "010201", + "SpanID": "030100", + "ParentSpanID": "04", + "ServiceName": "worker", + "Name": "operation", + "Kind": "SPAN_KIND_CLIENT", + "StartUnixNanos": 1790856000000000017, + "EndUnixNanos": 1790856000001000019, + "DurationMS": 1.000002, + "StatusCode": "STATUS_CODE_ERROR", + "StatusMsg": "failed", + "Resource": { + "deployment.environment": "test", + "nested": [ + 7, + false + ], + "service.name": "worker", + "service.version": "v2" + }, + "Attributes": { + "bytes": "AP8=", + "flag": true, + "http.method": "POST", + "http.request.method": "GET", + "http.route": "/orders", + "http.status_code": 503 + }, + "EventsJSON": "[{\"time_unix_nano\":1790856000000000020,\"name\":\"exception\",\"attributes\":{\"exception.message\":\"late\",\"exception.type\":\"Timeout\"}}]", + "LinksJSON": "[{\"trace_id\":\"05\",\"span_id\":\"06\",\"trace_state\":\"a=b\",\"attributes\":{\"link\":\"value\"}}]", + "TraceState": "vendor=state", + "Flags": 257, + "ScopeName": "jobs", + "ScopeVersion": "2.0", + "IngestedAt": 0, + "HTTPMethod": "POST", + "HTTPStatusCode": "503", + "HTTPRoute": "/orders", + "DBSystem": "", + "RPCMethod": "", + "RPCService": "", + "PeerService": "", + "ServiceVersion": "v2", + "DeploymentEnv": "test", + "ExceptionType": "Timeout", + "ExceptionMessage": "late" + } + ], + "Logs": [ + { + "Namespace": "shop", + "EventUnixNanos": 0, + "TimeUnixNanos": 1790856000000000017, + "ObservedTimeNanos": 1790856000000000019, + "Severity": "ERROR", + "SeverityNumber": 17, + "Body": "order 123 failed", + "ServiceName": "api", + "TraceID": "0102", + "SpanID": "0304", + "Flags": 1, + "Resource": { + "deployment.environment": "prod", + "nested": [ + 7, + false + ], + "service.name": "api", + "service.namespace": "shop", + "service.version": "v1" + }, + "Attributes": { + "bytes": "AP8=", + "flag": true, + "http.method": "POST", + "http.request.method": "GET", + "http.route": "/orders", + "http.status_code": 503 + }, + "ScopeName": "http", + "ScopeVersion": "1.0", + "IngestedAt": 0, + "BodyTemplate": "order \u003cnum\u003e failed" + }, + { + "Namespace": "shop", + "EventUnixNanos": 0, + "TimeUnixNanos": 1790856000000000017, + "ObservedTimeNanos": 1790856000000000019, + "Severity": "ERROR", + "SeverityNumber": 17, + "Body": "order 123 failed", + "ServiceName": "api", + "TraceID": "0102", + "SpanID": "0304", + "Flags": 1, + "Resource": { + "deployment.environment": "prod", + "nested": [ + 7, + false + ], + "service.name": "api", + "service.namespace": "shop", + "service.version": "v1" + }, + "Attributes": { + "bytes": "AP8=", + "flag": true, + "http.method": "POST", + "http.request.method": "GET", + "http.route": "/orders", + "http.status_code": 503 + }, + "ScopeName": "jobs", + "ScopeVersion": "2.0", + "IngestedAt": 0, + "BodyTemplate": "order \u003cnum\u003e failed" + }, + { + "Namespace": "fallback", + "EventUnixNanos": 0, + "TimeUnixNanos": 1790856000000000017, + "ObservedTimeNanos": 1790856000000000019, + "Severity": "ERROR", + "SeverityNumber": 17, + "Body": "order 123 failed", + "ServiceName": "worker", + "TraceID": "0102", + "SpanID": "0304", + "Flags": 1, + "Resource": { + "deployment.environment": "test", + "nested": [ + 7, + false + ], + "service.name": "worker", + "service.version": "v2" + }, + "Attributes": { + "bytes": "AP8=", + "flag": true, + "http.method": "POST", + "http.request.method": "GET", + "http.route": "/orders", + "http.status_code": 503 + }, + "ScopeName": "http", + "ScopeVersion": "1.0", + "IngestedAt": 0, + "BodyTemplate": "order \u003cnum\u003e failed" + }, + { + "Namespace": "fallback", + "EventUnixNanos": 0, + "TimeUnixNanos": 1790856000000000017, + "ObservedTimeNanos": 1790856000000000019, + "Severity": "ERROR", + "SeverityNumber": 17, + "Body": "order 123 failed", + "ServiceName": "worker", + "TraceID": "0102", + "SpanID": "0304", + "Flags": 1, + "Resource": { + "deployment.environment": "test", + "nested": [ + 7, + false + ], + "service.name": "worker", + "service.version": "v2" + }, + "Attributes": { + "bytes": "AP8=", + "flag": true, + "http.method": "POST", + "http.request.method": "GET", + "http.route": "/orders", + "http.status_code": 503 + }, + "ScopeName": "jobs", + "ScopeVersion": "2.0", + "IngestedAt": 0, + "BodyTemplate": "order \u003cnum\u003e failed" + } + ], + "Metrics": [ + { + "SeriesID": 2351255154244030761, + "Namespace": "shop", + "EventUnixNanos": 0, + "TimeUnixNanos": 1790856000000000017, + "Name": "gauge", + "Description": "description", + "Unit": "ms", + "Type": "gauge", + "Temporality": "", + "Monotonic": false, + "StartTimeUnixNanos": 1790855940000000009, + "ServiceName": "api", + "Value": 12, + "HistBoundsJSON": "", + "HistCountsJSON": "", + "HistCount": 0, + "HistSum": 0, + "ExemplarsJSON": "", + "Attributes": { + "bytes": "AP8=", + "flag": true, + "http.method": "POST", + "http.request.method": "GET", + "http.route": "/orders", + "http.status_code": 503 + }, + "Resource": { + "deployment.environment": "prod", + "nested": [ + 7, + false + ], + "service.name": "api", + "service.namespace": "shop", + "service.version": "v1" + }, + "ScopeName": "http", + "ScopeVersion": "1.0", + "IngestedAt": 0 + }, + { + "SeriesID": 8590532344666084636, + "Namespace": "shop", + "EventUnixNanos": 0, + "TimeUnixNanos": 1790856000000000017, + "Name": "sum", + "Description": "", + "Unit": "", + "Type": "sum", + "Temporality": "delta", + "Monotonic": true, + "StartTimeUnixNanos": 1790855940000000009, + "ServiceName": "api", + "Value": 12, + "HistBoundsJSON": "", + "HistCountsJSON": "", + "HistCount": 0, + "HistSum": 0, + "ExemplarsJSON": "", + "Attributes": { + "bytes": "AP8=", + "flag": true, + "http.method": "POST", + "http.request.method": "GET", + "http.route": "/orders", + "http.status_code": 503 + }, + "Resource": { + "deployment.environment": "prod", + "nested": [ + 7, + false + ], + "service.name": "api", + "service.namespace": "shop", + "service.version": "v1" + }, + "ScopeName": "http", + "ScopeVersion": "1.0", + "IngestedAt": 0 + }, + { + "SeriesID": 8703808910938826939, + "Namespace": "shop", + "EventUnixNanos": 0, + "TimeUnixNanos": 1790856000000000017, + "Name": "histogram", + "Description": "", + "Unit": "", + "Type": "histogram", + "Temporality": "cumulative", + "Monotonic": false, + "StartTimeUnixNanos": 0, + "ServiceName": "api", + "Value": 0, + "HistBoundsJSON": "[1]", + "HistCountsJSON": "[1,1]", + "HistCount": 2, + "HistSum": 0, + "ExemplarsJSON": "", + "Attributes": { + "bytes": "AP8=", + "flag": true, + "http.method": "POST", + "http.request.method": "GET", + "http.route": "/orders", + "http.status_code": 503 + }, + "Resource": { + "deployment.environment": "prod", + "nested": [ + 7, + false + ], + "service.name": "api", + "service.namespace": "shop", + "service.version": "v1" + }, + "ScopeName": "jobs", + "ScopeVersion": "2.0", + "IngestedAt": 0 + }, + { + "SeriesID": 8583487415274828271, + "Namespace": "fallback", + "EventUnixNanos": 0, + "TimeUnixNanos": 1790856000000000017, + "Name": "exponential", + "Description": "", + "Unit": "", + "Type": "exponential_histogram", + "Temporality": "delta", + "Monotonic": false, + "StartTimeUnixNanos": 0, + "ServiceName": "worker", + "Value": 0, + "HistBoundsJSON": "[-5e-324,0,1,2]", + "HistCountsJSON": "[0,0,0,1,0]", + "HistCount": 1, + "HistSum": 0, + "ExemplarsJSON": "", + "Attributes": { + "bytes": "AP8=", + "flag": true, + "http.method": "POST", + "http.request.method": "GET", + "http.route": "/orders", + "http.status_code": 503 + }, + "Resource": { + "deployment.environment": "test", + "nested": [ + 7, + false + ], + "service.name": "worker", + "service.version": "v2" + }, + "ScopeName": "http", + "ScopeVersion": "1.0", + "IngestedAt": 0 + }, + { + "SeriesID": 4088968036412745285, + "Namespace": "fallback", + "EventUnixNanos": 0, + "TimeUnixNanos": 1790856000000000017, + "Name": "summary", + "Description": "", + "Unit": "", + "Type": "summary", + "Temporality": "", + "Monotonic": false, + "StartTimeUnixNanos": 0, + "ServiceName": "worker", + "Value": 0, + "HistBoundsJSON": "[0.5]", + "HistCountsJSON": "[3]", + "HistCount": 1, + "HistSum": 3, + "ExemplarsJSON": "", + "Attributes": { + "bytes": "AP8=", + "flag": true, + "http.method": "POST", + "http.request.method": "GET", + "http.route": "/orders", + "http.status_code": 503 + }, + "Resource": { + "deployment.environment": "test", + "nested": [ + 7, + false + ], + "service.name": "worker", + "service.version": "v2" + }, + "ScopeName": "jobs", + "ScopeVersion": "2.0", + "IngestedAt": 0 + } + ] +} diff --git a/internal/query/annotation_maintenance_test.go b/internal/query/annotation_maintenance_test.go index ea433aa9..aed9eb60 100644 --- a/internal/query/annotation_maintenance_test.go +++ b/internal/query/annotation_maintenance_test.go @@ -433,3 +433,38 @@ func TestIdleVersionPassDoesNotWrite(t *testing.T) { t.Fatalf("unused watermark: %d %v", unused, err) } } + +func TestVersionMarkerTransferUsesOneInputSet(t *testing.T) { + for _, warm := range []bool{false, true} { + t.Run(fmt.Sprint(warm), func(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatal(err) + } + defer db.Close() + d := &Duck{DB: db, writeDB: db} + inputs := make([]string, 128) + for i := range inputs { + inputs[i] = fmt.Sprintf("input-%d", i) + } + count := len(inputs) + if !warm { + count-- + } + mock.ExpectBegin() + mock.ExpectQuery(`SELECT count\(\*\) FROM version_rollup_batches WHERE batch_id IN`).WillReturnRows(sqlmock.NewRows([]string{"count"}).AddRow(count)) + if warm { + mock.ExpectExec(`INSERT INTO version_rollup_batches`).WithArgs("output", int64(2)).WillReturnResult(sqlmock.NewResult(0, 1)) + mock.ExpectCommit() + } else { + mock.ExpectRollback() + } + if err := d.transferVersionMarker(t.Context(), telemetry.BatchMetadata{ID: "output", MaxIngestedNanos: 2}, inputs); err != nil { + t.Fatal(err) + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatal(err) + } + }) + } +} diff --git a/internal/query/annotations.go b/internal/query/annotations.go index b26d7960..d3cb542f 100644 --- a/internal/query/annotations.go +++ b/internal/query/annotations.go @@ -228,8 +228,8 @@ func (d *Duck) refreshVersionRollup(ctx context.Context, waitBudget time.Duratio return 0, err } defer func() { _ = tx.Rollback() }() - for _, id := range retired { - if _, err := tx.ExecContext(ctx, `DELETE FROM version_rollup_batches WHERE batch_id=?`, id); err != nil { + if len(retired) > 0 { + if _, err := tx.ExecContext(ctx, `DELETE FROM version_rollup_batches WHERE batch_id IN (`+idsSQL(retired)+`)`); err != nil { return 0, err } } @@ -513,14 +513,14 @@ func (d *Duck) transferVersionMarker(ctx context.Context, output telemetry.Batch return err } defer func() { _ = tx.Rollback() }() - for _, id := range inputs { - var count int - if err := tx.QueryRowContext(ctx, `SELECT count(*) FROM version_rollup_batches WHERE batch_id=?`, id).Scan(&count); err != nil { - return err - } - if count == 0 { - return nil - } // The full output must be read when any input is cold. + // The full output must be read when any input is cold. Check the complete + // input set in one statement while retaining the publication transaction. + var count int + if err := tx.QueryRowContext(ctx, `SELECT count(*) FROM version_rollup_batches WHERE batch_id IN (`+idsSQL(inputs)+`)`).Scan(&count); err != nil { + return err + } + if count != len(inputs) { + return nil } if _, err := tx.ExecContext(ctx, `INSERT INTO version_rollup_batches VALUES (?,?) ON CONFLICT(batch_id) DO UPDATE SET max_ingested=greatest(version_rollup_batches.max_ingested,excluded.max_ingested)`, output.ID, output.MaxIngestedNanos); err != nil { return err diff --git a/internal/query/duck.go b/internal/query/duck.go index e2f2c72f..0789ab23 100644 --- a/internal/query/duck.go +++ b/internal/query/duck.go @@ -635,38 +635,53 @@ func (d *Duck) runMaintenanceLoop(ctx context.Context) { } d.updateParquetStats(ctx) } + runMaintenanceSchedule(ctx, every, run, func() int { + compacted, err := d.runParquetCompactionCycle(ctx) + if err != nil && ctx.Err() == nil { + slog.Warn("telemetry compaction cycle failed", "err", err) + } + return compacted + }) +} + +// Keep compaction busy while a cycle makes progress, then poll at the normal +// idle cadence. The independent maintenance ticker retains the hourly pass. +func runMaintenanceSchedule(ctx context.Context, every time.Duration, run func(), compact func() int) { run() maintenanceTicker := time.NewTicker(every) defer maintenanceTicker.Stop() - var compactionTicker *time.Ticker + var compactionTimer *time.Timer var compactionTick <-chan time.Time if every > parquetCompactionCycle { - compactionTicker = time.NewTicker(parquetCompactionCycle) - compactionTick = compactionTicker.C - defer compactionTicker.Stop() + compactionTimer = time.NewTimer(parquetCompactionCycle) + compactionTick = compactionTimer.C + defer compactionTimer.Stop() } for { select { case <-maintenanceTicker.C: run() case <-compactionTick: - if err := d.runParquetCompactionCycle(ctx); err != nil && ctx.Err() == nil { - slog.Warn("telemetry compaction cycle failed", "err", err) + compacted := compact() + delay := parquetCompactionCycle + if compacted > 0 { + delay = 100 * time.Millisecond } + compactionTimer.Reset(delay) case <-ctx.Done(): return } } } -func (d *Duck) runParquetCompactionCycle(ctx context.Context) error { +func (d *Duck) runParquetCompactionCycle(ctx context.Context) (int, error) { if d.repository == nil { - return nil + return 0, nil } started := time.Now() if err := d.repository.RecoverParquet(ctx, d); err != nil { metrics.RecordTelemetryOperation(metrics.TelemetryCompaction, metrics.TelemetryError, time.Since(started).Seconds()) - return err + return 0, err } compacted, err := d.repository.CompactParquetPass(ctx, d, parquetMaintenanceBatchLimit, parquetCompactionCycleBudget) result := metrics.TelemetryNoop @@ -676,7 +691,7 @@ func (d *Duck) runParquetCompactionCycle(ctx context.Context) error { result = metrics.TelemetrySuccess } metrics.RecordTelemetryOperation(metrics.TelemetryCompaction, result, time.Since(started).Seconds()) - return err + return compacted, err } func (d *Duck) runRepositoryMaintenance(ctx context.Context) error { @@ -1496,25 +1511,31 @@ WITH affected AS ( bounds AS ( SELECT MIN(bucket) AS lo, MAX(bucket) AS hi FROM affected ), --- The parent side is range-filtered HERE rather than in the join's ON --- clause, which keeps the hash build off every span ever ingested. The --- affected window is one minute of newly ingested spans; the table behind --- this view is the whole retention period, so the two differ by orders of --- magnitude and the ON-clause form pays for the difference on every pass. --- --- Measured (TestEdgeRollupPlanUsesHashJoins, spans as a Parquet-backed view, --- 600k spans over 25 days, one affected minute): 14ms here against 33ms with --- the range predicate in the ON clause. Both plan a HASH_JOIN -- the cost is --- the size of the build side, not the join algorithm. --- --- Parents more than 1h outside the affected bucket range are dropped, which --- is acceptable for minute-bucket dependency edges. Caveat retained from the --- original: buckets come from span start_time while the window bounds --- ingested_unix_nano, so one late-arriving span with an old start_time --- widens this range for the pass that ingests it. +-- Only traces with children in affected buckets can contribute call edges. +-- Bound this distinct relation to the same child window as call_edges, before +-- building the parent side; unrelated traces in the two-hour range must not +-- occupy the parent hash table. Namespace is part of the trace identity. +child_traces AS ( + SELECT DISTINCT child.namespace, child.trace_id + FROM spans child + JOIN affected a + ON a.namespace = child.namespace + AND a.bucket = date_trunc('minute', child.start_time::TIMESTAMP_NS) + CROSS JOIN bounds + WHERE child.service IS NOT NULL + AND child.service != '' + AND child.start_time >= (bounds.lo)::TIMESTAMP_NS::TIMESTAMPTZ_NS + AND child.start_time < (bounds.hi + INTERVAL 1 MINUTE)::TIMESTAMP_NS::TIMESTAMPTZ_NS +), +-- Retain the existing +/-1h parent range, including parents outside the child +-- minute. Late arrivals can still widen the affected bucket range. parent_scope AS ( SELECT parent.namespace, parent.span_id, parent.trace_id, parent.service - FROM spans parent, bounds + FROM spans parent + SEMI JOIN child_traces child + ON parent.namespace = child.namespace + AND parent.trace_id = child.trace_id + CROSS JOIN bounds WHERE parent.start_time >= (bounds.lo - INTERVAL 1 HOUR)::TIMESTAMP_NS::TIMESTAMPTZ_NS AND parent.start_time <= (bounds.hi + INTERVAL 1 HOUR)::TIMESTAMP_NS::TIMESTAMPTZ_NS AND parent.service IS NOT NULL @@ -1538,14 +1559,9 @@ call_edges AS ( JOIN affected a ON a.namespace = child.namespace AND a.bucket = date_trunc('minute', child.start_time::TIMESTAMP_NS) - -- bounds joins last, and as an explicit CROSS JOIN. Written as - -- "FROM spans child, bounds JOIN parent_scope ON child...", the comma binds - -- looser than JOIN: that parses as "spans child" comma "(bounds JOIN - -- parent_scope ON ...)", so DuckDB materialises a CROSS_PRODUCT of the child - -- scan with bounds instead of folding the single bounds row into the filters. - -- Same rows either way, so nothing fails and no row assertion notices; on the - -- fixture above it is 21ms against 14ms. TestEdgeRollupPlanUsesHashJoins - -- asserts the CROSS_PRODUCT is absent here and present in the comma form. + -- Keep bounds as an explicit CROSS JOIN after the keyed joins, so the + -- planner can fold its single row into the child filters. The plan test + -- guards against Cartesian or correlated span-to-span joins. CROSS JOIN bounds WHERE child.service IS NOT NULL AND child.service != '' diff --git a/internal/query/edge_rollup_plan_test.go b/internal/query/edge_rollup_plan_test.go index 14aed023..959d8b62 100644 --- a/internal/query/edge_rollup_plan_test.go +++ b/internal/query/edge_rollup_plan_test.go @@ -48,6 +48,9 @@ func TestEdgeRollupPlanUsesHashJoins(t *testing.T) { t.Errorf("edge rollup plans a %s; the FROM clause has regressed to the comma form:\n%s", bad, shippedPlan) } } + if !strings.Contains(shippedPlan, "SEMI") { + t.Errorf("edge rollup lost the child-trace semi-join:\n%s", shippedPlan) + } if !strings.Contains(shippedPlan, "Hash Join") { t.Errorf("edge rollup plans no HASH_JOIN at all:\n%s", shippedPlan) } @@ -96,7 +99,8 @@ func edgeRollupCommaForm(t *testing.T) string { if !strings.Contains(form, "\n CROSS JOIN bounds\n") { t.Fatal("no CROSS JOIN bounds to remove") } - return strings.Replace(form, "\n CROSS JOIN bounds\n", "\n", 1) + i := strings.LastIndex(form, "\n CROSS JOIN bounds\n") + return form[:i] + "\n" + form[i+len("\n CROSS JOIN bounds\n"):] } func explain(t *testing.T, db *sql.DB, query string, args ...any) string { @@ -226,3 +230,53 @@ func seedEdgeRollupFixture(t *testing.T, db *sql.DB) { t.Fatal(err) } } + +func TestEdgeRollupParentTraceScope(t *testing.T) { + db := openTestDuck(t) + if _, err := db.Exec(createEdgeRollupTable); err != nil { + t.Fatal(err) + } + if _, err := db.Exec(`CREATE TABLE spans AS + SELECT namespace,trace_id,span_id,parent_span_id,service,kind,status,duration_ms, + start_time::TIMESTAMPTZ_NS start_time,ingested_unix_nano,NULL::VARCHAR messaging_system,NULL::VARCHAR messaging_destination + FROM (VALUES + ('shop','trace','parent','','upstream','SPAN_KIND_SERVER','OK',1.0,TIMESTAMP '2026-09-20 11:30:00',1), + ('other','trace','parent','','wrong-namespace','SPAN_KIND_SERVER','OK',1.0,TIMESTAMP '2026-09-20 11:30:00',1), + ('shop','trace','child','parent','downstream','SPAN_KIND_CLIENT','ERROR',30.0,TIMESTAMP '2026-09-20 12:00:00',100), + ('shop','trace','late-child','parent','downstream','SPAN_KIND_CLIENT','OK',10.0,TIMESTAMP '2026-09-20 12:00:01',1), + ('shop','outside','parent','','too-old','SPAN_KIND_SERVER','OK',1.0,TIMESTAMP '2026-09-20 10:59:59',1), + ('shop','outside','child','parent','downstream','SPAN_KIND_CLIENT','OK',1.0,TIMESTAMP '2026-09-20 12:00:00',100), + ('shop','unused','parent','','unrelated','SPAN_KIND_SERVER','OK',1.0,TIMESTAMP '2026-09-20 11:30:00',1), + ('shop','trace','empty-child','parent','','SPAN_KIND_CLIENT','OK',1.0,TIMESTAMP '2026-09-20 12:00:00',100) + ) t(namespace,trace_id,span_id,parent_span_id,service,kind,status,duration_ms,start_time,ingested_unix_nano)`); err != nil { + t.Fatal(err) + } + lo := time.Date(2026, 9, 20, 12, 0, 0, 0, time.UTC) + if _, err := db.Exec(edgeRollupInsertSQL, 50, 200, lo, lo.Add(time.Minute)); err != nil { + t.Fatal(err) + } + var namespace, caller, callee, edgeType string + var bucket time.Time + var calls int + var avg, errorRate float64 + rows, err := db.Query(`SELECT namespace,bucket,caller,callee,calls,avg_ms,error_rate,edge_type FROM edge_rollup`) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + if !rows.Next() { + t.Fatal("no call edge") + } + if err := rows.Scan(&namespace, &bucket, &caller, &callee, &calls, &avg, &errorRate, &edgeType); err != nil { + t.Fatal(err) + } + if namespace != "shop" || !bucket.Equal(lo) || caller != "upstream" || callee != "downstream" || calls != 2 || avg != 20 || errorRate != 0.5 || edgeType != "call" { + t.Fatalf("edge=%s %s %s %s %d %g %g %s", namespace, bucket, caller, callee, calls, avg, errorRate, edgeType) + } + if rows.Next() { + t.Fatal("unexpected edge from unrelated, cross-namespace, or out-of-range parent") + } + if err := rows.Err(); err != nil { + t.Fatal(err) + } +} diff --git a/internal/query/maintenance_schedule_test.go b/internal/query/maintenance_schedule_test.go new file mode 100644 index 00000000..7a00b1e7 --- /dev/null +++ b/internal/query/maintenance_schedule_test.go @@ -0,0 +1,64 @@ +package query + +import ( + "context" + "reflect" + "testing" + "testing/synctest" + "time" +) + +func TestMaintenanceCompactionDrainsBacklogThenIdles(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + start := time.Now() + var cycles []time.Duration + var maintenance []time.Duration + done := make(chan struct{}) + go func() { + defer close(done) + runMaintenanceSchedule(ctx, time.Hour, func() { maintenance = append(maintenance, time.Since(start)) }, func() int { + cycles = append(cycles, time.Since(start)) + if len(cycles) <= 2 { + return 1 + } + return 0 + }) + }() + time.Sleep(21 * time.Second) + synctest.Wait() + want := []time.Duration{10 * time.Second, 10100 * time.Millisecond, 10200 * time.Millisecond, 20200 * time.Millisecond} + if !reflect.DeepEqual(cycles, want) { + t.Fatalf("cycles=%v want=%v", cycles, want) + } + t.Logf("backlogged cycle gaps: %s, %s; idle gap: %s", cycles[1]-cycles[0], cycles[2]-cycles[1], cycles[3]-cycles[2]) + time.Sleep(time.Hour - 21*time.Second) + synctest.Wait() + if !reflect.DeepEqual(maintenance, []time.Duration{0, time.Hour}) { + t.Fatalf("maintenance=%v", maintenance) + } + cancel() + synctest.Wait() + select { + case <-done: + default: + t.Fatal("cancellation did not stop scheduler") + } + }) +} +func TestMaintenanceShortIntervalNeedsNoCompactionTimer(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + passes := 0 + go runMaintenanceSchedule(ctx, time.Second, func() { passes++ }, func() int { t.Error("unexpected independent compaction"); return 1 }) + time.Sleep(3 * time.Second) + synctest.Wait() + if passes != 4 { + t.Fatalf("passes=%d", passes) + } + cancel() + synctest.Wait() + }) +} diff --git a/internal/telemetry/compaction_index_test.go b/internal/telemetry/compaction_index_test.go new file mode 100644 index 00000000..c290a730 --- /dev/null +++ b/internal/telemetry/compaction_index_test.go @@ -0,0 +1,152 @@ +package telemetry + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "os" + "path/filepath" + "testing" + + "github.com/parquet-go/parquet-go" +) + +type traceParquetRow struct { + TraceHash uint64 `parquet:"_trace_hash"` + StartUnixNano int64 `parquet:"start_unix_nano"` +} + +func TestCompactionStreamingIndexMatchesReread(t *testing.T) { + store, err := OpenParquetStore(t.TempDir()) + if err != nil { + t.Fatal(err) + } + var inputs []string + for part := 0; part < 4; part++ { + spans := make([]Span, 5000) + 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. + n := part*len(spans) + i + spans[i] = Span{Namespace: fmt.Sprint("ns-", n%3), TraceID: fmt.Sprint("trace-", n%97), SpanID: fmt.Sprint("span-", n), StartUnixNanos: int64(n%11) - 5, EndUnixNanos: int64(n%11) + 5, IngestedAt: 100, Resource: map[string]any{"bytes": []byte{0, 255}, "typed": []any{int64(n), false}}, Attributes: map[string]any{"n": int64(n)}} + } + id := fmt.Sprint("input-", part) + if err := store.CommitBatch(t.Context(), BatchMetadata{ID: id}, spans, nil, nil); err != nil { + t.Fatal(err) + } + inputs = append(inputs, filepath.Join(store.BatchPath(id), "spans.parquet")) + } + dir := filepath.Join(t.TempDir(), "output"+BatchSuffix) + if err := os.Mkdir(dir, 0755); err != nil { + t.Fatal(err) + } + output := filepath.Join(dir, "spans.parquet") + stats, err := store.MergeParquet(t.Context(), "spans", inputs, output) + if err != nil { + t.Fatal(err) + } + if stats.Rows != 20000 || stats.MinSpanStartNanos != 1 || stats.MaxSpanStartNanos != 5 { + t.Fatalf("merge stats=%+v", stats) + } + // This is the old narrow reread, retained only as an independent oracle. + file, err := os.Open(output) + if err != nil { + t.Fatal(err) + } + defer file.Close() + reader := parquet.NewGenericReader[traceParquetRow](file) + defer reader.Close() + wantPath := filepath.Join(t.TempDir(), "trace.fidx") + index, err := newTraceIndexWriter(wantPath) + if err != nil { + t.Fatal(err) + } + defer index.Abort() + buffer := make([]traceParquetRow, 8192) + for { + n, err := reader.Read(buffer) + for _, row := range buffer[:n] { + if err := index.Append(row.TraceHash); err != nil { + t.Fatal(err) + } + } + if errors.Is(err, io.EOF) { + break + } + if err != nil { + t.Fatal(err) + } + if n == 0 { + t.Fatal(io.ErrNoProgress) + } + } + if err := index.Close(); err != nil { + t.Fatal(err) + } + want, err := os.ReadFile(wantPath) + if err != nil { + t.Fatal(err) + } + got, err := os.ReadFile(filepath.Join(dir, "trace.fidx")) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(got, want) { + t.Fatal("streaming trace.fidx differs from reread") + } + metadata := BatchMetadata{ID: "output", Spans: int(stats.Rows), MinSpanStartNanos: stats.MinSpanStartNanos, MaxSpanStartNanos: stats.MaxSpanStartNanos} + if err := store.PrepareReplacement(dir, metadata); err != nil { + t.Fatal(err) + } + verifier, err := NewBatchVerifier(t.Context(), filepath.Dir(dir)) + if err != nil { + t.Fatal(err) + } + defer verifier.Close() + if err := verifier.Validate(t.Context(), dir); err != nil { + t.Fatal(err) + } +} + +type cancelMergeContext struct { + context.Context + cancel context.CancelFunc + checks int +} + +func (c *cancelMergeContext) Err() error { + c.checks++ + if c.checks == 3 { + c.cancel() + } + return c.Context.Err() +} + +func TestCancelledSpanMergeRemovesOutputAndIndex(t *testing.T) { + store, err := OpenParquetStore(t.TempDir()) + if err != nil { + t.Fatal(err) + } + spans := make([]Span, 5000) + for i := range spans { + spans[i] = Span{TraceID: fmt.Sprint("trace-", i), SpanID: fmt.Sprint(i), StartUnixNanos: int64(i), IngestedAt: 100} + } + if err := store.CommitBatch(t.Context(), BatchMetadata{ID: "input"}, spans, nil, nil); err != nil { + t.Fatal(err) + } + base, cancel := context.WithCancel(t.Context()) + defer cancel() + ctx := &cancelMergeContext{Context: base, cancel: cancel} + dir := t.TempDir() + // Cancel between reads, after rows have reached both the output writer and + // the index. Neither partial artifact may survive a failed merge. + if _, err := store.MergeParquet(ctx, "spans", []string{filepath.Join(store.BatchPath("input"), "spans.parquet")}, filepath.Join(dir, "spans.parquet")); !errors.Is(err, context.Canceled) { + t.Fatalf("error=%v", err) + } + entries, err := os.ReadDir(dir) + if err != nil || len(entries) != 0 { + t.Fatalf("artifacts=%v error=%v", entries, err) + } +} diff --git a/internal/telemetry/metric_series_test.go b/internal/telemetry/metric_series_test.go index 84603697..7629ada8 100644 --- a/internal/telemetry/metric_series_test.go +++ b/internal/telemetry/metric_series_test.go @@ -43,7 +43,7 @@ func TestMetricSeriesIDSurvivesParquetCompaction(t *testing.T) { want = append(want, m.SeriesID) } output := filepath.Join(t.TempDir(), "metrics.parquet") - if err := store.MergeParquet(t.Context(), "metrics", inputs, output); err != nil { + if _, err := store.MergeParquet(t.Context(), "metrics", inputs, output); err != nil { t.Fatal(err) } rows, err := parquet.ReadFile[struct { diff --git a/internal/telemetry/parquet.go b/internal/telemetry/parquet.go index 08426cd2..4b8b2859 100644 --- a/internal/telemetry/parquet.go +++ b/internal/telemetry/parquet.go @@ -145,20 +145,32 @@ func (p *ParquetStore) StagingPath(id string) string { return filepath.Join(p.stagingDir, id) } +// ParquetMergeStats records row counts and the existing span metadata bounds +// without decoding the completed output a second time. +type ParquetMergeStats struct { + Rows int64 + MinSpanStartNanos int64 + MaxSpanStartNanos int64 +} + // MergeParquet rewrites immutable files into one native Parquet output. Span -// rows retain the index order required by trace.fidx; the other signals keep -// their input order. Re-encoding also consolidates tiny ingest row groups. -func (p *ParquetStore) MergeParquet(ctx context.Context, signal string, inputs []string, output string) error { +// rows retain the index order required by trace.fidx and build that sidecar as +// they are copied; the other signals keep their input order. Re-encoding also +// consolidates tiny ingest row groups. +func (p *ParquetStore) MergeParquet(ctx context.Context, signal string, inputs []string, output string) (ParquetMergeStats, error) { + var stats ParquetMergeStats + var err error switch signal { case "spans": - return mergeTypedParquet[spanParquetRow](ctx, inputs, output, true) + err = mergeTypedParquet[spanParquetRow](ctx, inputs, output, true, &stats) case "logs": - return mergeTypedParquet[logParquetRow](ctx, inputs, output, false) + err = mergeTypedParquet[logParquetRow](ctx, inputs, output, false, &stats) case "metrics": - return mergeTypedParquet[metricParquetRow](ctx, inputs, output, false) + err = mergeTypedParquet[metricParquetRow](ctx, inputs, output, false, &stats) default: - return fmt.Errorf("unsupported Parquet signal %q", signal) + err = fmt.Errorf("unsupported Parquet signal %q", signal) } + return stats, err } func (p *ParquetStore) BatchMetadata() []BatchMetadata { @@ -736,69 +748,10 @@ func (p *ParquetStore) Stats() (map[string]ParquetStats, error) { return stats, nil } -// PrepareReplacement completes metadata and the trace sidecar for Parquet -// files produced by the compactor. +// PrepareReplacement durably records metadata after all signal merges and the +// streaming trace index have completed. Span bounds come from the merge. func (p *ParquetStore) PrepareReplacement(dir string, metadata BatchMetadata) error { metadata.Version = batchMetadataVersion - metadata.MinSpanStartNanos = 0 - metadata.MaxSpanStartNanos = 0 - if metadata.Spans > 0 { - f, err := os.Open(filepath.Join(dir, "spans.parquet")) - if err != nil { - return err - } - reader := parquet.NewGenericReader[traceParquetRow](f) - index, err := newTraceIndexWriter(filepath.Join(dir, "trace.fidx")) - if err != nil { - _ = reader.Close() - _ = f.Close() - return err - } - rows := 0 - buffer := make([]traceParquetRow, min(8192, metadata.Spans)) - for { - n, readErr := reader.Read(buffer) - for i := range n { - if start := buffer[i].StartUnixNano; start > 0 && (metadata.MinSpanStartNanos == 0 || start < metadata.MinSpanStartNanos) { - metadata.MinSpanStartNanos = start - } - metadata.MaxSpanStartNanos = max(metadata.MaxSpanStartNanos, buffer[i].StartUnixNano) - if err := index.Append(buffer[i].TraceHash); err != nil { - index.Abort() - _ = reader.Close() - _ = f.Close() - return err - } - } - rows += n - if errors.Is(readErr, io.EOF) { - break - } - if readErr != nil { - index.Abort() - _ = reader.Close() - _ = f.Close() - return readErr - } - if n == 0 { - index.Abort() - _ = reader.Close() - _ = f.Close() - return io.ErrNoProgress - } - } - if err := errors.Join(reader.Close(), f.Close()); err != nil { - index.Abort() - return err - } - if rows != metadata.Spans { - index.Abort() - return fmt.Errorf("compacted span count: got %d want %d", rows, metadata.Spans) - } - if err := index.Close(); err != nil { - return err - } - } if err := writeJSONFile(filepath.Join(dir, "metadata.json"), metadata); err != nil { return err } @@ -1288,16 +1241,51 @@ func parquetWriterOptions(pageSize int, extra ...parquet.WriterOption) []parquet 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 } - return r.Rows.ReadRows(rows) + 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) (err error) { +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 } @@ -1352,14 +1340,41 @@ func mergeTypedParquet[T any](ctx context.Context, inputs []string, output strin } }() 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...)...) - _, copyErr := parquet.CopyRows(writer, contextParquetRows{Context: ctx, Rows: rows}) + 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 } @@ -1468,8 +1483,3 @@ func ValidateBatchID(id string) error { } return nil } - -type traceParquetRow struct { - TraceHash uint64 `parquet:"_trace_hash"` - StartUnixNano int64 `parquet:"start_unix_nano"` -} diff --git a/internal/telemetry/store/compaction.go b/internal/telemetry/store/compaction.go index bc048e8a..d7f31f38 100644 --- a/internal/telemetry/store/compaction.go +++ b/internal/telemetry/store/compaction.go @@ -213,9 +213,17 @@ func (r *Repository) CompactParquet(ctx context.Context, publisher ParquetPublis group, mergeCtx := errgroup.WithContext(ctx) for _, plan := range plans { group.Go(func() error { - if err := r.Parquet.MergeParquet(mergeCtx, plan.signal, plan.inputs, plan.output); err != nil { + stats, err := r.Parquet.MergeParquet(mergeCtx, plan.signal, plan.inputs, plan.output) + if err != nil { return fmt.Errorf("compact %s Parquet: %w", plan.signal, err) } + if plan.signal == "spans" { + if stats.Rows != int64(marker.Output.Spans) { + return fmt.Errorf("compacted span count: got %d want %d", stats.Rows, marker.Output.Spans) + } + marker.Output.MinSpanStartNanos = stats.MinSpanStartNanos + marker.Output.MaxSpanStartNanos = stats.MaxSpanStartNanos + } if err := mergeCtx.Err(); err != nil { return err } diff --git a/internal/telemetry/store/repository_test.go b/internal/telemetry/store/repository_test.go index b760fbbe..d26d54f5 100644 --- a/internal/telemetry/store/repository_test.go +++ b/internal/telemetry/store/repository_test.go @@ -626,9 +626,14 @@ func TestRepositoryRecoversInterruptedCompactionSwap(t *testing.T) { for i := range paths { paths[i] = filepath.Join(repository.Parquet.BatchPath(fmt.Sprintf("recover-%d", i)), signal+".parquet") } - if err := repository.Parquet.MergeParquet(context.Background(), signal, paths, filepath.Join(stage, signal+".parquet")); err != nil { + stats, err := repository.Parquet.MergeParquet(context.Background(), signal, paths, filepath.Join(stage, signal+".parquet")) + if err != nil { t.Fatal(err) } + if signal == "spans" { + output.MinSpanStartNanos = stats.MinSpanStartNanos + output.MaxSpanStartNanos = stats.MaxSpanStartNanos + } } if err := repository.Parquet.PrepareReplacement(stage, output); err != nil { t.Fatal(err) diff --git a/internal/telemetry/variant_test.go b/internal/telemetry/variant_test.go index 5e3c0c53..82629a14 100644 --- a/internal/telemetry/variant_test.go +++ b/internal/telemetry/variant_test.go @@ -41,7 +41,7 @@ func TestVariantRoundTripAcrossRowGroupsAndCompaction(t *testing.T) { t.Fatal(err) } merged := filepath.Join(dir, "merged.parquet") - if err := mergeTypedParquet[logParquetRow](context.Background(), []string{first, second}, merged, false); err != nil { + if err := mergeTypedParquet[logParquetRow](context.Background(), []string{first, second}, merged, false, &ParquetMergeStats{}); err != nil { t.Fatal(err) } checkVariantRows(t, merged, append(rows, rows[:3]...)) From 54fda8009395617fd73afab518715b7924c4be2a Mon Sep 17 00:00:00 2001 From: Vishal Rana Date: Thu, 15 Oct 2026 01:24:46 -0700 Subject: [PATCH 2/2] test(query): synchronize the maintenance schedule test's recorded cycles The compaction and maintenance callbacks append to slices that the test body reads; under the race detector that is a data race even inside a synctest bubble. Guard both with a mutex and read copies. Claude-Session: https://claude.ai/code/session_01XA52nWyQ3R5f5kK8kjpAQW --- internal/query/maintenance_schedule_test.go | 27 ++++++++++++++++----- 1 file changed, 21 insertions(+), 6 deletions(-) diff --git a/internal/query/maintenance_schedule_test.go b/internal/query/maintenance_schedule_test.go index 7a00b1e7..6d57ce11 100644 --- a/internal/query/maintenance_schedule_test.go +++ b/internal/query/maintenance_schedule_test.go @@ -3,6 +3,8 @@ package query import ( "context" "reflect" + "slices" + "sync" "testing" "testing/synctest" "time" @@ -13,12 +15,19 @@ func TestMaintenanceCompactionDrainsBacklogThenIdles(t *testing.T) { ctx, cancel := context.WithCancel(t.Context()) defer cancel() start := time.Now() + var mu sync.Mutex var cycles []time.Duration var maintenance []time.Duration done := make(chan struct{}) go func() { defer close(done) - runMaintenanceSchedule(ctx, time.Hour, func() { maintenance = append(maintenance, time.Since(start)) }, func() int { + runMaintenanceSchedule(ctx, time.Hour, func() { + mu.Lock() + defer mu.Unlock() + maintenance = append(maintenance, time.Since(start)) + }, func() int { + mu.Lock() + defer mu.Unlock() cycles = append(cycles, time.Since(start)) if len(cycles) <= 2 { return 1 @@ -28,15 +37,21 @@ func TestMaintenanceCompactionDrainsBacklogThenIdles(t *testing.T) { }() time.Sleep(21 * time.Second) synctest.Wait() + mu.Lock() + gotCycles := slices.Clone(cycles) + mu.Unlock() want := []time.Duration{10 * time.Second, 10100 * time.Millisecond, 10200 * time.Millisecond, 20200 * time.Millisecond} - if !reflect.DeepEqual(cycles, want) { - t.Fatalf("cycles=%v want=%v", cycles, want) + if !reflect.DeepEqual(gotCycles, want) { + t.Fatalf("cycles=%v want=%v", gotCycles, want) } - t.Logf("backlogged cycle gaps: %s, %s; idle gap: %s", cycles[1]-cycles[0], cycles[2]-cycles[1], cycles[3]-cycles[2]) + t.Logf("backlogged cycle gaps: %s, %s; idle gap: %s", gotCycles[1]-gotCycles[0], gotCycles[2]-gotCycles[1], gotCycles[3]-gotCycles[2]) time.Sleep(time.Hour - 21*time.Second) synctest.Wait() - if !reflect.DeepEqual(maintenance, []time.Duration{0, time.Hour}) { - t.Fatalf("maintenance=%v", maintenance) + mu.Lock() + gotMaintenance := slices.Clone(maintenance) + mu.Unlock() + if !reflect.DeepEqual(gotMaintenance, []time.Duration{0, time.Hour}) { + t.Fatalf("maintenance=%v", gotMaintenance) } cancel() synctest.Wait()