package core

// 本文件验证【值级溢出】: 行内容超 MaxInlineRowData 时, 未建索引的宽字符串列落到堆页,
// 行内只留 6B 堆引用 —— "一页两行"的政策由此从"建表时按声明宽度卡死"变成"写入时按实际
// 行宽动态保住"。要钉住的性质:
//
//  1. 写入: 超限的行不再整行拒绝, 而是真的被缩短到 MaxInlineRowData 以内 (行内出现引用)。
//  2. 读回: 溢出的列解引用后逐字节一致; 同行未溢出的列不受影响。
//  3. 逐行判定: 同一列的【不同行】可以一行内联、一行溢出, 过滤必须按行内 tag 走对分支
//     —— 这是最容易错的一处: 只看列类型就会把 6B 页号当成字符串内容去比, 症状是静默漏行。
//  4. 更新: 溢出行改写后仍可读回 (旧堆块复用/释放)。
//  5. 删除: 行与堆块一并消失。

import (
	"fmt"
	"strings"
	"testing"

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

func spillColName(i int) string { return fmt.Sprintf("S%d", i) }

// spillNewTable 建一张"宽字符串列都可溢出"的表: 1 个 8B 主键列 + n 个 2048 宽字符串列。
// 列 0 是定长主键 (恒内联, 偏移表基址依赖它), 其余字符串列都没建索引 ⇒ 都可溢出。
func spillNewTable(t *testing.T, db *DB, name string, n int) *Table {
	t.Helper()
	fields := []types.FieldInfo{{FieldType: types.FieldType_Uint64, FieldName: []byte("id")}}
	for i := 0; i < n; i++ {
		fields = append(fields, types.FieldInfo{
			FieldType: types.FieldType_String,
			Flag:      types.WidthCodeFor(types.MaxInlineColumnWidth),
			FieldName: []byte(spillColName(i)),
		})
	}
	tbl, err := db.CreateTable(name, fields)
	if err != nil {
		t.Fatalf("建表 %s: %v", name, err)
	}
	if _, err := tbl.CreateIndex("pk", page.IndexTypePrimary, []byte{0}); err != nil {
		t.Fatalf("建主键索引: %v", err)
	}
	return tbl
}

// spillRawRow 直接按 pkID 取回【盘上原始 rowData】(去掉 8B pkID 前缀)。精确到字节, 用来断言
// "行真的被缩短了", 而不是只看插入成不成功 —— 溢出若没生效, 行会变大但仍然插得进去。
func spillRawRow(t *testing.T, tbl *Table, pk uint64) []byte {
	t.Helper()
	meta, handler, ok := tbl.IndexMeta().rowStoreMeta()
	if !ok {
		t.Fatal("表上没有行存储索引")
	}
	ph, ok := handler.(*tsindex.PrimaryHandler)
	if !ok {
		t.Fatalf("行存储 handler 不是 PrimaryHandler: %T", handler)
	}
	store, found := ph.SearchByPK(tbl.handle, meta.EntryPage, pk)
	if !found {
		t.Fatalf("pk=%d 读不到行", pk)
	}
	return store[8:]
}

func spillValues(pk uint64, vals []string) []types.FieldValue {
	fs := make([]types.FieldValue, 0, 1+len(vals))
	fs = append(fs, types.FieldValue{FieldName: []byte("id"), Value: pk})
	for i, v := range vals {
		fs = append(fs, types.FieldValue{FieldName: []byte(spillColName(i)), Value: v})
	}
	return fs
}

func TestValueSpillRoundTrip(t *testing.T) {
	db := fixOpenDB(t)
	const n = 5
	tbl := spillNewTable(t, db, "SPILL", n)

	// 每列 1800 字节 × 5 = 9000 > MaxInlineRowData(8159): 不溢出就装不下。
	big := make([]string, n)
	for i := range big {
		big[i] = strings.Repeat(string(rune('a'+i)), 1800)
	}
	pkBig := tbl.AllocPKID()
	if _, err := tbl.InsertRowWithID(pkBig, spillValues(pkBig, big)); err != nil {
		t.Fatalf("插入超限行: %v", err)
	}
	// 同一批列的第二行故意【不溢出】: 覆盖"同列不同行、内联与溢出并存"的路径。
	small := make([]string, n)
	for i := range small {
		small[i] = strings.Repeat(string(rune('A'+i)), 40)
	}
	pkSmall := tbl.AllocPKID()
	if _, err := tbl.InsertRowWithID(pkSmall, spillValues(pkSmall, small)); err != nil {
		t.Fatalf("插入小行: %v", err)
	}

	// 1. 大行真的被缩短到行内上限以内 (未溢出时长度 ≈ 9022)。
	raw := spillRawRow(t, tbl, pkBig)
	if len(raw) > types.MaxInlineRowData {
		t.Fatalf("行内长度 %d 超过上限 %d: 值级溢出没有生效",
			len(raw), types.MaxInlineRowData)
	}
	// 小行不该被无谓地溢出 (它本来就在限内)。
	if rawSmall := spillRawRow(t, tbl, pkSmall); len(rawSmall) > types.MaxInlineRowData {
		t.Fatalf("小行长度 %d 超过上限: 不该溢出却溢出了", len(rawSmall))
	}

	// 2. 两行都读回, 每列逐字节一致 (溢出列解引用 + 内联列并存的错位探针)。
	rows, err := tbl.Query().Find()
	if err != nil {
		t.Fatalf("find: %v", err)
	}
	if len(rows) != 2 {
		t.Fatalf("应读出 2 行, 得到 %d", len(rows))
	}
	byID := map[uint64][]types.FieldValue{}
	for _, r := range rows {
		id, ok := newTypeRowVal(t, r, "id").(uint64)
		if !ok {
			t.Fatalf("id 列不是 uint64: %T", newTypeRowVal(t, r, "id"))
		}
		byID[id] = r
	}
	for _, tc := range []struct {
		pk   uint64
		want []string
	}{{pkBig, big}, {pkSmall, small}} {
		r, ok := byID[tc.pk]
		if !ok {
			t.Fatalf("pk=%d 没读回来", tc.pk)
		}
		for i, want := range tc.want {
			if got := stringTrim(newTypeRowVal(t, r, spillColName(i))); got != want {
				t.Errorf("pk=%d 列 %s 读回长度 %d, 期望 %d (溢出列解引用不一致)",
					tc.pk, spillColName(i), len(got), len(want))
			}
		}
	}

	// 3. 过滤: 对【未建索引】的列做等值过滤, 走的是行级过滤 (rowFieldForCompare) ——
	//    大行该列已溢出 (行内是 6B 引用), 小行该列内联, 两条路径都要给出正确答案。
	const probeCol = 2
	if got, err := tbl.Query().Where(Eq(spillColName(probeCol), big[probeCol], types.FieldType(0))).Count(); err != nil {
		t.Fatalf("按溢出列过滤: %v", err)
	} else if got != 1 {
		t.Errorf("溢出列等值过滤命中 %d 行, 期望 1 (把 6B 堆引用当成内容比了?)", got)
	}
	if got, err := tbl.Query().Where(Eq(spillColName(probeCol), small[probeCol], types.FieldType(0))).Count(); err != nil {
		t.Fatalf("按内联列过滤: %v", err)
	} else if got != 1 {
		t.Errorf("内联列等值过滤命中 %d 行, 期望 1", got)
	}
	// 前缀不该命中 (与内联时的语义一致: 按列宽补零比较)。
	if got, _ := tbl.Query().Where(Eq(spillColName(probeCol), big[probeCol][:8], types.FieldType(0))).Count(); got != 0 {
		t.Errorf("前缀命中 %d 行, 期望 0", got)
	}
	// LIKE 走的是另一条行级路径 (LikeCondition.Match), 同样要能解引用。
	if got, err := tbl.Query().Where(Like(spillColName(probeCol), big[probeCol][:8]+"%")).Count(); err != nil {
		t.Fatalf("LIKE: %v", err)
	} else if got != 1 {
		t.Errorf("LIKE 命中 %d 行, 期望 1", got)
	}

	// 4. 更新溢出行 (改短) 后仍可读回。
	newVals := make([]string, n)
	for i := range newVals {
		newVals[i] = strings.Repeat(string(rune('0'+i)), 1500)
	}
	if err := tbl.UpdateRowByPK(pkBig, spillValues(pkBig, newVals)); err != nil {
		t.Fatalf("更新溢出行: %v", err)
	}
	rows, err = tbl.Query().Find()
	if err != nil {
		t.Fatalf("find after update: %v", err)
	}
	for _, r := range rows {
		if id, _ := newTypeRowVal(t, r, "id").(uint64); id != pkBig {
			continue
		}
		for i, want := range newVals {
			if got := stringTrim(newTypeRowVal(t, r, spillColName(i))); got != want {
				t.Errorf("更新后列 %s 长度 %d, 期望 %d", spillColName(i), len(got), len(want))
			}
		}
	}

	// 5. 删除: 行与堆块一并消失。
	if err := tbl.DeleteRowByPK(pkBig); err != nil {
		t.Fatalf("删除: %v", err)
	}
	if got, _ := tbl.Query().Count(); got != 1 {
		t.Errorf("删除后应剩 1 行, 得到 %d", got)
	}
}