~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/btree_index.py

  • Committer: Aaron Bentley
  • Date: 2006-06-21 14:30:57 UTC
  • mfrom: (1801.1.1 bzr.dev)
  • mto: This revision was merged to the branch mainline in revision 1803.
  • Revision ID: abentley@panoramicfeedback.com-20060621143057-776e4b8d707e430e
Install benchmarks. (Jelmer Vernooij)

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2008 Canonical Ltd
2
 
#
3
 
# This program is free software; you can redistribute it and/or modify
4
 
# it under the terms of the GNU General Public License as published by
5
 
# the Free Software Foundation; either version 2 of the License, or
6
 
# (at your option) any later version.
7
 
#
8
 
# This program is distributed in the hope that it will be useful,
9
 
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
 
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
11
 
# GNU General Public License for more details.
12
 
#
13
 
# You should have received a copy of the GNU General Public License
14
 
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
 
#
17
 
 
18
 
"""B+Tree indices"""
19
 
 
20
 
from bisect import bisect_right
21
 
import math
22
 
import tempfile
23
 
import zlib
24
 
 
25
 
from bzrlib import (
26
 
    chunk_writer,
27
 
    debug,
28
 
    errors,
29
 
    fifo_cache,
30
 
    index,
31
 
    lru_cache,
32
 
    osutils,
33
 
    trace,
34
 
    )
35
 
from bzrlib.index import _OPTION_NODE_REFS, _OPTION_KEY_ELEMENTS, _OPTION_LEN
36
 
from bzrlib.transport import get_transport
37
 
 
38
 
 
39
 
_BTSIGNATURE = "B+Tree Graph Index 2\n"
40
 
_OPTION_ROW_LENGTHS = "row_lengths="
41
 
_LEAF_FLAG = "type=leaf\n"
42
 
_INTERNAL_FLAG = "type=internal\n"
43
 
_INTERNAL_OFFSET = "offset="
44
 
 
45
 
_RESERVED_HEADER_BYTES = 120
46
 
_PAGE_SIZE = 4096
47
 
 
48
 
# 4K per page: 4MB - 1000 entries
49
 
_NODE_CACHE_SIZE = 1000
50
 
 
51
 
 
52
 
class _BuilderRow(object):
53
 
    """The stored state accumulated while writing out a row in the index.
54
 
 
55
 
    :ivar spool: A temporary file used to accumulate nodes for this row
56
 
        in the tree.
57
 
    :ivar nodes: The count of nodes emitted so far.
58
 
    """
59
 
 
60
 
    def __init__(self):
61
 
        """Create a _BuilderRow."""
62
 
        self.nodes = 0
63
 
        self.spool = tempfile.TemporaryFile()
64
 
        self.writer = None
65
 
 
66
 
    def finish_node(self, pad=True):
67
 
        byte_lines, _, padding = self.writer.finish()
68
 
        if self.nodes == 0:
69
 
            # padded note:
70
 
            self.spool.write("\x00" * _RESERVED_HEADER_BYTES)
71
 
        skipped_bytes = 0
72
 
        if not pad and padding:
73
 
            del byte_lines[-1]
74
 
            skipped_bytes = padding
75
 
        self.spool.writelines(byte_lines)
76
 
        remainder = (self.spool.tell() + skipped_bytes) % _PAGE_SIZE
77
 
        if remainder != 0:
78
 
            raise AssertionError("incorrect node length: %d, %d"
79
 
                                 % (self.spool.tell(), remainder))
80
 
        self.nodes += 1
81
 
        self.writer = None
82
 
 
83
 
 
84
 
class _InternalBuilderRow(_BuilderRow):
85
 
    """The stored state accumulated while writing out internal rows."""
86
 
 
87
 
    def finish_node(self, pad=True):
88
 
        if not pad:
89
 
            raise AssertionError("Must pad internal nodes only.")
90
 
        _BuilderRow.finish_node(self)
91
 
 
92
 
 
93
 
class _LeafBuilderRow(_BuilderRow):
94
 
    """The stored state accumulated while writing out a leaf rows."""
95
 
 
96
 
 
97
 
class BTreeBuilder(index.GraphIndexBuilder):
98
 
    """A Builder for B+Tree based Graph indices.
99
 
 
100
 
    The resulting graph has the structure:
101
 
 
102
 
    _SIGNATURE OPTIONS NODES
103
 
    _SIGNATURE     := 'B+Tree Graph Index 1' NEWLINE
104
 
    OPTIONS        := REF_LISTS KEY_ELEMENTS LENGTH
105
 
    REF_LISTS      := 'node_ref_lists=' DIGITS NEWLINE
106
 
    KEY_ELEMENTS   := 'key_elements=' DIGITS NEWLINE
107
 
    LENGTH         := 'len=' DIGITS NEWLINE
108
 
    ROW_LENGTHS    := 'row_lengths' DIGITS (COMMA DIGITS)*
109
 
    NODES          := NODE_COMPRESSED*
110
 
    NODE_COMPRESSED:= COMPRESSED_BYTES{4096}
111
 
    NODE_RAW       := INTERNAL | LEAF
112
 
    INTERNAL       := INTERNAL_FLAG POINTERS
113
 
    LEAF           := LEAF_FLAG ROWS
114
 
    KEY_ELEMENT    := Not-whitespace-utf8
115
 
    KEY            := KEY_ELEMENT (NULL KEY_ELEMENT)*
116
 
    ROWS           := ROW*
117
 
    ROW            := KEY NULL ABSENT? NULL REFERENCES NULL VALUE NEWLINE
118
 
    ABSENT         := 'a'
119
 
    REFERENCES     := REFERENCE_LIST (TAB REFERENCE_LIST){node_ref_lists - 1}
120
 
    REFERENCE_LIST := (REFERENCE (CR REFERENCE)*)?
121
 
    REFERENCE      := KEY
122
 
    VALUE          := no-newline-no-null-bytes
123
 
    """
124
 
 
125
 
    def __init__(self, reference_lists=0, key_elements=1, spill_at=100000):
126
 
        """See GraphIndexBuilder.__init__.
127
 
 
128
 
        :param spill_at: Optional parameter controlling the maximum number
129
 
            of nodes that BTreeBuilder will hold in memory.
130
 
        """
131
 
        index.GraphIndexBuilder.__init__(self, reference_lists=reference_lists,
132
 
            key_elements=key_elements)
133
 
        self._spill_at = spill_at
134
 
        self._backing_indices = []
135
 
        # A map of {key: (node_refs, value)}
136
 
        self._nodes = {}
137
 
        # Indicate it hasn't been built yet
138
 
        self._nodes_by_key = None
139
 
        self._optimize_for_size = False
140
 
 
141
 
    def add_node(self, key, value, references=()):
142
 
        """Add a node to the index.
143
 
 
144
 
        If adding the node causes the builder to reach its spill_at threshold,
145
 
        disk spilling will be triggered.
146
 
 
147
 
        :param key: The key. keys are non-empty tuples containing
148
 
            as many whitespace-free utf8 bytestrings as the key length
149
 
            defined for this index.
150
 
        :param references: An iterable of iterables of keys. Each is a
151
 
            reference to another key.
152
 
        :param value: The value to associate with the key. It may be any
153
 
            bytes as long as it does not contain \0 or \n.
154
 
        """
155
 
        # we don't care about absent_references
156
 
        node_refs, _ = self._check_key_ref_value(key, references, value)
157
 
        if key in self._nodes:
158
 
            raise errors.BadIndexDuplicateKey(key, self)
159
 
        self._nodes[key] = (node_refs, value)
160
 
        self._keys.add(key)
161
 
        if self._nodes_by_key is not None and self._key_length > 1:
162
 
            self._update_nodes_by_key(key, value, node_refs)
163
 
        if len(self._keys) < self._spill_at:
164
 
            return
165
 
        self._spill_mem_keys_to_disk()
166
 
 
167
 
    def _spill_mem_keys_to_disk(self):
168
 
        """Write the in memory keys down to disk to cap memory consumption.
169
 
 
170
 
        If we already have some keys written to disk, we will combine them so
171
 
        as to preserve the sorted order.  The algorithm for combining uses
172
 
        powers of two.  So on the first spill, write all mem nodes into a
173
 
        single index. On the second spill, combine the mem nodes with the nodes
174
 
        on disk to create a 2x sized disk index and get rid of the first index.
175
 
        On the third spill, create a single new disk index, which will contain
176
 
        the mem nodes, and preserve the existing 2x sized index.  On the fourth,
177
 
        combine mem with the first and second indexes, creating a new one of
178
 
        size 4x. On the fifth create a single new one, etc.
179
 
        """
180
 
        if self._combine_backing_indices:
181
 
            (new_backing_file, size,
182
 
             backing_pos) = self._spill_mem_keys_and_combine()
183
 
        else:
184
 
            new_backing_file, size = self._spill_mem_keys_without_combining()
185
 
        dir_path, base_name = osutils.split(new_backing_file.name)
186
 
        # Note: The transport here isn't strictly needed, because we will use
187
 
        #       direct access to the new_backing._file object
188
 
        new_backing = BTreeGraphIndex(get_transport(dir_path),
189
 
                                      base_name, size)
190
 
        # GC will clean up the file
191
 
        new_backing._file = new_backing_file
192
 
        if self._combine_backing_indices:
193
 
            if len(self._backing_indices) == backing_pos:
194
 
                self._backing_indices.append(None)
195
 
            self._backing_indices[backing_pos] = new_backing
196
 
            for backing_pos in range(backing_pos):
197
 
                self._backing_indices[backing_pos] = None
198
 
        else:
199
 
            self._backing_indices.append(new_backing)
200
 
        self._keys = set()
201
 
        self._nodes = {}
202
 
        self._nodes_by_key = None
203
 
 
204
 
    def _spill_mem_keys_without_combining(self):
205
 
        return self._write_nodes(self._iter_mem_nodes(), allow_optimize=False)
206
 
 
207
 
    def _spill_mem_keys_and_combine(self):
208
 
        iterators_to_combine = [self._iter_mem_nodes()]
209
 
        pos = -1
210
 
        for pos, backing in enumerate(self._backing_indices):
211
 
            if backing is None:
212
 
                pos -= 1
213
 
                break
214
 
            iterators_to_combine.append(backing.iter_all_entries())
215
 
        backing_pos = pos + 1
216
 
        new_backing_file, size = \
217
 
            self._write_nodes(self._iter_smallest(iterators_to_combine),
218
 
                              allow_optimize=False)
219
 
        return new_backing_file, size, backing_pos
220
 
 
221
 
    def add_nodes(self, nodes):
222
 
        """Add nodes to the index.
223
 
 
224
 
        :param nodes: An iterable of (key, node_refs, value) entries to add.
225
 
        """
226
 
        if self.reference_lists:
227
 
            for (key, value, node_refs) in nodes:
228
 
                self.add_node(key, value, node_refs)
229
 
        else:
230
 
            for (key, value) in nodes:
231
 
                self.add_node(key, value)
232
 
 
233
 
    def _iter_mem_nodes(self):
234
 
        """Iterate over the nodes held in memory."""
235
 
        nodes = self._nodes
236
 
        if self.reference_lists:
237
 
            for key in sorted(nodes):
238
 
                references, value = nodes[key]
239
 
                yield self, key, value, references
240
 
        else:
241
 
            for key in sorted(nodes):
242
 
                references, value = nodes[key]
243
 
                yield self, key, value
244
 
 
245
 
    def _iter_smallest(self, iterators_to_combine):
246
 
        if len(iterators_to_combine) == 1:
247
 
            for value in iterators_to_combine[0]:
248
 
                yield value
249
 
            return
250
 
        current_values = []
251
 
        for iterator in iterators_to_combine:
252
 
            try:
253
 
                current_values.append(iterator.next())
254
 
            except StopIteration:
255
 
                current_values.append(None)
256
 
        last = None
257
 
        while True:
258
 
            # Decorate candidates with the value to allow 2.4's min to be used.
259
 
            candidates = [(item[1][1], item) for item
260
 
                in enumerate(current_values) if item[1] is not None]
261
 
            if not len(candidates):
262
 
                return
263
 
            selected = min(candidates)
264
 
            # undecorate back to (pos, node)
265
 
            selected = selected[1]
266
 
            if last == selected[1][1]:
267
 
                raise errors.BadIndexDuplicateKey(last, self)
268
 
            last = selected[1][1]
269
 
            # Yield, with self as the index
270
 
            yield (self,) + selected[1][1:]
271
 
            pos = selected[0]
272
 
            try:
273
 
                current_values[pos] = iterators_to_combine[pos].next()
274
 
            except StopIteration:
275
 
                current_values[pos] = None
276
 
 
277
 
    def _add_key(self, string_key, line, rows, allow_optimize=True):
278
 
        """Add a key to the current chunk.
279
 
 
280
 
        :param string_key: The key to add.
281
 
        :param line: The fully serialised key and value.
282
 
        :param allow_optimize: If set to False, prevent setting the optimize
283
 
            flag when writing out. This is used by the _spill_mem_keys_to_disk
284
 
            functionality.
285
 
        """
286
 
        if rows[-1].writer is None:
287
 
            # opening a new leaf chunk;
288
 
            for pos, internal_row in enumerate(rows[:-1]):
289
 
                # flesh out any internal nodes that are needed to
290
 
                # preserve the height of the tree
291
 
                if internal_row.writer is None:
292
 
                    length = _PAGE_SIZE
293
 
                    if internal_row.nodes == 0:
294
 
                        length -= _RESERVED_HEADER_BYTES # padded
295
 
                    if allow_optimize:
296
 
                        optimize_for_size = self._optimize_for_size
297
 
                    else:
298
 
                        optimize_for_size = False
299
 
                    internal_row.writer = chunk_writer.ChunkWriter(length, 0,
300
 
                        optimize_for_size=optimize_for_size)
301
 
                    internal_row.writer.write(_INTERNAL_FLAG)
302
 
                    internal_row.writer.write(_INTERNAL_OFFSET +
303
 
                        str(rows[pos + 1].nodes) + "\n")
304
 
            # add a new leaf
305
 
            length = _PAGE_SIZE
306
 
            if rows[-1].nodes == 0:
307
 
                length -= _RESERVED_HEADER_BYTES # padded
308
 
            rows[-1].writer = chunk_writer.ChunkWriter(length,
309
 
                optimize_for_size=self._optimize_for_size)
310
 
            rows[-1].writer.write(_LEAF_FLAG)
311
 
        if rows[-1].writer.write(line):
312
 
            # this key did not fit in the node:
313
 
            rows[-1].finish_node()
314
 
            key_line = string_key + "\n"
315
 
            new_row = True
316
 
            for row in reversed(rows[:-1]):
317
 
                # Mark the start of the next node in the node above. If it
318
 
                # doesn't fit then propagate upwards until we find one that
319
 
                # it does fit into.
320
 
                if row.writer.write(key_line):
321
 
                    row.finish_node()
322
 
                else:
323
 
                    # We've found a node that can handle the pointer.
324
 
                    new_row = False
325
 
                    break
326
 
            # If we reached the current root without being able to mark the
327
 
            # division point, then we need a new root:
328
 
            if new_row:
329
 
                # We need a new row
330
 
                if 'index' in debug.debug_flags:
331
 
                    trace.mutter('Inserting new global row.')
332
 
                new_row = _InternalBuilderRow()
333
 
                reserved_bytes = 0
334
 
                rows.insert(0, new_row)
335
 
                # This will be padded, hence the -100
336
 
                new_row.writer = chunk_writer.ChunkWriter(
337
 
                    _PAGE_SIZE - _RESERVED_HEADER_BYTES,
338
 
                    reserved_bytes,
339
 
                    optimize_for_size=self._optimize_for_size)
340
 
                new_row.writer.write(_INTERNAL_FLAG)
341
 
                new_row.writer.write(_INTERNAL_OFFSET +
342
 
                    str(rows[1].nodes - 1) + "\n")
343
 
                new_row.writer.write(key_line)
344
 
            self._add_key(string_key, line, rows, allow_optimize=allow_optimize)
345
 
 
346
 
    def _write_nodes(self, node_iterator, allow_optimize=True):
347
 
        """Write node_iterator out as a B+Tree.
348
 
 
349
 
        :param node_iterator: An iterator of sorted nodes. Each node should
350
 
            match the output given by iter_all_entries.
351
 
        :param allow_optimize: If set to False, prevent setting the optimize
352
 
            flag when writing out. This is used by the _spill_mem_keys_to_disk
353
 
            functionality.
354
 
        :return: A file handle for a temporary file containing a B+Tree for
355
 
            the nodes.
356
 
        """
357
 
        # The index rows - rows[0] is the root, rows[1] is the layer under it
358
 
        # etc.
359
 
        rows = []
360
 
        # forward sorted by key. In future we may consider topological sorting,
361
 
        # at the cost of table scans for direct lookup, or a second index for
362
 
        # direct lookup
363
 
        key_count = 0
364
 
        # A stack with the number of nodes of each size. 0 is the root node
365
 
        # and must always be 1 (if there are any nodes in the tree).
366
 
        self.row_lengths = []
367
 
        # Loop over all nodes adding them to the bottom row
368
 
        # (rows[-1]). When we finish a chunk in a row,
369
 
        # propagate the key that didn't fit (comes after the chunk) to the
370
 
        # row above, transitively.
371
 
        for node in node_iterator:
372
 
            if key_count == 0:
373
 
                # First key triggers the first row
374
 
                rows.append(_LeafBuilderRow())
375
 
            key_count += 1
376
 
            string_key, line = _btree_serializer._flatten_node(node,
377
 
                                    self.reference_lists)
378
 
            self._add_key(string_key, line, rows, allow_optimize=allow_optimize)
379
 
        for row in reversed(rows):
380
 
            pad = (type(row) != _LeafBuilderRow)
381
 
            row.finish_node(pad=pad)
382
 
        result = tempfile.NamedTemporaryFile(prefix='bzr-index-')
383
 
        lines = [_BTSIGNATURE]
384
 
        lines.append(_OPTION_NODE_REFS + str(self.reference_lists) + '\n')
385
 
        lines.append(_OPTION_KEY_ELEMENTS + str(self._key_length) + '\n')
386
 
        lines.append(_OPTION_LEN + str(key_count) + '\n')
387
 
        row_lengths = [row.nodes for row in rows]
388
 
        lines.append(_OPTION_ROW_LENGTHS + ','.join(map(str, row_lengths)) + '\n')
389
 
        result.writelines(lines)
390
 
        position = sum(map(len, lines))
391
 
        root_row = True
392
 
        if position > _RESERVED_HEADER_BYTES:
393
 
            raise AssertionError("Could not fit the header in the"
394
 
                                 " reserved space: %d > %d"
395
 
                                 % (position, _RESERVED_HEADER_BYTES))
396
 
        # write the rows out:
397
 
        for row in rows:
398
 
            reserved = _RESERVED_HEADER_BYTES # reserved space for first node
399
 
            row.spool.flush()
400
 
            row.spool.seek(0)
401
 
            # copy nodes to the finalised file.
402
 
            # Special case the first node as it may be prefixed
403
 
            node = row.spool.read(_PAGE_SIZE)
404
 
            result.write(node[reserved:])
405
 
            result.write("\x00" * (reserved - position))
406
 
            position = 0 # Only the root row actually has an offset
407
 
            copied_len = osutils.pumpfile(row.spool, result)
408
 
            if copied_len != (row.nodes - 1) * _PAGE_SIZE:
409
 
                if type(row) != _LeafBuilderRow:
410
 
                    raise AssertionError("Incorrect amount of data copied"
411
 
                        " expected: %d, got: %d"
412
 
                        % ((row.nodes - 1) * _PAGE_SIZE,
413
 
                           copied_len))
414
 
        result.flush()
415
 
        size = result.tell()
416
 
        result.seek(0)
417
 
        return result, size
418
 
 
419
 
    def finish(self):
420
 
        """Finalise the index.
421
 
 
422
 
        :return: A file handle for a temporary file containing the nodes added
423
 
            to the index.
424
 
        """
425
 
        return self._write_nodes(self.iter_all_entries())[0]
426
 
 
427
 
    def iter_all_entries(self):
428
 
        """Iterate over all keys within the index
429
 
 
430
 
        :return: An iterable of (index, key, value, reference_lists). There is
431
 
            no defined order for the result iteration - it will be in the most
432
 
            efficient order for the index (in this case dictionary hash order).
433
 
        """
434
 
        if 'evil' in debug.debug_flags:
435
 
            trace.mutter_callsite(3,
436
 
                "iter_all_entries scales with size of history.")
437
 
        # Doing serial rather than ordered would be faster; but this shouldn't
438
 
        # be getting called routinely anyway.
439
 
        iterators = [self._iter_mem_nodes()]
440
 
        for backing in self._backing_indices:
441
 
            if backing is not None:
442
 
                iterators.append(backing.iter_all_entries())
443
 
        if len(iterators) == 1:
444
 
            return iterators[0]
445
 
        return self._iter_smallest(iterators)
446
 
 
447
 
    def iter_entries(self, keys):
448
 
        """Iterate over keys within the index.
449
 
 
450
 
        :param keys: An iterable providing the keys to be retrieved.
451
 
        :return: An iterable of (index, key, value, reference_lists). There is no
452
 
            defined order for the result iteration - it will be in the most
453
 
            efficient order for the index (keys iteration order in this case).
454
 
        """
455
 
        keys = set(keys)
456
 
        local_keys = keys.intersection(self._keys)
457
 
        if self.reference_lists:
458
 
            for key in local_keys:
459
 
                node = self._nodes[key]
460
 
                yield self, key, node[1], node[0]
461
 
        else:
462
 
            for key in local_keys:
463
 
                node = self._nodes[key]
464
 
                yield self, key, node[1]
465
 
        # Find things that are in backing indices that have not been handled
466
 
        # yet.
467
 
        if not self._backing_indices:
468
 
            return # We won't find anything there either
469
 
        # Remove all of the keys that we found locally
470
 
        keys.difference_update(local_keys)
471
 
        for backing in self._backing_indices:
472
 
            if backing is None:
473
 
                continue
474
 
            if not keys:
475
 
                return
476
 
            for node in backing.iter_entries(keys):
477
 
                keys.remove(node[1])
478
 
                yield (self,) + node[1:]
479
 
 
480
 
    def iter_entries_prefix(self, keys):
481
 
        """Iterate over keys within the index using prefix matching.
482
 
 
483
 
        Prefix matching is applied within the tuple of a key, not to within
484
 
        the bytestring of each key element. e.g. if you have the keys ('foo',
485
 
        'bar'), ('foobar', 'gam') and do a prefix search for ('foo', None) then
486
 
        only the former key is returned.
487
 
 
488
 
        :param keys: An iterable providing the key prefixes to be retrieved.
489
 
            Each key prefix takes the form of a tuple the length of a key, but
490
 
            with the last N elements 'None' rather than a regular bytestring.
491
 
            The first element cannot be 'None'.
492
 
        :return: An iterable as per iter_all_entries, but restricted to the
493
 
            keys with a matching prefix to those supplied. No additional keys
494
 
            will be returned, and every match that is in the index will be
495
 
            returned.
496
 
        """
497
 
        # XXX: To much duplication with the GraphIndex class; consider finding
498
 
        # a good place to pull out the actual common logic.
499
 
        keys = set(keys)
500
 
        if not keys:
501
 
            return
502
 
        for backing in self._backing_indices:
503
 
            if backing is None:
504
 
                continue
505
 
            for node in backing.iter_entries_prefix(keys):
506
 
                yield (self,) + node[1:]
507
 
        if self._key_length == 1:
508
 
            for key in keys:
509
 
                # sanity check
510
 
                if key[0] is None:
511
 
                    raise errors.BadIndexKey(key)
512
 
                if len(key) != self._key_length:
513
 
                    raise errors.BadIndexKey(key)
514
 
                try:
515
 
                    node = self._nodes[key]
516
 
                except KeyError:
517
 
                    continue
518
 
                if self.reference_lists:
519
 
                    yield self, key, node[1], node[0]
520
 
                else:
521
 
                    yield self, key, node[1]
522
 
            return
523
 
        for key in keys:
524
 
            # sanity check
525
 
            if key[0] is None:
526
 
                raise errors.BadIndexKey(key)
527
 
            if len(key) != self._key_length:
528
 
                raise errors.BadIndexKey(key)
529
 
            # find what it refers to:
530
 
            key_dict = self._get_nodes_by_key()
531
 
            elements = list(key)
532
 
            # find the subdict to return
533
 
            try:
534
 
                while len(elements) and elements[0] is not None:
535
 
                    key_dict = key_dict[elements[0]]
536
 
                    elements.pop(0)
537
 
            except KeyError:
538
 
                # a non-existant lookup.
539
 
                continue
540
 
            if len(elements):
541
 
                dicts = [key_dict]
542
 
                while dicts:
543
 
                    key_dict = dicts.pop(-1)
544
 
                    # can't be empty or would not exist
545
 
                    item, value = key_dict.iteritems().next()
546
 
                    if type(value) == dict:
547
 
                        # push keys
548
 
                        dicts.extend(key_dict.itervalues())
549
 
                    else:
550
 
                        # yield keys
551
 
                        for value in key_dict.itervalues():
552
 
                            yield (self, ) + value
553
 
            else:
554
 
                yield (self, ) + key_dict
555
 
 
556
 
    def _get_nodes_by_key(self):
557
 
        if self._nodes_by_key is None:
558
 
            nodes_by_key = {}
559
 
            if self.reference_lists:
560
 
                for key, (references, value) in self._nodes.iteritems():
561
 
                    key_dict = nodes_by_key
562
 
                    for subkey in key[:-1]:
563
 
                        key_dict = key_dict.setdefault(subkey, {})
564
 
                    key_dict[key[-1]] = key, value, references
565
 
            else:
566
 
                for key, (references, value) in self._nodes.iteritems():
567
 
                    key_dict = nodes_by_key
568
 
                    for subkey in key[:-1]:
569
 
                        key_dict = key_dict.setdefault(subkey, {})
570
 
                    key_dict[key[-1]] = key, value
571
 
            self._nodes_by_key = nodes_by_key
572
 
        return self._nodes_by_key
573
 
 
574
 
    def key_count(self):
575
 
        """Return an estimate of the number of keys in this index.
576
 
 
577
 
        For InMemoryGraphIndex the estimate is exact.
578
 
        """
579
 
        return len(self._keys) + sum(backing.key_count() for backing in
580
 
            self._backing_indices if backing is not None)
581
 
 
582
 
    def validate(self):
583
 
        """In memory index's have no known corruption at the moment."""
584
 
 
585
 
 
586
 
class _LeafNode(object):
587
 
    """A leaf node for a serialised B+Tree index."""
588
 
 
589
 
    __slots__ = ('keys', 'min_key', 'max_key')
590
 
 
591
 
    def __init__(self, bytes, key_length, ref_list_length):
592
 
        """Parse bytes to create a leaf node object."""
593
 
        # splitlines mangles the \r delimiters.. don't use it.
594
 
        key_list = _btree_serializer._parse_leaf_lines(bytes,
595
 
            key_length, ref_list_length)
596
 
        if key_list:
597
 
            self.min_key = key_list[0][0]
598
 
            self.max_key = key_list[-1][0]
599
 
        else:
600
 
            self.min_key = self.max_key = None
601
 
        self.keys = dict(key_list)
602
 
 
603
 
 
604
 
class _InternalNode(object):
605
 
    """An internal node for a serialised B+Tree index."""
606
 
 
607
 
    __slots__ = ('keys', 'offset')
608
 
 
609
 
    def __init__(self, bytes):
610
 
        """Parse bytes to create an internal node object."""
611
 
        # splitlines mangles the \r delimiters.. don't use it.
612
 
        self.keys = self._parse_lines(bytes.split('\n'))
613
 
 
614
 
    def _parse_lines(self, lines):
615
 
        nodes = []
616
 
        self.offset = int(lines[1][7:])
617
 
        for line in lines[2:]:
618
 
            if line == '':
619
 
                break
620
 
            nodes.append(tuple(map(intern, line.split('\0'))))
621
 
        return nodes
622
 
 
623
 
 
624
 
class BTreeGraphIndex(object):
625
 
    """Access to nodes via the standard GraphIndex interface for B+Tree's.
626
 
 
627
 
    Individual nodes are held in a LRU cache. This holds the root node in
628
 
    memory except when very large walks are done.
629
 
    """
630
 
 
631
 
    def __init__(self, transport, name, size):
632
 
        """Create a B+Tree index object on the index name.
633
 
 
634
 
        :param transport: The transport to read data for the index from.
635
 
        :param name: The file name of the index on transport.
636
 
        :param size: Optional size of the index in bytes. This allows
637
 
            compatibility with the GraphIndex API, as well as ensuring that
638
 
            the initial read (to read the root node header) can be done
639
 
            without over-reading even on empty indices, and on small indices
640
 
            allows single-IO to read the entire index.
641
 
        """
642
 
        self._transport = transport
643
 
        self._name = name
644
 
        self._size = size
645
 
        self._file = None
646
 
        self._recommended_pages = self._compute_recommended_pages()
647
 
        self._root_node = None
648
 
        # Default max size is 100,000 leave values
649
 
        self._leaf_value_cache = None # lru_cache.LRUCache(100*1000)
650
 
        self._leaf_node_cache = lru_cache.LRUCache(_NODE_CACHE_SIZE)
651
 
        # We could limit this, but even a 300k record btree has only 3k leaf
652
 
        # nodes, and only 20 internal nodes. So the default of 100 nodes in an
653
 
        # LRU would mean we always cache everything anyway, no need to pay the
654
 
        # overhead of LRU
655
 
        self._internal_node_cache = fifo_cache.FIFOCache(100)
656
 
        self._key_count = None
657
 
        self._row_lengths = None
658
 
        self._row_offsets = None # Start of each row, [-1] is the end
659
 
 
660
 
    def __eq__(self, other):
661
 
        """Equal when self and other were created with the same parameters."""
662
 
        return (
663
 
            type(self) == type(other) and
664
 
            self._transport == other._transport and
665
 
            self._name == other._name and
666
 
            self._size == other._size)
667
 
 
668
 
    def __ne__(self, other):
669
 
        return not self.__eq__(other)
670
 
 
671
 
    def _get_and_cache_nodes(self, nodes):
672
 
        """Read nodes and cache them in the lru.
673
 
 
674
 
        The nodes list supplied is sorted and then read from disk, each node
675
 
        being inserted it into the _node_cache.
676
 
 
677
 
        Note: Asking for more nodes than the _node_cache can contain will
678
 
        result in some of the results being immediately discarded, to prevent
679
 
        this an assertion is raised if more nodes are asked for than are
680
 
        cachable.
681
 
 
682
 
        :return: A dict of {node_pos: node}
683
 
        """
684
 
        found = {}
685
 
        start_of_leaves = None
686
 
        for node_pos, node in self._read_nodes(sorted(nodes)):
687
 
            if node_pos == 0: # Special case
688
 
                self._root_node = node
689
 
            else:
690
 
                if start_of_leaves is None:
691
 
                    start_of_leaves = self._row_offsets[-2]
692
 
                if node_pos < start_of_leaves:
693
 
                    self._internal_node_cache.add(node_pos, node)
694
 
                else:
695
 
                    self._leaf_node_cache.add(node_pos, node)
696
 
            found[node_pos] = node
697
 
        return found
698
 
 
699
 
    def _compute_recommended_pages(self):
700
 
        """Convert transport's recommended_page_size into btree pages.
701
 
 
702
 
        recommended_page_size is in bytes, we want to know how many _PAGE_SIZE
703
 
        pages fit in that length.
704
 
        """
705
 
        recommended_read = self._transport.recommended_page_size()
706
 
        recommended_pages = int(math.ceil(recommended_read /
707
 
                                          float(_PAGE_SIZE)))
708
 
        return recommended_pages
709
 
 
710
 
    def _compute_total_pages_in_index(self):
711
 
        """How many pages are in the index.
712
 
 
713
 
        If we have read the header we will use the value stored there.
714
 
        Otherwise it will be computed based on the length of the index.
715
 
        """
716
 
        if self._size is None:
717
 
            raise AssertionError('_compute_total_pages_in_index should not be'
718
 
                                 ' called when self._size is None')
719
 
        if self._root_node is not None:
720
 
            # This is the number of pages as defined by the header
721
 
            return self._row_offsets[-1]
722
 
        # This is the number of pages as defined by the size of the index. They
723
 
        # should be indentical.
724
 
        total_pages = int(math.ceil(self._size / float(_PAGE_SIZE)))
725
 
        return total_pages
726
 
 
727
 
    def _expand_offsets(self, offsets):
728
 
        """Find extra pages to download.
729
 
 
730
 
        The idea is that we always want to make big-enough requests (like 64kB
731
 
        for http), so that we don't waste round trips. So given the entries
732
 
        that we already have cached and the new pages being downloaded figure
733
 
        out what other pages we might want to read.
734
 
 
735
 
        See also doc/developers/btree_index_prefetch.txt for more details.
736
 
 
737
 
        :param offsets: The offsets to be read
738
 
        :return: A list of offsets to download
739
 
        """
740
 
        if 'index' in debug.debug_flags:
741
 
            trace.mutter('expanding: %s\toffsets: %s', self._name, offsets)
742
 
 
743
 
        if len(offsets) >= self._recommended_pages:
744
 
            # Don't add more, we are already requesting more than enough
745
 
            if 'index' in debug.debug_flags:
746
 
                trace.mutter('  not expanding large request (%s >= %s)',
747
 
                             len(offsets), self._recommended_pages)
748
 
            return offsets
749
 
        if self._size is None:
750
 
            # Don't try anything, because we don't know where the file ends
751
 
            if 'index' in debug.debug_flags:
752
 
                trace.mutter('  not expanding without knowing index size')
753
 
            return offsets
754
 
        total_pages = self._compute_total_pages_in_index()
755
 
        cached_offsets = self._get_offsets_to_cached_pages()
756
 
        # If reading recommended_pages would read the rest of the index, just
757
 
        # do so.
758
 
        if total_pages - len(cached_offsets) <= self._recommended_pages:
759
 
            # Read whatever is left
760
 
            if cached_offsets:
761
 
                expanded = [x for x in xrange(total_pages)
762
 
                               if x not in cached_offsets]
763
 
            else:
764
 
                expanded = range(total_pages)
765
 
            if 'index' in debug.debug_flags:
766
 
                trace.mutter('  reading all unread pages: %s', expanded)
767
 
            return expanded
768
 
 
769
 
        if self._root_node is None:
770
 
            # ATM on the first read of the root node of a large index, we don't
771
 
            # bother pre-reading any other pages. This is because the
772
 
            # likelyhood of actually reading interesting pages is very low.
773
 
            # See doc/developers/btree_index_prefetch.txt for a discussion, and
774
 
            # a possible implementation when we are guessing that the second
775
 
            # layer index is small
776
 
            final_offsets = offsets
777
 
        else:
778
 
            tree_depth = len(self._row_lengths)
779
 
            if len(cached_offsets) < tree_depth and len(offsets) == 1:
780
 
                # We haven't read enough to justify expansion
781
 
                # If we are only going to read the root node, and 1 leaf node,
782
 
                # then it isn't worth expanding our request. Once we've read at
783
 
                # least 2 nodes, then we are probably doing a search, and we
784
 
                # start expanding our requests.
785
 
                if 'index' in debug.debug_flags:
786
 
                    trace.mutter('  not expanding on first reads')
787
 
                return offsets
788
 
            final_offsets = self._expand_to_neighbors(offsets, cached_offsets,
789
 
                                                      total_pages)
790
 
 
791
 
        final_offsets = sorted(final_offsets)
792
 
        if 'index' in debug.debug_flags:
793
 
            trace.mutter('expanded:  %s', final_offsets)
794
 
        return final_offsets
795
 
 
796
 
    def _expand_to_neighbors(self, offsets, cached_offsets, total_pages):
797
 
        """Expand requests to neighbors until we have enough pages.
798
 
 
799
 
        This is called from _expand_offsets after policy has determined that we
800
 
        want to expand.
801
 
        We only want to expand requests within a given layer. We cheat a little
802
 
        bit and assume all requests will be in the same layer. This is true
803
 
        given the current design, but if it changes this algorithm may perform
804
 
        oddly.
805
 
 
806
 
        :param offsets: requested offsets
807
 
        :param cached_offsets: offsets for pages we currently have cached
808
 
        :return: A set() of offsets after expansion
809
 
        """
810
 
        final_offsets = set(offsets)
811
 
        first = end = None
812
 
        new_tips = set(final_offsets)
813
 
        while len(final_offsets) < self._recommended_pages and new_tips:
814
 
            next_tips = set()
815
 
            for pos in new_tips:
816
 
                if first is None:
817
 
                    first, end = self._find_layer_first_and_end(pos)
818
 
                previous = pos - 1
819
 
                if (previous > 0
820
 
                    and previous not in cached_offsets
821
 
                    and previous not in final_offsets
822
 
                    and previous >= first):
823
 
                    next_tips.add(previous)
824
 
                after = pos + 1
825
 
                if (after < total_pages
826
 
                    and after not in cached_offsets
827
 
                    and after not in final_offsets
828
 
                    and after < end):
829
 
                    next_tips.add(after)
830
 
                # This would keep us from going bigger than
831
 
                # recommended_pages by only expanding the first offsets.
832
 
                # However, if we are making a 'wide' request, it is
833
 
                # reasonable to expand all points equally.
834
 
                # if len(final_offsets) > recommended_pages:
835
 
                #     break
836
 
            final_offsets.update(next_tips)
837
 
            new_tips = next_tips
838
 
        return final_offsets
839
 
 
840
 
    def external_references(self, ref_list_num):
841
 
        if self._root_node is None:
842
 
            self._get_root_node()
843
 
        if ref_list_num + 1 > self.node_ref_lists:
844
 
            raise ValueError('No ref list %d, index has %d ref lists'
845
 
                % (ref_list_num, self.node_ref_lists))
846
 
        keys = set()
847
 
        refs = set()
848
 
        for node in self.iter_all_entries():
849
 
            keys.add(node[1])
850
 
            refs.update(node[3][ref_list_num])
851
 
        return refs - keys
852
 
 
853
 
    def _find_layer_first_and_end(self, offset):
854
 
        """Find the start/stop nodes for the layer corresponding to offset.
855
 
 
856
 
        :return: (first, end)
857
 
            first is the first node in this layer
858
 
            end is the first node of the next layer
859
 
        """
860
 
        first = end = 0
861
 
        for roffset in self._row_offsets:
862
 
            first = end
863
 
            end = roffset
864
 
            if offset < roffset:
865
 
                break
866
 
        return first, end
867
 
 
868
 
    def _get_offsets_to_cached_pages(self):
869
 
        """Determine what nodes we already have cached."""
870
 
        cached_offsets = set(self._internal_node_cache.keys())
871
 
        cached_offsets.update(self._leaf_node_cache.keys())
872
 
        if self._root_node is not None:
873
 
            cached_offsets.add(0)
874
 
        return cached_offsets
875
 
 
876
 
    def _get_root_node(self):
877
 
        if self._root_node is None:
878
 
            # We may not have a root node yet
879
 
            self._get_internal_nodes([0])
880
 
        return self._root_node
881
 
 
882
 
    def _get_nodes(self, cache, node_indexes):
883
 
        found = {}
884
 
        needed = []
885
 
        for idx in node_indexes:
886
 
            if idx == 0 and self._root_node is not None:
887
 
                found[0] = self._root_node
888
 
                continue
889
 
            try:
890
 
                found[idx] = cache[idx]
891
 
            except KeyError:
892
 
                needed.append(idx)
893
 
        if not needed:
894
 
            return found
895
 
        needed = self._expand_offsets(needed)
896
 
        found.update(self._get_and_cache_nodes(needed))
897
 
        return found
898
 
 
899
 
    def _get_internal_nodes(self, node_indexes):
900
 
        """Get a node, from cache or disk.
901
 
 
902
 
        After getting it, the node will be cached.
903
 
        """
904
 
        return self._get_nodes(self._internal_node_cache, node_indexes)
905
 
 
906
 
    def _cache_leaf_values(self, nodes):
907
 
        """Cache directly from key => value, skipping the btree."""
908
 
        if self._leaf_value_cache is not None:
909
 
            for node in nodes.itervalues():
910
 
                for key, value in node.keys.iteritems():
911
 
                    if key in self._leaf_value_cache:
912
 
                        # Don't add the rest of the keys, we've seen this node
913
 
                        # before.
914
 
                        break
915
 
                    self._leaf_value_cache[key] = value
916
 
 
917
 
    def _get_leaf_nodes(self, node_indexes):
918
 
        """Get a bunch of nodes, from cache or disk."""
919
 
        found = self._get_nodes(self._leaf_node_cache, node_indexes)
920
 
        self._cache_leaf_values(found)
921
 
        return found
922
 
 
923
 
    def iter_all_entries(self):
924
 
        """Iterate over all keys within the index.
925
 
 
926
 
        :return: An iterable of (index, key, value) or (index, key, value, reference_lists).
927
 
            The former tuple is used when there are no reference lists in the
928
 
            index, making the API compatible with simple key:value index types.
929
 
            There is no defined order for the result iteration - it will be in
930
 
            the most efficient order for the index.
931
 
        """
932
 
        if 'evil' in debug.debug_flags:
933
 
            trace.mutter_callsite(3,
934
 
                "iter_all_entries scales with size of history.")
935
 
        if not self.key_count():
936
 
            return
937
 
        if self._row_offsets[-1] == 1:
938
 
            # There is only the root node, and we read that via key_count()
939
 
            if self.node_ref_lists:
940
 
                for key, (value, refs) in sorted(self._root_node.keys.items()):
941
 
                    yield (self, key, value, refs)
942
 
            else:
943
 
                for key, (value, refs) in sorted(self._root_node.keys.items()):
944
 
                    yield (self, key, value)
945
 
            return
946
 
        start_of_leaves = self._row_offsets[-2]
947
 
        end_of_leaves = self._row_offsets[-1]
948
 
        needed_offsets = range(start_of_leaves, end_of_leaves)
949
 
        if needed_offsets == [0]:
950
 
            # Special case when we only have a root node, as we have already
951
 
            # read everything
952
 
            nodes = [(0, self._root_node)]
953
 
        else:
954
 
            nodes = self._read_nodes(needed_offsets)
955
 
        # We iterate strictly in-order so that we can use this function
956
 
        # for spilling index builds to disk.
957
 
        if self.node_ref_lists:
958
 
            for _, node in nodes:
959
 
                for key, (value, refs) in sorted(node.keys.items()):
960
 
                    yield (self, key, value, refs)
961
 
        else:
962
 
            for _, node in nodes:
963
 
                for key, (value, refs) in sorted(node.keys.items()):
964
 
                    yield (self, key, value)
965
 
 
966
 
    @staticmethod
967
 
    def _multi_bisect_right(in_keys, fixed_keys):
968
 
        """Find the positions where each 'in_key' would fit in fixed_keys.
969
 
 
970
 
        This is equivalent to doing "bisect_right" on each in_key into
971
 
        fixed_keys
972
 
 
973
 
        :param in_keys: A sorted list of keys to match with fixed_keys
974
 
        :param fixed_keys: A sorted list of keys to match against
975
 
        :return: A list of (integer position, [key list]) tuples.
976
 
        """
977
 
        if not in_keys:
978
 
            return []
979
 
        if not fixed_keys:
980
 
            # no pointers in the fixed_keys list, which means everything must
981
 
            # fall to the left.
982
 
            return [(0, in_keys)]
983
 
 
984
 
        # TODO: Iterating both lists will generally take M + N steps
985
 
        #       Bisecting each key will generally take M * log2 N steps.
986
 
        #       If we had an efficient way to compare, we could pick the method
987
 
        #       based on which has the fewer number of steps.
988
 
        #       There is also the argument that bisect_right is a compiled
989
 
        #       function, so there is even more to be gained.
990
 
        # iter_steps = len(in_keys) + len(fixed_keys)
991
 
        # bisect_steps = len(in_keys) * math.log(len(fixed_keys), 2)
992
 
        if len(in_keys) == 1: # Bisect will always be faster for M = 1
993
 
            return [(bisect_right(fixed_keys, in_keys[0]), in_keys)]
994
 
        # elif bisect_steps < iter_steps:
995
 
        #     offsets = {}
996
 
        #     for key in in_keys:
997
 
        #         offsets.setdefault(bisect_right(fixed_keys, key),
998
 
        #                            []).append(key)
999
 
        #     return [(o, offsets[o]) for o in sorted(offsets)]
1000
 
        in_keys_iter = iter(in_keys)
1001
 
        fixed_keys_iter = enumerate(fixed_keys)
1002
 
        cur_in_key = in_keys_iter.next()
1003
 
        cur_fixed_offset, cur_fixed_key = fixed_keys_iter.next()
1004
 
 
1005
 
        class InputDone(Exception): pass
1006
 
        class FixedDone(Exception): pass
1007
 
 
1008
 
        output = []
1009
 
        cur_out = []
1010
 
 
1011
 
        # TODO: Another possibility is that rather than iterating on each side,
1012
 
        #       we could use a combination of bisecting and iterating. For
1013
 
        #       example, while cur_in_key < fixed_key, bisect to find its
1014
 
        #       point, then iterate all matching keys, then bisect (restricted
1015
 
        #       to only the remainder) for the next one, etc.
1016
 
        try:
1017
 
            while True:
1018
 
                if cur_in_key < cur_fixed_key:
1019
 
                    cur_keys = []
1020
 
                    cur_out = (cur_fixed_offset, cur_keys)
1021
 
                    output.append(cur_out)
1022
 
                    while cur_in_key < cur_fixed_key:
1023
 
                        cur_keys.append(cur_in_key)
1024
 
                        try:
1025
 
                            cur_in_key = in_keys_iter.next()
1026
 
                        except StopIteration:
1027
 
                            raise InputDone
1028
 
                    # At this point cur_in_key must be >= cur_fixed_key
1029
 
                # step the cur_fixed_key until we pass the cur key, or walk off
1030
 
                # the end
1031
 
                while cur_in_key >= cur_fixed_key:
1032
 
                    try:
1033
 
                        cur_fixed_offset, cur_fixed_key = fixed_keys_iter.next()
1034
 
                    except StopIteration:
1035
 
                        raise FixedDone
1036
 
        except InputDone:
1037
 
            # We consumed all of the input, nothing more to do
1038
 
            pass
1039
 
        except FixedDone:
1040
 
            # There was some input left, but we consumed all of fixed, so we
1041
 
            # have to add one more for the tail
1042
 
            cur_keys = [cur_in_key]
1043
 
            cur_keys.extend(in_keys_iter)
1044
 
            cur_out = (len(fixed_keys), cur_keys)
1045
 
            output.append(cur_out)
1046
 
        return output
1047
 
 
1048
 
    def _walk_through_internal_nodes(self, keys):
1049
 
        """Take the given set of keys, and find the corresponding LeafNodes.
1050
 
 
1051
 
        :param keys: An unsorted iterable of keys to search for
1052
 
        :return: (nodes, index_and_keys)
1053
 
            nodes is a dict mapping {index: LeafNode}
1054
 
            keys_at_index is a list of tuples of [(index, [keys for Leaf])]
1055
 
        """
1056
 
        # 6 seconds spent in miss_torture using the sorted() line.
1057
 
        # Even with out of order disk IO it seems faster not to sort it when
1058
 
        # large queries are being made.
1059
 
        keys_at_index = [(0, sorted(keys))]
1060
 
 
1061
 
        for row_pos, next_row_start in enumerate(self._row_offsets[1:-1]):
1062
 
            node_indexes = [idx for idx, s_keys in keys_at_index]
1063
 
            nodes = self._get_internal_nodes(node_indexes)
1064
 
 
1065
 
            next_nodes_and_keys = []
1066
 
            for node_index, sub_keys in keys_at_index:
1067
 
                node = nodes[node_index]
1068
 
                positions = self._multi_bisect_right(sub_keys, node.keys)
1069
 
                node_offset = next_row_start + node.offset
1070
 
                next_nodes_and_keys.extend([(node_offset + pos, s_keys)
1071
 
                                           for pos, s_keys in positions])
1072
 
            keys_at_index = next_nodes_and_keys
1073
 
        # We should now be at the _LeafNodes
1074
 
        node_indexes = [idx for idx, s_keys in keys_at_index]
1075
 
 
1076
 
        # TODO: We may *not* want to always read all the nodes in one
1077
 
        #       big go. Consider setting a max size on this.
1078
 
        nodes = self._get_leaf_nodes(node_indexes)
1079
 
        return nodes, keys_at_index
1080
 
 
1081
 
    def iter_entries(self, keys):
1082
 
        """Iterate over keys within the index.
1083
 
 
1084
 
        :param keys: An iterable providing the keys to be retrieved.
1085
 
        :return: An iterable as per iter_all_entries, but restricted to the
1086
 
            keys supplied. No additional keys will be returned, and every
1087
 
            key supplied that is in the index will be returned.
1088
 
        """
1089
 
        # 6 seconds spent in miss_torture using the sorted() line.
1090
 
        # Even with out of order disk IO it seems faster not to sort it when
1091
 
        # large queries are being made.
1092
 
        # However, now that we are doing multi-way bisecting, we need the keys
1093
 
        # in sorted order anyway. We could change the multi-way code to not
1094
 
        # require sorted order. (For example, it bisects for the first node,
1095
 
        # does an in-order search until a key comes before the current point,
1096
 
        # which it then bisects for, etc.)
1097
 
        keys = frozenset(keys)
1098
 
        if not keys:
1099
 
            return
1100
 
 
1101
 
        if not self.key_count():
1102
 
            return
1103
 
 
1104
 
        needed_keys = []
1105
 
        if self._leaf_value_cache is None:
1106
 
            needed_keys = keys
1107
 
        else:
1108
 
            for key in keys:
1109
 
                value = self._leaf_value_cache.get(key, None)
1110
 
                if value is not None:
1111
 
                    # This key is known not to be here, skip it
1112
 
                    value, refs = value
1113
 
                    if self.node_ref_lists:
1114
 
                        yield (self, key, value, refs)
1115
 
                    else:
1116
 
                        yield (self, key, value)
1117
 
                else:
1118
 
                    needed_keys.append(key)
1119
 
 
1120
 
        last_key = None
1121
 
        needed_keys = keys
1122
 
        if not needed_keys:
1123
 
            return
1124
 
        nodes, nodes_and_keys = self._walk_through_internal_nodes(needed_keys)
1125
 
        for node_index, sub_keys in nodes_and_keys:
1126
 
            if not sub_keys:
1127
 
                continue
1128
 
            node = nodes[node_index]
1129
 
            for next_sub_key in sub_keys:
1130
 
                if next_sub_key in node.keys:
1131
 
                    value, refs = node.keys[next_sub_key]
1132
 
                    if self.node_ref_lists:
1133
 
                        yield (self, next_sub_key, value, refs)
1134
 
                    else:
1135
 
                        yield (self, next_sub_key, value)
1136
 
 
1137
 
    def _find_ancestors(self, keys, ref_list_num, parent_map, missing_keys):
1138
 
        """Find the parent_map information for the set of keys.
1139
 
 
1140
 
        This populates the parent_map dict and missing_keys set based on the
1141
 
        queried keys. It also can fill out an arbitrary number of parents that
1142
 
        it finds while searching for the supplied keys.
1143
 
 
1144
 
        It is unlikely that you want to call this directly. See
1145
 
        "CombinedGraphIndex.find_ancestry()" for a more appropriate API.
1146
 
 
1147
 
        :param keys: A keys whose ancestry we want to return
1148
 
            Every key will either end up in 'parent_map' or 'missing_keys'.
1149
 
        :param ref_list_num: This index in the ref_lists is the parents we
1150
 
            care about.
1151
 
        :param parent_map: {key: parent_keys} for keys that are present in this
1152
 
            index. This may contain more entries than were in 'keys', that are
1153
 
            reachable ancestors of the keys requested.
1154
 
        :param missing_keys: keys which are known to be missing in this index.
1155
 
            This may include parents that were not directly requested, but we
1156
 
            were able to determine that they are not present in this index.
1157
 
        :return: search_keys    parents that were found but not queried to know
1158
 
            if they are missing or present. Callers can re-query this index for
1159
 
            those keys, and they will be placed into parent_map or missing_keys
1160
 
        """
1161
 
        if not self.key_count():
1162
 
            # We use key_count() to trigger reading the root node and
1163
 
            # determining info about this BTreeGraphIndex
1164
 
            # If we don't have any keys, then everything is missing
1165
 
            missing_keys.update(keys)
1166
 
            return set()
1167
 
        if ref_list_num >= self.node_ref_lists:
1168
 
            raise ValueError('No ref list %d, index has %d ref lists'
1169
 
                % (ref_list_num, self.node_ref_lists))
1170
 
 
1171
 
        # The main trick we are trying to accomplish is that when we find a
1172
 
        # key listing its parents, we expect that the parent key is also likely
1173
 
        # to sit on the same page. Allowing us to expand parents quickly
1174
 
        # without suffering the full stack of bisecting, etc.
1175
 
        nodes, nodes_and_keys = self._walk_through_internal_nodes(keys)
1176
 
 
1177
 
        # These are parent keys which could not be immediately resolved on the
1178
 
        # page where the child was present. Note that we may already be
1179
 
        # searching for that key, and it may actually be present [or known
1180
 
        # missing] on one of the other pages we are reading.
1181
 
        # TODO:
1182
 
        #   We could try searching for them in the immediate previous or next
1183
 
        #   page. If they occur "later" we could put them in a pending lookup
1184
 
        #   set, and then for each node we read thereafter we could check to
1185
 
        #   see if they are present.
1186
 
        #   However, we don't know the impact of keeping this list of things
1187
 
        #   that I'm going to search for every node I come across from here on
1188
 
        #   out.
1189
 
        #   It doesn't handle the case when the parent key is missing on a
1190
 
        #   page that we *don't* read. So we already have to handle being
1191
 
        #   re-entrant for that.
1192
 
        #   Since most keys contain a date string, they are more likely to be
1193
 
        #   found earlier in the file than later, but we would know that right
1194
 
        #   away (key < min_key), and wouldn't keep searching it on every other
1195
 
        #   page that we read.
1196
 
        #   Mostly, it is an idea, one which should be benchmarked.
1197
 
        parents_not_on_page = set()
1198
 
 
1199
 
        for node_index, sub_keys in nodes_and_keys:
1200
 
            if not sub_keys:
1201
 
                continue
1202
 
            # sub_keys is all of the keys we are looking for that should exist
1203
 
            # on this page, if they aren't here, then they won't be found
1204
 
            node = nodes[node_index]
1205
 
            node_keys = node.keys
1206
 
            parents_to_check = set()
1207
 
            for next_sub_key in sub_keys:
1208
 
                if next_sub_key not in node_keys:
1209
 
                    # This one is just not present in the index at all
1210
 
                    missing_keys.add(next_sub_key)
1211
 
                else:
1212
 
                    value, refs = node_keys[next_sub_key]
1213
 
                    parent_keys = refs[ref_list_num]
1214
 
                    parent_map[next_sub_key] = parent_keys
1215
 
                    parents_to_check.update(parent_keys)
1216
 
            # Don't look for things we've already found
1217
 
            parents_to_check = parents_to_check.difference(parent_map)
1218
 
            # this can be used to test the benefit of having the check loop
1219
 
            # inlined.
1220
 
            # parents_not_on_page.update(parents_to_check)
1221
 
            # continue
1222
 
            while parents_to_check:
1223
 
                next_parents_to_check = set()
1224
 
                for key in parents_to_check:
1225
 
                    if key in node_keys:
1226
 
                        value, refs = node_keys[key]
1227
 
                        parent_keys = refs[ref_list_num]
1228
 
                        parent_map[key] = parent_keys
1229
 
                        next_parents_to_check.update(parent_keys)
1230
 
                    else:
1231
 
                        # This parent either is genuinely missing, or should be
1232
 
                        # found on another page. Perf test whether it is better
1233
 
                        # to check if this node should fit on this page or not.
1234
 
                        # in the 'everything-in-one-pack' scenario, this *not*
1235
 
                        # doing the check is 237ms vs 243ms.
1236
 
                        # So slightly better, but I assume the standard 'lots
1237
 
                        # of packs' is going to show a reasonable improvement
1238
 
                        # from the check, because it avoids 'going around
1239
 
                        # again' for everything that is in another index
1240
 
                        # parents_not_on_page.add(key)
1241
 
                        # Missing for some reason
1242
 
                        if key < node.min_key:
1243
 
                            # in the case of bzr.dev, 3.4k/5.3k misses are
1244
 
                            # 'earlier' misses (65%)
1245
 
                            parents_not_on_page.add(key)
1246
 
                        elif key > node.max_key:
1247
 
                            # This parent key would be present on a different
1248
 
                            # LeafNode
1249
 
                            parents_not_on_page.add(key)
1250
 
                        else:
1251
 
                            # assert key != node.min_key and key != node.max_key
1252
 
                            # If it was going to be present, it would be on
1253
 
                            # *this* page, so mark it missing.
1254
 
                            missing_keys.add(key)
1255
 
                parents_to_check = next_parents_to_check.difference(parent_map)
1256
 
                # Might want to do another .difference() from missing_keys
1257
 
        # parents_not_on_page could have been found on a different page, or be
1258
 
        # known to be missing. So cull out everything that has already been
1259
 
        # found.
1260
 
        search_keys = parents_not_on_page.difference(
1261
 
            parent_map).difference(missing_keys)
1262
 
        return search_keys
1263
 
 
1264
 
    def iter_entries_prefix(self, keys):
1265
 
        """Iterate over keys within the index using prefix matching.
1266
 
 
1267
 
        Prefix matching is applied within the tuple of a key, not to within
1268
 
        the bytestring of each key element. e.g. if you have the keys ('foo',
1269
 
        'bar'), ('foobar', 'gam') and do a prefix search for ('foo', None) then
1270
 
        only the former key is returned.
1271
 
 
1272
 
        WARNING: Note that this method currently causes a full index parse
1273
 
        unconditionally (which is reasonably appropriate as it is a means for
1274
 
        thunking many small indices into one larger one and still supplies
1275
 
        iter_all_entries at the thunk layer).
1276
 
 
1277
 
        :param keys: An iterable providing the key prefixes to be retrieved.
1278
 
            Each key prefix takes the form of a tuple the length of a key, but
1279
 
            with the last N elements 'None' rather than a regular bytestring.
1280
 
            The first element cannot be 'None'.
1281
 
        :return: An iterable as per iter_all_entries, but restricted to the
1282
 
            keys with a matching prefix to those supplied. No additional keys
1283
 
            will be returned, and every match that is in the index will be
1284
 
            returned.
1285
 
        """
1286
 
        keys = sorted(set(keys))
1287
 
        if not keys:
1288
 
            return
1289
 
        # Load if needed to check key lengths
1290
 
        if self._key_count is None:
1291
 
            self._get_root_node()
1292
 
        # TODO: only access nodes that can satisfy the prefixes we are looking
1293
 
        # for. For now, to meet API usage (as this function is not used by
1294
 
        # current bzrlib) just suck the entire index and iterate in memory.
1295
 
        nodes = {}
1296
 
        if self.node_ref_lists:
1297
 
            if self._key_length == 1:
1298
 
                for _1, key, value, refs in self.iter_all_entries():
1299
 
                    nodes[key] = value, refs
1300
 
            else:
1301
 
                nodes_by_key = {}
1302
 
                for _1, key, value, refs in self.iter_all_entries():
1303
 
                    key_value = key, value, refs
1304
 
                    # For a key of (foo, bar, baz) create
1305
 
                    # _nodes_by_key[foo][bar][baz] = key_value
1306
 
                    key_dict = nodes_by_key
1307
 
                    for subkey in key[:-1]:
1308
 
                        key_dict = key_dict.setdefault(subkey, {})
1309
 
                    key_dict[key[-1]] = key_value
1310
 
        else:
1311
 
            if self._key_length == 1:
1312
 
                for _1, key, value in self.iter_all_entries():
1313
 
                    nodes[key] = value
1314
 
            else:
1315
 
                nodes_by_key = {}
1316
 
                for _1, key, value in self.iter_all_entries():
1317
 
                    key_value = key, value
1318
 
                    # For a key of (foo, bar, baz) create
1319
 
                    # _nodes_by_key[foo][bar][baz] = key_value
1320
 
                    key_dict = nodes_by_key
1321
 
                    for subkey in key[:-1]:
1322
 
                        key_dict = key_dict.setdefault(subkey, {})
1323
 
                    key_dict[key[-1]] = key_value
1324
 
        if self._key_length == 1:
1325
 
            for key in keys:
1326
 
                # sanity check
1327
 
                if key[0] is None:
1328
 
                    raise errors.BadIndexKey(key)
1329
 
                if len(key) != self._key_length:
1330
 
                    raise errors.BadIndexKey(key)
1331
 
                try:
1332
 
                    if self.node_ref_lists:
1333
 
                        value, node_refs = nodes[key]
1334
 
                        yield self, key, value, node_refs
1335
 
                    else:
1336
 
                        yield self, key, nodes[key]
1337
 
                except KeyError:
1338
 
                    pass
1339
 
            return
1340
 
        for key in keys:
1341
 
            # sanity check
1342
 
            if key[0] is None:
1343
 
                raise errors.BadIndexKey(key)
1344
 
            if len(key) != self._key_length:
1345
 
                raise errors.BadIndexKey(key)
1346
 
            # find what it refers to:
1347
 
            key_dict = nodes_by_key
1348
 
            elements = list(key)
1349
 
            # find the subdict whose contents should be returned.
1350
 
            try:
1351
 
                while len(elements) and elements[0] is not None:
1352
 
                    key_dict = key_dict[elements[0]]
1353
 
                    elements.pop(0)
1354
 
            except KeyError:
1355
 
                # a non-existant lookup.
1356
 
                continue
1357
 
            if len(elements):
1358
 
                dicts = [key_dict]
1359
 
                while dicts:
1360
 
                    key_dict = dicts.pop(-1)
1361
 
                    # can't be empty or would not exist
1362
 
                    item, value = key_dict.iteritems().next()
1363
 
                    if type(value) == dict:
1364
 
                        # push keys
1365
 
                        dicts.extend(key_dict.itervalues())
1366
 
                    else:
1367
 
                        # yield keys
1368
 
                        for value in key_dict.itervalues():
1369
 
                            # each value is the key:value:node refs tuple
1370
 
                            # ready to yield.
1371
 
                            yield (self, ) + value
1372
 
            else:
1373
 
                # the last thing looked up was a terminal element
1374
 
                yield (self, ) + key_dict
1375
 
 
1376
 
    def key_count(self):
1377
 
        """Return an estimate of the number of keys in this index.
1378
 
 
1379
 
        For BTreeGraphIndex the estimate is exact as it is contained in the
1380
 
        header.
1381
 
        """
1382
 
        if self._key_count is None:
1383
 
            self._get_root_node()
1384
 
        return self._key_count
1385
 
 
1386
 
    def _compute_row_offsets(self):
1387
 
        """Fill out the _row_offsets attribute based on _row_lengths."""
1388
 
        offsets = []
1389
 
        row_offset = 0
1390
 
        for row in self._row_lengths:
1391
 
            offsets.append(row_offset)
1392
 
            row_offset += row
1393
 
        offsets.append(row_offset)
1394
 
        self._row_offsets = offsets
1395
 
 
1396
 
    def _parse_header_from_bytes(self, bytes):
1397
 
        """Parse the header from a region of bytes.
1398
 
 
1399
 
        :param bytes: The data to parse.
1400
 
        :return: An offset, data tuple such as readv yields, for the unparsed
1401
 
            data. (which may be of length 0).
1402
 
        """
1403
 
        signature = bytes[0:len(self._signature())]
1404
 
        if not signature == self._signature():
1405
 
            raise errors.BadIndexFormatSignature(self._name, BTreeGraphIndex)
1406
 
        lines = bytes[len(self._signature()):].splitlines()
1407
 
        options_line = lines[0]
1408
 
        if not options_line.startswith(_OPTION_NODE_REFS):
1409
 
            raise errors.BadIndexOptions(self)
1410
 
        try:
1411
 
            self.node_ref_lists = int(options_line[len(_OPTION_NODE_REFS):])
1412
 
        except ValueError:
1413
 
            raise errors.BadIndexOptions(self)
1414
 
        options_line = lines[1]
1415
 
        if not options_line.startswith(_OPTION_KEY_ELEMENTS):
1416
 
            raise errors.BadIndexOptions(self)
1417
 
        try:
1418
 
            self._key_length = int(options_line[len(_OPTION_KEY_ELEMENTS):])
1419
 
        except ValueError:
1420
 
            raise errors.BadIndexOptions(self)
1421
 
        options_line = lines[2]
1422
 
        if not options_line.startswith(_OPTION_LEN):
1423
 
            raise errors.BadIndexOptions(self)
1424
 
        try:
1425
 
            self._key_count = int(options_line[len(_OPTION_LEN):])
1426
 
        except ValueError:
1427
 
            raise errors.BadIndexOptions(self)
1428
 
        options_line = lines[3]
1429
 
        if not options_line.startswith(_OPTION_ROW_LENGTHS):
1430
 
            raise errors.BadIndexOptions(self)
1431
 
        try:
1432
 
            self._row_lengths = map(int, [length for length in
1433
 
                options_line[len(_OPTION_ROW_LENGTHS):].split(',')
1434
 
                if len(length)])
1435
 
        except ValueError:
1436
 
            raise errors.BadIndexOptions(self)
1437
 
        self._compute_row_offsets()
1438
 
 
1439
 
        # calculate the bytes we have processed
1440
 
        header_end = (len(signature) + sum(map(len, lines[0:4])) + 4)
1441
 
        return header_end, bytes[header_end:]
1442
 
 
1443
 
    def _read_nodes(self, nodes):
1444
 
        """Read some nodes from disk into the LRU cache.
1445
 
 
1446
 
        This performs a readv to get the node data into memory, and parses each
1447
 
        node, then yields it to the caller. The nodes are requested in the
1448
 
        supplied order. If possible doing sort() on the list before requesting
1449
 
        a read may improve performance.
1450
 
 
1451
 
        :param nodes: The nodes to read. 0 - first node, 1 - second node etc.
1452
 
        :return: None
1453
 
        """
1454
 
        # may be the byte string of the whole file
1455
 
        bytes = None
1456
 
        # list of (offset, length) regions of the file that should, evenually
1457
 
        # be read in to data_ranges, either from 'bytes' or from the transport
1458
 
        ranges = []
1459
 
        for index in nodes:
1460
 
            offset = index * _PAGE_SIZE
1461
 
            size = _PAGE_SIZE
1462
 
            if index == 0:
1463
 
                # Root node - special case
1464
 
                if self._size:
1465
 
                    size = min(_PAGE_SIZE, self._size)
1466
 
                else:
1467
 
                    # The only case where we don't know the size, is for very
1468
 
                    # small indexes. So we read the whole thing
1469
 
                    bytes = self._transport.get_bytes(self._name)
1470
 
                    self._size = len(bytes)
1471
 
                    # the whole thing should be parsed out of 'bytes'
1472
 
                    ranges.append((0, len(bytes)))
1473
 
                    break
1474
 
            else:
1475
 
                if offset > self._size:
1476
 
                    raise AssertionError('tried to read past the end'
1477
 
                                         ' of the file %s > %s'
1478
 
                                         % (offset, self._size))
1479
 
                size = min(size, self._size - offset)
1480
 
            ranges.append((offset, size))
1481
 
        if not ranges:
1482
 
            return
1483
 
        elif bytes is not None:
1484
 
            # already have the whole file
1485
 
            data_ranges = [(start, bytes[start:start+_PAGE_SIZE])
1486
 
                           for start in xrange(0, len(bytes), _PAGE_SIZE)]
1487
 
        elif self._file is None:
1488
 
            data_ranges = self._transport.readv(self._name, ranges)
1489
 
        else:
1490
 
            data_ranges = []
1491
 
            for offset, size in ranges:
1492
 
                self._file.seek(offset)
1493
 
                data_ranges.append((offset, self._file.read(size)))
1494
 
        for offset, data in data_ranges:
1495
 
            if offset == 0:
1496
 
                # extract the header
1497
 
                offset, data = self._parse_header_from_bytes(data)
1498
 
                if len(data) == 0:
1499
 
                    continue
1500
 
            bytes = zlib.decompress(data)
1501
 
            if bytes.startswith(_LEAF_FLAG):
1502
 
                node = _LeafNode(bytes, self._key_length, self.node_ref_lists)
1503
 
            elif bytes.startswith(_INTERNAL_FLAG):
1504
 
                node = _InternalNode(bytes)
1505
 
            else:
1506
 
                raise AssertionError("Unknown node type for %r" % bytes)
1507
 
            yield offset / _PAGE_SIZE, node
1508
 
 
1509
 
    def _signature(self):
1510
 
        """The file signature for this index type."""
1511
 
        return _BTSIGNATURE
1512
 
 
1513
 
    def validate(self):
1514
 
        """Validate that everything in the index can be accessed."""
1515
 
        # just read and parse every node.
1516
 
        self._get_root_node()
1517
 
        if len(self._row_lengths) > 1:
1518
 
            start_node = self._row_offsets[1]
1519
 
        else:
1520
 
            # We shouldn't be reading anything anyway
1521
 
            start_node = 1
1522
 
        node_end = self._row_offsets[-1]
1523
 
        for node in self._read_nodes(range(start_node, node_end)):
1524
 
            pass
1525
 
 
1526
 
 
1527
 
try:
1528
 
    from bzrlib import _btree_serializer_pyx as _btree_serializer
1529
 
except ImportError:
1530
 
    from bzrlib import _btree_serializer_py as _btree_serializer