From bfd1ad301f030b2e1f2eba7330eef09dba8d4e1b Mon Sep 17 00:00:00 2001
From: Jackson Owens <jackson@cockroachlabs.com>
Date: Fri, 8 Nov 2024 10:31:10 -0500
Subject: [PATCH] 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/storage/mvcc_history_test.go |  5 +++--
 pkg/storage/pebble_key_schema.go |  3 +++
 pkg/storage/sst_writer.go        | 12 ++++++++----
 3 files changed, 14 insertions(+), 6 deletions(-)

diff --git a/pkg/storage/mvcc_history_test.go b/pkg/storage/mvcc_history_test.go
index 67ddd32ee84f..9b04c5d40be8 100644
--- a/pkg/storage/mvcc_history_test.go
+++ b/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
 				}
diff --git a/pkg/storage/pebble_key_schema.go b/pkg/storage/pebble_key_schema.go
index e3a03de2ee4d..4b0a3567336a 100644
--- a/pkg/storage/pebble_key_schema.go
+++ b/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])
diff --git a/pkg/storage/sst_writer.go b/pkg/storage/sst_writer.go
index b35790f72427..8ab2a83b249f 100644
--- a/pkg/storage/sst_writer.go
+++ b/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