Merge pull request #370 from parca-dev/storage-iterator-ranges

pkg/storage: Fix iterators to only get index range within chunks
parca-dev · Oct 25, 2021 · a6867a7 · a6867a7
2 parents 3ab314d + 3735c66
commit a6867a7
Show file tree

Hide file tree

Showing 6 changed files with 27 additions and 7 deletions.
diff --git a/pkg/query/query.go b/pkg/query/query.go
@@ -111,6 +111,9 @@ func (q *Query) QueryRange(ctx context.Context, req *pb.QueryRangeRequest) (*pb.
 		it := series.Iterator()
 		for it.Next() {
 			p := it.At()
+			if p.ProfileMeta().Timestamp == 0 {
+				return nil, status.Error(codes.Internal, "profile's timestamp is 0")
+			}
 			metricsSeries.Samples = append(metricsSeries.Samples, &pb.MetricsSample{
 				Timestamp: timestamppb.New(timestamp.Time(p.ProfileMeta().Timestamp)),
 				Value:     p.ProfileTree().RootCumulativeValue(),

diff --git a/pkg/storage/series_iterator.go b/pkg/storage/series_iterator.go
@@ -289,17 +289,21 @@ func (n *MemSeriesIteratorTreeNode) FlatValues() []*ProfileTreeValueNode {
 	return res
 }
 
-func getIndexRange(it MemSeriesValuesIterator, numSamples uint16, mint, maxt int64) (uint64, uint64, error) {
+func getIndexRange(it MemSeriesValuesIterator, numSamples uint64, mint, maxt int64) (uint64, uint64, error) {
 	// figure out the index of the first sample > mint and the last sample < maxt
 	start := uint64(0)
 	end := uint64(0)
-	i := uint16(0)
+	i := uint64(0)
 	for it.Next() {
 		if i == numSamples {
 			end++
 			break
 		}
 		t := it.At()
+		// MultiChunkIterator might return sparse values - shouldn't usually happen though.
+		if t == 0 {
+			break
+		}
 		if t < mint {
 			start++
 		}

diff --git a/pkg/storage/series_iterator_merge.go b/pkg/storage/series_iterator_merge.go
@@ -44,15 +44,18 @@ func (ms *MemMergeSeries) Iterator() ProfileSeriesIterator {
 		maxt = ms.s.maxTime
 	}
 
+	var numSamples uint64
+
 	chunkStart, chunkEnd := ms.s.timestamps.indexRange(mint, maxt)
 	timestamps := make([]chunkenc.Chunk, 0, chunkEnd-chunkStart)
 	for _, t := range ms.s.timestamps[chunkStart:chunkEnd] {
+		numSamples += uint64(t.chunk.NumSamples())
 		timestamps = append(timestamps, t.chunk)
 	}
 
 	sl := &SliceProfileSeriesIterator{i: -1}
 
-	start, end, err := getIndexRange(NewMultiChunkIterator(timestamps), ms.s.numSamples, mint, maxt)
+	start, end, err := getIndexRange(NewMultiChunkIterator(timestamps), numSamples, mint, maxt)
 	if err != nil {
 		sl.err = err
 		return sl

diff --git a/pkg/storage/series_iterator_range.go b/pkg/storage/series_iterator_range.go
@@ -37,14 +37,17 @@ func (rs *MemRangeSeries) Iterator() ProfileSeriesIterator {
 	rs.s.mu.RLock()
 	defer rs.s.mu.RUnlock()
 
+	var numSamples uint64
+
 	chunkStart, chunkEnd := rs.s.timestamps.indexRange(rs.mint, rs.maxt)
 	timestamps := make([]chunkenc.Chunk, 0, chunkEnd-chunkStart)
 	for _, t := range rs.s.timestamps[chunkStart:chunkEnd] {
+		numSamples += uint64(t.chunk.NumSamples())
 		timestamps = append(timestamps, t.chunk)
 	}
 
 	it := NewMultiChunkIterator(timestamps)
-	start, end, err := getIndexRange(it, rs.s.numSamples, rs.mint, rs.maxt)
+	start, end, err := getIndexRange(it, numSamples, rs.mint, rs.maxt)
 	if err != nil {
 		return &MemRangeSeriesIterator{err: err}
 	}
@@ -111,7 +114,6 @@ func (rs *MemRangeSeries) Iterator() ProfileSeriesIterator {
 		periodsIterator.Seek(start)
 	}
 
-	numSamples := uint64(rs.s.numSamples)
 	if end-start < numSamples {
 		numSamples = end - start - 1
 	}

diff --git a/pkg/storage/series_iterator_root.go b/pkg/storage/series_iterator_root.go
@@ -35,14 +35,17 @@ func (rs *MemRootSeries) Iterator() ProfileSeriesIterator {
 	rs.s.mu.RLock()
 	defer rs.s.mu.RUnlock()
 
+	var numSamples uint64
+
 	chunkStart, chunkEnd := rs.s.timestamps.indexRange(rs.mint, rs.maxt)
 	timestamps := make([]chunkenc.Chunk, 0, chunkEnd-chunkStart)
 	for _, t := range rs.s.timestamps[chunkStart:chunkEnd] {
+		numSamples += uint64(t.chunk.NumSamples())
 		timestamps = append(timestamps, t.chunk)
 	}
 
 	it := NewMultiChunkIterator(timestamps)
-	start, end, err := getIndexRange(it, rs.s.numSamples, rs.mint, rs.maxt)
+	start, end, err := getIndexRange(it, numSamples, rs.mint, rs.maxt)
 	if start == end {
 		return &MemRootSeriesIterator{err: fmt.Errorf("no samples within the time range")}
 	}
@@ -58,7 +61,7 @@ func (rs *MemRootSeries) Iterator() ProfileSeriesIterator {
 		rootIterator.Seek(start)
 	}
 
-	numSamples := uint64(rs.s.numSamples)
+	// Set numSamples correctly if only subset selected.
 	if end-start < numSamples {
 		numSamples = end - start - 1
 	}

diff --git a/pkg/storage/series_iterator_test.go b/pkg/storage/series_iterator_test.go
@@ -269,6 +269,11 @@ func TestGetIndexRange(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, uint64(2), start)
 	require.Equal(t, uint64(4), end)
+
+	start, end, err = getIndexRange(NewMultiChunkIterator([]chunkenc.Chunk{c}), 123, 1, 12)
+	require.NoError(t, err)
+	require.Equal(t, uint64(0), start)
+	require.Equal(t, uint64(5), end)
 }
 
 func TestIteratorRangeSum(t *testing.T) {