本教程涵盖 TsDB2 时序数据库的所有核心操作,包括建表、增删改查、条件查询、索引管理、事务控制等。
TsDB2 使用 Go struct 定义数据模型,通过 vrtx tag 声明字符串长度与索引:
package main
import (
"git.ycl.pub/yl32940/TsDB2/DB"
"fmt"
)
// 员工表模型
type Employee struct {
DB.BaseModel // 内嵌 → _id 主键 (uint64, 自增/可显式指定)
Name string `vrtx:"maxlen:32"` // 字符串字段,最大 32 字节
Age int64 // 数值类型无需 tag,按 Go 类型自动推断 (有符号 int64 → Int64)
Score int64 // 有符号 int64 → Int64 (保序编码存盘)
Dept string `vrtx:"maxlen:16"` // 字符串字段,最大 16 字节
}
// 自定义主键: 内嵌基类 (表必须有基类), 再用 vrtx:"pk" 覆盖默认的 _id 列
type Product struct {
DB.BaseModel // 表基类: 提供 Insert/Update/Delete/Save/Load/... 头部方法
SKU int64 `vrtx:"pk"` // 主键必须是 64 位整数 (int/uint/int64/uint64), 列名即字段名
Name string `vrtx:"maxlen:32"`
Price int64
}
// 订单表模型
type Order struct {
DB.BaseModel
EmployeeID int64 // 有符号 int64 → Int64
Amount int64 // 有符号 int64 → Int64
Status int64 // 0=pending, 1=done, 2=cancel
}
字段声明规则:
DB.BaseModel / DB.PrimaryModel(主键表,_id)或 DB.TimeSeriesModel(时序表,_ts)。基类既是"这是一张本框架的表"的标记,也提供对象头部的增删改查(Insert / Update / Delete / Save / Load / Reload / PK)。没有内嵌基类的结构体不是本框架的表:ParseModel / UseTable 直接报 DB.ErrNotModel(校验在建表之前,库里不会留下半成品表)。Int8/Int16/Int32/Int64,盘上用保序编码),无符号整数走无符号档(Uint8/Uint16/Uint32/Uint64,v9 起的语义名):int8 → Int8、int16 → Int16、int32 → Int32、int64/int → Int64;uint8 → Uint8、uint16 → Uint16、uint32 → Uint32、uint64 → Uint64。两档盘上字节不同、不可混用,否则范围查询 / 排序会出错。vrtx:"maxlen:N" 指定最大字节数,映射到最接近的定长档位(16~2048 共 24 档,约 1.25 倍递增就近向上取整,如 20→24B、200→224B;>2048 转动态列)。N 的上界是 16360 字节:超过 2048 字节的列转动态列(内容整体存进一个堆页),声明超过这个数的列在 ParseModel 阶段直接报错 —— 那种容量兑不了现,与其等写入时才失败,不如建模时就拒。index:名称(默认 B+ 树)或 index:名称,type:hash(Hash 桶式索引)。vrtx:"pk" 把基类默认的键列(_id / _ts)覆盖成自己的字段(_id / _ts 不再建列),字段类型必须是 int / uint / int64 / uint64;字符串、浮点、int32 等窄整数会在 ParseModel 时报 DB.ErrInvalidPrimaryKey。注意 pk 标签不能替代基类 —— 只写 pk 而不内嵌基类同样报 DB.ErrNotModel。vrtx:"maxlen:32,index:idx_name"。tsdb 为旧标签,仍向后兼容,但新代码请统一使用 vrtx。// 打开或创建数据库文件
db, err := DB.Open("mydb.db")
if err != nil {
panic(err)
}
defer db.Close() // 程序退出前关闭
// 设置自动提交延迟(防抖)
db.SetAutoCommitDelay(500 * time.Millisecond) // 500ms 后自动 Commit
// ✅ 推荐方式:自动推导表名(从类型名 Employee -> "employee")
empTbl, err := DB.UseTable[Employee](db)
if err != nil {
panic(err)
}
// 可选:手动指定表名
orderTbl, err := DB.UseTable[Order](db, "orders") // 自定义表名为 "orders"
if err != nil {
panic(err)
}
工作原理:
UseTable[Employee](db) → 自动推导表名为 "employee"(首字母小写)UseTable[OrderItem](db) → 自动推导表名为 "orderItem"UseTable[T](db, "custom_name") → 使用自定义表名T 必须内嵌表基类(DB.BaseModel / DB.PrimaryModel / DB.TimeSeriesModel),否则 UseTable 报 DB.ErrNotModel。这道校验发生在建表之前,所以被拒绝的类型不会留下一张半成品表。Table[T] 后两种写法都可用:empTbl.Insert(&Employee{...})(把对象交给表),或对象头部 emp.Insert()(见 12.7)。推荐方式:在模型上用 vrtx tag 声明,UseTable 会自动创建主键索引与所有二级索引:
type Employee struct {
DB.BaseModel // 内嵌 → 自动生成主键索引 pk_employee (_id)
Name string `vrtx:"maxlen:32"` // 无 index → 不建二级索引
Age int64 `vrtx:"index:idx_age,type:hash"` // Hash 二级索引 (等值 O(1))
Score int64 `vrtx:"index:idx_score"` // B+ 二级索引 (默认,支持范围/有序)
Dept string `vrtx:"maxlen:16,index:idx_dept"` // B+ 二级索引
}
底层方式:也可通过 Core 对象手动建索引(字段序号从 1 开始,按结构体字段顺序):
import "git.ycl.pub/yl32940/TsDB2/DB/core/page"
empCore, _ := db.CoreDB().FindTable("employee")
empCore.CreateIndex("idx_age", page.IndexTypeBTree, []byte{2}) // Age 字段索引
empCore.CreateIndex("idx_dept", page.IndexTypeBTree, []byte{4}) // Dept 字段索引
索引类型说明:
IndexTypePrimary (2): 主键索引,叶子节点存 [Key → DataPage](行数据,聚簇)IndexTypeBTree (5): 二级 B+ 树索引,叶子节点存 [Key → PKID],支持等值/范围/有序IndexTypeHash (6): Hash 二级索引,等值 O(1),范围查询需全桶扫描IndexTypeTime (1): 时序索引,优化时间范围查询IndexTypeCompose (4): 复合索引。处理器底层即 SBT 二级 B+ 树(与 IndexTypeBTree 同结构/同性能):最左前缀等值/范围 O(logN)、全局有序、MIN/MAX 端点快路径可用;各分量按真实宽度定长拼接,分量不限 8 字节(Decimal/Year/定长字符串列均可)。ORM 侧用默认的多字段同名 index:NAME 即为 BTree 复合键;写 type:compose 会报「未知字段类型」(ORM 只认 hash/btree),core API 可直接用 IndexTypeComposeerr := empTbl.Insert(&Employee{
Name: "张三",
Age: 25,
Score: 95,
Dept: "工程部",
})
if err != nil {
fmt.Printf("插入失败: %v\n", err)
}
// 插入 10000 条员工记录
for i := 1; i <= 10000; i++ {
err := empTbl.Insert(&Employee{
Name: fmt.Sprintf("Emp%07d", i),
Age: int64(20 + (i % 50)), // 年龄 20~69
Score: int64(i % 1000), // 分数 0~999
Dept: []string{"工程部", "销售部", "人事部", "财务部", "运维部"}[i%5],
})
if err != nil {
panic(fmt.Sprintf("插入第 %d 行失败: %v", i, err))
}
// 每 1000 行打印进度
if i%1000 == 0 {
fmt.Printf("已插入 %d 行\n", i)
}
}
// 禁用自动提交,手动控制 Commit 时机
db.SetAutoCommitDelay(0) // 禁用防抖
// 批量插入 10 万行
for i := 1; i <= 100000; i++ {
empTbl.Insert(&Employee{...})
}
// 一次性提交所有数据
fmt.Println("开始 Commit...")
t1 := time.Now()
if err := db.Commit(); err != nil {
panic(err)
}
fmt.Printf("Commit 完成,耗时: %.2fs\n", time.Since(t1).Seconds())
性能对比:
// 根据 ID 查询
emp, err := empTbl.Find(1)
if err != nil {
fmt.Printf("未找到 ID=1 的记录\n")
} else {
fmt.Printf("ID=%d, Name=%s, Age=%d, Score=%d, Dept=%s\n",
emp.ID, emp.Name, emp.Age, emp.Score, emp.Dept)
}
// WHERE Age = 25
rows, err := empTbl.Where("Age", "=", int64(25)).Find()
if err != nil {
panic(err)
}
fmt.Printf("找到 %d 名 25 岁的员工\n", len(rows))
for _, emp := range rows {
fmt.Printf(" - %s (Score: %d)\n", emp.Name, emp.Score)
}
// 全表扫描
allEmps, err := empTbl.FindAll() // 无条件,返回所有记录
if err != nil {
panic(err)
}
fmt.Printf("总共有 %d 名员工\n", len(allEmps))
// 循环处理
for i, emp := range allEmps {
if emp.Score >= 90 {
fmt.Printf("[%d] 优秀员工: %s, Score=%d\n", i+1, emp.Name, emp.Score)
}
}
// 只解码 Name / Age 两列;未选中的字段在结果对象里保持零值
rows, _ := empTbl.Where("Dept", "=", "工程部").Select("Name", "Age").Find()
for _, emp := range rows {
fmt.Printf(" %s: %d岁\n", emp.Name, emp.Age) // emp.Score / emp.Dept 等是零值
}
Select(cols ...string) 接受 Go 字段名,投影下推到引擎:未选中的列不做反序列化与堆页解引用。ORDER BY 列会被自动并入解码集,排序行为不受影响。反序列化失败会中断查询:行数据损坏(blob 链断裂、读取长度与引用不符等)时,
Find/Scan/Select返回 error 并停止扫描,不再静默跳过坏行。
// 更新 ID=1 的员工信息
err := empTbl.Update(&Employee{
BaseModel: DB.BaseModel{ID: 1}, // 必须指定 ID
Name: "李四",
Age: 30,
Score: 98,
Dept: "技术部",
})
if err != nil {
fmt.Printf("更新失败: %v\n", err)
}
// 验证更新
emp, _ := empTbl.Find(1)
fmt.Printf("更新后: Name=%s, Age=%d\n", emp.Name, emp.Age)
// 输出: 更新后: Name=李四, Age=30
// 给所有工程部员工加 10 分
emps, _ := empTbl.Where("Dept", "=", "工程部").Find()
for _, emp := range emps {
emp.Score += 10
empTbl.Update(emp)
}
fmt.Printf("已为 %d 名工程部员工加分\n", len(emps))
// 删除 ID=2 的员工
err := empTbl.Delete(2)
if err != nil {
fmt.Printf("删除失败: %v\n", err)
}
// 验证删除
_, err = empTbl.Find(2)
if err != nil {
fmt.Println("✅ ID=2 已成功删除")
}
// 删除所有分数低于 60 的员工
lowScores, _ := empTbl.Where("Score", "<", int64(60)).Find()
fmt.Printf("找到 %d 名低分员工\n", len(lowScores))
for _, emp := range lowScores {
empTbl.Delete(emp.ID)
}
fmt.Println("✅ 已删除所有低分员工")
// 验证
remaining, _ := empTbl.Count()
fmt.Printf("剩余员工数: %d\n", remaining)
// 删除部门时,同时删除该部门所有员工
deptToDelete := "临时工"
emps, _ := empTbl.Where("Dept", "=", deptToDelete).Find()
for _, emp := range emps {
empTbl.Delete(emp.ID)
}
fmt.Printf("已删除 %d 名 %s 员工\n", len(emps), deptToDelete)
// 等于 (=)
rows, _ := empTbl.Where("Age", "=", int64(25)).Find()
// 不等于 (!=)
rows, _ = empTbl.Where("Status", "!=", int64(0)).Find()
// 大于 (>)
rows, _ = empTbl.Where("Score", ">", int64(90)).Find()
// 大于等于 (>=)
rows, _ = empTbl.Where("Age", ">=", int64(30)).Find()
// 小于 (<)
rows, _ = empTbl.Where("Score", "<", int64(60)).Find()
// 小于等于 (<=)
rows, _ = empTbl.Where("Age", "<=", int64(25)).Find()
// WHERE Age >= 20 AND Age <= 30
rows, _ := empTbl.
Where("Age", ">=", int64(20)).
Where("Age", "<=", int64(30)).
Find()
fmt.Printf("20~30岁员工: %d 人\n", len(rows))
// WHERE Dept IN ("工程部", "技术部")
rows, _ := empTbl.Where("Dept", "in", []string{"工程部", "技术部"}).Find()
fmt.Printf("工程/技术部员工: %d 人\n", len(rows))
// 整数集合 (分量按列类型自动归一); 空集合恒假 (0 行, 不是全表)
rows, _ = empTbl.Where("Age", "in", []int64{20, 22, 24}).Find()
rows, _ = empTbl.Where("Age", "in", []int64{}).Find() // → 0 行
IN 与索引:IN 列上带索引(B+ / Hash / 复合索引的等值前缀)时,每个值展开一次索引定位, 不再退化为全表扫描;重复值自动去重。IN 与同列范围条件并存时,范围条件走索引区间、 IN 由行级过滤兜底(正确性不受影响)。
// WHERE Name LIKE "张%"
rows, _ := empTbl.Where("Name", "like", "张%").Find()
fmt.Printf("姓张的员工: %d 人\n", len(rows))
// WHERE Name LIKE "%三"
rows, _ = empTbl.Where("Name", "like", "%三").Find()
fmt.Printf("名字以'三'结尾的员工: %d 人\n", len(rows))
// WHERE Age > 25 AND Score >= 80 AND Dept = "工程部"
rows, _ := empTbl.
Where("Age", ">", int64(25)).
Where("Score", ">=", int64(80)).
Where("Dept", "=", "工程部").
Find()
fmt.Printf("符合条件的员工: %d 人\n", len(rows))
ORM 层的 Where 链只表达 AND;OR / NOT / 嵌套组合要到 core 层用条件构造器(core.Or / core.And / core.Not),把组合条件交给 core Query:
import (
"git.ycl.pub/yl32940/TsDB2/DB/core"
"git.ycl.pub/yl32940/TsDB2/DB/core/types"
)
// WHERE Dept = "工程部" OR Dept = "技术部" —— 等价于 ORM 的 Where("Dept","in",...)
empCore, _ := db.CoreDB().FindTable("employee")
cond := core.Or(
core.Eq("Dept", "工程部", types.FieldType(0)),
core.Eq("Dept", "技术部", types.FieldType(0)),
)
n, err := empCore.Query().Where(cond).Count()
core.Eq的第三个参数是列类型提示,types.FieldType(0)表示自动推断;条件构造器还有Gt/Ge/Lt/Le/Like/ILike/In,可继续用core.And/core.Or/core.Not任意嵌套。
// ORDER BY Age ASC
rows, _ := empTbl.OrderBy("Age").Find()
fmt.Println("按年龄升序:")
for _, emp := range rows[:5] { // 只看前5条
fmt.Printf(" %s: %d岁\n", emp.Name, emp.Age)
}
// ORDER BY Score DESC
rows, _ := empTbl.OrderBy("Score", true).Find() // true = 降序
fmt.Println("按分数降序(前5名):")
for i, emp := range rows[:5] {
fmt.Printf(" [%d] %s: %d分\n", i+1, emp.Name, emp.Score)
}
// ORDER BY Age ASC LIMIT 10
rows, _ := empTbl.OrderBy("Age").Limit(10).Find()
fmt.Printf("最年轻的 10 名员工:\n")
for _, emp := range rows {
fmt.Printf(" %s: %d岁\n", emp.Name, emp.Age)
}
ORDER BY + LIMIT 是流式 top-k:内存只保留
limit+offset行的小顶堆,免全量排序; 相等键的输出顺序与全量排序一致(稳定)。
// 每页 20 条
pageSize := 20
// 第1页
page1, _ := empTbl.OrderBy("Age").Offset(0).Limit(pageSize).Find()
fmt.Printf("第1页: %d 条\n", len(page1))
// 第2页
page2, _ := empTbl.OrderBy("Age").Offset(pageSize).Limit(pageSize).Find()
fmt.Printf("第2页: %d 条\n", len(page2))
// 第3页
page3, _ := empTbl.OrderBy("Age").Offset(pageSize*2).Limit(pageSize).Find()
fmt.Printf("第3页: %d 条\n", len(page3))
func printPage(tbl *DB.Table[Employee], pageNum, pageSize int) {
offset := (pageNum - 1) * pageSize
rows, _ := tbl.OrderBy("Age").Offset(offset).Limit(pageSize).Find()
fmt.Printf("\n=== 第 %d 页 (共 %d 条) ===\n", pageNum, len(rows))
for _, emp := range rows {
fmt.Printf(" %-15s %2d岁 %3d分 %s\n",
emp.Name, emp.Age, emp.Score, emp.Dept)
}
}
// 打印前3页
for page := 1; page <= 3; page++ {
printPage(empTbl, page, 20)
}
// SELECT COUNT(*) FROM employee
total, err := empTbl.Count()
if err != nil {
panic(err)
}
fmt.Printf("员工总数: %d\n", total)
// 带条件的 COUNT
youngCount, _ := empTbl.Where("Age", "<", int64(25)).Count()
fmt.Printf("25岁以下员工: %d 人\n", youngCount)
// 最小年龄
minAgeRows, _ := empTbl.OrderBy("Age").Limit(1).Find()
if len(minAgeRows) > 0 {
fmt.Printf("最小年龄: %d岁 (%s)\n", minAgeRows[0].Age, minAgeRows[0].Name)
}
// 最大分数
maxScoreRows, _ := empTbl.OrderBy("Score", true).Limit(1).Find()
if len(maxScoreRows) > 0 {
fmt.Printf("最高分数: %d分 (%s)\n", maxScoreRows[0].Score, maxScoreRows[0].Name)
}
// 计算平均分数
allEmps, _ := empTbl.Find()
var totalScore int64
for _, emp := range allEmps {
totalScore += emp.Score
}
avgScore := float64(totalScore) / float64(len(allEmps))
fmt.Printf("平均分数: %.2f\n", avgScore)
// 各部门平均分
deptMap := make(map[string][]int64)
for _, emp := range allEmps {
deptMap[emp.Dept] = append(deptMap[emp.Dept], emp.Score)
}
for dept, scores := range deptMap {
var sum int64
for _, s := range scores {
sum += s
}
avg := float64(sum) / float64(len(scores))
fmt.Printf(" %s: 平均 %.2f 分 (%d 人)\n", dept, avg, len(scores))
}
// 查询某员工的所有订单
employeeID := int64(1)
orders, _ := orderTbl.Where("EmployeeID", "=", employeeID).Find()
fmt.Printf("员工 %d 的订单数: %d\n", employeeID, len(orders))
for _, order := range orders {
fmt.Printf(" 订单ID=%d, 金额=%d, 状态=%d\n", order.ID, order.Amount, order.Status)
}
// 查询每个员工的订单总数
emps, _ := empTbl.Find()
for _, emp := range emps[:10] { // 只看前10名员工
orders, _ := orderTbl.Where("EmployeeID", "=", int64(emp.ID)).Find()
fmt.Printf("%-15s: %d 个订单\n", emp.Name, len(orders))
}
// 查询工程部员工的状态为"已完成"的订单
deptEmps, _ := empTbl.Where("Dept", "=", "工程部").Find()
var totalAmount int64
var orderCount int
for _, emp := range deptEmps {
orders, _ := orderTbl.
Where("EmployeeID", "=", int64(emp.ID)).
Where("Status", "=", int64(1)). // 1 = done
Find()
for _, order := range orders {
totalAmount += order.Amount
orderCount++
}
}
fmt.Printf("工程部已完成订单: %d 笔, 总金额: %d\n", orderCount, totalAmount)
// 禁用自动提交
db.SetAutoCommitDelay(0)
// 执行多个操作
empTbl.Insert(&Employee{Name: "A", Age: 25, Score: 90, Dept: "IT"})
empTbl.Insert(&Employee{Name: "B", Age: 30, Score: 85, Dept: "HR"})
empTbl.Delete(5)
// 一次性提交
if err := db.Commit(); err != nil {
fmt.Printf("提交失败: %v\n", err)
// 可以回滚或重试
} else {
fmt.Println("✅ 事务提交成功")
}
// 设置 500ms 防抖
db.SetAutoCommitDelay(500 * time.Millisecond)
// 连续插入,最后一次操作后 500ms 自动 Commit
for i := 1; i <= 100; i++ {
empTbl.Insert(&Employee{...})
}
// 等待自动提交
time.Sleep(1 * time.Second)
fmt.Println("数据已自动提交")
// 当脏页超过 500 页时自动触发 Commit
db.SetDirtyThreshold(500)
// 大量插入时会自动分批 Commit
for i := 1; i <= 100000; i++ {
empTbl.Insert(&Employee{...})
}
// 无需手动 Commit,系统会根据脏页阈值自动提交
TsDB2 提供三种预设配置,可根据业务场景选择:
// 方式1: 使用预设(最简单)
db.SetCommitPreset("batch") // 离线批量导入
db.SetCommitPreset("online") // 在线业务默认
db.SetCommitPreset("safe") // 金融/医疗关键数据
// 方式2: 手动配置
db.SetDirtyThreshold(500) // 脏页阈值(页数)
db.SetAutoCommitDelay(time.Second) // 防抖超时
| 预设 | 脏页阈值 | 内存占用 | 防抖超时 | 断电风险 | 适用场景 |
|---|---|---|---|---|---|
| batch | 500页 | 8MB | 1s | ~10,000行 | 离线导入、日志归档 |
| online | 100页 | 1.6MB | 500ms | ~2,000行 | Web应用、API服务 |
| safe | 20页 | 320KB | 100ms | ~400行 | 金融交易、医疗记录 |
离线批量导入(batch)
db, _ := DB.Open("import.db")
db.SetCommitPreset("batch") // 高吞吐
tbl, _ := DB.UseTable[Record](db)
for i := 1; i <= 1_000_000; i++ {
tbl.Insert(&Record{Name: fmt.Sprintf("Emp%07d", i)})
}
// 性能: ~28s 插入 100万行
// 风险: 断电最多丢失 1万行(可重跑脚本)
在线业务(online)
db, _ := DB.Open("app.db")
db.SetCommitPreset("online") // 平衡
tbl, _ := DB.UseTable[Order](db)
http.HandleFunc("/api/order", func(w http.ResponseWriter, r *http.Request) {
order := parseOrder(r)
tbl.Insert(order) // 每次请求插入
// 500ms 后自动落盘,断电最多丢失 2000行
})
关键数据(safe)
db, _ := DB.Open("finance.db")
db.SetCommitPreset("safe") // 最高安全
tbl, _ := DB.UseTable[Transaction](db)
func transfer(from, to string, amount float64) {
tbl.Insert(&Transaction{From: from, To: to, Amount: amount})
// 100ms 后自动落盘,断电最多丢失 400行
// 适合银行转账、医疗记录等关键业务
}
// 即使忘记 Commit,Close 时也会自动提交
db, _ := DB.Open("test.db")
tbl, _ := DB.UseTable[Employee](db, "test")
for i := 1; i <= 1000; i++ {
tbl.Insert(&Employee{...})
}
// 没有手动 Commit
db.Close() // ✅ Close 内部会调用 Commit,确保数据持久化
// 重新打开验证
db2, _ := DB.Open("test.db")
tbl2, _ := DB.UseTable[Employee](db2, "test")
count, _ := tbl2.Count()
fmt.Printf("Close 后数据保留: %d 条\n", count) // 输出: 1000 条
// V1: 初始版本
type EmployeeV1 struct {
DB.BaseModel
Name string `vrtx:"maxlen:32"`
Age int64
}
// V2: 添加新字段
type EmployeeV2 struct {
DB.BaseModel
Name string `vrtx:"maxlen:32"`
Age int64
Email string `vrtx:"maxlen:64"` // 新增字段
}
// TsDB2 自动处理 schema 版本迁移
tbl, _ := DB.UseTable[EmployeeV2](db, "employee")
// 标记字段为 dropped(不物理删除,保持兼容性)
empCore.DropField("Score") // Score 字段标记为已删除
// 查询时不再返回该字段
emp, _ := tbl.Find(1)
// emp.Score 将为默认值 0
// 删除整个表(回收所有页)
err := db.DropTable("order")
if err != nil {
fmt.Printf("删除表失败: %v\n", err)
}
// 验证删除
_, exists := db.CoreDB().FindTable("order")
if !exists {
fmt.Println("✅ order 表已删除")
}
// 查看所有索引
table, _ := db.CoreDB().FindTable("employee")
for _, idx := range table.GetIndexes() {
fmt.Printf("索引: %s, 类型: %d, 根页: %d\n",
idx.Name, idx.Type, idx.EntryPage)
}
// 删除索引
table.DropIndex("idx_age")
// 统计信息
fi, _ := os.Stat("mydb.db")
fmt.Printf("数据库大小: %.2f MB\n", float64(fi.Size())/1024/1024)
// 表统计
count, _ := empTbl.Count()
fmt.Printf("员工表行数: %d\n", count)
// BufferPool 统计
bp := db.CoreDB().GetBufferPool()
fmt.Printf("缓存页数: %d\n", bp.Len())
fmt.Printf("脏页数: %d\n", bp.DirtyCount())
BufferPool 是真正的 LRU + 读缓存:读/写命中都会刷新该页的热度,容量满时淘汰最久未使用的 干净页(脏页绝不淘汰,宁可临时超容量,靠脏页阈值触发的 Commit 回落);读未命中会把磁盘内容 作为干净页载入缓存,重复扫描/点查直接走内存。
二进制大对象只有一条存储路径:FieldType_Blob 列的行内是 8 字节引用
([TotalLen(4B) + FirstPageID(4B)]),内容落在链页上(PageType=BlobIndex,
PageHdr.NextPage 串接),单值上限 core.BlobChainMax = 256K 页 ≈ 4GB。
与 Dynamicbyte 的分界是"内容能不能跨页":后者行内 6B 堆引用、内容整块落一个堆页,
单值上限 core.MaxBlobSize ≈ 16KB;blob 列可以跨任意多页。
| 入口 | 你写什么 | 何时用 |
|---|---|---|
ORM []byte + vrtx:"type:blob" |
内容本身(ORM 自动落链,读回即内容) | 要与行一起走增删改查的小对象 |
ORM types.Blob(不写 tag 自动推断 blob 列) |
流式载体:Source io.Reader(文件/网络流直接进 struct),写侧 io.Copy 落链(内存峰值 ~128KB 与内容大小无关);读侧引擎只填回 8B 引用,row.Data.Open(tbl.BlobReader()) 按需拿 io.ReadCloser 流式消费 |
大对象(图片/视频/日志文件),不想把内容整块拉进内存 |
core Table.InsertRow / UpdateRow |
[]byte / string / io.Reader / types.Blob(不需要显式事务) |
迁移、手写 core 写入 |
core tx.BlobCommit → BlobRef(流式) |
io.Reader(可选 Snappy) |
超大对象、想自己保管引用 |
写进行内的引用由引擎随行回收(行删除 / 更新 / 表 drop 都会 BlobFree 旧链);
只有行外自管的引用(BlobCommit 的返回值没写进任何行)才需要你自己还链。
// ===== 独立引用 + 流式提交(core 层,文件/视频/备份等大对象)=====
tx := db.BeginTransaction()
// 提交: io.Reader 流式写入; io.Copy 内部 32KB buf + snappy 内部 ~64KB,
// 峰值内存 ≈ 128KB, 与文件大小无关
src, _ := os.Open("video.mp4")
defer src.Close()
ref, err := tx.BlobCommit(src, types.CompressSnappy) // types.CompressNone 或 CompressSnappy
// ref.TotalLen 原始字节数(解压后)
// ref.FirstPageID 链首页; (0,0) 表示空 blob, 不分配任何页
if err != nil { tx.Rollback(); return err }
if err := tx.Commit(); err != nil { return err }
// 读回: io.ReadAll → []byte (解压后), 长度 == ref.TotalLen
data, err := db.CoreDB().FindTable("files").BlobRead(ref, types.CompressSnappy)
if err != nil || uint32(len(data)) != ref.TotalLen {
return ErrBlobCorrupted
}
// 用完释放 (空 blob 是 no-op; 写进行内的引用不用你管)
db.CoreDB().FindTable("files").BlobFree(ref)
调用约定:
tx := db.BeginTransaction() + tx.Commit() 时中途失败可 tx.Rollback() 整链回收(新页都在 tx.allocated 里);不在事务里也走得通(db.BlobCommit,或直接往 blob 列写内容),代价是失败时已分配的页成为孤儿。compress 由调用方选择并保存:BlobRef 不携带压缩标志;BlobCommit 与 BlobRead 的 compress 必须一致(不一致 → core.ErrBlobCorrupted)。io.Reader.Read 返错 → 函数立即返错;已分配链页仍归本事务所有,tx.Rollback() 走 tx.allocated 路径整链回收;tx.Commit() 之后再丢,则需 tbl.BlobFree(ref) 显式还链。BlobChainMax = 256 * 1024 页 ≈ 4GB;超限返 core.ErrBlobTooLarge。行外引用要显式释放(行内引用不用):
row, _ := tbl.Find(id)
blobRef, _ := blobFromRow(row) // 你自己存到某列里的 BlobRef
tbl.Delete(id) // 行内 blob 列的链由引擎随行回收
tbl.BlobFree(blobRef) // 行外的独立句柄: 引擎不知道它还在
// ===== blob 列(ORM,与行一起走增删改查)=====
type File struct {
DB.BaseModel
Name string `vrtx:"maxlen:128"`
Cover []byte `vrtx:"type:blob"` // 载体一: []byte —— 一次性, 读回即内容本身
Data types.Blob // 载体二: 流式 —— 不写 tag 也自动推断为 blob 列
}
// []byte 载体: 增/改/查 与普通列无异; 索引不允许 (ErrIndexBlobField)
_ = tbl.Insert(&File{Name: "avatar.jpg", Cover: jpegBytes})
// 读回来就是内容本身: obj.Cover 是 []byte, 不是引用
// 流式载体: 写侧 Source 是任意 io.Reader (文件/网络流直接进 struct)
f, _ := os.Open("video.mp4")
_ = tbl.Insert(&File{Name: "video.mp4", Data: types.Blob{Source: f}})
// 读侧引擎只填回 8B 引用, 内容仍在链页; 按需开流消费, 不整块进内存
got, _ := tbl.Find(id)
rc, _ := got.Data.Open(tbl.BlobReader()) // io.ReadCloser
io.Copy(dst, rc)
// Insert/Update 成功后引用自动回填进对象; 之后只改普通字段的 Update 不会抹掉 blob 链
// 单值超 ~4GB: 写入报错 ErrBlobTooLarge, 含具体上限与写下的字节数
完整演示:go run ./demo/blob_stream(core 层 5 场景:空流 / 8KB 不压缩 / 1MB Snappy / 网络中断回滚 / 100MB 随机流)、go run ./demo/blob_orm(ORM 层 6 场景:32MB 流式写 / 流式读 CRC / []byte 对照 / 改名 Update 保链 / 重开持久化 / 删行归还链页)。
// TsDB2 不是线程安全的,需要在应用层加锁
var mu sync.Mutex
go func() {
mu.Lock()
defer mu.Unlock()
empTbl.Insert(&Employee{...})
}()
go func() {
mu.Lock()
defer mu.Unlock()
empTbl.Find(1)
}()
内嵌表基类之后,每个记录自己就带一套增删改查(父类定义方法、子类决定主键样式), 适合"读出来 → 改几个字段 → 写回去"的写法:
emp := empTbl.New() // New: 创建 + 绑定(知道自己属于哪张表)
emp.Name, emp.Age = "alice", 25
emp.Insert() // 增:主键为 0 时引擎发号并回写 emp.ID
emp.Age = 26
emp.Update() // 改:按 emp.ID 整行覆盖(未赋值字段会清零)
emp.Reload() // 查:按 emp.ID 重新读进自己
// 想要一个类型安全的新对象请用 empTbl.Find(id) —— 头部方法拿不到类型参数 T
emp.Delete() // 删:按 emp.ID
Insert / Find / FindAll / Select / Where(...).Find() 都会顺带绑定。
手工 &Employee{...} 构造且第一次动作就是 Update / Delete / Reload 时,先 empTbl.Bind(obj)。DB.ErrModelNotBound(不会崩、也不会写错表)。dup := *emp 之后 dup.Update() 更新的是 emp。
需要独立对象用 New + Load。vrtx:"pk" 换成表自己的字段(见 1.1),头部方法随之按那一列定位:
u.No = 7; u.Insert() 之后 u.Update() / u.Reload() / u.Delete() 都按 No 生效。| 基类 | 一级(行存储)索引 | 默认键列 | 基类字段 |
|---|---|---|---|
DB.BaseModel / DB.PrimaryModel(等价) |
Primary(B+ 树,聚簇) |
_id |
ID uint64 |
DB.TimeSeriesModel |
Time(分段) |
_ts |
TS uint64 |
BaseModel 与 TimeSeriesModel 报 DB.ErrInvalidPrimaryKey。ParseModel / UseTable 报 DB.ErrNotModel。
单写 vrtx:"pk" 不能替代基类 —— 它的语义是"覆盖基类默认的键列"。migrate 包(见 MIGRATION_GUIDE.md)走 core 层
db.CreateTable,不受这条约束:它建表时根本没有 Go 结构体。package main
import (
"git.ycl.pub/yl32940/TsDB2/DB"
"fmt"
"time"
)
type Employee struct {
DB.BaseModel
Name string `vrtx:"maxlen:32"`
Age int64 `vrtx:"index:idx_age"` // B+ 树二级索引
Score int64 `vrtx:"index:idx_score"` // B+ 树二级索引
Dept string `vrtx:"maxlen:16,index:idx_dept"`
}
func main() {
// 1. 打开数据库
db, _ := DB.Open("company.db")
defer db.Close()
db.SetAutoCommitDelay(0) // 手动控制
// 2. 建表 + 索引(UseTable 会按 vrtx tag 自动创建全部索引)
tbl, _ := DB.UseTable[Employee](db) // ✅ 无需传递 "employee"
// 3. 批量插入
fmt.Println("插入数据...")
depts := []string{"工程部", "销售部", "人事部", "财务部", "运维部"}
for i := 1; i <= 1000; i++ {
tbl.Insert(&Employee{
Name: fmt.Sprintf("Emp%04d", i),
Age: int64(20 + (i % 50)),
Score: int64(50 + (i % 51)),
Dept: depts[i%5],
})
}
db.Commit()
// 4. 查询
fmt.Println("\n=== 查询示例 ===")
// 4.1 主键查询
emp, _ := tbl.Find(1)
fmt.Printf("ID=1: %s, %d岁\n", emp.Name, emp.Age)
// 4.2 条件查询
young, _ := tbl.Where("Age", "<", int64(25)).Find()
fmt.Printf("25岁以下: %d 人\n", len(young))
// 4.3 排序 + 分页
top10, _ := tbl.OrderBy("Score", true).Limit(10).Find()
fmt.Println("分数Top 10:")
for i, e := range top10 {
fmt.Printf(" [%d] %s: %d分\n", i+1, e.Name, e.Score)
}
// 4.4 聚合
total, _ := tbl.Count()
fmt.Printf("总人数: %d\n", total)
// 5. 更新(Update 为整行覆盖:未赋值字段会被清零,因此先取出整行再改)
fmt.Println("\n=== 更新示例 ===")
row, _ := tbl.Find(1)
row.Score = 100
tbl.Update(row)
emp, _ = tbl.Find(1)
fmt.Printf("ID=1 更新后分数: %d\n", emp.Score)
// 6. 删除
fmt.Println("\n=== 删除示例 ===")
tbl.Delete(2)
_, err := tbl.Find(2)
if err != nil {
fmt.Println("ID=2 已删除")
}
// 7. 复杂查询
fmt.Println("\n=== 复杂查询 ===")
itHighScore, _ := tbl.
Where("Dept", "=", "工程部").
Where("Score", ">", int64(80)).
Find()
fmt.Printf("工程部高分员工: %d 人\n", len(itHighScore))
fmt.Println("\n✅ 所有操作完成!")
}
祝使用愉快! 🎉