Merge pull request #5489 from parca-dev/query-flamechart-sorting

pkg/query: Sort samples before generating flame charts

go.mod

@@ -35,7 +35,7 @@ require (
 	github.com/olekukonko/tablewriter v0.0.5
 	github.com/parquet-go/parquet-go v0.24.0
 	github.com/planetscale/vtprotobuf v0.6.1-0.20241121165744-79df5c4772f2
-	github.com/polarsignals/frostdb v0.0.0-20250107161604-e9ab6f315eae
+	github.com/polarsignals/frostdb v0.0.0-20250218183442-9c9972458112
 	github.com/polarsignals/iceberg-go v0.0.0-20240502213135-2ee70b71e76b
 	github.com/prometheus/client_golang v1.20.5
 	github.com/prometheus/common v0.62.0
@@ -55,7 +55,7 @@ require (
 	go.opentelemetry.io/proto/otlp v1.3.1
 	go.uber.org/atomic v1.11.0
 	go.uber.org/automaxprocs v1.6.0
-	golang.org/x/exp v0.0.0-20241215155358-4a5509556b9e
+	golang.org/x/exp v0.0.0-20250128182459-e0ece0dbea4c
 	golang.org/x/net v0.35.0
 	golang.org/x/oauth2 v0.26.0
 	golang.org/x/sync v0.11.0
@@ -238,7 +238,7 @@ require (
 	golang.org/x/term v0.29.0 // indirect
 	golang.org/x/text v0.22.0 // indirect
 	golang.org/x/time v0.10.0 // indirect
-	golang.org/x/tools v0.28.0 // indirect
+	golang.org/x/tools v0.29.0 // indirect
 	golang.org/x/xerrors v0.0.0-20231012003039-104605ab7028 // indirect
 	google.golang.org/genproto v0.0.0-20241118233622-e639e219e697 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20250207221924-e9438ea467c6 // indirect

go.sum

@@ -728,8 +728,8 @@ github.com/planetscale/vtprotobuf v0.6.1-0.20241121165744-79df5c4772f2/go.mod h1
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
 github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/polarsignals/frostdb v0.0.0-20250107161604-e9ab6f315eae h1:Pg4yVBVw4O5O4TR+XCEU9pE0oKra4MOkq7pDaJUq0NM=
-github.com/polarsignals/frostdb v0.0.0-20250107161604-e9ab6f315eae/go.mod h1:Pdv/t9Qrx2c2dTbQTwsOmrgCvsz9fpcktVoq/6Lnjm0=
+github.com/polarsignals/frostdb v0.0.0-20250218183442-9c9972458112 h1:nL8vkrs13TXKhLgOrYol3ziC4eH0UGFh1IjgrdwrsY0=
+github.com/polarsignals/frostdb v0.0.0-20250218183442-9c9972458112/go.mod h1:/uYMJOhlj30NEsZUSxZMi3XvoYEtmMIPqM81bPVPoDk=
 github.com/polarsignals/iceberg-go v0.0.0-20240502213135-2ee70b71e76b h1:Dbm5itapR0uYIMujR8OntWpDJ/nm5OM6JiaKauLcZ4Y=
 github.com/polarsignals/iceberg-go v0.0.0-20240502213135-2ee70b71e76b/go.mod h1:5T9ChEZjRNhAGGLwH1cqzDA7wXB84SmU+WkXQr/ZAjo=
 github.com/polarsignals/wal v0.0.0-20240619104840-9da940027f9c h1:ReFgEXqZ9/y+/9ZdNHOa1L62wqt8mWqoqrWutWj2x+A=
@@ -939,8 +939,8 @@ golang.org/x/crypto v0.33.0/go.mod h1:bVdXmD7IV/4GdElGPozy6U7lWdRXA4qyRVGJV57uQ5
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190306152737-a1d7652674e8/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20200331195152-e8c3332aa8e5/go.mod h1:4M0jN8W1tt0AVLNr8HDosyJCDCDuyL9N9+3m7wDWgKw=
-golang.org/x/exp v0.0.0-20241215155358-4a5509556b9e h1:4qufH0hlUYs6AO6XmZC3GqfDPGSXHVXUFR6OND+iJX4=
-golang.org/x/exp v0.0.0-20241215155358-4a5509556b9e/go.mod h1:qj5a5QZpwLU2NLQudwIN5koi3beDhSAlJwa67PuM98c=
+golang.org/x/exp v0.0.0-20250128182459-e0ece0dbea4c h1:KL/ZBHXgKGVmuZBZ01Lt57yE5ws8ZPSkkihmEyq7FXc=
+golang.org/x/exp v0.0.0-20250128182459-e0ece0dbea4c/go.mod h1:tujkw807nyEEAamNbDrEGzRav+ilXA7PCRAd6xsmwiU=
 golang.org/x/image v0.0.0-20190227222117-0694c2d4d067/go.mod h1:kZ7UVZpmo3dzQBMxlp+ypCbDeSB+sBbTgSJuh5dn5js=
 golang.org/x/image v0.0.0-20190802002840-cff245a6509b/go.mod h1:FeLwcggjj3mMvU+oOTbSwawSJRM1uh48EjtB4UJZlP0=
 golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
@@ -1106,8 +1106,8 @@ golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4f
 golang.org/x/tools v0.1.0/go.mod h1:xkSsbof2nBLbhDlRMhhhyNLN/zl3eTqcnHD5viDpcZ0=
 golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
 golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU=
-golang.org/x/tools v0.28.0 h1:WuB6qZ4RPCQo5aP3WdKZS7i595EdWqWR8vqJTlwTVK8=
-golang.org/x/tools v0.28.0/go.mod h1:dcIOrVd3mfQKTgrDVQHqCPMWy6lnhfhtX3hLXYVLfRw=
+golang.org/x/tools v0.29.0 h1:Xx0h3TtM9rzQpQuR4dKLrdglAmCEN5Oi+P74JdhdzXE=
+golang.org/x/tools v0.29.0/go.mod h1:KMQVMRsVxU6nHCFXrBPhDB8XncLNLM0lIy/F14RP588=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=

pkg/query/columnquery.go

@@ -34,6 +34,8 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
+	"github.com/polarsignals/frostdb/pqarrow/arrowutils"
+
 	metastorev1alpha1 "github.com/parca-dev/parca/gen/proto/go/parca/metastore/v1alpha1"
 	pb "github.com/parca-dev/parca/gen/proto/go/parca/query/v1alpha1"
 	sharepb "github.com/parca-dev/parca/gen/proto/go/parca/share/v1alpha1"
@@ -600,6 +602,55 @@ func RenderReport(
 			},
 		}, nil
 	case pb.QueryRequest_REPORT_TYPE_FLAMEGRAPH_ARROW, pb.QueryRequest_REPORT_TYPE_FLAMECHART:
+		if typ == pb.QueryRequest_REPORT_TYPE_FLAMECHART {
+			// Generating the flame chart assumes a single record that is sorted by timestamp.
+			for i, sample := range p.Samples {
+				indices := sample.Schema().FieldIndices(FlamegraphFieldTimestamp)
+				if len(indices) != 1 {
+					return nil, status.Errorf(codes.Internal, "invalid flame chart timestamp indices: %v", indices)
+				}
+				sortedIndices, err := arrowutils.SortRecord(sample, []arrowutils.SortingColumn{
+					{Index: indices[0], Direction: arrowutils.Ascending},
+				})
+				if err != nil {
+					return nil, status.Errorf(codes.Internal, "failed to sort flame chart record: %v", err.Error())
+				}
+
+				isSorted := true
+				for j := 0; j < sortedIndices.Len(); j++ {
+					if sortedIndices.Value(j) != int32(j) {
+						isSorted = false
+						break
+					}
+				}
+				if isSorted {
+					// Don't sort if the indices are already sorted.
+					continue
+				}
+
+				sorted, err := arrowutils.Take(ctx, sample, sortedIndices)
+				if err != nil {
+					return nil, status.Errorf(codes.Internal, "failed to sort flame chart record: %v", err.Error())
+				}
+
+				p.Samples[i] = sorted
+			}
+
+			if len(p.Samples) > 1 {
+				indices := p.Samples[0].Schema().FieldIndices(FlamegraphFieldTimestamp)
+				if len(indices) != 1 {
+					return nil, status.Errorf(codes.Internal, "invalid flame chart timestamp indices: %v", indices)
+				}
+				sorted, err := arrowutils.MergeRecords(mem, p.Samples, []arrowutils.SortingColumn{
+					{Index: indices[0], Direction: arrowutils.Ascending},
+				}, 0)
+				if err != nil {
+					return nil, status.Errorf(codes.Internal, "failed to merge flame chart records: %v", err.Error())
+				}
+				p.Samples = []arrow.Record{sorted}
+			}
+		}
+
 		fa, total, err := GenerateFlamegraphArrow(ctx, mem, tracer, p, groupBy, nodeTrimFraction)
 		if err != nil {
 			return nil, status.Errorf(codes.Internal, "failed to generate arrow flamegraph: %v", err.Error())

pkg/query/flamegraph_arrow.go

@@ -756,6 +756,9 @@ func (fb *flamegraphBuilder) mergeUnsymbolizedRows(
 }
 
 func matchRowsByTimestamp(compareTimestamp, compareDuration, timestamp, duration int64) (bool, error) {
+	if compareTimestamp > timestamp {
+		return false, fmt.Errorf("compareTimestamp > timestamp: %d > %d", compareTimestamp, timestamp)
+	}
 	if compareTimestamp == timestamp {
 		return false, fmt.Errorf("multiple samples for the same timestamp is not allowed: %d", timestamp)
 	}
@@ -764,8 +767,9 @@ func matchRowsByTimestamp(compareTimestamp, compareDuration, timestamp, duration
 	// We truncate 10% jitter. We use duration which usually is the period.
 	// For example, for 19hz sampling rate, we'll get a duration of 1000ms/19hz = 52.63ms
 	// and 10% are 5.2ms jitter that gets truncated.
-	truncated := difference.Truncate(time.Duration(duration / 10))
-	return truncated == 0, nil
+	jitter := time.Duration(duration / 10)
+	truncated := difference - jitter
+	return truncated <= 0, nil
 }
 
 func (fb *flamegraphBuilder) intersectLabels(

pkg/query/flamegraph_arrow_test.go

@@ -1459,3 +1459,24 @@ func drawFlamegraphToConsole(testing *testing.T, record arrow.Record) {
 	populateChild(t, 0)
 	fmt.Println(t)
 }
+
+func Test_matchRowsByTimestamp(t *testing.T) {
+	second := time.Second.Nanoseconds()
+
+	tests := []struct {
+		name         string
+		ct, cd, t, d int64
+		match        bool
+	}{
+		{"0", 0, second, second, second, true},
+		{"1/100", 0, second, second + second/100, second, true},
+		{"1/10", 0, second, second + second/10, second, true},
+		{"1/5", 0, second, second + second/5, second, false},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			m, _ := matchRowsByTimestamp(tc.ct, tc.cd, tc.t, tc.d)
+			require.Equal(t, tc.match, m)
+		})
+	}
+}