blob_writer.go 5.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152
  1. package core
  2. import (
  3. "errors"
  4. "git.ycl.pub/yl32940/TsDB2/DB/core/page"
  5. "git.ycl.pub/yl32940/TsDB2/DB/core/types"
  6. )
  7. // blobChainWriter 把 io.Writer.Write 调用直接转为"写 blob 链尾", 自动跨页 NewPage。
  8. //
  9. // 约束:
  10. // - 一页 = 一个 blob chunk (与 HeapPage 那种"一页多块"完全不同);
  11. // - 链由 PageHdr.NextPage 串, 最后一页 NextPage = 0;
  12. // - 不维护 RefCount (单 owner, 删除时整链 FreeChain 一次 O(1) 还链)。
  13. //
  14. // 所有页头统一用 PageHdr (共 20 字节, 与 heap/index/schema-ext 一致):
  15. // PageHdr 占每页前 20 字节, 后续 Data 区就是 raw blob 字节, 无 chunk 头、无长度前缀。
  16. //
  17. // 错误:
  18. //
  19. // - ErrBlobTooLarge: 单 blob 链长超过 BlobChainMax (uint32 上限);
  20. //
  21. // 调用方应中止 io.Copy 并 tx.Rollback(), 由事务模型自动还链。
  22. type blobChainWriter struct {
  23. db *DB
  24. firstPage uint32 // 链首
  25. tailPage uint32 // 链尾 (Write 时优先填这一页, 满了再 NewPage)
  26. tailUsed uint16 // tailPage.Data 区已用字节
  27. chainLen uint32 // 已分配的页数 (含第一页)
  28. totalBytes uint64 // 已写入字节数 (= TotalLen)
  29. }
  30. // newBlobChainWriter 预分配第一页并把 writer 初始化到该页。
  31. //
  32. // 之所以必须"先 NewPage 再 Write": snappy.Writer 在第一次 Write 之前会先写一个
  33. // stream header (几字节), 如果第一页还没 NewPage, snappy 的 Write 会落到 nil tailPage
  34. // 上 panic。预开第一页后, 写入路径与 snappy 不耦合。
  35. //
  36. // 参数是 *DB 而不是 *Transaction: 链页的分配 (NewPage) 与写盘 (file.WritePage) 都不需要
  37. // 事务参与 —— 事务只是在有 currentTxn 时顺带把新页登记进 allocated (回滚时归还)。
  38. // 于是"没有显式事务"的写入路径 (core 层 InsertRow 直接塞 []byte, 见 Table.blobFieldRef)
  39. // 也能复用同一个 writer。
  40. func newBlobChainWriter(db *DB) (*blobChainWriter, error) {
  41. pid := db.NewPage()
  42. buf := make([]byte, PageSize)
  43. hdr := page.HdrOf(buf)
  44. hdr.Init(page.BlobIndex, pid)
  45. hdr.NextPage = 0
  46. hdr.UseSize = page.PageHdrSize
  47. if err := db.file.WritePage(pid, buf); err != nil {
  48. return nil, err
  49. }
  50. return &blobChainWriter{
  51. db: db,
  52. firstPage: pid,
  53. tailPage: pid,
  54. tailUsed: 0,
  55. chainLen: 1,
  56. totalBytes: 0,
  57. }, nil
  58. }
  59. // Write 实现 io.Writer. 一次 Write 可能跨多页, 由内部 NewPage 衔接.
  60. //
  61. // 返回值语义: 总是返回 (len(p), nil) 或 (已写入字节, err).
  62. // 错误只可能是 ErrBlobTooLarge, 一旦发生, 链已被 NewPage 占用,
  63. // 调用方应中止 io.Copy 并 tx.Rollback() 走事务模型回收整链.
  64. func (w *blobChainWriter) Write(p []byte) (int, error) {
  65. const pageData = uint32(PageSize - page.PageHdrSize) // 16364
  66. written := 0
  67. for written < len(p) {
  68. // 1. 当前尾页已满? 开新页并挂链
  69. if uint32(w.tailUsed) >= pageData {
  70. if w.chainLen >= BlobChainMax {
  71. return written, ErrBlobTooLarge
  72. }
  73. newPID := w.db.NewPage()
  74. // 挂到老尾页的 NextPage
  75. tailBuf := make([]byte, PageSize)
  76. if err := w.db.file.ReadPage(w.tailPage, tailBuf); err != nil {
  77. return written, err
  78. }
  79. page.HdrOf(tailBuf).NextPage = newPID
  80. if err := w.db.file.WritePage(w.tailPage, tailBuf); err != nil {
  81. return written, err
  82. }
  83. // 初始化新页
  84. newBuf := make([]byte, PageSize)
  85. hdr := page.HdrOf(newBuf)
  86. hdr.Init(page.BlobIndex, newPID)
  87. hdr.NextPage = 0
  88. hdr.UseSize = page.PageHdrSize
  89. if err := w.db.file.WritePage(newPID, newBuf); err != nil {
  90. return written, err
  91. }
  92. // 推进尾指针
  93. w.tailPage = newPID
  94. w.tailUsed = 0
  95. w.chainLen++
  96. }
  97. // 2. 计算本批可写字节数 (min(本页剩余空间, p 剩余字节))
  98. space := uint32(pageData - uint32(w.tailUsed))
  99. remaining := uint32(len(p) - written)
  100. n := space
  101. if remaining < n {
  102. n = remaining
  103. }
  104. // 3. 读尾页 + 写本批
  105. tailBuf := make([]byte, PageSize)
  106. if err := w.db.file.ReadPage(w.tailPage, tailBuf); err != nil {
  107. return written, err
  108. }
  109. copy(tailBuf[page.PageHdrSize+uint32(w.tailUsed):], p[written:written+int(n)])
  110. page.HdrOf(tailBuf).UseSize = page.PageHdrSize + uint32(w.tailUsed) + n
  111. if err := w.db.file.WritePage(w.tailPage, tailBuf); err != nil {
  112. return written, err
  113. }
  114. w.tailUsed += uint16(n)
  115. written += int(n)
  116. w.totalBytes += uint64(n)
  117. }
  118. return written, nil
  119. }
  120. // ErrBlobTooLarge 单 blob 链长超过 BlobChainMax (256K 页 = ~4GB).
  121. var ErrBlobTooLarge = errors.New("blob chain exceeds maximum length")
  122. // BlobChainMax 单 blob 链页数上限, 值 256 * 1024.
  123. // 定义在 const.go (库级常量统一管理), 本文件直接引用.
  124. //
  125. // 上限由来: 链长用 uint32 计数 (避免 RowFieldRef 里的 2 字节截断隐患),
  126. // 单页 16KB, 256K 页 ≈ 4GB. 这个量级对绝大多数 binary 大对象够用;
  127. // 超大文件 (>4GB) 应改用对象存储而非单字段.
  128. // BlobRef 行内 8 字节引用. 类型权威定义见 types.BlobRef (本类型为 alias, 历史代码兼容)。
  129. //
  130. // [TotalLen 4B BE][FirstPageID 4B BE]
  131. //
  132. // TotalLen = 原始 (解压后) 字节数;
  133. // FirstPageID = 0 表示空 blob (与 TotalLen = 0 一同出现).
  134. //
  135. // 【为什么要 alias】types 包持有 BlobRef 是为了让 EncodeValue / DecodeValue 不必反向依赖
  136. // core 包 (core 已经依赖 types, 反过来会循环)。core 内的历史 API (BlobCommit / BlobRead
  137. // / BlobFree 等) 全部继续用这个名字, 直接 type 别名过去, 调用方代码零改动。
  138. //
  139. // types.BlobRef 已定义 Empty / SizeBytes 方法, alias 自动继承, 不在这里再写一次
  140. // (否则是 duplicate method)。
  141. type BlobRef = types.BlobRef