c2c: fix producer job proto

This patch fixes a papercut where we were storing pointers to spans in the
producer job proto. This isn't a correctness bug, as protocol buffers will
dereference pointers to objects during unmarshalling, but rather, it seems
dangerous to work with slices of pointers to spans.

Release note: None

Author: msbutler

pkg/ccl/streamingccl/streamproducer/producer_job.go

@@ -29,9 +29,9 @@ import (
 	"github.com/cockroachdb/errors"
 )
 
-func makeTenantSpan(tenantID uint64) *roachpb.Span {
+func makeTenantSpan(tenantID uint64) roachpb.Span {
 	prefix := keys.MakeTenantPrefix(roachpb.MustMakeTenantID(tenantID))
-	return &roachpb.Span{Key: prefix, EndKey: prefix.PrefixEnd()}
+	return roachpb.Span{Key: prefix, EndKey: prefix.PrefixEnd()}
 }
 
 func makeProducerJobRecord(
@@ -47,7 +47,7 @@ func makeProducerJobRecord(
 		Username:    user,
 		Details: jobspb.StreamReplicationDetails{
 			ProtectedTimestampRecordID: ptsID,
-			Spans:                      []*roachpb.Span{makeTenantSpan(tenantID)},
+			Spans:                      []roachpb.Span{makeTenantSpan(tenantID)},
 			TenantID:                   roachpb.MustMakeTenantID(tenantID),
 		},
 		Progress: jobspb.StreamReplicationProgress{

pkg/ccl/streamingccl/streamproducer/producer_job_test.go

@@ -160,7 +160,7 @@ func TestStreamReplicationProducerJob(t *testing.T) {
 		return insqlDB.Txn(ctx, func(
 			ctx context.Context, txn isql.Txn,
 		) error {
-			deprecatedTenantSpan := roachpb.Spans{*makeTenantSpan(30)}
+			deprecatedTenantSpan := roachpb.Spans{makeTenantSpan(30)}
 			tenantTarget := ptpb.MakeTenantsTarget([]roachpb.TenantID{roachpb.MustMakeTenantID(30)})
 			record := jobsprotectedts.MakeRecord(
 				ptsID, int64(jr.JobID), ts, deprecatedTenantSpan,

pkg/ccl/streamingccl/streamproducer/stream_lifetime.go

@@ -64,7 +64,7 @@ func startReplicationProducerJob(
 	statementTime := hlc.Timestamp{
 		WallTime: evalCtx.GetStmtTimestamp().UnixNano(),
 	}
-	deprecatedSpansToProtect := roachpb.Spans{*makeTenantSpan(tenantID)}
+	deprecatedSpansToProtect := roachpb.Spans{makeTenantSpan(tenantID)}
 	targetToProtect := ptpb.MakeTenantsTarget([]roachpb.TenantID{roachpb.MustMakeTenantID(tenantID)})
 	pts := jobsprotectedts.MakeRecord(ptsID, int64(jr.JobID), statementTime,
 		deprecatedSpansToProtect, jobsprotectedts.Jobs, targetToProtect)
@@ -236,12 +236,7 @@ func getReplicationStreamSpec(
 	if !ok {
 		return nil, errors.Errorf("job with id %d is not a replication stream job", streamID)
 	}
-	replicatedSpans := details.Spans
-	spans := make([]roachpb.Span, 0, len(replicatedSpans))
-	for _, span := range replicatedSpans {
-		spans = append(spans, *span)
-	}
-	spanPartitions, err := dsp.PartitionSpans(ctx, planCtx, spans)
+	spanPartitions, err := dsp.PartitionSpans(ctx, planCtx, details.Spans)
 	if err != nil {
 		return nil, err
 	}

pkg/jobs/jobspb/jobs.proto

@@ -184,7 +184,7 @@ message StreamIngestionProgress {
 
 message StreamReplicationDetails {
   // Key spans we are replicating
-  repeated roachpb.Span spans = 1;
+  repeated roachpb.Span spans = 1 [(gogoproto.nullable) = false];
 
   // ID of the protected timestamp record that protects the above spans
   bytes protected_timestamp_record_id = 2 [