Line data Source code
1 : // Copyright 2018 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 : "bytes"
9 : "cmp"
10 : "fmt"
11 : "iter"
12 : "math"
13 : "slices"
14 : "sort"
15 : "strings"
16 :
17 : "github.com/cockroachdb/errors"
18 : "github.com/cockroachdb/pebble/internal/base"
19 : "github.com/cockroachdb/pebble/internal/humanize"
20 : "github.com/cockroachdb/pebble/internal/invariants"
21 : "github.com/cockroachdb/pebble/internal/manifest"
22 : "github.com/cockroachdb/pebble/internal/problemspans"
23 : )
24 :
25 : // The minimum count for an intra-L0 compaction. This matches the RocksDB
26 : // heuristic.
27 : const minIntraL0Count = 4
28 :
29 : type compactionEnv struct {
30 : // diskAvailBytes holds a statistic on the number of bytes available on
31 : // disk, as reported by the filesystem. It's used to be more restrictive in
32 : // expanding compactions if available disk space is limited.
33 : //
34 : // The cached value (d.diskAvailBytes) is updated whenever a file is deleted
35 : // and whenever a compaction or flush completes. Since file removal is the
36 : // primary means of reclaiming space, there is a rough bound on the
37 : // statistic's staleness when available bytes is growing. Compactions and
38 : // flushes are longer, slower operations and provide a much looser bound
39 : // when available bytes is decreasing.
40 : diskAvailBytes uint64
41 : earliestUnflushedSeqNum base.SeqNum
42 : earliestSnapshotSeqNum base.SeqNum
43 : inProgressCompactions []compactionInfo
44 : readCompactionEnv readCompactionEnv
45 : // problemSpans is checked by the compaction picker to avoid compactions that
46 : // overlap an active "problem span". It can be nil when there are no problem
47 : // spans.
48 : problemSpans *problemspans.ByLevel
49 : }
50 :
51 : type compactionPickerMetrics struct {
52 : levels [numLevels]struct {
53 : score float64
54 : fillFactor float64
55 : compensatedFillFactor float64
56 : }
57 : }
58 :
59 : type compactionPicker interface {
60 : getMetrics([]compactionInfo) compactionPickerMetrics
61 : getBaseLevel() int
62 : estimatedCompactionDebt() uint64
63 : pickHighPrioritySpaceCompaction(env compactionEnv) pickedCompaction
64 : pickAutoScore(env compactionEnv) (pc pickedCompaction)
65 : pickAutoNonScore(env compactionEnv) (pc pickedCompaction)
66 : forceBaseLevel1()
67 : }
68 :
69 : // A pickedCompaction describes a potential compaction that the compaction
70 : // picker has selected, based on its heuristics. When a compaction begins to
71 : // execute, it is converted into a compaction struct by ConstructCompaction.
72 : type pickedCompaction interface {
73 : // ManualID returns the ID of the manual compaction, or 0 if the picked
74 : // compaction is not a result of a manual compaction.
75 : ManualID() uint64
76 : // ConstructCompaction creates a compaction from the picked compaction.
77 : ConstructCompaction(*DB, CompactionGrantHandle) compaction
78 : // WaitingCompaction returns a WaitingCompaction description of this
79 : // compaction for consumption by the compaction scheduler.
80 : WaitingCompaction() WaitingCompaction
81 : }
82 :
83 : // readCompactionEnv is used to hold data required to perform read compactions
84 : type readCompactionEnv struct {
85 : rescheduleReadCompaction *bool
86 : readCompactions *readCompactionQueue
87 : flushing bool
88 : }
89 :
90 : // Information about in-progress compactions provided to the compaction picker.
91 : // These are used to constrain the new compactions that will be picked.
92 : type compactionInfo struct {
93 : // versionEditApplied is true if this compaction's version edit has already
94 : // been committed. The compaction may still be in-progress deleting newly
95 : // obsolete files.
96 : versionEditApplied bool
97 : // kind indicates the kind of compaction.
98 : kind compactionKind
99 : inputs []compactionLevel
100 : outputLevel int
101 : // bounds may be nil if the compaction does not involve sstables
102 : // (specifically, a blob file rewrite).
103 : bounds *base.UserKeyBounds
104 : }
105 :
106 0 : func (info compactionInfo) String() string {
107 0 : var buf bytes.Buffer
108 0 : var largest int
109 0 : for i, in := range info.inputs {
110 0 : if i > 0 {
111 0 : fmt.Fprintf(&buf, " -> ")
112 0 : }
113 0 : fmt.Fprintf(&buf, "L%d", in.level)
114 0 : for f := range in.files.All() {
115 0 : fmt.Fprintf(&buf, " %s", f.TableNum)
116 0 : }
117 0 : if largest < in.level {
118 0 : largest = in.level
119 0 : }
120 : }
121 0 : if largest != info.outputLevel || len(info.inputs) == 1 {
122 0 : fmt.Fprintf(&buf, " -> L%d", info.outputLevel)
123 0 : }
124 0 : return buf.String()
125 : }
126 :
127 : // sublevelInfo is used to tag a LevelSlice for an L0 sublevel with the
128 : // sublevel.
129 : type sublevelInfo struct {
130 : manifest.LevelSlice
131 : sublevel manifest.Layer
132 : }
133 :
134 1 : func (cl sublevelInfo) Clone() sublevelInfo {
135 1 : return sublevelInfo{
136 1 : sublevel: cl.sublevel,
137 1 : LevelSlice: cl.LevelSlice,
138 1 : }
139 1 : }
140 0 : func (cl sublevelInfo) String() string {
141 0 : return fmt.Sprintf(`Sublevel %s; Levels %s`, cl.sublevel, cl.LevelSlice)
142 0 : }
143 :
144 : // generateSublevelInfo will generate the level slices for each of the sublevels
145 : // from the level slice for all of L0.
146 1 : func generateSublevelInfo(cmp base.Compare, levelFiles manifest.LevelSlice) []sublevelInfo {
147 1 : sublevelMap := make(map[uint64][]*manifest.TableMetadata)
148 1 : for f := range levelFiles.All() {
149 1 : sublevelMap[uint64(f.SubLevel)] = append(sublevelMap[uint64(f.SubLevel)], f)
150 1 : }
151 :
152 1 : var sublevels []int
153 1 : for level := range sublevelMap {
154 1 : sublevels = append(sublevels, int(level))
155 1 : }
156 1 : sort.Ints(sublevels)
157 1 :
158 1 : var levelSlices []sublevelInfo
159 1 : for _, sublevel := range sublevels {
160 1 : metas := sublevelMap[uint64(sublevel)]
161 1 : levelSlices = append(
162 1 : levelSlices,
163 1 : sublevelInfo{
164 1 : manifest.NewLevelSliceKeySorted(cmp, metas),
165 1 : manifest.L0Sublevel(sublevel),
166 1 : },
167 1 : )
168 1 : }
169 1 : return levelSlices
170 : }
171 :
172 : // pickedCompactionMetrics holds metrics related to the compaction picking process
173 : type pickedCompactionMetrics struct {
174 : // scores contains candidateLevelInfo.scores.
175 : scores []float64
176 : singleLevelOverlappingRatio float64
177 : multiLevelOverlappingRatio float64
178 : }
179 :
180 : // pickedTableCompaction contains information about a compaction of sstables
181 : // that has already been chosen, and is being constructed. Compaction
182 : // construction info lives in this struct, and is copied over into the
183 : // compaction struct in constructCompaction.
184 : type pickedTableCompaction struct {
185 : // score of the chosen compaction (candidateLevelInfo.score).
186 : score float64
187 : // kind indicates the kind of compaction.
188 : kind compactionKind
189 : // manualID > 0 iff this is a manual compaction. It exists solely for
190 : // internal bookkeeping.
191 : manualID uint64
192 : // startLevel is the level that is being compacted. Inputs from startLevel
193 : // and outputLevel will be merged to produce a set of outputLevel files.
194 : startLevel *compactionLevel
195 : // outputLevel is the level that files are being produced in. outputLevel is
196 : // equal to startLevel+1 except when:
197 : // - if startLevel is 0, the output level equals compactionPicker.baseLevel().
198 : // - in multilevel compaction, the output level is the lowest level involved in
199 : // the compaction
200 : outputLevel *compactionLevel
201 : // inputs contain levels involved in the compaction in ascending order
202 : inputs []compactionLevel
203 : // LBase at the time of compaction picking. Might be uninitialized for
204 : // intra-L0 compactions.
205 : baseLevel int
206 : // L0-specific compaction info. Set to a non-nil value for all compactions
207 : // where startLevel == 0 that were generated by L0Sublevels.
208 : lcf *manifest.L0CompactionFiles
209 :
210 : // The boundaries of the input data.
211 : bounds base.UserKeyBounds
212 : version *manifest.Version
213 : l0Organizer *manifest.L0Organizer
214 : pickerMetrics pickedCompactionMetrics
215 : }
216 :
217 : // Assert that *pickedTableCompaction implements pickedCompaction.
218 : var _ pickedCompaction = (*pickedTableCompaction)(nil)
219 :
220 : // ManualID returns the ID of the manual compaction, or 0 if the picked
221 : // compaction is not a result of a manual compaction.
222 1 : func (pc *pickedTableCompaction) ManualID() uint64 { return pc.manualID }
223 :
224 : // Kind returns the kind of compaction.
225 0 : func (pc *pickedTableCompaction) Kind() compactionKind { return pc.kind }
226 :
227 : // Score returns the score of the level at the time the compaction was picked.
228 0 : func (pc *pickedTableCompaction) Score() float64 { return pc.score }
229 :
230 : // ConstructCompaction creates a compaction struct from the
231 : // pickedTableCompaction.
232 : func (pc *pickedTableCompaction) ConstructCompaction(
233 : d *DB, grantHandle CompactionGrantHandle,
234 1 : ) compaction {
235 1 : return newCompaction(
236 1 : pc,
237 1 : d.opts,
238 1 : d.opts.private.timeNow(),
239 1 : d.ObjProvider(),
240 1 : grantHandle,
241 1 : d.shouldCreateShared(pc.outputLevel.level),
242 1 : d.determineCompactionValueSeparation)
243 1 : }
244 :
245 : // WaitingCompaction returns a WaitingCompaction description of this compaction
246 : // for consumption by the compaction scheduler.
247 1 : func (pc *pickedTableCompaction) WaitingCompaction() WaitingCompaction {
248 1 : if pc.manualID > 0 {
249 1 : return WaitingCompaction{Priority: manualCompactionPriority, Score: pc.score}
250 1 : }
251 1 : entry, ok := scheduledCompactionMap[pc.kind]
252 1 : if !ok {
253 0 : panic(errors.AssertionFailedf("unexpected compactionKind %s", pc.kind))
254 : }
255 1 : return WaitingCompaction{
256 1 : Optional: entry.optional,
257 1 : Priority: entry.priority,
258 1 : Score: pc.score,
259 1 : }
260 : }
261 :
262 1 : func defaultOutputLevel(startLevel, baseLevel int) int {
263 1 : outputLevel := startLevel + 1
264 1 : if startLevel == 0 {
265 1 : outputLevel = baseLevel
266 1 : }
267 1 : if outputLevel >= numLevels-1 {
268 1 : outputLevel = numLevels - 1
269 1 : }
270 1 : return outputLevel
271 : }
272 :
273 : func newPickedTableCompaction(
274 : opts *Options,
275 : cur *manifest.Version,
276 : l0Organizer *manifest.L0Organizer,
277 : startLevel, outputLevel, baseLevel int,
278 1 : ) *pickedTableCompaction {
279 1 : if baseLevel == 0 {
280 0 : panic("base level cannot be 0")
281 : }
282 1 : if startLevel > 0 && startLevel < baseLevel {
283 0 : panic(fmt.Sprintf("invalid compaction: start level %d should not be empty (base level %d)",
284 0 : startLevel, baseLevel))
285 : }
286 :
287 1 : pc := &pickedTableCompaction{
288 1 : version: cur,
289 1 : l0Organizer: l0Organizer,
290 1 : baseLevel: baseLevel,
291 1 : inputs: []compactionLevel{{level: startLevel}, {level: outputLevel}},
292 1 : }
293 1 : pc.startLevel = &pc.inputs[0]
294 1 : pc.outputLevel = &pc.inputs[1]
295 1 : return pc
296 : }
297 :
298 : // adjustedOutputLevel is the output level used for the purpose of
299 : // determining the target output file size, overlap bytes, and expanded
300 : // bytes, taking into account the base level.
301 0 : func adjustedOutputLevel(outputLevel int, baseLevel int) int {
302 0 : if outputLevel == 0 {
303 0 : return 0
304 0 : }
305 0 : if baseLevel == 0 {
306 0 : panic("base level cannot be 0")
307 : }
308 : // Output level is in the range [baseLevel, numLevels). For the purpose of
309 : // determining the target output file size, overlap bytes, and expanded
310 : // bytes, we want to adjust the range to [1, numLevels).
311 0 : return 1 + outputLevel - baseLevel
312 : }
313 :
314 : func newPickedCompactionFromL0(
315 : lcf *manifest.L0CompactionFiles,
316 : opts *Options,
317 : vers *manifest.Version,
318 : l0Organizer *manifest.L0Organizer,
319 : baseLevel int,
320 : isBase bool,
321 1 : ) *pickedTableCompaction {
322 1 : outputLevel := baseLevel
323 1 : if !isBase {
324 1 : outputLevel = 0 // Intra L0
325 1 : }
326 :
327 1 : pc := newPickedTableCompaction(opts, vers, l0Organizer, 0, outputLevel, baseLevel)
328 1 : pc.lcf = lcf
329 1 :
330 1 : // Manually build the compaction as opposed to calling
331 1 : // pickAutoHelper. This is because L0Sublevels has already added
332 1 : // any overlapping L0 SSTables that need to be added, and
333 1 : // because compactions built by L0SSTables do not necessarily
334 1 : // pick contiguous sequences of files in pc.version.Levels[0].
335 1 : pc.startLevel.files = manifest.NewLevelSliceSeqSorted(lcf.Files)
336 1 : return pc
337 : }
338 :
339 0 : func (pc *pickedTableCompaction) String() string {
340 0 : var builder strings.Builder
341 0 : builder.WriteString(fmt.Sprintf(`Score=%f, `, pc.score))
342 0 : builder.WriteString(fmt.Sprintf(`Kind=%s, `, pc.kind))
343 0 : builder.WriteString(fmt.Sprintf(`AdjustedOutputLevel=%d, `, adjustedOutputLevel(pc.outputLevel.level, pc.baseLevel)))
344 0 : builder.WriteString(fmt.Sprintf(`bounds=%s, `, pc.bounds))
345 0 : builder.WriteString(fmt.Sprintf(`version=%s, `, pc.version))
346 0 : builder.WriteString(fmt.Sprintf(`inputs=%s, `, pc.inputs))
347 0 : builder.WriteString(fmt.Sprintf(`startlevel=%s, `, pc.startLevel))
348 0 : builder.WriteString(fmt.Sprintf(`outputLevel=%s, `, pc.outputLevel))
349 0 : builder.WriteString(fmt.Sprintf(`l0SublevelInfo=%s, `, pc.startLevel.l0SublevelInfo))
350 0 : builder.WriteString(fmt.Sprintf(`lcf=%s`, pc.lcf))
351 0 : return builder.String()
352 0 : }
353 :
354 : // Clone creates a deep copy of the pickedCompaction
355 1 : func (pc *pickedTableCompaction) clone() *pickedTableCompaction {
356 1 :
357 1 : // Quickly copy over fields that do not require special deep copy care, and
358 1 : // set all fields that will require a deep copy to nil.
359 1 : newPC := &pickedTableCompaction{
360 1 : score: pc.score,
361 1 : kind: pc.kind,
362 1 : baseLevel: pc.baseLevel,
363 1 : bounds: pc.bounds.Clone(),
364 1 :
365 1 : // TODO(msbutler): properly clone picker metrics
366 1 : pickerMetrics: pc.pickerMetrics,
367 1 :
368 1 : // Both copies see the same manifest, therefore, it's ok for them to share
369 1 : // the same pc.version and pc.l0Organizer.
370 1 : version: pc.version,
371 1 : l0Organizer: pc.l0Organizer,
372 1 : }
373 1 :
374 1 : newPC.inputs = make([]compactionLevel, len(pc.inputs))
375 1 : for i := range pc.inputs {
376 1 : newPC.inputs[i] = pc.inputs[i].Clone()
377 1 : if i == 0 {
378 1 : newPC.startLevel = &newPC.inputs[i]
379 1 : } else if i == len(pc.inputs)-1 {
380 1 : newPC.outputLevel = &newPC.inputs[i]
381 1 : }
382 : }
383 :
384 1 : if len(pc.startLevel.l0SublevelInfo) > 0 {
385 1 : newPC.startLevel.l0SublevelInfo = make([]sublevelInfo, len(pc.startLevel.l0SublevelInfo))
386 1 : for i := range pc.startLevel.l0SublevelInfo {
387 1 : newPC.startLevel.l0SublevelInfo[i] = pc.startLevel.l0SublevelInfo[i].Clone()
388 1 : }
389 : }
390 1 : if pc.lcf != nil {
391 1 : newPC.lcf = pc.lcf.Clone()
392 1 : }
393 1 : return newPC
394 : }
395 :
396 : // setupInputs returns true if a compaction has been set up using the provided inputLevel and
397 : // pc.outputLevel. It returns false if a concurrent compaction is occurring on the start or
398 : // output level files. Note that inputLevel is not necessarily pc.startLevel. In multiLevel
399 : // compactions, inputs are set by calling setupInputs once for each adjacent pair of levels.
400 : // This will preserve level invariants when expanding the compaction. pc.bounds will be updated
401 : // to reflect the key range of the inputs.
402 : func (pc *pickedTableCompaction) setupInputs(
403 : opts *Options,
404 : diskAvailBytes uint64,
405 : inProgressCompactions []compactionInfo,
406 : inputLevel *compactionLevel,
407 : problemSpans *problemspans.ByLevel,
408 1 : ) bool {
409 1 : cmp := opts.Comparer.Compare
410 1 : if !canCompactTables(inputLevel.files, inputLevel.level, problemSpans) {
411 1 : return false
412 1 : }
413 1 : pc.bounds = manifest.ExtendKeyRange(cmp, pc.bounds, inputLevel.files.All())
414 1 :
415 1 : // Setup output files and attempt to grow the inputLevel files with
416 1 : // the expanded key range. No need to do this for intra-L0 compactions;
417 1 : // outputLevel.files is left empty for those.
418 1 : if inputLevel.level != pc.outputLevel.level {
419 1 : // Determine the sstables in the output level which overlap with the compaction
420 1 : // key range.
421 1 : pc.outputLevel.files = pc.version.Overlaps(pc.outputLevel.level, pc.bounds)
422 1 : if !canCompactTables(pc.outputLevel.files, pc.outputLevel.level, problemSpans) {
423 1 : return false
424 1 : }
425 1 : pc.bounds = manifest.ExtendKeyRange(cmp, pc.bounds, pc.outputLevel.files.All())
426 1 :
427 1 : // maxExpandedBytes is the maximum size of an expanded compaction. If
428 1 : // growing a compaction results in a larger size, the original compaction
429 1 : // is used instead.
430 1 : targetFileSize := opts.TargetFileSize(pc.outputLevel.level, pc.baseLevel)
431 1 : maxExpandedBytes := expandedCompactionByteSizeLimit(opts, targetFileSize, diskAvailBytes)
432 1 :
433 1 : // Grow the sstables in inputLevel.level as long as it doesn't affect the number
434 1 : // of sstables included from pc.outputLevel.level.
435 1 : if pc.lcf != nil && inputLevel.level == 0 {
436 1 : pc.maybeGrowL0ForBase(cmp, maxExpandedBytes)
437 1 : } else {
438 1 : pc.maybeGrow(cmp, pc.bounds, maxExpandedBytes, inputLevel, problemSpans)
439 1 : }
440 : }
441 :
442 1 : if inputLevel.level == 0 {
443 1 : // If L0 is involved, it should always be the startLevel of the compaction.
444 1 : pc.startLevel.l0SublevelInfo = generateSublevelInfo(cmp, pc.startLevel.files)
445 1 : }
446 :
447 1 : if outputKeyRangeAlreadyCompacting(cmp, inProgressCompactions, pc) {
448 1 : return false
449 1 : }
450 :
451 1 : return true
452 : }
453 :
454 : // maybeGrow grows the number of inputs at startLevel without changing the
455 : // number of pc.outputLevel files in the compaction, and returns whether the
456 : // inputs grew. sm and la are the smallest and largest InternalKeys in all
457 : // inputs.
458 : func (pc *pickedTableCompaction) maybeGrow(
459 : cmp base.Compare,
460 : bounds base.UserKeyBounds,
461 : maxExpandedBytes uint64,
462 : inputLevel *compactionLevel,
463 : problemSpans *problemspans.ByLevel,
464 1 : ) bool {
465 1 : if pc.outputLevel.files.Empty() {
466 1 : return false
467 1 : }
468 1 : expandedInputLevel := pc.version.Overlaps(inputLevel.level, bounds)
469 1 : if !canCompactTables(expandedInputLevel, inputLevel.level, problemSpans) {
470 1 : return false
471 1 : }
472 1 : if expandedInputLevel.Len() <= inputLevel.files.Len() {
473 1 : return false
474 1 : }
475 1 : if expandedInputLevel.AggregateSizeSum()+pc.outputLevel.files.AggregateSizeSum() >= maxExpandedBytes {
476 1 : return false
477 1 : }
478 : // Check that expanding the input level does not change the number of overlapping files in output level.
479 : // We need to include the outputLevel iter because without it, in a multiLevel scenario,
480 : // expandedInputLevel's key range not fully cover all files currently in pc.outputLevel,
481 : // since pc.outputLevel was created using the entire key range which includes higher levels.
482 1 : expandedOutputLevel := pc.version.Overlaps(pc.outputLevel.level,
483 1 : manifest.KeyRange(cmp, expandedInputLevel.All(), pc.outputLevel.files.All()))
484 1 : if expandedOutputLevel.Len() != pc.outputLevel.files.Len() {
485 1 : return false
486 1 : }
487 1 : if !canCompactTables(expandedOutputLevel, pc.outputLevel.level, problemSpans) {
488 0 : return false
489 0 : }
490 1 : inputLevel.files = expandedInputLevel
491 1 : // inputLevel was expanded, adjust key range if necessary.
492 1 : pc.bounds = manifest.ExtendKeyRange(cmp, pc.bounds, inputLevel.files.All())
493 1 : return true
494 : }
495 :
496 : // Similar logic as pc.maybeGrow. Additional L0 files are optionally added to
497 : // the compaction at this step. Note that the bounds passed in are not the
498 : // bounds of the compaction, but rather the smallest and largest internal keys
499 : // that the compaction cannot include from L0 without pulling in more Lbase
500 : // files. Consider this example:
501 : //
502 : // L0: c-d e+f g-h
503 : // Lbase: a-b e+f i-j
504 : //
505 : // a b c d e f g h i j
506 : //
507 : // The e-f files have already been chosen in the compaction. As pulling
508 : // in more LBase files is undesirable, the logic below will pass in
509 : // smallest = b and largest = i to ExtendL0ForBaseCompactionTo, which
510 : // will expand the compaction to include c-d and g-h from L0. The
511 : // bounds passed in are exclusive; the compaction cannot be expanded
512 : // to include files that "touch" it.
513 1 : func (pc *pickedTableCompaction) maybeGrowL0ForBase(cmp base.Compare, maxExpandedBytes uint64) {
514 1 : if invariants.Enabled {
515 1 : if pc.startLevel.level != 0 {
516 0 : panic(fmt.Sprintf("pc.startLevel.level is %d, expected 0", pc.startLevel.level))
517 : }
518 : }
519 :
520 1 : if pc.outputLevel.files.Empty() {
521 1 : // If there are no overlapping fields in the output level, we do not
522 1 : // attempt to expand the compaction to encourage move compactions.
523 1 : return
524 1 : }
525 :
526 1 : smallestBaseKey := base.InvalidInternalKey
527 1 : largestBaseKey := base.InvalidInternalKey
528 1 : // NB: We use Reslice to access the underlying level's files, but
529 1 : // we discard the returned slice. The pc.outputLevel.files slice
530 1 : // is not modified.
531 1 : _ = pc.outputLevel.files.Reslice(func(start, end *manifest.LevelIterator) {
532 1 : if sm := start.Prev(); sm != nil {
533 1 : smallestBaseKey = sm.Largest()
534 1 : }
535 1 : if la := end.Next(); la != nil {
536 1 : largestBaseKey = la.Smallest()
537 1 : }
538 : })
539 1 : oldLcf := pc.lcf.Clone()
540 1 : if !pc.l0Organizer.ExtendL0ForBaseCompactionTo(smallestBaseKey, largestBaseKey, pc.lcf) {
541 1 : return
542 1 : }
543 :
544 1 : var newStartLevelFiles []*manifest.TableMetadata
545 1 : iter := pc.version.Levels[0].Iter()
546 1 : var sizeSum uint64
547 1 : for j, f := 0, iter.First(); f != nil; j, f = j+1, iter.Next() {
548 1 : if pc.lcf.FilesIncluded[f.L0Index] {
549 1 : newStartLevelFiles = append(newStartLevelFiles, f)
550 1 : sizeSum += f.Size
551 1 : }
552 : }
553 :
554 1 : if sizeSum+pc.outputLevel.files.AggregateSizeSum() >= maxExpandedBytes {
555 1 : *pc.lcf = *oldLcf
556 1 : return
557 1 : }
558 :
559 1 : pc.startLevel.files = manifest.NewLevelSliceSeqSorted(newStartLevelFiles)
560 1 : pc.bounds = manifest.ExtendKeyRange(cmp, pc.bounds,
561 1 : pc.startLevel.files.All(), pc.outputLevel.files.All())
562 : }
563 :
564 : // estimatedInputSize returns an estimate of the size of the compaction's
565 : // inputs, including the estimated physical size of input tables' blob
566 : // references.
567 1 : func (pc *pickedTableCompaction) estimatedInputSize() uint64 {
568 1 : var bytesToCompact uint64
569 1 : for i := range pc.inputs {
570 1 : bytesToCompact += pc.inputs[i].files.AggregateSizeSum()
571 1 : }
572 1 : return bytesToCompact
573 : }
574 :
575 : // setupMultiLevelCandidate returns true if it successfully added another level
576 : // to the compaction.
577 : // Note that adding a new level will never change the startLevel inputs, but we
578 : // will attempt to expand the inputs of the intermediate level to the output key range,
579 : // if size constraints allow it.
580 : // For example, consider the following LSM structure, with the initial compaction
581 : // from L1->L2:
582 : // startLevel: L1 [a-b]
583 : // outputLevel: L2 [a-c]
584 : // L1: |a-b | d--e
585 : // L2: |a---c| d----f
586 : // L3: a---------e
587 : //
588 : // When adding L3, we'll expand L2 to include d-f via a call to setupInputs with
589 : // startLevel=L2. L1 will not be expanded.
590 : // startLevel: L1 [a-b]
591 : // intermediateLevel: L2 [a-c, d-f]
592 : // outputLevel: L3 [a-e]
593 : // L1: |a-b | d--e
594 : // L2: |a---c d----f|
595 : // L3: |a---------e |
596 1 : func (pc *pickedTableCompaction) setupMultiLevelCandidate(opts *Options, env compactionEnv) bool {
597 1 : pc.inputs = append(pc.inputs, compactionLevel{level: pc.outputLevel.level + 1})
598 1 :
599 1 : // Recalibrate startLevel and outputLevel:
600 1 : // - startLevel and outputLevel pointers may be obsolete after appending to pc.inputs.
601 1 : // - push outputLevel to extraLevels and move the new level to outputLevel
602 1 : pc.startLevel = &pc.inputs[0]
603 1 : pc.outputLevel = &pc.inputs[2]
604 1 : return pc.setupInputs(opts, env.diskAvailBytes, env.inProgressCompactions, &pc.inputs[1], nil /* TODO(radu) */)
605 1 : }
606 :
607 : // canCompactTables returns true if the tables in the level slice are not
608 : // compacting already and don't intersect any problem spans.
609 : func canCompactTables(
610 : inputs manifest.LevelSlice, level int, problemSpans *problemspans.ByLevel,
611 1 : ) bool {
612 1 : for f := range inputs.All() {
613 1 : if f.IsCompacting() {
614 1 : return false
615 1 : }
616 1 : if problemSpans != nil && problemSpans.Overlaps(level, f.UserKeyBounds()) {
617 0 : return false
618 0 : }
619 : }
620 1 : return true
621 : }
622 :
623 : // newCompactionPickerByScore creates a compactionPickerByScore associated with
624 : // the newest version. The picker is used under logLock (until a new version is
625 : // installed).
626 : func newCompactionPickerByScore(
627 : v *manifest.Version,
628 : lvs *latestVersionState,
629 : opts *Options,
630 : inProgressCompactions []compactionInfo,
631 1 : ) *compactionPickerByScore {
632 1 : p := &compactionPickerByScore{
633 1 : opts: opts,
634 1 : vers: v,
635 1 : latestVersionState: lvs,
636 1 : }
637 1 : p.initLevelMaxBytes(inProgressCompactions)
638 1 : return p
639 1 : }
640 :
641 : // Information about a candidate compaction level that has been identified by
642 : // the compaction picker.
643 : type candidateLevelInfo struct {
644 : // The fill factor of the level, calculated using uncompensated file sizes and
645 : // without any adjustments. A factor > 1 means that the level has more data
646 : // than the ideal size for that level.
647 : //
648 : // For L0, the fill factor is calculated based on the number of sublevels
649 : // (see calculateL0FillFactor).
650 : //
651 : // For L1+, the fill factor is the ratio between the total uncompensated file
652 : // size and the ideal size of the level (based on the total size of the DB).
653 : fillFactor float64
654 :
655 : // The score of the level, used to rank levels.
656 : //
657 : // If the level doesn't require compaction, the score is 0. Otherwise:
658 : // - for L6 the score is equal to the fillFactor;
659 : // - for L0-L5:
660 : // - if the fillFactor is < 1: the score is equal to the fillFactor;
661 : // - if the fillFactor is >= 1: the score is the ratio between the
662 : // fillFactor and the next level's fillFactor.
663 : score float64
664 :
665 : // The fill factor of the level after accounting for level size compensation.
666 : //
667 : // For L0, the compensatedFillFactor is equal to the fillFactor as we don't
668 : // account for level size compensation in L0.
669 : //
670 : // For l1+, the compensatedFillFactor takes into account the estimated
671 : // savings in the lower levels because of deletions.
672 : //
673 : // The compensated fill factor is used to determine if the level should be
674 : // compacted (see calculateLevelScores).
675 : compensatedFillFactor float64
676 :
677 : level int
678 : // The level to compact to.
679 : outputLevel int
680 : // The file in level that will be compacted. Additional files may be
681 : // picked by the compaction, and a pickedCompaction created for the
682 : // compaction.
683 : file manifest.LevelFile
684 : }
685 :
686 1 : func (c *candidateLevelInfo) shouldCompact() bool {
687 1 : return c.score > 0
688 1 : }
689 :
690 1 : func tableTombstoneCompensation(t *manifest.TableMetadata) uint64 {
691 1 : if stats, ok := t.Stats(); ok {
692 1 : return stats.PointDeletionsBytesEstimate + stats.RangeDeletionsBytesEstimate
693 1 : }
694 1 : return 0
695 : }
696 :
697 : // tableCompensatedSize returns t's size, including an estimate of the physical
698 : // size of its external references, and inflated according to compaction
699 : // priorities.
700 1 : func tableCompensatedSize(t *manifest.TableMetadata) uint64 {
701 1 : // Add in the estimate of disk space that may be reclaimed by compacting the
702 1 : // table's tombstones.
703 1 : return t.Size + t.EstimatedReferenceSize() + tableTombstoneCompensation(t)
704 1 : }
705 :
706 : // totalCompensatedSize computes the compensated size over a table metadata
707 : // iterator. Note that this function is linear in the files available to the
708 : // iterator. Use the compensatedSizeAnnotator if querying the total
709 : // compensated size of a level.
710 1 : func totalCompensatedSize(iter iter.Seq[*manifest.TableMetadata]) uint64 {
711 1 : var sz uint64
712 1 : for f := range iter {
713 1 : sz += tableCompensatedSize(f)
714 1 : }
715 1 : return sz
716 : }
717 :
718 : // compactionPickerByScore holds the state and logic for picking a compaction. A
719 : // compaction picker is associated with a single version. A new compaction
720 : // picker is created and initialized every time a new version is installed.
721 : type compactionPickerByScore struct {
722 : opts *Options
723 : vers *manifest.Version
724 : // Unlike vers, which is immutable and the latest version when this picker
725 : // is created, latestVersionState represents the mutable state of the latest
726 : // version. This means that at some point in the future a
727 : // compactionPickerByScore created in the past will have mutually
728 : // inconsistent state in vers and latestVersionState. This is not a problem
729 : // since (a) a new picker is created in UpdateVersionLocked when a new
730 : // version is installed, and (b) only the latest picker is used for picking
731 : // compactions. This is ensured by holding versionSet.logLock for both (a)
732 : // and (b).
733 : latestVersionState *latestVersionState
734 : // The level to target for L0 compactions. Levels L1 to baseLevel must be
735 : // empty.
736 : baseLevel int
737 : // levelMaxBytes holds the dynamically adjusted max bytes setting for each
738 : // level.
739 : levelMaxBytes [numLevels]int64
740 : dbSizeBytes uint64
741 : }
742 :
743 : var _ compactionPicker = &compactionPickerByScore{}
744 :
745 0 : func (p *compactionPickerByScore) getMetrics(inProgress []compactionInfo) compactionPickerMetrics {
746 0 : var m compactionPickerMetrics
747 0 : for _, info := range p.calculateLevelScores(inProgress) {
748 0 : m.levels[info.level].score = info.score
749 0 : m.levels[info.level].fillFactor = info.fillFactor
750 0 : m.levels[info.level].compensatedFillFactor = info.compensatedFillFactor
751 0 : }
752 0 : return m
753 : }
754 :
755 1 : func (p *compactionPickerByScore) getBaseLevel() int {
756 1 : if p == nil {
757 0 : return 1
758 0 : }
759 1 : return p.baseLevel
760 : }
761 :
762 : // estimatedCompactionDebt estimates the number of bytes which need to be
763 : // compacted before the LSM tree becomes stable.
764 1 : func (p *compactionPickerByScore) estimatedCompactionDebt() uint64 {
765 1 : if p == nil {
766 0 : return 0
767 0 : }
768 :
769 : // We assume that all the bytes in L0 need to be compacted to Lbase. This is
770 : // unlike the RocksDB logic that figures out whether L0 needs compaction.
771 1 : bytesAddedToNextLevel := p.vers.Levels[0].AggregateSize()
772 1 : lbaseSize := p.vers.Levels[p.baseLevel].AggregateSize()
773 1 :
774 1 : var compactionDebt uint64
775 1 : if bytesAddedToNextLevel > 0 && lbaseSize > 0 {
776 1 : // We only incur compaction debt if both L0 and Lbase contain data. If L0
777 1 : // is empty, no compaction is necessary. If Lbase is empty, a move-based
778 1 : // compaction from L0 would occur.
779 1 : compactionDebt += bytesAddedToNextLevel + lbaseSize
780 1 : }
781 :
782 : // loop invariant: At the beginning of the loop, bytesAddedToNextLevel is the
783 : // bytes added to `level` in the loop.
784 1 : for level := p.baseLevel; level < numLevels-1; level++ {
785 1 : levelSize := p.vers.Levels[level].AggregateSize() + bytesAddedToNextLevel
786 1 : nextLevelSize := p.vers.Levels[level+1].AggregateSize()
787 1 : if levelSize > uint64(p.levelMaxBytes[level]) {
788 1 : bytesAddedToNextLevel = levelSize - uint64(p.levelMaxBytes[level])
789 1 : if nextLevelSize > 0 {
790 1 : // We only incur compaction debt if the next level contains data. If the
791 1 : // next level is empty, a move-based compaction would be used.
792 1 : levelRatio := float64(nextLevelSize) / float64(levelSize)
793 1 : // The current level contributes bytesAddedToNextLevel to compactions.
794 1 : // The next level contributes levelRatio * bytesAddedToNextLevel.
795 1 : compactionDebt += uint64(float64(bytesAddedToNextLevel) * (levelRatio + 1))
796 1 : }
797 1 : } else {
798 1 : // We're not moving any bytes to the next level.
799 1 : bytesAddedToNextLevel = 0
800 1 : }
801 : }
802 1 : return compactionDebt
803 : }
804 :
805 1 : func (p *compactionPickerByScore) initLevelMaxBytes(inProgressCompactions []compactionInfo) {
806 1 : // The levelMaxBytes calculations here differ from RocksDB in two ways:
807 1 : //
808 1 : // 1. The use of dbSize vs maxLevelSize. RocksDB uses the size of the maximum
809 1 : // level in L1-L6, rather than determining the size of the bottom level
810 1 : // based on the total amount of data in the dB. The RocksDB calculation is
811 1 : // problematic if L0 contains a significant fraction of data, or if the
812 1 : // level sizes are roughly equal and thus there is a significant fraction
813 1 : // of data outside of the largest level.
814 1 : //
815 1 : // 2. Not adjusting the size of Lbase based on L0. RocksDB computes
816 1 : // baseBytesMax as the maximum of the configured LBaseMaxBytes and the
817 1 : // size of L0. This is problematic because baseBytesMax is used to compute
818 1 : // the max size of lower levels. A very large baseBytesMax will result in
819 1 : // an overly large value for the size of lower levels which will caused
820 1 : // those levels not to be compacted even when they should be
821 1 : // compacted. This often results in "inverted" LSM shapes where Ln is
822 1 : // larger than Ln+1.
823 1 :
824 1 : // Determine the first non-empty level and the total DB size.
825 1 : firstNonEmptyLevel := -1
826 1 : var dbSize uint64
827 1 : for level := 1; level < numLevels; level++ {
828 1 : if p.vers.Levels[level].AggregateSize() > 0 {
829 1 : if firstNonEmptyLevel == -1 {
830 1 : firstNonEmptyLevel = level
831 1 : }
832 1 : dbSize += p.vers.Levels[level].AggregateSize()
833 : }
834 : }
835 1 : for _, c := range inProgressCompactions {
836 1 : if c.outputLevel == 0 || c.outputLevel == -1 {
837 1 : continue
838 : }
839 1 : if c.inputs[0].level == 0 && (firstNonEmptyLevel == -1 || c.outputLevel < firstNonEmptyLevel) {
840 1 : firstNonEmptyLevel = c.outputLevel
841 1 : }
842 : }
843 :
844 : // Initialize the max-bytes setting for each level to "infinity" which will
845 : // disallow compaction for that level. We'll fill in the actual value below
846 : // for levels we want to allow compactions from.
847 1 : for level := 0; level < numLevels; level++ {
848 1 : p.levelMaxBytes[level] = math.MaxInt64
849 1 : }
850 :
851 1 : dbSizeBelowL0 := dbSize
852 1 : dbSize += p.vers.Levels[0].AggregateSize()
853 1 : p.dbSizeBytes = dbSize
854 1 : if dbSizeBelowL0 == 0 {
855 1 : // No levels for L1 and up contain any data. Target L0 compactions for the
856 1 : // last level or to the level to which there is an ongoing L0 compaction.
857 1 : p.baseLevel = numLevels - 1
858 1 : if firstNonEmptyLevel >= 0 {
859 1 : p.baseLevel = firstNonEmptyLevel
860 1 : }
861 1 : return
862 : }
863 :
864 1 : bottomLevelSize := dbSize - dbSize/uint64(p.opts.Experimental.LevelMultiplier)
865 1 :
866 1 : curLevelSize := bottomLevelSize
867 1 : for level := numLevels - 2; level >= firstNonEmptyLevel; level-- {
868 1 : curLevelSize = uint64(float64(curLevelSize) / float64(p.opts.Experimental.LevelMultiplier))
869 1 : }
870 :
871 : // Compute base level (where L0 data is compacted to).
872 1 : baseBytesMax := uint64(p.opts.LBaseMaxBytes)
873 1 : p.baseLevel = firstNonEmptyLevel
874 1 : for p.baseLevel > 1 && curLevelSize > baseBytesMax {
875 1 : p.baseLevel--
876 1 : curLevelSize = uint64(float64(curLevelSize) / float64(p.opts.Experimental.LevelMultiplier))
877 1 : }
878 :
879 1 : smoothedLevelMultiplier := 1.0
880 1 : if p.baseLevel < numLevels-1 {
881 1 : smoothedLevelMultiplier = math.Pow(
882 1 : float64(bottomLevelSize)/float64(baseBytesMax),
883 1 : 1.0/float64(numLevels-p.baseLevel-1))
884 1 : }
885 :
886 1 : levelSize := float64(baseBytesMax)
887 1 : for level := p.baseLevel; level < numLevels; level++ {
888 1 : if level > p.baseLevel && levelSize > 0 {
889 1 : levelSize *= smoothedLevelMultiplier
890 1 : }
891 : // Round the result since test cases use small target level sizes, which
892 : // can be impacted by floating-point imprecision + integer truncation.
893 1 : roundedLevelSize := math.Round(levelSize)
894 1 : if roundedLevelSize > float64(math.MaxInt64) {
895 0 : p.levelMaxBytes[level] = math.MaxInt64
896 1 : } else {
897 1 : p.levelMaxBytes[level] = int64(roundedLevelSize)
898 1 : }
899 : }
900 : }
901 :
902 : type levelSizeAdjust struct {
903 : incomingActualBytes uint64
904 : outgoingActualBytes uint64
905 : outgoingCompensatedBytes uint64
906 : }
907 :
908 1 : func (a levelSizeAdjust) compensated() uint64 {
909 1 : return a.incomingActualBytes - a.outgoingCompensatedBytes
910 1 : }
911 :
912 1 : func (a levelSizeAdjust) actual() uint64 {
913 1 : return a.incomingActualBytes - a.outgoingActualBytes
914 1 : }
915 :
916 1 : func calculateSizeAdjust(inProgressCompactions []compactionInfo) [numLevels]levelSizeAdjust {
917 1 : // Compute size adjustments for each level based on the in-progress
918 1 : // compactions. We sum the file sizes of all files leaving and entering each
919 1 : // level in in-progress compactions. For outgoing files, we also sum a
920 1 : // separate sum of 'compensated file sizes', which are inflated according
921 1 : // to deletion estimates.
922 1 : //
923 1 : // When we adjust a level's size according to these values during score
924 1 : // calculation, we subtract the compensated size of start level inputs to
925 1 : // account for the fact that score calculation uses compensated sizes.
926 1 : //
927 1 : // Since compensated file sizes may be compensated because they reclaim
928 1 : // space from the output level's files, we only add the real file size to
929 1 : // the output level.
930 1 : //
931 1 : // This is slightly different from RocksDB's behavior, which simply elides
932 1 : // compacting files from the level size calculation.
933 1 : var sizeAdjust [numLevels]levelSizeAdjust
934 1 : for i := range inProgressCompactions {
935 1 : c := &inProgressCompactions[i]
936 1 : // If this compaction's version edit has already been applied, there's
937 1 : // no need to adjust: The LSM we'll examine will already reflect the
938 1 : // new LSM state.
939 1 : if c.versionEditApplied {
940 1 : continue
941 : }
942 :
943 1 : for _, input := range c.inputs {
944 1 : actualSize := input.files.AggregateSizeSum()
945 1 : compensatedSize := totalCompensatedSize(input.files.All())
946 1 :
947 1 : if input.level != c.outputLevel {
948 1 : sizeAdjust[input.level].outgoingCompensatedBytes += compensatedSize
949 1 : sizeAdjust[input.level].outgoingActualBytes += actualSize
950 1 : if c.outputLevel != -1 {
951 1 : sizeAdjust[c.outputLevel].incomingActualBytes += actualSize
952 1 : }
953 : }
954 : }
955 : }
956 1 : return sizeAdjust
957 : }
958 :
959 : // calculateLevelScores calculates the candidateLevelInfo for all levels and
960 : // returns them in decreasing score order.
961 : func (p *compactionPickerByScore) calculateLevelScores(
962 : inProgressCompactions []compactionInfo,
963 1 : ) [numLevels]candidateLevelInfo {
964 1 : var scores [numLevels]candidateLevelInfo
965 1 : for i := range scores {
966 1 : scores[i].level = i
967 1 : scores[i].outputLevel = i + 1
968 1 : }
969 1 : l0FillFactor := calculateL0FillFactor(p.vers, p.latestVersionState.l0Organizer, p.opts, inProgressCompactions)
970 1 : scores[0] = candidateLevelInfo{
971 1 : outputLevel: p.baseLevel,
972 1 : fillFactor: l0FillFactor,
973 1 : compensatedFillFactor: l0FillFactor, // No compensation for L0.
974 1 : }
975 1 : sizeAdjust := calculateSizeAdjust(inProgressCompactions)
976 1 : for level := 1; level < numLevels; level++ {
977 1 : // Actual file size.
978 1 : compensatedLevelSize := p.vers.Levels[level].AggregateSize()
979 1 : // Deletions.
980 1 : delBytes := deletionBytesAnnotator.LevelAnnotation(p.vers.Levels[level])
981 1 : compensatedLevelSize += delBytes.PointDels + delBytes.RangeDels
982 1 : // Adjustments for in-progress compactions.
983 1 : compensatedLevelSize += sizeAdjust[level].compensated()
984 1 : scores[level].compensatedFillFactor = float64(compensatedLevelSize) / float64(p.levelMaxBytes[level])
985 1 : scores[level].fillFactor = float64(p.vers.Levels[level].AggregateSize()+sizeAdjust[level].actual()) / float64(p.levelMaxBytes[level])
986 1 : }
987 :
988 : // Adjust each level's fill factor by the fill factor of the next level to get
989 : // an (uncompensated) score; and each level's compensated fill factor by the
990 : // fill factor of the next level to get a compensated score.
991 : //
992 : // The compensated score is used to determine if the level should be compacted
993 : // at all. The (uncompensated) score is used as the value used to rank levels.
994 : //
995 : // If the next level has a high fill factor, and is thus a priority for
996 : // compaction, this reduces the priority for compacting the current level. If
997 : // the next level has a low fill factor (i.e. it is below its target size),
998 : // this increases the priority for compacting the current level.
999 : //
1000 : // The effect of this adjustment is to help prioritize compactions in lower
1001 : // levels. The following example shows the scores and the fill factors. In this
1002 : // scenario, L0 has 68 sublevels. L3 (a.k.a. Lbase) is significantly above its
1003 : // target size. The original score prioritizes compactions from those two
1004 : // levels, but doing so ends up causing a future problem: data piles up in the
1005 : // higher levels, starving L5->L6 compactions, and to a lesser degree starving
1006 : // L4->L5 compactions.
1007 : //
1008 : // Note that in the example shown there is no level size compensation so the
1009 : // compensatedFillFactor and fillFactor are the same for each level.
1010 : //
1011 : // score fillFactor compensatedFillFactor size max-size
1012 : // L0 3.2 68.0 68.0 2.2 G -
1013 : // L3 3.2 21.1 21.1 1.3 G 64 M
1014 : // L4 3.4 6.7 6.7 3.1 G 467 M
1015 : // L5 3.4 2.0 2.0 6.6 G 3.3 G
1016 : // L6 0 0.6 0.6 14 G 24 G
1017 : //
1018 : // TODO(radu): the way compensation works needs some rethinking. For example,
1019 : // if compacting L5 can free up a lot of space in L6, the score of L5 should
1020 : // go *up* with the fill factor of L6, not the other way around.
1021 1 : for level := 0; level < numLevels; level++ {
1022 1 : if level > 0 && level < p.baseLevel {
1023 1 : continue
1024 : }
1025 1 : const compensatedFillFactorThreshold = 1.0
1026 1 : if scores[level].compensatedFillFactor < compensatedFillFactorThreshold {
1027 1 : // No need to compact this level; score stays 0.
1028 1 : continue
1029 : }
1030 1 : score := scores[level].fillFactor
1031 1 : compensatedScore := scores[level].compensatedFillFactor
1032 1 : if level < numLevels-1 {
1033 1 : nextLevel := scores[level].outputLevel
1034 1 : // Avoid absurdly large scores by placing a floor on the factor that we'll
1035 1 : // adjust a level by. The value of 0.01 was chosen somewhat arbitrarily.
1036 1 : denominator := max(0.01, scores[nextLevel].fillFactor)
1037 1 : score /= denominator
1038 1 : compensatedScore /= denominator
1039 1 : }
1040 :
1041 : // We calculated a compensated score above by dividing the
1042 : // compensatedFillFactor by the next level's fill factor. Previous
1043 : // versions of Pebble had a default behavior of only considering levels
1044 : // with a compensatedScore >= 1.0 eligible for compaction. This wasn't a
1045 : // principled decision and has been experimentally observed to limit
1046 : // productive compactions that would reclaim disk space, but were
1047 : // prohibited because the output level's fill factor was > 1.0.
1048 : //
1049 : // We allow the use of old behavior through the
1050 : // UseDeprecatedCompensatedScore option, which if true, only considers
1051 : // the level eligible for compaction iff both compensatedFillFactor and
1052 : // compensatedScore are >= 1.0.
1053 : //
1054 : // Otherwise, only L0 requires the compensatedScore to be >= 1.0; all
1055 : // other levels only require the compensatedFillFactor to be >= 1.0. L0
1056 : // is treated exceptionally out of concern that a large, LBase that's
1057 : // being compacted into the next level may prevent L0->Lbase compactions
1058 : // and attempting to pick L0 compactions may result in intra-L0
1059 : // compactions.
1060 1 : const compensatedScoreThreshold = 1.0
1061 1 : if (level == 0 || p.opts.Experimental.UseDeprecatedCompensatedScore()) &&
1062 1 : compensatedScore < compensatedScoreThreshold {
1063 1 : // No need to compact this level; score stays 0.
1064 1 : continue
1065 : }
1066 1 : scores[level].score = score
1067 : }
1068 : // Sort by score (decreasing) and break ties by level (increasing).
1069 1 : slices.SortFunc(scores[:], func(a, b candidateLevelInfo) int {
1070 1 : if a.score != b.score {
1071 1 : return cmp.Compare(b.score, a.score)
1072 1 : }
1073 1 : return cmp.Compare(a.level, b.level)
1074 : })
1075 1 : return scores
1076 : }
1077 :
1078 : // calculateL0FillFactor calculates a float value representing the relative
1079 : // priority of compacting L0. A value less than 1 indicates that L0 does not
1080 : // need any compactions.
1081 : //
1082 : // L0 is special in that files within L0 may overlap one another, so a different
1083 : // set of heuristics that take into account read amplification apply.
1084 : func calculateL0FillFactor(
1085 : vers *manifest.Version,
1086 : l0Organizer *manifest.L0Organizer,
1087 : opts *Options,
1088 : inProgressCompactions []compactionInfo,
1089 1 : ) float64 {
1090 1 : // Use the sublevel count to calculate the score. The base vs intra-L0
1091 1 : // compaction determination happens in pickAuto, not here.
1092 1 : score := float64(2*l0Organizer.MaxDepthAfterOngoingCompactions()) /
1093 1 : float64(opts.L0CompactionThreshold)
1094 1 :
1095 1 : // Also calculate a score based on the file count but use it only if it
1096 1 : // produces a higher score than the sublevel-based one. This heuristic is
1097 1 : // designed to accommodate cases where L0 is accumulating non-overlapping
1098 1 : // files in L0. Letting too many non-overlapping files accumulate in few
1099 1 : // sublevels is undesirable, because:
1100 1 : // 1) we can produce a massive backlog to compact once files do overlap.
1101 1 : // 2) constructing L0 sublevels has a runtime that grows superlinearly with
1102 1 : // the number of files in L0 and must be done while holding D.mu.
1103 1 : noncompactingFiles := vers.Levels[0].Len()
1104 1 : for _, c := range inProgressCompactions {
1105 1 : for _, cl := range c.inputs {
1106 1 : if cl.level == 0 {
1107 1 : noncompactingFiles -= cl.files.Len()
1108 1 : }
1109 : }
1110 : }
1111 1 : fileScore := float64(noncompactingFiles) / float64(opts.L0CompactionFileThreshold)
1112 1 : if score < fileScore {
1113 1 : score = fileScore
1114 1 : }
1115 1 : return score
1116 : }
1117 :
1118 : // pickCompactionSeedFile picks a file from `level` in the `vers` to build a
1119 : // compaction around. Currently, this function implements a heuristic similar to
1120 : // RocksDB's kMinOverlappingRatio, seeking to minimize write amplification. This
1121 : // function is linear with respect to the number of files in `level` and
1122 : // `outputLevel`.
1123 : func pickCompactionSeedFile(
1124 : vers *manifest.Version,
1125 : virtualBackings *manifest.VirtualBackings,
1126 : opts *Options,
1127 : level, outputLevel int,
1128 : earliestSnapshotSeqNum base.SeqNum,
1129 : problemSpans *problemspans.ByLevel,
1130 1 : ) (manifest.LevelFile, bool) {
1131 1 : // Select the file within the level to compact. We want to minimize write
1132 1 : // amplification, but also ensure that (a) deletes are propagated to the
1133 1 : // bottom level in a timely fashion, and (b) virtual sstables that are
1134 1 : // pinning backing sstables where most of the data is garbage are compacted
1135 1 : // away. Doing (a) and (b) reclaims disk space. A table's smallest sequence
1136 1 : // number provides a measure of its age. The ratio of overlapping-bytes /
1137 1 : // table-size gives an indication of write amplification (a smaller ratio is
1138 1 : // preferrable).
1139 1 : //
1140 1 : // The current heuristic is based off the RocksDB kMinOverlappingRatio
1141 1 : // heuristic. It chooses the file with the minimum overlapping ratio with
1142 1 : // the target level, which minimizes write amplification.
1143 1 : //
1144 1 : // The heuristic uses a "compensated size" for the denominator, which is the
1145 1 : // file size inflated by (a) an estimate of the space that may be reclaimed
1146 1 : // through compaction, and (b) a fraction of the amount of garbage in the
1147 1 : // backing sstable pinned by this (virtual) sstable.
1148 1 : //
1149 1 : // TODO(peter): For concurrent compactions, we may want to try harder to
1150 1 : // pick a seed file whose resulting compaction bounds do not overlap with
1151 1 : // an in-progress compaction.
1152 1 :
1153 1 : cmp := opts.Comparer.Compare
1154 1 : startIter := vers.Levels[level].Iter()
1155 1 : outputIter := vers.Levels[outputLevel].Iter()
1156 1 :
1157 1 : var file manifest.LevelFile
1158 1 : smallestRatio := uint64(math.MaxUint64)
1159 1 :
1160 1 : outputFile := outputIter.First()
1161 1 :
1162 1 : for f := startIter.First(); f != nil; f = startIter.Next() {
1163 1 : var overlappingBytes uint64
1164 1 : if f.IsCompacting() {
1165 1 : // Move on if this file is already being compacted. We'll likely
1166 1 : // still need to move past the overlapping output files regardless,
1167 1 : // but in cases where all start-level files are compacting we won't.
1168 1 : continue
1169 : }
1170 1 : if problemSpans != nil && problemSpans.Overlaps(level, f.UserKeyBounds()) {
1171 0 : // File touches problem span which temporarily disallows auto compactions.
1172 0 : continue
1173 : }
1174 :
1175 : // Trim any output-level files smaller than f.
1176 1 : for outputFile != nil && sstableKeyCompare(cmp, outputFile.Largest(), f.Smallest()) < 0 {
1177 1 : outputFile = outputIter.Next()
1178 1 : }
1179 :
1180 1 : skip := false
1181 1 : for outputFile != nil && sstableKeyCompare(cmp, outputFile.Smallest(), f.Largest()) <= 0 {
1182 1 : overlappingBytes += outputFile.Size
1183 1 : if outputFile.IsCompacting() {
1184 1 : // If one of the overlapping files is compacting, we're not going to be
1185 1 : // able to compact f anyway, so skip it.
1186 1 : skip = true
1187 1 : break
1188 : }
1189 1 : if problemSpans != nil && problemSpans.Overlaps(outputLevel, outputFile.UserKeyBounds()) {
1190 0 : // Overlapping file touches problem span which temporarily disallows auto compactions.
1191 0 : skip = true
1192 0 : break
1193 : }
1194 :
1195 : // For files in the bottommost level of the LSM, the
1196 : // Stats.RangeDeletionsBytesEstimate field is set to the estimate
1197 : // of bytes /within/ the file itself that may be dropped by
1198 : // recompacting the file. These bytes from obsolete keys would not
1199 : // need to be rewritten if we compacted `f` into `outputFile`, so
1200 : // they don't contribute to write amplification. Subtracting them
1201 : // out of the overlapping bytes helps prioritize these compactions
1202 : // that are cheaper than their file sizes suggest.
1203 1 : if outputLevel == numLevels-1 && outputFile.LargestSeqNum < earliestSnapshotSeqNum {
1204 1 : if stats, ok := outputFile.Stats(); ok {
1205 1 : overlappingBytes -= stats.RangeDeletionsBytesEstimate
1206 1 : }
1207 : }
1208 :
1209 : // If the file in the next level extends beyond f's largest key,
1210 : // break out and don't advance outputIter because f's successor
1211 : // might also overlap.
1212 : //
1213 : // Note, we stop as soon as we encounter an output-level file with a
1214 : // largest key beyond the input-level file's largest bound. We
1215 : // perform a simple user key comparison here using sstableKeyCompare
1216 : // which handles the potential for exclusive largest key bounds.
1217 : // There's some subtlety when the bounds are equal (eg, equal and
1218 : // inclusive, or equal and exclusive). Current Pebble doesn't split
1219 : // user keys across sstables within a level (and in format versions
1220 : // FormatSplitUserKeysMarkedCompacted and later we guarantee no
1221 : // split user keys exist within the entire LSM). In that case, we're
1222 : // assured that neither the input level nor the output level's next
1223 : // file shares the same user key, so compaction expansion will not
1224 : // include them in any compaction compacting `f`.
1225 : //
1226 : // NB: If we /did/ allow split user keys, or we're running on an
1227 : // old database with an earlier format major version where there are
1228 : // existing split user keys, this logic would be incorrect. Consider
1229 : // L1: [a#120,a#100] [a#80,a#60]
1230 : // L2: [a#55,a#45] [a#35,a#25] [a#15,a#5]
1231 : // While considering the first file in L1, [a#120,a#100], we'd skip
1232 : // past all of the files in L2. When considering the second file in
1233 : // L1, we'd improperly conclude that the second file overlaps
1234 : // nothing in the second level and is cheap to compact, when in
1235 : // reality we'd need to expand the compaction to include all 5
1236 : // files.
1237 1 : if sstableKeyCompare(cmp, outputFile.Largest(), f.Largest()) > 0 {
1238 1 : break
1239 : }
1240 1 : outputFile = outputIter.Next()
1241 : }
1242 1 : if skip {
1243 1 : continue
1244 : }
1245 :
1246 1 : compSz := tableCompensatedSize(f) + responsibleForGarbageBytes(virtualBackings, f)
1247 1 : scaledRatio := overlappingBytes * 1024 / compSz
1248 1 : if scaledRatio < smallestRatio {
1249 1 : smallestRatio = scaledRatio
1250 1 : file = startIter.Take()
1251 1 : }
1252 : }
1253 1 : return file, file.TableMetadata != nil
1254 : }
1255 :
1256 : // responsibleForGarbageBytes returns the amount of garbage in the backing
1257 : // sstable that we consider the responsibility of this virtual sstable. For
1258 : // non-virtual sstables, this is of course 0. For virtual sstables, we equally
1259 : // distribute the responsibility of the garbage across all the virtual
1260 : // sstables that are referencing the same backing sstable. One could
1261 : // alternatively distribute this in proportion to the virtual sst sizes, but
1262 : // it isn't clear that more sophisticated heuristics are worth it, given that
1263 : // the garbage cannot be reclaimed until all the referencing virtual sstables
1264 : // are compacted.
1265 : func responsibleForGarbageBytes(
1266 : virtualBackings *manifest.VirtualBackings, m *manifest.TableMetadata,
1267 1 : ) uint64 {
1268 1 : if !m.Virtual {
1269 1 : return 0
1270 1 : }
1271 1 : useCount, virtualizedSize := virtualBackings.Usage(m.TableBacking.DiskFileNum)
1272 1 : // Since virtualizedSize is the sum of the estimated size of all virtual
1273 1 : // ssts, we allow for the possibility that virtualizedSize could exceed
1274 1 : // m.TableBacking.Size.
1275 1 : totalGarbage := int64(m.TableBacking.Size) - int64(virtualizedSize)
1276 1 : if totalGarbage <= 0 {
1277 1 : return 0
1278 1 : }
1279 1 : if useCount == 0 {
1280 0 : // This cannot happen if m exists in the latest version. The call to
1281 0 : // ResponsibleForGarbageBytes during compaction picking ensures that m
1282 0 : // exists in the latest version by holding versionSet.logLock.
1283 0 : panic(errors.AssertionFailedf("%s has zero useCount", m.String()))
1284 : }
1285 1 : return uint64(totalGarbage) / uint64(useCount)
1286 : }
1287 :
1288 1 : func (p *compactionPickerByScore) getCompactionConcurrency() int {
1289 1 : lower, upper := p.opts.CompactionConcurrencyRange()
1290 1 : if lower >= upper {
1291 1 : return upper
1292 1 : }
1293 : // Compaction concurrency is controlled by L0 read-amp. We allow one
1294 : // additional compaction per L0CompactionConcurrency sublevels, as well as
1295 : // one additional compaction per CompactionDebtConcurrency bytes of
1296 : // compaction debt. Compaction concurrency is tied to L0 sublevels as that
1297 : // signal is independent of the database size. We tack on the compaction
1298 : // debt as a second signal to prevent compaction concurrency from dropping
1299 : // significantly right after a base compaction finishes, and before those
1300 : // bytes have been compacted further down the LSM.
1301 : //
1302 : // Let n be the number of in-progress compactions.
1303 : //
1304 : // l0ReadAmp >= ccSignal1 then can run another compaction, where
1305 : // ccSignal1 = n * p.opts.Experimental.L0CompactionConcurrency
1306 : // Rearranging,
1307 : // n <= l0ReadAmp / p.opts.Experimental.L0CompactionConcurrency.
1308 : // So we can run up to
1309 : // l0ReadAmp / p.opts.Experimental.L0CompactionConcurrency extra compactions.
1310 1 : l0ReadAmpCompactions := 0
1311 1 : if p.opts.Experimental.L0CompactionConcurrency > 0 {
1312 1 : l0ReadAmp := p.latestVersionState.l0Organizer.MaxDepthAfterOngoingCompactions()
1313 1 : l0ReadAmpCompactions = (l0ReadAmp / p.opts.Experimental.L0CompactionConcurrency)
1314 1 : }
1315 : // compactionDebt >= ccSignal2 then can run another compaction, where
1316 : // ccSignal2 = uint64(n) * p.opts.Experimental.CompactionDebtConcurrency
1317 : // Rearranging,
1318 : // n <= compactionDebt / p.opts.Experimental.CompactionDebtConcurrency
1319 : // So we can run up to
1320 : // compactionDebt / p.opts.Experimental.CompactionDebtConcurrency extra
1321 : // compactions.
1322 1 : compactionDebtCompactions := 0
1323 1 : if p.opts.Experimental.CompactionDebtConcurrency > 0 {
1324 1 : compactionDebt := p.estimatedCompactionDebt()
1325 1 : compactionDebtCompactions = int(compactionDebt / p.opts.Experimental.CompactionDebtConcurrency)
1326 1 : }
1327 :
1328 1 : compactableGarbageCompactions := 0
1329 1 : garbageFractionLimit := p.opts.Experimental.CompactionGarbageFractionForMaxConcurrency()
1330 1 : if garbageFractionLimit > 0 && p.dbSizeBytes > 0 {
1331 1 : delBytes := deletionBytesAnnotator.MultiLevelAnnotation(p.vers.Levels[:])
1332 1 : compactableGarbageBytes := delBytes.PointDels + delBytes.RangeDels
1333 1 : garbageFraction := float64(compactableGarbageBytes) / float64(p.dbSizeBytes)
1334 1 : compactableGarbageCompactions =
1335 1 : int((garbageFraction / garbageFractionLimit) * float64(upper-lower))
1336 1 : }
1337 :
1338 1 : extraCompactions := max(l0ReadAmpCompactions, compactionDebtCompactions, compactableGarbageCompactions, 0)
1339 1 :
1340 1 : return min(lower+extraCompactions, upper)
1341 : }
1342 :
1343 : // TODO(sumeer): remove unless someone actually finds this useful.
1344 : func (p *compactionPickerByScore) logCompactionForTesting(
1345 : env compactionEnv, scores [numLevels]candidateLevelInfo, pc *pickedTableCompaction,
1346 0 : ) {
1347 0 : var buf bytes.Buffer
1348 0 : for i := 0; i < numLevels; i++ {
1349 0 : if i != 0 && i < p.baseLevel {
1350 0 : continue
1351 : }
1352 :
1353 0 : var info *candidateLevelInfo
1354 0 : for j := range scores {
1355 0 : if scores[j].level == i {
1356 0 : info = &scores[j]
1357 0 : break
1358 : }
1359 : }
1360 :
1361 0 : marker := " "
1362 0 : if pc.startLevel.level == info.level {
1363 0 : marker = "*"
1364 0 : }
1365 0 : fmt.Fprintf(&buf, " %sL%d: score:%5.1f fillFactor:%5.1f compensatedFillFactor:%5.1f %8s %8s",
1366 0 : marker, info.level, info.score, info.fillFactor, info.compensatedFillFactor,
1367 0 : humanize.Bytes.Int64(int64(totalCompensatedSize(
1368 0 : p.vers.Levels[info.level].All(),
1369 0 : ))),
1370 0 : humanize.Bytes.Int64(p.levelMaxBytes[info.level]),
1371 0 : )
1372 0 :
1373 0 : count := 0
1374 0 : for i := range env.inProgressCompactions {
1375 0 : c := &env.inProgressCompactions[i]
1376 0 : if c.inputs[0].level != info.level {
1377 0 : continue
1378 : }
1379 0 : count++
1380 0 : if count == 1 {
1381 0 : fmt.Fprintf(&buf, " [")
1382 0 : } else {
1383 0 : fmt.Fprintf(&buf, " ")
1384 0 : }
1385 0 : fmt.Fprintf(&buf, "L%d->L%d", c.inputs[0].level, c.outputLevel)
1386 : }
1387 0 : if count > 0 {
1388 0 : fmt.Fprintf(&buf, "]")
1389 0 : }
1390 0 : fmt.Fprintf(&buf, "\n")
1391 : }
1392 0 : p.opts.Logger.Infof("pickAuto: L%d->L%d\n%s",
1393 0 : pc.startLevel.level, pc.outputLevel.level, buf.String())
1394 : }
1395 :
1396 : // pickHighPrioritySpaceCompaction checks for a high-priority space reclamation
1397 : // compaction. Under some circumstances, we want to persue a compaction for the
1398 : // purpose of reclaiming disk space even when there are eligible default
1399 : // compactions.
1400 : func (p *compactionPickerByScore) pickHighPrioritySpaceCompaction(
1401 : env compactionEnv,
1402 1 : ) pickedCompaction {
1403 1 : if pc := p.pickBlobFileRewriteCompactionHighPriority(env); pc != nil {
1404 1 : return pc
1405 1 : }
1406 : // NB: We can't just return the above result because the above func returns
1407 : // a *pickedBlobFileCompaction, not a pickedCompaction. We need to return an
1408 : // untyped nil.
1409 1 : return nil
1410 : }
1411 :
1412 : // pickAutoScore picks the best score-based compaction, if any.
1413 : //
1414 : // On each call, pickAutoScore computes per-level size adjustments based on
1415 : // in-progress compactions, and computes a per-level score. The levels are
1416 : // iterated over in decreasing score order trying to find a valid compaction
1417 : // anchored at that level.
1418 : //
1419 : // If a score-based compaction cannot be found, pickAuto falls back to looking
1420 : // for an elision-only compaction to remove obsolete keys.
1421 1 : func (p *compactionPickerByScore) pickAutoScore(env compactionEnv) pickedCompaction {
1422 1 : scores := p.calculateLevelScores(env.inProgressCompactions)
1423 1 :
1424 1 : // Check for a score-based compaction. candidateLevelInfos are first sorted
1425 1 : // by whether they should be compacted, so if we find a level which shouldn't
1426 1 : // be compacted, we can break early.
1427 1 : for i := range scores {
1428 1 : info := &scores[i]
1429 1 : if !info.shouldCompact() {
1430 1 : break
1431 : }
1432 1 : if info.level == numLevels-1 {
1433 1 : continue
1434 : }
1435 :
1436 1 : if info.level == 0 {
1437 1 : ptc := pickL0(env, p.opts, p.vers, p.latestVersionState.l0Organizer, p.baseLevel)
1438 1 : if ptc != nil {
1439 1 : p.addScoresToPickedCompactionMetrics(ptc, scores)
1440 1 : ptc.score = info.score
1441 1 : if false {
1442 0 : p.logCompactionForTesting(env, scores, ptc)
1443 0 : }
1444 1 : return ptc
1445 : }
1446 1 : continue
1447 : }
1448 :
1449 : // info.level > 0
1450 1 : var ok bool
1451 1 : info.file, ok = pickCompactionSeedFile(p.vers, &p.latestVersionState.virtualBackings, p.opts, info.level, info.outputLevel, env.earliestSnapshotSeqNum, env.problemSpans)
1452 1 : if !ok {
1453 1 : continue
1454 : }
1455 :
1456 1 : pc := pickAutoLPositive(env, p.opts, p.vers, p.latestVersionState.l0Organizer, *info, p.baseLevel)
1457 1 : if pc != nil {
1458 1 : p.addScoresToPickedCompactionMetrics(pc, scores)
1459 1 : pc.score = info.score
1460 1 : if false {
1461 0 : p.logCompactionForTesting(env, scores, pc)
1462 0 : }
1463 1 : return pc
1464 : }
1465 : }
1466 1 : return nil
1467 : }
1468 :
1469 : // pickAutoNonScore picks the best non-score-based compaction, if any.
1470 1 : func (p *compactionPickerByScore) pickAutoNonScore(env compactionEnv) (pc pickedCompaction) {
1471 1 : // Check for files which contain excessive point tombstones that could slow
1472 1 : // down reads. Unlike elision-only compactions, these compactions may select
1473 1 : // a file at any level rather than only the lowest level.
1474 1 : if pc := p.pickTombstoneDensityCompaction(env); pc != nil {
1475 1 : return pc
1476 1 : }
1477 :
1478 : // Check for L6 files with tombstones that may be elided. These files may
1479 : // exist if a snapshot prevented the elision of a tombstone or because of
1480 : // a move compaction. These are low-priority compactions because they
1481 : // don't help us keep up with writes, just reclaim disk space.
1482 1 : if pc := p.pickElisionOnlyCompaction(env); pc != nil {
1483 1 : return pc
1484 1 : }
1485 :
1486 : // Check for virtual SST rewrites. These compactions materialize virtual tables
1487 : // to reclaim space in backing files with low utilization.
1488 1 : if pc := p.pickVirtualRewriteCompaction(env); pc != nil {
1489 1 : return pc
1490 1 : }
1491 :
1492 : // Check for blob file rewrites. These are low-priority compactions because
1493 : // they don't help us keep up with writes, just reclaim disk space.
1494 1 : if pc := p.pickBlobFileRewriteCompactionLowPriority(env); pc != nil {
1495 1 : return pc
1496 1 : }
1497 :
1498 1 : if pc := p.pickReadTriggeredCompaction(env); pc != nil {
1499 0 : return pc
1500 0 : }
1501 :
1502 : // NB: This should only be run if a read compaction wasn't
1503 : // scheduled.
1504 : //
1505 : // We won't be scheduling a read compaction right now, and in
1506 : // read heavy workloads, compactions won't be scheduled frequently
1507 : // because flushes aren't frequent. So we need to signal to the
1508 : // iterator to schedule a compaction when it adds compactions to
1509 : // the read compaction queue.
1510 : //
1511 : // We need the nil check here because without it, we have some
1512 : // tests which don't set that variable fail. Since there's a
1513 : // chance that one of those tests wouldn't want extra compactions
1514 : // to be scheduled, I added this check here, instead of
1515 : // setting rescheduleReadCompaction in those tests.
1516 1 : if env.readCompactionEnv.rescheduleReadCompaction != nil {
1517 1 : *env.readCompactionEnv.rescheduleReadCompaction = true
1518 1 : }
1519 :
1520 : // At the lowest possible compaction-picking priority, look for files marked
1521 : // for compaction. Pebble will mark files for compaction if they have atomic
1522 : // compaction units that span multiple files. While current Pebble code does
1523 : // not construct such sstables, RocksDB and earlier versions of Pebble may
1524 : // have created them. These split user keys form sets of files that must be
1525 : // compacted together for correctness (referred to as "atomic compaction
1526 : // units" within the code). Rewrite them in-place.
1527 : //
1528 : // It's also possible that a file may have been marked for compaction by
1529 : // even earlier versions of Pebble code, since TableMetadata's
1530 : // MarkedForCompaction field is persisted in the manifest. That's okay. We
1531 : // previously would've ignored the designation, whereas now we'll re-compact
1532 : // the file in place.
1533 1 : if p.vers.MarkedForCompaction.Count() > 0 {
1534 0 : if pc := p.pickRewriteCompaction(env); pc != nil {
1535 0 : return pc
1536 0 : }
1537 : }
1538 :
1539 1 : return nil
1540 : }
1541 :
1542 : func (p *compactionPickerByScore) addScoresToPickedCompactionMetrics(
1543 : pc *pickedTableCompaction, candInfo [numLevels]candidateLevelInfo,
1544 1 : ) {
1545 1 :
1546 1 : // candInfo is sorted by score, not by compaction level.
1547 1 : infoByLevel := [numLevels]candidateLevelInfo{}
1548 1 : for i := range candInfo {
1549 1 : level := candInfo[i].level
1550 1 : infoByLevel[level] = candInfo[i]
1551 1 : }
1552 : // Gather the compaction scores for the levels participating in the compaction.
1553 1 : pc.pickerMetrics.scores = make([]float64, len(pc.inputs))
1554 1 : inputIdx := 0
1555 1 : for i := range infoByLevel {
1556 1 : if pc.inputs[inputIdx].level == infoByLevel[i].level {
1557 1 : pc.pickerMetrics.scores[inputIdx] = infoByLevel[i].score
1558 1 : inputIdx++
1559 1 : }
1560 1 : if inputIdx == len(pc.inputs) {
1561 1 : break
1562 : }
1563 : }
1564 : }
1565 :
1566 : // elisionOnlyAnnotator is a manifest.TableAnnotator that annotates B-Tree
1567 : // nodes with the *fileMetadata of a file meeting the obsolete keys criteria
1568 : // for an elision-only compaction within the subtree. If multiple files meet
1569 : // the criteria, it chooses whichever file has the lowest LargestSeqNum. The
1570 : // lowest LargestSeqNum file will be the first eligible for an elision-only
1571 : // compaction once snapshots less than or equal to its LargestSeqNum are closed.
1572 : var elisionOnlyAnnotator = manifest.MakePickFileAnnotator(
1573 : manifest.NewTableAnnotationIdx(),
1574 : manifest.PickFileAnnotatorFuncs{
1575 1 : Filter: func(f *manifest.TableMetadata) (eligible bool, cacheOK bool) {
1576 1 : if f.IsCompacting() {
1577 1 : return false, true
1578 1 : }
1579 :
1580 1 : backingProps, backingPropsValid := f.TableBacking.Properties()
1581 1 : stats, statsValid := f.Stats()
1582 1 : if !backingPropsValid || !statsValid {
1583 1 : return false, false
1584 1 : }
1585 :
1586 : // Bottommost files are large and not worthwhile to compact just
1587 : // to remove a few tombstones. Consider a file eligible only if
1588 : // either its own range deletions delete at least 10% of its data or
1589 : // its deletion tombstones make at least 10% of its entries.
1590 : //
1591 : // TODO(jackson): This does not account for duplicate user keys
1592 : // which may be collapsed. Ideally, we would have 'obsolete keys'
1593 : // statistics that would include tombstones, the keys that are
1594 : // dropped by tombstones and duplicated user keys. See #847.
1595 : //
1596 : // Note that tables that contain exclusively range keys (i.e. no point keys,
1597 : // `NumEntries` and `RangeDeletionsBytesEstimate` are both zero) are excluded
1598 : // from elision-only compactions.
1599 : // TODO(travers): Consider an alternative heuristic for elision of range-keys.
1600 1 : eligible = stats.RangeDeletionsBytesEstimate*10 >= f.Size || backingProps.NumDeletions*10 > backingProps.NumEntries
1601 1 : return eligible, true
1602 : },
1603 1 : Compare: func(f1 *manifest.TableMetadata, f2 *manifest.TableMetadata) bool {
1604 1 : return f1.LargestSeqNum < f2.LargestSeqNum
1605 1 : },
1606 : },
1607 : )
1608 :
1609 : // pickedCompactionFromCandidateFile creates a pickedCompaction from a *fileMetadata
1610 : // with various checks to ensure that the file still exists in the expected level
1611 : // and isn't already being compacted.
1612 : func (p *compactionPickerByScore) pickedCompactionFromCandidateFile(
1613 : candidate *manifest.TableMetadata,
1614 : env compactionEnv,
1615 : startLevel int,
1616 : outputLevel int,
1617 : kind compactionKind,
1618 1 : ) *pickedTableCompaction {
1619 1 : if candidate == nil || candidate.IsCompacting() {
1620 1 : return nil
1621 1 : }
1622 :
1623 1 : var inputs manifest.LevelSlice
1624 1 : if startLevel == 0 && outputLevel > 0 {
1625 1 : // Overlapping L0 files must also be compacted alongside the candidate.
1626 1 : // Some compactions attempt to rewrite a file in place (e.g. virtual rewrite)
1627 1 : // so we only do this for L0->Lbase compactions.
1628 1 : inputs = p.vers.Overlaps(0, candidate.UserKeyBounds())
1629 1 : } else {
1630 1 : inputs = p.vers.Levels[startLevel].Find(p.opts.Comparer.Compare, candidate)
1631 1 : }
1632 1 : if invariants.Enabled {
1633 1 : found := false
1634 1 : for f := range inputs.All() {
1635 1 : if f.TableNum == candidate.TableNum {
1636 1 : found = true
1637 1 : }
1638 : }
1639 1 : if !found {
1640 0 : panic(fmt.Sprintf("file %s not found in level %d as expected", candidate.TableNum, startLevel))
1641 : }
1642 : }
1643 :
1644 1 : pc := newPickedTableCompaction(p.opts, p.vers, p.latestVersionState.l0Organizer,
1645 1 : startLevel, outputLevel, p.baseLevel)
1646 1 : pc.kind = kind
1647 1 : pc.startLevel.files = inputs
1648 1 :
1649 1 : if !pc.setupInputs(p.opts, env.diskAvailBytes, env.inProgressCompactions, pc.startLevel, env.problemSpans) {
1650 1 : return nil
1651 1 : }
1652 1 : return pc
1653 : }
1654 :
1655 : // pickElisionOnlyCompaction looks for compactions of sstables in the
1656 : // bottommost level containing obsolete records that may now be dropped.
1657 : func (p *compactionPickerByScore) pickElisionOnlyCompaction(
1658 : env compactionEnv,
1659 1 : ) (pc *pickedTableCompaction) {
1660 1 : if p.opts.private.disableElisionOnlyCompactions {
1661 1 : return nil
1662 1 : }
1663 1 : candidate := elisionOnlyAnnotator.LevelAnnotation(p.vers.Levels[numLevels-1])
1664 1 : if candidate == nil {
1665 1 : return nil
1666 1 : }
1667 1 : if candidate.LargestSeqNum >= env.earliestSnapshotSeqNum {
1668 1 : return nil
1669 1 : }
1670 1 : return p.pickedCompactionFromCandidateFile(candidate, env, numLevels-1, numLevels-1, compactionKindElisionOnly)
1671 : }
1672 :
1673 : // pickRewriteCompaction attempts to construct a compaction that
1674 : // rewrites a file marked for compaction. pickRewriteCompaction will
1675 : // pull in adjacent files in the file's atomic compaction unit if
1676 : // necessary. A rewrite compaction outputs files to the same level as
1677 : // the input level.
1678 : func (p *compactionPickerByScore) pickRewriteCompaction(
1679 : env compactionEnv,
1680 0 : ) (pc *pickedTableCompaction) {
1681 0 : for candidate, level := range p.vers.MarkedForCompaction.Ascending() {
1682 0 : if pc := p.pickedCompactionFromCandidateFile(candidate, env, level, level, compactionKindRewrite); pc != nil {
1683 0 : return pc
1684 0 : }
1685 : }
1686 0 : return nil
1687 : }
1688 :
1689 : // pickVirtualRewriteCompaction looks for backing tables that have a low percentage
1690 : // of referenced data and materializes their virtual sstables.
1691 : func (p *compactionPickerByScore) pickVirtualRewriteCompaction(
1692 : env compactionEnv,
1693 1 : ) *pickedTableCompaction {
1694 1 :
1695 1 : for _, c := range env.inProgressCompactions {
1696 1 : // Allow only one virtual rewrite compaction at a time.
1697 1 : if c.kind == compactionKindVirtualRewrite {
1698 1 : return nil
1699 1 : }
1700 : }
1701 :
1702 : // We'll pick one virtual table at a time to materialize. This works with our
1703 : // compaction system, which currently doesn't support outputting to multiple levels
1704 : // or selecting files that aren't contiguous in a level. Successfully materializing
1705 : // one of the backing's virtual table will also make the backing more likely to be
1706 : // picked again, since the space amp will increase.
1707 1 : referencedDataPct, _, vtablesByLevel := p.latestVersionState.virtualBackings.ReplacementCandidate()
1708 1 :
1709 1 : if 1-referencedDataPct < p.opts.Experimental.VirtualTableRewriteUnreferencedFraction() {
1710 1 : return nil
1711 1 : }
1712 :
1713 1 : for level, tables := range vtablesByLevel {
1714 1 : for _, vt := range tables {
1715 1 : if vt.IsCompacting() {
1716 1 : continue
1717 : }
1718 1 : if pc := p.pickedCompactionFromCandidateFile(vt, env, level, level, compactionKindVirtualRewrite); pc != nil {
1719 1 : return pc
1720 1 : }
1721 : }
1722 : }
1723 :
1724 1 : return nil
1725 : }
1726 :
1727 : // pickBlobFileRewriteCompactionHighPriority picks a compaction that rewrites a
1728 : // blob file to reclaim disk space if the heuristics for high-priority blob file
1729 : // rewrites are met.
1730 : func (p *compactionPickerByScore) pickBlobFileRewriteCompactionHighPriority(
1731 : env compactionEnv,
1732 1 : ) (pc *pickedBlobFileCompaction) {
1733 1 : policy := p.opts.Experimental.ValueSeparationPolicy()
1734 1 : if policy.GarbageRatioHighPriority >= 1.0 {
1735 1 : // High-priority blob file rewrite compactions are disabled.
1736 1 : return nil
1737 1 : }
1738 1 : aggregateStats, heuristicStats := p.latestVersionState.blobFiles.Stats()
1739 1 : if heuristicStats.CountFilesEligible == 0 && heuristicStats.CountFilesTooRecent == 0 {
1740 1 : // No blob files with any garbage to rewrite.
1741 1 : return nil
1742 1 : }
1743 :
1744 1 : garbagePct := float64(aggregateStats.ValueSize-aggregateStats.ReferencedValueSize) /
1745 1 : float64(aggregateStats.ValueSize)
1746 1 : if garbagePct <= policy.GarbageRatioHighPriority {
1747 1 : // Not enough garbage to warrant a rewrite compaction.
1748 1 : return nil
1749 1 : }
1750 1 : pc = p.pickBlobFileRewriteCandidate(env)
1751 1 : if pc != nil {
1752 1 : pc.highPriority = true
1753 1 : }
1754 1 : return pc
1755 : }
1756 :
1757 : // pickBlobFileRewriteCompactionLowPriority picks a compaction that rewrites a
1758 : // blob file to reclaim disk space if the heuristics for low-priority blob file
1759 : // rewrites are met.
1760 : func (p *compactionPickerByScore) pickBlobFileRewriteCompactionLowPriority(
1761 : env compactionEnv,
1762 1 : ) (pc *pickedBlobFileCompaction) {
1763 1 : aggregateStats, heuristicStats := p.latestVersionState.blobFiles.Stats()
1764 1 : if heuristicStats.CountFilesEligible == 0 && heuristicStats.CountFilesTooRecent == 0 {
1765 1 : // No blob files with any garbage to rewrite.
1766 1 : return nil
1767 1 : }
1768 1 : policy := p.opts.Experimental.ValueSeparationPolicy()
1769 1 : if policy.GarbageRatioLowPriority >= 1.0 {
1770 1 : // Blob file rewrite compactions are disabled.
1771 1 : return nil
1772 1 : }
1773 :
1774 : // We want to use ReferencedValueSize here instead of ReferencedBackingValueSize
1775 : // to get the estimate of live data in blob files. We'll check below to see if there
1776 : // are actually any candidates with garbage to reclaim.
1777 1 : garbagePct := float64(aggregateStats.ValueSize-aggregateStats.ReferencedValueSize) /
1778 1 : float64(aggregateStats.ValueSize)
1779 1 : if garbagePct <= policy.GarbageRatioLowPriority {
1780 1 : // Not enough garbage to warrant a rewrite compaction.
1781 1 : return nil
1782 1 : }
1783 1 : return p.pickBlobFileRewriteCandidate(env)
1784 : }
1785 :
1786 : func (p *compactionPickerByScore) pickBlobFileRewriteCandidate(
1787 : env compactionEnv,
1788 1 : ) (pc *pickedBlobFileCompaction) {
1789 1 : // Check if there is an ongoing blob file rewrite compaction. If there is,
1790 1 : // don't schedule a new one.
1791 1 : for _, c := range env.inProgressCompactions {
1792 1 : if c.kind == compactionKindBlobFileRewrite {
1793 1 : return nil
1794 1 : }
1795 : }
1796 1 : candidate, ok := p.latestVersionState.blobFiles.ReplacementCandidate()
1797 1 : if !ok {
1798 1 : // None meet the heuristic.
1799 1 : return nil
1800 1 : }
1801 1 : return &pickedBlobFileCompaction{
1802 1 : vers: p.vers,
1803 1 : file: candidate,
1804 1 : referencingTables: p.latestVersionState.blobFiles.ReferencingTables(candidate.FileID),
1805 1 : }
1806 : }
1807 :
1808 : // pickTombstoneDensityCompaction looks for a compaction that eliminates
1809 : // regions of extremely high point tombstone density. For each level, it picks
1810 : // a file where the ratio of tombstone-dense blocks is at least
1811 : // options.Experimental.MinTombstoneDenseRatio, prioritizing compaction of
1812 : // files with higher ratios of tombstone-dense blocks.
1813 : func (p *compactionPickerByScore) pickTombstoneDensityCompaction(
1814 : env compactionEnv,
1815 1 : ) (pc *pickedTableCompaction) {
1816 1 : threshold := p.opts.Experimental.TombstoneDenseCompactionThreshold()
1817 1 : if threshold <= 0 {
1818 0 : // Tombstone density compactions are disabled.
1819 0 : return nil
1820 0 : }
1821 :
1822 1 : var candidate *manifest.TableMetadata
1823 1 : var candidateTombstoneDenseBlocksRatio float64
1824 1 : var level int
1825 1 : // If a candidate file has a very high overlapping ratio, point tombstones
1826 1 : // in it are likely sparse in keyspace even if the sstable itself is tombstone
1827 1 : // dense. These tombstones likely wouldn't be slow to iterate over, so we exclude
1828 1 : // these files from tombstone density compactions. The threshold of 40.0 is
1829 1 : // chosen somewhat arbitrarily, after some observations around excessively large
1830 1 : // tombstone density compactions.
1831 1 : const maxOverlappingRatio = 40.0
1832 1 : // NB: we don't consider the lowest level because elision-only compactions
1833 1 : // handle that case.
1834 1 : lastNonEmptyLevel := numLevels - 1
1835 1 : for l := numLevels - 2; l >= 0; l-- {
1836 1 : iter := p.vers.Levels[l].Iter()
1837 1 : for f := iter.First(); f != nil; f = iter.Next() {
1838 1 : if f.IsCompacting() || f.Size == 0 {
1839 1 : continue
1840 : }
1841 1 : props, propsValid := f.TableBacking.Properties()
1842 1 : if !propsValid {
1843 1 : continue
1844 : }
1845 1 : if props.TombstoneDenseBlocksRatio < threshold {
1846 1 : continue
1847 : }
1848 1 : overlaps := p.vers.Overlaps(lastNonEmptyLevel, f.UserKeyBounds())
1849 1 : if float64(overlaps.AggregateSizeSum())/float64(f.Size) > maxOverlappingRatio {
1850 1 : continue
1851 : }
1852 1 : if candidate == nil || candidateTombstoneDenseBlocksRatio < props.TombstoneDenseBlocksRatio {
1853 1 : candidate = f
1854 1 : candidateTombstoneDenseBlocksRatio = props.TombstoneDenseBlocksRatio
1855 1 : level = l
1856 1 : }
1857 : }
1858 : // We prefer lower level (ie. L5) candidates over higher level (ie. L4) ones.
1859 1 : if candidate != nil {
1860 1 : break
1861 : }
1862 1 : if !p.vers.Levels[l].Empty() {
1863 1 : lastNonEmptyLevel = l
1864 1 : }
1865 : }
1866 :
1867 1 : return p.pickedCompactionFromCandidateFile(candidate, env, level, defaultOutputLevel(level, p.baseLevel), compactionKindTombstoneDensity)
1868 : }
1869 :
1870 : // pickAutoLPositive picks an automatic compaction for the candidate
1871 : // file in a positive-numbered level. This function must not be used for
1872 : // L0.
1873 : func pickAutoLPositive(
1874 : env compactionEnv,
1875 : opts *Options,
1876 : vers *manifest.Version,
1877 : l0Organizer *manifest.L0Organizer,
1878 : cInfo candidateLevelInfo,
1879 : baseLevel int,
1880 1 : ) (pc *pickedTableCompaction) {
1881 1 : if cInfo.level == 0 {
1882 0 : panic("pebble: pickAutoLPositive called for L0")
1883 : }
1884 :
1885 1 : pc = newPickedTableCompaction(opts, vers, l0Organizer, cInfo.level, defaultOutputLevel(cInfo.level, baseLevel), baseLevel)
1886 1 : if pc.outputLevel.level != cInfo.outputLevel {
1887 0 : panic("pebble: compaction picked unexpected output level")
1888 : }
1889 1 : pc.startLevel.files = cInfo.file.Slice()
1890 1 :
1891 1 : if !pc.setupInputs(opts, env.diskAvailBytes, env.inProgressCompactions, pc.startLevel, env.problemSpans) {
1892 1 : return nil
1893 1 : }
1894 1 : return pc.maybeAddLevel(opts, env)
1895 : }
1896 :
1897 : // maybeAddLevel maybe adds a level to the picked compaction.
1898 : // Multilevel compactions are only allowed if the max compaction concurrency
1899 : // is greater than 1, and there are no in-progress multi-level compactions.
1900 : func (pc *pickedTableCompaction) maybeAddLevel(
1901 : opts *Options, env compactionEnv,
1902 1 : ) *pickedTableCompaction {
1903 1 : pc.pickerMetrics.singleLevelOverlappingRatio = pc.overlappingRatio()
1904 1 : if pc.outputLevel.level == numLevels-1 {
1905 1 : // Don't add a level if the current output level is in L6.
1906 1 : return pc
1907 1 : }
1908 : // We allow at most one in-progress multiLevel compaction at any time.
1909 1 : for _, c := range env.inProgressCompactions {
1910 1 : if len(c.inputs) > 2 {
1911 1 : return pc
1912 1 : }
1913 : }
1914 1 : _, upper := opts.CompactionConcurrencyRange()
1915 1 : if upper == 1 {
1916 1 : // If the maximum compaction concurrency is 1, avoid picking a multi-level compactions
1917 1 : // as they could block compactions from L0.
1918 1 : return pc
1919 1 : }
1920 1 : if !opts.Experimental.MultiLevelCompactionHeuristic().allowL0() && pc.startLevel.level == 0 {
1921 1 : return pc
1922 1 : }
1923 1 : targetFileSize := opts.TargetFileSize(pc.outputLevel.level, pc.baseLevel)
1924 1 : if pc.estimatedInputSize() > expandedCompactionByteSizeLimit(opts, targetFileSize, env.diskAvailBytes) {
1925 0 : // Don't add a level if the current compaction exceeds the compaction size limit
1926 0 : return pc
1927 0 : }
1928 1 : return opts.Experimental.MultiLevelCompactionHeuristic().pick(pc, opts, env)
1929 : }
1930 :
1931 : // MultiLevelHeuristic evaluates whether to add files from the next level into the compaction.
1932 : type MultiLevelHeuristic interface {
1933 : // Evaluate returns the preferred compaction.
1934 : pick(pc *pickedTableCompaction, opts *Options, env compactionEnv) *pickedTableCompaction
1935 :
1936 : // Returns if the heuristic allows L0 to be involved in ML compaction
1937 : allowL0() bool
1938 :
1939 : // String implements fmt.Stringer.
1940 : String() string
1941 : }
1942 :
1943 : // NoMultiLevel will never add an additional level to the compaction.
1944 : type NoMultiLevel struct{}
1945 :
1946 : var _ MultiLevelHeuristic = (*NoMultiLevel)(nil)
1947 :
1948 1 : func OptionNoMultiLevel() MultiLevelHeuristic {
1949 1 : return NoMultiLevel{}
1950 1 : }
1951 :
1952 : func (nml NoMultiLevel) pick(
1953 : pc *pickedTableCompaction, opts *Options, env compactionEnv,
1954 1 : ) *pickedTableCompaction {
1955 1 : return pc
1956 1 : }
1957 :
1958 1 : func (nml NoMultiLevel) allowL0() bool { return false }
1959 1 : func (nml NoMultiLevel) String() string { return "none" }
1960 :
1961 1 : func (pc *pickedTableCompaction) predictedWriteAmp() float64 {
1962 1 : var bytesToCompact uint64
1963 1 : var higherLevelBytes uint64
1964 1 : for i := range pc.inputs {
1965 1 : levelSize := pc.inputs[i].files.AggregateSizeSum()
1966 1 : bytesToCompact += levelSize
1967 1 : if i != len(pc.inputs)-1 {
1968 1 : higherLevelBytes += levelSize
1969 1 : }
1970 : }
1971 1 : return float64(bytesToCompact) / float64(higherLevelBytes)
1972 : }
1973 :
1974 1 : func (pc *pickedTableCompaction) overlappingRatio() float64 {
1975 1 : var higherLevelBytes uint64
1976 1 : var lowestLevelBytes uint64
1977 1 : for i := range pc.inputs {
1978 1 : levelSize := pc.inputs[i].files.AggregateSizeSum()
1979 1 : if i == len(pc.inputs)-1 {
1980 1 : lowestLevelBytes += levelSize
1981 1 : continue
1982 : }
1983 1 : higherLevelBytes += levelSize
1984 : }
1985 1 : return float64(lowestLevelBytes) / float64(higherLevelBytes)
1986 : }
1987 :
1988 : // WriteAmpHeuristic defines a multi level compaction heuristic which will add
1989 : // an additional level to the picked compaction if it reduces predicted write
1990 : // amp of the compaction + the addPropensity constant.
1991 : type WriteAmpHeuristic struct {
1992 : // addPropensity is a constant that affects the propensity to conduct multilevel
1993 : // compactions. If positive, a multilevel compaction may get picked even if
1994 : // the single level compaction has lower write amp, and vice versa.
1995 : AddPropensity float64
1996 :
1997 : // AllowL0 if true, allow l0 to be involved in a ML compaction.
1998 : AllowL0 bool
1999 : }
2000 :
2001 : var _ MultiLevelHeuristic = (*WriteAmpHeuristic)(nil)
2002 :
2003 : // Default write amp heuristic with no propensity towards multi-level
2004 : // and no multilevel compactions involving L0.
2005 : var defaultWriteAmpHeuristic = &WriteAmpHeuristic{}
2006 :
2007 1 : func OptionWriteAmpHeuristic() MultiLevelHeuristic {
2008 1 : return defaultWriteAmpHeuristic
2009 1 : }
2010 :
2011 : // TODO(msbutler): microbenchmark the extent to which multilevel compaction
2012 : // picking slows down the compaction picking process. This should be as fast as
2013 : // possible since Compaction-picking holds d.mu, which prevents WAL rotations,
2014 : // in-progress flushes and compactions from completing, etc. Consider ways to
2015 : // deduplicate work, given that setupInputs has already been called.
2016 : func (wa WriteAmpHeuristic) pick(
2017 : pcOrig *pickedTableCompaction, opts *Options, env compactionEnv,
2018 1 : ) *pickedTableCompaction {
2019 1 : pcMulti := pcOrig.clone()
2020 1 : if !pcMulti.setupMultiLevelCandidate(opts, env) {
2021 1 : return pcOrig
2022 1 : }
2023 : // We consider the addition of a level as an "expansion" of the compaction.
2024 : // If pcMulti is past the expanded compaction byte size limit already,
2025 : // we don't consider it.
2026 1 : targetFileSize := opts.TargetFileSize(pcMulti.outputLevel.level, pcMulti.baseLevel)
2027 1 : if pcMulti.estimatedInputSize() >= expandedCompactionByteSizeLimit(opts, targetFileSize, env.diskAvailBytes) {
2028 0 : return pcOrig
2029 0 : }
2030 1 : picked := pcOrig
2031 1 : if pcMulti.predictedWriteAmp() <= pcOrig.predictedWriteAmp()+wa.AddPropensity {
2032 1 : picked = pcMulti
2033 1 : }
2034 : // Regardless of what compaction was picked, log the multilevelOverlapping ratio.
2035 1 : picked.pickerMetrics.multiLevelOverlappingRatio = pcMulti.overlappingRatio()
2036 1 : return picked
2037 : }
2038 :
2039 1 : func (wa WriteAmpHeuristic) allowL0() bool {
2040 1 : return wa.AllowL0
2041 1 : }
2042 :
2043 : // String implements fmt.Stringer.
2044 1 : func (wa WriteAmpHeuristic) String() string {
2045 1 : return fmt.Sprintf("wamp(%.2f, %t)", wa.AddPropensity, wa.AllowL0)
2046 1 : }
2047 :
2048 : // Helper method to pick compactions originating from L0. Uses information about
2049 : // sublevels to generate a compaction.
2050 : func pickL0(
2051 : env compactionEnv,
2052 : opts *Options,
2053 : vers *manifest.Version,
2054 : l0Organizer *manifest.L0Organizer,
2055 : baseLevel int,
2056 1 : ) *pickedTableCompaction {
2057 1 : // It is important to pass information about Lbase files to L0Sublevels
2058 1 : // so it can pick a compaction that does not conflict with an Lbase => Lbase+1
2059 1 : // compaction. Without this, we observed reduced concurrency of L0=>Lbase
2060 1 : // compactions, and increasing read amplification in L0.
2061 1 : //
2062 1 : // TODO(bilal) Remove the minCompactionDepth parameter once fixing it at 1
2063 1 : // has been shown to not cause a performance regression.
2064 1 : lcf := l0Organizer.PickBaseCompaction(opts.Logger, 1, vers.Levels[baseLevel].Slice(), baseLevel, env.problemSpans)
2065 1 : if lcf != nil {
2066 1 : pc := newPickedCompactionFromL0(lcf, opts, vers, l0Organizer, baseLevel, true)
2067 1 : if pc.setupInputs(opts, env.diskAvailBytes, env.inProgressCompactions, pc.startLevel, env.problemSpans) {
2068 1 : if pc.startLevel.files.Empty() {
2069 0 : opts.Logger.Errorf("%v", base.AssertionFailedf("empty compaction chosen"))
2070 0 : }
2071 1 : return pc.maybeAddLevel(opts, env)
2072 : }
2073 : // TODO(radu): investigate why this happens.
2074 : // opts.Logger.Errorf("%v", base.AssertionFailedf("setupInputs failed"))
2075 : }
2076 :
2077 : // Couldn't choose a base compaction. Try choosing an intra-L0
2078 : // compaction. Note that we pass in L0CompactionThreshold here as opposed to
2079 : // 1, since choosing a single sublevel intra-L0 compaction is
2080 : // counterproductive.
2081 1 : lcf = l0Organizer.PickIntraL0Compaction(env.earliestUnflushedSeqNum, minIntraL0Count, env.problemSpans)
2082 1 : if lcf != nil {
2083 1 : pc := newPickedCompactionFromL0(lcf, opts, vers, l0Organizer, baseLevel, false)
2084 1 : if pc.setupInputs(opts, env.diskAvailBytes, env.inProgressCompactions, pc.startLevel, env.problemSpans) {
2085 1 : if pc.startLevel.files.Empty() {
2086 0 : opts.Logger.Fatalf("empty compaction chosen")
2087 0 : }
2088 : // A single-file intra-L0 compaction is unproductive.
2089 1 : if iter := pc.startLevel.files.Iter(); iter.First() != nil && iter.Next() != nil {
2090 1 : pc.bounds = manifest.KeyRange(opts.Comparer.Compare, pc.startLevel.files.All())
2091 1 : return pc
2092 1 : }
2093 0 : } else {
2094 0 : // TODO(radu): investigate why this happens.
2095 0 : // opts.Logger.Errorf("%v", base.AssertionFailedf("setupInputs failed"))
2096 0 : }
2097 : }
2098 1 : return nil
2099 : }
2100 :
2101 : func newPickedManualCompaction(
2102 : vers *manifest.Version,
2103 : l0Organizer *manifest.L0Organizer,
2104 : opts *Options,
2105 : env compactionEnv,
2106 : baseLevel int,
2107 : manual *manualCompaction,
2108 1 : ) (pc *pickedTableCompaction, retryLater bool) {
2109 1 : outputLevel := manual.level + 1
2110 1 : if manual.level == 0 {
2111 1 : outputLevel = baseLevel
2112 1 : } else if manual.level < baseLevel {
2113 1 : // The start level for a compaction must be >= Lbase. A manual
2114 1 : // compaction could have been created adhering to that condition, and
2115 1 : // then an automatic compaction came in and compacted all of the
2116 1 : // sstables in Lbase to Lbase+1 which caused Lbase to change. Simply
2117 1 : // ignore this manual compaction as there is nothing to do (manual.level
2118 1 : // points to an empty level).
2119 1 : return nil, false
2120 1 : }
2121 : // This conflictsWithInProgress call is necessary for the manual compaction to
2122 : // be retried when it conflicts with an ongoing automatic compaction. Without
2123 : // it, the compaction is dropped due to pc.setupInputs returning false since
2124 : // the input/output range is already being compacted, and the manual
2125 : // compaction ends with a non-compacted LSM.
2126 1 : if conflictsWithInProgress(manual, outputLevel, env.inProgressCompactions, opts.Comparer.Compare) {
2127 1 : return nil, true
2128 1 : }
2129 1 : pc = newPickedTableCompaction(opts, vers, l0Organizer, manual.level, defaultOutputLevel(manual.level, baseLevel), baseLevel)
2130 1 : pc.manualID = manual.id
2131 1 : manual.outputLevel = pc.outputLevel.level
2132 1 : pc.startLevel.files = vers.Overlaps(manual.level, base.UserKeyBoundsInclusive(manual.start, manual.end))
2133 1 : if pc.startLevel.files.Empty() {
2134 1 : // Nothing to do
2135 1 : return nil, false
2136 1 : }
2137 : // We use nil problemSpans because we don't want problem spans to prevent
2138 : // manual compactions.
2139 1 : if !pc.setupInputs(opts, env.diskAvailBytes, env.inProgressCompactions, pc.startLevel, nil /* problemSpans */) {
2140 1 : // setupInputs returned false indicating there's a conflicting
2141 1 : // concurrent compaction.
2142 1 : return nil, true
2143 1 : }
2144 1 : if pc = pc.maybeAddLevel(opts, env); pc == nil {
2145 0 : return nil, false
2146 0 : }
2147 1 : if pc.outputLevel.level != outputLevel {
2148 1 : if len(pc.inputs) > 2 {
2149 1 : // Multilevel compactions relax this invariant.
2150 1 : } else {
2151 0 : panic("pebble: compaction picked unexpected output level")
2152 : }
2153 : }
2154 1 : return pc, false
2155 : }
2156 :
2157 : // pickDownloadCompaction picks a download compaction for the downloadSpan,
2158 : // which could be specified as being performed either by a copy compaction of
2159 : // the backing file or a rewrite compaction.
2160 : func pickDownloadCompaction(
2161 : vers *manifest.Version,
2162 : l0Organizer *manifest.L0Organizer,
2163 : opts *Options,
2164 : env compactionEnv,
2165 : baseLevel int,
2166 : kind compactionKind,
2167 : level int,
2168 : file *manifest.TableMetadata,
2169 1 : ) (pc *pickedTableCompaction) {
2170 1 : // Check if the file is compacting already.
2171 1 : if file.CompactionState == manifest.CompactionStateCompacting {
2172 0 : return nil
2173 0 : }
2174 1 : if kind != compactionKindCopy && kind != compactionKindRewrite {
2175 0 : panic("invalid download/rewrite compaction kind")
2176 : }
2177 1 : pc = newPickedTableCompaction(opts, vers, l0Organizer, level, level, baseLevel)
2178 1 : pc.kind = kind
2179 1 : pc.startLevel.files = manifest.NewLevelSliceKeySorted(opts.Comparer.Compare, []*manifest.TableMetadata{file})
2180 1 : if !pc.setupInputs(opts, env.diskAvailBytes, env.inProgressCompactions, pc.startLevel, nil /* problemSpans */) {
2181 0 : // setupInputs returned false indicating there's a conflicting
2182 0 : // concurrent compaction.
2183 0 : return nil
2184 0 : }
2185 1 : if pc.outputLevel.level != level {
2186 0 : panic("pebble: download compaction picked unexpected output level")
2187 : }
2188 1 : return pc
2189 : }
2190 :
2191 : func (p *compactionPickerByScore) pickReadTriggeredCompaction(
2192 : env compactionEnv,
2193 1 : ) (pc *pickedTableCompaction) {
2194 1 : // If a flush is in-progress or expected to happen soon, it means more writes are taking place. We would
2195 1 : // soon be scheduling more write focussed compactions. In this case, skip read compactions as they are
2196 1 : // lower priority.
2197 1 : if env.readCompactionEnv.flushing || env.readCompactionEnv.readCompactions == nil {
2198 1 : return nil
2199 1 : }
2200 1 : for env.readCompactionEnv.readCompactions.size > 0 {
2201 0 : rc := env.readCompactionEnv.readCompactions.remove()
2202 0 : if pc = pickReadTriggeredCompactionHelper(p, rc, env); pc != nil {
2203 0 : break
2204 : }
2205 : }
2206 1 : return pc
2207 : }
2208 :
2209 : func pickReadTriggeredCompactionHelper(
2210 : p *compactionPickerByScore, rc *readCompaction, env compactionEnv,
2211 0 : ) (pc *pickedTableCompaction) {
2212 0 : overlapSlice := p.vers.Overlaps(rc.level, base.UserKeyBoundsInclusive(rc.start, rc.end))
2213 0 : var fileMatches bool
2214 0 : for f := range overlapSlice.All() {
2215 0 : if f.TableNum == rc.tableNum {
2216 0 : fileMatches = true
2217 0 : break
2218 : }
2219 : }
2220 0 : if !fileMatches {
2221 0 : return nil
2222 0 : }
2223 :
2224 0 : pc = newPickedTableCompaction(p.opts, p.vers, p.latestVersionState.l0Organizer,
2225 0 : rc.level, defaultOutputLevel(rc.level, p.baseLevel), p.baseLevel)
2226 0 :
2227 0 : pc.startLevel.files = overlapSlice
2228 0 : if !pc.setupInputs(p.opts, env.diskAvailBytes, env.inProgressCompactions, pc.startLevel, env.problemSpans) {
2229 0 : return nil
2230 0 : }
2231 0 : pc.kind = compactionKindRead
2232 0 :
2233 0 : if outputOverlaps := pc.version.Overlaps(pc.outputLevel.level, pc.bounds); !outputOverlaps.Empty() {
2234 0 : // Prevent read compactions which are too wide.
2235 0 : targetFileSize := p.opts.TargetFileSize(pc.outputLevel.level, pc.baseLevel)
2236 0 : if outputOverlaps.AggregateSizeSum() > maxReadCompactionBytes(targetFileSize) {
2237 0 : return nil
2238 0 : }
2239 :
2240 : // Prevent compactions which start with a small seed file of size X, but
2241 : // overlap with files in the output layer that have total size larger than
2242 : // allowedCompactionWidth * X.
2243 0 : const allowedCompactionWidth = 35
2244 0 : if outputOverlaps.AggregateSizeSum() > overlapSlice.AggregateSizeSum()*allowedCompactionWidth {
2245 0 : return nil
2246 0 : }
2247 : }
2248 :
2249 0 : return pc
2250 : }
2251 :
2252 0 : func (p *compactionPickerByScore) forceBaseLevel1() {
2253 0 : p.baseLevel = 1
2254 0 : }
2255 :
2256 : // outputKeyRangeAlreadyCompacting checks if the input range of the picked
2257 : // compaction is already being written to by an in-progress compaction.
2258 : func outputKeyRangeAlreadyCompacting(
2259 : cmp base.Compare, inProgressCompactions []compactionInfo, pc *pickedTableCompaction,
2260 1 : ) bool {
2261 1 : // Look for active compactions outputting to the same region of the key
2262 1 : // space in the same output level. Two potential compactions may conflict
2263 1 : // without sharing input files if there are no files in the output level
2264 1 : // that overlap with the intersection of the compactions' key spaces.
2265 1 : //
2266 1 : // Consider an active L0->Lbase compaction compacting two L0 files one
2267 1 : // [a-f] and the other [t-z] into Lbase.
2268 1 : //
2269 1 : // L0
2270 1 : // ↦ 000100 ↤ ↦ 000101 ↤
2271 1 : // L1
2272 1 : // ↦ 000004 ↤
2273 1 : // a b c d e f g h i j k l m n o p q r s t u v w x y z
2274 1 : //
2275 1 : // If a new file 000102 [j-p] is flushed while the existing compaction is
2276 1 : // still ongoing, new file would not be in any compacting sublevel
2277 1 : // intervals and would not overlap with any Lbase files that are also
2278 1 : // compacting. However, this compaction cannot be picked because the
2279 1 : // compaction's output key space [j-p] would overlap the existing
2280 1 : // compaction's output key space [a-z].
2281 1 : //
2282 1 : // L0
2283 1 : // ↦ 000100* ↤ ↦ 000102 ↤ ↦ 000101* ↤
2284 1 : // L1
2285 1 : // ↦ 000004* ↤
2286 1 : // a b c d e f g h i j k l m n o p q r s t u v w x y z
2287 1 : //
2288 1 : // * - currently compacting
2289 1 : if pc.outputLevel != nil && pc.outputLevel.level != 0 {
2290 1 : for _, c := range inProgressCompactions {
2291 1 : if pc.outputLevel.level != c.outputLevel {
2292 1 : continue
2293 : }
2294 1 : if !c.bounds.Overlaps(cmp, &pc.bounds) {
2295 1 : continue
2296 : }
2297 : // The picked compaction and the in-progress compaction c are
2298 : // outputting to the same region of the key space of the same
2299 : // level.
2300 1 : return true
2301 : }
2302 : }
2303 1 : return false
2304 : }
2305 :
2306 : // conflictsWithInProgress checks if there are any in-progress compactions with overlapping keyspace.
2307 : func conflictsWithInProgress(
2308 : manual *manualCompaction, outputLevel int, inProgressCompactions []compactionInfo, cmp Compare,
2309 1 : ) bool {
2310 1 : for _, c := range inProgressCompactions {
2311 1 : if (c.outputLevel == manual.level || c.outputLevel == outputLevel) &&
2312 1 : areUserKeysOverlapping(manual.start, manual.end, c.bounds.Start, c.bounds.End.Key, cmp) {
2313 1 : return true
2314 1 : }
2315 1 : for _, in := range c.inputs {
2316 1 : if in.files.Empty() {
2317 1 : continue
2318 : }
2319 1 : iter := in.files.Iter()
2320 1 : smallest := iter.First().Smallest().UserKey
2321 1 : largest := iter.Last().Largest().UserKey
2322 1 : if (in.level == manual.level || in.level == outputLevel) &&
2323 1 : areUserKeysOverlapping(manual.start, manual.end, smallest, largest, cmp) {
2324 1 : return true
2325 1 : }
2326 : }
2327 : }
2328 1 : return false
2329 : }
2330 :
2331 1 : func areUserKeysOverlapping(x1, x2, y1, y2 []byte, cmp Compare) bool {
2332 1 : return cmp(x1, y2) <= 0 && cmp(y1, x2) <= 0
2333 1 : }
|
