perf(v2): static parquet page buffer size

Author: kolesnikovae

pkg/experiment/block/block.go

@@ -23,29 +23,10 @@ const (
 	symbolsPrefetchSize = 32 << 10
 )
 
-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))
-}
+const (
+	parquetReadBufferSize      = 2 << 20
+	parquetPageWriteBufferSize = 4 << 20
+)
 
 func estimateFooterSize(size int64) int64 {
 	var s int64
@@ -67,17 +48,3 @@ 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,13 +374,7 @@ func (m *datasetCompaction) registerSampleObserver(observer SampleObserver) {
 }
 
 func (m *datasetCompaction) open(ctx context.Context, w io.Writer) (err error) {
-	var estimatedProfileTableSize int64
-	for _, ds := range m.datasets {
-		estimatedProfileTableSize += ds.sectionSize(SectionProfiles)
-	}
-	pageBufferSize := estimatePageBufferSize(estimatedProfileTableSize)
-	m.profilesWriter = newProfileWriter(pageBufferSize, w)
-
+	m.profilesWriter = newProfileWriter(w)
 	m.indexRewriter = newIndexRewriter()
 	m.symbolsRewriter = newSymbolsRewriter()
 

pkg/experiment/block/section_profiles.go

@@ -43,7 +43,7 @@ func openProfileTable(_ context.Context, s *Dataset) (err error) {
 			parquet.SkipBloomFilters(true),
 			parquet.SkipPageIndex(true),
 			parquet.FileReadMode(parquet.ReadModeAsync),
-			parquet.ReadBufferSize(estimateReadBufferSize(size)))
+			parquet.ReadBufferSize(parquetReadBufferSize))
 	}
 	if err != nil {
 		return fmt.Errorf("opening profile parquet table: %w", err)
@@ -163,12 +163,12 @@ type profilesWriter struct {
 	profiles uint64
 }
 
-func newProfileWriter(pageBufferSize int, w io.Writer) *profilesWriter {
+func newProfileWriter(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(pageBufferSize),
+			parquet.PageBufferSize(parquetPageWriteBufferSize),
 			// 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 = 32 << 10
+const segmentsParquetWriteBufferSize = 256 << 10
 
 func WriteProfiles(metrics *HeadMetrics, profiles []v1.InMemoryProfile) ([]byte, error) {
 	buf := &bytes.Buffer{}