TsDB2/DB/core/deleter.go

package core

import (
	"TsDB2/DB/core/index"
	"TsDB2/DB/core/index/btree"
	"TsDB2/DB/core/index/datapage"
	"TsDB2/DB/core/page"
	"TsDB2/DB/core/types"
	"encoding/binary"
	"runtime/debug"
	"sort"
	"sync"
)

// Deleter 删除操作对象
type Deleter struct {
	table *Table
	where Condition
}

// Delete 返回删除操作对象
func (table *Table) Delete() *Deleter {
	return &Deleter{table: table}
}

// Where 设置删除条件
func (d *Deleter) Where(cond Condition) *Deleter {
	d.where = cond
	return d
}

// Exec 执行删除
// SQLite 风格优化: 先软删除 DataPage 行 (极快), 再重建索引 (流式, 低内存)
// 避免逐行级联删除 B+ 树的高昂开销
func (d *Deleter) Exec() (int64, error) {
	// 找到主键索引
	var pkMeta *IndexMeta
	for i := range d.table.indexMeta.metas {
		m := &d.table.indexMeta.metas[i]
		if m.Flag&0x80 != 0 {
			continue
		}
		if m.IndexType == page.IndexTypePrimary || m.IndexType == page.IndexTypeTime {
			pkMeta = &d.table.indexMeta.metas[i]
			break
		}
	}
	if pkMeta == nil {
		return 0, nil
	}

	// 用 Scan 流式收集匹配行的 pkID (避免 Find 反序列化全行)
	pkIDs := make(map[uint64]bool)
	err := d.table.Query().Where(d.where).Scan(func(row []types.FieldValue) bool {
		for _, fv := range row {
			if v, ok := fv.Value.(uint64); ok {
				pkIDs[v] = true
				break
			}
		}
		return true
	})
	if err != nil {
		return 0, err
	}
	if len(pkIDs) == 0 {
		return 0, nil
	}

	// Phase 1: 软删除 DataPage 中的行 (只标记 DeletedFlag, 极快!)
	pkHandler, _ := index.Compose.GetHandler(pkMeta.IndexType)
	dpIDs := pkHandler.GetDataPageIDs(d.table.handle, pkMeta.EntryPage)

	var count int64
	pkKeyBE := make([]byte, 8) // 复用缓冲区
	for _, dpID := range dpIDs {
		dp, err := datapage.Load(d.table.handle, dpID)
		if err != nil {
			continue
		}
		changed := false
		rows := dp.GetAllRows()
		for _, storeRow := range rows {
			if len(storeRow) < 8 {
				continue
			}
			pkID := binary.BigEndian.Uint64(storeRow[:8])
			if pkIDs[pkID] {
				binary.BigEndian.PutUint64(pkKeyBE, pkID)
				if dp.DeleteByPK(pkKeyBE) {
					changed = true
					count++
				}
			}
		}
		if changed {
			dp.Save()
		}
	}

	// 释放 pkIDs 内存 (Phase 2 不需要)
	pkIDs = nil

	// Phase 2: 流式重建所有索引 (逐个索引从 DataPage 读取, 不缓存全部行)
	if err := d.rebuildIndexes(pkMeta, dpIDs); err != nil {
		return count, err
	}

	// 更新行数
	d.table.info.TotalRows -= uint64(count)
	d.table.saveTableInfo()

	// 失效缓存
	d.table.indexMeta.primaryDPs = nil

	return count, nil
}

// rebuildIndexes 批量重建所有索引
// 主键索引: 顺序创建 DataPage + 批量构建 B+ 树 (避免逐行搜索)
// 二级索引: 收集条目 → 排序 → 批量构建 B+ 树
func (d *Deleter) rebuildIndexes(pkMeta *IndexMeta, dpIDs []uint32) error {
	fields := d.table.Field
	rowSize := int(d.table.info.RowSize)

	// 内存池: 复用行缓冲区
	rowPool := &sync.Pool{
		New: func() interface{} {
			b := make([]byte, 8+rowSize)
			return &b
		},
	}

	// ===== 重建主键索引: 顺序扫描 + 批量构建 =====
	var leafEntries [][]byte
	var curDP *datapage.DataPage
	var curDPID uint32

	for _, dpID := range dpIDs {
		dp, err := datapage.Load(d.table.handle, dpID)
		if err != nil {
			continue
		}
		rows := dp.GetAllRowsPooled(rowPool)
		for _, storeRow := range rows {
			if len(storeRow) < 8 {
				continue
			}
			rowData := storeRow[8:]
			indexKey := extractIndexKey(rowData, pkMeta, fields, d.table.activeVersion, d.table.versions)
			if indexKey == nil {
				continue
			}
			// 确保有可用的 DataPage
			if curDP == nil {
				curDP = datapage.New(d.table.handle)
				curDPID = curDP.PageID()
			}
			// 当前 DataPage 已满则创建新的
			if curDP.IsFull(uint32(len(storeRow))) {
				curDP = datapage.New(d.table.handle)
				curDPID = curDP.PageID()
			}
			curDP.AppendRow(storeRow)
			curDP.Save()
			// 收集叶条目: [pkID(8B BE) + dpID(4B BE)]
			pkID := binary.BigEndian.Uint64(storeRow[:8])
			pkKeyBE := make([]byte, 8)
			binary.BigEndian.PutUint64(pkKeyBE, pkID)
			entry := make([]byte, 12)
			copy(entry[:8], pkKeyBE)
			binary.BigEndian.PutUint32(entry[8:], curDPID)
			leafEntries = append(leafEntries, entry)
		}
		// 归还缓冲区到池
		for i := range rows {
			rowPool.Put(&rows[i])
		}
	}

	// 批量构建主键 B+ 树 (一次创建所有节点, 避免逐行搜索)
	pkMeta.EntryPage = btree.BuildFromSorted(leafEntries, d.table.handle)
	d.table.indexMeta.saveMeta(pkMeta)
	leafEntries = nil // 释放内存
	debug.FreeOSMemory()

	// ===== 重建每个二级索引: 收集 → 排序 → 批量构建 =====
	for i := range d.table.indexMeta.metas {
		m := &d.table.indexMeta.metas[i]
		if m.Flag&0x80 != 0 {
			continue
		}
		if m.IndexType == page.IndexTypePrimary || m.IndexType == page.IndexTypeTime {
			continue
		}

		// 收集所有二级索引条目
		var secEntries [][]byte
		for _, dpID := range dpIDs {
			dp, err := datapage.Load(d.table.handle, dpID)
			if err != nil {
				continue
			}
			rows := dp.GetAllRowsPooled(rowPool)
			for _, storeRow := range rows {
				if len(storeRow) < 8 {
					continue
				}
				pkID := binary.BigEndian.Uint64(storeRow[:8])
				rowData := storeRow[8:]
				secKey := extractIndexKey(rowData, m, fields, d.table.activeVersion, d.table.versions)
				if secKey == nil {
					continue
				}
				// 构建二级条目: [secKey + pkID(8B BE)]
				secEntry := make([]byte, len(secKey)+8)
				copy(secEntry[:len(secKey)], secKey)
				binary.BigEndian.PutUint64(secEntry[len(secKey):], pkID)
				secEntries = append(secEntries, secEntry)
			}
			for i := range rows {
				rowPool.Put(&rows[i])
			}
		}

		// 按 secKey 排序 (稳定排序保持同 key 的 pkID 顺序)
		sort.SliceStable(secEntries, func(i, j int) bool {
			ki := secEntries[i][:len(secEntries[i])-8]
			kj := secEntries[j][:len(secEntries[j])-8]
			return compareBytes(ki, kj) < 0
		})

		// 批量构建二级 B+ 树
		m.EntryPage = btree.SBTBuildFromSorted(secEntries, d.table.handle)
		d.table.indexMeta.saveMeta(m)
		secEntries = nil
		debug.FreeOSMemory()
	}

	return nil
}

// compareBytes 按字典序比较两个字节切片
func compareBytes(a, b []byte) int {
	ml := len(a)
	if len(b) < ml {
		ml = len(b)
	}
	for i := 0; i < ml; i++ {
		if a[i] < b[i] {
			return -1
		}
		if a[i] > b[i] {
			return 1
		}
	}
	if len(a) < len(b) {
		return -1
	}
	if len(a) > len(b) {
		return 1
	}
	return 0
}