Support for executing plan from a specified step

Signed-off-by: Sergen Yalçın <[email protected]>

Author: sergenyalcin

pkg/migration/plan_executor.go

@@ -26,9 +26,11 @@ const (
 // PlanExecutor drives the execution of a plan's steps and
 // uses the configured `executors` to execute those steps.
 type PlanExecutor struct {
-	executors []Executor
-	plan      Plan
-	callback  ExecutorCallback
+	executors          []Executor
+	plan               Plan
+	callback           ExecutorCallback
+	LastSuccessfulStep int
+	StartIndex         int
 }
 
 // Action represents an action to be taken by the PlanExecutor.
@@ -69,6 +71,12 @@ func WithExecutorCallback(cb ExecutorCallback) PlanExecutorOption {
 	}
 }
 
+func WithStartIndex(startIndex int) PlanExecutorOption {
+	return func(pe *PlanExecutor) {
+		pe.StartIndex = startIndex
+	}
+}
+
 // ExecutorCallback is the interface for the callback implementations
 // to be notified while executing each Step of a migration Plan.
 type ExecutorCallback interface {
@@ -103,14 +111,15 @@ func NewPlanExecutor(plan Plan, executors []Executor, opts ...PlanExecutorOption
 
 func (pe *PlanExecutor) Execute() error { //nolint:gocyclo // easier to follow this way
 	ctx := make(map[string]any)
-	for i := 0; i < len(pe.plan.Spec.Steps); i++ {
+	for i := pe.StartIndex; i < len(pe.plan.Spec.Steps); i++ {
 		var r CallbackResult
 		if pe.callback != nil {
 			r = pe.callback.StepToExecute(pe.plan.Spec.Steps[i], i)
 			switch r.Action {
 			case ActionCancel:
 				return nil
 			case ActionSkip:
+				pe.LastSuccessfulStep = i
 				continue
 			case ActionContinue, ActionRepeat:
 			}
@@ -124,6 +133,7 @@ func (pe *PlanExecutor) Execute() error { //nolint:gocyclo // easier to follow t
 			}
 		} else if pe.callback != nil {
 			r = pe.callback.StepSucceeded(pe.plan.Spec.Steps[i], i, diag)
+			pe.LastSuccessfulStep = i
 		}
 
 		switch r.Action {

pkg/migration/plan_generator_test.go

@@ -293,7 +293,7 @@ func TestGeneratePlan(t *testing.T) {
 			if err != nil {
 				t.Fatalf("Failed to load plan file from path %s: %v", tt.want.migrationPlanPath, err)
 			}
-			if diff := cmp.Diff(p, &pg.Plan, cmpopts.IgnoreUnexported(Spec{})); diff != "" {
+			if diff := cmp.Diff(p, &pg.Plan, cmpopts.IgnoreUnexported(Plan{}, Spec{})); diff != "" {
 				t.Errorf("GeneratePlan(): -wantPlan, +gotPlan: %s", diff)
 			}
 			// compare generated migration files with the expected ones