storage: add storage.sstable.compression_algorithm cluster setting

Introduce a new cluster setting that allows the operator to configure the
compression algorithm used when compressing sstable blocks. This allows
operators to opt into use of zstd (as opposed to the previous setting of
snappy). ZSTD typically achieves better compression ratios than snappy, and
operators may find that they can achieve higher node densities through enabling
zstd. Future releases may change the default compression algorithm.

In a side-by-side comparison of a 10000-warehouse tpcc import, the zstd cluster
achieved a higher import speed of 146 MiB/s versus snappy's 135 MiB/s. The zstd
cluster's physical database size was significantly less.

Epic: none
Release note (ops change): Add `storage.sstable.compression_algorithm` cluster
setting that configures the compression algorithm to use when compressing
sstable blocks.

pkg/storage/pebble.go

@@ -132,6 +132,44 @@ var IngestAsFlushable = settings.RegisterBoolSetting(
 	util.ConstantWithMetamorphicTestBool(
 		"storage.ingest_as_flushable.enabled", true))
 
+const (
+	compressionAlgorithmSnappy int64 = 1
+	compressionAlgorithmZstd   int64 = 2
+)
+
+// compressionAlgorithm determines the compression algorithm used to compress
+// data blocks when writing sstables. Users should call getCompressionAlgorithm
+// rather than calling compressionAlgorithm.Get directly.
+var compressionAlgorithm = settings.RegisterEnumSetting(
+	settings.SystemVisible,
+	"storage.sstable.compression_algorithm",
+	`determines the compression algorithm to use when compressing sstable data blocks;`+
+		` supported values: "snappy", "zstd"`,
+	"snappy",
+	map[int64]string{
+		compressionAlgorithmSnappy: "snappy",
+		compressionAlgorithmZstd:   "zstd",
+	},
+	settings.WithPublic,
+)
+
+func getCompressionAlgorithm(settings *cluster.Settings) pebble.Compression {
+	switch compressionAlgorithm.Get(&settings.SV) {
+	case compressionAlgorithmSnappy:
+		return pebble.SnappyCompression
+	case compressionAlgorithmZstd:
+		// Pre-24.1 Pebble's implementation of zstd had bugs that could cause
+		// in-memory corruption. We require that the 24.1 version be finalized
+		// before zstd will have any effect.
+		if settings.Version.ActiveVersionOrEmpty(context.TODO()).IsActive(clusterversion.V24_1Start) {
+			return pebble.ZstdCompression
+		}
+		return pebble.DefaultCompression
+	default:
+		return pebble.DefaultCompression
+	}
+}
+
 // DO NOT set storage.single_delete.crash_on_invariant_violation.enabled or
 // storage.single_delete.crash_on_ineffectual.enabled to true.
 //
@@ -1025,6 +1063,11 @@ func newPebble(ctx context.Context, cfg PebbleConfig) (p *Pebble, err error) {
 	}
 	opts.FS = cfg.Env
 	opts.Lock = cfg.Env.DirectoryLock
+	for _, l := range opts.Levels {
+		l.Compression = func() sstable.Compression {
+			return getCompressionAlgorithm(cfg.Settings)
+		}
+	}
 	opts.EnsureDefaults()
 
 	// The context dance here is done so that we have a clean context without

pkg/storage/sst_writer.go

@@ -82,6 +82,7 @@ func MakeIngestionWriterOptions(ctx context.Context, cs *cluster.Settings) sstab
 		format = sstable.TableFormatPebblev4
 	}
 	opts := DefaultPebbleOptions().MakeWriterOptions(0, format)
+	opts.Compression = getCompressionAlgorithm(cs)
 	opts.MergerName = "nullptr"
 	return opts
 }