TsDB2/DB/core/vacuum_identity_test.go

package core

import (
	"testing"

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

// TestVACUUMPreservesRowStoreIdentity 行存储索引必须按原始名字/类型重建:
// 旧实现无条件建成 ("pk", Primary) —— ORM 按 pk_<表名> 寻址, VACUUM 后
// Find/Delete/Update 全部 ErrFieldNotFound; 时序表被静默改成主键表。
func TestVACUUMPreservesRowStoreIdentity(t *testing.T) {
	db := fixOpenDB(t)

	// 主键表: 行存储索引用 ORM 风格命名 pk_people
	tbl, err := db.CreateTable("people", []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_people", page.IndexTypePrimary, []byte{0}); err != nil {
		t.Fatalf("建主键索引: %v", err)
	}
	for i := 1; i <= 50; i++ {
		if _, err := tbl.InsertRowWithID(uint64(i), []types.FieldValue{
			{FieldName: []byte("id"), Value: uint64(i)},
			{FieldName: []byte("age"), Value: int64(i * 2)},
		}); 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)
	}

	// 行存储索引名字/类型必须原样保留
	ntbl, ok := db.FindTable("people")
	if !ok {
		t.Fatalf("VACUUM 后表丢失")
	}
	m, h, ok := ntbl.IndexMeta().rowStoreMeta()
	if !ok {
		t.Fatalf("VACUUM 后行存储索引丢失")
	}
	if m.Name() != "pk_people" {
		t.Fatalf("行存储索引名 = %q, 想要 pk_people", m.Name())
	}
	if m.IndexType != page.IndexTypePrimary {
		t.Fatalf("行存储索引类型 = %v, 想要 Primary", m.IndexType)
	}
	_ = h

	// 按原索引名寻址 (ORM 的路径) 必须可用
	row, found := ntbl.IndexMeta().SearchByIndex("pk_people", func() []byte {
		k := make([]byte, 8)
		types.EncodeValue(types.FieldType_Uint64, 0, uint64(7), k)
		return k
	}())
	if !found || len(row) < 8 {
		t.Fatalf("VACUUM 后按 pk_people 查不到 id=7")
	}
	if got := row[8:][1:][0]; got != 0 { // rowData[0]=schemaVer
		_ = got
	}
	// 数据完整性
	if c := vacCountOf(t, ntbl); c != 50 {
		t.Fatalf("VACUUM 后行数 = %d, 想要 50", c)
	}
}

// TestVACUUMPreservesNULL VACUUM 不得把 NULL 物化成零值: 快照保 NULL,
// 重建后 NULL 位图仍在, IS-NULL 语义不变。
func TestVACUUMPreservesNULL(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"), Flag: types.FlagNullable},
	})
	if err != nil {
		t.Fatalf("建表: %v", err)
	}
	if _, err := tbl.CreateIndex("pk", page.IndexTypePrimary, []byte{0}); err != nil {
		t.Fatalf("建主键索引: %v", err)
	}
	// 3 行: NULL / 有值 / NULL
	for i := 1; i <= 3; i++ {
		vals := []types.FieldValue{
			{FieldName: []byte("id"), Value: uint64(i)},
		}
		if i == 2 {
			vals = append(vals, types.FieldValue{FieldName: []byte("v"), Value: int64(99)})
		} else {
			vals = append(vals, types.FieldValue{FieldName: []byte("v"), Value: nil})
		}
		if _, err := tbl.InsertRowWithID(uint64(i), vals); 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)
	}

	ntbl, _ := db.FindTable("t")
	// id=1 的 v 必须仍是 NULL (查不到值), id=2 是 99
	r1, _ := ntbl.FindByPK(1)
	r2, _ := ntbl.FindByPK(2)
	if hasColVal(r1, "v") {
		t.Fatalf("id=1 的 v 应为 NULL, 得到 %+v", r1)
	}
	if !hasColVal(r2, "v") {
		t.Fatalf("id=2 的 v 应为 99, 得到 %+v", r2)
	}
	if c := vacCountOf(t, ntbl); c != 3 {
		t.Fatalf("行数 = %d, 想要 3", c)
	}
}

// TestVACUUMDroppedColumnTable "删过列、且被索引列在被删列之后"的表,
// VACUUM 必须成功 (FieldIDs 重映射), 且索引仍可用。
func TestVACUUMDroppedColumnTable(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("dead")},
		{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{2}); err != nil {
		t.Fatalf("建 idx_age: %v", err)
	}
	for i := 1; i <= 30; i++ {
		if _, err := tbl.InsertRowWithID(uint64(i), []types.FieldValue{
			{FieldName: []byte("id"), Value: uint64(i)},
			{FieldName: []byte("dead"), Value: int64(i)},
			{FieldName: []byte("age"), Value: int64(i % 40)},
		}); err != nil {
			t.Fatalf("插入 %d: %v", i, err)
		}
	}
	if err := tbl.DropField("dead"); err != nil {
		t.Fatalf("DropField: %v", err)
	}
	if err := db.Commit(); err != nil {
		t.Fatalf("commit: %v", err)
	}

	if _, err := db.VACUUM(); err != nil {
		t.Fatalf("删过列的表 VACUUM 失败: %v", err)
	}

	ntbl, _ := db.FindTable("t")
	if c := vacCountOf(t, ntbl); c != 30 {
		t.Fatalf("VACUUM 后行数 = %d, 想要 30", c)
	}
	// 重建的 idx_age 必须仍按新下标工作: age=5 → id ∈ {5, 45} 中 45>30 即仅 id=5
	n, err := ntbl.Query().Where(Eq("age", int64(5), 0)).Count()
	if err != nil || n != 1 {
		t.Fatalf("VACUUM 后 idx_age 等值 count=%d err=%v, 想要 1", n, err)
	}
}

// TestVACUUMTimeSeriesKeepsTimeIndex 时序表 VACUUM 后仍是时序索引,
// 时间范围查询正常。
func TestVACUUMTimeSeriesKeepsTimeIndex(t *testing.T) {
	db := fixOpenDB(t)
	tbl, err := db.CreateTable("tick", []types.FieldInfo{
		{FieldType: types.FieldType_Uint64, FieldName: []byte("ts")},
		{FieldType: types.FieldType_Float64, FieldName: []byte("price")},
	})
	if err != nil {
		t.Fatalf("建表: %v", err)
	}
	if _, err := tbl.CreateIndex("ts_tick", page.IndexTypeTime, []byte{0}); err != nil {
		t.Fatalf("建时序索引: %v", err)
	}
	for i := 1; i <= 100; i++ {
		if _, err := tbl.InsertRowWithID(uint64(i), []types.FieldValue{
			{FieldName: []byte("ts"), Value: uint64(i * 10)},
			{FieldName: []byte("price"), Value: float64(i) + 0.5},
		}); 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)
	}

	ntbl, _ := db.FindTable("tick")
	m, _, ok := ntbl.IndexMeta().rowStoreMeta()
	if !ok || m.IndexType != page.IndexTypeTime {
		t.Fatalf("VACUUM 后行存储索引类型 = %+v, 想要 Time", m)
	}
	if m.Name() != "ts_tick" {
		t.Fatalf("VACUUM 后时序索引名 = %q, 想要 ts_tick", m.Name())
	}
	// 时间范围查询
	n, err := ntbl.Query().Where(And(
		Ge("ts", uint64(300), 0), Le("ts", uint64(500), 0))).Count()
	if err != nil || n != 21 {
		t.Fatalf("时序范围 count=%d err=%v, 想要 21 (300..500 步进 10)", n, err)
	}
}

// vacCountOf 拿表行数 (测试辅助)。
func vacCountOf(t *testing.T, tbl *Table) int64 {
	t.Helper()
	c, err := tbl.Query().Count()
	if err != nil {
		t.Fatalf("count: %v", err)
	}
	return c
}

// hasColVal 行字段值里是否有指定列 (NULL 列不在其中)。
func hasColVal(vals []types.FieldValue, name string) bool {
	for _, fv := range vals {
		if string(fv.FieldName) == name {
			return true
		}
	}
	return false
}