Bug#11815557 60269: MYSQL SHOULD REJECT ATTEMPTS TO CREATE SYSTEM

                    TABLES IN INCORRECT ENGINE

PROBLEM:
  CREATE/ALTER TABLE currently can move system tables like
mysql.db, user, host etc, to engines other than MyISAM. This is not
completely supported as of now, by mysqld. When some of system tables
like plugin, servers, event, func, *_priv, time_zone* are moved
to innodb, mysqld restart crashes. Currently system tables
can be moved to BLACKHOLE also!!!.

ANALYSIS:
  The problem is that there is no check before creating or moving
a system table to some particular engine.

  System tables are suppose to be residing in MyISAM. We can think
of restricting system tables to exist only in MyISAM. But, there could
be future needs of these system tables to be part of other engines
by design. For eg, NDB cluster expects some tables to be on innodb
or ndb engine. This calls for a solution, by which system
tables can be supported by any desired engine, with minimal effort.

FIX:
  The solution provides a handlerton interface using which,
mysqld server can query particular storage engine handlerton for
system tables that it supports. This way each storage engine
layer can define their own system database and system tables.

  The check_engine() function uses the new handlerton function
ha_check_if_supported_system_table() to check if db.tablename
provided in the DDL is supported by the SE.

Note: This fix has modified a test in help.test, which was moving
mysql.help_* to innodb. The primary intention of the test was not
to move them between engines.

10 files changed, 628 insertions, 134 deletions

diff --git a/mysql-test/r/alter_table.result b/mysql-test/r/alter_table.result
index ab6793235c7..236d07bccbc 100644
--- a/mysql-test/r/alter_table.result
+++ b/mysql-test/r/alter_table.result
@@ -1433,3 +1433,58 @@ rename table t2 to t1;
 execute stmt1;
 deallocate prepare stmt1;
 drop table t2;
+#
+# Bug#11815557 60269: MYSQL SHOULD REJECT ATTEMPTS TO CREATE SYSTEM
+#                     TABLES IN INCORRECT ENGINE
+#
+# Note: This test assumes that only MyISAM supports system tables.
+#       If other engines are made to support system tables,
+#       then this test needs to be updated
+#
+use mysql;
+ALTER TABLE db ENGINE=innodb;
+ERROR HY000: Storage engine 'InnoDB' does not support system tables. [mysql.db]
+ALTER TABLE user ENGINE=memory;
+ERROR HY000: Storage engine 'MEMORY' does not support system tables. [mysql.user]
+ALTER TABLE proc ENGINE=heap;
+ERROR HY000: Storage engine 'MEMORY' does not support system tables. [mysql.proc]
+ALTER TABLE func ENGINE=csv;
+ERROR HY000: Storage engine 'CSV' does not support system tables. [mysql.func]
+ALTER TABLE event ENGINE=merge;
+ERROR HY000: Storage engine 'MRG_MYISAM' does not support system tables. [mysql.event]
+ALTER TABLE servers ENGINE=innodb;
+ERROR HY000: Storage engine 'InnoDB' does not support system tables. [mysql.servers]
+ALTER TABLE procs_priv ENGINE=memory;
+ERROR HY000: Storage engine 'MEMORY' does not support system tables. [mysql.procs_priv]
+ALTER TABLE tables_priv ENGINE=heap;
+ERROR HY000: Storage engine 'MEMORY' does not support system tables. [mysql.tables_priv]
+ALTER TABLE columns_priv ENGINE=csv;
+ERROR HY000: Storage engine 'CSV' does not support system tables. [mysql.columns_priv]
+ALTER TABLE time_zone ENGINE=merge;
+ERROR HY000: Storage engine 'MRG_MYISAM' does not support system tables. [mysql.time_zone]
+ALTER TABLE help_topic ENGINE=innodb;
+ERROR HY000: Storage engine 'InnoDB' does not support system tables. [mysql.help_topic]
+CREATE TABLE db (dummy int) ENGINE=innodb;
+ERROR HY000: Storage engine 'InnoDB' does not support system tables. [mysql.db]
+CREATE TABLE user (dummy int) ENGINE=memory;
+ERROR HY000: Storage engine 'MEMORY' does not support system tables. [mysql.user]
+CREATE TABLE proc (dummy int) ENGINE=heap;
+ERROR HY000: Storage engine 'MEMORY' does not support system tables. [mysql.proc]
+CREATE TABLE func (dummy int) ENGINE=csv;
+ERROR HY000: Storage engine 'CSV' does not support system tables. [mysql.func]
+CREATE TABLE event (dummy int) ENGINE=merge;
+ERROR HY000: Storage engine 'MRG_MYISAM' does not support system tables. [mysql.event]
+CREATE TABLE servers (dummy int) ENGINE=innodb;
+ERROR HY000: Storage engine 'InnoDB' does not support system tables. [mysql.servers]
+CREATE TABLE procs_priv (dummy int) ENGINE=memory;
+ERROR HY000: Storage engine 'MEMORY' does not support system tables. [mysql.procs_priv]
+CREATE TABLE tables_priv (dummy int) ENGINE=heap;
+ERROR HY000: Storage engine 'MEMORY' does not support system tables. [mysql.tables_priv]
+CREATE TABLE columns_priv (dummy int) ENGINE=csv;
+ERROR HY000: Storage engine 'CSV' does not support system tables. [mysql.columns_priv]
+CREATE TABLE time_zone (dummy int) ENGINE=merge;
+ERROR HY000: Storage engine 'MRG_MYISAM' does not support system tables. [mysql.time_zone]
+CREATE TABLE help_topic (dummy int) ENGINE=innodb;
+ERROR HY000: Storage engine 'InnoDB' does not support system tables. [mysql.help_topic]
+use test;
+# End of Bug#11815557
diff --git a/mysql-test/r/help.result b/mysql-test/r/help.result
index 16719cc8193..8ec69d0c7be 100644
--- a/mysql-test/r/help.result
+++ b/mysql-test/r/help.result
@@ -148,100 +148,6 @@ help 'impossible_category_1';
 source_category_name	name	is_it_category
 impossible_category_1	impossible_function_1	N
 impossible_category_1	impossible_function_2	N
-alter table mysql.help_relation engine=innodb;
-alter table mysql.help_keyword engine=innodb;
-alter table mysql.help_topic engine=innodb;
-alter table mysql.help_category engine=innodb;
-help 'function_of_my_dream';
-name	is_it_category
-help '%possible_f%';
-name	is_it_category
-impossible_function_1	N
-impossible_function_2	N
-impossible_function_3	N
-impossible_function_4	N
-impossible_function_7	N
-help 'impossible_func%';
-name	is_it_category
-impossible_function_1	N
-impossible_function_2	N
-impossible_function_3	N
-impossible_function_4	N
-impossible_function_7	N
-help 'impossible_category%';
-name	is_it_category
-impossible_category_1	Y
-impossible_category_2	Y
-impossible_category_3	Y
-help 'impossible_%';
-name	is_it_category
-impossible_function_1	N
-impossible_function_2	N
-impossible_function_3	N
-impossible_function_4	N
-impossible_function_7	N
-impossible_category_1	Y
-impossible_category_2	Y
-impossible_category_3	Y
-help '%function_1';
-name	description	example
-impossible_function_1	description of 
- impossible_function1
-	example of 
- impossible_function1
-help '%function_2';
-name	description	example
-impossible_function_2	description of 
- impossible_function2
-	example of 
- impossible_function2
-help '%function_3';
-name	description	example
-impossible_function_3	description of 
- impossible_function3
-	example of 
- impossible_function3
-help '%function_4';
-name	description	example
-impossible_function_4	description of 
- impossible_function4
-	example of 
- impossible_function4
-help '%function_5';
-name	description	example
-impossible_function_1	description of 
- impossible_function1
-	example of 
- impossible_function1
-help '%function_6';
-name	is_it_category
-impossible_function_3	N
-impossible_function_4	N
-help '%function_7';
-name	description	example
-impossible_function_7	description of 
- impossible_function5
-	example of 
- impossible_function7
-help '%category_2';
-source_category_name	name	is_it_category
-impossible_category_2	impossible_function_3	N
-impossible_category_2	impossible_function_4	N
-impossible_category_2	impossible_category_3	Y
-help 'impossible_function_1';
-name	description	example
-impossible_function_1	description of 
- impossible_function1
-	example of 
- impossible_function1
-help 'impossible_category_1';
-source_category_name	name	is_it_category
-impossible_category_1	impossible_function_1	N
-impossible_category_1	impossible_function_2	N
-alter table mysql.help_relation engine=myisam;
-alter table mysql.help_keyword engine=myisam;
-alter table mysql.help_topic engine=myisam;
-alter table mysql.help_category engine=myisam;
 delete from mysql.help_topic where help_topic_id=@topic1_id;
 delete from mysql.help_topic where help_topic_id=@topic2_id;
 delete from mysql.help_topic where help_topic_id=@topic3_id;
diff --git a/mysql-test/t/alter_table.test b/mysql-test/t/alter_table.test
index 5cdd00b6afc..f5d5538c217 100644
--- a/mysql-test/t/alter_table.test
+++ b/mysql-test/t/alter_table.test
@@ -1203,3 +1203,61 @@ rename table t2 to t1;
 execute stmt1;
 deallocate prepare stmt1;
 drop table t2;
+
+--echo #
+--echo # Bug#11815557 60269: MYSQL SHOULD REJECT ATTEMPTS TO CREATE SYSTEM
+--echo #                     TABLES IN INCORRECT ENGINE
+--echo #
+--echo # Note: This test assumes that only MyISAM supports system tables.
+--echo #       If other engines are made to support system tables,
+--echo #       then this test needs to be updated
+--echo #
+
+use mysql;
+--error ER_UNSUPPORTED_ENGINE
+ALTER TABLE db ENGINE=innodb;
+--error ER_UNSUPPORTED_ENGINE
+ALTER TABLE user ENGINE=memory;
+--error ER_UNSUPPORTED_ENGINE
+ALTER TABLE proc ENGINE=heap;
+--error ER_UNSUPPORTED_ENGINE
+ALTER TABLE func ENGINE=csv;
+--error ER_UNSUPPORTED_ENGINE
+ALTER TABLE event ENGINE=merge;
+--error ER_UNSUPPORTED_ENGINE
+ALTER TABLE servers ENGINE=innodb;
+--error ER_UNSUPPORTED_ENGINE
+ALTER TABLE procs_priv ENGINE=memory;
+--error ER_UNSUPPORTED_ENGINE
+ALTER TABLE tables_priv ENGINE=heap;
+--error ER_UNSUPPORTED_ENGINE
+ALTER TABLE columns_priv ENGINE=csv;
+--error ER_UNSUPPORTED_ENGINE
+ALTER TABLE time_zone ENGINE=merge;
+--error ER_UNSUPPORTED_ENGINE
+ALTER TABLE help_topic ENGINE=innodb;
+
+--error ER_UNSUPPORTED_ENGINE
+CREATE TABLE db (dummy int) ENGINE=innodb;
+--error ER_UNSUPPORTED_ENGINE
+CREATE TABLE user (dummy int) ENGINE=memory;
+--error ER_UNSUPPORTED_ENGINE
+CREATE TABLE proc (dummy int) ENGINE=heap;
+--error ER_UNSUPPORTED_ENGINE
+CREATE TABLE func (dummy int) ENGINE=csv;
+--error ER_UNSUPPORTED_ENGINE
+CREATE TABLE event (dummy int) ENGINE=merge;
+--error ER_UNSUPPORTED_ENGINE
+CREATE TABLE servers (dummy int) ENGINE=innodb;
+--error ER_UNSUPPORTED_ENGINE
+CREATE TABLE procs_priv (dummy int) ENGINE=memory;
+--error ER_UNSUPPORTED_ENGINE
+CREATE TABLE tables_priv (dummy int) ENGINE=heap;
+--error ER_UNSUPPORTED_ENGINE
+CREATE TABLE columns_priv (dummy int) ENGINE=csv;
+--error ER_UNSUPPORTED_ENGINE
+CREATE TABLE time_zone (dummy int) ENGINE=merge;
+--error ER_UNSUPPORTED_ENGINE
+CREATE TABLE help_topic (dummy int) ENGINE=innodb;
+use test;
+--echo # End of Bug#11815557
diff --git a/mysql-test/t/help.test b/mysql-test/t/help.test
index 71821e46771..6fe3a71f23f 100644
--- a/mysql-test/t/help.test
+++ b/mysql-test/t/help.test
@@ -63,38 +63,6 @@ help 'impossible_function_1';
 help 'impossible_category_1';
 ##############
 
---disable_warnings
-alter table mysql.help_relation engine=innodb;
-alter table mysql.help_keyword engine=innodb;
-alter table mysql.help_topic engine=innodb;
-alter table mysql.help_category engine=innodb;
---enable_warnings
-
-##############
-help 'function_of_my_dream';
-help '%possible_f%';
-help 'impossible_func%';
-help 'impossible_category%';
-help 'impossible_%';
-
-help '%function_1';
-help '%function_2';
-help '%function_3';
-help '%function_4';
-help '%function_5';
-help '%function_6';
-help '%function_7';
-
-help '%category_2';
-help 'impossible_function_1';
-help 'impossible_category_1';
-##############
-
-alter table mysql.help_relation engine=myisam;
-alter table mysql.help_keyword engine=myisam;
-alter table mysql.help_topic engine=myisam;
-alter table mysql.help_category engine=myisam;
-
 delete from mysql.help_topic where help_topic_id=@topic1_id;
 delete from mysql.help_topic where help_topic_id=@topic2_id;
 delete from mysql.help_topic where help_topic_id=@topic3_id;
diff --git a/sql/handler.cc b/sql/handler.cc
index b6df46ed48d..c3c70109daa 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -45,6 +45,7 @@
 #include "ha_partition.h"
 #endif
 
+
 /*
   While we have legacy_db_type, we have this array to
   check for dups and to find handlerton from legacy_db_type.
@@ -90,6 +91,83 @@ TYPELIB tx_isolation_typelib= {array_elements(tx_isolation_names)-1,"",
 static TYPELIB known_extensions= {0,"known_exts", NULL, NULL};
 uint known_extensions_id= 0;
 
+/**
+  Database name that hold most of mysqld system tables.
+  Current code assumes that, there exists only some
+  specific "database name" designated as system database.
+*/
+const char* mysqld_system_database= "mysql";
+
+// System tables that belong to mysqld_system_database.
+st_system_tablename mysqld_system_tables[]= {
+  {mysqld_system_database, "db"},
+  {mysqld_system_database, "user"},
+  {mysqld_system_database, "host"},
+  {mysqld_system_database, "func"},
+  {mysqld_system_database, "proc"},
+  {mysqld_system_database, "event"},
+  {mysqld_system_database, "plugin"},
+  {mysqld_system_database, "servers"},
+  {mysqld_system_database, "procs_priv"},
+  {mysqld_system_database, "tables_priv"},
+  {mysqld_system_database, "proxies_priv"},
+  {mysqld_system_database, "columns_priv"},
+  {mysqld_system_database, "time_zone"},
+  {mysqld_system_database, "time_zone_name"},
+  {mysqld_system_database, "time_zone_leap_second"},
+  {mysqld_system_database, "time_zone_transition"},
+  {mysqld_system_database, "time_zone_transition_type"},
+  {mysqld_system_database, "help_category"},
+  {mysqld_system_database, "help_keyword"},
+  {mysqld_system_database, "help_relation"},
+  {mysqld_system_database, "help_topic"},
+  {(const char *)NULL, (const char *)NULL} /* This must be at the end */
+};
+
+/**
+  This static pointer holds list of system databases from SQL layer and
+  various SE's. The required memory is allocated once, and never freed.
+*/
+static const char **known_system_databases= NULL;
+static const char **ha_known_system_databases();
+
+// Called for each SE to get SE specific system database.
+static my_bool system_databases_handlerton(THD *unused, plugin_ref plugin,
+                                           void *arg);
+
+// Called for each SE to check if given db.table_name is a system table.
+static my_bool check_engine_system_table_handlerton(THD *unused,
+                                                    plugin_ref plugin,
+                                                    void *arg);
+
+/**
+  Structure used by SE during check for system table.
+  This structure is passed to each SE handlerton and the status (OUT param)
+  is collected.
+*/
+struct st_sys_tbl_chk_params
+{
+  const char *db;                             // IN param
+  const char *table_name;                     // IN param
+  bool is_sql_layer_system_table;             // IN param
+  legacy_db_type db_type;                     // IN param
+
+  enum enum_sys_tbl_chk_status
+  {
+    // db.table_name is not a supported system table.
+    NOT_KNOWN_SYSTEM_TABLE,
+    /*
+      db.table_name is a system table,
+      but may not be supported by SE.
+    */
+    KNOWN_SYSTEM_TABLE,
+    /*
+      db.table_name is a system table,
+      and is supported by SE.
+    */
+    SUPPORTED_SYSTEM_TABLE
+  }status;                                    // OUT param
+};
 
 
 static plugin_ref ha_default_plugin(THD *thd)
@@ -587,6 +665,14 @@ int ha_init()
   */
   opt_using_transactions= total_ha>(ulong)opt_bin_log;
   savepoint_alloc_size+= sizeof(SAVEPOINT);
+
+  /*
+    Initialize system database name cache.
+    This cache is used to do a quick check if a given
+    db.tablename is a system table.
+  */
+  known_system_databases= ha_known_system_databases();
+
   DBUG_RETURN(error);
 }
 
@@ -3784,6 +3870,227 @@ ha_check_if_table_exists(THD* thd, const char *db, const char *name,
   DBUG_RETURN(FALSE);
 }
 
+/**
+  @brief Check if a given table is a system table.
+
+  @details The primary purpose of introducing this function is to stop system
+  tables to be created or being moved to undesired storage engines.
+
+  @todo There is another function called is_system_table_name() used by
+        get_table_category(), which is used to set TABLE_SHARE table_category.
+        It checks only a subset of table name like proc, event and time*.
+        We cannot use below function in get_table_category(),
+        as that affects locking mechanism. If we need to
+        unify these functions, we need to fix locking issues generated.
+
+  @param   hton                  Handlerton of new engine.
+  @param   db                    Database name.
+  @param   table_name            Table name to be checked.
+
+  @return Operation status
+    @retval  true                If the table name is a valid system table
+                                 or if its a valid user table.
+
+    @retval  false               If the table name is a system table name
+                                 and does not belong to engine specified
+                                 in the command.
+*/
+bool ha_check_if_supported_system_table(handlerton *hton, const char *db,
+                                        const char *table_name)
+{
+  DBUG_ENTER("ha_check_if_supported_system_table");
+  st_sys_tbl_chk_params check_params;
+  bool is_system_database= false;
+  const char **names;
+  st_system_tablename *systab;
+
+  // Check if we have a system database name in the command.
+  DBUG_ASSERT(known_system_databases != NULL);
+  names= known_system_databases;
+  while (names && *names)
+  {
+    if (strcmp(*names, db) == 0)
+    {
+      /* Used to compare later, will be faster */
+      check_params.db= *names;
+      is_system_database= true;
+      break;
+    }
+    names++;
+  }
+  if (!is_system_database)
+    DBUG_RETURN(true); // It's a user table name.
+
+  // Check if this is SQL layer system tables.
+  systab= mysqld_system_tables;
+  check_params.is_sql_layer_system_table= false;
+  while (systab && systab->db)
+  {
+    if (systab->db == check_params.db &&
+        strcmp(systab->tablename, table_name) == 0)
+    {
+      check_params.is_sql_layer_system_table= true;
+      break;
+    }
+    systab++;
+  }
+
+  // Check if this is a system table and if some engine supports it.
+  check_params.status= check_params.is_sql_layer_system_table ?
+    st_sys_tbl_chk_params::KNOWN_SYSTEM_TABLE :
+    st_sys_tbl_chk_params::NOT_KNOWN_SYSTEM_TABLE;
+  check_params.db_type= hton->db_type;
+  check_params.table_name= table_name;
+  plugin_foreach(NULL, check_engine_system_table_handlerton,
+                 MYSQL_STORAGE_ENGINE_PLUGIN, &check_params);
+
+  // SE does not support this system table.
+  if (check_params.status == st_sys_tbl_chk_params::KNOWN_SYSTEM_TABLE)
+    DBUG_RETURN(false);
+
+  // It's a system table or a valid user table.
+  DBUG_RETURN(true);
+}
+
+/**
+  @brief Called for each SE to check if given db, tablename is a system table.
+
+  @details The primary purpose of introducing this function is to stop system
+  tables to be created or being moved to undesired storage engines.
+
+  @param   unused  unused THD*
+  @param   plugin  Points to specific SE.
+  @param   arg     Is of type struct st_sys_tbl_chk_params.
+
+  @note
+    args->status   Indicates OUT param,
+                   see struct st_sys_tbl_chk_params definition for more info.
+
+  @return Operation status
+    @retval  true  There was a match found.
+                   This will stop doing checks with other SE's.
+
+    @retval  false There was no match found.
+                   Other SE's will be checked to find a match.
+*/
+static my_bool check_engine_system_table_handlerton(THD *unused,
+                                                    plugin_ref plugin,
+                                                    void *arg)
+{
+  st_sys_tbl_chk_params *check_params= (st_sys_tbl_chk_params*) arg;
+  handlerton *hton= plugin_data(plugin, handlerton *);
+
+  // Do we already know that the table is a system table?
+  if (check_params->status == st_sys_tbl_chk_params::KNOWN_SYSTEM_TABLE)
+  {
+    /*
+      If this is the same SE specified in the command, we can
+      simply ask the SE if it supports it stop the search regardless.
+    */
+    if (hton->db_type == check_params->db_type)
+    {
+      if (hton->is_supported_system_table &&
+          hton->is_supported_system_table(check_params->db,
+                                       check_params->table_name,
+                                       check_params->is_sql_layer_system_table))
+        check_params->status= st_sys_tbl_chk_params::SUPPORTED_SYSTEM_TABLE;
+      return TRUE;
+    }
+    /*
+      If this is a different SE, there is no point in asking the SE
+      since we already know it's a system table and we don't care
+      if it is supported or not.
+    */
+    return FALSE;
+  }
+
+  /*
+    We don't yet know if the table is a system table or not.
+    We therefore must always ask the SE.
+  */
+  if (hton->is_supported_system_table &&
+      hton->is_supported_system_table(check_params->db,
+                                      check_params->table_name,
+                                      check_params->is_sql_layer_system_table))
+  {
+    /*
+      If this is the same SE specified in the command, we know it's a
+      supported system table and can stop the search.
+    */
+    if (hton->db_type == check_params->db_type)
+    {
+      check_params->status= st_sys_tbl_chk_params::SUPPORTED_SYSTEM_TABLE;
+      return TRUE;
+    }
+    else
+      check_params->status= st_sys_tbl_chk_params::KNOWN_SYSTEM_TABLE;
+  }
+
+  return FALSE;
+}
+
+/*
+  Prepare list of all known system database names
+  current we just have 'mysql' as system database name.
+
+  Later ndbcluster, innodb SE's can define some new database
+  name which can store system tables specific to SE.
+*/
+const char** ha_known_system_databases(void)
+{
+  I_List<i_string> found_databases;
+  const char **databases, **database;
+
+  // Get mysqld system database name.
+  found_databases.push_back(new i_string(mysqld_system_database));
+
+  // Get system database names from every specific storage engine.
+  plugin_foreach(NULL, system_databases_handlerton,
+                 MYSQL_STORAGE_ENGINE_PLUGIN, &found_databases);
+
+  int element_count= 0;
+  I_List_iterator<i_string> iter(found_databases);
+  while (iter++) element_count++;
+  databases= (const char **) my_once_alloc(sizeof(char *)*
+                                           (element_count+1),
+                                           MYF(MY_WME | MY_FAE));
+  DBUG_ASSERT(databases != NULL);
+
+  database= databases;
+  i_string *tmp;
+  while ((tmp= found_databases.get()))
+  {
+    *database++= tmp->ptr;
+    delete tmp;
+  }
+  *database= NULL; // Last element.
+
+  return databases;
+}
+
+/**
+  @brief Fetch system database name specific to SE.
+
+  @details This function is invoked by plugin_foreach() from
+           ha_known_system_databases(), for each storage engine.
+*/
+static my_bool system_databases_handlerton(THD *unused, plugin_ref plugin,
+                                           void *arg)
+{
+  I_List<i_string> *found_databases= (I_List<i_string> *) arg;
+  const char *db;
+
+  handlerton *hton= plugin_data(plugin, handlerton *);
+  if (hton->system_database)
+  {
+    db= hton->system_database();
+    if (db)
+      found_databases->push_back(new i_string(db));
+  }
+
+  return FALSE;
+}
+
 
 void st_ha_check_opt::init()
 {
diff --git a/sql/handler.h b/sql/handler.h
index cbdfc231ed5..0b970a1349d 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -382,6 +382,25 @@ enum enum_binlog_command {
 /** Unused. Reserved for future versions. */
 #define HA_CREATE_USED_PAGE_CHECKSUM    (1L << 21)
 
+
+/*
+  This is master database for most of system tables. However there
+  can be other databases which can hold system tables. Respective
+  storage engines define their own system database names.
+*/
+extern const char *mysqld_system_database;
+
+/*
+  Structure to hold list of system_database.system_table.
+  This is used at both mysqld and storage engine layer.
+*/
+struct st_system_tablename
+{
+  const char *db;
+  const char *tablename;
+};
+
+
 typedef ulonglong my_xid; // this line is the same as in log_event.h
 #define MYSQL_XID_PREFIX "MySQLXid"
 #define MYSQL_XID_PREFIX_LEN 8 // must be a multiple of 8
@@ -800,6 +819,39 @@ struct handlerton
                      const char *wild, bool dir, List<LEX_STRING> *files);
    int (*table_exists_in_engine)(handlerton *hton, THD* thd, const char *db,
                                  const char *name);
+
+  /**
+    List of all system tables specific to the SE.
+    Array element would look like below,
+     { "<database_name>", "<system table name>" },
+    The last element MUST be,
+     { (const char*)NULL, (const char*)NULL }
+
+    @see ha_example_system_tables in ha_example.cc
+
+    This interface is optional, so every SE need not implement it.
+  */
+  const char* (*system_database)();
+
+  /**
+    Check if the given db.tablename is a system table for this SE.
+
+    @param db                         Database name to check.
+    @param table_name                 table name to check.
+    @param is_sql_layer_system_table  if the supplied db.table_name is a SQL
+                                      layer system table.
+
+    @see example_is_supported_system_table in ha_example.cc
+
+    is_sql_layer_system_table is supplied to make more efficient
+    checks possible for SEs that support all SQL layer tables.
+
+    This interface is optional, so every SE need not implement it.
+  */
+  bool (*is_supported_system_table)(const char *db,
+                                    const char *table_name,
+                                    bool is_sql_layer_system_table);
+
    uint32 license; /* Flag for Engine License */
    void *data; /* Location for engines to keep personal structures */
 };
@@ -2238,6 +2290,8 @@ int ha_discover(THD* thd, const char* dbname, const char* name,
 int ha_find_files(THD *thd,const char *db,const char *path,
                   const char *wild, bool dir, List<LEX_STRING>* files);
 int ha_table_exists_in_engine(THD* thd, const char* db, const char* name);
+bool ha_check_if_supported_system_table(handlerton *hton, const char* db, 
+                                        const char* table_name);
 
 /* key cache */
 extern "C" int ha_init_key_cache(const char *name, KEY_CACHE *key_cache);
diff --git a/sql/share/errmsg-utf8.txt b/sql/share/errmsg-utf8.txt
index 587a70d50bd..4434ecafc29 100644
--- a/sql/share/errmsg-utf8.txt
+++ b/sql/share/errmsg-utf8.txt
@@ -6497,6 +6497,9 @@ ER_BINLOG_UNSAFE_CREATE_SELECT_AUTOINC
 ER_BINLOG_UNSAFE_INSERT_TWO_KEYS
   eng "INSERT... ON DUPLICATE KEY UPDATE  on a table with more than one UNIQUE KEY is unsafe"
 
+ER_UNSUPPORTED_ENGINE
+  eng "Storage engine '%s' does not support system tables. [%s.%s]"
+
 #
 #  End of 5.5 error messages.
 #
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index b9ec70f6560..0f4bac47cab 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -69,7 +69,10 @@ static int copy_data_between_tables(TABLE *from,TABLE *to,
                                     bool error_if_not_empty);
 
 static bool prepare_blob_field(THD *thd, Create_field *sql_field);
-static bool check_engine(THD *, const char *, HA_CREATE_INFO *);
+static bool check_engine(THD *thd, const char *db_name,
+                         const char *table_name,
+                         HA_CREATE_INFO *create_info);
+
 static int
 mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
                            Alter_info *alter_info,
@@ -3940,7 +3943,7 @@ bool mysql_create_table_no_lock(THD *thd,
                MYF(0));
     DBUG_RETURN(TRUE);
   }
-  if (check_engine(thd, table_name, create_info))
+  if (check_engine(thd, db, table_name, create_info))
     DBUG_RETURN(TRUE);
 
   set_table_default_charset(thd, create_info, (char*) db);
@@ -5923,7 +5926,7 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
       create_info->db_type= old_db_type;
   }
 
-  if (check_engine(thd, new_name, create_info))
+  if (check_engine(thd, new_db, new_name, create_info))
     goto err;
   new_db_type= create_info->db_type;
 
@@ -7342,16 +7345,32 @@ err:
   DBUG_RETURN(TRUE);
 }
 
-static bool check_engine(THD *thd, const char *table_name,
-                         HA_CREATE_INFO *create_info)
+/**
+  @brief Check if the table can be created in the specified storage engine.
+
+  Checks if the storage engine is enabled and supports the given table
+  type (e.g. normal, temporary, system). May do engine substitution
+  if the requested engine is disabled.
+
+  @param thd          Thread descriptor.
+  @param db_name      Database name.
+  @param table_name   Name of table to be created.
+  @param create_info  Create info from parser, including engine.
+
+  @retval true  Engine not available/supported, error has been reported.
+  @retval false Engine available/supported.
+*/
+static bool check_engine(THD *thd, const char *db_name,
+                         const char *table_name, HA_CREATE_INFO *create_info)
 {
+  DBUG_ENTER("check_engine");
   handlerton **new_engine= &create_info->db_type;
   handlerton *req_engine= *new_engine;
   bool no_substitution=
         test(thd->variables.sql_mode & MODE_NO_ENGINE_SUBSTITUTION);
   if (!(*new_engine= ha_checktype(thd, ha_legacy_type(req_engine),
                                   no_substitution, 1)))
-    return TRUE;
+    DBUG_RETURN(true);
 
   if (req_engine && req_engine != *new_engine)
   {
@@ -7369,9 +7388,23 @@ static bool check_engine(THD *thd, const char *table_name,
       my_error(ER_ILLEGAL_HA_CREATE_OPTION, MYF(0),
                ha_resolve_storage_engine_name(*new_engine), "TEMPORARY");
       *new_engine= 0;
-      return TRUE;
+      DBUG_RETURN(true);
     }
     *new_engine= myisam_hton;
   }
-  return FALSE;
+
+  /*
+    Check, if the given table name is system table, and if the storage engine 
+    does supports it.
+  */
+  if ((create_info->used_fields & HA_CREATE_USED_ENGINE) &&
+      !ha_check_if_supported_system_table(*new_engine, db_name, table_name))
+  {
+    my_error(ER_UNSUPPORTED_ENGINE, MYF(0),
+             ha_resolve_storage_engine_name(*new_engine), db_name, table_name);
+    *new_engine= NULL;
+    DBUG_RETURN(true);
+  }
+
+  DBUG_RETURN(false);
 }
diff --git a/storage/example/ha_example.cc b/storage/example/ha_example.cc
index 3b0100510b8..c4ed4d5d2c2 100644
--- a/storage/example/ha_example.cc
+++ b/storage/example/ha_example.cc
@@ -103,6 +103,12 @@ static handler *example_create_handler(handlerton *hton,
 
 handlerton *example_hton;
 
+/* Interface to mysqld, to check system tables supported by SE */
+static const char* example_system_database();
+static bool example_is_supported_system_table(const char *db,
+                                      const char *table_name,
+                                      bool is_sql_layer_system_table);
+
 /* Variables for example share methods */
 
 /* 
@@ -165,6 +171,8 @@ static int example_init_func(void *p)
   example_hton->state=   SHOW_OPTION_YES;
   example_hton->create=  example_create_handler;
   example_hton->flags=   HTON_CAN_RECREATE;
+  example_hton->system_database=   example_system_database;
+  example_hton->is_supported_system_table= example_is_supported_system_table;
 
   DBUG_RETURN(0);
 }
@@ -298,6 +306,65 @@ const char **ha_example::bas_ext() const
   return ha_example_exts;
 }
 
+/*
+  Following handler function provides access to
+  system database specific to SE. This interface
+  is optional, so every SE need not implement it.
+*/
+const char* ha_example_system_database= NULL;
+const char* example_system_database()
+{
+  return ha_example_system_database;
+}
+
+/*
+  List of all system tables specific to the SE.
+  Array element would look like below,
+     { "<database_name>", "<system table name>" },
+  The last element MUST be,
+     { (const char*)NULL, (const char*)NULL }
+
+  This array is optional, so every SE need not implement it.
+*/
+static st_system_tablename ha_example_system_tables[]= {
+  {(const char*)NULL, (const char*)NULL}
+};
+
+/**
+  @brief Check if the given db.tablename is a system table for this SE.
+
+  @param db                         Database name to check.
+  @param table_name                 table name to check.
+  @param is_sql_layer_system_table  if the supplied db.table_name is a SQL
+                                    layer system table.
+
+  @return
+    @retval TRUE   Given db.table_name is supported system table.
+    @retval FALSE  Given db.table_name is not a supported system table.
+*/
+static bool example_is_supported_system_table(const char *db,
+                                              const char *table_name,
+                                              bool is_sql_layer_system_table)
+{
+  st_system_tablename *systab;
+
+  // Does this SE support "ALL" SQL layer system tables ?
+  if (is_sql_layer_system_table)
+    return false;
+
+  // Check if this is SE layer system tables
+  systab= ha_example_system_tables;
+  while (systab && systab->db)
+  {
+    if (systab->db == db &&
+        strcmp(systab->tablename, table_name) == 0)
+      return true;
+    systab++;
+  }
+
+  return false;
+}
+
 
 /**
   @brief
diff --git a/storage/myisam/ha_myisam.cc b/storage/myisam/ha_myisam.cc
index 4d8f613b454..34532d39443 100644
--- a/storage/myisam/ha_myisam.cc
+++ b/storage/myisam/ha_myisam.cc
@@ -37,6 +37,11 @@
 ulonglong myisam_recover_options;
 static ulong opt_myisam_block_size;
 
+/* Interface to mysqld, to check system tables supported by SE */
+static bool myisam_is_supported_system_table(const char *db,
+                                             const char *table_name,
+                                             bool is_sql_layer_system_table);
+
 /* bits in myisam_recover_options */
 const char *myisam_recover_names[] =
 { "DEFAULT", "BACKUP", "FORCE", "QUICK", "OFF", NullS};
@@ -665,6 +670,42 @@ const char **ha_myisam::bas_ext() const
   return ha_myisam_exts;
 }
 
+/**
+  @brief Check if the given db.tablename is a system table for this SE.
+
+  @param db                         Database name to check.
+  @param table_name                 table name to check.
+  @param is_sql_layer_system_table  if the supplied db.table_name is a SQL
+                                    layer system table.
+
+  @note Currently, only MYISAM engine supports all the SQL layer
+        system tables, and hence it returns true, when
+        is_sql_layer_system_table is set.
+
+  @note In case there is a need to define MYISAM specific system
+        database, then please see reference implementation in
+        ha_example.cc.
+
+  @return
+    @retval TRUE   Given db.table_name is supported system table.
+    @retval FALSE  Given db.table_name is not a supported system table.
+*/
+static bool myisam_is_supported_system_table(const char *db,
+                                             const char *table_name,
+                                             bool is_sql_layer_system_table)
+{
+  // Does MYISAM support "ALL" SQL layer system tables ?
+  if (is_sql_layer_system_table)
+    return true;
+
+  /*
+    Currently MYISAM does not support any other SE specific
+    system tables. If in future it does, please see ha_example.cc
+    for reference implementation.
+  */
+
+  return false;
+}
 
 const char *ha_myisam::index_type(uint key_number)
 {
@@ -2070,6 +2111,8 @@ static int myisam_init(void *p)
   myisam_hton->create= myisam_create_handler;
   myisam_hton->panic= myisam_panic;
   myisam_hton->flags= HTON_CAN_RECREATE | HTON_SUPPORT_LOG_TABLES;
+  myisam_hton->is_supported_system_table= myisam_is_supported_system_table;
+
   return 0;
 }