package core import ( "fmt" "git.ycl.pub/yl32940/TsDB2/DB/core/page" "git.ycl.pub/yl32940/TsDB2/DB/core/types" ) // ======================== 高层 CRUD ======================== // InsertRow 插入一行数据, pkID 由引擎按自增游标分配 (只增不减, 见 AllocPKID)。 func (table *Table) InsertRow(values []types.FieldValue) (uint64, error) { if err := table.errIfReadOnly(); err != nil { return 0, err } return table.InsertRowWithID(0, values) } // InsertRowWithID 插入一行数据, pkID == 0 表示由引擎自增分配。 // // 允许调用方指定主键: 上层 (ORM/导入) 往往已经拿到一个明确的行标识, 旧实现 // 无条件用 TotalRows+1 覆盖, 结果是"调用方以为按 _id 写进去了, 之后按 _id 却 // 查不到" —— 行确实存在, 只是不可达 (数据静默错位的一类)。 // // 指定的主键若已存在则报 ErrDuplicatePK, 避免同一 pkID 在主键树里出现两条记录。 // 例外: 引擎刚分配并"预留"的号 (AllocPKID) 必然唯一, 跳过这次查重 —— // 查重走的是回表 ResolveByPK, 会重建 O(页数) 的定位缓存, 而每次插入又使其失效; // 逐行插入将因此退化成 O(N²) (千万级写入从分钟级变成不可完成)。 func (table *Table) InsertRowWithID(pkID uint64, values []types.FieldValue) (uint64, error) { if err := table.errIfReadOnly(); err != nil { return 0, err } if pkID == 0 { pkID = table.AllocPKID() } reserved := table.hasPendingPK && table.pendingPK == pkID table.hasPendingPK = false if !reserved { if table.indexMeta.rowStoreHasPK(pkID) { return 0, fmt.Errorf("%w: pkID=%d", ErrDuplicatePK, pkID) } } // 显式指定更大的主键时把游标顶到它之后, 避免后续自增再发一次同样的号 if pkID+1 > table.nextPK { table.nextPK = pkID + 1 } rowData, err := table.serializeRow(values, nil) if err != nil { return 0, err } if err := table.indexMeta.InsertRow(table.Field, pkID, rowData, table.activeVersion, table.versions); err != nil { return 0, err } table.info.TotalRows++ table.saveTableInfo() if table.onWrite != nil { table.onWrite() } return pkID, nil } // InsertRowsWithIDs 批量插入。pkIDs[i]==0 的行由引擎自增分配 (号先拿到、写进 // _id 列后再序列化, 与 ORM 单行路径同一约定); pkIDs[i]!=0 为显式主键, 逐个走 // 查重。返回实际使用的 pkID 列表 (与入参等长)。 // // 与逐行 InsertRowWithID 的语义差异只在"摊销", 不在"语义": // - 自增号来自只增不减的游标, 整批天然唯一, 跳过逐行查重 (单行路径靠 // pendingPK 预留跳过同一件事; 批量下预留槽只有一格, 改为按"号段"整批豁免); // - 显式主键照旧 rowStoreHasPK 查重, 另补批内查重 (单行路径里批内冲突会 // 在第 i+1 行的查重中被发现, 这里一次性比出来); // - 序列化 (含动态堆写入) 仍逐行做 —— 堆页的写入粒度是页级的, 堆块追加 // 本身已经近似批量; saveTableInfo / onWrite 摊到批末一次。 // // 失败语义: 序列化阶段报错时, 先前行写入的堆块成为无主垃圾 (泄漏而非错乱, // 行数据并未落任何索引, 不会被读到); 索引阶段报错时部分索引已写入 —— 与 // 单行路径"写了一半报错"一致, 靠事务回滚兜底。 func (table *Table) InsertRowsWithIDs(pkIDs []uint64, values [][]types.FieldValue) ([]uint64, error) { if err := table.errIfReadOnly(); err != nil { return nil, err } return table.insertRows(pkIDs, values, false) } // InsertRowsReserved 批量插入"引擎预留号"的行: pkIDs 全部来自 AllocPKID // (ORM 先要号填进对象、再连同行数据一起送回来), 唯一性由只增不减的游标保证, // 逐行查重整个跳过 —— 语义同单行路径的 pendingPK 预留豁免, 只是预留号不止 // 一格, 按号段整批豁免。批内查重仍保留 (号段自身不可能重复, 这层是防御)。 // // 注意: 走这个入口的前提是调用方【没有】夹带任何非预留号 —— 夹带了请用 // InsertRowsWithIDs (显式号走逐行查重)。 func (table *Table) InsertRowsReserved(pkIDs []uint64, values [][]types.FieldValue) ([]uint64, error) { if err := table.errIfReadOnly(); err != nil { return nil, err } return table.insertRows(pkIDs, values, true) } func (table *Table) insertRows(pkIDs []uint64, values [][]types.FieldValue, allReserved bool) ([]uint64, error) { n := len(values) if n == 0 { return nil, nil } if len(pkIDs) != n { return nil, fmt.Errorf("tsdb: 批量插入: pkIDs (%d) 与 values (%d) 数量不一致", len(pkIDs), n) } // 1) 补齐自增号 auto := make([]bool, n) for i := 0; i < n; i++ { if pkIDs[i] == 0 { pkIDs[i] = table.AllocPKID() auto[i] = true } } table.hasPendingPK = false // 预留号已被本批整体消费 // 2) 批内主键查重 (显式 vs 显式, 以及显式撞上自增号段) { seen := make(map[uint64]struct{}, n) for i := 0; i < n; i++ { if _, dup := seen[pkIDs[i]]; dup { return nil, fmt.Errorf("%w: pkID=%d (批内重复)", ErrDuplicatePK, pkIDs[i]) } seen[pkIDs[i]] = struct{}{} } } // 3) 显式主键对既有行查重 (自增/预留号必然唯一, 跳过) if !allReserved { for i := 0; i < n; i++ { if auto[i] { continue } if table.indexMeta.rowStoreHasPK(pkIDs[i]) { return nil, fmt.Errorf("%w: pkID=%d", ErrDuplicatePK, pkIDs[i]) } } } // 4) 显式大号顶起自增游标 (与单行路径一致; 预留号本就来自游标, 无需) if !allReserved { for i := 0; i < n; i++ { if !auto[i] && pkIDs[i]+1 > table.nextPK { table.nextPK = pkIDs[i] + 1 } } } // 5) 逐行序列化 (动态堆写入在这一步发生) rowDatas := make([][]byte, n) for i := 0; i < n; i++ { rowData, err := table.serializeRow(values[i], nil) if err != nil { return nil, err } rowDatas[i] = rowData } // 6) 索引层批量写入 if err := table.indexMeta.InsertRows(table.Field, pkIDs, rowDatas, table.activeVersion, table.versions); err != nil { return nil, err } table.info.TotalRows += uint64(n) table.saveTableInfo() if table.onWrite != nil { table.onWrite() } return pkIDs, nil } // ======================== 按行号 (pkID) 直接读写 ======================== // // 索引键只决定"落在哪个分段/哪一批", 行号才唯一确定"是哪一行"。 // 时序表里同一个时间戳允许存在多行 (同一时刻的多个测点/多次事件), 而上面那组 // "按索引键取第一条"的入口只能碰到第一行 —— 其余行存得进、读不到、改不了、删不掉。 // 下面这组入口按行号直达, 供 ORM (与任何需要精确定位一行的上层) 使用。 // FindByPK 按行号读取一行的字段值。 func (table *Table) FindByPK(pkID uint64) ([]types.FieldValue, error) { storeRow, found := table.indexMeta.ResolveByPK(pkID) if !found || len(storeRow) < 8 { return nil, page.ErrNotFound } return table.deserializeRow(storeRow[8:]), nil } // DeleteRowByPK 按行号删除一行, 并级联清掉它在全部索引里的条目。 func (table *Table) DeleteRowByPK(pkID uint64) error { if err := table.errIfReadOnly(); err != nil { return err } // 一次回表: 既拿到行数据用于释放动态字段的堆块, 也下传给索引层定位 pkID。 storeRow, found := table.indexMeta.ResolveByPK(pkID) if !found || len(storeRow) < 8 { return page.ErrNotFound } name, _, ok := table.indexMeta.rowStoreMeta() if !ok { return ErrFieldNotFound } // indexKey 传 nil: 行数据在手, 每个索引都能自己从行里推出键 (时序键/主键列)。 if err := table.indexMeta.DeleteByKey(name.Name(), nil, storeRow, table.Field, table.activeVersion, table.versions); err != nil { return err } table.heapReleaseRow(storeRow[8:]) table.info.TotalRows-- table.saveTableInfo() if table.onWrite != nil { table.onWrite() } return nil } // UpdateRowByPK 按行号更新一行, 并级联更新它在全部索引里的条目。 func (table *Table) UpdateRowByPK(pkID uint64, values []types.FieldValue) error { if err := table.errIfReadOnly(); err != nil { return err } // 旧行既用于复用动态字段堆块, 也用于让索引层提取"旧键" (时序键变了要搬分段)。 storeRow, found := table.indexMeta.ResolveByPK(pkID) if !found || len(storeRow) < 8 { return ErrFieldNotFound } // 唯一性预检放在序列化之前: 见 checkUniqueUpdate。 if err := table.checkUniqueUpdate(values, storeRow[8:], pkID); err != nil { return err } newRowData, err := table.serializeRow(values, storeRow[8:]) if err != nil { return err } if err := table.indexMeta.UpdateByKey(nil, newRowData, pkID, storeRow, table.Field, table.activeVersion, table.versions); err != nil { return err } if table.onWrite != nil { table.onWrite() } return nil }