forked from cockroachdb/pebble
-
Notifications
You must be signed in to change notification settings - Fork 0
/
compaction_picker.go
2107 lines (1932 loc) · 77.2 KB
/
compaction_picker.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 2018 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"
"fmt"
"math"
"sort"
"strings"
"github.com/cockroachdb/pebble/internal/base"
"github.com/cockroachdb/pebble/internal/humanize"
"github.com/cockroachdb/pebble/internal/manifest"
)
// The minimum count for an intra-L0 compaction. This matches the RocksDB
// heuristic.
const minIntraL0Count = 4
type compactionEnv struct {
// diskAvailBytes holds a statistic on the number of bytes available on
// disk, as reported by the filesystem. It's used to be more restrictive in
// expanding compactions if available disk space is limited.
//
// The cached value (d.diskAvailBytes) is updated whenever a file is deleted
// and whenever a compaction or flush completes. Since file removal is the
// primary means of reclaiming space, there is a rough bound on the
// statistic's staleness when available bytes is growing. Compactions and
// flushes are longer, slower operations and provide a much looser bound
// when available bytes is decreasing.
diskAvailBytes uint64
earliestUnflushedSeqNum uint64
earliestSnapshotSeqNum uint64
inProgressCompactions []compactionInfo
readCompactionEnv readCompactionEnv
}
type compactionPicker interface {
getScores([]compactionInfo) [numLevels]float64
getBaseLevel() int
estimatedCompactionDebt(l0ExtraSize uint64) uint64
pickAuto(env compactionEnv) (pc *pickedCompaction)
pickElisionOnlyCompaction(env compactionEnv) (pc *pickedCompaction)
pickRewriteCompaction(env compactionEnv) (pc *pickedCompaction)
pickReadTriggeredCompaction(env compactionEnv) (pc *pickedCompaction)
forceBaseLevel1()
}
// readCompactionEnv is used to hold data required to perform read compactions
type readCompactionEnv struct {
rescheduleReadCompaction *bool
readCompactions *readCompactionQueue
flushing bool
}
// Information about in-progress compactions provided to the compaction picker.
// These are used to constrain the new compactions that will be picked.
type compactionInfo struct {
// versionEditApplied is true if this compaction's version edit has already
// been committed. The compaction may still be in-progress deleting newly
// obsolete files.
versionEditApplied bool
inputs []compactionLevel
outputLevel int
smallest InternalKey
largest InternalKey
}
func (info compactionInfo) String() string {
var buf bytes.Buffer
var largest int
for i, in := range info.inputs {
if i > 0 {
fmt.Fprintf(&buf, " -> ")
}
fmt.Fprintf(&buf, "L%d", in.level)
in.files.Each(func(m *fileMetadata) {
fmt.Fprintf(&buf, " %s", m.FileNum)
})
if largest < in.level {
largest = in.level
}
}
if largest != info.outputLevel || len(info.inputs) == 1 {
fmt.Fprintf(&buf, " -> L%d", info.outputLevel)
}
return buf.String()
}
type sortCompactionLevelsByPriority []candidateLevelInfo
func (s sortCompactionLevelsByPriority) Len() int {
return len(s)
}
// A level should be picked for compaction if the compensatedScoreRatio is >= the
// compactionScoreThreshold.
const compactionScoreThreshold = 1
// Less should return true if s[i] must be placed earlier than s[j] in the final
// sorted list. The candidateLevelInfo for the level placed earlier is more likely
// to be picked for a compaction.
func (s sortCompactionLevelsByPriority) Less(i, j int) bool {
iShouldCompact := s[i].compensatedScoreRatio >= compactionScoreThreshold
jShouldCompact := s[j].compensatedScoreRatio >= compactionScoreThreshold
// Ordering is defined as decreasing on (shouldCompact, uncompensatedScoreRatio)
// where shouldCompact is 1 for true and 0 for false.
if iShouldCompact && !jShouldCompact {
return true
}
if !iShouldCompact && jShouldCompact {
return false
}
if s[i].uncompensatedScoreRatio != s[j].uncompensatedScoreRatio {
return s[i].uncompensatedScoreRatio > s[j].uncompensatedScoreRatio
}
return s[i].level < s[j].level
}
func (s sortCompactionLevelsByPriority) Swap(i, j int) {
s[i], s[j] = s[j], s[i]
}
// sublevelInfo is used to tag a LevelSlice for an L0 sublevel with the
// sublevel.
type sublevelInfo struct {
manifest.LevelSlice
sublevel manifest.Level
}
func (cl sublevelInfo) Clone() sublevelInfo {
return sublevelInfo{
sublevel: cl.sublevel,
LevelSlice: cl.LevelSlice.Reslice(func(start, end *manifest.LevelIterator) {}),
}
}
func (cl sublevelInfo) String() string {
return fmt.Sprintf(`Sublevel %s; Levels %s`, cl.sublevel, cl.LevelSlice)
}
// generateSublevelInfo will generate the level slices for each of the sublevels
// from the level slice for all of L0.
func generateSublevelInfo(cmp base.Compare, levelFiles manifest.LevelSlice) []sublevelInfo {
sublevelMap := make(map[uint64][]*fileMetadata)
it := levelFiles.Iter()
for f := it.First(); f != nil; f = it.Next() {
sublevelMap[uint64(f.SubLevel)] = append(sublevelMap[uint64(f.SubLevel)], f)
}
var sublevels []int
for level := range sublevelMap {
sublevels = append(sublevels, int(level))
}
sort.Ints(sublevels)
var levelSlices []sublevelInfo
for _, sublevel := range sublevels {
metas := sublevelMap[uint64(sublevel)]
levelSlices = append(
levelSlices,
sublevelInfo{
manifest.NewLevelSliceKeySorted(cmp, metas),
manifest.L0Sublevel(sublevel),
},
)
}
return levelSlices
}
// compactionPickerMetrics holds metrics related to the compaction picking process
type compactionPickerMetrics struct {
// scores contains the compensatedScoreRatio from the candidateLevelInfo.
scores []float64
singleLevelOverlappingRatio float64
multiLevelOverlappingRatio float64
}
// pickedCompaction contains information about a compaction that has already
// been chosen, and is being constructed. Compaction construction info lives in
// this struct, and is copied over into the compaction struct when that's
// created.
type pickedCompaction struct {
cmp Compare
// score of the chosen compaction. This is the same as the
// compensatedScoreRatio in the candidateLevelInfo.
score float64
// kind indicates the kind of compaction.
kind compactionKind
// 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
outputLevel *compactionLevel
// extraLevels contain additional levels in between the input and output
// levels that get compacted in multi level compactions
extraLevels []*compactionLevel
inputs []compactionLevel
// LBase at the time of compaction picking.
baseLevel int
// L0-specific compaction info. Set to a non-nil value for all compactions
// where startLevel == 0 that were generated by L0Sublevels.
lcf *manifest.L0CompactionFiles
// 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
// maxReadCompactionBytes is the maximum bytes a read compaction is allowed to
// overlap in its output level with. If the overlap is greater than
// maxReadCompaction bytes, then we don't proceed with the compaction.
maxReadCompactionBytes uint64
// The boundaries of the input data.
smallest InternalKey
largest InternalKey
version *version
pickerMetrics compactionPickerMetrics
}
func defaultOutputLevel(startLevel, baseLevel int) int {
outputLevel := startLevel + 1
if startLevel == 0 {
outputLevel = baseLevel
}
if outputLevel >= numLevels-1 {
outputLevel = numLevels - 1
}
return outputLevel
}
func newPickedCompaction(
opts *Options, cur *version, startLevel, outputLevel, baseLevel int,
) *pickedCompaction {
if startLevel > 0 && startLevel < baseLevel {
panic(fmt.Sprintf("invalid compaction: start level %d should not be empty (base level %d)",
startLevel, baseLevel))
}
adjustedLevel := adjustedOutputLevel(outputLevel, baseLevel)
pc := &pickedCompaction{
cmp: opts.Comparer.Compare,
version: cur,
baseLevel: baseLevel,
inputs: []compactionLevel{{level: startLevel}, {level: outputLevel}},
maxOutputFileSize: uint64(opts.Level(adjustedLevel).TargetFileSize),
maxOverlapBytes: maxGrandparentOverlapBytes(opts, adjustedLevel),
maxReadCompactionBytes: maxReadCompactionBytes(opts, adjustedLevel),
}
pc.startLevel = &pc.inputs[0]
pc.outputLevel = &pc.inputs[1]
return pc
}
// adjustedOutputLevel is the output level used for the purpose of
// determining the target output file size, overlap bytes, and expanded
// bytes, taking into account the base level.
func adjustedOutputLevel(outputLevel int, baseLevel int) int {
adjustedOutputLevel := outputLevel
if adjustedOutputLevel > 0 {
// Output level is in the range [baseLevel, numLevels]. For the purpose of
// determining the target output file size, overlap bytes, and expanded
// bytes, we want to adjust the range to [1,numLevels].
adjustedOutputLevel = 1 + outputLevel - baseLevel
}
return adjustedOutputLevel
}
func newPickedCompactionFromL0(
lcf *manifest.L0CompactionFiles, opts *Options, vers *version, baseLevel int, isBase bool,
) *pickedCompaction {
outputLevel := baseLevel
if !isBase {
outputLevel = 0 // Intra L0
}
pc := newPickedCompaction(opts, vers, 0, outputLevel, baseLevel)
pc.lcf = lcf
pc.outputLevel.level = outputLevel
// Manually build the compaction as opposed to calling
// pickAutoHelper. This is because L0Sublevels has already added
// any overlapping L0 SSTables that need to be added, and
// because compactions built by L0SSTables do not necessarily
// pick contiguous sequences of files in pc.version.Levels[0].
files := make([]*manifest.FileMetadata, 0, len(lcf.Files))
iter := vers.Levels[0].Iter()
for f := iter.First(); f != nil; f = iter.Next() {
if lcf.FilesIncluded[f.L0Index] {
files = append(files, f)
}
}
pc.startLevel.files = manifest.NewLevelSliceSeqSorted(files)
return pc
}
func (pc *pickedCompaction) String() string {
var builder strings.Builder
builder.WriteString(fmt.Sprintf(`Score=%f, `, pc.score))
builder.WriteString(fmt.Sprintf(`Kind=%s, `, pc.kind))
builder.WriteString(fmt.Sprintf(`AdjustedOutputLevel=%d, `, adjustedOutputLevel(pc.outputLevel.level, pc.baseLevel)))
builder.WriteString(fmt.Sprintf(`maxOutputFileSize=%d, `, pc.maxOutputFileSize))
builder.WriteString(fmt.Sprintf(`maxReadCompactionBytes=%d, `, pc.maxReadCompactionBytes))
builder.WriteString(fmt.Sprintf(`smallest=%s, `, pc.smallest))
builder.WriteString(fmt.Sprintf(`largest=%s, `, pc.largest))
builder.WriteString(fmt.Sprintf(`version=%s, `, pc.version))
builder.WriteString(fmt.Sprintf(`inputs=%s, `, pc.inputs))
builder.WriteString(fmt.Sprintf(`startlevel=%s, `, pc.startLevel))
builder.WriteString(fmt.Sprintf(`outputLevel=%s, `, pc.outputLevel))
builder.WriteString(fmt.Sprintf(`extraLevels=%s, `, pc.extraLevels))
builder.WriteString(fmt.Sprintf(`l0SublevelInfo=%s, `, pc.startLevel.l0SublevelInfo))
builder.WriteString(fmt.Sprintf(`lcf=%s`, pc.lcf))
return builder.String()
}
// Clone creates a deep copy of the pickedCompaction
func (pc *pickedCompaction) clone() *pickedCompaction {
// Quickly copy over fields that do not require special deep copy care, and
// set all fields that will require a deep copy to nil.
newPC := &pickedCompaction{
cmp: pc.cmp,
score: pc.score,
kind: pc.kind,
baseLevel: pc.baseLevel,
maxOutputFileSize: pc.maxOutputFileSize,
maxOverlapBytes: pc.maxOverlapBytes,
maxReadCompactionBytes: pc.maxReadCompactionBytes,
smallest: pc.smallest.Clone(),
largest: pc.largest.Clone(),
// TODO(msbutler): properly clone picker metrics
pickerMetrics: pc.pickerMetrics,
// Both copies see the same manifest, therefore, it's ok for them to se
// share the same pc. version.
version: pc.version,
}
newPC.inputs = make([]compactionLevel, len(pc.inputs))
newPC.extraLevels = make([]*compactionLevel, 0, len(pc.extraLevels))
for i := range pc.inputs {
newPC.inputs[i] = pc.inputs[i].Clone()
if i == 0 {
newPC.startLevel = &newPC.inputs[i]
} else if i == len(pc.inputs)-1 {
newPC.outputLevel = &newPC.inputs[i]
} else {
newPC.extraLevels = append(newPC.extraLevels, &newPC.inputs[i])
}
}
if len(pc.startLevel.l0SublevelInfo) > 0 {
newPC.startLevel.l0SublevelInfo = make([]sublevelInfo, len(pc.startLevel.l0SublevelInfo))
for i := range pc.startLevel.l0SublevelInfo {
newPC.startLevel.l0SublevelInfo[i] = pc.startLevel.l0SublevelInfo[i].Clone()
}
}
if pc.lcf != nil {
newPC.lcf = pc.lcf.Clone()
}
return newPC
}
// maybeExpandedBounds is a helper function for setupInputs which ensures the
// pickedCompaction's smallest and largest internal keys are updated iff
// the candidate keys expand the key span. This avoids a bug for multi-level
// compactions: during the second call to setupInputs, the picked compaction's
// smallest and largest keys should not decrease the key span.
func (pc *pickedCompaction) maybeExpandBounds(smallest InternalKey, largest InternalKey) {
emptyKey := InternalKey{}
if base.InternalCompare(pc.cmp, smallest, emptyKey) == 0 {
if base.InternalCompare(pc.cmp, largest, emptyKey) != 0 {
panic("either both candidate keys are empty or neither are empty")
}
return
}
if base.InternalCompare(pc.cmp, pc.smallest, emptyKey) == 0 {
if base.InternalCompare(pc.cmp, pc.largest, emptyKey) != 0 {
panic("either both pc keys are empty or neither are empty")
}
pc.smallest = smallest
pc.largest = largest
return
}
if base.InternalCompare(pc.cmp, pc.smallest, smallest) >= 0 {
pc.smallest = smallest
}
if base.InternalCompare(pc.cmp, pc.largest, largest) <= 0 {
pc.largest = largest
}
}
// setupInputs returns true if a compaction has been set up. It returns false if
// a concurrent compaction is occurring on the start or output level files.
func (pc *pickedCompaction) setupInputs(
opts *Options, diskAvailBytes uint64, startLevel *compactionLevel,
) bool {
// maxExpandedBytes is the maximum size of an expanded compaction. If
// growing a compaction results in a larger size, the original compaction
// is used instead.
maxExpandedBytes := expandedCompactionByteSizeLimit(
opts, adjustedOutputLevel(pc.outputLevel.level, pc.baseLevel), diskAvailBytes,
)
// Expand the initial inputs to a clean cut.
var isCompacting bool
startLevel.files, isCompacting = expandToAtomicUnit(pc.cmp, startLevel.files, false /* disableIsCompacting */)
if isCompacting {
return false
}
pc.maybeExpandBounds(manifest.KeyRange(pc.cmp, startLevel.files.Iter()))
// Determine the sstables in the output level which overlap with the input
// sstables, and then expand those tables to a clean cut. No need to do
// this for intra-L0 compactions; outputLevel.files is left empty for those.
if startLevel.level != pc.outputLevel.level {
pc.outputLevel.files = pc.version.Overlaps(pc.outputLevel.level, pc.cmp, pc.smallest.UserKey,
pc.largest.UserKey, pc.largest.IsExclusiveSentinel())
pc.outputLevel.files, isCompacting = expandToAtomicUnit(pc.cmp, pc.outputLevel.files,
false /* disableIsCompacting */)
if isCompacting {
return false
}
pc.maybeExpandBounds(manifest.KeyRange(pc.cmp,
startLevel.files.Iter(), pc.outputLevel.files.Iter()))
}
// Grow the sstables in startLevel.level as long as it doesn't affect the number
// of sstables included from pc.outputLevel.level.
if pc.lcf != nil && startLevel.level == 0 && pc.outputLevel.level != 0 {
// Call the L0-specific compaction extension method. Similar logic as
// pc.grow. Additional L0 files are optionally added to the compaction at
// this step. Note that the bounds passed in are not the bounds of the
// compaction, but rather the smallest and largest internal keys that
// the compaction cannot include from L0 without pulling in more Lbase
// files. Consider this example:
//
// L0: c-d e+f g-h
// Lbase: a-b e+f i-j
// a b c d e f g h i j
//
// The e-f files have already been chosen in the compaction. As pulling
// in more LBase files is undesirable, the logic below will pass in
// smallest = b and largest = i to ExtendL0ForBaseCompactionTo, which
// will expand the compaction to include c-d and g-h from L0. The
// bounds passed in are exclusive; the compaction cannot be expanded
// to include files that "touch" it.
smallestBaseKey := base.InvalidInternalKey
largestBaseKey := base.InvalidInternalKey
if pc.outputLevel.files.Empty() {
baseIter := pc.version.Levels[pc.outputLevel.level].Iter()
if sm := baseIter.SeekLT(pc.cmp, pc.smallest.UserKey); sm != nil {
smallestBaseKey = sm.Largest
}
if la := baseIter.SeekGE(pc.cmp, pc.largest.UserKey); la != nil {
largestBaseKey = la.Smallest
}
} else {
// NB: We use Reslice to access the underlying level's files, but
// we discard the returned slice. The pc.outputLevel.files slice
// is not modified.
_ = pc.outputLevel.files.Reslice(func(start, end *manifest.LevelIterator) {
if sm := start.Prev(); sm != nil {
smallestBaseKey = sm.Largest
}
if la := end.Next(); la != nil {
largestBaseKey = la.Smallest
}
})
}
oldLcf := pc.lcf.Clone()
if pc.version.L0Sublevels.ExtendL0ForBaseCompactionTo(smallestBaseKey, largestBaseKey, pc.lcf) {
var newStartLevelFiles []*fileMetadata
iter := pc.version.Levels[0].Iter()
var sizeSum uint64
for j, f := 0, iter.First(); f != nil; j, f = j+1, iter.Next() {
if pc.lcf.FilesIncluded[f.L0Index] {
newStartLevelFiles = append(newStartLevelFiles, f)
sizeSum += f.Size
}
}
if sizeSum+pc.outputLevel.files.SizeSum() < maxExpandedBytes {
startLevel.files = manifest.NewLevelSliceSeqSorted(newStartLevelFiles)
pc.smallest, pc.largest = manifest.KeyRange(pc.cmp,
startLevel.files.Iter(), pc.outputLevel.files.Iter())
} else {
*pc.lcf = *oldLcf
}
}
} else if pc.grow(pc.smallest, pc.largest, maxExpandedBytes, startLevel) {
pc.maybeExpandBounds(manifest.KeyRange(pc.cmp,
startLevel.files.Iter(), pc.outputLevel.files.Iter()))
}
if pc.startLevel.level == 0 {
// We don't change the input files for the compaction beyond this point.
pc.startLevel.l0SublevelInfo = generateSublevelInfo(pc.cmp, pc.startLevel.files)
}
return true
}
// grow grows the number of inputs at c.level without changing the number of
// c.level+1 files in the compaction, and returns whether the inputs grew. sm
// and la are the smallest and largest InternalKeys in all of the inputs.
func (pc *pickedCompaction) grow(
sm, la InternalKey, maxExpandedBytes uint64, startLevel *compactionLevel,
) bool {
if pc.outputLevel.files.Empty() {
return false
}
grow0 := pc.version.Overlaps(startLevel.level, pc.cmp, sm.UserKey,
la.UserKey, la.IsExclusiveSentinel())
grow0, isCompacting := expandToAtomicUnit(pc.cmp, grow0, false /* disableIsCompacting */)
if isCompacting {
return false
}
if grow0.Len() <= startLevel.files.Len() {
return false
}
if grow0.SizeSum()+pc.outputLevel.files.SizeSum() >= maxExpandedBytes {
return false
}
// We need to include the outputLevel iter because without it, in a multiLevel scenario,
// sm1 and la1 could shift the output level keyspace when pc.outputLevel.files is set to grow1.
sm1, la1 := manifest.KeyRange(pc.cmp, grow0.Iter(), pc.outputLevel.files.Iter())
grow1 := pc.version.Overlaps(pc.outputLevel.level, pc.cmp, sm1.UserKey,
la1.UserKey, la1.IsExclusiveSentinel())
grow1, isCompacting = expandToAtomicUnit(pc.cmp, grow1, false /* disableIsCompacting */)
if isCompacting {
return false
}
if grow1.Len() != pc.outputLevel.files.Len() {
return false
}
startLevel.files = grow0
pc.outputLevel.files = grow1
return true
}
func (pc *pickedCompaction) compactionSize() uint64 {
var bytesToCompact uint64
for i := range pc.inputs {
bytesToCompact += pc.inputs[i].files.SizeSum()
}
return bytesToCompact
}
// setupMultiLevelCandidated returns true if it successfully added another level
// to the compaction.
func (pc *pickedCompaction) setupMultiLevelCandidate(opts *Options, diskAvailBytes uint64) bool {
pc.inputs = append(pc.inputs, compactionLevel{level: pc.outputLevel.level + 1})
// Recalibrate startLevel and outputLevel:
// - startLevel and outputLevel pointers may be obsolete after appending to pc.inputs.
// - push outputLevel to extraLevels and move the new level to outputLevel
pc.startLevel = &pc.inputs[0]
pc.extraLevels = []*compactionLevel{&pc.inputs[1]}
pc.outputLevel = &pc.inputs[2]
return pc.setupInputs(opts, diskAvailBytes, pc.extraLevels[len(pc.extraLevels)-1])
}
// expandToAtomicUnit expands the provided level slice within its level both
// forwards and backwards to its "atomic compaction unit" boundaries, if
// necessary.
//
// While picking compaction inputs, this is required to maintain the invariant
// that the versions of keys at level+1 are older than the versions of keys at
// level. Tables are added to the right of the current slice tables such that
// the rightmost table has a "clean cut". A clean cut is either a change in
// user keys, or when the largest key in the left sstable is a range tombstone
// sentinel key (InternalKeyRangeDeleteSentinel).
//
// In addition to maintaining the seqnum invariant, expandToAtomicUnit is used
// to provide clean boundaries for range tombstone truncation during
// compaction. In order to achieve these clean boundaries, expandToAtomicUnit
// needs to find a "clean cut" on the left edge of the compaction as well.
// This is necessary in order for "atomic compaction units" to always be
// compacted as a unit. Failure to do this leads to a subtle bug with
// truncation of range tombstones to atomic compaction unit boundaries.
// Consider the scenario:
//
// L3:
// 12:[a#2,15-b#1,1]
// 13:[b#0,15-d#72057594037927935,15]
//
// These sstables contain a range tombstone [a-d)#2 which spans the two
// sstables. The two sstables need to always be kept together. Compacting
// sstable 13 independently of sstable 12 would result in:
//
// L3:
// 12:[a#2,15-b#1,1]
// L4:
// 14:[b#0,15-d#72057594037927935,15]
//
// This state is still ok, but when sstable 12 is next compacted, its range
// tombstones will be truncated at "b" (the largest key in its atomic
// compaction unit). In the scenario here, that could result in b#1 becoming
// visible when it should be deleted.
//
// isCompacting is returned true for any atomic units that contain files that
// have in-progress compactions, i.e. FileMetadata.Compacting == true. If
// disableIsCompacting is true, isCompacting always returns false. This helps
// avoid spurious races from being detected when this method is used outside
// of compaction picking code.
//
// TODO(jackson): Compactions and flushes no longer split a user key between two
// sstables. We could perform a migration, re-compacting any sstables with split
// user keys, which would allow us to remove atomic compaction unit expansion
// code.
func expandToAtomicUnit(
cmp Compare, inputs manifest.LevelSlice, disableIsCompacting bool,
) (slice manifest.LevelSlice, isCompacting bool) {
// NB: Inputs for L0 can't be expanded and *version.Overlaps guarantees
// that we get a 'clean cut.' For L0, Overlaps will return a slice without
// access to the rest of the L0 files, so it's OK to try to reslice.
if inputs.Empty() {
// Nothing to expand.
return inputs, false
}
// TODO(jackson): Update to avoid use of LevelIterator.Current(). The
// Reslice interface will require some tweaking, because we currently rely
// on Reslice having already positioned the LevelIterator appropriately.
inputs = inputs.Reslice(func(start, end *manifest.LevelIterator) {
iter := start.Clone()
iter.Prev()
for cur, prev := start.Current(), iter.Current(); prev != nil; cur, prev = start.Prev(), iter.Prev() {
if cur.IsCompacting() {
isCompacting = true
}
if cmp(prev.Largest.UserKey, cur.Smallest.UserKey) < 0 {
break
}
if prev.Largest.IsExclusiveSentinel() {
// The table prev has a largest key indicating that the user key
// prev.largest.UserKey doesn't actually exist in the table.
break
}
// prev.Largest.UserKey == cur.Smallest.UserKey, so we need to
// include prev in the compaction.
}
iter = end.Clone()
iter.Next()
for cur, next := end.Current(), iter.Current(); next != nil; cur, next = end.Next(), iter.Next() {
if cur.IsCompacting() {
isCompacting = true
}
if cmp(cur.Largest.UserKey, next.Smallest.UserKey) < 0 {
break
}
if cur.Largest.IsExclusiveSentinel() {
// The table cur has a largest key indicating that the user key
// cur.largest.UserKey doesn't actually exist in the table.
break
}
// cur.Largest.UserKey == next.Smallest.UserKey, so we need to
// include next in the compaction.
}
})
inputIter := inputs.Iter()
isCompacting = !disableIsCompacting &&
(isCompacting || inputIter.First().IsCompacting() || inputIter.Last().IsCompacting())
return inputs, isCompacting
}
func newCompactionPicker(
v *version, opts *Options, inProgressCompactions []compactionInfo,
) compactionPicker {
p := &compactionPickerByScore{
opts: opts,
vers: v,
}
p.initLevelMaxBytes(inProgressCompactions)
return p
}
// Information about a candidate compaction level that has been identified by
// the compaction picker.
type candidateLevelInfo struct {
// The compensatedScore of the level after adjusting according to the other
// levels' sizes. For L0, the compensatedScoreRatio is equivalent to the
// uncompensatedScoreRatio as we don't account for level size compensation in
// L0.
compensatedScoreRatio float64
// The score of the level after accounting for level size compensation before
// adjusting according to other levels' sizes. For L0, the compensatedScore
// is equivalent to the uncompensatedScore as we don't account for level
// size compensation in L0.
compensatedScore float64
// The score of the level to be compacted, calculated using uncompensated file
// sizes and without any adjustments.
uncompensatedScore float64
// uncompensatedScoreRatio is the uncompensatedScore adjusted according to
// the other levels' sizes.
uncompensatedScoreRatio float64
level int
// The level to compact to.
outputLevel int
// The file in level that will be compacted. Additional files may be
// picked by the compaction, and a pickedCompaction created for the
// compaction.
file manifest.LevelFile
}
func (c *candidateLevelInfo) shouldCompact() bool {
return c.compensatedScoreRatio >= compactionScoreThreshold
}
func fileCompensation(f *fileMetadata) uint64 {
return uint64(f.Stats.PointDeletionsBytesEstimate) + f.Stats.RangeDeletionsBytesEstimate
}
// compensatedSize returns f's file size, inflated according to compaction
// priorities.
func compensatedSize(f *fileMetadata) uint64 {
// Add in the estimate of disk space that may be reclaimed by compacting the
// file's tombstones.
return f.Size + fileCompensation(f)
}
// compensatedSizeAnnotator implements manifest.Annotator, annotating B-Tree
// nodes with the sum of the files' compensated sizes. Its annotation type is
// a *uint64. Compensated sizes may change once a table's stats are loaded
// asynchronously, so its values are marked as cacheable only if a file's
// stats have been loaded.
type compensatedSizeAnnotator struct {
}
var _ manifest.Annotator = compensatedSizeAnnotator{}
func (a compensatedSizeAnnotator) Zero(dst interface{}) interface{} {
if dst == nil {
return new(uint64)
}
v := dst.(*uint64)
*v = 0
return v
}
func (a compensatedSizeAnnotator) Accumulate(
f *fileMetadata, dst interface{},
) (v interface{}, cacheOK bool) {
vptr := dst.(*uint64)
*vptr = *vptr + compensatedSize(f)
return vptr, f.StatsValid()
}
func (a compensatedSizeAnnotator) Merge(src interface{}, dst interface{}) interface{} {
srcV := src.(*uint64)
dstV := dst.(*uint64)
*dstV = *dstV + *srcV
return dstV
}
// totalCompensatedSize computes the compensated size over a file metadata
// iterator. Note that this function is linear in the files available to the
// iterator. Use the compensatedSizeAnnotator if querying the total
// compensated size of a level.
func totalCompensatedSize(iter manifest.LevelIterator) uint64 {
var sz uint64
for f := iter.First(); f != nil; f = iter.Next() {
sz += compensatedSize(f)
}
return sz
}
// compactionPickerByScore holds the state and logic for picking a compaction. A
// compaction picker is associated with a single version. A new compaction
// picker is created and initialized every time a new version is installed.
type compactionPickerByScore struct {
opts *Options
vers *version
// The level to target for L0 compactions. Levels L1 to baseLevel must be
// empty.
baseLevel int
// levelMaxBytes holds the dynamically adjusted max bytes setting for each
// level.
levelMaxBytes [numLevels]int64
}
var _ compactionPicker = &compactionPickerByScore{}
func (p *compactionPickerByScore) getScores(inProgress []compactionInfo) [numLevels]float64 {
var scores [numLevels]float64
for _, info := range p.calculateLevelScores(inProgress) {
scores[info.level] = info.compensatedScoreRatio
}
return scores
}
func (p *compactionPickerByScore) getBaseLevel() int {
if p == nil {
return 1
}
return p.baseLevel
}
// estimatedCompactionDebt estimates the number of bytes which need to be
// compacted before the LSM tree becomes stable.
func (p *compactionPickerByScore) estimatedCompactionDebt(l0ExtraSize uint64) uint64 {
if p == nil {
return 0
}
// We assume that all the bytes in L0 need to be compacted to Lbase. This is
// unlike the RocksDB logic that figures out whether L0 needs compaction.
bytesAddedToNextLevel := l0ExtraSize + p.vers.Levels[0].Size()
lbaseSize := p.vers.Levels[p.baseLevel].Size()
var compactionDebt uint64
if bytesAddedToNextLevel > 0 && lbaseSize > 0 {
// We only incur compaction debt if both L0 and Lbase contain data. If L0
// is empty, no compaction is necessary. If Lbase is empty, a move-based
// compaction from L0 would occur.
compactionDebt += bytesAddedToNextLevel + lbaseSize
}
// loop invariant: At the beginning of the loop, bytesAddedToNextLevel is the
// bytes added to `level` in the loop.
for level := p.baseLevel; level < numLevels-1; level++ {
levelSize := p.vers.Levels[level].Size() + bytesAddedToNextLevel
nextLevelSize := p.vers.Levels[level+1].Size()
if levelSize > uint64(p.levelMaxBytes[level]) {
bytesAddedToNextLevel = levelSize - uint64(p.levelMaxBytes[level])
if nextLevelSize > 0 {
// We only incur compaction debt if the next level contains data. If the
// next level is empty, a move-based compaction would be used.
levelRatio := float64(nextLevelSize) / float64(levelSize)
// The current level contributes bytesAddedToNextLevel to compactions.
// The next level contributes levelRatio * bytesAddedToNextLevel.
compactionDebt += uint64(float64(bytesAddedToNextLevel) * (levelRatio + 1))
}
} else {
// We're not moving any bytes to the next level.
bytesAddedToNextLevel = 0
}
}
return compactionDebt
}
func (p *compactionPickerByScore) initLevelMaxBytes(inProgressCompactions []compactionInfo) {
// The levelMaxBytes calculations here differ from RocksDB in two ways:
//
// 1. The use of dbSize vs maxLevelSize. RocksDB uses the size of the maximum
// level in L1-L6, rather than determining the size of the bottom level
// based on the total amount of data in the dB. The RocksDB calculation is
// problematic if L0 contains a significant fraction of data, or if the
// level sizes are roughly equal and thus there is a significant fraction
// of data outside of the largest level.
//
// 2. Not adjusting the size of Lbase based on L0. RocksDB computes
// baseBytesMax as the maximum of the configured LBaseMaxBytes and the
// size of L0. This is problematic because baseBytesMax is used to compute
// the max size of lower levels. A very large baseBytesMax will result in
// an overly large value for the size of lower levels which will caused
// those levels not to be compacted even when they should be
// compacted. This often results in "inverted" LSM shapes where Ln is
// larger than Ln+1.
// Determine the first non-empty level and the total DB size.
firstNonEmptyLevel := -1
var dbSize uint64
for level := 1; level < numLevels; level++ {
if p.vers.Levels[level].Size() > 0 {
if firstNonEmptyLevel == -1 {
firstNonEmptyLevel = level
}
dbSize += p.vers.Levels[level].Size()
}
}
for _, c := range inProgressCompactions {
if c.outputLevel == 0 || c.outputLevel == -1 {
continue
}
if c.inputs[0].level == 0 && (firstNonEmptyLevel == -1 || c.outputLevel < firstNonEmptyLevel) {
firstNonEmptyLevel = c.outputLevel
}
}
// Initialize the max-bytes setting for each level to "infinity" which will
// disallow compaction for that level. We'll fill in the actual value below
// for levels we want to allow compactions from.
for level := 0; level < numLevels; level++ {
p.levelMaxBytes[level] = math.MaxInt64
}
if dbSize == 0 {
// No levels for L1 and up contain any data. Target L0 compactions for the
// last level or to the level to which there is an ongoing L0 compaction.
p.baseLevel = numLevels - 1
if firstNonEmptyLevel >= 0 {
p.baseLevel = firstNonEmptyLevel
}
return
}
dbSize += p.vers.Levels[0].Size()
bottomLevelSize := dbSize - dbSize/uint64(p.opts.Experimental.LevelMultiplier)
curLevelSize := bottomLevelSize
for level := numLevels - 2; level >= firstNonEmptyLevel; level-- {
curLevelSize = uint64(float64(curLevelSize) / float64(p.opts.Experimental.LevelMultiplier))
}
// Compute base level (where L0 data is compacted to).
baseBytesMax := uint64(p.opts.LBaseMaxBytes)
p.baseLevel = firstNonEmptyLevel
for p.baseLevel > 1 && curLevelSize > baseBytesMax {
p.baseLevel--
curLevelSize = uint64(float64(curLevelSize) / float64(p.opts.Experimental.LevelMultiplier))
}
smoothedLevelMultiplier := 1.0
if p.baseLevel < numLevels-1 {
smoothedLevelMultiplier = math.Pow(
float64(bottomLevelSize)/float64(baseBytesMax),
1.0/float64(numLevels-p.baseLevel-1))
}
levelSize := float64(baseBytesMax)
for level := p.baseLevel; level < numLevels; level++ {
if level > p.baseLevel && levelSize > 0 {
levelSize *= smoothedLevelMultiplier
}
// Round the result since test cases use small target level sizes, which
// can be impacted by floating-point imprecision + integer truncation.
roundedLevelSize := math.Round(levelSize)
if roundedLevelSize > float64(math.MaxInt64) {
p.levelMaxBytes[level] = math.MaxInt64
} else {
p.levelMaxBytes[level] = int64(roundedLevelSize)
}
}
}
type levelSizeAdjust struct {
incomingActualBytes uint64
outgoingActualBytes uint64
outgoingCompensatedBytes uint64
}
func (a levelSizeAdjust) compensated() uint64 {
return a.incomingActualBytes - a.outgoingCompensatedBytes
}
func (a levelSizeAdjust) actual() uint64 {
return a.incomingActualBytes - a.outgoingActualBytes
}
func calculateSizeAdjust(inProgressCompactions []compactionInfo) [numLevels]levelSizeAdjust {
// Compute size adjustments for each level based on the in-progress
// compactions. We sum the file sizes of all files leaving and entering each
// level in in-progress compactions. For outgoing files, we also sum a
// separate sum of 'compensated file sizes', which are inflated according
// to deletion estimates.
//
// When we adjust a level's size according to these values during score
// calculation, we subtract the compensated size of start level inputs to
// account for the fact that score calculation uses compensated sizes.
//
// Since compensated file sizes may be compensated because they reclaim
// space from the output level's files, we only add the real file size to
// the output level.
//
// This is slightly different from RocksDB's behavior, which simply elides
// compacting files from the level size calculation.
var sizeAdjust [numLevels]levelSizeAdjust
for i := range inProgressCompactions {
c := &inProgressCompactions[i]
// If this compaction's version edit has already been applied, there's
// no need to adjust: The LSM we'll examine will already reflect the
// new LSM state.
if c.versionEditApplied {
continue
}
for _, input := range c.inputs {
actualSize := input.files.SizeSum()
compensatedSize := totalCompensatedSize(input.files.Iter())
if input.level != c.outputLevel {
sizeAdjust[input.level].outgoingCompensatedBytes += compensatedSize
sizeAdjust[input.level].outgoingActualBytes += actualSize
if c.outputLevel != -1 {
sizeAdjust[c.outputLevel].incomingActualBytes += actualSize
}
}
}
}