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

2 files changed, 251 insertions, 4 deletions

diff --git a/client/client_priv.h b/client/client_priv.h
index 1d85791fa73..e96e187fb34 100644
--- a/client/client_priv.h
+++ b/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,
diff --git a/client/mysqlbinlog.cc b/client/mysqlbinlog.cc
index 558dbebc89c..39a879bc619 100644
--- a/client/mysqlbinlog.cc
+++ b/client/mysqlbinlog.cc
@@ -66,6 +66,10 @@ Rpl_filter *binlog_filter= 0;
 /* Needed for Rpl_filter */
 CHARSET_INFO* system_charset_info= &my_charset_utf8_general_ci;
 
+/* Needed for Flashback */
+DYNAMIC_ARRAY binlog_events; // Storing the events output string
+String stop_event_string; // Storing the STOP_EVENT output string
+
 char server_version[SERVER_VERSION_LENGTH];
 ulong server_id = 0;
 
@@ -89,7 +93,7 @@ static const char *load_groups[]=
 static void error(const char *format, ...) ATTRIBUTE_FORMAT(printf, 1, 2);
 static void warning(const char *format, ...) ATTRIBUTE_FORMAT(printf, 1, 2);
 
-static bool one_database=0, to_last_remote_log= 0, disable_log_bin= 0;
+static bool one_database=0, one_table=0, to_last_remote_log= 0, disable_log_bin= 0;
 static bool opt_hexdump= 0, opt_version= 0;
 const char *base64_output_mode_names[]=
 {"NEVER", "AUTO", "ALWAYS", "UNSPEC", "DECODE-ROWS", NullS};
@@ -99,6 +103,7 @@ TYPELIB base64_output_mode_typelib=
 static enum_base64_output_mode opt_base64_output_mode= BASE64_OUTPUT_UNSPEC;
 static char *opt_base64_output_mode_str= NullS;
 static char* database= 0;
+static char* table= 0;
 static my_bool force_opt= 0, short_form= 0, remote_opt= 0;
 static my_bool debug_info_flag, debug_check_flag;
 static my_bool force_if_open_opt= 1;
@@ -132,6 +137,12 @@ static MYSQL* mysql = NULL;
 static const char* dirname_for_local_load= 0;
 static bool opt_skip_annotate_row_events= 0;
 
+static my_bool opt_flashback;
+#ifdef WHEN_FLASHBACK_REVIEW_READY
+static my_bool opt_flashback_review;
+static char *flashback_review_dbname, *flashback_review_tablename;
+#endif
+
 /**
   Pointer to the Format_description_log_event of the currently active binlog.
 
@@ -791,6 +802,23 @@ print_skip_replication_statement(PRINT_EVENT_INFO *pinfo, const Log_event *ev)
 }
 
 /**
+  Indicates whether the given table should be filtered out,
+  according to the --table=X option.
+
+  @param log_tblname Name of table.
+
+  @return nonzero if the table with the given name should be
+  filtered out, 0 otherwise.
+*/
+static bool shall_skip_table(const char *log_tblname)
+{
+  return one_table &&
+         (log_tblname != NULL) &&
+         strcmp(log_tblname, table);
+}
+
+
+/**
   Prints the given event in base64 format.
 
   The header is printed to the head cache and the body is printed to
@@ -952,6 +980,12 @@ Exit_status process_event(PRINT_EVENT_INFO *print_event_info, Log_event *ev,
   Exit_status retval= OK_CONTINUE;
   IO_CACHE *const head= &print_event_info->head_cache;
 
+  /* Bypass flashback settings to event */
+  ev->is_flashback= opt_flashback;
+#ifdef WHEN_FLASHBACK_REVIEW_READY
+  ev->need_flashback_review= opt_flashback_review;
+#endif
+
   /*
     Format events are not concerned by --offset and such, we always need to
     read them to be able to process the wanted events.
@@ -988,7 +1022,7 @@ Exit_status process_event(PRINT_EVENT_INFO *print_event_info, Log_event *ev,
       retval= OK_STOP;
       goto end;
     }
-    if (!short_form)
+    if (!short_form && !opt_flashback)
       fprintf(result_file, "# at %s\n",llstr(pos,ll_buff));
 
     if (!opt_hexdump)
@@ -1214,12 +1248,128 @@ Exit_status process_event(PRINT_EVENT_INFO *print_event_info, Log_event *ev,
     case TABLE_MAP_EVENT:
     {
       Table_map_log_event *map= ((Table_map_log_event *)ev);
-      if (shall_skip_database(map->get_db_name()))
+      if (shall_skip_database(map->get_db_name()) ||
+          shall_skip_table(map->get_table_name()))
       {
         print_event_info->m_table_map_ignored.set_table(map->get_table_id(), map);
         destroy_evt= FALSE;
         goto end;
       }
+#ifdef WHEN_FLASHBACK_REVIEW_READY
+      /* Create review table for Flashback */
+      if (opt_flashback_review)
+      {
+        // Check if the table was already created?
+        Table_map_log_event *exist_table;
+        exist_table= print_event_info->m_table_map.get_table(map->get_table_id());
+
+        if (!exist_table)
+        {
+
+          MYSQL *conn;
+          MYSQL_RES *res;
+          MYSQL_ROW row;
+          char tmp_sql[8096];
+          int  tmp_sql_offset;
+
+          conn = mysql_init(NULL);
+          if (!mysql_real_connect(conn, host, user, pass,
+                map->get_db_name(), port, sock, 0))
+          {
+            fprintf(stderr, "%s\n", mysql_error(conn));
+            exit(1);
+          }
+
+          if (mysql_query(conn, "SET group_concat_max_len=10000;"))
+          {
+            fprintf(stderr, "%s\n", mysql_error(conn));
+            exit(1);
+          }
+
+          memset(tmp_sql, 0, sizeof(tmp_sql));
+          sprintf(tmp_sql, " "
+                  "SELECT Group_concat(cols) "
+                  "FROM   (SELECT 'op_type char(1)' cols "
+                  "  UNION ALL "
+                  "  SELECT Concat('`', column_name, '_old` ', column_type, ' ', "
+                  "    IF(character_set_name IS NOT NULL, "
+                  "      Concat('character set ', character_set_name, ' '), ' '), "
+                  "    IF(collation_name IS NOT NULL, "
+                  "      Concat('collate ', collation_name, ' '), ' ')) cols "
+                  "  FROM   information_schema.columns "
+                  "  WHERE  table_schema = '%s' "
+                  "  AND table_name = '%s' "
+                  "  UNION ALL "
+                  "  SELECT Concat('`', column_name, '_new` ', column_type, ' ', "
+                  "    IF(character_set_name IS NOT NULL, "
+                  "      Concat('character set ', character_set_name, ' '), ' '), "
+                  "    IF(collation_name IS NOT NULL, "
+                  "      Concat('collate ', collation_name, ' '), ' ')) cols "
+                  "  FROM   information_schema.columns "
+                  "  WHERE  table_schema = '%s' "
+                  "  AND table_name = '%s') tmp;",
+            map->get_db_name(), map->get_table_name(),
+            map->get_db_name(), map->get_table_name());
+
+          if (mysql_query(conn, tmp_sql))
+          {
+            fprintf(stderr, "%s\n", mysql_error(conn));
+            exit(1);
+          }
+          res = mysql_use_result(conn);
+          if ((row = mysql_fetch_row(res)) != NULL)  // only one row
+          {
+            if (flashback_review_dbname)
+            {
+              ev->set_flashback_review_dbname(flashback_review_dbname);
+            }
+            else
+            {
+              ev->set_flashback_review_dbname(map->get_db_name());
+            }
+            if (flashback_review_tablename)
+            {
+              ev->set_flashback_review_tablename(flashback_review_tablename);
+            }
+            else
+            {
+              memset(tmp_sql, 0, sizeof(tmp_sql));
+              sprintf(tmp_sql, "__%s", map->get_table_name());
+              ev->set_flashback_review_tablename(tmp_sql);
+            }
+            memset(tmp_sql, 0, sizeof(tmp_sql));
+            tmp_sql_offset= sprintf(tmp_sql, "CREATE TABLE IF NOT EXISTS");
+            tmp_sql_offset+= sprintf(tmp_sql + tmp_sql_offset, " `%s`.`%s` (%s) %s",
+                                     ev->get_flashback_review_dbname(),
+                                     ev->get_flashback_review_tablename(),
+                                     row[0],
+                                     print_event_info->delimiter);
+          }
+          fprintf(result_file, "%s\n", tmp_sql);
+          mysql_free_result(res);
+          mysql_close(conn);
+        }
+        else
+        {
+          char tmp_str[128];
+
+          if (flashback_review_dbname)
+            ev->set_flashback_review_dbname(flashback_review_dbname);
+          else
+            ev->set_flashback_review_dbname(map->get_db_name());
+
+          if (flashback_review_tablename)
+            ev->set_flashback_review_tablename(flashback_review_tablename);
+          else
+          {
+            memset(tmp_str, 0, sizeof(tmp_str));
+            sprintf(tmp_str, "__%s", map->get_table_name());
+            ev->set_flashback_review_tablename(tmp_str);
+          }
+        }
+      }
+#endif
+
       /*
         The Table map is to be printed, so it's just the time when we may
         print the kept Annotate event (if there is any).
@@ -1294,6 +1444,38 @@ end:
   */
   if (ev)
   {
+    /* Holding event output if needed */
+    if (!ev->output_buf.is_empty())
+    {
+      LEX_STRING tmp_str;
+
+      tmp_str.length= ev->output_buf.length();
+      tmp_str.str=    ev->output_buf.release();
+
+      if (opt_flashback)
+      {
+        if (ev_type == STOP_EVENT)
+          stop_event_string.reset(tmp_str.str, tmp_str.length, tmp_str.length,
+                                  &my_charset_bin);
+        else
+        {
+          if (push_dynamic(&binlog_events, (uchar *) &tmp_str))
+          {
+            error("Out of memory: can't allocate memory to store the flashback events.");
+            exit(1);
+          }
+        }
+      }
+      else
+      {
+        my_fwrite(result_file, (const uchar *) tmp_str.str, tmp_str.length,
+                  MYF(MY_NABP));
+        my_free(tmp_str.str);
+      }
+    }
+
+    if (remote_opt)
+      ev->temp_buf= 0;
     if (destroy_evt) /* destroy it later if not set (ignored table map) */
       delete ev;
   }
@@ -1352,6 +1534,13 @@ static struct my_option my_options[] =
     "already have. NOTE: you will need a SUPER privilege to use this option.",
    &disable_log_bin, &disable_log_bin, 0, GET_BOOL,
    NO_ARG, 0, 0, 0, 0, 0, 0},
+  {"flashback", 'B', "Flashback feature can rollback you committed data to a special time point.",
+#ifdef WHEN_FLASHBACK_REVIEW_READY
+   "before Flashback feature writing a row, original row can insert to review-dbname.review-tablename,"
+   "and mysqlbinlog will login mysql by user(-u) and password(-p) and host(-h).",
+#endif
+   &opt_flashback, &opt_flashback, 0, GET_BOOL, NO_ARG, 0, 0, 0, 0,
+   0, 0},
   {"force-if-open", 'F', "Force if binlog was not closed properly.",
    &force_if_open_opt, &force_if_open_opt, 0, GET_BOOL, NO_ARG,
    1, 0, 0, 0, 0, 0},
@@ -1395,6 +1584,19 @@ static struct my_option my_options[] =
    "prefix for the file names.",
    &result_file_name, &result_file_name, 0, GET_STR, REQUIRED_ARG,
    0, 0, 0, 0, 0, 0},
+#ifdef WHEN_FLASHBACK_REVIEW_READY
+  {"review", opt_flashback_review, "Print review sql in output file.",
+   &opt_flashback_review, &opt_flashback_review, 0, GET_BOOL, NO_ARG, 0, 0, 0, 0,
+   0, 0},
+  {"review-dbname", opt_flashback_flashback_review_dbname,
+   "Writing flashback original row data into this db",
+   &flashback_review_dbname, &flashback_review_dbname,
+   0, GET_STR_ALLOC, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
+  {"review-tablename", opt_flashback_flashback_review_tablename,
+   "Writing flashback original row data into this table",
+   &flashback_review_tablename, &flashback_review_tablename,
+   0, GET_STR_ALLOC, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
+#endif
   {"server-id", 0,
    "Extract only binlog entries created by the server having the given id.",
    &server_id, &server_id, 0, GET_ULONG,
@@ -1458,6 +1660,9 @@ static struct my_option my_options[] =
    &stop_position, &stop_position, 0, GET_ULL,
    REQUIRED_ARG, (longlong)(~(my_off_t)0), BIN_LOG_HEADER_SIZE,
    (ulonglong)(~(my_off_t)0), 0, 0, 0},
+  {"table", 'T', "List entries for just this table (local log only).",
+   &table, &table, 0, GET_STR_ALLOC, REQUIRED_ARG,
+   0, 0, 0, 0, 0, 0},
   {"to-last-log", 't', "Requires -R. Will not stop at the end of the \
 requested binlog but rather continue printing until the end of the last \
 binlog of the MySQL server. If you send the output to the same MySQL server, \
@@ -1567,6 +1772,7 @@ static void cleanup()
 {
   my_free(pass);
   my_free(database);
+  my_free(table);
   my_free(host);
   my_free(user);
   my_free(const_cast<char*>(dirname_for_local_load));
@@ -1637,6 +1843,9 @@ get_one_option(int optid, const struct my_option *opt __attribute__((unused)),
     break;
 #endif
 #include <sslopt-case.h>
+  case 'B':
+    opt_flashback= 1;
+    break;
   case 'd':
     one_database = 1;
     break;
@@ -1658,10 +1867,18 @@ get_one_option(int optid, const struct my_option *opt __attribute__((unused)),
   case 'R':
     remote_opt= 1;
     break;
+  case 'T':
+    one_table= 1;
+    break;
   case OPT_MYSQL_PROTOCOL:
     opt_protocol= find_type_or_exit(argument, &sql_protocol_typelib,
                                     opt->name);
     break;
+#ifdef WHEN_FLASHBACK_REVIEW_READY
+  case opt_flashback_review:
+    opt_flashback_review= 1;
+    break;
+#endif
   case OPT_START_DATETIME:
     start_datetime= convert_str_to_timestamp(start_datetime_str);
     break;
@@ -1862,7 +2079,7 @@ static Exit_status dump_log_entries(const char* logname)
        dump_local_log_entries(&print_event_info, logname));
 
   /* Set delimiter back to semicolon */
-  if (!opt_raw_mode)
+  if (!opt_raw_mode && !opt_flashback)
     fprintf(result_file, "DELIMITER ;\n");
   strmov(print_event_info.delimiter, ";");
   return rc;
@@ -2660,6 +2877,9 @@ int main(int argc, char** argv)
   DBUG_ENTER("main");
   DBUG_PROCESS(argv[0]);
 
+  (void) my_init_dynamic_array(&binlog_events, sizeof(LEX_STRING), 1024, 1024,
+                               MYF(0));
+
   my_init_time(); // for time functions
   tzset(); // set tzname
 
@@ -2795,6 +3015,29 @@ int main(int argc, char** argv)
     start_position= BIN_LOG_HEADER_SIZE;
   }
 
+  /*
+    If enable flashback, need to print the events from the end to the
+    beginning
+  */
+  if (opt_flashback)
+  {
+    for (uint i= binlog_events.elements; i > 0; --i)
+    {
+      LEX_STRING *event_str= dynamic_element(&binlog_events, i - 1,
+                                             LEX_STRING*);
+      fprintf(result_file, "%s", event_str->str);
+      my_free(event_str->str);
+    }
+    fprintf(result_file, "COMMIT\n/*!*/;\n");
+  }
+  delete_dynamic(&binlog_events);
+
+  /* Set delimiter back to semicolon */
+  if (!stop_event_string.is_empty())
+    fprintf(result_file, "%s", stop_event_string.ptr());
+  if (!opt_raw_mode && opt_flashback)
+    fprintf(result_file, "DELIMITER ;\n");
+
   if (!opt_raw_mode)
   {
     /*