MDEV-16355 Add option for mysqldump to read data as of specific timestamp from system-versioned tables

Based on Aleksey Midenkov's patch

mysqldump changes:

* --as-of option specifies historical point;
* query forging protection for --as-of parameter.

system versioned tables are detected by querying I_S.TABLES:

* it transfers much less data when the full table definition is not needed
* it does not give false positives on x TEXT DEFAULT 'WITH SYSTEM VERSIONING'

Author: vuvova

client/client_priv.h

@@ -104,6 +104,7 @@ enum options_client
   OPT_SHUTDOWN_WAIT_FOR_SLAVES,
   OPT_COPY_S3_TABLES,
   OPT_PRINT_TABLE_METADATA,
+  OPT_ASOF_TIMESTAMP,
   OPT_MAX_CLIENT_OPTION /* should be always the last */
 };
 

client/mysqldump.c

@@ -129,7 +129,7 @@ static my_bool  verbose= 0, opt_no_create_info= 0, opt_no_data= 0, opt_no_data_m
                 opt_include_master_host_port= 0,
                 opt_events= 0, opt_comments_used= 0,
                 opt_alltspcs=0, opt_notspcs= 0, opt_logging,
-                opt_drop_trigger= 0 ;
+                opt_drop_trigger= 0;
 #define OPT_SYSTEM_ALL 1
 #define OPT_SYSTEM_USERS 2
 #define OPT_SYSTEM_PLUGINS 4
@@ -155,7 +155,7 @@ static char  *opt_password=0,*current_user=0,
              *lines_terminated=0, *enclosed=0, *opt_enclosed=0, *escaped=0,
              *where=0, *order_by=0,
              *err_ptr= 0,
-             *log_error_file= NULL;
+             *log_error_file= NULL, *opt_asof_timestamp= NULL;
 static const char *opt_compatible_mode_str= 0;
 static char **defaults_argv= 0;
 static char compatible_mode_normal_str[255];
@@ -278,6 +278,9 @@ static struct my_option my_long_options[] =
    "Adds 'STOP SLAVE' prior to 'CHANGE MASTER' and 'START SLAVE' to bottom of dump.",
    &opt_slave_apply, &opt_slave_apply, 0, GET_BOOL, NO_ARG,
    0, 0, 0, 0, 0, 0},
+  {"as-of", OPT_ASOF_TIMESTAMP,
+   "Dump system versioned table as of specified timestamp.",
+   &opt_asof_timestamp, &opt_asof_timestamp, 0, GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
   {"character-sets-dir", OPT_CHARSETS_DIR,
    "Directory for character set files.", (char **)&charsets_dir,
    (char **)&charsets_dir, 0, GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
@@ -1325,6 +1328,12 @@ static int get_options(int *argc, char ***argv)
 	    my_progname_short);
     return(EX_USAGE);
   }
+  if (opt_asof_timestamp && strchr(opt_asof_timestamp, '\''))
+  {
+    fprintf(stderr, "%s: Incorrect DATETIME value: '%s'\n",
+            my_progname_short, opt_asof_timestamp);
+    return(EX_USAGE);
+  }
   if (strcmp(default_charset, MYSQL_AUTODETECT_CHARSET_NAME) &&
       !(charset_info= get_charset_by_csname(default_charset,
                                             MY_CS_PRIMARY,
@@ -3039,7 +3048,7 @@ static void get_sequence_structure(const char *seq, const char *db)
 */
 
 static uint get_table_structure(const char *table, const char *db, char *table_type,
-                                char *ignore_flag)
+                                char *ignore_flag, my_bool *versioned)
 {
   my_bool    init=0, delayed, write_data, complete_insert;
   my_ulonglong num_fields;
@@ -3106,6 +3115,26 @@ static uint get_table_structure(const char *table, const char *db, char *table_t
 
   verbose_msg("-- Retrieving table structure for table %s...\n", table);
 
+  if (versioned)
+  {
+    if (!opt_asof_timestamp)
+      versioned= NULL;
+    else
+    {
+      my_snprintf(query_buff, sizeof(query_buff), "select 1 from"
+                  " information_schema.tables where table_schema=database()"
+                  " and table_name=%s and table_type='SYSTEM VERSIONED'",
+                  quote_for_equal(table, table_buff));
+      if (!mysql_query_with_error_report(mysql, &result, query_buff))
+      {
+        *versioned= result->row_count > 0;
+        mysql_free_result(result);
+      }
+      else
+        *versioned= 0;
+    }
+  }
+
   len= my_snprintf(query_buff, sizeof(query_buff),
                    "SET SQL_QUOTE_SHOW_CREATE=%d",
                    (opt_quoted || opt_keywords));
@@ -3991,6 +4020,15 @@ static char *alloc_query_str(size_t size)
 }
 
 
+static void vers_append_system_time(DYNAMIC_STRING* query_string)
+{
+  DBUG_ASSERT(opt_asof_timestamp);
+  dynstr_append_checked(query_string, " FOR SYSTEM_TIME AS OF TIMESTAMP '");
+  dynstr_append_checked(query_string, opt_asof_timestamp);
+  dynstr_append_checked(query_string, "'");
+}
+
+
 /*
 
  SYNOPSIS
@@ -4018,6 +4056,7 @@ static void dump_table(const char *table, const char *db, const uchar *hash_key,
   ulong         rownr, row_break;
   uint num_fields;
   size_t total_length, init_length;
+  my_bool versioned= 0;
 
   MYSQL_RES     *res;
   MYSQL_FIELD   *field;
@@ -4028,7 +4067,7 @@ static void dump_table(const char *table, const char *db, const uchar *hash_key,
     Make sure you get the create table info before the following check for
     --no-data flag below. Otherwise, the create table info won't be printed.
   */
-  num_fields= get_table_structure(table, db, table_type, &ignore_flag);
+  num_fields= get_table_structure(table, db, table_type, &ignore_flag, &versioned);
 
   /*
     The "table" could be a view.  If so, we don't do anything here.
@@ -4135,6 +4174,8 @@ static void dump_table(const char *table, const char *db, const uchar *hash_key,
 
     dynstr_append_checked(&query_string, " FROM ");
     dynstr_append_checked(&query_string, result_table);
+    if (versioned)
+      vers_append_system_time(&query_string);
 
     if (where)
     {
@@ -4167,6 +4208,8 @@ static void dump_table(const char *table, const char *db, const uchar *hash_key,
     dynstr_append_checked(&query_string, select_field_names.str);
     dynstr_append_checked(&query_string, " FROM ");
     dynstr_append_checked(&query_string, result_table);
+    if (versioned)
+      vers_append_system_time(&query_string);
 
     if (where)
     {
@@ -5637,21 +5680,21 @@ static int dump_all_tables_in_db(char *database)
     if (general_log_table_exists)
     {
       if (!get_table_structure((char *) "general_log",
-                               database, table_type, &ignore_flag) )
+                               database, table_type, &ignore_flag, NULL) )
         verbose_msg("-- Warning: get_table_structure() failed with some internal "
                     "error for 'general_log' table\n");
     }
     if (slow_log_table_exists)
     {
       if (!get_table_structure((char *) "slow_log",
-                               database, table_type, &ignore_flag) )
+                               database, table_type, &ignore_flag, NULL) )
         verbose_msg("-- Warning: get_table_structure() failed with some internal "
                     "error for 'slow_log' table\n");
     }
     if (transaction_registry_table_exists)
     {
       if (!get_table_structure((char *) "transaction_registry",
-                               database, table_type, &ignore_flag) )
+                               database, table_type, &ignore_flag, NULL) )
         verbose_msg("-- Warning: get_table_structure() failed with some internal "
                     "error for 'transaction_registry' table\n");
     }

mysql-test/suite/versioning/r/data.result

@@ -0,0 +1,50 @@
+#
+# MDEV-16355 Add option for mysqldump to read data as of specific timestamp from system-versioned tables
+#
+create or replace table t1 (x int) with system versioning;
+set timestamp=unix_timestamp('1990-01-01 00:00');
+insert t1 (x) values (1),(2),(3);
+set timestamp=unix_timestamp('1990-08-03 00:00');
+delete from t1 where x=1;
+set timestamp=unix_timestamp('1991-01-02 00:00');
+delete from t1 where x=2;
+set timestamp=default;
+#MYSQL_DUMP --compact test
+/*!40101 SET @saved_cs_client     = @@character_set_client */;
+/*!40101 SET character_set_client = utf8 */;
+CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING;
+/*!40101 SET character_set_client = @saved_cs_client */;
+INSERT INTO `t1` VALUES (3);
+#MYSQL_DUMP --compact --as-of="1990-01-02 00:00" test
+/*!40101 SET @saved_cs_client     = @@character_set_client */;
+/*!40101 SET character_set_client = utf8 */;
+CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING;
+/*!40101 SET character_set_client = @saved_cs_client */;
+INSERT INTO `t1` VALUES (1),(2),(3);
+#MYSQL_DUMP --compact --as-of="1990-08-02 00:00" --databases test
+
+CREATE DATABASE /*!32312 IF NOT EXISTS*/ `test` /*!40100 DEFAULT CHARACTER SET latin1 */;
+
+USE `test`;
+/*!40101 SET @saved_cs_client     = @@character_set_client */;
+/*!40101 SET character_set_client = utf8 */;
+CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING;
+/*!40101 SET character_set_client = @saved_cs_client */;
+INSERT INTO `t1` VALUES (1),(2),(3);
+#MYSQL_DUMP --compact --as-of="1990-08-04 00:00" test t1
+/*!40101 SET @saved_cs_client     = @@character_set_client */;
+/*!40101 SET character_set_client = utf8 */;
+CREATE TABLE `t1` (
+  `x` int(11) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1 WITH SYSTEM VERSIONING;
+/*!40101 SET character_set_client = @saved_cs_client */;
+INSERT INTO `t1` VALUES (2),(3);
+#MYSQL_DUMP --compact --as-of="1990-08-04 00:00' where 'abc" test 2>&1
+mysqldump: Incorrect DATETIME value: '1990-08-04 00:00' where 'abc'
+drop tables t1;

mysql-test/suite/versioning/t/data.test

@@ -0,0 +1,32 @@
+--source include/not_embedded.inc
+--source suite/versioning/common.inc
+
+--echo #
+--echo # MDEV-16355 Add option for mysqldump to read data as of specific timestamp from system-versioned tables
+--echo #
+create or replace table t1 (x int) with system versioning;
+set timestamp=unix_timestamp('1990-01-01 00:00');
+insert t1 (x) values (1),(2),(3);
+set timestamp=unix_timestamp('1990-08-03 00:00');
+delete from t1 where x=1;
+set timestamp=unix_timestamp('1991-01-02 00:00');
+delete from t1 where x=2;
+set timestamp=default;
+
+--echo #MYSQL_DUMP --compact test
+--exec $MYSQL_DUMP --compact test
+--echo #MYSQL_DUMP --compact --as-of="1990-01-02 00:00" test
+--exec $MYSQL_DUMP --compact --as-of="1990-01-02 00:00" test
+--echo #MYSQL_DUMP --compact --as-of="1990-08-02 00:00" --databases test
+--exec $MYSQL_DUMP --compact --as-of="1990-08-02 00:00" --databases test
+--echo #MYSQL_DUMP --compact --as-of="1990-08-04 00:00" test t1
+--exec $MYSQL_DUMP --compact --as-of="1990-08-04 00:00" test t1
+## Forged query protection
+--echo #MYSQL_DUMP --compact --as-of="1990-08-04 00:00' where 'abc" test 2>&1
+--replace_result mysqldump.exe mysqldump
+--error 1
+--exec $MYSQL_DUMP --compact --as-of="1990-08-04 00:00' where 'abc" test 2>&1
+
+drop tables t1;
+
+--source suite/versioning/common_finish.inc