[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:

1 files changed, 74 insertions, 2 deletions

diff --git a/sql/log_event.h b/sql/log_event.h
index 7b8704636af..c2829a9bb10 100644
--- a/sql/log_event.h
+++ b/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 @@ public:
   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 @@ public:
 #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 @@ protected:
   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 @@ public:
 };
 
 
+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)
 {