~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/repository.py

merge 2.0 branch rev 4647

Show diffs side-by-side

added added

removed removed

Lines of Context:
31
31
    gpg,
32
32
    graph,
33
33
    inventory,
 
34
    inventory_delta,
34
35
    lazy_regex,
35
36
    lockable_files,
36
37
    lockdir,
57
58
    entry_factory,
58
59
    )
59
60
from bzrlib import registry
60
 
from bzrlib.symbol_versioning import (
61
 
        deprecated_method,
62
 
        )
63
61
from bzrlib.trace import (
64
62
    log_exception_quietly, note, mutter, mutter_callsite, warning)
65
63
 
467
465
            if content_summary[2] is None:
468
466
                raise ValueError("Files must not have executable = None")
469
467
            if not store:
470
 
                if (# if the file length changed we have to store:
471
 
                    parent_entry.text_size != content_summary[1] or
472
 
                    # if the exec bit has changed we have to store:
 
468
                # We can't trust a check of the file length because of content
 
469
                # filtering...
 
470
                if (# if the exec bit has changed we have to store:
473
471
                    parent_entry.executable != content_summary[2]):
474
472
                    store = True
475
473
                elif parent_entry.text_sha1 == content_summary[3]:
494
492
            ie.executable = content_summary[2]
495
493
            file_obj, stat_value = tree.get_file_with_stat(ie.file_id, path)
496
494
            try:
497
 
                lines = file_obj.readlines()
 
495
                text = file_obj.read()
498
496
            finally:
499
497
                file_obj.close()
500
498
            try:
501
499
                ie.text_sha1, ie.text_size = self._add_text_to_weave(
502
 
                    ie.file_id, lines, heads, nostore_sha)
 
500
                    ie.file_id, text, heads, nostore_sha)
503
501
                # Let the caller know we generated a stat fingerprint.
504
502
                fingerprint = (ie.text_sha1, stat_value)
505
503
            except errors.ExistingContent:
517
515
                # carry over:
518
516
                ie.revision = parent_entry.revision
519
517
                return self._get_delta(ie, basis_inv, path), False, None
520
 
            lines = []
521
 
            self._add_text_to_weave(ie.file_id, lines, heads, None)
 
518
            self._add_text_to_weave(ie.file_id, '', heads, None)
522
519
        elif kind == 'symlink':
523
520
            current_link_target = content_summary[3]
524
521
            if not store:
532
529
                ie.symlink_target = parent_entry.symlink_target
533
530
                return self._get_delta(ie, basis_inv, path), False, None
534
531
            ie.symlink_target = current_link_target
535
 
            lines = []
536
 
            self._add_text_to_weave(ie.file_id, lines, heads, None)
 
532
            self._add_text_to_weave(ie.file_id, '', heads, None)
537
533
        elif kind == 'tree-reference':
538
534
            if not store:
539
535
                if content_summary[3] != parent_entry.reference_revision:
544
540
                ie.revision = parent_entry.revision
545
541
                return self._get_delta(ie, basis_inv, path), False, None
546
542
            ie.reference_revision = content_summary[3]
547
 
            lines = []
548
 
            self._add_text_to_weave(ie.file_id, lines, heads, None)
 
543
            if ie.reference_revision is None:
 
544
                raise AssertionError("invalid content_summary for nested tree: %r"
 
545
                    % (content_summary,))
 
546
            self._add_text_to_weave(ie.file_id, '', heads, None)
549
547
        else:
550
548
            raise NotImplementedError('unknown kind')
551
549
        ie.revision = self._new_revision_id
745
743
                        entry.executable = True
746
744
                    else:
747
745
                        entry.executable = False
748
 
                    if (carry_over_possible and 
 
746
                    if (carry_over_possible and
749
747
                        parent_entry.executable == entry.executable):
750
748
                            # Check the file length, content hash after reading
751
749
                            # the file.
754
752
                        nostore_sha = None
755
753
                    file_obj, stat_value = tree.get_file_with_stat(file_id, change[1][1])
756
754
                    try:
757
 
                        lines = file_obj.readlines()
 
755
                        text = file_obj.read()
758
756
                    finally:
759
757
                        file_obj.close()
760
758
                    try:
761
759
                        entry.text_sha1, entry.text_size = self._add_text_to_weave(
762
 
                            file_id, lines, heads, nostore_sha)
 
760
                            file_id, text, heads, nostore_sha)
763
761
                        yield file_id, change[1][1], (entry.text_sha1, stat_value)
764
762
                    except errors.ExistingContent:
765
763
                        # No content change against a carry_over parent
774
772
                        parent_entry.symlink_target == entry.symlink_target):
775
773
                        carried_over = True
776
774
                    else:
777
 
                        self._add_text_to_weave(change[0], [], heads, None)
 
775
                        self._add_text_to_weave(change[0], '', heads, None)
778
776
                elif kind == 'directory':
779
777
                    if carry_over_possible:
780
778
                        carried_over = True
782
780
                        # Nothing to set on the entry.
783
781
                        # XXX: split into the Root and nonRoot versions.
784
782
                        if change[1][1] != '' or self.repository.supports_rich_root():
785
 
                            self._add_text_to_weave(change[0], [], heads, None)
 
783
                            self._add_text_to_weave(change[0], '', heads, None)
786
784
                elif kind == 'tree-reference':
787
785
                    if not self.repository._format.supports_tree_reference:
788
786
                        # This isn't quite sane as an error, but we shouldn't
791
789
                        # references.
792
790
                        raise errors.UnsupportedOperation(tree.add_reference,
793
791
                            self.repository)
794
 
                    entry.reference_revision = \
795
 
                        tree.get_reference_revision(change[0])
 
792
                    reference_revision = tree.get_reference_revision(change[0])
 
793
                    entry.reference_revision = reference_revision
796
794
                    if (carry_over_possible and
797
795
                        parent_entry.reference_revision == reference_revision):
798
796
                        carried_over = True
799
797
                    else:
800
 
                        self._add_text_to_weave(change[0], [], heads, None)
 
798
                        self._add_text_to_weave(change[0], '', heads, None)
801
799
                else:
802
800
                    raise AssertionError('unknown kind %r' % kind)
803
801
                if not carried_over:
812
810
                seen_root = True
813
811
        self.new_inventory = None
814
812
        if len(inv_delta):
 
813
            # This should perhaps be guarded by a check that the basis we
 
814
            # commit against is the basis for the commit and if not do a delta
 
815
            # against the basis.
815
816
            self._any_changes = True
816
817
        if not seen_root:
817
818
            # housekeeping root entry changes do not affect no-change commits.
818
819
            self._require_root_change(tree)
819
820
        self.basis_delta_revision = basis_revision_id
820
821
 
821
 
    def _add_text_to_weave(self, file_id, new_lines, parents, nostore_sha):
822
 
        # Note: as we read the content directly from the tree, we know its not
823
 
        # been turned into unicode or badly split - but a broken tree
824
 
        # implementation could give us bad output from readlines() so this is
825
 
        # not a guarantee of safety. What would be better is always checking
826
 
        # the content during test suite execution. RBC 20070912
827
 
        parent_keys = tuple((file_id, parent) for parent in parents)
828
 
        return self.repository.texts.add_lines(
829
 
            (file_id, self._new_revision_id), parent_keys, new_lines,
830
 
            nostore_sha=nostore_sha, random_id=self.random_revid,
831
 
            check_content=False)[0:2]
 
822
    def _add_text_to_weave(self, file_id, new_text, parents, nostore_sha):
 
823
        parent_keys = tuple([(file_id, parent) for parent in parents])
 
824
        return self.repository.texts._add_text(
 
825
            (file_id, self._new_revision_id), parent_keys, new_text,
 
826
            nostore_sha=nostore_sha, random_id=self.random_revid)[0:2]
832
827
 
833
828
 
834
829
class RootCommitBuilder(CommitBuilder):
860
855
######################################################################
861
856
# Repositories
862
857
 
 
858
 
863
859
class Repository(object):
864
860
    """Repository holding history for one or more branches.
865
861
 
935
931
        """
936
932
        if self._write_group is not self.get_transaction():
937
933
            # has an unlock or relock occured ?
 
934
            if suppress_errors:
 
935
                mutter(
 
936
                '(suppressed) mismatched lock context and write group. %r, %r',
 
937
                self._write_group, self.get_transaction())
 
938
                return
938
939
            raise errors.BzrError(
939
940
                'mismatched lock context and write group. %r, %r' %
940
941
                (self._write_group, self.get_transaction()))
1027
1028
                               parents, basis_inv=None, propagate_caches=False):
1028
1029
        """Add a new inventory expressed as a delta against another revision.
1029
1030
 
 
1031
        See the inventory developers documentation for the theory behind
 
1032
        inventory deltas.
 
1033
 
1030
1034
        :param basis_revision_id: The inventory id the delta was created
1031
1035
            against. (This does not have to be a direct parent.)
1032
1036
        :param delta: The inventory delta (see Inventory.apply_delta for
1071
1075
        check_content=True):
1072
1076
        """Store lines in inv_vf and return the sha1 of the inventory."""
1073
1077
        parents = [(parent,) for parent in parents]
1074
 
        return self.inventories.add_lines((revision_id,), parents, lines,
 
1078
        result = self.inventories.add_lines((revision_id,), parents, lines,
1075
1079
            check_content=check_content)[0]
 
1080
        self.inventories._access.flush()
 
1081
        return result
1076
1082
 
1077
1083
    def add_revision(self, revision_id, rev, inv=None, config=None):
1078
1084
        """Add rev to the revision store as revision_id.
1154
1160
        # The old API returned a list, should this actually be a set?
1155
1161
        return parent_map.keys()
1156
1162
 
 
1163
    def _check_inventories(self, checker):
 
1164
        """Check the inventories found from the revision scan.
 
1165
        
 
1166
        This is responsible for verifying the sha1 of inventories and
 
1167
        creating a pending_keys set that covers data referenced by inventories.
 
1168
        """
 
1169
        bar = ui.ui_factory.nested_progress_bar()
 
1170
        try:
 
1171
            self._do_check_inventories(checker, bar)
 
1172
        finally:
 
1173
            bar.finished()
 
1174
 
 
1175
    def _do_check_inventories(self, checker, bar):
 
1176
        """Helper for _check_inventories."""
 
1177
        revno = 0
 
1178
        keys = {'chk_bytes':set(), 'inventories':set(), 'texts':set()}
 
1179
        kinds = ['chk_bytes', 'texts']
 
1180
        count = len(checker.pending_keys)
 
1181
        bar.update("inventories", 0, 2)
 
1182
        current_keys = checker.pending_keys
 
1183
        checker.pending_keys = {}
 
1184
        # Accumulate current checks.
 
1185
        for key in current_keys:
 
1186
            if key[0] != 'inventories' and key[0] not in kinds:
 
1187
                checker._report_items.append('unknown key type %r' % (key,))
 
1188
            keys[key[0]].add(key[1:])
 
1189
        if keys['inventories']:
 
1190
            # NB: output order *should* be roughly sorted - topo or
 
1191
            # inverse topo depending on repository - either way decent
 
1192
            # to just delta against. However, pre-CHK formats didn't
 
1193
            # try to optimise inventory layout on disk. As such the
 
1194
            # pre-CHK code path does not use inventory deltas.
 
1195
            last_object = None
 
1196
            for record in self.inventories.check(keys=keys['inventories']):
 
1197
                if record.storage_kind == 'absent':
 
1198
                    checker._report_items.append(
 
1199
                        'Missing inventory {%s}' % (record.key,))
 
1200
                else:
 
1201
                    last_object = self._check_record('inventories', record,
 
1202
                        checker, last_object,
 
1203
                        current_keys[('inventories',) + record.key])
 
1204
            del keys['inventories']
 
1205
        else:
 
1206
            return
 
1207
        bar.update("texts", 1)
 
1208
        while (checker.pending_keys or keys['chk_bytes']
 
1209
            or keys['texts']):
 
1210
            # Something to check.
 
1211
            current_keys = checker.pending_keys
 
1212
            checker.pending_keys = {}
 
1213
            # Accumulate current checks.
 
1214
            for key in current_keys:
 
1215
                if key[0] not in kinds:
 
1216
                    checker._report_items.append('unknown key type %r' % (key,))
 
1217
                keys[key[0]].add(key[1:])
 
1218
            # Check the outermost kind only - inventories || chk_bytes || texts
 
1219
            for kind in kinds:
 
1220
                if keys[kind]:
 
1221
                    last_object = None
 
1222
                    for record in getattr(self, kind).check(keys=keys[kind]):
 
1223
                        if record.storage_kind == 'absent':
 
1224
                            checker._report_items.append(
 
1225
                                'Missing %s {%s}' % (kind, record.key,))
 
1226
                        else:
 
1227
                            last_object = self._check_record(kind, record,
 
1228
                                checker, last_object, current_keys[(kind,) + record.key])
 
1229
                    keys[kind] = set()
 
1230
                    break
 
1231
 
 
1232
    def _check_record(self, kind, record, checker, last_object, item_data):
 
1233
        """Check a single text from this repository."""
 
1234
        if kind == 'inventories':
 
1235
            rev_id = record.key[0]
 
1236
            inv = self.deserialise_inventory(rev_id,
 
1237
                record.get_bytes_as('fulltext'))
 
1238
            if last_object is not None:
 
1239
                delta = inv._make_delta(last_object)
 
1240
                for old_path, path, file_id, ie in delta:
 
1241
                    if ie is None:
 
1242
                        continue
 
1243
                    ie.check(checker, rev_id, inv)
 
1244
            else:
 
1245
                for path, ie in inv.iter_entries():
 
1246
                    ie.check(checker, rev_id, inv)
 
1247
            if self._format.fast_deltas:
 
1248
                return inv
 
1249
        elif kind == 'chk_bytes':
 
1250
            # No code written to check chk_bytes for this repo format.
 
1251
            checker._report_items.append(
 
1252
                'unsupported key type chk_bytes for %s' % (record.key,))
 
1253
        elif kind == 'texts':
 
1254
            self._check_text(record, checker, item_data)
 
1255
        else:
 
1256
            checker._report_items.append(
 
1257
                'unknown key type %s for %s' % (kind, record.key))
 
1258
 
 
1259
    def _check_text(self, record, checker, item_data):
 
1260
        """Check a single text."""
 
1261
        # Check it is extractable.
 
1262
        # TODO: check length.
 
1263
        if record.storage_kind == 'chunked':
 
1264
            chunks = record.get_bytes_as(record.storage_kind)
 
1265
            sha1 = osutils.sha_strings(chunks)
 
1266
            length = sum(map(len, chunks))
 
1267
        else:
 
1268
            content = record.get_bytes_as('fulltext')
 
1269
            sha1 = osutils.sha_string(content)
 
1270
            length = len(content)
 
1271
        if item_data and sha1 != item_data[1]:
 
1272
            checker._report_items.append(
 
1273
                'sha1 mismatch: %s has sha1 %s expected %s referenced by %s' %
 
1274
                (record.key, sha1, item_data[1], item_data[2]))
 
1275
 
1157
1276
    @staticmethod
1158
1277
    def create(a_bzrdir):
1159
1278
        """Construct the current default format repository in a_bzrdir."""
1192
1311
        self._inventory_entry_cache = fifo_cache.FIFOCache(10*1024)
1193
1312
 
1194
1313
    def __repr__(self):
1195
 
        return '%s(%r)' % (self.__class__.__name__,
1196
 
                           self.base)
 
1314
        if self._fallback_repositories:
 
1315
            return '%s(%r, fallback_repositories=%r)' % (
 
1316
                self.__class__.__name__,
 
1317
                self.base,
 
1318
                self._fallback_repositories)
 
1319
        else:
 
1320
            return '%s(%r)' % (self.__class__.__name__,
 
1321
                               self.base)
 
1322
 
 
1323
    def _has_same_fallbacks(self, other_repo):
 
1324
        """Returns true if the repositories have the same fallbacks."""
 
1325
        my_fb = self._fallback_repositories
 
1326
        other_fb = other_repo._fallback_repositories
 
1327
        if len(my_fb) != len(other_fb):
 
1328
            return False
 
1329
        for f, g in zip(my_fb, other_fb):
 
1330
            if not f.has_same_location(g):
 
1331
                return False
 
1332
        return True
1197
1333
 
1198
1334
    def has_same_location(self, other):
1199
1335
        """Returns a boolean indicating if this repository is at the same
1407
1543
        """Commit the contents accrued within the current write group.
1408
1544
 
1409
1545
        :seealso: start_write_group.
 
1546
        
 
1547
        :return: it may return an opaque hint that can be passed to 'pack'.
1410
1548
        """
1411
1549
        if self._write_group is not self.get_transaction():
1412
1550
            # has an unlock or relock occured ?
1413
1551
            raise errors.BzrError('mismatched lock context %r and '
1414
1552
                'write group %r.' %
1415
1553
                (self.get_transaction(), self._write_group))
1416
 
        self._commit_write_group()
 
1554
        result = self._commit_write_group()
1417
1555
        self._write_group = None
 
1556
        return result
1418
1557
 
1419
1558
    def _commit_write_group(self):
1420
1559
        """Template method for per-repository write group cleanup.
1537
1676
            raise errors.InternalBzrError(
1538
1677
                "May not fetch while in a write group.")
1539
1678
        # fast path same-url fetch operations
1540
 
        if self.has_same_location(source) and fetch_spec is None:
 
1679
        # TODO: lift out to somewhere common with RemoteRepository
 
1680
        # <https://bugs.edge.launchpad.net/bzr/+bug/401646>
 
1681
        if (self.has_same_location(source)
 
1682
            and fetch_spec is None
 
1683
            and self._has_same_fallbacks(source)):
1541
1684
            # check that last_revision is in 'from' and then return a
1542
1685
            # no-operation.
1543
1686
            if (revision_id is not None and
1568
1711
        :param revprops: Optional dictionary of revision properties.
1569
1712
        :param revision_id: Optional revision id.
1570
1713
        """
 
1714
        if self._fallback_repositories:
 
1715
            raise errors.BzrError("Cannot commit from a lightweight checkout "
 
1716
                "to a stacked branch. See "
 
1717
                "https://bugs.launchpad.net/bzr/+bug/375013 for details.")
1571
1718
        result = self._commit_builder_class(self, parents, config,
1572
1719
            timestamp, timezone, committer, revprops, revision_id)
1573
1720
        self.start_write_group()
1700
1847
 
1701
1848
    @needs_read_lock
1702
1849
    def get_revisions(self, revision_ids):
1703
 
        """Get many revisions at once."""
 
1850
        """Get many revisions at once.
 
1851
        
 
1852
        Repositories that need to check data on every revision read should 
 
1853
        subclass this method.
 
1854
        """
1704
1855
        return self._get_revisions(revision_ids)
1705
1856
 
1706
1857
    @needs_read_lock
1707
1858
    def _get_revisions(self, revision_ids):
1708
1859
        """Core work logic to get many revisions without sanity checks."""
1709
 
        for rev_id in revision_ids:
1710
 
            if not rev_id or not isinstance(rev_id, basestring):
1711
 
                raise errors.InvalidRevisionId(revision_id=rev_id, branch=self)
 
1860
        revs = {}
 
1861
        for revid, rev in self._iter_revisions(revision_ids):
 
1862
            if rev is None:
 
1863
                raise errors.NoSuchRevision(self, revid)
 
1864
            revs[revid] = rev
 
1865
        return [revs[revid] for revid in revision_ids]
 
1866
 
 
1867
    def _iter_revisions(self, revision_ids):
 
1868
        """Iterate over revision objects.
 
1869
 
 
1870
        :param revision_ids: An iterable of revisions to examine. None may be
 
1871
            passed to request all revisions known to the repository. Note that
 
1872
            not all repositories can find unreferenced revisions; for those
 
1873
            repositories only referenced ones will be returned.
 
1874
        :return: An iterator of (revid, revision) tuples. Absent revisions (
 
1875
            those asked for but not available) are returned as (revid, None).
 
1876
        """
 
1877
        if revision_ids is None:
 
1878
            revision_ids = self.all_revision_ids()
 
1879
        else:
 
1880
            for rev_id in revision_ids:
 
1881
                if not rev_id or not isinstance(rev_id, basestring):
 
1882
                    raise errors.InvalidRevisionId(revision_id=rev_id, branch=self)
1712
1883
        keys = [(key,) for key in revision_ids]
1713
1884
        stream = self.revisions.get_record_stream(keys, 'unordered', True)
1714
 
        revs = {}
1715
1885
        for record in stream:
 
1886
            revid = record.key[0]
1716
1887
            if record.storage_kind == 'absent':
1717
 
                raise errors.NoSuchRevision(self, record.key[0])
1718
 
            text = record.get_bytes_as('fulltext')
1719
 
            rev = self._serializer.read_revision_from_string(text)
1720
 
            revs[record.key[0]] = rev
1721
 
        return [revs[revid] for revid in revision_ids]
 
1888
                yield (revid, None)
 
1889
            else:
 
1890
                text = record.get_bytes_as('fulltext')
 
1891
                rev = self._serializer.read_revision_from_string(text)
 
1892
                yield (revid, rev)
1722
1893
 
1723
1894
    @needs_read_lock
1724
1895
    def get_revision_xml(self, revision_id):
1919
2090
                    yield line, revid
1920
2091
 
1921
2092
    def _find_file_ids_from_xml_inventory_lines(self, line_iterator,
1922
 
        revision_ids):
 
2093
        revision_keys):
1923
2094
        """Helper routine for fileids_altered_by_revision_ids.
1924
2095
 
1925
2096
        This performs the translation of xml lines to revision ids.
1926
2097
 
1927
2098
        :param line_iterator: An iterator of lines, origin_version_id
1928
 
        :param revision_ids: The revision ids to filter for. This should be a
 
2099
        :param revision_keys: The revision ids to filter for. This should be a
1929
2100
            set or other type which supports efficient __contains__ lookups, as
1930
 
            the revision id from each parsed line will be looked up in the
1931
 
            revision_ids filter.
 
2101
            the revision key from each parsed line will be looked up in the
 
2102
            revision_keys filter.
1932
2103
        :return: a dictionary mapping altered file-ids to an iterable of
1933
2104
        revision_ids. Each altered file-ids has the exact revision_ids that
1934
2105
        altered it listed explicitly.
1935
2106
        """
1936
2107
        seen = set(self._find_text_key_references_from_xml_inventory_lines(
1937
2108
                line_iterator).iterkeys())
1938
 
        # Note that revision_ids are revision keys.
1939
 
        parent_maps = self.revisions.get_parent_map(revision_ids)
1940
 
        parents = set()
1941
 
        map(parents.update, parent_maps.itervalues())
1942
 
        parents.difference_update(revision_ids)
 
2109
        parent_keys = self._find_parent_keys_of_revisions(revision_keys)
1943
2110
        parent_seen = set(self._find_text_key_references_from_xml_inventory_lines(
1944
 
            self._inventory_xml_lines_for_keys(parents)))
 
2111
            self._inventory_xml_lines_for_keys(parent_keys)))
1945
2112
        new_keys = seen - parent_seen
1946
2113
        result = {}
1947
2114
        setdefault = result.setdefault
1949
2116
            setdefault(key[0], set()).add(key[-1])
1950
2117
        return result
1951
2118
 
 
2119
    def _find_parent_ids_of_revisions(self, revision_ids):
 
2120
        """Find all parent ids that are mentioned in the revision graph.
 
2121
 
 
2122
        :return: set of revisions that are parents of revision_ids which are
 
2123
            not part of revision_ids themselves
 
2124
        """
 
2125
        parent_map = self.get_parent_map(revision_ids)
 
2126
        parent_ids = set()
 
2127
        map(parent_ids.update, parent_map.itervalues())
 
2128
        parent_ids.difference_update(revision_ids)
 
2129
        parent_ids.discard(_mod_revision.NULL_REVISION)
 
2130
        return parent_ids
 
2131
 
 
2132
    def _find_parent_keys_of_revisions(self, revision_keys):
 
2133
        """Similar to _find_parent_ids_of_revisions, but used with keys.
 
2134
 
 
2135
        :param revision_keys: An iterable of revision_keys.
 
2136
        :return: The parents of all revision_keys that are not already in
 
2137
            revision_keys
 
2138
        """
 
2139
        parent_map = self.revisions.get_parent_map(revision_keys)
 
2140
        parent_keys = set()
 
2141
        map(parent_keys.update, parent_map.itervalues())
 
2142
        parent_keys.difference_update(revision_keys)
 
2143
        parent_keys.discard(_mod_revision.NULL_REVISION)
 
2144
        return parent_keys
 
2145
 
1952
2146
    def fileids_altered_by_revision_ids(self, revision_ids, _inv_weave=None):
1953
2147
        """Find the file ids and versions affected by revisions.
1954
2148
 
2056
2250
                batch_size]
2057
2251
            if not to_query:
2058
2252
                break
2059
 
            for rev_tree in self.revision_trees(to_query):
2060
 
                revision_id = rev_tree.get_revision_id()
 
2253
            for revision_id in to_query:
2061
2254
                parent_ids = ancestors[revision_id]
2062
2255
                for text_key in revision_keys[revision_id]:
2063
2256
                    pb.update("Calculating text parents", processed_texts)
2163
2356
        """Get Inventory object by revision id."""
2164
2357
        return self.iter_inventories([revision_id]).next()
2165
2358
 
2166
 
    def iter_inventories(self, revision_ids):
 
2359
    def iter_inventories(self, revision_ids, ordering=None):
2167
2360
        """Get many inventories by revision_ids.
2168
2361
 
2169
2362
        This will buffer some or all of the texts used in constructing the
2171
2364
        time.
2172
2365
 
2173
2366
        :param revision_ids: The expected revision ids of the inventories.
 
2367
        :param ordering: optional ordering, e.g. 'topological'.  If not
 
2368
            specified, the order of revision_ids will be preserved (by
 
2369
            buffering if necessary).
2174
2370
        :return: An iterator of inventories.
2175
2371
        """
2176
2372
        if ((None in revision_ids)
2177
2373
            or (_mod_revision.NULL_REVISION in revision_ids)):
2178
2374
            raise ValueError('cannot get null revision inventory')
2179
 
        return self._iter_inventories(revision_ids)
 
2375
        return self._iter_inventories(revision_ids, ordering)
2180
2376
 
2181
 
    def _iter_inventories(self, revision_ids):
 
2377
    def _iter_inventories(self, revision_ids, ordering):
2182
2378
        """single-document based inventory iteration."""
2183
 
        for text, revision_id in self._iter_inventory_xmls(revision_ids):
 
2379
        inv_xmls = self._iter_inventory_xmls(revision_ids, ordering)
 
2380
        for text, revision_id in inv_xmls:
2184
2381
            yield self.deserialise_inventory(revision_id, text)
2185
2382
 
2186
 
    def _iter_inventory_xmls(self, revision_ids):
 
2383
    def _iter_inventory_xmls(self, revision_ids, ordering):
 
2384
        if ordering is None:
 
2385
            order_as_requested = True
 
2386
            ordering = 'unordered'
 
2387
        else:
 
2388
            order_as_requested = False
2187
2389
        keys = [(revision_id,) for revision_id in revision_ids]
2188
 
        stream = self.inventories.get_record_stream(keys, 'unordered', True)
 
2390
        if not keys:
 
2391
            return
 
2392
        if order_as_requested:
 
2393
            key_iter = iter(keys)
 
2394
            next_key = key_iter.next()
 
2395
        stream = self.inventories.get_record_stream(keys, ordering, True)
2189
2396
        text_chunks = {}
2190
2397
        for record in stream:
2191
2398
            if record.storage_kind != 'absent':
2192
 
                text_chunks[record.key] = record.get_bytes_as('chunked')
 
2399
                chunks = record.get_bytes_as('chunked')
 
2400
                if order_as_requested:
 
2401
                    text_chunks[record.key] = chunks
 
2402
                else:
 
2403
                    yield ''.join(chunks), record.key[-1]
2193
2404
            else:
2194
2405
                raise errors.NoSuchRevision(self, record.key)
2195
 
        for key in keys:
2196
 
            chunks = text_chunks.pop(key)
2197
 
            yield ''.join(chunks), key[-1]
 
2406
            if order_as_requested:
 
2407
                # Yield as many results as we can while preserving order.
 
2408
                while next_key in text_chunks:
 
2409
                    chunks = text_chunks.pop(next_key)
 
2410
                    yield ''.join(chunks), next_key[-1]
 
2411
                    try:
 
2412
                        next_key = key_iter.next()
 
2413
                    except StopIteration:
 
2414
                        # We still want to fully consume the get_record_stream,
 
2415
                        # just in case it is not actually finished at this point
 
2416
                        next_key = None
 
2417
                        break
2198
2418
 
2199
2419
    def deserialise_inventory(self, revision_id, xml):
2200
2420
        """Transform the xml into an inventory object.
2221
2441
    @needs_read_lock
2222
2442
    def get_inventory_xml(self, revision_id):
2223
2443
        """Get inventory XML as a file object."""
2224
 
        texts = self._iter_inventory_xmls([revision_id])
 
2444
        texts = self._iter_inventory_xmls([revision_id], 'unordered')
2225
2445
        try:
2226
2446
            text, revision_id = texts.next()
2227
2447
        except StopIteration:
2234
2454
        """
2235
2455
        return self.get_revision(revision_id).inventory_sha1
2236
2456
 
 
2457
    def get_rev_id_for_revno(self, revno, known_pair):
 
2458
        """Return the revision id of a revno, given a later (revno, revid)
 
2459
        pair in the same history.
 
2460
 
 
2461
        :return: if found (True, revid).  If the available history ran out
 
2462
            before reaching the revno, then this returns
 
2463
            (False, (closest_revno, closest_revid)).
 
2464
        """
 
2465
        known_revno, known_revid = known_pair
 
2466
        partial_history = [known_revid]
 
2467
        distance_from_known = known_revno - revno
 
2468
        if distance_from_known < 0:
 
2469
            raise ValueError(
 
2470
                'requested revno (%d) is later than given known revno (%d)'
 
2471
                % (revno, known_revno))
 
2472
        try:
 
2473
            _iter_for_revno(
 
2474
                self, partial_history, stop_index=distance_from_known)
 
2475
        except errors.RevisionNotPresent, err:
 
2476
            if err.revision_id == known_revid:
 
2477
                # The start revision (known_revid) wasn't found.
 
2478
                raise
 
2479
            # This is a stacked repository with no fallbacks, or a there's a
 
2480
            # left-hand ghost.  Either way, even though the revision named in
 
2481
            # the error isn't in this repo, we know it's the next step in this
 
2482
            # left-hand history.
 
2483
            partial_history.append(err.revision_id)
 
2484
        if len(partial_history) <= distance_from_known:
 
2485
            # Didn't find enough history to get a revid for the revno.
 
2486
            earliest_revno = known_revno - len(partial_history) + 1
 
2487
            return (False, (earliest_revno, partial_history[-1]))
 
2488
        if len(partial_history) - 1 > distance_from_known:
 
2489
            raise AssertionError('_iter_for_revno returned too much history')
 
2490
        return (True, partial_history[-1])
 
2491
 
2237
2492
    def iter_reverse_revision_history(self, revision_id):
2238
2493
        """Iterate backwards through revision ids in the lefthand history
2239
2494
 
2369
2624
            keys = tsort.topo_sort(parent_map)
2370
2625
        return [None] + list(keys)
2371
2626
 
2372
 
    def pack(self):
 
2627
    def pack(self, hint=None):
2373
2628
        """Compress the data within the repository.
2374
2629
 
2375
2630
        This operation only makes sense for some repository types. For other
2378
2633
        This stub method does not require a lock, but subclasses should use
2379
2634
        @needs_write_lock as this is a long running call its reasonable to
2380
2635
        implicitly lock for the user.
 
2636
 
 
2637
        :param hint: If not supplied, the whole repository is packed.
 
2638
            If supplied, the repository may use the hint parameter as a
 
2639
            hint for the parts of the repository to pack. A hint can be
 
2640
            obtained from the result of commit_write_group(). Out of
 
2641
            date hints are simply ignored, because concurrent operations
 
2642
            can obsolete them rapidly.
2381
2643
        """
2382
2644
 
2383
2645
    def get_transaction(self):
2417
2679
                [parents_provider, other_repository._make_parents_provider()])
2418
2680
        return graph.Graph(parents_provider)
2419
2681
 
2420
 
    def _get_versioned_file_checker(self, text_key_references=None):
 
2682
    def _get_versioned_file_checker(self, text_key_references=None,
 
2683
        ancestors=None):
2421
2684
        """Return an object suitable for checking versioned files.
2422
2685
        
2423
2686
        :param text_key_references: if non-None, an already built
2425
2688
            to whether they were referred to by the inventory of the
2426
2689
            revision_id that they contain. If None, this will be
2427
2690
            calculated.
 
2691
        :param ancestors: Optional result from
 
2692
            self.get_graph().get_parent_map(self.all_revision_ids()) if already
 
2693
            available.
2428
2694
        """
2429
2695
        return _VersionedFileChecker(self,
2430
 
            text_key_references=text_key_references)
 
2696
            text_key_references=text_key_references, ancestors=ancestors)
2431
2697
 
2432
2698
    def revision_ids_to_search_result(self, result_set):
2433
2699
        """Convert a set of revision ids to a graph SearchResult."""
2483
2749
        return record.get_bytes_as('fulltext')
2484
2750
 
2485
2751
    @needs_read_lock
2486
 
    def check(self, revision_ids=None):
 
2752
    def check(self, revision_ids=None, callback_refs=None, check_repo=True):
2487
2753
        """Check consistency of all history of given revision_ids.
2488
2754
 
2489
2755
        Different repository implementations should override _check().
2490
2756
 
2491
2757
        :param revision_ids: A non-empty list of revision_ids whose ancestry
2492
2758
             will be checked.  Typically the last revision_id of a branch.
 
2759
        :param callback_refs: A dict of check-refs to resolve and callback
 
2760
            the check/_check method on the items listed as wanting the ref.
 
2761
            see bzrlib.check.
 
2762
        :param check_repo: If False do not check the repository contents, just 
 
2763
            calculate the data callback_refs requires and call them back.
2493
2764
        """
2494
 
        return self._check(revision_ids)
 
2765
        return self._check(revision_ids, callback_refs=callback_refs,
 
2766
            check_repo=check_repo)
2495
2767
 
2496
 
    def _check(self, revision_ids):
2497
 
        result = check.Check(self)
2498
 
        result.check()
 
2768
    def _check(self, revision_ids, callback_refs, check_repo):
 
2769
        result = check.Check(self, check_repo=check_repo)
 
2770
        result.check(callback_refs)
2499
2771
        return result
2500
2772
 
2501
2773
    def _warn_if_deprecated(self):
2786
3058
    # Does this format have < O(tree_size) delta generation. Used to hint what
2787
3059
    # code path for commit, amongst other things.
2788
3060
    fast_deltas = None
 
3061
    # Does doing a pack operation compress data? Useful for the pack UI command
 
3062
    # (so if there is one pack, the operation can still proceed because it may
 
3063
    # help), and for fetching when data won't have come from the same
 
3064
    # compressor.
 
3065
    pack_compresses = False
 
3066
    # Does the repository inventory storage understand references to trees?
 
3067
    supports_tree_reference = None
2789
3068
 
2790
3069
    def __str__(self):
2791
3070
        return "<%s>" % self.__class__.__name__
2895
3174
        raise NotImplementedError(self.network_name)
2896
3175
 
2897
3176
    def check_conversion_target(self, target_format):
2898
 
        raise NotImplementedError(self.check_conversion_target)
 
3177
        if self.rich_root_data and not target_format.rich_root_data:
 
3178
            raise errors.BadConversionTarget(
 
3179
                'Does not support rich root data.', target_format,
 
3180
                from_format=self)
 
3181
        if (self.supports_tree_reference and 
 
3182
            not getattr(target_format, 'supports_tree_reference', False)):
 
3183
            raise errors.BadConversionTarget(
 
3184
                'Does not support nested trees', target_format,
 
3185
                from_format=self)
2899
3186
 
2900
3187
    def open(self, a_bzrdir, _found=False):
2901
3188
        """Return an instance of this format for the bzrdir a_bzrdir.
3418
3705
        return self.source.revision_ids_to_search_result(result_set)
3419
3706
 
3420
3707
 
3421
 
class InterPackRepo(InterSameDataRepository):
3422
 
    """Optimised code paths between Pack based repositories."""
3423
 
 
3424
 
    @classmethod
3425
 
    def _get_repo_format_to_test(self):
3426
 
        from bzrlib.repofmt import pack_repo
3427
 
        return pack_repo.RepositoryFormatKnitPack6RichRoot()
3428
 
 
3429
 
    @staticmethod
3430
 
    def is_compatible(source, target):
3431
 
        """Be compatible with known Pack formats.
3432
 
 
3433
 
        We don't test for the stores being of specific types because that
3434
 
        could lead to confusing results, and there is no need to be
3435
 
        overly general.
3436
 
 
3437
 
        InterPackRepo does not support CHK based repositories.
3438
 
        """
3439
 
        from bzrlib.repofmt.pack_repo import RepositoryFormatPack
3440
 
        from bzrlib.repofmt.groupcompress_repo import RepositoryFormatCHK1
3441
 
        try:
3442
 
            are_packs = (isinstance(source._format, RepositoryFormatPack) and
3443
 
                isinstance(target._format, RepositoryFormatPack))
3444
 
            not_packs = (isinstance(source._format, RepositoryFormatCHK1) or
3445
 
                isinstance(target._format, RepositoryFormatCHK1))
3446
 
        except AttributeError:
3447
 
            return False
3448
 
        if not_packs or not are_packs:
3449
 
            return False
3450
 
        return InterRepository._same_model(source, target)
3451
 
 
3452
 
    @needs_write_lock
3453
 
    def fetch(self, revision_id=None, pb=None, find_ghosts=False,
3454
 
            fetch_spec=None):
3455
 
        """See InterRepository.fetch()."""
3456
 
        if (len(self.source._fallback_repositories) > 0 or
3457
 
            len(self.target._fallback_repositories) > 0):
3458
 
            # The pack layer is not aware of fallback repositories, so when
3459
 
            # fetching from a stacked repository or into a stacked repository
3460
 
            # we use the generic fetch logic which uses the VersionedFiles
3461
 
            # attributes on repository.
3462
 
            from bzrlib.fetch import RepoFetcher
3463
 
            fetcher = RepoFetcher(self.target, self.source, revision_id,
3464
 
                    pb, find_ghosts, fetch_spec=fetch_spec)
3465
 
        if fetch_spec is not None:
3466
 
            if len(list(fetch_spec.heads)) != 1:
3467
 
                raise AssertionError(
3468
 
                    "InterPackRepo.fetch doesn't support "
3469
 
                    "fetching multiple heads yet.")
3470
 
            revision_id = list(fetch_spec.heads)[0]
3471
 
            fetch_spec = None
3472
 
        if revision_id is None:
3473
 
            # TODO:
3474
 
            # everything to do - use pack logic
3475
 
            # to fetch from all packs to one without
3476
 
            # inventory parsing etc, IFF nothing to be copied is in the target.
3477
 
            # till then:
3478
 
            source_revision_ids = frozenset(self.source.all_revision_ids())
3479
 
            revision_ids = source_revision_ids - \
3480
 
                frozenset(self.target.get_parent_map(source_revision_ids))
3481
 
            revision_keys = [(revid,) for revid in revision_ids]
3482
 
            index = self.target._pack_collection.revision_index.combined_index
3483
 
            present_revision_ids = set(item[1][0] for item in
3484
 
                index.iter_entries(revision_keys))
3485
 
            revision_ids = set(revision_ids) - present_revision_ids
3486
 
            # implementing the TODO will involve:
3487
 
            # - detecting when all of a pack is selected
3488
 
            # - avoiding as much as possible pre-selection, so the
3489
 
            # more-core routines such as create_pack_from_packs can filter in
3490
 
            # a just-in-time fashion. (though having a HEADS list on a
3491
 
            # repository might make this a lot easier, because we could
3492
 
            # sensibly detect 'new revisions' without doing a full index scan.
3493
 
        elif _mod_revision.is_null(revision_id):
3494
 
            # nothing to do:
3495
 
            return (0, [])
3496
 
        else:
3497
 
            revision_ids = self.search_missing_revision_ids(revision_id,
3498
 
                find_ghosts=find_ghosts).get_keys()
3499
 
            if len(revision_ids) == 0:
3500
 
                return (0, [])
3501
 
        return self._pack(self.source, self.target, revision_ids)
3502
 
 
3503
 
    def _pack(self, source, target, revision_ids):
3504
 
        from bzrlib.repofmt.pack_repo import Packer
3505
 
        packs = source._pack_collection.all_packs()
3506
 
        pack = Packer(self.target._pack_collection, packs, '.fetch',
3507
 
            revision_ids).pack()
3508
 
        if pack is not None:
3509
 
            self.target._pack_collection._save_pack_names()
3510
 
            copied_revs = pack.get_revision_count()
3511
 
            # Trigger an autopack. This may duplicate effort as we've just done
3512
 
            # a pack creation, but for now it is simpler to think about as
3513
 
            # 'upload data, then repack if needed'.
3514
 
            self.target._pack_collection.autopack()
3515
 
            return (copied_revs, [])
3516
 
        else:
3517
 
            return (0, [])
3518
 
 
3519
 
    @needs_read_lock
3520
 
    def search_missing_revision_ids(self, revision_id=None, find_ghosts=True):
3521
 
        """See InterRepository.missing_revision_ids().
3522
 
 
3523
 
        :param find_ghosts: Find ghosts throughout the ancestry of
3524
 
            revision_id.
3525
 
        """
3526
 
        if not find_ghosts and revision_id is not None:
3527
 
            return self._walk_to_common_revisions([revision_id])
3528
 
        elif revision_id is not None:
3529
 
            # Find ghosts: search for revisions pointing from one repository to
3530
 
            # the other, and vice versa, anywhere in the history of revision_id.
3531
 
            graph = self.target.get_graph(other_repository=self.source)
3532
 
            searcher = graph._make_breadth_first_searcher([revision_id])
3533
 
            found_ids = set()
3534
 
            while True:
3535
 
                try:
3536
 
                    next_revs, ghosts = searcher.next_with_ghosts()
3537
 
                except StopIteration:
3538
 
                    break
3539
 
                if revision_id in ghosts:
3540
 
                    raise errors.NoSuchRevision(self.source, revision_id)
3541
 
                found_ids.update(next_revs)
3542
 
                found_ids.update(ghosts)
3543
 
            found_ids = frozenset(found_ids)
3544
 
            # Double query here: should be able to avoid this by changing the
3545
 
            # graph api further.
3546
 
            result_set = found_ids - frozenset(
3547
 
                self.target.get_parent_map(found_ids))
3548
 
        else:
3549
 
            source_ids = self.source.all_revision_ids()
3550
 
            # source_ids is the worst possible case we may need to pull.
3551
 
            # now we want to filter source_ids against what we actually
3552
 
            # have in target, but don't try to check for existence where we know
3553
 
            # we do not have a revision as that would be pointless.
3554
 
            target_ids = set(self.target.all_revision_ids())
3555
 
            result_set = set(source_ids).difference(target_ids)
3556
 
        return self.source.revision_ids_to_search_result(result_set)
3557
 
 
3558
 
 
3559
3708
class InterDifferingSerializer(InterRepository):
3560
3709
 
3561
3710
    @classmethod
3568
3717
        # This is redundant with format.check_conversion_target(), however that
3569
3718
        # raises an exception, and we just want to say "False" as in we won't
3570
3719
        # support converting between these formats.
 
3720
        if 'IDS_never' in debug.debug_flags:
 
3721
            return False
3571
3722
        if source.supports_rich_root() and not target.supports_rich_root():
3572
3723
            return False
3573
3724
        if (source._format.supports_tree_reference
3574
3725
            and not target._format.supports_tree_reference):
3575
3726
            return False
 
3727
        if target._fallback_repositories and target._format.supports_chks:
 
3728
            # IDS doesn't know how to copy CHKs for the parent inventories it
 
3729
            # adds to stacked repos.
 
3730
            return False
 
3731
        if 'IDS_always' in debug.debug_flags:
 
3732
            return True
 
3733
        # Only use this code path for local source and target.  IDS does far
 
3734
        # too much IO (both bandwidth and roundtrips) over a network.
 
3735
        if not source.bzrdir.transport.base.startswith('file:///'):
 
3736
            return False
 
3737
        if not target.bzrdir.transport.base.startswith('file:///'):
 
3738
            return False
3576
3739
        return True
3577
3740
 
3578
 
    def _get_delta_for_revision(self, tree, parent_ids, basis_id, cache):
 
3741
    def _get_trees(self, revision_ids, cache):
 
3742
        possible_trees = []
 
3743
        for rev_id in revision_ids:
 
3744
            if rev_id in cache:
 
3745
                possible_trees.append((rev_id, cache[rev_id]))
 
3746
            else:
 
3747
                # Not cached, but inventory might be present anyway.
 
3748
                try:
 
3749
                    tree = self.source.revision_tree(rev_id)
 
3750
                except errors.NoSuchRevision:
 
3751
                    # Nope, parent is ghost.
 
3752
                    pass
 
3753
                else:
 
3754
                    cache[rev_id] = tree
 
3755
                    possible_trees.append((rev_id, tree))
 
3756
        return possible_trees
 
3757
 
 
3758
    def _get_delta_for_revision(self, tree, parent_ids, possible_trees):
3579
3759
        """Get the best delta and base for this revision.
3580
3760
 
3581
3761
        :return: (basis_id, delta)
3582
3762
        """
3583
 
        possible_trees = [(parent_id, cache[parent_id])
3584
 
                          for parent_id in parent_ids
3585
 
                           if parent_id in cache]
3586
 
        if len(possible_trees) == 0:
3587
 
            # There either aren't any parents, or the parents aren't in the
3588
 
            # cache, so just use the last converted tree
3589
 
            possible_trees.append((basis_id, cache[basis_id]))
3590
3763
        deltas = []
 
3764
        # Generate deltas against each tree, to find the shortest.
 
3765
        texts_possibly_new_in_tree = set()
3591
3766
        for basis_id, basis_tree in possible_trees:
3592
3767
            delta = tree.inventory._make_delta(basis_tree.inventory)
 
3768
            for old_path, new_path, file_id, new_entry in delta:
 
3769
                if new_path is None:
 
3770
                    # This file_id isn't present in the new rev, so we don't
 
3771
                    # care about it.
 
3772
                    continue
 
3773
                if not new_path:
 
3774
                    # Rich roots are handled elsewhere...
 
3775
                    continue
 
3776
                kind = new_entry.kind
 
3777
                if kind != 'directory' and kind != 'file':
 
3778
                    # No text record associated with this inventory entry.
 
3779
                    continue
 
3780
                # This is a directory or file that has changed somehow.
 
3781
                texts_possibly_new_in_tree.add((file_id, new_entry.revision))
3593
3782
            deltas.append((len(delta), basis_id, delta))
3594
3783
        deltas.sort()
3595
3784
        return deltas[0][1:]
3596
3785
 
3597
 
    def _get_parent_keys(self, root_key, parent_map):
3598
 
        """Get the parent keys for a given root id."""
3599
 
        root_id, rev_id = root_key
3600
 
        # Include direct parents of the revision, but only if they used
3601
 
        # the same root_id and are heads.
3602
 
        parent_keys = []
3603
 
        for parent_id in parent_map[rev_id]:
3604
 
            if parent_id == _mod_revision.NULL_REVISION:
3605
 
                continue
3606
 
            if parent_id not in self._revision_id_to_root_id:
3607
 
                # We probably didn't read this revision, go spend the
3608
 
                # extra effort to actually check
3609
 
                try:
3610
 
                    tree = self.source.revision_tree(parent_id)
3611
 
                except errors.NoSuchRevision:
3612
 
                    # Ghost, fill out _revision_id_to_root_id in case we
3613
 
                    # encounter this again.
3614
 
                    # But set parent_root_id to None since we don't really know
3615
 
                    parent_root_id = None
3616
 
                else:
3617
 
                    parent_root_id = tree.get_root_id()
3618
 
                self._revision_id_to_root_id[parent_id] = None
3619
 
            else:
3620
 
                parent_root_id = self._revision_id_to_root_id[parent_id]
3621
 
            if root_id == parent_root_id:
3622
 
                # With stacking we _might_ want to refer to a non-local
3623
 
                # revision, but this code path only applies when we have the
3624
 
                # full content available, so ghosts really are ghosts, not just
3625
 
                # the edge of local data.
3626
 
                parent_keys.append((parent_id,))
3627
 
            else:
3628
 
                # root_id may be in the parent anyway.
3629
 
                try:
3630
 
                    tree = self.source.revision_tree(parent_id)
3631
 
                except errors.NoSuchRevision:
3632
 
                    # ghost, can't refer to it.
3633
 
                    pass
3634
 
                else:
3635
 
                    try:
3636
 
                        parent_keys.append((tree.inventory[root_id].revision,))
3637
 
                    except errors.NoSuchId:
3638
 
                        # not in the tree
3639
 
                        pass
3640
 
        g = graph.Graph(self.source.revisions)
3641
 
        heads = g.heads(parent_keys)
3642
 
        selected_keys = []
3643
 
        for key in parent_keys:
3644
 
            if key in heads and key not in selected_keys:
3645
 
                selected_keys.append(key)
3646
 
        return tuple([(root_id,)+ key for key in selected_keys])
 
3786
    def _fetch_parent_invs_for_stacking(self, parent_map, cache):
 
3787
        """Find all parent revisions that are absent, but for which the
 
3788
        inventory is present, and copy those inventories.
3647
3789
 
3648
 
    def _new_root_data_stream(self, root_keys_to_create, parent_map):
3649
 
        for root_key in root_keys_to_create:
3650
 
            parent_keys = self._get_parent_keys(root_key, parent_map)
3651
 
            yield versionedfile.FulltextContentFactory(root_key,
3652
 
                parent_keys, None, '')
 
3790
        This is necessary to preserve correctness when the source is stacked
 
3791
        without fallbacks configured.  (Note that in cases like upgrade the
 
3792
        source may be not have _fallback_repositories even though it is
 
3793
        stacked.)
 
3794
        """
 
3795
        parent_revs = set()
 
3796
        for parents in parent_map.values():
 
3797
            parent_revs.update(parents)
 
3798
        present_parents = self.source.get_parent_map(parent_revs)
 
3799
        absent_parents = set(parent_revs).difference(present_parents)
 
3800
        parent_invs_keys_for_stacking = self.source.inventories.get_parent_map(
 
3801
            (rev_id,) for rev_id in absent_parents)
 
3802
        parent_inv_ids = [key[-1] for key in parent_invs_keys_for_stacking]
 
3803
        for parent_tree in self.source.revision_trees(parent_inv_ids):
 
3804
            current_revision_id = parent_tree.get_revision_id()
 
3805
            parents_parents_keys = parent_invs_keys_for_stacking[
 
3806
                (current_revision_id,)]
 
3807
            parents_parents = [key[-1] for key in parents_parents_keys]
 
3808
            basis_id = _mod_revision.NULL_REVISION
 
3809
            basis_tree = self.source.revision_tree(basis_id)
 
3810
            delta = parent_tree.inventory._make_delta(basis_tree.inventory)
 
3811
            self.target.add_inventory_by_delta(
 
3812
                basis_id, delta, current_revision_id, parents_parents)
 
3813
            cache[current_revision_id] = parent_tree
3653
3814
 
3654
3815
    def _fetch_batch(self, revision_ids, basis_id, cache):
3655
3816
        """Fetch across a few revisions.
3669
3830
        pending_deltas = []
3670
3831
        pending_revisions = []
3671
3832
        parent_map = self.source.get_parent_map(revision_ids)
 
3833
        self._fetch_parent_invs_for_stacking(parent_map, cache)
3672
3834
        for tree in self.source.revision_trees(revision_ids):
 
3835
            # Find a inventory delta for this revision.
 
3836
            # Find text entries that need to be copied, too.
3673
3837
            current_revision_id = tree.get_revision_id()
3674
3838
            parent_ids = parent_map.get(current_revision_id, ())
 
3839
            parent_trees = self._get_trees(parent_ids, cache)
 
3840
            possible_trees = list(parent_trees)
 
3841
            if len(possible_trees) == 0:
 
3842
                # There either aren't any parents, or the parents are ghosts,
 
3843
                # so just use the last converted tree.
 
3844
                possible_trees.append((basis_id, cache[basis_id]))
3675
3845
            basis_id, delta = self._get_delta_for_revision(tree, parent_ids,
3676
 
                                                           basis_id, cache)
 
3846
                                                           possible_trees)
3677
3847
            if self._converting_to_rich_root:
3678
3848
                self._revision_id_to_root_id[current_revision_id] = \
3679
3849
                    tree.get_root_id()
3680
 
            # Find text entries that need to be copied
 
3850
            # Determine which texts are in present in this revision but not in
 
3851
            # any of the available parents.
 
3852
            texts_possibly_new_in_tree = set()
3681
3853
            for old_path, new_path, file_id, entry in delta:
3682
 
                if new_path is not None:
3683
 
                    if not new_path:
3684
 
                        # This is the root
3685
 
                        if not self.target.supports_rich_root():
3686
 
                            # The target doesn't support rich root, so we don't
3687
 
                            # copy
3688
 
                            continue
3689
 
                        if self._converting_to_rich_root:
3690
 
                            # This can't be copied normally, we have to insert
3691
 
                            # it specially
3692
 
                            root_keys_to_create.add((file_id, entry.revision))
3693
 
                            continue
3694
 
                    text_keys.add((file_id, entry.revision))
 
3854
                if new_path is None:
 
3855
                    # This file_id isn't present in the new rev
 
3856
                    continue
 
3857
                if not new_path:
 
3858
                    # This is the root
 
3859
                    if not self.target.supports_rich_root():
 
3860
                        # The target doesn't support rich root, so we don't
 
3861
                        # copy
 
3862
                        continue
 
3863
                    if self._converting_to_rich_root:
 
3864
                        # This can't be copied normally, we have to insert
 
3865
                        # it specially
 
3866
                        root_keys_to_create.add((file_id, entry.revision))
 
3867
                        continue
 
3868
                kind = entry.kind
 
3869
                texts_possibly_new_in_tree.add((file_id, entry.revision))
 
3870
            for basis_id, basis_tree in possible_trees:
 
3871
                basis_inv = basis_tree.inventory
 
3872
                for file_key in list(texts_possibly_new_in_tree):
 
3873
                    file_id, file_revision = file_key
 
3874
                    try:
 
3875
                        entry = basis_inv[file_id]
 
3876
                    except errors.NoSuchId:
 
3877
                        continue
 
3878
                    if entry.revision == file_revision:
 
3879
                        texts_possibly_new_in_tree.remove(file_key)
 
3880
            text_keys.update(texts_possibly_new_in_tree)
3695
3881
            revision = self.source.get_revision(current_revision_id)
3696
3882
            pending_deltas.append((basis_id, delta,
3697
3883
                current_revision_id, revision.parent_ids))
3702
3888
        from_texts = self.source.texts
3703
3889
        to_texts = self.target.texts
3704
3890
        if root_keys_to_create:
3705
 
            root_stream = self._new_root_data_stream(root_keys_to_create,
3706
 
                                                     parent_map)
 
3891
            from bzrlib.fetch import _new_root_data_stream
 
3892
            root_stream = _new_root_data_stream(
 
3893
                root_keys_to_create, self._revision_id_to_root_id, parent_map,
 
3894
                self.source)
3707
3895
            to_texts.insert_record_stream(root_stream)
3708
3896
        to_texts.insert_record_stream(from_texts.get_record_stream(
3709
3897
            text_keys, self.target._format._fetch_order,
3716
3904
            # for the new revisions that we are about to insert.  We do this
3717
3905
            # before adding the revisions so that no revision is added until
3718
3906
            # all the inventories it may depend on are added.
 
3907
            # Note that this is overzealous, as we may have fetched these in an
 
3908
            # earlier batch.
3719
3909
            parent_ids = set()
3720
3910
            revision_ids = set()
3721
3911
            for revision in pending_revisions:
3724
3914
            parent_ids.difference_update(revision_ids)
3725
3915
            parent_ids.discard(_mod_revision.NULL_REVISION)
3726
3916
            parent_map = self.source.get_parent_map(parent_ids)
3727
 
            for parent_tree in self.source.revision_trees(parent_ids):
3728
 
                basis_id, delta = self._get_delta_for_revision(tree, parent_ids, basis_id, cache)
 
3917
            # we iterate over parent_map and not parent_ids because we don't
 
3918
            # want to try copying any revision which is a ghost
 
3919
            for parent_tree in self.source.revision_trees(parent_map):
3729
3920
                current_revision_id = parent_tree.get_revision_id()
3730
3921
                parents_parents = parent_map[current_revision_id]
 
3922
                possible_trees = self._get_trees(parents_parents, cache)
 
3923
                if len(possible_trees) == 0:
 
3924
                    # There either aren't any parents, or the parents are
 
3925
                    # ghosts, so just use the last converted tree.
 
3926
                    possible_trees.append((basis_id, cache[basis_id]))
 
3927
                basis_id, delta = self._get_delta_for_revision(parent_tree,
 
3928
                    parents_parents, possible_trees)
3731
3929
                self.target.add_inventory_by_delta(
3732
3930
                    basis_id, delta, current_revision_id, parents_parents)
3733
3931
        # insert signatures and revisions
3747
3945
 
3748
3946
        :param revision_ids: The list of revisions to fetch. Must be in
3749
3947
            topological order.
3750
 
        :param pb: A ProgressBar
 
3948
        :param pb: A ProgressTask
3751
3949
        :return: None
3752
3950
        """
3753
3951
        basis_id, basis_tree = self._get_basis(revision_ids[0])
3755
3953
        cache = lru_cache.LRUCache(100)
3756
3954
        cache[basis_id] = basis_tree
3757
3955
        del basis_tree # We don't want to hang on to it here
 
3956
        hints = []
3758
3957
        for offset in range(0, len(revision_ids), batch_size):
3759
3958
            self.target.start_write_group()
3760
3959
            try:
3766
3965
                self.target.abort_write_group()
3767
3966
                raise
3768
3967
            else:
3769
 
                self.target.commit_write_group()
 
3968
                hint = self.target.commit_write_group()
 
3969
                if hint:
 
3970
                    hints.extend(hint)
 
3971
        if hints and self.target._format.pack_compresses:
 
3972
            self.target.pack(hint=hints)
3770
3973
        pb.update('Transferring revisions', len(revision_ids),
3771
3974
                  len(revision_ids))
3772
3975
 
3793
3996
        # Walk though all revisions; get inventory deltas, copy referenced
3794
3997
        # texts that delta references, insert the delta, revision and
3795
3998
        # signature.
3796
 
        first_rev = self.source.get_revision(revision_ids[0])
3797
3999
        if pb is None:
3798
4000
            my_pb = ui.ui_factory.nested_progress_bar()
3799
4001
            pb = my_pb
3836
4038
InterRepository.register_optimiser(InterSameDataRepository)
3837
4039
InterRepository.register_optimiser(InterWeaveRepo)
3838
4040
InterRepository.register_optimiser(InterKnitRepo)
3839
 
InterRepository.register_optimiser(InterPackRepo)
3840
4041
 
3841
4042
 
3842
4043
class CopyConverter(object):
3923
4124
 
3924
4125
class _VersionedFileChecker(object):
3925
4126
 
3926
 
    def __init__(self, repository, text_key_references=None):
 
4127
    def __init__(self, repository, text_key_references=None, ancestors=None):
3927
4128
        self.repository = repository
3928
4129
        self.text_index = self.repository._generate_text_key_index(
3929
 
            text_key_references=text_key_references)
 
4130
            text_key_references=text_key_references, ancestors=ancestors)
3930
4131
 
3931
4132
    def calculate_file_version_parents(self, text_key):
3932
4133
        """Calculate the correct parents for a file version according to
3950
4151
            revision_id) tuples for versions that are present in this versioned
3951
4152
            file, but not used by the corresponding inventory.
3952
4153
        """
 
4154
        local_progress = None
 
4155
        if progress_bar is None:
 
4156
            local_progress = ui.ui_factory.nested_progress_bar()
 
4157
            progress_bar = local_progress
 
4158
        try:
 
4159
            return self._check_file_version_parents(texts, progress_bar)
 
4160
        finally:
 
4161
            if local_progress:
 
4162
                local_progress.finished()
 
4163
 
 
4164
    def _check_file_version_parents(self, texts, progress_bar):
 
4165
        """See check_file_version_parents."""
3953
4166
        wrong_parents = {}
3954
4167
        self.file_ids = set([file_id for file_id, _ in
3955
4168
            self.text_index.iterkeys()])
3956
4169
        # text keys is now grouped by file_id
3957
 
        n_weaves = len(self.file_ids)
3958
 
        files_in_revisions = {}
3959
 
        revisions_of_files = {}
3960
4170
        n_versions = len(self.text_index)
3961
4171
        progress_bar.update('loading text store', 0, n_versions)
3962
4172
        parent_map = self.repository.texts.get_parent_map(self.text_index)
3964
4174
        text_keys = self.repository.texts.keys()
3965
4175
        unused_keys = frozenset(text_keys) - set(self.text_index)
3966
4176
        for num, key in enumerate(self.text_index.iterkeys()):
3967
 
            if progress_bar is not None:
3968
 
                progress_bar.update('checking text graph', num, n_versions)
 
4177
            progress_bar.update('checking text graph', num, n_versions)
3969
4178
            correct_parents = self.calculate_file_version_parents(key)
3970
4179
            try:
3971
4180
                knit_parents = parent_map[key]
4056
4265
            else:
4057
4266
                new_pack.set_write_cache_size(1024*1024)
4058
4267
        for substream_type, substream in stream:
 
4268
            if 'stream' in debug.debug_flags:
 
4269
                mutter('inserting substream: %s', substream_type)
4059
4270
            if substream_type == 'texts':
4060
4271
                self.target_repo.texts.insert_record_stream(substream)
4061
4272
            elif substream_type == 'inventories':
4065
4276
                else:
4066
4277
                    self._extract_and_insert_inventories(
4067
4278
                        substream, src_serializer)
 
4279
            elif substream_type == 'inventory-deltas':
 
4280
                self._extract_and_insert_inventory_deltas(
 
4281
                    substream, src_serializer)
4068
4282
            elif substream_type == 'chk_bytes':
4069
4283
                # XXX: This doesn't support conversions, as it assumes the
4070
4284
                #      conversion was done in the fetch code.
4115
4329
                # missing keys can handle suspending a write group).
4116
4330
                write_group_tokens = self.target_repo.suspend_write_group()
4117
4331
                return write_group_tokens, missing_keys
4118
 
        self.target_repo.commit_write_group()
 
4332
        hint = self.target_repo.commit_write_group()
 
4333
        if (to_serializer != src_serializer and
 
4334
            self.target_repo._format.pack_compresses):
 
4335
            self.target_repo.pack(hint=hint)
4119
4336
        return [], set()
4120
4337
 
4121
 
    def _extract_and_insert_inventories(self, substream, serializer):
 
4338
    def _extract_and_insert_inventory_deltas(self, substream, serializer):
 
4339
        target_rich_root = self.target_repo._format.rich_root_data
 
4340
        target_tree_refs = self.target_repo._format.supports_tree_reference
 
4341
        for record in substream:
 
4342
            # Insert the delta directly
 
4343
            inventory_delta_bytes = record.get_bytes_as('fulltext')
 
4344
            deserialiser = inventory_delta.InventoryDeltaDeserializer()
 
4345
            try:
 
4346
                parse_result = deserialiser.parse_text_bytes(
 
4347
                    inventory_delta_bytes)
 
4348
            except inventory_delta.IncompatibleInventoryDelta, err:
 
4349
                trace.mutter("Incompatible delta: %s", err.msg)
 
4350
                raise errors.IncompatibleRevision(self.target_repo._format)
 
4351
            basis_id, new_id, rich_root, tree_refs, inv_delta = parse_result
 
4352
            revision_id = new_id
 
4353
            parents = [key[0] for key in record.parents]
 
4354
            self.target_repo.add_inventory_by_delta(
 
4355
                basis_id, inv_delta, revision_id, parents)
 
4356
 
 
4357
    def _extract_and_insert_inventories(self, substream, serializer,
 
4358
            parse_delta=None):
4122
4359
        """Generate a new inventory versionedfile in target, converting data.
4123
4360
 
4124
4361
        The inventory is retrieved from the source, (deserializing it), and
4125
4362
        stored in the target (reserializing it in a different format).
4126
4363
        """
 
4364
        target_rich_root = self.target_repo._format.rich_root_data
 
4365
        target_tree_refs = self.target_repo._format.supports_tree_reference
4127
4366
        for record in substream:
 
4367
            # It's not a delta, so it must be a fulltext in the source
 
4368
            # serializer's format.
4128
4369
            bytes = record.get_bytes_as('fulltext')
4129
4370
            revision_id = record.key[0]
4130
4371
            inv = serializer.read_inventory_from_string(bytes, revision_id)
4131
4372
            parents = [key[0] for key in record.parents]
4132
4373
            self.target_repo.add_inventory(revision_id, inv, parents)
 
4374
            # No need to keep holding this full inv in memory when the rest of
 
4375
            # the substream is likely to be all deltas.
 
4376
            del inv
4133
4377
 
4134
4378
    def _extract_and_insert_revisions(self, substream, serializer):
4135
4379
        for record in substream:
4184
4428
        return [('signatures', signatures), ('revisions', revisions)]
4185
4429
 
4186
4430
    def _generate_root_texts(self, revs):
4187
 
        """This will be called by __fetch between fetching weave texts and
 
4431
        """This will be called by get_stream between fetching weave texts and
4188
4432
        fetching the inventory weave.
4189
 
 
4190
 
        Subclasses should override this if they need to generate root texts
4191
 
        after fetching weave texts.
4192
4433
        """
4193
4434
        if self._rich_root_upgrade():
4194
4435
            import bzrlib.fetch
4201
4442
        phase = 'file'
4202
4443
        revs = search.get_keys()
4203
4444
        graph = self.from_repository.get_graph()
4204
 
        revs = list(graph.iter_topo_order(revs))
 
4445
        revs = tsort.topo_sort(graph.get_parent_map(revs))
4205
4446
        data_to_fetch = self.from_repository.item_keys_introduced_by(revs)
4206
4447
        text_keys = []
4207
4448
        for knit_kind, file_id, revisions in data_to_fetch:
4226
4467
                # will be valid.
4227
4468
                for _ in self._generate_root_texts(revs):
4228
4469
                    yield _
4229
 
                # NB: This currently reopens the inventory weave in source;
4230
 
                # using a single stream interface instead would avoid this.
4231
 
                from_weave = self.from_repository.inventories
4232
4470
                # we fetch only the referenced inventories because we do not
4233
4471
                # know for unselected inventories whether all their required
4234
4472
                # texts are present in the other repository - it could be
4273
4511
            if not keys:
4274
4512
                # No need to stream something we don't have
4275
4513
                continue
 
4514
            if substream_kind == 'inventories':
 
4515
                # Some missing keys are genuinely ghosts, filter those out.
 
4516
                present = self.from_repository.inventories.get_parent_map(keys)
 
4517
                revs = [key[0] for key in present]
 
4518
                # Get the inventory stream more-or-less as we do for the
 
4519
                # original stream; there's no reason to assume that records
 
4520
                # direct from the source will be suitable for the sink.  (Think
 
4521
                # e.g. 2a -> 1.9-rich-root).
 
4522
                for info in self._get_inventory_stream(revs, missing=True):
 
4523
                    yield info
 
4524
                continue
 
4525
 
4276
4526
            # Ask for full texts always so that we don't need more round trips
4277
4527
            # after this stream.
4278
4528
            # Some of the missing keys are genuinely ghosts, so filter absent
4293
4543
        return (not self.from_repository._format.rich_root_data and
4294
4544
            self.to_format.rich_root_data)
4295
4545
 
4296
 
    def _get_inventory_stream(self, revision_ids):
 
4546
    def _get_inventory_stream(self, revision_ids, missing=False):
4297
4547
        from_format = self.from_repository._format
4298
 
        if (from_format.supports_chks and self.to_format.supports_chks
4299
 
            and (from_format._serializer == self.to_format._serializer)):
4300
 
            # Both sides support chks, and they use the same serializer, so it
4301
 
            # is safe to transmit the chk pages and inventory pages across
4302
 
            # as-is.
4303
 
            return self._get_chk_inventory_stream(revision_ids)
4304
 
        elif (not from_format.supports_chks):
4305
 
            # Source repository doesn't support chks. So we can transmit the
4306
 
            # inventories 'as-is' and either they are just accepted on the
4307
 
            # target, or the Sink will properly convert it.
4308
 
            return self._get_simple_inventory_stream(revision_ids)
 
4548
        if (from_format.supports_chks and self.to_format.supports_chks and
 
4549
            from_format.network_name() == self.to_format.network_name()):
 
4550
            raise AssertionError(
 
4551
                "this case should be handled by GroupCHKStreamSource")
 
4552
        elif 'forceinvdeltas' in debug.debug_flags:
 
4553
            return self._get_convertable_inventory_stream(revision_ids,
 
4554
                    delta_versus_null=missing)
 
4555
        elif from_format.network_name() == self.to_format.network_name():
 
4556
            # Same format.
 
4557
            return self._get_simple_inventory_stream(revision_ids,
 
4558
                    missing=missing)
 
4559
        elif (not from_format.supports_chks and not self.to_format.supports_chks
 
4560
                and from_format._serializer == self.to_format._serializer):
 
4561
            # Essentially the same format.
 
4562
            return self._get_simple_inventory_stream(revision_ids,
 
4563
                    missing=missing)
4309
4564
        else:
4310
 
            # XXX: Hack to make not-chk->chk fetch: copy the inventories as
4311
 
            #      inventories. Note that this should probably be done somehow
4312
 
            #      as part of bzrlib.repository.StreamSink. Except JAM couldn't
4313
 
            #      figure out how a non-chk repository could possibly handle
4314
 
            #      deserializing an inventory stream from a chk repo, as it
4315
 
            #      doesn't have a way to understand individual pages.
4316
 
            return self._get_convertable_inventory_stream(revision_ids)
 
4565
            # Any time we switch serializations, we want to use an
 
4566
            # inventory-delta based approach.
 
4567
            return self._get_convertable_inventory_stream(revision_ids,
 
4568
                    delta_versus_null=missing)
4317
4569
 
4318
 
    def _get_simple_inventory_stream(self, revision_ids):
 
4570
    def _get_simple_inventory_stream(self, revision_ids, missing=False):
 
4571
        # NB: This currently reopens the inventory weave in source;
 
4572
        # using a single stream interface instead would avoid this.
4319
4573
        from_weave = self.from_repository.inventories
 
4574
        if missing:
 
4575
            delta_closure = True
 
4576
        else:
 
4577
            delta_closure = not self.delta_on_metadata()
4320
4578
        yield ('inventories', from_weave.get_record_stream(
4321
4579
            [(rev_id,) for rev_id in revision_ids],
4322
 
            self.inventory_fetch_order(),
4323
 
            not self.delta_on_metadata()))
4324
 
 
4325
 
    def _get_chk_inventory_stream(self, revision_ids):
4326
 
        """Fetch the inventory texts, along with the associated chk maps."""
4327
 
        # We want an inventory outside of the search set, so that we can filter
4328
 
        # out uninteresting chk pages. For now we use
4329
 
        # _find_revision_outside_set, but if we had a Search with cut_revs, we
4330
 
        # could use that instead.
4331
 
        start_rev_id = self.from_repository._find_revision_outside_set(
4332
 
                            revision_ids)
4333
 
        start_rev_key = (start_rev_id,)
4334
 
        inv_keys_to_fetch = [(rev_id,) for rev_id in revision_ids]
4335
 
        if start_rev_id != _mod_revision.NULL_REVISION:
4336
 
            inv_keys_to_fetch.append((start_rev_id,))
4337
 
        # Any repo that supports chk_bytes must also support out-of-order
4338
 
        # insertion. At least, that is how we expect it to work
4339
 
        # We use get_record_stream instead of iter_inventories because we want
4340
 
        # to be able to insert the stream as well. We could instead fetch
4341
 
        # allowing deltas, and then iter_inventories, but we don't know whether
4342
 
        # source or target is more 'local' anway.
4343
 
        inv_stream = self.from_repository.inventories.get_record_stream(
4344
 
            inv_keys_to_fetch, 'unordered',
4345
 
            True) # We need them as full-texts so we can find their references
4346
 
        uninteresting_chk_roots = set()
4347
 
        interesting_chk_roots = set()
4348
 
        def filter_inv_stream(inv_stream):
4349
 
            for idx, record in enumerate(inv_stream):
4350
 
                ### child_pb.update('fetch inv', idx, len(inv_keys_to_fetch))
4351
 
                bytes = record.get_bytes_as('fulltext')
4352
 
                chk_inv = inventory.CHKInventory.deserialise(
4353
 
                    self.from_repository.chk_bytes, bytes, record.key)
4354
 
                if record.key == start_rev_key:
4355
 
                    uninteresting_chk_roots.add(chk_inv.id_to_entry.key())
4356
 
                    p_id_map = chk_inv.parent_id_basename_to_file_id
4357
 
                    if p_id_map is not None:
4358
 
                        uninteresting_chk_roots.add(p_id_map.key())
4359
 
                else:
4360
 
                    yield record
4361
 
                    interesting_chk_roots.add(chk_inv.id_to_entry.key())
4362
 
                    p_id_map = chk_inv.parent_id_basename_to_file_id
4363
 
                    if p_id_map is not None:
4364
 
                        interesting_chk_roots.add(p_id_map.key())
4365
 
        ### pb.update('fetch inventory', 0, 2)
4366
 
        yield ('inventories', filter_inv_stream(inv_stream))
4367
 
        # Now that we have worked out all of the interesting root nodes, grab
4368
 
        # all of the interesting pages and insert them
4369
 
        ### pb.update('fetch inventory', 1, 2)
4370
 
        interesting = chk_map.iter_interesting_nodes(
4371
 
            self.from_repository.chk_bytes, interesting_chk_roots,
4372
 
            uninteresting_chk_roots)
4373
 
        def to_stream_adapter():
4374
 
            """Adapt the iter_interesting_nodes result to a single stream.
4375
 
 
4376
 
            iter_interesting_nodes returns records as it processes them, along
4377
 
            with keys. However, we only want to return the records themselves.
4378
 
            """
4379
 
            for record, items in interesting:
4380
 
                if record is not None:
4381
 
                    yield record
4382
 
        # XXX: We could instead call get_record_stream(records.keys())
4383
 
        #      ATM, this will always insert the records as fulltexts, and
4384
 
        #      requires that you can hang on to records once you have gone
4385
 
        #      on to the next one. Further, it causes the target to
4386
 
        #      recompress the data. Testing shows it to be faster than
4387
 
        #      requesting the records again, though.
4388
 
        yield ('chk_bytes', to_stream_adapter())
4389
 
        ### pb.update('fetch inventory', 2, 2)
4390
 
 
4391
 
    def _get_convertable_inventory_stream(self, revision_ids):
4392
 
        # XXX: One of source or target is using chks, and they don't have
4393
 
        #      compatible serializations. The StreamSink code expects to be
4394
 
        #      able to convert on the target, so we need to put
4395
 
        #      bytes-on-the-wire that can be converted
4396
 
        yield ('inventories', self._stream_invs_as_fulltexts(revision_ids))
4397
 
 
4398
 
    def _stream_invs_as_fulltexts(self, revision_ids):
 
4580
            self.inventory_fetch_order(), delta_closure))
 
4581
 
 
4582
    def _get_convertable_inventory_stream(self, revision_ids,
 
4583
                                          delta_versus_null=False):
 
4584
        # The source is using CHKs, but the target either doesn't or it has a
 
4585
        # different serializer.  The StreamSink code expects to be able to
 
4586
        # convert on the target, so we need to put bytes-on-the-wire that can
 
4587
        # be converted.  That means inventory deltas (if the remote is <1.19,
 
4588
        # RemoteStreamSink will fallback to VFS to insert the deltas).
 
4589
        yield ('inventory-deltas',
 
4590
           self._stream_invs_as_deltas(revision_ids,
 
4591
                                       delta_versus_null=delta_versus_null))
 
4592
 
 
4593
    def _stream_invs_as_deltas(self, revision_ids, delta_versus_null=False):
 
4594
        """Return a stream of inventory-deltas for the given rev ids.
 
4595
 
 
4596
        :param revision_ids: The list of inventories to transmit
 
4597
        :param delta_versus_null: Don't try to find a minimal delta for this
 
4598
            entry, instead compute the delta versus the NULL_REVISION. This
 
4599
            effectively streams a complete inventory. Used for stuff like
 
4600
            filling in missing parents, etc.
 
4601
        """
4399
4602
        from_repo = self.from_repository
4400
 
        from_serializer = from_repo._format._serializer
4401
4603
        revision_keys = [(rev_id,) for rev_id in revision_ids]
4402
4604
        parent_map = from_repo.inventories.get_parent_map(revision_keys)
4403
 
        for inv in self.from_repository.iter_inventories(revision_ids):
4404
 
            # XXX: This is a bit hackish, but it works. Basically,
4405
 
            #      CHKSerializer 'accidentally' supports
4406
 
            #      read/write_inventory_to_string, even though that is never
4407
 
            #      the format that is stored on disk. It *does* give us a
4408
 
            #      single string representation for an inventory, so live with
4409
 
            #      it for now.
4410
 
            #      This would be far better if we had a 'serialized inventory
4411
 
            #      delta' form. Then we could use 'inventory._make_delta', and
4412
 
            #      transmit that. This would both be faster to generate, and
4413
 
            #      result in fewer bytes-on-the-wire.
4414
 
            as_bytes = from_serializer.write_inventory_to_string(inv)
 
4605
        # XXX: possibly repos could implement a more efficient iter_inv_deltas
 
4606
        # method...
 
4607
        inventories = self.from_repository.iter_inventories(
 
4608
            revision_ids, 'topological')
 
4609
        format = from_repo._format
 
4610
        invs_sent_so_far = set([_mod_revision.NULL_REVISION])
 
4611
        inventory_cache = lru_cache.LRUCache(50)
 
4612
        null_inventory = from_repo.revision_tree(
 
4613
            _mod_revision.NULL_REVISION).inventory
 
4614
        # XXX: ideally the rich-root/tree-refs flags would be per-revision, not
 
4615
        # per-repo (e.g.  streaming a non-rich-root revision out of a rich-root
 
4616
        # repo back into a non-rich-root repo ought to be allowed)
 
4617
        serializer = inventory_delta.InventoryDeltaSerializer(
 
4618
            versioned_root=format.rich_root_data,
 
4619
            tree_references=format.supports_tree_reference)
 
4620
        for inv in inventories:
4415
4621
            key = (inv.revision_id,)
4416
4622
            parent_keys = parent_map.get(key, ())
 
4623
            delta = None
 
4624
            if not delta_versus_null and parent_keys:
 
4625
                # The caller did not ask for complete inventories and we have
 
4626
                # some parents that we can delta against.  Make a delta against
 
4627
                # each parent so that we can find the smallest.
 
4628
                parent_ids = [parent_key[0] for parent_key in parent_keys]
 
4629
                for parent_id in parent_ids:
 
4630
                    if parent_id not in invs_sent_so_far:
 
4631
                        # We don't know that the remote side has this basis, so
 
4632
                        # we can't use it.
 
4633
                        continue
 
4634
                    if parent_id == _mod_revision.NULL_REVISION:
 
4635
                        parent_inv = null_inventory
 
4636
                    else:
 
4637
                        parent_inv = inventory_cache.get(parent_id, None)
 
4638
                        if parent_inv is None:
 
4639
                            parent_inv = from_repo.get_inventory(parent_id)
 
4640
                    candidate_delta = inv._make_delta(parent_inv)
 
4641
                    if (delta is None or
 
4642
                        len(delta) > len(candidate_delta)):
 
4643
                        delta = candidate_delta
 
4644
                        basis_id = parent_id
 
4645
            if delta is None:
 
4646
                # Either none of the parents ended up being suitable, or we
 
4647
                # were asked to delta against NULL
 
4648
                basis_id = _mod_revision.NULL_REVISION
 
4649
                delta = inv._make_delta(null_inventory)
 
4650
            invs_sent_so_far.add(inv.revision_id)
 
4651
            inventory_cache[inv.revision_id] = inv
 
4652
            delta_serialized = ''.join(
 
4653
                serializer.delta_to_lines(basis_id, key[-1], delta))
4417
4654
            yield versionedfile.FulltextContentFactory(
4418
 
                key, parent_keys, None, as_bytes)
 
4655
                key, parent_keys, None, delta_serialized)
 
4656
 
 
4657
 
 
4658
def _iter_for_revno(repo, partial_history_cache, stop_index=None,
 
4659
                    stop_revision=None):
 
4660
    """Extend the partial history to include a given index
 
4661
 
 
4662
    If a stop_index is supplied, stop when that index has been reached.
 
4663
    If a stop_revision is supplied, stop when that revision is
 
4664
    encountered.  Otherwise, stop when the beginning of history is
 
4665
    reached.
 
4666
 
 
4667
    :param stop_index: The index which should be present.  When it is
 
4668
        present, history extension will stop.
 
4669
    :param stop_revision: The revision id which should be present.  When
 
4670
        it is encountered, history extension will stop.
 
4671
    """
 
4672
    start_revision = partial_history_cache[-1]
 
4673
    iterator = repo.iter_reverse_revision_history(start_revision)
 
4674
    try:
 
4675
        #skip the last revision in the list
 
4676
        iterator.next()
 
4677
        while True:
 
4678
            if (stop_index is not None and
 
4679
                len(partial_history_cache) > stop_index):
 
4680
                break
 
4681
            if partial_history_cache[-1] == stop_revision:
 
4682
                break
 
4683
            revision_id = iterator.next()
 
4684
            partial_history_cache.append(revision_id)
 
4685
    except StopIteration:
 
4686
        # No more history
 
4687
        return
4419
4688