MDEV-27753 Incorrect ENGINE type of table after crash for CONNECT table

whenever possible, partitioning should use the full
partition plugin name, not the one byte legacy code.

Normally, ha_partition can get the engine plugin from
table_share->default_part_plugin.

But in some cases, e.g. in DROP TABLE, the table isn't
opened, table_share is NULL, and ha_partition has to parse
the frm, much like dd_frm_type() does.

temporary_tables.cc, sql_table.cc:

When dropping a table, it must be deleted in the engine
first, then frm file. Because frm can be the only true
source of metadata that the engine might need for DROP.

table.cc:

when opening a partitioned table, if the engine for
partitions is not found, do not fallback to MyISAM.

mysql-test/main/drop_bad_db_type.result

@@ -3,34 +3,50 @@ SET debug_dbug='+d,unstable_db_type';
 install soname 'ha_archive';
 create table t1 (a int) engine=archive;
 insert t1 values (1),(2),(3);
+create table t2 (a int) engine=archive partition by hash(a) partitions 3;
 flush tables;
 uninstall soname 'ha_archive';
-select table_schema, table_name from information_schema.tables where table_name like 't1';
-table_schema	test
-table_name	t1
-select table_schema, table_name, engine, version from information_schema.tables where table_name like 't1';
-table_schema	test
-table_name	t1
-engine	ARCHIVE
-version	NULL
+select table_schema, table_name from information_schema.tables where table_name like 't_' order by 1,2;
+table_schema	table_name
+test	t1
+test	t2
+select table_schema, table_name, engine, version from information_schema.tables where table_name like 't_' order by 1,2;
+table_schema	table_name	engine	version
+test	t1	ARCHIVE	NULL
+test	t2	NULL	NULL
 Warnings:
-Level	Warning
-Code	1286
-Message	Unknown storage engine 'ARCHIVE'
-select table_schema, table_name, engine, row_format from information_schema.tables where table_name like 't1';
-table_schema	test
-table_name	t1
-engine	ARCHIVE
-row_format	NULL
+Warning	1033	Incorrect information in file: './test/t2.frm'
+Warning	1286	Unknown storage engine 'ARCHIVE'
+select table_schema, table_name, engine, row_format from information_schema.tables where table_name like 't_' order by 1,2;
+table_schema	table_name	engine	row_format
+test	t1	ARCHIVE	NULL
+test	t2	NULL	NULL
 Warnings:
-Level	Warning
-Code	1286
-Message	Unknown storage engine 'ARCHIVE'
+Warning	1033	Incorrect information in file: './test/t2.frm'
+Warning	1286	Unknown storage engine 'ARCHIVE'
 install soname 'ha_archive';
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` int(11) DEFAULT NULL
+) ENGINE=ARCHIVE DEFAULT CHARSET=latin1
+show create table t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `a` int(11) DEFAULT NULL
+) ENGINE=ARCHIVE DEFAULT CHARSET=latin1
+ PARTITION BY HASH (`a`)
+PARTITIONS 3
 db.opt
 t1.ARZ
 t1.frm
+t2#P#p0.ARZ
+t2#P#p1.ARZ
+t2#P#p2.ARZ
+t2.frm
+t2.par
 drop table t1;
+drop table t2;
 db.opt
 uninstall soname 'ha_archive';
 SET debug_dbug=@saved_dbug;

mysql-test/main/drop_bad_db_type.test

-
+--source include/have_partition.inc
 --source include/have_debug.inc
 
 if (!$HA_ARCHIVE_SO) {
@@ -13,18 +13,25 @@ SET debug_dbug='+d,unstable_db_type';
 install soname 'ha_archive';
 create table t1 (a int) engine=archive;
 insert t1 values (1),(2),(3);
+
+create table t2 (a int) engine=archive partition by hash(a) partitions 3;
+
 flush tables;
 uninstall soname 'ha_archive';
 
---vertical_results
-select table_schema, table_name from information_schema.tables where table_name like 't1';
-select table_schema, table_name, engine, version from information_schema.tables where table_name like 't1';
-select table_schema, table_name, engine, row_format from information_schema.tables where table_name like 't1';
---horizontal_results
+select table_schema, table_name from information_schema.tables where table_name like 't_' order by 1,2;
+--replace_result $mysqld_datadir ./
+select table_schema, table_name, engine, version from information_schema.tables where table_name like 't_' order by 1,2;
+--replace_result $mysqld_datadir ./
+select table_schema, table_name, engine, row_format from information_schema.tables where table_name like 't_' order by 1,2;
 
 install soname 'ha_archive';
+show create table t1;
+show create table t2;
+
 --list_files $mysqld_datadir/test
 drop table t1;
+drop table t2;
 --list_files $mysqld_datadir/test
 uninstall soname 'ha_archive';
 

mysql-test/main/partition_not_blackhole.result

@@ -9,7 +9,7 @@ SHOW TABLES;
 Tables_in_test
 t1
 SHOW CREATE TABLE t1;
-ERROR HY000: Failed to read from the .par file
+ERROR HY000: Incorrect information in file: './test/t1.frm'
 DROP TABLE t1;
 ERROR HY000: Got error 1 "Operation not permitted" from storage engine partition
 t1.frm

mysql-test/main/partition_not_blackhole.test

@@ -17,7 +17,7 @@ let $MYSQLD_DATADIR= `SELECT @@datadir`;
 --copy_file std_data/parts/t1_blackhole.par $MYSQLD_DATADIR/test/t1.par
 SHOW TABLES;
 --replace_result $MYSQLD_DATADIR ./
---error ER_FAILED_READ_FROM_PAR_FILE
+--error ER_NOT_FORM_FILE
 SHOW CREATE TABLE t1;
 
 # The replace is needed for Solaris

sql/ha_partition.cc

@@ -3059,11 +3059,12 @@ bool ha_partition::read_par_file(const char *name)
     @retval true   Failure
 */
 
-bool ha_partition::setup_engine_array(MEM_ROOT *mem_root)
+bool ha_partition::setup_engine_array(MEM_ROOT *mem_root,
+                                      handlerton* first_engine)
 {
   uint i;
   uchar *buff;
-  handlerton **engine_array, *first_engine;
+  handlerton **engine_array;
   enum legacy_db_type db_type, first_db_type;
 
   DBUG_ASSERT(!m_file);
@@ -3073,11 +3074,8 @@ bool ha_partition::setup_engine_array(MEM_ROOT *mem_root)
     DBUG_RETURN(true);
 
   buff= (uchar *) (m_file_buffer + PAR_ENGINES_OFFSET);
-  first_db_type= (enum legacy_db_type) buff[0];
-  first_engine= ha_resolve_by_legacy_type(ha_thd(), first_db_type);
-  if (!first_engine)
-    goto err;
 
+  first_db_type= (enum legacy_db_type) buff[0];
   if (!(m_engine_array= (plugin_ref*)
         alloc_root(&m_mem_root, m_tot_parts * sizeof(plugin_ref))))
     goto err;
@@ -3118,6 +3116,74 @@ bool ha_partition::setup_engine_array(MEM_ROOT *mem_root)
 }
 
 
+handlerton *ha_partition::get_def_part_engine(const char *name)
+{
+  if (table_share)
+  {
+    if (table_share->default_part_plugin)
+      return plugin_data(table_share->default_part_plugin, handlerton *);
+  }
+  else
+  {
+    // DROP TABLE, for example
+    char buff[FN_REFLEN];
+    File file;
+    MY_STAT state;
+    uchar *frm_image= 0;
+    handlerton *hton= 0;
+    bool use_legacy_type= false;
+
+    fn_format(buff, name, "", reg_ext, MY_APPEND_EXT);
+
+    file= mysql_file_open(key_file_frm, buff, O_RDONLY | O_SHARE, MYF(0));
+    if (file < 0)
+      return NULL;
+
+    if (mysql_file_fstat(file, &state, MYF(MY_WME)))
+      goto err;
+    if (state.st_size <= 64)
+      goto err;
+    if (!(frm_image= (uchar*)my_malloc(state.st_size, MYF(MY_WME))))
+      goto err;
+    if (mysql_file_read(file, frm_image, state.st_size, MYF(MY_NABP)))
+      goto err;
+
+    if (frm_image[64] != '/')
+    {
+      const uchar *e2= frm_image + 64;
+      const uchar *e2end = e2 + uint2korr(frm_image + 4);
+      if (e2end > frm_image + state.st_size)
+        goto err;
+      while (e2 + 3 < e2end)
+      {
+        uchar type= *e2++;
+        size_t length= extra2_read_len(&e2, e2end);
+        if (!length)
+          goto err;
+        if (type == EXTRA2_DEFAULT_PART_ENGINE)
+        {
+          LEX_CSTRING name= { (char*)e2, length };
+          plugin_ref plugin= ha_resolve_by_name(ha_thd(), &name, false);
+          if (plugin)
+            hton= plugin_data(plugin, handlerton *);
+          goto err;
+        }
+        e2+= length;
+      }
+    }
+    use_legacy_type= true;
+err:
+    my_free(frm_image);
+    mysql_file_close(file, MYF(0));
+    if (!use_legacy_type)
+      return hton;
+  }
+
+  return ha_resolve_by_legacy_type(ha_thd(),
+                  (enum legacy_db_type)m_file_buffer[PAR_ENGINES_OFFSET]);
+}
+
+
 /**
   Get info about partition engines and their names from the .par file
 
@@ -3145,7 +3211,11 @@ bool ha_partition::get_from_handler_file(const char *name, MEM_ROOT *mem_root,
   if (read_par_file(name))
     DBUG_RETURN(true);
 
-  if (!is_clone && setup_engine_array(mem_root))
+  handlerton *default_engine= get_def_part_engine(name);
+  if (!default_engine)
+    DBUG_RETURN(true);
+
+  if (!is_clone && setup_engine_array(mem_root, default_engine))
     DBUG_RETURN(true);
 
   DBUG_RETURN(false);

sql/ha_partition.h

@@ -580,8 +580,9 @@ class ha_partition :public handler
     And one method to read it in.
   */
   bool create_handler_file(const char *name);
-  bool setup_engine_array(MEM_ROOT *mem_root);
+  bool setup_engine_array(MEM_ROOT *mem_root, handlerton *first_engine);
   bool read_par_file(const char *name);
+  handlerton *get_def_part_engine(const char *name);
   bool get_from_handler_file(const char *name, MEM_ROOT *mem_root,
                              bool is_clone);
   bool new_handlers_from_part_info(MEM_ROOT *mem_root);

sql/sql_table.cc

@@ -2802,15 +2802,13 @@ bool quick_rm_table(THD *thd, handlerton *base, const LEX_CSTRING *db,
                     const LEX_CSTRING *table_name, uint flags, const char *table_path)
 {
   char path[FN_REFLEN + 1];
+  const size_t pathmax = sizeof(path) - 1 - reg_ext_length;
   int error= 0;
   DBUG_ENTER("quick_rm_table");
 
   size_t path_length= table_path ?
-    (strxnmov(path, sizeof(path) - 1, table_path, reg_ext, NullS) - path) :
-    build_table_filename(path, sizeof(path)-1, db->str, table_name->str, reg_ext, flags);
-  if (mysql_file_delete(key_file_frm, path, MYF(0)))
-    error= 1; /* purecov: inspected */
-  path[path_length - reg_ext_length]= '\0'; // Remove reg_ext
+    (strxnmov(path, pathmax, table_path, NullS) - path) :
+    build_table_filename(path, pathmax, db->str, table_name->str, "", flags);
   if (flags & NO_HA_TABLE)
   {
     handler *file= get_new_handler((TABLE_SHARE*) 0, thd->mem_root, base);
@@ -2822,6 +2820,10 @@ bool quick_rm_table(THD *thd, handlerton *base, const LEX_CSTRING *db,
   if (!(flags & (FRM_ONLY|NO_HA_TABLE)))
     error|= ha_delete_table(current_thd, base, path, db, table_name, 0);
 
+  memcpy(path + path_length, reg_ext, reg_ext_length + 1);
+  if (mysql_file_delete(key_file_frm, path, MYF(0)))
+    error= 1;
+
   if (likely(error == 0))
   {
     PSI_CALL_drop_table_share(flags & FN_IS_TMP, db->str, (uint)db->length,

sql/table.cc

@@ -1370,7 +1370,7 @@ int TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
   if (frm_image[61] && !share->default_part_plugin)
   {
     enum legacy_db_type db_type= (enum legacy_db_type) (uint) frm_image[61];
-    share->default_part_plugin= ha_lock_engine(NULL, ha_checktype(thd, db_type));
+    share->default_part_plugin= ha_lock_engine(NULL, ha_checktype(thd, db_type, 1));
     if (!share->default_part_plugin)
       goto err;
   }

sql/temporary_tables.cc

@@ -698,19 +698,20 @@ bool THD::rm_temporary_table(handlerton *base, const char *path)
   char frm_path[FN_REFLEN + 1];
 
   strxnmov(frm_path, sizeof(frm_path) - 1, path, reg_ext, NullS);
-  if (mysql_file_delete(key_file_frm, frm_path, MYF(0)))
-  {
-    error= true;
-  }
-  file= get_new_handler((TABLE_SHARE*) 0, current_thd->mem_root, base);
+
+  file= get_new_handler((TABLE_SHARE*) 0, mem_root, base);
   if (file && file->ha_delete_table(path))
   {
     error= true;
     sql_print_warning("Could not remove temporary table: '%s', error: %d",
                       path, my_errno);
   }
-
   delete file;
+
+  if (mysql_file_delete(key_file_frm, frm_path, MYF(0)))
+  {
+    error= true;
+  }
   DBUG_RETURN(error);
 }
 

sql/unireg.h

@@ -188,6 +188,21 @@ enum extra2_field_flags {
   VERS_OPTIMIZED_UPDATE= 1 << INVISIBLE_MAX_BITS
 };
 
+static inline size_t extra2_read_len(const uchar **extra2, const uchar *end)
+{
+  size_t length= *(*extra2)++;
+  if (length)
+    return length;
+
+  if ((*extra2) + 2 >= end)
+    return 0;
+  length= uint2korr(*extra2);
+  (*extra2)+= 2;
+  if (length < 256 || *extra2 + length > end)
+    return 0;
+  return length;
+}
+
 int rea_create_table(THD *thd, LEX_CUSTRING *frm,
                      const char *path, const char *db, const char *table_name,
                      HA_CREATE_INFO *create_info, handler *file,