db: add compaction interface

In preparation for the introduction of blob file rewrite compactions, rename
the existing compaction struct to tableCompaction and add a compaction
interface.

This change is largely mechanical. The interface is not as simple as it could
be and should be improved, but this is deferred for now.

Author: jbowens

compaction.go

@@ -73,7 +73,7 @@ type pickedCompaction interface {
 	// compaction is not a result of a manual compaction.
 	ManualID() uint64
 	// ConstructCompaction creates a compaction from the picked compaction.
-	ConstructCompaction(*DB, CompactionGrantHandle) *compaction
+	ConstructCompaction(*DB, CompactionGrantHandle) compaction
 	// WaitingCompaction returns a WaitingCompaction description of this
 	// compaction for consumption by the compaction scheduler.
 	WaitingCompaction() WaitingCompaction
@@ -236,7 +236,7 @@ func (pc *pickedTableCompaction) Score() float64 { return pc.score }
 // pickedTableCompaction.
 func (pc *pickedTableCompaction) ConstructCompaction(
 	d *DB, grantHandle CompactionGrantHandle,
-) *compaction {
+) compaction {
 	return newCompaction(
 		pc,
 		d.opts,

compaction_picker.go

@@ -1654,7 +1654,7 @@ func TestCompactionPickerScores(t *testing.T) {
 						d.mu.Lock()
 						wait := len(d.mu.compact.inProgress) > 0
 						for c := range d.mu.compact.inProgress {
-							wait = wait && !c.versionEditApplied
+							wait = wait && !c.VersionEditApplied()
 						}
 						d.mu.Unlock()
 						if !wait {

compaction_picker_test.go

@@ -8,7 +8,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
 )
 
@@ -149,17 +148,6 @@ func init() {
 		compactionOptionalAndPriority{optional: true, priority: 30}
 }
 
-func makeWaitingCompaction(manual bool, kind compactionKind, score float64) WaitingCompaction {
-	if manual {
-		return WaitingCompaction{Priority: manualCompactionPriority, Score: score}
-	}
-	entry, ok := scheduledCompactionMap[kind]
-	if !ok {
-		panic(errors.AssertionFailedf("unexpected compactionKind %s", kind))
-	}
-	return WaitingCompaction{Optional: entry.optional, Priority: entry.priority, Score: score}
-}
-
 // noopGrantHandle is used in cases that don't interact with a CompactionScheduler.
 type noopGrantHandle struct{}
 

compaction_scheduler.go

@@ -915,8 +915,8 @@ func TestCompaction(t *testing.T) {
 	}
 
 	// d.mu must be held when calling.
-	createOngoingCompaction := func(start, end []byte, startLevel, outputLevel int) (ongoingCompaction *compaction) {
-		ongoingCompaction = &compaction{
+	createOngoingCompaction := func(start, end []byte, startLevel, outputLevel int) (ongoingCompaction *tableCompaction) {
+		ongoingCompaction = &tableCompaction{
 			inputs: []compactionLevel{{level: startLevel}, {level: outputLevel}},
 			bounds: base.UserKeyBoundsInclusive(start, end),
 		}
@@ -937,7 +937,7 @@ func TestCompaction(t *testing.T) {
 	}
 
 	// d.mu must be held when calling.
-	deleteOngoingCompaction := func(ongoingCompaction *compaction) {
+	deleteOngoingCompaction := func(ongoingCompaction *tableCompaction) {
 		for _, cl := range ongoingCompaction.inputs {
 			for f := range cl.files.All() {
 				f.CompactionState = manifest.CompactionStateNotCompacting
@@ -949,7 +949,7 @@ func TestCompaction(t *testing.T) {
 
 	runTest := func(t *testing.T, testData string, minVersion, maxVersion FormatMajorVersion, verbose bool) {
 		reset(minVersion, maxVersion)
-		var ongoingCompaction *compaction
+		var ongoingCompaction *tableCompaction
 		datadriven.RunTest(t, testData, func(t *testing.T, td *datadriven.TestData) string {
 			switch td.Cmd {
 			case "reset":
@@ -2091,7 +2091,7 @@ func TestCompactionAllowZeroSeqNum(t *testing.T) {
 
 			case "allow-zero-seqnum":
 				d.mu.Lock()
-				c := &compaction{
+				c := &tableCompaction{
 					comparer: d.opts.Comparer,
 					version:  d.mu.versions.currentVersion(),
 					inputs:   []compactionLevel{{}, {}},
@@ -2173,7 +2173,7 @@ func TestCompactionErrorOnUserKeyOverlap(t *testing.T) {
 		func(t *testing.T, d *datadriven.TestData) string {
 			switch d.Cmd {
 			case "error-on-user-key-overlap":
-				c := &compaction{comparer: DefaultComparer}
+				c := &tableCompaction{comparer: DefaultComparer}
 				var files []manifest.NewTableEntry
 				tableNum := base.TableNum(1)
 
@@ -2300,7 +2300,7 @@ func TestCompactionCheckOrdering(t *testing.T) {
 		func(t *testing.T, d *datadriven.TestData) string {
 			switch d.Cmd {
 			case "check-ordering":
-				c := &compaction{
+				c := &tableCompaction{
 					comparer: DefaultComparer,
 					logger:   panicLogger{},
 					inputs:   []compactionLevel{{level: -1}, {level: -1}},
@@ -2552,7 +2552,7 @@ func TestAdjustGrandparentOverlapBytesForFlush(t *testing.T) {
 	}
 	for _, tc := range testCases {
 		t.Run("", func(t *testing.T) {
-			c := compaction{
+			c := tableCompaction{
 				grandparents:      ls,
 				maxOverlapBytes:   maxOverlapBytes,
 				maxOutputFileSize: maxOutputFileSize,

compaction_test.go

@@ -948,7 +948,7 @@ func runDBDefineCmdReuseFS(td *datadriven.TestData, opts *Options) (*DB, error)
 		if err != nil {
 			return err
 		}
-		c.getValueSeparation = func(JobID, *compaction, sstable.TableFormat) compact.ValueSeparation {
+		c.getValueSeparation = func(JobID, *tableCompaction, sstable.TableFormat) compact.ValueSeparation {
 			return valueSeparator
 		}
 		// NB: define allows the test to exactly specify which keys go
@@ -1018,12 +1018,12 @@ func runDBDefineCmdReuseFS(td *datadriven.TestData, opts *Options) (*DB, error)
 	}
 
 	// Example, compact: a-c.
-	parseCompaction := func(outputLevel int, s string) (*compaction, error) {
+	parseCompaction := func(outputLevel int, s string) (*tableCompaction, error) {
 		m, err := parseMeta(s)
 		if err != nil {
 			return nil, err
 		}
-		c := &compaction{
+		c := &tableCompaction{
 			inputs: []compactionLevel{{}, {level: outputLevel}},
 			bounds: m.UserKeyBounds(),
 		}

data_test.go

@@ -433,7 +433,7 @@ type DB struct {
 			// sublevels' state. Some of the compactions contained within this
 			// map may have already committed an edit to the version but are
 			// lingering performing cleanup, like deleting obsolete files.
-			inProgress map[*compaction]struct{}
+			inProgress map[compaction]struct{}
 
 			// rescheduleReadCompaction indicates to an iterator that a read compaction
 			// should be scheduled.
@@ -2080,8 +2080,8 @@ func (d *DB) Metrics() *Metrics {
 	metrics.Compact.MarkedFiles = vers.Stats.MarkedForCompaction
 	metrics.Compact.Duration = d.mu.compact.duration
 	for c := range d.mu.compact.inProgress {
-		if c.kind != compactionKindFlush && c.kind != compactionKindIngestedFlushable {
-			metrics.Compact.Duration += d.timeNow().Sub(c.metrics.beganAt)
+		if !c.IsFlush() {
+			metrics.Compact.Duration += d.timeNow().Sub(c.BeganAt())
 		}
 	}
 	metrics.Compact.NumProblemSpans = d.problemSpans.Len()
@@ -2860,19 +2860,10 @@ func (d *DB) getEarliestUnflushedSeqNumLocked() base.SeqNum {
 	return seqNum
 }
 
-func (d *DB) getInProgressCompactionInfoLocked(finishing *compaction) (rv []compactionInfo) {
+func (d *DB) getInProgressCompactionInfoLocked(finishing *tableCompaction) (rv []compactionInfo) {
 	for c := range d.mu.compact.inProgress {
-		if len(c.flush.flushables) == 0 && (finishing == nil || c != finishing) {
-			info := compactionInfo{
-				versionEditApplied: c.versionEditApplied,
-				inputs:             c.inputs,
-				bounds:             c.bounds,
-				outputLevel:        -1,
-			}
-			if c.outputLevel != nil {
-				info.outputLevel = c.outputLevel.level
-			}
-			rv = append(rv, info)
+		if !c.IsFlush() && (finishing == nil || c != finishing) {
+			rv = append(rv, c.Info())
 		}
 	}
 	return

db.go

@@ -449,7 +449,7 @@ func (d *DB) tryLaunchDownloadForFile(
 	c := newCompaction(pc, d.opts, d.timeNow(), d.objProvider, noopGrantHandle{}, d.TableFormat(), d.determineCompactionValueSeparation)
 	c.isDownload = true
 	d.mu.compact.downloadingCount++
-	d.addInProgressCompaction(c)
+	c.AddInProgressLocked(d)
 	go d.compact(c, doneCh)
 	return doneCh, true
 }

download.go

@@ -912,7 +912,7 @@ func ingestTargetLevel(
 	cmp base.Compare,
 	lsmOverlap overlap.WithLSM,
 	baseLevel int,
-	compactions map[*compaction]struct{},
+	compactions map[compaction]struct{},
 	meta *manifest.TableMetadata,
 	suggestSplit bool,
 ) (targetLevel int, splitFile *manifest.TableMetadata, err error) {
@@ -1027,10 +1027,14 @@ func ingestTargetLevel(
 		// unless necessary.
 		overlaps := false
 		for c := range compactions {
-			if c.outputLevel == nil || level != c.outputLevel.level {
+			tblCompaction, ok := c.(*tableCompaction)
+			if !ok {
+				continue
+			}
+			if tblCompaction.outputLevel == nil || level != tblCompaction.outputLevel.level {
 				continue
 			}
-			if metaBounds.Overlaps(cmp, &c.bounds) {
+			if metaBounds.Overlaps(cmp, tblCompaction.Bounds()) {
 				overlaps = true
 				break
 			}
@@ -2101,7 +2105,7 @@ func (d *DB) ingestApply(
 		}
 		if len(filesToSplit) > 0 || exciseSpan.Valid() {
 			for c := range d.mu.compact.inProgress {
-				if c.versionEditApplied {
+				if c.VersionEditApplied() {
 					continue
 				}
 				// Check if this compaction overlaps with the excise span. Note that just
@@ -2111,8 +2115,8 @@ func (d *DB) ingestApply(
 				// doing a [c,d) excise at the same time as this compaction, we will have
 				// to error out the whole compaction as we can't guarantee it hasn't/won't
 				// write a file overlapping with the excise span.
-				if c.bounds.Overlaps(d.cmp, &exciseBounds) {
-					c.cancel.Store(true)
+				if c.Bounds().Overlaps(d.cmp, &exciseBounds) {
+					c.Cancel()
 				}
 				// Check if this compaction's inputs have been replaced due to an
 				// ingest-time split. In that case, cancel the compaction as a newly picked
@@ -2121,12 +2125,10 @@ func (d *DB) ingestApply(
 				// file that was ingest-split as an input, even if it started before this
 				// ingestion.
 				if checkCompactions {
-					for i := range c.inputs {
-						for f := range c.inputs[i].files.All() {
-							if _, ok := replacedTables[f.TableNum]; ok {
-								c.cancel.Store(true)
-								break
-							}
+					for _, table := range c.Tables() {
+						if _, ok := replacedTables[table.TableNum]; ok {
+							c.Cancel()
+							break
 						}
 					}
 				}

ingest.go

@@ -274,7 +274,7 @@ func Open(dirname string, opts *Options) (db *DB, err error) {
 		d.mu.mem.nextSize = initialMemTableSize
 	}
 	d.mu.compact.cond.L = &d.mu.Mutex
-	d.mu.compact.inProgress = make(map[*compaction]struct{})
+	d.mu.compact.inProgress = make(map[compaction]struct{})
 	d.mu.compact.noOngoingFlushStartTime = crtime.NowMono()
 	d.mu.snapshots.init()
 	// logSeqNum is the next sequence number that will be assigned.