Merge #155292 #155537

155292: logical: external conn should not require db name r=msbutler a=kev-cao

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.

155537: sql/opt: fix RLS metadata clearing r=spilchen a=spilchen

This commit addresses two issues in the handling of RLS metadata in the sql/opt package:

- The RLS metadata was not being properly cleared after calling the Clear() function. This function is only used in test.
- Emitting RLS information during explain could theoretically result in a nil pointer dereference. While this path cannot be reached with a nil pointer under current conditions, the code was caught by an LLM. This change adds a defensive safeguard.

Closes #153192

Release note: none
Epic: none

Co-authored-by: Kevin Cao <39608887+kev-cao@users.noreply.github.com>
Co-authored-by: Matt Spilchen <matt.spilchen@cockroachlabs.com>

Author: 3 people

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)
+}

pkg/sql/opt/exec/explain/emit.go

@@ -1509,14 +1509,16 @@ func (e *emitter) emitPolicies(ob *OutputBuilder, table cat.Table, n *Node) {
 		ob.AddField("policies", "row-level security enabled, no policies applied.")
 	} else {
 		var sb strings.Builder
-		policies := table.Policies()
-		for _, grp := range [][]cat.Policy{policies.Permissive, policies.Restrictive} {
-			for _, policy := range grp {
-				if applied.Policies.Contains(policy.ID) {
-					if sb.Len() > 0 {
-						sb.WriteString(", ")
+		if table != nil {
+			policies := table.Policies()
+			for _, grp := range [][]cat.Policy{policies.Permissive, policies.Restrictive} {
+				for _, policy := range grp {
+					if applied.Policies.Contains(policy.ID) {
+						if sb.Len() > 0 {
+							sb.WriteString(", ")
+						}
+						sb.WriteString(policy.Name.Normalize())
 					}
-					sb.WriteString(policy.Name.Normalize())
 				}
 			}
 		}

pkg/sql/opt/memo/memo_test.go

@@ -672,7 +672,12 @@ func TestMemoIsStale(t *testing.T) {
 	stale()
 	evalCtx.SessionData().UserProto = oldUser
 	notStale()
+
+	// User changes (after RLS was reinitialized)
 	o.Memo().Metadata().ClearRLSEnabled()
+	evalCtx.SessionData().UserProto = newUser
+	notStale()
+	evalCtx.SessionData().UserProto = oldUser
 	notStale()
 
 	// Stale row_security.

pkg/sql/opt/row_level_security.go

@@ -48,7 +48,10 @@ func (r *RowLevelSecurityMeta) MaybeInit(user username.SQLUsername, hasAdminRole
 
 // Clear unsets the initialized property. This is used as a test helper.
 func (r *RowLevelSecurityMeta) Clear() {
-	r = &RowLevelSecurityMeta{}
+	if r == nil {
+		return
+	}
+	*r = RowLevelSecurityMeta{}
 }
 
 // AddTableUse indicates that an RLS-enabled table was encountered while