show.go

/*
 * Copyright 2018 Xiaomi, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package database

import (
	"fmt"
	"github.com/XiaoMi/soar/common"
	"regexp"
	"strconv"
	"strings"
)

// SHOW TABLE STATUS Syntax
// https://dev.mysql.com/doc/refman/5.7/en/show-table-status.html

// TableStatInfo 用以保存 show table status 之后获取的table信息
type TableStatInfo struct {
	Name string
	Rows []tableStatusRow
}

// tableStatusRow 用于 show table status value
// use []byte instead of string, because []byte allow to be null, string not
type tableStatusRow struct {
	Name         string // 表名
	Engine       []byte // 该表使用的存储引擎
	Version      []byte // 该表的 .frm 文件版本号
	RowFormat    []byte // 该表使用的行存储格式
	Rows         int64  // 表行数, InnoDB 引擎中为预估值，甚至可能会有40%~50%的数值偏差
	AvgRowLength int    // 平均行长度

	// MyISAM: Data_length 为数据文件的大小，单位为 bytes
	// InnoDB: Data_length 为聚簇索引分配的近似内存量，单位为 bytes, 计算方式为聚簇索引数量乘以 InnoDB 页面大小
	// 其他不同的存储引擎中该值的意义可能不尽相同
	DataLength int

	// MyISAM: Max_data_length 为数据文件长度的最大值。这是在给定使用的数据指针大小的情况下，可以存储在表中的数据的最大字节数
	// InnoDB: 未使用
	// 其他不同的存储引擎中该值的意义可能不尽相同
	MaxDataLength int

	// MyISAM: Index_length 为 index 文件的大小，单位为 bytes
	// InnoDB: Index_length 为非聚簇索引分配的近似内存量，单位为 bytes，计算方式为非聚簇索引数量乘以 InnoDB 页面大小
	// 其他不同的存储引擎中该值的意义可能不尽相同
	IndexLength int

	DataFree      int    // 已分配但未使用的字节数
	AutoIncrement []byte // 下一个自增值
	CreateTime    []byte // 创建时间
	UpdateTime    []byte // 最近一次更新时间，该值不准确
	CheckTime     []byte // 上次检查时间
	Collation     []byte // 字符集及排序规则信息
	Checksum      []byte // 校验和
	CreateOptions []byte // 创建表的时候的时候一切其他属性
	Comment       []byte // 注释
}

// newTableStat 构造 table Stat 对象
func newTableStat(tableName string) *TableStatInfo {
	return &TableStatInfo{
		Name: tableName,
		Rows: make([]tableStatusRow, 0),
	}
}

// ShowTables 执行 show tables
func (db *Connector) ShowTables() ([]string, error) {
	defer func() {
		err := recover()
		if err != nil {
			common.Log.Error("recover ShowTables()", err)
		}
	}()

	// 执行 show table status
	res, err := db.Query("show tables")
	if err != nil {
		return []string{}, err
	}
	if res.Error != nil {
		return []string{}, res.Error
	}

	// 获取值
	var tables []string
	for res.Rows.Next() {
		var table string
		err = res.Rows.Scan(&table)
		if err != nil {
			return []string{}, err
		}
		tables = append(tables, table)
	}
	return tables, err
}

// ShowTableStatus 执行 show table status
func (db *Connector) ShowTableStatus(tableName string) (*TableStatInfo, error) {
	// 初始化struct
	tbStatus := newTableStat(tableName)

	// 执行 show table status
	res, err := db.Query(fmt.Sprintf("show table status where name = '%s'", tbStatus.Name))
	if err != nil {
		return tbStatus, err
	}
	if res.Error != nil {
		return tbStatus, res.Error
	}

	ts := tableStatusRow{}
	statusFields := make([]interface{}, 0)
	fields := map[string]interface{}{
		"Name":            &ts.Name,
		"Engine":          &ts.Engine,
		"Version":         &ts.Version,
		"Row_format":      &ts.RowFormat,
		"Rows":            &ts.Rows,
		"Avg_row_length":  &ts.AvgRowLength,
		"Data_length":     &ts.DataLength,
		"Max_data_length": &ts.MaxDataLength,
		"Index_length":    &ts.IndexLength,
		"Data_free":       &ts.DataFree,
		"Auto_increment":  &ts.AutoIncrement,
		"Create_time":     &ts.CreateTime,
		"Update_time":     &ts.UpdateTime,
		"Check_time":      &ts.CheckTime,
		"Collation":       &ts.Collation,
		"Checksum":        &ts.Checksum,
		"Create_options":  &ts.CreateOptions,
		"Comment":         &ts.Comment,
	}
	cols, err := res.Rows.Columns()
	common.LogIfError(err, "")
	for _, col := range cols {
		statusFields = append(statusFields, fields[col])
	}
	// 获取值
	for res.Rows.Next() {
		res.Rows.Scan(statusFields...)
		tbStatus.Rows = append(tbStatus.Rows, ts)
	}
	return tbStatus, err
}

// https://dev.mysql.com/doc/refman/5.7/en/show-index.html

// TableIndexInfo 用以保存 show index 之后获取的 index 信息
type TableIndexInfo struct {
	TableName string
	Rows      []TableIndexRow
}

// TableIndexRow 用以存放show index之后获取的每一条index信息
type TableIndexRow struct {
	Table        string // 表名
	NonUnique    int    // 0：unique key，1：not unique
	KeyName      string // index的名称，如果是主键则为 "PRIMARY"
	SeqInIndex   int    // 该列在索引中的位置。计数从 1 开始
	ColumnName   string // 列名
	Collation    string // A or Null
	Cardinality  int    // 索引中唯一值的数量，"ANALYZE TABLE" 可更新该值
	SubPart      int    // 索引前缀字节数
	Packed       int
	Null         string // 表示该列是否可以为空，如果可以为 'YES'，反之''
	IndexType    string // BTREE, FULLTEXT, HASH, RTREE
	Comment      string
	IndexComment string
	Visible      string
}

// NewTableIndexInfo 构造 TableIndexInfo
func NewTableIndexInfo(tableName string) *TableIndexInfo {
	return &TableIndexInfo{
		TableName: tableName,
		Rows:      make([]TableIndexRow, 0),
	}
}

// ShowIndex show Index
func (db *Connector) ShowIndex(tableName string) (*TableIndexInfo, error) {
	tbIndex := NewTableIndexInfo(tableName)

	// 执行 show create table
	res, err := db.Query(fmt.Sprintf("show index from `%s`.`%s`", db.Database, tableName))
	if err != nil {
		return nil, err
	}
	if res.Error != nil {
		return nil, res.Error
	}

	// 获取值
	for res.Rows.Next() {
		var ti TableIndexRow
		res.Rows.Scan(&ti.Table,
			&ti.NonUnique,
			&ti.KeyName,
			&ti.SeqInIndex,
			&ti.ColumnName,
			&ti.Collation,
			&ti.Cardinality,
			&ti.SubPart,
			&ti.Packed,
			&ti.Null,
			&ti.IndexType,
			&ti.Comment,
			&ti.IndexComment,
			&ti.Visible)
		tbIndex.Rows = append(tbIndex.Rows, ti)
	}
	return tbIndex, err
}

// IndexSelectKey 用以对 TableIndexInfo 进行查询
type IndexSelectKey string

// 索引相关
const (
	IndexKeyName    = IndexSelectKey("KeyName")    // 索引名称
	IndexColumnName = IndexSelectKey("ColumnName") // 索引列名称
	IndexIndexType  = IndexSelectKey("IndexType")  // 索引类型
	IndexNonUnique  = IndexSelectKey("NonUnique")  // 唯一索引
)

// FindIndex 获取 TableIndexInfo 中需要的索引
func (tbIndex *TableIndexInfo) FindIndex(arg IndexSelectKey, value string) []TableIndexRow {
	var result []TableIndexRow
	if tbIndex == nil {
		return result
	}

	value = strings.ToLower(value)

	switch arg {
	case IndexKeyName:
		for _, index := range tbIndex.Rows {
			if strings.ToLower(index.KeyName) == value {
				result = append(result, index)
			}
		}

	case IndexColumnName:
		for _, index := range tbIndex.Rows {
			if strings.ToLower(index.ColumnName) == value {
				result = append(result, index)
			}
		}

	case IndexIndexType:
		for _, index := range tbIndex.Rows {
			if strings.ToLower(index.IndexType) == value {
				result = append(result, index)
			}
		}

	case IndexNonUnique:
		for _, index := range tbIndex.Rows {
			unique := strconv.Itoa(index.NonUnique)
			if unique == value {
				result = append(result, index)
			}
		}

	default:
		common.Log.Error("no such args: TableIndexRow")
	}

	return result
}

// desc table
// https://dev.mysql.com/doc/refman/5.7/en/show-columns.html

// TableDesc show columns from rental;
type TableDesc struct {
	Name       string
	DescValues []TableDescValue
}

// TableDescValue 含有每一列的属性
type TableDescValue struct {
	Field      string // 列名
	Type       string // 数据类型
	Collation  []byte // 字符集
	Null       string // 是否有NULL（NO、YES）
	Key        string // 键类型
	Default    []byte // 默认值
	Extra      string // 其他
	Privileges string // 权限
	Comment    string // 备注
}

// NewTableDesc 初始化一个*TableDesc
func NewTableDesc(tableName string) *TableDesc {
	return &TableDesc{
		Name:       tableName,
		DescValues: make([]TableDescValue, 0),
	}
}

// ShowColumns 获取 DB 中所有的 columns
func (db *Connector) ShowColumns(tableName string) (*TableDesc, error) {
	tbDesc := NewTableDesc(tableName)

	// 执行 show create table
	res, err := db.Query(fmt.Sprintf("show full columns from `%s`.`%s`", db.Database, tableName))
	if err != nil {
		return nil, err
	}
	if res.Error != nil {
		return nil, res.Error
	}

	// 获取值
	for res.Rows.Next() {
		var tc TableDescValue
		res.Rows.Scan(&tc.Field,
			&tc.Type,
			&tc.Collation,
			&tc.Null,
			&tc.Key,
			&tc.Default,
			&tc.Extra,
			&tc.Privileges,
			&tc.Comment)
		tbDesc.DescValues = append(tbDesc.DescValues, tc)
	}
	return tbDesc, err
}

// Columns 用于获取TableDesc中所有列的名称
func (td TableDesc) Columns() []string {
	var cols []string
	for _, col := range td.DescValues {
		cols = append(cols, col.Field)
	}
	return cols
}

// showCreate show create
func (db *Connector) showCreate(createType, name string) (string, error) {
	// 执行 show create table
	res, err := db.Query(fmt.Sprintf("show create %s `%s`", createType, name))
	if err != nil {
		return "", err
	}
	if res.Error != nil {
		return "", res.Error
	}

	// 获取 CREATE TABLE 语句
	var tableName, createTable string
	for res.Rows.Next() {
		res.Rows.Scan(&tableName, &createTable)
	}

	return createTable, err
}

// ShowCreateDatabase show create database
func (db *Connector) ShowCreateDatabase(dbName string) (string, error) {
	defer func() {
		err := recover()
		if err != nil {
			common.Log.Error("recover ShowCreateDatabase()", err)
		}
	}()
	return db.showCreate("database", dbName)
}

// ShowCreateTable show create table
func (db *Connector) ShowCreateTable(tableName string) (string, error) {
	defer func() {
		err := recover()
		if err != nil {
			common.Log.Error("recover ShowCreateTable()", err)
		}
	}()

	ddl, err := db.showCreate("table", tableName)

	// 去除外键关联条件
	var noConstraint []string
	relationReg, _ := regexp.Compile("CONSTRAINT")
	for _, line := range strings.Split(ddl, "\n") {

		if relationReg.Match([]byte(line)) {
			continue
		}

		// 去除外键语句会使DDL中多一个','导致语法错误，要把多余的逗号去除
		if strings.Index(line, ")") == 0 {
			lineWrongSyntax := noConstraint[len(noConstraint)-1]
			// 如果')'前一句的末尾是',' 删除 ',' 保证语法正确性
			if strings.Index(lineWrongSyntax, ",") == len(lineWrongSyntax)-1 {
				noConstraint[len(noConstraint)-1] = lineWrongSyntax[:len(lineWrongSyntax)-1]
			}
		}

		noConstraint = append(noConstraint, line)
	}

	return strings.Join(noConstraint, "\n"), err
}

// FindColumn find column
func (db *Connector) FindColumn(name, dbName string, tables ...string) ([]*common.Column, error) {
	// 执行 show create table
	var columns []*common.Column
	sql := fmt.Sprintf("SELECT "+
		"c.TABLE_NAME,c.TABLE_SCHEMA,c.COLUMN_TYPE,c.CHARACTER_SET_NAME, c.COLLATION_NAME "+
		"FROM `INFORMATION_SCHEMA`.`COLUMNS` as c where c.COLUMN_NAME = '%s' ", name)

	if dbName != "" {
		sql += fmt.Sprintf(" and c.table_schema = '%s'", dbName)
	}

	if len(tables) > 0 {
		var tmp []string
		for _, table := range tables {
			tmp = append(tmp, "'"+table+"'")
		}
		sql += fmt.Sprintf(" and c.table_name in (%s)", strings.Join(tmp, ","))
	}

	common.Log.Debug("FindColumn, execute SQL: %s", sql)
	res, err := db.Query(sql)
	if err != nil {
		common.Log.Error("(db *Connector) FindColumn Error : ", err)
		return columns, err
	}
	if res.Error != nil {
		common.Log.Error("(db *Connector) FindColumn Error : ", res.Error)
		return columns, res.Error
	}

	var col common.Column
	for res.Rows.Next() {
		res.Rows.Scan(&col.Table,
			&col.DB,
			&col.DataType,
			&col.Character,
			&col.Collation)

		// 填充字符集和排序规则
		if col.Character == "" {
			// 当从`INFORMATION_SCHEMA`.`COLUMNS`表中查询不到相关列的character和collation的信息时
			// 认为该列使用的character和collation与其所处的表一致
			// 由于`INFORMATION_SCHEMA`.`TABLES`表中未找到表的character，所以从按照MySQL中collation的规则从中截取character

			sql = fmt.Sprintf("SELECT `t`.`TABLE_COLLATION` FROM `INFORMATION_SCHEMA`.`TABLES` AS `t` "+
				"WHERE `t`.`TABLE_NAME`='%s' AND `t`.`TABLE_SCHEMA` = '%s'", col.Table, col.DB)

			common.Log.Debug("FindColumn, execute SQL: %s", sql)
			var newRes QueryResult
			newRes, err = db.Query(sql)
			if err != nil {
				common.Log.Error("(db *Connector) FindColumn Error : ", err)
				return columns, err
			}
			if res.Error != nil {
				common.Log.Error("(db *Connector) FindColumn Error : ", res.Error)
				return columns, res.Error
			}

			var tbCollation string
			if newRes.Rows.Next() {
				newRes.Rows.Scan(&tbCollation)
			}
			if tbCollation != "" {
				col.Character = strings.Split(tbCollation, "_")[0]
				col.Collation = tbCollation
			}
		}
		columns = append(columns, &col)
	}
	return columns, err
}

// IsForeignKey 判断列是否是外键
func (db *Connector) IsForeignKey(dbName, tbName, column string) bool {
	sql := fmt.Sprintf("SELECT REFERENCED_COLUMN_NAME FROM INFORMATION_SCHEMA.KEY_COLUMN_USAGE C "+
		"WHERE REFERENCED_TABLE_SCHEMA <> 'NULL' AND"+
		" TABLE_NAME='%s' AND"+
		" TABLE_SCHEMA='%s' AND"+
		" COLUMN_NAME='%s'", tbName, dbName, column)

	common.Log.Debug("IsForeignKey, execute SQL: %s", sql)
	res, err := db.Query(sql)
	if err != nil {
		common.Log.Error("IsForeignKey, Error: %s", err.Error())
		return false
	}
	if res.Error != nil {
		common.Log.Error("IsForeignKey, Error: %s", res.Error.Error())
		return false
	}
	if res.Rows.Next() {
		return true
	}

	return false
}

// Reference 用于存储关系
type Reference map[string][]ReferenceValue

// ReferenceValue 用于处理表之间的关系
type ReferenceValue struct {
	ReferencedTableSchema string // 夫表所属数据库
	ReferencedTableName   string // 父表
	TableSchema           string // 子表所属数据库
	TableName             string // 子表
	ConstraintName        string // 关系名称
}

// ShowReference 查找所有的外键信息
func (db *Connector) ShowReference(dbName string, tbName ...string) ([]ReferenceValue, error) {
	var referenceValues []ReferenceValue
	sql := `SELECT C.REFERENCED_TABLE_SCHEMA,C.REFERENCED_TABLE_NAME,C.TABLE_SCHEMA,C.TABLE_NAME,C.CONSTRAINT_NAME
FROM INFORMATION_SCHEMA.KEY_COLUMN_USAGE C JOIN INFORMATION_SCHEMA. TABLES T ON T.TABLE_NAME = C.TABLE_NAME
WHERE C.REFERENCED_TABLE_NAME IS NOT NULL`
	sql = sql + fmt.Sprintf(` AND C.TABLE_SCHEMA = "%s"`, dbName)

	if len(tbName) > 0 {
		extra := fmt.Sprintf(` AND C.TABLE_NAME IN ("%s")`, strings.Join(tbName, `","`))
		sql = sql + extra
	}

	common.Log.Debug("ShowReference, execute SQL: %s", sql)
	// 执行SQL查找外键关联关系
	res, err := db.Query(sql)
	if err != nil {
		return referenceValues, err
	}
	if res.Error != nil {
		return referenceValues, res.Error
	}

	// 获取值
	for res.Rows.Next() {
		var rv ReferenceValue
		res.Rows.Scan(&rv.ReferencedTableSchema,
			&rv.ReferencedTableName,
			&rv.TableSchema,
			&rv.TableName,
			&rv.ConstraintName)
		referenceValues = append(referenceValues, rv)
	}

	return referenceValues, err
}