Fixes an edge case in the batch chunk iterator. (#4550)

* no-cache

Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com>

* found a bug to fix.

Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com>

* Fixes a edge case in the batch chunk iterator.

Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com>

pkg/storage/batch.go

@@ -235,11 +235,20 @@ func (it *batchChunkIterator) nextBatch() (res *chunkBatch) {
 				through = through.Add(time.Nanosecond)
 			}
 		} else {
+
 			from = time.Unix(0, headChunk.Chunk.From.UnixNano())
 
-			// when clipping the from it should never be before the start or equal to the end.
+			// if the start of the batch is equal to the end of the query, since the end is not inclusive we can discard that batch.
+			if from.Equal(it.end) {
+				if it.end != it.start {
+					return nil
+				}
+				// unless end and start are equal in which case start and end are both inclusive.
+				from = it.end
+			}
+			// when clipping the from it should never be before the start.
 			// Doing so would include entries not requested.
-			if from.Before(it.start) || from.Equal(it.end) {
+			if from.Before(it.start) {
 				from = it.start
 			}
 		}

pkg/storage/batch_test.go

@@ -1162,6 +1162,121 @@ func Test_newSampleBatchChunkIterator(t *testing.T) {
 			from, from.Add(3 * time.Millisecond),
 			2,
 		},
+		"forward last chunk boundaries equal to end": {
+			[]*LazyChunk{
+				newLazyChunk(logproto.Stream{
+					Labels: fooLabelsWithName,
+					Entries: []logproto.Entry{
+						{
+							Timestamp: time.Unix(1, 0),
+							Line:      "1",
+						},
+						{
+							Timestamp: time.Unix(2, 0),
+							Line:      "2",
+						},
+					},
+				}),
+				newLazyChunk(logproto.Stream{
+					Labels: fooLabelsWithName,
+					Entries: []logproto.Entry{
+
+						{
+							Timestamp: time.Unix(2, 0),
+							Line:      "2",
+						},
+						{
+							Timestamp: time.Unix(3, 0),
+							Line:      "3",
+						},
+					},
+				}),
+				newLazyChunk(logproto.Stream{
+					Labels: fooLabelsWithName,
+					Entries: []logproto.Entry{
+						{
+							Timestamp: time.Unix(3, 0),
+							Line:      "3",
+						},
+						{
+							Timestamp: time.Unix(4, 0),
+							Line:      "4",
+						},
+					},
+				}),
+			},
+			[]logproto.Series{
+				{
+					Labels: fooLabels,
+					Samples: []logproto.Sample{
+						{
+							Timestamp: time.Unix(1, 0).UnixNano(),
+							Hash:      xxhash.Sum64String("1"),
+							Value:     1.,
+						},
+						{
+							Timestamp: time.Unix(2, 0).UnixNano(),
+							Hash:      xxhash.Sum64String("2"),
+							Value:     1.,
+						},
+					},
+				},
+			},
+			fooLabelsWithName,
+			time.Unix(1, 0), time.Unix(3, 0),
+			2,
+		},
+		"forward last chunk boundaries equal to end and start": {
+			[]*LazyChunk{
+				newLazyChunk(logproto.Stream{
+					Labels: fooLabelsWithName,
+					Entries: []logproto.Entry{
+						{
+							Timestamp: time.Unix(1, 0),
+							Line:      "1",
+						},
+						{
+							Timestamp: time.Unix(1, 0),
+							Line:      "2",
+						},
+					},
+				}),
+				newLazyChunk(logproto.Stream{
+					Labels: fooLabelsWithName,
+					Entries: []logproto.Entry{
+
+						{
+							Timestamp: time.Unix(1, 0),
+							Line:      "2",
+						},
+						{
+							Timestamp: time.Unix(2, 0),
+							Line:      "3",
+						},
+					},
+				}),
+			},
+			[]logproto.Series{
+				{
+					Labels: fooLabels,
+					Samples: []logproto.Sample{
+						{
+							Timestamp: time.Unix(1, 0).UnixNano(),
+							Hash:      xxhash.Sum64String("1"),
+							Value:     1.,
+						},
+						{
+							Timestamp: time.Unix(1, 0).UnixNano(),
+							Hash:      xxhash.Sum64String("2"),
+							Value:     1.,
+						},
+					},
+				},
+			},
+			fooLabelsWithName,
+			time.Unix(1, 0), time.Unix(1, 0),
+			2,
+		},
 		"forward without overlap": {
 			[]*LazyChunk{
 				newLazyChunk(logproto.Stream{