types.go 26 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654
  1. // Package migrate 把 VRTX 库与 MySQL / SQLite 库互相转换 (结构 + 数据 + 索引 + 主键)。
  2. //
  3. // 设计要点: 两端各实现 Source (读结构 + 流式读行) 与 Sink (建结构 + 批量写行),
  4. // 中间统一表示是「VRTX 的 FieldType」+ 一组按列序排列的 interface{} 值。于是任意
  5. // 方向都走同一条代码路径:
  6. //
  7. // VRTX → SQLite / MySQL 导出
  8. // SQLite / MySQL → VRTX 导入
  9. // VRTX → VRTX 顺带就是一个整库拷贝 / 压实工具
  10. //
  11. // 保真: 导出到 SQL 时会把原始结构写进一张 __vrtx_meta 表 (Options.EmbedMeta 可关)。
  12. // 带着它回来, 那些 SQL 方言表达不了的细节 —— 定长串的档位 (16/32/64/128/256 字节)、
  13. // 索引类型 (Hash/BTree/时序)、主键索引挂在哪个列上 —— 都能原样还原; 没有它则按 SQL
  14. // 列类型就近推断 (给外部库用)。
  15. package migrate
  16. import (
  17. "git.ycl.pub/yl32940/TsDB2/DB/core/page"
  18. "git.ycl.pub/yl32940/TsDB2/DB/core/types"
  19. "fmt"
  20. "log"
  21. "strings"
  22. "time"
  23. )
  24. // Kind 一端库的种类。
  25. type Kind string
  26. const (
  27. KindVRTX Kind = "vrtx"
  28. KindMySQL Kind = "mysql"
  29. KindSQLite Kind = "sqlite"
  30. )
  31. // Column 列定义 (中间表示)。
  32. //
  33. // 统一用 VRTX 的 FieldType 而不是中性类型: 它已经覆盖了这里需要的全部宽度档位
  34. // (1/2/4/8/16/32/64/128/256 字节定长 + 浮点 + 变长), SQL 侧的 varchar(n)/int/bigint
  35. // 都能无损映射进来。反向映射只能"就近取档", 这也是 __vrtx_meta 存在的理由。
  36. type Column struct {
  37. Name string
  38. Type types.FieldType
  39. // Param 列级类型参数, 按类型解释 (与 VRTX 的 Flag 低 5 位同源, 见 types.FieldSize):
  40. //
  41. // - Decimal: 小数位数 scale。它的保序编码把 scale 写进了行内字节, "同一列 scale
  42. // 必须一致"是它能被 memcmp 比较的前提 —— 所以 scale 必须跟着列走
  43. // (__vrtx_meta 里也要存), 不能只活在声明文本里, 否则导出再导入就丢了小数位。
  44. // - FieldType_String (与变长): 行内宽度码。maxlen 声明出来的列真实
  45. // 宽度由它决定, GetSize 只给类型默认档。
  46. // - 其余类型恒为 0 —— 由 ValidateColumnParam 强制。
  47. Param byte
  48. // TimeKind 这列在源 SQL 里的时间语义: "datetime" / "date" / "timestamp", 空即普通列。
  49. //
  50. // 为什么还要单独记: 时间列在 VRTX 侧落成引擎的专用类型 FieldType_DateTime
  51. // (行内 8 字节 Unix 毫秒, 与 gormvrtx / SQL driver 同一约定), 盘上只能看出
  52. // "这是时间", 分不出当初声明的是 DATETIME / DATE 还是 TIMESTAMP —— 带着这个
  53. // 标记, 导回 SQL 时才能还原成原来的声明, 而不是一律 DATETIME。
  54. //
  55. // 它还管着【旧库兼容】的半条命: 更早的实现把 SQL 时间列迁成 Uint64 存 Unix
  56. // 纳秒, 那些存量表的 meta 里同样打着这个标记。读路径按列的 FieldType 分岔
  57. // (DateTime → 毫秒, Uint64+TimeKind → 旧纳秒数据), 两条路都在 (见 value.go)。
  58. //
  59. // 标记放在哪, 以及它管不到哪一段: 目前只写进 SQL 侧的保真表 __vrtx_meta,
  60. // VRTX 的 schema 里【没有地方放它】——
  61. //
  62. // - types.FieldInfo 的 Flag 一个字节已经排满: 低 5 位是宽度码 / scale,
  63. // 5/6/7 位分别是 Unique / Nullable / Deleted;
  64. // - 借 Param 也不行, ValidateColumnParam 明确禁止非 Decimal、非字符串类型
  65. // 带列级参数 ("uint64 的 Param 恒为 0"), 改这条就是把一个格式不变量拆了。
  66. //
  67. // 所以: SQL → VRTX → SQL 能还原成时间列 (标记跟着 __vrtx_meta 往返),
  68. // 但【从 VRTX 直接导出】拿不到标记 —— DateTime 列会被声明成 DATETIME(6)
  69. // (sqlTypeName 的兜底还原), DATE / TIMESTAMP 当初的细分丢失。要补上这一段,
  70. // 得给 VRTX 侧一个正经的存放位置 (加 FieldType / 内部元数据表),
  71. // 那是一次格式版本决策, 不适合顺手做。
  72. //
  73. // 精度说明: DateTime 列是毫秒精度, formatSQLTime 恰好写到毫秒, 转时间文本
  74. // 无损; 只有存量 Uint64+纳秒的旧表会按微秒截断导出 (MySQL DATETIME(6) 的上限)
  75. // —— 那是旧盘面自身的历史包袱, 不是现在的损失。
  76. TimeKind string
  77. // Nullable 该列允许 NULL (VRTX 的 FlagNullable ⇄ SQL 的可空声明)。
  78. //
  79. // 两边都必须显式带上: 导出时可空列不能写 NOT NULL (否则 NULL 值根本插不进去,
  80. // 只能被改成 0/空串), 导入时 SQL 的可空性也要落到 VRTX 的 schema 上 ——
  81. // __vrtx_meta 里同样存一份, 因为 SQLite 的 PRAGMA 看得见 notnull 而
  82. // SQL 的 TEXT 保真表看不出"整数列能不能为空"。
  83. Nullable bool
  84. // Unique 该列声明了唯一约束 (VRTX 的 FlagUnique ⇄ SQL 的列级 UNIQUE)。
  85. //
  86. // 只是【声明】, 真正拦重复的是唯一索引 (见 IndexSchema.Unique) —— 与 VRTX 的
  87. // 分工一致: 列位告诉读者/工具"这列的意图", 约束由索引执行。往返时两个标记
  88. // 都要带, 否则导入 VRTX 后列上的声明就丢了 (索引还在, 但 ORM 反推时会看不出
  89. // 这列本该唯一)。
  90. //
  91. // 导出到 SQL 时【不】把它写成列级 UNIQUE: SQL 的 UNIQUE 会真的生成一条唯一
  92. // 索引, 也就是一个会拦写入的约束, 而这里的声明在 VRTX 侧什么都不拦 —— 写了
  93. // 等于凭空造约束, 再导回来还会多出一条索引。它只进 __vrtx_meta 保真表
  94. // (见 metaCol.Uniq), 纯导出时确实不落 DDL, 这是有意为之。
  95. Unique bool
  96. }
  97. // Flags 列上的可空/唯一声明, 拼成给人看的一小段后缀 (Describe / 日志用)。
  98. func (c Column) Flags() string {
  99. var s string
  100. if c.Nullable {
  101. s += " NULL"
  102. }
  103. if c.Unique {
  104. s += " UNIQUE"
  105. }
  106. return s
  107. }
  108. // IsTime 这列是否承载 SQL 的时间语义。
  109. func (c Column) IsTime() bool { return c.TimeKind != "" }
  110. // Size 列在 VRTX 行内的字节宽度 (变长列为堆引用宽度)。
  111. //
  112. // 走 FieldSize(c.Type, c.Param): 字符串列的真实宽度由列级宽度码决定, GetSize 只给
  113. // 类型默认档 (对 maxlen 声明出来的列会算窄)。
  114. func (c Column) Size() int { return int(types.FieldSize(c.Type, c.Param)) }
  115. // IsDynamic 变长列 (超过 MaxInlineColumnWidth 的字符串 / 二进制)。
  116. func (c Column) IsDynamic() bool { return types.IsDynamic(c.Type) }
  117. func (c Column) String() string {
  118. return fmt.Sprintf("%s %s%s", c.Name, columnTypeLabel(c), c.Flags())
  119. }
  120. // columnTypeLabel 列类型的可读名: Decimal 带上小数位 (decimal(2)), 其余用 TypeName。
  121. func columnTypeLabel(c Column) string {
  122. if c.Type == types.FieldType_Decimal {
  123. return fmt.Sprintf("decimal(%d)", c.Param)
  124. }
  125. return TypeName(c.Type)
  126. }
  127. // TypeName VRTX 类型的可读名字 (与 types 注册表里的名字同源; 写进 __vrtx_meta)。
  128. func TypeName(t types.FieldType) string {
  129. switch t {
  130. case types.FieldType_Bool:
  131. return "bool"
  132. case types.FieldType_Uint8:
  133. return "uint8"
  134. case types.FieldType_Int8:
  135. return "int8"
  136. case types.FieldType_Uint16:
  137. return "uint16"
  138. case types.FieldType_Int16:
  139. return "int16"
  140. case types.FieldType_Year:
  141. return "year"
  142. case types.FieldType_Uint32:
  143. return "uint32"
  144. case types.FieldType_Int32:
  145. return "int32"
  146. case types.FieldType_Float32:
  147. return "float32"
  148. case types.FieldType_Uint64:
  149. return "uint64"
  150. case types.FieldType_DateTime:
  151. return "datetime"
  152. case types.FieldType_Int64:
  153. return "int64"
  154. case types.FieldType_Float64:
  155. return "float64"
  156. case types.FieldType_Int128:
  157. return "int128"
  158. case types.FieldType_UUID:
  159. return "uuid"
  160. case types.FieldType_Decimal:
  161. return "decimal"
  162. case types.FieldType_String:
  163. return "string"
  164. case types.FieldType_Dynamicbyte:
  165. return "dynamic"
  166. case types.FieldType_Blob:
  167. return "blob"
  168. }
  169. return fmt.Sprintf("unknown(%d)", byte(t))
  170. }
  171. // IndexSchema 索引定义。
  172. //
  173. // Type 直接用引擎的 page.IndexType, 不另造一套枚举: 跨库往返时 Hash / BTree / 时序
  174. // 的差异本来就要记 (SQL 侧只分"索引"), 记在 __vrtx_meta 里回来还原, 比自定义枚举
  175. // 少一层翻译。
  176. type IndexSchema struct {
  177. Name string
  178. Type page.IndexType
  179. Columns []string
  180. // Unique 唯一索引: 同一索引键只允许一行 (NULL 免除 —— 任一分量为 NULL 的
  181. // 行不进该索引, 与 core.IndexFlagUnique 的语义完全一致)。
  182. //
  183. // SQL 侧只有"唯一索引"这一个结构对应物, 不给它, 把 VRTX 的唯一约束导出到
  184. // SQL 就只剩一条普通索引: 重复值随便进, 再导回 VRTX 时约束已经悄悄没了 ——
  185. // 结构变了却没人报错, 所以这个标记必须往返 (__vrtx_meta 里也存一份)。
  186. Unique bool
  187. }
  188. // IsRowStore 是否承载行数据本身 (VRTX 的主键索引 / 时序索引), 即 SQL 侧的主键。
  189. func (i IndexSchema) IsRowStore() bool {
  190. return i.Type == page.IndexTypePrimary || i.Type == page.IndexTypeTime
  191. }
  192. // SameColumns 分量列表是否一致 (顺序敏感)。
  193. func (i IndexSchema) SameColumns(cols []string) bool {
  194. if len(i.Columns) != len(cols) {
  195. return false
  196. }
  197. for k := range cols {
  198. if i.Columns[k] != cols[k] {
  199. return false
  200. }
  201. }
  202. return true
  203. }
  204. func (i IndexSchema) String() string {
  205. kind := IndexTypeName(i.Type)
  206. if i.Unique {
  207. kind += " unique"
  208. }
  209. return fmt.Sprintf("%s(%s) [%s]", i.Name, strings.Join(i.Columns, ","), kind)
  210. }
  211. // IndexTypeName 索引类型的可读名字 (与 page.IndexType 对齐)。
  212. func IndexTypeName(t page.IndexType) string {
  213. switch t {
  214. case page.IndexTypeTime:
  215. return "time"
  216. case page.IndexTypePrimary:
  217. return "primary"
  218. case page.IndexTypeStr:
  219. return "str"
  220. case page.IndexTypeCompose:
  221. return "compose"
  222. case page.IndexTypeBTree:
  223. return "btree"
  224. case page.IndexTypeHash:
  225. return "hash"
  226. }
  227. return fmt.Sprintf("unknown(%d)", byte(t))
  228. }
  229. // TableSchema 一张表的完整结构。
  230. type TableSchema struct {
  231. Name string
  232. Columns []Column
  233. // PKColumn 主键列名。它同时是"行号列": VRTX 的行数据以 pkID 定位, 这个列就是
  234. // pkID 的镜像, 导出到 SQL 时充当 PRIMARY KEY。空表示源表没有主键 (SQL 侧无
  235. // PRIMARY KEY), 目标 VRTX 表只能由引擎分配行号。
  236. PKColumn string
  237. // Indexes 不含主键索引本身 (主键由 PKColumn + PKIndexType 表达)。
  238. Indexes []IndexSchema
  239. // PKIndexType 行存储索引在 VRTX 里的类型: primary (普通自增主键) 或 time (时序)。
  240. // 从 SQL 导入时按 primary 处理, 除非 __vrtx_meta 里记着 time。
  241. PKIndexType page.IndexType
  242. // RowStoreColumns 行存储索引的分量列 (VRTX 侧建索引时用)。
  243. //
  244. // 为空时就是 [PKColumn]; 只有时序索引会显式给出: 它的键是时间列, 既不能当
  245. // SQL 主键也不能承载行号, 于是 PKColumn 留空 (SQL 侧不设主键), 行存储索引
  246. // 靠这个字段记住要建在哪些列上。
  247. RowStoreColumns []string
  248. // PKIndexName 行存储索引名 (VRTX 侧默认 pk_<表名>)。SQL 侧主键索引固定叫 PRIMARY,
  249. // 这个名字只在"往 VRTX 建表"时用得上, 靠 __vrtx_meta 带回来。
  250. PKIndexName string
  251. // Rows 源表行数估计, 仅供计划展示 (以真实扫描计数为准)。
  252. Rows uint64
  253. // Synthetic 记录"源表里没有、由迁移过程补出来的列"。
  254. // 读端扫描时必须跳过它们 (SQL 上 SELECT 一个不存在的列会直接报错),
  255. // 写端则要在写入时给它们填值 (填行号)。
  256. Synthetic map[string]bool
  257. // SkippedIndexes 目标端建不出来的索引, 形如 "idx_ctime: 键列 ctime 是动态列"。
  258. // 由 Sink 建表时填: 源库里合法的索引在目标端可能无处安放 (VRTX 的行内只存
  259. // 动态列的堆引用, 不能拿它当索引键)。静默丢索引等于悄悄改结构, 所以记下来
  260. // 报给用户 —— 而不是让整张表跟着失败。
  261. SkippedIndexes []string
  262. // CappedColumns 源库里声明了超过 VRTX 单值上限 (types.MaxStringSize) 的列,
  263. // 形如 "Note: 声明容量 60000 字节超过单值上限 16360 字节 (...)"。
  264. //
  265. // 由读端填 (只有它拿得到 SQL 的声明原文, 见 capacityNote): 声明 varchar(60000)
  266. // 的列落到 VRTX 只能是动态列, 而动态列装不下那么多 —— 声明兑现不了。这与
  267. // SkippedIndexes 是同一条原则: 静默把容量改小等于悄悄改结构, 必须报出来,
  268. // 但也不该让整张表的数据陪葬 (超限只是"这一列的上限比声明的小", 大部分数据照样迁得动)。
  269. CappedColumns []string
  270. // DegradedColumns 源库里能建、但到 VRTX 只能【降级表达】因而语义变了的列,
  271. // 形如 "Amount: DECIMAL(65,30) 无法精确表达 (精度 65 超过上限 38), 已退化成 float64"。
  272. //
  273. // 与 CappedColumns 的分工: 那个是"类型不变、上限变小"; 这个是"类型都换了"。
  274. // 同样由读端填 (只有它拿得到 SQL 的声明原文, 见 degradeNote), 同样只报不拒 ——
  275. // 降级后的列照样装得下数据, 只是比较/聚合语义与原库不同 (定点 vs 浮点)。
  276. DegradedColumns []string
  277. // PKNote 源表有主键、但这个主键 VRTX 承载不了时的说明 (目前只有复合主键)。
  278. //
  279. // 读端在放弃 PKColumn 时必须填: "主键是哪几列"这个信息只存在于读端, 到了下游就只剩
  280. // 一个空的 PKColumn —— 再报"源表没有主键"就是假消息 (源表明明有主键, 只是多列)。
  281. PKNote string
  282. // SourcePKColumn 源库的主键列名 —— 仅在它【承载不了行号】、被降级成普通列时填
  283. // (见 run.go 的降级分支: MySQL 的 int(11) 主键只有 4 字节, VRTX 的行号是 8 字节)。
  284. //
  285. // 为什么非要单独存: 那种情况下 PKColumn 已经换成补出来的 _id, 而 _id 是引擎分配的行号,
  286. // 跟源库的行对不上号 —— 于是写失败时只能报"批内第 N 行", 用户还得自己回源库猜。原主键
  287. // 列虽然降成了普通列, 但它的值照样在行数据里, 记下这个列名就能把源库真认得的
  288. // "id = 12345" 报出来。降级说明 (DegradedColumns) 是给人看的散文, 这个字段是给代码用的。
  289. SourcePKColumn string
  290. }
  291. // MarkSynthetic 标记一列是迁移过程补出来的。
  292. func (s *TableSchema) MarkSynthetic(name string) {
  293. if s.Synthetic == nil {
  294. s.Synthetic = map[string]bool{}
  295. }
  296. s.Synthetic[name] = true
  297. }
  298. // IsSynthetic 该列是否为迁移过程补出来的。
  299. func (s *TableSchema) IsSynthetic(name string) bool {
  300. return s.Synthetic[name]
  301. }
  302. // ColumnIdx 按列名找下标 (大小写敏感 —— VRTX 列名就是 Go 字段名, 不做转换)。
  303. func (s *TableSchema) ColumnIdx(name string) (int, bool) {
  304. for i := range s.Columns {
  305. if s.Columns[i].Name == name {
  306. return i, true
  307. }
  308. }
  309. return -1, false
  310. }
  311. // HasColumn 列是否存在。
  312. func (s *TableSchema) HasColumn(name string) bool {
  313. _, ok := s.ColumnIdx(name)
  314. return ok
  315. }
  316. // Clone 深拷贝 (迁移过程中会往 schema 里补列, 不能污染调用方那份)。
  317. func (s *TableSchema) Clone() *TableSchema {
  318. c := *s
  319. c.Columns = append([]Column(nil), s.Columns...)
  320. c.RowStoreColumns = append([]string(nil), s.RowStoreColumns...)
  321. c.Indexes = make([]IndexSchema, len(s.Indexes))
  322. for i, idx := range s.Indexes {
  323. idx.Columns = append([]string(nil), idx.Columns...)
  324. c.Indexes[i] = idx
  325. }
  326. if s.Synthetic != nil {
  327. c.Synthetic = make(map[string]bool, len(s.Synthetic))
  328. for k, v := range s.Synthetic {
  329. c.Synthetic[k] = v
  330. }
  331. }
  332. c.SkippedIndexes = append([]string(nil), s.SkippedIndexes...)
  333. c.CappedColumns = append([]string(nil), s.CappedColumns...)
  334. c.DegradedColumns = append([]string(nil), s.DegradedColumns...)
  335. c.PKNote = s.PKNote
  336. return &c
  337. }
  338. // RowStore 行存储索引要建的列: 未显式指定时就是 [PKColumn]。
  339. func (s *TableSchema) RowStore() []string {
  340. if len(s.RowStoreColumns) > 0 {
  341. return s.RowStoreColumns
  342. }
  343. if s.PKColumn == "" {
  344. return nil
  345. }
  346. return []string{s.PKColumn}
  347. }
  348. // Describe 多行结构摘要 (CLI -list / 日志用)。
  349. func (s *TableSchema) Describe() string {
  350. var b strings.Builder
  351. fmt.Fprintf(&b, "表 %s (%d 列, 主键 %s[%s])\n", s.Name, len(s.Columns), orNone(s.PKColumn), IndexTypeName(s.PKIndexType))
  352. for _, c := range s.Columns {
  353. mark := " "
  354. if c.Name == s.PKColumn {
  355. mark = "PK"
  356. }
  357. fmt.Fprintf(&b, " %s %-28s %s%s\n", mark, c.Name, columnTypeLabel(c), c.Flags())
  358. }
  359. for _, idx := range s.Indexes {
  360. fmt.Fprintf(&b, " IDX %-27s %s\n", idx.String(), "")
  361. }
  362. return b.String()
  363. }
  364. func orNone(s string) string {
  365. if s == "" {
  366. return "<无>"
  367. }
  368. return s
  369. }
  370. // Row 一行数据。Values 与 TableSchema.Columns 同序同长。
  371. //
  372. // PK 是"行主键号": VRTX 侧就是 pkID, SQL 侧是主键列的整数值。HasPK=false 表示这行
  373. // 没有可用主键 (源表无主键, 或主键列为空), 目标侧自行分配。
  374. type Row struct {
  375. PK uint64
  376. HasPK bool
  377. Values []any
  378. }
  379. // Source 读端。
  380. type Source interface {
  381. Kind() Kind
  382. // Identity 唯一标识这个库 (路径 / DSN), 用于拦住 src==dst 的自杀式迁移。
  383. Identity() string
  384. // Tables 列出可迁移的表 (不含内部表)。
  385. Tables() ([]string, error)
  386. // Schema 读一张表的结构。
  387. Schema(table string) (*TableSchema, error)
  388. // Scan 流式扫描整表, 每行回调一次 (Row.Values 与 s.Columns 同序)。
  389. //
  390. // 用回调而不是 Next() 迭代器: core 的扫描本身就是回调式、无法中途暂停,
  391. // 而 SQL 侧 rows.Next() 也天然适配 —— 编排器在回调里攒够一批就写, 内存
  392. // 占用与表的大小无关。回调返回 error 会立刻中止扫描并原样返回。
  393. Scan(table string, s *TableSchema, fn func(Row) error) error
  394. Close() error
  395. }
  396. // Sink 写端。
  397. type Sink interface {
  398. Kind() Kind
  399. Identity() string
  400. // Exists 目标表是否已存在。
  401. Exists(table string) (bool, error)
  402. // CreateTable 建表 + 建索引 (含主键)。recreate=true 时先 DROP。
  403. CreateTable(s *TableSchema, recreate bool) error
  404. // Open 准备写入 (建 prepared statement / 开事务), 调用方负责 Close。
  405. Open(table string, s *TableSchema) (RowWriter, error)
  406. Close() error
  407. }
  408. // RowWriter 批量写入器。
  409. type RowWriter interface {
  410. // Write 写一批行 (行数由 Options.Batch 控制); 实现内部可自行分事务提交。
  411. Write(rows []Row) error
  412. // Close 落盘并结束 (VRTX 侧是 Commit, SQL 侧是提交事务)。
  413. Close() error
  414. }
  415. // Discarder 写入器的可选接口: 放弃尚未提交的内容 (出错路径专用)。
  416. //
  417. // 为什么需要它: Close 是【成功语义】(SQL 侧 flush 剩余缓冲 + Commit)。扫描中途失败
  418. // 或 flush 出错时再用 Close 收尾, 等于把半截数据当成功提交 —— 调用方拿到错误、以为
  419. // 这张表没进去, 库里却已经躺了半张表。实现 Discarder 后, 出错路径会走 Discard
  420. // (SQL 侧 Rollback), 缓冲区里那批行原样丢弃。
  421. //
  422. // 未实现的写入器 (如 VRTX) 仍退回 Close: 它的 Close 不提交 (提交由 Committer 在
  423. // 整表成功后单独调用), 所以没有这个问题。
  424. type Discarder interface {
  425. // Discard 丢弃当前事务与缓冲区里尚未提交的行。
  426. Discard() error
  427. }
  428. // Options 迁移选项。
  429. //
  430. // 用法约定: 从 DefaultOptions() 出发改自己关心的字段。
  431. //
  432. // opts := migrate.DefaultOptions()
  433. // opts.Recreate = true
  434. // migrate.Run(src, dst, opts)
  435. //
  436. // Run(src, dst, nil) 等价于传 DefaultOptions()。直接传裸 &Options{} 是"激进档":
  437. // 不建表、不写保真表、不做值校验 —— 给已经有目标表的批处理脚本用。
  438. type Options struct {
  439. // Tables 只迁这些表; 空 = 源库全部表。
  440. Tables []string
  441. // Create 目标表不存在时建表。
  442. Create bool
  443. // Recreate 目标表已存在时先 DROP 再建 (false: 复用已有表并追加数据)。
  444. Recreate bool
  445. // Batch 一次写入的行数, 也是 SQL 侧每次提交的行数 (<=0 取 500)。
  446. Batch int
  447. // EmbedMeta 向 SQL 侧写 __vrtx_meta 保真表 (回来时能还原定长档位与索引类型)。
  448. EmbedMeta bool
  449. // AutoPK 源表没有整数主键时, 让目标 VRTX 表按引擎自增号分配 (false: 报错)。
  450. AutoPK bool
  451. // ForceInspect VRTX 端打开时启用抢救模式: 头页校验失败仍只读打开 (写禁用)。
  452. // 只影响 vrtx: 源/目标, 供头页损坏后抢救数据。
  453. ForceInspect bool
  454. // Limit 每张表最多迁移多少行 (0 = 不限, 便于先试跑)。
  455. Limit uint64
  456. // StrictValue 值装不进目标列时报错 (false: 按 VRTX 语义静默截断, 会丢数据)。
  457. StrictValue bool
  458. // TruncateOversize 动态列的值超过【单值上限】(types.MaxStringSize, 即引擎的
  459. // core.MaxBlobSize, 当前 16360 字节) 时截到上限继续写, 并逐行报出截了什么。
  460. // false (默认) 时报错, 不丢一个字节。
  461. //
  462. // 与 StrictValue 的分工: StrictValue 管"值装不进这一列的宽度"(定长列声明多宽就多宽,
  463. // 截断后仍合原列的声明); 这个管"值超过引擎单个动态值能装的上限"。后者是引擎当下的
  464. // 能力边界: Dynamicbyte 的内容整块落进一个堆页, 多页分块存储尚未实现 —— 也就是说
  465. // 这跟源库怎么声明该列无关, 一个本来合法的值照样会卡在这里 (见 capacityNote 里那条
  466. // "容量封顶"的提示)。
  467. //
  468. // 只管 Dynamicbyte: blob 列 (FieldType_Blob) 的内容落在可跨页的链上 (单值约 4GB),
  469. // 不受这里的单堆页上限约束, 也不需要这个开关兜底。
  470. //
  471. // 这是【明确可选的数据丢失】, 所以默认关着: 打开之后每截一行报一行 (表 / 行标识 /
  472. // 列名 / 原长 / 截后长), 用户据此回源库补那一行的内容, 而不用满表翻。文本列会退到
  473. // UTF-8 字符边界再截, 不留半个汉字。
  474. TruncateOversize bool
  475. // DryRun 只读结构不搬数据。
  476. DryRun bool
  477. // ContinueOnError 单张表迁移失败时跳过它、继续迁其余表 (false: 第一个错误就整体中止)。
  478. // 失败的表会记进 TableReport.Error, 全部跑完后 Run 仍返回一个汇总错误 ——
  479. // 继续不等于吞掉错误。
  480. ContinueOnError bool
  481. // SchemaOnly 只同步结构 (建表 + 建索引), 不搬数据 —— 两阶段迁移的第一步。
  482. SchemaOnly bool
  483. // EachRow 每读到一行就回调一次 (表名, 该表累计行数, 该行)。
  484. // 给"看得见每一行在流动"的界面用; 开销随行数线性上涨, 大表慎用, nil 即关闭。
  485. EachRow func(table string, n uint64, row Row)
  486. // Verbose 打印每张表的进度。
  487. Verbose bool
  488. // Log 日志输出; nil 时用 log.Printf。
  489. Log func(format string, args ...any)
  490. // Progress 迁移中途的进度回调 (表名, 已写行数, 已用时), 按秒级节流地调。
  491. // 界面靠它原地刷一行 —— 否则百万行的表只有"开始"和"结束"两条日志,
  492. // 看起来和死锁没区别。nil 时退回 Verbose 文本 (CLI -v)。
  493. Progress func(table string, rows uint64, elapsed time.Duration)
  494. }
  495. // DefaultOptions 默认档: 建表 + 全表 + 批 500 + 写保真表 + 严格校验。
  496. func DefaultOptions() *Options {
  497. return &Options{
  498. Create: true,
  499. EmbedMeta: true,
  500. StrictValue: true,
  501. Batch: 500,
  502. }
  503. }
  504. // normalized 把 nil 当成 DefaultOptions() (Options 里的开关都是 bool, 没法在事后
  505. // 区分"没设过"和"显式设成 false", 所以只能在这一层兜底), 并补上 Batch 这类数值型
  506. // 默认值 —— 调用方显式设过的开关一律不动。
  507. func (o *Options) normalized() *Options {
  508. if o == nil {
  509. return DefaultOptions()
  510. }
  511. out := *o
  512. if out.Batch <= 0 {
  513. out.Batch = 500
  514. }
  515. return &out
  516. }
  517. // wants 表是否在本次迁移范围内。
  518. func (o *Options) wants(table string) bool {
  519. if len(o.Tables) == 0 {
  520. return true
  521. }
  522. for _, t := range o.Tables {
  523. if t == table {
  524. return true
  525. }
  526. }
  527. return false
  528. }
  529. // Wants 表是否在本次迁移范围内 (与内部同一套判断; 供 CLI / 调用方预筛表用)。
  530. func (o *Options) Wants(table string) bool { return o.wants(table) }
  531. func (o *Options) logf(format string, args ...any) {
  532. if o.Log != nil {
  533. o.Log(format, args...)
  534. return
  535. }
  536. log.Printf(format, args...)
  537. }
  538. // vlogf 只在 Verbose 打开时说话 (进度类信息, 默认安静)。
  539. func (o *Options) vlogf(format string, args ...any) {
  540. if o.Verbose {
  541. o.logf(format, args...)
  542. }
  543. }
  544. // progress 报一次"这张表迁到哪儿了"。结构化回调优先 —— 界面要原地刷新一行,
  545. // 不该去解析文本; 没有回调时退化成 Verbose 文本, CLI 开着 -v 照样看得见。
  546. func (o *Options) progress(table string, rows uint64, elapsed time.Duration) {
  547. if o.Progress != nil {
  548. o.Progress(table, rows, elapsed)
  549. return
  550. }
  551. o.vlogf("[migrate] 表 %s: 已写 %d 行 (%s)", table, rows, elapsed.Round(time.Millisecond))
  552. }
  553. // TableReport 单表迁移结果。
  554. type TableReport struct {
  555. Table string
  556. Rows uint64
  557. Indexes int
  558. Columns int
  559. Elapsed time.Duration
  560. Skipped bool
  561. Note string
  562. // Error 非空表示这张表失败了 (无论这趟是跳过它继续、还是就此中止, 原因都在)。
  563. // 此时 Rows 是失败前已经写进去的行数。
  564. Error string
  565. }
  566. // Report 整体结果。
  567. type Report struct {
  568. From Kind
  569. To Kind
  570. Tables []TableReport
  571. Rows uint64
  572. Elapsed time.Duration
  573. }
  574. // Line 单行摘要 (CLI 输出)。
  575. func (r *Report) Line(tr TableReport) string {
  576. s := fmt.Sprintf("%-24s %8d 行 %3d 列 %2d 索引 %8s", tr.Table, tr.Rows, tr.Columns, tr.Indexes, tr.Elapsed.Round(time.Millisecond))
  577. if tr.Skipped {
  578. s += " [跳过]"
  579. }
  580. if tr.Error != "" {
  581. s += " [失败] " + tr.Error
  582. }
  583. if tr.Note != "" {
  584. s += " " + tr.Note
  585. }
  586. return s
  587. }
  588. // Failed 报告里失败的表数 (用它快速判断这趟是否干净)。
  589. func (r *Report) Failed() int {
  590. n := 0
  591. for _, tr := range r.Tables {
  592. if tr.Error != "" {
  593. n++
  594. }
  595. }
  596. return n
  597. }
  598. // String 整份报告的文本形式 (逐表一行 + 汇总)。
  599. func (r *Report) String() string {
  600. var b strings.Builder
  601. for _, tr := range r.Tables {
  602. b.WriteString(r.Line(tr))
  603. b.WriteByte('\n')
  604. }
  605. fmt.Fprintf(&b, "%s → %s: %d 张表 / %d 行, 用时 %s",
  606. r.From, r.To, len(r.Tables), r.Rows, r.Elapsed.Round(time.Millisecond))
  607. if n := r.Failed(); n > 0 {
  608. fmt.Fprintf(&b, " (%d 张失败)", n)
  609. }
  610. return b.String()
  611. }