vrtx.go 23 KB

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