Fixes a bug in MatrixStepper when sharding queries.

pkg/logcli/query/query.go

@@ -58,7 +58,6 @@ type Query struct {
 
 // DoQuery executes the query and prints out the results
 func (q *Query) DoQuery(c client.Client, out output.LogOutput, statistics bool) {
-
 	if q.LocalConfig != "" {
 		if err := q.DoLocalQuery(out, statistics, c.GetOrgID()); err != nil {
 			log.Fatalf("Query failed: %+v", err)
@@ -149,7 +148,6 @@ func (q *Query) DoQuery(c client.Client, out output.LogOutput, statistics bool)
 
 		}
 	}
-
 }
 
 func (q *Query) printResult(value loghttp.ResultValue, out output.LogOutput, lastEntry []*loghttp.Entry) (int, []*loghttp.Entry) {
@@ -172,7 +170,6 @@ func (q *Query) printResult(value loghttp.ResultValue, out output.LogOutput, las
 
 // DoLocalQuery executes the query against the local store using a Loki configuration file.
 func (q *Query) DoLocalQuery(out output.LogOutput, statistics bool, orgID string) error {
-
 	var conf loki.Config
 	conf.RegisterFlags(flag.CommandLine)
 	if q.LocalConfig == "" {
@@ -255,7 +252,7 @@ func (q *Query) SetInstant(time time.Time) {
 }
 
 func (q *Query) isInstant() bool {
-	return q.Start == q.End
+	return q.Start == q.End && q.Step == 0
 }
 
 func (q *Query) printStream(streams loghttp.Streams, out output.LogOutput, lastEntry []*loghttp.Entry) (int, []*loghttp.Entry) {
@@ -369,7 +366,6 @@ func (q *Query) printMatrix(matrix loghttp.Matrix) {
 	// it gives us more flexibility with regard to output types in the future.  initially we are supporting just formatted json but eventually
 	// we might add output options such as render to an image file on disk
 	bytes, err := json.MarshalIndent(matrix, "", "  ")
-
 	if err != nil {
 		log.Fatalf("Error marshalling matrix: %v", err)
 	}
@@ -379,7 +375,6 @@ func (q *Query) printMatrix(matrix loghttp.Matrix) {
 
 func (q *Query) printVector(vector loghttp.Vector) {
 	bytes, err := json.MarshalIndent(vector, "", "  ")
-
 	if err != nil {
 		log.Fatalf("Error marshalling vector: %v", err)
 	}
@@ -389,7 +384,6 @@ func (q *Query) printVector(vector loghttp.Vector) {
 
 func (q *Query) printScalar(scalar loghttp.Scalar) {
 	bytes, err := json.MarshalIndent(scalar, "", "  ")
-
 	if err != nil {
 		log.Fatalf("Error marshalling scalar: %v", err)
 	}

pkg/logql/matrix.go

@@ -33,7 +33,7 @@ func NewMatrixStepper(start, end time.Time, step time.Duration, m promql.Matrix)
 
 func (m *MatrixStepper) Next() (bool, int64, promql.Vector) {
 	m.ts = m.ts.Add(m.step)
-	if !m.ts.Before(m.end) {
+	if m.ts.After(m.end) {
 		return false, 0, nil
 	}
 

pkg/logql/matrix_test.go

@@ -20,7 +20,7 @@ func TestMatrixStepper(t *testing.T) {
 		promql.Series{
 			Metric: labels.Labels{{Name: "foo", Value: "bar"}},
 			Points: []promql.Point{
-				{T: start.UnixNano() / int64(step), V: 0},
+				{T: start.UnixNano(), V: 0},
 				{T: start.Add(step).UnixNano() / int64(time.Millisecond), V: 1},
 				{T: start.Add(2*step).UnixNano() / int64(time.Millisecond), V: 2},
 				{T: start.Add(3*step).UnixNano() / int64(time.Millisecond), V: 3},
@@ -42,11 +42,11 @@ func TestMatrixStepper(t *testing.T) {
 	expected := []promql.Vector{
 		{
 			promql.Sample{
-				Point:  promql.Point{T: start.UnixNano() / int64(step), V: 0},
+				Point:  promql.Point{T: start.UnixNano(), V: 0},
 				Metric: labels.Labels{{Name: "foo", Value: "bar"}},
 			},
 			promql.Sample{
-				Point:  promql.Point{T: start.UnixNano() / int64(step), V: 0},
+				Point:  promql.Point{T: start.UnixNano(), V: 0},
 				Metric: labels.Labels{{Name: "bazz", Value: "buzz"}},
 			},
 		},
@@ -100,9 +100,19 @@ func TestMatrixStepper(t *testing.T) {
 				Metric: labels.Labels{{Name: "bazz", Value: "buzz"}},
 			},
 		},
+		{
+			promql.Sample{
+				Point:  promql.Point{T: start.Add(6*step).UnixNano() / int64(time.Millisecond), V: 0},
+				Metric: labels.Labels{{Name: "foo", Value: "bar"}},
+			},
+			promql.Sample{
+				Point:  promql.Point{T: start.Add(6*step).UnixNano() / int64(time.Millisecond), V: 0},
+				Metric: labels.Labels{{Name: "bazz", Value: "buzz"}},
+			},
+		},
 	}
 
-	for i := 0; i < int(end.Sub(start)/step); i++ {
+	for i := 0; i <= int(end.Sub(start)/step); i++ {
 		ok, ts, vec := s.Next()
 		require.Equal(t, ok, true)
 		require.Equal(t, start.Add(step*time.Duration(i)).UnixNano()/int64(time.Millisecond), ts)
@@ -113,3 +123,33 @@ func TestMatrixStepper(t *testing.T) {
 
 	require.Equal(t, ok, false)
 }
+
+func Test_SingleStepMatrix(t *testing.T) {
+	var (
+		start = time.Unix(0, 0)
+		end   = time.Unix(0, 0)
+		step  = time.Second
+	)
+
+	m := promql.Matrix{
+		promql.Series{
+			Metric: labels.Labels{},
+			Points: []promql.Point{
+				{T: start.UnixNano(), V: 10},
+			},
+		},
+	}
+
+	s := NewMatrixStepper(start, end, step, m)
+
+	ok, ts, vec := s.Next()
+	require.True(t, ok)
+	require.Equal(t, start.UnixNano(), ts)
+	require.Equal(t, promql.Vector{promql.Sample{
+		Point:  promql.Point{T: start.UnixNano(), V: 10},
+		Metric: labels.Labels{},
+	}}, vec)
+
+	ok, _, _ = s.Next()
+	require.False(t, ok)
+}

pkg/logql/test_utils.go

@@ -201,7 +201,7 @@ outer:
 	return iter.NewTimeRangedSampleIterator(
 		iter.NewMultiSeriesIterator(ctx, filtered),
 		req.Start.UnixNano(),
-		req.End.UnixNano(),
+		req.End.UnixNano()+1,
 	), nil
 }
 
@@ -232,7 +232,6 @@ func (m MockDownstreamer) Downstream(ctx context.Context, queries []DownstreamQu
 		results = append(results, res)
 	}
 	return results, nil
-
 }
 
 // create nStreams of nEntries with labelNames each where each label value
@@ -256,7 +255,7 @@ func randomStreams(nStreams, nEntries, nShards int, labelNames []string) (stream
 				Value: fmt.Sprintf("%d", shard),
 			})
 		}
-		for j := 0; j < nEntries; j++ {
+		for j := 0; j <= nEntries; j++ {
 			stream.Entries = append(stream.Entries, logproto.Entry{
 				Timestamp: time.Unix(0, int64(j*int(time.Second))),
 				Line:      fmt.Sprintf("line number: %d", j),
@@ -267,7 +266,6 @@ func randomStreams(nStreams, nEntries, nShards int, labelNames []string) (stream
 		streams = append(streams, stream)
 	}
 	return streams
-
 }
 
 func mustParseLabels(s string) labels.Labels {