TsDB2/DB/core/compose_varwidth_test.go

package core

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

// TestVarWidthComposeCorrectness 非 8 字节分量的复合索引 (SBT B+ 树) 端到端正确性:
func TestVarWidthComposeCorrectness(t *testing.T) {
	db := fixOpenDB(t)
	tbl, err := db.CreateTable("t", []types.FieldInfo{
		{FieldType: types.FieldType_Uint64, FieldName: []byte("id")},
		{FieldType: types.FieldType_Decimal, FieldName: []byte("amt"), Flag: 2}, // 18B
		{FieldType: types.FieldType_Year, FieldName: []byte("yr")},              // 2B
	})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := tbl.CreateIndex("pk", page.IndexTypePrimary, []byte{0}); err != nil {
		t.Fatal(err)
	}
	// 复合索引 (amt, yr): keyLen = 18+2 = 20, 非 8 对齐
	if _, err := tbl.CreateIndex("idx_ay", page.IndexTypeCompose, []byte{1, 2}); err != nil {
		t.Fatal(err)
	}
	// 造数据: amt 在尾部字节不同 (1.01 vs 1.02 在 18B 编码的尾部有差异)
	type row struct {
		id  uint64
		amt string
		yr  uint16
	}
	rows := []row{}
	for i := 1; i <= 300; i++ {
		r := row{id: uint64(i), amt: "1.01", yr: uint16(2000 + i%50)}
		if i%2 == 0 {
			r.amt = "2.50"
		}
		if i%3 == 0 {
			r.amt = "12345.67"
		}
		rows = append(rows, r)
	}
	for _, r := range rows {
		amt, err := types.ParseDecimal(r.amt)
		if err != nil {
			t.Fatal(err)
		}
		if _, err := tbl.InsertRowWithID(r.id, []types.FieldValue{
			{FieldName: []byte("id"), Value: r.id},
			{FieldName: []byte("amt"), Value: amt},
			{FieldName: []byte("yr"), Value: r.yr},
		}); err != nil {
			t.Fatal(err)
		}
	}
	if err := db.Commit(); err != nil {
		t.Fatal(err)
	}

	// 等值查 amt=2.50: 与全表过滤比对
	amt250, _ := types.ParseDecimal("2.50")
	want := 0
	for _, r := range rows {
		if r.amt == "2.50" {
			want++
		}
	}
	n, err := tbl.Query().Where(Eq("amt", amt250, 0)).Count()
	if err != nil || n != int64(want) {
		t.Fatalf("等值 amt=2.50: 索引 count=%d err=%v, 全表过滤 %d", n, err, want)
	}
	// 复合等值 amt=2.50 AND yr=2002
	want2 := 0
	for _, r := range rows {
		if r.amt == "2.50" && r.yr == 2002 {
			want2++
		}
	}
	yrVal := uint16(2002)
	n, err = tbl.Query().Where(And(Eq("amt", amt250, 0), Eq("yr", yrVal, 0))).Count()
	if err != nil || n != int64(want2) {
		t.Fatalf("复合等值 amt+yr: 索引 count=%d err=%v, 全表过滤 %d", n, err, want2)
	}
	// 范围查 amt ∈ [2.00, 3.00]
	lo, _ := types.ParseDecimal("2.00")
	hi, _ := types.ParseDecimal("3.00")
	want3 := 0
	for _, r := range rows {
		if r.amt == "2.50" {
			want3++
		}
	}
	n, err = tbl.Query().Where(And(Ge("amt", lo, 0), Le("amt", hi, 0))).Count()
	if err != nil || n != int64(want3) {
		t.Fatalf("范围 amt∈[2,3]: 索引 count=%d err=%v, 全表过滤 %d", n, err, want3)
	}
	t.Logf("非 8 字节分量复合索引端到端正确: 等值/复合等值/范围全部与全表过滤一致")
}