vrtx.go 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623
  1. package migrate
  2. import (
  3. "TsDB2/DB/core"
  4. "TsDB2/DB/core/page"
  5. "TsDB2/DB/core/types"
  6. "TsDB2/DB/core/util"
  7. "errors"
  8. "fmt"
  9. "sort"
  10. "strings"
  11. "unicode/utf8"
  12. )
  13. // VRTX 一侧。同一个对象既能当 Source 也能当 Sink (方向由调用方决定), 于是任意
  14. // 方向 (VRTX→SQL / SQL→VRTX / VRTX→VRTX) 都只需要这一套读写实现。
  15. //
  16. // 这里刻意走 core 层而不是 ORM (package DB): 表结构在迁移时来自源库的元数据
  17. // (列名 / 类型 / 行存储索引列), 运行时才知道, 根本没有一个 Go 结构体可以内嵌表基类 ——
  18. // 而 ORM 的入口 UseTable[T] 恰恰要求 T 内嵌 DB.BaseModel / DB.TimeSeriesModel
  19. // (见 DB 包的 requireBase)。core 层只认 []types.FieldInfo, 没有"先有结构体"这个前提,
  20. // 所以"按库里的元数据建表"只能落在 core 层。ORM 是给"写代码时就知道表结构"的场景用的。
  21. type VRTX struct {
  22. db *core.DB
  23. path string
  24. opts *Options
  25. }
  26. // OpenVRTX 打开 (不存在则新建) 一个 VRTX 库文件。
  27. func OpenVRTX(path string, opts *Options) (*VRTX, error) {
  28. db, err := core.Open(path)
  29. if err != nil {
  30. return nil, fmt.Errorf("打开 VRTX 库 %s: %w", path, err)
  31. }
  32. return &VRTX{db: db, path: path, opts: opts.normalized()}, nil
  33. }
  34. func (v *VRTX) Kind() Kind { return KindVRTX }
  35. func (v *VRTX) Identity() string { return "vrtx:" + v.path }
  36. // DB 暴露底层句柄 (调用方可能需要调缓冲池 / 手动提交)。
  37. func (v *VRTX) DB() *core.DB { return v.db }
  38. // Commit 把当前脏页落盘。编排器每迁完一张表调一次: VRTX 的插入是攒在缓冲池里的,
  39. // 千万行的库若要等最后 Close 才提交, 中途崩一次就是全部白干, 内存也会一直涨。
  40. func (v *VRTX) Commit() error { return v.db.Commit() }
  41. // Close 落盘并关闭。
  42. func (v *VRTX) Close() error {
  43. // 迁移是批量写, 收尾必须显式提交一次: 否则最后一批脏页只剩防抖定时器兜底,
  44. // 进程立刻退出就会丢。
  45. if err := v.db.Commit(); err != nil {
  46. _ = v.db.Close()
  47. return err
  48. }
  49. return v.db.Close()
  50. }
  51. // Tables 列出全部表 (core 已跳过软删除的表)。
  52. func (v *VRTX) Tables() ([]string, error) {
  53. metas := v.db.GetAllTables()
  54. names := make([]string, 0, len(metas))
  55. for _, m := range metas {
  56. names = append(names, m.Name)
  57. }
  58. sort.Strings(names) // 稳定顺序: 迁移过程可复现、可对照
  59. return names, nil
  60. }
  61. // Schema 读一张表的结构。
  62. //
  63. // 列取 master registry 里未被软删除的那些 —— 注意用 master 顺序而不是
  64. // GetActiveFieldNames(): 后者是"当前 schema 版本"的列序, 而 master registry 的
  65. // 下标才是索引项里 FieldIDs 引用的那个下标, 换算列名必须用它。
  66. func (v *VRTX) Schema(table string) (*TableSchema, error) {
  67. tbl, ok := v.db.FindTable(table)
  68. if !ok {
  69. return nil, fmt.Errorf("VRTX 库 %s 里没有表 %s", v.path, table)
  70. }
  71. fields := tbl.GetFields() // master registry: 含已软删除的列 (删除标记见 types.FlagDeleted)
  72. s := &TableSchema{Name: table}
  73. for _, f := range fields {
  74. if f.IsDropped() {
  75. continue
  76. }
  77. s.Columns = append(s.Columns, Column{
  78. Name: util.BytesToString(f.FieldName),
  79. Type: f.FieldType,
  80. Param: f.Flag & types.FlagScaleMask, // Decimal 的列级 scale, 必须跟着列走
  81. // NULL / 唯一声明: 前者决定导出时写不写 NOT NULL, 后者是列级声明
  82. // (约束本身由索引执行, 见下面 GetIndexes 的 m.Unique())。
  83. Nullable: f.IsNullable(),
  84. Unique: f.IsUnique(),
  85. })
  86. }
  87. im := tbl.IndexMeta()
  88. pk, hasPK := im.GetRowStoreIndex()
  89. if !hasPK {
  90. return nil, fmt.Errorf("VRTX 表 %s 没有行存储索引 (主键/时序), 行数据无处安放", table)
  91. }
  92. s.PKIndexName = pk.Name()
  93. s.PKIndexType = pk.IndexType
  94. pkCols := make([]string, 0, pk.FieldCnt)
  95. for _, fid := range pk.Fields() {
  96. if int(fid) >= len(fields) {
  97. continue
  98. }
  99. if util.IsHighBitSet(byte(fields[fid].FieldType)) {
  100. continue
  101. }
  102. pkCols = append(pkCols, util.BytesToString(fields[fid].FieldName))
  103. }
  104. switch len(pkCols) {
  105. case 0:
  106. // 行存储索引的分量列全被删了: 保留 PKColumn 为空, 由下面补列
  107. case 1:
  108. s.PKColumn = pkCols[0]
  109. default:
  110. return nil, fmt.Errorf("VRTX 表 %s 的行存储索引 %s 有 %d 个分量, 不是合法的行存储索引",
  111. table, pk.Name(), len(pkCols))
  112. }
  113. for _, m := range im.GetIndexes() {
  114. if m.Name() == s.PKIndexName {
  115. continue // 主键索引由 PKColumn / PKIndexType 表达
  116. }
  117. cols := make([]string, 0, m.FieldCnt)
  118. broken := false
  119. for _, fid := range m.Fields() {
  120. if int(fid) >= len(fields) || util.IsHighBitSet(byte(fields[fid].FieldType)) {
  121. broken = true // 分量列已被删除: 索引名存实亡, 导过去只会是个坏索引
  122. break
  123. }
  124. cols = append(cols, util.BytesToString(fields[fid].FieldName))
  125. }
  126. if broken || len(cols) == 0 {
  127. v.opts.logf("[vrtx] 表 %s 的索引 %s 有分量列已被删除, 跳过", table, m.Name())
  128. continue
  129. }
  130. s.Indexes = append(s.Indexes, IndexSchema{
  131. Name: m.Name(),
  132. Type: m.IndexType,
  133. Columns: cols,
  134. Unique: m.Unique(), // IndexMeta.Flag 的唯一位: 约束的执行者
  135. })
  136. }
  137. // 时序索引: 行存储的键是"时间列", 与行号无关 —— 既不能当 SQL 主键 (主键得是整数),
  138. // 也不能把行号往时间列里写。这类表导出到 SQL 时不设主键, 时间列照常是普通列;
  139. // 行存储索引的类型和键列记在 __vrtx_meta 里, 回来时原样重建。
  140. //
  141. // 顺带说明为什么不给时间列补 _id 或普通索引: 时序索引自己维护的 pkID 并不可靠
  142. // (首个分段落盘时会把首列数据当成 pkID 存, 见 datapage.CreateDataPage), 所以这条
  143. // 路径上干脆不读它, 让目标端按自增号分配; 时间列是行存储键, 建普通索引也是重复。
  144. if s.PKIndexType == page.IndexTypeTime && s.PKColumn != "" &&
  145. pinRowStorePK(s, s.PKColumn) {
  146. s.RowStoreColumns = append([]string(nil), pkCols...)
  147. s.PKColumn = ""
  148. return s, nil
  149. }
  150. // 其余情况行主键必须能承载行号 (SQL 的 PRIMARY KEY 只能是整数, VRTX 侧这个列
  151. // 也是 pkID 的镜像)。行存储索引挂在定长串这类非 8 字节列上时, 该列的值与行号
  152. // 不是一回事 —— 补一个 _id 列专门承载行号, 原列退化成普通列。
  153. //
  154. // pinRowStorePK 顺带把"有符号 8 字节整数"钉回 uint64 档 (见 rowStorePKType)。
  155. if !pinRowStorePK(s, s.PKColumn) {
  156. name := freeColumnName(s, "_id")
  157. s.Columns = append(s.Columns, Column{Name: name, Type: types.FieldType_Uint64})
  158. s.MarkSynthetic(name)
  159. if s.PKColumn != "" {
  160. v.opts.logf("[vrtx] 表 %s 的行存储索引挂在 %s (%s) 上, 不是 8 字节整数列, "+
  161. "补列 %s 承载行号", table, s.PKColumn, TypeName(typeOfColumn(s, s.PKColumn)), name)
  162. // 原列变成普通列, 不参与主键; 但"按它查"是原来的用法, 补个索引续上
  163. ensureColumnIndex(s, s.PKColumn)
  164. }
  165. s.PKIndexType = page.IndexTypePrimary
  166. s.PKColumn = name
  167. }
  168. return s, nil
  169. }
  170. // Scan 全表扫描。回调拿到的 Row.Values 与 s.Columns 同序。
  171. func (v *VRTX) Scan(table string, s *TableSchema, fn func(Row) error) error {
  172. tbl, ok := v.db.FindTable(table)
  173. if !ok {
  174. return fmt.Errorf("VRTX 库 %s 里没有表 %s", v.path, table)
  175. }
  176. pos := make(map[string]int, len(s.Columns))
  177. for i, c := range s.Columns {
  178. pos[c.Name] = i
  179. }
  180. pkIdx := columnIndex(s, s.PKColumn)
  181. diverged := false
  182. var fnErr error
  183. scanErr := tbl.Query().ScanWithPKID(func(pkID uint64, vals []types.FieldValue) bool {
  184. row := Row{PK: pkID, HasPK: true, Values: make([]any, len(s.Columns))}
  185. for _, fv := range vals {
  186. i, ok := pos[util.BytesToString(fv.FieldName)]
  187. if !ok {
  188. continue // 已删除的列 / 不在导出范围内的列
  189. }
  190. // Blob 列的行内值只有 8 字节 BlobRef (指向链页), 而中间表示要的是【内容本身】:
  191. // 在这一层解引用一次, 下游的 coerceForSQL / coerceForVRTX 就只面对 []byte,
  192. // 不必各自认识链引用。
  193. if ref, ok := fv.Value.(types.BlobRef); ok {
  194. data, err := tbl.BlobRead(ref, types.CompressNone)
  195. if err != nil {
  196. fnErr = fmt.Errorf("表 %s 读回 blob 列 %s (ref=%+v): %w",
  197. table, util.BytesToString(fv.FieldName), ref, err)
  198. return false
  199. }
  200. if data == nil {
  201. data = []byte{}
  202. }
  203. row.Values[i] = data
  204. continue
  205. }
  206. row.Values[i] = fv.Value
  207. }
  208. // 主键列以行号为准: 它本来就是 pkID 的镜像。真有对不上的 (绕过 ORM 直接
  209. // 用 core 写入、主键列给了别的值), 说一声, 免得以为是导错了。
  210. if pkIdx >= 0 {
  211. if old := row.Values[pkIdx]; old != nil {
  212. if n, err := asUint64(old); err == nil && n != pkID && !diverged {
  213. diverged = true
  214. v.opts.logf("[vrtx] 表 %s 有行的主键列 %s 与行号不一致 (如 %v ≠ %d), "+
  215. "导出时以行号为准", table, s.PKColumn, old, pkID)
  216. }
  217. }
  218. row.Values[pkIdx] = pkID
  219. }
  220. if err := fn(row); err != nil {
  221. fnErr = err
  222. return false
  223. }
  224. return true
  225. })
  226. if fnErr != nil {
  227. return fnErr
  228. }
  229. if scanErr != nil {
  230. return fmt.Errorf("扫描 VRTX 表 %s: %w", table, scanErr)
  231. }
  232. return nil
  233. }
  234. // ============================ Sink ============================
  235. // Exists 目标表是否存在。
  236. func (v *VRTX) Exists(table string) (bool, error) {
  237. _, ok := v.db.FindTable(table)
  238. return ok, nil
  239. }
  240. // CreateTable 建表 + 建索引 (主键索引必须建, 否则行数据无处安放)。
  241. func (v *VRTX) CreateTable(s *TableSchema, recreate bool) error {
  242. exists, err := v.Exists(s.Name)
  243. if err != nil {
  244. return err
  245. }
  246. if exists && recreate {
  247. if err := v.db.DropTable(s.Name); err != nil {
  248. return fmt.Errorf("DROP VRTX 表 %s: %w", s.Name, err)
  249. }
  250. exists = false
  251. }
  252. if exists {
  253. return nil // 复用已有表: 列与索引保持原样, 数据追加
  254. }
  255. if len(s.RowStore()) == 0 {
  256. return fmt.Errorf("VRTX 表 %s 没有行存储索引的列 (主键列为空), 无法建表", s.Name)
  257. }
  258. if len(s.Name) > core.TableNameSize {
  259. return fmt.Errorf("表名 %s 有 %d 字节, 超过 VRTX 上限 %d", s.Name, len(s.Name), core.TableNameSize)
  260. }
  261. // 单列唯一索引 ⇒ 这一列同时是"唯一列", 把列上的声明位补上。
  262. //
  263. // 声明与约束是两件事 (列位给读者看, 索引负责拦), 但只有索引这一半能从 SQL 读
  264. // 出来 —— SQLite 的列级 UNIQUE 在 PRAGMA table_info 里毫无痕迹, 只生成一条
  265. // autoindex。补位不会凭空多出约束 (约束一直是索引给的), 少了它却会让从外部库
  266. // 迁进来的表"看不出这列本该唯一"。复合唯一索引不补: (A,B) 唯一不等于 B 唯一。
  267. for _, idx := range s.Indexes {
  268. if !idx.Unique || len(idx.Columns) != 1 {
  269. continue
  270. }
  271. if ci := columnIndex(s, idx.Columns[0]); ci >= 0 {
  272. s.Columns[ci].Unique = true
  273. }
  274. }
  275. tbl, err := v.db.CreateTable(s.Name, vrtxFields(s))
  276. if err != nil {
  277. return fmt.Errorf("建 VRTX 表 %s: %w", s.Name, err)
  278. }
  279. rsCols := s.RowStore()
  280. if len(rsCols) == 0 {
  281. return fmt.Errorf("表 %s 没有行存储索引的列 (主键列为空)", s.Name)
  282. }
  283. pkType := s.PKIndexType
  284. if pkType != page.IndexTypeTime {
  285. pkType = page.IndexTypePrimary
  286. }
  287. ids := make([]byte, 0, len(rsCols))
  288. for _, col := range rsCols {
  289. ci := columnIndex(s, col)
  290. if ci < 0 {
  291. return fmt.Errorf("表 %s 的行存储索引列 %s 不在列清单里", s.Name, col)
  292. }
  293. if pkType == page.IndexTypeTime && s.Columns[ci].Type != types.FieldType_Uint64 {
  294. // 时序索引的键就是 8 字节时间戳, 宽度不对会在写入时直接崩。
  295. return fmt.Errorf("表 %s 的行存储索引列 %s 是 %s, 时序索引要求 8 字节整数列",
  296. s.Name, col, TypeName(s.Columns[ci].Type))
  297. }
  298. ids = append(ids, byte(ci))
  299. }
  300. pkName := s.PKIndexName
  301. if pkName == "" {
  302. pkName = "pk_" + s.Name
  303. }
  304. if err := checkIndexName(pkName); err != nil {
  305. return err
  306. }
  307. if _, err := tbl.CreateIndex(pkName, pkType, ids); err != nil {
  308. return fmt.Errorf("建主键索引 %s: %w", pkName, err)
  309. }
  310. for _, idx := range s.Indexes {
  311. if idx.Name == pkName || idx.IsRowStore() {
  312. continue
  313. }
  314. // 索引名超长 / 分量过多: 与下面的"动态列当键"同一处理 —— 跳过并记进报告。
  315. // 这两种都是"源库合法、VRTX 装不下" (SQLite 的索引名与列数几乎没有限制), 但都
  316. // 只影响这条索引, 不该让整张表的数据陪葬。以前这里直接 return err, 于是一个
  317. // 名字太长的索引能把整张表拦下来。
  318. if bad := auditIndexForVRTX(idx); bad != "" {
  319. s.SkippedIndexes = append(s.SkippedIndexes, idx.Name+": "+bad)
  320. v.opts.logf("[vrtx] 表 %s 的索引 %s 建不了, 跳过 (%s)", s.Name, idx.Name, bad)
  321. continue
  322. }
  323. ids := make([]byte, 0, len(idx.Columns))
  324. bad := ""
  325. for _, col := range idx.Columns {
  326. ci := columnIndex(s, col)
  327. if ci < 0 {
  328. bad = "列 " + col + " 不在表中"
  329. break
  330. }
  331. if s.Columns[ci].IsDynamic() {
  332. // 源库里它是条合法索引 (MySQL 的长 varchar / text 都能建), 但 VRTX
  333. // 行内只存动态列的堆引用, 按值查不到, 引擎会直接拒 (ErrIndexDynamicField)。
  334. // 跳过这个索引, 而不是让整张表失败 —— 索引丢了还能补, 数据没迁才是真丢。
  335. bad = fmt.Sprintf("键列 %s 是 %s 动态列, VRTX 不能拿它当索引键",
  336. col, TypeName(s.Columns[ci].Type))
  337. break
  338. }
  339. if s.Columns[ci].Type == types.FieldType_Blob {
  340. // 同理 (v10 起 Blob 的行内是 8 字节 BlobRef): 索引里会变成页号而不是内容,
  341. // 引擎在 CreateIndex 直接拒 (ErrIndexBlobField)。这里提前跳过, 免得一个
  342. // BLOB 列上的索引把整张表的迁移拦下来。
  343. bad = fmt.Sprintf("键列 %s 是 blob 列, VRTX 不能拿它当索引键 (行内只有 8 字节链引用)",
  344. col)
  345. break
  346. }
  347. ids = append(ids, byte(ci))
  348. }
  349. if bad != "" {
  350. s.SkippedIndexes = append(s.SkippedIndexes, idx.Name+": "+bad)
  351. v.opts.logf("[vrtx] 表 %s 的索引 %s 建不了, 跳过 (%s)", s.Name, idx.Name, bad)
  352. continue
  353. }
  354. idxType := idx.Type
  355. if idxType == 0 || idxType == page.IndexTypePrimary || idxType == page.IndexTypeTime {
  356. idxType = page.IndexTypeBTree
  357. }
  358. if _, err := tbl.CreateIndex(idx.Name, idxType, ids, idx.Unique); err != nil {
  359. return fmt.Errorf("建索引 %s: %w", idx.Name, err)
  360. }
  361. }
  362. return nil
  363. }
  364. // Open 准备写入。
  365. func (v *VRTX) Open(table string, s *TableSchema) (RowWriter, error) {
  366. tbl, ok := v.db.FindTable(table)
  367. if !ok {
  368. return nil, fmt.Errorf("VRTX 库 %s 里没有表 %s (要先建表)", v.path, table)
  369. }
  370. return &vrtxWriter{tbl: tbl, schema: s, opts: v.opts, pkIdx: columnIndex(s, s.PKColumn)}, nil
  371. }
  372. type vrtxWriter struct {
  373. tbl *core.Table
  374. schema *TableSchema
  375. opts *Options
  376. pkIdx int
  377. }
  378. // Write 逐行写入。VRTX 没有"批量插入"接口 (每行都要维护 B+ 树与索引项),
  379. // 所以这里就是 for 循环 —— 批的意义在于 SQL 侧的往返次数。
  380. func (w *vrtxWriter) Write(rows []Row) error {
  381. vals := make([]types.FieldValue, len(w.schema.Columns))
  382. for i, c := range w.schema.Columns {
  383. vals[i].FieldName = []byte(c.Name)
  384. }
  385. for ri := range rows {
  386. r := &rows[ri]
  387. pkID := r.PK
  388. if !r.HasPK || pkID == 0 {
  389. pkID = w.tbl.AllocPKID()
  390. }
  391. for i, c := range w.schema.Columns {
  392. var raw any
  393. if i < len(r.Values) {
  394. raw = r.Values[i]
  395. }
  396. if i == w.pkIdx {
  397. raw = pkID // 主键列是行号的镜像
  398. }
  399. cv, err := coerceForVRTX(c.Type, c.Param, raw, c.Name, w.opts.StrictValue, c.Nullable)
  400. if err != nil {
  401. return fmt.Errorf("%s: %w", w.rowTag(r, ri), err)
  402. }
  403. if w.opts.TruncateOversize {
  404. cv = w.truncateOversize(cv, c, r, ri)
  405. }
  406. vals[i].Value = cv
  407. }
  408. if _, err := w.tbl.InsertRowWithID(pkID, vals); err != nil {
  409. return fmt.Errorf("%s: 写入 VRTX 表 %s: %w", w.rowTag(r, ri), w.schema.Name, err)
  410. }
  411. }
  412. return nil
  413. }
  414. // rowTag 给写失败的那一行拼一个"回源库找得到它"的标识。
  415. //
  416. // 迁移报错时用户真正要做的动作是回源库处理那一行 (把列改大 / 把值裁小 / 剔掉), 只报
  417. // "表 x 的 y 列超限"等于让人拿 LENGTH(y) 去全表捞 —— 动态列的容量上限与源库的声明无关,
  418. // 光看列名反推不出是哪一行。
  419. //
  420. // 三种身份按可靠程度排:
  421. // 1. 源库主键列的值: 源库自己认得, 能直接 SELECT ... WHERE id = ?。主键不是 8 字节整数、
  422. // 行号被换成引擎分配的 _id 的表 (MySQL 常见的 int(11) 主键都走这条) 只能靠它;
  423. // 2. VRTX 行号: 源库主键本身就是行号时, 它等于源库主键值, 一样能定位;
  424. // 3. 批内序号: 前两条都拿不到 (源表没有整型主键) 时的兜底, 只能把范围缩到那一批。
  425. func (w *vrtxWriter) rowTag(r *Row, batchPos int) string {
  426. if i := columnIndex(w.schema, w.schema.SourcePKColumn); i >= 0 && i < len(r.Values) {
  427. if v := r.Values[i]; v != nil {
  428. if b, ok := v.([]byte); ok {
  429. return fmt.Sprintf("源表主键 %s", b)
  430. }
  431. return fmt.Sprintf("源表主键 %v", v)
  432. }
  433. }
  434. if r.HasPK {
  435. return fmt.Sprintf("主键 %d", r.PK)
  436. }
  437. return fmt.Sprintf("批内第 %d 行", batchPos+1)
  438. }
  439. // truncateOversize 把超过单值上限的 Dynamicbyte 值截到上限, 并逐行报出截了什么 (见
  440. // Options.TruncateOversize 说明为什么这不是默认行为)。
  441. //
  442. // 只动 Dynamicbyte, 也只动"超过单值上限"这一种: 定长列装不下是另一码事, 归 StrictValue /
  443. // -loose 管; blob 列的内容在链上 (单值约 4GB), 没有这个上限。
  444. func (w *vrtxWriter) truncateOversize(v any, c Column, r *Row, batchPos int) any {
  445. if !types.IsDynamic(c.Type) {
  446. return v
  447. }
  448. b, ok := v.([]byte)
  449. if !ok || uint32(len(b)) <= core.MaxBlobSize {
  450. return v
  451. }
  452. // 用的是引擎那道闸门的同一个常量 (core.MaxBlobSize), 不是迁移侧自己的推算 ——
  453. // 截到"闸门允许的最大值"才既装得下又不白丢。
  454. cut := runeSafeCut(b, int(core.MaxBlobSize))
  455. w.opts.logf("[migrate] 表 %s %s 列 %s 的值 %d 字节超过单值上限 %d, 已截成 %d 字节 (丢了 %d 字节)",
  456. w.schema.Name, w.rowTag(r, batchPos), c.Name, len(b), core.MaxBlobSize, cut, len(b)-cut)
  457. return b[:cut]
  458. }
  459. // runeSafeCut 求一个不超过 limit 的截断位置, 并且不把一个 UTF-8 字符切成两半。
  460. //
  461. // 判据是"被排除的第一个字节是不是字符起始字节": 原值本身就是合法 UTF-8 时, 从字符边界
  462. // 切开得到的必然是合法文本; 二进制列 (本来就不是文本) 退到最近一个非续字节即可, 少
  463. // 1..3 个字节对它没有意义。
  464. func runeSafeCut(b []byte, limit int) int {
  465. if limit >= len(b) {
  466. return len(b)
  467. }
  468. cut := limit
  469. for cut > 0 && !utf8.RuneStart(b[cut]) {
  470. cut--
  471. }
  472. if cut == 0 {
  473. // 上限前面全是续字节 —— 这不是文本, 没有字符边界可退, 老实按字节截,
  474. // 免得把整个值截成空的。
  475. return limit
  476. }
  477. return cut
  478. }
  479. // Close 提交这一批 (VRTX 的提交由 DB 层统一做, 这里只留扩展点)。
  480. func (w *vrtxWriter) Close() error { return nil }
  481. // ============================ 工具 ============================
  482. func checkIndexName(name string) error {
  483. if name == "" {
  484. return errors.New("索引名为空")
  485. }
  486. if len(name) > core.IndexNameSize {
  487. return fmt.Errorf("索引名 %s 有 %d 字节, 超过 VRTX 上限 %d (换短一点的名字)",
  488. name, len(name), core.IndexNameSize)
  489. }
  490. return nil
  491. }
  492. // columnFlag 列的 VRTX 标志位: 低 5 位 scale + NULL / 唯一声明。
  493. //
  494. // 三者共用一个字节 (布局见 types 的 Flag* 常量), 少写任何一位都是静默改结构:
  495. // 丢了 FlagNullable, 往这列写 NULL 会被当成"给个零值"塞 0; 丢了 FlagUnique,
  496. // 反向推 schema 的工具 (以及 ORM 的 tag) 就看不出这列本该唯一。
  497. func columnFlag(c Column) byte {
  498. f := c.Param & types.FlagScaleMask
  499. if c.Nullable {
  500. f |= types.FlagNullable
  501. }
  502. if c.Unique {
  503. f |= types.FlagUnique
  504. }
  505. return f
  506. }
  507. // vrtxFields 把列清单翻成 core 建表要的 []types.FieldInfo。
  508. //
  509. // 抽出来是因为它有两个用途, 两处必须逐字节一致 —— 建表用它, 行宽闸门 (更宽的行装不进
  510. // 一页两行) 也用它算行宽 (见 fitRowWidthForVRTX)。各写一份的话, 一旦某天只改了建表那份
  511. // (比如新增一个列标志位), 行宽闸门算的就是另一张表: 闸门放过的表建不出来。
  512. func vrtxFields(s *TableSchema) []types.FieldInfo {
  513. fields := make([]types.FieldInfo, 0, len(s.Columns))
  514. for _, c := range s.Columns {
  515. fields = append(fields, types.FieldInfo{
  516. FieldType: c.Type,
  517. FieldName: []byte(c.Name),
  518. Flag: columnFlag(c), // decimal scale + NULL / 唯一声明
  519. })
  520. }
  521. return fields
  522. }
  523. func columnIndex(s *TableSchema, name string) int {
  524. if i, ok := s.ColumnIdx(name); ok {
  525. return i
  526. }
  527. return -1
  528. }
  529. func typeOfColumn(s *TableSchema, name string) types.FieldType {
  530. if i, ok := s.ColumnIdx(name); ok {
  531. return s.Columns[i].Type
  532. }
  533. return 0
  534. }
  535. // rowStorePKType 判断某列能否直接承载行号 (pkID 的镜像), 并给出它在 VRTX 侧必须使用的类型。
  536. //
  537. // 行存储索引的键就是一个 8 字节整数列: 它的值即引擎行主键号, 主键 B+ 树按它的字节序排列;
  538. // 时序表里这一列同时承载时间戳, 时序索引按 int64 补码直读这 8 个字节。因此它在 VRTX 侧
  539. // 必须落在【无符号 uint64 档】上 —— 盘上就是补码本身。
  540. //
  541. // 而 SQL 的整数默认被 inferSQLType 收编到有符号档 (Int64, 盘上是保序编码): 宽度同为 8 字节,
  542. // 但字节布局不同, 直接拿去当行主键会让主键树错序、当时间键会让时间轴错位。这里把它钉回
  543. // uint64; 行号本身非负, 值不会因此改变。1/2/4 字节的整数列装不下行号, 返回 false,
  544. // 由调用方补 _id 列。
  545. func rowStorePKType(t types.FieldType) (types.FieldType, bool) {
  546. switch t {
  547. case types.FieldType_Uint64:
  548. return types.FieldType_Uint64, true
  549. case types.FieldType_Int64:
  550. return types.FieldType_Uint64, true
  551. }
  552. return 0, false
  553. }
  554. // pinRowStorePK 就地把指定列钉到行存储主键要求的档位; 列不存在或宽度不够时返回 false。
  555. func pinRowStorePK(s *TableSchema, col string) bool {
  556. i, ok := s.ColumnIdx(col)
  557. if !ok {
  558. return false
  559. }
  560. ft, ok := rowStorePKType(s.Columns[i].Type)
  561. if !ok {
  562. return false
  563. }
  564. s.Columns[i].Type = ft
  565. return true
  566. }
  567. // freeColumnName 找一个没被占用的列名 (优先用 want)。
  568. func freeColumnName(s *TableSchema, want string) string {
  569. if !s.HasColumn(want) {
  570. return want
  571. }
  572. for i := 2; ; i++ {
  573. cand := fmt.Sprintf("%s_%d", want, i)
  574. if !s.HasColumn(cand) {
  575. return cand
  576. }
  577. }
  578. }
  579. // vrtxTableName 把外部表名收拾成 VRTX 能接受的样子 (名字里有空格/奇怪字符时)。
  580. func vrtxTableName(name string) string {
  581. return strings.TrimSpace(name)
  582. }