Influxdb中基于磁盘的倒排索引文件TSI结构解析

IndexFileTrailerSize = IndexFileVersionSize +
        8 + 8 + // measurement block offset + size
        8 + 8 + // series id set offset + size
        8 + 8 + // tombstone series id set offset + size
        8 + 8 + // series sketch offset + size
        8 + 8 + // tombstone series sketch offset + size
        0

type MeasurementBlock struct {
    data     []byte
    hashData []byte

    // Measurement sketch and tombstone sketch for cardinality estimation.
    sketchData, tSketchData []byte

    version int // block version
}

//遍历每个tag key
for tagkey in tagkeys {
   每个tag key对应多个tag value，遍历
   每个tag key都生成一个tag key entry对象，记录下tag key entry的offset，然后将这个tag key entry放入数组TagKeyEntrys备用
   for tagValue in tagValuesBytagKey {
      buildTagValueBlock
      写这个tagValueBlock的offset和tag value入hash表
   }
   写入这个tag value到tagValueBlockOffset的hash表
}

 for tagkeyEntry in TagKeyEntrys {
   构建tagKeyBlock
   写这个tagKeyBlock的offset和tag key入hash表
}
写入这个tag key到tagKeyBlock Offset的hash表

type IndexFile struct {
    wg   sync.WaitGroup // ref count
    data []byte

    // Components
    sfile *tsdb.SeriesFile // 对应的Seriesile文件
    tblks map[string]*TagBlock //包含的所有TagBlock
    mblk  MeasurementBlock //MeasurementBlock

    // Raw series set data.
    seriesIDSetData          []byte
    tombstoneSeriesIDSetData []byte

    // Series sketch data.
    sketchData, tSketchData []byte

    // Sortable identifier & filepath to the log file.
    level int
    id    int

    mu sync.RWMutex
    // Compaction tracking.
    compacting bool

    // Path to data file.
    path string
}

type IndexFiles []*IndexFile

func (p IndexFiles) MeasurementIterator() MeasurementIterator
func (p *IndexFiles) TagKeyIterator(name []byte) (TagKeyIterator, error)
func (p IndexFiles) MeasurementSeriesIDIterator(name []byte) tsdb.SeriesIDIterator
func (p IndexFiles) TagValueSeriesIDSet(name, key, value []byte) (*tsdb.SeriesIDSet, error) 

type FileSet struct {
    levels       []CompactionLevel
    sfile        *tsdb.SeriesFile
    files        []File // 按最后更改时间从小到大排列
    manifestSize int64 // Size of the manifest file in bytes.
}

func (fs *FileSet) MustReplace(oldFiles []File, newFile File) *FileSet {
    assert(len(oldFiles) > 0, "cannot replace empty files")

    // Find index of first old file.
    var i int
    for ; i < len(fs.files); i++ {
        if fs.files[i] == oldFiles[0] {
            break
        } else if i == len(fs.files)-1 {
            panic("first replacement file not found")
        }
    }

    // Ensure all old files are contiguous.
    for j := range oldFiles {
        if fs.files[i+j] != oldFiles[j] {
            panic(fmt.Sprintf("cannot replace non-contiguous files: subset=%+v, fileset=%+v", Files(oldFiles).IDs(), Files(fs.files).IDs()))
        }
    }

    // Copy to new fileset.
    other := make([]File, len(fs.files)-len(oldFiles)+1)
    copy(other[:i], fs.files[:i])
    other[i] = newFile
    copy(other[i+1:], fs.files[i+len(oldFiles):])

    // Build new fileset and rebuild changed filters.
    return &FileSet{
        levels: fs.levels,
        files:  other,
    }
}

func (p *Partition) compact() {
    if p.isClosing() {
        return
    } else if !p.compactionsEnabled() {
        return
    }
    interrupt := p.compactionInterrupt

    fs := p.retainFileSet()
    defer fs.Release()

    //influxdb的每个partition中tsi的层次是固定的L0-L7,其中L0是wal,这个方法不涉及它的compact
    //L7为最高层，它也不会再被compact了
    //所以这个compact方法需要处理的是L1-L6层
    minLevel, maxLevel := 1, len(p.levels)-2
    for level := minLevel; level <= maxLevel; level++ {
        //如果正在被compact则跳过
        if p.levelCompacting[level] {
            continue
        }

        // 获取当前level中的相邻的index文件列表，按文件更改时间由新到旧排,每次最多compact两个文件，少于两个的不作compact
        files := fs.LastContiguousIndexFilesByLevel(level)
        if len(files) < 2 {
            continue
        } else if len(files) > MaxIndexMergeCount {
            files = files[len(files)-MaxIndexMergeCount:]
        }

        // Retain files during compaction.
        IndexFiles(files).Retain()

        // Mark the level as compacting.
        p.levelCompacting[level] = true

        // 开goroutine作compact
        func(files []*IndexFile, level int) {
            // Start compacting in a separate goroutine.
            p.wg.Add(1)
            go func() {

                // compact到高一级.
                p.compactToLevel(files, level+1, interrupt)

                // Ensure compaction lock for the level is released.
                p.mu.Lock()
                p.levelCompacting[level] = false
                p.mu.Unlock()
                p.wg.Done()

                // Check for new compactions
                p.Compact()
            }()
        }(files, level)
    }

type seriesIDCacheElement struct {
    name        []byte
    key         []byte
    value       []byte
    SeriesIDSet *tsdb.SeriesIDSet
}