summaryrefslogtreecommitdiff
path: root/sql
diff options
context:
space:
mode:
authorAndrei Elkin <aelkin@mysql.com>2009-10-01 19:44:53 +0300
committerAndrei Elkin <aelkin@mysql.com>2009-10-01 19:44:53 +0300
commitf68119a74541479ec38164be788dbc7ace664564 (patch)
treee2849e3cfc039882029b2eb6059cefa89b85f0de /sql
parent8acf951e223a8cd68d2d6f795ddb27ffa93f0133 (diff)
downloadmariadb-git-f68119a74541479ec38164be788dbc7ace664564.tar.gz
backporting bug@27808 fixes
Diffstat (limited to 'sql')
-rw-r--r--sql/lex.h1
-rw-r--r--sql/rpl_mi.cc88
-rw-r--r--sql/rpl_mi.h9
-rw-r--r--sql/share/errmsg.txt3
-rw-r--r--sql/slave.cc160
-rw-r--r--sql/sql_lex.h3
-rw-r--r--sql/sql_repl.cc58
-rw-r--r--sql/sql_yacc.yy23
8 files changed, 313 insertions, 32 deletions
diff --git a/sql/lex.h b/sql/lex.h
index 790808a8c14..0a61a92fd60 100644
--- a/sql/lex.h
+++ b/sql/lex.h
@@ -244,6 +244,7 @@ static SYMBOL symbols[] = {
{ "IDENTIFIED", SYM(IDENTIFIED_SYM)},
{ "IF", SYM(IF)},
{ "IGNORE", SYM(IGNORE_SYM)},
+ { "IGNORE_SERVER_IDS", SYM(IGNORE_SERVER_IDS_SYM)},
{ "IMPORT", SYM(IMPORT)},
{ "IN", SYM(IN_SYM)},
{ "INDEX", SYM(INDEX_SYM)},
diff --git a/sql/rpl_mi.cc b/sql/rpl_mi.cc
index 77f7b7e1929..a50f75d53e3 100644
--- a/sql/rpl_mi.cc
+++ b/sql/rpl_mi.cc
@@ -27,17 +27,19 @@ 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);
int init_floatvar_from_file(float* var, IO_CACHE* f, float default_val);
+int init_dynarray_intvar_from_file(DYNAMIC_ARRAY* arr, IO_CACHE* f);
Master_info::Master_info()
:Slave_reporting_capability("I/O"),
ssl(0), ssl_verify_server_cert(0), fd(-1), io_thd(0), inited(0),
abort_slave(0),slave_running(0), slave_run_id(0),
- heartbeat_period(0), received_heartbeats(0)
+ heartbeat_period(0), received_heartbeats(0), master_id(0)
{
host[0] = 0; user[0] = 0; password[0] = 0;
ssl_ca[0]= 0; ssl_capath[0]= 0; ssl_cert[0]= 0;
ssl_cipher[0]= 0; ssl_key[0]= 0;
+ my_init_dynamic_array(&ignore_server_ids, sizeof(::server_id), 16, 16);
bzero((char*) &file, sizeof(file));
pthread_mutex_init(&run_lock, MY_MUTEX_INIT_FAST);
pthread_mutex_init(&data_lock, MY_MUTEX_INIT_FAST);
@@ -48,6 +50,7 @@ Master_info::Master_info()
Master_info::~Master_info()
{
+ delete_dynamic(&ignore_server_ids);
pthread_mutex_destroy(&run_lock);
pthread_mutex_destroy(&data_lock);
pthread_cond_destroy(&data_cond);
@@ -55,6 +58,43 @@ Master_info::~Master_info()
pthread_cond_destroy(&stop_cond);
}
+/**
+ A comparison function to be supplied as argument to @c sort_dynamic()
+ and @c bsearch()
+
+ @return -1 if first argument is less, 0 if it equal to, 1 if it is greater
+ than the second
+*/
+int change_master_server_id_cmp(ulong *id1, ulong *id2)
+{
+ return *id1 < *id2? -1 : (*id1 > *id2? 1 : 0);
+}
+
+
+/**
+ Reports if the s_id server has been configured to ignore events
+ it generates with
+
+ CHANGE MASTER IGNORE_SERVER_IDS= ( list of server ids )
+
+ Method is called from the io thread event receiver filtering.
+
+ @param s_id the master server identifier
+
+ @retval TRUE if s_id is in the list of ignored master servers,
+ @retval FALSE otherwise.
+ */
+bool Master_info::shall_ignore_server_id(ulong s_id)
+{
+ if (likely(ignore_server_ids.elements == 1))
+ return (* (ulong*) dynamic_array_ptr(&ignore_server_ids, 0)) == s_id;
+ else
+ return bsearch((const ulong *) &s_id,
+ ignore_server_ids.buffer,
+ ignore_server_ids.elements, sizeof(ulong),
+ (int (*) (const void*, const void*)) change_master_server_id_cmp)
+ != NULL;
+}
void init_master_info_with_options(Master_info* mi)
{
@@ -105,12 +145,12 @@ enum {
/* 5.1.16 added value of master_ssl_verify_server_cert */
LINE_FOR_MASTER_SSL_VERIFY_SERVER_CERT= 15,
-
/* 6.0 added value of master_heartbeat_period */
LINE_FOR_MASTER_HEARTBEAT_PERIOD= 16,
-
+ /* 6.0 added value of master_ignore_server_id */
+ LINE_FOR_REPLICATE_IGNORE_SERVER_IDS= 17,
/* Number of lines currently used when saving master info file */
- LINES_IN_MASTER_INFO= LINE_FOR_MASTER_HEARTBEAT_PERIOD
+ LINES_IN_MASTER_INFO= LINE_FOR_REPLICATE_IGNORE_SERVER_IDS
};
int init_master_info(Master_info* mi, const char* master_info_fname,
@@ -304,6 +344,16 @@ file '%s')", fname);
if (lines >= LINE_FOR_MASTER_HEARTBEAT_PERIOD &&
init_floatvar_from_file(&master_heartbeat_period, &mi->file, 0.0))
goto errwithmsg;
+ /*
+ Starting from 6.0 list of server_id of ignorable servers might be
+ in the file
+ */
+ if (lines >= LINE_FOR_REPLICATE_IGNORE_SERVER_IDS &&
+ init_dynarray_intvar_from_file(&mi->ignore_server_ids, &mi->file))
+ {
+ sql_print_error("Failed to initialize master info ignore_server_ids");
+ goto errwithmsg;
+ }
}
#ifndef HAVE_OPENSSL
@@ -384,7 +434,29 @@ int flush_master_info(Master_info* mi, bool flush_relay_log_cache)
if (flush_relay_log_cache &&
flush_io_cache(mi->rli.relay_log.get_log_file()))
DBUG_RETURN(2);
-
+
+ /*
+ produce a line listing the total number and all the ignored server_id:s
+ */
+ char* ignore_server_ids_buf;
+ {
+ ignore_server_ids_buf=
+ (char *) my_malloc((sizeof(::server_id) * 3 + 1) *
+ (1 + mi->ignore_server_ids.elements), MYF(MY_WME));
+ if (!ignore_server_ids_buf)
+ DBUG_RETURN(1);
+ for (ulong i= 0, cur_len= my_sprintf(ignore_server_ids_buf,
+ (ignore_server_ids_buf, "%u",
+ mi->ignore_server_ids.elements));
+ i < mi->ignore_server_ids.elements; i++)
+ {
+ ulong s_id;
+ get_dynamic(&mi->ignore_server_ids, (uchar*) &s_id, i);
+ cur_len +=my_sprintf(ignore_server_ids_buf + cur_len,
+ (ignore_server_ids_buf + cur_len,
+ " %lu", s_id));
+ }
+ }
/*
We flushed the relay log BEFORE the master.info file, because if we crash
now, we will get a duplicate event in the relay log at restart. If we
@@ -405,14 +477,16 @@ int flush_master_info(Master_info* mi, bool flush_relay_log_cache)
my_sprintf(heartbeat_buf, (heartbeat_buf, "%.3f", mi->heartbeat_period));
my_b_seek(file, 0L);
my_b_printf(file,
- "%u\n%s\n%s\n%s\n%s\n%s\n%d\n%d\n%d\n%s\n%s\n%s\n%s\n%s\n%d\n%s\n",
+ "%u\n%s\n%s\n%s\n%s\n%s\n%d\n%d\n%d\n%s\n%s\n%s\n%s\n%s\n%d\n%s\n%s\n",
LINES_IN_MASTER_INFO,
mi->master_log_name, llstr(mi->master_log_pos, lbuf),
mi->host, mi->user,
mi->password, mi->port, mi->connect_retry,
(int)(mi->ssl), mi->ssl_ca, mi->ssl_capath, mi->ssl_cert,
mi->ssl_cipher, mi->ssl_key, mi->ssl_verify_server_cert,
- heartbeat_buf);
+ heartbeat_buf, ignore_server_ids_buf);
+
+ my_free(ignore_server_ids_buf, MYF(0));
DBUG_RETURN(-flush_io_cache(file));
}
diff --git a/sql/rpl_mi.h b/sql/rpl_mi.h
index 35e18414932..0caa6904da4 100644
--- a/sql/rpl_mi.h
+++ b/sql/rpl_mi.h
@@ -20,6 +20,7 @@
#include "rpl_rli.h"
#include "rpl_reporting.h"
+#include "my_sys.h"
/*****************************************************************************
@@ -60,6 +61,7 @@ class Master_info : public Slave_reporting_capability
public:
Master_info();
~Master_info();
+ bool shall_ignore_server_id(ulong s_id);
/* the variables below are needed because we can change masters on the fly */
char master_log_name[FN_REFLEN];
@@ -83,8 +85,6 @@ class Master_info : public Slave_reporting_capability
Relay_log_info rli;
uint port;
uint connect_retry;
- float heartbeat_period; // interface with CHANGE MASTER or master.info
- ulonglong received_heartbeats; // counter of received heartbeat events
#ifndef DBUG_OFF
int events_till_disconnect;
#endif
@@ -102,6 +102,10 @@ class Master_info : public Slave_reporting_capability
*/
long clock_diff_with_master;
+ float heartbeat_period; // interface with CHANGE MASTER or master.info
+ ulonglong received_heartbeats; // counter of received heartbeat events
+ DYNAMIC_ARRAY ignore_server_ids;
+ ulong master_id;
};
void init_master_info_with_options(Master_info* mi);
@@ -111,6 +115,7 @@ int init_master_info(Master_info* mi, const char* master_info_fname,
int thread_mask);
void end_master_info(Master_info* mi);
int flush_master_info(Master_info* mi, bool flush_relay_log_cache);
+int change_master_server_id_cmp(ulong *id1, ulong *id2);
#endif /* HAVE_REPLICATION */
#endif /* RPL_MI_H */
diff --git a/sql/share/errmsg.txt b/sql/share/errmsg.txt
index 3aba434b284..18d3a41424a 100644
--- a/sql/share/errmsg.txt
+++ b/sql/share/errmsg.txt
@@ -6163,6 +6163,9 @@ ER_SLAVE_HEARTBEAT_FAILURE
ER_SLAVE_HEARTBEAT_VALUE_OUT_OF_RANGE
eng "The requested value for the heartbeat period %s %s"
+ER_SLAVE_IGNORE_SERVER_IDS
+ eng "The requested server id %d clashes with the slave startup option --replicate-same-server-id"
+
ER_NDB_REPLICATION_SCHEMA_ERROR
eng "Bad schema for mysql.ndb_replication table. Message: %-.64s"
ER_CONFLICT_FN_PARSE_ERROR
diff --git a/sql/slave.cc b/sql/slave.cc
index bbdfb8e633f..fe4a2f6ba0a 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -880,6 +880,95 @@ int init_floatvar_from_file(float* var, IO_CACHE* f, float default_val)
DBUG_RETURN(1);
}
+
+/**
+ A master info read method
+
+ This function is called from @c init_master_info() along with
+ relatives to restore some of @c active_mi members.
+ Particularly, this function is responsible for restoring
+ IGNORE_SERVER_IDS list of servers whose events the slave is
+ going to ignore (to not log them in the relay log).
+ Items being read are supposed to be decimal output of values of a
+ type shorter or equal of @c long and separated by the single space.
+
+ @param arr @c DYNAMIC_ARRAY pointer to storage for servers id
+ @param f @c IO_CACHE pointer to the source file
+
+ @retval 0 All OK
+ @retval non-zero An error
+*/
+
+int init_dynarray_intvar_from_file(DYNAMIC_ARRAY* arr, IO_CACHE* f)
+{
+ int ret= 0;
+ char buf[16 * (sizeof(long)*4 + 1)]; // static buffer to use most of times
+ char *buf_act= buf; // actual buffer can be dynamic if static is short
+ char *token, *last;
+ uint num_items; // number of items of `arr'
+ size_t read_size;
+ DBUG_ENTER("init_dynarray_intvar_from_file");
+
+ if ((read_size= my_b_gets(f, buf_act, sizeof(buf))) == 0)
+ {
+ return 0; // no line in master.info
+ }
+ if (read_size + 1 == sizeof(buf) && buf[sizeof(buf) - 2] != '\n')
+ {
+ /*
+ short read happend; allocate sufficient memory and make the 2nd read
+ */
+ char buf_work[(sizeof(long)*3 + 1)*16];
+ memcpy(buf_work, buf, sizeof(buf_work));
+ num_items= atoi(strtok_r(buf_work, " ", &last));
+ size_t snd_size;
+ /*
+ max size lower bound approximate estimation bases on the formula:
+ (the items number + items themselves) *
+ (decimal size + space) - 1 + `\n' + '\0'
+ */
+ size_t max_size= (1 + num_items) * (sizeof(long)*3 + 1) + 1;
+ buf_act= (char*) my_malloc(max_size, MYF(MY_WME));
+ memcpy(buf_act, buf, read_size);
+ snd_size= my_b_gets(f, buf_act + read_size, max_size - read_size);
+ if (snd_size == 0 ||
+ (snd_size + 1 == max_size - read_size) && buf[max_size - 2] != '\n')
+ {
+ /*
+ failure to make the 2nd read or short read again
+ */
+ ret= 1;
+ goto err;
+ }
+ }
+ token= strtok_r(buf_act, " ", &last);
+ if (token == NULL)
+ {
+ ret= 1;
+ goto err;
+ }
+ num_items= atoi(token);
+ for (uint i=0; i < num_items; i++)
+ {
+ token= strtok_r(NULL, " ", &last);
+ if (token == NULL)
+ {
+ ret= 1;
+ goto err;
+ }
+ else
+ {
+ ulong val= atol(token);
+ insert_dynamic(arr, (uchar *) &val);
+ }
+ }
+err:
+ if (buf_act != buf)
+ my_free(buf_act, MYF(0));
+ DBUG_RETURN(ret);
+}
+
+
static bool check_io_slave_killed(THD *thd, Master_info *mi, const char *info)
{
if (io_slave_killed(thd, mi))
@@ -1058,7 +1147,7 @@ static int get_master_version_and_clock(MYSQL* mysql, Master_info* mi)
(master_res= mysql_store_result(mysql)) &&
(master_row= mysql_fetch_row(master_res)))
{
- if ((::server_id == strtoul(master_row[1], 0, 10)) &&
+ if ((::server_id == (mi->master_id= strtoul(master_row[1], 0, 10))) &&
!mi->rli.replicate_same_server_id)
{
errmsg= "The slave I/O thread stops because master and slave have equal \
@@ -1096,6 +1185,13 @@ maybe it is a *VERY OLD MASTER*.");
mysql_free_result(master_res);
master_res= NULL;
}
+ if (mi->master_id == 0 && mi->ignore_server_ids.elements > 0)
+ {
+ errmsg= "Slave configured with server id filtering could not detect the master server id.";
+ err_code= ER_SLAVE_FATAL_ERROR;
+ sprintf(err_buff, ER(err_code), errmsg);
+ goto err;
+ }
/*
Check that the master's global character_set_server and ours are the same.
@@ -1659,6 +1755,10 @@ bool show_master_info(THD* thd, Master_info* mi)
field_list.push_back(new Item_empty_string("Last_IO_Error", 20));
field_list.push_back(new Item_return_int("Last_SQL_Errno", 4, MYSQL_TYPE_LONG));
field_list.push_back(new Item_empty_string("Last_SQL_Error", 20));
+ field_list.push_back(new Item_empty_string("Replicate_Ignore_Server_Ids",
+ FN_REFLEN));
+ field_list.push_back(new Item_return_int("Master_Server_Id", sizeof(ulong),
+ MYSQL_TYPE_LONG));
if (protocol->send_fields(&field_list,
Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF))
@@ -1780,6 +1880,32 @@ bool show_master_info(THD* thd, Master_info* mi)
protocol->store(mi->rli.last_error().number);
// Last_SQL_Error
protocol->store(mi->rli.last_error().message, &my_charset_bin);
+ // Replicate_Ignore_Server_Ids
+ {
+ char buff[FN_REFLEN];
+ ulong i, cur_len;
+ for (i= 0, buff[0]= 0, cur_len= 0;
+ i < mi->ignore_server_ids.elements; i++)
+ {
+ ulong s_id, slen;
+ char sbuff[FN_REFLEN];
+ get_dynamic(&mi->ignore_server_ids, (uchar*) &s_id, i);
+ slen= my_sprintf(sbuff, (sbuff, (i==0? "%lu" : ", %lu"), s_id));
+ if (cur_len + slen + 4 > FN_REFLEN)
+ {
+ /*
+ break the loop whenever remained space could not fit
+ ellipses on the next cycle
+ */
+ my_sprintf(buff + cur_len, (buff + cur_len, "..."));
+ break;
+ }
+ cur_len += my_sprintf(buff + cur_len, (buff + cur_len, "%s", sbuff));
+ }
+ protocol->store(buff, &my_charset_bin);
+ }
+ // Master_Server_id
+ protocol->store((uint32) mi->master_id);
pthread_mutex_unlock(&mi->rli.err_lock);
pthread_mutex_unlock(&mi->err_lock);
@@ -3599,6 +3725,7 @@ static int queue_event(Master_info* mi,const char* buf, ulong event_len)
ulong inc_pos;
Relay_log_info *rli= &mi->rli;
pthread_mutex_t *log_lock= rli->relay_log.get_log_lock();
+ ulong s_id;
DBUG_ENTER("queue_event");
LINT_INIT(inc_pos);
@@ -3745,9 +3872,20 @@ static int queue_event(Master_info* mi,const char* buf, ulong event_len)
*/
pthread_mutex_lock(log_lock);
-
- if ((uint4korr(buf + SERVER_ID_OFFSET) == ::server_id) &&
- !mi->rli.replicate_same_server_id)
+ s_id= uint4korr(buf + SERVER_ID_OFFSET);
+ if ((s_id == ::server_id && !mi->rli.replicate_same_server_id) ||
+ /*
+ the following conjunction deals with IGNORE_SERVER_IDS, if set
+ If the master is on the ignore list, execution of
+ format description log events and rotate events is necessary.
+ */
+ (mi->ignore_server_ids.elements > 0 &&
+ mi->shall_ignore_server_id(s_id) &&
+ /* everything is filtered out from non-master */
+ (s_id != mi->master_id ||
+ /* for the master meta information is necessary */
+ buf[EVENT_TYPE_OFFSET] != FORMAT_DESCRIPTION_EVENT &&
+ buf[EVENT_TYPE_OFFSET] != ROTATE_EVENT)))
{
/*
Do not write it to the relay log.
@@ -3762,10 +3900,14 @@ static int queue_event(Master_info* mi,const char* buf, ulong event_len)
But events which were generated by this slave and which do not exist in
the master's binlog (i.e. Format_desc, Rotate & Stop) should not increment
mi->master_log_pos.
+ If the event is originated remotely and is being filtered out by
+ IGNORE_SERVER_IDS it increments mi->master_log_pos
+ as well as rli->group_relay_log_pos.
*/
- if (buf[EVENT_TYPE_OFFSET]!=FORMAT_DESCRIPTION_EVENT &&
- buf[EVENT_TYPE_OFFSET]!=ROTATE_EVENT &&
- buf[EVENT_TYPE_OFFSET]!=STOP_EVENT)
+ if (!(s_id == ::server_id && !mi->rli.replicate_same_server_id) ||
+ buf[EVENT_TYPE_OFFSET] != FORMAT_DESCRIPTION_EVENT &&
+ buf[EVENT_TYPE_OFFSET] != ROTATE_EVENT &&
+ buf[EVENT_TYPE_OFFSET] != STOP_EVENT)
{
mi->master_log_pos+= inc_pos;
memcpy(rli->ign_master_log_name_end, mi->master_log_name, FN_REFLEN);
@@ -3773,8 +3915,8 @@ static int queue_event(Master_info* mi,const char* buf, ulong event_len)
rli->ign_master_log_pos_end= mi->master_log_pos;
}
rli->relay_log.signal_update(); // the slave SQL thread needs to re-check
- DBUG_PRINT("info", ("master_log_pos: %lu, event originating from the same server, ignored",
- (ulong) mi->master_log_pos));
+ DBUG_PRINT("info", ("master_log_pos: %lu, event originating from %u server, ignored",
+ (ulong) mi->master_log_pos, uint4korr(buf + SERVER_ID_OFFSET)));
}
else
{
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index f6effab93a4..c4ab3091d6a 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -214,10 +214,11 @@ typedef struct st_lex_master_info
changed variable or if it should be left at old value
*/
enum {LEX_MI_UNCHANGED, LEX_MI_DISABLE, LEX_MI_ENABLE}
- ssl, ssl_verify_server_cert, heartbeat_opt;
+ ssl, ssl_verify_server_cert, heartbeat_opt, repl_ignore_server_ids_opt;
char *ssl_key, *ssl_cert, *ssl_ca, *ssl_capath, *ssl_cipher;
char *relay_log_name;
ulong relay_log_pos;
+ DYNAMIC_ARRAY repl_ignore_server_ids;
} LEX_MASTER_INFO;
diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc
index cde713b1b40..eadb3362882 100644
--- a/sql/sql_repl.cc
+++ b/sql/sql_repl.cc
@@ -1266,26 +1266,27 @@ bool change_master(THD* thd, Master_info* mi)
int thread_mask;
const char* errmsg= 0;
bool need_relay_log_purge= 1;
+ bool ret= FALSE;
DBUG_ENTER("change_master");
lock_slave_threads(mi);
init_thread_mask(&thread_mask,mi,0 /*not inverse*/);
+ LEX_MASTER_INFO* lex_mi= &thd->lex->mi;
if (thread_mask) // We refuse if any slave thread is running
{
my_message(ER_SLAVE_MUST_STOP, ER(ER_SLAVE_MUST_STOP), MYF(0));
- unlock_slave_threads(mi);
- DBUG_RETURN(TRUE);
+ ret= TRUE;
+ goto err;
}
thd_proc_info(thd, "Changing master");
- LEX_MASTER_INFO* lex_mi= &thd->lex->mi;
// TODO: see if needs re-write
if (init_master_info(mi, master_info_file, relay_log_info_file, 0,
thread_mask))
{
my_message(ER_MASTER_INFO, ER(ER_MASTER_INFO), MYF(0));
- unlock_slave_threads(mi);
- DBUG_RETURN(TRUE);
+ ret= TRUE;
+ goto err;
}
/*
@@ -1330,6 +1331,34 @@ bool change_master(THD* thd, Master_info* mi)
mi->heartbeat_period= (float) min(SLAVE_MAX_HEARTBEAT_PERIOD,
(slave_net_timeout/2.0));
mi->received_heartbeats= LL(0); // counter lives until master is CHANGEd
+ /*
+ reset the last time server_id list if the current CHANGE MASTER
+ is mentioning IGNORE_SERVER_IDS= (...)
+ */
+ if (lex_mi->repl_ignore_server_ids_opt == LEX_MASTER_INFO::LEX_MI_ENABLE)
+ reset_dynamic(&mi->ignore_server_ids);
+ for (uint i= 0; i < lex_mi->repl_ignore_server_ids.elements; i++)
+ {
+ ulong s_id;
+ get_dynamic(&lex_mi->repl_ignore_server_ids, (uchar*) &s_id, i);
+ if (s_id == ::server_id && replicate_same_server_id)
+ {
+ my_error(ER_SLAVE_IGNORE_SERVER_IDS, MYF(0), s_id);
+ ret= TRUE;
+ goto err;
+ }
+ else
+ {
+ if (bsearch((const ulong *) &s_id,
+ mi->ignore_server_ids.buffer,
+ mi->ignore_server_ids.elements, sizeof(ulong),
+ (int (*) (const void*, const void*))
+ change_master_server_id_cmp) == NULL)
+ insert_dynamic(&mi->ignore_server_ids, (uchar*) &s_id);
+ }
+ }
+ sort_dynamic(&mi->ignore_server_ids, (qsort_cmp) change_master_server_id_cmp);
+
if (lex_mi->ssl != LEX_MASTER_INFO::LEX_MI_UNCHANGED)
mi->ssl= (lex_mi->ssl == LEX_MASTER_INFO::LEX_MI_ENABLE);
@@ -1407,8 +1436,8 @@ bool change_master(THD* thd, Master_info* mi)
if (flush_master_info(mi, 0))
{
my_error(ER_RELAY_LOG_INIT, MYF(0), "Failed to flush master info file");
- unlock_slave_threads(mi);
- DBUG_RETURN(TRUE);
+ ret= TRUE;
+ goto err;
}
if (need_relay_log_purge)
{
@@ -1419,8 +1448,8 @@ bool change_master(THD* thd, Master_info* mi)
&errmsg))
{
my_error(ER_RELAY_LOG_FAIL, MYF(0), errmsg);
- unlock_slave_threads(mi);
- DBUG_RETURN(TRUE);
+ ret= TRUE;
+ goto err;
}
}
else
@@ -1435,8 +1464,8 @@ bool change_master(THD* thd, Master_info* mi)
&msg, 0))
{
my_error(ER_RELAY_LOG_INIT, MYF(0), msg);
- unlock_slave_threads(mi);
- DBUG_RETURN(TRUE);
+ ret= TRUE;
+ goto err;
}
}
/*
@@ -1473,10 +1502,13 @@ bool change_master(THD* thd, Master_info* mi)
pthread_cond_broadcast(&mi->data_cond);
pthread_mutex_unlock(&mi->rli.data_lock);
+err:
unlock_slave_threads(mi);
thd_proc_info(thd, 0);
- my_ok(thd);
- DBUG_RETURN(FALSE);
+ if (ret == FALSE)
+ my_ok(thd);
+ delete_dynamic(&lex_mi->repl_ignore_server_ids); //freeing of parser-time alloc
+ DBUG_RETURN(ret);
}
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 50395d386e8..7dff91befb0 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -740,6 +740,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
%token IDENT_QUOTED
%token IF
%token IGNORE_SYM
+%token IGNORE_SERVER_IDS_SYM
%token IMPORT
%token INDEXES
%token INDEX_SYM
@@ -1559,6 +1560,12 @@ change:
LEX *lex = Lex;
lex->sql_command = SQLCOM_CHANGE_MASTER;
bzero((char*) &lex->mi, sizeof(lex->mi));
+ /*
+ resetting flags that can left from the previous CHANGE MASTER
+ */
+ lex->mi.repl_ignore_server_ids_opt= LEX_MASTER_INFO::LEX_MI_UNCHANGED;
+ my_init_dynamic_array(&Lex->mi.repl_ignore_server_ids,
+ sizeof(::server_id), 16, 16);
}
master_defs
{}
@@ -1661,10 +1668,26 @@ master_def:
}
Lex->mi.heartbeat_opt= LEX_MASTER_INFO::LEX_MI_ENABLE;
}
+ | IGNORE_SERVER_IDS_SYM EQ '(' ignore_server_id_list ')'
+ {
+ Lex->mi.repl_ignore_server_ids_opt= LEX_MASTER_INFO::LEX_MI_ENABLE;
+ }
|
master_file_def
;
+ignore_server_id_list:
+ /* Empty */
+ | ignore_server_id
+ | ignore_server_id_list ',' ignore_server_id
+ ;
+
+ignore_server_id:
+ ulong_num
+ {
+ insert_dynamic(&Lex->mi.repl_ignore_server_ids, (uchar*) &($1));
+ }
+
master_file_def:
MASTER_LOG_FILE_SYM EQ TEXT_STRING_sys
{