TsDB2/migrate/sqlite.go

package migrate

import (
	"git.ycl.pub/yl32940/TsDB2/DB/core/page"
	"database/sql"
	"fmt"
	"strings"

	_ "modernc.org/sqlite" // 纯 Go 的 SQLite 实现, 不依赖 CGO
)

// OpenSQLite 打开 (不存在则新建) 一个 SQLite 库文件。
//
// path 可以是普通路径, 也可以是 "file:xxx.db?_pragma=busy_timeout(5000)" 这种 URI。
// 只留 1 个连接: SQLite 同一时刻只有一个写者, 多连接更容易撞 SQLITE_BUSY。
func OpenSQLite(path string, opts *Options) (*SQL, error) {
	db, err := sql.Open("sqlite", path)
	if err != nil {
		return nil, fmt.Errorf("打开 SQLite %s: %w", path, err)
	}
	db.SetMaxOpenConns(1)
	if err := db.Ping(); err != nil {
		_ = db.Close()
		return nil, fmt.Errorf("打开 SQLite %s: %w", path, err)
	}
	return newSQL(db, KindSQLite, path, sqliteDial{}, opts), nil
}

type sqliteDial struct{}

func (sqliteDial) kind() Kind { return KindSQLite }

func (sqliteDial) quote(name string) string {
	return `"` + strings.ReplaceAll(name, `"`, `""`) + `"`
}

func (d sqliteDial) columnDDL(c Column) string {
	return d.quote(c.Name) + " " + columnTypeName(c, false) + nullClause(c)
}

func (d sqliteDial) createTable(s *TableSchema) string {
	parts := make([]string, 0, len(s.Columns)+1)
	for _, c := range s.Columns {
		parts = append(parts, "  "+d.columnDDL(c))
	}
	if s.PKColumn != "" {
		parts = append(parts, "  PRIMARY KEY ("+d.quote(s.PKColumn)+")")
	}
	return fmt.Sprintf("CREATE TABLE %s (\n%s\n)", d.quote(s.Name), strings.Join(parts, ",\n"))
}

func (d sqliteDial) dropTable(table string) string {
	return "DROP TABLE IF EXISTS " + d.quote(table)
}

func (d sqliteDial) addColumn(table string, c Column) string {
	return fmt.Sprintf("ALTER TABLE %s ADD COLUMN %s", d.quote(table), d.columnDDL(c))
}

func (d sqliteDial) createIndex(table string, idx IndexSchema) string {
	cols := make([]string, len(idx.Columns))
	for i, c := range idx.Columns {
		cols[i] = d.quote(c)
	}
	// SQLite 的索引名是库级唯一的 (不像 MySQL 按表命名空间), 撞名时退让:
	// 迁移过来的 VRTX 索引名通常自带表名, 这里只是兜底。
	return fmt.Sprintf("%s IF NOT EXISTS %s ON %s (%s)",
		createIndexKind(idx), d.quote(idx.Name), d.quote(table), strings.Join(cols, ", "))
}

func (d sqliteDial) listTables(db *sql.DB) ([]string, error) {
	rows, err := db.Query(`SELECT name FROM sqlite_master
		WHERE type = 'table' AND name NOT LIKE 'sqlite_%' ORDER BY name`)
	if err != nil {
		return nil, fmt.Errorf("列 SQLite 表: %w", err)
	}
	defer rows.Close()
	var out []string
	for rows.Next() {
		var name string
		if err := rows.Scan(&name); err != nil {
			return nil, err
		}
		out = append(out, name)
	}
	return out, rows.Err()
}

// columns 读列与主键列名 (PRAGMA table_info; pk 字段是该列在主键里的序号)。
// 第三个返回值是"声明容量超过单值上限"的列 (见 capacityNote), 由 SQL.Schema 收进报告。
func (d sqliteDial) columns(db *sql.DB, table string) ([]Column, string, sourceNotes, error) {
	rows, err := db.Query("PRAGMA table_info(" + d.quote(table) + ")")
	if err != nil {
		return nil, "", sourceNotes{}, err
	}
	defer rows.Close()

	type pkCol struct {
		seq  int
		name string
	}
	var (
		cols  []Column
		pks   []pkCol
		notes sourceNotes
	)
	for rows.Next() {
		var (
			cid, notNull, pk int
			name, decl       string
			dflt             any
		)
		if err := rows.Scan(&cid, &name, &decl, &notNull, &dflt, &pk); err != nil {
			return nil, "", sourceNotes{}, err
		}
		ft, param, note := inferSQLTypeWithNote(decl, false)
		if note != "" {
			notes.Capped = append(notes.Capped, name+": "+note)
		}
		if d := degradeNote(decl); d != "" {
			notes.Degraded = append(notes.Degraded, name+": "+d)
		}
		cols = append(cols, Column{
			Name:     name,
			Type:     ft,
			Param:    param,
			Decl:     decl,
			TimeKind: sqlTimeKind(decl),
			Nullable: notNull == 0, // PRAGMA 的 notnull: 0 = 该列可为空
		})
		if pk > 0 {
			pks = append(pks, pkCol{seq: pk, name: name})
		}
	}
	if err := rows.Err(); err != nil {
		return nil, "", sourceNotes{}, err
	}
	if len(pks) == 1 {
		// SQLite 的 `id INTEGER PRIMARY KEY` 是 rowid 别名, PRAGMA 里 notnull = 0
		// (非空由"自动生成 rowid"保证, 不由 NOT NULL 约束表达)。而 VRTX 的行存储列
		// 承载 pkID, 必须有值 —— 不在这里钉住, 外部库导进来就会得到一张"主键列可为空"
		// 的 VRTX 表。注意这是 SQLite 特有的: 我们导出的表会显式写 NOT NULL, 读回
		// notnull = 1, 所以这一步只对"外部建的库"生效, 不影响往返。
		for i := range cols {
			if cols[i].Name == pks[0].name {
				cols[i].Nullable = false
			}
		}
		return cols, pks[0].name, notes, nil
	}
	if len(pks) > 1 {
		// `PRIMARY KEY (a, b)` 是 SQLite 的表级约束, PRAGMA 里表现为多列的 pk 序号。
		// 与 MySQL 同名场景一样: VRTX 的行存储索引只能单列, 但"源表主键是哪几列"这个
		// 事实只有这里知道, 记下来供下游报准话。
		ordered := make([]string, len(pks))
		for _, p := range pks {
			if p.seq >= 1 && p.seq <= len(pks) {
				ordered[p.seq-1] = p.name
			}
		}
		notes.PK = compositePKNote(table, ordered)
	}
	// 复合主键 / 无主键: 都当作"没有单列主键", 由调用方决定自增还是报错
	return cols, "", notes, nil
}

// indexes 读二级索引, 唯一性照搬 (PRAGMA index_list 的 unique 列)。
//
// origin='c' 是显式 CREATE INDEX (普通或唯一); origin='u' 是 UNIQUE 列/表约束
// 自动生成的 sqlite_autoindex_* —— 名字不是用户起的, 但它就是一条唯一索引, 忽略
// 它等于把约束丢掉, 所以照收并改个能重建的名字 (sqlite_ 前缀是保留的, 不能拿来
// 建索引)。origin='pk' 是主键索引, 由 PKColumn 表达, 跳过。
func (d sqliteDial) indexes(db *sql.DB, table string) ([]IndexSchema, error) {
	rows, err := db.Query("PRAGMA index_list(" + d.quote(table) + ")")
	if err != nil {
		return nil, err
	}
	type entry struct {
		name   string
		origin string
		unique bool
	}
	var list []entry
	for rows.Next() {
		var (
			seq, unique, partial int
			name, origin         string
		)
		if err := rows.Scan(&seq, &name, &unique, &origin, &partial); err != nil {
			rows.Close()
			return nil, err
		}
		list = append(list, entry{name: name, origin: origin, unique: unique != 0})
	}
	if err := rows.Err(); err != nil {
		rows.Close()
		return nil, err
	}
	rows.Close()

	var out []IndexSchema
	uqSeq := 0
	for _, e := range list {
		if e.origin != "c" && e.origin != "u" {
			continue
		}
		cols, err := d.indexColumns(db, e.name)
		if err != nil {
			return nil, err
		}
		if len(cols) == 0 {
			continue
		}
		name := e.name
		if e.origin == "u" {
			uqSeq++
			name = fmt.Sprintf("uq_%s_%d", table, uqSeq)
		}
		out = append(out, IndexSchema{Name: name, Type: page.IndexTypeBTree, Columns: cols, Unique: e.unique})
	}
	return out, nil
}

func (d sqliteDial) indexColumns(db *sql.DB, index string) ([]string, error) {
	rows, err := db.Query("PRAGMA index_info(" + d.quote(index) + ")")
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var cols []string
	for rows.Next() {
		var (
			seqno, cid int
			name       string
		)
		if err := rows.Scan(&seqno, &cid, &name); err != nil {
			return nil, err
		}
		cols = append(cols, name)
	}
	return cols, rows.Err()
}

func (d sqliteDial) metaDDL() string {
	return fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (tbl TEXT PRIMARY KEY, payload TEXT NOT NULL)",
		d.quote(MetaTable))
}

func (d sqliteDial) metaUpsert() string {
	return fmt.Sprintf("INSERT INTO %s (tbl, payload) VALUES (?, ?) "+
		"ON CONFLICT(tbl) DO UPDATE SET payload = excluded.payload", d.quote(MetaTable))
}

func (d sqliteDial) metaSelect() string {
	return fmt.Sprintf("SELECT payload FROM %s WHERE tbl = ? LIMIT 1", d.quote(MetaTable))
}

// maxPlaceholders SQLite 的 SQLITE_MAX_VARIABLE_NUMBER 现代版本默认 32766。
func (d sqliteDial) maxPlaceholders() int { return 30000 }

func (d sqliteDial) textLenExpr(col string) string {
	// SQLite 的 length() 对文本数的是【字符】数 (到首个 NUL 为止), 与按字节计容量的
	// 宽度码对不上 —— "25 个汉字" 会采成 25 而实际 75 字节, 建窄的列后面全写不进去。
	// CAST 成 BLOB 后 length() 返回原始字节数, 与宽度码同口径。
	return "LENGTH(CAST(" + d.quote(col) + " AS BLOB))"
}