db: move latest version state into a shared struct

Consolidate the various fields that hold mutable state describing the latest
version. Consolidating helps highlight the distinction that this state is
expected to be mutated, unlike an immutable manifest.Version.

Additonally, this prepares for passing additional mutable state into the
compaction picker so that the compaction picker is capable of picking blob file
rewrite compactions.

Author: jbowens

checkpoint.go

@@ -197,10 +197,10 @@ func (d *DB) Checkpoint(
 	optionsFileNum := d.optionsFileNum
 
 	virtualBackingFiles := make(map[base.DiskFileNum]struct{})
-	d.mu.versions.virtualBackings.ForEach(func(backing *manifest.TableBacking) {
+	d.mu.versions.latest.virtualBackings.ForEach(func(backing *manifest.TableBacking) {
 		virtualBackingFiles[backing.DiskFileNum] = struct{}{}
 	})
-	versionBlobFiles := d.mu.versions.blobFiles.Metadatas()
+	versionBlobFiles := d.mu.versions.latest.blobFiles.Metadatas()
 
 	// Acquire the logs while holding mutexes to ensure we don't race with a
 	// flush that might mark a log that's relevant to `current` as obsolete

checkpoint_test.go

@@ -151,7 +151,7 @@ func testCheckpointImpl(t *testing.T, ddFile string, createOnShared bool) {
 			d := dbs[td.CmdArgs[0].String()]
 			d.mu.Lock()
 			d.mu.versions.logLock()
-			fileNums := d.mu.versions.virtualBackings.DiskFileNums()
+			fileNums := d.mu.versions.latest.virtualBackings.DiskFileNums()
 			d.mu.versions.logUnlock()
 			d.mu.Unlock()
 

compaction.go

@@ -1179,7 +1179,7 @@ func (d *DB) addInProgressCompaction(c *compaction) {
 		if isIntraL0 {
 			l0Inputs = append(l0Inputs, c.outputLevel.files)
 		}
-		if err := d.mu.versions.l0Organizer.UpdateStateForStartedCompaction(l0Inputs, isBase); err != nil {
+		if err := d.mu.versions.latest.l0Organizer.UpdateStateForStartedCompaction(l0Inputs, isBase); err != nil {
 			d.opts.Logger.Fatalf("could not update state for compaction: %s", err)
 		}
 	}
@@ -1236,7 +1236,7 @@ func (d *DB) clearCompactingState(c *compaction, rollback bool) {
 		// may be able to pick a better compaction (though when this compaction
 		// succeeded we've also cleared the cache in UpdateVersionLocked).
 		defer d.mu.versions.logUnlockAndInvalidatePickedCompactionCache()
-		d.mu.versions.l0Organizer.InitCompactingFileInfo(l0InProgress)
+		d.mu.versions.latest.l0Organizer.InitCompactingFileInfo(l0InProgress)
 	}()
 }
 
@@ -1593,7 +1593,7 @@ func (d *DB) flush1() (bytesFlushed uint64, err error) {
 		}
 	}
 
-	c, err := newFlush(d.opts, d.mu.versions.currentVersion(), d.mu.versions.l0Organizer,
+	c, err := newFlush(d.opts, d.mu.versions.currentVersion(), d.mu.versions.latest.l0Organizer,
 		d.mu.versions.picker.getBaseLevel(), d.mu.mem.queue[:n], d.timeNow(), d.TableFormat(), d.determineCompactionValueSeparation)
 	if err != nil {
 		return 0, err
@@ -2061,7 +2061,7 @@ func (d *DB) tryScheduleDownloadCompactions(env compactionEnv, maxConcurrentDown
 			break
 		}
 		download := d.mu.compact.downloads[i]
-		switch d.tryLaunchDownloadCompaction(download, vers, d.mu.versions.l0Organizer, env, maxConcurrentDownloads) {
+		switch d.tryLaunchDownloadCompaction(download, vers, d.mu.versions.latest.l0Organizer, env, maxConcurrentDownloads) {
 		case launchedCompaction:
 			started = true
 			continue
@@ -2080,7 +2080,8 @@ func (d *DB) pickManualCompaction(env compactionEnv) (pc *pickedCompaction) {
 	v := d.mu.versions.currentVersion()
 	for len(d.mu.compact.manual) > 0 {
 		manual := d.mu.compact.manual[0]
-		pc, retryLater := newPickedManualCompaction(v, d.mu.versions.l0Organizer, d.opts, env, d.mu.versions.picker.getBaseLevel(), manual)
+		pc, retryLater := newPickedManualCompaction(v, d.mu.versions.latest.l0Organizer,
+			d.opts, env, d.mu.versions.picker.getBaseLevel(), manual)
 		if pc != nil {
 			return pc
 		}

compaction_picker.go

@@ -599,16 +599,14 @@ func canCompactTables(
 // installed).
 func newCompactionPickerByScore(
 	v *manifest.Version,
-	l0Organizer *manifest.L0Organizer,
-	virtualBackings *manifest.VirtualBackings,
+	lvs *latestVersionState,
 	opts *Options,
 	inProgressCompactions []compactionInfo,
 ) *compactionPickerByScore {
 	p := &compactionPickerByScore{
-		opts:            opts,
-		vers:            v,
-		l0Organizer:     l0Organizer,
-		virtualBackings: virtualBackings,
+		opts:               opts,
+		vers:               v,
+		latestVersionState: lvs,
 	}
 	p.initLevelMaxBytes(inProgressCompactions)
 	return p
@@ -694,16 +692,16 @@ func totalCompensatedSize(iter iter.Seq[*manifest.TableMetadata]) uint64 {
 type compactionPickerByScore struct {
 	opts *Options
 	vers *manifest.Version
-	// Unlike vers, which is immutable and the latest version when this picker is
-	// created, l0Organizer represents the mutable L0 state of the latest version.
-	// This means that at some point in the future a compactionPickerByScore
-	// created in the past will have mutually inconsistent state in vers and
-	// l0Organizer. This is not a problem since (a) a new picker is created in
-	// UpdateVersionLocked when a new version is installed, and (b) only the
-	// latest picker is used for picking compactions. This is ensured by holding
-	// versionSet.logLock for both (a) and (b).
-	l0Organizer     *manifest.L0Organizer
-	virtualBackings *manifest.VirtualBackings
+	// Unlike vers, which is immutable and the latest version when this picker
+	// is created, latestVersionState represents the mutable state of the latest
+	// version. This means that at some point in the future a
+	// compactionPickerByScore created in the past will have mutually
+	// inconsistent state in vers and latestVersionState. This is not a problem
+	// since (a) a new picker is created in UpdateVersionLocked when a new
+	// version is installed, and (b) only the latest picker is used for picking
+	// compactions. This is ensured by holding versionSet.logLock for both (a)
+	// and (b).
+	latestVersionState *latestVersionState
 	// The level to target for L0 compactions. Levels L1 to baseLevel must be
 	// empty.
 	baseLevel int
@@ -939,7 +937,7 @@ func (p *compactionPickerByScore) calculateLevelScores(
 		scores[i].level = i
 		scores[i].outputLevel = i + 1
 	}
-	l0FillFactor := calculateL0FillFactor(p.vers, p.l0Organizer, p.opts, inProgressCompactions)
+	l0FillFactor := calculateL0FillFactor(p.vers, p.latestVersionState.l0Organizer, p.opts, inProgressCompactions)
 	scores[0] = candidateLevelInfo{
 		outputLevel:           p.baseLevel,
 		fillFactor:            l0FillFactor,
@@ -1268,7 +1266,7 @@ func (p *compactionPickerByScore) getCompactionConcurrency() int {
 	// l0ReadAmp / p.opts.Experimental.L0CompactionConcurrency extra compactions.
 	l0ReadAmpCompactions := 0
 	if p.opts.Experimental.L0CompactionConcurrency > 0 {
-		l0ReadAmp := p.l0Organizer.MaxDepthAfterOngoingCompactions()
+		l0ReadAmp := p.latestVersionState.l0Organizer.MaxDepthAfterOngoingCompactions()
 		l0ReadAmpCompactions = (l0ReadAmp / p.opts.Experimental.L0CompactionConcurrency)
 	}
 	// compactionDebt >= ccSignal2 then can run another compaction, where
@@ -1378,7 +1376,7 @@ func (p *compactionPickerByScore) pickAutoScore(env compactionEnv) (pc *pickedCo
 		}
 
 		if info.level == 0 {
-			pc = pickL0(env, p.opts, p.vers, p.l0Organizer, p.baseLevel)
+			pc = pickL0(env, p.opts, p.vers, p.latestVersionState.l0Organizer, p.baseLevel)
 			// Fail-safe to protect against compacting the same sstable
 			// concurrently.
 			if pc != nil && !inputRangeAlreadyCompacting(env, pc) {
@@ -1394,12 +1392,12 @@ func (p *compactionPickerByScore) pickAutoScore(env compactionEnv) (pc *pickedCo
 
 		// info.level > 0
 		var ok bool
-		info.file, ok = pickCompactionSeedFile(p.vers, p.virtualBackings, p.opts, info.level, info.outputLevel, env.earliestSnapshotSeqNum, env.problemSpans)
+		info.file, ok = pickCompactionSeedFile(p.vers, &p.latestVersionState.virtualBackings, p.opts, info.level, info.outputLevel, env.earliestSnapshotSeqNum, env.problemSpans)
 		if !ok {
 			continue
 		}
 
-		pc := pickAutoLPositive(env, p.opts, p.vers, p.l0Organizer, *info, p.baseLevel)
+		pc := pickAutoLPositive(env, p.opts, p.vers, p.latestVersionState.l0Organizer, *info, p.baseLevel)
 		// Fail-safe to protect against compacting the same sstable concurrently.
 		if pc != nil && !inputRangeAlreadyCompacting(env, pc) {
 			p.addScoresToPickedCompactionMetrics(pc, scores)
@@ -1583,7 +1581,8 @@ func (p *compactionPickerByScore) pickedCompactionFromCandidateFile(
 		}
 	}
 
-	pc := newPickedCompaction(p.opts, p.vers, p.l0Organizer, startLevel, outputLevel, p.baseLevel)
+	pc := newPickedCompaction(p.opts, p.vers, p.latestVersionState.l0Organizer,
+		startLevel, outputLevel, p.baseLevel)
 	pc.kind = kind
 	pc.startLevel.files = inputs
 	pc.smallest, pc.largest = manifest.KeyRange(pc.cmp, pc.startLevel.files.All())
@@ -2030,7 +2029,8 @@ func pickReadTriggeredCompactionHelper(
 		return nil
 	}
 
-	pc = newPickedCompaction(p.opts, p.vers, p.l0Organizer, rc.level, defaultOutputLevel(rc.level, p.baseLevel), p.baseLevel)
+	pc = newPickedCompaction(p.opts, p.vers, p.latestVersionState.l0Organizer,
+		rc.level, defaultOutputLevel(rc.level, p.baseLevel), p.baseLevel)
 
 	pc.startLevel.files = overlapSlice
 	if !pc.setupInputs(p.opts, env.diskAvailBytes, pc.startLevel, env.problemSpans) {