WIP: Symbols compaction

pkg/iter/iter.go

@@ -101,6 +101,20 @@ func NewSliceSeekIterator[A constraints.Ordered](s []A) SeekIterator[A, A] {
 	}
 }
 
+type slicePositionIterator[T constraints.Integer, M any] struct {
+	i Iterator[T]
+	s []M
+}
+
+func NewSliceIndexIterator[T constraints.Integer, M any](s []M, i Iterator[T]) Iterator[M] {
+	return slicePositionIterator[T, M]{s: s, i: i}
+}
+
+func (i slicePositionIterator[T, M]) Next() bool   { return i.i.Next() }
+func (i slicePositionIterator[T, M]) At() M        { return i.s[i.i.At()] }
+func (i slicePositionIterator[T, M]) Err() error   { return i.i.Err() }
+func (i slicePositionIterator[T, M]) Close() error { return i.i.Close() }
+
 type sliceSeekIterator[A constraints.Ordered] struct {
 	*sliceIterator[A]
 }

pkg/phlaredb/block_querier.go

@@ -305,7 +305,7 @@ type singleBlockQuerier struct {
 type StacktraceDB interface {
 	Open(ctx context.Context) error
 	Close() error
-	Resolve(ctx context.Context, mapping uint64, locs locationsIdsByStacktraceID, stacktraceIDs []uint32) error
+	Resolve(ctx context.Context, partition uint64, locs symdb.StacktraceInserter, stacktraceIDs []uint32) error
 }
 
 type stacktraceResolverV1 struct {
@@ -321,14 +321,17 @@ func (r *stacktraceResolverV1) Close() error {
 	return r.stacktraces.Close()
 }
 
-func (r *stacktraceResolverV1) Resolve(ctx context.Context, mapping uint64, locs locationsIdsByStacktraceID, stacktraceIDs []uint32) error {
+func (r *stacktraceResolverV1) Resolve(ctx context.Context, _ uint64, locs symdb.StacktraceInserter, stacktraceIDs []uint32) error {
 	stacktraces := repeatedColumnIter(ctx, r.stacktraces.file, "LocationIDs.list.element", iter.NewSliceIterator(stacktraceIDs))
 	defer stacktraces.Close()
-
+	t := make([]int32, 0, 64)
 	for stacktraces.Next() {
+		t = t[:0]
 		s := stacktraces.At()
-		locs.addFromParquet(int64(s.Row), s.Values)
-
+		for i, v := range s.Values {
+			t[i] = v.Int32()
+		}
+		locs.InsertStacktrace(s.Row, t)
 	}
 	return stacktraces.Err()
 }
@@ -351,19 +354,14 @@ func (r *stacktraceResolverV2) Close() error {
 	return nil
 }
 
-func (r *stacktraceResolverV2) Resolve(ctx context.Context, mapping uint64, locs locationsIdsByStacktraceID, stacktraceIDs []uint32) error {
+func (r *stacktraceResolverV2) Resolve(ctx context.Context, mapping uint64, locs symdb.StacktraceInserter, stacktraceIDs []uint32) error {
 	mr, ok := r.reader.MappingReader(mapping)
 	if !ok {
 		return nil
 	}
 	resolver := mr.StacktraceResolver()
 	defer resolver.Release()
-
-	return resolver.ResolveStacktraces(ctx, symdb.StacktraceInserterFn(
-		func(stacktraceID uint32, locations []int32) {
-			locs.add(int64(stacktraceID), locations)
-		},
-	), stacktraceIDs)
+	return resolver.ResolveStacktraces(ctx, locs, stacktraceIDs)
 }
 
 func NewSingleBlockQuerierFromMeta(phlarectx context.Context, bucketReader phlareobj.Bucket, meta *block.Meta) *singleBlockQuerier {
@@ -435,6 +433,11 @@ func (b *singleBlockQuerier) Index() IndexReader {
 	return b.index
 }
 
+func (b *singleBlockQuerier) Symbols() SymbolsReader {
+	// TODO(kolesnikovae)
+	return nil
+}
+
 func (b *singleBlockQuerier) Meta() block.Meta {
 	if b.meta == nil {
 		return block.Meta{}

pkg/phlaredb/block_symbols_appender.go

@@ -0,0 +1,19 @@
+package phlaredb
+
+import schemav1 "github.com/grafana/phlare/pkg/phlaredb/schemas/v1"
+
+// TODO(kolesnikovae): Refactor to symdb.
+
+type SymbolsWriter interface {
+	SymbolsAppender(partition uint64) (SymbolsAppender, error)
+}
+
+type SymbolsAppender interface {
+	AppendStacktrace([]int32) uint32
+	AppendLocation(*schemav1.InMemoryLocation) uint32
+	AppendMapping(*schemav1.InMemoryMapping) uint32
+	AppendFunction(*schemav1.InMemoryFunction) uint32
+	AppendString(string) uint32
+
+	Flush() error
+}

pkg/phlaredb/block_symbols_reader.go

@@ -0,0 +1,92 @@
+package phlaredb
+
+import (
+	"context"
+
+	"github.com/grafana/phlare/pkg/iter"
+	schemav1 "github.com/grafana/phlare/pkg/phlaredb/schemas/v1"
+	"github.com/grafana/phlare/pkg/phlaredb/symdb"
+)
+
+// TODO(kolesnikovae): Refactor to symdb.
+
+type SymbolsReader interface {
+	SymbolsResolver(partition uint64) (SymbolsResolver, error)
+}
+
+type SymbolsResolver interface {
+	ResolveStacktraces(ctx context.Context, dst symdb.StacktraceInserter, stacktraces []uint32) error
+
+	Locations(iter.Iterator[uint32]) iter.Iterator[*schemav1.InMemoryLocation]
+	Mappings(iter.Iterator[uint32]) iter.Iterator[*schemav1.InMemoryMapping]
+	Functions(iter.Iterator[uint32]) iter.Iterator[*schemav1.InMemoryFunction]
+	Strings(iter.Iterator[uint32]) iter.Iterator[string]
+
+	WriteStats(*SymbolStats)
+}
+
+type SymbolStats struct {
+	StacktracesTotal int
+	LocationsTotal   int
+	MappingsTotal    int
+	FunctionsTotal   int
+	StringsTotal     int
+}
+
+type inMemorySymbolsReader struct {
+	partitions map[uint64]*inMemorySymbolsResolver
+
+	// TODO(kolesnikovae): Split into partitions.
+	strings     inMemoryparquetReader[string, *schemav1.StringPersister]
+	functions   inMemoryparquetReader[*schemav1.InMemoryFunction, *schemav1.FunctionPersister]
+	locations   inMemoryparquetReader[*schemav1.InMemoryLocation, *schemav1.LocationPersister]
+	mappings    inMemoryparquetReader[*schemav1.InMemoryMapping, *schemav1.MappingPersister]
+	stacktraces StacktraceDB
+}
+
+func (r *inMemorySymbolsReader) Symbols(partition uint64) SymbolsResolver {
+	p, ok := r.partitions[partition]
+	if !ok {
+		p = &inMemorySymbolsResolver{
+			partition: 0,
+			ctx:       nil,
+			reader:    nil,
+		}
+		r.partitions[partition] = p
+	}
+	return p
+}
+
+type inMemorySymbolsResolver struct {
+	partition uint64
+	ctx       context.Context
+	reader    *inMemorySymbolsReader
+}
+
+func (s inMemorySymbolsResolver) ResolveStacktraces(ctx context.Context, dst symdb.StacktraceInserter, stacktraces []uint32) error {
+	return s.reader.stacktraces.Resolve(ctx, s.partition, dst, stacktraces)
+}
+
+func (s inMemorySymbolsResolver) Locations(i iter.Iterator[uint32]) iter.Iterator[*schemav1.InMemoryLocation] {
+	return iter.NewSliceIndexIterator(s.reader.locations.cache, i)
+}
+
+func (s inMemorySymbolsResolver) Mappings(i iter.Iterator[uint32]) iter.Iterator[*schemav1.InMemoryMapping] {
+	return iter.NewSliceIndexIterator(s.reader.mappings.cache, i)
+}
+
+func (s inMemorySymbolsResolver) Functions(i iter.Iterator[uint32]) iter.Iterator[*schemav1.InMemoryFunction] {
+	return iter.NewSliceIndexIterator(s.reader.functions.cache, i)
+}
+
+func (s inMemorySymbolsResolver) Strings(i iter.Iterator[uint32]) iter.Iterator[string] {
+	return iter.NewSliceIndexIterator(s.reader.strings.cache, i)
+}
+
+func (s inMemorySymbolsResolver) WriteStats(stats *SymbolStats) {
+	stats.StacktracesTotal = 0 // TODO
+	stats.LocationsTotal = int(s.reader.locations.NumRows())
+	stats.MappingsTotal = int(s.reader.mappings.NumRows())
+	stats.FunctionsTotal = int(s.reader.functions.NumRows())
+	stats.StringsTotal = int(s.reader.strings.NumRows())
+}