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., 59 Temple Place, Suite 330, Boston, MA 02111-1307 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
dir_path, base_name = osutils.split(new_backing_file.name)
194
# Note: The transport here isn't strictly needed, because we will use
195
# direct access to the new_backing._file object
196
new_backing = BTreeGraphIndex(get_transport(dir_path),
198
# GC will clean up the file
199
new_backing._file = new_backing_file
200
if len(self._backing_indices) == backing_pos:
201
self._backing_indices.append(None)
202
self._backing_indices[backing_pos] = new_backing
203
for pos in range(backing_pos):
204
self._backing_indices[pos] = None
207
self._nodes_by_key = None
209
def add_nodes(self, nodes):
210
"""Add nodes to the index.
212
:param nodes: An iterable of (key, node_refs, value) entries to add.
214
if self.reference_lists:
215
for (key, value, node_refs) in nodes:
216
self.add_node(key, value, node_refs)
218
for (key, value) in nodes:
219
self.add_node(key, value)
221
def _iter_mem_nodes(self):
222
"""Iterate over the nodes held in memory."""
224
if self.reference_lists:
225
for key in sorted(nodes):
226
references, value = nodes[key]
227
yield self, key, value, references
229
for key in sorted(nodes):
230
references, value = nodes[key]
231
yield self, key, value
233
def _iter_smallest(self, iterators_to_combine):
234
if len(iterators_to_combine) == 1:
235
for value in iterators_to_combine[0]:
239
for iterator in iterators_to_combine:
241
current_values.append(iterator.next())
242
except StopIteration:
243
current_values.append(None)
246
# Decorate candidates with the value to allow 2.4's min to be used.
247
candidates = [(item[1][1], item) for item
248
in enumerate(current_values) if item[1] is not None]
249
if not len(candidates):
251
selected = min(candidates)
252
# undecorate back to (pos, node)
253
selected = selected[1]
254
if last == selected[1][1]:
255
raise errors.BadIndexDuplicateKey(last, self)
256
last = selected[1][1]
257
# Yield, with self as the index
258
yield (self,) + selected[1][1:]
261
current_values[pos] = iterators_to_combine[pos].next()
262
except StopIteration:
263
current_values[pos] = None
265
def _add_key(self, string_key, line, rows):
266
"""Add a key to the current chunk.
268
:param string_key: The key to add.
269
:param line: The fully serialised key and value.
271
if rows[-1].writer is None:
272
# opening a new leaf chunk;
273
for pos, internal_row in enumerate(rows[:-1]):
274
# flesh out any internal nodes that are needed to
275
# preserve the height of the tree
276
if internal_row.writer is None:
278
if internal_row.nodes == 0:
279
length -= _RESERVED_HEADER_BYTES # padded
280
internal_row.writer = chunk_writer.ChunkWriter(length, 0,
281
optimize_for_size=self._optimize_for_size)
282
internal_row.writer.write(_INTERNAL_FLAG)
283
internal_row.writer.write(_INTERNAL_OFFSET +
284
str(rows[pos + 1].nodes) + "\n")
287
if rows[-1].nodes == 0:
288
length -= _RESERVED_HEADER_BYTES # padded
289
rows[-1].writer = chunk_writer.ChunkWriter(length,
290
optimize_for_size=self._optimize_for_size)
291
rows[-1].writer.write(_LEAF_FLAG)
292
if rows[-1].writer.write(line):
293
# this key did not fit in the node:
294
rows[-1].finish_node()
295
key_line = string_key + "\n"
297
for row in reversed(rows[:-1]):
298
# Mark the start of the next node in the node above. If it
299
# doesn't fit then propogate upwards until we find one that
301
if row.writer.write(key_line):
304
# We've found a node that can handle the pointer.
307
# If we reached the current root without being able to mark the
308
# division point, then we need a new root:
311
if 'index' in debug.debug_flags:
312
trace.mutter('Inserting new global row.')
313
new_row = _InternalBuilderRow()
315
rows.insert(0, new_row)
316
# This will be padded, hence the -100
317
new_row.writer = chunk_writer.ChunkWriter(
318
_PAGE_SIZE - _RESERVED_HEADER_BYTES,
320
optimize_for_size=self._optimize_for_size)
321
new_row.writer.write(_INTERNAL_FLAG)
322
new_row.writer.write(_INTERNAL_OFFSET +
323
str(rows[1].nodes - 1) + "\n")
324
new_row.writer.write(key_line)
325
self._add_key(string_key, line, rows)
327
def _write_nodes(self, node_iterator):
328
"""Write node_iterator out as a B+Tree.
330
:param node_iterator: An iterator of sorted nodes. Each node should
331
match the output given by iter_all_entries.
332
:return: A file handle for a temporary file containing a B+Tree for
335
# The index rows - rows[0] is the root, rows[1] is the layer under it
338
# forward sorted by key. In future we may consider topological sorting,
339
# at the cost of table scans for direct lookup, or a second index for
342
# A stack with the number of nodes of each size. 0 is the root node
343
# and must always be 1 (if there are any nodes in the tree).
344
self.row_lengths = []
345
# Loop over all nodes adding them to the bottom row
346
# (rows[-1]). When we finish a chunk in a row,
347
# propogate the key that didn't fit (comes after the chunk) to the
348
# row above, transitively.
349
for node in node_iterator:
351
# First key triggers the first row
352
rows.append(_LeafBuilderRow())
354
string_key, line = _btree_serializer._flatten_node(node,
355
self.reference_lists)
356
self._add_key(string_key, line, rows)
357
for row in reversed(rows):
358
pad = (type(row) != _LeafBuilderRow)
359
row.finish_node(pad=pad)
360
result = tempfile.NamedTemporaryFile()
361
lines = [_BTSIGNATURE]
362
lines.append(_OPTION_NODE_REFS + str(self.reference_lists) + '\n')
363
lines.append(_OPTION_KEY_ELEMENTS + str(self._key_length) + '\n')
364
lines.append(_OPTION_LEN + str(key_count) + '\n')
365
row_lengths = [row.nodes for row in rows]
366
lines.append(_OPTION_ROW_LENGTHS + ','.join(map(str, row_lengths)) + '\n')
367
result.writelines(lines)
368
position = sum(map(len, lines))
370
if position > _RESERVED_HEADER_BYTES:
371
raise AssertionError("Could not fit the header in the"
372
" reserved space: %d > %d"
373
% (position, _RESERVED_HEADER_BYTES))
374
# write the rows out:
376
reserved = _RESERVED_HEADER_BYTES # reserved space for first node
379
# copy nodes to the finalised file.
380
# Special case the first node as it may be prefixed
381
node = row.spool.read(_PAGE_SIZE)
382
result.write(node[reserved:])
383
result.write("\x00" * (reserved - position))
384
position = 0 # Only the root row actually has an offset
385
copied_len = osutils.pumpfile(row.spool, result)
386
if copied_len != (row.nodes - 1) * _PAGE_SIZE:
387
if type(row) != _LeafBuilderRow:
388
raise AssertionError("Incorrect amount of data copied"
389
" expected: %d, got: %d"
390
% ((row.nodes - 1) * _PAGE_SIZE,
398
"""Finalise the index.
400
:return: A file handle for a temporary file containing the nodes added
403
return self._write_nodes(self.iter_all_entries())[0]
405
def iter_all_entries(self):
406
"""Iterate over all keys within the index
408
:return: An iterable of (index, key, reference_lists, value). There is no
409
defined order for the result iteration - it will be in the most
410
efficient order for the index (in this case dictionary hash order).
412
if 'evil' in debug.debug_flags:
413
trace.mutter_callsite(3,
414
"iter_all_entries scales with size of history.")
415
# Doing serial rather than ordered would be faster; but this shouldn't
416
# be getting called routinely anyway.
417
iterators = [self._iter_mem_nodes()]
418
for backing in self._backing_indices:
419
if backing is not None:
420
iterators.append(backing.iter_all_entries())
421
if len(iterators) == 1:
423
return self._iter_smallest(iterators)
425
def iter_entries(self, keys):
426
"""Iterate over keys within the index.
428
:param keys: An iterable providing the keys to be retrieved.
429
:return: An iterable of (index, key, value, reference_lists). There is no
430
defined order for the result iteration - it will be in the most
431
efficient order for the index (keys iteration order in this case).
434
if self.reference_lists:
435
for key in keys.intersection(self._keys):
436
node = self._nodes[key]
437
yield self, key, node[1], node[0]
439
for key in keys.intersection(self._keys):
440
node = self._nodes[key]
441
yield self, key, node[1]
442
keys.difference_update(self._keys)
443
for backing in self._backing_indices:
448
for node in backing.iter_entries(keys):
450
yield (self,) + node[1:]
452
def iter_entries_prefix(self, keys):
453
"""Iterate over keys within the index using prefix matching.
455
Prefix matching is applied within the tuple of a key, not to within
456
the bytestring of each key element. e.g. if you have the keys ('foo',
457
'bar'), ('foobar', 'gam') and do a prefix search for ('foo', None) then
458
only the former key is returned.
460
:param keys: An iterable providing the key prefixes to be retrieved.
461
Each key prefix takes the form of a tuple the length of a key, but
462
with the last N elements 'None' rather than a regular bytestring.
463
The first element cannot be 'None'.
464
:return: An iterable as per iter_all_entries, but restricted to the
465
keys with a matching prefix to those supplied. No additional keys
466
will be returned, and every match that is in the index will be
469
# XXX: To much duplication with the GraphIndex class; consider finding
470
# a good place to pull out the actual common logic.
474
for backing in self._backing_indices:
477
for node in backing.iter_entries_prefix(keys):
478
yield (self,) + node[1:]
479
if self._key_length == 1:
483
raise errors.BadIndexKey(key)
484
if len(key) != self._key_length:
485
raise errors.BadIndexKey(key)
487
node = self._nodes[key]
490
if self.reference_lists:
491
yield self, key, node[1], node[0]
493
yield self, key, node[1]
498
raise errors.BadIndexKey(key)
499
if len(key) != self._key_length:
500
raise errors.BadIndexKey(key)
501
# find what it refers to:
502
key_dict = self._get_nodes_by_key()
504
# find the subdict to return
506
while len(elements) and elements[0] is not None:
507
key_dict = key_dict[elements[0]]
510
# a non-existant lookup.
515
key_dict = dicts.pop(-1)
516
# can't be empty or would not exist
517
item, value = key_dict.iteritems().next()
518
if type(value) == dict:
520
dicts.extend(key_dict.itervalues())
523
for value in key_dict.itervalues():
524
yield (self, ) + value
526
yield (self, ) + key_dict
528
def _get_nodes_by_key(self):
529
if self._nodes_by_key is None:
531
if self.reference_lists:
532
for key, (references, value) in self._nodes.iteritems():
533
key_dict = nodes_by_key
534
for subkey in key[:-1]:
535
key_dict = key_dict.setdefault(subkey, {})
536
key_dict[key[-1]] = key, value, references
538
for key, (references, value) in self._nodes.iteritems():
539
key_dict = nodes_by_key
540
for subkey in key[:-1]:
541
key_dict = key_dict.setdefault(subkey, {})
542
key_dict[key[-1]] = key, value
543
self._nodes_by_key = nodes_by_key
544
return self._nodes_by_key
547
"""Return an estimate of the number of keys in this index.
549
For InMemoryGraphIndex the estimate is exact.
551
return len(self._keys) + sum(backing.key_count() for backing in
552
self._backing_indices if backing is not None)
555
"""In memory index's have no known corruption at the moment."""
558
class _LeafNode(object):
559
"""A leaf node for a serialised B+Tree index."""
561
def __init__(self, bytes, key_length, ref_list_length):
562
"""Parse bytes to create a leaf node object."""
563
# splitlines mangles the \r delimiters.. don't use it.
564
self.keys = dict(_btree_serializer._parse_leaf_lines(bytes,
565
key_length, ref_list_length))
568
class _InternalNode(object):
569
"""An internal node for a serialised B+Tree index."""
571
def __init__(self, bytes):
572
"""Parse bytes to create an internal node object."""
573
# splitlines mangles the \r delimiters.. don't use it.
574
self.keys = self._parse_lines(bytes.split('\n'))
576
def _parse_lines(self, lines):
578
self.offset = int(lines[1][7:])
579
for line in lines[2:]:
582
nodes.append(tuple(line.split('\0')))
586
class BTreeGraphIndex(object):
587
"""Access to nodes via the standard GraphIndex interface for B+Tree's.
589
Individual nodes are held in a LRU cache. This holds the root node in
590
memory except when very large walks are done.
593
def __init__(self, transport, name, size):
594
"""Create a B+Tree index object on the index name.
596
:param transport: The transport to read data for the index from.
597
:param name: The file name of the index on transport.
598
:param size: Optional size of the index in bytes. This allows
599
compatibility with the GraphIndex API, as well as ensuring that
600
the initial read (to read the root node header) can be done
601
without over-reading even on empty indices, and on small indices
602
allows single-IO to read the entire index.
604
self._transport = transport
608
self._recommended_pages = self._compute_recommended_pages()
609
self._root_node = None
610
# Default max size is 100,000 leave values
611
self._leaf_value_cache = None # lru_cache.LRUCache(100*1000)
612
self._leaf_node_cache = lru_cache.LRUCache(_NODE_CACHE_SIZE)
613
self._internal_node_cache = lru_cache.LRUCache()
614
self._key_count = None
615
self._row_lengths = None
616
self._row_offsets = None # Start of each row, [-1] is the end
618
def __eq__(self, other):
619
"""Equal when self and other were created with the same parameters."""
621
type(self) == type(other) and
622
self._transport == other._transport and
623
self._name == other._name and
624
self._size == other._size)
626
def __ne__(self, other):
627
return not self.__eq__(other)
629
def _get_and_cache_nodes(self, nodes):
630
"""Read nodes and cache them in the lru.
632
The nodes list supplied is sorted and then read from disk, each node
633
being inserted it into the _node_cache.
635
Note: Asking for more nodes than the _node_cache can contain will
636
result in some of the results being immediately discarded, to prevent
637
this an assertion is raised if more nodes are asked for than are
640
:return: A dict of {node_pos: node}
643
start_of_leaves = None
644
for node_pos, node in self._read_nodes(sorted(nodes)):
645
if node_pos == 0: # Special case
646
self._root_node = node
648
if start_of_leaves is None:
649
start_of_leaves = self._row_offsets[-2]
650
if node_pos < start_of_leaves:
651
self._internal_node_cache.add(node_pos, node)
653
self._leaf_node_cache.add(node_pos, node)
654
found[node_pos] = node
657
def _compute_recommended_pages(self):
658
"""Convert transport's recommended_page_size into btree pages.
660
recommended_page_size is in bytes, we want to know how many _PAGE_SIZE
661
pages fit in that length.
663
recommended_read = self._transport.recommended_page_size()
664
recommended_pages = int(math.ceil(recommended_read /
666
return recommended_pages
668
def _compute_total_pages_in_index(self):
669
"""How many pages are in the index.
671
If we have read the header we will use the value stored there.
672
Otherwise it will be computed based on the length of the index.
674
if self._size is None:
675
raise AssertionError('_compute_total_pages_in_index should not be'
676
' called when self._size is None')
677
if self._root_node is not None:
678
# This is the number of pages as defined by the header
679
return self._row_offsets[-1]
680
# This is the number of pages as defined by the size of the index. They
681
# should be indentical.
682
total_pages = int(math.ceil(self._size / float(_PAGE_SIZE)))
685
def _expand_offsets(self, offsets):
686
"""Find extra pages to download.
688
The idea is that we always want to make big-enough requests (like 64kB
689
for http), so that we don't waste round trips. So given the entries
690
that we already have cached and the new pages being downloaded figure
691
out what other pages we might want to read.
693
See also doc/developers/btree_index_prefetch.txt for more details.
695
:param offsets: The offsets to be read
696
:return: A list of offsets to download
698
if 'index' in debug.debug_flags:
699
trace.mutter('expanding: %s\toffsets: %s', self._name, offsets)
701
if len(offsets) >= self._recommended_pages:
702
# Don't add more, we are already requesting more than enough
703
if 'index' in debug.debug_flags:
704
trace.mutter(' not expanding large request (%s >= %s)',
705
len(offsets), self._recommended_pages)
707
if self._size is None:
708
# Don't try anything, because we don't know where the file ends
709
if 'index' in debug.debug_flags:
710
trace.mutter(' not expanding without knowing index size')
712
total_pages = self._compute_total_pages_in_index()
713
cached_offsets = self._get_offsets_to_cached_pages()
714
# If reading recommended_pages would read the rest of the index, just
716
if total_pages - len(cached_offsets) <= self._recommended_pages:
717
# Read whatever is left
719
expanded = [x for x in xrange(total_pages)
720
if x not in cached_offsets]
722
expanded = range(total_pages)
723
if 'index' in debug.debug_flags:
724
trace.mutter(' reading all unread pages: %s', expanded)
727
if self._root_node is None:
728
# ATM on the first read of the root node of a large index, we don't
729
# bother pre-reading any other pages. This is because the
730
# likelyhood of actually reading interesting pages is very low.
731
# See doc/developers/btree_index_prefetch.txt for a discussion, and
732
# a possible implementation when we are guessing that the second
733
# layer index is small
734
final_offsets = offsets
736
tree_depth = len(self._row_lengths)
737
if len(cached_offsets) < tree_depth and len(offsets) == 1:
738
# We haven't read enough to justify expansion
739
# If we are only going to read the root node, and 1 leaf node,
740
# then it isn't worth expanding our request. Once we've read at
741
# least 2 nodes, then we are probably doing a search, and we
742
# start expanding our requests.
743
if 'index' in debug.debug_flags:
744
trace.mutter(' not expanding on first reads')
746
final_offsets = self._expand_to_neighbors(offsets, cached_offsets,
749
final_offsets = sorted(final_offsets)
750
if 'index' in debug.debug_flags:
751
trace.mutter('expanded: %s', final_offsets)
754
def _expand_to_neighbors(self, offsets, cached_offsets, total_pages):
755
"""Expand requests to neighbors until we have enough pages.
757
This is called from _expand_offsets after policy has determined that we
759
We only want to expand requests within a given layer. We cheat a little
760
bit and assume all requests will be in the same layer. This is true
761
given the current design, but if it changes this algorithm may perform
764
:param offsets: requested offsets
765
:param cached_offsets: offsets for pages we currently have cached
766
:return: A set() of offsets after expansion
768
final_offsets = set(offsets)
770
new_tips = set(final_offsets)
771
while len(final_offsets) < self._recommended_pages and new_tips:
775
first, end = self._find_layer_first_and_end(pos)
778
and previous not in cached_offsets
779
and previous not in final_offsets
780
and previous >= first):
781
next_tips.add(previous)
783
if (after < total_pages
784
and after not in cached_offsets
785
and after not in final_offsets
788
# This would keep us from going bigger than
789
# recommended_pages by only expanding the first offsets.
790
# However, if we are making a 'wide' request, it is
791
# reasonable to expand all points equally.
792
# if len(final_offsets) > recommended_pages:
794
final_offsets.update(next_tips)
798
def _find_layer_first_and_end(self, offset):
799
"""Find the start/stop nodes for the layer corresponding to offset.
801
:return: (first, end)
802
first is the first node in this layer
803
end is the first node of the next layer
806
for roffset in self._row_offsets:
813
def _get_offsets_to_cached_pages(self):
814
"""Determine what nodes we already have cached."""
815
cached_offsets = set(self._internal_node_cache.keys())
816
cached_offsets.update(self._leaf_node_cache.keys())
817
if self._root_node is not None:
818
cached_offsets.add(0)
819
return cached_offsets
821
def _get_root_node(self):
822
if self._root_node is None:
823
# We may not have a root node yet
824
self._get_internal_nodes([0])
825
return self._root_node
827
def _get_nodes(self, cache, node_indexes):
830
for idx in node_indexes:
831
if idx == 0 and self._root_node is not None:
832
found[0] = self._root_node
835
found[idx] = cache[idx]
840
needed = self._expand_offsets(needed)
841
found.update(self._get_and_cache_nodes(needed))
844
def _get_internal_nodes(self, node_indexes):
845
"""Get a node, from cache or disk.
847
After getting it, the node will be cached.
849
return self._get_nodes(self._internal_node_cache, node_indexes)
851
def _get_leaf_nodes(self, node_indexes):
852
"""Get a bunch of nodes, from cache or disk."""
853
found = self._get_nodes(self._leaf_node_cache, node_indexes)
854
if self._leaf_value_cache is not None:
855
for node in found.itervalues():
856
for key, value in node.keys.iteritems():
857
if key in self._leaf_value_cache:
858
# Don't add the rest of the keys, we've seen this node
861
self._leaf_value_cache[key] = value
864
def iter_all_entries(self):
865
"""Iterate over all keys within the index.
867
:return: An iterable of (index, key, value) or (index, key, value, reference_lists).
868
The former tuple is used when there are no reference lists in the
869
index, making the API compatible with simple key:value index types.
870
There is no defined order for the result iteration - it will be in
871
the most efficient order for the index.
873
if 'evil' in debug.debug_flags:
874
trace.mutter_callsite(3,
875
"iter_all_entries scales with size of history.")
876
if not self.key_count():
878
start_of_leaves = self._row_offsets[-2]
879
end_of_leaves = self._row_offsets[-1]
880
needed_nodes = range(start_of_leaves, end_of_leaves)
881
# We iterate strictly in-order so that we can use this function
882
# for spilling index builds to disk.
883
if self.node_ref_lists:
884
for _, node in self._read_nodes(needed_nodes):
885
for key, (value, refs) in sorted(node.keys.items()):
886
yield (self, key, value, refs)
888
for _, node in self._read_nodes(needed_nodes):
889
for key, (value, refs) in sorted(node.keys.items()):
890
yield (self, key, value)
893
def _multi_bisect_right(in_keys, fixed_keys):
894
"""Find the positions where each 'in_key' would fit in fixed_keys.
896
This is equivalent to doing "bisect_right" on each in_key into
899
:param in_keys: A sorted list of keys to match with fixed_keys
900
:param fixed_keys: A sorted list of keys to match against
901
:return: A list of (integer position, [key list]) tuples.
906
# no pointers in the fixed_keys list, which means everything must
908
return [(0, in_keys)]
910
# TODO: Iterating both lists will generally take M + N steps
911
# Bisecting each key will generally take M * log2 N steps.
912
# If we had an efficient way to compare, we could pick the method
913
# based on which has the fewer number of steps.
914
# There is also the argument that bisect_right is a compiled
915
# function, so there is even more to be gained.
916
# iter_steps = len(in_keys) + len(fixed_keys)
917
# bisect_steps = len(in_keys) * math.log(len(fixed_keys), 2)
918
if len(in_keys) == 1: # Bisect will always be faster for M = 1
919
return [(bisect_right(fixed_keys, in_keys[0]), in_keys)]
920
# elif bisect_steps < iter_steps:
922
# for key in in_keys:
923
# offsets.setdefault(bisect_right(fixed_keys, key),
925
# return [(o, offsets[o]) for o in sorted(offsets)]
926
in_keys_iter = iter(in_keys)
927
fixed_keys_iter = enumerate(fixed_keys)
928
cur_in_key = in_keys_iter.next()
929
cur_fixed_offset, cur_fixed_key = fixed_keys_iter.next()
931
class InputDone(Exception): pass
932
class FixedDone(Exception): pass
937
# TODO: Another possibility is that rather than iterating on each side,
938
# we could use a combination of bisecting and iterating. For
939
# example, while cur_in_key < fixed_key, bisect to find its
940
# point, then iterate all matching keys, then bisect (restricted
941
# to only the remainder) for the next one, etc.
944
if cur_in_key < cur_fixed_key:
946
cur_out = (cur_fixed_offset, cur_keys)
947
output.append(cur_out)
948
while cur_in_key < cur_fixed_key:
949
cur_keys.append(cur_in_key)
951
cur_in_key = in_keys_iter.next()
952
except StopIteration:
954
# At this point cur_in_key must be >= cur_fixed_key
955
# step the cur_fixed_key until we pass the cur key, or walk off
957
while cur_in_key >= cur_fixed_key:
959
cur_fixed_offset, cur_fixed_key = fixed_keys_iter.next()
960
except StopIteration:
963
# We consumed all of the input, nothing more to do
966
# There was some input left, but we consumed all of fixed, so we
967
# have to add one more for the tail
968
cur_keys = [cur_in_key]
969
cur_keys.extend(in_keys_iter)
970
cur_out = (len(fixed_keys), cur_keys)
971
output.append(cur_out)
974
def iter_entries(self, keys):
975
"""Iterate over keys within the index.
977
:param keys: An iterable providing the keys to be retrieved.
978
:return: An iterable as per iter_all_entries, but restricted to the
979
keys supplied. No additional keys will be returned, and every
980
key supplied that is in the index will be returned.
982
# 6 seconds spent in miss_torture using the sorted() line.
983
# Even with out of order disk IO it seems faster not to sort it when
984
# large queries are being made.
985
# However, now that we are doing multi-way bisecting, we need the keys
986
# in sorted order anyway. We could change the multi-way code to not
987
# require sorted order. (For example, it bisects for the first node,
988
# does an in-order search until a key comes before the current point,
989
# which it then bisects for, etc.)
990
keys = frozenset(keys)
994
if not self.key_count():
998
if self._leaf_value_cache is None:
1002
value = self._leaf_value_cache.get(key, None)
1003
if value is not None:
1004
# This key is known not to be here, skip it
1006
if self.node_ref_lists:
1007
yield (self, key, value, refs)
1009
yield (self, key, value)
1011
needed_keys.append(key)
1017
# 6 seconds spent in miss_torture using the sorted() line.
1018
# Even with out of order disk IO it seems faster not to sort it when
1019
# large queries are being made.
1020
needed_keys = sorted(needed_keys)
1022
nodes_and_keys = [(0, needed_keys)]
1024
for row_pos, next_row_start in enumerate(self._row_offsets[1:-1]):
1025
node_indexes = [idx for idx, s_keys in nodes_and_keys]
1026
nodes = self._get_internal_nodes(node_indexes)
1028
next_nodes_and_keys = []
1029
for node_index, sub_keys in nodes_and_keys:
1030
node = nodes[node_index]
1031
positions = self._multi_bisect_right(sub_keys, node.keys)
1032
node_offset = next_row_start + node.offset
1033
next_nodes_and_keys.extend([(node_offset + pos, s_keys)
1034
for pos, s_keys in positions])
1035
nodes_and_keys = next_nodes_and_keys
1036
# We should now be at the _LeafNodes
1037
node_indexes = [idx for idx, s_keys in nodes_and_keys]
1039
# TODO: We may *not* want to always read all the nodes in one
1040
# big go. Consider setting a max size on this.
1042
nodes = self._get_leaf_nodes(node_indexes)
1043
for node_index, sub_keys in nodes_and_keys:
1046
node = nodes[node_index]
1047
for next_sub_key in sub_keys:
1048
if next_sub_key in node.keys:
1049
value, refs = node.keys[next_sub_key]
1050
if self.node_ref_lists:
1051
yield (self, next_sub_key, value, refs)
1053
yield (self, next_sub_key, value)
1055
def iter_entries_prefix(self, keys):
1056
"""Iterate over keys within the index using prefix matching.
1058
Prefix matching is applied within the tuple of a key, not to within
1059
the bytestring of each key element. e.g. if you have the keys ('foo',
1060
'bar'), ('foobar', 'gam') and do a prefix search for ('foo', None) then
1061
only the former key is returned.
1063
WARNING: Note that this method currently causes a full index parse
1064
unconditionally (which is reasonably appropriate as it is a means for
1065
thunking many small indices into one larger one and still supplies
1066
iter_all_entries at the thunk layer).
1068
:param keys: An iterable providing the key prefixes to be retrieved.
1069
Each key prefix takes the form of a tuple the length of a key, but
1070
with the last N elements 'None' rather than a regular bytestring.
1071
The first element cannot be 'None'.
1072
:return: An iterable as per iter_all_entries, but restricted to the
1073
keys with a matching prefix to those supplied. No additional keys
1074
will be returned, and every match that is in the index will be
1077
keys = sorted(set(keys))
1080
# Load if needed to check key lengths
1081
if self._key_count is None:
1082
self._get_root_node()
1083
# TODO: only access nodes that can satisfy the prefixes we are looking
1084
# for. For now, to meet API usage (as this function is not used by
1085
# current bzrlib) just suck the entire index and iterate in memory.
1087
if self.node_ref_lists:
1088
if self._key_length == 1:
1089
for _1, key, value, refs in self.iter_all_entries():
1090
nodes[key] = value, refs
1093
for _1, key, value, refs in self.iter_all_entries():
1094
key_value = key, value, refs
1095
# For a key of (foo, bar, baz) create
1096
# _nodes_by_key[foo][bar][baz] = key_value
1097
key_dict = nodes_by_key
1098
for subkey in key[:-1]:
1099
key_dict = key_dict.setdefault(subkey, {})
1100
key_dict[key[-1]] = key_value
1102
if self._key_length == 1:
1103
for _1, key, value in self.iter_all_entries():
1107
for _1, key, value in self.iter_all_entries():
1108
key_value = key, value
1109
# For a key of (foo, bar, baz) create
1110
# _nodes_by_key[foo][bar][baz] = key_value
1111
key_dict = nodes_by_key
1112
for subkey in key[:-1]:
1113
key_dict = key_dict.setdefault(subkey, {})
1114
key_dict[key[-1]] = key_value
1115
if self._key_length == 1:
1119
raise errors.BadIndexKey(key)
1120
if len(key) != self._key_length:
1121
raise errors.BadIndexKey(key)
1123
if self.node_ref_lists:
1124
value, node_refs = nodes[key]
1125
yield self, key, value, node_refs
1127
yield self, key, nodes[key]
1134
raise errors.BadIndexKey(key)
1135
if len(key) != self._key_length:
1136
raise errors.BadIndexKey(key)
1137
# find what it refers to:
1138
key_dict = nodes_by_key
1139
elements = list(key)
1140
# find the subdict whose contents should be returned.
1142
while len(elements) and elements[0] is not None:
1143
key_dict = key_dict[elements[0]]
1146
# a non-existant lookup.
1151
key_dict = dicts.pop(-1)
1152
# can't be empty or would not exist
1153
item, value = key_dict.iteritems().next()
1154
if type(value) == dict:
1156
dicts.extend(key_dict.itervalues())
1159
for value in key_dict.itervalues():
1160
# each value is the key:value:node refs tuple
1162
yield (self, ) + value
1164
# the last thing looked up was a terminal element
1165
yield (self, ) + key_dict
1167
def key_count(self):
1168
"""Return an estimate of the number of keys in this index.
1170
For BTreeGraphIndex the estimate is exact as it is contained in the
1173
if self._key_count is None:
1174
self._get_root_node()
1175
return self._key_count
1177
def _compute_row_offsets(self):
1178
"""Fill out the _row_offsets attribute based on _row_lengths."""
1181
for row in self._row_lengths:
1182
offsets.append(row_offset)
1184
offsets.append(row_offset)
1185
self._row_offsets = offsets
1187
def _parse_header_from_bytes(self, bytes):
1188
"""Parse the header from a region of bytes.
1190
:param bytes: The data to parse.
1191
:return: An offset, data tuple such as readv yields, for the unparsed
1192
data. (which may be of length 0).
1194
signature = bytes[0:len(self._signature())]
1195
if not signature == self._signature():
1196
raise errors.BadIndexFormatSignature(self._name, BTreeGraphIndex)
1197
lines = bytes[len(self._signature()):].splitlines()
1198
options_line = lines[0]
1199
if not options_line.startswith(_OPTION_NODE_REFS):
1200
raise errors.BadIndexOptions(self)
1202
self.node_ref_lists = int(options_line[len(_OPTION_NODE_REFS):])
1204
raise errors.BadIndexOptions(self)
1205
options_line = lines[1]
1206
if not options_line.startswith(_OPTION_KEY_ELEMENTS):
1207
raise errors.BadIndexOptions(self)
1209
self._key_length = int(options_line[len(_OPTION_KEY_ELEMENTS):])
1211
raise errors.BadIndexOptions(self)
1212
options_line = lines[2]
1213
if not options_line.startswith(_OPTION_LEN):
1214
raise errors.BadIndexOptions(self)
1216
self._key_count = int(options_line[len(_OPTION_LEN):])
1218
raise errors.BadIndexOptions(self)
1219
options_line = lines[3]
1220
if not options_line.startswith(_OPTION_ROW_LENGTHS):
1221
raise errors.BadIndexOptions(self)
1223
self._row_lengths = map(int, [length for length in
1224
options_line[len(_OPTION_ROW_LENGTHS):].split(',')
1227
raise errors.BadIndexOptions(self)
1228
self._compute_row_offsets()
1230
# calculate the bytes we have processed
1231
header_end = (len(signature) + sum(map(len, lines[0:4])) + 4)
1232
return header_end, bytes[header_end:]
1234
def _read_nodes(self, nodes):
1235
"""Read some nodes from disk into the LRU cache.
1237
This performs a readv to get the node data into memory, and parses each
1238
node, the yields it to the caller. The nodes are requested in the
1239
supplied order. If possible doing sort() on the list before requesting
1240
a read may improve performance.
1242
:param nodes: The nodes to read. 0 - first node, 1 - second node etc.
1247
offset = index * _PAGE_SIZE
1250
# Root node - special case
1252
size = min(_PAGE_SIZE, self._size)
1254
stream = self._transport.get(self._name)
1255
start = stream.read(_PAGE_SIZE)
1256
# Avoid doing this again
1257
self._size = len(start)
1258
size = min(_PAGE_SIZE, self._size)
1260
if offset > self._size:
1261
raise AssertionError('tried to read past the end'
1262
' of the file %s > %s'
1263
% (offset, self._size))
1264
size = min(size, self._size - offset)
1265
ranges.append((offset, size))
1268
if self._file is None:
1269
data_ranges = self._transport.readv(self._name, ranges)
1272
for offset, size in ranges:
1273
self._file.seek(offset)
1274
data_ranges.append((offset, self._file.read(size)))
1275
for offset, data in data_ranges:
1277
# extract the header
1278
offset, data = self._parse_header_from_bytes(data)
1281
bytes = zlib.decompress(data)
1282
if bytes.startswith(_LEAF_FLAG):
1283
node = _LeafNode(bytes, self._key_length, self.node_ref_lists)
1284
elif bytes.startswith(_INTERNAL_FLAG):
1285
node = _InternalNode(bytes)
1287
raise AssertionError("Unknown node type for %r" % bytes)
1288
yield offset / _PAGE_SIZE, node
1290
def _signature(self):
1291
"""The file signature for this index type."""
1295
"""Validate that everything in the index can be accessed."""
1296
# just read and parse every node.
1297
self._get_root_node()
1298
if len(self._row_lengths) > 1:
1299
start_node = self._row_offsets[1]
1301
# We shouldn't be reading anything anyway
1303
node_end = self._row_offsets[-1]
1304
for node in self._read_nodes(range(start_node, node_end)):
1309
from bzrlib import _btree_serializer_c as _btree_serializer
1311
from bzrlib import _btree_serializer_py as _btree_serializer
added
removed
bzrlib/btree_index.py
