MDEV-23844 Atomic DROP TABLE (single table)

Logging logic:
- Log tables, just before they are dropped, to the ddl log
- After the last table for the statement is dropped, log an xid for the
  whole ddl log event

In case of crash:
- Remove first any active DROP TABLE events from the ddl log that matches
  xids found in binary log (this mean the drop was successful and was
  propery logged).
- Loop over all active DROP TABLE events
  - Ensure that the table is completely dropped
- Write a DROP TABLE entry to the binary log with the dropped tables.

Other things:
- Added code to ha_drop_table() to be able to tell the difference if
  a get_new_handler() failed because of out-of-memory or because the
  handler refused/was not able to create a a handler. This was needed
  to get sequences to work as sequences needs a share object to be passed
  to get_new_handler()
- TC_LOG_BINLOG::recover() was changed to always collect Xid's from the
  binary log and always call ddl_log_close_binlogged_events(). This was
  needed to be able to collect DROP TABLE events with embedded Xid's
  (used by ddl log).
- Added a new variable "$grep_script" to binlog filter to be able to find
  only rows that matches a regexp.
- Had to adjust some test that changed because drop statements are a bit
  larger in the binary log than before (as we have to store the xid)

Other things:
- MDEV-25588 Atomic DDL: Binlog query event written upon recovery is corrupt
  fixed (in the original commit).

1 files changed, 277 insertions, 28 deletions

diff --git a/sql/ddl_log.cc b/sql/ddl_log.cc
index 1da6b5e52d5..c7e579f8a2e 100644
--- a/sql/ddl_log.cc
+++ b/sql/ddl_log.cc
@@ -26,6 +26,7 @@
 #include "sql_statistics.h"                     // rename_table_in_stats_tables
 #include "sql_view.h"                           // mysql_rename_view()
 #include "strfunc.h"                            // strconvert
+#include "sql_show.h"                           // append_identifier()
 #include <mysys_err.h>                          // EE_LINK
 
 
@@ -81,17 +82,19 @@ uchar ddl_log_file_magic[]=
 
 /* Action names for ddl_log_action_code */
 
-const char *ddl_log_action_name[DDL_LOG_LAST_ACTION]=
+const char *ddl_log_action_name[]=
 {
   "Unknown", "partitioning delete", "partitioning rename",
   "partitioning replace", "partitioning exchange",
-  "rename table", "rename view"
+  "rename table", "rename view",
+  "initialize drop table", "drop table",
+  "initialize drop view", "drop view"
 };
 
 /* Number of phases per entry */
 const uchar ddl_log_entry_phases[DDL_LOG_LAST_ACTION]=
 {
-  1, 1, 2, 3, 4, 1
+  0, 1, 1, 2, 3, 4, 1, 1, 3, 1, 1
 };
 
 
@@ -109,6 +112,7 @@ struct st_global_ddl_log
 };
 
 st_global_ddl_log global_ddl_log;
+String ddl_drop_query;                      // Used during startup recovery
 
 mysql_mutex_t LOCK_gdl;
 
@@ -285,6 +289,19 @@ static bool update_phase(uint entry_pos, uchar phase)
 }
 
 
+static bool update_next_entry_pos(uint entry_pos, uint next_entry)
+{
+  uchar buff[4];
+  DBUG_ENTER("update_next_entry_pos");
+
+  int4store(buff, next_entry);
+  DBUG_RETURN(mysql_file_pwrite(global_ddl_log.file_id, buff, sizeof(buff),
+                                global_ddl_log.io_size * entry_pos +
+                                DDL_LOG_NEXT_ENTRY_POS,
+                                MYF(MY_WME | MY_NABP)));
+}
+
+
 static bool update_xid(uint entry_pos, ulonglong xid)
 {
   uchar buff[8];
@@ -1130,6 +1147,122 @@ 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_DROP_TABLE_INIT_ACTION:
+  {
+    LEX_CSTRING *comment= &ddl_log_entry->tmp_name;
+    ddl_drop_query.length(0);
+    ddl_drop_query.set_charset(system_charset_info);
+    ddl_drop_query.append(STRING_WITH_LEN("DROP TABLE IF EXISTS "));
+    if (comment->length)
+    {
+      ddl_drop_query.append(comment);
+      ddl_drop_query.append(' ');
+    }
+    /* We don't increment phase as we want to retry this in case of crash */
+    break;
+  }
+  case DDL_LOG_DROP_TABLE_ACTION:
+  {
+    LEX_CSTRING db, table, path;
+    db=     ddl_log_entry->db;
+    table=  ddl_log_entry->name;
+    /* Note that path is without .frm extension */
+    path=   ddl_log_entry->tmp_name;
+
+    switch (ddl_log_entry->phase) {
+    case DDL_DROP_PHASE_TABLE:
+      if (hton)
+      {
+        if ((error= hton->drop_table(hton, path.str)))
+        {
+          if (!non_existing_table_error(error))
+            break;
+          error= -1;
+        }
+      }
+      else
+        error= ha_delete_table_force(thd, path.str, &db, &table);
+      if (error <= 0)
+      {
+        /* Not found or already deleted. Delete .frm if it exists */
+        strxnmov(to_path, sizeof(to_path)-1, path.str, reg_ext, NullS);
+        mysql_file_delete(key_file_frm, to_path, MYF(MY_WME|MY_IGNORE_ENOENT));
+      }
+      if (ddl_log_increment_phase_no_lock(entry_pos))
+        break;
+      (void) ddl_log_sync_no_lock();
+      /* Fall through */
+    case DDL_DROP_PHASE_TRIGGER:
+      Table_triggers_list::drop_all_triggers(thd, &db, &table,
+                                             MYF(MY_WME | MY_IGNORE_ENOENT));
+      if (ddl_log_increment_phase_no_lock(entry_pos))
+        break;
+      (void) ddl_log_sync_no_lock();
+      /* Fall through */
+
+    case DDL_DROP_PHASE_BINLOG:
+      append_identifier(thd, &ddl_drop_query, &db);
+      ddl_drop_query.append('.');
+      append_identifier(thd, &ddl_drop_query, &table);
+      ddl_drop_query.append(',');
+      /* We don't increment phase as we want to retry this in case of crash */
+
+      if (!ddl_log_entry->next_entry && mysql_bin_log.is_open())
+      {
+        /* Last drop table. Write query to binlog */
+        LEX_CSTRING end_comment=
+          { STRING_WITH_LEN(" /* generated by ddl recovery */")};
+        ddl_drop_query.length(ddl_drop_query.length()-1);
+        ddl_drop_query.append(&end_comment);
+
+        mysql_mutex_unlock(&LOCK_gdl);
+        (void) thd->binlog_query(THD::STMT_QUERY_TYPE, ddl_drop_query.ptr(),
+                                 ddl_drop_query.length(), TRUE, FALSE,
+                                 FALSE, 0);
+        mysql_mutex_lock(&LOCK_gdl);
+      }
+      break;
+    }
+    break;
+  }
+  case DDL_LOG_DROP_VIEW_INIT_ACTION:
+  {
+    ddl_drop_query.length(0);
+    ddl_drop_query.set_charset(system_charset_info);
+    ddl_drop_query.append(STRING_WITH_LEN("DROP VIEW IF EXISTS "));
+    /* We don't increment phase as we want to retry this in case of crash */
+    break;
+  }
+  case DDL_LOG_DROP_VIEW_ACTION:
+  {
+    LEX_CSTRING db, table, path;
+    db=     ddl_log_entry->db;
+    table=  ddl_log_entry->name;
+    /* Note that for views path is WITH .frm extension */
+    path=   ddl_log_entry->tmp_name;
+
+    mysql_file_delete(key_file_frm, path.str, MYF(MY_WME|MY_IGNORE_ENOENT));
+    append_identifier(thd, &ddl_drop_query, &db);
+    ddl_drop_query.append('.');
+    append_identifier(thd, &ddl_drop_query, &table);
+    ddl_drop_query.append(',');
+
+    if (!ddl_log_entry->next_entry)
+    {
+      /* Last drop view. Write query to binlog */
+      LEX_CSTRING end_comment=
+        { STRING_WITH_LEN(" /* generated by ddl recovery */")};
+      ddl_drop_query.length(ddl_drop_query.length()-1);
+      ddl_drop_query.append(&end_comment);
+
+      mysql_mutex_unlock(&LOCK_gdl);
+      (void) thd->binlog_query(THD::STMT_QUERY_TYPE, ddl_drop_query.ptr(),
+                               ddl_drop_query.length(), TRUE, FALSE,
+                               FALSE, 0);
+      mysql_mutex_lock(&LOCK_gdl);
+    }
+    break;
+  }
   default:
     DBUG_ASSERT(0);
     break;
@@ -1298,9 +1431,14 @@ bool ddl_log_write_entry(DDL_LOG_ENTRY *ddl_log_entry,
   uchar *pos, *end;
   DBUG_ENTER("ddl_log_write_entry");
 
+  *active_entry= 0;
   mysql_mutex_assert_owner(&LOCK_gdl);
-  if (!global_ddl_log.open)
+  DBUG_ASSERT(global_ddl_log.open);
+  if (unlikely(!global_ddl_log.open))
+  {
+    my_error(ER_INTERNAL_ERROR, MYF(0), "ddl log not initialized");
     DBUG_RETURN(TRUE);
+  }
 
   ddl_log_entry->entry_type= DDL_LOG_ENTRY_CODE;
   set_global_from_ddl_log_entry(ddl_log_entry);
@@ -1383,7 +1521,7 @@ bool ddl_log_write_execute_entry(uint first_entry,
     if (ddl_log_get_free_entry(active_entry))
       DBUG_RETURN(TRUE);
     got_free_entry= TRUE;
-  }
+ }
   if (write_ddl_log_file_entry((*active_entry)->entry_pos))
   {
     if (got_free_entry)
@@ -1509,6 +1647,7 @@ bool ddl_log_close_binlogged_events(HASH *xids)
   {
     if (read_ddl_log_entry(i, &ddl_log_entry))
       break;                                    // Read error. Stop reading
+    DBUG_PRINT("xid",("xid: %llu", ddl_log_entry.xid));
     if (ddl_log_entry.entry_type == DDL_LOG_EXECUTE_CODE &&
         ddl_log_entry.xid != 0 &&
         my_hash_search(xids, (uchar*) &ddl_log_entry.xid,
@@ -1562,6 +1701,7 @@ int ddl_log_execute_recovery()
   thd->store_globals();
   thd->init();                                  // Needed for error messages
   thd->log_all_errors= (global_system_variables.log_warnings >= 3);
+  ddl_drop_query.free();
 
   thd->set_query(recover_query_string, strlen(recover_query_string));
 
@@ -1599,6 +1739,7 @@ int ddl_log_execute_recovery()
       count++;
     }
   }
+  ddl_drop_query.free();
   close_ddl_log();
   mysql_mutex_unlock(&LOCK_gdl);
   thd->reset_query();
@@ -1682,6 +1823,7 @@ void ddl_log_release_entries(DDL_LOG_STATE *ddl_log_state)
     next= log_entry->next_active_log_entry;
     ddl_log_release_memory_entry(log_entry);
   }
+  ddl_log_state->list= 0;
 
   if (ddl_log_state->execute_entry)
   {
@@ -1779,6 +1921,33 @@ bool ddl_log_update_xid(DDL_LOG_STATE *state, ulonglong xid)
 }
 
 
+/*
+  Write ddl_log_entry and write or update ddl_execute_entry
+*/
+
+static bool ddl_log_write(DDL_LOG_STATE *ddl_state,
+                          DDL_LOG_ENTRY *ddl_log_entry)
+{
+  int error;
+  DDL_LOG_MEMORY_ENTRY *log_entry;
+  DBUG_ENTER("ddl_log_write");
+
+  mysql_mutex_lock(&LOCK_gdl);
+  error= ((ddl_log_write_entry(ddl_log_entry, &log_entry)) ||
+          ddl_log_write_execute_entry(log_entry->entry_pos,
+                                      &ddl_state->execute_entry));
+  mysql_mutex_unlock(&LOCK_gdl);
+  if (error)
+  {
+    if (log_entry)
+      ddl_log_release_memory_entry(log_entry);
+    DBUG_RETURN(1);
+  }
+  add_log_entry(ddl_state, log_entry);
+  DBUG_RETURN(0);
+}
+
+
 /**
    Logging of rename table
 */
@@ -1791,13 +1960,10 @@ bool ddl_log_rename_table(THD *thd, DDL_LOG_STATE *ddl_state,
                           const LEX_CSTRING *new_alias)
 {
   DDL_LOG_ENTRY ddl_log_entry;
-  DDL_LOG_MEMORY_ENTRY *log_entry;
   DBUG_ENTER("ddl_log_rename_file");
 
   bzero(&ddl_log_entry, sizeof(ddl_log_entry));
 
-  mysql_mutex_lock(&LOCK_gdl);
-
   ddl_log_entry.action_type=  DDL_LOG_RENAME_TABLE_ACTION;
   ddl_log_entry.next_entry=   ddl_state->list ? ddl_state->list->entry_pos : 0;
   lex_string_set(&ddl_log_entry.handler_name,
@@ -1808,20 +1974,7 @@ bool ddl_log_rename_table(THD *thd, DDL_LOG_STATE *ddl_state,
   ddl_log_entry.from_name=    *const_cast<LEX_CSTRING*>(org_alias);
   ddl_log_entry.phase=        DDL_RENAME_PHASE_TABLE;
 
-  if (ddl_log_write_entry(&ddl_log_entry, &log_entry))
-    goto error;
-
-  if (ddl_log_write_execute_entry(log_entry->entry_pos,
-                                  &ddl_state->execute_entry))
-    goto error;
-
-  add_log_entry(ddl_state, log_entry);
-  mysql_mutex_unlock(&LOCK_gdl);
-  DBUG_RETURN(0);
-
-error:
-  mysql_mutex_unlock(&LOCK_gdl);
-  DBUG_RETURN(1);
+  DBUG_RETURN(ddl_log_write(ddl_state, &ddl_log_entry));
 }
 
 /*
@@ -1835,13 +1988,10 @@ bool ddl_log_rename_view(THD *thd, DDL_LOG_STATE *ddl_state,
                          const LEX_CSTRING *new_alias)
 {
   DDL_LOG_ENTRY ddl_log_entry;
-  DDL_LOG_MEMORY_ENTRY *log_entry;
   DBUG_ENTER("ddl_log_rename_file");
 
   bzero(&ddl_log_entry, sizeof(ddl_log_entry));
 
-  mysql_mutex_lock(&LOCK_gdl);
-
   ddl_log_entry.action_type=  DDL_LOG_RENAME_VIEW_ACTION;
   ddl_log_entry.next_entry=   ddl_state->list ? ddl_state->list->entry_pos : 0;
   ddl_log_entry.db=           *const_cast<LEX_CSTRING*>(new_db);
@@ -1849,18 +1999,117 @@ bool ddl_log_rename_view(THD *thd, DDL_LOG_STATE *ddl_state,
   ddl_log_entry.from_db=      *const_cast<LEX_CSTRING*>(org_db);
   ddl_log_entry.from_name=    *const_cast<LEX_CSTRING*>(org_alias);
 
+  DBUG_RETURN(ddl_log_write(ddl_state, &ddl_log_entry));
+}
+
+
+/**
+   Logging of DROP TABLE and DROP VIEW
+
+   Note that in contrast to rename, which are re-done in reverse order,
+   deletes are stored in a linked list according to delete order. This
+   is to ensure that the tables, for the query generated for binlog,
+   is in original delete order.
+*/
+
+static bool ddl_log_drop_init(THD *thd, DDL_LOG_STATE *ddl_state,
+                              ddl_log_action_code action_code,
+                              const LEX_CSTRING *comment)
+{
+  DDL_LOG_ENTRY ddl_log_entry;
+  DBUG_ENTER("ddl_log_drop_file");
+
+  bzero(&ddl_log_entry, sizeof(ddl_log_entry));
+
+  ddl_log_entry.action_type=  action_code;
+  ddl_log_entry.next_entry=   0;
+  ddl_log_entry.tmp_name=     *const_cast<LEX_CSTRING*>(comment);
+  ddl_log_entry.phase=        0;
+
+  DBUG_RETURN(ddl_log_write(ddl_state, &ddl_log_entry));
+}
+
+
+bool ddl_log_drop_table_init(THD *thd, DDL_LOG_STATE *ddl_state,
+                             const LEX_CSTRING *comment)
+{
+  return ddl_log_drop_init(thd, ddl_state, DDL_LOG_DROP_TABLE_INIT_ACTION,
+                           comment);
+}
+
+bool ddl_log_drop_view_init(THD *thd, DDL_LOG_STATE *ddl_state)
+{
+  LEX_CSTRING comment= {0,0};
+  return ddl_log_drop_init(thd, ddl_state, DDL_LOG_DROP_VIEW_INIT_ACTION,
+                           &comment);
+}
+
+static bool ddl_log_drop(THD *thd, DDL_LOG_STATE *ddl_state,
+                         ddl_log_action_code action_code,
+                         uint phase,
+                         handlerton *hton,
+                         const LEX_CSTRING *path,
+                         const LEX_CSTRING *db,
+                         const LEX_CSTRING *table)
+{
+  DDL_LOG_ENTRY ddl_log_entry;
+  DDL_LOG_MEMORY_ENTRY *log_entry;
+  DBUG_ENTER("ddl_log_drop");
+
+  DBUG_ASSERT(ddl_state->list);
+  bzero(&ddl_log_entry, sizeof(ddl_log_entry));
+
+  ddl_log_entry.action_type=  action_code;
+  if (hton)
+    lex_string_set(&ddl_log_entry.handler_name,
+                   ha_resolve_storage_engine_name(hton));
+  ddl_log_entry.db=           *const_cast<LEX_CSTRING*>(db);
+  ddl_log_entry.name=         *const_cast<LEX_CSTRING*>(table);
+  ddl_log_entry.tmp_name=     *const_cast<LEX_CSTRING*>(path);
+  ddl_log_entry.phase=        (uchar) phase;
+
+  mysql_mutex_lock(&LOCK_gdl);
   if (ddl_log_write_entry(&ddl_log_entry, &log_entry))
     goto error;
 
-  if (ddl_log_write_execute_entry(log_entry->entry_pos,
-                                  &ddl_state->execute_entry))
+  (void) ddl_log_sync_no_lock();
+  if (update_next_entry_pos(ddl_state->list->entry_pos,
+                            log_entry->entry_pos))
+  {
+    ddl_log_release_memory_entry(log_entry);
     goto error;
+  }
 
-  add_log_entry(ddl_state, log_entry);
   mysql_mutex_unlock(&LOCK_gdl);
+  add_log_entry(ddl_state, log_entry);
   DBUG_RETURN(0);
 
 error:
   mysql_mutex_unlock(&LOCK_gdl);
   DBUG_RETURN(1);
 }
+
+
+bool ddl_log_drop_table(THD *thd, DDL_LOG_STATE *ddl_state,
+                        handlerton *hton,
+                        const LEX_CSTRING *path,
+                        const LEX_CSTRING *db,
+                        const LEX_CSTRING *table)
+{
+  DBUG_ENTER("ddl_log_drop_table");
+  DBUG_RETURN(ddl_log_drop(thd, ddl_state,
+                           DDL_LOG_DROP_TABLE_ACTION, DDL_DROP_PHASE_TABLE,
+                           hton, path, db, table));
+}
+
+
+bool ddl_log_drop_view(THD *thd, DDL_LOG_STATE *ddl_state,
+                        const LEX_CSTRING *path,
+                        const LEX_CSTRING *db,
+                        const LEX_CSTRING *table)
+{
+  DBUG_ENTER("ddl_log_drop_view");
+  DBUG_RETURN(ddl_log_drop(thd, ddl_state,
+                           DDL_LOG_DROP_VIEW_ACTION, 0,
+                           (handlerton*) 0, path, db, table));
+}