[yugabyte#25709] xCluster: Handle tables with same name in non-DBScoped DR xCluster

Summary:
If PITR is enabled we hide the table, instead of immediately deleting it when dropped.
This causes non-DBScoped DR xCluster to fail if a table or index of the same name is immediately created again.

- Moved `snapshot_test_util.cc` from `ql-dml-test-base` to `yb_client_test_util`
Jira: DB-14966

Test Plan:
XClusterDbScopedYsqlIndexTest.CreateDropIndexWithPITR
XClusterYSqlTestConsistentTransactionsTest.CreateDropTableAndIndexWithPITR

Reviewers: jhe, xCluster

Reviewed By: jhe

Subscribers: ybase

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

Author: hari90

src/yb/client/CMakeLists.txt

@@ -119,11 +119,11 @@ endif()
 # part of the client itself (ie we don't want to ship it to customers,
 # and therefore don't need to worry about export strictness)
 ADD_YB_LIBRARY(yb_client_test_util
-               SRCS client-test-util.cc
+               SRCS client-test-util.cc snapshot_test_util.cc
                DEPS gmock gtest yb_client yb_test_util)
 
 ADD_YB_LIBRARY(ql-dml-test-base
-               SRCS ql-dml-test-base.cc txn-test-base.cc snapshot_test_util.cc
+               SRCS ql-dml-test-base.cc txn-test-base.cc
                DEPS integration-tests)
 
 # Tests

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

@@ -13,7 +13,6 @@
 
 #include <gmock/gmock.h>
 
-#include "yb/client/snapshot_test_util.h"
 #include "yb/client/table.h"
 #include "yb/client/xcluster_client.h"
 #include "yb/client/yb_table_name.h"
@@ -68,19 +67,6 @@ class XClusterDBScopedTest : public XClusterYsqlTestBase {
       const NamespaceId& namespace_id) {
     return GetXClusterStreams(namespace_id, /*table_names=*/{}, /*pg_schema_names=*/{});
   }
-
-  Status EnablePITROnClusters() {
-    return RunOnBothClusters([this](Cluster* cluster) -> Status {
-      client::SnapshotTestUtil snapshot_util;
-      snapshot_util.SetProxy(&cluster->client_->proxy_cache());
-      snapshot_util.SetCluster(cluster->mini_cluster_.get());
-
-      RETURN_NOT_OK(snapshot_util.CreateSchedule(
-          nullptr, YQL_DATABASE_PGSQL, namespace_name, client::WaitSnapshot::kTrue,
-          2s * kTimeMultiplier, 20h));
-      return Status::OK();
-    });
-  }
 };
 
 TEST_F(XClusterDBScopedTest, TestCreateWithCheckpoint) {

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

@@ -3196,4 +3196,49 @@ TEST_F(XClusterYsqlTest, TransactionalBidirectionalWithTwoDBs) {
   ASSERT_OK(WriteWorkload(250, 350, &consumer_cluster_, consumer_tables_[1]->name()));
 }
 
+// Create and drop tables and indexes in a loop with PITR which will keep the dropped tables in
+// hidden state.
+TEST_F(XClusterYSqlTestConsistentTransactionsTest, CreateDropTableAndIndexWithPITR) {
+  const auto kBatchSize = 10;
+  ASSERT_OK(CreateTableAndSetupReplication());
+  ASSERT_OK(EnablePITROnClusters());
+
+  auto producer_conn =
+      EXPECT_RESULT(producer_cluster_.ConnectToDB(producer_table_->name().namespace_name()));
+  auto consumer_conn =
+      EXPECT_RESULT(consumer_cluster_.ConnectToDB(consumer_table_->name().namespace_name()));
+
+  for (int run_count = 0; run_count < 5; run_count++) {
+    auto producer_table_name = ASSERT_RESULT(CreateYsqlTable(
+        /*idx=*/1, /*num_tablets=*/1, &producer_cluster_));
+    std::shared_ptr<client::YBTable> new_producer_table;
+    ASSERT_OK(producer_client()->OpenTable(producer_table_name, &new_producer_table));
+
+    auto consumer_table_name = ASSERT_RESULT(CreateYsqlTable(
+        /*idx=*/1, /*num_tablets=*/1, &consumer_cluster_));
+    std::shared_ptr<client::YBTable> new_consumer_table;
+    ASSERT_OK(consumer_client()->OpenTable(consumer_table_name, &new_consumer_table));
+
+    ASSERT_OK(
+        AlterUniverseReplication(kReplicationGroupId, {new_producer_table}, true /* add_tables */));
+
+    const auto create_index_stmt = Format(
+        "CREATE INDEX my_idx ON $0 ($1 ASC)", new_producer_table->name().table_name(),
+        kKeyColumnName);
+
+    ASSERT_OK(producer_conn.Execute(create_index_stmt));
+    ASSERT_OK(consumer_conn.Execute(create_index_stmt));
+
+    ASSERT_OK(InsertRowsInProducer(
+        run_count * kBatchSize, (run_count + 1) * kBatchSize, new_producer_table));
+    ASSERT_OK(VerifyWrittenRecords(new_producer_table, new_consumer_table));
+
+    ASSERT_OK(DropYsqlTable(producer_cluster_, *new_producer_table.get()));
+    ASSERT_OK(DropYsqlTable(consumer_cluster_, *new_consumer_table.get()));
+  }
+
+  ASSERT_OK(InsertRowsInProducer(0, 10, producer_table_));
+  ASSERT_OK(VerifyWrittenRecords());
+}
+
 }  // namespace yb

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

@@ -44,7 +44,6 @@ namespace yb {
 const string kTableName = "test_table";
 const string kIndexName = "test_index";
 const auto kInsertStmtFormat = Format("INSERT INTO $0 VALUES($1, $1)", kTableName, "$0");
-const auto kDropIndexStmt = Format("DROP INDEX $0", kIndexName);
 const auto kId1CountStmt = Format("SELECT COUNT(*) FROM $0 WHERE id1 >= 0", kTableName);
 const auto kId2CountStmt = Format("SELECT COUNT(*) FROM $0 WHERE id2 >= 0", kTableName);
 const auto kSelectAllId12Stmt = Format("SELECT id1, id2 FROM $0 ORDER BY id1, id2", kTableName);
@@ -157,6 +156,10 @@ class XClusterYsqlIndexTest : public XClusterYsqlTestBase {
     return conn.Execute(Format("CREATE INDEX $0 ON $1 (id2 ASC)", kIndexName, kTableName));
   }
 
+  virtual Status DropIndex(pgwrapper::PGConn& conn) {
+    return conn.Execute(Format("DROP INDEX $0", kIndexName));
+  }
+
   auto GetAllRows(pgwrapper::PGConn* conn) {
     return conn->FetchRows<int32_t, int32_t>(kSelectAllId12Stmt);
   }
@@ -541,6 +544,30 @@ TEST_F(XClusterDbScopedYsqlIndexTest, CreateIndexWithWorkload) {
   ASSERT_OK(TestCreateIndexConcurrentWorkload());
 }
 
+// Create and drop indexes in a loop with PITR which will keep the dropped tables in
+// hidden state.
+TEST_F(XClusterDbScopedYsqlIndexTest, CreateDropIndexWithPITR) {
+  ASSERT_OK(EnablePITROnClusters());
+
+  for (int run_count = 0; run_count < 2; run_count++) {
+    LOG(INFO) << "Run count: " << run_count;
+
+    ASSERT_OK(CreateIndex(*producer_conn_));
+    ASSERT_OK(CreateIndex(*consumer_conn_));
+
+    // Insert more rows and validate.
+    for (int i = 0; i < 10; i++, row_count_++) {
+      ASSERT_OK(producer_conn_->ExecuteFormat(kInsertStmtFormat, row_count_));
+    }
+
+    ASSERT_OK(WaitForSafeTimeToAdvanceToNow());
+    ASSERT_OK(ValidateRows());
+
+    ASSERT_OK(DropIndex(*producer_conn_));
+    ASSERT_OK(DropIndex(*consumer_conn_));
+  }
+}
+
 class XClusterYsqlIndexProducerOnlyTest : public XClusterYsqlIndexTest {
   void SetUp() override {
     XClusterYsqlTestBase::SetUp();

src/yb/integration-tests/xcluster/xcluster_ysql_test_base.cc

@@ -14,6 +14,7 @@
 #include "yb/integration-tests/xcluster/xcluster_ysql_test_base.h"
 
 #include "yb/client/client.h"
+#include "yb/client/snapshot_test_util.h"
 #include "yb/client/table.h"
 #include "yb/client/xcluster_client.h"
 #include "yb/client/yb_table_name.h"
@@ -1112,4 +1113,16 @@ Status XClusterYsqlTestBase::VerifyDDLExtensionTablesDeletion(
   });
 }
 
+Status XClusterYsqlTestBase::EnablePITROnClusters() {
+  return RunOnBothClusters([this](Cluster* cluster) -> Status {
+    client::SnapshotTestUtil snapshot_util;
+    snapshot_util.SetProxy(&cluster->client_->proxy_cache());
+    snapshot_util.SetCluster(cluster->mini_cluster_.get());
+
+    RETURN_NOT_OK(snapshot_util.CreateSchedule(
+        nullptr, YQL_DATABASE_PGSQL, namespace_name, client::WaitSnapshot::kTrue,
+        2s * kTimeMultiplier, 20h));
+    return Status::OK();
+  });
+}
 }  // namespace yb

src/yb/integration-tests/xcluster/xcluster_ysql_test_base.h

@@ -185,6 +185,8 @@ class XClusterYsqlTestBase : public XClusterTestBase {
   Status VerifyDDLExtensionTablesCreation(const NamespaceName& db_name, bool only_source = false);
   Status VerifyDDLExtensionTablesDeletion(const NamespaceName& db_name, bool only_source = false);
 
+  Status EnablePITROnClusters();
+
  protected:
   void TestReplicationWithSchemaChanges(TableId producer_table_id, bool bootstrap);
 

src/yb/master/xrepl_catalog_manager.cc

@@ -35,6 +35,7 @@
 #include "yb/util/is_operation_done_result.h"
 #include "yb/master/xcluster/xcluster_manager.h"
 #include "yb/master/xcluster/xcluster_replication_group.h"
+#include "yb/master/xcluster/master_xcluster_util.h"
 #include "yb/master/xcluster_consumer_registry_service.h"
 #include "yb/master/xcluster_rpc_tasks.h"
 #include "yb/master/master.h"
@@ -3546,9 +3547,7 @@ Status CatalogManager::BootstrapProducer(
   SCHECK(
       pg_database_type || req->db_type() == YQL_DATABASE_CQL, InvalidArgument,
       "Invalid database type");
-  SCHECK(
-      req->has_namespace_name() && !req->namespace_name().empty(), InvalidArgument,
-      "No namespace specified");
+  SCHECK_PB_FIELDS_NOT_EMPTY(*req, namespace_name);
   SCHECK_GT(req->table_name_size(), 0, InvalidArgument, "No tables specified");
   if (pg_database_type) {
     SCHECK_EQ(
@@ -3560,21 +3559,38 @@ Status CatalogManager::BootstrapProducer(
         "Pg Schema does not apply to CQL databases");
   }
 
+  NamespaceIdentifierPB ns_id;
+  ns_id.set_database_type(req->db_type());
+  ns_id.set_name(req->namespace_name());
+  auto ns = VERIFY_RESULT(FindNamespace(ns_id));
+  // We can skip sequence data, since that is only used in db scoped xCluster which does not call
+  // this function.
+  auto all_tables = VERIFY_RESULT(
+      GetTablesEligibleForXClusterReplication(*this, ns->id(), /*include_sequences_data= */ false));
+
   cdc::BootstrapProducerRequestPB bootstrap_req;
   master::TSDescriptorPtr ts = nullptr;
   for (int i = 0; i < req->table_name_size(); i++) {
     string pg_schema_name = pg_database_type ? req->pg_schema_name(i) : "";
-    auto table_info = GetTableInfoFromNamespaceNameAndTableName(
-        req->db_type(), req->namespace_name(), req->table_name(i), pg_schema_name);
+
+    auto table_designator = std::find_if(
+        all_tables.begin(), all_tables.end(),
+        [&table_name = req->table_name(i),
+         &pg_schema_name](const TableDesignator& table_designator) {
+          return table_designator.name() == table_name &&
+                 table_designator.pgschema_name() == pg_schema_name;
+        });
     SCHECK(
-        table_info, NotFound, Format("Table $0.$1$2 not found"), req->namespace_name(),
-        (pg_schema_name.empty() ? "" : pg_schema_name + "."), req->table_name(i));
+        table_designator != all_tables.end(), NotFound, Format("Table $0.$1$2 not found"),
+        req->namespace_name(), (pg_schema_name.empty() ? "" : pg_schema_name + "."),
+        req->table_name(i));
 
-    bootstrap_req.add_table_ids(table_info->id());
-    resp->add_table_ids(table_info->id());
+    bootstrap_req.add_table_ids(table_designator->id);
+    resp->add_table_ids(table_designator->id);
 
     // Pick a valid tserver to bootstrap from.
     if (!ts) {
+      auto table_info = VERIFY_RESULT(FindTableById(table_designator->id));
       ts = VERIFY_RESULT(VERIFY_RESULT(table_info->GetTablets()).front()->GetLeader());
     }
   }