[MDEV-10570] Add Flashback support

==== Description ====

Flashback can rollback the instances/databases/tables to an old snapshot.
It's implement on Server-Level by full image format binary logs (--binlog-row-image=FULL), so it supports all engines.
Currently, it’s a feature inside mysqlbinlog tool (with --flashback arguments).

Because the flashback binlog events will store in the memory, you should check if there is enough memory in your machine.

==== New Arguments to mysqlbinlog ====

--flashback (-B)
It will let mysqlbinlog to work on FLASHBACK mode.

==== New Arguments to mysqld ====

--flashback

Setup the server to use flashback. This enables binary log in row mode
and will enable extra logging for DDL's needed by flashback feature

==== Example ====

I have a table "t" in database "test", we can compare the output with "--flashback" and without.

#client/mysqlbinlog /data/mysqldata_10.0/binlog/mysql-bin.000001 -vv -d test -T t --start-datetime="2013-03-27 14:54:00" > /tmp/1.sql
#client/mysqlbinlog /data/mysqldata_10.0/binlog/mysql-bin.000001 -vv -d test -T t --start-datetime="2013-03-27 14:54:00" -B > /tmp/2.sql

Then, importing the output flashback file (/tmp/2.log), it can flashback your database/table to the special time (--start-datetime).
And if you know the exact postion, "--start-postion" is also works, mysqlbinlog will output the flashback logs that can flashback to "--start-postion" position.

==== Implement ====

1. As we know, if binlog_format is ROW (binlog-row-image=FULL in 10.1 and later), all columns value are store in the row event, so we can get the data before mis-operation.

2. Just do following things:

  2.1 Change Event Type, INSERT->DELETE, DELETE->INSERT.
  For example:
    INSERT INTO t VALUES (...)  ---> DELETE FROM t WHERE ...
    DELETE FROM t ... ---> INSERT INTO t VALUES (...)

  2.2 For Update_Event, swapping the SET part and WHERE part.
  For example:
    UPDATE t SET cols1 = vals1 WHERE cols2 = vals2
    --->
    UPDATE t SET cols2 = vals2 WHERE cols1 = vals1

  2.3 For Multi-Rows Event, reverse the rows sequence, from the last row to the first row.
  For example:
    DELETE FROM t WHERE id=1; DELETE FROM t WHERE id=2; ...; DELETE FROM t WHERE id=n;
    --->
    DELETE FROM t WHERE id=n; ...; DELETE FROM t WHERE id=2; DELETE FROM t WHERE id=1;

  2.4 Output those events from the last one to the first one which mis-operation happened.
  For example:

client/client_priv.h

@@ -66,6 +66,10 @@ enum options_client
   OPT_MYSQLDUMP_SLAVE_APPLY,
   OPT_MYSQLDUMP_SLAVE_DATA,
   OPT_MYSQLDUMP_INCLUDE_MASTER_HOST_PORT,
+#ifdef WHEN_FLASHBACK_REVIEW_READY
+  OPT_REVIEW,
+  OPT_REVIEW_DBNAME, OPT_REVIEW_TABLENAME,
+#endif
   OPT_SLAP_CSV, OPT_SLAP_CREATE_STRING,
   OPT_SLAP_AUTO_GENERATE_SQL_LOAD_TYPE, OPT_SLAP_AUTO_GENERATE_WRITE_NUM,
   OPT_SLAP_AUTO_GENERATE_ADD_AUTO,

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

@@ -208,6 +208,9 @@ The following options may be given as the first argument:
  --extra-port=#      Extra port number to use for tcp connections in a
  one-thread-per-connection manner. 0 means don't use
  another port
+ --flashback         Setup the server to use flashback. This enables binary
+ log in row mode and will enable extra logging for DDL's
+ needed by flashback feature
  --flush             Flush MyISAM tables to disk between SQL commands
  --flush-time=#      A dedicated thread is created to flush all tables at the
  given interval
@@ -1233,6 +1236,7 @@ explicit-defaults-for-timestamp FALSE
 external-locking FALSE
 extra-max-connections 1
 extra-port 0
+flashback FALSE
 flush FALSE
 flush-time 0
 ft-boolean-syntax + -><()~*:""&|

mysql-test/suite/binlog/t/flashback-master.opt

+--flashback
+--timezone=GMT-8

mysql-test/suite/binlog/t/flashback.test

+--source include/have_log_bin.inc
+--source include/have_innodb.inc
+
+--echo #
+--echo # Preparatory cleanup.
+--echo #
+--disable_warnings
+DROP TABLE IF EXISTS t1;
+--enable_warnings
+
+--echo #
+--echo # We need a fixed timestamp to avoid varying results.
+--echo #
+SET timestamp=1000000000;
+
+--echo #
+--echo # Delete all existing binary logs.
+--echo #
+RESET MASTER;
+
+
+CREATE TABLE t1 (
+  c01 tinyint,
+  c02 smallint,
+  c03 mediumint,
+  c04 int,
+  c05 bigint,
+  c06 char(10),
+  c07 varchar(20),
+  c08 TEXT
+) ENGINE=InnoDB;
+
+--echo #
+--echo # Insert data to t1
+--echo #
+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));
+
+
+--echo #
+--echo # Update t1
+--echo #
+UPDATE t1 SET c01=100 WHERE c02=0 OR c03=3;
+
+--echo #
+--echo # Clear t1
+--echo #
+DELETE FROM t1;
+
+FLUSH LOGS;
+
+--echo #
+--echo # Show mysqlbinlog result without -B
+--echo #
+
+let $MYSQLD_DATADIR= `select @@datadir`;
+--replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+--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 # Show mysqlbinlog result with -B
+--echo #
+
+let $MYSQLD_DATADIR= `select @@datadir`;
+--replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+--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 # Insert data to t1
+--echo #
+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));
+
+--echo #
+--echo # Delete all existing binary logs.
+--echo #
+RESET MASTER;
+SELECT * FROM t1;
+
+--echo #
+--echo # Operate some data
+--echo #
+
+UPDATE t1 SET c01=20;
+UPDATE t1 SET c02=200;
+UPDATE t1 SET c03=2000;
+
+DELETE FROM t1;
+
+FLUSH LOGS;
+
+--echo #
+--echo # Flashback & Check the result
+--echo #
+
+let $MYSQLD_DATADIR= `select @@datadir`;
+--replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+--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;"
+
+SELECT * FROM t1;
+
+RESET MASTER;
+
+--echo #
+--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;
+COMMIT;
+
+FLUSH LOGS;
+
+--echo #
+--echo # Flashback & Check the result
+--echo #
+
+let $MYSQLD_DATADIR= `select @@datadir`;
+--replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+--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;"
+
+SELECT * FROM t1;
+
+DROP TABLE t1;
+
+--echo #
+--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;
+INSERT INTO t1 VALUES (1, NULL);
+INSERT INTO t1 VALUES (2, 1), (3, 2), (4, 3);
+COMMIT;
+
+SELECT * FROM t1;
+
+# New binlog
+RESET MASTER;
+
+DELETE FROM t1 ORDER BY a DESC;
+
+FLUSH LOGS;
+
+--echo #
+--echo # Flashback & Check the result
+--echo #
+
+let $MYSQLD_DATADIR= `select @@datadir`;
+--replace_result $MYSQLTEST_VARDIR MYSQLTEST_VARDIR
+--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;

mysql-test/suite/wsrep/r/binlog_format.result

@@ -8,20 +8,20 @@ Variable_name	Value
 binlog_format	ROW
 SET binlog_format=STATEMENT;
 Warnings:
-Warning	1105	MariaDB Galera does not support binlog format: STATEMENT
+Warning	1105	MariaDB Galera and flashback does not support binlog format: STATEMENT
 SHOW WARNINGS;
 Level	Code	Message
-Warning	1105	MariaDB Galera does not support binlog format: STATEMENT
+Warning	1105	MariaDB Galera and flashback does not support binlog format: STATEMENT
 SHOW VARIABLES LIKE 'binlog_format';
 Variable_name	Value
 binlog_format	STATEMENT
 CREATE TABLE IF NOT EXISTS test.t1 AS SELECT * FROM information_schema.routines WHERE 1 = 0;
 SET binlog_format=MIXED;
 Warnings:
-Warning	1105	MariaDB Galera does not support binlog format: MIXED
+Warning	1105	MariaDB Galera and flashback does not support binlog format: MIXED
 SHOW WARNINGS;
 Level	Code	Message
-Warning	1105	MariaDB Galera does not support binlog format: MIXED
+Warning	1105	MariaDB Galera and flashback does not support binlog format: MIXED
 SHOW VARIABLES LIKE 'binlog_format';
 Variable_name	Value
 binlog_format	MIXED

sql/log_event.h

@@ -41,6 +41,7 @@
 #include "rpl_utility.h"
 #include "hash.h"
 #include "rpl_tblmap.h"
+#include "sql_string.h"
 #endif
 
 #ifdef MYSQL_SERVER
@@ -52,7 +53,9 @@
 #include "rpl_gtid.h"
 
 /* Forward declarations */
+#ifndef MYSQL_CLIENT
 class String;
+#endif
 
 #define PREFIX_SQL_LOAD "SQL_LOAD-"
 #define LONG_FIND_ROW_THRESHOLD 60 /* seconds */
@@ -845,9 +848,16 @@ typedef struct st_print_event_info
   ~st_print_event_info() {
     close_cached_file(&head_cache);
     close_cached_file(&body_cache);
+#ifdef WHEN_FLASHBACK_REVIEW_READY
+    close_cached_file(&review_sql_cache);
+#endif
   }
   bool init_ok() /* tells if construction was successful */
-    { return my_b_inited(&head_cache) && my_b_inited(&body_cache); }
+    { return my_b_inited(&head_cache) && my_b_inited(&body_cache)
+#ifdef WHEN_FLASHBACK_REVIEW_READY
+      && my_b_inited(&review_sql_cache)
+#endif
+    ; }
 
 
   /* Settings on how to print the events */
@@ -875,6 +885,10 @@ typedef struct st_print_event_info
    */
   IO_CACHE head_cache;
   IO_CACHE body_cache;
+#ifdef WHEN_FLASHBACK_REVIEW_READY
+  /* Storing the SQL for reviewing */
+  IO_CACHE review_sql_cache;
+#endif
 } PRINT_EVENT_INFO;
 #endif
 
@@ -1223,6 +1237,37 @@ class Log_event
   void print_base64(IO_CACHE* file, PRINT_EVENT_INFO* print_event_info,
                     bool is_more);
 #endif
+
+  /* The following code used for Flashback */
+#ifdef MYSQL_CLIENT
+  my_bool is_flashback;
+  my_bool need_flashback_review;
+  String  output_buf; // Storing the event output
+#ifdef WHEN_FLASHBACK_REVIEW_READY
+  String  m_review_dbname;
+  String  m_review_tablename;
+
+  void set_review_dbname(const char *name)
+  {
+    if (name)
+    {
+      m_review_dbname.free();
+      m_review_dbname.append(name);
+    }
+  }
+  void set_review_tablename(const char *name)
+  {
+    if (name)
+    {
+      m_review_tablename.free();
+      m_review_tablename.append(name);
+    }
+  }
+  const char *get_review_dbname() const { return m_review_dbname.ptr(); }
+  const char *get_review_tablename() const { return m_review_tablename.ptr(); }
+#endif
+#endif
+
   /*
     read_log_event() functions read an event from a binlog or relay
     log; used by SHOW BINLOG EVENTS, the binlog_dump thread on the
@@ -4362,12 +4407,14 @@ class Rows_log_event : public Log_event
 #ifdef MYSQL_CLIENT
   /* not for direct call, each derived has its own ::print() */
   virtual void print(FILE *file, PRINT_EVENT_INFO *print_event_info)= 0;
+  void change_to_flashback_event(PRINT_EVENT_INFO *print_event_info, uchar *rows_buff, Log_event_type ev_type);
   void print_verbose(IO_CACHE *file,
                      PRINT_EVENT_INFO *print_event_info);
   size_t print_verbose_one_row(IO_CACHE *file, table_def *td,
                                PRINT_EVENT_INFO *print_event_info,
                                MY_BITMAP *cols_bitmap,
-                               const uchar *ptr, const uchar *prefix);
+                               const uchar *ptr, const uchar *prefix,
+                               const my_bool no_fill_output= 0); // if no_fill_output=1, then print result is unnecessary
 #endif
 
 #ifdef MYSQL_SERVER
@@ -4506,6 +4553,8 @@ class Rows_log_event : public Log_event
   uchar    *m_rows_cur;		/* One-after the end of the data */
   uchar    *m_rows_end;		/* One-after the end of the allocated space */
 
+  size_t   m_rows_before_size;  /* The length before m_rows_buf */
+
   flag_set m_flags;		/* Flags for row-level events */
 
   Log_event_type m_type;        /* Actual event type */
@@ -5040,6 +5089,29 @@ class Ignorable_log_event : public Log_event {
 };
 
 
+static inline bool copy_event_cache_to_string_and_reinit(IO_CACHE *cache, LEX_STRING *to)
+{
+  String tmp;
+
+  reinit_io_cache(cache, READ_CACHE, 0L, FALSE, FALSE);
+  if (tmp.append(cache, cache->end_of_file))
+    goto err;
+  reinit_io_cache(cache, WRITE_CACHE, 0, FALSE, TRUE);
+
+  /*
+    Can't change the order, because the String::release() will clear the
+    length.
+  */
+  to->length= tmp.length();
+  to->str=    tmp.release();
+
+  return false;
+
+err:
+  perror("Out of memory: can't allocate memory in copy_event_cache_to_string_and_reinit().");
+  return true;
+}
+
 static inline bool copy_event_cache_to_file_and_reinit(IO_CACHE *cache,
                                                        FILE *file)
 {

sql/mysqld.cc

@@ -393,7 +393,7 @@ bool opt_bin_log_compress;
 uint opt_bin_log_compress_min_len;
 my_bool opt_log, debug_assert_if_crashed_table= 0, opt_help= 0;
 my_bool debug_assert_on_not_freed_memory= 0;
-my_bool disable_log_notes;
+my_bool disable_log_notes, opt_support_flashback= 0;
 static my_bool opt_abort;
 ulonglong log_output_options;
 my_bool opt_userstat_running;
@@ -7413,6 +7413,10 @@ struct my_option my_long_options[]=
    0, GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
   /* We must always support the next option to make scripts like mysqltest
      easier to do */
+  {"flashback", 0,
+   "Setup the server to use flashback. This enables binary log in row mode and will enable extra logging for DDL's needed by flashback feature",
+   &opt_support_flashback, &opt_support_flashback,
+   0, GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
   {"gdb", 0,
    "Set up signals usable for debugging. Deprecated, use --debug-gdb instead.",
    &opt_debugging, &opt_debugging,
@@ -9587,6 +9591,18 @@ static int get_options(int *argc_ptr, char ***argv_ptr)
   else
     global_system_variables.option_bits&= ~OPTION_BIG_SELECTS;
 
+  if (opt_support_flashback)
+  {
+    /* Force binary logging */
+    if (!opt_bin_logname)
+      opt_bin_logname= (char*) "";                  // Use default name
+    opt_bin_log= opt_bin_log_used= 1;
+
+    /* Force format to row */
+    binlog_format_used= 1;
+    global_system_variables.binlog_format= BINLOG_FORMAT_ROW;
+  }
+
   if (!opt_bootstrap && WSREP_PROVIDER_EXISTS &&
       global_system_variables.binlog_format != BINLOG_FORMAT_ROW)
   {

sql/mysqld.h

@@ -114,6 +114,7 @@ extern uint opt_bin_log_compress_min_len;
 extern my_bool opt_log, opt_bootstrap;
 extern my_bool opt_backup_history_log;
 extern my_bool opt_backup_progress_log;
+extern my_bool opt_support_flashback;
 extern ulonglong log_output_options;
 extern ulong log_backup_output_options;
 extern my_bool opt_log_queries_not_using_indexes;

sql/sys_vars.cc

@@ -449,16 +449,17 @@ static bool binlog_format_check(sys_var *self, THD *thd, set_var *var)
   /*
     MariaDB Galera does not support STATEMENT or MIXED binlog format currently.
   */
-  if (WSREP(thd) && var->save_result.ulonglong_value != BINLOG_FORMAT_ROW)
+  if ((WSREP(thd) || opt_support_flashback) &&
+      var->save_result.ulonglong_value != BINLOG_FORMAT_ROW)
   {
     // Push a warning to the error log.
     push_warning_printf(thd, Sql_condition::WARN_LEVEL_WARN, ER_UNKNOWN_ERROR,
-                        "MariaDB Galera does not support binlog format: %s",
+                        "MariaDB Galera and flashback does not support binlog format: %s",
                         binlog_format_names[var->save_result.ulonglong_value]);
 
     if (var->type == OPT_GLOBAL)
     {
-      WSREP_ERROR("MariaDB Galera does not support binlog format: %s",
+      WSREP_ERROR("MariaDB Galera and flashback does not support binlog format: %s",
                   binlog_format_names[var->save_result.ulonglong_value]);
       return true;
     }