storage: ingest columnar-block sstables when enabled

When the active cluster verison is sufficiently high and the columnar blocks
cluster setting is enabled, build sstables for ingestion in table format
TableFormatPebblev5 with columnar blocks.

Epic: none
Release note: none

pkg/kv/kvnemesis/operations.go

@@ -296,8 +296,8 @@ func (op DeleteRangeUsingTombstoneOperation) format(w *strings.Builder, fctx for
 }
 
 func (op AddSSTableOperation) format(w *strings.Builder, fctx formatCtx) {
-	fmt.Fprintf(w, `%s.AddSSTable(%s%s, %s, ... /* @%s */) // %d bytes`,
-		fctx.receiver, fctx.maybeCtx(), fmtKey(op.Span.Key), fmtKey(op.Span.EndKey), op.Seq, len(op.Data))
+	fmt.Fprintf(w, `%s.AddSSTable(%s%s, %s, ... /* @%s */)`,
+		fctx.receiver, fctx.maybeCtx(), fmtKey(op.Span.Key), fmtKey(op.Span.EndKey), op.Seq)
 	if op.AsWrites {
 		fmt.Fprintf(w, ` (as writes)`)
 	}

pkg/kv/kvnemesis/testdata/TestApplier/addsstable

@@ -1,6 +1,6 @@
 echo
 ----
-db0.AddSSTable(ctx, tk(1), tk(4), ... /* @s1 */) // 1004 bytes (as writes)
+db0.AddSSTable(ctx, tk(1), tk(4), ... /* @s1 */) (as writes)
 // ^-- tk(1) -> sv(s1): /Table/100/"0000000000000001"/<ts> -> /BYTES/v1
 // ^-- tk(2) -> sv(s1): /Table/100/"0000000000000002"/<ts> -> /<empty>
 // ^-- [tk(3), tk(4)) -> sv(s1): /Table/100/"000000000000000{3"-4"} -> /<empty>

pkg/kv/kvnemesis/testdata/TestOperationsFormat/4

@@ -1,6 +1,6 @@
 echo
 ----
-···db0.AddSSTable(ctx, tk(1), tk(4), ... /* @s1 */) // 1004 bytes (as writes)
+···db0.AddSSTable(ctx, tk(1), tk(4), ... /* @s1 */) (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>

pkg/kv/kvnemesis/testdata/TestValidate/addsstable_ingestion

@@ -1,6 +1,6 @@
 echo
 ----
-db0.AddSSTable(ctx, tk(1), tk(4), ... /* @s1 */) // 1004 bytes
+db0.AddSSTable(ctx, tk(1), tk(4), ... /* @s1 */)
 // ^-- 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>

pkg/kv/kvnemesis/testdata/TestValidate/addsstable_ingestion_shadowing_scan

@@ -4,7 +4,7 @@ db0.Put(ctx, tk(1), sv(1)) // @0.000000001,0 <nil>
 db0.Put(ctx, tk(2), sv(1)) // @0.000000001,0 <nil>
 db0.Put(ctx, tk(3), sv(1)) // @0.000000001,0 <nil>
 db0.Put(ctx, tk(4), sv(1)) // @0.000000001,0 <nil>
-db0.AddSSTable(ctx, tk(1), tk(4), ... /* @s2 */) // 1005 bytes
+db0.AddSSTable(ctx, tk(1), tk(4), ... /* @s2 */)
 // ^-- tk(1) -> sv(s2): /Table/100/"0000000000000001"/0.000000001,0 -> /BYTES/v2
 // ^-- tk(2) -> sv(s2): /Table/100/"0000000000000002"/0.000000001,0 -> /<empty>
 // ^-- [tk(3), tk(4)) -> sv(s2): /Table/100/"000000000000000{3"-4"} -> /<empty>

pkg/kv/kvserver/batcheval/cmd_add_sstable_test.go

@@ -178,6 +178,7 @@ func TestEvalAddSSTable(t *testing.T) {
 			expectErr: []string{
 				`unexpected timestamp 2.000000000,0 (expected 1.000000000,0) for key "c"`,
 				`key has suffix "\x00\x00\x00\x00w5\x94\x00\t", expected "\x00\x00\x00\x00;\x9a\xca\x00\t"`,
+				`has suffix 0x000000007735940009; require 0x000000003b9aca0009`,
 			},
 		},
 		"SSTTimestampToRequestTimestamp rejects incorrect SST timestamp for range keys": {
@@ -187,6 +188,7 @@ func TestEvalAddSSTable(t *testing.T) {
 			expectErr: []string{
 				`unexpected timestamp 2.000000000,0 (expected 1.000000000,0) for range key {c-d}`,
 				`key has suffix "\x00\x00\x00\x00w5\x94\x00\t", expected "\x00\x00\x00\x00;\x9a\xca\x00\t"`,
+				`has suffix 0x000000007735940009; require 0x000000003b9aca0009`,
 			},
 		},
 		"SSTTimestampToRequestTimestamp rejects incorrect 0 SST timestamp": {
@@ -196,6 +198,7 @@ func TestEvalAddSSTable(t *testing.T) {
 			expectErr: []string{
 				`unexpected timestamp 0,0 (expected 1.000000000,0) for key "c"`,
 				`key has suffix "", expected "\x00\x00\x00\x00;\x9a\xca\x00\t"`,
+				`has suffix 0x; require 0x000000003b9aca0009`,
 			},
 			expectErrRace: `SST contains inline value or intent for key "c"/0,0`,
 		},

pkg/storage/mvcc_history_test.go

@@ -262,7 +262,8 @@ func TestMVCCHistories(t *testing.T) {
 		// SST iterator in order to accurately represent the raw SST data.
 		reportSSTEntries := func(buf *redact.StringBuilder, name string, sst []byte) error {
 			r, err := sstable.NewMemReader(sst, sstable.ReaderOptions{
-				Comparer: storage.EngineComparer,
+				Comparer:   storage.EngineComparer,
+				KeySchemas: sstable.MakeKeySchemas(storage.KeySchemas...),
 			})
 			if err != nil {
 				return err
@@ -275,7 +276,7 @@ func TestMVCCHistories(t *testing.T) {
 				return err
 			}
 			defer func() { _ = iter.Close() }()
-			for kv := iter.SeekGE(nil, 0); kv != nil; kv = iter.Next() {
+			for kv := iter.First(); kv != nil; kv = iter.Next() {
 				if err := iter.Error(); err != nil {
 					return err
 				}

pkg/storage/pebble_key_schema.go

@@ -387,6 +387,9 @@ func (ks *cockroachKeySeeker) IsLowerBound(k []byte, syntheticSuffix []byte) boo
 func (ks *cockroachKeySeeker) SeekGE(
 	key []byte, boundRow int, searchDir int8,
 ) (row int, equalPrefix bool) {
+	if buildutil.CrdbTestBuild && len(key) == 0 {
+		panic(errors.AssertionFailedf("seeking to empty key"))
+	}
 	// TODO(jackson): Inline EngineKeySplit.
 	si := EngineKeySplit(key)
 	row, eq := ks.roachKeys.Search(key[:si-1])

pkg/storage/sst_writer.go

@@ -10,6 +10,7 @@ import (
 	"context"
 	"io"
 
+	"github.com/cockroachdb/cockroach/pkg/clusterversion"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/settings"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
@@ -80,11 +81,14 @@ func (*noopFinishAbort) Abort() {}
 // MakeIngestionWriterOptions returns writer options suitable for writing SSTs
 // that will subsequently be ingested (e.g. with AddSSTable).
 func MakeIngestionWriterOptions(ctx context.Context, cs *cluster.Settings) sstable.WriterOptions {
-	// By default, take a conservative approach and assume we don't have newer
-	// table features available. Upgrade to an appropriate version only if the
-	// cluster supports it. Currently, all supported versions understand
-	// TableFormatPebblev4.
+	// All supported versions understand TableFormatPebblev4. If columnar blocks
+	// are enabled and the active cluster version is at least 24.3, use
+	// TableFormatPebblev5.
 	format := sstable.TableFormatPebblev4
+	if cs.Version.IsActive(ctx, clusterversion.V24_3) && columnarBlocksEnabled.Get(&cs.SV) {
+		format = sstable.TableFormatPebblev5
+	}
+
 	opts := DefaultPebbleOptions().MakeWriterOptions(0, format)
 	// By default, compress with the algorithm used for storage in a Pebble store.
 	// There are other, more specific, use cases that may call for a different

pkg/storage/sst_writer_test.go

@@ -87,6 +87,7 @@ func TestMakeIngestionWriterOptions(t *testing.T) {
 			st: func() *cluster.Settings {
 				st := cluster.MakeTestingClusterSettings()
 				IngestionValueBlocksEnabled.Override(context.Background(), &st.SV, true)
+				columnarBlocksEnabled.Override(context.Background(), &st.SV, false)
 				return st
 			}(),
 			want: want{
@@ -99,13 +100,40 @@ func TestMakeIngestionWriterOptions(t *testing.T) {
 			st: func() *cluster.Settings {
 				st := cluster.MakeTestingClusterSettings()
 				IngestionValueBlocksEnabled.Override(context.Background(), &st.SV, false)
+				columnarBlocksEnabled.Override(context.Background(), &st.SV, false)
 				return st
 			}(),
 			want: want{
 				format:             sstable.TableFormatPebblev4,
 				disableValueBlocks: true,
 			},
 		},
+		{
+			name: "enable columnar blocks",
+			st: func() *cluster.Settings {
+				st := cluster.MakeTestingClusterSettings()
+				IngestionValueBlocksEnabled.Override(context.Background(), &st.SV, false)
+				columnarBlocksEnabled.Override(context.Background(), &st.SV, true)
+				return st
+			}(),
+			want: want{
+				format:             sstable.TableFormatPebblev5,
+				disableValueBlocks: true,
+			},
+		},
+		{
+			name: "enable columnar blocks with value blocks",
+			st: func() *cluster.Settings {
+				st := cluster.MakeTestingClusterSettings()
+				IngestionValueBlocksEnabled.Override(context.Background(), &st.SV, true)
+				columnarBlocksEnabled.Override(context.Background(), &st.SV, true)
+				return st
+			}(),
+			want: want{
+				format:             sstable.TableFormatPebblev5,
+				disableValueBlocks: false,
+			},
+		},
 	}
 
 	for _, tc := range testCases {