dialector.go 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472
  1. // Package gormvrtx 把 TsDB2 (vrtxsql database/sql 驱动) 接入 GORM:
  2. // CRUD 走 GORM 生成 SQL → vrtx 驱动执行; schema 自省 (AutoMigrate) 不经过 SQL,
  3. // 由 Migrator 直接调引擎 API。
  4. //
  5. // 引擎是嵌入式单写者存储, 与 mysql/pgx 这类网络库有两个本质差别, 本包据此裁剪:
  6. // - 无版本协商、无嵌套事务 (SavePoint/RollbackTo 不声明, 让 GORM 默认行为生效);
  7. // - 行存储即主键索引, 表必须有整型主键列承载。
  8. package gormvrtx
  9. import (
  10. "context"
  11. "database/sql"
  12. "fmt"
  13. "reflect"
  14. "regexp"
  15. "strconv"
  16. "strings"
  17. "time"
  18. "gorm.io/gorm"
  19. "gorm.io/gorm/callbacks"
  20. "gorm.io/gorm/clause"
  21. "gorm.io/gorm/schema"
  22. "git.ycl.pub/yl32940/TsDB2/DB/core"
  23. "git.ycl.pub/yl32940/TsDB2/DB/core/types"
  24. vrtxsql "git.ycl.pub/yl32940/TsDB2/DB/driver"
  25. )
  26. // DriverName 是 vrtxsql 注册的 database/sql 驱动名。
  27. const DriverName = "vrtx"
  28. // Config Dialector 配置。
  29. type Config struct {
  30. DriverName string // database/sql 驱动名, 默认 "vrtx"
  31. DSN string // 数据库文件路径 (可带 ?mode=ro / ?cache=private)
  32. }
  33. // Dialector 实现 gorm.Dialector。
  34. type Dialector struct {
  35. *Config
  36. db *core.DB // Initialize 时经 SharedDB 取得的引擎实例, Migrator 直接用
  37. sqlDB *sql.DB // 连接池本体 (供调用方 Close 时正确释放)
  38. }
  39. var _ gorm.Dialector = (*Dialector)(nil)
  40. // Open 以默认配置打开: DriverName="vrtx", DSN=文件路径。
  41. func Open(dsn string) gorm.Dialector {
  42. return &Dialector{Config: &Config{DriverName: DriverName, DSN: dsn}}
  43. }
  44. // New 按配置构造 Dialector; DriverName 为空时用 "vrtx"。
  45. func New(config Config) gorm.Dialector {
  46. if config.DriverName == "" {
  47. config.DriverName = DriverName
  48. }
  49. return &Dialector{Config: &config}
  50. }
  51. // Name 方言名。
  52. func (d *Dialector) Name() string { return "vrtx" }
  53. // Initialize 打开驱动连接池与引擎实例。不做任何 SELECT VERSION() 探测 ——
  54. // 引擎无版本概念, 打开时的 CRC/WAL 恢复即完整性校验。
  55. //
  56. // 两个引擎相关的适配点:
  57. // - SetMaxOpenConns(1): 引擎是单写者模型, 连接池压成 1 让所有 SQL 串行;
  58. // - ConnPool 包一层 rewritePool: 内联 LIMIT/OFFSET 整数参数 (驱动解析器支持限定名)
  59. // (旧解析器只认裸列名/字面量 LIMIT, 该包装让 GORM 的 SQL 形态稳定可用)。
  60. func (d *Dialector) Initialize(db *gorm.DB) error {
  61. if d.DriverName == "" {
  62. d.DriverName = DriverName
  63. }
  64. sqlDB, err := sql.Open(d.DriverName, d.DSN)
  65. if err != nil {
  66. return err
  67. }
  68. sqlDB.SetMaxOpenConns(1)
  69. eng, err := vrtxsql.SharedDB(d.DSN)
  70. if err != nil {
  71. _ = sqlDB.Close()
  72. return fmt.Errorf("vrtx: 打开引擎 %s: %w", d.DSN, err)
  73. }
  74. d.db = eng
  75. d.sqlDB = sqlDB
  76. db.ConnPool = rewritePool{DB: sqlDB}
  77. // GORM v1.25 的 Open 不注册默认回调 —— 由 Dialector 负责 (与 mysql/sqlite 驱动一致)。
  78. callbacks.RegisterDefaultCallbacks(db, &callbacks.Config{})
  79. // vrtx 的 SQL INSERT 语义要求显式提供整型主键 (引擎行存储定位键), 不像
  80. // AUTO_INCREMENT 那样省略列由库自增回填; 而 gorm 只有 tag 显式 autoIncrement
  81. // 才认为主键有库侧默认值 (LastInsertId 回填路径只对这类字段生效)。
  82. // 折衷: 建 create 前回调, 用引擎游标预分配主键写进对象 —— INSERT 带上该值走
  83. // 显式主键路径 (预留号免查重), 行为对齐"自增主键 + 回填"的直觉。
  84. db.Callback().Create().Before("gorm:create").Register("vrtx:assign_pk", d.assignPKCallback)
  85. return nil
  86. }
  87. // assignPKCallback create 前回调: 零值整型主键 → 引擎 AllocPKID 预分配并写回对象。
  88. func (d *Dialector) assignPKCallback(tx *gorm.DB) {
  89. stmt := tx.Statement
  90. if stmt == nil || stmt.Schema == nil || stmt.DryRun {
  91. return
  92. }
  93. pk := stmt.Schema.PrioritizedPrimaryField
  94. if pk == nil {
  95. return
  96. }
  97. kind := pk.IndirectFieldType.Kind()
  98. isUint := kind == reflect.Uint || kind == reflect.Uint8 || kind == reflect.Uint16 ||
  99. kind == reflect.Uint32 || kind == reflect.Uint64
  100. isInt := kind == reflect.Int || kind == reflect.Int8 || kind == reflect.Int16 ||
  101. kind == reflect.Int32 || kind == reflect.Int64
  102. if !isUint && !isInt {
  103. return
  104. }
  105. // 列被 Select/Omit 排除时跳过 (INSERT 不会带它, 预分配反而写错对象)。
  106. selectColumns, restricted := stmt.SelectAndOmitColumns(true, false)
  107. if v, present := selectColumns[pk.DBName]; present {
  108. if !v {
  109. return
  110. }
  111. } else if restricted {
  112. return
  113. }
  114. tbl, ok := d.db.FindTable(stmt.Table)
  115. if !ok {
  116. return // 表不存在: 让 INSERT 自己报错
  117. }
  118. // ⭐ AllocPKID 必须在【逐行闭包内】调用: 批量创建 (切片) 时每行都要独立
  119. // 主键号 — 在循环外只取一次, 切片所有行会共用同一个号, 第二行起被引擎
  120. // 按重复主键拒掉 (实测: 监控批量保存 Create(&metrics) 3 行共用一号全灭)。
  121. // 号只增不减, 逐行分配产生的跳号无害。
  122. assign := func(rv reflect.Value) {
  123. _, isZero := pk.ValueOf(stmt.Context, rv)
  124. if !isZero {
  125. return
  126. }
  127. id := tbl.AllocPKID()
  128. if isUint {
  129. tx.AddError(pk.Set(stmt.Context, rv, id))
  130. } else {
  131. tx.AddError(pk.Set(stmt.Context, rv, int64(id)))
  132. }
  133. }
  134. rv := stmt.ReflectValue
  135. for rv.Kind() == reflect.Ptr {
  136. if rv.IsNil() {
  137. return
  138. }
  139. rv = rv.Elem()
  140. }
  141. switch rv.Kind() {
  142. case reflect.Struct:
  143. assign(rv)
  144. case reflect.Slice, reflect.Array:
  145. for i := 0; i < rv.Len(); i++ {
  146. assign(rv.Index(i))
  147. }
  148. }
  149. }
  150. // Migrator 返回自定义 Migrator: schema 自省直接调引擎 API (不经 SQL)。
  151. func (d *Dialector) Migrator(db *gorm.DB) gorm.Migrator {
  152. return Migrator{db: db, eng: d.db, dsn: d.DSN, dialector: d}
  153. }
  154. // explicitTypeMapping 小型 SQL 类型映射: 把 gorm:"type:..." 标签里用户显式声明的
  155. // SQL 类型归一化成引擎侧类型名 (前缀 = 空白或 '(' 之前的部分, 故 "bigint unsigned"
  156. // 归 BIGINT、"varchar(500)" 归 VARCHAR 且宽度取括号值)。独立实现, 不依赖 DB/driver
  157. // 的 SQL 文本解析。第三个返回值为 false 表示不认识 (decimal/json/...), 调用方落回
  158. // Go 类型的默认映射并注释说明。
  159. func explicitTypeMapping(tag string) (string, int, bool) {
  160. t := strings.ToLower(strings.TrimSpace(tag))
  161. if t == "" {
  162. return "", 0, false
  163. }
  164. name := t
  165. if i := strings.IndexAny(name, "( "); i >= 0 {
  166. name = name[:i]
  167. }
  168. size := 0
  169. if i := strings.IndexByte(t, '('); i >= 0 {
  170. if j := strings.IndexByte(t[i:], ')'); j > 0 {
  171. size, _ = strconv.Atoi(strings.TrimSpace(t[i+1 : i+j]))
  172. }
  173. }
  174. switch name {
  175. case "bigint":
  176. return "BIGINT", 0, true
  177. case "int":
  178. return "INT", 0, true
  179. case "varchar", "char":
  180. return "VARCHAR", size, true
  181. case "text":
  182. return "TEXT", 0, true
  183. case "double":
  184. return "DOUBLE", 0, true
  185. case "bool", "boolean":
  186. return "BOOLEAN", 0, true
  187. case "datetime":
  188. return "DATETIME", 0, true
  189. }
  190. return "", 0, false
  191. }
  192. // DataTypeOf GORM 字段 → 引擎理解的 SQL 类型名 (供 FullDataTypeOf / 高级用法;
  193. // 建表本身走引擎 API, 不经 SQL)。
  194. //
  195. // 显式 gorm:"type:..." 标签优先 (报告 #10 的逃生门: string 字段默认无 size 会落成
  196. // Blob 链式列, 带索引时引擎拒绝建索引 —— 用 type:varchar(500) 精确声明即可绕过
  197. // 默认映射): 按 explicitTypeMapping 解析, 认识 → 按其返回; 不认识 → 保持默认映射
  198. // 并附注释说明。
  199. // 默认映射: string 带尺寸 → VARCHAR(size), 超过引擎行内定长上限 2048 时截断声明并
  200. // 注释说明; 无尺寸 → BLOB (链式存储)。
  201. //
  202. // 无尺寸 string 落 BLOB 而不是旧行为的 TEXT (动态列): 动态列单值上限 16360 字节
  203. // (types.MaxStringSize), 运行时写入更大的文本会撞墙; Blob 列行内 8B 引用 + 链页,
  204. // 单值上限约 4GB。读回时驱动把链内容物化成 []byte, GORM 照常扫进 string。
  205. // 权衡: 该列失去 LIKE 与索引能力 (行内只有引用, 按值查不到) —— 需要可搜索的大文本
  206. // 请给 size:1..2048 走定长档, 或把检索字段拆出来单独建列。
  207. func (d *Dialector) DataTypeOf(field *schema.Field) string {
  208. if tag, ok := field.TagSettings["TYPE"]; ok {
  209. if name, size, ok := explicitTypeMapping(tag); ok {
  210. switch name {
  211. case "TEXT":
  212. // 显式 type:text 与无 size 默认同一归宿: Blob 链式列 (权衡见上)。
  213. return "BLOB /* 不定长大文本: 链式存储, 单值上限 ~4GB; 失去 LIKE/索引能力, 需要可搜索文本请用 size:1..2048 */"
  214. case "VARCHAR":
  215. switch {
  216. case size <= 0:
  217. return "VARCHAR /* 声明无宽度, 建列按动态列处理 */"
  218. case size > types.MaxInlineColumnWidth:
  219. return fmt.Sprintf("VARCHAR(2048) /* 声明 %d 超引擎行内上限 2048, 截断 */", size)
  220. default:
  221. return fmt.Sprintf("VARCHAR(%d)", size)
  222. }
  223. default:
  224. return name
  225. }
  226. }
  227. return fmt.Sprintf("%s /* 未知 type:%s, 按 Go 类型默认映射 */", d.defaultDataTypeOf(field), tag)
  228. }
  229. return d.defaultDataTypeOf(field)
  230. }
  231. // defaultDataTypeOf 按 Go 类型给默认 SQL 类型名 (DataTypeOf 的回落分支)。
  232. //
  233. // 无尺寸 string 落 BLOB (链式存储): 动态列单值上限 16360 字节, 不定长大文本运行时会
  234. // 撞墙; Blob 列行内 8B 引用 + 链页 (~4GB)。代价是失去 LIKE/索引能力 —— 需要可搜索
  235. // 文本时给 size:1..2048 走定长档。读回经驱动物化成 []byte, GORM 可扫进 string。
  236. func (d *Dialector) defaultDataTypeOf(field *schema.Field) string {
  237. switch field.DataType {
  238. case schema.Uint:
  239. return "BIGINT UNSIGNED"
  240. case schema.Int:
  241. return "BIGINT"
  242. case schema.Float:
  243. return "DOUBLE"
  244. case schema.String:
  245. if field.Size > 0 {
  246. if field.Size > types.MaxInlineColumnWidth {
  247. return fmt.Sprintf("VARCHAR(2048) /* 声明 %d 超引擎行内上限 2048, 截断 */", field.Size)
  248. }
  249. return fmt.Sprintf("VARCHAR(%d)", field.Size)
  250. }
  251. return "BLOB"
  252. case schema.Bool:
  253. return "BOOLEAN"
  254. case schema.Time:
  255. return "DATETIME" // 引擎约定: 行内 8B Unix 毫秒
  256. case schema.Bytes:
  257. return "BLOB"
  258. default:
  259. // struct / JSON 等: 序列化为文本存动态列
  260. return "TEXT"
  261. }
  262. }
  263. // DefaultValueOf 引擎不支持列级 DEFAULT (未提供的列按 NULL/零值裁决), 给出占位表达式。
  264. func (d *Dialector) DefaultValueOf(field *schema.Field) clause.Expression {
  265. return clause.Expr{SQL: "DEFAULT"}
  266. }
  267. // BindVarTo 占位符: 与 mysql 一致, 一律 "?" (驱动只认 ? 占位符)。
  268. func (d *Dialector) BindVarTo(w clause.Writer, stmt *gorm.Statement, v interface{}) {
  269. w.WriteByte('?')
  270. }
  271. // QuoteTo 反引号包裹; 含 "." 时按段分别引用: t.col → `t`.`col`。
  272. func (d *Dialector) QuoteTo(w clause.Writer, s string) {
  273. for i, part := range strings.Split(s, ".") {
  274. if i > 0 {
  275. w.WriteByte('.')
  276. }
  277. w.WriteByte('`')
  278. w.WriteString(strings.ReplaceAll(part, "`", "``"))
  279. w.WriteByte('`')
  280. }
  281. }
  282. // Explain 把 "?" 按序替换为字面值 (字符串单引号包裹并转义, 数字直写,
  283. // bool → 1/0, nil → NULL)。仅用于 SQL 日志展示。
  284. func (d *Dialector) Explain(sql string, vars ...interface{}) string {
  285. var sb strings.Builder
  286. idx := 0
  287. for i := 0; i < len(sql); i++ {
  288. c := sql[i]
  289. if c != '?' || idx >= len(vars) {
  290. sb.WriteByte(c)
  291. continue
  292. }
  293. sb.WriteString(literalOf(vars[idx]))
  294. idx++
  295. }
  296. return sb.String()
  297. }
  298. // literalOf 把一个参数值格式化为 SQL 字面量。
  299. func literalOf(v interface{}) string {
  300. switch x := v.(type) {
  301. case nil:
  302. return "NULL"
  303. case string:
  304. return "'" + strings.ReplaceAll(x, "'", "''") + "'"
  305. case []byte:
  306. return "'" + strings.ReplaceAll(string(x), "'", "''") + "'"
  307. case bool:
  308. if x {
  309. return "1"
  310. }
  311. return "0"
  312. case time.Time:
  313. return "'" + x.Format("2006-01-02 15:04:05.999") + "'"
  314. }
  315. rv := reflect.ValueOf(v)
  316. switch rv.Kind() {
  317. case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64:
  318. return strconv.FormatInt(rv.Int(), 10)
  319. case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64:
  320. return strconv.FormatUint(rv.Uint(), 10)
  321. case reflect.Float32, reflect.Float64:
  322. return strconv.FormatFloat(rv.Float(), 'g', -1, 64)
  323. }
  324. return "'" + strings.ReplaceAll(fmt.Sprint(v), "'", "''") + "'"
  325. }
  326. // =============================================================================
  327. // rewritePool: SQL 进驱动前的轻量改写 (内联 LIMIT/OFFSET 整数参数)
  328. // =============================================================================
  329. // 注: 早期驱动解析器不支持表限定列名, 这里曾剥掉 "`t`." 前缀; 驱动现已原生
  330. // 支持限定名 (单表剥限定 + JOIN 按别名归属), 剥前缀反而会制造歧义 (JOIN 下
  331. // 两表同名列), 故只保留 LIMIT/OFFSET 参数内联。
  332. // limitClauseRe 匹配 LIMIT/OFFSET 占位符形态 (GORM 把 Limit(1) 编成 "LIMIT ?",
  333. // 而 vrtx 解析器只接受字面量整数, 需要按参数内联)。
  334. var limitClauseRe = regexp.MustCompile(`(?i)\b(LIMIT|OFFSET)[ \t\r\n]+\?`)
  335. // rewriteSQL 进驱动前的统一改写: 内联 LIMIT/OFFSET 整数参数 (限定名保持原样)。
  336. func rewriteSQL(query string, args []interface{}) (string, []interface{}) {
  337. if !limitClauseRe.MatchString(query) {
  338. return query, args
  339. }
  340. // 找出 LIMIT/OFFSET 子句里 '?' 是整条 SQL 的第几个 '?' (即参数下标),
  341. // 把它替换成字面量并把对应参数从 args 剔除 (后面的参数依次前移)。
  342. mark := map[int]string{} // 第k个 '?' → 字面量
  343. for _, loc := range limitClauseRe.FindAllStringSubmatchIndex(query, -1) {
  344. qpos := strings.IndexByte(query[loc[0]:loc[1]], '?')
  345. if qpos < 0 {
  346. continue
  347. }
  348. abs := loc[0] + qpos
  349. k := 0
  350. for i := 0; i < abs; i++ {
  351. if query[i] == '?' {
  352. k++
  353. }
  354. }
  355. if k >= len(args) {
  356. continue
  357. }
  358. mark[k] = intLiteral(args[k])
  359. }
  360. if len(mark) == 0 {
  361. return query, args
  362. }
  363. var sb strings.Builder
  364. argIdx := 0
  365. for i := 0; i < len(query); i++ {
  366. c := query[i]
  367. if c != '?' {
  368. sb.WriteByte(c)
  369. continue
  370. }
  371. if lit, ok := mark[argIdx]; ok {
  372. sb.WriteString(lit)
  373. } else {
  374. sb.WriteByte('?')
  375. }
  376. argIdx++
  377. }
  378. kept := make([]interface{}, 0, len(args)-len(mark))
  379. for i, a := range args {
  380. if _, ok := mark[i]; !ok {
  381. kept = append(kept, a)
  382. }
  383. }
  384. return sb.String(), kept
  385. }
  386. // intLiteral 把 LIMIT/OFFSET 参数格式化为整数字面量; 不认识的类型原样占位返回 "0"。
  387. func intLiteral(v interface{}) string {
  388. switch x := v.(type) {
  389. case int:
  390. return strconv.Itoa(x)
  391. case int8:
  392. return strconv.FormatInt(int64(x), 10)
  393. case int16:
  394. return strconv.FormatInt(int64(x), 10)
  395. case int32:
  396. return strconv.FormatInt(int64(x), 10)
  397. case int64:
  398. return strconv.FormatInt(x, 10)
  399. case uint:
  400. return strconv.FormatUint(uint64(x), 10)
  401. case uint8:
  402. return strconv.FormatUint(uint64(x), 10)
  403. case uint16:
  404. return strconv.FormatUint(uint64(x), 10)
  405. case uint32:
  406. return strconv.FormatUint(uint64(x), 10)
  407. case uint64:
  408. return strconv.FormatUint(x, 10)
  409. case float64:
  410. return strconv.FormatInt(int64(x), 10)
  411. case string:
  412. return x
  413. }
  414. return "0"
  415. }
  416. // rewritePool 包装 *sql.DB, SQL 进驱动前统一过 rewriteSQL (剥 “ `t`. “ 限定前缀、
  417. // 内联 LIMIT/OFFSET)。嵌入 *sql.DB 顺带获得 BeginTx (TxBeginner) 等全部池能力。
  418. type rewritePool struct {
  419. *sql.DB
  420. }
  421. func (p rewritePool) PrepareContext(ctx context.Context, query string) (*sql.Stmt, error) {
  422. q, _ := rewriteSQL(query, nil)
  423. return p.DB.PrepareContext(ctx, q)
  424. }
  425. func (p rewritePool) ExecContext(ctx context.Context, query string, args ...interface{}) (sql.Result, error) {
  426. q, a := rewriteSQL(query, args)
  427. return p.DB.ExecContext(ctx, q, a...)
  428. }
  429. func (p rewritePool) QueryContext(ctx context.Context, query string, args ...interface{}) (*sql.Rows, error) {
  430. q, a := rewriteSQL(query, args)
  431. return p.DB.QueryContext(ctx, q, a...)
  432. }
  433. func (p rewritePool) QueryRowContext(ctx context.Context, query string, args ...interface{}) *sql.Row {
  434. q, a := rewriteSQL(query, args)
  435. return p.DB.QueryRowContext(ctx, q, a...)
  436. }