Don't convert chunks to matrixes and then merges them...

cmd/lite/main.go

@@ -11,7 +11,6 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/common/route"
 	"github.com/prometheus/prometheus/config"
-	"github.com/prometheus/prometheus/promql"
 	"github.com/prometheus/prometheus/web/api/v1"
 	"github.com/prometheus/tsdb"
 	"google.golang.org/grpc"
@@ -124,8 +123,7 @@ func main() {
 	tableManager.Start()
 	defer tableManager.Stop()
 
-	engine := promql.NewEngine(util.Logger, nil, querierConfig.MaxConcurrent, querierConfig.Timeout)
-	queryable := querier.NewQueryable(dist, chunkStore)
+	queryable, engine := querier.Make(querierConfig, dist, chunkStore)
 
 	if configStoreConfig.ConfigsAPIURL.String() != "" || configStoreConfig.DBConfig.URI != "" {
 		rulesAPI, err := ruler.NewRulesAPI(configStoreConfig)
@@ -186,7 +184,7 @@ func main() {
 
 	subrouter := server.HTTP.PathPrefix("/api/prom").Subrouter()
 	subrouter.PathPrefix("/api/v1").Handler(activeMiddleware.Wrap(promRouter))
-	subrouter.Path("/read").Handler(activeMiddleware.Wrap(http.HandlerFunc(queryable.RemoteReadHandler)))
+	subrouter.Path("/read").Handler(activeMiddleware.Wrap(querier.RemoteReadHandler(queryable)))
 	subrouter.Path("/validate_expr").Handler(activeMiddleware.Wrap(http.HandlerFunc(dist.ValidateExprHandler)))
 	subrouter.Path("/user_stats").Handler(activeMiddleware.Wrap(http.HandlerFunc(dist.UserStatsHandler)))
 

cmd/querier/main.go

@@ -11,7 +11,6 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/common/route"
 	"github.com/prometheus/prometheus/config"
-	"github.com/prometheus/prometheus/promql"
 	"github.com/prometheus/prometheus/web/api/v1"
 	"github.com/prometheus/tsdb"
 
@@ -88,9 +87,7 @@ func main() {
 	}
 	defer chunkStore.Stop()
 
-	queryable := querier.NewQueryable(dist, chunkStore)
-
-	engine := promql.NewEngine(util.Logger, nil, querierConfig.MaxConcurrent, querierConfig.Timeout)
+	queryable, engine := querier.Make(querierConfig, dist, chunkStore)
 	api := v1.NewAPI(
 		engine,
 		queryable,
@@ -107,7 +104,7 @@ func main() {
 
 	subrouter := server.HTTP.PathPrefix("/api/prom").Subrouter()
 	subrouter.PathPrefix("/api/v1").Handler(middleware.AuthenticateUser.Wrap(promRouter))
-	subrouter.Path("/read").Handler(middleware.AuthenticateUser.Wrap(http.HandlerFunc(queryable.RemoteReadHandler)))
+	subrouter.Path("/read").Handler(middleware.AuthenticateUser.Wrap(querier.RemoteReadHandler(queryable)))
 	subrouter.Path("/validate_expr").Handler(middleware.AuthenticateUser.Wrap(http.HandlerFunc(dist.ValidateExprHandler)))
 	subrouter.Path("/user_stats").Handler(middleware.AuthenticateUser.Wrap(http.HandlerFunc(dist.UserStatsHandler)))
 

cmd/ruler/main.go

@@ -36,6 +36,7 @@ func main() {
 		schemaConfig      chunk.SchemaConfig
 		storageConfig     storage.Config
 		configStoreConfig ruler.ConfigStoreConfig
+		querierConfig     querier.Config
 		logLevel          util.LogLevel
 	)
 
@@ -44,7 +45,8 @@ func main() {
 	defer trace.Close()
 
 	util.RegisterFlags(&serverConfig, &ringConfig, &distributorConfig,
-		&rulerConfig, &chunkStoreConfig, &storageConfig, &schemaConfig, &configStoreConfig, &logLevel)
+		&rulerConfig, &chunkStoreConfig, &storageConfig, &schemaConfig, &configStoreConfig,
+		&querierConfig, &logLevel)
 	flag.Parse()
 
 	util.InitLogger(logLevel.AllowedLevel)
@@ -78,7 +80,7 @@ func main() {
 	prometheus.MustRegister(dist)
 
 	engine := promql.NewEngine(util.Logger, prometheus.DefaultRegisterer, rulerConfig.NumWorkers, rulerConfig.GroupTimeout)
-	queryable := querier.NewQueryable(dist, chunkStore)
+	queryable := querier.NewQueryable(dist, chunkStore, querierConfig.Iterators)
 
 	rlr, err := ruler.NewRuler(rulerConfig, engine, queryable, dist)
 	if err != nil {

pkg/chunk/chunk.go

@@ -319,36 +319,31 @@ func equalByKey(a, b Chunk) bool {
 		a.From == b.From && a.Through == b.Through && a.Checksum == b.Checksum
 }
 
-func chunksToMatrix(ctx context.Context, chunks []Chunk, from, through model.Time) (model.Matrix, error) {
+// ChunksToMatrix converts a set of chunks to a model.Matrix.
+func ChunksToMatrix(ctx context.Context, chunks []Chunk, from, through model.Time) (model.Matrix, error) {
 	sp, ctx := ot.StartSpanFromContext(ctx, "chunksToMatrix")
 	defer sp.Finish()
 	sp.LogFields(otlog.Int("chunks", len(chunks)))
 
 	// Group chunks by series, sort and dedupe samples.
-	sampleStreams := map[model.Fingerprint]*model.SampleStream{}
+	metrics := map[model.Fingerprint]model.Metric{}
+	samplesBySeries := map[model.Fingerprint][][]model.SamplePair{}
 	for _, c := range chunks {
-		ss, ok := sampleStreams[c.Fingerprint]
-		if !ok {
-			ss = &model.SampleStream{
-				Metric: c.Metric,
-			}
-			sampleStreams[c.Fingerprint] = ss
-		}
-
-		samples, err := c.Samples(from, through)
+		ss, err := c.Samples(from, through)
 		if err != nil {
 			return nil, err
 		}
 
-		ss.Values = util.MergeSampleSets(ss.Values, samples)
+		metrics[c.Fingerprint] = c.Metric
+		samplesBySeries[c.Fingerprint] = append(samplesBySeries[c.Fingerprint], ss)
 	}
-	sp.LogFields(otlog.Int("sample streams", len(sampleStreams)))
+	sp.LogFields(otlog.Int("series", len(samplesBySeries)))
 
-	matrix := make(model.Matrix, 0, len(sampleStreams))
-	for _, ss := range sampleStreams {
+	matrix := make(model.Matrix, 0, len(samplesBySeries))
+	for fp, ss := range samplesBySeries {
 		matrix = append(matrix, &model.SampleStream{
-			Metric: ss.Metric,
-			Values: ss.Values,
+			Metric: metrics[fp],
+			Values: util.MergeNSampleSets(ss...),
 		})
 	}
 

pkg/chunk/chunk_store.go

@@ -195,7 +195,7 @@ func (s *spanLogger) Log(kvps ...interface{}) error {
 }
 
 // Get implements ChunkStore
-func (c *Store) Get(ctx context.Context, from, through model.Time, allMatchers ...*labels.Matcher) (model.Matrix, error) {
+func (c *Store) Get(ctx context.Context, from, through model.Time, allMatchers ...*labels.Matcher) ([]Chunk, error) {
 	log, ctx := newSpanLogger(ctx, "ChunkStore.Get")
 	defer log.Span.Finish()
 
@@ -227,19 +227,11 @@ func (c *Store) Get(ctx context.Context, from, through model.Time, allMatchers .
 	metricNameMatcher, matchers, ok := extract.MetricNameMatcherFromMatchers(allMatchers)
 	if ok && metricNameMatcher.Type == labels.MatchEqual {
 		log.Span.SetTag("metric", metricNameMatcher.Value)
-		return c.getMetricNameMatrix(ctx, from, through, matchers, metricNameMatcher.Value)
+		return c.getMetricNameChunks(ctx, from, through, matchers, metricNameMatcher.Value)
 	}
 
 	// Otherwise we consult the metric name index first and then create queries for each matching metric name.
-	return c.getSeriesMatrix(ctx, from, through, matchers, metricNameMatcher)
-}
-
-func (c *Store) getMetricNameMatrix(ctx context.Context, from, through model.Time, allMatchers []*labels.Matcher, metricName string) (model.Matrix, error) {
-	chunks, err := c.getMetricNameChunks(ctx, from, through, allMatchers, metricName)
-	if err != nil {
-		return nil, err
-	}
-	return chunksToMatrix(ctx, chunks, from, through)
+	return c.getSeriesChunks(ctx, from, through, matchers, metricNameMatcher)
 }
 
 func (c *Store) getMetricNameChunks(ctx context.Context, from, through model.Time, allMatchers []*labels.Matcher, metricName string) ([]Chunk, error) {
@@ -345,7 +337,7 @@ func ProcessCacheResponse(chunks []Chunk, keys []string, bufs [][]byte) (found [
 	return
 }
 
-func (c *Store) getSeriesMatrix(ctx context.Context, from, through model.Time, allMatchers []*labels.Matcher, metricNameMatcher *labels.Matcher) (model.Matrix, error) {
+func (c *Store) getSeriesChunks(ctx context.Context, from, through model.Time, allMatchers []*labels.Matcher, metricNameMatcher *labels.Matcher) ([]Chunk, error) {
 	// Get all series from the index
 	userID, err := user.ExtractOrgID(ctx)
 	if err != nil {
@@ -406,7 +398,7 @@ outer:
 			}
 		}
 	}
-	return chunksToMatrix(ctx, chunks, from, through)
+	return chunks, nil
 }
 
 func (c *Store) lookupChunksByMetricName(ctx context.Context, from, through model.Time, matchers []*labels.Matcher, metricName string) ([]Chunk, error) {

pkg/chunk/chunk_store_test.go

@@ -218,7 +218,10 @@ func TestChunkStore_Get(t *testing.T) {
 				}
 
 				// Query with ordinary time-range
-				matrix1, err := store.Get(ctx, now.Add(-time.Hour), now, matchers...)
+				chunks1, err := store.Get(ctx, now.Add(-time.Hour), now, matchers...)
+				require.NoError(t, err)
+
+				matrix1, err := ChunksToMatrix(ctx, chunks1, now.Add(-time.Hour), now)
 				require.NoError(t, err)
 
 				sort.Sort(ByFingerprint(matrix1))
@@ -229,7 +232,10 @@ func TestChunkStore_Get(t *testing.T) {
 				}
 
 				// Pushing end of time-range into future should yield exact same resultset
-				matrix2, err := store.Get(ctx, now.Add(-time.Hour), now.Add(time.Hour*24*30), matchers...)
+				chunks2, err := store.Get(ctx, now.Add(-time.Hour), now.Add(time.Hour*24*30), matchers...)
+				require.NoError(t, err)
+
+				matrix2, err := ChunksToMatrix(ctx, chunks2, now.Add(-time.Hour), now)
 				require.NoError(t, err)
 
 				sort.Sort(ByFingerprint(matrix2))

pkg/chunk/chunk_test.go

@@ -188,7 +188,7 @@ func TestChunksToMatrix(t *testing.T) {
 			},
 		},
 	} {
-		matrix, err := chunksToMatrix(context.Background(), c.chunks, chunk1.From, chunk3.Through)
+		matrix, err := ChunksToMatrix(context.Background(), c.chunks, chunk1.From, chunk3.Through)
 		require.NoError(t, err)
 
 		sort.Sort(matrix)

pkg/ingester/ingester_test.go

@@ -3,6 +3,7 @@ package ingester
 import (
 	"context"
 	"fmt"
+	"math"
 	"net/http"
 	"sort"
 	"sync"
@@ -18,7 +19,6 @@ import (
 	"github.com/weaveworks/common/user"
 	"github.com/weaveworks/cortex/pkg/chunk"
 	"github.com/weaveworks/cortex/pkg/ingester/client"
-	"github.com/weaveworks/cortex/pkg/util"
 )
 
 type testStore struct {
@@ -86,36 +86,6 @@ func matrixToSamples(m model.Matrix) []model.Sample {
 	return samples
 }
 
-// chunksToMatrix converts a slice of chunks into a model.Matrix.
-func chunksToMatrix(chunks []chunk.Chunk) (model.Matrix, error) {
-	// Group chunks by series, sort and dedupe samples.
-	sampleStreams := map[model.Fingerprint]*model.SampleStream{}
-	for _, c := range chunks {
-		fp := c.Metric.Fingerprint()
-		ss, ok := sampleStreams[fp]
-		if !ok {
-			ss = &model.SampleStream{
-				Metric: c.Metric,
-			}
-			sampleStreams[fp] = ss
-		}
-
-		samples, err := c.Samples(c.From, c.Through)
-		if err != nil {
-			return nil, err
-		}
-
-		ss.Values = util.MergeSampleSets(ss.Values, samples)
-	}
-
-	matrix := make(model.Matrix, 0, len(sampleStreams))
-	for _, ss := range sampleStreams {
-		matrix = append(matrix, ss)
-	}
-
-	return matrix, nil
-}
-
 func TestIngesterAppend(t *testing.T) {
 	store, ing := newTestStore(t, defaultIngesterTestConfig())
 
@@ -154,7 +124,7 @@ func TestIngesterAppend(t *testing.T) {
 	// Read samples back via chunk store.
 	ing.Shutdown()
 	for _, userID := range userIDs {
-		res, err := chunksToMatrix(store.chunks[userID])
+		res, err := chunk.ChunksToMatrix(context.Background(), store.chunks[userID], model.Time(0), model.Time(math.MaxInt64))
 		require.NoError(t, err)
 		sort.Sort(res)
 		assert.Equal(t, testData[userID], res)

pkg/ingester/lifecycle_test.go

@@ -2,6 +2,7 @@ package ingester
 
 import (
 	"io"
+	"math"
 	"reflect"
 	"runtime"
 	"testing"
@@ -17,6 +18,7 @@ import (
 	"github.com/prometheus/prometheus/pkg/labels"
 
 	"github.com/weaveworks/common/user"
+	"github.com/weaveworks/cortex/pkg/chunk"
 	"github.com/weaveworks/cortex/pkg/ingester/client"
 	"github.com/weaveworks/cortex/pkg/ring"
 	"github.com/weaveworks/cortex/pkg/util"
@@ -314,7 +316,7 @@ func TestIngesterFlush(t *testing.T) {
 	})
 
 	// And check the store has the chunk
-	res, err := chunksToMatrix(store.chunks[userID])
+	res, err := chunk.ChunksToMatrix(context.Background(), store.chunks[userID], model.Time(0), model.Time(math.MaxInt64))
 	require.NoError(t, err)
 	assert.Equal(t, model.Matrix{
 		&model.SampleStream{

pkg/querier/benchmark_test.go

@@ -0,0 +1,27 @@
+package querier
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/prometheus/prometheus/promql"
+)
+
+var result *promql.Result
+
+func BenchmarkChunkQueryable(b *testing.B) {
+	for _, encoding := range encodings {
+		store, from := makeMockChunkStore(b, 24*30, encoding.e)
+
+		for _, q := range queryables {
+			b.Run(fmt.Sprintf("%s/%s", q.name, encoding.name), func(b *testing.B) {
+				queryable := q.f(store)
+				var r *promql.Result
+				for n := 0; n < b.N; n++ {
+					r = testQuery(b, queryable, from)
+				}
+				result = r
+			})
+		}
+	}
+}

pkg/querier/chunk_iterator.go

@@ -0,0 +1,59 @@
+package querier
+
+import (
+	"github.com/prometheus/common/model"
+	"github.com/weaveworks/cortex/pkg/chunk"
+	promchunk "github.com/weaveworks/cortex/pkg/prom1/storage/local/chunk"
+)
+
+type chunkIterator struct {
+	chunk.Chunk
+	it promchunk.Iterator
+
+	// At() is called often in the heap code, so caching its result seems like
+	// a good idea.
+	cacheValid  bool
+	cachedTime  int64
+	cachedValue float64
+}
+
+// Seek advances the iterator forward to the value at or after
+// the given timestamp.
+func (i *chunkIterator) Seek(t int64) bool {
+	i.cacheValid = false
+
+	// We assume seeks only care about a specific window; if this chunk doesn't
+	// contain samples in that window, we can shortcut.
+	if int64(i.Through) < t {
+		return false
+	}
+
+	return i.it.FindAtOrAfter(model.Time(t))
+}
+
+func (i *chunkIterator) AtTime() int64 {
+	if !i.cacheValid {
+		v := i.it.Value()
+		i.cachedTime, i.cachedValue = int64(v.Timestamp), float64(v.Value)
+		i.cacheValid = true
+	}
+	return i.cachedTime
+}
+
+func (i *chunkIterator) At() (int64, float64) {
+	if !i.cacheValid {
+		v := i.it.Value()
+		i.cachedTime, i.cachedValue = int64(v.Timestamp), float64(v.Value)
+		i.cacheValid = true
+	}
+	return i.cachedTime, i.cachedValue
+}
+
+func (i *chunkIterator) Next() bool {
+	i.cacheValid = false
+	return i.it.Scan()
+}
+
+func (i *chunkIterator) Err() error {
+	return i.it.Err()
+}