InfluxDB 当中保存三组不同类型文件

TSM：保存时序数据，通过 SeriesKey 可以查询到保存的时序数据
TSI：时序数据的「倒排索引」，可以快速找到某个 measurement 下某个 tag 里包含某个 value 的 SeriesID 有哪些
Series：保存 SeriesID 和 SeriesKey 之间的映射

可以通过 SeriesKey 找到对应的 SeriesID
可以从 SeriesID 找到 SeriesKey 在其「日志文件」中的位置（offset），再通过位置从文件中读取 SeriesKey

其实 InfluxDB 文档中，上面提到的 TSI 和 Series 合在一起才是其所谓的 TSI，不过由于 Series 本身也有一套自己的文件和 Compact 算法，所以这里把它们俩拆分开了。

分工

在处理查询时：

首先，通过 TSI 查询符合条件的 SeriesID
再从 SeriesFile 当中找到对应的 SeriesKey
找到 SeriesKey 之后，再筛选出包含 SeriesKey 数据的 TSM 文件（一般还会根据时间范围进行筛选，过滤掉时间范围不符合条件的文件）
最后，合并 TSM 文件中的数据，用于做进一步的计算（window、aggregation 等等）

SeriesID

刚开始看这个分工的时候，可能会觉得有些奇怪，这个流程似乎有些繁琐，要先找到 SeriesID 之后再找 SeriesKey，为什么不直接通过索引找到 SeriesKey？

实际上，SeriesID 对于 InfluxDB 的用户而言应该也是感知不到的。用户会知道系统里现在一共有多少 SeriesKey，但是不需要知道对应的 SeriesID 是什么。

之所以引入 SeriesID，我想是为了提高内存使用效率。试想，如果直接索引 SeriesKey 的话，SeriesKey 是一个 byte 数组，格式上有个多个 tagKey 和 tagValue 拼接而成，是很占内存的。而 SeriesID 是自增的 64 位整形，索引 SeriesID 的话内存占用最少会降低一个数量级，通过位图可以进一步提高内存使用效率。

不过，增加了 SeriesID，就必须要增加 SeriesID 和 SeriesKey 映射关系的存储，也就增加了系统的复杂度。这应该是权衡之后的结果。

查询示例分析

在处理下面这个查询语句的时候：

from(bucket:"pageload")
  |> range(start:-1000h)
  |> filter(fn: (r) =>
      r._field == "duration" and
      r.system=="ios"
  )
  |> window(every: 1h)
  |> median()

第一步，系统会先根据 from 和 filter 中的条件，从 TSI 中获取符合条件的 SeriesID，如果索引数据已经被存储在内存缓存中的话，会执行下面这些代码查找 SeriesID：

// tsdb/ts11/cache.go
type TagValueSeriesIDCache struct {
  //...
  // measurement/tagKey/tagValue 到 SeriesID 的映射
  cache   map[string]map[string]map[string]*list.Element
  //...
}

//...
func (c *TagValueSeriesIDCache) get(name, key, value []byte) *tsdb.SeriesIDSet {
  if mmap, ok := c.cache[string(name)]; ok { // 从映射中找到 "pageload" 的数据
    if tkmap, ok := mmap[string(key)]; ok { // 在下一层找 tagKey 为 "system" 的数据
      if ele, ok := tkmap[string(value)]; ok { // 在下一层找 system=="ios" 的数据
        //...
        return ele.Value.(*seriesIDCacheElement).SeriesIDSet // 返回找到的 SeriesID 集合
      }
    }
  }
  //...
  return nil
}

第二步，找到 SeriesID 后，继续从 Series 当中查询对应的 SeriesKey，这部分主要由 tsdb.SeriesFile 负责，入口代码如下：

// tsdb/series_file.go
func (f *SeriesFile) SeriesKey(id SeriesID) []byte {
    //...
  p := f.SeriesIDPartition(id)
  //...
  return p.SeriesKey(id)
}

// tsdb/series_partition.go
func (p *SeriesPartition) SeriesKey(id SeriesID) []byte {
  //...
  key := p.seriesKeyByOffset(p.index.FindOffsetByID(id))
  //...
  return key
}

第三步，找到 SeriesKey 之后，继续从 TSM 文件当中找到对应的文件并读取数据，这个环节还会使用 range 中设置的时间范围对 TSM 文件进行筛选。

这部分的代码逻辑可以通过下面的测试用例来研究一下，里面涵盖了通过 seriesKey 和时间范围筛选 TSM 文件并从多个文件当中 merge 数据的过程：

// tsdb/tsm1/array_cursor_test.go
func TestFileStore_DuplicatePoints(t *testing.T) {
    //...

    data := []keyValues{
        {"m,_field=v#!~#v", makeVals(21)},
        {"m,_field=v#!~#v", makeVals(44)},
        {"m,_field=v#!~#v", makeVals(40, 46)},
        {"m,_field=v#!~#v", makeVals(46, 51)},
    }

    files, err := newFiles(dir, data...)
    //...

    t.Run("ascending", func(t *testing.T) {
        const START, END = 21, 100
        kc := fs.KeyCursor(context.Background(), []byte("m,_field=v#!~#v"), START, true)
        defer kc.Close()
        cur := newFloatArrayAscendingCursor()
        cur.reset(START, END, nil, kc)

        var got []int64
        ar := cur.Next()
        for ar.Len() >  {
            got = append(got, ar.Timestamps...)
            ar = cur.Next()
        }

        if exp := []int64{21, 40, 44, 46, 51}; !cmp.Equal(got, exp) {
            t.Errorf("unexpected values; -got/+exp\n%s", cmp.Diff(got, exp))
        }
    })
    //..
}

写入流程分析

数据写入的应用层代码大致如下：

func (e *Engine) WritePoints(ctx context.Context, points []models.Point) error {
    // 对 points 进行一系列的转换，转换成 values
    // ...

    // 先写入到 TSM 的 WAL
    if _, err := e.wal.WriteMulti(ctx, values); err != nil {
        return err
    }

    // 写入数据
    return e.writePointsLocked(ctx, collection, values)
}

func (e *Engine) writePointsLocked(ctx context.Context, collection *tsdb.SeriesCollection, values map[string][]value.Value) error {
    //...

    // 创建 TSI 如果 Series 还不存在
    if err := e.index.CreateSeriesListIfNotExists(collection); err != nil {
        return err
    }

    //...

    // 数据写入到 TSM，默认会先写入缓存
    if err := e.engine.WriteValues(values); err != nil {
        return err
    }

    return collection.PartialWriteError()
}

其中

代码第 18 行，会尝试创建 TSI 和 Series，如果 Series 不存在的话
代码第 25 行将数据写入到 TSM 引擎，会先写入到缓存，当缓存超出阈值的时候，会触发 Compact
创建 TSI 和 Series 的过程也会有类似的 Compact 机制，后面会对这些过程做具体介绍

小结

上面描述了三种不同文件 TSM、TSI、Series 之间的分工，以及查询和写入数据时三者之间的协作关系。它们都实现了类似 LSM 的结构，都含有内存缓存、文件合并的工作机制，后面具体整理一下它们的具体结构和工作细节。

ITPub博客

InfluxDB 源码学习：文件结构（一）

分工

SeriesID

查询示例分析

写入流程分析

小结

最新文章