Add Rollback

Author: Peng Gu

libspectrum/include/spectrum.h

@@ -40,7 +40,8 @@ enum event_type_enum {
    UPDATE_ROW,
    DELETE_ROW,
    PREPARE,
-   COMMIT
+   COMMIT,
+   ROLLBACK
 };
 }
 
@@ -188,6 +189,7 @@ extern int spectrum_compute_update_row(THD *thd, TABLE *table, const uchar *old_
 extern int spectrum_compute_delete_row(THD *thd, TABLE *table, const uchar *record);
 extern int spectrum_compute_prepare(THD *thd, bool all);
 extern int spectrum_compute_commit(THD *thd, bool all);
+extern int spectrum_compute_rollback(THD *thd, bool all);
 extern int spectrum_compute_begin_attachable_transaction(THD *thd, bool readonly);
 extern int spectrum_compute_end_attachable_transaction(THD *thd);
 extern int spectrum_compute_acquire_mdl(THD *thd, MDL_ticket *ticket);
@@ -209,6 +211,7 @@ extern int spectrum_log_update_row(THD *thd, TABLE *table, uchar *new_row, uchar
 extern int spectrum_log_delete_row(THD *thd, TABLE *table, uchar *row);
 extern int spectrum_log_prepare(THD *thd, bool all, bool real_trans);
 extern int spectrum_log_commit(THD *thd, bool all, bool real_trans);
+extern int spectrum_log_rollback(THD *thd, bool all, bool real_trans);
 extern int spectrum_log_write_commit(THD *thd, commit_id_t commit_id, my_xid xid);
 extern int spectrum_log_write_event(THD *thd, spectrum::Event *event);
 

libspectrum/src/protobuf/spectrum.proto

@@ -34,6 +34,7 @@ service StorageNode {
 
   rpc Prepare(PrepareRequest) returns (PrepareResponse) {}
   rpc Commit(CommitRequest) returns (CommitResponse) {} 
+  rpc Rollback(RollbackRequest) returns (RollbackResponse) {}
 }
 
 service StorageReplicaNode {
@@ -342,6 +343,15 @@ message CommitRequest {
 message CommitResponse {
 }
 
+message RollbackRequest {
+  Thread thread = 1;
+  bool all = 2;
+  uint64 commit_id = 3;
+}
+
+message RollbackResponse {
+}
+
 message BeginAttachableTransactionRequest {
   Thread thread = 1;
   bool readonly = 2;

libspectrum/src/spectrum_compute.cc

@@ -644,6 +644,30 @@ int spectrum_compute_commit(THD *thd, bool all) {
   return 0;
 }
 
+int spectrum_compute_rollback(THD *thd, bool all) {
+  spectrum::RollbackRequest request;
+  spectrum::RollbackResponse response;
+
+  if (thd->spectrum_compute_disabled) {
+    return 0;
+  }
+
+  sql_print_information("spectrum_rollback[%d]: all=%d", thd->spectrum_thread_id, all);
+
+  spectrum::Thread *spectrum_thread = request.mutable_thread();
+  spectrum_thread_fill(thd, spectrum_thread);
+  request.set_all(all);
+
+  grpc::ClientContext context;
+  grpc::Status status = get_storage_primary_client()->Rollback(&context, request, &response);
+  if (!status.ok()) {
+    sql_print_error("spectrum_rollback[%d]: error=%s", thd->spectrum_thread_id, status.error_message().c_str());
+    assert(false);
+  }
+
+  return 0;
+}
+
 int spectrum_compute_begin_attachable_transaction(THD *thd, bool readonly) {
   spectrum::BeginAttachableTransactionRequest request;
   spectrum::BeginAttachableTransactionResponse response;

libspectrum/src/spectrum_log.cc

@@ -499,8 +499,6 @@ class ReplicationStream {
       commit_id_t max_commit_id;
       spectrum::CommitList commits;
       std::set<commit_id_t> prepared_commit_ids_copy;
-      TABLE *commit_table;
-      TABLE *event_table;
 
       spectrum::InitReplicationStreamRequest request;
       spectrum::InitReplicationStreamResponse response;
@@ -761,9 +759,13 @@ int spectrum_log_prepare(THD *thd, bool all, bool real_trans) {
     mysql_mutex_lock(&prepare_lock);
     prepare_locked = true;
     commit_id = next_commit_id();
-    spectrum_log_write_commit(thd, commit_id, xid);
+
+    mysql_mutex_lock(&commit_lock);
     prepared_commit_ids.insert(commit_id);
+    mysql_mutex_unlock(&commit_lock);
+
     storage_thd_context->set_commit_id(commit_id);
+    spectrum_log_write_commit(thd, commit_id, xid);
   }
 
   spectrum::PrepareRequest event_body;
@@ -795,8 +797,9 @@ int spectrum_log_commit(THD *thd, bool all, bool real_trans) {
 
   if (real_trans) {
     mysql_mutex_lock(&commit_lock);
-    commit_locked = true;
     prepared_commit_ids.erase(commit_id);
+    mysql_mutex_unlock(&commit_lock);
+
     storage_thd_context->clear_commit_id();
   }
 
@@ -810,9 +813,35 @@ int spectrum_log_commit(THD *thd, bool all, bool real_trans) {
   if (replication_stream->write(thd, &event, false)) {
     sql_print_error("spectrum_log_commit: stream write error");
   }
+  return 0;
+}
+
+int spectrum_log_rollback(THD *thd, bool all, bool real_trans) {
+  spectrum_storage::THD_context *storage_thd_context = thd->spectrum_storage_context();
+  my_xid xid = storage_thd_context->xid();
+  event_id_t event_id = storage_thd_context->next_event_id();
+  commit_id_t commit_id = storage_thd_context->commit_id();
+  bool commit_locked = false;
+
+  sql_print_information("spectrum_log_rollback: all=%d, real_trans=%d, xid=%d, commit_id=%d", all, real_trans, xid, commit_id);
 
-  if (commit_locked) {
+  if (real_trans) {
+    mysql_mutex_lock(&commit_lock);
+    prepared_commit_ids.erase(commit_id);
     mysql_mutex_unlock(&commit_lock);
+
+    storage_thd_context->clear_commit_id();
+  }
+
+  spectrum::RollbackRequest event_body;
+  spectrum_thread_fill(thd, event_body.mutable_thread());
+  event_body.set_all(all);
+  event_body.set_commit_id(commit_id);
+
+  spectrum::Event event;
+  spectrum_log_build_event(xid, event_id, spectrum::event_type_enum::ROLLBACK, event_body, &event);
+  if (replication_stream->write(thd, &event, false)) {
+    sql_print_error("spectrum_log_rollback: stream write error");
   }
   return 0;
 }

libspectrum/src/spectrum_storage.cc

@@ -101,10 +101,15 @@ THD *create_thd(const spectrum::Thread &spectrum_thread)
   } else {
     thd->query_id = (query_id_t)spectrum_thread.query_id();
   }
-
   return (thd);
 }
 
+void release_thd(THD *thd) {
+  thd->release_resources();
+  Global_THD_manager::get_instance()->remove_thd(thd);
+  delete thd;
+}
+
 bool check_and_coalesce_trx_read_write(THD *thd, bool all) {
   Transaction_ctx::enum_trx_scope trx_scope =
           all ? Transaction_ctx::SESSION : Transaction_ctx::STMT;
@@ -603,6 +608,34 @@ class StorageNodeImpl final : public spectrum::StorageNode::Service {
       return grpc::Status::OK; 
     }
 
+    ::grpc::Status Rollback(::grpc::ServerContext* context, const ::spectrum::RollbackRequest* request, ::spectrum::RollbackResponse* response) {
+      THD *thd = create_thd(request->thread());
+      Transaction_ctx *trn_ctx = thd->get_transaction();
+      bool all = request->all();
+      bool real_trans = (all || !trn_ctx->is_active(Transaction_ctx::SESSION));
+      spectrum_storage::THD_context *thd_storage_context = thd->spectrum_storage_context();
+
+      sql_print_information("Rollback[%d]: all=%d, real_trans=%d", thd->spectrum_thread_id, all, real_trans);
+
+      if (check_and_coalesce_trx_read_write(thd, all)) {
+        spectrum_log_open(thd);
+        spectrum_log_rollback(thd, all, real_trans);
+      } else {
+        sql_print_information("Rollback[%d]: skip spectrum log rollback for readonly transaction", thd->spectrum_thread_id);
+      }
+  
+      ha_rollback_low(thd, all);
+      thd_storage_context->set_prepared(false);
+
+      spectrum_log_close(thd);
+
+      if (real_trans) {
+        trn_ctx->cleanup();
+        thd->tx_priority = 0;
+      }
+      return grpc::Status::OK;
+    }
+
     ::grpc::Status BeginAttachableTransaction(::grpc::ServerContext* context, const ::spectrum::BeginAttachableTransactionRequest* request, ::spectrum::BeginAttachableTransactionResponse* response) {
       THD *thd;
 
@@ -932,7 +965,6 @@ class StorageReplicaNodeImpl final : public spectrum::StorageReplicaNode::Servic
         ha_prepare_low(thd, false);
         ha_commit_low(thd, false);
       }
-
       ha_prepare_low(thd, all);
       thd_storage_context->set_prepared(true);
       return 0;
@@ -972,6 +1004,30 @@ class StorageReplicaNodeImpl final : public spectrum::StorageReplicaNode::Servic
       return 0;
     }
 
+    int Rollback(spectrum::Event &event) {
+      spectrum::RollbackRequest request;
+      google::protobuf::TextFormat::ParseFromString(event.body(), &request);
+
+      THD *thd = create_thd(request.thread());
+      spectrum_storage::THD_context *thd_storage_context = thd->spectrum_storage_context();
+      bool all = request.all();
+      commit_id_t commit_id = request.commit_id();
+
+      sql_print_information("Rollback: all=%d, commit_id=%d", all, commit_id);
+
+      close_thread_tables(thd);
+
+      ha_rollback_low(thd, all);
+      thd_storage_context->set_prepared(false);
+
+      spectrum_log_close(thd);
+
+      if (all) {
+        thd->mdl_context.release_transactional_locks();
+      }
+      return 0;
+    }
+
     ::grpc::Status InitReplicationStream(::grpc::ServerContext* context, const ::spectrum::InitReplicationStreamRequest* request, ::spectrum::InitReplicationStreamResponse* response) {
       mysql_mutex_lock(&replication_lock);
 
@@ -1019,6 +1075,10 @@ class StorageReplicaNodeImpl final : public spectrum::StorageReplicaNode::Servic
             Commit(event);
             response.set_event_id(event.id());
             stream->Write(response);
+          } else if (event_type == spectrum::event_type_enum::ROLLBACK) {
+            Rollback(event);
+            response.set_event_id(event.id());
+            stream->Write(response);
           }
 
           mysql_mutex_unlock(&replication_lock);

mysql-test/r/spectrum.result

@@ -24,3 +24,11 @@ select * from employees where id=1;
 id	first_name	last_name	hire_date	salary
 1	John	Doe	2024-12-01	555.00
 commit;
+update employees set salary=999 where id=1;
+select * from employees where id=1;
+id	first_name	last_name	hire_date	salary
+1	John	Doe	2024-12-01	999.00
+rollback;
+select * from employees where id=1;
+id	first_name	last_name	hire_date	salary
+1	John	Doe	2024-12-01	888.00

mysql-test/t/spectrum.test

@@ -32,4 +32,9 @@ set session transaction isolation level READ COMMITTED;
 select * from employees where id=1;
 
 connection compute0;
-commit;
+commit;
+
+update employees set salary=999 where id=1;
+select * from employees where id=1;
+rollback;
+select * from employees where id=1;

storage/innobase/handler/ha_innodb.cc

@@ -5942,6 +5942,10 @@ static int innobase_rollback(handlerton *hton, /*!< in: InnoDB handlerton */
   assert(hton == innodb_hton_ptr);
   DBUG_PRINT("trans", ("aborting transaction"));
 
+  if (is_spectrum_compute()) {
+    return spectrum_compute_rollback(thd, rollback_trx);
+  }
+
   trx_t *trx = check_trx_exists(thd);
 
   TrxInInnoDB trx_in_innodb(trx);