[#19184] xClusterDDLRepl: Mark extension tables as system tables

Summary:
Mark ddl_queue and replicated_ddls tables as system tables.
Move the non-colocation and single tablet creation rules to catalog manager.
Blocking tablet splits on these tables.

Jira: DB-7986

Test Plan:
```
ybd --cxx-test xcluster_ddl_replication-test --gtest_filter "XClusterDDLReplicationTest.CreateTable"
ybd --cxx-test xcluster_ddl_replication-test --gtest_filter "XClusterDDLReplicationTest.DisableSplitting"
```

Also tested locally and confirmed that these tables show up as system tables in the master ui.

Reviewers: hsunder, xCluster

Reviewed By: hsunder

Subscribers: asrivastava, yql, ybase, xCluster

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

src/postgres/yb-extensions/yb_xcluster_ddl_replication/yb_xcluster_ddl_replication--1.0.sql

@@ -8,17 +8,13 @@ CREATE TABLE yb_xcluster_ddl_replication.ddl_queue(
   start_time bigint NOT NULL,
   query_id bigint NOT NULL,
   yb_data jsonb NOT NULL,
-  PRIMARY KEY (start_time, query_id))
-WITH (COLOCATION = false)
-SPLIT INTO 1 TABLETS;
+  PRIMARY KEY (start_time, query_id));
 
 CREATE TABLE yb_xcluster_ddl_replication.replicated_ddls(
   start_time bigint NOT NULL,
   query_id bigint NOT NULL,
   yb_data jsonb NOT NULL,
-  PRIMARY KEY (start_time, query_id))
-WITH (COLOCATION = false)
-SPLIT INTO 1 TABLETS;
+  PRIMARY KEY (start_time, query_id));
 
 /* ------------------------------------------------------------------------- */
 /* Create event triggers. */

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

@@ -75,9 +75,23 @@ class XClusterDDLReplicationTest : public XClusterYsqlTestBase {
         xcluster::kDDLQueuePgSchemaName));
     auto c_conn = VERIFY_RESULT(consumer_cluster_.Connect());
     RETURN_NOT_OK(c_conn.ExecuteFormat("CREATE EXTENSION $0", xcluster::kDDLQueuePgSchemaName));
-    return c_conn.ExecuteFormat(
+    RETURN_NOT_OK(c_conn.ExecuteFormat(
         "ALTER DATABASE $0 SET $1.replication_role = TARGET", namespace_name,
-        xcluster::kDDLQueuePgSchemaName);
+        xcluster::kDDLQueuePgSchemaName));
+
+    // Ensure that tables are properly created with only one tablet each.
+    RETURN_NOT_OK(RunOnBothClusters([&](Cluster* cluster) -> Status {
+      for (const auto& table_name :
+           {xcluster::kDDLQueueTableName, xcluster::kDDLReplicatedTableName}) {
+        auto yb_table_name = VERIFY_RESULT(
+            GetYsqlTable(cluster, namespace_name, xcluster::kDDLQueuePgSchemaName, table_name));
+        std::shared_ptr<client::YBTable> table;
+        RETURN_NOT_OK(producer_client()->OpenTable(yb_table_name, &table));
+        SCHECK_EQ(table->GetPartitionCount(), 1, IllegalState, "Expected 1 tablet");
+      }
+      return Status::OK();
+    }));
+    return Status::OK();
   }
 
   void InsertRowsIntoProducerTableAndVerifyConsumer(
@@ -107,6 +121,27 @@ class XClusterDDLReplicationTest : public XClusterYsqlTestBase {
   }
 };
 
+TEST_F(XClusterDDLReplicationTest, DisableSplitting) {
+  // Ensure that splitting of xCluster DDL Replication tables is disabled on both sides.
+  ASSERT_OK(SetUpClusters());
+  ASSERT_OK(EnableDDLReplicationExtension());
+
+  for (auto* cluster : {&producer_cluster_, &consumer_cluster_}) {
+    for (const auto& table : {xcluster::kDDLQueueTableName, xcluster::kDDLReplicatedTableName}) {
+      auto yb_table_name = ASSERT_RESULT(
+          GetYsqlTable(cluster, namespace_name, xcluster::kDDLQueuePgSchemaName, table));
+
+      google::protobuf::RepeatedPtrField<master::TabletLocationsPB> tablets;
+      ASSERT_OK(cluster->client_->GetTabletsFromTableId(yb_table_name.table_id(), 1, &tablets));
+
+      auto res = CallAdmin(cluster->mini_cluster_.get(), "split_tablet", tablets[0].tablet_id());
+      ASSERT_NOK(res);
+      ASSERT_TRUE(res.status().message().Contains(
+          "Tablet splitting is not supported for xCluster DDL Replication tables"));
+    }
+  }
+}
+
 TEST_F(XClusterDDLReplicationTest, CreateTable) {
   ASSERT_OK(SetUpClusters());
   ASSERT_OK(EnableDDLReplicationExtension());

src/yb/master/catalog_entity_info.cc

@@ -34,6 +34,7 @@
 
 #include <string>
 
+#include "yb/cdc/xcluster_types.h"
 #include "yb/common/colocated_util.h"
 #include "yb/common/doc_hybrid_time.h"
 #include "yb/dockv/partition.h"
@@ -984,6 +985,16 @@ bool TableInfo::IsSequencesSystemTable() const {
   return false;
 }
 
+bool TableInfo::IsXClusterDDLReplicationDDLQueueTable() const {
+  return GetTableType() == PGSQL_TABLE_TYPE && pgschema_name() == xcluster::kDDLQueuePgSchemaName &&
+         name() == xcluster::kDDLQueueTableName;
+}
+
+bool TableInfo::IsXClusterDDLReplicationReplicatedDDLsTable() const {
+  return GetTableType() == PGSQL_TABLE_TYPE && pgschema_name() == xcluster::kDDLQueuePgSchemaName &&
+         name() == xcluster::kDDLReplicatedTableName;
+}
+
 TablespaceId TableInfo::TablespaceIdForTableCreation() const {
   SharedLock<decltype(lock_)> l(lock_);
   return tablespace_id_for_table_creation_;

src/yb/master/catalog_entity_info.h

@@ -731,6 +731,11 @@ class TableInfo : public RefCountedThreadSafe<TableInfo>,
   bool IsTablegroupParentTable() const;
   bool IsColocatedUserTable() const;
   bool IsSequencesSystemTable() const;
+  bool IsXClusterDDLReplicationDDLQueueTable() const;
+  bool IsXClusterDDLReplicationReplicatedDDLsTable() const;
+  bool IsXClusterDDLReplicationTable() const {
+    return IsXClusterDDLReplicationDDLQueueTable() || IsXClusterDDLReplicationReplicatedDDLsTable();
+  }
 
   // Provides the ID of the tablespace that will be used to determine
   // where the tablets for this table should be placed when the table

src/yb/master/catalog_manager.cc

@@ -3853,6 +3853,11 @@ Status CatalogManager::CreateTable(const CreateTableRequestPB* orig_req,
         indexed_table->LockForRead(), resp));
   }
 
+  // Validate schema.
+  Schema schema;
+  RETURN_NOT_OK(SchemaFromPB(req.schema(), &schema));
+  RETURN_NOT_OK(ValidateCreateTableSchema(schema, resp));
+
   // Pre-colocation GA colocated tables in a legacy colocated database are colocated via database,
   // but after GA, colocated tables are colocated via tablegroups.
   bool is_colocated_via_database;
@@ -3873,7 +3878,9 @@ Status CatalogManager::CreateTable(const CreateTableRequestPB* orig_req,
         // Opt out of colocation if the request says so.
         (!req.has_is_colocated_via_database() || req.is_colocated_via_database()) &&
         // Opt out of colocation if the indexed table opted out of colocation.
-        (!indexed_table || indexed_table->colocated());
+        (!indexed_table || indexed_table->colocated()) &&
+        // Any tables created in the xCluster DDL replication extension should not be colocated.
+        schema.SchemaName() != xcluster::kDDLQueuePgSchemaName;
   }
 
   const bool colocated = is_colocated_via_database || req.has_tablegroup_id();
@@ -3898,11 +3905,6 @@ Status CatalogManager::CreateTable(const CreateTableRequestPB* orig_req,
     }
   }
 
-  // Validate schema.
-  Schema schema;
-  RETURN_NOT_OK(SchemaFromPB(req.schema(), &schema));
-  RETURN_NOT_OK(ValidateCreateTableSchema(schema, resp));
-
   // checking that referenced user-defined types (if any) exist.
   {
     SharedLock lock(mutex_);
@@ -8017,6 +8019,7 @@ bool CatalogManager::IsUserCreatedTable(const TableInfo& table) const {
 bool CatalogManager::IsUserCreatedTableUnlocked(const TableInfo& table) const {
   if (table.GetTableType() == PGSQL_TABLE_TYPE || table.GetTableType() == YQL_TABLE_TYPE) {
     if (!IsSystemTable(table) && !table.IsSequencesSystemTable() &&
+        !table.IsXClusterDDLReplicationTable() &&
         GetNamespaceNameUnlocked(table.namespace_id()) != kSystemNamespaceName &&
         !table.IsColocationParentTable()) {
       return true;
@@ -12092,6 +12095,14 @@ Result<int> CatalogManager::CalculateNumTabletsForTableCreation(
     LOG(INFO) << "Setting default tablets to " << num_tablets << " with " << num_live_tservers
               << " primary servers";
   }
+
+  if (schema.SchemaName() == xcluster::kDDLQueuePgSchemaName &&
+      (request.name() == xcluster::kDDLQueueTableName ||
+       request.name() == xcluster::kDDLReplicatedTableName)) {
+    // xCluster DDL queue tables need to be single tablet tables - This ensures that we have a
+    // singular stream of DDLs which simplifies ordering guarantees.
+    num_tablets = 1;
+  }
   return num_tablets;
 }
 

src/yb/master/tablet_split_manager.cc

@@ -249,6 +249,12 @@ Status TabletSplitManager::ValidateSplitCandidateTable(
         NotSupported,
         "Tablet splitting is not supported for YEDIS tables, table_id: $0", table->id());
   }
+  if (table->IsXClusterDDLReplicationTable()) {
+    return STATUS_FORMAT(
+        NotSupported,
+        "Tablet splitting is not supported for xCluster DDL Replication tables, table_id: $0",
+        table->id());
+  }
 
   auto replication_info = VERIFY_RESULT(filter_->GetTableReplicationInfo(table));
   auto s = filter_->CanAddPartitionsToTable(

src/yb/master/xcluster/master_xcluster_util.cc

@@ -46,8 +46,7 @@ bool IsTableEligibleForXClusterReplication(const master::TableInfo& table) {
     return false;
   }
 
-  if (table.name() == xcluster::kDDLReplicatedTableName &&
-      table.pgschema_name() == xcluster::kDDLQueuePgSchemaName) {
+  if (table.IsXClusterDDLReplicationReplicatedDDLsTable()) {
     // replicated_ddls is only used on the target, so we do not want to replicate it.
     return false;
   }