sstable

package
v0.0.0-...-8909e19 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jun 13, 2026 License: Apache-2.0 Imports: 21 Imported by: 0

Documentation

Overview

internal/storage/shard/sstable/index.go

Index

Constants

View Source 
const (
	CompressionNone   = types.CompressionNone
	CompressionSnappy = types.CompressionSnappy
	CompressionLZ4    = types.CompressionLZ4
)

便捷常量别名

View Source 
const (
	FileVersionCurrent  uint32 = FileVersion
	FileVersionSIDMajor uint32 = FileVersion
)
View Source 
const (
	FlagSorted     uint16 = 0x0000
	FlagUnordered  uint16 = 0x0001
	FlagHasZoneMap uint16 = 0x0002
	FlagSIDMajor   uint16 = 0x0004
)

FlagSorted 表示 SSTable 是有序的（默认值，与旧格式兼容）。 FlagUnordered 表示 SSTable 是无序的，需要排序后才能 compact。 FlagHasZoneMap 表示 SSTable 包含 Zone Map section，支持谓词下推块跳过。 FlagSIDMajor 表示 SSTable 按 SID/time 聚簇布局写入。

View Source 
const BlockSize = 64 * 1024

BlockSize 默认块大小 64KB。

View Source 
const FileVersion uint32 = 4

FileVersion 单文件 SSTable 格式版本。 v4: 强制 SID-major 布局，所有 block 追加 4B CRC32C 校验和。

View Source 
const HeaderSize = 64

HeaderSize 文件头固定大小 (64 字节)。

View Source 
const IndexVersion = 1

IndexVersion 索引版本

Variables

View Source 
var ErrInvalidIndex = &IndexError{msg: "invalid index file"}

ErrInvalidIndex 无效索引错误

View Source 
var IndexMagic = [8]byte{0x54, 0x53, 0x49, 0x44, 0x58, 0x30, 0x30, 0x31}

IndexMagic 是索引文件的魔数。

值："TSIDX001"

View Source 
var Magic = [8]byte{0x54, 0x53, 0x45, 0x52, 0x53, 0x54, 0x42, 0x4C}

Magic 是 SSTable 文件的魔数 "TSERSTBL"。

Functions

func CompressBlock 

func CompressBlock(data []byte, algo CompressionAlgorithm) ([]byte, error)

CompressBlock 压缩已编码的 block 数据并追加 CRC32C 校验和。

输出格式：

  • CompressionNone: [raw_data][crc32:4B BigEndian]
  • CompressionSnappy/LZ4: [uncompressedLen:4B BigEndian][compressed_data][crc32:4B BigEndian]

CRC32C 覆盖整个 payload（不含末尾 4B CRC 自身）：

  • 压缩块：CRC([uncompressedLen:4B] + [compressed_data])
  • 无压缩：CRC([raw_data])

func DecompressBlock 

func DecompressBlock(data []byte, algo CompressionAlgorithm) ([]byte, error)

DecompressBlock 解压 CompressBlock 压缩的数据，并验证 CRC32C 校验和。

返回解压后的原始数据；CRC 不匹配时返回错误。

func MarshalSIDMetaIndex 

func MarshalSIDMetaIndex(metas []SIDBlockMeta) []byte

Types

type BlockIndex 

type BlockIndex struct {
	// contains filtered or unexported fields
}

BlockIndex 是 SSTable 的数据块索引。

为每个数据块维护索引信息，支持快速定位时间范围对应的数据块。

索引条目： 

FirstTimestamp:  Block 中第一个时间戳
LastTimestamp:   Block 中最后一个时间戳
Offset:          累计行号，Block 起始行在已排序数据中的偏移（非字节偏移）
RowCount:        Block 中的行数

查询优化： 

使用二分查找（FindBlock）定位目标 Block。
避免扫描整个文件。

文件格式： 

[magic:8][version:4][count:4][entries...]
每个 entry 24 字节。

func NewBlockIndex 

func NewBlockIndex() *BlockIndex

NewBlockIndex 创建空的 BlockIndex。

func (*BlockIndex) Add 

func (idx *BlockIndex) Add(firstTs, lastTs int64, offset uint32, rowCount uint32)

Add 添加一个 Block 的索引条目。

参数：

  • firstTs: Block 起始时间
  • lastTs: Block 结束时间
  • offset: Block 在文件中的偏移量
  • rowCount: Block 行数

注意： 

应该按时间顺序调用 Add，添加的 Block 时间不应重叠。

func (*BlockIndex) Entry 

func (idx *BlockIndex) Entry(i int) BlockIndexEntry

Entry 返回指定索引的 Block 条目。

参数：

  • i: Block 索引

返回：

  • BlockIndexEntry: Block 信息

panic： 

如果 i < 0 或 i >= Len()，会导致 panic。

func (*BlockIndex) FindBlock 

func (idx *BlockIndex) FindBlock(target int64) int

FindBlock 二分查找第一个包含或晚于目标时间的 Block。

参数：

  • target: 目标时间戳

返回：

  • int: Block 索引，如果不存在返回 len(entries)

查找逻辑： 

查找第一个 LastTimestamp >= target 的 Block。
因为 Block 按时间排序且互不重叠，这一定位是有效的。

使用示例： 

idx := blockIndex.FindBlock(startTime)
for i := idx; i < blockIndex.Len(); i++ {
    entry := blockIndex.Entry(i)
    if entry.FirstTimestamp > endTime {
        break // 超出范围
    }
    // 读取此 block...
}

func (*BlockIndex) Len 

func (idx *BlockIndex) Len() int

Len 返回索引条目数量。

返回：

  • int: Block 数量

func (*BlockIndex) Read 

func (idx *BlockIndex) Read(file string) error

Read 从文件读取索引。

参数：

  • file: 源文件路径

返回：

  • error: 读取失败时返回错误

验证： 

检查文件大小、魔数和版本。
任何验证失败都返回 ErrInvalidIndex。

func (*BlockIndex) Write 

func (idx *BlockIndex) Write(file string) error

Write 将索引写入文件。

参数：

  • file: 目标文件路径

返回：

  • error: 写入失败时返回错误

文件格式： 

[magic:8][version:4][count:4][entries...]
每个 entry 24 字节

错误处理： 

如果写入失败，可能产生不完整的文件。
调用方应考虑原子写入或删除不完整文件。

type BlockIndexEntry 

type BlockIndexEntry struct {
	FirstTimestamp int64  // block 内第一个时间戳
	LastTimestamp  int64  // block 内最后一个时间戳
	Offset         uint32 // 累计行号（非字节偏移），用于定位 block 在已排序数据中的起始行
	RowCount       uint32 // 该 block 的行数
}

BlockIndexEntry 单个 block 的索引条目

type BlockSectionMap 

type BlockSectionMap struct {
	Sections []BlockSectionOffsets
}

BlockSectionMap 记录每个 section 内各 block 的字节偏移，支持按 block 粒度定位数据。 通过 _block_map section 持久化到文件。

func UnmarshalBlockSectionMap 

func UnmarshalBlockSectionMap(data []byte) (*BlockSectionMap, error)

UnmarshalBlockSectionMap 从字节反序列化。

func (*BlockSectionMap) Lookup 

func (m *BlockSectionMap) Lookup(name string) *BlockSectionOffsets

Lookup 按 section 名称查找。

func (*BlockSectionMap) Marshal 

func (m *BlockSectionMap) Marshal() []byte

Marshal 序列化 BlockSectionMap。

格式: 

[section_count:2B]
For each section:
  [name_len:1B][name:variable]
  [block_count:4B]
  [offsets: (block_count+1)*8B]

type BlockSectionOffsets 

type BlockSectionOffsets struct {
	Name    string
	Offsets []uint64
}

BlockSectionOffsets 是单个 section 内各 block 的字节偏移。

Offsets 长度为 blockCount+1，最后一个元素为 sentinel（section 总字节数）， 方便计算最后一个 block 的大小：size = Offsets[i+1] - Offsets[i]。

func (*BlockSectionOffsets) BlockCount 

func (bso *BlockSectionOffsets) BlockCount() int

BlockCount 返回 block 数量。

func (*BlockSectionOffsets) BlockRange 

func (bso *BlockSectionOffsets) BlockRange(blockIdx int) (offset, size uint64)

BlockRange 返回第 blockIdx 个 block 的 (offset, size)，blockIdx 从 0 开始。

type BlockZoneMap 

type BlockZoneMap struct {
	FieldZMaps []ZoneMapEntry
}

BlockZoneMap 单 block 内所有字段的 ZoneMap 汇总。

type CompressionAlgorithm 

type CompressionAlgorithm = types.CompressionAlgorithm

CompressionAlgorithm 通用块压缩算法（类型别名，定义在 proto 中）。

type EncodingType 

type EncodingType uint8

EncodingType 段编码类型。 

const (
	EncodingRaw          EncodingType = 0 // 原始编码
	EncodingDeltaVarint  EncodingType = 1 // Delta-of-Delta + Varint (时间戳)
	EncodingVarint       EncodingType = 2 // Varint (SID)
	EncodingZigZagVarint EncodingType = 3 // ZigZag + Varint (int64)
	EncodingXORFloat     EncodingType = 4 // XOR 浮点 (float64)
	EncodingDictString   EncodingType = 5 // 字典编码 (string)
	EncodingBitmapBool   EncodingType = 6 // 位图 (bool)
)

type FieldType 

type FieldType string

FieldType 字段类型。 

const (
	FieldTypeFloat64 FieldType = "float64"
	FieldTypeInt64   FieldType = "int64"
	FieldTypeString  FieldType = "string"
	FieldTypeBool    FieldType = "bool"
)

type FileHeader 

type FileHeader struct {
	Magic              [8]byte // "TSERSTBL"
	Version            uint32
	RowCount           uint32
	FieldCount         uint16
	BlockCount         uint16
	BlockSize          uint16
	Flags              uint16 // 0=有序, 1=无序 (原 padding 位)
	TimestampsOffset   uint64
	SidsOffset         uint64
	BlockIndexOffset   uint64
	SectionTableOffset uint64
	// contains filtered or unexported fields
}

FileHeader 是单文件 SSTable 的文件头 (64 字节)。

func UnmarshalFileHeader 

func UnmarshalFileHeader(data [HeaderSize]byte) (FileHeader, error)

UnmarshalFileHeader 从 64 字节反序列化 FileHeader。

func (*FileHeader) Marshal 

func (h *FileHeader) Marshal() [HeaderSize]byte

Marshal 将 FileHeader 序列化为 64 字节。

type FilterCondition 

type FilterCondition struct {
	Field string
	Op    int32
	Value float64
}

FilterCondition 用于 ZoneMap 块跳过的简化过滤条件（避免循环导入 types）。

type IndexError 

type IndexError struct {
	// contains filtered or unexported fields
}

IndexError 表示索引操作错误。

包含具体的错误信息。

func (*IndexError) Error 

func (e *IndexError) Error() string

type Iterator 

type Iterator struct {
	// contains filtered or unexported fields
}

Iterator 是 SSTable 的流式迭代器。

func (*Iterator) CurrentBlockFirstTimestamp 

func (it *Iterator) CurrentBlockFirstTimestamp() int64

CurrentBlockFirstTimestamp 返回当前 Block 的起始时间。

func (*Iterator) CurrentBlockLastTimestamp 

func (it *Iterator) CurrentBlockLastTimestamp() int64

CurrentBlockLastTimestamp 返回当前 Block 的结束时间。

func (*Iterator) CurrentMemPoint 

func (it *Iterator) CurrentMemPoint(db, meas string) types.MemPoint

CurrentMemPoint 直接构造 MemPoint，不分配 PointRow/FieldEntry/map。 复用 Iterator 内部的惰性解码缓存，直接将字段值序列化为 FieldData 格式。

func (*Iterator) Done 

func (it *Iterator) Done() bool

Done 返回是否已经遍历完所有数据。

func (*Iterator) Err 

func (it *Iterator) Err() error

Err 返回迭代过程中遇到的底层读取或解码错误。

func (*Iterator) Next 

func (it *Iterator) Next() bool

Next 移动到下一个数据点。

func (*Iterator) Point 

func (it *Iterator) Point() *types.PointRow

Point 返回当前迭代位置的数据点。

func (*Iterator) SeekToTime 

func (it *Iterator) SeekToTime(target int64) error

SeekToTime 定位到指定时间的 Block。

type IteratorOption 

type IteratorOption func(*iteratorOptions)

IteratorOption 调整 SSTable 迭代器读取行为。

func WithQueryBounds 

func WithQueryBounds(startTime, endTime int64, maxRows int) IteratorOption

WithQueryBounds 将查询时间范围和最大候选行数下推到底层 SSTable 迭代器。

func WithSIDFilter 

func WithSIDFilter(filter *seriesfilter.Filter) IteratorOption

WithSIDFilter 将 SSTable 迭代器限制到指定 SID 集合。

type Layout 

type Layout uint8
const (
	LayoutTimeMajor Layout = iota
	LayoutSIDMajor
)

type MergeIterator 

type MergeIterator struct {
	// contains filtered or unexported fields
}

MergeIterator 合并多个 SSTable 文件，按时间戳升序输出。 内部对每个文件使用 Iterator 进行块级流式读取。

func NewMergeIterator 

func NewMergeIterator(filePaths []string, startTime, endTime int64, schema Schema, refMgr SSTableRefManager, fields []string, filterConds []FilterCondition) (*MergeIterator, error)

NewMergeIterator 创建多 SSTable 合并迭代器。

参数：

  • filePaths: SSTable 文件路径列表
  • startTime: 查询起始时间（包含）
  • endTime: 查询结束时间（不包含），<=0 表示无限制
  • schema: 解码所需的 Schema
  • refMgr: 引用计数管理器（Shard 实现），可为 nil
  • fields: 需要投影的字段（nil=全部字段）

func NewMergeIteratorWithLimitOptions 

func NewMergeIteratorWithLimitOptions(filePaths []string, startTime, endTime int64, maxRows int, schema Schema, refMgr SSTableRefManager, fields []string, filterConds []FilterCondition, opts ...IteratorOption) (*MergeIterator, error)

NewMergeIteratorWithLimitOptions 创建多 SSTable 合并迭代器，并将查询边界下推到底层 Iterator。

func NewMergeIteratorWithOptions 

func NewMergeIteratorWithOptions(filePaths []string, startTime, endTime int64, schema Schema, refMgr SSTableRefManager, fields []string, filterConds []FilterCondition, opts ...IteratorOption) (*MergeIterator, error)

NewMergeIteratorWithOptions 创建多 SSTable 合并迭代器，并允许下推底层 Iterator 选项。

func (*MergeIterator) Close 

func (mi *MergeIterator) Close() error

Close 关闭所有底层 Reader 并释放引用。

func (*MergeIterator) Err 

func (mi *MergeIterator) Err() error

Err 返回底层 SSTable reader/iterator 暴露的错误。

func (*MergeIterator) Next 

func (mi *MergeIterator) Next() bool

Next 移动到下一个时间戳最小的行。

func (*MergeIterator) Point 

func (mi *MergeIterator) Point() *types.PointRow

Point 返回当前行。

type Reader 

type Reader struct {
	Flags uint16 // 文件头标志位 (FlagSorted=0 有序, FlagUnordered=1 无序)
	// contains filtered or unexported fields
}

Reader 是 SSTable 的读取器。

func NewReader 

func NewReader(filePath string, schema Schema) (*Reader, error)

NewReader 创建 SSTable 读取器，接收 .bin 文件路径和 schema。

func (*Reader) Close 

func (r *Reader) Close() error

Close 关闭读取器，可重复调用。

func (*Reader) GetBlockIndex 

func (r *Reader) GetBlockIndex() *BlockIndex

GetBlockIndex 返回 BlockIndex。

func (*Reader) NewIterator 

func (r *Reader) NewIterator(fields []string, filterConds []FilterCondition, opts ...IteratorOption) (*Iterator, error)

NewIterator 创建新的流式迭代器。fields 指定投影字段（nil=全部），filterConds 用于 ZoneMap 块跳过。

func (*Reader) ReadAll 

func (r *Reader) ReadAll(fields []string) ([]*types.PointRow, error)

ReadAll 读取 SSTable 中的所有数据，使用编码感知的解码器。

func (*Reader) ReadAllDecodedFieldSections 

func (r *Reader) ReadAllDecodedFieldSections(fields []string, rowCount int) (map[string][]*types.FieldValue, error)

ReadAllDecodedFieldSections 读取并解码所有字段段。

func (*Reader) ReadRange 

func (r *Reader) ReadRange(startTime, endTime int64, maxRows int) ([]*types.PointRow, error)

ReadRange 读取指定时间范围内的数据。 maxRows 限制返回行数（0 表示无限制）。

func (*Reader) SidsOffset 

func (r *Reader) SidsOffset() uint64

SidsOffset 返回 sids 段的文件偏移量。

func (*Reader) SingleSID 

func (r *Reader) SingleSID() (uint64, bool)

SingleSID 返回 SID-major 文件是否只包含单个 SID。

func (*Reader) TimestampsOffset 

func (r *Reader) TimestampsOffset() uint64

TimestampsOffset 返回 timestamps 段的文件偏移量。

type SIDBlockMeta 

type SIDBlockMeta struct {
	SID          uint64
	MinTimestamp int64
	MaxTimestamp int64
	BlockIndex   uint32
	RowStart     uint32
	RowCount     uint32
}

func UnmarshalSIDMetaIndex 

func UnmarshalSIDMetaIndex(data []byte) ([]SIDBlockMeta, error)

type SIDMetaIndex 

type SIDMetaIndex struct {
	// contains filtered or unexported fields
}

func NewSIDMetaIndex 

func NewSIDMetaIndex(metas []SIDBlockMeta) *SIDMetaIndex

func (*SIDMetaIndex) SelectBlocks 

func (idx *SIDMetaIndex) SelectBlocks(sids map[uint64]struct{}, startTime, endTime int64) []int

func (*SIDMetaIndex) SingleSID 

func (idx *SIDMetaIndex) SingleSID() (uint64, bool)

type SSTableRefManager 

type SSTableRefManager interface {
	AcquireSSTRef(path string) bool
	ReleaseSSTRef(path string)
}

SSTableRefManager 管理 SSTable 文件的引用计数。 由 Shard 实现，用于防止 Compaction 删除正在读取的文件。

type Schema 

type Schema struct {
	Fields map[string]FieldType `json:"fields"`
}

Schema 描述 SSTable 的字段结构。

type SectionEntry 

type SectionEntry struct {
	Type        SectionType
	Name        string // 仅字段段有值
	Offset      uint64
	Size        uint64
	Encoding    EncodingType
	Compression CompressionAlgorithm
}

SectionEntry 描述文件中一个段的元数据。

type SectionTable 

type SectionTable struct {
	Entries []SectionEntry
	// contains filtered or unexported fields
}

SectionTable 是文件末尾的段目录。

func UnmarshalSectionTable 

func UnmarshalSectionTable(data []byte) (SectionTable, error)

UnmarshalSectionTable 反序列化 SectionTable。

func (*SectionTable) FieldNames 

func (st *SectionTable) FieldNames() []string

FieldNames 返回所有字段段名称。

func (*SectionTable) Lookup 

func (st *SectionTable) Lookup(name string) (offset, size uint64)

Lookup 按名称查找段，返回 offset 和 size。O(1) 索引查找。未找到返回 0,0。

func (*SectionTable) LookupByType 

func (st *SectionTable) LookupByType(typ SectionType) (offset, size uint64, name string)

LookupByType 按类型查找段。O(1) 索引查找。

func (*SectionTable) LookupCompression 

func (st *SectionTable) LookupCompression(name string) CompressionAlgorithm

LookupCompression 按名称查找段压缩算法。未找到返回 CompressionNone。

func (*SectionTable) LookupEncoding 

func (st *SectionTable) LookupEncoding(name string) EncodingType

LookupEncoding 按名称查找段编码。O(1) 索引查找。未找到返回 EncodingRaw。

func (*SectionTable) Marshal 

func (st *SectionTable) Marshal() []byte

Marshal 序列化 SectionTable。 格式: [count:2B][reserved:2B][entries...] 每个 entry: [type:1B][encoding:1B][compression:1B][nameLen:1B][offset:8B][size:8B][name:nameLen]

type SectionType 

type SectionType uint8

SectionType 段类型。 

const (
	SectionTimestamps SectionType = 0
	SectionSids       SectionType = 1
	SectionIndex      SectionType = 2
	SectionField      SectionType = 3
)

type Writer 

type Writer struct {
	// contains filtered or unexported fields
}

Writer SSTable 写入器。

func NewWriter 

func NewWriter(shardDir string, seq uint64, blockSize int, compressAlgo CompressionAlgorithm, flags uint16) (*Writer, error)

NewWriter 创建 SSTable Writer。 在 shardDir/data/ 下创建 sst_{seq}.bin 单文件。

func NewWriterWithLayout 

func NewWriterWithLayout(
	shardDir string,
	seq uint64,
	blockSize int,
	compressAlgo CompressionAlgorithm,
	flags uint16,
	layout Layout,
) (*Writer, error)

func (*Writer) Close 

func (w *Writer) Close() error

func (*Writer) Schema 

func (w *Writer) Schema() Schema

Schema 返回写入过程中检测到的 schema。

func (*Writer) SetSyncOnClose 

func (w *Writer) SetSyncOnClose(v bool)

SetSyncOnClose 设置 Close 时是否执行 fsync。flush 场景可设为 false， 由 WAL + tmp/rename 模式保证原子性和持久性。

func (*Writer) WithContext 

func (w *Writer) WithContext(ctx context.Context) *Writer

WithContext 设置 Writer 的 context，用于支持超时取消。 取消 context 后，Close() 中正在进行的 I/O 操作会尽快返回错误。

func (*Writer) WriteMemPoints 

func (w *Writer) WriteMemPoints(points []types.MemPoint) error

WriteMemPoints 直接写入 MemPoint 切片，从 FieldData 字节流解码字段值到列式缓冲区。 跳过 MemPoint → InternalPoint → WritePoints 中间态，消除 deserializeFieldData + NewFieldValue 分配。

func (*Writer) WritePointRows 

func (w *Writer) WritePointRows(rows []*types.PointRow) error

WritePointRows 直接写入 PointRow 切片，跳过 InternalField 中间转换。 首次调用执行字段发现和文件映射建立，后续调用复用缓存跳过 O(N*F) 字段扫描。

func (*Writer) WritePointRowsSIDMajor 

func (w *Writer) WritePointRowsSIDMajor(rows []*types.PointRow) error

WritePointRowsSIDMajor 写入已按 (sid,timestamp) 排序的 PointRow。 调用方必须保证跨批次也保持 SID-major 顺序；该路径用于单 SID level merge 等已验证顺序的场景，避免将大 compaction 输出缓存在 pendingRows。

func (*Writer) WritePoints 

func (w *Writer) WritePoints(points []types.InternalPoint) error

WritePoints 写入一批 InternalPoint 到 SSTable。 首次调用执行字段发现，后续调用复用缓存跳过扫描。

type ZoneMapEntry 

type ZoneMapEntry struct {
	FieldName string
	Min       float64
	Max       float64
}

ZoneMapEntry 记录单个字段在单个 block 中的 min/max 值。

type ZoneMapIndex 

type ZoneMapIndex struct {
	Blocks []BlockZoneMap
}

ZoneMapIndex 所有 block 的 ZoneMap 汇总。

func UnmarshalZoneMapIndex 

func UnmarshalZoneMapIndex(data []byte) (*ZoneMapIndex, error)

UnmarshalZoneMapIndex 反序列化 ZoneMapIndex。

func (*ZoneMapIndex) Lookup 

func (zm *ZoneMapIndex) Lookup(blockIdx int, fieldName string) (ZoneMapEntry, bool)

Lookup 按 block 索引和字段名查找 ZoneMap 条目。

func (*ZoneMapIndex) Marshal 

func (zm *ZoneMapIndex) Marshal() []byte

Marshal 序列化 ZoneMapIndex。 格式: [block_count:4B][for each block: field_count:2B][for each field: name_len:2B, name:var, min:8B, max:8B]

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.
⌂ 换网址