TsDB2/DB/core/db.go

package core

import (
	"git.ycl.pub/yl32940/TsDB2/DB/core/index"
	"git.ycl.pub/yl32940/TsDB2/DB/core/page"
	"git.ycl.pub/yl32940/TsDB2/DB/core/types"
	"git.ycl.pub/yl32940/TsDB2/DB/core/util"
	"encoding/binary"
	"errors"
	"fmt"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"unsafe"
)

type TableMeta struct {
	Name       string
	FieldCount uint32
	TotalRows  uint64
	FirstHeap  uint32
	LastHeap   uint32
	FirstIndex uint32
	LastIndex  uint32
	FirstData  uint32
	LastData   uint32
	RowSize    uint16
}

type DB struct {
	header *HeaderPage
	tables *TablePage

	file *File

	// path 数据库文件路径 (VACUUM / 自动重打开使用)。
	// 由 Open(path) 赋值, 运行时不变。
	path string

	// 防抖自动提交
	autoCommitDelay time.Duration
	autoCommitTimer *time.Timer
	autoCommitMu    sync.Mutex

	// commitMu 提交互斥: 防抖定时器 / 脏页阈值回调 / 用户显式提交 / 事务提交共用。
	// 只串行化"提交"这一步, 不参与写入路径 —— 详见 DB.Commit 的说明。
	commitMu sync.Mutex

	// accMu 操作级可重入串行锁: 引擎不支持并发读写, 所有公开操作入口都过它,
	// 把多 goroutine 的误并发自动串行化(防止改坏结构 / 读到中间态)。
	// 显式事务在 Begin 持有、Commit/Rollback 释放; 事务内同 goroutine 重入不阻塞。
	// 详见 access_lock.go。
	accMu accessLock

	// 当前活跃事务（nil 表示无事务）。
	// 用 atomic.Pointer: 写线程 (Begin/Commit/Rollback) 与自动提交的定时器
	// goroutine (autoCommit/NotifyWrite) 会并发读写它, 裸指针是 data race。
	// 注意这只保证【指针本身】的可见性 —— Transaction 内部字段仍受
	// "单写入者模型"约束 (同一时刻只允许一个 goroutine 跑写入路径)。
	currentTxn atomic.Pointer[Transaction]

	// opDepth 行级多页写操作的嵌套深度 (InsertRow/UpdateRow/DeleteRow/批量/建索引/批量删)。
	// >0 时阈值/防抖提交一律不触发: 否则 Commit 会把同一操作的脏页切成两笔 WAL,
	// kill 后烧出半成品 (DockerManager 实测: 数据页已提交、索引页未提交, 行按主键
	// 在、按二级索引永远查不到)。深度归零时由收尾闭包补一次提交检查。
	opDepth atomic.Int32

	// pendingMu 保护 pendingFree / pendingFreeChain: 写入路径登记、提交路径出队,
	// 两条路径并发 (写线程插入/删除 + 后台提交线程) 时必须串行化, 否则切片并发读写
	// 会丢登记项 / 触发数据竞争。
	pendingMu sync.Mutex
	// pendingFree 无事务上下文时积压的"待回收页"(堆页摘链 / 批量删除后作废的旧 DataPage)。
	// 不在修改中途直写盘, 而是等 Commit 成功 / Close 后再真正还链 —— 见 DeferFreePage 的说明。
	// currentTxn != nil 时改入 txn.pendingFree (Rollback 时整体丢弃)。
	pendingFree []uint32
	// pendingFreeChain 无事务上下文时积压的"待回收整链"(见 DeferFreeChain)。
	// currentTxn != nil 时改入 txn.pendingFreeChain (Rollback 时整体丢弃)。
	pendingFreeChain []pendingChain
	// freePage 真正把页还给空闲链表 (直写盘), 供上面的队列出队时调用
	freePage func(pageID uint32)
	// freeChain 真正把整链还给空闲链表 (直写盘), 供上面的队列出队时调用
	freeChain func(head, tail, count uint32) error

	// 自动 VACUUM 调度 (SetAutoVACUUM 启用)
	autoVacuumMu      sync.Mutex
	autoVacuumTimer   *time.Timer
	autoVacuumEnabled bool
	autoVacuumOpts    VACUUMOptions
	lastVACUUM        time.Time

	// vacuumMu 串行化 VACUUM 本体: 手动与自动调度可能并发触发,
	// 双开会对同一文件做 Close/Open 交错 (自动调度抢到锁才跑, 否则本轮跳过)。
	vacuumMu sync.Mutex

	// dirtyThreshold 脏页阈值的用户配置 (内存态, wireRuntime 重绑阈值回调时用)。
	dirtyThreshold int
}

// pendingChain 一条待回收的整链: head…tail 共 count 页, 链内 next 在 PageHdr.NextPage
type pendingChain struct {
	head, tail, count uint32
}

// OpenOptions 打开数据库的选项。
type OpenOptions struct {
	// ForceInspect 抢救模式: 头页校验 (Magic/XXH32) 失败时仍打开 —— 打印警告、
	// 只读 (一切写操作返回 ErrReadOnly)、跳过 WAL 恢复 (恢复会写盘)。
	// 用途: 头页损坏后抢救数据 / analyzer 类工具手工勘验。
	ForceInspect bool

	// ReadOnly 只读打开: 文件以 O_RDONLY 打开、加【共享】跨进程锁 (可与其它只读
	// 进程共存, 与读写进程互斥)、跳过 WAL 恢复与一切写盘。用于报表 / 分析等不
	// 修改数据的连接 (DSN mode=ro)。
	ReadOnly bool
}

// Open 打开数据库 (等价 OpenWithOptions 零值)。
func Open(path string) (*DB, error) {
	return OpenWithOptions(path, OpenOptions{})
}

// OpenWithOptions 打开数据库。新建库走 newHeader + PrepareWrite (Version=11,
// CRC 随首次落盘生成); 已有库先 Validate (Magic 必查, v11 且 CRC==0 的旧库
// 跳过, 其余校验 XXH32), 校验失败默认拒绝打开, ForceInspect 时只读抢救。
func OpenWithOptions(path string, opt OpenOptions) (*DB, error) {
	var (
		ff  *File
		err error
	)
	if opt.ReadOnly {
		ff, err = OpenFileReadOnly(path)
	} else {
		ff, err = OpenFile(path)
	}
	if err != nil {
		return nil, err
	}

	// 启用页面缓冲池 (只读时仅作读缓存, 不会产生脏页)
	ff.InitBufferPool(DefaultBufferPoolCapacity)

	stat, err := ff.f.Stat()
	if err != nil {
		return nil, err
	}
	if opt.ReadOnly && stat.Size() == 0 {
		_ = ff.f.Close()
		return nil, fmt.Errorf("只读打开失败: 数据库文件 %q 不存在或为空", path)
	}

	// 自适应缓冲池容量: 按文件总页数 1/4 配置 —— 整库可放内存时最大化命中率,
	// 巨型库也不会无限制占内存。上界 8000 页 (128MB) 截断, 下界保留默认值。
	// 旧固定 16MB 对 214MB+ 的真实库欠配, 1M 行 Delete 全程冷读导致 ~50% 时间
	// 花在 pread 上; 扩到 1/4 后, 全表 COUNT(*) / 随机点查的主数据页基本全驻。
	if stat.Size() > 0 {
		filePages := int(stat.Size() / int64(PageSize))
		auto := filePages / 4
		if auto > 8000 {
			auto = 8000
		}
		if auto > DefaultBufferPoolCapacity {
			ff.SetBufferPoolCapacity(auto)
		}
	}

	// 跨进程文件锁: 只读用共享锁 (多个只读可共存), 读写用排他锁; 在读头页 /
	// WAL 恢复之前先占锁, 阻止读写进程与读者/写者同时打开。
	if stat.Size() > 0 {
		var lockErr error
		if opt.ReadOnly {
			lockErr = ff.tryRLock()
		} else {
			lockErr = ff.tryLock()
		}
		if lockErr != nil {
			_ = ff.f.Close() // 释放句柄 (锁随之释放)
			return nil, lockErr
		}
	}

	var header *HeaderPage
	if stat.Size() == 0 {
		// 建库: 未提供介绍 → 头页 Comment 默认写入 VersionComment
		// (当前版本的功能介绍与修复点); 建库时间见 CreateTime 字段
		header, err = newHeader("")
		if err != nil {
			return nil, err
		}
		if err := ff.WriteHeader(header); err != nil {
			return nil, err
		}
		// 新库头页写入后文件已有内容, 立即加跨进程文件锁。
		if err := ff.tryLock(); err != nil {
			_ = ff.f.Close()
			return nil, err
		}
	} else {
		header, err = ff.ReadHeader()
		if err != nil {
			return nil, err
		}
		// 头页校验: Magic 必查; CRC 对 v11-旧库 (HeaderCrc==0) 跳过, 其余校验。
		if err := header.Validate(); err != nil {
			// 只读打开本就不写; 抢救模式显式要求 —— 两者都降级为只读, 其余情况拒绝。
			if !opt.ForceInspect && !opt.ReadOnly {
				_ = ff.f.Close() // 校验失败必须释放文件句柄, 否则 Windows 上文件被占用
				return nil, err
			}
			fmt.Printf("警告: %v\n以只读模式打开 (写操作被禁用)\n", err)
			ff.readOnly = true
		}
		// 版本闸门: 低于当前版本的库结构对不上, 一律拒绝 (抢救模式也不例外)
		if header.Version < Version {
			_ = ff.f.Close()
			return nil, fmt.Errorf("不兼容的数据库文件版本 %d (当前支持 >= %d), 请重建数据库", header.Version, Version)
		}
	}
	// ==========================
	// 初始化表页
	// ==========================

	// realFreePage 真正把页还给空闲链表。
	// 必须直接写盘: 走 BufferPool 会被 ClearCache 清掉; 且该页在缓存里通常仍是
	// "脏"的旧内容, 不覆盖的话 Commit 收集脏页时会把刚写好的空闲链表指针反向
	// 覆盖掉, 导致空闲链错乱/页被重复分配。
	realFreePage := func(pageID uint32) {
		ff.withHeader(func() {
			header.recyclePage(pageID, ff.WritePageDirect)
			syncBuf := make([]byte, PageSize)
			_ = ff.readPageDisk(pageID, syncBuf)
			ff.SyncCache(pageID, syncBuf)
			_ = ff.writeHeaderLocked(header)
		})
	}

	// realFreeChain 把一条已用 PageHdr.NextPage 串好的整链一次性还链 (O(1))。
	//
	// 与 realFreePage 的区别只是"链 vs 单页": 这里只改写一条链尾页 + 更新头页,
	// 代价与链长无关。逐页遍历回收 N 页需要 2N 次 IO (每页一次读 + 一次写),
	// 拼接只要 3 次 —— DropTable 的堆链、WAL 提交后的 WAL 链都走这里。
	realFreeChain := func(head, tail, count uint32) error {
		var err error
		ff.withHeader(func() {
			err = recycleChainWithCache(ff, header, head, tail, count)
			if err != nil {
				return
			}
			err = ff.writeHeaderLocked(header)
		})
		return err
	}

	// 启动期的句柄用 nil 接收者构造: Init 阶段只会用到 ReadPage/SavePage,
	// 涉及 db 的闭包有 nil 守卫且此阶段不会触发; db 构造完成后以真实接收者重建
	// (wireRuntime), 闭包从此绑定本句柄 —— 见 buildHandleDb 的说明。
	tablePage := &TablePage{
		handle: (*DB)(nil).buildHandleDb(ff, header, realFreePage, realFreeChain),
	}

	// ==========================
	// 初始化索引元数据页
	// ==========================
	var db *DB

	saveHeader := false
	//如果索引页和表页都是空的话则直接创建
	if header.TablePage == 0 {
		if opt.ReadOnly {
			_ = ff.f.Close()
			return nil, fmt.Errorf("只读打开的数据库缺少表页 (TablePage=0): 不是有效数据库, 无法在只读模式初始化")
		}
		// 用带读盘能力的分配: 即便此时空闲链非空 (例如 TablePage 丢失的旧库),
		// 也能正确从空闲链摘页; 新库空闲链为空, 走 NextFreePage 递增分配。
		header.TablePage = ff.allocPage(header, true)
		tablePage.InitHeader(header.TablePage)
		saveHeader = true
	}
	if saveHeader {
		err = ff.WriteHeader(header)
		if err != nil {
			return nil, err
		}
	}

	// ==========================
	// WAL 恢复: 检查是否有未完成的 WAL 事务
	// ==========================
	// WAL 恢复会写盘 —— 抢救模式 (只读) 下跳过: 待恢复事务保持挂起, 数据页
	// 仍是盘上原状, 读路径不受影响 (这正是抢救模式要的效果)。
	//
	// 必须在 tablePage.Init 之前: 恢复会把已提交事务的表信息页镜像 (TotalRows /
	// 堆链指针 / RowSize 等) 重放回磁盘, 而 Init 会把这些元数据【载入内存 Table
	// 对象】。先 Init 再恢复, 内存里就是旧元数据: 崩溃恢复后 TotalRows 对不上、
	// 堆链指针过期, 后续写入按过期结构分配 = 损坏 (wal_crash_chaos 压测实锤)。
	if !ff.readOnly {
		if err := RecoverWAL(ff, header); err != nil {
			_ = ff.f.Close() // 恢复失败必须释放文件句柄, 否则 Windows 上文件被占用
			return nil, err
		}
		// 孤儿 WAL 链自愈: kill 恰落在"WAL 页已写盘、头页指针未发布(或已被后续
		// 提交清掉)"的窗口时, 盘上会留下 FirstWALPage 找不到的 WAL 链页, 永不
		// 回收。这里把它们整条捡回空闲链 (不重放: 已提交事务的回写必先于指针
		// 清零落盘, 重放只是幂等; 未提交事务的镜像本就该丢弃 —— 回收即丢弃)。
		if err := healOrphanWALPages(ff, header); err != nil {
			_ = ff.f.Close()
			return nil, err
		}
	}

	// 恢复完成后再载入表元数据: 上面重放过的表页此刻才是最新状态
	tablePage.Init(header.TablePage)

	db = &DB{
		file:            ff,
		header:          header,
		tables:          tablePage,
		path:            path,
		autoCommitDelay: DefaultAutoCommitDelay,
		freePage:        realFreePage,
		freeChain:       realFreeChain,
		dirtyThreshold:  DefaultDirtyThreshold,
	}
	// 反向注入, TablePage.Create 时拿到 db 指针; bindDB 顺带给 Init 阶段已经从盘上
	// 加载出来的表补上 db (那时 DB 还没构造完, 只能现在回填)。
	db.tables.bindDB(db)
	db.startAutoCommit()
	// 运行期回调与句柄闭包以真实接收者重绑 (启动期那版绑定的是 nil 接收者)。
	db.wireRuntime()
	return db, nil
}

// buildHandleDb 构造 TablePage 的运行期句柄, 闭包绑定接收者 db。
//
// Open 在 db 存在之前先用 nil 接收者构造一版 (Init 阶段用不到 db 闭包,
// 且涉及 db 的闭包都有 nil 守卫); db 构造完成后以真实接收者重建 —— 闭包
// 从此绑定本句柄的 currentTxn/pendingFree。VACUUM 的 adoptEngineState 换过
// 引擎状态后也必须用它重绑, 否则闭包仍指向旧句柄: 新句柄上的事务不会被
// markTouched (Rollback 静默失效), 新分配的页登记到旧句柄 (回滚后成孤儿页)。
func (db *DB) buildHandleDb(ff *File, header *HeaderPage,
	realFreePage func(uint32), realFreeChain func(uint32, uint32, uint32) error) page.HandleDb {
	return page.HandleDb{
		NewPage: func() uint32 {
			// allocPage(erase=true): headerMu 内摘空闲链 / 递增游标并推进头页; 擦页在
			// 锁外走 WritePage —— 调用方紧接着的整页写入会覆盖池里同一份缓存, 因此不
			// 额外产生物理 IO。读空闲链直接读盘 (绕过 BufferPool)。
			pid := ff.allocPage(header, true)
			// 事务内新分配的页登记到事务: 分配已就地推进 header 游标 (头页延迟到
			// Commit 落盘, 不可回退), Rollback 时必须把这些页还给空闲链, 否则成为
			// 永久泄漏的孤儿页。
			if db != nil {
				if txn := db.currentTxn.Load(); txn != nil {
					txn.allocated = append(txn.allocated, pid)
				}
			}
			return pid
		},
		FreePage: func(pageID uint32) {
			// 事务内被立即还链的页, 从 allocated 撤销登记, 避免 Rollback 时二次归还。
			if db != nil {
				if txn := db.currentTxn.Load(); txn != nil {
					txn.undoAllocated(pageID)
				}
			}
			realFreePage(pageID)
		},
		// 延迟回收: 堆页摘链 / 旧 DataPage 作废时只登记, 等提交成功后再还链。
		// 中途直写盘会让"页已清零"先于 WAL 落盘 —— 崩溃或 Rollback 后原页变空页,
		// 而堆链/索引仍指向它。事务内入 txn.pendingFree (回滚即丢弃),
		// 事务外入 db.pendingFree (下次提交/关闭时统一还链)。
		DeferFreePage: func(pageID uint32) {
			if db == nil {
				realFreePage(pageID) // 启动期(如 WAL 恢复前)没有事务上下文, 保持原行为
				return
			}
			if txn := db.currentTxn.Load(); txn != nil {
				txn.pendingFree = append(txn.pendingFree, pageID)
				return
			}
			db.pendingMu.Lock()
			db.pendingFree = append(db.pendingFree, pageID)
			db.pendingMu.Unlock()
		},
		ReadPage: ff.ReadPage,
		SavePage: ff.WritePage,
		// 整链回收: DropTable 的堆链 / WAL 提交后的 WAL 链 / 哈希索引的桶目录链与桶
		// 溢出链, 都是「首尾和链长已知、且链内 next 与空闲链同用 PageHdr.NextPage」
		// 的链 —— 一次拼接 O(1) 还链, 不必逐页遍历 (逐页 = 每页读+写各一次)。
		FreeChain: realFreeChain,
		// 延迟整链回收: 与 DeferFreePage 同一时序约定 (提交成功后才真正还链),
		// 但一条链只登记一次, 提交后一次 FreeChain 还链。
		DeferFreeChain: func(head, tail, count uint32) {
			if head == 0 || count == 0 {
				return
			}
			if db == nil {
				_ = realFreeChain(head, tail, count) // 启动期没有事务上下文, 保持原行为
				return
			}
			c := pendingChain{head: head, tail: tail, count: count}
			if txn := db.currentTxn.Load(); txn != nil {
				txn.pendingFreeChain = append(txn.pendingFreeChain, c)
				return
			}
			db.pendingMu.Lock()
			db.pendingFreeChain = append(db.pendingFreeChain, c)
			db.pendingMu.Unlock()
		},
		// ValidPage: 页号合法性 = 落在已分配区间 [2, NextFreePage)。
		// 0 是头页、1 是建库预留的表页; >= NextFreePage 的页从未分配过, 文件里没有
		// 它的位置 —— 放进空闲链, 下一次 NewPage 就会拿着它写到几十 GB 偏移处
		// (实测 DROP TABLE 假死 52s、稀疏文件暴涨到 40GB 的直接原因)。
		ValidPage: func(pageID uint32) bool {
			ff.headerMu.Lock()
			limit := header.NextFreePage
			ff.headerMu.Unlock()
			return pageID >= 2 && pageID < limit
		},
	}
}

// errIfReadOnly 抢救模式 (force-inspect) 下写入入口的统一闸门。
// File 层的 readOnly 守卫保证"任何数据都不可能落盘"; 这里的入口闸门让写入
// 在第一时间就拿到清晰的 ErrReadOnly, 而不是缓存一串脏页到提交时才失败。
func (db *DB) errIfReadOnly() error {
	if db.file.readOnly {
		return ErrReadOnly
	}
	return nil
}

// errIfReadOnly 表级写入入口的闸门 (经 table.db 转发)。
func (table *Table) errIfReadOnly() error {
	if table.db == nil {
		return nil
	}
	return table.db.errIfReadOnly()
}

// wireRuntime 把运行期回调绑定到本句柄: 事务页登记 (onPageWrite)、脏页阈值
// 自动提交回调、表页句柄闭包。VACUUM 的 adoptEngineState 换引擎状态后必须重绑,
// 否则这些闭包仍指向旧句柄。
func (db *DB) wireRuntime() {
	db.file.SetOnPageWriteCallback(func(pageID uint32) {
		if txn := db.currentTxn.Load(); txn != nil {
			txn.markTouched(pageID)
		}
	})
	// 脏页阈值: 超过时自动触发 Commit 刷盘, 防止内存暴涨。
	// 走 autoCommit: 有活跃事务时不能从中途提交 (会破坏事务原子性)。
	db.file.SetDirtyThreshold(db.dirtyThreshold, db.autoCommit)
	db.tables.handle = db.buildHandleDb(db.file, db.header, db.freePage, db.freeChain)
}

// NewPage 分配一个新页号, 走空闲链 / 头页游标的同一路径, 并按事务内已分配登记.
//
// 这条路径与 HandleDb.NewPage 闭包完全等价 —— 后者只绑给 TablePage 的几个调用点,
// 其它地方 (如 BlobCommit) 不持有 handle, 必须走这里. 返回的是页号, 调用方拿到后
// 仍要自己用 db.file.ReadPage/WritePage 读写内容.
func (db *DB) NewPage() uint32 {
	pid := db.file.allocPage(db.header, true)
	if txn := db.currentTxn.Load(); txn != nil {
		txn.allocated = append(txn.allocated, pid)
	}
	return pid
}

func (db *DB) updateTime() {
	db.file.withHeader(func() {
		db.header.ModifyTime = time.Now().UnixNano()
		_ = db.file.writeHeaderLocked(db.header)
	})
}

// Comment 返回头页里的数据库介绍区内容 (建库时自动写入一段创建说明, 之后可用
// SetComment 改写)。返回的是去掉尾部零字节后的字符串。
func (db *DB) Comment() string {
	db.file.headerMu.Lock()
	comment := db.header.Comment
	db.file.headerMu.Unlock()
	return strings.TrimRight(string(comment[:]), "\x00")
}

// SetComment 改写数据库介绍区 (头页 Comment 字段, ≤512 字节, 超长报错)。
// 写入走 withHeader + writeHeaderLocked (与 updateTime 同模式), 并更新 ModifyTime。
func (db *DB) SetComment(desc string) error {
	if err := db.errIfReadOnly(); err != nil {
		return err
	}
	defer db.lockOp()()
	if len(desc) > len(db.header.Comment) {
		return fmt.Errorf("描述字符串长度 %d 超过最大限制 %d", len(desc), len(db.header.Comment))
	}
	var werr error
	db.file.withHeader(func() {
		var comment [PageSize4K - 332]byte
		copy(comment[:], desc)
		db.header.Comment = comment
		db.header.ModifyTime = time.Now().UnixNano()
		werr = db.file.writeHeaderLocked(db.header)
	})
	return werr
}

func (db *DB) FindTable(name string) (*Table, bool) {
	tbl, ok := db.tables.Find(name)
	if ok && tbl.onWrite == nil {
		tbl.SetOnWrite(db.NotifyWrite)
	}
	return tbl, ok
}

func (db *DB) CreateTable(name string, fields []types.FieldInfo) (*Table, error) {
	if db.HasActiveTransaction() {
		return nil, errDDLInTransaction("CreateTable")
	}
	defer db.lockOp()() // DDL 全程串行; 内部 db.Commit 同 goroutine 重入
	tbl, err := db.tables.Create(name, fields)
	if err != nil {
		return nil, err
	}
	tbl.SetOnWrite(db.NotifyWrite)
	db.file.withHeader(func() {
		db.header.TableCount++
		_ = db.file.writeHeaderLocked(db.header)
	})
	db.updateTime()
	// DDL 隐式提交: 把新表钉到盘上 —— DDL 不具有事务性 (与 Oracle/MySQL 一致),
	// 未提交的 DDL 随回滚消失后, 内存元数据 (TableInfo/IndexMeta) 会指向已回
	// 收的页, 后续读写立即迷路 (txn_crash_chaos 压测实锤)
	if err := db.Commit(); err != nil {
		return nil, err
	}
	return tbl, nil
}

// errDDLInTransaction 事务内执行 DDL 的明确拒绝。
//
// DDL 一旦允许进入事务, 回滚它就会出现"内存元数据 vs 磁盘真值"的分叉:
// Rollback 按页丢弃缓存, 而 Table.info / IndexMeta.metas 是独立于页缓存的
// Go 堆副本, 不会随缓存一起消失 —— 未提交的建表/建索引回滚后, 它们仍指向
// 事务中分配、已被回收的页号, 下一个操作立刻 load failed / 写穿到空闲页。
// 因此这里宁可显式拒绝, 也不让调用方踩进这个不可恢复的坑。
func errDDLInTransaction(op string) error {
	return fmt.Errorf("%s: DDL 操作不支持在事务中执行 (DDL 为隐式提交语义, 请先 Commit 或 Rollback 当前事务)", op)
}

// ddlBoundary DDL 收尾: 无活跃事务时隐式提交, 有活跃事务时应已被入口拒绝
// (防御性再查一次)。返回 error 时 DDL 已成功但落盘失败 —— 调用方应视 DDL 为未完成。
func (db *DB) ddlBoundary(op string) error {
	if db.HasActiveTransaction() {
		return errDDLInTransaction(op)
	}
	return db.Commit()
}

func (db *DB) DropTable(name string) error {
	if db.HasActiveTransaction() {
		return errDDLInTransaction("DropTable")
	}
	defer db.lockOp()() // DDL 全程串行; 内部 Flush/Commit 同 goroutine 重入
	// 先刷脏页再清缓存: ClearCache 会把缓存里的页统统标成"干净",
	// 而随后的 FlushBufferPool(CollectDirty) 会把它们直接移出缓存 ——
	// 如果缓存里还有尚未落盘的页(手动提交模式下很常见), 就会被静默丢掉。
	_ = db.file.FlushBufferPool()
	db.file.ClearCache()
	if err := db.tables.Drop(name); err != nil {
		return err
	}
	db.file.withHeader(func() {
		db.header.TableCount--
		_ = db.file.writeHeaderLocked(db.header)
	})
	db.updateTime()
	// 先刷盘再清缓存, 防止 Drop 中标记的 Flag(0x80) 丢失
	_ = db.file.FlushBufferPool()
	db.file.ClearCache()
	return db.ddlBoundary("DropTable")
}

// TruncateTable 清空一张表的全部数据但保留表结构与索引定义 (DDL, 隐式提交)。
// 事务内拒绝 (与其它 DDL 一致)。
func (db *DB) TruncateTable(name string) error {
	if err := db.errIfReadOnly(); err != nil {
		return err
	}
	if db.HasActiveTransaction() {
		return errDDLInTransaction("TruncateTable")
	}
	defer db.lockOp()()
	if err := db.tables.Truncate(name); err != nil {
		return err
	}
	return db.ddlBoundary("TruncateTable")
}

// RenameTable 重命名表 (DDL, 隐式提交)。只改 TableInfo.Name 元数据字符串,
// 不涉及数据页 / 索引页迁移或重建。事务内拒绝。
func (db *DB) RenameTable(oldName, newName string) error {
	if err := db.errIfReadOnly(); err != nil {
		return err
	}
	if db.HasActiveTransaction() {
		return errDDLInTransaction("RenameTable")
	}
	defer db.lockOp()()
	if err := db.tables.RenameTable(oldName, newName); err != nil {
		return err
	}
	return db.ddlBoundary("RenameTable")
}

// RenameColumn 重命名列 (DDL, 隐式提交)。只改 FieldInfo.FieldName + schema 落盘,
// 不涉及列偏移 / 行数据 / 索引键重建。事务内拒绝。
func (db *DB) RenameColumn(tableName, oldCol, newCol string) error {
	if err := db.errIfReadOnly(); err != nil {
		return err
	}
	if db.HasActiveTransaction() {
		return errDDLInTransaction("RenameColumn")
	}
	defer db.lockOp()()
	tbl, ok := db.tables.Find(tableName)
	if !ok {
		return ErrTableNotFound
	}
	return tbl.RenameColumn(oldCol, newCol)
}

func (db *DB) GetAllTables() []TableMeta {
	res := make([]TableMeta, 0, len(db.tables.tables))
	for _, t := range db.tables.tables {
		if util.IsHighBitSet(t.info.Flag) {
			continue
		}
		res = append(res, TableMeta{
			Name:       util.BytesToString(t.info.Name[:]),
			FieldCount: uint32(t.info.FieldCount),
			TotalRows:  t.info.TotalRows,
			FirstHeap:  t.info.FirstHeap,
			LastHeap:   t.info.LastHeap,
			FirstData:  t.info.FirstData,
			LastData:   t.info.LastData,
			RowSize:    t.info.RowSize,
		})
	}
	return res
}

func (db *DB) Close() error {
	defer db.lockOp()() // 等待在飞操作结束, 并在关闭全程挡住新操作
	// 停止防抖定时器
	db.stopAutoCommit()
	// 只读实例: 无脏页/无待写, 直接关句柄 —— Commit/写头/fsync 在只读 fd 上
	// (Windows) 会报 Access is denied, 且没有任何需要持久化的内容。
	if db.file.readOnly {
		return db.file.Close()
	}
	// 先提交剩余脏页 (走 WAL, 提交路径会把头页一并落盘)
	errCommit := db.Commit()
	// 头页兜底: Commit 失败时游标变更还悬着, 必须先于缓冲池刷盘落下,
	// 否则盘上出现页号超过 NextFreePage 的脏页, 重开后会被二次分配覆盖
	errHeader := db.file.FlushHeader(db.header)
	// BufferPool 安全网 (理论上已无脏页); 写盘失败必须带出去 —— 失败页仍脏,
	// 但 Close 之后无人再刷, 吞掉错误等于静默丢数据
	errFlush := db.file.FlushBufferPool()
	// 积压的待回收页: 此时链指针已落盘, 可以安全还链
	db.drainPendingFree()
	// 收尾 fsync: 保证 Header / 数据页真正落到持久介质,
	// 否则进程退出后立刻断电可能丢掉最后一次提交
	errSync := db.file.sync()
	errClose := db.file.Close()
	// 任何一个环节出错都要带出去 (errors.Join 忽略 nil)
	return errors.Join(errCommit, errHeader, errFlush, errSync, errClose)
}

// freeNow 把一批待回收页真正还给空闲链表 (直写盘)
func (db *DB) freeNow(ids []uint32) {
	for _, pid := range ids {
		if db.freePage != nil {
			db.freePage(pid)
		}
	}
}

// freeChainNow 把一批待回收整链真正还给空闲链表 (每条链一次 O(1) 拼接)
func (db *DB) freeChainNow(chains []pendingChain) {
	for _, c := range chains {
		if db.freeChain != nil {
			_ = db.freeChain(c.head, c.tail, c.count)
		}
	}
}

// drainPendingFree 回收无事务上下文时积压的页与整链。
// 只能在"链指针已经持久化"的时刻调用 —— 即提交成功之后, 或 Close 收尾时。
func (db *DB) drainPendingFree() {
	// 先原子地摘下当前队列 (写入路径可能正在并发登记, 只取走此刻已入队的部分),
	// 再在锁外真正还链 —— 还链会走 headerMu + 磁盘 IO, 不该占着 pendingMu。
	db.pendingMu.Lock()
	ids := db.pendingFree
	db.pendingFree = nil
	chains := db.pendingFreeChain
	db.pendingFreeChain = nil
	db.pendingMu.Unlock()

	if len(ids) == 0 && len(chains) == 0 {
		return
	}
	db.freeNow(ids)
	db.freeChainNow(chains)
}

// Commit 显式提交 — 通过 WAL 保证原子性
// 同时重置防抖定时器
//
// commitMu 保证同一时刻只有一笔提交在飞: 防抖定时器、脏页阈值回调(在写入 goroutine
// 内同步触发) 和用户显式调用都可能撞在一起, 而两笔提交同时收集/回写同一批脏页会把
// 页的归属搞乱 (前一笔的快照会把后一笔刚改出来的内容判成"已落盘")。
// 这把锁只串行化提交本身, 不参与写入路径: 写入只在 BufferPool.WritePage 里等
// writeLock (见 LockWrites), 两者没有嵌套获取, 因此不会互相死锁。
func (db *DB) Commit() error {
	if err := db.errIfReadOnly(); err != nil {
		return err
	}
	defer db.lockOp()() // 操作级串行: 提交期间挡住其他 goroutine 的读写(独立调用时)
	db.commitMu.Lock()
	defer db.commitMu.Unlock()

	// 取消待执行的自动提交
	db.autoCommitMu.Lock()
	if db.autoCommitTimer != nil {
		db.autoCommitTimer.Stop()
	}
	db.autoCommitMu.Unlock()

	// 1. 收集所有 BufferPool 脏页 (TablePage + IndexPage + DataPage 等)
	dirty := db.file.CollectDirty()
	if len(dirty) == 0 {
		// 无脏页 = 修改已持久, 积压的待回收页可以安全还链
		db.drainPendingFree()
		return nil // 无脏页, 跳过
	}

	// 2. 创建事务
	wal := db.BeginTxn()

	// 3. 记录所有脏页
	for _, d := range dirty {
		wal.RecordPage(d.PageID, d.Data)
	}

	// 4. HeaderPage 只在有【真实】变更时纳入 (pageID=0): allocPage 推进分配游标
	//    会置 headerDirty, 新页号必须随本笔 WAL 落盘 —— 否则崩溃重放后盘上旧游标
	//    会把这批新页二次分配。纯改写已有页的提交不动头页, 省掉 4K 镜像与回写;
	//    NextTxnID / ModifyTime 这类诊断字段只随内存游标前进, 不强制头页落盘。
	if db.file.IsHeaderDirty() {
		//    marshalHeader 要读整张头页 (含提交路径正在改的 WAL 链指针), 必须在 headerMu 内拍快照。
		db.file.withHeader(func() {
			wal.RecordPage(0, marshalHeader(db.header))
		})
	}

	// 5. 原子提交
	if err := wal.Commit(); err != nil {
		// 提交失败: 这批页必须继续留在缓存里当脏页。CollectDirty 现在不再把它们移出
		// 缓存, 所以正常情况下它们本来就在; RestoreDirty 只是兜底 (错误路径上页回收
		// 等动作也可能碰过这些页), 而且只补缺失条目, 不会用旧快照盖掉更新的内容。
		db.file.RestoreDirty(dirty)
		return err
	}

	// 6. 提交成功 → 把这一批页收成干净页。
	//    必须按快照的版本号逐页判定: 提交窗口里写入线程可能已经把某些页又改了一遍
	//    (缓存里是更新的内容, 本次 WAL 记的却是快照那份), 那些页要继续留脏, 否则新
	//    内容既不会被写盘, 脏标记还报告"已持久化" —— 该页一旦被淘汰就永久丢失。
	//    (旧实现这里是 ClearAll, 无差别地把整个缓存标干净, 恰好会抹掉这种页。)
	db.file.MarkClean(dirty)

	// 7. 提交成功 → 链指针已持久化, 积压的待回收页现在可以真正还链
	db.drainPendingFree()
	return nil
}

// ======================== 防抖自动提交 ========================

// SetAutoCommitDelay 设置防抖自动提交延迟
// delay=0 表示禁用自动提交 (仅手动 Commit)
func (db *DB) SetAutoCommitDelay(d time.Duration) {
	db.autoCommitMu.Lock()
	defer db.autoCommitMu.Unlock()
	db.autoCommitDelay = d
	if db.autoCommitTimer != nil {
		db.autoCommitTimer.Stop()
		db.autoCommitTimer = nil
	}
}

// NotifyWrite 写操作通知 (InsertRow/DeleteRow/UpdateRow 完成后调用)
// 重置防抖定时器: 如果延迟期间没有新操作, 则自动 Commit
func (db *DB) NotifyWrite() {
	if db.autoCommitDelay <= 0 {
		return
	}
	// 有活跃事务时不启动自动提交: 否则定时器到点会把事务从中途截断提交,
	// 破坏事务原子性, 并使后续 Rollback 无法真正撤销
	if db.currentTxn.Load() != nil {
		return
	}
	db.autoCommitMu.Lock()
	defer db.autoCommitMu.Unlock()

	// 复用同一个 timer 对象，避免频繁创建/销毁
	if db.autoCommitTimer == nil {
		// 首次创建
		db.autoCommitTimer = time.AfterFunc(db.autoCommitDelay, db.autoCommit)
	} else {
		// 后续重置（无需重新分配对象，减少 GC 压力）
		db.autoCommitTimer.Reset(db.autoCommitDelay)
	}
}

// beginOp 标记一个行级多页写操作的开始, 返回闭包在操作结束时调用 (defer 形态,
// 嵌套安全: 深度计数, 最外层收尾)。收尾只做一件事: 重新武装防抖提交 —— 提交
// 时机仍由防抖/阈值策略决定, 这里保证的是"任何提交都不早于操作结束"。
func (db *DB) beginOp() func() {
	db.opDepth.Add(1)
	return func() {
		if db.opDepth.Add(-1) == 0 {
			if db.file.DirtyOverThreshold() {
				db.autoCommit() // 阈值在操作期间被越过 (被栅栏压住): 立即补提交
			} else {
				db.NotifyWrite() // 否则重新武装防抖定时器
			}
		}
	}
}

// autoCommit 自动提交入口 (防抖定时器 / 脏页阈值共用)
// 有活跃事务时直接跳过, 避免把事务从中途截断提交; 行级写操作中途同样跳过
// (opDepth>0), 等操作收尾时由 beginOp 的闭包统一补提交。
func (db *DB) autoCommit() {
	if db.currentTxn.Load() != nil {
		return
	}
	if db.opDepth.Load() > 0 {
		return
	}
	_ = db.Commit()
}

// startAutoCommit 启动防抖自动提交
func (db *DB) startAutoCommit() {
	// 不立即启动定时器, 等待第一次 NotifyWrite
}

// stopAutoCommit 停止防抖自动提交
func (db *DB) stopAutoCommit() {
	db.autoCommitMu.Lock()
	defer db.autoCommitMu.Unlock()
	if db.autoCommitTimer != nil {
		db.autoCommitTimer.Stop()
		db.autoCommitTimer = nil
	}
}

// SetDirtyThreshold 设置脏页阈值（页数），超过此值自动触发 Commit
func (db *DB) SetDirtyThreshold(threshold int) {
	db.dirtyThreshold = threshold
	db.file.SetDirtyThreshold(threshold, db.autoCommit)
}

// LockWrites 锁定写入 —— 实际锁的是"页落盘"(BufferPool.WritePage 会 waitForWriteLock)。
// 锁定期 Insert/Update/Delete 仍可改内存中的页并标记脏, 只是要写磁盘时阻塞; 没有
// "写入被拒绝"的错误码。Commit/Flush 期间由引擎内部调用, 以取得稳定的磁盘视图。
func (db *DB) LockWrites() {
	db.file.LockWrites()
}

// UnlockWrites 解锁写入 (见 LockWrites 的语义说明)
func (db *DB) UnlockWrites() {
	db.file.UnlockWrites()
}

// SetBufferPoolCapacity 设置 BufferPool 容量（页数）
func (db *DB) SetBufferPoolCapacity(capacity int) {
	db.file.SetBufferPoolCapacity(capacity)
}

// marshalHeader 将 HeaderPage 序列化为字节切片 (填充到 PageSize)
func marshalHeader(h *HeaderPage) []byte {
	buf := make([]byte, PageSize)
	// 在【副本】上重算 CRC 再快照: 内存中的 HeaderCrc 可能滞后于游标 / 链
	// 指针的内存内变更 (allocPage 只标脏不落盘), 快照字节必须与快照内容自洽,
	// 否则崩溃恢复重放出头页后 CRC 与内容对不上。
	snap := *h
	snap.PrepareWrite()
	src := (*[PageSize4K]byte)(unsafe.Pointer(&snap))[:]
	copy(buf, src)
	return buf
}

// ======================== 调试诊断 ========================

// DebugDBState 打印数据库调试信息
func (db *DB) DebugDBState() {
	db.file.withHeader(func() {
		fmt.Printf("[DEBUG] Header: TablePage=%d NextFreePage=%d FreePageStart=%d FirstWAL=%d LastWAL=%d NextTxn=%d\n",
			db.header.TablePage, db.header.NextFreePage, db.header.FreePageStart,
			db.header.FirstWALPage, db.header.LastWALPage, db.header.NextTxnID)
	})
	fmt.Printf("[DEBUG] BufferPool: total=%d dirty=%d\n", db.file.bp.Len(), db.file.bp.DirtyCount())
}

// DebugTableState 打印表调试信息 (从磁盘读取, 绕过 BufferPool)
func (db *DB) DebugTableState(tableName string) {
	tbl, ok := db.tables.Find(tableName)
	if !ok {
		fmt.Printf("[DEBUG] Table '%s' not found\n", tableName)
		return
	}
	fmt.Printf("[DEBUG] Table '%s': TotalRows=%d (memory)\n", tableName, tbl.info.TotalRows)

	// 从磁盘读取 TablePage 检查 TotalRows
	buf := make([]byte, PageSize)
	if err := db.file.ReadPage(tbl.TablePageID(), buf); err == nil {
		info := (*TableInfo)(unsafe.Pointer(&buf[tbl.TableStartOffset()]))
		fmt.Printf("[DEBUG] TablePage on disk: TotalRows=%d FieldCount=%d\n", info.TotalRows, info.FieldCount)
	}

	// 检查索引元数据
	im := tbl.IndexMeta()
	if im != nil {
		for i := range im.metas {
			m := &im.metas[i]
			if m.Flag&0x80 != 0 {
				continue
			}
			fmt.Printf("[DEBUG] Index '%s': EntryPage=%d IndexType=%d\n",
				string(m.IndexName[:]), m.EntryPage, m.IndexType)

			// 从磁盘读取 IndexMetaPage 检查 EntryPage
			imBuf := make([]byte, PageSize)
			if err := db.file.ReadPage(im.PageID(), imBuf); err == nil {
				// 读取磁盘上的 EntryPage
				diskMeta := (*IndexMeta)(unsafe.Pointer(&imBuf[im.HdrSize()]))
				fmt.Printf("[DEBUG] IndexMetaPage on disk: EntryPage=%d\n", diskMeta.EntryPage)
			}

			// 统计 B+ 树 DataPage 数量
			if indexStoresRows(m) {
				if handler, ok := index.Compose.GetHandler(m.IndexType); ok {
					if rs, ok := handler.(index.RowStoreHandler); ok {
						dpIDs := rs.GetDataPageIDs(tbl.handle, m.EntryPage)
						totalRows := 0
						for _, dpID := range dpIDs {
							dpBuf := make([]byte, PageSize)
							if err := db.file.ReadPage(dpID, dpBuf); err == nil {
								// 读取 DataPage 的行数 (在 offset 18 处的 4 字节 Count)
								count := binary.LittleEndian.Uint32(dpBuf[18:])
								totalRows += int(count)
							}
						}
						fmt.Printf("[DEBUG] B+ tree: %d DataPages, %d total rows\n", len(dpIDs), totalRows)
					}
				}
			}
		}
	}
}

// DebugFreePageCount 返回空闲页数量 (调试用)
func (db *DB) DebugFreePageCount() int {
	db.file.headerMu.Lock()
	n := db.header.FreePageCount
	db.file.headerMu.Unlock()
	return int(n)
}