Revert "perf(v2): static parquet page buffer size (#4208)" (#4221)

This reverts commit 77754da.

Author: kolesnikovae

pkg/experiment/block/block.go

@@ -23,11 +23,29 @@ const (
 	symbolsPrefetchSize = 32 << 10
 )
 
-const (
-	// Each 2MB translates to an I/O read op.
-	parquetReadBufferSize      = 2 << 20
-	parquetPageWriteBufferSize = 1 << 20
-)
+func estimateReadBufferSize(s int64) int {
+	const minSize = 64 << 10
+	const maxSize = 1 << 20
+	// Parquet has global buffer map, where buffer size is key,
+	// so we want a low cardinality here.
+	e := nextPowerOfTwo(uint32(s / 10))
+	if e < minSize {
+		return minSize
+	}
+	return int(min(e, maxSize))
+}
+
+// This is a verbatim copy of estimateReadBufferSize.
+// It's kept for the sake of clarity and to avoid confusion.
+func estimatePageBufferSize(s int64) int {
+	const minSize = 64 << 10
+	const maxSize = 1 << 20
+	e := nextPowerOfTwo(uint32(s / 10))
+	if e < minSize {
+		return minSize
+	}
+	return int(min(e, maxSize))
+}
 
 func estimateFooterSize(size int64) int64 {
 	var s int64
@@ -49,3 +67,17 @@ func estimateFooterSize(size int64) int64 {
 	}
 	return s
 }
+
+func nextPowerOfTwo(n uint32) uint32 {
+	if n == 0 {
+		return 1
+	}
+	n--
+	n |= n >> 1
+	n |= n >> 2
+	n |= n >> 4
+	n |= n >> 8
+	n |= n >> 16
+	n++
+	return n
+}

pkg/experiment/block/compaction.go

@@ -374,7 +374,13 @@ func (m *datasetCompaction) registerSampleObserver(observer SampleObserver) {
 }
 
 func (m *datasetCompaction) open(ctx context.Context, w io.Writer) (err error) {
-	m.profilesWriter = newProfileWriter(w)
+	var estimatedProfileTableSize int64
+	for _, ds := range m.datasets {
+		estimatedProfileTableSize += ds.sectionSize(SectionProfiles)
+	}
+	pageBufferSize := estimatePageBufferSize(estimatedProfileTableSize)
+	m.profilesWriter = newProfileWriter(pageBufferSize, w)
+
 	m.indexRewriter = newIndexRewriter()
 	m.symbolsRewriter = newSymbolsRewriter()
 

pkg/experiment/block/object.go

@@ -196,8 +196,6 @@ func (obj *Object) Download(ctx context.Context) error {
 	return nil
 }
 
-func (obj *Object) Path() string { return obj.path }
-
 func (obj *Object) Metadata() *metastorev1.BlockMeta { return obj.meta }
 
 func (obj *Object) SetMetadata(md *metastorev1.BlockMeta) { obj.meta = md }

pkg/experiment/block/section_profiles.go

@@ -41,7 +41,7 @@ func openProfileTable(_ context.Context, s *Dataset) (err error) {
 			estimateFooterSize(size),
 			parquet.SkipBloomFilters(true),
 			parquet.FileReadMode(parquet.ReadModeAsync),
-			parquet.ReadBufferSize(parquetReadBufferSize))
+			parquet.ReadBufferSize(estimateReadBufferSize(size)))
 	}
 	if err != nil {
 		return fmt.Errorf("opening profile parquet table: %w", err)
@@ -183,12 +183,12 @@ type profilesWriter struct {
 	profiles uint64
 }
 
-func newProfileWriter(w io.Writer) *profilesWriter {
+func newProfileWriter(pageBufferSize int, w io.Writer) *profilesWriter {
 	return &profilesWriter{
 		buf: make([]parquet.Row, 1),
 		GenericWriter: parquet.NewGenericWriter[*schemav1.Profile](w,
 			parquet.CreatedBy("github.com/grafana/pyroscope/", build.Version, build.Revision),
-			parquet.PageBufferSize(parquetPageWriteBufferSize),
+			parquet.PageBufferSize(pageBufferSize),
 			// Note that parquet keeps ALL RG pages in memory (ColumnPageBuffers).
 			parquet.MaxRowsPerRowGroup(maxRowsPerRowGroup),
 			schemav1.ProfilesSchema,

pkg/experiment/ingester/memdb/profiles.go

@@ -10,7 +10,7 @@ import (
 	"github.com/grafana/pyroscope/pkg/util/build"
 )
 
-const segmentsParquetWriteBufferSize = 256 << 10
+const segmentsParquetWriteBufferSize = 32 << 10
 
 func WriteProfiles(metrics *HeadMetrics, profiles []v1.InMemoryProfile) ([]byte, error) {
 	buf := &bytes.Buffer{}

pkg/experiment/query_backend/query.go

@@ -86,9 +86,7 @@ type blockContext struct {
 
 func (b *blockContext) execute() error {
 	var span opentracing.Span
-	span, b.ctx = opentracing.StartSpanFromContext(b.ctx, "blockContext.execute", opentracing.Tags{
-		"object_name": b.obj.Path(),
-	})
+	span, b.ctx = opentracing.StartSpanFromContext(b.ctx, "blockContext.execute")
 	defer span.Finish()
 
 	if idx := b.datasetIndex(); idx != nil {