crl-release-24.3: backport a few fixes

.github/workflows/ci.yaml

@@ -125,7 +125,7 @@ jobs:
     strategy:
       matrix:
         go: ['1.22', '1.23']
-    runs-on: macos-12
+    runs-on: macos-14
     steps:
       - uses: actions/checkout@v2
 

checkpoint.go

@@ -5,6 +5,7 @@
 package pebble
 
 import (
+	"bytes"
 	"io"
 	"os"
 
@@ -239,10 +240,10 @@ func (d *DB) Checkpoint(
 	}
 
 	{
-		// Link or copy the OPTIONS.
+		// Copy the OPTIONS.
 		srcPath := base.MakeFilepath(fs, d.dirname, fileTypeOptions, optionsFileNum)
 		destPath := fs.PathJoin(destDir, fs.PathBase(srcPath))
-		ckErr = vfs.LinkOrCopy(fs, srcPath, destPath)
+		ckErr = copyCheckpointOptions(fs, srcPath, destPath)
 		if ckErr != nil {
 			return ckErr
 		}
@@ -373,6 +374,58 @@ func (d *DB) Checkpoint(
 	return ckErr
 }
 
+// copyCheckpointOptions copies an OPTIONS file, commenting out some options
+// that existed on the original database but no longer apply to the checkpointed
+// database. For example, the entire [WAL Failover] stanza is commented out
+// because Checkpoint will copy all WAL segment files from both the primary and
+// secondary WAL directories into the checkpoint.
+func copyCheckpointOptions(fs vfs.FS, srcPath, dstPath string) error {
+	var buf bytes.Buffer
+	f, err := fs.Open(srcPath)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+	b, err := io.ReadAll(f)
+	if err != nil {
+		return err
+	}
+	// Copy the OPTIONS file verbatim, but commenting out the [WAL Failover]
+	// section.
+	err = parseOptions(string(b), parseOptionsFuncs{
+		visitNewSection: func(startOff, endOff int, section string) error {
+			if section == "WAL Failover" {
+				buf.WriteString("# ")
+			}
+			buf.Write(b[startOff:endOff])
+			return nil
+		},
+		visitKeyValue: func(startOff, endOff int, section, key, value string) error {
+			if section == "WAL Failover" {
+				buf.WriteString("# ")
+			}
+			buf.Write(b[startOff:endOff])
+			return nil
+		},
+		visitCommentOrWhitespace: func(startOff, endOff int, line string) error {
+			buf.Write(b[startOff:endOff])
+			return nil
+		},
+	})
+	if err != nil {
+		return err
+	}
+	nf, err := fs.Create(dstPath, vfs.WriteCategoryUnspecified)
+	if err != nil {
+		return err
+	}
+	_, err = io.Copy(nf, &buf)
+	if err != nil {
+		return err
+	}
+	return errors.CombineErrors(nf.Sync(), nf.Close())
+}
+
 func (d *DB) writeCheckpointManifest(
 	fs vfs.FS,
 	formatVers FormatMajorVersion,

checkpoint_test.go

@@ -8,6 +8,7 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"io"
 	"math/rand/v2"
 	"runtime"
 	"sort"
@@ -35,22 +36,25 @@ func testCheckpointImpl(t *testing.T, ddFile string, createOnShared bool) {
 	mem := vfs.NewMem()
 	var memLog base.InMemLogger
 	remoteMem := remote.NewInMem()
-	opts := &Options{
-		FS:                          vfs.WithLogging(mem, memLog.Infof),
-		FormatMajorVersion:          internalFormatNewest,
-		L0CompactionThreshold:       10,
-		DisableAutomaticCompactions: true,
-		Logger:                      testLogger{t},
-	}
-	opts.Experimental.EnableColumnarBlocks = func() bool { return true }
-	opts.Experimental.RemoteStorage = remote.MakeSimpleFactory(map[remote.Locator]remote.Storage{
-		"": remoteMem,
-	})
-	if createOnShared {
-		opts.Experimental.CreateOnShared = remote.CreateOnSharedAll
+	makeOptions := func() *Options {
+		opts := &Options{
+			FS:                          vfs.WithLogging(mem, memLog.Infof),
+			FormatMajorVersion:          internalFormatNewest,
+			L0CompactionThreshold:       10,
+			DisableAutomaticCompactions: true,
+			Logger:                      testLogger{t},
+		}
+		opts.Experimental.EnableColumnarBlocks = func() bool { return true }
+		opts.Experimental.RemoteStorage = remote.MakeSimpleFactory(map[remote.Locator]remote.Storage{
+			"": remoteMem,
+		})
+		if createOnShared {
+			opts.Experimental.CreateOnShared = remote.CreateOnSharedAll
+		}
+		opts.DisableTableStats = true
+		opts.private.testingAlwaysWaitForCleanup = true
+		return opts
 	}
-	opts.DisableTableStats = true
-	opts.private.testingAlwaysWaitForCleanup = true
 
 	datadriven.RunTest(t, ddFile, func(t *testing.T, td *datadriven.TestData) string {
 		switch td.Cmd {
@@ -70,7 +74,7 @@ func testCheckpointImpl(t *testing.T, ddFile string, createOnShared bool) {
 			return memLog.String()
 
 		case "checkpoint":
-			if !(len(td.CmdArgs) == 2 || (len(td.CmdArgs) == 3 && td.CmdArgs[2].Key == "restrict")) {
+			if len(td.CmdArgs) < 2 {
 				return "checkpoint <db> <dir> [restrict=(start-end, ...)]"
 			}
 			var opts []CheckpointOption
@@ -93,6 +97,10 @@ func testCheckpointImpl(t *testing.T, ddFile string, createOnShared bool) {
 			if err := d.Checkpoint(td.CmdArgs[1].String(), opts...); err != nil {
 				return err.Error()
 			}
+			if td.HasArg("nondeterministic") {
+				memLog.Reset()
+				return ""
+			}
 			return memLog.String()
 
 		case "ingest-and-excise":
@@ -184,19 +192,19 @@ func testCheckpointImpl(t *testing.T, ddFile string, createOnShared bool) {
 			return fmt.Sprintf("%s\n", strings.Join(paths, "\n"))
 
 		case "open":
-			if len(td.CmdArgs) != 1 && len(td.CmdArgs) != 2 {
+			if len(td.CmdArgs) < 1 {
 				return "open <dir> [readonly]"
 			}
-			opts.ReadOnly = false
-			if len(td.CmdArgs) == 2 {
-				if td.CmdArgs[1].String() != "readonly" {
-					return "open <dir> [readonly]"
-				}
-				opts.ReadOnly = true
-			}
+			opts := makeOptions()
+			require.NoError(t, parseDBOptionsArgs(opts, td.CmdArgs[1:]))
 
 			memLog.Reset()
 			dir := td.CmdArgs[0].String()
+			if _, ok := dbs[dir]; ok {
+				require.NoError(t, dbs[dir].Close())
+				dbs[dir] = nil
+			}
+
 			d, err := Open(dir, opts)
 			if err != nil {
 				return err.Error()
@@ -208,6 +216,12 @@ func testCheckpointImpl(t *testing.T, ddFile string, createOnShared bool) {
 					return err.Error()
 				}
 			}
+			waitForCompactionsAndTableStats(d)
+
+			if td.HasArg("nondeterministic") {
+				memLog.Reset()
+				return ""
+			}
 			return memLog.String()
 
 		case "scan":
@@ -232,6 +246,33 @@ func testCheckpointImpl(t *testing.T, ddFile string, createOnShared bool) {
 	})
 }
 
+func TestCopyCheckpointOptions(t *testing.T) {
+	fs := vfs.NewMem()
+	datadriven.RunTest(t, "testdata/copy_checkpoint_options", func(t *testing.T, td *datadriven.TestData) string {
+		switch td.Cmd {
+		case "copy":
+			f, err := fs.Create("old", vfs.WriteCategoryUnspecified)
+			require.NoError(t, err)
+			_, err = io.WriteString(f, td.Input)
+			require.NoError(t, err)
+			require.NoError(t, f.Close())
+
+			if err := copyCheckpointOptions(fs, "old", "new"); err != nil {
+				return err.Error()
+			}
+
+			f, err = fs.Open("new")
+			require.NoError(t, err)
+			newFile, err := io.ReadAll(f)
+			require.NoError(t, err)
+			require.NoError(t, f.Close())
+			return string(newFile)
+		default:
+			panic(fmt.Sprintf("unrecognized command %q", td.Cmd))
+		}
+	})
+}
+
 func TestCheckpoint(t *testing.T) {
 	t.Run("shared=false", func(t *testing.T) {
 		testCheckpointImpl(t, "testdata/checkpoint", false /* createOnShared */)

compaction.go

@@ -2262,6 +2262,9 @@ func (d *DB) compact1(c *compaction, errChannel chan error) (err error) {
 			// the manifest lock, we don't expect this bool to change its value
 			// as only the holder of the manifest lock will ever write to it.
 			if c.cancel.Load() {
+				d.mu.versions.metrics.Compact.CancelledCount++
+				d.mu.versions.metrics.Compact.CancelledBytes += c.bytesWritten
+
 				err = firstError(err, ErrCancelledCompaction)
 				// This is the first time we've seen a cancellation during the
 				// life of this compaction (or the original condition on err == nil
@@ -2294,6 +2297,10 @@ func (d *DB) compact1(c *compaction, errChannel chan error) (err error) {
 	// NB: clearing compacting state must occur before updating the read state;
 	// L0Sublevels initialization depends on it.
 	d.clearCompactingState(c, err != nil)
+	if err != nil && errors.Is(err, ErrCancelledCompaction) {
+		d.mu.versions.metrics.Compact.CancelledCount++
+		d.mu.versions.metrics.Compact.CancelledBytes += c.bytesWritten
+	}
 	d.mu.versions.incrementCompactions(c.kind, c.extraLevels, c.pickerMetrics)
 	d.mu.versions.incrementCompactionBytes(-c.bytesWritten)
 

data_test.go

@@ -33,6 +33,7 @@ import (
 	"github.com/cockroachdb/pebble/sstable"
 	"github.com/cockroachdb/pebble/vfs"
 	"github.com/cockroachdb/pebble/vfs/errorfs"
+	"github.com/cockroachdb/pebble/wal"
 	"github.com/ghemawat/stream"
 	"github.com/stretchr/testify/require"
 )
@@ -1473,6 +1474,8 @@ func parseDBOptionsArgs(opts *Options, args []datadriven.CmdArg) error {
 			default:
 				return errors.Newf("unrecognized Merger %q\n", cmdArg.Vals[0])
 			}
+		case "readonly":
+			opts.ReadOnly = true
 		case "target-file-sizes":
 			if len(opts.Levels) < len(cmdArg.Vals) {
 				opts.Levels = slices.Grow(opts.Levels, len(cmdArg.Vals)-len(opts.Levels))[0:len(cmdArg.Vals)]
@@ -1484,6 +1487,15 @@ func parseDBOptionsArgs(opts *Options, args []datadriven.CmdArg) error {
 				}
 				opts.Levels[i].TargetFileSize = size
 			}
+		case "wal-failover":
+			if v := cmdArg.Vals[0]; v == "off" || v == "disabled" {
+				opts.WALFailover = nil
+				continue
+			}
+			opts.WALFailover = &WALFailoverOptions{
+				Secondary: wal.Dir{FS: opts.FS, Dirname: cmdArg.Vals[0]},
+			}
+			opts.WALFailover.EnsureDefaults()
 		}
 	}
 	return nil

metrics.go

@@ -178,6 +178,11 @@ type Metrics struct {
 		InProgressBytes int64
 		// Number of compactions that are in-progress.
 		NumInProgress int64
+		// Number of compactions that were cancelled.
+		CancelledCount int64
+		// CancelledBytes the number of bytes written by compactions that were
+		// cancelled.
+		CancelledBytes int64
 		// MarkedFiles is a count of files that are marked for
 		// compaction. Such files are compacted in a rewrite compaction
 		// when no other compactions are picked.

open.go

@@ -742,17 +742,19 @@ func GetVersion(dir string, fs vfs.FS) (string, error) {
 			if err != nil {
 				return "", err
 			}
-			err = parseOptions(string(data), func(section, key, value string) error {
-				switch {
-				case section == "Version":
-					switch key {
-					case "pebble_version":
-						version = value
-					case "rocksdb_version":
-						version = fmt.Sprintf("rocksdb v%s", value)
+			err = parseOptions(string(data), parseOptionsFuncs{
+				visitKeyValue: func(i, j int, section, key, value string) error {
+					switch {
+					case section == "Version":
+						switch key {
+						case "pebble_version":
+							version = value
+						case "rocksdb_version":
+							version = fmt.Sprintf("rocksdb v%s", value)
+						}
 					}
-				}
-				return nil
+					return nil
+				},
 			})
 			if err != nil {
 				return "", err