[GitHub] [incubator-devlake] abeizn commented on a diff in pull request #3711: fix(tapd): resolve issues impacted incremental collection

[GitBox <gi...@apache.org>]

abeizn commented on code in PR #3711:
URL: https://github.com/apache/incubator-devlake/pull/3711#discussion_r1018697206


##########
plugins/tapd/models/migrationscripts/20221109_add_init_tables.go:
##########
@@ -136,7 +136,7 @@ func (*addInitTables) Up(basicRes core.BasicRes) errors.Error {
 }
 
 func (*addInitTables) Version() uint64 {
-	return 20221013201138

Review Comment:
   need updated version?



##########
plugins/tapd/tasks/story_bug_collector.go:
##########
@@ -42,52 +38,33 @@ func CollectStoryBugs(taskCtx core.SubTaskContext) errors.Error {
 	logger := taskCtx.GetLogger()
 	logger.Info("collect storyBugs")
 	since := data.Since
-	incremental := false
-	if since == nil {
-		// user didn't specify a time range to sync, try load from database
-		var latestUpdated models.TapdStoryCommit
-		clauses := []dal.Clause{
-			dal.Where("connection_id = ? and workspace_id = ?", data.Options.ConnectionId, data.Options.WorkspaceId),
-			dal.Orderby("created DESC"),
-		}
-		err := db.First(&latestUpdated, clauses...)
-		if err != nil && !goerror.Is(err, gorm.ErrRecordNotFound) {
-			return errors.NotFound.Wrap(err, "failed to get latest tapd changelog record")
-		}
-		if latestUpdated.Id > 0 {
-			since = (*time.Time)(latestUpdated.Created)
-			incremental = true
-		}
-	}
-
 	clauses := []dal.Clause{
-		dal.Select("id"),
+		dal.Select("id as issue_id, modified as update_time"),
 		dal.From(&models.TapdStory{}),
-		dal.Where("connection_id = ? and workspace_id = ?", data.Options.ConnectionId, data.Options.WorkspaceId),
+		dal.Where("_tool_tapd_stories.connection_id = ? and _tool_tapd_stories.workspace_id = ? ", data.Options.ConnectionId, data.Options.WorkspaceId),
 	}
 	if since != nil {
 		clauses = append(clauses, dal.Where("modified > ?", since))
 	}
-
 	cursor, err := db.Cursor(clauses...)
 	if err != nil {
 		return err
 	}
-	iterator, err := helper.NewDalCursorIterator(db, cursor, reflect.TypeOf(SimpleStory{}))
+	iterator, err := helper.NewDalCursorIterator(db, cursor, reflect.TypeOf(models.Input{}))
 	if err != nil {
 		return err
 	}
 	collector, err := helper.NewApiCollector(helper.ApiCollectorArgs{
 		RawDataSubTaskArgs: *rawDataSubTaskArgs,
 		ApiClient:          data.ApiClient,
-		Incremental:        incremental,
+		Incremental:        since == nil,

Review Comment:
   same as above



##########
helpers/migrationhelper/migrationhelper.go:
##########
@@ -183,7 +183,7 @@ func TransformTable[S any, D any](
 ) (err errors.Error) {
 	db := basicRes.GetDal()
 	tmpTableName := fmt.Sprintf("%s_%s", tableName, hashScript(script))
-
+	db.Dialect()

Review Comment:
   What is the point of adding this line?



##########
plugins/tapd/tasks/task_commit_collector.go:
##########
@@ -78,23 +52,23 @@ func CollectTaskCommits(taskCtx core.SubTaskContext) errors.Error {
 	if err != nil {
 		return err
 	}
-	iterator, err := helper.NewDalCursorIterator(db, cursor, reflect.TypeOf(SimpleTask{}))
+	iterator, err := helper.NewDalCursorIterator(db, cursor, reflect.TypeOf(models.Input{}))
 	if err != nil {
 		return err
 	}
 	collector, err := helper.NewApiCollector(helper.ApiCollectorArgs{
 		RawDataSubTaskArgs: *rawDataSubTaskArgs,
-		Incremental:        incremental,
+		Incremental:        since == nil,

Review Comment:
   same



##########
plugins/tapd/tasks/bug_commit_collector.go:
##########
@@ -77,24 +52,24 @@ func CollectBugCommits(taskCtx core.SubTaskContext) errors.Error {
 	if err != nil {
 		return err
 	}
-	iterator, err := helper.NewDalCursorIterator(db, cursor, reflect.TypeOf(SimpleBug{}))
+	iterator, err := helper.NewDalCursorIterator(db, cursor, reflect.TypeOf(models.Input{}))
 
 	if err != nil {
 		return err
 	}
 	collector, err := helper.NewApiCollector(helper.ApiCollectorArgs{
 		RawDataSubTaskArgs: *rawDataSubTaskArgs,
-		Incremental:        incremental,
+		Incremental:        since == nil,

Review Comment:
   since != nil ?



##########
plugins/tapd/tasks/story_status_collector.go:
##########
@@ -38,7 +38,6 @@ func CollectStoryStatus(taskCtx core.SubTaskContext) errors.Error {
 	collector, err := helper.NewApiCollector(helper.ApiCollectorArgs{
 		RawDataSubTaskArgs: *rawDataSubTaskArgs,
 		ApiClient:          data.ApiClient,
-		PageSize:           100,

Review Comment:
   why?



##########
plugins/tapd/tasks/iteration_collector.go:
##########
@@ -65,6 +65,7 @@ func CollectIterations(taskCtx core.SubTaskContext) errors.Error {
 		Incremental:        incremental,
 		ApiClient:          data.ApiClient,
 		PageSize:           100,
+		Concurrency:        3,

Review Comment:
   why need to set Concurrency?



##########
plugins/tapd/tasks/story_commit_collector.go:
##########
@@ -78,23 +52,23 @@ func CollectStoryCommits(taskCtx core.SubTaskContext) errors.Error {
 	if err != nil {
 		return err
 	}
-	iterator, err := helper.NewDalCursorIterator(db, cursor, reflect.TypeOf(SimpleStory{}))
+	iterator, err := helper.NewDalCursorIterator(db, cursor, reflect.TypeOf(models.Input{}))
 	if err != nil {
 		return err
 	}
 	collector, err := helper.NewApiCollector(helper.ApiCollectorArgs{
 		RawDataSubTaskArgs: *rawDataSubTaskArgs,
-		Incremental:        incremental,
+		Incremental:        since == nil,

Review Comment:
   same



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@devlake.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org