ptstorage: change Protect and GetRecord to work with target column

Protected timestamp records now apply to a `ptpb.Target` instead
of `roachpb.Spans`. Protect will now write the `Target` on the record to the
`target` column in the system.pts_records table, and `GetRecord` will read
the `target` column when constructing a `ptpb.Record` to return to the
caller.

If we are running in a mixed version state where the relevant migration
to the system.pts_records table has not run, we use the existing logic
to protect Spans.

Once the migration has run, all calls to protect must
have a non-nil `Target` field on the record. While the new subsystem is
under development, we write both the `Spans` and the `Target` post migration.
Once the GC queue has been taught to use the `Target` field when making GC decisions,
we will write an empty `Spans` field in the `Protect` method, before persisting
state in the system table. All new records will no longer be reliant on
the Spans column, paving the way for deleting all Span related logic
in a future release.

This change tweaks the tests in `ptstorage_test` to populate the `Target`
field. All other test changes in `ptverifier`, `ptcache`, `kvserver` are
only to get past the "target is nil" validation step. These tests will
be changed as and when we develop those pieces of the subsystem.

Informs: cockroachdb#73727

Release note: None

pkg/ccl/backupccl/backup_test.go

@@ -43,6 +43,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/cloud"
 	"github.com/cockroachdb/cockroach/pkg/cloud/amazon"
 	_ "github.com/cockroachdb/cockroach/pkg/cloud/impl" // register cloud storage providers
+	"github.com/cockroachdb/cockroach/pkg/clusterversion"
 	"github.com/cockroachdb/cockroach/pkg/config"
 	"github.com/cockroachdb/cockroach/pkg/config/zonepb"
 	"github.com/cockroachdb/cockroach/pkg/jobs"
@@ -53,6 +54,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/security"
+	"github.com/cockroachdb/cockroach/pkg/server"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
@@ -6742,16 +6744,28 @@ func TestRestoreErrorPropagates(t *testing.T) {
 
 // TestProtectedTimestampsFailDueToLimits ensures that when creating a protected
 // timestamp record fails, we return the correct error.
+//
+// TODO(adityamaru): Remove in 22.2 once no records protect spans.
 func TestProtectedTimestampsFailDueToLimits(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
 	ctx := context.Background()
 	dir, dirCleanupFn := testutils.TempDir(t)
 	defer dirCleanupFn()
-	params := base.TestClusterArgs{}
-	params.ServerArgs.ExternalIODir = dir
-	tc := testcluster.StartTestCluster(t, 1, params)
+	clusterArgs := base.TestClusterArgs{
+		ServerArgs: base.TestServerArgs{
+			Knobs: base.TestingKnobs{
+				Server: &server.TestingKnobs{
+					DisableAutomaticVersionUpgrade: 1,
+					BinaryVersionOverride: clusterversion.ByKey(
+						clusterversion.AlterSystemProtectedTimestampAddColumn - 1),
+				},
+			},
+		},
+	}
+	clusterArgs.ServerArgs.ExternalIODir = dir
+	tc := testcluster.StartTestCluster(t, 1, clusterArgs)
 	defer tc.Stopper().Stop(ctx)
 	db := tc.ServerConn(0)
 	runner := sqlutils.MakeSQLRunner(db)

pkg/jobs/jobsprotectedts/jobs_protected_ts_test.go

@@ -79,7 +79,7 @@ func TestJobsProtectedTimestamp(t *testing.T) {
 			if j, err = jr.CreateJobWithTxn(ctx, mkJobRec(), jobID, txn); err != nil {
 				return err
 			}
-			deprecatedSpansToProtect := roachpb.Spans{{keys.MinKey, keys.MaxKey}}
+			deprecatedSpansToProtect := roachpb.Spans{{Key: keys.MinKey, EndKey: keys.MaxKey}}
 			targetToProtect := ptpb.MakeRecordClusterTarget()
 			rec = jobsprotectedts.MakeRecord(uuid.MakeV4(), int64(jobID), ts,
 				deprecatedSpansToProtect, jobsprotectedts.Jobs, targetToProtect)
@@ -165,7 +165,7 @@ func TestSchedulesProtectedTimestamp(t *testing.T) {
 		require.NoError(t, s0.DB().Txn(ctx, func(ctx context.Context, txn *kv.Txn) (err error) {
 			sj = mkScheduledJobRec(scheduleLabel)
 			require.NoError(t, sj.Create(ctx, s0.InternalExecutor().(sqlutil.InternalExecutor), txn))
-			deprecatedSpansToProtect := roachpb.Spans{{keys.MinKey, keys.MaxKey}}
+			deprecatedSpansToProtect := roachpb.Spans{{Key: keys.MinKey, EndKey: keys.MaxKey}}
 			targetToProtect := ptpb.MakeRecordClusterTarget()
 			rec = jobsprotectedts.MakeRecord(uuid.MakeV4(), sj.ScheduleID(), ts,
 				deprecatedSpansToProtect, jobsprotectedts.Schedules, targetToProtect)

pkg/kv/kvserver/client_protectedts_test.go

@@ -140,6 +140,7 @@ func TestProtectedTimestamps(t *testing.T) {
 	pts := ptstorage.New(s0.ClusterSettings(), s0.InternalExecutor().(*sql.InternalExecutor))
 	ptsWithDB := ptstorage.WithDatabase(pts, s0.DB())
 	startKey := getTableStartKey("foo")
+	unusedTarget := ptpb.MakeRecordClusterTarget()
 	ptsRec := ptpb.Record{
 		ID:        uuid.MakeV4().GetBytes(),
 		Timestamp: s0.Clock().Now(),
@@ -150,6 +151,7 @@ func TestProtectedTimestamps(t *testing.T) {
 				EndKey: startKey.PrefixEnd(),
 			},
 		},
+		Target: unusedTarget,
 	}
 	require.NoError(t, ptsWithDB.Protect(ctx, nil /* txn */, &ptsRec))
 	upsertUntilBackpressure()

pkg/kv/kvserver/client_replica_test.go

@@ -3372,10 +3372,12 @@ func TestStrictGCEnforcement(t *testing.T) {
 		tableKey      = keys.SystemSQLCodec.TablePrefix(tableID)
 		tableSpan     = roachpb.Span{Key: tableKey, EndKey: tableKey.PrefixEnd()}
 		mkRecord      = func() ptpb.Record {
+			unusedTarget := ptpb.MakeRecordClusterTarget()
 			return ptpb.Record{
 				ID:              uuid.MakeV4().GetBytes(),
 				Timestamp:       tenSecondsAgo.Add(-10*time.Second.Nanoseconds(), 0),
 				DeprecatedSpans: []roachpb.Span{tableSpan},
+				Target:          unusedTarget,
 			}
 		}
 		mkStaleTxn = func() *kv.Txn {

pkg/kv/kvserver/protectedts/ptcache/cache_test.go

@@ -430,11 +430,13 @@ func protect(
 	t *testing.T, s serverutils.TestServerInterface, p protectedts.Storage, spans ...roachpb.Span,
 ) (r *ptpb.Record, createdAt hlc.Timestamp) {
 	protectTS := s.Clock().Now()
+	unusedTarget := ptpb.MakeRecordClusterTarget()
 	r = &ptpb.Record{
 		ID:              uuid.MakeV4().GetBytes(),
 		Timestamp:       protectTS,
 		Mode:            ptpb.PROTECT_AFTER,
 		DeprecatedSpans: spans,
+		Target:          unusedTarget,
 	}
 	ctx := context.Background()
 	txn := s.DB().NewTxn(ctx, "test")

pkg/kv/kvserver/protectedts/ptreconcile/reconciler_test.go

@@ -79,6 +79,7 @@ func TestReconciler(t *testing.T) {
 		Mode:      ptpb.PROTECT_AFTER,
 		MetaType:  testTaskType,
 		Meta:      []byte(recMeta),
+		Target:    ptpb.MakeRecordClusterTarget(),
 		DeprecatedSpans: []roachpb.Span{
 			{Key: keys.MinKey, EndKey: keys.MaxKey},
 		},

pkg/kv/kvserver/protectedts/ptstorage/BUILD.bazel

@@ -13,6 +13,7 @@ go_library(
     importpath = "github.com/cockroachdb/cockroach/pkg/kv/kvserver/protectedts/ptstorage",
     visibility = ["//visibility:public"],
     deps = [
+        "//pkg/clusterversion",
         "//pkg/kv",
         "//pkg/kv/kvserver/protectedts",
         "//pkg/kv/kvserver/protectedts/ptpb",
@@ -40,6 +41,7 @@ go_test(
     embed = [":ptstorage"],
     deps = [
         "//pkg/base",
+        "//pkg/clusterversion",
         "//pkg/keys",
         "//pkg/kv",
         "//pkg/kv/kvserver/protectedts",
@@ -48,12 +50,15 @@ go_test(
         "//pkg/security",
         "//pkg/security/securitytest",
         "//pkg/server",
+        "//pkg/settings/cluster",
         "//pkg/sql",
         "//pkg/sql/catalog",
         "//pkg/sql/catalog/colinfo",
+        "//pkg/sql/catalog/descpb",
         "//pkg/sql/sem/tree",
         "//pkg/sql/sessiondata",
         "//pkg/sql/sqlutil",
+        "//pkg/sql/tests",
         "//pkg/testutils",
         "//pkg/testutils/serverutils",
         "//pkg/testutils/testcluster",

pkg/kv/kvserver/protectedts/ptstorage/sql.go

@@ -35,6 +35,18 @@ WITH
     checks AS (` + protectChecksCTE + `),
     updated_meta AS (` + protectUpsertMetaCTE + `),
     new_record AS (` + protectInsertRecordCTE + `)
+SELECT
+    failed,
+    version AS prev_version
+FROM
+    checks, current_meta;`
+
+	protectQueryWithoutTarget = `
+WITH
+    current_meta AS (` + currentMetaCTE + `),
+    checks AS (` + protectChecksCTE + `),
+    updated_meta AS (` + protectUpsertMetaCTE + `),
+    new_record AS (` + protectInsertRecordWithoutTargetCTE + `)
 SELECT
     failed,
     num_spans AS prev_spans,
@@ -84,6 +96,20 @@ RETURNING
 
 	protectInsertRecordCTE = `
 INSERT
+INTO
+    system.protected_ts_records (id, ts, meta_type, meta, num_spans, spans, target)
+(
+    SELECT
+        $4, $5, $6, $7, $8, $9, $10
+    WHERE
+        NOT EXISTS(SELECT * FROM checks WHERE failed)
+)
+RETURNING
+    id
+`
+
+	protectInsertRecordWithoutTargetCTE = `
+INSERT
 INTO
     system.protected_ts_records (id, ts, meta_type, meta, num_spans, spans)
 (
@@ -96,12 +122,23 @@ RETURNING
     id
 `
 
-	getRecordsQueryBase = `
+	getRecordsWithoutTargetQueryBase = `
 SELECT
     id, ts, meta_type, meta, spans, verified
 FROM
     system.protected_ts_records`
 
+	getRecordsWithoutTargetQuery = getRecordsWithoutTargetQueryBase + ";"
+	getRecordWithoutTargetQuery  = getRecordsWithoutTargetQueryBase + `
+WHERE
+    id = $1;`
+
+	getRecordsQueryBase = `
+SELECT
+    id, ts, meta_type, meta, spans, verified, target
+FROM
+    system.protected_ts_records`
+
 	getRecordsQuery = getRecordsQueryBase + ";"
 	getRecordQuery  = getRecordsQueryBase + `
 WHERE