[#28956] YSQL: Fix transaction self abort with rollback to savepoint

Summary:
With DDL savepoint support, users can rollback DDL operations as part of rollback to savepoint operations.

When a table is created as part of a savepoint and then the savepoint is rolled back, the table must get deleted. In such scenarios, statements
executed after the rollback to savepoint operation are receiving the transaction expired / conflict error. Example:

```
CREATE TABLE projects (id INT PRIMARY KEY, project_name VARCHAR(100));
INSERT INTO projects (id, project_name) VALUES (101, 'Project Alpha');
-- Conflict
BEGIN ISOLATION LEVEL REPEATABLE READ;
UPDATE projects SET project_name = 'Project Phoenix' WHERE id = 101;
SAVEPOINT sp_mixed;
CREATE TABLE project_logs (log_id INT, entry TEXT);
INSERT INTO project_logs VALUES (1, 'Initial log');
ROLLBACK TO SAVEPOINT sp_mixed;
SELECT project_name FROM projects WHERE id = 101; -- gets expired/conflict error
```

```
ERROR:  current transaction is expired or aborted (query layer retry isn't possible because this is not the first command in the transaction. Consider
using READ COMMITTED isolation level.)
```

#### Issue
When the table is deleted, we make DeleteTabletRequestPB request to delete all tablets of the table. As part of the tablet deletion process there is
logic to abort all transactions that are active on the tablet. This leads to the transaction self-abort i.e. the in-flight transaction that initiated
the delete table is also aborted. This leads to future statements within the transaction receiving the transaction expired / conflict error.

#### Why not earlier
This is the first time in YSQL where a tablet deletion is done with an active transaction. Before ddl savepoints, we always deleted tables after the
transaction committed (roll-forward). So the transaction abort during tablet deletion was a no-op.

#### Fix
This revision fixes the issue by not aborting the current transaction if the table is being deleted due to rollback to savepoint operation. We populate the transaction_id in the DeleteTablet request to the tablet server. In the tablet server, we exclude aborting the transaction_id if provided in the request.

**Upgrade/Rollback safety:**
New field is only populated during rollback to savepoint operation which is a new feature and protected under the TEST flag `FLAGS_TEST_ysql_yb_enable_ddl_savepoint_support`.
Jira: DB-18693

Test Plan: ./yb_build.sh --java-test 'org.yb.pgsql.TestDdlSavepoints'

Reviewers: bkolagani, amitanand

Reviewed By: bkolagani

Subscribers: myang, hsunder, svc_phabricator, yql, ybase

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

Author: dr0pdb

src/postgres/src/test/regress/expected/yb.orig.ddl_savepoint.out

@@ -118,3 +118,19 @@ ROLLBACK;
 -- #28955: CREATE INDEX (separate DDL transaction) works when savepoint is enabled.
 CREATE TABLE employees(id INT PRIMARY KEY, code VARCHAR(20) UNIQUE, department VARCHAR(50));
 CREATE INDEX idx_department ON employees(department);
+-- #28956: Transaction doesn't self abort
+CREATE TABLE projects (id INT PRIMARY KEY, project_name VARCHAR(100));
+INSERT INTO projects (id, project_name) VALUES (101, 'Project Alpha');
+BEGIN ISOLATION LEVEL REPEATABLE READ;
+UPDATE projects SET project_name = 'Project Phoenix' WHERE id = 101;
+SAVEPOINT sp_mixed;
+CREATE TABLE project_logs (log_id INT, entry TEXT);
+INSERT INTO project_logs VALUES (1, 'Initial log');
+-- This will drop table project_logs but shouldn't abort the entire transaction.
+ROLLBACK TO SAVEPOINT sp_mixed;
+SELECT project_name FROM projects WHERE id = 101;
+  project_name   
+-----------------
+ Project Phoenix
+(1 row)
+

src/postgres/src/test/regress/sql/yb.orig.ddl_savepoint.sql

@@ -69,3 +69,15 @@ ROLLBACK;
 -- #28955: CREATE INDEX (separate DDL transaction) works when savepoint is enabled.
 CREATE TABLE employees(id INT PRIMARY KEY, code VARCHAR(20) UNIQUE, department VARCHAR(50));
 CREATE INDEX idx_department ON employees(department);
+
+-- #28956: Transaction doesn't self abort
+CREATE TABLE projects (id INT PRIMARY KEY, project_name VARCHAR(100));
+INSERT INTO projects (id, project_name) VALUES (101, 'Project Alpha');
+BEGIN ISOLATION LEVEL REPEATABLE READ;
+UPDATE projects SET project_name = 'Project Phoenix' WHERE id = 101;
+SAVEPOINT sp_mixed;
+CREATE TABLE project_logs (log_id INT, entry TEXT);
+INSERT INTO project_logs VALUES (1, 'Initial log');
+-- This will drop table project_logs but shouldn't abort the entire transaction.
+ROLLBACK TO SAVEPOINT sp_mixed;
+SELECT project_name FROM projects WHERE id = 101;

src/yb/master/async_rpc_tasks.cc

@@ -467,6 +467,9 @@ bool AsyncDeleteReplica::SendRequest(int attempt) {
   bool should_abort_active_txns = !table() ||
                                   table()->LockForRead()->started_deleting();
   req.set_should_abort_active_txns(should_abort_active_txns);
+  if (should_abort_active_txns && exclude_aborting_transaction_id_.has_value()) {
+    req.set_transaction_id(exclude_aborting_transaction_id_->ToString());
+  }
 
   ts_admin_proxy_->DeleteTabletAsync(req, &resp_, &rpc_, BindRpcCallback());
   VLOG_WITH_PREFIX(1) << "Send delete tablet request to " << permanent_uuid_

src/yb/master/async_rpc_tasks.h

@@ -217,6 +217,10 @@ class AsyncDeleteReplica : public RetrySpecificTSRpcTaskWithTable {
     keep_data_ = value;
   }
 
+  void set_exclude_aborting_transaction_id(TransactionId value) {
+    exclude_aborting_transaction_id_ = value;
+  }
+
   TabletId tablet_id() const override { return tablet_id_; }
 
  protected:
@@ -232,6 +236,7 @@ class AsyncDeleteReplica : public RetrySpecificTSRpcTaskWithTable {
   tserver::DeleteTabletResponsePB resp_;
   bool hide_only_ = false;
   bool keep_data_ = false;
+  std::optional<TransactionId> exclude_aborting_transaction_id_{std::nullopt};
 
  private:
   Status SetPendingDelete(AddPendingDelete add_pending_delete);

src/yb/master/catalog_manager.cc

@@ -11007,6 +11007,10 @@ void CatalogManager::SendDeleteTabletRequest(
   }
   if (table != nullptr) {
     table->AddTask(call);
+    TransactionId txn_id = TransactionId::Nil();
+    if (IsTableDeletionDueToRollbackToSubTxn(table, txn_id)) {
+      call->set_exclude_aborting_transaction_id(txn_id);
+    }
   }
 
   auto status = ScheduleTask(call);

src/yb/master/catalog_manager.h

@@ -531,6 +531,14 @@ class CatalogManager : public CatalogManagerIf, public SnapshotCoordinatorContex
                                           const SubTransactionId sub_txn_id,
                                           const LeaderEpoch& epoch);
 
+  // Returns true if the table deletion is happening due to a rollback to sub-transaction operation.
+  // If returning true, the txn_id field is populated with the transaction that it undergoing the
+  // rollback to sub-transaction operation.
+  // NOTE: This function must only be called when we know that the table is being deleted i.e. it
+  // assumes that the table is being deleted.
+  bool IsTableDeletionDueToRollbackToSubTxn(
+      const scoped_refptr<TableInfo>& table, TransactionId& txn_id);
+
   // Rollback all the DDL state changes made by the YSQL transaction from the end till
   // rollback_till_ddl_state_index of ysql_ddl_txn_verifier_state i.e.
   // ysql_ddl_txn_verifier_state[rollback_till_ddl_state_index, end)

src/yb/master/ysql_ddl_handler.cc

@@ -1210,4 +1210,70 @@ Status CatalogManager::IsRollbackDocdbSchemaToSubtxnDone(
              : Status::OK();
 }
 
+bool CatalogManager::IsTableDeletionDueToRollbackToSubTxn(
+    const scoped_refptr<TableInfo>& table, TransactionId& txn_id) {
+  if (!FLAGS_TEST_ysql_yb_enable_ddl_savepoint_support) {
+    return false;
+  }
+
+  // Get the transaction id (if exists) that is modifying the table.
+  {
+    auto l = table->LockForRead();
+    auto txn_id_res = l->GetCurrentDdlTransactionId();
+    if (!txn_id_res.ok() || txn_id_res->IsNil()) {
+      // Transaction ID will always be present in case of rollback to sub-transaction operation.
+      return false;
+    }
+    txn_id = *txn_id_res;
+  }
+
+  SubTransactionId rolled_back_sub_transaction_id;
+  {
+    LockGuard lock(ddl_txn_verifier_mutex_);
+    const auto rollback_to_subtxn_state =
+        FindOrNull(ysql_ddl_txn_undergoing_subtransaction_rollback_map_, txn_id);
+    if (rollback_to_subtxn_state == nullptr) {
+      // No rollback to sub-txn operation is in progress for this transaction.
+      return false;
+    }
+
+    auto tables_itr = std::find_if(
+        rollback_to_subtxn_state->tables.begin(), rollback_to_subtxn_state->tables.end(),
+        [&table](const TableInfoPtr& table_info) {
+          return table_info->id() == table->id(); });
+    if (tables_itr == rollback_to_subtxn_state->tables.end()) {
+      // This table is not undergoing rollback to sub-transaction operation.
+      // Either it has already completed or it wasn't affected.
+      return false;
+    }
+
+    rolled_back_sub_transaction_id = rollback_to_subtxn_state->sub_txn;
+  }
+
+  auto l = table->LockForRead();
+  // Ensure this table is still being modified by the same transaction id.
+  auto txn_id_res = l->GetCurrentDdlTransactionId();
+  if (!txn_id_res.ok() || txn_id_res->IsNil() || txn_id != *txn_id_res) {
+    // Table is now:
+    // 1. Not under any DDL anymore i.e. completed
+    // 2. Undergoing DDL via a different transaction
+    // None of these cases should happen because this would indicate that the transaction finished
+    // and got cleaned up even before the table deletion finished.
+    // Crash in DEBUG so that we can find out cases where this is happening.
+    // Return false in RELEASE to avoid crash. There is no correctness issue with returning false as
+    // this function is used to avoid self-abort of transaction. When we return false, no txn are
+    // excluded i.e. all get aborted as part of tablet deletion.
+    LOG(DFATAL) << "DDL transaction id for table: " << table
+                << " changed while DeleteTable was in progress. old: " << txn_id
+                << ", new: " << txn_id_res;
+    return false;
+  }
+
+  // The table must be created within or after the rolled back sub-transaction.
+  const YsqlDdlTxnVerifierStatePB& first_ddl_state = l->ysql_ddl_txn_verifier_state_first();
+  return first_ddl_state.contains_create_table_op() &&
+         first_ddl_state.has_sub_transaction_id() &&
+         first_ddl_state.sub_transaction_id() >= rolled_back_sub_transaction_id;
+}
+
 } // namespace yb::master

src/yb/tablet/tablet.cc

@@ -5319,14 +5319,16 @@ std::string IncrementedCopy(Slice key) {
 
 } // namespace
 
-Status Tablet::AbortActiveTransactions(CoarseTimePoint deadline) const {
+Status Tablet::AbortActiveTransactions(
+    CoarseTimePoint deadline, std::optional<TransactionId>&& exclude_txn_id) const {
   if (transaction_participant() == nullptr) {
     return Status::OK();
   }
   HybridTime max_cutoff = HybridTime::kMax;
   LOG(INFO) << "Aborting transactions that started prior to " << max_cutoff << " for tablet id "
             << tablet_id();
-  return transaction_participant()->StopActiveTxnsPriorTo(max_cutoff, deadline);
+  return transaction_participant()->StopActiveTxnsPriorTo(
+      max_cutoff, deadline, exclude_txn_id.has_value() ? &*exclude_txn_id : nullptr);
 }
 
 Status Tablet::GetTabletKeyRanges(

src/yb/tablet/tablet.h

@@ -971,7 +971,8 @@ class Tablet : public AbstractTablet,
         max_key_length, std::move(callback), colocated_table_id);
   }
 
-  Status AbortActiveTransactions(CoarseTimePoint deadline) const;
+  Status AbortActiveTransactions(
+      CoarseTimePoint deadline, std::optional<TransactionId>&& exclude_txn_id = std::nullopt) const;
 
   // TODO: Move mutex to private section.
   // Lock used to serialize the creation of RocksDB checkpoints.

src/yb/tablet/tablet_peer.cc

@@ -614,14 +614,15 @@ void TabletPeer::WaitUntilShutdown() {
 
 Status TabletPeer::Shutdown(
     ShouldAbortActiveTransactions should_abort_active_txns,
-    DisableFlushOnShutdown disable_flush_on_shutdown) {
+    DisableFlushOnShutdown disable_flush_on_shutdown,
+    std::optional<TransactionId>&& exclude_aborting_txn_id) {
   auto is_shutdown_initiated = StartShutdown();
 
   if (should_abort_active_txns) {
     // Once raft group state enters QUIESCING state,
     // new queries cannot be processed from then onwards.
     // Aborting any remaining active transactions in the tablet.
-    AbortActiveTransactions();
+    AbortActiveTransactions(std::move(exclude_aborting_txn_id));
   }
 
   if (is_shutdown_initiated) {
@@ -632,14 +633,15 @@ Status TabletPeer::Shutdown(
   return Status::OK();
 }
 
-void TabletPeer::AbortActiveTransactions() const {
+void TabletPeer::AbortActiveTransactions(
+    std::optional<TransactionId>&& exclude_aborting_txn_id) const {
   if (!tablet_) {
     return;
   }
   auto deadline =
       CoarseMonoClock::Now() + MonoDelta::FromMilliseconds(FLAGS_ysql_transaction_abort_timeout_ms);
   WARN_NOT_OK(
-      tablet_->AbortActiveTransactions(deadline),
+      tablet_->AbortActiveTransactions(deadline, std::move(exclude_aborting_txn_id)),
       "Cannot abort transactions for tablet " + tablet_->tablet_id());
 }