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.

3 files changed, 211 insertions, 7 deletions

diff --git a/sql/ddl_log.cc b/sql/ddl_log.cc
index db2c386f800..49fdb861f8d 100644
--- a/sql/ddl_log.cc
+++ b/sql/ddl_log.cc
@@ -89,7 +89,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 */
@@ -98,7 +99,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,
 };
 
 
@@ -378,6 +380,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.
@@ -1551,6 +1573,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);
@@ -2501,3 +2587,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 368b887b6c5..38c0129f4bf 100644
--- a/sql/ddl_log.h
+++ b/sql/ddl_log.h
@@ -83,6 +83,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 */
 };
 
@@ -125,6 +127,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
   the phase to the maximum phase (..PHASE_END) for an entry.
@@ -266,5 +276,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 36e85bc2ea0..59b8962a84a 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;