Finalize storage engine plugins

Give BerkeleyDB savepoints
Remove "enum db_type" from most of the code


storage/example/ha_example.h:
  Rename: sql/examples/ha_example.h -> storage/example/ha_example.h
storage/csv/ha_tina.h:
  Rename: sql/examples/ha_tina.h -> storage/csv/ha_tina.h
config/ac-macros/storage.m4:
  if hton name is "no", then we don't install it as a builtin
configure.in:
  pluggable changes
include/plugin.h:
  version field
mysql-test/r/bdb.result:
  savepoint results copied from innodb test
mysql-test/r/information_schema.result:
  PLUGINS information schema
mysql-test/r/information_schema_db.result:
  PLUGINS information schema
mysql-test/t/bdb.test:
  savepoint test copied from innodb test
sql/Makefile.am:
  tina and example are not here anymore
sql/authors.h:
  minor tweek
sql/ha_archive.cc:
  remove unwanted handlerton entries
sql/ha_berkeley.cc:
  remove unwanted handlerton entries
  support for savepoints
  changes to show logs
sql/ha_blackhole.cc:
  remove unwanted handlerton entries
sql/ha_federated.cc:
  remove unwanted handlerton entries
sql/ha_heap.cc:
  remove unwanted handlerton entries
sql/ha_innodb.cc:
  remove unwanted handlerton entries
  changes for show status
sql/ha_myisam.cc:
  remove unwanted handlerton entries
sql/ha_myisammrg.cc:
  remove unwanted handlerton entries
sql/ha_ndbcluster.cc:
  remove unwanted handlerton entries
  changes to stat_print
sql/ha_partition.cc:
  remove unwanted handlerton entries
  bye bye enum db_type
sql/ha_partition.h:
  bye bye enum db_type
sql/handler.cc:
  remove unwanted handlerton entries
  bye bye enum db_type
sql/handler.h:
  remove unwanted handlerton entries
  bye bye enum db_type
  changes to stat_print_fn
sql/item_sum.cc:
  bye bye enum db_type
sql/log.cc:
  remove unwanted handlerton entries
sql/mysql_priv.h:
  bye bye enum db_type
sql/mysqld.cc:
  bye bye enum db_type
  reorder plugin initialization
sql/set_var.cc:
  bye bye enum db_type
sql/set_var.h:
  bye bye enum db_type
sql/sql_base.cc:
  bye bye enum db_type
sql/sql_cache.cc:
  bye bye enum db_type
sql/sql_class.h:
  bye bye enum db_type
sql/sql_delete.cc:
  bye bye enum db_type
sql/sql_insert.cc:
  bye bye enum db_type
sql/sql_lex.h:
  show plugin
sql/sql_parse.cc:
  bye bye enum db_type
sql/sql_partition.cc:
  bye bye enum db_type
sql/sql_plugin.cc:
  loadable storage engines
sql/sql_plugin.h:
  loadable storage engines
sql/sql_rename.cc:
  bye bye enum db_type
sql/sql_select.cc:
  bye bye enum db_type
sql/sql_show.cc:
  SHOW PLUGIN
  PLUGINS information schema
  changes to show engines
sql/sql_table.cc:
  bye bye enum db_type
sql/sql_view.cc:
  bye bye enum db_type
sql/sql_view.h:
  bye bye enum db_type
sql/sql_yacc.yy:
  bye bye enum db_type
sql/table.cc:
  bye bye enum db_type
sql/table.h:
  bye bye enum db_type
sql/unireg.cc:
  bye bye enum db_type
storage/csv/ha_tina.cc:
  make tina into a loadable plugin
storage/example/ha_example.cc:
  make into a plugin
storage/csv/Makefile.am:
  New BitKeeper file ``storage/csv/Makefile.am''
storage/example/Makefile.am:
  New BitKeeper file ``storage/example/Makefile.am''

45 files changed, 1237 insertions, 2566 deletions

diff --git a/sql/Makefile.am b/sql/Makefile.am
index 1090c6d48c7..a5b81ffdcdb 100644
--- a/sql/Makefile.am
+++ b/sql/Makefile.am
@@ -101,9 +101,7 @@ EXTRA_mysqld_SOURCES =	ha_innodb.cc ha_berkeley.cc ha_archive.cc \
 			ha_innodb.h  ha_berkeley.h  ha_archive.h \
 			ha_blackhole.cc ha_federated.cc ha_ndbcluster.cc \
 			ha_blackhole.h  ha_federated.h  ha_ndbcluster.h \
-			ha_partition.cc ha_partition.h \
-			examples/ha_tina.cc examples/ha_example.cc \
-			examples/ha_tina.h  examples/ha_example.h
+			ha_partition.cc ha_partition.h
 mysqld_DEPENDENCIES =	@mysql_se_objs@
 gen_lex_hash_SOURCES =	gen_lex_hash.cc
 gen_lex_hash_LDADD =	$(LDADD) $(CXXLDFLAGS)
diff --git a/sql/authors.h b/sql/authors.h
index 1c6e1d468c4..fde1806f4be 100644
--- a/sql/authors.h
+++ b/sql/authors.h
@@ -62,7 +62,8 @@ struct show_table_authors_st show_table_authors[]= {
   { "Albert Chin-A-Young", "",
     "Tru64 port, large file support, better TCP wrappers support" },
   { "Jorge del Conde", "Mexico City, Mexico", "Windows development" },
-  { "Antony T. Curtis", "Norwalk, CA, USA", "Parser, port to OS/2" },
+  { "Antony T. Curtis", "Norwalk, CA, USA",
+    "Parser, port to OS/2, storage engines and some random stuff" },
   { "Yuri Dario", "", "OS/2 port" },
   { "Sergei Golubchik", "Kerpen, Germany",
     "Full-text search, precision math" },
diff --git a/sql/examples/ha_example.cc b/sql/examples/ha_example.cc
deleted file mode 100644
index 5b32b549e8d..00000000000
--- a/sql/examples/ha_example.cc
+++ /dev/null
@@ -1,715 +0,0 @@
-/* Copyright (C) 2003 MySQL AB
-
-  This program is free software; you can redistribute it and/or modify
-  it under the terms of the GNU General Public License as published by
-  the Free Software Foundation; either version 2 of the License, or
-  (at your option) any later version.
-
-  This program is distributed in the hope that it will be useful,
-  but WITHOUT ANY WARRANTY; without even the implied warranty of
-  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-  GNU General Public License for more details.
-
-  You should have received a copy of the GNU General Public License
-  along with this program; if not, write to the Free Software
-  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
-
-/*
-  ha_example is a stubbed storage engine. It does nothing at this point. It
-  will let you create/open/delete tables but that is all. You can enable it
-  in your buld by doing the following during your build process:
-  ./configure --with-example-storage-engine
-
-  Once this is done mysql will let you create tables with:
-  CREATE TABLE A (...) ENGINE=EXAMPLE;
-
-  The example is setup to use table locks. It implements an example "SHARE"
-  that is inserted into a hash by table name. You can use this to store
-  information of state that any example handler object will be able to see
-  if it is using the same table.
-
-  Please read the object definition in ha_example.h before reading the rest
-  if this file.
-
-  To get an idea of what occurs here is an example select that would do a
-  scan of an entire table:
-  ha_example::store_lock
-  ha_example::external_lock
-  ha_example::info
-  ha_example::rnd_init
-  ha_example::extra
-  ENUM HA_EXTRA_CACHE   Cash record in HA_rrnd()
-  ha_example::rnd_next
-  ha_example::rnd_next
-  ha_example::rnd_next
-  ha_example::rnd_next
-  ha_example::rnd_next
-  ha_example::rnd_next
-  ha_example::rnd_next
-  ha_example::rnd_next
-  ha_example::rnd_next
-  ha_example::extra
-  ENUM HA_EXTRA_NO_CACHE   End cacheing of records (def)
-  ha_example::external_lock
-  ha_example::extra
-  ENUM HA_EXTRA_RESET   Reset database to after open
-
-  In the above example has 9 row called before rnd_next signalled that it was
-  at the end of its data. In the above example the table was already opened
-  (or you would have seen a call to ha_example::open(). Calls to
-  ha_example::extra() are hints as to what will be occuring to the request.
-
-  Happy coding!
-    -Brian
-*/
-
-#ifdef USE_PRAGMA_IMPLEMENTATION
-#pragma implementation        // gcc: Class implementation
-#endif
-
-#include "../mysql_priv.h"
-
-#include "ha_example.h"
-
-static handler* example_create_handler(TABLE_SHARE *table);
-
-handlerton example_hton= {
-  "EXAMPLE",
-  SHOW_OPTION_YES,
-  "Example storage engine", 
-  DB_TYPE_EXAMPLE_DB,
-  NULL,    /* We do need to write one! */
-  0,       /* slot */
-  0,       /* savepoint size. */
-  NULL,    /* close_connection */
-  NULL,    /* savepoint */
-  NULL,    /* rollback to savepoint */
-  NULL,    /* release savepoint */
-  NULL,    /* commit */
-  NULL,    /* rollback */
-  NULL,    /* prepare */
-  NULL,    /* recover */
-  NULL,    /* commit_by_xid */
-  NULL,    /* rollback_by_xid */
-  NULL,    /* create_cursor_read_view */
-  NULL,    /* set_cursor_read_view */
-  NULL,    /* close_cursor_read_view */
-  example_create_handler,    /* Create a new handler */
-  NULL,    /* Drop a database */
-  NULL,    /* Panic call */
-  NULL,    /* Release temporary latches */
-  NULL,    /* Update Statistics */
-  NULL,    /* Start Consistent Snapshot */
-  NULL,    /* Flush logs */
-  NULL,    /* Show status */
-  NULL,    /* Replication Report Sent Binlog */
-  HTON_CAN_RECREATE
-};
-
-/* Variables for example share methods */
-static HASH example_open_tables; // Hash used to track open tables
-pthread_mutex_t example_mutex;   // This is the mutex we use to init the hash
-static int example_init= 0;      // Variable for checking the init state of hash
-
-
-/*
-  Function we use in the creation of our hash to get key.
-*/
-static byte* example_get_key(EXAMPLE_SHARE *share,uint *length,
-                             my_bool not_used __attribute__((unused)))
-{
-  *length=share->table_name_length;
-  return (byte*) share->table_name;
-}
-
-
-/*
-  Example of simple lock controls. The "share" it creates is structure we will
-  pass to each example handler. Do you have to have one of these? Well, you have
-  pieces that are used for locking, and they are needed to function.
-*/
-static EXAMPLE_SHARE *get_share(const char *table_name, TABLE *table)
-{
-  EXAMPLE_SHARE *share;
-  uint length;
-  char *tmp_name;
-
-  /*
-    So why does this exist? There is no way currently to init a storage engine.
-    Innodb and BDB both have modifications to the server to allow them to
-    do this. Since you will not want to do this, this is probably the next
-    best method.
-  */
-  if (!example_init)
-  {
-    /* Hijack a mutex for init'ing the storage engine */
-    pthread_mutex_lock(&LOCK_mysql_create_db);
-    if (!example_init)
-    {
-      example_init++;
-      VOID(pthread_mutex_init(&example_mutex,MY_MUTEX_INIT_FAST));
-      (void) hash_init(&example_open_tables,system_charset_info,32,0,0,
-                       (hash_get_key) example_get_key,0,0);
-    }
-    pthread_mutex_unlock(&LOCK_mysql_create_db);
-  }
-  pthread_mutex_lock(&example_mutex);
-  length=(uint) strlen(table_name);
-
-  if (!(share=(EXAMPLE_SHARE*) hash_search(&example_open_tables,
-                                           (byte*) table_name,
-                                           length)))
-  {
-    if (!(share=(EXAMPLE_SHARE *)
-          my_multi_malloc(MYF(MY_WME | MY_ZEROFILL),
-                          &share, sizeof(*share),
-                          &tmp_name, length+1,
-                          NullS)))
-    {
-      pthread_mutex_unlock(&example_mutex);
-      return NULL;
-    }
-
-    share->use_count=0;
-    share->table_name_length=length;
-    share->table_name=tmp_name;
-    strmov(share->table_name,table_name);
-    if (my_hash_insert(&example_open_tables, (byte*) share))
-      goto error;
-    thr_lock_init(&share->lock);
-    pthread_mutex_init(&share->mutex,MY_MUTEX_INIT_FAST);
-  }
-  share->use_count++;
-  pthread_mutex_unlock(&example_mutex);
-
-  return share;
-
-error:
-  pthread_mutex_destroy(&share->mutex);
-  pthread_mutex_unlock(&example_mutex);
-  my_free((gptr) share, MYF(0));
-
-  return NULL;
-}
-
-
-/*
-  Free lock controls. We call this whenever we close a table. If the table had
-  the last reference to the share then we free memory associated with it.
-*/
-static int free_share(EXAMPLE_SHARE *share)
-{
-  pthread_mutex_lock(&example_mutex);
-  if (!--share->use_count)
-  {
-    hash_delete(&example_open_tables, (byte*) share);
-    thr_lock_delete(&share->lock);
-    pthread_mutex_destroy(&share->mutex);
-    my_free((gptr) share, MYF(0));
-  }
-  pthread_mutex_unlock(&example_mutex);
-
-  return 0;
-}
-
-
-static handler* example_create_handler(TABLE_SHARE *table)
-{
-  return new ha_example(table);
-}
-
-
-ha_example::ha_example(TABLE_SHARE *table_arg)
-  :handler(&example_hton, table_arg)
-{}
-
-/*
-  If frm_error() is called then we will use this to to find out what file extentions
-  exist for the storage engine. This is also used by the default rename_table and
-  delete_table method in handler.cc.
-*/
-static const char *ha_example_exts[] = {
-  NullS
-};
-
-const char **ha_example::bas_ext() const
-{
-  return ha_example_exts;
-}
-
-
-/*
-  Used for opening tables. The name will be the name of the file.
-  A table is opened when it needs to be opened. For instance
-  when a request comes in for a select on the table (tables are not
-  open and closed for each request, they are cached).
-
-  Called from handler.cc by handler::ha_open(). The server opens all tables by
-  calling ha_open() which then calls the handler specific open().
-*/
-int ha_example::open(const char *name, int mode, uint test_if_locked)
-{
-  DBUG_ENTER("ha_example::open");
-
-  if (!(share = get_share(name, table)))
-    DBUG_RETURN(1);
-  thr_lock_data_init(&share->lock,&lock,NULL);
-
-  DBUG_RETURN(0);
-}
-
-
-/*
-  Closes a table. We call the free_share() function to free any resources
-  that we have allocated in the "shared" structure.
-
-  Called from sql_base.cc, sql_select.cc, and table.cc.
-  In sql_select.cc it is only used to close up temporary tables or during
-  the process where a temporary table is converted over to being a
-  myisam table.
-  For sql_base.cc look at close_data_tables().
-*/
-int ha_example::close(void)
-{
-  DBUG_ENTER("ha_example::close");
-  DBUG_RETURN(free_share(share));
-}
-
-
-/*
-  write_row() inserts a row. No extra() hint is given currently if a bulk load
-  is happeneding. buf() is a byte array of data. You can use the field
-  information to extract the data from the native byte array type.
-  Example of this would be:
-  for (Field **field=table->field ; *field ; field++)
-  {
-    ...
-  }
-
-  See ha_tina.cc for an example of extracting all of the data as strings.
-  ha_berekly.cc has an example of how to store it intact by "packing" it
-  for ha_berkeley's own native storage type.
-
-  See the note for update_row() on auto_increments and timestamps. This
-  case also applied to write_row().
-
-  Called from item_sum.cc, item_sum.cc, sql_acl.cc, sql_insert.cc,
-  sql_insert.cc, sql_select.cc, sql_table.cc, sql_udf.cc, and sql_update.cc.
-*/
-int ha_example::write_row(byte * buf)
-{
-  DBUG_ENTER("ha_example::write_row");
-  DBUG_RETURN(HA_ERR_WRONG_COMMAND);
-}
-
-
-/*
-  Yes, update_row() does what you expect, it updates a row. old_data will have
-  the previous row record in it, while new_data will have the newest data in
-  it.
-  Keep in mind that the server can do updates based on ordering if an ORDER BY
-  clause was used. Consecutive ordering is not guarenteed.
-  Currently new_data will not have an updated auto_increament record, or
-  and updated timestamp field. You can do these for example by doing these:
-  if (table->timestamp_field_type & TIMESTAMP_AUTO_SET_ON_UPDATE)
-    table->timestamp_field->set_time();
-  if (table->next_number_field && record == table->record[0])
-    update_auto_increment();
-
-  Called from sql_select.cc, sql_acl.cc, sql_update.cc, and sql_insert.cc.
-*/
-int ha_example::update_row(const byte * old_data, byte * new_data)
-{
-
-  DBUG_ENTER("ha_example::update_row");
-  DBUG_RETURN(HA_ERR_WRONG_COMMAND);
-}
-
-
-/*
-  This will delete a row. buf will contain a copy of the row to be deleted.
-  The server will call this right after the current row has been called (from
-  either a previous rnd_nexT() or index call).
-  If you keep a pointer to the last row or can access a primary key it will
-  make doing the deletion quite a bit easier.
-  Keep in mind that the server does no guarentee consecutive deletions. ORDER BY
-  clauses can be used.
-
-  Called in sql_acl.cc and sql_udf.cc to manage internal table information.
-  Called in sql_delete.cc, sql_insert.cc, and sql_select.cc. In sql_select it is
-  used for removing duplicates while in insert it is used for REPLACE calls.
-*/
-int ha_example::delete_row(const byte * buf)
-{
-  DBUG_ENTER("ha_example::delete_row");
-  DBUG_RETURN(HA_ERR_WRONG_COMMAND);
-}
-
-
-/*
-  Positions an index cursor to the index specified in the handle. Fetches the
-  row if available. If the key value is null, begin at the first key of the
-  index.
-*/
-int ha_example::index_read(byte * buf, const byte * key,
-                           uint key_len __attribute__((unused)),
-                           enum ha_rkey_function find_flag
-                           __attribute__((unused)))
-{
-  DBUG_ENTER("ha_example::index_read");
-  DBUG_RETURN(HA_ERR_WRONG_COMMAND);
-}
-
-
-/*
-  Positions an index cursor to the index specified in key. Fetches the
-  row if any.  This is only used to read whole keys.
-*/
-int ha_example::index_read_idx(byte * buf, uint index, const byte * key,
-                               uint key_len __attribute__((unused)),
-                               enum ha_rkey_function find_flag
-                               __attribute__((unused)))
-{
-  DBUG_ENTER("ha_example::index_read_idx");
-  DBUG_RETURN(HA_ERR_WRONG_COMMAND);
-}
-
-
-/*
-  Used to read forward through the index.
-*/
-int ha_example::index_next(byte * buf)
-{
-  DBUG_ENTER("ha_example::index_next");
-  DBUG_RETURN(HA_ERR_WRONG_COMMAND);
-}
-
-
-/*
-  Used to read backwards through the index.
-*/
-int ha_example::index_prev(byte * buf)
-{
-  DBUG_ENTER("ha_example::index_prev");
-  DBUG_RETURN(HA_ERR_WRONG_COMMAND);
-}
-
-
-/*
-  index_first() asks for the first key in the index.
-
-  Called from opt_range.cc, opt_sum.cc, sql_handler.cc,
-  and sql_select.cc.
-*/
-int ha_example::index_first(byte * buf)
-{
-  DBUG_ENTER("ha_example::index_first");
-  DBUG_RETURN(HA_ERR_WRONG_COMMAND);
-}
-
-
-/*
-  index_last() asks for the last key in the index.
-
-  Called from opt_range.cc, opt_sum.cc, sql_handler.cc,
-  and sql_select.cc.
-*/
-int ha_example::index_last(byte * buf)
-{
-  DBUG_ENTER("ha_example::index_last");
-  DBUG_RETURN(HA_ERR_WRONG_COMMAND);
-}
-
-
-/*
-  rnd_init() is called when the system wants the storage engine to do a table
-  scan.
-  See the example in the introduction at the top of this file to see when
-  rnd_init() is called.
-
-  Called from filesort.cc, records.cc, sql_handler.cc, sql_select.cc, sql_table.cc,
-  and sql_update.cc.
-*/
-int ha_example::rnd_init(bool scan)
-{
-  DBUG_ENTER("ha_example::rnd_init");
-  DBUG_RETURN(HA_ERR_WRONG_COMMAND);
-}
-
-int ha_example::rnd_end()
-{
-  DBUG_ENTER("ha_example::rnd_end");
-  DBUG_RETURN(0);
-}
-
-/*
-  This is called for each row of the table scan. When you run out of records
-  you should return HA_ERR_END_OF_FILE. Fill buff up with the row information.
-  The Field structure for the table is the key to getting data into buf
-  in a manner that will allow the server to understand it.
-
-  Called from filesort.cc, records.cc, sql_handler.cc, sql_select.cc, sql_table.cc,
-  and sql_update.cc.
-*/
-int ha_example::rnd_next(byte *buf)
-{
-  DBUG_ENTER("ha_example::rnd_next");
-  DBUG_RETURN(HA_ERR_END_OF_FILE);
-}
-
-
-/*
-  position() is called after each call to rnd_next() if the data needs
-  to be ordered. You can do something like the following to store
-  the position:
-  my_store_ptr(ref, ref_length, current_position);
-
-  The server uses ref to store data. ref_length in the above case is
-  the size needed to store current_position. ref is just a byte array
-  that the server will maintain. If you are using offsets to mark rows, then
-  current_position should be the offset. If it is a primary key like in
-  BDB, then it needs to be a primary key.
-
-  Called from filesort.cc, sql_select.cc, sql_delete.cc and sql_update.cc.
-*/
-void ha_example::position(const byte *record)
-{
-  DBUG_ENTER("ha_example::position");
-  DBUG_VOID_RETURN;
-}
-
-
-/*
-  This is like rnd_next, but you are given a position to use
-  to determine the row. The position will be of the type that you stored in
-  ref. You can use ha_get_ptr(pos,ref_length) to retrieve whatever key
-  or position you saved when position() was called.
-  Called from filesort.cc records.cc sql_insert.cc sql_select.cc sql_update.cc.
-*/
-int ha_example::rnd_pos(byte * buf, byte *pos)
-{
-  DBUG_ENTER("ha_example::rnd_pos");
-  DBUG_RETURN(HA_ERR_WRONG_COMMAND);
-}
-
-
-/*
-  ::info() is used to return information to the optimizer.
-  see my_base.h for the complete description
-
-  Currently this table handler doesn't implement most of the fields
-  really needed. SHOW also makes use of this data
-  Another note, you will probably want to have the following in your
-  code:
-  if (records < 2)
-    records = 2;
-  The reason is that the server will optimize for cases of only a single
-  record. If in a table scan you don't know the number of records
-  it will probably be better to set records to two so you can return
-  as many records as you need.
-  Along with records a few more variables you may wish to set are:
-    records
-    deleted
-    data_file_length
-    index_file_length
-    delete_length
-    check_time
-  Take a look at the public variables in handler.h for more information.
-
-  Called in:
-    filesort.cc
-    ha_heap.cc
-    item_sum.cc
-    opt_sum.cc
-    sql_delete.cc
-    sql_delete.cc
-    sql_derived.cc
-    sql_select.cc
-    sql_select.cc
-    sql_select.cc
-    sql_select.cc
-    sql_select.cc
-    sql_show.cc
-    sql_show.cc
-    sql_show.cc
-    sql_show.cc
-    sql_table.cc
-    sql_union.cc
-    sql_update.cc
-
-*/
-void ha_example::info(uint flag)
-{
-  DBUG_ENTER("ha_example::info");
-  DBUG_VOID_RETURN;
-}
-
-
-/*
-  extra() is called whenever the server wishes to send a hint to
-  the storage engine. The myisam engine implements the most hints.
-  ha_innodb.cc has the most exhaustive list of these hints.
-*/
-int ha_example::extra(enum ha_extra_function operation)
-{
-  DBUG_ENTER("ha_example::extra");
-  DBUG_RETURN(0);
-}
-
-
-/*
-  Deprecated and likely to be removed in the future. Storage engines normally
-  just make a call like:
-  ha_example::extra(HA_EXTRA_RESET);
-  to handle it.
-*/
-int ha_example::reset(void)
-{
-  DBUG_ENTER("ha_example::reset");
-  DBUG_RETURN(0);
-}
-
-
-/*
-  Used to delete all rows in a table. Both for cases of truncate and
-  for cases where the optimizer realizes that all rows will be
-  removed as a result of a SQL statement.
-
-  Called from item_sum.cc by Item_func_group_concat::clear(),
-  Item_sum_count_distinct::clear(), and Item_func_group_concat::clear().
-  Called from sql_delete.cc by mysql_delete().
-  Called from sql_select.cc by JOIN::reinit().
-  Called from sql_union.cc by st_select_lex_unit::exec().
-*/
-int ha_example::delete_all_rows()
-{
-  DBUG_ENTER("ha_example::delete_all_rows");
-  DBUG_RETURN(HA_ERR_WRONG_COMMAND);
-}
-
-
-/*
-  First you should go read the section "locking functions for mysql" in
-  lock.cc to understand this.
-  This create a lock on the table. If you are implementing a storage engine
-  that can handle transacations look at ha_berkely.cc to see how you will
-  want to goo about doing this. Otherwise you should consider calling flock()
-  here.
-
-  Called from lock.cc by lock_external() and unlock_external(). Also called
-  from sql_table.cc by copy_data_between_tables().
-*/
-int ha_example::external_lock(THD *thd, int lock_type)
-{
-  DBUG_ENTER("ha_example::external_lock");
-  DBUG_RETURN(0);
-}
-
-
-/*
-  The idea with handler::store_lock() is the following:
-
-  The statement decided which locks we should need for the table
-  for updates/deletes/inserts we get WRITE locks, for SELECT... we get
-  read locks.
-
-  Before adding the lock into the table lock handler (see thr_lock.c)
-  mysqld calls store lock with the requested locks.  Store lock can now
-  modify a write lock to a read lock (or some other lock), ignore the
-  lock (if we don't want to use MySQL table locks at all) or add locks
-  for many tables (like we do when we are using a MERGE handler).
-
-  Berkeley DB for example  changes all WRITE locks to TL_WRITE_ALLOW_WRITE
-  (which signals that we are doing WRITES, but we are still allowing other
-  reader's and writer's.
-
-  When releasing locks, store_lock() are also called. In this case one
-  usually doesn't have to do anything.
-
-  In some exceptional cases MySQL may send a request for a TL_IGNORE;
-  This means that we are requesting the same lock as last time and this
-  should also be ignored. (This may happen when someone does a flush
-  table when we have opened a part of the tables, in which case mysqld
-  closes and reopens the tables and tries to get the same locks at last
-  time).  In the future we will probably try to remove this.
-
-  Called from lock.cc by get_lock_data().
-*/
-THR_LOCK_DATA **ha_example::store_lock(THD *thd,
-                                       THR_LOCK_DATA **to,
-                                       enum thr_lock_type lock_type)
-{
-  if (lock_type != TL_IGNORE && lock.type == TL_UNLOCK)
-    lock.type=lock_type;
-  *to++= &lock;
-  return to;
-}
-
-/*
-  Used to delete a table. By the time delete_table() has been called all
-  opened references to this table will have been closed (and your globally
-  shared references released. The variable name will just be the name of
-  the table. You will need to remove any files you have created at this point.
-
-  If you do not implement this, the default delete_table() is called from
-  handler.cc and it will delete all files with the file extentions returned
-  by bas_ext().
-
-  Called from handler.cc by delete_table and  ha_create_table(). Only used
-  during create if the table_flag HA_DROP_BEFORE_CREATE was specified for
-  the storage engine.
-*/
-int ha_example::delete_table(const char *name)
-{
-  DBUG_ENTER("ha_example::delete_table");
-  /* This is not implemented but we want someone to be able that it works. */
-  DBUG_RETURN(0);
-}
-
-/*
-  Renames a table from one name to another from alter table call.
-
-  If you do not implement this, the default rename_table() is called from
-  handler.cc and it will delete all files with the file extentions returned
-  by bas_ext().
-
-  Called from sql_table.cc by mysql_rename_table().
-*/
-int ha_example::rename_table(const char * from, const char * to)
-{
-  DBUG_ENTER("ha_example::rename_table ");
-  DBUG_RETURN(HA_ERR_WRONG_COMMAND);
-}
-
-/*
-  Given a starting key, and an ending key estimate the number of rows that
-  will exist between the two. end_key may be empty which in case determine
-  if start_key matches any rows.
-
-  Called from opt_range.cc by check_quick_keys().
-*/
-ha_rows ha_example::records_in_range(uint inx, key_range *min_key,
-                                     key_range *max_key)
-{
-  DBUG_ENTER("ha_example::records_in_range");
-  DBUG_RETURN(10);                         // low number to force index usage
-}
-
-
-/*
-  create() is called to create a database. The variable name will have the name
-  of the table. When create() is called you do not need to worry about opening
-  the table. Also, the FRM file will have already been created so adjusting
-  create_info will not do you any good. You can overwrite the frm file at this
-  point if you wish to change the table definition, but there are no methods
-  currently provided for doing that.
-
-  Called from handle.cc by ha_create_table().
-*/
-int ha_example::create(const char *name, TABLE *table_arg,
-                       HA_CREATE_INFO *create_info)
-{
-  DBUG_ENTER("ha_example::create");
-  /* This is not implemented but we want someone to be able that it works. */
-  DBUG_RETURN(0);
-}
diff --git a/sql/examples/ha_example.h b/sql/examples/ha_example.h
deleted file mode 100644
index 139a50a3281..00000000000
--- a/sql/examples/ha_example.h
+++ /dev/null
@@ -1,155 +0,0 @@
-/* Copyright (C) 2003 MySQL AB
-
-  This program is free software; you can redistribute it and/or modify
-  it under the terms of the GNU General Public License as published by
-  the Free Software Foundation; either version 2 of the License, or
-  (at your option) any later version.
-
-  This program is distributed in the hope that it will be useful,
-  but WITHOUT ANY WARRANTY; without even the implied warranty of
-  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-  GNU General Public License for more details.
-
-  You should have received a copy of the GNU General Public License
-  along with this program; if not, write to the Free Software
-  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
-
-/*
-  Please read ha_exmple.cc before reading this file.
-  Please keep in mind that the example storage engine implements all methods
-  that are required to be implemented. handler.h has a full list of methods
-  that you can implement.
-*/
-
-#ifdef USE_PRAGMA_INTERFACE
-#pragma interface			/* gcc class implementation */
-#endif
-
-/*
-  EXAMPLE_SHARE is a structure that will be shared amoung all open handlers
-  The example implements the minimum of what you will probably need.
-*/
-typedef struct st_example_share {
-  char *table_name;
-  uint table_name_length,use_count;
-  pthread_mutex_t mutex;
-  THR_LOCK lock;
-} EXAMPLE_SHARE;
-
-/*
-  Class definition for the storage engine
-*/
-class ha_example: public handler
-{
-  THR_LOCK_DATA lock;      /* MySQL lock */
-  EXAMPLE_SHARE *share;    /* Shared lock info */
-
-public:
-  ha_example(TABLE_SHARE *table_arg);
-  ~ha_example()
-  {
-  }
-  /* The name that will be used for display purposes */
-  const char *table_type() const { return "EXAMPLE"; }
-  /*
-    The name of the index type that will be used for display
-    don't implement this method unless you really have indexes
-   */
-  const char *index_type(uint inx) { return "HASH"; }
-  const char **bas_ext() const;
-  /*
-    This is a list of flags that says what the storage engine
-    implements. The current table flags are documented in
-    handler.h
-  */
-  ulong table_flags() const
-  {
-    return 0;
-  }
-  /*
-    This is a bitmap of flags that says how the storage engine
-    implements indexes. The current index flags are documented in
-    handler.h. If you do not implement indexes, just return zero
-    here.
-
-    part is the key part to check. First key part is 0
-    If all_parts it's set, MySQL want to know the flags for the combined
-    index up to and including 'part'.
-  */
-  ulong index_flags(uint inx, uint part, bool all_parts) const
-  {
-    return 0;
-  }
-  /*
-    unireg.cc will call the following to make sure that the storage engine can
-    handle the data it is about to send.
-
-    Return *real* limits of your storage engine here. MySQL will do
-    min(your_limits, MySQL_limits) automatically
-
-    There is no need to implement ..._key_... methods if you don't suport
-    indexes.
-  */
-  uint max_supported_record_length() const { return HA_MAX_REC_LENGTH; }
-  uint max_supported_keys()          const { return 0; }
-  uint max_supported_key_parts()     const { return 0; }
-  uint max_supported_key_length()    const { return 0; }
-  /*
-    Called in test_quick_select to determine if indexes should be used.
-  */
-  virtual double scan_time() { return (double) (records+deleted) / 20.0+10; }
-  /*
-    The next method will never be called if you do not implement indexes.
-  */
-  virtual double read_time(ha_rows rows) { return (double) rows /  20.0+1; }
-
-  /*
-    Everything below are methods that we implment in ha_example.cc.
-
-    Most of these methods are not obligatory, skip them and
-    MySQL will treat them as not implemented
-  */
-  int open(const char *name, int mode, uint test_if_locked);    // required
-  int close(void);                                              // required
-
-  int write_row(byte * buf);
-  int update_row(const byte * old_data, byte * new_data);
-  int delete_row(const byte * buf);
-  int index_read(byte * buf, const byte * key,
-                 uint key_len, enum ha_rkey_function find_flag);
-  int index_read_idx(byte * buf, uint idx, const byte * key,
-                     uint key_len, enum ha_rkey_function find_flag);
-  int index_next(byte * buf);
-  int index_prev(byte * buf);
-  int index_first(byte * buf);
-  int index_last(byte * buf);
-  /*
-    unlike index_init(), rnd_init() can be called two times
-    without rnd_end() in between (it only makes sense if scan=1).
-    then the second call should prepare for the new table scan
-    (e.g if rnd_init allocates the cursor, second call should
-    position it to the start of the table, no need to deallocate
-    and allocate it again
-  */
-  int rnd_init(bool scan);                                      //required
-  int rnd_end();
-  int rnd_next(byte *buf);                                      //required
-  int rnd_pos(byte * buf, byte *pos);                           //required
-  void position(const byte *record);                            //required
-  void info(uint);                                              //required
-
-  int extra(enum ha_extra_function operation);
-  int reset(void);
-  int external_lock(THD *thd, int lock_type);                   //required
-  int delete_all_rows(void);
-  ha_rows records_in_range(uint inx, key_range *min_key,
-                           key_range *max_key);
-  int delete_table(const char *from);
-  int rename_table(const char * from, const char * to);
-  int create(const char *name, TABLE *form,
-             HA_CREATE_INFO *create_info);                      //required
-
-  THR_LOCK_DATA **store_lock(THD *thd, THR_LOCK_DATA **to,
-                             enum thr_lock_type lock_type);     //required
-};
-
diff --git a/sql/examples/ha_tina.cc b/sql/examples/ha_tina.cc
deleted file mode 100644
index 4f2dc0b5cac..00000000000
--- a/sql/examples/ha_tina.cc
+++ /dev/null
@@ -1,935 +0,0 @@
-/* Copyright (C) 2003 MySQL AB
-
-  This program is free software; you can redistribute it and/or modify
-  it under the terms of the GNU General Public License as published by
-  the Free Software Foundation; either version 2 of the License, or
-  (at your option) any later version.
-
-  This program is distributed in the hope that it will be useful,
-  but WITHOUT ANY WARRANTY; without even the implied warranty of
-  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-  GNU General Public License for more details.
-
-  You should have received a copy of the GNU General Public License
-  along with this program; if not, write to the Free Software
-  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
-
-/*
-  Make sure to look at ha_tina.h for more details.
-
-  First off, this is a play thing for me, there are a number of things
-  wrong with it:
-    *) It was designed for csv and therefore its performance is highly
-       questionable.
-    *) Indexes have not been implemented. This is because the files can
-       be traded in and out of the table directory without having to worry
-       about rebuilding anything.
-    *) NULLs and "" are treated equally (like a spreadsheet).
-    *) There was in the beginning no point to anyone seeing this other
-       then me, so there is a good chance that I haven't quite documented
-       it well.
-    *) Less design, more "make it work"
-
-  Now there are a few cool things with it:
-    *) Errors can result in corrupted data files.
-    *) Data files can be read by spreadsheets directly.
-
-TODO:
- *) Move to a block system for larger files
- *) Error recovery, its all there, just need to finish it
- *) Document how the chains work.
-
- -Brian
-*/
-
-#ifdef USE_PRAGMA_IMPLEMENTATION
-#pragma implementation        // gcc: Class implementation
-#endif
-
-#include "mysql_priv.h"
-
-#include "ha_tina.h"
-#include <sys/mman.h>
-
-/* Stuff for shares */
-pthread_mutex_t tina_mutex;
-static HASH tina_open_tables;
-static int tina_init= 0;
-static handler *tina_create_handler(TABLE_SHARE *table);
-
-handlerton tina_hton= {
-  "CSV",
-  SHOW_OPTION_YES,
-  "CSV storage engine", 
-  DB_TYPE_CSV_DB,
-  NULL,    /* One needs to be written! */
-  0,       /* slot */
-  0,       /* savepoint size. */
-  NULL,    /* close_connection */
-  NULL,    /* savepoint */
-  NULL,    /* rollback to savepoint */
-  NULL,    /* release savepoint */
-  NULL,    /* commit */
-  NULL,    /* rollback */
-  NULL,    /* prepare */
-  NULL,    /* recover */
-  NULL,    /* commit_by_xid */
-  NULL,    /* rollback_by_xid */
-  NULL,    /* create_cursor_read_view */
-  NULL,    /* set_cursor_read_view */
-  NULL,    /* close_cursor_read_view */
-  tina_create_handler,    /* Create a new handler */
-  NULL,    /* Drop a database */
-  tina_end,    /* Panic call */
-  NULL,    /* Release temporary latches */
-  NULL,    /* Update Statistics */
-  NULL,    /* Start Consistent Snapshot */
-  NULL,    /* Flush logs */
-  NULL,    /* Show status */
-  NULL,    /* Replication Report Sent Binlog */
-  HTON_CAN_RECREATE
-};
-
-/*****************************************************************************
- ** TINA tables
- *****************************************************************************/
-
-/*
-  Used for sorting chains with qsort().
-*/
-int sort_set (tina_set *a, tina_set *b)
-{
-  /*
-    We assume that intervals do not intersect. So, it is enought to compare
-    any two points. Here we take start of intervals for comparison.
-  */
-  return ( a->begin > b->begin ? -1 : ( a->begin < b->begin ? 1 : 0 ) );
-}
-
-static byte* tina_get_key(TINA_SHARE *share,uint *length,
-                          my_bool not_used __attribute__((unused)))
-{
-  *length=share->table_name_length;
-  return (byte*) share->table_name;
-}
-
-/*
-  Reloads the mmap file.
-*/
-int get_mmap(TINA_SHARE *share, int write)
-{
-  DBUG_ENTER("ha_tina::get_mmap");
-  if (share->mapped_file && my_munmap(share->mapped_file,
-                                      share->file_stat.st_size))
-    DBUG_RETURN(1);
-
-  if (my_fstat(share->data_file, &share->file_stat, MYF(MY_WME)) == -1)
-    DBUG_RETURN(1);
-
-  if (share->file_stat.st_size)
-  {
-    if (write)
-      share->mapped_file= (byte *)my_mmap(NULL, share->file_stat.st_size,
-                                          PROT_READ|PROT_WRITE, MAP_SHARED,
-                                          share->data_file, 0);
-    else
-      share->mapped_file= (byte *)my_mmap(NULL, share->file_stat.st_size,
-                                          PROT_READ, MAP_PRIVATE,
-                                          share->data_file, 0);
-    if ((share->mapped_file ==(caddr_t)-1))
-    {
-      /*
-        Bad idea you think? See the problem is that nothing actually checks
-        the return value of ::rnd_init(), so tossing an error is about
-        it for us.
-        Never going to happen right? :)
-      */
-      my_message(errno, "Woops, blew up opening a mapped file", 0);
-      DBUG_ASSERT(0);
-      DBUG_RETURN(1);
-    }
-  }
-  else
-    share->mapped_file= NULL;
-
-  DBUG_RETURN(0);
-}
-
-/*
-  Simple lock controls.
-*/
-static TINA_SHARE *get_share(const char *table_name, TABLE *table)
-{
-  TINA_SHARE *share;
-  char *tmp_name;
-  uint length;
-
-  if (!tina_init)
-  {
-    /* Hijack a mutex for init'ing the storage engine */
-    pthread_mutex_lock(&LOCK_mysql_create_db);
-    if (!tina_init)
-    {
-      tina_init++;
-      VOID(pthread_mutex_init(&tina_mutex,MY_MUTEX_INIT_FAST));
-      (void) hash_init(&tina_open_tables,system_charset_info,32,0,0,
-                       (hash_get_key) tina_get_key,0,0);
-    }
-    pthread_mutex_unlock(&LOCK_mysql_create_db);
-  }
-  pthread_mutex_lock(&tina_mutex);
-  length=(uint) strlen(table_name);
-  if (!(share=(TINA_SHARE*) hash_search(&tina_open_tables,
-                                        (byte*) table_name,
-                                        length)))
-  {
-    char data_file_name[FN_REFLEN];
-    if (!my_multi_malloc(MYF(MY_WME | MY_ZEROFILL),
-                         &share, sizeof(*share),
-                         &tmp_name, length+1,
-                         NullS))
-    {
-      pthread_mutex_unlock(&tina_mutex);
-      return NULL;
-    }
-
-    share->use_count= 0;
-    share->table_name_length= length;
-    share->table_name= tmp_name;
-    strmov(share->table_name, table_name);
-    fn_format(data_file_name, table_name, "", ".CSV",
-              MY_REPLACE_EXT|MY_UNPACK_FILENAME);
-    if (my_hash_insert(&tina_open_tables, (byte*) share))
-      goto error;
-    thr_lock_init(&share->lock);
-    pthread_mutex_init(&share->mutex,MY_MUTEX_INIT_FAST);
-
-    if ((share->data_file= my_open(data_file_name, O_RDWR|O_APPEND,
-                                   MYF(0))) == -1)
-      goto error2;
-
-    /*
-      We only use share->data_file for writing, so we scan to
-      the end to append
-    */
-    if (my_seek(share->data_file, 0, SEEK_END, MYF(0)) == MY_FILEPOS_ERROR)
-      goto error2;
-
-    share->mapped_file= NULL; // We don't know the state as we just allocated it
-    if (get_mmap(share, 0) > 0)
-      goto error3;
-  }
-  share->use_count++;
-  pthread_mutex_unlock(&tina_mutex);
-
-  return share;
-
-error3:
-  my_close(share->data_file,MYF(0));
-error2:
-  thr_lock_delete(&share->lock);
-  pthread_mutex_destroy(&share->mutex);
-error:
-  pthread_mutex_unlock(&tina_mutex);
-  my_free((gptr) share, MYF(0));
-
-  return NULL;
-}
-
-
-/*
-  Free lock controls.
-*/
-static int free_share(TINA_SHARE *share)
-{
-  DBUG_ENTER("ha_tina::free_share");
-  pthread_mutex_lock(&tina_mutex);
-  int result_code= 0;
-  if (!--share->use_count){
-    /* Drop the mapped file */
-    if (share->mapped_file)
-      my_munmap(share->mapped_file, share->file_stat.st_size);
-    result_code= my_close(share->data_file,MYF(0));
-    hash_delete(&tina_open_tables, (byte*) share);
-    thr_lock_delete(&share->lock);
-    pthread_mutex_destroy(&share->mutex);
-    my_free((gptr) share, MYF(0));
-  }
-  pthread_mutex_unlock(&tina_mutex);
-
-  DBUG_RETURN(result_code);
-}
-
-int tina_end(ha_panic_function type)
-{
-  if (tina_init)
-  {
-    hash_free(&tina_open_tables);
-    VOID(pthread_mutex_destroy(&tina_mutex));
-  }
-  tina_init= 0;
-  return 0;
-}
-
-/*
-  Finds the end of a line.
-  Currently only supports files written on a UNIX OS.
-*/
-byte * find_eoln(byte *data, off_t begin, off_t end)
-{
-  for (off_t x= begin; x < end; x++)
-    if (data[x] == '\n')
-      return data + x;
-
-  return 0;
-}
-
-
-static handler *tina_create_handler(TABLE_SHARE *table)
-{
-  return new ha_tina(table);
-}
-
-
-ha_tina::ha_tina(TABLE_SHARE *table_arg)
-  :handler(&tina_hton, table_arg),
-  /*
-    These definitions are found in handler.h
-    They are not probably completely right.
-  */
-  current_position(0), next_position(0), chain_alloced(0),
-  chain_size(DEFAULT_CHAIN_LENGTH), records_is_known(0)
-{
-  /* Set our original buffers from pre-allocated memory */
-  buffer.set(byte_buffer, IO_SIZE, system_charset_info);
-  chain= chain_buffer;
-}
-
-/*
-  Encode a buffer into the quoted format.
-*/
-
-int ha_tina::encode_quote(byte *buf)
-{
-  char attribute_buffer[1024];
-  String attribute(attribute_buffer, sizeof(attribute_buffer), &my_charset_bin);
-
-  buffer.length(0);
-  for (Field **field=table->field ; *field ; field++)
-  {
-    const char *ptr;
-    const char *end_ptr;
-
-    (*field)->val_str(&attribute,&attribute);
-    ptr= attribute.ptr();
-    end_ptr= attribute.length() + ptr;
-
-    buffer.append('"');
-
-    while (ptr < end_ptr) 
-    {
-      if (*ptr == '"')
-      {
-        buffer.append('\\');
-        buffer.append('"');
-        *ptr++;
-      }
-      else if (*ptr == '\r')
-      {
-        buffer.append('\\');
-        buffer.append('r');
-        *ptr++;
-      }
-      else if (*ptr == '\\')
-      {
-        buffer.append('\\');
-        buffer.append('\\');
-        *ptr++;
-      }
-      else if (*ptr == '\n')
-      {
-        buffer.append('\\');
-        buffer.append('n');
-        *ptr++;
-      }
-      else
-        buffer.append(*ptr++);
-    }
-    buffer.append('"');
-    buffer.append(',');
-  }
-  // Remove the comma, add a line feed
-  buffer.length(buffer.length() - 1);
-  buffer.append('\n');
-  //buffer.replace(buffer.length(), 0, "\n", 1);
-
-  return (buffer.length());
-}
-
-/*
-  chain_append() adds delete positions to the chain that we use to keep
-  track of space. Then the chain will be used to cleanup "holes", occured
-  due to deletes and updates.
-*/
-int ha_tina::chain_append()
-{
-  if ( chain_ptr != chain && (chain_ptr -1)->end == current_position)
-    (chain_ptr -1)->end= next_position;
-  else
-  {
-    /* We set up for the next position */
-    if ((off_t)(chain_ptr - chain) == (chain_size -1))
-    {
-      off_t location= chain_ptr - chain;
-      chain_size += DEFAULT_CHAIN_LENGTH;
-      if (chain_alloced)
-      {
-        /* Must cast since my_malloc unlike malloc doesn't have a void ptr */
-        if ((chain= (tina_set *) my_realloc((gptr)chain,
-                                            chain_size, MYF(MY_WME))) == NULL)
-          return -1;
-      }
-      else
-      {
-        tina_set *ptr= (tina_set *) my_malloc(chain_size * sizeof(tina_set),
-                                              MYF(MY_WME));
-        memcpy(ptr, chain, DEFAULT_CHAIN_LENGTH * sizeof(tina_set));
-        chain= ptr;
-        chain_alloced++;
-      }
-      chain_ptr= chain + location;
-    }
-    chain_ptr->begin= current_position;
-    chain_ptr->end= next_position;
-    chain_ptr++;
-  }
-
-  return 0;
-}
-
-
-/*
-  Scans for a row.
-*/
-int ha_tina::find_current_row(byte *buf)
-{
-  byte *mapped_ptr= (byte *)share->mapped_file + current_position;
-  byte *end_ptr;
-  DBUG_ENTER("ha_tina::find_current_row");
-
-  /* EOF should be counted as new line */
-  if ((end_ptr=  find_eoln(share->mapped_file, current_position,
-                           share->file_stat.st_size)) == 0)
-    DBUG_RETURN(HA_ERR_END_OF_FILE);
-
-  for (Field **field=table->field ; *field ; field++)
-  {
-    buffer.length(0);
-    mapped_ptr++; // Increment past the first quote
-    for(;mapped_ptr != end_ptr; mapped_ptr++)
-    {
-      // Need to convert line feeds!
-      if (*mapped_ptr == '"' &&
-          (((mapped_ptr[1] == ',') && (mapped_ptr[2] == '"')) ||
-           (mapped_ptr == end_ptr -1 )))
-      {
-        mapped_ptr += 2; // Move past the , and the "
-        break;
-      }
-      if (*mapped_ptr == '\\' && mapped_ptr != (end_ptr - 1))
-      {
-        mapped_ptr++;
-        if (*mapped_ptr == 'r')
-          buffer.append('\r');
-        else if (*mapped_ptr == 'n' )
-          buffer.append('\n');
-        else if ((*mapped_ptr == '\\') || (*mapped_ptr == '"'))
-          buffer.append(*mapped_ptr);
-        else  /* This could only happed with an externally created file */
-        {
-          buffer.append('\\');
-          buffer.append(*mapped_ptr);
-        }
-      }
-      else
-        buffer.append(*mapped_ptr);
-    }
-    (*field)->store(buffer.ptr(), buffer.length(), system_charset_info);
-  }
-  next_position= (end_ptr - share->mapped_file)+1;
-  /* Maybe use \N for null? */
-  memset(buf, 0, table->s->null_bytes); /* We do not implement nulls! */
-
-  DBUG_RETURN(0);
-}
-
-/*
-  If frm_error() is called in table.cc this is called to find out what file
-  extensions exist for this handler.
-*/
-static const char *ha_tina_exts[] = {
-  ".CSV",
-  NullS
-};
-
-const char **ha_tina::bas_ext() const
-{
-  return ha_tina_exts;
-}
-
-
-/*
-  Open a database file. Keep in mind that tables are caches, so
-  this will not be called for every request. Any sort of positions
-  that need to be reset should be kept in the ::extra() call.
-*/
-int ha_tina::open(const char *name, int mode, uint test_if_locked)
-{
-  DBUG_ENTER("ha_tina::open");
-
-  if (!(share= get_share(name, table)))
-    DBUG_RETURN(1);
-  thr_lock_data_init(&share->lock,&lock,NULL);
-  ref_length=sizeof(off_t);
-
-  DBUG_RETURN(0);
-}
-
-
-/*
-  Close a database file. We remove ourselves from the shared strucutre.
-  If it is empty we destroy it and free the mapped file.
-*/
-int ha_tina::close(void)
-{
-  DBUG_ENTER("ha_tina::close");
-  DBUG_RETURN(free_share(share));
-}
-
-/*
-  This is an INSERT. At the moment this handler just seeks to the end
-  of the file and appends the data. In an error case it really should
-  just truncate to the original position (this is not done yet).
-*/
-int ha_tina::write_row(byte * buf)
-{
-  int size;
-  DBUG_ENTER("ha_tina::write_row");
-
-  statistic_increment(table->in_use->status_var.ha_write_count, &LOCK_status);
-
-  if (table->timestamp_field_type & TIMESTAMP_AUTO_SET_ON_INSERT)
-    table->timestamp_field->set_time();
-
-  size= encode_quote(buf);
-
-  if (my_write(share->data_file, buffer.ptr(), size, MYF(MY_WME | MY_NABP)))
-    DBUG_RETURN(-1);
-
-  /*
-    Ok, this is means that we will be doing potentially bad things
-    during a bulk insert on some OS'es. What we need is a cleanup
-    call for ::write_row that would let us fix up everything after the bulk
-    insert. The archive handler does this with an extra mutx call, which
-    might be a solution for this.
-  */
-  if (get_mmap(share, 0) > 0)
-    DBUG_RETURN(-1);
-  records++;
-  DBUG_RETURN(0);
-}
-
-
-/*
-  This is called for an update.
-  Make sure you put in code to increment the auto increment, also
-  update any timestamp data. Currently auto increment is not being
-  fixed since autoincrements have yet to be added to this table handler.
-  This will be called in a table scan right before the previous ::rnd_next()
-  call.
-*/
-int ha_tina::update_row(const byte * old_data, byte * new_data)
-{
-  int size;
-  DBUG_ENTER("ha_tina::update_row");
-
-  statistic_increment(table->in_use->status_var.ha_read_rnd_next_count,
-		      &LOCK_status);
-
-  if (table->timestamp_field_type & TIMESTAMP_AUTO_SET_ON_UPDATE)
-    table->timestamp_field->set_time();
-
-  size= encode_quote(new_data);
-
-  if (chain_append())
-    DBUG_RETURN(-1);
-
-  if (my_write(share->data_file, buffer.ptr(), size, MYF(MY_WME | MY_NABP)))
-    DBUG_RETURN(-1);
-  DBUG_RETURN(0);
-}
-
-
-/*
-  Deletes a row. First the database will find the row, and then call this
-  method. In the case of a table scan, the previous call to this will be
-  the ::rnd_next() that found this row.
-  The exception to this is an ORDER BY. This will cause the table handler
-  to walk the table noting the positions of all rows that match a query.
-  The table will then be deleted/positioned based on the ORDER (so RANDOM,
-  DESC, ASC).
-*/
-int ha_tina::delete_row(const byte * buf)
-{
-  DBUG_ENTER("ha_tina::delete_row");
-  statistic_increment(table->in_use->status_var.ha_delete_count,
-                      &LOCK_status);
-
-  if (chain_append())
-    DBUG_RETURN(-1);
-
-  --records;
-
-  DBUG_RETURN(0);
-}
-
-/*
-  Fill buf with value from key. Simply this is used for a single index read
-  with a key.
-*/
-int ha_tina::index_read(byte * buf, const byte * key,
-                        uint key_len __attribute__((unused)),
-                        enum ha_rkey_function find_flag
-                        __attribute__((unused)))
-{
-  DBUG_ENTER("ha_tina::index_read");
-  DBUG_ASSERT(0);
-  DBUG_RETURN(HA_ADMIN_NOT_IMPLEMENTED);
-}
-
-/*
-  Fill buf with value from key. Simply this is used for a single index read
-  with a key.
-  Whatever the current key is we will use it. This is what will be in "index".
-*/
-int ha_tina::index_read_idx(byte * buf, uint index, const byte * key,
-                            uint key_len __attribute__((unused)),
-                            enum ha_rkey_function find_flag
-                            __attribute__((unused)))
-{
-  DBUG_ENTER("ha_tina::index_read_idx");
-  DBUG_ASSERT(0);
-  DBUG_RETURN(HA_ADMIN_NOT_IMPLEMENTED);
-}
-
-
-/*
-  Read the next position in the index.
-*/
-int ha_tina::index_next(byte * buf)
-{
-  DBUG_ENTER("ha_tina::index_next");
-  DBUG_ASSERT(0);
-  DBUG_RETURN(HA_ADMIN_NOT_IMPLEMENTED);
-}
-
-/*
-  Read the previous position in the index.
-*/
-int ha_tina::index_prev(byte * buf)
-{
-  DBUG_ENTER("ha_tina::index_prev");
-  DBUG_ASSERT(0);
-  DBUG_RETURN(HA_ADMIN_NOT_IMPLEMENTED);
-}
-
-/*
-  Read the first position in the index
-*/
-int ha_tina::index_first(byte * buf)
-{
-  DBUG_ENTER("ha_tina::index_first");
-  DBUG_ASSERT(0);
-  DBUG_RETURN(HA_ADMIN_NOT_IMPLEMENTED);
-}
-
-/*
-  Read the last position in the index
-  With this we don't need to do a filesort() with index.
-  We just read the last row and call previous.
-*/
-int ha_tina::index_last(byte * buf)
-{
-  DBUG_ENTER("ha_tina::index_last");
-  DBUG_ASSERT(0);
-  DBUG_RETURN(HA_ADMIN_NOT_IMPLEMENTED);
-}
-
-/*
-  All table scans call this first.
-  The order of a table scan is:
-
-  ha_tina::store_lock
-  ha_tina::external_lock
-  ha_tina::info
-  ha_tina::rnd_init
-  ha_tina::extra
-  ENUM HA_EXTRA_CACHE   Cash record in HA_rrnd()
-  ha_tina::rnd_next
-  ha_tina::rnd_next
-  ha_tina::rnd_next
-  ha_tina::rnd_next
-  ha_tina::rnd_next
-  ha_tina::rnd_next
-  ha_tina::rnd_next
-  ha_tina::rnd_next
-  ha_tina::rnd_next
-  ha_tina::extra
-  ENUM HA_EXTRA_NO_CACHE   End cacheing of records (def)
-  ha_tina::external_lock
-  ha_tina::extra
-  ENUM HA_EXTRA_RESET   Reset database to after open
-
-  Each call to ::rnd_next() represents a row returned in the can. When no more
-  rows can be returned, rnd_next() returns a value of HA_ERR_END_OF_FILE.
-  The ::info() call is just for the optimizer.
-
-*/
-
-int ha_tina::rnd_init(bool scan)
-{
-  DBUG_ENTER("ha_tina::rnd_init");
-
-  current_position= next_position= 0;
-  records= 0;
-  records_is_known= 0;
-  chain_ptr= chain;
-#ifdef HAVE_MADVISE
-  if (scan)
-    (void) madvise(share->mapped_file, share->file_stat.st_size,
-                   MADV_SEQUENTIAL);
-#endif
-
-  DBUG_RETURN(0);
-}
-
-/*
-  ::rnd_next() does all the heavy lifting for a table scan. You will need to
-  populate *buf with the correct field data. You can walk the field to
-  determine at what position you should store the data (take a look at how
-  ::find_current_row() works). The structure is something like:
-  0Foo  Dog  Friend
-  The first offset is for the first attribute. All space before that is
-  reserved for null count.
-  Basically this works as a mask for which rows are nulled (compared to just
-  empty).
-  This table handler doesn't do nulls and does not know the difference between
-  NULL and "". This is ok since this table handler is for spreadsheets and
-  they don't know about them either :)
-*/
-int ha_tina::rnd_next(byte *buf)
-{
-  DBUG_ENTER("ha_tina::rnd_next");
-
-  statistic_increment(table->in_use->status_var.ha_read_rnd_next_count,
-		      &LOCK_status);
-
-  current_position= next_position;
-  if (!share->mapped_file)
-    DBUG_RETURN(HA_ERR_END_OF_FILE);
-  if (HA_ERR_END_OF_FILE == find_current_row(buf) )
-    DBUG_RETURN(HA_ERR_END_OF_FILE);
-
-  records++;
-  DBUG_RETURN(0);
-}
-
-/*
-  In the case of an order by rows will need to be sorted.
-  ::position() is called after each call to ::rnd_next(),
-  the data it stores is to a byte array. You can store this
-  data via my_store_ptr(). ref_length is a variable defined to the
-  class that is the sizeof() of position being stored. In our case
-  its just a position. Look at the bdb code if you want to see a case
-  where something other then a number is stored.
-*/
-void ha_tina::position(const byte *record)
-{
-  DBUG_ENTER("ha_tina::position");
-  my_store_ptr(ref, ref_length, current_position);
-  DBUG_VOID_RETURN;
-}
-
-
-/*
-  Used to fetch a row from a posiion stored with ::position().
-  my_get_ptr() retrieves the data for you.
-*/
-
-int ha_tina::rnd_pos(byte * buf, byte *pos)
-{
-  DBUG_ENTER("ha_tina::rnd_pos");
-  statistic_increment(table->in_use->status_var.ha_read_rnd_next_count,
-		      &LOCK_status);
-  current_position= my_get_ptr(pos,ref_length);
-  DBUG_RETURN(find_current_row(buf));
-}
-
-/*
-  ::info() is used to return information to the optimizer.
-  Currently this table handler doesn't implement most of the fields
-  really needed. SHOW also makes use of this data
-*/
-void ha_tina::info(uint flag)
-{
-  DBUG_ENTER("ha_tina::info");
-  /* This is a lie, but you don't want the optimizer to see zero or 1 */
-  if (!records_is_known && records < 2) 
-    records= 2;
-  DBUG_VOID_RETURN;
-}
-
-/*
-  Grab bag of flags that are sent to the able handler every so often.
-  HA_EXTRA_RESET and HA_EXTRA_RESET_STATE are the most frequently called.
-  You are not required to implement any of these.
-*/
-int ha_tina::extra(enum ha_extra_function operation)
-{
-  DBUG_ENTER("ha_tina::extra");
-  DBUG_RETURN(0);
-}
-
-/*
-  This is no longer used.
-*/
-int ha_tina::reset(void)
-{
-  DBUG_ENTER("ha_tina::reset");
-  ha_tina::extra(HA_EXTRA_RESET);
-  DBUG_RETURN(0);
-}
-
-
-/*
-  Called after each table scan. In particular after deletes,
-  and updates. In the last case we employ chain of deleted
-  slots to clean up all of the dead space we have collected while
-  performing deletes/updates.
-*/
-int ha_tina::rnd_end()
-{
-  DBUG_ENTER("ha_tina::rnd_end");
-
-  records_is_known= 1;
-
-  /* First position will be truncate position, second will be increment */
-  if ((chain_ptr - chain)  > 0)
-  {
-    tina_set *ptr;
-    off_t length;
-
-    /*
-      Setting up writable map, this will contain all of the data after the
-      get_mmap call that we have added to the file.
-    */
-    if (get_mmap(share, 1) > 0)
-      DBUG_RETURN(-1);
-    length= share->file_stat.st_size;
-
-    /*
-      The sort handles updates/deletes with random orders.
-      It also sorts so that we move the final blocks to the
-      beginning so that we move the smallest amount of data possible.
-    */
-    qsort(chain, (size_t)(chain_ptr - chain), sizeof(tina_set),
-          (qsort_cmp)sort_set);
-    for (ptr= chain; ptr < chain_ptr; ptr++)
-    {
-      memmove(share->mapped_file + ptr->begin, share->mapped_file + ptr->end,
-              length - (size_t)ptr->end);
-      length= length - (size_t)(ptr->end - ptr->begin);
-    }
-
-    /* Truncate the file to the new size */
-    if (my_chsize(share->data_file, length, 0, MYF(MY_WME)))
-      DBUG_RETURN(-1);
-
-    if (my_munmap(share->mapped_file, length))
-      DBUG_RETURN(-1);
-
-    /* We set it to null so that get_mmap() won't try to unmap it */
-    share->mapped_file= NULL;
-    if (get_mmap(share, 0) > 0)
-      DBUG_RETURN(-1);
-  }
-
-  DBUG_RETURN(0);
-}
-
-
-/* 
-  DELETE without WHERE calls this
-*/
-
-int ha_tina::delete_all_rows()
-{
-  DBUG_ENTER("ha_tina::delete_all_rows");
-
-  if (!records_is_known)
-    return (my_errno=HA_ERR_WRONG_COMMAND);
-
-  int rc= my_chsize(share->data_file, 0, 0, MYF(MY_WME));
-
-  if (get_mmap(share, 0) > 0)
-    DBUG_RETURN(-1);
-
-  records=0;
-  DBUG_RETURN(rc);
-}
-
-/*
-  Always called by the start of a transaction (or by "lock tables");
-*/
-int ha_tina::external_lock(THD *thd, int lock_type)
-{
-  DBUG_ENTER("ha_tina::external_lock");
-  DBUG_RETURN(0);          // No external locking
-}
-
-/*
-  Called by the database to lock the table. Keep in mind that this
-  is an internal lock.
-*/
-THR_LOCK_DATA **ha_tina::store_lock(THD *thd,
-                                    THR_LOCK_DATA **to,
-                                    enum thr_lock_type lock_type)
-{
-  if (lock_type != TL_IGNORE && lock.type == TL_UNLOCK)
-    lock.type=lock_type;
-  *to++= &lock;
-  return to;
-}
-
-/* 
-  Create a table. You do not want to leave the table open after a call to
-  this (the database will call ::open() if it needs to).
-*/
-
-int ha_tina::create(const char *name, TABLE *table_arg,
-                    HA_CREATE_INFO *create_info)
-{
-  char name_buff[FN_REFLEN];
-  File create_file;
-  DBUG_ENTER("ha_tina::create");
-
-  if ((create_file= my_create(fn_format(name_buff, name, "", ".CSV",
-                                        MY_REPLACE_EXT|MY_UNPACK_FILENAME),0,
-                              O_RDWR | O_TRUNC,MYF(MY_WME))) < 0)
-    DBUG_RETURN(-1);
-
-  my_close(create_file,MYF(0));
-
-  DBUG_RETURN(0);
-}
-
diff --git a/sql/examples/ha_tina.h b/sql/examples/ha_tina.h
deleted file mode 100644
index c46750fb703..00000000000
--- a/sql/examples/ha_tina.h
+++ /dev/null
@@ -1,130 +0,0 @@
-/* Copyright (C) 2003 MySQL AB
-
-   This program is free software; you can redistribute it and/or modify
-   it under the terms of the GNU General Public License as published by
-   the Free Software Foundation; either version 2 of the License, or
-   (at your option) any later version.
-
-   This program is distributed in the hope that it will be useful,
-   but WITHOUT ANY WARRANTY; without even the implied warranty of
-   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-   GNU General Public License for more details.
-
-   You should have received a copy of the GNU General Public License
-   along with this program; if not, write to the Free Software
-   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
-
-#include <sys/types.h>
-#include <sys/stat.h>
-#include <my_dir.h>
-
-#define DEFAULT_CHAIN_LENGTH 512
-
-typedef struct st_tina_share {
-  char *table_name;
-  byte *mapped_file;                /* mapped region of file */
-  uint table_name_length,use_count;
-  MY_STAT file_stat;                /* Stat information for the data file */
-  File data_file;                   /* Current open data file */
-  pthread_mutex_t mutex;
-  THR_LOCK lock;
-} TINA_SHARE;
-
-typedef struct tina_set {
-	off_t begin;
-	off_t end;
-};
-
-class ha_tina: public handler
-{
-  THR_LOCK_DATA lock;      /* MySQL lock */
-  TINA_SHARE *share;       /* Shared lock info */
-  off_t current_position;  /* Current position in the file during a file scan */
-  off_t next_position;     /* Next position in the file scan */
-  byte byte_buffer[IO_SIZE];
-  String buffer;
-  /*
-    The chain contains "holes" in the file, occured because of
-    deletes/updates. It is used in rnd_end() to get rid of them
-    in the end of the query.
-  */
-  tina_set chain_buffer[DEFAULT_CHAIN_LENGTH];
-  tina_set *chain;
-  tina_set *chain_ptr;
-  byte chain_alloced;
-  uint32 chain_size;
-  bool records_is_known;
-
-public:
-  ha_tina(TABLE_SHARE *table_arg);
-  ~ha_tina() 
-  {
-    if (chain_alloced)
-      my_free((gptr)chain,0);
-  }
-  const char *table_type() const { return "CSV"; }
-  const char *index_type(uint inx) { return "NONE"; }
-  const char **bas_ext() const;
-  ulong table_flags() const
-  {
-    return (HA_REC_NOT_IN_SEQ | HA_NOT_EXACT_COUNT | 
-      HA_NO_AUTO_INCREMENT );
-  }
-  ulong index_flags(uint idx, uint part, bool all_parts) const
-  {
-    /* We will never have indexes so this will never be called(AKA we return zero) */
-    return 0;
-  }
-  uint max_record_length() const { return HA_MAX_REC_LENGTH; }
-  uint max_keys()          const { return 0; }
-  uint max_key_parts()     const { return 0; }
-  uint max_key_length()    const { return 0; }
-  /*
-     Called in test_quick_select to determine if indexes should be used.
-   */
-  virtual double scan_time() { return (double) (records+deleted) / 20.0+10; }
-  /* The next method will never be called */
-  virtual bool fast_key_read() { return 1;}
-  /* 
-    TODO: return actual upper bound of number of records in the table.
-    (e.g. save number of records seen on full table scan and/or use file size
-    as upper bound)
-  */
-  ha_rows estimate_rows_upper_bound() { return HA_POS_ERROR; }
-
-  int open(const char *name, int mode, uint test_if_locked);
-  int close(void);
-  int write_row(byte * buf);
-  int update_row(const byte * old_data, byte * new_data);
-  int delete_row(const byte * buf);
-  int index_read(byte * buf, const byte * key,
-      uint key_len, enum ha_rkey_function find_flag);
-  int index_read_idx(byte * buf, uint idx, const byte * key,
-      uint key_len, enum ha_rkey_function find_flag);
-  int index_next(byte * buf);
-  int index_prev(byte * buf);
-  int index_first(byte * buf);
-  int index_last(byte * buf);
-  int rnd_init(bool scan=1);
-  int rnd_next(byte *buf);
-  int rnd_pos(byte * buf, byte *pos);
-  int rnd_end();
-  void position(const byte *record);
-  void info(uint);
-  int extra(enum ha_extra_function operation);
-  int reset(void);
-  int external_lock(THD *thd, int lock_type);
-  int delete_all_rows(void);
-  int create(const char *name, TABLE *form, HA_CREATE_INFO *create_info);
-
-  THR_LOCK_DATA **store_lock(THD *thd, THR_LOCK_DATA **to,
-      enum thr_lock_type lock_type);
-
-  /* The following methods were added just for TINA */
-  int encode_quote(byte *buf);
-  int find_current_row(byte *buf);
-  int chain_append();
-};
-
-int tina_end(ha_panic_function type);
-
diff --git a/sql/ha_archive.cc b/sql/ha_archive.cc
index b5c2f2a6555..e5f93ea063c 100644
--- a/sql/ha_archive.cc
+++ b/sql/ha_archive.cc
@@ -140,6 +140,7 @@ static handler *archive_create_handler(TABLE_SHARE *table);
 
 /* dummy handlerton - only to have something to return from archive_db_init */
 handlerton archive_hton = {
+  MYSQL_HANDLERTON_INTERFACE_VERSION,
   "ARCHIVE",
   SHOW_OPTION_YES,
   "Archive storage engine", 
@@ -163,12 +164,9 @@ handlerton archive_hton = {
   archive_create_handler,    /* Create a new handler */
   NULL,    /* Drop a database */
   archive_db_end,    /* Panic call */
-  NULL,    /* Release temporary latches */
-  NULL,    /* Update Statistics */
   NULL,    /* Start Consistent Snapshot */
   NULL,    /* Flush logs */
   NULL,    /* Show status */
-  NULL,    /* Replication Report Sent Binlog */
   HTON_NO_FLAGS
 };
 
diff --git a/sql/ha_berkeley.cc b/sql/ha_berkeley.cc
index 7ef7b6bbf0f..b5d231ee8f8 100644
--- a/sql/ha_berkeley.cc
+++ b/sql/ha_berkeley.cc
@@ -114,20 +114,24 @@ static void berkeley_noticecall(DB_ENV *db_env, db_notices notice);
 static int berkeley_close_connection(THD *thd);
 static int berkeley_commit(THD *thd, bool all);
 static int berkeley_rollback(THD *thd, bool all);
+static int berkeley_rollback_to_savepoint(THD* thd, void *savepoint);
+static int berkeley_savepoint(THD* thd, void *savepoint);
+static int berkeley_release_savepoint(THD* thd, void *savepoint);
 static handler *berkeley_create_handler(TABLE_SHARE *table);
 
 handlerton berkeley_hton = {
+  MYSQL_HANDLERTON_INTERFACE_VERSION,
   "BerkeleyDB",
   SHOW_OPTION_YES,
   "Supports transactions and page-level locking", 
   DB_TYPE_BERKELEY_DB,
   berkeley_init,
   0, /* slot */
-  0, /* savepoint size */
+  sizeof(DB_TXN *), /* savepoint size */
   berkeley_close_connection,
-  NULL, /* savepoint_set */
-  NULL, /* savepoint_rollback */
-  NULL, /* savepoint_release */
+  berkeley_savepoint, /* savepoint_set */
+  berkeley_rollback_to_savepoint, /* savepoint_rollback */
+  berkeley_release_savepoint, /* savepoint_release */
   berkeley_commit,
   berkeley_rollback,
   NULL, /* prepare */
@@ -140,12 +144,9 @@ handlerton berkeley_hton = {
   berkeley_create_handler, /* Create a new handler */
   NULL, /* Drop a database */
   berkeley_end, /* Panic call */
-  NULL, /* Release temporary latches */
-  NULL, /* Update Statistics */
   NULL, /* Start Consistent Snapshot */
   berkeley_flush_logs, /* Flush logs */
   berkeley_show_status, /* Show status */
-  NULL, /* Replication Report Sent Binlog */
   HTON_CLOSE_CURSORS_AT_COMMIT | HTON_FLUSH_AFTER_RENAME
 };
 
@@ -157,6 +158,7 @@ handler *berkeley_create_handler(TABLE_SHARE *table)
 typedef struct st_berkeley_trx_data {
   DB_TXN *all;
   DB_TXN *stmt;
+  DB_TXN *sp_level;
   uint bdb_lock_count;
 } berkeley_trx_data;
 
@@ -309,10 +311,53 @@ static int berkeley_rollback(THD *thd, bool all)
   DBUG_RETURN(error);
 }
 
+static int berkeley_savepoint(THD* thd, void *savepoint)
+{
+  int error;
+  DB_TXN **save_txn= (DB_TXN**) savepoint;
+  DBUG_ENTER("berkeley_savepoint");
+  berkeley_trx_data *trx=(berkeley_trx_data *)thd->ha_data[berkeley_hton.slot];
+  if (!(error= db_env->txn_begin(db_env, trx->sp_level, save_txn, 0)))
+  {
+    trx->sp_level= *save_txn;
+  }
+  DBUG_RETURN(error);
+}
+
+static int berkeley_rollback_to_savepoint(THD* thd, void *savepoint)
+{
+  int error;
+  DB_TXN *parent, **save_txn= (DB_TXN**) savepoint;
+  DBUG_ENTER("berkeley_rollback_to_savepoint");
+  berkeley_trx_data *trx=(berkeley_trx_data *)thd->ha_data[berkeley_hton.slot];
+  parent= (*save_txn)->parent;
+  if (!(error= (*save_txn)->abort(*save_txn)))
+  {
+    trx->sp_level= parent;
+    error= berkeley_savepoint(thd, savepoint);
+  }
+  DBUG_RETURN(error);
+}
+
+static int berkeley_release_savepoint(THD* thd, void *savepoint)
+{
+  int error;
+  DB_TXN *parent, **save_txn= (DB_TXN**) savepoint;
+  DBUG_ENTER("berkeley_release_savepoint");
+  berkeley_trx_data *trx=(berkeley_trx_data *)thd->ha_data[berkeley_hton.slot];
+  parent= (*save_txn)->parent;
+  if (!(error= (*save_txn)->commit(*save_txn,0)))
+  {
+    trx->sp_level= parent;
+    *save_txn= 0;
+  }
+  DBUG_RETURN(error);
+}
 
 static bool berkeley_show_logs(THD *thd, stat_print_fn *stat_print)
 {
   char **all_logs, **free_logs, **a, **f;
+  uint hton_name_len= strlen(berkeley_hton.name);
   int error=1;
   MEM_ROOT **root_ptr= my_pthread_getspecific_ptr(MEM_ROOT**,THR_MALLOC);
   MEM_ROOT show_logs_root, *old_mem_root= *root_ptr;
@@ -337,19 +382,20 @@ static bool berkeley_show_logs(THD *thd, stat_print_fn *stat_print)
   {
     for (a = all_logs, f = free_logs; *a; ++a)
     {
-      const char *status;
       if (f && *f && strcmp(*a, *f) == 0)
       {
         f++;
-        status= SHOW_LOG_STATUS_FREE;
+        if ((error= stat_print(thd, berkeley_hton.name, hton_name_len,
+                               *a, strlen(*a),
+                               STRING_WITH_LEN(SHOW_LOG_STATUS_FREE))))
+          break;
       }
       else
-        status= SHOW_LOG_STATUS_INUSE;
-  
-      if (stat_print(thd, berkeley_hton.name, *a, status))
       {
-        error=1;
-        goto err;
+        if ((error= stat_print(thd, berkeley_hton.name, hton_name_len,
+                               *a, strlen(*a),
+                               STRING_WITH_LEN(SHOW_LOG_STATUS_INUSE))))
+          break;
       }
     }
   }
@@ -1882,6 +1928,8 @@ int ha_berkeley::external_lock(THD *thd, int lock_type)
     if (!trx)
       DBUG_RETURN(1);
   }
+  if (trx->all == 0)
+    trx->sp_level= 0;
   if (lock_type != F_UNLCK)
   {
     if (!trx->bdb_lock_count++)
@@ -1900,12 +1948,13 @@ int ha_berkeley::external_lock(THD *thd, int lock_type)
           trx->bdb_lock_count--;        // We didn't get the lock
           DBUG_RETURN(error);
 	}
+        trx->sp_level= trx->all;
         trans_register_ha(thd, TRUE, &berkeley_hton);
 	if (thd->in_lock_tables)
 	  DBUG_RETURN(0);			// Don't create stmt trans
       }
       DBUG_PRINT("trans",("starting transaction stmt"));
-      if ((error= db_env->txn_begin(db_env, trx->all, &trx->stmt, 0)))
+      if ((error= db_env->txn_begin(db_env, trx->sp_level, &trx->stmt, 0)))
       {
 	/* We leave the possible master transaction open */
         trx->bdb_lock_count--;                  // We didn't get the lock
@@ -1959,7 +2008,7 @@ int ha_berkeley::start_stmt(THD *thd, thr_lock_type lock_type)
   if (!trx->stmt)
   {
     DBUG_PRINT("trans",("starting transaction stmt"));
-    error= db_env->txn_begin(db_env, trx->all, &trx->stmt, 0);
+    error= db_env->txn_begin(db_env, trx->sp_level, &trx->stmt, 0);
     trans_register_ha(thd, FALSE, &berkeley_hton);
   }
   transaction= trx->stmt;
diff --git a/sql/ha_blackhole.cc b/sql/ha_blackhole.cc
index e2b0fc29d86..615836b9867 100644
--- a/sql/ha_blackhole.cc
+++ b/sql/ha_blackhole.cc
@@ -30,6 +30,7 @@ static handler *blackhole_create_handler(TABLE_SHARE *table);
 /* Blackhole storage engine handlerton */
 
 handlerton blackhole_hton= {
+  MYSQL_HANDLERTON_INTERFACE_VERSION,
   "BLACKHOLE",
   SHOW_OPTION_YES,
   "/dev/null storage engine (anything you write to it disappears)",
@@ -53,12 +54,9 @@ handlerton blackhole_hton= {
   blackhole_create_handler,    /* Create a new handler */
   NULL,    /* Drop a database */
   NULL,    /* Panic call */
-  NULL,    /* Release temporary latches */
-  NULL,    /* Update Statistics */
   NULL,    /* Start Consistent Snapshot */
   NULL,    /* Flush logs */
   NULL,    /* Show status */
-  NULL,    /* Replication Report Sent Binlog */
   HTON_CAN_RECREATE
 };
 
diff --git a/sql/ha_federated.cc b/sql/ha_federated.cc
index e0f0b6ee35a..d90f2506e2b 100644
--- a/sql/ha_federated.cc
+++ b/sql/ha_federated.cc
@@ -367,6 +367,7 @@ static int federated_rollback(THD *thd, bool all);
 /* Federated storage engine handlerton */
 
 handlerton federated_hton= {
+  MYSQL_HANDLERTON_INTERFACE_VERSION,
   "FEDERATED",
   SHOW_OPTION_YES,
   "Federated MySQL storage engine", 
@@ -390,12 +391,9 @@ handlerton federated_hton= {
   federated_create_handler,    /* Create a new handler */
   NULL,    /* Drop a database */
   federated_db_end,    /* Panic call */
-  NULL,    /* Release temporary latches */
-  NULL,    /* Update Statistics */
   NULL,    /* Start Consistent Snapshot */
   NULL,    /* Flush logs */
   NULL,    /* Show status */
-  NULL,    /* Replication Report Sent Binlog */
   HTON_ALTER_NOT_SUPPORTED
 };
 
diff --git a/sql/ha_heap.cc b/sql/ha_heap.cc
index ddc6c1bfb8f..a83a95ac863 100644
--- a/sql/ha_heap.cc
+++ b/sql/ha_heap.cc
@@ -27,6 +27,7 @@
 static handler *heap_create_handler(TABLE_SHARE *table);
 
 handlerton heap_hton= {
+  MYSQL_HANDLERTON_INTERFACE_VERSION,
   "MEMORY",
   SHOW_OPTION_YES,
   "Hash based, stored in memory, useful for temporary tables", 
@@ -50,12 +51,9 @@ handlerton heap_hton= {
   heap_create_handler,    /* Create a new handler */
   NULL,    /* Drop a database */
   heap_panic,    /* Panic call */
-  NULL,    /* Release temporary latches */
-  NULL,    /* Update Statistics */
   NULL,    /* Start Consistent Snapshot */
   NULL,    /* Flush logs */
   NULL,    /* Show status */
-  NULL,    /* Replication Report Sent Binlog */
   HTON_CAN_RECREATE
 };
 
diff --git a/sql/ha_innodb.cc b/sql/ha_innodb.cc
index 1562803aed7..c2e2b64201d 100644
--- a/sql/ha_innodb.cc
+++ b/sql/ha_innodb.cc
@@ -208,6 +208,7 @@ static int innobase_release_savepoint(THD* thd, void *savepoint);
 static handler *innobase_create_handler(TABLE_SHARE *table);
 
 handlerton innobase_hton = {
+  MYSQL_HANDLERTON_INTERFACE_VERSION,
   "InnoDB",
   SHOW_OPTION_YES,
   "Supports transactions, row-level locking, and foreign keys", 
@@ -231,16 +232,9 @@ handlerton innobase_hton = {
   innobase_create_handler,	/* Create a new handler */
   innobase_drop_database,	/* Drop a database */
   innobase_end,			/* Panic call */
-  innobase_release_temporary_latches,    /* Release temporary latches */
-  innodb_export_status,		/* Update Statistics */
   innobase_start_trx_and_assign_read_view,    /* Start Consistent Snapshot */
   innobase_flush_logs,		/* Flush logs */
   innobase_show_status,		/* Show status */
-#ifdef HAVE_REPLICATION
-  innobase_repl_report_sent_binlog,    /* Replication Report Sent Binlog */
-#else
-  NULL,
-#endif
   HTON_NO_FLAGS
 };
 
@@ -1977,6 +1971,11 @@ innobase_repl_report_sent_binlog(
         int     cmp;
         ibool   can_release_threads     = 0;
 
+	if (!innodb_inited) {
+		
+		return 0;
+	}
+
         /* If synchronous replication is not switched on, or this thd is
         sending binlog to a slave where we do not need synchronous replication,
         then return immediately */
@@ -6472,10 +6471,11 @@ ha_innobase::transactional_table_lock(
 Here we export InnoDB status variables to MySQL.  */
 
 int
-innodb_export_status(void)
+innodb_export_status()
 /*======================*/
 {
-  srv_export_innodb_status();
+  if (innodb_inited)
+    srv_export_innodb_status();
   return 0;
 }
 
@@ -6562,7 +6562,8 @@ innodb_show_status(
 
 	bool result = FALSE;
 
-	if (stat_print(thd, innobase_hton.name, "", str)) {
+	if (stat_print(thd, innobase_hton.name, strlen(innobase_hton.name), 
+                       STRING_WITH_LEN(""), str, flen)) {
 		result= TRUE;
 	}
 	my_free(str, MYF(0));
@@ -6587,6 +6588,7 @@ innodb_mutex_show_status(
   ulint   rw_lock_count_os_wait= 0;
   ulint   rw_lock_count_os_yield= 0;
   ulonglong rw_lock_wait_time= 0;
+  uint    hton_name_len= strlen(innobase_hton.name), buf1len, buf2len;
   DBUG_ENTER("innodb_mutex_show_status");
 
 #ifdef MUTEX_PROTECT_TO_BE_ADDED_LATER
@@ -6601,16 +6603,17 @@ innodb_mutex_show_status(
     {
       if (mutex->count_using > 0)
       {
-	my_snprintf(buf1, sizeof(buf1), "%s:%s",
-		    mutex->cmutex_name, mutex->cfile_name);
-	my_snprintf(buf2, sizeof(buf2),
-		    "count=%lu, spin_waits=%lu, spin_rounds=%lu, "
-		    "os_waits=%lu, os_yields=%lu, os_wait_times=%lu",
-		    mutex->count_using, mutex->count_spin_loop,
-		    mutex->count_spin_rounds,
-		    mutex->count_os_wait, mutex->count_os_yield,
-		    mutex->lspent_time/1000);
-	if (stat_print(thd, innobase_hton.name, buf1, buf2))
+	buf1len= my_snprintf(buf1, sizeof(buf1), "%s:%s",
+		             mutex->cmutex_name, mutex->cfile_name);
+	buf2len= my_snprintf(buf2, sizeof(buf2),
+		             "count=%lu, spin_waits=%lu, spin_rounds=%lu, "
+		             "os_waits=%lu, os_yields=%lu, os_wait_times=%lu",
+		             mutex->count_using, mutex->count_spin_loop,
+		             mutex->count_spin_rounds,
+		             mutex->count_os_wait, mutex->count_os_yield,
+		             mutex->lspent_time/1000);
+	if (stat_print(thd, innobase_hton.name, hton_name_len, 
+                       buf1, buf1len, buf2, buf2len))
         {
 #ifdef MUTEX_PROTECT_TO_BE_ADDED_LATER
           mutex_exit(&mutex_list_mutex);
@@ -6632,15 +6635,16 @@ innodb_mutex_show_status(
     mutex = UT_LIST_GET_NEXT(list, mutex);
   }
 
-  my_snprintf(buf2, sizeof(buf2),
-	      "count=%lu, spin_waits=%lu, spin_rounds=%lu, "
-	      "os_waits=%lu, os_yields=%lu, os_wait_times=%lu",
-	      rw_lock_count, rw_lock_count_spin_loop,
-	      rw_lock_count_spin_rounds,
-	      rw_lock_count_os_wait, rw_lock_count_os_yield,
-	      rw_lock_wait_time/1000);
+  buf2len= my_snprintf(buf2, sizeof(buf2),
+	               "count=%lu, spin_waits=%lu, spin_rounds=%lu, "
+	               "os_waits=%lu, os_yields=%lu, os_wait_times=%lu",
+	               rw_lock_count, rw_lock_count_spin_loop,
+	               rw_lock_count_spin_rounds,
+	               rw_lock_count_os_wait, rw_lock_count_os_yield,
+	               rw_lock_wait_time/1000);
 
-  if (stat_print(thd, innobase_hton.name, "rw_lock_mutexes", buf2))
+  if (stat_print(thd, innobase_hton.name, hton_name_len, 
+                 STRING_WITH_LEN("rw_lock_mutexes"), buf2, buf2len))
   {
     DBUG_RETURN(1);
   }
diff --git a/sql/ha_myisam.cc b/sql/ha_myisam.cc
index 02d814a9451..6bb0c7838ee 100644
--- a/sql/ha_myisam.cc
+++ b/sql/ha_myisam.cc
@@ -55,6 +55,7 @@ static handler *myisam_create_handler(TABLE_SHARE *table);
 /* MyISAM handlerton */
 
 handlerton myisam_hton= {
+  MYSQL_HANDLERTON_INTERFACE_VERSION,
   "MyISAM",
   SHOW_OPTION_YES,
   "Default engine as of MySQL 3.23 with great performance", 
@@ -82,12 +83,9 @@ handlerton myisam_hton= {
   myisam_create_handler,    /* Create a new handler */
   NULL,    /* Drop a database */
   mi_panic,/* Panic call */
-  NULL,    /* Release temporary latches */
-  NULL,    /* Update Statistics */
   NULL,    /* Start Consistent Snapshot */
   NULL,    /* Flush logs */
   NULL,    /* Show status */
-  NULL,    /* Replication Report Sent Binlog */
   HTON_CAN_RECREATE
 };
 
diff --git a/sql/ha_myisammrg.cc b/sql/ha_myisammrg.cc
index 37dfe34b40c..1be4fb62fa1 100644
--- a/sql/ha_myisammrg.cc
+++ b/sql/ha_myisammrg.cc
@@ -37,6 +37,7 @@ static handler *myisammrg_create_handler(TABLE_SHARE *table);
 /* MyISAM MERGE handlerton */
 
 handlerton myisammrg_hton= {
+  MYSQL_HANDLERTON_INTERFACE_VERSION,
   "MRG_MYISAM",
   SHOW_OPTION_YES,
   "Collection of identical MyISAM tables", 
@@ -60,12 +61,9 @@ handlerton myisammrg_hton= {
   myisammrg_create_handler,    /* Create a new handler */
   NULL,    /* Drop a database */
   myrg_panic,    /* Panic call */
-  NULL,    /* Release temporary latches */
-  NULL,    /* Update Statistics */
   NULL,    /* Start Consistent Snapshot */
   NULL,    /* Flush logs */
   NULL,    /* Show status */
-  NULL,    /* Replication Report Sent Binlog */
   HTON_CAN_RECREATE
 };
 
diff --git a/sql/ha_ndbcluster.cc b/sql/ha_ndbcluster.cc
index 75ee136c916..ebafbfa6431 100644
--- a/sql/ha_ndbcluster.cc
+++ b/sql/ha_ndbcluster.cc
@@ -59,6 +59,7 @@ static int ndbcluster_rollback(THD *thd, bool all);
 static handler* ndbcluster_create_handler(TABLE_SHARE *table);
 
 handlerton ndbcluster_hton = {
+  MYSQL_HANDLERTON_INTERFACE_VERSION,
   "ndbcluster",
   SHOW_OPTION_YES,
   "Clustered, fault-tolerant, memory-based tables", 
@@ -82,12 +83,9 @@ handlerton ndbcluster_hton = {
   ndbcluster_create_handler, /* Create a new handler */
   ndbcluster_drop_database, /* Drop a database */
   ndbcluster_end, /* Panic call */
-  NULL, /* Release temporary latches */
-  NULL, /* Update Statistics */
   NULL, /* Start Consistent Snapshot */
   NULL, /* Flush logs */
   ndbcluster_show_status, /* Show status */
-  NULL, /* Replication Report Sent Binlog */
   HTON_NO_FLAGS
 };
 
@@ -8060,10 +8058,12 @@ ndbcluster_show_status(THD* thd, stat_print_fn *stat_print,
     Ndb::Free_list_usage tmp; tmp.m_name= 0;
     while (ndb->get_free_list_usage(&tmp))
     {
-      my_snprintf(buf, sizeof(buf),
+      uint buflen=
+        my_snprintf(buf, sizeof(buf),
                   "created=%u, free=%u, sizeof=%u",
                   tmp.m_created, tmp.m_free, tmp.m_sizeof);
-      if (stat_print(thd, ndbcluster_hton.name, tmp.m_name, buf))
+      if (stat_print(thd, ndbcluster_hton.name, strlen(ndbcluster_hton.name),
+                     tmp.m_name, strlen(tmp.m_name), buf, buflen))
         DBUG_RETURN(TRUE);
     }
   }
diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index 650830832cb..2c6fde77abd 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -69,6 +69,7 @@ static PARTITION_SHARE *get_share(const char *table_name, TABLE * table);
 static handler *partition_create_handler(TABLE_SHARE *share);
 
 handlerton partition_hton = {
+  MYSQL_HANDLERTON_INTERFACE_VERSION,
   "partition",
   SHOW_OPTION_YES,
   "Partition Storage Engine Helper", /* A comment used by SHOW to describe an engine */
@@ -92,12 +93,9 @@ handlerton partition_hton = {
   partition_create_handler, /* Create a new handler */
   NULL, /* Drop a database */
   NULL, /* Panic call */
-  NULL, /* Release temporary latches */
-  NULL, /* Update Statistics */
   NULL, /* Start Consistent Snapshot */
   NULL, /* Flush logs */
   NULL, /* Show status */
-  NULL, /* Replication Report Sent Binlog */
   HTON_NOT_USER_SELECTABLE
 };
 
@@ -697,6 +695,7 @@ bool ha_partition::create_handler_file(const char *name)
 void ha_partition::clear_handler_file()
 {
   my_free((char*) m_file_buffer, MYF(MY_ALLOW_ZERO_PTR));
+  my_free((char*) m_engine_array, MYF(MY_ALLOW_ZERO_PTR));
   m_file_buffer= NULL;
   m_name_buffer_ptr= NULL;
   m_engine_array= NULL;
@@ -715,18 +714,19 @@ bool ha_partition::create_handlers()
   for (i= 0; i < m_tot_parts; i++)
   {
     if (!(m_file[i]= get_new_handler(table_share, current_thd->mem_root,
-                                     (enum db_type) m_engine_array[i])))
+                                     m_engine_array[i])))
       DBUG_RETURN(TRUE);
     DBUG_PRINT("info", ("engine_type: %u", m_engine_array[i]));
   }
   m_file[m_tot_parts]= 0;
   /* For the moment we only support partition over the same table engine */
-  if (m_engine_array[0] == (uchar) DB_TYPE_MYISAM)
+  if (m_engine_array[0] == &myisam_hton)
   {
     DBUG_PRINT("info", ("MyISAM"));
     m_myisam= TRUE;
   }
-  else if (m_engine_array[0] == (uchar) DB_TYPE_INNODB)
+  /* INNODB may not be compiled in... */
+  else if (ha_legacy_type(m_engine_array[0]) == DB_TYPE_INNODB)
   {
     DBUG_PRINT("info", ("InnoDB"));
     m_innodb= TRUE;
@@ -761,7 +761,7 @@ bool ha_partition::new_handlers_from_part_info()
     if (!(m_file[i]= get_new_handler(table_share, thd->mem_root,
                                      part_elem->engine_type)))
       goto error;
-    DBUG_PRINT("info", ("engine_type: %u", (uint) part_elem->engine_type));
+    DBUG_PRINT("info", ("engine_type: %u", (uint) ha_legacy_type(part_elem->engine_type)));
     if (m_is_sub_partitioned)
     {
       for (j= 0; j < m_part_info->no_subparts; j++)
@@ -769,11 +769,11 @@ bool ha_partition::new_handlers_from_part_info()
 	if (!(m_file[i]= get_new_handler(table_share, thd->mem_root,
                                          part_elem->engine_type)))
           goto error;
-	DBUG_PRINT("info", ("engine_type: %u", (uint) part_elem->engine_type));
+	DBUG_PRINT("info", ("engine_type: %u", (uint) ha_legacy_type(part_elem->engine_type)));
       }
     }
   } while (++i < m_part_info->no_parts);
-  if (part_elem->engine_type == DB_TYPE_MYISAM)
+  if (part_elem->engine_type == &myisam_hton)
   {
     DBUG_PRINT("info", ("MyISAM"));
     m_myisam= TRUE;
@@ -795,7 +795,7 @@ bool ha_partition::get_from_handler_file(const char *name)
   char buff[FN_REFLEN], *address_tot_name_len;
   File file;
   char *file_buffer, *name_buffer_ptr;
-  uchar *engine_array;
+  handlerton **engine_array;
   uint i, len_bytes, len_words, tot_partition_words, tot_name_words, chksum;
   DBUG_ENTER("ha_partition::get_from_handler_file");
   DBUG_PRINT("enter", ("table name: '%s'", name));
@@ -824,7 +824,11 @@ bool ha_partition::get_from_handler_file(const char *name)
     goto err2;
   m_tot_parts= uint4korr((file_buffer) + 8);
   tot_partition_words= (m_tot_parts + 3) / 4;
-  engine_array= (uchar *) ((file_buffer) + 12);
+  if (!(engine_array= (handlerton **) my_malloc(m_tot_parts * sizeof(handlerton*),MYF(0))))
+    goto err2;
+  for (i= 0; i < m_tot_parts; i++)
+    engine_array[i]= ha_resolve_by_legacy_type(current_thd, 
+                (enum legacy_db_type) *(uchar *) ((file_buffer) + 12 + i));
   address_tot_name_len= file_buffer + 12 + 4 * tot_partition_words;
   tot_name_words= (uint4korr(address_tot_name_len) + 3) / 4;
   if (len_words != (tot_partition_words + tot_name_words + 4))
diff --git a/sql/ha_partition.h b/sql/ha_partition.h
index a727a278103..760f99ad8aa 100644
--- a/sql/ha_partition.h
+++ b/sql/ha_partition.h
@@ -46,7 +46,7 @@ private:
   /* Data for the partition handler */
   char *m_file_buffer;                  // Buffer with names
   char *m_name_buffer_ptr;		// Pointer to first partition name
-  uchar *m_engine_array;                // Array of types of the handlers
+  handlerton **m_engine_array;          // Array of types of the handlers
   handler **m_file;                     // Array of references to handler inst.
   partition_info *m_part_info;          // local reference to partition
   byte *m_start_key_ref;                // Reference of start key in current
diff --git a/sql/handler.cc b/sql/handler.cc
index 7f825ffda35..1a6122aebbd 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -37,11 +37,32 @@
 #ifdef WITH_PARTITION_STORAGE_ENGINE
 #include "ha_partition.h"
 #endif
+#ifdef WITH_INNOBASE_STORAGE_ENGINE
+#include "ha_innodb.h"
+#endif
 
 extern handlerton *sys_table_types[];
-  
+
 /* static functions defined in this file */
 
+static handler *create_default(TABLE_SHARE *table);
+
+const handlerton default_hton =
+{
+  MYSQL_HANDLERTON_INTERFACE_VERSION,
+  "DEFAULT",
+  SHOW_OPTION_YES,
+  NULL,
+  DB_TYPE_DEFAULT,
+  NULL,
+  0, 0,
+  NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL,
+  NULL, NULL, NULL,
+  create_default,
+  NULL, NULL, NULL, NULL, NULL,
+  HTON_NO_FLAGS
+};
+
 static SHOW_COMP_OPTION have_yes= SHOW_OPTION_YES;
 
 /* number of entries in handlertons[] */
@@ -53,12 +74,12 @@ ulong savepoint_alloc_size;
 
 struct show_table_alias_st sys_table_aliases[]=
 {
-  {"INNOBASE",	"InnoDB"},
-  {"NDB", "NDBCLUSTER"},
-  {"BDB", "BERKELEYDB"},
-  {"HEAP", "MEMORY"},
-  {"MERGE", "MRG_MYISAM"},
-  {NullS, NullS}
+  {"INNOBASE",  DB_TYPE_INNODB},
+  {"NDB",       DB_TYPE_NDBCLUSTER},
+  {"BDB",       DB_TYPE_BERKELEY_DB},
+  {"HEAP",      DB_TYPE_HEAP},
+  {"MERGE",     DB_TYPE_MRG_MYISAM},
+  {NullS,       DB_TYPE_UNKNOWN}
 };
 
 const char *ha_row_type[] = {
@@ -74,26 +95,22 @@ TYPELIB tx_isolation_typelib= {array_elements(tx_isolation_names)-1,"",
 static TYPELIB known_extensions= {0,"known_exts", NULL, NULL};
 uint known_extensions_id= 0;
 
-enum db_type ha_resolve_by_name(const char *name, uint namelen)
+handlerton *ha_resolve_by_name(THD *thd, LEX_STRING *name)
 {
-  THD *thd= current_thd;
   show_table_alias_st *table_alias;
-  handlerton **types;
+  st_plugin_int *plugin;
 
   if (thd && !my_strnncoll(&my_charset_latin1,
-                           (const uchar *)name, namelen,
+                           (const uchar *)name->str, name->length,
                            (const uchar *)"DEFAULT", 7))
-    return (enum db_type) thd->variables.table_type;
+    return ha_resolve_by_legacy_type(thd, DB_TYPE_DEFAULT);
 
-retest:
-  for (types= sys_table_types; *types; types++)
+  if ((plugin= plugin_lock(name, MYSQL_STORAGE_ENGINE_PLUGIN)))
   {
-    if ((!my_strnncoll(&my_charset_latin1,
-                       (const uchar *)name, namelen,
-                       (const uchar *)(*types)->name,
-                        strlen((*types)->name))) &&
-        !((*types)->flags & HTON_NOT_USER_SELECTABLE))
-      return (enum db_type) (*types)->db_type;
+    handlerton *hton= (handlerton *) plugin->plugin->info;
+    if (!(hton->flags & HTON_NOT_USER_SELECTABLE))
+      return hton;
+    plugin_unlock(plugin);
   }
 
   /*
@@ -102,63 +119,99 @@ retest:
   for (table_alias= sys_table_aliases; table_alias->type; table_alias++)
   {
     if (!my_strnncoll(&my_charset_latin1,
-                      (const uchar *)name, namelen,
+                      (const uchar *)name->str, name->length,
                       (const uchar *)table_alias->alias,
                       strlen(table_alias->alias)))
-    {
-      name= table_alias->type;
-      namelen= strlen(name);
-      goto retest;
-    }
+      return ha_resolve_by_legacy_type(thd, table_alias->type);
   }
 
-  return DB_TYPE_UNKNOWN;
+  return NULL;
 }
 
 
-const char *ha_get_storage_engine(enum db_type db_type)
+struct plugin_find_dbtype_st
 {
-  handlerton **types;
-  for (types= sys_table_types; *types; types++)
+  enum legacy_db_type db_type;
+  handlerton *hton;
+};
+
+
+static my_bool plugin_find_dbtype(THD *unused, st_plugin_int *plugin,
+                                  void *arg)
+{
+  handlerton *types= (handlerton *) plugin->plugin->info;
+  if (types->db_type == ((struct plugin_find_dbtype_st *)arg)->db_type)
   {
-    if (db_type == (*types)->db_type)
-      return (*types)->name;
+    ((struct plugin_find_dbtype_st *)arg)->hton= types;
+    return TRUE;
   }
-  return "*NONE*";
+  return FALSE;
 }
 
 
-bool ha_check_storage_engine_flag(enum db_type db_type, uint32 flag)
+const char *ha_get_storage_engine(enum legacy_db_type db_type)
 {
-  handlerton **types;
-  for (types= sys_table_types; *types; types++)
+  struct plugin_find_dbtype_st info;
+  
+  switch (db_type)
   {
-    if (db_type == (*types)->db_type)
-      return test((*types)->flags & flag);
+  case DB_TYPE_DEFAULT:
+    return "DEFAULT";
+  case DB_TYPE_UNKNOWN:
+    return "UNKNOWN";
+  default:
+    info.db_type= db_type;
+
+    if (!plugin_foreach(NULL, plugin_find_dbtype, 
+                        MYSQL_STORAGE_ENGINE_PLUGIN, &info))
+      return "*NONE*";
+
+    return info.hton->name;
   }
-  return FALSE;                                 // No matching engine
 }
 
 
-my_bool ha_storage_engine_is_enabled(enum db_type database_type)
+static handler *create_default(TABLE_SHARE *table)
 {
-  handlerton **types;
-  for (types= sys_table_types; *types; types++)
+  handlerton *hton=ha_resolve_by_legacy_type(current_thd, DB_TYPE_DEFAULT);
+  return (hton && hton != &default_hton && hton->create) ? 
+        hton->create(table) : NULL;
+}
+
+
+handlerton *ha_resolve_by_legacy_type(THD *thd, enum legacy_db_type db_type)
+{
+  struct plugin_find_dbtype_st info;
+
+  switch (db_type)
   {
-    if (database_type == (*types)->db_type)
-      return ((*types)->state == SHOW_OPTION_YES) ? TRUE : FALSE;
+  case DB_TYPE_DEFAULT:
+    return (thd->variables.table_type != NULL) ?
+            thd->variables.table_type :
+            (global_system_variables.table_type != NULL ?
+             global_system_variables.table_type : &myisam_hton);
+  case DB_TYPE_UNKNOWN:
+    return NULL;
+  default:
+    info.db_type= db_type;
+    if (!plugin_foreach(NULL, plugin_find_dbtype, 
+                        MYSQL_STORAGE_ENGINE_PLUGIN, &info))
+      return NULL;
+
+    return info.hton;
   }
-  return FALSE;
 }
 
 
 /* Use other database handler if databasehandler is not compiled in */
 
-enum db_type ha_checktype(THD *thd, enum db_type database_type,
+handlerton *ha_checktype(THD *thd, enum legacy_db_type database_type,
                           bool no_substitute, bool report_error)
 {
-  if (ha_storage_engine_is_enabled(database_type))
-    return database_type;
+  handlerton *hton= ha_resolve_by_legacy_type(thd, database_type);
+  if (ha_storage_engine_is_enabled(hton))
+    return hton;
+
   if (no_substitute)
   {
     if (report_error)
@@ -166,34 +219,28 @@ enum db_type ha_checktype(THD *thd, enum db_type database_type,
       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;
+    return NULL;
   }
 
   switch (database_type) {
 #ifndef NO_HASH
   case DB_TYPE_HASH:
-    return (database_type);
+    return ha_resolve_by_legacy_type(thd, DB_TYPE_HASH);
 #endif
   case DB_TYPE_MRG_ISAM:
-    return (DB_TYPE_MRG_MYISAM);
+    return ha_resolve_by_legacy_type(thd, DB_TYPE_MRG_MYISAM);
   default:
     break;
   }
-  
-  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 !=
-           DB_TYPE_UNKNOWN ?
-           (enum db_type) global_system_variables.table_type : DB_TYPE_MYISAM)
-          );
+
+  return ha_resolve_by_legacy_type(thd, DB_TYPE_DEFAULT);  
 } /* ha_checktype */
 
 
 handler *get_new_handler(TABLE_SHARE *share, MEM_ROOT *alloc,
-                         enum db_type db_type)
+                         handlerton *db_type)
 {
   handler *file= NULL;
-  handlerton **types;
   /*
     handlers are allocated with new in the handlerton create() function
     we need to set the thd mem_root for these to be allocated correctly
@@ -201,20 +248,15 @@ handler *get_new_handler(TABLE_SHARE *share, MEM_ROOT *alloc,
   THD *thd= current_thd;
   MEM_ROOT *thd_save_mem_root= thd->mem_root;
   thd->mem_root= alloc;
-  for (types= sys_table_types; *types; types++)
-  {
-    if (db_type == (*types)->db_type && (*types)->create)
-    {
-      file= ((*types)->state == SHOW_OPTION_YES) ?
-		(*types)->create(share) : NULL;
-      break;
-    }
-  }
+
+  if (db_type != NULL && db_type->state == SHOW_OPTION_YES && db_type->create)
+    file= db_type->create(share);
+
   thd->mem_root= thd_save_mem_root;
 
   if (!file)
   {
-    enum db_type def=(enum db_type) current_thd->variables.table_type;
+    handlerton *def= current_thd->variables.table_type;
     /* Try first with 'default table type' */
     if (db_type != def)
       return get_new_handler(share, alloc, def);
@@ -341,16 +383,72 @@ static int ha_finish_errors(void)
 }
 
 
-static inline void ha_was_inited_ok(handlerton **ht)
+static void ha_was_inited_ok(handlerton *ht)
 {
-  uint tmp= (*ht)->savepoint_offset;
-  (*ht)->savepoint_offset= savepoint_alloc_size;
+  uint tmp= ht->savepoint_offset;
+  ht->savepoint_offset= savepoint_alloc_size;
   savepoint_alloc_size+= tmp;
-  (*ht)->slot= total_ha++;
-  if ((*ht)->prepare)
+  ht->slot= total_ha++;
+  if (ht->prepare)
     total_ha_2pc++;
 }
 
+
+int ha_initialize_handlerton(handlerton *hton)
+{
+  DBUG_ENTER("ha_initialize_handlerton");
+
+  if (hton == NULL)
+    DBUG_RETURN(1);
+
+  /* check major version */
+  if ((hton->interface_version>>24) != (MYSQL_HANDLERTON_INTERFACE_VERSION>>24))
+  {
+    sql_print_error("handlerton major version incompatible");
+    DBUG_PRINT("warning", ("handlerton major version incompatible"));
+    DBUG_RETURN(1);
+  }
+
+  /* check minor version */
+  if ((hton->interface_version>>16)&0xff < 
+        (MYSQL_HANDLERTON_INTERFACE_VERSION>>16)&0xff)
+  {
+    sql_print_error("handlerton minor version incompatible");
+    DBUG_PRINT("warning", ("handlerton minor version incompatible"));
+    DBUG_RETURN(1);
+  }
+
+  switch (hton->state)
+  {
+  case SHOW_OPTION_NO:
+    break;
+  case SHOW_OPTION_YES:
+    if (!hton->init || !hton->init())
+    {
+      ha_was_inited_ok(hton);
+      break;
+    }
+    /* fall through */
+  default:
+    hton->state= SHOW_OPTION_DISABLED;
+    break;
+  }
+  DBUG_RETURN(0);
+}
+
+
+static my_bool init_handlerton(THD *unused1, st_plugin_int *plugin,
+                               void *unused2)
+{
+  if (plugin->state == PLUGIN_IS_UNINITIALIZED)
+  {
+    ha_initialize_handlerton((handlerton *) plugin->plugin->info);
+    plugin->state= PLUGIN_IS_READY;
+  }
+  return FALSE;
+}
+
+
 int ha_init()
 {
   int error= 0;
@@ -361,16 +459,8 @@ int ha_init()
   if (ha_init_errors())
     return 1;
 
-  /*
-    We now initialize everything here.
-  */
-  for (types= sys_table_types; *types; types++)
-  {
-    if (!(*types)->init || !(*types)->init())
-      ha_was_inited_ok(types); 
-    else
-      (*types)->state= SHOW_OPTION_DISABLED;
-  }
+  if (plugin_foreach(NULL, init_handlerton, MYSQL_STORAGE_ENGINE_PLUGIN, 0))
+    return 1;
 
   DBUG_ASSERT(total_ha < MAX_HA);
   /*
@@ -383,43 +473,97 @@ int ha_init()
   return error;
 }
 
+
+int ha_register_builtin_plugins()
+{
+  handlerton **hton;
+  uint size= 0;
+  struct st_mysql_plugin *plugin;
+  DBUG_ENTER("ha_register_builtin_plugins");
+
+  for (hton= sys_table_types; *hton; hton++)
+    size+= sizeof(struct st_mysql_plugin);
+  
+  if (!(plugin= (struct st_mysql_plugin *)
+        my_once_alloc(size, MYF(MY_WME | MY_ZEROFILL))))
+    DBUG_RETURN(1);
+  
+  for (hton= sys_table_types; *hton; hton++, plugin++)
+  {
+    plugin->type= MYSQL_STORAGE_ENGINE_PLUGIN;
+    plugin->info= *hton;
+    plugin->version= 0;
+    plugin->name= (*hton)->name;
+    plugin->author= NULL;
+    plugin->descr= (*hton)->comment;
+    
+    if (plugin_register_builtin(plugin))
+      DBUG_RETURN(1);
+  }
+  DBUG_RETURN(0);
+}
+
+
+
+
 	/* close, flush or restart databases */
 	/* Ignore this for other databases than ours */
 
-int ha_panic(enum ha_panic_function flag)
+static my_bool panic_handlerton(THD *unused1, st_plugin_int *plugin,
+                               void *arg)
 {
-  int error=0;
-  handlerton **types;
+  handlerton *hton= (handlerton *) plugin->plugin->info;  
+  if (hton->state == SHOW_OPTION_YES && hton->panic)
+    ((int*)arg)[0]|= hton->panic((enum ha_panic_function)((int*)arg)[1]);
+  return FALSE;
+}
 
-  for (types= sys_table_types; *types; types++)
-  {
-    if ((*types)->state == SHOW_OPTION_YES && (*types)->panic)
-        error|= (*types)->panic(flag);
-  }
-  if (ha_finish_errors())
-    error= 1;
-  return error;
+
+int ha_panic(enum ha_panic_function flag)
+{
+  int error[2];
+  
+  error[0]= 0; error[1]= (int)flag;
+  plugin_foreach(NULL, panic_handlerton, MYSQL_STORAGE_ENGINE_PLUGIN, error);
+  
+  if (flag == HA_PANIC_CLOSE && ha_finish_errors())
+    error[0]= 1;
+  return error[0];
 } /* ha_panic */
 
+static my_bool dropdb_handlerton(THD *unused1, st_plugin_int *plugin,
+                                 void *path)
+{
+  handlerton *hton= (handlerton *) plugin->plugin->info;  
+  if (hton->state == SHOW_OPTION_YES && hton->drop_database)
+    hton->drop_database((char *)path);
+  return FALSE;
+}
+
+
 void ha_drop_database(char* path)
 {
-  handlerton **types;
+  plugin_foreach(NULL, dropdb_handlerton, MYSQL_STORAGE_ENGINE_PLUGIN, path);
+}
 
-  for (types= sys_table_types; *types; types++)
-  {
-    if ((*types)->state == SHOW_OPTION_YES && (*types)->drop_database)
-      (*types)->drop_database(path);
-  }
+
+static my_bool closecon_handlerton(THD *thd, st_plugin_int *plugin,
+                                   void *unused)
+{
+  handlerton *hton= (handlerton *) plugin->plugin->info;  
+  /* there's no need to rollback here as all transactions must 
+     be rolled back already */
+  if (hton->state == SHOW_OPTION_YES && hton->close_connection &&
+      thd->ha_data[hton->slot])
+    hton->close_connection(thd);
+  return FALSE;
 }
 
+
 /* don't bother to rollback here, it's done already */
 void ha_close_connection(THD* thd)
 {
-  handlerton **types;
-  for (types= sys_table_types; *types; types++)
-	/* XXX Maybe do a rollback if close_connection == NULL ? */
-    if (thd->ha_data[(*types)->slot] && (*types)->close_connection)
-      (*types)->close_connection(thd);
+  plugin_foreach(thd, closecon_handlerton, MYSQL_STORAGE_ENGINE_PLUGIN, 0);
 }
 
 /* ========================================================================
@@ -729,21 +873,46 @@ int ha_autocommit_or_rollback(THD *thd, int error)
 }
 
 
-int ha_commit_or_rollback_by_xid(XID *xid, bool commit)
+struct xahton_st {
+  XID *xid;
+  int result;
+};
+
+static my_bool xacommit_handlerton(THD *unused1, st_plugin_int *plugin,
+                                   void *arg)
 {
-  handlerton **types;
-  int res= 1;
+  handlerton *hton= (handlerton *) plugin->plugin->info;
+  if (hton->state == SHOW_OPTION_YES && hton->recover)
+  {
+    hton->commit_by_xid(((struct xahton_st *)arg)->xid);
+    ((struct xahton_st *)arg)->result= 0;
+  }
+  return FALSE;
+}
 
-  for (types= sys_table_types; *types; types++)
+static my_bool xarollback_handlerton(THD *unused1, st_plugin_int *plugin,
+                                     void *arg)
+{
+  handlerton *hton= (handlerton *) plugin->plugin->info;
+  if (hton->state == SHOW_OPTION_YES && hton->recover)
   {
-    if ((*types)->state == SHOW_OPTION_YES && (*types)->recover)
-    {
-      if ((*(commit ? (*types)->commit_by_xid :
-             (*types)->rollback_by_xid))(xid));
-      res= 0;
-    }
+    hton->rollback_by_xid(((struct xahton_st *)arg)->xid);
+    ((struct xahton_st *)arg)->result= 0;
   }
-  return res;
+  return FALSE;
+}
+
+
+int ha_commit_or_rollback_by_xid(XID *xid, bool commit)
+{
+  struct xahton_st xaop;
+  xaop.xid= xid;
+  xaop.result= 1;
+  
+  plugin_foreach(NULL, commit ? xacommit_handlerton : xarollback_handlerton,
+                 MYSQL_STORAGE_ENGINE_PLUGIN, &xaop);
+
+  return xaop.result;
 }
 
 
@@ -819,99 +988,123 @@ static char* xid_to_str(char *buf, XID *xid)
      in this case commit_list==0, tc_heuristic_recover == 0
      there should be no prepared transactions in this case.
 */
-int ha_recover(HASH *commit_list)
-{
-  int len, got, found_foreign_xids=0, found_my_xids=0;
-  handlerton **types;
-  XID *list=0;
-  bool dry_run=(commit_list==0 && tc_heuristic_recover==0);
-  DBUG_ENTER("ha_recover");
 
-  /* commit_list and tc_heuristic_recover cannot be set both */
-  DBUG_ASSERT(commit_list==0 || tc_heuristic_recover==0);
-  /* if either is set, total_ha_2pc must be set too */
-  DBUG_ASSERT(dry_run || total_ha_2pc>(ulong)opt_bin_log);
-
-  if (total_ha_2pc <= (ulong)opt_bin_log)
-    DBUG_RETURN(0);
-
-  if (commit_list)
-    sql_print_information("Starting crash recovery...");
-
-#ifndef WILL_BE_DELETED_LATER
-  /*
-    for now, only InnoDB supports 2pc. It means we can always safely
-    rollback all pending transactions, without risking inconsistent data
-  */
-  DBUG_ASSERT(total_ha_2pc == (ulong) opt_bin_log+1); // only InnoDB and binlog
-  tc_heuristic_recover= TC_HEURISTIC_RECOVER_ROLLBACK; // forcing ROLLBACK
-  dry_run=FALSE;
-#endif
-
-  for (len= MAX_XID_LIST_SIZE ; list==0 && len > MIN_XID_LIST_SIZE; len/=2)
-  {
-    list=(XID *)my_malloc(len*sizeof(XID), MYF(0));
-  }
-  if (!list)
-  {
-    sql_print_error(ER(ER_OUTOFMEMORY), len*sizeof(XID));
-    DBUG_RETURN(1);
-  }
+struct xarecover_st
+{
+  int len, found_foreign_xids, found_my_xids;
+  XID *list;
+  HASH *commit_list;
+  bool dry_run;
+};
 
-  for (types= sys_table_types; *types; types++)
+static my_bool xarecover_handlerton(THD *unused, st_plugin_int *plugin,
+                                    void *arg)
+{
+  handlerton *hton= (handlerton *) plugin->plugin->info;
+  struct xarecover_st *info= (struct xarecover_st *) arg;
+  int got;
+  
+  if (hton->state == SHOW_OPTION_YES && hton->recover)
   {
-    if ((*types)->state != SHOW_OPTION_YES || !(*types)->recover)
-      continue;
-    while ((got=(*(*types)->recover)(list, len)) > 0 )
+    while ((got= hton->recover(info->list, info->len)) > 0 )
     {
       sql_print_information("Found %d prepared transaction(s) in %s",
-                            got, (*types)->name);
+                            got, hton->name);
       for (int i=0; i < got; i ++)
       {
-        my_xid x=list[i].get_my_xid();
+        my_xid x=info->list[i].get_my_xid();
         if (!x) // not "mine" - that is generated by external TM
         {
 #ifndef DBUG_OFF
           char buf[XIDDATASIZE*4+6]; // see xid_to_str
-          sql_print_information("ignore xid %s", xid_to_str(buf, list+i));
+          sql_print_information("ignore xid %s", xid_to_str(buf, info->list+i));
 #endif
-          xid_cache_insert(list+i, XA_PREPARED);
-          found_foreign_xids++;
+          xid_cache_insert(info->list+i, XA_PREPARED);
+          info->found_foreign_xids++;
           continue;
         }
-        if (dry_run)
+        if (info->dry_run)
         {
-          found_my_xids++;
+          info->found_my_xids++;
           continue;
         }
         // recovery mode
-        if (commit_list ?
-            hash_search(commit_list, (byte *)&x, sizeof(x)) != 0 :
+        if (info->commit_list ?
+            hash_search(info->commit_list, (byte *)&x, sizeof(x)) != 0 :
             tc_heuristic_recover == TC_HEURISTIC_RECOVER_COMMIT)
         {
 #ifndef DBUG_OFF
           char buf[XIDDATASIZE*4+6]; // see xid_to_str
-          sql_print_information("commit xid %s", xid_to_str(buf, list+i));
+          sql_print_information("commit xid %s", xid_to_str(buf, info->list+i));
 #endif
-          (*(*types)->commit_by_xid)(list+i);
+          hton->commit_by_xid(info->list+i);
         }
         else
         {
 #ifndef DBUG_OFF
           char buf[XIDDATASIZE*4+6]; // see xid_to_str
-          sql_print_information("rollback xid %s", xid_to_str(buf, list+i));
+          sql_print_information("rollback xid %s",
+                                xid_to_str(buf, info->list+i));
 #endif
-          (*(*types)->rollback_by_xid)(list+i);
+          hton->rollback_by_xid(info->list+i);
         }
       }
-      if (got < len)
+      if (got < info->len)
         break;
     }
   }
-  my_free((gptr)list, MYF(0));
-  if (found_foreign_xids)
-    sql_print_warning("Found %d prepared XA transactions", found_foreign_xids);
-  if (dry_run && found_my_xids)
+  return FALSE;
+}
+
+int ha_recover(HASH *commit_list)
+{
+  struct xarecover_st info;
+  DBUG_ENTER("ha_recover");
+  info.found_foreign_xids= info.found_my_xids= 0;
+  info.commit_list= commit_list;
+  info.dry_run= (info.commit_list==0 && tc_heuristic_recover==0);
+  info.list= NULL;
+
+  /* commit_list and tc_heuristic_recover cannot be set both */
+  DBUG_ASSERT(info.commit_list==0 || tc_heuristic_recover==0);
+  /* if either is set, total_ha_2pc must be set too */
+  DBUG_ASSERT(info.dry_run || total_ha_2pc>(ulong)opt_bin_log);
+
+  if (total_ha_2pc <= (ulong)opt_bin_log)
+    DBUG_RETURN(0);
+
+  if (info.commit_list)
+    sql_print_information("Starting crash recovery...");
+
+#ifndef WILL_BE_DELETED_LATER
+  /*
+    for now, only InnoDB supports 2pc. It means we can always safely
+    rollback all pending transactions, without risking inconsistent data
+  */
+  DBUG_ASSERT(total_ha_2pc == (ulong) opt_bin_log+1); // only InnoDB and binlog
+  tc_heuristic_recover= TC_HEURISTIC_RECOVER_ROLLBACK; // forcing ROLLBACK
+  info.dry_run=FALSE;
+#endif
+
+  for (info.len= MAX_XID_LIST_SIZE ; 
+       info.list==0 && info.len > MIN_XID_LIST_SIZE; info.len/=2)
+  {
+    info.list=(XID *)my_malloc(info.len*sizeof(XID), MYF(0));
+  }
+  if (!info.list)
+  {
+    sql_print_error(ER(ER_OUTOFMEMORY), info.len*sizeof(XID));
+    DBUG_RETURN(1);
+  }
+
+  plugin_foreach(NULL, xarecover_handlerton, 
+                 MYSQL_STORAGE_ENGINE_PLUGIN, &info);
+
+  my_free((gptr)info.list, MYF(0));
+  if (info.found_foreign_xids)
+    sql_print_warning("Found %d prepared XA transactions", 
+                      info.found_foreign_xids);
+  if (info.dry_run && info.found_my_xids)
   {
     sql_print_error("Found %d prepared transactions! It means that mysqld was "
                     "not shut down properly last time and critical recovery "
@@ -919,10 +1112,10 @@ int ha_recover(HASH *commit_list)
                     "after a crash. You have to start mysqld with "
                     "--tc-heuristic-recover switch to commit or rollback "
                     "pending transactions.",
-                    found_my_xids, opt_tc_log_file);
+                    info.found_my_xids, opt_tc_log_file);
     DBUG_RETURN(1);
   }
-  if (commit_list)
+  if (info.commit_list)
     sql_print_information("Crash recovery finished.");
   DBUG_RETURN(0);
 }
@@ -995,32 +1188,17 @@ bool mysql_xa_recover(THD *thd)
 
 int ha_release_temporary_latches(THD *thd)
 {
-  handlerton **types;
-
-  for (types= sys_table_types; *types; types++)
-  {
-    if ((*types)->state == SHOW_OPTION_YES &&
-        (*types)->release_temporary_latches)
-      (*types)->release_temporary_latches(thd);
-  }
-  return 0;
+#ifdef WITH_INNOBASE_STORAGE_ENGINE
+  innobase_release_temporary_latches(thd);
+#endif
 }
 
 
-/* 
-  Export statistics for different engines. Currently we use it only for
-  InnoDB.
-*/
-
 int ha_update_statistics()
 {
-  handlerton **types;
-
-  for (types= sys_table_types; *types; types++)
-  {
-    if ((*types)->state == SHOW_OPTION_YES && (*types)->update_statistics)
-      (*types)->update_statistics();
-  }
+#ifdef WITH_INNOBASE_STORAGE_ENGINE
+  innodb_export_status();
+#endif
   return 0;
 }
 
@@ -1129,20 +1307,25 @@ int ha_release_savepoint(THD *thd, SAVEPOINT *sv)
 }
 
 
+static my_bool snapshot_handlerton(THD *thd, st_plugin_int *plugin,
+                                   void *arg)
+{
+  handlerton *hton= (handlerton *) plugin->plugin->info;
+  if (hton->state == SHOW_OPTION_YES &&
+      hton->start_consistent_snapshot)
+  {
+    hton->start_consistent_snapshot(thd);
+    *((bool *)arg)= false;
+  }
+  return FALSE;
+}
+
 int ha_start_consistent_snapshot(THD *thd)
 {
   bool warn= true;
-  handlerton **types;
 
-  for (types= sys_table_types; *types; types++)
-  {
-    if ((*types)->state == SHOW_OPTION_YES &&
-        (*types)->start_consistent_snapshot)
-    {
-      (*types)->start_consistent_snapshot(thd);
-      warn= false; /* hope user is using engine */
-    }
-  }
+  plugin_foreach(thd, snapshot_handlerton, MYSQL_STORAGE_ENGINE_PLUGIN, &warn);
+
   /*
     Same idea as when one wants to CREATE TABLE in one engine which does not
     exist:
@@ -1155,22 +1338,31 @@ int ha_start_consistent_snapshot(THD *thd)
 }
 
 
-bool ha_flush_logs(enum db_type db_type)
+static my_bool flush_handlerton(THD *thd, st_plugin_int *plugin,
+                                void *arg)
 {
-  bool result=0;
-  handlerton **types;
+  handlerton *hton= (handlerton *) plugin->plugin->info;
+  if (hton->state == SHOW_OPTION_YES && hton->flush_logs && hton->flush_logs())
+    return TRUE;
+  return FALSE;
+}
+
 
-  for (types= sys_table_types; *types; types++)
+bool ha_flush_logs(handlerton *db_type)
+{
+  if (db_type == NULL)
   {
-    if ((*types)->state == SHOW_OPTION_YES && 
-        (db_type == DB_TYPE_DEFAULT || db_type == (*types)->db_type) &&
-        (*types)->flush_logs)
-    {
-      if ((*types)->flush_logs())
-        result= 1;
-    }
+    if (plugin_foreach(NULL, flush_handlerton,
+                          MYSQL_STORAGE_ENGINE_PLUGIN, 0))
+      return TRUE;
   }
-  return result;
+  else
+  {
+    if (db_type->state != SHOW_OPTION_YES ||
+        (db_type->flush_logs && db_type->flush_logs()))
+      return TRUE;
+  }
+  return FALSE;
 }
 
 /*
@@ -1178,7 +1370,7 @@ bool ha_flush_logs(enum db_type db_type)
   The .frm file will be deleted only if we return 0 or ENOENT
 */
 
-int ha_delete_table(THD *thd, enum db_type table_type, const char *path,
+int ha_delete_table(THD *thd, handlerton *table_type, const char *path,
                     const char *db, const char *alias, bool generate_warning)
 {
   handler *file;
@@ -1193,7 +1385,7 @@ int ha_delete_table(THD *thd, enum db_type table_type, const char *path,
   dummy_table.s= &dummy_share;
 
   /* DB_TYPE_UNKNOWN is used in ALTER TABLE when renaming only .frm files */
-  if (table_type == DB_TYPE_UNKNOWN ||
+  if (table_type == NULL ||
       ! (file=get_new_handler(&dummy_share, thd->mem_root, table_type)))
     DBUG_RETURN(ENOENT);
 
@@ -2486,40 +2678,50 @@ int handler::index_read_idx(byte * buf, uint index, const byte * key,
     pointer		pointer to TYPELIB structure
 */
 
+static my_bool exts_handlerton(THD *unused, st_plugin_int *plugin,
+                               void *arg)
+{
+  List<char> *found_exts= (List<char> *) arg;
+  handlerton *hton= (handlerton *) plugin->plugin->info;
+  handler *file;
+  if (hton->state == SHOW_OPTION_YES && hton->create &&
+      (file= hton->create((TABLE_SHARE*) 0)))
+  {
+    List_iterator_fast<char> it(*found_exts);
+    const char **ext, *old_ext;
+    
+    for (ext= file->bas_ext(); *ext; ext++)
+    {
+      while ((old_ext= it++))
+      {
+        if (!strcmp(old_ext, *ext))
+	  break;
+      }
+      if (!old_ext)
+        found_exts->push_back((char *) *ext);
+
+      it.rewind();
+    }
+    delete file;
+  }
+  return FALSE;
+}
+
 TYPELIB *ha_known_exts(void)
 {
   MEM_ROOT *mem_root= current_thd->mem_root;
   if (!known_extensions.type_names || mysys_usage_id != known_extensions_id)
   {
-    handlerton **types;
     List<char> found_exts;
-    List_iterator_fast<char> it(found_exts);
     const char **ext, *old_ext;
 
     known_extensions_id= mysys_usage_id;
     found_exts.push_back((char*) triggers_file_ext);
     found_exts.push_back((char*) trigname_file_ext);
-    for (types= sys_table_types; *types; types++)
-    {
-      if ((*types)->state == SHOW_OPTION_YES)
-      {
-	handler *file= get_new_handler((TABLE_SHARE*) 0, mem_root,
-                                       (enum db_type) (*types)->db_type);
-	for (ext= file->bas_ext(); *ext; ext++)
-	{
-	  while ((old_ext= it++))
-          {
-	    if (!strcmp(old_ext, *ext))
-	      break;
-          }
-	  if (!old_ext)
-	    found_exts.push_back((char *) *ext);
-
-	  it.rewind();
-	}
-	delete file;
-      }
-    }
+    
+    plugin_foreach(NULL, exts_handlerton, 
+                   MYSQL_STORAGE_ENGINE_PLUGIN, &found_exts);
+
     ext= (const char **) my_once_alloc(sizeof(char *)*
                                        (found_exts.elements+1),
                                        MYF(MY_WME | MY_FAE));
@@ -2528,6 +2730,7 @@ TYPELIB *ha_known_exts(void)
     known_extensions.count= found_exts.elements;
     known_extensions.type_names= ext;
 
+    List_iterator_fast<char> it(found_exts);
     while ((old_ext= it++))
       *ext++= old_ext;
     *ext= 0;
@@ -2535,24 +2738,37 @@ TYPELIB *ha_known_exts(void)
   return &known_extensions;
 }
 
-static bool stat_print(THD *thd, const char *type, const char *file,
-                       const char *status)
+static bool stat_print(THD *thd, const char *type, uint type_len,
+                       const char *file, uint file_len,
+                       const char *status, uint status_len)
 {
   Protocol *protocol= thd->protocol;
   protocol->prepare_for_resend();
-  protocol->store(type, system_charset_info);
-  protocol->store(file, system_charset_info);
-  protocol->store(status, system_charset_info);
+  protocol->store(type, type_len, system_charset_info);
+  protocol->store(file, file_len, system_charset_info);
+  protocol->store(status, status_len, system_charset_info);
   if (protocol->write())
     return TRUE;
   return FALSE;
 }
 
-bool ha_show_status(THD *thd, enum db_type db_type, enum ha_stat_type stat)
+
+static my_bool showstat_handlerton(THD *thd, st_plugin_int *plugin,
+                                   void *arg)
+{
+  enum ha_stat_type stat= *(enum ha_stat_type *) arg;
+  handlerton *hton= (handlerton *) plugin->plugin->info;
+  if (hton->state == SHOW_OPTION_YES && hton->show_status &&
+      hton->show_status(thd, stat_print, stat))
+    return TRUE;
+  return FALSE;
+}
+
+bool ha_show_status(THD *thd, handlerton *db_type, enum ha_stat_type stat)
 {
-  handlerton **types;
   List<Item> field_list;
   Protocol *protocol= thd->protocol;
+  bool result;
 
   field_list.push_back(new Item_empty_string("Type",10));
   field_list.push_back(new Item_empty_string("Name",FN_REFLEN));
@@ -2562,25 +2778,24 @@ bool ha_show_status(THD *thd, enum db_type db_type, enum ha_stat_type stat)
                             Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF))
     return TRUE;
 
-  for (types= sys_table_types; *types; types++)
+  if (db_type == NULL)
   {
-    if ((*types)->state == SHOW_OPTION_YES &&
-        (db_type == DB_TYPE_DEFAULT || db_type == (*types)->db_type) &&
-        (*types)->show_status)
-    {
-      if ((*types)->show_status(thd, stat_print, stat))
-        return TRUE;
-    }
-    else if (db_type == (*types)->db_type &&
-             (*types)->state != SHOW_OPTION_YES)
-    {
-      if (stat_print(thd, (*types)->name, "", "DISABLED"))
-	return TRUE;
-    }
+    result= plugin_foreach(thd, showstat_handlerton, 
+                           MYSQL_STORAGE_ENGINE_PLUGIN, &stat);
+  }
+  else
+  {
+    if (db_type->state != SHOW_OPTION_YES)
+      result= stat_print(thd, db_type->name, strlen(db_type->name), 
+                         "", 0, "DISABLED", 8) ? 1 : 0;
+    else
+      result= db_type->show_status && 
+              db_type->show_status(thd, stat_print, stat) ? 1 : 0;
   }
 
-  send_eof(thd);
-  return FALSE;
+  if (!result)
+    send_eof(thd);
+  return result;
 }
 
 
@@ -2606,19 +2821,10 @@ bool ha_show_status(THD *thd, enum db_type db_type, enum ha_stat_type stat)
 int ha_repl_report_sent_binlog(THD *thd, char *log_file_name,
                                my_off_t end_offset)
 {
-  int result= 0;
-  handlerton **types;
-
-  for (types= sys_table_types; *types; types++)
-  {
-    if ((*types)->state == SHOW_OPTION_YES &&
-        (*types)->repl_report_sent_binlog)
-    {
-      (*types)->repl_report_sent_binlog(thd,log_file_name,end_offset);
-      result= 0;
-    }
-  }
-  return result;
+#ifdef WITH_INNOBASE_STORAGE_ENGINE
+  innobase_repl_report_sent_binlog(thd, log_file_name, end_offset);
+#endif
+  return 0;
 }
 
 
@@ -2643,3 +2849,4 @@ int ha_repl_report_replication_stop(THD *thd)
   return 0;
 }
 #endif /* HAVE_REPLICATION */
+
diff --git a/sql/handler.h b/sql/handler.h
index b4b90cbfaa8..9519ba06390 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -178,7 +178,7 @@
 /* Options of START TRANSACTION statement (and later of SET TRANSACTION stmt) */
 #define MYSQL_START_TRANS_OPT_WITH_CONS_SNAPSHOT 1
 
-enum db_type
+enum legacy_db_type
 {
   DB_TYPE_UNKNOWN=0,DB_TYPE_DIAB_ISAM=1,
   DB_TYPE_HASH,DB_TYPE_MISAM,DB_TYPE_PISAM,
@@ -191,7 +191,7 @@ enum db_type
   DB_TYPE_BLACKHOLE_DB,
   DB_TYPE_PARTITION_DB,
   DB_TYPE_BINLOG,
-  DB_TYPE_DEFAULT // Must be last
+  DB_TYPE_DEFAULT=127 // Must be last
 };
 
 enum row_type { ROW_TYPE_NOT_USED=-1, ROW_TYPE_DEFAULT, ROW_TYPE_FIXED,
@@ -315,8 +315,9 @@ typedef struct st_table TABLE;
 typedef struct st_table_share TABLE_SHARE;
 struct st_foreign_key_info;
 typedef struct st_foreign_key_info FOREIGN_KEY_INFO;
-typedef bool (stat_print_fn)(THD *thd, const char *type, const char *file,
-                             const char *status);
+typedef bool (stat_print_fn)(THD *thd, const char *type, uint type_len,
+                             const char *file, uint file_len,
+                             const char *status, uint status_len);
 enum ha_stat_type { HA_ENGINE_STATUS, HA_ENGINE_LOGS, HA_ENGINE_MUTEX };
 
 /*
@@ -333,6 +334,13 @@ enum ha_stat_type { HA_ENGINE_STATUS, HA_ENGINE_LOGS, HA_ENGINE_MUTEX };
 typedef struct
 {
   /*
+    handlerton structure version
+   */
+  const int interface_version;
+#define MYSQL_HANDLERTON_INTERFACE_VERSION 0x00000000
+
+
+  /*
     storage engine name as it should be printed to a user
   */
   const char *name;
@@ -351,7 +359,7 @@ typedef struct
     Historical number used for frm file to determine the correct storage engine.
     This is going away and new engines will just use "name" for this.
   */
-  enum db_type db_type;
+  enum legacy_db_type db_type;
   /* 
     Method that initizlizes a storage engine
   */
@@ -416,19 +424,17 @@ typedef struct
    handler *(*create)(TABLE_SHARE *table);
    void (*drop_database)(char* path);
    int (*panic)(enum ha_panic_function flag);
-   int (*release_temporary_latches)(THD *thd);
-   int (*update_statistics)();
    int (*start_consistent_snapshot)(THD *thd);
    bool (*flush_logs)();
    bool (*show_status)(THD *thd, stat_print_fn *print, enum ha_stat_type stat);
-   int (*repl_report_sent_binlog)(THD *thd, char *log_file_name,
-                                  my_off_t end_offset);
    uint32 flags;                                /* global handler flags */
 } handlerton;
 
+extern const handlerton default_hton;
+
 struct show_table_alias_st {
   const char *alias;
-  const char *type;
+  enum legacy_db_type type;
 };
 
 /* Possible flags of a handlerton */
@@ -496,7 +502,7 @@ public:
   char* part_comment;
   char* data_file_name;
   char* index_file_name;
-  enum db_type engine_type;
+  handlerton *engine_type;
   enum partition_state part_state;
   uint16 nodegroup_id;
   
@@ -504,7 +510,7 @@ public:
   : part_max_rows(0), part_min_rows(0), partition_name(NULL),
     tablespace_name(NULL), range_value(0), part_comment(NULL),
     data_file_name(NULL), index_file_name(NULL),
-    engine_type(DB_TYPE_UNKNOWN),part_state(PART_NORMAL),
+    engine_type(NULL),part_state(PART_NORMAL),
     nodegroup_id(UNDEF_NODEGROUP)
   {
     subpartitions.empty();
@@ -567,7 +573,7 @@ public:
   key_map all_fields_in_PF, all_fields_in_PPF, all_fields_in_SPF;
   key_map some_fields_in_PF;
 
-  enum db_type default_engine_type;
+  handlerton *default_engine_type;
   Item_result part_result_type;
   partition_type part_type;
   partition_type subpart_type;
@@ -608,7 +614,7 @@ public:
     part_info_string(NULL),
     part_func_string(NULL), subpart_func_string(NULL),
     curr_part_elem(NULL), current_partition(NULL),
-    default_engine_type(DB_TYPE_UNKNOWN),
+    default_engine_type(NULL),
     part_result_type(INT_RESULT),
     part_type(NOT_A_PARTITION), subpart_type(NOT_A_PARTITION),
     part_info_len(0), part_func_len(0), subpart_func_len(0),
@@ -683,7 +689,7 @@ typedef struct st_ha_create_information
   ulong raid_chunksize;
   ulong used_fields;
   SQL_LIST merge_list;
-  enum db_type db_type;
+  handlerton *db_type;
   enum row_type row_type;
   uint null_bits;                       /* NULL bits at start of record */
   uint options;				/* OR of HA_CREATE_ options */
@@ -727,7 +733,7 @@ int get_parts_for_update(const byte *old_data, byte *new_data,
                          uint32 *old_part_id, uint32 *new_part_id);
 int get_part_for_delete(const byte *buf, const byte *rec0,
                         partition_info *part_info, uint32 *part_id);
-bool check_partition_info(partition_info *part_info,enum db_type eng_type,
+bool check_partition_info(partition_info *part_info,handlerton *eng_type,
                           handler *file, ulonglong max_rows);
 bool fix_partition_func(THD *thd, const char *name, TABLE *table);
 char *generate_partition_syntax(partition_info *part_info,
@@ -743,7 +749,7 @@ void get_full_part_id_from_key(const TABLE *table, byte *buf,
                                part_id_range *part_spec);
 bool mysql_unpack_partition(THD *thd, const uchar *part_buf,
                             uint part_info_len, TABLE *table,
-                            enum db_type default_db_type);
+                            handlerton *default_db_type);
 #endif
 
 
@@ -1413,32 +1419,56 @@ extern ulong total_ha, total_ha_2pc;
 #define ha_rollback(thd) (ha_rollback_trans((thd), TRUE))
 
 /* lookups */
-enum db_type ha_resolve_by_name(const char *name, uint namelen);
-const char *ha_get_storage_engine(enum db_type db_type);
+handlerton *ha_resolve_by_name(THD *thd, LEX_STRING *name);
+handlerton *ha_resolve_by_legacy_type(THD *thd, enum legacy_db_type db_type);
+const char *ha_get_storage_engine(enum legacy_db_type db_type);
 handler *get_new_handler(TABLE_SHARE *share, MEM_ROOT *alloc,
-                         enum db_type db_type);
-enum db_type ha_checktype(THD *thd, enum db_type database_type,
+                         handlerton *db_type);
+handlerton *ha_checktype(THD *thd, enum legacy_db_type database_type,
                           bool no_substitute, bool report_error);
-bool ha_check_storage_engine_flag(enum db_type db_type, uint32 flag);
+
+
+inline enum legacy_db_type ha_legacy_type(const handlerton *db_type)
+{
+  return (db_type == NULL) ? DB_TYPE_UNKNOWN : db_type->db_type;
+}
+
+inline const char *ha_resolve_storage_engine_name(const handlerton *db_type)
+{
+  return db_type == NULL ? "UNKNOWN" : db_type->name;
+}
+
+inline bool ha_check_storage_engine_flag(const handlerton *db_type, uint32 flag)
+{
+  return db_type == NULL ? FALSE : test(db_type->flags & flag);
+}
+
+inline bool ha_storage_engine_is_enabled(const handlerton *db_type)
+{
+  return (db_type && db_type->create) ? 
+         (db_type->state == SHOW_OPTION_YES) : FALSE;
+}
 
 /* basic stuff */
 int ha_init(void);
+int ha_register_builtin_plugins();
+int ha_initialize_handlerton(handlerton *hton);
+
 TYPELIB *ha_known_exts(void);
 int ha_panic(enum ha_panic_function flag);
 int ha_update_statistics();
 void ha_close_connection(THD* thd);
-my_bool ha_storage_engine_is_enabled(enum db_type database_type);
-bool ha_flush_logs(enum db_type db_type=DB_TYPE_DEFAULT);
+bool ha_flush_logs(handlerton *db_type);
 void ha_drop_database(char* path);
 int ha_create_table(THD *thd, const char *path,
                     const char *db, const char *table_name,
                     HA_CREATE_INFO *create_info,
 		    bool update_create_info);
-int ha_delete_table(THD *thd, enum db_type db_type, const char *path,
+int ha_delete_table(THD *thd, handlerton *db_type, const char *path,
                     const char *db, const char *alias, bool generate_warning);
 
 /* statistics and info */
-bool ha_show_status(THD *thd, enum db_type db_type, enum ha_stat_type stat);
+bool ha_show_status(THD *thd, handlerton *db_type, enum ha_stat_type stat);
 
 /* discovery */
 int ha_create_table_from_engine(THD* thd, const char *db, const char *name);
diff --git a/sql/item_sum.cc b/sql/item_sum.cc
index 03050a99ff1..9425351e682 100644
--- a/sql/item_sum.cc
+++ b/sql/item_sum.cc
@@ -2321,7 +2321,7 @@ bool Item_sum_count_distinct::setup(THD *thd)
   table->file->extra(HA_EXTRA_NO_ROWS);		// Don't update rows
   table->no_rows=1;
 
-  if (table->s->db_type == DB_TYPE_HEAP)
+  if (table->s->db_type == &heap_hton)
   {
     /*
       No blobs, otherwise it would have been MyISAM: set up a compare
diff --git a/sql/log.cc b/sql/log.cc
index 1fc7ba1f1fd..d30cf3266f9 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -48,6 +48,7 @@ static int binlog_rollback(THD *thd, bool all);
 static int binlog_prepare(THD *thd, bool all);
 
 handlerton binlog_hton = {
+  MYSQL_HANDLERTON_INTERFACE_VERSION,
   "binlog",
   SHOW_OPTION_YES,
   "This is a meta storage engine to represent the binlog in a transaction",
@@ -71,12 +72,9 @@ handlerton binlog_hton = {
   NULL,                         /* Create a new handler */
   NULL,                         /* Drop a database */
   NULL,                         /* Panic call */
-  NULL,                         /* Release temporary latches */
-  NULL,                         /* Update Statistics */
   NULL,                         /* Start Consistent Snapshot */
   NULL,                         /* Flush logs */
   NULL,                         /* Show status */
-  NULL,                         /* Replication Report Sent Binlog */
   HTON_NOT_USER_SELECTABLE | HTON_HIDDEN
 };
 
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index 4b7a1fc00a8..4a7c6e8bbcf 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -602,7 +602,7 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
 int mysql_rm_table_part2_with_lock(THD *thd, TABLE_LIST *tables,
 				   bool if_exists, bool drop_temporary,
 				   bool log_query);
-bool quick_rm_table(enum db_type base,const char *db,
+bool quick_rm_table(handlerton *base,const char *db,
                     const char *table_name);
 void close_cached_table(THD *thd, TABLE *table);
 bool mysql_rename_tables(THD *thd, TABLE_LIST *table_list);
@@ -748,7 +748,7 @@ bool mysql_recreate_table(THD *thd, TABLE_LIST *table_list, bool do_send_ok);
 bool mysql_create_like_table(THD *thd, TABLE_LIST *table,
                              HA_CREATE_INFO *create_info,
                              Table_ident *src_table);
-bool mysql_rename_table(enum db_type base,
+bool mysql_rename_table(handlerton *base,
 			const char *old_db,
 			const char * old_name,
 			const char *new_db,
@@ -1001,7 +1001,7 @@ bool open_normal_and_derived_tables(THD *thd, TABLE_LIST *tables, uint flags);
 int lock_tables(THD *thd, TABLE_LIST *tables, uint counter, bool *need_reopen);
 TABLE *open_temporary_table(THD *thd, const char *path, const char *db,
 			    const char *table_name, bool link_in_list);
-bool rm_temporary_table(enum db_type base, char *path);
+bool rm_temporary_table(handlerton *base, char *path);
 void free_io_cache(TABLE *entry);
 void intern_close_table(TABLE *entry);
 bool close_thread_table(THD *thd, TABLE **table_ptr);
@@ -1329,6 +1329,10 @@ extern handlerton partition_hton;
 extern SHOW_COMP_OPTION have_partition_db;
 #endif
 
+extern handlerton myisam_hton;
+extern handlerton myisammrg_hton;
+extern handlerton heap_hton;
+
 extern SHOW_COMP_OPTION have_isam;
 extern SHOW_COMP_OPTION have_raid, have_openssl, have_symlink;
 extern SHOW_COMP_OPTION have_query_cache;
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index f344becffde..544fe23e270 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -2622,6 +2622,18 @@ static int init_common_variables(const char *conf_file_name, int argc,
   strmake(pidfile_name, glob_hostname, sizeof(pidfile_name)-5);
   strmov(fn_ext(pidfile_name),".pid");		// Add proper extension
 
+  if (plugin_init())
+  {
+    sql_print_error("Failed to init plugins.");
+    return 1;
+  }
+
+  if (ha_register_builtin_plugins())
+  {
+    sql_print_error("Failed to register built-in storage engines.");
+    return 1;
+  }
+
   load_defaults(conf_file_name, groups, &argc, &argv);
   defaults_argv=argv;
   get_options(argc,argv);
@@ -3105,17 +3117,15 @@ server.");
   /*
     Check that the default storage engine is actually available.
   */
-  if (!ha_storage_engine_is_enabled((enum db_type)
-                                    global_system_variables.table_type))
+  if (!ha_storage_engine_is_enabled(global_system_variables.table_type))
   {
     if (!opt_bootstrap)
     {
       sql_print_error("Default storage engine (%s) is not available",
-                      ha_get_storage_engine((enum db_type)
-                                            global_system_variables.table_type));
+                      global_system_variables.table_type->name);
       unireg_abort(1);
     }
-    global_system_variables.table_type= DB_TYPE_MYISAM;
+    global_system_variables.table_type= &myisam_hton;
   }
 
   tc_log= (total_ha_2pc > 1 ? (opt_bin_log  ?
@@ -3470,7 +3480,7 @@ we force server id to 2, but this MySQL server will not act as a slave.");
 
   if (!opt_noacl)
   {
-    plugin_init();
+    plugin_load();
 #ifdef HAVE_DLOPEN
     udf_init();
 #endif
@@ -6102,6 +6112,7 @@ struct show_var_st status_vars[]= {
   {"Com_show_master_status",   (char*) offsetof(STATUS_VAR, com_stat[(uint) SQLCOM_SHOW_MASTER_STAT]), SHOW_LONG_STATUS},
   {"Com_show_new_master",      (char*) offsetof(STATUS_VAR, com_stat[(uint) SQLCOM_SHOW_NEW_MASTER]), SHOW_LONG_STATUS},
   {"Com_show_open_tables",     (char*) offsetof(STATUS_VAR, com_stat[(uint) SQLCOM_SHOW_OPEN_TABLES]), SHOW_LONG_STATUS},
+  {"Com_show_plugins",         (char*) offsetof(STATUS_VAR, com_stat[(uint) SQLCOM_SHOW_PLUGINS]), SHOW_LONG_STATUS},
   {"Com_show_privileges",      (char*) offsetof(STATUS_VAR, com_stat[(uint) SQLCOM_SHOW_PRIVILEGES]), SHOW_LONG_STATUS},
   {"Com_show_processlist",     (char*) offsetof(STATUS_VAR, com_stat[(uint) SQLCOM_SHOW_PROCESSLIST]), SHOW_LONG_STATUS},
   {"Com_show_slave_hosts",     (char*) offsetof(STATUS_VAR, com_stat[(uint) SQLCOM_SHOW_SLAVE_HOSTS]), SHOW_LONG_STATUS},
@@ -6412,7 +6423,7 @@ static void mysql_init_variables(void)
 
 
   /* Set default values for some option variables */
-  global_system_variables.table_type=   DB_TYPE_MYISAM;
+  global_system_variables.table_type= &myisam_hton;
   global_system_variables.tx_isolation= ISO_REPEATABLE_READ;
   global_system_variables.select_limit= (ulonglong) HA_POS_ERROR;
   max_system_variables.select_limit=    (ulonglong) HA_POS_ERROR;
@@ -6805,9 +6816,9 @@ get_one_option(int optid, const struct my_option *opt __attribute__((unused)),
     break;
   case OPT_STORAGE_ENGINE:
   {
-    if ((enum db_type)((global_system_variables.table_type=
-                        ha_resolve_by_name(argument, strlen(argument)))) ==
-        DB_TYPE_UNKNOWN)
+    LEX_STRING name= { argument, strlen(argument) };
+    if ((global_system_variables.table_type=
+                        ha_resolve_by_name(current_thd, &name)) == NULL)
     {
       fprintf(stderr,"Unknown/unsupported table type: %s\n",argument);
       exit(1);
diff --git a/sql/set_var.cc b/sql/set_var.cc
index 19dc2265347..735dc82dbed 100644
--- a/sql/set_var.cc
+++ b/sql/set_var.cc
@@ -3034,11 +3034,12 @@ bool sys_var_thd_storage_engine::check(THD *thd, set_var *var)
 
   if (var->value->result_type() == STRING_RESULT)
   {
-    enum db_type db_type;
+    LEX_STRING name;
+    handlerton *db_type;
     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(thd, db_type, 1, 0) != db_type)
+        !(name.str= (char *)res->ptr()) || !(name.length= res->length()) ||
+	!(var->save_result.hton= db_type= ha_resolve_by_name(thd, &name)) ||
+        ha_checktype(thd, ha_legacy_type(db_type), 1, 0) != db_type)
     {
       value= res ? res->c_ptr() : "NULL";
       goto err;
@@ -3056,29 +3057,28 @@ err:
 byte *sys_var_thd_storage_engine::value_ptr(THD *thd, enum_var_type type,
 					    LEX_STRING *base)
 {
-  ulong val;
-  val= ((type == OPT_GLOBAL) ? global_system_variables.*offset :
-        thd->variables.*offset);
-  const char *table_type= ha_get_storage_engine((enum db_type)val);
-  return (byte *) table_type;
+  handlerton *val;
+  val= (type == OPT_GLOBAL) ? global_system_variables.*offset :
+        thd->variables.*offset;
+  return (byte *) val->name;
 }
 
 
 void sys_var_thd_storage_engine::set_default(THD *thd, enum_var_type type)
 {
   if (type == OPT_GLOBAL)
-    global_system_variables.*offset= (ulong) DB_TYPE_MYISAM;
+    global_system_variables.*offset= &myisam_hton;
   else
-    thd->variables.*offset= (ulong) (global_system_variables.*offset);
+    thd->variables.*offset= global_system_variables.*offset;
 }
 
 
 bool sys_var_thd_storage_engine::update(THD *thd, set_var *var)
 {
-  if (var->type == OPT_GLOBAL)
-    global_system_variables.*offset= var->save_result.ulong_value;
-  else
-    thd->variables.*offset= var->save_result.ulong_value;
+  handlerton **value= &(global_system_variables.*offset);
+  if (var->type != OPT_GLOBAL)
+    value= &(thd->variables.*offset);
+  *value= var->save_result.hton;
   return 0;
 }
 
diff --git a/sql/set_var.h b/sql/set_var.h
index 14059f7e9b7..838037b6e20 100644
--- a/sql/set_var.h
+++ b/sql/set_var.h
@@ -379,9 +379,9 @@ public:
 class sys_var_thd_storage_engine :public sys_var_thd
 {
 protected:
-  ulong SV::*offset;
+  handlerton *SV::*offset;
 public:
-  sys_var_thd_storage_engine(const char *name_arg, ulong SV::*offset_arg)
+  sys_var_thd_storage_engine(const char *name_arg, handlerton *SV::*offset_arg)
     :sys_var_thd(name_arg), offset(offset_arg)
   {}
   bool check(THD *thd, set_var *var);
@@ -398,7 +398,7 @@ SHOW_TYPE type() { return SHOW_CHAR; }
 class sys_var_thd_table_type :public sys_var_thd_storage_engine
 {
 public:
-  sys_var_thd_table_type(const char *name_arg, ulong SV::*offset_arg)
+  sys_var_thd_table_type(const char *name_arg, handlerton *SV::*offset_arg)
     :sys_var_thd_storage_engine(name_arg, offset_arg)
   {}
   void warn_deprecated(THD *thd);
@@ -812,6 +812,7 @@ public:
     CHARSET_INFO *charset;
     ulong ulong_value;
     ulonglong ulonglong_value;
+    handlerton *hton;
     DATE_TIME_FORMAT *date_time_format;
     Time_zone *time_zone;
   } save_result;
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 4ced556fd11..5baf7cbd80b 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -1433,7 +1433,7 @@ void close_temporary_table(THD *thd, TABLE *table,
 
 void close_temporary(TABLE *table, bool free_share, bool delete_table)
 {
-  db_type table_type= table->s->db_type;
+  handlerton *table_type= table->s->db_type;
   DBUG_ENTER("close_temporary");
 
   free_io_cache(table);
@@ -1802,7 +1802,7 @@ TABLE *open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
     */
     {
       char path[FN_REFLEN];
-      db_type not_used;
+      enum legacy_db_type not_used;
       strxnmov(path, FN_REFLEN-1, mysql_data_home, "/", table_list->db, "/",
                table_list->table_name, reg_ext, NullS);
       (void) unpack_filename(path, path);
@@ -3278,7 +3278,7 @@ TABLE *open_temporary_table(THD *thd, const char *path, const char *db,
 }
 
 
-bool rm_temporary_table(enum db_type base, char *path)
+bool rm_temporary_table(handlerton *base, char *path)
 {
   bool error=0;
   handler *file;
diff --git a/sql/sql_cache.cc b/sql/sql_cache.cc
index 0f45b348d11..9a8e1d26ebd 100644
--- a/sql/sql_cache.cc
+++ b/sql/sql_cache.cc
@@ -2232,7 +2232,7 @@ Query_cache::register_tables_from_list(TABLE_LIST *tables_used,
                         tables_used->engine_data))
         DBUG_RETURN(0);
 
-      if (tables_used->table->s->db_type == DB_TYPE_MRG_MYISAM)
+      if (tables_used->table->s->db_type == &myisammrg_hton)
       {
         ha_myisammrg *handler = (ha_myisammrg *) tables_used->table->file;
         MYRG_INFO *file = handler->myrg_info();
@@ -2860,7 +2860,7 @@ static TABLE_COUNTER_TYPE process_and_count_tables(TABLE_LIST *tables_used,
                     "other non-cacheable table(s)"));
         DBUG_RETURN(0);
       }
-      if (tables_used->table->s->db_type == DB_TYPE_MRG_MYISAM)
+      if (tables_used->table->s->db_type == &myisammrg_hton)
       {
         ha_myisammrg *handler = (ha_myisammrg *)tables_used->table->file;
         MYRG_INFO *file = handler->myrg_info();
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 3ac95bc1ad8..84670e97b19 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -527,7 +527,7 @@ struct system_variables
   ulong read_rnd_buff_size;
   ulong div_precincrement;
   ulong sortbuff_size;
-  ulong table_type;
+  handlerton *table_type;
   ulong tmp_table_size;
   ulong tx_isolation;
   ulong completion_type;
diff --git a/sql/sql_delete.cc b/sql/sql_delete.cc
index d5543ac4e06..c58aa9d9e4f 100644
--- a/sql/sql_delete.cc
+++ b/sql/sql_delete.cc
@@ -822,7 +822,7 @@ bool mysql_truncate(THD *thd, TABLE_LIST *table_list, bool dont_send_ok)
   /* If it is a temporary table, close and regenerate it */
   if (!dont_send_ok && (table= find_temporary_table(thd, table_list)))
   {
-    db_type table_type= table->s->db_type;
+    handlerton *table_type= table->s->db_type;
     TABLE_SHARE *share= table->s;
     if (!ha_check_storage_engine_flag(table_type, HTON_CAN_RECREATE))
       goto trunc_by_del;
@@ -852,7 +852,7 @@ bool mysql_truncate(THD *thd, TABLE_LIST *table_list, bool dont_send_ok)
 
   if (!dont_send_ok)
   {
-    db_type table_type;
+    enum legacy_db_type table_type;
     mysql_frm_type(thd, path, &table_type);
     if (table_type == DB_TYPE_UNKNOWN)
     {
@@ -860,7 +860,8 @@ bool mysql_truncate(THD *thd, TABLE_LIST *table_list, bool dont_send_ok)
                table_list->db, table_list->table_name);
       DBUG_RETURN(TRUE);
     }
-    if (!ha_check_storage_engine_flag(table_type, HTON_CAN_RECREATE)
+    if (!ha_check_storage_engine_flag(ha_resolve_by_legacy_type(thd, table_type),
+                                      HTON_CAN_RECREATE)
         || thd->lex->sphead)
       goto trunc_by_del;
     if (lock_and_wait_for_table_name(thd, table_list))
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 85c8fad40d4..d23ad54b3f1 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -2585,7 +2585,7 @@ void select_create::abort()
   if (table)
   {
     table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
-    enum db_type table_type=table->s->db_type;
+    handlerton *table_type=table->s->db_type;
     if (!table->s->tmp_table)
     {
       ulong version= table->s->version;
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index d7ad28a95f5..f2618dd8bce 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -93,7 +93,7 @@ enum enum_sql_command {
   SQLCOM_XA_COMMIT, SQLCOM_XA_ROLLBACK, SQLCOM_XA_RECOVER,
   SQLCOM_SHOW_PROC_CODE, SQLCOM_SHOW_FUNC_CODE,
   SQLCOM_INSTALL_PLUGIN, SQLCOM_UNINSTALL_PLUGIN,
-  SQLCOM_SHOW_AUTHORS,
+  SQLCOM_SHOW_AUTHORS, SQLCOM_SHOW_PLUGINS,
   /* This should be the last !!! */
 
   SQLCOM_END
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 2e2a6f4d1b1..8dbce3cb78e 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -6788,7 +6788,7 @@ bool reload_acl_and_cache(THD *thd, ulong options, TABLE_LIST *tables,
     rotate_relay_log(active_mi);
     pthread_mutex_unlock(&LOCK_active_mi);
 #endif
-    if (ha_flush_logs())
+    if (ha_flush_logs(NULL))
       result=1;
     if (flush_error_log())
       result=1;
@@ -7104,7 +7104,7 @@ bool mysql_create_index(THD *thd, TABLE_LIST *table_list, List<Key> &keys)
   HA_CREATE_INFO create_info;
   DBUG_ENTER("mysql_create_index");
   bzero((char*) &create_info,sizeof(create_info));
-  create_info.db_type=DB_TYPE_DEFAULT;
+  create_info.db_type= (handlerton*) &default_hton;
   create_info.default_table_charset= thd->variables.collation_database;
   DBUG_RETURN(mysql_alter_table(thd,table_list->db,table_list->table_name,
 				&create_info, table_list,
@@ -7120,7 +7120,7 @@ bool mysql_drop_index(THD *thd, TABLE_LIST *table_list, ALTER_INFO *alter_info)
   HA_CREATE_INFO create_info;
   DBUG_ENTER("mysql_drop_index");
   bzero((char*) &create_info,sizeof(create_info));
-  create_info.db_type=DB_TYPE_DEFAULT;
+  create_info.db_type= (handlerton*) &default_hton;
   create_info.default_table_charset= thd->variables.collation_database;
   alter_info->clear();
   alter_info->flags= ALTER_DROP_INDEX;
diff --git a/sql/sql_partition.cc b/sql/sql_partition.cc
index dd4f2d9246a..cc2eb468662 100644
--- a/sql/sql_partition.cc
+++ b/sql/sql_partition.cc
@@ -599,7 +599,7 @@ static bool set_up_default_partitions(partition_info *part_info,
     partition_element *part_elem= new partition_element();
     if (likely(part_elem != 0))
     {
-      part_elem->engine_type= DB_TYPE_UNKNOWN;
+      part_elem->engine_type= NULL;
       part_elem->partition_name= default_name;
       default_name+=MAX_PART_NAME_SIZE;
       part_info->partitions.push_back(part_elem);
@@ -671,7 +671,7 @@ static bool set_up_default_subpartitions(partition_info *part_info,
       partition_element *subpart_elem= new partition_element();
       if (likely(subpart_elem != 0))
       {
-        subpart_elem->engine_type= DB_TYPE_UNKNOWN;
+        subpart_elem->engine_type= NULL;
         subpart_elem->partition_name= name_ptr;
         name_ptr+= MAX_PART_NAME_SIZE;
         part_elem->subpartitions.push_back(subpart_elem);
@@ -731,7 +731,7 @@ bool set_up_defaults_for_partitioning(partition_info *part_info,
     FALSE                  Ok, no mixed engines
 */
 
-static bool check_engine_mix(u_char *engine_array, uint no_parts)
+static bool check_engine_mix(handlerton **engine_array, uint no_parts)
 {
   /*
     Current check verifies only that all handlers are the same.
@@ -772,10 +772,10 @@ static bool check_engine_mix(u_char *engine_array, uint no_parts)
     This code is used early in the CREATE TABLE and ALTER TABLE process.
 */
 
-bool check_partition_info(partition_info *part_info,enum db_type eng_type,
+bool check_partition_info(partition_info *part_info,handlerton *eng_type,
                           handler *file, ulonglong max_rows)
 {
-  u_char *engine_array= NULL;
+  handlerton **engine_array= NULL;
   uint part_count= 0, i, no_parts, tot_partitions;
   bool result= TRUE;
   List_iterator<partition_element> part_it(part_info->partitions);
@@ -803,7 +803,8 @@ bool check_partition_info(partition_info *part_info,enum db_type eng_type,
     my_error(ER_SAME_NAME_PARTITION, MYF(0));
     goto end;
   }
-  engine_array= (u_char*)my_malloc(tot_partitions, MYF(MY_WME));
+  engine_array= (handlerton**)my_malloc(tot_partitions * sizeof(handlerton *), 
+                                        MYF(MY_WME));
   if (unlikely(!engine_array))
     goto end;
   i= 0;
@@ -813,10 +814,10 @@ bool check_partition_info(partition_info *part_info,enum db_type eng_type,
     partition_element *part_elem= part_it++;
     if (!is_sub_partitioned(part_info))
     {
-      if (part_elem->engine_type == DB_TYPE_UNKNOWN)
+      if (part_elem->engine_type == NULL)
         part_elem->engine_type= eng_type;
-      DBUG_PRINT("info", ("engine = %u",(uint)part_elem->engine_type));
-      engine_array[part_count++]= (u_char)part_elem->engine_type;
+      DBUG_PRINT("info", ("engine = %s", part_elem->engine_type->name));
+      engine_array[part_count++]= part_elem->engine_type;
     }
     else
     {
@@ -825,10 +826,10 @@ bool check_partition_info(partition_info *part_info,enum db_type eng_type,
       do
       {
         part_elem= sub_it++;
-        if (part_elem->engine_type == DB_TYPE_UNKNOWN)
+        if (part_elem->engine_type == NULL)
           part_elem->engine_type= eng_type;
-        DBUG_PRINT("info", ("engine = %u",(uint)part_elem->engine_type));
-        engine_array[part_count++]= (u_char)part_elem->engine_type;
+        DBUG_PRINT("info", ("engine = %s", part_elem->engine_type->name));
+        engine_array[part_count++]= part_elem->engine_type;
       } while (++j < no_subparts);
     }
   } while (++i < part_info->no_parts);
@@ -1895,9 +1896,9 @@ static int add_keyword_int(File fptr, const char *keyword, longlong num)
   return err + add_space(fptr);
 }
 
-static int add_engine(File fptr, enum db_type engine_type)
+static int add_engine(File fptr, handlerton *engine_type)
 {
-  const char *engine_str= ha_get_storage_engine(engine_type);
+  const char *engine_str= engine_type->name;
   int err= add_string(fptr, "ENGINE = ");
   return err + add_string(fptr, engine_str);
   return err;
@@ -3092,7 +3093,7 @@ void get_partition_set(const TABLE *table, byte *buf, const uint index,
 
 bool mysql_unpack_partition(THD *thd, const uchar *part_buf,
                             uint part_info_len, TABLE* table,
-                            enum db_type default_db_type)
+                            handlerton *default_db_type)
 {
   Item *thd_free_list= thd->free_list;
   bool result= TRUE;
@@ -3129,7 +3130,7 @@ bool mysql_unpack_partition(THD *thd, const uchar *part_buf,
   part_info= lex.part_info;
   table->part_info= part_info;
   table->file->set_part_info(part_info);
-  if (part_info->default_engine_type == DB_TYPE_UNKNOWN)
+  if (part_info->default_engine_type == NULL)
     part_info->default_engine_type= default_db_type;
   else
   {
diff --git a/sql/sql_plugin.cc b/sql/sql_plugin.cc
index 7e78b58e6cf..591289f6ee1 100644
--- a/sql/sql_plugin.cc
+++ b/sql/sql_plugin.cc
@@ -33,7 +33,6 @@ static HASH plugin_hash[MYSQL_MAX_PLUGIN_TYPE_NUM];
 static rw_lock_t THR_LOCK_plugin;
 static bool initialized= 0;
 
-
 static struct st_plugin_dl *plugin_dl_find(LEX_STRING *dl)
 {
   uint i;
@@ -348,6 +347,43 @@ void plugin_unlock(struct st_plugin_int *plugin)
 }
 
 
+static int plugin_initialize(struct st_plugin_int *plugin)
+{
+  DBUG_ENTER("plugin_initialize");
+  
+  if (plugin->plugin->init)
+  {
+    if (plugin->plugin->init())
+    {
+      sql_print_error("Plugin '%s' init function returned error.",
+                      plugin->name.str);
+      DBUG_PRINT("warning", ("Plugin '%s' init function returned error.",
+                             plugin->name.str))
+      goto err;
+    }
+  }
+  
+  switch (plugin->plugin->type)
+  {
+  case MYSQL_STORAGE_ENGINE_PLUGIN:
+    if (ha_initialize_handlerton((handlerton*) plugin->plugin->info))
+    {
+      sql_print_error("Plugin '%s' handlerton init returned error.",
+                      plugin->name.str);
+      DBUG_PRINT("warning", ("Plugin '%s' handlerton init returned error.",
+                             plugin->name.str))
+      goto err;
+    }
+    break;
+  default:
+    break;
+  }
+
+  DBUG_RETURN(0);
+err:
+  DBUG_RETURN(1);
+}
+
 static void plugin_call_initializer(void)
 {
   uint i;
@@ -356,20 +392,13 @@ static void plugin_call_initializer(void)
   {
     struct st_plugin_int *tmp= dynamic_element(&plugin_array, i,
                                                struct st_plugin_int *);
-    if (tmp->state == PLUGIN_IS_UNINITIALIZED && tmp->plugin->init)
+    if (tmp->state == PLUGIN_IS_UNINITIALIZED)
     {
-      DBUG_PRINT("info", ("Initializing plugin: '%s'", tmp->name.str));
-      if (tmp->plugin->init())
-      {
-        sql_print_error("Plugin '%s' init function returned error.",
-                        tmp->name.str);
-        DBUG_PRINT("warning", ("Plugin '%s' init function returned error.",
-                               tmp->name.str))
+      if (plugin_initialize(tmp))
         plugin_del(&tmp->name);
-      }
+      else
+        tmp->state= PLUGIN_IS_READY;
     }
-    if (tmp->state == PLUGIN_IS_UNINITIALIZED)
-      tmp->state= PLUGIN_IS_READY;
   }
   DBUG_VOID_RETURN;
 }
@@ -410,42 +439,84 @@ static byte *get_hash_key(const byte *buff, uint *length,
 }
 
 
-void plugin_init(void)
+int plugin_init(void)
+{
+  int i;
+  DBUG_ENTER("plugin_init");
+
+  if (initialized)
+    DBUG_RETURN(0);
+
+  my_rwlock_init(&THR_LOCK_plugin, NULL);
+
+  if (my_init_dynamic_array(&plugin_dl_array,
+                            sizeof(struct st_plugin_dl),16,16) ||
+      my_init_dynamic_array(&plugin_array,
+                            sizeof(struct st_plugin_int),16,16))
+    goto err;
+
+  for (i= 0; i < MYSQL_MAX_PLUGIN_TYPE_NUM; i++)
+  {
+    if (hash_init(&plugin_hash[i], system_charset_info, 16, 0, 0,
+                  get_hash_key, NULL, 0))
+      goto err;
+  }
+  
+  initialized= 1;
+
+  DBUG_RETURN(0);
+  
+err:
+  DBUG_RETURN(1);
+}
+
+
+my_bool plugin_register_builtin(struct st_mysql_plugin *plugin)
+{
+  struct st_plugin_int tmp;
+  DBUG_ENTER("plugin_register_builtin");
+
+  tmp.plugin= plugin;
+  tmp.name.str= (char *)plugin->name;
+  tmp.name.length= strlen(plugin->name);
+  tmp.state= PLUGIN_IS_UNINITIALIZED;
+
+  /* Cannot be unloaded */
+  tmp.ref_count= 1;
+  tmp.plugin_dl= 0;
+
+  if (insert_dynamic(&plugin_array, (gptr)&tmp))
+    DBUG_RETURN(1);
+
+  if (my_hash_insert(&plugin_hash[plugin->type],
+                     (byte*)dynamic_element(&plugin_array,
+                                            plugin_array.elements - 1,
+                                            struct st_plugin_int *)))
+    DBUG_RETURN(1);
+
+  DBUG_RETURN(0);
+}
+
+
+void plugin_load(void)
 {
   TABLE_LIST tables;
   TABLE *table;
   READ_RECORD read_record_info;
   int error, i;
   MEM_ROOT mem;
-  DBUG_ENTER("plugin_init");
-  if (initialized)
-    DBUG_VOID_RETURN;
-  my_rwlock_init(&THR_LOCK_plugin, NULL);
-  THD *new_thd = new THD;
-  if (!new_thd ||
-      my_init_dynamic_array(&plugin_dl_array,sizeof(struct st_plugin_dl),16,16) ||
-      my_init_dynamic_array(&plugin_array,sizeof(struct st_plugin_int),16,16))
+  THD *new_thd;
+  DBUG_ENTER("plugin_load");
+  
+  DBUG_ASSERT(initialized);
+  
+  if (!(new_thd= new THD))
   {
     sql_print_error("Can't allocate memory for plugin structures");
     delete new_thd;
-    delete_dynamic(&plugin_dl_array);
-    delete_dynamic(&plugin_array);
     DBUG_VOID_RETURN;
   }
-  for (i= 0; i < MYSQL_MAX_PLUGIN_TYPE_NUM; i++)
-  {
-    if (hash_init(&plugin_hash[i], system_charset_info, 16, 0, 0,
-                  get_hash_key, NULL, 0))
-    {
-      sql_print_error("Can't allocate memory for plugin structures");
-      delete new_thd;
-      delete_dynamic(&plugin_dl_array);
-      delete_dynamic(&plugin_array);
-      DBUG_VOID_RETURN;
-    }
-  }
   init_sql_alloc(&mem, 1024, 0);
-  initialized= 1;
   new_thd->thread_stack= (char*) &tables;
   new_thd->store_globals();
   new_thd->db= my_strdup("mysql", MYF(0));
@@ -458,10 +529,6 @@ void plugin_init(void)
   {
     DBUG_PRINT("error",("Can't open plugin table"));
     sql_print_error("Can't open the mysql.plugin table. Please run the mysql_install_db script to create it.");
-    delete_dynamic(&plugin_dl_array);
-    delete_dynamic(&plugin_array);
-    for (i= 0; i < MYSQL_MAX_PLUGIN_TYPE_NUM; i++)
-      hash_free(&plugin_hash[i]);
     goto end;
   }
   table= tables.table;
@@ -530,27 +597,31 @@ my_bool mysql_install_plugin(THD *thd, LEX_STRING *name, LEX_STRING *dl)
   int error;
   struct st_plugin_int *tmp;
   DBUG_ENTER("mysql_install_plugin");
+
   bzero(&tables, sizeof(tables));
   tables.db= (char *)"mysql";
   tables.table_name= tables.alias= (char *)"plugin";
   if (check_table_access(thd, INSERT_ACL, &tables, 0))
     DBUG_RETURN(TRUE);
+    
+  /* need to open before acquiring THR_LOCK_plugin or it will deadlock */
+  if (! (table = open_ltable(thd, &tables, TL_WRITE)))
+    DBUG_RETURN(TRUE);
+
   rw_wrlock(&THR_LOCK_plugin);
   if (plugin_add(name, dl, REPORT_TO_USER))
     goto err;
   tmp= plugin_find_internal(name, MYSQL_ANY_PLUGIN);
-  if (tmp->plugin->init)
+  
+  if (plugin_initialize(tmp))
   {
-    if (tmp->plugin->init())
-    {
-      my_error(ER_CANT_INITIALIZE_UDF, MYF(0), name->str,
-               "Plugin initialization function failed.");
-      goto err;
-    }
-    tmp->state= PLUGIN_IS_READY;
+    my_error(ER_CANT_INITIALIZE_UDF, MYF(0), name->str,
+             "Plugin initialization function failed.");
+    goto err;
   }
-  if (! (table = open_ltable(thd, &tables, TL_WRITE)))
-    goto deinit;
+
+  tmp->state= PLUGIN_IS_READY;
+
   restore_record(table, s->default_values);
   table->field[0]->store(name->str, name->length, system_charset_info);
   table->field[1]->store(dl->str, dl->length, files_charset_info);
@@ -560,6 +631,7 @@ my_bool mysql_install_plugin(THD *thd, LEX_STRING *name, LEX_STRING *dl)
     table->file->print_error(error, MYF(0));
     goto deinit;
   }
+  
   rw_unlock(&THR_LOCK_plugin);
   DBUG_RETURN(FALSE);
 deinit:
@@ -578,12 +650,29 @@ my_bool mysql_uninstall_plugin(THD *thd, LEX_STRING *name)
   TABLE_LIST tables;
   struct st_plugin_int *plugin;
   DBUG_ENTER("mysql_uninstall_plugin");
+
+  bzero(&tables, sizeof(tables));
+  tables.db= (char *)"mysql";
+  tables.table_name= tables.alias= (char *)"plugin";
+
+  /* need to open before acquiring THR_LOCK_plugin or it will deadlock */
+  if (! (table= open_ltable(thd, &tables, TL_WRITE)))
+    DBUG_RETURN(TRUE);
+
   rw_wrlock(&THR_LOCK_plugin);
-  if (! (plugin= plugin_find_internal(name, MYSQL_ANY_PLUGIN)))
+  if (!(plugin= plugin_find_internal(name, MYSQL_ANY_PLUGIN)))
   {
     my_error(ER_SP_DOES_NOT_EXIST, MYF(0), "PLUGIN", name->str);
     goto err;
   }
+  if (!plugin->plugin_dl)
+  {
+    push_warning(thd, MYSQL_ERROR::WARN_LEVEL_WARN, 0,
+                 "Built-in plugins cannot be deleted,.");
+    my_error(ER_SP_DOES_NOT_EXIST, MYF(0), "PLUGIN", name->str);
+    goto err;
+  }
+  
   if (plugin->ref_count)
   {
     plugin->state= PLUGIN_IS_DELETED;
@@ -596,11 +685,7 @@ my_bool mysql_uninstall_plugin(THD *thd, LEX_STRING *name)
       plugin->plugin->deinit();
     plugin_del(name);
   }
-  bzero(&tables, sizeof(tables));
-  tables.db= (char *)"mysql";
-  tables.table_name= tables.alias= (char *)"plugin";
-  if (! (table= open_ltable(thd, &tables, TL_WRITE)))
-    goto err;
+
   table->field[0]->store(name->str, name->length, system_charset_info);
   table->file->extra(HA_EXTRA_RETRIEVE_ALL_COLS);
   if (! table->file->index_read_idx(table->record[0], 0,
@@ -621,3 +706,44 @@ err:
   rw_unlock(&THR_LOCK_plugin);
   DBUG_RETURN(TRUE);
 }
+
+
+my_bool plugin_foreach(THD *thd, plugin_foreach_func *func,
+                       int type, void *arg)
+{
+  uint idx;
+  struct st_plugin_int *plugin;
+  DBUG_ENTER("mysql_uninstall_plugin");
+  rw_rdlock(&THR_LOCK_plugin);
+  
+  if (type == MYSQL_ANY_PLUGIN)
+  {
+    for (idx= 0; idx < plugin_array.elements; idx++)
+    {
+      plugin= dynamic_element(&plugin_array, idx, struct st_plugin_int *);
+      
+      /* FREED records may have garbage pointers */
+      if ((plugin->state != PLUGIN_IS_FREED) &&
+          func(thd, plugin, arg))
+        goto err;
+    }
+  }
+  else
+  {
+    HASH *hash= &plugin_hash[type];
+    for (idx= 0; idx < hash->records; idx++)
+    {
+      plugin= (struct st_plugin_int *) hash_element(hash, idx);
+      if ((plugin->state != PLUGIN_IS_FREED) &&
+          (plugin->state != PLUGIN_IS_DELETED) &&
+          func(thd, plugin, arg))
+        goto err;
+    }
+  }
+
+  rw_unlock(&THR_LOCK_plugin);
+  DBUG_RETURN(FALSE);
+err:
+  rw_unlock(&THR_LOCK_plugin);
+  DBUG_RETURN(TRUE);
+}
diff --git a/sql/sql_plugin.h b/sql/sql_plugin.h
index 8fb186b62de..24ba02367f5 100644
--- a/sql/sql_plugin.h
+++ b/sql/sql_plugin.h
@@ -53,11 +53,20 @@ struct st_plugin_int
 
 extern char *opt_plugin_dir_ptr;
 extern char opt_plugin_dir[FN_REFLEN];
-extern void plugin_init(void);
+extern int plugin_init(void);
+extern void plugin_load(void);
 extern void plugin_free(void);
 extern my_bool plugin_is_ready(LEX_STRING *name, int type);
 extern st_plugin_int *plugin_lock(LEX_STRING *name, int type);
 extern void plugin_unlock(struct st_plugin_int *plugin);
 extern my_bool mysql_install_plugin(THD *thd, LEX_STRING *name, LEX_STRING *dl);
 extern my_bool mysql_uninstall_plugin(THD *thd, LEX_STRING *name);
+
+extern my_bool plugin_register_builtin(struct st_mysql_plugin *plugin);
+
+typedef my_bool (plugin_foreach_func)(THD *thd, 
+                                      st_plugin_int *plugin,
+                                      void *arg);
+extern my_bool plugin_foreach(THD *thd, plugin_foreach_func *func,
+                              int type, void *arg);
 #endif
diff --git a/sql/sql_rename.cc b/sql/sql_rename.cc
index 80fcb973028..2c8c732fe86 100644
--- a/sql/sql_rename.cc
+++ b/sql/sql_rename.cc
@@ -134,7 +134,7 @@ rename_tables(THD *thd, TABLE_LIST *table_list, bool skip_error)
 {
   TABLE_LIST *ren_table,*new_table;
   frm_type_enum frm_type;
-  db_type table_type;
+  enum legacy_db_type table_type;
 
   DBUG_ENTER("rename_tables");
 
@@ -176,7 +176,8 @@ rename_tables(THD *thd, TABLE_LIST *table_list, bool skip_error)
         if (table_type == DB_TYPE_UNKNOWN) 
           my_error(ER_FILE_NOT_FOUND, MYF(0), name, my_errno);
         else
-          rc= mysql_rename_table(table_type, ren_table->db, old_alias,
+          rc= mysql_rename_table(ha_resolve_by_legacy_type(thd, table_type),
+                                 ren_table->db, old_alias,
                                  new_table->db, new_alias);
         break;
       }
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index c6f0ca03538..b221706d1f8 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -8589,7 +8589,7 @@ create_tmp_table(THD *thd,TMP_TABLE_PARAM *param,List<Item> &fields,
       OPTION_BIG_TABLES || (select_options & TMP_TABLE_FORCE_MYISAM))
   {
     table->file= get_new_handler(share, &table->mem_root,
-                                 share->db_type= DB_TYPE_MYISAM);
+                                 share->db_type= &myisam_hton);
     if (group &&
 	(param->group_parts > table->file->max_key_parts() ||
 	 param->group_length > table->file->max_key_length()))
@@ -8598,7 +8598,7 @@ create_tmp_table(THD *thd,TMP_TABLE_PARAM *param,List<Item> &fields,
   else
   {
     table->file= get_new_handler(share, &table->mem_root,
-                                 share->db_type= DB_TYPE_HEAP);
+                                 share->db_type= &heap_hton);
   }
   if (!table->file)
     goto err;
@@ -8728,7 +8728,7 @@ create_tmp_table(THD *thd,TMP_TABLE_PARAM *param,List<Item> &fields,
   if (thd->variables.tmp_table_size == ~(ulong) 0)		// No limit
     share->max_rows= ~(ha_rows) 0;
   else
-    share->max_rows= (((share->db_type == DB_TYPE_HEAP) ?
+    share->max_rows= (((share->db_type == &heap_hton) ?
                           min(thd->variables.tmp_table_size,
                               thd->variables.max_heap_table_size) :
                           thd->variables.tmp_table_size)/ share->reclength);
@@ -8868,7 +8868,7 @@ create_tmp_table(THD *thd,TMP_TABLE_PARAM *param,List<Item> &fields,
   if (thd->is_fatal_error)				// If end of memory
     goto err;					 /* purecov: inspected */
   share->db_record_offset= 1;
-  if (share->db_type == DB_TYPE_MYISAM)
+  if (share->db_type == &myisam_hton)
   {
     if (create_myisam_tmp_table(table,param,select_options))
       goto err;
@@ -9175,7 +9175,8 @@ bool create_myisam_from_heap(THD *thd, TABLE *table, TMP_TABLE_PARAM *param,
   int write_err;
   DBUG_ENTER("create_myisam_from_heap");
 
-  if (table->s->db_type != DB_TYPE_HEAP || error != HA_ERR_RECORD_FILE_FULL)
+  if (table->s->db_type != &heap_hton || 
+      error != HA_ERR_RECORD_FILE_FULL)
   {
     table->file->print_error(error,MYF(0));
     DBUG_RETURN(1);
@@ -9183,9 +9184,9 @@ bool create_myisam_from_heap(THD *thd, TABLE *table, TMP_TABLE_PARAM *param,
   new_table= *table;
   share= *table->s;
   new_table.s= &share;
-  new_table.s->db_type= DB_TYPE_MYISAM;
+  new_table.s->db_type= &myisam_hton;
   if (!(new_table.file= get_new_handler(&share, &new_table.mem_root,
-                                        DB_TYPE_MYISAM)))
+                                        &myisam_hton)))
     DBUG_RETURN(1);				// End of memory
 
   save_proc_info=thd->proc_info;
@@ -11615,7 +11616,7 @@ remove_duplicates(JOIN *join, TABLE *entry,List<Item> &fields, Item *having)
 
   free_io_cache(entry);				// Safety
   entry->file->info(HA_STATUS_VARIABLE);
-  if (entry->s->db_type == DB_TYPE_HEAP ||
+  if (entry->s->db_type == &heap_hton ||
       (!entry->s->blob_fields &&
        ((ALIGN_SIZE(reclength) + HASH_OVERHEAD) * entry->file->records <
 	thd->variables.sortbuff_size)))
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index eaf10a3b074..a26892654b7 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -48,6 +48,32 @@ static bool schema_table_store_record(THD *thd, TABLE *table);
 ** List all table types supported 
 ***************************************************************************/
 
+static my_bool show_handlerton(THD *thd, st_plugin_int *plugin,
+                               void *arg)
+{
+  handlerton *default_type= (handlerton *) arg;
+  Protocol *protocol= thd->protocol;
+  handlerton *hton= (handlerton *) plugin->plugin->info;
+
+  if (!(hton->flags & HTON_HIDDEN))
+  {
+    protocol->prepare_for_resend();
+    protocol->store(hton->name, system_charset_info);
+    const char *option_name= show_comp_option_name[(int) hton->state];
+
+    if (hton->state == SHOW_OPTION_YES && default_type == hton)
+      option_name= "DEFAULT";
+    protocol->store(option_name, system_charset_info);
+    protocol->store(hton->comment, system_charset_info);
+    protocol->store(hton->commit ? "YES" : "NO", system_charset_info);
+    protocol->store(hton->prepare ? "YES" : "NO", system_charset_info);
+    protocol->store(hton->savepoint_set ? "YES" : "NO", system_charset_info);
+    
+    return protocol->write() ? 1 : 0;
+  }
+  return 0;  
+}
+
 bool mysqld_show_storage_engines(THD *thd)
 {
   List<Item> field_list;
@@ -65,34 +91,126 @@ bool mysqld_show_storage_engines(THD *thd)
                             Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF))
     DBUG_RETURN(TRUE);
 
-  const char *default_type_name= 
-    ha_get_storage_engine((enum db_type)thd->variables.table_type);
+  if (plugin_foreach(thd, show_handlerton, 
+                     MYSQL_STORAGE_ENGINE_PLUGIN, thd->variables.table_type))
+    DBUG_RETURN(TRUE);
 
-  handlerton **types;
-  for (types= sys_table_types; *types; types++)
-  {
-    if (!((*types)->flags & HTON_HIDDEN))
-    {
-      protocol->prepare_for_resend();
-      protocol->store((*types)->name, system_charset_info);
-      const char *option_name= show_comp_option_name[(int) (*types)->state];
-
-      if ((*types)->state == SHOW_OPTION_YES &&
-          !my_strcasecmp(system_charset_info, default_type_name, (*types)->name))
-        option_name= "DEFAULT";
-      protocol->store(option_name, system_charset_info);
-      protocol->store((*types)->comment, system_charset_info);
-      protocol->store((*types)->commit ? "YES" : "NO", system_charset_info);
-      protocol->store((*types)->prepare ? "YES" : "NO", system_charset_info);
-      protocol->store((*types)->savepoint_set ? "YES" : "NO", system_charset_info);
-      if (protocol->write())
-        DBUG_RETURN(TRUE);
-    }
-  }
   send_eof(thd);
   DBUG_RETURN(FALSE);
 }
 
+static int make_version_string(char *buf, int buf_length, uint version)
+{
+  return my_snprintf(buf, buf_length, "%d.%d.%d",
+                     (version>>24)&0xff, (version>>16)&0xff,version&0xffff);
+}
+
+static my_bool show_plugins(THD *thd, st_plugin_int *plugin,
+                            void *arg)
+{
+  TABLE *table= (TABLE*) arg;
+  struct st_mysql_plugin *plug= plugin->plugin;
+  Protocol *protocol= thd->protocol;
+  CHARSET_INFO *cs= system_charset_info;
+  char version_buf[20];
+
+  restore_record(table, s->default_values);
+
+  table->field[0]->store(plugin->name.str, plugin->name.length, cs);
+    
+  switch (plugin->state)
+  {
+  /* case PLUGIN_IS_FREED: does not happen */
+  case PLUGIN_IS_DELETED:
+    table->field[1]->store(STRING_WITH_LEN("DELETED"), cs);
+    break;
+  case PLUGIN_IS_UNINITIALIZED:
+    table->field[1]->store(STRING_WITH_LEN("INACTIVE"), cs);
+    break;
+  case PLUGIN_IS_READY:
+    table->field[1]->store(STRING_WITH_LEN("ACTIVE"), cs);
+    break;
+  default:
+    DBUG_ASSERT(0);
+  }
+
+  switch (plug->type)
+  {
+  case MYSQL_UDF_PLUGIN:
+    table->field[2]->store(STRING_WITH_LEN("UDF"), cs);
+    break;
+  case MYSQL_STORAGE_ENGINE_PLUGIN:
+    table->field[2]->store(STRING_WITH_LEN("STORAGE"), cs);
+    break;
+  case MYSQL_FTPARSER_PLUGIN:
+    table->field[2]->store(STRING_WITH_LEN("FTPARSER"), cs);
+    break;
+  default:
+    table->field[2]->store(STRING_WITH_LEN("UNKNOWN"), cs);
+    break;
+  }
+
+  if (plug->version)
+  {
+    table->field[3]->store(version_buf,
+          make_version_string(version_buf, sizeof(version_buf), plug->version),
+          cs);
+    table->field[3]->set_notnull();
+  }
+  else
+    table->field[3]->set_null();
+
+  if (plug->info)
+  {
+    table->field[4]->store(version_buf,
+          make_version_string(version_buf, sizeof(version_buf), 
+                              *(uint *)plug->info), cs);
+    table->field[4]->set_notnull();
+  }
+  else
+    table->field[4]->set_null();
+
+  if (plugin->plugin_dl)
+  {
+    table->field[5]->store(plugin->plugin_dl->dl.str, 
+                           plugin->plugin_dl->dl.length, cs);
+    table->field[5]->set_notnull();
+  }
+  else
+    table->field[5]->set_null();
+  
+  if (plug->author)
+  {
+    table->field[6]->store(plug->author, strlen(plug->author), cs);
+    table->field[6]->set_notnull();
+  }
+  else
+    table->field[6]->set_null();
+
+  if (plug->descr)
+  {
+    table->field[7]->store(plug->descr, strlen(plug->descr), cs);
+    table->field[7]->set_notnull();
+  }
+  else
+    table->field[7]->set_null();
+
+  return schema_table_store_record(thd, table);
+}
+
+
+int fill_plugins(THD *thd, TABLE_LIST *tables, COND *cond)
+{
+  DBUG_ENTER("fill_plugins");
+  TABLE *table= tables->table;
+
+  if (plugin_foreach(thd, show_plugins, MYSQL_ANY_PLUGIN, table))
+    DBUG_RETURN(1);
+    
+  DBUG_RETURN(0);
+}
+
+
 /***************************************************************************
 ** List all Authors.
 ** If you can update it, you get to be in it :)
@@ -1005,8 +1123,8 @@ store_create_info(THD *thd, TABLE_LIST *table_list, String *packet)
       packet->append(STRING_WITH_LEN(" ENGINE="));
 #ifdef WITH_PARTITION_STORAGE_ENGINE
     if (table->part_info)
-      packet->append(ha_get_storage_engine(table->part_info->
-                                           default_engine_type));
+      packet->append(ha_resolve_storage_engine_name(
+                        table->part_info->default_engine_type));
     else
       packet->append(file->table_type());
 #else
@@ -2090,7 +2208,7 @@ int get_all_tables(THD *thd, TABLE_LIST *tables, COND *cond)
   Security_context *sctx= thd->security_ctx;
   uint derived_tables= lex->derived_tables; 
   int error= 1;
-  db_type not_used;
+  enum legacy_db_type not_used;
   Open_tables_state open_tables_state_backup;
   DBUG_ENTER("get_all_tables");
 
@@ -4172,6 +4290,20 @@ ST_FIELD_INFO variables_fields_info[]=
 };
 
 
+ST_FIELD_INFO plugin_fields_info[]=
+{
+  {"PLUGIN_NAME", NAME_LEN, MYSQL_TYPE_STRING, 0, 0, "Name"},
+  {"PLUGIN_STATUS", 10, MYSQL_TYPE_STRING, 0, 0, "Status"},
+  {"PLUGIN_TYPE", 10, MYSQL_TYPE_STRING, 0, 0, "Type"},
+  {"PLUGIN_VERSION", 20, MYSQL_TYPE_STRING, 0, 1, 0},
+  {"PLUGIN_TYPE_VERSION", 20, MYSQL_TYPE_STRING, 0, 1, 0},
+  {"PLUGIN_LIBRARY", NAME_LEN, MYSQL_TYPE_STRING, 0, 1, "Library"},
+  {"PLUGIN_AUTHOR", NAME_LEN, MYSQL_TYPE_STRING, 0, 1, 0},
+  {"PLUGIN_DESCRIPTION", 65535, MYSQL_TYPE_STRING, 0, 1, 0},
+  {0, 0, MYSQL_TYPE_STRING, 0, 0, 0}
+};
+
+
 /*
   Description of ST_FIELD_INFO in table.h
 */
@@ -4192,6 +4324,8 @@ ST_SCHEMA_TABLE schema_tables[]=
     get_all_tables, 0, get_schema_key_column_usage_record, 4, 5, 0},
   {"OPEN_TABLES", open_tables_fields_info, create_schema_table,
    fill_open_tables, make_old_format, 0, -1, -1, 1},
+  {"PLUGINS", plugin_fields_info, create_schema_table,
+    fill_plugins, make_old_format, 0, -1, -1, 0},
   {"ROUTINES", proc_fields_info, create_schema_table, 
     fill_schema_proc, make_proc_old_format, 0, -1, -1, 0},
   {"SCHEMATA", schema_fields_info, create_schema_table,
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index ff562cf2a39..c9a997a33db 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -27,7 +27,6 @@
 #include <io.h>
 #endif
 
-
 const char *primary_key_name="PRIMARY";
 
 static bool check_if_keyname_exists(const char *name,KEY *start, KEY *end);
@@ -40,7 +39,7 @@ static int copy_data_between_tables(TABLE *from,TABLE *to,
 				    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);                             
+                         handlerton **new_engine);                             
 
 /*
   SYNOPSIS
@@ -290,7 +289,7 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
   for (table= tables; table; table= table->next_local)
   {
     TABLE_SHARE *share;
-    table->db_type= DB_TYPE_UNKNOWN;
+    table->db_type= NULL;
     if ((share= get_cached_table_share(table->db, table->table_name)))
       table->db_type= share->db_type;
   }
@@ -304,7 +303,8 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
   for (table= tables; table; table= table->next_local)
   {
     char *db=table->db;
-    db_type table_type;
+    handlerton *table_type;
+    enum legacy_db_type frm_db_type;
 
     mysql_ha_flush(thd, table, MYSQL_HA_CLOSE_FINAL, TRUE);
     if (!close_temporary_table(thd, table))
@@ -331,12 +331,12 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
       /* remove .frm file and engine files */
       build_table_path(path, sizeof(path), db, alias, reg_ext);
     }
-    if (table_type == DB_TYPE_UNKNOWN &&
+    if (table_type == NULL &&
         (drop_temporary ||
          (access(path, F_OK) &&
           ha_create_table_from_engine(thd, db, alias)) ||
          (!drop_view &&
-          mysql_frm_type(thd, path, &table_type) != FRMTYPE_TABLE)))
+          mysql_frm_type(thd, path, &frm_db_type) != FRMTYPE_TABLE)))
     {
       // Table was not found on disk and table can't be created from engine
       if (if_exists)
@@ -349,13 +349,16 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
     else
     {
       char *end;
-      if (table_type == DB_TYPE_UNKNOWN)
-	mysql_frm_type(thd, path, &table_type);
+      if (table_type == NULL)
+      {
+	mysql_frm_type(thd, path, &frm_db_type);
+        table_type= ha_resolve_by_legacy_type(thd, frm_db_type);
+      }
       *(end=fn_ext(path))=0;			// Remove extension for delete
       error= ha_delete_table(thd, table_type, path, db, table->table_name,
                              !dont_log_query);
       if ((error == ENOENT || error == HA_ERR_NO_SUCH_TABLE) && 
-	  (if_exists || table_type == DB_TYPE_UNKNOWN))
+	  (if_exists || table_type == NULL))
 	error= 0;
       if (error == HA_ERR_ROW_IS_REFERENCED)
       {
@@ -413,7 +416,7 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
 }
 
 
-bool quick_rm_table(enum db_type base,const char *db,
+bool quick_rm_table(handlerton *base,const char *db,
 		   const char *table_name)
 {
   char path[FN_REFLEN];
@@ -1670,10 +1673,10 @@ bool mysql_create_table(THD *thd,const char *db, const char *table_name,
     this information in the default_db_type variable, it is either
     DB_TYPE_DEFAULT or the engine set in the ALTER TABLE command.
     */
-    enum db_type part_engine_type= create_info->db_type;
+    handlerton *part_engine_type= create_info->db_type;
     char *part_syntax_buf;
     uint syntax_len;
-    if (part_engine_type == DB_TYPE_PARTITION_DB)
+    if (part_engine_type == &partition_hton)
     {
       /*
         This only happens at ALTER TABLE.
@@ -1681,7 +1684,7 @@ bool mysql_create_table(THD *thd,const char *db, const char *table_name,
         TABLE command.
       */
       part_engine_type= ha_checktype(thd,
-                                     part_info->default_engine_type, 0, 0);
+                        ha_legacy_type(part_info->default_engine_type), 0, 0);
     }
     else
     {
@@ -1702,7 +1705,7 @@ bool mysql_create_table(THD *thd,const char *db, const char *table_name,
     part_info->part_info_string= part_syntax_buf;
     part_info->part_info_len= syntax_len;
     if ((!(file->partition_flags() & HA_CAN_PARTITION)) ||
-        create_info->db_type == DB_TYPE_PARTITION_DB)
+        create_info->db_type == &partition_hton)
     {
       /*
         The handler assigned to the table cannot handle partitioning.
@@ -1711,7 +1714,7 @@ bool mysql_create_table(THD *thd,const char *db, const char *table_name,
       DBUG_PRINT("info", ("db_type: %d  part_flag: %d",
                           create_info->db_type,file->partition_flags()));
       delete file;
-      create_info->db_type= DB_TYPE_PARTITION_DB;
+      create_info->db_type= &partition_hton;
       if (!(file= get_ha_partition(part_info)))
       {
         DBUG_RETURN(TRUE);
@@ -1928,8 +1931,9 @@ TABLE *create_table_from_items(THD *thd, HA_CREATE_INFO *create_info,
 
   tmp_table.s->db_create_options=0;
   tmp_table.s->blob_ptr_size= portable_sizeof_char_ptr;
-  tmp_table.s->db_low_byte_first= test(create_info->db_type == DB_TYPE_MYISAM ||
-                                       create_info->db_type == DB_TYPE_HEAP);
+  tmp_table.s->db_low_byte_first= 
+        test(create_info->db_type == &myisam_hton ||
+             create_info->db_type == &heap_hton);
   tmp_table.null_row=tmp_table.maybe_null=0;
 
   while ((item=it++))
@@ -2006,7 +2010,7 @@ TABLE *create_table_from_items(THD *thd, HA_CREATE_INFO *create_info,
 ****************************************************************************/
 
 bool
-mysql_rename_table(enum db_type base,
+mysql_rename_table(handlerton *base,
 		   const char *old_db,
 		   const char *old_name,
 		   const char *new_db,
@@ -2020,7 +2024,7 @@ mysql_rename_table(enum db_type base,
   int error=0;
   DBUG_ENTER("mysql_rename_table");
 
-  file= (base == DB_TYPE_UNKNOWN ? 0 :
+  file= (base == NULL ? 0 :
          get_new_handler((TABLE_SHARE*) 0, thd->mem_root, base));
 
   build_table_path(from, sizeof(from), old_db, old_name, "");
@@ -2865,7 +2869,7 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table,
   char *src_table= table_ident->table.str;
   int  err;
   bool res= TRUE;
-  db_type not_used;
+  enum legacy_db_type not_used;
 
   TABLE_LIST src_tables_list;
   DBUG_ENTER("mysql_create_like_table");
@@ -3153,7 +3157,7 @@ int mysql_create_indexes(THD *thd, TABLE_LIST *table_list, List<Key> &keys)
     fields.push_back(c_fld);
   }
   bzero((char*) &create_info,sizeof(create_info));
-  create_info.db_type=DB_TYPE_DEFAULT;
+  create_info.db_type= (handlerton*) &default_hton;
   create_info.default_table_charset= thd->variables.collation_database;
   db_options= 0;
   if (mysql_prepare_table(thd, &create_info, &fields,
@@ -3177,7 +3181,7 @@ int mysql_create_indexes(THD *thd, TABLE_LIST *table_list, List<Key> &keys)
   {
     /* Re-initialize the create_info, which was changed by prepare table. */
     bzero((char*) &create_info,sizeof(create_info));
-    create_info.db_type=DB_TYPE_DEFAULT;
+    create_info.db_type= (handlerton*) &default_hton;
     create_info.default_table_charset= thd->variables.collation_database;
     /* Cleanup the fields list. We do not want to create existing fields. */
     fields.delete_elements();
@@ -3272,7 +3276,7 @@ int mysql_drop_indexes(THD *thd, TABLE_LIST *table_list,
   }
 
   bzero((char*) &create_info,sizeof(create_info));
-  create_info.db_type=DB_TYPE_DEFAULT;
+  create_info.db_type= (handlerton*) &default_hton;
   create_info.default_table_charset= thd->variables.collation_database;
 
   if ((drop_key)|| (drop.elements<= 0))
@@ -3488,13 +3492,13 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
   ha_rows copied,deleted;
   ulonglong next_insert_id;
   uint db_create_options, used_fields;
-  enum db_type old_db_type,new_db_type;
+  handlerton *old_db_type, *new_db_type;
   uint need_copy_table= 0;
 #ifdef WITH_PARTITION_STORAGE_ENGINE
   bool online_add_empty_partition= FALSE;
   bool online_drop_partition= FALSE;
   bool partition_changed= FALSE;
-  enum db_type default_engine_type;
+  handlerton *default_engine_type;
 #endif
   DBUG_ENTER("mysql_alter_table");
 
@@ -3571,7 +3575,7 @@ 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)
+  if (create_info->db_type == (handlerton*) &default_hton)
     create_info->db_type= old_db_type;
 
 #ifdef WITH_PARTITION_STORAGE_ENGINE
@@ -3675,7 +3679,7 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
           DBUG_RETURN(TRUE);
         }
       }
-      create_info->db_type= DB_TYPE_PARTITION_DB;
+      create_info->db_type= &partition_hton;
       thd->lex->part_info= tab_part_info;
       if (table->file->alter_table_flags() & HA_ONLINE_ADD_EMPTY_PARTITION &&
           (tab_part_info->part_type == RANGE_PARTITION ||
@@ -3905,7 +3909,7 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
       }
     }
     partition_changed= TRUE;
-    create_info->db_type= DB_TYPE_PARTITION_DB;
+    create_info->db_type= &partition_hton;
     thd->lex->part_info= tab_part_info;
     if (alter_info->flags == ALTER_ADD_PARTITION ||
         alter_info->flags == ALTER_REORGANISE_PARTITION)
@@ -3976,9 +3980,9 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
       */
       if (thd->lex->part_info != table->part_info)
         partition_changed= TRUE;
-      if (create_info->db_type != DB_TYPE_PARTITION_DB)
+      if (create_info->db_type != &partition_hton)
         thd->lex->part_info->default_engine_type= create_info->db_type;
-      create_info->db_type= DB_TYPE_PARTITION_DB;
+      create_info->db_type= &partition_hton;
     }
   }
 #endif
@@ -4669,7 +4673,7 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
 
   error=0;
   if (!need_copy_table)
-    new_db_type=old_db_type=DB_TYPE_UNKNOWN; // this type cannot happen in regular ALTER
+    new_db_type=old_db_type= NULL; // this type cannot happen in regular ALTER
   if (mysql_rename_table(old_db_type,db,table_name,db,old_name))
   {
     error=1;
@@ -4999,7 +5003,7 @@ bool mysql_recreate_table(THD *thd, TABLE_LIST *table_list,
   lex->col_list.empty();
   lex->alter_info.reset();
   bzero((char*) &create_info,sizeof(create_info));
-  create_info.db_type=DB_TYPE_DEFAULT;
+  create_info.db_type= (handlerton*) &default_hton;
   create_info.row_type=ROW_TYPE_NOT_USED;
   create_info.default_table_charset=default_charset_info;
   /* Force alter table to recreate table */
@@ -5131,21 +5135,21 @@ bool mysql_checksum_table(THD *thd, TABLE_LIST *tables, HA_CHECK_OPT *check_opt)
 }
 
 static bool check_engine(THD *thd, const char *table_name,
-                         enum db_type *new_engine)
+                         handlerton **new_engine)
 {
-  enum db_type req_engine= *new_engine;
+  handlerton *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)
+  if (!(*new_engine= ha_checktype(thd, ha_legacy_type(req_engine),
+                                  no_substitution, 1)))
     return TRUE;
 
-  if (req_engine != *new_engine)
+  if (req_engine != (handlerton*) &default_hton && 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),
+                       ha_resolve_storage_engine_name(*new_engine),
                        table_name);
   }
   return FALSE;
diff --git a/sql/sql_view.cc b/sql/sql_view.cc
index 02bbe2bc3dc..0f9393f8857 100644
--- a/sql/sql_view.cc
+++ b/sql/sql_view.cc
@@ -1169,7 +1169,7 @@ bool mysql_drop_view(THD *thd, TABLE_LIST *views, enum_drop_mode drop_mode)
 {
   char path[FN_REFLEN];
   TABLE_LIST *view;
-  db_type not_used;
+  enum legacy_db_type not_used;
   DBUG_ENTER("mysql_drop_view");
 
   for (view= views; view; view= view->next_local)
@@ -1242,7 +1242,7 @@ err:
     FRMTYPE_VIEW	view
 */
 
-frm_type_enum mysql_frm_type(THD *thd, char *path, db_type *dbt)
+frm_type_enum mysql_frm_type(THD *thd, char *path, enum legacy_db_type *dbt)
 {
   File file;
   uchar header[10];	//"TYPE=VIEW\n" it is 10 characters
@@ -1271,7 +1271,7 @@ frm_type_enum mysql_frm_type(THD *thd, char *path, db_type *dbt)
        (header[2] < FRM_VER+3 || header[2] > FRM_VER+4)))
     DBUG_RETURN(FRMTYPE_TABLE);
 
-  *dbt= ha_checktype(thd, (enum db_type) (uint) *(header + 3), 0, 0);
+  *dbt= (enum legacy_db_type) (uint) *(header + 3);
   DBUG_RETURN(FRMTYPE_TABLE);                   // Is probably a .frm table
 }
 
diff --git a/sql/sql_view.h b/sql/sql_view.h
index cd61d7e9e71..1e3e5f4aa73 100644
--- a/sql/sql_view.h
+++ b/sql/sql_view.h
@@ -27,7 +27,7 @@ bool check_key_in_view(THD *thd, TABLE_LIST * view);
 
 bool insert_view_fields(THD *thd, List<Item> *list, TABLE_LIST *view);
 
-frm_type_enum mysql_frm_type(THD *thd, char *path, db_type *dbt);
+frm_type_enum mysql_frm_type(THD *thd, char *path, enum legacy_db_type *dbt);
 
 int view_checksum(THD *thd, TABLE_LIST *view);
 
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index c79ad97be4e..01491ba8795 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -92,7 +92,7 @@ inline Item *is_truth_value(Item *A, bool v1, bool v2)
   enum enum_var_type var_type;
   Key::Keytype key_type;
   enum ha_key_alg key_alg;
-  enum db_type db_type;
+  handlerton *db_type;
   enum row_type row_type;
   enum ha_rkey_function ha_rkey_mode;
   enum enum_tx_isolation tx_isolation;
@@ -1174,7 +1174,7 @@ create:
 	  lex->change=NullS;
 	  bzero((char*) &lex->create_info,sizeof(lex->create_info));
 	  lex->create_info.options=$2 | $4;
-	  lex->create_info.db_type= (enum db_type) lex->thd->variables.table_type;
+	  lex->create_info.db_type= lex->thd->variables.table_type;
 	  lex->create_info.default_table_charset= NULL;
 	  lex->name=0;
 	}
@@ -2781,7 +2781,7 @@ part_definition:
           part_info->current_partition= p_elem;
           part_info->use_default_partitions= FALSE;
           part_info->partitions.push_back(p_elem);
-          p_elem->engine_type= DB_TYPE_UNKNOWN;
+          p_elem->engine_type= NULL;
           part_info->count_curr_parts++;
         }
         part_name {}
@@ -2964,7 +2964,7 @@ sub_part_definition:
           part_info->current_partition->subpartitions.push_back(p_elem);
           part_info->use_default_subpartitions= FALSE;
           part_info->count_curr_subparts++;
-          p_elem->engine_type= DB_TYPE_UNKNOWN;
+          p_elem->engine_type= NULL;
         }
         sub_name opt_part_options {}
         ;
@@ -3193,8 +3193,8 @@ default_collation:
 storage_engines:
 	ident_or_text
 	{
-	  $$ = ha_resolve_by_name($1.str,$1.length);
-	  if ($$ == DB_TYPE_UNKNOWN &&
+	  $$ = ha_resolve_by_name(YYTHD, &$1);
+	  if ($$ == NULL &&
 	      test(YYTHD->variables.sql_mode & MODE_NO_ENGINE_SUBSTITUTION))
 	  {
 	    my_error(ER_UNKNOWN_STORAGE_ENGINE, MYF(0), $1.str);
@@ -3845,7 +3845,7 @@ alter:
           lex->select_lex.init_order();
 	  lex->select_lex.db=lex->name=0;
 	  bzero((char*) &lex->create_info,sizeof(lex->create_info));
-	  lex->create_info.db_type= DB_TYPE_DEFAULT;
+	  lex->create_info.db_type= (handlerton*) &default_hton;
 	  lex->create_info.default_table_charset= NULL;
 	  lex->create_info.row_type= ROW_TYPE_NOT_USED;
 	  lex->alter_info.reset();
@@ -7010,11 +7010,19 @@ show_param:
             if (prepare_schema_table(YYTHD, lex, 0, SCH_OPEN_TABLES))
               YYABORT;
 	  }
+        | PLUGIN_SYM
+	  {
+	    LEX *lex= Lex;
+            lex->sql_command= SQLCOM_SELECT;
+            lex->orig_sql_command= SQLCOM_SHOW_PLUGINS;
+            if (prepare_schema_table(YYTHD, lex, 0, SCH_PLUGINS))
+              YYABORT;
+	  }
 	| ENGINE_SYM storage_engines 
 	  { Lex->create_info.db_type= $2; }
 	  show_engine_param
 	| ENGINE_SYM ALL 
-	  { Lex->create_info.db_type= DB_TYPE_DEFAULT; }
+	  { Lex->create_info.db_type= NULL; }
 	  show_engine_param
 	| opt_full COLUMNS from_or_in table_ident opt_db wild_and_where
 	  {
@@ -7106,14 +7114,24 @@ show_param:
           {
             LEX *lex= Lex;
             lex->sql_command = SQLCOM_SHOW_ENGINE_STATUS;
-            lex->create_info.db_type= DB_TYPE_INNODB;
+            if (!(lex->create_info.db_type=
+                  ha_resolve_by_legacy_type(YYTHD, DB_TYPE_INNODB)))
+            {
+	      my_error(ER_UNKNOWN_STORAGE_ENGINE, MYF(0), "InnoDB");
+	      YYABORT;
+            }
             WARN_DEPRECATED("SHOW INNODB STATUS", "SHOW ENGINE INNODB STATUS");
 	  }
         | MUTEX_SYM STATUS_SYM
           {
 	    LEX *lex= Lex;
             lex->sql_command = SQLCOM_SHOW_ENGINE_MUTEX; 
-            lex->create_info.db_type= DB_TYPE_INNODB;
+            if (!(lex->create_info.db_type=
+                  ha_resolve_by_legacy_type(YYTHD, DB_TYPE_INNODB)))
+            {
+	      my_error(ER_UNKNOWN_STORAGE_ENGINE, MYF(0), "InnoDB");
+	      YYABORT;
+            }
             WARN_DEPRECATED("SHOW MUTEX STATUS", "SHOW ENGINE INNODB MUTEX");
 	  }
 	| opt_full PROCESSLIST_SYM
@@ -7147,14 +7165,24 @@ show_param:
 	  {
 	    LEX *lex= Lex;
 	    lex->sql_command= SQLCOM_SHOW_ENGINE_LOGS;
-	    lex->create_info.db_type= DB_TYPE_BERKELEY_DB;
+            if (!(lex->create_info.db_type=
+                  ha_resolve_by_legacy_type(YYTHD, DB_TYPE_BERKELEY_DB)))
+            {
+	      my_error(ER_UNKNOWN_STORAGE_ENGINE, MYF(0), "BerkeleyDB");
+	      YYABORT;
+            }
 	    WARN_DEPRECATED("SHOW BDB LOGS", "SHOW ENGINE BDB LOGS");
 	  }
 	| LOGS_SYM
 	  {
 	    LEX *lex= Lex;
 	    lex->sql_command= SQLCOM_SHOW_ENGINE_LOGS;
-	    lex->create_info.db_type= DB_TYPE_BERKELEY_DB;
+            if (!(lex->create_info.db_type=
+                  ha_resolve_by_legacy_type(YYTHD, DB_TYPE_BERKELEY_DB)))
+            {
+	      my_error(ER_UNKNOWN_STORAGE_ENGINE, MYF(0), "BerkeleyDB");
+	      YYABORT;
+            }
 	    WARN_DEPRECATED("SHOW LOGS", "SHOW ENGINE BDB LOGS");
 	  }
 	| GRANTS
@@ -9828,7 +9856,7 @@ opt_migrate:
     ;
 
 install:
-  INSTALL_SYM PLUGIN_SYM IDENT_sys SONAME_SYM TEXT_STRING_sys
+  INSTALL_SYM PLUGIN_SYM ident SONAME_SYM TEXT_STRING_sys
   {
     LEX *lex= Lex;
     lex->sql_command= SQLCOM_INSTALL_PLUGIN;
@@ -9837,7 +9865,7 @@ install:
   };
 
 uninstall:
-  UNINSTALL_SYM PLUGIN_SYM IDENT_sys
+  UNINSTALL_SYM PLUGIN_SYM ident
   {
     LEX *lex= Lex;
     lex->sql_command= SQLCOM_UNINSTALL_PLUGIN;
diff --git a/sql/table.cc b/sql/table.cc
index ff29a33ef03..91fbf915dc5 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -327,6 +327,7 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
   SQL_CRYPT *crypted=0;
   Field  **field_ptr, *reg_field;
   const char **interval_array;
+  enum legacy_db_type legacy_db_type;
   DBUG_ENTER("open_binary_frm");
 
   new_field_pack_flag= head[27];
@@ -349,11 +350,11 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
     share->frm_version= FRM_VER_TRUE_VARCHAR;
 
 #ifdef WITH_PARTITION_STORAGE_ENGINE
-  share->default_part_db_type= ha_checktype(thd,
-                                            (enum db_type) (uint) *(head+61),0,
-                                            0);
+  share->default_part_db_type= 
+        ha_checktype(thd, (enum legacy_db_type) (uint) *(head+61), 0, 0);
 #endif
-  share->db_type= ha_checktype(thd, (enum db_type) (uint) *(head+3),0,0);
+  legacy_db_type= (enum legacy_db_type) (uint) *(head+3);
+  share->db_type= ha_checktype(thd, legacy_db_type, 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);
@@ -385,7 +386,7 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
   if (db_create_options & HA_OPTION_LONG_BLOB_PTR)
     share->blob_ptr_size= portable_sizeof_char_ptr;
   /* Set temporarily a good value for db_low_byte_first */
-  share->db_low_byte_first= test(share->db_type != DB_TYPE_ISAM);
+  share->db_low_byte_first= test(legacy_db_type != DB_TYPE_ISAM);
   error=4;
   share->max_rows= uint4korr(head+18);
   share->min_rows= uint4korr(head+22);
@@ -513,14 +514,14 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
     if (next_chunk + 2 < buff_end)
     {
       uint str_db_type_length= uint2korr(next_chunk);
-      enum db_type tmp_db_type= ha_resolve_by_name(next_chunk + 2,
-                                                   str_db_type_length);
-      if (tmp_db_type != DB_TYPE_UNKNOWN)
+      LEX_STRING name= { next_chunk + 2, str_db_type_length };
+      handlerton *tmp_db_type= ha_resolve_by_name(thd, &name);
+      if (tmp_db_type != NULL)
       {
         share->db_type= tmp_db_type;
         DBUG_PRINT("info", ("setting dbtype to '%.*s' (%d)",
                             str_db_type_length, next_chunk + 2,
-                            share->db_type));
+                            ha_legacy_type(share->db_type)));
       }
 #ifdef WITH_PARTITION_STORAGE_ENGINE
       else
@@ -528,10 +529,10 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
         if (!strncmp(next_chunk + 2, "partition", str_db_type_length))
         {
           /* Use partition handler */
-          share->db_type= DB_TYPE_PARTITION_DB;
+          share->db_type= &partition_hton;
           DBUG_PRINT("info", ("setting dbtype to '%.*s' (%d)",
                               str_db_type_length, next_chunk + 2,
-                              share->db_type));
+                              ha_legacy_type(share->db_type)));
         }
       }
 #endif
@@ -1634,7 +1635,7 @@ void open_table_error(TABLE_SHARE *share, int error, int db_errno, int errarg)
     handler *file= 0;
     const char *datext= "";
     
-    if (share->db_type != DB_TYPE_UNKNOWN)
+    if (share->db_type != NULL)
     {
       if ((file= get_new_handler(share, current_thd->mem_root,
                                  share->db_type)))
@@ -1900,7 +1901,8 @@ File create_frm(THD *thd, const char *name, const char *db,
     fileinfo[1]= 1;
     fileinfo[2]= FRM_VER+3+ test(create_info->varchar);
 
-    fileinfo[3]= (uchar) ha_checktype(thd,create_info->db_type,0,0);
+    fileinfo[3]= (uchar) ha_legacy_type(
+          ha_checktype(thd,ha_legacy_type(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;
diff --git a/sql/table.h b/sql/table.h
index 81c13552a04..ef8ef5242d0 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -153,7 +153,7 @@ typedef struct st_table_share
   ulong   timestamp_offset;		/* Set to offset+1 of record */
   ulong   reclength;			/* Recordlength */
 
-  enum db_type db_type;			/* table_type for handler */
+  handlerton *db_type;			/* table_type for handler */
   enum row_type row_type;		/* How rows are stored */
   enum tmp_table_type tmp_table;
 
@@ -200,7 +200,7 @@ typedef struct st_table_share
 #ifdef WITH_PARTITION_STORAGE_ENGINE
   const uchar *partition_info;
   uint  partition_info_len;
-  enum db_type default_part_db_type;
+  handlerton *default_part_db_type;
 #endif
 } TABLE_SHARE;
 
@@ -324,6 +324,7 @@ enum enum_schema_tables
   SCH_COLUMN_PRIVILEGES,
   SCH_KEY_COLUMN_USAGE,
   SCH_OPEN_TABLES,
+  SCH_PLUGINS,
   SCH_PROCEDURES,
   SCH_SCHEMATA,
   SCH_SCHEMA_PRIVILEGES,
@@ -642,7 +643,7 @@ typedef struct st_table_list
   bool          where_processed;
   /* FRMTYPE_ERROR if any type is acceptable */
   enum frm_type_enum required_type;
-  enum db_type  db_type;		/* table_type for handler */
+  handlerton	*db_type;		/* table_type for handler */
   char		timestamp_buffer[20];	/* buffer for timestamp (19+1) */
   /*
     This TABLE_LIST object is just placeholder for prelocking, it will be
diff --git a/sql/unireg.cc b/sql/unireg.cc
index ebbf1177953..66be20736e8 100644
--- a/sql/unireg.cc
+++ b/sql/unireg.cc
@@ -35,7 +35,7 @@ static uchar * pack_screens(List<create_field> &create_fields,
 			    uint *info_length, uint *screens, bool small_file);
 static uint pack_keys(uchar *keybuff,uint key_count, KEY *key_info,
                       ulong data_offset);
-static bool pack_header(uchar *forminfo,enum db_type table_type,
+static bool pack_header(uchar *forminfo,enum legacy_db_type table_type,
 			List<create_field> &create_fields,
 			uint info_length, uint screens, uint table_options,
 			ulong data_offset, handler *file);
@@ -43,7 +43,7 @@ static uint get_interval_id(uint *int_count,List<create_field> &create_fields,
 			    create_field *last_field);
 static bool pack_fields(File file, List<create_field> &create_fields,
                         ulong data_offset);
-static bool make_empty_rec(THD *thd, int file, enum db_type table_type,
+static bool make_empty_rec(THD *thd, int file, enum legacy_db_type table_type,
 			   uint table_options,
 			   List<create_field> &create_fields,
 			   uint reclength, ulong data_offset,
@@ -103,7 +103,8 @@ bool mysql_create_frm(THD *thd, const char *file_name,
     create_info->null_bits++;
   data_offset= (create_info->null_bits + 7) / 8;
 
-  if (pack_header(forminfo, create_info->db_type,create_fields,info_length,
+  if (pack_header(forminfo, ha_legacy_type(create_info->db_type),
+                  create_fields,info_length,
 		  screens, create_info->table_options,
                   data_offset, db_file))
   {
@@ -115,7 +116,8 @@ bool mysql_create_frm(THD *thd, const char *file_name,
     thd->net.last_error[0]=0;
     if (!(screen_buff=pack_screens(create_fields,&info_length,&screens,1)))
       DBUG_RETURN(1);
-    if (pack_header(forminfo, create_info->db_type, create_fields,info_length,
+    if (pack_header(forminfo, ha_legacy_type(create_info->db_type),
+                    create_fields,info_length,
 		    screens, create_info->table_options, data_offset, db_file))
     {
       my_free((gptr) screen_buff,MYF(0));
@@ -125,7 +127,7 @@ bool mysql_create_frm(THD *thd, const char *file_name,
   reclength=uint2korr(forminfo+266);
 
   /* Calculate extra data segment length */
-  str_db_type.str= (char *) ha_get_storage_engine(create_info->db_type);
+  str_db_type.str= (char *) ha_resolve_storage_engine_name(create_info->db_type);
   str_db_type.length= strlen(str_db_type.str);
   /* str_db_type */
   create_info->extra_size= (2 + str_db_type.length +
@@ -168,7 +170,7 @@ bool mysql_create_frm(THD *thd, const char *file_name,
 
 #ifdef WITH_PARTITION_STORAGE_ENGINE
   if (part_info)
-    fileinfo[61]= (uchar) part_info->default_engine_type;
+    fileinfo[61]= (uchar) ha_legacy_type(part_info->default_engine_type);
 #endif
   int2store(fileinfo+59,db_file->extra_rec_buf_length());
   if (my_pwrite(file,(byte*) fileinfo,64,0L,MYF_RW) ||
@@ -178,7 +180,8 @@ bool mysql_create_frm(THD *thd, const char *file_name,
   VOID(my_seek(file,
 	       (ulong) uint2korr(fileinfo+6)+ (ulong) key_buff_length,
 	       MY_SEEK_SET,MYF(0)));
-  if (make_empty_rec(thd,file,create_info->db_type,create_info->table_options,
+  if (make_empty_rec(thd,file,ha_legacy_type(create_info->db_type),
+                     create_info->table_options,
 		     create_fields,reclength, data_offset, db_file))
     goto err;
 
@@ -480,7 +483,7 @@ static uint pack_keys(uchar *keybuff, uint key_count, KEY *keyinfo,
 
 	/* Make formheader */
 
-static bool pack_header(uchar *forminfo, enum db_type table_type,
+static bool pack_header(uchar *forminfo, enum legacy_db_type table_type,
 			List<create_field> &create_fields,
                         uint info_length, uint screens, uint table_options,
                         ulong data_offset, handler *file)
@@ -739,7 +742,7 @@ static bool pack_fields(File file, List<create_field> &create_fields,
 
 	/* save an empty record on start of formfile */
 
-static bool make_empty_rec(THD *thd, File file,enum db_type table_type,
+static bool make_empty_rec(THD *thd, File file,enum legacy_db_type table_type,
 			   uint table_options,
 			   List<create_field> &create_fields,
 			   uint reclength,