crdb_internal: filter in replication spans view

Release note: none.
Epic: CRDB-51855.

Author: dt

pkg/crosscluster/physical/alter_replication_job_test.go

@@ -533,6 +533,10 @@ func TestTenantReplicationStatus(t *testing.T) {
 	_, status, err = getReplicationStatsAndStatus(ctx, registry, nil, jobspb.JobID(producerJobID))
 	require.ErrorContains(t, err, "is not a stream ingestion job")
 	require.Equal(t, "replication error", status)
+	c.DestSysSQL.CheckQueryResults(t, "SELECT count(*) > 0 FROM crdb_internal.cluster_replication_spans", [][]string{{"true"}})
+	c.DestSysSQL.Exec(t, fmt.Sprintf("CREATE USER %s", username.TestUser))
+	noPrivs := sqlutils.MakeSQLRunner(c.DestSysServer.SQLConn(t, serverutils.User(username.TestUser)))
+	noPrivs.CheckQueryResults(t, "SELECT count(*) > 0 FROM crdb_internal.cluster_replication_spans", [][]string{{"false"}})
 }
 
 // TestAlterTenantHandleFutureProtectedTimestamp verifies that cutting over "TO

pkg/sql/crdb_internal.go

@@ -9290,7 +9290,7 @@ var crdbInternalClusterReplicationResolvedView = virtualSchemaView{
 			SELECT
 				j.id AS job_id, jsonb_array_elements(crdb_internal.pb_to_json('progress', i.value)->'streamIngest'->'checkpoint'->'resolvedSpans') AS s
 			FROM system.jobs j LEFT JOIN system.job_info i ON j.id = i.job_id AND i.info_key = 'legacy_progress'
-			WHERE j.job_type = 'REPLICATION STREAM INGESTION'
+			WHERE j.job_type = 'REPLICATION STREAM INGESTION' AND pg_has_role(current_user, 'admin', 'member')
 			) SELECT
 				job_id,
 				crdb_internal.pretty_key(decode(s->'span'->>'key', 'base64'), 0) AS start_key,
@@ -9313,7 +9313,7 @@ var crdbInternalLogicalReplicationResolvedView = virtualSchemaView{
 			SELECT
 				j.id AS job_id, jsonb_array_elements(crdb_internal.pb_to_json('progress', i.value)->'LogicalReplication'->'checkpoint'->'resolvedSpans') AS s
 			FROM system.jobs j LEFT JOIN system.job_info i ON j.id = i.job_id AND i.info_key = 'legacy_progress'
-			WHERE j.job_type = 'LOGICAL REPLICATION'
+			WHERE j.job_type = 'LOGICAL REPLICATION' AND pg_has_role(current_user, 'admin', 'member')
 			) SELECT
 				job_id,
 				crdb_internal.pretty_key(decode(s->'span'->>'key', 'base64'), 0) AS start_key,