TsDB2/DB/core/audit_round3_test.go

package core

import (
	"testing"

	"git.ycl.pub/yl32940/TsDB2/DB/core/page"
	"git.ycl.pub/yl32940/TsDB2/DB/core/types"
)

// TestTxnRollbackAfterEmptyingPage 事务内把一页删空再回滚:
// 旧实现删空即直写盘还链, 回滚后树仍引用该页 —— 后续分配把它复用后,
// 回滚"恢复"的行读到别人的页 (数据损坏)。延迟回收下回滚必须完整复活数据页。
func TestTxnRollbackAfterEmptyingPage(t *testing.T) {
	db := fixOpenDB(t)
	tbl, err := db.CreateTable("t", []types.FieldInfo{
		{FieldType: types.FieldType_Uint64, FieldName: []byte("id")},
		{FieldType: types.FieldType_Int64, FieldName: []byte("v")},
	})
	if err != nil {
		t.Fatalf("建表: %v", err)
	}
	if _, err := tbl.CreateIndex("pk", page.IndexTypePrimary, []byte{0}); err != nil {
		t.Fatalf("建主键索引: %v", err)
	}
	// 100 行, 单页装得下
	for i := 1; i <= 100; i++ {
		if _, err := tbl.InsertRowWithID(uint64(i), []types.FieldValue{
			{FieldName: []byte("id"), Value: uint64(i)},
			{FieldName: []byte("v"), Value: int64(i)},
		}); err != nil {
			t.Fatalf("插入 %d: %v", i, err)
		}
	}
	if err := db.Commit(); err != nil {
		t.Fatalf("commit: %v", err)
	}

	// 事务内删空
	tx := db.BeginTransaction()
	for i := 1; i <= 100; i++ {
		if err := tbl.DeleteRowByPK(uint64(i)); err != nil {
			t.Fatalf("事务内删除 %d: %v", i, err)
		}
	}
	if err := tx.Rollback(); err != nil {
		t.Fatalf("rollback: %v", err)
	}

	// 回滚后 100 行必须完整回来
	if c := vacCountOf(t, tbl); c != 100 {
		t.Fatalf("回滚后行数 = %d, 想要 100 (删空页的直写盘还链会让回滚丢数据)", c)
	}
	row, found := tbl.IndexMeta().ResolveByPK(1)
	if !found || len(row) < 8 {
		t.Fatalf("回滚后 FindByPK(1) 失败 (树引用了被复用的页)")
	}

	// 继续写入: 若页被误还链, 这里会复用它并踩坏回滚的数据
	for i := 101; i <= 150; i++ {
		if _, err := tbl.InsertRowWithID(uint64(i), []types.FieldValue{
			{FieldName: []byte("id"), Value: uint64(i)},
			{FieldName: []byte("v"), Value: int64(i)},
		}); err != nil {
			t.Fatalf("插入 %d: %v", i, err)
		}
	}
	if err := db.Commit(); err != nil {
		t.Fatalf("commit: %v", err)
	}
	if c := vacCountOf(t, tbl); c != 150 {
		t.Fatalf("继续写入后行数 = %d, 想要 150", c)
	}
	if _, found := tbl.IndexMeta().ResolveByPK(1); !found {
		t.Fatalf("复用后 FindByPK(1) 丢失 (空闲页与活页双重分配)")
	}
	if _, found := tbl.IndexMeta().ResolveByPK(150); !found {
		t.Fatalf("FindByPK(150) 丢失")
	}
}

// TestTxnWorksAfterVacuum 回归: VACUUM 换引擎状态后, 事务的 Rollback 必须
// 真正生效 (markTouched/allocated/pendingFree 闭包重绑到当前句柄)。
func TestTxnWorksAfterVacuum(t *testing.T) {
	db := fixOpenDB(t)
	tbl, err := db.CreateTable("t", []types.FieldInfo{
		{FieldType: types.FieldType_Uint64, FieldName: []byte("id")},
		{FieldType: types.FieldType_Int64, FieldName: []byte("v")},
	})
	if err != nil {
		t.Fatalf("建表: %v", err)
	}
	if _, err := tbl.CreateIndex("pk", page.IndexTypePrimary, []byte{0}); err != nil {
		t.Fatalf("建主键索引: %v", err)
	}
	for i := 1; i <= 10; i++ {
		if _, err := tbl.InsertRowWithID(uint64(i), []types.FieldValue{
			{FieldName: []byte("id"), Value: uint64(i)},
			{FieldName: []byte("v"), Value: int64(i)},
		}); err != nil {
			t.Fatalf("插入 %d: %v", i, err)
		}
	}
	if err := db.Commit(); err != nil {
		t.Fatalf("commit: %v", err)
	}
	if _, err := db.VACUUM(); err != nil {
		t.Fatalf("VACUUM: %v", err)
	}
	// VACUUM 换了引擎状态, 须按当前句柄重新取表 (旧表对象的句柄/闭包属于旧状态)
	var ok bool
	tbl, ok = db.FindTable("t")
	if !ok {
		t.Fatalf("VACUUM 后表丢失")
	}

	// VACUUM 后: 事务插入再回滚 → 行必须不存在
	tx := db.BeginTransaction()
	if _, err := tbl.InsertRowWithID(999, []types.FieldValue{
		{FieldName: []byte("id"), Value: uint64(999)},
		{FieldName: []byte("v"), Value: int64(999)},
	}); err != nil {
		t.Fatalf("事务内插入: %v", err)
	}
	if err := tx.Rollback(); err != nil {
		t.Fatalf("rollback: %v", err)
	}
	if _, found := tbl.IndexMeta().ResolveByPK(999); found {
		t.Fatalf("VACUUM 后 Rollback 失效: 已回滚的行仍在 (闭包未重绑, markTouched 落空)")
	}
	if c := vacCountOf(t, tbl); c != 10 {
		t.Fatalf("回滚后行数 = %d, 想要 10", c)
	}

	// VACUUM 后: 事务插入再提交 → 行必须存在
	tx2 := db.BeginTransaction()
	if _, err := tbl.InsertRowWithID(1000, []types.FieldValue{
		{FieldName: []byte("id"), Value: uint64(1000)},
		{FieldName: []byte("v"), Value: int64(1000)},
	}); err != nil {
		t.Fatalf("事务内插入: %v", err)
	}
	if err := tx2.Commit(); err != nil {
		t.Fatalf("commit: %v", err)
	}
	if _, found := tbl.IndexMeta().ResolveByPK(1000); !found {
		t.Fatalf("VACUUM 后事务提交的行丢失")
	}
}

// TestInDuplicateValueAggCount IN 重复值不得让聚合重复计数:
// 行交付路径有去重, raw 聚合路径 (Count/Sum) 依赖规划期区间去重。
func TestInDuplicateValueAggCount(t *testing.T) {
	db := fixOpenDB(t)
	tbl, err := db.CreateTable("t", []types.FieldInfo{
		{FieldType: types.FieldType_Uint64, FieldName: []byte("id")},
		{FieldType: types.FieldType_Int64, FieldName: []byte("age")},
	})
	if err != nil {
		t.Fatalf("建表: %v", err)
	}
	if _, err := tbl.CreateIndex("pk", page.IndexTypePrimary, []byte{0}); err != nil {
		t.Fatalf("建主键索引: %v", err)
	}
	if _, err := tbl.CreateIndex("idx_age", page.IndexTypeBTree, []byte{1}); err != nil {
		t.Fatalf("建二级索引: %v", err)
	}
	for i := 1; i <= 100; i++ {
		if _, err := tbl.InsertRowWithID(uint64(i), []types.FieldValue{
			{FieldName: []byte("id"), Value: uint64(i)},
			{FieldName: []byte("age"), Value: int64(i % 50)},
		}); err != nil {
			t.Fatalf("插入 %d: %v", i, err)
		}
	}
	if err := db.Commit(); err != nil {
		t.Fatalf("commit: %v", err)
	}

	// age=7 → id ∈ {7,57} 两行; 重复写三遍仍是 2
	n, err := tbl.Query().Where(In("age", []interface{}{int64(7), int64(7), int64(7)}, 0)).Count()
	if err != nil || n != 2 {
		t.Fatalf("重复 IN 的 Count = %d (err=%v), 想要 2", n, err)
	}
	// 行交付路径同样不得重复
	got := collectAges(t, tbl.Query().Where(In("age", []interface{}{int64(7), int64(7)}, 0)))
	if len(got) != 2 {
		t.Fatalf("重复 IN 的行交付 = %d 行, 想要 2", len(got))
	}
}

// TestDropRowStoreComponentRejected 行存储索引的分量列不得删除:
// 删了之后新行提不出键、不进任何索引, 插入"成功"但行静默丢失。
func TestDropRowStoreComponentRejected(t *testing.T) {
	db := fixOpenDB(t)
	tbl, err := db.CreateTable("t", []types.FieldInfo{
		{FieldType: types.FieldType_Uint64, FieldName: []byte("id")},
		{FieldType: types.FieldType_Int64, FieldName: []byte("v")},
	})
	if err != nil {
		t.Fatalf("建表: %v", err)
	}
	if _, err := tbl.CreateIndex("pk", page.IndexTypePrimary, []byte{0}); err != nil {
		t.Fatalf("建主键索引: %v", err)
	}
	if err := tbl.DropField("id"); err == nil {
		t.Fatalf("删除行存储索引分量列必须报错, 却成功了")
	}
	// 普通列仍可删除
	if err := tbl.DropField("v"); err != nil {
		t.Fatalf("删除普通列不应报错: %v", err)
	}
}