[yugabyte#27532] xCluster: Suport major YSQL upgrade with xCluster

Summary:
Major YSQL upgrade involves a soft delete and recreate of the databases during pg_upgrade. Normally drop database is blocker if the database is under xCluster.
This only affects the `yugabyte` and internal dbs created during the initdb phase.
Relaxing this check for the major upgrade since no real database is dropped.
Jira: DB-17092

Test Plan: TEST_P(XClusterUpgradeTest, UpgradeHashPartitionedTable) {

Reviewers: slingam

Reviewed By: slingam

Subscribers: ybase

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

Author: hari90

src/yb/integration-tests/upgrade-tests/xcluster_upgrade-itest.cc

@@ -42,39 +42,45 @@ const auto kCountIdxRowsStmt = Format("SELECT COUNT(1) FROM $0 WHERE $1 IN (0,1,
 
 YB_STRONGLY_TYPED_BOOL(RangePartitioned);
 
-class XClusterUpgradeTest : public UpgradeTestBase {
+class XClusterUpgradeTestBase : public UpgradeTestBase {
  public:
-  XClusterUpgradeTest() : UpgradeTestBase(kBuild_2_25_0_0) {}
+  explicit XClusterUpgradeTestBase(const std::string& from_version)
+      : UpgradeTestBase(from_version) {}
 
-  void TearDown() override {
-    SwitchToConsumerCluster();
-    TearDownCluster();
+  void SetUp() override {
+    TEST_SETUP_SUPER(UpgradeTestBase);
 
-    SwitchToProducerCluster();
-    UpgradeTestBase::TearDown();
-  }
-
-  Status StartClustersInOldVersion(RangePartitioned ranged_partitioned = RangePartitioned::kFalse) {
     ExternalMiniClusterOptions opts;
     opts.num_masters = 3;
     opts.num_tablet_servers = 3;
 
     opts.cluster_id = "producer_cluster";
     opts.cluster_short_name = "P";
-    RETURN_NOT_OK(StartClusterInOldVersion(opts));
+    ASSERT_OK(StartClusterInOldVersion(opts));
     producer_cluster_ = cluster_.get();
     producer_client_ = client_.get();
 
+#if !defined(NDEBUG)
+    if (IsYsqlMajorVersionUpgrade()) {
+      GTEST_SKIP() << "xCluster major YSQL Upgrade testing not supported in debug mode";
+    }
+#endif
+
     SwitchToConsumerCluster();
     opts.cluster_id = "consumer_cluster";
     opts.cluster_short_name = "C";
-    RETURN_NOT_OK(StartClusterInOldVersion(opts));
+    ASSERT_OK(StartClusterInOldVersion(opts));
     consumer_cluster_ = cluster_.get();
     consumer_client_ = client_.get();
     SwitchToProducerCluster();
+  }
 
-    RETURN_NOT_OK(CreateTablesAndSetupXCluster(ranged_partitioned));
-    return Status::OK();
+  void TearDown() override {
+    SwitchToConsumerCluster();
+    TearDownCluster();
+
+    SwitchToProducerCluster();
+    UpgradeTestBase::TearDown();
   }
 
   Status RunOnBothClusters(const std::function<Status(ExternalMiniCluster*)>& run_on_cluster) {
@@ -245,7 +251,7 @@ class XClusterUpgradeTest : public UpgradeTestBase {
   }
 
   void SimpleReplicationTest(RangePartitioned ranged_partitioned) {
-    ASSERT_OK(StartClustersInOldVersion());
+    ASSERT_OK(CreateTablesAndSetupXCluster(ranged_partitioned));
 
     TestThreadHolder thread_holder;
     std::atomic<int64_t> rows_inserted = 0;
@@ -307,13 +313,24 @@ class XClusterUpgradeTest : public UpgradeTestBase {
   client::YBClient *producer_client_, *consumer_client_;
 };
 
-TEST_F(XClusterUpgradeTest, UpgradeHashPartitionedTable) {
+class XClusterUpgradeTest : public XClusterUpgradeTestBase,
+                            public ::testing::WithParamInterface<std::string> {
+ public:
+  XClusterUpgradeTest() : XClusterUpgradeTestBase(GetParam()) {}
+};
+
+INSTANTIATE_TEST_SUITE_P(
+    UpgradeFrom_2024_2_4_0, XClusterUpgradeTest, ::testing::Values(kBuild_2024_2_4_0));
+INSTANTIATE_TEST_SUITE_P(
+    UpgradeFrom_2_25_0_0, XClusterUpgradeTest, ::testing::Values(kBuild_2_25_0_0));
+
+TEST_P(XClusterUpgradeTest, UpgradeHashPartitionedTable) {
   ASSERT_NO_FATAL_FAILURE(SimpleReplicationTest(RangePartitioned::kFalse));
 }
 
 // #27380 added support for range partitioned tables in xCluster replication.
 // Enable this test once the from build with the fix is available.
-TEST_F(XClusterUpgradeTest, YB_DISABLE_TEST(UpgradeRangePartitionedTable)) {
+TEST_P(XClusterUpgradeTest, YB_DISABLE_TEST(UpgradeRangePartitionedTable)) {
   ASSERT_NO_FATAL_FAILURE(SimpleReplicationTest(RangePartitioned::kTrue));
 }
 

src/yb/master/catalog_manager.cc

@@ -9247,7 +9247,10 @@ Status CatalogManager::DeleteYsqlDatabase(const DeleteNamespaceRequestPB* req,
   // Only allow YSQL database deletion if it does not contain any replicated tables. No need to
   // check this for YCQL keyspaces, as YCQL does not allow drops of non-empty keyspaces, regardless
   // of their replication status.
-  RETURN_NOT_OK(CheckIfDatabaseHasReplication(database));
+  // Skip this check during major YSQL upgrade since we are not actually dropping the database.
+  if (!ysql_manager_->IsMajorUpgradeInProgress()) {
+    RETURN_NOT_OK(CheckIfDatabaseHasReplication(database));
+  }
 
   // Set the Namespace to DELETING.
   TRACE("Locking database");