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,23 @@ func (s *seriesServer) Send(r *storepb.SeriesResponse) error {
 		return nil
 	}
 
+	if r.GetCompressedSeries() != nil {
+		s.compressedSeriesSet = append(s.compressedSeriesSet, *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 +430,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

@@ -333,7 +333,7 @@ type BucketStore struct {
 	postingOffsetsInMemSampling int
 
 	// Enables hints in the Series() response.
-	enableSeriesResponseHints bool
+	enableQueriedBlocksHints bool
 
 	enableChunkHashCalculation bool
 }
@@ -454,7 +454,7 @@ func NewBucketStore(
 		partitioner:                 partitioner,
 		enableCompatibilityLabel:    enableCompatibilityLabel,
 		postingOffsetsInMemSampling: postingOffsetsInMemSampling,
-		enableSeriesResponseHints:   enableSeriesResponseHints,
+		enableQueriedBlocksHints:    enableSeriesResponseHints,
 		enableChunkHashCalculation:  enableChunkHashCalculation,
 	}
 
@@ -1083,6 +1083,8 @@ func (s *BucketStore) Series(req *storepb.SeriesRequest, srv storepb.Store_Serie
 		}
 	}
 
+	lookupTable := newLookupTableBuilder(req.MaximumStringSlots)
+
 	s.mtx.RLock()
 	for _, bs := range s.blockSets {
 		blockMatchers, ok := bs.labelMatchers(matchers...)
@@ -1100,7 +1102,7 @@ func (s *BucketStore) Series(req *storepb.SeriesRequest, srv storepb.Store_Serie
 			b := b
 			gctx := gctx
 
-			if s.enableSeriesResponseHints {
+			if s.enableQueriedBlocksHints {
 				// Keep track of queried blocks.
 				resHints.AddQueriedBlock(b.meta.ULID)
 			}
@@ -1231,9 +1233,38 @@ func (s *BucketStore) Series(req *storepb.SeriesRequest, srv storepb.Store_Serie
 				s.metrics.chunkSizeBytes.Observe(float64(chunksSize(series.Chunks)))
 			}
 			series.Labels = labelpb.ZLabelsFromPromLabels(lset)
-			if err = srv.Send(storepb.NewSeriesResponse(&series)); err != nil {
-				err = status.Error(codes.Unknown, errors.Wrap(err, "send series response").Error())
-				return
+
+			var compressedResponse bool
+			compressedLabels := make([]labelpb.CompressedLabel, 0, len(lset))
+
+			for _, lbl := range lset {
+				nameRef, nerr := lookupTable.putString(lbl.Name)
+				valueRef, verr := lookupTable.putString(lbl.Value)
+
+				if nerr != nil || verr != nil {
+					compressedResponse = false
+					break
+				} else if compressedResponse && nerr == nil && verr == nil {
+					compressedLabels = append(compressedLabels, labelpb.CompressedLabel{
+						NameRef:  nameRef,
+						ValueRef: valueRef,
+					})
+				}
+			}
+
+			if compressedResponse {
+				if err = srv.Send(storepb.NewCompressedSeriesResponse(&storepb.CompressedSeries{
+					Labels: compressedLabels,
+					Chunks: series.Chunks,
+				})); err != nil {
+					err = status.Error(codes.Unknown, errors.Wrap(err, "send series response").Error())
+					return
+				}
+			} else {
+				if err = srv.Send(storepb.NewSeriesResponse(&series)); err != nil {
+					err = status.Error(codes.Unknown, errors.Wrap(err, "send series response").Error())
+					return
+				}
 			}
 		}
 		if set.Err() != nil {
@@ -1246,7 +1277,9 @@ func (s *BucketStore) Series(req *storepb.SeriesRequest, srv storepb.Store_Serie
 		err = nil
 	})
 
-	if s.enableSeriesResponseHints {
+	resHints.StringSymbolTable = lookupTable.getTable()
+
+	{
 		var anyHints *types.Any
 
 		if anyHints, err = types.MarshalAny(resHints); err != nil {

pkg/store/bucket_test.go

@@ -1379,6 +1379,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...)