TsDB2/DB/core/limits_test.go

package core

import (
	"errors"
	"strconv"
	"strings"
	"testing"

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

// 本文件集中钉死"每一处限制在触发时都必须有报错或提示"。
//
// 这些闸门原先有的缺口是【静默】的 —— 越界不报错, 而是把 1 字节的 FieldCount /
// 字段索引 / schemaVer 写回绕掉, 读回来整表错位。所以每个用例除了"必须报错",
// 还检查"报错里要有用户能据此改代码的信息"(上限与实际值 / 列名)。

// TestFieldCountGate 列数上限: 建表与加列必须卡同一条。
//
// 只在 TablePage.Create 卡住而 AddField 不限, 就能反复加列把 master registry 推过
// 255, 落盘时把 FieldCount 与字段索引 (都是 1 字节) 截断成错位 schema。
func TestFieldCountGate(t *testing.T) {
	if err := checkFieldCount(0); err == nil || !errors.Is(err, ErrInvalidFieldCount) {
		t.Fatalf("0 列应报 ErrInvalidFieldCount, 得到 %v", err)
	}
	if err := checkFieldCount(MaxFieldsPerTable); err != nil {
		t.Fatalf("%d 列应当放行: %v", MaxFieldsPerTable, err)
	}
	err := checkFieldCount(MaxFieldsPerTable + 1)
	if err == nil || !errors.Is(err, ErrInvalidFieldCount) {
		t.Fatalf("%d 列应报 ErrInvalidFieldCount, 得到 %v", MaxFieldsPerTable+1, err)
	}
	// 只说"字段数量非法"没法用: 用户要知道上限是多少、本次是多少
	for _, want := range []string{strconv.Itoa(MaxFieldsPerTable), strconv.Itoa(MaxFieldsPerTable + 1)} {
		if !strings.Contains(err.Error(), want) {
			t.Errorf("列数超限的报错缺少 %q: %v", want, err)
		}
	}

	// 加列路径也必须走同一条闸门
	db := fixOpenDB(t)
	fields := make([]types.FieldInfo, MaxFieldsPerTable)
	for i := range fields {
		fields[i] = types.FieldInfo{
			FieldType: types.FieldType_Uint8,
			FieldName: []byte("c" + strconv.Itoa(i)),
		}
	}
	if _, err := db.CreateTable("FULL", fields); err != nil {
		t.Fatalf("建一张 %d 列的表应当成功: %v", MaxFieldsPerTable, err)
	}
	tbl, ok := db.FindTable("FULL")
	if !ok {
		t.Fatal("找不到 FULL")
	}
	err = tbl.AddField(types.FieldInfo{
		FieldType: types.FieldType_Uint8, FieldName: []byte("oneTooMany"),
	})
	if err == nil || !errors.Is(err, ErrInvalidFieldCount) {
		t.Fatalf("超过 %d 列后 AddField 应报 ErrInvalidFieldCount, 得到 %v", MaxFieldsPerTable, err)
	}
	if _, present := tbl.GetActiveFieldType("oneTooMany"); present {
		t.Error("被拒的列不能进 schema")
	}
}

// TestSchemaVersionGate schema 版本号是每行行首 1 字节, 版本条数也是 1 字节: 到上限再演进
// 会回绕成 0, 而 0 不是任何已知版本 → getVersion 返回 nil, 该行 NULL 位图取不到、所有列
// 静默读成 0/NULL。必须在演进之前拒绝, 而不是"写进去再读不出来"。
func TestSchemaVersionGate(t *testing.T) {
	db := fixOpenDB(t)
	if _, err := db.CreateTable("VER", []types.FieldInfo{
		{FieldType: types.FieldType_Uint8, FieldName: []byte("A")},
	}); err != nil {
		t.Fatalf("create: %v", err)
	}
	tbl, ok := db.FindTable("VER")
	if !ok {
		t.Fatal("找不到 VER")
	}

	realVer := tbl.activeVersion
	tbl.activeVersion = 0xFF // 推到"下一版就会回绕"的状态
	if _, err := tbl.nextSchemaVersion(); err == nil {
		t.Error("activeVersion=255 时 nextSchemaVersion 应当报错")
	}

	addErr := tbl.AddField(types.FieldInfo{
		FieldType: types.FieldType_Uint8, FieldName: []byte("B"),
	})
	dropErr := tbl.DropField("A")
	if addErr == nil {
		t.Error("版本耗尽后 AddField 应当被拒 (否则版本号回绕成 0)")
	}
	if dropErr == nil {
		t.Error("版本耗尽后 DropField 应当被拒 (删列同样新增一版)")
	}

	// 被拒的调用不能改动任何 in-memory 状态, 否则表会停在"改了一半"
	if len(tbl.Field) != 1 {
		t.Errorf("被拒的 AddField 改动了 master registry: %d 列", len(tbl.Field))
	}
	if tbl.Field[0].IsDropped() {
		t.Error("被拒的 DropField 把列标记成了已删除")
	}

	// 恢复真实版本后, 表必须仍然可用 (闸门不能把表弄坏)
	tbl.activeVersion = realVer
	if _, err := tbl.InsertRow([]types.FieldValue{
		{FieldName: []byte("A"), Value: uint8(1)},
	}); err != nil {
		t.Errorf("版本闸门拒绝后表应当仍可写: %v", err)
	}
}

// TestFieldNameLengthGate 列名在 schema 里是 [nameLen(1B)][name...]:
// 空名会让解析侧把下一个字段的类型字节当名字, 超过 255 字节会让 nameLen 截断并让其后
// 所有字段整体错位。两者都必须在建列时报错。
func TestFieldNameLengthGate(t *testing.T) {
	db := fixOpenDB(t)

	_, err := db.CreateTable("EMPTYNAME", []types.FieldInfo{
		{FieldType: types.FieldType_Uint8, FieldName: nil},
	})
	if err == nil {
		t.Error("长度为 0 的列名必须被拒")
	}

	tooLong := make([]byte, FieldNameSize+1)
	for i := range tooLong {
		tooLong[i] = 'a'
	}
	_, err = db.CreateTable("LONGNAME", []types.FieldInfo{
		{FieldType: types.FieldType_Uint8, FieldName: tooLong},
	})
	if err == nil {
		t.Fatalf("列名超过 %d 字节必须被拒", FieldNameSize)
	}
	// 报错要能直接改代码: 既要有上限, 也要有用户写下的那个长度
	for _, want := range []string{strconv.Itoa(FieldNameSize), strconv.Itoa(FieldNameSize + 1)} {
		if !strings.Contains(err.Error(), want) {
			t.Errorf("列名超长的报错缺少 %q: %v", want, err)
		}
	}
}

// TestIndexFieldCountGate 索引分量个数上限。
//
// 旧实现用 `byte(len(fieldIDs))` 判断, 257 个分量会回绕成 1 而"通过", 于是越界的调用
// 反而建成一个只覆盖 1 列却声称覆盖全部的索引。所以先按 int 判, 再转 byte。
func TestIndexFieldCountGate(t *testing.T) {
	db := fixOpenDB(t)
	if _, err := db.CreateTable("IDXCNT", []types.FieldInfo{
		{FieldType: types.FieldType_Uint64, FieldName: []byte("A")},
	}); err != nil {
		t.Fatalf("create: %v", err)
	}
	tbl, ok := db.FindTable("IDXCNT")
	if !ok {
		t.Fatal("找不到 IDXCNT")
	}
	if _, err := tbl.CreateIndex("pk", page.IndexTypePrimary, []byte{0}); err != nil {
		t.Fatalf("create pk: %v", err)
	}

	if _, err := tbl.CreateIndex("none", page.IndexTypeBTree, nil); err == nil {
		t.Error("0 个分量必须被拒")
	}
	if _, err := tbl.CreateIndex("many", page.IndexTypeBTree, make([]byte, MaxIndexFields+1)); err == nil {
		t.Errorf("%d 个分量必须被拒", MaxIndexFields+1)
	}
	// 257 个分量: 走 byte 会回绕成 1, 必须在这条闸门被挡住
	if _, err := tbl.CreateIndex("overflow", page.IndexTypeBTree, make([]byte, 257)); err == nil {
		t.Error("257 个分量 (byte 回绕) 必须被拒")
	}
}

// TestDynamicValueSizeGate Dynamicbyte 的单值上限, 以及"这条上限只管 Dynamicbyte"。
//
// Dynamicbyte 的内容整块落进一个堆页 (行内 6B 堆引用), 单值上限就是 MaxBlobSize。超限值
// 必须在 serialize 这一层就报出列名与上限, 否则用户只会拿到底层那句不带列名、不带字节数的
// ErrHeapPageTooLarge, 看不出是哪一列、超了多少。
//
// blob 列 (FieldType_Blob) 不在闸门内: v10 起它的行内是 8B BlobRef, 内容落在可跨页的链上
// (上限约 4GB, 见 BlobChainMax), 同一个超过 MaxBlobSize 的值写进 blob 列是合法的 ——
// 链上的内容能不能真的读回来由 TestBlobColumnChainRoundTrip 钉。
func TestDynamicValueSizeGate(t *testing.T) {
	db := fixOpenDB(t)
	if _, err := db.CreateTable("DYN", []types.FieldInfo{
		{FieldType: types.FieldType_Uint64, FieldName: []byte("I")},
		{FieldType: types.FieldType_Dynamicbyte, FieldName: []byte("D")},
	}); err != nil {
		t.Fatalf("create: %v", err)
	}
	tbl, ok := db.FindTable("DYN")
	if !ok {
		t.Fatal("找不到 DYN")
	}
	// 主键要落在定长列 I 上: 动态列不能作为索引键
	if _, err := tbl.CreateIndex("pk", page.IndexTypePrimary, []byte{0}); err != nil {
		t.Fatalf("create pk: %v", err)
	}

	huge := make([]byte, MaxBlobSize+1)
	_, err := tbl.InsertRow([]types.FieldValue{
		{FieldName: []byte("I"), Value: uint64(1)},
		{FieldName: []byte("D"), Value: huge},
	})
	if err == nil {
		t.Fatalf("D: 超过单块上限 %d 的写入必须被拒", MaxBlobSize)
	}
	if !errors.Is(err, types.ErrValueRange) {
		t.Errorf("D: 超限应报 ErrValueRange, 得到 %v", err)
	}
	if !strings.Contains(err.Error(), "D") {
		t.Errorf("D: 超限报错应包含列名: %v", err)
	}
	if !strings.Contains(err.Error(), strconv.Itoa(int(MaxBlobSize))) {
		t.Errorf("D: 超限报错应包含上限 %d: %v", MaxBlobSize, err)
	}
	// 报错要指出出路: 同一个值换个列类型就装得下。
	if !strings.Contains(err.Error(), "blob") {
		t.Errorf("D: 超限报错应提示可改用 blob 列: %v", err)
	}
}

// TestNotNullGate 非空列写 NULL 必须报错, 且定长列与动态列口径一致。
//
// 原先只有动态列会"静默成功" (行内引用全 0 且 NULL 位图不置位, 读回来是空内容),
// 而定长列写 nil 会在 EncodeValue 里报错 —— 同一句"写 nil"在两种列上行为相反。
func TestNotNullGate(t *testing.T) {
	db := fixOpenDB(t)
	if _, err := db.CreateTable("NN", []types.FieldInfo{
		{FieldType: types.FieldType_Uint64, FieldName: []byte("I")},
		{FieldType: types.FieldType_Dynamicbyte, FieldName: []byte("D")},
		{FieldType: types.FieldType_Dynamicbyte, FieldName: []byte("N"), Flag: types.FlagNullable},
	}); err != nil {
		t.Fatalf("create: %v", err)
	}
	tbl, ok := db.FindTable("NN")
	if !ok {
		t.Fatal("找不到 NN")
	}
	if _, err := tbl.CreateIndex("pk", page.IndexTypePrimary, []byte{0}); err != nil {
		t.Fatalf("create pk: %v", err)
	}

	// 非空动态列写 NULL → 必须报错并点名列
	_, err := tbl.InsertRow([]types.FieldValue{
		{FieldName: []byte("I"), Value: uint64(1)},
		{FieldName: []byte("D"), Value: nil},
	})
	if err == nil {
		t.Fatal("非空动态列写 NULL 必须被拒")
	}
	if !strings.Contains(err.Error(), "D") {
		t.Errorf("非空列写 NULL 的报错应点名列: %v", err)
	}

	// 可空列写 NULL → 放行
	if _, err := tbl.InsertRow([]types.FieldValue{
		{FieldName: []byte("I"), Value: uint64(2)},
		{FieldName: []byte("N"), Value: nil},
	}); err != nil {
		t.Errorf("可空列写 NULL 应当放行: %v", err)
	}

	// 省略非空列 (本次不传) → 放行, 按零值/空内容写
	if _, err := tbl.InsertRow([]types.FieldValue{
		{FieldName: []byte("I"), Value: uint64(3)},
	}); err != nil {
		t.Errorf("省略非空列应当放行: %v", err)
	}
}