[raft] ability to remove gtid from logged gtids

Summary:
port D21375044 (facebook@7fc9eaa) (facebook@7fc9eaa) and D17291883 (facebook@6a6a35f) (facebook@6a6a35f)
**Summary**
* As part of leadership changes, we will trim binlogs/relay logs on a
running mysql instance. This needs the ability to remove gtids from executed_gtid set.
This diff adds the ability to do the same. One can enqueue an event in the raft
listener queue and the raft thread in the server will take care of removing and
updating executed_gtid on a running instance.
* When raft logs are trimmed as part of TruncateOpsAfter(), the gtid of the trimmed trxs are also removed from executed/logged gtids. However, when binlog files are rotated, the previous gtid written to the new file depends on what files are rotated. During  binlog rotation the prev gtid are the logged/executed gtids of the instance. During relay log rotation prev gtids are the retrieved gtids of the instance. Hence, in addition to trimming logged gtids/executed gtids, we should also trim retrieved gtids. This prevents creation of holes in the instances executed_gtid set as the replicaset goes through multiple promotions over its lifetime.

Reviewed By: luqun

Differential Revision: D24690007

---------------------------------------------------------------------------------------

Use DBUG_TRACE instead of DBUG_ENTER

Summary:
several testcase failed in asan debug,

==106552==ERROR: AddressSanitizer: stack-buffer-underflow on address 0x7f38f3590078 at pc 0x0000093aa495 bp 0x7f38f358ffc0 sp 0x7f38f358ffb8
READ of size 4 at 0x7f38f3590078 thread T203
    #0 0x93aa494 in _db_enter_(char const*, int, char const*, unsigned int, _db_stack_frame_*)
    #1 0x8d0b9bb in Relay_log_info::remove_logged_gtids
    #2 0x8b1ec3f in trim_logged_gtid
    #3 0x8c767cf in process_raft_queue

If Use DBUG_ENTER, then you need to use DBUG_RETURN to pop current frame in CODE_STATE.
If use DBUG_TRACE, it will pop current frame during .dtor

ps. small refactor changes for sql/rpl_handler.cc

Reviewed By: bhatvinay

Differential Revision: D28809418

sql/binlog.cc

@@ -14246,6 +14246,29 @@ void MYSQL_BIN_LOG::finish_transaction_in_engines(THD *thd, bool all, bool run_a
   }
 }
 
+int trim_logged_gtid(const std::vector<std::string> &trimmed_gtids) {
+  if (trimmed_gtids.empty()) return 0;
+
+  global_tsid_lock->rdlock();
+  int error = gtid_state->remove_logged_gtid_on_trim(trimmed_gtids);
+  Master_info *active_mi = nullptr;
+  if (!get_and_lock_master_info(&active_mi)) {
+    // NO_LINT_DEBUG
+    sql_print_information(
+        "active_mi or rli is not set. Hence not trimming "
+        "logged gtids from rli");
+  }
+  if (active_mi && active_mi->rli) {
+    // Remove rli logged gtids. Note that retrieved gtid is not cleared here
+    // since it is going to be updated when the next gtid is fetched
+    error = active_mi->rli->remove_logged_gtids(trimmed_gtids);
+    unlock_master_info(active_mi);
+  }
+  global_tsid_lock->unlock();
+
+  return error;
+}
+
 struct st_mysql_storage_engine binlog_storage_engine = {
     MYSQL_HANDLERTON_INTERFACE_VERSION};
 

sql/binlog.h

@@ -32,6 +32,7 @@
 #include <mutex>
 #include <unordered_map>
 #include <utility>
+#include <vector>
 
 #include "my_dbug.h"
 #include "my_inttypes.h"
@@ -1643,6 +1644,7 @@ int query_error_code(const THD *thd, bool not_killed);
 bool show_raft_status(THD *thd);
 bool get_and_lock_master_info(Master_info **master_info);
 void unlock_master_info(Master_info *master_info);
+int trim_logged_gtid(const std::vector<std::string> &trimmed_gtids);
 
 void set_valid_pos(my_off_t *valid_pos, const std::string &cur_binlog_file,
                    my_off_t first_gtid_start, char *engine_binlog_file,

sql/rpl_gtid.h

@@ -3161,6 +3161,13 @@ class Gtid_state {
   */
   int32 get_gtid_wait_count() { return atomic_gtid_wait_count; }
 
+  /**
+    Remove gtid from logged_gtid when binlog gets trimmed.
+    @param trimmed_gtids The gtids to remove from logged_gtids
+  */
+  enum_return_status remove_logged_gtid_on_trim(
+      const std::vector<std::string> &trimmed_gtids);
+
 #endif  // ifdef MYSQL_SERVER
   /**
     Computes the next available GNO.

sql/rpl_gtid_state.cc

@@ -22,6 +22,9 @@
 
 #include <time.h>
 #include <atomic>
+#include <cinttypes>
+#include <string>
+#include <vector>
 
 #include "lex_string.h"
 #include "my_compiler.h"
@@ -38,6 +41,7 @@
 #include "sql/binlog.h"
 #include "sql/current_thd.h"
 #include "sql/debug_sync.h"  // DEBUG_SYNC
+#include "sql/log.h"
 #include "sql/mdl.h"
 #include "sql/mysqld.h"  // opt_bin_log
 #include "sql/rpl_context.h"
@@ -964,3 +968,47 @@ enum_return_status Gtid_state::ensure_commit_group_sidnos(rpl_sidno sidno) {
   BINLOG_ERROR(("Out of memory."), (ER_OUT_OF_RESOURCES, MYF(0)));
   RETURN_REPORTED_ERROR;
 }
+
+enum_return_status Gtid_state::remove_logged_gtid_on_trim(
+    const std::vector<std::string> &trimmed_gtids) {
+  DBUG_TRACE;
+  global_tsid_lock->assert_some_lock();
+
+  if (trimmed_gtids.empty()) RETURN_OK;
+
+  const auto &first_gtid = trimmed_gtids.front();
+
+  Gtid gtid;
+  if (gtid.parse(global_tsid_map, first_gtid.c_str()) !=
+      mysql::utils::Return_status::ok) {
+    // NO_LINT_DEBUG
+    sql_print_error("Failed to parse gtid %s", first_gtid.c_str());
+    RETURN_REPORTED_ERROR;
+  }
+
+  rpl_sidno first_sidno = gtid.sidno;
+  tsid_locks.lock(first_sidno);
+
+  for (const auto &trimmed_gtid : trimmed_gtids) {
+    if (gtid.parse(global_tsid_map, trimmed_gtid.c_str()) !=
+        mysql::utils::Return_status::ok) {
+      // NO_LINT_DEBUG
+      sql_print_error("Failed to parse gtid %s", trimmed_gtid.c_str());
+      tsid_locks.unlock(first_sidno);
+      RETURN_REPORTED_ERROR;
+    }
+    assert(first_sidno == gtid.sidno);
+    if (gtid.sidno > 0) {
+      /* Remove Gtid from logged_gtid set. */
+      DBUG_PRINT("info",
+                 ("Removing gtid(sidno:%d, gno:%" PRId64 ") from logged gtids",
+                  gtid.sidno, gtid.gno));
+      // TODO(pgl) - confirm once if the below code is good?
+      executed_gtids._remove_gtid(gtid);
+      gtids_only_in_table._remove_gtid(gtid);
+    }
+  }
+
+  tsid_locks.unlock(first_sidno);
+  RETURN_OK;
+}

sql/rpl_handler.cc

@@ -83,6 +83,7 @@ extern int raft_stop_io_thread(THD *thd);
 extern int raft_start_sql_thread(THD *thd);
 extern int rli_relay_log_raft_reset(
     std::pair<std::string, unsigned long long> raft_log_applied_upto_pos);
+extern int trim_logged_gtid(const std::vector<std::string> &trimmed_gtids);
 /** end of raft related extern funtion declarations  **/
 
 Trans_delegate *transaction_delegate;
@@ -1769,6 +1770,10 @@ extern "C" void *process_raft_queue(void *) {
         result.error = handle_read_only(element.arg.val_sys_var_uint);
         break;
       }
+      case RaftListenerCallbackType::TRIM_LOGGED_GTIDS: {
+        result.error = trim_logged_gtid(element.arg.trim_gtids);
+        break;
+      }
       case RaftListenerCallbackType::ROTATE_BINLOG: {
         result.error = rotate_binlog_file(current_thd);
         break;

sql/rpl_rli.cc

@@ -29,6 +29,7 @@
 #include <algorithm>
 #include <chrono>
 #include <regex>
+#include <vector>
 
 #include "mutex_lock.h"  // Mutex_lock
 #include "my_bitmap.h"
@@ -1902,6 +1903,34 @@ int Relay_log_info::rli_init_info(bool skip_received_gtid_set_recovery) {
   return error;
 }
 
+int Relay_log_info::remove_logged_gtids(
+    const std::vector<std::string> &trimmed_gtids) {
+  DBUG_TRACE;
+  global_tsid_lock->assert_some_lock();
+
+  if (trimmed_gtids.empty()) RETURN_OK;
+
+  Gtid gtid;
+  for (const auto &trimmed_gtid : trimmed_gtids) {
+    if (gtid.parse(global_tsid_map, trimmed_gtid.c_str()) !=
+        mysql::utils::Return_status::ok) {
+      // NO_LINT_DEBUG
+      sql_print_error("Failed to parse gtid %s", trimmed_gtid.c_str());
+      RETURN_REPORTED_ERROR;
+    }
+
+    if (gtid.sidno > 0) {
+      /* Remove Gtid from logged_gtid set. */
+      DBUG_PRINT("info", ("Removing gtid(sidno:%d, gno:%" PRId64
+                          ") from rli logged gtids",
+                          gtid.sidno, gtid.gno));
+      gtid_set->_remove_gtid(gtid);
+    }
+  }
+
+  RETURN_OK;
+}
+
 void Relay_log_info::end_info() {
   DBUG_TRACE;
 

sql/rpl_rli.h

@@ -862,6 +862,8 @@ class Relay_log_info : public Rpl_info {
   */
   enum_return_status add_gtid_set(const Gtid_set *gtid_set);
 
+  int remove_logged_gtids(const std::vector<std::string> &trimmed_gtids);
+
   const Gtid_set *get_gtid_set() const { return gtid_set; }
 
   bool reinit_sql_thread_io_cache(const char *log, bool need_data_lock);