types.go 23 KB

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