type CompactionPlanner interface {Plan(lastWrite time.Time) []CompactionGroupPlanLevel(level int) []CompactionGroupPlanOptimize() []CompactionGroupRelease(group []CompactionGroup)FullyCompacted() bool// ForceFull causes the planner to return a full compaction plan the next// time Plan() is called if there are files that could be compacted.ForceFull()SetFileStore(fs *FileStore)
}

type tsmGeneration struct {id            int // Generationfiles         []FileStat //包含的tsm文件的信息, 并且这个files是按文件名从小到大排序好的parseFileName ParseFileNameFunc //这个函数用来从tsm文件名中解析出Generation和Sequence number
}

func (c *DefaultPlanner) PlanLevel(level int) []CompactionGroup {... // Determine the generations from all files on disk.  We need to treat// a generation conceptually as a single file even though it may be// split across several files in sequence.// 将相同generation id的tsm文件放在一起// generations -> tsmGenerationsgenerations := c.findGenerations(true)...// 按level把tsmGenerations分组// 这些分完组后groups中的tsmGenerations的level从大到小排列的var currentGen tsmGenerationsvar groups []tsmGenerationsfor i := 0; i < len(generations); i++ {cur := generations[i]// See if this generation is orphan'd which would prevent it from being further// compacted until a final full compactin runs.if i < len(generations)-1 {if cur.level() < generations[i+1].level() {currentGen = append(currentGen, cur)continue}}if len(currentGen) == 0 || currentGen.level() == cur.level() {currentGen = append(currentGen, cur)continue}groups = append(groups, currentGen)currentGen = tsmGenerations{}currentGen = append(currentGen, cur)}if len(currentGen) > 0 {groups = append(groups, currentGen)}// Remove any groups in the wrong level// level是这个函数传进来的参数，指明要compact哪一level的file,这里作个过滤// cur.level()返回的是这个tmsGeneration中所有fileState中最小的level, 这样作// 合适吗？var levelGroups []tsmGenerationsfor _, cur := range groups {if cur.level() == level {levelGroups = append(levelGroups, cur)}}minGenerations := 4if level == 1 {//对于level至少要有8个文件，才会compactminGenerations = 8}//type CompactionGroup []stringvar cGroups []CompactionGroupfor _, group := range levelGroups {// 将每个tsmGenerations中的tsmGeneration按给定大小分堆for _, chunk := range group.chunk(minGenerations) {var cGroup CompactionGroupvar hasTombstones boolfor _, gen := range chunk {if gen.hasTombstones() {hasTombstones = true}for _, file := range gen.files {//cGroup里存需要被分组compact的file.PathcGroup = append(cGroup, file.Path)}}// 如果当前的chunk里的tsmGeneration数不够minGeneration大小，// 需要用下一个chunk来凑够这个数// hasTombstones为true, 说明有标记删除的，需要通过 compact 真正删除掉if len(chunk) < minGenerations && !hasTombstones {continue}cGroups = append(cGroups, cGroup)}}if !c.acquire(cGroups) {return nil}return cGroups
}

func (c *DefaultPlanner) Plan(lastWrite time.Time) []CompactionGroup {generations := c.findGenerations(true)for _, v := range generations {fmt.Printf("xxx | generations: %vn", v)}c.mu.RLock()forceFull := c.forceFullc.mu.RUnlock()// first check if we should be doing a full compaction because nothing has been written in a long time// fullCompact是有时间间隔的，这里作判断// 这部分处理fullCompact的情况if forceFull || c.compactFullWriteColdDuration > 0 && time.Since(lastWrite) > c.compactFullWriteColdDuration && len(generations) > 1 {// Reset the full schedule if we planned because of it.if forceFull {c.mu.Lock()c.forceFull = falsec.mu.Unlock()}var tsmFiles []stringvar genCount intfor i, group := range generations {var skip bool// Skip the file if it's over the max size and contains a full block and it does not have any tombstonesif len(generations) > 2 && group.size() > uint64(maxTSMFileSize) &&c.FileStore.BlockCount(group.files[0].Path, 1) == tsdb.DefaultMaxPointsPerBlock &&!group.hasTombstones() {skip = true}// compressed files.if i < len(generations)-1 {if generations[i+1].level() <= 3 {skip = false}}if skip {continue}for _, f := range group.files {tsmFiles = append(tsmFiles, f.Path)}genCount += 1}sort.Strings(tsmFiles)// Make sure we have more than 1 file and more than 1 generationif len(tsmFiles) <= 1 || genCount <= 1 {return nil}group := []CompactionGroup{tsmFiles}if !c.acquire(group) {return nil}return group}// don't plan if nothing has changed in the filestoreif c.lastPlanCheck.After(c.FileStore.LastModified()) && !generations.hasTombstones() {return nil}c.lastPlanCheck = time.Now()// If there is only one generation, return early to avoid re-compacting the same file// over and over again.if len(generations) <= 1 && !generations.hasTombstones() {return nil}// Need to find the ending point for level 4 files.  They will be the oldest files. We scan// each generation in descending break once we see a file less than 4.end := 0start := 0// 因为finGeneratons返回的是按level从大到小排序的// 这里找到level >= 4的截至点for i, g := range generations {if g.level() <= 3 {break}end = i + 1}// As compactions run, the oldest files get bigger.  We don't want to re-compact them during// this planning if they are maxed out so skip over any we see.var hasTombstones boolfor i, g := range generations[:end] {if g.hasTombstones() {hasTombstones = true}if hasTombstones {continue}// 下面这部分主要是跳到过大的tsm文件// Skip the file if it's over the max size and contains a full block or the generation is split// over multiple files.  In the latter case, that would mean the data in the file spilled over// the 2GB limit.if g.size() > uint64(maxTSMFileSize) &&c.FileStore.BlockCount(g.files[0].Path, 1) == tsdb.DefaultMaxPointsPerBlock {start = i + 1}// This is an edge case that can happen after multiple compactions run.  The files at the beginning// can become larger faster than ones after them.  We want to skip those really big ones and just// compact the smaller ones until they are closer in size.if i > 0 {if g.size()*2 < generations[i-1].size() {start = ibreak}}}// step is how may files to compact in a group.  We want to clamp it at 4 but also stil// return groups smaller than 4.step := 4if step > end {step = end}// slice off the generations that we'll examinegenerations = generations[start:end]// 下面这些代码主要就是将generations分堆，也就是最后要将tsm文件分堆，以便并行作compaction// Loop through the generations in groups of size step and see if we can compact all (or// some of them as group)groups := []tsmGenerations{}for i := 0; i < len(generations); i += step {var skipGroup boolstartIndex := ifor j := i; j < i+step && j < len(generations); j++ {gen := generations[j]lvl := gen.level()// Skip compacting this group if there happens to be any lower level files in the// middle.  These will get picked up by the level compactors.if lvl <= 3 {fmt.Printf("xxx | lvl <= 3")skipGroup = truebreak}// Skip the file if it's over the max size and it contains a full blockif gen.size() >= uint64(maxTSMFileSize) && c.FileStore.BlockCount(gen.files[0].Path, 1) == tsdb.DefaultMaxPointsPerBlock && !gen.hasTombstones() {startIndex++continue}}if skipGroup {continue}endIndex := i + stepif endIndex > len(generations) {endIndex = len(generations)}if endIndex-startIndex > 0 {groups = append(groups, generations[startIndex:endIndex])}}if len(groups) == 0 {return nil}// With the groups, we need to evaluate whether the group as a whole can be compactedcompactable := []tsmGenerations{}for _, group := range groups {//if we don't have enough generations to compact, skip itif len(group) < 4 && !group.hasTombstones() {continue}compactable = append(compactable, group)}// All the files to be compacted must be compacted in order.  We need to convert each// group to the actual set of files in that group to be compacted.var tsmFiles []CompactionGroupfor _, c := range compactable {var cGroup CompactionGroupfor _, group := range c {for _, f := range group.files {cGroup = append(cGroup, f.Path)}}sort.Strings(cGroup)tsmFiles = append(tsmFiles, cGroup)}if !c.acquire(tsmFiles) {return nil}return tsmFiles
}

func (c *cacheKeyIterator) encode() {concurrency := runtime.GOMAXPROCS(0)n := len(c.ready)// Divide the keyset across each CPUchunkSize := 1idx := uint64(0)// 启动多个goroutine来作encodefor i := 0; i < concurrency; i++ {// Run one goroutine per CPU and encode a section of the key space concurrentlygo func() {// 获取Time, Float, Boolean, Unsigned, String, Iterger的编码器tenc := getTimeEncoder(tsdb.DefaultMaxPointsPerBlock)fenc := getFloatEncoder(tsdb.DefaultMaxPointsPerBlock)benc := getBooleanEncoder(tsdb.DefaultMaxPointsPerBlock)uenc := getUnsignedEncoder(tsdb.DefaultMaxPointsPerBlock)senc := getStringEncoder(tsdb.DefaultMaxPointsPerBlock)ienc := getIntegerEncoder(tsdb.DefaultMaxPointsPerBlock)defer putTimeEncoder(tenc)defer putFloatEncoder(fenc)defer putBooleanEncoder(benc)defer putUnsignedEncoder(uenc)defer putStringEncoder(senc)defer putIntegerEncoder(ienc)for {i := int(atomic.AddUint64(&idx, uint64(chunkSize))) - chunkSizeif i >= n {break}key := c.order[i]values := c.cache.values(key)for len(values) > 0 {//每次最多编码c.size个valueend := len(values)if end > c.size {end = c.size}minTime, maxTime := values[0].UnixNano(), values[end-1].UnixNano()var b []bytevar err errorswitch values[0].(type) {case FloatValue:b, err = encodeFloatBlockUsing(nil, values[:end], tenc, fenc)case IntegerValue:b, err = encodeIntegerBlockUsing(nil, values[:end], tenc, ienc)case UnsignedValue:b, err = encodeUnsignedBlockUsing(nil, values[:end], tenc, uenc)case BooleanValue:b, err = encodeBooleanBlockUsing(nil, values[:end], tenc, benc)case StringValue:b, err = encodeStringBlockUsing(nil, values[:end], tenc, senc)default:b, err = Values(values[:end]).Encode(nil)}// 更新values为剩余未编码的values = values[end:]// 每个key对应c.blocks中的一项，里面存储的是cacheBlockc.blocks[i] = append(c.blocks[i], cacheBlock{k:       key,minTime: minTime,maxTime: maxTime,b:       b,err:     err,})if err != nil {c.err = err}}// Notify this key is fully encoded// 对于每个key, 如果全部编码完成，就向这个key对应的chan中写入数据，通知其编码完成c.ready[i] <- struct{}{}}}()}
}

// 将Cache的内容写入到 *.tsm.tmp文件中
// cache中value过多的话，会将cache作split成多个cache,并行处理，每个splited cache有自己的generation
func (c *Compactor) WriteSnapshot(cache *Cache) ([]string, error) {c.mu.RLock()enabled := c.snapshotsEnabledintC := c.snapshotsInterruptc.mu.RUnlock()if !enabled {return nil, errSnapshotsDisabled}start := time.Now()// cache.Count() 返回cache的所有的 value的个数card := cache.Count()// Enable throttling if we have lower cardinality or snapshots are going fast.// 3e6 = 3 x 10的6次方// compaction过程是否要作流控throttle := card < 3e6 && c.snapshotLatencies.avg() < 15*time.Second// Write snapshost concurrently if cardinality is relatively high.concurrency := card / 2e6if concurrency < 1 {concurrency = 1}// Special case very high cardinality, use max concurrency and don't throttle writes.if card >= 3e6 {concurrency = 4throttle = false}splits := cache.Split(concurrency)type res struct {files []stringerr   error}resC := make(chan res, concurrency)for i := 0; i < concurrency; i++ {go func(sp *Cache) {iter := NewCacheKeyIterator(sp, tsdb.DefaultMaxPointsPerBlock, intC)files, err := c.writeNewFiles(c.FileStore.NextGeneration(), 0, nil, iter, throttle)resC <- res{files: files, err: err}}(splits[i])}var err errorfiles := make([]string, 0, concurrency)for i := 0; i < concurrency; i++ {result := <-resCif result.err != nil {err = result.err}files = append(files, result.files...)}... return files, err
}

func (c *Compactor) writeNewFiles(generation, sequence int, src []string, iter KeyIterator, throttle bool) ([]string, error) {// These are the new TSM files writtenvar files []stringfor {// sequence + 1, 这个sequence其实就是 levelsequence++// 这里写入的文件的命名为 *.tsm.tmp// 它在作fullCompact时被重命名为 *.tsmfileName := filepath.Join(c.Dir, c.formatFileName(generation, sequence)+"."+TSMFileExtension+"."+TmpTSMFileExtension)// Write as much as possible to this file// c.write实现了实际的写入操作err := c.write(fileName, iter, throttle)// We've hit the max file limit and there is more to write.  Create a new file// and continue.// 写入的文件大小或block数达到上限，就切下一个文件，sequence + 1if err == errMaxFileExceeded || err == ErrMaxBlocksExceeded {files = append(files, fileName)continue} else if err == ErrNoValues {// ErrNoValues意味着没有有效的value, 只有tombstoned entires, 就不写入文件// If the file only contained tombstoned entries, then it would be a 0 length// file that we can drop.if err := os.RemoveAll(fileName); err != nil {return nil, err}break} else if _, ok := err.(errCompactionInProgress); ok {// Don't clean up the file as another compaction is using it.  This should not happen as the// planner keeps track of which files are assigned to compaction plans now.return nil, err} else if err != nil {// Remove any tmp files we already completedfor _, f := range files {if err := os.RemoveAll(f); err != nil {return nil, err}}// We hit an error and didn't finish the compaction.  Remove the temp file and abort.if err := os.RemoveAll(fileName); err != nil {return nil, err}return nil, err}files = append(files, fileName)break}return files, nil
}

func (k *tsmBatchKeyIterator) Next() bool {
RETRY:// Any merged blocks pending?if len(k.merged) > 0 {k.merged = k.merged[1:]if len(k.merged) > 0 {return true}}// Any merged values pending?if k.hasMergedValues() {k.merge()if len(k.merged) > 0 || k.hasMergedValues() {return true}}// If we still have blocks from the last read, merge themif len(k.blocks) > 0 {k.merge()if len(k.merged) > 0 || k.hasMergedValues() {return true}}// Read the next block from each TSM iterator// 读每一个tsm文件，将其第一组block都存到k.buf里，看起来是要合并排序// 每个tsm文件对应一个blocks// 这个blocks和tsm的index是一样的，是按key从小到大排序的for i, v := range k.buf {if len(v) != 0 {continue}iter := k.iterators[i]if iter.Next() {key, minTime, maxTime, typ, _, b, err := iter.Read()if err != nil {k.err = err}// This block may have ranges of time removed from it that would// reduce the block min and max time.// 这个tombstones是[]TimeRangetombstones := iter.r.TombstoneRange(key)var blk *block// k.buf[i]的类型是[]blocks -> [][]block// 下面这段逻辑，就是不断向k.buf[i]中append新的bolck// 如果k.buf[i]需要扩容，就在append时扩，扩为原有cap的二倍if cap(k.buf[i]) > len(k.buf[i]) {k.buf[i] = k.buf[i][:len(k.buf[i])+1]blk = k.buf[i][len(k.buf[i])-1]if blk == nil {blk = &block{}k.buf[i][len(k.buf[i])-1] = blk}} else {blk = &block{}k.buf[i] = append(k.buf[i], blk)}blk.minTime = minTimeblk.maxTime = maxTimeblk.key = keyblk.typ = typblk.b = bblk.tombstones = tombstonesblk.readMin = math.MaxInt64blk.readMax = math.MinInt64blockKey := key// 如果这两个key相等，说明还没有遍历完当前的blockfor bytes.Equal(iter.PeekNext(), blockKey) {iter.Next()key, minTime, maxTime, typ, _, b, err := iter.Read()if err != nil {k.err = err}tombstones := iter.r.TombstoneRange(key)var blk *blockif cap(k.buf[i]) > len(k.buf[i]) {k.buf[i] = k.buf[i][:len(k.buf[i])+1]blk = k.buf[i][len(k.buf[i])-1]if blk == nil {blk = &block{}k.buf[i][len(k.buf[i])-1] = blk}} else {blk = &block{}k.buf[i] = append(k.buf[i], blk)}blk.minTime = minTimeblk.maxTime = maxTimeblk.key = keyblk.typ = typblk.b = bblk.tombstones = tombstonesblk.readMin = math.MaxInt64blk.readMax = math.MinInt64}}if iter.Err() != nil {k.err = iter.Err()}}// Each reader could have a different key that it's currently at, need to find// the next smallest one to keep the sort ordering.// 找出当前最小的key(series key + field)// 因为k.buf中的每个blocks都是按key从小到大排好的，// 所以这里只需看每个blocks[0]var minKey []bytevar minType bytefor _, b := range k.buf {// block could be nil if the iterator has been exhausted for that fileif len(b) == 0 {continue}if len(minKey) == 0 || bytes.Compare(b[0].key, minKey) < 0 {minKey = b[0].keyminType = b[0].typ}}k.key = minKeyk.typ = minType// Now we need to find all blocks that match the min key so we can combine and dedupe// the blocks if necessary// 把key都等于上面获取的minKey的block放到k.blocks中for i, b := range k.buf {if len(b) == 0 {continue}//b[0]即为当前的k.buf[i][0], 是一个block// b是[]blockif bytes.Equal(b[0].key, k.key) {//k.blocks => []block// b => []blockk.blocks = append(k.blocks, b...)//k.buf[i]的length被reset为0, 即已有的数据被清掉k.buf[i] = k.buf[i][:0]}}if len(k.blocks) == 0 {return false}k.merge()// After merging all the values for this key, we might not have any.  (e.g. they were all deleted// through many tombstones).  In this case, move on to the next key instead of ending iteration.if len(k.merged) == 0 {goto RETRY}return len(k.merged) > 0
}

func (k *tsmBatchKeyIterator) combineFloat(dedup bool) blocks {if dedup {//实现了按minTime来排序,去重for k.mergedFloatValues.Len() < k.size && len(k.blocks) > 0 {// 去除已经读取过的blockfor len(k.blocks) > 0 && k.blocks[0].read() {k.blocks = k.blocks[1:]}if len(k.blocks) == 0 {break}first := k.blocks[0]minTime := first.minTimemaxTime := first.maxTime// Adjust the min time to the start of any overlapping blocks.// 其实i可以从1开始// 为了按minTime排序，需要确定一个全局最小范围的[minTime, maxTime]for i := 0; i < len(k.blocks); i++ {if k.blocks[i].overlapsTimeRange(minTime, maxTime) && !k.blocks[i].read() {if k.blocks[i].minTime < minTime {minTime = k.blocks[i].minTime}// 将最大值减小if k.blocks[i].maxTime > minTime && k.blocks[i].maxTime < maxTime {maxTime = k.blocks[i].maxTime}}}// We have some overlapping blocks so decode all, append in order and then dedup// 按上面确定的[minTime, maxTime]在所有的blocks中捞数据for i := 0; i < len(k.blocks); i++ {if !k.blocks[i].overlapsTimeRange(minTime, maxTime) || k.blocks[i].read() {continue}var v tsdb.FloatArrayvar err errorif err = DecodeFloatArrayBlock(k.blocks[i].b, &v); err != nil {k.err = errreturn nil}// Remove values we already readv.Exclude(k.blocks[i].readMin, k.blocks[i].readMax)// Filter out only the values for overlapping block// 这个Include是不是可以不用调用v.Include(minTime, maxTime)if v.Len() > 0 {// Record that we read a subset of the blockk.blocks[i].markRead(v.MinTime(), v.MaxTime())}// Apply each tombstone to the blockfor _, ts := range k.blocks[i].tombstones {v.Exclude(ts.Min, ts.Max)}k.mergedFloatValues.Merge(&v)}}// Since we combined multiple blocks, we could have more values than we should put into// a single block.  We need to chunk them up into groups and re-encode them.return k.chunkFloat(nil)}var i intfor i < len(k.blocks) {// skip this block if it's values were already readif k.blocks[i].read() {i++continue}// If we this block is already full, just add it as is// 遇到一个不full的Block就break, 那如果后续还有full的block怎么办?if BlockCount(k.blocks[i].b) >= k.size {k.merged = append(k.merged, k.blocks[i])} else {break}i++}if k.fast {for i < len(k.blocks) {// skip this block if it's values were already readif k.blocks[i].read() {i++continue}k.merged = append(k.merged, k.blocks[i])i++}}// If we only have 1 blocks left, just append it as is and avoid decoding/recodingif i == len(k.blocks)-1 {if !k.blocks[i].read() {k.merged = append(k.merged, k.blocks[i])}i++}// The remaining blocks can be combined and we know that they do not overlap and// so we can just append each, sort and re-encode.for i < len(k.blocks) && k.mergedFloatValues.Len() < k.size {if k.blocks[i].read() {i++continue}var v tsdb.FloatArrayif err := DecodeFloatArrayBlock(k.blocks[i].b, &v); err != nil {k.err = errreturn nil}// Apply each tombstone to the blockfor _, ts := range k.blocks[i].tombstones {v.Exclude(ts.Min, ts.Max)}k.blocks[i].markRead(k.blocks[i].minTime, k.blocks[i].maxTime)k.mergedFloatValues.Merge(&v)i++}k.blocks = k.blocks[i:]return k.chunkFloat(k.merged)
}