PXB-3034 - Handle prepare

https://jira.percona.com/browse/PXB-3034

Adjusted DDL tracking to produce correct files at the end of backup and
handle those files during prepare.

storage/innobase/log/log0recv.cc

@@ -1688,8 +1688,10 @@ static byte *recv_parse_or_apply_log_rec_body(
       return fil_tablespace_redo_delete(
           ptr, end_ptr, page_id_t(space_id, page_no), parsed_bytes,
           recv_sys->bytes_to_ignore_before_checkpoint !=
-              0 IF_XB(|| recv_sys->recovered_lsn + parsed_bytes <
-                             backup_redo_log_flushed_lsn));
+              0 IF_XB(||
+                      recv_sys->recovered_lsn + parsed_bytes <
+                          backup_redo_log_flushed_lsn ||
+                      opt_lock_ddl == LOCK_DDL_REDUCED));
 
     case MLOG_FILE_CREATE:
 
@@ -1702,8 +1704,10 @@ static byte *recv_parse_or_apply_log_rec_body(
       return fil_tablespace_redo_create(
           ptr, end_ptr, page_id_t(space_id, page_no), parsed_bytes,
           recv_sys->bytes_to_ignore_before_checkpoint !=
-              0 IF_XB(|| recv_sys->recovered_lsn + parsed_bytes <
-                             backup_redo_log_flushed_lsn));
+              0 IF_XB(||
+                      recv_sys->recovered_lsn + parsed_bytes <
+                          backup_redo_log_flushed_lsn ||
+                      opt_lock_ddl == LOCK_DDL_REDUCED));
 
     case MLOG_FILE_RENAME:
 #ifdef XTRABACKUP
@@ -1715,8 +1719,10 @@ static byte *recv_parse_or_apply_log_rec_body(
       return fil_tablespace_redo_rename(
           ptr, end_ptr, page_id_t(space_id, page_no), parsed_bytes,
           recv_sys->bytes_to_ignore_before_checkpoint !=
-              0 IF_XB(|| recv_sys->recovered_lsn + parsed_bytes <
-                             backup_redo_log_flushed_lsn));
+              0 IF_XB(||
+                      recv_sys->recovered_lsn + parsed_bytes <
+                          backup_redo_log_flushed_lsn ||
+                      opt_lock_ddl == LOCK_DDL_REDUCED));
 
     case MLOG_FILE_EXTEND:
 

storage/innobase/xtrabackup/src/backup_copy.cc

@@ -1395,6 +1395,35 @@ bool backup_start(Backup_context &context) {
         return (false);
       }
     }
+    /* LTFB/LIFB has to be executed before copying MyISAM */
+    if (ddl_tracker != nullptr) {
+      if (!lock_tables_for_backup(mysql_connection, opt_backup_lock_timeout,
+                                  opt_backup_lock_retry_count)) {
+        return (false);
+      }
+      /**
+       * Gather log_status for ddl_tracker. We want to ensure that when we start
+       * processing DDL entries from ddl_tracker, no new DDL will happen in the
+       * server and we have parsed all DDL from before LTFB. We instruct InnoDB
+       * to flush the log ensuring the log_status query sees the most recent
+       * updates. We gather log_status here (even before copying MyISAM tables)
+       * so we allow the background redo thread to catchup later when we start
+       * to process the DDL's all we care is that redo has parsed at least up to
+       * the LSN where new DDL's are possible, as this will can change the
+       * in-memory structure of ddl_tracker. Ensuring no new DDL happens when we
+       * process handle_ddl_operations allows PXB to operate without a mutex
+       * here.
+       */
+      if (have_flush_engine_logs) {
+        xb::info() << "Executing FLUSH NO_WRITE_TO_BINLOG ENGINE LOGS...";
+        xb_mysql_query(mysql_connection, "FLUSH NO_WRITE_TO_BINLOG ENGINE LOGS",
+                       false);
+      }
+      log_status_get(mysql_connection, true);
+      xb::info() << "DDL tracking :  log_status current checkpoint lsn is "
+                 << log_status.lsn_checkpoint << " and current lsn is "
+                 << log_status.lsn;
+    }
 
     if (!backup_files(MySQL_datadir_path.path().c_str(), true, context)) {
       return (false);
@@ -1451,11 +1480,25 @@ bool backup_start(Backup_context &context) {
     context.myrocks_checkpoint.create(mysql_connection, true);
   }
 
+  if (ddl_tracker != nullptr) {
+    std::this_thread::sleep_for(
+        std::chrono::milliseconds(context.redo_mgr->get_copy_interval()));
+    while (context.redo_mgr->get_scanned_lsn() < log_status.lsn &&
+           !context.redo_mgr->has_parsed_lsn(log_status.lsn)) {
+      xb::info() << "Waiting for redo thread to catchup to LSN "
+                 << log_status.lsn << " (currently parsing at "
+                 << context.redo_mgr->get_parsed_lsn() << ")";
+      std::this_thread::sleep_for(
+          std::chrono::milliseconds(context.redo_mgr->get_copy_interval()));
+    }
+    ddl_tracker->handle_ddl_operations();
+  }
+
   xb::info() << "Executing FLUSH NO_WRITE_TO_BINLOG BINARY LOGS";
   xb_mysql_query(mysql_connection, "FLUSH NO_WRITE_TO_BINLOG BINARY LOGS",
                  false);
 
-  log_status_get(mysql_connection);
+  log_status_get(mysql_connection, false);
 
   /* Wait until we have checkpoint LSN greater than the page tracking start LSN.
   Page tracking start LSN is system LSN (lets say 105) and Backup End LSN is
@@ -1485,7 +1528,7 @@ bool backup_start(Backup_context &context) {
                    << " to reach to page tracking start lsn "
                    << page_tracking_start_lsn;
         std::this_thread::sleep_for(std::chrono::seconds(1));
-        log_status_get(mysql_connection);
+        log_status_get(mysql_connection, false);
       }
     }
   }

storage/innobase/xtrabackup/src/backup_mysql.cc

@@ -1690,10 +1690,7 @@ static void log_status_local_parse(const char *s, log_status_t &log_status) {
   }
 }
 
-/** Read binary log position, InnoDB LSN and other storage engine information
-from p_s.log_status and update global log_status variable.
-@param[in]   conn         mysql connection handle */
-void log_status_get(MYSQL *conn) {
+void log_status_get(MYSQL *conn, bool consumer_can_advance) {
   xb::info() << "Selecting LSN and binary log position from p_s.log_status";
 
   debug_sync_point("log_status_get");
@@ -1702,7 +1699,7 @@ void log_status_get(MYSQL *conn) {
         opt_no_lock || opt_lock_ddl_per_table);
 
   if (xtrabackup_register_redo_log_consumer)
-    redo_log_consumer_can_advance.store(false);
+    redo_log_consumer_can_advance.store(consumer_can_advance);
 
   const char *query =
       "SELECT server_uuid, local, replication, "
@@ -1798,52 +1795,56 @@ char *get_xtrabackup_info(MYSQL *connection) {
   ut_a(uuid);
   ut_a(server_version);
   char *result = NULL;
-  int ret = asprintf(&result,
-                     "uuid = %s\n"
-                     "name = %s\n"
-                     "tool_name = %s\n"
-                     "tool_command = %s\n"
-                     "tool_version = %s\n"
-                     "ibbackup_version = %s\n"
-                     "server_version = %s\n"
-                     "server_flavor = %s\n"
-                     "start_time = %s\n"
-                     "end_time = %s\n"
-                     "lock_time = %ld\n"
-                     "binlog_pos = %s\n"
-                     "innodb_from_lsn = " LSN_PF
-                     "\n"
-                     "innodb_to_lsn = " LSN_PF
-                     "\n"
-                     "partial = %s\n"
-                     "incremental = %s\n"
-                     "format = %s\n"
-                     "compressed = %s\n"
-                     "encrypted = %s\n",
-                     uuid,                           /* uuid */
-                     opt_history ? opt_history : "", /* name */
-                     tool_name,                      /* tool_name */
-                     tool_args,                      /* tool_command */
-                     XTRABACKUP_VERSION,             /* tool_version */
-                     XTRABACKUP_VERSION,             /* ibbackup_version */
-                     server_version,                 /* server_version */
-                     mysql_server_version_comment.c_str(), /* server_flavor */
-                     buf_start_time,                       /* start_time */
-                     buf_end_time,                         /* end_time */
-                     (long int)history_lock_time,          /* lock_time */
-                     mysql_binlog_position.c_str(),        /* binlog_pos */
-                     incremental_lsn,                      /* innodb_from_lsn */
-                     metadata_to_lsn,                      /* innodb_to_lsn */
-                     (xtrabackup_tables                    /* partial */
-                      || xtrabackup_tables_exclude || xtrabackup_tables_file ||
-                      xtrabackup_databases || xtrabackup_databases_exclude ||
-                      xtrabackup_databases_file)
-                         ? "Y"
-                         : "N",
-                     xtrabackup_incremental ? "Y" : "N", /* incremental */
-                     xb_stream_format_name[xtrabackup_stream_fmt], /* format */
-                     xtrabackup_compress ? "compressed" : "N", /* compressed */
-                     xtrabackup_encrypt ? "Y" : "N");          /* encrypted */
+  int ret =
+      asprintf(&result,
+               "uuid = %s\n"
+               "name = %s\n"
+               "tool_name = %s\n"
+               "tool_command = %s\n"
+               "tool_version = %s\n"
+               "ibbackup_version = %s\n"
+               "server_version = %s\n"
+               "server_flavor = %s\n"
+               "start_time = %s\n"
+               "end_time = %s\n"
+               "lock_time = %ld\n"
+               "binlog_pos = %s\n"
+               "innodb_from_lsn = " LSN_PF
+               "\n"
+               "innodb_to_lsn = " LSN_PF
+               "\n"
+               "partial = %s\n"
+               "incremental = %s\n"
+               "format = %s\n"
+               "compressed = %s\n"
+               "encrypted = %s\n"
+               "lock_ddl_type = %s\n",
+               uuid,                                 /* uuid */
+               opt_history ? opt_history : "",       /* name */
+               tool_name,                            /* tool_name */
+               tool_args,                            /* tool_command */
+               XTRABACKUP_VERSION,                   /* tool_version */
+               XTRABACKUP_VERSION,                   /* ibbackup_version */
+               server_version,                       /* server_version */
+               mysql_server_version_comment.c_str(), /* server_flavor */
+               buf_start_time,                       /* start_time */
+               buf_end_time,                         /* end_time */
+               (long int)history_lock_time,          /* lock_time */
+               mysql_binlog_position.c_str(),        /* binlog_pos */
+               incremental_lsn,                      /* innodb_from_lsn */
+               metadata_to_lsn,                      /* innodb_to_lsn */
+               (xtrabackup_tables                    /* partial */
+                || xtrabackup_tables_exclude || xtrabackup_tables_file ||
+                xtrabackup_databases || xtrabackup_databases_exclude ||
+                xtrabackup_databases_file)
+                   ? "Y"
+                   : "N",
+               xtrabackup_incremental ? "Y" : "N",           /* incremental */
+               xb_stream_format_name[xtrabackup_stream_fmt], /* format */
+               xtrabackup_compress ? "compressed" : "N",     /* compressed */
+               xtrabackup_encrypt ? "Y" : "N",               /* encrypted */
+               ddl_lock_type_to_str(static_cast<lock_ddl_type_t>(opt_lock_ddl))
+                   .c_str()); /* lock-ddl */
 
   ut_a(ret != 0);
 

storage/innobase/xtrabackup/src/backup_mysql.h

@@ -205,8 +205,9 @@ bool write_current_binlog_file(MYSQL *connection);
 
 /** Read binary log position, InnoDB LSN and other storage engine information
 from p_s.log_status and update global log_status variable.
-@param[in]   conn         mysql connection handle */
-void log_status_get(MYSQL *conn);
+@param[in]   conn                     mysql connection handle
+@param[in]   consumer_can_advance     bool to define if consumer can advance */
+void log_status_get(MYSQL *conn, bool consumer_can_advance);
 
 /*********************************************************************/ /**
  Retrieves MySQL binlog position and

storage/innobase/xtrabackup/src/ddl_tracker.cc

@@ -118,3 +118,100 @@ void ddl_tracker_t::add_table(const space_id_t &space_id, std::string name) {
   }
   tables_in_backup[space_id] = name;
 }
+
+void ddl_tracker_t::handle_ddl_operations() {
+  // TODO: Make copy multi thread
+
+  xb::info() << "DDL tracking :  handling DDL operations";
+
+  if (new_tables.empty() && renames.empty() && drops.empty() &&
+      recopy_tables.empty()) {
+    xb::info()
+        << "DDL tracking : Finished handling DDL operations - No changes";
+    return;
+  }
+  dberr_t err;
+
+  /* Some tables might get to the new list if the DDL happen in between
+   * redo_mgr.start and xb_load_tablespaces. This causes we ending up with two
+   * tablespaces with the same spaceID. Remove them from new tables */
+  for (auto &table : tables_in_backup) {
+    if (new_tables.find(table.first) != new_tables.end()) {
+      new_tables.erase(table.first);
+    }
+  }
+
+  /* recopy_tables will be handled as follow:
+    * not in the backup - nothign to do. This is a new table that was created
+     during the backup. It will be re-copied anyway as .new in the backup.
+    * in the backup - we add it to be recopied if renamed - we delete the old
+    file during prepare rename logic will then instruct adjust the proper file
+    name to be copied
+  */
+  for (auto &table : recopy_tables) {
+    if (tables_in_backup.find(table) != tables_in_backup.end()) {
+      if (renames.find(table) != renames.end()) {
+        backup_file_printf((renames[table].first + ".del").c_str(), "%s", "");
+      }
+      string name = tables_in_backup[table];
+      new_tables[table] = name;
+    }
+  }
+
+  for (auto &table : drops) {
+    if (check_if_skip_table(table.second.c_str())) {
+      continue;
+    }
+    /* Remove from rename */
+    renames.erase(table.first);
+
+    /* Remove from new tables and skip drop*/
+    if (new_tables.find(table.first) != new_tables.end()) {
+      new_tables.erase(table.first);
+      continue;
+    }
+    backup_file_printf((table.second + ".del").c_str(), "%s", "");
+  }
+
+  for (auto &table : renames) {
+    if (check_if_skip_table(table.second.second.c_str())) {
+      continue;
+    }
+    if (check_if_skip_table(table.second.first.c_str())) {
+      continue;
+    }
+    /* renamed new table. update new table entry to renamed table name */
+    if (new_tables.find(table.first) != new_tables.end()) {
+      new_tables[table.first] = table.second.second;
+      continue;
+    }
+    backup_file_printf((table.second.first + ".ren").c_str(), "%s",
+                       table.second.second.c_str());
+  }
+
+  fil_close_all_files();
+  for (auto table = new_tables.begin(); table != new_tables.end();) {
+    if (check_if_skip_table(table->second.c_str())) {
+      table = new_tables.erase(table);
+      continue;
+    }
+    std::tie(err, std::ignore) = fil_open_for_xtrabackup(
+        table->second, table->second.substr(0, table->second.length() - 4));
+    table++;
+  }
+
+  datafiles_iter_t *it = datafiles_iter_new(nullptr);
+  while (fil_node_t *node = datafiles_iter_next(it)) {
+    if (new_tables.find(node->space->id) == new_tables.end()) {
+      continue;
+    }
+    if (check_if_skip_table(node->name)) {
+      continue;
+    }
+    std::string dest_name = node->name;
+    dest_name.append(".new");
+    xtrabackup_copy_datafile(node, 0, dest_name.c_str());
+  }
+  datafiles_iter_free(it);
+  xb::info() << "DDL tracking :  Finished handling DDL operations";
+}

storage/innobase/xtrabackup/src/ddl_tracker.h

@@ -50,5 +50,7 @@ class ddl_tracker_t {
   @param[in] start_lsn  lsn for REDO record */
   void backup_file_op(uint32_t space_id, mlog_id_t type, const byte *buf,
                       ulint len, lsn_t start_lsn);
+  /** Function responsible to generate files based on DDL operations */
+  void handle_ddl_operations();
 };
 #endif  // DDL_TRACKER_H