[BACKPORT 2024.2][#25392] YSQL: make alter database statement non-glo…

…bal impact

Summary:
Currently alter database statement has global impact when incrementing catalog
version because it writes pg_database which is a shared relation. As a result it
can cause symptom such as:

Scheduled run_maintenance job for PG Partman failed on one database while a
backup was being restored on another.

The error looks like
```
2024-12-16 16:54:39.544 UTC [101015] ERROR:  The catalog snapshot used for this transaction has been invalidated: expected: 124, got: 123: MISMATCHED_SCHEMA
```

The above error was caused by a global-impact alter database DDL:

```
ALTER DATABASE "db_timestamp_1" OWNER TO postgres
```

In reality, there is no need for the alter database statement to increment the
catalog version of all databases. There is only one database that is altered and
we should increment the catalog version of "db_timestamp_1", which may not be
MyDatabaseId (OID of the database that the current session is connected to).

This diff makes an optimization to only increment the catalog version of the
database that is being altered for a set of common alter database statements.

Note that this is an optimization in the sense that it is not incorrect to
increment catalog versions of all the databases: in that case we will be
rejecting more queries than necessary but will not wrongly accept more queries.

Added new unit tests that do some sanity tests to show the same behavior as
native PG (except when more than one nodes are involved).
Jira: DB-14621

Original commit: 3b31c0b / D40959

Test Plan:
./yb_build.sh --java-test 'org.yb.pgsql.TestPgCatalogConsistency#testAlterDatabaseSetResetGucTestDb'
./yb_build.sh --java-test 'org.yb.pgsql.TestPgCatalogConsistency#testAlterDatabaseSetResetGucDefaultDb'
./yb_build.sh --cxx-test pg_catalog_version-test

Reviewers: kfranz, mihnea, loginov, #db-approvers

Reviewed By: kfranz, #db-approvers

Subscribers: yql

Differential Revision: https://phorge.dev.yugabyte.com/D41212

java/yb-pgsql/src/test/java/org/yb/pgsql/TestPgCatalogConsistency.java

@@ -148,4 +148,63 @@ public void testAlterTable() throws Exception {
       assertGreaterThan(catalogCheckSuccessfulCycles.get(), 0);
     }
   }
+
+  public void testAlterDatabaseSetResetGucHelper(String database) throws Exception {
+    try (Statement statement = connection.createStatement()) {
+      if (!database.equals(DEFAULT_PG_DATABASE)) {
+        statement.execute("CREATE DATABASE tdb");
+      }
+      statement.execute("ALTER DATABASE " + database + " SET temp_file_limit = 4096");
+      waitForTServerHeartbeat();
+
+      // Test 2 new connections.
+      Connection connection1 = getConnectionBuilder().withDatabase(database).connect();
+      Statement statement1 = connection1.createStatement();
+      assertQuery(statement1, "SHOW temp_file_limit", new Row("4MB"));
+
+      Connection connection2 = getConnectionBuilder().withDatabase(database).connect();
+      Statement statement2 = connection2.createStatement();
+      assertQuery(statement2, "SHOW temp_file_limit", new Row("4MB"));
+
+      // Change 'temp_file_limit'.
+      statement.execute("ALTER DATABASE " + database + " SET temp_file_limit = 1024");
+      waitForTServerHeartbeat();
+
+      // Test a new connection.
+      Connection connection3 = getConnectionBuilder().withDatabase(database).connect();
+      Statement statement3 = connection3.createStatement();
+      assertQuery(statement3, "SHOW temp_file_limit", new Row("1MB"));
+
+      // Test old connections.
+      assertQuery(statement1, "SHOW temp_file_limit", new Row("4MB"));
+      assertQuery(statement2, "SHOW temp_file_limit", new Row("4MB"));
+
+      // Reset 'temp_file_limit'.
+      statement.execute("ALTER DATABASE " + database + " RESET temp_file_limit");
+      waitForTServerHeartbeat();
+
+      // Test a new connection.
+      Connection connection4 = getConnectionBuilder().withDatabase(database).connect();
+      Statement statement4 = connection4.createStatement();
+      assertQuery(statement4, "SHOW temp_file_limit", new Row("1GB"));
+
+      // Test old connections.
+      assertQuery(statement1, "SHOW temp_file_limit", new Row("4MB"));
+      assertQuery(statement2, "SHOW temp_file_limit", new Row("4MB"));
+      assertQuery(statement3, "SHOW temp_file_limit", new Row("1MB"));
+
+      connection4.close();
+      connection3.close();
+      connection2.close();
+      connection1.close();
+    }
+  }
+  @Test
+  public void testAlterDatabaseSetResetGucTestDb() throws Exception {
+    testAlterDatabaseSetResetGucHelper("tdb");
+  }
+  @Test
+  public void testAlterDatabaseSetResetGucDefaultDb() throws Exception {
+    testAlterDatabaseSetResetGucHelper(DEFAULT_PG_DATABASE);
+  }
 }

src/postgres/src/backend/catalog/yb_catalog/yb_catalog_version.c

@@ -328,13 +328,14 @@ bool YbIncrementMasterCatalogVersionTableEntry(bool is_breaking_change,
 {
 	if (YbGetCatalogVersionType() != CATALOG_VERSION_CATALOG_TABLE)
 		return false;
-	/*
-	 * TemplateDbOid row is for global catalog version when not in per-db mode.
-	 */
-	YbIncrementMasterDBCatalogVersionTableEntryImpl(
-		YBIsDBCatalogVersionMode() ? MyDatabaseId : TemplateDbOid,
-		is_breaking_change, is_global_ddl, command_tag);
 
+	Oid database_oid = YbGetDatabaseOidToIncrementCatalogVersion();
+	Assert(OidIsValid(database_oid));
+
+	YbIncrementMasterDBCatalogVersionTableEntryImpl(database_oid,
+													is_breaking_change,
+													is_global_ddl,
+													command_tag);
 	if (yb_test_fail_next_inc_catalog_version)
 	{
 		yb_test_fail_next_inc_catalog_version = false;

src/postgres/src/backend/utils/misc/pg_yb_utils.c

@@ -1620,6 +1620,7 @@ typedef struct DdlTransactionState
 	bool is_global_ddl;
 	NodeTag original_node_tag;
 	const char *original_ddl_command_tag;
+	Oid database_oid;
 } DdlTransactionState;
 
 static DdlTransactionState ddl_transaction_state = {0};
@@ -1807,6 +1808,7 @@ YBDecrementDdlNestingLevel()
 			is_silent_altering = (mode == YB_DDL_MODE_SILENT_ALTERING);
 		}
 
+		Oid database_oid = YbGetDatabaseOidToIncrementCatalogVersion();
 		ddl_transaction_state = (DdlTransactionState) {};
 
 		YBCForceAllowCatalogModifications(false);
@@ -1822,7 +1824,11 @@ YBDecrementDdlNestingLevel()
 		 */
 		if (increment_done)
 		{
-			YbUpdateCatalogCacheVersion(YbGetCatalogCacheVersion() + 1);
+			if (database_oid == MyDatabaseId || !YBIsDBCatalogVersionMode())
+				YbUpdateCatalogCacheVersion(YbGetCatalogCacheVersion() + 1);
+			else
+				elog(LOG, "skipped optimization %u %u", database_oid, MyDatabaseId);
+
 			if (YbIsClientYsqlConnMgr())
 			{
 				/* Wait for tserver hearbeat */
@@ -2410,14 +2416,64 @@ YbDdlModeOptional YbGetDdlMode(
 }
 
 static void
-YBTxnDdlProcessUtility(
-	PlannedStmt *pstmt,
-	const char *queryString,
-	ProcessUtilityContext context,
-	ParamListInfo params,
-	QueryEnvironment *queryEnv,
-	DestReceiver *dest,
-	char *completionTag)
+CheckAlterDatabaseDdl(PlannedStmt *pstmt)
+{
+	Node *const parsetree = GetActualStmtNode(pstmt);
+	char *dbname = NULL;
+	switch (nodeTag(parsetree))
+	{
+		case T_AlterDatabaseSetStmt:
+			dbname = castNode(AlterDatabaseSetStmt, parsetree)->dbname;
+			break;
+		case T_AlterDatabaseStmt:
+			dbname = castNode(AlterDatabaseStmt, parsetree)->dbname;
+			break;
+		case T_RenameStmt:
+		{
+			const RenameStmt *const stmt = castNode(RenameStmt, parsetree);
+			if (stmt->renameType == OBJECT_DATABASE)
+			{
+				/*
+				 * At this point, old database name is already renamed to newname
+				 * in the current DDL transaction, so we will not be able to find
+				 * the database oid using the old name.
+				 */
+				dbname = stmt->newname;
+			}
+			break;
+		}
+		case T_AlterOwnerStmt:
+		{
+			const AlterOwnerStmt *const stmt = castNode(AlterOwnerStmt, parsetree);
+			if (stmt->objectType == OBJECT_DATABASE)
+				dbname = strVal(stmt->object);
+			break;
+		}
+		default:
+			break;
+	}
+	if (dbname)
+	{
+		/*
+		 * Alter database does not need to be a global impact DDL, it only needs
+		 * to increment the catalog version of the database that is altered,
+		 * which may not be the same as MyDatabaseId.
+		 */
+		ddl_transaction_state.database_oid = get_database_oid(dbname, false);
+		ddl_transaction_state.is_global_ddl = false;
+	}
+	else
+		ddl_transaction_state.database_oid = InvalidOid;
+}
+
+static void
+YBTxnDdlProcessUtility(PlannedStmt *pstmt,
+					   const char *queryString,
+					   ProcessUtilityContext context,
+					   ParamListInfo params,
+					   QueryEnvironment *queryEnv,
+					   DestReceiver *dest,
+					   char *completionTag)
 {
 
 	const YbDdlModeOptional ddl_mode = YbGetDdlMode(pstmt, context);
@@ -2462,7 +2518,10 @@ YBTxnDdlProcessUtility(
 									dest, completionTag);
 
 		if (is_ddl)
+		{
+			CheckAlterDatabaseDdl(pstmt);
 			YBDecrementDdlNestingLevel();
+		}
 	}
 	PG_CATCH();
 	{
@@ -5431,3 +5490,13 @@ YbGetIndexAttnum(Relation index, AttrNumber table_attno)
 	}
 	elog(ERROR, "column is not in index");
 }
+
+Oid
+YbGetDatabaseOidToIncrementCatalogVersion()
+{
+	if (!YBIsDBCatalogVersionMode())
+		return TemplateDbOid;
+	if (OidIsValid(ddl_transaction_state.database_oid))
+		return ddl_transaction_state.database_oid;
+	return MyDatabaseId;
+}

src/postgres/src/include/pg_yb_utils.h

@@ -1233,4 +1233,6 @@ extern bool yb_ysql_conn_mgr_superuser_existed;
 
 extern AttrNumber YbGetIndexAttnum(Relation index, AttrNumber table_attno);
 
+extern Oid YbGetDatabaseOidToIncrementCatalogVersion();
+
 #endif /* PG_YB_UTILS_H */

src/yb/yql/pgwrapper/pg_catalog_version-test.cc

@@ -1691,5 +1691,76 @@ TEST_F(PgCatalogVersionTest, DisableNopAlterRoleOptimization) {
   ASSERT_EQ(v3, v2 + 1);
 }
 
+TEST_F(PgCatalogVersionTest, AlterDatabaseCatalogVersionIncrement) {
+  PGConn conn = ASSERT_RESULT(Connect());
+  // Create a test db and a test user.
+  ASSERT_OK(conn.Execute("CREATE DATABASE test_db"));
+  ASSERT_OK(conn.Execute("CREATE USER test_user"));
+  auto v1_yugabyte = ASSERT_RESULT(GetCatalogVersion(&conn));
+
+  // Connect to the test db as the test user.
+  PGConn conn_test1 = ASSERT_RESULT(ConnectToDBAsUser("test_db" /* db_name */, "test_user"));
+  auto v1_test_db = ASSERT_RESULT(GetCatalogVersion(&conn_test1));
+
+  // Try to perform alter database test_db as the test user, which isn't the owner.
+  auto status = conn_test1.Execute("ALTER DATABASE test_db SET statement_timeout = 100");
+  ASSERT_TRUE(status.IsNetworkError()) << status;
+  ASSERT_STR_CONTAINS(status.ToString(), "must be owner of database");
+  status = conn_test1.Execute("ALTER DATABASE test_db SET temp_file_limit = 1024");
+  ASSERT_TRUE(status.IsNetworkError()) << status;
+  ASSERT_STR_CONTAINS(status.ToString(), "must be owner of database");
+  status = conn_test1.Execute("ALTER DATABASE test_db RENAME TO test_db_renamed");
+  ASSERT_TRUE(status.IsNetworkError()) << status;
+  ASSERT_STR_CONTAINS(status.ToString(), "must be owner of database");
+
+  ASSERT_OK(conn.Execute("ALTER DATABASE test_db OWNER TO test_user"));
+  auto v2_yugabyte = ASSERT_RESULT(GetCatalogVersion(&conn));
+  auto v2_test_db = ASSERT_RESULT(GetCatalogVersion(&conn_test1));
+  ASSERT_EQ(v2_yugabyte, v1_yugabyte);
+  ASSERT_EQ(v2_test_db, v1_test_db + 1);
+  WaitForCatalogVersionToPropagate();
+  ASSERT_OK(conn_test1.Execute("ALTER DATABASE test_db SET statement_timeout = 100"));
+  auto v3_yugabyte = ASSERT_RESULT(GetCatalogVersion(&conn));
+  auto v3_test_db = ASSERT_RESULT(GetCatalogVersion(&conn_test1));
+  ASSERT_EQ(v3_yugabyte, v2_yugabyte);
+  ASSERT_EQ(v3_test_db, v2_test_db + 1);
+  // temp_file_limit requires PGC_SUSET, test_user only has PGC_USERSET.
+  status = conn_test1.Execute("ALTER DATABASE test_db SET temp_file_limit = 1024");
+  ASSERT_TRUE(status.IsNetworkError()) << status;
+  ASSERT_STR_CONTAINS(status.ToString(), "permission denied to set parameter");
+
+  // Rename database requires createdb priviledge.
+  status = conn_test1.Execute("ALTER DATABASE test_db RENAME TO test_db_renamed");
+  ASSERT_TRUE(status.IsNetworkError()) << status;
+  ASSERT_STR_CONTAINS(status.ToString(), "permission denied to rename database");
+
+  // Grant createdb priviledge to test user.
+  ASSERT_OK(conn.Execute("ALTER USER test_user CREATEDB"));
+  auto v4_yugabyte = ASSERT_RESULT(GetCatalogVersion(&conn));
+  auto v4_test_db = ASSERT_RESULT(GetCatalogVersion(&conn_test1));
+  // Alter user is a global-impact DDL.
+  ASSERT_EQ(v4_yugabyte, v3_yugabyte + 1);
+  ASSERT_EQ(v4_test_db, v3_test_db + 1);
+  WaitForCatalogVersionToPropagate();
+  status = conn_test1.Execute("ALTER DATABASE test_db RENAME TO test_db_renamed");
+  ASSERT_TRUE(status.IsNetworkError()) << status;
+  ASSERT_STR_CONTAINS(status.ToString(), "current database cannot be renamed");
+
+  PGConn conn_test2 = ASSERT_RESULT(ConnectToDBAsUser(
+      "yugabyte" /* db_name */, "test_user"));
+  status = conn_test2.Execute("ALTER DATABASE test_db RENAME TO test_db_renamed");
+  ASSERT_TRUE(status.IsNetworkError()) << status;
+  // The error is only detected on connection to the same node.
+  ASSERT_STR_CONTAINS(status.ToString(), "is being accessed by other users");
+
+  // Make a connection to the second node as test user.
+  pg_ts = cluster_->tablet_server(1);
+  PGConn conn_test3 = ASSERT_RESULT(ConnectToDBAsUser(
+      "yugabyte" /* db_name */, "test_user"));
+  // The error is not detected on connection to the a different node, this is
+  // unique for YB.
+  ASSERT_OK(conn_test3.Execute("ALTER DATABASE test_db RENAME TO test_db_renamed"));
+}
+
 } // namespace pgwrapper
 } // namespace yb