TsDB2/DB/Table.go

package DB

import (
	"git.ycl.pub/yl32940/TsDB2/DB/core"
	"git.ycl.pub/yl32940/TsDB2/DB/core/types"
	"encoding/binary"
	"fmt"
	"time"
)

// Table 泛型表封装。
//
// T 必须内嵌 DB.BaseModel / DB.PrimaryModel / DB.TimeSeriesModel —— 由 UseTable 在
// 建表之前强制校验 (见 types.go 的 requireBase), 所以拿到一个 *Table[T] 就意味着
// T 一定有对象头部的增删改查。
//
// db 是反向指针, ORM Insert/Update/Delete 自动开 tx (tx.BlobCommit) 时用来从 ORM 层
// 拿 *core.Transaction。UseTable 在创建时填充, 之后只读。
type Table[T any] struct {
	db            *DB
	coreTable     *core.Table
	model         *ModelInfo
	name          string          // 表名
	activeColumns map[string]bool // 当前 DB schema 活跃列集合
}

// refreshActiveColumns 从 core table 刷新活跃列集合
func (t *Table[T]) refreshActiveColumns() {
	t.activeColumns = make(map[string]bool)
	for _, name := range t.coreTable.GetActiveFieldNames() {
		t.activeColumns[name] = true
	}
}

// SchemaColumns 返回当前 DB 中的所有活跃列名
func (t *Table[T]) SchemaColumns() []string {
	names := make([]string, 0, len(t.activeColumns))
	for name := range t.activeColumns {
		names = append(names, name)
	}
	return names
}

// Insert 插入一条记录
//
// 顺带把对象绑定到本表, 这样即使对象是外部构造的, 插完之后也能直接用头部的
// Update / Delete / Reload (见 model.go); 同时把引擎分配的行号带回对象, 之后
// obj.Update() / obj.Delete() 才能精确到"这一行"。
//
// Blob 列的落链由 t.db.coreDB (无事务版 BlobCommitter) 承接: 用户不需要手动开事务,
// 写入的内容照样自动落链; 用户若已显式开事务, 链页会经 DB.NewPage 登记进该事务,
// Rollback 整链回收。这里【不】自动 begin-tx + 立即 Commit —— tx.Commit 只要缓冲池
// 里有脏页 (单行插入必然有) 就走完整 WAL + fsync (~37ms/行), 逐行插入会比防抖提交
// 慢三个数量级, 且 SetCommitPreset 的批量策略被整体绕过 (a56b71d 的性能回退根因)。
func (t *Table[T]) Insert(obj *T) error {
	t.bindModel(obj)

	// 主键显式给了就用它; 没给 (==0) 就向引擎要一个自增号。
	// 旧实现无条件用 TotalRows+1 覆盖: 一是调用方填的 _id 被改写 → "按 _id 插入的行
	// 按 _id 查不到"; 二是删除会让行数回退 → 号回退并撞上仍存活的 pkID (探针 T4)。
	// 现在号一律来自 core 的只增不减游标, 拿到号后再序列化, 保证主键列与 pkID 一致。
	// 主键字段可以是内嵌 BaseModel 的 ID, 也可以是用户用 vrtx:"pk" 指定的普通字段。
	//
	// 时序表是唯一的例外, 见下面 IsTimeSeries 分支。
	pkID := uint64(0)
	if t.model.IsTimeSeries {
		// 时间键不是行号, 两者必须分开: 同一个时间戳下允许有多行 (同一时刻的多个
		// 测点 / 多次事件), 这是时序数据的常态。若沿用"时间键即行号", 第二行会被
		// 当成重复主键拒掉 —— 数据直接丢, 而这正是引入行号要解决的问题。
		// 所以时序表永远向引擎要一个新行号, 时间键保持用户给的原值, 只当索引键用。
		pkID = t.coreTable.AllocPKID()
		if t.model.PKInStruct {
			ts, err := getPKValue(obj, t.model)
			if err != nil {
				return fmt.Errorf("tsdb: insert: %w", err)
			}
			if ts == 0 {
				// 完全没给时间键: 补一个"写入这一刻"的纳秒级时间戳。
				//
				// 时序键就是时间戳本身, 默认值必须与显式传入的时间戳同量纲 (纳秒),
				// 否则它会成为一个远小于真实时间戳的怪值, 顺序写入下立刻被判为乱序。
				// 同一纳秒内连插多行的 TS 会相同, 这是允许的 —— 同刻多行由行号区分。
				setPKValue(obj, t.model, uint64(time.Now().UnixNano()))
			}
		}
	} else if t.model.PKInStruct {
		var err error
		if pkID, err = getPKValue(obj, t.model); err != nil {
			return fmt.Errorf("tsdb: insert: %w", err)
		}
		if pkID == 0 {
			pkID = t.coreTable.AllocPKID()
			setPKValue(obj, t.model, pkID)
		}
	}

	fvs, err := structToFieldValues(obj, t.model, t.activeColumns, t.db.coreDB)
	if err != nil {
		return fmt.Errorf("tsdb: insert: %w", err)
	}

	// pkID == 0 (结构体没有主键字段) 时由 core 自行分配
	assigned, err := t.coreTable.InsertRowWithID(pkID, fvs)
	if err != nil {
		return fmt.Errorf("tsdb: insert: %w", err)
	}
	setRowIDOf(obj, assigned)
	backfillBlobRefs(obj, t.model, fvs)
	return nil
}

// BatchInsert 批量插入
//
// 真批量: 走 core 的 InsertRowsWithIDs —— 索引层按"排序灌叶"写入, 数据页/树页
// 落盘按页摊销, saveTableInfo 与防抖提交通知只发一次。灌数吞吐相对逐行 Insert
// 提升一个数量级 (逐行路径每行 4 棵树各下降一次 + 每次页保存整页回写)。
//
// 两种情形退回逐行 Insert, 语义与旧实现完全一致:
//   - 时序表: assertTimeOrder 逐行把关写入顺序;
//   - 任一对象显式携带主键: 逐行查重路径 (显式主键本身就是小众场景)。
func (t *Table[T]) BatchInsert(objs []*T) error {
	if len(objs) == 0 {
		return nil
	}
	if t.model.IsTimeSeries {
		for _, obj := range objs {
			if err := t.Insert(obj); err != nil {
				return err
			}
		}
		return nil
	}

	// 先扫一遍: 有显式主键就整批退回逐行 (避免"半批半单"两条语义)
	if t.model.PKInStruct {
		anyExplicit := false
		for _, obj := range objs {
			t.bindModel(obj)
			if pkID, err := getPKValue(obj, t.model); err == nil && pkID != 0 {
				anyExplicit = true
				break
			}
		}
		if anyExplicit {
			for _, obj := range objs {
				if err := t.Insert(obj); err != nil {
					return err
				}
			}
			return nil
		}
	}

	pkIDs := make([]uint64, len(objs))
	fvs := make([][]types.FieldValue, len(objs))

	// 事务边界: 用户已持有活跃事务 (显式 BeginTransaction) 就整体复用 —— 既不顶掉
	// 用户事务、也不在批量中途替用户全局提交 (那样用户随后的 Rollback 会失去原子性);
	// 没有活跃事务才自开自提。core.BeginTransaction 对"已有活跃事务再开"直接 panic,
	// 所以必须先查 HasActiveTransaction, 顺序不能反。复用期间 blob 列传 coreDB 版
	// committer: 有活跃事务时链页经 NewPage 自动登记进该事务, Rollback 一样整链回收。
	ownTx := false
	var tx *core.Transaction
	var blobc types.BlobCommitter = t.db.coreDB
	if !t.db.HasActiveTransaction() {
		tx = t.db.BeginTransaction()
		ownTx = true
		blobc = tx
	}
	for i, obj := range objs {
		t.bindModel(obj)
		// 号先拿到、填进对象再序列化: 行里存的 _id 与引擎行号才一致 (与单行同约定)
		pkID := t.coreTable.AllocPKID()
		if t.model.PKInStruct {
			setPKValue(obj, t.model, pkID)
		}
		pkIDs[i] = pkID
		fv, err := structToFieldValues(obj, t.model, t.activeColumns, blobc)
		if err != nil {
			if ownTx {
				_ = tx.Rollback()
			}
			return fmt.Errorf("tsdb: batch insert: %w", err)
		}
		fvs[i] = fv
	}

	// 号全部来自 AllocPKID 预留号段: 走 InsertRowsReserved 跳过逐行查重
	// (走 WithIDs 会被当显式主键逐行 rowStoreHasPK —— 正是单行注释里警告过的
	// O(N²) 退化, 5 万行实测多烧 1.2GB 分配)
	assigned, err := t.coreTable.InsertRowsReserved(pkIDs, fvs)
	if err != nil {
		if ownTx {
			_ = tx.Rollback()
		}
		return fmt.Errorf("tsdb: batch insert: %w", err)
	}
	if ownTx {
		if cerr := tx.Commit(); cerr != nil {
			return fmt.Errorf("tsdb: batch insert: tx commit: %w", cerr)
		}
	}
	for i, obj := range objs {
		setRowIDOf(obj, assigned[i])
		backfillBlobRefs(obj, t.model, fvs[i])
	}
	return nil
}

// Update 更新一条记录
//
// 优先按对象携带的引擎行号定位 (从表里读出来或插入过的对象都有), 这样同一时间戳
// 下有多行时改的确实是"自己那一行"; 行号未知时退回按主键列定位 (旧行为)。
//
// Blob 列的落链与旧链回收都由引擎在 UpdateRow 内完成 (新链走 coreDB 的无事务
// BlobCommitter, 旧链走 deferred free), 与 Insert 同一套边界 —— 不自动开 tx。
func (t *Table[T]) Update(obj *T) error {
	t.bindModel(obj)

	fvs, err := structToFieldValues(obj, t.model, t.activeColumns, t.db.coreDB)
	if err != nil {
		return fmt.Errorf("tsdb: update: %w", err)
	}

	if id := rowIDOf(obj); id != 0 {
		if uerr := t.coreTable.UpdateRowByPK(id, fvs); uerr != nil {
			return fmt.Errorf("tsdb: update: %w", uerr)
		}
		backfillBlobRefs(obj, t.model, fvs)
		return nil
	}

	pk, err := getPKValue(obj, t.model)
	if err != nil {
		return fmt.Errorf("tsdb: update: %w", err)
	}
	if pk == 0 {
		return fmt.Errorf("tsdb: update: primary key is 0")
	}

	if uerr := t.coreTable.UpdateRow(t.pkIndexName(), encodePKKey(pk), fvs); uerr != nil {
		return fmt.Errorf("tsdb: update: %w", uerr)
	}
	backfillBlobRefs(obj, t.model, fvs)
	return nil
}

// Delete 按主键删除一条记录
//
// 引擎走 heapReleaseRow + releaseRowBlobRefs 回收堆块与 blob 链, 登记进 deferred
// free 队列 (有活跃事务时随 tx, 无事务时由 Commit/Close 收尾) —— 不自动开 tx。
func (t *Table[T]) Delete(id uint64) error {
	return t.coreTable.DeleteRow(t.pkIndexName(), encodePKKey(id))
}

// convertRow 反序列化一行 core 值 → *T, 并回写行号 + 绑定表。
// Find / Scan / Select 三个入口共用。
func (t *Table[T]) convertRow(pkID uint64, row []types.FieldValue) (*T, error) {
	obj := new(T)
	if err := fieldValuesToStruct(row, t.model, obj, t.coreTable); err != nil {
		return nil, err
	}
	setRowIDOf(obj, pkID)
	t.bindModel(obj)
	return obj, nil
}

// Find 按主键查找一条记录
func (t *Table[T]) Find(id uint64) (*T, error) {
	// SearchByIndexWithPK 与 SearchByIndex 取的是同一行 (同一条索引键 → 同一次定位),
	// 只是额外把行的引擎行号带回来 —— 没有它就没法对同刻多行做精确的改 / 删。
	pkID, row, err := t.coreTable.SearchByIndexWithPK(t.pkIndexName(), encodePKKey(id))
	if err != nil {
		return nil, err
	}

	obj := new(T)
	if err := fieldValuesToStruct(row, t.model, obj, t.coreTable); err != nil {
		return nil, err
	}
	setRowIDOf(obj, pkID)
	t.bindModel(obj)
	return obj, nil
}

// FindAll 返回所有记录
func (t *Table[T]) FindAll() ([]*T, error) {
	return t.Query().Find()
}

// Count 返回总行数
func (t *Table[T]) Count() (int64, error) {
	return t.coreTable.Query().Count()
}

// Select 流式遍历所有记录
func (t *Table[T]) Select(callback func(*T) bool) error {
	var convErr error
	err := t.coreTable.Query().ScanWithPKID(func(pkID uint64, row []types.FieldValue) bool {
		obj, err := t.convertRow(pkID, row)
		if err != nil {
			// 反序列化失败: 停止扫描并让 Select 返回错误, 不能静默跳过该行
			if convErr == nil {
				convErr = err
			}
			return false
		}
		return callback(obj)
	})
	if err != nil {
		return err
	}
	return convErr
}

// Where 开始条件查询
func (t *Table[T]) Where(field string, op string, value interface{}) *Query[T] {
	q := &Query[T]{table: t}
	q.addWhere(field, op, value)
	return q
}

// OrderBy 排序
func (t *Table[T]) OrderBy(field string, desc ...bool) *Query[T] {
	q := &Query[T]{table: t}
	q.OrderBy(field, desc...)
	return q
}

// Limit 限制返回行数
func (t *Table[T]) Limit(n int) *Query[T] {
	q := &Query[T]{table: t}
	q.Limit(n)
	return q
}

// Offset 跳过行数
func (t *Table[T]) Offset(n int) *Query[T] {
	q := &Query[T]{table: t}
	q.Offset(n)
	return q
}

// Query 创建一个新的查询构建器
func (t *Table[T]) Query() *Query[T] {
	return &Query[T]{table: t}
}

// BlobReader 返回本表的 blob 读取入口 (core.Table, 实现了 types.BlobReader)。
//
// 用途: 流式载体 types.Blob 的读回 —— row.Body.Open(tbl.BlobReader()) 拿
// io.ReadCloser, 按流消费 (io.Copy / 逐块), 不把内容整块拉进内存; 适合大对象。
// []byte 载体不需要它 (ORM 读回时自动解链)。
func (t *Table[T]) BlobReader() types.BlobReader {
	return t.coreTable
}

func (t *Table[T]) pkIndexName() string {
	return pkIndexPrefix + t.name
}

// ===== 模型绑定: 头部增删改查的入口 =====

// Bind 把对象绑定到本表 —— 绑定之后才能用对象头部的 Insert / Update / Delete /
// Save / Load / Reload。Insert / Update / Find / FindAll / Select / 条件查询的返回值
// 都会自动绑定, 所以只有"外部手工构造、且第一件事就是 Update/Delete/Reload"的对象
// 需要显式绑一次 (那种情况下用 New 更省事)。
//
// 注意: 绑定信息 (属于哪张表、外层对象是谁) 存在对象内部, 因此
//
//	u2 := *u      // 拷贝一个已绑定的对象
//	u2.Update()   // 更新的是 u, 不是 u2
//
// 会更新原对象。需要一份独立的对象请用 Table.New + Load, 或 Table.Find。
func (t *Table[T]) Bind(obj *T) error {
	if !t.bindModel(obj) {
		// UseTable 已经把没有基类的 T 挡在门外, 走到这里只可能是 T 的绑定器被特殊的
		// 嵌入方式挡住了 (例如把基类嵌在一个指针结构体里)。给同样的错误, 别让它静默失效。
		return fmt.Errorf("tsdb: %w: %T 需要内嵌 DB.BaseModel / DB.PrimaryModel / "+
			"DB.TimeSeriesModel 才能使用头部方法", ErrNotModel, obj)
	}
	return nil
}

// New 创建一个已经绑定到本表的新对象, 随后可以直接 obj.Load(pk) 读入。
//
// 与"手写 &T{} 再 Bind"的区别: 后者要先拿到 T 的类型; 在泛型代码里 (不知道 T 的具体
// 类型时) 只有 Table 能造出 T, New 就是那个入口。
func (t *Table[T]) New() *T {
	obj := new(T)
	t.bindModel(obj)
	return obj
}

// bindModel 把"这个对象属于哪张表"写回对象内嵌的表基类, 返回是否绑定成功。
//
// 走接口断言而不是反射: 每条被读写的记录都要过一遍, 反射在这里是白付的成本。
// 内嵌的表基类会把 modelBinder 提升到 *T 上, 所以一次类型断言就够了。
func (t *Table[T]) bindModel(obj *T) bool {
	b, ok := any(obj).(modelBinder)
	if !ok {
		return false
	}
	b.bindTo(t, obj)
	return true
}

// ===== session: Table[T] 的类型擦除视图 =====
//
// 模型基类不认识 T, 头部方法只能把请求交回这里; 这一层认识 T, 于是把参数断言回 *T。
// 断言失败只可能是"把 A 表的对象交给了 B 表", 所以返回错误而不是 panic。

// sessionObj 断言出 *T, 并顺手重新绑定 —— 对象可能是刚从一个临时副本拷过来的。
func (t *Table[T]) sessionObj(op string, obj any) (*T, error) {
	o, ok := obj.(*T)
	if !ok {
		return nil, fmt.Errorf("tsdb: %s: 表 %s 收到了 %T, 类型不匹配", op, t.name, obj)
	}
	t.bindModel(o)
	return o, nil
}

func (t *Table[T]) insert(obj any) error {
	o, err := t.sessionObj("insert", obj)
	if err != nil {
		return err
	}
	return t.Insert(o)
}

func (t *Table[T]) update(obj any) error {
	o, err := t.sessionObj("update", obj)
	if err != nil {
		return err
	}
	return t.Update(o)
}

func (t *Table[T]) pkOf(obj any) (uint64, error) {
	o, err := t.sessionObj("pk", obj)
	if err != nil {
		return 0, err
	}
	return getPKValue(o, t.model)
}

// loadInto 把主键 pk 对应的整行读进 obj 指向的对象。
//
// 用 *o = *got 整块赋值而不是逐字段搬: 两边是同一个 T, 字段集合一致, 省掉一次反射。
// 代价是 got 的绑定信息也被拷了过来 (self 还指着 got), 所以紧接着必须重新绑定到 o。
func (t *Table[T]) loadInto(pk uint64, obj any) error {
	o, err := t.sessionObj("load", obj)
	if err != nil {
		return err
	}
	got, err := t.Find(pk)
	if err != nil {
		return err
	}
	*o = *got
	t.bindModel(o)
	return nil
}

// reload 重新读一遍 obj 所在的那一行。
//
// 行号已知 → 按行号读 (同刻多行也不会读串); 行号未知 → 退回按可见主键读。
func (t *Table[T]) reload(obj any) error {
	o, err := t.sessionObj("reload", obj)
	if err != nil {
		return err
	}
	if id := rowIDOf(o); id != 0 {
		return t.loadByRowID(id, o)
	}
	pk, err := getPKValue(o, t.model)
	if err != nil {
		return fmt.Errorf("tsdb: reload: %w", err)
	}
	return t.loadInto(pk, o)
}

// loadByRowID 把行号 pkID 对应的整行读进 obj 指向的对象。
func (t *Table[T]) loadByRowID(pkID uint64, obj any) error {
	o, err := t.sessionObj("load", obj)
	if err != nil {
		return err
	}
	got, err := t.findByRowID(pkID)
	if err != nil {
		return err
	}
	*o = *got
	t.bindModel(o)
	return nil
}

// findByRowID 按引擎行号读出一个新对象 (行号是内部句柄, 所以不对外暴露)。
func (t *Table[T]) findByRowID(pkID uint64) (*T, error) {
	row, err := t.coreTable.FindByPK(pkID)
	if err != nil {
		return nil, err
	}
	obj := new(T)
	if err := fieldValuesToStruct(row, t.model, obj, t.coreTable); err != nil {
		return nil, err
	}
	setRowIDOf(obj, pkID)
	t.bindModel(obj)
	return obj, nil
}

func (t *Table[T]) deleteByPK(pk uint64) error { return t.Delete(pk) }

// deleteByRowID 按引擎行号删除一行 —— 同刻多行时唯一能精确指认"这一行"的方式。
//
// 自动开 tx 与 Delete 同源: heapReleaseRow / releaseRowBlobRefs 都在 deferred.
//
// 事务边界同 BatchInsert: 用户已持有活跃事务则复用、绝不替用户提交 (否则
// 用户的 Rollback 会失去原子性); 无活跃事务才自开自提。必须先查
// HasActiveTransaction 再决定 —— core 对"已有活跃事务再开"直接 panic。
func (t *Table[T]) deleteByRowID(pkID uint64) error {
	ownTx := false
	var tx *core.Transaction
	if !t.db.HasActiveTransaction() {
		tx = t.db.BeginTransaction()
		ownTx = true
	}
	if err := t.coreTable.DeleteRowByPK(pkID); err != nil {
		if ownTx {
			_ = tx.Rollback()
		}
		return err
	}
	if ownTx {
		if err := tx.Commit(); err != nil {
			return fmt.Errorf("tsdb: delete: tx commit: %w", err)
		}
	}
	return nil
}

func (t *Table[T]) tableName() string { return t.name }

func encodePKKey(pk uint64) []byte {
	buf := make([]byte, 8)
	binary.BigEndian.PutUint64(buf, pk)
	return buf
}