| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430 |
- package core
- import (
- "encoding/binary"
- "sync"
- "unsafe"
- "git.ycl.pub/yl32940/TsDB2/DB/core/page"
- "git.ycl.pub/yl32940/TsDB2/DB/core/types"
- "git.ycl.pub/yl32940/TsDB2/DB/core/util"
- )
- // SchemaFragHdr 共享扩展页里每个 schema 续块的头 (定长 8 字节, 紧随 payload)
- // 警告: 该结构体仅用于计算大小/表达语义, 写盘一律用 putSchemaFragHdr,
- // 不要做 `*(*SchemaFragHdr)(p) = v` 的结构体赋值。
- type SchemaFragHdr struct {
- NextPage uint32 // 下一个续块所在页 (0 = 本表续块链结束)
- NextOff uint16 // 下一个续块在页内的物理偏移
- DataLen uint16 // 本续块 payload 字节数
- }
- // schemaFragHdrSize schema 续块头的物理大小 (字节)
- const schemaFragHdrSize = 8
- // putSchemaFragHdr 写入续块头 (大端: NextPage 4B + NextOff 2B + DataLen 2B)
- func putSchemaFragHdr(buf []byte, off int, h SchemaFragHdr) {
- binary.BigEndian.PutUint32(buf[off:off+4], h.NextPage)
- binary.BigEndian.PutUint16(buf[off+4:off+6], h.NextOff)
- binary.BigEndian.PutUint16(buf[off+6:off+8], h.DataLen)
- }
- // readSchemaFragHdr 读取续块头
- func readSchemaFragHdr(buf []byte, off int) SchemaFragHdr {
- return SchemaFragHdr{
- NextPage: binary.BigEndian.Uint32(buf[off : off+4]),
- NextOff: binary.BigEndian.Uint16(buf[off+4 : off+6]),
- DataLen: binary.BigEndian.Uint16(buf[off+6 : off+8]),
- }
- }
- // SchemaVersion 表的一个 schema 版本定义
- type SchemaVersion struct {
- Version byte
- FieldIndices []byte // 每个字段在 masterFields 中的索引
- }
- type TableInfo struct {
- Name [TableNameSize]byte
- Flag byte // 表的状态是否被删除
- FieldCount byte // 当前字段的总数量(包含已删除,永不减少)byte=255上限
- ActiveFieldCount byte // 实际活跃字段数量(不包含已删除)
- _ byte // 1 显式对齐填充 (TotalRows 需要 8 字节对齐)
- // HeapPageCount 本表堆链 (FirstHeap→LastHeap) 上的页数。
- //
- // 塞进 TotalRows 前的对齐填充里 (那 4 字节本来就是编译器要插入的空洞), 不额外
- // 增加 TableInfo 的体积。
- // 用途: DropTable 把整条堆链 O(1) 拼进空闲链时, 不遍历也有链长可累加空闲页计数。
- HeapPageCount uint32 // 4 堆链页数
- TotalRows uint64 // 当前表行的总数量
- FirstHeap uint32 // 堆链首页
- LastHeap uint32 // 堆链尾页
- FirstData uint32 // 数据链首页
- LastData uint32 // 数据链尾
- RowSize uint16 // 单条记录大小(字节)
- FieldAllocateSize uint16 // 当前列总分配大小
- IndexCount byte // 这张表有几个索引
- FirstIndexID uint32 // 本表第一个索引的元数据位置
- LastIndexID uint32 // 本表最后一个索引的元数据位置
- // ExtRef schema 续块在「全库共享扩展页」中的位置
- // (复用原 FirstOverflow/LastOverflow 占用的 8 字节: ExtPage 落在 8 字节对齐处,
- // 尾部 2 字节显式填充, 使 TableInfo 整体大小为 8 的倍数)
- ExtPage uint32 // 4 首个续块所在扩展页 ID (0 = schema 全部内联, 无续块)
- ExtOff uint16 // 2 首个续块在页内的物理偏移
- _ uint16 // 2 显式填充, 使整体大小为 8 的倍数
- }
- // 编译期断言: 钉死 TableInfo 的盘上大小与字段布局。
- //
- // TableNameSize 变化会整体平移 Name 之后所有字段的偏移 —— analyzer 那套硬编码偏移
- // (analyzer/main.go 的 TI_*) 必须同步修改, 改漏了会静默读错数据。这里把大小钉死,
- // 让「改了布局却忘了改断言」直接编译失败。
- //
- // 布局 (TableNameSize=48) = 104 字节:
- //
- // Name[48] + Flag/FieldCount/ActiveFieldCount/pad(4) + HeapPageCount(4) + TotalRows(8)
- // + FirstHeap/LastHeap/FirstData/LastData(16) + RowSize/FieldAllocateSize(4)
- // + IndexCount/pad(4) + FirstIndexID/LastIndexID/ExtPage(12) + ExtOff/pad(4)
- var (
- _ [unsafe.Sizeof(TableInfo{}) - 104]struct{}
- _ [104 - unsafe.Sizeof(TableInfo{})]struct{}
- )
- // readSchema 读取一张表完整的 schema 定义
- // 数据来源: 表页内的固定内联块 (allocSize 字节) + 共享扩展页里的续块链
- // 新格式: [activeVer(1B)] [masterCount(1B)] [type+len+name]×masterCount [verCount(1B)] [verDefs...]
- func readSchema(handle page.HandleDb, pageID uint32, startOff uint16, allocSize uint16,
- extPage uint32, extOff uint16) (byte, []types.FieldInfo, []SchemaVersion) {
- // 1. 内联块 (表页内, 固定 allocSize 字节, 无跳转槽)
- var data []byte
- buf := make([]byte, PageSize)
- if err := handle.ReadPage(pageID, buf); err == nil {
- end := int(startOff) + int(allocSize)
- if end > len(buf) {
- end = len(buf)
- }
- if int(startOff) < end {
- data = append(data, buf[startOff:end]...)
- }
- }
- // 2. 续块链 (共享扩展页): [SchemaFragHdr 8B][payload]
- for pid, off := extPage, extOff; pid != 0; {
- fb := make([]byte, PageSize)
- if err := handle.ReadPage(pid, fb); err != nil {
- break
- }
- if int(off)+schemaFragHdrSize > len(fb) {
- break
- }
- frag := readSchemaFragHdr(fb, int(off))
- start := int(off) + schemaFragHdrSize
- end := start + int(frag.DataLen)
- if end > len(fb) {
- break
- }
- data = append(data, fb[start:end]...)
- pid, off = frag.NextPage, frag.NextOff
- }
- if len(data) < 2 {
- return 0, nil, nil
- }
- pos := 0
- activeVer := data[pos]
- pos++
- masterCount := int(data[pos])
- pos++
- // 读取 master fields: [type(1B)][flag(1B)][nameLen(1B)][name...]
- // flag 是列级标志 (删除/可空/唯一 + decimal 的 scale), 见 types.FieldInfo.Flag。
- // 语义变了但字节位置与顺序不变: 必须与 serializeSchema 严格对照。
- var masterFields []types.FieldInfo
- for i := 0; i < masterCount && pos+3 <= len(data); i++ {
- typ := types.FieldType(data[pos])
- flag := data[pos+1]
- nameLen := int(data[pos+2])
- pos += 3
- if pos+nameLen > len(data) {
- break
- }
- name := make([]byte, nameLen)
- copy(name, data[pos:pos+nameLen])
- pos += nameLen
- masterFields = append(masterFields, types.FieldInfo{FieldType: typ, FieldName: name, Flag: flag})
- }
- // 读取版本定义
- var versions []SchemaVersion
- if pos < len(data) {
- verCount := int(data[pos])
- pos++
- for i := 0; i < verCount && pos < len(data); i++ {
- verNo := data[pos]
- pos++
- if pos >= len(data) {
- break
- }
- fieldCount := int(data[pos])
- pos++
- indices := make([]byte, fieldCount)
- if pos+fieldCount <= len(data) {
- copy(indices, data[pos:pos+fieldCount])
- pos += fieldCount
- }
- versions = append(versions, SchemaVersion{Version: verNo, FieldIndices: indices})
- }
- }
- return activeVer, masterFields, versions
- }
- type Table struct {
- info *TableInfo
- Field []types.FieldInfo // master registry (所有字段,含已删除,删除标记见 types.FlagDeleted)
- // db 所属 DB (BlobRead/BlobFree 需要它取 file / pendingFreeChain).
- // 创建时由 db.tables.Create 注入, 运行时不变.
- db *DB
- handle page.HandleDb
- indexMeta *IndexMetaPage
- tablePageID uint32 // 表ID
- tableStartOffset uint32 // 表所在页的偏移
- // schema 续块所在的「全库共享扩展页」分配器 (由 TablePage 注入, 各表同一实例)
- schemaExt *SchemaExtAllocator
- // schema 版本管理
- activeVersion byte
- versions []SchemaVersion
- // 写操作通知回调 (防抖自动提交)
- onWrite func()
- // 自增主键游标 (内存态, 详见 AllocPKID/ensureNextPK):
- // 不能再现算 TotalRows+1 —— 删除会让行数回退, 编号随之回退并撞上存活的 pkID。
- nextPK uint64
- nextPKInit bool
- pendingPK uint64 // AllocPKID 刚预留、等待 InsertRowWithID 消费的号
- hasPendingPK bool
- // pkMu 保护自增主键游标的读-改-写 (ensureNextPK / nextPK++ / pendingPK 预留):
- // 多写者并发 AllocPKID 时, 无锁的 nextPK++ 会给两行发出同一个号 (重复主键)。
- pkMu sync.Mutex
- // fieldIdxCache: 列名 → table.Field 下标的惰性缓存。条件求值
- // (Eq/In/Gt... 的 Match) 每行每条件都查一次字段位置, 线性扫是 O(列数)/行;
- // 表字段在建模后基本不变, 缓存后 O(1)。Field 切片身份变化 (AddField 追加 /
- // ReloadInfos 整体换片) 时按首元素指针 + 长度自动重建, 无需手工失效。
- // fieldIdxMu 保护首次构建: 引擎允许并发读, 两个读 goroutine 可能同时建缓存。
- fieldIdxMu sync.Mutex
- fieldIdxCache map[string]int
- fieldIdxSrc []types.FieldInfo // 建缓存时的 Field 切片 (身份比对用)
- }
- // TotalRows 返回表的总行数
- func (table *Table) TotalRows() uint64 { return table.info.TotalRows }
- // TablePageID 返回表页 ID (调试用)
- func (table *Table) TablePageID() uint32 { return table.tablePageID }
- // TableStartOffset 返回表信息在页中的偏移 (调试用)
- func (table *Table) TableStartOffset() uint32 { return table.tableStartOffset }
- // IndexMeta 返回索引元数据页 (调试用)
- func (table *Table) IndexMeta() *IndexMetaPage { return table.indexMeta }
- // SetOnWrite 设置写操作通知回调 (防抖自动提交)
- func (table *Table) SetOnWrite(fn func()) { table.onWrite = fn }
- // findFieldIndex 查找字段在 master registry 中的索引
- //
- // 走 fieldIndexMap 的惰性缓存: Match 类条件是"每行每条件一次"的热路径, 线性扫
- // O(列数) 在宽表 + 全表扫描下会累积成可见开销。缓存按 Field 切片身份
- // (首元素指针 + 长度) 自动失效, AddField / ReloadInfos 后下一次调用即重建。
- func (table *Table) findFieldIndex(name string) int {
- if idx, ok := table.fieldIndexMap()[name]; ok {
- return idx
- }
- return -1
- }
- // fieldIndexMap 返回 (必要时重建) 列名 → Field 下标缓存。
- func (table *Table) fieldIndexMap() map[string]int {
- src := table.Field
- if table.fieldIdxCache != nil && len(table.fieldIdxSrc) == len(src) &&
- (len(src) == 0 || &table.fieldIdxSrc[0] == &src[0]) {
- return table.fieldIdxCache
- }
- table.fieldIdxMu.Lock()
- defer table.fieldIdxMu.Unlock()
- // 双检: 等待锁期间另一读线程可能已建好
- if table.fieldIdxCache != nil && len(table.fieldIdxSrc) == len(src) &&
- (len(src) == 0 || &table.fieldIdxSrc[0] == &src[0]) {
- return table.fieldIdxCache
- }
- m := make(map[string]int, len(src))
- for i := range src {
- m[string(src[i].FieldName)] = i
- }
- table.fieldIdxCache = m
- table.fieldIdxSrc = src
- return m
- }
- // fieldWidth 取某列的行内宽度 (字节) —— 引擎内所有行偏移计算的唯一入口。
- //
- // 为什么不能直接 types.GetSize(f.FieldType): 字符串列的行内宽度由【列自己声明】
- // (Flag 低 5 位的宽度码, 上限 types.MaxInlineColumnWidth), 只看类型会一律算成 256,
- // 于是该列之后的所有列都从错位偏移解码。这里统一经 FieldInfo.Width()。
- //
- // 返回 0 表示索引越界或列级参数非法 —— 调用方必须当错误处理: 0 宽度会让整行错位。
- func (table *Table) fieldWidth(masterIdx int) uint16 {
- if masterIdx < 0 || masterIdx >= len(table.Field) {
- return 0
- }
- return table.Field[masterIdx].Width()
- }
- func (table *Table) GetFieldCount() byte { return table.info.FieldCount }
- func (table *Table) GetActiveFieldCount() byte { return table.info.ActiveFieldCount }
- // GetActiveFieldNames 返回当前活跃版本中包含的所有字段名 (不含已删除)
- func (table *Table) GetActiveFieldNames() []string {
- sv := table.getVersion(table.activeVersion)
- if sv == nil {
- return nil
- }
- names := make([]string, 0, len(sv.FieldIndices))
- for _, idx := range sv.FieldIndices {
- if table.Field[idx].IsDropped() {
- continue // 跳过已删除
- }
- names = append(names, string(table.Field[idx].FieldName))
- }
- return names
- }
- // GetActiveFieldType 返回指定字段的 FieldType (基于 master registry)
- func (table *Table) GetActiveFieldType(name string) (types.FieldType, bool) {
- for _, f := range table.Field {
- if string(f.FieldName) == name && !f.IsDropped() {
- return f.FieldType, true
- }
- }
- return 0, false
- }
- // GetActiveFieldFlag 返回指定字段的列级标志 (含 Decimal 的 scale 与可空位)。
- //
- // 与 GetActiveFieldType 配对: 类型一致但 scale 不一致是"同列不同 scale"这种会被
- // 静默破坏保序前提的情况, 调用方 (ORM 重开表时) 必须能读到它来做一致性校验。
- func (table *Table) GetActiveFieldFlag(name string) (byte, bool) {
- for i := range table.Field {
- f := &table.Field[i]
- if string(f.FieldName) == name && !f.IsDropped() {
- return f.Flag, true
- }
- }
- return 0, false
- }
- func (table *Table) GetFieldAllocateSize() uint16 { return table.info.FieldAllocateSize }
- func (table *Table) GetFirstData() uint32 { return table.info.FirstData }
- // GetSchemaExtRef 返回本表首个 schema 续块的位置 (pageID, off); pageID == 0 表示无续块
- func (table *Table) GetSchemaExtRef() (uint32, uint16) {
- return table.info.ExtPage, table.info.ExtOff
- }
- func (table *Table) GetFields() []types.FieldInfo {
- return table.Field
- }
- func (table *Table) findFieldByName(name []byte) *types.FieldInfo {
- target := util.BytesToString(name)
- for i := range table.Field {
- if util.BytesToString(table.Field[i].FieldName) == target {
- return &table.Field[i]
- }
- }
- return nil
- }
- // fieldParam 取列级类型参数 (Flag 低 5 位): decimal 是 scale, 字符串家族是行内宽度码,
- // 其余类型为 0 (由 types.ValidateColumnParam 保证)。
- //
- // 以前这里叫 fieldScale 且只服务 decimal; 字符串列也用上同一格之后, "取参数"与
- // "算宽度"必须成对 —— 只传 scale 会漏掉宽度码, 编出来的字节宽度与行内不一致。
- func (table *Table) fieldParam(masterIdx int) byte {
- if masterIdx < 0 || masterIdx >= len(table.Field) {
- return 0
- }
- return table.Field[masterIdx].Param()
- }
- // AllocPKID 分配并预留下一个自增 pkID。
- //
- // 供"必须先把号写进对象、再序列化整行"的上层使用 (典型是 ORM:
- // 只有先拿到 pkID 填进 _id 列, 存进行里的 _id 才与真实 pkID 一致)。
- // 分配即推进游标并预留该号, 紧随其后的 InsertRowWithID 会跳过查重。
- // 若最终未插入, 该号只是被跳过 —— 游标只增不减, 不会与后续插入冲突。
- func (table *Table) AllocPKID() uint64 {
- // 整个"惰性初始化 → 取号 → 游标推进 → 预留登记"在锁内完成:
- // 并发写者各拿各的号, 号段互不重叠。
- table.pkMu.Lock()
- defer table.pkMu.Unlock()
- table.ensureNextPK()
- pk := table.nextPK
- table.nextPK++
- table.pendingPK = pk
- table.hasPendingPK = true
- return pk
- }
- // ensureNextPK 惰性初始化自增游标: 只增不减, 从行存储里当前最大的存活 pkID 之后继续。
- //
- // 旧实现 (TotalRows+1) 删除后会回退编号, 与仍存活的 pkID 撞车 (探针 T4)。
- // 首次使用时取一次当前最大 pkID —— 每次开表只算一次, 之后 O(1)。
- func (table *Table) ensureNextPK() {
- if table.nextPKInit {
- return
- }
- table.nextPK = table.maxRowStorePK() + 1
- if table.nextPK == 0 {
- table.nextPK = 1
- }
- table.nextPKInit = true
- }
- // maxRowStorePK 返回行存储索引中当前最大的存活 pkID (无行或没有行存储索引时为 0)。
- // DataPage 沿 B+ 树叶链按 pkID 升序排列, 故从最后一页往回找第一个非空页即可;
- // 每页内取最大 pkID 以兼容页内乱序的极端情况。
- func (table *Table) maxRowStorePK() uint64 {
- primMeta, primHandler, ok := table.indexMeta.rowStoreMeta()
- if !ok {
- return 0
- }
- ids := primHandler.GetDataPageIDs(table.handle, primMeta.EntryPage)
- for i := len(ids) - 1; i >= 0; i-- {
- rows := primHandler.ReadDataPageRows(table.handle, ids[i])
- var maxPK uint64
- for _, r := range rows {
- if len(r) < 8 {
- continue
- }
- if pk := binary.BigEndian.Uint64(r[:8]); pk > maxPK {
- maxPK = pk
- }
- }
- if maxPK != 0 {
- return maxPK
- }
- }
- return 0
- }
|