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) } }