rpl_rli.cc 45.6 KB
Newer Older
1
/* Copyright (c) 2006, 2011, Oracle and/or its affiliates. All rights reserved.
2 3 4

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
unknown's avatar
unknown committed
5
   the Free Software Foundation; version 2 of the License.
6 7 8 9 10 11 12 13

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
14
   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA */
15

16 17
#include "sql_priv.h"
#include "unireg.h"                             // HAVE_*
18
#include "rpl_mi.h"
19
#include "rpl_rli.h"
20
#include "sql_base.h"                        // close_thread_tables
21 22
#include <my_dir.h>    // For MY_STAT
#include "sql_repl.h"  // For check_binlog_magic
23 24 25
#include "log_event.h" // Format_description_log_event, Log_event,
                       // FORMAT_DESCRIPTION_LOG_EVENT, ROTATE_EVENT,
                       // PREFIX_SQL_LOAD
26
#include "rpl_utility.h"
Konstantin Osipov's avatar
Konstantin Osipov committed
27
#include "transaction.h"
28
#include "sql_parse.h"                          // end_trans, ROLLBACK
29 30
#include <mysql/plugin.h>
#include <mysql/service_thd_wait.h>
31

32
static int count_relay_log_space(Relay_log_info* rli);
33 34 35 36 37 38

// Defined in slave.cc
int init_intvar_from_file(int* var, IO_CACHE* f, int default_val);
int init_strvar_from_file(char *var, int max_size, IO_CACHE *f,
			  const char *default_val);

39
Relay_log_info::Relay_log_info(bool is_slave_recovery)
40
  :Slave_reporting_capability("SQL"),
41
   no_storage(FALSE), replicate_same_server_id(::replicate_same_server_id),
42
   info_fd(-1), cur_log_fd(-1), relay_log(&sync_relaylog_period),
43
   sync_counter(0), is_relay_log_recovery(is_slave_recovery),
44 45
   save_temporary_tables(0), cur_log_old_open_count(0), group_relay_log_pos(0), 
   event_relay_log_pos(0),
46 47 48
#if HAVE_purify
   is_fake(FALSE),
#endif
49 50
   group_master_log_pos(0), log_space_total(0), ignore_log_space_limit(0),
   last_master_timestamp(0), slave_skip_counter(0),
51
   abort_pos_wait(0), slave_run_id(0), sql_thd(0),
52 53 54
   inited(0), abort_slave(0), slave_running(0), until_condition(UNTIL_NONE),
   until_log_pos(0), retried_trans(0),
   tables_to_lock(0), tables_to_lock_count(0),
55
   last_event_start_time(0), m_flags(0)
56
{
57
  DBUG_ENTER("Relay_log_info::Relay_log_info");
58

59 60 61 62 63 64 65
#ifdef HAVE_PSI_INTERFACE
  relay_log.set_psi_keys(key_RELAYLOG_LOCK_index,
                         key_RELAYLOG_update_cond,
                         key_file_relaylog,
                         key_file_relaylog_index);
#endif

66 67
  group_relay_log_name[0]= event_relay_log_name[0]=
    group_master_log_name[0]= 0;
68
  until_log_name[0]= ign_master_log_name_end[0]= 0;
69 70 71
  bzero((char*) &info_file, sizeof(info_file));
  bzero((char*) &cache_buf, sizeof(cache_buf));
  cached_charset_invalidate();
Marc Alff's avatar
Marc Alff committed
72 73 74 75 76 77 78 79 80
  mysql_mutex_init(key_relay_log_info_run_lock, &run_lock, MY_MUTEX_INIT_FAST);
  mysql_mutex_init(key_relay_log_info_data_lock,
                   &data_lock, MY_MUTEX_INIT_FAST);
  mysql_mutex_init(key_relay_log_info_log_space_lock,
                   &log_space_lock, MY_MUTEX_INIT_FAST);
  mysql_cond_init(key_relay_log_info_data_cond, &data_cond, NULL);
  mysql_cond_init(key_relay_log_info_start_cond, &start_cond, NULL);
  mysql_cond_init(key_relay_log_info_stop_cond, &stop_cond, NULL);
  mysql_cond_init(key_relay_log_info_log_space_cond, &log_space_cond, NULL);
81 82 83 84 85
  relay_log.init_pthread_objects();
  DBUG_VOID_RETURN;
}


86
Relay_log_info::~Relay_log_info()
87
{
88
  DBUG_ENTER("Relay_log_info::~Relay_log_info");
89

Marc Alff's avatar
Marc Alff committed
90 91 92 93 94 95 96
  mysql_mutex_destroy(&run_lock);
  mysql_mutex_destroy(&data_lock);
  mysql_mutex_destroy(&log_space_lock);
  mysql_cond_destroy(&data_cond);
  mysql_cond_destroy(&start_cond);
  mysql_cond_destroy(&stop_cond);
  mysql_cond_destroy(&log_space_cond);
97 98 99 100 101
  relay_log.cleanup();
  DBUG_VOID_RETURN;
}


102
int init_relay_log_info(Relay_log_info* rli,
103 104 105 106 107 108 109 110 111 112 113 114
			const char* info_fname)
{
  char fname[FN_REFLEN+128];
  int info_fd;
  const char* msg = 0;
  int error = 0;
  DBUG_ENTER("init_relay_log_info");
  DBUG_ASSERT(!rli->no_storage);         // Don't init if there is no storage

  if (rli->inited)                       // Set if this function called
    DBUG_RETURN(0);
  fn_format(fname, info_fname, mysql_data_home, "", 4+32);
Marc Alff's avatar
Marc Alff committed
115
  mysql_mutex_lock(&rli->data_lock);
116 117 118 119 120 121 122 123 124
  info_fd = rli->info_fd;
  rli->cur_log_fd = -1;
  rli->slave_skip_counter=0;
  rli->abort_pos_wait=0;
  rli->log_space_limit= relay_log_space_limit;
  rli->log_space_total= 0;
  rli->tables_to_lock= 0;
  rli->tables_to_lock_count= 0;

125
  char pattern[FN_REFLEN];
126 127
  (void) my_realpath(pattern, slave_load_tmpdir, 0);
  if (fn_format(pattern, PREFIX_SQL_LOAD, pattern, "",
128 129
            MY_SAFE_PATH | MY_RETURN_REAL_PATH) == NullS)
  {
Marc Alff's avatar
Marc Alff committed
130
    mysql_mutex_unlock(&rli->data_lock);
131 132 133 134 135
    sql_print_error("Unable to use slave's temporary directory %s",
                    slave_load_tmpdir);
    DBUG_RETURN(1);
  }
  unpack_filename(rli->slave_patternload_file, pattern);
136 137
  rli->slave_patternload_file_size= strlen(rli->slave_patternload_file);

138 139 140
  /*
    The relay log will now be opened, as a SEQ_READ_APPEND IO_CACHE.
    Note that the I/O thread flushes it to disk after writing every
141
    event, in flush_master_info(mi, 1, ?).
142 143 144 145 146 147 148 149 150 151 152 153
  */

  /*
    For the maximum log size, we choose max_relay_log_size if it is
    non-zero, max_binlog_size otherwise. If later the user does SET
    GLOBAL on one of these variables, fix_max_binlog_size and
    fix_max_relay_log_size will reconsider the choice (for example
    if the user changes max_relay_log_size to zero, we have to
    switch to using max_binlog_size for the relay log) and update
    rli->relay_log.max_size (and mysql_bin_log.max_size).
  */
  {
154 155 156 157 158
    /* Reports an error and returns, if the --relay-log's path 
       is a directory.*/
    if (opt_relay_logname && 
        opt_relay_logname[strlen(opt_relay_logname) - 1] == FN_LIBCHAR)
    {
Marc Alff's avatar
Marc Alff committed
159
      mysql_mutex_unlock(&rli->data_lock);
160 161 162 163 164 165 166 167 168 169 170
      sql_print_error("Path '%s' is a directory name, please specify \
a file name for --relay-log option", opt_relay_logname);
      DBUG_RETURN(1);
    }

    /* Reports an error and returns, if the --relay-log-index's path 
       is a directory.*/
    if (opt_relaylog_index_name && 
        opt_relaylog_index_name[strlen(opt_relaylog_index_name) - 1] 
        == FN_LIBCHAR)
    {
Marc Alff's avatar
Marc Alff committed
171
      mysql_mutex_unlock(&rli->data_lock);
172 173 174 175 176
      sql_print_error("Path '%s' is a directory name, please specify \
a file name for --relay-log-index option", opt_relaylog_index_name);
      DBUG_RETURN(1);
    }

177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202
    char buf[FN_REFLEN];
    const char *ln;
    static bool name_warning_sent= 0;
    ln= rli->relay_log.generate_name(opt_relay_logname, "-relay-bin",
                                     1, buf);
    /* We send the warning only at startup, not after every RESET SLAVE */
    if (!opt_relay_logname && !opt_relaylog_index_name && !name_warning_sent)
    {
      /*
        User didn't give us info to name the relay log index file.
        Picking `hostname`-relay-bin.index like we do, causes replication to
        fail if this slave's hostname is changed later. So, we would like to
        instead require a name. But as we don't want to break many existing
        setups, we only give warning, not error.
      */
      sql_print_warning("Neither --relay-log nor --relay-log-index were used;"
                        " so replication "
                        "may break when this MySQL server acts as a "
                        "slave and has his hostname changed!! Please "
                        "use '--relay-log=%s' to avoid this problem.", ln);
      name_warning_sent= 1;
    }
    /*
      note, that if open() fails, we'll still have index file open
      but a destructor will take care of that
    */
203
    if (rli->relay_log.open_index_file(opt_relaylog_index_name, ln, TRUE) ||
204 205
        rli->relay_log.open(ln, LOG_BIN, 0, SEQ_READ_APPEND, 0,
                            (max_relay_log_size ? max_relay_log_size :
206
                            max_binlog_size), 1, TRUE))
207
    {
Marc Alff's avatar
Marc Alff committed
208
      mysql_mutex_unlock(&rli->data_lock);
209 210 211
      sql_print_error("Failed in open_log() called from init_relay_log_info()");
      DBUG_RETURN(1);
    }
212
    rli->relay_log.is_relay_log= TRUE;
213 214 215 216 217 218 219 220 221 222
  }

  /* if file does not exist */
  if (access(fname,F_OK))
  {
    /*
      If someone removed the file from underneath our feet, just close
      the old descriptor and re-create the old file
    */
    if (info_fd >= 0)
Marc Alff's avatar
Marc Alff committed
223 224 225
      mysql_file_close(info_fd, MYF(MY_WME));
    if ((info_fd= mysql_file_open(key_file_relay_log_info,
                                  fname, O_CREAT|O_RDWR|O_BINARY, MYF(MY_WME))) < 0)
226 227 228
    {
      sql_print_error("Failed to create a new relay log info file (\
file '%s', errno %d)", fname, my_errno);
Marc Alff's avatar
Marc Alff committed
229
      msg= current_thd->stmt_da->message();
230 231 232 233 234 235 236
      goto err;
    }
    if (init_io_cache(&rli->info_file, info_fd, IO_SIZE*2, READ_CACHE, 0L,0,
                      MYF(MY_WME)))
    {
      sql_print_error("Failed to create a cache on relay log info file '%s'",
                      fname);
Marc Alff's avatar
Marc Alff committed
237
      msg= current_thd->stmt_da->message();
238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258
      goto err;
    }

    /* Init relay log with first entry in the relay index file */
    if (init_relay_log_pos(rli,NullS,BIN_LOG_HEADER_SIZE,0 /* no data lock */,
                           &msg, 0))
    {
      sql_print_error("Failed to open the relay log 'FIRST' (relay_log_pos 4)");
      goto err;
    }
    rli->group_master_log_name[0]= 0;
    rli->group_master_log_pos= 0;
    rli->info_fd= info_fd;
  }
  else // file exists
  {
    if (info_fd >= 0)
      reinit_io_cache(&rli->info_file, READ_CACHE, 0L,0,0);
    else
    {
      int error=0;
Marc Alff's avatar
Marc Alff committed
259 260
      if ((info_fd= mysql_file_open(key_file_relay_log_info,
                                    fname, O_RDWR|O_BINARY, MYF(MY_WME))) < 0)
261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276
      {
        sql_print_error("\
Failed to open the existing relay log info file '%s' (errno %d)",
                        fname, my_errno);
        error= 1;
      }
      else if (init_io_cache(&rli->info_file, info_fd,
                             IO_SIZE*2, READ_CACHE, 0L, 0, MYF(MY_WME)))
      {
        sql_print_error("Failed to create a cache on relay log info file '%s'",
                        fname);
        error= 1;
      }
      if (error)
      {
        if (info_fd >= 0)
Marc Alff's avatar
Marc Alff committed
277
          mysql_file_close(info_fd, MYF(0));
278 279
        rli->info_fd= -1;
        rli->relay_log.close(LOG_CLOSE_INDEX | LOG_CLOSE_STOP_EVENT);
Marc Alff's avatar
Marc Alff committed
280
        mysql_mutex_unlock(&rli->data_lock);
281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304
        DBUG_RETURN(1);
      }
    }

    rli->info_fd = info_fd;
    int relay_log_pos, master_log_pos;
    if (init_strvar_from_file(rli->group_relay_log_name,
                              sizeof(rli->group_relay_log_name),
                              &rli->info_file, "") ||
       init_intvar_from_file(&relay_log_pos,
                             &rli->info_file, BIN_LOG_HEADER_SIZE) ||
       init_strvar_from_file(rli->group_master_log_name,
                             sizeof(rli->group_master_log_name),
                             &rli->info_file, "") ||
       init_intvar_from_file(&master_log_pos, &rli->info_file, 0))
    {
      msg="Error reading slave log configuration";
      goto err;
    }
    strmake(rli->event_relay_log_name,rli->group_relay_log_name,
            sizeof(rli->event_relay_log_name)-1);
    rli->group_relay_log_pos= rli->event_relay_log_pos= relay_log_pos;
    rli->group_master_log_pos= master_log_pos;

305 306 307
    if (rli->is_relay_log_recovery && init_recovery(rli->mi, &msg))
      goto err;

308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338
    if (init_relay_log_pos(rli,
                           rli->group_relay_log_name,
                           rli->group_relay_log_pos,
                           0 /* no data lock*/,
                           &msg, 0))
    {
      char llbuf[22];
      sql_print_error("Failed to open the relay log '%s' (relay_log_pos %s)",
                      rli->group_relay_log_name,
                      llstr(rli->group_relay_log_pos, llbuf));
      goto err;
    }
  }

#ifndef DBUG_OFF
  {
    char llbuf1[22], llbuf2[22];
    DBUG_PRINT("info", ("my_b_tell(rli->cur_log)=%s rli->event_relay_log_pos=%s",
                        llstr(my_b_tell(rli->cur_log),llbuf1),
                        llstr(rli->event_relay_log_pos,llbuf2)));
    DBUG_ASSERT(rli->event_relay_log_pos >= BIN_LOG_HEADER_SIZE);
    DBUG_ASSERT(my_b_tell(rli->cur_log) == rli->event_relay_log_pos);
  }
#endif

  /*
    Now change the cache from READ to WRITE - must do this
    before flush_relay_log_info
  */
  reinit_io_cache(&rli->info_file, WRITE_CACHE,0L,0,1);
  if ((error= flush_relay_log_info(rli)))
339 340 341 342
  {
    msg= "Failed to flush relay log info file";
    goto err;
  }
343 344 345 346 347 348
  if (count_relay_log_space(rli))
  {
    msg="Error counting relay log space";
    goto err;
  }
  rli->inited= 1;
Marc Alff's avatar
Marc Alff committed
349
  mysql_mutex_unlock(&rli->data_lock);
350 351 352
  DBUG_RETURN(error);

err:
353
  sql_print_error("%s", msg);
354 355
  end_io_cache(&rli->info_file);
  if (info_fd >= 0)
Marc Alff's avatar
Marc Alff committed
356
    mysql_file_close(info_fd, MYF(0));
357 358
  rli->info_fd= -1;
  rli->relay_log.close(LOG_CLOSE_INDEX | LOG_CLOSE_STOP_EVENT);
Marc Alff's avatar
Marc Alff committed
359
  mysql_mutex_unlock(&rli->data_lock);
360 361 362 363
  DBUG_RETURN(1);
}


364
static inline int add_relay_log(Relay_log_info* rli,LOG_INFO* linfo)
365 366 367
{
  MY_STAT s;
  DBUG_ENTER("add_relay_log");
368
  if (!mysql_file_stat(key_file_relaylog,
Marc Alff's avatar
Marc Alff committed
369
                       linfo->log_file_name, &s, MYF(0)))
370 371 372 373 374 375 376 377 378 379 380 381 382 383
  {
    sql_print_error("log %s listed in the index, but failed to stat",
                    linfo->log_file_name);
    DBUG_RETURN(1);
  }
  rli->log_space_total += s.st_size;
#ifndef DBUG_OFF
  char buf[22];
  DBUG_PRINT("info",("log_space_total: %s", llstr(rli->log_space_total,buf)));
#endif
  DBUG_RETURN(0);
}


384
static int count_relay_log_space(Relay_log_info* rli)
385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409
{
  LOG_INFO linfo;
  DBUG_ENTER("count_relay_log_space");
  rli->log_space_total= 0;
  if (rli->relay_log.find_log_pos(&linfo, NullS, 1))
  {
    sql_print_error("Could not find first log while counting relay log space");
    DBUG_RETURN(1);
  }
  do
  {
    if (add_relay_log(rli,&linfo))
      DBUG_RETURN(1);
  } while (!rli->relay_log.find_next_log(&linfo, 1));
  /*
     As we have counted everything, including what may have written in a
     preceding write, we must reset bytes_written, or we may count some space
     twice.
  */
  rli->relay_log.reset_bytes_written();
  DBUG_RETURN(0);
}


/*
410
   Reset UNTIL condition for Relay_log_info
411 412 413

   SYNOPSYS
    clear_until_condition()
414
      rli - Relay_log_info structure where UNTIL condition should be reset
415 416
 */

417
void Relay_log_info::clear_until_condition()
418 419 420
{
  DBUG_ENTER("clear_until_condition");

421
  until_condition= Relay_log_info::UNTIL_NONE;
422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459
  until_log_name[0]= 0;
  until_log_pos= 0;
  DBUG_VOID_RETURN;
}


/*
  Open the given relay log

  SYNOPSIS
    init_relay_log_pos()
    rli                 Relay information (will be initialized)
    log                 Name of relay log file to read from. NULL = First log
    pos                 Position in relay log file
    need_data_lock      Set to 1 if this functions should do mutex locks
    errmsg              Store pointer to error message here
    look_for_description_event
                        1 if we should look for such an event. We only need
                        this when the SQL thread starts and opens an existing
                        relay log and has to execute it (possibly from an
                        offset >4); then we need to read the first event of
                        the relay log to be able to parse the events we have
                        to execute.

  DESCRIPTION
  - Close old open relay log files.
  - If we are using the same relay log as the running IO-thread, then set
    rli->cur_log to point to the same IO_CACHE entry.
  - If not, open the 'log' binary file.

  TODO
    - check proper initialization of group_master_log_name/group_master_log_pos

  RETURN VALUES
    0   ok
    1   error.  errmsg is set to point to the error message
*/

460
int init_relay_log_pos(Relay_log_info* rli,const char* log,
461 462 463 464 465
                       ulonglong pos, bool need_data_lock,
                       const char** errmsg,
                       bool look_for_description_event)
{
  DBUG_ENTER("init_relay_log_pos");
unknown's avatar
unknown committed
466
  DBUG_PRINT("info", ("pos: %lu", (ulong) pos));
467 468

  *errmsg=0;
Marc Alff's avatar
Marc Alff committed
469
  mysql_mutex_t *log_lock= rli->relay_log.get_log_lock();
470 471

  if (need_data_lock)
Marc Alff's avatar
Marc Alff committed
472
    mysql_mutex_lock(&rli->data_lock);
473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491

  /*
    Slave threads are not the only users of init_relay_log_pos(). CHANGE MASTER
    is, too, and init_slave() too; these 2 functions allocate a description
    event in init_relay_log_pos, which is not freed by the terminating SQL slave
    thread as that thread is not started by these functions. So we have to free
    the description_event here, in case, so that there is no memory leak in
    running, say, CHANGE MASTER.
  */
  delete rli->relay_log.description_event_for_exec;
  /*
    By default the relay log is in binlog format 3 (4.0).
    Even if format is 4, this will work enough to read the first event
    (Format_desc) (remember that format 4 is just lenghtened compared to format
    3; format 3 is a prefix of format 4).
  */
  rli->relay_log.description_event_for_exec= new
    Format_description_log_event(3);

Marc Alff's avatar
Marc Alff committed
492
  mysql_mutex_lock(log_lock);
493 494 495 496 497

  /* Close log file and free buffers if it's already open */
  if (rli->cur_log_fd >= 0)
  {
    end_io_cache(&rli->cache_buf);
Marc Alff's avatar
Marc Alff committed
498
    mysql_file_close(rli->cur_log_fd, MYF(MY_WME));
499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631
    rli->cur_log_fd = -1;
  }

  rli->group_relay_log_pos = rli->event_relay_log_pos = pos;

  /*
    Test to see if the previous run was with the skip of purging
    If yes, we do not purge when we restart
  */
  if (rli->relay_log.find_log_pos(&rli->linfo, NullS, 1))
  {
    *errmsg="Could not find first log during relay log initialization";
    goto err;
  }

  if (log && rli->relay_log.find_log_pos(&rli->linfo, log, 1))
  {
    *errmsg="Could not find target log during relay log initialization";
    goto err;
  }
  strmake(rli->group_relay_log_name,rli->linfo.log_file_name,
          sizeof(rli->group_relay_log_name)-1);
  strmake(rli->event_relay_log_name,rli->linfo.log_file_name,
          sizeof(rli->event_relay_log_name)-1);
  if (rli->relay_log.is_active(rli->linfo.log_file_name))
  {
    /*
      The IO thread is using this log file.
      In this case, we will use the same IO_CACHE pointer to
      read data as the IO thread is using to write data.
    */
    my_b_seek((rli->cur_log=rli->relay_log.get_log_file()), (off_t)0);
    if (check_binlog_magic(rli->cur_log,errmsg))
      goto err;
    rli->cur_log_old_open_count=rli->relay_log.get_open_count();
  }
  else
  {
    /*
      Open the relay log and set rli->cur_log to point at this one
    */
    if ((rli->cur_log_fd=open_binlog(&rli->cache_buf,
                                     rli->linfo.log_file_name,errmsg)) < 0)
      goto err;
    rli->cur_log = &rli->cache_buf;
  }
  /*
    In all cases, check_binlog_magic() has been called so we're at offset 4 for
    sure.
  */
  if (pos > BIN_LOG_HEADER_SIZE) /* If pos<=4, we stay at 4 */
  {
    Log_event* ev;
    while (look_for_description_event)
    {
      /*
        Read the possible Format_description_log_event; if position
        was 4, no need, it will be read naturally.
      */
      DBUG_PRINT("info",("looking for a Format_description_log_event"));

      if (my_b_tell(rli->cur_log) >= pos)
        break;

      /*
        Because of we have rli->data_lock and log_lock, we can safely read an
        event
      */
      if (!(ev=Log_event::read_log_event(rli->cur_log,0,
                                         rli->relay_log.description_event_for_exec)))
      {
        DBUG_PRINT("info",("could not read event, rli->cur_log->error=%d",
                           rli->cur_log->error));
        if (rli->cur_log->error) /* not EOF */
        {
          *errmsg= "I/O error reading event at position 4";
          goto err;
        }
        break;
      }
      else if (ev->get_type_code() == FORMAT_DESCRIPTION_EVENT)
      {
        DBUG_PRINT("info",("found Format_description_log_event"));
        delete rli->relay_log.description_event_for_exec;
        rli->relay_log.description_event_for_exec= (Format_description_log_event*) ev;
        /*
          As ev was returned by read_log_event, it has passed is_valid(), so
          my_malloc() in ctor worked, no need to check again.
        */
        /*
          Ok, we found a Format_description event. But it is not sure that this
          describes the whole relay log; indeed, one can have this sequence
          (starting from position 4):
          Format_desc (of slave)
          Rotate (of master)
          Format_desc (of master)
          So the Format_desc which really describes the rest of the relay log
          is the 3rd event (it can't be further than that, because we rotate
          the relay log when we queue a Rotate event from the master).
          But what describes the Rotate is the first Format_desc.
          So what we do is:
          go on searching for Format_description events, until you exceed the
          position (argument 'pos') or until you find another event than Rotate
          or Format_desc.
        */
      }
      else
      {
        DBUG_PRINT("info",("found event of another type=%d",
                           ev->get_type_code()));
        look_for_description_event= (ev->get_type_code() == ROTATE_EVENT);
        delete ev;
      }
    }
    my_b_seek(rli->cur_log,(off_t)pos);
#ifndef DBUG_OFF
  {
    char llbuf1[22], llbuf2[22];
    DBUG_PRINT("info", ("my_b_tell(rli->cur_log)=%s rli->event_relay_log_pos=%s",
                        llstr(my_b_tell(rli->cur_log),llbuf1),
                        llstr(rli->event_relay_log_pos,llbuf2)));
  }
#endif

  }

err:
  /*
    If we don't purge, we can't honour relay_log_space_limit ;
    silently discard it
  */
  if (!relay_log_purge)
    rli->log_space_limit= 0;
Marc Alff's avatar
Marc Alff committed
632
  mysql_cond_broadcast(&rli->data_cond);
633

Marc Alff's avatar
Marc Alff committed
634
  mysql_mutex_unlock(log_lock);
635 636

  if (need_data_lock)
Marc Alff's avatar
Marc Alff committed
637
    mysql_mutex_unlock(&rli->data_lock);
638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669
  if (!rli->relay_log.description_event_for_exec->is_valid() && !*errmsg)
    *errmsg= "Invalid Format_description log event; could be out of memory";

  DBUG_RETURN ((*errmsg) ? 1 : 0);
}


/*
  Waits until the SQL thread reaches (has executed up to) the
  log/position or timed out.

  SYNOPSIS
    wait_for_pos()
    thd             client thread that sent SELECT MASTER_POS_WAIT
    log_name        log name to wait for
    log_pos         position to wait for
    timeout         timeout in seconds before giving up waiting

  NOTES
    timeout is longlong whereas it should be ulong ; but this is
    to catch if the user submitted a negative timeout.

  RETURN VALUES
    -2          improper arguments (log_pos<0)
                or slave not running, or master info changed
                during the function's execution,
                or client thread killed. -2 is translated to NULL by caller
    -1          timed out
    >=0         number of log events the function had to wait
                before reaching the desired log/position
 */

670
int Relay_log_info::wait_for_pos(THD* thd, String* log_name,
671 672 673 674 675 676 677 678
                                    longlong log_pos,
                                    longlong timeout)
{
  int event_count = 0;
  ulong init_abort_pos_wait;
  int error=0;
  struct timespec abstime; // for timeout checking
  const char *msg;
679
  DBUG_ENTER("Relay_log_info::wait_for_pos");
680 681

  if (!inited)
unknown's avatar
unknown committed
682
    DBUG_RETURN(-2);
683 684 685 686 687

  DBUG_PRINT("enter",("log_name: '%s'  log_pos: %lu  timeout: %lu",
                      log_name->c_ptr(), (ulong) log_pos, (ulong) timeout));

  set_timespec(abstime,timeout);
Marc Alff's avatar
Marc Alff committed
688
  mysql_mutex_lock(&data_lock);
689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800
  msg= thd->enter_cond(&data_cond, &data_lock,
                       "Waiting for the slave SQL thread to "
                       "advance position");
  /*
     This function will abort when it notices that some CHANGE MASTER or
     RESET MASTER has changed the master info.
     To catch this, these commands modify abort_pos_wait ; We just monitor
     abort_pos_wait and see if it has changed.
     Why do we have this mechanism instead of simply monitoring slave_running
     in the loop (we do this too), as CHANGE MASTER/RESET SLAVE require that
     the SQL thread be stopped?
     This is becasue if someones does:
     STOP SLAVE;CHANGE MASTER/RESET SLAVE; START SLAVE;
     the change may happen very quickly and we may not notice that
     slave_running briefly switches between 1/0/1.
  */
  init_abort_pos_wait= abort_pos_wait;

  /*
    We'll need to
    handle all possible log names comparisons (e.g. 999 vs 1000).
    We use ulong for string->number conversion ; this is no
    stronger limitation than in find_uniq_filename in sql/log.cc
  */
  ulong log_name_extension;
  char log_name_tmp[FN_REFLEN]; //make a char[] from String

  strmake(log_name_tmp, log_name->ptr(), min(log_name->length(), FN_REFLEN-1));

  char *p= fn_ext(log_name_tmp);
  char *p_end;
  if (!*p || log_pos<0)
  {
    error= -2; //means improper arguments
    goto err;
  }
  // Convert 0-3 to 4
  log_pos= max(log_pos, BIN_LOG_HEADER_SIZE);
  /* p points to '.' */
  log_name_extension= strtoul(++p, &p_end, 10);
  /*
    p_end points to the first invalid character.
    If it equals to p, no digits were found, error.
    If it contains '\0' it means conversion went ok.
  */
  if (p_end==p || *p_end)
  {
    error= -2;
    goto err;
  }

  /* The "compare and wait" main loop */
  while (!thd->killed &&
         init_abort_pos_wait == abort_pos_wait &&
         slave_running)
  {
    bool pos_reached;
    int cmp_result= 0;

    DBUG_PRINT("info",
               ("init_abort_pos_wait: %ld  abort_pos_wait: %ld",
                init_abort_pos_wait, abort_pos_wait));
    DBUG_PRINT("info",("group_master_log_name: '%s'  pos: %lu",
                       group_master_log_name, (ulong) group_master_log_pos));

    /*
      group_master_log_name can be "", if we are just after a fresh
      replication start or after a CHANGE MASTER TO MASTER_HOST/PORT
      (before we have executed one Rotate event from the master) or
      (rare) if the user is doing a weird slave setup (see next
      paragraph).  If group_master_log_name is "", we assume we don't
      have enough info to do the comparison yet, so we just wait until
      more data. In this case master_log_pos is always 0 except if
      somebody (wrongly) sets this slave to be a slave of itself
      without using --replicate-same-server-id (an unsupported
      configuration which does nothing), then group_master_log_pos
      will grow and group_master_log_name will stay "".
    */
    if (*group_master_log_name)
    {
      char *basename= (group_master_log_name +
                       dirname_length(group_master_log_name));
      /*
        First compare the parts before the extension.
        Find the dot in the master's log basename,
        and protect against user's input error :
        if the names do not match up to '.' included, return error
      */
      char *q= (char*)(fn_ext(basename)+1);
      if (strncmp(basename, log_name_tmp, (int)(q-basename)))
      {
        error= -2;
        break;
      }
      // Now compare extensions.
      char *q_end;
      ulong group_master_log_name_extension= strtoul(q, &q_end, 10);
      if (group_master_log_name_extension < log_name_extension)
        cmp_result= -1 ;
      else
        cmp_result= (group_master_log_name_extension > log_name_extension) ? 1 : 0 ;

      pos_reached= ((!cmp_result && group_master_log_pos >= (ulonglong)log_pos) ||
                    cmp_result > 0);
      if (pos_reached || thd->killed)
        break;
    }

    //wait for master update, with optional timeout.

    DBUG_PRINT("info",("Waiting for master update"));
    /*
Marc Alff's avatar
Marc Alff committed
801
      We are going to mysql_cond_(timed)wait(); if the SQL thread stops it
802 803
      will wake us up.
    */
804
    thd_wait_begin(thd, THD_WAIT_BINLOG);
805 806 807
    if (timeout > 0)
    {
      /*
Marc Alff's avatar
Marc Alff committed
808
        Note that mysql_cond_timedwait checks for the timeout
809 810 811 812 813
        before for the condition ; i.e. it returns ETIMEDOUT
        if the system time equals or exceeds the time specified by abstime
        before the condition variable is signaled or broadcast, _or_ if
        the absolute time specified by abstime has already passed at the time
        of the call.
Marc Alff's avatar
Marc Alff committed
814
        For that reason, mysql_cond_timedwait will do the "timeoutting" job
815 816 817
        even if its condition is always immediately signaled (case of a loaded
        master).
      */
Marc Alff's avatar
Marc Alff committed
818
      error= mysql_cond_timedwait(&data_cond, &data_lock, &abstime);
819 820
    }
    else
Marc Alff's avatar
Marc Alff committed
821
      mysql_cond_wait(&data_cond, &data_lock);
822
    thd_wait_end(thd);
823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851
    DBUG_PRINT("info",("Got signal of master update or timed out"));
    if (error == ETIMEDOUT || error == ETIME)
    {
      error= -1;
      break;
    }
    error=0;
    event_count++;
    DBUG_PRINT("info",("Testing if killed or SQL thread not running"));
  }

err:
  thd->exit_cond(msg);
  DBUG_PRINT("exit",("killed: %d  abort: %d  slave_running: %d \
improper_arguments: %d  timed_out: %d",
                     thd->killed_errno(),
                     (int) (init_abort_pos_wait != abort_pos_wait),
                     (int) slave_running,
                     (int) (error == -2),
                     (int) (error == -1)));
  if (thd->killed || init_abort_pos_wait != abort_pos_wait ||
      !slave_running)
  {
    error= -2;
  }
  DBUG_RETURN( error ? error : event_count );
}


852
void Relay_log_info::inc_group_relay_log_pos(ulonglong log_pos,
853 854
                                                bool skip_lock)
{
855
  DBUG_ENTER("Relay_log_info::inc_group_relay_log_pos");
856 857

  if (!skip_lock)
Marc Alff's avatar
Marc Alff committed
858
    mysql_mutex_lock(&data_lock);
859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901
  inc_event_relay_log_pos();
  group_relay_log_pos= event_relay_log_pos;
  strmake(group_relay_log_name,event_relay_log_name,
          sizeof(group_relay_log_name)-1);

  notify_group_relay_log_name_update();

  /*
    If the slave does not support transactions and replicates a transaction,
    users should not trust group_master_log_pos (which they can display with
    SHOW SLAVE STATUS or read from relay-log.info), because to compute
    group_master_log_pos the slave relies on log_pos stored in the master's
    binlog, but if we are in a master's transaction these positions are always
    the BEGIN's one (excepted for the COMMIT), so group_master_log_pos does
    not advance as it should on the non-transactional slave (it advances by
    big leaps, whereas it should advance by small leaps).
  */
  /*
    In 4.x we used the event's len to compute the positions here. This is
    wrong if the event was 3.23/4.0 and has been converted to 5.0, because
    then the event's len is not what is was in the master's binlog, so this
    will make a wrong group_master_log_pos (yes it's a bug in 3.23->4.0
    replication: Exec_master_log_pos is wrong). Only way to solve this is to
    have the original offset of the end of the event the relay log. This is
    what we do in 5.0: log_pos has become "end_log_pos" (because the real use
    of log_pos in 4.0 was to compute the end_log_pos; so better to store
    end_log_pos instead of begin_log_pos.
    If we had not done this fix here, the problem would also have appeared
    when the slave and master are 5.0 but with different event length (for
    example the slave is more recent than the master and features the event
    UID). It would give false MASTER_POS_WAIT, false Exec_master_log_pos in
    SHOW SLAVE STATUS, and so the user would do some CHANGE MASTER using this
    value which would lead to badly broken replication.
    Even the relay_log_pos will be corrupted in this case, because the len is
    the relay log is not "val".
    With the end_log_pos solution, we avoid computations involving lengthes.
  */
  DBUG_PRINT("info", ("log_pos: %lu  group_master_log_pos: %lu",
                      (long) log_pos, (long) group_master_log_pos));
  if (log_pos) // 3.23 binlogs don't have log_posx
  {
    group_master_log_pos= log_pos;
  }
Marc Alff's avatar
Marc Alff committed
902
  mysql_cond_broadcast(&data_cond);
903
  if (!skip_lock)
Marc Alff's avatar
Marc Alff committed
904
    mysql_mutex_unlock(&data_lock);
905 906 907 908
  DBUG_VOID_RETURN;
}


909
void Relay_log_info::close_temporary_tables()
910 911
{
  TABLE *table,*next;
912
  DBUG_ENTER("Relay_log_info::close_temporary_tables");
913 914 915 916 917 918 919 920

  for (table=save_temporary_tables ; table ; table=next)
  {
    next=table->next;
    /*
      Don't ask for disk deletion. For now, anyway they will be deleted when
      slave restarts, but it is a better intention to not delete them.
    */
unknown's avatar
unknown committed
921
    DBUG_PRINT("info", ("table: 0x%lx", (long) table));
922 923 924 925 926 927 928 929 930 931 932 933 934 935
    close_temporary(table, 1, 0);
  }
  save_temporary_tables= 0;
  slave_open_temp_tables= 0;
  DBUG_VOID_RETURN;
}

/*
  purge_relay_logs()

  NOTES
    Assumes to have a run lock on rli and that no slave thread are running.
*/

936
int purge_relay_logs(Relay_log_info* rli, THD *thd, bool just_reset,
937 938 939 940 941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972
                     const char** errmsg)
{
  int error=0;
  DBUG_ENTER("purge_relay_logs");

  /*
    Even if rli->inited==0, we still try to empty rli->master_log_* variables.
    Indeed, rli->inited==0 does not imply that they already are empty.
    It could be that slave's info initialization partly succeeded :
    for example if relay-log.info existed but *relay-bin*.*
    have been manually removed, init_relay_log_info reads the old
    relay-log.info and fills rli->master_log_*, then init_relay_log_info
    checks for the existence of the relay log, this fails and
    init_relay_log_info leaves rli->inited to 0.
    In that pathological case, rli->master_log_pos* will be properly reinited
    at the next START SLAVE (as RESET SLAVE or CHANGE
    MASTER, the callers of purge_relay_logs, will delete bogus *.info files
    or replace them with correct files), however if the user does SHOW SLAVE
    STATUS before START SLAVE, he will see old, confusing rli->master_log_*.
    In other words, we reinit rli->master_log_* for SHOW SLAVE STATUS
    to display fine in any case.
  */

  rli->group_master_log_name[0]= 0;
  rli->group_master_log_pos= 0;

  if (!rli->inited)
  {
    DBUG_PRINT("info", ("rli->inited == 0"));
    DBUG_RETURN(0);
  }

  DBUG_ASSERT(rli->slave_running == 0);
  DBUG_ASSERT(rli->mi->slave_running == 0);

  rli->slave_skip_counter=0;
Marc Alff's avatar
Marc Alff committed
973
  mysql_mutex_lock(&rli->data_lock);
974 975 976 977 978 979 980 981 982 983

  /*
    we close the relay log fd possibly left open by the slave SQL thread,
    to be able to delete it; the relay log fd possibly left open by the slave
    I/O thread will be closed naturally in reset_logs() by the
    close(LOG_CLOSE_TO_BE_OPENED) call
  */
  if (rli->cur_log_fd >= 0)
  {
    end_io_cache(&rli->cache_buf);
Marc Alff's avatar
Marc Alff committed
984
    mysql_file_close(rli->cur_log_fd, MYF(MY_WME));
985 986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000 1001 1002
    rli->cur_log_fd= -1;
  }

  if (rli->relay_log.reset_logs(thd))
  {
    *errmsg = "Failed during log reset";
    error=1;
    goto err;
  }
  /* Save name of used relay log file */
  strmake(rli->group_relay_log_name, rli->relay_log.get_log_fname(),
          sizeof(rli->group_relay_log_name)-1);
  strmake(rli->event_relay_log_name, rli->relay_log.get_log_fname(),
          sizeof(rli->event_relay_log_name)-1);
  rli->group_relay_log_pos= rli->event_relay_log_pos= BIN_LOG_HEADER_SIZE;
  if (count_relay_log_space(rli))
  {
    *errmsg= "Error counting relay log space";
1003
    error=1;
1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015
    goto err;
  }
  if (!just_reset)
    error= init_relay_log_pos(rli, rli->group_relay_log_name,
                              rli->group_relay_log_pos,
                              0 /* do not need data lock */, errmsg, 0);

err:
#ifndef DBUG_OFF
  char buf[22];
#endif
  DBUG_PRINT("info",("log_space_total: %s",llstr(rli->log_space_total,buf)));
Marc Alff's avatar
Marc Alff committed
1016
  mysql_mutex_unlock(&rli->data_lock);
1017 1018 1019 1020 1021 1022 1023
  DBUG_RETURN(error);
}


/*
     Check if condition stated in UNTIL clause of START SLAVE is reached.
   SYNOPSYS
1024
     Relay_log_info::is_until_satisfied()
1025 1026 1027 1028 1029
     master_beg_pos    position of the beginning of to be executed event
                       (not log_pos member of the event that points to the
                        beginning of the following event)


1030 1031 1032 1033
   DESCRIPTION
     Checks if UNTIL condition is reached. Uses caching result of last
     comparison of current log file name and target log file name. So cached
     value should be invalidated if current log file name changes
1034
     (see Relay_log_info::notify_... functions).
1035 1036 1037 1038 1039

     This caching is needed to avoid of expensive string comparisons and
     strtol() conversions needed for log names comparison. We don't need to
     compare them each time this function is called, we only need to do this
     when current log name changes. If we have UNTIL_MASTER_POS condition we
1040
     need to do this only after Rotate_log_event::do_apply_event() (which is
1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053
     rare, so caching gives real benifit), and if we have UNTIL_RELAY_POS
     condition then we should invalidate cached comarison value after
     inc_group_relay_log_pos() which called for each group of events (so we
     have some benefit if we have something like queries that use
     autoincrement or if we have transactions).

     Should be called ONLY if until_condition != UNTIL_NONE !
   RETURN VALUE
     true - condition met or error happened (condition seems to have
            bad log file name)
     false - condition not met
*/

1054
bool Relay_log_info::is_until_satisfied(THD *thd, Log_event *ev)
1055 1056 1057
{
  const char *log_name;
  ulonglong log_pos;
1058
  DBUG_ENTER("Relay_log_info::is_until_satisfied");
1059 1060 1061 1062 1063

  DBUG_ASSERT(until_condition != UNTIL_NONE);

  if (until_condition == UNTIL_MASTER_POS)
  {
1064
    if (ev && ev->server_id == (uint32) ::server_id && !replicate_same_server_id)
1065
      DBUG_RETURN(FALSE);
1066
    log_name= group_master_log_name;
1067
    log_pos= (!ev)? group_master_log_pos :
1068
      ((thd->variables.option_bits & OPTION_BEGIN || !ev->log_pos) ?
1069
       group_master_log_pos : ev->log_pos - ev->data_written);
1070 1071 1072 1073 1074 1075 1076
  }
  else
  { /* until_condition == UNTIL_RELAY_POS */
    log_name= group_relay_log_name;
    log_pos= group_relay_log_pos;
  }

1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092
#ifndef DBUG_OFF
  {
    char buf[32];
    DBUG_PRINT("info", ("group_master_log_name='%s', group_master_log_pos=%s",
                        group_master_log_name, llstr(group_master_log_pos, buf)));
    DBUG_PRINT("info", ("group_relay_log_name='%s', group_relay_log_pos=%s",
                        group_relay_log_name, llstr(group_relay_log_pos, buf)));
    DBUG_PRINT("info", ("(%s) log_name='%s', log_pos=%s",
                        until_condition == UNTIL_MASTER_POS ? "master" : "relay",
                        log_name, llstr(log_pos, buf)));
    DBUG_PRINT("info", ("(%s) until_log_name='%s', until_log_pos=%s",
                        until_condition == UNTIL_MASTER_POS ? "master" : "relay",
                        until_log_name, llstr(until_log_pos, buf)));
  }
#endif

1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 1119 1120 1121 1122 1123 1124 1125 1126 1127 1128 1129 1130 1131 1132 1133 1134 1135 1136 1137
  if (until_log_names_cmp_result == UNTIL_LOG_NAMES_CMP_UNKNOWN)
  {
    /*
      We have no cached comparison results so we should compare log names
      and cache result.
      If we are after RESET SLAVE, and the SQL slave thread has not processed
      any event yet, it could be that group_master_log_name is "". In that case,
      just wait for more events (as there is no sensible comparison to do).
    */

    if (*log_name)
    {
      const char *basename= log_name + dirname_length(log_name);

      const char *q= (const char*)(fn_ext(basename)+1);
      if (strncmp(basename, until_log_name, (int)(q-basename)) == 0)
      {
        /* Now compare extensions. */
        char *q_end;
        ulong log_name_extension= strtoul(q, &q_end, 10);
        if (log_name_extension < until_log_name_extension)
          until_log_names_cmp_result= UNTIL_LOG_NAMES_CMP_LESS;
        else
          until_log_names_cmp_result=
            (log_name_extension > until_log_name_extension) ?
            UNTIL_LOG_NAMES_CMP_GREATER : UNTIL_LOG_NAMES_CMP_EQUAL ;
      }
      else
      {
        /* Probably error so we aborting */
        sql_print_error("Slave SQL thread is stopped because UNTIL "
                        "condition is bad.");
        DBUG_RETURN(TRUE);
      }
    }
    else
      DBUG_RETURN(until_log_pos == 0);
  }

  DBUG_RETURN(((until_log_names_cmp_result == UNTIL_LOG_NAMES_CMP_EQUAL &&
           log_pos >= until_log_pos) ||
          until_log_names_cmp_result == UNTIL_LOG_NAMES_CMP_GREATER));
}


1138
void Relay_log_info::cached_charset_invalidate()
1139
{
1140
  DBUG_ENTER("Relay_log_info::cached_charset_invalidate");
1141 1142 1143 1144 1145 1146 1147

  /* Full of zeroes means uninitialized. */
  bzero(cached_charset, sizeof(cached_charset));
  DBUG_VOID_RETURN;
}


1148
bool Relay_log_info::cached_charset_compare(char *charset) const
1149
{
1150
  DBUG_ENTER("Relay_log_info::cached_charset_compare");
1151

1152
  if (memcmp(cached_charset, charset, sizeof(cached_charset)))
1153
  {
1154
    memcpy(const_cast<char*>(cached_charset), charset, sizeof(cached_charset));
1155 1156 1157 1158 1159 1160
    DBUG_RETURN(1);
  }
  DBUG_RETURN(0);
}


1161
void Relay_log_info::stmt_done(my_off_t event_master_log_pos,
1162
                                  time_t event_creation_time)
1163
{
1164 1165 1166
#ifndef DBUG_OFF
  extern uint debug_not_change_ts_if_art_event;
#endif
1167 1168 1169 1170 1171 1172 1173 1174 1175 1176 1177 1178 1179 1180 1181 1182 1183 1184 1185 1186 1187 1188 1189 1190 1191
  clear_flag(IN_STMT);

  /*
    If in a transaction, and if the slave supports transactions, just
    inc_event_relay_log_pos(). We only have to check for OPTION_BEGIN
    (not OPTION_NOT_AUTOCOMMIT) as transactions are logged with
    BEGIN/COMMIT, not with SET AUTOCOMMIT= .

    CAUTION: opt_using_transactions means innodb || bdb ; suppose the
    master supports InnoDB and BDB, but the slave supports only BDB,
    problems will arise: - suppose an InnoDB table is created on the
    master, - then it will be MyISAM on the slave - but as
    opt_using_transactions is true, the slave will believe he is
    transactional with the MyISAM table. And problems will come when
    one does START SLAVE; STOP SLAVE; START SLAVE; (the slave will
    resume at BEGIN whereas there has not been any rollback).  This is
    the problem of using opt_using_transactions instead of a finer
    "does the slave support _transactional handler used on the
    master_".

    More generally, we'll have problems when a query mixes a
    transactional handler and MyISAM and STOP SLAVE is issued in the
    middle of the "transaction". START SLAVE will resume at BEGIN
    while the MyISAM table has already been updated.
  */
1192
  if ((sql_thd->variables.option_bits & OPTION_BEGIN) && opt_using_transactions)
1193 1194 1195 1196 1197 1198 1199 1200 1201 1202 1203 1204 1205
    inc_event_relay_log_pos();
  else
  {
    inc_group_relay_log_pos(event_master_log_pos);
    flush_relay_log_info(this);
    /*
      Note that Rotate_log_event::do_apply_event() does not call this
      function, so there is no chance that a fake rotate event resets
      last_master_timestamp.  Note that we update without mutex
      (probably ok - except in some very rare cases, only consequence
      is that value may take some time to display in
      Seconds_Behind_Master - not critical).
    */
1206 1207 1208 1209 1210 1211
#ifndef DBUG_OFF
    if (!(event_creation_time == 0 && debug_not_change_ts_if_art_event > 0))
#else
      if (event_creation_time != 0)
#endif
        last_master_timestamp= event_creation_time;
1212 1213 1214
  }
}

1215
#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
1216
void Relay_log_info::cleanup_context(THD *thd, bool error)
1217
{
1218
  DBUG_ENTER("Relay_log_info::cleanup_context");
1219 1220 1221

  DBUG_ASSERT(sql_thd == thd);
  /*
1222
    1) Instances of Table_map_log_event, if ::do_apply_event() was called on them,
1223 1224 1225 1226
    may have opened tables, which we cannot be sure have been closed (because
    maybe the Rows_log_event have not been found or will not be, because slave
    SQL thread is stopping, or relay log has a missing tail etc). So we close
    all thread's tables. And so the table mappings have to be cancelled.
1227
    2) Rows_log_event::do_apply_event() may even have started statements or
1228 1229 1230 1231 1232 1233 1234
    transactions on them, which we need to rollback in case of error.
    3) If finding a Format_description_log_event after a BEGIN, we also need
    to rollback before continuing with the next events.
    4) so we need this "context cleanup" function.
  */
  if (error)
  {
Konstantin Osipov's avatar
Konstantin Osipov committed
1235 1236
    trans_rollback_stmt(thd); // if a "statement transaction"
    trans_rollback(thd);      // if a "real transaction"
1237 1238
  }
  m_table_map.clear_tables();
1239
  slave_close_thread_tables(thd);
1240 1241
  if (error)
    thd->mdl_context.release_transactional_locks();
1242
  clear_flag(IN_STMT);
1243 1244 1245
  /*
    Cleanup for the flags that have been set at do_apply_event.
  */
1246 1247
  thd->variables.option_bits&= ~OPTION_NO_FOREIGN_KEY_CHECKS;
  thd->variables.option_bits&= ~OPTION_RELAXED_UNIQUE_CHECKS;
1248 1249 1250 1251 1252 1253 1254 1255 1256

  /*
    Reset state related to long_find_row notes in the error log:
    - timestamp
    - flag that decides whether the slave prints or not
  */
  reset_row_stmt_start_timestamp();
  unset_long_find_row_note_printed();

1257 1258
  DBUG_VOID_RETURN;
}
1259

1260
void Relay_log_info::clear_tables_to_lock()
1261 1262 1263
{
  while (tables_to_lock)
  {
1264
    uchar* to_free= reinterpret_cast<uchar*>(tables_to_lock);
1265 1266 1267 1268 1269
    if (tables_to_lock->m_tabledef_valid)
    {
      tables_to_lock->m_tabledef.table_def::~table_def();
      tables_to_lock->m_tabledef_valid= FALSE;
    }
1270 1271 1272 1273 1274 1275 1276 1277 1278 1279

    /*
      If blob fields were used during conversion of field values 
      from the master table into the slave table, then we need to 
      free the memory used temporarily to store their values before
      copying into the slave's table.
    */
    if (tables_to_lock->m_conv_table)
      free_blobs(tables_to_lock->m_conv_table);

1280 1281 1282
    tables_to_lock=
      static_cast<RPL_TABLE_LIST*>(tables_to_lock->next_global);
    tables_to_lock_count--;
1283
    my_free(to_free);
1284 1285 1286 1287
  }
  DBUG_ASSERT(tables_to_lock == NULL && tables_to_lock_count == 0);
}

1288 1289
void Relay_log_info::slave_close_thread_tables(THD *thd)
{
1290 1291 1292 1293
  thd->stmt_da->can_overwrite_status= TRUE;
  thd->is_error() ? trans_rollback_stmt(thd) : trans_commit_stmt(thd);
  thd->stmt_da->can_overwrite_status= FALSE;

1294
  close_thread_tables(thd);
1295 1296 1297 1298 1299 1300 1301 1302 1303 1304 1305 1306
  /*
    - If inside a multi-statement transaction,
    defer the release of metadata locks until the current
    transaction is either committed or rolled back. This prevents
    other statements from modifying the table for the entire
    duration of this transaction.  This provides commit ordering
    and guarantees serializability across multiple transactions.
    - If in autocommit mode, or outside a transactional context,
    automatically release metadata locks of the current statement.
  */
  if (! thd->in_multi_stmt_transaction_mode())
    thd->mdl_context.release_transactional_locks();
1307 1308 1309
  else
    thd->mdl_context.release_statement_locks();

1310 1311
  clear_tables_to_lock();
}
1312
#endif