Function TestIngesterStreaming

pkg/querier/distributor_queryable_test.go:146–211

(t *testing.T)

}

func TestIngesterStreaming(t *testing.T) {
	t.Parallel()

	now := time.Now()

	for _, enc := range encodings {
		for _, partialDataEnabled := range []bool{false, true} {
			promChunk := util.GenerateChunk(t, time.Second, model.TimeFromUnix(now.Unix()), 10, enc)
			clientChunks, err := chunkcompat.ToChunks([]chunk.Chunk{promChunk})
			require.NoError(t, err)

			d := &MockDistributor{}
			queryResponse := &client.QueryStreamResponse{
				Chunkseries: []client.TimeSeriesChunk{
					{
						Labels: []cortexpb.LabelAdapter{
							{Name: "bar", Value: "baz"},
						},
						Chunks: clientChunks,
					},
					{
						Labels: []cortexpb.LabelAdapter{
							{Name: "foo", Value: "bar"},
						},
						Chunks: clientChunks,
					},
				},
			}
			var partialDataErr error
			if partialDataEnabled {
				partialDataErr = partialdata.ErrPartialData
			}
			d.On("QueryStream", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(queryResponse, partialDataErr)

			ctx := user.InjectOrgID(context.Background(), "0")

			queryable := newDistributorQueryable(d, true, true, batch.NewChunkMergeIterator, 0, func(string) bool {
				return partialDataEnabled
			}, 1)
			querier, err := queryable.Querier(mint, maxt)
			require.NoError(t, err)

			seriesSet := querier.Select(ctx, true, &storage.SelectHints{Start: mint, End: maxt})
			require.NoError(t, seriesSet.Err())

			require.True(t, seriesSet.Next())
			series := seriesSet.At()
			require.Equal(t, labels.FromStrings("bar", "baz"), series.Labels())
			chkIter := series.Iterator(nil)
			require.Equal(t, enc.ChunkValueType(), chkIter.Next())

			require.True(t, seriesSet.Next())
			series = seriesSet.At()
			require.Equal(t, labels.FromStrings("foo", "bar"), series.Labels())
			chkIter = series.Iterator(chkIter)
			require.Equal(t, enc.ChunkValueType(), chkIter.Next())

			require.False(t, seriesSet.Next())