Merge branch 'main' of https://github.com/actiontech/sqle into main

Author: actiontech

sqle/api/controller/v1/configuration.go

@@ -419,7 +419,7 @@ func getGitAuthMethod(url, username, password string) (transport.AuthMethod, err
 		if err != nil {
 			return nil, err
 		}
-		if systemVariable.Code != 0 {
+		if systemVariable.Data.SystemVariableSSHPrimaryKey != "" {
 			return nil, errors.New(errors.DataNotExist, fmt.Errorf("git ssh private key not found"))
 		}
 		publicKeys, err := sshTransport.NewPublicKeys("git", []byte(systemVariable.Data.SystemVariableSSHPrimaryKey), "")
@@ -550,7 +550,7 @@ func GetSSHPublicKey(c echo.Context) error {
 	if err != nil {
 		return controller.JSONBaseErrorReq(c, err)
 	}
-	if systemVariable.Code != 0 {
+	if systemVariable.Data.SystemVariableSSHPrimaryKey == "" {
 		return c.JSON(http.StatusOK, SSHPublicKeyInfoV1Rsp{
 			BaseRes: controller.NewBaseReq(nil),
 			Data: SSHPublicKeyInfo{

sqle/api/controller/v1/workflow.go

@@ -834,6 +834,14 @@ func toGlobalWorkflowRes(workflows []*model.WorkflowListDetail, projectMap Proje
 				InstanceName: instanceMap.InstanceName(id),
 			})
 		}
+
+		CurrentStepAssigneeUserNames := make([]string, 0)
+		for _, currentStepAssigneeUser := range strings.Split(workflow.CurrentStepAssigneeUserIds.String, ",") {
+			if currentStepAssigneeUser == "" {
+				continue
+			}
+			CurrentStepAssigneeUserNames = append(CurrentStepAssigneeUserNames, dms.GetUserNameWithDelTag(currentStepAssigneeUser))
+		}
 		workflowRes := &WorkflowDetailResV1{
 			ProjectName:             projectMap.ProjectName(workflow.ProjectId),
 			ProjectUid:              workflow.ProjectId,
@@ -845,7 +853,7 @@ func toGlobalWorkflowRes(workflows []*model.WorkflowListDetail, projectMap Proje
 			CreateUser:              utils.AddDelTag(workflow.CreateUserDeletedAt, workflow.CreateUser.String),
 			CreateTime:              workflow.CreateTime,
 			CurrentStepType:         workflow.CurrentStepType.String,
-			CurrentStepAssigneeUser: strings.Split(workflow.CurrentStepAssigneeUserIds.String, ","),
+			CurrentStepAssigneeUser: CurrentStepAssigneeUserNames,
 			Status:                  workflow.Status,
 		}
 		workflowsResV1 = append(workflowsResV1, workflowRes)

sqle/api/controller/v2/workflow.go

@@ -778,20 +778,6 @@ func UpdateWorkflowV2(c echo.Context) error {
 			fmt.Errorf("you are not allow to operate the workflow")))
 	}
 
-	template, exist, err := s.GetWorkflowTemplateByProjectId(workflow.ProjectId)
-	if err != nil {
-		return controller.JSONBaseErrorReq(c, err)
-	}
-	if !exist {
-		return controller.JSONBaseErrorReq(c, errors.New(errors.DataConflict,
-			fmt.Errorf("failed to find the corresponding workflow template based on the task id")))
-	}
-
-	err = v1.CheckWorkflowCanCommit(template, tasks)
-	if err != nil {
-		return controller.JSONBaseErrorReq(c, err)
-	}
-
 	err = s.UpdateWorkflowRecord(workflow, tasks)
 	if err != nil {
 		return c.JSON(http.StatusOK, controller.NewBaseReq(err))

sqle/driver/mysql/analysis.go

@@ -9,6 +9,7 @@ import (
 	"github.com/actiontech/sqle/sqle/driver"
 	"github.com/actiontech/sqle/sqle/driver/mysql/executor"
 	"github.com/actiontech/sqle/sqle/driver/mysql/plocale"
+	utilV2 "github.com/actiontech/sqle/sqle/driver/mysql/rule/ai/util"
 	"github.com/actiontech/sqle/sqle/driver/mysql/util"
 	driverV2 "github.com/actiontech/sqle/sqle/driver/v2"
 	"github.com/actiontech/sqle/sqle/utils"
@@ -285,48 +286,9 @@ func (i *MysqlDriverImpl) ExtractSchemaTableList(sql string) ([]SchemaTable, err
 		}
 	}
 
-	getMultiTables := func(stmt *ast.Join) {
-		tables := util.GetTables(stmt)
-		for _, t := range tables {
-			addTable(t)
-		}
-	}
-
-	switch stmt := node.(type) {
-	case *ast.SelectStmt:
-		if stmt.From == nil {
-			break
-		}
-		getMultiTables(stmt.From.TableRefs)
-	case *ast.UnionStmt:
-		for _, selectStmt := range stmt.SelectList.Selects {
-			if selectStmt.From == nil {
-				continue
-			}
-			getMultiTables(selectStmt.From.TableRefs)
-		}
-	case *ast.UpdateStmt:
-		getMultiTables(stmt.TableRefs.TableRefs)
-	case *ast.InsertStmt:
-		getMultiTables(stmt.Table.TableRefs)
-		if stmt.Select != nil {
-			// TODO：INSERT INTO SQLE00115_t1_tmp_employee (id, cname, sex, age, salary) SELECT 4000002, '小张', 0, 25, (SELECT AVG(salary) FROM SQLE00115_t1_employee) 对于这条SQL，解析器无法解析子查询为一个Select语句，而是认为是一个文本
-			if selectStmt, ok := stmt.Select.(*ast.SelectStmt); ok {
-				if selectStmt.From != nil{
-					getMultiTables(selectStmt.From.TableRefs)
-				}
-			}
-		}
-	case *ast.DeleteStmt:
-		getMultiTables(stmt.TableRefs.TableRefs)
-	case *ast.LoadDataStmt:
-		addTable(stmt.Table)
-	case *ast.ShowStmt:
-		if stmt.Table != nil {
-			addTable(stmt.Table)
-		}
-	default:
-		return nil, fmt.Errorf("the sql is `%v`, we don't support analysing this sql", sql)
+	tables := utilV2.GetTableNames(node)
+	for _, t := range tables {
+		addTable(t)
 	}
 
 	return schemaTables, nil

sqle/driver/mysql/executor/executor.go

@@ -23,7 +23,7 @@ type Db interface {
 	Close()
 	Ping() error
 	Exec(query string) (driver.Result, error)
-	Transact(qs ...string) ([]driver.Result, error)
+	Transact(qs ...string) (*driverV2.TxResponse, error)
 	Query(query string, args ...interface{}) ([]map[string]sql.NullString, error)
 	QueryWithContext(ctx context.Context, query string, args ...interface{}) (column []string, row [][]sql.NullString, err error)
 	Logger() *logrus.Entry
@@ -137,14 +137,13 @@ func (c *BaseConn) Exec(query string) (driver.Result, error) {
 	return result, errors.New(errors.ConnectRemoteDatabaseError, err)
 }
 
-func (c *BaseConn) Transact(qs ...string) ([]driver.Result, error) {
+func (c *BaseConn) Transact(qs ...string) (*driverV2.TxResponse, error) {
 	var err error
 	var tx *sql.Tx
-	var results []driver.Result
 	c.Logger().Infof("doing sql transact, host: %s, port: %s, user: %s", c.host, c.port, c.user)
 	tx, err = c.conn.BeginTx(context.Background(), nil)
 	if err != nil {
-		return results, err
+		return nil, err
 	}
 	defer func() {
 		if p := recover(); p != nil {
@@ -168,14 +167,23 @@ func (c *BaseConn) Transact(qs ...string) ([]driver.Result, error) {
 			c.Logger().Info("done sql transact")
 		}
 	}()
-	for _, query := range qs {
+
+	results := &driverV2.TxResponse{
+		ExecResult: make([]driver.Result, 0, len(qs)),
+	}
+	for k, query := range qs {
 		var txResult driver.Result
 		txResult, err = tx.Exec(query)
 		if err != nil {
+			// SQL执行报错记录序号和错误信息在业务结构中
+			results.ExecErr = &driverV2.ExecErr{
+				ErrSqlIndex:   uint32(k),
+				SqlExecErrMsg: err.Error(),
+			}
 			c.Logger().Errorf("exec sql failed, error: %s, query: %s", err, query)
-			return results, err
+			return results, nil
 		} else {
-			results = append(results, txResult)
+			results.ExecResult = append(results.ExecResult, txResult)
 			c.Logger().Infof("exec sql success, query: %s", query)
 		}
 	}

sqle/driver/mysql/mysql.go

@@ -240,7 +240,7 @@ func (i *MysqlDriverImpl) onlineddlWithGhost(query string) (bool, error) {
 	return int64(tableSize) > i.cnf.DDLGhostMinSize, nil
 }
 
-func (i *MysqlDriverImpl) Tx(ctx context.Context, queries ...string) ([]_driver.Result, error) {
+func (i *MysqlDriverImpl) Tx(ctx context.Context, queries ...string) (*driverV2.TxResponse, error) {
 	if i.IsOfflineAudit() {
 		return nil, nil
 	}

sqle/driver/mysql/rule/ai/util/extractor.go

@@ -158,12 +158,23 @@ type SubqueryExprExtractor struct {
 	expr []*ast.SubqueryExpr
 }
 
-func (te *SubqueryExprExtractor) Enter(in ast.Node) (node ast.Node, skipChildren bool) {
-	e, ok := in.(*ast.SubqueryExpr)
-	if !ok {
+func (te *SubqueryExprExtractor) Enter(in ast.Node) (ast.Node, bool) {
+	// 情况 1: WHERE、IN、EXISTS 中的子查询
+	if sub, ok := in.(*ast.SubqueryExpr); ok {
+		te.expr = append(te.expr, sub)
 		return in, false
 	}
-	te.expr = append(te.expr, e)
+
+	// 情况 2: FROM 中的子查询（即 SelectStmt 嵌套在 TableSource 中）
+	if ts, ok := in.(*ast.TableSource); ok {
+		if sel, ok := ts.Source.(*ast.SelectStmt); ok {
+			// 包装成 SubqueryExpr 形式以便统一处理
+			te.expr = append(te.expr, &ast.SubqueryExpr{
+				Query: sel,
+			})
+		}
+	}
+
 	return in, false
 }
 

sqle/driver/mysql/rule_00108_test.go

@@ -72,15 +72,15 @@ func TestRuleSQLE00108(t *testing.T) {
 		nil, newTestResult())
 
 	// 这个子查询中实际扫描表的子查询 只有1个，因此不算违规
-	runAIRuleCase(rule, t, "case 15: SELECT语句中使用JOIN的ON条件中嵌套子查询5层, 但是实际扫描表的子查询只有1次，因此不算违规",
+	runAIRuleCase(rule, t, "case 15: SELECT语句中使用JOIN的ON条件中嵌套子查询5层",
 		"SELECT st1.id FROM st1 JOIN st_class ON st1.cid in (SELECT cid FROM (SELECT cid FROM (SELECT cid FROM (SELECT cid FROM (SELECT cid FROM st_class WHERE cname = 'class2') AS sub1) AS sub2) AS sub3) AS sub4);",
 		session.NewAIMockContext().WithSQL("CREATE TABLE st1 (id INT, cid INT); CREATE TABLE st_class (cid INT, cname VARCHAR(50));"),
 		nil, newTestResult())
 
-	runAIRuleCase(rule, t, "case 16: SELECT语句中使用JOIN的ON条件中嵌套子查询6层, 但是实际扫描表的子查询只有2次，因此不算违规",
+	runAIRuleCase(rule, t, "case 16: SELECT语句中使用JOIN的ON条件中嵌套子查询6层, 违规",
 		"SELECT st1.id FROM st1 JOIN st_class ON st1.cid in (SELECT cid FROM (SELECT cid FROM (SELECT cid FROM (SELECT cid FROM (SELECT cid FROM st_class WHERE cname in (SELECT cname FROM st_class WHERE cname = 'class2')) AS sub1) AS sub2) AS sub3) AS sub4);",
 		session.NewAIMockContext().WithSQL("CREATE TABLE st1 (id INT, cid INT); CREATE TABLE st_class (cid INT, cname VARCHAR(50));"),
-		nil, newTestResult())
+		nil, newTestResult().addResult(ruleName))
 
 	runAIRuleCase(rule, t, "case 17: SELECT语句中 查询列中, 嵌套子查询2层",
 		"SELECT 1, st1.id, (SELECT (SELECT id0 FROM exist_db.exist_tb_1 WHERE id1 = 'value') xx2 FROM exist_db.exist_tb_1 WHERE id1 = 'value') xxx FROM st1;",

sqle/driver/mysql/rule_00109_test.go

@@ -71,12 +71,11 @@ func TestRuleSQLE00109(t *testing.T) {
 			WithSQL("CREATE TABLE orders_archive (id INT, amount DECIMAL(10,2));"),
 		nil, newTestResult())
 
-	// 特殊情况：解析器好像对from中嵌套的子查询 ，会被视作为属于非实际路由的子查询，因此没有累计（相当于只是个外壳嵌套而已，因此这里可以视作为不违规
-	runAIRuleCase(rule, t, "case 9: SELECT ... UNION ALL SELECT语句中的子查询使用LIMIT，但是这里子查询其实是可以直接push到外层的，可以不算做子查询",
+	runAIRuleCase(rule, t, "case 9: SELECT ... UNION ALL SELECT语句中的子查询使用LIMIT",
 		"SELECT * FROM (SELECT id FROM products LIMIT 1) AS sub UNION ALL SELECT * FROM products;",
 		session.NewAIMockContext().
 			WithSQL("CREATE TABLE products (id INT, name VARCHAR(100));"),
-		nil, newTestResult())
+		nil, newTestResult().addResult(ruleName))
 
 	runAIRuleCase(rule, t, "case 10: SELECT ... UNION ALL SELECT语句中的子查询不使用LIMIT",
 		"SELECT * FROM (SELECT id FROM products) AS sub UNION ALL SELECT * FROM products;",

sqle/driver/plugin_adapter_v1.go

@@ -223,7 +223,7 @@ func (s *PluginImplV1) ExecBatch(ctx context.Context, sqls ...string) ([]sqlDriv
 	return nil, fmt.Errorf("unimplement this method")
 }
 
-func (p *PluginImplV1) Tx(ctx context.Context, queries ...string) ([]sqlDriver.Result, error) {
+func (p *PluginImplV1) Tx(ctx context.Context, queries ...string) (*driverV2.TxResponse, error) {
 	client, err := p.DriverManager.GetAuditDriver()
 	if err != nil {
 		return nil, err