Precalculate and reuse the vParquet3 schema before opening blocks

CHANGELOG.md

@@ -46,6 +46,7 @@
 * [ENHANCEMENT] Add variable expansion support to overrides configuration [#3175](https://github.com/grafana/tempo/pull/3175) (@mapno)
 * [ENHANCEMENT] Update memcached default image in jsonnet for multiple CVE [#3310](https://github.com/grafana/tempo/pull/3310) (@zalegrala)
 * [ENHANCEMENT] Add HTML pages /status/overrides and /status/overrides/{tenant} [#3244](https://github.com/grafana/tempo/pull/3244) [#3332](https://github.com/grafana/tempo/pull/3332) (@kvrhdn)
+* [ENHANCEMENT] Precalculate and reuse the vParquet3 schema before opening blocks [#3367](https://github.com/grafana/tempo/pull/3367) (@stoewer)
 * [BUGFIX] Prevent building parquet iterators that would loop forever. [#3159](https://github.com/grafana/tempo/pull/3159) (@mapno)
 * [BUGFIX] Sanitize name in mapped dimensions in span-metrics processor [#3171](https://github.com/grafana/tempo/pull/3171) (@mapno)
 * [BUGFIX] Fixed an issue where cached footers were requested then ignored. [#3196](https://github.com/grafana/tempo/pull/3196) (@joe-elliott)

tempodb/encoding/vparquet3/block_iterator.go

@@ -19,7 +19,13 @@ func (b *backendBlock) open(ctx context.Context) (*parquet.File, *parquet.Reader
 	// 128 MB memory buffering
 	br := tempo_io.NewBufferedReaderAt(rr, int64(b.meta.Size), 2*1024*1024, 64)
 
-	pf, err := parquet.OpenFile(br, int64(b.meta.Size), parquet.SkipBloomFilters(true), parquet.SkipPageIndex(true))
+	o := []parquet.FileOption{
+		parquet.SkipBloomFilters(true),
+		parquet.SkipPageIndex(true),
+		parquet.FileSchema(parquetSchema),
+	}
+
+	pf, err := parquet.OpenFile(br, int64(b.meta.Size), o...)
 	if err != nil {
 		return nil, nil, err
 	}

tempodb/encoding/vparquet3/block_search.go

@@ -68,6 +68,7 @@ func (b *backendBlock) openForSearch(ctx context.Context, opts common.SearchOpti
 		parquet.SkipBloomFilters(true),
 		parquet.SkipPageIndex(true),
 		parquet.FileReadMode(parquet.ReadModeAsync),
+		parquet.FileSchema(parquetSchema),
 	}
 
 	// if the read buffer size provided is <= 0 then we'll use the parquet default

tempodb/encoding/vparquet3/schema.go

@@ -3,14 +3,15 @@ package vparquet3
 import (
 	"bytes"
 
-	"github.com/grafana/tempo/tempodb/backend"
-
 	"github.com/golang/protobuf/jsonpb" //nolint:all //deprecated
+	"github.com/parquet-go/parquet-go"
+
 	"github.com/grafana/tempo/pkg/tempopb"
 	v1 "github.com/grafana/tempo/pkg/tempopb/common/v1"
 	v1_resource "github.com/grafana/tempo/pkg/tempopb/resource/v1"
 	v1_trace "github.com/grafana/tempo/pkg/tempopb/trace/v1"
 	"github.com/grafana/tempo/pkg/util"
+	"github.com/grafana/tempo/tempodb/backend"
 	"github.com/grafana/tempo/tempodb/encoding/common"
 )
 
@@ -103,6 +104,8 @@ var (
 		LabelHTTPUrl:        "rs.list.element.ss.list.element.Spans.list.element.HttpUrl",
 		LabelHTTPStatusCode: "rs.list.element.ss.list.element.Spans.list.element.HttpStatusCode",
 	}
+
+	parquetSchema = parquet.SchemaOf(&Trace{})
 )
 
 type Attribute struct {

tempodb/encoding/vparquet3/wal_block.go

@@ -35,9 +35,6 @@ const defaultRowPoolSize = 100000
 // completeBlockRowPool is used by the wal iterators and complete block logic to pool rows
 var completeBlockRowPool = newRowPool(defaultRowPoolSize)
 
-// walSchema is a shared schema that all wals use. it comes with minor cpu and memory improvements
-var walSchema = parquet.SchemaOf(&Trace{})
-
 // path + filename = folder to create
 //   path/folder/00001
 //   	        /00002
@@ -227,7 +224,13 @@ func (w *walBlockFlush) file(ctx context.Context) (*pageFile, error) {
 	size := info.Size()
 
 	wr := newWalReaderAt(ctx, file)
-	pf, err := parquet.OpenFile(wr, size, parquet.SkipBloomFilters(true), parquet.SkipPageIndex(true), parquet.FileSchema(walSchema))
+	o := []parquet.FileOption{
+		parquet.SkipBloomFilters(true),
+		parquet.SkipPageIndex(true),
+		parquet.FileSchema(parquetSchema),
+	}
+
+	pf, err := parquet.OpenFile(wr, size, o...)
 	if err != nil {
 		return nil, fmt.Errorf("error opening parquet file: %w", err)
 	}
@@ -387,7 +390,7 @@ func (b *walBlock) openWriter() (err error) {
 
 	if b.writer == nil {
 		b.writer = parquet.NewGenericWriter[*Trace](b.file, &parquet.WriterConfig{
-			Schema: walSchema,
+			Schema: parquetSchema,
 			// setting this value low massively reduces the amount of static memory we hold onto in highly multi-tenant environments at the cost of
 			// cutting pages more aggressively when writing column chunks
 			PageBufferSize: 1024,