TsDB2/DB/core/stringwidth_test.go

package core

import (
	"bytes"
	"errors"
	"fmt"
	"strings"
	"testing"

	"TsDB2/DB/core/page"
	"TsDB2/DB/core/types"
)

// 本文件是"字符串列宽度由列级参数决定"这一改动的端到端验证。
//
// 要钉住的性质只有一条, 但它极其容易静默失效: **行内字节与索引键字节同源**。
// 宽度算错的后果不是报错, 而是该列之后的每一个字段都从错位偏移解码 —— 行照样能读,
// 字段也都有值, 只是全是别的字段的内容。所以这里除了校验被加宽的列本身, 还特意放了
// 两个"错位探针" (After / Idx): 只有它们的值也对, 才说明布局是对的。

// stringTrim 定长字符串列解码出来是"补满列宽"的字节, 比较前去掉补零。
func stringTrim(v interface{}) string {
	b, ok := v.([]byte)
	if !ok {
		return fmt.Sprintf("<非字节载体 %T>", v)
	}
	return string(bytes.TrimRight(b, "\x00"))
}

// TestStringColumnWidthEndToEnd 宽度码列: 写入/读取/索引/超宽报错。
func TestStringColumnWidthEndToEnd(t *testing.T) {
	db := fixOpenDB(t)

	code := types.WidthCodeFor(1024) // 声明 1024 字节
	width := int(types.FieldSize(types.FieldType_256byte, code))
	if width < 1024 {
		t.Fatalf("宽度码 %d 只给出 %d 字节, 装不下声明的 1024", code, width)
	}

	tbl, err := db.CreateTable("WSTR", []types.FieldInfo{
		{FieldType: types.FieldType_8byte, FieldName: []byte("ID")},
		{FieldType: types.FieldType_256byte, Flag: code, FieldName: []byte("Wide")},
		// Wide2 与 Wide 同宽但【不建索引】: 用来比对"索引路径"与"全表过滤路径"
		{FieldType: types.FieldType_256byte, Flag: code, FieldName: []byte("Wide2")},
		// After / Idx 是错位探针: Wide 的宽度算错时先烂的是它们
		{FieldType: types.FieldType_64byte, FieldName: []byte("After")},
		{FieldType: types.FieldType_8byte, FieldName: []byte("Idx")},
	})
	if err != nil {
		t.Fatalf("create table: %v", err)
	}
	// 主键索引 = 承载行数据的行存储: 没有它 InsertRow 无处落行 (静默丢行), 全表扫也读不到
	if _, err := tbl.CreateIndex("pk", page.IndexTypePrimary, []byte{0}); err != nil {
		t.Fatalf("create pk: %v", err)
	}
	if _, err := tbl.CreateIndex("idx_wide", page.IndexTypeBTree, []byte{1}); err != nil {
		t.Fatalf("在宽度码列上建索引: %v", err)
	}

	// 1 字节 / 500 字节 / 正好填满列宽: 最后一档覆盖"边界处不允许溢出"
	vals := []string{"a", strings.Repeat("b", 500), strings.Repeat("c", width)}
	after := func(i int) string { return fmt.Sprintf("%012d", 1000+i) }
	for i, v := range vals {
		if _, err := tbl.InsertRow([]types.FieldValue{
			{FieldName: []byte("ID"), Value: uint64(i + 1)},
			{FieldName: []byte("Wide"), Value: v},
			{FieldName: []byte("Wide2"), Value: v},
			{FieldName: []byte("After"), Value: after(i)},
			{FieldName: []byte("Idx"), Value: uint64(7 + i)},
		}); err != nil {
			t.Fatalf("insert %d: %v", i, err)
		}
	}

	rows, err := tbl.Query().OrderBy("Idx").Find()
	if err != nil {
		t.Fatalf("find: %v", err)
	}
	if len(rows) != len(vals) {
		t.Fatalf("应读出 %d 行, 得到 %d", len(vals), len(rows))
	}
	for i, row := range rows {
		if got := stringTrim(newTypeRowVal(t, row, "Wide")); got != vals[i] {
			t.Errorf("第 %d 行 Wide 长度 %d, 期望长度 %d (内容不一致)",
				i, len(got), len(vals[i]))
		}
		if got := stringTrim(newTypeRowVal(t, row, "Wide2")); got != vals[i] {
			t.Errorf("第 %d 行 Wide2 长度 %d, 期望长度 %d", i, len(got), len(vals[i]))
		}
		// 这两行才是"宽度正确"的真正证据
		if got := stringTrim(newTypeRowVal(t, row, "After")); got != after(i) {
			t.Errorf("第 %d 行 After = %q, 期望 %q —— 宽度算错时这里会先烂",
				i, got, after(i))
		}
		if got := newTypeRowVal(t, row, "Idx"); got != uint64(7+i) {
			t.Errorf("第 %d 行 Idx = %v, 期望 %d", i, got, 7+i)
		}
	}

	// 索引路径 (Wide) 与全表过滤路径 (Wide2) 必须给出同一个答案
	probe := vals[1]
	viaIdx, err := tbl.Query().Where(Eq("Wide", probe, types.FieldType(0))).Count()
	if err != nil {
		t.Fatalf("索引路径查询: %v", err)
	}
	viaScan, err := tbl.Query().Where(Eq("Wide2", probe, types.FieldType(0))).Count()
	if err != nil {
		t.Fatalf("全表过滤查询: %v", err)
	}
	if viaIdx != 1 || viaScan != 1 {
		t.Fatalf("索引路径 %d 行 / 全表过滤 %d 行, 都应为 1 (不一致即行内字节与索引键不同源)",
			viaIdx, viaScan)
	}

	// 定长列的比较是【整列宽】比较: 128 字节的 'b' 前缀不能命中 500 字节的行
	if n, _ := tbl.Query().Where(Eq("Wide", strings.Repeat("b", 128), types.FieldType(0))).Count(); n != 0 {
		t.Errorf("前缀 %q 不该命中 500 字节的行, 却命中了 %d 行 (比较没有按列宽补零)",
			strings.Repeat("b", 8)+"...", n)
	}

	// ===== 超宽: 报错, 不截断 (旧行为是静默截断: 写 20 字节的列声明 10 也会成功落库) =====
	before, _ := tbl.Query().Count()
	_, err = tbl.InsertRow([]types.FieldValue{
		{FieldName: []byte("ID"), Value: uint64(99)},
		{FieldName: []byte("Wide"), Value: strings.Repeat("d", width+1)},
		{FieldName: []byte("Wide2"), Value: "x"},
		{FieldName: []byte("After"), Value: after(99)},
		{FieldName: []byte("Idx"), Value: uint64(99)},
	})
	if !errors.Is(err, types.ErrValueRange) {
		t.Fatalf("写入 %d 字节 (列宽 %d) 应报 ErrValueRange, 得到 %v", width+1, width, err)
	}
	// 失败不能留下半条行, 也不能让表变得不可用
	if after, _ := tbl.Query().Count(); after != before {
		t.Errorf("写失败后行数从 %d 变成 %d: 失败的行有残留", before, after)
	}
	if n, _ := tbl.Query().Where(Eq("Idx", uint64(99), types.FieldType(0))).Count(); n != 0 {
		t.Errorf("被判超宽的行竟然进了表 (命中 %d 行)", n)
	}
	if _, err := tbl.InsertRow([]types.FieldValue{
		{FieldName: []byte("ID"), Value: uint64(100)},
		{FieldName: []byte("Wide"), Value: "ok"},
		{FieldName: []byte("Wide2"), Value: "ok"},
		{FieldName: []byte("After"), Value: after(100)},
		{FieldName: []byte("Idx"), Value: uint64(100)},
	}); err != nil {
		t.Errorf("超宽写入失败之后, 正常写入也失败了: %v", err)
	}
}

// TestRowSizeGate 整行占用闸门: 建表 / 加列 / 写行三处都要拦, 且错误要说清哪几列占了多少。
//
// 判据不是"字段区 ≤ 8K": 4 × 2048 = 8192 的字段区看着正好 8K, 加上每行固定开销与 NULL
// 位图之后就装不下两行了, 所以必须拒。
func TestRowSizeGate(t *testing.T) {
	db := fixOpenDB(t)
	code := types.WidthCodeFor(types.MaxInlineColumnWidth) // 每列 2048 字节
	wide := func(name string) types.FieldInfo {
		return types.FieldInfo{
			FieldType: types.FieldType_256byte, Flag: code, FieldName: []byte(name),
		}
	}

	// 3 × 2048 = 6144 字段区: 加上开销仍装得下两行, 允许建表
	if _, err := db.CreateTable("NARROW", []types.FieldInfo{
		wide("A"), wide("B"), wide("C"),
	}); err != nil {
		t.Fatalf("3 × 2048 建表: %v", err)
	}
	at, ok := db.FindTable("NARROW")
	if !ok {
		t.Fatal("找不到 NARROW")
	}
	// 行存储索引 (主键): 下面"宽行仍能写入"必须真的落行, 否则 InsertRow 会静默丢行
	if _, err := at.CreateIndex("pk", page.IndexTypePrimary, []byte{0}); err != nil {
		t.Fatalf("create pk: %v", err)
	}
	// 6144 + 2048 = 8192 字段区: 一页装不下两行, AddField 必须被拒
	if err := at.AddField(wide("D")); err == nil {
		t.Errorf("4 × 2048 = 8192 的字段区装不下两行 (上限 %d), AddField 必须被拒", types.MaxRowDataSize)
	} else if !strings.Contains(err.Error(), "行宽") {
		t.Errorf("超宽错误应说明是行宽问题: %v", err)
	}
	// 被拒的列不能进 schema, 表本身也必须还能用
	if _, present := at.GetActiveFieldType("D"); present {
		t.Errorf("行宽超限的列 D 被加进了 schema")
	}
	if _, err := at.InsertRow([]types.FieldValue{
		{FieldName: []byte("A"), Value: "a"},
		{FieldName: []byte("B"), Value: "b"},
		{FieldName: []byte("C"), Value: "c"},
	}); err != nil {
		t.Errorf("闸门内的表应当仍能写入: %v", err)
	} else if n, _ := at.Query().Count(); n != 1 {
		t.Errorf("写入成功却读不到行 (读到 %d 行): 宽行没有真正落盘", n)
	}

	// 整表建的时候就超: 必须拒绝, 并列出各列宽度 (只说"超了"没法用, 用户要知道砍哪一列)
	_, err := db.CreateTable("OVER", []types.FieldInfo{
		wide("A"), wide("B"), wide("C"), wide("D"),
	})
	if err == nil {
		t.Fatalf("4 × 2048 = 8192 的字段区超过 %d, 必须拒绝建表", types.MaxRowDataSize)
	}
	for _, want := range []string{"行宽", "D=2048", "8192"} {
		if !strings.Contains(err.Error(), want) {
			t.Errorf("建表超宽的错误信息缺少 %q: %v", want, err)
		}
	}
}

// TestMaxRowDataSizeFitsTwoRows 建列闸门推出的上限必须真的"一页两行"。
//
// 这是 types.MaxRowDataSize 的【唯一来由】(推导见那里的注释), 但那个上限算在 types 包里、
// 用的是页面布局的镜像值, 所以在这里用真实的 PageSize / DataPageHdrSize 复核一遍: 镜像一旦
// 与布局漂移, 上限就变成空头承诺 —— 建表时放行, 写第二行时才发现装不下。
//
// 顺带把 8192 钉在"拦得住"的那一侧: 8K 是个好看的数, 但每行还有固定开销与 NULL 位图,
// 两行合计会越过页里能用的字节数 —— 旧上限就是这么来的, 这个断言防止它被改回去。
func TestMaxRowDataSizeFitsTwoRows(t *testing.T) {
	// 一页能放行的行字节数: 末字节用不上 (InsertRowSortedByOffset 用 ">=" 提前判失败)
	avail := int(PageSize) - int(page.DataPageHdrSize) - 1

	// 上限本身: 两行正好装得下
	if 2*types.MaxRowDataSize > avail {
		t.Errorf("上限 %d 的两行 (%d 字节) 装不进一页 (%d)",
			types.MaxRowDataSize, 2*types.MaxRowDataSize, avail)
	}
	// 且它是满足该性质的【最大】值 —— 否则上限偏小, 白白少用页里一半空间
	if 2*(types.MaxRowDataSize+1) <= avail {
		t.Errorf("上限 %d 偏小: 再宽 1 字节的两行 (%d) 仍装得进一页 (%d)",
			types.MaxRowDataSize, 2*(types.MaxRowDataSize+1), avail)
	}
	// 旧上限 (8192 字节字段区): 加上每行开销后两行装不下 —— 它从来没兑现过这条政策
	oldWidestRow := 8192 + 4 + 8 + 1 + types.NullBitmapLen(1)
	if 2*oldWidestRow <= avail {
		t.Errorf("8192 字节字段区竟然装得下两行 (%d × 2 ≤ %d): 那 8192 本来就没问题",
			oldWidestRow, avail)
	}
}

// TestReviveFieldKeepsWidth 复活已删除的列时不允许改类型/宽度。
//
// master registry 里的类型与宽度码是【所有历史版本共用】的行偏移依据: 复活时把它换掉,
// 已经落盘的旧行仍按旧宽度排布, 却会被按新宽度解析 —— 该列之后的字段全部静默错位。
func TestReviveFieldKeepsWidth(t *testing.T) {
	db := fixOpenDB(t)
	tbl, err := db.CreateTable("REVIVE", []types.FieldInfo{
		{FieldType: types.FieldType_8byte, FieldName: []byte("ID")},
		{FieldType: types.FieldType_64byte, FieldName: []byte("Gone")},
		{FieldType: types.FieldType_8byte, FieldName: []byte("Probe")},
	})
	if err != nil {
		t.Fatalf("create table: %v", err)
	}
	if _, err := tbl.CreateIndex("pk", page.IndexTypePrimary, []byte{0}); err != nil {
		t.Fatalf("create pk: %v", err)
	}
	if _, err := tbl.InsertRow([]types.FieldValue{
		{FieldName: []byte("ID"), Value: uint64(1)},
		{FieldName: []byte("Gone"), Value: "keepme"},
		{FieldName: []byte("Probe"), Value: uint64(42)},
	}); err != nil {
		t.Fatalf("insert: %v", err)
	}

	if err := tbl.DropField("Gone"); err != nil {
		t.Fatalf("drop: %v", err)
	}
	err = tbl.AddField(types.FieldInfo{FieldType: types.FieldType_32byte, FieldName: []byte("Gone")})
	if err == nil {
		t.Fatalf("复活时把 64byte 的列改成 32byte 必须被拒")
	}
	if !strings.Contains(err.Error(), "宽度") && !strings.Contains(err.Error(), "类型") {
		t.Errorf("拒因应说明是类型/宽度不一致: %v", err)
	}

	// 同类型复活: 允许, 且旧行 (按旧版本布局) 与新行 (按新版本布局) 都要读得对
	if err := tbl.AddField(types.FieldInfo{FieldType: types.FieldType_64byte, FieldName: []byte("Gone")}); err != nil {
		t.Fatalf("同类型复活应当成功: %v", err)
	}
	if _, err := tbl.InsertRow([]types.FieldValue{
		{FieldName: []byte("ID"), Value: uint64(2)},
		{FieldName: []byte("Gone"), Value: "second"},
		{FieldName: []byte("Probe"), Value: uint64(43)},
	}); err != nil {
		t.Fatalf("复活后 insert: %v", err)
	}

	rows, err := tbl.Query().OrderBy("Probe").Find()
	if err != nil {
		t.Fatalf("find: %v", err)
	}
	if len(rows) != 2 {
		t.Fatalf("应读出 2 行, 得到 %d", len(rows))
	}
	wantGone := []string{"keepme", "second"}
	for i, row := range rows {
		if got := stringTrim(newTypeRowVal(t, row, "Gone")); got != wantGone[i] {
			t.Errorf("第 %d 行 Gone = %q, 期望 %q (旧行/新行用了不同版本的布局)",
				i, got, wantGone[i])
		}
		if got := newTypeRowVal(t, row, "Probe"); got != uint64(42+i) {
			t.Errorf("第 %d 行 Probe = %v, 期望 %d", i, got, 42+i)
		}
	}
}