sql.go 20 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672
  1. package migrate
  2. import (
  3. "git.ycl.pub/yl32940/TsDB2/DB/core/page"
  4. "git.ycl.pub/yl32940/TsDB2/DB/core/types"
  5. "database/sql"
  6. "encoding/json"
  7. "errors"
  8. "fmt"
  9. "sort"
  10. "strings"
  11. )
  12. // MetaTable 保真表的名字。它在目标库里只是一张普通表, 别的工具读到也不影响;
  13. // 删掉它, 迁移依然能跑 (类型/索引类型按 SQL 声明就近推断), 只是定长档位与
  14. // Hash/时序这些 SQL 表达不了的细节会丢。
  15. const MetaTable = "__vrtx_meta"
  16. // dialect SQL 方言。MySQL 与 SQLite 只有这些差异, 读写流程完全共用。
  17. // nullClause 建表时的可空声明。
  18. //
  19. // 只有"不可空"才写 NOT NULL —— 可空列不能靠"什么都不写"来表达吗? 能, 但显式
  20. // 写一个 NULL 更好: 生成出来的 DDL 一眼能看出这列是【有意可空】, 而不是从
  21. // VRTX 导过来时被漏掉了可空声明 (两者的 DDL 长得一样, 只有写了 NULL 才区分得开)。
  22. func nullClause(c Column) string {
  23. if c.Nullable {
  24. return " NULL"
  25. }
  26. return " NOT NULL"
  27. }
  28. // createIndexKind 索引 DDL 的动词前缀。唯一索引在 SQL 侧就是 UNIQUE INDEX,
  29. // 没有别的表达方式 —— 少了它, VRTX 的唯一约束导出到 SQL 就只剩普通索引。
  30. func createIndexKind(idx IndexSchema) string {
  31. if idx.Unique {
  32. return "CREATE UNIQUE INDEX"
  33. }
  34. return "CREATE INDEX"
  35. }
  36. type dialect interface {
  37. kind() Kind
  38. quote(name string) string
  39. // createTable 生成含主键的 CREATE TABLE 语句 (二级索引另建)。
  40. createTable(s *TableSchema) string
  41. dropTable(table string) string
  42. createIndex(table string, idx IndexSchema) string
  43. // columnDDL 单列的列定义 (名字 + 类型)。
  44. columnDDL(c Column) string
  45. // addColumn 给已有表补列的语句 (复用已有表时用)。
  46. addColumn(table string, c Column) string
  47. // listTables 列出用户表 (不含内部表)。
  48. listTables(db *sql.DB) ([]string, error)
  49. // columns 读列 (按表内顺序)、主键列名, 以及"源库的结构 VRTX 承载不了"的说明。
  50. //
  51. // 第三个返回值是读端才看得见的东西 (只有这里拿得到 SQL 的声明原文与主键构成),
  52. // 由 SQL.Schema 落到 TableSchema 的 CappedColumns / DegradedColumns / PKNote 上,
  53. // 最终进迁移报告 —— 见 sourceNotes。
  54. columns(db *sql.DB, table string) ([]Column, string, sourceNotes, error)
  55. // indexes 读二级索引 (不含主键索引)。
  56. indexes(db *sql.DB, table string) ([]IndexSchema, error)
  57. // metaDDL 保真表建表语句。
  58. metaDDL() string
  59. // metaUpsert 写一行保真数据的语句 (两个 ? 占位: 表名, payload)。
  60. metaUpsert() string
  61. // metaSelect 读一张表的保真数据。
  62. metaSelect() string
  63. // maxPlaceholders 单条语句允许的最大占位符数量 (决定多行 INSERT 一次插几行)。
  64. maxPlaceholders() int
  65. }
  66. // SQL 一侧 (MySQL / SQLite 共用)。
  67. type SQL struct {
  68. db *sql.DB
  69. kind Kind
  70. dsn string
  71. dial dialect
  72. opts *Options
  73. tables []string
  74. hasMeta map[string]*metaDTO
  75. metaRead map[string]bool
  76. }
  77. func newSQL(db *sql.DB, kind Kind, dsn string, d dialect, opts *Options) *SQL {
  78. return &SQL{
  79. db: db,
  80. kind: kind,
  81. dsn: dsn,
  82. dial: d,
  83. opts: opts.normalized(),
  84. metaRead: map[string]bool{},
  85. }
  86. }
  87. func (s *SQL) Kind() Kind { return s.kind }
  88. func (s *SQL) Identity() string { return string(s.kind) + ":" + s.dsn }
  89. // DB 暴露底层句柄。
  90. func (s *SQL) DB() *sql.DB { return s.db }
  91. func (s *SQL) Close() error { return s.db.Close() }
  92. // Tables 列出可迁移的表 (不含保真表; 除非调用方明确点名要它)。
  93. func (s *SQL) Tables() ([]string, error) {
  94. if s.tables == nil {
  95. names, err := s.dial.listTables(s.db)
  96. if err != nil {
  97. return nil, err
  98. }
  99. s.tables = names
  100. }
  101. out := make([]string, 0, len(s.tables))
  102. for _, n := range s.tables {
  103. if n == MetaTable && !s.opts.wants(MetaTable) {
  104. continue
  105. }
  106. out = append(out, n)
  107. }
  108. sort.Strings(out)
  109. return out, nil
  110. }
  111. // Schema 读一张表的结构: SQL 声明 → VRTX 类型, 再用保真表 (若有) 覆盖回来。
  112. func (s *SQL) Schema(table string) (*TableSchema, error) {
  113. cols, pk, notes, err := s.dial.columns(s.db, table)
  114. if err != nil {
  115. return nil, fmt.Errorf("读 %s 表 %s 的结构: %w", s.kind, table, err)
  116. }
  117. if len(cols) == 0 {
  118. return nil, fmt.Errorf("%s 表 %s 不存在或没有列", s.kind, table)
  119. }
  120. schema := &TableSchema{
  121. Name: table,
  122. Columns: cols,
  123. PKColumn: pk,
  124. PKIndexType: page.IndexTypePrimary,
  125. CappedColumns: notes.Capped,
  126. DegradedColumns: notes.Degraded,
  127. PKNote: notes.PK,
  128. }
  129. if idx, err := s.dial.indexes(s.db, table); err == nil {
  130. schema.Indexes = idx
  131. } else {
  132. s.opts.logf("[%s] 读表 %s 的索引列表失败 (忽略): %v", s.kind, table, err)
  133. }
  134. // 保真表优先: 列宽档位、索引类型、索引名都按当初导出时记的还原。
  135. if s.opts.EmbedMeta {
  136. if m, ok := s.readMeta(table); ok {
  137. applyMeta(schema, m)
  138. }
  139. }
  140. return schema, nil
  141. }
  142. // Scan 全表扫描 (不带 WHERE, 顺序由存储决定)。
  143. func (s *SQL) Scan(table string, sc *TableSchema, fn func(Row) error) error {
  144. // 补出来的列 (源 SQL 表里并不存在) 不能出现在 SELECT 里。
  145. // 同时记下每个被选中的列在 schema 里的下标 —— 扫描结果要按它放回原位,
  146. // 补列的位置留 nil (由写端自己填)。
  147. sel := make([]int, 0, len(sc.Columns))
  148. for i, c := range sc.Columns {
  149. if sc.IsSynthetic(c.Name) {
  150. continue
  151. }
  152. sel = append(sel, i)
  153. }
  154. if len(sel) == 0 {
  155. return fmt.Errorf("%s 表 %s 没有可读的列", s.kind, table)
  156. }
  157. names := make([]string, len(sel))
  158. for k, i := range sel {
  159. names[k] = s.dial.quote(sc.Columns[i].Name)
  160. }
  161. q := fmt.Sprintf("SELECT %s FROM %s", strings.Join(names, ", "), s.dial.quote(table))
  162. rows, err := s.db.Query(q)
  163. if err != nil {
  164. return fmt.Errorf("查询 %s 表 %s: %w", s.kind, table, err)
  165. }
  166. defer rows.Close()
  167. pkIdx := columnIndex(sc, sc.PKColumn)
  168. dests := make([]any, len(sel))
  169. for i := range dests {
  170. var v any
  171. dests[i] = &v
  172. }
  173. for rows.Next() {
  174. if err := rows.Scan(dests...); err != nil {
  175. return fmt.Errorf("读 %s 表 %s 的行: %w", s.kind, table, err)
  176. }
  177. // 注意: 驱动给的值只在本次迭代内有效 (可能复用底层缓冲), 回调必须当场消费。
  178. vals := make([]any, len(sc.Columns))
  179. for k, i := range sel {
  180. vals[i] = *(dests[k].(*any))
  181. }
  182. row := Row{Values: vals}
  183. if pkIdx >= 0 && vals[pkIdx] != nil {
  184. if n, err := asUint64(vals[pkIdx]); err == nil {
  185. row.PK, row.HasPK = n, true
  186. } else if s.opts.Verbose {
  187. s.opts.logf("[%s] 表 %s 有行的主键 %v 不是整数, "+
  188. "该行将按目标库的自增号插入", s.kind, table, vals[pkIdx])
  189. }
  190. }
  191. if err := fn(row); err != nil {
  192. return err
  193. }
  194. }
  195. if err := rows.Err(); err != nil {
  196. return fmt.Errorf("遍历 %s 表 %s: %w", s.kind, table, err)
  197. }
  198. return nil
  199. }
  200. // Exists 目标表是否已存在。
  201. func (s *SQL) Exists(table string) (bool, error) {
  202. if s.tables == nil {
  203. if _, err := s.Tables(); err != nil {
  204. return false, err
  205. }
  206. }
  207. for _, n := range s.tables {
  208. if n == table {
  209. return true, nil
  210. }
  211. }
  212. return false, nil
  213. }
  214. // CreateTable 建表 + 建二级索引 + 写保真表。
  215. func (s *SQL) CreateTable(sc *TableSchema, recreate bool) error {
  216. exists, err := s.Exists(sc.Name)
  217. if err != nil {
  218. return err
  219. }
  220. if exists && recreate {
  221. if _, err := s.db.Exec(s.dial.dropTable(sc.Name)); err != nil {
  222. return fmt.Errorf("DROP %s 表 %s: %w", s.kind, sc.Name, err)
  223. }
  224. exists = false
  225. s.tables = nil
  226. }
  227. if !exists {
  228. if _, err := s.db.Exec(s.dial.createTable(sc)); err != nil {
  229. return fmt.Errorf("建 %s 表 %s: %w\nSQL: %s", s.kind, sc.Name, err, s.dial.createTable(sc))
  230. }
  231. s.tables = nil
  232. for _, idx := range sc.Indexes {
  233. if idx.IsRowStore() || idx.Name == "" {
  234. continue
  235. }
  236. if _, err := s.db.Exec(s.dial.createIndex(sc.Name, idx)); err != nil {
  237. return fmt.Errorf("建索引 %s: %w\nSQL: %s", idx.Name, err, s.dial.createIndex(sc.Name, idx))
  238. }
  239. }
  240. } else if err := s.ensureColumns(sc); err != nil {
  241. return err
  242. }
  243. if s.opts.EmbedMeta {
  244. if err := s.writeMeta(sc); err != nil {
  245. return err
  246. }
  247. }
  248. return nil
  249. }
  250. // ensureColumns 复用已有表时补上缺的列 (SQL 的 ADD COLUMN), 免得 INSERT 撞上未知列。
  251. // VRTX 侧的 schema 同步是同一套思路: 结构向"源"看齐, 但不删目标已有的列。
  252. func (s *SQL) ensureColumns(sc *TableSchema) error {
  253. // 容量说明只在读整表结构时报一次 (这里是补列的辅助路径, 丢掉即可)。
  254. cols, _, _, err := s.dial.columns(s.db, sc.Name)
  255. if err != nil {
  256. return err
  257. }
  258. have := make(map[string]bool, len(cols))
  259. for _, c := range cols {
  260. have[c.Name] = true
  261. }
  262. missing := 0
  263. for _, c := range sc.Columns {
  264. if have[c.Name] {
  265. continue
  266. }
  267. // 主键列不能后补 (SQL 里加主键是另一件事, 而且要重建表), 缺了直接报错
  268. if c.Name == sc.PKColumn {
  269. return fmt.Errorf("目标表 %s 里没有主键列 %s, 无法追加", sc.Name, c.Name)
  270. }
  271. if _, err := s.db.Exec(s.dial.addColumn(sc.Name, c)); err != nil {
  272. return fmt.Errorf("给表 %s 补列 %s: %w", sc.Name, c.Name, err)
  273. }
  274. missing++
  275. }
  276. if missing > 0 {
  277. s.opts.logf("[%s] 表 %s 补了 %d 个缺列", s.kind, sc.Name, missing)
  278. }
  279. return nil
  280. }
  281. // Open 准备批量写: 拼多行 INSERT (MySQL/SQLite 的往返次数直接决定导入速度)。
  282. func (s *SQL) Open(table string, sc *TableSchema) (RowWriter, error) {
  283. n := len(sc.Columns)
  284. if n == 0 {
  285. return nil, fmt.Errorf("表 %s 没有列", table)
  286. }
  287. rowsPerStmt := s.dial.maxPlaceholders() / n
  288. if rowsPerStmt < 1 {
  289. rowsPerStmt = 1
  290. }
  291. if rowsPerStmt > s.opts.Batch {
  292. rowsPerStmt = s.opts.Batch
  293. }
  294. commitEvery := uint64(s.opts.Batch) * 8
  295. if commitEvery < 1000 {
  296. commitEvery = 1000
  297. }
  298. tx, err := s.db.Begin()
  299. if err != nil {
  300. return nil, fmt.Errorf("开事务: %w", err)
  301. }
  302. return &sqlWriter{
  303. s: s,
  304. table: table,
  305. cols: sc.Columns,
  306. rowsPerStmt: rowsPerStmt,
  307. commitEvery: commitEvery,
  308. tx: tx,
  309. stmts: map[int]*sql.Stmt{},
  310. }, nil
  311. }
  312. type sqlWriter struct {
  313. s *SQL
  314. table string
  315. cols []Column
  316. rowsPerStmt int
  317. commitEvery uint64
  318. tx *sql.Tx
  319. stmts map[int]*sql.Stmt
  320. pending []any
  321. pendingRows int
  322. written uint64
  323. closed bool
  324. }
  325. func (w *sqlWriter) insertSQL(rows, cols int) string {
  326. one := "(" + strings.TrimSuffix(strings.Repeat("?,", cols), ",") + ")"
  327. vals := strings.TrimSuffix(strings.Repeat(one+",", rows), ",")
  328. names := make([]string, cols)
  329. for i, c := range w.cols {
  330. names[i] = w.s.dial.quote(c.Name)
  331. }
  332. return fmt.Sprintf("INSERT INTO %s (%s) VALUES %s",
  333. w.s.dial.quote(w.table), strings.Join(names, ", "), vals)
  334. }
  335. func (w *sqlWriter) Write(rows []Row) error {
  336. for ri := range rows {
  337. r := &rows[ri]
  338. for i, c := range w.cols {
  339. var raw any
  340. if i < len(r.Values) {
  341. raw = r.Values[i]
  342. }
  343. v, err := coerceForSQL(c, raw, w.s.kind == KindMySQL)
  344. if err != nil {
  345. return err
  346. }
  347. w.pending = append(w.pending, v)
  348. }
  349. w.pendingRows++
  350. if w.pendingRows >= w.rowsPerStmt {
  351. if err := w.flush(); err != nil {
  352. return err
  353. }
  354. }
  355. }
  356. return nil
  357. }
  358. func (w *sqlWriter) flush() error {
  359. if w.pendingRows == 0 {
  360. return nil
  361. }
  362. stmt, ok := w.stmts[w.pendingRows]
  363. if !ok {
  364. s, err := w.tx.Prepare(w.insertSQL(w.pendingRows, len(w.cols)))
  365. if err != nil {
  366. return fmt.Errorf("准备 INSERT (表 %s): %w", w.table, err)
  367. }
  368. w.stmts[w.pendingRows] = s
  369. stmt = s
  370. }
  371. if _, err := stmt.Exec(w.pending...); err != nil {
  372. return fmt.Errorf("写入 %s 表 %s: %w", w.s.kind, w.table, err)
  373. }
  374. w.written += uint64(w.pendingRows)
  375. w.pending = w.pending[:0]
  376. w.pendingRows = 0
  377. if w.written >= w.commitEvery {
  378. if err := w.commit(); err != nil {
  379. return err
  380. }
  381. w.written = 0
  382. }
  383. return nil
  384. }
  385. // commit 收掉当前事务并开新的 (长事务会撑爆 undo/日志, 也拖慢 MySQL)。
  386. func (w *sqlWriter) commit() error {
  387. for _, s := range w.stmts {
  388. _ = s.Close()
  389. }
  390. w.stmts = map[int]*sql.Stmt{}
  391. if err := w.tx.Commit(); err != nil {
  392. return fmt.Errorf("提交 %s 表 %s: %w", w.s.kind, w.table, err)
  393. }
  394. tx, err := w.s.db.Begin()
  395. if err != nil {
  396. return fmt.Errorf("重新开事务: %w", err)
  397. }
  398. w.tx = tx
  399. return nil
  400. }
  401. func (w *sqlWriter) Close() error {
  402. if w.closed {
  403. return nil
  404. }
  405. w.closed = true
  406. if err := w.flush(); err != nil {
  407. _ = w.tx.Rollback()
  408. return err
  409. }
  410. for _, s := range w.stmts {
  411. _ = s.Close()
  412. }
  413. if err := w.tx.Commit(); err != nil {
  414. return fmt.Errorf("提交 %s 表 %s: %w", w.s.kind, w.table, err)
  415. }
  416. return nil
  417. }
  418. // Discard 放弃尚未提交的内容 (migrate.Discarder, 出错路径用)。
  419. //
  420. // 与 Close 的区别只有收尾那一步: Close 会把缓冲区 flush 掉再 Commit, Discard 直接
  421. // Rollback 并丢掉缓冲。已经跨过 commitEvery 阈值提交过的批次留在库里 —— 那是分批
  422. // 提交的既有语义 (长事务会撑爆 undo/日志), 能挽回的是当前事务和缓冲区里这一批;
  423. // 报告里的 Rows 会如实说明写到哪儿了。
  424. func (w *sqlWriter) Discard() error {
  425. if w.closed {
  426. return nil
  427. }
  428. w.closed = true
  429. for _, s := range w.stmts {
  430. _ = s.Close()
  431. }
  432. w.stmts = map[int]*sql.Stmt{}
  433. w.pending = w.pending[:0]
  434. w.pendingRows = 0
  435. if err := w.tx.Rollback(); err != nil && !errors.Is(err, sql.ErrTxDone) {
  436. return fmt.Errorf("回滚 %s 表 %s: %w", w.s.kind, w.table, err)
  437. }
  438. return nil
  439. }
  440. // ============================ 保真表 ============================
  441. // metaDTO 保真表里一行 (JSON) 的内容。字段名特意取短: 这张表会跟着数据一起躺在
  442. // 别人的库里, 越不占地方越好; 版本号留着以后改格式时兜底。
  443. type metaDTO struct {
  444. Version int `json:"v"`
  445. PK string `json:"pk,omitempty"`
  446. PKIndexName string `json:"pkn,omitempty"`
  447. PKIndexType byte `json:"pkt,omitempty"`
  448. RowStore []string `json:"rs,omitempty"`
  449. Columns []metaCol `json:"cols"`
  450. Indexes []metaIdx `json:"idx,omitempty"`
  451. }
  452. type metaCol struct {
  453. Name string `json:"n"`
  454. Type byte `json:"t"`
  455. // Param 列级类型参数 (目前是 Decimal 的 scale)。SQL 的 DECIMAL(38,2) 声明虽然
  456. // 带着小数位, 但 SQLite 的 TEXT 列看不出来 —— 存进保真表才能原样还原。
  457. Param byte `json:"p,omitempty"`
  458. // Time 源 SQL 的时间语义 ("datetime" / "date" / "timestamp"), 空即普通列。
  459. // 新迁的表盘上它是 FieldType_DateTime (Unix 毫秒), 靠这个标记导回 SQL 时才还原成
  460. // 当初的 DATETIME / DATE / TIMESTAMP —— 光看 FieldType 只能分出"是时间"。
  461. // 旧库可能是 Uint64+纳秒 (更早的实现), 读路径按列的 FieldType 分岔兼容, 两条路都在。
  462. Time string `json:"tm,omitempty"`
  463. // Null 该列可空。SQL 侧的可空性虽然能从 PRAGMA / information_schema 读出来,
  464. // 但那是【建表时写下的声明】; 我们导出时按 VRTX 的 NULL 位图写 NULL / NOT NULL,
  465. // 带一份在保真表里, 回来就不必依赖各家 SQL 方言的可空推断。
  466. Null bool `json:"nl,omitempty"`
  467. // Uniq 该列声明了唯一约束。列位与索引唯一位是两件事 (声明 vs 约束),
  468. // 各存各的: 只存索引的话, 导回 VRTX 后列上的声明就没了。
  469. Uniq bool `json:"uq,omitempty"`
  470. }
  471. type metaIdx struct {
  472. Name string `json:"n"`
  473. Type byte `json:"t"`
  474. Columns []string `json:"c"`
  475. // Uniq 唯一索引 (SQL 侧的 UNIQUE INDEX / VRTX 的 IndexFlagUnique)。
  476. Uniq bool `json:"u,omitempty"`
  477. }
  478. const metaVersion = 1
  479. func (s *SQL) ensureMetaTable() error {
  480. if _, err := s.db.Exec(s.dial.metaDDL()); err != nil {
  481. return fmt.Errorf("建保真表 %s: %w", MetaTable, err)
  482. }
  483. return nil
  484. }
  485. func (s *SQL) writeMeta(sc *TableSchema) error {
  486. if err := s.ensureMetaTable(); err != nil {
  487. return err
  488. }
  489. dto := metaDTO{
  490. Version: metaVersion,
  491. PK: sc.PKColumn,
  492. PKIndexName: sc.PKIndexName,
  493. PKIndexType: byte(sc.PKIndexType),
  494. RowStore: sc.RowStoreColumns,
  495. }
  496. for _, c := range sc.Columns {
  497. dto.Columns = append(dto.Columns, metaCol{
  498. Name: c.Name, Type: byte(c.Type), Param: c.Param, Time: c.TimeKind,
  499. Null: c.Nullable, Uniq: c.Unique,
  500. })
  501. }
  502. for _, idx := range sc.Indexes {
  503. dto.Indexes = append(dto.Indexes, metaIdx{
  504. Name: idx.Name, Type: byte(idx.Type), Columns: idx.Columns, Uniq: idx.Unique,
  505. })
  506. }
  507. payload, err := json.Marshal(dto)
  508. if err != nil {
  509. return err
  510. }
  511. if _, err := s.db.Exec(s.dial.metaUpsert(), sc.Name, string(payload)); err != nil {
  512. return fmt.Errorf("写保真表 %s: %w", MetaTable, err)
  513. }
  514. return nil
  515. }
  516. func (s *SQL) readMeta(table string) (*metaDTO, bool) {
  517. if s.hasMeta == nil {
  518. s.hasMeta = map[string]*metaDTO{}
  519. }
  520. if m, ok := s.hasMeta[table]; ok {
  521. return m, m != nil
  522. }
  523. // 保真表不存在是常态 (外部库), 静默跳过
  524. if ok, err := s.Exists(MetaTable); err != nil || !ok {
  525. s.hasMeta[table] = nil
  526. return nil, false
  527. }
  528. var payload string
  529. err := s.db.QueryRow(s.dial.metaSelect(), table).Scan(&payload)
  530. if err != nil {
  531. if !errors.Is(err, sql.ErrNoRows) {
  532. s.opts.logf("[%s] 读保真表 %s 失败 (按 SQL 类型推断): %v", s.kind, MetaTable, err)
  533. }
  534. s.hasMeta[table] = nil
  535. return nil, false
  536. }
  537. var dto metaDTO
  538. if err := json.Unmarshal([]byte(payload), &dto); err != nil {
  539. s.opts.logf("[%s] 保真表里表 %s 的内容解析失败 (按 SQL 类型推断): %v", s.kind, table, err)
  540. s.hasMeta[table] = nil
  541. return nil, false
  542. }
  543. s.hasMeta[table] = &dto
  544. return &dto, true
  545. }
  546. // applyMeta 用保真数据覆盖推断出来的结构。
  547. //
  548. // 只覆盖"双方都有"的列: 目标 SQL 表里后来手工加的列照常按 SQL 类型推断,
  549. // 不会因为保真表里没有它就丢列。
  550. func applyMeta(s *TableSchema, m *metaDTO) {
  551. typeOf := make(map[string]types.FieldType, len(m.Columns))
  552. timeOf := make(map[string]string, len(m.Columns))
  553. paramOf := make(map[string]byte, len(m.Columns))
  554. nullOf := make(map[string]bool, len(m.Columns))
  555. uniqOf := make(map[string]bool, len(m.Columns))
  556. for _, c := range m.Columns {
  557. typeOf[c.Name] = types.FieldType(c.Type)
  558. if c.Param != 0 {
  559. paramOf[c.Name] = c.Param
  560. }
  561. if c.Time != "" {
  562. timeOf[c.Name] = c.Time
  563. }
  564. nullOf[c.Name] = c.Null
  565. uniqOf[c.Name] = c.Uniq
  566. }
  567. for i := range s.Columns {
  568. if t, ok := typeOf[s.Columns[i].Name]; ok {
  569. s.Columns[i].Type = t
  570. // NULL / 唯一声明无条件取自保真表: 旧格式 (没这两个字段) 解出来是 false,
  571. // 而旧格式导出的表当年就是"全列 NOT NULL + 无唯一约束", 语义正好对上。
  572. s.Columns[i].Nullable = nullOf[s.Columns[i].Name]
  573. s.Columns[i].Unique = uniqOf[s.Columns[i].Name]
  574. }
  575. // Param 无条件覆盖: 保真表里没记 (旧格式) 时是 0, 而推断出来的 SQL DECIMAL
  576. // 也可能带 scale —— 以保真表为准才叫"还原当初的结构"。
  577. if p, ok := paramOf[s.Columns[i].Name]; ok {
  578. s.Columns[i].Param = p
  579. } else if _, ok := typeOf[s.Columns[i].Name]; ok {
  580. s.Columns[i].Param = 0
  581. }
  582. if k, ok := timeOf[s.Columns[i].Name]; ok {
  583. s.Columns[i].TimeKind = k
  584. }
  585. }
  586. if m.PK != "" && s.HasColumn(m.PK) {
  587. s.PKColumn = m.PK
  588. }
  589. if len(m.RowStore) > 0 {
  590. s.RowStoreColumns = append([]string(nil), m.RowStore...)
  591. }
  592. if m.PKIndexName != "" {
  593. s.PKIndexName = m.PKIndexName
  594. }
  595. if m.PKIndexType != 0 {
  596. s.PKIndexType = page.IndexType(m.PKIndexType)
  597. }
  598. // 索引: 按名字合并, 保真表里的类型更可信 (SQL 侧根本区分不出 Hash/BTree)
  599. byName := make(map[string]metaIdx, len(m.Indexes))
  600. for _, idx := range m.Indexes {
  601. byName[idx.Name] = idx
  602. }
  603. for i := range s.Indexes {
  604. if mi, ok := byName[s.Indexes[i].Name]; ok {
  605. s.Indexes[i].Type = page.IndexType(mi.Type)
  606. s.Indexes[i].Unique = mi.Uniq
  607. if len(mi.Columns) > 0 {
  608. s.Indexes[i].Columns = mi.Columns
  609. }
  610. delete(byName, mi.Name)
  611. }
  612. }
  613. for _, mi := range m.Indexes {
  614. if _, ok := byName[mi.Name]; !ok {
  615. continue
  616. }
  617. if isRowStoreType(page.IndexType(mi.Type)) {
  618. continue
  619. }
  620. s.Indexes = append(s.Indexes, IndexSchema{
  621. Name: mi.Name,
  622. Type: page.IndexType(mi.Type),
  623. Columns: mi.Columns,
  624. Unique: mi.Uniq,
  625. })
  626. }
  627. }
  628. // isRowStoreType 主键/时序这类"承载行数据"的索引类型。
  629. func isRowStoreType(t page.IndexType) bool {
  630. return t == page.IndexTypePrimary || t == page.IndexTypeTime
  631. }