Commit 26787260 authored by lars@mysql.com's avatar lars@mysql.com

WL#2472: Refactoring of replication and binlog filtering.

Needed to be able to have multiple masters connected 
to same slave server with different filters.
parent b6a2b5f7
...@@ -58,7 +58,7 @@ noinst_HEADERS = item.h item_func.h item_sum.h item_cmpfunc.h \ ...@@ -58,7 +58,7 @@ noinst_HEADERS = item.h item_func.h item_sum.h item_cmpfunc.h \
log_event.h sql_repl.h slave.h \ log_event.h sql_repl.h slave.h \
stacktrace.h sql_sort.h sql_cache.h set_var.h \ stacktrace.h sql_sort.h sql_cache.h set_var.h \
spatial.h gstream.h client_settings.h tzfile.h \ spatial.h gstream.h client_settings.h tzfile.h \
tztime.h \ table_filter.h tztime.h \
sp_head.h sp_pcontext.h sp_rcontext.h sp.h sp_cache.h \ sp_head.h sp_pcontext.h sp_rcontext.h sp.h sp_cache.h \
parse_file.h sql_view.h sql_trigger.h \ parse_file.h sql_view.h sql_trigger.h \
examples/ha_example.h examples/ha_archive.h \ examples/ha_example.h examples/ha_archive.h \
...@@ -95,7 +95,7 @@ mysqld_SOURCES = sql_lex.cc sql_handler.cc \ ...@@ -95,7 +95,7 @@ mysqld_SOURCES = sql_lex.cc sql_handler.cc \
stacktrace.c repl_failsafe.h repl_failsafe.cc \ stacktrace.c repl_failsafe.h repl_failsafe.cc \
sql_olap.cc sql_view.cc \ sql_olap.cc sql_view.cc \
gstream.cc spatial.cc sql_help.cc protocol_cursor.cc \ gstream.cc spatial.cc sql_help.cc protocol_cursor.cc \
tztime.cc my_time.c \ table_filter.cc tztime.cc my_time.c \
sp_head.cc sp_pcontext.cc sp_rcontext.cc sp.cc \ sp_head.cc sp_pcontext.cc sp_rcontext.cc sp.cc \
sp_cache.cc parse_file.cc sql_trigger.cc \ sp_cache.cc parse_file.cc sql_trigger.cc \
examples/ha_example.cc examples/ha_archive.cc \ examples/ha_example.cc examples/ha_archive.cc \
......
...@@ -24,6 +24,7 @@ ...@@ -24,6 +24,7 @@
#include "mysql_priv.h" #include "mysql_priv.h"
#include "sql_repl.h" #include "sql_repl.h"
#include "table_filter.h"
#include "ha_innodb.h" // necessary to cut the binlog when crash recovery #include "ha_innodb.h" // necessary to cut the binlog when crash recovery
#include <my_dir.h> #include <my_dir.h>
...@@ -34,6 +35,8 @@ ...@@ -34,6 +35,8 @@
#include "message.h" #include "message.h"
#endif #endif
extern Table_filter *binlog_filter;
MYSQL_LOG mysql_log, mysql_slow_log, mysql_bin_log; MYSQL_LOG mysql_log, mysql_slow_log, mysql_bin_log;
ulong sync_binlog_counter= 0; ulong sync_binlog_counter= 0;
...@@ -1325,10 +1328,11 @@ bool MYSQL_LOG::write(Log_event* event_info) ...@@ -1325,10 +1328,11 @@ bool MYSQL_LOG::write(Log_event* event_info)
binlog_[wild_]{do|ignore}_table?" (WL#1049)" binlog_[wild_]{do|ignore}_table?" (WL#1049)"
*/ */
if ((thd && !(thd->options & OPTION_BIN_LOG)) || if ((thd && !(thd->options & OPTION_BIN_LOG)) ||
(local_db && !db_ok(local_db, binlog_do_db, binlog_ignore_db))) (local_db && !binlog_filter->db_ok(local_db)))
{ {
VOID(pthread_mutex_unlock(&LOCK_log)); VOID(pthread_mutex_unlock(&LOCK_log));
DBUG_PRINT("error",("!db_ok('%s')", local_db)); DBUG_PRINT("info",("db_ok('%s')==%d", local_db,
binlog_filter->db_ok(local_db)));
DBUG_RETURN(0); DBUG_RETURN(0);
} }
#endif /* HAVE_REPLICATION */ #endif /* HAVE_REPLICATION */
......
...@@ -21,7 +21,9 @@ ...@@ -21,7 +21,9 @@
#endif #endif
#include "mysql_priv.h" #include "mysql_priv.h"
#include "slave.h" #include "slave.h"
#include "table_filter.h"
#include <my_dir.h> #include <my_dir.h>
extern Table_filter *rpl_filter;
#endif /* MYSQL_CLIENT */ #endif /* MYSQL_CLIENT */
#define log_cs &my_charset_latin1 #define log_cs &my_charset_latin1
...@@ -1370,7 +1372,7 @@ int Query_log_event::exec_event(struct st_relay_log_info* rli) ...@@ -1370,7 +1372,7 @@ int Query_log_event::exec_event(struct st_relay_log_info* rli)
*/ */
thd->catalog= (char*) catalog; thd->catalog= (char*) catalog;
thd->db_length= db_len; thd->db_length= db_len;
thd->db= (char*) rewrite_db(db, &thd->db_length); thd->db= (char *) rpl_filter->get_rewrite_db(db, &thd->db_length);
thd->variables.auto_increment_increment= auto_increment_increment; thd->variables.auto_increment_increment= auto_increment_increment;
thd->variables.auto_increment_offset= auto_increment_offset; thd->variables.auto_increment_offset= auto_increment_offset;
...@@ -1389,7 +1391,7 @@ int Query_log_event::exec_event(struct st_relay_log_info* rli) ...@@ -1389,7 +1391,7 @@ int Query_log_event::exec_event(struct st_relay_log_info* rli)
clear_all_errors(thd, rli); clear_all_errors(thd, rli);
if (db_ok(thd->db, replicate_do_db, replicate_ignore_db)) if (rpl_filter->db_ok(thd->db))
{ {
thd->set_time((time_t)when); thd->set_time((time_t)when);
thd->query_length= q_len; thd->query_length= q_len;
...@@ -2449,7 +2451,7 @@ int Load_log_event::exec_event(NET* net, struct st_relay_log_info* rli, ...@@ -2449,7 +2451,7 @@ int Load_log_event::exec_event(NET* net, struct st_relay_log_info* rli,
{ {
char *load_data_query= 0; char *load_data_query= 0;
thd->db_length= db_len; thd->db_length= db_len;
thd->db= (char*) rewrite_db(db, &thd->db_length); thd->db= (char *) rpl_filter->get_rewrite_db(db, &thd->db_length);
DBUG_ASSERT(thd->query == 0); DBUG_ASSERT(thd->query == 0);
thd->query_length= 0; // Should not be needed thd->query_length= 0; // Should not be needed
thd->query_error= 0; thd->query_error= 0;
...@@ -2478,7 +2480,7 @@ int Load_log_event::exec_event(NET* net, struct st_relay_log_info* rli, ...@@ -2478,7 +2480,7 @@ int Load_log_event::exec_event(NET* net, struct st_relay_log_info* rli,
al. Another way is do the filtering in the I/O thread (more efficient: no al. Another way is do the filtering in the I/O thread (more efficient: no
disk writes at all). disk writes at all).
*/ */
if (db_ok(thd->db, replicate_do_db, replicate_ignore_db)) if (rpl_filter->db_ok(thd->db))
{ {
thd->set_time((time_t)when); thd->set_time((time_t)when);
VOID(pthread_mutex_lock(&LOCK_thread_count)); VOID(pthread_mutex_lock(&LOCK_thread_count));
...@@ -2500,7 +2502,7 @@ int Load_log_event::exec_event(NET* net, struct st_relay_log_info* rli, ...@@ -2500,7 +2502,7 @@ int Load_log_event::exec_event(NET* net, struct st_relay_log_info* rli,
tables.updating= 1; tables.updating= 1;
// the table will be opened in mysql_load // the table will be opened in mysql_load
if (table_rules_on && !tables_ok(thd, &tables)) if (rpl_filter->is_on() && !rpl_filter->tables_ok(thd->db, &tables))
{ {
// TODO: this is a bug - this needs to be moved to the I/O thread // TODO: this is a bug - this needs to be moved to the I/O thread
if (net) if (net)
......
...@@ -1072,7 +1072,6 @@ extern KNOWN_DATE_TIME_FORMAT known_date_time_formats[]; ...@@ -1072,7 +1072,6 @@ extern KNOWN_DATE_TIME_FORMAT known_date_time_formats[];
extern String null_string; extern String null_string;
extern HASH open_cache; extern HASH open_cache;
extern TABLE *unused_tables; extern TABLE *unused_tables;
extern I_List<i_string> binlog_do_db, binlog_ignore_db;
extern const char* any_db; extern const char* any_db;
extern struct my_option my_long_options[]; extern struct my_option my_long_options[];
......
...@@ -19,6 +19,7 @@ ...@@ -19,6 +19,7 @@
#include <my_dir.h> #include <my_dir.h>
#include "slave.h" #include "slave.h"
#include "sql_repl.h" #include "sql_repl.h"
#include "table_filter.h"
#include "repl_failsafe.h" #include "repl_failsafe.h"
#include "stacktrace.h" #include "stacktrace.h"
#include "mysqld_suffix.h" #include "mysqld_suffix.h"
...@@ -382,12 +383,10 @@ Le_creator le_creator; ...@@ -382,12 +383,10 @@ Le_creator le_creator;
FILE *bootstrap_file; FILE *bootstrap_file;
I_List<i_string_pair> replicate_rewrite_db;
I_List<i_string> replicate_do_db, replicate_ignore_db;
// allow the user to tell us which db to replicate and which to ignore
I_List<i_string> binlog_do_db, binlog_ignore_db;
I_List<THD> threads,thread_cache; I_List<THD> threads,thread_cache;
I_List<NAMED_LIST> key_caches; I_List<NAMED_LIST> key_caches;
Table_filter* rpl_filter;
Table_filter* binlog_filter;
struct system_variables global_system_variables; struct system_variables global_system_variables;
struct system_variables max_system_variables; struct system_variables max_system_variables;
...@@ -994,11 +993,6 @@ void clean_up(bool print_message) ...@@ -994,11 +993,6 @@ void clean_up(bool print_message)
free_max_user_conn(); free_max_user_conn();
#ifdef HAVE_REPLICATION #ifdef HAVE_REPLICATION
end_slave_list(); end_slave_list();
free_list(&replicate_do_db);
free_list(&replicate_ignore_db);
free_list(&binlog_do_db);
free_list(&binlog_ignore_db);
free_list(&replicate_rewrite_db);
#endif #endif
#ifdef HAVE_OPENSSL #ifdef HAVE_OPENSSL
if (ssl_acceptor_fd) if (ssl_acceptor_fd)
...@@ -2967,9 +2961,16 @@ int win_main(int argc, char **argv) ...@@ -2967,9 +2961,16 @@ int win_main(int argc, char **argv)
int main(int argc, char **argv) int main(int argc, char **argv)
#endif #endif
{ {
DEBUGGER_OFF; DEBUGGER_OFF;
rpl_filter= new Table_filter;
binlog_filter= new Table_filter;
if (!rpl_filter || !binlog_filter)
{
sql_perror("Could not allocate replication and binlog filters");
exit(1);
}
MY_INIT(argv[0]); // init my_sys library & pthreads MY_INIT(argv[0]); // init my_sys library & pthreads
#ifdef _CUSTOMSTARTUPCONFIG_ #ifdef _CUSTOMSTARTUPCONFIG_
...@@ -3214,6 +3215,9 @@ we force server id to 2, but this MySQL server will not act as a slave."); ...@@ -3214,6 +3215,9 @@ we force server id to 2, but this MySQL server will not act as a slave.");
clean_up_mutexes(); clean_up_mutexes();
my_end(opt_endinfo ? MY_CHECK_ERROR | MY_GIVE_INFO : 0); my_end(opt_endinfo ? MY_CHECK_ERROR | MY_GIVE_INFO : 0);
delete rpl_filter;
delete binlog_filter;
exit(0); exit(0);
return(0); /* purecov: deadcode */ return(0); /* purecov: deadcode */
} }
...@@ -3309,7 +3313,6 @@ default_service_handling(char **argv, ...@@ -3309,7 +3313,6 @@ default_service_handling(char **argv,
int main(int argc, char **argv) int main(int argc, char **argv)
{ {
/* When several instances are running on the same machine, we /* When several instances are running on the same machine, we
need to have an unique named hEventShudown through the need to have an unique named hEventShudown through the
application PID e.g.: MySQLShutdown1890; MySQLShutdown2342 application PID e.g.: MySQLShutdown1890; MySQLShutdown2342
...@@ -5875,13 +5878,6 @@ static void mysql_init_variables(void) ...@@ -5875,13 +5878,6 @@ static void mysql_init_variables(void)
default_key_cache_base.length))) default_key_cache_base.length)))
exit(1); exit(1);
/* Initialize structures that is used when processing options */
replicate_rewrite_db.empty();
replicate_do_db.empty();
replicate_ignore_db.empty();
binlog_do_db.empty();
binlog_ignore_db.empty();
/* Set directory paths */ /* Set directory paths */
strmake(language, LANGUAGE, sizeof(language)-1); strmake(language, LANGUAGE, sizeof(language)-1);
strmake(mysql_real_data_home, get_relative_path(DATADIR), strmake(mysql_real_data_home, get_relative_path(DATADIR),
...@@ -6136,14 +6132,12 @@ get_one_option(int optid, const struct my_option *opt __attribute__((unused)), ...@@ -6136,14 +6132,12 @@ get_one_option(int optid, const struct my_option *opt __attribute__((unused)),
} }
case (int)OPT_REPLICATE_IGNORE_DB: case (int)OPT_REPLICATE_IGNORE_DB:
{ {
i_string *db = new i_string(argument); rpl_filter->add_ignore_db(argument);
replicate_ignore_db.push_back(db);
break; break;
} }
case (int)OPT_REPLICATE_DO_DB: case (int)OPT_REPLICATE_DO_DB:
{ {
i_string *db = new i_string(argument); rpl_filter->add_do_db(argument);
replicate_do_db.push_back(db);
break; break;
} }
case (int)OPT_REPLICATE_REWRITE_DB: case (int)OPT_REPLICATE_REWRITE_DB:
...@@ -6176,71 +6170,54 @@ get_one_option(int optid, const struct my_option *opt __attribute__((unused)), ...@@ -6176,71 +6170,54 @@ get_one_option(int optid, const struct my_option *opt __attribute__((unused)),
exit(1); exit(1);
} }
i_string_pair *db_pair = new i_string_pair(key, val); rpl_filter->add_db_rewrite(key, val);
replicate_rewrite_db.push_back(db_pair);
break; break;
} }
case (int)OPT_BINLOG_IGNORE_DB: case (int)OPT_BINLOG_IGNORE_DB:
{ {
i_string *db = new i_string(argument); binlog_filter->add_ignore_db(argument);
binlog_ignore_db.push_back(db);
break; break;
} }
case (int)OPT_BINLOG_DO_DB: case (int)OPT_BINLOG_DO_DB:
{ {
i_string *db = new i_string(argument); binlog_filter->add_do_db(argument);
binlog_do_db.push_back(db);
break; break;
} }
case (int)OPT_REPLICATE_DO_TABLE: case (int)OPT_REPLICATE_DO_TABLE:
{ {
if (!do_table_inited) if (rpl_filter->add_do_table(argument))
init_table_rule_hash(&replicate_do_table, &do_table_inited);
if (add_table_rule(&replicate_do_table, argument))
{ {
fprintf(stderr, "Could not add do table rule '%s'!\n", argument); fprintf(stderr, "Could not add do table rule '%s'!\n", argument);
exit(1); exit(1);
} }
table_rules_on = 1;
break; break;
} }
case (int)OPT_REPLICATE_WILD_DO_TABLE: case (int)OPT_REPLICATE_WILD_DO_TABLE:
{ {
if (!wild_do_table_inited) if (rpl_filter->add_wild_do_table(argument))
init_table_rule_array(&replicate_wild_do_table,
&wild_do_table_inited);
if (add_wild_table_rule(&replicate_wild_do_table, argument))
{ {
fprintf(stderr, "Could not add do table rule '%s'!\n", argument); fprintf(stderr, "Could not add do table rule '%s'!\n", argument);
exit(1); exit(1);
} }
table_rules_on = 1;
break; break;
} }
case (int)OPT_REPLICATE_WILD_IGNORE_TABLE: case (int)OPT_REPLICATE_WILD_IGNORE_TABLE:
{ {
if (!wild_ignore_table_inited) if (rpl_filter->add_wild_ignore_table(argument))
init_table_rule_array(&replicate_wild_ignore_table,
&wild_ignore_table_inited);
if (add_wild_table_rule(&replicate_wild_ignore_table, argument))
{ {
fprintf(stderr, "Could not add ignore table rule '%s'!\n", argument); fprintf(stderr, "Could not add ignore table rule '%s'!\n", argument);
exit(1); exit(1);
} }
table_rules_on = 1;
break; break;
} }
case (int)OPT_REPLICATE_IGNORE_TABLE: case (int)OPT_REPLICATE_IGNORE_TABLE:
{ {
if (!ignore_table_inited) if (rpl_filter->add_ignore_table(argument))
init_table_rule_hash(&replicate_ignore_table, &ignore_table_inited);
if (add_table_rule(&replicate_ignore_table, argument))
{ {
fprintf(stderr, "Could not add ignore table rule '%s'!\n", argument); fprintf(stderr, "Could not add ignore table rule '%s'!\n", argument);
exit(1); exit(1);
} }
table_rules_on = 1;
break; break;
} }
#endif /* HAVE_REPLICATION */ #endif /* HAVE_REPLICATION */
......
...@@ -20,9 +20,12 @@ ...@@ -20,9 +20,12 @@
#include "repl_failsafe.h" #include "repl_failsafe.h"
#include "sql_repl.h" #include "sql_repl.h"
#include "slave.h" #include "slave.h"
#include "table_filter.h"
#include "log_event.h" #include "log_event.h"
#include <mysql.h> #include <mysql.h>
extern Table_filter *rpl_filter;
#define SLAVE_LIST_CHUNK 128 #define SLAVE_LIST_CHUNK 128
#define SLAVE_ERRMSG_SIZE (FN_REFLEN+64) #define SLAVE_ERRMSG_SIZE (FN_REFLEN+64)
...@@ -735,14 +738,14 @@ static int fetch_db_tables(THD *thd, MYSQL *mysql, const char *db, ...@@ -735,14 +738,14 @@ static int fetch_db_tables(THD *thd, MYSQL *mysql, const char *db,
TABLE_LIST table; TABLE_LIST table;
const char* table_name= row[0]; const char* table_name= row[0];
int error; int error;
if (table_rules_on) if (rpl_filter->is_on())
{ {
bzero((char*) &table, sizeof(table)); //just for safe bzero((char*) &table, sizeof(table)); //just for safe
table.db= (char*) db; table.db= (char*) db;
table.table_name= (char*) table_name; table.table_name= (char*) table_name;
table.updating= 1; table.updating= 1;
if (!tables_ok(thd, &table)) if (!rpl_filter->tables_ok(thd->db, &table))
continue; continue;
} }
/* download master's table and overwrite slave's table */ /* download master's table and overwrite slave's table */
...@@ -860,8 +863,8 @@ bool load_master_data(THD* thd) ...@@ -860,8 +863,8 @@ bool load_master_data(THD* thd)
data from master data from master
*/ */
if (!db_ok(db, replicate_do_db, replicate_ignore_db) || if (!rpl_filter->db_ok(db) ||
!db_ok_with_wild_table(db) || !rpl_filter->db_ok_with_wild_table(db) ||
!strcmp(db,"mysql")) !strcmp(db,"mysql"))
{ {
*cur_table_res = 0; *cur_table_res = 0;
......
This diff is collapsed.
...@@ -443,15 +443,6 @@ typedef struct st_master_info ...@@ -443,15 +443,6 @@ typedef struct st_master_info
int queue_event(MASTER_INFO* mi,const char* buf,ulong event_len); int queue_event(MASTER_INFO* mi,const char* buf,ulong event_len);
typedef struct st_table_rule_ent
{
char* db;
char* tbl_name;
uint key_len;
} TABLE_RULE_ENT;
#define TABLE_RULE_HASH_SIZE 16
#define TABLE_RULE_ARR_SIZE 16
#define MAX_SLAVE_ERRMSG 1024 #define MAX_SLAVE_ERRMSG 1024
#define RPL_LOG_NAME (rli->group_master_log_name[0] ? rli->group_master_log_name :\ #define RPL_LOG_NAME (rli->group_master_log_name[0] ? rli->group_master_log_name :\
...@@ -505,27 +496,9 @@ int mysql_table_dump(THD* thd, const char* db, ...@@ -505,27 +496,9 @@ int mysql_table_dump(THD* thd, const char* db,
int fetch_master_table(THD* thd, const char* db_name, const char* table_name, int fetch_master_table(THD* thd, const char* db_name, const char* table_name,
MASTER_INFO* mi, MYSQL* mysql, bool overwrite); MASTER_INFO* mi, MYSQL* mysql, bool overwrite);
void table_rule_ent_hash_to_str(String* s, HASH* h);
void table_rule_ent_dynamic_array_to_str(String* s, DYNAMIC_ARRAY* a);
bool show_master_info(THD* thd, MASTER_INFO* mi); bool show_master_info(THD* thd, MASTER_INFO* mi);
bool show_binlog_info(THD* thd); bool show_binlog_info(THD* thd);
/* See if the query uses any tables that should not be replicated */
bool tables_ok(THD* thd, TABLE_LIST* tables);
/*
Check to see if the database is ok to operate on with respect to the
do and ignore lists - used in replication
*/
int db_ok(const char* db, I_List<i_string> &do_list,
I_List<i_string> &ignore_list );
int db_ok_with_wild_table(const char *db);
int add_table_rule(HASH* h, const char* table_spec);
int add_wild_table_rule(DYNAMIC_ARRAY* a, const char* table_spec);
void init_table_rule_hash(HASH* h, bool* h_inited);
void init_table_rule_array(DYNAMIC_ARRAY* a, bool* a_inited);
const char *rewrite_db(const char* db, uint32 *new_db_len);
const char *print_slave_db_safe(const char *db); const char *print_slave_db_safe(const char *db);
int check_expected_error(THD* thd, RELAY_LOG_INFO* rli, int error_code); int check_expected_error(THD* thd, RELAY_LOG_INFO* rli, int error_code);
void skip_load_data_infile(NET* net); void skip_load_data_infile(NET* net);
...@@ -559,11 +532,7 @@ extern "C" pthread_handler_decl(handle_slave_sql,arg); ...@@ -559,11 +532,7 @@ extern "C" pthread_handler_decl(handle_slave_sql,arg);
extern bool volatile abort_loop; extern bool volatile abort_loop;
extern MASTER_INFO main_mi, *active_mi; /* active_mi for multi-master */ extern MASTER_INFO main_mi, *active_mi; /* active_mi for multi-master */
extern LIST master_list; extern LIST master_list;
extern HASH replicate_do_table, replicate_ignore_table; extern bool replicate_same_server_id;
extern DYNAMIC_ARRAY replicate_wild_do_table, replicate_wild_ignore_table;
extern bool do_table_inited, ignore_table_inited,
wild_do_table_inited, wild_ignore_table_inited;
extern bool table_rules_on, replicate_same_server_id;
extern int disconnect_slave_event_count, abort_slave_event_count ; extern int disconnect_slave_event_count, abort_slave_event_count ;
...@@ -577,8 +546,6 @@ extern my_bool master_ssl; ...@@ -577,8 +546,6 @@ extern my_bool master_ssl;
extern my_string master_ssl_ca, master_ssl_capath, master_ssl_cert, extern my_string master_ssl_ca, master_ssl_capath, master_ssl_cert,
master_ssl_cipher, master_ssl_key; master_ssl_cipher, master_ssl_key;
extern I_List<i_string> replicate_do_db, replicate_ignore_db;
extern I_List<i_string_pair> replicate_rewrite_db;
extern I_List<THD> threads; extern I_List<THD> threads;
#endif #endif
......
...@@ -28,7 +28,8 @@ ...@@ -28,7 +28,8 @@
#include "mysql_priv.h" #include "mysql_priv.h"
#include "hash_filo.h" #include "hash_filo.h"
#ifdef HAVE_REPLICATION #ifdef HAVE_REPLICATION
#include "sql_repl.h" //for tables_ok() #include "table_filter.h" //for tables_ok()
extern Table_filter *rpl_filter;
#endif #endif
#include <m_ctype.h> #include <m_ctype.h>
#include <stdarg.h> #include <stdarg.h>
...@@ -1475,7 +1476,7 @@ static bool update_user_table(THD *thd, const char *host, const char *user, ...@@ -1475,7 +1476,7 @@ static bool update_user_table(THD *thd, const char *host, const char *user,
GRANT and REVOKE are applied the slave in/exclusion rules as they are GRANT and REVOKE are applied the slave in/exclusion rules as they are
some kind of updates to the mysql.% tables. some kind of updates to the mysql.% tables.
*/ */
if (thd->slave_thread && table_rules_on) if (thd->slave_thread && rpl_filter->is_on())
{ {
/* /*
The tables must be marked "updating" so that tables_ok() takes them into The tables must be marked "updating" so that tables_ok() takes them into
...@@ -1483,7 +1484,7 @@ static bool update_user_table(THD *thd, const char *host, const char *user, ...@@ -1483,7 +1484,7 @@ static bool update_user_table(THD *thd, const char *host, const char *user,
*/ */
tables.updating= 1; tables.updating= 1;
/* Thanks to bzero, tables.next==0 */ /* Thanks to bzero, tables.next==0 */
if (!tables_ok(0, &tables)) if (!rpl_filter->tables_ok(0, &tables))
DBUG_RETURN(0); DBUG_RETURN(0);
} }
#endif #endif
...@@ -2669,14 +2670,14 @@ bool mysql_table_grant(THD *thd, TABLE_LIST *table_list, ...@@ -2669,14 +2670,14 @@ bool mysql_table_grant(THD *thd, TABLE_LIST *table_list,
GRANT and REVOKE are applied the slave in/exclusion rules as they are GRANT and REVOKE are applied the slave in/exclusion rules as they are
some kind of updates to the mysql.% tables. some kind of updates to the mysql.% tables.
*/ */
if (thd->slave_thread && table_rules_on) if (thd->slave_thread && rpl_filter->is_on())
{ {
/* /*
The tables must be marked "updating" so that tables_ok() takes them into The tables must be marked "updating" so that tables_ok() takes them into
account in tests. account in tests.
*/ */
tables[0].updating= tables[1].updating= tables[2].updating= 1; tables[0].updating= tables[1].updating= tables[2].updating= 1;
if (!tables_ok(0, tables)) if (!rpl_filter->tables_ok(0, tables))
DBUG_RETURN(FALSE); DBUG_RETURN(FALSE);
} }
#endif #endif
...@@ -2872,14 +2873,14 @@ bool mysql_procedure_grant(THD *thd, TABLE_LIST *table_list, ...@@ -2872,14 +2873,14 @@ bool mysql_procedure_grant(THD *thd, TABLE_LIST *table_list,
GRANT and REVOKE are applied the slave in/exclusion rules as they are GRANT and REVOKE are applied the slave in/exclusion rules as they are
some kind of updates to the mysql.% tables. some kind of updates to the mysql.% tables.
*/ */
if (thd->slave_thread && table_rules_on) if (thd->slave_thread && rpl_filter->is_on())
{ {
/* /*
The tables must be marked "updating" so that tables_ok() takes them into The tables must be marked "updating" so that tables_ok() takes them into
account in tests. account in tests.
*/ */
tables[0].updating= tables[1].updating= 1; tables[0].updating= tables[1].updating= 1;
if (!tables_ok(0, tables)) if (!rpl_filter->tables_ok(0, tables))
DBUG_RETURN(FALSE); DBUG_RETURN(FALSE);
} }
#endif #endif
...@@ -3001,14 +3002,14 @@ bool mysql_grant(THD *thd, const char *db, List <LEX_USER> &list, ...@@ -3001,14 +3002,14 @@ bool mysql_grant(THD *thd, const char *db, List <LEX_USER> &list,
GRANT and REVOKE are applied the slave in/exclusion rules as they are GRANT and REVOKE are applied the slave in/exclusion rules as they are
some kind of updates to the mysql.% tables. some kind of updates to the mysql.% tables.
*/ */
if (thd->slave_thread && table_rules_on) if (thd->slave_thread && rpl_filter->is_on())
{ {
/* /*
The tables must be marked "updating" so that tables_ok() takes them into The tables must be marked "updating" so that tables_ok() takes them into
account in tests. account in tests.
*/ */
tables[0].updating= tables[1].updating= 1; tables[0].updating= tables[1].updating= 1;
if (!tables_ok(0, tables)) if (!rpl_filter->tables_ok(0, tables))
DBUG_RETURN(FALSE); DBUG_RETURN(FALSE);
} }
#endif #endif
...@@ -4192,7 +4193,7 @@ int open_grant_tables(THD *thd, TABLE_LIST *tables) ...@@ -4192,7 +4193,7 @@ int open_grant_tables(THD *thd, TABLE_LIST *tables)
GRANT and REVOKE are applied the slave in/exclusion rules as they are GRANT and REVOKE are applied the slave in/exclusion rules as they are
some kind of updates to the mysql.% tables. some kind of updates to the mysql.% tables.
*/ */
if (thd->slave_thread && table_rules_on) if (thd->slave_thread && rpl_filter->is_on())
{ {
/* /*
The tables must be marked "updating" so that tables_ok() takes them into The tables must be marked "updating" so that tables_ok() takes them into
...@@ -4200,7 +4201,7 @@ int open_grant_tables(THD *thd, TABLE_LIST *tables) ...@@ -4200,7 +4201,7 @@ int open_grant_tables(THD *thd, TABLE_LIST *tables)
*/ */
tables[0].updating=tables[1].updating=tables[2].updating= tables[0].updating=tables[1].updating=tables[2].updating=
tables[3].updating=tables[4].updating=1; tables[3].updating=tables[4].updating=1;
if (!tables_ok(0, tables)) if (!rpl_filter->tables_ok(0, tables))
DBUG_RETURN(1); DBUG_RETURN(1);
tables[0].updating=tables[1].updating=tables[2].updating= tables[0].updating=tables[1].updating=tables[2].updating=
tables[3].updating=tables[4].updating=0;; tables[3].updating=tables[4].updating=0;;
......
...@@ -327,19 +327,20 @@ class LEX_COLUMN : public Sql_alloc ...@@ -327,19 +327,20 @@ class LEX_COLUMN : public Sql_alloc
class i_string: public ilink class i_string: public ilink
{ {
public: public:
char* ptr; const char* ptr;
i_string():ptr(0) { } i_string():ptr(0) { }
i_string(char* s) : ptr(s) {} i_string(const char* s) : ptr(s) {}
}; };
/* needed for linked list of two strings for replicate-rewrite-db */ /* needed for linked list of two strings for replicate-rewrite-db */
class i_string_pair: public ilink class i_string_pair: public ilink
{ {
public: public:
char* key; const char* key;
char* val; const char* val;
i_string_pair():key(0),val(0) { } i_string_pair():key(0),val(0) { }
i_string_pair(char* key_arg, char* val_arg) : key(key_arg),val(val_arg) {} i_string_pair(const char* key_arg, const char* val_arg) :
key(key_arg),val(val_arg) {}
}; };
......
...@@ -16,11 +16,14 @@ ...@@ -16,11 +16,14 @@
#include "mysql_priv.h" #include "mysql_priv.h"
#include "sql_repl.h" #include "sql_repl.h"
#include "table_filter.h"
#include "repl_failsafe.h" #include "repl_failsafe.h"
#include <m_ctype.h> #include <m_ctype.h>
#include <myisam.h> #include <myisam.h>
#include <my_dir.h> #include <my_dir.h>
extern Table_filter *rpl_filter;
#ifdef HAVE_INNOBASE_DB #ifdef HAVE_INNOBASE_DB
#include "ha_innodb.h" #include "ha_innodb.h"
#endif #endif
...@@ -137,10 +140,12 @@ static bool end_active_trans(THD *thd) ...@@ -137,10 +140,12 @@ static bool end_active_trans(THD *thd)
#ifdef HAVE_REPLICATION #ifdef HAVE_REPLICATION
inline bool all_tables_not_ok(THD *thd, TABLE_LIST *tables) inline bool all_tables_not_ok(THD *thd, TABLE_LIST *tables)
{ {
return (table_rules_on && tables && !tables_ok(thd,tables) && return (rpl_filter->is_on() && tables &&
!rpl_filter->tables_ok(thd->db, tables) &&
((thd->lex->sql_command != SQLCOM_DELETE_MULTI) || ((thd->lex->sql_command != SQLCOM_DELETE_MULTI) ||
!tables_ok(thd, !rpl_filter->tables_ok(thd->db,
(TABLE_LIST *)thd->lex->auxilliary_table_list.first))); (TABLE_LIST *)
thd->lex->auxilliary_table_list.first)));
} }
#endif #endif
...@@ -3283,9 +3288,9 @@ mysql_execute_command(THD *thd) ...@@ -3283,9 +3288,9 @@ mysql_execute_command(THD *thd)
above was not called. So we have to check rules again here. above was not called. So we have to check rules again here.
*/ */
#ifdef HAVE_REPLICATION #ifdef HAVE_REPLICATION
if (thd->slave_thread && if (thd->slave_thread &&
(!db_ok(lex->name, replicate_do_db, replicate_ignore_db) || (!rpl_filter->db_ok(lex->name) ||
!db_ok_with_wild_table(lex->name))) !rpl_filter->db_ok_with_wild_table(lex->name)))
{ {
my_message(ER_SLAVE_IGNORED_TABLE, ER(ER_SLAVE_IGNORED_TABLE), MYF(0)); my_message(ER_SLAVE_IGNORED_TABLE, ER(ER_SLAVE_IGNORED_TABLE), MYF(0));
break; break;
...@@ -3314,8 +3319,8 @@ mysql_execute_command(THD *thd) ...@@ -3314,8 +3319,8 @@ mysql_execute_command(THD *thd)
*/ */
#ifdef HAVE_REPLICATION #ifdef HAVE_REPLICATION
if (thd->slave_thread && if (thd->slave_thread &&
(!db_ok(lex->name, replicate_do_db, replicate_ignore_db) || (!rpl_filter->db_ok(lex->name) ||
!db_ok_with_wild_table(lex->name))) !rpl_filter->db_ok_with_wild_table(lex->name)))
{ {
my_message(ER_SLAVE_IGNORED_TABLE, ER(ER_SLAVE_IGNORED_TABLE), MYF(0)); my_message(ER_SLAVE_IGNORED_TABLE, ER(ER_SLAVE_IGNORED_TABLE), MYF(0));
break; break;
...@@ -3355,8 +3360,8 @@ mysql_execute_command(THD *thd) ...@@ -3355,8 +3360,8 @@ mysql_execute_command(THD *thd)
*/ */
#ifdef HAVE_REPLICATION #ifdef HAVE_REPLICATION
if (thd->slave_thread && if (thd->slave_thread &&
(!db_ok(db, replicate_do_db, replicate_ignore_db) || (!rpl_filter->db_ok(lex->name) ||
!db_ok_with_wild_table(db))) !rpl_filter->db_ok_with_wild_table(lex->name)))
{ {
my_message(ER_SLAVE_IGNORED_TABLE, ER(ER_SLAVE_IGNORED_TABLE), MYF(0)); my_message(ER_SLAVE_IGNORED_TABLE, ER(ER_SLAVE_IGNORED_TABLE), MYF(0));
break; break;
......
...@@ -19,8 +19,11 @@ ...@@ -19,8 +19,11 @@
#include "sql_repl.h" #include "sql_repl.h"
#include "log_event.h" #include "log_event.h"
#include "table_filter.h"
#include <my_dir.h> #include <my_dir.h>
extern Table_filter *binlog_filter;
int max_binlog_dump_events = 0; // unlimited int max_binlog_dump_events = 0; // unlimited
my_bool opt_sporadic_binlog_dump_fail = 0; my_bool opt_sporadic_binlog_dump_fail = 0;
static int binlog_dump_count = 0; static int binlog_dump_count = 0;
...@@ -1442,8 +1445,8 @@ bool show_binlog_info(THD* thd) ...@@ -1442,8 +1445,8 @@ bool show_binlog_info(THD* thd)
int dir_len = dirname_length(li.log_file_name); int dir_len = dirname_length(li.log_file_name);
protocol->store(li.log_file_name + dir_len, &my_charset_bin); protocol->store(li.log_file_name + dir_len, &my_charset_bin);
protocol->store((ulonglong) li.pos); protocol->store((ulonglong) li.pos);
protocol->store(&binlog_do_db); protocol->store(binlog_filter->get_do_db());
protocol->store(&binlog_ignore_db); protocol->store(binlog_filter->get_ignore_db());
if (protocol->write()) if (protocol->write())
DBUG_RETURN(TRUE); DBUG_RETURN(TRUE);
} }
......
...@@ -31,7 +31,6 @@ typedef struct st_slave_info ...@@ -31,7 +31,6 @@ typedef struct st_slave_info
extern my_bool opt_show_slave_auth_info; extern my_bool opt_show_slave_auth_info;
extern char *master_host, *master_info_file; extern char *master_host, *master_info_file;
extern bool server_id_supplied; extern bool server_id_supplied;
extern I_List<i_string> binlog_do_db, binlog_ignore_db;
extern int max_binlog_dump_events; extern int max_binlog_dump_events;
extern my_bool opt_sporadic_binlog_dump_fail; extern my_bool opt_sporadic_binlog_dump_fail;
......
This diff is collapsed.
/* Copyright (C) 2000-2003 MySQL AB
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
the Free Software Foundation; either version 2 of the License, or
(at your option) any later version.
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
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA */
#ifndef __TABLE_FILTER_H__
#define __TABLE_FILTER_H__
#include "mysql.h"
#include "my_list.h"
typedef struct st_table_rule_ent
{
char* db;
char* tbl_name;
uint key_len;
} TABLE_RULE_ENT;
/*
Table_filter
Inclusion and exclusion rules of tables and databases.
Also handles rewrites of db.
Used for replication and binlogging.
*/
class Table_filter
{
public:
Table_filter();
~Table_filter();
/* Checks - returns true if ok to replicate/log */
bool tables_ok(const char* db, TABLE_LIST* tables);
bool db_ok(const char* db);
bool db_ok_with_wild_table(const char *db);
bool is_on();
/* Setters - add filtering rules */
int add_do_table(const char* table_spec);
int add_ignore_table(const char* table_spec);
int add_wild_do_table(const char* table_spec);
int add_wild_ignore_table(const char* table_spec);
void add_do_db(const char* db_spec);
void add_ignore_db(const char* db_spec);
void add_db_rewrite(const char* from_db, const char* to_db);
/* Getters - to get information about current rules */
void get_do_table(String* str);
void get_ignore_table(String* str);
void get_wild_do_table(String* str);
void get_wild_ignore_table(String* str);
const char* get_rewrite_db(const char* db, uint32 *new_len);
I_List<i_string>* get_do_db();
I_List<i_string>* get_ignore_db();
private:
bool table_rules_on;
void init_table_rule_hash(HASH* h, bool* h_inited);
void init_table_rule_array(DYNAMIC_ARRAY* a, bool* a_inited);
int add_table_rule(HASH* h, const char* table_spec);
int add_wild_table_rule(DYNAMIC_ARRAY* a, const char* table_spec);
void free_string_array(DYNAMIC_ARRAY *a);
void table_rule_ent_hash_to_str(String* s, HASH* h);
void table_rule_ent_dynamic_array_to_str(String* s, DYNAMIC_ARRAY* a);
TABLE_RULE_ENT* find_wild(DYNAMIC_ARRAY *a, const char* key, int len);
HASH do_table;
HASH ignore_table;
DYNAMIC_ARRAY wild_do_table;
DYNAMIC_ARRAY wild_ignore_table;
bool do_table_inited;
bool ignore_table_inited;
bool wild_do_table_inited;
bool wild_ignore_table_inited;
I_List<i_string> do_db;
I_List<i_string> ignore_db;
I_List<i_string_pair> rewrite_db;
};
#endif // __TABLE_FILTER_H__
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment