Query storage by iterating through chunks by batches.

pkg/chunkenc/lazy_chunk.go

@@ -2,6 +2,7 @@ package chunkenc
 
 import (
 	"context"
+	"errors"
 	"time"
 
 	"github.com/cortexproject/cortex/pkg/chunk"
@@ -16,14 +17,6 @@ type LazyChunk struct {
 	Fetcher *chunk.Fetcher
 }
 
-func (c *LazyChunk) getChunk(ctx context.Context) (Chunk, error) {
-	chunks, err := c.Fetcher.FetchChunks(ctx, []chunk.Chunk{c.Chunk}, []string{c.Chunk.ExternalKey()})
-	if err != nil {
-		return nil, err
-	}
-	return chunks[0].Data.(*Facade).LokiChunk(), nil
-}
-
 // Iterator returns an entry iterator.
 func (c *LazyChunk) Iterator(ctx context.Context, from, through time.Time, direction logproto.Direction, filter logql.Filter) (iter.EntryIterator, error) {
 	// If the chunk is already loaded, then use that.
@@ -32,77 +25,5 @@ func (c *LazyChunk) Iterator(ctx context.Context, from, through time.Time, direc
 		return lokiChunk.Iterator(from, through, direction, filter)
 	}
 
-	return &lazyIterator{
-		chunk:  c,
-		filter: filter,
-
-		from:      from,
-		through:   through,
-		direction: direction,
-		context:   ctx,
-	}, nil
-}
-
-type lazyIterator struct {
-	iter.EntryIterator
-
-	chunk *LazyChunk
-	err   error
-
-	from, through time.Time
-	direction     logproto.Direction
-	context       context.Context
-	filter        logql.Filter
-
-	closed bool
-}
-
-func (it *lazyIterator) Next() bool {
-	if it.err != nil {
-		return false
-	}
-
-	if it.closed {
-		return false
-	}
-
-	if it.EntryIterator != nil {
-		next := it.EntryIterator.Next()
-		if !next {
-			it.Close()
-		}
-		return next
-	}
-
-	chk, err := it.chunk.getChunk(it.context)
-	if err != nil {
-		it.err = err
-		return false
-	}
-	it.EntryIterator, it.err = chk.Iterator(it.from, it.through, it.direction, it.filter)
-	return it.Next()
-}
-
-func (it *lazyIterator) Labels() string {
-	return it.chunk.Chunk.Metric.String()
-}
-
-func (it *lazyIterator) Error() error {
-	if it.err != nil {
-		return it.err
-	}
-	if it.EntryIterator != nil {
-		return it.EntryIterator.Error()
-	}
-	return nil
-}
-
-func (it *lazyIterator) Close() error {
-	if it.EntryIterator != nil {
-		it.closed = true
-		err := it.EntryIterator.Close()
-		it.EntryIterator = nil
-		return err
-	}
-	return nil
+	return nil, errors.New("chunk is not loaded")
 }

pkg/ingester/instance.go

@@ -18,7 +18,6 @@ import (
 	"github.com/grafana/loki/pkg/iter"
 	"github.com/grafana/loki/pkg/logproto"
 	"github.com/grafana/loki/pkg/logql"
-	"github.com/grafana/loki/pkg/querier"
 	"github.com/grafana/loki/pkg/util"
 )
 
@@ -257,7 +256,7 @@ func isDone(ctx context.Context) bool {
 func sendBatches(i iter.EntryIterator, queryServer logproto.Querier_QueryServer, limit uint32) error {
 	sent := uint32(0)
 	for sent < limit && !isDone(queryServer.Context()) {
-		batch, batchSize, err := querier.ReadBatch(i, helpers.MinUint32(queryBatchSize, limit-sent))
+		batch, batchSize, err := iter.ReadBatch(i, helpers.MinUint32(queryBatchSize, limit-sent))
 		if err != nil {
 			return err
 		}

pkg/iter/iterator.go

@@ -466,3 +466,28 @@ func (i *entryIteratorBackward) Error() error { return nil }
 func (i *entryIteratorBackward) Labels() string {
 	return ""
 }
+
+// ReadBatch reads a set of entries off an iterator.
+func ReadBatch(i EntryIterator, size uint32) (*logproto.QueryResponse, uint32, error) {
+	streams := map[string]*logproto.Stream{}
+	respSize := uint32(0)
+	for ; respSize < size && i.Next(); respSize++ {
+		labels, entry := i.Labels(), i.Entry()
+		stream, ok := streams[labels]
+		if !ok {
+			stream = &logproto.Stream{
+				Labels: labels,
+			}
+			streams[labels] = stream
+		}
+		stream.Entries = append(stream.Entries, entry)
+	}
+
+	result := logproto.QueryResponse{
+		Streams: make([]*logproto.Stream, 0, len(streams)),
+	}
+	for _, stream := range streams {
+		result.Streams = append(result.Streams, stream)
+	}
+	return &result, respSize, i.Error()
+}

pkg/loki/loki.go

@@ -9,7 +9,6 @@ import (
 	"google.golang.org/grpc"
 
 	"github.com/cortexproject/cortex/pkg/chunk"
-	"github.com/cortexproject/cortex/pkg/chunk/storage"
 	"github.com/cortexproject/cortex/pkg/ring"
 	"github.com/cortexproject/cortex/pkg/util"
 	"github.com/cortexproject/cortex/pkg/util/validation"
@@ -20,7 +19,7 @@ import (
 	"github.com/grafana/loki/pkg/ingester"
 	"github.com/grafana/loki/pkg/ingester/client"
 	"github.com/grafana/loki/pkg/querier"
-	loki_storage "github.com/grafana/loki/pkg/storage"
+	"github.com/grafana/loki/pkg/storage"
 )
 
 // Config is the root config for Loki.
@@ -70,7 +69,7 @@ type Loki struct {
 	distributor  *distributor.Distributor
 	ingester     *ingester.Ingester
 	querier      *querier.Querier
-	store        loki_storage.Store
+	store        storage.Store
 	tableManager *chunk.TableManager
 
 	httpAuthMiddleware middleware.Interface

pkg/loki/modules.go

@@ -197,12 +197,12 @@ func (t *Loki) initTableManager() error {
 		os.Exit(1)
 	}
 
-	tableClient, err := storage.NewTableClient(lastConfig.IndexType, t.cfg.StorageConfig)
+	tableClient, err := storage.NewTableClient(lastConfig.IndexType, t.cfg.StorageConfig.Config)
 	if err != nil {
 		return err
 	}
 
-	bucketClient, err := storage.NewBucketClient(t.cfg.StorageConfig)
+	bucketClient, err := storage.NewBucketClient(t.cfg.StorageConfig.Config)
 	util.CheckFatal("initializing bucket client", err)
 
 	t.tableManager, err = chunk.NewTableManager(t.cfg.TableManager, t.cfg.SchemaConfig, maxChunkAgeForTableManager, tableClient, bucketClient)

pkg/querier/querier.go

@@ -129,7 +129,7 @@ func (q *Querier) Query(ctx context.Context, req *logproto.QueryRequest) (*logpr
 	iterator := iter.NewHeapIterator(iterators, req.Direction)
 	defer helpers.LogError("closing iterator", iterator.Close)
 
-	resp, _, err := ReadBatch(iterator, req.Limit)
+	resp, _, err := iter.ReadBatch(iterator, req.Limit)
 	return resp, err
 }
 
@@ -201,31 +201,6 @@ func (q *Querier) Label(ctx context.Context, req *logproto.LabelRequest) (*logpr
 	}, nil
 }
 
-// ReadBatch reads a set of entries off an iterator.
-func ReadBatch(i iter.EntryIterator, size uint32) (*logproto.QueryResponse, uint32, error) {
-	streams := map[string]*logproto.Stream{}
-	respSize := uint32(0)
-	for ; respSize < size && i.Next(); respSize++ {
-		labels, entry := i.Labels(), i.Entry()
-		stream, ok := streams[labels]
-		if !ok {
-			stream = &logproto.Stream{
-				Labels: labels,
-			}
-			streams[labels] = stream
-		}
-		stream.Entries = append(stream.Entries, entry)
-	}
-
-	result := logproto.QueryResponse{
-		Streams: make([]*logproto.Stream, 0, len(streams)),
-	}
-	for _, stream := range streams {
-		result.Streams = append(result.Streams, stream)
-	}
-	return &result, respSize, i.Error()
-}
-
 // Check implements the grpc healthcheck
 func (*Querier) Check(_ context.Context, _ *grpc_health_v1.HealthCheckRequest) (*grpc_health_v1.HealthCheckResponse, error) {
 	return &grpc_health_v1.HealthCheckResponse{Status: grpc_health_v1.HealthCheckResponse_SERVING}, nil

pkg/storage/hack/main.go

@@ -40,9 +40,11 @@ func main() {
 
 func getStore() (lstore.Store, error) {
 	store, err := lstore.NewStore(
-		storage.Config{
-			BoltDBConfig: local.BoltDBConfig{Directory: "/tmp/benchmark/index"},
-			FSConfig:     local.FSConfig{Directory: "/tmp/benchmark/chunks"},
+		lstore.Config{
+			Config: storage.Config{
+				BoltDBConfig: local.BoltDBConfig{Directory: "/tmp/benchmark/index"},
+				FSConfig:     local.FSConfig{Directory: "/tmp/benchmark/chunks"},
+			},
 		},
 		chunk.StoreConfig{},
 		chunk.SchemaConfig{