func()

in v1storage/series.go [435:502]

• 60 lines of code
• 17 McCabe index (conditional complexity)

func (s *memorySeries) preloadChunksForRange(
	fp model.Fingerprint,
	from model.Time, through model.Time,
	mss *MemorySeriesStorage,
) (SeriesIterator, error) {
	firstChunkDescTime := model.Latest
	if len(s.chunkDescs) > 0 {
		firstChunkDescTime = s.chunkDescs[0].FirstTime()
	}
	if s.chunkDescsOffset != 0 && from.Before(firstChunkDescTime) {
		cds, err := mss.loadChunkDescs(fp, s.persistWatermark)
		if err != nil {
			return nopIter, err
		}
		if s.chunkDescsOffset != -1 && len(cds) != s.chunkDescsOffset {
			return nopIter, fmt.Errorf(
				"unexpected number of chunk descs loaded for fingerprint %v: expected %d, got %d",
				fp, s.chunkDescsOffset, len(cds),
			)
		}
		s.persistWatermark += len(cds)
		s.chunkDescs = append(cds, s.chunkDescs...)
		s.chunkDescsOffset = 0
		if len(s.chunkDescs) > 0 {
			firstChunkDescTime = s.chunkDescs[0].FirstTime()
		}
	}

	if len(s.chunkDescs) == 0 || through.Before(firstChunkDescTime) {
		return nopIter, nil
	}

	// Find first chunk with start time after "from".
	fromIdx := sort.Search(len(s.chunkDescs), func(i int) bool {
		return s.chunkDescs[i].FirstTime().After(from)
	})
	// Find first chunk with start time after "through".
	throughIdx := sort.Search(len(s.chunkDescs), func(i int) bool {
		return s.chunkDescs[i].FirstTime().After(through)
	})
	if fromIdx == len(s.chunkDescs) {
		// Even the last chunk starts before "from". Find out if the
		// series ends before "from" and we don't need to do anything.
		lt, err := s.chunkDescs[len(s.chunkDescs)-1].LastTime()
		if err != nil {
			return nopIter, err
		}
		if lt.Before(from) {
			return nopIter, nil
		}
	}
	if fromIdx > 0 {
		fromIdx--
	}
	if throughIdx == len(s.chunkDescs) {
		throughIdx--
	}
	if fromIdx > throughIdx {
		// Guard against nonsensical result. The caller will quarantine the series with a meaningful log entry.
		return nopIter, fmt.Errorf("fromIdx=%d is greater than throughIdx=%d, likely caused by data corruption", fromIdx, throughIdx)
	}

	pinIndexes := make([]int, 0, throughIdx-fromIdx+1)
	for i := fromIdx; i <= throughIdx; i++ {
		pinIndexes = append(pinIndexes, i)
	}
	return s.preloadChunks(pinIndexes, fp, mss)
}