MDEV-24607 Atomic CREATE VIEW

The logic of the new code is:
- Log CREATE view to DDL log, with a marker if old view existed
- If old view exists (in case of CREATE or REPLACE view), make a copy
  of the old view as view_name.frm-
- Create the new view definition file
- Delete copy of view if it was created.

Crash recovery:
- Delete view_name.frm~ file (Temporary file for view definition)
- If query was logged to binary log
  - Delete copy of view if it exists
- else
   -rename the copy of the view over the .frm file (restoring the
    old definition)

One benefit of the new code is that CREATE OR REPLACE VIEW for an
existing view is no fully atomic: Either the view will be replaced or
the old one will be left unchanged.

5 files changed, 380 insertions, 7 deletions

diff --git a/mysql-test/suite/atomic/create_view.result b/mysql-test/suite/atomic/create_view.result
new file mode 100644
index 00000000000..3c38a6516e2
--- /dev/null
+++ b/mysql-test/suite/atomic/create_view.result
@@ -0,0 +1,82 @@
+query: CREATE VIEW t1 as select "new"
+crash point: ddl_log_create_before_copy_view
+t2.frm
+old
+old
+crash point: ddl_log_create_before_create_view
+t2.frm
+old
+old
+crash point: definition_file_after_create
+t2.frm
+old
+old
+crash point: ddl_log_create_after_create_view
+t2.frm
+old
+old
+crash point: ddl_log_create_before_binlog
+t2.frm
+old
+old
+crash point: ddl_log_create_after_binlog
+t1.frm
+t2.frm
+old
+old
+master-bin.000001	#	Query	#	#	use `test`; CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `t1` AS select "new"
+query: CREATE OR REPLACE VIEW t1 as select "new"
+crash point: ddl_log_create_before_copy_view
+t2.frm
+old
+old
+crash point: ddl_log_create_before_create_view
+t2.frm
+old
+old
+crash point: definition_file_after_create
+t2.frm
+old
+old
+crash point: ddl_log_create_after_create_view
+t2.frm
+old
+old
+crash point: ddl_log_create_before_binlog
+t2.frm
+old
+old
+crash point: ddl_log_create_after_binlog
+t1.frm
+t2.frm
+old
+old
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `t1` AS select "new"
+query: CREATE OR REPLACE VIEW t2 as select "new"
+crash point: ddl_log_create_before_copy_view
+t2.frm
+old
+old
+crash point: ddl_log_create_before_create_view
+t2.frm
+old
+old
+crash point: definition_file_after_create
+t2.frm
+old
+old
+crash point: ddl_log_create_after_create_view
+t2.frm
+old
+old
+crash point: ddl_log_create_before_binlog
+t2.frm
+old
+old
+crash point: ddl_log_create_after_binlog
+t2.frm
+new
+new
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `t2` AS select "new"
+Warnings:
+Note	4092	Unknown VIEW: 'test.t1,test.t2'
diff --git a/mysql-test/suite/atomic/create_view.test b/mysql-test/suite/atomic/create_view.test
new file mode 100644
index 00000000000..a51f8df7438
--- /dev/null
+++ b/mysql-test/suite/atomic/create_view.test
@@ -0,0 +1,87 @@
+--source include/have_debug.inc
+--source include/have_sequence.inc
+--source include/have_log_bin.inc
+--source include/not_valgrind.inc
+
+#
+# Testing of atomic create view with crashes in a lot of different places
+
+--disable_query_log
+call mtr.add_suppression("InnoDB: .* does not exist in the InnoDB internal");
+--enable_query_log
+let $MYSQLD_DATADIR= `SELECT @@datadir`;
+
+let $crash_count=6;
+let $crash_points='ddl_log_create_before_copy_view', 'ddl_log_create_before_create_view', 'definition_file_after_create','ddl_log_create_after_create_view', 'ddl_log_create_before_binlog', 'ddl_log_create_after_binlog';
+
+let $statement_count=3;
+let $statements='CREATE VIEW t1 as select "new"',
+                'CREATE OR REPLACE VIEW t1 as select "new"',
+                'CREATE OR REPLACE VIEW t2 as select "new"';
+
+let $old_debug=`select @@debug_dbug`;
+
+let $e=0;
+let $keep_include_silent=1;
+let $grep_script=CREATE|DROP;
+--disable_query_log
+
+while ($e < 1)
+{
+  inc $e;
+
+  let $r=0;
+  while ($r < $statement_count)
+  {
+    inc $r;
+
+    let $statement=`select ELT($r, $statements)`;
+    --echo query: $statement
+
+    let $c=0;
+    while ($c < $crash_count)
+    {
+      inc $c;
+      let $crash=`select ELT($c, $crash_points)`;
+
+      create view t2 as select "old";
+
+      RESET MASTER;
+      --echo crash point: $crash
+      --exec echo "restart" > $MYSQLTEST_VARDIR/tmp/mysqld.1.expect
+      --disable_reconnect
+      --eval set @@debug_dbug="+d,$crash",@debug_crash_counter=1
+      let $errno=0;
+      --error 0,2013
+      --eval $statement;
+      let $error=$errno;
+      --enable_reconnect
+      --source include/wait_until_connected_again.inc
+      --disable_query_log
+      --eval set @@debug_dbug="$old_debug"
+
+      if ($error == 0)
+      {
+        echo "No crash!";
+      }
+      # Check which tables still exists
+      --list_files $MYSQLD_DATADIR/test t*
+      select * from t2;
+
+      --let $binlog_file=master-bin.000001
+      --source include/show_binlog_events.inc
+      if ($error)
+      {
+        --let $binlog_file=master-bin.000002
+        --source include/show_binlog_events.inc
+      }
+      # Drop the tables. The warnings will show what was dropped
+      --disable_warnings
+      drop view if exists t1,t2;
+      --enable_warnings
+    }
+  }
+}
+drop view if exists t1,t2;
+
+--enable_query_log
diff --git a/sql/ddl_log.cc b/sql/ddl_log.cc
index 56e40128132..990c31d8034 100644
--- a/sql/ddl_log.cc
+++ b/sql/ddl_log.cc
@@ -87,7 +87,8 @@ const char *ddl_log_action_name[DDL_LOG_LAST_ACTION]=
   "partitioning replace", "partitioning exchange",
   "rename table", "rename view",
   "initialize drop table", "drop table",
-  "drop view", "drop trigger", "drop db", "create table",
+  "drop view", "drop trigger", "drop db", "create table", "create view",
+  "delete tmp file",
 };
 
 /* Number of phases per entry */
@@ -96,7 +97,8 @@ const uchar ddl_log_entry_phases[DDL_LOG_LAST_ACTION]=
   0, 1, 1, 2,
   (uchar) EXCH_PHASE_END, (uchar) DDL_RENAME_PHASE_END, 1, 1,
   (uchar) DDL_DROP_PHASE_END, 1, 1,
-  (uchar) DDL_DROP_DB_PHASE_END, (uchar) DDL_CREATE_TABLE_PHASE_END
+  (uchar) DDL_DROP_DB_PHASE_END, (uchar) DDL_CREATE_TABLE_PHASE_END,
+  (uchar) DDL_CREATE_VIEW_PHASE_END, 0,
 };
 
 
@@ -356,6 +358,26 @@ bool ddl_log_disable_execute_entry(DDL_LOG_MEMORY_ENTRY **active_entry)
 }
 
 
+/*
+  Check if an executive entry is active
+
+  @return 0  Entry is active
+  @return 1  Entry is not active
+*/
+
+static bool is_execute_entry_active(uint entry_pos)
+{
+  uchar buff[1];
+  DBUG_ENTER("disable_execute_entry");
+
+  if (mysql_file_pread(global_ddl_log.file_id, buff, sizeof(buff),
+                       global_ddl_log.io_size * entry_pos +
+                       DDL_LOG_ENTRY_TYPE_POS,
+                       MYF(MY_WME | MY_NABP)))
+    DBUG_RETURN(1);
+  DBUG_RETURN(buff[0] == (uchar) DDL_LOG_EXECUTE_CODE);
+}
+
 
 /**
   Read header of ddl log file.
@@ -1513,6 +1535,70 @@ static int ddl_log_execute_action(THD *thd, MEM_ROOT *mem_root,
     (void) update_phase(entry_pos, DDL_LOG_FINAL_PHASE);
     break;
   }
+  case DDL_LOG_CREATE_VIEW_ACTION:
+  {
+    char *path= to_path;
+    size_t path_length= ddl_log_entry->tmp_name.length;
+    memcpy(path, ddl_log_entry->tmp_name.str, path_length+1);
+    path[path_length+1]= 0;               // Prepare for extending
+
+    /* Remove temporary parser file */
+    path[path_length]='~';
+    mysql_file_delete(key_file_fileparser, path,
+                      MYF(MY_WME|MY_IGNORE_ENOENT));
+    path[path_length]= 0;
+
+    switch (ddl_log_entry->phase) {
+    case DDL_CREATE_VIEW_PHASE_NO_OLD_VIEW:
+    {
+      /*
+        No old view exists, so we can just delete the .frm and temporary files
+      */
+      path[path_length]='-';
+      mysql_file_delete(key_file_fileparser, path,
+                        MYF(MY_WME|MY_IGNORE_ENOENT));
+      path[path_length]= 0;
+      mysql_file_delete(key_file_frm, path, MYF(MY_WME|MY_IGNORE_ENOENT));
+      break;
+    }
+    case DDL_CREATE_VIEW_PHASE_DELETE_VIEW_COPY:
+    {
+      /*
+        Old view existed. We crashed before we had done a copy and change
+        state to DDL_CREATE_VIEW_PHASE_OLD_VIEW_COPIED
+      */
+      path[path_length]='-';
+      mysql_file_delete(key_file_fileparser, path,
+                        MYF(MY_WME|MY_IGNORE_ENOENT));
+      path[path_length]= 0;
+      break;
+    }
+    case DDL_CREATE_VIEW_PHASE_OLD_VIEW_COPIED:
+    {
+      /*
+        Old view existed copied to '-' file. Restore it
+      */
+      memcpy(from_path, path, path_length+2);
+      from_path[path_length]='-';
+      if (!access(from_path, F_OK))
+        mysql_file_rename(key_file_fileparser, from_path, path, MYF(MY_WME));
+      break;
+    }
+    }
+    (void) update_phase(entry_pos, DDL_LOG_FINAL_PHASE);
+    break;
+  }
+  case DDL_LOG_DELETE_TMP_FILE_ACTION:
+  {
+    LEX_CSTRING path= ddl_log_entry->tmp_name;
+    DBUG_ASSERT(ddl_log_entry->unique_id <= UINT_MAX32);
+    if (!ddl_log_entry->unique_id ||
+        !is_execute_entry_active((uint) ddl_log_entry->unique_id))
+      mysql_file_delete(key_file_fileparser, path.str,
+                        MYF(MY_WME|MY_IGNORE_ENOENT));
+    (void) update_phase(entry_pos, DDL_LOG_FINAL_PHASE);
+    break;
+  }
   break;
   default:
     DBUG_ASSERT(0);
@@ -2439,3 +2525,49 @@ bool ddl_log_create_table(THD *thd, DDL_LOG_STATE *ddl_state,
 
   DBUG_RETURN(ddl_log_write(ddl_state, &ddl_log_entry));
 }
+
+
+/**
+   Log CREATE VIEW
+*/
+
+bool ddl_log_create_view(THD *thd, DDL_LOG_STATE *ddl_state,
+                         const LEX_CSTRING *path,
+                         enum_ddl_log_create_view_phase phase)
+{
+  DDL_LOG_ENTRY ddl_log_entry;
+  DBUG_ENTER("ddl_log_create_view");
+
+  bzero(&ddl_log_entry, sizeof(ddl_log_entry));
+  ddl_log_entry.action_type=  DDL_LOG_CREATE_VIEW_ACTION;
+  ddl_log_entry.tmp_name=     *const_cast<LEX_CSTRING*>(path);
+  ddl_log_entry.phase=        (uchar) phase;
+  DBUG_RETURN(ddl_log_write(ddl_state, &ddl_log_entry));
+}
+
+
+/**
+  Log creation of temporary file that should be deleted during recovery
+
+  @param thd             Thread handler
+  @param ddl_log_state   ddl_state
+  @param path            Path to file to be deleted
+  @param depending_state If not NULL, then do not delete the temp file if this
+                         entry exists and is active.
+*/
+
+bool ddl_log_delete_tmp_file(THD *thd, DDL_LOG_STATE *ddl_state,
+                             const LEX_CSTRING *path,
+                             DDL_LOG_STATE *depending_state)
+{
+  DDL_LOG_ENTRY ddl_log_entry;
+  DBUG_ENTER("ddl_log_delete_tmp_file");
+
+  bzero(&ddl_log_entry, sizeof(ddl_log_entry));
+  ddl_log_entry.action_type=  DDL_LOG_DELETE_TMP_FILE_ACTION;
+  ddl_log_entry.next_entry=   ddl_state->list ? ddl_state->list->entry_pos : 0;
+  ddl_log_entry.tmp_name=     *const_cast<LEX_CSTRING*>(path);
+  if (depending_state)
+    ddl_log_entry.unique_id=  depending_state->execute_entry->entry_pos;
+  DBUG_RETURN(ddl_log_write(ddl_state, &ddl_log_entry));
+}
diff --git a/sql/ddl_log.h b/sql/ddl_log.h
index a0b2e103617..09ad7cd7ba9 100644
--- a/sql/ddl_log.h
+++ b/sql/ddl_log.h
@@ -84,6 +84,8 @@ enum ddl_log_action_code
   DDL_LOG_DROP_TRIGGER_ACTION= 10,
   DDL_LOG_DROP_DB_ACTION=11,
   DDL_LOG_CREATE_TABLE_ACTION=12,
+  DDL_LOG_CREATE_VIEW_ACTION=13,
+  DDL_LOG_DELETE_TMP_FILE_ACTION=14,
   DDL_LOG_LAST_ACTION                          /* End marker */
 };
 
@@ -126,6 +128,14 @@ enum enum_ddl_log_create_table_phase {
   DDL_CREATE_TABLE_PHASE_END
 };
 
+enum enum_ddl_log_create_view_phase {
+  DDL_CREATE_VIEW_PHASE_NO_OLD_VIEW,
+  DDL_CREATE_VIEW_PHASE_DELETE_VIEW_COPY,
+  DDL_CREATE_VIEW_PHASE_OLD_VIEW_COPIED,
+  DDL_CREATE_VIEW_PHASE_END
+};
+
+
 /*
   Setting ddl_log_entry.phase to this has the same effect as setting
   action_type to DDL_IGNORE_LOG_ENTRY_CODE
@@ -247,5 +257,11 @@ bool ddl_log_create_table(THD *thd, DDL_LOG_STATE *ddl_state,
                           const LEX_CSTRING *db,
                           const LEX_CSTRING *table,
                           bool only_frm);
+bool ddl_log_create_view(THD *thd, DDL_LOG_STATE *ddl_state,
+                         const LEX_CSTRING *path,
+                         enum_ddl_log_create_view_phase phase);
+bool ddl_log_delete_tmp_file(THD *thd, DDL_LOG_STATE *ddl_state,
+                             const LEX_CSTRING *path,
+                             DDL_LOG_STATE *depending_state);
 extern mysql_mutex_t LOCK_gdl;
 #endif /* DDL_LOG_INCLUDED */
diff --git a/sql/sql_view.cc b/sql/sql_view.cc
index 5cd8ad07c58..e737b4c89ec 100644
--- a/sql/sql_view.cc
+++ b/sql/sql_view.cc
@@ -44,7 +44,9 @@
 
 const LEX_CSTRING view_type= { STRING_WITH_LEN("VIEW") };
 
-static int mysql_register_view(THD *, TABLE_LIST *, enum_view_create_mode);
+static int mysql_register_view(THD *thd, DDL_LOG_STATE *ddl_log_state,
+                               TABLE_LIST *view, enum_view_create_mode mode,
+                               char *backup_file_name);
 
 /*
   Make a unique name for an anonymous view column
@@ -406,6 +408,8 @@ bool mysql_create_view(THD *thd, TABLE_LIST *views,
   SELECT_LEX *select_lex= lex->first_select_lex();
   SELECT_LEX *sl;
   SELECT_LEX_UNIT *unit= &lex->unit;
+  DDL_LOG_STATE ddl_log_state, ddl_log_state_tmp_file;
+  char backup_file_name[FN_REFLEN+2];
   bool res= FALSE;
   DBUG_ENTER("mysql_create_view");
 
@@ -413,6 +417,9 @@ bool mysql_create_view(THD *thd, TABLE_LIST *views,
   DBUG_ASSERT(!lex->proc_list.first && !lex->result &&
               !lex->param_list.elements);
 
+  bzero(&ddl_log_state, sizeof(ddl_log_state));
+  bzero(&ddl_log_state_tmp_file, sizeof(ddl_log_state_tmp_file));
+  backup_file_name[0]= 0;
   /*
     We can't allow taking exclusive meta-data locks of unlocked view under
     LOCK TABLES since this might lead to deadlock. Since at the moment we
@@ -649,7 +656,7 @@ bool mysql_create_view(THD *thd, TABLE_LIST *views,
   }
 #endif
 
-  res= mysql_register_view(thd, view, mode);
+  res= mysql_register_view(thd, &ddl_log_state, view, mode, backup_file_name);
 
   /*
     View TABLE_SHARE must be removed from the table definition cache in order to
@@ -710,10 +717,21 @@ bool mysql_create_view(THD *thd, TABLE_LIST *views,
       with statement based replication
     */
     thd->reset_unsafe_warnings();
+    thd->binlog_xid= thd->query_id;
+    ddl_log_update_xid(&ddl_log_state, thd->binlog_xid);
+    if (backup_file_name[0])
+    {
+      LEX_CSTRING cpath= {backup_file_name, strlen(backup_file_name) };
+      ddl_log_delete_tmp_file(thd, &ddl_log_state_tmp_file, &cpath,
+                              &ddl_log_state);
+    }
+    debug_crash_here("ddl_log_create_before_binlog");
     if (thd->binlog_query(THD::STMT_QUERY_TYPE,
                           buff.ptr(), buff.length(), FALSE, FALSE, FALSE,
                           errcode) > 0)
       res= TRUE;
+    thd->binlog_xid= 0;
+    debug_crash_here("ddl_log_create_after_binlog");
   }
 
   if (mode != VIEW_CREATE_NEW)
@@ -721,8 +739,14 @@ bool mysql_create_view(THD *thd, TABLE_LIST *views,
   if (res)
     goto err;
 
+  if (backup_file_name[0] &&
+      mysql_file_delete(key_file_fileparser, backup_file_name, MYF(MY_WME)))
+    goto err;                                   // Should be impossible
+
   my_ok(thd);
   lex->link_first_table_back(view, link_to_local);
+  ddl_log_complete(&ddl_log_state);
+  ddl_log_complete(&ddl_log_state_tmp_file);
   DBUG_RETURN(0);
 
 #ifdef WITH_WSREP
@@ -735,6 +759,10 @@ err:
   lex->link_first_table_back(view, link_to_local);
 err_no_relink:
   unit->cleanup();
+  if (backup_file_name[0])
+    mysql_file_delete(key_file_fileparser, backup_file_name, MYF(MY_WME));
+  ddl_log_complete(&ddl_log_state);
+  ddl_log_complete(&ddl_log_state_tmp_file);
   DBUG_RETURN(res || thd->is_error());
 }
 
@@ -891,6 +919,7 @@ int mariadb_fix_view(THD *thd, TABLE_LIST *view, bool wrong_checksum,
     thd		- thread handler
     view	- view description
     mode	- VIEW_CREATE_NEW, VIEW_ALTER, VIEW_CREATE_OR_REPLACE
+    backup_file_name  - Store name for backup of old view definition here
 
   RETURN
      0	OK
@@ -898,8 +927,9 @@ int mariadb_fix_view(THD *thd, TABLE_LIST *view, bool wrong_checksum,
      1	Error and error message given
 */
 
-static int mysql_register_view(THD *thd, TABLE_LIST *view,
-			       enum_view_create_mode mode)
+static int mysql_register_view(THD *thd, DDL_LOG_STATE *ddl_log_state,
+                               TABLE_LIST *view, enum_view_create_mode mode,
+                               char *backup_file_name)
 {
   LEX *lex= thd->lex;
 
@@ -936,11 +966,13 @@ static int mysql_register_view(THD *thd, TABLE_LIST *view,
   char dir_buff[FN_REFLEN + 1], path_buff[FN_REFLEN + 1];
   LEX_CSTRING dir, file, path;
   int error= 0;
+  bool old_view_exists= 0;
   DBUG_ENTER("mysql_register_view");
 
   /* Generate view definition and IS queries. */
   view_query.length(0);
   is_query.length(0);
+  backup_file_name[0]= 0;
   {
     Sql_mode_instant_remove sms(thd, MODE_ANSI_QUOTES);
 
@@ -1042,6 +1074,7 @@ loop_out:
 
     if (ha_table_exists(thd, &view->db, &view->table_name))
     {
+      old_view_exists= 1;
       if (lex->create_info.if_not_exists())
       {
         push_warning_printf(thd, Sql_condition::WARN_LEVEL_NOTE,
@@ -1077,7 +1110,7 @@ loop_out:
       */
     }
     else
-   {
+    {
       if (mode == VIEW_ALTER)
       {
 	my_error(ER_NO_SUCH_TABLE, MYF(0), view->db.str, view->alias.str);
@@ -1137,12 +1170,35 @@ loop_out:
     goto err;
   }
 
+  ddl_log_create_view(thd, ddl_log_state, &path, old_view_exists ?
+                      DDL_CREATE_VIEW_PHASE_DELETE_VIEW_COPY :
+                      DDL_CREATE_VIEW_PHASE_NO_OLD_VIEW);
+
+  debug_crash_here("ddl_log_create_before_copy_view");
+
+  if (old_view_exists)
+  {
+    /* Make a backup that we can restore in case of crash */
+    memcpy(backup_file_name, path.str, path.length);
+    backup_file_name[path.length]='-';
+    backup_file_name[path.length+1]= 0;
+    if (my_copy(path.str, backup_file_name, MYF(MY_WME)))
+    {
+      error= 1;
+      goto err;
+    }
+    ddl_log_update_phase(ddl_log_state, DDL_CREATE_VIEW_PHASE_OLD_VIEW_COPIED);
+  }
+
+  debug_crash_here("ddl_log_create_before_create_view");
   if (sql_create_definition_file(&dir, &file, view_file_type,
 				 (uchar*)view, view_parameters))
   {
     error= thd->is_error() ? -1 : 1;
     goto err;
   }
+  debug_crash_here("ddl_log_create_after_create_view");
+
   DBUG_RETURN(0);
 err:
   view->select_stmt.str= NULL;