Function newMergeIterator

pkg/querier/batch/merge.go:26–62

(it iterator, cs []GenericChunk)

}

func newMergeIterator(it iterator, cs []GenericChunk) *mergeIterator {
	css := partitionChunks(cs)

	var c *mergeIterator

	if mIterator, ok := it.(*mergeIterator); ok && cap(mIterator.its) >= len(css) {
		c = mIterator.Reset(len(css))
	} else {
		c = &mergeIterator{
			h:          make(iteratorHeap, 0, len(css)),
			batches:    make(batchStream, 0, len(css)),
			batchesBuf: make(batchStream, len(css)),
		}
	}

	if cap(c.its) < len(css) {
		c.its = make([]*nonOverlappingIterator, 0, len(css))
	}

	for _, cs := range css {
		c.its = append(c.its, newNonOverlappingIterator(cs))
	}

	for _, iter := range c.its {
		if iter.Next(1) != chunkenc.ValNone {
			c.h = append(c.h, iter)
			continue
		}

		if err := iter.Err(); err != nil {
			c.currErr = err
		}
	}

	heap.Init(&c.h)
	return c
}

func (c *mergeIterator) Reset(size int) *mergeIterator {
	c.its = c.its[:0]