feat: clear destination #243

vishalm0509 · 2025-11-02T18:09:16Z

Description

This PR introduces the Clear Destination feature that allows users to manually or automatically clear data from a destination linked to a job when configuration or stream changes are detected.

Added two new API endpoints:

POST /api/v1/project/:projectid/jobs/:id/clear-destination – Triggers a clear destination workflow for the specified job.
GET /api/v1/project/:projectid/jobs/:id/clear-destination – Returns the current status of the clear destination workflow (running or not running).

Type of change

New feature (non-breaking change which adds functionality)
This change requires a documentation update

How Has This Been Tested?

Run clear destination manually
Update job and check if configured streams are getting dropped from the destination

… into refactor/bff

…er operations

…o refactor/worker

…into feat/clear_v0

…ke-ui into feat/clear_v0

…into feat/clear_v0

* fix: send id in job edit * fix: add modal for spec error, fix double error messages * fix: integration test fix * fix: minor fix

…into feat/clear_v0

vishalm0509 · 2025-11-05T10:40:19Z

Even after resuming a paused job, the options are blocked. @deepanshupal09-datazip

Pause a job
Resume it

vikaxsh · 2025-11-05T09:02:54Z

server/internal/services/etl/job.go

+	if clearRunning, _ := isWorkflowRunning(ctx, s.temporal, projectID, jobID, temporal.ClearDestination); clearRunning {
+		return fmt.Errorf("clear-destination is in progress, cannot update job")
+	}
+
+	// Cancel sync before updating the job
+	if syncRunning, _ := isWorkflowRunning(ctx, s.temporal, projectID, jobID, temporal.Sync); syncRunning {
+		logger.Infof("sync is running for job %d, cancelling sync workflow", jobID)
+		jobSlice := []*models.Job{existingJob}
+		if err := cancelAllJobWorkflows(ctx, s.temporal, jobSlice, projectID); err != nil {
+			return fmt.Errorf("failed to cancel sync: %s", err)
+		}
+		logger.Infof("successfully cancelled sync for job %d", jobID)
+	}


can we return type as well from isWorkflow runing, and cancel bases on that, would reduce db queries

I have removed that check, so extra db query won't happen

Before

// Cancel sync before updating the job if syncRunning, _ := isWorkflowRunning(ctx, s.temporal, projectID, jobID, temporal.Sync); syncRunning { logger.Infof("sync is running for job %d, cancelling sync workflow", jobID) jobSlice := []*models.Job{existingJob} if err := cancelAllJobWorkflows(ctx, s.temporal, jobSlice, projectID); err != nil { return fmt.Errorf("failed to cancel sync: %s", err) } logger.Infof("successfully cancelled sync for job %d", jobID) }

After

if err := cancelAllJobWorkflows(ctx, s.temporal, jobSlice, projectID); err != nil { return fmt.Errorf("failed to cancel sync: %s", err) }

vikaxsh · 2025-11-05T09:04:43Z

server/internal/services/etl/job.go

+
+		if len(diffCatalog) > 0 {
+			logger.Infof("stream difference detected for job %d, running clear destination workflow", existingJob.ID)
+			if _, err := s.ClearDestination(ctx, projectID, jobID, req.DifferenceStreams, 10*time.Second); err != nil {


are you sure clear destination would only run after any of the workflow is not running

Fixing this...

server/internal/services/etl/job.go

vikaxsh · 2025-11-05T09:21:07Z

server/internal/services/etl/job.go

+	// Check if sync is running and wait for it to stop
+	if running, _ := isWorkflowRunning(ctx, s.temporal, projectID, jobID, temporal.Sync); running {
+		if err := waitForSyncToStop(ctx, s.temporal, projectID, jobID, syncWaitTime); err != nil {
+			return nil, fmt.Errorf("sync is in progress, please cancel it before running clear-destination")
+		}
+	}


since it is part of same workflow at a time any one of them would run , why to keep this check

As we need to wait for sync to cancel after cancelling it in updateJob flow

vikaxsh · 2025-11-05T09:23:12Z

server/internal/services/etl/job.go

+		return nil, fmt.Errorf("job not found: %s", err)
+	}
+
+	diffCatalog, err := s.temporal.GetDifferenceStreams(ctx, job, job.StreamsConfig, req.UpdatedStreamsConfig)


Suggested change

diffCatalog, err := s.temporal.GetDifferenceStreams(ctx, job, job.StreamsConfig, req.UpdatedStreamsConfig)

diffCatalog, err := s.temporal.GetStreamDifference(ctx, job, job.StreamsConfig, req.UpdatedStreamsConfig)

vikaxsh · 2025-11-05T09:51:32Z

server/internal/services/etl/utils.go

 	for _, job := range jobs {
 		conditions = append(conditions, fmt.Sprintf(
-			"(WorkflowId BETWEEN 'sync-%s-%d' AND 'sync-%s-%d-~')",
+			"(WorkflowId BETWEEN 'sync-%s-%d' AND 'sync-%s-%d-~' AND OperationType = 'sync')",


would it be backward compatible ,i f not or you can list OperationType != 'clear'

OperationType != 'clear' and ``OperationType = 'sync'` Both will work

Anyways if a workflow is running it will definitely have operationType since we are registering it at the start of the workflow.

doc: https://docs.temporal.io/develop/go/observability#remove-search-attribute

vikaxsh · 2025-11-05T09:52:34Z

server/internal/services/etl/utils.go

+}
+
+// waitForSyncToStop waits for sync workflows to stop with timeout
+func waitForSyncToStop(ctx context.Context, tempClient *temporal.Temporal, projectID string, jobID int, maxWaitTime time.Duration) error {


we should not fire db queries in loop

I checked we can't use the DescribeWorkflowExecution so I have increased timeout to 1s and also removed time.After check and added a context timeout instead

Checking...

vikaxsh · 2025-11-05T10:07:04Z

server/internal/services/etl/utils.go

+		projectID, jobID, projectID, jobID, opType,
+	)
+
+	resp, err := tempClient.ListWorkflow(ctx, &workflowservice.ListWorkflowExecutionsRequest{


see if can use tempClient.Client.DescribeWorkflowExecution()

tempClient.Client.DescribeWorkflowExecution()

for this we need to provide workflowID, since the workflowID will change

Checking...

vikaxsh · 2025-11-05T10:23:52Z

server/internal/services/temporal/client.go

+// UpdateScheduleSpec updates an existing schedule's spec
+func (t *Temporal) UpdateScheduleSpec(ctx context.Context, frequency, projectID string, jobID int) error {
 	cronExpression := utils.ToCron(frequency)
 	_, scheduleID := t.WorkflowAndScheduleID(projectID, jobID)

 	handle := t.Client.ScheduleClient().GetHandle(ctx, scheduleID)
 	return handle.Update(ctx, client.ScheduleUpdateOptions{
 		DoUpdate: func(input client.ScheduleUpdateInput) (*client.ScheduleUpdate, error) {
 			input.Description.Schedule.Spec = &client.ScheduleSpec{
 				CronExpressions: []string{cronExpression},
 			}
 			return &client.ScheduleUpdate{
 				Schedule: &input.Description.Schedule,
 			}, nil
 		},
 	})
 }

+// UpdateScheduleAction updates the action (workflow args) of an existing schedule
+func (t *Temporal) UpdateScheduleAction(ctx context.Context, projectID string, jobID int, req interface{}) error {


can we use only one function for schedule update?

vikaxsh · 2025-11-05T10:33:38Z

server/internal/services/temporal/utils.go

+	case ClearDestination:
+		return time.Hour * 24 * 30


clear destination wont take that much time

Up for discussion

let us discuss and close this

deepanshupal09-datazip · 2025-11-05T12:47:47Z

Even after resuming a paused job, the options are blocked. @deepanshupal09-datazip

Pause a job

Resume it

not able to reproduce this

… into feat/clear_v0

vikaxsh · 2025-11-06T10:14:18Z

server/internal/handlers/job.go

+	})
+}
+
+// @router /project/:projectid/jobs/:id/stream-difference [get]


Suggested change

// @router /project/:projectid/jobs/:id/stream-difference [get]

// @router /project/:projectid/jobs/:id/clear-destination [get]

vikaxsh · 2025-11-06T10:47:02Z

server/internal/services/etl/job.go

+
+		if len(diffCatalog) > 0 {
+			logger.Infof("stream difference detected for job %d, running clear destination workflow", existingJob.ID)
+			if _, err := s.ClearDestination(ctx, projectID, jobID, req.DifferenceStreams, 30*time.Second); err != nil {


make it constant and let us discuss about max wait time

vikaxsh · 2025-11-06T10:49:42Z

server/internal/services/etl/job.go

+
+		if len(diffCatalog) > 0 {
+			logger.Infof("stream difference detected for job %d, running clear destination workflow", existingJob.ID)
+			if _, err := s.ClearDestination(ctx, projectID, jobID, req.DifferenceStreams, 30*time.Second); err != nil {


what if cleat destination failed, how would we recover and clear destination for only these streams req.DifferenceStreams??

Added TODO

// TODO: handle clear-destination workflow failure

vikaxsh · 2025-11-06T10:53:55Z

server/internal/services/etl/job.go

+
+	// Check if sync is running and wait for it to stop
+	if running, _, _ := isWorkflowRunning(ctx, s.temporal, projectID, jobID, temporal.Sync); running {
+		if err := waitForSyncToStop(ctx, s.temporal, projectID, jobID, syncWaitTime); err != nil {


what if sync is running , ans user click clear destination?, why to wait in that case

We won't wait because in that case we are passing syncWaitTIme=0

vikaxsh · 2025-11-06T11:21:19Z

server/internal/services/etl/utils.go

+	)
+
+	resp, err := tempClient.ListWorkflow(ctx, &workflowservice.ListWorkflowExecutionsRequest{
+		Query: query,


pass pageSize:1, Avoids loading unnecessary workflow metadata

* chore: remove temporalAddr constant and make retry simple * chore: remove logs

…into feat/clear_v0

hash-data · 2025-11-07T09:36:58Z

server/go.mod

 require github.com/beego/beego/v2 v2.3.8

 require (
+	github.com/apache/spark-connect-go/v35 v35.0.0-20250317154112-ffd832059443


why these are being added in this pr?

Used the one from staging.

Not needed

hash-data · 2025-11-07T09:38:27Z

server/internal/services/temporal/utils.go

+	case ClearDestination:
+		return time.Hour * 24 * 30


let us discuss and close this

hash-data · 2025-11-07T09:51:49Z

server/internal/services/temporal/utils.go

+
 // extractWorkflowResponse extracts and parses the JSON response from a workflow execution result
 func ExtractWorkflowResponse(ctx context.Context, run client.WorkflowRun) (map[string]interface{}, error) {
 	var result map[string]interface{}


Suggested change

var result map[string]interface{}

result := make(map[string]interface)

hash-data · 2025-11-07T10:20:18Z

server/internal/services/temporal/execute.go

+		"--difference", "/mnt/config/new_streams.json",
+	}
+	if encryptionKey, _ := web.AppConfig.String(constants.ConfEncryptionKey); encryptionKey != "" {
+		cmdArgs = append(cmdArgs, "--encryption-key", encryptionKey)


can we create constants for command args or add a todo.

hash-data · 2025-11-07T10:21:09Z

server/internal/services/temporal/execute.go

+	}
+
+	req := &ExecutionRequest{
+		Type:          "docker",


why type is required?

hash-data · 2025-11-07T11:40:57Z

server/internal/handlers/job.go

+		utils.ErrorResponse(&h.Controller, http.StatusBadRequest, fmt.Sprintf("failed to validate request: %s", err), err)
+		return
+	}
+	result, err := h.etl.ClearDestination(h.Ctx.Request.Context(), projectID, id, "", 0)


why sending empty stream config

In case of clearing the entire data form streams.json we are passing it. Else we can pass difference_streams in case of streams edit

hash-data · 2025-11-07T11:42:04Z

server/internal/handlers/job.go

+	if err != nil {
+		utils.ErrorResponse(&h.Controller, http.StatusInternalServerError, "Failed to get stream difference", err)
+		return
+	}


let us also print diff streams, so that we can have log till it persist

hash-data · 2025-11-07T11:44:32Z

server/internal/services/etl/job.go

+	}
+
+	// Handle stream difference if provided
+	// TODO: handle clear-destination workflow failure


let us add a recover flow for it

hash-data · 2025-11-07T11:49:49Z

server/internal/services/etl/job.go

 }

+func (s *ETLService) ClearDestination(ctx context.Context, projectID string, jobID int, streamsConfig string, syncWaitTime time.Duration) (map[string]interface{}, error) {
+	job, err := s.db.GetJobByID(jobID, true)


let us log the diff streams that is being passed so that we have an idea while debugging that clear destination run for these streams

hash-data · 2025-11-07T11:54:16Z

api-contract.md

+    "success": "boolean",
+    "message": "string",
+    "data": {
+      "message": "string"


does this message required?

Not needed. Removed

vikaxsh and others added 30 commits June 12, 2025 11:26

fix(api): refactor raw queries and add constants

0ca9bab

chore(api): Adding service layer

b6c00d6

chore(api): adding service layer

190b89f

Merge branch 'staging' of https://github.com/datazip-inc/olake-frontend…

922c2da

… into refactor/bff

fix: resolve conflict errors

9a00631

feat: improve Worker design by passing existing Client

b4bb565

Merge branch 'staging' of https://github.com/datazip-inc/olake-frontend…

679e4ce

… into refactor/bff

feat: improving error handling a

934cb1e

feat: add info logging

d152357

fix: changes after testing

a1142de

Merge branch 'staging' of https://github.com/datazip-inc/olake-frontend…

3bdd367

… into refactor/bff

feat(docker): use Docker Go SDK instead of shell commands for contain…

c4e93e8

…er operations

feat: updated GoSec security ci

489a2ea

feat: delete config files

5942237

chore: update project name to olake-ui

376a12b

chore: addressed review comments

8213098

chore: merge staging branch

61bfd98

chore: creatign dto packaghe

abc47c8

chore: using GetIDFromPath function for id

33640af

feat: implement dto layer and validation for request

6ab7a04

fix: commands -> check, discover

c65e735

fix: sync worklfow

8cc3be8

refactor: remove worker from olake-ui

f3b792a

chore: remove worker files & add timeout utility

83e844a

chore: update branch name in worklfow to release an image

6bc05d1

fix: remove worker from release.sh

cc10970

fix: add pre and post sync worker callback

7810c46

fix: add telemetry event for sync workflow

ae740f5

Merge branch 'staging' of https://github.com/datazip-inc/olake-ui int…

8ce4037

…o refactor/worker

fix: latest changes

b50c466

vishalm0509 and others added 10 commits November 3, 2025 22:41

fix: lint issue

7b62dd2

Merge branch 'refactor/bff' of https://github.com/datazip-inc/olake-ui …

6f0d821

…into feat/clear_v0

remove unpause and update req from bff

94c3f24

fix: send id in job edit

e4072e8

Merge branch 'fix/job-edit-api' of https://github.com/datazip-inc/ola…

766e029

…ke-ui into feat/clear_v0

fix: error in login and test destiantion

e5dfb84

Merge branch 'refactor/bff' of https://github.com/datazip-inc/olake-ui …

6dd7a9f

…into feat/clear_v0

fix: add retry logic for temporal client init

a849ec9

fix: send id in job edit (#245)

4091964

* fix: send id in job edit * fix: add modal for spec error, fix double error messages * fix: integration test fix * fix: minor fix

Merge branch 'refactor/bff' of https://github.com/datazip-inc/olake-ui …

3a530a8

…into feat/clear_v0

vishalm0509 mentioned this pull request Nov 5, 2025

feat: clear destination for iceberg/s3 #238

Open

4 tasks

vikaxsh reviewed Nov 5, 2025

View reviewed changes

deepanshupal09-datazip and others added 7 commits November 5, 2025 19:19

fix: fix destination name edit

51a1f2c

fix: resolve comments

d6ffbdb

Merge branch 'feat/clear_v0' of https://github.com/datazip-inc/olake-ui…

b691603

… into feat/clear_v0

fix: add get clear destination status endpoint

4ebf8cb

fix: add stream_difference and update timeout to 30s

71bebf2

fix: proper sync wait error handling

296bf46

resolve comments - add proper sync wait method without polling

a4ddb28

vikaxsh reviewed Nov 6, 2025

View reviewed changes

vishalm0509 and others added 6 commits November 6, 2025 18:01

fix: resolve comments

5e5d147

chore: update worker tag in docker-compose for int. test

a825e8c

chore: remove temporal address from constant (#246)

029ebfb

* chore: remove temporalAddr constant and make retry simple * chore: remove logs

chore: resolve review comments

f54819f

chore: remove unused messages

2c09805

Merge branch 'refactor/bff' of https://github.com/datazip-inc/olake-ui …

8d4e7af

…into feat/clear_v0

hash-data reviewed Nov 7, 2025

View reviewed changes

Base automatically changed from refactor/bff to staging November 7, 2025 12:39

	diffCatalog, err := s.temporal.GetDifferenceStreams(ctx, job, job.StreamsConfig, req.UpdatedStreamsConfig)
	diffCatalog, err := s.temporal.GetStreamDifference(ctx, job, job.StreamsConfig, req.UpdatedStreamsConfig)

	// @router /project/:projectid/jobs/:id/stream-difference [get]
	// @router /project/:projectid/jobs/:id/clear-destination [get]

	var result map[string]interface{}
	result := make(map[string]interface)

feat: clear destination #243

Are you sure you want to change the base?

feat: clear destination #243

Conversation

vishalm0509 commented Nov 2, 2025

Description

Type of change

How Has This Been Tested?

Uh oh!

vishalm0509 commented Nov 5, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

vishalm0509 Nov 5, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

deepanshupal09-datazip commented Nov 5, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

vishalm0509 Nov 5, 2025 •

edited

Loading