From 5e9cc226ff76732ff165d755b940127752176a2e Mon Sep 17 00:00:00 2001
From: sumeerbhola <sumeer@cockroachlabs.com>
Date: Fri, 26 Jul 2024 11:54:45 -0400
Subject: [PATCH] storage: fix storage.value_blocks.enabled usage in
 MakeIngestionWriterOptions

This will allow the existing cluster setting to be used to disable value
blocks even for ingestion, as was intended.

Informs #127678

Epic: none

Release note: None
---
 pkg/kv/kvnemesis/testdata/TestOperationsFormat/4 |  2 +-
 pkg/storage/pebble.go                            |  5 +++--
 pkg/storage/sst_writer.go                        | 15 +++++++++------
 pkg/storage/sst_writer_test.go                   |  9 +++++++++
 4 files changed, 22 insertions(+), 9 deletions(-)

diff --git a/pkg/kv/kvnemesis/testdata/TestOperationsFormat/4 b/pkg/kv/kvnemesis/testdata/TestOperationsFormat/4
index ef926f053e3f..5f2a811172c4 100644
--- a/pkg/kv/kvnemesis/testdata/TestOperationsFormat/4
+++ b/pkg/kv/kvnemesis/testdata/TestOperationsFormat/4
@@ -1,6 +1,6 @@
 echo
 ----
-···db0.AddSSTable(ctx, tk(1), tk(4), ... /* @s1 */) // 1144 bytes (as writes)
+···db0.AddSSTable(ctx, tk(1), tk(4), ... /* @s1 */) // 1114 bytes (as writes)
 ···// ^-- tk(1) -> sv(s1): /Table/100/"0000000000000001"/0.000000001,0 -> /BYTES/v1
 ···// ^-- tk(2) -> sv(s1): /Table/100/"0000000000000002"/0.000000001,0 -> /<empty>
 ···// ^-- [tk(3), tk(4)) -> sv(s1): /Table/100/"000000000000000{3"-4"} -> /<empty>
diff --git a/pkg/storage/pebble.go b/pkg/storage/pebble.go
index 79ddad1b1cf4..d1f2dfadca0b 100644
--- a/pkg/storage/pebble.go
+++ b/pkg/storage/pebble.go
@@ -91,8 +91,9 @@ var MaxSyncDurationFatalOnExceeded = settings.RegisterBoolSetting(
 // ValueBlocksEnabled controls whether older versions of MVCC keys in the same
 // sstable will have their values written to value blocks. This only affects
 // sstables that will be written in the future, as part of flushes or
-// compactions, and does not eagerly change the encoding of existing sstables.
-// Reads can correctly read both kinds of sstables.
+// compactions inside Pebble, and for ingestions (outside Pebble), and does
+// not eagerly change the encoding of existing sstables. Reads can correctly
+// read both kinds of sstables.
 var ValueBlocksEnabled = settings.RegisterBoolSetting(
 	settings.ApplicationLevel, // used for temp storage in virtual cluster servers
 	"storage.value_blocks.enabled",
diff --git a/pkg/storage/sst_writer.go b/pkg/storage/sst_writer.go
index 19aa0d24a466..8bdf7633eddf 100644
--- a/pkg/storage/sst_writer.go
+++ b/pkg/storage/sst_writer.go
@@ -69,12 +69,15 @@ func MakeIngestionWriterOptions(ctx context.Context, cs *cluster.Settings) sstab
 	// table features available. Upgrade to an appropriate version only if the
 	// cluster supports it.
 	format := sstable.TableFormatPebblev2
-	if cs.Version.IsActive(ctx, clusterversion.V23_1EnablePebbleFormatSSTableValueBlocks) &&
-		ValueBlocksEnabled.Get(&cs.SV) {
-		format = sstable.TableFormatPebblev3
-	}
-	if cs.Version.IsActive(ctx, clusterversion.V23_2_EnablePebbleFormatVirtualSSTables) {
-		format = sstable.TableFormatPebblev4
+	// Don't ratchet up the format if value blocks are disabled, since the later
+	// formats always enable value blocks.
+	if ValueBlocksEnabled.Get(&cs.SV) {
+		if cs.Version.IsActive(ctx, clusterversion.V23_1EnablePebbleFormatSSTableValueBlocks) {
+			format = sstable.TableFormatPebblev3
+		}
+		if cs.Version.IsActive(ctx, clusterversion.V23_2_EnablePebbleFormatVirtualSSTables) {
+			format = sstable.TableFormatPebblev4
+		}
 	}
 	opts := DefaultPebbleOptions().MakeWriterOptions(0, format)
 	opts.MergerName = "nullptr"
diff --git a/pkg/storage/sst_writer_test.go b/pkg/storage/sst_writer_test.go
index 8e69680bca35..f904cc0e50be 100644
--- a/pkg/storage/sst_writer_test.go
+++ b/pkg/storage/sst_writer_test.go
@@ -91,6 +91,15 @@ func TestMakeIngestionWriterOptions(t *testing.T) {
 			}(),
 			want: sstable.TableFormatPebblev4,
 		},
+		{
+			name: "disable value blocks",
+			st: func() *cluster.Settings {
+				st := cluster.MakeTestingClusterSettings()
+				ValueBlocksEnabled.Override(context.Background(), &st.SV, false)
+				return st
+			}(),
+			want: sstable.TableFormatPebblev2,
+		},
 	}
 
 	for _, tc := range testCases {