Bug#47343: InnoDB fails to clean-up after lock wait timeout on

           REORGANIZE PARTITION

There were several problems which lead to this this,
all related to bad error handling.

1) There was several bugs preventing the ddl-log to be used for
   cleaning up created files on error.

2) The error handling after the copy partition rows did not close
   and unlock the tables, resulting in deletion of partitions
   which were in use, which lead InnoDB to put the partition to
   drop in a background queue.

sql/ha_partition.cc:
  Bug#47343: InnoDB fails to clean-up after lock wait timeout on
             REORGANIZE PARTITION
  
  Better error handling, if partition has been created/opened/locked
  then make sure it is unlocked and closed before returning error.
  The delete of the newly created partition is handled by the ddl-log.
sql/sql_parse.cc:
  Bug#47343: InnoDB fails to clean-up after lock wait timeout on
             REORGANIZE PARTITION
  
  Fix a bug found when experimenting, thd could really be NULL here,
  as mentioned in the function header.
sql/sql_partition.cc:
  Bug#47343: InnoDB fails to clean-up after lock wait timeout on
             REORGANIZE PARTITION
  
  Used the correct .frm shadow name to put into the ddl-log.
  Really use the ddl-log to handle errors.
sql/sql_table.cc:
  Bug#47343: InnoDB fails to clean-up after lock wait timeout on
             REORGANIZE PARTITION
  
  Fixes of the ddl-log when used as error recovery (no crash).
  When executing an entry from memory (not read from disk)
  the name_len was not set correctly.

4 files changed, 68 insertions, 27 deletions

diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index b854e270029..555a0b97ddd 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -1215,17 +1215,28 @@ int ha_partition::prepare_new_partition(TABLE *tbl,
                                         partition_element *p_elem)
 {
   int error;
-  bool create_flag= FALSE;
   DBUG_ENTER("prepare_new_partition");
 
   if ((error= set_up_table_before_create(tbl, part_name, create_info,
                                          0, p_elem)))
-    goto error;
+    goto error_create;
   if ((error= file->ha_create(part_name, tbl, create_info)))
-    goto error;
-  create_flag= TRUE;
+  {
+    /*
+      Added for safety, InnoDB reports HA_ERR_FOUND_DUPP_KEY
+      if the table/partition already exists.
+      If we return that error code, then print_error would try to
+      get_dup_key on a non-existing partition.
+      So return a more reasonable error code.
+    */
+    if (error == HA_ERR_FOUND_DUPP_KEY)
+      error= HA_ERR_TABLE_EXIST;
+    goto error_create;
+  }
+  DBUG_PRINT("info", ("partition %s created", part_name));
   if ((error= file->ha_open(tbl, part_name, m_mode, m_open_test_lock)))
-    goto error;
+    goto error_open;
+  DBUG_PRINT("info", ("partition %s opened", part_name));
   /*
     Note: if you plan to add another call that may return failure,
     better to do it before external_lock() as cleanup_new_partition()
@@ -1233,12 +1244,15 @@ int ha_partition::prepare_new_partition(TABLE *tbl,
     Otherwise see description for cleanup_new_partition().
   */
   if ((error= file->ha_external_lock(ha_thd(), m_lock_type)))
-    goto error;
+    goto error_external_lock;
+  DBUG_PRINT("info", ("partition %s external locked", part_name));
 
   DBUG_RETURN(0);
-error:
-  if (create_flag)
-    VOID(file->ha_delete_table(part_name));
+error_external_lock:
+  VOID(file->close());
+error_open:
+  VOID(file->ha_delete_table(part_name));
+error_create:
   DBUG_RETURN(error);
 }
 
@@ -1272,19 +1286,23 @@ error:
 
 void ha_partition::cleanup_new_partition(uint part_count)
 {
-  handler **save_m_file= m_file;
   DBUG_ENTER("ha_partition::cleanup_new_partition");
 
-  if (m_added_file && m_added_file[0])
+  if (m_added_file)
   {
-    m_file= m_added_file;
-    m_added_file= NULL;
+    THD *thd= ha_thd();
+    handler **file= m_added_file;
+    while ((part_count > 0) && (*file))
+    {
+      (*file)->ha_external_lock(thd, F_UNLCK);
+      (*file)->close();
 
-    external_lock(ha_thd(), F_UNLCK);
-    /* delete_table also needed, a bit more complex */
-    close();
+      /* Leave the (*file)->ha_delete_table(part_name) to the ddl-log */
 
-    m_file= save_m_file;
+      file++;
+      part_count--;
+    }
+    m_added_file= NULL;
   }
   DBUG_VOID_RETURN;
 }
@@ -1590,7 +1608,15 @@ int ha_partition::change_partitions(HA_CREATE_INFO *create_info,
     part_elem->part_state= PART_TO_BE_DROPPED;
   }
   m_new_file= new_file_array;
-  DBUG_RETURN(copy_partitions(copied, deleted));
+  if ((error= copy_partitions(copied, deleted)))
+  {
+    /*
+      Close and unlock the new temporary partitions.
+      They will later be deleted through the ddl-log.
+    */
+    cleanup_new_partition(part_count);
+  }
+  DBUG_RETURN(error);
 }
 
 
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index eaea6e55b09..7eabc1887f9 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -6954,7 +6954,7 @@ bool reload_acl_and_cache(THD *thd, ulong options, TABLE_LIST *tables,
  /*
    If the query was killed then this function must fail.
  */
- return result || thd->killed;
+ return result || (thd ? thd->killed : 0);
 }
 
 
diff --git a/sql/sql_partition.cc b/sql/sql_partition.cc
index e0697f563ea..441adc9ddd9 100644
--- a/sql/sql_partition.cc
+++ b/sql/sql_partition.cc
@@ -5708,8 +5708,7 @@ static bool write_log_drop_partition(ALTER_PARTITION_PARAM_TYPE *lpt)
   part_info->first_log_entry= NULL;
   build_table_filename(path, sizeof(path) - 1, lpt->db,
                        lpt->table_name, "", 0);
-  build_table_filename(tmp_path, sizeof(tmp_path) - 1, lpt->db,
-                       lpt->table_name, "#", 0);
+  build_table_shadow_filename(tmp_path, sizeof(tmp_path) - 1, lpt);
   pthread_mutex_lock(&LOCK_gdl);
   if (write_log_dropped_partitions(lpt, &next_entry, (const char*)path,
                                    FALSE))
@@ -5765,8 +5764,7 @@ static bool write_log_add_change_partition(ALTER_PARTITION_PARAM_TYPE *lpt)
 
   build_table_filename(path, sizeof(path) - 1, lpt->db,
                        lpt->table_name, "", 0);
-  build_table_filename(tmp_path, sizeof(tmp_path) - 1, lpt->db,
-                       lpt->table_name, "#", 0);
+  build_table_shadow_filename(tmp_path, sizeof(tmp_path) - 1, lpt);
   pthread_mutex_lock(&LOCK_gdl);
   if (write_log_dropped_partitions(lpt, &next_entry, (const char*)path,
                                    FALSE))
@@ -5991,7 +5989,7 @@ void handle_alter_part_error(ALTER_PARTITION_PARAM_TYPE *lpt,
   partition_info *part_info= lpt->part_info;
   DBUG_ENTER("handle_alter_part_error");
 
-  if (!part_info->first_log_entry &&
+  if (part_info->first_log_entry &&
       execute_ddl_log_entry(current_thd,
                             part_info->first_log_entry->entry_pos))
   {
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 3b7354111ba..0e36ecfcb46 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -647,7 +647,7 @@ static bool read_ddl_log_file_entry(uint entry_no)
   Write one entry from ddl log file
   SYNOPSIS
     write_ddl_log_file_entry()
-    entry_no                     Entry number to read
+    entry_no                     Entry number to write
   RETURN VALUES
     TRUE                         Error
     FALSE                        Success
@@ -748,10 +748,10 @@ static uint read_ddl_log_header()
     else
       successful_open= TRUE;
   }
-  entry_no= uint4korr(&file_entry_buf[DDL_LOG_NUM_ENTRY_POS]);
-  global_ddl_log.name_len= uint4korr(&file_entry_buf[DDL_LOG_NAME_LEN_POS]);
   if (successful_open)
   {
+    entry_no= uint4korr(&file_entry_buf[DDL_LOG_NUM_ENTRY_POS]);
+    global_ddl_log.name_len= uint4korr(&file_entry_buf[DDL_LOG_NAME_LEN_POS]);
     global_ddl_log.io_size= uint4korr(&file_entry_buf[DDL_LOG_IO_SIZE_POS]);
     DBUG_ASSERT(global_ddl_log.io_size <=
                 sizeof(global_ddl_log.file_entry_buf));
@@ -832,6 +832,7 @@ static bool init_ddl_log()
     goto end;
 
   global_ddl_log.io_size= IO_SIZE;
+  global_ddl_log.name_len= FN_LEN;
   create_ddl_log_file_name(file_name);
   if ((global_ddl_log.file_id= my_create(file_name,
                                          CREATE_MODE,
@@ -884,6 +885,13 @@ static int execute_ddl_log_action(THD *thd, DDL_LOG_ENTRY *ddl_log_entry)
   {
     DBUG_RETURN(FALSE);
   }
+  DBUG_PRINT("ddl_log",
+             ("execute type %c next %u name '%s' from_name '%s' handler '%s'",
+             ddl_log_entry->action_type,
+             ddl_log_entry->next_entry,
+             ddl_log_entry->name,
+             ddl_log_entry->from_name,
+             ddl_log_entry->handler_name));
   handler_name.str= (char*)ddl_log_entry->handler_name;
   handler_name.length= strlen(ddl_log_entry->handler_name);
   init_sql_alloc(&mem_root, TABLE_ALLOC_BLOCK_SIZE, 0); 
@@ -1091,6 +1099,15 @@ bool write_ddl_log_entry(DDL_LOG_ENTRY *ddl_log_entry,
     DBUG_RETURN(TRUE);
   }
   error= FALSE;
+  DBUG_PRINT("ddl_log",
+             ("write type %c next %u name '%s' from_name '%s' handler '%s'",
+             (char) global_ddl_log.file_entry_buf[DDL_LOG_ACTION_TYPE_POS],
+             ddl_log_entry->next_entry,
+             (char*) &global_ddl_log.file_entry_buf[DDL_LOG_NAME_POS],
+             (char*) &global_ddl_log.file_entry_buf[DDL_LOG_NAME_POS
+                                                    + FN_LEN],
+             (char*) &global_ddl_log.file_entry_buf[DDL_LOG_NAME_POS
+                                                    + (2*FN_LEN)]));
   if (write_ddl_log_file_entry((*active_entry)->entry_pos))
   {
     error= TRUE;