feat(github): Add incremental data collection

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,