InfluxDB中的inmem内存索引结构解析

type tagKeyValueEntry struct {
    m map[uint64]struct{} // series id set
    a seriesIDs           // lazily sorted list of series.这两个字段存储的是相同的series id的集合
}

func (e *tagKeyValueEntry) ids() seriesIDs {
    if e == nil {
        return nil
    }
    
    //首先调用`ids`方法时才给a赋值
    if len(e.a) == len(e.m) {
        return e.a
    }
    
    a := make(seriesIDs, 0, len(e.m))
    for id := range e.m {
        a = append(a, id)
    }
    
    //基数排序
    radix.SortUint64s(a)
    
    e.a = a
    return e.a
}

type tagKeyValue struct {
    mu      sync.RWMutex
    entries map[string]*tagKeyValueEntry
}

type measurement struct {
    Database  string
    Name      string `json:"name,omitempty"`
    NameBytes []byte // cached version as []byte

    mu         sync.RWMutex
    fieldNames map[string]struct{} //存储了当前measurement的所有字段

    // in-memory index fields
    seriesByID          map[uint64]*series      // series id到series的映射
    seriesByTagKeyValue map[string]*tagKeyValue // tagkey -> tag value -> series id set

    // lazyily created sorted series IDs
    sortedSeriesIDs seriesIDs // sorted list of series IDs in this measurement

    // Indicates whether the seriesByTagKeyValueMap needs to be rebuilt as it contains deleted series
    // that waste memory.
    dirty bool
}

func (m *measurement) TagKeys() []string {
    m.mu.RLock()
    keys := make([]string, 0, len(m.seriesByTagKeyValue))
    for k := range m.seriesByTagKeyValue {
        keys = append(keys, k)
    }
    m.mu.RUnlock()
    sort.Strings(keys)
    return keys
}

func (m *measurement) TagValues(auth query.Authorizer, key string) []string {
    m.mu.RLock()
    defer m.mu.RUnlock()
    values := make([]string, 0, m.seriesByTagKeyValue[key].Cardinality())

    //遍历所有的tagKeyValue
    m.seriesByTagKeyValue[key].RangeAll(func(k string, a seriesIDs) {
        if query.AuthorizerIsOpen(auth) {
            values = append(values, k)
        } else {
            for _, sid := range a {
                s := m.seriesByID[sid]
                if s == nil {
                    continue
                }
                if auth.AuthorizeSeriesRead(m.Database, m.NameBytes, s.Tags) {
                    values = append(values, k)
                    return
                }
            }
        }
    })
    return values
}

func (m *measurement) SeriesIDsByTagKey(key []byte) seriesIDs {
    tagVals := m.seriesByTagKeyValue[string(key)]
    if tagVals == nil {
        return nil
    }

    var ids seriesIDs
    tagVals.RangeAll(func(_ string, a seriesIDs) {
        ids = append(ids, a...)
    })
    sort.Sort(ids)
    return ids
}

func (m *measurement) SeriesIDsByTagValue(key, value []byte) seriesIDs {
    tagVals := m.seriesByTagKeyValue[string(key)]
    if tagVals == nil {
        return nil
    }
    return tagVals.Load(string(value))
}

type Index struct {
    mu sync.RWMutex

    database string //数据库名字
    sfile    *tsdb.SeriesFile  //_series 目录下series file
    fieldset *tsdb.MeasurementFieldSet

    // In-memory metadata index, built on load and updated when new series come in
    measurements map[string]*measurement // measurement name to object and index
    series       map[string]*series      // map series key to the Series object

    seriesSketch, seriesTSSketch             estimator.Sketch
    measurementsSketch, measurementsTSSketch estimator.Sketch

    // Mutex to control rebuilds of the index
    rebuildQueue sync.Mutex
}

type IndexSet struct {
    Indexes    []Index                // The set of indexes comprising this IndexSet.
    SeriesFile *SeriesFile            // The Series File associated with the db for this set.
    fieldSets  []*MeasurementFieldSet // field sets for _all_ indexes in this set's DB.
}