TsDB2/DB/core/blob_test.go

package core

import (
	"bytes"
	"crypto/rand"
	"errors"
	"io"
	"strings"
	"testing"

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

// TestBlobEmptyRoundTrip 验证"空 blob" = (FirstPageID=0, TotalLen=0) 全程不分配页.
func TestBlobEmptyRoundTrip(t *testing.T) {
	db := fixOpenDB(t)
	_, err := db.CreateTable("empty_blob", []types.FieldInfo{
		{FieldType: types.FieldType_Uint64, FieldName: []byte("id")},
		{FieldType: types.FieldType_Blob, FieldName: []byte("data")},
	})
	if err != nil {
		t.Fatalf("CreateTable: %v", err)
	}

	tx := db.BeginTransaction()
	ref, err := tx.BlobCommit(nil, types.CompressNone)
	if err != nil {
		t.Fatalf("BlobCommit(nil): %v", err)
	}
	if !ref.Empty() {
		t.Errorf("nil io.Reader -> ref %+v, want Empty", ref)
	}

	ref2, err := tx.BlobCommit(bytes.NewReader(nil), types.CompressNone)
	if err != nil {
		t.Fatalf("BlobCommit(empty stream): %v", err)
	}
	if !ref2.Empty() {
		t.Errorf("empty stream -> ref %+v, want Empty", ref2)
	}

	if err := tx.Commit(); err != nil {
		t.Fatalf("Commit: %v", err)
	}
}

// TestBlobSinglePageRoundTrip 单页 (16364 字节以内) 的写读一致.
func TestBlobSinglePageRoundTrip(t *testing.T) {
	db := fixOpenDB(t)
	tbl, _ := db.CreateTable("t1", []types.FieldInfo{
		{FieldType: types.FieldType_Uint64, FieldName: []byte("id")},
		{FieldType: types.FieldType_Blob, FieldName: []byte("data")},
	})

	payload := make([]byte, 8000)
	for i := range payload {
		payload[i] = byte(i % 251)
	}

	tx := db.BeginTransaction()
	ref, err := tx.BlobCommit(bytes.NewReader(payload), types.CompressNone)
	if err != nil {
		t.Fatalf("BlobCommit: %v", err)
	}
	if ref.TotalLen != uint32(len(payload)) {
		t.Errorf("TotalLen = %d, want %d", ref.TotalLen, len(payload))
	}
	if err := tx.Commit(); err != nil {
		t.Fatalf("Commit: %v", err)
	}

	got, err := tbl.BlobRead(ref, types.CompressNone)
	if err != nil {
		t.Fatalf("BlobRead: %v", err)
	}
	if !bytes.Equal(got, payload) {
		t.Errorf("BlobRead bytes mismatch (len got=%d want=%d)", len(got), len(payload))
	}
}

// TestBlobMultiPageRoundTrip 跨页 (4 页, ~50KB).
func TestBlobMultiPageRoundTrip(t *testing.T) {
	db := fixOpenDB(t)
	tbl, _ := db.CreateTable("t2", []types.FieldInfo{
		{FieldType: types.FieldType_Uint64, FieldName: []byte("id")},
		{FieldType: types.FieldType_Blob, FieldName: []byte("data")},
	})

	payload := make([]byte, 50_000)
	if _, err := rand.Read(payload); err != nil {
		t.Fatalf("rand.Read: %v", err)
	}

	tx := db.BeginTransaction()
	ref, err := tx.BlobCommit(bytes.NewReader(payload), types.CompressNone)
	if err != nil {
		t.Fatalf("BlobCommit: %v", err)
	}
	if err := tx.Commit(); err != nil {
		t.Fatalf("Commit: %v", err)
	}

	got, err := tbl.BlobRead(ref, types.CompressNone)
	if err != nil {
		t.Fatalf("BlobRead: %v", err)
	}
	if !bytes.Equal(got, payload) {
		t.Errorf("mismatch: got %d bytes, want %d", len(got), len(payload))
	}
}

// TestBlobSnappyRoundTrip Snappy 压缩全程 (压缩-解压字节完全一致).
func TestBlobSnappyRoundTrip(t *testing.T) {
	db := fixOpenDB(t)
	tbl, _ := db.CreateTable("t3", []types.FieldInfo{
		{FieldType: types.FieldType_Uint64, FieldName: []byte("id")},
		{FieldType: types.FieldType_Blob, FieldName: []byte("data")},
	})

	// 高度可压缩的 1MB (少量字节重复)
	payload := make([]byte, 1024*1024)
	for i := range payload {
		payload[i] = byte(i % 4)
	}

	tx := db.BeginTransaction()
	ref, err := tx.BlobCommit(bytes.NewReader(payload), types.CompressSnappy)
	if err != nil {
		t.Fatalf("BlobCommit snappy: %v", err)
	}
	if ref.TotalLen != uint32(len(payload)) {
		t.Errorf("TotalLen = %d, want %d", ref.TotalLen, len(payload))
	}
	if err := tx.Commit(); err != nil {
		t.Fatalf("Commit: %v", err)
	}

	got, err := tbl.BlobRead(ref, types.CompressSnappy)
	if err != nil {
		t.Fatalf("BlobRead snappy: %v", err)
	}
	if !bytes.Equal(got, payload) {
		t.Errorf("snappy round-trip mismatch (got %d, want %d)", len(got), len(payload))
	}

	// 错误压缩方式 -> 解出的字节数与 TotalLen 不一致, 应报 ErrBlobCorrupted
	_, err = tbl.BlobRead(ref, types.CompressNone)
	if !errors.Is(err, ErrBlobCorrupted) {
		t.Errorf("wrong compress -> got err=%v, want ErrBlobCorrupted", err)
	}
}

// TestBlobInterruptedThenRollback 流中断 + Rollback -> 链页全部回收 (无副作用).
func TestBlobInterruptedThenRollback(t *testing.T) {
	db := fixOpenDB(t)
	_, _ = db.CreateTable("t4", []types.FieldInfo{
		{FieldType: types.FieldType_Uint64, FieldName: []byte("id")},
		{FieldType: types.FieldType_Blob, FieldName: []byte("data")},
	})

	// 模拟 100KB 网络流, 中途"网线断"返回 io.ErrUnexpectedEOF
	r := &brokenReader{remaining: 100_000, failAt: 50_000}

	tx := db.BeginTransaction()
	_, err := tx.BlobCommit(r, types.CompressNone)
	if err == nil {
		t.Fatal("expected error from broken stream, got nil")
	}
	if err := tx.Rollback(); err != nil {
		t.Fatalf("Rollback: %v", err)
	}

	// Rollback 后库应回到事务前状态: 再 Commit 应能成功
	if err := db.Commit(); err != nil {
		t.Fatalf("Commit after rollback: %v", err)
	}
}

// TestBlobCommitAndFree 写读一致 + Free 后 BlobFree(Empty) 应是 no-op.
func TestBlobCommitAndFree(t *testing.T) {
	db := fixOpenDB(t)
	tbl, _ := db.CreateTable("t5", []types.FieldInfo{
		{FieldType: types.FieldType_Uint64, FieldName: []byte("id")},
		{FieldType: types.FieldType_Blob, FieldName: []byte("data")},
	})

	payload := []byte("hello blob world")
	tx := db.BeginTransaction()
	ref, err := tx.BlobCommit(bytes.NewReader(payload), types.CompressNone)
	if err != nil {
		t.Fatalf("BlobCommit: %v", err)
	}
	if err := tx.Commit(); err != nil {
		t.Fatalf("Commit: %v", err)
	}

	got, err := tbl.BlobRead(ref, types.CompressNone)
	if err != nil || !bytes.Equal(got, payload) {
		t.Fatalf("BlobRead: got=%v err=%v", got, err)
	}

	// Free 整链
	if err := tbl.BlobFree(ref); err != nil {
		t.Fatalf("BlobFree: %v", err)
	}
	if err := db.Commit(); err != nil {
		t.Fatalf("Commit after free: %v", err)
	}

	// BlobFree(Empty) 应是 no-op
	if err := tbl.BlobFree(BlobRef{}); err != nil {
		t.Errorf("BlobFree(empty): %v", err)
	}
}

// TestBlobSnappyIncompressibleMultiBlock 回归: 不可压缩数据 + Snappy 时,
// snappy.Reader 内部 io.ReadFull 会以 >64KB 的 buffer 调用 chainReader.Read;
// 旧实现 uint16(len(p)-total) 在 >65535 时截断为 0, 导致跳页丢数据 -> corrupt input.
func TestBlobSnappyIncompressibleMultiBlock(t *testing.T) {
	db := fixOpenDB(t)
	tbl, _ := db.CreateTable("t6", []types.FieldInfo{
		{FieldType: types.FieldType_Uint64, FieldName: []byte("id")},
		{FieldType: types.FieldType_Blob, FieldName: []byte("data")},
	})

	// 随机数据 = snappy 几乎不压缩, 每块编码后 >65536 字节, 触发大 buffer Read.
	for _, size := range []int{65_556, 200_000, 1_024_000} {
		payload := make([]byte, size)
		if _, err := rand.Read(payload); err != nil {
			t.Fatalf("rand.Read: %v", err)
		}
		tx := db.BeginTransaction()
		ref, err := tx.BlobCommit(bytes.NewReader(payload), types.CompressSnappy)
		if err != nil {
			t.Fatalf("BlobCommit(%d): %v", size, err)
		}
		if err := tx.Commit(); err != nil {
			t.Fatalf("Commit: %v", err)
		}

		got, err := tbl.BlobRead(ref, types.CompressSnappy)
		if err != nil {
			t.Fatalf("BlobRead(%d): %v", size, err)
		}
		if !bytes.Equal(got, payload) {
			t.Fatalf("size=%d: round-trip mismatch (got %d bytes)", size, len(got))
		}
	}
}

// brokenReader N 字节后返错, 模拟流中断.
type brokenReader struct {
	remaining int
	failAt    int
	consumed  int
}

func (b *brokenReader) Read(p []byte) (int, error) {
	if b.consumed >= b.failAt {
		return 0, io.ErrUnexpectedEOF
	}
	left := b.failAt - b.consumed
	if left > len(p) {
		left = len(p)
	}
	n := b.remaining
	if n > left {
		n = left
	}
	for i := 0; i < n; i++ {
		p[i] = byte(i)
	}
	b.consumed += n
	b.remaining -= n
	if b.remaining == 0 {
		return n, io.EOF
	}
	return n, nil
}

// TestBlobColumnChainRoundTrip blob 列直接写"内容本身"([]byte / string / io.Reader) 时,
// 引擎自动落链页, 行内只留 8 字节引用; 读回逐字节一致。
//
// 这条钉的是 types/blob.go 里那句设计目标的执行落点 (Table.blobFieldRef): 用户不需要自己
// 开事务、自己调 BlobCommit 就能写 blob 列。同时钉住"Dynamicbyte 的单值上限不管 blob"——
// 超过一整块堆页 (MaxBlobSize) 的内容照样写得进, 因为链能跨页。
//
// 空内容 (长度为 0) 是特例: 不分配任何链页, 引用为空 (0/0), 读回来是"读不到值"。
func TestBlobColumnChainRoundTrip(t *testing.T) {
	db := fixOpenDB(t)
	tbl, err := db.CreateTable("blobcol", []types.FieldInfo{
		{FieldType: types.FieldType_Uint64, FieldName: []byte("id")},
		{FieldType: types.FieldType_Blob, FieldName: []byte("data")},
	})
	if err != nil {
		t.Fatalf("CreateTable: %v", err)
	}
	if _, err := tbl.CreateIndex("pk", page.IndexTypePrimary, []byte{0}); err != nil {
		t.Fatalf("CreateIndex: %v", err)
	}

	// ② 特意超过 Dynamicbyte 的单值上限: 只有链式存储装得下。
	big := make([]byte, MaxBlobSize*2+7)
	for i := range big {
		big[i] = byte(i * 7 % 251)
	}
	streamed := strings.Repeat("stream-", 3000)

	cases := []struct {
		pk   uint64
		val  interface{}
		want []byte
	}{
		{1, []byte("hello blob"), []byte("hello blob")},
		{2, big, big},
		{3, "string 载体", []byte("string 载体")},
		{4, strings.NewReader(streamed), []byte(streamed)},
		{5, []byte{}, []byte{}},
	}
	for _, c := range cases {
		if _, err := tbl.InsertRow([]types.FieldValue{
			{FieldName: []byte("id"), Value: c.pk},
			{FieldName: []byte("data"), Value: c.val},
		}); err != nil {
			t.Fatalf("InsertRow(pk=%d, %T): %v", c.pk, c.val, err)
		}
	}

	for _, c := range cases {
		vals, err := tbl.FindByPK(c.pk)
		if err != nil {
			t.Fatalf("FindByPK(%d): %v", c.pk, err)
		}
		var ref types.BlobRef
		var got []byte
		for _, fv := range vals {
			if string(fv.FieldName) != "data" {
				continue
			}
			r, ok := fv.Value.(types.BlobRef)
			if !ok {
				t.Fatalf("pk=%d: blob 列读出的值是 %T, 期望 types.BlobRef (行内只存 8 字节引用)",
					c.pk, fv.Value)
			}
			ref = r
			got, err = tbl.BlobRead(r, types.CompressNone)
			if err != nil {
				t.Fatalf("pk=%d: BlobRead(%+v): %v", c.pk, r, err)
			}
		}
		if !bytes.Equal(got, c.want) {
			t.Errorf("pk=%d: 读回 %d 字节, 期望 %d 字节 (内容不一致)", c.pk, len(got), len(c.want))
		}
		if int(ref.TotalLen) != len(c.want) {
			t.Errorf("pk=%d: ref.TotalLen = %d, 期望 %d", c.pk, ref.TotalLen, len(c.want))
		}
		// "超过堆页上限的内容"必须真的落在多页链上, 否则这条用例没验到跨页。
		if len(c.want) > int(MaxBlobSize) && ref.FirstPageID == 0 {
			t.Errorf("pk=%d: %d 字节的内容竟然没有链首页", c.pk, len(c.want))
		}
	}

	// 行内宽度必须是引用宽度 (8 字节), 不是"装得下多少内容" —— 这是能跨页的前提。
	if got := types.FieldSize(types.FieldType_Blob, 0); got != types.BlobRefSize {
		t.Errorf("blob 列行内宽度 = %d, 期望引用宽度 %d", got, types.BlobRefSize)
	}
}

// TestBlobColumnNullableRowLayout 带 NULL blob 列的行布局必须自洽。
//
// blob 列的行内是固定 8 字节引用槽位: NULL 只是不拷字节, 槽位照样要推进。布局预算
// (Pass 2) 与布局落地 (Pass 3) 必须用同一个占用公式 —— 预算按 NULL 记 0、落地却推进 8,
// 末尾那句"行布局长度失配"会让"写一行带 NULL blob 列"直接失败。
//
// 尾列 (变长) 用来验对齐: 槽位错位的话它读回来的就不是写下去的那串字节。
func TestBlobColumnNullableRowLayout(t *testing.T) {
	db := fixOpenDB(t)
	tbl, err := db.CreateTable("blobnull", []types.FieldInfo{
		{FieldType: types.FieldType_Uint64, FieldName: []byte("id")},
		{FieldType: types.FieldType_Blob, FieldName: []byte("data"), Flag: types.FlagNullable},
		{FieldType: types.FieldType_String, FieldName: []byte("tail"), Flag: types.WidthCodeFor(32)},
	})
	if err != nil {
		t.Fatalf("CreateTable: %v", err)
	}
	if _, err := tbl.CreateIndex("pk", page.IndexTypePrimary, []byte{0}); err != nil {
		t.Fatalf("CreateIndex: %v", err)
	}

	rows := []struct {
		pk   uint64
		data interface{}
		tail string
	}{
		{1, nil, "after-null-blob"},             // NULL blob
		{2, []byte("has-value"), "after-value"}, // 有值 blob, 与上一行布局共存
		{3, []byte{}, "after-empty"},            // 空 blob (零页)
	}
	for _, r := range rows {
		if _, err := tbl.InsertRow([]types.FieldValue{
			{FieldName: []byte("id"), Value: r.pk},
			{FieldName: []byte("data"), Value: r.data},
			{FieldName: []byte("tail"), Value: r.tail},
		}); err != nil {
			t.Fatalf("InsertRow(pk=%d, data=%T): %v", r.pk, r.data, err)
		}
	}

	for _, r := range rows {
		fvs, err := tbl.FindByPK(r.pk)
		if err != nil {
			t.Fatalf("FindByPK(%d): %v", r.pk, err)
		}
		var tail string
		var hasBlobValue bool
		for _, fv := range fvs {
			switch string(fv.FieldName) {
			case "tail":
				b, _ := fv.Value.([]byte)
				tail = string(b)
			case "data":
				hasBlobValue = true
			}
		}
		if tail != r.tail {
			t.Errorf("pk=%d: tail = %q, 期望 %q (NULL blob 的 8B 槽位没对齐?)", r.pk, tail, r.tail)
		}
		// 值为 nil 或空 blob 时, 行内是空引用, 解码侧不产出这一列的值。
		if r.data == nil && hasBlobValue {
			t.Errorf("pk=%d: NULL blob 列不该解出值", r.pk)
		}
	}
}

// TestBlobColumnRejectsUnsupportedValue blob 列收到引擎无法理解的值时必须报错点名。
//
// 宁可报错也不要"把 %T 序列化成某个字节串" —— 那会把一个整数静默变成 blob 内容, 用户
// 直到读回来才发现存的不是自己写的东西。
func TestBlobColumnRejectsUnsupportedValue(t *testing.T) {
	db := fixOpenDB(t)
	tbl, err := db.CreateTable("blobbad", []types.FieldInfo{
		{FieldType: types.FieldType_Uint64, FieldName: []byte("id")},
		{FieldType: types.FieldType_Blob, FieldName: []byte("data")},
	})
	if err != nil {
		t.Fatalf("CreateTable: %v", err)
	}
	if _, err := tbl.CreateIndex("pk", page.IndexTypePrimary, []byte{0}); err != nil {
		t.Fatalf("CreateIndex: %v", err)
	}

	_, err = tbl.InsertRow([]types.FieldValue{
		{FieldName: []byte("id"), Value: uint64(1)},
		{FieldName: []byte("data"), Value: int64(42)},
	})
	if err == nil {
		t.Fatal("blob 列写 int64 必须被拒")
	}
	if !strings.Contains(err.Error(), "data") {
		t.Errorf("报错应点出列名: %v", err)
	}
}