Method cloneTablesWithSlot

flow/workflows/snapshot_flow.go:294–330

(
	ctx workflow.Context,
	sessionCtx workflow.Context,
	numTablesInParallel int,
)

}

func (s *SnapshotFlowExecution) cloneTablesWithSlot(
	ctx workflow.Context,
	sessionCtx workflow.Context,
	numTablesInParallel int,
) error {
	slotInfo, err := s.setupReplication(sessionCtx)
	if err != nil {
		return fmt.Errorf("failed to setup replication: %w", err)
	}
	defer func() {
		dCtx, cancel := workflow.NewDisconnectedContext(sessionCtx)
		defer cancel()
		if err := s.closeSlotKeepAlive(dCtx); err != nil {
			s.logger.Error("failed to close slot keep alive", slog.Any("error", err))
		}
	}()
	var slotName string
	var snapshotName string
	if slotInfo != nil {
		slotName = slotInfo.SlotName
		snapshotName = slotInfo.SnapshotName
	}

	s.logger.Info("cloning tables in parallel", slog.Int("parallelism", numTablesInParallel))
	if err := s.cloneTables(ctx,
		SNAPSHOT_TYPE_SLOT,
		slotName,
		snapshotName,
		"",
		numTablesInParallel,
	); err != nil {
		s.logger.Error("failed to clone tables", slog.Any("error", err))
		return fmt.Errorf("failed to clone tables: %w", err)
	}

	return nil
}

func SnapshotFlowWorkflow(
	ctx workflow.Context,