sql.go 23 KB

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