Skip to content

feat(resync): Add support to update snapshot config (max_parallel_workers, num_tables, rows_per_partition) during resync signal #4276

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
masterashu merged 13 commits into from
May 29, 2026
Merged

feat(resync): Add support to update snapshot config (max_parallel_workers, num_tables, rows_per_partition) during resync signal #4276

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
c614269
feat(resync): Add support to update snapshot config (max_parallel_wor…
masterashu May 5, 2026
57c205f
Fix snapshot override parameter not picked up during existing resync
masterashu May 11, 2026
44e8ffb
resolved comments
masterashu May 14, 2026
6924494
resolve comments
masterashu May 18, 2026
114a505
remove unused function
masterashu May 20, 2026
b4253db
add e2e test
masterashu May 22, 2026
9050b10
fix(test): Updated test for resync with snapshot parameters
masterashu May 27, 2026
8439ee7
Handle snapshot completed pipes
masterashu May 27, 2026
33d8eca
Fix resync during snapshot not overriding params
masterashu May 27, 2026
1dd91bf
add tests
masterashu May 27, 2026
f586af9
Fix unsupported multi row insert query for mysql
masterashu May 27, 2026
3790e5c
fix lint error
masterashu May 28, 2026
0370fd8
resolevd comments
masterashu May 29, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 17 additions & 0 deletions flow/cmd/handler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -503,6 +503,9 @@ func (h *FlowRequestHandler) FlowStateChange(

config.Resync = true
config.DoInitialSnapshot = true
// Override Snapshot Parameters
overrideSnapshotParametersInResync(req, config)

@jgao54 jgao54 May 14, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: i may regret saying this (lol) but for now we are not validating these snapshot parameters so we could just remove it altogether.

maybe can add a comment here mentioning that. this avoids having update these config fields.


// validate mirror first because once the mirror is dropped, there's no going back
if _, err := h.ValidateCDCMirror(ctx, &protos.CreateCDCFlowRequest{
ConnectionConfigs: config,
Expand Down Expand Up @@ -549,6 +552,20 @@ func (h *FlowRequestHandler) FlowStateChange(
return &protos.FlowStateChangeResponse{}, nil
}

func overrideSnapshotParametersInResync(req *protos.FlowStateChangeRequest, configs *protos.FlowConnectionConfigs) {
Comment thread
masterashu marked this conversation as resolved.
Outdated
if u := req.GetFlowConfigUpdate().GetCdcFlowConfigUpdate(); u != nil {
if u.SnapshotMaxParallelWorkers > 0 {
configs.SnapshotMaxParallelWorkers = u.SnapshotMaxParallelWorkers
}
if u.SnapshotNumTablesInParallel > 0 {
configs.SnapshotNumTablesInParallel = u.SnapshotNumTablesInParallel
}
if u.SnapshotNumRowsPerPartition > 0 {
configs.SnapshotNumRowsPerPartition = u.SnapshotNumRowsPerPartition
}
}
}

func (h *FlowRequestHandler) handleCancelWorkflow(ctx context.Context, workflowID, runID string) error {
ctxWithTimeout, cancel := context.WithTimeout(ctx, 2*time.Minute)
defer cancel()
Expand Down
19 changes: 19 additions & 0 deletions flow/shared/telemetry/activity_logging.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,6 +93,25 @@ func LogActivityStartFlowConfigUpdate(ctx context.Context, flowName string, upda
changes = append(changes, fmt.Sprintf("tables removed: %v", removedTables))
}

if update.BatchSize > 0 {
changes = append(changes, fmt.Sprintf("batchSize: %v", update.BatchSize))
}
if update.IdleTimeout > 0 {
changes = append(changes, fmt.Sprintf("idleTimeout: %v", update.IdleTimeout))
}
if update.SnapshotNumPartitionsOverride > 0 {
changes = append(changes, fmt.Sprintf("snapshotNumPartitionsOverride: %v", update.SnapshotNumPartitionsOverride))
}
if update.SnapshotMaxParallelWorkers > 0 {

@jgao54 jgao54 May 14, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

thanks for adding this to activity logging as well!

@jgao54 jgao54 May 14, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: for same of completion, let's include batchSize, idleTimeout, snapshotNumPartitionsOverride to match LogActivityUpdateFlowConfig

@ilidemi ilidemi May 28, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

// Only logging fields that don't need old values to compare to

The caller here will actually pass all values, changed and unchanged, and the flow-api can't distinguish which ones to log. Logging the list of tables has to happen here because they're not applied to the config until snapshotting is done, so keeping just tables and leaving the rest to the worker is better.

@masterashu masterashu May 29, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Okay

changes = append(changes, fmt.Sprintf("snapshotMaxParallelWorkers: %v", update.SnapshotMaxParallelWorkers))
}
if update.SnapshotNumTablesInParallel > 0 {
changes = append(changes, fmt.Sprintf("snapshotNumTablesInParallel: %v", update.SnapshotNumTablesInParallel))
}
if update.SnapshotNumRowsPerPartition > 0 {
changes = append(changes, fmt.Sprintf("snapshotNumRowsPerPartition: %v", update.SnapshotNumRowsPerPartition))
}

logActivity(ctx, ActionStartFlowConfigUpdate,
slog.String("flowName", flowName),
slog.String("activityDetails", strings.Join(changes, ", ")))
Expand Down
29 changes: 29 additions & 0 deletions flow/workflows/cdc_flow.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -192,6 +192,9 @@ func handleFlowSignalStateChange(
// we should ContinueAsNew after the first signal in the selector, but just in case
cfg.Resync = true
cfg.DoInitialSnapshot = true
// This is just for in-memory
// Override Snapshot Parameters if request in State Change request
overrideSnapshotParametersInState(val, state)
state.DropFlowInput = &protos.DropFlowInput{
// to be filled in just before ContinueAsNew
FlowJobName: cfg.FlowJobName,
Expand All @@ -206,6 +209,25 @@ func handleFlowSignalStateChange(
}
}

func overrideSnapshotParametersInState(req *protos.FlowStateChangeRequest, s *cdc_state.CDCFlowWorkflowState) {
u := req.GetFlowConfigUpdate().GetCdcFlowConfigUpdate()
if u == nil {
return
}
if u.SnapshotMaxParallelWorkers > 0 {
s.SnapshotMaxParallelWorkers = u.SnapshotMaxParallelWorkers
}
if u.SnapshotNumTablesInParallel > 0 {
s.SnapshotNumTablesInParallel = u.SnapshotNumTablesInParallel
}
if u.SnapshotNumRowsPerPartition > 0 {
s.SnapshotNumRowsPerPartition = u.SnapshotNumRowsPerPartition
}
if u.SnapshotNumPartitionsOverride > 0 {
s.SnapshotNumPartitionsOverride = u.SnapshotNumPartitionsOverride
}
}

func processTableAdditions(
ctx workflow.Context,
logger log.Logger,
Expand Down Expand Up @@ -494,6 +516,8 @@ func CDCFlowWorkflow(
state.ActiveSignal = model.ResyncSignal
cfg.Resync = true
cfg.DoInitialSnapshot = true
// Update State with snapshot parameters
overrideSnapshotParametersInState(val, state)
resyncCfg := syncStateToConfigProtoInCatalog(ctx, cfg, state)
state.DropFlowInput = &protos.DropFlowInput{
FlowJobName: resyncCfg.FlowJobName,
Expand Down Expand Up @@ -627,6 +651,10 @@ func CDCFlowWorkflow(
cfg.TableMappings = originalTableMappings
// this is the only place where we can have a resync during a resync
// so we need to NOT sync the tableMappings to catalog to preserve original names

// We still override the snapshot parameters (when resync with updated values)
overrideSnapshotParametersInState(val, state)
syncStateToConfigProtoInCatalog(ctx, cfg, state)
uploadConfigToCatalog(ctx, cfg)
Comment thread
jgao54 marked this conversation as resolved.
state.DropFlowInput = &protos.DropFlowInput{
FlowJobName: cfg.FlowJobName,
Expand Down Expand Up @@ -862,6 +890,7 @@ func CDCFlowWorkflow(
state.ActiveSignal = model.ResyncSignal
cfg.Resync = true
cfg.DoInitialSnapshot = true
overrideSnapshotParametersInState(val, state)
resyncCfg := syncStateToConfigProtoInCatalog(ctx, cfg, state)
state.DropFlowInput = &protos.DropFlowInput{
FlowJobName: resyncCfg.FlowJobName,
Expand Down
Loading