Line data Source code
1 : // Copyright 2019 The LevelDB-Go and Pebble Authors. All rights reserved. Use
2 : // of this source code is governed by a BSD-style license that can be found in
3 : // the LICENSE file.
4 :
5 : package pebble
6 :
7 : import (
8 : "context"
9 : "fmt"
10 : "io"
11 : "sort"
12 :
13 : "github.com/cockroachdb/errors"
14 : "github.com/cockroachdb/pebble/internal/base"
15 : "github.com/cockroachdb/pebble/internal/keyspan"
16 : "github.com/cockroachdb/pebble/internal/manifest"
17 : )
18 :
19 : // This file implements DB.CheckLevels() which checks that every entry in the
20 : // DB is consistent with respect to the level invariant: any point (or the
21 : // infinite number of points in a range tombstone) has a seqnum such that a
22 : // point with the same UserKey at a lower level has a lower seqnum. This is an
23 : // expensive check since it involves iterating over all the entries in the DB,
24 : // hence only intended for tests or tools.
25 : //
26 : // If we ignore range tombstones, the consistency checking of points can be
27 : // done with a simplified version of mergingIter. simpleMergingIter is that
28 : // simplified version of mergingIter that only needs to step through points
29 : // (analogous to only doing Next()). It can also easily accommodate
30 : // consistency checking of points relative to range tombstones.
31 : // simpleMergingIter does not do any seek optimizations present in mergingIter
32 : // (it minimally needs to seek the range delete iterators to position them at
33 : // or past the current point) since it does not want to miss points for
34 : // purposes of consistency checking.
35 : //
36 : // Mutual consistency of range tombstones is non-trivial to check. One needs
37 : // to detect inversions of the form [a, c)#8 at higher level and [b, c)#10 at
38 : // a lower level. The start key of the former is not contained in the latter
39 : // and we can't use the exclusive end key, c, for a containment check since it
40 : // is the sentinel key. We observe that if these tombstones were fragmented
41 : // wrt each other we would have [a, b)#8 and [b, c)#8 at the higher level and
42 : // [b, c)#10 at the lower level and then it is is trivial to compare the two
43 : // [b, c) tombstones. Note that this fragmentation needs to take into account
44 : // that tombstones in a file may be untruncated and need to act within the
45 : // bounds of the file. This checking is performed by checkRangeTombstones()
46 : // and its helper functions.
47 :
48 : // The per-level structure used by simpleMergingIter.
49 : type simpleMergingIterLevel struct {
50 : iter internalIterator
51 : rangeDelIter keyspan.FragmentIterator
52 : levelIterBoundaryContext
53 :
54 : iterKey *InternalKey
55 : iterValue base.LazyValue
56 : tombstone *keyspan.Span
57 : }
58 :
59 : type simpleMergingIter struct {
60 : levels []simpleMergingIterLevel
61 : snapshot uint64
62 : heap simpleMergingIterHeap
63 : // The last point's key and level. For validation.
64 : lastKey InternalKey
65 : lastLevel int
66 : lastIterMsg string
67 : // A non-nil valueMerger means MERGE record processing is ongoing.
68 : valueMerger base.ValueMerger
69 : // The first error will cause step() to return false.
70 : err error
71 : numPoints int64
72 : merge Merge
73 : formatKey base.FormatKey
74 : }
75 :
76 : func (m *simpleMergingIter) init(
77 : merge Merge,
78 : cmp Compare,
79 : snapshot uint64,
80 : formatKey base.FormatKey,
81 : levels ...simpleMergingIterLevel,
82 1 : ) {
83 1 : m.levels = levels
84 1 : m.formatKey = formatKey
85 1 : m.merge = merge
86 1 : m.snapshot = snapshot
87 1 : m.lastLevel = -1
88 1 : m.heap.cmp = cmp
89 1 : m.heap.items = make([]simpleMergingIterItem, 0, len(levels))
90 1 : for i := range m.levels {
91 1 : l := &m.levels[i]
92 1 : l.iterKey, l.iterValue = l.iter.First()
93 1 : if l.iterKey != nil {
94 1 : item := simpleMergingIterItem{
95 1 : index: i,
96 1 : value: l.iterValue,
97 1 : }
98 1 : item.key.Trailer = l.iterKey.Trailer
99 1 : item.key.UserKey = append(item.key.UserKey[:0], l.iterKey.UserKey...)
100 1 : m.heap.items = append(m.heap.items, item)
101 1 : }
102 : }
103 1 : m.heap.init()
104 1 :
105 1 : if m.heap.len() == 0 {
106 1 : return
107 1 : }
108 1 : m.positionRangeDels()
109 : }
110 :
111 : // Positions all the rangedel iterators at or past the current top of the
112 : // heap, using SeekGE().
113 1 : func (m *simpleMergingIter) positionRangeDels() {
114 1 : item := &m.heap.items[0]
115 1 : for i := range m.levels {
116 1 : l := &m.levels[i]
117 1 : if l.rangeDelIter == nil {
118 1 : continue
119 : }
120 1 : l.tombstone = l.rangeDelIter.SeekGE(item.key.UserKey)
121 : }
122 : }
123 :
124 : // Returns true if not yet done.
125 1 : func (m *simpleMergingIter) step() bool {
126 1 : if m.heap.len() == 0 || m.err != nil {
127 1 : return false
128 1 : }
129 1 : item := &m.heap.items[0]
130 1 : l := &m.levels[item.index]
131 1 : // Sentinels are not relevant for this point checking.
132 1 : if !item.key.IsExclusiveSentinel() && item.key.Visible(m.snapshot, base.InternalKeySeqNumMax) {
133 1 : m.numPoints++
134 1 : keyChanged := m.heap.cmp(item.key.UserKey, m.lastKey.UserKey) != 0
135 1 : if !keyChanged {
136 1 : // At the same user key. We will see them in decreasing seqnum
137 1 : // order so the lastLevel must not be lower.
138 1 : if m.lastLevel > item.index {
139 0 : m.err = errors.Errorf("found InternalKey %s in %s and InternalKey %s in %s",
140 0 : item.key.Pretty(m.formatKey), l.iter, m.lastKey.Pretty(m.formatKey),
141 0 : m.lastIterMsg)
142 0 : return false
143 0 : }
144 1 : m.lastLevel = item.index
145 1 : } else {
146 1 : // The user key has changed.
147 1 : m.lastKey.Trailer = item.key.Trailer
148 1 : m.lastKey.UserKey = append(m.lastKey.UserKey[:0], item.key.UserKey...)
149 1 : m.lastLevel = item.index
150 1 : }
151 : // Ongoing series of MERGE records ends with a MERGE record.
152 1 : if keyChanged && m.valueMerger != nil {
153 1 : var closer io.Closer
154 1 : _, closer, m.err = m.valueMerger.Finish(true /* includesBase */)
155 1 : if m.err == nil && closer != nil {
156 0 : m.err = closer.Close()
157 0 : }
158 1 : m.valueMerger = nil
159 : }
160 1 : itemValue, _, err := item.value.Value(nil)
161 1 : if err != nil {
162 0 : m.err = err
163 0 : return false
164 0 : }
165 1 : if m.valueMerger != nil {
166 1 : // Ongoing series of MERGE records.
167 1 : switch item.key.Kind() {
168 1 : case InternalKeyKindSingleDelete, InternalKeyKindDelete, InternalKeyKindDeleteSized:
169 1 : var closer io.Closer
170 1 : _, closer, m.err = m.valueMerger.Finish(true /* includesBase */)
171 1 : if m.err == nil && closer != nil {
172 0 : m.err = closer.Close()
173 0 : }
174 1 : m.valueMerger = nil
175 1 : case InternalKeyKindSet, InternalKeyKindSetWithDelete:
176 1 : m.err = m.valueMerger.MergeOlder(itemValue)
177 1 : if m.err == nil {
178 1 : var closer io.Closer
179 1 : _, closer, m.err = m.valueMerger.Finish(true /* includesBase */)
180 1 : if m.err == nil && closer != nil {
181 0 : m.err = closer.Close()
182 0 : }
183 : }
184 1 : m.valueMerger = nil
185 1 : case InternalKeyKindMerge:
186 1 : m.err = m.valueMerger.MergeOlder(itemValue)
187 0 : default:
188 0 : m.err = errors.Errorf("pebble: invalid internal key kind %s in %s",
189 0 : item.key.Pretty(m.formatKey),
190 0 : l.iter)
191 0 : return false
192 : }
193 1 : } else if item.key.Kind() == InternalKeyKindMerge && m.err == nil {
194 1 : // New series of MERGE records.
195 1 : m.valueMerger, m.err = m.merge(item.key.UserKey, itemValue)
196 1 : }
197 1 : if m.err != nil {
198 0 : m.err = errors.Wrapf(m.err, "merge processing error on key %s in %s",
199 0 : item.key.Pretty(m.formatKey), l.iter)
200 0 : return false
201 0 : }
202 : // Is this point covered by a tombstone at a lower level? Note that all these
203 : // iterators must be positioned at a key > item.key. So the Largest key bound
204 : // of the sstable containing the tombstone >= item.key. So the upper limit of
205 : // the tombstone cannot be file-bounds-constrained to < item.key. But it is
206 : // possible that item.key < smallest key bound of the sstable, in which case
207 : // this tombstone should be ignored.
208 1 : for level := item.index + 1; level < len(m.levels); level++ {
209 1 : lvl := &m.levels[level]
210 1 : if lvl.rangeDelIter == nil || lvl.tombstone.Empty() {
211 1 : continue
212 : }
213 1 : if (lvl.smallestUserKey == nil || m.heap.cmp(lvl.smallestUserKey, item.key.UserKey) <= 0) &&
214 1 : lvl.tombstone.Contains(m.heap.cmp, item.key.UserKey) {
215 1 : if lvl.tombstone.CoversAt(m.snapshot, item.key.SeqNum()) {
216 0 : m.err = errors.Errorf("tombstone %s in %s deletes key %s in %s",
217 0 : lvl.tombstone.Pretty(m.formatKey), lvl.iter, item.key.Pretty(m.formatKey),
218 0 : l.iter)
219 0 : return false
220 0 : }
221 : }
222 : }
223 : }
224 :
225 : // The iterator for the current level may be closed in the following call to
226 : // Next(). We save its debug string for potential use after it is closed -
227 : // either in this current step() invocation or on the next invocation.
228 1 : m.lastIterMsg = l.iter.String()
229 1 :
230 1 : // Step to the next point.
231 1 : if l.iterKey, l.iterValue = l.iter.Next(); l.iterKey != nil {
232 1 : // Check point keys in an sstable are ordered. Although not required, we check
233 1 : // for memtables as well. A subtle check here is that successive sstables of
234 1 : // L1 and higher levels are ordered. This happens when levelIter moves to the
235 1 : // next sstable in the level, in which case item.key is previous sstable's
236 1 : // last point key.
237 1 : if base.InternalCompare(m.heap.cmp, item.key, *l.iterKey) >= 0 {
238 0 : m.err = errors.Errorf("out of order keys %s >= %s in %s",
239 0 : item.key.Pretty(m.formatKey), l.iterKey.Pretty(m.formatKey), l.iter)
240 0 : return false
241 0 : }
242 1 : item.key.Trailer = l.iterKey.Trailer
243 1 : item.key.UserKey = append(item.key.UserKey[:0], l.iterKey.UserKey...)
244 1 : item.value = l.iterValue
245 1 : if m.heap.len() > 1 {
246 1 : m.heap.fix(0)
247 1 : }
248 1 : } else {
249 1 : m.err = l.iter.Close()
250 1 : l.iter = nil
251 1 : m.heap.pop()
252 1 : }
253 1 : if m.err != nil {
254 0 : return false
255 0 : }
256 1 : if m.heap.len() == 0 {
257 1 : // Last record was a MERGE record.
258 1 : if m.valueMerger != nil {
259 1 : var closer io.Closer
260 1 : _, closer, m.err = m.valueMerger.Finish(true /* includesBase */)
261 1 : if m.err == nil && closer != nil {
262 0 : m.err = closer.Close()
263 0 : }
264 1 : if m.err != nil {
265 0 : m.err = errors.Wrapf(m.err, "merge processing error on key %s in %s",
266 0 : item.key.Pretty(m.formatKey), m.lastIterMsg)
267 0 : }
268 1 : m.valueMerger = nil
269 : }
270 1 : return false
271 : }
272 1 : m.positionRangeDels()
273 1 : return true
274 : }
275 :
276 : // Checking that range tombstones are mutually consistent is performed by checkRangeTombstones().
277 : // See the overview comment at the top of the file.
278 : //
279 : // We do this check as follows:
280 : // - For each level that can have untruncated tombstones, compute the atomic compaction
281 : // bounds (getAtomicUnitBounds()) and use them to truncate tombstones.
282 : // - Now that we have a set of truncated tombstones for each level, put them into one
283 : // pool of tombstones along with their level information (addTombstonesFromIter()).
284 : // - Collect the start and end user keys from all these tombstones (collectAllUserKey()) and use
285 : // them to fragment all the tombstones (fragmentUsingUserKey()).
286 : // - Sort tombstones by start key and decreasing seqnum (tombstonesByStartKeyAndSeqnum) -- all
287 : // tombstones that have the same start key will have the same end key because they have been
288 : // fragmented.
289 : // - Iterate and check (iterateAndCheckTombstones()).
290 : // Note that this simple approach requires holding all the tombstones across all levels in-memory.
291 : // A more sophisticated incremental approach could be devised, if necessary.
292 :
293 : // A tombstone and the corresponding level it was found in.
294 : type tombstoneWithLevel struct {
295 : keyspan.Span
296 : level int
297 : // The level in LSM. A -1 means it's a memtable.
298 : lsmLevel int
299 : fileNum FileNum
300 : }
301 :
302 : // For sorting tombstoneWithLevels in increasing order of start UserKey and
303 : // for the same start UserKey in decreasing order of seqnum.
304 : type tombstonesByStartKeyAndSeqnum struct {
305 : cmp Compare
306 : buf []tombstoneWithLevel
307 : }
308 :
309 1 : func (v *tombstonesByStartKeyAndSeqnum) Len() int { return len(v.buf) }
310 1 : func (v *tombstonesByStartKeyAndSeqnum) Less(i, j int) bool {
311 1 : less := v.cmp(v.buf[i].Start, v.buf[j].Start)
312 1 : if less == 0 {
313 1 : return v.buf[i].LargestSeqNum() > v.buf[j].LargestSeqNum()
314 1 : }
315 1 : return less < 0
316 : }
317 1 : func (v *tombstonesByStartKeyAndSeqnum) Swap(i, j int) {
318 1 : v.buf[i], v.buf[j] = v.buf[j], v.buf[i]
319 1 : }
320 :
321 : func iterateAndCheckTombstones(
322 : cmp Compare, formatKey base.FormatKey, tombstones []tombstoneWithLevel,
323 1 : ) error {
324 1 : sortBuf := tombstonesByStartKeyAndSeqnum{
325 1 : cmp: cmp,
326 1 : buf: tombstones,
327 1 : }
328 1 : sort.Sort(&sortBuf)
329 1 :
330 1 : // For a sequence of tombstones that share the same start UserKey, we will
331 1 : // encounter them in non-increasing seqnum order and so should encounter them
332 1 : // in non-decreasing level order.
333 1 : lastTombstone := tombstoneWithLevel{}
334 1 : for _, t := range tombstones {
335 1 : if cmp(lastTombstone.Start, t.Start) == 0 && lastTombstone.level > t.level {
336 0 : return errors.Errorf("encountered tombstone %s in %s"+
337 0 : " that has a lower seqnum than the same tombstone in %s",
338 0 : t.Span.Pretty(formatKey), levelOrMemtable(t.lsmLevel, t.fileNum),
339 0 : levelOrMemtable(lastTombstone.lsmLevel, lastTombstone.fileNum))
340 0 : }
341 1 : lastTombstone = t
342 : }
343 1 : return nil
344 : }
345 :
346 : type checkConfig struct {
347 : logger Logger
348 : cmp Compare
349 : readState *readState
350 : newIters tableNewIters
351 : seqNum uint64
352 : stats *CheckLevelsStats
353 : merge Merge
354 : formatKey base.FormatKey
355 : }
356 :
357 1 : func checkRangeTombstones(c *checkConfig) error {
358 1 : var level int
359 1 : var tombstones []tombstoneWithLevel
360 1 : var err error
361 1 :
362 1 : memtables := c.readState.memtables
363 1 : for i := len(memtables) - 1; i >= 0; i-- {
364 1 : iter := memtables[i].newRangeDelIter(nil)
365 1 : if iter == nil {
366 1 : continue
367 : }
368 1 : if tombstones, err = addTombstonesFromIter(iter, level, -1, 0, tombstones,
369 1 : c.seqNum, c.cmp, c.formatKey, nil); err != nil {
370 0 : return err
371 0 : }
372 1 : level++
373 : }
374 :
375 1 : current := c.readState.current
376 1 : addTombstonesFromLevel := func(files manifest.LevelIterator, lsmLevel int) error {
377 1 : for f := files.First(); f != nil; f = files.Next() {
378 1 : lf := files.Take()
379 1 : atomicUnit, _ := expandToAtomicUnit(c.cmp, lf.Slice(), true /* disableIsCompacting */)
380 1 : lower, upper := manifest.KeyRange(c.cmp, atomicUnit.Iter())
381 1 : iterToClose, iter, err := c.newIters(
382 1 : context.Background(), lf.FileMetadata, &IterOptions{level: manifest.Level(lsmLevel)}, internalIterOpts{})
383 1 : if err != nil {
384 0 : return err
385 0 : }
386 1 : iterToClose.Close()
387 1 : if iter == nil {
388 1 : continue
389 : }
390 1 : truncate := func(t keyspan.Span) keyspan.Span {
391 1 : // Same checks as in keyspan.Truncate.
392 1 : if c.cmp(t.Start, lower.UserKey) < 0 {
393 0 : t.Start = lower.UserKey
394 0 : }
395 1 : if c.cmp(t.End, upper.UserKey) > 0 {
396 0 : t.End = upper.UserKey
397 0 : }
398 1 : if c.cmp(t.Start, t.End) >= 0 {
399 0 : // Remove the keys.
400 0 : t.Keys = t.Keys[:0]
401 0 : }
402 1 : return t
403 : }
404 1 : if tombstones, err = addTombstonesFromIter(iter, level, lsmLevel, f.FileNum,
405 1 : tombstones, c.seqNum, c.cmp, c.formatKey, truncate); err != nil {
406 0 : return err
407 0 : }
408 : }
409 1 : return nil
410 : }
411 : // Now the levels with untruncated tombsones.
412 1 : for i := len(current.L0SublevelFiles) - 1; i >= 0; i-- {
413 1 : if current.L0SublevelFiles[i].Empty() {
414 0 : continue
415 : }
416 1 : err := addTombstonesFromLevel(current.L0SublevelFiles[i].Iter(), 0)
417 1 : if err != nil {
418 0 : return err
419 0 : }
420 1 : level++
421 : }
422 1 : for i := 1; i < len(current.Levels); i++ {
423 1 : if err := addTombstonesFromLevel(current.Levels[i].Iter(), i); err != nil {
424 0 : return err
425 0 : }
426 1 : level++
427 : }
428 1 : if c.stats != nil {
429 0 : c.stats.NumTombstones = len(tombstones)
430 0 : }
431 : // We now have truncated tombstones.
432 : // Fragment them all.
433 1 : userKeys := collectAllUserKeys(c.cmp, tombstones)
434 1 : tombstones = fragmentUsingUserKeys(c.cmp, tombstones, userKeys)
435 1 : return iterateAndCheckTombstones(c.cmp, c.formatKey, tombstones)
436 : }
437 :
438 0 : func levelOrMemtable(lsmLevel int, fileNum FileNum) string {
439 0 : if lsmLevel == -1 {
440 0 : return "memtable"
441 0 : }
442 0 : return fmt.Sprintf("L%d: fileNum=%s", lsmLevel, fileNum)
443 : }
444 :
445 : func addTombstonesFromIter(
446 : iter keyspan.FragmentIterator,
447 : level int,
448 : lsmLevel int,
449 : fileNum FileNum,
450 : tombstones []tombstoneWithLevel,
451 : seqNum uint64,
452 : cmp Compare,
453 : formatKey base.FormatKey,
454 : truncate func(tombstone keyspan.Span) keyspan.Span,
455 1 : ) (_ []tombstoneWithLevel, err error) {
456 1 : defer func() {
457 1 : err = firstError(err, iter.Close())
458 1 : }()
459 :
460 1 : var prevTombstone keyspan.Span
461 1 : for tomb := iter.First(); tomb != nil; tomb = iter.Next() {
462 1 : t := tomb.Visible(seqNum)
463 1 : if t.Empty() {
464 1 : continue
465 : }
466 1 : t = t.DeepClone()
467 1 : // This is mainly a test for rangeDelV2 formatted blocks which are expected to
468 1 : // be ordered and fragmented on disk. But we anyways check for memtables,
469 1 : // rangeDelV1 as well.
470 1 : if cmp(prevTombstone.End, t.Start) > 0 {
471 0 : return nil, errors.Errorf("unordered or unfragmented range delete tombstones %s, %s in %s",
472 0 : prevTombstone.Pretty(formatKey), t.Pretty(formatKey), levelOrMemtable(lsmLevel, fileNum))
473 0 : }
474 1 : prevTombstone = t
475 1 :
476 1 : // Truncation of a tombstone must happen after checking its ordering,
477 1 : // fragmentation wrt previous tombstone. Since it is possible that after
478 1 : // truncation the tombstone is ordered, fragmented when it originally wasn't.
479 1 : if truncate != nil {
480 1 : t = truncate(t)
481 1 : }
482 1 : if !t.Empty() {
483 1 : tombstones = append(tombstones, tombstoneWithLevel{
484 1 : Span: t,
485 1 : level: level,
486 1 : lsmLevel: lsmLevel,
487 1 : fileNum: fileNum,
488 1 : })
489 1 : }
490 : }
491 1 : return tombstones, nil
492 : }
493 :
494 : type userKeysSort struct {
495 : cmp Compare
496 : buf [][]byte
497 : }
498 :
499 1 : func (v *userKeysSort) Len() int { return len(v.buf) }
500 1 : func (v *userKeysSort) Less(i, j int) bool {
501 1 : return v.cmp(v.buf[i], v.buf[j]) < 0
502 1 : }
503 1 : func (v *userKeysSort) Swap(i, j int) {
504 1 : v.buf[i], v.buf[j] = v.buf[j], v.buf[i]
505 1 : }
506 1 : func collectAllUserKeys(cmp Compare, tombstones []tombstoneWithLevel) [][]byte {
507 1 : keys := make([][]byte, 0, len(tombstones)*2)
508 1 : for _, t := range tombstones {
509 1 : keys = append(keys, t.Start)
510 1 : keys = append(keys, t.End)
511 1 : }
512 1 : sorter := userKeysSort{
513 1 : cmp: cmp,
514 1 : buf: keys,
515 1 : }
516 1 : sort.Sort(&sorter)
517 1 : var last, curr int
518 1 : for last, curr = -1, 0; curr < len(keys); curr++ {
519 1 : if last < 0 || cmp(keys[last], keys[curr]) != 0 {
520 1 : last++
521 1 : keys[last] = keys[curr]
522 1 : }
523 : }
524 1 : keys = keys[:last+1]
525 1 : return keys
526 : }
527 :
528 : func fragmentUsingUserKeys(
529 : cmp Compare, tombstones []tombstoneWithLevel, userKeys [][]byte,
530 1 : ) []tombstoneWithLevel {
531 1 : var buf []tombstoneWithLevel
532 1 : for _, t := range tombstones {
533 1 : // Find the first position with tombstone start < user key
534 1 : i := sort.Search(len(userKeys), func(i int) bool {
535 1 : return cmp(t.Start, userKeys[i]) < 0
536 1 : })
537 1 : for ; i < len(userKeys); i++ {
538 1 : if cmp(userKeys[i], t.End) >= 0 {
539 1 : break
540 : }
541 1 : tPartial := t
542 1 : tPartial.End = userKeys[i]
543 1 : buf = append(buf, tPartial)
544 1 : t.Start = userKeys[i]
545 : }
546 1 : buf = append(buf, t)
547 : }
548 1 : return buf
549 : }
550 :
551 : // CheckLevelsStats provides basic stats on points and tombstones encountered.
552 : type CheckLevelsStats struct {
553 : NumPoints int64
554 : NumTombstones int
555 : }
556 :
557 : // CheckLevels checks:
558 : // - Every entry in the DB is consistent with the level invariant. See the
559 : // comment at the top of the file.
560 : // - Point keys in sstables are ordered.
561 : // - Range delete tombstones in sstables are ordered and fragmented.
562 : // - Successful processing of all MERGE records.
563 1 : func (d *DB) CheckLevels(stats *CheckLevelsStats) error {
564 1 : // Grab and reference the current readState.
565 1 : readState := d.loadReadState()
566 1 : defer readState.unref()
567 1 :
568 1 : // Determine the seqnum to read at after grabbing the read state (current and
569 1 : // memtables) above.
570 1 : seqNum := d.mu.versions.visibleSeqNum.Load()
571 1 :
572 1 : checkConfig := &checkConfig{
573 1 : logger: d.opts.Logger,
574 1 : cmp: d.cmp,
575 1 : readState: readState,
576 1 : newIters: d.newIters,
577 1 : seqNum: seqNum,
578 1 : stats: stats,
579 1 : merge: d.merge,
580 1 : formatKey: d.opts.Comparer.FormatKey,
581 1 : }
582 1 : return checkLevelsInternal(checkConfig)
583 1 : }
584 :
585 1 : func checkLevelsInternal(c *checkConfig) (err error) {
586 1 : // Phase 1: Use a simpleMergingIter to step through all the points and ensure
587 1 : // that points with the same user key at different levels are not inverted
588 1 : // wrt sequence numbers and the same holds for tombstones that cover points.
589 1 : // To do this, one needs to construct a simpleMergingIter which is similar to
590 1 : // how one constructs a mergingIter.
591 1 :
592 1 : // Add mem tables from newest to oldest.
593 1 : var mlevels []simpleMergingIterLevel
594 1 : defer func() {
595 1 : for i := range mlevels {
596 1 : l := &mlevels[i]
597 1 : if l.iter != nil {
598 1 : err = firstError(err, l.iter.Close())
599 1 : l.iter = nil
600 1 : }
601 1 : if l.rangeDelIter != nil {
602 1 : err = firstError(err, l.rangeDelIter.Close())
603 1 : l.rangeDelIter = nil
604 1 : }
605 : }
606 : }()
607 :
608 1 : memtables := c.readState.memtables
609 1 : for i := len(memtables) - 1; i >= 0; i-- {
610 1 : mem := memtables[i]
611 1 : mlevels = append(mlevels, simpleMergingIterLevel{
612 1 : iter: mem.newIter(nil),
613 1 : rangeDelIter: mem.newRangeDelIter(nil),
614 1 : })
615 1 : }
616 :
617 1 : current := c.readState.current
618 1 : // Determine the final size for mlevels so that there are no more
619 1 : // reallocations. levelIter will hold a pointer to elements in mlevels.
620 1 : start := len(mlevels)
621 1 : for sublevel := len(current.L0SublevelFiles) - 1; sublevel >= 0; sublevel-- {
622 1 : if current.L0SublevelFiles[sublevel].Empty() {
623 0 : continue
624 : }
625 1 : mlevels = append(mlevels, simpleMergingIterLevel{})
626 : }
627 1 : for level := 1; level < len(current.Levels); level++ {
628 1 : if current.Levels[level].Empty() {
629 1 : continue
630 : }
631 1 : mlevels = append(mlevels, simpleMergingIterLevel{})
632 : }
633 1 : mlevelAlloc := mlevels[start:]
634 1 : // Add L0 files by sublevel.
635 1 : for sublevel := len(current.L0SublevelFiles) - 1; sublevel >= 0; sublevel-- {
636 1 : if current.L0SublevelFiles[sublevel].Empty() {
637 0 : continue
638 : }
639 1 : manifestIter := current.L0SublevelFiles[sublevel].Iter()
640 1 : iterOpts := IterOptions{logger: c.logger}
641 1 : li := &levelIter{}
642 1 : li.init(context.Background(), iterOpts, c.cmp, nil /* split */, c.newIters, manifestIter,
643 1 : manifest.L0Sublevel(sublevel), internalIterOpts{})
644 1 : li.initRangeDel(&mlevelAlloc[0].rangeDelIter)
645 1 : li.initBoundaryContext(&mlevelAlloc[0].levelIterBoundaryContext)
646 1 : mlevelAlloc[0].iter = li
647 1 : mlevelAlloc = mlevelAlloc[1:]
648 : }
649 1 : for level := 1; level < len(current.Levels); level++ {
650 1 : if current.Levels[level].Empty() {
651 1 : continue
652 : }
653 :
654 1 : iterOpts := IterOptions{logger: c.logger}
655 1 : li := &levelIter{}
656 1 : li.init(context.Background(), iterOpts, c.cmp, nil /* split */, c.newIters,
657 1 : current.Levels[level].Iter(), manifest.Level(level), internalIterOpts{})
658 1 : li.initRangeDel(&mlevelAlloc[0].rangeDelIter)
659 1 : li.initBoundaryContext(&mlevelAlloc[0].levelIterBoundaryContext)
660 1 : mlevelAlloc[0].iter = li
661 1 : mlevelAlloc = mlevelAlloc[1:]
662 : }
663 :
664 1 : mergingIter := &simpleMergingIter{}
665 1 : mergingIter.init(c.merge, c.cmp, c.seqNum, c.formatKey, mlevels...)
666 1 : for cont := mergingIter.step(); cont; cont = mergingIter.step() {
667 1 : }
668 1 : if err := mergingIter.err; err != nil {
669 0 : return err
670 0 : }
671 1 : if c.stats != nil {
672 0 : c.stats.NumPoints = mergingIter.numPoints
673 0 : }
674 :
675 : // Phase 2: Check that the tombstones are mutually consistent.
676 1 : return checkRangeTombstones(c)
677 : }
678 :
679 : type simpleMergingIterItem struct {
680 : index int
681 : key InternalKey
682 : value base.LazyValue
683 : }
684 :
685 : type simpleMergingIterHeap struct {
686 : cmp Compare
687 : reverse bool
688 : items []simpleMergingIterItem
689 : }
690 :
691 1 : func (h *simpleMergingIterHeap) len() int {
692 1 : return len(h.items)
693 1 : }
694 :
695 1 : func (h *simpleMergingIterHeap) less(i, j int) bool {
696 1 : ikey, jkey := h.items[i].key, h.items[j].key
697 1 : if c := h.cmp(ikey.UserKey, jkey.UserKey); c != 0 {
698 1 : if h.reverse {
699 0 : return c > 0
700 0 : }
701 1 : return c < 0
702 : }
703 1 : if h.reverse {
704 0 : return ikey.Trailer < jkey.Trailer
705 0 : }
706 1 : return ikey.Trailer > jkey.Trailer
707 : }
708 :
709 1 : func (h *simpleMergingIterHeap) swap(i, j int) {
710 1 : h.items[i], h.items[j] = h.items[j], h.items[i]
711 1 : }
712 :
713 : // init, fix, up and down are copied from the go stdlib.
714 1 : func (h *simpleMergingIterHeap) init() {
715 1 : // heapify
716 1 : n := h.len()
717 1 : for i := n/2 - 1; i >= 0; i-- {
718 1 : h.down(i, n)
719 1 : }
720 : }
721 :
722 1 : func (h *simpleMergingIterHeap) fix(i int) {
723 1 : if !h.down(i, h.len()) {
724 1 : h.up(i)
725 1 : }
726 : }
727 :
728 1 : func (h *simpleMergingIterHeap) pop() *simpleMergingIterItem {
729 1 : n := h.len() - 1
730 1 : h.swap(0, n)
731 1 : h.down(0, n)
732 1 : item := &h.items[n]
733 1 : h.items = h.items[:n]
734 1 : return item
735 1 : }
736 :
737 1 : func (h *simpleMergingIterHeap) up(j int) {
738 1 : for {
739 1 : i := (j - 1) / 2 // parent
740 1 : if i == j || !h.less(j, i) {
741 1 : break
742 : }
743 0 : h.swap(i, j)
744 0 : j = i
745 : }
746 : }
747 :
748 1 : func (h *simpleMergingIterHeap) down(i0, n int) bool {
749 1 : i := i0
750 1 : for {
751 1 : j1 := 2*i + 1
752 1 : if j1 >= n || j1 < 0 { // j1 < 0 after int overflow
753 1 : break
754 : }
755 1 : j := j1 // left child
756 1 : if j2 := j1 + 1; j2 < n && h.less(j2, j1) {
757 1 : j = j2 // = 2*i + 2 // right child
758 1 : }
759 1 : if !h.less(j, i) {
760 1 : break
761 : }
762 1 : h.swap(i, j)
763 1 : i = j
764 : }
765 1 : return i > i0
766 : }
|
