release-23.2: sql/sem/builtins: validate engine keys in storage builtins

Ensure that the user-provided engine keys are valid before using them to seek,
in storage's SQL builtins.  The Comparer will now panic if there's an attempt
to compare using an invalid key. These built-ins now attempt to decode and
validate the keys as engine keys. If they validate, it uses them as-is.
Otherwise it encodes them as version-less keys.

Epic: none
Fix cockroachdb#128757.
Release note: none

pkg/sql/sem/builtins/BUILD.bazel

@@ -99,6 +99,7 @@ go_library(
         "//pkg/sql/storageparam/indexstorageparam",
         "//pkg/sql/syntheticprivilege",
         "//pkg/sql/types",
+        "//pkg/storage",
         "//pkg/storage/enginepb",
         "//pkg/util",
         "//pkg/util/arith",

pkg/sql/sem/builtins/builtins.go

@@ -76,6 +76,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlstats/persistedsqlstats/sqlstatsutil"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqltelemetry"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
+	"github.com/cockroachdb/cockroach/pkg/storage"
 	"github.com/cockroachdb/cockroach/pkg/util"
 	"github.com/cockroachdb/cockroach/pkg/util/bitarray"
 	"github.com/cockroachdb/cockroach/pkg/util/duration"
@@ -5234,6 +5235,7 @@ DO NOT USE -- USE 'CREATE VIRTUAL CLUSTER' INSTEAD`,
 			Volatility: volatility.Stable,
 		},
 	),
+
 	"crdb_internal.redact_descriptor": makeBuiltin(
 		tree.FunctionProperties{
 			Category:         builtinconstants.CategorySystemInfo,
@@ -6976,6 +6978,14 @@ Parameters:` + randgencfg.ConfigDoc,
 				storeID := int32(tree.MustBeDInt(args[1]))
 				startKey := []byte(tree.MustBeDBytes(args[2]))
 				endKey := []byte(tree.MustBeDBytes(args[3]))
+
+				if ek, ok := storage.DecodeEngineKey(startKey); !ok || ek.Validate() != nil {
+					startKey = storage.EncodeMVCCKey(storage.MVCCKey{Key: startKey})
+				}
+				if ek, ok := storage.DecodeEngineKey(endKey); !ok || ek.Validate() != nil {
+					endKey = storage.EncodeMVCCKey(storage.MVCCKey{Key: endKey})
+				}
+				log.Infof(ctx, "crdb_internal.compact_engine_span called for nodeID=%d, storeID=%d, range[startKey=%s, endKey=%s]", nodeID, storeID, startKey, endKey)
 				if err := evalCtx.CompactEngineSpan(
 					ctx, nodeID, storeID, startKey, endKey); err != nil {
 					return nil, err

pkg/sql/sem/builtins/generator_builtins.go

@@ -37,6 +37,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/sessiondata"
 	"github.com/cockroachdb/cockroach/pkg/sql/syntheticprivilege"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
+	"github.com/cockroachdb/cockroach/pkg/storage"
 	"github.com/cockroachdb/cockroach/pkg/storage/enginepb"
 	"github.com/cockroachdb/cockroach/pkg/util/arith"
 	"github.com/cockroachdb/cockroach/pkg/util/duration"
@@ -3297,7 +3298,17 @@ func makeTableMetricsGenerator(
 	storeID := int32(tree.MustBeDInt(args[1]))
 	start := []byte(tree.MustBeDBytes(args[2]))
 	end := []byte(tree.MustBeDBytes(args[3]))
-
+	// We use the keys as-is if they are valid engine keys, otherwise we encode
+	// them as a version-less key. This preserves the ability to pass in
+	// manually constructed engine keys, but also allows for the common practice
+	// of passing in keys extracted from the ranges table. It's possible
+	// (although somewhat unlikely) for a user key to validate as an engine key.
+	if ek, ok := storage.DecodeEngineKey(start); !ok || ek.Validate() != nil {
+		start = storage.EncodeMVCCKey(storage.MVCCKey{Key: start})
+	}
+	if ek, ok := storage.DecodeEngineKey(end); !ok || ek.Validate() != nil {
+		end = storage.EncodeMVCCKey(storage.MVCCKey{Key: start})
+	}
 	return newTableMetricsIterator(evalCtx, nodeID, storeID, start, end), nil
 }
 
@@ -3404,6 +3415,17 @@ func makeStorageInternalKeysGenerator(
 	storeID := int32(tree.MustBeDInt(args[1]))
 	start := []byte(tree.MustBeDBytes(args[2]))
 	end := []byte(tree.MustBeDBytes(args[3]))
+	// We use the keys as-is if they are valid engine keys, otherwise we encode
+	// them as a version-less key. This preserves the ability to pass in
+	// manually constructed engine keys, but also allows for the common practice
+	// of passing in keys extracted from the ranges table. It's possible
+	// (although somewhat unlikely) for a user key to validate as an engine key.
+	if ek, ok := storage.DecodeEngineKey(start); !ok || ek.Validate() != nil {
+		start = storage.EncodeMVCCKey(storage.MVCCKey{Key: start})
+	}
+	if ek, ok := storage.DecodeEngineKey(end); !ok || ek.Validate() != nil {
+		end = storage.EncodeMVCCKey(storage.MVCCKey{Key: end})
+	}
 
 	var megabytesPerSecond int64
 	if len(args) > 4 {

pkg/storage/pebble.go

@@ -41,7 +41,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util"
 	"github.com/cockroachdb/cockroach/pkg/util/buildutil"
 	"github.com/cockroachdb/cockroach/pkg/util/envutil"
-	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/humanizeutil"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/protoutil"
@@ -2451,9 +2450,20 @@ func (p *Pebble) Compact() error {
 
 // CompactRange implements the Engine interface.
 func (p *Pebble) CompactRange(start, end roachpb.Key) error {
-	bufStart := EncodeMVCCKey(MVCCKey{start, hlc.Timestamp{}})
-	bufEnd := EncodeMVCCKey(MVCCKey{end, hlc.Timestamp{}})
-	return p.db.Compact(bufStart, bufEnd, true /* parallel */)
+	// TODO(jackson): Consider changing Engine.CompactRange's signature to take
+	// in EngineKeys so that it's unambiguous that the arguments have already
+	// been encoded as engine keys. We do need to encode these keys in protocol
+	// buffers when they're sent over the wire during the
+	// crdb_internal.compact_engine_span builtin. Maybe we should have a
+	// roachpb.Key equivalent for EngineKey so we don't lose that type
+	// information?
+	if ek, ok := DecodeEngineKey(start); !ok || ek.Validate() != nil {
+		return errors.Errorf("invalid start key: %q", start)
+	}
+	if ek, ok := DecodeEngineKey(end); !ok || ek.Validate() != nil {
+		return errors.Errorf("invalid end key: %q", end)
+	}
+	return p.db.Compact(start, end, true /* parallel */)
 }
 
 // RegisterFlushCompletedCallback implements the Engine interface.