vParquet5: virtual span row numbers

CHANGELOG.md

@@ -9,6 +9,7 @@
 * [FEATURE] Added validation mode and tests for tempo-vulture [#5605](https://github.com/grafana/tempo/pull/5605)
 * [FEATURE] Add query_end_cutoff setting to clamp near-“now” queries (default 30s) for consistent Tempo results [#5682](https://github.com/grafana/tempo/pull/5682) (@javiermolinar)
 * [FEATURE] New block encoding vParquet5-preview4 with array support for dedicated columns. This is a preview, breaking changes are expected. [#5760](https://github.com/grafana/tempo/pull/5760) (@stoewer)
+* [FEATURE] New block encoding vParquet5-preview5 with virtual span row numbers. This is a preview, breaking changes are expected. [#5943](https://github.com/grafana/tempo/pull/5943) (@stoewer)
 * [FEATURE] Add SSE-C encryption support to S3 backend [#5789](https://github.com/grafana/tempo/pull/5789) (@steffsas)
 * [ENHANCEMENT] docs: Add explicit notes about authentication [#5735](https://github.com/grafana/tempo/pull/5735) (@electron0zero)
 * [ENHANCEMENT] On startup, first record for live store to consume is not older than two complete block timeouts [#5693](https://github.com/grafana/tempo/pull/5693) (@ruslan-mikhailov)

tempodb/encoding/vparquet5/block_autocomplete_test.go

@@ -132,6 +132,7 @@ func TestFetchTagNames(t *testing.T) {
 								attr("scope-attr-str-1", "scope-attr-1"),
 							},
 						},
+						SpanCount: 2,
 						Spans: []Span{
 							{
 								SpanID:        []byte("0101"),
@@ -197,6 +198,7 @@ func TestFetchTagNames(t *testing.T) {
 								attr("scope-attr-str-2", "scope-attr-2"),
 							},
 						},
+						SpanCount: 1,
 						Spans: []Span{
 							{
 								SpanID:        []byte("0201"),

tempodb/encoding/vparquet5/block_findtracebyid_test.go

@@ -52,6 +52,7 @@ func TestBackendBlockFindTraceByID(t *testing.T) {
 					},
 					ScopeSpans: []ScopeSpans{
 						{
+							SpanCount: 1,
 							Spans: []Span{
 								{
 									Name: "hello",

tempodb/encoding/vparquet5/block_iterator.go

@@ -34,7 +34,7 @@ func (b *backendBlock) openForIteration(ctx context.Context) (*parquet.File, *pa
 		return nil, nil, err
 	}
 
-	r := parquet.NewReader(pf, parquet.SchemaOf(&Trace{}))
+	r := parquet.NewReader(pf, pf.Schema())
 	return pf, r, nil
 }
 

tempodb/encoding/vparquet5/block_search.go

@@ -397,38 +397,6 @@ func makeNilIterFunc(ctx context.Context, rgs []parquet.RowGroup, pf *parquet.Fi
 	}
 }
 
-type rowNumberIterator struct {
-	rowNumbers []pq.RowNumber
-}
-
-var _ pq.Iterator = (*rowNumberIterator)(nil)
-
-func (r *rowNumberIterator) String() string {
-	return "rowNumberIterator()"
-}
-
-func (r *rowNumberIterator) Next() (*pq.IteratorResult, error) {
-	if len(r.rowNumbers) == 0 {
-		return nil, nil
-	}
-
-	res := &pq.IteratorResult{RowNumber: r.rowNumbers[0]}
-	r.rowNumbers = r.rowNumbers[1:]
-	return res, nil
-}
-
-func (r *rowNumberIterator) SeekTo(to pq.RowNumber, definitionLevel int) (*pq.IteratorResult, error) {
-	var at *pq.IteratorResult
-
-	for at, _ = r.Next(); r != nil && at != nil && pq.CompareRowNumbers(definitionLevel, at.RowNumber, to) < 0; {
-		at, _ = r.Next()
-	}
-
-	return at, nil
-}
-
-func (r *rowNumberIterator) Close() {}
-
 // reportValuesPredicate is a "fake" predicate that uses existing iterator logic to find all values in a given column
 type reportValuesPredicate struct {
 	cb common.TagValuesCallbackV2

tempodb/encoding/vparquet5/block_search_test.go

@@ -55,6 +55,7 @@ func TestBackendBlockSearch(t *testing.T) {
 				},
 				ScopeSpans: []ScopeSpans{
 					{
+						SpanCount: 1,
 						Spans: []Span{
 							{
 								Name:         "hello",

tempodb/encoding/vparquet5/block_traceql.go

@@ -914,6 +914,7 @@ const (
 	columnPathResourceAttrBool        = "rs.list.element.Resource.Attrs.list.element.ValueBool.list.element"
 	columnPathResourceServiceName     = "rs.list.element.Resource.ServiceName"
 
+	columnPathScopeSpansSpanCount       = "rs.list.element.ss.list.element.SpanCount"
 	columnPathInstrumentationName       = "rs.list.element.ss.list.element.Scope.Name"
 	columnPathInstrumentationVersion    = "rs.list.element.ss.list.element.Scope.Version"
 	columnPathInstrumentationAttrKey    = "rs.list.element.ss.list.element.Scope.Attrs.list.element.Key"
@@ -2213,20 +2214,14 @@ func createSpanIterator(makeIter, makeNilIter makeIterFn, innerIterators []parqu
 		iters = nil
 	}
 
-	// if there are no direct conditions imposed on the span/span attributes level we are purposefully going to request the "Kind" column
-	//  b/c it is extremely cheap to retrieve. retrieving matching spans in this case will allow aggregates such as "count" to be computed
-	//  how do we know to pull duration for things like | avg(duration) > 1s? look at avg(span.http.status_code) it pushes a column request down here
-	//  the entire engine is built around spans. we have to return at least one entry for every span to the layers above for things to work
-	// TODO: note that if the query is { kind = client } the fetch layer will actually create two iterators over the kind column. this is evidence
-	//  this spaniterator code could be tightened up
-	// Also note that this breaks optimizations related to requireAtLeastOneMatch and requireAtLeastOneMatchOverall b/c it will add a kind attribute
-	//  to the span attributes map in spanCollector
+	// If there are no direct conditions imposed on the span level we are evaluating the SpanCount column to
+	// calculate all span row numbers.
 	if len(required) == 0 {
 		var pred parquetquery.Predicate
 		if sampler != nil {
 			pred = newSamplingPredicate(sampler, nil)
 		}
-		required = []parquetquery.Iterator{makeIter(columnPathSpanStatusCode, pred, "")}
+		required = []parquetquery.Iterator{newVirtualRowNumberIterator(makeIter(columnPathScopeSpansSpanCount, pred, "spanCount"), DefinitionLevelResourceSpansILSSpan)}
 	}
 
 	// Left join here means the span id/start/end iterators + 1 are required,

tempodb/encoding/vparquet5/block_traceql_test.go

@@ -369,6 +369,7 @@ func TestBackendNilValueBlockSearchTraceQL(t *testing.T) {
 					{
 						ScopeSpans: []ScopeSpans{
 							{
+								SpanCount: 2,
 								Spans: []Span{
 									{
 										// this span has nil values for everything
@@ -418,6 +419,7 @@ func TestBackendNilValueBlockSearchTraceQL(t *testing.T) {
 						},
 						ScopeSpans: []ScopeSpans{
 							{
+								SpanCount: 1,
 								Spans: []Span{
 									{
 										SpanID: []byte("nil-test-span-2"),
@@ -1033,6 +1035,7 @@ func fullyPopulatedTestTraceWithOption(id common.ID, parentIDTest bool) *Trace {
 								attr("scope-attr-bool", true),
 							},
 						},
+						SpanCount: 1,
 						Spans: []Span{
 							{
 								SpanID:                 []byte("spanid"),
@@ -1137,6 +1140,7 @@ func fullyPopulatedTestTraceWithOption(id common.ID, parentIDTest bool) *Trace {
 								attr("scope-attr-str", "scope-attr-2"),
 							},
 						},
+						SpanCount: 1,
 						Spans: []Span{
 							{
 								SpanID:                 []byte("spanid2"),

tempodb/encoding/vparquet5/combiner.go

@@ -111,6 +111,7 @@ func (c *Combiner) ConsumeWithFinal(tr *Trace, final bool) (spanCount int) {
 
 			if len(notFoundSpans) > 0 {
 				ils.Spans = notFoundSpans
+				ils.SpanCount = int32(len(ils.Spans))
 				notFoundILS = append(notFoundILS, ils)
 			}
 

tempodb/encoding/vparquet5/combiner_test.go

@@ -153,6 +153,7 @@ func TestCombiner(t *testing.T) {
 						},
 						ScopeSpans: []ScopeSpans{
 							{
+								SpanCount: 1,
 								Spans: []Span{
 									{
 										SpanID:         []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01},
@@ -176,6 +177,7 @@ func TestCombiner(t *testing.T) {
 						},
 						ScopeSpans: []ScopeSpans{
 							{
+								SpanCount: 1,
 								Spans: []Span{
 									{
 										SpanID:       []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02},
@@ -207,6 +209,7 @@ func TestCombiner(t *testing.T) {
 						},
 						ScopeSpans: []ScopeSpans{
 							{
+								SpanCount: 1,
 								Spans: []Span{
 									{
 										SpanID:         []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01},
@@ -225,6 +228,7 @@ func TestCombiner(t *testing.T) {
 						},
 						ScopeSpans: []ScopeSpans{
 							{
+								SpanCount: 1,
 								Spans: []Span{
 									{
 										SpanID:         []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02},
@@ -253,6 +257,7 @@ func TestCombiner(t *testing.T) {
 						},
 						ScopeSpans: []ScopeSpans{
 							{
+								SpanCount: 1,
 								Spans: []Span{
 									{
 										SpanID:         []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01},
@@ -276,6 +281,7 @@ func TestCombiner(t *testing.T) {
 						},
 						ScopeSpans: []ScopeSpans{
 							{
+								SpanCount: 2,
 								Spans: []Span{
 									{
 										SpanID:       []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02},
@@ -313,6 +319,7 @@ func TestCombiner(t *testing.T) {
 						},
 						ScopeSpans: []ScopeSpans{
 							{
+								SpanCount: 1,
 								Spans: []Span{
 									{
 										SpanID:         []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01},
@@ -331,6 +338,7 @@ func TestCombiner(t *testing.T) {
 						},
 						ScopeSpans: []ScopeSpans{
 							{
+								SpanCount: 2,
 								Spans: []Span{
 									{
 										SpanID:         []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02},
@@ -412,6 +420,7 @@ func TestCombinerReturnsDuplicates(t *testing.T) {
 						},
 						ScopeSpans: []ScopeSpans{
 							{
+								SpanCount: 1,
 								Spans: []Span{
 									{
 										SpanID:         []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01},
@@ -435,6 +444,7 @@ func TestCombinerReturnsDuplicates(t *testing.T) {
 						},
 						ScopeSpans: []ScopeSpans{
 							{
+								SpanCount: 1,
 								Spans: []Span{
 									{
 										SpanID:       []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02},
@@ -461,6 +471,7 @@ func TestCombinerReturnsDuplicates(t *testing.T) {
 						},
 						ScopeSpans: []ScopeSpans{
 							{
+								SpanCount: 1,
 								Spans: []Span{
 									{
 										SpanID:         []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01},
@@ -484,6 +495,7 @@ func TestCombinerReturnsDuplicates(t *testing.T) {
 						},
 						ScopeSpans: []ScopeSpans{
 							{
+								SpanCount: 2,
 								Spans: []Span{
 									{
 										SpanID:     []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01},

tempodb/encoding/vparquet5/encoding.go

@@ -9,7 +9,7 @@ import (
 	"github.com/grafana/tempo/tempodb/encoding/common"
 )
 
-const VersionString = "vParquet5-preview4"
+const VersionString = "vParquet5-preview5"
 
 type Encoding struct{}
 

tempodb/encoding/vparquet5/rebatch_trace.go

@@ -77,6 +77,7 @@ func rebatchTrace(trace *Trace) {
 			// else, merge the Spans with the existing identical ScopeSpans
 			rebatchSS := &rs.ScopeSpans[idx]
 			rebatchSS.Spans = append(rebatchSS.Spans, ss.Spans...)
+			rebatchSS.SpanCount = int32(len(rebatchSS.Spans))
 
 			// the append above creates copies of Spans, we have to clear the originals otherwise
 			// we will have multiple copies of the same slices in different Spans
@@ -233,6 +234,7 @@ func clearScopeSpans(sss []ScopeSpans) []ScopeSpans {
 		ss := &sss[i]
 		ss.Scope.Attrs = nil
 		ss.Spans = nil
+		ss.SpanCount = 0
 	}
 	return sss[:0]
 }