dialector.go 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452
  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. id := tbl.AllocPKID()
  119. assign := func(rv reflect.Value) {
  120. _, isZero := pk.ValueOf(stmt.Context, rv)
  121. if !isZero {
  122. return
  123. }
  124. if isUint {
  125. tx.AddError(pk.Set(stmt.Context, rv, id))
  126. } else {
  127. tx.AddError(pk.Set(stmt.Context, rv, int64(id)))
  128. }
  129. }
  130. rv := stmt.ReflectValue
  131. for rv.Kind() == reflect.Ptr {
  132. if rv.IsNil() {
  133. return
  134. }
  135. rv = rv.Elem()
  136. }
  137. switch rv.Kind() {
  138. case reflect.Struct:
  139. assign(rv)
  140. case reflect.Slice, reflect.Array:
  141. for i := 0; i < rv.Len(); i++ {
  142. assign(rv.Index(i))
  143. }
  144. }
  145. }
  146. // Migrator 返回自定义 Migrator: schema 自省直接调引擎 API (不经 SQL)。
  147. func (d *Dialector) Migrator(db *gorm.DB) gorm.Migrator {
  148. return Migrator{db: db, eng: d.db, dsn: d.DSN, dialector: d}
  149. }
  150. // explicitTypeMapping 小型 SQL 类型映射: 把 gorm:"type:..." 标签里用户显式声明的
  151. // SQL 类型归一化成引擎侧类型名 (前缀 = 空白或 '(' 之前的部分, 故 "bigint unsigned"
  152. // 归 BIGINT、"varchar(500)" 归 VARCHAR 且宽度取括号值)。独立实现, 不依赖 DB/driver
  153. // 的 SQL 文本解析。第三个返回值为 false 表示不认识 (decimal/json/...), 调用方落回
  154. // Go 类型的默认映射并注释说明。
  155. func explicitTypeMapping(tag string) (string, int, bool) {
  156. t := strings.ToLower(strings.TrimSpace(tag))
  157. if t == "" {
  158. return "", 0, false
  159. }
  160. name := t
  161. if i := strings.IndexAny(name, "( "); i >= 0 {
  162. name = name[:i]
  163. }
  164. size := 0
  165. if i := strings.IndexByte(t, '('); i >= 0 {
  166. if j := strings.IndexByte(t[i:], ')'); j > 0 {
  167. size, _ = strconv.Atoi(strings.TrimSpace(t[i+1 : i+j]))
  168. }
  169. }
  170. switch name {
  171. case "bigint":
  172. return "BIGINT", 0, true
  173. case "int":
  174. return "INT", 0, true
  175. case "varchar", "char":
  176. return "VARCHAR", size, true
  177. case "text":
  178. return "TEXT", 0, true
  179. case "double":
  180. return "DOUBLE", 0, true
  181. case "bool", "boolean":
  182. return "BOOLEAN", 0, true
  183. case "datetime":
  184. return "DATETIME", 0, true
  185. }
  186. return "", 0, false
  187. }
  188. // DataTypeOf GORM 字段 → 引擎理解的 SQL 类型名 (供 FullDataTypeOf / 高级用法;
  189. // 建表本身走引擎 API, 不经 SQL)。
  190. //
  191. // 显式 gorm:"type:..." 标签优先 (报告 #10 的逃生门: string 字段默认无 size 会落成
  192. // 动态列, 带索引时引擎拒绝建索引 —— 用 type:varchar(500) 精确声明即可绕过默认映射):
  193. // 按 explicitTypeMapping 解析, 认识 → 按其返回; 不认识 → 保持默认映射并附注释说明。
  194. // 默认映射: string 带尺寸 → VARCHAR(size), 超过引擎行内定长上限 2048 时截断声明并
  195. // 注释说明; 无尺寸 → TEXT (动态列)。
  196. func (d *Dialector) DataTypeOf(field *schema.Field) string {
  197. if tag, ok := field.TagSettings["TYPE"]; ok {
  198. if name, size, ok := explicitTypeMapping(tag); ok {
  199. if name != "VARCHAR" {
  200. return name
  201. }
  202. switch {
  203. case size <= 0:
  204. return "VARCHAR /* 声明无宽度, 建列按动态列处理 */"
  205. case size > types.MaxInlineColumnWidth:
  206. return fmt.Sprintf("VARCHAR(2048) /* 声明 %d 超引擎行内上限 2048, 截断 */", size)
  207. default:
  208. return fmt.Sprintf("VARCHAR(%d)", size)
  209. }
  210. }
  211. return fmt.Sprintf("%s /* 未知 type:%s, 按 Go 类型默认映射 */", d.defaultDataTypeOf(field), tag)
  212. }
  213. return d.defaultDataTypeOf(field)
  214. }
  215. // defaultDataTypeOf 按 Go 类型给默认 SQL 类型名 (DataTypeOf 的回落分支)。
  216. func (d *Dialector) defaultDataTypeOf(field *schema.Field) string {
  217. switch field.DataType {
  218. case schema.Uint:
  219. return "BIGINT UNSIGNED"
  220. case schema.Int:
  221. return "BIGINT"
  222. case schema.Float:
  223. return "DOUBLE"
  224. case schema.String:
  225. if field.Size > 0 {
  226. if field.Size > types.MaxInlineColumnWidth {
  227. return fmt.Sprintf("VARCHAR(2048) /* 声明 %d 超引擎行内上限 2048, 截断 */", field.Size)
  228. }
  229. return fmt.Sprintf("VARCHAR(%d)", field.Size)
  230. }
  231. return "TEXT"
  232. case schema.Bool:
  233. return "BOOLEAN"
  234. case schema.Time:
  235. return "DATETIME" // 引擎约定: 行内 8B Unix 毫秒
  236. case schema.Bytes:
  237. return "BLOB"
  238. default:
  239. // struct / JSON 等: 序列化为文本存动态列
  240. return "TEXT"
  241. }
  242. }
  243. // DefaultValueOf 引擎不支持列级 DEFAULT (未提供的列按 NULL/零值裁决), 给出占位表达式。
  244. func (d *Dialector) DefaultValueOf(field *schema.Field) clause.Expression {
  245. return clause.Expr{SQL: "DEFAULT"}
  246. }
  247. // BindVarTo 占位符: 与 mysql 一致, 一律 "?" (驱动只认 ? 占位符)。
  248. func (d *Dialector) BindVarTo(w clause.Writer, stmt *gorm.Statement, v interface{}) {
  249. w.WriteByte('?')
  250. }
  251. // QuoteTo 反引号包裹; 含 "." 时按段分别引用: t.col → `t`.`col`。
  252. func (d *Dialector) QuoteTo(w clause.Writer, s string) {
  253. for i, part := range strings.Split(s, ".") {
  254. if i > 0 {
  255. w.WriteByte('.')
  256. }
  257. w.WriteByte('`')
  258. w.WriteString(strings.ReplaceAll(part, "`", "``"))
  259. w.WriteByte('`')
  260. }
  261. }
  262. // Explain 把 "?" 按序替换为字面值 (字符串单引号包裹并转义, 数字直写,
  263. // bool → 1/0, nil → NULL)。仅用于 SQL 日志展示。
  264. func (d *Dialector) Explain(sql string, vars ...interface{}) string {
  265. var sb strings.Builder
  266. idx := 0
  267. for i := 0; i < len(sql); i++ {
  268. c := sql[i]
  269. if c != '?' || idx >= len(vars) {
  270. sb.WriteByte(c)
  271. continue
  272. }
  273. sb.WriteString(literalOf(vars[idx]))
  274. idx++
  275. }
  276. return sb.String()
  277. }
  278. // literalOf 把一个参数值格式化为 SQL 字面量。
  279. func literalOf(v interface{}) string {
  280. switch x := v.(type) {
  281. case nil:
  282. return "NULL"
  283. case string:
  284. return "'" + strings.ReplaceAll(x, "'", "''") + "'"
  285. case []byte:
  286. return "'" + strings.ReplaceAll(string(x), "'", "''") + "'"
  287. case bool:
  288. if x {
  289. return "1"
  290. }
  291. return "0"
  292. case time.Time:
  293. return "'" + x.Format("2006-01-02 15:04:05.999") + "'"
  294. }
  295. rv := reflect.ValueOf(v)
  296. switch rv.Kind() {
  297. case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64:
  298. return strconv.FormatInt(rv.Int(), 10)
  299. case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64:
  300. return strconv.FormatUint(rv.Uint(), 10)
  301. case reflect.Float32, reflect.Float64:
  302. return strconv.FormatFloat(rv.Float(), 'g', -1, 64)
  303. }
  304. return "'" + strings.ReplaceAll(fmt.Sprint(v), "'", "''") + "'"
  305. }
  306. // =============================================================================
  307. // rewritePool: SQL 进驱动前的轻量改写 (内联 LIMIT/OFFSET 整数参数)
  308. // =============================================================================
  309. // 注: 早期驱动解析器不支持表限定列名, 这里曾剥掉 "`t`." 前缀; 驱动现已原生
  310. // 支持限定名 (单表剥限定 + JOIN 按别名归属), 剥前缀反而会制造歧义 (JOIN 下
  311. // 两表同名列), 故只保留 LIMIT/OFFSET 参数内联。
  312. // limitClauseRe 匹配 LIMIT/OFFSET 占位符形态 (GORM 把 Limit(1) 编成 "LIMIT ?",
  313. // 而 vrtx 解析器只接受字面量整数, 需要按参数内联)。
  314. var limitClauseRe = regexp.MustCompile(`(?i)\b(LIMIT|OFFSET)[ \t\r\n]+\?`)
  315. // rewriteSQL 进驱动前的统一改写: 内联 LIMIT/OFFSET 整数参数 (限定名保持原样)。
  316. func rewriteSQL(query string, args []interface{}) (string, []interface{}) {
  317. if !limitClauseRe.MatchString(query) {
  318. return query, args
  319. }
  320. // 找出 LIMIT/OFFSET 子句里 '?' 是整条 SQL 的第几个 '?' (即参数下标),
  321. // 把它替换成字面量并把对应参数从 args 剔除 (后面的参数依次前移)。
  322. mark := map[int]string{} // 第k个 '?' → 字面量
  323. for _, loc := range limitClauseRe.FindAllStringSubmatchIndex(query, -1) {
  324. qpos := strings.IndexByte(query[loc[0]:loc[1]], '?')
  325. if qpos < 0 {
  326. continue
  327. }
  328. abs := loc[0] + qpos
  329. k := 0
  330. for i := 0; i < abs; i++ {
  331. if query[i] == '?' {
  332. k++
  333. }
  334. }
  335. if k >= len(args) {
  336. continue
  337. }
  338. mark[k] = intLiteral(args[k])
  339. }
  340. if len(mark) == 0 {
  341. return query, args
  342. }
  343. var sb strings.Builder
  344. argIdx := 0
  345. for i := 0; i < len(query); i++ {
  346. c := query[i]
  347. if c != '?' {
  348. sb.WriteByte(c)
  349. continue
  350. }
  351. if lit, ok := mark[argIdx]; ok {
  352. sb.WriteString(lit)
  353. } else {
  354. sb.WriteByte('?')
  355. }
  356. argIdx++
  357. }
  358. kept := make([]interface{}, 0, len(args)-len(mark))
  359. for i, a := range args {
  360. if _, ok := mark[i]; !ok {
  361. kept = append(kept, a)
  362. }
  363. }
  364. return sb.String(), kept
  365. }
  366. // intLiteral 把 LIMIT/OFFSET 参数格式化为整数字面量; 不认识的类型原样占位返回 "0"。
  367. func intLiteral(v interface{}) string {
  368. switch x := v.(type) {
  369. case int:
  370. return strconv.Itoa(x)
  371. case int8:
  372. return strconv.FormatInt(int64(x), 10)
  373. case int16:
  374. return strconv.FormatInt(int64(x), 10)
  375. case int32:
  376. return strconv.FormatInt(int64(x), 10)
  377. case int64:
  378. return strconv.FormatInt(x, 10)
  379. case uint:
  380. return strconv.FormatUint(uint64(x), 10)
  381. case uint8:
  382. return strconv.FormatUint(uint64(x), 10)
  383. case uint16:
  384. return strconv.FormatUint(uint64(x), 10)
  385. case uint32:
  386. return strconv.FormatUint(uint64(x), 10)
  387. case uint64:
  388. return strconv.FormatUint(x, 10)
  389. case float64:
  390. return strconv.FormatInt(int64(x), 10)
  391. case string:
  392. return x
  393. }
  394. return "0"
  395. }
  396. // rewritePool 包装 *sql.DB, SQL 进驱动前统一过 rewriteSQL (剥 “ `t`. “ 限定前缀、
  397. // 内联 LIMIT/OFFSET)。嵌入 *sql.DB 顺带获得 BeginTx (TxBeginner) 等全部池能力。
  398. type rewritePool struct {
  399. *sql.DB
  400. }
  401. func (p rewritePool) PrepareContext(ctx context.Context, query string) (*sql.Stmt, error) {
  402. q, _ := rewriteSQL(query, nil)
  403. return p.DB.PrepareContext(ctx, q)
  404. }
  405. func (p rewritePool) ExecContext(ctx context.Context, query string, args ...interface{}) (sql.Result, error) {
  406. q, a := rewriteSQL(query, args)
  407. return p.DB.ExecContext(ctx, q, a...)
  408. }
  409. func (p rewritePool) QueryContext(ctx context.Context, query string, args ...interface{}) (*sql.Rows, error) {
  410. q, a := rewriteSQL(query, args)
  411. return p.DB.QueryContext(ctx, q, a...)
  412. }
  413. func (p rewritePool) QueryRowContext(ctx context.Context, query string, args ...interface{}) *sql.Row {
  414. q, a := rewriteSQL(query, args)
  415. return p.DB.QueryRowContext(ctx, q, a...)
  416. }