BUG#42941: --database paramater to mysqlbinlog fails with RBR

            
mysqlbinlog --database parameter was being ignored when processing
row events. As such no event filtering would take place.
            
This patch addresses this by deploying a call to shall_skip_database
when table_map_events are handled (as these contain also the name of
the database). All other rows events referencing the table id for the
filtered map event, will also be skipped.

client/mysqlbinlog.cc:
  Added shall_skip_database call to the part of the code that handles 
  Table_map_log_events. It inspects the database name and decides whether
  to filter the event or not. Furthermore, if table map event is filtered
  next events referencing the table id in the table map event, will also
  be filtered.
mysql-test/suite/binlog/t/binlog_row_mysqlbinlog_db_filter.test:
  Test case that checks if row events are actually filtered out.
sql/log_event.h:
  Added a map for holding the currently ignored table map events.
  Table map events are inserted when they shall be skipped and removed
  once the last row event in the statement is processed.

4 files changed, 242 insertions, 1 deletions

diff --git a/client/mysqlbinlog.cc b/client/mysqlbinlog.cc
index 1864786e3c6..042a7957956 100644
--- a/client/mysqlbinlog.cc
+++ b/client/mysqlbinlog.cc
@@ -681,6 +681,7 @@ Exit_status process_event(PRINT_EVENT_INFO *print_event_info, Log_event *ev,
 {
   char ll_buff[21];
   Log_event_type ev_type= ev->get_type_code();
+  my_bool destroy_evt= TRUE;
   DBUG_ENTER("process_event");
   print_event_info->short_form= short_form;
   Exit_status retval= OK_CONTINUE;
@@ -871,12 +872,63 @@ Exit_status process_event(PRINT_EVENT_INFO *print_event_info, Log_event *ev,
       break;
     }
     case TABLE_MAP_EVENT:
+    {
+      Table_map_log_event *map= ((Table_map_log_event *)ev);
+      if (shall_skip_database(map->get_db_name()))
+      {
+        print_event_info->m_table_map_ignored.set_table(map->get_table_id(), map);
+        destroy_evt= FALSE;
+        goto end;
+      }
+    }
     case WRITE_ROWS_EVENT:
     case DELETE_ROWS_EVENT:
     case UPDATE_ROWS_EVENT:
     case PRE_GA_WRITE_ROWS_EVENT:
     case PRE_GA_DELETE_ROWS_EVENT:
     case PRE_GA_UPDATE_ROWS_EVENT:
+    {
+      if (ev_type != TABLE_MAP_EVENT)
+      {
+        Rows_log_event *e= (Rows_log_event*) ev;
+        Table_map_log_event *ignored_map= 
+          print_event_info->m_table_map_ignored.get_table(e->get_table_id());
+        bool skip_event= (ignored_map != NULL);
+
+        /* 
+           end of statement check:
+             i) destroy/free ignored maps
+            ii) if skip event, flush cache now
+         */
+        if (e->get_flags(Rows_log_event::STMT_END_F))
+        {
+          /* 
+            Now is safe to clear ignored map (clear_tables will also
+            delete original table map events stored in the map).
+          */
+          if (print_event_info->m_table_map_ignored.count() > 0)
+            print_event_info->m_table_map_ignored.clear_tables();
+
+          /* 
+             One needs to take into account an event that gets
+             filtered but was last event in the statement. If this is
+             the case, previous rows events that were written into
+             IO_CACHEs still need to be copied from cache to
+             result_file (as it would happen in ev->print(...) if
+             event was not skipped).
+          */
+          if (skip_event)
+          {
+            if ((copy_event_cache_to_file_and_reinit(&print_event_info->head_cache, result_file) ||
+                copy_event_cache_to_file_and_reinit(&print_event_info->body_cache, result_file)))
+              goto err;
+          }
+        }
+
+        /* skip the event check */
+        if (skip_event)
+          goto end;
+      }
       /*
         These events must be printed in base64 format, if printed.
         base64 format requires a FD event to be safe, so if no FD
@@ -900,6 +952,7 @@ Exit_status process_event(PRINT_EVENT_INFO *print_event_info, Log_event *ev,
         goto err;
       }
       /* FALL THROUGH */
+    }
     default:
       ev->print(result_file, print_event_info);
     }
@@ -919,7 +972,8 @@ end:
   {
     if (remote_opt)
       ev->temp_buf= 0;
-    delete ev;
+    if (destroy_evt) /* destroy it later if not set (ignored table map) */
+      delete ev;
   }
   DBUG_RETURN(retval);
 }
diff --git a/mysql-test/suite/binlog/r/binlog_row_mysqlbinlog_db_filter.result b/mysql-test/suite/binlog/r/binlog_row_mysqlbinlog_db_filter.result
new file mode 100644
index 00000000000..354fd832fb3
--- /dev/null
+++ b/mysql-test/suite/binlog/r/binlog_row_mysqlbinlog_db_filter.result
@@ -0,0 +1,43 @@
+RESET MASTER;
+CREATE TABLE t1 (id int);
+CREATE TABLE t2 (id int);
+CREATE TABLE t3 (txt TEXT);
+CREATE TABLE t4 (a int) ENGINE= InnoDB;
+INSERT INTO t1 VALUES (1);
+INSERT INTO t1 VALUES (2);
+INSERT INTO t2 VALUES (1);
+INSERT INTO t2 VALUES (2);
+INSERT INTO t1 VALUES (3);
+LOAD DATA INFILE 'MYSQLTEST_VARDIR/std_data/words.dat' INTO TABLE t3;
+INSERT INTO t1 VALUES (4);
+CREATE DATABASE b42941;
+use b42941;
+CREATE TABLE t1 (id int);
+CREATE TABLE t2 (id int);
+CREATE TABLE t3 (txt TEXT);
+CREATE TABLE t4 (a int) ENGINE= InnoDB;
+INSERT INTO t1 VALUES (1);
+INSERT INTO t1 VALUES (2);
+INSERT INTO t2 VALUES (1);
+INSERT INTO t2 VALUES (2);
+INSERT INTO t1 VALUES (3);
+LOAD DATA INFILE 'MYSQLTEST_VARDIR/std_data/words.dat' INTO TABLE t3;
+INSERT INTO t1 VALUES (4);
+INSERT INTO test.t1 VALUES (5);
+FLUSH LOGS;
+UPDATE test.t1 t11, b42941.t1 t12 SET t11.id=10, t12.id=100;
+BEGIN;
+INSERT INTO test.t4 VALUES (1);
+INSERT INTO b42941.t4 VALUES (1);
+UPDATE test.t4 tn4, b42941.t4 tt4 SET tn4.a= 10, tt4.a= 100;
+COMMIT;
+FLUSH LOGS;
+SET @b42941_output.1= LOAD_FILE('MYSQLTEST_VARDIR/tmp/b42941-mysqlbinlog.1');
+SET @b42941_output.2= LOAD_FILE('MYSQLTEST_VARDIR/tmp/b42941-mysqlbinlog.2');
+SET @b42941_output.1= LOAD_FILE('MYSQLTEST_VARDIR/tmp/b42941-mysqlbinlog.1');
+SET @b42941_output.2= LOAD_FILE('MYSQLTEST_VARDIR/tmp/b42941-mysqlbinlog.2');
+SET @b42941_output.1= LOAD_FILE('MYSQLTEST_VARDIR/tmp/b42941-mysqlbinlog.1');
+SET @b42941_output.2= LOAD_FILE('MYSQLTEST_VARDIR/tmp/b42941-mysqlbinlog.2');
+DROP DATABASE b42941;
+use test;
+DROP TABLE t1, t2, t3, t4;
diff --git a/mysql-test/suite/binlog/t/binlog_row_mysqlbinlog_db_filter.test b/mysql-test/suite/binlog/t/binlog_row_mysqlbinlog_db_filter.test
new file mode 100644
index 00000000000..0422c204270
--- /dev/null
+++ b/mysql-test/suite/binlog/t/binlog_row_mysqlbinlog_db_filter.test
@@ -0,0 +1,143 @@
+# BUG#42941: --database parameter to mysqlbinlog fails with RBR
+#
+# WHAT
+# ====
+#
+#  This test aims at checking whether a rows log event is printed or
+#  not when --database parameter is used to filter events from one
+#  given database.
+#
+# HOW
+# ===
+#
+#  The test is implemented as follows: 
+#
+#    i) Some operations are done in two different databases:
+#       'test' and 'b42941';
+#   ii) mysqlbinlog is used to dump the contents of the binlog file
+#       filtering only events from 'b42941'. The result of the dump is
+#       stored in a temporary file. (This is done with and without
+#       --verbose/hexdump flag);
+#  iii) The contents of the dump are loaded into a session variable;
+#   iv) The variable contents are searched for 'test' and 'b42941';
+#    v) Should 'test' be found, an ERROR is reported. Should 'b42941' be
+#       absent, an ERROR is reported.
+
+-- source include/have_log_bin.inc
+-- source include/have_binlog_format_row.inc
+-- source include/have_innodb.inc
+
+RESET MASTER;
+-- let $MYSQLD_DATADIR= `select @@datadir`
+
+CREATE TABLE t1 (id int);
+CREATE TABLE t2 (id int);
+CREATE TABLE t3 (txt TEXT);
+CREATE TABLE t4 (a int) ENGINE= InnoDB;
+INSERT INTO t1 VALUES (1);
+INSERT INTO t1 VALUES (2);
+INSERT INTO t2 VALUES (1);
+INSERT INTO t2 VALUES (2);
+INSERT INTO t1 VALUES (3);
+-- replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+-- eval LOAD DATA INFILE '$MYSQLTEST_VARDIR/std_data/words.dat' INTO TABLE t3
+INSERT INTO t1 VALUES (4);
+
+CREATE DATABASE b42941;
+use b42941;
+CREATE TABLE t1 (id int);
+CREATE TABLE t2 (id int);
+CREATE TABLE t3 (txt TEXT);
+CREATE TABLE t4 (a int) ENGINE= InnoDB;
+INSERT INTO t1 VALUES (1);
+INSERT INTO t1 VALUES (2);
+INSERT INTO t2 VALUES (1);
+INSERT INTO t2 VALUES (2);
+INSERT INTO t1 VALUES (3);
+-- replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+-- eval LOAD DATA INFILE '$MYSQLTEST_VARDIR/std_data/words.dat' INTO TABLE t3
+INSERT INTO t1 VALUES (4);
+
+INSERT INTO test.t1 VALUES (5);
+
+FLUSH LOGS;
+
+UPDATE test.t1 t11, b42941.t1 t12 SET t11.id=10, t12.id=100;
+
+BEGIN;
+INSERT INTO test.t4 VALUES (1);
+INSERT INTO b42941.t4 VALUES (1);
+UPDATE test.t4 tn4, b42941.t4 tt4 SET tn4.a= 10, tt4.a= 100;
+COMMIT;
+
+FLUSH LOGS;
+
+-- let $log_file1= $MYSQLD_DATADIR/master-bin.000001
+-- let $log_file2= $MYSQLD_DATADIR/master-bin.000002
+-- let $outfile= $MYSQLTEST_VARDIR/tmp/b42941-mysqlbinlog
+-- let $cmd= $MYSQL_BINLOG
+
+let $i= 3;
+while($i)
+{
+  -- let $flags=--database=b42941
+
+  # construct CLI for mysqlbinlog
+  if(`SELECT $i=3`)
+  {
+    -- let $flags= $flags --verbose --hexdump
+  }
+
+  if(`SELECT $i=2`)
+  {
+    -- let $flags= $flags --verbose
+  }
+
+#  if(`SELECT $i=1`)
+#  {
+    # do nothing $flags is already set as it should be
+#  }
+
+  # execute mysqlbinlog on the two available master binlog files
+  -- exec $cmd $flags $log_file1 > $outfile.1
+  -- exec $cmd $flags $log_file2 > $outfile.2
+
+  # load outputs into a variable
+  -- replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+  -- eval SET @b42941_output.1= LOAD_FILE('$outfile.1')
+
+  -- replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+  -- eval SET @b42941_output.2= LOAD_FILE('$outfile.2')
+
+  # remove unecessary files
+  -- remove_file $outfile.1
+  -- remove_file $outfile.2
+
+  # assertion: events for database test are filtered
+  if (`SELECT INSTR(@b42941_output.1, 'test')`)
+  {
+    -- echo **** ERROR **** Database name 'test' FOUND in mysqlbinlog output ($flags $outfile.1).
+  }
+
+  if (`SELECT INSTR(@b42941_output.2, 'test')`)
+  {
+    -- echo **** ERROR **** Database name 'test' FOUND in mysqlbinlog output ($flags $outfile.2).
+  }
+
+  # assertion: events for database b42941 are not filtered
+  if (!`SELECT INSTR(@b42941_output.1, 'b42941')`)
+  {
+    -- echo **** ERROR **** Database name 'b42941' NOT FOUND in mysqlbinlog output ($flags $outfile.1).
+  }
+
+  if (!`SELECT INSTR(@b42941_output.2, 'b42941')`)
+  {
+    -- echo **** ERROR **** Database name 'b42941' NOT FOUND in mysqlbinlog output ($flags $outfile.2).
+  }
+
+  dec $i;
+}
+
+DROP DATABASE b42941;
+use test;
+DROP TABLE t1, t2, t3, t4;
diff --git a/sql/log_event.h b/sql/log_event.h
index bda53da8ab0..c3cf6cf297a 100644
--- a/sql/log_event.h
+++ b/sql/log_event.h
@@ -676,6 +676,7 @@ typedef struct st_print_event_info
 #ifdef MYSQL_CLIENT
   uint verbose;
   table_mapping m_table_map;
+  table_mapping m_table_map_ignored;
 #endif
 
   /*