-
Notifications
You must be signed in to change notification settings - Fork 6.3k
/
merging_iterator.cc
1755 lines (1653 loc) · 73.3 KB
/
merging_iterator.cc
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
// Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
// This source code is licensed under both the GPLv2 (found in the
// COPYING file in the root directory) and Apache 2.0 License
// (found in the LICENSE.Apache file in the root directory).
//
// Copyright (c) 2011 The LevelDB Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file. See the AUTHORS file for names of contributors.
#include "table/merging_iterator.h"
#include "db/arena_wrapped_db_iter.h"
namespace ROCKSDB_NAMESPACE {
// MergingIterator uses a min/max heap to combine data from point iterators.
// Range tombstones can be added and keys covered by range tombstones will be
// skipped.
//
// The following are implementation details and can be ignored by user.
// For merging iterator to process range tombstones, it treats the start and end
// keys of a range tombstone as two keys and put them into minHeap_ or maxHeap_
// together with regular point keys. Each range tombstone is active only within
// its internal key range [start_key, end_key). An `active_` set is used to
// track levels that have an active range tombstone. Take forward scanning
// for example. Level j is in active_ if its current range tombstone has its
// start_key popped from minHeap_ and its end_key in minHeap_. If the top of
// minHeap_ is a point key from level L, we can determine if the point key is
// covered by any range tombstone by checking if there is an l <= L in active_.
// The case of l == L also involves checking range tombstone's sequence number.
//
// The following (non-exhaustive) list of invariants are maintained by
// MergingIterator during forward scanning. After each InternalIterator API,
// i.e., Seek*() and Next(), and FindNextVisibleKey(), if minHeap_ is not empty:
// (1) minHeap_.top().type == ITERATOR
// (2) minHeap_.top()->key() is not covered by any range tombstone.
//
// After each call to SeekImpl() in addition to the functions mentioned above:
// (3) For all level i and j <= i, range_tombstone_iters_[j].prev.end_key() <
// children_[i].iter.key(). That is, range_tombstone_iters_[j] is at or before
// the first range tombstone from level j with end_key() >
// children_[i].iter.key().
// (4) For all level i and j <= i, if j in active_, then
// range_tombstone_iters_[j]->start_key() < children_[i].iter.key().
// - When range_tombstone_iters_[j] is !Valid(), we consider its `prev` to be
// the last range tombstone from that range tombstone iterator.
// - When referring to range tombstone start/end keys, assume it is the value of
// HeapItem::tombstone_pik. This value has op_type = kMaxValid, which makes
// range tombstone keys have distinct values from point keys.
//
// Applicable class variables have their own (forward scanning) invariants
// listed in the comments above their definition.
class MergingIterator : public InternalIterator {
public:
MergingIterator(const InternalKeyComparator* comparator,
InternalIterator** children, int n, bool is_arena_mode,
bool prefix_seek_mode,
const Slice* iterate_upper_bound = nullptr)
: is_arena_mode_(is_arena_mode),
prefix_seek_mode_(prefix_seek_mode),
direction_(kForward),
comparator_(comparator),
current_(nullptr),
minHeap_(MinHeapItemComparator(comparator_)),
pinned_iters_mgr_(nullptr),
iterate_upper_bound_(iterate_upper_bound) {
children_.resize(n);
for (int i = 0; i < n; i++) {
children_[i].level = i;
children_[i].iter.Set(children[i]);
}
}
void considerStatus(Status s) {
if (!s.ok() && status_.ok()) {
status_ = s;
}
}
virtual void AddIterator(InternalIterator* iter) {
children_.emplace_back(children_.size(), iter);
if (pinned_iters_mgr_) {
iter->SetPinnedItersMgr(pinned_iters_mgr_);
}
// Invalidate to ensure `Seek*()` is called to construct the heaps before
// use.
current_ = nullptr;
}
// There must be either no range tombstone iterator or the same number of
// range tombstone iterators as point iterators after all iters are added.
// The i-th added range tombstone iterator and the i-th point iterator
// must point to the same LSM level.
// Merging iterator takes ownership of `iter` and is responsible for freeing
// it. One exception to this is when a LevelIterator moves to a different SST
// file or when Iterator::Refresh() is called, the range tombstone iterator
// could be updated. In that case, this merging iterator is only responsible
// for freeing the new range tombstone iterator that it has pointers to in
// range_tombstone_iters_.
void AddRangeTombstoneIterator(
std::unique_ptr<TruncatedRangeDelIterator>&& iter) {
range_tombstone_iters_.emplace_back(std::move(iter));
}
// Called by MergingIteratorBuilder when all point iterators and range
// tombstone iterators are added. Initializes HeapItems for range tombstone
// iterators.
void Finish() {
if (!range_tombstone_iters_.empty()) {
assert(range_tombstone_iters_.size() == children_.size());
pinned_heap_item_.resize(range_tombstone_iters_.size());
for (size_t i = 0; i < range_tombstone_iters_.size(); ++i) {
pinned_heap_item_[i].level = i;
// Range tombstone end key is exclusive. If a point internal key has the
// same user key and sequence number as the start or end key of a range
// tombstone, the order will be start < end key < internal key with the
// following op_type change. This is helpful to ensure keys popped from
// heap are in expected order since range tombstone start/end keys will
// be distinct from point internal keys. Strictly speaking, this is only
// needed for tombstone end points that are truncated in
// TruncatedRangeDelIterator since untruncated tombstone end points
// always have kMaxSequenceNumber and kTypeRangeDeletion (see
// TruncatedRangeDelIterator::start_key()/end_key()).
pinned_heap_item_[i].tombstone_pik.type = kTypeMaxValid;
}
}
}
~MergingIterator() override {
range_tombstone_iters_.clear();
for (auto& child : children_) {
child.iter.DeleteIter(is_arena_mode_);
}
status_.PermitUncheckedError();
}
void SetRangeDelReadSeqno(SequenceNumber read_seqno) override {
for (auto& child : children_) {
// This should only be needed for LevelIterator (iterators from L1+).
child.iter.SetRangeDelReadSeqno(read_seqno);
}
for (auto& child : range_tombstone_iters_) {
if (child) {
child->SetRangeDelReadSeqno(read_seqno);
}
}
}
bool Valid() const override { return current_ != nullptr && status_.ok(); }
Status status() const override { return status_; }
// Add range_tombstone_iters_[level] into min heap.
// Updates active_ if the end key of a range tombstone is inserted.
// pinned_heap_items_[level].type is updated based on `start_key`.
//
// If range_tombstone_iters_[level] is after iterate_upper_bound_,
// it is removed from the heap.
// @param start_key specifies which end point of the range tombstone to add.
void InsertRangeTombstoneToMinHeap(size_t level, bool start_key = true,
bool replace_top = false) {
assert(!range_tombstone_iters_.empty() &&
range_tombstone_iters_[level]->Valid());
// Maintains Invariant(phi)
if (start_key) {
pinned_heap_item_[level].type = HeapItem::Type::DELETE_RANGE_START;
ParsedInternalKey pik = range_tombstone_iters_[level]->start_key();
// iterate_upper_bound does not have timestamp
if (iterate_upper_bound_ &&
comparator_->user_comparator()->CompareWithoutTimestamp(
pik.user_key, true /* a_has_ts */, *iterate_upper_bound_,
false /* b_has_ts */) >= 0) {
if (replace_top) {
// replace_top implies this range tombstone iterator is still in
// minHeap_ and at the top.
minHeap_.pop();
}
return;
}
pinned_heap_item_[level].SetTombstoneKey(std::move(pik));
// Checks Invariant(active_)
assert(active_.count(level) == 0);
} else {
// allow end key to go over upper bound (if present) since start key is
// before upper bound and the range tombstone could still cover a
// range before upper bound.
// Maintains Invariant(active_)
pinned_heap_item_[level].SetTombstoneKey(
range_tombstone_iters_[level]->end_key());
pinned_heap_item_[level].type = HeapItem::Type::DELETE_RANGE_END;
active_.insert(level);
}
if (replace_top) {
minHeap_.replace_top(&pinned_heap_item_[level]);
} else {
minHeap_.push(&pinned_heap_item_[level]);
}
}
// Add range_tombstone_iters_[level] into max heap.
// Updates active_ if the start key of a range tombstone is inserted.
// @param end_key specifies which end point of the range tombstone to add.
void InsertRangeTombstoneToMaxHeap(size_t level, bool end_key = true,
bool replace_top = false) {
assert(!range_tombstone_iters_.empty() &&
range_tombstone_iters_[level]->Valid());
if (end_key) {
pinned_heap_item_[level].SetTombstoneKey(
range_tombstone_iters_[level]->end_key());
pinned_heap_item_[level].type = HeapItem::Type::DELETE_RANGE_END;
assert(active_.count(level) == 0);
} else {
pinned_heap_item_[level].SetTombstoneKey(
range_tombstone_iters_[level]->start_key());
pinned_heap_item_[level].type = HeapItem::Type::DELETE_RANGE_START;
active_.insert(level);
}
if (replace_top) {
maxHeap_->replace_top(&pinned_heap_item_[level]);
} else {
maxHeap_->push(&pinned_heap_item_[level]);
}
}
// Remove HeapItems from top of minHeap_ that are of type DELETE_RANGE_START
// until minHeap_ is empty or the top of the minHeap_ is not of type
// DELETE_RANGE_START. Each such item means a range tombstone becomes active,
// so `active_` is updated accordingly.
void PopDeleteRangeStart() {
while (!minHeap_.empty() &&
minHeap_.top()->type == HeapItem::Type::DELETE_RANGE_START) {
TEST_SYNC_POINT_CALLBACK("MergeIterator::PopDeleteRangeStart", nullptr);
// Invariant(rti) holds since
// range_tombstone_iters_[minHeap_.top()->level] is still valid, and
// parameter `replace_top` is set to true here to ensure only one such
// HeapItem is in minHeap_.
InsertRangeTombstoneToMinHeap(
minHeap_.top()->level, false /* start_key */, true /* replace_top */);
}
}
// Remove HeapItems from top of maxHeap_ that are of type DELETE_RANGE_END
// until maxHeap_ is empty or the top of the maxHeap_ is not of type
// DELETE_RANGE_END. Each such item means a range tombstone becomes active,
// so `active_` is updated accordingly.
void PopDeleteRangeEnd() {
while (!maxHeap_->empty() &&
maxHeap_->top()->type == HeapItem::Type::DELETE_RANGE_END) {
// insert start key of this range tombstone and updates active_
InsertRangeTombstoneToMaxHeap(maxHeap_->top()->level, false /* end_key */,
true /* replace_top */);
}
}
void SeekToFirst() override {
ClearHeaps();
status_ = Status::OK();
for (auto& child : children_) {
child.iter.SeekToFirst();
AddToMinHeapOrCheckStatus(&child);
}
for (size_t i = 0; i < range_tombstone_iters_.size(); ++i) {
if (range_tombstone_iters_[i]) {
range_tombstone_iters_[i]->SeekToFirst();
if (range_tombstone_iters_[i]->Valid()) {
// It is possible to be invalid due to snapshots.
InsertRangeTombstoneToMinHeap(i);
}
}
}
FindNextVisibleKey();
direction_ = kForward;
current_ = CurrentForward();
}
void SeekToLast() override {
ClearHeaps();
InitMaxHeap();
status_ = Status::OK();
for (auto& child : children_) {
child.iter.SeekToLast();
AddToMaxHeapOrCheckStatus(&child);
}
for (size_t i = 0; i < range_tombstone_iters_.size(); ++i) {
if (range_tombstone_iters_[i]) {
range_tombstone_iters_[i]->SeekToLast();
if (range_tombstone_iters_[i]->Valid()) {
// It is possible to be invalid due to snapshots.
InsertRangeTombstoneToMaxHeap(i);
}
}
}
FindPrevVisibleKey();
direction_ = kReverse;
current_ = CurrentReverse();
}
// Position this merging iterator at the first key >= target (internal key).
// If range tombstones are present, keys covered by range tombstones are
// skipped, and this merging iter points to the first non-range-deleted key >=
// target after Seek(). If !Valid() and status().ok() then this iterator
// reaches the end.
//
// If range tombstones are present, cascading seeks may be called (an
// optimization adapted from Pebble https://github.com/cockroachdb/pebble).
// Roughly, if there is a range tombstone [start, end) that covers the
// target user key at level L, then this range tombstone must cover the range
// [target key, end) in all levels > L. So for all levels > L, we can pretend
// the target key is `end`. This optimization is applied at each level and
// hence the name "cascading seek".
void Seek(const Slice& target) override {
// Define LevelNextVisible(i, k) to be the first key >= k in level i that is
// not covered by any range tombstone.
// After SeekImpl(target, 0), invariants (3) and (4) hold.
// For all level i, target <= children_[i].iter.key() <= LevelNextVisible(i,
// target). By the contract of FindNextVisibleKey(), Invariants (1)-(4)
// holds after this call, and minHeap_.top().iter points to the
// first key >= target among children_ that is not covered by any range
// tombstone.
status_ = Status::OK();
SeekImpl(target);
FindNextVisibleKey();
direction_ = kForward;
{
PERF_TIMER_GUARD(seek_min_heap_time);
current_ = CurrentForward();
}
}
void SeekForPrev(const Slice& target) override {
assert(range_tombstone_iters_.empty() ||
range_tombstone_iters_.size() == children_.size());
status_ = Status::OK();
SeekForPrevImpl(target);
FindPrevVisibleKey();
direction_ = kReverse;
{
PERF_TIMER_GUARD(seek_max_heap_time);
current_ = CurrentReverse();
}
}
void Next() override {
assert(Valid());
// Ensure that all children are positioned after key().
// If we are moving in the forward direction, it is already
// true for all the non-current children since current_ is
// the smallest child and key() == current_->key().
if (direction_ != kForward) {
// The loop advanced all non-current children to be > key() so current_
// should still be strictly the smallest key.
SwitchToForward();
}
// For the heap modifications below to be correct, current_ must be the
// current top of the heap.
assert(current_ == CurrentForward());
// as the current points to the current record. move the iterator forward.
current_->Next();
if (current_->Valid()) {
// current is still valid after the Next() call above. Call
// replace_top() to restore the heap property. When the same child
// iterator yields a sequence of keys, this is cheap.
assert(current_->status().ok());
minHeap_.replace_top(minHeap_.top());
} else {
// current stopped being valid, remove it from the heap.
considerStatus(current_->status());
minHeap_.pop();
}
// Invariants (3) and (4) hold when after advancing current_.
// Let k be the smallest key among children_[i].iter.key().
// k <= children_[i].iter.key() <= LevelNextVisible(i, k) holds for all
// level i. After FindNextVisible(), Invariants (1)-(4) hold and
// minHeap_.top()->key() is the first key >= k from any children_ that is
// not covered by any range tombstone.
FindNextVisibleKey();
current_ = CurrentForward();
}
bool NextAndGetResult(IterateResult* result) override {
Next();
bool is_valid = Valid();
if (is_valid) {
result->key = key();
result->bound_check_result = UpperBoundCheckResult();
result->value_prepared = current_->IsValuePrepared();
}
return is_valid;
}
void Prev() override {
assert(Valid());
// Ensure that all children are positioned before key().
// If we are moving in the reverse direction, it is already
// true for all the non-current children since current_ is
// the largest child and key() == current_->key().
if (direction_ != kReverse) {
// Otherwise, retreat the non-current children. We retreat current_
// just after the if-block.
SwitchToBackward();
}
// For the heap modifications below to be correct, current_ must be the
// current top of the heap.
assert(current_ == CurrentReverse());
current_->Prev();
if (current_->Valid()) {
// current is still valid after the Prev() call above. Call
// replace_top() to restore the heap property. When the same child
// iterator yields a sequence of keys, this is cheap.
assert(current_->status().ok());
maxHeap_->replace_top(maxHeap_->top());
} else {
// current stopped being valid, remove it from the heap.
considerStatus(current_->status());
maxHeap_->pop();
}
FindPrevVisibleKey();
current_ = CurrentReverse();
}
Slice key() const override {
assert(Valid());
return current_->key();
}
uint64_t write_unix_time() const override {
assert(Valid());
return current_->write_unix_time();
}
Slice value() const override {
assert(Valid());
return current_->value();
}
bool PrepareValue() override {
assert(Valid());
if (current_->PrepareValue()) {
return true;
}
considerStatus(current_->status());
assert(!status_.ok());
return false;
}
// Here we simply relay MayBeOutOfLowerBound/MayBeOutOfUpperBound result
// from current child iterator. Potentially as long as one of child iterator
// report out of bound is not possible, we know current key is within bound.
bool MayBeOutOfLowerBound() override {
assert(Valid());
return current_->MayBeOutOfLowerBound();
}
IterBoundCheck UpperBoundCheckResult() override {
assert(Valid());
return current_->UpperBoundCheckResult();
}
void SetPinnedItersMgr(PinnedIteratorsManager* pinned_iters_mgr) override {
pinned_iters_mgr_ = pinned_iters_mgr;
for (auto& child : children_) {
child.iter.SetPinnedItersMgr(pinned_iters_mgr);
}
}
bool IsKeyPinned() const override {
assert(Valid());
return pinned_iters_mgr_ && pinned_iters_mgr_->PinningEnabled() &&
current_->IsKeyPinned();
}
bool IsValuePinned() const override {
assert(Valid());
return pinned_iters_mgr_ && pinned_iters_mgr_->PinningEnabled() &&
current_->IsValuePinned();
}
private:
// Represents an element in the min/max heap. Each HeapItem corresponds to a
// point iterator or a range tombstone iterator, differentiated by
// HeapItem::type.
struct HeapItem {
HeapItem() = default;
// corresponding point iterator
IteratorWrapper iter;
size_t level = 0;
// corresponding range tombstone iterator's start or end key value
// depending on value of `type`.
ParsedInternalKey tombstone_pik;
// Will be overwritten before use, initialize here so compiler does not
// complain.
enum class Type { ITERATOR, DELETE_RANGE_START, DELETE_RANGE_END };
Type type = Type::ITERATOR;
explicit HeapItem(size_t _level, InternalIteratorBase<Slice>* _iter)
: level(_level), type(Type::ITERATOR) {
iter.Set(_iter);
}
void SetTombstoneKey(ParsedInternalKey&& pik) {
// op_type is already initialized in MergingIterator::Finish().
tombstone_pik.user_key = pik.user_key;
tombstone_pik.sequence = pik.sequence;
}
};
class MinHeapItemComparator {
public:
explicit MinHeapItemComparator(const InternalKeyComparator* comparator)
: comparator_(comparator) {}
bool operator()(HeapItem* a, HeapItem* b) const {
if (LIKELY(a->type == HeapItem::Type::ITERATOR)) {
if (LIKELY(b->type == HeapItem::Type::ITERATOR)) {
return comparator_->Compare(a->iter.key(), b->iter.key()) > 0;
} else {
return comparator_->Compare(a->iter.key(), b->tombstone_pik) > 0;
}
} else {
if (LIKELY(b->type == HeapItem::Type::ITERATOR)) {
return comparator_->Compare(a->tombstone_pik, b->iter.key()) > 0;
} else {
return comparator_->Compare(a->tombstone_pik, b->tombstone_pik) > 0;
}
}
}
private:
const InternalKeyComparator* comparator_;
};
class MaxHeapItemComparator {
public:
explicit MaxHeapItemComparator(const InternalKeyComparator* comparator)
: comparator_(comparator) {}
bool operator()(HeapItem* a, HeapItem* b) const {
if (LIKELY(a->type == HeapItem::Type::ITERATOR)) {
if (LIKELY(b->type == HeapItem::Type::ITERATOR)) {
return comparator_->Compare(a->iter.key(), b->iter.key()) < 0;
} else {
return comparator_->Compare(a->iter.key(), b->tombstone_pik) < 0;
}
} else {
if (LIKELY(b->type == HeapItem::Type::ITERATOR)) {
return comparator_->Compare(a->tombstone_pik, b->iter.key()) < 0;
} else {
return comparator_->Compare(a->tombstone_pik, b->tombstone_pik) < 0;
}
}
}
private:
const InternalKeyComparator* comparator_;
};
using MergerMinIterHeap = BinaryHeap<HeapItem*, MinHeapItemComparator>;
using MergerMaxIterHeap = BinaryHeap<HeapItem*, MaxHeapItemComparator>;
friend class MergeIteratorBuilder;
// Clears heaps for both directions, used when changing direction or seeking
void ClearHeaps(bool clear_active = true);
// Ensures that maxHeap_ is initialized when starting to go in the reverse
// direction
void InitMaxHeap();
// Advance this merging iterator until the current key (minHeap_.top()) is
// from a point iterator and is not covered by any range tombstone,
// or that there is no more keys (heap is empty). SeekImpl() may be called
// to seek to the end of a range tombstone as an optimization.
void FindNextVisibleKey();
void FindPrevVisibleKey();
// Advance this merging iterators to the first key >= `target` for all
// components from levels >= starting_level. All iterators before
// starting_level are untouched.
//
// @param range_tombstone_reseek Whether target is some range tombstone
// end, i.e., whether this SeekImpl() call is a part of a "cascading seek".
// This is used only for recoding relevant perf_context.
void SeekImpl(const Slice& target, size_t starting_level = 0,
bool range_tombstone_reseek = false);
// Seek to fist key <= target key (internal key) for
// children_[starting_level:].
void SeekForPrevImpl(const Slice& target, size_t starting_level = 0,
bool range_tombstone_reseek = false);
bool is_arena_mode_;
bool prefix_seek_mode_;
// Which direction is the iterator moving?
enum Direction : uint8_t { kForward, kReverse };
Direction direction_;
const InternalKeyComparator* comparator_;
// HeapItem for all child point iterators.
// Invariant(children_): children_[i] is in minHeap_ iff
// children_[i].iter.Valid(), and at most one children_[i] is in minHeap_.
// TODO: We could use an autovector with a larger reserved size.
std::vector<HeapItem> children_;
// HeapItem for range tombstone start and end keys.
// pinned_heap_item_[i] corresponds to range_tombstone_iters_[i].
// Invariant(phi): If range_tombstone_iters_[i]->Valid(),
// pinned_heap_item_[i].tombstone_pik is equal to
// range_tombstone_iters_[i]->start_key() when
// pinned_heap_item_[i].type is DELETE_RANGE_START and
// range_tombstone_iters_[i]->end_key() when
// pinned_heap_item_[i].type is DELETE_RANGE_END (ignoring op_type which is
// kMaxValid for all pinned_heap_item_.tombstone_pik).
// pinned_heap_item_[i].type is either DELETE_RANGE_START or DELETE_RANGE_END.
std::vector<HeapItem> pinned_heap_item_;
// range_tombstone_iters_[i] contains range tombstones in the sorted run that
// corresponds to children_[i]. range_tombstone_iters_.empty() means not
// handling range tombstones in merging iterator. range_tombstone_iters_[i] ==
// nullptr means the sorted run of children_[i] does not have range
// tombstones.
// Invariant(rti): pinned_heap_item_[i] is in minHeap_ iff
// range_tombstone_iters_[i]->Valid() and at most one pinned_heap_item_[i] is
// in minHeap_.
std::vector<std::unique_ptr<TruncatedRangeDelIterator>>
range_tombstone_iters_;
// Levels (indices into range_tombstone_iters_/children_ ) that currently have
// "active" range tombstones. See comments above MergingIterator for meaning
// of "active".
// Invariant(active_): i is in active_ iff range_tombstone_iters_[i]->Valid()
// and pinned_heap_item_[i].type == DELETE_RANGE_END.
std::set<size_t> active_;
bool SkipNextDeleted();
bool SkipPrevDeleted();
// Invariant: at the end of each InternalIterator API,
// current_ points to minHeap_.top().iter (maxHeap_ if backward scanning)
// or nullptr if no child iterator is valid.
// This follows from that current_ = CurrentForward()/CurrentReverse() is
// called at the end of each InternalIterator API.
IteratorWrapper* current_;
// If any of the children have non-ok status, this is one of them.
Status status_;
// Invariant: min heap property is maintained (parent is always <= child).
// This holds by using only BinaryHeap APIs to modify heap. One
// exception is to modify heap top item directly (by caller iter->Next()), and
// it should be followed by a call to replace_top() or pop().
MergerMinIterHeap minHeap_;
// Max heap is used for reverse iteration, which is way less common than
// forward. Lazily initialize it to save memory.
std::unique_ptr<MergerMaxIterHeap> maxHeap_;
PinnedIteratorsManager* pinned_iters_mgr_;
// Used to bound range tombstones. For point keys, DBIter and SSTable iterator
// take care of boundary checking.
const Slice* iterate_upper_bound_;
// In forward direction, process a child that is not in the min heap.
// If valid, add to the min heap. Otherwise, check status.
void AddToMinHeapOrCheckStatus(HeapItem*);
// In backward direction, process a child that is not in the max heap.
// If valid, add to the min heap. Otherwise, check status.
void AddToMaxHeapOrCheckStatus(HeapItem*);
void SwitchToForward();
// Switch the direction from forward to backward without changing the
// position. Iterator should still be valid.
void SwitchToBackward();
IteratorWrapper* CurrentForward() const {
assert(direction_ == kForward);
assert(minHeap_.empty() ||
minHeap_.top()->type == HeapItem::Type::ITERATOR);
return !minHeap_.empty() ? &minHeap_.top()->iter : nullptr;
}
IteratorWrapper* CurrentReverse() const {
assert(direction_ == kReverse);
assert(maxHeap_);
assert(maxHeap_->empty() ||
maxHeap_->top()->type == HeapItem::Type::ITERATOR);
return !maxHeap_->empty() ? &maxHeap_->top()->iter : nullptr;
}
};
// Pre-condition:
// - Invariants (3) and (4) hold for i < starting_level
// - For i < starting_level, range_tombstone_iters_[i].prev.end_key() <
// `target`.
// - For i < starting_level, if i in active_, then
// range_tombstone_iters_[i]->start_key() < `target`.
//
// Post-condition:
// - Invariants (3) and (4) hold for all level i.
// - (*) target <= children_[i].iter.key() <= LevelNextVisible(i, target)
// for i >= starting_level
// - (**) target < pinned_heap_item_[i].tombstone_pik if
// range_tombstone_iters_[i].Valid() for i >= starting_level
//
// Proof sketch:
// Invariant (3) holds for all level i.
// For j <= i < starting_level, it follows from Pre-condition that (3) holds
// and that SeekImpl(-, starting_level) does not update children_[i] or
// range_tombstone_iters_[j].
// For j < starting_level and i >= starting_level, it follows from
// - Pre-condition that range_tombstone_iters_[j].prev.end_key() < `target`
// - range_tombstone_iters_[j] is not updated in SeekImpl(), and
// - children_[i].iter.Seek(current_search_key) is called with
// current_search_key >= target (shown below).
// When current_search_key is updated, it is updated to some
// range_tombstone_iter->end_key() after
// range_tombstone_iter->SeekInternalKey(current_search_key) was called. So
// current_search_key increases if updated and >= target.
// For starting_level <= j <= i:
// children_[i].iter.Seek(k1) and range_tombstone_iters_[j]->SeekInternalKey(k2)
// are called in SeekImpl(). Seek(k1) positions children_[i] at the first key >=
// k1 from level i. SeekInternalKey(k2) positions range_tombstone_iters_[j] at
// the first range tombstone from level j with end_key() > k2. It suffices to
// show that k1 >= k2. Since k1 and k2 are values of current_search_key where
// k1 = k2 or k1 is value of a later current_search_key than k2, so k1 >= k2.
//
// Invariant (4) holds for all level >= 0.
// By Pre-condition Invariant (4) holds for i < starting_level.
// Since children_[i], range_tombstone_iters_[i] and contents of active_ for
// i < starting_level do not change (4) holds for j <= i < starting_level.
// By Pre-condition: for all j < starting_level, if j in active_, then
// range_tombstone_iters_[j]->start_key() < target. For i >= starting_level,
// children_[i].iter.Seek(k) is called for k >= target. So
// children_[i].iter.key() >= target > range_tombstone_iters_[j]->start_key()
// for j < starting_level and i >= starting_level. So invariant (4) holds for
// j < starting_level and i >= starting_level.
// For starting_level <= j <= i, j is added to active_ only if
// - range_tombstone_iters_[j]->SeekInternalKey(k1) was called
// - range_tombstone_iters_[j]->start_key() <= k1
// Since children_[i].iter.Seek(k2) is called for some k2 >= k1 and for all
// starting_level <= j <= i, (4) also holds for all starting_level <= j <= i.
//
// Post-condition (*): target <= children_[i].iter.key() <= LevelNextVisible(i,
// target) for i >= starting_level.
// target <= children_[i].iter.key() follows from that Seek() is called on some
// current_search_key >= target for children_[i].iter. If current_search_key
// is updated from k1 to k2 when level = i, we show that the range [k1, k2) is
// not visible for children_[j] for any j > i. When current_search_key is
// updated from k1 to k2,
// - range_tombstone_iters_[i]->SeekInternalKey(k1) was called
// - range_tombstone_iters_[i]->Valid()
// - range_tombstone_iters_[i]->start_key().user_key <= k1.user_key
// - k2 = range_tombstone_iters_[i]->end_key()
// We assume that range_tombstone_iters_[i]->start_key() has a higher sequence
// number compared to any key from levels > i that has the same user key. So no
// point key from levels > i in range [k1, k2) is visible. So
// children_[i].iter.key() <= LevelNextVisible(i, target).
//
// Post-condition (**) target < pinned_heap_item_[i].tombstone_pik for i >=
// starting_level if range_tombstone_iters_[i].Valid(). This follows from that
// SeekInternalKey() being called for each range_tombstone_iters_ with some key
// >= `target` and that we pick start/end key that is > `target` to insert to
// minHeap_.
void MergingIterator::SeekImpl(const Slice& target, size_t starting_level,
bool range_tombstone_reseek) {
// active range tombstones before `starting_level` remain active
ClearHeaps(false /* clear_active */);
ParsedInternalKey pik;
if (!range_tombstone_iters_.empty()) {
// pik is only used in InsertRangeTombstoneToMinHeap().
ParseInternalKey(target, &pik, false).PermitUncheckedError();
}
// TODO: perhaps we could save some upheap cost by add all child iters first
// and then do a single heapify.
// Invariant(children_) for level < starting_level
for (size_t level = 0; level < starting_level; ++level) {
PERF_TIMER_GUARD(seek_min_heap_time);
AddToMinHeapOrCheckStatus(&children_[level]);
}
if (!range_tombstone_iters_.empty()) {
// Add range tombstones from levels < starting_level. We can insert from
// pinned_heap_item_ for the following reasons:
// - pinned_heap_item_[level] is in minHeap_ iff
// range_tombstone_iters[level]->Valid().
// - If `level` is in active_, then range_tombstone_iters_[level]->Valid()
// and pinned_heap_item_[level] is of type RANGE_DELETION_END.
for (size_t level = 0; level < starting_level; ++level) {
// Restores Invariants(rti), (phi) and (active_) for level <
// starting_level
if (range_tombstone_iters_[level] &&
range_tombstone_iters_[level]->Valid()) {
// use an iterator on active_ if performance becomes an issue here
if (active_.count(level) > 0) {
assert(pinned_heap_item_[level].type ==
HeapItem::Type::DELETE_RANGE_END);
// if it was active, then start key must be within upper_bound,
// so we can add to minHeap_ directly.
minHeap_.push(&pinned_heap_item_[level]);
} else {
assert(pinned_heap_item_[level].type ==
HeapItem::Type::DELETE_RANGE_START);
// this takes care of checking iterate_upper_bound, but with an extra
// key comparison if range_tombstone_iters_[level] was already out of
// bound. Consider using a new HeapItem type or some flag to remember
// boundary checking result.
InsertRangeTombstoneToMinHeap(level);
}
} else {
assert(!active_.count(level));
}
}
// levels >= starting_level will be reseeked below, so clearing their active
// state here.
active_.erase(active_.lower_bound(starting_level), active_.end());
}
IterKey current_search_key;
current_search_key.SetInternalKey(target, false /* copy */);
// Seek target might change to some range tombstone end key, so
// we need to remember them for async requests.
// (level, target) pairs
autovector<std::pair<size_t, std::string>> prefetched_target;
for (auto level = starting_level; level < children_.size(); ++level) {
{
PERF_TIMER_GUARD(seek_child_seek_time);
children_[level].iter.Seek(current_search_key.GetInternalKey());
}
PERF_COUNTER_ADD(seek_child_seek_count, 1);
if (!range_tombstone_iters_.empty()) {
if (range_tombstone_reseek) {
// This seek is to some range tombstone end key.
// Should only happen when there are range tombstones.
PERF_COUNTER_ADD(internal_range_del_reseek_count, 1);
}
if (children_[level].iter.status().IsTryAgain()) {
prefetched_target.emplace_back(
level, current_search_key.GetInternalKey().ToString());
}
UnownedPtr<TruncatedRangeDelIterator> range_tombstone_iter =
range_tombstone_iters_[level].get();
if (range_tombstone_iter) {
range_tombstone_iter->SeekInternalKey(
current_search_key.GetInternalKey());
// Invariants (rti) and (phi)
if (range_tombstone_iter->Valid()) {
// If range tombstone starts after `current_search_key`,
// we should insert start key to heap as the range tombstone is not
// active yet.
InsertRangeTombstoneToMinHeap(
level, comparator_->Compare(range_tombstone_iter->start_key(),
pik) > 0 /* start_key */);
// current_search_key < end_key guaranteed by the SeekInternalKey()
// and Valid() calls above. Here we only need to compare user_key
// since if target.user_key ==
// range_tombstone_iter->start_key().user_key and target <
// range_tombstone_iter->start_key(), no older level would have any
// key in range [target, range_tombstone_iter->start_key()], so no
// keys in range [target, range_tombstone_iter->end_key()) from older
// level would be visible. So it is safe to seek to
// range_tombstone_iter->end_key().
//
// TODO: range_tombstone_iter->Seek() finds the max covering
// sequence number, can make it cheaper by not looking for max.
if (comparator_->user_comparator()->Compare(
range_tombstone_iter->start_key().user_key,
current_search_key.GetUserKey()) <= 0) {
range_tombstone_reseek = true;
// Note that for prefix seek case, it is possible that the prefix
// is not the same as the original target, it should not affect
// correctness. Besides, in most cases, range tombstone start and
// end key should have the same prefix?
current_search_key.SetInternalKey(range_tombstone_iter->end_key());
}
}
}
}
// child.iter.status() is set to Status::TryAgain indicating asynchronous
// request for retrieval of data blocks has been submitted. So it should
// return at this point and Seek should be called again to retrieve the
// requested block and add the child to min heap.
if (children_[level].iter.status().IsTryAgain()) {
continue;
}
{
// Strictly, we timed slightly more than min heap operation,
// but these operations are very cheap.
PERF_TIMER_GUARD(seek_min_heap_time);
AddToMinHeapOrCheckStatus(&children_[level]);
}
}
if (range_tombstone_iters_.empty()) {
for (auto& child : children_) {
if (child.iter.status().IsTryAgain()) {
child.iter.Seek(target);
{
PERF_TIMER_GUARD(seek_min_heap_time);
AddToMinHeapOrCheckStatus(&child);
}
PERF_COUNTER_ADD(number_async_seek, 1);
}
}
} else {
for (auto& prefetch : prefetched_target) {
// (level, target) pairs
children_[prefetch.first].iter.Seek(prefetch.second);
{
PERF_TIMER_GUARD(seek_min_heap_time);
AddToMinHeapOrCheckStatus(&children_[prefetch.first]);
}
PERF_COUNTER_ADD(number_async_seek, 1);
}
}
}
// Returns true iff the current key (min heap top) should not be returned
// to user (of the merging iterator). This can be because the current key
// is deleted by some range tombstone, the current key is some fake file
// boundary sentinel key, or the current key is an end point of a range
// tombstone. Advance the iterator at heap top if needed. Heap order is restored
// and `active_` is updated accordingly.
// See FindNextVisibleKey() for more detail on internal implementation
// of advancing child iters.
// When false is returned, if minHeap is not empty, then minHeap_.top().type
// == ITERATOR
//
// REQUIRES:
// - min heap is currently not empty, and iter is in kForward direction.
// - minHeap_ top is not DELETE_RANGE_START (so that `active_` is current).
bool MergingIterator::SkipNextDeleted() {
// 3 types of keys:
// - point key
// - file boundary sentinel keys
// - range deletion end key
auto current = minHeap_.top();
if (current->type == HeapItem::Type::DELETE_RANGE_END) {
// Invariant(active_): range_tombstone_iters_[current->level] is about to
// become !Valid() or that its start key is going to be added to minHeap_.
active_.erase(current->level);
assert(range_tombstone_iters_[current->level] &&
range_tombstone_iters_[current->level]->Valid());
range_tombstone_iters_[current->level]->Next();
// Maintain Invariants (rti) and (phi)
if (range_tombstone_iters_[current->level]->Valid()) {
InsertRangeTombstoneToMinHeap(current->level, true /* start_key */,
true /* replace_top */);
} else {
// TruncatedRangeDelIterator does not have status
minHeap_.pop();
}
return true /* current key deleted */;
}
if (current->iter.IsDeleteRangeSentinelKey()) {
// If the file boundary is defined by a range deletion, the range
// tombstone's end key must come before this sentinel key (see op_type in
// SetTombstoneKey()).
assert(ExtractValueType(current->iter.key()) != kTypeRangeDeletion ||
active_.count(current->level) == 0);
// When entering a new file, range tombstone iter from the old file is
// freed, but the last key from that range tombstone iter may still be in
// the heap. We need to ensure the data underlying its corresponding key
// Slice is still alive. We do so by popping the range tombstone key from
// heap before calling iter->Next(). Technically, this change is not needed:
// if there is a range tombstone end key that is after file boundary
// sentinel key in minHeap_, the range tombstone end key must have been
// truncated at file boundary. The underlying data of the range tombstone
// end key Slice is the SST file's largest internal key stored as file
// metadata in Version. However, since there are too many implicit
// assumptions made, it is safer to just ensure range tombstone iter is
// still alive.
minHeap_.pop();
// Remove last SST file's range tombstone end key if there is one.
// This means file boundary is before range tombstone end key,
// which could happen when a range tombstone and a user key
// straddle two SST files. Note that in TruncatedRangeDelIterator
// constructor, parsed_largest.sequence is decremented 1 in this case.
// Maintains Invariant(rti) that at most one
// pinned_heap_item_[current->level] is in minHeap_.
if (range_tombstone_iters_[current->level] &&
range_tombstone_iters_[current->level]->Valid()) {
if (!minHeap_.empty() && minHeap_.top()->level == current->level) {
assert(minHeap_.top()->type == HeapItem::Type::DELETE_RANGE_END);
minHeap_.pop();
// Invariant(active_): we are about to enter a new SST file with new
// range_tombstone_iters[current->level]. Either it is !Valid() or its
// start key is going to be added to minHeap_.
active_.erase(current->level);
} else {
// range tombstone is still valid, but it is not on heap.
// This should only happen if the range tombstone is over iterator
// upper bound.
assert(iterate_upper_bound_ &&
comparator_->user_comparator()->CompareWithoutTimestamp(
range_tombstone_iters_[current->level]->start_key().user_key,
true /* a_has_ts */, *iterate_upper_bound_,