Merge pull request #3131 from actiontech/new-rule-2.2306.x

iwanghc · web-flow · commit 826bca7bcc4d · 2025-09-05T17:38:27.000+08:00
New rule 2.2306.x
diff --git a/sqle/driver/mysql/audit_test.go b/sqle/driver/mysql/audit_test.go
@@ -129,7 +129,8 @@ func runDefaultRulesInspectCase(t *testing.T, desc string, i *MysqlDriverImpl, s
 		rulepkg.DMLCheckAlias:                               {},
 		rulepkg.DMLCheckAffectedRows:                        {},
 		rulepkg.DMLCheckSortColumnLength:                    {},
-		rulepkg.DDLCheckAllIndexNotNullConstraint:			 {},
+		rulepkg.DDLCheckAllIndexNotNullConstraint:           {},
+		rulepkg.DDLCheckTransactionNotCommitted:             {},
 	}
 	for i := range rulepkg.RuleHandlers {
 		handler := rulepkg.RuleHandlers[i]
@@ -5330,7 +5331,6 @@ func TestDDLCheckAllIndexNotNullConstraint(t *testing.T) {
 	)
 }
 
-
 func TestDMLCheckSameTableJoinedMultipleTimes(t *testing.T) {
 	rule := rulepkg.RuleHandlerMap[rulepkg.DMLCheckSameTableJoinedMultipleTimes].Rule
 
diff --git a/sqle/driver/mysql/rule/rule.go b/sqle/driver/mysql/rule/rule.go
@@ -46,6 +46,7 @@ const (
 
 // inspector DDL rules
 const (
+	DDLCheckTransactionNotCommitted                    = "ddl_check_transactions_not_committed"
 	DDLCheckPKWithoutIfNotExists                       = "ddl_check_table_without_if_not_exists"
 	DDLCheckObjectNameLength                           = "ddl_check_object_name_length"
 	DDLCheckObjectNameUsingKeyword                     = "ddl_check_object_name_using_keyword"
@@ -432,6 +433,18 @@ var RuleHandlers = []RuleHandler{
 	},
 
 	// rule
+	{
+		Rule: driverV2.Rule{
+			Name:       DDLCheckTransactionNotCommitted,
+			Desc:       "DDL执行前存在事务未提交",
+			Annotation: "检查可能产生锁冲突的DDL操作执行前是否存在事务未提交，MySQL8之前版本会查询information_schema.innodb_trx所有记录，MySQL8版本会查询performance_schema.data_locks相关记录",
+			Level:      driverV2.RuleLevelWarn,
+			Category:   RuleTypeUsageSuggestion,
+		},
+		Message:      "DDL执行前存在事务未提交, %s",
+		AllowOffline: false,
+		Func:         checkTransactionNotCommittedBeforeDDL,
+	},
 	{
 		Rule: driverV2.Rule{
 			Name:       DDLCheckPKWithoutIfNotExists,
@@ -5909,7 +5922,7 @@ func getTableNameWithSchema(stmt *ast.TableName, c *session.Context) string {
 	} else {
 		tableWithSchema = fmt.Sprintf("`%s`.`%s`", stmt.Schema, stmt.Name)
 	}
-	
+
 	if c.IsLowerCaseTableName() {
 		tableWithSchema = strings.ToLower(tableWithSchema)
 	}
@@ -5919,7 +5932,7 @@ func getTableNameWithSchema(stmt *ast.TableName, c *session.Context) string {
 
 func checkSameTableJoinedMultipleTimes(input *RuleHandlerInput) error {
 	var repeatTables []string
-	
+
 	if _, ok := input.Node.(ast.DMLNode); ok {
 		selectVisitor := &util.SelectVisitor{}
 		input.Node.Accept(selectVisitor)
@@ -5973,3 +5986,139 @@ func checkAllIndexNotNullConstraint(input *RuleHandlerInput) error {
 	}
 	return nil
 }
+
+type checkTransactionNotCommittedBeforeDDLTableInfo struct {
+	Schema string
+	Table  string
+}
+
+func (c checkTransactionNotCommittedBeforeDDLTableInfo) String() string {
+	if c.Table == "" {
+		return c.Schema
+	}
+	if c.Schema == "" {
+		return c.Table
+	}
+	return fmt.Sprintf("%s.%s", c.Schema, c.Table)
+}
+
+func checkTransactionNotCommittedBeforeDDL(input *RuleHandlerInput) error {
+	// 跳过非DDL语句
+	switch input.Node.(type) {
+	case ast.DDLNode:
+	default:
+		return nil
+	}
+
+	tables := extractDDLSchemaAndTable(input.Node)
+	if len(tables) == 0 {
+		return nil
+	}
+
+	const transactionExecSecs = 600
+
+	// 首先检测MySQL版本
+	version, err := input.Ctx.GetMySQLMajorVersion()
+	if err != nil {
+		return err
+	}
+
+	for _, table := range tables {
+		if table.Schema == "" {
+			table.Schema = input.Ctx.CurrentSchema()
+		}
+		if table.Schema == "" {
+			// 如果没有提取到schema，跳过检查
+			continue
+		}
+
+		if input.Ctx.IsLowerCaseTableName() {
+			table.Schema, table.Table = strings.ToLower(table.Schema), strings.ToLower(table.Table)
+		}
+
+		// 根据版本选择不同的策略
+		if version >= 8 {
+			// MySQL 8
+			count, err := input.Ctx.CheckTableRelatedTransactionNotCommittedMySQL8(table.Schema, table.Table)
+			if err != nil {
+				return err
+			}
+			if count > 0 {
+				addResult(input.Res, input.Rule, input.Rule.Name,
+					fmt.Sprintf("performance_schema.data_locks存在%d条%s的相关记录", count, table))
+				return nil
+			}
+		} else {
+			// MySQL 5
+			count, ExecTimeoutCount, err := input.Ctx.CheckTransactionNotCommittedMySQL5(transactionExecSecs)
+			if err != nil {
+				return err
+			}
+			if count > 0 || ExecTimeoutCount > 0 {
+				addResult(input.Res, input.Rule, input.Rule.Name,
+					fmt.Sprintf("information_schema.innodb_trx存在%d条记录, %d条执行时长超过%d秒", count, ExecTimeoutCount, transactionExecSecs))
+				return nil
+			}
+		}
+	}
+
+	return nil
+}
+
+// extractDDLSchemaAndTable 从DDL语句中提取schema和table name
+func extractDDLSchemaAndTable(node ast.Node) []checkTransactionNotCommittedBeforeDDLTableInfo {
+	var tables []checkTransactionNotCommittedBeforeDDLTableInfo
+
+	switch stmt := node.(type) {
+	case *ast.CreateTableStmt, *ast.CreateIndexStmt, *ast.CreateViewStmt, *ast.CreateSequenceStmt, *ast.CreateDatabaseStmt:
+		// 跳过CREATE类型的DDL
+		return nil
+	case *ast.AlterTableStmt:
+		tables = append(tables, checkTransactionNotCommittedBeforeDDLTableInfo{
+			Schema: stmt.Table.Schema.O,
+			Table:  stmt.Table.Name.O,
+		})
+	case *ast.DropTableStmt:
+		for _, table := range stmt.Tables {
+			tables = append(tables, checkTransactionNotCommittedBeforeDDLTableInfo{
+				Schema: table.Schema.O,
+				Table:  table.Name.O,
+			})
+		}
+	case *ast.DropIndexStmt:
+		tables = append(tables, checkTransactionNotCommittedBeforeDDLTableInfo{
+			Schema: stmt.Table.Schema.O,
+			Table:  stmt.Table.Name.O,
+		})
+	case *ast.DropSequenceStmt:
+		for _, sequence := range stmt.Sequences {
+			tables = append(tables, checkTransactionNotCommittedBeforeDDLTableInfo{
+				Schema: sequence.Schema.O,
+				Table:  sequence.Name.L,
+			})
+		}
+	case *ast.RenameTableStmt:
+		// 对于RENAME TABLE，检查旧表
+		tables = append(tables, checkTransactionNotCommittedBeforeDDLTableInfo{
+			Schema: stmt.OldTable.Schema.O,
+			Table:  stmt.OldTable.Name.O,
+		})
+	case *ast.TruncateTableStmt:
+		tables = append(tables, checkTransactionNotCommittedBeforeDDLTableInfo{
+			Schema: stmt.Table.Schema.O,
+			Table:  stmt.Table.Name.O,
+		})
+	case *ast.RepairTableStmt:
+		tables = append(tables, checkTransactionNotCommittedBeforeDDLTableInfo{
+			Schema: stmt.Table.Schema.O,
+			Table:  stmt.Table.Name.O,
+		})
+	case *ast.DropDatabaseStmt:
+		// 数据库级别的DDL，返回数据库名作为schema
+		tables = append(tables, checkTransactionNotCommittedBeforeDDLTableInfo{
+			Schema: stmt.Name,
+		})
+	}
+
+	return tables
+}
diff --git a/sqle/driver/mysql/session/context.go b/sqle/driver/mysql/session/context.go
@@ -1,6 +1,7 @@
 package session
 
 import (
+	"database/sql"
 	"fmt"
 	"strconv"
 	"strings"
@@ -187,6 +188,35 @@ func (c *Context) IsLowerCaseTableName() bool {
 	return lowerCaseTableNames != "0"
 }
 
+// GetMySQLVersion 获取MySQL版本号
+func (c *Context) GetMySQLVersion() string {
+	version, err := c.GetSystemVariable(SysVarVersion)
+	if err != nil {
+		log.NewEntry().Errorf("fail to load system variable version, error: %v", err)
+		return ""
+	}
+	return version
+}
+
+// GetMySQLMajorVersion 获取MySQL主版本号
+func (c *Context) GetMySQLMajorVersion() (int, error) {
+	versionStr := c.GetMySQLVersion()
+	if versionStr == "" {
+		return 0, fmt.Errorf("unable to get MySQL version")
+	}
+
+	// 解析版本号，例如 "8.0.33" -> 8
+	versionParts := strings.Split(versionStr, ".")
+	if len(versionParts) > 0 {
+		majorVersion, err := strconv.Atoi(versionParts[0])
+		if err == nil {
+			return majorVersion, nil
+		}
+	}
+
+	return 0, fmt.Errorf("unable to parse MySQL version: %s", versionStr)
+}
+
 func (c *Context) getSchema(schemaName string) (*SchemaInfo, bool) {
 	if c.IsLowerCaseTableName() {
 		schemaName = strings.ToLower(schemaName)
@@ -478,6 +508,7 @@ func (c *Context) IsTableExist(stmt *ast.TableName) (bool, error) {
 
 const (
 	SysVarLowerCaseTableNames = "lower_case_table_names"
+	SysVarVersion             = "version"
 )
 
 // GetSystemVariable get system variable.
@@ -567,27 +598,31 @@ func (c *Context) GetCreateTableStmt(stmt *ast.TableName) (*ast.CreateTableStmt,
 /*
 建表语句可能如下:
 CREATE TABLE `__all_server_event_history` (
-  `gmt_create` timestamp(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
-  `svr_ip` varchar(46) NOT NULL,
-  `svr_port` bigint(20) NOT NULL,
-  `module` varchar(64) NOT NULL,
-  `event` varchar(64) NOT NULL,
-  `name1` varchar(256) DEFAULT '',
-  `value1` varchar(256) DEFAULT '',
-  `name2` varchar(256) DEFAULT '',
-  `value2` longtext DEFAULT NULL,
-  `name3` varchar(256) DEFAULT '',
-  `value3` varchar(256) DEFAULT '',
-  `name4` varchar(256) DEFAULT '',
-  `value4` varchar(256) DEFAULT '',
-  `name5` varchar(256) DEFAULT '',
-  `value5` varchar(256) DEFAULT '',
-  `name6` varchar(256) DEFAULT '',
-  `value6` varchar(256) DEFAULT '',
-  `extra_info` varchar(512) DEFAULT '',
-  PRIMARY KEY (`gmt_create`, `svr_ip`, `svr_port`)
+
+	`gmt_create` timestamp(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
+	`svr_ip` varchar(46) NOT NULL,
+	`svr_port` bigint(20) NOT NULL,
+	`module` varchar(64) NOT NULL,
+	`event` varchar(64) NOT NULL,
+	`name1` varchar(256) DEFAULT '',
+	`value1` varchar(256) DEFAULT '',
+	`name2` varchar(256) DEFAULT '',
+	`value2` longtext DEFAULT NULL,
+	`name3` varchar(256) DEFAULT '',
+	`value3` varchar(256) DEFAULT '',
+	`name4` varchar(256) DEFAULT '',
+	`value4` varchar(256) DEFAULT '',
+	`name5` varchar(256) DEFAULT '',
+	`value5` varchar(256) DEFAULT '',
+	`name6` varchar(256) DEFAULT '',
+	`value6` varchar(256) DEFAULT '',
+	`extra_info` varchar(512) DEFAULT '',
+	PRIMARY KEY (`gmt_create`, `svr_ip`, `svr_port`)
+
 ) DEFAULT CHARSET = utf8mb4 ROW_FORMAT = COMPACT COMPRESSION = 'none' REPLICA_NUM = 1 BLOCK_SIZE = 16384 USE_BLOOM_FILTER = FALSE TABLET_SIZE = 134217728 PCTFREE = 10 TABLEGROUP = 'oceanbase'
- partition by key_v2(svr_ip, svr_port)
+
+	partition by key_v2(svr_ip, svr_port)
+
 (partition p0,
 partition p1,
 partition p2,
@@ -606,7 +641,6 @@ partition p14,
 partition p15)
 
 建表语句后半段是options，oceanbase mysql模式下的show create table结果返回的options中包含mysql不支持的options, 为了能解析, 方法将会倒着遍历建表语句, 每次找到右括号时截断后面的部分, 然后尝试解析一次, 直到解析成功, 此时剩余的建表语句将不在包含OB特有options
-
 */
 func (c *Context) parseCreateTableSqlCompatibly(createTableSql string) (*ast.CreateTableStmt, error) {
 	for i := len(createTableSql) - 1; i >= 0; i-- {
@@ -889,3 +923,65 @@ func (c *Context) GetColumnCardinality(tn *ast.TableName, columnName string) (in
 func (c *Context) GetExecutor() *executor.Executor {
 	return c.e
 }
+
+func (c *Context) CheckTransactionNotCommittedMySQL5(execSecs int) (int, int, error) {
+	var count, timeoutCount int
+	query := `SELECT COUNT(*) FROM information_schema.innodb_trx`
+	results, err := c.GetExecutor().Db.Query(query)
+	if err != nil {
+		return 0, 0, err
+	}
+	if len(results) > 0 && len(results[0]) > 0 {
+		countStr := results[0]["COUNT(*)"]
+		count, err = strconv.Atoi(countStr.String)
+		if err != nil {
+			return 0, 0, err
+		}
+	}
+
+	query = `SELECT COUNT(*) FROM information_schema.innodb_trx WHERE TIMESTAMPDIFF(SECOND, trx_started, NOW()) > ?`
+	results, err = c.GetExecutor().Db.Query(query, execSecs)
+	if err != nil {
+		return 0, 0, err
+	}
+	if len(results) > 0 && len(results[0]) > 0 {
+		countStr := results[0]["COUNT(*)"]
+		timeoutCount, err = strconv.Atoi(countStr.String)
+		if err != nil {
+			return 0, 0, err
+		}
+	}
+
+	return count, timeoutCount, nil
+
+}
+
+func (c *Context) CheckTableRelatedTransactionNotCommittedMySQL8(schema, table string) (int, error) {
+	if schema == "" {
+		return 0, nil
+	}
+
+	var results []map[string]sql.NullString
+	var err error
+
+	if table == "" {
+		query := `SELECT COUNT(*) FROM performance_schema.data_locks WHERE object_schema = ?`
+		results, err = c.GetExecutor().Db.Query(query, schema)
+	} else {
+		query := `SELECT COUNT(*) FROM performance_schema.data_locks WHERE object_schema = ? AND object_name = ?`
+		results, err = c.GetExecutor().Db.Query(query, schema, table)
+	}
+
+	if err != nil {
+		return 0, err
+	}
+
+	if len(results) > 0 && len(results[0]) > 0 {
+		countStr := results[0]["COUNT(*)"]
+		if countStr.Valid {
+			return strconv.Atoi(countStr.String)
+		}
+	}
+
+	return 0, nil
+}

Original file line number	Diff line number	Diff line change
`@@ -129,7 +129,8 @@ func runDefaultRulesInspectCase(t testing.T, desc string, i MysqlDriverImpl, s`
`129`	`129`	`rulepkg.DMLCheckAlias: {},`
`130`	`130`	`rulepkg.DMLCheckAffectedRows: {},`
`131`	`131`	`rulepkg.DMLCheckSortColumnLength: {},`
`132`		`- rulepkg.DDLCheckAllIndexNotNullConstraint: {},`
	`132`	`+ rulepkg.DDLCheckAllIndexNotNullConstraint: {},`
	`133`	`+ rulepkg.DDLCheckTransactionNotCommitted: {},`
`133`	`134`	`}`
`134`	`135`	`for i := range rulepkg.RuleHandlers {`
`135`	`136`	`handler := rulepkg.RuleHandlers[i]`
`@@ -5330,7 +5331,6 @@ func TestDDLCheckAllIndexNotNullConstraint(t *testing.T) {`
`5330`	`5331`	`)`
`5331`	`5332`	`}`
`5332`	`5333`
`5333`		`-`
`5334`	`5334`	`func TestDMLCheckSameTableJoinedMultipleTimes(t *testing.T) {`
`5335`	`5335`	`rule := rulepkg.RuleHandlerMap[rulepkg.DMLCheckSameTableJoinedMultipleTimes].Rule`
`5336`	`5336`