opt: check privileges after data sources in memo staleness check

Previously, the `opt.Metadata` staleness check would check the access
privileges for a data source immediately after checking that the data
source still resolves as it originally did. This meant that the privilege
check for one data source could occur before checking staleness of other
data sources. Before #96045 this wasn't a problem, since these checks
would happen in the order in which the data sources were resolved. But
after #96045, the data sources were stored in a map instead of a slice,
which made the order in which the checks are performed nondeterministic.

This change in behavior can cause queries to return insufficient privilege
errors even when the current user has access privileges on all data sources
referenced by the query. This is because the query may *implicitly*
reference a table by ID if there is a foreign key relation from an explicitly
referenced data source. If the database is changed, this ID reference will
still resolve the same, but if the user is changed to one without privileges
on that table, the staleness check will result in an error. This wasn't a
problem before because the referencing table would always be checked first,
causing the staleness check to finish without an error.

This patch fixes the bug by moving the privilege checks until after all
data source resolution checks have succeeded.

Fixes #102375

Release note (bug fix): Fixed a bug introduced in versions 22.1.19,
22.2.8, and pre-release versions of 23.1 that could cause queries to
return spurious insufficient privilege errors. For the bug to occur,
two databases would need to have duplicate tables each with a foreign key
reference to another table. The error would then occur if the same SQL
string was executed against both databases concurrently by users that have
privileges over only one of the tables.

pkg/sql/logictest/testdata/logic_test/schema

@@ -1259,3 +1259,68 @@ SELECT abs(1);
 1
 
 subtest end
+
+# Regression test for #102375 - check that all data sources resolve to the same
+# schema objects before checking access privileges.
+subtest privileges
+
+user root
+
+statement ok
+RESET search_path;
+CREATE DATABASE testdb1;
+USE testdb1;
+CREATE TABLE ab (a INT PRIMARY KEY, b INT);
+INSERT INTO ab VALUES (1, 1);
+CREATE TABLE xy (x INT, y INT, FOREIGN KEY (x) REFERENCES ab(a));
+GRANT ALL ON xy TO testuser;
+GRANT ALL ON ab TO testuser;
+
+statement ok
+CREATE DATABASE testdb2;
+USE testdb2;
+CREATE TABLE ab (a INT PRIMARY KEY, b INT);
+INSERT INTO ab VALUES (1, 1);
+CREATE TABLE xy (x INT, y INT, FOREIGN KEY (x) REFERENCES ab(a));
+CREATE USER testuser2;
+GRANT ALL ON xy TO testuser2;
+GRANT ALL ON ab TO testuser2;
+
+user testuser
+
+statement ok
+USE testdb1
+
+statement ok
+INSERT into xy VALUES(1, 1)
+
+user testuser2
+
+statement ok
+USE testdb2
+
+statement ok
+INSERT into xy VALUES(1, 1)
+
+statement ok
+INSERT into xy VALUES(1, 1)
+
+user testuser
+
+statement ok
+USE testdb1
+
+statement ok
+INSERT into xy VALUES(1, 1)
+
+user root
+
+statement ok
+USE test;
+REVOKE ALL ON testdb2.xy FROM testuser2;
+REVOKE ALL ON testdb2.ab FROM testuser2;
+DROP USER testuser2;
+DROP DATABASE testdb1 CASCADE;
+DROP DATABASE testdb2 CASCADE;
+
+subtest end

pkg/sql/opt/memo/memo_test.go

@@ -360,6 +360,15 @@ func TestMemoIsStale(t *testing.T) {
 	evalCtx.SessionData().OptimizerHoistUncorrelatedEqualitySubqueries = false
 	notStale()
 
+	// User no longer has access to view.
+	catalog.View(tree.NewTableNameWithSchema("t", tree.PublicSchemaName, "abcview")).Revoked = true
+	_, err = o.Memo().IsStale(ctx, &evalCtx, catalog)
+	if exp := "user does not have privilege"; !testutils.IsError(err, exp) {
+		t.Fatalf("expected %q error, but got %+v", exp, err)
+	}
+	catalog.View(tree.NewTableNameWithSchema("t", tree.PublicSchemaName, "abcview")).Revoked = false
+	notStale()
+
 	// Stale data sources and schema. Create new catalog so that data sources are
 	// recreated and can be modified independently.
 	catalog = testcat.New()
@@ -372,15 +381,6 @@ func TestMemoIsStale(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	// User no longer has access to view.
-	catalog.View(tree.NewTableNameWithSchema("t", tree.PublicSchemaName, "abcview")).Revoked = true
-	_, err = o.Memo().IsStale(ctx, &evalCtx, catalog)
-	if exp := "user does not have privilege"; !testutils.IsError(err, exp) {
-		t.Fatalf("expected %q error, but got %+v", exp, err)
-	}
-	catalog.View(tree.NewTableNameWithSchema("t", tree.PublicSchemaName, "abcview")).Revoked = false
-	notStale()
-
 	// Table ID changes.
 	catalog.Table(tree.NewTableNameWithSchema("t", tree.PublicSchemaName, "abc")).TabID = 1
 	stale()

pkg/sql/opt/metadata.go

@@ -393,10 +393,11 @@ func (md *Metadata) CheckDependencies(
 				return false, maybeSwallowMetadataResolveErr(err)
 			}
 		}
-		// Ensure that all required privileges for the data source are still valid.
-		if err := md.checkDataSourcePrivileges(ctx, optCatalog, toCheck); err != nil {
-			return false, err
-		}
+	}
+
+	// Ensure that all required privileges for the data sources are still valid.
+	if err := md.checkDataSourcePrivileges(ctx, optCatalog); err != nil {
+		return false, err
 	}
 
 	// Check that no referenced user defined types have changed.
@@ -481,27 +482,24 @@ func maybeSwallowMetadataResolveErr(err error) error {
 }
 
 // checkDataSourcePrivileges checks that none of the privileges required by the
-// query for the given data source have been revoked.
-func (md *Metadata) checkDataSourcePrivileges(
-	ctx context.Context, optCatalog cat.Catalog, dataSource cat.DataSource,
-) error {
-	if dataSource == nil {
-		panic(errors.AssertionFailedf("expected data source for privilege check to be non-nil"))
-	}
-	privileges := md.privileges[dataSource.ID()]
-	for privs := privileges; privs != 0; {
-		// Strip off each privilege bit and make call to CheckPrivilege for it.
-		// Note that priv == 0 can occur when a dependency was added with
-		// privilege.Kind = 0 (e.g. for a table within a view, where the table
-		// privileges do not need to be checked). Ignore the "zero privilege".
-		priv := privilege.Kind(bits.TrailingZeros32(uint32(privs)))
-		if priv != 0 {
-			if err := optCatalog.CheckPrivilege(ctx, dataSource, priv); err != nil {
-				return err
+// query for the referenced data sources have been revoked.
+func (md *Metadata) checkDataSourcePrivileges(ctx context.Context, optCatalog cat.Catalog) error {
+	for _, dataSource := range md.dataSourceDeps {
+		privileges := md.privileges[dataSource.ID()]
+		for privs := privileges; privs != 0; {
+			// Strip off each privilege bit and make call to CheckPrivilege for it.
+			// Note that priv == 0 can occur when a dependency was added with
+			// privilege.Kind = 0 (e.g. for a table within a view, where the table
+			// privileges do not need to be checked). Ignore the "zero privilege".
+			priv := privilege.Kind(bits.TrailingZeros32(uint32(privs)))
+			if priv != 0 {
+				if err := optCatalog.CheckPrivilege(ctx, dataSource, priv); err != nil {
+					return err
+				}
 			}
+			// Set the just-handled privilege bit to zero and look for next.
+			privs &= ^(1 << priv)
 		}
-		// Set the just-handled privilege bit to zero and look for next.
-		privs &= ^(1 << priv)
 	}
 	return nil
 }