464 lines
16 KiB
Go
464 lines
16 KiB
Go
package mysql
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"fmt"
|
|
"regexp"
|
|
"strings"
|
|
|
|
mysqlDriver "github.com/go-sql-driver/mysql"
|
|
|
|
"git.warky.dev/wdevs/relspecgo/pkg/mariadb"
|
|
"git.warky.dev/wdevs/relspecgo/pkg/models"
|
|
"git.warky.dev/wdevs/relspecgo/pkg/readers"
|
|
)
|
|
|
|
type Reader struct {
|
|
options *readers.ReaderOptions
|
|
db *sql.DB
|
|
ctx context.Context
|
|
}
|
|
|
|
func NewReader(options *readers.ReaderOptions) *Reader {
|
|
return &Reader{options: options, ctx: context.Background()}
|
|
}
|
|
|
|
func (r *Reader) ReadDatabase() (*models.Database, error) {
|
|
if r.options == nil || r.options.ConnectionString == "" {
|
|
return nil, fmt.Errorf("connection string is required")
|
|
}
|
|
if err := r.connect(); err != nil {
|
|
return nil, fmt.Errorf("failed to connect: %w", err)
|
|
}
|
|
defer r.close()
|
|
var name, version, sqlMode string
|
|
if err := r.db.QueryRowContext(r.ctx, "SELECT DATABASE(), @@SESSION.sql_mode").Scan(&name, &sqlMode); err != nil {
|
|
return nil, fmt.Errorf("failed to get database name: %w", err)
|
|
}
|
|
_ = r.db.QueryRowContext(r.ctx, "SELECT VERSION()").Scan(&version)
|
|
db := models.InitDatabase(name)
|
|
db.Metadata["mysql_sql_mode"] = sqlMode
|
|
db.DatabaseType, db.SourceFormat, db.DatabaseVersion = models.MySQLDatabaseType, "mysql", version
|
|
schemas, err := r.querySchemas(name)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to query schemas: %w", err)
|
|
}
|
|
for _, schema := range schemas {
|
|
tables, err := r.queryTables(schema.Name)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
schema.Tables = tables
|
|
schema.Metadata["mysql_sql_mode"] = sqlMode
|
|
for _, table := range tables {
|
|
table.Metadata["mysql_sql_mode"] = sqlMode
|
|
table.Columns, err = r.queryColumns(schema.Name, table)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
table.Constraints, err = r.queryConstraints(schema.Name, table.Name)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
table.Indexes, err = r.queryIndexes(schema.Name, table.Name)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
table.RefSchema = schema
|
|
for _, c := range table.Constraints {
|
|
if c.Type == models.PrimaryKeyConstraint {
|
|
for _, n := range c.Columns {
|
|
if col := table.Columns[n]; col != nil {
|
|
col.IsPrimaryKey = true
|
|
}
|
|
}
|
|
}
|
|
}
|
|
for _, c := range table.Constraints {
|
|
if c.Type == models.ForeignKeyConstraint {
|
|
r.deriveRelationship(table, c)
|
|
}
|
|
}
|
|
}
|
|
schema.RefDatabase = db
|
|
db.Schemas = append(db.Schemas, schema)
|
|
}
|
|
if err := r.readUnsupported(db, name); err != nil {
|
|
return nil, err
|
|
}
|
|
return db, nil
|
|
}
|
|
|
|
func (r *Reader) ReadSchema() (*models.Schema, error) {
|
|
db, err := r.ReadDatabase()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if len(db.Schemas) == 0 {
|
|
return nil, fmt.Errorf("no schemas found in database")
|
|
}
|
|
return db.Schemas[0], nil
|
|
}
|
|
|
|
func (r *Reader) ReadTable() (*models.Table, error) {
|
|
s, err := r.ReadSchema()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if len(s.Tables) == 0 {
|
|
return nil, fmt.Errorf("no tables found in schema")
|
|
}
|
|
return s.Tables[0], nil
|
|
}
|
|
|
|
func (r *Reader) connect() error {
|
|
cfg, err := mysqlDriver.ParseDSN(r.options.ConnectionString)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if cfg.Params == nil {
|
|
cfg.Params = map[string]string{}
|
|
}
|
|
// SHOW CREATE parsing requires quoted column identifiers, including when the
|
|
// caller's metadata connection would otherwise disable show-create quoting.
|
|
cfg.Params["sql_quote_show_create"] = "1"
|
|
db, err := sql.Open("mysql", cfg.FormatDSN())
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err = db.PingContext(r.ctx); err != nil {
|
|
db.Close()
|
|
return err
|
|
}
|
|
r.db = db
|
|
return nil
|
|
}
|
|
|
|
func (r *Reader) close() {
|
|
if r.db != nil {
|
|
_ = r.db.Close()
|
|
}
|
|
}
|
|
func (r *Reader) mapDataType(t string) string { return mariadb.ConvertMariaDBToCanonical(t) }
|
|
|
|
func (r *Reader) querySchemas(current string) ([]*models.Schema, error) {
|
|
rows, err := r.db.QueryContext(r.ctx, "SELECT SCHEMA_NAME FROM information_schema.SCHEMATA WHERE SCHEMA_NAME = ?", current)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var out []*models.Schema
|
|
for rows.Next() {
|
|
var n string
|
|
if err := rows.Scan(&n); err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, models.InitSchema(n))
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
func (r *Reader) queryTables(schema string) ([]*models.Table, error) {
|
|
rows, err := r.db.QueryContext(r.ctx, "SELECT TABLE_NAME, ENGINE, TABLE_COLLATION, TABLE_COMMENT, CREATE_OPTIONS, AUTO_INCREMENT FROM information_schema.TABLES WHERE TABLE_SCHEMA = ? AND TABLE_TYPE = 'BASE TABLE' ORDER BY TABLE_NAME", schema)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var out []*models.Table
|
|
for rows.Next() {
|
|
var n, engine, collation, comment, options string
|
|
var auto sql.NullString
|
|
if err := rows.Scan(&n, &engine, &collation, &comment, &options, &auto); err != nil {
|
|
return nil, err
|
|
}
|
|
t := models.InitTable(n, schema)
|
|
t.Comment = comment
|
|
t.Metadata["mysql_engine"], t.Metadata["mysql_collation"], t.Metadata["mysql_create_options"] = engine, collation, options
|
|
if auto.Valid {
|
|
t.Metadata["mysql_auto_increment"] = auto.String
|
|
}
|
|
out = append(out, t)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
var onUpdateExtra = regexp.MustCompile(`on update current_timestamp(?:\(\d*\))?`)
|
|
|
|
func (r *Reader) queryColumns(schema string, table *models.Table) (map[string]*models.Column, error) {
|
|
rows, err := r.db.QueryContext(r.ctx, `SELECT COLUMN_NAME, COLUMN_TYPE, IS_NULLABLE, COLUMN_DEFAULT, ORDINAL_POSITION, EXTRA, COLUMN_COMMENT,
|
|
CHARACTER_MAXIMUM_LENGTH, NUMERIC_PRECISION, NUMERIC_SCALE, DATETIME_PRECISION, COLLATION_NAME, CHARACTER_SET_NAME, GENERATION_EXPRESSION, SRS_ID
|
|
FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=? AND TABLE_NAME=? ORDER BY ORDINAL_POSITION`, schema, table.Name)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("columns %s.%s: %w", schema, table.Name, err)
|
|
}
|
|
defer rows.Close()
|
|
out := map[string]*models.Column{}
|
|
for rows.Next() {
|
|
var name, typ, nullable, extra, comment string
|
|
var def, collation, charset, generation sql.NullString
|
|
var length, precision, scale, datetime, srid sql.NullInt64
|
|
var pos int
|
|
if err := rows.Scan(&name, &typ, &nullable, &def, &pos, &extra, &comment, &length, &precision, &scale, &datetime, &collation, &charset, &generation, &srid); err != nil {
|
|
return nil, err
|
|
}
|
|
c := models.InitColumn(name, table.Name, schema)
|
|
c.Type, c.NotNull, c.Sequence, c.Comment = r.mapDataType(typ), nullable == "NO", uint(pos), comment
|
|
// The native declaration preserves mediumint, enum/set, binary/blob sizes,
|
|
// unsigned/zerofill and timestamp versus datetime. Canonical dimensions remain editable.
|
|
c.Length, c.Precision, c.Scale = int(length.Int64), int(precision.Int64), int(scale.Int64)
|
|
if datetime.Valid {
|
|
c.Precision = int(datetime.Int64)
|
|
}
|
|
c.Collation = collation.String
|
|
c.Metadata["mysql_type"] = typ
|
|
c.Metadata["mysql_type_signature"] = typeSignature(c)
|
|
c.Metadata["mysql_charset"] = charset.String
|
|
c.Metadata["mysql_datetime_precision"] = datetime.Int64
|
|
c.Metadata["mysql_default_expression"] = strings.Contains(extra, "DEFAULT_GENERATED") || strings.HasPrefix(strings.ToLower(def.String), "current_timestamp") && (c.Type == "timestamp")
|
|
if def.Valid {
|
|
c.Default = def.String
|
|
if strings.Contains(extra, "DEFAULT_GENERATED") {
|
|
c.Default = unescapeExpression(def.String)
|
|
}
|
|
if strings.HasPrefix(typ, "bit(") && def.Valid {
|
|
c.Metadata["mysql_default_expression"] = true
|
|
}
|
|
}
|
|
c.AutoIncrement = strings.Contains(extra, "auto_increment")
|
|
c.Generated, c.GenerationExpression = generation.String != "", unescapeExpression(generation.String)
|
|
if strings.Contains(extra, "STORED GENERATED") {
|
|
c.Metadata["mysql_generation_storage"] = "STORED"
|
|
} else if c.Generated {
|
|
c.Metadata["mysql_generation_storage"] = "VIRTUAL"
|
|
}
|
|
c.Metadata["mysql_invisible"] = strings.Contains(extra, "INVISIBLE")
|
|
remaining := onUpdateExtra.ReplaceAllString(strings.ToLower(extra), "")
|
|
for _, known := range []string{"auto_increment", "default_generated", "stored generated", "virtual generated", "invisible"} {
|
|
remaining = strings.ReplaceAll(remaining, known, "")
|
|
}
|
|
if remaining = strings.TrimSpace(remaining); remaining != "" {
|
|
c.Metadata["mysql_unsupported_extra"] = remaining
|
|
}
|
|
if i := strings.Index(strings.ToLower(extra), "on update "); i >= 0 {
|
|
c.Metadata["mysql_on_update"] = strings.Fields(extra[i+10:])[0]
|
|
}
|
|
if srid.Valid {
|
|
c.Metadata["mysql_srs_id"] = srid.Int64
|
|
}
|
|
out[name] = c
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
var tableName, create string
|
|
qualified := "`" + strings.ReplaceAll(schema, "`", "``") + "`.`" + strings.ReplaceAll(table.Name, "`", "``") + "`"
|
|
if err := r.db.QueryRowContext(r.ctx, "SHOW CREATE TABLE "+qualified).Scan(&tableName, &create); err != nil {
|
|
return nil, fmt.Errorf("native defaults %s.%s: %w", schema, table.Name, err)
|
|
}
|
|
// information_schema AUTO_INCREMENT may be stale because of statistics caching.
|
|
for _, c := range out {
|
|
if c.AutoIncrement {
|
|
table.Metadata["mysql_auto_increment"] = autoIncrementValue(create)
|
|
break
|
|
}
|
|
}
|
|
if options := unsupportedNativeOptions(create); len(options) > 0 {
|
|
table.Metadata["mysql_unsupported_options"] = strings.Join(options, ", ")
|
|
}
|
|
for name, sql := range columnDefaults(create) {
|
|
c := out[name]
|
|
if c == nil || c.Default == nil {
|
|
continue
|
|
}
|
|
if c.Type == "bytea" {
|
|
c.Default = sql
|
|
}
|
|
c.Metadata["mysql_default_sql"] = sql
|
|
c.Metadata["mysql_default_value"] = fmt.Sprint(c.Default)
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func typeSignature(c *models.Column) string {
|
|
return fmt.Sprintf("%s/%d/%d/%d", c.Type, c.Length, c.Precision, c.Scale)
|
|
}
|
|
|
|
func (r *Reader) queryConstraints(schema, table string) (map[string]*models.Constraint, error) {
|
|
rows, err := r.db.QueryContext(r.ctx, `SELECT k.CONSTRAINT_NAME, t.CONSTRAINT_TYPE, k.COLUMN_NAME,
|
|
k.REFERENCED_TABLE_SCHEMA, k.REFERENCED_TABLE_NAME, k.REFERENCED_COLUMN_NAME, k.ORDINAL_POSITION,
|
|
rc.DELETE_RULE, rc.UPDATE_RULE FROM information_schema.KEY_COLUMN_USAGE k
|
|
JOIN information_schema.TABLE_CONSTRAINTS t USING (CONSTRAINT_SCHEMA,TABLE_NAME,CONSTRAINT_NAME)
|
|
LEFT JOIN information_schema.REFERENTIAL_CONSTRAINTS rc ON rc.CONSTRAINT_SCHEMA=k.CONSTRAINT_SCHEMA AND rc.TABLE_NAME=k.TABLE_NAME AND rc.CONSTRAINT_NAME=k.CONSTRAINT_NAME
|
|
WHERE k.TABLE_SCHEMA=? AND k.TABLE_NAME=? ORDER BY k.CONSTRAINT_NAME,k.ORDINAL_POSITION`, schema, table)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("constraints %s.%s: %w", schema, table, err)
|
|
}
|
|
out := map[string]*models.Constraint{}
|
|
for rows.Next() {
|
|
var name, typ string
|
|
var col, rs, rt, rc, del, upd sql.NullString
|
|
var pos int
|
|
if err := rows.Scan(&name, &typ, &col, &rs, &rt, &rc, &pos, &del, &upd); err != nil {
|
|
rows.Close()
|
|
return nil, err
|
|
}
|
|
c := out[name]
|
|
if c == nil {
|
|
ct := models.UniqueConstraint
|
|
if typ == "PRIMARY KEY" {
|
|
ct = models.PrimaryKeyConstraint
|
|
}
|
|
if typ == "FOREIGN KEY" {
|
|
ct = models.ForeignKeyConstraint
|
|
}
|
|
c = models.InitConstraint(name, ct)
|
|
c.Schema, c.Table = schema, table
|
|
c.ReferencedSchema, c.ReferencedTable = rs.String, rt.String
|
|
c.OnDelete, c.OnUpdate = del.String, upd.String
|
|
out[name] = c
|
|
}
|
|
if col.Valid {
|
|
c.Columns = append(c.Columns, col.String)
|
|
}
|
|
if rc.Valid {
|
|
c.ReferencedColumns = append(c.ReferencedColumns, rc.String)
|
|
}
|
|
}
|
|
err = rows.Err()
|
|
rows.Close()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
rows, err = r.db.QueryContext(r.ctx, `SELECT t.CONSTRAINT_NAME,c.CHECK_CLAUSE,t.ENFORCED FROM information_schema.TABLE_CONSTRAINTS t JOIN information_schema.CHECK_CONSTRAINTS c USING (CONSTRAINT_SCHEMA,CONSTRAINT_NAME) WHERE t.TABLE_SCHEMA=? AND t.TABLE_NAME=? AND t.CONSTRAINT_TYPE='CHECK' ORDER BY t.CONSTRAINT_NAME`, schema, table)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("checks %s.%s (requires MySQL 8.0.16+): %w", schema, table, err)
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var name, expr, enforced string
|
|
if err := rows.Scan(&name, &expr, &enforced); err != nil {
|
|
return nil, err
|
|
}
|
|
c := models.InitConstraint(name, models.CheckConstraint)
|
|
c.Schema, c.Table, c.Expression = schema, table, unescapeExpression(expr)
|
|
c.Metadata = map[string]any{"mysql_enforced": enforced == "YES"}
|
|
out[name] = c
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
func (r *Reader) queryIndexes(schema, table string) (map[string]*models.Index, error) {
|
|
rows, err := r.db.QueryContext(r.ctx, `SELECT INDEX_NAME, NON_UNIQUE, COLUMN_NAME, SEQ_IN_INDEX, INDEX_TYPE, EXPRESSION, SUB_PART, COLLATION, IS_VISIBLE, INDEX_COMMENT FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=? AND TABLE_NAME=? ORDER BY INDEX_NAME,SEQ_IN_INDEX`, schema, table)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("indexes %s.%s (requires MySQL 8.0.13+): %w", schema, table, err)
|
|
}
|
|
defer rows.Close()
|
|
out := map[string]*models.Index{}
|
|
for rows.Next() {
|
|
var name, typ, visible, comment string
|
|
var col, expr, direction sql.NullString
|
|
var prefix sql.NullInt64
|
|
var non, seq int
|
|
if err := rows.Scan(&name, &non, &col, &seq, &typ, &expr, &prefix, &direction, &visible, &comment); err != nil {
|
|
return nil, err
|
|
}
|
|
i := out[name]
|
|
if i == nil {
|
|
i = models.InitIndex(name, table, schema)
|
|
i.Unique = non == 0
|
|
i.Type = strings.ToLower(typ)
|
|
i.Comment = comment
|
|
i.Metadata["mysql_visible"] = visible == "YES"
|
|
out[name] = i
|
|
}
|
|
i.Columns = append(i.Columns, col.String)
|
|
i.KeyParts = append(i.KeyParts, models.IndexPart{Name: col.String, Expression: unescapeExpression(expr.String), Length: int(prefix.Int64), Descending: direction.String == "D"})
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
// Record objects outside the supported table-schema contract rather than silently
|
|
// discarding them. Writers can warn or reject the conversion before any output.
|
|
func (r *Reader) readUnsupported(db *models.Database, schema string) error {
|
|
queries := []struct{ kind, sql string }{
|
|
{"view", `SELECT TABLE_NAME FROM information_schema.VIEWS WHERE TABLE_SCHEMA=?`},
|
|
{"trigger", `SELECT TRIGGER_NAME FROM information_schema.TRIGGERS WHERE TRIGGER_SCHEMA=?`},
|
|
{"routine", `SELECT ROUTINE_NAME FROM information_schema.ROUTINES WHERE ROUTINE_SCHEMA=?`},
|
|
{"event", `SELECT EVENT_NAME FROM information_schema.EVENTS WHERE EVENT_SCHEMA=?`},
|
|
{"partitioned table", `SELECT DISTINCT TABLE_NAME FROM information_schema.PARTITIONS WHERE TABLE_SCHEMA=? AND PARTITION_NAME IS NOT NULL`},
|
|
}
|
|
var unsupported []string
|
|
for _, q := range queries {
|
|
rows, err := r.db.QueryContext(r.ctx, q.sql, schema)
|
|
if err != nil {
|
|
return fmt.Errorf("discover %s: %w", q.kind, err)
|
|
}
|
|
for rows.Next() {
|
|
var name string
|
|
if err := rows.Scan(&name); err != nil {
|
|
rows.Close()
|
|
return err
|
|
}
|
|
unsupported = append(unsupported, q.kind+" "+schema+"."+name)
|
|
}
|
|
err = rows.Err()
|
|
rows.Close()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if len(unsupported) > 0 {
|
|
db.Metadata["mysql_unsupported"] = unsupported
|
|
for _, s := range db.Schemas {
|
|
s.Metadata["mysql_unsupported"] = unsupported
|
|
}
|
|
if r.options.Progress != nil {
|
|
for _, item := range unsupported {
|
|
r.options.Progress("warning: unsupported MySQL " + item)
|
|
}
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (r *Reader) deriveRelationship(t *models.Table, c *models.Constraint) {
|
|
n := fmt.Sprintf("%s_%s", t.Name, c.Name)
|
|
rel := models.InitRelationship(n, models.OneToMany)
|
|
rel.FromTable, rel.FromSchema = t.Name, t.Schema
|
|
rel.FromColumns = append([]string(nil), c.Columns...)
|
|
rel.ToTable, rel.ToSchema = c.ReferencedTable, c.ReferencedSchema
|
|
rel.ToColumns = append([]string(nil), c.ReferencedColumns...)
|
|
rel.ForeignKey = c.Name
|
|
t.Relationships[n] = rel
|
|
}
|
|
|
|
// MySQL information_schema escapes SQL expressions by an additional layer
|
|
// (including quote delimiters). This covers defaults, generated columns, checks,
|
|
// and functional indexes. Literal defaults are already values and bypass this.
|
|
func unescapeExpression(s string) string {
|
|
var out strings.Builder
|
|
for i := 0; i < len(s); i++ {
|
|
if s[i] == '\\' && i+1 < len(s) {
|
|
i++
|
|
switch s[i] {
|
|
case 'n':
|
|
out.WriteByte('\n')
|
|
case 'r':
|
|
out.WriteByte('\r')
|
|
case 't':
|
|
out.WriteByte('\t')
|
|
case '0':
|
|
out.WriteByte(0)
|
|
case 'Z':
|
|
out.WriteByte(26)
|
|
default:
|
|
out.WriteByte(s[i])
|
|
}
|
|
} else {
|
|
out.WriteByte(s[i])
|
|
}
|
|
}
|
|
return out.String()
|
|
}
|