TsDB2/DB/core/pk_int64_test.go

package core

// Int64 (及更窄整型) 主键镜像列的回归测试。
//
// 背景: 主键 B+ 树的键必须是 pkID(BE) —— 查询/删除/更新侧都按 pkID 定位。
// 旧实现插入/删除/更新统一用 extractIndexKey (取镜像列的【存储字节】) 当树键:
// Uint64 镜像"存储字节 == BE"侥幸正确, 而 Int64 等带 sort-key 编码的类型
// 存储字节 ≠ pkID-BE —— 插入把行挂到错键上, 按 pkID 查找退化成全叶扫描,
// 删除/更新按错键静默失效 (ErrNotFound 按索引逐个吞掉, 不报错也不动数据)。
// 修复: 主键行存储的键恒为 pkID(BE) (rowStoreInsertKey), 与时序索引的
// "时间分量"键源区分; 三侧 (插/删/改) 同一键源。

import (
	"testing"

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

func int64PKTable(t *testing.T, db *DB, name string) *Table {
	t.Helper()
	tbl, err := db.CreateTable(name, []types.FieldInfo{
		{FieldType: types.FieldType_Int64, 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)
	}
	return tbl
}

func TestInt64PKDeleteUpdate(t *testing.T) {
	db := fixOpenDB(t)
	tbl := int64PKTable(t, db, "t")

	// 单行 + 批量两条路径都覆盖
	tbl.InsertRowWithID(1, []types.FieldValue{
		{FieldName: []byte("id"), Value: int64(1)},
		{FieldName: []byte("v"), Value: int64(10)},
	})
	if _, err := tbl.InsertRowsWithIDs([]uint64{2, 3}, [][]types.FieldValue{
		{{FieldName: []byte("id"), Value: int64(2)}, {FieldName: []byte("v"), Value: int64(20)}},
		{{FieldName: []byte("id"), Value: int64(3)}, {FieldName: []byte("v"), Value: int64(30)}},
	}); err != nil {
		t.Fatalf("批量插入: %v", err)
	}
	if err := db.Commit(); err != nil {
		t.Fatal(err)
	}

	// 按 pkID 删除: 必须真正消失 (修复前: 静默失效)
	if err := tbl.DeleteRowByPK(2); err != nil {
		t.Fatalf("delete: %v", err)
	}
	if _, err := tbl.FindByPK(2); err == nil {
		t.Fatalf("pkID=2 删除后仍可见 — Int64 主键删除静默失效")
	}
	if _, err := tbl.FindByPK(1); err != nil {
		t.Fatalf("pkID=1 被误删: %v", err)
	}
	if got := tbl.TotalRows(); got != 2 {
		t.Fatalf("TotalRows=%d, 想要 2", got)
	}

	// 更新: 修复前同样按错键静默失效
	if err := tbl.UpdateRowByPK(1, []types.FieldValue{
		{FieldName: []byte("v"), Value: int64(99)},
	}); err != nil {
		t.Fatalf("update: %v", err)
	}
	vals, err := tbl.FindByPK(1)
	if err != nil {
		t.Fatalf("find: %v", err)
	}
	var v int64
	for _, fv := range vals {
		if string(fv.FieldName) == "v" {
			v, _ = fv.Value.(int64)
		}
	}
	if v != 99 {
		t.Fatalf("更新未生效: v=%d, 想要 99", v)
	}

	// 二级索引不受键源修复影响 (仍按分量列字节编键, 插入/查询同侧一致)
	if _, err := tbl.CreateIndex("idx_v", page.IndexTypeBTree, []byte{1}); err != nil {
		t.Fatalf("建二级索引: %v", err)
	}
	if _, err := tbl.InsertRowWithID(4, []types.FieldValue{
		{FieldName: []byte("id"), Value: int64(4)},
		{FieldName: []byte("v"), Value: int64(99)}, // 与 id=1 同 v, 非唯一索引允许
	}); err != nil {
		t.Fatalf("insert 4: %v", err)
	}
	n := 0
	if err := tbl.Query().Where(Eq("v", int64(99), types.FieldType(0))).Scan(func(row []types.FieldValue) bool {
		n++
		return true
	}); err != nil {
		t.Fatal(err)
	}
	if n != 2 {
		t.Fatalf("二级索引应命中 2 行 (v=99), 实际 %d", n)
	}
}