Function processTableAdditions

flow/workflows/cdc_flow.go:233–349

(
	ctx workflow.Context,
	logger log.Logger,
	cfg *protos.FlowConnectionConfigsCore,
	state *cdc_state.CDCFlowWorkflowState,
	mirrorNameSearch temporal.SearchAttributes,
)

}

func processTableAdditions(
	ctx workflow.Context,
	logger log.Logger,
	cfg *protos.FlowConnectionConfigsCore,
	state *cdc_state.CDCFlowWorkflowState,
	mirrorNameSearch temporal.SearchAttributes,
) error {
	flowConfigUpdate := state.FlowConfigUpdate
	if len(flowConfigUpdate.AdditionalTables) == 0 {
		syncStateToConfigProtoInCatalog(ctx, cfg, state)
		return nil
	}
	checkDestinationOverlap := !getClickHouseInitialLoadAllowNonEmptyTables(ctx, logger, cfg.Env)
	if internal.AdditionalTablesHasOverlap(
		state.SyncFlowOptions.TableMappings, flowConfigUpdate.AdditionalTables, checkDestinationOverlap,
	) {
		logger.Warn("duplicate source/destination tables found in additionalTables")
		syncStateToConfigProtoInCatalog(ctx, cfg, state)
		return nil
	}
	state.UpdateStatus(ctx, logger, protos.FlowStatus_STATUS_SNAPSHOT)

	addTablesSelector := workflow.NewNamedSelector(ctx, "AddTables")
	addTablesSelector.AddReceive(ctx.Done(), func(_ workflow.ReceiveChannel, _ bool) {})
	flowSignalStateChangeChan := model.FlowSignalStateChange.GetSignalChannel(ctx)
	flowSignalStateChangeChan.AddToSelector(addTablesSelector, handleFlowSignalStateChange(ctx, cfg, state, logger, "AddTables"))

	logger.Info("altering publication for additional tables")
	alterPublicationAddAdditionalTablesCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
		StartToCloseTimeout: 1 * time.Hour,
		HeartbeatTimeout:    5 * time.Minute,
	})
	alterPublicationAddAdditionalTablesFuture := workflow.ExecuteActivity(
		alterPublicationAddAdditionalTablesCtx,
		flowable.AddTablesToPublication,
		cfg, flowConfigUpdate.AdditionalTables)

	var res *CDCFlowWorkflowResult
	var addTablesFlowErr error
	addTablesSelector.AddFuture(alterPublicationAddAdditionalTablesFuture, func(f workflow.Future) {
		addTablesFlowErr = f.Get(alterPublicationAddAdditionalTablesCtx, f)
		if addTablesFlowErr == nil {
			logger.Info("additional tables added to publication")
			additionalTablesUUID := GetUUID(ctx)
			childAdditionalTablesCDCFlowID := GetChildWorkflowID(
				additionalTablesCDCFlowPrefix,
				cfg.FlowJobName,
				additionalTablesUUID,
			)
			additionalTablesCfg := proto.CloneOf(cfg)
			additionalTablesCfg.DoInitialSnapshot = !flowConfigUpdate.SkipInitialSnapshotForTableAdditions
			additionalTablesCfg.InitialSnapshotOnly = true
			additionalTablesCfg.TableMappings = flowConfigUpdate.AdditionalTables
			additionalTablesCfg.Resync = false
			if state.SnapshotNumRowsPerPartition > 0 {
				additionalTablesCfg.SnapshotNumRowsPerPartition = state.SnapshotNumRowsPerPartition
			}
			if state.SnapshotNumPartitionsOverride > 0 {