-
Notifications
You must be signed in to change notification settings - Fork 452
/
compaction.go
3137 lines (2905 loc) · 112 KB
/
compaction.go
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 2013 The LevelDB-Go and Pebble Authors. All rights reserved. Use
// of this source code is governed by a BSD-style license that can be found in
// the LICENSE file.
package pebble
import (
"bytes"
"context"
"fmt"
"math"
"runtime/pprof"
"slices"
"sort"
"sync/atomic"
"time"
"github.com/cockroachdb/crlib/crtime"
"github.com/cockroachdb/errors"
"github.com/cockroachdb/pebble/internal/base"
"github.com/cockroachdb/pebble/internal/compact"
"github.com/cockroachdb/pebble/internal/keyspan"
"github.com/cockroachdb/pebble/internal/keyspan/keyspanimpl"
"github.com/cockroachdb/pebble/internal/manifest"
"github.com/cockroachdb/pebble/internal/sstableinternal"
"github.com/cockroachdb/pebble/objstorage"
"github.com/cockroachdb/pebble/objstorage/objstorageprovider/objiotracing"
"github.com/cockroachdb/pebble/objstorage/remote"
"github.com/cockroachdb/pebble/sstable"
"github.com/cockroachdb/pebble/vfs"
)
var errEmptyTable = errors.New("pebble: empty table")
// ErrCancelledCompaction is returned if a compaction is cancelled by a
// concurrent excise or ingest-split operation.
var ErrCancelledCompaction = errors.New("pebble: compaction cancelled by a concurrent operation, will retry compaction")
var compactLabels = pprof.Labels("pebble", "compact")
var flushLabels = pprof.Labels("pebble", "flush")
var gcLabels = pprof.Labels("pebble", "gc")
// expandedCompactionByteSizeLimit is the maximum number of bytes in all
// compacted files. We avoid expanding the lower level file set of a compaction
// if it would make the total compaction cover more than this many bytes.
func expandedCompactionByteSizeLimit(opts *Options, level int, availBytes uint64) uint64 {
v := uint64(25 * opts.Level(level).TargetFileSize)
// Never expand a compaction beyond half the available capacity, divided
// by the maximum number of concurrent compactions. Each of the concurrent
// compactions may expand up to this limit, so this attempts to limit
// compactions to half of available disk space. Note that this will not
// prevent compaction picking from pursuing compactions that are larger
// than this threshold before expansion.
diskMax := (availBytes / 2) / uint64(opts.MaxConcurrentCompactions())
if v > diskMax {
v = diskMax
}
return v
}
// maxGrandparentOverlapBytes is the maximum bytes of overlap with level+1
// before we stop building a single file in a level-1 to level compaction.
func maxGrandparentOverlapBytes(opts *Options, level int) uint64 {
return uint64(10 * opts.Level(level).TargetFileSize)
}
// maxReadCompactionBytes is used to prevent read compactions which
// are too wide.
func maxReadCompactionBytes(opts *Options, level int) uint64 {
return uint64(10 * opts.Level(level).TargetFileSize)
}
// noCloseIter wraps around a FragmentIterator, intercepting and eliding
// calls to Close. It is used during compaction to ensure that rangeDelIters
// are not closed prematurely.
type noCloseIter struct {
keyspan.FragmentIterator
}
func (i *noCloseIter) Close() {}
type compactionLevel struct {
level int
files manifest.LevelSlice
// l0SublevelInfo contains information about L0 sublevels being compacted.
// It's only set for the start level of a compaction starting out of L0 and
// is nil for all other compactions.
l0SublevelInfo []sublevelInfo
}
func (cl compactionLevel) Clone() compactionLevel {
newCL := compactionLevel{
level: cl.level,
files: cl.files,
}
return newCL
}
func (cl compactionLevel) String() string {
return fmt.Sprintf(`Level %d, Files %s`, cl.level, cl.files)
}
// compactionWritable is a objstorage.Writable wrapper that, on every write,
// updates a metric in `versions` on bytes written by in-progress compactions so
// far. It also increments a per-compaction `written` int.
type compactionWritable struct {
objstorage.Writable
versions *versionSet
written *int64
}
// Write is part of the objstorage.Writable interface.
func (c *compactionWritable) Write(p []byte) error {
if err := c.Writable.Write(p); err != nil {
return err
}
*c.written += int64(len(p))
c.versions.incrementCompactionBytes(int64(len(p)))
return nil
}
type compactionKind int
const (
compactionKindDefault compactionKind = iota
compactionKindFlush
// compactionKindMove denotes a move compaction where the input file is
// retained and linked in a new level without being obsoleted.
compactionKindMove
// compactionKindCopy denotes a copy compaction where the input file is
// copied byte-by-byte into a new file with a new FileNum in the output level.
compactionKindCopy
// compactionKindDeleteOnly denotes a compaction that only deletes input
// files. It can occur when wide range tombstones completely contain sstables.
compactionKindDeleteOnly
compactionKindElisionOnly
compactionKindRead
compactionKindTombstoneDensity
compactionKindRewrite
compactionKindIngestedFlushable
)
func (k compactionKind) String() string {
switch k {
case compactionKindDefault:
return "default"
case compactionKindFlush:
return "flush"
case compactionKindMove:
return "move"
case compactionKindDeleteOnly:
return "delete-only"
case compactionKindElisionOnly:
return "elision-only"
case compactionKindRead:
return "read"
case compactionKindTombstoneDensity:
return "tombstone-density"
case compactionKindRewrite:
return "rewrite"
case compactionKindIngestedFlushable:
return "ingested-flushable"
case compactionKindCopy:
return "copy"
}
return "?"
}
// compaction is a table compaction from one level to the next, starting from a
// given version.
type compaction struct {
// cancel is a bool that can be used by other goroutines to signal a compaction
// to cancel, such as if a conflicting excise operation raced it to manifest
// application. Only holders of the manifest lock will write to this atomic.
cancel atomic.Bool
kind compactionKind
// isDownload is true if this compaction was started as part of a Download
// operation. In this case kind is compactionKindCopy or
// compactionKindRewrite.
isDownload bool
cmp Compare
equal Equal
comparer *base.Comparer
formatKey base.FormatKey
logger Logger
version *version
stats base.InternalIteratorStats
beganAt time.Time
// versionEditApplied is set to true when a compaction has completed and the
// resulting version has been installed (if successful), but the compaction
// goroutine is still cleaning up (eg, deleting obsolete files).
versionEditApplied bool
bufferPool sstable.BufferPool
// startLevel is the level that is being compacted. Inputs from startLevel
// and outputLevel will be merged to produce a set of outputLevel files.
startLevel *compactionLevel
// outputLevel is the level that files are being produced in. outputLevel is
// equal to startLevel+1 except when:
// - if startLevel is 0, the output level equals compactionPicker.baseLevel().
// - in multilevel compaction, the output level is the lowest level involved in
// the compaction
// A compaction's outputLevel is nil for delete-only compactions.
outputLevel *compactionLevel
// extraLevels point to additional levels in between the input and output
// levels that get compacted in multilevel compactions
extraLevels []*compactionLevel
inputs []compactionLevel
// maxOutputFileSize is the maximum size of an individual table created
// during compaction.
maxOutputFileSize uint64
// maxOverlapBytes is the maximum number of bytes of overlap allowed for a
// single output table with the tables in the grandparent level.
maxOverlapBytes uint64
// flushing contains the flushables (aka memtables) that are being flushed.
flushing flushableList
// bytesWritten contains the number of bytes that have been written to outputs.
bytesWritten int64
// The boundaries of the input data.
smallest InternalKey
largest InternalKey
// A list of fragment iterators to close when the compaction finishes. Used by
// input iteration to keep rangeDelIters open for the lifetime of the
// compaction, and only close them when the compaction finishes.
closers []*noCloseIter
// grandparents are the tables in level+2 that overlap with the files being
// compacted. Used to determine output table boundaries. Do not assume that the actual files
// in the grandparent when this compaction finishes will be the same.
grandparents manifest.LevelSlice
// Boundaries at which flushes to L0 should be split. Determined by
// L0Sublevels. If nil, flushes aren't split.
l0Limits [][]byte
delElision compact.TombstoneElision
rangeKeyElision compact.TombstoneElision
// allowedZeroSeqNum is true if seqnums can be zeroed if there are no
// snapshots requiring them to be kept. This determination is made by
// looking for an sstable which overlaps the bounds of the compaction at a
// lower level in the LSM during runCompaction.
allowedZeroSeqNum bool
// deletionHints are set if this is a compactionKindDeleteOnly. Used to figure
// out whether an input must be deleted in its entirety, or excised into
// virtual sstables.
deletionHints []deleteCompactionHint
// exciseEnabled is set to true if this is a compactionKindDeleteOnly and
// this compaction is allowed to excise files.
exciseEnabled bool
metrics map[int]*LevelMetrics
pickerMetrics compactionPickerMetrics
}
// inputLargestSeqNumAbsolute returns the maximum LargestSeqNumAbsolute of any
// input sstables.
func (c *compaction) inputLargestSeqNumAbsolute() base.SeqNum {
var seqNum base.SeqNum
for _, cl := range c.inputs {
cl.files.Each(func(m *manifest.FileMetadata) {
seqNum = max(seqNum, m.LargestSeqNumAbsolute)
})
}
return seqNum
}
func (c *compaction) makeInfo(jobID JobID) CompactionInfo {
info := CompactionInfo{
JobID: int(jobID),
Reason: c.kind.String(),
Input: make([]LevelInfo, 0, len(c.inputs)),
Annotations: []string{},
}
if c.isDownload {
info.Reason = "download," + info.Reason
}
for _, cl := range c.inputs {
inputInfo := LevelInfo{Level: cl.level, Tables: nil}
iter := cl.files.Iter()
for m := iter.First(); m != nil; m = iter.Next() {
inputInfo.Tables = append(inputInfo.Tables, m.TableInfo())
}
info.Input = append(info.Input, inputInfo)
}
if c.outputLevel != nil {
info.Output.Level = c.outputLevel.level
// If there are no inputs from the output level (eg, a move
// compaction), add an empty LevelInfo to info.Input.
if len(c.inputs) > 0 && c.inputs[len(c.inputs)-1].level != c.outputLevel.level {
info.Input = append(info.Input, LevelInfo{Level: c.outputLevel.level})
}
} else {
// For a delete-only compaction, set the output level to L6. The
// output level is not meaningful here, but complicating the
// info.Output interface with a pointer doesn't seem worth the
// semantic distinction.
info.Output.Level = numLevels - 1
}
for i, score := range c.pickerMetrics.scores {
info.Input[i].Score = score
}
info.SingleLevelOverlappingRatio = c.pickerMetrics.singleLevelOverlappingRatio
info.MultiLevelOverlappingRatio = c.pickerMetrics.multiLevelOverlappingRatio
if len(info.Input) > 2 {
info.Annotations = append(info.Annotations, "multilevel")
}
return info
}
func (c *compaction) userKeyBounds() base.UserKeyBounds {
return base.UserKeyBoundsFromInternal(c.smallest, c.largest)
}
func newCompaction(
pc *pickedCompaction, opts *Options, beganAt time.Time, provider objstorage.Provider,
) *compaction {
c := &compaction{
kind: compactionKindDefault,
cmp: pc.cmp,
equal: opts.Comparer.Equal,
comparer: opts.Comparer,
formatKey: opts.Comparer.FormatKey,
inputs: pc.inputs,
smallest: pc.smallest,
largest: pc.largest,
logger: opts.Logger,
version: pc.version,
beganAt: beganAt,
maxOutputFileSize: pc.maxOutputFileSize,
maxOverlapBytes: pc.maxOverlapBytes,
pickerMetrics: pc.pickerMetrics,
}
c.startLevel = &c.inputs[0]
if pc.startLevel.l0SublevelInfo != nil {
c.startLevel.l0SublevelInfo = pc.startLevel.l0SublevelInfo
}
c.outputLevel = &c.inputs[1]
if len(pc.extraLevels) > 0 {
c.extraLevels = pc.extraLevels
c.outputLevel = &c.inputs[len(c.inputs)-1]
}
// Compute the set of outputLevel+1 files that overlap this compaction (these
// are the grandparent sstables).
if c.outputLevel.level+1 < numLevels {
c.grandparents = c.version.Overlaps(c.outputLevel.level+1, c.userKeyBounds())
}
c.delElision, c.rangeKeyElision = compact.SetupTombstoneElision(
c.cmp, c.version, c.outputLevel.level, base.UserKeyBoundsFromInternal(c.smallest, c.largest),
)
c.kind = pc.kind
if c.kind == compactionKindDefault && c.outputLevel.files.Empty() && !c.hasExtraLevelData() &&
c.startLevel.files.Len() == 1 && c.grandparents.SizeSum() <= c.maxOverlapBytes {
// This compaction can be converted into a move or copy from one level
// to the next. We avoid such a move if there is lots of overlapping
// grandparent data. Otherwise, the move could create a parent file
// that will require a very expensive merge later on.
iter := c.startLevel.files.Iter()
meta := iter.First()
isRemote := false
// We should always be passed a provider, except in some unit tests.
if provider != nil {
isRemote = !objstorage.IsLocalTable(provider, meta.FileBacking.DiskFileNum)
}
// Avoid a trivial move or copy if all of these are true, as rewriting a
// new file is better:
//
// 1) The source file is a virtual sstable
// 2) The existing file `meta` is on non-remote storage
// 3) The output level prefers shared storage
mustCopy := !isRemote && remote.ShouldCreateShared(opts.Experimental.CreateOnShared, c.outputLevel.level)
if mustCopy {
// If the source is virtual, it's best to just rewrite the file as all
// conditions in the above comment are met.
if !meta.Virtual {
c.kind = compactionKindCopy
}
} else {
c.kind = compactionKindMove
}
}
return c
}
func newDeleteOnlyCompaction(
opts *Options,
cur *version,
inputs []compactionLevel,
beganAt time.Time,
hints []deleteCompactionHint,
exciseEnabled bool,
) *compaction {
c := &compaction{
kind: compactionKindDeleteOnly,
cmp: opts.Comparer.Compare,
equal: opts.Comparer.Equal,
comparer: opts.Comparer,
formatKey: opts.Comparer.FormatKey,
logger: opts.Logger,
version: cur,
beganAt: beganAt,
inputs: inputs,
deletionHints: hints,
exciseEnabled: exciseEnabled,
}
// Set c.smallest, c.largest.
files := make([]manifest.LevelIterator, 0, len(inputs))
for _, in := range inputs {
files = append(files, in.files.Iter())
}
c.smallest, c.largest = manifest.KeyRange(opts.Comparer.Compare, files...)
return c
}
func adjustGrandparentOverlapBytesForFlush(c *compaction, flushingBytes uint64) {
// Heuristic to place a lower bound on compaction output file size
// caused by Lbase. Prior to this heuristic we have observed an L0 in
// production with 310K files of which 290K files were < 10KB in size.
// Our hypothesis is that it was caused by L1 having 2600 files and
// ~10GB, such that each flush got split into many tiny files due to
// overlapping with most of the files in Lbase.
//
// The computation below is general in that it accounts
// for flushing different volumes of data (e.g. we may be flushing
// many memtables). For illustration, we consider the typical
// example of flushing a 64MB memtable. So 12.8MB output,
// based on the compression guess below. If the compressed bytes
// guess is an over-estimate we will end up with smaller files,
// and if an under-estimate we will end up with larger files.
// With a 2MB target file size, 7 files. We are willing to accept
// 4x the number of files, if it results in better write amplification
// when later compacting to Lbase, i.e., ~450KB files (target file
// size / 4).
//
// Note that this is a pessimistic heuristic in that
// fileCountUpperBoundDueToGrandparents could be far from the actual
// number of files produced due to the grandparent limits. For
// example, in the extreme, consider a flush that overlaps with 1000
// files in Lbase f0...f999, and the initially calculated value of
// maxOverlapBytes will cause splits at f10, f20,..., f990, which
// means an upper bound file count of 100 files. Say the input bytes
// in the flush are such that acceptableFileCount=10. We will fatten
// up maxOverlapBytes by 10x to ensure that the upper bound file count
// drops to 10. However, it is possible that in practice, even without
// this change, we would have produced no more than 10 files, and that
// this change makes the files unnecessarily wide. Say the input bytes
// are distributed such that 10% are in f0...f9, 10% in f10...f19, ...
// 10% in f80...f89 and 10% in f990...f999. The original value of
// maxOverlapBytes would have actually produced only 10 sstables. But
// by increasing maxOverlapBytes by 10x, we may produce 1 sstable that
// spans f0...f89, i.e., a much wider sstable than necessary.
//
// We could produce a tighter estimate of
// fileCountUpperBoundDueToGrandparents if we had knowledge of the key
// distribution of the flush. The 4x multiplier mentioned earlier is
// a way to try to compensate for this pessimism.
//
// TODO(sumeer): we don't have compression info for the data being
// flushed, but it is likely that existing files that overlap with
// this flush in Lbase are representative wrt compression ratio. We
// could store the uncompressed size in FileMetadata and estimate
// the compression ratio.
const approxCompressionRatio = 0.2
approxOutputBytes := approxCompressionRatio * float64(flushingBytes)
approxNumFilesBasedOnTargetSize :=
int(math.Ceil(approxOutputBytes / float64(c.maxOutputFileSize)))
acceptableFileCount := float64(4 * approxNumFilesBasedOnTargetSize)
// The byte calculation is linear in numGrandparentFiles, but we will
// incur this linear cost in compact.Runner.TableSplitLimit() too, so we are
// also willing to pay it now. We could approximate this cheaply by using the
// mean file size of Lbase.
grandparentFileBytes := c.grandparents.SizeSum()
fileCountUpperBoundDueToGrandparents :=
float64(grandparentFileBytes) / float64(c.maxOverlapBytes)
if fileCountUpperBoundDueToGrandparents > acceptableFileCount {
c.maxOverlapBytes = uint64(
float64(c.maxOverlapBytes) *
(fileCountUpperBoundDueToGrandparents / acceptableFileCount))
}
}
func newFlush(
opts *Options, cur *version, baseLevel int, flushing flushableList, beganAt time.Time,
) (*compaction, error) {
c := &compaction{
kind: compactionKindFlush,
cmp: opts.Comparer.Compare,
equal: opts.Comparer.Equal,
comparer: opts.Comparer,
formatKey: opts.Comparer.FormatKey,
logger: opts.Logger,
version: cur,
beganAt: beganAt,
inputs: []compactionLevel{{level: -1}, {level: 0}},
maxOutputFileSize: math.MaxUint64,
maxOverlapBytes: math.MaxUint64,
flushing: flushing,
}
c.startLevel = &c.inputs[0]
c.outputLevel = &c.inputs[1]
if len(flushing) > 0 {
if _, ok := flushing[0].flushable.(*ingestedFlushable); ok {
if len(flushing) != 1 {
panic("pebble: ingestedFlushable must be flushed one at a time.")
}
c.kind = compactionKindIngestedFlushable
return c, nil
}
}
// Make sure there's no ingestedFlushable after the first flushable in the
// list.
for _, f := range flushing {
if _, ok := f.flushable.(*ingestedFlushable); ok {
panic("pebble: flushing shouldn't contain ingestedFlushable flushable")
}
}
if cur.L0Sublevels != nil {
c.l0Limits = cur.L0Sublevels.FlushSplitKeys()
}
smallestSet, largestSet := false, false
updatePointBounds := func(iter internalIterator) {
if kv := iter.First(); kv != nil {
if !smallestSet ||
base.InternalCompare(c.cmp, c.smallest, kv.K) > 0 {
smallestSet = true
c.smallest = kv.K.Clone()
}
}
if kv := iter.Last(); kv != nil {
if !largestSet ||
base.InternalCompare(c.cmp, c.largest, kv.K) < 0 {
largestSet = true
c.largest = kv.K.Clone()
}
}
}
updateRangeBounds := func(iter keyspan.FragmentIterator) error {
// File bounds require s != nil && !s.Empty(). We only need to check for
// s != nil here, as the memtable's FragmentIterator would never surface
// empty spans.
if s, err := iter.First(); err != nil {
return err
} else if s != nil {
if key := s.SmallestKey(); !smallestSet ||
base.InternalCompare(c.cmp, c.smallest, key) > 0 {
smallestSet = true
c.smallest = key.Clone()
}
}
if s, err := iter.Last(); err != nil {
return err
} else if s != nil {
if key := s.LargestKey(); !largestSet ||
base.InternalCompare(c.cmp, c.largest, key) < 0 {
largestSet = true
c.largest = key.Clone()
}
}
return nil
}
var flushingBytes uint64
for i := range flushing {
f := flushing[i]
updatePointBounds(f.newIter(nil))
if rangeDelIter := f.newRangeDelIter(nil); rangeDelIter != nil {
if err := updateRangeBounds(rangeDelIter); err != nil {
return nil, err
}
}
if rangeKeyIter := f.newRangeKeyIter(nil); rangeKeyIter != nil {
if err := updateRangeBounds(rangeKeyIter); err != nil {
return nil, err
}
}
flushingBytes += f.inuseBytes()
}
if opts.FlushSplitBytes > 0 {
c.maxOutputFileSize = uint64(opts.Level(0).TargetFileSize)
c.maxOverlapBytes = maxGrandparentOverlapBytes(opts, 0)
c.grandparents = c.version.Overlaps(baseLevel, c.userKeyBounds())
adjustGrandparentOverlapBytesForFlush(c, flushingBytes)
}
// We don't elide tombstones for flushes.
c.delElision, c.rangeKeyElision = compact.NoTombstoneElision(), compact.NoTombstoneElision()
return c, nil
}
func (c *compaction) hasExtraLevelData() bool {
if len(c.extraLevels) == 0 {
// not a multi level compaction
return false
} else if c.extraLevels[0].files.Empty() {
// a multi level compaction without data in the intermediate input level;
// e.g. for a multi level compaction with levels 4,5, and 6, this could
// occur if there is no files to compact in 5, or in 5 and 6 (i.e. a move).
return false
}
return true
}
// errorOnUserKeyOverlap returns an error if the last two written sstables in
// this compaction have revisions of the same user key present in both sstables,
// when it shouldn't (eg. when splitting flushes).
func (c *compaction) errorOnUserKeyOverlap(ve *versionEdit) error {
if n := len(ve.NewFiles); n > 1 {
meta := ve.NewFiles[n-1].Meta
prevMeta := ve.NewFiles[n-2].Meta
if !prevMeta.Largest.IsExclusiveSentinel() &&
c.cmp(prevMeta.Largest.UserKey, meta.Smallest.UserKey) >= 0 {
return errors.Errorf("pebble: compaction split user key across two sstables: %s in %s and %s",
prevMeta.Largest.Pretty(c.formatKey),
prevMeta.FileNum,
meta.FileNum)
}
}
return nil
}
// allowZeroSeqNum returns true if seqnum's can be zeroed if there are no
// snapshots requiring them to be kept. It performs this determination by
// looking at the TombstoneElision values which are set up based on sstables
// which overlap the bounds of the compaction at a lower level in the LSM.
func (c *compaction) allowZeroSeqNum() bool {
// TODO(peter): we disable zeroing of seqnums during flushing to match
// RocksDB behavior and to avoid generating overlapping sstables during
// DB.replayWAL. When replaying WAL files at startup, we flush after each
// WAL is replayed building up a single version edit that is
// applied. Because we don't apply the version edit after each flush, this
// code doesn't know that L0 contains files and zeroing of seqnums should
// be disabled. That is fixable, but it seems safer to just match the
// RocksDB behavior for now.
return len(c.flushing) == 0 && c.delElision.ElidesEverything() && c.rangeKeyElision.ElidesEverything()
}
// newInputIters returns an iterator over all the input tables in a compaction.
func (c *compaction) newInputIters(
newIters tableNewIters, newRangeKeyIter keyspanimpl.TableNewSpanIter,
) (
pointIter internalIterator,
rangeDelIter, rangeKeyIter keyspan.FragmentIterator,
retErr error,
) {
// Validate the ordering of compaction input files for defense in depth.
if len(c.flushing) == 0 {
if c.startLevel.level >= 0 {
err := manifest.CheckOrdering(c.cmp, c.formatKey,
manifest.Level(c.startLevel.level), c.startLevel.files.Iter())
if err != nil {
return nil, nil, nil, err
}
}
err := manifest.CheckOrdering(c.cmp, c.formatKey,
manifest.Level(c.outputLevel.level), c.outputLevel.files.Iter())
if err != nil {
return nil, nil, nil, err
}
if c.startLevel.level == 0 {
if c.startLevel.l0SublevelInfo == nil {
panic("l0SublevelInfo not created for compaction out of L0")
}
for _, info := range c.startLevel.l0SublevelInfo {
err := manifest.CheckOrdering(c.cmp, c.formatKey,
info.sublevel, info.Iter())
if err != nil {
return nil, nil, nil, err
}
}
}
if len(c.extraLevels) > 0 {
if len(c.extraLevels) > 1 {
panic("n>2 multi level compaction not implemented yet")
}
interLevel := c.extraLevels[0]
err := manifest.CheckOrdering(c.cmp, c.formatKey,
manifest.Level(interLevel.level), interLevel.files.Iter())
if err != nil {
return nil, nil, nil, err
}
}
}
// There are three classes of keys that a compaction needs to process: point
// keys, range deletion tombstones and range keys. Collect all iterators for
// all these classes of keys from all the levels. We'll aggregate them
// together farther below.
//
// numInputLevels is an approximation of the number of iterator levels. Due
// to idiosyncrasies in iterator construction, we may (rarely) exceed this
// initial capacity.
numInputLevels := max(len(c.flushing), len(c.inputs))
iters := make([]internalIterator, 0, numInputLevels)
rangeDelIters := make([]keyspan.FragmentIterator, 0, numInputLevels)
rangeKeyIters := make([]keyspan.FragmentIterator, 0, numInputLevels)
// If construction of the iterator inputs fails, ensure that we close all
// the consitutent iterators.
defer func() {
if retErr != nil {
for _, iter := range iters {
if iter != nil {
iter.Close()
}
}
for _, rangeDelIter := range rangeDelIters {
rangeDelIter.Close()
}
}
}()
iterOpts := IterOptions{
Category: categoryCompaction,
logger: c.logger,
}
// Populate iters, rangeDelIters and rangeKeyIters with the appropriate
// constituent iterators. This depends on whether this is a flush or a
// compaction.
if len(c.flushing) != 0 {
// If flushing, we need to build the input iterators over the memtables
// stored in c.flushing.
for i := range c.flushing {
f := c.flushing[i]
iters = append(iters, f.newFlushIter(nil))
rangeDelIter := f.newRangeDelIter(nil)
if rangeDelIter != nil {
rangeDelIters = append(rangeDelIters, rangeDelIter)
}
if rangeKeyIter := f.newRangeKeyIter(nil); rangeKeyIter != nil {
rangeKeyIters = append(rangeKeyIters, rangeKeyIter)
}
}
} else {
addItersForLevel := func(level *compactionLevel, l manifest.Layer) error {
// Add a *levelIter for point iterators. Because we don't call
// initRangeDel, the levelIter will close and forget the range
// deletion iterator when it steps on to a new file. Surfacing range
// deletions to compactions are handled below.
iters = append(iters, newLevelIter(context.Background(),
iterOpts, c.comparer, newIters, level.files.Iter(), l, internalIterOpts{
compaction: true,
bufferPool: &c.bufferPool,
}))
// TODO(jackson): Use keyspanimpl.LevelIter to avoid loading all the range
// deletions into memory upfront. (See #2015, which reverted this.) There
// will be no user keys that are split between sstables within a level in
// Cockroach 23.1, which unblocks this optimization.
// Add the range deletion iterator for each file as an independent level
// in mergingIter, as opposed to making a levelIter out of those. This
// is safer as levelIter expects all keys coming from underlying
// iterators to be in order. Due to compaction / tombstone writing
// logic in finishOutput(), it is possible for range tombstones to not
// be strictly ordered across all files in one level.
//
// Consider this example from the metamorphic tests (also repeated in
// finishOutput()), consisting of three L3 files with their bounds
// specified in square brackets next to the file name:
//
// ./000240.sst [tmgc#391,MERGE-tmgc#391,MERGE]
// tmgc#391,MERGE [786e627a]
// tmgc-udkatvs#331,RANGEDEL
//
// ./000241.sst [tmgc#384,MERGE-tmgc#384,MERGE]
// tmgc#384,MERGE [666c7070]
// tmgc-tvsalezade#383,RANGEDEL
// tmgc-tvsalezade#331,RANGEDEL
//
// ./000242.sst [tmgc#383,RANGEDEL-tvsalezade#72057594037927935,RANGEDEL]
// tmgc-tvsalezade#383,RANGEDEL
// tmgc#375,SET [72646c78766965616c72776865676e79]
// tmgc-tvsalezade#356,RANGEDEL
//
// Here, the range tombstone in 000240.sst falls "after" one in
// 000241.sst, despite 000240.sst being ordered "before" 000241.sst for
// levelIter's purposes. While each file is still consistent before its
// bounds, it's safer to have all rangedel iterators be visible to
// mergingIter.
iter := level.files.Iter()
for f := iter.First(); f != nil; f = iter.Next() {
rangeDelIter, err := c.newRangeDelIter(newIters, iter.Take(), iterOpts, l)
if err != nil {
// The error will already be annotated with the BackingFileNum, so
// we annotate it with the FileNum.
return errors.Wrapf(err, "pebble: could not open table %s", errors.Safe(f.FileNum))
}
if rangeDelIter == nil {
continue
}
rangeDelIters = append(rangeDelIters, rangeDelIter)
c.closers = append(c.closers, rangeDelIter)
}
// Check if this level has any range keys.
hasRangeKeys := false
for f := iter.First(); f != nil; f = iter.Next() {
if f.HasRangeKeys {
hasRangeKeys = true
break
}
}
if hasRangeKeys {
newRangeKeyIterWrapper := func(ctx context.Context, file *manifest.FileMetadata, iterOptions keyspan.SpanIterOptions) (keyspan.FragmentIterator, error) {
rangeKeyIter, err := newRangeKeyIter(ctx, file, iterOptions)
if err != nil {
return nil, err
} else if rangeKeyIter == nil {
return emptyKeyspanIter, nil
}
// Ensure that the range key iter is not closed until the compaction is
// finished. This is necessary because range key processing
// requires the range keys to be held in memory for up to the
// lifetime of the compaction.
noCloseIter := &noCloseIter{rangeKeyIter}
c.closers = append(c.closers, noCloseIter)
// We do not need to truncate range keys to sstable boundaries, or
// only read within the file's atomic compaction units, unlike with
// range tombstones. This is because range keys were added after we
// stopped splitting user keys across sstables, so all the range keys
// in this sstable must wholly lie within the file's bounds.
return noCloseIter, err
}
li := keyspanimpl.NewLevelIter(
context.Background(), keyspan.SpanIterOptions{}, c.cmp,
newRangeKeyIterWrapper, level.files.Iter(), l, manifest.KeyTypeRange,
)
rangeKeyIters = append(rangeKeyIters, li)
}
return nil
}
for i := range c.inputs {
// If the level is annotated with l0SublevelInfo, expand it into one
// level per sublevel.
// TODO(jackson): Perform this expansion even earlier when we pick the
// compaction?
if len(c.inputs[i].l0SublevelInfo) > 0 {
for _, info := range c.startLevel.l0SublevelInfo {
sublevelCompactionLevel := &compactionLevel{0, info.LevelSlice, nil}
if err := addItersForLevel(sublevelCompactionLevel, info.sublevel); err != nil {
return nil, nil, nil, err
}
}
continue
}
if err := addItersForLevel(&c.inputs[i], manifest.Level(c.inputs[i].level)); err != nil {
return nil, nil, nil, err
}
}
}
// If there's only one constituent point iterator, we can avoid the overhead
// of a *mergingIter. This is possible, for example, when performing a flush
// of a single memtable. Otherwise, combine all the iterators into a merging
// iter.
pointIter = iters[0]
if len(iters) > 1 {
pointIter = newMergingIter(c.logger, &c.stats, c.cmp, nil, iters...)
}
// In normal operation, levelIter iterates over the point operations in a
// level, and initializes a rangeDelIter pointer for the range deletions in
// each table. During compaction, we want to iterate over the merged view of
// point operations and range deletions. In order to do this we create one
// levelIter per level to iterate over the point operations, and collect up
// all the range deletion files.
//
// The range deletion levels are combined with a keyspanimpl.MergingIter. The
// resulting merged rangedel iterator is then included using an
// InterleavingIter.
// TODO(jackson): Consider using a defragmenting iterator to stitch together
// logical range deletions that were fragmented due to previous file
// boundaries.
if len(rangeDelIters) > 0 {
mi := &keyspanimpl.MergingIter{}
mi.Init(c.comparer, keyspan.NoopTransform, new(keyspanimpl.MergingBuffers), rangeDelIters...)
rangeDelIter = mi
}
// If there are range key iterators, we need to combine them using
// keyspanimpl.MergingIter, and then interleave them among the points.
if len(rangeKeyIters) > 0 {
mi := &keyspanimpl.MergingIter{}
mi.Init(c.comparer, keyspan.NoopTransform, new(keyspanimpl.MergingBuffers), rangeKeyIters...)
// TODO(radu): why do we have a defragmenter here but not above?
di := &keyspan.DefragmentingIter{}
di.Init(c.comparer, mi, keyspan.DefragmentInternal, keyspan.StaticDefragmentReducer, new(keyspan.DefragmentingBuffers))
rangeKeyIter = di
}
return pointIter, rangeDelIter, rangeKeyIter, nil
}
func (c *compaction) newRangeDelIter(
newIters tableNewIters, f manifest.LevelFile, opts IterOptions, l manifest.Layer,
) (*noCloseIter, error) {
opts.layer = l
iterSet, err := newIters(context.Background(), f.FileMetadata, &opts,
internalIterOpts{
compaction: true,
bufferPool: &c.bufferPool,
}, iterRangeDeletions)
if err != nil {
return nil, err
} else if iterSet.rangeDeletion == nil {
// The file doesn't contain any range deletions.
return nil, nil
}
// Ensure that rangeDelIter is not closed until the compaction is
// finished. This is necessary because range tombstone processing
// requires the range tombstones to be held in memory for up to the
// lifetime of the compaction.
return &noCloseIter{iterSet.rangeDeletion}, nil
}
func (c *compaction) String() string {
if len(c.flushing) != 0 {
return "flush\n"
}
var buf bytes.Buffer
for level := c.startLevel.level; level <= c.outputLevel.level; level++ {
i := level - c.startLevel.level
fmt.Fprintf(&buf, "%d:", level)
iter := c.inputs[i].files.Iter()
for f := iter.First(); f != nil; f = iter.Next() {
fmt.Fprintf(&buf, " %s:%s-%s", f.FileNum, f.Smallest, f.Largest)
}
fmt.Fprintf(&buf, "\n")
}
return buf.String()
}
type manualCompaction struct {
// Count of the retries either due to too many concurrent compactions, or a
// concurrent compaction to overlapping levels.
retries int
level int
outputLevel int
done chan error
start []byte
end []byte
split bool
}
type readCompaction struct {
level int
// [start, end] key ranges are used for de-duping.
start []byte
end []byte
// The file associated with the compaction.
// If the file no longer belongs in the same
// level, then we skip the compaction.
fileNum base.FileNum
}
func (d *DB) addInProgressCompaction(c *compaction) {
d.mu.compact.inProgress[c] = struct{}{}
var isBase, isIntraL0 bool
for _, cl := range c.inputs {
iter := cl.files.Iter()
for f := iter.First(); f != nil; f = iter.Next() {
if f.IsCompacting() {
d.opts.Logger.Fatalf("L%d->L%d: %s already being compacted", c.startLevel.level, c.outputLevel.level, f.FileNum)
}
f.SetCompactionState(manifest.CompactionStateCompacting)
if c.startLevel != nil && c.outputLevel != nil && c.startLevel.level == 0 {
if c.outputLevel.level == 0 {
f.IsIntraL0Compacting = true
isIntraL0 = true
} else {
isBase = true
}
}