crosscluster/logical: auth PlanLogicalReplication with job id

This patch simplifies the implementation of source side table level auth
checks.

Epic: none

Release note: none

Author: msbutler

pkg/crosscluster/logical/logical_replication_job.go

@@ -488,6 +488,7 @@ func (p *logicalReplicationPlanner) generatePlanImpl(
 		// During an offline initial scan, we need to replicate the whole table, not
 		// just the primary keys.
 		UseTableSpan: payload.CreateTable && progress.ReplicatedTime.IsEmpty(),
+		StreamID:     streampb.StreamID(payload.StreamID),
 	}
 	for _, pair := range payload.ReplicationPairs {
 		req.TableIDs = append(req.TableIDs, pair.SrcDescriptorID)

pkg/repstream/streampb/stream.proto

@@ -170,6 +170,7 @@ message LogicalReplicationPlanRequest {
     // PlanAsOf is the time as of which the plan should be produced.
     util.hlc.Timestamp plan_as_of = 2 [(gogoproto.nullable) = false];
     bool use_table_span = 3;
+    int64 stream_id = 4 [(gogoproto.customname) = "StreamID", (gogoproto.casttype) = "StreamID"];
 }
 
 // SourcePartition contains per partition information for a replication plan.

pkg/sql/sem/builtins/replication_builtins.go

@@ -511,7 +511,14 @@ var replicationBuiltins = map[string]builtinDefinition{
 				if err := protoutil.Unmarshal(reqBytes, &req); err != nil {
 					return nil, err
 				}
-				if err := mgr.AuthorizeViaReplicationPriv(ctx); err != nil {
+				if req.StreamID != 0 {
+					if err := mgr.AuthorizeViaJob(ctx, req.StreamID); err != nil {
+						return nil, err
+					}
+					// Auth via replication priv exists to ensure a user that planned their
+					// job pre 25.2, which will not send a stream id, can still plan their
+					// distsql flow.
+				} else if err := mgr.AuthorizeViaReplicationPriv(ctx); err != nil {
 					return nil, err
 				}