Clean up old converted data when no convertible files found

dataviewer/workflows/activity.go

@@ -44,6 +44,7 @@ type DataViewerActivity interface {
 	UploadParquetFiles(ctx context.Context, uploadReq dvCom.UploadParquetReq) (*dvCom.CardData, error)
 	UpdateCardData(ctx context.Context, cardReq dvCom.UpdateCardReq) error
 	CleanUp(ctx context.Context, req types.UpdateViewerReq) error
+	CleanupOldViewerData(ctx context.Context, req types.UpdateViewerReq) error
 	UpdateWorkflowStatus(ctx context.Context, status dvCom.UpdateWorkflowStatusReq) error
 	CalcStatistics(cardReq *dvCom.UpdateCardReq)
 }
@@ -1108,3 +1109,46 @@ func (dva *dataViewerActivityImpl) UpdateWorkflowStatus(ctx context.Context, sta
 	}
 	return nil
 }
+
+func (dva *dataViewerActivityImpl) CleanupOldViewerData(ctx context.Context, req types.UpdateViewerReq) error {
+	// Delete old parquet branch if exists
+	findReq := gitserver.GetBranchReq{
+		Namespace: req.Namespace,
+		Name:      req.Name,
+		Ref:       dvCom.ParquetBranch,
+		RepoType:  req.RepoType,
+	}
+	branch, err := dva.gitServer.GetRepoBranchByName(ctx, findReq)
+	if err != nil {
+		slog.WarnContext(ctx, "get parquet branch for cleanup", slog.Any("req", req), slog.Any("error", err))
+	}
+	if err == nil && branch != nil {
+		deleteReq := gitserver.DeleteBranchReq{
+			Namespace: req.Namespace,
+			Name:      req.Name,
+			Ref:       dvCom.ParquetBranch,
+			RepoType:  req.RepoType,
+			Username:  GitDefaultUserName,
+			Email:     GitDefaultUserEmail,
+		}
+		if err := dva.gitServer.DeleteRepoBranch(ctx, deleteReq); err != nil {
+			return fmt.Errorf("delete old parquet branch %s for cleanup, cause: %w", dvCom.ParquetBranch, err)
+		}
+	}
+
+	// Clear card data in the current job
+	wfCtx := activity.GetInfo(ctx)
+	workflowID := wfCtx.WorkflowExecution.ID
+	job, err := dva.viewerStore.GetJob(ctx, workflowID)
+	if err != nil {
+		return fmt.Errorf("get job by workflow id %s for cleanup, cause: %w", workflowID, err)
+	}
+	job.AutoCard = true
+	job.CardData = ""
+	job.CardMD5 = ""
+	if _, err := dva.viewerStore.UpdateJob(ctx, *job); err != nil {
+		return fmt.Errorf("update job for cleanup, cause: %w", err)
+	}
+
+	return nil
+}

dataviewer/workflows/workflow.go

@@ -36,6 +36,7 @@ func BuildDataViewerRunWorker(tc temporal.Client, cfg *config.Config, act DataVi
 	wr.RegisterActivity(act.UploadParquetFiles)
 	wr.RegisterActivity(act.UpdateCardData)
 	wr.RegisterActivity(act.CleanUp)
+	wr.RegisterActivity(act.CleanupOldViewerData)
 	wr.RegisterActivity(act.UpdateWorkflowStatus)
 
 	wr.RegisterWorkflow(DataViewerUpdateWorkflow)
@@ -131,6 +132,16 @@ func runWorkFlow(sessionCtx workflow.Context, updateWorkflow dvCom.WorkflowUpdat
 		return false, fmt.Errorf("run data viewer activity DetermineCardData error: %w", err)
 	}
 
+	// cleanup old viewer data if no convertible files found
+	if len(repoDataType) < 1 {
+		slog.Warn("no convertible files found, cleanup old viewer data")
+		err = workflow.ExecuteActivity(sessionCtx, DataViewerActivity.CleanupOldViewerData, updateWorkflow.Req).Get(sessionCtx, nil)
+		if err != nil {
+			return false, fmt.Errorf("run data viewer activity CleanupOldViewerData error: %w", err)
+		}
+		return true, nil
+	}
+
 	var needRebuild bool
 	err = workflow.ExecuteActivity(sessionCtx, DataViewerActivity.CheckIfNeedRebuild, dvCom.CheckBuildReq{
 		Req:  updateWorkflow.Req,

dataviewer/workflows/workflow_test.go

@@ -229,4 +229,64 @@ func TestWorkflow_DataviewerWorkflow(t *testing.T) {
 		require.NoError(t, tester.env.GetWorkflowError())
 	})
 
+	t.Run("no convertible files/cleanup old data", func(t *testing.T) {
+		tester, err := newWorkflowTester(t)
+		require.NoError(t, err)
+
+		cfg, err := config.LoadConfig()
+		require.Nil(t, err)
+
+		req := types.UpdateViewerReq{
+			Namespace: "test",
+			Name:      "test",
+			Branch:    "test",
+			RepoType:  types.DatasetRepo,
+		}
+
+		tester.mocks.mockact.EXPECT().BeginViewerJob(mock.Anything).Return(nil)
+		tester.mocks.mockact.EXPECT().GetCardFromReadme(mock.Anything, req).Return(&dvCom.CardData{}, nil)
+		tester.mocks.mockact.EXPECT().ScanRepoFiles(mock.Anything,
+			dvCom.ScanRepoFileReq{
+				Req:              req,
+				ConvertLimitSize: cfg.DataViewer.ConvertLimitSize,
+				MaxFileNum:       cfg.DataViewer.ScanFileNumLimit,
+			}).Return(&dvCom.RepoFilesClass{
+			AllFiles:     map[string]*dvCom.RepoFile{},
+			ParquetFiles: map[string]*dvCom.RepoFile{},
+			JsonlFiles:   map[string]*dvCom.RepoFile{},
+			CsvFiles:     map[string]*dvCom.RepoFile{},
+		}, nil)
+
+		// DetermineCardData returns nil card (Configs == nil) when no scope files found
+		tester.mocks.mockact.EXPECT().DetermineCardData(mock.Anything,
+			dvCom.DetermineCardReq{
+				Card: dvCom.CardData{},
+				Class: dvCom.RepoFilesClass{
+					AllFiles:     map[string]*dvCom.RepoFile{},
+					ParquetFiles: map[string]*dvCom.RepoFile{},
+					JsonlFiles:   map[string]*dvCom.RepoFile{},
+					CsvFiles:     map[string]*dvCom.RepoFile{},
+				},
+				RepoDataType: "",
+			},
+		).Return(nil, nil)
+
+		// Activity: cleanup old viewer data
+		tester.mocks.mockact.EXPECT().CleanupOldViewerData(mock.Anything, req).Return(nil)
+
+		// Update workflow status: success, should update viewer
+		tester.mocks.mockact.EXPECT().UpdateWorkflowStatus(mock.Anything, dvCom.UpdateWorkflowStatusReq{
+			Req:                req,
+			WorkflowErrMsg:     "",
+			ShouldUpdateViewer: true,
+		}).Return(nil)
+
+		tester.env.ExecuteWorkflow(DataViewerUpdateWorkflow, dvCom.WorkflowUpdateParams{
+			Req:    req,
+			Config: cfg,
+		})
+
+		require.True(t, tester.env.IsWorkflowCompleted())
+		require.NoError(t, tester.env.GetWorkflowError())
+	})
 }