[#26052] xCluster: bump up normal OID counter at start of automatic mode

Summary:
In more detail, when xCluster automatic mode replication starts, we now increase the normal space OID counter above any normal space OIDs in use that we might need to preserve across universes as part of xCluster replication.

A new function in sys_catalog, ReadHighestNormalPreservableOid, runs on the master leader in the source universe to compute the needed OID to bump to.

Did a small amount of tidying/improving.

Fixes #26052
Jira: DB-15379

Test Plan:
New "unit test" for ReadHighestNormalPreservableOid function:
```
ybd --cxx-test sys_catalog-itest --gtest_filter '*.ReadHighestNormalPreservableOid'
```
Part of this test is currently disabled because the `pg_catalog.binary_upgrade_set_next_heap_relfilenode` directive isn't honored yet; this will be fixed shortly.

In the meantime, I temporarily manually changed the Postgres source to honor the directive by editing:
```
~/code/yugabyte-db/src/postgres/src/backend/catalog/heap.c:1435:
				/*
				 * YB: The parent partition has DocDB storage, so we preserve
				 * its relfilenode when upgrading.
				 */
				if ((RELKIND_HAS_STORAGE(relkind) ||
					 (IsYugaByteEnabled() && relkind == RELKIND_PARTITIONED_TABLE))
					&& !yb_binary_restore)
```
to remove the `&& !yb_binary_restore` part at the end.  After doing that, the test passes without disabling anything.

New test for checking that the bumping is done correctly end-to-end:
```
ybd  --cxx-test xcluster_ddl_replication-test --gtest_filter '*.ReplicationSetUpBumpsOidCounter'
```
I have verified this test fails if I don't bump or if I don't invalidate.

Reviewers: xCluster, hsunder

Reviewed By: hsunder

Subscribers: ybase

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

Author: mdbridge

src/yb/integration-tests/CMakeLists.txt

@@ -225,6 +225,8 @@ ADD_YB_TEST(load_balancer_multi_table-test)
 ADD_YB_TEST(load_balancer_colocated_tables-test)
 ADD_YB_TEST(load_balancer_respect_affinity-test)
 ADD_YB_TEST(load_balancer_placement_policy-test)
+ADD_YB_TEST(sys_catalog-itest)
+YB_TEST_TARGET_LINK_LIBRARIES(sys_catalog-itest pg_wrapper_test_base)
 ADD_YB_TEST(sys_catalog_respect_affinity-test)
 ADD_YB_TEST(restart-test)
 ADD_YB_TEST(yb-ts-cli-itest)

src/yb/integration-tests/sys_catalog-itest.cc

@@ -0,0 +1,147 @@
+// Copyright (c) YugabyteDB, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
+// in compliance with the License.  You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software distributed under the License
+// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+// or implied.  See the License for the specific language governing permissions and limitations
+// under the License.
+
+#include <regex>
+#include <string>
+
+#include <gtest/gtest.h>
+
+#include "yb/master/catalog_manager_if.h"
+#include "yb/master/master_ddl.pb.h"
+#include "yb/master/sys_catalog.h"
+
+#include "yb/yql/pgwrapper/pg_mini_test_base.h"
+
+namespace yb::master {
+
+class SysCatalogITest : public pgwrapper::PgMiniTestBase {
+ public:
+  void SetUp() override {
+    TEST_SETUP_SUPER(pgwrapper::PgMiniTestBase);
+
+    master::GetNamespaceInfoResponsePB resp;
+    ASSERT_OK(client_->GetNamespaceInfo(
+        /*namespace_id=*/std::string(), namespace_name, YQL_DATABASE_PGSQL, &resp));
+    namespace_id_ = resp.namespace_().id();
+
+    google::SetVLOGLevel("catalog_manager*", 1);
+  }
+
+  const std::string namespace_name = "yugabyte";
+  NamespaceId namespace_id_;
+};
+
+TEST_F(SysCatalogITest, ReadHighestNormalPreservableOid) {
+  auto conn = ASSERT_RESULT(ConnectToDB(namespace_name));
+  auto database_oid = ASSERT_RESULT(GetPgsqlDatabaseOid(namespace_id_));
+  auto sys_catalog = ASSERT_RESULT(catalog_manager())->sys_catalog();
+
+  auto original_highest_oid =
+      ASSERT_RESULT(sys_catalog->ReadHighestNormalPreservableOid(database_oid));
+  // Make sure all the OIDs we are going to use are higher than any of the starting OIDs.
+  ASSERT_LT(original_highest_oid, 20'000);
+
+  // Create secondary space OIDs of each of the kinds we preserve.
+  {
+    // Here @ will be replaced by the lowest secondary space OID.
+    std::string command = R"(
+       SET yb_binary_restore = true;
+       SET yb_ignore_pg_class_oids = false;
+       SELECT pg_catalog.binary_upgrade_set_next_pg_type_oid((@)::pg_catalog.oid);
+       SELECT pg_catalog.binary_upgrade_set_next_array_pg_type_oid((@+1)::pg_catalog.oid);
+       CREATE TYPE high_enum AS ENUM ();
+       SELECT pg_catalog.binary_upgrade_set_next_pg_enum_oid((@+2)::pg_catalog.oid);
+       ALTER TYPE high_enum ADD VALUE 'red';
+       SELECT pg_catalog.binary_upgrade_set_next_heap_pg_class_oid((@+3)::pg_catalog.oid);
+       SELECT pg_catalog.binary_upgrade_set_next_heap_relfilenode((@+4)::pg_catalog.oid);
+       CREATE SEQUENCE high_sequence;
+      )";
+    ASSERT_OK(conn.Execute(std::regex_replace(
+        command, std::regex{"@"}, std::to_string(kPgFirstSecondarySpaceObjectId))));
+    auto oid = ASSERT_RESULT(sys_catalog->ReadHighestNormalPreservableOid(database_oid));
+    // The addition of these OIDs should not have changed the highest normal space OID at all.
+    ASSERT_EQ(oid, original_highest_oid);
+  }
+
+  {
+    // Run CREATE TYPE new_enum AS ENUM ('red', 'orange'); but using 50000-50001 as pg_enum OIDs.
+    ASSERT_OK(conn.Execute(R"(
+       SET yb_binary_restore = true;
+       SET yb_ignore_pg_class_oids = false;
+       SELECT pg_catalog.binary_upgrade_set_next_pg_type_oid('30000'::pg_catalog.oid);
+       SELECT pg_catalog.binary_upgrade_set_next_array_pg_type_oid('30001'::pg_catalog.oid);
+
+       CREATE TYPE new_enum AS ENUM ();
+       SELECT pg_catalog.binary_upgrade_set_next_pg_enum_oid('50000'::pg_catalog.oid);
+       ALTER TYPE new_enum ADD VALUE 'red';
+       SELECT pg_catalog.binary_upgrade_set_next_pg_enum_oid('50001'::pg_catalog.oid);
+       ALTER TYPE new_enum ADD VALUE 'orange';
+      )"));
+    auto oid = ASSERT_RESULT(sys_catalog->ReadHighestNormalPreservableOid(database_oid));
+    EXPECT_EQ(oid, 50'001);
+  }
+
+  {
+    // Run CREATE SEQUENCE new_sequence; using pg_class OID 60000.
+    ASSERT_OK(conn.Execute(R"(
+       SET yb_binary_restore = true;
+       SET yb_ignore_pg_class_oids = false;
+       SELECT pg_catalog.binary_upgrade_set_next_heap_pg_class_oid('60000'::pg_catalog.oid);
+       SELECT pg_catalog.binary_upgrade_set_next_heap_relfilenode('30010'::pg_catalog.oid);
+
+       CREATE SEQUENCE new_sequence;
+      )"));
+    auto oid = ASSERT_RESULT(sys_catalog->ReadHighestNormalPreservableOid(database_oid));
+    EXPECT_EQ(oid, 60000);
+  }
+
+  {
+    // Run CREATE TYPE new_composite AS (x int); using pg_type OID 70000-70001.
+    ASSERT_OK(conn.Execute(R"(
+       SET yb_binary_restore = true;
+       SET yb_ignore_pg_class_oids = false;
+       SELECT pg_catalog.binary_upgrade_set_next_pg_type_oid('70000'::pg_catalog.oid);
+       SELECT pg_catalog.binary_upgrade_set_next_array_pg_type_oid('70001'::pg_catalog.oid);
+       SELECT pg_catalog.binary_upgrade_set_next_heap_pg_class_oid('30020'::pg_catalog.oid);
+
+       CREATE TYPE new_composite AS (x int);
+      )"));
+    auto oid = ASSERT_RESULT(sys_catalog->ReadHighestNormalPreservableOid(database_oid));
+    EXPECT_EQ(oid, 70'001);
+  }
+
+  {
+    // Run CREATE TABLE public.my_table (x integer); attempting to use pg_relfilenode 80000.
+    ASSERT_OK(conn.Execute(R"(
+       SET yb_binary_restore = true;
+       SET yb_ignore_pg_class_oids = false;
+       SELECT pg_catalog.binary_upgrade_set_next_pg_type_oid('30030'::pg_catalog.oid);
+       SELECT pg_catalog.binary_upgrade_set_next_array_pg_type_oid('30031'::pg_catalog.oid);
+       SELECT pg_catalog.binary_upgrade_set_next_heap_pg_class_oid('30032'::pg_catalog.oid);
+       SELECT pg_catalog.binary_upgrade_set_next_heap_relfilenode('80000'::pg_catalog.oid);
+
+       CREATE TABLE my_table (x integer);
+      )"));
+
+    // TODO(yhaddad): fix Postgres to honor binary_upgrade_set_next_heap_relfilenode directive then
+    // update this test to expect 80'000.
+    //
+    // (Currently this directive is not honored, which prevents this test from testing the
+    // relfilenode case.)
+    LOG(INFO) << ASSERT_RESULT(conn.FetchAllAsString("SELECT oid, relfilenode FROM pg_class;"));
+    auto oid = ASSERT_RESULT(sys_catalog->ReadHighestNormalPreservableOid(database_oid));
+    // EXPECT_EQ(oid, 80'000);
+    EXPECT_EQ(oid, 70'001);
+  }
+}
+
+}  // namespace yb::master

src/yb/integration-tests/xcluster/xcluster_ddl_replication-test.cc

@@ -1318,6 +1318,80 @@ TEST_F(XClusterDDLReplicationSwitchoverTest, SwitchoverBumpsAboveUsedOids) {
   ASSERT_OK(WaitForSafeTimeToAdvanceToNow());
 }
 
+using XClusterDDLReplicationSetupTest = XClusterDDLReplicationSwitchoverTest;
+
+TEST_F(XClusterDDLReplicationSetupTest, ReplicationSetUpBumpsOidCounter) {
+  if (!UseYbController()) {
+    GTEST_SKIP() << "This test does not work with yb_backup.py";
+  }
+
+  // The resulting statement will consume 100 pg_enum OIDs.
+  auto CreateGiantEnumStatement = [](std::string name) {
+    std::string result = Format("CREATE TYPE $0 AS ENUM ('l0'", name);
+    for (int i = 1; i < 100; i++) {
+      result += Format(", 'l$0'", i);
+    }
+    return result + ");";
+  };
+
+  google::SetVLOGLevel("catalog_manager*", 1);
+  google::SetVLOGLevel("pg_client_service*", 1);
+  google::SetVLOGLevel("xcluster_source_manager", 1);
+  // Cache only 30 OIDs at a time; see below for why this value was chosen.
+  ANNOTATE_UNPROTECTED_WRITE(FLAGS_ysql_oid_cache_prefetch_size) = 30;
+
+  ASSERT_OK(SetUpClusters(/*is_colocated=*/false, /*start_yb_controller_servers=*/true));
+
+  // Create a giant enum on cluster A.
+  {
+    auto conn = ASSERT_RESULT(cluster_A_->ConnectToDB(namespace_name));
+    ASSERT_OK(conn.Execute(CreateGiantEnumStatement("first_enum")));
+    // Backup requires at least one table so create one.
+    ASSERT_OK(conn.Execute("CREATE TABLE my_table (x INT);"));
+  }
+
+  // Reset OID counters on A by backing up then restoring the database on cluster A.
+  ASSERT_OK(BackupFromProducer());
+  SetReplicationDirection(ReplicationDirection::BToA);
+  ASSERT_OK(RestoreToConsumer());
+  SetReplicationDirection(ReplicationDirection::AToB);
+
+  // Allocate a few OIDs on A to force caching of normal space OIDs.
+  {
+    auto conn = ASSERT_RESULT(cluster_A_->ConnectToDB(namespace_name));
+    ASSERT_OK(conn.Execute("CREATE TABLE exercise_cache (x INT);"));
+  }
+
+  // Set up xCluster replication with a nonempty database.
+  ASSERT_OK(CheckpointReplicationGroupOnNamespaces({namespace_name}));
+  ASSERT_OK(BackupFromProducer());
+  ASSERT_OK(RestoreToConsumer());
+  ASSERT_OK(CreateReplicationFromCheckpoint());
+
+  // At this point, in the absence of OID cache invalidation, we would still have OIDs cached on A
+  // from before replication was set up.  (Replication setup does create some objects, consuming
+  // OIDs, but not enough to exhaust the cache size of 30 we have set.)
+  //
+  // Importantly, we do not have enough OIDs cached to handle an entire giant enum.
+
+  {
+    // Drop via manual DDL replication the enum on only cluster A.
+    auto conn = ASSERT_RESULT(cluster_A_->ConnectToDB(namespace_name));
+    ASSERT_OK(conn.Execute(R"(
+                     SET yb_xcluster_ddl_replication.enable_manual_ddl_replication=1;
+                     DROP TYPE first_enum;
+                     SET yb_xcluster_ddl_replication.enable_manual_ddl_replication=0;)"));
+    ASSERT_OK(WaitForSafeTimeToAdvanceToNow());
+
+    // Now create a new giant enum via normal DDL replication.  If we did not bump up the normal
+    // space OID counter and invalidate on A then this will cause a OID collision on cluster B
+    // because the new enum on A will use OIDs freed up by dropping the previous enum but those OIDs
+    // are still in use on B because the previous enum still exists there.
+    ASSERT_OK(conn.Execute(CreateGiantEnumStatement("second_enum")));
+    ASSERT_OK(WaitForSafeTimeToAdvanceToNow());
+  }
+}
+
 class XClusterDDLReplicationAddDropColumnTest : public XClusterDDLReplicationTest {
  public:
   void SetUp() override {