TsDB2/DB/core/table_crud.go

package core

import (
	"errors"
	"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
	}
	defer table.db.lockOp()()
	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
	}
	defer table.db.lockOp()()
	defer table.db.beginOp()() // 插入 = 数据页+各级索引的多页写, 中途禁止阈值提交截断
	if pkID == 0 {
		pkID = table.AllocPKID()
	}
	// 游标状态 (预留消费 + 顶起) 统一在 pkMu 内读写, 与 AllocPKID 的
	// 取号/预留登记互斥: 并发写者交错时避免游标回退发重号或预留错位。
	table.pkMu.Lock()
	reserved := table.hasPendingPK && table.pendingPK == pkID
	table.hasPendingPK = false
	// 查重短路: nextPK 是"历史发号上界" (只增不减, 删除行不回退), 初始化自
	// 当前最大存活 pkID。pkID >= nextPK ⇒ 这个号从未发放过 ⇒ 必然不存在,
	// 跳过 rowStoreHasPK 回表查重。时序行存储按 pkID 取行是 O(段数+页内行扫),
	// 顺序写入时这笔开销占写入 CPU 约 36% (bench_ts profile); 短路必须在
	// 游标顶起之前判定 —— 顶起后 pkID < nextPK 恒成立, 上界信息就被覆盖了。
	possiblyExists := true
	if !reserved {
		table.ensureNextPK()
		possiblyExists = pkID < table.nextPK
	}
	// 显式指定更大的主键时把游标顶到它之后, 避免后续自增再发一次同样的号
	if pkID+1 > table.nextPK {
		table.nextPK = pkID + 1
	}
	table.pkMu.Unlock()
	if !reserved && possiblyExists {
		if table.indexMeta.rowStoreHasPK(pkID) {
			return 0, fmt.Errorf("%w: pkID=%d", ErrDuplicatePK, pkID)
		}
	}

	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 {
		// 唯一冲突的预检先于任何索引写入 (index_meta_page.InsertRow): 此时行没落进
		// 任何索引, 序列化时已提交的 blob 链成为孤儿, 必须还掉。其它错误 (页满 / IO)
		// 可能已部分写入, 不能动链 —— 那种半成品只能靠事务回滚兜底。
		if errors.Is(err, ErrDuplicatedKey) {
			_ = table.releaseRowBlobRefs(rowData)
		}
		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 摊到批末一次。
//
// 失败语义: 序列化阶段报错时, 先前行写入的堆块成为无主垃圾 (泄漏而非错乱,
// 行数据并未落任何索引, 不会被读到); 各先行行的 blob 链在报错时会被 serializeRow
// 的兜底回收 (见 serializeRowInternal 的 newBlobRefs), 不随堆块一起泄漏。
// 索引阶段报错时部分索引已写入 —— 与单行路径"写了一半报错"一致, 唯一冲突
// (预检先于任何写入) 时整批 blob 链会回收, 其余错误靠事务回滚兜底。
func (table *Table) InsertRowsWithIDs(pkIDs []uint64, values [][]types.FieldValue) ([]uint64, error) {
	if err := table.errIfReadOnly(); err != nil {
		return nil, err
	}
	defer table.db.lockOp()()
	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
	}
	defer table.db.lockOp()()
	return table.insertRows(pkIDs, values, true)
}

func (table *Table) insertRows(pkIDs []uint64, values [][]types.FieldValue, allReserved bool) ([]uint64, error) {
	defer table.db.beginOp()() // 整批是一个操作: 批内任何一点都不许阈值提交
	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
		}
	}
	// 预留槽消费与游标顶起都在 pkMu 内, 与 AllocPKID/单行路径互斥
	// skipDup[i]=true 表示"号从未发放" (pkID >= 下界), 第 3) 步免回表查重 —
	// 判定必须在游标顶起之前做 (与单行路径同一短路, 见 InsertRowWithID)。
	table.pkMu.Lock()
	table.hasPendingPK = false // 预留号已被本批整体消费
	skipDup := make([]bool, n)
	if !allReserved {
		table.ensureNextPK()
		for i := 0; i < n; i++ {
			skipDup[i] = !auto[i] && pkIDs[i] >= table.nextPK
		}
		for i := 0; i < n; i++ {
			if !auto[i] && pkIDs[i]+1 > table.nextPK {
				table.nextPK = pkIDs[i] + 1
			}
		}
	}
	table.pkMu.Unlock()

	// 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] || skipDup[i] {
				continue
			}
			if table.indexMeta.rowStoreHasPK(pkIDs[i]) {
				return nil, fmt.Errorf("%w: pkID=%d", ErrDuplicatePK, pkIDs[i])
			}
		}
	}
	// 4) 逐行序列化 (动态堆写入在这一步发生)
	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 {
		// 唯一预检是"全部查完才动第一笔写入" (index_meta_page.InsertRows):
		// 唯一冲突意味着零写入, 每行序列化时已落的 blob 链整批还掉。
		// 非唯一错误可能已部分写入, 链不能动 (同单行路径)。
		if errors.Is(err, ErrDuplicatedKey) {
			for _, rd := range rowDatas {
				_ = table.releaseRowBlobRefs(rd)
			}
		}
		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) {
	defer table.db.lockOp()()
	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
	}
	defer table.db.lockOp()()
	defer table.db.beginOp()() // 删除 = 数据页+全部索引的多页写, 中途禁止阈值提交
	// 一次回表: 既拿到行数据用于释放动态字段的堆块, 也下传给索引层定位 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
	}
	defer table.db.lockOp()()
	defer table.db.beginOp()() // 更新 = 数据页+全部索引的多页写, 中途禁止阈值提交
	// 旧行既用于复用动态字段堆块, 也用于让索引层提取"旧键" (时序键变了要搬分段)。
	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
}