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
39
from bzrlib.index import _OPTION_NODE_REFS, _OPTION_KEY_ELEMENTS, _OPTION_LEN
40
from bzrlib.transport import get_transport
43
_BTSIGNATURE = "B+Tree Graph Index 2\n"
44
_OPTION_ROW_LENGTHS = "row_lengths="
45
_LEAF_FLAG = "type=leaf\n"
46
_INTERNAL_FLAG = "type=internal\n"
47
_INTERNAL_OFFSET = "offset="
49
_RESERVED_HEADER_BYTES = 120
52
# 4K per page: 4MB - 1000 entries
53
_NODE_CACHE_SIZE = 1000
56
class _BuilderRow(object):
57
"""The stored state accumulated while writing out a row in the index.
59
:ivar spool: A temporary file used to accumulate nodes for this row
61
:ivar nodes: The count of nodes emitted so far.
65
"""Create a _BuilderRow."""
67
self.spool = tempfile.TemporaryFile()
70
def finish_node(self, pad=True):
71
byte_lines, _, padding = self.writer.finish()
74
self.spool.write("\x00" * _RESERVED_HEADER_BYTES)
76
if not pad and padding:
78
skipped_bytes = padding
79
self.spool.writelines(byte_lines)
80
remainder = (self.spool.tell() + skipped_bytes) % _PAGE_SIZE
82
raise AssertionError("incorrect node length: %d, %d"
83
% (self.spool.tell(), remainder))
88
class _InternalBuilderRow(_BuilderRow):
89
"""The stored state accumulated while writing out internal rows."""
91
def finish_node(self, pad=True):
93
raise AssertionError("Must pad internal nodes only.")
94
_BuilderRow.finish_node(self)
97
class _LeafBuilderRow(_BuilderRow):
98
"""The stored state accumulated while writing out a leaf rows."""
101
class BTreeBuilder(index.GraphIndexBuilder):
102
"""A Builder for B+Tree based Graph indices.
104
The resulting graph has the structure:
106
_SIGNATURE OPTIONS NODES
107
_SIGNATURE := 'B+Tree Graph Index 1' NEWLINE
108
OPTIONS := REF_LISTS KEY_ELEMENTS LENGTH
109
REF_LISTS := 'node_ref_lists=' DIGITS NEWLINE
110
KEY_ELEMENTS := 'key_elements=' DIGITS NEWLINE
111
LENGTH := 'len=' DIGITS NEWLINE
112
ROW_LENGTHS := 'row_lengths' DIGITS (COMMA DIGITS)*
113
NODES := NODE_COMPRESSED*
114
NODE_COMPRESSED:= COMPRESSED_BYTES{4096}
115
NODE_RAW := INTERNAL | LEAF
116
INTERNAL := INTERNAL_FLAG POINTERS
117
LEAF := LEAF_FLAG ROWS
118
KEY_ELEMENT := Not-whitespace-utf8
119
KEY := KEY_ELEMENT (NULL KEY_ELEMENT)*
121
ROW := KEY NULL ABSENT? NULL REFERENCES NULL VALUE NEWLINE
123
REFERENCES := REFERENCE_LIST (TAB REFERENCE_LIST){node_ref_lists - 1}
124
REFERENCE_LIST := (REFERENCE (CR REFERENCE)*)?
126
VALUE := no-newline-no-null-bytes
129
def __init__(self, reference_lists=0, key_elements=1, spill_at=100000):
130
"""See GraphIndexBuilder.__init__.
132
:param spill_at: Optional parameter controlling the maximum number
133
of nodes that BTreeBuilder will hold in memory.
135
index.GraphIndexBuilder.__init__(self, reference_lists=reference_lists,
136
key_elements=key_elements)
137
self._spill_at = spill_at
138
self._backing_indices = []
139
# A map of {key: (node_refs, value)}
141
# Indicate it hasn't been built yet
142
self._nodes_by_key = None
143
self._optimize_for_size = False
145
def add_node(self, key, value, references=()):
146
"""Add a node to the index.
148
If adding the node causes the builder to reach its spill_at threshold,
149
disk spilling will be triggered.
151
:param key: The key. keys are non-empty tuples containing
152
as many whitespace-free utf8 bytestrings as the key length
153
defined for this index.
154
:param references: An iterable of iterables of keys. Each is a
155
reference to another key.
156
:param value: The value to associate with the key. It may be any
157
bytes as long as it does not contain \0 or \n.
159
# we don't care about absent_references
160
node_refs, _ = self._check_key_ref_value(key, references, value)
161
if key in self._nodes:
162
raise errors.BadIndexDuplicateKey(key, self)
163
self._nodes[key] = (node_refs, value)
165
if self._nodes_by_key is not None and self._key_length > 1:
166
self._update_nodes_by_key(key, value, node_refs)
167
if len(self._keys) < self._spill_at:
169
self._spill_mem_keys_to_disk()
171
def _spill_mem_keys_to_disk(self):
172
"""Write the in memory keys down to disk to cap memory consumption.
174
If we already have some keys written to disk, we will combine them so
175
as to preserve the sorted order. The algorithm for combining uses
176
powers of two. So on the first spill, write all mem nodes into a
177
single index. On the second spill, combine the mem nodes with the nodes
178
on disk to create a 2x sized disk index and get rid of the first index.
179
On the third spill, create a single new disk index, which will contain
180
the mem nodes, and preserve the existing 2x sized index. On the fourth,
181
combine mem with the first and second indexes, creating a new one of
182
size 4x. On the fifth create a single new one, etc.
184
if self._combine_backing_indices:
185
(new_backing_file, size,
186
backing_pos) = self._spill_mem_keys_and_combine()
188
new_backing_file, size = self._spill_mem_keys_without_combining()
189
dir_path, base_name = osutils.split(new_backing_file.name)
190
# Note: The transport here isn't strictly needed, because we will use
191
# direct access to the new_backing._file object
192
new_backing = BTreeGraphIndex(get_transport(dir_path),
194
# GC will clean up the file
195
new_backing._file = new_backing_file
196
if self._combine_backing_indices:
197
if len(self._backing_indices) == backing_pos:
198
self._backing_indices.append(None)
199
self._backing_indices[backing_pos] = new_backing
200
for backing_pos in range(backing_pos):
201
self._backing_indices[backing_pos] = None
203
self._backing_indices.append(new_backing)
206
self._nodes_by_key = None
208
def _spill_mem_keys_without_combining(self):
209
return self._write_nodes(self._iter_mem_nodes(), allow_optimize=False)
211
def _spill_mem_keys_and_combine(self):
212
iterators_to_combine = [self._iter_mem_nodes()]
214
for pos, backing in enumerate(self._backing_indices):
218
iterators_to_combine.append(backing.iter_all_entries())
219
backing_pos = pos + 1
220
new_backing_file, size = \
221
self._write_nodes(self._iter_smallest(iterators_to_combine),
222
allow_optimize=False)
223
return new_backing_file, size, backing_pos
225
def add_nodes(self, nodes):
226
"""Add nodes to the index.
228
:param nodes: An iterable of (key, node_refs, value) entries to add.
230
if self.reference_lists:
231
for (key, value, node_refs) in nodes:
232
self.add_node(key, value, node_refs)
234
for (key, value) in nodes:
235
self.add_node(key, value)
237
def _iter_mem_nodes(self):
238
"""Iterate over the nodes held in memory."""
240
if self.reference_lists:
241
for key in sorted(nodes):
242
references, value = nodes[key]
243
yield self, key, value, references
245
for key in sorted(nodes):
246
references, value = nodes[key]
247
yield self, key, value
249
def _iter_smallest(self, iterators_to_combine):
250
if len(iterators_to_combine) == 1:
251
for value in iterators_to_combine[0]:
255
for iterator in iterators_to_combine:
257
current_values.append(iterator.next())
258
except StopIteration:
259
current_values.append(None)
262
# Decorate candidates with the value to allow 2.4's min to be used.
263
candidates = [(item[1][1], item) for item
264
in enumerate(current_values) if item[1] is not None]
265
if not len(candidates):
267
selected = min(candidates)
268
# undecorate back to (pos, node)
269
selected = selected[1]
270
if last == selected[1][1]:
271
raise errors.BadIndexDuplicateKey(last, self)
272
last = selected[1][1]
273
# Yield, with self as the index
274
yield (self,) + selected[1][1:]
277
current_values[pos] = iterators_to_combine[pos].next()
278
except StopIteration:
279
current_values[pos] = None
281
def _add_key(self, string_key, line, rows, allow_optimize=True):
282
"""Add a key to the current chunk.
284
:param string_key: The key to add.
285
:param line: The fully serialised key and value.
286
:param allow_optimize: If set to False, prevent setting the optimize
287
flag when writing out. This is used by the _spill_mem_keys_to_disk
290
if rows[-1].writer is None:
291
# opening a new leaf chunk;
292
for pos, internal_row in enumerate(rows[:-1]):
293
# flesh out any internal nodes that are needed to
294
# preserve the height of the tree
295
if internal_row.writer is None:
297
if internal_row.nodes == 0:
298
length -= _RESERVED_HEADER_BYTES # padded
300
optimize_for_size = self._optimize_for_size
302
optimize_for_size = False
303
internal_row.writer = chunk_writer.ChunkWriter(length, 0,
304
optimize_for_size=optimize_for_size)
305
internal_row.writer.write(_INTERNAL_FLAG)
306
internal_row.writer.write(_INTERNAL_OFFSET +
307
str(rows[pos + 1].nodes) + "\n")
310
if rows[-1].nodes == 0:
311
length -= _RESERVED_HEADER_BYTES # padded
312
rows[-1].writer = chunk_writer.ChunkWriter(length,
313
optimize_for_size=self._optimize_for_size)
314
rows[-1].writer.write(_LEAF_FLAG)
315
if rows[-1].writer.write(line):
316
# this key did not fit in the node:
317
rows[-1].finish_node()
318
key_line = string_key + "\n"
320
for row in reversed(rows[:-1]):
321
# Mark the start of the next node in the node above. If it
322
# doesn't fit then propagate upwards until we find one that
324
if row.writer.write(key_line):
327
# We've found a node that can handle the pointer.
330
# If we reached the current root without being able to mark the
331
# division point, then we need a new root:
334
if 'index' in debug.debug_flags:
335
trace.mutter('Inserting new global row.')
336
new_row = _InternalBuilderRow()
338
rows.insert(0, new_row)
339
# This will be padded, hence the -100
340
new_row.writer = chunk_writer.ChunkWriter(
341
_PAGE_SIZE - _RESERVED_HEADER_BYTES,
343
optimize_for_size=self._optimize_for_size)
344
new_row.writer.write(_INTERNAL_FLAG)
345
new_row.writer.write(_INTERNAL_OFFSET +
346
str(rows[1].nodes - 1) + "\n")
347
new_row.writer.write(key_line)
348
self._add_key(string_key, line, rows, allow_optimize=allow_optimize)
350
def _write_nodes(self, node_iterator, allow_optimize=True):
351
"""Write node_iterator out as a B+Tree.
353
:param node_iterator: An iterator of sorted nodes. Each node should
354
match the output given by iter_all_entries.
355
:param allow_optimize: If set to False, prevent setting the optimize
356
flag when writing out. This is used by the _spill_mem_keys_to_disk
358
:return: A file handle for a temporary file containing a B+Tree for
361
# The index rows - rows[0] is the root, rows[1] is the layer under it
364
# forward sorted by key. In future we may consider topological sorting,
365
# at the cost of table scans for direct lookup, or a second index for
368
# A stack with the number of nodes of each size. 0 is the root node
369
# and must always be 1 (if there are any nodes in the tree).
370
self.row_lengths = []
371
# Loop over all nodes adding them to the bottom row
372
# (rows[-1]). When we finish a chunk in a row,
373
# propagate the key that didn't fit (comes after the chunk) to the
374
# row above, transitively.
375
for node in node_iterator:
377
# First key triggers the first row
378
rows.append(_LeafBuilderRow())
380
string_key, line = _btree_serializer._flatten_node(node,
381
self.reference_lists)
382
self._add_key(string_key, line, rows, allow_optimize=allow_optimize)
383
for row in reversed(rows):
384
pad = (type(row) != _LeafBuilderRow)
385
row.finish_node(pad=pad)
386
result = tempfile.NamedTemporaryFile(prefix='bzr-index-')
387
lines = [_BTSIGNATURE]
388
lines.append(_OPTION_NODE_REFS + str(self.reference_lists) + '\n')
389
lines.append(_OPTION_KEY_ELEMENTS + str(self._key_length) + '\n')
390
lines.append(_OPTION_LEN + str(key_count) + '\n')
391
row_lengths = [row.nodes for row in rows]
392
lines.append(_OPTION_ROW_LENGTHS + ','.join(map(str, row_lengths)) + '\n')
393
result.writelines(lines)
394
position = sum(map(len, lines))
396
if position > _RESERVED_HEADER_BYTES:
397
raise AssertionError("Could not fit the header in the"
398
" reserved space: %d > %d"
399
% (position, _RESERVED_HEADER_BYTES))
400
# write the rows out:
402
reserved = _RESERVED_HEADER_BYTES # reserved space for first node
405
# copy nodes to the finalised file.
406
# Special case the first node as it may be prefixed
407
node = row.spool.read(_PAGE_SIZE)
408
result.write(node[reserved:])
409
result.write("\x00" * (reserved - position))
410
position = 0 # Only the root row actually has an offset
411
copied_len = osutils.pumpfile(row.spool, result)
412
if copied_len != (row.nodes - 1) * _PAGE_SIZE:
413
if type(row) != _LeafBuilderRow:
414
raise AssertionError("Incorrect amount of data copied"
415
" expected: %d, got: %d"
416
% ((row.nodes - 1) * _PAGE_SIZE,
424
"""Finalise the index.
426
:return: A file handle for a temporary file containing the nodes added
429
return self._write_nodes(self.iter_all_entries())[0]
431
def iter_all_entries(self):
432
"""Iterate over all keys within the index
434
:return: An iterable of (index, key, reference_lists, value). There is no
435
defined order for the result iteration - it will be in the most
436
efficient order for the index (in this case dictionary hash order).
438
if 'evil' in debug.debug_flags:
439
trace.mutter_callsite(3,
440
"iter_all_entries scales with size of history.")
441
# Doing serial rather than ordered would be faster; but this shouldn't
442
# be getting called routinely anyway.
443
iterators = [self._iter_mem_nodes()]
444
for backing in self._backing_indices:
445
if backing is not None:
446
iterators.append(backing.iter_all_entries())
447
if len(iterators) == 1:
449
return self._iter_smallest(iterators)
451
def iter_entries(self, keys):
452
"""Iterate over keys within the index.
454
:param keys: An iterable providing the keys to be retrieved.
455
:return: An iterable of (index, key, value, reference_lists). There is no
456
defined order for the result iteration - it will be in the most
457
efficient order for the index (keys iteration order in this case).
460
local_keys = keys.intersection(self._keys)
461
if self.reference_lists:
462
for key in local_keys:
463
node = self._nodes[key]
464
yield self, key, node[1], node[0]
466
for key in local_keys:
467
node = self._nodes[key]
468
yield self, key, node[1]
469
# Find things that are in backing indices that have not been handled
471
if not self._backing_indices:
472
return # We won't find anything there either
473
# Remove all of the keys that we found locally
474
keys.difference_update(local_keys)
475
for backing in self._backing_indices:
480
for node in backing.iter_entries(keys):
482
yield (self,) + node[1:]
484
def iter_entries_prefix(self, keys):
485
"""Iterate over keys within the index using prefix matching.
487
Prefix matching is applied within the tuple of a key, not to within
488
the bytestring of each key element. e.g. if you have the keys ('foo',
489
'bar'), ('foobar', 'gam') and do a prefix search for ('foo', None) then
490
only the former key is returned.
492
:param keys: An iterable providing the key prefixes to be retrieved.
493
Each key prefix takes the form of a tuple the length of a key, but
494
with the last N elements 'None' rather than a regular bytestring.
495
The first element cannot be 'None'.
496
:return: An iterable as per iter_all_entries, but restricted to the
497
keys with a matching prefix to those supplied. No additional keys
498
will be returned, and every match that is in the index will be
501
# XXX: To much duplication with the GraphIndex class; consider finding
502
# a good place to pull out the actual common logic.
506
for backing in self._backing_indices:
509
for node in backing.iter_entries_prefix(keys):
510
yield (self,) + node[1:]
511
if self._key_length == 1:
515
raise errors.BadIndexKey(key)
516
if len(key) != self._key_length:
517
raise errors.BadIndexKey(key)
519
node = self._nodes[key]
522
if self.reference_lists:
523
yield self, key, node[1], node[0]
525
yield self, key, node[1]
530
raise errors.BadIndexKey(key)
531
if len(key) != self._key_length:
532
raise errors.BadIndexKey(key)
533
# find what it refers to:
534
key_dict = self._get_nodes_by_key()
536
# find the subdict to return
538
while len(elements) and elements[0] is not None:
539
key_dict = key_dict[elements[0]]
542
# a non-existant lookup.
547
key_dict = dicts.pop(-1)
548
# can't be empty or would not exist
549
item, value = key_dict.iteritems().next()
550
if type(value) == dict:
552
dicts.extend(key_dict.itervalues())
555
for value in key_dict.itervalues():
556
yield (self, ) + value
558
yield (self, ) + key_dict
560
def _get_nodes_by_key(self):
561
if self._nodes_by_key is None:
563
if self.reference_lists:
564
for key, (references, value) in self._nodes.iteritems():
565
key_dict = nodes_by_key
566
for subkey in key[:-1]:
567
key_dict = key_dict.setdefault(subkey, {})
568
key_dict[key[-1]] = key, value, references
570
for key, (references, value) in self._nodes.iteritems():
571
key_dict = nodes_by_key
572
for subkey in key[:-1]:
573
key_dict = key_dict.setdefault(subkey, {})
574
key_dict[key[-1]] = key, value
575
self._nodes_by_key = nodes_by_key
576
return self._nodes_by_key
579
"""Return an estimate of the number of keys in this index.
581
For InMemoryGraphIndex the estimate is exact.
583
return len(self._keys) + sum(backing.key_count() for backing in
584
self._backing_indices if backing is not None)
587
"""In memory index's have no known corruption at the moment."""
590
class _LeafNode(object):
591
"""A leaf node for a serialised B+Tree index."""
593
__slots__ = ('keys',)
595
def __init__(self, bytes, key_length, ref_list_length):
596
"""Parse bytes to create a leaf node object."""
597
# splitlines mangles the \r delimiters.. don't use it.
598
self.keys = dict(_btree_serializer._parse_leaf_lines(bytes,
599
key_length, ref_list_length))
602
class _InternalNode(object):
603
"""An internal node for a serialised B+Tree index."""
605
__slots__ = ('keys', 'offset')
607
def __init__(self, bytes):
608
"""Parse bytes to create an internal node object."""
609
# splitlines mangles the \r delimiters.. don't use it.
610
self.keys = self._parse_lines(bytes.split('\n'))
612
def _parse_lines(self, lines):
614
self.offset = int(lines[1][7:])
615
for line in lines[2:]:
618
nodes.append(tuple(map(intern, line.split('\0'))))
622
class BTreeGraphIndex(object):
623
"""Access to nodes via the standard GraphIndex interface for B+Tree's.
625
Individual nodes are held in a LRU cache. This holds the root node in
626
memory except when very large walks are done.
629
def __init__(self, transport, name, size):
630
"""Create a B+Tree index object on the index name.
632
:param transport: The transport to read data for the index from.
633
:param name: The file name of the index on transport.
634
:param size: Optional size of the index in bytes. This allows
635
compatibility with the GraphIndex API, as well as ensuring that
636
the initial read (to read the root node header) can be done
637
without over-reading even on empty indices, and on small indices
638
allows single-IO to read the entire index.
640
self._transport = transport
644
self._recommended_pages = self._compute_recommended_pages()
645
self._root_node = None
646
# Default max size is 100,000 leave values
647
self._leaf_value_cache = None # lru_cache.LRUCache(100*1000)
648
self._leaf_node_cache = lru_cache.LRUCache(_NODE_CACHE_SIZE)
649
# We could limit this, but even a 300k record btree has only 3k leaf
650
# nodes, and only 20 internal nodes. So the default of 100 nodes in an
651
# LRU would mean we always cache everything anyway, no need to pay the
653
self._internal_node_cache = fifo_cache.FIFOCache(100)
654
self._key_count = None
655
self._row_lengths = None
656
self._row_offsets = None # Start of each row, [-1] is the end
658
def __eq__(self, other):
659
"""Equal when self and other were created with the same parameters."""
661
type(self) == type(other) and
662
self._transport == other._transport and
663
self._name == other._name and
664
self._size == other._size)
666
def __ne__(self, other):
667
return not self.__eq__(other)
669
def _get_and_cache_nodes(self, nodes):
670
"""Read nodes and cache them in the lru.
672
The nodes list supplied is sorted and then read from disk, each node
673
being inserted it into the _node_cache.
675
Note: Asking for more nodes than the _node_cache can contain will
676
result in some of the results being immediately discarded, to prevent
677
this an assertion is raised if more nodes are asked for than are
680
:return: A dict of {node_pos: node}
683
start_of_leaves = None
684
for node_pos, node in self._read_nodes(sorted(nodes)):
685
if node_pos == 0: # Special case
686
self._root_node = node
688
if start_of_leaves is None:
689
start_of_leaves = self._row_offsets[-2]
690
if node_pos < start_of_leaves:
691
self._internal_node_cache.add(node_pos, node)
693
self._leaf_node_cache.add(node_pos, node)
694
found[node_pos] = node
697
def _compute_recommended_pages(self):
698
"""Convert transport's recommended_page_size into btree pages.
700
recommended_page_size is in bytes, we want to know how many _PAGE_SIZE
701
pages fit in that length.
703
recommended_read = self._transport.recommended_page_size()
704
recommended_pages = int(math.ceil(recommended_read /
706
return recommended_pages
708
def _compute_total_pages_in_index(self):
709
"""How many pages are in the index.
711
If we have read the header we will use the value stored there.
712
Otherwise it will be computed based on the length of the index.
714
if self._size is None:
715
raise AssertionError('_compute_total_pages_in_index should not be'
716
' called when self._size is None')
717
if self._root_node is not None:
718
# This is the number of pages as defined by the header
719
return self._row_offsets[-1]
720
# This is the number of pages as defined by the size of the index. They
721
# should be indentical.
722
total_pages = int(math.ceil(self._size / float(_PAGE_SIZE)))
725
def _expand_offsets(self, offsets):
726
"""Find extra pages to download.
728
The idea is that we always want to make big-enough requests (like 64kB
729
for http), so that we don't waste round trips. So given the entries
730
that we already have cached and the new pages being downloaded figure
731
out what other pages we might want to read.
733
See also doc/developers/btree_index_prefetch.txt for more details.
735
:param offsets: The offsets to be read
736
:return: A list of offsets to download
738
if 'index' in debug.debug_flags:
739
trace.mutter('expanding: %s\toffsets: %s', self._name, offsets)
741
if len(offsets) >= self._recommended_pages:
742
# Don't add more, we are already requesting more than enough
743
if 'index' in debug.debug_flags:
744
trace.mutter(' not expanding large request (%s >= %s)',
745
len(offsets), self._recommended_pages)
747
if self._size is None:
748
# Don't try anything, because we don't know where the file ends
749
if 'index' in debug.debug_flags:
750
trace.mutter(' not expanding without knowing index size')
752
total_pages = self._compute_total_pages_in_index()
753
cached_offsets = self._get_offsets_to_cached_pages()
754
# If reading recommended_pages would read the rest of the index, just
756
if total_pages - len(cached_offsets) <= self._recommended_pages:
757
# Read whatever is left
759
expanded = [x for x in xrange(total_pages)
760
if x not in cached_offsets]
762
expanded = range(total_pages)
763
if 'index' in debug.debug_flags:
764
trace.mutter(' reading all unread pages: %s', expanded)
767
if self._root_node is None:
768
# ATM on the first read of the root node of a large index, we don't
769
# bother pre-reading any other pages. This is because the
770
# likelyhood of actually reading interesting pages is very low.
771
# See doc/developers/btree_index_prefetch.txt for a discussion, and
772
# a possible implementation when we are guessing that the second
773
# layer index is small
774
final_offsets = offsets
776
tree_depth = len(self._row_lengths)
777
if len(cached_offsets) < tree_depth and len(offsets) == 1:
778
# We haven't read enough to justify expansion
779
# If we are only going to read the root node, and 1 leaf node,
780
# then it isn't worth expanding our request. Once we've read at
781
# least 2 nodes, then we are probably doing a search, and we
782
# start expanding our requests.
783
if 'index' in debug.debug_flags:
784
trace.mutter(' not expanding on first reads')
786
final_offsets = self._expand_to_neighbors(offsets, cached_offsets,
789
final_offsets = sorted(final_offsets)
790
if 'index' in debug.debug_flags:
791
trace.mutter('expanded: %s', final_offsets)
794
def _expand_to_neighbors(self, offsets, cached_offsets, total_pages):
795
"""Expand requests to neighbors until we have enough pages.
797
This is called from _expand_offsets after policy has determined that we
799
We only want to expand requests within a given layer. We cheat a little
800
bit and assume all requests will be in the same layer. This is true
801
given the current design, but if it changes this algorithm may perform
804
:param offsets: requested offsets
805
:param cached_offsets: offsets for pages we currently have cached
806
:return: A set() of offsets after expansion
808
final_offsets = set(offsets)
810
new_tips = set(final_offsets)
811
while len(final_offsets) < self._recommended_pages and new_tips:
815
first, end = self._find_layer_first_and_end(pos)
818
and previous not in cached_offsets
819
and previous not in final_offsets
820
and previous >= first):
821
next_tips.add(previous)
823
if (after < total_pages
824
and after not in cached_offsets
825
and after not in final_offsets
828
# This would keep us from going bigger than
829
# recommended_pages by only expanding the first offsets.
830
# However, if we are making a 'wide' request, it is
831
# reasonable to expand all points equally.
832
# if len(final_offsets) > recommended_pages:
834
final_offsets.update(next_tips)
838
def external_references(self, ref_list_num):
839
if self._root_node is None:
840
self._get_root_node()
841
if ref_list_num + 1 > self.node_ref_lists:
842
raise ValueError('No ref list %d, index has %d ref lists'
843
% (ref_list_num, self.node_ref_lists))
846
for node in self.iter_all_entries():
848
refs.update(node[3][ref_list_num])
851
def _find_layer_first_and_end(self, offset):
852
"""Find the start/stop nodes for the layer corresponding to offset.
854
:return: (first, end)
855
first is the first node in this layer
856
end is the first node of the next layer
859
for roffset in self._row_offsets:
866
def _get_offsets_to_cached_pages(self):
867
"""Determine what nodes we already have cached."""
868
cached_offsets = set(self._internal_node_cache.keys())
869
cached_offsets.update(self._leaf_node_cache.keys())
870
if self._root_node is not None:
871
cached_offsets.add(0)
872
return cached_offsets
874
def _get_root_node(self):
875
if self._root_node is None:
876
# We may not have a root node yet
877
self._get_internal_nodes([0])
878
return self._root_node
880
def _get_nodes(self, cache, node_indexes):
883
for idx in node_indexes:
884
if idx == 0 and self._root_node is not None:
885
found[0] = self._root_node
888
found[idx] = cache[idx]
893
needed = self._expand_offsets(needed)
894
found.update(self._get_and_cache_nodes(needed))
897
def _get_internal_nodes(self, node_indexes):
898
"""Get a node, from cache or disk.
900
After getting it, the node will be cached.
902
return self._get_nodes(self._internal_node_cache, node_indexes)
904
def _cache_leaf_values(self, nodes):
905
"""Cache directly from key => value, skipping the btree."""
906
if self._leaf_value_cache is not None:
907
for node in nodes.itervalues():
908
for key, value in node.keys.iteritems():
909
if key in self._leaf_value_cache:
910
# Don't add the rest of the keys, we've seen this node
913
self._leaf_value_cache[key] = value
915
def _get_leaf_nodes(self, node_indexes):
916
"""Get a bunch of nodes, from cache or disk."""
917
found = self._get_nodes(self._leaf_node_cache, node_indexes)
918
self._cache_leaf_values(found)
921
def iter_all_entries(self):
922
"""Iterate over all keys within the index.
924
:return: An iterable of (index, key, value) or (index, key, value, reference_lists).
925
The former tuple is used when there are no reference lists in the
926
index, making the API compatible with simple key:value index types.
927
There is no defined order for the result iteration - it will be in
928
the most efficient order for the index.
930
if 'evil' in debug.debug_flags:
931
trace.mutter_callsite(3,
932
"iter_all_entries scales with size of history.")
933
if not self.key_count():
935
if self._row_offsets[-1] == 1:
936
# There is only the root node, and we read that via key_count()
937
if self.node_ref_lists:
938
for key, (value, refs) in sorted(self._root_node.keys.items()):
939
yield (self, key, value, refs)
941
for key, (value, refs) in sorted(self._root_node.keys.items()):
942
yield (self, key, value)
944
start_of_leaves = self._row_offsets[-2]
945
end_of_leaves = self._row_offsets[-1]
946
needed_offsets = range(start_of_leaves, end_of_leaves)
947
if needed_offsets == [0]:
948
# Special case when we only have a root node, as we have already
950
nodes = [(0, self._root_node)]
952
nodes = self._read_nodes(needed_offsets)
953
# We iterate strictly in-order so that we can use this function
954
# for spilling index builds to disk.
955
if self.node_ref_lists:
956
for _, node in nodes:
957
for key, (value, refs) in sorted(node.keys.items()):
958
yield (self, key, value, refs)
960
for _, node in nodes:
961
for key, (value, refs) in sorted(node.keys.items()):
962
yield (self, key, value)
965
def _multi_bisect_right(in_keys, fixed_keys):
966
"""Find the positions where each 'in_key' would fit in fixed_keys.
968
This is equivalent to doing "bisect_right" on each in_key into
971
:param in_keys: A sorted list of keys to match with fixed_keys
972
:param fixed_keys: A sorted list of keys to match against
973
:return: A list of (integer position, [key list]) tuples.
978
# no pointers in the fixed_keys list, which means everything must
980
return [(0, in_keys)]
982
# TODO: Iterating both lists will generally take M + N steps
983
# Bisecting each key will generally take M * log2 N steps.
984
# If we had an efficient way to compare, we could pick the method
985
# based on which has the fewer number of steps.
986
# There is also the argument that bisect_right is a compiled
987
# function, so there is even more to be gained.
988
# iter_steps = len(in_keys) + len(fixed_keys)
989
# bisect_steps = len(in_keys) * math.log(len(fixed_keys), 2)
990
if len(in_keys) == 1: # Bisect will always be faster for M = 1
991
return [(bisect_right(fixed_keys, in_keys[0]), in_keys)]
992
# elif bisect_steps < iter_steps:
994
# for key in in_keys:
995
# offsets.setdefault(bisect_right(fixed_keys, key),
997
# return [(o, offsets[o]) for o in sorted(offsets)]
998
in_keys_iter = iter(in_keys)
999
fixed_keys_iter = enumerate(fixed_keys)
1000
cur_in_key = in_keys_iter.next()
1001
cur_fixed_offset, cur_fixed_key = fixed_keys_iter.next()
1003
class InputDone(Exception): pass
1004
class FixedDone(Exception): pass
1009
# TODO: Another possibility is that rather than iterating on each side,
1010
# we could use a combination of bisecting and iterating. For
1011
# example, while cur_in_key < fixed_key, bisect to find its
1012
# point, then iterate all matching keys, then bisect (restricted
1013
# to only the remainder) for the next one, etc.
1016
if cur_in_key < cur_fixed_key:
1018
cur_out = (cur_fixed_offset, cur_keys)
1019
output.append(cur_out)
1020
while cur_in_key < cur_fixed_key:
1021
cur_keys.append(cur_in_key)
1023
cur_in_key = in_keys_iter.next()
1024
except StopIteration:
1026
# At this point cur_in_key must be >= cur_fixed_key
1027
# step the cur_fixed_key until we pass the cur key, or walk off
1029
while cur_in_key >= cur_fixed_key:
1031
cur_fixed_offset, cur_fixed_key = fixed_keys_iter.next()
1032
except StopIteration:
1035
# We consumed all of the input, nothing more to do
1038
# There was some input left, but we consumed all of fixed, so we
1039
# have to add one more for the tail
1040
cur_keys = [cur_in_key]
1041
cur_keys.extend(in_keys_iter)
1042
cur_out = (len(fixed_keys), cur_keys)
1043
output.append(cur_out)
1046
def iter_entries(self, keys):
1047
"""Iterate over keys within the index.
1049
:param keys: An iterable providing the keys to be retrieved.
1050
:return: An iterable as per iter_all_entries, but restricted to the
1051
keys supplied. No additional keys will be returned, and every
1052
key supplied that is in the index will be returned.
1054
# 6 seconds spent in miss_torture using the sorted() line.
1055
# Even with out of order disk IO it seems faster not to sort it when
1056
# large queries are being made.
1057
# However, now that we are doing multi-way bisecting, we need the keys
1058
# in sorted order anyway. We could change the multi-way code to not
1059
# require sorted order. (For example, it bisects for the first node,
1060
# does an in-order search until a key comes before the current point,
1061
# which it then bisects for, etc.)
1062
keys = frozenset(keys)
1066
if not self.key_count():
1070
if self._leaf_value_cache is None:
1074
value = self._leaf_value_cache.get(key, None)
1075
if value is not None:
1076
# This key is known not to be here, skip it
1078
if self.node_ref_lists:
1079
yield (self, key, value, refs)
1081
yield (self, key, value)
1083
needed_keys.append(key)
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
needed_keys = sorted(needed_keys)
1094
nodes_and_keys = [(0, needed_keys)]
1096
for row_pos, next_row_start in enumerate(self._row_offsets[1:-1]):
1097
node_indexes = [idx for idx, s_keys in nodes_and_keys]
1098
nodes = self._get_internal_nodes(node_indexes)
1100
next_nodes_and_keys = []
1101
for node_index, sub_keys in nodes_and_keys:
1102
node = nodes[node_index]
1103
positions = self._multi_bisect_right(sub_keys, node.keys)
1104
node_offset = next_row_start + node.offset
1105
next_nodes_and_keys.extend([(node_offset + pos, s_keys)
1106
for pos, s_keys in positions])
1107
nodes_and_keys = next_nodes_and_keys
1108
# We should now be at the _LeafNodes
1109
node_indexes = [idx for idx, s_keys in nodes_and_keys]
1111
# TODO: We may *not* want to always read all the nodes in one
1112
# big go. Consider setting a max size on this.
1114
nodes = self._get_leaf_nodes(node_indexes)
1115
for node_index, sub_keys in nodes_and_keys:
1118
node = nodes[node_index]
1119
for next_sub_key in sub_keys:
1120
if next_sub_key in node.keys:
1121
value, refs = node.keys[next_sub_key]
1122
if self.node_ref_lists:
1123
yield (self, next_sub_key, value, refs)
1125
yield (self, next_sub_key, value)
1127
def iter_entries_prefix(self, keys):
1128
"""Iterate over keys within the index using prefix matching.
1130
Prefix matching is applied within the tuple of a key, not to within
1131
the bytestring of each key element. e.g. if you have the keys ('foo',
1132
'bar'), ('foobar', 'gam') and do a prefix search for ('foo', None) then
1133
only the former key is returned.
1135
WARNING: Note that this method currently causes a full index parse
1136
unconditionally (which is reasonably appropriate as it is a means for
1137
thunking many small indices into one larger one and still supplies
1138
iter_all_entries at the thunk layer).
1140
:param keys: An iterable providing the key prefixes to be retrieved.
1141
Each key prefix takes the form of a tuple the length of a key, but
1142
with the last N elements 'None' rather than a regular bytestring.
1143
The first element cannot be 'None'.
1144
:return: An iterable as per iter_all_entries, but restricted to the
1145
keys with a matching prefix to those supplied. No additional keys
1146
will be returned, and every match that is in the index will be
1149
keys = sorted(set(keys))
1152
# Load if needed to check key lengths
1153
if self._key_count is None:
1154
self._get_root_node()
1155
# TODO: only access nodes that can satisfy the prefixes we are looking
1156
# for. For now, to meet API usage (as this function is not used by
1157
# current bzrlib) just suck the entire index and iterate in memory.
1159
if self.node_ref_lists:
1160
if self._key_length == 1:
1161
for _1, key, value, refs in self.iter_all_entries():
1162
nodes[key] = value, refs
1165
for _1, key, value, refs in self.iter_all_entries():
1166
key_value = key, value, refs
1167
# For a key of (foo, bar, baz) create
1168
# _nodes_by_key[foo][bar][baz] = key_value
1169
key_dict = nodes_by_key
1170
for subkey in key[:-1]:
1171
key_dict = key_dict.setdefault(subkey, {})
1172
key_dict[key[-1]] = key_value
1174
if self._key_length == 1:
1175
for _1, key, value in self.iter_all_entries():
1179
for _1, key, value in self.iter_all_entries():
1180
key_value = key, value
1181
# For a key of (foo, bar, baz) create
1182
# _nodes_by_key[foo][bar][baz] = key_value
1183
key_dict = nodes_by_key
1184
for subkey in key[:-1]:
1185
key_dict = key_dict.setdefault(subkey, {})
1186
key_dict[key[-1]] = key_value
1187
if self._key_length == 1:
1191
raise errors.BadIndexKey(key)
1192
if len(key) != self._key_length:
1193
raise errors.BadIndexKey(key)
1195
if self.node_ref_lists:
1196
value, node_refs = nodes[key]
1197
yield self, key, value, node_refs
1199
yield self, key, nodes[key]
1206
raise errors.BadIndexKey(key)
1207
if len(key) != self._key_length:
1208
raise errors.BadIndexKey(key)
1209
# find what it refers to:
1210
key_dict = nodes_by_key
1211
elements = list(key)
1212
# find the subdict whose contents should be returned.
1214
while len(elements) and elements[0] is not None:
1215
key_dict = key_dict[elements[0]]
1218
# a non-existant lookup.
1223
key_dict = dicts.pop(-1)
1224
# can't be empty or would not exist
1225
item, value = key_dict.iteritems().next()
1226
if type(value) == dict:
1228
dicts.extend(key_dict.itervalues())
1231
for value in key_dict.itervalues():
1232
# each value is the key:value:node refs tuple
1234
yield (self, ) + value
1236
# the last thing looked up was a terminal element
1237
yield (self, ) + key_dict
1239
def key_count(self):
1240
"""Return an estimate of the number of keys in this index.
1242
For BTreeGraphIndex the estimate is exact as it is contained in the
1245
if self._key_count is None:
1246
self._get_root_node()
1247
return self._key_count
1249
def _compute_row_offsets(self):
1250
"""Fill out the _row_offsets attribute based on _row_lengths."""
1253
for row in self._row_lengths:
1254
offsets.append(row_offset)
1256
offsets.append(row_offset)
1257
self._row_offsets = offsets
1259
def _parse_header_from_bytes(self, bytes):
1260
"""Parse the header from a region of bytes.
1262
:param bytes: The data to parse.
1263
:return: An offset, data tuple such as readv yields, for the unparsed
1264
data. (which may be of length 0).
1266
signature = bytes[0:len(self._signature())]
1267
if not signature == self._signature():
1268
raise errors.BadIndexFormatSignature(self._name, BTreeGraphIndex)
1269
lines = bytes[len(self._signature()):].splitlines()
1270
options_line = lines[0]
1271
if not options_line.startswith(_OPTION_NODE_REFS):
1272
raise errors.BadIndexOptions(self)
1274
self.node_ref_lists = int(options_line[len(_OPTION_NODE_REFS):])
1276
raise errors.BadIndexOptions(self)
1277
options_line = lines[1]
1278
if not options_line.startswith(_OPTION_KEY_ELEMENTS):
1279
raise errors.BadIndexOptions(self)
1281
self._key_length = int(options_line[len(_OPTION_KEY_ELEMENTS):])
1283
raise errors.BadIndexOptions(self)
1284
options_line = lines[2]
1285
if not options_line.startswith(_OPTION_LEN):
1286
raise errors.BadIndexOptions(self)
1288
self._key_count = int(options_line[len(_OPTION_LEN):])
1290
raise errors.BadIndexOptions(self)
1291
options_line = lines[3]
1292
if not options_line.startswith(_OPTION_ROW_LENGTHS):
1293
raise errors.BadIndexOptions(self)
1295
self._row_lengths = map(int, [length for length in
1296
options_line[len(_OPTION_ROW_LENGTHS):].split(',')
1299
raise errors.BadIndexOptions(self)
1300
self._compute_row_offsets()
1302
# calculate the bytes we have processed
1303
header_end = (len(signature) + sum(map(len, lines[0:4])) + 4)
1304
return header_end, bytes[header_end:]
1306
def _read_nodes(self, nodes):
1307
"""Read some nodes from disk into the LRU cache.
1309
This performs a readv to get the node data into memory, and parses each
1310
node, then yields it to the caller. The nodes are requested in the
1311
supplied order. If possible doing sort() on the list before requesting
1312
a read may improve performance.
1314
:param nodes: The nodes to read. 0 - first node, 1 - second node etc.
1317
# may be the byte string of the whole file
1319
# list of (offset, length) regions of the file that should, evenually
1320
# be read in to data_ranges, either from 'bytes' or from the transport
1323
offset = index * _PAGE_SIZE
1326
# Root node - special case
1328
size = min(_PAGE_SIZE, self._size)
1330
# The only case where we don't know the size, is for very
1331
# small indexes. So we read the whole thing
1332
bytes = self._transport.get_bytes(self._name)
1333
self._size = len(bytes)
1334
# the whole thing should be parsed out of 'bytes'
1335
ranges.append((0, len(bytes)))
1338
if offset > self._size:
1339
raise AssertionError('tried to read past the end'
1340
' of the file %s > %s'
1341
% (offset, self._size))
1342
size = min(size, self._size - offset)
1343
ranges.append((offset, size))
1346
elif bytes is not None:
1347
# already have the whole file
1348
data_ranges = [(start, bytes[start:start+_PAGE_SIZE])
1349
for start in xrange(0, len(bytes), _PAGE_SIZE)]
1350
elif self._file is None:
1351
data_ranges = self._transport.readv(self._name, ranges)
1354
for offset, size in ranges:
1355
self._file.seek(offset)
1356
data_ranges.append((offset, self._file.read(size)))
1357
for offset, data in data_ranges:
1359
# extract the header
1360
offset, data = self._parse_header_from_bytes(data)
1363
bytes = zlib.decompress(data)
1364
if bytes.startswith(_LEAF_FLAG):
1365
node = _LeafNode(bytes, self._key_length, self.node_ref_lists)
1366
elif bytes.startswith(_INTERNAL_FLAG):
1367
node = _InternalNode(bytes)
1369
raise AssertionError("Unknown node type for %r" % bytes)
1370
yield offset / _PAGE_SIZE, node
1372
def _signature(self):
1373
"""The file signature for this index type."""
1377
"""Validate that everything in the index can be accessed."""
1378
# just read and parse every node.
1379
self._get_root_node()
1380
if len(self._row_lengths) > 1:
1381
start_node = self._row_offsets[1]
1383
# We shouldn't be reading anything anyway
1385
node_end = self._row_offsets[-1]
1386
for node in self._read_nodes(range(start_node, node_end)):
1391
from bzrlib import _btree_serializer_c as _btree_serializer
1393
from bzrlib import _btree_serializer_py as _btree_serializer
added
removed
bzrlib/btree_index.py
