engine

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: 29 Imported by: 0

Documentation

Overview

Package engine 实现微时序数据库的存储引擎。

Engine 是数据库的核心组件，负责协调写入和查询操作。 它管理 Shard 的创建和回收，以及元数据的访问。

架构说明： 

Engine → Flusher → ShardManager → Shards → SSTable
Engine → Catalog / SeriesStore / ShardIndex → metadata.Manager
Engine → FlushCoordinator(全局WAL+全局MemTable) → unordered

Engine 是并发安全的，所有公共方法都可以从多个 goroutine 调用。

Package engine 定义存储引擎核心接口。

本文件定义了 Flusher、Catalog、SeriesStore、ShardIndex 四个核心接口，用于解耦 Engine 具体实现与外部依赖。

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrNilPoint              = errors.New("point is nil")
	ErrEmptyDatabase         = errors.New("database name is empty")
	ErrEmptyMeasurement      = errors.New("measurement name is empty")
	ErrInvalidTimestamp      = errors.New("timestamp is negative")
	ErrDatabaseNotFound      = errors.New("database not found")
	ErrDatabaseAlreadyExists = errors.New("database already exists")
	ErrMeasurementNotFound   = errors.New("measurement not found")
	ErrRecoveryFailed        = errors.New("startup recovery failed")
)

错误定义

View Source 
var ErrFlushBackpressure = errors.New("flush backpressure debt exceeds policy")

Functions

func IteratorWithMemTable 

func IteratorWithMemTable(ctx context.Context, shards []*shard.Shard, wmt *memtable.MemTable, extSeriesStore shard.SeriesStore, req *types.QueryRangeRequest) *query.Iterator

IteratorWithMemTable 是包内使用的辅助函数（供测试等场景使用）。

Types

type Catalog 

type Catalog interface {
	CreateDatabase(name string) error
	DropDatabase(name string) error
	ListDatabases() []string
	DatabaseExists(name string) bool
	CreateMeasurement(database, name string) error
	DropMeasurement(database, name string) error
	ListMeasurements(database string) ([]string, error)
	MeasurementExists(database, name string) bool
	GetSchema(database, measurement string) (*metadata.Schema, error)
	SetSchema(database, measurement string, s *metadata.Schema) error
	GetRetention(database, measurement string) (time.Duration, error)
	SetRetention(database, measurement string, d time.Duration) error
	GetDatabaseRetention(database string) (time.Duration, error)
	SetDatabaseRetention(database string, d time.Duration) error
	GetDownsampleConfig(database string) (*types.DownsampleConfig, error)
	SetDownsampleConfig(database string, cfg *types.DownsampleConfig) error
}

Catalog 管理 database/measurement/schema。

type Compactor 

type Compactor interface {
	Compact() error
}

Compactor 是 flush 后需要执行的 unordered→L0 compaction 接口。

type Config 

type Config struct {
	DataDir                string
	ShardDuration          time.Duration
	MemTableCfg            *types.MemTableConfig
	CompactionCfg          *compaction.Config
	LevelCompactionCfg     *compaction.LevelConfig
	AllowDegradedStartup   bool
	CompressionAlgorithm   types.CompressionAlgorithm
	RetentionPeriod        time.Duration
	RetentionCheckInterval time.Duration
	MetadataCacheCfg       metadata.CacheConfig
	WALSyncInterval        time.Duration
	MemoryLimitBytes       int64
}

Config 定义存储引擎的配置。

type Engine 

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

Engine 是微时序数据库的存储引擎。

func New 

func New(cfg *Config) (*Engine, error)

New 创建新的存储引擎实例。

func (*Engine) Close 

func (e *Engine) Close() error

Close 关闭引擎，释放所有资源。

func (*Engine) CreateDatabase 

func (e *Engine) CreateDatabase(database string, retention time.Duration, downsample *types.DownsampleConfig) error

CreateDatabase 创建一个新的数据库。

retention 为数据保留时间，为 0 时使用全局默认值或不启用过期清理。 downsample 为降采样配置，为 nil 时不启用降采样。

func (*Engine) CreateMeasurement 

func (e *Engine) CreateMeasurement(database, measurement string) (bool, error)

CreateMeasurement 在指定数据库中创建一个新的 Measurement。

func (*Engine) DataDir 

func (e *Engine) DataDir() string

DataDir 返回引擎的数据目录。

func (*Engine) DropDatabase 

func (e *Engine) DropDatabase(database string) error

DropDatabase 删除指定的数据库。

func (*Engine) DropMeasurement 

func (e *Engine) DropMeasurement(database, measurement string) (bool, error)

DropMeasurement 删除指定的 Measurement。

func (*Engine) Execute 

func (e *Engine) Execute(ctx context.Context, plan *types.QueryPlan) (*query.RowIterator, error)

Execute 执行查询计划，返回算子 Pipeline 迭代器。

func (*Engine) ExecuteWithOptions 

func (e *Engine) ExecuteWithOptions(ctx context.Context, plan *types.QueryPlan, optionFns ...ExecuteOption) (*query.RowIterator, error)

ExecuteWithOptions 执行查询计划，并允许调用方控制内部执行策略。

func (*Engine) Flush 

func (e *Engine) Flush() error

Flush 将全局 MemTable 的数据刷写到 SSTable。

func (*Engine) ForceDownsample 

func (e *Engine) ForceDownsample()

ForceDownsample 手动触发一次降采样处理。

func (*Engine) Health 

func (e *Engine) Health(ctx context.Context) error

Health 检查引擎是否已完成启动恢复且未关闭。

func (*Engine) InvalidateQueryCacheRange 

func (e *Engine) InvalidateQueryCacheRange(database, measurement string, startTime, endTime int64) int

InvalidateQueryCacheRange 清理与给定时间范围重叠的查询结果缓存。

func (*Engine) Iterator 

func (e *Engine) Iterator(ctx context.Context, req *types.QueryRangeRequest) (*query.Iterator, error)

Iterator 返回流式查询迭代器。 合并全局 MemTable（未刷盘数据）、Shard SSTable（已刷盘数据）和 unordered 文件（未 compaction 数据）。 当 req.DownsampleWindowNanos > 0 时，读取降采样数据而非原始数据。

func (*Engine) IteratorWithOptions 

func (e *Engine) IteratorWithOptions(ctx context.Context, req *types.QueryRangeRequest, optionFns ...IteratorOption) (*query.Iterator, error)

IteratorWithOptions 返回可调整内部行为的流式查询迭代器。

func (*Engine) ListDatabases 

func (e *Engine) ListDatabases() []string

ListDatabases 列出所有数据库名称。

func (*Engine) ListMeasurements 

func (e *Engine) ListMeasurements(database string) ([]string, error)

ListMeasurements 列出指定数据库中的所有 Measurement 名称。

func (*Engine) ResetQueryCache 

func (e *Engine) ResetQueryCache() int

ResetQueryCache 清空 Engine 查询结果缓存，返回删除的条目数。

func (*Engine) SetConfig 

func (e *Engine) SetConfig(config *compaction.Config)

SetConfig 运行时更新所有 Shard 的 Compaction 配置。

func (*Engine) Write 

func (e *Engine) Write(ctx context.Context, point *types.Point) error

Write 先写入 WAL，再写入 MemTable 使单点在查询中可见。

func (*Engine) WriteBatch 

func (e *Engine) WriteBatch(ctx context.Context, points []*types.Point) error

WriteBatch 先批量写入 WAL，再逐点写入 MemTable。 批量写入不是原子操作，已经写入成功的点不会回滚。

type ExecuteOption 

type ExecuteOption func(*executeOptions)

ExecuteOption 调整 Engine.Execute 的内部行为。

func ExecuteNoCache 

func ExecuteNoCache() ExecuteOption

ExecuteNoCache 跳过 query result cache 的读取和写入。

type FlushCoordinator 

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

FlushCoordinator 编排全局 MemTable → unordered 的异步刷盘流程。

func NewFlushCoordinator 

func NewFlushCoordinator(mt *memtable.MemTable, w *wal.WAL, flusher Flusher, compactor Compactor, dataDir string, compression types.CompressionAlgorithm) *FlushCoordinator

NewFlushCoordinator 创建新的 FlushCoordinator。

func (*FlushCoordinator) Close 

func (fc *FlushCoordinator) Close()

Close 停止周期性检查并等待 goroutine 退出。

func (*FlushCoordinator) FlushAll 

func (fc *FlushCoordinator) FlushAll() error

FlushAll 强制刷写所有数据。

func (*FlushCoordinator) FlushAllForWrite 

func (fc *FlushCoordinator) FlushAllForWrite() error

FlushAllForWrite 是写路径的刷盘入口。 当 MemTable 已达到硬上限时优先释放 active 内存，避免 debt policy 将写入卡死。

func (*FlushCoordinator) FlushAllForce 

func (fc *FlushCoordinator) FlushAllForce() error

FlushAllForce 强制刷写所有数据，跳过 debt policy，用于 Close/显式 Flush。

func (*FlushCoordinator) FlushAllWithDebtPolicy 

func (fc *FlushCoordinator) FlushAllWithDebtPolicy() error

FlushAllWithDebtPolicy 按 debt policy 刷写所有数据；debt 超限时返回背压错误。

func (*FlushCoordinator) MemTable 

func (fc *FlushCoordinator) MemTable() *memtable.MemTable

MemTable 返回全局 MemTable。

func (*FlushCoordinator) QuerySnapshot 

func (fc *FlushCoordinator) QuerySnapshot(fn func())

QuerySnapshot 在 flush 可见性提交窗口外构造查询快照。

func (*FlushCoordinator) SetCompactionDebtProvider 

func (fc *FlushCoordinator) SetCompactionDebtProvider(provider compactionDebtProvider)

SetCompactionDebtProvider 注入 level compaction debt 统计来源。

func (*FlushCoordinator) SetPassiveFieldDataReleasePredicate 

func (fc *FlushCoordinator) SetPassiveFieldDataReleasePredicate(fn func() bool)

SetPassiveFieldDataReleasePredicate 设置 passive FieldData 释放条件。

func (*FlushCoordinator) SetPendingSeg 

func (fc *FlushCoordinator) SetPendingSeg(seg uint64)

SetPendingSeg 设置 pending WAL segment，该 segment 内的 entry 不会被 truncation。 用于 MemTable 写入失败但 WAL 已写入的场景，保证重启后 WAL replay 可恢复数据。

func (*FlushCoordinator) StartPeriodicCheck 

func (fc *FlushCoordinator) StartPeriodicCheck(interval time.Duration)

StartPeriodicCheck 启动周期性检查（每 1 秒检查是否需要刷盘）。

func (*FlushCoordinator) VisibilityLocker 

func (fc *FlushCoordinator) VisibilityLocker() sync.Locker

VisibilityLocker 返回 flush/query/compaction 共享的可见性写锁。

type Flusher 

type Flusher interface {
	Flush(points []types.MemPoint) error
	Compact(startTime int64) error
	GetShards(db, measurement string, startTime, endTime int64) []*shard.Shard
	CloseAll() error
	SetConfig(config *compaction.Config)
}

Flusher 管理 SSTable 写入和 Compaction。

type IteratorOption 

type IteratorOption func(*iteratorOptions)

IteratorOption 调整 Engine.Iterator 的内部行为。

func IteratorWithoutTags 

func IteratorWithoutTags() IteratorOption

IteratorWithoutTags 在无 tag filter 的流式查询中跳过 tags 返回。

type SeriesStore 

type SeriesStore interface {
	AllocateSID(database, measurement string, tags map[string]string) (uint64, error)
	GetTags(database, measurement string, sid uint64) (map[string]string, bool)
	GetSIDsByTag(database, measurement string, tagKey, tagValue string) []uint64
	SeriesCount(database, measurement string) int
}

SeriesStore 管理 Series ID 分配和标签索引。

type ShardIndex 

type ShardIndex interface {
	RegisterShard(database, measurement string, info ShardInfo) error
	UnregisterShard(database, measurement string, shardID string) error
	QueryShards(database, measurement string, startTime, endTime int64) []ShardInfo
	ListShards(database, measurement string) []ShardInfo
	UpdateShardStats(database, measurement, shardID string, sstableCount int, totalSize int64) error
}

ShardIndex 管理 Shard 时间范围索引。

type ShardInfo 

type ShardInfo = metadata.ShardInfo

ShardInfo 是 ShardIndex 使用的 Shard 元数据类型。

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.