21
21
that has merged into it. As the first step of a merge, pull, or
22
22
branch operation we copy history from the source into the destination
25
The copying is done in a slightly complicated order. We don't want to
26
add a revision to the store until everything it refers to is also
27
stored, so that if a revision is present we can totally recreate it.
28
However, we can't know what files are included in a revision until we
29
read its inventory. So we query the inventory store of the source for
30
the ids we need, and then pull those ids and then return to the inventories.
36
import bzrlib.errors as errors
37
33
from bzrlib.errors import InstallFailed
38
34
from bzrlib.progress import ProgressPhase
39
35
from bzrlib.revision import NULL_REVISION
40
36
from bzrlib.tsort import topo_sort
41
37
from bzrlib.trace import mutter
43
from bzrlib.versionedfile import filter_absent, FulltextContentFactory
45
# TODO: Avoid repeatedly opening weaves so many times.
47
# XXX: This doesn't handle ghost (not present in branch) revisions at
48
# all yet. I'm not sure they really should be supported.
50
# NOTE: This doesn't copy revisions which may be present but not
51
# merged into the last revision. I'm not sure we want to do that.
53
# - get a list of revisions that need to be pulled in
54
# - for each one, pull in that revision file
55
# and get the inventory, and store the inventory with right
57
# - and get the ancestry, and store that with right parents too
58
# - and keep a note of all file ids and version seen
59
# - then go through all files; for each one get the weave,
60
# and add in all file versions
39
from bzrlib.versionedfile import FulltextContentFactory
63
42
class RepoFetcher(object):
64
43
"""Pull revisions and texts from one repository to another.
67
if set, try to limit to the data this revision references.
70
count_copied -- number of revisions copied
72
45
This should not be used directly, it's essential a object to encapsulate
73
46
the logic in InterRepository.fetch().
76
def __init__(self, to_repository, from_repository, last_revision=None, pb=None,
49
def __init__(self, to_repository, from_repository, last_revision=None,
50
pb=None, find_ghosts=True, fetch_spec=None):
78
51
"""Create a repo fetcher.
53
:param last_revision: If set, try to limit to the data this revision
80
55
:param find_ghosts: If True search the entire history for ghosts.
81
56
:param _write_group_acquired_callable: Don't use; this parameter only
82
57
exists to facilitate a hack done in InterPackRepo.fetch. We would
83
58
like to remove this parameter.
59
:param pb: ProgressBar object to use; deprecated and ignored.
60
This method will just create one on top of the stack.
86
self.failed_revisions = []
63
symbol_versioning.warn(
64
symbol_versioning.deprecated_in((1, 14, 0))
65
% "pb parameter to RepoFetcher.__init__")
66
# and for simplicity it is in fact ignored
88
67
if to_repository.has_same_location(from_repository):
89
68
# repository.fetch should be taking care of this case.
90
69
raise errors.BzrError('RepoFetcher run '
95
74
self.sink = to_repository._get_sink()
96
75
# must not mutate self._last_revision as its potentially a shared instance
97
76
self._last_revision = last_revision
77
self._fetch_spec = fetch_spec
98
78
self.find_ghosts = find_ghosts
100
self.pb = bzrlib.ui.ui_factory.nested_progress_bar()
101
self.nested_pb = self.pb
104
self.nested_pb = None
105
79
self.from_repository.lock_read()
80
mutter("Using fetch logic to copy between %s(%s) and %s(%s)",
81
self.from_repository, self.from_repository._format,
82
self.to_repository, self.to_repository._format)
110
if self.nested_pb is not None:
111
self.nested_pb.finished()
113
86
self.from_repository.unlock()
126
99
# assert not missing
127
100
self.count_total = 0
128
101
self.file_ids_names = {}
129
pp = ProgressPhase('Transferring', 4, self.pb)
102
pb = bzrlib.ui.ui_factory.nested_progress_bar()
103
pb.show_pct = pb.show_count = False
105
pb.update("Finding revisions", 0, 2)
132
106
search = self._revids_to_fetch()
133
107
if search is None:
135
self._fetch_everything_for_search(search, pp)
109
pb.update("Fetching revisions", 1, 2)
110
self._fetch_everything_for_search(search)
139
def _fetch_everything_for_search(self, search, pp):
114
def _fetch_everything_for_search(self, search):
140
115
"""Fetch all data for the given set of revisions."""
141
116
# The first phase is "file". We pass the progress bar for it directly
142
117
# into item_keys_introduced_by, which has more information about how
146
121
# item_keys_introduced_by should have a richer API than it does at the
147
122
# moment, so that it can feed the progress information back to this
149
self.pb = bzrlib.ui.ui_factory.nested_progress_bar()
124
if (self.from_repository._format.rich_root_data and
125
not self.to_repository._format.rich_root_data):
126
raise errors.IncompatibleRepositories(
127
self.from_repository, self.to_repository,
128
"different rich-root support")
129
pb = bzrlib.ui.ui_factory.nested_progress_bar()
131
pb.update("Get stream source")
132
source = self.from_repository._get_source(
133
self.to_repository._format)
134
stream = source.get_stream(search)
151
135
from_format = self.from_repository._format
152
stream = self.get_stream(search, pp)
136
pb.update("Inserting stream")
153
137
resume_tokens, missing_keys = self.sink.insert_stream(
154
138
stream, from_format, [])
156
stream = self.get_stream_for_missing_keys(missing_keys)
140
pb.update("Missing keys")
141
stream = source.get_stream_for_missing_keys(missing_keys)
142
pb.update("Inserting missing keys")
157
143
resume_tokens, missing_keys = self.sink.insert_stream(
158
144
stream, from_format, resume_tokens)
164
150
raise AssertionError(
165
151
"second push failed to commit the fetch %r." % (
153
pb.update("Finishing stream")
167
154
self.sink.finished()
169
if self.pb is not None:
172
def get_stream(self, search, pp):
174
revs = search.get_keys()
175
graph = self.from_repository.get_graph()
176
revs = list(graph.iter_topo_order(revs))
177
data_to_fetch = self.from_repository.item_keys_introduced_by(
180
for knit_kind, file_id, revisions in data_to_fetch:
181
if knit_kind != phase:
183
# Make a new progress bar for this phase
186
self.pb = bzrlib.ui.ui_factory.nested_progress_bar()
187
if knit_kind == "file":
188
# Accumulate file texts
189
text_keys.extend([(file_id, revision) for revision in
191
elif knit_kind == "inventory":
192
# Now copy the file texts.
193
from_texts = self.from_repository.texts
194
yield ('texts', from_texts.get_record_stream(
195
text_keys, self.to_repository._fetch_order,
196
not self.to_repository._fetch_uses_deltas))
197
# Cause an error if a text occurs after we have done the
200
# Before we process the inventory we generate the root
201
# texts (if necessary) so that the inventories references
203
for _ in self._generate_root_texts(revs):
205
# NB: This currently reopens the inventory weave in source;
206
# using a single stream interface instead would avoid this.
207
self.pb.update("fetch inventory", 0, 1)
208
from_weave = self.from_repository.inventories
209
# we fetch only the referenced inventories because we do not
210
# know for unselected inventories whether all their required
211
# texts are present in the other repository - it could be
213
yield ('inventories', from_weave.get_record_stream(
214
[(rev_id,) for rev_id in revs],
215
self.inventory_fetch_order(),
216
not self.delta_on_metadata()))
217
elif knit_kind == "signatures":
218
# Nothing to do here; this will be taken care of when
219
# _fetch_revision_texts happens.
221
elif knit_kind == "revisions":
222
for _ in self._fetch_revision_texts(revs, self.pb):
225
raise AssertionError("Unknown knit kind %r" % knit_kind)
226
self.count_copied += len(revs)
228
def get_stream_for_missing_keys(self, missing_keys):
229
# missing keys can only occur when we are byte copying and not
230
# translating (because translation means we don't send
231
# unreconstructable deltas ever).
233
keys['texts'] = set()
234
keys['revisions'] = set()
235
keys['inventories'] = set()
236
keys['signatures'] = set()
237
for key in missing_keys:
238
keys[key[0]].add(key[1:])
239
if len(keys['revisions']):
240
# If we allowed copying revisions at this point, we could end up
241
# copying a revision without copying its required texts: a
242
# violation of the requirements for repository integrity.
243
raise AssertionError(
244
'cannot copy revisions to fill in missing deltas %s' % (
246
for substream_kind, keys in keys.iteritems():
247
vf = getattr(self.from_repository, substream_kind)
248
# Ask for full texts always so that we don't need more round trips
250
stream = vf.get_record_stream(keys,
251
self.to_repository._fetch_order, True)
252
yield substream_kind, stream
254
158
def _revids_to_fetch(self):
255
159
"""Determines the exact revisions needed from self.from_repository to
271
177
except errors.NoSuchRevision, e:
272
178
raise InstallFailed([self._last_revision])
274
def _fetch_revision_texts(self, revs, pb):
275
# fetch signatures first and then the revision texts
276
# may need to be a InterRevisionStore call here.
277
from_sf = self.from_repository.signatures
278
# A missing signature is just skipped.
279
keys = [(rev_id,) for rev_id in revs]
280
signatures = filter_absent(from_sf.get_record_stream(
282
self.to_repository._fetch_order,
283
not self.to_repository._fetch_uses_deltas))
284
# If a revision has a delta, this is actually expanded inside the
285
# insert_record_stream code now, which is an alternate fix for
287
from_rf = self.from_repository.revisions
288
revisions = from_rf.get_record_stream(
290
self.to_repository._fetch_order,
291
not self.delta_on_metadata())
292
return [('signatures', signatures), ('revisions', revisions)]
294
def _generate_root_texts(self, revs):
295
"""This will be called by __fetch between fetching weave texts and
296
fetching the inventory weave.
298
Subclasses should override this if they need to generate root texts
299
after fetching weave texts.
303
def inventory_fetch_order(self):
304
return self.to_repository._fetch_order
306
def delta_on_metadata(self):
307
src_serializer = self.from_repository._format._serializer
308
target_serializer = self.to_repository._format._serializer
309
return (self.to_repository._fetch_uses_deltas and
310
src_serializer == target_serializer)
313
181
class Inter1and2Helper(object):
314
182
"""Helper for operations that convert data from model 1 and 2
397
265
rev_id_to_root_id.get(parent, root_id) == root_id)
398
266
yield FulltextContentFactory(key, parent_keys, None, '')
399
267
return [('texts', yield_roots())]
402
class Model1toKnit2Fetcher(RepoFetcher):
403
"""Fetch from a Model1 repository into a Knit2 repository
405
def __init__(self, to_repository, from_repository, last_revision=None,
406
pb=None, find_ghosts=True):
407
self.helper = Inter1and2Helper(from_repository)
408
RepoFetcher.__init__(self, to_repository, from_repository,
409
last_revision, pb, find_ghosts)
411
def _generate_root_texts(self, revs):
412
return self.helper.generate_root_texts(revs)
414
def inventory_fetch_order(self):
417
Knit1to2Fetcher = Model1toKnit2Fetcher
added
removed
bzrlib/fetch.py
