Bug#14871 mysqldump: invalid view dump output

 - Add comments with embeded veriosn info around the parts of the view syntax that are only supported by a certain version of MySQL Server


client/mysqldump.c:
  Use information_schema.views to gather information about the view, then replace some parts of the output from "SHOW CREATE VIEW" with comment markers with version, to make thos parts of the view syntax become parsed only of MySQL servers that supports it.
  Create common function "open_sql_file_for_table" to open the individual .sql file where to dump the table or view.
mysql-test/r/mysqldump.result:
  Update results
mysql-test/t/mysqldump.test:
  Add test to see that views can be deumped and reloaded alos when they contain "SECURITY TYPE", "CHECK OPTION" and "DEFINER"

client/mysqldump.c

@@ -851,6 +851,27 @@ static int mysql_query_with_error_report(MYSQL *mysql_con, MYSQL_RES **res,
   return 0;
 }
 
+/*
+  Open a new .sql file to dump the table or view into
+
+  SYNOPSIS
+    open_sql_file_for_table
+    name      name of the table or view
+
+  RETURN VALUES
+    0        Failed to open file
+    > 0      Handle of the open file
+*/
+static FILE* open_sql_file_for_table(const char* table)
+{
+  FILE* res;
+  char filename[FN_REFLEN], tmp_path[FN_REFLEN];
+  convert_dirname(tmp_path,path,NullS);
+  res= my_fopen(fn_format(filename, table, tmp_path, ".sql", 4),
+		O_WRONLY, MYF(MY_WME));
+  return res;
+}
+
 
 static void safe_exit(int error)
 {
@@ -1402,11 +1423,7 @@ static uint get_table_structure(char *table, char *db, char *table_type,
 
       if (path)
       {
-        char filename[FN_REFLEN], tmp_path[FN_REFLEN];
-        convert_dirname(tmp_path,path,NullS);
-        sql_file= my_fopen(fn_format(filename, table, tmp_path, ".sql", 4),
-				 O_WRONLY, MYF(MY_WME));
-        if (!sql_file)			/* If file couldn't be opened */
+        if (!(sql_file= open_sql_file_for_table(table)))
         {
 	  safe_exit(EX_MYSQLERR);
 	  DBUG_RETURN(0);
@@ -1568,11 +1585,7 @@ static uint get_table_structure(char *table, char *db, char *table_type,
     {
       if (path)
       {
-        char filename[FN_REFLEN], tmp_path[FN_REFLEN];
-        convert_dirname(tmp_path,path,NullS);
-        sql_file= my_fopen(fn_format(filename, table, tmp_path, ".sql", 4),
-				 O_WRONLY, MYF(MY_WME));
-        if (!sql_file)			/* If file couldn't be opened */
+        if (!(sql_file= open_sql_file_for_table(table)))
         {
 	  safe_exit(EX_MYSQLERR);
 	  DBUG_RETURN(0);
@@ -3202,6 +3215,41 @@ cleanup:
 }
 
 
+/*
+  Replace a substring
+
+  SYNOPSIS
+    replace
+    ds_str      The string to search and perform the replace in
+    search_str  The string to search for
+    search_len  Length of the string to search for
+    replace_str The string to replace with
+    replace_len Length of the string to replace with
+
+  RETURN
+    0 String replaced
+    1 Could not find search_str in str
+*/
+
+static int replace(DYNAMIC_STRING *ds_str,
+                   const char *search_str, ulong search_len,
+                   const char *replace_str, ulong replace_len)
+{
+  const char *start= strstr(ds_str->str, search_str);
+  if (!start)
+    return 1;
+  DYNAMIC_STRING ds_tmp;
+  init_dynamic_string(&ds_tmp, "",
+		      ds_str->length + replace_len, 256);
+  dynstr_append_mem(&ds_tmp, ds_str->str, start - ds_str->str);
+  dynstr_append_mem(&ds_tmp, replace_str, replace_len);
+  dynstr_append(&ds_tmp, start + search_len);
+  dynstr_set(ds_str, ds_tmp.str);
+  dynstr_free(&ds_tmp);
+  return 0;
+}
+
+
 /*
   Getting VIEW structure
 
@@ -3223,11 +3271,11 @@ static my_bool get_view_structure(char *table, char* db)
   char	     *result_table, *opt_quoted_table;
   char	     table_buff[NAME_LEN*2+3];
   char	     table_buff2[NAME_LEN*2+3];
-  char       buff[20+FN_REFLEN];
+  char       query[QUERY_LENGTH];
   FILE       *sql_file = md_result_file;
   DBUG_ENTER("get_view_structure");
 
-  if (tFlag)
+  if (tFlag) /* Don't write table creation info */
     DBUG_RETURN(0);
 
   if (verbose)
@@ -3241,36 +3289,32 @@ static my_bool get_view_structure(char *table, char* db)
   result_table=     quote_name(table, table_buff, 1);
   opt_quoted_table= quote_name(table, table_buff2, 0);
 
-  sprintf(buff,"show create table %s", result_table);
-  if (mysql_query(sock, buff))
+  snprintf(query, sizeof(query), "SHOW CREATE TABLE %s", result_table);
+  if (mysql_query_with_error_report(sock, &table_res, query))
   {
-    fprintf(stderr, "%s: Can't get CREATE TABLE for view %s (%s)\n",
-            my_progname, result_table, mysql_error(sock));
     safe_exit(EX_MYSQLERR);
     DBUG_RETURN(0);
   }
 
+  /* Check if this is a view */
+  field= mysql_fetch_field_direct(table_res, 0);
+  if (strcmp(field->name, "View") != 0)
+  {
+    if (verbose)
+      fprintf(stderr, "-- It's base table, skipped\n");
+    DBUG_RETURN(0);
+  }
+
+  /* If requested, open separate .sql file for this view */
   if (path)
   {
-    char filename[FN_REFLEN], tmp_path[FN_REFLEN];
-    convert_dirname(tmp_path,path,NullS);
-    sql_file= my_fopen(fn_format(filename, table, tmp_path, ".sql", 4),
-                       O_WRONLY, MYF(MY_WME));
-    if (!sql_file)			/* If file couldn't be opened */
+    if (!(sql_file= open_sql_file_for_table(table)))
     {
       safe_exit(EX_MYSQLERR);
       DBUG_RETURN(1);
     }
     write_header(sql_file, db);
   }
-  table_res= mysql_store_result(sock);
-  field= mysql_fetch_field_direct(table_res, 0);
-  if (strcmp(field->name, "View") != 0)
-  {
-    if (verbose)
-      fprintf(stderr, "-- It's base table, skipped\n");
-    DBUG_RETURN(0);
-  }
 
   if (!opt_xml && opt_comments)
   {
@@ -3287,11 +3331,102 @@ static my_bool get_view_structure(char *table, char* db)
     check_io(sql_file);
   }
 
-  row= mysql_fetch_row(table_res);
-  fprintf(sql_file, "/*!50001 %s*/;\n", row[1]);
-  check_io(sql_file);
-  mysql_free_result(table_res);
 
+  snprintf(query, sizeof(query),
+           "SELECT CHECK_OPTION, DEFINER, SECURITY_TYPE "               \
+           "FROM information_schema.views "                             \
+           "WHERE table_name=\"%s\" AND table_schema=\"%s\"", table, db);
+  if (mysql_query(sock, query))
+  {
+    /*
+      Use the raw output from SHOW CREATE TABLE if
+       information_schema query fails.
+     */
+    row= mysql_fetch_row(table_res);
+    fprintf(sql_file, "/*!50001 %s */;\n", row[1]);
+    check_io(sql_file);
+    mysql_free_result(table_res);
+  }
+  else
+  {
+    char *ptr;
+    ulong *lengths;
+    char search_buf[256], replace_buf[256];
+    ulong search_len, replace_len;
+    DYNAMIC_STRING ds_view;
+
+    /* Save the result of SHOW CREATE TABLE in ds_view */
+    row= mysql_fetch_row(table_res);
+    lengths= mysql_fetch_lengths(table_res);
+    init_dynamic_string(&ds_view, row[1], lengths[1] + 1, 1024);
+    mysql_free_result(table_res);
+
+    /* Get the result from "select ... information_schema" */
+    if (!(table_res= mysql_store_result(sock)))
+    {
+      safe_exit(EX_MYSQLERR);
+      DBUG_RETURN(1);
+    }
+    row= mysql_fetch_row(table_res);
+    lengths= mysql_fetch_lengths(table_res);
+
+    /*
+      "WITH %s CHECK OPTION" is available from 5.0.2
+      Surround it with !50002 comments
+    */
+    if (strcmp(row[0], "NONE"))
+    {
+
+      ptr= search_buf;
+      search_len= (ulong)(strxmov(ptr, "WITH ", row[0],
+                                  " CHECK OPTION", NullS) - ptr);
+      ptr= replace_buf;
+      replace_len=(ulong)(strxmov(ptr, "*/\n/*!50002 WITH ", row[0],
+                                  " CHECK OPTION", NullS) - ptr);
+      replace(&ds_view, search_buf, search_len, replace_buf, replace_len);
+    }
+
+    /*
+      "DEFINER=%s SQL SECURITY %s" is available from 5.0.13
+      Surround it with !50013 comments
+    */
+    {
+      uint       user_name_len;
+      char       user_name_str[USERNAME_LENGTH + 1];
+      char       quoted_user_name_str[USERNAME_LENGTH * 2 + 3];
+      uint       host_name_len;
+      char       host_name_str[HOSTNAME_LENGTH + 1];
+      char       quoted_host_name_str[HOSTNAME_LENGTH * 2 + 3];
+
+      parse_user(row[1], lengths[1], user_name_str, &user_name_len,
+		 host_name_str, &host_name_len);
+
+      ptr= search_buf;
+      search_len=
+        (ulong)(strxmov(ptr, "DEFINER=",
+                        quote_name(user_name_str, quoted_user_name_str, FALSE),
+                        "@",
+                        quote_name(host_name_str, quoted_host_name_str, FALSE),
+                        " SQL SECURITY ", row[2], NullS) - ptr);
+      ptr= replace_buf;
+      replace_len=
+        (ulong)(strxmov(ptr, "*/\n/*!50013 DEFINER=",
+                        quote_name(user_name_str, quoted_user_name_str, FALSE),
+                        "@",
+                        quote_name(host_name_str, quoted_host_name_str, FALSE),
+                        " SQL SECURITY ", row[2],
+                        " */\n/*!50001", NullS) - ptr);
+      replace(&ds_view, search_buf, search_len, replace_buf, replace_len);
+    }
+
+    /* Dump view structure to file */
+    fprintf(sql_file, "/*!50001 %s */;\n", ds_view.str);
+    check_io(sql_file);
+    mysql_free_result(table_res);
+    dynstr_free(&ds_view);
+  }
+
+  /* If a separate .sql file was opened, close it now */
   if (sql_file != md_result_file)
   {
     fputs("\n", sql_file);

mysql-test/r/mysqldump.result

@@ -1464,7 +1464,10 @@ DROP TABLE IF EXISTS `v2`;
 ) */;
 /*!50001 DROP TABLE IF EXISTS `v2`*/;
 /*!50001 DROP VIEW IF EXISTS `v2`*/;
-/*!50001 CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `v2` AS select `t2`.`a` AS `a` from `t2` where (`t2`.`a` like _latin1'a%') WITH CASCADED CHECK OPTION*/;
+/*!50001 CREATE ALGORITHM=UNDEFINED */
+/*!50013 DEFINER=`root`@`localhost` SQL SECURITY DEFINER */
+/*!50001 VIEW `v2` AS select `t2`.`a` AS `a` from `t2` where (`t2`.`a` like _latin1'a%') */
+/*!50002 WITH CASCADED CHECK OPTION */;
 /*!40103 SET TIME_ZONE=@OLD_TIME_ZONE */;
 
 /*!40101 SET SQL_MODE=@OLD_SQL_MODE */;
@@ -1728,7 +1731,9 @@ DROP TABLE IF EXISTS `v1`;
 ) */;
 /*!50001 DROP TABLE IF EXISTS `v1`*/;
 /*!50001 DROP VIEW IF EXISTS `v1`*/;
-/*!50001 CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `v1` AS select `t1`.`a` AS `a` from `t1`*/;
+/*!50001 CREATE ALGORITHM=UNDEFINED */
+/*!50013 DEFINER=`root`@`localhost` SQL SECURITY DEFINER */
+/*!50001 VIEW `v1` AS select `t1`.`a` AS `a` from `t1` */;
 /*!40103 SET TIME_ZONE=@OLD_TIME_ZONE */;
 
 /*!40101 SET SQL_MODE=@OLD_SQL_MODE */;
@@ -1784,7 +1789,10 @@ DROP TABLE IF EXISTS `v2`;
 ) */;
 /*!50001 DROP TABLE IF EXISTS `v2`*/;
 /*!50001 DROP VIEW IF EXISTS `v2`*/;
-/*!50001 CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `v2` AS select `t2`.`a` AS `a` from `t2` where (`t2`.`a` like _latin1'a%') WITH CASCADED CHECK OPTION*/;
+/*!50001 CREATE ALGORITHM=UNDEFINED */
+/*!50013 DEFINER=`root`@`localhost` SQL SECURITY DEFINER */
+/*!50001 VIEW `v2` AS select `t2`.`a` AS `a` from `t2` where (`t2`.`a` like _latin1'a%') */
+/*!50002 WITH CASCADED CHECK OPTION */;
 /*!40103 SET TIME_ZONE=@OLD_TIME_ZONE */;
 
 /*!40101 SET SQL_MODE=@OLD_SQL_MODE */;
@@ -1890,13 +1898,19 @@ DROP TABLE IF EXISTS `v3`;
 ) */;
 /*!50001 DROP TABLE IF EXISTS `v1`*/;
 /*!50001 DROP VIEW IF EXISTS `v1`*/;
-/*!50001 CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `v1` AS select `v3`.`a` AS `a`,`v3`.`b` AS `b`,`v3`.`c` AS `c` from `v3` where (`v3`.`b` in (1,2,3,4,5,6,7))*/;
+/*!50001 CREATE ALGORITHM=UNDEFINED */
+/*!50013 DEFINER=`root`@`localhost` SQL SECURITY DEFINER */
+/*!50001 VIEW `v1` AS select `v3`.`a` AS `a`,`v3`.`b` AS `b`,`v3`.`c` AS `c` from `v3` where (`v3`.`b` in (1,2,3,4,5,6,7)) */;
 /*!50001 DROP TABLE IF EXISTS `v2`*/;
 /*!50001 DROP VIEW IF EXISTS `v2`*/;
-/*!50001 CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `v2` AS select `v3`.`a` AS `a` from (`v3` join `v1`) where ((`v1`.`a` = `v3`.`a`) and (`v3`.`b` = 3)) limit 1*/;
+/*!50001 CREATE ALGORITHM=UNDEFINED */
+/*!50013 DEFINER=`root`@`localhost` SQL SECURITY DEFINER */
+/*!50001 VIEW `v2` AS select `v3`.`a` AS `a` from (`v3` join `v1`) where ((`v1`.`a` = `v3`.`a`) and (`v3`.`b` = 3)) limit 1 */;
 /*!50001 DROP TABLE IF EXISTS `v3`*/;
 /*!50001 DROP VIEW IF EXISTS `v3`*/;
-/*!50001 CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `v3` AS select `t1`.`a` AS `a`,`t1`.`b` AS `b`,`t1`.`c` AS `c` from `t1`*/;
+/*!50001 CREATE ALGORITHM=UNDEFINED */
+/*!50013 DEFINER=`root`@`localhost` SQL SECURITY DEFINER */
+/*!50001 VIEW `v3` AS select `t1`.`a` AS `a`,`t1`.`b` AS `b`,`t1`.`c` AS `c` from `t1` */;
 /*!40103 SET TIME_ZONE=@OLD_TIME_ZONE */;
 
 /*!40101 SET SQL_MODE=@OLD_SQL_MODE */;
@@ -2465,13 +2479,19 @@ CREATE DATABASE /*!32312 IF NOT EXISTS*/ `test` /*!40100 DEFAULT CHARACTER SET l
 USE `test`;
 /*!50001 DROP TABLE IF EXISTS `v0`*/;
 /*!50001 DROP VIEW IF EXISTS `v0`*/;
-/*!50001 CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `v0` AS select `v1`.`a` AS `a`,`v1`.`b` AS `b`,`v1`.`c` AS `c` from `v1`*/;
+/*!50001 CREATE ALGORITHM=UNDEFINED */
+/*!50013 DEFINER=`root`@`localhost` SQL SECURITY DEFINER */
+/*!50001 VIEW `v0` AS select `v1`.`a` AS `a`,`v1`.`b` AS `b`,`v1`.`c` AS `c` from `v1` */;
 /*!50001 DROP TABLE IF EXISTS `v1`*/;
 /*!50001 DROP VIEW IF EXISTS `v1`*/;
-/*!50001 CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `v1` AS select `t1`.`a` AS `a`,`t1`.`b` AS `b`,`t1`.`c` AS `c` from `t1`*/;
+/*!50001 CREATE ALGORITHM=UNDEFINED */
+/*!50013 DEFINER=`root`@`localhost` SQL SECURITY DEFINER */
+/*!50001 VIEW `v1` AS select `t1`.`a` AS `a`,`t1`.`b` AS `b`,`t1`.`c` AS `c` from `t1` */;
 /*!50001 DROP TABLE IF EXISTS `v2`*/;
 /*!50001 DROP VIEW IF EXISTS `v2`*/;
-/*!50001 CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `v2` AS select `v0`.`a` AS `a`,`v0`.`b` AS `b`,`v0`.`c` AS `c` from `v0`*/;
+/*!50001 CREATE ALGORITHM=UNDEFINED */
+/*!50013 DEFINER=`root`@`localhost` SQL SECURITY DEFINER */
+/*!50001 VIEW `v2` AS select `v0`.`a` AS `a`,`v0`.`b` AS `b`,`v0`.`c` AS `c` from `v0` */;
 /*!40103 SET TIME_ZONE=@OLD_TIME_ZONE */;
 
 /*!40101 SET SQL_MODE=@OLD_SQL_MODE */;
@@ -2611,3 +2631,29 @@ UNLOCK TABLES;
 /*!40111 SET SQL_NOTES=@OLD_SQL_NOTES */;
 
 drop table t1;
+create table t1 (a int);
+insert into t1 values (289), (298), (234), (456), (789);
+create definer = CURRENT_USER view v1 as select * from t1;
+create SQL SECURITY INVOKER view v2 as select * from t1;
+create view v3 as select * from t1 with local check option;
+create algorithm=merge view v4 as select * from t1 with cascaded check option;
+create algorithm =temptable view v5 as select * from t1;
+drop table t1;
+drop view v1, v2, v3, v4, v5;
+show tables;
+Tables_in_test
+t1
+v1
+v2
+v3
+v4
+v5
+select * from v3 order by a;
+a
+234
+289
+298
+456
+789
+drop table t1;
+drop view v1, v2, v3, v4, v5;

mysql-test/t/mysqldump.test

@@ -1036,3 +1036,33 @@ insert into t1 values ('','');
 drop table t1;
 
 # End of 4.1 tests
+
+#
+# Bug 14871 Invalid view dump output
+#
+
+create table t1 (a int);
+insert into t1 values (289), (298), (234), (456), (789);
+create definer = CURRENT_USER view v1 as select * from t1;
+create SQL SECURITY INVOKER view v2 as select * from t1;
+create view v3 as select * from t1 with local check option;
+create algorithm=merge view v4 as select * from t1 with cascaded check option;
+create algorithm =temptable view v5 as select * from t1;
+
+# dump tables and views
+--exec $MYSQL_DUMP test > var/tmp/bug14871.sql
+
+# drop the db, tables and views
+drop table t1;
+drop view v1, v2, v3, v4, v5;
+
+# Reload dump
+--exec $MYSQL test < var/tmp/bug14871.sql
+
+# check that all tables and views could be created
+show tables;
+select * from v3 order by a;
+
+drop table t1;
+drop view v1, v2, v3, v4, v5;
+