Merge

VERSION

 MYSQL_VERSION_MAJOR=5
 MYSQL_VERSION_MINOR=5
-MYSQL_VERSION_PATCH=20
+MYSQL_VERSION_PATCH=21
 MYSQL_VERSION_EXTRA=

client/mysqlbinlog.cc

@@ -750,6 +750,31 @@ print_use_stmt(PRINT_EVENT_INFO* pinfo, const Query_log_event *ev)
 }
 
 
+/**
+   Print "SET skip_replication=..." statement when needed.
+
+   Not all servers support this (only MariaDB from some version on). So we
+   mark the SET to only execute from the version of MariaDB that supports it,
+   and also only output it if we actually see events with the flag set, to not
+   get spurious errors on MySQL@Oracle servers of higher version that do not
+   support the flag.
+
+   So we start out assuming @@skip_replication is 0, and only output a SET
+   statement when it changes.
+*/
+static void
+print_skip_replication_statement(PRINT_EVENT_INFO *pinfo, const Log_event *ev)
+{
+  int cur_val;
+
+  cur_val= (ev->flags & LOG_EVENT_SKIP_REPLICATION_F) != 0;
+  if (cur_val == pinfo->skip_replication)
+    return;                                     /* Not changed. */
+  fprintf(result_file, "/*!50521 SET skip_replication=%d*/%s\n",
+          cur_val, pinfo->delimiter);
+  pinfo->skip_replication= cur_val;
+}
+
 /**
   Prints the given event in base64 format.
 
@@ -893,7 +918,10 @@ Exit_status process_event(PRINT_EVENT_INFO *print_event_info, Log_event *ev,
           goto end;
       }
       else
+      {
+        print_skip_replication_statement(print_event_info, ev);
         ev->print(result_file, print_event_info);
+      }
       break;
     }
 
@@ -923,7 +951,10 @@ Exit_status process_event(PRINT_EVENT_INFO *print_event_info, Log_event *ev,
           goto end;
       }
       else
+      {
+        print_skip_replication_statement(print_event_info, ev);
         ce->print(result_file, print_event_info, TRUE);
+      }
 
       // If this binlog is not 3.23 ; why this test??
       if (glob_description_event->binlog_version >= 3)
@@ -1027,6 +1058,7 @@ Exit_status process_event(PRINT_EVENT_INFO *print_event_info, Log_event *ev,
         if (fname)
         {
           convert_path_to_forward_slashes(fname);
+          print_skip_replication_statement(print_event_info, ev);
           exlq->print(result_file, print_event_info, fname);
         }
         else
@@ -1156,6 +1188,7 @@ Exit_status process_event(PRINT_EVENT_INFO *print_event_info, Log_event *ev,
       /* FALL THROUGH */
     }
     default:
+      print_skip_replication_statement(print_event_info, ev);
       ev->print(result_file, print_event_info);
     }
   }

mysql-test/include/rpl_udf.inc

@@ -11,6 +11,10 @@
 --source include/have_udf.inc
 --source include/master-slave.inc
 
+disable_query_log;
+call mtr.add_suppression("Unsafe statement written to the binary log using statement format since BINLOG_FORMAT = STATEMENT");
+enable_query_log;
+
 #
 # To run this tests the "sql/udf_example.c" need to be compiled into
 # udf_example.so and LD_LIBRARY_PATH should be setup to point out where

mysql-test/r/mysqld--help-win.result.THIS

@@ -620,6 +620,14 @@ The following options may be given as the first argument:
  directive multiple times, once for each table. This will
  work for cross-database updates, in contrast to
  replicate-do-db.
+ --replicate-events-marked-for-skip=name 
+ 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).
  --replicate-ignore-db=name 
  Tells the slave thread to not replicate to the specified
  database. To specify more than one database to ignore,
@@ -1029,6 +1037,7 @@ relay-log-purge TRUE
 relay-log-recovery FALSE
 relay-log-space-limit 0
 replicate-annotate-row-events FALSE
+replicate-events-marked-for-skip replicate
 replicate-same-server-id FALSE
 report-host (No default value)
 report-password (No default value)

mysql-test/r/mysqld--help.result

@@ -620,6 +620,14 @@ The following options may be given as the first argument:
  directive multiple times, once for each table. This will
  work for cross-database updates, in contrast to
  replicate-do-db.
+ --replicate-events-marked-for-skip=name 
+ 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).
  --replicate-ignore-db=name 
  Tells the slave thread to not replicate to the specified
  database. To specify more than one database to ignore,
@@ -1021,6 +1029,7 @@ relay-log-purge TRUE
 relay-log-recovery FALSE
 relay-log-space-limit 0
 replicate-annotate-row-events FALSE
+replicate-events-marked-for-skip replicate
 replicate-same-server-id FALSE
 report-host (No default value)
 report-password (No default value)

mysql-test/suite/innodb/t/mdev-117.test

@@ -18,11 +18,11 @@ SELECT col_int_key FROM t1;
 --send DELETE IGNORE FROM t1;
 
 --connection con1
---error 0,1213
+--error 0,ER_LOCK_DEADLOCK,ER_LOCK_WAIT_TIMEOUT
 DELETE FROM t1 WHERE col_int_key IN (1, 40000000);
 
 --connection default
---error 0,1213
+--error 0,ER_LOCK_DEADLOCK,ER_LOCK_WAIT_TIMEOUT
 --reap 
 
 --disconnect con1

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

+include/master-slave.inc
+[connection master]
+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=FILTER_ON_MASTER;
+ERROR 42000: Access denied; you need (at least one of) the SUPER privilege(s) for this operation
+DROP USER'nonsuperuser'@'127.0.0.1';
+SELECT @@global.replicate_events_marked_for_skip;
+@@global.replicate_events_marked_for_skip
+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
+replicate
+STOP SLAVE;
+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
+replicate
+SET GLOBAL replicate_events_marked_for_skip=FILTER_ON_MASTER;
+SELECT @@global.replicate_events_marked_for_skip;
+@@global.replicate_events_marked_for_skip
+filter_on_master
+START SLAVE;
+SELECT @@skip_replication;
+@@skip_replication
+0
+SET GLOBAL skip_replication=1;
+ERROR HY000: Variable 'skip_replication' is a SESSION variable and can't be used with SET GLOBAL
+SELECT @@skip_replication;
+@@skip_replication
+0
+CREATE TABLE t1 (a INT PRIMARY KEY, b INT) ENGINE=myisam;
+CREATE TABLE t2 (a INT PRIMARY KEY, b INT) ENGINE=innodb;
+INSERT INTO t1(a) VALUES (1);
+INSERT INTO t2(a) VALUES (1);
+SET skip_replication=1;
+CREATE TABLE t3 (a INT PRIMARY KEY, b INT) ENGINE=myisam;
+INSERT INTO t1(a) VALUES (2);
+INSERT INTO t2(a) VALUES (2);
+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=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;
+a	b
+2	NULL
+DROP TABLE t3;
+TRUNCATE t1;
+RESET MASTER;
+SET skip_replication=0;
+INSERT INTO t1 VALUES (1,0);
+SET skip_replication=1;
+INSERT INTO t1 VALUES (2,0);
+SET skip_replication=0;
+INSERT INTO t1 VALUES (3,0);
+SELECT * FROM t1 ORDER by a;
+a	b
+1	0
+2	0
+3	0
+STOP SLAVE;
+SET GLOBAL replicate_events_marked_for_skip=FILTER_ON_MASTER;
+TRUNCATE t1;
+SELECT * FROM t1 ORDER by a;
+a	b
+1	0
+2	0
+3	0
+START SLAVE;
+SELECT * FROM t1 ORDER by a;
+a	b
+1	0
+3	0
+TRUNCATE t1;
+STOP SLAVE;
+SET GLOBAL sql_slave_skip_counter=6;
+SET GLOBAL replicate_events_marked_for_skip=FILTER_ON_SLAVE;
+START SLAVE;
+SET @old_binlog_format= @@binlog_format;
+SET binlog_format= statement;
+SET skip_replication=0;
+INSERT INTO t1 VALUES (1,5);
+SET skip_replication=1;
+INSERT INTO t1 VALUES (2,5);
+SET skip_replication=0;
+INSERT INTO t1 VALUES (3,5);
+INSERT INTO t1 VALUES (4,5);
+SET binlog_format= @old_binlog_format;
+SELECT * FROM t1;
+a	b
+4	5
+include/stop_slave.inc
+SET @old_slave_binlog_format= @@global.binlog_format;
+SET GLOBAL binlog_format= row;
+include/start_slave.inc
+TRUNCATE t1;
+SET @old_binlog_format= @@binlog_format;
+SET binlog_format= row;
+BINLOG 'wlZOTw8BAAAA8QAAAPUAAAAAAAQANS41LjIxLU1hcmlhREItZGVidWctbG9nAAAAAAAAAAAAAAAA
+AAAAAAAAAAAAAAAAAAAAAAAAEzgNAAgAEgAEBAQEEgAA2QAEGggAAAAICAgCAAAAAAAAAAAAAAAA
+AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA
+AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA
+AAAAAAAAAAAA371saA==';
+BINLOG 'wlZOTxMBAAAAKgAAAGMBAAAAgCkAAAAAAAEABHRlc3QAAnQxAAIDAwAC
+wlZOTxcBAAAAJgAAAIkBAAAAgCkAAAAAAAEAAv/8AQAAAAgAAAA=';
+BINLOG 'wlZOTxMBAAAAKgAAADwCAAAAACkAAAAAAAEABHRlc3QAAnQxAAIDAwAC
+wlZOTxcBAAAAJgAAAGICAAAAACkAAAAAAAEAAv/8AgAAAAgAAAA=';
+SET binlog_format= @old_binlog_format;
+SELECT * FROM t1 ORDER BY a;
+a	b
+1	8
+2	8
+SELECT * FROM t1 ORDER by a;
+a	b
+2	8
+include/stop_slave.inc
+SET GLOBAL binlog_format= @old_slave_binlog_format;
+include/start_slave.inc
+SET skip_replication=0;
+BEGIN;
+SET skip_replication=0;
+ERROR HY000: Cannot modify @@session.skip_replication inside a transaction
+SET skip_replication=1;
+ERROR HY000: Cannot modify @@session.skip_replication inside a transaction
+ROLLBACK;
+SET skip_replication=1;
+BEGIN;
+SET skip_replication=0;
+ERROR HY000: Cannot modify @@session.skip_replication inside a transaction
+SET skip_replication=1;
+ERROR HY000: Cannot modify @@session.skip_replication inside a transaction
+COMMIT;
+SET autocommit=0;
+INSERT INTO t2(a) VALUES(100);
+SET skip_replication=1;
+ERROR HY000: Cannot modify @@session.skip_replication inside a transaction
+ROLLBACK;
+SET autocommit=1;
+SET skip_replication=1;
+CREATE FUNCTION foo (x INT) RETURNS INT BEGIN SET SESSION skip_replication=x; RETURN x; END|
+CREATE PROCEDURE bar(x INT) BEGIN SET SESSION skip_replication=x; END|
+CREATE FUNCTION baz (x INT) RETURNS INT BEGIN CALL bar(x); RETURN x; END|
+SELECT foo(0);
+ERROR HY000: Cannot modify @@session.skip_replication inside a stored function or trigger
+SELECT baz(0);
+ERROR HY000: Cannot modify @@session.skip_replication inside a stored function or trigger
+SET @a= foo(1);
+ERROR HY000: Cannot modify @@session.skip_replication inside a stored function or trigger
+SET @a= baz(1);
+ERROR HY000: Cannot modify @@session.skip_replication inside a stored function or trigger
+UPDATE t2 SET b=foo(0);
+ERROR HY000: Cannot modify @@session.skip_replication inside a stored function or trigger
+UPDATE t2 SET b=baz(0);
+ERROR HY000: Cannot modify @@session.skip_replication inside a stored function or trigger
+INSERT INTO t1 VALUES (101, foo(1));
+ERROR HY000: Cannot modify @@session.skip_replication inside a stored function or trigger
+INSERT INTO t1 VALUES (101, baz(0));
+ERROR HY000: Cannot modify @@session.skip_replication inside a stored function or trigger
+SELECT @@skip_replication;
+@@skip_replication
+1
+CALL bar(0);
+SELECT @@skip_replication;
+@@skip_replication
+0
+CALL bar(1);
+SELECT @@skip_replication;
+@@skip_replication
+1
+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=REPLICATE;
+START SLAVE;
+include/rpl_end.inc

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

+--source include/master-slave.inc
+--source include/have_innodb.inc
+
+connection slave;
+# Test that SUPER is required to change @@replicate_events_marked_for_skip.
+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';
+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=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=FILTER_ON_SLAVE;
+SELECT @@global.replicate_events_marked_for_skip;
+STOP SLAVE;
+--error ER_GLOBAL_VARIABLE
+SET SESSION replicate_events_marked_for_skip=FILTER_ON_MASTER;
+SELECT @@global.replicate_events_marked_for_skip;
+SET GLOBAL replicate_events_marked_for_skip=FILTER_ON_MASTER;
+SELECT @@global.replicate_events_marked_for_skip;
+START SLAVE;
+
+connection master;
+SELECT @@skip_replication;
+--error ER_LOCAL_VARIABLE
+SET GLOBAL skip_replication=1;
+SELECT @@skip_replication;
+
+CREATE TABLE t1 (a INT PRIMARY KEY, b INT) ENGINE=myisam;
+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;
+INSERT INTO t1(a) VALUES (2);
+INSERT INTO t2(a) VALUES (2);
+
+# 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;
+
+
+# Test that slave-side filtering works.
+connection slave;
+STOP SLAVE;
+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;
+connection slave;
+SELECT * FROM t3;
+connection master;
+DROP TABLE t3;
+
+#
+# Test that the slave will preserve the @@skip_replication flag in its
+# own binlog.
+#
+
+TRUNCATE t1;
+sync_slave_with_master;
+connection slave;
+RESET MASTER;
+
+connection master;
+SET skip_replication=0;
+INSERT INTO t1 VALUES (1,0);
+SET skip_replication=1;
+INSERT INTO t1 VALUES (2,0);
+SET skip_replication=0;
+INSERT INTO t1 VALUES (3,0);
+
+sync_slave_with_master;
+connection slave;
+# 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=FILTER_ON_MASTER;
+let $SLAVE_DATADIR= `select @@datadir`;
+
+connection master;
+TRUNCATE t1;
+
+# Now apply the slave binlog to the master, to check that both the slave
+# and mysqlbinlog will preserve the @@skip_replication flag.
+--exec $MYSQL_BINLOG $SLAVE_DATADIR/slave-bin.000001 > $MYSQLTEST_VARDIR/tmp/rpl_skip_replication.binlog
+--exec $MYSQL test < $MYSQLTEST_VARDIR/tmp/rpl_skip_replication.binlog
+
+# The master should have all three events.
+SELECT * FROM t1 ORDER by a;
+
+# The slave should be missing event 2, which is marked with the
+# @@skip_replication flag.
+
+connection slave;
+START SLAVE;
+
+connection master;
+sync_slave_with_master;
+
+connection slave;
+SELECT * FROM t1 ORDER by a;
+
+#
+# Test that @@sql_slave_skip_counter does not count skipped @@skip_replication
+# events.
+#
+
+connection master;
+TRUNCATE t1;
+
+sync_slave_with_master;
+connection slave;
+STOP SLAVE;
+# We will skip two INSERTs (in addition to any skipped due to
+# @@skip_replication). Since from 5.5 every statement is wrapped in
+# BEGIN ... END, we need to skip 6 events for this.
+SET GLOBAL sql_slave_skip_counter=6;
+SET GLOBAL replicate_events_marked_for_skip=FILTER_ON_SLAVE;
+START SLAVE;
+
+connection master;
+# Need to fix @@binlog_format to get consistent event count.
+SET @old_binlog_format= @@binlog_format;
+SET binlog_format= statement;
+SET skip_replication=0;
+INSERT INTO t1 VALUES (1,5);
+SET skip_replication=1;
+INSERT INTO t1 VALUES (2,5);
+SET skip_replication=0;
+INSERT INTO t1 VALUES (3,5);
+INSERT INTO t1 VALUES (4,5);
+SET binlog_format= @old_binlog_format;
+
+sync_slave_with_master;
+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=FILTER_ON_SLAVE). So only number 4
+# should be left.
+SELECT * FROM t1;
+
+
+#
+# Check that BINLOG statement preserves the @@skip_replication flag.
+#
+connection slave;
+# Need row @@binlog_format for BINLOG statements containing row events.
+--source include/stop_slave.inc
+SET @old_slave_binlog_format= @@global.binlog_format;
+SET GLOBAL binlog_format= row;
+--source include/start_slave.inc
+
+connection master;
+TRUNCATE t1;
+
+SET @old_binlog_format= @@binlog_format;
+SET binlog_format= row;
+# Format description log event.
+BINLOG 'wlZOTw8BAAAA8QAAAPUAAAAAAAQANS41LjIxLU1hcmlhREItZGVidWctbG9nAAAAAAAAAAAAAAAA
+AAAAAAAAAAAAAAAAAAAAAAAAEzgNAAgAEgAEBAQEEgAA2QAEGggAAAAICAgCAAAAAAAAAAAAAAAA
+AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA
+AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA
+AAAAAAAAAAAA371saA==';
+# INSERT INTO t1 VALUES (1,8)  # with @@skip_replication=1
+BINLOG 'wlZOTxMBAAAAKgAAAGMBAAAAgCkAAAAAAAEABHRlc3QAAnQxAAIDAwAC
+wlZOTxcBAAAAJgAAAIkBAAAAgCkAAAAAAAEAAv/8AQAAAAgAAAA=';
+# INSERT INTO t1 VALUES (2,8)  # with @@skip_replication=0
+BINLOG 'wlZOTxMBAAAAKgAAADwCAAAAACkAAAAAAAEABHRlc3QAAnQxAAIDAwAC
+wlZOTxcBAAAAJgAAAGICAAAAACkAAAAAAAEAAv/8AgAAAAgAAAA=';
+SET binlog_format= @old_binlog_format;
+
+SELECT * FROM t1 ORDER BY a;
+sync_slave_with_master;
+connection slave;
+# Slave should have only the second insert, the first should be ignored due to
+# the @@skip_replication flag.
+SELECT * FROM t1 ORDER by a;
+
+--source include/stop_slave.inc
+SET GLOBAL binlog_format= @old_slave_binlog_format;
+--source include/start_slave.inc
+
+
+# Test that it is not possible to change @@skip_replication inside a
+# transaction or statement, thereby replicating only parts of statements
+# or transactions.
+connection master;
+SET skip_replication=0;
+
+BEGIN;
+--error ER_INSIDE_TRANSACTION_PREVENTS_SWITCH_SKIP_REPLICATION
+SET skip_replication=0;
+--error ER_INSIDE_TRANSACTION_PREVENTS_SWITCH_SKIP_REPLICATION
+SET skip_replication=1;
+ROLLBACK;
+SET skip_replication=1;
+BEGIN;
+--error ER_INSIDE_TRANSACTION_PREVENTS_SWITCH_SKIP_REPLICATION
+SET skip_replication=0;
+--error ER_INSIDE_TRANSACTION_PREVENTS_SWITCH_SKIP_REPLICATION
+SET skip_replication=1;
+COMMIT;
+SET autocommit=0;
+INSERT INTO t2(a) VALUES(100);
+--error ER_INSIDE_TRANSACTION_PREVENTS_SWITCH_SKIP_REPLICATION
+SET skip_replication=1;
+ROLLBACK;
+SET autocommit=1;
+
+SET skip_replication=1;
+--delimiter |
+CREATE FUNCTION foo (x INT) RETURNS INT BEGIN SET SESSION skip_replication=x; RETURN x; END|
+CREATE PROCEDURE bar(x INT) BEGIN SET SESSION skip_replication=x; END|
+CREATE FUNCTION baz (x INT) RETURNS INT BEGIN CALL bar(x); RETURN x; END|
+--delimiter ;
+--error ER_STORED_FUNCTION_PREVENTS_SWITCH_SKIP_REPLICATION
+SELECT foo(0);
+--error ER_STORED_FUNCTION_PREVENTS_SWITCH_SKIP_REPLICATION
+SELECT baz(0);
+--error ER_STORED_FUNCTION_PREVENTS_SWITCH_SKIP_REPLICATION
+SET @a= foo(1);
+--error ER_STORED_FUNCTION_PREVENTS_SWITCH_SKIP_REPLICATION
+SET @a= baz(1);
+--error ER_STORED_FUNCTION_PREVENTS_SWITCH_SKIP_REPLICATION
+UPDATE t2 SET b=foo(0);
+--error ER_STORED_FUNCTION_PREVENTS_SWITCH_SKIP_REPLICATION
+UPDATE t2 SET b=baz(0);
+--error ER_STORED_FUNCTION_PREVENTS_SWITCH_SKIP_REPLICATION
+INSERT INTO t1 VALUES (101, foo(1));
+--error ER_STORED_FUNCTION_PREVENTS_SWITCH_SKIP_REPLICATION
+INSERT INTO t1 VALUES (101, baz(0));
+SELECT @@skip_replication;
+CALL bar(0);
+SELECT @@skip_replication;
+CALL bar(1);
+SELECT @@skip_replication;
+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=REPLICATE;
+START SLAVE;
+
+--source include/rpl_end.inc

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

@@ -5,10 +5,6 @@
 # statement based format. This tests work completed in WL#3629.   #
 ###################################################################
 
-disable_query_log;
-call mtr.add_suppression("Unsafe statement written to the binary log using statement format since BINLOG_FORMAT = STATEMENT");
-enable_query_log;
-
 let $engine_type= MyISAM;
 --source include/rpl_udf.inc
 

mysql-test/suite/sys_vars/r/replicate_events_marked_for_skip_basic.result

+#
+# Basic testing of replicate_events_marked_for_skip.
+#
+SET @save_replicate_events_marked_for_skip = @@GLOBAL.replicate_events_marked_for_skip;
+SELECT @save_replicate_events_marked_for_skip;
+@save_replicate_events_marked_for_skip
+replicate
+# Scope.
+SET @@SESSION.replicate_events_marked_for_skip = "";
+ERROR HY000: Variable 'replicate_events_marked_for_skip' is a GLOBAL variable and should be set with SET GLOBAL
+SELECT @@SESSION.replicate_events_marked_for_skip;
+ERROR HY000: Variable 'replicate_events_marked_for_skip' is a GLOBAL variable
+# Argument syntax.
+SET @@GLOBAL.replicate_events_marked_for_skip=filter_on_master;
+SELECT @@GLOBAL.replicate_events_marked_for_skip;
+@@GLOBAL.replicate_events_marked_for_skip
+filter_on_master
+SET @@GLOBAL.replicate_events_marked_for_skip=filter_on_slave;
+SELECT @@GLOBAL.replicate_events_marked_for_skip;
+@@GLOBAL.replicate_events_marked_for_skip
+filter_on_slave
+SET @@GLOBAL.replicate_events_marked_for_skip=replicate;
+SELECT @@GLOBAL.replicate_events_marked_for_skip;
+@@GLOBAL.replicate_events_marked_for_skip
+replicate
+SET @@GLOBAL.replicate_events_marked_for_skip=filter;
+ERROR 42000: Variable 'replicate_events_marked_for_skip' can't be set to the value of 'filter'
+SELECT @@GLOBAL.replicate_events_marked_for_skip;
+@@GLOBAL.replicate_events_marked_for_skip
+replicate
+SELECT * FROM INFORMATION_SCHEMA.GLOBAL_VARIABLES WHERE VARIABLE_NAME='replicate_events_marked_for_skip';
+VARIABLE_NAME	VARIABLE_VALUE
+REPLICATE_EVENTS_MARKED_FOR_SKIP	replicate
+# Cleanup.
+SET @@GLOBAL.replicate_events_marked_for_skip = @save_replicate_events_marked_for_skip;

mysql-test/suite/sys_vars/r/skip_replication_basic.result

+select @@global.skip_replication;
+ERROR HY000: Variable 'skip_replication' is a SESSION variable
+select @@session.skip_replication between 1 and 10000;
+@@session.skip_replication between 1 and 10000
+0
+should be empty
+show global variables like 'skip_replication';
+Variable_name	Value
+show session variables like 'skip_replication';
+Variable_name	Value
+skip_replication	OFF
+should be empty
+select * from information_schema.global_variables where variable_name='skip_replication';
+VARIABLE_NAME	VARIABLE_VALUE
+select @@session.skip_replication = variable_value from information_schema.session_variables where variable_name='skip_replication';
+@@session.skip_replication = variable_value
+1
+Warnings:
+Warning	1292	Truncated incorrect DOUBLE value: 'OFF'
+set session skip_replication=0;
+select @@session.skip_replication;
+@@session.skip_replication
+0
+set session skip_replication=1;
+select @@session.skip_replication;
+@@session.skip_replication
+1
+select * from information_schema.global_variables where variable_name='skip_replication';
+VARIABLE_NAME	VARIABLE_VALUE
+select variable_value from information_schema.session_variables where variable_name='skip_replication';
+variable_value
+ON
+set global skip_replication=1;
+ERROR HY000: Variable 'skip_replication' is a SESSION variable and can't be used with SET GLOBAL

mysql-test/suite/sys_vars/t/replicate_events_marked_for_skip_basic.test

+source include/not_embedded.inc;
+
+--echo #
+--echo # Basic testing of replicate_events_marked_for_skip.
+--echo #
+
+SET @save_replicate_events_marked_for_skip = @@GLOBAL.replicate_events_marked_for_skip;
+SELECT @save_replicate_events_marked_for_skip;
+
+--echo # Scope.
+
+--error ER_GLOBAL_VARIABLE
+SET @@SESSION.replicate_events_marked_for_skip = "";
+--error ER_INCORRECT_GLOBAL_LOCAL_VAR
+SELECT @@SESSION.replicate_events_marked_for_skip;
+
+--echo # Argument syntax.
+
+SET @@GLOBAL.replicate_events_marked_for_skip=filter_on_master;
+SELECT @@GLOBAL.replicate_events_marked_for_skip;
+SET @@GLOBAL.replicate_events_marked_for_skip=filter_on_slave;
+SELECT @@GLOBAL.replicate_events_marked_for_skip;
+SET @@GLOBAL.replicate_events_marked_for_skip=replicate;
+SELECT @@GLOBAL.replicate_events_marked_for_skip;
+--error ER_WRONG_VALUE_FOR_VAR
+SET @@GLOBAL.replicate_events_marked_for_skip=filter;
+SELECT @@GLOBAL.replicate_events_marked_for_skip;
+SELECT * FROM INFORMATION_SCHEMA.GLOBAL_VARIABLES WHERE VARIABLE_NAME='replicate_events_marked_for_skip';
+
+--echo # Cleanup.
+SET @@GLOBAL.replicate_events_marked_for_skip = @save_replicate_events_marked_for_skip;

mysql-test/suite/sys_vars/t/skip_replication_basic.test

+# exists as a session only
+ 
+--error ER_INCORRECT_GLOBAL_LOCAL_VAR
+select @@global.skip_replication;
+
+# Check the variable has a valid numeric value (assumed to be less then 10000)
+select @@session.skip_replication between 1 and 10000;
+
+--echo should be empty
+show global variables like 'skip_replication';
+show session variables like 'skip_replication';
+
+# Global I_S variable is empty
+--echo should be empty
+select * from information_schema.global_variables where variable_name='skip_replication';
+
+# Check that I_S value is same as variable
+select @@session.skip_replication = variable_value from information_schema.session_variables where variable_name='skip_replication';
+
+#
+# show that it's writable
+#
+set session skip_replication=0;
+select @@session.skip_replication;
+set session skip_replication=1;
+select @@session.skip_replication;
+select * from information_schema.global_variables where variable_name='skip_replication';
+select variable_value from information_schema.session_variables where variable_name='skip_replication';
+--error ER_LOCAL_VARIABLE
+set global skip_replication=1;

sql/log_event.cc

@@ -729,7 +729,7 @@ const char* Log_event::get_type_str()
 
 #ifndef MYSQL_CLIENT
 Log_event::Log_event(THD* thd_arg, uint16 flags_arg, bool using_trans)
-  :log_pos(0), temp_buf(0), exec_time(0), flags(flags_arg),
+  :log_pos(0), temp_buf(0), exec_time(0),
    crc(0), thd(thd_arg),
    checksum_alg(BINLOG_CHECKSUM_ALG_UNDEF)
 {
@@ -741,6 +741,9 @@ Log_event::Log_event(THD* thd_arg, uint16 flags_arg, bool using_trans)
     cache_type= Log_event::EVENT_TRANSACTIONAL_CACHE;
   else
     cache_type= Log_event::EVENT_STMT_CACHE;
+  flags= flags_arg |
+    (thd->variables.option_bits & OPTION_SKIP_REPLICATION ?
+     LOG_EVENT_SKIP_REPLICATION_F : 0);
 }
 
 /**
@@ -891,7 +894,9 @@ Log_event::do_shall_skip(Relay_log_info *rli)
                       rli->replicate_same_server_id,
                       rli->slave_skip_counter));
   if ((server_id == ::server_id && !rli->replicate_same_server_id) ||
-      (rli->slave_skip_counter == 1 && rli->is_in_group()))
+      (rli->slave_skip_counter == 1 && rli->is_in_group()) ||
+      (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;
@@ -3901,6 +3906,14 @@ Query_log_event::do_shall_skip(Relay_log_info *rli)
   DBUG_PRINT("debug", ("query: %s; q_len: %d", query, q_len));
   DBUG_ASSERT(query && q_len > 0);
 
+  /*
+    An event skipped due to @@skip_replication must not be counted towards the
+    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 != RPL_SKIP_REPLICATE)
+    DBUG_RETURN(Log_event::EVENT_SKIP_IGNORE);
+
   if (rli->slave_skip_counter > 0)
   {
     if (strcmp("BEGIN", query) == 0)
@@ -10806,7 +10819,7 @@ st_print_event_info::st_print_event_info()
    auto_increment_increment(0),auto_increment_offset(0), charset_inited(0),
    lc_time_names_number(~0),
    charset_database_number(ILLEGAL_CHARSET_INFO_NUMBER),
-   thread_id(0), thread_id_printed(false),
+   thread_id(0), thread_id_printed(false), skip_replication(0),
    base64_output_mode(BASE64_OUTPUT_UNSPEC), printed_fd_event(FALSE)
 {
   /*

sql/log_event.h

@@ -504,6 +504,19 @@ struct sql_ex_info
 */
 #define LOG_EVENT_RELAY_LOG_F 0x40
 
+/**
+   @def LOG_EVENT_SKIP_REPLICATION_F
+
+   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 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.
+*/
+#define LOG_EVENT_SKIP_REPLICATION_F 0x8000
+
+
 /**
   @def OPTIONS_WRITTEN_TO_BIN_LOG
 
@@ -701,6 +714,11 @@ typedef struct st_print_event_info
   uint charset_database_number;
   uint thread_id;
   bool thread_id_printed;
+  /*
+    Track when @@skip_replication changes so we need to output a SET
+    statement for it.
+  */
+  int skip_replication;
 
   st_print_event_info();
 
@@ -993,8 +1011,8 @@ public:
 
   /**
     Some 16 flags. See the definitions above for LOG_EVENT_TIME_F,
-    LOG_EVENT_FORCED_ROTATE_F, LOG_EVENT_THREAD_SPECIFIC_F, and
-    LOG_EVENT_SUPPRESS_USE_F for notes.
+    LOG_EVENT_FORCED_ROTATE_F, LOG_EVENT_THREAD_SPECIFIC_F,
+    LOG_EVENT_SUPPRESS_USE_F, and LOG_EVENT_SKIP_REPLICATION_F for notes.
   */
   uint16 flags;
 
@@ -4143,6 +4161,8 @@ public:
     m_message.str= NULL;                    /* Just as a precaution */
     m_message.length= 0;
     set_direct_logging();
+    /* Replicate the incident irregardless of @@skip_replication. */
+    flags&= ~LOG_EVENT_SKIP_REPLICATION_F;
     DBUG_VOID_RETURN;
   }
 
@@ -4153,6 +4173,8 @@ public:
     DBUG_PRINT("enter", ("m_incident: %d", m_incident));
     m_message= msg;
     set_direct_logging();
+    /* Replicate the incident irregardless of @@skip_replication. */
+    flags&= ~LOG_EVENT_SKIP_REPLICATION_F;
     DBUG_VOID_RETURN;
   }
 #endif

sql/mysqld.cc

@@ -437,6 +437,8 @@ uint   opt_large_page_size= 0;
 MYSQL_PLUGIN_IMPORT uint    opt_debug_sync_timeout= 0;
 #endif /* defined(ENABLED_DEBUG_SYNC) */
 my_bool opt_old_style_user_limits= 0, trust_function_creators= 0;
+ulong opt_replicate_events_marked_for_skip;
+
 /*
   True if there is at least one per-hour limit for some user, so we should
   check them before each query (and possibly reset counters when hour is

sql/mysqld.h

@@ -109,6 +109,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 ulong opt_replicate_events_marked_for_skip;
 extern char *default_tz_name;
 extern Time_zone *default_tz;
 extern char *default_storage_engine;

sql/set_var.h

@@ -171,6 +171,7 @@ protected:
 
 #include "sql_plugin.h"                    /* SHOW_HA_ROWS, SHOW_MY_BOOL */
 
+
 /****************************************************************************
   Classes for parsing of the SET command
 ****************************************************************************/

sql/share/errmsg-utf8.txt

@@ -6557,4 +6557,7 @@ ER_CONNECTION_KILLED 70100
         eng "Connection was killed"
 ER_INTERNAL_ERROR
         eng "Internal error: '%-.192s'"
-
+ER_INSIDE_TRANSACTION_PREVENTS_SWITCH_SKIP_REPLICATION
+        eng "Cannot modify @@session.skip_replication inside a transaction"
+ER_STORED_FUNCTION_PREVENTS_SWITCH_SKIP_REPLICATION
+        eng "Cannot modify @@session.skip_replication inside a stored function or trigger"

sql/slave.cc

@@ -1710,6 +1710,48 @@ when it try to get the value of TIME_ZONE global variable from master.";
 past_checksum:
 #endif
 
+  /*
+    Request the master to filter away events with the @@skip_replication flag
+    set, if we are running with
+    --replicate-events-marked-for-skip=FILTER_ON_MASTER.
+  */
+  if (opt_replicate_events_marked_for_skip == RPL_SKIP_FILTER_ON_MASTER)
+  {
+    if (mysql_real_query(mysql, STRING_WITH_LEN("SET skip_replication=1")))
+    {
+      err_code= mysql_errno(mysql);
+      if (is_network_error(err_code))
+      {
+        mi->report(ERROR_LEVEL, err_code,
+                   "Setting master-side filtering of @@skip_replication failed "
+                   "with error: %s", mysql_error(mysql));
+        goto network_err;
+      }
+      else if (err_code == ER_UNKNOWN_SYSTEM_VARIABLE)
+      {
+        /*
+          The master is older than the slave and does not support the
+          @@skip_replication feature.
+          This is not a problem, as such master will not generate events with
+          the @@skip_replication flag set in the first place. We will still
+          do slave-side filtering of such events though, to handle the (rare)
+          case of downgrading a master and receiving old events generated from
+          before the downgrade with the @@skip_replication flag set.
+        */
+        DBUG_PRINT("info", ("Old master does not support master-side filtering "
+                            "of @@skip_replication events."));
+      }
+      else
+      {
+        /* Fatal error */
+        errmsg= "The slave I/O thread stops because a fatal error is "
+          "encountered when it tries to request filtering of events marked "
+          "with the @@skip_replication flag.";
+        sprintf(err_buff, "%s Error: %s", errmsg, mysql_error(mysql));
+        goto err;
+      }
+    }
+  }
 err:
   if (errmsg)
   {
@@ -2498,6 +2540,9 @@ int apply_event_and_update_pos(Log_event* ev, THD* thd, Relay_log_info* rli)
     ev->when= hrtime_to_my_time(hrtime);
     ev->when_sec_part= hrtime_sec_part(hrtime);
   }
+  thd->variables.option_bits=
+    (thd->variables.option_bits & ~OPTION_SKIP_REPLICATION) |
+    (ev->flags & LOG_EVENT_SKIP_REPLICATION_F ? OPTION_SKIP_REPLICATION : 0);
   ev->thd = thd; // because up to this point, ev->thd == 0
 
   int reason= ev->shall_skip(rli);
@@ -4062,6 +4107,7 @@ static int queue_event(Master_info* mi,const char* buf, ulong event_len)
   int error= 0;
   String error_msg;
   ulong inc_pos;
+  ulong event_pos;
   Relay_log_info *rli= &mi->rli;
   mysql_mutex_t *log_lock= rli->relay_log.get_log_lock();
   ulong s_id;
@@ -4134,7 +4180,6 @@ static int queue_event(Master_info* mi,const char* buf, ulong event_len)
       (uchar)buf[EVENT_TYPE_OFFSET] != FORMAT_DESCRIPTION_EVENT /* a way to escape */)
     DBUG_RETURN(queue_old_event(mi,buf,event_len));
 
-  LINT_INIT(inc_pos);
   mysql_mutex_lock(&mi->data_lock);
 
   switch ((uchar)buf[EVENT_TYPE_OFFSET]) {
@@ -4326,6 +4371,23 @@ static int queue_event(Master_info* mi,const char* buf, ulong event_len)
     break;
   }
 
+  /*
+    If we filter events master-side (eg. @@skip_replication), we will see holes
+    in the event positions from the master. If we see such a hole, adjust
+    mi->master_log_pos accordingly so we maintain the correct position (for
+    reconnect, MASTER_POS_WAIT(), etc.)
+  */
+  if (inc_pos > 0 &&
+      event_len >= LOG_POS_OFFSET+4 &&
+      (event_pos= uint4korr(buf+LOG_POS_OFFSET)) > mi->master_log_pos + inc_pos)
+  {
+    inc_pos= event_pos - mi->master_log_pos;
+    DBUG_PRINT("info", ("Adjust master_log_pos %lu->%lu to account for "
+                        "master-side filtering",
+                        (unsigned long)(mi->master_log_pos + inc_pos),
+                        event_pos));
+  }
+
   /*
      If this event is originating from this server, don't queue it.
      We don't check this for 3.23 events because it's simpler like this; 3.23

sql/slave.h

@@ -152,6 +152,15 @@ extern ulonglong relay_log_space_limit;
 */
 #define SLAVE_FORCE_ALL 4
 
+/*
+  Values for the option --replicate-events-marked-for-skip.
+  Must match the names in replicate_events_marked_for_skip_names in sys_vars.cc
+*/
+#define RPL_SKIP_REPLICATE 0
+#define RPL_SKIP_FILTER_ON_SLAVE 1
+#define RPL_SKIP_FILTER_ON_MASTER 2
+
+
 int init_slave();
 int init_recovery(Master_info* mi, const char** errmsg);
 void init_slave_skip_errors(const char* arg);

sql/sql_binlog.cc

@@ -44,6 +44,7 @@
 
 void mysql_client_binlog_statement(THD* thd)
 {
+  ulonglong save_skip_replication;
   DBUG_ENTER("mysql_client_binlog_statement");
   DBUG_PRINT("info",("binlog base64: '%*s'",
                      (int) (thd->lex->comment.length < 2048 ?
@@ -225,7 +226,17 @@ void mysql_client_binlog_statement(THD* thd)
         reporting.
       */
 #if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
+      save_skip_replication= thd->variables.option_bits&OPTION_SKIP_REPLICATION;
+      thd->variables.option_bits=
+        (thd->variables.option_bits & ~OPTION_SKIP_REPLICATION) |
+        (ev->flags & LOG_EVENT_SKIP_REPLICATION_F ?
+         OPTION_SKIP_REPLICATION : 0);
+
       err= ev->apply_event(rli);
+
+      thd->variables.option_bits=
+        (thd->variables.option_bits & ~OPTION_SKIP_REPLICATION) |
+        save_skip_replication;
 #else
       err= 0;
 #endif

sql/sql_priv.h

@@ -151,6 +151,7 @@
   Note! Reserved for use in MySQL Cluster
 */
 #define OPTION_ALLOW_BATCH              (ULL(1) << 36) // THD, intern (slave)
+#define OPTION_SKIP_REPLICATION         (ULL(1) << 37) // THD, user
 
 /* The rest of the file is included in the server only */
 #ifndef MYSQL_CLIENT

sql/sql_repl.cc

@@ -556,8 +556,60 @@ static int send_heartbeat_event(NET* net, String* packet,
 
 
 /*
-  TODO: Clean up loop to only have one call to send_file()
+  Helper function for mysql_binlog_send() to write an event down the slave
+  connection.
+
+  Returns NULL on success, error message string on error.
 */
+static const char *
+send_event_to_slave(THD *thd, NET *net, String* const packet, ushort flags,
+                    Log_event_type event_type, char *log_file_name,
+                    IO_CACHE *log)
+{
+  my_off_t pos;
+
+  /* Do not send annotate_rows events unless slave requested it. */
+  if (event_type == ANNOTATE_ROWS_EVENT &&
+      !(flags & BINLOG_SEND_ANNOTATE_ROWS_EVENT))
+    return NULL;
+
+  /*
+    Skip events with the @@skip_replication flag set, if slave requested
+    skipping of such events.
+  */
+  if (thd->variables.option_bits & OPTION_SKIP_REPLICATION)
+  {
+    /*
+      The first byte of the packet is a '\0' to distinguish it from an error
+      packet. So the actual event starts at offset +1.
+    */
+    uint16 event_flags= uint2korr(&((*packet)[FLAGS_OFFSET+1]));
+    if (event_flags & LOG_EVENT_SKIP_REPLICATION_F)
+      return NULL;
+  }
+
+  thd_proc_info(thd, "Sending binlog event to slave");
+
+  pos= my_b_tell(log);
+  if (RUN_HOOK(binlog_transmit, before_send_event,
+               (thd, flags, packet, log_file_name, pos)))
+    return "run 'before_send_event' hook failed";
+
+  if (my_net_write(net, (uchar*) packet->ptr(), packet->length()))
+    return "Failed on my_net_write()";
+
+  DBUG_PRINT("info", ("log event code %d", (*packet)[LOG_EVENT_OFFSET+1] ));
+  if (event_type == LOAD_EVENT)
+  {
+    if (send_file(thd))
+      return "failed in send_file()";
+  }
+
+  if (RUN_HOOK(binlog_transmit, after_send_event, (thd, flags, packet)))
+    return "Failed to run hook 'after_send_event'";
+
+  return NULL;    /* Success */
+}
 
 void mysql_binlog_send(THD* thd, char* log_ident, my_off_t pos,
 		       ushort flags)
@@ -570,9 +622,9 @@ void mysql_binlog_send(THD* thd, char* log_ident, my_off_t pos,
 
   IO_CACHE log;
   File file = -1;
-  String* packet = &thd->packet;
+  String* const packet = &thd->packet;
   int error;
-  const char *errmsg = "Unknown error";
+  const char *errmsg = "Unknown error", *tmp_msg;
   const char *fmt= "%s; the last event was read from '%s' at %s, the last byte read was read from '%s' at %s.";
   char llbuff1[22], llbuff2[22];
   char error_text[MAX_SLAVE_ERRMSG]; // to be send to slave via my_message()
@@ -889,51 +941,21 @@ impossible position";
         (*packet)[FLAGS_OFFSET+ev_offset] &= ~LOG_EVENT_BINLOG_IN_USE_F;
       }
 
-      if (event_type != ANNOTATE_ROWS_EVENT ||
-          (flags & BINLOG_SEND_ANNOTATE_ROWS_EVENT))
+      if ((tmp_msg= send_event_to_slave(thd, net, packet, flags, event_type,
+                                        log_file_name, &log)))
       {
-        pos = my_b_tell(&log);
-        if (RUN_HOOK(binlog_transmit, before_send_event,
-                     (thd, flags, packet, log_file_name, pos)))
-        {
-          my_errno= ER_UNKNOWN_ERROR;
-          errmsg= "run 'before_send_event' hook failed";
-          goto err;
-        }
-
-        if (my_net_write(net, (uchar*) packet->ptr(), packet->length()))
-        {
-          errmsg = "Failed on my_net_write()";
-          my_errno= ER_UNKNOWN_ERROR;
-          goto err;
-        }
+        errmsg= tmp_msg;
+        my_errno= ER_UNKNOWN_ERROR;
+        goto err;
+      }
 
-        DBUG_EXECUTE_IF("dump_thread_wait_before_send_xid",
+      DBUG_EXECUTE_IF("dump_thread_wait_before_send_xid",
+                      {
+                        if (event_type == XID_EVENT)
                         {
-                          if (event_type == XID_EVENT)
-                          {
-                            net_flush(net);
-                          }
-                        });
-
-        DBUG_PRINT("info", ("log event code %d", event_type));
-        if (event_type == LOAD_EVENT)
-        {
-          if (send_file(thd))
-          {
-            errmsg = "failed in send_file()";
-            my_errno= ER_UNKNOWN_ERROR;
-            goto err;
-          }
-        }
-
-        if (RUN_HOOK(binlog_transmit, after_send_event, (thd, flags, packet)))
-        {
-          errmsg= "Failed to run hook 'after_send_event'";
-          my_errno= ER_UNKNOWN_ERROR;
-          goto err;
-        }
-      }
+                          net_flush(net);
+                        }
+                      });
 
       /* reset transmit packet for next loop */
       if (reset_transmit_packet(thd, flags, &ev_offset, &errmsg))
@@ -1078,43 +1100,13 @@ impossible position";
           goto err;
 	}
 
-	if (read_packet &&
-            (event_type != ANNOTATE_ROWS_EVENT ||
-             (flags & BINLOG_SEND_ANNOTATE_ROWS_EVENT)))
+        if (read_packet &&
+            (tmp_msg= send_event_to_slave(thd, net, packet, flags, event_type,
+                                          log_file_name, &log)))
         {
-          thd_proc_info(thd, "Sending binlog event to slave");
-          pos = my_b_tell(&log);
-          if (RUN_HOOK(binlog_transmit, before_send_event,
-                       (thd, flags, packet, log_file_name, pos)))
-          {
-            my_errno= ER_UNKNOWN_ERROR;
-            errmsg= "run 'before_send_event' hook failed";
-            goto err;
-          }
-	  
-	  if (my_net_write(net, (uchar*) packet->ptr(), packet->length()) )
-	  {
-	    errmsg = "Failed on my_net_write()";
-	    my_errno= ER_UNKNOWN_ERROR;
-	    goto err;
-	  }
-
-	  if (event_type == LOAD_EVENT)
-	  {
-	    if (send_file(thd))
-	    {
-	      errmsg = "failed in send_file()";
-	      my_errno= ER_UNKNOWN_ERROR;
-	      goto err;
-	    }
-	  }
-
-          if (RUN_HOOK(binlog_transmit, after_send_event, (thd, flags, packet)))
-          {
-            my_errno= ER_UNKNOWN_ERROR;
-            errmsg= "Failed to run hook 'after_send_event'";
-            goto err;
-          }
+          errmsg= tmp_msg;
+          my_errno= ER_UNKNOWN_ERROR;
+          goto err;
 	}
 
 	log.error=0;

sql/sys_vars.cc

@@ -231,6 +231,35 @@ static Sys_var_ulonglong Sys_binlog_stmt_cache_size(
        CMD_LINE(REQUIRED_ARG),
        VALID_RANGE(IO_SIZE, ULONGLONG_MAX), DEFAULT(32768), BLOCK_SIZE(IO_SIZE));
 
+/*
+  Some variables like @sql_log_bin and @binlog_format change how/if binlogging
+  is done. We must not change them inside a running transaction or statement,
+  otherwise the event group eventually written to the binlog may become
+  incomplete or otherwise garbled.
+
+  This function does the appropriate check.
+
+  It returns true if an error is caused by incorrect usage, false if ok.
+*/
+static bool
+error_if_in_trans_or_substatement(THD *thd, int in_substatement_error,
+                                  int in_transaction_error)
+{
+  if (thd->in_sub_stmt)
+  {
+    my_error(in_substatement_error, MYF(0));
+    return true;
+  }
+
+  if (thd->in_active_multi_stmt_transaction())
+  {
+    my_error(in_transaction_error, MYF(0));
+    return true;
+  }
+
+  return false;
+}
+
 static bool check_has_super(sys_var *self, THD *thd, set_var *var)
 {
   DBUG_ASSERT(self->scope() != sys_var::GLOBAL);// don't abuse check_has_super()
@@ -271,22 +300,10 @@ static bool binlog_format_check(sys_var *self, THD *thd, set_var *var)
     return true;
   }
 
-  /*
-    if in a stored function/trigger, it's too late to change mode
-  */
-  if (thd->in_sub_stmt)
-  {
-    my_error(ER_STORED_FUNCTION_PREVENTS_SWITCH_BINLOG_FORMAT, MYF(0));
+  if (error_if_in_trans_or_substatement(thd,
+         ER_STORED_FUNCTION_PREVENTS_SWITCH_BINLOG_FORMAT,
+         ER_INSIDE_TRANSACTION_PREVENTS_SWITCH_BINLOG_FORMAT))
     return true;
-  }
-  /*
-    Make the session variable 'binlog_format' read-only inside a transaction.
-  */
-  if (thd->in_active_multi_stmt_transaction())
-  {
-    my_error(ER_INSIDE_TRANSACTION_PREVENTS_SWITCH_BINLOG_FORMAT, MYF(0));
-    return true;
-  }
 
   return false;
 }
@@ -322,24 +339,10 @@ static bool binlog_direct_check(sys_var *self, THD *thd, set_var *var)
   if (var->type == OPT_GLOBAL)
     return false;
 
-   /*
-     Makes the session variable 'binlog_direct_non_transactional_updates'
-     read-only if within a procedure, trigger or function.
-   */
-   if (thd->in_sub_stmt)
-   {
-     my_error(ER_STORED_FUNCTION_PREVENTS_SWITCH_BINLOG_DIRECT, MYF(0));
-     return true;
-   }
-   /*
-     Makes the session variable 'binlog_direct_non_transactional_updates'
-     read-only inside a transaction.
-   */
-   if (thd->in_active_multi_stmt_transaction())
-   {
-     my_error(ER_INSIDE_TRANSACTION_PREVENTS_SWITCH_BINLOG_DIRECT, MYF(0));
+  if (error_if_in_trans_or_substatement(thd,
+          ER_STORED_FUNCTION_PREVENTS_SWITCH_BINLOG_DIRECT,
+          ER_INSIDE_TRANSACTION_PREVENTS_SWITCH_BINLOG_DIRECT))
      return true;
-   }
 
   return false;
 }
@@ -2017,6 +2020,67 @@ static Sys_var_mybool Sys_master_verify_checksum(
        "SHOW BINLOG EVENTS",
        GLOBAL_VAR(opt_master_verify_checksum), CMD_LINE(OPT_ARG),
        DEFAULT(FALSE));
+
+/* 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
+};
+static bool
+replicate_events_marked_for_skip_check(sys_var *self, THD *thd,
+                                                set_var *var)
+{
+  int thread_mask;
+  DBUG_ENTER("sys_var_replicate_events_marked_for_skip_check");
+
+  /* Slave threads must be stopped to change the variable. */
+  mysql_mutex_lock(&LOCK_active_mi);
+  lock_slave_threads(active_mi);
+  init_thread_mask(&thread_mask, active_mi, 0 /*not inverse*/);
+  unlock_slave_threads(active_mi);
+  mysql_mutex_unlock(&LOCK_active_mi);
+
+  if (thread_mask) // We refuse if any slave thread is running
+  {
+    my_error(ER_SLAVE_MUST_STOP, MYF(0));
+    DBUG_RETURN(true);
+  }
+  DBUG_RETURN(false);
+}
+bool
+Sys_var_replicate_events_marked_for_skip::global_update(THD *thd, set_var *var)
+{
+  bool result;
+  int thread_mask;
+  DBUG_ENTER("Sys_var_replicate_events_marked_for_skip::global_update");
+
+  /* Slave threads must be stopped to change the variable. */
+  mysql_mutex_lock(&LOCK_active_mi);
+  lock_slave_threads(active_mi);
+  init_thread_mask(&thread_mask, active_mi, 0 /*not inverse*/);
+  if (thread_mask) // We refuse if any slave thread is running
+  {
+    my_error(ER_SLAVE_MUST_STOP, MYF(0));
+    result= true;
+  }
+  else
+    result= Sys_var_enum::global_update(thd, var);
+
+  unlock_slave_threads(active_mi);
+  mysql_mutex_unlock(&LOCK_active_mi);
+  DBUG_RETURN(result);
+}
+static Sys_var_replicate_events_marked_for_skip Replicate_events_marked_for_skip
+   ("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).",
+   GLOBAL_VAR(opt_replicate_events_marked_for_skip), CMD_LINE(REQUIRED_ARG),
+   replicate_events_marked_for_skip_names, DEFAULT(RPL_SKIP_REPLICATE),
+   NO_MUTEX_GUARD, NOT_IN_BINLOG,
+   ON_CHECK(replicate_events_marked_for_skip_check));
 #endif
 
 
@@ -2571,18 +2635,10 @@ static bool check_sql_log_bin(sys_var *self, THD *thd, set_var *var)
   if (var->type == OPT_GLOBAL)
     return FALSE;
 
-  /* If in a stored function/trigger, it's too late to change sql_log_bin. */
-  if (thd->in_sub_stmt)
-  {
-    my_error(ER_STORED_FUNCTION_PREVENTS_SWITCH_SQL_LOG_BIN, MYF(0));
+  if (error_if_in_trans_or_substatement(thd,
+          ER_STORED_FUNCTION_PREVENTS_SWITCH_SQL_LOG_BIN,
+          ER_INSIDE_TRANSACTION_PREVENTS_SWITCH_SQL_LOG_BIN))
     return TRUE;
-  }
-  /* Make the session variable 'sql_log_bin' read-only inside a transaction. */
-  if (thd->in_active_multi_stmt_transaction())
-  {
-    my_error(ER_INSIDE_TRANSACTION_PREVENTS_SWITCH_SQL_LOG_BIN, MYF(0));
-    return TRUE;
-  }
 
   return FALSE;
 }
@@ -2647,6 +2703,40 @@ static Sys_var_ulong Sys_profiling_history_size(
        VALID_RANGE(0, 100), DEFAULT(15), BLOCK_SIZE(1));
 #endif
 
+/*
+  When this is set by a connection, binlogged events will be marked with a
+  corresponding flag. The slave can be configured to not replicate events
+  so marked.
+  In the binlog dump thread on the master, this variable is re-used for a
+  related purpose: The slave sets this flag when connecting to the master to
+  request that the master filter out (ie. not send) any events with the flag
+  set, thus saving network traffic on events that would be ignored by the
+  slave anyway.
+*/
+static bool check_skip_replication(sys_var *self, THD *thd, set_var *var)
+{
+  /*
+    We must not change @@skip_replication in the middle of a transaction or
+    statement, as that could result in only part of the transaction / statement
+    being replicated.
+    (This would be particularly serious if we were to replicate eg.
+    Rows_log_event without Table_map_log_event or transactional updates without
+    the COMMIT).
+  */
+  if (error_if_in_trans_or_substatement(thd,
+          ER_STORED_FUNCTION_PREVENTS_SWITCH_SKIP_REPLICATION,
+          ER_INSIDE_TRANSACTION_PREVENTS_SWITCH_SKIP_REPLICATION))
+    return 1;
+
+  return 0;
+}
+
+static Sys_var_bit Sys_skip_replication(
+       "skip_replication", "skip_replication",
+       SESSION_ONLY(option_bits), NO_CMD_LINE, OPTION_SKIP_REPLICATION,
+       DEFAULT(FALSE), NO_MUTEX_GUARD, NOT_IN_BINLOG,
+       ON_CHECK(check_skip_replication));
+
 static Sys_var_harows Sys_select_limit(
        "sql_select_limit",
        "The maximum number of rows to return from SELECT statements",
@@ -3523,7 +3613,7 @@ static Sys_var_mybool Sys_query_cache_strip_comments(
 
 static ulonglong in_transaction(THD *thd)
 {
-  return test(thd->server_status & SERVER_STATUS_IN_TRANS);
+  return test(thd->in_active_multi_stmt_transaction());
 }
 static Sys_var_session_special Sys_in_transaction(
        "in_transaction", "Whether there is an active transaction",

sql/sys_vars.h

@@ -1800,6 +1800,26 @@ public:
   }
 };
 
+/*
+  Class for replicate_events_marked_for_skip.
+  We need a custom update function that ensures the slave is stopped when
+  the update is happening.
+*/
+class Sys_var_replicate_events_marked_for_skip: public Sys_var_enum
+{
+public:
+  Sys_var_replicate_events_marked_for_skip(const char *name_arg,
+          const char *comment, int flag_args, ptrdiff_t off, size_t size,
+          CMD_LINE getopt,
+          const char *values[], uint def_val, PolyLock *lock,
+          enum binlog_status_enum binlog_status_arg,
+          on_check_function on_check_func)
+    :Sys_var_enum(name_arg, comment, flag_args, off, size, getopt,
+                  values, def_val, lock, binlog_status_arg, on_check_func)
+  {}
+  bool global_update(THD *thd, set_var *var);
+};
+
 /****************************************************************************
   Used templates
 ****************************************************************************/