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/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