diff --git a/src/yb/common/schema.h b/src/yb/common/schema.h
index 31c9feb45733..0b6b9b623f2a 100644
--- a/src/yb/common/schema.h
+++ b/src/yb/common/schema.h
@@ -133,6 +133,10 @@ class ColumnSchema {
     return a.order_ == b.order_;
   }
 
+  static bool CompMarkedForDeletion(const ColumnSchema &a, const ColumnSchema &b) {
+    return a.marked_for_deletion_ == b.marked_for_deletion_;
+  }
+
   // Combined comparators.
   static bool CompareType(const ColumnSchema &a, const ColumnSchema &b) {
     return CompNullable(a, b) && CompKind(a, b) && CompTypeInfo(a, b);
@@ -142,6 +146,10 @@ class ColumnSchema {
     return CompareType(a, b) && CompName(a, b);
   }
 
+  static bool CompareDdlAtomicity(const ColumnSchema &a, const ColumnSchema &b) {
+    return CompareByDefault(a, b) && CompMarkedForDeletion(a, b);
+  }
+
   // name: column name
   // type: column type (e.g. UINT8, INT32, STRING, MAP<INT32, STRING> ...)
   // is_nullable: true if a row value can be null
diff --git a/src/yb/master/catalog_entity_info.cc b/src/yb/master/catalog_entity_info.cc
index 50f605ac88ec..bd79a5a2196e 100644
--- a/src/yb/master/catalog_entity_info.cc
+++ b/src/yb/master/catalog_entity_info.cc
@@ -519,16 +519,24 @@ TableType TableInfo::GetTableType() const {
   return LockForRead()->pb.table_type();
 }
 
-bool TableInfo::IsBeingDroppedDueToDdlTxn(const std::string& pb_txn_id, bool txn_success) const {
+bool TableInfo::IsBeingDroppedDueToDdlTxn(
+    const std::string& pb_txn_id, std::optional<bool> txn_success) const {
   auto l = LockForRead();
   if (l->pb_transaction_id() != pb_txn_id) {
     return false;
   }
+  if (!txn_success.has_value()) {
+    // This means the DDL txn only increments the schema version and does not change
+    // the DocDB schema at all. It cannot be one of the following 2 cases.
+    DCHECK(!l->is_being_created_by_ysql_ddl_txn());
+    DCHECK(!l->is_being_deleted_by_ysql_ddl_txn());
+    return false;
+  }
   // The table can be dropped in 2 cases due to a DDL:
   // 1. This table was created by a transaction that subsequently aborted.
   // 2. This is a successful transaction that DROPs the table.
-  return (l->is_being_created_by_ysql_ddl_txn() && !txn_success) ||
-         (l->is_being_deleted_by_ysql_ddl_txn() && txn_success);
+  return (l->is_being_created_by_ysql_ddl_txn() && !*txn_success) ||
+         (l->is_being_deleted_by_ysql_ddl_txn() && *txn_success);
 }
 
 Status TableInfo::AddTablet(const TabletInfoPtr& tablet) {
diff --git a/src/yb/master/catalog_entity_info.h b/src/yb/master/catalog_entity_info.h
index 1cbcedc0b7a0..9444d2e7fba4 100644
--- a/src/yb/master/catalog_entity_info.h
+++ b/src/yb/master/catalog_entity_info.h
@@ -614,7 +614,8 @@ class TableInfo : public RefCountedThreadSafe<TableInfo>,
     return GetTableType() == REDIS_TABLE_TYPE;
   }
 
-  bool IsBeingDroppedDueToDdlTxn(const std::string& txn_id_pb, bool txn_success) const;
+  bool IsBeingDroppedDueToDdlTxn(
+      const std::string& txn_id_pb, std::optional<bool> txn_success) const;
 
   // Add a tablet to this table.
   Status AddTablet(const TabletInfoPtr& tablet);
diff --git a/src/yb/master/catalog_manager.cc b/src/yb/master/catalog_manager.cc
index b75f0bd3f790..4e5c8f7ab370 100644
--- a/src/yb/master/catalog_manager.cc
+++ b/src/yb/master/catalog_manager.cc
@@ -4385,7 +4385,7 @@ Status CatalogManager::CreateTableIfNotFound(
 void CatalogManager::ScheduleVerifyTablePgLayer(TransactionMetadata txn,
                                                 const TableInfoPtr& table,
                                                 const LeaderEpoch& epoch) {
-  auto when_done = [this, table, epoch](Result<bool> exists) {
+  auto when_done = [this, table, epoch](Result<std::optional<bool>> exists) {
     WARN_NOT_OK(VerifyTablePgLayer(table, exists, epoch), "Failed to verify table");
   };
   TableSchemaVerificationTask::CreateAndStartTask(
@@ -4394,10 +4394,13 @@ void CatalogManager::ScheduleVerifyTablePgLayer(TransactionMetadata txn,
 }
 
 Status CatalogManager::VerifyTablePgLayer(
-    scoped_refptr<TableInfo> table, Result<bool> exists, const LeaderEpoch& epoch) {
+    scoped_refptr<TableInfo> table, Result<std::optional<bool>> exists, const LeaderEpoch& epoch) {
   if (!exists.ok()) {
     return exists.status();
   }
+  auto opt_exists = exists.get();
+  SCHECK(opt_exists.has_value(), IllegalState,
+         Substitute("Unexpected opt_exists for $0", table->ToString()));
   // Upon Transaction completion, check pg system table using OID to ensure SUCCESS.
   auto l = table->LockForWrite();
   auto* mutable_table_info = table->mutable_metadata()->mutable_dirty();
@@ -4409,7 +4412,7 @@ Status CatalogManager::VerifyTablePgLayer(
           "Unexpected table state ($0), abandoning transaction GC work for $1",
           SysTablesEntryPB_State_Name(metadata.state()), table->ToString()));
 
-  if (exists.get()) {
+  if (*opt_exists) {
     // Remove the transaction from the entry since we're done processing it.
     metadata.clear_transaction();
     RETURN_NOT_OK(sys_catalog_->Upsert(epoch, table));
diff --git a/src/yb/master/catalog_manager.h b/src/yb/master/catalog_manager.h
index 47582dca646d..5d283a369a59 100644
--- a/src/yb/master/catalog_manager.h
+++ b/src/yb/master/catalog_manager.h
@@ -183,7 +183,13 @@ YB_DEFINE_ENUM(YsqlDdlVerificationState,
     (kDdlPostProcessing)
     (kDdlPostProcessingFailed));
 
-YB_DEFINE_ENUM(TxnState, (kUnknown) (kCommitted) (kAborted));
+// kNoChange is used when a PG DDL statement only increments the table's schema version
+// without any real DocDB table schema change (e.g., alter table t alter column c set not null).
+// In this case we cannot decide whether the PG DDL txn has committed or aborted by doing
+// schema comparison of DocDB current schema or previous schema against the PG catalog schema.
+// That's fine because whether the PG DDL txn has committed or aborted makes no difference for
+// this table's DocDB schema.
+YB_DEFINE_ENUM(TxnState, (kUnknown) (kCommitted) (kAborted) (kNoChange));
 
 struct YsqlTableDdlTxnState;
 
@@ -352,7 +358,7 @@ class CatalogManager : public tserver::TabletPeerLookupIf,
       const TableInfoPtr& table, const LeaderEpoch& epoch);
 
   // Called when transaction associated with table create finishes. Verifies postgres layer present.
-  Status VerifyTablePgLayer(scoped_refptr<TableInfo> table, Result<bool> exists,
+  Status VerifyTablePgLayer(scoped_refptr<TableInfo> table, Result<std::optional<bool>> exists,
     const LeaderEpoch& epoch);
 
   // Truncate the specified table.
@@ -474,16 +480,17 @@ class CatalogManager : public tserver::TabletPeerLookupIf,
 
   Status YsqlTableSchemaChecker(TableInfoPtr table,
                                 const std::string& pb_txn_id,
-                                Result<bool> is_committed,
+                                Result<std::optional<bool>> is_committed,
                                 const LeaderEpoch& epoch);
 
   Status YsqlDdlTxnCompleteCallback(const std::string& pb_txn_id,
-                                    bool is_committed,
+                                    std::optional<bool> is_committed,
                                     const LeaderEpoch& epoch,
                                     const std::string& debug_caller_info);
 
   Status YsqlDdlTxnCompleteCallbackInternal(
-      TableInfo* table, const TransactionId& txn_id, bool success, const LeaderEpoch& epoch);
+      TableInfo* table, const TransactionId& txn_id, std::optional<bool> success,
+      const LeaderEpoch& epoch);
 
   Status HandleSuccessfulYsqlDdlTxn(const YsqlTableDdlTxnState txn_data);
 
diff --git a/src/yb/master/ysql_ddl_handler.cc b/src/yb/master/ysql_ddl_handler.cc
index 7ad4bb3feda3..3f3e94321856 100644
--- a/src/yb/master/ysql_ddl_handler.cc
+++ b/src/yb/master/ysql_ddl_handler.cc
@@ -140,7 +140,7 @@ Status CatalogManager::ScheduleVerifyTransaction(
             << " id: " << table->id() << " schema version: " << l->pb.version()
             << " for transaction " << txn;
   const string txn_id_pb = l->pb_transaction_id();
-  auto when_done = [this, table, txn_id_pb, epoch](Result<bool> is_committed) {
+  auto when_done = [this, table, txn_id_pb, epoch](Result<std::optional<bool>> is_committed) {
     WARN_NOT_OK(YsqlTableSchemaChecker(table, txn_id_pb, is_committed, epoch),
                 "YsqlTableSchemaChecker failed");
   };
@@ -152,7 +152,7 @@ Status CatalogManager::ScheduleVerifyTransaction(
 
 Status CatalogManager::YsqlTableSchemaChecker(TableInfoPtr table,
                                               const string& pb_txn_id,
-                                              Result<bool> is_committed,
+                                              Result<std::optional<bool>> is_committed,
                                               const LeaderEpoch& epoch) {
   if (!is_committed.ok()) {
     auto txn = VERIFY_RESULT(FullyDecodeTransactionId(pb_txn_id));
@@ -178,7 +178,7 @@ Status CatalogManager::YsqlTableSchemaChecker(TableInfoPtr table,
 }
 
 Status CatalogManager::YsqlDdlTxnCompleteCallback(const string& pb_txn_id,
-                                                  bool is_committed,
+                                                  std::optional<bool> is_committed,
                                                   const LeaderEpoch& epoch,
                                                   const std::string& debug_caller_info) {
   SCHECK(!pb_txn_id.empty(), IllegalState,
@@ -188,7 +188,8 @@ Status CatalogManager::YsqlDdlTxnCompleteCallback(const string& pb_txn_id,
 
   auto txn = VERIFY_RESULT(FullyDecodeTransactionId(pb_txn_id));
   LOG(INFO) << "YsqlDdlTxnCompleteCallback for transaction "
-            << txn << " is_committed: " << (is_committed ? "true" : "false")
+            << txn << " is_committed: "
+            << (is_committed.has_value() ? (*is_committed ? "true" : "false") : "nullopt")
             << ", debug_caller_info " << debug_caller_info;
 
   vector<TableInfoPtr> tables;
@@ -202,7 +203,8 @@ Status CatalogManager::YsqlDdlTxnCompleteCallback(const string& pb_txn_id,
     }
 
     auto state = verifier_state->state;
-    auto txn_state = is_committed ? TxnState::kCommitted : TxnState::kAborted;
+    auto txn_state = is_committed.has_value() ?
+        (*is_committed ? TxnState::kCommitted : TxnState::kAborted) : TxnState::kNoChange;
     if (state == YsqlDdlVerificationState::kDdlPostProcessing) {
       // We used to return Status::OK() here on the grounds that the txn is
       // already being verified and we assumed verifier_state->tables represent
@@ -220,10 +222,18 @@ Status CatalogManager::YsqlDdlTxnCompleteCallback(const string& pb_txn_id,
       // to process t3. It is fine to reprocess t1 and t2, that will result in a
       // no-op, except for delete table operation for which we'll detect and avoid
       // reprocessing.
-      SCHECK_EQ(txn_state, verifier_state->txn_state, IllegalState,
-                Format("Mismatch in txn_state for transaction $0", txn));
+      // Some alter table DDL statements only increment table schema version and does not make
+      // any table schema change, this is represented by TxnState::kNoChange. In this case
+      // it does not matter whether the PG DDL transaction is committed or aborted.
+      if (txn_state != verifier_state->txn_state &&
+          txn_state != TxnState::kNoChange && verifier_state->txn_state != TxnState::kNoChange) {
+        return STATUS_FORMAT(IllegalState, "Mismatch in txn_state for transaction $0", txn);
+      }
     } else {
-      verifier_state->txn_state = txn_state;
+      if (verifier_state->txn_state != TxnState::kCommitted &&
+          verifier_state->txn_state != TxnState::kAborted) {
+        verifier_state->txn_state = txn_state;
+      }
       verifier_state->state = YsqlDdlVerificationState::kDdlPostProcessing;
     }
     tables = verifier_state->tables;
@@ -301,7 +311,8 @@ struct YsqlTableDdlTxnState {
 };
 
 Status CatalogManager::YsqlDdlTxnCompleteCallbackInternal(
-    TableInfo* table, const TransactionId& txn_id, bool success, const LeaderEpoch& epoch) {
+    TableInfo* table, const TransactionId& txn_id,
+    std::optional<bool> success, const LeaderEpoch& epoch) {
 
   TEST_PAUSE_IF_FLAG(TEST_pause_ddl_rollback);
 
@@ -315,7 +326,8 @@ Status CatalogManager::YsqlDdlTxnCompleteCallbackInternal(
     return Status::OK();
   }
   LOG_WITH_FUNC(INFO) << id << " for transaction " << txn_id
-                      << ": Success: " << (success ? "true" : "false")
+                      << ": Success: "
+                      << (success.has_value() ? (*success ? "true" : "false") : "nullopt")
                       << " ysql_ddl_txn_verifier_state: "
                       << l->ysql_ddl_txn_verifier_state().DebugString();
 
@@ -332,10 +344,17 @@ Status CatalogManager::YsqlDdlTxnCompleteCallbackInternal(
     .ddl_txn_id = txn_id
   };
 
-  if (success) {
-    RETURN_NOT_OK(HandleSuccessfulYsqlDdlTxn(txn_data));
+  if (success.has_value()) {
+    if (*success) {
+      RETURN_NOT_OK(HandleSuccessfulYsqlDdlTxn(txn_data));
+    } else {
+      RETURN_NOT_OK(HandleAbortedYsqlDdlTxn(txn_data));
+    }
   } else {
-    RETURN_NOT_OK(HandleAbortedYsqlDdlTxn(txn_data));
+    // If success is nullopt, it represents a PG DDL statement that only increments the schema
+    // version of this table without any table schema change. There is nothing to do but to
+    // cleanup.
+    RETURN_NOT_OK(ClearYsqlDdlTxnState(txn_data));
   }
   return Status::OK();
 }
@@ -664,7 +683,7 @@ Status CatalogManager::TriggerDdlVerificationIfNeeded(
             << " for transaction " << txn;
 
   const string txn_id_pb = l->pb_transaction_id();
-  auto when_done = [this, table, txn_id_pb, epoch](Result<bool> is_committed) {
+  auto when_done = [this, table, txn_id_pb, epoch](Result<std::optional<bool>> is_committed) {
     WARN_NOT_OK(YsqlTableSchemaChecker(table, txn_id_pb, is_committed, epoch),
                 "YsqlTableSchemaChecker failed");
   };
diff --git a/src/yb/master/ysql_ddl_verification_task.cc b/src/yb/master/ysql_ddl_verification_task.cc
index 2f7289904a0f..8445bc2c4d80 100644
--- a/src/yb/master/ysql_ddl_verification_task.cc
+++ b/src/yb/master/ysql_ddl_verification_task.cc
@@ -101,7 +101,7 @@ Status PgEntryExistsWithReadTime(
 Status PgSchemaCheckerWithReadTime(SysCatalogTable* sys_catalog,
                                    const scoped_refptr<TableInfo>& table,
                                    const ReadHybridTime& read_time,
-                                   bool* result,
+                                   std::optional<bool>* result,
                                    HybridTime* read_restart_ht);
 
 bool MatchPgDocDBSchemaColumns(const scoped_refptr<TableInfo>& table,
@@ -205,8 +205,9 @@ Status PgEntryExistsWithReadTime(
   return Status::OK();
 }
 
-Result<bool> PgSchemaChecker(SysCatalogTable& sys_catalog, const scoped_refptr<TableInfo>& table) {
-  bool result = false;
+Result<std::optional<bool>> PgSchemaChecker(
+    SysCatalogTable& sys_catalog, const scoped_refptr<TableInfo>& table) {
+  std::optional<bool> result{std::nullopt};
   RETURN_NOT_OK(sys_catalog.ReadWithRestarts(
     std::bind(&PgSchemaCheckerWithReadTime, &sys_catalog, table, std::placeholders::_1, &result,
       std::placeholders::_2)));
@@ -216,7 +217,7 @@ Result<bool> PgSchemaChecker(SysCatalogTable& sys_catalog, const scoped_refptr<T
 Status PgSchemaCheckerWithReadTime(SysCatalogTable* sys_catalog,
                                    const scoped_refptr<TableInfo>& table,
                                    const ReadHybridTime& read_time,
-                                   bool* result,
+                                   std::optional<bool>* result,
                                    HybridTime* read_restart_ht) {
   PgOid oid = kPgInvalidOid;
   string pg_catalog_table_id, name_col;
@@ -339,6 +340,7 @@ Status PgSchemaCheckerWithReadTime(SysCatalogTable* sys_catalog,
 
   // Table was being altered. Check whether its current DocDB schema matches
   // that of PG catalog.
+  VLOG(3) << "Comparing with the PG schema for alter table";
   CHECK(l->ysql_ddl_txn_verifier_state().contains_alter_table_op());
   const auto& relname_col = row.GetValue(relname_col_id);
   const string& table_name = relname_col->string_value();
@@ -361,17 +363,28 @@ Status PgSchemaCheckerWithReadTime(SysCatalogTable* sys_catalog,
 
   Schema schema;
   RETURN_NOT_OK(table->GetSchema(&schema));
+  Schema previous_schema;
+  RETURN_NOT_OK(SchemaFromPB(l->ysql_ddl_txn_verifier_state().previous_schema(), &previous_schema));
+  // CompareDdlAtomicity takes marked_for_deletion() into comparison. If a column is marked for
+  // deletion in the current schema and not in the previous schema, then CompareByDefault would
+  // return true which isn't right for correct handling.
+  if (schema.Equals(previous_schema, ColumnSchema::CompareDdlAtomicity)) {
+    VLOG(3) << "The current DocDB schema is the same as the previous DocDB schema";
+    *result = std::nullopt;
+    return Status::OK();
+  }
+
   if (MatchPgDocDBSchemaColumns(table, schema, pg_cols)) {
     // The PG catalog schema matches the current DocDB schema. The transaction was a success.
+    VLOG(3) << "PG schema matches the current DocDB schema";
     *result = true;
     return Status::OK();
   }
 
-  Schema previous_schema;
-  RETURN_NOT_OK(SchemaFromPB(l->ysql_ddl_txn_verifier_state().previous_schema(), &previous_schema));
   if (MatchPgDocDBSchemaColumns(table, previous_schema, pg_cols)) {
     // The PG catalog schema matches the DocDB schema of the table prior to this transaction. The
     // transaction must have aborted.
+    VLOG(3) << "PG schema matches the previous DocDB schema";
     *result = false;
     return Status::OK();
   }
@@ -719,7 +732,8 @@ void NamespaceVerificationTask::PerformAbort() {
 
 TableSchemaVerificationTask::TableSchemaVerificationTask(
     CatalogManager& catalog_manager, scoped_refptr<TableInfo> table,
-    const TransactionMetadata& transaction, std::function<void(Result<bool>)> complete_callback,
+    const TransactionMetadata& transaction,
+    std::function<void(Result<std::optional<bool>>)> complete_callback,
     SysCatalogTable* sys_catalog, std::shared_future<client::YBClient*> client_future,
     rpc::Messenger& messenger, const LeaderEpoch& epoch, bool ddl_atomicity_enabled)
     : MultiStepTableTaskBase(
@@ -741,7 +755,7 @@ void TableSchemaVerificationTask::CreateAndStartTask(
     CatalogManager& catalog_manager,
     scoped_refptr<TableInfo> table,
     const TransactionMetadata& transaction,
-    std::function<void(Result<bool>)> complete_callback,
+    std::function<void(Result<std::optional<bool>>)> complete_callback,
     SysCatalogTable* sys_catalog,
     std::shared_future<client::YBClient*> client_future,
     rpc::Messenger& messenger,
@@ -794,7 +808,7 @@ void TableSchemaVerificationTask::FinishPollTransaction(Status status) {
   }, "Compare Schema");
 }
 
-Status TableSchemaVerificationTask::FinishTask(Result<bool> is_committed) {
+Status TableSchemaVerificationTask::FinishTask(Result<std::optional<bool>> is_committed) {
   RETURN_NOT_OK(is_committed);
 
   is_committed_ = *is_committed;
diff --git a/src/yb/master/ysql_ddl_verification_task.h b/src/yb/master/ysql_ddl_verification_task.h
index 908544315555..d9d76279ba66 100644
--- a/src/yb/master/ysql_ddl_verification_task.h
+++ b/src/yb/master/ysql_ddl_verification_task.h
@@ -149,7 +149,7 @@ class TableSchemaVerificationTask : public MultiStepTableTaskBase,
     CatalogManager& catalog_manager,
     scoped_refptr<TableInfo> table,
     const TransactionMetadata& transaction,
-    std::function<void(Result<bool>)> complete_callback,
+    std::function<void(Result<std::optional<bool>>)> complete_callback,
     SysCatalogTable* sys_catalog,
     std::shared_future<client::YBClient*> client_future,
     rpc::Messenger& messenger,
@@ -172,7 +172,7 @@ class TableSchemaVerificationTask : public MultiStepTableTaskBase,
     CatalogManager& catalog_manager,
     scoped_refptr<TableInfo> table,
     const TransactionMetadata& transaction,
-    std::function<void(Result<bool>)> complete_callback,
+    std::function<void(Result<std::optional<bool>>)> complete_callback,
     SysCatalogTable* sys_catalog,
     std::shared_future<client::YBClient*> client_future,
     rpc::Messenger& messenger,
@@ -185,14 +185,14 @@ class TableSchemaVerificationTask : public MultiStepTableTaskBase,
   Status ValidateRunnable() override;
   Status CheckTableExists(Status s);
   Status CompareSchema(Status s);
-  Status FinishTask(Result<bool> is_committed);
+  Status FinishTask(Result<std::optional<bool>> is_committed);
   void FinishPollTransaction(Status s) override;
   void TaskCompleted(const Status& status) override;
   void PerformAbort() override;
 
   SysCatalogTable& sys_catalog_;
   bool ddl_atomicity_enabled_;
-  bool is_committed_ = false;
+  std::optional<bool> is_committed_{std::nullopt};
 };
 
 }  // namespace master