feat(github): Add incremental data collection (#8858)

* fix(github): use RFC 3339 format for since= incremental API parameters

The since= query parameter passed to the GitHub API in four collectors
was formatted using Go's time.Time.String(), which produces a
human-readable string (e.g. "2024-01-15 10:30:00 +0000 UTC") rather
than the ISO 8601 / RFC 3339 format the GitHub API requires
(e.g. "2024-01-15T10:30:00Z").

The GitHub API silently ignores malformed date strings, causing these
collectors to perform full re-scans on every incremental run despite
appearing to filter correctly. Fix by using .UTC().Format(time.RFC3339)
in all four affected collectors:
- comment_collector.go
- issue_collector.go
- commit_collector.go
- pr_review_comment_collector.go

(cherry picked from commit cf3cb621462d8ae661df5fcb8e1b47c70564cd60)

* feat(github): migrate event extractor to StatefulApiExtractor

Switch Extract Events from the legacy full-scan NewApiExtractor to
NewStatefulApiExtractor, which filters _raw_github_api_events by
created_at >= last_run_start on incremental syncs. This table had
13,828 rows in a representative production run and took ~117s to
process on every run regardless of how few new events were collected.
After this change incremental runs process only newly collected rows.

(cherry picked from commit 84f324f65e8576cd2ff0ca8b74a2588ee2600e12)

* feat(github): migrate PR extractor to StatefulApiExtractor

Switch Extract Pull Requests from legacy full-scan NewApiExtractor to
NewStatefulApiExtractor. The _raw_github_api_pull_requests table had
12,448 rows in a representative production run, taking ~108s on every
incremental sync. After this change only newly collected PR rows are
processed.

SubtaskConfig captures prType and prComponent regex strings so that a
scope config change automatically triggers a full re-extract. BeforeExtract
deletes GithubPrLabel rows for the current PR before re-inserting them
in incremental mode, preventing stale labels from persisting when labels
are removed upstream.

(cherry picked from commit 80dc76d4e5f1ad6c65d48d42f60f50b87c9dad2d)

* feat(github): migrate workflow run and PR commit extractors to StatefulApiExtractor

Switch Extract Workflow Runs (~7,364 raw rows, ~65s) and Extract PR
Commits (~6,982 raw rows, ~60s) from legacy full-scan NewApiExtractor
to NewStatefulApiExtractor. Both extractors are simple mappings with
no scope-config dependency, so no SubtaskConfig or BeforeExtract needed.

(cherry picked from commit ea4f158a65ad0f7a2c23ba7bbc9932059e2ca408)

* feat(github): migrate remaining high-volume extractors to StatefulApiExtractor

Migrate Extract Jobs (~5,369 rows, ~48s), Extract PR Reviews (~3,073
rows, ~27s), and Extract PR Review Comments (~1,820 rows, ~16s) from
legacy full-scan NewApiExtractor to NewStatefulApiExtractor.

Also moves prUrlRegex compilation in pr_review_comment_extractor.go
from inside the Extract closure (recompiled on every raw row) to before
the extractor is created, eliminating redundant regexp compilation.

(cherry picked from commit 54d601587bc74ebd0f1103345c545571146739b5)

* feat(github): migrate remaining low-volume extractors to StatefulApiExtractor

Migrate the final seven GitHub extractors to NewStatefulApiExtractor:
issue, comment, account, account_org, milestone, commit, commit_stats.

issue_extractor gains SubtaskConfig (issue classification regex strings)
so scope config changes trigger automatic full re-extraction, and
BeforeExtract cleanup for GithubIssueLabel and GithubIssueAssignee rows
in incremental mode to prevent stale labels/assignees persisting after
upstream removal.

All other extractors in this commit are simple migrations with no
config-sensitivity or child record cleanup needed.

With this commit all 14 GitHub plugin extractors are now incremental.
Combined with the collector fixes in earlier commits, incremental
collection runs that previously took 9+ minutes in the extract phase
will now complete in seconds when few or no new records were collected.

(cherry picked from commit 606acea88caef63662733162faa47a6c6d3155cc)

* feat(github): migrate CICD converters to StatefulDataConverter

Convert Workflow Runs and Convert Jobs now use NewStatefulDataConverter,
skipping records unchanged since last run on incremental pipelines.
Jobs are filtered via JOIN on _tool_github_runs.github_updated_at.

(cherry picked from commit 75a909efccab1d04bfae4058f4674663b00762a6)

* feat(github): migrate PR supporting data converters to StatefulDataConverter

Convert PR Commits, Convert PR Comments, and Convert PR Reviews now use
NewStatefulDataConverter. Child-of-PR records are filtered incrementally
via JOIN on _tool_github_pull_requests.github_updated_at; PR comments
additionally filter on their own github_updated_at.

(cherry picked from commit 17cbc2e4caff7349fa2d3389d7cafcc5d0edee71)

* feat(github): migrate PR main and cross converters to StatefulDataConverter

Convert Pull Requests filters by GithubPullRequest.github_updated_at.
Convert Reviews and Convert PR Issues filter via JOIN on pull_requests
github_updated_at since reviewers and pr_issues have no own timestamp.

(cherry picked from commit 47437ae49d30b967b443c964a4b1baca34344acc)

* feat(github): migrate remaining data converters to StatefulDataConverter

Migrate the last 9 converters from DataConverter to StatefulDataConverter
so they skip already-processed records on incremental runs:

- issue_convertor: filter on github_updated_at
- issue_comment_convertor: filter on github_updated_at
- issue_label_convertor: JOIN to issues, filter on issues.github_updated_at
- issue_assignee_convertor: JOIN to issues, filter on issues.github_updated_at
- pr_label_convertor: JOIN to pull_requests, filter on pr.github_updated_at
- account_convertor: filter on updated_at
- release_convertor: filter on updated_at
- repo_convertor: filter on updated_at; retain GithubApiRepo struct (used by pr_extractor)
- commit_convertor: filter on authored_date

(cherry picked from commit 7bcc9da4676d34d9cd24e41a68765cf85723ac81)

* fix(github): keep incremental commit and PR-issue processing accurate

* fix(github): bootstrap workflow runs incremental window from tool state

* feat(runner): include processed record count in subtask finish logs

* fix(stateful): bootstrap subtask state from collector checkpoints

* fix(github): optimize incremental Convert Jobs query

* fix(github): resolve lint and staticcheck issues in converters

* fix(stateful): tolerate missing collector state table in bootstrap

* fix(github): address CI failures in state bootstrap and run tests

* fix(server): use interface{} for store handler swag annotations (#8859)

Signed-off-by: yamoyamoto <yamo7yamoto@gmail.com>

* fix(github): stabilize e2e state bootstrap and issue assignee join

* fix(stateful): skip collector bootstrap lookup when state table is absent

* test(stateful): use generic scope params in bootstrap tests

* fix(github): keep issue assignee conversion full-sync safe

---------

Signed-off-by: yamoyamoto <yamo7yamoto@gmail.com>
Co-authored-by: Tomoya Kawaguchi <68677002+yamoyamoto@users.noreply.github.com>

Author: lrf-nitro

backend/core/plugin/plugin_task.go

@@ -53,6 +53,7 @@ type ExecContext interface {
 	GetData() interface{}
 	SetProgress(current int, total int)
 	IncProgress(quantity int)
+	GetProgress() int
 }
 
 // SubTaskContext This interface define all resources that needed for subtask execution

backend/core/runner/run_task.go

@@ -336,7 +336,7 @@ func RunPluginSubTasks(
 			logger.Info("executing subtask %s", subtaskMeta.Name)
 			start := time.Now()
 			err = runSubtask(basicRes, subtaskCtx, task.ID, subtaskNumber, subtaskMeta.EntryPoint)
-			logger.Info("subtask %s finished in %d ms", subtaskMeta.Name, time.Since(start).Milliseconds())
+			logger.Info("subtask %s finished in %d ms, records processed: %d", subtaskMeta.Name, time.Since(start).Milliseconds(), subtaskCtx.GetProgress())
 			if err != nil {
 				err = errors.SubtaskErr.Wrap(err, fmt.Sprintf("subtask %s ended unexpectedly", subtaskMeta.Name), errors.WithData(&subtaskMeta))
 				logger.Error(err, "")

backend/helpers/pluginhelper/api/subtask_state_manager.go

@@ -20,6 +20,7 @@ package api
 import (
 	"fmt"
 	"reflect"
+	"strings"
 	"time"
 
 	"github.com/apache/incubator-devlake/core/dal"
@@ -88,6 +89,10 @@ func NewSubtaskStateManager(args *SubtaskCommonArgs) (stateManager *SubtaskState
 	if err != nil {
 		return
 	}
+	preState, err = bootstrapStateFromCollectorStateIfNeeded(db, preState, args)
+	if err != nil {
+		return
+	}
 
 	isIncremental, since := calculateStateManagerIncrementalMode(syncPolicy, preState, utils.ToJsonString(args.SubtaskConfig))
 
@@ -127,6 +132,56 @@ func loadPreviousState(db dal.Dal, plugin, subtask, params string) (*models.Subt
 	return preState, nil
 }
 
+func bootstrapStateFromCollectorStateIfNeeded(db dal.Dal, preState *models.SubtaskState, args *SubtaskCommonArgs) (*models.SubtaskState, errors.Error) {
+	if preState == nil || preState.PrevStartedAt != nil {
+		return preState, nil
+	}
+	if args == nil || args.Table == "" {
+		return preState, nil
+	}
+	if !db.HasTable(&models.CollectorLatestState{}) {
+		return preState, nil
+	}
+
+	rawTable := args.GetRawDataTable()
+	if rawTable == "" {
+		return preState, nil
+	}
+
+	collectorState := &models.CollectorLatestState{}
+	err := db.First(
+		collectorState,
+		dal.Where("raw_data_table = ? AND raw_data_params = ?", rawTable, preState.Params),
+	)
+	if err != nil {
+		if db.IsErrorNotFound(err) || isStateTableNotReadyError(err) {
+			return preState, nil
+		}
+		return nil, errors.Default.Wrap(err, "failed to load collector state for subtask bootstrap")
+	}
+
+	if collectorState.LatestSuccessStart != nil {
+		preState.PrevStartedAt = collectorState.LatestSuccessStart
+	}
+	if preState.TimeAfter == nil && collectorState.TimeAfter != nil {
+		preState.TimeAfter = collectorState.TimeAfter
+	}
+
+	return preState, nil
+}
+
+func isStateTableNotReadyError(err error) bool {
+	if err == nil {
+		return false
+	}
+	msg := strings.ToLower(err.Error())
+	return strings.Contains(msg, "_devlake_collector_latest_state") &&
+		(strings.Contains(msg, "doesn't exist") ||
+			strings.Contains(msg, "does not exist") ||
+			strings.Contains(msg, "unknown table") ||
+			strings.Contains(msg, "no such table"))
+}
+
 // calculateStateManagerIncrementalMode tries to calculate whether state manager should run in incremental mode and returns the state manager's 'since' time.
 func calculateStateManagerIncrementalMode(syncPolicy *models.SyncPolicy, preState *models.SubtaskState, newSubtaskConfig string) (bool, *time.Time) {
 	if preState == nil || syncPolicy == nil {

backend/helpers/pluginhelper/api/subtask_state_manager_test.go

@@ -30,6 +30,8 @@ import (
 	"github.com/stretchr/testify/mock"
 )
 
+const testGithubScopeParams = `{"ConnectionId":1,"Name":"test/repo"}`
+
 func TestSubtaskStateManager(t *testing.T) {
 	time0 := errors.Must1(time.Parse(time.RFC3339, "2020-01-01T00:00:00Z"))
 	time1 := errors.Must1(time.Parse(time.RFC3339, "2021-01-01T00:00:00Z"))
@@ -187,3 +189,170 @@ func TestSubtaskStateManager(t *testing.T) {
 		})
 	}
 }
+
+func TestBootstrapStateFromCollectorStateIfNeeded(t *testing.T) {
+	latest := errors.Must1(time.Parse(time.RFC3339, "2026-04-27T18:05:31Z"))
+	timeAfter := errors.Must1(time.Parse(time.RFC3339, "2024-12-20T00:00:00Z"))
+
+	t.Run("bootstrap prev_started_at and time_after from collector state", func(t *testing.T) {
+		mockDal := new(mockdal.Dal)
+		mockDal.On("HasTable", mock.Anything).Return(true).Once()
+		mockDal.On("First", mock.Anything, mock.Anything).Run(func(args mock.Arguments) {
+			dst := args.Get(0).(*models.CollectorLatestState)
+			dst.LatestSuccessStart = &latest
+			dst.TimeAfter = &timeAfter
+		}).Return(nil).Once()
+
+		state := &models.SubtaskState{
+			Plugin:  "github",
+			Subtask: "Convert Jobs",
+			Params:  testGithubScopeParams,
+		}
+		args := &SubtaskCommonArgs{Table: "github_api_jobs"}
+
+		bootstrapped, err := bootstrapStateFromCollectorStateIfNeeded(mockDal, state, args)
+		assert.Nil(t, err)
+		if assert.NotNil(t, bootstrapped.PrevStartedAt) {
+			assert.True(t, bootstrapped.PrevStartedAt.Equal(latest))
+		}
+		if assert.NotNil(t, bootstrapped.TimeAfter) {
+			assert.True(t, bootstrapped.TimeAfter.Equal(timeAfter))
+		}
+		mockDal.AssertExpectations(t)
+	})
+
+	t.Run("do not overwrite existing prev_started_at", func(t *testing.T) {
+		mockDal := new(mockdal.Dal)
+
+		existing := errors.Must1(time.Parse(time.RFC3339, "2026-04-27T18:06:37Z"))
+		state := &models.SubtaskState{
+			Plugin:        "github",
+			Subtask:       "Convert Jobs",
+			Params:        testGithubScopeParams,
+			PrevStartedAt: &existing,
+		}
+		args := &SubtaskCommonArgs{Table: "github_api_jobs"}
+
+		bootstrapped, err := bootstrapStateFromCollectorStateIfNeeded(mockDal, state, args)
+		assert.Nil(t, err)
+		if assert.NotNil(t, bootstrapped.PrevStartedAt) {
+			assert.True(t, bootstrapped.PrevStartedAt.Equal(existing))
+		}
+		mockDal.AssertNotCalled(t, "HasTable", mock.Anything)
+		mockDal.AssertNotCalled(t, "First", mock.Anything, mock.Anything)
+	})
+
+	t.Run("ignore not found collector state", func(t *testing.T) {
+		mockDal := new(mockdal.Dal)
+		notFoundErr := errors.Default.New("record not found")
+		mockDal.On("HasTable", mock.Anything).Return(true).Once()
+		mockDal.On("First", mock.Anything, mock.Anything).Return(notFoundErr).Once()
+		mockDal.On("IsErrorNotFound", notFoundErr).Return(true).Once()
+
+		state := &models.SubtaskState{
+			Plugin:  "github",
+			Subtask: "Convert Jobs",
+			Params:  testGithubScopeParams,
+		}
+		args := &SubtaskCommonArgs{Table: "github_api_jobs"}
+
+		bootstrapped, err := bootstrapStateFromCollectorStateIfNeeded(mockDal, state, args)
+		assert.Nil(t, err)
+		assert.Nil(t, bootstrapped.PrevStartedAt)
+		assert.Nil(t, bootstrapped.TimeAfter)
+		mockDal.AssertExpectations(t)
+	})
+
+	t.Run("return error when collector state query fails", func(t *testing.T) {
+		mockDal := new(mockdal.Dal)
+		dbErr := errors.Default.New("db unavailable")
+		mockDal.On("HasTable", mock.Anything).Return(true).Once()
+		mockDal.On("First", mock.Anything, mock.Anything).Return(dbErr).Once()
+		mockDal.On("IsErrorNotFound", dbErr).Return(false).Once()
+
+		state := &models.SubtaskState{
+			Plugin:  "github",
+			Subtask: "Convert Jobs",
+			Params:  testGithubScopeParams,
+		}
+		args := &SubtaskCommonArgs{Table: "github_api_jobs"}
+
+		bootstrapped, err := bootstrapStateFromCollectorStateIfNeeded(mockDal, state, args)
+		assert.Nil(t, bootstrapped)
+		assert.NotNil(t, err)
+		assert.Contains(t, err.Error(), "failed to load collector state for subtask bootstrap")
+		mockDal.AssertExpectations(t)
+	})
+
+	t.Run("ignore missing collector table errors", func(t *testing.T) {
+		mockDal := new(mockdal.Dal)
+		mockDal.On("HasTable", mock.Anything).Return(true).Once()
+		tableErr := errors.Default.New("Error 1146 (42S02): Table 'lake._devlake_collector_latest_state' doesn't exist")
+		mockDal.On("First", mock.Anything, mock.Anything).Return(tableErr).Once()
+		mockDal.On("IsErrorNotFound", tableErr).Return(false).Once()
+
+		state := &models.SubtaskState{
+			Plugin:  "github",
+			Subtask: "Convert Jobs",
+			Params:  testGithubScopeParams,
+		}
+		args := &SubtaskCommonArgs{Table: "github_api_jobs"}
+
+		bootstrapped, err := bootstrapStateFromCollectorStateIfNeeded(mockDal, state, args)
+		assert.Nil(t, err)
+		assert.NotNil(t, bootstrapped)
+		assert.Nil(t, bootstrapped.PrevStartedAt)
+		assert.Nil(t, bootstrapped.TimeAfter)
+		mockDal.AssertExpectations(t)
+	})
+
+	t.Run("do not ignore missing unrelated table errors", func(t *testing.T) {
+		mockDal := new(mockdal.Dal)
+		mockDal.On("HasTable", mock.Anything).Return(true).Once()
+		tableErr := errors.Default.New("Error 1146 (42S02): Table 'lake._tool_github_issues' doesn't exist")
+		mockDal.On("First", mock.Anything, mock.Anything).Return(tableErr).Once()
+		mockDal.On("IsErrorNotFound", tableErr).Return(false).Once()
+
+		state := &models.SubtaskState{
+			Plugin:  "github",
+			Subtask: "Convert Jobs",
+			Params:  testGithubScopeParams,
+		}
+		args := &SubtaskCommonArgs{Table: "github_api_jobs"}
+
+		bootstrapped, err := bootstrapStateFromCollectorStateIfNeeded(mockDal, state, args)
+		assert.Nil(t, bootstrapped)
+		assert.NotNil(t, err)
+		assert.Contains(t, err.Error(), "failed to load collector state for subtask bootstrap")
+		mockDal.AssertExpectations(t)
+	})
+
+	t.Run("skip collector lookup when collector state table does not exist", func(t *testing.T) {
+		mockDal := new(mockdal.Dal)
+		mockDal.On("HasTable", mock.Anything).Return(false).Once()
+
+		state := &models.SubtaskState{
+			Plugin:  "github",
+			Subtask: "Convert Jobs",
+			Params:  testGithubScopeParams,
+		}
+		args := &SubtaskCommonArgs{Table: "github_api_jobs"}
+
+		bootstrapped, err := bootstrapStateFromCollectorStateIfNeeded(mockDal, state, args)
+		assert.Nil(t, err)
+		assert.NotNil(t, bootstrapped)
+		assert.Nil(t, bootstrapped.PrevStartedAt)
+		assert.Nil(t, bootstrapped.TimeAfter)
+		mockDal.AssertNotCalled(t, "First", mock.Anything, mock.Anything)
+		mockDal.AssertExpectations(t)
+	})
+}
+
+func TestIsStateTableNotReadyError(t *testing.T) {
+	assert.False(t, isStateTableNotReadyError(nil))
+	assert.True(t, isStateTableNotReadyError(errors.Default.New("Error 1146 (42S02): Table 'lake._devlake_collector_latest_state' doesn't exist")))
+	assert.True(t, isStateTableNotReadyError(errors.Default.New("pq: relation \"_devlake_collector_latest_state\" does not exist")))
+	assert.True(t, isStateTableNotReadyError(errors.Default.New("no such table: _devlake_collector_latest_state")))
+	assert.False(t, isStateTableNotReadyError(errors.Default.New("Error 1146 (42S02): Table 'lake._tool_github_issues' doesn't exist")))
+	assert.False(t, isStateTableNotReadyError(errors.Default.New("db unavailable")))
+}

backend/impls/context/default_exec_context.go

@@ -96,6 +96,10 @@ func (c *defaultExecContext) IncProgress(progressType plugin.ProgressType, quant
 	}
 }
 
+func (c *defaultExecContext) GetProgress() int {
+	return int(atomic.LoadInt64(&c.current))
+}
+
 func (c *defaultExecContext) fork(name string) *defaultExecContext {
 	return newDefaultExecContext(
 		c.ctx,

backend/plugins/github/tasks/account_convertor.go

@@ -18,12 +18,10 @@ limitations under the License.
 package tasks
 
 import (
-	"reflect"
 	"strings"
 
 	"github.com/apache/incubator-devlake/core/dal"
 	"github.com/apache/incubator-devlake/core/errors"
-	"github.com/apache/incubator-devlake/core/models/common"
 	"github.com/apache/incubator-devlake/core/models/domainlayer"
 	"github.com/apache/incubator-devlake/core/models/domainlayer/crossdomain"
 	"github.com/apache/incubator-devlake/core/models/domainlayer/didgen"
@@ -49,50 +47,44 @@ var ConvertAccountsMeta = plugin.SubTaskMeta{
 	ProductTables: []string{crossdomain.Account{}.TableName()},
 }
 
-type GithubAccountWithOrg struct {
-	models.GithubAccount
-	Login string `json:"login" gorm:"type:varchar(255)"`
-	common.NoPKModel
-}
-
 func ConvertAccounts(taskCtx plugin.SubTaskContext) errors.Error {
 	db := taskCtx.GetDal()
 	data := taskCtx.GetData().(*GithubTaskData)
 
-	cursor, err := db.Cursor(
-		dal.Select("_tool_github_accounts.*"),
-		dal.From(&models.GithubAccount{}),
-		dal.Where(
-			"repo_github_id = ? and _tool_github_accounts.connection_id=?",
-			data.Options.GithubId,
-			data.Options.ConnectionId,
-		),
-		dal.Join(`left join _tool_github_repo_accounts gra on (
-			_tool_github_accounts.connection_id = gra.connection_id
-			AND _tool_github_accounts.id = gra.account_id
-		)`),
-	)
-	if err != nil {
-		return err
-	}
-	defer cursor.Close()
-
 	accountIdGen := didgen.NewDomainIdGenerator(&models.GithubAccount{})
 
-	converter, err := api.NewDataConverter(api.DataConverterArgs{
-		InputRowType: reflect.TypeOf(models.GithubAccount{}),
-		Input:        cursor,
-		RawDataSubTaskArgs: api.RawDataSubTaskArgs{
-			Ctx: taskCtx,
+	converter, err := api.NewStatefulDataConverter(&api.StatefulDataConverterArgs[models.GithubAccount]{
+		SubtaskCommonArgs: &api.SubtaskCommonArgs{
+			SubTaskContext: taskCtx,
+			Table:          RAW_ACCOUNT_TABLE,
 			Params: GithubApiParams{
 				ConnectionId: data.Options.ConnectionId,
 				Name:         data.Options.Name,
 			},
-			Table: RAW_ACCOUNT_TABLE,
 		},
-		Convert: func(inputRow interface{}) ([]interface{}, errors.Error) {
-			githubUser := inputRow.(*models.GithubAccount)
-
+		Input: func(stateManager *api.SubtaskStateManager) (dal.Rows, errors.Error) {
+			clauses := []dal.Clause{
+				dal.Select("_tool_github_accounts.*"),
+				dal.From(&models.GithubAccount{}),
+				dal.Where(
+					"repo_github_id = ? and _tool_github_accounts.connection_id=?",
+					data.Options.GithubId,
+					data.Options.ConnectionId,
+				),
+				dal.Join(`left join _tool_github_repo_accounts gra on (
+					_tool_github_accounts.connection_id = gra.connection_id
+					AND _tool_github_accounts.id = gra.account_id
+				)`),
+			}
+			if stateManager.IsIncremental() {
+				since := stateManager.GetSince()
+				if since != nil {
+					clauses = append(clauses, dal.Where("_tool_github_accounts.updated_at >= ?", since))
+				}
+			}
+			return db.Cursor(clauses...)
+		},
+		Convert: func(githubUser *models.GithubAccount) ([]interface{}, errors.Error) {
 			// query related orgs
 			var orgs []string
 			err := db.Pluck(`org_login`, &orgs,