TsDB2/DB/core/stringwidth_test.go

package core

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

	"git.ycl.pub/yl32940/TsDB2/DB/core/page"
	"git.ycl.pub/yl32940/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_String, code))
	if width < 1024 {
		t.Fatalf("宽度码 %d 只给出 %d 字节, 装不下声明的 1024", code, width)
	}

	tbl, err := db.CreateTable("WSTR", []types.FieldInfo{
		{FieldType: types.FieldType_Uint64, FieldName: []byte("ID")},
		{FieldType: types.FieldType_String, Flag: code, FieldName: []byte("Wide")},
		// Wide2 与 Wide 同宽但【不建索引】: 用来比对"索引路径"与"全表过滤路径"
		{FieldType: types.FieldType_String, Flag: code, FieldName: []byte("Wide2")},
		// After / Idx 是错位探针: Wide 的宽度算错时先烂的是它们
		{FieldType: types.FieldType_String, FieldName: []byte("After")},
		{FieldType: types.FieldType_Uint64, 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 行宽闸门: 建表 / 加列按【溢出下界】放行, 真正卡死的是"不可溢出列撑满"。
//
// 与旧语义的区别正在宽字符串列上: 它们可溢出, 所以 4 × 2048 = 8192 的【声明】字段区不再拒
// (旧闸门会拒, 于是宽字符串表根本建不出来); 而一旦这些列【建了索引】就不能再溢出, 下界抬过
// MaxInlineRowData 就必须拒 (ErrIndexRowTooWide)。
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_String, Flag: code, FieldName: []byte(name),
		}
	}
	id := types.FieldInfo{FieldType: types.FieldType_Uint64, FieldName: []byte("id")}

	// 4 × 2048 = 8192 的声明字段区: 这些列都没建索引 ⇒ 都可溢出 ⇒ 允许建表 (旧闸门会拒)
	at, err := db.CreateTable("SPILLABLE", []types.FieldInfo{
		id, wide("A"), wide("B"), wide("C"), wide("D"),
	})
	if err != nil {
		t.Fatalf("4 × 2048 的可溢出字符串列应当允许建表: %v", err)
	}
	// 行存储索引 (主键): 下面"宽行仍能写入"必须真的落行, 否则会静默丢行
	if _, err := at.CreateIndex("pk", page.IndexTypePrimary, []byte{0}); err != nil {
		t.Fatalf("create pk: %v", err)
	}
	// 满值写入 (4 × 2048 = 8192 内容) 必须能落行且读得回: "装不下两行"的压力由值级溢出承担
	pk := at.AllocPKID()
	if _, err := at.InsertRowWithID(pk, []types.FieldValue{
		{FieldName: []byte("id"), Value: pk},
		{FieldName: []byte("A"), Value: strings.Repeat("a", 2048)},
		{FieldName: []byte("B"), Value: strings.Repeat("b", 2048)},
		{FieldName: []byte("C"), Value: strings.Repeat("c", 2048)},
		{FieldName: []byte("D"), Value: strings.Repeat("d", 2048)},
	}); err != nil {
		t.Fatalf("宽字符串行应当能写入 (靠溢出): %v", err)
	}
	if n, _ := at.Query().Count(); n != 1 {
		t.Fatalf("写入成功却读不到行 (读到 %d 行): 宽行没有真正落盘", n)
	}

	// 加列同理: 新列是可溢出字符串 ⇒ 放行 (加列闸门也按溢出下界算)
	if err := at.AddField(wide("E")); err != nil {
		t.Fatalf("加一列可溢出字符串应当放行: %v", err)
	}
	if _, present := at.GetActiveFieldType("E"); !present {
		t.Errorf("放行的新列 E 没进 schema")
	}

	// ===== 溢出闸门: 分量列建了索引就不能再溢出, 下界抬过上限必须拒 =====
	// id 是主键 (不可溢出)。再给 A/B/C 建索引后下界 = 1 + id8 + 3×2048 + D6 + E6 + 偏移表12
	// + 位图1 = 6178, 仍放行; 再给 D 建索引 ⇒ 4 列全宽 ⇒ 8220 > MaxInlineRowData, 必须拒。
	if _, err := at.CreateIndex("idx_a", page.IndexTypeBTree, []byte{1}); err != nil {
		t.Fatalf("给第 2 列建索引应当放行: %v", err)
	}
	if _, err := at.CreateIndex("idx_b", page.IndexTypeBTree, []byte{2}); err != nil {
		t.Fatalf("给第 3 列建索引应当放行: %v", err)
	}
	if _, err := at.CreateIndex("idx_c", page.IndexTypeBTree, []byte{3}); err != nil {
		t.Fatalf("给第 4 列建索引应当放行: %v", err)
	}
	if _, err := at.CreateIndex("idx_d", page.IndexTypeBTree, []byte{4}); !errors.Is(err, ErrIndexRowTooWide) {
		t.Fatalf("第 5 列再建索引会让不可溢出下界超 %d, 应当报 ErrIndexRowTooWide, 得到 %v",
			types.MaxInlineRowData, 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_Uint64, FieldName: []byte("ID")},
		{FieldType: types.FieldType_String, Flag: types.WidthCodeFor(64), FieldName: []byte("Gone")},
		{FieldType: types.FieldType_Uint64, 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_String, Flag: types.WidthCodeFor(32), FieldName: []byte("Gone")})
	if err == nil {
		t.Fatalf("复活时把 64 字节的列改成 32 字节必须被拒")
	}
	if !strings.Contains(err.Error(), "宽度") && !strings.Contains(err.Error(), "类型") {
		t.Errorf("拒因应说明是类型/宽度不一致: %v", err)
	}

	// 同类型复活: 允许, 且旧行 (按旧版本布局) 与新行 (按新版本布局) 都要读得对
	if err := tbl.AddField(types.FieldInfo{FieldType: types.FieldType_String, Flag: types.WidthCodeFor(64), 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)
		}
	}
}