package core import ( "bytes" "errors" "io" "github.com/golang/snappy" "git.ycl.pub/yl32940/TsDB2/DB/core/types" ) // BlobCommit 把 io.Reader 流式写入 blob 链, 一遍跑完即得到行内引用. // // 调用约定 (用户必须遵守): // 1. 必须在 Transaction 上下文中调用, 即 tx := db.BeginTransaction(); // 2. 流可被任意次 Read 消费; 库内部 io.Copy(buf=32KB) 循环, 全程峰值 ~128KB; // 3. Snappy 压缩按块 (snappy.NewBufferedWriter) 自动分段, 调用方无感; // 4. 调用方必须保证 compress 与将来的 BlobRead 一致 (库不持久化压缩标志). // // 中断保护: // - 流中途断 (conn closed, EOF 提前, Read 返错): 函数返 err, 此时 NewPage 出来的页 // 已在 BufferPool (脏页不刷盘) + tx.allocated 列表里; // - 调用方应 tx.Rollback(): 整链随 allocated 走"事务回滚归还空闲链"路径回收; // - 若 tx.Commit() 已被调用, 则链随提交落盘, 调用方需自管生命周期 (BlobFree). // // 内存峰值固定 (与 blob 大小无关): // // io.Copy buf 32KB + snappy 内部缓冲 ~64KB + 链适配器临时页 buf 16KB ≈ 128KB. // // 容量上限: // // BlobChainMax = 256K 页 ≈ 4 GB (原始数据); 超出返 ErrBlobTooLarge. func (tx *Transaction) BlobCommit(r io.Reader, compress types.FieldCompress) (BlobRef, error) { if tx == nil || tx.db == nil { return BlobRef{}, errors.New("BlobCommit: tx is nil") } return tx.db.blobCommit(r, compress) } // BlobCommit 无事务上下文版的 BlobCommit —— 让 *DB 直接满足 types.BlobCommitter。 // // 用途: ORM 层 Table.Insert/Update/Delete 把 *core.DB 当 blobc 传给 structToFieldValues, // 用户不需要手动开事务也能写 Blob 列 (内容自动落链, 行内只存 8B 引用)。 // // 事务语义: 调用时若恰有活跃事务 (db.currentTxn != nil), 链页经 DB.NewPage 自动登记进 // 该事务的 allocated 列表, Rollback 一样能整链回收; 无事务时链页与普通脏页同命运, // 随 Commit / Close 落盘。因此对调用方而言语义与 tx.BlobCommit 一致。 // // 千万不要为了拿这个入口回到"ORM 每行自动 begin-tx + 立即 Commit"的老路: tx.Commit // 只要缓冲池里有脏页 (单行插入必然有) 就走完整 WAL + fsync, ~37ms/行 —— 逐行插入 // 10000 行要 6 分钟, SetCommitPreset 的防抖/阈值机制被整体绕过 (a56b71d 引入、本次 // 修复的 1500 倍性能回退就是它)。 func (db *DB) BlobCommit(r io.Reader, compress types.FieldCompress) (BlobRef, error) { if db == nil { return BlobRef{}, errors.New("BlobCommit: db is nil") } return db.blobCommit(r, compress) } // blobCommit 是 BlobCommit 的无事务版本: 同样的流式写入, 但不要求调用方持有 Transaction. // // 为什么需要它: FieldType_Blob 列的行内只有 8 字节 BlobRef, 而用户往这一列写的是 // []byte / io.Reader 这种"内容本身" (见 Table.blobFieldRef). core 的 InsertRow / UpdateRow // 在无显式事务时也必须能落地 —— 否则"写一行 Blob 列"就得先手动开事务, 与 // types/blob.go 里"引擎自动走链页"的设计目标背离. // // 与 tx.BlobCommit 的唯一差别在失败回收: 有 currentTxn 时新页登记进 tx.allocated, 回滚即回收; // 无事务时没有 allocated 可回滚 —— 函数返回 err 后, 已分配的页按"调用方中止写入"语义处理: // 调用方是 core 行写入 (serializeRowInternal) 时, 整行写入失败会自动把本次新落的链还掉 // (见那里的 newBlobRefs 兜底); 调用方直接调 BlobCommit 的 (如 ORM 流式写), 仍需自管 // 生命周期 —— 成功拿 ref 后归调用方, 流中断返 err 时这批页只能等上层处置。 func (db *DB) blobCommit(r io.Reader, compress types.FieldCompress) (BlobRef, error) { if db == nil { return BlobRef{}, errors.New("BlobCommit: db is nil") } defer db.lockOp()() // 一处覆盖 tx.BlobCommit / db.BlobCommit 两个入口; 被 CRUD 路径调用时重入 if !compress.IsValid() { return BlobRef{}, errors.New("BlobCommit: unknown compress option") } if r == nil { // 与"空流"等价: 给空 blob return BlobRef{FirstPageID: 0, TotalLen: 0}, nil } // 0. 先探一个字节: 空输入连第一页都不分配 (否则空 blob 会白占一页 —— 提前开页是为了接住 // snappy 的 stream header, 但空输入压根没有 header 要接). var head [1]byte n, rerr := r.Read(head[:]) if rerr != nil && rerr != io.EOF { return BlobRef{}, rerr } if n == 0 { return BlobRef{FirstPageID: 0, TotalLen: 0}, nil } src := io.MultiReader(bytes.NewReader(head[:n]), r) // 1. 预开第一页 (snappy.Writer 第一次 Write 会先写 stream header, 必须有页接住) w, err := newBlobChainWriter(db) if err != nil { return BlobRef{}, err } // 2. io.Copy 链: r -> [snappy?] -> blobChainWriter // io.Copy 内部 32KB buf, snappy.NewBufferedWriter 内部 ~64KB, 都是固定大小. var totalRaw int64 if compress == types.CompressSnappy { sw := snappy.NewBufferedWriter(w) totalRaw, err = io.Copy(sw, src) if cerr := sw.Close(); cerr != nil && err == nil { err = cerr // Close 才能 flush snappy 末尾 } } else { totalRaw, err = io.Copy(w, src) } if err != nil { // 流中断 / io 错误: 链已分配但未 Commit, 由 tx.Rollback() 走 allocated 路径回收 return BlobRef{}, err } if totalRaw == 0 { // 有输入却一个字节都没落地 (压缩后为空的极端情况): 把预开的页还回去, 给空引用. db.deferFreeBlobChain(w.firstPage, w.tailPage, w.chainLen) return BlobRef{FirstPageID: 0, TotalLen: 0}, nil } if uint64(totalRaw) > uint64(^uint32(0)) { return BlobRef{}, ErrBlobTooLarge // 单 blob 上限 4GB } return BlobRef{ FirstPageID: w.firstPage, TotalLen: uint32(totalRaw), }, nil }