MWL#234: Implement option to switch between master-side and client-side...

MWL#234: Implement option to switch between master-side and client-side filtering of @@skip_replication events.

mysql-test/suite/rpl/r/rpl_skip_replication.result

@@ -3,27 +3,27 @@ include/master-slave.inc
 CREATE USER 'nonsuperuser'@'127.0.0.1';
 GRANT ALTER,CREATE,DELETE,DROP,EVENT,INSERT,PROCESS,REPLICATION SLAVE,
 SELECT,UPDATE ON *.* TO 'nonsuperuser'@'127.0.0.1';
-SET GLOBAL replicate_events_marked_for_skip=0;
+SET GLOBAL replicate_events_marked_for_skip=FILTER_ON_MASTER;
 ERROR 42000: Access denied; you need the SUPER privilege for this operation
 DROP USER'nonsuperuser'@'127.0.0.1';
 SELECT @@global.replicate_events_marked_for_skip;
 @@global.replicate_events_marked_for_skip
-1
-SET GLOBAL replicate_events_marked_for_skip=0;
+replicate
+SET GLOBAL replicate_events_marked_for_skip=FILTER_ON_SLAVE;
 ERROR HY000: This operation cannot be performed with a running slave; run STOP SLAVE first
 SELECT @@global.replicate_events_marked_for_skip;
 @@global.replicate_events_marked_for_skip
-1
+replicate
 STOP SLAVE;
-SET SESSION replicate_events_marked_for_skip=0;
+SET SESSION replicate_events_marked_for_skip=FILTER_ON_MASTER;
 ERROR HY000: Variable 'replicate_events_marked_for_skip' is a GLOBAL variable and should be set with SET GLOBAL
 SELECT @@global.replicate_events_marked_for_skip;
 @@global.replicate_events_marked_for_skip
-1
-SET GLOBAL replicate_events_marked_for_skip=0;
+replicate
+SET GLOBAL replicate_events_marked_for_skip=FILTER_ON_MASTER;
 SELECT @@global.replicate_events_marked_for_skip;
 @@global.replicate_events_marked_for_skip
-0
+filter_on_master
 START SLAVE;
 SELECT @@skip_replication;
 @@skip_replication
@@ -55,8 +55,29 @@ a	b
 DROP TABLE t3;
 FLUSH NO_WRITE_TO_BINLOG LOGS;
 STOP SLAVE;
-SET GLOBAL replicate_events_marked_for_skip=1;
+SET GLOBAL replicate_events_marked_for_skip=FILTER_ON_SLAVE;
 START SLAVE;
+SET skip_replication=1;
+CREATE TABLE t3 (a INT PRIMARY KEY, b INT) ENGINE=myisam;
+INSERT INTO t1(a) VALUES (3);
+INSERT INTO t2(a) VALUES (3);
+FLUSH NO_WRITE_TO_BINLOG LOGS;
+SHOW TABLES;
+Tables_in_test
+t1
+t2
+SELECT * FROM t1;
+a	b
+1	NULL
+SELECT * FROM t2;
+a	b
+1	NULL
+DROP TABLE t3;
+FLUSH NO_WRITE_TO_BINLOG LOGS;
+STOP SLAVE;
+SET GLOBAL replicate_events_marked_for_skip=REPLICATE;
+START SLAVE;
+SET skip_replication=1;
 CREATE TABLE t3 (a INT PRIMARY KEY, b INT) ENGINE=myisam;
 INSERT INTO t3(a) VALUES(2);
 SELECT * FROM t3;
@@ -77,7 +98,7 @@ a	b
 2	0
 3	0
 STOP SLAVE;
-SET GLOBAL replicate_events_marked_for_skip=0;
+SET GLOBAL replicate_events_marked_for_skip=FILTER_ON_MASTER;
 TRUNCATE t1;
 SELECT * FROM t1 ORDER by a;
 a	b
@@ -92,7 +113,7 @@ a	b
 TRUNCATE t1;
 STOP SLAVE;
 SET GLOBAL sql_slave_skip_counter=2;
-SET GLOBAL replicate_events_marked_for_skip=0;
+SET GLOBAL replicate_events_marked_for_skip=FILTER_ON_SLAVE;
 START SLAVE;
 SET @old_binlog_format= @@binlog_format;
 SET binlog_format= statement;
@@ -175,9 +196,44 @@ SELECT @@skip_replication;
 DROP FUNCTION foo;
 DROP PROCEDURE bar;
 DROP FUNCTION baz;
+SET skip_replication= 0;
+TRUNCATE t1;
+STOP SLAVE;
+SET GLOBAL replicate_events_marked_for_skip=FILTER_ON_MASTER;
+START SLAVE IO_THREAD;
+SET skip_replication= 1;
+INSERT INTO t1(a) VALUES (1);
+SET skip_replication= 0;
+INSERT INTO t1(a) VALUES (2);
+include/save_master_pos.inc
+include/sync_io_with_master.inc
+STOP SLAVE IO_THREAD;
+SET GLOBAL replicate_events_marked_for_skip=REPLICATE;
+START SLAVE;
+SELECT * FROM t1;
+a	b
+2	NULL
+SET skip_replication= 0;
+TRUNCATE t1;
+STOP SLAVE;
+SET GLOBAL replicate_events_marked_for_skip=FILTER_ON_SLAVE;
+START SLAVE IO_THREAD;
+SET skip_replication= 1;
+INSERT INTO t1(a) VALUES (1);
+SET skip_replication= 0;
+INSERT INTO t1(a) VALUES (2);
+include/save_master_pos.inc
+include/sync_io_with_master.inc
+STOP SLAVE IO_THREAD;
+SET GLOBAL replicate_events_marked_for_skip=REPLICATE;
+START SLAVE;
+SELECT * FROM t1 ORDER BY a;
+a	b
+1	NULL
+2	NULL
 SET skip_replication=0;
 DROP TABLE t1,t2;
 STOP SLAVE;
-SET GLOBAL replicate_events_marked_for_skip=1;
+SET GLOBAL replicate_events_marked_for_skip=REPLICATE;
 START SLAVE;
 include/rpl_end.inc

mysql-test/suite/rpl/t/rpl_skip_replication.test

@@ -9,20 +9,20 @@ GRANT ALTER,CREATE,DELETE,DROP,EVENT,INSERT,PROCESS,REPLICATION SLAVE,
 connect(nonpriv, 127.0.0.1, nonsuperuser,, test, $SLAVE_MYPORT,);
 connection nonpriv;
 --error ER_SPECIFIC_ACCESS_DENIED_ERROR
-SET GLOBAL replicate_events_marked_for_skip=0;
+SET GLOBAL replicate_events_marked_for_skip=FILTER_ON_MASTER;
 disconnect nonpriv;
 connection slave;
 DROP USER'nonsuperuser'@'127.0.0.1';
 
 SELECT @@global.replicate_events_marked_for_skip;
 --error ER_SLAVE_MUST_STOP
-SET GLOBAL replicate_events_marked_for_skip=0;
+SET GLOBAL replicate_events_marked_for_skip=FILTER_ON_SLAVE;
 SELECT @@global.replicate_events_marked_for_skip;
 STOP SLAVE;
 --error ER_GLOBAL_VARIABLE
-SET SESSION replicate_events_marked_for_skip=0;
+SET SESSION replicate_events_marked_for_skip=FILTER_ON_MASTER;
 SELECT @@global.replicate_events_marked_for_skip;
-SET GLOBAL replicate_events_marked_for_skip=0;
+SET GLOBAL replicate_events_marked_for_skip=FILTER_ON_MASTER;
 SELECT @@global.replicate_events_marked_for_skip;
 START SLAVE;
 
@@ -37,6 +37,8 @@ CREATE TABLE t2 (a INT PRIMARY KEY, b INT) ENGINE=innodb;
 INSERT INTO t1(a) VALUES (1);
 INSERT INTO t2(a) VALUES (1);
 
+
+# Test that master-side filtering works.
 SET skip_replication=1;
 
 CREATE TABLE t3 (a INT PRIMARY KEY, b INT) ENGINE=myisam;
@@ -59,12 +61,46 @@ DROP TABLE t3;
 
 FLUSH NO_WRITE_TO_BINLOG LOGS;
 sync_slave_with_master;
+
+
+# Test that slave-side filtering works.
 connection slave;
 STOP SLAVE;
-SET GLOBAL replicate_events_marked_for_skip=1;
+SET GLOBAL replicate_events_marked_for_skip=FILTER_ON_SLAVE;
 START SLAVE;
 
 connection master;
+SET skip_replication=1;
+CREATE TABLE t3 (a INT PRIMARY KEY, b INT) ENGINE=myisam;
+INSERT INTO t1(a) VALUES (3);
+INSERT INTO t2(a) VALUES (3);
+
+# Inject a rotate event in the binlog stream sent to slave (otherwise we will
+# fail sync_slave_with_master as the last event on the master is not present
+# on the slave).
+FLUSH NO_WRITE_TO_BINLOG LOGS;
+
+sync_slave_with_master;
+connection slave;
+SHOW TABLES;
+SELECT * FROM t1;
+SELECT * FROM t2;
+
+connection master;
+DROP TABLE t3;
+
+FLUSH NO_WRITE_TO_BINLOG LOGS;
+sync_slave_with_master;
+connection slave;
+STOP SLAVE;
+SET GLOBAL replicate_events_marked_for_skip=REPLICATE;
+START SLAVE;
+
+
+# Test that events with @@skip_replication=1 are not filtered when filtering is
+# not set on slave.
+connection master;
+SET skip_replication=1;
 CREATE TABLE t3 (a INT PRIMARY KEY, b INT) ENGINE=myisam;
 INSERT INTO t3(a) VALUES(2);
 sync_slave_with_master;
@@ -93,12 +129,12 @@ INSERT INTO t1 VALUES (3,0);
 
 sync_slave_with_master;
 connection slave;
-# Since slave has @@replicate_events_marked_for_skip=1, it should have
+# Since slave has @@replicate_events_marked_for_skip=REPLICATE, it should have
 # applied all events.
 SELECT * FROM t1 ORDER by a;
 
 STOP SLAVE;
-SET GLOBAL replicate_events_marked_for_skip=0;
+SET GLOBAL replicate_events_marked_for_skip=FILTER_ON_MASTER;
 let $SLAVE_DATADIR= `select @@datadir`;
 
 connection master;
@@ -136,7 +172,7 @@ sync_slave_with_master;
 connection slave;
 STOP SLAVE;
 SET GLOBAL sql_slave_skip_counter=2;
-SET GLOBAL replicate_events_marked_for_skip=0;
+SET GLOBAL replicate_events_marked_for_skip=FILTER_ON_SLAVE;
 START SLAVE;
 
 connection master;
@@ -157,7 +193,8 @@ connection slave;
 
 # The slave should have skipped the first three inserts (number 1 and 3 due
 # to @@sql_slave_skip_counter=2, number 2 due to
-# @@replicate_events_marked_for_skip=0). So only number 4 should be left.
+# @@replicate_events_marked_for_skip=FILTER_ON_SLAVE). So only number 4
+# should be left.
 SELECT * FROM t1;
 
 
@@ -242,13 +279,79 @@ DROP FUNCTION foo;
 DROP PROCEDURE bar;
 DROP FUNCTION baz;
 
+
+# Test that master-side filtering happens on the master side, and that
+# slave-side filtering happens on the slave.
+
+# First test that events do not reach the slave when master-side filtering
+# is configured. Do this by replicating first with only the IO thread running
+# and master-side filtering; then change to no filtering and start the SQL
+# thread. This should still skip the events, as master-side filtering
+# means the events never reached the slave.
+connection master;
+SET skip_replication= 0;
+TRUNCATE t1;
+sync_slave_with_master;
+connection slave;
+STOP SLAVE;
+SET GLOBAL replicate_events_marked_for_skip=FILTER_ON_MASTER;
+START SLAVE IO_THREAD;
+connection master;
+SET skip_replication= 1;
+INSERT INTO t1(a) VALUES (1);
+SET skip_replication= 0;
+INSERT INTO t1(a) VALUES (2);
+--source include/save_master_pos.inc
+connection slave;
+--source include/sync_io_with_master.inc
+STOP SLAVE IO_THREAD;
+SET GLOBAL replicate_events_marked_for_skip=REPLICATE;
+START SLAVE;
+connection master;
+sync_slave_with_master;
+connection slave;
+# Now only the second insert of (2) should be visible, as the first was
+# filtered on the master, so even though the SQL thread ran without skipping
+# events, it will never see the event in the first place.
+SELECT * FROM t1;
+
+# Now tests that when slave-side filtering is configured, events _do_ reach
+# the slave.
+connection master;
+SET skip_replication= 0;
+TRUNCATE t1;
+sync_slave_with_master;
+connection slave;
+STOP SLAVE;
+SET GLOBAL replicate_events_marked_for_skip=FILTER_ON_SLAVE;
+START SLAVE IO_THREAD;
+connection master;
+SET skip_replication= 1;
+INSERT INTO t1(a) VALUES (1);
+SET skip_replication= 0;
+INSERT INTO t1(a) VALUES (2);
+--source include/save_master_pos.inc
+connection slave;
+--source include/sync_io_with_master.inc
+STOP SLAVE IO_THREAD;
+SET GLOBAL replicate_events_marked_for_skip=REPLICATE;
+START SLAVE;
+connection master;
+sync_slave_with_master;
+connection slave;
+# Now both inserts should be visible. Since filtering was configured to be
+# slave-side, the event is in the relay log, and when the SQL thread ran we
+# had disabled filtering again.
+SELECT * FROM t1 ORDER BY a;
+
+
 # Clean up.
 connection master;
 SET skip_replication=0;
 DROP TABLE t1,t2;
 connection slave;
 STOP SLAVE;
-SET GLOBAL replicate_events_marked_for_skip=1;
+SET GLOBAL replicate_events_marked_for_skip=REPLICATE;
 START SLAVE;
 
 --source include/rpl_end.inc

sql/log_event.cc

@@ -828,8 +828,8 @@ Log_event::do_shall_skip(Relay_log_info *rli)
                       rli->slave_skip_counter));
   if ((server_id == ::server_id && !rli->replicate_same_server_id) ||
       (rli->slave_skip_counter == 1 && rli->is_in_group()) ||
-      (flags & LOG_EVENT_SKIP_REPLICATION_F
-       && !opt_replicate_events_marked_for_skip))
+      (flags & LOG_EVENT_SKIP_REPLICATION_F &&
+       opt_replicate_events_marked_for_skip != RPL_SKIP_REPLICATE))
     return EVENT_SKIP_IGNORE;
   if (rli->slave_skip_counter > 0)
     return EVENT_SKIP_COUNT;
@@ -3492,7 +3492,7 @@ Query_log_event::do_shall_skip(Relay_log_info *rli)
     number of events to be skipped due to @@sql_slave_skip_counter.
   */
   if (flags & LOG_EVENT_SKIP_REPLICATION_F &&
-      !opt_replicate_events_marked_for_skip)
+      opt_replicate_events_marked_for_skip != RPL_SKIP_REPLICATE)
     DBUG_RETURN(Log_event::EVENT_SKIP_IGNORE);
 
   if (rli->slave_skip_counter > 0)

sql/log_event.h

@@ -495,7 +495,7 @@ struct sql_ex_info
 
    Flag set by application creating the event (with @@skip_replication); the
    slave will skip replication of such events if
-   --replicate-events-marked-for-skip is false.
+   --replicate-events-marked-for-skip is not set to REPLICATE.
 
    This is a MariaDB flag; we allocate it from the end of the available
    values to reduce risk of conflict with new MySQL flags.

sql/mysql_priv.h

@@ -2065,7 +2065,7 @@ extern my_bool opt_old_style_user_limits, trust_function_creators;
 extern uint opt_crash_binlog_innodb;
 extern char *shared_memory_base_name, *mysqld_unix_port;
 extern my_bool opt_enable_shared_memory;
-extern my_bool opt_replicate_events_marked_for_skip;
+extern uint opt_replicate_events_marked_for_skip;
 extern char *default_tz_name;
 #endif /* MYSQL_SERVER */
 #if defined MYSQL_SERVER || defined INNODB_COMPATIBILITY_HOOKS

sql/mysqld.cc

@@ -389,6 +389,16 @@ TYPELIB plugin_maturity_values=
 
 const int server_maturity= MariaDB_PLUGIN_MATURITY_UNKNOWN;
 
+/* These names must match RPL_SKIP_XXX #defines in slave.h. */
+static const char *replicate_events_marked_for_skip_names[]=
+{ "replicate", "filter_on_slave", "filter_on_master", 0 };
+
+TYPELIB replicate_events_marked_for_skip_typelib=
+{
+  array_elements(replicate_events_marked_for_skip_names) - 1, "",
+  replicate_events_marked_for_skip_names, 0
+};
+
 const char *first_keyword= "first", *binary_keyword= "BINARY";
 const char *my_localhost= "localhost", *delayed_user= "DELAYED";
 #if SIZEOF_OFF_T > 4 && defined(BIG_TABLES)
@@ -553,7 +563,7 @@ uint    opt_large_page_size= 0;
 uint    opt_debug_sync_timeout= 0;
 #endif /* defined(ENABLED_DEBUG_SYNC) */
 my_bool opt_old_style_user_limits= 0, trust_function_creators= 0;
-my_bool opt_replicate_events_marked_for_skip;
+uint opt_replicate_events_marked_for_skip;
 
 /*
   True if there is at least one per-hour limit for some user, so we should
@@ -6785,12 +6795,6 @@ each time the SQL thread starts.",
    "cross database updates. If you need cross database updates to work, "
    "make sure you have 3.23.28 or later, and use replicate-wild-ignore-"
    "table=db_name.%. ", 0, 0, 0, GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
-  {"replicate-events-marked-for-skip", OPT_REPLICATE_EVENTS_MARKED_FOR_SKIP,
-   "Tells the slave thread to replicate events that were created with"
-   "@@skip_replication=1. Default true. If set to false, such events will not"
-   "be replicated.", &opt_replicate_events_marked_for_skip,
-   &opt_replicate_events_marked_for_skip, 0, GET_BOOL, NO_ARG,
-   1, 0, 0 ,0, 0, 0},
   {"replicate-ignore-table", OPT_REPLICATE_IGNORE_TABLE,
    "Tells the slave thread to not replicate to the specified table. To specify "
    "more than one table to ignore, use the directive multiple times, once for "
@@ -6807,6 +6811,16 @@ each time the SQL thread starts.",
    "Can't be set to 1 if --log-slave-updates is used.",
    &replicate_same_server_id, &replicate_same_server_id,
    0, GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
+  {"replicate-events-marked-for-skip", OPT_REPLICATE_EVENTS_MARKED_FOR_SKIP,
+   "Whether the slave should replicate events that were created with "
+   "@@skip_replication=1 on the master. Default REPLICATE (no events are "
+   "skipped). Other values are FILTER_ON_SLAVE (events will be sent by the "
+   "master but ignored by the slave) and FILTER_ON_MASTER (events marked with "
+    "@@skip_replication=1 will be filtered on the master and never be sent to "
+   "the slave).", &opt_replicate_events_marked_for_skip,
+   &opt_replicate_events_marked_for_skip,
+   &replicate_events_marked_for_skip_typelib, GET_ENUM, REQUIRED_ARG,
+   RPL_SKIP_REPLICATE, 0, 0 ,0, 0, 0},
 #endif
   {"replicate-wild-do-table", OPT_REPLICATE_WILD_DO_TABLE,
    "Tells the slave thread to restrict replication to the tables that match "

sql/set_var.cc

@@ -925,7 +925,9 @@ static sys_var_thd_set sys_log_slow_verbosity(&vars,
 static sys_var_replicate_events_marked_for_skip
   sys_replicate_events_marked_for_skip(&vars,
                                        "replicate_events_marked_for_skip",
-                                       &opt_replicate_events_marked_for_skip);
+                                       &opt_replicate_events_marked_for_skip,
+                                       &replicate_events_marked_for_skip_typelib,
+                                       NULL);
 #endif
 
 /* Global read-only variable containing hostname */
@@ -4469,7 +4471,7 @@ bool sys_var_replicate_events_marked_for_skip::update(THD *thd, set_var *var)
     result= TRUE;
   }
   else
-    result= sys_var_bool_ptr::update(thd, var);
+    result= sys_var_enum::update(thd, var);
 
   unlock_slave_threads(active_mi);
   pthread_mutex_unlock(&LOCK_active_mi);

sql/set_var.h

@@ -31,7 +31,8 @@ typedef struct system_variables SV;
 typedef struct my_locale_st MY_LOCALE;
 
 extern TYPELIB bool_typelib, delay_key_write_typelib, sql_mode_typelib,
-  optimizer_switch_typelib, slave_exec_mode_typelib;
+  optimizer_switch_typelib, slave_exec_mode_typelib,
+  replicate_events_marked_for_skip_typelib;
 
 typedef int (*sys_check_func)(THD *,  set_var *);
 typedef bool (*sys_update_func)(THD *, set_var *);
@@ -1289,13 +1290,15 @@ public:
   Handler for setting the system variable --replicate-events-marked-for-skip.
 */
 
-class sys_var_replicate_events_marked_for_skip :public sys_var_bool_ptr
+class sys_var_replicate_events_marked_for_skip :public sys_var_enum
 {
 public:
   sys_var_replicate_events_marked_for_skip(sys_var_chain *chain,
                                            const char *name_arg,
-                                           my_bool *value_arg) :
-    sys_var_bool_ptr(chain, name_arg, value_arg) {};
+                                           uint *value_arg,
+                                           TYPELIB *typelib,
+                                           sys_after_update_func func) :
+    sys_var_enum(chain, name_arg, value_arg, typelib, func) {};
   ~sys_var_replicate_events_marked_for_skip() {};
   bool update(THD *thd, set_var *var);
 };

sql/slave.cc

@@ -1178,9 +1178,10 @@ when it try to get the value of TIME_ZONE global variable from master.";
 
   /*
     Request the master to filter away events with the @@skip_replication flag
-    set, if we are running with --replicate-events-marked-for-skip=0.
+    set, if we are running with
+    --replicate-events-marked-for-skip=FILTER_ON_MASTER.
   */
-  if (!opt_replicate_events_marked_for_skip)
+  if (opt_replicate_events_marked_for_skip == RPL_SKIP_FILTER_ON_MASTER)
   {
     if (mysql_real_query(mysql, STRING_WITH_LEN("SET skip_replication=1")))
     {

sql/slave.h

@@ -134,6 +134,15 @@ extern ulonglong relay_log_space_limit;
 */
 #define SLAVE_FORCE_ALL 4
 
+/*
+  Values for the option --replicate-events-marked-for-skip.
+  Must match the typelib replicate_events_marked_for_skip_typelib in mysqld.cc
+*/
+#define RPL_SKIP_REPLICATE 0
+#define RPL_SKIP_FILTER_ON_SLAVE 1
+#define RPL_SKIP_FILTER_ON_MASTER 2
+
+
 int init_slave();
 void init_slave_skip_errors(const char* arg);
 bool flush_relay_log_info(Relay_log_info* rli);