WL#7353 Recover relay log by truncating half written transactions [1/3]

Step 1. Log sanitizer - common class for relay/binary log recovery

This step refactors Binlog_recovery class and methods to be re-usable
during the relay log recovery process. Introduced changes:
New Log_sanitizer class, containing utilities for:
- iteration over events in existing log file/files and extracting:
  - the last valid log position
  - the last valid source log file and position
- removal of log files containing only partially-written transactions from disk
  and the index file
- truncation of the last log file to contain only fully written transactions

Some of the functionalities are used only for relay log recovery. Binary log
recovery will never remove binlog file, since binary log file always start
after the last finished transaction. When relay log recovery is considered, we
need to find a start place, which will be either:
- a Query Event containing the end of the transaction:
  - COMMIT / ROLLBACK / XA COMMIT / XA ROLLBACK
- a Query Event containing an atomic DDL
- a source Rotate Event
- a Xid Event
Afterwards, searching for the last valid position begins. If no valid position
can be found in relay log files (e.g. we cannot decrypt relay log files),
relay log recovery won't perform any action.

Change-Id: If76694d2dfa0c62ff7b7243cff9bea451117b376

Author: Karolina Szczepankiewicz

share/messages_to_error_log.txt

@@ -12686,6 +12686,9 @@ ER_IB_LONG_ROLLBACK
 ER_INVALID_FILE_FORMAT
   eng "Invalid format of file '%s'."
 
+ER_LOG_SANITIZATION
+  eng "Relay log sanitization: %s"
+
 # DO NOT add server-to-client messages here;
 # they go in messages_to_clients.txt
 # in the same directory as this file.

sql/CMakeLists.txt

@@ -1020,6 +1020,7 @@ SET(BINLOG_SOURCE
   basic_istream.cc
   basic_ostream.cc
   binlog/global.cc
+  binlog/log_sanitizer.cc
   binlog/recovery.cc
   binlog/group_commit/bgc_ticket_manager.cc
   binlog/group_commit/bgc_ticket.cc
@@ -1078,6 +1079,7 @@ TARGET_LINK_LIBRARIES(rpl_source binlog sql_main)
 
 SET (RPL_REPLICA_SRCS
   dynamic_ids.cc
+  binlog/log_sanitizer.cc
   changestreams/apply/commit_order_queue.cc
   changestreams/apply/replication_thread_status.cc
   rpl_applier_reader.cc

sql/binlog.cc

@@ -1903,26 +1903,43 @@ int MYSQL_BIN_LOG::gtid_end_transaction(THD *thd) {
   return 0;
 }
 
+std::pair<std::list<std::string>, mysql::utils::Error>
+MYSQL_BIN_LOG::get_filename_list() {
+  std::pair<std::list<std::string>, mysql::utils::Error> result;
+  auto &[filename_list, internal_error] = result;
+  LOG_INFO linfo;
+  int error = 0;
+  std::string error_message;
+  MUTEX_LOCK(g, &LOCK_index);
+  try {
+    for (error = find_log_pos(&linfo, nullptr, false); !error;
+         error = find_next_log(&linfo, false)) {
+      filename_list.push_back(string(linfo.log_file_name));
+    }
+  } catch (std::bad_alloc &) {
+    internal_error = mysql::utils::Error("MYSQL_BIN_LOG", __FILE__, __LINE__,
+                                         "Out of memory");
+  }
+  if (error != LOG_INFO_EOF) {
+    internal_error = mysql::utils::Error("MYSQL_BIN_LOG", __FILE__, __LINE__,
+                                         "Error while reading index file");
+  }
+  return result;
+}
+
 bool MYSQL_BIN_LOG::reencrypt_logs() {
   DBUG_TRACE;
 
   if (!is_open()) return false;
 
   std::string error_message;
-  /* Gather the set of files to be accessed. */
-  list<string> filename_list;
   LOG_INFO linfo;
   int error = 0;
   list<string>::reverse_iterator rit;
 
   /* Read binary/relay log file names from index file. */
-  mysql_mutex_lock(&LOCK_index);
-  for (error = find_log_pos(&linfo, nullptr, false); !error;
-       error = find_next_log(&linfo, false)) {
-    filename_list.push_back(string(linfo.log_file_name));
-  }
-  mysql_mutex_unlock(&LOCK_index);
-  if (error != LOG_INFO_EOF ||
+  auto [filename_list, internal_error] = get_filename_list();
+  if (internal_error.is_error() ||
       DBUG_EVALUATE_IF("fail_to_open_index_file", true, false)) {
     error_message.assign("I/O error reading index file '");
     error_message.append(index_file_name);
@@ -3296,37 +3313,41 @@ int query_error_code(const THD *thd, bool not_killed) {
   buffers would just make things slower and more complicated.
   In most cases the copy loop should only do one read.
 
-  @param from          File to copy.
-  @param to            File to copy to.
-  @param offset        Offset in 'from' file.
+  @param from        File to copy.
+  @param to          File to copy to.
+  @param offset      Offset in 'from' file.
+  @param end_pos     End position in 'from' file up to which content should be
+                     copied; 0 means copying till the end of file
 
 
   @retval
     0    ok
   @retval
     -1    error
 */
-static bool copy_file(IO_CACHE *from, IO_CACHE *to, my_off_t offset) {
+static bool copy_file(IO_CACHE *from, IO_CACHE *to, my_off_t offset,
+                      my_off_t end_pos = 0) {
   int bytes_read;
   uchar io_buf[IO_SIZE * 2];
   DBUG_TRACE;
+  unsigned int bytes_written = 0;
 
   mysql_file_seek(from->file, offset, MY_SEEK_SET, MYF(0));
   while (true) {
     if ((bytes_read = (int)mysql_file_read(from->file, io_buf, sizeof(io_buf),
                                            MYF(MY_WME))) < 0)
-      goto err;
+      return true;
     if (DBUG_EVALUATE_IF("fault_injection_copy_part_file", 1, 0))
       bytes_read = bytes_read / 2;
     if (!bytes_read) break;  // end of file
+    if (end_pos != 0 && (bytes_written + bytes_read > end_pos - offset)) {
+      bytes_read = end_pos - offset - bytes_written;
+    }
+    bytes_written += bytes_read;
     if (mysql_file_write(to->file, io_buf, bytes_read, MYF(MY_WME | MY_NABP)))
-      goto err;
+      return true;
   }
-
   return false;
-
-err:
-  return true;
 }
 
 /**
@@ -5868,59 +5889,57 @@ int MYSQL_BIN_LOG::close_crash_safe_index_file() {
   return error;
 }
 
-/**
-  Remove logs from index file.
-
-  - To make it crash safe, we copy the content of the index file
-  from index_file_start_offset recorded in log_info to a
-  crash safe index file first and then move the crash
-  safe index file to the index file.
+int MYSQL_BIN_LOG::remove_logs_outside_range_from_index(
+    const std::string &first, const std::string &last) {
+  LOG_INFO first_linfo;
+  LOG_INFO last_linfo;
 
-  @param log_info               Store here the found log file name and
-                                position to the NEXT log file name in
-                                the index file.
-
-  @param need_update_threads    If we want to update the log coordinates
-                                of all threads. False for relay logs,
-                                true otherwise.
+  MUTEX_LOCK(g, &LOCK_index);
+  int error = find_log_pos(&first_linfo, first.c_str(), false);
+  if (error) return error;
+  error = find_log_pos(&last_linfo, last.c_str(), false);
+  if (error) return error;
+  return remove_logs_outside_range_from_index(&first_linfo, is_relay_log,
+                                              &last_linfo);
+}
 
-  @retval
-    0    ok
-  @retval
-    LOG_INFO_IO    Got IO error while reading/writing file
-*/
-int MYSQL_BIN_LOG::remove_logs_from_index(LOG_INFO *log_info,
-                                          bool need_update_threads) {
+int MYSQL_BIN_LOG::remove_logs_outside_range_from_index(
+    LOG_INFO *start_log_info, bool need_update_threads,
+    LOG_INFO *last_log_info) {
+  my_off_t end_offset = 0;
   if (open_crash_safe_index_file()) {
     LogErr(ERROR_LEVEL, ER_BINLOG_CANT_OPEN_TMP_INDEX,
-           "MYSQL_BIN_LOG::remove_logs_from_index");
+           "MYSQL_BIN_LOG::remove_logs_outside_range_from_index");
     goto err;
   }
+  if (last_log_info != nullptr) {
+    end_offset = last_log_info->index_file_offset;
+  }
 
   if (copy_file(&index_file, &crash_safe_index_file,
-                log_info->index_file_start_offset)) {
+                start_log_info->index_file_start_offset, end_offset)) {
     LogErr(ERROR_LEVEL, ER_BINLOG_CANT_COPY_INDEX_TO_TMP,
-           "MYSQL_BIN_LOG::remove_logs_from_index");
+           "MYSQL_BIN_LOG::remove_logs_outside_range_from_index");
     goto err;
   }
 
   if (close_crash_safe_index_file()) {
     LogErr(ERROR_LEVEL, ER_BINLOG_CANT_CLOSE_TMP_INDEX,
-           "MYSQL_BIN_LOG::remove_logs_from_index");
+           "MYSQL_BIN_LOG::remove_logs_outside_range_from_index");
     goto err;
   }
   DBUG_EXECUTE_IF("fault_injection_copy_part_file", DBUG_SUICIDE(););
 
   if (move_crash_safe_index_file_to_index_file(
           false /*need_lock_index=false*/)) {
     LogErr(ERROR_LEVEL, ER_BINLOG_CANT_MOVE_TMP_TO_INDEX,
-           "MYSQL_BIN_LOG::remove_logs_from_index");
+           "MYSQL_BIN_LOG::remove_logs_outside_range_from_index");
     goto err;
   }
 
   // now update offsets in index file for running threads
   if (need_update_threads)
-    adjust_linfo_offsets(log_info->index_file_start_offset);
+    adjust_linfo_offsets(start_log_info->index_file_start_offset);
   return 0;
 
 err:
@@ -6028,7 +6047,8 @@ int MYSQL_BIN_LOG::purge_logs(const char *to_log, bool included,
   }
 
   /* We know how many files to delete. Update index file. */
-  if ((error = remove_logs_from_index(&log_info, need_update_threads))) {
+  if ((error = remove_logs_outside_range_from_index(&log_info,
+                                                    need_update_threads))) {
     LogErr(ERROR_LEVEL, ER_BINLOG_PURGE_LOGS_CANT_UPDATE_INDEX_FILE);
     goto err;
   }

sql/binlog.h

@@ -319,6 +319,10 @@ class MYSQL_BIN_LOG : public TC_LOG {
                             char *buff);
   bool is_open() const { return atomic_log_state != LOG_CLOSED; }
 
+  /// @brief Obtains the list of logs from the index file
+  /// @return List of log filenames
+  std::pair<std::list<std::string>, mysql::utils::Error> get_filename_list();
+
   /* This is relay log */
   bool is_relay_log;
 
@@ -857,7 +861,36 @@ class MYSQL_BIN_LOG : public TC_LOG {
  public:
   void make_log_name(char *buf, const char *log_ident);
   bool is_active(const char *log_file_name) const;
-  int remove_logs_from_index(LOG_INFO *linfo, bool need_update_threads);
+
+  /// @brief Remove logs from index file, except files between 'start' and
+  /// 'last'
+  /// @details To make it crash safe, we copy the content of the index file
+  /// from index_file_start_offset recorded in log_info to a
+  /// crash safe index file first and then move the crash
+  /// safe index file to the index file.
+  /// @param start_log_info         Metadata of the first log to be kept
+  ///                               in the index file
+  /// @param need_update_threads    If we want to update the log coordinates
+  ///                               of all threads. False for relay logs,
+  ///                               true otherwise.
+  /// @param last_log_info Metadata of the last log to be kept in the index
+  /// file; nullptr means that all logs after start_log_info will be kept
+  /// @retval
+  ///   0    ok
+  /// @retval
+  ///   LOG_INFO_IO    Got IO error while reading/writing file
+  int remove_logs_outside_range_from_index(LOG_INFO *start_log_info,
+                                           bool need_update_threads,
+                                           LOG_INFO *last_log_info = nullptr);
+  /// @brief Remove logs from index file except logs between first and last
+  /// @param first Filename of the first relay log to be kept in index file
+  /// @param last Filename of the last relay log to be kept in index file
+
+  /// @retval 0 OK
+  /// @retval LOG_INFO_IO    Got IO error while reading/writing file
+  /// @retval LOG_INFO_EOF   Could not find requested log file (first or last)
+  int remove_logs_outside_range_from_index(const std::string &first,
+                                           const std::string &last);
   int rotate(bool force_rotate, bool *check_purge);
 
   /**