Add vparquet4 encoding

.golangci.yml

@@ -63,6 +63,8 @@ linters-settings:
     exclude: ./.errcheck-exclude.txt
   errorlint:
     errorf: false # don't check whether fmt.Errorf uses %w to wrap errors
+  goconst:
+    ignore-tests: true
 
 issues:
   exclude:

CHANGELOG.md

@@ -1,6 +1,7 @@
 ## main / unreleased
 
 * [ENHANCEMENT] Add tracing integration to profiling endpoints [#3276](https://github.com/grafana/tempo/pull/3276) (@cyriltovena)
+* [FEATURE] New block encoding vParquet4 with support for links, events, and arrays [#3368](https://github.com/grafana/tempo/pull/3368) (@stoewer @ie-pham @andreasgerstmayr)
 * [FEATURE] Add configuration on tempo-query plugin for fetch services older than complete_block_timeout [#3262](https://github.com/grafana/tempo/pull/3350) (@rubenvp8510)
 * [FEATURE] Add support for multi-tenant queries in streaming search [#3262](https://github.com/grafana/tempo/pull/3262) (@electron0zero)
 * [FEATURE] TraceQL metrics queries [#3227](https://github.com/grafana/tempo/pull/3227) [#3252](https://github.com/grafana/tempo/pull/3252) [#3258](https://github.com/grafana/tempo/pull/3258) (@mdisibio @zalegrala)

modules/frontend/search_progress.go

@@ -159,6 +159,17 @@ func (r *searchProgress) result() *shardedSearchResults {
 			SpanSet:           copySpanset(m.SpanSet),
 		}
 
+		// copy service stats
+		if len(m.ServiceStats) > 0 {
+			mCopy.ServiceStats = make(map[string]*tempopb.ServiceStats, len(m.ServiceStats))
+			for service, stat := range m.ServiceStats {
+				mCopy.ServiceStats[service] = &tempopb.ServiceStats{
+					SpanCount:  stat.SpanCount,
+					ErrorCount: stat.ErrorCount,
+				}
+			}
+		}
+
 		// now copy spansets
 		if len(m.SpanSets) > 0 {
 			mCopy.SpanSets = make([]*tempopb.SpanSet, 0, len(m.SpanSets))

pkg/tempopb/tempo.proto

@@ -113,6 +113,12 @@ message TraceSearchMetadata {
   uint32 durationMs = 5;
   SpanSet spanSet = 6; // deprecated. use SpanSets field below
   repeated SpanSet spanSets = 7;
+  map<string, ServiceStats> serviceStats = 8;
+}
+
+message ServiceStats {
+  uint32 spanCount = 1;
+  uint32 errorCount = 2;
 }
 
 message SpanSet {

pkg/traceql/combine.go

@@ -70,6 +70,17 @@ func combineSearchResults(existing *tempopb.TraceSearchMetadata, incoming *tempo
 		existing.DurationMs = incoming.DurationMs
 	}
 
+	// Merge service stats
+	for service, incomingStats := range incoming.ServiceStats {
+		existingStats, ok := existing.ServiceStats[service]
+		if !ok {
+			existingStats = &tempopb.ServiceStats{}
+			existing.ServiceStats[service] = existingStats
+		}
+		existingStats.SpanCount += incomingStats.SpanCount
+		existingStats.ErrorCount += incomingStats.ErrorCount
+	}
+
 	// make a map of existing Spansets
 	existingSS := make(map[string]*tempopb.SpanSet)
 	for _, ss := range existing.SpanSets {

pkg/traceql/engine.go

@@ -231,9 +231,17 @@ func (e *Engine) asTraceSearchMetadata(spanset *Spanset) *tempopb.TraceSearchMet
 		RootTraceName:     spanset.RootSpanName,
 		StartTimeUnixNano: spanset.StartTimeUnixNanos,
 		DurationMs:        uint32(spanset.DurationNanos / 1_000_000),
+		ServiceStats:      make(map[string]*tempopb.ServiceStats, len(spanset.ServiceStats)),
 		SpanSet:           &tempopb.SpanSet{},
 	}
 
+	for service, stats := range spanset.ServiceStats {
+		metadata.ServiceStats[service] = &tempopb.ServiceStats{
+			SpanCount:  stats.SpanCount,
+			ErrorCount: stats.ErrorCount,
+		}
+	}
+
 	for _, span := range spanset.Spans {
 		tempopbSpan := &tempopb.Span{
 			SpanID:            util.SpanIDToHexString(span.ID()),

pkg/traceql/engine_test.go

@@ -167,6 +167,7 @@ func TestEngine_Execute(t *testing.T) {
 			TraceID:         "1",
 			RootServiceName: "my-service",
 			RootTraceName:   "HTTP GET",
+			ServiceStats:    map[string]*tempopb.ServiceStats{},
 			SpanSet:         expectedSpanset,
 			SpanSets:        []*tempopb.SpanSet{expectedSpanset},
 		},
@@ -321,6 +322,7 @@ func TestEngine_asTraceSearchMetadata(t *testing.T) {
 		RootTraceName:     "HTTP GET",
 		StartTimeUnixNano: 1000,
 		DurationMs:        uint32(time.Second.Milliseconds()),
+		ServiceStats:      map[string]*tempopb.ServiceStats{},
 		SpanSet:           expectedSpanset,
 		SpanSets:          []*tempopb.SpanSet{expectedSpanset},
 	}

pkg/traceql/storage.go

@@ -131,6 +131,11 @@ type SpansetAttribute struct {
 	Val  Static
 }
 
+type ServiceStats struct {
+	SpanCount  uint32
+	ErrorCount uint32
+}
+
 type Spanset struct {
 	// these fields are actually used by the engine to evaluate queries
 	Scalar Static
@@ -141,6 +146,7 @@ type Spanset struct {
 	RootServiceName    string
 	StartTimeUnixNanos uint64
 	DurationNanos      uint64
+	ServiceStats       map[string]ServiceStats
 	Attributes         []*SpansetAttribute
 
 	// Set this function to provide upstream callers with a method to

tempodb/encoding/versioned.go

@@ -14,6 +14,7 @@ import (
 	"github.com/grafana/tempo/tempodb/encoding/vparquet"
 	"github.com/grafana/tempo/tempodb/encoding/vparquet2"
 	"github.com/grafana/tempo/tempodb/encoding/vparquet3"
+	"github.com/grafana/tempo/tempodb/encoding/vparquet4"
 )
 
 // VersionedEncoding represents a backend block version, and the methods to
@@ -66,6 +67,8 @@ func FromVersion(v string) (VersionedEncoding, error) {
 		return vparquet2.Encoding{}, nil
 	case vparquet3.VersionString:
 		return vparquet3.Encoding{}, nil
+	case vparquet4.VersionString:
+		return vparquet4.Encoding{}, nil
 	default:
 		return nil, fmt.Errorf("%s is not a valid block version", v)
 	}
@@ -78,7 +81,7 @@ func DefaultEncoding() VersionedEncoding {
 
 // LatestEncoding returns the most recent encoding.
 func LatestEncoding() VersionedEncoding {
-	return vparquet3.Encoding{}
+	return vparquet4.Encoding{}
 }
 
 // AllEncodings returns all encodings
@@ -88,6 +91,7 @@ func AllEncodings() []VersionedEncoding {
 		vparquet.Encoding{},
 		vparquet2.Encoding{},
 		vparquet3.Encoding{},
+		vparquet4.Encoding{},
 	}
 }
 

tempodb/encoding/vparquet3/block_traceql_test.go

@@ -564,22 +564,26 @@ func BenchmarkBackendBlockTraceQL(b *testing.B) {
 		query string
 	}{
 		// span
-		{"spanAttValNoMatch", "{ span.bloom = `bar` }"},
-		{"spanAttIntrinsicNoMatch", "{ name = `asdfasdf` }"},
+		{"spanAttValMatch", "{ span.component = `net/http` }"},
+		{"spanAttValNoMatch", "{ span.bloom = `does-not-exit-6c2408325a45` }"},
+		{"spanAttIntrinsicMatch", "{ name = `/cortex.Ingester/Push` }"},
+		{"spanAttIntrinsicNoMatch", "{ name = `does-not-exit-6c2408325a45` }"},
 
 		// resource
-		{"resourceAttValNoMatch", "{ resource.module.path = `bar` }"},
-		{"resourceAttIntrinsicMatch", "{ resource.service.name = `tempo-query-frontend` }"},
+		{"resourceAttValMatch", "{ resource.opencensus.exporterversion = `Jaeger-Go-2.30.0` }"},
+		{"resourceAttValNoMatch", "{ resource.module.path = `does-not-exit-6c2408325a45` }"},
+		{"resourceAttIntrinsicMatch", "{ resource.service.name = `tempo-gateway` }"},
+		{"resourceAttIntrinsicMatch", "{ resource.service.name = `does-not-exit-6c2408325a45` }"},
 
 		// mixed
-		{"mixedValNoMatch", "{ .bloom = `bar` }"},
+		{"mixedValNoMatch", "{ .bloom = `does-not-exit-6c2408325a45` }"},
 		{"mixedValMixedMatchAnd", "{ resource.foo = `bar` && name = `gcs.ReadRange` }"},
 		{"mixedValMixedMatchOr", "{ resource.foo = `bar` || name = `gcs.ReadRange` }"},
 
 		{"count", "{ } | count() > 1"},
-		{"struct", "{ resource.service.name != `loki-querier` } >> { resource.service.name = `loki-querier` && status = error }"},
-		{"||", "{ resource.service.name = `loki-querier` } || { resource.service.name = `loki-ingester` }"},
-		{"mixed", `{resource.namespace!="" && resource.service.name="loki-distributor" && duration>2s && resource.cluster=~"prod.*"}`},
+		{"struct", "{ resource.service.name != `loki-querier` } >> { resource.service.name = `loki-gateway` && status = error }"},
+		{"||", "{ resource.service.name = `loki-querier` } || { resource.service.name = `loki-gateway` }"},
+		{"mixed", `{resource.namespace!="" && resource.service.name="cortex-gateway" && duration>50ms && resource.cluster=~"prod.*"}`},
 	}
 
 	ctx := context.TODO()
@@ -598,8 +602,8 @@ func BenchmarkBackendBlockTraceQL(b *testing.B) {
 	require.NoError(b, err)
 
 	opts := common.DefaultSearchOptions()
-	opts.StartPage = 10
-	opts.TotalPages = 1
+	opts.StartPage = 3
+	opts.TotalPages = 2
 
 	block := newBackendBlock(meta, rr)
 	_, _, err = block.openForSearch(ctx, opts)

tempodb/encoding/vparquet4/block.go

@@ -0,0 +1,32 @@
+package vparquet4
+
+import (
+	"sync"
+
+	"github.com/grafana/tempo/tempodb/backend"
+	"github.com/grafana/tempo/tempodb/encoding/common"
+)
+
+const (
+	DataFileName = "data.parquet"
+)
+
+type backendBlock struct {
+	meta *backend.BlockMeta
+	r    backend.Reader
+
+	openMtx sync.Mutex
+}
+
+var _ common.BackendBlock = (*backendBlock)(nil)
+
+func newBackendBlock(meta *backend.BlockMeta, r backend.Reader) *backendBlock {
+	return &backendBlock{
+		meta: meta,
+		r:    r,
+	}
+}
+
+func (b *backendBlock) BlockMeta() *backend.BlockMeta {
+	return b.meta
+}