package DB import ( "bytes" "hash/crc32" "io" "path/filepath" "testing" "git.ycl.pub/yl32940/TsDB2/DB/core" "git.ycl.pub/yl32940/TsDB2/DB/core/types" ) // blobStreamRow 流式载体: 字段本身就是"持有流的对象" (types.Blob), // 不写 type:blob tag 也会被 goTypeToFieldType 自动推断为 blob 列。 type blobStreamRow struct { BaseModel Name string `vrtx:"maxlen:32"` Data types.Blob `vrtx:"type:blob"` } // TestBlobStreamCarrierRoundTrip 流式载体 (types.Blob) 的完整生命周期: // // 1. 写: Data.Source 是 io.Reader, 引擎流式落链 (不整块进内存); // 2. 回填: Insert 成功后 ref (FirstPageID/TotalLen) 已回填进对象, Source 清空; // 3. 读: Find 只填 ref, Open(table) 按需拿 io.ReadCloser 流式消费; // 4. 保链: 只改普通字段的 Update 不抹掉 blob (读回镜像分支); // 5. 重开: 关库重开照样流式读得回。 func TestBlobStreamCarrierRoundTrip(t *testing.T) { dbPath := filepath.Join(t.TempDir(), "blob_stream_orm.db") db, err := Open(dbPath) if err != nil { t.Fatalf("打开: %v", err) } defer func() { _ = db.Close() }() db.SetAutoCommitDelay(0) tbl, err := UseTable[blobStreamRow](db, "stream") if err != nil { t.Fatalf("建表: %v", err) } content := make([]byte, core.MaxBlobSize*3+7) // 跨 3+ 页, 单页链测不出链指针问题 for i := range content { content[i] = byte(i % 251) } wantCRC := crc32.ChecksumIEEE(content) row := &blobStreamRow{Name: "stream", Data: types.Blob{Source: bytes.NewReader(content)}} if err := tbl.Insert(row); err != nil { t.Fatalf("Insert: %v", err) } if err := db.Commit(); err != nil { t.Fatalf("Commit: %v", err) } // 2. ref 已回填, Source 已清 (再 Update 不会把死流灌一遍) if row.Data.FirstPageID == 0 || row.Data.TotalLen != uint32(len(content)) { t.Fatalf("Insert 后 ref 未回填: %+v", row.Data) } if row.Data.Source != nil { t.Fatalf("Insert 后 Source 应被清掉") } // 3. 读回: ref + Open 流式消费 got, err := tbl.Find(row.ID) if err != nil { t.Fatalf("Find: %v", err) } rc, err := got.Data.Open(tbl.BlobReader()) if err != nil { t.Fatalf("Open: %v", err) } data, err := io.ReadAll(rc) if err != nil { t.Fatalf("流式读: %v", err) } _ = rc.Close() if crc32.ChecksumIEEE(data) != wantCRC { t.Fatalf("流式读内容不一致") } // 4. 只改 Name 的 Update: blob 链必须原样保留 (曾因 blobFieldToRef 判空顺序 // 把镜像当空 blob, Update 一发整条链被抹掉) freeBefore := db.CoreDB().DebugFreePageCount() got.Name = "renamed" if err := tbl.Update(got); err != nil { t.Fatalf("Update: %v", err) } if err := db.Commit(); err != nil { t.Fatalf("Commit: %v", err) } renamed, err := tbl.Find(got.ID) if err != nil { t.Fatalf("Find: %v", err) } if renamed.Data.TotalLen != uint32(len(content)) { t.Fatalf("改名 Update 后 blob 丢失: TotalLen=%d", renamed.Data.TotalLen) } rc2, err := renamed.Data.Open(tbl.BlobReader()) if err != nil { t.Fatalf("Open: %v", err) } again, err := io.ReadAll(rc2) if err != nil { t.Fatalf("流式读: %v", err) } _ = rc2.Close() if crc32.ChecksumIEEE(again) != wantCRC { t.Fatalf("改名 Update 后内容不一致") } // 链页没被回收也没新开 (Update 未触碰 blob) if n := db.CoreDB().DebugFreePageCount(); n != freeBefore { t.Logf("注意: Update 前后空闲页 %d -> %d", freeBefore, n) } // 5. 重开库 if err := db.Close(); err != nil { t.Fatalf("关库: %v", err) } db2, err := Open(dbPath) if err != nil { t.Fatalf("重开: %v", err) } defer func() { _ = db2.Close() }() tbl2, err := UseTable[blobStreamRow](db2, "stream") if err != nil { t.Fatalf("重开建表: %v", err) } reloaded, err := tbl2.Find(row.ID) if err != nil { t.Fatalf("重开 Find: %v", err) } rc3, err := reloaded.Data.Open(tbl2.BlobReader()) if err != nil { t.Fatalf("重开 Open: %v", err) } persist, err := io.ReadAll(rc3) if err != nil { t.Fatalf("重开流式读: %v", err) } _ = rc3.Close() if crc32.ChecksumIEEE(persist) != wantCRC { t.Fatalf("重开后内容不一致") } }