logical: external conn should not require db name

LDR currently erroneously requires the db name in the external
connection with setting up a bidirectional connection. This was
determined to be caused by the fact that the privilege lookup on the
source cluster was using unqualified table names during the reverse
stream setup. This commit fixes the issue by using fully qualified table
names.

Fixes: #152395

Release note: LDR no longer requires the database name to be specified
in the external connection URI when setting up a bidirectional stream.

Author: kev-cao

pkg/crosscluster/logical/create_logical_replication_stmt.go

@@ -65,8 +65,8 @@ SELECT
 			id AS job_id,
 			crdb_internal.pb_to_json(
 				'cockroach.sql.jobs.jobspb.Payload',
-				payload)->'logicalReplicationDetails'->>'parentId' AS parent_id 
-		FROM crdb_internal.system_jobs 
+				payload)->'logicalReplicationDetails'->>'parentId' AS parent_id
+		FROM crdb_internal.system_jobs
 		WHERE job_type = 'LOGICAL REPLICATION'
 	) AS t
 	WHERE t.parent_id = $1
@@ -359,10 +359,12 @@ func (r *ResolvedDestObjects) TargetDescription() string {
 	return targetDescription
 }
 
+// TargetTableNames returns the fully qualified names of the resolved target
+// tables.
 func (r *ResolvedDestObjects) TargetTableNames() []string {
-	var targetTableNames []string
+	targetTableNames := make([]string, len(r.TableNames))
 	for i := range r.TableNames {
-		targetTableNames = append(targetTableNames, r.TableNames[i].Table())
+		targetTableNames[i] = r.TableNames[i].FQString()
 	}
 	return targetTableNames
 }

pkg/crosscluster/logical/logical_replication_job_test.go

@@ -1727,12 +1727,12 @@ func GetReverseJobID(
 	var jobID jobspb.JobID
 	testutils.SucceedsSoon(t, func() error {
 		err := db.DB.QueryRowContext(ctx, `
-            SELECT id 
-            FROM system.jobs 
-            WHERE job_type = 'LOGICAL REPLICATION' 
+            SELECT id
+            FROM system.jobs
+            WHERE job_type = 'LOGICAL REPLICATION'
             AND id != $1
             AND created > $2
-            ORDER BY created DESC 
+            ORDER BY created DESC
             LIMIT 1`,
 			parentID, created).Scan(&jobID)
 		if err != nil {
@@ -2905,3 +2905,46 @@ func TestGetWriterType(t *testing.T) {
 		require.Equal(t, sqlclustersettings.LDRWriterTypeSQL, wt)
 	})
 }
+
+func TestLogicalReplicationExternalConnWithoutDBName(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	ctx := context.Background()
+
+	server, s, dbA, dbB := setupLogicalTestServer(t, ctx, testClusterBaseClusterArgs, 1)
+	defer server.Stopper().Stop(ctx)
+
+	dbA.Exec(t, "CREATE TABLE a.public.foo (x INT PRIMARY KEY)")
+	dbA.Exec(t, "INSERT INTO a.public.foo SELECT * FROM generate_series(1, 10)")
+	dbA.Exec(t, "CREATE USER userA WITH PASSWORD '123'")
+	dbA.Exec(t, "GRANT REPLICATIONSOURCE, REPLICATIONDEST ON TABLE a.public.foo TO userA")
+	dbAURL := replicationtestutils.GetExternalConnectionURI(
+		t, s, s, serverutils.ClientCerts(false), serverutils.UserPassword("userA", "123"),
+	)
+
+	dbB.Exec(t, "CREATE USER userB WITH PASSWORD '123'")
+	dbB.Exec(t, "GRANT CREATE ON DATABASE b TO userB")
+	dbBURL := replicationtestutils.GetExternalConnectionURI(
+		t, s, s, serverutils.ClientCerts(false), serverutils.UserPassword("userB", "123"),
+	)
+
+	dbBAsUser := sqlutils.MakeSQLRunner(s.SQLConn(
+		t,
+		serverutils.DBName("b"),
+		serverutils.ClientCerts(false),
+		serverutils.UserPassword("userB", "123"),
+	))
+
+	var jobID jobspb.JobID
+	dbBAsUser.QueryRow(
+		t,
+		"CREATE LOGICALLY REPLICATED TABLE b.public.foo FROM TABLE a.public.foo ON $1 WITH BIDIRECTIONAL ON $2",
+		dbAURL.String(),
+		dbBURL.String(),
+	).Scan(&jobID)
+	WaitUntilReplicatedTime(t, s.Clock().Now(), dbB, jobID)
+
+	reverseJobID := GetReverseJobID(ctx, t, dbA, jobID)
+	WaitUntilReplicatedTime(t, s.Clock().Now(), dbA, reverseJobID)
+}