Merge pull request #175 from soul-F/feature-musthavecolumnindex

添加审核项 columns_must_have_index

config/config.go

@@ -359,6 +359,8 @@ type Inc struct {
 
 	// 建表必须创建的列. 可指定多个列,以逗号分隔.列类型可选. 格式: 列名 [列类型,可选],...
 	MustHaveColumns string `toml:"must_have_columns" json:"must_have_columns"`
+	// 如果表包含以下列，列必须有索引。可指定多个列,以逗号分隔.列类型可选.   格式: 列名 [列类型,可选],...
+	ColumnsMustHaveIndex string `toml:"columns_must_have_index" json:"columns_must_have_index"`
 
 	// 是否跳过用户权限校验
 	SkipGrantTable bool `toml:"skip_grant_table" json:"skip_grant_table"`
@@ -621,6 +623,8 @@ type IncLevel struct {
 	ER_INVALID_DATA_TYPE            int8 `toml:"er_invalid_data_type"`
 	ER_INVALID_IDENT                int8 `toml:"er_invalid_ident"`
 	ER_MUST_HAVE_COLUMNS            int8 `toml:"er_must_have_columns"`
+	ErrColumnsMustHaveIndex         int8 `toml:"er_columns_must_have_index"`
+	ErrColumnsMustHaveIndexTypeErr  int8 `toml:"er_columns_must_have_index_type_err"`
 	ER_NO_WHERE_CONDITION           int8 `toml:"er_no_where_condition"`
 	ER_NOT_ALLOWED_NULLABLE         int8 `toml:"er_not_allowed_nullable"`
 	ER_ORDERY_BY_RAND               int8 `toml:"er_ordery_by_rand"`
@@ -833,6 +837,8 @@ var defaultConf = Config{
 		ER_INVALID_DATA_TYPE:            1,
 		ER_INVALID_IDENT:                1,
 		ER_MUST_HAVE_COLUMNS:            1,
+		ErrColumnsMustHaveIndex:         1,
+		ErrColumnsMustHaveIndexTypeErr:  1,
 		ER_NO_WHERE_CONDITION:           1,
 		ER_NOT_ALLOWED_NULLABLE:         1,
 		ER_ORDERY_BY_RAND:               1,

config/config.toml.default

@@ -198,6 +198,8 @@ er_invalid_ident = 1
 er_join_no_on_condition = 1
 er_json_type_support = 2
 er_must_have_columns = 1
+er_columns_must_have_index = 1
+er_columns_must_have_index_type_err = 1
 er_no_where_condition = 1
 er_not_allowed_nullable = 1
 er_ordery_by_rand = 1

config/config.toml.example

@@ -327,6 +327,8 @@ er_invalid_ident = 1
 er_join_no_on_condition = 1
 er_json_type_support = 2
 er_must_have_columns = 1
+er_columns_must_have_index = 1
+er_columns_must_have_index_type_err = 1
 er_no_where_condition = 1
 er_not_allowed_nullable = 1
 er_ordery_by_rand = 1

session/errors.go

@@ -177,6 +177,8 @@ const (
 	ER_CANT_SET_ENGINE
 	ER_MUST_AT_LEAST_ONE_COLUMN
 	ER_MUST_HAVE_COLUMNS
+	ErrColumnsMustHaveIndex
+	ErrColumnsMustHaveIndexTypeErr
 	ER_PRIMARY_CANT_HAVE_NULL
 	ErrCantRemoveAllFields
 	ErrNotFoundTableInfo
@@ -348,6 +350,8 @@ var ErrorsDefault = map[ErrorCode]string{
 	ER_CANT_SET_ENGINE:                     "Cannot set engine '%s'",
 	ER_MUST_AT_LEAST_ONE_COLUMN:            "A table must have at least 1 column.",
 	ER_MUST_HAVE_COLUMNS:                   "Must have the specified column: '%s'.",
+	ErrColumnsMustHaveIndex:                "The specified column: '%s' must have index.",
+	ErrColumnsMustHaveIndexTypeErr:         "The specified column: '%s' type must be '%s',current is '%s'.",
 	ER_PRIMARY_CANT_HAVE_NULL:              "All parts of a PRIMARY KEY must be NOT NULL; if you need NULL in a key, use UNIQUE instead",
 	ErrCantRemoveAllFields:                 "You can't delete all columns with ALTER TABLE; use DROP TABLE instead",
 	ErrNotFoundTableInfo:                   "Skip backup because there is no table structure information.",
@@ -519,6 +523,8 @@ var ErrorsChinese = map[ErrorCode]string{
 	ER_CANT_SET_ENGINE:                     "禁止指定存储引擎:'%s'",
 	ER_MUST_AT_LEAST_ONE_COLUMN:            "表至少需要有一个列.",
 	ER_MUST_HAVE_COLUMNS:                   "表必须包含以下列: '%s'.",
+	ErrColumnsMustHaveIndex:                "列: '%s' 必须建索引.",
+	ErrColumnsMustHaveIndexTypeErr:         "列: '%s' 类型必须为 '%s',当前为 '%s'",
 	ER_PRIMARY_CANT_HAVE_NULL:              "主键的所有列必须为NOT NULL,如需要NULL列,请改用唯一索引",
 	ErrCantRemoveAllFields:                 "禁止删除表的所有列.",
 	ErrNotFoundTableInfo:                   "没有表结构信息,跳过备份.",
@@ -567,6 +573,8 @@ func GetErrorLevel(code ErrorCode) uint8 {
 		ER_INVALID_DATA_TYPE,
 		ER_INVALID_IDENT,
 		ER_MUST_HAVE_COLUMNS,
+		ErrColumnsMustHaveIndex,
+		ErrColumnsMustHaveIndexTypeErr,
 		ER_NO_WHERE_CONDITION,
 		ErrJoinNoOnCondition,
 		ER_NOT_ALLOWED_NULLABLE,
@@ -983,6 +991,10 @@ func (e ErrorCode) String() string {
 		return "er_must_at_least_one_column"
 	case ER_MUST_HAVE_COLUMNS:
 		return "er_must_have_columns"
+	case ErrColumnsMustHaveIndex:
+		return "er_columns_must_have_index"
+	case ErrColumnsMustHaveIndexTypeErr:
+		return "er_columns_must_have_index_type_err"
 	case ER_PRIMARY_CANT_HAVE_NULL:
 		return "er_primary_cant_have_null"
 	case ErrCantRemoveAllFields:

session/session_inception.go

@@ -3585,6 +3585,12 @@ func (s *session) checkCreateTable(node *ast.CreateTableStmt, sql string) {
 									{Column: field.Name,
 										Length: types.UnspecifiedLength},
 								}, nil, table, true, ast.ConstraintPrimaryKey)
+						case ast.ColumnOptionUniqKey:
+							s.checkCreateIndex(nil, field.Name.String(),
+								[]*ast.IndexColName{
+									{Column: field.Name, Length: types.UnspecifiedLength},
+								}, nil, table, true, ast.ConstraintUniq)
+
 						}
 					}
 
@@ -3680,6 +3686,10 @@ func (s *session) checkCreateTable(node *ast.CreateTableStmt, sql string) {
 		}
 	}
 
+	if s.Inc.ColumnsMustHaveIndex != "" {
+		s.checkColumnsMustHaveindex(table)
+	}
+
 	if !s.hasError() && s.opt.execute {
 		s.myRecord.DDLRollback = fmt.Sprintf("DROP TABLE `%s`.`%s`;", table.Schema, table.Name)
 	}
@@ -3762,6 +3772,57 @@ func (s *session) checkMustHaveColumns(table *TableInfo) {
 	}
 }
 
+func (s *session) checkColumnsMustHaveindex(table *TableInfo) {
+	columns := strings.Split(s.Inc.ColumnsMustHaveIndex, ",")
+	if len(columns) == 0 {
+		return
+	}
+	if table == nil {
+		return
+	}
+	var mustHaveNotHaveIndexCol []string
+	for _, mustIndexCol := range columns {
+		mustIndexCol = strings.TrimSpace(mustIndexCol)
+		col_name := mustIndexCol
+		col_type := ""
+		if strings.Contains(mustIndexCol, " ") {
+			column_name_type := strings.Fields(mustIndexCol)
+			if len(column_name_type) > 1 {
+				col_name = column_name_type[0]
+				col_type = GetDataTypeBase(column_name_type[1])
+			}
+		}
+
+		inTable := false
+		haveIndex := false
+		for _, field := range table.Fields {
+			//表内包含必须有索引的列
+			if strings.EqualFold(col_name, field.Field) {
+				inTable = true
+				for _, indexColName := range table.Indexes {
+					if strings.EqualFold(col_name, indexColName.ColumnName) && indexColName.Seq == 1 {
+						haveIndex = true
+					}
+				}
+
+				if col_type != "" && !strings.EqualFold(col_type, GetDataTypeBase(field.Type)) {
+					s.AppendErrorNo(ErrColumnsMustHaveIndexTypeErr, col_name, col_type, GetDataTypeBase(field.Type))
+				}
+			}
+		}
+
+		//col_name 在表中，并且没有索引
+		if inTable == true && haveIndex == false {
+			mustHaveNotHaveIndexCol = append(mustHaveNotHaveIndexCol, col_name)
+		}
+	}
+
+	if len(mustHaveNotHaveIndexCol) > 0 {
+		s.AppendErrorNo(ErrColumnsMustHaveIndex, strings.Join(mustHaveNotHaveIndexCol, ","))
+	}
+
+}
+
 func (s *session) buildTableInfo(node *ast.CreateTableStmt) *TableInfo {
 	log.Debug("buildTableInfo")
 
@@ -3964,6 +4025,11 @@ func (s *session) checkAlterTable(node *ast.AlterTableStmt, sql string) {
 		}
 	}
 
+	if s.Inc.ColumnsMustHaveIndex != "" {
+		tableCopy := s.getTableFromCache(node.Table.Schema.O, node.Table.Name.O, true)
+		s.checkColumnsMustHaveindex(tableCopy)
+	}
+
 	// 生成alter回滚语句,多个时逆向
 	if !s.hasError() && s.opt.execute && s.opt.backup {
 		if hasRenameTable {

session/session_inception_test.go

@@ -782,6 +782,40 @@ primary key(id)) comment 'test';`
 
 	config.GetGlobalConfig().Inc.MustHaveColumns = ""
 
+	// 如果表包含以下列，列必须有索引。
+	config.GetGlobalConfig().Inc.ColumnsMustHaveIndex = "c1 , c2 int "
+	sql = `drop table if exists t1;CREATE TABLE t1(id int, c1 int, c2 int, index idx_c1 (c1));`
+	s.testErrorCode(c, sql,
+		session.NewErr(session.ErrColumnsMustHaveIndex, "c2"))
+
+	sql = `drop table if exists t1;CREATE TABLE t1(id int, c1 int, c2 varchar (20), index idx_c1 (c1));`
+	s.testErrorCode(c, sql,
+		session.NewErr(session.ErrColumnsMustHaveIndexTypeErr, "c2", "int", "varchar"),
+		session.NewErr(session.ErrColumnsMustHaveIndex, "c2"))
+
+	sql = `drop table if exists t1;create table t1(id int,c2 int unique);`
+	s.testErrorCode(c, sql)
+
+	sql = `drop table if exists t1;create table t1(c1 int primary key,c2 int unique)`
+	s.testErrorCode(c, sql)
+
+	sql = `drop table if exists t1;CREATE TABLE t1(id int, c1 int, c2 int, index idx_1 (c1,c2));`
+	s.testErrorCode(c, sql,
+		session.NewErr(session.ErrColumnsMustHaveIndex, "c2"))
+
+	sql = `drop table if exists t1;CREATE TABLE t1(id int,c1 int ,c2 int,index idx_c1(c1),index idx_c2(c2));`
+	s.testErrorCode(c, sql)
+
+	sql = `drop table if exists t1;CREATE TABLE t1(id int);ALTER TABLE t1 ADD COLUMN c2 varchar(20);`
+	s.testErrorCode(c, sql,
+		session.NewErr(session.ErrColumnsMustHaveIndexTypeErr, "c2", "int", "varchar"),
+		session.NewErr(session.ErrColumnsMustHaveIndex, "c2"))
+
+	sql = `drop table if exists t1;CREATE TABLE t1(id int);ALTER TABLE t1 ADD COLUMN c2 int,add index idx_c2(c2);`
+	s.testErrorCode(c, sql)
+
+	config.GetGlobalConfig().Inc.ColumnsMustHaveIndex = ""
+
 	config.GetGlobalConfig().Inc.CheckInsertField = false
 	config.GetGlobalConfig().IncLevel.ER_WITH_INSERT_FIELD = 0