[#26633] DocDB: Drop invalid index DDL halts the replication in xCluster

Summary:
### Problem
Dropping an INVALID index halts xCluster DDL replication. When CREATE UNIQUE INDEX fails due to a duplicate key violation, the index is left in an INVALID state (`indisvalid=false`) on the source but is never created on the target. Dropping this index causes replication to fail with:
- `index "<invalid_index_name>" does not exist` on the target
- `DDL replication is paused due to repeated failures`

### Root Cause
There is a state mismatch between the source and target catalogs. The CREATE INDEX that failed was never replicated to the target (since ddl_command_end doesn't execute for failed DDLs). However, the subsequent DROP INDEX was replicated, which causes the target to try dropping an index that doesn't exist. This causes the state mismatch between the source and target catalogs: the source has an invalid object and the target has no object. The replication stream sends DDLs assuming the object exists on the target, violating the "exactly one" semantics required for the target.

### Solution
On the target side, when `yb_xcluster_automatic_mode_target_ddl` is set (running in xCluster automatic mode), we implement a "silent skip" for DROP INDEX, RENAME INDEX and attribute/tablespace changes like SET STATISTICS, SET TABLESPACE. By treating these "phantom" objects (failed creation on the source and never existed on the target) as no-ops, the DDL applier can progress through the replication stream without halting. Valid objects are still dropped normally.

Test Plan:
### Automated Tests
```
./yb_build.sh release --cxx-test xcluster_ddl_replication-test --gtest_filter='XClusterDDLReplicationTest.DropInvalidIndexDoesNotHaltReplication'
./yb_build.sh release --cxx-test xcluster_ddl_replication-test --gtest_filter='XClusterDDLReplicationTest.DropPartitionedIndexOnOnlyReplicates'
./yb_build.sh release --cxx-test xcluster_ddl_replication-test --gtest_filter='XClusterDDLReplicationTest.AlterPhantomIndexLifecycleDoesNotHaltReplication'
```

Test 1: DropInvalidIndexDoesNotHaltReplication
- CREATE UNIQUE INDEX fails due to duplicate key, meaning the index is INVALID and not replicated
- DROP INDEX is replicated to target, which fails with "does not exist", and we validate that the error is ignored
- Then verify that replication continues normally

Test 2: DropPartitionedIndexOnOnlyReplicates
- CREATE INDEX ON ONLY creates an INVALID index that IS replicated, meaning the CREATE succeeded
- DROP INDEX is replicated to target, which succeeds since the index exists
- Then verify that we don't incorrectly skip DROP for valid cases

Test 3: AlterPhantomIndexLifecycleDoesNotHaltReplication
- CREATE UNIQUE INDEX (on an expression) fails on source and is not created on target
- ALTER INDEX ... SET STATISTICS and SET TABLESPACE are replicated for the phantom index
- ALTER INDEX ... RENAME TO is replicated, changing the phantom's name on the source
- ALTER TABLE ... SET SCHEMA is replicated, triggering an implicit schema move for the phantom index
- DROP INDEX using the new name is replicated
- Then verify that the target handles the entire lifecycle of attribute, name, and namespace changes on the missing index gracefully, and replication continues
(Note: SET TABLESPACE here goes through the same code path as SET STATISTICS: `T_AlterTableStmt`)

The fixed code passes all four tests. The old code fails Test 1 (because the "does not exist" error halts replication) and Test, but passes Test 2 (because the index was successfully created and replicated, so the DROP succeeds normally).
This is because Test 2 validates the fix doesn't break successful CREATEs of INVALID indexes (meaning the index was replicated).

Did not test DROP on multiple indices since this isn't supported yet in YB. See [issue #880](#880).

Reviewers: jhe, xCluster, hsunder

Reviewed By: jhe

Subscribers: ybase, yql

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

Author: Sid1279

src/postgres/yb-extensions/yb_xcluster_ddl_replication/yb_xcluster_ddl_replication.c

@@ -480,11 +480,52 @@ HandleSourceDDLStart(EventTriggerData *trig_data)
 	ClearRewrittenTableOidList();
 }
 
+/*
+ * xCluster: Prevent replication from halting on phantom indexes (invalid
+ * indexes that exist on the source but not the target). DDLs referencing them
+ * are silently skipped rather than forcing replication to halt.
+ */
+static void
+yb_xcluster_handle_phantom_indexes(Node *parsetree)
+{
+	if (yb_xcluster_automatic_mode_target_ddl)
+	{
+		switch (nodeTag(parsetree))
+		{
+			case T_DropStmt:
+			{
+				DropStmt *stmt = (DropStmt *) parsetree;
+				if (stmt->removeType == OBJECT_INDEX)
+					stmt->missing_ok = true;
+				break;
+			}
+			case T_RenameStmt:
+			{
+				RenameStmt *stmt = (RenameStmt *) parsetree;
+				if (stmt->renameType == OBJECT_INDEX)
+					stmt->missing_ok = true;
+				break;
+			}
+			case T_AlterTableStmt:
+			{
+				AlterTableStmt *stmt = (AlterTableStmt *) parsetree;
+				if (stmt->objtype == OBJECT_INDEX)
+					stmt->missing_ok = true;
+				break;
+			}
+			default:
+				break;
+		}
+	}
+}
+
 void
 HandleTargetDDLStart(EventTriggerData *trig_data)
 {
 	yb_xcluster_target_ddl_bypass = false;
 
+	yb_xcluster_handle_phantom_indexes(parsetree);
+
 	/* Bypass DDLs executed in manual mode, or from the target poller. */
 	if (enable_manual_ddl_replication || yb_xcluster_automatic_mode_target_ddl)
 	{

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

@@ -926,6 +926,161 @@ TEST_F(XClusterDDLReplicationTest, NonconcurrentBackfillsWithPartitions) {
   ASSERT_OK(WaitForSafeTimeToAdvanceToNow());
 }
 
+TEST_F(XClusterDDLReplicationTest, DropInvalidIndexDoesNotHaltReplication) {
+  ASSERT_OK(SetUpClustersAndReplication());
+
+  const std::string kBaseTableName = "test_invalid_idx";
+  const std::string kIndexName = "idx_invalid";
+  const std::string kColumn2Name = "value";
+
+  ASSERT_OK(producer_conn_->ExecuteFormat(
+      "CREATE TABLE $0($1 int PRIMARY KEY, $2 int);", kBaseTableName, kKeyColumnName,
+      kColumn2Name));
+  ASSERT_OK(WaitForSafeTimeToAdvanceToNow());
+
+  // Insert rows with duplicate values to cause unique index creation to fail.
+  ASSERT_OK(producer_conn_->ExecuteFormat(
+      "INSERT INTO $0 VALUES (1, 10), (2, 10), (3, 20);", kBaseTableName));
+  ASSERT_OK(WaitForSafeTimeToAdvanceToNow());
+  ASSERT_OK(VerifyWrittenRecords({kBaseTableName}));
+
+  // Create unique index fails due to duplicate key violation.
+  auto create_result = producer_conn_->ExecuteFormat(
+      "CREATE UNIQUE INDEX $0 ON $1($2)", kIndexName, kBaseTableName, kColumn2Name);
+  ASSERT_NOK(create_result);
+  ASSERT_STR_CONTAINS(create_result.ToString(), "duplicate key");
+
+  // Verify index is invalid on producer.
+  ASSERT_FALSE(ASSERT_RESULT(IsIndexValid(*producer_conn_, kIndexName)))
+      << "Index should be INVALID";
+
+  // Verify invalid index was not replicated to consumer using consumer_conn_ directly.
+  auto consumer_index_exists = ASSERT_RESULT(consumer_conn_->FetchRow<bool>(Format(
+      "SELECT EXISTS(SELECT 1 FROM pg_class WHERE relname = '$0')", kIndexName)));
+  ASSERT_FALSE(consumer_index_exists) << "INVALID index should not exist on consumer";
+
+  // Drop the invalid index.
+  ASSERT_OK(producer_conn_->ExecuteFormat("DROP INDEX $0", kIndexName));
+
+  // Verify replication continues by writing more rows to the original table.
+  ASSERT_OK(producer_conn_->ExecuteFormat(
+      "INSERT INTO $0 VALUES (4, 30), (5, 40);", kBaseTableName));
+  ASSERT_OK(WaitForSafeTimeToAdvanceToNow());
+  ASSERT_OK(VerifyWrittenRecords({kBaseTableName}));
+}
+
+
+TEST_F(XClusterDDLReplicationTest, DropPartitionedIndexOnOnlyReplicates) {
+  ASSERT_OK(SetUpClustersAndReplication());
+
+  const std::string kParentTableName = "test_partitioned";
+  const std::string kPartitionName = "test_partitioned_p1";
+  const std::string kIndexName = "idx_partitioned_only";
+  const std::string kColumn2Name = "value";
+
+  ASSERT_OK(producer_conn_->ExecuteFormat(
+      "CREATE TABLE $0($1 int PRIMARY KEY, $2 int) PARTITION BY RANGE ($1);",
+      kParentTableName, kKeyColumnName, kColumn2Name));
+  ASSERT_OK(WaitForSafeTimeToAdvanceToNow());
+
+  ASSERT_OK(producer_conn_->ExecuteFormat(
+      "CREATE TABLE $0 PARTITION OF $1 FOR VALUES FROM (1) TO (100);",
+      kPartitionName, kParentTableName));
+  ASSERT_OK(WaitForSafeTimeToAdvanceToNow());
+
+  // ON ONLY creates an index marked invalid until attached to a partition index.
+  ASSERT_OK(producer_conn_->ExecuteFormat(
+      "CREATE INDEX $0 ON ONLY $1($2)", kIndexName, kParentTableName, kColumn2Name));
+  ASSERT_OK(WaitForSafeTimeToAdvanceToNow());
+
+  // Verify index is invalid on producer.
+  ASSERT_FALSE(ASSERT_RESULT(IsIndexValid(*producer_conn_, kIndexName)))
+      << "Index should be INVALID on producer";
+
+  // Verify the invalid index was replicated to consumer.
+  auto consumer_index_exists = ASSERT_RESULT(consumer_conn_->FetchRow<bool>(Format(
+      "SELECT EXISTS(SELECT 1 FROM pg_class WHERE relname = '$0')", kIndexName)));
+  ASSERT_TRUE(consumer_index_exists) << "Partitioned index should exist on consumer";
+
+  // Create a child index and attach it to verify the ATTACH PARTITION path.
+  const std::string kPartIdxName = "idx_partition_child";
+  ASSERT_OK(producer_conn_->ExecuteFormat(
+      "CREATE INDEX $0 ON $1($2)", kPartIdxName, kPartitionName, kColumn2Name));
+  ASSERT_OK(producer_conn_->ExecuteFormat(
+      "ALTER INDEX $0 ATTACH PARTITION $1", kIndexName, kPartIdxName));
+  ASSERT_OK(WaitForSafeTimeToAdvanceToNow());
+
+  // Verify parent index is now valid on both clusters.
+  ASSERT_TRUE(ASSERT_RESULT(IsIndexValid(*producer_conn_, kIndexName)));
+  ASSERT_TRUE(ASSERT_RESULT(IsIndexValid(*consumer_conn_, kIndexName)))
+      << "Attached index did not become valid on consumer";
+
+  // Cleanup: drop the parent index.
+  ASSERT_OK(producer_conn_->ExecuteFormat("DROP INDEX $0", kIndexName));
+  ASSERT_OK(WaitForSafeTimeToAdvanceToNow());
+
+  // Verify the drop was replicated.
+  consumer_index_exists = ASSERT_RESULT(consumer_conn_->FetchRow<bool>(Format(
+      "SELECT EXISTS(SELECT 1 FROM pg_class WHERE relname = '$0')", kIndexName)));
+  ASSERT_FALSE(consumer_index_exists) << "Dropped index should not exist on consumer";
+
+  // Verify replication continues.
+  const std::string kSecondTableName = "verify_replication_works2";
+  ASSERT_OK(producer_conn_->ExecuteFormat(
+      "CREATE TABLE $0($1 int PRIMARY KEY);", kSecondTableName, kKeyColumnName));
+  ASSERT_OK(producer_conn_->ExecuteFormat(
+      "INSERT INTO $0 VALUES (1), (2), (3);", kSecondTableName));
+
+  ASSERT_OK(WaitForSafeTimeToAdvanceToNow());
+  ASSERT_OK(VerifyWrittenRecords({kSecondTableName}));
+}
+
+TEST_F(XClusterDDLReplicationTest, AlterPhantomIndexLifecycleDoesNotHaltReplication) {
+  ASSERT_OK(SetUpClustersAndReplication());
+
+  const std::string kTableName = "schema_test_table";
+  const std::string kIdxName = "idx_phantom_schema";
+  const std::string kIdxNewName = "idx_phantom_renamed";
+  const std::string kSchemaName = "new_index_schema";
+
+  // Setup table with duplicate data and a new schema.
+  ASSERT_OK(producer_conn_->ExecuteFormat(
+      "CREATE TABLE $0(key int PRIMARY KEY, val int);", kTableName));
+  ASSERT_OK(producer_conn_->ExecuteFormat("INSERT INTO $0 VALUES (1, 10), (2, 10);", kTableName));
+  ASSERT_OK(producer_conn_->ExecuteFormat("CREATE SCHEMA $0;", kSchemaName));
+  ASSERT_OK(WaitForSafeTimeToAdvanceToNow());
+
+  // Create unique index fails on producer and is not created on consumer.
+  ASSERT_NOK_STR_CONTAINS(
+      producer_conn_->ExecuteFormat(
+          "CREATE UNIQUE INDEX $0 ON $1((val + 0));", kIdxName, kTableName),
+      "duplicate key");
+
+  // ALTER INDEX attribute change on the phantom index.
+  ASSERT_OK(producer_conn_->ExecuteFormat(
+      "ALTER INDEX $0 ALTER COLUMN 1 SET STATISTICS 100;", kIdxName));
+
+  // ALTER INDEX tablespace change on the phantom index.
+  ASSERT_OK(producer_conn_->Execute("CREATE TABLESPACE dummy_ts LOCATION '/tmp/dummy_path';"));
+  ASSERT_OK(producer_conn_->ExecuteFormat("ALTER INDEX $0 SET TABLESPACE dummy_ts;", kIdxName));
+
+  // Rename the phantom index.
+  ASSERT_OK(producer_conn_->ExecuteFormat("ALTER INDEX $0 RENAME TO $1;", kIdxName, kIdxNewName));
+
+  // Move the parent table to a new schema (indexes follow implicitly).
+  ASSERT_OK(producer_conn_->ExecuteFormat(
+      "ALTER TABLE $0 SET SCHEMA $1;", kTableName, kSchemaName));
+
+  // Drop the phantom index (renamed, now in the new schema).
+  ASSERT_OK(producer_conn_->ExecuteFormat("DROP INDEX $0.$1;", kSchemaName, kIdxNewName));
+
+  // Verify replication continues by writing more rows to the original table.
+  ASSERT_OK(producer_conn_->ExecuteFormat(
+      "INSERT INTO $0.$1 VALUES (3, 30), (4, 40);", kSchemaName, kTableName));
+  ASSERT_OK(WaitForSafeTimeToAdvanceToNow());
+  ASSERT_OK(VerifyWrittenRecords({kTableName}, /*database_name=*/"", kSchemaName));
+}
+
 TEST_F(XClusterDDLReplicationTest, ExactlyOnceReplication) {
   // Test that DDLs are only replicated exactly once.
   const int kNumTablets = 3;

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

@@ -507,6 +507,12 @@ Result<pgwrapper::PGResultPtr> XClusterYsqlTestBase::ScanToStrings(
   return result;
 }
 
+Result<bool> XClusterYsqlTestBase::IsIndexValid(
+    pgwrapper::PGConn& conn, const std::string& index_name) {
+  return conn.FetchRow<bool>(Format(
+      "SELECT indisvalid FROM pg_index WHERE indexrelid = '$0'::regclass", index_name));
+}
+
 Result<int> XClusterYsqlTestBase::GetRowCount(
     const YBTableName& table_name, Cluster* cluster, bool read_latest) {
   auto conn = VERIFY_RESULT(

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

@@ -109,6 +109,8 @@ class XClusterYsqlTestBase : public XClusterTestBase {
   static Result<pgwrapper::PGResultPtr> ScanToStrings(
       const client::YBTableName& table_name, Cluster* cluster);
 
+  static Result<bool> IsIndexValid(pgwrapper::PGConn& conn, const std::string& index_name);
+
   static Result<int> GetRowCount(
       const client::YBTableName& table_name, Cluster* cluster, bool read_latest = false);