[#22243] YSQL: Fix incorrect log message when a column is dropped suc…

…cessfully

Summary:
To reproduce the bug:

```
yugabyte=# create table foo(id int);
CREATE TABLE
yugabyte=# alter table foo drop column id;
ALTER TABLE
```

Note that the ALTER TABLE statement completed successfully. Look at yb-master.INFO, saw:

```
I0502 19:26:37.713428 27752 ysql_ddl_handler.cc:396] Sending Alter Table request as part of rollback for table foo
```

The log says `rollback` which suggests the DDL operation failed. But in this
case `roll forward` is more appropriate. We first mark the column id to be dropped
in DocDB metadata for table `foo`, then drop the column from PG catalog metadata
for table `foo`. Once the PG side has successfully committed, we then roll forward
the DocDB metadata change by dropping the column. This completes the entire DDL
operation.

I made a fix to log `roll forward` if the PG side commits successfully, and
`rollback` if the PG side aborts.

Also made some small test code cleanup for DDL atomicity because --ysql_yb_ddl_rollback_enabled
is true by default now.
Jira: DB-11161

Test Plan:
(1) Manually do the test described in the summary, saw the new log:

```
I0502 20:12:33.628252  4412 ysql_ddl_handler.cc:399] Sending Alter Table request as part of roll forward for table foo
```
(2) ./yb_build.sh release --java-test 'org.yb.pgsql.TestPgUniqueConstraint'
(3) YB_ENABLE_YSQL_CONN_MGR_IN_TESTS=true ./yb_build.sh release --java-test 'org.yb.pgsql.TestPgUniqueConstraint'

Reviewers: fizaa

Reviewed By: fizaa

Subscribers: yql, ybase

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

java/yb-pgsql/src/test/java/org/yb/pgsql/TestPgUniqueConstraint.java

@@ -401,13 +401,11 @@ public void createIndexViolatingUniqueness() throws Exception {
       assertTrue(miniCluster.getClient().setFlag(hp,
           "enable_transactional_ddl_gc", "false"));
     }
-    for (HostAndPort hp : miniCluster.getTabletServers().keySet()) {
-      if (isTestRunningWithConnectionManager())
-        assertTrue(miniCluster.getClient().setFlag(hp,
-            "allowed_preview_flags_csv", "ysql_yb_ddl_rollback_enabled,enable_ysql_conn_mgr"));
-      else
-        assertTrue(miniCluster.getClient().setFlag(hp,
-            "allowed_preview_flags_csv", "ysql_yb_ddl_rollback_enabled=true"));
+    if (isTestRunningWithConnectionManager()) {
+      for (HostAndPort hp : miniCluster.getTabletServers().keySet()) {
+          assertTrue(miniCluster.getClient().setFlag(hp,
+              "allowed_preview_flags_csv", "enable_ysql_conn_mgr"));
+      }
     }
     try (Statement stmt = connection.createStatement()) {
       runInvalidQuery(

src/yb/master/catalog_manager.h

@@ -485,7 +485,8 @@ class CatalogManager : public tserver::TabletPeerLookupIf,
 
   Status YsqlDdlTxnAlterTableHelper(const YsqlTableDdlTxnState txn_data,
                                     const std::vector<DdlLogEntry>& ddl_log_entries,
-                                    const std::string& new_table_name);
+                                    const std::string& new_table_name,
+                                    bool success);
 
   Status YsqlDdlTxnDropTableHelper(const YsqlTableDdlTxnState txn_data);
 

src/yb/master/ysql_ddl_handler.cc

@@ -325,7 +325,8 @@ Status CatalogManager::HandleSuccessfulYsqlDdlTxn(
           Format("Drop column $0", col));
   }
   SchemaToPB(builder.Build(), mutable_pb.mutable_schema());
-  return YsqlDdlTxnAlterTableHelper(txn_data, ddl_log_entries, "" /* new_table_name */);
+  return YsqlDdlTxnAlterTableHelper(
+      txn_data, ddl_log_entries, "" /* new_table_name */, true /* success */);
 }
 
 Status CatalogManager::HandleAbortedYsqlDdlTxn(const YsqlTableDdlTxnState txn_data) {
@@ -345,7 +346,8 @@ Status CatalogManager::HandleAbortedYsqlDdlTxn(const YsqlTableDdlTxnState txn_da
     mutable_pb.mutable_schema()->CopyFrom(ddl_state.previous_schema());
     const string new_table_name = ddl_state.previous_table_name();
     mutable_pb.set_name(new_table_name);
-    return YsqlDdlTxnAlterTableHelper(txn_data, ddl_log_entries, new_table_name);
+    return YsqlDdlTxnAlterTableHelper(
+        txn_data, ddl_log_entries, new_table_name, false /* success */);
   }
 
   // This must be a failed Delete transaction.
@@ -367,7 +369,8 @@ Status CatalogManager::ClearYsqlDdlTxnState(const YsqlTableDdlTxnState txn_data)
 
 Status CatalogManager::YsqlDdlTxnAlterTableHelper(const YsqlTableDdlTxnState txn_data,
                                                   const std::vector<DdlLogEntry>& ddl_log_entries,
-                                                  const string& new_table_name) {
+                                                  const string& new_table_name,
+                                                  bool success) {
   auto& table_pb = txn_data.write_lock.mutable_data()->pb;
   const int target_schema_version = table_pb.version() + 1;
   table_pb.set_version(target_schema_version);
@@ -393,7 +396,9 @@ Status CatalogManager::YsqlDdlTxnAlterTableHelper(const YsqlTableDdlTxnState txn
   auto table = txn_data.table;
   table->AddDdlTxnWaitingForSchemaVersion(target_schema_version, txn_data.ddl_txn_id);
 
-  LOG(INFO) << "Sending Alter Table request as part of rollback for table " << table->name();
+  auto action = success ? "roll forward" : "rollback";
+  LOG(INFO) << "Sending Alter Table request as part of " << action
+            << " for table " << table->name();
   return SendAlterTableRequestInternal(table, TransactionId::Nil(), txn_data.epoch);
 }