最终版 (final branch) — 合并自 experiment (写入优化)、tablet_refine (Tablet 写入优化)、read_opt (读取优化)
TsFile C++ 是 Apache TsFile 时序数据文件格式的 C++ 实现，提供高性能的时序数据读写能力。
代码规模：157 个头文件 + 54 个源文件 = 211 个文件
src/ ├── common/ # 公共基础设施 │ ├── allocator/ # 内存管理 (ByteStream, PageArena, MyString) │ ├── cache/ # LRU 缓存 │ ├── config/ # 全局配置 │ ├── constant/ # TsFile 常量定义 │ ├── container/ # 容器 (BitMap, Array, HashTable, List) │ ├── datatype/ # 日期转换等数据类型工具 │ ├── logger/ # 日志 (elog) │ ├── mutex/ # 互斥锁封装 │ └── tsblock/ # TsBlock 列式数据块 │ └── vector/ # 列向量 (FixedLengthVector, VariableLengthVector) ├── compress/ # 压缩 (LZ4, Snappy, LZO, Uncompressed) ├── cwrapper/ # C 语言封装层 ├── encoding/ # 编码 (Plain, RLE, TS2Diff, Gorilla, Sprintz, Dictionary, ZigZag) ├── file/ # 文件 I/O (ReadFile, WriteFile, TsFileIOReader/Writer) ├── parser/ # 路径解析 (ANTLR4 生成) ├── reader/ # 读取引擎 │ ├── block/ # TsBlock 读取器 │ ├── filter/ # 时间/值过滤器 │ └── task/ # 查询任务 ├── utils/ # 工具函数 └── writer/ # 写入引擎
Tablet 是用户向 TsFile 写入数据的主要接口。 内存布局: ┌──────────────┐ │ timestamps_ │ → int64_t[] 时间戳数组 ├──────────────┤ │ value_matrix_│ → ValueMatrixEntry[] 列值数组 │ ├─ bool* │ (每列一个 entry，类型由 union 区分) │ ├─ int32* │ │ ├─ int64* │ │ ├─ float* │ │ ├─ double* │ │ └─ StringColumn* ← [优化] Arrow 风格连续字符串存储 ├──────────────┤ │ bitmaps_ │ → BitMap[] 空值位图 (bit=1 表示 null) └──────────────┘
StringColumn 优化 (来自 experiment 分支):
String* 数组 (每个字符串独立分配)offsets[] + buffer 连续存储string[i] = buffer + offsets[i], len = offsets[i+1] - offsets[i]append() 自动扩容，reset() 零拷贝重置set_column_string_repeated() 批量填充相同字符串批量写入 API:
set_timestamps(ptr, count) — 批量拷贝时间戳set_column_values(idx, data, bitmap, count) — 批量拷贝列数据set_column_string_repeated(idx, str, len, count) — 批量填充字符串列TsBlock 是列式内存数据块，用于读取路径返回数据。 ┌─────────────┐ │ TupleDesc │ → 列描述 (列名 + 数据类型) ├─────────────┤ │ Columns[] │ → FixedLengthVector / VariableLengthVector │ └─ data │ 每列存储对应类型的数组 │ └─ nulls │ 空值位图 ├─────────────┤ │ row_count │ → 当前行数 │ capacity │ → 最大容量 └─────────────┘ ColAppender → 向列追加数据 RowAppender → 管理行提交 ColIterator → 读取列数据
ByteStream 是一个链式分页缓冲区，用于编码和 I/O。 ┌──────┐ ┌──────┐ ┌──────┐ │ Page │──→│ Page │──→│ Page │──→ NULL │ buf │ │ buf │ │ buf │ └──────┘ └──────┘ └──────┘ ↑ head ↑ tail 特性: - OptionalAtomic<uint32_t> total_size_ (可选原子操作，支持并发写入) - 支持 wrap_from() 零拷贝包装外部缓冲区 - page_mask_ 位运算取模优化 (代替 %) - AllocModID 追踪内存分配模块
BitMap 用于标记 null 值 (bit=1 表示 null)。 [优化] 来自 read_opt 分支: - count_set_bits(): 使用 __builtin_popcount 快速统计 null 数量 - next_set_bit(from, total): 使用 __builtin_ctz 快速跳过非 null 字节 → 用于读取时跳过 null 行，时间复杂度与 null 字节数成正比
用户接口 写入引擎 文件格式
┌──────────┐ ┌───────────────┐ ┌──────────────┐
│ Tablet / │ write_ │ TsFileWriter │ flush │ TsFileIO │
│ TsRecord │──tablet──→│ │─────────→│ Writer │
└──────────┘ write_ │ ThreadPool │ │ │
table │ (6 threads) │ │ WriteFile │
└───────┬───────┘ └──────────────┘
│
┌─────────────┼─────────────┐
↓ ↓ ↓
┌─────────────┐ ┌────────────┐ ┌──────────────┐
│ ChunkWriter │ │ TimeChunk │ │ ValueChunk │
│ (non-align) │ │ Writer │ │ Writer │
└──────┬──────┘ └─────┬──────┘ └──────┬───────┘
↓ ↓ ↓
┌─────────────┐ ┌────────────┐ ┌──────────────┐
│ PageWriter │ │ TimePage │ │ ValuePage │
│ │ │ Writer │ │ Writer │
└─────────────┘ └────────────┘ └──────────────┘
三种写入模式:
| 模式 | 方法 | Chunk 类型 | 说明 |
|---|---|---|---|
| Tree 非对齐 | write_tree() / write_tablet() | ChunkWriter | 每个 measurement 独立时间戳 |
| Tree 对齐 | write_tablet_aligned() | TimeChunkWriter + ValueChunkWriter | 共享时间戳列 |
| Table 模型 | write_table() | TimeChunkWriter + ValueChunkWriter | 按 TAG 列自动分组设备 |
Table 写入流程:
split_tablet_by_device() — 根据 TAG 列 (ID columns) 找到设备边界find_all_device_boundaries() — 使用 uint64 位图 + __builtin_ctzll 快速扫描边界time_write_column_batch() + value_write_column_batch()write_tablet() / write_table()
│
├─ [快速路径] 无 null → write_column_batch()
│ └─ chunk_writer->write_batch(timestamps, values, count)
│ └─ page_writer_.write_batch(timestamps, values, count)
│ └─ 直接 memcpy 或批量编码
│
└─ [慢速路径] 有 null → write_column() (逐行)
└─ for each row: chunk_writer->write(timestamp, value, is_null)
value_write_column_batch():
└─ value_chunk_writer->write_batch(timestamps, values, bitmap, start, count)
└─ 支持所有基本类型的批量写入
└─ STRING/TEXT/BLOB 回退到逐行
// 写入路径使用通用线程池 (6 线程) common::ThreadPool thread_pool_{6}; // 支持 std::future 获取结果 auto future = thread_pool_.submit([&] { return do_work(); }); auto result = future.get();
用户接口 读取引擎 文件格式
┌──────────────┐ ┌─────────────────┐ ┌───────────────┐
│ TsFileReader │───────→│ QueryExecutor │───────→│ TsFileIO │
│ │ query │ QueryDataSet │ read │ Reader │
└──────────────┘ └────────┬────────┘ │ │
│ │ ReadFile │
┌─────────────────┤ └───────────────┘
↓ ↓
┌─────────────────┐ ┌───────────────────┐
│ TsBlockReader │ │ QdsWithTime │
│ (Table 模型) │ │ Generator │
│ │ │ (Tree 模型) │
└────────┬────────┘ └───────────────────┘
↓
┌──────────────────────────┐
│ SingleDeviceTsBlockReader│
│ ├─ MeasurementColumnCtx │
│ │ ├─ SingleMeasurement │ → 单列
│ │ └─ VectorMeasurement │ → 多列 (Aligned)
│ └─ IdColumnCtx │ → TAG 列
└──────────┬───────────────┘
↓
┌─────────────────────────┐
│ TsFileSeriesScanIterator│ ← 时间序列扫描迭代器
│ ├─ ChunkReader │ (非对齐)
│ └─ AlignedChunkReader │ (对齐)
└─────────────────────────┘
当所有查询列都来自 aligned chunk group 时:
has_next() → has_next_aligned()
│
└─ 避免逐行 merge-sort
└─ 批量 bulk_copy_into() 直接拷贝 TsBlock 数据
└─ 利用 available_rows() 计算可批量拷贝的行数
TsFileSeriesScanIterator:
set_row_range(offset, limit)
│
├─ should_skip_chunk_by_offset(cm)
│ └─ 利用 chunk 统计信息 (count) 跳过整个 chunk
│
└─ get_next_page() 6-param overload
└─ row_offset, row_limit 传入 AlignedChunkReader
└─ 利用 page 统计信息跳过整个 page
AlignedChunkReader:
load_by_aligned_meta_multi(time_meta, value_metas[])
│
└─ 一次加载 time chunk + N 个 value chunk
└─ get_next_page_multi() → 批量解码多列
└─ 避免 N 次独立 chunk 加载开销
// 读取路径使用轻量级解码线程池 class DecodeThreadPool { submit(task); // 提交解码任务 wait_all(); // 等待所有解码完成 };
Encoder (base) ├── PlainEncoder # 无编码，直接存储 ├── RleEncoder # Run-Length Encoding ├── Ts2DiffEncoder # 时间序列二阶差分 + zigzag ├── GorillaEncoder # Gorilla 压缩 (浮点数 XOR) ├── SprintzEncoder # Sprintz 编码 │ ├── Int32SprintzEncoder │ ├── Int64SprintzEncoder │ ├── FloatSprintzEncoder │ └── DoubleSprintzEncoder ├── DictionaryEncoder # 字典编码 └── ZigZagEncoder # ZigZag 变长编码 Decoder (对称结构) ├── PlainDecoder ├── RleDecoder ├── Ts2DiffDecoder ├── GorillaDecoder ├── SprintzDecoder ├── DictionaryDecoder └── ZigZagDecoder
PlainEncoder:
write_batch(values, count) → 直接 memcpy
其他编码器:
支持 write_batch() → 内部循环调用 encode()
通过 simde (SIMDe) 库提供跨平台 SIMD 支持，用于编码/解码加速。
Compressor (base) ├── UncompressedCompressor # 无压缩 ├── LZ4Compressor # LZ4 压缩 ├── SnappyCompressor # Snappy 压缩 └── LZOCompressor # LZO 压缩
AllocBase (g_base_allocator)
│
├─ mem_alloc(size, mod_id) # 带模块标记的分配
├─ mem_free(ptr) # 释放
├─ mem_realloc(ptr, size) # 重分配
│
└─ PageArena # 页面级内存池
├─ alloc(size) # 从当前页分配
└─ reset() # 整体回收 (不逐个释放)
AllocModID (内存模块标记):
MOD_DEFAULT, MOD_TABLET, MOD_CW_PAGES_DATA,
MOD_TSFILE_WRITER, ...
ByteStream(AllocModID mid)
└─ get_mid() → 传播给内部分配
└─ SerializationUtil 使用 mid 进行所有中间分配
┌───────────────────┐
│ Magic ("TsFile") │ 6 bytes
├───────────────────┤
│ Chunk Group 1 │ ← 一个设备的数据
│ ├─ Chunk 1 │ ← 一个 measurement 的数据
│ │ ├─ Header │
│ │ ├─ Page 1 │ ← 一个数据页
│ │ ├─ Page 2 │
│ │ └─ ... │
│ ├─ Chunk 2 │
│ └─ ... │
├───────────────────┤
│ Chunk Group 2 │
├───────────────────┤
│ ... │
├───────────────────┤
│ Index Area │ ← 时间序列索引
│ ├─ MetaData │
│ ├─ BloomFilter │
│ └─ IndexTree │
├───────────────────┤
│ Footer │
│ └─ Magic │
└───────────────────┘
对齐写入时，一个设备的所有 measurement 共享时间戳:
Aligned Chunk Group:
├── TimeChunk → 时间戳列 (只存一份)
│ ├── TimePage 1
│ └── TimePage 2
├── ValueChunk (col1) → 值列1
│ ├── ValuePage 1
│ └── ValuePage 2
└── ValueChunk (col2) → 值列2
├── ValuePage 1
└── ValuePage 2
| 来源分支 | 主要变更 |
|---|---|
| experiment | Tablet StringColumn 优化、批量写入 API (set_timestamps, set_column_values, set_column_string_repeated)、write_column_batch/time_write_column_batch/value_write_column_batch 快速路径、ThreadPool (6 线程写入并发)、ByteStream 构造器参数化 mid、SerializationUtil 使用 mem_alloc 替代 malloc |
| tablet_refine | Tablet 写入优化 (已被 experiment 完全覆盖) |
| read_opt | BitMap count_set_bits/next_set_bit 优化、AlignedChunkReader 6-param get_next_page (offset/limit 下推)、aligned fast path (has_next_aligned + bulk_copy_into)、多值列合并读取 (load_by_aligned_meta_multi)、DecodeThreadPool、TsFileSeriesScanIterator row_range 支持 |
1. 用户创建 Tablet，调用 set_timestamps() + set_column_values() 2. TsFileWriter::write_table(tablet) 3. split_tablet_by_device() → 按 TAG 列分组 4. 对每个设备: a. do_check_schema_table() → 获取/创建 ChunkWriter b. time_write_column_batch() → 批量写入时间戳 c. value_write_column_batch() → 批量写入各列值 5. check_memory_size_and_may_flush_chunks() → 内存超限则 flush 6. flush() → TsFileIOWriter 写入磁盘 7. close() → 写入索引和 footer
1. TsFileReader::query(columns, filter) 2. QueryExecutor → SingleDeviceTsBlockReader 3. 对每个 measurement: a. 创建 MeasurementColumnContext b. 初始化 TsFileSeriesScanIterator c. 加载 TimeseriesIndex 4. has_next(): a. [aligned fast path] → bulk_copy_into() 批量拷贝 b. [merge path] → 逐行合并多列 5. next() → 返回 TsBlock 6. 用户遍历 TsBlock 的 ColIterator