fix(fetch): fix wrong use of db.Fetch (#6669)

backend/core/models/migrationscripts/20230223_remove_createddateafter_from_collector_state.go

@@ -40,13 +40,13 @@ type removeCreatedDateAfterFromCollectorMeta20230223 struct{}
 func (script *removeCreatedDateAfterFromCollectorMeta20230223) Up(basicRes context.BasicRes) errors.Error {
 	db := basicRes.GetDal()
 	// step 1: rename bp.settings.createdDateAfter to timeAfter
-	bp := &blueprint20230223{}
-	cursor, err := db.Cursor(dal.From(bp), dal.Where("mode = ?", "NORMAL"))
+	cursor, err := db.Cursor(dal.From(&blueprint20230223{}), dal.Where("mode = ?", "NORMAL"))
 	if err != nil {
 		return err
 	}
 	defer cursor.Close()
 	for cursor.Next() {
+		bp := &blueprint20230223{}
 		err = db.Fetch(cursor, bp)
 		if err != nil {
 			return err

backend/core/models/migrationscripts/20230829_normalize_bp_settings.go

@@ -73,12 +73,12 @@ func (script *normalizeBpSettings) Up(basicRes context.BasicRes) errors.Error {
 		return errors.BadInput.New("invalid encKey")
 	}
 	db := basicRes.GetDal()
-	bp := &blueprint20230829{}
-	cursor := errors.Must1(db.Cursor(dal.From(bp)))
+	cursor := errors.Must1(db.Cursor(dal.From(&blueprint20230829{})))
 	defer cursor.Close()
 
 	for cursor.Next() {
 		// load row
+		bp := &blueprint20230829{}
 		errors.Must(db.Fetch(cursor, bp))
 		// decrypt and unmarshal settings
 		settingsJson := errors.Must1(plugin.Decrypt(encKey, bp.Settings))

backend/helpers/migrationhelper/migrationhelper.go

@@ -155,8 +155,8 @@ func TransformColumns[S any, D any](
 				return errors.Default.Wrap(err, fmt.Sprintf("failed to instantiate BatchSave for table [%s]", tableName))
 			}
 			defer batch.Close()
-			src := new(S)
 			for cursor.Next() {
+				src := new(S)
 				err = db.Fetch(cursor, src)
 				if err != nil {
 					return errors.Default.Wrap(err, fmt.Sprintf("fail to load record from table [%s]", tableName))

backend/helpers/pluginhelper/api/api_extractor.go

@@ -93,8 +93,6 @@ func (extractor *ApiExtractor) Execute() errors.Error {
 	}
 	logger.Info("get data from %s where params=%s and got %d", extractor.table, extractor.params, count)
 	defer cursor.Close()
-	row := &RawData{}
-
 	// batch save divider
 	divider := NewBatchSaveDivider(extractor.args.Ctx, extractor.args.BatchSize, extractor.table, extractor.params)
 
@@ -108,6 +106,7 @@ func (extractor *ApiExtractor) Execute() errors.Error {
 			return errors.Convert(ctx.Err())
 		default:
 		}
+		row := &RawData{}
 		err = db.Fetch(cursor, row)
 		if err != nil {
 			return errors.Default.Wrap(err, "error fetching row")

backend/helpers/pluginhelper/api/graphql_collector.go

@@ -309,9 +309,8 @@ func (collector *GraphqlCollector) ExtractExistRawData(divider *BatchSaveDivider
 	}
 	logger.Info("get data from %s where params=%s and got %d", collector.table, collector.params, count)
 	defer cursor.Close()
-	row := &RawData{}
 
-	// prgress
+	// progress
 	collector.args.Ctx.SetProgress(0, -1)
 	ctx := collector.args.Ctx.GetContext()
 	// iterate all rows
@@ -323,6 +322,7 @@ func (collector *GraphqlCollector) ExtractExistRawData(divider *BatchSaveDivider
 		}
 		// get the type of query and variables. For each iteration, the query should be a different object
 		query, variables, _ := collector.args.BuildQuery(nil)
+		row := &RawData{}
 		err = db.Fetch(cursor, row)
 		if err != nil {
 			return errors.Default.Wrap(err, "error fetching row")

backend/plugins/ae/tasks/commits_convertor.go

@@ -28,12 +28,9 @@ import (
 func ConvertCommits(taskCtx plugin.SubTaskContext) errors.Error {
 	db := taskCtx.GetDal()
 	data := taskCtx.GetData().(*AeTaskData)
-
-	aeCommit := &aeModels.AECommit{}
-
 	// Get all the commits from the domain layer
 	cursor, err := db.Cursor(
-		dal.From(aeCommit),
+		dal.From(&aeModels.AECommit{}),
 		dal.Where("ae_project_id = ?", data.Options.ProjectId),
 	)
 	if err != nil {
@@ -53,7 +50,7 @@ func ConvertCommits(taskCtx plugin.SubTaskContext) errors.Error {
 		}
 		// uncomment following line if you want to test out canceling feature for this task
 		//time.Sleep(1 * time.Second)
-
+		aeCommit := &aeModels.AECommit{}
 		err = db.Fetch(cursor, aeCommit)
 		if err != nil {
 			return err

backend/plugins/refdiff/tasks/commit_diff_calculator.go

@@ -76,7 +76,6 @@ func CalculateCommitsDiff(taskCtx plugin.SubTaskContext) errors.Error {
 	insertCountLimitOfCommitsDiff := int(65535 / reflect.ValueOf(code.CommitsDiff{}).NumField())
 
 	// load commits from db
-	commitParent := &code.CommitParent{}
 	cursor, err := db.Cursor(
 		dal.Select("cp.*"),
 		dal.Join("LEFT JOIN repo_commits rc ON (rc.commit_sha = cp.commit_sha)"),
@@ -94,6 +93,7 @@ func CalculateCommitsDiff(taskCtx plugin.SubTaskContext) errors.Error {
 			return errors.Convert(ctx.Err())
 		default:
 		}
+		commitParent := &code.CommitParent{}
 		err = db.Fetch(cursor, commitParent)
 		if err != nil {
 			return errors.Default.Wrap(err, "failed to read commit from database")

backend/plugins/refdiff/tasks/deployment_commit_diff_calculator.go

@@ -156,13 +156,13 @@ func loadCommitGraph(ctx context.Context, db dal.Dal, data *RefdiffTaskData) (*u
 	}
 	defer cursor.Close()
 
-	commitParent := &code.CommitParent{}
 	for cursor.Next() {
 		select {
 		case <-ctx.Done():
 			return nil, errors.Convert(ctx.Err())
 		default:
 		}
+		commitParent := &code.CommitParent{}
 		err = db.Fetch(cursor, commitParent)
 		if err != nil {
 			return nil, err

backend/plugins/refdiff/tasks/refs_pr_cherry_pick_calculator.go

@@ -68,7 +68,6 @@ func CalculatePrCherryPick(taskCtx plugin.SubTaskContext) errors.Error {
 
 	defer cursor.Close()
 
-	pr := &code.PullRequest{}
 	var parentPrKeyInt int
 	taskCtx.SetProgress(0, -1)
 
@@ -79,7 +78,7 @@ func CalculatePrCherryPick(taskCtx plugin.SubTaskContext) errors.Error {
 			return errors.Convert(ctx.Err())
 		default:
 		}
-
+		pr := &code.PullRequest{}
 		err = db.Fetch(cursor, pr)
 		if err != nil {
 			return errors.Convert(err)

backend/server/services/pipeline_runner.go

@@ -145,9 +145,8 @@ func ComputePipelineStatus(pipeline *models.Pipeline, isCancelled bool) (string,
 
 // GetLatestTasksOfPipeline returns latest tasks (reran tasks are excluding) of specified pipeline
 func GetLatestTasksOfPipeline(pipeline *models.Pipeline) ([]*models.Task, errors.Error) {
-	task := &models.Task{}
 	cursor, err := db.Cursor(
-		dal.From(task),
+		dal.From(&models.Task{}),
 		dal.Where("pipeline_id = ?", pipeline.ID),
 		dal.Orderby("id DESC"), // sort it by id so we can hit the latest task first for the RERUNed row/col
 	)
@@ -160,6 +159,7 @@ func GetLatestTasksOfPipeline(pipeline *models.Pipeline) ([]*models.Task, errors
 	type rowcol struct{ row, col int }
 	memorized := make(map[rowcol]bool)
 	for cursor.Next() {
+		task := &models.Task{}
 		if e := db.Fetch(cursor, task); e != nil {
 			return nil, errors.Convert(e)
 		}