[flyteadmin] add delete execution phase api

flyteadmin/pkg/manager/impl/execution_manager.go

@@ -1959,3 +1959,37 @@
 	}
 	return filters, nil
 }
+func (m *ExecutionManager) DeleteExecutionPhase(ctx context.Context, req *admin.ExecutionPhaseDeleteRequest) (*admin.ExecutionPhaseDeleteResponse, error) {
+	executionPhase := req.GetPhase()
+	if executionPhase == core.WorkflowExecution_UNDEFINED {
+		return nil, fmt.Errorf("execution phase cannot be undefined")
+	}
+
+	workflowExecutionID := req.GetId()
+	if workflowExecutionID == nil {
+		return nil, fmt.Errorf("workflow execution identifier cannot be nil")
+	}
+
+	if workflowExecutionID.GetProject() == "" || workflowExecutionID.GetDomain() == "" {
+		return nil, fmt.Errorf("workflow execution identifier must have project, domain")
+	}
+
+	// Wrap executionPhase in ExecutionPhaseDeleteInput
+	input := repositoryInterfaces.ExecutionPhaseDeleteInput{
+		Project:        workflowExecutionID.GetProject(),
+		Domain:         workflowExecutionID.GetDomain(),
+		ExecutionPhase: executionPhase,
+	}
+
+	err := m.db.ExecutionRepo().Delete(ctx, input)
+	if err != nil {
+		return nil, err
+	}
+
+	return &admin.ExecutionPhaseDeleteResponse{
+		Message: fmt.Sprintf("Execution phase %s for workflow %s/%s deleted successfully",
+			executionPhase.String(),
+			workflowExecutionID.GetProject(),
+			workflowExecutionID.GetDomain()),
+	}, nil
+}

flyteadmin/pkg/manager/impl/execution_manager_test.go

@@ -37,6 +37,7 @@ import (
 	"github.com/flyteorg/flyte/flyteadmin/pkg/manager/mocks"
 	managerMocks "github.com/flyteorg/flyte/flyteadmin/pkg/manager/mocks"
 	"github.com/flyteorg/flyte/flyteadmin/pkg/repositories/interfaces"
+	repositoryInterfaces "github.com/flyteorg/flyte/flyteadmin/pkg/repositories/interfaces"
 	repositoryMocks "github.com/flyteorg/flyte/flyteadmin/pkg/repositories/mocks"
 	"github.com/flyteorg/flyte/flyteadmin/pkg/repositories/models"
 	"github.com/flyteorg/flyte/flyteadmin/pkg/repositories/transformers"
@@ -6029,3 +6030,68 @@ func TestQueryTemplate(t *testing.T) {
 		assert.Error(t, err)
 	})
 }
+
+func TestDeleteExecutionPhase(t *testing.T) {
+	t.Run("successful deletion", func(t *testing.T) {
+		repository := repositoryMocks.NewMockRepository()
+
+		deleteExecutionPhaseFunc := func(ctx context.Context, input repositoryInterfaces.ExecutionPhaseDeleteInput) error {
+			assert.Equal(t, "project", input.Project)
+			assert.Equal(t, "domain", input.Domain)
+			assert.Equal(t, core.WorkflowExecution_SUCCEEDED, input.ExecutionPhase)
+			return nil
+		}
+
+		mockExecutionRepo := repository.ExecutionRepo().(*repositoryMocks.MockExecutionRepo)
+		mockExecutionRepo.SetDeleteCallback(deleteExecutionPhaseFunc)
+
+		r := plugins.NewRegistry()
+		r.RegisterDefault(plugins.PluginIDWorkflowExecutor, &defaultTestExecutor)
+		execManager := NewExecutionManager(repository, r, getMockExecutionsConfigProvider(),
+			getMockStorageForExecTest(context.Background()), mockScope.NewTestScope(),
+			mockScope.NewTestScope(), &mockPublisher, mockExecutionRemoteURL, nil, nil, nil, nil,
+			&eventWriterMocks.WorkflowExecutionEventWriter{})
+
+		resp, err := execManager.DeleteExecutionPhase(context.Background(), &admin.ExecutionPhaseDeleteRequest{
+			Id: &core.WorkflowExecutionIdentifier{
+				Project: "project",
+				Domain:  "domain",
+			},
+			Phase: core.WorkflowExecution_SUCCEEDED,
+		})
+
+		assert.NoError(t, err)
+		assert.NotNil(t, resp)
+	})
+
+	t.Run("failed deletion", func(t *testing.T) {
+		repository := repositoryMocks.NewMockRepository()
+
+		deleteExecutionPhaseFunc := func(ctx context.Context, input repositoryInterfaces.ExecutionPhaseDeleteInput) error {
+			assert.Equal(t, "project", input.Project)
+			assert.Equal(t, core.WorkflowExecution_SUCCEEDED, input.ExecutionPhase)
+			return nil
+		}
+
+		mockExecutionRepo := repository.ExecutionRepo().(*repositoryMocks.MockExecutionRepo)
+		mockExecutionRepo.SetDeleteCallback(deleteExecutionPhaseFunc)
+
+		r := plugins.NewRegistry()
+		r.RegisterDefault(plugins.PluginIDWorkflowExecutor, &defaultTestExecutor)
+		execManager := NewExecutionManager(repository, r, getMockExecutionsConfigProvider(),
+			getMockStorageForExecTest(context.Background()), mockScope.NewTestScope(),
+			mockScope.NewTestScope(), &mockPublisher, mockExecutionRemoteURL, nil, nil, nil, nil,
+			&eventWriterMocks.WorkflowExecutionEventWriter{})
+
+		resp, err := execManager.DeleteExecutionPhase(context.Background(), &admin.ExecutionPhaseDeleteRequest{
+			Id: &core.WorkflowExecutionIdentifier{
+				Project: "project",
+			},
+			Phase: core.WorkflowExecution_SUCCEEDED,
+		})
+
+		assert.Error(t, err)
+		assert.Nil(t, resp)
+		assert.Contains(t, err.Error(), "workflow execution identifier must have project, domain")
+	})
+}

flyteadmin/pkg/manager/interfaces/execution.go

@@ -30,4 +30,5 @@ type ExecutionInterface interface {
 	ListExecutions(ctx context.Context, request *admin.ResourceListRequest) (*admin.ExecutionList, error)
 	TerminateExecution(
 		ctx context.Context, request *admin.ExecutionTerminateRequest) (*admin.ExecutionTerminateResponse, error)
+	DeleteExecutionPhase(ctx context.Context, request *admin.ExecutionPhaseDeleteRequest) (*admin.ExecutionPhaseDeleteResponse, error)
 }

flyteadmin/pkg/repositories/gormimpl/execution_repo.go

@@ -170,3 +170,17 @@
 		metrics:          metrics,
 	}
 }
+
+func (r *ExecutionRepo) Delete(ctx context.Context, input interfaces.ExecutionPhaseDeleteInput) error {
+	result := r.db.Delete(&models.Execution{},
+		"execution_project = ? AND execution_domain = ? AND phase = ?",
+		input.Project,
+		input.Domain,
+		input.ExecutionPhase.String())
+
+	if result.Error != nil {
+		return result.Error
+	}
+
+	return nil
+}

flyteadmin/pkg/repositories/interfaces/common.go

@@ -2,6 +2,7 @@ package interfaces
 
 import (
 	"github.com/flyteorg/flyte/flyteadmin/pkg/common"
+	"github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/core"
 )
 
 // Parameters for getting an individual resource.
@@ -44,3 +45,9 @@ type CountResourceInput struct {
 	// the count query. This enables filtering on non-primary entity attributes.
 	JoinTableEntities map[common.Entity]bool
 }
+
+type ExecutionPhaseDeleteInput struct {
+	Project        string
+	Domain         string
+	ExecutionPhase core.WorkflowExecution_Phase
+}

flyteadmin/pkg/repositories/interfaces/execution_repo.go

@@ -18,6 +18,7 @@ type ExecutionRepoInterface interface {
 	List(ctx context.Context, input ListResourceInput) (ExecutionCollectionOutput, error)
 	// Returns count of executions matching query parameters.
 	Count(ctx context.Context, input CountResourceInput) (int64, error)
+	Delete(ctx context.Context, input ExecutionPhaseDeleteInput) error
 }
 
 // Response format for a query on workflows.

flyteadmin/pkg/repositories/mocks/execution_repo.go

@@ -13,13 +13,15 @@ type GetExecutionFunc func(ctx context.Context, input interfaces.Identifier) (mo
 type ListExecutionFunc func(ctx context.Context, input interfaces.ListResourceInput) (
 	interfaces.ExecutionCollectionOutput, error)
 type CountExecutionFunc func(ctx context.Context, input interfaces.CountResourceInput) (int64, error)
+type DeleteExecutionFunc func(ctx context.Context, input interfaces.ExecutionPhaseDeleteInput) error
 
 type MockExecutionRepo struct {
 	createFunction CreateExecutionFunc
 	updateFunction UpdateExecutionFunc
 	getFunction    GetExecutionFunc
 	listFunction   ListExecutionFunc
 	countFunction  CountExecutionFunc
+	deleteFunction DeleteExecutionFunc
 }
 
 func (r *MockExecutionRepo) Create(ctx context.Context, input models.Execution, executionTagModel []*models.ExecutionTag) error {
@@ -81,3 +83,14 @@ func (r *MockExecutionRepo) SetCountCallback(countFunction CountExecutionFunc) {
 func NewMockExecutionRepo() interfaces.ExecutionRepoInterface {
 	return &MockExecutionRepo{}
 }
+
+func (r *MockExecutionRepo) Delete(ctx context.Context, input interfaces.ExecutionPhaseDeleteInput) error {
+	if r.deleteFunction != nil {
+		return r.deleteFunction(ctx, input)
+	}
+	return nil
+}
+
+func (r *MockExecutionRepo) SetDeleteCallback(deleteFunction func(context.Context, interfaces.ExecutionPhaseDeleteInput) error) {
+	r.deleteFunction = deleteFunction
+}

flyteadmin/pkg/rpc/adminservice/execution.go

@@ -152,3 +152,17 @@
 	m.Metrics.executionEndpointMetrics.terminate.Success()
 	return response, nil
 }
+
+func (m *AdminService) DeleteExecutionPhase(
+	ctx context.Context, request *admin.ExecutionPhaseDeleteRequest) (*admin.ExecutionPhaseDeleteResponse, error) {
+	var response *admin.ExecutionPhaseDeleteResponse
+	var err error
+	m.Metrics.executionEndpointMetrics.terminate.Time(func() {
+		response, err = m.ExecutionManager.DeleteExecutionPhase(ctx, request)
+	})
+	if err != nil {
+		return nil, util.TransformAndRecordError(err, &m.Metrics.executionEndpointMetrics.terminate)
+	}
+	m.Metrics.executionEndpointMetrics.terminate.Success()
+	return response, nil
+}