MDEV-12067 flashback does not correcly revert update/replace statementsbb-10.2-MDEV-12067

Problem
-------
For one-statement contains multiple row events, Flashback didn't reverse the
sequence of row events inside one-statement.

Solution
--------
Using a new array 'events_in_stmt' to store the row events of one-statement,
when parsed the last one event, then print from the last one to the first one.

In the same time, fixed another bug, without -vv will not insert the table_map
into print_event_info->m_table_map, then change_to_flashback_event() will not
execute because of Table_map_log_event is empty.

5 files changed, 458 insertions, 46 deletions

diff --git a/client/mysqlbinlog.cc b/client/mysqlbinlog.cc
index 34e810f7b6b..584f6955453 100644
--- a/client/mysqlbinlog.cc
+++ b/client/mysqlbinlog.cc
@@ -68,6 +68,7 @@ CHARSET_INFO* system_charset_info= &my_charset_utf8_general_ci;
 
 /* Needed for Flashback */
 DYNAMIC_ARRAY binlog_events; // Storing the events output string
+DYNAMIC_ARRAY events_in_stmt; // Storing the events that in one statement
 String stop_event_string; // Storing the STOP_EVENT output string
 
 char server_version[SERVER_VERSION_LENGTH];
@@ -894,6 +895,25 @@ static bool print_row_event(PRINT_EVENT_INFO *print_event_info, Log_event *ev,
     print_event_info->m_table_map_ignored.get_table(table_id);
   bool skip_event= (ignored_map != NULL);
 
+  if (opt_flashback)
+  {
+    Rows_log_event *e= (Rows_log_event*) ev;
+    // The last Row_log_event will be the first event in Flashback
+    if (is_stmt_end)
+      e->clear_flags(Rows_log_event::STMT_END_F);
+    // The first Row_log_event will be the last event in Flashback
+    if (events_in_stmt.elements == 0)
+      e->set_flags(Rows_log_event::STMT_END_F);
+    // Update the temp_buf
+    e->update_flags();
+
+    if (insert_dynamic(&events_in_stmt, (uchar *) &ev))
+    {
+      error("Out of memory: can't allocate memory to store the flashback events.");
+      exit(1);
+    }
+  }
+
   /* 
      end of statement check:
        i) destroy/free ignored maps
@@ -945,7 +965,36 @@ static bool print_row_event(PRINT_EVENT_INFO *print_event_info, Log_event *ev,
   if (skip_event)
     return 0;
 
-  return print_base64(print_event_info, ev);
+  if (!opt_flashback)
+    return print_base64(print_event_info, ev);
+  else
+  {
+    if (is_stmt_end)
+    {
+      bool res= false;
+      Log_event *e= NULL;
+
+      // Print the row_event from the last one to the first one
+      for (uint i= events_in_stmt.elements; i > 0; --i)
+      {
+        e= *(dynamic_element(&events_in_stmt, i - 1, Log_event**));
+        res= res || print_base64(print_event_info, e);
+      }
+      // Copy all output into the Log_event
+      ev->output_buf.copy(e->output_buf);
+      // Delete Log_event
+      for (uint i= 0; i < events_in_stmt.elements-1; ++i)
+      {
+        e= *(dynamic_element(&events_in_stmt, i, Log_event**));
+        delete e;
+      }
+      reset_dynamic(&events_in_stmt);
+
+      return res;
+    }
+  }
+
+  return 0;
 }
 
 
@@ -1386,6 +1435,8 @@ Exit_status process_event(PRINT_EVENT_INFO *print_event_info, Log_event *ev,
       }
       if (print_base64(print_event_info, ev))
         goto err;
+      if (opt_flashback)
+        reset_dynamic(&events_in_stmt);
       break;
     }
     case WRITE_ROWS_EVENT:
@@ -1402,9 +1453,12 @@ Exit_status process_event(PRINT_EVENT_INFO *print_event_info, Log_event *ev,
     case DELETE_ROWS_COMPRESSED_EVENT_V1:
     {
       Rows_log_event *e= (Rows_log_event*) ev;
+      bool is_stmt_end= e->get_flags(Rows_log_event::STMT_END_F);
       if (print_row_event(print_event_info, ev, e->get_table_id(),
                           e->get_flags(Rows_log_event::STMT_END_F)))
         goto err;
+      if (!is_stmt_end)
+        destroy_evt= FALSE;
       break;
     }
     case PRE_GA_WRITE_ROWS_EVENT:
@@ -1412,9 +1466,12 @@ Exit_status process_event(PRINT_EVENT_INFO *print_event_info, Log_event *ev,
     case PRE_GA_UPDATE_ROWS_EVENT:
     {
       Old_rows_log_event *e= (Old_rows_log_event*) ev;
+      bool is_stmt_end= e->get_flags(Rows_log_event::STMT_END_F);
       if (print_row_event(print_event_info, ev, e->get_table_id(),
                           e->get_flags(Old_rows_log_event::STMT_END_F)))
         goto err;
+      if (!is_stmt_end)
+        destroy_evt= FALSE;
       break;
     }
     case START_ENCRYPTION_EVENT:
@@ -1459,7 +1516,7 @@ end:
                                   &my_charset_bin);
         else
         {
-          if (push_dynamic(&binlog_events, (uchar *) &tmp_str))
+          if (insert_dynamic(&binlog_events, (uchar *) &tmp_str))
           {
             error("Out of memory: can't allocate memory to store the flashback events.");
             exit(1);
@@ -2915,9 +2972,12 @@ int main(int argc, char** argv)
   my_set_max_open_files(open_files_limit);
 
   if (opt_flashback)
+  {
     my_init_dynamic_array(&binlog_events, sizeof(LEX_STRING), 1024, 1024,
                           MYF(0));
-
+    my_init_dynamic_array(&events_in_stmt, sizeof(Rows_log_event*), 1024, 1024,
+                          MYF(0));
+  }
   if (opt_stop_never)
     to_last_remote_log= TRUE;
 
@@ -3031,6 +3091,7 @@ int main(int argc, char** argv)
     }
     fprintf(result_file, "COMMIT\n/*!*/;\n");
     delete_dynamic(&binlog_events);
+    delete_dynamic(&events_in_stmt);
   }
 
   /* Set delimiter back to semicolon */
diff --git a/mysql-test/suite/binlog/r/flashback.result b/mysql-test/suite/binlog/r/flashback.result
index ae8b583fc7a..96729d3bc3c 100644
--- a/mysql-test/suite/binlog/r/flashback.result
+++ b/mysql-test/suite/binlog/r/flashback.result
@@ -6,7 +6,7 @@ DROP TABLE IF EXISTS t1;
 # We need a fixed timestamp to avoid varying results.
 #
 SET timestamp=1000000000;
-#
+# < CASE 1 >
 # Delete all existing binary logs.
 #
 RESET MASTER;
@@ -20,22 +20,22 @@ c06 char(10),
 c07 varchar(20),
 c08 TEXT
 ) ENGINE=InnoDB;
-#
+# < CASE 1 >
 # Insert data to t1
 #
 INSERT INTO t1 VALUES(0,0,0,0,0,'','','');
 INSERT INTO t1 VALUES(1,2,3,4,5, "abc", "abcdefg", "abcedfghijklmnopqrstuvwxyz");
 INSERT INTO t1 VALUES(127, 32767, 8388607, 2147483647, 9223372036854775807, repeat('a', 10), repeat('a', 20), repeat('a', 255));
-#
+# < CASE 1 >
 # Update t1
 #
 UPDATE t1 SET c01=100 WHERE c02=0 OR c03=3;
-#
+# < CASE 1 >
 # Clear t1
 #
 DELETE FROM t1;
 FLUSH LOGS;
-#
+# < CASE 1 >
 # Show mysqlbinlog result without -B
 #
 /*!50530 SET @@SESSION.PSEUDO_SLAVE_MODE=1*/;
@@ -258,7 +258,7 @@ DELIMITER ;
 ROLLBACK /* added by mysqlbinlog */;
 /*!50003 SET COMPLETION_TYPE=@OLD_COMPLETION_TYPE*/;
 /*!50530 SET @@SESSION.PSEUDO_SLAVE_MODE=0*/;
-#
+# < CASE 1 >
 # Show mysqlbinlog result with -B
 #
 /*!50530 SET @@SESSION.PSEUDO_SLAVE_MODE=1*/;
@@ -426,14 +426,14 @@ DELIMITER ;
 ROLLBACK /* added by mysqlbinlog */;
 /*!50003 SET COMPLETION_TYPE=@OLD_COMPLETION_TYPE*/;
 /*!50530 SET @@SESSION.PSEUDO_SLAVE_MODE=0*/;
-#
+# < CASE 1 >
 # Insert data to t1
 #
 TRUNCATE TABLE t1;
 INSERT INTO t1 VALUES(0,0,0,0,0,'','','');
 INSERT INTO t1 VALUES(1,2,3,4,5, "abc", "abcdefg", "abcedfghijklmnopqrstuvwxyz");
 INSERT INTO t1 VALUES(127, 32767, 8388607, 2147483647, 9223372036854775807, repeat('a', 10), repeat('a', 20), repeat('a', 60));
-#
+# < CASE 1 >
 # Delete all existing binary logs.
 #
 RESET MASTER;
@@ -442,7 +442,7 @@ c01	c02	c03	c04	c05	c06	c07	c08
 0	0	0	0	0			
 1	2	3	4	5	abc	abcdefg	abcedfghijklmnopqrstuvwxyz
 127	32767	8388607	2147483647	9223372036854775807	aaaaaaaaaa	aaaaaaaaaaaaaaaaaaaa	aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa
-#
+# < CASE 1 >
 # Operate some data
 #
 UPDATE t1 SET c01=20;
@@ -450,7 +450,7 @@ UPDATE t1 SET c02=200;
 UPDATE t1 SET c03=2000;
 DELETE FROM t1;
 FLUSH LOGS;
-#
+# < CASE 1 >
 # Flashback & Check the result
 #
 SELECT * FROM t1;
@@ -459,7 +459,7 @@ c01	c02	c03	c04	c05	c06	c07	c08
 1	2	3	4	5	abc	abcdefg	abcedfghijklmnopqrstuvwxyz
 0	0	0	0	0			
 RESET MASTER;
-#
+# < CASE 2 >
 # UPDATE multi-rows in one event
 #
 BEGIN;
@@ -467,7 +467,7 @@ UPDATE t1 SET c01=10 WHERE c01=0;
 UPDATE t1 SET c01=20 WHERE c01=10;
 COMMIT;
 FLUSH LOGS;
-#
+# < CASE 2 >
 # Flashback & Check the result
 #
 SELECT * FROM t1;
@@ -476,7 +476,7 @@ c01	c02	c03	c04	c05	c06	c07	c08
 1	2	3	4	5	abc	abcdefg	abcedfghijklmnopqrstuvwxyz
 0	0	0	0	0			
 DROP TABLE t1;
-#
+# < CASE 3 >
 # Self-referencing foreign keys
 #
 CREATE TABLE t1 (a INT PRIMARY KEY, b INT, FOREIGN KEY my_fk(b) REFERENCES t1(a)) ENGINE=InnoDB;
@@ -493,7 +493,110 @@ a	b
 RESET MASTER;
 DELETE FROM t1 ORDER BY a DESC;
 FLUSH LOGS;
+# < CASE 3 >
+# Flashback & Check the result
 #
+SELECT * FROM t1;
+a	b
+1	NULL
+2	1
+3	2
+4	3
+DROP TABLE t1;
+# < CASE 4 >
+# Trigger
+#
+CREATE TABLE t1 (a INT PRIMARY KEY, b INT) ENGINE=InnoDB;
+CREATE TABLE t2 (a INT PRIMARY KEY, b INT) ENGINE=InnoDB;
+BEGIN;
+INSERT INTO t1 VALUES (1, NULL);
+INSERT INTO t1 VALUES (2, 1), (3, 2), (4, 3);
+INSERT INTO t2 VALUES (6, 7), (7, 8), (8, 9);
+COMMIT;
+SELECT * FROM t1;
+a	b
+1	NULL
+2	1
+3	2
+4	3
+SELECT * FROM t2;
+a	b
+6	7
+7	8
+8	9
+CREATE TRIGGER trg1 BEFORE INSERT ON t1 FOR EACH ROW DELETE FROM t2 WHERE a = NEW.b;
+RESET MASTER;
+INSERT INTO t1 VALUES (5, 6), (7, 8);
+SELECT * FROM t1;
+a	b
+1	NULL
+2	1
+3	2
+4	3
+5	6
+7	8
+SELECT * FROM t2;
+a	b
+7	8
+FLUSH LOGS;
+# < CASE 4 >
+# Flashback & Check the result
+#
+SELECT * FROM t1;
+a	b
+1	NULL
+2	1
+3	2
+4	3
+SELECT * FROM t2;
+a	b
+6	7
+7	8
+8	9
+DROP TRIGGER trg1;
+DROP TABLE t1;
+DROP TABLE t2;
+# < CASE 5 >
+# REPLCAE Queries
+#
+CREATE TABLE t1 (a INT PRIMARY KEY, b INT, UNIQUE uk(b)) ENGINE=InnoDB;
+BEGIN;
+INSERT INTO t1 VALUES (1, NULL);
+INSERT INTO t1 VALUES (2, 1), (3, 2), (4, 3);
+INSERT INTO t1 VALUES (5, 4), (6, 5), (7, 6);
+COMMIT;
+SELECT * FROM t1;
+a	b
+1	NULL
+2	1
+3	2
+4	3
+5	4
+6	5
+7	6
+RESET MASTER;
+REPLACE INTO t1 VALUES (3, 100);
+REPLACE INTO t1 SET a=4, b=200;
+SELECT * FROM t1;
+a	b
+1	NULL
+2	1
+5	4
+6	5
+7	6
+3	100
+4	200
+REPLACE INTO t1 VALUES (5,5);
+SELECT * FROM t1;
+a	b
+1	NULL
+2	1
+5	5
+7	6
+3	100
+4	200
+FLUSH LOGS;
+# < CASE 5 >
 # Flashback & Check the result
 #
 SELECT * FROM t1;
@@ -502,9 +605,77 @@ a	b
 2	1
 3	2
 4	3
+5	4
+6	5
+7	6
+DROP TABLE t1;
+# < CASE 6 >
+# Test Case from MDEV-21067
+#
+CREATE DATABASE world;
+CREATE TABLE world.City (
+ID INT AUTO_INCREMENT PRIMARY KEY,
+Name VARCHAR(64),
+CountryCode VARCHAR(64),
+District VARCHAR(64),
+Population INT
+) ENGINE=InnoDB;
+CREATE TABLE test.test (
+ID INT AUTO_INCREMENT PRIMARY KEY,
+REC VARCHAR(64),
+ts TIMESTAMP
+) ENGINE=InnoDB;
+INSERT INTO world.City VALUES (NULL, 'Davenport', 'USA', 'Iowa', 100);
+INSERT INTO world.City VALUES (NULL, 'Boulder', 'USA', 'Colorado', 1000);
+INSERT INTO world.City VALUES (NULL, 'Gweru', 'ZWE', 'Midlands', 10000);
+RESET MASTER;
+CHECKSUM TABLE world.City;
+Table	Checksum
+world.City	563256876
+INSERT INTO test.test VALUES (NULL, 'Good record 1', CURRENT_TIMESTAMP());
+INSERT INTO world.City VALUES (NULL, 'Wrong value 1', '000', 'Wrong', 0);
+INSERT INTO world.City VALUES (NULL, 'Wrong value 2', '000', 'Wrong', 0) , (NULL, 'Wrong value 3', '000', 'Wrong', 0);
+INSERT INTO test.test VALUES (NULL, 'Good record 2', CURRENT_TIMESTAMP());
+UPDATE world.City SET Population = 99999999 WHERE ID IN (1, 2, 3);
+INSERT INTO test.test VALUES (NULL, 'Good record 3', CURRENT_TIMESTAMP());
+DELETE FROM world.City WHERE ID BETWEEN 1 AND 2;
+INSERT INTO test.test VALUES (NULL, 'Good record 5', CURRENT_TIMESTAMP());
+REPLACE INTO world.City VALUES (4074, 'Wrong value 4', '000', 'Wrong', 0);
+REPLACE INTO world.City VALUES (4078, 'Wrong value 5', '000', 'Wrong', 0), (NULL, 'Wrong value 6', '000', 'Wrong', 0);
+INSERT INTO test.test VALUES (NULL, 'Good record 6', CURRENT_TIMESTAMP());
+INSERT INTO world.City
+SELECT NULL, Name, CountryCode, District, Population FROM world.City WHERE ID BETWEEN 2 AND 10;
+INSERT INTO test.test VALUES (NULL, 'Good record 7', CURRENT_TIMESTAMP());
+INSERT INTO test.test VALUES (NULL, 'Good record 8', CURRENT_TIMESTAMP());
+DELETE FROM world.City;
+INSERT INTO test.test VALUES (NULL, 'Good record 9', CURRENT_TIMESTAMP());
+FLUSH LOGS;
+# < CASE 6 >
+# Flashback & Check the result
+#
+SELECT * FROM world.City;
+ID	Name	CountryCode	District	Population
+1	Davenport	USA	Iowa	100
+2	Boulder	USA	Colorado	1000
+3	Gweru	ZWE	Midlands	10000
+SELECT * FROM test.test;
+ID	REC	ts
+1	Good record 1	2001-09-09 09:46:40
+2	Good record 2	2001-09-09 09:46:40
+3	Good record 3	2001-09-09 09:46:40
+4	Good record 5	2001-09-09 09:46:40
+5	Good record 6	2001-09-09 09:46:40
+6	Good record 7	2001-09-09 09:46:40
+7	Good record 8	2001-09-09 09:46:40
+8	Good record 9	2001-09-09 09:46:40
+CHECKSUM TABLE world.City;
+Table	Checksum
+world.City	563256876
+DROP TABLE test.test;
+DROP TABLE world.City;
+DROP DATABASE world;
 SET binlog_format=statement;
 Warnings:
 Warning	1105	MariaDB Galera and flashback do not support binlog format: STATEMENT
 SET GLOBAL binlog_format=statement;
 ERROR HY000: Flashback does not support binlog_format STATEMENT
-DROP TABLE t1;
diff --git a/mysql-test/suite/binlog/t/flashback.test b/mysql-test/suite/binlog/t/flashback.test
index 8ef38dbcb9f..3636276a028 100644
--- a/mysql-test/suite/binlog/t/flashback.test
+++ b/mysql-test/suite/binlog/t/flashback.test
@@ -13,12 +13,11 @@ DROP TABLE IF EXISTS t1;
 --echo #
 SET timestamp=1000000000;
 
---echo #
+--echo # < CASE 1 >
 --echo # Delete all existing binary logs.
 --echo #
 RESET MASTER;
 
-
 CREATE TABLE t1 (
   c01 tinyint,
   c02 smallint,
@@ -30,7 +29,7 @@ CREATE TABLE t1 (
   c08 TEXT
 ) ENGINE=InnoDB;
 
---echo #
+--echo # < CASE 1 >
 --echo # Insert data to t1
 --echo #
 INSERT INTO t1 VALUES(0,0,0,0,0,'','','');
@@ -38,19 +37,19 @@ INSERT INTO t1 VALUES(1,2,3,4,5, "abc", "abcdefg", "abcedfghijklmnopqrstuvwxyz")
 INSERT INTO t1 VALUES(127, 32767, 8388607, 2147483647, 9223372036854775807, repeat('a', 10), repeat('a', 20), repeat('a', 255));
 
 
---echo #
+--echo # < CASE 1 >
 --echo # Update t1
 --echo #
 UPDATE t1 SET c01=100 WHERE c02=0 OR c03=3;
 
---echo #
+--echo # < CASE 1 >
 --echo # Clear t1
 --echo #
 DELETE FROM t1;
 
 FLUSH LOGS;
 
---echo #
+--echo # < CASE 1 >
 --echo # Show mysqlbinlog result without -B
 --echo #
 
@@ -59,7 +58,7 @@ let $MYSQLD_DATADIR= `select @@datadir`;
 --replace_regex /SQL_LOAD_MB-[0-9]-[0-9]/SQL_LOAD_MB-#-#/ /exec_time=[0-9]*/exec_time=#/ /end_log_pos [0-9]*/end_log_pos #/ /# at [0-9]*/# at #/ /Xid = [0-9]*/Xid = #/ /thread_id=[0-9]*/thread_id=#/ /table id [0-9]*/table id #/ /mapped to number [0-9]*/mapped to number #/ /server v [^ ]*/server v #.##.##/ /CRC32 0x[0-9a-f]*/CRC32 XXX/
 --exec $MYSQL_BINLOG --base64-output=decode-rows -v -v $MYSQLD_DATADIR/master-bin.000001
 
---echo #
+--echo # < CASE 1 >
 --echo # Show mysqlbinlog result with -B
 --echo #
 
@@ -68,7 +67,7 @@ let $MYSQLD_DATADIR= `select @@datadir`;
 --replace_regex /SQL_LOAD_MB-[0-9]-[0-9]/SQL_LOAD_MB-#-#/ /exec_time=[0-9]*/exec_time=#/ /end_log_pos [0-9]*/end_log_pos #/ /# at [0-9]*/# at #/ /Xid = [0-9]*/Xid = #/ /thread_id=[0-9]*/thread_id=#/ /table id [0-9]*/table id #/ /mapped to number [0-9]*/mapped to number #/ /server v [^ ]*/server v #.##.##/ /CRC32 0x[0-9a-f]*/CRC32 XXX/
 --exec $MYSQL_BINLOG -B --base64-output=decode-rows -v -v $MYSQLD_DATADIR/master-bin.000001
 
---echo #
+--echo # < CASE 1 >
 --echo # Insert data to t1
 --echo #
 TRUNCATE TABLE t1;
@@ -76,13 +75,13 @@ INSERT INTO t1 VALUES(0,0,0,0,0,'','','');
 INSERT INTO t1 VALUES(1,2,3,4,5, "abc", "abcdefg", "abcedfghijklmnopqrstuvwxyz");
 INSERT INTO t1 VALUES(127, 32767, 8388607, 2147483647, 9223372036854775807, repeat('a', 10), repeat('a', 20), repeat('a', 60));
 
---echo #
+--echo # < CASE 1 >
 --echo # Delete all existing binary logs.
 --echo #
 RESET MASTER;
 SELECT * FROM t1;
 
---echo #
+--echo # < CASE 1 >
 --echo # Operate some data
 --echo #
 
@@ -94,12 +93,13 @@ DELETE FROM t1;
 
 FLUSH LOGS;
 
---echo #
+--echo # < CASE 1 >
 --echo # Flashback & Check the result
 --echo #
 
 let $MYSQLD_DATADIR= `select @@datadir`;
 --replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+--exec $MYSQL_BINLOG -vv $MYSQLD_DATADIR/master-bin.000001 > $MYSQLTEST_VARDIR/tmp/mysqlbinlog_row_flashback_original_1.sql
 --exec $MYSQL_BINLOG -B -vv $MYSQLD_DATADIR/master-bin.000001 > $MYSQLTEST_VARDIR/tmp/mysqlbinlog_row_flashback_1.sql
 --exec $MYSQL -e "SET binlog_format= ROW; source $MYSQLTEST_VARDIR/tmp/mysqlbinlog_row_flashback_1.sql;"
 
@@ -107,9 +107,10 @@ SELECT * FROM t1;
 
 RESET MASTER;
 
---echo #
+--echo # < CASE 2 >
 --echo # UPDATE multi-rows in one event
 --echo #
+
 BEGIN;
 UPDATE t1 SET c01=10 WHERE c01=0;
 UPDATE t1 SET c01=20 WHERE c01=10;
@@ -117,12 +118,13 @@ COMMIT;
 
 FLUSH LOGS;
 
---echo #
+--echo # < CASE 2 >
 --echo # Flashback & Check the result
 --echo #
 
 let $MYSQLD_DATADIR= `select @@datadir`;
 --replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+--exec $MYSQL_BINLOG -vv $MYSQLD_DATADIR/master-bin.000001 > $MYSQLTEST_VARDIR/tmp/mysqlbinlog_row_flashback_original_2.sql
 --exec $MYSQL_BINLOG -B -vv $MYSQLD_DATADIR/master-bin.000001 > $MYSQLTEST_VARDIR/tmp/mysqlbinlog_row_flashback_2.sql
 --exec $MYSQL -e "SET binlog_format= ROW; source $MYSQLTEST_VARDIR/tmp/mysqlbinlog_row_flashback_2.sql;"
 
@@ -130,9 +132,10 @@ SELECT * FROM t1;
 
 DROP TABLE t1;
 
---echo #
+--echo # < CASE 3 >
 --echo # Self-referencing foreign keys
 --echo #
+
 CREATE TABLE t1 (a INT PRIMARY KEY, b INT, FOREIGN KEY my_fk(b) REFERENCES t1(a)) ENGINE=InnoDB;
 
 BEGIN;
@@ -149,19 +152,191 @@ DELETE FROM t1 ORDER BY a DESC;
 
 FLUSH LOGS;
 
---echo #
+--echo # < CASE 3 >
 --echo # Flashback & Check the result
 --echo #
 
 let $MYSQLD_DATADIR= `select @@datadir`;
 --replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+--exec $MYSQL_BINLOG -vv $MYSQLD_DATADIR/master-bin.000001 > $MYSQLTEST_VARDIR/tmp/mysqlbinlog_row_flashback_original_3.sql
 --exec $MYSQL_BINLOG -B -vv $MYSQLD_DATADIR/master-bin.000001 > $MYSQLTEST_VARDIR/tmp/mysqlbinlog_row_flashback_3.sql
 --exec $MYSQL -e "SET binlog_format= ROW; source $MYSQLTEST_VARDIR/tmp/mysqlbinlog_row_flashback_3.sql;"
 
 SELECT * FROM t1;
 
+DROP TABLE t1;
+
+--echo # < CASE 4 >
+--echo # Trigger
+--echo #
+
+CREATE TABLE t1 (a INT PRIMARY KEY, b INT) ENGINE=InnoDB;
+CREATE TABLE t2 (a INT PRIMARY KEY, b INT) ENGINE=InnoDB;
+
+BEGIN;
+INSERT INTO t1 VALUES (1, NULL);
+INSERT INTO t1 VALUES (2, 1), (3, 2), (4, 3);
+INSERT INTO t2 VALUES (6, 7), (7, 8), (8, 9);
+COMMIT;
+
+SELECT * FROM t1;
+SELECT * FROM t2;
+
+CREATE TRIGGER trg1 BEFORE INSERT ON t1 FOR EACH ROW DELETE FROM t2 WHERE a = NEW.b;
+
+# New binlog
+RESET MASTER;
+
+INSERT INTO t1 VALUES (5, 6), (7, 8);
+
+SELECT * FROM t1;
+SELECT * FROM t2;
+
+FLUSH LOGS;
+
+--echo # < CASE 4 >
+--echo # Flashback & Check the result
+--echo #
+
+let $MYSQLD_DATADIR= `select @@datadir`;
+--replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+--exec $MYSQL_BINLOG -vv $MYSQLD_DATADIR/master-bin.000001 > $MYSQLTEST_VARDIR/tmp/mysqlbinlog_row_flashback_original_4.sql
+--exec $MYSQL_BINLOG -B $MYSQLD_DATADIR/master-bin.000001 > $MYSQLTEST_VARDIR/tmp/mysqlbinlog_row_flashback_4.sql
+--exec $MYSQL -e "SET binlog_format= ROW; source $MYSQLTEST_VARDIR/tmp/mysqlbinlog_row_flashback_4.sql;"
+
+SELECT * FROM t1;
+SELECT * FROM t2;
+
+DROP TRIGGER trg1;
+DROP TABLE t1;
+DROP TABLE t2;
+
+--echo # < CASE 5 >
+--echo # REPLCAE Queries
+--echo #
+
+CREATE TABLE t1 (a INT PRIMARY KEY, b INT, UNIQUE uk(b)) ENGINE=InnoDB;
+
+BEGIN;
+INSERT INTO t1 VALUES (1, NULL);
+INSERT INTO t1 VALUES (2, 1), (3, 2), (4, 3);
+INSERT INTO t1 VALUES (5, 4), (6, 5), (7, 6);
+COMMIT;
+
+SELECT * FROM t1;
+
+# New binlog
+RESET MASTER;
+
+REPLACE INTO t1 VALUES (3, 100);
+REPLACE INTO t1 SET a=4, b=200;
+
+SELECT * FROM t1;
+
+REPLACE INTO t1 VALUES (5,5);
+
+SELECT * FROM t1;
+
+FLUSH LOGS;
+
+--echo # < CASE 5 >
+--echo # Flashback & Check the result
+--echo #
+
+let $MYSQLD_DATADIR= `select @@datadir`;
+--replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+--exec $MYSQL_BINLOG -vv $MYSQLD_DATADIR/master-bin.000001 > $MYSQLTEST_VARDIR/tmp/mysqlbinlog_row_flashback_original_5.sql
+--exec $MYSQL_BINLOG -B $MYSQLD_DATADIR/master-bin.000001 > $MYSQLTEST_VARDIR/tmp/mysqlbinlog_row_flashback_5.sql
+--exec $MYSQL -e "SET binlog_format= ROW; source $MYSQLTEST_VARDIR/tmp/mysqlbinlog_row_flashback_5.sql;"
+
+SELECT * FROM t1;
+
+DROP TABLE t1;
+
+
+--echo # < CASE 6 >
+--echo # Test Case from MDEV-21067
+--echo #
+
+# Init Structure
+CREATE DATABASE world;
+CREATE TABLE world.City (
+     ID INT AUTO_INCREMENT PRIMARY KEY,
+     Name VARCHAR(64),
+     CountryCode VARCHAR(64),
+     District VARCHAR(64),
+     Population INT
+) ENGINE=InnoDB;
+CREATE TABLE test.test (
+     ID INT AUTO_INCREMENT PRIMARY KEY,
+     REC VARCHAR(64),
+     ts TIMESTAMP
+) ENGINE=InnoDB;
+
+INSERT INTO world.City VALUES (NULL, 'Davenport', 'USA', 'Iowa', 100);
+INSERT INTO world.City VALUES (NULL, 'Boulder', 'USA', 'Colorado', 1000);
+INSERT INTO world.City VALUES (NULL, 'Gweru', 'ZWE', 'Midlands', 10000);
+
+RESET MASTER;
+
+CHECKSUM TABLE world.City;
+
+# Insert test data
+INSERT INTO test.test VALUES (NULL, 'Good record 1', CURRENT_TIMESTAMP());
+
+INSERT INTO world.City VALUES (NULL, 'Wrong value 1', '000', 'Wrong', 0);
+INSERT INTO world.City VALUES (NULL, 'Wrong value 2', '000', 'Wrong', 0) , (NULL, 'Wrong value 3', '000', 'Wrong', 0);
+
+INSERT INTO test.test VALUES (NULL, 'Good record 2', CURRENT_TIMESTAMP());
+
+UPDATE world.City SET Population = 99999999 WHERE ID IN (1, 2, 3);
+
+INSERT INTO test.test VALUES (NULL, 'Good record 3', CURRENT_TIMESTAMP());
+
+DELETE FROM world.City WHERE ID BETWEEN 1 AND 2;
+
+INSERT INTO test.test VALUES (NULL, 'Good record 5', CURRENT_TIMESTAMP());
+
+REPLACE INTO world.City VALUES (4074, 'Wrong value 4', '000', 'Wrong', 0);
+REPLACE INTO world.City VALUES (4078, 'Wrong value 5', '000', 'Wrong', 0), (NULL, 'Wrong value 6', '000', 'Wrong', 0);
+
+INSERT INTO test.test VALUES (NULL, 'Good record 6', CURRENT_TIMESTAMP());
+
+INSERT INTO world.City
+SELECT NULL, Name, CountryCode, District, Population FROM world.City WHERE ID BETWEEN 2 AND 10;
+
+INSERT INTO test.test VALUES (NULL, 'Good record 7', CURRENT_TIMESTAMP());
+
+INSERT INTO test.test VALUES (NULL, 'Good record 8', CURRENT_TIMESTAMP());
+
+DELETE FROM world.City;
+
+INSERT INTO test.test VALUES (NULL, 'Good record 9', CURRENT_TIMESTAMP());
+
+FLUSH LOGS;
+
+--echo # < CASE 6 >
+--echo # Flashback & Check the result
+--echo #
+
+let $MYSQLD_DATADIR= `select @@datadir`;
+--replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+--exec $MYSQL_BINLOG --database=world --table=City -vv $MYSQLD_DATADIR/master-bin.000001 > $MYSQLTEST_VARDIR/tmp/mysqlbinlog_row_flashback_original_6.sql
+--exec $MYSQL_BINLOG --database=world --table=City -B $MYSQLD_DATADIR/master-bin.000001 > $MYSQLTEST_VARDIR/tmp/mysqlbinlog_row_flashback_6.sql
+--exec $MYSQL -e "SET binlog_format= ROW; source $MYSQLTEST_VARDIR/tmp/mysqlbinlog_row_flashback_6.sql;"
+
+SELECT * FROM world.City;
+
+SELECT * FROM test.test;
+
+CHECKSUM TABLE world.City;
+
+DROP TABLE test.test;
+DROP TABLE world.City;
+DROP DATABASE world;
+
+## Clear
+
 SET binlog_format=statement;
 --error ER_FLASHBACK_NOT_SUPPORTED
 SET GLOBAL binlog_format=statement;
-
-DROP TABLE t1;
diff --git a/sql/log_event.cc b/sql/log_event.cc
index 6ff0211e536..79e8f7e0a94 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -3477,7 +3477,8 @@ void Log_event::print_base64(IO_CACHE* file,
 #ifdef WHEN_FLASHBACK_REVIEW_READY
   if (print_event_info->verbose || need_flashback_review)
 #else
-  if (print_event_info->verbose)
+  // Flashback need the table_map to parse the event
+  if (print_event_info->verbose || is_flashback)
 #endif
   {
     Rows_log_event *ev= NULL;
@@ -3564,7 +3565,8 @@ void Log_event::print_base64(IO_CACHE* file,
         close_cached_file(&tmp_cache);
       }
 #else
-      ev->print_verbose(file, print_event_info);
+      if (print_event_info->verbose)
+        ev->print_verbose(file, print_event_info);
 #endif
       delete ev;
     }
@@ -10251,6 +10253,7 @@ Rows_log_event::Rows_log_event(const char *buf, uint event_len,
     post_start+= RW_FLAGS_OFFSET;
   }
 
+  m_flags_pos= post_start - buf;
   m_flags= uint2korr(post_start);
   post_start+= 2;
 
@@ -11299,18 +11302,18 @@ void Rows_log_event::print_helper(FILE *file,
 
   if (get_flags(STMT_END_F))
   {
-    reinit_io_cache(head, READ_CACHE, 0L, FALSE, FALSE);
-    output_buf.append(head, head->end_of_file);
-    reinit_io_cache(head, WRITE_CACHE, 0, FALSE, TRUE);
-
-    reinit_io_cache(body, READ_CACHE, 0L, FALSE, FALSE);
-    output_buf.append(body, body->end_of_file);
-    reinit_io_cache(body, WRITE_CACHE, 0, FALSE, TRUE);
+    LEX_STRING tmp_str;
 
+    copy_event_cache_to_string_and_reinit(head, &tmp_str);
+    output_buf.append(&tmp_str);
+    my_free(tmp_str.str);
+    copy_event_cache_to_string_and_reinit(body, &tmp_str);
+    output_buf.append(&tmp_str);
+    my_free(tmp_str.str);
 #ifdef WHEN_FLASHBACK_REVIEW_READY
-    reinit_io_cache(sql, READ_CACHE, 0L, FALSE, FALSE);
-    output_buf.append(sql, sql->end_of_file);
-    reinit_io_cache(sql, WRITE_CACHE, 0, FALSE, TRUE);
+    copy_event_cache_to_string_and_reinit(sql, &tmp_str);
+    output_buf.append(&tmp_str);
+    my_free(tmp_str.str);
 #endif
   }
 }
diff --git a/sql/log_event.h b/sql/log_event.h
index 6c6dce7e18e..a06781aebeb 100644
--- a/sql/log_event.h
+++ b/sql/log_event.h
@@ -4397,6 +4397,7 @@ public:
   void set_flags(flag_set flags_arg) { m_flags |= flags_arg; }
   void clear_flags(flag_set flags_arg) { m_flags &= ~flags_arg; }
   flag_set get_flags(flag_set flags_arg) const { return m_flags & flags_arg; }
+  void update_flags() { int2store(temp_buf + m_flags_pos, m_flags); }
 
   Log_event_type get_type_code() { return m_type; } /* Specific type (_V1 etc) */
   virtual Log_event_type get_general_type_code() = 0; /* General rows op type, no version */
@@ -4555,6 +4556,7 @@ protected:
   uchar    *m_rows_end;		/* One-after the end of the allocated space */
 
   size_t   m_rows_before_size;  /* The length before m_rows_buf */
+  size_t   m_flags_pos; /* The position of the m_flags */
 
   flag_set m_flags;		/* Flags for row-level events */