types.go 27 KB

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