228
228
return self.index_name('text', name)
230
230
def _replace_index_with_readonly(self, index_type):
231
unlimited_cache = False
232
if index_type == 'chk':
233
unlimited_cache = True
231
234
setattr(self, index_type + '_index',
232
235
self.index_class(self.index_transport,
233
236
self.index_name(index_type, self.name),
234
self.index_sizes[self.index_offset(index_type)]))
237
self.index_sizes[self.index_offset(index_type)],
238
unlimited_cache=unlimited_cache))
237
241
class ExistingPack(Pack):
269
273
def __init__(self, name, revision_index, inventory_index, text_index,
270
274
signature_index, upload_transport, pack_transport, index_transport,
275
pack_collection, chk_index=None):
272
276
"""Create a ResumedPack object."""
273
277
ExistingPack.__init__(self, pack_transport, name, revision_index,
274
inventory_index, text_index, signature_index)
278
inventory_index, text_index, signature_index,
275
280
self.upload_transport = upload_transport
276
281
self.index_transport = index_transport
277
282
self.index_sizes = [None, None, None, None]
301
309
self.upload_transport.delete(self.file_name())
302
310
indices = [self.revision_index, self.inventory_index, self.text_index,
303
311
self.signature_index]
312
if self.chk_index is not None:
313
indices.append(self.chk_index)
304
314
for index in indices:
305
315
index._transport.delete(index._name)
307
317
def finish(self):
308
318
self._check_references()
309
new_name = '../packs/' + self.file_name()
310
self.upload_transport.rename(self.file_name(), new_name)
311
for index_type in ['revision', 'inventory', 'text', 'signature']:
319
index_types = ['revision', 'inventory', 'text', 'signature']
320
if self.chk_index is not None:
321
index_types.append('chk')
322
for index_type in index_types:
312
323
old_name = self.index_name(index_type, self.name)
313
324
new_name = '../indices/' + old_name
314
325
self.upload_transport.rename(old_name, new_name)
315
326
self._replace_index_with_readonly(index_type)
327
new_name = '../packs/' + self.file_name()
328
self.upload_transport.rename(self.file_name(), new_name)
316
329
self._state = 'finished'
318
331
def _get_external_refs(self, index):
332
"""Return compression parents for this index that are not present.
334
This returns any compression parents that are referenced by this index,
335
which are not contained *in* this index. They may be present elsewhere.
319
337
return index.external_references(1)
564
589
flush_func=flush_func)
565
590
self.add_callback = None
567
def replace_indices(self, index_to_pack, indices):
568
"""Replace the current mappings with fresh ones.
570
This should probably not be used eventually, rather incremental add and
571
removal of indices. It has been added during refactoring of existing
574
:param index_to_pack: A mapping from index objects to
575
(transport, name) tuples for the pack file data.
576
:param indices: A list of indices.
578
# refresh the revision pack map dict without replacing the instance.
579
self.index_to_pack.clear()
580
self.index_to_pack.update(index_to_pack)
581
# XXX: API break - clearly a 'replace' method would be good?
582
self.combined_index._indices[:] = indices
583
# the current add nodes callback for the current writable index if
585
self.add_callback = None
587
592
def add_index(self, index, pack):
588
593
"""Add index to the aggregate, which is an index for Pack pack.
622
627
self.data_access.set_writer(None, None, (None, None))
623
628
self.index_to_pack.clear()
624
629
del self.combined_index._indices[:]
630
del self.combined_index._index_names[:]
625
631
self.add_callback = None
627
def remove_index(self, index, pack):
633
def remove_index(self, index):
628
634
"""Remove index from the indices used to answer queries.
630
636
:param index: An index from the pack parameter.
631
:param pack: A Pack instance.
633
638
del self.index_to_pack[index]
634
self.combined_index._indices.remove(index)
639
pos = self.combined_index._indices.index(index)
640
del self.combined_index._indices[pos]
641
del self.combined_index._index_names[pos]
635
642
if (self.add_callback is not None and
636
643
getattr(index, 'add_nodes', None) == self.add_callback):
637
644
self.add_callback = None
1392
1400
self.inventory_index = AggregateIndex(self.reload_pack_names, flush)
1393
1401
self.text_index = AggregateIndex(self.reload_pack_names, flush)
1394
1402
self.signature_index = AggregateIndex(self.reload_pack_names, flush)
1403
all_indices = [self.revision_index, self.inventory_index,
1404
self.text_index, self.signature_index]
1395
1405
if use_chk_index:
1396
1406
self.chk_index = AggregateIndex(self.reload_pack_names, flush)
1407
all_indices.append(self.chk_index)
1398
1409
# used to determine if we're using a chk_index elsewhere.
1399
1410
self.chk_index = None
1411
# Tell all the CombinedGraphIndex objects about each other, so they can
1412
# share hints about which pack names to search first.
1413
all_combined = [agg_idx.combined_index for agg_idx in all_indices]
1414
for combined_idx in all_combined:
1415
combined_idx.set_sibling_indices(
1416
set(all_combined).difference([combined_idx]))
1400
1417
# resumed packs
1401
1418
self._resumed_packs = []
1421
return '%s(%r)' % (self.__class__.__name__, self.repo)
1403
1423
def add_pack_to_memory(self, pack):
1404
1424
"""Make a Pack object available to the repository to satisfy queries.
1486
1506
'containing %d revisions. Packing %d files into %d affecting %d'
1487
1507
' revisions', self, total_packs, total_revisions, num_old_packs,
1488
1508
num_new_packs, num_revs_affected)
1489
self._execute_pack_operations(pack_operations,
1509
result = self._execute_pack_operations(pack_operations,
1490
1510
reload_func=self._restart_autopack)
1491
1511
mutter('Auto-packing repository %s completed', self)
1494
1514
def _execute_pack_operations(self, pack_operations, _packer_class=Packer,
1495
1515
reload_func=None):
1519
1539
self._remove_pack_from_memory(pack)
1520
1540
# record the newly available packs and stop advertising the old
1522
self._save_pack_names(clear_obsolete_packs=True)
1523
# Move the old packs out of the way now they are no longer referenced.
1524
for revision_count, packs in pack_operations:
1525
self._obsolete_packs(packs)
1542
to_be_obsoleted = []
1543
for _, packs in pack_operations:
1544
to_be_obsoleted.extend(packs)
1545
result = self._save_pack_names(clear_obsolete_packs=True,
1546
obsolete_packs=to_be_obsoleted)
1527
1549
def _flush_new_pack(self):
1528
1550
if self._new_pack is not None:
1539
1561
def _already_packed(self):
1540
1562
"""Is the collection already packed?"""
1541
return len(self._names) < 2
1563
return not (self.repo._format.pack_compresses or (len(self._names) > 1))
1565
def pack(self, hint=None, clean_obsolete_packs=False):
1544
1566
"""Pack the pack collection totally."""
1545
1567
self.ensure_loaded()
1546
1568
total_packs = len(self._names)
1547
1569
if self._already_packed():
1548
# This is arguably wrong because we might not be optimal, but for
1549
# now lets leave it in. (e.g. reconcile -> one pack. But not
1552
1571
total_revisions = self.revision_index.combined_index.key_count()
1553
1572
# XXX: the following may want to be a class, to pack with a given
1555
1574
mutter('Packing repository %s, which has %d pack files, '
1556
'containing %d revisions into 1 packs.', self, total_packs,
1575
'containing %d revisions with hint %r.', self, total_packs,
1576
total_revisions, hint)
1558
1577
# determine which packs need changing
1559
pack_distribution = [1]
1560
1578
pack_operations = [[0, []]]
1561
1579
for pack in self.all_packs():
1562
pack_operations[-1][0] += pack.get_revision_count()
1563
pack_operations[-1][1].append(pack)
1580
if hint is None or pack.name in hint:
1581
# Either no hint was provided (so we are packing everything),
1582
# or this pack was included in the hint.
1583
pack_operations[-1][0] += pack.get_revision_count()
1584
pack_operations[-1][1].append(pack)
1564
1585
self._execute_pack_operations(pack_operations, OptimisingPacker)
1587
if clean_obsolete_packs:
1588
self._clear_obsolete_packs()
1566
1590
def plan_autopack_combinations(self, existing_packs, pack_distribution):
1567
1591
"""Plan a pack operation.
1680
1704
inv_index = self._make_index(name, '.iix', resume=True)
1681
1705
txt_index = self._make_index(name, '.tix', resume=True)
1682
1706
sig_index = self._make_index(name, '.six', resume=True)
1683
result = ResumedPack(name, rev_index, inv_index, txt_index,
1684
sig_index, self._upload_transport, self._pack_transport,
1685
self._index_transport, self)
1707
if self.chk_index is not None:
1708
chk_index = self._make_index(name, '.cix', resume=True,
1709
unlimited_cache=True)
1712
result = self.resumed_pack_factory(name, rev_index, inv_index,
1713
txt_index, sig_index, self._upload_transport,
1714
self._pack_transport, self._index_transport, self,
1715
chk_index=chk_index)
1686
1716
except errors.NoSuchFile, e:
1687
1717
raise errors.UnresumableWriteGroup(self.repo, [name], str(e))
1688
1718
self.add_pack_to_memory(result)
1755
1786
:param return: None.
1757
1788
for pack in packs:
1758
pack.pack_transport.rename(pack.file_name(),
1759
'../obsolete_packs/' + pack.file_name())
1790
pack.pack_transport.rename(pack.file_name(),
1791
'../obsolete_packs/' + pack.file_name())
1792
except (errors.PathError, errors.TransportError), e:
1793
# TODO: Should these be warnings or mutters?
1794
mutter("couldn't rename obsolete pack, skipping it:\n%s"
1760
1796
# TODO: Probably needs to know all possible indices for this pack
1761
1797
# - or maybe list the directory and move all indices matching this
1762
1798
# name whether we recognize it or not?
1764
1800
if self.chk_index is not None:
1765
1801
suffixes.append('.cix')
1766
1802
for suffix in suffixes:
1767
self._index_transport.rename(pack.name + suffix,
1768
'../obsolete_packs/' + pack.name + suffix)
1804
self._index_transport.rename(pack.name + suffix,
1805
'../obsolete_packs/' + pack.name + suffix)
1806
except (errors.PathError, errors.TransportError), e:
1807
mutter("couldn't rename obsolete index, skipping it:\n%s"
1770
1810
def pack_distribution(self, total_revisions):
1771
1811
"""Generate a list of the number of revisions to put in each pack.
1797
1837
self._remove_pack_indices(pack)
1798
1838
self.packs.remove(pack)
1800
def _remove_pack_indices(self, pack):
1801
"""Remove the indices for pack from the aggregated indices."""
1802
self.revision_index.remove_index(pack.revision_index, pack)
1803
self.inventory_index.remove_index(pack.inventory_index, pack)
1804
self.text_index.remove_index(pack.text_index, pack)
1805
self.signature_index.remove_index(pack.signature_index, pack)
1806
if self.chk_index is not None:
1807
self.chk_index.remove_index(pack.chk_index, pack)
1840
def _remove_pack_indices(self, pack, ignore_missing=False):
1841
"""Remove the indices for pack from the aggregated indices.
1843
:param ignore_missing: Suppress KeyErrors from calling remove_index.
1845
for index_type in Pack.index_definitions.keys():
1846
attr_name = index_type + '_index'
1847
aggregate_index = getattr(self, attr_name)
1848
if aggregate_index is not None:
1849
pack_index = getattr(pack, attr_name)
1851
aggregate_index.remove_index(pack_index)
1809
1857
def reset(self):
1810
1858
"""Clear all cached data."""
1811
1859
# cached revision data
1812
self.repo._revision_knit = None
1813
1860
self.revision_index.clear()
1814
1861
# cached signature data
1815
self.repo._signature_knit = None
1816
1862
self.signature_index.clear()
1817
1863
# cached file text data
1818
1864
self.text_index.clear()
1819
self.repo._text_knit = None
1820
1865
# cached inventory data
1821
1866
self.inventory_index.clear()
1822
1867
# cached chk data
1921
1967
:param clear_obsolete_packs: If True, clear out the contents of the
1922
1968
obsolete_packs directory.
1969
:param obsolete_packs: Packs that are obsolete once the new pack-names
1970
file has been written.
1971
:return: A list of the names saved that were not previously on disk.
1973
already_obsolete = []
1924
1974
self.lock_names()
1926
1976
builder = self._index_builder_class()
1927
disk_nodes, deleted_nodes, new_nodes = self._diff_pack_names()
1977
(disk_nodes, deleted_nodes, new_nodes,
1978
orig_disk_nodes) = self._diff_pack_names()
1928
1979
# TODO: handle same-name, index-size-changes here -
1929
1980
# e.g. use the value from disk, not ours, *unless* we're the one
1932
1983
builder.add_node(key, value)
1933
1984
self.transport.put_file('pack-names', builder.finish(),
1934
1985
mode=self.repo.bzrdir._get_file_mode())
1935
# move the baseline forward
1936
1986
self._packs_at_load = disk_nodes
1937
1987
if clear_obsolete_packs:
1938
self._clear_obsolete_packs()
1990
to_preserve = set([o.name for o in obsolete_packs])
1991
already_obsolete = self._clear_obsolete_packs(to_preserve)
1940
1993
self._unlock_names()
1941
1994
# synchronise the memory packs list with what we just wrote:
1942
1995
self._syncronize_pack_names_from_disk_nodes(disk_nodes)
1997
# TODO: We could add one more condition here. "if o.name not in
1998
# orig_disk_nodes and o != the new_pack we haven't written to
1999
# disk yet. However, the new pack object is not easily
2000
# accessible here (it would have to be passed through the
2001
# autopacking code, etc.)
2002
obsolete_packs = [o for o in obsolete_packs
2003
if o.name not in already_obsolete]
2004
self._obsolete_packs(obsolete_packs)
2005
return [new_node[0][0] for new_node in new_nodes]
1944
2007
def reload_pack_names(self):
1945
2008
"""Sync our pack listing with what is present in the repository.
1961
2024
# out the new value.
1962
disk_nodes, _, _ = self._diff_pack_names()
1963
self._packs_at_load = disk_nodes
2025
(disk_nodes, deleted_nodes, new_nodes,
2026
orig_disk_nodes) = self._diff_pack_names()
2027
# _packs_at_load is meant to be the explicit list of names in
2028
# 'pack-names' at then start. As such, it should not contain any
2029
# pending names that haven't been written out yet.
2030
self._packs_at_load = orig_disk_nodes
1964
2031
(removed, added,
1965
2032
modified) = self._syncronize_pack_names_from_disk_nodes(disk_nodes)
1966
2033
if removed or added or modified:
1976
2043
raise errors.RetryAutopack(self.repo, False, sys.exc_info())
1978
def _clear_obsolete_packs(self):
2045
def _clear_obsolete_packs(self, preserve=None):
1979
2046
"""Delete everything from the obsolete-packs directory.
2048
:return: A list of pack identifiers (the filename without '.pack') that
2049
were found in obsolete_packs.
1981
2052
obsolete_pack_transport = self.transport.clone('obsolete_packs')
2053
if preserve is None:
1982
2055
for filename in obsolete_pack_transport.list_dir('.'):
2056
name, ext = osutils.splitext(filename)
2059
if name in preserve:
1984
2062
obsolete_pack_transport.delete(filename)
1985
2063
except (errors.PathError, errors.TransportError), e:
1986
warning("couldn't delete obsolete pack, skipping it:\n%s" % (e,))
2064
warning("couldn't delete obsolete pack, skipping it:\n%s"
1988
2068
def _start_write_group(self):
1989
2069
# Do not permit preparation for writing if we're not in a 'write lock'.
2016
2096
# FIXME: just drop the transient index.
2017
2097
# forget what names there are
2018
2098
if self._new_pack is not None:
2020
self._new_pack.abort()
2022
# XXX: If we aborted while in the middle of finishing the write
2023
# group, _remove_pack_indices can fail because the indexes are
2024
# already gone. If they're not there we shouldn't fail in this
2025
# case. -- mbp 20081113
2026
self._remove_pack_indices(self._new_pack)
2027
self._new_pack = None
2099
operation = cleanup.OperationWithCleanups(self._new_pack.abort)
2100
operation.add_cleanup(setattr, self, '_new_pack', None)
2101
# If we aborted while in the middle of finishing the write
2102
# group, _remove_pack_indices could fail because the indexes are
2103
# already gone. But they're not there we shouldn't fail in this
2104
# case, so we pass ignore_missing=True.
2105
operation.add_cleanup(self._remove_pack_indices, self._new_pack,
2106
ignore_missing=True)
2107
operation.run_simple()
2028
2108
for resumed_pack in self._resumed_packs:
2030
resumed_pack.abort()
2032
# See comment in previous finally block.
2034
self._remove_pack_indices(resumed_pack)
2109
operation = cleanup.OperationWithCleanups(resumed_pack.abort)
2110
# See comment in previous finally block.
2111
operation.add_cleanup(self._remove_pack_indices, resumed_pack,
2112
ignore_missing=True)
2113
operation.run_simple()
2037
2114
del self._resumed_packs[:]
2038
self.repo._text_knit = None
2040
2116
def _remove_resumed_pack_indices(self):
2041
2117
for resumed_pack in self._resumed_packs:
2042
2118
self._remove_pack_indices(resumed_pack)
2043
2119
del self._resumed_packs[:]
2121
def _check_new_inventories(self):
2122
"""Detect missing inventories in this write group.
2124
:returns: list of strs, summarising any problems found. If the list is
2125
empty no problems were found.
2127
# The base implementation does no checks. GCRepositoryPackCollection
2045
2131
def _commit_write_group(self):
2046
2132
all_missing = set()
2047
2133
for prefix, versioned_file in (
2056
2142
raise errors.BzrCheckError(
2057
2143
"Repository %s has missing compression parent(s) %r "
2058
2144
% (self.repo, sorted(all_missing)))
2145
problems = self._check_new_inventories()
2147
problems_summary = '\n'.join(problems)
2148
raise errors.BzrCheckError(
2149
"Cannot add revision(s) to repository: " + problems_summary)
2059
2150
self._remove_pack_indices(self._new_pack)
2060
should_autopack = False
2151
any_new_content = False
2061
2152
if self._new_pack.data_inserted():
2062
2153
# get all the data to disk and read to use
2063
2154
self._new_pack.finish()
2064
2155
self.allocate(self._new_pack)
2065
2156
self._new_pack = None
2066
should_autopack = True
2157
any_new_content = True
2068
2159
self._new_pack.abort()
2069
2160
self._new_pack = None
2150
2242
self.revisions = KnitVersionedFiles(
2151
2243
_KnitGraphIndex(self._pack_collection.revision_index.combined_index,
2152
2244
add_callback=self._pack_collection.revision_index.add_callback,
2153
deltas=False, parents=True, is_locked=self.is_locked),
2245
deltas=False, parents=True, is_locked=self.is_locked,
2246
track_external_parent_refs=True),
2154
2247
data_access=self._pack_collection.revision_index.data_access,
2155
2248
max_delta_chain=0)
2156
2249
self.signatures = KnitVersionedFiles(
2189
2282
self._reconcile_fixes_text_parents = True
2190
2283
self._reconcile_backsup_inventory = False
2192
def _warn_if_deprecated(self):
2285
def _warn_if_deprecated(self, branch=None):
2193
2286
# This class isn't deprecated, but one sub-format is
2194
2287
if isinstance(self._format, RepositoryFormatKnitPack5RichRootBroken):
2195
from bzrlib import repository
2196
if repository._deprecation_warning_done:
2198
repository._deprecation_warning_done = True
2199
warning("Format %s for %s is deprecated - please use"
2200
" 'bzr upgrade --1.6.1-rich-root'"
2201
% (self._format, self.bzrdir.transport.base))
2288
super(KnitPackRepository, self)._warn_if_deprecated(branch)
2203
2290
def _abort_write_group(self):
2291
self.revisions._index._key_dependencies.clear()
2204
2292
self._pack_collection._abort_write_group()
2206
def _find_inconsistent_revision_parents(self):
2207
"""Find revisions with incorrectly cached parents.
2209
:returns: an iterator yielding tuples of (revison-id, parents-in-index,
2210
parents-in-revision).
2212
if not self.is_locked():
2213
raise errors.ObjectNotLocked(self)
2214
pb = ui.ui_factory.nested_progress_bar()
2217
revision_nodes = self._pack_collection.revision_index \
2218
.combined_index.iter_all_entries()
2219
index_positions = []
2220
# Get the cached index values for all revisions, and also the
2221
# location in each index of the revision text so we can perform
2223
for index, key, value, refs in revision_nodes:
2224
node = (index, key, value, refs)
2225
index_memo = self.revisions._index._node_to_position(node)
2226
if index_memo[0] != index:
2227
raise AssertionError('%r != %r' % (index_memo[0], index))
2228
index_positions.append((index_memo, key[0],
2229
tuple(parent[0] for parent in refs[0])))
2230
pb.update("Reading revision index", 0, 0)
2231
index_positions.sort()
2233
pb.update("Checking cached revision graph", 0,
2234
len(index_positions))
2235
for offset in xrange(0, len(index_positions), 1000):
2236
pb.update("Checking cached revision graph", offset)
2237
to_query = index_positions[offset:offset + batch_size]
2240
rev_ids = [item[1] for item in to_query]
2241
revs = self.get_revisions(rev_ids)
2242
for revision, item in zip(revs, to_query):
2243
index_parents = item[2]
2244
rev_parents = tuple(revision.parent_ids)
2245
if index_parents != rev_parents:
2246
result.append((revision.revision_id, index_parents,
2294
def _get_source(self, to_format):
2295
if to_format.network_name() == self._format.network_name():
2296
return KnitPackStreamSource(self, to_format)
2297
return super(KnitPackRepository, self)._get_source(to_format)
2252
2299
def _make_parents_provider(self):
2253
2300
return graph.CachingParentsProvider(self)
2261
2308
self._pack_collection._start_write_group()
2263
2310
def _commit_write_group(self):
2264
return self._pack_collection._commit_write_group()
2311
hint = self._pack_collection._commit_write_group()
2312
self.revisions._index._key_dependencies.clear()
2266
2315
def suspend_write_group(self):
2267
2316
# XXX check self._write_group is self.get_transaction()?
2268
2317
tokens = self._pack_collection._suspend_write_group()
2318
self.revisions._index._key_dependencies.clear()
2269
2319
self._write_group = None
2272
2322
def _resume_write_group(self, tokens):
2273
2323
self._start_write_group()
2274
self._pack_collection._resume_write_group(tokens)
2325
self._pack_collection._resume_write_group(tokens)
2326
except errors.UnresumableWriteGroup:
2327
self._abort_write_group()
2329
for pack in self._pack_collection._resumed_packs:
2330
self.revisions._index.scan_unvalidated_index(pack.revision_index)
2276
2332
def get_transaction(self):
2277
2333
if self._write_lock_count:
2286
2342
return self._write_lock_count
2288
2344
def lock_write(self, token=None):
2345
"""Lock the repository for writes.
2347
:return: A bzrlib.repository.RepositoryWriteLockResult.
2289
2349
locked = self.is_locked()
2290
2350
if not self._write_lock_count and locked:
2291
2351
raise errors.ReadOnlyError(self)
2292
2352
self._write_lock_count += 1
2293
2353
if self._write_lock_count == 1:
2294
2354
self._transaction = transactions.WriteTransaction()
2356
if 'relock' in debug.debug_flags and self._prev_lock == 'w':
2357
note('%r was write locked again', self)
2358
self._prev_lock = 'w'
2295
2359
for repo in self._fallback_repositories:
2296
2360
# Writes don't affect fallback repos
2297
2361
repo.lock_read()
2299
2362
self._refresh_data()
2363
return RepositoryWriteLockResult(self.unlock, None)
2301
2365
def lock_read(self):
2366
"""Lock the repository for reads.
2368
:return: A bzrlib.lock.LogicalLockResult.
2302
2370
locked = self.is_locked()
2303
2371
if self._write_lock_count:
2304
2372
self._write_lock_count += 1
2306
2374
self.control_files.lock_read()
2376
if 'relock' in debug.debug_flags and self._prev_lock == 'r':
2377
note('%r was read locked again', self)
2378
self._prev_lock = 'r'
2307
2379
for repo in self._fallback_repositories:
2308
# Writes don't affect fallback repos
2309
2380
repo.lock_read()
2311
2381
self._refresh_data()
2382
return LogicalLockResult(self.unlock)
2313
2384
def leave_lock_in_place(self):
2314
2385
# not supported - raise an error
2353
2425
transaction = self._transaction
2354
2426
self._transaction = None
2355
2427
transaction.finish()
2356
for repo in self._fallback_repositories:
2359
2429
self.control_files.unlock()
2431
if not self.is_locked():
2360
2432
for repo in self._fallback_repositories:
2436
class KnitPackStreamSource(StreamSource):
2437
"""A StreamSource used to transfer data between same-format KnitPack repos.
2439
This source assumes:
2440
1) Same serialization format for all objects
2441
2) Same root information
2442
3) XML format inventories
2443
4) Atomic inserts (so we can stream inventory texts before text
2448
def __init__(self, from_repository, to_format):
2449
super(KnitPackStreamSource, self).__init__(from_repository, to_format)
2450
self._text_keys = None
2451
self._text_fetch_order = 'unordered'
2453
def _get_filtered_inv_stream(self, revision_ids):
2454
from_repo = self.from_repository
2455
parent_ids = from_repo._find_parent_ids_of_revisions(revision_ids)
2456
parent_keys = [(p,) for p in parent_ids]
2457
find_text_keys = from_repo._find_text_key_references_from_xml_inventory_lines
2458
parent_text_keys = set(find_text_keys(
2459
from_repo._inventory_xml_lines_for_keys(parent_keys)))
2460
content_text_keys = set()
2461
knit = KnitVersionedFiles(None, None)
2462
factory = KnitPlainFactory()
2463
def find_text_keys_from_content(record):
2464
if record.storage_kind not in ('knit-delta-gz', 'knit-ft-gz'):
2465
raise ValueError("Unknown content storage kind for"
2466
" inventory text: %s" % (record.storage_kind,))
2467
# It's a knit record, it has a _raw_record field (even if it was
2468
# reconstituted from a network stream).
2469
raw_data = record._raw_record
2470
# read the entire thing
2471
revision_id = record.key[-1]
2472
content, _ = knit._parse_record(revision_id, raw_data)
2473
if record.storage_kind == 'knit-delta-gz':
2474
line_iterator = factory.get_linedelta_content(content)
2475
elif record.storage_kind == 'knit-ft-gz':
2476
line_iterator = factory.get_fulltext_content(content)
2477
content_text_keys.update(find_text_keys(
2478
[(line, revision_id) for line in line_iterator]))
2479
revision_keys = [(r,) for r in revision_ids]
2480
def _filtered_inv_stream():
2481
source_vf = from_repo.inventories
2482
stream = source_vf.get_record_stream(revision_keys,
2484
for record in stream:
2485
if record.storage_kind == 'absent':
2486
raise errors.NoSuchRevision(from_repo, record.key)
2487
find_text_keys_from_content(record)
2489
self._text_keys = content_text_keys - parent_text_keys
2490
return ('inventories', _filtered_inv_stream())
2492
def _get_text_stream(self):
2493
# Note: We know we don't have to handle adding root keys, because both
2494
# the source and target are the identical network name.
2495
text_stream = self.from_repository.texts.get_record_stream(
2496
self._text_keys, self._text_fetch_order, False)
2497
return ('texts', text_stream)
2499
def get_stream(self, search):
2500
revision_ids = search.get_keys()
2501
for stream_info in self._fetch_revision_texts(revision_ids):
2503
self._revision_keys = [(rev_id,) for rev_id in revision_ids]
2504
yield self._get_filtered_inv_stream(revision_ids)
2505
yield self._get_text_stream()
2364
2509
class RepositoryFormatPack(MetaDirRepositoryFormat):
2365
2510
"""Format logic for pack structured repositories.
2413
2558
utf8_files = [('format', self.get_format_string())]
2415
2560
self._upload_blank_content(a_bzrdir, dirs, files, utf8_files, shared)
2416
return self.open(a_bzrdir=a_bzrdir, _found=True)
2561
repository = self.open(a_bzrdir=a_bzrdir, _found=True)
2562
self._run_post_repo_init_hooks(repository, a_bzrdir, shared)
2418
2565
def open(self, a_bzrdir, _found=False, _override_transport=None):
2419
2566
"""See RepositoryFormat.open().
2503
2648
_matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
2505
def check_conversion_target(self, target_format):
2506
if not target_format.rich_root_data:
2507
raise errors.BadConversionTarget(
2508
'Does not support rich root data.', target_format)
2509
if not getattr(target_format, 'supports_tree_reference', False):
2510
raise errors.BadConversionTarget(
2511
'Does not support nested trees', target_format)
2513
2650
def get_format_string(self):
2514
2651
"""See RepositoryFormat.get_format_string()."""
2515
2652
return "Bazaar pack repository format 1 with subtree support (needs bzr 0.92)\n"
2812
2924
_matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
2814
def check_conversion_target(self, target_format):
2815
if not target_format.rich_root_data:
2816
raise errors.BadConversionTarget(
2817
'Does not support rich root data.', target_format)
2818
if not getattr(target_format, 'supports_tree_reference', False):
2819
raise errors.BadConversionTarget(
2820
'Does not support nested trees', target_format)
2822
2926
def get_format_string(self):
2823
2927
"""See RepositoryFormat.get_format_string()."""
2824
2928
return ("Bazaar development format 2 with subtree support "
added
removed
bzrlib/repofmt/pack_repo.py
