/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/repofmt/groupcompress_repo.py

  • Committer: John Arbash Meinel
  • Date: 2009-06-12 18:05:15 UTC
  • mto: (4371.4.5 vila-better-heads)
  • mto: This revision was merged to the branch mainline in revision 4449.
  • Revision ID: john@arbash-meinel.com-20090612180515-t0cwbjsnve094oik
Add a failing test for handling nodes that are in the same linear chain.

It fails because the ancestry skipping causes us to miss the fact that the two nodes
are actually directly related. We could check at the beginning, as the 
code used to do, but I think that will be incomplete for the more-than-two
heads cases.

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2008, 2009, 2010 Canonical Ltd
 
1
# Copyright (C) 2008, 2009 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
29
29
    knit,
30
30
    osutils,
31
31
    pack,
 
32
    remote,
 
33
    repository,
32
34
    revision as _mod_revision,
33
35
    trace,
34
36
    ui,
37
39
    BTreeGraphIndex,
38
40
    BTreeBuilder,
39
41
    )
 
42
from bzrlib.index import GraphIndex, GraphIndexBuilder
40
43
from bzrlib.groupcompress import (
41
44
    _GCGraphIndex,
42
45
    GroupCompressVersionedFiles,
45
48
    Pack,
46
49
    NewPack,
47
50
    KnitPackRepository,
48
 
    KnitPackStreamSource,
49
51
    PackRootCommitBuilder,
50
52
    RepositoryPackCollection,
51
53
    RepositoryFormatPack,
52
54
    ResumedPack,
53
55
    Packer,
54
56
    )
55
 
from bzrlib.static_tuple import StaticTuple
56
57
 
57
58
 
58
59
class GCPack(NewPack):
154
155
        self._writer.begin()
155
156
        # what state is the pack in? (open, finished, aborted)
156
157
        self._state = 'open'
157
 
        # no name until we finish writing the content
158
 
        self.name = None
159
158
 
160
159
    def _check_references(self):
161
160
        """Make sure our external references are present.
218
217
            p_id_roots_set = set()
219
218
            stream = source_vf.get_record_stream(keys, 'groupcompress', True)
220
219
            for idx, record in enumerate(stream):
221
 
                # Inventories should always be with revisions; assume success.
222
220
                bytes = record.get_bytes_as('fulltext')
223
221
                chk_inv = inventory.CHKInventory.deserialise(None, bytes,
224
222
                                                             record.key)
295
293
                    stream = source_vf.get_record_stream(cur_keys,
296
294
                                                         'as-requested', True)
297
295
                    for record in stream:
298
 
                        if record.storage_kind == 'absent':
299
 
                            # An absent CHK record: we assume that the missing
300
 
                            # record is in a different pack - e.g. a page not
301
 
                            # altered by the commit we're packing.
302
 
                            continue
303
296
                        bytes = record.get_bytes_as('fulltext')
304
297
                        # We don't care about search_key_func for this code,
305
298
                        # because we only care about external references.
352
345
        """Build a VersionedFiles instance on top of this group of packs."""
353
346
        index_name = index_name + '_index'
354
347
        index_to_pack = {}
355
 
        access = knit._DirectPackAccess(index_to_pack,
356
 
                                        reload_func=self._reload_func)
 
348
        access = knit._DirectPackAccess(index_to_pack)
357
349
        if for_write:
358
350
            # Use new_pack
359
351
            if self.new_pack is None:
413
405
 
414
406
    def _copy_inventory_texts(self):
415
407
        source_vf, target_vf = self._build_vfs('inventory', True, True)
416
 
        # It is not sufficient to just use self.revision_keys, as stacked
417
 
        # repositories can have more inventories than they have revisions.
418
 
        # One alternative would be to do something with
419
 
        # get_parent_map(self.revision_keys), but that shouldn't be any faster
420
 
        # than this.
421
 
        inventory_keys = source_vf.keys()
422
 
        missing_inventories = set(self.revision_keys).difference(inventory_keys)
423
 
        if missing_inventories:
424
 
            missing_inventories = sorted(missing_inventories)
425
 
            raise ValueError('We are missing inventories for revisions: %s'
426
 
                % (missing_inventories,))
427
 
        self._copy_stream(source_vf, target_vf, inventory_keys,
 
408
        self._copy_stream(source_vf, target_vf, self.revision_keys,
428
409
                          'inventories', self._get_filtered_inv_stream, 2)
429
410
 
430
411
    def _copy_chk_texts(self):
457
438
        #      is grabbing too many keys...
458
439
        text_keys = source_vf.keys()
459
440
        self._copy_stream(source_vf, target_vf, text_keys,
460
 
                          'texts', self._get_progress_stream, 4)
 
441
                          'text', self._get_progress_stream, 4)
461
442
 
462
443
    def _copy_signature_texts(self):
463
444
        source_vf, target_vf = self._build_vfs('signature', False, False)
480
461
        if not self._use_pack(self.new_pack):
481
462
            self.new_pack.abort()
482
463
            return None
483
 
        self.new_pack.finish_content()
484
 
        if len(self.packs) == 1:
485
 
            old_pack = self.packs[0]
486
 
            if old_pack.name == self.new_pack._hash.hexdigest():
487
 
                # The single old pack was already optimally packed.
488
 
                trace.mutter('single pack %s was already optimally packed',
489
 
                    old_pack.name)
490
 
                self.new_pack.abort()
491
 
                return None
492
464
        self.pb.update('finishing repack', 6, 7)
493
465
        self.new_pack.finish()
494
466
        self._pack_collection.allocate(self.new_pack)
585
557
    pack_factory = GCPack
586
558
    resumed_pack_factory = ResumedGCPack
587
559
 
588
 
    def _check_new_inventories(self):
589
 
        """Detect missing inventories or chk root entries for the new revisions
590
 
        in this write group.
591
 
 
592
 
        :returns: list of strs, summarising any problems found.  If the list is
593
 
            empty no problems were found.
594
 
        """
595
 
        # Ensure that all revisions added in this write group have:
596
 
        #   - corresponding inventories,
597
 
        #   - chk root entries for those inventories,
598
 
        #   - and any present parent inventories have their chk root
599
 
        #     entries too.
600
 
        # And all this should be independent of any fallback repository.
601
 
        problems = []
602
 
        key_deps = self.repo.revisions._index._key_dependencies
603
 
        new_revisions_keys = key_deps.get_new_keys()
604
 
        no_fallback_inv_index = self.repo.inventories._index
605
 
        no_fallback_chk_bytes_index = self.repo.chk_bytes._index
606
 
        no_fallback_texts_index = self.repo.texts._index
607
 
        inv_parent_map = no_fallback_inv_index.get_parent_map(
608
 
            new_revisions_keys)
609
 
        # Are any inventories for corresponding to the new revisions missing?
610
 
        corresponding_invs = set(inv_parent_map)
611
 
        missing_corresponding = set(new_revisions_keys)
612
 
        missing_corresponding.difference_update(corresponding_invs)
613
 
        if missing_corresponding:
614
 
            problems.append("inventories missing for revisions %s" %
615
 
                (sorted(missing_corresponding),))
616
 
            return problems
617
 
        # Are any chk root entries missing for any inventories?  This includes
618
 
        # any present parent inventories, which may be used when calculating
619
 
        # deltas for streaming.
620
 
        all_inv_keys = set(corresponding_invs)
621
 
        for parent_inv_keys in inv_parent_map.itervalues():
622
 
            all_inv_keys.update(parent_inv_keys)
623
 
        # Filter out ghost parents.
624
 
        all_inv_keys.intersection_update(
625
 
            no_fallback_inv_index.get_parent_map(all_inv_keys))
626
 
        parent_invs_only_keys = all_inv_keys.symmetric_difference(
627
 
            corresponding_invs)
628
 
        all_missing = set()
629
 
        inv_ids = [key[-1] for key in all_inv_keys]
630
 
        parent_invs_only_ids = [key[-1] for key in parent_invs_only_keys]
631
 
        root_key_info = _build_interesting_key_sets(
632
 
            self.repo, inv_ids, parent_invs_only_ids)
633
 
        expected_chk_roots = root_key_info.all_keys()
634
 
        present_chk_roots = no_fallback_chk_bytes_index.get_parent_map(
635
 
            expected_chk_roots)
636
 
        missing_chk_roots = expected_chk_roots.difference(present_chk_roots)
637
 
        if missing_chk_roots:
638
 
            problems.append("missing referenced chk root keys: %s"
639
 
                % (sorted(missing_chk_roots),))
640
 
            # Don't bother checking any further.
641
 
            return problems
642
 
        # Find all interesting chk_bytes records, and make sure they are
643
 
        # present, as well as the text keys they reference.
644
 
        chk_bytes_no_fallbacks = self.repo.chk_bytes.without_fallbacks()
645
 
        chk_bytes_no_fallbacks._search_key_func = \
646
 
            self.repo.chk_bytes._search_key_func
647
 
        chk_diff = chk_map.iter_interesting_nodes(
648
 
            chk_bytes_no_fallbacks, root_key_info.interesting_root_keys,
649
 
            root_key_info.uninteresting_root_keys)
650
 
        bytes_to_info = inventory.CHKInventory._bytes_to_utf8name_key
651
 
        text_keys = set()
652
 
        try:
653
 
            for record in _filter_text_keys(chk_diff, text_keys, bytes_to_info):
654
 
                pass
655
 
        except errors.NoSuchRevision, e:
656
 
            # XXX: It would be nice if we could give a more precise error here.
657
 
            problems.append("missing chk node(s) for id_to_entry maps")
658
 
        chk_diff = chk_map.iter_interesting_nodes(
659
 
            chk_bytes_no_fallbacks, root_key_info.interesting_pid_root_keys,
660
 
            root_key_info.uninteresting_pid_root_keys)
661
 
        try:
662
 
            for interesting_rec, interesting_map in chk_diff:
663
 
                pass
664
 
        except errors.NoSuchRevision, e:
665
 
            problems.append(
666
 
                "missing chk node(s) for parent_id_basename_to_file_id maps")
667
 
        present_text_keys = no_fallback_texts_index.get_parent_map(text_keys)
668
 
        missing_text_keys = text_keys.difference(present_text_keys)
669
 
        if missing_text_keys:
670
 
            problems.append("missing text keys: %r"
671
 
                % (sorted(missing_text_keys),))
672
 
        return problems
 
560
    def _already_packed(self):
 
561
        """Is the collection already packed?"""
 
562
        # Always repack GC repositories for now
 
563
        return False
673
564
 
674
565
    def _execute_pack_operations(self, pack_operations,
675
566
                                 _packer_class=GCCHKPacker,
689
580
            packer = GCCHKPacker(self, packs, '.autopack',
690
581
                                 reload_func=reload_func)
691
582
            try:
692
 
                result = packer.pack()
 
583
                packer.pack()
693
584
            except errors.RetryWithNewPacks:
694
585
                # An exception is propagating out of this context, make sure
695
586
                # this packer has cleaned up. Packer() doesn't set its new_pack
698
589
                if packer.new_pack is not None:
699
590
                    packer.new_pack.abort()
700
591
                raise
701
 
            if result is None:
702
 
                return
703
592
            for pack in packs:
704
593
                self._remove_pack_from_memory(pack)
705
594
        # record the newly available packs and stop advertising the old
706
595
        # packs
707
 
        to_be_obsoleted = []
708
 
        for _, packs in pack_operations:
709
 
            to_be_obsoleted.extend(packs)
710
 
        result = self._save_pack_names(clear_obsolete_packs=True,
711
 
                                       obsolete_packs=to_be_obsoleted)
712
 
        return result
 
596
        self._save_pack_names(clear_obsolete_packs=True)
 
597
        # Move the old packs out of the way now they are no longer referenced.
 
598
        for revision_count, packs in pack_operations:
 
599
            self._obsolete_packs(packs)
713
600
 
714
601
 
715
602
class CHKInventoryRepository(KnitPackRepository):
733
620
        self.inventories = GroupCompressVersionedFiles(
734
621
            _GCGraphIndex(self._pack_collection.inventory_index.combined_index,
735
622
                add_callback=self._pack_collection.inventory_index.add_callback,
736
 
                parents=True, is_locked=self.is_locked,
737
 
                inconsistency_fatal=False),
 
623
                parents=True, is_locked=self.is_locked),
738
624
            access=self._pack_collection.inventory_index.data_access)
739
625
        self.revisions = GroupCompressVersionedFiles(
740
626
            _GCGraphIndex(self._pack_collection.revision_index.combined_index,
741
627
                add_callback=self._pack_collection.revision_index.add_callback,
742
628
                parents=True, is_locked=self.is_locked,
743
 
                track_external_parent_refs=True, track_new_keys=True),
 
629
                track_external_parent_refs=True),
744
630
            access=self._pack_collection.revision_index.data_access,
745
631
            delta=False)
746
632
        self.signatures = GroupCompressVersionedFiles(
747
633
            _GCGraphIndex(self._pack_collection.signature_index.combined_index,
748
634
                add_callback=self._pack_collection.signature_index.add_callback,
749
 
                parents=False, is_locked=self.is_locked,
750
 
                inconsistency_fatal=False),
 
635
                parents=False, is_locked=self.is_locked),
751
636
            access=self._pack_collection.signature_index.data_access,
752
637
            delta=False)
753
638
        self.texts = GroupCompressVersionedFiles(
754
639
            _GCGraphIndex(self._pack_collection.text_index.combined_index,
755
640
                add_callback=self._pack_collection.text_index.add_callback,
756
 
                parents=True, is_locked=self.is_locked,
757
 
                inconsistency_fatal=False),
 
641
                parents=True, is_locked=self.is_locked),
758
642
            access=self._pack_collection.text_index.data_access)
759
643
        # No parents, individual CHK pages don't have specific ancestry
760
644
        self.chk_bytes = GroupCompressVersionedFiles(
761
645
            _GCGraphIndex(self._pack_collection.chk_index.combined_index,
762
646
                add_callback=self._pack_collection.chk_index.add_callback,
763
 
                parents=False, is_locked=self.is_locked,
764
 
                inconsistency_fatal=False),
 
647
                parents=False, is_locked=self.is_locked),
765
648
            access=self._pack_collection.chk_index.data_access)
766
 
        search_key_name = self._format._serializer.search_key_name
767
 
        search_key_func = chk_map.search_key_registry.get(search_key_name)
768
 
        self.chk_bytes._search_key_func = search_key_func
769
649
        # True when the repository object is 'write locked' (as opposed to the
770
650
        # physical lock only taken out around changes to the pack-names list.)
771
651
        # Another way to represent this would be a decorator around the control
794
674
        return self._inventory_add_lines(revision_id, parents,
795
675
            inv_lines, check_content=False)
796
676
 
797
 
    def _create_inv_from_null(self, delta, revision_id):
798
 
        """This will mutate new_inv directly.
799
 
 
800
 
        This is a simplified form of create_by_apply_delta which knows that all
801
 
        the old values must be None, so everything is a create.
802
 
        """
803
 
        serializer = self._format._serializer
804
 
        new_inv = inventory.CHKInventory(serializer.search_key_name)
805
 
        new_inv.revision_id = revision_id
806
 
        entry_to_bytes = new_inv._entry_to_bytes
807
 
        id_to_entry_dict = {}
808
 
        parent_id_basename_dict = {}
809
 
        for old_path, new_path, file_id, entry in delta:
810
 
            if old_path is not None:
811
 
                raise ValueError('Invalid delta, somebody tried to delete %r'
812
 
                                 ' from the NULL_REVISION'
813
 
                                 % ((old_path, file_id),))
814
 
            if new_path is None:
815
 
                raise ValueError('Invalid delta, delta from NULL_REVISION has'
816
 
                                 ' no new_path %r' % (file_id,))
817
 
            if new_path == '':
818
 
                new_inv.root_id = file_id
819
 
                parent_id_basename_key = StaticTuple('', '').intern()
820
 
            else:
821
 
                utf8_entry_name = entry.name.encode('utf-8')
822
 
                parent_id_basename_key = StaticTuple(entry.parent_id,
823
 
                                                     utf8_entry_name).intern()
824
 
            new_value = entry_to_bytes(entry)
825
 
            # Populate Caches?
826
 
            # new_inv._path_to_fileid_cache[new_path] = file_id
827
 
            key = StaticTuple(file_id).intern()
828
 
            id_to_entry_dict[key] = new_value
829
 
            parent_id_basename_dict[parent_id_basename_key] = file_id
830
 
 
831
 
        new_inv._populate_from_dicts(self.chk_bytes, id_to_entry_dict,
832
 
            parent_id_basename_dict, maximum_size=serializer.maximum_size)
833
 
        return new_inv
834
 
 
835
677
    def add_inventory_by_delta(self, basis_revision_id, delta, new_revision_id,
836
678
                               parents, basis_inv=None, propagate_caches=False):
837
679
        """Add a new inventory expressed as a delta against another revision.
857
699
            repository format specific) of the serialized inventory, and the
858
700
            resulting inventory.
859
701
        """
 
702
        if basis_revision_id == _mod_revision.NULL_REVISION:
 
703
            return KnitPackRepository.add_inventory_by_delta(self,
 
704
                basis_revision_id, delta, new_revision_id, parents)
860
705
        if not self.is_in_write_group():
861
706
            raise AssertionError("%r not in write group" % (self,))
862
707
        _mod_revision.check_not_reserved_id(new_revision_id)
863
 
        basis_tree = None
864
 
        if basis_inv is None:
865
 
            if basis_revision_id == _mod_revision.NULL_REVISION:
866
 
                new_inv = self._create_inv_from_null(delta, new_revision_id)
867
 
                inv_lines = new_inv.to_lines()
868
 
                return self._inventory_add_lines(new_revision_id, parents,
869
 
                    inv_lines, check_content=False), new_inv
870
 
            else:
871
 
                basis_tree = self.revision_tree(basis_revision_id)
872
 
                basis_tree.lock_read()
 
708
        basis_tree = self.revision_tree(basis_revision_id)
 
709
        basis_tree.lock_read()
 
710
        try:
 
711
            if basis_inv is None:
873
712
                basis_inv = basis_tree.inventory
874
 
        try:
875
713
            result = basis_inv.create_by_apply_delta(delta, new_revision_id,
876
714
                propagate_caches=propagate_caches)
877
715
            inv_lines = result.to_lines()
878
716
            return self._inventory_add_lines(new_revision_id, parents,
879
717
                inv_lines, check_content=False), result
880
718
        finally:
881
 
            if basis_tree is not None:
882
 
                basis_tree.unlock()
883
 
 
884
 
    def _deserialise_inventory(self, revision_id, bytes):
885
 
        return inventory.CHKInventory.deserialise(self.chk_bytes, bytes,
886
 
            (revision_id,))
887
 
 
888
 
    def _iter_inventories(self, revision_ids, ordering):
 
719
            basis_tree.unlock()
 
720
 
 
721
    def _iter_inventories(self, revision_ids):
889
722
        """Iterate over many inventory objects."""
890
 
        if ordering is None:
891
 
            ordering = 'unordered'
892
723
        keys = [(revision_id,) for revision_id in revision_ids]
893
 
        stream = self.inventories.get_record_stream(keys, ordering, True)
 
724
        stream = self.inventories.get_record_stream(keys, 'unordered', True)
894
725
        texts = {}
895
726
        for record in stream:
896
727
            if record.storage_kind != 'absent':
900
731
        for key in keys:
901
732
            yield inventory.CHKInventory.deserialise(self.chk_bytes, texts[key], key)
902
733
 
903
 
    def _iter_inventory_xmls(self, revision_ids, ordering):
904
 
        # Without a native 'xml' inventory, this method doesn't make sense.
905
 
        # However older working trees, and older bundles want it - so we supply
906
 
        # it allowing _get_inventory_xml to work. Bundles currently use the
907
 
        # serializer directly; this also isn't ideal, but there isn't an xml
908
 
        # iteration interface offered at all for repositories. We could make
909
 
        # _iter_inventory_xmls be part of the contract, even if kept private.
910
 
        inv_to_str = self._serializer.write_inventory_to_string
911
 
        for inv in self.iter_inventories(revision_ids, ordering=ordering):
912
 
            yield inv_to_str(inv), inv.revision_id
913
 
 
914
 
    def _find_present_inventory_keys(self, revision_keys):
915
 
        parent_map = self.inventories.get_parent_map(revision_keys)
916
 
        present_inventory_keys = set(k for k in parent_map)
917
 
        return present_inventory_keys
 
734
    def _iter_inventory_xmls(self, revision_ids):
 
735
        # Without a native 'xml' inventory, this method doesn't make sense, so
 
736
        # make it raise to trap naughty direct users.
 
737
        raise NotImplementedError(self._iter_inventory_xmls)
 
738
 
 
739
    def _find_parent_ids_of_revisions(self, revision_ids):
 
740
        # TODO: we probably want to make this a helper that other code can get
 
741
        #       at
 
742
        parent_map = self.get_parent_map(revision_ids)
 
743
        parents = set()
 
744
        map(parents.update, parent_map.itervalues())
 
745
        parents.difference_update(revision_ids)
 
746
        parents.discard(_mod_revision.NULL_REVISION)
 
747
        return parents
 
748
 
 
749
    def _find_present_inventory_ids(self, revision_ids):
 
750
        keys = [(r,) for r in revision_ids]
 
751
        parent_map = self.inventories.get_parent_map(keys)
 
752
        present_inventory_ids = set(k[-1] for k in parent_map)
 
753
        return present_inventory_ids
918
754
 
919
755
    def fileids_altered_by_revision_ids(self, revision_ids, _inv_weave=None):
920
756
        """Find the file ids and versions affected by revisions.
931
767
        file_id_revisions = {}
932
768
        pb = ui.ui_factory.nested_progress_bar()
933
769
        try:
934
 
            revision_keys = [(r,) for r in revision_ids]
935
 
            parent_keys = self._find_parent_keys_of_revisions(revision_keys)
936
 
            # TODO: instead of using _find_present_inventory_keys, change the
937
 
            #       code paths to allow missing inventories to be tolerated.
938
 
            #       However, we only want to tolerate missing parent
939
 
            #       inventories, not missing inventories for revision_ids
940
 
            present_parent_inv_keys = self._find_present_inventory_keys(
941
 
                                        parent_keys)
942
 
            present_parent_inv_ids = set(
943
 
                [k[-1] for k in present_parent_inv_keys])
944
 
            inventories_to_read = set(revision_ids)
945
 
            inventories_to_read.update(present_parent_inv_ids)
946
 
            root_key_info = _build_interesting_key_sets(
947
 
                self, inventories_to_read, present_parent_inv_ids)
948
 
            interesting_root_keys = root_key_info.interesting_root_keys
949
 
            uninteresting_root_keys = root_key_info.uninteresting_root_keys
 
770
            parent_ids = self._find_parent_ids_of_revisions(revision_ids)
 
771
            present_parent_inv_ids = self._find_present_inventory_ids(parent_ids)
 
772
            uninteresting_root_keys = set()
 
773
            interesting_root_keys = set()
 
774
            inventories_to_read = set(present_parent_inv_ids)
 
775
            inventories_to_read.update(revision_ids)
 
776
            for inv in self.iter_inventories(inventories_to_read):
 
777
                entry_chk_root_key = inv.id_to_entry.key()
 
778
                if inv.revision_id in present_parent_inv_ids:
 
779
                    uninteresting_root_keys.add(entry_chk_root_key)
 
780
                else:
 
781
                    interesting_root_keys.add(entry_chk_root_key)
 
782
 
950
783
            chk_bytes = self.chk_bytes
951
784
            for record, items in chk_map.iter_interesting_nodes(chk_bytes,
952
785
                        interesting_root_keys, uninteresting_root_keys,
953
786
                        pb=pb):
954
787
                for name, bytes in items:
955
788
                    (name_utf8, file_id, revision_id) = bytes_to_info(bytes)
956
 
                    # TODO: consider interning file_id, revision_id here, or
957
 
                    #       pushing that intern() into bytes_to_info()
958
 
                    # TODO: rich_root should always be True here, for all
959
 
                    #       repositories that support chk_bytes
960
789
                    if not rich_root and name_utf8 == '':
961
790
                        continue
962
791
                    try:
1005
834
 
1006
835
    def _get_source(self, to_format):
1007
836
        """Return a source for streaming from this repository."""
1008
 
        if self._format._serializer == to_format._serializer:
 
837
        if isinstance(to_format, remote.RemoteRepositoryFormat):
 
838
            # Can't just check attributes on to_format with the current code,
 
839
            # work around this:
 
840
            to_format._ensure_real()
 
841
            to_format = to_format._custom_format
 
842
        if to_format.__class__ is self._format.__class__:
1009
843
            # We must be exactly the same format, otherwise stuff like the chk
1010
 
            # page layout might be different.
1011
 
            # Actually, this test is just slightly looser than exact so that
1012
 
            # CHK2 <-> 2a transfers will work.
 
844
            # page layout might be different
1013
845
            return GroupCHKStreamSource(self, to_format)
1014
846
        return super(CHKInventoryRepository, self)._get_source(to_format)
1015
847
 
1016
848
 
1017
 
class GroupCHKStreamSource(KnitPackStreamSource):
 
849
class GroupCHKStreamSource(repository.StreamSource):
1018
850
    """Used when both the source and target repo are GroupCHK repos."""
1019
851
 
1020
852
    def __init__(self, from_repository, to_format):
1022
854
        super(GroupCHKStreamSource, self).__init__(from_repository, to_format)
1023
855
        self._revision_keys = None
1024
856
        self._text_keys = None
1025
 
        self._text_fetch_order = 'groupcompress'
1026
857
        self._chk_id_roots = None
1027
858
        self._chk_p_id_roots = None
1028
859
 
1067
898
            p_id_roots_set.clear()
1068
899
        return ('inventories', _filtered_inv_stream())
1069
900
 
1070
 
    def _get_filtered_chk_streams(self, excluded_revision_keys):
 
901
    def _find_present_inventories(self, revision_ids):
 
902
        revision_keys = [(r,) for r in revision_ids]
 
903
        inventories = self.from_repository.inventories
 
904
        present_inventories = inventories.get_parent_map(revision_keys)
 
905
        return [p[-1] for p in present_inventories]
 
906
 
 
907
    def _get_filtered_chk_streams(self, excluded_revision_ids):
1071
908
        self._text_keys = set()
1072
 
        excluded_revision_keys.discard(_mod_revision.NULL_REVISION)
1073
 
        if not excluded_revision_keys:
 
909
        excluded_revision_ids.discard(_mod_revision.NULL_REVISION)
 
910
        if not excluded_revision_ids:
1074
911
            uninteresting_root_keys = set()
1075
912
            uninteresting_pid_root_keys = set()
1076
913
        else:
1078
915
            # actually present
1079
916
            # TODO: Update Repository.iter_inventories() to add
1080
917
            #       ignore_missing=True
1081
 
            present_keys = self.from_repository._find_present_inventory_keys(
1082
 
                            excluded_revision_keys)
1083
 
            present_ids = [k[-1] for k in present_keys]
 
918
            present_ids = self.from_repository._find_present_inventory_ids(
 
919
                            excluded_revision_ids)
 
920
            present_ids = self._find_present_inventories(excluded_revision_ids)
1084
921
            uninteresting_root_keys = set()
1085
922
            uninteresting_pid_root_keys = set()
1086
923
            for inv in self.from_repository.iter_inventories(present_ids):
1090
927
        bytes_to_info = inventory.CHKInventory._bytes_to_utf8name_key
1091
928
        chk_bytes = self.from_repository.chk_bytes
1092
929
        def _filter_id_to_entry():
1093
 
            interesting_nodes = chk_map.iter_interesting_nodes(chk_bytes,
1094
 
                        self._chk_id_roots, uninteresting_root_keys)
1095
 
            for record in _filter_text_keys(interesting_nodes, self._text_keys,
1096
 
                    bytes_to_info):
 
930
            for record, items in chk_map.iter_interesting_nodes(chk_bytes,
 
931
                        self._chk_id_roots, uninteresting_root_keys):
 
932
                for name, bytes in items:
 
933
                    # Note: we don't care about name_utf8, because we are always
 
934
                    # rich-root = True
 
935
                    _, file_id, revision_id = bytes_to_info(bytes)
 
936
                    self._text_keys.add((file_id, revision_id))
1097
937
                if record is not None:
1098
938
                    yield record
1099
939
            # Consumed
1108
948
            self._chk_p_id_roots = None
1109
949
        yield 'chk_bytes', _get_parent_id_basename_to_file_id_pages()
1110
950
 
 
951
    def _get_text_stream(self):
 
952
        # Note: We know we don't have to handle adding root keys, because both
 
953
        # the source and target are GCCHK, and those always support rich-roots
 
954
        # We may want to request as 'unordered', in case the source has done a
 
955
        # 'split' packing
 
956
        return ('texts', self.from_repository.texts.get_record_stream(
 
957
                            self._text_keys, 'groupcompress', False))
 
958
 
1111
959
    def get_stream(self, search):
1112
960
        revision_ids = search.get_keys()
1113
961
        for stream_info in self._fetch_revision_texts(revision_ids):
1114
962
            yield stream_info
1115
963
        self._revision_keys = [(rev_id,) for rev_id in revision_ids]
1116
 
        self.from_repository.revisions.clear_cache()
1117
 
        self.from_repository.signatures.clear_cache()
1118
964
        yield self._get_inventory_stream(self._revision_keys)
1119
 
        self.from_repository.inventories.clear_cache()
1120
965
        # TODO: The keys to exclude might be part of the search recipe
1121
966
        # For now, exclude all parents that are at the edge of ancestry, for
1122
967
        # which we have inventories
1123
968
        from_repo = self.from_repository
1124
 
        parent_keys = from_repo._find_parent_keys_of_revisions(
1125
 
                        self._revision_keys)
1126
 
        for stream_info in self._get_filtered_chk_streams(parent_keys):
 
969
        parent_ids = from_repo._find_parent_ids_of_revisions(revision_ids)
 
970
        for stream_info in self._get_filtered_chk_streams(parent_ids):
1127
971
            yield stream_info
1128
 
        self.from_repository.chk_bytes.clear_cache()
1129
972
        yield self._get_text_stream()
1130
 
        self.from_repository.texts.clear_cache()
1131
973
 
1132
974
    def get_stream_for_missing_keys(self, missing_keys):
1133
975
        # missing keys can only occur when we are byte copying and not
1142
984
            missing_inventory_keys.add(key[1:])
1143
985
        if self._chk_id_roots or self._chk_p_id_roots:
1144
986
            raise AssertionError('Cannot call get_stream_for_missing_keys'
1145
 
                ' until all of get_stream() has been consumed.')
 
987
                ' untill all of get_stream() has been consumed.')
1146
988
        # Yield the inventory stream, so we can find the chk stream
1147
989
        # Some of the missing_keys will be missing because they are ghosts.
1148
990
        # As such, we can ignore them. The Sink is required to verify there are
1149
991
        # no unavailable texts when the ghost inventories are not filled in.
1150
992
        yield self._get_inventory_stream(missing_inventory_keys,
1151
993
                                         allow_absent=True)
1152
 
        # We use the empty set for excluded_revision_keys, to make it clear
1153
 
        # that we want to transmit all referenced chk pages.
 
994
        # We use the empty set for excluded_revision_ids, to make it clear that
 
995
        # we want to transmit all referenced chk pages.
1154
996
        for stream_info in self._get_filtered_chk_streams(set()):
1155
997
            yield stream_info
1156
998
 
1157
999
 
1158
 
class _InterestingKeyInfo(object):
1159
 
    def __init__(self):
1160
 
        self.interesting_root_keys = set()
1161
 
        self.interesting_pid_root_keys = set()
1162
 
        self.uninteresting_root_keys = set()
1163
 
        self.uninteresting_pid_root_keys = set()
1164
 
 
1165
 
    def all_interesting(self):
1166
 
        return self.interesting_root_keys.union(self.interesting_pid_root_keys)
1167
 
 
1168
 
    def all_uninteresting(self):
1169
 
        return self.uninteresting_root_keys.union(
1170
 
            self.uninteresting_pid_root_keys)
1171
 
 
1172
 
    def all_keys(self):
1173
 
        return self.all_interesting().union(self.all_uninteresting())
1174
 
 
1175
 
 
1176
 
def _build_interesting_key_sets(repo, inventory_ids, parent_only_inv_ids):
1177
 
    result = _InterestingKeyInfo()
1178
 
    for inv in repo.iter_inventories(inventory_ids, 'unordered'):
1179
 
        root_key = inv.id_to_entry.key()
1180
 
        pid_root_key = inv.parent_id_basename_to_file_id.key()
1181
 
        if inv.revision_id in parent_only_inv_ids:
1182
 
            result.uninteresting_root_keys.add(root_key)
1183
 
            result.uninteresting_pid_root_keys.add(pid_root_key)
1184
 
        else:
1185
 
            result.interesting_root_keys.add(root_key)
1186
 
            result.interesting_pid_root_keys.add(pid_root_key)
1187
 
    return result
1188
 
 
1189
 
 
1190
 
def _filter_text_keys(interesting_nodes_iterable, text_keys, bytes_to_info):
1191
 
    """Iterate the result of iter_interesting_nodes, yielding the records
1192
 
    and adding to text_keys.
1193
 
    """
1194
 
    for record, items in interesting_nodes_iterable:
1195
 
        for name, bytes in items:
1196
 
            # Note: we don't care about name_utf8, because groupcompress repos
1197
 
            # are always rich-root, so there are no synthesised root records to
1198
 
            # ignore.
1199
 
            _, file_id, revision_id = bytes_to_info(bytes)
1200
 
            file_id = intern(file_id)
1201
 
            revision_id = intern(revision_id)
1202
 
            text_keys.add(StaticTuple(file_id, revision_id).intern())
1203
 
        yield record
1204
 
 
1205
 
 
1206
 
 
1207
 
 
1208
1000
class RepositoryFormatCHK1(RepositoryFormatPack):
1209
1001
    """A hashed CHK+group compress pack repository."""
1210
1002
 
1229
1021
    _fetch_order = 'unordered'
1230
1022
    _fetch_uses_deltas = False # essentially ignored by the groupcompress code.
1231
1023
    fast_deltas = True
1232
 
    pack_compresses = True
1233
1024
 
1234
1025
    def _get_matching_bzrdir(self):
1235
1026
        return bzrdir.format_registry.make_bzrdir('development6-rich-root')
1249
1040
        return ("Development repository format - rich roots, group compression"
1250
1041
            " and chk inventories")
1251
1042
 
 
1043
    def check_conversion_target(self, target_format):
 
1044
        if not target_format.rich_root_data:
 
1045
            raise errors.BadConversionTarget(
 
1046
                'Does not support rich root data.', target_format)
 
1047
        if not getattr(target_format, 'supports_tree_reference', False):
 
1048
            raise errors.BadConversionTarget(
 
1049
                'Does not support nested trees', target_format)
 
1050
 
 
1051
 
1252
1052
 
1253
1053
class RepositoryFormatCHK2(RepositoryFormatCHK1):
1254
1054
    """A CHK repository that uses the bencode revision serializer."""
1266
1066
    def get_format_string(self):
1267
1067
        """See RepositoryFormat.get_format_string()."""
1268
1068
        return ('Bazaar development format - chk repository with bencode '
1269
 
                'revision serialization (needs bzr.dev from 1.16)\n')
1270
 
 
1271
 
 
1272
 
class RepositoryFormat2a(RepositoryFormatCHK2):
1273
 
    """A CHK repository that uses the bencode revision serializer.
1274
 
 
1275
 
    This is the same as RepositoryFormatCHK2 but with a public name.
1276
 
    """
1277
 
 
1278
 
    _serializer = chk_serializer.chk_bencode_serializer
1279
 
 
1280
 
    def _get_matching_bzrdir(self):
1281
 
        return bzrdir.format_registry.make_bzrdir('2a')
1282
 
 
1283
 
    def _ignore_setting_bzrdir(self, format):
1284
 
        pass
1285
 
 
1286
 
    _matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
1287
 
 
1288
 
    def get_format_string(self):
1289
 
        return ('Bazaar repository format 2a (needs bzr 1.16 or later)\n')
1290
 
 
1291
 
    def get_format_description(self):
1292
 
        """See RepositoryFormat.get_format_description()."""
1293
 
        return ("Repository format 2a - rich roots, group compression"
1294
 
            " and chk inventories")
 
1069
                'revision serialization (needs bzr.dev from 1.15)\n')
 
1070
 
 
1071