Bug#6877 MySQL should give an error if the requested table type is not available

  Implement new SQL mode - NO_ENGINE_SUBSTITUTION

Author: acurtis@xiphis.org

mysql-test/r/sql_mode.result

@@ -402,4 +402,22 @@ a\b	a\"b	a'\b	a'\"b
 SELECT "a\\b", "a\\\'b", "a""\\b", "a""\\\'b";
 a\b	a\'b	a"\b	a"\'b
 a\b	a\'b	a"\b	a"\'b
+set session sql_mode = 'NO_ENGINE_SUBSTITUTION';
+create table t1 (a int) engine=isam;
+ERROR HY000: The 'ISAM' feature is disabled; you need MySQL built with 'ISAM' to have it working
+show create table t1;
+ERROR 42S02: Table 'test.t1' doesn't exist
+drop table if exists t1;
+Warnings:
+Note	1051	Unknown table 't1'
+set session sql_mode = '';
+create table t1 (a int) engine=isam;
+Warnings:
+Warning	1266	Using storage engine MyISAM for table 't1'
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` int(11) default NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+drop table t1;
 SET @@SQL_MODE=@OLD_SQL_MODE;

mysql-test/t/sql_mode.test

@@ -189,4 +189,24 @@ SET @@SQL_MODE='';
 SELECT 'a\\b', 'a\\\"b', 'a''\\b', 'a''\\\"b';
 SELECT "a\\b", "a\\\'b", "a""\\b", "a""\\\'b";
 
+
+#
+# Bug#6877: MySQL should give an error if the requested table type 
+#           is not available
+#
+
+set session sql_mode = 'NO_ENGINE_SUBSTITUTION';
+--error 1289
+create table t1 (a int) engine=isam;
+--error 1146
+show create table t1;
+drop table if exists t1;
+
+# for comparison, lets see the warnings...
+set session sql_mode = '';
+create table t1 (a int) engine=isam;
+show create table t1;
+drop table t1;
+
+
 SET @@SQL_MODE=@OLD_SQL_MODE;

sql/handler.cc

@@ -165,12 +165,22 @@ my_bool ha_storage_engine_is_enabled(enum db_type database_type)
 
 	/* Use other database handler if databasehandler is not incompiled */
 
-enum db_type ha_checktype(enum db_type database_type)
+enum db_type ha_checktype(THD *thd, enum db_type database_type,
+                          bool no_substitute, bool report_error)
 {
-  THD *thd;
   if (ha_storage_engine_is_enabled(database_type))
     return database_type;
 
+  if (no_substitute)
+  {
+    if (report_error)
+    {
+      const char *engine_name= ha_get_storage_engine(database_type);
+      my_error(ER_FEATURE_DISABLED,MYF(0),engine_name,engine_name);
+    }
+    return DB_TYPE_UNKNOWN;
+  }
+
   switch (database_type) {
 #ifndef NO_HASH
   case DB_TYPE_HASH:
@@ -182,7 +192,6 @@ enum db_type ha_checktype(enum db_type database_type)
     break;
   }
 
-  thd= current_thd;
   return ((enum db_type) thd->variables.table_type != DB_TYPE_UNKNOWN ?
           (enum db_type) thd->variables.table_type :
           ((enum db_type) global_system_variables.table_type !=

sql/handler.h

@@ -813,7 +813,8 @@ extern ulong total_ha, total_ha_2pc;
 enum db_type ha_resolve_by_name(const char *name, uint namelen);
 const char *ha_get_storage_engine(enum db_type db_type);
 handler *get_new_handler(TABLE *table, enum db_type db_type);
-enum db_type ha_checktype(enum db_type database_type);
+enum db_type ha_checktype(THD *thd, enum db_type database_type,
+                          bool no_substitute, bool report_error);
 
 /* basic stuff */
 int ha_init(void);

sql/mysql_priv.h

@@ -325,6 +325,7 @@ extern CHARSET_INFO *national_charset_info, *table_alias_charset;
 #define MODE_TRADITIONAL		(MODE_ERROR_FOR_DIVISION_BY_ZERO*2)
 #define MODE_NO_AUTO_CREATE_USER	(MODE_TRADITIONAL*2)
 #define MODE_HIGH_NOT_PRECEDENCE	(MODE_NO_AUTO_CREATE_USER*2)
+#define MODE_NO_ENGINE_SUBSTITUTION     (MODE_HIGH_NOT_PRECEDENCE*2)
 /*
   Replication uses 8 bytes to store SQL_MODE in the binary log. The day you
   use strictly more than 64 bits by adding one more define above, you should
@@ -1232,7 +1233,7 @@ int openfrm(THD *thd, const char *name,const char *alias,uint filestat,
 int readfrm(const char *name, const void** data, uint* length);
 int writefrm(const char* name, const void* data, uint len);
 int closefrm(TABLE *table);
-db_type get_table_type(const char *name);
+db_type get_table_type(THD *thd, const char *name);
 int read_string(File file, gptr *to, uint length);
 void free_blobs(TABLE *table);
 int set_zone(int nr,int min_zone,int max_zone);
@@ -1289,7 +1290,7 @@ ulong make_new_entry(File file,uchar *fileinfo,TYPELIB *formnames,
 		     const char *newname);
 ulong next_io_size(ulong pos);
 void append_unescaped(String *res, const char *pos, uint length);
-int create_frm(char *name,uint reclength,uchar *fileinfo,
+int create_frm(THD *thd, char *name,uint reclength,uchar *fileinfo,
 	       HA_CREATE_INFO *create_info, uint keys);
 void update_create_info_from_table(HA_CREATE_INFO *info, TABLE *form);
 int rename_file_ext(const char * from,const char * to,const char * ext);

sql/mysqld.cc

@@ -229,6 +229,7 @@ static const char *sql_mode_names[] =
   "NO_AUTO_VALUE_ON_ZERO", "NO_BACKSLASH_ESCAPES", "STRICT_TRANS_TABLES", "STRICT_ALL_TABLES",
   "NO_ZERO_IN_DATE", "NO_ZERO_DATE", "ALLOW_INVALID_DATES", "ERROR_FOR_DIVISION_BY_ZERO",
   "TRADITIONAL", "NO_AUTO_CREATE_USER", "HIGH_NOT_PRECEDENCE",
+  "NO_ENGINE_SUBSTITUTION",
   NullS
 };
 TYPELIB sql_mode_typelib= { array_elements(sql_mode_names)-1,"",

sql/set_var.cc

@@ -3131,7 +3131,7 @@ bool sys_var_thd_storage_engine::check(THD *thd, set_var *var)
     if (!(res=var->value->val_str(&str)) ||
 	!(var->save_result.ulong_value=
           (ulong) (db_type= ha_resolve_by_name(res->ptr(), res->length()))) ||
-        ha_checktype(db_type) != db_type)
+        ha_checktype(thd, db_type, 1, 0) != db_type)
     {
       value= res ? res->c_ptr() : "NULL";
       goto err;

sql/sql_delete.cc

@@ -790,7 +790,7 @@ bool mysql_truncate(THD *thd, TABLE_LIST *table_list, bool dont_send_ok)
   if (!dont_send_ok)
   {
     db_type table_type;
-    if ((table_type=get_table_type(path)) == DB_TYPE_UNKNOWN)
+    if ((table_type=get_table_type(thd, path)) == DB_TYPE_UNKNOWN)
     {
       my_error(ER_NO_SUCH_TABLE, MYF(0),
                table_list->db, table_list->table_name);

sql/sql_rename.cc

@@ -164,7 +164,7 @@ rename_tables(THD *thd, TABLE_LIST *table_list, bool skip_error)
 	    ren_table->db, old_alias,
 	    reg_ext);
     unpack_filename(name, name);
-    if ((table_type=get_table_type(name)) == DB_TYPE_UNKNOWN)
+    if ((table_type=get_table_type(thd, name)) == DB_TYPE_UNKNOWN)
     {
       my_error(ER_FILE_NOT_FOUND, MYF(0), name, my_errno);
       if (!skip_error)

sql/sql_table.cc

@@ -39,6 +39,8 @@ static int copy_data_between_tables(TABLE *from,TABLE *to,
 				    uint order_num, ORDER *order,
 				    ha_rows *copied,ha_rows *deleted);
 static bool prepare_blob_field(THD *thd, create_field *sql_field);
+static bool check_engine(THD *thd, const char *table_name,
+                         enum db_type *new_engine);                             
 
 
 /*
@@ -268,7 +270,7 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
     else
     {
       char *end;
-      db_type table_type= get_table_type(path);
+      db_type table_type= get_table_type(thd, path);
       *(end=fn_ext(path))=0;			// Remove extension for delete
       error= ha_delete_table(thd, table_type, path, table->table_name,
                              !dont_log_query);
@@ -1490,7 +1492,6 @@ bool mysql_create_table(THD *thd,const char *db, const char *table_name,
   KEY		*key_info_buffer;
   handler	*file;
   bool		error= TRUE;
-  enum db_type	new_db_type;
   DBUG_ENTER("mysql_create_table");
 
   /* Check for duplicate fields and check type of table to create */
@@ -1500,16 +1501,8 @@ bool mysql_create_table(THD *thd,const char *db, const char *table_name,
                MYF(0));
     DBUG_RETURN(TRUE);
   }
-  if ((new_db_type= ha_checktype(create_info->db_type)) !=
-      create_info->db_type)
-  {
-    create_info->db_type= new_db_type;
-    push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
-			ER_WARN_USING_OTHER_HANDLER,
-			ER(ER_WARN_USING_OTHER_HANDLER),
-			ha_get_storage_engine(new_db_type),
-			table_name);
-  }
+  if (check_engine(thd, table_name, &create_info->db_type))
+    DBUG_RETURN(TRUE);
   db_options= create_info->table_options;
   if (create_info->row_type == ROW_TYPE_DYNAMIC)
     db_options|=HA_OPTION_PACK_RECORD;
@@ -3125,16 +3118,9 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
   old_db_type= table->s->db_type;
   if (create_info->db_type == DB_TYPE_DEFAULT)
     create_info->db_type= old_db_type;
-  if ((new_db_type= ha_checktype(create_info->db_type)) !=
-      create_info->db_type)
-  {
-    create_info->db_type= new_db_type;
-    push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
-			ER_WARN_USING_OTHER_HANDLER,
-			ER(ER_WARN_USING_OTHER_HANDLER),
-			ha_get_storage_engine(new_db_type),
-			new_name);
-  }
+  if (check_engine(thd, new_name, &create_info->db_type))
+    DBUG_RETURN(TRUE);
+  new_db_type= create_info->db_type;
   if (create_info->row_type == ROW_TYPE_NOT_USED)
     create_info->row_type= table->s->row_type;
 
@@ -4132,3 +4118,24 @@ bool mysql_checksum_table(THD *thd, TABLE_LIST *tables, HA_CHECK_OPT *check_opt)
     table->table=0;
   DBUG_RETURN(TRUE);
 }
+
+static bool check_engine(THD *thd, const char *table_name,
+                         enum db_type *new_engine)
+{
+  enum db_type req_engine= *new_engine;
+  bool no_substitution= 
+        test(thd->variables.sql_mode & MODE_NO_ENGINE_SUBSTITUTION);
+  if ((*new_engine= 
+       ha_checktype(thd, req_engine, no_substitution, 1)) == DB_TYPE_UNKNOWN)
+    return TRUE;
+
+  if (req_engine != *new_engine)
+  {
+    push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
+                       ER_WARN_USING_OTHER_HANDLER,
+                       ER(ER_WARN_USING_OTHER_HANDLER),
+                       ha_get_storage_engine(*new_engine),
+                       table_name);
+  }
+  return FALSE;
+}

sql/table.cc

@@ -163,7 +163,7 @@ int openfrm(THD *thd, const char *name, const char *alias, uint db_stat,
   if (share->frm_version == FRM_VER_TRUE_VARCHAR -1 && head[33] == 5)
     share->frm_version= FRM_VER_TRUE_VARCHAR;
 
-  share->db_type= ha_checktype((enum db_type) (uint) *(head+3));
+  share->db_type= ha_checktype(thd,(enum db_type) (uint) *(head+3),0,0);
   share->db_create_options= db_create_options=uint2korr(head+30);
   share->db_options_in_use= share->db_create_options;
   share->mysql_version= uint4korr(head+51);
@@ -1341,8 +1341,8 @@ void append_unescaped(String *res, const char *pos, uint length)
 
 	/* Create a .frm file */
 
-File create_frm(register my_string name, uint reclength, uchar *fileinfo,
-		HA_CREATE_INFO *create_info, uint keys)
+File create_frm(THD *thd, register my_string name, uint reclength,
+		uchar *fileinfo, HA_CREATE_INFO *create_info, uint keys)
 {
   register File file;
   ulong length;
@@ -1375,7 +1375,7 @@ File create_frm(register my_string name, uint reclength, uchar *fileinfo,
     fileinfo[1]= 1;
     fileinfo[2]= FRM_VER+3+ test(create_info->varchar);
 
-    fileinfo[3]= (uchar) ha_checktype(create_info->db_type);
+    fileinfo[3]= (uchar) ha_checktype(thd,create_info->db_type,0,0);
     fileinfo[4]=1;
     int2store(fileinfo+6,IO_SIZE);		/* Next block starts here */
     key_length=keys*(7+NAME_LEN+MAX_REF_PARTS*9)+16;
@@ -1637,7 +1637,7 @@ bool check_column_name(const char *name)
 ** Get type of table from .frm file
 */
 
-db_type get_table_type(const char *name)
+db_type get_table_type(THD *thd, const char *name)
 {
   File	 file;
   uchar head[4];
@@ -1653,7 +1653,7 @@ db_type get_table_type(const char *name)
       (head[2] != FRM_VER && head[2] != FRM_VER+1 &&
        (head[2] < FRM_VER+3 || head[2] > FRM_VER+4)))
     DBUG_RETURN(DB_TYPE_UNKNOWN);
-  DBUG_RETURN(ha_checktype((enum db_type) (uint) *(head+3)));
+  DBUG_RETURN(ha_checktype(thd,(enum db_type) (uint) *(head+3),0,0));
 }
 
 

sql/unireg.cc

@@ -113,7 +113,7 @@ bool mysql_create_frm(THD *thd, my_string file_name,
   }
   reclength=uint2korr(forminfo+266);
 
-  if ((file=create_frm(file_name, reclength, fileinfo,
+  if ((file=create_frm(thd, file_name, reclength, fileinfo,
 		       create_info, keys)) < 0)
   {
     my_free((gptr) screen_buff,MYF(0));