MDEV-8469 Add RESET MASTER TO x to allow specification of binlog file nr

Other things:
- Avoid calling init_and_set_log_file_name() when opening binary log.
- Remove newlines early when reading from index file.
- Ensure that reset_logs() will work even if thd is 0 (Can happen on startup)
- Added thd to sart_slave_threads() for better error handling.

mysql-test/r/create_drop_binlog.result

@@ -333,3 +333,15 @@ Log_name	Pos	Event_type	Server_id	End_log_pos	Info
 #	#	Gtid	1	#	GTID #-#-#
 #	#	Query	1	#	use `test`; DROP TABLE `t1` /* generated by server */
 RESET MASTER;
+RESET MASTER;
+SHOW MASTER STATUS;
+File	Position	Binlog_Do_DB	Binlog_Ignore_DB
+master-bin.000001	313		
+RESET MASTER TO 100;
+SHOW MASTER STATUS;
+File	Position	Binlog_Do_DB	Binlog_Ignore_DB
+master-bin.000100	313		
+RESET MASTER;
+SHOW MASTER STATUS;
+File	Position	Binlog_Do_DB	Binlog_Ignore_DB
+master-bin.000001	313		

mysql-test/t/create_drop_binlog.test

@@ -160,3 +160,12 @@ DROP TABLE t1;
 --replace_regex /xid=[0-9]+/xid=XX/ /GTID [0-9]+-[0-9]+-[0-9]+/GTID #-#-#/ /Server.ver.*/VER/
 SHOW BINLOG EVENTS;
 RESET MASTER;
+
+# Test RESET MASTER TO
+
+RESET MASTER;
+SHOW MASTER STATUS;
+RESET MASTER TO 100;
+SHOW MASTER STATUS;
+RESET MASTER;
+SHOW MASTER STATUS;

sql/log.h

@@ -313,19 +313,22 @@ public:
 #endif
             const char *log_name,
             enum_log_type log_type,
-            const char *new_name,
+            const char *new_name, ulong next_file_number,
             enum cache_type io_cache_type_arg);
   bool init_and_set_log_file_name(const char *log_name,
                                   const char *new_name,
+                                  ulong next_log_number,
                                   enum_log_type log_type_arg,
                                   enum cache_type io_cache_type_arg);
   void init(enum_log_type log_type_arg,
             enum cache_type io_cache_type_arg);
   void close(uint exiting);
   inline bool is_open() { return log_state != LOG_CLOSED; }
-  const char *generate_name(const char *log_name, const char *suffix,
+  const char *generate_name(const char *log_name,
+                            const char *suffix,
                             bool strip_ext, char *buff);
-  int generate_new_name(char *new_name, const char *log_name);
+  int generate_new_name(char *new_name, const char *log_name,
+                        ulong next_log_number);
  protected:
   /* LOCK_log is inited by init_pthread_objects() */
   mysql_mutex_t LOCK_log;
@@ -366,7 +369,7 @@ public:
                 key_file_slow_log,
 #endif
                 generate_name(log_name, "-slow.log", 0, buf),
-                LOG_NORMAL, 0, WRITE_CACHE);
+                LOG_NORMAL, 0, 0, WRITE_CACHE);
   }
   bool open_query_log(const char *log_name)
   {
@@ -376,7 +379,7 @@ public:
                 key_file_query_log,
 #endif
                 generate_name(log_name, ".log", 0, buf),
-                LOG_NORMAL, 0, WRITE_CACHE);
+                LOG_NORMAL, 0, 0, WRITE_CACHE);
   }
 
 private:
@@ -707,6 +710,7 @@ public:
   bool open(const char *log_name,
             enum_log_type log_type,
             const char *new_name,
+            ulong next_log_number,
 	    enum cache_type io_cache_type_arg,
 	    ulong max_size,
             bool null_created,
@@ -780,7 +784,8 @@ public:
   int purge_index_entry(THD *thd, ulonglong *decrease_log_space,
                         bool need_mutex);
   bool reset_logs(THD* thd, bool create_new_log,
-                  rpl_gtid *init_state, uint32 init_state_len);
+                  rpl_gtid *init_state, uint32 init_state_len,
+                  ulong next_log_number);
   void close(uint exiting);
   void clear_inuse_flag_when_closing(File file);
 

sql/mysqld.cc

@@ -5292,7 +5292,7 @@ static int init_server_components()
      * but to be able to have mysql_mutex_assert_owner() in code,
      * we do it anyway */
     mysql_mutex_lock(mysql_bin_log.get_log_lock());
-    if (mysql_bin_log.open(opt_bin_logname, LOG_BIN, 0,
+    if (mysql_bin_log.open(opt_bin_logname, LOG_BIN, 0, 0,
                            WRITE_CACHE, max_binlog_size, 0, TRUE))
       unireg_abort(1);
     mysql_mutex_unlock(mysql_bin_log.get_log_lock());

sql/rpl_mi.cc

@@ -1127,12 +1127,13 @@ bool Master_info_index::init_all_master_info()
 
       if (!opt_skip_slave_start)
       {
-        if (start_slave_threads(1 /* need mutex */,
-              0 /* no wait for start*/,
-              mi,
-              buf_master_info_file,
-              buf_relay_log_info_file,
-              SLAVE_IO | SLAVE_SQL))
+        if (start_slave_threads(current_thd,
+                                1 /* need mutex */,
+                                0 /* no wait for start*/,
+                                mi,
+                                buf_master_info_file,
+                                buf_relay_log_info_file,
+                                SLAVE_IO | SLAVE_SQL))
         {
           sql_print_error("Failed to create slave threads for connection '%.*s'",
                           (int) connection_name.length,
@@ -1455,7 +1456,7 @@ bool Master_info_index::start_all_slaves(THD *thd)
         if (error < 0)                            // fatal error
           break;
       }
-      else
+      else if (thd)
         push_warning_printf(thd, Sql_condition::WARN_LEVEL_NOTE,
                             ER_SLAVE_STARTED, ER_THD(thd, ER_SLAVE_STARTED),
                             (int) mi->connection_name.length,
@@ -1471,6 +1472,8 @@ bool Master_info_index::start_all_slaves(THD *thd)
 
    Start all slaves that was not running.
 
+   @param	thread id from user
+
    @return
    TRUE  	Error
    FALSE	Everything ok.
@@ -1481,6 +1484,7 @@ bool Master_info_index::stop_all_slaves(THD *thd)
   bool result= FALSE;
   DBUG_ENTER("warn_if_slave_running");
   mysql_mutex_assert_owner(&LOCK_active_mi);
+  DBUG_ASSERT(thd);
 
   for (uint i= 0; i< master_info_hash.records; ++i)
   {

sql/rpl_rli.cc

@@ -228,7 +228,7 @@ a file name for --relay-log-index option", opt_relaylog_index_name);
       but a destructor will take care of that
     */
     if (rli->relay_log.open_index_file(buf_relaylog_index_name, ln, TRUE) ||
-        rli->relay_log.open(ln, LOG_BIN, 0, SEQ_READ_APPEND,
+        rli->relay_log.open(ln, LOG_BIN, 0, 0, SEQ_READ_APPEND,
                             mi->rli.max_relay_log_size, 1, TRUE))
     {
       mysql_mutex_unlock(&rli->data_lock);
@@ -1076,6 +1076,9 @@ void Relay_log_info::close_temporary_tables()
 /*
   purge_relay_logs()
 
+  @param rli		Relay log information
+  @param thd		thread id. May be zero during startup
+
   NOTES
     Assumes to have a run lock on rli and that no slave thread are running.
 */
@@ -1131,7 +1134,7 @@ int purge_relay_logs(Relay_log_info* rli, THD *thd, bool just_reset,
     rli->cur_log_fd= -1;
   }
 
-  if (rli->relay_log.reset_logs(thd, !just_reset, NULL, 0))
+  if (rli->relay_log.reset_logs(thd, !just_reset, NULL, 0, 0))
   {
     *errmsg = "Failed during log reset";
     error=1;

sql/slave.cc

@@ -430,7 +430,8 @@ int init_slave()
 
   if (active_mi->host[0] && !opt_skip_slave_start)
   {
-    if (start_slave_threads(1 /* need mutex */,
+    if (start_slave_threads(0, /* No active thd */
+                            1 /* need mutex */,
                             0 /* no wait for start*/,
                             active_mi,
                             master_info_file,
@@ -887,7 +888,8 @@ int start_slave_thread(
     started the threads that were not previously running
 */
 
-int start_slave_threads(bool need_slave_mutex, bool wait_for_start,
+int start_slave_threads(THD *thd,
+                        bool need_slave_mutex, bool wait_for_start,
                         Master_info* mi, const char* master_info_fname,
                         const char* slave_info_fname, int thread_mask)
 {
@@ -933,7 +935,7 @@ int start_slave_threads(bool need_slave_mutex, bool wait_for_start,
                                 mi->rli.group_master_log_pos);
     strmake(mi->master_log_name, mi->rli.group_master_log_name,
             sizeof(mi->master_log_name)-1);
-    purge_relay_logs(&mi->rli, NULL, 0, &errmsg);
+    purge_relay_logs(&mi->rli, thd, 0, &errmsg);
     mi->rli.group_master_log_pos= mi->master_log_pos;
     strmake(mi->rli.group_master_log_name, mi->master_log_name,
             sizeof(mi->rli.group_master_log_name)-1);

sql/slave.h

@@ -176,7 +176,8 @@ bool flush_relay_log_info(Relay_log_info* rli);
 int register_slave_on_master(MYSQL* mysql);
 int terminate_slave_threads(Master_info* mi, int thread_mask,
 			     bool skip_lock = 0);
-int start_slave_threads(bool need_slave_mutex, bool wait_for_start,
+int start_slave_threads(THD *thd,
+                        bool need_slave_mutex, bool wait_for_start,
 			Master_info* mi, const char* master_info_fname,
 			const char* slave_info_fname, int thread_mask);
 /*

sql/sql_lex.h

@@ -2489,6 +2489,7 @@ public:
   USER_RESOURCES mqh;
   LEX_RESET_SLAVE reset_slave_info;
   ulonglong type;
+  ulong next_binlog_file_number;
   /* The following is used by KILL */
   killed_state kill_signal;
   killed_type  kill_type;

sql/sql_reload.cc

@@ -334,7 +334,7 @@ bool reload_acl_and_cache(THD *thd, unsigned long long options,
   {
     DBUG_ASSERT(thd);
     tmp_write_to_binlog= 0;
-    if (reset_master(thd, NULL, 0))
+    if (reset_master(thd, NULL, 0, thd->lex->next_binlog_file_number))
     {
       /* NOTE: my_error() has been already called by reset_master(). */
       result= 1;

sql/sql_repl.cc

@@ -3110,7 +3110,8 @@ int start_slave(THD* thd , Master_info* mi,  bool net_report)
                      ER_THD(thd, ER_UNTIL_COND_IGNORED));
 
       if (!slave_errno)
-        slave_errno = start_slave_threads(0 /*no mutex */,
+        slave_errno = start_slave_threads(thd,
+                                          0 /*no mutex */,
                                           1 /* wait for start */,
                                           mi,
                                           master_info_file_tmp,
@@ -3772,7 +3773,8 @@ err:
   @retval 0 success
   @retval 1 error
 */
-int reset_master(THD* thd, rpl_gtid *init_state, uint32 init_state_len)
+int reset_master(THD* thd, rpl_gtid *init_state, uint32 init_state_len,
+                 ulong next_log_number)
 {
   if (!mysql_bin_log.is_open())
   {
@@ -3782,7 +3784,8 @@ int reset_master(THD* thd, rpl_gtid *init_state, uint32 init_state_len)
     return 1;
   }
 
-  if (mysql_bin_log.reset_logs(thd, 1, init_state, init_state_len))
+  if (mysql_bin_log.reset_logs(thd, 1, init_state, init_state_len,
+                               next_log_number))
     return 1;
   RUN_HOOK(binlog_transmit, after_reset_master, (thd, 0 /* flags */));
   return 0;

sql/sql_repl.h

@@ -46,7 +46,8 @@ int stop_slave(THD* thd, Master_info* mi, bool net_report);
 bool change_master(THD* thd, Master_info* mi, bool *master_info_added);
 bool mysql_show_binlog_events(THD* thd);
 int reset_slave(THD *thd, Master_info* mi);
-int reset_master(THD* thd, rpl_gtid *init_state, uint32 init_state_len);
+int reset_master(THD* thd, rpl_gtid *init_state, uint32 init_state_len,
+                 ulong next_log_number);
 bool purge_master_logs(THD* thd, const char* to_log);
 bool purge_master_logs_before_date(THD* thd, time_t purge_time);
 bool log_in_use(const char* log_name);

sql/sql_yacc.yy

@@ -13037,7 +13037,12 @@ reset_option:
           SLAVE               { Lex->type|= REFRESH_SLAVE; }
           optional_connection_name
           slave_reset_options { }
-        | MASTER_SYM          { Lex->type|= REFRESH_MASTER; }
+        | MASTER_SYM
+          {
+             Lex->type|= REFRESH_MASTER;
+             Lex->next_binlog_file_number= 0;
+          }
+          master_reset_options
         | QUERY_SYM CACHE_SYM { Lex->type|= REFRESH_QUERY_CACHE;}
         ;
 
@@ -13046,6 +13051,14 @@ slave_reset_options:
         | ALL         { Lex->reset_slave_info.all= true; }
         ;
 
+master_reset_options:
+          /* empty */ {}
+        | TO_SYM ulong_num
+          {
+            Lex->next_binlog_file_number = $2;
+          }
+        ;
+
 purge:
           PURGE
           {

sql/sys_vars.cc

@@ -1715,7 +1715,7 @@ Sys_var_gtid_binlog_state::global_update(THD *thd, set_var *var)
   struct gtid_binlog_state_data *data=
     (struct gtid_binlog_state_data *)var->save_result.ptr;
   mysql_mutex_unlock(&LOCK_global_system_variables);
-  res= (0 != reset_master(thd, data->list, data->list_len));
+  res= (reset_master(thd, data->list, data->list_len, 0) != 0);
   mysql_mutex_lock(&LOCK_global_system_variables);
   my_free(data->list);
   my_free(data);

sql/wsrep_mysqld.cc

@@ -502,12 +502,13 @@ static void wsrep_synced_cb(void* app_ctx)
     wsrep_restart_slave_activated= FALSE;
 
     mysql_mutex_lock(&LOCK_active_mi);
-    if ((rcode = start_slave_threads(1 /* need mutex */,
-                            0 /* no wait for start*/,
-                            active_mi,
-                            master_info_file,
-                            relay_log_info_file,
-                            SLAVE_SQL)))
+    if ((rcode = start_slave_threads(0,
+                                     1 /* need mutex */,
+                                     0 /* no wait for start*/,
+                                     active_mi,
+                                     master_info_file,
+                                     relay_log_info_file,
+                                     SLAVE_SQL)))
     {
       WSREP_WARN("Failed to create slave threads: %d", rcode);
     }