store: add initial symbol tables support

pkg/query/querier.go

@@ -5,12 +5,14 @@ package query
 
 import (
 	"context"
+	"fmt"
 	"sort"
 	"strings"
 	"sync"
 	"time"
 
 	"github.com/go-kit/log"
+	"github.com/gogo/protobuf/types"
 	"github.com/opentracing/opentracing-go"
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
@@ -24,6 +26,7 @@ import (
 	"github.com/thanos-io/thanos/pkg/extprom"
 	"github.com/thanos-io/thanos/pkg/gate"
 	"github.com/thanos-io/thanos/pkg/store"
+	"github.com/thanos-io/thanos/pkg/store/hintspb"
 	"github.com/thanos-io/thanos/pkg/store/labelpb"
 	"github.com/thanos-io/thanos/pkg/store/storepb"
 	"github.com/thanos-io/thanos/pkg/tracing"
@@ -208,6 +211,47 @@ type seriesServer struct {
 	seriesSet      []storepb.Series
 	seriesSetStats storepb.SeriesStatsCounter
 	warnings       []string
+	symbolTables   []map[uint64]string
+
+	compressedSeriesSet []storepb.CompressedSeries
+}
+
+func (s *seriesServer) DecompressSeries() error {
+	for _, cs := range s.compressedSeriesSet {
+		newSeries := &storepb.Series{
+			Chunks: cs.Chunks,
+		}
+
+		lbls := labels.Labels{}
+
+		for _, cLabel := range cs.Labels {
+			var name, val string
+			for _, symTable := range s.symbolTables {
+				if foundName, ok := symTable[uint64(cLabel.NameRef)]; ok {
+					name = foundName
+				}
+
+				if foundValue, ok := symTable[uint64(cLabel.ValueRef)]; ok {
+					val = foundValue
+				}
+			}
+			if name == "" {
+				return fmt.Errorf("found no reference for name ref %d", cLabel.NameRef)
+			}
+			if val == "" {
+				return fmt.Errorf("found no reference for value ref %d", cLabel.ValueRef)
+			}
+
+			lbls = append(lbls, labels.Label{
+				Name:  name,
+				Value: val,
+			})
+		}
+
+		newSeries.Labels = labelpb.ZLabelsFromPromLabels(lbls)
+		s.seriesSet = append(s.seriesSet, *newSeries)
+	}
+	return nil
 }
 
 func (s *seriesServer) Send(r *storepb.SeriesResponse) error {
@@ -222,6 +266,24 @@ func (s *seriesServer) Send(r *storepb.SeriesResponse) error {
 		return nil
 	}
 
+	if r.GetCompressedSeries() != nil {
+		s.compressedSeriesSet = append(s.compressedSeriesSet, *r.GetCompressedSeries())
+		s.seriesSetStats.Count(r.GetCompressedSeries())
+		return nil
+	}
+
+	if r.GetHints() != nil {
+		var seriesResponseHints hintspb.SeriesResponseHints
+
+		// Some other, unknown type. Skip it.
+		if err := types.UnmarshalAny(r.GetHints(), &seriesResponseHints); err != nil {
+			return nil
+		}
+
+		s.symbolTables = append(s.symbolTables, seriesResponseHints.StringSymbolTable)
+		return nil
+	}
+
 	// Unsupported field, skip.
 	return nil
 }
@@ -369,6 +431,10 @@ func (q *querier) selectFn(ctx context.Context, hints *storage.SelectHints, ms .
 		warns = append(warns, errors.New(w))
 	}
 
+	if err := resp.DecompressSeries(); err != nil {
+		return nil, storepb.SeriesStatsCounter{}, errors.Wrap(err, "decompressing series")
+	}
+
 	// Delete the metric's name from the result because that's what the
 	// PromQL does either way and we want our iterator to work with data
 	// that was either pushed down or not.

pkg/store/bucket.go

@@ -350,7 +350,7 @@ type BucketStore struct {
 	postingOffsetsInMemSampling int
 
 	// Enables hints in the Series() response.
-	enableSeriesResponseHints bool
+	enableQueriedBlocksHints bool
 
 	enableChunkHashCalculation bool
 }
@@ -477,7 +477,7 @@ func NewBucketStore(
 		partitioner:                 partitioner,
 		enableCompatibilityLabel:    enableCompatibilityLabel,
 		postingOffsetsInMemSampling: postingOffsetsInMemSampling,
-		enableSeriesResponseHints:   enableSeriesResponseHints,
+		enableQueriedBlocksHints:    enableSeriesResponseHints,
 		enableChunkHashCalculation:  enableChunkHashCalculation,
 		seriesBatchSize:             SeriesBatchSize,
 	}
@@ -1186,6 +1186,7 @@ func (s *BucketStore) Series(req *storepb.SeriesRequest, srv storepb.Store_Serie
 	}
 
 	s.mtx.RLock()
+
 	for _, bs := range s.blockSets {
 		blockMatchers, ok := bs.labelMatchers(matchers...)
 		if !ok {
@@ -1202,7 +1203,7 @@ func (s *BucketStore) Series(req *storepb.SeriesRequest, srv storepb.Store_Serie
 			blk := b
 			gctx := gctx
 
-			if s.enableSeriesResponseHints {
+			if s.enableQueriedBlocksHints {
 				// Keep track of queried blocks.
 				resHints.AddQueriedBlock(blk.meta.ULID)
 			}
@@ -1250,6 +1251,8 @@ func (s *BucketStore) Series(req *storepb.SeriesRequest, srv storepb.Store_Serie
 					shardMatcher,
 					false,
 					s.metrics.emptyPostingCount,
+					req.MaximumStringSlots,
+					nil,
 				)
 
 				mtx.Lock()
@@ -1311,6 +1314,8 @@ func (s *BucketStore) Series(req *storepb.SeriesRequest, srv storepb.Store_Serie
 		s.metrics.seriesBlocksQueried.Observe(float64(stats.blocksQueried))
 	}
 
+	symbolTableBuilder := newSymbolTableBuilder(req.MaximumStringSlots)
+
 	// Merge the sub-results from each selected block.
 	tracing.DoInSpan(ctx, "bucket_store_merge_all", func(ctx context.Context) {
 		defer func() {
@@ -1338,7 +1343,35 @@ func (s *BucketStore) Series(req *storepb.SeriesRequest, srv storepb.Store_Serie
 					stats.mergedChunksCount += len(series.Chunks)
 					s.metrics.chunkSizeBytes.Observe(float64(chunksSize(series.Chunks)))
 				}
+
+				lset := series.Labels
+
+				var compressedResponse bool = true
+				compressedLabels := make([]labelpb.CompressedLabel, 0, len(lset))
+
+				for _, lbl := range lset {
+					nameRef, nok := symbolTableBuilder.getOrStoreString(lbl.Name)
+					valueRef, vok := symbolTableBuilder.getOrStoreString(lbl.Value)
+
+					if !nok || !vok {
+						compressedResponse = false
+						break
+					} else if compressedResponse && nok && vok {
+						compressedLabels = append(compressedLabels, labelpb.CompressedLabel{
+							NameRef:  nameRef,
+							ValueRef: valueRef,
+						})
+					}
+				}
+
+				if compressedResponse {
+					at = storepb.NewCompressedSeriesResponse(&storepb.CompressedSeries{
+						Labels: compressedLabels,
+						Chunks: series.Chunks,
+					})
+				}
 			}
+
 			if err = srv.Send(at); err != nil {
 				err = status.Error(codes.Unknown, errors.Wrap(err, "send series response").Error())
 				return
@@ -1353,7 +1386,9 @@ func (s *BucketStore) Series(req *storepb.SeriesRequest, srv storepb.Store_Serie
 		return err
 	}
 
-	if s.enableSeriesResponseHints {
+	resHints.StringSymbolTable = symbolTableBuilder.getTable()
+
+	{
 		var anyHints *types.Any
 
 		if anyHints, err = types.MarshalAny(resHints); err != nil {

pkg/store/bucket_test.go

@@ -1378,6 +1378,9 @@ func benchBucketSeries(t testutil.TB, skipChunk bool, samplesPerSeries, totalSer
 			},
 			// This does not cut chunks properly, but those are assured against for non benchmarks only, where we use 100% case only.
 			ExpectedSeries: series[:seriesCut],
+			ExpectedHints: []hintspb.SeriesResponseHints{
+				{},
+			},
 		})
 	}
 	storetestutil.TestServerSeries(t, st, bCases...)