Merge pull request #2808 from actiontech/2028-6

feat: implement of backup according to max rows

go.mod

@@ -5,7 +5,7 @@ go 1.19
 require (
 	github.com/DATA-DOG/go-sqlmock v1.5.0
 	github.com/Masterminds/semver/v3 v3.1.1
-	github.com/actiontech/dms v0.0.0-20241122070816-a943dbc68714
+	github.com/actiontech/dms v0.0.0-20241205071755-3b81d30f71c3
 	github.com/actiontech/java-sql-extractor v0.0.0-20231103015812-cdd5fc040f62
 	github.com/actiontech/mybatis-mapper-2-sql v0.5.1-0.20240806065717-29cde7000ef5
 	github.com/agiledragon/gomonkey v2.0.2+incompatible

go.sum

@@ -46,8 +46,8 @@ github.com/VividCortex/ewma v1.1.1 h1:MnEK4VOv6n0RSY4vtRe3h11qjxL3+t0B8yOL8iMXdc
 github.com/VividCortex/ewma v1.1.1/go.mod h1:2Tkkvm3sRDVXaiyucHiACn4cqf7DpdyLvmxzcbUokwA=
 github.com/acomagu/bufpipe v1.0.4 h1:e3H4WUzM3npvo5uv95QuJM3cQspFNtFBzvJ2oNjKIDQ=
 github.com/acomagu/bufpipe v1.0.4/go.mod h1:mxdxdup/WdsKVreO5GpW4+M/1CE2sMG4jeGJ2sYmHc4=
-github.com/actiontech/dms v0.0.0-20241122070816-a943dbc68714 h1:Zb69GHQvI5thzLjoKEE5mdaaKBEd8J9T0R5SXmGgWa0=
-github.com/actiontech/dms v0.0.0-20241122070816-a943dbc68714/go.mod h1:Jm+tDoBYmX8R8zqgPJRq1NwZXdxO/F+HxVRJVF65WXA=
+github.com/actiontech/dms v0.0.0-20241205071755-3b81d30f71c3 h1:96fKxlsA2ZOqRlfB1oseZsuu6HedEBf2yHRme+M4644=
+github.com/actiontech/dms v0.0.0-20241205071755-3b81d30f71c3/go.mod h1:Jm+tDoBYmX8R8zqgPJRq1NwZXdxO/F+HxVRJVF65WXA=
 github.com/actiontech/java-sql-extractor v0.0.0-20231103015812-cdd5fc040f62 h1:JM7WnLzlvXOGE90KKd+aigi+qUDS+U5dLwQMNpTKZxE=
 github.com/actiontech/java-sql-extractor v0.0.0-20231103015812-cdd5fc040f62/go.mod h1:adDZHhAf2LRMx2h0JzofPXn12x2XlyQjVE116KXquwo=
 github.com/actiontech/mybatis-mapper-2-sql v0.5.1-0.20240806065717-29cde7000ef5 h1:vyQVrkYPzUV9d7gSvOWoezwWMTiC4jc3f3Hpianefq0=

sqle/api/controller/v1/dms_handle.go

@@ -33,7 +33,7 @@ func OperateDataResourceHandle(c echo.Context) error {
 	}
 	h := dmsobject.GetOperateHandle(fmt.Sprintf("%s_%s_%s", req.OperationTiming, req.OperationType, req.DataResourceType))
 
-	if err := h.Handle(c.Request().Context(), "", req.DataResourceUid); err != nil {
+	if err := h.Handle(c.Request().Context(), "", req.DataResourceUid, req.ExtraParams); err != nil {
 		return c.JSON(http.StatusOK, dmsV1.OperateDataResourceHandleReply{GenericResp: baseV1.GenericResp{Code: http.StatusBadRequest, Message: err.Error()}})
 	}
 
@@ -58,7 +58,7 @@ type BeforeDeleteDbService struct {
 type AfterUpdateDbService struct {
 }
 
-func (h BeforeDeleteDbService) Handle(ctx context.Context, currentUserId string, instanceIdStr string) error {
+func (h BeforeDeleteDbService) Handle(ctx context.Context, currentUserId string, instanceIdStr string, extraParams string) error {
 	instanceId, err := strconv.ParseInt(instanceIdStr, 10, 64)
 	if err != nil {
 		return err
@@ -67,6 +67,6 @@ func (h BeforeDeleteDbService) Handle(ctx context.Context, currentUserId string,
 	return common.CheckDeleteInstance(instanceId)
 }
 
-func (h AfterUpdateDbService) Handle(ctx context.Context, currentUserId string, instanceIdStr string) error {
+func (h AfterUpdateDbService) Handle(ctx context.Context, currentUserId string, instanceIdStr string, extraParams string) error {
 	return nil
 }

sqle/api/controller/v1/dms_handle_ce.go

@@ -7,15 +7,15 @@ import (
 	"context"
 )
 
-func (h BeforeArchiveProject) Handle(ctx context.Context, currentUserId string, dataResourceId string) error {
+func (h BeforeArchiveProject) Handle(ctx context.Context, currentUserId string, objId string, extraParams string) error {
 	return nil
 }
-func (h AfterDeleteProject) Handle(ctx context.Context, currentUserId string, dataResourceId string) error {
+func (h AfterDeleteProject) Handle(ctx context.Context, currentUserId string, objId string, extraParams string) error {
 	return nil
 }
-func (h BeforeDeleteProject) Handle(ctx context.Context, currentUserId string, dataResourceId string) error {
+func (h BeforeDeleteProject) Handle(ctx context.Context, currentUserId string, objId string, extraParams string) error {
 	return nil
 }
-func (h AfterCreateProject) Handle(ctx context.Context, currentUserId string, dataResourceId string) error {
+func (h AfterCreateProject) Handle(ctx context.Context, currentUserId string, objId string, extraParams string) error {
 	return nil
 }

sqle/api/controller/v1/instance.go

@@ -430,6 +430,7 @@ func GetInstanceTips(c echo.Context) error {
 			Port:                    inst.Port,
 			WorkflowTemplateId:      uint32(template.ID),
 			EnableBackup:            inst.EnableBackup,
+			BackupMaxRows:           inst.BackupMaxRows,
 			SupportedBackupStrategy: svc.SupportedBackupStrategy(inst.DbType),
 		}
 		instanceTipsResV1 = append(instanceTipsResV1, instanceTipRes)

sqle/api/controller/v1/task.go

@@ -41,7 +41,7 @@ type CreateAuditTaskReqV1 struct {
 	Sql             string  `json:"sql" form:"sql" example:"alter table tb1 drop columns c1"`
 	ExecMode        string  `json:"exec_mode" form:"exec_mode" enums:"sql_file,sqls"`
 	EnableBackup    bool    `json:"enable_backup" form:"enable_backup"`
-	BackupMaxRows   *uint64 `json:"backup_max_rows,omitempty"`
+	BackupMaxRows   *uint64 `json:"backup_max_rows,omitempty" form:"backup_max_rows"`
 	FileOrderMethod string  `json:"file_order_method" form:"file_order_method"`
 }
 
@@ -89,6 +89,7 @@ func convertTaskToRes(task *model.Task) *AuditTaskResV1 {
 		ExecEndTime:                task.ExecEndAt,
 		ExecMode:                   task.ExecMode,
 		EnableBackup:               task.EnableBackup,
+		BackupMaxRows:              task.BackupMaxRows,
 		BackupConflictWithInstance: server.BackupService{}.IsBackupConflictWithInstance(task.EnableBackup, task.InstanceEnableBackup),
 		FileOrderMethod:            task.FileOrderMethod,
 		AuditFiles:                 convertToAuditFileResp(task.AuditFiles),
@@ -351,8 +352,8 @@ func CreateAndAuditTask(c echo.Context) error {
 
 	task.ExecMode = req.ExecMode
 	task.FileOrderMethod = req.FileOrderMethod
+	backupService := server.BackupService{}
 	if req.EnableBackup {
-		backupService := server.BackupService{}
 		err = backupService.CheckBackupConflictWithExecMode(req.EnableBackup, req.ExecMode)
 		if err != nil {
 			return controller.JSONBaseErrorReq(c, err)
@@ -363,6 +364,7 @@ func CreateAndAuditTask(c echo.Context) error {
 		}
 		task.EnableBackup = req.EnableBackup
 	}
+	task.BackupMaxRows = backupService.AutoChooseBackupMaxRows(req.EnableBackup, req.BackupMaxRows, tmpInst)
 	task.InstanceEnableBackup = tmpInst.EnableBackup
 
 	err = convertSQLSourceEncodingFromTask(task)
@@ -959,7 +961,7 @@ type AuditTaskGroupReqV1 struct {
 	TaskGroupId   uint    `json:"task_group_id" form:"task_group_id" valid:"required"`
 	Sql           string  `json:"sql" form:"sql" example:"alter table tb1 drop columns c1"`
 	EnableBackup  bool    `json:"enable_backup" form:"enable_backup"`
-	BackupMaxRows *uint64 `json:"backup_max_rows,omitempty"`
+	BackupMaxRows *uint64 `json:"backup_max_rows,omitempty" form:"backup_max_rows"`
 }
 
 type AuditTaskGroupRes struct {
@@ -1064,8 +1066,8 @@ func AuditTaskGroupV1(c echo.Context) error {
 
 		for _, task := range tasks {
 			task.SQLSource = sqls.SourceType
+			backupService := server.BackupService{}
 			if req.EnableBackup {
-				backupService := server.BackupService{}
 				err = backupService.CheckBackupConflictWithExecMode(req.EnableBackup, task.ExecMode)
 				if err != nil {
 					return controller.JSONBaseErrorReq(c, err)
@@ -1077,6 +1079,7 @@ func AuditTaskGroupV1(c echo.Context) error {
 				task.EnableBackup = req.EnableBackup
 			}
 			if instance, exist := instanceMap[task.InstanceId]; exist {
+				task.BackupMaxRows = backupService.AutoChooseBackupMaxRows(req.EnableBackup, req.BackupMaxRows, *instance)
 				task.InstanceEnableBackup = instance.EnableBackup
 			} else {
 				return controller.JSONBaseErrorReq(c, fmt.Errorf("can not find instance in task"))

sqle/dms/instance.go

@@ -140,6 +140,7 @@ func convertInstance(instance *dmsV1.ListDBService) (*model.Instance, error) {
 		SqlQueryConfig:    sqlQueryConfig,
 		Business:          instance.Business,
 		EnableBackup:      instance.EnableBackup,
+		BackupMaxRows:     instance.BackupMaxRows,
 	}, nil
 }
 

sqle/driver/mysql/backup_ce.go

@@ -12,7 +12,7 @@ import (
 
 var ErrUnsupportedBackup error = fmt.Errorf("backup is unsupported for sqle community version")
 
-func (i *MysqlDriverImpl) Backup(ctx context.Context, backupStrategy string, sql string) (BackupSql []string, ExecuteInfo string, err error) {
+func (i *MysqlDriverImpl) Backup(ctx context.Context, backupStrategy string, sql string, backupMaxRows uint64) (backupSqls []string, executeResult string, err error) {
 	return nil, "", ErrUnsupportedBackup
 }
 

sqle/driver/mysql/mysql.go

@@ -455,26 +455,7 @@ func (i *MysqlDriverImpl) audit(ctx context.Context, sql string) (*driverV2.Audi
 }
 
 func (i *MysqlDriverImpl) GenRollbackSQL(ctx context.Context, sql string) (string, i18nPkg.I18nStr, error) {
-	if i.IsOfflineAudit() {
-		return "", nil, nil
-	}
-	if i.HasInvalidSql {
-		return "", nil, nil
-	}
-
-	nodes, err := i.ParseSql(sql)
-	if err != nil {
-		return "", nil, err
-	}
-
-	rollback, i18nReason, err := i.GenerateRollbackSql(nodes[0])
-	if err != nil {
-		return "", nil, err
-	}
-
-	i.Ctx.UpdateContext(nodes[0])
-
-	return rollback, i18nReason, nil
+	return "", nil, nil
 }
 
 func (i *MysqlDriverImpl) Close(ctx context.Context) {
@@ -633,7 +614,6 @@ func (p *PluginProcessor) GetDriverMetas() (*driverV2.DriverMetas, error) {
 		Rules:                    allRules,
 		DatabaseAdditionalParams: params.Params{},
 		EnabledOptionalModule: []driverV2.OptionalModule{
-			driverV2.OptionalModuleGenRollbackSQL,
 			driverV2.OptionalModuleQuery,
 			driverV2.OptionalModuleExplain,
 			driverV2.OptionalModuleGetTableMeta,

sqle/driver/mysql/mysql_test.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"testing"
 
-	"github.com/actiontech/dms/pkg/dms-common/i18nPkg"
 	"github.com/actiontech/sqle/sqle/driver/mysql/util"
 	driverV2 "github.com/actiontech/sqle/sqle/driver/v2"
 	"github.com/stretchr/testify/assert"
@@ -160,24 +159,6 @@ func TestInspect_onlineddlWithGhost(t *testing.T) {
 	}
 }
 
-func TestInspect_GenRollbackSQL(t *testing.T) {
-	i := DefaultMysqlInspect()
-
-	rollback, reason, err := i.GenRollbackSQL(context.TODO(), "create table t1(id int, c1 int)")
-	assert.NoError(t, err)
-	assert.Equal(t, "", reason[i18nPkg.DefaultLang])
-	assert.Equal(t, "DROP TABLE IF EXISTS `exist_db`.`t1`", rollback)
-
-	rollback, reason, err = i.GenRollbackSQL(context.TODO(), "alter table t1 drop column c1")
-	assert.NoError(t, err)
-	assert.Equal(t, "", reason[i18nPkg.DefaultLang])
-	assert.Equal(t, "ALTER TABLE `exist_db`.`t1`\nADD COLUMN `c1` int(11);", rollback)
-
-	rollback, reason, err = i.GenRollbackSQL(context.TODO(), "alter table t1 add column c1 int")
-	assert.NoError(t, err)
-	assert.Equal(t, "", reason[i18nPkg.DefaultLang])
-	assert.Equal(t, "ALTER TABLE `exist_db`.`t1`\nDROP COLUMN `c1`;", rollback)
-}
 func TestInspect_assertSQLType(t *testing.T) {
 	args := []struct {
 		Name string