Make Replication filter settings dynamic.

Make the slave options --replicate-* dynamic variables so that these
options can be changed dynamically while the server is running,
which enables users to modify replication filtering rules without
having to stop and restart the server.

This is accomplished by just requiring that the slave threads are
stopped when these options are set dynamically. Since filtering
rules are only used by the SQL slave thread, setting them while the
thread is not running avoids the need for locking.

1 files changed, 226 insertions, 10 deletions

diff --git a/sql/rpl_filter.cc b/sql/rpl_filter.cc
index 5f5473e09ab..0380bc323a3 100644
--- a/sql/rpl_filter.cc
+++ b/sql/rpl_filter.cc
@@ -42,8 +42,8 @@ Rpl_filter::~Rpl_filter()
     free_string_array(&wild_do_table);
   if (wild_ignore_table_inited)
     free_string_array(&wild_ignore_table);
-  free_list(&do_db);
-  free_list(&ignore_db);
+  free_string_list(&do_db);
+  free_string_list(&ignore_db);
   free_list(&rewrite_db);
 }
 
@@ -263,6 +263,57 @@ Rpl_filter::is_on()
 }
 
 
+/**
+  Parse and add the given comma-separated sequence of filter rules.
+
+  @param  spec  Comma-separated sequence of filter rules.
+  @param  add   Callback member function to add a filter rule.
+
+  @return true if error, false otherwise.
+*/
+
+int
+Rpl_filter::parse_filter_rule(const char* spec, Add_filter add)
+{
+  int status= 0;
+  char *arg, *ptr, *pstr;
+
+  if (! (ptr= my_strdup(spec, MYF(MY_WME))))
+    return true;
+
+  pstr= ptr;
+
+  while (pstr)
+  {
+    arg= pstr;
+
+    /* Parse token string. */
+    pstr= strpbrk(arg, ",");
+
+    /* NUL terminate the token string. */
+    if (pstr)
+      *pstr++= '\0';
+
+    /* Skip an empty token string. */
+    if (arg[0] == '\0')
+      continue;
+
+    /* Skip leading spaces.  */
+    while (my_isspace(system_charset_info, *arg))
+      arg++;
+
+    status= (this->*add)(arg);
+
+    if (status)
+      break;
+  }
+
+  my_free(ptr);
+
+  return status;
+}
+
+
 int 
 Rpl_filter::add_do_table(const char* table_spec) 
 {
@@ -285,6 +336,46 @@ Rpl_filter::add_ignore_table(const char* table_spec)
 }
 
 
+int
+Rpl_filter::set_do_table(const char* table_spec)
+{
+  int status;
+
+  if (do_table_inited)
+    my_hash_reset(&do_table);
+
+  status= parse_filter_rule(table_spec, &Rpl_filter::add_do_table);
+
+  if (!do_table.records)
+  {
+    my_hash_free(&do_table);
+    do_table_inited= 0;
+  }
+
+  return status;
+}
+
+
+int
+Rpl_filter::set_ignore_table(const char* table_spec)
+{
+  int status;
+
+  if (ignore_table_inited)
+    my_hash_reset(&ignore_table);
+
+  status= parse_filter_rule(table_spec, &Rpl_filter::add_ignore_table);
+
+  if (!ignore_table.records)
+  {
+    my_hash_free(&ignore_table);
+    ignore_table_inited= 0;
+  }
+
+  return status;
+}
+
+
 int 
 Rpl_filter::add_wild_do_table(const char* table_spec)
 {
@@ -307,6 +398,46 @@ Rpl_filter::add_wild_ignore_table(const char* table_spec)
 }
 
 
+int
+Rpl_filter::set_wild_do_table(const char* table_spec)
+{
+  int status;
+
+  if (wild_do_table_inited)
+    free_string_array(&wild_do_table);
+
+  status= parse_filter_rule(table_spec, &Rpl_filter::add_wild_do_table);
+
+  if (!wild_do_table.elements)
+  {
+    delete_dynamic(&wild_do_table);
+    wild_do_table_inited= 0;
+  }
+
+  return status;
+}
+
+
+int
+Rpl_filter::set_wild_ignore_table(const char* table_spec)
+{
+  int status;
+
+  if (wild_ignore_table_inited)
+    free_string_array(&wild_ignore_table);
+
+  status= parse_filter_rule(table_spec, &Rpl_filter::add_wild_ignore_table);
+
+  if (!wild_ignore_table.elements)
+  {
+    delete_dynamic(&wild_ignore_table);
+    wild_ignore_table_inited= 0;
+  }
+
+  return status;
+}
+
+
 void
 Rpl_filter::add_db_rewrite(const char* from_db, const char* to_db)
 {
@@ -355,25 +486,59 @@ Rpl_filter::add_wild_table_rule(DYNAMIC_ARRAY* a, const char* table_spec)
 }
 
 
-void
+int
+Rpl_filter::add_string_list(I_List<i_string> *list, const char* spec)
+{
+  char *str;
+  i_string *node;
+
+  if (! (str= my_strdup(spec, MYF(MY_WME))))
+    return true;
+
+  if (! (node= new i_string(str)))
+  {
+    my_free(str);
+    return true;
+  }
+
+  list->push_back(node);
+
+  return false;
+}
+
+
+int
 Rpl_filter::add_do_db(const char* table_spec)
 {
   DBUG_ENTER("Rpl_filter::add_do_db");
-  i_string *db = new i_string(table_spec);
-  do_db.push_back(db);
-  DBUG_VOID_RETURN;
+  DBUG_RETURN(add_string_list(&do_db, table_spec));
 }
 
 
-void
+int
 Rpl_filter::add_ignore_db(const char* table_spec)
 {
   DBUG_ENTER("Rpl_filter::add_ignore_db");
-  i_string *db = new i_string(table_spec);
-  ignore_db.push_back(db);
-  DBUG_VOID_RETURN;
+  DBUG_RETURN(add_string_list(&ignore_db, table_spec));
 }
 
+
+int
+Rpl_filter::set_do_db(const char* db_spec)
+{
+  free_string_list(&do_db);
+  return parse_filter_rule(db_spec, &Rpl_filter::add_do_db);
+}
+
+
+int
+Rpl_filter::set_ignore_db(const char* db_spec)
+{
+  free_string_list(&ignore_db);
+  return parse_filter_rule(db_spec, &Rpl_filter::add_ignore_db);
+}
+
+
 extern "C" uchar *get_table_key(const uchar *, size_t *, my_bool);
 extern "C" void free_table_ent(void* a);
 
@@ -448,6 +613,23 @@ Rpl_filter::free_string_array(DYNAMIC_ARRAY *a)
 }
 
 
+void
+Rpl_filter::free_string_list(I_List<i_string> *l)
+{
+  void *ptr;
+  i_string *tmp;
+
+  while ((tmp= l->get()))
+  {
+    ptr= (void *) tmp->ptr;
+    my_free(ptr);
+    delete tmp;
+  }
+
+  l->empty();
+}
+
+
 /*
   Builds a String from a HASH of TABLE_RULE_ENT. Cannot be used for any other 
   hash, as it assumes that the hash entries are TABLE_RULE_ENT.
@@ -564,3 +746,37 @@ Rpl_filter::get_ignore_db()
 {
   return &ignore_db;
 }
+
+
+void
+Rpl_filter::db_rule_ent_list_to_str(String* str, I_List<i_string>* list)
+{
+  I_List_iterator<i_string> it(*list);
+  i_string* s;
+
+  str->length(0);
+
+  while ((s= it++))
+  {
+    str->append(s->ptr);
+    str->append(',');
+  }
+
+  // Remove last ','
+  if (!str->is_empty())
+    str->chop();
+}
+
+
+void
+Rpl_filter::get_do_db(String* str)
+{
+  db_rule_ent_list_to_str(str, get_do_db());
+}
+
+
+void
+Rpl_filter::get_ignore_db(String* str)
+{
+  db_rule_ent_list_to_str(str, get_ignore_db());
+}