pebble: extract manifest rotation logic into record.RotationHelper

db_test.go

@@ -936,7 +936,7 @@ func TestRollManifest(t *testing.T) {
 	sizeRolloverState := func() (int64, int64) {
 		d.mu.Lock()
 		defer d.mu.Unlock()
-		return d.mu.versions.lastSnapshotFileCount, d.mu.versions.editsSinceLastSnapshotFileCount
+		return d.mu.versions.rotationHelper.DebugInfo()
 	}
 
 	current := func() string {

record/rotation.go

@@ -0,0 +1,81 @@
+// Copyright 2023 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 record
+
+// RotationHelper is a type used to inform the decision of rotating a record log
+// file.
+//
+// The assumption is that multiple records can be coalesced into a single record
+// (called a snapshot). Starting a new file, where the first record is a
+// snapshot of the current state is referred to as "rotating" the log.
+//
+// Normally we rotate files when a certain file size is reached. But in certain
+// cases (e.g. contents become very large), this can result in too frequent
+// rotation. This helper contains logic to impose extra conditions on the
+// rotation.
+//
+// The rotation helper uses "size" as a unit-less estimation that is correlated
+// with the on-disk size of a record or snapshot.
+type RotationHelper struct {
+	// lastSnapshotSize is the size of the last snapshot.
+	lastSnapshotSize int64
+	// sizeSinceLastSnapshot is the sum of sizes of records applied since the last
+	// snapshot.
+	sizeSinceLastSnapshot int64
+	lastRecordSize        int64
+}
+
+// AddRecord makes the rotation helper aware of a new record.
+func (rh *RotationHelper) AddRecord(recordSize int64) {
+	rh.sizeSinceLastSnapshot += recordSize
+	rh.lastRecordSize = recordSize
+}
+
+// ShouldRotate returns whether we should start a new log file (with a snapshot).
+// Does not need to be called if other rotation factors (log file size) are not
+// satisfied.
+func (rh *RotationHelper) ShouldRotate(nextSnapshotSize int64) bool {
+	// The primary goal is to ensure that when reopening a log file, the number of
+	// edits that need to be replayed on top of the snapshot is "sane" while
+	// keeping the rotation frequency as low as possible.
+	//
+	// For the purposes of this description, we assume that the log is mainly
+	// storing a collection of "entries", with edits adding or removing entries.
+	// Consider the following cases:
+	//
+	// - The number of live entries is roughly stable: after writing the snapshot
+	//   (with S entries), we require that there be enough edits such that the
+	//   cumulative number of entries in those edits, E, be greater than S. This
+	//   will ensure that at most 50% of data written out is due to rotation.
+	//
+	// - The number of live entries K in the DB is shrinking drastically, say from
+	//   S to S/10: After this shrinking, E = 0.9F, and so if we used the previous
+	//   snapshot entry count, S, as the threshold that needs to be exceeded, we
+	//   will further delay the snapshot writing. Which means on reopen we will
+	//   need to replay 0.9S edits to get to a version with 0.1S entries. It would
+	//   be better to create a new snapshot when E exceeds the number of entries in
+	//   the current version.
+	//
+	// - The number of live entries L in the DB is growing; say the last snapshot
+	//   had S entries, and now we have 10S entries, so E = 9S. We will further
+	//   delay writing a new snapshot.
+	//
+	// The logic below uses the min of the last snapshot size count and the size
+	// count in the current version.
+	return rh.sizeSinceLastSnapshot > rh.lastSnapshotSize || rh.sizeSinceLastSnapshot > nextSnapshotSize
+}
+
+// Rotate makes the rotation helper aware that we are rotating to a new snapshot
+// (to which we will apply the latest edit).
+func (rh *RotationHelper) Rotate(snapshotSize int64) {
+	rh.lastSnapshotSize = snapshotSize
+	rh.sizeSinceLastSnapshot = rh.lastRecordSize
+}
+
+// DebugInfo returns the last snapshot size and size of the edits since the last
+// snapshot; used for testing and debugging.
+func (rh *RotationHelper) DebugInfo() (lastSnapshotSize int64, sizeSinceLastSnapshot int64) {
+	return rh.lastSnapshotSize, rh.sizeSinceLastSnapshot
+}

record/rotation_test.go

@@ -0,0 +1,49 @@
+// Copyright 2023 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 record
+
+import (
+	"fmt"
+	"strconv"
+	"testing"
+
+	"github.com/cockroachdb/datadriven"
+)
+
+func TestRotation(t *testing.T) {
+	var rh RotationHelper
+	datadriven.RunTest(t, "testdata/rotation", func(t *testing.T, td *datadriven.TestData) string {
+		oneIntArg := func() int64 {
+			if len(td.CmdArgs) != 1 {
+				td.Fatalf(t, "expected one integer argument")
+			}
+			n, err := strconv.Atoi(td.CmdArgs[0].String())
+			if err != nil {
+				td.Fatalf(t, "expected one integer argument")
+			}
+			return int64(n)
+		}
+		switch td.Cmd {
+		case "add":
+			size := oneIntArg()
+			rh.AddRecord(size)
+
+		case "should-rotate":
+			nextSnapshotSize := oneIntArg()
+			return fmt.Sprint(rh.ShouldRotate(nextSnapshotSize))
+
+		case "rotate":
+			snapshotSize := oneIntArg()
+			rh.Rotate(snapshotSize)
+
+		default:
+			td.Fatalf(t, "unknown command %s", td.Cmd)
+		}
+
+		// For commands with no output, show the debug info.
+		a, b := rh.DebugInfo()
+		return fmt.Sprintf("last-snapshot-size: %d\nsize-since-last-snapshot: %d", a, b)
+	})
+}

record/testdata/rotation

@@ -0,0 +1,65 @@
+rotate 100
+----
+last-snapshot-size: 100
+size-since-last-snapshot: 0
+
+add 10
+----
+last-snapshot-size: 100
+size-since-last-snapshot: 10
+
+# We should only rotate if the next snapshot is much smaller.
+should-rotate 100
+----
+false
+
+should-rotate 5
+----
+true
+
+add 50
+----
+last-snapshot-size: 100
+size-since-last-snapshot: 60
+
+add 50
+----
+last-snapshot-size: 100
+size-since-last-snapshot: 110
+
+add 50
+----
+last-snapshot-size: 100
+size-since-last-snapshot: 160
+
+# We exceeded the last snapshot size, we should rotate regardless.
+should-rotate 1
+----
+true
+
+should-rotate 1000
+----
+true
+
+add 1
+----
+last-snapshot-size: 100
+size-since-last-snapshot: 161
+
+rotate 10
+----
+last-snapshot-size: 10
+size-since-last-snapshot: 1
+
+add 5
+----
+last-snapshot-size: 10
+size-since-last-snapshot: 6
+
+should-rotate 5
+----
+true
+
+should-rotate 100
+----
+false

version_set.go

@@ -111,8 +111,7 @@ type versionSet struct {
 	writing    bool
 	writerCond sync.Cond
 	// State for deciding when to write a snapshot. Protected by mu.
-	lastSnapshotFileCount           int64
-	editsSinceLastSnapshotFileCount int64
+	rotationHelper record.RotationHelper
 }
 
 func (vs *versionSet) init(
@@ -433,37 +432,22 @@ func (vs *versionSet) logAndApply(
 	//
 	// The logic below uses the min of the last snapshot file count and the file
 	// count in the current version.
-	editCount := int64(len(ve.DeletedFiles) + len(ve.NewFiles))
-	vs.editsSinceLastSnapshotFileCount += editCount
+	vs.rotationHelper.AddRecord(int64(len(ve.DeletedFiles) + len(ve.NewFiles)))
 	sizeExceeded := vs.manifest.Size() >= vs.opts.MaxManifestFileSize
 	requireRotation := forceRotation || vs.manifest == nil
-	computeNextSnapshotFileCount := func() int64 {
-		var count int64
-		for i := range vs.metrics.Levels {
-			count += vs.metrics.Levels[i].NumFiles
-		}
-		return count
+
+	var nextSnapshotFilecount int64
+	for i := range vs.metrics.Levels {
+		nextSnapshotFilecount += vs.metrics.Levels[i].NumFiles
 	}
-	var nextSnapshotFileCount int64
 	if sizeExceeded && !requireRotation {
-		if vs.editsSinceLastSnapshotFileCount > vs.lastSnapshotFileCount {
-			requireRotation = true
-		} else {
-			nextSnapshotFileCount = computeNextSnapshotFileCount()
-			if vs.editsSinceLastSnapshotFileCount > nextSnapshotFileCount {
-				requireRotation = true
-			}
-		}
+		requireRotation = vs.rotationHelper.ShouldRotate(nextSnapshotFilecount)
 	}
 	var newManifestFileNum FileNum
 	var prevManifestFileSize uint64
 	if requireRotation {
 		newManifestFileNum = vs.getNextFileNum()
 		prevManifestFileSize = uint64(vs.manifest.Size())
-		if nextSnapshotFileCount == 0 {
-			// Haven't computed it, or happens to be 0.
-			nextSnapshotFileCount = computeNextSnapshotFileCount()
-		}
 	}
 
 	// Grab certain values before releasing vs.mu, in case createManifest() needs
@@ -542,8 +526,7 @@ func (vs *versionSet) logAndApply(
 
 	if requireRotation {
 		// Successfully rotated.
-		vs.lastSnapshotFileCount = nextSnapshotFileCount
-		vs.editsSinceLastSnapshotFileCount = editCount
+		vs.rotationHelper.Rotate(nextSnapshotFilecount)
 	}
 	// Now that DB.mu is held again, initialize compacting file info in
 	// L0Sublevels.