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

…o refactor/worker

…into refactor/worker

…e-ui into feat/clear_destination

…tion

…rogress

* feat: show error logs in test connection in source and destination * fix: fix test connection fail modal * fix: resolve comments * chore: add comments (#224) * chore: add comments * chore: 🚨 lint fix * feat: add error logs in check connection for source and destination (#195) * feat: add error logs i check conenction for source and destination * fix: parse json message as string * chore: update response key * chore: refactor * chore: remove logEntry struct * fix: log parsing * fix: log parsing * fix: logs parsing logic * docs: updated api contract * test: ✅ base setup for automation testing using play… (#203) * test: ✅ base setup for automation testing using playwright , add test cases fro major flows * fix: integration workflow changes * fix: ci changes * fix: ci changes * fix: ci changes * fix: ci changes * fix: ci changes * fix: ci changes * fix: ci changes * fix: ci changes * fix: ci changes * fix: ci changes * fix: ci changes * fix: ci changes * fix: ci changes * fix: ci changes * fix: ci changes * fix: ci changes * test: add postgres flow * fix: lint changes * test: add iceberg specific test * fix: tests * fix: stream name * fix: test fix * fix: test fix * fix: test fix * fix: test fix * fix: changes * fix: ci changes * fix: ci changes * fix: ci changes * test: fail test on test connection failure * fix: ci changes * fix: changes * fix: changes * fix: changes * fix: changes * fix: changes * fix: changes * fix: test fix * fix: changes * fix: changes * fix: test-data fix * fix: test-data fix * fix: changes * fix: changes * fix: changes * fix: host and port changes * fix: changes * fix: change * test: login test updated * test: refactor create source and destination page * test: refactor tests * fix: test fix * fix: changes * fix: fix changes * fix: changes * fix: changes * fix: review changes * refactor: refactor tests to use authenticated state of playwright * fix: changes * docs: update tests readme * test: intercept and log destination spec response * docs: update readme * fix: update timeouts for playwright tests * test: update readme * fix: add retries in playwright config * test: fix timeout duration * fix: update readme * test: add connector data test-id * fix: resolve comments * refactor(test): make and use enums instead of strings * fix: fix test data builder * fix: test data job name * fix: add todo --------- Co-authored-by: Duke Dhal <[email protected]> Co-authored-by: deepanshupal09-datazip <[email protected]> * feat: show refetch success message in job history page (#227) * Added success message on Job History refetch * Update ui/src/modules/jobs/pages/JobHistory.tsx Co-authored-by: deepanshupal09-datazip <[email protected]> --------- Co-authored-by: vikash choudhary <[email protected]> Co-authored-by: deepanshupal09-datazip <[email protected]> * fix: replace deprecated Phosphor icons with latest *Icon alternatives (#216) * fix: replace deprecated Phosphor icons with latest *Icon alternatives * fix: update Phosphor icons to latest alternatives in DestinationEdit and SourceEdit components and fix Lint issue * refactor: remove EditSourceModal * fix: update icon import --------- Co-authored-by: vikash choudhary <[email protected]> * feat: add clear destination * fix: merge conflicts * fix: build error fix * feat: append only mode (#232) * feat: add append only mode in ui * fix: data filter bug * fix: fix color of append only mode switch * fix: add custom option in all streams ingestion mode change * fix: use clsx for conditional styling * fix: fix icons * chore: add utils --------- Co-authored-by: Taraka Swathi <[email protected]> Co-authored-by: Ankit Sharma <[email protected]> Co-authored-by: vikash choudhary <[email protected]> Co-authored-by: Duke Dhal <[email protected]> Co-authored-by: SARTHAK RANA <[email protected]> Co-authored-by: vishalm0509 <[email protected]>

…nc/olake-ui into feat/clear_v0

…into feat/clear_v0

…ke-ui into feat/clear_v0

…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

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

vishalm0509 added 30 commits September 24, 2025 16:42

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

chore: remove stale code

a480a71

fix: update env var - container_persistence_dir and host_persistence_dir

1b7e87a

chore: remove worker job api

0ca69b6

chore: cleanup files

e5168a2

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

de465d4

…into refactor/worker

fix: fix refactor bff changes

805458f

chore: move encryptionkey to app.conf

15b3d8c

chore: add unmarshal-and-validate util

89664a2

chore: add robust error logging (resolve comments)

1e4dbec

feat: add clear-dest bff

36f4691

chore: remove container_persistent_path env var

3a2e1c9

Merge branch 'refactor/worker' of https://github.com/datazip-inc/olak…

22c6626

…e-ui into feat/clear_destination

fix: add hash to clear-destination workflowID

42961a7

chore: add api-contract and minor refactoring

062fb45

chore: remove hardcoded version used for testing

c2676bd

fix: add proper error handling

33962d6

chore: update api contract for job task and logs

b30d579

fix: add response type for clear-destination status

7ecd1bb

fix: add cancel sync in job update & remove cancel from clear destina…

6c620fa

…tion

fix: add stream difference endpoint

242ef5d

vishalm0509 and others added 17 commits October 27, 2025 16:28

chore: remove clear-destination status api endpoint

b380856

fix: resolve comments and add proper error for clear-destination in p…

de00a75

…rogress

chore: add proper error for SyncJob

40ddd64

fix: block sync/clear if job is paused

892282e

feat: add clear destination

c291eb6

Merge branch 'feat/clear_destination' of https://github.com/datazip-i…

a08cb8a

…nc/olake-ui into feat/clear_v0

fix: resolve merge conflicts

9ca6dd5

Merge branch 'refactor/bff' into feat/clear_v0

34a0268

fix: add syncWaitTime to clearDestination service

f2f3546

fix: fix clear destination api changes

f41f45a

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

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

6dd7a9f

…into feat/clear_v0

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

fix: resolve comments

5e5d147

	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]

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!

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!

Choose a reason for hiding this comment

Uh oh!

vishalm0509 Nov 5, 2025 •

edited

Loading