-
Notifications
You must be signed in to change notification settings - Fork 563
Expand file tree
/
Copy pathcompaction.go
More file actions
3164 lines (2946 loc) · 116 KB
/
compaction.go
File metadata and controls
3164 lines (2946 loc) · 116 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
// Copyright 2013 The LevelDB-Go and Pebble Authors. All rights reserved. Use
// of this source code is governed by a BSD-style license that can be found in
// the LICENSE file.
package pebble
import (
"bytes"
"context"
"fmt"
"iter"
"math"
"runtime/pprof"
"slices"
"sync/atomic"
"time"
"unsafe"
"github.com/cockroachdb/crlib/crtime"
"github.com/cockroachdb/errors"
"github.com/cockroachdb/pebble/internal/base"
"github.com/cockroachdb/pebble/internal/compact"
"github.com/cockroachdb/pebble/internal/keyspan"
"github.com/cockroachdb/pebble/internal/keyspan/keyspanimpl"
"github.com/cockroachdb/pebble/internal/manifest"
"github.com/cockroachdb/pebble/internal/problemspans"
"github.com/cockroachdb/pebble/internal/sstableinternal"
"github.com/cockroachdb/pebble/internal/tombspan"
"github.com/cockroachdb/pebble/objstorage"
"github.com/cockroachdb/pebble/objstorage/objstorageprovider/objiotracing"
"github.com/cockroachdb/pebble/sstable"
"github.com/cockroachdb/pebble/sstable/blob"
"github.com/cockroachdb/pebble/sstable/block"
"github.com/cockroachdb/pebble/sstable/block/blockkind"
"github.com/cockroachdb/pebble/valsep"
"github.com/cockroachdb/pebble/vfs"
"github.com/cockroachdb/redact"
)
var errEmptyTable = errors.New("pebble: empty table")
// ErrCancelledCompaction is returned if a compaction is cancelled by a
// concurrent excise or ingest-split operation.
var ErrCancelledCompaction = errors.New("pebble: compaction cancelled by a concurrent operation, will retry compaction")
var flushLabels = pprof.Labels("pebble", "flush", "output-level", "L0")
// expandedCompactionByteSizeLimit is the maximum number of bytes in all
// compacted files. We avoid expanding the lower level file set of a compaction
// if it would make the total compaction cover more than this many bytes.
func expandedCompactionByteSizeLimit(
opts *Options, targetFileSize int64, availBytes uint64,
) uint64 {
v := uint64(25 * targetFileSize)
// Never expand a compaction beyond half the available capacity, divided
// by the maximum number of concurrent compactions. Each of the concurrent
// compactions may expand up to this limit, so this attempts to limit
// compactions to half of available disk space. Note that this will not
// prevent compaction picking from pursuing compactions that are larger
// than this threshold before expansion.
//
// NB: this heuristic is an approximation since we may run more compactions
// than the upper concurrency limit.
_, maxConcurrency := opts.CompactionConcurrencyRange()
diskMax := (availBytes / 2) / uint64(maxConcurrency)
if v > diskMax {
v = diskMax
}
return v
}
// maxGrandparentOverlapBytes is the maximum bytes of overlap with level+1
// before we stop building a single file in a level-1 to level compaction.
func maxGrandparentOverlapBytes(targetFileSize int64) uint64 {
return uint64(10 * targetFileSize)
}
// maxReadCompactionBytes is used to prevent read compactions which
// are too wide.
func maxReadCompactionBytes(targetFileSize int64) uint64 {
return uint64(10 * targetFileSize)
}
// noCloseIter wraps around a FragmentIterator, intercepting and eliding
// calls to Close. It is used during compaction to ensure that rangeDelIters
// are not closed prematurely.
type noCloseIter struct {
keyspan.FragmentIterator
}
func (i *noCloseIter) Close() {}
type compactionLevel struct {
level int
files manifest.LevelSlice
// l0SublevelInfo contains information about L0 sublevels being compacted.
// It's only set for the start level of a compaction starting out of L0 and
// is nil for all other compactions.
l0SublevelInfo []sublevelInfo
}
func (cl compactionLevel) Clone() compactionLevel {
newCL := compactionLevel{
level: cl.level,
files: cl.files,
}
return newCL
}
func (cl compactionLevel) String() string {
return fmt.Sprintf(`Level %d, Files %s`, cl.level, cl.files)
}
// compactionWritable is a objstorage.Writable wrapper that, on every write,
// updates a metric in `versions` on bytes written by in-progress compactions so
// far. It also increments a per-compaction `written` atomic int.
type compactionWritable struct {
objstorage.Writable
versions *versionSet
written *atomic.Int64
}
// Write is part of the objstorage.Writable interface.
func (c *compactionWritable) Write(p []byte) error {
if err := c.Writable.Write(p); err != nil {
return err
}
c.written.Add(int64(len(p)))
c.versions.incrementCompactionBytes(int64(len(p)))
return nil
}
type compactionKind int
const (
compactionKindDefault compactionKind = iota
compactionKindFlush
// compactionKindMove denotes a move compaction where the input file is
// retained and linked in a new level without being obsoleted.
compactionKindMove
// compactionKindCopy denotes a copy compaction where the input file is
// copied byte-by-byte into a new file with a new TableNum in the output level.
compactionKindCopy
// compactionKindDeleteOnly denotes a compaction that only deletes input
// files. It can occur when wide range tombstones completely contain sstables.
compactionKindDeleteOnly
compactionKindElisionOnly
compactionKindRead
compactionKindTombstoneDensity
compactionKindRewrite
compactionKindIngestedFlushable
compactionKindBlobFileRewrite
// compactionKindVirtualRewrite must be the last compactionKind.
// If a new kind has to be added after VirtualRewrite,
// update AllCompactionKindStrings() accordingly.
compactionKindVirtualRewrite
)
func (k compactionKind) String() string {
switch k {
case compactionKindDefault:
return "default"
case compactionKindFlush:
return "flush"
case compactionKindMove:
return "move"
case compactionKindDeleteOnly:
return "delete-only"
case compactionKindElisionOnly:
return "elision-only"
case compactionKindRead:
return "read"
case compactionKindTombstoneDensity:
return "tombstone-density"
case compactionKindRewrite:
return "rewrite"
case compactionKindIngestedFlushable:
return "ingested-flushable"
case compactionKindCopy:
return "copy"
case compactionKindBlobFileRewrite:
return "blob-file-rewrite"
case compactionKindVirtualRewrite:
return "virtual-sst-rewrite"
}
return "?"
}
func (k compactionKind) SafeFormat(w redact.SafePrinter, _ rune) {
w.Print(redact.SafeString(k.String()))
}
// compactingOrFlushing returns "flushing" if the compaction kind is a flush,
// otherwise it returns "compacting".
func (k compactionKind) compactingOrFlushing() string {
if k == compactionKindFlush {
return "flushing"
}
return "compacting"
}
// AllCompactionKindStrings returns all compaction kind string representations
// for testing purposes. Used by tool/logs/compaction_test.go to verify the
// compaction summary tool stays in sync with new compaction types.
//
// NOTE: This function iterates up to compactionKindVirtualRewrite. If a new
// compactionKind is added after VirtualRewrite, update this function accordingly.
func AllCompactionKindStrings() map[string]bool {
kinds := make(map[string]bool)
for k := compactionKindDefault; k <= compactionKindVirtualRewrite; k++ {
kinds[k.String()] = true
}
return kinds
}
type compaction interface {
AddInProgressLocked(*DB)
BeganAt() time.Time
Bounds() *base.UserKeyBounds
Cancel()
Execute(JobID, *DB) error
GrantHandle() CompactionGrantHandle
Info() compactionInfo
IsDownload() bool
IsFlush() bool
PprofLabels(UserKeyCategories) pprof.LabelSet
RecordError(*problemspans.ByLevel, error)
Tables() iter.Seq2[int, *manifest.TableMetadata]
UsesBurstConcurrency() bool
VersionEditApplied() bool
}
// tableCompaction is a table compaction from one level to the next, starting
// from a given version. It implements the compaction interface.
type tableCompaction struct {
ctx context.Context
// cancel is a bool that can be used by other goroutines to signal a compaction
// to cancel, such as if a conflicting excise operation raced it to manifest
// application. Only holders of the manifest lock will write to this atomic.
cancel atomic.Bool
// kind indicates the kind of compaction. Different compaction kinds have
// different semantics and mechanics. Some may have additional fields.
kind compactionKind
// isDownload is true if this compaction was started as part of a Download
// operation. In this case kind is compactionKindCopy or
// compactionKindRewrite.
isDownload bool
comparer *base.Comparer
logger Logger
version *manifest.Version
// versionEditApplied is set to true when a compaction has completed and the
// resulting version has been installed (if successful), but the compaction
// goroutine is still cleaning up (eg, deleting obsolete files).
versionEditApplied bool
// getValueSeparation constructs a valsep.ValueSeparation for use in a
// compaction. It implements heuristics around choosing whether a compaction
// should:
//
// a) preserve existing blob references: The compaction does not write any
// new blob files, but propagates existing references to blob files.This
// conserves write bandwidth by avoiding rewriting the referenced values. It
// also reduces the locality of the referenced values which can reduce scan
// performance because a scan must load values from more unique blob files.
// It can also delay reclamation of disk space if some of the references to
// blob values are elided by the compaction, increasing space amplification.
//
// b) rewrite blob files: The compaction will write eligible values to new
// blob files. This consumes more write bandwidth because all values are
// rewritten. However it restores locality.
getValueSeparation func(JobID, *tableCompaction) valsep.ValueSeparation
// startLevel is the level that is being compacted. Inputs from startLevel
// and outputLevel will be merged to produce a set of outputLevel files.
startLevel *compactionLevel
// outputLevel is the level that files are being produced in. outputLevel is
// equal to startLevel+1 except when:
// - if startLevel is 0, the output level equals compactionPicker.baseLevel().
// - in multilevel compaction, the output level is the lowest level involved in
// the compaction
// A compaction's outputLevel is nil for delete-only compactions.
outputLevel *compactionLevel
// extraLevels point to additional levels in between the input and output
// levels that get compacted in multilevel compactions
extraLevels []*compactionLevel
inputs []compactionLevel
// eventualOutputLevel is normally outputLevel.level, unless
// outputLevel.level+1 has no overlap with the compaction bounds (in which
// case it is the bottom-most consecutive level with no such overlap).
//
// Because of move compactions, we know that any sstables produced by this
// compaction will be later moved to eventualOutputLevel. So we use
// eventualOutputLevel when determining the target file size, compression
// options, etc.
eventualOutputLevel int
// maxOutputFileSize is the maximum size of an individual table created
// during compaction.
maxOutputFileSize uint64
// maxOverlapBytes is the maximum number of bytes of overlap allowed for a
// single output table with the tables in the grandparent level.
maxOverlapBytes uint64
// The boundaries of the input data.
bounds base.UserKeyBounds
// grandparents are the tables in eventualOutputLevel+2 that overlap with the
// files being compacted. Used to determine output table boundaries. Do not
// assume that the actual files in the grandparent when this compaction
// finishes will be the same.
grandparents manifest.LevelSlice
delElision compact.TombstoneElision
rangeKeyElision compact.TombstoneElision
// deleteOnly contains information specific to compactions with kind
// compactionKindDeleteOnly. A delete-only compaction is a special
// compaction that does not merge or write sstables. Instead, it only
// performs deletions either through removing whole sstables from the LSM or
// virtualizing them into virtual sstables.
deleteOnly tombspan.DeleteOnlyCompaction
// flush contains information specific to flushes (compactionKindFlush and
// compactionKindIngestedFlushable). A flush is modeled by a compaction
// because it has similar mechanics to a default compaction.
flush struct {
// flushables contains the flushables (aka memtables, large batches,
// flushable ingestions, etc) that are being flushed.
flushables flushableList
// Boundaries at which sstables flushed to L0 should be split.
// Determined by L0Sublevels. If nil, ignored.
l0Limits [][]byte
}
// iterationState contains state used during compaction iteration.
iterationState struct {
// bufferPool is a pool of buffers used when reading blocks. Compactions
// do not populate the block cache under the assumption that the blocks
// we read will soon be irrelevant when their containing sstables are
// removed from the LSM.
bufferPool sstable.BufferPool
// keyspanIterClosers is a list of fragment iterators to close when the
// compaction finishes. As iteration opens new keyspan iterators,
// elements are appended. Keyspan iterators must remain open for the
// lifetime of the compaction, so they're accumulated here. When the
// compaction finishes, all the underlying keyspan iterators are closed.
keyspanIterClosers []*noCloseIter
// valueFetcher is used to fetch values from blob files. It's propagated
// down the iterator tree through the internal iterator options.
valueFetcher blob.ValueFetcher
}
// metrics encapsulates various metrics collected during a compaction.
metrics compactionMetrics
grantHandle CompactionGrantHandle
objCreateOpts objstorage.CreateOptions
annotations []string
}
// Assert that tableCompaction implements the compaction interface.
var _ compaction = (*tableCompaction)(nil)
func (c *tableCompaction) AddInProgressLocked(d *DB) {
d.mu.compact.inProgress[c] = struct{}{}
if c.UsesBurstConcurrency() {
d.mu.compact.burstConcurrency.Add(1)
}
var isBase, isIntraL0 bool
for _, cl := range c.inputs {
for f := range cl.files.All() {
if f.IsCompacting() {
d.opts.Logger.Fatalf("L%d->L%d: %s already being compacted", c.startLevel.level, c.outputLevel.level, f.TableNum)
}
f.SetCompactionState(manifest.CompactionStateCompacting)
if c.startLevel != nil && c.outputLevel != nil && c.startLevel.level == 0 {
if c.outputLevel.level == 0 {
f.IsIntraL0Compacting = true
isIntraL0 = true
} else {
isBase = true
}
}
}
}
if isIntraL0 || isBase {
l0Inputs := []manifest.LevelSlice{c.startLevel.files}
if isIntraL0 {
l0Inputs = append(l0Inputs, c.outputLevel.files)
}
if err := d.mu.versions.latest.l0Organizer.UpdateStateForStartedCompaction(l0Inputs, isBase); err != nil {
d.opts.Logger.Fatalf("could not update state for compaction: %s", err)
}
}
}
func (c *tableCompaction) BeganAt() time.Time { return c.metrics.beganAt }
func (c *tableCompaction) Bounds() *base.UserKeyBounds { return &c.bounds }
func (c *tableCompaction) Cancel() { c.cancel.Store(true) }
func (c *tableCompaction) Execute(jobID JobID, d *DB) error {
c.grantHandle.Started()
err := d.compact1(jobID, c)
// The version stored in the compaction is ref'd when the compaction is
// created. We're responsible for un-refing it when the compaction is
// complete.
if c.version != nil {
c.version.UnrefLocked()
}
return err
}
func (c *tableCompaction) RecordError(problemSpans *problemspans.ByLevel, err error) {
// Record problem spans for a short duration, unless the error is a
// corruption.
expiration := 30 * time.Second
if IsCorruptionError(err) {
// TODO(radu): ideally, we should be using the corruption reporting
// mechanism which has a tighter span for the corruption. We would need to
// somehow plumb the level of the file.
expiration = 5 * time.Minute
}
for i := range c.inputs {
level := c.inputs[i].level
if level == 0 {
// We do not set problem spans on L0, as they could block flushes.
continue
}
it := c.inputs[i].files.Iter()
for f := it.First(); f != nil; f = it.Next() {
problemSpans.Add(level, f.UserKeyBounds(), expiration)
}
}
}
func (c *tableCompaction) GrantHandle() CompactionGrantHandle { return c.grantHandle }
func (c *tableCompaction) IsDownload() bool { return c.isDownload }
func (c *tableCompaction) IsFlush() bool { return len(c.flush.flushables) > 0 }
func (c *tableCompaction) Info() compactionInfo {
info := compactionInfo{
versionEditApplied: c.versionEditApplied,
kind: c.kind,
inputs: c.inputs,
bounds: &c.bounds,
outputLevel: -1,
}
if c.outputLevel != nil {
info.outputLevel = c.outputLevel.level
}
return info
}
func (c *tableCompaction) PprofLabels(kc UserKeyCategories) pprof.LabelSet {
activity := "compact"
if len(c.flush.flushables) != 0 {
activity = "flush"
}
level := "L?"
// Delete-only compactions don't have an output level.
if c.outputLevel != nil {
level = fmt.Sprintf("L%d", c.outputLevel.level)
}
if kc.Len() > 0 {
cat := kc.CategorizeKeyRange(c.bounds.Start, c.bounds.End.Key)
return pprof.Labels("pebble", activity, "output-level", level, "key-type", cat)
}
return pprof.Labels("pebble", activity, "output-level", level)
}
func (c *tableCompaction) Tables() iter.Seq2[int, *manifest.TableMetadata] {
return func(yield func(int, *manifest.TableMetadata) bool) {
for _, cl := range c.inputs {
for f := range cl.files.All() {
if !yield(cl.level, f) {
return
}
}
}
}
}
func (c *tableCompaction) UsesBurstConcurrency() bool { return false }
func (c *tableCompaction) VersionEditApplied() bool { return c.versionEditApplied }
// compactionMetrics contians metrics surrounding a compaction.
type compactionMetrics struct {
// beganAt is the time when the compaction began.
beganAt time.Time
// bytesWritten contains the number of bytes that have been written to
// outputs. It's updated whenever the compaction outputs'
// objstorage.Writables receive new writes. See newCompactionOutputObj.
bytesWritten atomic.Int64
// internalIterStats contains statistics from the internal iterators used by
// the compaction.
//
// TODO(jackson): Use these to power the compaction BytesRead metric.
internalIterStats base.InternalIteratorStats
// perLevel contains metrics for each level involved in the compaction.
perLevel levelMetricsDelta
// picker contains metrics from the compaction picker when the compaction
// was picked.
picker pickedCompactionMetrics
}
// inputLargestSeqNumAbsolute returns the maximum LargestSeqNumAbsolute of any
// input sstables.
func (c *tableCompaction) inputLargestSeqNumAbsolute() base.SeqNum {
var seqNum base.SeqNum
for _, cl := range c.inputs {
for m := range cl.files.All() {
seqNum = max(seqNum, m.LargestSeqNumAbsolute)
}
}
return seqNum
}
func (c *tableCompaction) makeInfo(jobID JobID) CompactionInfo {
info := CompactionInfo{
JobID: int(jobID),
Reason: c.kind.String(),
Input: make([]LevelInfo, 0, len(c.inputs)),
Annotations: []string{},
}
if c.isDownload {
info.Reason = "download," + info.Reason
}
for _, cl := range c.inputs {
inputInfo := LevelInfo{Level: cl.level, Tables: nil}
for m := range cl.files.All() {
inputInfo.Tables = append(inputInfo.Tables, m.TableInfo())
}
info.Input = append(info.Input, inputInfo)
}
if c.outputLevel != nil {
info.Output.Level = c.outputLevel.level
// If there are no inputs from the output level (eg, a move
// compaction), add an empty LevelInfo to info.Input.
if len(c.inputs) > 0 && c.inputs[len(c.inputs)-1].level != c.outputLevel.level {
info.Input = append(info.Input, LevelInfo{Level: c.outputLevel.level})
}
} else {
// For a delete-only compaction, set the output level to L6. The
// output level is not meaningful here, but complicating the
// info.Output interface with a pointer doesn't seem worth the
// semantic distinction.
info.Output.Level = numLevels - 1
}
for i, score := range c.metrics.picker.scores {
info.Input[i].Score = score
}
info.SingleLevelOverlappingRatio = c.metrics.picker.singleLevelOverlappingRatio
info.MultiLevelOverlappingRatio = c.metrics.picker.multiLevelOverlappingRatio
if len(info.Input) > 2 {
info.Annotations = append(info.Annotations, "multilevel")
}
return info
}
type getValueSeparation func(JobID, *tableCompaction) valsep.ValueSeparation
// newCompaction constructs a compaction from the provided picked compaction.
//
// The compaction is created with a reference to its version that must be
// released when the compaction is complete.
func newCompaction(
ctx context.Context,
pc *pickedTableCompaction,
opts *Options,
beganAt time.Time,
provider objstorage.Provider,
grantHandle CompactionGrantHandle,
preferSharedStorage bool,
getValueSeparation getValueSeparation,
) *tableCompaction {
c := &tableCompaction{
ctx: ctx,
kind: compactionKindDefault,
comparer: opts.Comparer,
inputs: pc.inputs,
bounds: pc.bounds,
logger: opts.Logger,
version: pc.version,
getValueSeparation: getValueSeparation,
metrics: compactionMetrics{
beganAt: beganAt,
picker: pc.pickerMetrics,
},
grantHandle: grantHandle,
}
// Determine eventual output level.
c.eventualOutputLevel = pc.outputLevel.level
// TODO(radu): for intra-L0 compactions, we could check if the compaction
// includes all L0 files within the bounds.
if pc.outputLevel.level != 0 {
for c.eventualOutputLevel < manifest.NumLevels-1 && !c.version.HasOverlap(c.eventualOutputLevel+1, c.bounds) {
// All output tables are guaranteed to be moved down.
c.eventualOutputLevel++
}
}
targetFileSize := opts.TargetFileSize(c.eventualOutputLevel, pc.baseLevel)
c.maxOutputFileSize = uint64(targetFileSize)
c.maxOverlapBytes = maxGrandparentOverlapBytes(targetFileSize)
// Acquire a reference to the version to ensure that files and in-memory
// version state necessary for reading files remain available. Ignoring
// excises, this isn't strictly necessary for reading the sstables that are
// inputs to the compaction because those files are 'marked as compacting'
// and shouldn't be subject to any competing compactions. However with
// excises, a concurrent excise may remove a compaction's file from the
// Version and then cancel the compaction. The file shouldn't be physically
// removed until the cancelled compaction stops reading it.
//
// Additionally, we need any blob files referenced by input sstables to
// remain available, even if the blob file is rewritten. Maintaining a
// reference ensures that all these files remain available for the
// compaction's reads.
c.version.Ref()
c.startLevel = &c.inputs[0]
if pc.startLevel.l0SublevelInfo != nil {
c.startLevel.l0SublevelInfo = pc.startLevel.l0SublevelInfo
}
c.outputLevel = &c.inputs[len(c.inputs)-1]
if len(pc.inputs) > 2 {
// TODO(xinhaoz): Look into removing extraLevels on the compaction struct.
c.extraLevels = make([]*compactionLevel, 0, len(pc.inputs)-2)
for i := 1; i < len(pc.inputs)-1; i++ {
c.extraLevels = append(c.extraLevels, &c.inputs[i])
}
}
// Compute the set of outputLevel+1 files that overlap this compaction (these
// are the grandparent sstables).
if c.eventualOutputLevel < manifest.NumLevels-1 {
c.grandparents = c.version.Overlaps(max(c.eventualOutputLevel+1, pc.baseLevel), c.bounds)
}
c.delElision, c.rangeKeyElision = compact.SetupTombstoneElision(
c.comparer.Compare, c.version, pc.l0Organizer, c.outputLevel.level, c.bounds,
)
c.kind = pc.kind
c.maybeSwitchToMoveOrCopy(preferSharedStorage, provider)
c.objCreateOpts = objstorage.CreateOptions{
PreferSharedStorage: preferSharedStorage,
WriteCategory: getDiskWriteCategoryForCompaction(opts, c.kind),
}
if preferSharedStorage {
c.getValueSeparation = neverSeparateValues
}
return c
}
// maybeSwitchToMoveOrCopy decides if the compaction can be changed into a move
// or copy compaction, in which case c.kind is updated.
func (c *tableCompaction) maybeSwitchToMoveOrCopy(
preferSharedStorage bool, provider objstorage.Provider,
) {
// Only non-multi-level compactions with a single input file can be
// considered.
if c.startLevel.files.Len() != 1 || !c.outputLevel.files.Empty() || c.hasExtraLevelData() {
return
}
// In addition to the default compaction, we also check whether a tombstone
// density compaction can be optimized into a move compaction. However, we
// want to avoid performing a move compaction into the lowest level, since the
// goal there is to actually remove the tombstones.
//
// Tombstone density compaction is meant to address cases where tombstones
// don't reclaim much space but are still expensive to scan over. We can only
// remove the tombstones once there's nothing at all underneath them.
switch c.kind {
case compactionKindDefault:
// Proceed.
case compactionKindTombstoneDensity:
// Tombstone density compaction can be optimized into a move compaction.
// However, we want to avoid performing a move compaction into the lowest
// level, since the goal there is to actually remove the tombstones; even if
// they don't prevent a lot of space from being reclaimed, tombstones can
// still be expensive to scan over.
if c.outputLevel.level == numLevels-1 {
return
}
default:
// Other compaction kinds not supported.
return
}
// We avoid a move or copy if there is lots of overlapping grandparent data.
// Otherwise, the move could create a parent file that will require a very
// expensive merge later on.
//
// Note that if eventualOutputLevel != outputLevel, there are no
// "grandparents" on the output level.
if c.eventualOutputLevel == c.outputLevel.level && c.grandparents.AggregateSizeSum() > c.maxOverlapBytes {
return
}
iter := c.startLevel.files.Iter()
meta := iter.First()
// We should always be passed a provider, except in some unit tests.
isRemote := provider != nil && !objstorage.IsLocalTable(provider, meta.TableBacking.DiskFileNum)
// Shared and external tables can always be moved. We can also move a local
// table unless we need the result to be on shared storage.
if isRemote || !preferSharedStorage {
c.kind = compactionKindMove
return
}
// We can rewrite the table (regular compaction) or we can use a copy compaction.
switch {
case meta.Virtual:
// We want to avoid a copy compaction if the table is virtual, as we may end
// up copying a lot more data than necessary.
case meta.BlobReferenceDepth != 0:
// We also want to avoid copy compactions for tables with blob references,
// as we currently lack a mechanism to propagate blob references along with
// the sstable.
default:
c.kind = compactionKindCopy
}
}
func adjustGrandparentOverlapBytesForFlush(c *tableCompaction, flushingBytes uint64) {
// Heuristic to place a lower bound on compaction output file size
// caused by Lbase. Prior to this heuristic we have observed an L0 in
// production with 310K files of which 290K files were < 10KB in size.
// Our hypothesis is that it was caused by L1 having 2600 files and
// ~10GB, such that each flush got split into many tiny files due to
// overlapping with most of the files in Lbase.
//
// The computation below is general in that it accounts
// for flushing different volumes of data (e.g. we may be flushing
// many memtables). For illustration, we consider the typical
// example of flushing a 64MB memtable. So 12.8MB output,
// based on the compression guess below. If the compressed bytes
// guess is an over-estimate we will end up with smaller files,
// and if an under-estimate we will end up with larger files.
// With a 2MB target file size, 7 files. We are willing to accept
// 4x the number of files, if it results in better write amplification
// when later compacting to Lbase, i.e., ~450KB files (target file
// size / 4).
//
// Note that this is a pessimistic heuristic in that
// fileCountUpperBoundDueToGrandparents could be far from the actual
// number of files produced due to the grandparent limits. For
// example, in the extreme, consider a flush that overlaps with 1000
// files in Lbase f0...f999, and the initially calculated value of
// maxOverlapBytes will cause splits at f10, f20,..., f990, which
// means an upper bound file count of 100 files. Say the input bytes
// in the flush are such that acceptableFileCount=10. We will fatten
// up maxOverlapBytes by 10x to ensure that the upper bound file count
// drops to 10. However, it is possible that in practice, even without
// this change, we would have produced no more than 10 files, and that
// this change makes the files unnecessarily wide. Say the input bytes
// are distributed such that 10% are in f0...f9, 10% in f10...f19, ...
// 10% in f80...f89 and 10% in f990...f999. The original value of
// maxOverlapBytes would have actually produced only 10 sstables. But
// by increasing maxOverlapBytes by 10x, we may produce 1 sstable that
// spans f0...f89, i.e., a much wider sstable than necessary.
//
// We could produce a tighter estimate of
// fileCountUpperBoundDueToGrandparents if we had knowledge of the key
// distribution of the flush. The 4x multiplier mentioned earlier is
// a way to try to compensate for this pessimism.
//
// TODO(sumeer): we don't have compression info for the data being
// flushed, but it is likely that existing files that overlap with
// this flush in Lbase are representative wrt compression ratio. We
// could store the uncompressed size in TableMetadata and estimate
// the compression ratio.
const approxCompressionRatio = 0.2
approxOutputBytes := approxCompressionRatio * float64(flushingBytes)
approxNumFilesBasedOnTargetSize :=
int(math.Ceil(approxOutputBytes / float64(c.maxOutputFileSize)))
acceptableFileCount := float64(4 * approxNumFilesBasedOnTargetSize)
// The byte calculation is linear in numGrandparentFiles, but we will
// incur this linear cost in compact.Runner.TableSplitLimit() too, so we are
// also willing to pay it now. We could approximate this cheaply by using the
// mean file size of Lbase.
grandparentFileBytes := c.grandparents.AggregateSizeSum()
fileCountUpperBoundDueToGrandparents :=
float64(grandparentFileBytes) / float64(c.maxOverlapBytes)
if fileCountUpperBoundDueToGrandparents > acceptableFileCount {
c.maxOverlapBytes = uint64(
float64(c.maxOverlapBytes) *
(fileCountUpperBoundDueToGrandparents / acceptableFileCount))
}
}
// newFlush creates the state necessary for a flush (modeled with the compaction
// struct).
//
// newFlush takes the current Version in order to populate grandparent flushing
// limits, but it does not reference the version.
//
// TODO(jackson): Consider maintaining a reference to the version anyways since
// in the future in-memory Version state may only be available while a Version
// is referenced (eg, if we start recycling B-Tree nodes once they're no longer
// referenced). There's subtlety around unref'ing the version at the right
// moment, so we defer it for now.
func newFlush(
opts *Options,
cur *manifest.Version,
l0Organizer *manifest.L0Organizer,
baseLevel int,
flushing flushableList,
beganAt time.Time,
preferSharedStorage bool,
getValueSeparation getValueSeparation,
) (*tableCompaction, error) {
c := &tableCompaction{
kind: compactionKindFlush,
comparer: opts.Comparer,
logger: opts.Logger,
inputs: []compactionLevel{{level: -1}, {level: 0}},
getValueSeparation: getValueSeparation,
// TODO(radu): consider calculating the eventual output level for flushes.
// We expect the bounds to be very wide in practice, but perhaps we can do a
// finer-grained overlap analysis.
eventualOutputLevel: 0,
maxOutputFileSize: math.MaxUint64,
maxOverlapBytes: math.MaxUint64,
grantHandle: noopGrantHandle{},
metrics: compactionMetrics{
beganAt: beganAt,
},
}
c.flush.flushables = flushing
c.flush.l0Limits = l0Organizer.FlushSplitKeys()
c.startLevel = &c.inputs[0]
c.outputLevel = &c.inputs[1]
if len(flushing) > 0 {
if _, ok := flushing[0].flushable.(*ingestedFlushable); ok {
if len(flushing) != 1 {
panic(errors.AssertionFailedf("pebble: ingestedFlushable must be flushed one at a time."))
}
c.kind = compactionKindIngestedFlushable
return c, nil
} else {
// Make sure there's no ingestedFlushable after the first flushable
// in the list.
for _, f := range c.flush.flushables[1:] {
if _, ok := f.flushable.(*ingestedFlushable); ok {
panic(errors.AssertionFailedf("pebble: flushables shouldn't contain ingestedFlushable"))
}
}
}
}
c.objCreateOpts = objstorage.CreateOptions{
PreferSharedStorage: preferSharedStorage,
WriteCategory: getDiskWriteCategoryForCompaction(opts, c.kind),
}
if preferSharedStorage {
c.getValueSeparation = neverSeparateValues
}
cmp := c.comparer.Compare
updatePointBounds := func(iter internalIterator) {
if kv := iter.First(); kv != nil {
if c.bounds.Start == nil || cmp(c.bounds.Start, kv.K.UserKey) > 0 {
c.bounds.Start = slices.Clone(kv.K.UserKey)
}
}
if kv := iter.Last(); kv != nil {
if c.bounds.End.Key == nil || !c.bounds.End.IsUpperBoundForInternalKey(cmp, kv.K) {
c.bounds.End = base.UserKeyExclusiveIf(slices.Clone(kv.K.UserKey), kv.K.IsExclusiveSentinel())
}
}
}
updateRangeBounds := func(iter keyspan.FragmentIterator) error {
// File bounds require s != nil && !s.Empty(). We only need to check for
// s != nil here, as the memtable's FragmentIterator would never surface
// empty spans.
if s, err := iter.First(); err != nil {
return err
} else if s != nil {
c.bounds = c.bounds.Union(cmp, s.Bounds().Clone())
}
if s, err := iter.Last(); err != nil {
return err
} else if s != nil {
c.bounds = c.bounds.Union(cmp, s.Bounds().Clone())
}
return nil
}
var flushingBytes uint64
for i := range flushing {
f := flushing[i]
updatePointBounds(f.newIter(nil))
if rangeDelIter := f.newRangeDelIter(nil); rangeDelIter != nil {
if err := updateRangeBounds(rangeDelIter); err != nil {
return nil, err
}
}
if rangeKeyIter := f.newRangeKeyIter(nil); rangeKeyIter != nil {
if err := updateRangeBounds(rangeKeyIter); err != nil {
return nil, err
}
}
flushingBytes += f.inuseBytes()
}
if opts.FlushSplitBytes > 0 {
c.maxOutputFileSize = uint64(opts.TargetFileSizes[0])
c.maxOverlapBytes = maxGrandparentOverlapBytes(opts.TargetFileSizes[0])
c.grandparents = cur.Overlaps(baseLevel, c.bounds)
adjustGrandparentOverlapBytesForFlush(c, flushingBytes)
}
// We don't elide tombstones for flushes.
c.delElision, c.rangeKeyElision = compact.NoTombstoneElision(), compact.NoTombstoneElision()
return c, nil
}
func (c *tableCompaction) hasExtraLevelData() bool {
if len(c.extraLevels) == 0 {
// not a multi level compaction
return false
} else if c.extraLevels[0].files.Empty() {
// a multi level compaction without data in the intermediate input level;
// e.g. for a multi level compaction with levels 4,5, and 6, this could
// occur if there is no files to compact in 5, or in 5 and 6 (i.e. a move).
return false
}
return true
}
// errorOnUserKeyOverlap returns an error if the last two written sstables in
// this compaction have revisions of the same user key present in both sstables,
// when it shouldn't (eg. when splitting flushes).
func (c *tableCompaction) errorOnUserKeyOverlap(ve *manifest.VersionEdit) error {
if n := len(ve.NewTables); n > 1 {
meta := ve.NewTables[n-1].Meta
prevMeta := ve.NewTables[n-2].Meta
if !prevMeta.Largest().IsExclusiveSentinel() &&
c.comparer.Compare(prevMeta.Largest().UserKey, meta.Smallest().UserKey) >= 0 {
return errors.Errorf("pebble: compaction split user key across two sstables: %s in %s and %s",
prevMeta.Largest().Pretty(c.comparer.FormatKey),
prevMeta.TableNum,
meta.TableNum)
}
}
return nil
}
// isBottommostDataLayer returns true if the compaction's inputs are known to be
// the bottommost layer of data for the compaction's key range. If true, this
// allows the compaction iterator to perform transformations to keys such as
// setting a key's sequence number to zero.
//
// This function performs this determination by looking at the TombstoneElision
// values which are set up based on sstables which overlap the bounds of the
// compaction at a lower level in the LSM. This function always returns false
// for flushes.
func (c *tableCompaction) isBottommostDataLayer() bool {
// TODO(peter): we disable zeroing of seqnums during flushing to match
// RocksDB behavior and to avoid generating overlapping sstables during
// DB.replayWAL. When replaying WAL files at startup, we flush after each
// WAL is replayed building up a single version edit that is
// applied. Because we don't apply the version edit after each flush, this
// code doesn't know that L0 contains files and zeroing of seqnums should
// be disabled. That is fixable, but it seems safer to just match the
// RocksDB behavior for now.
return len(c.flush.flushables) == 0 && c.delElision.ElidesEverything() && c.rangeKeyElision.ElidesEverything()
}
// newInputIters returns an iterator over all the input tables in a compaction.
func (c *tableCompaction) newInputIters(
newIters tableNewIters, iiopts internalIterOpts,
) (
pointIter internalIterator,
rangeDelIter, rangeKeyIter keyspan.FragmentIterator,
retErr error,
) {
ctx := c.ctx
cmp := c.comparer.Compare
// Validate the ordering of compaction input files for defense in depth.
if len(c.flush.flushables) == 0 {
if c.startLevel.level >= 0 {
err := manifest.CheckOrdering(c.comparer, manifest.Level(c.startLevel.level),