querier: Added regressions tests for counter missed bug.

PR with just tests, not fix yet.

Reproduces: #2401

* Added regressions tests for CounterSeriesIterator; Simplified aggregators.
* Fixes edge dedup cases for Next and added tests for deduplication.
* Refactored downsampling tests, added more realistic cases.
* Added check for duplicated chunks during downsampling.

Signed-off-by: Bartlomiej Plotka <[email protected]>

pkg/compact/downsample/downsample.go

@@ -116,6 +116,13 @@ func Downsample(
 		if err := indexr.Series(postings.At(), &lset, &chks); err != nil {
 			return id, errors.Wrapf(err, "get series %d", postings.At())
 		}
+
+		for i, c := range chks[1:] {
+			if chks[i].MaxTime >= c.MinTime {
+				return id, errors.Errorf("found overlapping chunks within series %d. Chunks expected to be ordered by min time and non-overlapping, got: %v", postings.At(), chks)
+			}
+		}
+
 		// While #183 exists, we sanitize the chunks we retrieved from the block
 		// before retrieving their samples.
 		for i, c := range chks {
@@ -129,6 +136,9 @@ func Downsample(
 		// Raw and already downsampled data need different processing.
 		if origMeta.Thanos.Downsample.Resolution == 0 {
 			for _, c := range chks {
+				// TODO(bwplotka): We can optimze this further by using in WriteSeries iterators of each chunk instead of
+				// samples. Also ensure 120 sample limit, otherwise we have gigantic chunks.
+				// https://github.com/thanos-io/thanos/issues/2542
 				if err := expandChunkIterator(c.Chunk.Iterator(reuseIt), &all); err != nil {
 					return id, errors.Wrapf(err, "expand chunk %d, series %d", c.Ref, postings.At())
 				}
@@ -139,7 +149,11 @@ func Downsample(
 		} else {
 			// Downsample a block that contains aggregated chunks already.
 			for _, c := range chks {
-				aggrChunks = append(aggrChunks, c.Chunk.(*AggrChunk))
+				ac, ok := c.Chunk.(*AggrChunk)
+				if !ok {
+					return id, errors.Errorf("expected downsampled chunk (*downsample.AggrChunk) got %T instead for series: %d", c.Chunk, postings.At())
+				}
+				aggrChunks = append(aggrChunks, ac)
 			}
 			downsampledChunks, err := downsampleAggr(
 				aggrChunks,
@@ -334,12 +348,12 @@ func downsampleRawLoop(data []sample, resolution int64, numChunks int) []chunks.
 
 		ab := newAggrChunkBuilder()
 
-		// Encode first raw value; see CounterSeriesIterator.
+		// Encode first raw value; see CounterAggrChunksIterator.
 		ab.apps[AggrCounter].Append(batch[0].t, batch[0].v)
 
 		lastT := downsampleBatch(batch, resolution, ab.add)
 
-		// Encode last raw value; see CounterSeriesIterator.
+		// Encode last raw value; see CounterAggrChunksIterator.
 		ab.apps[AggrCounter].Append(lastT, batch[len(batch)-1].v)
 
 		chks = append(chks, ab.encode())
@@ -524,7 +538,7 @@ func downsampleAggrBatch(chks []*AggrChunk, buf *[]sample, resolution int64) (ch
 	ab.chunks[AggrCounter] = chunkenc.NewXORChunk()
 	ab.apps[AggrCounter], _ = ab.chunks[AggrCounter].Appender()
 
-	// Retain first raw value; see CounterSeriesIterator.
+	// Retain first raw value; see CounterAggrChunksIterator.
 	ab.apps[AggrCounter].Append((*buf)[0].t, (*buf)[0].v)
 
 	lastT := downsampleBatch(*buf, resolution, func(t int64, a *aggregator) {
@@ -536,7 +550,7 @@ func downsampleAggrBatch(chks []*AggrChunk, buf *[]sample, resolution int64) (ch
 		ab.apps[AggrCounter].Append(t, a.counter)
 	})
 
-	// Retain last raw value; see CounterSeriesIterator.
+	// Retain last raw value; see CounterAggrChunksIterator.
 	ab.apps[AggrCounter].Append(lastT, it.lastV)
 
 	ab.mint = mint
@@ -551,16 +565,21 @@ type sample struct {
 
 // CounterSeriesIterator generates monotonically increasing values by iterating
 // over an ordered sequence of chunks, which should be raw or aggregated chunks
-// of counter values.  The generated samples can be used by PromQL functions
-// like 'rate' that calculate differences between counter values.
+// of counter values. The generated samples can be used by PromQL functions
+// like 'rate' that calculate differences between counter values. Stale Markers
+// are removed as well.
 //
 // Counter aggregation chunks must have the first and last values from their
 // original raw series: the first raw value should be the first value encoded
 // in the chunk, and the last raw value is encoded by the duplication of the
-// previous sample's timestamp.  As iteration occurs between chunks, the
+// previous sample's timestamp. As iteration occurs between chunks, the
 // comparison between the last raw value of the earlier chunk and the first raw
 // value of the later chunk ensures that counter resets between chunks are
 // recognized and that the correct value delta is calculated.
+//
+// It handles overlapped chunks (removes overlaps.
+// NOTE: It is important to deduplicate with care ensuring that you don't hit
+// issue https://github.com/thanos-io/thanos/issues/2401#issuecomment-621958839.
 type CounterSeriesIterator struct {
 	chks   []chunkenc.Iterator
 	i      int     // Current chunk.