Merge pull request #2789 from actiontech/2028-1

modify: backup compatible with other plugins

sqle/driver/mysql/audit_test.go

@@ -114,7 +114,8 @@ func NewMockInspectWithIsExecutedSQL(e *executor.Executor) *MysqlDriverImpl {
 			Password:     "123456",
 			DatabaseName: "mysql",
 		},
-		Ctx: session.NewMockContext(e),
+		isConnected: true,
+		Ctx:         session.NewMockContext(e),
 		cnf: &Config{
 			DDLOSCMinSize:      16,
 			DDLGhostMinSize:    16,

sqle/driver/mysql/rollback.go

@@ -415,7 +415,7 @@ func (i *MysqlDriverImpl) generateInsertRollbackSql(stmt *ast.InsertStmt) (strin
 			for n, name := range columnsName {
 				_, isPk := pkColumnsName[name]
 				if isPk {
-					where = append(where, fmt.Sprintf("%s = '%s'", name, util.ExprFormat(value[n])))
+					where = append(where, fmt.Sprintf("%s = '%s'", name, restore(value[n])))
 				}
 			}
 			if len(where) != len(pkColumnsName) {
@@ -437,7 +437,7 @@ func (i *MysqlDriverImpl) generateInsertRollbackSql(stmt *ast.InsertStmt) (strin
 			name := setExpr.Column.Name.String()
 			_, isPk := pkColumnsName[name]
 			if isPk {
-				where = append(where, fmt.Sprintf("%s = '%s'", name, util.ExprFormat(setExpr.Expr)))
+				where = append(where, fmt.Sprintf("%s = '%s'", name, restore(setExpr.Expr)))
 			}
 		}
 		if len(where) != len(pkColumnsName) {
@@ -617,7 +617,7 @@ func (i *MysqlDriverImpl) generateUpdateRollbackSql(stmt *ast.UpdateStmt) (strin
 					colChanged = true
 					if isPk {
 						isPkChanged = true
-						pkValue = util.ExprFormat(l.Expr)
+						pkValue = restore(l.Expr)
 					}
 				}
 			}
@@ -702,12 +702,12 @@ func (i *MysqlDriverImpl) generateGetRecordsSql(expr string, tableName *ast.Tabl
 		recordSql = fmt.Sprintf("%s AS %s", recordSql, tableAlias)
 	}
 	if where != nil {
-		recordSql = fmt.Sprintf("%s WHERE %s", recordSql, util.ExprFormat(where))
+		recordSql = fmt.Sprintf("%s WHERE %s", recordSql, restore(where))
 	}
 	if order != nil {
 		recordSql = fmt.Sprintf("%s ORDER BY", recordSql)
 		for _, item := range order.Items {
-			recordSql = fmt.Sprintf("%s %s", recordSql, util.ExprFormat(item.Expr))
+			recordSql = fmt.Sprintf("%s %s", recordSql, restore(item.Expr))
 			if item.Desc {
 				recordSql = fmt.Sprintf("%s DESC", recordSql)
 			}

sqle/server/backup.go

@@ -2,17 +2,13 @@ package server
 
 import "github.com/actiontech/sqle/sqle/model"
 
-type BackupTask interface {
-	Backup() error
-}
-
 type BackupStrategy string
 
 const (
 	BackupStrategyNone        BackupStrategy = "none"         // 不备份(不支持备份、无需备份、选择不备份)
 	BackupStrategyReverseSql  BackupStrategy = "reverse_sql"  // 备份为反向SQL
 	BackupStrategyOriginalRow BackupStrategy = "original_row" // 备份为原始行
-	BackupStrategyManually    BackupStrategy = "manual"     // 标记为人工备份
+	BackupStrategyManually    BackupStrategy = "manual"       // 标记为人工备份
 	BackupRowsAffectedLimit   int            = 1000           // SQL影响行数上限，超过该上限的SQL不进行备份
 )
 
@@ -63,4 +59,4 @@ func (m backupTaskMap) AddBackupTask(backupTask *model.BackupTask) {
 	if _, exist := m[backupTask.ExecuteSqlId]; !exist {
 		m[backupTask.ExecuteSqlId] = backupTask
 	}
-}
+}

sqle/server/backup_ce.go

@@ -18,18 +18,18 @@ func (BackupService) CheckIsDbTypeSupportEnableBackup(dbType string) error {
 	return nil
 }
 
-type BaseBackupTask struct{}
+type BackupManager struct{}
 
-func (t BaseBackupTask) Backup() error {
+func (t BackupManager) Backup() error {
 	return nil
 }
 
 func initModelBackupTask(p driver.Plugin, task *model.Task, sql *model.ExecuteSQL) *model.BackupTask {
 	return &model.BackupTask{}
 }
 
-func toBackupTask(a driver.Plugin, sql *model.ExecuteSQL) (BackupTask, error) {
-	return &BaseBackupTask{}, nil
+func getBackupManager(p driver.Plugin, sql *model.ExecuteSQL, dbType string) (*BackupManager, error) {
+	return &BackupManager{}, nil
 }
 
 func (BackupService) GetRollbackSqlsMap(taskId uint) (map[uint][]string, error) {

sqle/server/sqled.go

@@ -445,8 +445,8 @@ func (a *action) GetTaskStatus(st *model.Storage) string {
 }
 
 func (a *action) execTask() (err error) {
-	// TODO  if enable backup and plugin support backup
-	if a.task.EnableBackup {
+	svc := BackupService{}
+	if svc.CheckCanTaskBackup(a.task) {
 		err = a.backupAndExecSql()
 		if err != nil {
 			return err
@@ -483,19 +483,16 @@ backupAndExecSql() 备份与执行SQL：
 	按照顺序，先根据一条SQL备份，再执行该SQL。备份过程中涉及连库查询和保存数据。
 */
 func (a *action) backupAndExecSql() error {
-	svc := BackupService{}
 	for _, executeSQL := range a.task.ExecuteSQLs {
-		if svc.CheckCanTaskBackup(a.task) {
-			backupTask, err := toBackupTask(a.plugin, executeSQL)
-			if err != nil {
-				return fmt.Errorf("in backupAndExecSql when convert toBackupTask, err %w , backup task: %v, task: %v", err, executeSQL.BackupTask.ID, a.task.ID)
-			}
-			if err = backupTask.Backup(); err != nil {
-				return fmt.Errorf("in backupAndExecSql when backupTask Backup, err %w, backup task: %v, task: %v", err, executeSQL.BackupTask.ID, a.task.ID)
-			}
+		backupMgr, err := getBackupManager(a.plugin, executeSQL, a.task.DBType)
+		if err != nil {
+			return fmt.Errorf("in backupAndExecSql when getBackupManager, err %w , task: %v", err, a.task.ID)
+		}
+		if err = backupMgr.Backup(); err != nil {
+			return fmt.Errorf("in backupAndExecSql when backupMgr Backup, err %w, backup manager: %v, task: %v", err, backupMgr, a.task.ID)
 		}
 		if err := a.execSQL(executeSQL); err != nil {
-			return fmt.Errorf("in backupAndExecSql when execSQL %v, err %w, backup task: %v, task: %v", executeSQL, err, executeSQL.BackupTask.ID, a.task.ID)
+			return fmt.Errorf("in backupAndExecSql when execSQL %v, err %w, backup manager: %v, task: %v", executeSQL, err, backupMgr, a.task.ID)
 		}
 	}
 	return nil