1
# Copyright (C) 2008 Canonical Ltd
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.
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.
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
22
from bisect import bisect_right
23
from copy import deepcopy
38
from bzrlib.index import _OPTION_NODE_REFS, _OPTION_KEY_ELEMENTS, _OPTION_LEN
39
from bzrlib.transport import get_transport
42
_BTSIGNATURE = "B+Tree Graph Index 2\n"
43
_OPTION_ROW_LENGTHS = "row_lengths="
44
_LEAF_FLAG = "type=leaf\n"
45
_INTERNAL_FLAG = "type=internal\n"
46
_INTERNAL_OFFSET = "offset="
48
_RESERVED_HEADER_BYTES = 120
51
# 4K per page: 4MB - 1000 entries
52
_NODE_CACHE_SIZE = 1000
55
class _BuilderRow(object):
56
"""The stored state accumulated while writing out a row in the index.
58
:ivar spool: A temporary file used to accumulate nodes for this row
60
:ivar nodes: The count of nodes emitted so far.
64
"""Create a _BuilderRow."""
66
self.spool = tempfile.TemporaryFile()
69
def finish_node(self, pad=True):
70
byte_lines, _, padding = self.writer.finish()
73
self.spool.write("\x00" * _RESERVED_HEADER_BYTES)
75
if not pad and padding:
77
skipped_bytes = padding
78
self.spool.writelines(byte_lines)
79
remainder = (self.spool.tell() + skipped_bytes) % _PAGE_SIZE
81
raise AssertionError("incorrect node length: %d, %d"
82
% (self.spool.tell(), remainder))
87
class _InternalBuilderRow(_BuilderRow):
88
"""The stored state accumulated while writing out internal rows."""
90
def finish_node(self, pad=True):
92
raise AssertionError("Must pad internal nodes only.")
93
_BuilderRow.finish_node(self)
96
class _LeafBuilderRow(_BuilderRow):
97
"""The stored state accumulated while writing out a leaf rows."""
100
class BTreeBuilder(index.GraphIndexBuilder):
101
"""A Builder for B+Tree based Graph indices.
103
The resulting graph has the structure:
105
_SIGNATURE OPTIONS NODES
106
_SIGNATURE := 'B+Tree Graph Index 1' NEWLINE
107
OPTIONS := REF_LISTS KEY_ELEMENTS LENGTH
108
REF_LISTS := 'node_ref_lists=' DIGITS NEWLINE
109
KEY_ELEMENTS := 'key_elements=' DIGITS NEWLINE
110
LENGTH := 'len=' DIGITS NEWLINE
111
ROW_LENGTHS := 'row_lengths' DIGITS (COMMA DIGITS)*
112
NODES := NODE_COMPRESSED*
113
NODE_COMPRESSED:= COMPRESSED_BYTES{4096}
114
NODE_RAW := INTERNAL | LEAF
115
INTERNAL := INTERNAL_FLAG POINTERS
116
LEAF := LEAF_FLAG ROWS
117
KEY_ELEMENT := Not-whitespace-utf8
118
KEY := KEY_ELEMENT (NULL KEY_ELEMENT)*
120
ROW := KEY NULL ABSENT? NULL REFERENCES NULL VALUE NEWLINE
122
REFERENCES := REFERENCE_LIST (TAB REFERENCE_LIST){node_ref_lists - 1}
123
REFERENCE_LIST := (REFERENCE (CR REFERENCE)*)?
125
VALUE := no-newline-no-null-bytes
128
def __init__(self, reference_lists=0, key_elements=1, spill_at=100000):
129
"""See GraphIndexBuilder.__init__.
131
:param spill_at: Optional parameter controlling the maximum number
132
of nodes that BTreeBuilder will hold in memory.
134
index.GraphIndexBuilder.__init__(self, reference_lists=reference_lists,
135
key_elements=key_elements)
136
self._spill_at = spill_at
137
self._backing_indices = []
138
# A map of {key: (node_refs, value)}
140
# Indicate it hasn't been built yet
141
self._nodes_by_key = None
142
self._optimize_for_size = False
144
def add_node(self, key, value, references=()):
145
"""Add a node to the index.
147
If adding the node causes the builder to reach its spill_at threshold,
148
disk spilling will be triggered.
150
:param key: The key. keys are non-empty tuples containing
151
as many whitespace-free utf8 bytestrings as the key length
152
defined for this index.
153
:param references: An iterable of iterables of keys. Each is a
154
reference to another key.
155
:param value: The value to associate with the key. It may be any
156
bytes as long as it does not contain \0 or \n.
158
# we don't care about absent_references
159
node_refs, _ = self._check_key_ref_value(key, references, value)
160
if key in self._nodes:
161
raise errors.BadIndexDuplicateKey(key, self)
162
self._nodes[key] = (node_refs, value)
164
if self._nodes_by_key is not None and self._key_length > 1:
165
self._update_nodes_by_key(key, value, node_refs)
166
if len(self._keys) < self._spill_at:
168
self._spill_mem_keys_to_disk()
170
def _spill_mem_keys_to_disk(self):
171
"""Write the in memory keys down to disk to cap memory consumption.
173
If we already have some keys written to disk, we will combine them so
174
as to preserve the sorted order. The algorithm for combining uses
175
powers of two. So on the first spill, write all mem nodes into a
176
single index. On the second spill, combine the mem nodes with the nodes
177
on disk to create a 2x sized disk index and get rid of the first index.
178
On the third spill, create a single new disk index, which will contain
179
the mem nodes, and preserve the existing 2x sized index. On the fourth,
180
combine mem with the first and second indexes, creating a new one of
181
size 4x. On the fifth create a single new one, etc.
183
iterators_to_combine = [self._iter_mem_nodes()]
185
for pos, backing in enumerate(self._backing_indices):
189
iterators_to_combine.append(backing.iter_all_entries())
190
backing_pos = pos + 1
191
new_backing_file, size = \
192
self._write_nodes(self._iter_smallest(iterators_to_combine),
193
allow_optimize=False)
194
dir_path, base_name = osutils.split(new_backing_file.name)
195
# Note: The transport here isn't strictly needed, because we will use
196
# direct access to the new_backing._file object
197
new_backing = BTreeGraphIndex(get_transport(dir_path),
199
# GC will clean up the file
200
new_backing._file = new_backing_file
201
if len(self._backing_indices) == backing_pos:
202
self._backing_indices.append(None)
203
self._backing_indices[backing_pos] = new_backing
204
for pos in range(backing_pos):
205
self._backing_indices[pos] = None
208
self._nodes_by_key = None
210
def add_nodes(self, nodes):
211
"""Add nodes to the index.
213
:param nodes: An iterable of (key, node_refs, value) entries to add.
215
if self.reference_lists:
216
for (key, value, node_refs) in nodes:
217
self.add_node(key, value, node_refs)
219
for (key, value) in nodes:
220
self.add_node(key, value)
222
def _iter_mem_nodes(self):
223
"""Iterate over the nodes held in memory."""
225
if self.reference_lists:
226
for key in sorted(nodes):
227
references, value = nodes[key]
228
yield self, key, value, references
230
for key in sorted(nodes):
231
references, value = nodes[key]
232
yield self, key, value
234
def _iter_smallest(self, iterators_to_combine):
235
if len(iterators_to_combine) == 1:
236
for value in iterators_to_combine[0]:
240
for iterator in iterators_to_combine:
242
current_values.append(iterator.next())
243
except StopIteration:
244
current_values.append(None)
247
# Decorate candidates with the value to allow 2.4's min to be used.
248
candidates = [(item[1][1], item) for item
249
in enumerate(current_values) if item[1] is not None]
250
if not len(candidates):
252
selected = min(candidates)
253
# undecorate back to (pos, node)
254
selected = selected[1]
255
if last == selected[1][1]:
256
raise errors.BadIndexDuplicateKey(last, self)
257
last = selected[1][1]
258
# Yield, with self as the index
259
yield (self,) + selected[1][1:]
262
current_values[pos] = iterators_to_combine[pos].next()
263
except StopIteration:
264
current_values[pos] = None
266
def _add_key(self, string_key, line, rows, allow_optimize=True):
267
"""Add a key to the current chunk.
269
:param string_key: The key to add.
270
:param line: The fully serialised key and value.
271
:param allow_optimize: If set to False, prevent setting the optimize
272
flag when writing out. This is used by the _spill_mem_keys_to_disk
275
if rows[-1].writer is None:
276
# opening a new leaf chunk;
277
for pos, internal_row in enumerate(rows[:-1]):
278
# flesh out any internal nodes that are needed to
279
# preserve the height of the tree
280
if internal_row.writer is None:
282
if internal_row.nodes == 0:
283
length -= _RESERVED_HEADER_BYTES # padded
285
optimize_for_size = self._optimize_for_size
287
optimize_for_size = False
288
internal_row.writer = chunk_writer.ChunkWriter(length, 0,
289
optimize_for_size=optimize_for_size)
290
internal_row.writer.write(_INTERNAL_FLAG)
291
internal_row.writer.write(_INTERNAL_OFFSET +
292
str(rows[pos + 1].nodes) + "\n")
295
if rows[-1].nodes == 0:
296
length -= _RESERVED_HEADER_BYTES # padded
297
rows[-1].writer = chunk_writer.ChunkWriter(length,
298
optimize_for_size=self._optimize_for_size)
299
rows[-1].writer.write(_LEAF_FLAG)
300
if rows[-1].writer.write(line):
301
# this key did not fit in the node:
302
rows[-1].finish_node()
303
key_line = string_key + "\n"
305
for row in reversed(rows[:-1]):
306
# Mark the start of the next node in the node above. If it
307
# doesn't fit then propogate upwards until we find one that
309
if row.writer.write(key_line):
312
# We've found a node that can handle the pointer.
315
# If we reached the current root without being able to mark the
316
# division point, then we need a new root:
319
if 'index' in debug.debug_flags:
320
trace.mutter('Inserting new global row.')
321
new_row = _InternalBuilderRow()
323
rows.insert(0, new_row)
324
# This will be padded, hence the -100
325
new_row.writer = chunk_writer.ChunkWriter(
326
_PAGE_SIZE - _RESERVED_HEADER_BYTES,
328
optimize_for_size=self._optimize_for_size)
329
new_row.writer.write(_INTERNAL_FLAG)
330
new_row.writer.write(_INTERNAL_OFFSET +
331
str(rows[1].nodes - 1) + "\n")
332
new_row.writer.write(key_line)
333
self._add_key(string_key, line, rows, allow_optimize=allow_optimize)
335
def _write_nodes(self, node_iterator, allow_optimize=True):
336
"""Write node_iterator out as a B+Tree.
338
:param node_iterator: An iterator of sorted nodes. Each node should
339
match the output given by iter_all_entries.
340
:param allow_optimize: If set to False, prevent setting the optimize
341
flag when writing out. This is used by the _spill_mem_keys_to_disk
343
:return: A file handle for a temporary file containing a B+Tree for
346
# The index rows - rows[0] is the root, rows[1] is the layer under it
349
# forward sorted by key. In future we may consider topological sorting,
350
# at the cost of table scans for direct lookup, or a second index for
353
# A stack with the number of nodes of each size. 0 is the root node
354
# and must always be 1 (if there are any nodes in the tree).
355
self.row_lengths = []
356
# Loop over all nodes adding them to the bottom row
357
# (rows[-1]). When we finish a chunk in a row,
358
# propogate the key that didn't fit (comes after the chunk) to the
359
# row above, transitively.
360
for node in node_iterator:
362
# First key triggers the first row
363
rows.append(_LeafBuilderRow())
365
string_key, line = _btree_serializer._flatten_node(node,
366
self.reference_lists)
367
self._add_key(string_key, line, rows, allow_optimize=allow_optimize)
368
for row in reversed(rows):
369
pad = (type(row) != _LeafBuilderRow)
370
row.finish_node(pad=pad)
371
result = tempfile.NamedTemporaryFile(prefix='bzr-index-')
372
lines = [_BTSIGNATURE]
373
lines.append(_OPTION_NODE_REFS + str(self.reference_lists) + '\n')
374
lines.append(_OPTION_KEY_ELEMENTS + str(self._key_length) + '\n')
375
lines.append(_OPTION_LEN + str(key_count) + '\n')
376
row_lengths = [row.nodes for row in rows]
377
lines.append(_OPTION_ROW_LENGTHS + ','.join(map(str, row_lengths)) + '\n')
378
result.writelines(lines)
379
position = sum(map(len, lines))
381
if position > _RESERVED_HEADER_BYTES:
382
raise AssertionError("Could not fit the header in the"
383
" reserved space: %d > %d"
384
% (position, _RESERVED_HEADER_BYTES))
385
# write the rows out:
387
reserved = _RESERVED_HEADER_BYTES # reserved space for first node
390
# copy nodes to the finalised file.
391
# Special case the first node as it may be prefixed
392
node = row.spool.read(_PAGE_SIZE)
393
result.write(node[reserved:])
394
result.write("\x00" * (reserved - position))
395
position = 0 # Only the root row actually has an offset
396
copied_len = osutils.pumpfile(row.spool, result)
397
if copied_len != (row.nodes - 1) * _PAGE_SIZE:
398
if type(row) != _LeafBuilderRow:
399
raise AssertionError("Incorrect amount of data copied"
400
" expected: %d, got: %d"
401
% ((row.nodes - 1) * _PAGE_SIZE,
409
"""Finalise the index.
411
:return: A file handle for a temporary file containing the nodes added
414
return self._write_nodes(self.iter_all_entries())[0]
416
def iter_all_entries(self):
417
"""Iterate over all keys within the index
419
:return: An iterable of (index, key, reference_lists, value). There is no
420
defined order for the result iteration - it will be in the most
421
efficient order for the index (in this case dictionary hash order).
423
if 'evil' in debug.debug_flags:
424
trace.mutter_callsite(3,
425
"iter_all_entries scales with size of history.")
426
# Doing serial rather than ordered would be faster; but this shouldn't
427
# be getting called routinely anyway.
428
iterators = [self._iter_mem_nodes()]
429
for backing in self._backing_indices:
430
if backing is not None:
431
iterators.append(backing.iter_all_entries())
432
if len(iterators) == 1:
434
return self._iter_smallest(iterators)
436
def iter_entries(self, keys):
437
"""Iterate over keys within the index.
439
:param keys: An iterable providing the keys to be retrieved.
440
:return: An iterable of (index, key, value, reference_lists). There is no
441
defined order for the result iteration - it will be in the most
442
efficient order for the index (keys iteration order in this case).
445
local_keys = keys.intersection(self._keys)
446
if self.reference_lists:
447
for key in local_keys:
448
node = self._nodes[key]
449
yield self, key, node[1], node[0]
451
for key in local_keys:
452
node = self._nodes[key]
453
yield self, key, node[1]
454
# Find things that are in backing indices that have not been handled
456
if not self._backing_indices:
457
return # We won't find anything there either
458
# Remove all of the keys that we found locally
459
keys.difference_update(local_keys)
460
for backing in self._backing_indices:
465
for node in backing.iter_entries(keys):
467
yield (self,) + node[1:]
469
def iter_entries_prefix(self, keys):
470
"""Iterate over keys within the index using prefix matching.
472
Prefix matching is applied within the tuple of a key, not to within
473
the bytestring of each key element. e.g. if you have the keys ('foo',
474
'bar'), ('foobar', 'gam') and do a prefix search for ('foo', None) then
475
only the former key is returned.
477
:param keys: An iterable providing the key prefixes to be retrieved.
478
Each key prefix takes the form of a tuple the length of a key, but
479
with the last N elements 'None' rather than a regular bytestring.
480
The first element cannot be 'None'.
481
:return: An iterable as per iter_all_entries, but restricted to the
482
keys with a matching prefix to those supplied. No additional keys
483
will be returned, and every match that is in the index will be
486
# XXX: To much duplication with the GraphIndex class; consider finding
487
# a good place to pull out the actual common logic.
491
for backing in self._backing_indices:
494
for node in backing.iter_entries_prefix(keys):
495
yield (self,) + node[1:]
496
if self._key_length == 1:
500
raise errors.BadIndexKey(key)
501
if len(key) != self._key_length:
502
raise errors.BadIndexKey(key)
504
node = self._nodes[key]
507
if self.reference_lists:
508
yield self, key, node[1], node[0]
510
yield self, key, node[1]
515
raise errors.BadIndexKey(key)
516
if len(key) != self._key_length:
517
raise errors.BadIndexKey(key)
518
# find what it refers to:
519
key_dict = self._get_nodes_by_key()
521
# find the subdict to return
523
while len(elements) and elements[0] is not None:
524
key_dict = key_dict[elements[0]]
527
# a non-existant lookup.
532
key_dict = dicts.pop(-1)
533
# can't be empty or would not exist
534
item, value = key_dict.iteritems().next()
535
if type(value) == dict:
537
dicts.extend(key_dict.itervalues())
540
for value in key_dict.itervalues():
541
yield (self, ) + value
543
yield (self, ) + key_dict
545
def _get_nodes_by_key(self):
546
if self._nodes_by_key is None:
548
if self.reference_lists:
549
for key, (references, value) in self._nodes.iteritems():
550
key_dict = nodes_by_key
551
for subkey in key[:-1]:
552
key_dict = key_dict.setdefault(subkey, {})
553
key_dict[key[-1]] = key, value, references
555
for key, (references, value) in self._nodes.iteritems():
556
key_dict = nodes_by_key
557
for subkey in key[:-1]:
558
key_dict = key_dict.setdefault(subkey, {})
559
key_dict[key[-1]] = key, value
560
self._nodes_by_key = nodes_by_key
561
return self._nodes_by_key
564
"""Return an estimate of the number of keys in this index.
566
For InMemoryGraphIndex the estimate is exact.
568
return len(self._keys) + sum(backing.key_count() for backing in
569
self._backing_indices if backing is not None)
572
"""In memory index's have no known corruption at the moment."""
575
class _LeafNode(object):
576
"""A leaf node for a serialised B+Tree index."""
578
def __init__(self, bytes, key_length, ref_list_length):
579
"""Parse bytes to create a leaf node object."""
580
# splitlines mangles the \r delimiters.. don't use it.
581
self.keys = dict(_btree_serializer._parse_leaf_lines(bytes,
582
key_length, ref_list_length))
585
class _InternalNode(object):
586
"""An internal node for a serialised B+Tree index."""
588
def __init__(self, bytes):
589
"""Parse bytes to create an internal node object."""
590
# splitlines mangles the \r delimiters.. don't use it.
591
self.keys = self._parse_lines(bytes.split('\n'))
593
def _parse_lines(self, lines):
595
self.offset = int(lines[1][7:])
596
for line in lines[2:]:
599
nodes.append(tuple(line.split('\0')))
603
class BTreeGraphIndex(object):
604
"""Access to nodes via the standard GraphIndex interface for B+Tree's.
606
Individual nodes are held in a LRU cache. This holds the root node in
607
memory except when very large walks are done.
610
def __init__(self, transport, name, size):
611
"""Create a B+Tree index object on the index name.
613
:param transport: The transport to read data for the index from.
614
:param name: The file name of the index on transport.
615
:param size: Optional size of the index in bytes. This allows
616
compatibility with the GraphIndex API, as well as ensuring that
617
the initial read (to read the root node header) can be done
618
without over-reading even on empty indices, and on small indices
619
allows single-IO to read the entire index.
621
self._transport = transport
625
self._recommended_pages = self._compute_recommended_pages()
626
self._root_node = None
627
# Default max size is 100,000 leave values
628
self._leaf_value_cache = None # lru_cache.LRUCache(100*1000)
629
self._leaf_node_cache = lru_cache.LRUCache(_NODE_CACHE_SIZE)
630
self._internal_node_cache = lru_cache.LRUCache()
631
self._key_count = None
632
self._row_lengths = None
633
self._row_offsets = None # Start of each row, [-1] is the end
635
def __eq__(self, other):
636
"""Equal when self and other were created with the same parameters."""
638
type(self) == type(other) and
639
self._transport == other._transport and
640
self._name == other._name and
641
self._size == other._size)
643
def __ne__(self, other):
644
return not self.__eq__(other)
646
def _get_and_cache_nodes(self, nodes):
647
"""Read nodes and cache them in the lru.
649
The nodes list supplied is sorted and then read from disk, each node
650
being inserted it into the _node_cache.
652
Note: Asking for more nodes than the _node_cache can contain will
653
result in some of the results being immediately discarded, to prevent
654
this an assertion is raised if more nodes are asked for than are
657
:return: A dict of {node_pos: node}
660
start_of_leaves = None
661
for node_pos, node in self._read_nodes(sorted(nodes)):
662
if node_pos == 0: # Special case
663
self._root_node = node
665
if start_of_leaves is None:
666
start_of_leaves = self._row_offsets[-2]
667
if node_pos < start_of_leaves:
668
self._internal_node_cache.add(node_pos, node)
670
self._leaf_node_cache.add(node_pos, node)
671
found[node_pos] = node
674
def _compute_recommended_pages(self):
675
"""Convert transport's recommended_page_size into btree pages.
677
recommended_page_size is in bytes, we want to know how many _PAGE_SIZE
678
pages fit in that length.
680
recommended_read = self._transport.recommended_page_size()
681
recommended_pages = int(math.ceil(recommended_read /
683
return recommended_pages
685
def _compute_total_pages_in_index(self):
686
"""How many pages are in the index.
688
If we have read the header we will use the value stored there.
689
Otherwise it will be computed based on the length of the index.
691
if self._size is None:
692
raise AssertionError('_compute_total_pages_in_index should not be'
693
' called when self._size is None')
694
if self._root_node is not None:
695
# This is the number of pages as defined by the header
696
return self._row_offsets[-1]
697
# This is the number of pages as defined by the size of the index. They
698
# should be indentical.
699
total_pages = int(math.ceil(self._size / float(_PAGE_SIZE)))
702
def _expand_offsets(self, offsets):
703
"""Find extra pages to download.
705
The idea is that we always want to make big-enough requests (like 64kB
706
for http), so that we don't waste round trips. So given the entries
707
that we already have cached and the new pages being downloaded figure
708
out what other pages we might want to read.
710
See also doc/developers/btree_index_prefetch.txt for more details.
712
:param offsets: The offsets to be read
713
:return: A list of offsets to download
715
if 'index' in debug.debug_flags:
716
trace.mutter('expanding: %s\toffsets: %s', self._name, offsets)
718
if len(offsets) >= self._recommended_pages:
719
# Don't add more, we are already requesting more than enough
720
if 'index' in debug.debug_flags:
721
trace.mutter(' not expanding large request (%s >= %s)',
722
len(offsets), self._recommended_pages)
724
if self._size is None:
725
# Don't try anything, because we don't know where the file ends
726
if 'index' in debug.debug_flags:
727
trace.mutter(' not expanding without knowing index size')
729
total_pages = self._compute_total_pages_in_index()
730
cached_offsets = self._get_offsets_to_cached_pages()
731
# If reading recommended_pages would read the rest of the index, just
733
if total_pages - len(cached_offsets) <= self._recommended_pages:
734
# Read whatever is left
736
expanded = [x for x in xrange(total_pages)
737
if x not in cached_offsets]
739
expanded = range(total_pages)
740
if 'index' in debug.debug_flags:
741
trace.mutter(' reading all unread pages: %s', expanded)
744
if self._root_node is None:
745
# ATM on the first read of the root node of a large index, we don't
746
# bother pre-reading any other pages. This is because the
747
# likelyhood of actually reading interesting pages is very low.
748
# See doc/developers/btree_index_prefetch.txt for a discussion, and
749
# a possible implementation when we are guessing that the second
750
# layer index is small
751
final_offsets = offsets
753
tree_depth = len(self._row_lengths)
754
if len(cached_offsets) < tree_depth and len(offsets) == 1:
755
# We haven't read enough to justify expansion
756
# If we are only going to read the root node, and 1 leaf node,
757
# then it isn't worth expanding our request. Once we've read at
758
# least 2 nodes, then we are probably doing a search, and we
759
# start expanding our requests.
760
if 'index' in debug.debug_flags:
761
trace.mutter(' not expanding on first reads')
763
final_offsets = self._expand_to_neighbors(offsets, cached_offsets,
766
final_offsets = sorted(final_offsets)
767
if 'index' in debug.debug_flags:
768
trace.mutter('expanded: %s', final_offsets)
771
def _expand_to_neighbors(self, offsets, cached_offsets, total_pages):
772
"""Expand requests to neighbors until we have enough pages.
774
This is called from _expand_offsets after policy has determined that we
776
We only want to expand requests within a given layer. We cheat a little
777
bit and assume all requests will be in the same layer. This is true
778
given the current design, but if it changes this algorithm may perform
781
:param offsets: requested offsets
782
:param cached_offsets: offsets for pages we currently have cached
783
:return: A set() of offsets after expansion
785
final_offsets = set(offsets)
787
new_tips = set(final_offsets)
788
while len(final_offsets) < self._recommended_pages and new_tips:
792
first, end = self._find_layer_first_and_end(pos)
795
and previous not in cached_offsets
796
and previous not in final_offsets
797
and previous >= first):
798
next_tips.add(previous)
800
if (after < total_pages
801
and after not in cached_offsets
802
and after not in final_offsets
805
# This would keep us from going bigger than
806
# recommended_pages by only expanding the first offsets.
807
# However, if we are making a 'wide' request, it is
808
# reasonable to expand all points equally.
809
# if len(final_offsets) > recommended_pages:
811
final_offsets.update(next_tips)
815
def external_references(self, ref_list_num):
816
if self._root_node is None:
817
self._get_root_node()
818
if ref_list_num + 1 > self.node_ref_lists:
819
raise ValueError('No ref list %d, index has %d ref lists'
820
% (ref_list_num, self.node_ref_lists))
823
for node in self.iter_all_entries():
825
refs.update(node[3][ref_list_num])
828
def _find_layer_first_and_end(self, offset):
829
"""Find the start/stop nodes for the layer corresponding to offset.
831
:return: (first, end)
832
first is the first node in this layer
833
end is the first node of the next layer
836
for roffset in self._row_offsets:
843
def _get_offsets_to_cached_pages(self):
844
"""Determine what nodes we already have cached."""
845
cached_offsets = set(self._internal_node_cache.keys())
846
cached_offsets.update(self._leaf_node_cache.keys())
847
if self._root_node is not None:
848
cached_offsets.add(0)
849
return cached_offsets
851
def _get_root_node(self):
852
if self._root_node is None:
853
# We may not have a root node yet
854
self._get_internal_nodes([0])
855
return self._root_node
857
def _get_nodes(self, cache, node_indexes):
860
for idx in node_indexes:
861
if idx == 0 and self._root_node is not None:
862
found[0] = self._root_node
865
found[idx] = cache[idx]
870
needed = self._expand_offsets(needed)
871
found.update(self._get_and_cache_nodes(needed))
874
def _get_internal_nodes(self, node_indexes):
875
"""Get a node, from cache or disk.
877
After getting it, the node will be cached.
879
return self._get_nodes(self._internal_node_cache, node_indexes)
881
def _cache_leaf_values(self, nodes):
882
"""Cache directly from key => value, skipping the btree."""
883
if self._leaf_value_cache is not None:
884
for node in nodes.itervalues():
885
for key, value in node.keys.iteritems():
886
if key in self._leaf_value_cache:
887
# Don't add the rest of the keys, we've seen this node
890
self._leaf_value_cache[key] = value
892
def _get_leaf_nodes(self, node_indexes):
893
"""Get a bunch of nodes, from cache or disk."""
894
found = self._get_nodes(self._leaf_node_cache, node_indexes)
895
self._cache_leaf_values(found)
898
def iter_all_entries(self):
899
"""Iterate over all keys within the index.
901
:return: An iterable of (index, key, value) or (index, key, value, reference_lists).
902
The former tuple is used when there are no reference lists in the
903
index, making the API compatible with simple key:value index types.
904
There is no defined order for the result iteration - it will be in
905
the most efficient order for the index.
907
if 'evil' in debug.debug_flags:
908
trace.mutter_callsite(3,
909
"iter_all_entries scales with size of history.")
910
if not self.key_count():
912
if self._row_offsets[-1] == 1:
913
# There is only the root node, and we read that via key_count()
914
if self.node_ref_lists:
915
for key, (value, refs) in sorted(self._root_node.keys.items()):
916
yield (self, key, value, refs)
918
for key, (value, refs) in sorted(self._root_node.keys.items()):
919
yield (self, key, value)
921
start_of_leaves = self._row_offsets[-2]
922
end_of_leaves = self._row_offsets[-1]
923
needed_offsets = range(start_of_leaves, end_of_leaves)
924
if needed_offsets == [0]:
925
# Special case when we only have a root node, as we have already
927
nodes = [(0, self._root_node)]
929
nodes = self._read_nodes(needed_offsets)
930
# We iterate strictly in-order so that we can use this function
931
# for spilling index builds to disk.
932
if self.node_ref_lists:
933
for _, node in nodes:
934
for key, (value, refs) in sorted(node.keys.items()):
935
yield (self, key, value, refs)
937
for _, node in nodes:
938
for key, (value, refs) in sorted(node.keys.items()):
939
yield (self, key, value)
942
def _multi_bisect_right(in_keys, fixed_keys):
943
"""Find the positions where each 'in_key' would fit in fixed_keys.
945
This is equivalent to doing "bisect_right" on each in_key into
948
:param in_keys: A sorted list of keys to match with fixed_keys
949
:param fixed_keys: A sorted list of keys to match against
950
:return: A list of (integer position, [key list]) tuples.
955
# no pointers in the fixed_keys list, which means everything must
957
return [(0, in_keys)]
959
# TODO: Iterating both lists will generally take M + N steps
960
# Bisecting each key will generally take M * log2 N steps.
961
# If we had an efficient way to compare, we could pick the method
962
# based on which has the fewer number of steps.
963
# There is also the argument that bisect_right is a compiled
964
# function, so there is even more to be gained.
965
# iter_steps = len(in_keys) + len(fixed_keys)
966
# bisect_steps = len(in_keys) * math.log(len(fixed_keys), 2)
967
if len(in_keys) == 1: # Bisect will always be faster for M = 1
968
return [(bisect_right(fixed_keys, in_keys[0]), in_keys)]
969
# elif bisect_steps < iter_steps:
971
# for key in in_keys:
972
# offsets.setdefault(bisect_right(fixed_keys, key),
974
# return [(o, offsets[o]) for o in sorted(offsets)]
975
in_keys_iter = iter(in_keys)
976
fixed_keys_iter = enumerate(fixed_keys)
977
cur_in_key = in_keys_iter.next()
978
cur_fixed_offset, cur_fixed_key = fixed_keys_iter.next()
980
class InputDone(Exception): pass
981
class FixedDone(Exception): pass
986
# TODO: Another possibility is that rather than iterating on each side,
987
# we could use a combination of bisecting and iterating. For
988
# example, while cur_in_key < fixed_key, bisect to find its
989
# point, then iterate all matching keys, then bisect (restricted
990
# to only the remainder) for the next one, etc.
993
if cur_in_key < cur_fixed_key:
995
cur_out = (cur_fixed_offset, cur_keys)
996
output.append(cur_out)
997
while cur_in_key < cur_fixed_key:
998
cur_keys.append(cur_in_key)
1000
cur_in_key = in_keys_iter.next()
1001
except StopIteration:
1003
# At this point cur_in_key must be >= cur_fixed_key
1004
# step the cur_fixed_key until we pass the cur key, or walk off
1006
while cur_in_key >= cur_fixed_key:
1008
cur_fixed_offset, cur_fixed_key = fixed_keys_iter.next()
1009
except StopIteration:
1012
# We consumed all of the input, nothing more to do
1015
# There was some input left, but we consumed all of fixed, so we
1016
# have to add one more for the tail
1017
cur_keys = [cur_in_key]
1018
cur_keys.extend(in_keys_iter)
1019
cur_out = (len(fixed_keys), cur_keys)
1020
output.append(cur_out)
1023
def iter_entries(self, keys):
1024
"""Iterate over keys within the index.
1026
:param keys: An iterable providing the keys to be retrieved.
1027
:return: An iterable as per iter_all_entries, but restricted to the
1028
keys supplied. No additional keys will be returned, and every
1029
key supplied that is in the index will be returned.
1031
# 6 seconds spent in miss_torture using the sorted() line.
1032
# Even with out of order disk IO it seems faster not to sort it when
1033
# large queries are being made.
1034
# However, now that we are doing multi-way bisecting, we need the keys
1035
# in sorted order anyway. We could change the multi-way code to not
1036
# require sorted order. (For example, it bisects for the first node,
1037
# does an in-order search until a key comes before the current point,
1038
# which it then bisects for, etc.)
1039
keys = frozenset(keys)
1043
if not self.key_count():
1047
if self._leaf_value_cache is None:
1051
value = self._leaf_value_cache.get(key, None)
1052
if value is not None:
1053
# This key is known not to be here, skip it
1055
if self.node_ref_lists:
1056
yield (self, key, value, refs)
1058
yield (self, key, value)
1060
needed_keys.append(key)
1066
# 6 seconds spent in miss_torture using the sorted() line.
1067
# Even with out of order disk IO it seems faster not to sort it when
1068
# large queries are being made.
1069
needed_keys = sorted(needed_keys)
1071
nodes_and_keys = [(0, needed_keys)]
1073
for row_pos, next_row_start in enumerate(self._row_offsets[1:-1]):
1074
node_indexes = [idx for idx, s_keys in nodes_and_keys]
1075
nodes = self._get_internal_nodes(node_indexes)
1077
next_nodes_and_keys = []
1078
for node_index, sub_keys in nodes_and_keys:
1079
node = nodes[node_index]
1080
positions = self._multi_bisect_right(sub_keys, node.keys)
1081
node_offset = next_row_start + node.offset
1082
next_nodes_and_keys.extend([(node_offset + pos, s_keys)
1083
for pos, s_keys in positions])
1084
nodes_and_keys = next_nodes_and_keys
1085
# We should now be at the _LeafNodes
1086
node_indexes = [idx for idx, s_keys in nodes_and_keys]
1088
# TODO: We may *not* want to always read all the nodes in one
1089
# big go. Consider setting a max size on this.
1091
nodes = self._get_leaf_nodes(node_indexes)
1092
for node_index, sub_keys in nodes_and_keys:
1095
node = nodes[node_index]
1096
for next_sub_key in sub_keys:
1097
if next_sub_key in node.keys:
1098
value, refs = node.keys[next_sub_key]
1099
if self.node_ref_lists:
1100
yield (self, next_sub_key, value, refs)
1102
yield (self, next_sub_key, value)
1104
def iter_entries_prefix(self, keys):
1105
"""Iterate over keys within the index using prefix matching.
1107
Prefix matching is applied within the tuple of a key, not to within
1108
the bytestring of each key element. e.g. if you have the keys ('foo',
1109
'bar'), ('foobar', 'gam') and do a prefix search for ('foo', None) then
1110
only the former key is returned.
1112
WARNING: Note that this method currently causes a full index parse
1113
unconditionally (which is reasonably appropriate as it is a means for
1114
thunking many small indices into one larger one and still supplies
1115
iter_all_entries at the thunk layer).
1117
:param keys: An iterable providing the key prefixes to be retrieved.
1118
Each key prefix takes the form of a tuple the length of a key, but
1119
with the last N elements 'None' rather than a regular bytestring.
1120
The first element cannot be 'None'.
1121
:return: An iterable as per iter_all_entries, but restricted to the
1122
keys with a matching prefix to those supplied. No additional keys
1123
will be returned, and every match that is in the index will be
1126
keys = sorted(set(keys))
1129
# Load if needed to check key lengths
1130
if self._key_count is None:
1131
self._get_root_node()
1132
# TODO: only access nodes that can satisfy the prefixes we are looking
1133
# for. For now, to meet API usage (as this function is not used by
1134
# current bzrlib) just suck the entire index and iterate in memory.
1136
if self.node_ref_lists:
1137
if self._key_length == 1:
1138
for _1, key, value, refs in self.iter_all_entries():
1139
nodes[key] = value, refs
1142
for _1, key, value, refs in self.iter_all_entries():
1143
key_value = key, value, refs
1144
# For a key of (foo, bar, baz) create
1145
# _nodes_by_key[foo][bar][baz] = key_value
1146
key_dict = nodes_by_key
1147
for subkey in key[:-1]:
1148
key_dict = key_dict.setdefault(subkey, {})
1149
key_dict[key[-1]] = key_value
1151
if self._key_length == 1:
1152
for _1, key, value in self.iter_all_entries():
1156
for _1, key, value in self.iter_all_entries():
1157
key_value = key, value
1158
# For a key of (foo, bar, baz) create
1159
# _nodes_by_key[foo][bar][baz] = key_value
1160
key_dict = nodes_by_key
1161
for subkey in key[:-1]:
1162
key_dict = key_dict.setdefault(subkey, {})
1163
key_dict[key[-1]] = key_value
1164
if self._key_length == 1:
1168
raise errors.BadIndexKey(key)
1169
if len(key) != self._key_length:
1170
raise errors.BadIndexKey(key)
1172
if self.node_ref_lists:
1173
value, node_refs = nodes[key]
1174
yield self, key, value, node_refs
1176
yield self, key, nodes[key]
1183
raise errors.BadIndexKey(key)
1184
if len(key) != self._key_length:
1185
raise errors.BadIndexKey(key)
1186
# find what it refers to:
1187
key_dict = nodes_by_key
1188
elements = list(key)
1189
# find the subdict whose contents should be returned.
1191
while len(elements) and elements[0] is not None:
1192
key_dict = key_dict[elements[0]]
1195
# a non-existant lookup.
1200
key_dict = dicts.pop(-1)
1201
# can't be empty or would not exist
1202
item, value = key_dict.iteritems().next()
1203
if type(value) == dict:
1205
dicts.extend(key_dict.itervalues())
1208
for value in key_dict.itervalues():
1209
# each value is the key:value:node refs tuple
1211
yield (self, ) + value
1213
# the last thing looked up was a terminal element
1214
yield (self, ) + key_dict
1216
def key_count(self):
1217
"""Return an estimate of the number of keys in this index.
1219
For BTreeGraphIndex the estimate is exact as it is contained in the
1222
if self._key_count is None:
1223
self._get_root_node()
1224
return self._key_count
1226
def _compute_row_offsets(self):
1227
"""Fill out the _row_offsets attribute based on _row_lengths."""
1230
for row in self._row_lengths:
1231
offsets.append(row_offset)
1233
offsets.append(row_offset)
1234
self._row_offsets = offsets
1236
def _parse_header_from_bytes(self, bytes):
1237
"""Parse the header from a region of bytes.
1239
:param bytes: The data to parse.
1240
:return: An offset, data tuple such as readv yields, for the unparsed
1241
data. (which may be of length 0).
1243
signature = bytes[0:len(self._signature())]
1244
if not signature == self._signature():
1245
raise errors.BadIndexFormatSignature(self._name, BTreeGraphIndex)
1246
lines = bytes[len(self._signature()):].splitlines()
1247
options_line = lines[0]
1248
if not options_line.startswith(_OPTION_NODE_REFS):
1249
raise errors.BadIndexOptions(self)
1251
self.node_ref_lists = int(options_line[len(_OPTION_NODE_REFS):])
1253
raise errors.BadIndexOptions(self)
1254
options_line = lines[1]
1255
if not options_line.startswith(_OPTION_KEY_ELEMENTS):
1256
raise errors.BadIndexOptions(self)
1258
self._key_length = int(options_line[len(_OPTION_KEY_ELEMENTS):])
1260
raise errors.BadIndexOptions(self)
1261
options_line = lines[2]
1262
if not options_line.startswith(_OPTION_LEN):
1263
raise errors.BadIndexOptions(self)
1265
self._key_count = int(options_line[len(_OPTION_LEN):])
1267
raise errors.BadIndexOptions(self)
1268
options_line = lines[3]
1269
if not options_line.startswith(_OPTION_ROW_LENGTHS):
1270
raise errors.BadIndexOptions(self)
1272
self._row_lengths = map(int, [length for length in
1273
options_line[len(_OPTION_ROW_LENGTHS):].split(',')
1276
raise errors.BadIndexOptions(self)
1277
self._compute_row_offsets()
1279
# calculate the bytes we have processed
1280
header_end = (len(signature) + sum(map(len, lines[0:4])) + 4)
1281
return header_end, bytes[header_end:]
1283
def _read_nodes(self, nodes):
1284
"""Read some nodes from disk into the LRU cache.
1286
This performs a readv to get the node data into memory, and parses each
1287
node, then yields it to the caller. The nodes are requested in the
1288
supplied order. If possible doing sort() on the list before requesting
1289
a read may improve performance.
1291
:param nodes: The nodes to read. 0 - first node, 1 - second node etc.
1294
# may be the byte string of the whole file
1296
# list of (offset, length) regions of the file that should, evenually
1297
# be read in to data_ranges, either from 'bytes' or from the transport
1300
offset = index * _PAGE_SIZE
1303
# Root node - special case
1305
size = min(_PAGE_SIZE, self._size)
1307
# The only case where we don't know the size, is for very
1308
# small indexes. So we read the whole thing
1309
bytes = self._transport.get_bytes(self._name)
1310
self._size = len(bytes)
1311
# the whole thing should be parsed out of 'bytes'
1312
ranges.append((0, len(bytes)))
1315
if offset > self._size:
1316
raise AssertionError('tried to read past the end'
1317
' of the file %s > %s'
1318
% (offset, self._size))
1319
size = min(size, self._size - offset)
1320
ranges.append((offset, size))
1323
elif bytes is not None:
1324
# already have the whole file
1325
data_ranges = [(start, bytes[start:start+_PAGE_SIZE])
1326
for start in xrange(0, len(bytes), _PAGE_SIZE)]
1327
elif self._file is None:
1328
data_ranges = self._transport.readv(self._name, ranges)
1331
for offset, size in ranges:
1332
self._file.seek(offset)
1333
data_ranges.append((offset, self._file.read(size)))
1334
for offset, data in data_ranges:
1336
# extract the header
1337
offset, data = self._parse_header_from_bytes(data)
1340
bytes = zlib.decompress(data)
1341
if bytes.startswith(_LEAF_FLAG):
1342
node = _LeafNode(bytes, self._key_length, self.node_ref_lists)
1343
elif bytes.startswith(_INTERNAL_FLAG):
1344
node = _InternalNode(bytes)
1346
raise AssertionError("Unknown node type for %r" % bytes)
1347
yield offset / _PAGE_SIZE, node
1349
def _signature(self):
1350
"""The file signature for this index type."""
1354
"""Validate that everything in the index can be accessed."""
1355
# just read and parse every node.
1356
self._get_root_node()
1357
if len(self._row_lengths) > 1:
1358
start_node = self._row_offsets[1]
1360
# We shouldn't be reading anything anyway
1362
node_end = self._row_offsets[-1]
1363
for node in self._read_nodes(range(start_node, node_end)):
1368
from bzrlib import _btree_serializer_c as _btree_serializer
1370
from bzrlib import _btree_serializer_py as _btree_serializer
added
removed
bzrlib/btree_index.py
