TsDB2/DB/Table.go

package DB

import (
	"TsDB2/DB/core"
	"TsDB2/DB/core/types"
	"encoding/binary"
	"fmt"
)

// Table 泛型表封装
type Table[T any] struct {
	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 插入一条记录
func (t *Table[T]) Insert(obj *T) error {
	// 如果主键在结构体中且值为 0，预分配 pkID (与 core 的 TotalRows+1 保持一致)
	if t.model.PKInStruct && getPKValue(obj, t.model) == 0 {
		nextPK := t.coreTable.TotalRows() + 1
		setPKValue(obj, t.model, nextPK)
	}

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

	_, err = t.coreTable.InsertRow(fvs)
	if err != nil {
		return fmt.Errorf("tsdb: insert: %w", err)
	}
	return nil
}

// BatchInsert 批量插入
func (t *Table[T]) BatchInsert(objs []*T) error {
	for _, obj := range objs {
		if err := t.Insert(obj); err != nil {
			return err
		}
	}
	return nil
}

// Update 按主键更新一条记录
func (t *Table[T]) Update(obj *T) error {
	pk := getPKValue(obj, t.model)
	if pk == 0 {
		return fmt.Errorf("tsdb: update: primary key is 0")
	}

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

	return t.coreTable.UpdateRow(t.pkIndexName(), encodePKKey(pk), fvs)
}

// Delete 按主键删除一条记录
func (t *Table[T]) Delete(id uint64) error {
	return t.coreTable.DeleteRow(t.pkIndexName(), encodePKKey(id))
}

// Find 按主键查找一条记录
func (t *Table[T]) Find(id uint64) (*T, error) {
	row, err := t.coreTable.SearchByIndex(t.pkIndexName(), encodePKKey(id))
	if err != nil {
		return nil, err
	}

	obj := new(T)
	if err := fieldValuesToStruct(row, t.model, obj); err != nil {
		return nil, err
	}
	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 {
	return t.coreTable.Query().Scan(func(row []types.FieldValue) bool {
		obj := new(T)
		if err := fieldValuesToStruct(row, t.model, obj); err != nil {
			return true
		}
		return callback(obj)
	})
}

// 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}
}

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

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