TsDB2/migrate/widerow_test.go

package migrate

import (
	"git.ycl.pub/yl32940/TsDB2/DB/core/page"
	"git.ycl.pub/yl32940/TsDB2/DB/core/types"
	"bytes"
	"fmt"
	"path/filepath"
	"strings"
	"testing"
)

// wideColumn 造一个"源库声明了 n 字节容量"的字符串列。
//
// 走 bucketColumn (读端把 SQL 长度收编成列的那条路径) 而不是手写类型 + 宽度码:
// 测试要验的正是"读端读出来的列, 到行宽闸门这里怎么处理", 手写一份等于把被测的映射
// 又抄一遍 —— 映射改了测试不会跟着变。
func wideColumn(name string, n int) Column {
	ft, code := bucketColumn(n)
	return Column{Name: name, Type: ft, Param: code}
}

// wideSchema 造一张行宽超限的表, 与线上撞过闸门的那张同构: 几列 2048 / 1024 顶到上限。
//
// 【为什么必须带索引】v9 起未建索引的宽字符串列会在写入时值级溢出 (行内只留 6B 引用),
// 光靠"声明得宽"已经撞不到闸门了; 真正撑满行内下界的是【建了索引的列】—— 索引键按行内
// 字节拼, 一溢出键就成了页号。所以样本必须在 2×2048 + 4×1024 上挂索引, 才是线上那张表。
//
// 末两列是边界样本: resAlias(224) 与 resName(256) 都在 256 这条线以内, 必须留在定长档 ——
// 一起降掉就等于把"可索引、可定长比较"的短列白白拖进堆。
func wideSchema() *TableSchema {
	cols := []Column{{Name: "_id", Type: types.FieldType_Uint64}}
	for _, c := range []struct {
		name string
		size int
	}{
		{"resOnlinePatch", 2048},
		{"resChatCrowd", 2048},
		{"resExtendParam1", 1024},
		{"resExtendParam2", 1024},
		{"resExtendParam3", 1024},
		{"resExtendParam4", 1024},
		{"resExtendParam5", 1024},
		{"resExtendParam6", 1024},
		{"resExtendParam7", 1024},
		{"resAppID", 320}, // 320 > 256: 降
		{"resAlias", 224}, // <= 256: 不降
		{"resName", 256},  // 边界: 也不降
	} {
		cols = append(cols, wideColumn(c.name, c.size))
	}
	sc := &TableSchema{Name: "mesh_resource_info", Columns: cols, PKColumn: "_id"}
	// 这些列建了索引 ⇒ 不能值级溢出 ⇒ 行内下界 = 1 + id8 + 2×2048 + 4×1024 + 开销 > 8159
	for _, name := range []string{
		"resOnlinePatch", "resChatCrowd",
		"resExtendParam1", "resExtendParam2", "resExtendParam3", "resExtendParam4",
	} {
		sc.Indexes = append(sc.Indexes, IndexSchema{
			Name: "idx_" + name, Type: page.IndexTypeBTree, Columns: []string{name},
		})
	}
	return sc
}

func TestFitRowWidthDemotesWideColumns(t *testing.T) {
	sc := wideSchema()
	if floor := rowSpillFloor(sc); floor <= types.MaxInlineRowData {
		t.Fatalf("样本表本来就该超限 (下界 %d <= %d), 否则这个用例什么都没验",
			floor, types.MaxInlineRowData)
	}

	n, err := fitRowWidthForVRTX(sc)
	if err != nil {
		t.Fatalf("可降级的表不该报错: %v", err)
	}
	if want := 10; n != want {
		t.Errorf("降级列数 = %d, 期望 %d (宽度 > 256 的那些)", n, want)
	}
	if floor := rowSpillFloor(sc); floor > types.MaxInlineRowData {
		t.Fatalf("降级后行内下界 %d 应当装得下 (上限 %d)", floor, types.MaxInlineRowData)
	}

	byName := make(map[string]Column, len(sc.Columns))
	for _, c := range sc.Columns {
		byName[c.Name] = c
	}
	// 宽列: 变成动态列, 且宽度码必须清零 —— 留着原宽度码 FieldSize 仍按定长算, 行宽又变回去。
	for _, name := range []string{"resOnlinePatch", "resChatCrowd", "resExtendParam1", "resAppID"} {
		c, ok := byName[name]
		if !ok {
			t.Fatalf("列 %s 丢了", name)
		}
		if !c.IsDynamic() {
			t.Errorf("列 %s 应当降为动态列, 实际 %v", name, c.Type)
		}
		if c.Param != 0 {
			t.Errorf("列 %s 的宽度码应当清零, 实际 %d", name, c.Param)
		}
		if c.Size() != types.DynamicRefSize {
			t.Errorf("列 %s 的行内宽度 = %d, 期望堆引用 %d", name, c.Size(), types.DynamicRefSize)
		}
	}
	// 窄列与行主键: 一个都不许动。
	for name, want := range map[string]int{"resAlias": 224, "resName": 256, "_id": 8} {
		c := byName[name]
		if c.IsDynamic() {
			t.Errorf("列 %s (行内 %d 字节) 不该降为动态列", name, c.Size())
		}
		if c.Size() != want {
			t.Errorf("列 %s 的行内宽度 = %d, 期望 %d", name, c.Size(), want)
		}
	}
	// 降级是改类型, 必须报出来 (runTable 靠它汇进迁移报告)。
	if len(sc.DegradedColumns) != n {
		t.Fatalf("DegradedColumns = %d 条, 期望 %d 条", len(sc.DegradedColumns), n)
	}
	for _, note := range sc.DegradedColumns {
		if !strings.Contains(note, "动态列") || !strings.Contains(note, "索引键") {
			t.Errorf("降级说明没讲清后果: %q", note)
		}
	}
	if !strings.Contains(strings.Join(sc.DegradedColumns, "|"), "resChatCrowd") {
		t.Errorf("降级说明里应当点名具体列: %v", sc.DegradedColumns)
	}
}

// 没有可降的列 (全是 <= 256 的列) 时, 只能按引擎的口径拒绝 —— 降无可降, 硬装就是骗人。
//
// 这里全靠索引把下界顶上去: 未建索引的 256 列是可溢出的 (每个只占 6B), 40 列也才几百字节,
// 引擎靠溢出就装下了, 根本不需要降级; 只有它们【都建了索引】不再可溢出, 下界才会超限。
func TestFitRowWidthRejectsWhenNothingToDemote(t *testing.T) {
	sc := &TableSchema{Name: "flat_wide", Columns: []Column{{Name: "_id", Type: types.FieldType_Uint64}}}
	for i := 0; i < 40; i++ {
		name := fmt.Sprintf("c%02d", i)
		sc.Columns = append(sc.Columns, wideColumn(name, 256))
		sc.Indexes = append(sc.Indexes, IndexSchema{
			Name: "idx_" + name, Type: page.IndexTypeBTree, Columns: []string{name},
		})
	}
	sc.PKColumn = "_id"

	n, err := fitRowWidthForVRTX(sc)
	if err == nil {
		t.Fatal("降无可降时必须报错, 否则会建出一张每行都写不进去的表")
	}
	if n != 0 {
		t.Errorf("降级列数 = %d, 期望 0 (这些列都在 256 以内)", n)
	}
	if !strings.Contains(err.Error(), "行宽") {
		t.Errorf("报错应当保留引擎的口径 (行宽/上限/各列宽度): %v", err)
	}
	if len(sc.DegradedColumns) != 0 {
		t.Errorf("没降就不该有降级说明: %v", sc.DegradedColumns)
	}
}

// 幂等: 同一个 schema 跑第二遍不能再降一次 (否则 DegradedColumns 会重复记账)。
func TestFitRowWidthIsIdempotent(t *testing.T) {
	sc := wideSchema()
	if _, err := fitRowWidthForVRTX(sc); err != nil {
		t.Fatalf("第一次降级: %v", err)
	}
	notes := len(sc.DegradedColumns)

	n, err := fitRowWidthForVRTX(sc)
	if err != nil {
		t.Fatalf("第二次不该报错: %v", err)
	}
	if n != 0 {
		t.Errorf("第二次降级列数 = %d, 期望 0 (已经装得下了)", n)
	}
	if len(sc.DegradedColumns) != notes {
		t.Errorf("降级说明重复记账: %d != %d", len(sc.DegradedColumns), notes)
	}
}

// 端到端: 降级后的 schema 必须真的能在 VRTX 里立起来并接受大值 ——
// 降级前这张表连 CreateTable 都过不去, 这正是要解决的问题。
func TestDemotedSchemaBuildsAndAcceptsWideValues(t *testing.T) {
	sc := wideSchema()
	if _, err := fitRowWidthForVRTX(sc); err != nil {
		t.Fatalf("行宽闸门: %v", err)
	}
	// 被降级的列上原有的索引: 动态列不能做索引键, 该索引应当被跳过并报进 SkippedIndexes,
	// 而不是让整张表失败 —— 索引丢了能补, 数据没迁才是真丢。
	// (样本里的 6 条索引都挂在被降级的宽列上, 建表时应当全部跳过。)

	v, err := OpenVRTX(filepath.Join(t.TempDir(), "wide.db"), quiet())
	if err != nil {
		t.Fatalf("开库: %v", err)
	}
	defer v.Close()
	if err := v.CreateTable(sc, true); err != nil {
		t.Fatalf("降级后建表: %v", err)
	}
	if len(sc.SkippedIndexes) == 0 {
		t.Error("降级列上的索引应当被跳过并记进 SkippedIndexes")
	}

	tbl, ok := v.DB().FindTable("mesh_resource_info")
	if !ok {
		t.Fatal("建完表却找不到")
	}
	// 2000 字节 (超过 2048 档装不下、但动态列装得下) 的值必须原样进得去。
	big := bytes.Repeat([]byte{'x'}, 2000)
	vals := make([]types.FieldValue, 0, len(sc.Columns))
	for _, c := range sc.Columns {
		var val any
		switch c.Name {
		case "_id":
			val = uint64(1)
		case "resChatCrowd":
			val = big
		case "resAlias", "resName":
			val = []byte(c.Name)
		default:
			val = []byte(nil)
		}
		vals = append(vals, types.FieldValue{FieldName: []byte(c.Name), Value: val})
	}
	if _, err := tbl.InsertRowWithID(1, vals); err != nil {
		t.Fatalf("往降级后的列写 2000 字节: %v", err)
	}

	got, err := tbl.FindByPK(1)
	if err != nil {
		t.Fatalf("回读: %v", err)
	}
	for _, fv := range got {
		if string(fv.FieldName) != "resChatCrowd" {
			continue
		}
		raw, ok := fv.Value.([]byte)
		if !ok {
			t.Fatalf("动态列回读值类型 = %T, 期望 []byte", fv.Value)
		}
		if !bytes.Equal(raw, big) {
			t.Errorf("动态列回读 %d 字节, 与写入的 %d 字节不一致", len(raw), len(big))
		}
		return
	}
	t.Error("回读结果里没有 resChatCrowd")
}