Bug#42438: Crash ha_partition::change_table_ptr

There was two problems:
The first was the symptom, caused by bad error handling in
ha_partition. It did not handle print_error etc. when
having no partitions (when used by dummy handler).

The second was the real problem that when dropping tables
it reused the table type (storage engine) from when the lock
was asked for, not the table type that it had when gaining
the exclusive name lock. So that it tried to delete tables
from wrong storage engines.

Solutions for the first problem was to accept some handler
calls to the partitioning handler even if it was not setup
with any partitions, and also if possible fallback
to use the base handler's default functions.

Solution for the second problem was to remove the optimization
to reuse the definition from the cache, instead always check
the frm-file when holding the LOCK_open mutex

(updated with a fix for a debug print crash and better
comments as required by reviewer, and removed optimization
to avoid reading the frm-file).

mysql-test/r/partition_debug_sync.result:
  Bug#42438: Crash ha_partition::change_table_ptr
  
  New result file using DEBUG_SYNC for deterministic results.
mysql-test/t/partition_debug_sync.test:
  Bug#42438: Crash ha_partition::change_table_ptr
  
  New test file using DEBUG_SYNC for deterministic results.
sql/ha_partition.cc:
  Bug#42438: Crash ha_partition::change_table_ptr
  
  allow some handler calls, used by error handling, even when
  no partitions are setup. Fallback to default handling if possible.
sql/sql_base.cc:
  Bug#42438: Crash ha_partition::change_table_ptr
  
  Added DEBUG_SYNC point for deterministic test cases.
sql/sql_table.cc:
  Bug#42438: Crash ha_partition::change_table_ptr
  
  Always use the table type written in the .frm-file
  (i.e. the current table type) when deleting a table.
  
  Moved the check for log-table to not depend of the cache.
  
  Added DEBUG_SYNC points for deterministic test cases.

1 files changed, 32 insertions, 31 deletions

diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 869ae42c98c..5097c0b4945 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -22,6 +22,9 @@
 #include "sp_head.h"
 #include "sql_trigger.h"
 #include "sql_show.h"
+#if defined(ENABLED_DEBUG_SYNC)
+#include "debug_sync.h"
+#endif
 
 #ifdef __WIN__
 #include <io.h>
@@ -1870,30 +1873,6 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
 
   pthread_mutex_lock(&LOCK_open);
 
-  /*
-    If we have the table in the definition cache, we don't have to check the
-    .frm file to find if the table is a normal table (not view) and what
-    engine to use.
-  */
-
-  for (table= tables; table; table= table->next_local)
-  {
-    TABLE_SHARE *share;
-    table->db_type= NULL;
-    if ((share= get_cached_table_share(table->db, table->table_name)))
-      table->db_type= share->db_type();
-
-    /* Disable drop of enabled log tables */
-    if (share && (share->table_category == TABLE_CATEGORY_PERFORMANCE) &&
-        check_if_log_table(table->db_length, table->db,
-                           table->table_name_length, table->table_name, 1))
-    {
-      my_error(ER_BAD_LOG_STATEMENT, MYF(0), "DROP");
-      pthread_mutex_unlock(&LOCK_open);
-      DBUG_RETURN(1);
-    }
-  }
-
   if (!drop_temporary && lock_table_names_exclusively(thd, tables))
   {
     pthread_mutex_unlock(&LOCK_open);
@@ -1904,7 +1883,7 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
   {
     char *db=table->db;
     handlerton *table_type;
-    enum legacy_db_type frm_db_type;
+    enum legacy_db_type frm_db_type= DB_TYPE_UNKNOWN;
 
     DBUG_PRINT("table", ("table_l: '%s'.'%s'  table: 0x%lx  s: 0x%lx",
                          table->db, table->table_name, (long) table->table,
@@ -1945,6 +1924,15 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
       error= 0;
     }
 
+    /* Disable drop of enabled log tables */
+    if (check_if_log_table(table->db_length, table->db,
+                           table->table_name_length, table->table_name, 1))
+    {
+      my_error(ER_BAD_LOG_STATEMENT, MYF(0), "DROP");
+      pthread_mutex_unlock(&LOCK_open);
+      DBUG_RETURN(1);
+    }
+
     /*
       If row-based replication is used and the table is not a
       temporary table, we add the table name to the drop statement
@@ -1969,7 +1957,6 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
       built_query.append("`,");
     }
 
-    table_type= table->db_type;
     if (!drop_temporary)
     {
       TABLE *locked_table;
@@ -1996,9 +1983,9 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
                                         table->internal_tmp_table ?
                                         FN_IS_TMP : 0);
     }
+    DEBUG_SYNC(thd, "rm_table_part2_before_delete_table");
     if (drop_temporary ||
-        ((table_type == NULL &&        
-         access(path, F_OK) &&
+        ((access(path, F_OK) &&
           ha_create_table_from_engine(thd, db, alias)) ||
          (!drop_view &&
           mysql_frm_type(thd, path, &frm_db_type) != FRMTYPE_TABLE)))
@@ -2014,15 +2001,25 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
     else
     {
       char *end;
-      if (table_type == NULL)
+      /*
+        Cannot use the db_type from the table, since that might have changed
+        while waiting for the exclusive name lock. We are under LOCK_open,
+        so reading from the frm-file is safe.
+      */
+      if (frm_db_type == DB_TYPE_UNKNOWN)
       {
-	mysql_frm_type(thd, path, &frm_db_type);
-        table_type= ha_resolve_by_legacy_type(thd, frm_db_type);
+        mysql_frm_type(thd, path, &frm_db_type);
+        DBUG_PRINT("info", ("frm_db_type %d from %s", frm_db_type, path));
       }
+      table_type= ha_resolve_by_legacy_type(thd, frm_db_type);
       // Remove extension for delete
       *(end= path + path_length - reg_ext_length)= '\0';
+      DBUG_PRINT("info", ("deleting table of type %d",
+                          (table_type ? table_type->db_type : 0)));
       error= ha_delete_table(thd, table_type, path, db, table->table_name,
                              !dont_log_query);
+
+      /* No error if non existent table and 'IF EXIST' clause or view */
       if ((error == ENOENT || error == HA_ERR_NO_SUCH_TABLE) && 
 	  (if_exists || table_type == NULL))
       {
@@ -2062,6 +2059,7 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
     on the table name.
   */
   pthread_mutex_unlock(&LOCK_open);
+  DEBUG_SYNC(thd, "rm_table_part2_before_binlog");
   thd->thread_specific_used|= tmp_table_deleted;
   error= 0;
   if (wrong_tables.length())
@@ -7097,6 +7095,7 @@ view_err:
   else
     create_info->data_file_name=create_info->index_file_name=0;
 
+  DEBUG_SYNC(thd, "alter_table_before_create_table_no_lock");
   /*
     Create a table with a temporary name.
     With create_info->frm_only == 1 this creates a .frm file only.
@@ -7296,6 +7295,7 @@ view_err:
     intern_close_table(new_table);
     my_free(new_table,MYF(0));
   }
+  DEBUG_SYNC(thd, "alter_table_before_rename_result_table");
   VOID(pthread_mutex_lock(&LOCK_open));
   if (error)
   {
@@ -7438,6 +7438,7 @@ view_err:
   thd_proc_info(thd, "end");
 
   DBUG_EXECUTE_IF("sleep_alter_before_main_binlog", my_sleep(6000000););
+  DEBUG_SYNC(thd, "alter_table_before_main_binlog");
 
   ha_binlog_log_query(thd, create_info->db_type, LOGCOM_ALTER_TABLE,
                       thd->query(), thd->query_length(),