Merge tnurnberg@bk-internal.mysql.com:/home/bk/mysql-5.1

into  salvation.intern.azundris.com:/home/tnurnberg/mysql-5.1

34 files changed, 6893 insertions, 181 deletions

diff --git a/storage/archive/CMakeLists.txt b/storage/archive/CMakeLists.txt
index a631f194b1a..127942d4043 100644
--- a/storage/archive/CMakeLists.txt
+++ b/storage/archive/CMakeLists.txt
@@ -3,6 +3,7 @@ SET(CMAKE_C_FLAGS_DEBUG "${CMAKE_C_FLAGS_DEBUG} -DSAFEMALLOC -DSAFE_MUTEX")
 
 INCLUDE_DIRECTORIES(${CMAKE_SOURCE_DIR}/include ${CMAKE_SOURCE_DIR}/zlib
                     ${CMAKE_SOURCE_DIR}/sql
+                    ${CMAKE_SOURCE_DIR}/regex
                     ${CMAKE_SOURCE_DIR}/extra/yassl/include)
 ADD_LIBRARY(archive azio.c ha_archive.cc ha_archive.h)
 TARGET_LINK_LIBRARIES(archive zlib mysys dbug strings)
diff --git a/storage/blackhole/CMakeLists.txt b/storage/blackhole/CMakeLists.txt
index ea3a7eae38e..a90f8e14ca0 100644
--- a/storage/blackhole/CMakeLists.txt
+++ b/storage/blackhole/CMakeLists.txt
@@ -2,5 +2,6 @@ SET(CMAKE_CXX_FLAGS_DEBUG "${CMAKE_CXX_FLAGS_DEBUG} -DSAFEMALLOC -DSAFE_MUTEX")
 SET(CMAKE_C_FLAGS_DEBUG "${CMAKE_C_FLAGS_DEBUG} -DSAFEMALLOC -DSAFE_MUTEX")
 
 INCLUDE_DIRECTORIES(${CMAKE_SOURCE_DIR}/include ${CMAKE_SOURCE_DIR}/sql
+                    ${CMAKE_SOURCE_DIR}/regex
                     ${CMAKE_SOURCE_DIR}/extra/yassl/include)
 ADD_LIBRARY(blackhole ha_blackhole.cc ha_blackhole.h)
diff --git a/storage/blackhole/plug.in b/storage/blackhole/plug.in
new file mode 100644
index 00000000000..725db0facba
--- /dev/null
+++ b/storage/blackhole/plug.in
@@ -0,0 +1,6 @@
+MYSQL_STORAGE_ENGINE(blackhole,,[Blackhole Storage Engine],
+        [Basic Write-only Read-never tables], [max,max-no-ndb])
+MYSQL_PLUGIN_DIRECTORY(blackhole, [storage/blackhole])
+MYSQL_PLUGIN_STATIC(blackhole,  [libblackhole.a])
+MYSQL_PLUGIN_DYNAMIC(blackhole, [ha_blackhole.la])
+
diff --git a/storage/csv/CMakeLists.txt b/storage/csv/CMakeLists.txt
index 28748527cc3..55e9b50fbfc 100644
--- a/storage/csv/CMakeLists.txt
+++ b/storage/csv/CMakeLists.txt
@@ -2,5 +2,6 @@ SET(CMAKE_CXX_FLAGS_DEBUG "${CMAKE_CXX_FLAGS_DEBUG} -DSAFEMALLOC -DSAFE_MUTEX")
 SET(CMAKE_C_FLAGS_DEBUG "${CMAKE_C_FLAGS_DEBUG} -DSAFEMALLOC -DSAFE_MUTEX")
 
 INCLUDE_DIRECTORIES(${CMAKE_SOURCE_DIR}/include ${CMAKE_SOURCE_DIR}/sql
+                    ${CMAKE_SOURCE_DIR}/regex
                     ${CMAKE_SOURCE_DIR}/extra/yassl/include)
 ADD_LIBRARY(csv ha_tina.cc ha_tina.h)
diff --git a/storage/csv/ha_tina.cc b/storage/csv/ha_tina.cc
index bec236becd0..2fe2afeb470 100644
--- a/storage/csv/ha_tina.cc
+++ b/storage/csv/ha_tina.cc
@@ -157,6 +157,7 @@ static int tina_init_func()
     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);
+    bzero(&tina_hton, sizeof(handlerton));
     tina_hton.state= SHOW_OPTION_YES;
     tina_hton.db_type= DB_TYPE_CSV_DB;
     tina_hton.create= tina_create_handler;
@@ -229,6 +230,11 @@ static TINA_SHARE *get_share(const char *table_name, TABLE *table)
               MY_REPLACE_EXT|MY_UNPACK_FILENAME);
     fn_format(meta_file_name, table_name, "", CSM_EXT,
               MY_REPLACE_EXT|MY_UNPACK_FILENAME);
+
+    if (my_stat(share->data_file_name, &file_stat, MYF(MY_WME)) == NULL)
+      goto error;
+    share->saved_data_file_length= file_stat.st_size;
+
     if (my_hash_insert(&tina_open_tables, (byte*) share))
       goto error;
     thr_lock_init(&share->lock);
@@ -250,10 +256,6 @@ static TINA_SHARE *get_share(const char *table_name, TABLE *table)
     */
     if (read_meta_file(share->meta_file, &share->rows_recorded))
       share->crashed= TRUE;
-
-    if (my_stat(share->data_file_name, &file_stat, MYF(MY_WME)) == NULL)
-      goto error2;
-    share->saved_data_file_length= file_stat.st_size;
   }
   share->use_count++;
   pthread_mutex_unlock(&tina_mutex);
diff --git a/storage/csv/plug.in b/storage/csv/plug.in
new file mode 100644
index 00000000000..bbc69680fcd
--- /dev/null
+++ b/storage/csv/plug.in
@@ -0,0 +1,5 @@
+MYSQL_STORAGE_ENGINE(csv,,      [CSV Storage Engine],
+        [Stores tables in text CSV format])
+MYSQL_PLUGIN_DIRECTORY(csv,     [storage/csv])
+MYSQL_PLUGIN_STATIC(csv,        [libcsv.a])
+MYSQL_PLUGIN_MANDATORY(csv)     dnl Used for logging
diff --git a/storage/example/CMakeLists.txt b/storage/example/CMakeLists.txt
index f4579aa0c66..384631a66c4 100644
--- a/storage/example/CMakeLists.txt
+++ b/storage/example/CMakeLists.txt
@@ -2,5 +2,6 @@ SET(CMAKE_CXX_FLAGS_DEBUG "${CMAKE_CXX_FLAGS_DEBUG} -DSAFEMALLOC -DSAFE_MUTEX")
 SET(CMAKE_C_FLAGS_DEBUG "${CMAKE_C_FLAGS_DEBUG} -DSAFEMALLOC -DSAFE_MUTEX")
 
 INCLUDE_DIRECTORIES(${CMAKE_SOURCE_DIR}/include ${CMAKE_SOURCE_DIR}/sql
+                    ${CMAKE_SOURCE_DIR}/regex
                     ${CMAKE_SOURCE_DIR}/extra/yassl/include)
 ADD_LIBRARY(example ha_example.cc)
diff --git a/storage/example/ha_example.cc b/storage/example/ha_example.cc
index feabad2e356..704ea757749 100644
--- a/storage/example/ha_example.cc
+++ b/storage/example/ha_example.cc
@@ -67,6 +67,7 @@
 #pragma implementation        // gcc: Class implementation
 #endif
 
+#define MYSQL_SERVER 1
 #include "mysql_priv.h"
 #include "ha_example.h"
 
diff --git a/storage/federated/CMakeLists.txt b/storage/federated/CMakeLists.txt
new file mode 100644
index 00000000000..97a4f318a11
--- /dev/null
+++ b/storage/federated/CMakeLists.txt
@@ -0,0 +1,7 @@
+SET(CMAKE_CXX_FLAGS_DEBUG "${CMAKE_CXX_FLAGS_DEBUG} -DSAFEMALLOC -DSAFE_MUTEX")
+SET(CMAKE_C_FLAGS_DEBUG "${CMAKE_C_FLAGS_DEBUG} -DSAFEMALLOC -DSAFE_MUTEX")
+
+INCLUDE_DIRECTORIES(${CMAKE_SOURCE_DIR}/include ${CMAKE_SOURCE_DIR}/sql
+                    ${CMAKE_SOURCE_DIR}/regex
+                    ${CMAKE_SOURCE_DIR}/extra/yassl/include)
+ADD_LIBRARY(federated ha_federated.cc)
diff --git a/storage/federated/Makefile.am b/storage/federated/Makefile.am
new file mode 100644
index 00000000000..813455ed5c7
--- /dev/null
+++ b/storage/federated/Makefile.am
@@ -0,0 +1,52 @@
+# Copyright (C) 2000 MySQL AB & MySQL Finland AB & TCX DataKonsult 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
+
+#called from the top level Makefile
+
+MYSQLDATAdir =          $(localstatedir)
+MYSQLSHAREdir =         $(pkgdatadir)
+MYSQLBASEdir=           $(prefix)
+MYSQLLIBdir=            $(pkglibdir)
+INCLUDES =              -I$(top_srcdir)/include -I$(top_builddir)/include \
+			-I$(top_srcdir)/regex \
+			-I$(top_srcdir)/sql \
+                        -I$(srcdir)
+WRAPLIBS=
+
+LDADD =
+
+DEFS =                  @DEFS@
+
+noinst_HEADERS =	ha_federated.h
+
+EXTRA_LTLIBRARIES =	ha_federated.la
+pkglib_LTLIBRARIES =	@plugin_federated_shared_target@
+ha_federated_la_LDFLAGS =	-module -rpath $(MYSQLLIBdir)
+ha_federated_la_CXXFLAGS=	$(AM_CFLAGS) -DMYSQL_DYNAMIC_PLUGIN
+ha_federated_la_CFLAGS =	$(AM_CFLAGS) -DMYSQL_DYNAMIC_PLUGIN
+ha_federated_la_SOURCES =	ha_federated.cc
+
+
+EXTRA_LIBRARIES =	libfederated.a
+noinst_LIBRARIES =	@plugin_federated_static_target@
+libfederated_a_CXXFLAGS =	$(AM_CFLAGS)
+libfederated_a_CFLAGS =	$(AM_CFLAGS)
+libfederated_a_SOURCES=	ha_federated.cc
+
+
+EXTRA_DIST =		CMakeLists.txt
+# Don't update the files from bitkeeper
+%::SCCS/s.%
diff --git a/storage/federated/ha_federated.cc b/storage/federated/ha_federated.cc
new file mode 100644
index 00000000000..98f48b09ba6
--- /dev/null
+++ b/storage/federated/ha_federated.cc
@@ -0,0 +1,2902 @@
+/* Copyright (C) 2004 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 */
+
+/*
+
+  MySQL Federated Storage Engine
+
+  ha_federated.cc - MySQL Federated Storage Engine
+  Patrick Galbraith and Brian Aker, 2004
+
+  This is a handler which uses a foreign database as the data file, as
+  opposed to a handler like MyISAM, which uses .MYD files locally.
+
+  How this handler works
+  ----------------------------------
+  Normal database files are local and as such: You create a table called
+  'users', a file such as 'users.MYD' is created. A handler reads, inserts,
+  deletes, updates data in this file. The data is stored in particular format,
+  so to read, that data has to be parsed into fields, to write, fields have to
+  be stored in this format to write to this data file.
+
+  With MySQL Federated storage engine, there will be no local files
+  for each table's data (such as .MYD). A foreign database will store
+  the data that would normally be in this file. This will necessitate
+  the use of MySQL client API to read, delete, update, insert this
+  data. The data will have to be retrieve via an SQL call "SELECT *
+  FROM users". Then, to read this data, it will have to be retrieved
+  via mysql_fetch_row one row at a time, then converted from the
+  column in this select into the format that the handler expects.
+
+  The create table will simply create the .frm file, and within the
+  "CREATE TABLE" SQL, there SHALL be any of the following :
+
+  comment=scheme://username:password@hostname:port/database/tablename
+  comment=scheme://username@hostname/database/tablename
+  comment=scheme://username:password@hostname/database/tablename
+  comment=scheme://username:password@hostname/database/tablename
+
+  An example would be:
+
+  comment=mysql://username:password@hostname:port/database/tablename
+
+  ***IMPORTANT***
+
+  This is a first release, conceptual release
+  Only 'mysql://' is supported at this release.
+
+
+  This comment connection string is necessary for the handler to be
+  able to connect to the foreign server.
+
+
+  The basic flow is this:
+
+  SQL calls issues locally ->
+  mysql handler API (data in handler format) ->
+  mysql client API (data converted to SQL calls) ->
+  foreign database -> mysql client API ->
+  convert result sets (if any) to handler format ->
+  handler API -> results or rows affected to local
+
+  What this handler does and doesn't support
+  ------------------------------------------
+  * Tables MUST be created on the foreign server prior to any action on those
+    tables via the handler, first version. IMPORTANT: IF you MUST use the
+    federated storage engine type on the REMOTE end, MAKE SURE [ :) ] That
+    the table you connect to IS NOT a table pointing BACK to your ORIGNAL
+    table! You know  and have heard the screaching of audio feedback? You
+    know putting two mirror in front of each other how the reflection
+    continues for eternity? Well, need I say more?!
+  * There will not be support for transactions.
+  * There is no way for the handler to know if the foreign database or table
+    has changed. The reason for this is that this database has to work like a
+    data file that would never be written to by anything other than the
+    database. The integrity of the data in the local table could be breached
+    if there was any change to the foreign database.
+  * Support for SELECT, INSERT, UPDATE , DELETE, indexes.
+  * No ALTER TABLE, DROP TABLE or any other Data Definition Language calls.
+  * Prepared statements will not be used in the first implementation, it
+    remains to to be seen whether the limited subset of the client API for the
+    server supports this.
+  * This uses SELECT, INSERT, UPDATE, DELETE and not HANDLER for its
+    implementation.
+  * This will not work with the query cache.
+
+   Method calls
+
+   A two column table, with one record:
+
+   (SELECT)
+
+   "SELECT * FROM foo"
+    ha_federated::info
+    ha_federated::scan_time:
+    ha_federated::rnd_init: share->select_query SELECT * FROM foo
+    ha_federated::extra
+
+    <for every row of data retrieved>
+    ha_federated::rnd_next
+    ha_federated::convert_row_to_internal_format
+    ha_federated::rnd_next
+    </for every row of data retrieved>
+
+    ha_federated::rnd_end
+    ha_federated::extra
+    ha_federated::reset
+
+    (INSERT)
+
+    "INSERT INTO foo (id, ts) VALUES (2, now());"
+
+    ha_federated::write_row
+
+    ha_federated::reset
+
+    (UPDATE)
+
+    "UPDATE foo SET ts = now() WHERE id = 1;"
+
+    ha_federated::index_init
+    ha_federated::index_read
+    ha_federated::index_read_idx
+    ha_federated::rnd_next
+    ha_federated::convert_row_to_internal_format
+    ha_federated::update_row
+
+    ha_federated::extra
+    ha_federated::extra
+    ha_federated::extra
+    ha_federated::external_lock
+    ha_federated::reset
+
+
+    How do I use this handler?
+    --------------------------
+    First of all, you need to build this storage engine:
+
+      ./configure --with-federated-storage-engine
+      make
+
+    Next, to use this handler, it's very simple. You must
+    have two databases running, either both on the same host, or
+    on different hosts.
+
+    One the server that will be connecting to the foreign
+    host (client), you create your table as such:
+
+    CREATE TABLE test_table (
+      id     int(20) NOT NULL auto_increment,
+      name   varchar(32) NOT NULL default '',
+      other  int(20) NOT NULL default '0',
+      PRIMARY KEY  (id),
+      KEY name (name),
+      KEY other_key (other))
+       ENGINE="FEDERATED"
+       DEFAULT CHARSET=latin1
+       COMMENT='root@127.0.0.1:9306/federated/test_federated';
+
+   Notice the "COMMENT" and "ENGINE" field? This is where you
+   respectively set the engine type, "FEDERATED" and foreign
+   host information, this being the database your 'client' database
+   will connect to and use as the "data file". Obviously, the foreign
+   database is running on port 9306, so you want to start up your other
+   database so that it is indeed on port 9306, and your federated
+   database on a port other than that. In my setup, I use port 5554
+   for federated, and port 5555 for the foreign database.
+
+   Then, on the foreign database:
+
+   CREATE TABLE test_table (
+     id     int(20) NOT NULL auto_increment,
+     name   varchar(32) NOT NULL default '',
+     other  int(20) NOT NULL default '0',
+     PRIMARY KEY  (id),
+     KEY name (name),
+     KEY other_key (other))
+     ENGINE="<NAME>" <-- whatever you want, or not specify
+     DEFAULT CHARSET=latin1 ;
+
+    This table is exactly the same (and must be exactly the same),
+    except that it is not using the federated handler and does
+    not need the URL.
+
+
+    How to see the handler in action
+    --------------------------------
+
+    When developing this handler, I compiled the federated database with
+    debugging:
+
+    ./configure --with-federated-storage-engine
+    --prefix=/home/mysql/mysql-build/federated/ --with-debug
+
+    Once compiled, I did a 'make install' (not for the purpose of installing
+    the binary, but to install all the files the binary expects to see in the
+    diretory I specified in the build with --prefix,
+    "/home/mysql/mysql-build/federated".
+
+    Then, I started the foreign server:
+
+    /usr/local/mysql/bin/mysqld_safe
+    --user=mysql --log=/tmp/mysqld.5555.log -P 5555
+
+    Then, I went back to the directory containing the newly compiled mysqld,
+    <builddir>/sql/, started up gdb:
+
+    gdb ./mysqld
+
+    Then, withn the (gdb) prompt:
+    (gdb) run --gdb --port=5554 --socket=/tmp/mysqld.5554 --skip-innodb --debug
+
+    Next, I open several windows for each:
+
+    1. Tail the debug trace: tail -f /tmp/mysqld.trace|grep ha_fed
+    2. Tail the SQL calls to the foreign database: tail -f /tmp/mysqld.5555.log
+    3. A window with a client open to the federated server on port 5554
+    4. A window with a client open to the federated server on port 5555
+
+    I would create a table on the client to the foreign server on port
+    5555, and then to the federated server on port 5554. At this point,
+    I would run whatever queries I wanted to on the federated server,
+    just always remembering that whatever changes I wanted to make on
+    the table, or if I created new tables, that I would have to do that
+    on the foreign server.
+
+    Another thing to look for is 'show variables' to show you that you have
+    support for federated handler support:
+
+    show variables like '%federat%'
+
+    and:
+
+    show storage engines;
+
+    Both should display the federated storage handler.
+
+
+    Testing
+    -------
+
+    There is a test for MySQL Federated Storage Handler in ./mysql-test/t,
+    federatedd.test It starts both a slave and master database using
+    the same setup that the replication tests use, with the exception that
+    it turns off replication, and sets replication to ignore the test tables.
+    After ensuring that you actually do have support for the federated storage
+    handler, numerous queries/inserts/updates/deletes are run, many derived
+    from the MyISAM tests, plus som other tests which were meant to reveal
+    any issues that would be most likely to affect this handler. All tests
+    should work! ;)
+
+    To run these tests, go into ./mysql-test (based in the directory you
+    built the server in)
+
+    ./mysql-test-run federatedd
+
+    To run the test, or if you want to run the test and have debug info:
+
+    ./mysql-test-run --debug federated
+
+    This will run the test in debug mode, and you can view the trace and
+    log files in the ./mysql-test/var/log directory
+
+    ls -l mysql-test/var/log/
+    -rw-r--r--  1 patg  patg        17  4 Dec 12:27 current_test
+    -rw-r--r--  1 patg  patg       692  4 Dec 12:52 manager.log
+    -rw-rw----  1 patg  patg     21246  4 Dec 12:51 master-bin.000001
+    -rw-rw----  1 patg  patg        68  4 Dec 12:28 master-bin.index
+    -rw-r--r--  1 patg  patg      1620  4 Dec 12:51 master.err
+    -rw-rw----  1 patg  patg     23179  4 Dec 12:51 master.log
+    -rw-rw----  1 patg  patg  16696550  4 Dec 12:51 master.trace
+    -rw-r--r--  1 patg  patg         0  4 Dec 12:28 mysqltest-time
+    -rw-r--r--  1 patg  patg   2024051  4 Dec 12:51 mysqltest.trace
+    -rw-rw----  1 patg  patg     94992  4 Dec 12:51 slave-bin.000001
+    -rw-rw----  1 patg  patg        67  4 Dec 12:28 slave-bin.index
+    -rw-rw----  1 patg  patg       249  4 Dec 12:52 slave-relay-bin.000003
+    -rw-rw----  1 patg  patg        73  4 Dec 12:28 slave-relay-bin.index
+    -rw-r--r--  1 patg  patg      1349  4 Dec 12:51 slave.err
+    -rw-rw----  1 patg  patg     96206  4 Dec 12:52 slave.log
+    -rw-rw----  1 patg  patg  15706355  4 Dec 12:51 slave.trace
+    -rw-r--r--  1 patg  patg         0  4 Dec 12:51 warnings
+
+    Of course, again, you can tail the trace log:
+
+    tail -f mysql-test/var/log/master.trace |grep ha_fed
+
+    As well as the slave query log:
+
+    tail -f mysql-test/var/log/slave.log
+
+    Files that comprise the test suit
+    ---------------------------------
+    mysql-test/t/federated.test
+    mysql-test/r/federated.result
+    mysql-test/r/have_federated_db.require
+    mysql-test/include/have_federated_db.inc
+
+
+    Other tidbits
+    -------------
+
+    These were the files that were modified or created for this
+    Federated handler to work:
+
+    ./configure.in
+    ./sql/Makefile.am
+    ./config/ac_macros/ha_federated.m4
+    ./sql/handler.cc
+    ./sql/mysqld.cc
+    ./sql/set_var.cc
+    ./sql/field.h
+    ./sql/sql_string.h
+    ./mysql-test/mysql-test-run(.sh)
+    ./mysql-test/t/federated.test
+    ./mysql-test/r/federated.result
+    ./mysql-test/r/have_federated_db.require
+    ./mysql-test/include/have_federated_db.inc
+    ./sql/ha_federated.cc
+    ./sql/ha_federated.h
+
+*/
+
+
+#define MYSQL_SERVER 1
+#include "mysql_priv.h"
+#include <mysql/plugin.h>
+
+#ifdef USE_PRAGMA_IMPLEMENTATION
+#pragma implementation                          // gcc: Class implementation
+#endif
+
+#include "ha_federated.h"
+
+#include "m_string.h"
+
+#include <mysql/plugin.h>
+
+/* Variables for federated share methods */
+static HASH federated_open_tables;              // To track open tables
+pthread_mutex_t federated_mutex;                // To init the hash
+static int federated_init= FALSE;               // Checking the state of hash
+
+/* Variables used when chopping off trailing characters */
+static const uint sizeof_trailing_comma= sizeof(", ") - 1;
+static const uint sizeof_trailing_closeparen= sizeof(") ") - 1;
+static const uint sizeof_trailing_and= sizeof(" AND ") - 1;
+static const uint sizeof_trailing_where= sizeof(" WHERE ") - 1;
+
+/* Static declaration for handerton */
+static handler *federated_create_handler(TABLE_SHARE *table,
+                                         MEM_ROOT *mem_root);
+static int federated_commit(THD *thd, bool all);
+static int federated_rollback(THD *thd, bool all);
+
+/* Federated storage engine handlerton */
+
+handlerton federated_hton;
+
+static handler *federated_create_handler(TABLE_SHARE *table,
+                                         MEM_ROOT *mem_root)
+{
+  return new (mem_root) ha_federated(table);
+}
+
+
+/* Function we use in the creation of our hash to get key */
+
+static byte *federated_get_key(FEDERATED_SHARE *share, uint *length,
+                               my_bool not_used __attribute__ ((unused)))
+{
+  *length= share->connect_string_length;
+  return (byte*) share->scheme;
+}
+
+/*
+  Initialize the federated handler.
+
+  SYNOPSIS
+    federated_db_init()
+    void
+
+  RETURN
+    FALSE       OK
+    TRUE        Error
+*/
+
+int federated_db_init()
+{
+  DBUG_ENTER("federated_db_init");
+
+  federated_hton.state= SHOW_OPTION_YES;
+  federated_hton.db_type= DB_TYPE_FEDERATED_DB;
+  federated_hton.commit= federated_commit;
+  federated_hton.rollback= federated_rollback;
+  federated_hton.create= federated_create_handler;
+  federated_hton.panic= federated_db_end;
+  federated_hton.flags= HTON_ALTER_NOT_SUPPORTED;
+
+  if (pthread_mutex_init(&federated_mutex, MY_MUTEX_INIT_FAST))
+    goto error;
+  if (!hash_init(&federated_open_tables, &my_charset_bin, 32, 0, 0,
+                    (hash_get_key) federated_get_key, 0, 0))
+  {
+    federated_init= TRUE;
+    DBUG_RETURN(FALSE);
+  }
+
+  VOID(pthread_mutex_destroy(&federated_mutex));
+error:
+  have_federated_db= SHOW_OPTION_DISABLED;	// If we couldn't use handler
+  DBUG_RETURN(TRUE);
+}
+
+
+/*
+  Release the federated handler.
+
+  SYNOPSIS
+    federated_db_end()
+
+  RETURN
+    FALSE       OK
+*/
+
+int federated_db_end(ha_panic_function type)
+{
+  if (federated_init)
+  {
+    hash_free(&federated_open_tables);
+    VOID(pthread_mutex_destroy(&federated_mutex));
+  }
+  federated_init= 0;
+  return 0;
+}
+
+
+/*
+ Check (in create) whether the tables exists, and that it can be connected to
+
+  SYNOPSIS
+    check_foreign_data_source()
+      share               pointer to FEDERATED share
+      table_create_flag   tells us that ::create is the caller, 
+                          therefore, return CANT_CREATE_FEDERATED_TABLE
+
+  DESCRIPTION
+    This method first checks that the connection information that parse url
+    has populated into the share will be sufficient to connect to the foreign
+    table, and if so, does the foreign table exist.
+*/
+
+static int check_foreign_data_source(FEDERATED_SHARE *share,
+                                     bool table_create_flag)
+{
+  char escaped_table_name[NAME_LEN*2];
+  char query_buffer[FEDERATED_QUERY_BUFFER_SIZE];
+  char error_buffer[FEDERATED_QUERY_BUFFER_SIZE];
+  uint error_code;
+  String query(query_buffer, sizeof(query_buffer), &my_charset_bin);
+  MYSQL *mysql;
+  DBUG_ENTER("ha_federated::check_foreign_data_source");
+
+  /* Zero the length, otherwise the string will have misc chars */
+  query.length(0);
+
+  /* error out if we can't alloc memory for mysql_init(NULL) (per Georg) */
+  if (!(mysql= mysql_init(NULL)))
+    DBUG_RETURN(HA_ERR_OUT_OF_MEM);
+  /* check if we can connect */
+  if (!mysql_real_connect(mysql,
+                          share->hostname,
+                          share->username,
+                          share->password,
+                          share->database,
+                          share->port,
+                          share->socket, 0))
+  {
+    /*
+      we want the correct error message, but it to return
+      ER_CANT_CREATE_FEDERATED_TABLE if called by ::create
+    */
+    error_code= (table_create_flag ?
+                 ER_CANT_CREATE_FEDERATED_TABLE :
+                 ER_CONNECT_TO_FOREIGN_DATA_SOURCE);
+
+    my_sprintf(error_buffer,
+               (error_buffer,
+                "database: '%s'  username: '%s'  hostname: '%s'",
+                share->database, share->username, share->hostname));
+
+    my_error(ER_CONNECT_TO_FOREIGN_DATA_SOURCE, MYF(0), error_buffer);
+    goto error;
+  }
+  else
+  {
+    int escaped_table_name_length= 0;
+    /*
+      Since we do not support transactions at this version, we can let the 
+      client API silently reconnect. For future versions, we will need more 
+      logic to deal with transactions
+    */
+    mysql->reconnect= 1;
+    /*
+      Note: I am not using INORMATION_SCHEMA because this needs to work with 
+      versions prior to 5.0
+      
+      if we can connect, then make sure the table exists 
+
+      the query will be: SELECT * FROM `tablename` WHERE 1=0
+    */
+    query.append(STRING_WITH_LEN("SELECT * FROM `"));
+    escaped_table_name_length=
+      escape_string_for_mysql(&my_charset_bin, (char*)escaped_table_name,
+                            sizeof(escaped_table_name),
+                            share->table_name,
+                            share->table_name_length);
+    query.append(escaped_table_name, escaped_table_name_length);
+    query.append(STRING_WITH_LEN("` WHERE 1=0"));
+
+    if (mysql_real_query(mysql, query.ptr(), query.length()))
+    {
+      error_code= table_create_flag ?
+        ER_CANT_CREATE_FEDERATED_TABLE : ER_FOREIGN_DATA_SOURCE_DOESNT_EXIST;
+      my_sprintf(error_buffer, (error_buffer, "error: %d  '%s'",
+                                mysql_errno(mysql), mysql_error(mysql)));
+
+      my_error(error_code, MYF(0), error_buffer);
+      goto error;
+    }
+  }
+  error_code=0;
+
+error:
+    mysql_close(mysql);
+    DBUG_RETURN(error_code);
+}
+
+
+static int parse_url_error(FEDERATED_SHARE *share, TABLE *table, int error_num)
+{
+  char buf[FEDERATED_QUERY_BUFFER_SIZE];
+  int buf_len;
+  DBUG_ENTER("ha_federated parse_url_error");
+
+  if (share->scheme)
+  {
+    DBUG_PRINT("info",
+               ("error: parse_url. Returning error code %d \
+                freeing share->scheme %lx", error_num, share->scheme));
+    my_free((gptr) share->scheme, MYF(0));
+    share->scheme= 0;
+  }
+  buf_len= min(table->s->connect_string.length,
+               FEDERATED_QUERY_BUFFER_SIZE-1);
+  strmake(buf, table->s->connect_string.str, buf_len);
+  my_error(error_num, MYF(0), buf);
+  DBUG_RETURN(error_num);
+}
+
+/*
+  Parse connection info from table->s->connect_string
+
+  SYNOPSIS
+    parse_url()
+    share               pointer to FEDERATED share
+    table               pointer to current TABLE class
+    table_create_flag   determines what error to throw
+
+  DESCRIPTION
+    Populates the share with information about the connection
+    to the foreign database that will serve as the data source.
+    This string must be specified (currently) in the "comment" field,
+    listed in the CREATE TABLE statement.
+
+    This string MUST be in the format of any of these:
+
+    scheme://username:password@hostname:port/database/table
+    scheme://username@hostname/database/table
+    scheme://username@hostname:port/database/table
+    scheme://username:password@hostname/database/table
+
+  An Example:
+
+  mysql://joe:joespass@192.168.1.111:9308/federated/testtable
+
+  ***IMPORTANT***
+  Currently, only "mysql://" is supported.
+
+  'password' and 'port' are both optional.
+
+  RETURN VALUE
+    0           success
+    error_num   particular error code 
+
+*/
+
+static int parse_url(FEDERATED_SHARE *share, TABLE *table,
+                     uint table_create_flag)
+{
+  uint error_num= (table_create_flag ?
+                   ER_FOREIGN_DATA_STRING_INVALID_CANT_CREATE :
+                   ER_FOREIGN_DATA_STRING_INVALID);
+  DBUG_ENTER("ha_federated::parse_url");
+
+  share->port= 0;
+  share->socket= 0;
+  DBUG_PRINT("info", ("Length: %d", table->s->connect_string.length));
+  DBUG_PRINT("info", ("String: '%.*s'", table->s->connect_string.length, 
+                      table->s->connect_string.str));
+  share->scheme= my_strndup(table->s->connect_string.str,
+                            table->s->connect_string.length,
+                            MYF(0));
+
+  share->connect_string_length= table->s->connect_string.length;
+  DBUG_PRINT("info",("parse_url alloced share->scheme %lx", share->scheme));
+
+  /*
+    remove addition of null terminator and store length
+    for each string  in share
+  */
+  if (!(share->username= strstr(share->scheme, "://")))
+    goto error;
+  share->scheme[share->username - share->scheme]= '\0';
+
+  if (strcmp(share->scheme, "mysql") != 0)
+    goto error;
+
+  share->username+= 3;
+
+  if (!(share->hostname= strchr(share->username, '@')))
+    goto error;
+    
+  share->username[share->hostname - share->username]= '\0';
+  share->hostname++;
+
+  if ((share->password= strchr(share->username, ':')))
+  {
+    share->username[share->password - share->username]= '\0';
+    share->password++;
+    share->username= share->username;
+    /* make sure there isn't an extra / or @ */
+    if ((strchr(share->password, '/') || strchr(share->hostname, '@')))
+      goto error;
+    /*
+      Found that if the string is:
+      user:@hostname:port/database/table
+      Then password is a null string, so set to NULL
+    */
+    if ((share->password[0] == '\0'))
+      share->password= NULL;
+  }
+  else
+    share->username= share->username;
+
+  /* make sure there isn't an extra / or @ */
+  if ((strchr(share->username, '/')) || (strchr(share->hostname, '@')))
+    goto error;
+
+  if (!(share->database= strchr(share->hostname, '/')))
+    goto error;
+  share->hostname[share->database - share->hostname]= '\0';
+  share->database++;
+
+  if ((share->sport= strchr(share->hostname, ':')))
+  {
+    share->hostname[share->sport - share->hostname]= '\0';
+    share->sport++;
+    if (share->sport[0] == '\0')
+      share->sport= NULL;
+    else
+      share->port= atoi(share->sport);
+  }
+
+  if (!(share->table_name= strchr(share->database, '/')))
+    goto error;
+  share->database[share->table_name - share->database]= '\0';
+  share->table_name++;
+
+  share->table_name_length= strlen(share->table_name);
+ 
+  /* make sure there's not an extra / */
+  if ((strchr(share->table_name, '/')))
+    goto error;
+
+  if (share->hostname[0] == '\0')
+    share->hostname= NULL;
+
+  if (!share->port)
+  {
+    if (strcmp(share->hostname, my_localhost) == 0)
+      share->socket= my_strdup(MYSQL_UNIX_ADDR, MYF(0));
+    else
+      share->port= MYSQL_PORT;
+  }
+
+  DBUG_PRINT("info",
+             ("scheme: %s  username: %s  password: %s \
+               hostname: %s  port: %d  database: %s  tablename: %s",
+              share->scheme, share->username, share->password,
+              share->hostname, share->port, share->database,
+              share->table_name));
+
+  DBUG_RETURN(0);
+
+error:
+  DBUG_RETURN(parse_url_error(share, table, error_num));
+}
+
+
+/*****************************************************************************
+** FEDERATED tables
+*****************************************************************************/
+
+ha_federated::ha_federated(TABLE_SHARE *table_arg)
+  :handler(&federated_hton, table_arg),
+  mysql(0), stored_result(0)
+{
+  trx_next= 0;
+}
+
+
+/*
+  Convert MySQL result set row to handler internal format
+
+  SYNOPSIS
+    convert_row_to_internal_format()
+      record    Byte pointer to record
+      row       MySQL result set row from fetchrow()
+      result	Result set to use
+
+  DESCRIPTION
+    This method simply iterates through a row returned via fetchrow with
+    values from a successful SELECT , and then stores each column's value
+    in the field object via the field object pointer (pointing to the table's
+    array of field object pointers). This is how the handler needs the data
+    to be stored to then return results back to the user
+
+  RETURN VALUE
+    0   After fields have had field values stored from record
+*/
+
+uint ha_federated::convert_row_to_internal_format(byte *record,
+                                                  MYSQL_ROW row,
+                                                  MYSQL_RES *result)
+{
+  ulong *lengths;
+  Field **field;
+  my_bitmap_map *old_map= dbug_tmp_use_all_columns(table, table->write_set);
+  DBUG_ENTER("ha_federated::convert_row_to_internal_format");
+
+  lengths= mysql_fetch_lengths(result);
+
+  for (field= table->field; *field; field++, row++, lengths++)
+  {
+    /*
+      index variable to move us through the row at the
+      same iterative step as the field
+    */
+    my_ptrdiff_t old_ptr;
+    old_ptr= (my_ptrdiff_t) (record - table->record[0]);
+    (*field)->move_field_offset(old_ptr);
+    if (!*row)
+      (*field)->set_null();
+    else
+    {
+      if (bitmap_is_set(table->read_set, (*field)->field_index))
+      {
+        (*field)->set_notnull();
+        (*field)->store(*row, *lengths, &my_charset_bin);
+      }
+    }
+    (*field)->move_field_offset(-old_ptr);
+  }
+  dbug_tmp_restore_column_map(table->write_set, old_map);
+  DBUG_RETURN(0);
+}
+
+static bool emit_key_part_name(String *to, KEY_PART_INFO *part)
+{
+  DBUG_ENTER("emit_key_part_name");
+  if (to->append(STRING_WITH_LEN("`")) ||
+      to->append(part->field->field_name) ||
+      to->append(STRING_WITH_LEN("`")))
+    DBUG_RETURN(1);                           // Out of memory
+  DBUG_RETURN(0);
+}
+
+static bool emit_key_part_element(String *to, KEY_PART_INFO *part,
+                                  bool needs_quotes, bool is_like,
+                                  const byte *ptr, uint len)
+{
+  Field *field= part->field;
+  DBUG_ENTER("emit_key_part_element");
+
+  if (needs_quotes && to->append(STRING_WITH_LEN("'")))
+    DBUG_RETURN(1);
+
+  if (part->type == HA_KEYTYPE_BIT)
+  {
+    char buff[STRING_BUFFER_USUAL_SIZE], *buf= buff;
+
+    *buf++= '0';
+    *buf++= 'x';
+    buf= octet2hex(buf, (char*) ptr, len);
+    if (to->append((char*) buff, (uint)(buf - buff)))
+      DBUG_RETURN(1);
+  }
+  else if (part->key_part_flag & HA_BLOB_PART)
+  {
+    String blob;
+    uint blob_length= uint2korr(ptr);
+    blob.set_quick((char*) ptr+HA_KEY_BLOB_LENGTH,
+                   blob_length, &my_charset_bin);
+    if (append_escaped(to, &blob))
+      DBUG_RETURN(1);
+  }
+  else if (part->key_part_flag & HA_VAR_LENGTH_PART)
+  {
+    String varchar;
+    uint var_length= uint2korr(ptr);
+    varchar.set_quick((char*) ptr+HA_KEY_BLOB_LENGTH,
+                      var_length, &my_charset_bin);
+    if (append_escaped(to, &varchar))
+      DBUG_RETURN(1);
+  }
+  else
+  {
+    char strbuff[MAX_FIELD_WIDTH];
+    String str(strbuff, sizeof(strbuff), part->field->charset()), *res;
+
+    res= field->val_str(&str, (char *)ptr);
+
+    if (field->result_type() == STRING_RESULT)
+    {
+      if (append_escaped(to, res))
+        DBUG_RETURN(1);
+    }
+    else if (to->append(res->ptr(), res->length()))
+      DBUG_RETURN(1);
+  }
+
+  if (is_like && to->append(STRING_WITH_LEN("%")))
+    DBUG_RETURN(1);
+
+  if (needs_quotes && to->append(STRING_WITH_LEN("'")))
+    DBUG_RETURN(1);
+
+  DBUG_RETURN(0);
+}
+
+/*
+  Create a WHERE clause based off of values in keys
+  Note: This code was inspired by key_copy from key.cc
+
+  SYNOPSIS
+    create_where_from_key ()
+      to          String object to store WHERE clause
+      key_info    KEY struct pointer
+      key         byte pointer containing key
+      key_length  length of key
+      range_type  0 - no range, 1 - min range, 2 - max range
+                  (see enum range_operation)
+
+  DESCRIPTION
+    Using iteration through all the keys via a KEY_PART_INFO pointer,
+    This method 'extracts' the value of each key in the byte pointer
+    *key, and for each key found, constructs an appropriate WHERE clause
+
+  RETURN VALUE
+    0   After all keys have been accounted for to create the WHERE clause
+    1   No keys found
+
+    Range flags Table per Timour:
+
+   -----------------
+   - start_key:
+     * ">"  -> HA_READ_AFTER_KEY
+     * ">=" -> HA_READ_KEY_OR_NEXT
+     * "="  -> HA_READ_KEY_EXACT
+
+   - end_key:
+     * "<"  -> HA_READ_BEFORE_KEY
+     * "<=" -> HA_READ_AFTER_KEY
+
+   records_in_range:
+   -----------------
+   - start_key:
+     * ">"  -> HA_READ_AFTER_KEY
+     * ">=" -> HA_READ_KEY_EXACT
+     * "="  -> HA_READ_KEY_EXACT
+
+   - end_key:
+     * "<"  -> HA_READ_BEFORE_KEY
+     * "<=" -> HA_READ_AFTER_KEY
+     * "="  -> HA_READ_AFTER_KEY
+
+0 HA_READ_KEY_EXACT,              Find first record else error
+1 HA_READ_KEY_OR_NEXT,            Record or next record
+2 HA_READ_KEY_OR_PREV,            Record or previous
+3 HA_READ_AFTER_KEY,              Find next rec. after key-record
+4 HA_READ_BEFORE_KEY,             Find next rec. before key-record
+5 HA_READ_PREFIX,                 Key which as same prefix
+6 HA_READ_PREFIX_LAST,            Last key with the same prefix
+7 HA_READ_PREFIX_LAST_OR_PREV,    Last or prev key with the same prefix
+
+Flags that I've found:
+
+id, primary key, varchar
+
+id = 'ccccc'
+records_in_range: start_key 0 end_key 3
+read_range_first: start_key 0 end_key NULL
+
+id > 'ccccc'
+records_in_range: start_key 3 end_key NULL
+read_range_first: start_key 3 end_key NULL
+
+id < 'ccccc'
+records_in_range: start_key NULL end_key 4
+read_range_first: start_key NULL end_key 4
+
+id <= 'ccccc'
+records_in_range: start_key NULL end_key 3
+read_range_first: start_key NULL end_key 3
+
+id >= 'ccccc'
+records_in_range: start_key 0 end_key NULL
+read_range_first: start_key 1 end_key NULL
+
+id like 'cc%cc'
+records_in_range: start_key 0 end_key 3
+read_range_first: start_key 1 end_key 3
+
+id > 'aaaaa' and id < 'ccccc'
+records_in_range: start_key 3 end_key 4
+read_range_first: start_key 3 end_key 4
+
+id >= 'aaaaa' and id < 'ccccc';
+records_in_range: start_key 0 end_key 4
+read_range_first: start_key 1 end_key 4
+
+id >= 'aaaaa' and id <= 'ccccc';
+records_in_range: start_key 0 end_key 3
+read_range_first: start_key 1 end_key 3
+
+id > 'aaaaa' and id <= 'ccccc';
+records_in_range: start_key 3 end_key 3
+read_range_first: start_key 3 end_key 3
+
+numeric keys:
+
+id = 4
+index_read_idx: start_key 0 end_key NULL 
+
+id > 4
+records_in_range: start_key 3 end_key NULL
+read_range_first: start_key 3 end_key NULL
+
+id >= 4
+records_in_range: start_key 0 end_key NULL
+read_range_first: start_key 1 end_key NULL
+
+id < 4
+records_in_range: start_key NULL end_key 4
+read_range_first: start_key NULL end_key 4
+
+id <= 4
+records_in_range: start_key NULL end_key 3
+read_range_first: start_key NULL end_key 3
+
+id like 4
+full table scan, select * from
+
+id > 2 and id < 8
+records_in_range: start_key 3 end_key 4
+read_range_first: start_key 3 end_key 4
+
+id >= 2 and id < 8
+records_in_range: start_key 0 end_key 4
+read_range_first: start_key 1 end_key 4
+
+id >= 2 and id <= 8
+records_in_range: start_key 0 end_key 3
+read_range_first: start_key 1 end_key 3
+
+id > 2 and id <= 8
+records_in_range: start_key 3 end_key 3
+read_range_first: start_key 3 end_key 3
+
+multi keys (id int, name varchar, other varchar)
+
+id = 1;
+records_in_range: start_key 0 end_key 3
+read_range_first: start_key 0 end_key NULL
+
+id > 4;
+id > 2 and name = '333'; remote: id > 2
+id > 2 and name > '333'; remote: id > 2
+id > 2 and name > '333' and other < 'ddd'; remote: id > 2 no results
+id > 2 and name >= '333' and other < 'ddd'; remote: id > 2 1 result
+id >= 4 and name = 'eric was here' and other > 'eeee';
+records_in_range: start_key 3 end_key NULL
+read_range_first: start_key 3 end_key NULL
+
+id >= 4;
+id >= 2 and name = '333' and other < 'ddd';
+remote: `id`  >= 2 AND `name`  >= '333';
+records_in_range: start_key 0 end_key NULL
+read_range_first: start_key 1 end_key NULL
+
+id < 4;
+id < 3 and name = '222' and other <= 'ccc'; remote: id < 3
+records_in_range: start_key NULL end_key 4
+read_range_first: start_key NULL end_key 4
+
+id <= 4;
+records_in_range: start_key NULL end_key 3
+read_range_first: start_key NULL end_key 3
+
+id like 4;
+full table scan
+
+id  > 2 and id < 4;
+records_in_range: start_key 3 end_key 4
+read_range_first: start_key 3 end_key 4
+
+id >= 2 and id < 4;
+records_in_range: start_key 0 end_key 4
+read_range_first: start_key 1 end_key 4
+
+id >= 2 and id <= 4;
+records_in_range: start_key 0 end_key 3
+read_range_first: start_key 1 end_key 3
+
+id > 2 and id <= 4;
+id = 6 and name = 'eric was here' and other > 'eeee';
+remote: (`id`  > 6 AND `name`  > 'eric was here' AND `other`  > 'eeee')
+AND (`id`  <= 6) AND ( AND `name`  <= 'eric was here')
+no results
+records_in_range: start_key 3 end_key 3
+read_range_first: start_key 3 end_key 3
+
+Summary:
+
+* If the start key flag is 0 the max key flag shouldn't even be set, 
+  and if it is, the query produced would be invalid.
+* Multipart keys, even if containing some or all numeric columns,
+  are treated the same as non-numeric keys
+
+  If the query is " = " (quotes or not):
+  - records in range start key flag HA_READ_KEY_EXACT,
+    end key flag HA_READ_AFTER_KEY (incorrect)
+  - any other: start key flag HA_READ_KEY_OR_NEXT,
+    end key flag HA_READ_AFTER_KEY (correct)
+
+* 'like' queries (of key)
+  - Numeric, full table scan
+  - Non-numeric
+      records_in_range: start_key 0 end_key 3
+      other : start_key 1 end_key 3
+
+* If the key flag is HA_READ_AFTER_KEY:
+   if start_key, append >
+   if end_key, append <=
+
+* If create_where_key was called by records_in_range:
+
+ - if the key is numeric:
+    start key flag is 0 when end key is NULL, end key flag is 3 or 4
+ - if create_where_key was called by any other function:
+    start key flag is 1 when end key is NULL, end key flag is 3 or 4
+ - if the key is non-numeric, or multipart
+    When the query is an exact match, the start key flag is 0,
+    end key flag is 3 for what should be a no-range condition where
+    you should have 0 and max key NULL, which it is if called by
+    read_range_first
+
+Conclusion:
+
+1. Need logic to determin if a key is min or max when the flag is
+HA_READ_AFTER_KEY, and handle appending correct operator accordingly
+
+2. Need a boolean flag to pass to create_where_from_key, used in the
+switch statement. Add 1 to the flag if:
+  - start key flag is HA_READ_KEY_EXACT and the end key is NULL
+
+*/
+
+bool ha_federated::create_where_from_key(String *to,
+                                         KEY *key_info,
+                                         const key_range *start_key,
+                                         const key_range *end_key,
+                                         bool records_in_range,
+                                         bool eq_range)
+{
+  bool both_not_null=
+    (start_key != NULL && end_key != NULL) ? TRUE : FALSE;
+  const byte *ptr;
+  uint remainder, length;
+  char tmpbuff[FEDERATED_QUERY_BUFFER_SIZE];
+  String tmp(tmpbuff, sizeof(tmpbuff), system_charset_info);
+  const key_range *ranges[2]= { start_key, end_key };
+  my_bitmap_map *old_map;
+  DBUG_ENTER("ha_federated::create_where_from_key");
+
+  tmp.length(0); 
+  if (start_key == NULL && end_key == NULL)
+    DBUG_RETURN(1);
+
+  old_map= dbug_tmp_use_all_columns(table, table->write_set);
+  for (uint i= 0; i <= 1; i++)
+  {
+    bool needs_quotes;
+    KEY_PART_INFO *key_part;
+    if (ranges[i] == NULL)
+      continue;
+
+    if (both_not_null)
+    {
+      if (i > 0)
+        tmp.append(STRING_WITH_LEN(") AND ("));
+      else
+        tmp.append(STRING_WITH_LEN(" ("));
+    }
+
+    for (key_part= key_info->key_part,
+         remainder= key_info->key_parts,
+         length= ranges[i]->length,
+         ptr= ranges[i]->key; ;
+         remainder--,
+         key_part++)
+    {
+      Field *field= key_part->field;
+      uint store_length= key_part->store_length;
+      uint part_length= min(store_length, length);
+      needs_quotes= field->str_needs_quotes();
+      DBUG_DUMP("key, start of loop", (char *) ptr, length);
+
+      if (key_part->null_bit)
+      {
+        if (*ptr++)
+        {
+          if (emit_key_part_name(&tmp, key_part) ||
+              tmp.append(STRING_WITH_LEN(" IS NULL ")))
+            goto err;
+          continue;
+        }
+      }
+
+      if (tmp.append(STRING_WITH_LEN(" (")))
+        goto err;
+
+      switch (ranges[i]->flag) {
+      case HA_READ_KEY_EXACT:
+        DBUG_PRINT("info", ("federated HA_READ_KEY_EXACT %d", i));
+        if (store_length >= length ||
+            !needs_quotes ||
+            key_part->type == HA_KEYTYPE_BIT ||
+            field->result_type() != STRING_RESULT)
+        {
+          if (emit_key_part_name(&tmp, key_part))
+            goto err;
+
+          if (records_in_range)
+          {
+            if (tmp.append(STRING_WITH_LEN(" >= ")))
+              goto err;
+          }
+          else
+          {
+            if (tmp.append(STRING_WITH_LEN(" = ")))
+              goto err;
+          }
+
+          if (emit_key_part_element(&tmp, key_part, needs_quotes, 0, ptr,
+                                    part_length))
+            goto err;
+        }
+        else
+        {
+          /* LIKE */
+          if (emit_key_part_name(&tmp, key_part) ||
+              tmp.append(STRING_WITH_LEN(" LIKE ")) ||
+              emit_key_part_element(&tmp, key_part, needs_quotes, 1, ptr,
+                                    part_length))
+            goto err;
+        }
+        break;
+      case HA_READ_AFTER_KEY:
+        if (eq_range)
+        {
+          if (tmp.append("1=1"))                // Dummy
+            goto err;
+          break;
+        }
+        DBUG_PRINT("info", ("federated HA_READ_AFTER_KEY %d", i));
+        if (store_length >= length) /* end key */
+        {
+          if (emit_key_part_name(&tmp, key_part))
+            goto err;
+
+          if (i > 0) /* end key */
+          {
+            if (tmp.append(STRING_WITH_LEN(" <= ")))
+              goto err;
+          }
+          else /* start key */
+          {
+            if (tmp.append(STRING_WITH_LEN(" > ")))
+              goto err;
+          }
+
+          if (emit_key_part_element(&tmp, key_part, needs_quotes, 0, ptr,
+                                    part_length))
+          {
+            goto err;
+          }
+          break;
+        }
+      case HA_READ_KEY_OR_NEXT:
+        DBUG_PRINT("info", ("federated HA_READ_KEY_OR_NEXT %d", i));
+        if (emit_key_part_name(&tmp, key_part) ||
+            tmp.append(STRING_WITH_LEN(" >= ")) ||
+            emit_key_part_element(&tmp, key_part, needs_quotes, 0, ptr,
+              part_length))
+          goto err;
+        break;
+      case HA_READ_BEFORE_KEY:
+        DBUG_PRINT("info", ("federated HA_READ_BEFORE_KEY %d", i));
+        if (store_length >= length)
+        {
+          if (emit_key_part_name(&tmp, key_part) ||
+              tmp.append(STRING_WITH_LEN(" < ")) ||
+              emit_key_part_element(&tmp, key_part, needs_quotes, 0, ptr,
+                                    part_length))
+            goto err;
+          break;
+        }
+      case HA_READ_KEY_OR_PREV:
+        DBUG_PRINT("info", ("federated HA_READ_KEY_OR_PREV %d", i));
+        if (emit_key_part_name(&tmp, key_part) ||
+            tmp.append(STRING_WITH_LEN(" <= ")) ||
+            emit_key_part_element(&tmp, key_part, needs_quotes, 0, ptr,
+                                  part_length))
+          goto err;
+        break;
+      default:
+        DBUG_PRINT("info",("cannot handle flag %d", ranges[i]->flag));
+        goto err;
+      }
+      if (tmp.append(STRING_WITH_LEN(") ")))
+        goto err;
+
+next_loop:
+      if (store_length >= length)
+        break;
+      DBUG_PRINT("info", ("remainder %d", remainder));
+      DBUG_ASSERT(remainder > 1);
+      length-= store_length;
+      ptr+= store_length;
+      if (tmp.append(STRING_WITH_LEN(" AND ")))
+        goto err;
+
+      DBUG_PRINT("info",
+                 ("create_where_from_key WHERE clause: %s",
+                  tmp.c_ptr_quick()));
+    }
+  }
+  dbug_tmp_restore_column_map(table->write_set, old_map);
+
+  if (both_not_null)
+    if (tmp.append(STRING_WITH_LEN(") ")))
+      DBUG_RETURN(1);
+
+  if (to->append(STRING_WITH_LEN(" WHERE ")))
+    DBUG_RETURN(1);
+
+  if (to->append(tmp))
+    DBUG_RETURN(1);
+
+  DBUG_RETURN(0);
+
+err:
+  dbug_tmp_restore_column_map(table->write_set, old_map);
+  DBUG_RETURN(1);
+}
+
+/*
+  Example of simple lock controls. The "share" it creates is structure we will
+  pass to each federated 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 FEDERATED_SHARE *get_share(const char *table_name, TABLE *table)
+{
+  char *select_query;
+  char query_buffer[FEDERATED_QUERY_BUFFER_SIZE];
+  Field **field;
+  String query(query_buffer, sizeof(query_buffer), &my_charset_bin);
+  FEDERATED_SHARE *share= NULL, tmp_share;
+  /*
+    In order to use this string, we must first zero it's length,
+    or it will contain garbage
+  */
+  query.length(0);
+
+  pthread_mutex_lock(&federated_mutex);
+
+  if (parse_url(&tmp_share, table, 0))
+    goto error;
+
+  /* TODO: change tmp_share.scheme to LEX_STRING object */
+  if (!(share= (FEDERATED_SHARE *) hash_search(&federated_open_tables,
+                                               (byte*) tmp_share.scheme,
+                                               tmp_share.
+                                               connect_string_length)))
+  {
+    query.set_charset(system_charset_info);
+    query.append(STRING_WITH_LEN("SELECT "));
+    for (field= table->field; *field; field++)
+    {
+      query.append(STRING_WITH_LEN("`"));
+      query.append((*field)->field_name);
+      query.append(STRING_WITH_LEN("`, "));
+    }
+    /* chops off trailing comma */
+    query.length(query.length() - sizeof_trailing_comma);
+
+    query.append(STRING_WITH_LEN(" FROM `"));
+
+    if (!(share= (FEDERATED_SHARE *)
+          my_multi_malloc(MYF(MY_WME),
+                          &share, sizeof(*share),
+                          &select_query,
+                          query.length()+table->s->connect_string.length+1,
+                          NullS)))
+      goto error;
+
+    memcpy(share, &tmp_share, sizeof(tmp_share));
+
+    share->table_name_length= strlen(share->table_name);
+    /* TODO: share->table_name to LEX_STRING object */
+    query.append(share->table_name, share->table_name_length);
+    query.append(STRING_WITH_LEN("`"));
+    share->select_query= select_query;
+    strmov(share->select_query, query.ptr());
+    share->use_count= 0;
+    DBUG_PRINT("info",
+               ("share->select_query %s", share->select_query));
+
+    if (my_hash_insert(&federated_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(&federated_mutex);
+
+  return share;
+
+error:
+  pthread_mutex_unlock(&federated_mutex);
+  my_free((gptr) tmp_share.scheme, MYF(MY_ALLOW_ZERO_PTR));
+  my_free((gptr) share, MYF(MY_ALLOW_ZERO_PTR));
+  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(FEDERATED_SHARE *share)
+{
+  DBUG_ENTER("free_share");
+
+  pthread_mutex_lock(&federated_mutex);
+  if (!--share->use_count)
+  {
+    hash_delete(&federated_open_tables, (byte*) share);
+    my_free((gptr) share->scheme, MYF(MY_ALLOW_ZERO_PTR));
+    my_free((gptr) share->socket, MYF(MY_ALLOW_ZERO_PTR));
+    thr_lock_delete(&share->lock);
+    VOID(pthread_mutex_destroy(&share->mutex));
+    my_free((gptr) share, MYF(0));
+  }
+  pthread_mutex_unlock(&federated_mutex);
+
+  DBUG_RETURN(0);
+}
+
+
+ha_rows ha_federated::records_in_range(uint inx, key_range *start_key,
+                                   key_range *end_key)
+{
+  /*
+
+  We really want indexes to be used as often as possible, therefore
+  we just need to hard-code the return value to a very low number to
+  force the issue
+
+*/
+  DBUG_ENTER("ha_federated::records_in_range");
+  DBUG_RETURN(FEDERATED_RECORDS_IN_RANGE);
+}
+/*
+  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.
+*/
+
+const char **ha_federated::bas_ext() const
+{
+  static const char *ext[]=
+  {
+    NullS
+  };
+  return ext;
+}
+
+
+/*
+  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_federated::open(const char *name, int mode, uint test_if_locked)
+{
+  DBUG_ENTER("ha_federated::open");
+
+  if (!(share= get_share(name, table)))
+    DBUG_RETURN(1);
+  thr_lock_data_init(&share->lock, &lock, NULL);
+
+  /* Connect to foreign database mysql_real_connect() */
+  mysql= mysql_init(0);
+  if (!mysql || !mysql_real_connect(mysql,
+                                   share->hostname,
+                                   share->username,
+                                   share->password,
+                                   share->database,
+                                   share->port,
+                                   share->socket, 0))
+  {
+    free_share(share);
+    DBUG_RETURN(stash_remote_error());
+  }
+  /*
+    Since we do not support transactions at this version, we can let the client
+    API silently reconnect. For future versions, we will need more logic to
+    deal with transactions
+  */
+  mysql->reconnect= 1;
+
+  ref_length= (table->s->primary_key != MAX_KEY ?
+               table->key_info[table->s->primary_key].key_length :
+               table->s->reclength);
+  DBUG_PRINT("info", ("ref_length: %u", ref_length));
+
+  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_federated::close(void)
+{
+  int retval;
+  DBUG_ENTER("ha_federated::close");
+
+  /* free the result set */
+  if (stored_result)
+  {
+    mysql_free_result(stored_result);
+    stored_result= 0;
+  }
+  /* Disconnect from mysql */
+  if (mysql)                                    // QQ is this really needed
+    mysql_close(mysql);
+  retval= free_share(share);
+  DBUG_RETURN(retval);
+
+}
+
+/*
+
+  Checks if a field in a record is SQL NULL.
+
+  SYNOPSIS
+    field_in_record_is_null()
+      table     TABLE pointer, MySQL table object
+      field     Field pointer, MySQL field object
+      record    char pointer, contains record
+
+    DESCRIPTION
+      This method uses the record format information in table to track
+      the null bit in record.
+
+    RETURN VALUE
+      1    if NULL
+      0    otherwise
+*/
+
+inline uint field_in_record_is_null(TABLE *table,
+                                    Field *field,
+                                    char *record)
+{
+  int null_offset;
+  DBUG_ENTER("ha_federated::field_in_record_is_null");
+
+  if (!field->null_ptr)
+    DBUG_RETURN(0);
+
+  null_offset= (uint) ((char*)field->null_ptr - (char*)table->record[0]);
+
+  if (record[null_offset] & field->null_bit)
+    DBUG_RETURN(1);
+
+  DBUG_RETURN(0);
+}
+
+
+/*
+  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++)
+  {
+    ...
+  }
+
+  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_federated::write_row(byte *buf)
+{
+  /*
+    I need a bool again, in 5.0, I used table->s->fields to accomplish this.
+    This worked as a flag that says there are fields with values or not.
+    In 5.1, this value doesn't work the same, and I end up with the code
+    truncating open parenthesis:
+
+    the statement "INSERT INTO t1 VALUES ()" ends up being first built
+    in two strings
+      "INSERT INTO t1 ("
+      and
+      " VALUES ("
+
+    If there are fields with values, they get appended, with commas, and 
+    the last loop, a trailing comma is there
+
+    "INSERT INTO t1 ( col1, col2, colN, "
+
+    " VALUES ( 'val1', 'val2', 'valN', "
+
+    Then, if there are fields, it should decrement the string by ", " length.
+
+    "INSERT INTO t1 ( col1, col2, colN"
+    " VALUES ( 'val1', 'val2', 'valN'"
+
+    Then it adds a close paren to both - if there are fields
+
+    "INSERT INTO t1 ( col1, col2, colN)"
+    " VALUES ( 'val1', 'val2', 'valN')"
+
+    Then appends both together
+    "INSERT INTO t1 ( col1, col2, colN) VALUES ( 'val1', 'val2', 'valN')"
+
+    So... the problem, is if you have the original statement:
+
+    "INSERT INTO t1 VALUES ()"
+
+    Which is legitimate, but if the code thinks there are fields
+
+    "INSERT INTO t1 ("
+    " VALUES ( "
+
+    If the field flag is set, but there are no commas, reduces the 
+    string by strlen(", ")
+
+    "INSERT INTO t1 "
+    " VALUES "
+
+    Then adds the close parenthesis
+
+    "INSERT INTO t1  )"
+    " VALUES  )"
+
+    So, I have to use a bool as before, set in the loop where fields and commas
+    are appended to the string
+  */
+  my_bool commas_added= FALSE;
+  char insert_buffer[FEDERATED_QUERY_BUFFER_SIZE];
+  char values_buffer[FEDERATED_QUERY_BUFFER_SIZE];
+  char insert_field_value_buffer[STRING_BUFFER_USUAL_SIZE];
+  Field **field;
+
+  /* The main insert query string */
+  String insert_string(insert_buffer, sizeof(insert_buffer), &my_charset_bin);
+  /* The string containing the values to be added to the insert */
+  String values_string(values_buffer, sizeof(values_buffer), &my_charset_bin);
+  /* The actual value of the field, to be added to the values_string */
+  String insert_field_value_string(insert_field_value_buffer,
+                                   sizeof(insert_field_value_buffer),
+                                   &my_charset_bin);
+  my_bitmap_map *old_map= dbug_tmp_use_all_columns(table, table->read_set);
+  DBUG_ENTER("ha_federated::write_row");
+
+  values_string.length(0);
+  insert_string.length(0);
+  insert_field_value_string.length(0);
+  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();
+
+  /*
+    start both our field and field values strings
+  */
+  insert_string.append(STRING_WITH_LEN("INSERT INTO `"));
+  insert_string.append(share->table_name, share->table_name_length);
+  insert_string.append('`');
+  insert_string.append(STRING_WITH_LEN(" ("));
+
+  values_string.append(STRING_WITH_LEN(" VALUES "));
+  values_string.append(STRING_WITH_LEN(" ("));
+
+  /*
+    loop through the field pointer array, add any fields to both the values
+    list and the fields list that is part of the write set
+  */
+  for (field= table->field; *field; field++)
+  {
+    if (bitmap_is_set(table->write_set, (*field)->field_index))
+    {
+      commas_added= TRUE;
+      if ((*field)->is_null())
+        values_string.append(STRING_WITH_LEN(" NULL "));
+      else
+      {
+        bool needs_quote= (*field)->str_needs_quotes();
+        (*field)->val_str(&insert_field_value_string);
+        if (needs_quote)
+          values_string.append('\'');
+        insert_field_value_string.print(&values_string);
+        if (needs_quote)
+          values_string.append('\'');
+
+        insert_field_value_string.length(0);
+      }
+      /* append the field name */
+      insert_string.append((*field)->field_name);
+
+      /* append commas between both fields and fieldnames */
+      /*
+        unfortunately, we can't use the logic if *(fields + 1) to
+        make the following appends conditional as we don't know if the
+        next field is in the write set
+      */
+      insert_string.append(STRING_WITH_LEN(", "));
+      values_string.append(STRING_WITH_LEN(", "));
+    }
+  }
+  dbug_tmp_restore_column_map(table->read_set, old_map);
+
+  /*
+    if there were no fields, we don't want to add a closing paren
+    AND, we don't want to chop off the last char '('
+    insert will be "INSERT INTO t1 VALUES ();"
+  */
+  if (commas_added)
+  {
+    insert_string.length(insert_string.length() - sizeof_trailing_comma);
+    /* chops off leading commas */
+    values_string.length(values_string.length() - sizeof_trailing_comma);
+    insert_string.append(STRING_WITH_LEN(") "));
+  }
+  else
+  {
+    /* chops off trailing ) */
+    insert_string.length(insert_string.length() - sizeof_trailing_closeparen);
+  }
+
+  /* we always want to append this, even if there aren't any fields */
+  values_string.append(STRING_WITH_LEN(") "));
+
+  /* add the values */
+  insert_string.append(values_string);
+
+  if (mysql_real_query(mysql, insert_string.ptr(), insert_string.length()))
+  {
+    DBUG_RETURN(stash_remote_error());
+  }
+  /*
+    If the table we've just written a record to contains an auto_increment
+    field, then store the last_insert_id() value from the foreign server
+  */
+  if (table->next_number_field)
+    update_auto_increment();
+
+  DBUG_RETURN(0);
+}
+
+/*
+  ha_federated::update_auto_increment
+
+  This method ensures that last_insert_id() works properly. What it simply does
+  is calls last_insert_id() on the foreign database immediately after insert
+  (if the table has an auto_increment field) and sets the insert id via
+  thd->insert_id(ID)).
+*/
+void ha_federated::update_auto_increment(void)
+{
+  THD *thd= current_thd;
+  DBUG_ENTER("ha_federated::update_auto_increment");
+
+  thd->first_successful_insert_id_in_cur_stmt= 
+    mysql->last_used_con->insert_id;
+  DBUG_PRINT("info",("last_insert_id %d", stats.auto_increment_value));
+
+  DBUG_VOID_RETURN;
+}
+
+int ha_federated::optimize(THD* thd, HA_CHECK_OPT* check_opt)
+{
+  char query_buffer[STRING_BUFFER_USUAL_SIZE];
+  String query(query_buffer, sizeof(query_buffer), &my_charset_bin);
+  DBUG_ENTER("ha_federated::optimize");
+  
+  query.length(0);
+
+  query.set_charset(system_charset_info);
+  query.append(STRING_WITH_LEN("OPTIMIZE TABLE `"));
+  query.append(share->table_name, share->table_name_length);
+  query.append(STRING_WITH_LEN("`"));
+
+  if (mysql_real_query(mysql, query.ptr(), query.length()))
+  {
+    DBUG_RETURN(stash_remote_error());
+  }
+
+  DBUG_RETURN(0);
+}
+
+
+int ha_federated::repair(THD* thd, HA_CHECK_OPT* check_opt)
+{
+  char query_buffer[STRING_BUFFER_USUAL_SIZE];
+  String query(query_buffer, sizeof(query_buffer), &my_charset_bin);
+  DBUG_ENTER("ha_federated::repair");
+
+  query.length(0);
+
+  query.set_charset(system_charset_info);
+  query.append(STRING_WITH_LEN("REPAIR TABLE `"));
+  query.append(share->table_name, share->table_name_length);
+  query.append(STRING_WITH_LEN("`"));
+  if (check_opt->flags & T_QUICK)
+    query.append(STRING_WITH_LEN(" QUICK"));
+  if (check_opt->flags & T_EXTEND)
+    query.append(STRING_WITH_LEN(" EXTENDED"));
+  if (check_opt->sql_flags & TT_USEFRM)
+    query.append(STRING_WITH_LEN(" USE_FRM"));
+
+  if (mysql_real_query(mysql, query.ptr(), query.length()))
+  {
+    DBUG_RETURN(stash_remote_error());
+  }
+
+  DBUG_RETURN(0);
+}
+
+
+/*
+  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 guaranteed.
+  Currently new_data will not have an updated auto_increament record, or
+  and updated timestamp field. You can do these for federated by doing these:
+  if (table->timestamp_on_update_now)
+    update_timestamp(new_row+table->timestamp_on_update_now-1);
+  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_federated::update_row(const byte *old_data, byte *new_data)
+{
+  /*
+    This used to control how the query was built. If there was a
+    primary key, the query would be built such that there was a where
+    clause with only that column as the condition. This is flawed,
+    because if we have a multi-part primary key, it would only use the
+    first part! We don't need to do this anyway, because
+    read_range_first will retrieve the correct record, which is what
+    is used to build the WHERE clause. We can however use this to
+    append a LIMIT to the end if there is NOT a primary key. Why do
+    this? Because we only are updating one record, and LIMIT enforces
+    this.
+  */
+  bool has_a_primary_key= test(table->s->primary_key != MAX_KEY);
+  
+  /*
+    buffers for following strings
+  */
+  char field_value_buffer[STRING_BUFFER_USUAL_SIZE];
+  char update_buffer[FEDERATED_QUERY_BUFFER_SIZE];
+  char where_buffer[FEDERATED_QUERY_BUFFER_SIZE];
+
+  /* Work area for field values */
+  String field_value(field_value_buffer, sizeof(field_value_buffer),
+                     &my_charset_bin);
+  /* stores the update query */
+  String update_string(update_buffer,
+                       sizeof(update_buffer),
+                       &my_charset_bin);
+  /* stores the WHERE clause */
+  String where_string(where_buffer,
+                      sizeof(where_buffer),
+                      &my_charset_bin);
+  DBUG_ENTER("ha_federated::update_row");
+  /*
+    set string lengths to 0 to avoid misc chars in string
+  */
+  field_value.length(0);
+  update_string.length(0);
+  where_string.length(0);
+
+  update_string.append(STRING_WITH_LEN("UPDATE `"));
+  update_string.append(share->table_name);
+  update_string.append(STRING_WITH_LEN("` SET "));
+
+  /*
+    In this loop, we want to match column names to values being inserted
+    (while building INSERT statement).
+
+    Iterate through table->field (new data) and share->old_field (old_data)
+    using the same index to create an SQL UPDATE statement. New data is
+    used to create SET field=value and old data is used to create WHERE
+    field=oldvalue
+  */
+
+  for (Field **field= table->field; *field; field++)
+  {
+    if (bitmap_is_set(table->write_set, (*field)->field_index))
+    {
+      update_string.append((*field)->field_name);
+      update_string.append(STRING_WITH_LEN(" = "));
+
+      if ((*field)->is_null())
+        update_string.append(STRING_WITH_LEN(" NULL "));
+      else
+      {
+        /* otherwise = */
+        my_bitmap_map *old_map= tmp_use_all_columns(table, table->read_set);
+        bool needs_quote= (*field)->str_needs_quotes();
+	(*field)->val_str(&field_value);
+        if (needs_quote)
+          update_string.append('\'');
+        field_value.print(&update_string);
+        if (needs_quote)
+          update_string.append('\'');
+        field_value.length(0);
+        tmp_restore_column_map(table->read_set, old_map);
+      }
+      update_string.append(STRING_WITH_LEN(", "));
+    }
+
+    if (bitmap_is_set(table->read_set, (*field)->field_index))
+    {
+      where_string.append((*field)->field_name);
+      if (field_in_record_is_null(table, *field, (char*) old_data))
+        where_string.append(STRING_WITH_LEN(" IS NULL "));
+      else
+      {
+        bool needs_quote= (*field)->str_needs_quotes();
+        where_string.append(STRING_WITH_LEN(" = "));
+        (*field)->val_str(&field_value,
+                          (char*) (old_data + (*field)->offset()));
+        if (needs_quote)
+          where_string.append('\'');
+        field_value.print(&where_string);
+        if (needs_quote)
+          where_string.append('\'');
+        field_value.length(0);
+      }
+      where_string.append(STRING_WITH_LEN(" AND "));
+    }
+  }
+
+  /* Remove last ', '. This works as there must be at least on updated field */
+  update_string.length(update_string.length() - sizeof_trailing_comma);
+
+  if (where_string.length())
+  {
+    /* chop off trailing AND */
+    where_string.length(where_string.length() - sizeof_trailing_and);
+    update_string.append(STRING_WITH_LEN(" WHERE "));
+    update_string.append(where_string);
+  }
+
+  /*
+    If this table has not a primary key, then we could possibly
+    update multiple rows. We want to make sure to only update one!
+  */
+  if (!has_a_primary_key)
+    update_string.append(STRING_WITH_LEN(" LIMIT 1"));
+
+  if (mysql_real_query(mysql, update_string.ptr(), update_string.length()))
+  {
+    DBUG_RETURN(stash_remote_error());
+  }
+  DBUG_RETURN(0);
+}
+
+/*
+  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_federated::delete_row(const byte *buf)
+{
+  char delete_buffer[FEDERATED_QUERY_BUFFER_SIZE];
+  char data_buffer[FEDERATED_QUERY_BUFFER_SIZE];
+  String delete_string(delete_buffer, sizeof(delete_buffer), &my_charset_bin);
+  String data_string(data_buffer, sizeof(data_buffer), &my_charset_bin);
+  uint found= 0;
+  DBUG_ENTER("ha_federated::delete_row");
+
+  delete_string.length(0);
+  delete_string.append(STRING_WITH_LEN("DELETE FROM `"));
+  delete_string.append(share->table_name);
+  delete_string.append(STRING_WITH_LEN("` WHERE "));
+
+  for (Field **field= table->field; *field; field++)
+  {
+    Field *cur_field= *field;
+    found++;
+    if (bitmap_is_set(table->read_set, cur_field->field_index))
+    {
+      data_string.length(0);
+      delete_string.append(cur_field->field_name);
+      if (cur_field->is_null())
+      {
+        delete_string.append(STRING_WITH_LEN(" IS NULL "));
+      }
+      else
+      {
+        bool needs_quote= cur_field->str_needs_quotes();
+        delete_string.append(STRING_WITH_LEN(" = "));
+        cur_field->val_str(&data_string);
+        if (needs_quote)
+          delete_string.append('\'');
+        data_string.print(&delete_string);
+        if (needs_quote)
+          delete_string.append('\'');
+      }
+      delete_string.append(STRING_WITH_LEN(" AND "));
+    }
+  }
+
+  // Remove trailing AND
+  delete_string.length(delete_string.length() - sizeof_trailing_and);
+  if (!found)
+    delete_string.length(delete_string.length() - sizeof_trailing_where);
+
+  delete_string.append(STRING_WITH_LEN(" LIMIT 1"));
+  DBUG_PRINT("info",
+             ("Delete sql: %s", delete_string.c_ptr_quick()));
+  if (mysql_real_query(mysql, delete_string.ptr(), delete_string.length()))
+  {
+    DBUG_RETURN(stash_remote_error());
+  }
+  stats.deleted+= mysql->affected_rows;
+  stats.records-= mysql->affected_rows;
+  DBUG_PRINT("info",
+             ("rows deleted %d rows deleted for all time %d",
+             int(mysql->affected_rows), stats.deleted));
+
+  DBUG_RETURN(0);
+}
+
+
+/*
+  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. This method, which is called in the case of an SQL statement having
+  a WHERE clause on a non-primary key index, simply calls index_read_idx.
+*/
+
+int ha_federated::index_read(byte *buf, const byte *key,
+                             uint key_len, ha_rkey_function find_flag)
+{
+  DBUG_ENTER("ha_federated::index_read");
+
+  if (stored_result)
+    mysql_free_result(stored_result);
+  DBUG_RETURN(index_read_idx_with_result_set(buf, active_index, key,
+                                             key_len, find_flag,
+                                             &stored_result));
+}
+
+
+/*
+  Positions an index cursor to the index specified in key. Fetches the
+  row if any.  This is only used to read whole keys.
+
+  This method is called via index_read in the case of a WHERE clause using
+  a primary key index OR is called DIRECTLY when the WHERE clause
+  uses a PRIMARY KEY index.
+
+  NOTES
+    This uses an internal result set that is deleted before function
+    returns.  We need to be able to be calable from ha_rnd_pos()
+*/
+
+int ha_federated::index_read_idx(byte *buf, uint index, const byte *key,
+                                 uint key_len, enum ha_rkey_function find_flag)
+{
+  int retval;
+  MYSQL_RES *mysql_result;
+  DBUG_ENTER("ha_federated::index_read_idx");
+
+  if ((retval= index_read_idx_with_result_set(buf, index, key,
+                                              key_len, find_flag,
+                                              &mysql_result)))
+    DBUG_RETURN(retval);
+  mysql_free_result(mysql_result);
+  DBUG_RETURN(retval);
+}
+
+
+/*
+  Create result set for rows matching query and return first row
+
+  RESULT
+    0	ok     In this case *result will contain the result set
+	       table->status == 0 
+    #   error  In this case *result will contain 0
+               table->status == STATUS_NOT_FOUND
+*/
+
+int ha_federated::index_read_idx_with_result_set(byte *buf, uint index,
+                                                 const byte *key,
+                                                 uint key_len,
+                                                 ha_rkey_function find_flag,
+                                                 MYSQL_RES **result)
+{
+  int retval;
+  char error_buffer[FEDERATED_QUERY_BUFFER_SIZE];
+  char index_value[STRING_BUFFER_USUAL_SIZE];
+  char sql_query_buffer[FEDERATED_QUERY_BUFFER_SIZE];
+  String index_string(index_value,
+                      sizeof(index_value),
+                      &my_charset_bin);
+  String sql_query(sql_query_buffer,
+                   sizeof(sql_query_buffer),
+                   &my_charset_bin);
+  key_range range;
+  DBUG_ENTER("ha_federated::index_read_idx_with_result_set");
+
+  *result= 0;                                   // In case of errors
+  index_string.length(0);
+  sql_query.length(0);
+  statistic_increment(table->in_use->status_var.ha_read_key_count,
+                      &LOCK_status);
+
+  sql_query.append(share->select_query);
+
+  range.key= key;
+  range.length= key_len;
+  range.flag= find_flag;
+  create_where_from_key(&index_string,
+                        &table->key_info[index],
+                        &range,
+                        NULL, 0, 0);
+  sql_query.append(index_string);
+
+  if (mysql_real_query(mysql, sql_query.ptr(), sql_query.length()))
+  {
+    my_sprintf(error_buffer, (error_buffer, "error: %d '%s'",
+                              mysql_errno(mysql), mysql_error(mysql)));
+    retval= ER_QUERY_ON_FOREIGN_DATA_SOURCE;
+    goto error;
+  }
+  if (!(*result= mysql_store_result(mysql)))
+  {
+    retval= HA_ERR_END_OF_FILE;
+    goto error;
+  }
+  if (!(retval= read_next(buf, *result)))
+    DBUG_RETURN(retval);
+
+  mysql_free_result(*result);
+  *result= 0;
+  table->status= STATUS_NOT_FOUND;
+  DBUG_RETURN(retval);
+
+error:
+  table->status= STATUS_NOT_FOUND;
+  my_error(retval, MYF(0), error_buffer);
+  DBUG_RETURN(retval);
+}
+
+
+/* Initialized at each key walk (called multiple times unlike rnd_init()) */
+
+int ha_federated::index_init(uint keynr, bool sorted)
+{
+  DBUG_ENTER("ha_federated::index_init");
+  DBUG_PRINT("info", ("table: '%s'  key: %u", table->s->table_name, keynr));
+  active_index= keynr;
+  DBUG_RETURN(0);
+}
+
+
+/*
+  Read first range
+*/
+
+int ha_federated::read_range_first(const key_range *start_key,
+                                   const key_range *end_key,
+                                   bool eq_range, bool sorted)
+{
+  char sql_query_buffer[FEDERATED_QUERY_BUFFER_SIZE];
+  int retval;
+  String sql_query(sql_query_buffer,
+                   sizeof(sql_query_buffer),
+                   &my_charset_bin);
+  DBUG_ENTER("ha_federated::read_range_first");
+
+  DBUG_ASSERT(!(start_key == NULL && end_key == NULL));
+
+  sql_query.length(0);
+  sql_query.append(share->select_query);
+  create_where_from_key(&sql_query,
+                        &table->key_info[active_index],
+                        start_key, end_key, 0, eq_range);
+
+  if (stored_result)
+  {
+    mysql_free_result(stored_result);
+    stored_result= 0;
+  }
+  if (mysql_real_query(mysql, sql_query.ptr(), sql_query.length()))
+  {
+    retval= ER_QUERY_ON_FOREIGN_DATA_SOURCE;
+    goto error;
+  }
+  sql_query.length(0);
+
+  if (!(stored_result= mysql_store_result(mysql)))
+  {
+    retval= HA_ERR_END_OF_FILE;
+    goto error;
+  }
+
+  retval= read_next(table->record[0], stored_result);
+  DBUG_RETURN(retval);
+
+error:
+  table->status= STATUS_NOT_FOUND;
+  DBUG_RETURN(retval);
+}
+
+
+int ha_federated::read_range_next()
+{
+  int retval;
+  DBUG_ENTER("ha_federated::read_range_next");
+  retval= rnd_next(table->record[0]);
+  DBUG_RETURN(retval);
+}
+
+
+/* Used to read forward through the index.  */
+int ha_federated::index_next(byte *buf)
+{
+  DBUG_ENTER("ha_federated::index_next");
+  statistic_increment(table->in_use->status_var.ha_read_next_count,
+		      &LOCK_status);
+  DBUG_RETURN(read_next(buf, stored_result));
+}
+
+
+/*
+  rnd_init() is called when the system wants the storage engine to do a table
+  scan.
+
+  This is the method that gets data for the SELECT calls.
+
+  See the federated 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_federated::rnd_init(bool scan)
+{
+  DBUG_ENTER("ha_federated::rnd_init");
+  /*
+    The use of the 'scan' flag is incredibly important for this handler
+    to work properly, especially with updates containing WHERE clauses
+    using indexed columns.
+
+    When the initial query contains a WHERE clause of the query using an
+    indexed column, it's index_read_idx that selects the exact record from
+    the foreign database.
+
+    When there is NO index in the query, either due to not having a WHERE
+    clause, or the WHERE clause is using columns that are not indexed, a
+    'full table scan' done by rnd_init, which in this situation simply means
+    a 'select * from ...' on the foreign table.
+
+    In other words, this 'scan' flag gives us the means to ensure that if
+    there is an index involved in the query, we want index_read_idx to
+    retrieve the exact record (scan flag is 0), and do not  want rnd_init
+    to do a 'full table scan' and wipe out that result set.
+
+    Prior to using this flag, the problem was most apparent with updates.
+
+    An initial query like 'UPDATE tablename SET anything = whatever WHERE
+    indexedcol = someval', index_read_idx would get called, using a query
+    constructed with a WHERE clause built from the values of index ('indexcol'
+    in this case, having a value of 'someval').  mysql_store_result would
+    then get called (this would be the result set we want to use).
+
+    After this rnd_init (from sql_update.cc) would be called, it would then
+    unecessarily call "select * from table" on the foreign table, then call
+    mysql_store_result, which would wipe out the correct previous result set
+    from the previous call of index_read_idx's that had the result set
+    containing the correct record, hence update the wrong row!
+
+  */
+
+  if (scan)
+  {
+    if (stored_result)
+    {
+      mysql_free_result(stored_result);
+      stored_result= 0;
+    }
+
+    if (mysql_real_query(mysql,
+                         share->select_query,
+                         strlen(share->select_query)))
+      goto error;
+
+    stored_result= mysql_store_result(mysql);
+    if (!stored_result)
+      goto error;
+  }
+  DBUG_RETURN(0);
+
+error:
+  DBUG_RETURN(stash_remote_error());
+}
+
+
+int ha_federated::rnd_end()
+{
+  DBUG_ENTER("ha_federated::rnd_end");
+  DBUG_RETURN(index_end());
+}
+
+
+int ha_federated::index_end(void)
+{
+  DBUG_ENTER("ha_federated::index_end");
+  if (stored_result)
+  {
+    mysql_free_result(stored_result);
+    stored_result= 0;
+  }
+  active_index= MAX_KEY;
+  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_federated::rnd_next(byte *buf)
+{
+  DBUG_ENTER("ha_federated::rnd_next");
+
+  if (stored_result == 0)
+  {
+    /*
+      Return value of rnd_init is not always checked (see records.cc),
+      so we can get here _even_ if there is _no_ pre-fetched result-set!
+      TODO: fix it. We can delete this in 5.1 when rnd_init() is checked.
+    */
+    DBUG_RETURN(1);
+  }
+  DBUG_RETURN(read_next(buf, stored_result));
+}
+
+
+/*
+  ha_federated::read_next
+
+  reads from a result set and converts to mysql internal
+  format
+
+  SYNOPSIS
+    field_in_record_is_null()
+      buf       byte pointer to record 
+      result    mysql result set 
+
+    DESCRIPTION
+     This method is a wrapper method that reads one record from a result
+     set and converts it to the internal table format
+
+    RETURN VALUE
+      1    error
+      0    no error 
+*/
+
+int ha_federated::read_next(byte *buf, MYSQL_RES *result)
+{
+  int retval;
+  my_ulonglong num_rows;
+  MYSQL_ROW row;
+  DBUG_ENTER("ha_federated::read_next");
+
+  table->status= STATUS_NOT_FOUND;              // For easier return
+
+  /* Fetch a row, insert it back in a row format. */
+  if (!(row= mysql_fetch_row(result)))
+    DBUG_RETURN(HA_ERR_END_OF_FILE);
+
+  if (!(retval= convert_row_to_internal_format(buf, row, result)))
+    table->status= 0;
+
+  DBUG_RETURN(retval);
+}
+
+
+/*
+  store reference to current row so that we can later find it for
+  a re-read, update or delete.
+
+  In case of federated, a reference is either a primary key or
+  the whole record.
+
+  Called from filesort.cc, sql_select.cc, sql_delete.cc and sql_update.cc.
+*/
+
+void ha_federated::position(const byte *record)
+{
+  DBUG_ENTER("ha_federated::position");
+  if (table->s->primary_key != MAX_KEY)
+    key_copy(ref, (byte *)record, table->key_info + table->s->primary_key,
+             ref_length);
+  else
+    memcpy(ref, record, ref_length);
+  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.
+
+  This method is required for an ORDER BY
+
+  Called from filesort.cc records.cc sql_insert.cc sql_select.cc sql_update.cc.
+*/
+
+int ha_federated::rnd_pos(byte *buf, byte *pos)
+{
+  int result;
+  DBUG_ENTER("ha_federated::rnd_pos");
+  statistic_increment(table->in_use->status_var.ha_read_rnd_count,
+                      &LOCK_status);
+  if (table->s->primary_key != MAX_KEY)
+  {
+    /* We have a primary key, so use index_read_idx to find row */
+    result= index_read_idx(buf, table->s->primary_key, pos,
+                           ref_length, HA_READ_KEY_EXACT);
+  }
+  else
+  {
+    /* otherwise, get the old record ref as obtained in ::position */
+    memcpy(buf, pos, ref_length);
+    result= 0;
+  }
+  table->status= result ? STATUS_NOT_FOUND : 0;
+  DBUG_RETURN(result);
+}
+
+
+/*
+  ::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
+  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_federated::info(uint flag)
+{
+  char error_buffer[FEDERATED_QUERY_BUFFER_SIZE];
+  char status_buf[FEDERATED_QUERY_BUFFER_SIZE];
+  char escaped_table_name[FEDERATED_QUERY_BUFFER_SIZE];
+  int error;
+  uint error_code;
+  MYSQL_RES *result= 0;
+  MYSQL_ROW row;
+  String status_query_string(status_buf, sizeof(status_buf), &my_charset_bin);
+  DBUG_ENTER("ha_federated::info");
+
+  error_code= ER_QUERY_ON_FOREIGN_DATA_SOURCE;
+  /* we want not to show table status if not needed to do so */
+  if (flag & (HA_STATUS_VARIABLE | HA_STATUS_CONST))
+  {
+    status_query_string.length(0);
+    status_query_string.append(STRING_WITH_LEN("SHOW TABLE STATUS LIKE '"));
+    escape_string_for_mysql(&my_charset_bin, (char *)escaped_table_name,
+                            sizeof(escaped_table_name),
+                            share->table_name,
+                            share->table_name_length);
+    status_query_string.append(escaped_table_name);
+    status_query_string.append(STRING_WITH_LEN("'"));
+
+    if (mysql_real_query(mysql, status_query_string.ptr(),
+                         status_query_string.length()))
+      goto error;
+
+    status_query_string.length(0);
+
+    result= mysql_store_result(mysql);
+    if (!result)
+      goto error;
+
+    if (!mysql_num_rows(result))
+      goto error;
+
+    if (!(row= mysql_fetch_row(result)))
+      goto error;
+
+    if (flag & HA_STATUS_VARIABLE | HA_STATUS_CONST)
+    {
+      /*
+        deleted is set in ha_federated::info
+      */
+      /*
+        need to figure out what this means as far as federated is concerned,
+        since we don't have a "file"
+
+        data_file_length = ?
+        index_file_length = ?
+        delete_length = ?
+      */
+      if (row[4] != NULL)
+        stats.records=   (ha_rows) my_strtoll10(row[4], (char**) 0,
+                                                       &error);
+      if (row[5] != NULL)
+        stats.mean_rec_length= (ha_rows) my_strtoll10(row[5], (char**) 0, &error);
+
+      stats.data_file_length= stats.records * stats.mean_rec_length;
+
+      if (row[12] != NULL)
+        stats.update_time=     (ha_rows) my_strtoll10(row[12], (char**) 0,
+                                                      &error);
+      if (row[13] != NULL)
+        stats.check_time=      (ha_rows) my_strtoll10(row[13], (char**) 0,
+                                                      &error);
+    }
+    /*
+      size of IO operations (This is based on a good guess, no high science
+      involved)
+    */
+    if (flag & HA_STATUS_CONST)
+      stats.block_size= 4096;
+
+  }
+
+  if (result)
+    mysql_free_result(result);
+
+  DBUG_VOID_RETURN;
+
+error:
+  if (result)
+    mysql_free_result(result);
+
+  my_sprintf(error_buffer, (error_buffer, ": %d : %s",
+                            mysql_errno(mysql), mysql_error(mysql)));
+  my_error(error_code, MYF(0), error_buffer);
+  DBUG_VOID_RETURN;
+}
+
+
+/*
+  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_federated::delete_all_rows()
+{
+  char query_buffer[FEDERATED_QUERY_BUFFER_SIZE];
+  String query(query_buffer, sizeof(query_buffer), &my_charset_bin);
+  DBUG_ENTER("ha_federated::delete_all_rows");
+
+  query.length(0);
+
+  query.set_charset(system_charset_info);
+  query.append(STRING_WITH_LEN("TRUNCATE `"));
+  query.append(share->table_name);
+  query.append(STRING_WITH_LEN("`"));
+
+  /*
+    TRUNCATE won't return anything in mysql_affected_rows
+  */
+  if (mysql_real_query(mysql, query.ptr(), query.length()))
+  {
+    DBUG_RETURN(stash_remote_error());
+  }
+  stats.deleted+= stats.records;
+  stats.records= 0;
+  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 federated  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_federated::store_lock(THD *thd,
+                                         THR_LOCK_DATA **to,
+                                         enum thr_lock_type lock_type)
+{
+  DBUG_ENTER("ha_federated::store_lock");
+  if (lock_type != TL_IGNORE && lock.type == TL_UNLOCK)
+  {
+    /*
+      Here is where we get into the guts of a row level lock.
+      If TL_UNLOCK is set
+      If we are not doing a LOCK TABLE or DISCARD/IMPORT
+      TABLESPACE, then allow multiple writers
+    */
+
+    if ((lock_type >= TL_WRITE_CONCURRENT_INSERT &&
+         lock_type <= TL_WRITE) && !thd->in_lock_tables)
+      lock_type= TL_WRITE_ALLOW_WRITE;
+
+    /*
+      In queries of type INSERT INTO t1 SELECT ... FROM t2 ...
+      MySQL would use the lock TL_READ_NO_INSERT on t2, and that
+      would conflict with TL_WRITE_ALLOW_WRITE, blocking all inserts
+      to t2. Convert the lock to a normal read lock to allow
+      concurrent inserts to t2.
+    */
+
+    if (lock_type == TL_READ_NO_INSERT && !thd->in_lock_tables)
+      lock_type= TL_READ;
+
+    lock.type= lock_type;
+  }
+
+  *to++= &lock;
+
+  DBUG_RETURN(to);
+}
+
+/*
+  create() does nothing, since we have no local setup of our own.
+  FUTURE: We should potentially connect to the foreign database and
+*/
+
+int ha_federated::create(const char *name, TABLE *table_arg,
+                         HA_CREATE_INFO *create_info)
+{
+  int retval;
+  FEDERATED_SHARE tmp_share; // Only a temporary share, to test the url
+  DBUG_ENTER("ha_federated::create");
+
+  if (!(retval= parse_url(&tmp_share, table_arg, 1)))
+    retval= check_foreign_data_source(&tmp_share, 1);
+
+  my_free((gptr) tmp_share.scheme, MYF(MY_ALLOW_ZERO_PTR));
+  DBUG_RETURN(retval);
+
+}
+
+
+int ha_federated::stash_remote_error()
+{
+  DBUG_ENTER("ha_federated::stash_remote_error()");
+  remote_error_number= mysql_errno(mysql);
+  strmake(remote_error_buf, mysql_error(mysql), sizeof(remote_error_buf)-1);
+  DBUG_RETURN(HA_FEDERATED_ERROR_WITH_REMOTE_SYSTEM);
+}
+
+
+bool ha_federated::get_error_message(int error, String* buf)
+{
+  DBUG_ENTER("ha_federated::get_error_message");
+  DBUG_PRINT("enter", ("error: %d", error));
+  if (error == HA_FEDERATED_ERROR_WITH_REMOTE_SYSTEM)
+  {
+    buf->append(STRING_WITH_LEN("Error on remote system: "));
+    buf->qs_append(remote_error_number);
+    buf->append(STRING_WITH_LEN(": "));
+    buf->append(remote_error_buf);
+
+    remote_error_number= 0;
+    remote_error_buf[0]= '\0';
+  }
+  DBUG_PRINT("exit", ("message: %s", buf->ptr()));
+  DBUG_RETURN(FALSE);
+}
+
+int ha_federated::external_lock(THD *thd, int lock_type)
+{
+  int error= 0;
+  ha_federated *trx= (ha_federated *)thd->ha_data[federated_hton.slot];
+  DBUG_ENTER("ha_federated::external_lock");
+
+  if (lock_type != F_UNLCK)
+  {
+    DBUG_PRINT("info",("federated not lock F_UNLCK"));
+    if (!(thd->options & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN))) 
+    {
+      DBUG_PRINT("info",("federated autocommit"));
+      /* 
+        This means we are doing an autocommit
+      */
+      error= connection_autocommit(TRUE);
+      if (error)
+      {
+        DBUG_PRINT("info", ("error setting autocommit TRUE: %d", error));
+        DBUG_RETURN(error);
+      }
+      trans_register_ha(thd, FALSE, &federated_hton);
+    }
+    else 
+    { 
+      DBUG_PRINT("info",("not autocommit"));
+      if (!trx)
+      {
+        /* 
+          This is where a transaction gets its start
+        */
+        error= connection_autocommit(FALSE);
+        if (error)
+        { 
+          DBUG_PRINT("info", ("error setting autocommit FALSE: %d", error));
+          DBUG_RETURN(error);
+        }
+        thd->ha_data[federated_hton.slot]= this;
+        trans_register_ha(thd, TRUE, &federated_hton);
+        /*
+          Send a lock table to the remote end.
+          We do not support this at the moment
+        */
+        if (thd->options & (OPTION_TABLE_LOCK))
+        {
+          DBUG_PRINT("info", ("We do not support lock table yet"));
+        }
+      }
+      else
+      {
+        ha_federated *ptr;
+        for (ptr= trx; ptr; ptr= ptr->trx_next)
+          if (ptr == this)
+            break;
+          else if (!ptr->trx_next)
+            ptr->trx_next= this;
+      }
+    }
+  }
+  DBUG_RETURN(0);
+}
+
+
+static int federated_commit(THD *thd, bool all)
+{
+  int return_val= 0;
+  ha_federated *trx= (ha_federated *)thd->ha_data[federated_hton.slot];
+  DBUG_ENTER("federated_commit");
+
+  if (all)
+  {
+    int error= 0;
+    ha_federated *ptr, *old= NULL;
+    for (ptr= trx; ptr; old= ptr, ptr= ptr->trx_next)
+    {
+      if (old)
+        old->trx_next= NULL;
+      error= ptr->connection_commit();
+      if (error && !return_val);
+        return_val= error;
+    }
+    thd->ha_data[federated_hton.slot]= NULL;
+  }
+
+  DBUG_PRINT("info", ("error val: %d", return_val));
+  DBUG_RETURN(return_val);
+}
+
+
+static int federated_rollback(THD *thd, bool all)
+{
+  int return_val= 0;
+  ha_federated *trx= (ha_federated *)thd->ha_data[federated_hton.slot];
+  DBUG_ENTER("federated_rollback");
+
+  if (all)
+  {
+    int error= 0;
+    ha_federated *ptr, *old= NULL;
+    for (ptr= trx; ptr; old= ptr, ptr= ptr->trx_next)
+    {
+      if (old)
+        old->trx_next= NULL;
+      error= ptr->connection_rollback();
+      if (error && !return_val)
+        return_val= error;
+    }
+    thd->ha_data[federated_hton.slot]= NULL;
+  }
+
+  DBUG_PRINT("info", ("error val: %d", return_val));
+  DBUG_RETURN(return_val);
+}
+
+int ha_federated::connection_commit()
+{
+  DBUG_ENTER("ha_federated::connection_commit");
+  DBUG_RETURN(execute_simple_query("COMMIT", 6));
+}
+
+
+int ha_federated::connection_rollback()
+{
+  DBUG_ENTER("ha_federated::connection_rollback");
+  DBUG_RETURN(execute_simple_query("ROLLBACK", 8));
+}
+
+
+int ha_federated::connection_autocommit(bool state)
+{
+  const char *text;
+  DBUG_ENTER("ha_federated::connection_autocommit");
+  text= (state == true) ? "SET AUTOCOMMIT=1" : "SET AUTOCOMMIT=0";
+  DBUG_RETURN(execute_simple_query(text, 16));
+}
+
+
+int ha_federated::execute_simple_query(const char *query, int len)
+{
+  DBUG_ENTER("ha_federated::execute_simple_query");
+
+  if (mysql_real_query(mysql, query, len))
+  {
+    DBUG_RETURN(stash_remote_error());
+  }
+  DBUG_RETURN(0);
+}
+
+struct st_mysql_storage_engine federated_storage_engine=
+{ MYSQL_HANDLERTON_INTERFACE_VERSION, &federated_hton };
+
+mysql_declare_plugin(federated)
+{
+  MYSQL_STORAGE_ENGINE_PLUGIN,
+  &federated_storage_engine,
+  "FEDERATED",
+  "Patrick Galbraith and Brian Aker, MySQL AB",
+  "Federated MySQL storage engine",
+  federated_db_init, /* Plugin Init */
+  NULL, /* Plugin Deinit */
+  0x0100 /* 1.0 */,
+  0
+}
+mysql_declare_plugin_end;
+
diff --git a/storage/federated/ha_federated.h b/storage/federated/ha_federated.h
new file mode 100644
index 00000000000..ebdc775d3bf
--- /dev/null
+++ b/storage/federated/ha_federated.h
@@ -0,0 +1,241 @@
+/* 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 federated 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
+
+#include <mysql.h>
+
+/* 
+  handler::print_error has a case statement for error numbers.
+  This value is (10000) is far out of range and will envoke the 
+  default: case.  
+  (Current error range is 120-159 from include/my_base.h)
+*/
+#define HA_FEDERATED_ERROR_WITH_REMOTE_SYSTEM 10000
+
+#define FEDERATED_QUERY_BUFFER_SIZE STRING_BUFFER_USUAL_SIZE * 5
+#define FEDERATED_RECORDS_IN_RANGE 2
+
+/*
+  FEDERATED_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_federated_share {
+  /*
+    the primary select query to be used in rnd_init
+  */
+  char *select_query;
+  /*
+    remote host info, parse_url supplies
+  */
+  char *scheme;
+  char *connect_string;
+  char *hostname;
+  char *username;
+  char *password;
+  char *database;
+  char *table_name;
+  char *table;
+  char *socket;
+  char *sport;
+  ushort port;
+  uint table_name_length, connect_string_length, use_count;
+  pthread_mutex_t mutex;
+  THR_LOCK lock;
+} FEDERATED_SHARE;
+
+/*
+  Class definition for the storage engine
+*/
+class ha_federated: public handler
+{
+  THR_LOCK_DATA lock;      /* MySQL lock */
+  FEDERATED_SHARE *share;    /* Shared lock info */
+  MYSQL *mysql; /* MySQL connection */
+  MYSQL_RES *stored_result;
+  uint fetch_num; // stores the fetch num
+  MYSQL_ROW_OFFSET current_position;  // Current position used by ::position()
+  int remote_error_number;
+  char remote_error_buf[FEDERATED_QUERY_BUFFER_SIZE];
+
+private:
+  /*
+      return 0 on success
+      return errorcode otherwise
+  */
+  uint convert_row_to_internal_format(byte *buf, MYSQL_ROW row,
+                                      MYSQL_RES *result);
+  bool create_where_from_key(String *to, KEY *key_info, 
+                             const key_range *start_key,
+                             const key_range *end_key,
+                             bool records_in_range, bool eq_range);
+  int stash_remote_error();
+
+public:
+  ha_federated(TABLE_SHARE *table_arg);
+  ~ha_federated() {}
+  /* The name that will be used for display purposes */
+  const char *table_type() const { return "FEDERATED"; }
+  /*
+    Next pointer used in transaction
+  */
+  ha_federated *trx_next;
+  /*
+    The name of the index type that will be used for display
+    don't implement this method unless you really have indexes
+   */
+  // perhaps get index type
+  const char *index_type(uint inx) { return "REMOTE"; }
+  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
+  */
+  ulonglong table_flags() const
+  {
+    /* fix server to be able to get remote server table flags */
+    return (HA_PRIMARY_KEY_IN_READ_INDEX | HA_FILE_BASED
+            | HA_REC_NOT_IN_SEQ | HA_AUTO_PART_KEY | HA_CAN_INDEX_BLOBS |
+            HA_NO_PREFIX_CHAR_KEYS | HA_PRIMARY_KEY_REQUIRED_FOR_DELETE |
+            HA_PARTIAL_COLUMN_READ | HA_NULL_IN_KEY);
+  }
+  /*
+    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'.
+  */
+    /* fix server to be able to get remote server index flags */
+  ulong index_flags(uint inx, uint part, bool all_parts) const
+  {
+    return (HA_READ_NEXT | HA_READ_RANGE | HA_READ_AFTER_KEY);
+  }
+  uint max_supported_record_length() const { return HA_MAX_REC_LENGTH; }
+  uint max_supported_keys()          const { return MAX_KEY; }
+  uint max_supported_key_parts()     const { return MAX_REF_PARTS; }
+  uint max_supported_key_length()    const { return MAX_KEY_LENGTH; }
+  /*
+    Called in test_quick_select to determine if indexes should be used.
+    Normally, we need to know number of blocks . For federated we need to
+    know number of blocks on remote side, and number of packets and blocks
+    on the network side (?)
+    Talk to Kostja about this - how to get the
+    number of rows * ...
+    disk scan time on other side (block size, size of the row) + network time ...
+    The reason for "records * 1000" is that such a large number forces 
+    this to use indexes "
+  */
+  double scan_time()
+  {
+    DBUG_PRINT("info", ("records %lu", (ulong) stats.records));
+    return (double)(stats.records*1000); 
+  }
+  /*
+    The next method will never be called if you do not implement indexes.
+  */
+  double read_time(uint index, uint ranges, ha_rows rows) 
+  {
+    /*
+      Per Brian, this number is bugus, but this method must be implemented,
+      and at a later date, he intends to document this issue for handler code
+    */
+    return (double) rows /  20.0+1;
+  }
+
+  const key_map *keys_to_use_for_scanning() { return &key_map_full; }
+  /*
+    Everything below are methods that we implment in ha_federated.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_init(uint keynr, bool sorted);
+  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_end();
+  int read_range_first(const key_range *start_key,
+                               const key_range *end_key,
+                               bool eq_range, bool sorted);
+  int read_range_next();
+  /*
+    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
+
+  void update_auto_increment(void);
+  int repair(THD* thd, HA_CHECK_OPT* check_opt);
+  int optimize(THD* thd, HA_CHECK_OPT* check_opt);
+
+  int delete_all_rows(void);
+  int create(const char *name, TABLE *form,
+             HA_CREATE_INFO *create_info);                      //required
+  ha_rows records_in_range(uint inx, key_range *start_key,
+                                   key_range *end_key);
+  uint8 table_cache_type() { return HA_CACHE_TBL_NOCACHE; }
+
+  THR_LOCK_DATA **store_lock(THD *thd, THR_LOCK_DATA **to,
+                             enum thr_lock_type lock_type);     //required
+  virtual bool get_error_message(int error, String *buf);
+  int external_lock(THD *thd, int lock_type);
+  int connection_commit();
+  int connection_rollback();
+  int connection_autocommit(bool state);
+  int execute_simple_query(const char *query, int len);
+
+  int read_next(byte *buf, MYSQL_RES *result);
+  int index_read_idx_with_result_set(byte *buf, uint index,
+                                     const byte *key,
+                                     uint key_len,
+                                     ha_rkey_function find_flag,
+                                     MYSQL_RES **result);
+};
+
+int federated_db_init(void);
+int federated_db_end(ha_panic_function type);
+
diff --git a/storage/federated/plug.in b/storage/federated/plug.in
new file mode 100644
index 00000000000..81c56cb672f
--- /dev/null
+++ b/storage/federated/plug.in
@@ -0,0 +1,4 @@
+MYSQL_STORAGE_ENGINE(federated,,[Federated Storage Engine],
+        [Connects to tables on remote MySQL servers], [max,max-no-ndb])
+MYSQL_PLUGIN_STATIC(federated,    [libfederated.a])
+MYSQL_PLUGIN_DYNAMIC(federated,   [ha_federated.la])
diff --git a/storage/heap/CMakeLists.txt b/storage/heap/CMakeLists.txt
index db5fb8b2981..720bd7228f5 100644
--- a/storage/heap/CMakeLists.txt
+++ b/storage/heap/CMakeLists.txt
@@ -1,8 +1,12 @@
 SET(CMAKE_CXX_FLAGS_DEBUG "${CMAKE_CXX_FLAGS_DEBUG} -DSAFEMALLOC -DSAFE_MUTEX")
 SET(CMAKE_C_FLAGS_DEBUG "${CMAKE_C_FLAGS_DEBUG} -DSAFEMALLOC -DSAFE_MUTEX")
 
-INCLUDE_DIRECTORIES(${CMAKE_SOURCE_DIR}/include)
+INCLUDE_DIRECTORIES(${CMAKE_SOURCE_DIR}/include ${CMAKE_SOURCE_DIR}/zlib
+                    ${CMAKE_SOURCE_DIR}/sql
+                    ${CMAKE_SOURCE_DIR}/regex
+                    ${CMAKE_SOURCE_DIR}/extra/yassl/include)
 ADD_LIBRARY(heap _check.c _rectest.c hp_block.c hp_clear.c hp_close.c hp_create.c
+				ha_heap.cc
 				hp_delete.c hp_extra.c hp_hash.c hp_info.c hp_open.c hp_panic.c
 				hp_rename.c hp_rfirst.c hp_rkey.c hp_rlast.c hp_rnext.c hp_rprev.c
 				hp_rrnd.c hp_rsame.c hp_scan.c hp_static.c hp_update.c hp_write.c)
diff --git a/storage/heap/Makefile.am b/storage/heap/Makefile.am
index 68dce9bca5f..46565126b65 100644
--- a/storage/heap/Makefile.am
+++ b/storage/heap/Makefile.am
@@ -14,22 +14,42 @@
 # along with this program; if not, write to the Free Software
 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
-INCLUDES =		-I$(top_builddir)/include -I$(top_srcdir)/include 
-LDADD =			libheap.a \
-			$(top_builddir)/mysys/libmysys.a \
-			$(top_builddir)/dbug/libdbug.a \
-			$(top_builddir)/strings/libmystrings.a
+MYSQLDATAdir =          $(localstatedir)
+MYSQLSHAREdir =         $(pkgdatadir)
+MYSQLBASEdir=           $(prefix)
+MYSQLLIBdir=            $(pkglibdir)
+INCLUDES =              -I$(top_srcdir)/include -I$(top_builddir)/include \
+			-I$(top_srcdir)/regex \
+			-I$(top_srcdir)/sql \
+                        -I$(srcdir)
+WRAPLIBS=
+
+LDADD =
+
+DEFS =                  @DEFS@
 pkglib_LIBRARIES =	libheap.a
 noinst_PROGRAMS	=	hp_test1 hp_test2
+noinst_LIBRARIES =	libheap.a
 hp_test1_LDFLAGS = @NOINST_LDFLAGS@
+hp_test1_LDADD =	libheap.a \
+			$(top_builddir)/mysys/libmysys.a \
+			$(top_builddir)/dbug/libdbug.a \
+			$(top_builddir)/strings/libmystrings.a
 hp_test2_LDFLAGS = @NOINST_LDFLAGS@
-noinst_HEADERS =	heapdef.h
+hp_test2_LDADD =	libheap.a \
+			$(top_builddir)/mysys/libmysys.a \
+			$(top_builddir)/dbug/libdbug.a \
+			$(top_builddir)/strings/libmystrings.a
+noinst_HEADERS =	heapdef.h ha_heap.h
 libheap_a_SOURCES =	hp_open.c hp_extra.c hp_close.c hp_panic.c hp_info.c \
 			hp_rrnd.c hp_scan.c hp_update.c hp_write.c hp_delete.c \
 			hp_rsame.c hp_create.c hp_rename.c hp_rfirst.c \
 			hp_rnext.c hp_rlast.c hp_rprev.c hp_clear.c \
 			hp_rkey.c hp_block.c \
+			ha_heap.cc \
 			hp_hash.c _check.c _rectest.c hp_static.c
+
+
 EXTRA_DIST =		CMakeLists.txt
 
 # Don't update the files from bitkeeper
diff --git a/storage/heap/ha_heap.cc b/storage/heap/ha_heap.cc
new file mode 100644
index 00000000000..317f85d26f2
--- /dev/null
+++ b/storage/heap/ha_heap.cc
@@ -0,0 +1,713 @@
+/* Copyright (C) 2000,2004 MySQL AB & MySQL Finland AB & TCX DataKonsult 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 */
+
+
+#ifdef USE_PRAGMA_IMPLEMENTATION
+#pragma implementation				// gcc: Class implementation
+#endif
+
+#define MYSQL_SERVER 1
+#include "mysql_priv.h"
+#include <mysql/plugin.h>
+#include "ha_heap.h"
+
+
+static handler *heap_create_handler(TABLE_SHARE *table, MEM_ROOT *mem_root);
+
+handlerton heap_hton;
+
+int heap_init()
+{
+  heap_hton.state=      SHOW_OPTION_YES;
+  heap_hton.db_type=    DB_TYPE_HEAP;
+  heap_hton.create=     heap_create_handler;
+  heap_hton.panic=      heap_panic;
+  heap_hton.flags=      HTON_CAN_RECREATE;
+  return 0;
+}
+
+static handler *heap_create_handler(TABLE_SHARE *table, MEM_ROOT *mem_root)
+{
+  return new (mem_root) ha_heap(table);
+}
+
+
+/*****************************************************************************
+** HEAP tables
+*****************************************************************************/
+
+ha_heap::ha_heap(TABLE_SHARE *table_arg)
+  :handler(&heap_hton, table_arg), file(0), records_changed(0),
+  key_stat_version(0)
+{}
+
+
+static const char *ha_heap_exts[] = {
+  NullS
+};
+
+const char **ha_heap::bas_ext() const
+{
+  return ha_heap_exts;
+}
+
+/*
+  Hash index statistics is updated (copied from HP_KEYDEF::hash_buckets to 
+  rec_per_key) after 1/HEAP_STATS_UPDATE_THRESHOLD fraction of table records 
+  have been inserted/updated/deleted. delete_all_rows() and table flush cause 
+  immediate update.
+
+  NOTE
+   hash index statistics must be updated when number of table records changes
+   from 0 to non-zero value and vice versa. Otherwise records_in_range may 
+   erroneously return 0 and 'range' may miss records.
+*/
+#define HEAP_STATS_UPDATE_THRESHOLD 10
+
+int ha_heap::open(const char *name, int mode, uint test_if_locked)
+{
+  if (!(file= heap_open(name, mode)) && my_errno == ENOENT)
+  {
+    HA_CREATE_INFO create_info;
+    bzero(&create_info, sizeof(create_info));
+    if (!create(name, table, &create_info))
+    {
+      file= heap_open(name, mode);
+      implicit_emptied= 1;
+    }
+  }
+  ref_length= sizeof(HEAP_PTR);
+  if (file)
+  {
+    /* Initialize variables for the opened table */
+    set_keys_for_scanning();
+    /*
+      We cannot run update_key_stats() here because we do not have a
+      lock on the table. The 'records' count might just be changed
+      temporarily at this moment and we might get wrong statistics (Bug
+      #10178). Instead we request for update. This will be done in
+      ha_heap::info(), which is always called before key statistics are
+      used.
+    */
+    key_stat_version= file->s->key_stat_version-1;
+  }
+  return (file ? 0 : 1);
+}
+
+int ha_heap::close(void)
+{
+  return heap_close(file);
+}
+
+
+/*
+  Compute which keys to use for scanning
+
+  SYNOPSIS
+    set_keys_for_scanning()
+    no parameter
+
+  DESCRIPTION
+    Set the bitmap btree_keys, which is used when the upper layers ask
+    which keys to use for scanning. For each btree index the
+    corresponding bit is set.
+
+  RETURN
+    void
+*/
+
+void ha_heap::set_keys_for_scanning(void)
+{
+  btree_keys.clear_all();
+  for (uint i= 0 ; i < table->s->keys ; i++)
+  {
+    if (table->key_info[i].algorithm == HA_KEY_ALG_BTREE)
+      btree_keys.set_bit(i);
+  }
+}
+
+
+void ha_heap::update_key_stats()
+{
+  for (uint i= 0; i < table->s->keys; i++)
+  {
+    KEY *key=table->key_info+i;
+    if (!key->rec_per_key)
+      continue;
+    if (key->algorithm != HA_KEY_ALG_BTREE)
+    {
+      if (key->flags & HA_NOSAME)
+        key->rec_per_key[key->key_parts-1]= 1;
+      else
+      {
+        ha_rows hash_buckets= file->s->keydef[i].hash_buckets;
+        uint no_records= hash_buckets ? file->s->records/hash_buckets : 2;
+        if (no_records < 2)
+          no_records= 2;
+        key->rec_per_key[key->key_parts-1]= no_records;
+      }
+    }
+  }
+  records_changed= 0;
+  /* At the end of update_key_stats() we can proudly claim they are OK. */
+  key_stat_version= file->s->key_stat_version;
+}
+
+
+int ha_heap::write_row(byte * buf)
+{
+  int res;
+  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();
+  if (table->next_number_field && buf == table->record[0])
+    update_auto_increment();
+  res= heap_write(file,buf);
+  if (!res && (++records_changed*HEAP_STATS_UPDATE_THRESHOLD > 
+               file->s->records))
+  {
+    /*
+       We can perform this safely since only one writer at the time is
+       allowed on the table.
+    */
+    file->s->key_stat_version++;
+  }
+  return res;
+}
+
+int ha_heap::update_row(const byte * old_data, byte * new_data)
+{
+  int res;
+  statistic_increment(table->in_use->status_var.ha_update_count,&LOCK_status);
+  if (table->timestamp_field_type & TIMESTAMP_AUTO_SET_ON_UPDATE)
+    table->timestamp_field->set_time();
+  res= heap_update(file,old_data,new_data);
+  if (!res && ++records_changed*HEAP_STATS_UPDATE_THRESHOLD > 
+              file->s->records)
+  {
+    /*
+       We can perform this safely since only one writer at the time is
+       allowed on the table.
+    */
+    file->s->key_stat_version++;
+  }
+  return res;
+}
+
+int ha_heap::delete_row(const byte * buf)
+{
+  int res;
+  statistic_increment(table->in_use->status_var.ha_delete_count,&LOCK_status);
+  res= heap_delete(file,buf);
+  if (!res && table->s->tmp_table == NO_TMP_TABLE && 
+      ++records_changed*HEAP_STATS_UPDATE_THRESHOLD > file->s->records)
+  {
+    /*
+       We can perform this safely since only one writer at the time is
+       allowed on the table.
+    */
+    file->s->key_stat_version++;
+  }
+  return res;
+}
+
+int ha_heap::index_read(byte * buf, const byte * key, uint key_len,
+			enum ha_rkey_function find_flag)
+{
+  DBUG_ASSERT(inited==INDEX);
+  statistic_increment(table->in_use->status_var.ha_read_key_count,
+		      &LOCK_status);
+  int error = heap_rkey(file,buf,active_index, key, key_len, find_flag);
+  table->status = error ? STATUS_NOT_FOUND : 0;
+  return error;
+}
+
+int ha_heap::index_read_last(byte *buf, const byte *key, uint key_len)
+{
+  DBUG_ASSERT(inited==INDEX);
+  statistic_increment(table->in_use->status_var.ha_read_key_count,
+		      &LOCK_status);
+  int error= heap_rkey(file, buf, active_index, key, key_len,
+		       HA_READ_PREFIX_LAST);
+  table->status= error ? STATUS_NOT_FOUND : 0;
+  return error;
+}
+
+int ha_heap::index_read_idx(byte * buf, uint index, const byte * key,
+			    uint key_len, enum ha_rkey_function find_flag)
+{
+  statistic_increment(table->in_use->status_var.ha_read_key_count,
+		      &LOCK_status);
+  int error = heap_rkey(file, buf, index, key, key_len, find_flag);
+  table->status = error ? STATUS_NOT_FOUND : 0;
+  return error;
+}
+
+int ha_heap::index_next(byte * buf)
+{
+  DBUG_ASSERT(inited==INDEX);
+  statistic_increment(table->in_use->status_var.ha_read_next_count,
+		      &LOCK_status);
+  int error=heap_rnext(file,buf);
+  table->status=error ? STATUS_NOT_FOUND: 0;
+  return error;
+}
+
+int ha_heap::index_prev(byte * buf)
+{
+  DBUG_ASSERT(inited==INDEX);
+  statistic_increment(table->in_use->status_var.ha_read_prev_count,
+		      &LOCK_status);
+  int error=heap_rprev(file,buf);
+  table->status=error ? STATUS_NOT_FOUND: 0;
+  return error;
+}
+
+int ha_heap::index_first(byte * buf)
+{
+  DBUG_ASSERT(inited==INDEX);
+  statistic_increment(table->in_use->status_var.ha_read_first_count,
+		      &LOCK_status);
+  int error=heap_rfirst(file, buf, active_index);
+  table->status=error ? STATUS_NOT_FOUND: 0;
+  return error;
+}
+
+int ha_heap::index_last(byte * buf)
+{
+  DBUG_ASSERT(inited==INDEX);
+  statistic_increment(table->in_use->status_var.ha_read_last_count,
+		      &LOCK_status);
+  int error=heap_rlast(file, buf, active_index);
+  table->status=error ? STATUS_NOT_FOUND: 0;
+  return error;
+}
+
+int ha_heap::rnd_init(bool scan)
+{
+  return scan ? heap_scan_init(file) : 0;
+}
+
+int ha_heap::rnd_next(byte *buf)
+{
+  statistic_increment(table->in_use->status_var.ha_read_rnd_next_count,
+		      &LOCK_status);
+  int error=heap_scan(file, buf);
+  table->status=error ? STATUS_NOT_FOUND: 0;
+  return error;
+}
+
+int ha_heap::rnd_pos(byte * buf, byte *pos)
+{
+  int error;
+  HEAP_PTR position;
+  statistic_increment(table->in_use->status_var.ha_read_rnd_count,
+		      &LOCK_status);
+  memcpy_fixed((char*) &position,pos,sizeof(HEAP_PTR));
+  error=heap_rrnd(file, buf, position);
+  table->status=error ? STATUS_NOT_FOUND: 0;
+  return error;
+}
+
+void ha_heap::position(const byte *record)
+{
+  *(HEAP_PTR*) ref= heap_position(file);	// Ref is aligned
+}
+
+void ha_heap::info(uint flag)
+{
+  HEAPINFO info;
+  (void) heap_info(file,&info,flag);
+
+  errkey= info.errkey;
+  stats.records = info.records;
+  stats.deleted = info.deleted;
+  stats.mean_rec_length=info.reclength;
+  stats.data_file_length=info.data_length;
+  stats.index_file_length=info.index_length;
+  stats.max_data_file_length= info.max_records* info.reclength;
+  stats.delete_length= info.deleted * info.reclength;
+  if (flag & HA_STATUS_AUTO)
+    stats.auto_increment_value= info.auto_increment;
+  /*
+    If info() is called for the first time after open(), we will still
+    have to update the key statistics. Hoping that a table lock is now
+    in place.
+  */
+  if (key_stat_version != file->s->key_stat_version)
+    update_key_stats();
+}
+
+
+int ha_heap::extra(enum ha_extra_function operation)
+{
+  return heap_extra(file,operation);
+}
+
+
+int ha_heap::reset()
+{
+  return heap_reset(file);
+}
+
+
+int ha_heap::delete_all_rows()
+{
+  heap_clear(file);
+  if (table->s->tmp_table == NO_TMP_TABLE)
+  {
+    /*
+       We can perform this safely since only one writer at the time is
+       allowed on the table.
+    */
+    file->s->key_stat_version++;
+  }
+  return 0;
+}
+
+int ha_heap::external_lock(THD *thd, int lock_type)
+{
+  return 0;					// No external locking
+}
+
+
+/*
+  Disable indexes.
+
+  SYNOPSIS
+    disable_indexes()
+    mode        mode of operation:
+                HA_KEY_SWITCH_NONUNIQ      disable all non-unique keys
+                HA_KEY_SWITCH_ALL          disable all keys
+                HA_KEY_SWITCH_NONUNIQ_SAVE dis. non-uni. and make persistent
+                HA_KEY_SWITCH_ALL_SAVE     dis. all keys and make persistent
+
+  DESCRIPTION
+    Disable indexes and clear keys to use for scanning.
+
+  IMPLEMENTATION
+    HA_KEY_SWITCH_NONUNIQ       is not implemented.
+    HA_KEY_SWITCH_NONUNIQ_SAVE  is not implemented with HEAP.
+    HA_KEY_SWITCH_ALL_SAVE      is not implemented with HEAP.
+
+  RETURN
+    0  ok
+    HA_ERR_WRONG_COMMAND  mode not implemented.
+*/
+
+int ha_heap::disable_indexes(uint mode)
+{
+  int error;
+
+  if (mode == HA_KEY_SWITCH_ALL)
+  {
+    if (!(error= heap_disable_indexes(file)))
+      set_keys_for_scanning();
+  }
+  else
+  {
+    /* mode not implemented */
+    error= HA_ERR_WRONG_COMMAND;
+  }
+  return error;
+}
+
+
+/*
+  Enable indexes.
+
+  SYNOPSIS
+    enable_indexes()
+    mode        mode of operation:
+                HA_KEY_SWITCH_NONUNIQ      enable all non-unique keys
+                HA_KEY_SWITCH_ALL          enable all keys
+                HA_KEY_SWITCH_NONUNIQ_SAVE en. non-uni. and make persistent
+                HA_KEY_SWITCH_ALL_SAVE     en. all keys and make persistent
+
+  DESCRIPTION
+    Enable indexes and set keys to use for scanning.
+    The indexes might have been disabled by disable_index() before.
+    The function works only if both data and indexes are empty,
+    since the heap storage engine cannot repair the indexes.
+    To be sure, call handler::delete_all_rows() before.
+
+  IMPLEMENTATION
+    HA_KEY_SWITCH_NONUNIQ       is not implemented.
+    HA_KEY_SWITCH_NONUNIQ_SAVE  is not implemented with HEAP.
+    HA_KEY_SWITCH_ALL_SAVE      is not implemented with HEAP.
+
+  RETURN
+    0  ok
+    HA_ERR_CRASHED  data or index is non-empty. Delete all rows and retry.
+    HA_ERR_WRONG_COMMAND  mode not implemented.
+*/
+
+int ha_heap::enable_indexes(uint mode)
+{
+  int error;
+
+  if (mode == HA_KEY_SWITCH_ALL)
+  {
+    if (!(error= heap_enable_indexes(file)))
+      set_keys_for_scanning();
+  }
+  else
+  {
+    /* mode not implemented */
+    error= HA_ERR_WRONG_COMMAND;
+  }
+  return error;
+}
+
+
+/*
+  Test if indexes are disabled.
+
+  SYNOPSIS
+    indexes_are_disabled()
+    no parameters
+
+  RETURN
+    0  indexes are not disabled
+    1  all indexes are disabled
+   [2  non-unique indexes are disabled - NOT YET IMPLEMENTED]
+*/
+
+int ha_heap::indexes_are_disabled(void)
+{
+  return heap_indexes_are_disabled(file);
+}
+
+THR_LOCK_DATA **ha_heap::store_lock(THD *thd,
+				    THR_LOCK_DATA **to,
+				    enum thr_lock_type lock_type)
+{
+  if (lock_type != TL_IGNORE && file->lock.type == TL_UNLOCK)
+    file->lock.type=lock_type;
+  *to++= &file->lock;
+  return to;
+}
+
+/*
+  We have to ignore ENOENT entries as the HEAP table is created on open and
+  not when doing a CREATE on the table.
+*/
+
+int ha_heap::delete_table(const char *name)
+{
+  int error= heap_delete_table(name);
+  return error == ENOENT ? 0 : error;
+}
+
+
+void ha_heap::drop_table(const char *name)
+{
+  heap_drop_table(file);
+  close();
+}
+
+
+int ha_heap::rename_table(const char * from, const char * to)
+{
+  return heap_rename(from,to);
+}
+
+
+ha_rows ha_heap::records_in_range(uint inx, key_range *min_key,
+                                  key_range *max_key)
+{
+  KEY *key=table->key_info+inx;
+  if (key->algorithm == HA_KEY_ALG_BTREE)
+    return hp_rb_records_in_range(file, inx, min_key, max_key);
+
+  if (!min_key || !max_key ||
+      min_key->length != max_key->length ||
+      min_key->length != key->key_length ||
+      min_key->flag != HA_READ_KEY_EXACT ||
+      max_key->flag != HA_READ_AFTER_KEY)
+    return HA_POS_ERROR;			// Can only use exact keys
+
+  if (stats.records <= 1)
+    return stats.records;
+
+  /* Assert that info() did run. We need current statistics here. */
+  DBUG_ASSERT(key_stat_version == file->s->key_stat_version);
+  return key->rec_per_key[key->key_parts-1];
+}
+
+
+int ha_heap::create(const char *name, TABLE *table_arg,
+		    HA_CREATE_INFO *create_info)
+{
+  uint key, parts, mem_per_row= 0, keys= table_arg->s->keys;
+  uint auto_key= 0, auto_key_type= 0;
+  ha_rows max_rows;
+  HP_KEYDEF *keydef;
+  HA_KEYSEG *seg;
+  int error;
+  TABLE_SHARE *share= table_arg->s;
+  bool found_real_auto_increment= 0;
+
+  for (key= parts= 0; key < keys; key++)
+    parts+= table_arg->key_info[key].key_parts;
+
+  if (!(keydef= (HP_KEYDEF*) my_malloc(keys * sizeof(HP_KEYDEF) +
+				       parts * sizeof(HA_KEYSEG),
+				       MYF(MY_WME))))
+    return my_errno;
+  seg= my_reinterpret_cast(HA_KEYSEG*) (keydef + keys);
+  for (key= 0; key < keys; key++)
+  {
+    KEY *pos= table_arg->key_info+key;
+    KEY_PART_INFO *key_part=     pos->key_part;
+    KEY_PART_INFO *key_part_end= key_part + pos->key_parts;
+
+    keydef[key].keysegs=   (uint) pos->key_parts;
+    keydef[key].flag=      (pos->flags & (HA_NOSAME | HA_NULL_ARE_EQUAL));
+    keydef[key].seg=       seg;
+
+    switch (pos->algorithm) {
+    case HA_KEY_ALG_UNDEF:
+    case HA_KEY_ALG_HASH:
+      keydef[key].algorithm= HA_KEY_ALG_HASH;
+      mem_per_row+= sizeof(char*) * 2; // = sizeof(HASH_INFO)
+      break;
+    case HA_KEY_ALG_BTREE:
+      keydef[key].algorithm= HA_KEY_ALG_BTREE;
+      mem_per_row+=sizeof(TREE_ELEMENT)+pos->key_length+sizeof(char*);
+      break;
+    default:
+      DBUG_ASSERT(0); // cannot happen
+    }
+
+    for (; key_part != key_part_end; key_part++, seg++)
+    {
+      Field *field= key_part->field;
+
+      if (pos->algorithm == HA_KEY_ALG_BTREE)
+	seg->type= field->key_type();
+      else
+      {
+        if ((seg->type = field->key_type()) != (int) HA_KEYTYPE_TEXT &&
+            seg->type != HA_KEYTYPE_VARTEXT1 &&
+            seg->type != HA_KEYTYPE_VARTEXT2 &&
+            seg->type != HA_KEYTYPE_VARBINARY1 &&
+            seg->type != HA_KEYTYPE_VARBINARY2)
+          seg->type= HA_KEYTYPE_BINARY;
+      }
+      seg->start=   (uint) key_part->offset;
+      seg->length=  (uint) key_part->length;
+      seg->flag=    key_part->key_part_flag;
+
+      seg->charset= field->charset();
+      if (field->null_ptr)
+      {
+	seg->null_bit= field->null_bit;
+	seg->null_pos= (uint) (field->null_ptr - (uchar*) table_arg->record[0]);
+      }
+      else
+      {
+	seg->null_bit= 0;
+	seg->null_pos= 0;
+      }
+      if (field->flags & AUTO_INCREMENT_FLAG &&
+          table_arg->found_next_number_field &&
+          key == share->next_number_index)
+      {
+        /*
+          Store key number and type for found auto_increment key
+          We have to store type as seg->type can differ from it
+        */
+        auto_key= key+ 1;
+	auto_key_type= field->key_type();
+      }
+    }
+  }
+  mem_per_row+= MY_ALIGN(share->reclength + 1, sizeof(char*));
+  max_rows = (ha_rows) (table_arg->in_use->variables.max_heap_table_size /
+			mem_per_row);
+  if (table_arg->found_next_number_field)
+  {
+    keydef[share->next_number_index].flag|= HA_AUTO_KEY;
+    found_real_auto_increment= share->next_number_key_offset == 0;
+  }
+  HP_CREATE_INFO hp_create_info;
+  hp_create_info.auto_key= auto_key;
+  hp_create_info.auto_key_type= auto_key_type;
+  hp_create_info.auto_increment= (create_info->auto_increment_value ?
+				  create_info->auto_increment_value - 1 : 0);
+  hp_create_info.max_table_size=current_thd->variables.max_heap_table_size;
+  hp_create_info.with_auto_increment= found_real_auto_increment;
+  max_rows = (ha_rows) (hp_create_info.max_table_size / mem_per_row);
+  error= heap_create(name,
+		     keys, keydef, share->reclength,
+		     (ulong) ((share->max_rows < max_rows &&
+			       share->max_rows) ? 
+			      share->max_rows : max_rows),
+		     (ulong) share->min_rows, &hp_create_info);
+  my_free((gptr) keydef, MYF(0));
+  if (file)
+    info(HA_STATUS_NO_LOCK | HA_STATUS_CONST | HA_STATUS_VARIABLE);
+  return (error);
+}
+
+
+void ha_heap::update_create_info(HA_CREATE_INFO *create_info)
+{
+  table->file->info(HA_STATUS_AUTO);
+  if (!(create_info->used_fields & HA_CREATE_USED_AUTO))
+    create_info->auto_increment_value= stats.auto_increment_value;
+}
+
+void ha_heap::get_auto_increment(ulonglong offset, ulonglong increment,
+                                 ulonglong nb_desired_values,
+                                 ulonglong *first_value,
+                                 ulonglong *nb_reserved_values)
+{
+  ha_heap::info(HA_STATUS_AUTO);
+  *first_value= stats.auto_increment_value;
+  /* such table has only table-level locking so reserves up to +inf */
+  *nb_reserved_values= ULONGLONG_MAX;
+}
+
+
+bool ha_heap::check_if_incompatible_data(HA_CREATE_INFO *info,
+					 uint table_changes)
+{
+  /* Check that auto_increment value was not changed */
+  if ((table_changes != IS_EQUAL_YES &&
+       info->used_fields & HA_CREATE_USED_AUTO) &&
+      info->auto_increment_value != 0)
+    return COMPATIBLE_DATA_NO;
+  return COMPATIBLE_DATA_YES;
+}
+
+struct st_mysql_storage_engine heap_storage_engine=
+{ MYSQL_HANDLERTON_INTERFACE_VERSION, &heap_hton};
+
+mysql_declare_plugin(heap)
+{
+  MYSQL_STORAGE_ENGINE_PLUGIN,
+  &heap_storage_engine,
+  "MEMORY",
+  "MySQL AB",
+  "Hash based, stored in memory, useful for temporary tables",
+  heap_init,
+  NULL,
+  0x0100, /* 1.0 */
+  0
+}
+mysql_declare_plugin_end;
diff --git a/storage/heap/ha_heap.h b/storage/heap/ha_heap.h
new file mode 100644
index 00000000000..00e59856f26
--- /dev/null
+++ b/storage/heap/ha_heap.h
@@ -0,0 +1,118 @@
+/* Copyright (C) 2000,2004 MySQL AB & MySQL Finland AB & TCX DataKonsult 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 */
+
+
+#ifdef USE_PRAGMA_INTERFACE
+#pragma interface			/* gcc class implementation */
+#endif
+
+/* class for the the heap handler */
+
+#include <heap.h>
+
+class ha_heap: public handler
+{
+  HP_INFO *file;
+  key_map btree_keys;
+  /* number of records changed since last statistics update */
+  uint    records_changed;
+  uint    key_stat_version;
+public:
+  ha_heap(TABLE_SHARE *table);
+  ~ha_heap() {}
+  const char *table_type() const
+  {
+    return (table->in_use->variables.sql_mode & MODE_MYSQL323) ?
+           "HEAP" : "MEMORY";
+  }
+  const char *index_type(uint inx)
+  {
+    return ((table_share->key_info[inx].algorithm == HA_KEY_ALG_BTREE) ?
+            "BTREE" : "HASH");
+  }
+  /* Rows also use a fixed-size format */
+  enum row_type get_row_type() const { return ROW_TYPE_FIXED; }
+  const char **bas_ext() const;
+  ulonglong table_flags() const
+  {
+    return (HA_FAST_KEY_READ | HA_NO_BLOBS | HA_NULL_IN_KEY |
+            HA_REC_NOT_IN_SEQ | HA_CAN_INSERT_DELAYED | HA_NO_TRANSACTIONS |
+            HA_HAS_RECORDS | HA_STATS_RECORDS_IS_EXACT);
+  }
+  ulong index_flags(uint inx, uint part, bool all_parts) const
+  {
+    return ((table_share->key_info[inx].algorithm == HA_KEY_ALG_BTREE) ?
+	    HA_READ_NEXT | HA_READ_PREV | HA_READ_ORDER | HA_READ_RANGE :
+	    HA_ONLY_WHOLE_INDEX);
+  }
+  const key_map *keys_to_use_for_scanning() { return &btree_keys; }
+  uint max_supported_keys()          const { return MAX_KEY; }
+  uint max_supported_key_part_length() const { return MAX_KEY_LENGTH; }
+  double scan_time()
+  { return (double) (stats.records+stats.deleted) / 20.0+10; }
+  double read_time(uint index, uint ranges, ha_rows rows)
+  { return (double) rows /  20.0+1; }
+
+  int open(const char *name, int mode, uint test_if_locked);
+  int close(void);
+  void set_keys_for_scanning(void);
+  int write_row(byte * buf);
+  int update_row(const byte * old_data, byte * new_data);
+  int delete_row(const byte * buf);
+  virtual void get_auto_increment(ulonglong offset, ulonglong increment,
+                                  ulonglong nb_desired_values,
+                                  ulonglong *first_value,
+                                  ulonglong *nb_reserved_values);
+  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_read_last(byte * buf, const byte * key, uint key_len);
+  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);
+  int rnd_next(byte *buf);
+  int rnd_pos(byte * buf, byte *pos);
+  void position(const byte *record);
+  void info(uint);
+  int extra(enum ha_extra_function operation);
+  int reset();
+  int external_lock(THD *thd, int lock_type);
+  int delete_all_rows(void);
+  int disable_indexes(uint mode);
+  int enable_indexes(uint mode);
+  int indexes_are_disabled(void);
+  ha_rows records_in_range(uint inx, key_range *min_key, key_range *max_key);
+  int delete_table(const char *from);
+  void drop_table(const char *name);
+  int rename_table(const char * from, const char * to);
+  int create(const char *name, TABLE *form, HA_CREATE_INFO *create_info);
+  void update_create_info(HA_CREATE_INFO *create_info);
+
+  THR_LOCK_DATA **store_lock(THD *thd, THR_LOCK_DATA **to,
+			     enum thr_lock_type lock_type);
+  int cmp_ref(const byte *ref1, const byte *ref2)
+  {
+    HEAP_PTR ptr1=*(HEAP_PTR*)ref1;
+    HEAP_PTR ptr2=*(HEAP_PTR*)ref2;
+    return ptr1 < ptr2? -1 : (ptr1 > ptr2? 1 : 0);
+  }
+  bool check_if_incompatible_data(HA_CREATE_INFO *info, uint table_changes);
+private:
+  void update_key_stats();
+};
diff --git a/storage/heap/plug.in b/storage/heap/plug.in
new file mode 100644
index 00000000000..9e744b6ac0d
--- /dev/null
+++ b/storage/heap/plug.in
@@ -0,0 +1,6 @@
+MYSQL_STORAGE_ENGINE(heap,no,   [Memory Storage Engine],
+        [Volatile memory based tables])
+MYSQL_PLUGIN_DIRECTORY(heap,    [storage/heap])
+MYSQL_PLUGIN_STATIC(heap,       [libheap.a])
+MYSQL_PLUGIN_MANDATORY(heap)    dnl Memory tables
+
diff --git a/storage/innobase/Makefile.am b/storage/innobase/Makefile.am
index 3f46f059fc0..a5b84e0a908 100644
--- a/storage/innobase/Makefile.am
+++ b/storage/innobase/Makefile.am
@@ -17,10 +17,20 @@
 
 # Process this file with automake to create Makefile.in
 
+MYSQLDATAdir =          $(localstatedir)
+MYSQLSHAREdir =         $(pkgdatadir)
+MYSQLBASEdir=           $(prefix)
+MYSQLLIBdir=            $(pkglibdir)
+INCLUDES =              -I$(top_srcdir)/include -I$(top_builddir)/include \
+			-I$(top_srcdir)/regex \
+			-I$(top_srcdir)/storage/innobase/include \
+			-I$(top_srcdir)/sql \
+                        -I$(srcdir)
+
 AUTOMAKE_OPTIONS =	foreign
 TAR =			gtar
 
-noinst_HEADERS = ib_config.h
+noinst_HEADERS = 
 
 SUBDIRS =		os ut btr buf data dict dyn eval fil fsp fut \
 			ha ibuf lock log mach mem mtr page \
diff --git a/storage/innobase/configure.in b/storage/innobase/configure.in
deleted file mode 100644
index 4aaa28da89e..00000000000
--- a/storage/innobase/configure.in
+++ /dev/null
@@ -1,156 +0,0 @@
-# Process this file with autoconf to produce a configure script
-AC_INIT
-AC_CANONICAL_SYSTEM
-AM_MAINTAINER_MODE
-AM_CONFIG_HEADER(ib_config.h)
-AM_INIT_AUTOMAKE(ib, 0.90)
-
-# This is need before AC_PROG_CC
-#
-
-if test "x${CFLAGS-}" = x ; then
-  cflags_is_set=no
-else
-  cflags_is_set=yes
-fi
-
-if test "x${CPPFLAGS-}" = x ; then
-  cppflags_is_set=no
-else
-  cppflags_is_set=yes
-fi
-
-if test "x${LDFLAGS-}" = x ; then
-  ldflags_is_set=no
-else
-  ldflags_is_set=yes
-fi
-
-# The following hack should ensure that configure doesn't add optimizing
-# or debugging flags to CFLAGS or CXXFLAGS
-CFLAGS="$CFLAGS "
-CXXFLAGS="$CXXFLAGS "
-
-AC_PROG_CC
-AC_PROG_RANLIB
-AC_PROG_INSTALL
-AC_PROG_LIBTOOL
-AC_CHECK_HEADERS(aio.h sched.h)
-AC_CHECK_SIZEOF(int, 4)
-AC_CHECK_SIZEOF(long, 4)
-AC_CHECK_SIZEOF(void*, 4)
-AC_CHECK_FUNCS(sched_yield)
-AC_CHECK_FUNCS(fdatasync)
-AC_CHECK_FUNCS(localtime_r)
-#AC_CHECK_FUNCS(readdir_r) MySQL checks that it has also the right args.
-# Some versions of Unix only take 2 arguments.
-#AC_C_INLINE  Already checked in MySQL
-AC_C_BIGENDIAN
-
-# Build optimized or debug version ?
-# First check for gcc and g++
-if test "$ac_cv_prog_gcc" = "yes"
-then
-  DEBUG_CFLAGS="-g"
-  DEBUG_OPTIMIZE_CC="-O"
-  OPTIMIZE_CFLAGS="$MAX_C_OPTIMIZE"
-else
-  DEBUG_CFLAGS="-g"
-  DEBUG_OPTIMIZE_CC=""
-  OPTIMIZE_CFLAGS="-O"
-fi
-if test "$ac_cv_prog_cxx_g" = "yes"
-then
-  DEBUG_CXXFLAGS="-g"
-  DEBUG_OPTIMIZE_CXX="-O"
-  OPTIMIZE_CXXFLAGS="-O3"
-else
-  DEBUG_CXXFLAGS="-g"
-  DEBUG_OPTIMIZE_CXX=""
-  OPTIMIZE_CXXFLAGS="-O"
-fi
-AC_ARG_WITH(debug,
-    [  --without-debug         Build a production version without debugging code],
-    [with_debug=$withval],
-    [with_debug=no])
-if test "$with_debug" = "yes"
-then
-  # Medium debug.
-  CFLAGS="$DEBUG_CFLAGS $DEBUG_OPTIMIZE_CC -DDBUG_ON -DSAFE_MUTEX $CFLAGS"
-  CXXFLAGS="$DEBUG_CXXFLAGS $DEBUG_OPTIMIZE_CXX -DSAFE_MUTEX $CXXFLAGS"
-elif test "$with_debug" = "full"
-then
-  # Full debug. Very slow in some cases
-  CFLAGS="$DEBUG_CFLAGS -DDBUG_ON -DSAFE_MUTEX -DSAFEMALLOC $CFLAGS"
-  CXXFLAGS="$DEBUG_CXXFLAGS -DSAFE_MUTEX -DSAFEMALLOC $CXXFLAGS"
-else
-  # Optimized version. No debug
-  CFLAGS="$OPTIMIZE_CFLAGS -DDBUG_OFF $CFLAGS -DDEBUG_OFF"
-  CXXFLAGS="$OPTIMIZE_CXXFLAGS -DDBUG_OFF $CXXFLAGS -DDEBUG_OFF"
-fi
-
-# NOTE: The flags below are disabled by default since we can't easily get
-# rid of the "string over 509 characters in length" warnings, and thus can't
-# add -Werror. But it's a good idea to enable these for a test compile
-# before shipping a new snapshot to MySQL to catch errors that could make
-# the compile fail on non-C99 compilers.
-
-# If using gcc, disallow usage of C99 features to avoid accidentally
-# introducing problems on compilers that only implement C89.
-#if test "$ac_cv_prog_gcc" = "yes"
-#then
-#  CFLAGS="$CFLAGS -std=c89 -ansi -pedantic -Wno-long-long"
-#fi
-
-# If using gcc, add some extra warning flags.
-if test "$ac_cv_prog_gcc" = "yes"
-then
-  CFLAGS="$CFLAGS -Werror-implicit-function-declaration -Wpointer-arith"
-fi
-
-case "$target_os" in
-       lin*)
-	 CFLAGS="$CFLAGS -DUNIV_LINUX";;
-       hpux10*)
-	 CFLAGS="$CFLAGS -DUNIV_MUST_NOT_INLINE -DUNIV_HPUX -DUNIV_HPUX10";;
-       hp*)
-	 CFLAGS="$CFLAGS -DUNIV_MUST_NOT_INLINE -DUNIV_HPUX";;
-       aix*)
-         CFLAGS="$CFLAGS -DUNIV_AIX";;
-       irix*)
-	 CFLAGS="$CFLAGS -DUNIV_MUST_NOT_INLINE";;
-       osf*)
-	 CFLAGS="$CFLAGS -DUNIV_MUST_NOT_INLINE";;
-       sysv5uw7*)
-	 # Problem when linking on SCO
-	 CFLAGS="$CFLAGS -DUNIV_MUST_NOT_INLINE";;
-       openbsd*)
-         CFLAGS="$CFLAGS -DUNIV_MUST_NOT_INLINE";;
-esac
-
-case "$target" in
-       i[[4567]]86-*-*)
-	 CFLAGS="$CFLAGS -DUNIV_INTEL_X86";;
-	 # The compiler on Linux/S390 does not seem to have inlining
-       s390-*-*)
-	 CFLAGS="$CFLAGS -DUNIV_MUST_NOT_INLINE";;
-esac
-
-# must go in pair with AR as set by MYSQL_CHECK_AR
-if test -z "$ARFLAGS"
-then
-  ARFLAGS="cru"
-fi
-AC_SUBST(ARFLAGS)
-
-AC_OUTPUT(Makefile os/Makefile ut/Makefile btr/Makefile dnl
-		buf/Makefile data/Makefile dnl
-		dict/Makefile dyn/Makefile dnl
-		eval/Makefile fil/Makefile fsp/Makefile fut/Makefile dnl
-		ha/Makefile ibuf/Makefile dnl
-		lock/Makefile log/Makefile dnl
-		mach/Makefile mem/Makefile mtr/Makefile dnl
-		page/Makefile pars/Makefile que/Makefile dnl
-		read/Makefile rem/Makefile row/Makefile dnl
-		srv/Makefile sync/Makefile thr/Makefile trx/Makefile dnl
-		usr/Makefile)
diff --git a/storage/innobase/include/Makefile.i b/storage/innobase/include/Makefile.i
index 87952a7abc8..db436c702ff 100644
--- a/storage/innobase/include/Makefile.i
+++ b/storage/innobase/include/Makefile.i
@@ -1,6 +1,10 @@
 # Makefile included in Makefile.am in every subdirectory
 
-INCLUDES =		-I$(top_srcdir)/include -I$(top_srcdir)/../../include
+INCLUDES =              -I$(top_srcdir)/include -I$(top_builddir)/include \
+			-I$(top_srcdir)/regex \
+			-I$(top_srcdir)/storage/innobase/include \
+			-I$(top_srcdir)/sql \
+                        -I$(srcdir)
 
 # Don't update the files from bitkeeper
 %::SCCS/s.%
diff --git a/storage/innobase/include/univ.i b/storage/innobase/include/univ.i
index f2dafbc3a70..c1f028ef4a3 100644
--- a/storage/innobase/include/univ.i
+++ b/storage/innobase/include/univ.i
@@ -41,7 +41,7 @@ if we are compiling on Windows. */
 
 /* Include the header file generated by GNU autoconf */
 #ifndef __WIN__
-#include "../ib_config.h"
+#include "config.h"
 #endif
 
 #ifdef HAVE_SCHED_H
diff --git a/storage/innobase/plug.in b/storage/innobase/plug.in
new file mode 100644
index 00000000000..fc1d758fd87
--- /dev/null
+++ b/storage/innobase/plug.in
@@ -0,0 +1,70 @@
+MYSQL_STORAGE_ENGINE(innobase,  innodb, [InnoDB Storage Engine],
+        [Transactional Tables using InnoDB], [max,max-no-ndb])
+MYSQL_PLUGIN_DIRECTORY(innobase, [storage/innobase])
+MYSQL_PLUGIN_STATIC(innobase,   [libinnobase.a])
+MYSQL_PLUGIN_ACTIONS(innobase,  [
+  AC_CHECK_LIB(rt, aio_read, [innodb_system_libs="-lrt"])
+  AC_SUBST(innodb_system_libs)
+  AC_PROG_CC
+  AC_PROG_RANLIB
+  AC_PROG_INSTALL
+  AC_PROG_LIBTOOL
+  AC_CHECK_HEADERS(aio.h sched.h)
+  AC_CHECK_SIZEOF(int, 4)
+  AC_CHECK_SIZEOF(long, 4)
+  AC_CHECK_SIZEOF(void*, 4)
+  AC_CHECK_FUNCS(sched_yield)
+  AC_CHECK_FUNCS(fdatasync)
+  AC_CHECK_FUNCS(localtime_r)
+  AC_C_BIGENDIAN
+  case "$target_os" in
+         lin*)
+           CFLAGS="$CFLAGS -DUNIV_LINUX";;
+         hpux10*)
+           CFLAGS="$CFLAGS -DUNIV_MUST_NOT_INLINE -DUNIV_HPUX -DUNIV_HPUX10";;
+         hp*)
+           CFLAGS="$CFLAGS -DUNIV_MUST_NOT_INLINE -DUNIV_HPUX";;
+         aix*)
+           CFLAGS="$CFLAGS -DUNIV_AIX";;
+         irix*)
+           CFLAGS="$CFLAGS -DUNIV_MUST_NOT_INLINE";;
+         osf*)
+           CFLAGS="$CFLAGS -DUNIV_MUST_NOT_INLINE";;
+         sysv5uw7*)
+           # Problem when linking on SCO
+           CFLAGS="$CFLAGS -DUNIV_MUST_NOT_INLINE";;
+         openbsd*)
+           CFLAGS="$CFLAGS -DUNIV_MUST_NOT_INLINE";;
+  esac
+  AC_CONFIG_FILES(
+      storage/innobase/ut/Makefile
+      storage/innobase/btr/Makefile
+      storage/innobase/buf/Makefile
+      storage/innobase/data/Makefile
+      storage/innobase/dict/Makefile
+      storage/innobase/dyn/Makefile
+      storage/innobase/eval/Makefile
+      storage/innobase/fil/Makefile
+      storage/innobase/fsp/Makefile
+      storage/innobase/fut/Makefile
+      storage/innobase/ha/Makefile
+      storage/innobase/ibuf/Makefile
+      storage/innobase/lock/Makefile
+      storage/innobase/log/Makefile
+      storage/innobase/mach/Makefile
+      storage/innobase/mem/Makefile
+      storage/innobase/mtr/Makefile
+      storage/innobase/os/Makefile
+      storage/innobase/page/Makefile
+      storage/innobase/pars/Makefile
+      storage/innobase/que/Makefile
+      storage/innobase/read/Makefile
+      storage/innobase/rem/Makefile
+      storage/innobase/row/Makefile
+      storage/innobase/srv/Makefile
+      storage/innobase/sync/Makefile
+      storage/innobase/thr/Makefile
+      storage/innobase/trx/Makefile
+      storage/innobase/usr/Makefile)
+  ])
+
diff --git a/storage/myisam/CMakeLists.txt b/storage/myisam/CMakeLists.txt
index 3ba7aba4555..046e4fe28cd 100644
--- a/storage/myisam/CMakeLists.txt
+++ b/storage/myisam/CMakeLists.txt
@@ -1,8 +1,12 @@
 SET(CMAKE_CXX_FLAGS_DEBUG "${CMAKE_CXX_FLAGS_DEBUG} -DSAFEMALLOC -DSAFE_MUTEX")
 SET(CMAKE_C_FLAGS_DEBUG "${CMAKE_C_FLAGS_DEBUG} -DSAFEMALLOC -DSAFE_MUTEX")
 
-INCLUDE_DIRECTORIES(${CMAKE_SOURCE_DIR}/include)
+INCLUDE_DIRECTORIES(${CMAKE_SOURCE_DIR}/include ${CMAKE_SOURCE_DIR}/zlib
+                    ${CMAKE_SOURCE_DIR}/sql
+                    ${CMAKE_SOURCE_DIR}/regex
+                    ${CMAKE_SOURCE_DIR}/extra/yassl/include)
 ADD_LIBRARY(myisam ft_boolean_search.c ft_nlq_search.c ft_parser.c ft_static.c ft_stem.c
+                                ha_myisam.cc
 				ft_stopwords.c ft_update.c mi_cache.c mi_changed.c mi_check.c
 				mi_checksum.c mi_close.c mi_create.c mi_dbug.c mi_delete.c 
 				mi_delete_all.c mi_delete_table.c mi_dynrec.c mi_extra.c mi_info.c
diff --git a/storage/myisam/Makefile.am b/storage/myisam/Makefile.am
index 081d7facf3a..fdccb1f5b19 100644
--- a/storage/myisam/Makefile.am
+++ b/storage/myisam/Makefile.am
@@ -14,29 +14,76 @@
 # along with this program; if not, write to the Free Software
 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
+MYSQLDATAdir =          $(localstatedir)
+MYSQLSHAREdir =         $(pkgdatadir)
+MYSQLBASEdir=           $(prefix)
+MYSQLLIBdir=            $(pkglibdir)
+INCLUDES =              -I$(top_srcdir)/include -I$(top_builddir)/include \
+			-I$(top_srcdir)/regex \
+			-I$(top_srcdir)/sql \
+                        -I$(srcdir)
+WRAPLIBS=
+
+LDADD =
+
+DEFS =                  @DEFS@
+
 EXTRA_DIST =		mi_test_all.sh mi_test_all.res ft_stem.c CMakeLists.txt
 pkgdata_DATA =		mi_test_all mi_test_all.res
 
-INCLUDES =		-I$(top_builddir)/include -I$(top_srcdir)/include
-LDADD =			@CLIENT_EXTRA_LDFLAGS@ libmyisam.a \
-			$(top_builddir)/mysys/libmysys.a \
-			$(top_builddir)/dbug/libdbug.a \
-			$(top_builddir)/strings/libmystrings.a @ZLIB_LIBS@
 pkglib_LIBRARIES =	libmyisam.a
 bin_PROGRAMS =		myisamchk myisamlog myisampack myisam_ftdump
 myisamchk_DEPENDENCIES=	$(LIBRARIES)
+myisamchk_LDADD=		@CLIENT_EXTRA_LDFLAGS@ libmyisam.a \
+			$(top_builddir)/mysys/libmysys.a \
+			$(top_builddir)/dbug/libdbug.a \
+			$(top_builddir)/strings/libmystrings.a @ZLIB_LIBS@
 myisamlog_DEPENDENCIES=	$(LIBRARIES)
+myisamlog_LDADD=		@CLIENT_EXTRA_LDFLAGS@ libmyisam.a \
+			$(top_builddir)/mysys/libmysys.a \
+			$(top_builddir)/dbug/libdbug.a \
+			$(top_builddir)/strings/libmystrings.a @ZLIB_LIBS@
 myisampack_DEPENDENCIES=$(LIBRARIES)
+myisampack_LDADD=		@CLIENT_EXTRA_LDFLAGS@ libmyisam.a \
+			$(top_builddir)/mysys/libmysys.a \
+			$(top_builddir)/dbug/libdbug.a \
+			$(top_builddir)/strings/libmystrings.a @ZLIB_LIBS@
 noinst_PROGRAMS =	mi_test1 mi_test2 mi_test3 rt_test sp_test #ft_test1 ft_eval
-noinst_HEADERS =	myisamdef.h rt_index.h rt_key.h rt_mbr.h sp_defs.h fulltext.h ftdefs.h ft_test1.h ft_eval.h
+noinst_HEADERS =	myisamdef.h rt_index.h rt_key.h rt_mbr.h sp_defs.h \
+			fulltext.h ftdefs.h ft_test1.h ft_eval.h \
+			ha_myisam.h
 mi_test1_DEPENDENCIES=	$(LIBRARIES)
+mi_test1_LDADD=		@CLIENT_EXTRA_LDFLAGS@ libmyisam.a \
+			$(top_builddir)/mysys/libmysys.a \
+			$(top_builddir)/dbug/libdbug.a \
+			$(top_builddir)/strings/libmystrings.a @ZLIB_LIBS@
 mi_test2_DEPENDENCIES=	$(LIBRARIES)
+mi_test2_LDADD=		@CLIENT_EXTRA_LDFLAGS@ libmyisam.a \
+			$(top_builddir)/mysys/libmysys.a \
+			$(top_builddir)/dbug/libdbug.a \
+			$(top_builddir)/strings/libmystrings.a @ZLIB_LIBS@
 mi_test3_DEPENDENCIES=	$(LIBRARIES)
+mi_test3_LDADD=		@CLIENT_EXTRA_LDFLAGS@ libmyisam.a \
+			$(top_builddir)/mysys/libmysys.a \
+			$(top_builddir)/dbug/libdbug.a \
+			$(top_builddir)/strings/libmystrings.a @ZLIB_LIBS@
 #ft_test1_DEPENDENCIES=	$(LIBRARIES)
 #ft_eval_DEPENDENCIES=	$(LIBRARIES)
 myisam_ftdump_DEPENDENCIES=	$(LIBRARIES)
+myisam_ftdump_LDADD =			@CLIENT_EXTRA_LDFLAGS@ libmyisam.a \
+			$(top_builddir)/mysys/libmysys.a \
+			$(top_builddir)/dbug/libdbug.a \
+			$(top_builddir)/strings/libmystrings.a @ZLIB_LIBS@
 rt_test_DEPENDENCIES=	$(LIBRARIES)
+rt_test_LDADD =			@CLIENT_EXTRA_LDFLAGS@ libmyisam.a \
+			$(top_builddir)/mysys/libmysys.a \
+			$(top_builddir)/dbug/libdbug.a \
+			$(top_builddir)/strings/libmystrings.a @ZLIB_LIBS@
 sp_test_DEPENDENCIES=	$(LIBRARIES)
+sp_test_LDADD =			@CLIENT_EXTRA_LDFLAGS@ libmyisam.a \
+			$(top_builddir)/mysys/libmysys.a \
+			$(top_builddir)/dbug/libdbug.a \
+			$(top_builddir)/strings/libmystrings.a @ZLIB_LIBS@
 libmyisam_a_SOURCES =	mi_open.c mi_extra.c mi_info.c mi_rkey.c \
 			mi_rnext.c mi_rnext_same.c \
 			mi_search.c mi_page.c mi_key.c mi_locking.c \
@@ -52,9 +99,9 @@ libmyisam_a_SOURCES =	mi_open.c mi_extra.c mi_info.c mi_rkey.c \
 			mi_keycache.c mi_preload.c \
 			ft_parser.c ft_stopwords.c ft_static.c \
 			ft_update.c ft_boolean_search.c ft_nlq_search.c sort.c \
+			ha_myisam.cc \
 			rt_index.c rt_key.c rt_mbr.c rt_split.c sp_key.c
 CLEANFILES =		test?.MY? FT?.MY? isam.log mi_test_all rt_test.MY? sp_test.MY?
-DEFS =			-DMAP_TO_USE_RAID
 
 # Move to automake rules ?
 prolint:;	plparse -b -u -hF1 "-width(0,0)" "-format=%f:%l:\s%t:%n\s%m" \
diff --git a/storage/myisam/ha_myisam.cc b/storage/myisam/ha_myisam.cc
new file mode 100644
index 00000000000..209478ee9a5
--- /dev/null
+++ b/storage/myisam/ha_myisam.cc
@@ -0,0 +1,1806 @@
+/* Copyright (C) 2000,2004 MySQL AB & MySQL Finland AB & TCX DataKonsult 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 */
+
+
+#ifdef USE_PRAGMA_IMPLEMENTATION
+#pragma implementation				// gcc: Class implementation
+#endif
+
+#define MYSQL_SERVER 1
+#include "mysql_priv.h"
+#include <mysql/plugin.h>
+#include <m_ctype.h>
+#include <myisampack.h>
+#include "ha_myisam.h"
+#include <stdarg.h>
+#include "myisamdef.h"
+#include "rt_index.h"
+
+ulong myisam_recover_options= HA_RECOVER_NONE;
+
+/* bits in myisam_recover_options */
+const char *myisam_recover_names[] =
+{ "DEFAULT", "BACKUP", "FORCE", "QUICK", NullS};
+TYPELIB myisam_recover_typelib= {array_elements(myisam_recover_names)-1,"",
+				 myisam_recover_names, NULL};
+
+const char *myisam_stats_method_names[] = {"nulls_unequal", "nulls_equal",
+                                           "nulls_ignored", NullS};
+TYPELIB myisam_stats_method_typelib= {
+  array_elements(myisam_stats_method_names) - 1, "",
+  myisam_stats_method_names, NULL};
+
+
+/*****************************************************************************
+** MyISAM tables
+*****************************************************************************/
+
+static handler *myisam_create_handler(TABLE_SHARE *table, MEM_ROOT *mem_root)
+{
+  return new (mem_root) ha_myisam(table);
+}
+
+// collect errors printed by mi_check routines
+
+static void mi_check_print_msg(MI_CHECK *param,	const char* msg_type,
+			       const char *fmt, va_list args)
+{
+  THD* thd = (THD*)param->thd;
+  Protocol *protocol= thd->protocol;
+  uint length, msg_length;
+  char msgbuf[MI_MAX_MSG_BUF];
+  char name[NAME_LEN*2+2];
+
+  msg_length= my_vsnprintf(msgbuf, sizeof(msgbuf), fmt, args);
+  msgbuf[sizeof(msgbuf) - 1] = 0; // healthy paranoia
+
+  DBUG_PRINT(msg_type,("message: %s",msgbuf));
+
+  if (!thd->vio_ok())
+  {
+    sql_print_error(msgbuf);
+    return;
+  }
+
+  if (param->testflag & (T_CREATE_MISSING_KEYS | T_SAFE_REPAIR |
+			 T_AUTO_REPAIR))
+  {
+    my_message(ER_NOT_KEYFILE,msgbuf,MYF(MY_WME));
+    return;
+  }
+  length=(uint) (strxmov(name, param->db_name,".",param->table_name,NullS) -
+		 name);
+  protocol->prepare_for_resend();
+  protocol->store(name, length, system_charset_info);
+  protocol->store(param->op_name, system_charset_info);
+  protocol->store(msg_type, system_charset_info);
+  protocol->store(msgbuf, msg_length, system_charset_info);
+  if (protocol->write())
+    sql_print_error("Failed on my_net_write, writing to stderr instead: %s\n",
+		    msgbuf);
+  return;
+}
+
+extern "C" {
+
+volatile int *killed_ptr(MI_CHECK *param)
+{
+  /* In theory Unsafe conversion, but should be ok for now */
+  return (int*) &(((THD *)(param->thd))->killed);
+}
+
+void mi_check_print_error(MI_CHECK *param, const char *fmt,...)
+{
+  param->error_printed|=1;
+  param->out_flag|= O_DATA_LOST;
+  va_list args;
+  va_start(args, fmt);
+  mi_check_print_msg(param, "error", fmt, args);
+  va_end(args);
+}
+
+void mi_check_print_info(MI_CHECK *param, const char *fmt,...)
+{
+  va_list args;
+  va_start(args, fmt);
+  mi_check_print_msg(param, "info", fmt, args);
+  va_end(args);
+}
+
+void mi_check_print_warning(MI_CHECK *param, const char *fmt,...)
+{
+  param->warning_printed=1;
+  param->out_flag|= O_DATA_LOST;
+  va_list args;
+  va_start(args, fmt);
+  mi_check_print_msg(param, "warning", fmt, args);
+  va_end(args);
+}
+
+}
+
+
+ha_myisam::ha_myisam(TABLE_SHARE *table_arg)
+  :handler(&myisam_hton, table_arg), file(0),
+  int_table_flags(HA_NULL_IN_KEY | HA_CAN_FULLTEXT | HA_CAN_SQL_HANDLER |
+                  HA_DUPLICATE_POS | HA_CAN_INDEX_BLOBS | HA_AUTO_PART_KEY |
+                  HA_FILE_BASED | HA_CAN_GEOMETRY | HA_NO_TRANSACTIONS |
+                  HA_CAN_INSERT_DELAYED | HA_CAN_BIT_FIELD | HA_CAN_RTREEKEYS |
+                  HA_HAS_RECORDS | HA_STATS_RECORDS_IS_EXACT),
+   can_enable_indexes(1)
+{}
+
+
+static const char *ha_myisam_exts[] = {
+  ".MYI",
+  ".MYD",
+  NullS
+};
+
+const char **ha_myisam::bas_ext() const
+{
+  return ha_myisam_exts;
+}
+
+
+const char *ha_myisam::index_type(uint key_number)
+{
+  return ((table->key_info[key_number].flags & HA_FULLTEXT) ? 
+	  "FULLTEXT" :
+	  (table->key_info[key_number].flags & HA_SPATIAL) ?
+	  "SPATIAL" :
+	  (table->key_info[key_number].algorithm == HA_KEY_ALG_RTREE) ?
+	  "RTREE" :
+	  "BTREE");
+}
+
+#ifdef HAVE_REPLICATION
+int ha_myisam::net_read_dump(NET* net)
+{
+  int data_fd = file->dfile;
+  int error = 0;
+
+  my_seek(data_fd, 0L, MY_SEEK_SET, MYF(MY_WME));
+  for (;;)
+  {
+    ulong packet_len = my_net_read(net);
+    if (!packet_len)
+      break ; // end of file
+    if (packet_len == packet_error)
+    {
+      sql_print_error("ha_myisam::net_read_dump - read error ");
+      error= -1;
+      goto err;
+    }
+    if (my_write(data_fd, (byte*)net->read_pos, (uint) packet_len,
+		 MYF(MY_WME|MY_FNABP)))
+    {
+      error = errno;
+      goto err;
+    }
+  }
+err:
+  return error;
+}
+
+
+int ha_myisam::dump(THD* thd, int fd)
+{
+  MYISAM_SHARE* share = file->s;
+  NET* net = &thd->net;
+  uint blocksize = share->blocksize;
+  my_off_t bytes_to_read = share->state.state.data_file_length;
+  int data_fd = file->dfile;
+  byte * buf = (byte*) my_malloc(blocksize, MYF(MY_WME));
+  if (!buf)
+    return ENOMEM;
+
+  int error = 0;
+  my_seek(data_fd, 0L, MY_SEEK_SET, MYF(MY_WME));
+  for (; bytes_to_read > 0;)
+  {
+    uint bytes = my_read(data_fd, buf, blocksize, MYF(MY_WME));
+    if (bytes == MY_FILE_ERROR)
+    {
+      error = errno;
+      goto err;
+    }
+
+    if (fd >= 0)
+    {
+      if (my_write(fd, buf, bytes, MYF(MY_WME | MY_FNABP)))
+      {
+	error = errno ? errno : EPIPE;
+	goto err;
+      }
+    }
+    else
+    {
+      if (my_net_write(net, (char*) buf, bytes))
+      {
+	error = errno ? errno : EPIPE;
+	goto err;
+      }
+    }
+    bytes_to_read -= bytes;
+  }
+
+  if (fd < 0)
+  {
+    if (my_net_write(net, "", 0))
+      error = errno ? errno : EPIPE;
+    net_flush(net);
+  }
+
+err:
+  my_free((gptr) buf, MYF(0));
+  return error;
+}
+#endif /* HAVE_REPLICATION */
+
+
+bool ha_myisam::check_if_locking_is_allowed(uint sql_command,
+                                            ulong type, TABLE *table,
+                                            uint count,
+                                            bool called_by_logger_thread)
+{
+  /*
+    To be able to open and lock for reading system tables like 'mysql.proc',
+    when we already have some tables opened and locked, and avoid deadlocks
+    we have to disallow write-locking of these tables with any other tables.
+  */
+  if (table->s->system_table &&
+      table->reginfo.lock_type >= TL_WRITE_ALLOW_WRITE &&
+      count != 1)
+  {
+    my_error(ER_WRONG_LOCK_OF_SYSTEM_TABLE, MYF(0), table->s->db.str,
+             table->s->table_name.str);
+    return FALSE;
+  }
+
+  /*
+    Deny locking of the log tables, which is incompatible with
+    concurrent insert. Unless called from a logger THD:
+    general_log_thd or slow_log_thd.
+  */
+  if (!called_by_logger_thread)
+    return check_if_log_table_locking_is_allowed(sql_command, type, table);
+
+  return TRUE;
+}
+
+	/* Name is here without an extension */
+
+int ha_myisam::open(const char *name, int mode, uint test_if_locked)
+{
+  uint i;
+  if (!(file=mi_open(name, mode, test_if_locked | HA_OPEN_FROM_SQL_LAYER)))
+    return (my_errno ? my_errno : -1);
+  
+  if (test_if_locked & (HA_OPEN_IGNORE_IF_LOCKED | HA_OPEN_TMP_TABLE))
+    VOID(mi_extra(file, HA_EXTRA_NO_WAIT_LOCK, 0));
+
+  if (!(test_if_locked & HA_OPEN_TMP_TABLE) && opt_myisam_use_mmap)
+    VOID(mi_extra(file, HA_EXTRA_MMAP, 0));
+
+  info(HA_STATUS_NO_LOCK | HA_STATUS_VARIABLE | HA_STATUS_CONST);
+  if (!(test_if_locked & HA_OPEN_WAIT_IF_LOCKED))
+    VOID(mi_extra(file, HA_EXTRA_WAIT_LOCK, 0));
+  if (!table->s->db_record_offset)
+    int_table_flags|=HA_REC_NOT_IN_SEQ;
+  if (file->s->options & (HA_OPTION_CHECKSUM | HA_OPTION_COMPRESS_RECORD))
+    int_table_flags|=HA_HAS_CHECKSUM;
+
+  for (i= 0; i < table->s->keys; i++)
+  {
+    struct st_plugin_int *parser= table->key_info[i].parser;
+    if (table->key_info[i].flags & HA_USES_PARSER)
+      file->s->keyinfo[i].parser=
+        (struct st_mysql_ftparser *)parser->plugin->info;
+    table->key_info[i].block_size= file->s->keyinfo[i].block_length;
+  }
+  return (0);
+}
+
+int ha_myisam::close(void)
+{
+  MI_INFO *tmp=file;
+  file=0;
+  return mi_close(tmp);
+}
+
+int ha_myisam::write_row(byte * buf)
+{
+  statistic_increment(table->in_use->status_var.ha_write_count,&LOCK_status);
+
+  /* If we have a timestamp column, update it to the current time */
+  if (table->timestamp_field_type & TIMESTAMP_AUTO_SET_ON_INSERT)
+    table->timestamp_field->set_time();
+
+  /*
+    If we have an auto_increment column and we are writing a changed row
+    or a new row, then update the auto_increment value in the record.
+  */
+  if (table->next_number_field && buf == table->record[0])
+    update_auto_increment();
+  return mi_write(file,buf);
+}
+
+int ha_myisam::check(THD* thd, HA_CHECK_OPT* check_opt)
+{
+  if (!file) return HA_ADMIN_INTERNAL_ERROR;
+  int error;
+  MI_CHECK param;
+  MYISAM_SHARE* share = file->s;
+  const char *old_proc_info=thd->proc_info;
+
+  thd->proc_info="Checking table";
+  myisamchk_init(&param);
+  param.thd = thd;
+  param.op_name =   "check";
+  param.db_name=    table->s->db.str;
+  param.table_name= table->alias;
+  param.testflag = check_opt->flags | T_CHECK | T_SILENT;
+  param.stats_method= (enum_mi_stats_method)thd->variables.myisam_stats_method;
+
+  if (!(table->db_stat & HA_READ_ONLY))
+    param.testflag|= T_STATISTICS;
+  param.using_global_keycache = 1;
+
+  if (!mi_is_crashed(file) &&
+      (((param.testflag & T_CHECK_ONLY_CHANGED) &&
+	!(share->state.changed & (STATE_CHANGED | STATE_CRASHED |
+				  STATE_CRASHED_ON_REPAIR)) &&
+	share->state.open_count == 0) ||
+       ((param.testflag & T_FAST) && (share->state.open_count ==
+				      (uint) (share->global_changed ? 1 : 0)))))
+    return HA_ADMIN_ALREADY_DONE;
+
+  error = chk_status(&param, file);		// Not fatal
+  error = chk_size(&param, file);
+  if (!error)
+    error |= chk_del(&param, file, param.testflag);
+  if (!error)
+    error = chk_key(&param, file);
+  if (!error)
+  {
+    if ((!(param.testflag & T_QUICK) &&
+	 ((share->options &
+	   (HA_OPTION_PACK_RECORD | HA_OPTION_COMPRESS_RECORD)) ||
+	  (param.testflag & (T_EXTEND | T_MEDIUM)))) ||
+	mi_is_crashed(file))
+    {
+      uint old_testflag=param.testflag;
+      param.testflag|=T_MEDIUM;
+      if (!(error= init_io_cache(&param.read_cache, file->dfile,
+                                 my_default_record_cache_size, READ_CACHE,
+                                 share->pack.header_length, 1, MYF(MY_WME))))
+      {
+        error= chk_data_link(&param, file, param.testflag & T_EXTEND);
+        end_io_cache(&(param.read_cache));
+      }
+      param.testflag= old_testflag;
+    }
+  }
+  if (!error)
+  {
+    if ((share->state.changed & (STATE_CHANGED |
+				 STATE_CRASHED_ON_REPAIR |
+				 STATE_CRASHED | STATE_NOT_ANALYZED)) ||
+	(param.testflag & T_STATISTICS) ||
+	mi_is_crashed(file))
+    {
+      file->update|=HA_STATE_CHANGED | HA_STATE_ROW_CHANGED;
+      pthread_mutex_lock(&share->intern_lock);
+      share->state.changed&= ~(STATE_CHANGED | STATE_CRASHED |
+			       STATE_CRASHED_ON_REPAIR);
+      if (!(table->db_stat & HA_READ_ONLY))
+	error=update_state_info(&param,file,UPDATE_TIME | UPDATE_OPEN_COUNT |
+				UPDATE_STAT);
+      pthread_mutex_unlock(&share->intern_lock);
+      info(HA_STATUS_NO_LOCK | HA_STATUS_TIME | HA_STATUS_VARIABLE |
+	   HA_STATUS_CONST);
+    }
+  }
+  else if (!mi_is_crashed(file) && !thd->killed)
+  {
+    mi_mark_crashed(file);
+    file->update |= HA_STATE_CHANGED | HA_STATE_ROW_CHANGED;
+  }
+
+  thd->proc_info=old_proc_info;
+  return error ? HA_ADMIN_CORRUPT : HA_ADMIN_OK;
+}
+
+
+/*
+  analyze the key distribution in the table
+  As the table may be only locked for read, we have to take into account that
+  two threads may do an analyze at the same time!
+*/
+
+int ha_myisam::analyze(THD *thd, HA_CHECK_OPT* check_opt)
+{
+  int error=0;
+  MI_CHECK param;
+  MYISAM_SHARE* share = file->s;
+
+  myisamchk_init(&param);
+  param.thd = thd;
+  param.op_name=    "analyze";
+  param.db_name=    table->s->db.str;
+  param.table_name= table->alias;
+  param.testflag= (T_FAST | T_CHECK | T_SILENT | T_STATISTICS |
+                   T_DONT_CHECK_CHECKSUM);
+  param.using_global_keycache = 1;
+  param.stats_method= (enum_mi_stats_method)thd->variables.myisam_stats_method;
+
+  if (!(share->state.changed & STATE_NOT_ANALYZED))
+    return HA_ADMIN_ALREADY_DONE;
+
+  error = chk_key(&param, file);
+  if (!error)
+  {
+    pthread_mutex_lock(&share->intern_lock);
+    error=update_state_info(&param,file,UPDATE_STAT);
+    pthread_mutex_unlock(&share->intern_lock);
+  }
+  else if (!mi_is_crashed(file) && !thd->killed)
+    mi_mark_crashed(file);
+  return error ? HA_ADMIN_CORRUPT : HA_ADMIN_OK;
+}
+
+
+int ha_myisam::restore(THD* thd, HA_CHECK_OPT *check_opt)
+{
+  HA_CHECK_OPT tmp_check_opt;
+  char *backup_dir= thd->lex->backup_dir;
+  char src_path[FN_REFLEN], dst_path[FN_REFLEN];
+  char table_name[FN_REFLEN];
+  int error;
+  const char* errmsg;
+  DBUG_ENTER("restore");
+
+  VOID(tablename_to_filename(table->s->table_name.str, table_name,
+                             sizeof(table_name)));
+
+  if (fn_format_relative_to_data_home(src_path, table_name, backup_dir,
+				      MI_NAME_DEXT))
+    DBUG_RETURN(HA_ADMIN_INVALID);
+
+  strxmov(dst_path, table->s->normalized_path.str, MI_NAME_DEXT, NullS);
+  if (my_copy(src_path, dst_path, MYF(MY_WME)))
+  {
+    error= HA_ADMIN_FAILED;
+    errmsg= "Failed in my_copy (Error %d)";
+    goto err;
+  }
+
+  tmp_check_opt.init();
+  tmp_check_opt.flags |= T_VERY_SILENT | T_CALC_CHECKSUM | T_QUICK;
+  DBUG_RETURN(repair(thd, &tmp_check_opt));
+
+ err:
+  {
+    MI_CHECK param;
+    myisamchk_init(&param);
+    param.thd= thd;
+    param.op_name=    "restore";
+    param.db_name=    table->s->db.str;
+    param.table_name= table->s->table_name.str;
+    param.testflag= 0;
+    mi_check_print_error(&param, errmsg, my_errno);
+    DBUG_RETURN(error);
+  }
+}
+
+
+int ha_myisam::backup(THD* thd, HA_CHECK_OPT *check_opt)
+{
+  char *backup_dir= thd->lex->backup_dir;
+  char src_path[FN_REFLEN], dst_path[FN_REFLEN];
+  char table_name[FN_REFLEN];
+  int error;
+  const char *errmsg;
+  DBUG_ENTER("ha_myisam::backup");
+
+  VOID(tablename_to_filename(table->s->table_name.str, table_name,
+                             sizeof(table_name)));
+
+  if (fn_format_relative_to_data_home(dst_path, table_name, backup_dir,
+				      reg_ext))
+  {
+    errmsg= "Failed in fn_format() for .frm file (errno: %d)";
+    error= HA_ADMIN_INVALID;
+    goto err;
+  }
+
+  strxmov(src_path, table->s->normalized_path.str, reg_ext, NullS);
+  if (my_copy(src_path, dst_path,
+	      MYF(MY_WME | MY_HOLD_ORIGINAL_MODES | MY_DONT_OVERWRITE_FILE)))
+  {
+    error = HA_ADMIN_FAILED;
+    errmsg = "Failed copying .frm file (errno: %d)";
+    goto err;
+  }
+
+  /* Change extension */
+  if (fn_format_relative_to_data_home(dst_path, table_name, backup_dir,
+                                      MI_NAME_DEXT))
+  {
+    errmsg = "Failed in fn_format() for .MYD file (errno: %d)";
+    error = HA_ADMIN_INVALID;
+    goto err;
+  }
+
+  strxmov(src_path, table->s->normalized_path.str, MI_NAME_DEXT, NullS);
+  if (my_copy(src_path, dst_path,
+	      MYF(MY_WME | MY_HOLD_ORIGINAL_MODES | MY_DONT_OVERWRITE_FILE)))
+  {
+    errmsg = "Failed copying .MYD file (errno: %d)";
+    error= HA_ADMIN_FAILED;
+    goto err;
+  }
+  DBUG_RETURN(HA_ADMIN_OK);
+
+ err:
+  {
+    MI_CHECK param;
+    myisamchk_init(&param);
+    param.thd=        thd;
+    param.op_name=    "backup";
+    param.db_name=    table->s->db.str;
+    param.table_name= table->s->table_name.str;
+    param.testflag =  0;
+    mi_check_print_error(&param,errmsg, my_errno);
+    DBUG_RETURN(error);
+  }
+}
+
+
+int ha_myisam::repair(THD* thd, HA_CHECK_OPT *check_opt)
+{
+  int error;
+  MI_CHECK param;
+  ha_rows start_records;
+
+  if (!file) return HA_ADMIN_INTERNAL_ERROR;
+
+  myisamchk_init(&param);
+  param.thd = thd;
+  param.op_name=  "repair";
+  param.testflag= ((check_opt->flags & ~(T_EXTEND)) |
+                   T_SILENT | T_FORCE_CREATE | T_CALC_CHECKSUM |
+                   (check_opt->flags & T_EXTEND ? T_REP : T_REP_BY_SORT));
+  param.sort_buffer_length=  check_opt->sort_buffer_size;
+  start_records=file->state->records;
+  while ((error=repair(thd,param,0)) && param.retry_repair)
+  {
+    param.retry_repair=0;
+    if (test_all_bits(param.testflag,
+		      (uint) (T_RETRY_WITHOUT_QUICK | T_QUICK)))
+    {
+      param.testflag&= ~T_RETRY_WITHOUT_QUICK;
+      sql_print_information("Retrying repair of: '%s' without quick",
+                            table->s->path);
+      continue;
+    }
+    param.testflag&= ~T_QUICK;
+    if ((param.testflag & T_REP_BY_SORT))
+    {
+      param.testflag= (param.testflag & ~T_REP_BY_SORT) | T_REP;
+      sql_print_information("Retrying repair of: '%s' with keycache",
+                            table->s->path);
+      continue;
+    }
+    break;
+  }
+  if (!error && start_records != file->state->records &&
+      !(check_opt->flags & T_VERY_SILENT))
+  {
+    char llbuff[22],llbuff2[22];
+    sql_print_information("Found %s of %s rows when repairing '%s'",
+                          llstr(file->state->records, llbuff),
+                          llstr(start_records, llbuff2),
+                          table->s->path);
+  }
+  return error;
+}
+
+int ha_myisam::optimize(THD* thd, HA_CHECK_OPT *check_opt)
+{
+  int error;
+  if (!file) return HA_ADMIN_INTERNAL_ERROR;
+  MI_CHECK param;
+
+  myisamchk_init(&param);
+  param.thd = thd;
+  param.op_name= "optimize";
+  param.testflag= (check_opt->flags | T_SILENT | T_FORCE_CREATE |
+                   T_REP_BY_SORT | T_STATISTICS | T_SORT_INDEX);
+  param.sort_buffer_length=  check_opt->sort_buffer_size;
+  if ((error= repair(thd,param,1)) && param.retry_repair)
+  {
+    sql_print_warning("Warning: Optimize table got errno %d, retrying",
+                      my_errno);
+    param.testflag&= ~T_REP_BY_SORT;
+    error= repair(thd,param,1);
+  }
+  return error;
+}
+
+
+int ha_myisam::repair(THD *thd, MI_CHECK &param, bool optimize)
+{
+  int error=0;
+  uint local_testflag=param.testflag;
+  bool optimize_done= !optimize, statistics_done=0;
+  const char *old_proc_info=thd->proc_info;
+  char fixed_name[FN_REFLEN];
+  MYISAM_SHARE* share = file->s;
+  ha_rows rows= file->state->records;
+  DBUG_ENTER("ha_myisam::repair");
+
+  param.db_name=    table->s->db.str;
+  param.table_name= table->alias;
+  param.tmpfile_createflag = O_RDWR | O_TRUNC;
+  param.using_global_keycache = 1;
+  param.thd= thd;
+  param.tmpdir= &mysql_tmpdir_list;
+  param.out_flag= 0;
+  strmov(fixed_name,file->filename);
+
+  // Don't lock tables if we have used LOCK TABLE
+  if (!thd->locked_tables && 
+      mi_lock_database(file, table->s->tmp_table ? F_EXTRA_LCK : F_WRLCK))
+  {
+    mi_check_print_error(&param,ER(ER_CANT_LOCK),my_errno);
+    DBUG_RETURN(HA_ADMIN_FAILED);
+  }
+
+  if (!optimize ||
+      ((file->state->del || share->state.split != file->state->records) &&
+       (!(param.testflag & T_QUICK) ||
+	!(share->state.changed & STATE_NOT_OPTIMIZED_KEYS))))
+  {
+    ulonglong key_map= ((local_testflag & T_CREATE_MISSING_KEYS) ?
+			mi_get_mask_all_keys_active(share->base.keys) :
+			share->state.key_map);
+    uint testflag=param.testflag;
+    if (mi_test_if_sort_rep(file,file->state->records,key_map,0) &&
+	(local_testflag & T_REP_BY_SORT))
+    {
+      local_testflag|= T_STATISTICS;
+      param.testflag|= T_STATISTICS;		// We get this for free
+      statistics_done=1;
+      if (thd->variables.myisam_repair_threads>1)
+      {
+        char buf[40];
+        /* TODO: respect myisam_repair_threads variable */
+        my_snprintf(buf, 40, "Repair with %d threads", my_count_bits(key_map));
+        thd->proc_info=buf;
+        error = mi_repair_parallel(&param, file, fixed_name,
+            param.testflag & T_QUICK);
+        thd->proc_info="Repair done"; // to reset proc_info, as
+                                      // it was pointing to local buffer
+      }
+      else
+      {
+        thd->proc_info="Repair by sorting";
+        error = mi_repair_by_sort(&param, file, fixed_name,
+            param.testflag & T_QUICK);
+      }
+    }
+    else
+    {
+      thd->proc_info="Repair with keycache";
+      param.testflag &= ~T_REP_BY_SORT;
+      error=  mi_repair(&param, file, fixed_name,
+			param.testflag & T_QUICK);
+    }
+    param.testflag=testflag;
+    optimize_done=1;
+  }
+  if (!error)
+  {
+    if ((local_testflag & T_SORT_INDEX) &&
+	(share->state.changed & STATE_NOT_SORTED_PAGES))
+    {
+      optimize_done=1;
+      thd->proc_info="Sorting index";
+      error=mi_sort_index(&param,file,fixed_name);
+    }
+    if (!statistics_done && (local_testflag & T_STATISTICS))
+    {
+      if (share->state.changed & STATE_NOT_ANALYZED)
+      {
+	optimize_done=1;
+	thd->proc_info="Analyzing";
+	error = chk_key(&param, file);
+      }
+      else
+	local_testflag&= ~T_STATISTICS;		// Don't update statistics
+    }
+  }
+  thd->proc_info="Saving state";
+  if (!error)
+  {
+    if ((share->state.changed & STATE_CHANGED) || mi_is_crashed(file))
+    {
+      share->state.changed&= ~(STATE_CHANGED | STATE_CRASHED |
+			       STATE_CRASHED_ON_REPAIR);
+      file->update|=HA_STATE_CHANGED | HA_STATE_ROW_CHANGED;
+    }
+    /*
+      the following 'if', thought conceptually wrong,
+      is a useful optimization nevertheless.
+    */
+    if (file->state != &file->s->state.state)
+      file->s->state.state = *file->state;
+    if (file->s->base.auto_key)
+      update_auto_increment_key(&param, file, 1);
+    if (optimize_done)
+      error = update_state_info(&param, file,
+				UPDATE_TIME | UPDATE_OPEN_COUNT |
+				(local_testflag &
+				 T_STATISTICS ? UPDATE_STAT : 0));
+    info(HA_STATUS_NO_LOCK | HA_STATUS_TIME | HA_STATUS_VARIABLE |
+	 HA_STATUS_CONST);
+    if (rows != file->state->records && ! (param.testflag & T_VERY_SILENT))
+    {
+      char llbuff[22],llbuff2[22];
+      mi_check_print_warning(&param,"Number of rows changed from %s to %s",
+			     llstr(rows,llbuff),
+			     llstr(file->state->records,llbuff2));
+    }
+  }
+  else
+  {
+    mi_mark_crashed_on_repair(file);
+    file->update |= HA_STATE_CHANGED | HA_STATE_ROW_CHANGED;
+    update_state_info(&param, file, 0);
+  }
+  thd->proc_info=old_proc_info;
+  if (!thd->locked_tables)
+    mi_lock_database(file,F_UNLCK);
+  DBUG_RETURN(error ? HA_ADMIN_FAILED :
+	      !optimize_done ? HA_ADMIN_ALREADY_DONE : HA_ADMIN_OK);
+}
+
+
+/*
+  Assign table indexes to a specific key cache.
+*/
+
+int ha_myisam::assign_to_keycache(THD* thd, HA_CHECK_OPT *check_opt)
+{
+  KEY_CACHE *new_key_cache= check_opt->key_cache;
+  const char *errmsg= 0;
+  int error= HA_ADMIN_OK;
+  ulonglong map= ~(ulonglong) 0;
+  TABLE_LIST *table_list= table->pos_in_table_list;
+  DBUG_ENTER("ha_myisam::assign_to_keycache");
+
+  /* Check validity of the index references */
+  if (table_list->use_index)
+  {
+    /* We only come here when the user did specify an index map */
+    key_map kmap;
+    if (get_key_map_from_key_list(&kmap, table, table_list->use_index))
+    {
+      errmsg= thd->net.last_error;
+      error= HA_ADMIN_FAILED;
+      goto err;
+    }
+    map= kmap.to_ulonglong();
+  }
+
+  if ((error= mi_assign_to_key_cache(file, map, new_key_cache)))
+  { 
+    char buf[STRING_BUFFER_USUAL_SIZE];
+    my_snprintf(buf, sizeof(buf),
+		"Failed to flush to index file (errno: %d)", error);
+    errmsg= buf;
+    error= HA_ADMIN_CORRUPT;
+  }
+
+ err:
+  if (error != HA_ADMIN_OK)
+  {
+    /* Send error to user */
+    MI_CHECK param;
+    myisamchk_init(&param);
+    param.thd= thd;
+    param.op_name=    "assign_to_keycache";
+    param.db_name=    table->s->db.str;
+    param.table_name= table->s->table_name.str;
+    param.testflag= 0;
+    mi_check_print_error(&param, errmsg);
+  }
+  DBUG_RETURN(error);
+}
+
+
+/*
+  Preload pages of the index file for a table into the key cache.
+*/
+
+int ha_myisam::preload_keys(THD* thd, HA_CHECK_OPT *check_opt)
+{
+  int error;
+  const char *errmsg;
+  ulonglong map= ~(ulonglong) 0;
+  TABLE_LIST *table_list= table->pos_in_table_list;
+  my_bool ignore_leaves= table_list->ignore_leaves;
+
+  DBUG_ENTER("ha_myisam::preload_keys");
+
+  /* Check validity of the index references */
+  if (table_list->use_index)
+  {
+    key_map kmap;
+    get_key_map_from_key_list(&kmap, table, table_list->use_index);
+    if (kmap.is_set_all())
+    {
+      errmsg= thd->net.last_error;
+      error= HA_ADMIN_FAILED;
+      goto err;
+    }
+    if (!kmap.is_clear_all())
+      map= kmap.to_ulonglong();
+  }
+
+  mi_extra(file, HA_EXTRA_PRELOAD_BUFFER_SIZE,
+           (void *) &thd->variables.preload_buff_size);
+
+  if ((error= mi_preload(file, map, ignore_leaves)))
+  {
+    switch (error) {
+    case HA_ERR_NON_UNIQUE_BLOCK_SIZE:
+      errmsg= "Indexes use different block sizes";
+      break;
+    case HA_ERR_OUT_OF_MEM:
+      errmsg= "Failed to allocate buffer";
+      break;
+    default:
+      char buf[ERRMSGSIZE+20];
+      my_snprintf(buf, ERRMSGSIZE,
+                  "Failed to read from index file (errno: %d)", my_errno);
+      errmsg= buf;
+    }
+    error= HA_ADMIN_FAILED;
+    goto err;
+  }
+
+  DBUG_RETURN(HA_ADMIN_OK);
+
+ err:
+  {
+    MI_CHECK param;
+    myisamchk_init(&param);
+    param.thd= thd;
+    param.op_name=    "preload_keys";
+    param.db_name=    table->s->db.str;
+    param.table_name= table->s->table_name.str;
+    param.testflag=   0;
+    mi_check_print_error(&param, errmsg);
+    DBUG_RETURN(error);
+  }
+}
+
+
+/*
+  Disable indexes, making it persistent if requested.
+
+  SYNOPSIS
+    disable_indexes()
+    mode        mode of operation:
+                HA_KEY_SWITCH_NONUNIQ      disable all non-unique keys
+                HA_KEY_SWITCH_ALL          disable all keys
+                HA_KEY_SWITCH_NONUNIQ_SAVE dis. non-uni. and make persistent
+                HA_KEY_SWITCH_ALL_SAVE     dis. all keys and make persistent
+
+  IMPLEMENTATION
+    HA_KEY_SWITCH_NONUNIQ       is not implemented.
+    HA_KEY_SWITCH_ALL_SAVE      is not implemented.
+
+  RETURN
+    0  ok
+    HA_ERR_WRONG_COMMAND  mode not implemented.
+*/
+
+int ha_myisam::disable_indexes(uint mode)
+{
+  int error;
+
+  if (mode == HA_KEY_SWITCH_ALL)
+  {
+    /* call a storage engine function to switch the key map */
+    error= mi_disable_indexes(file);
+  }
+  else if (mode == HA_KEY_SWITCH_NONUNIQ_SAVE)
+  {
+    mi_extra(file, HA_EXTRA_NO_KEYS, 0);
+    info(HA_STATUS_CONST);                        // Read new key info
+    error= 0;
+  }
+  else
+  {
+    /* mode not implemented */
+    error= HA_ERR_WRONG_COMMAND;
+  }
+  return error;
+}
+
+
+/*
+  Enable indexes, making it persistent if requested.
+
+  SYNOPSIS
+    enable_indexes()
+    mode        mode of operation:
+                HA_KEY_SWITCH_NONUNIQ      enable all non-unique keys
+                HA_KEY_SWITCH_ALL          enable all keys
+                HA_KEY_SWITCH_NONUNIQ_SAVE en. non-uni. and make persistent
+                HA_KEY_SWITCH_ALL_SAVE     en. all keys and make persistent
+
+  DESCRIPTION
+    Enable indexes, which might have been disabled by disable_index() before.
+    The modes without _SAVE work only if both data and indexes are empty,
+    since the MyISAM repair would enable them persistently.
+    To be sure in these cases, call handler::delete_all_rows() before.
+
+  IMPLEMENTATION
+    HA_KEY_SWITCH_NONUNIQ       is not implemented.
+    HA_KEY_SWITCH_ALL_SAVE      is not implemented.
+
+  RETURN
+    0  ok
+    !=0  Error, among others:
+    HA_ERR_CRASHED  data or index is non-empty. Delete all rows and retry.
+    HA_ERR_WRONG_COMMAND  mode not implemented.
+*/
+
+int ha_myisam::enable_indexes(uint mode)
+{
+  int error;
+
+  if (mi_is_all_keys_active(file->s->state.key_map, file->s->base.keys))
+  {
+    /* All indexes are enabled already. */
+    return 0;
+  }
+
+  if (mode == HA_KEY_SWITCH_ALL)
+  {
+    error= mi_enable_indexes(file);
+    /*
+       Do not try to repair on error,
+       as this could make the enabled state persistent,
+       but mode==HA_KEY_SWITCH_ALL forbids it.
+    */
+  }
+  else if (mode == HA_KEY_SWITCH_NONUNIQ_SAVE)
+  {
+    THD *thd=current_thd;
+    MI_CHECK param;
+    const char *save_proc_info=thd->proc_info;
+    thd->proc_info="Creating index";
+    myisamchk_init(&param);
+    param.op_name= "recreating_index";
+    param.testflag= (T_SILENT | T_REP_BY_SORT | T_QUICK |
+                     T_CREATE_MISSING_KEYS);
+    param.myf_rw&= ~MY_WAIT_IF_FULL;
+    param.sort_buffer_length=  thd->variables.myisam_sort_buff_size;
+    param.stats_method= (enum_mi_stats_method)thd->variables.myisam_stats_method;
+    param.tmpdir=&mysql_tmpdir_list;
+    if ((error= (repair(thd,param,0) != HA_ADMIN_OK)) && param.retry_repair)
+    {
+      sql_print_warning("Warning: Enabling keys got errno %d, retrying",
+                        my_errno);
+      /* Repairing by sort failed. Now try standard repair method. */
+      param.testflag&= ~(T_REP_BY_SORT | T_QUICK);
+      error= (repair(thd,param,0) != HA_ADMIN_OK);
+      /*
+        If the standard repair succeeded, clear all error messages which
+        might have been set by the first repair. They can still be seen
+        with SHOW WARNINGS then.
+      */
+      if (! error)
+        thd->clear_error();
+    }
+    info(HA_STATUS_CONST);
+    thd->proc_info=save_proc_info;
+  }
+  else
+  {
+    /* mode not implemented */
+    error= HA_ERR_WRONG_COMMAND;
+  }
+  return error;
+}
+
+
+/*
+  Test if indexes are disabled.
+
+
+  SYNOPSIS
+    indexes_are_disabled()
+      no parameters
+
+
+  RETURN
+    0  indexes are not disabled
+    1  all indexes are disabled
+   [2  non-unique indexes are disabled - NOT YET IMPLEMENTED]
+*/
+
+int ha_myisam::indexes_are_disabled(void)
+{
+  
+  return mi_indexes_are_disabled(file);
+}
+
+
+/*
+  prepare for a many-rows insert operation
+  e.g. - disable indexes (if they can be recreated fast) or
+  activate special bulk-insert optimizations
+
+  SYNOPSIS
+    start_bulk_insert(rows)
+    rows        Rows to be inserted
+                0 if we don't know
+
+  NOTICE
+    Do not forget to call end_bulk_insert() later!
+*/
+
+void ha_myisam::start_bulk_insert(ha_rows rows)
+{
+  DBUG_ENTER("ha_myisam::start_bulk_insert");
+  THD *thd= current_thd;
+  ulong size= min(thd->variables.read_buff_size,
+                  table->s->avg_row_length*rows);
+  DBUG_PRINT("info",("start_bulk_insert: rows %lu size %lu",
+                     (ulong) rows, size));
+
+  /* don't enable row cache if too few rows */
+  if (! rows || (rows > MI_MIN_ROWS_TO_USE_WRITE_CACHE))
+    mi_extra(file, HA_EXTRA_WRITE_CACHE, (void*) &size);
+
+  can_enable_indexes= mi_is_all_keys_active(file->s->state.key_map,
+                                            file->s->base.keys);
+
+  if (!(specialflag & SPECIAL_SAFE_MODE))
+  {
+    /*
+      Only disable old index if the table was empty and we are inserting
+      a lot of rows.
+      We should not do this for only a few rows as this is slower and
+      we don't want to update the key statistics based of only a few rows.
+    */
+    if (file->state->records == 0 && can_enable_indexes &&
+        (!rows || rows >= MI_MIN_ROWS_TO_DISABLE_INDEXES))
+      mi_disable_non_unique_index(file,rows);
+    else
+    if (!file->bulk_insert &&
+        (!rows || rows >= MI_MIN_ROWS_TO_USE_BULK_INSERT))
+    {
+      mi_init_bulk_insert(file, thd->variables.bulk_insert_buff_size, rows);
+    }
+  }
+  DBUG_VOID_RETURN;
+}
+
+/*
+  end special bulk-insert optimizations,
+  which have been activated by start_bulk_insert().
+
+  SYNOPSIS
+    end_bulk_insert()
+    no arguments
+
+  RETURN
+    0     OK
+    != 0  Error
+*/
+
+int ha_myisam::end_bulk_insert()
+{
+  mi_end_bulk_insert(file);
+  int err=mi_extra(file, HA_EXTRA_NO_CACHE, 0);
+  return err ? err : can_enable_indexes ?
+                     enable_indexes(HA_KEY_SWITCH_NONUNIQ_SAVE) : 0;
+}
+
+
+bool ha_myisam::check_and_repair(THD *thd)
+{
+  int error=0;
+  int marked_crashed;
+  char *old_query;
+  uint old_query_length;
+  HA_CHECK_OPT check_opt;
+  DBUG_ENTER("ha_myisam::check_and_repair");
+
+  check_opt.init();
+  check_opt.flags= T_MEDIUM | T_AUTO_REPAIR;
+  // Don't use quick if deleted rows
+  if (!file->state->del && (myisam_recover_options & HA_RECOVER_QUICK))
+    check_opt.flags|=T_QUICK;
+  sql_print_warning("Checking table:   '%s'",table->s->path);
+
+  old_query= thd->query;
+  old_query_length= thd->query_length;
+  pthread_mutex_lock(&LOCK_thread_count);
+  thd->query=        table->s->table_name.str;
+  thd->query_length= table->s->table_name.length;
+  pthread_mutex_unlock(&LOCK_thread_count);
+
+  if ((marked_crashed= mi_is_crashed(file)) || check(thd, &check_opt))
+  {
+    sql_print_warning("Recovering table: '%s'",table->s->path);
+    check_opt.flags=
+      ((myisam_recover_options & HA_RECOVER_BACKUP ? T_BACKUP_DATA : 0) |
+       (marked_crashed                             ? 0 : T_QUICK) |
+       (myisam_recover_options & HA_RECOVER_FORCE  ? 0 : T_SAFE_REPAIR) |
+       T_AUTO_REPAIR);
+    if (repair(thd, &check_opt))
+      error=1;
+  }
+  pthread_mutex_lock(&LOCK_thread_count);
+  thd->query= old_query;
+  thd->query_length= old_query_length;
+  pthread_mutex_unlock(&LOCK_thread_count);
+  DBUG_RETURN(error);
+}
+
+bool ha_myisam::is_crashed() const
+{
+  return (file->s->state.changed & STATE_CRASHED ||
+	  (my_disable_locking && file->s->state.open_count));
+}
+
+int ha_myisam::update_row(const byte * old_data, byte * new_data)
+{
+  statistic_increment(table->in_use->status_var.ha_update_count,&LOCK_status);
+  if (table->timestamp_field_type & TIMESTAMP_AUTO_SET_ON_UPDATE)
+    table->timestamp_field->set_time();
+  return mi_update(file,old_data,new_data);
+}
+
+int ha_myisam::delete_row(const byte * buf)
+{
+  statistic_increment(table->in_use->status_var.ha_delete_count,&LOCK_status);
+  return mi_delete(file,buf);
+}
+
+int ha_myisam::index_read(byte * buf, const byte * key,
+			  uint key_len, enum ha_rkey_function find_flag)
+{
+  DBUG_ASSERT(inited==INDEX);
+  statistic_increment(table->in_use->status_var.ha_read_key_count,
+		      &LOCK_status);
+  int error=mi_rkey(file,buf,active_index, key, key_len, find_flag);
+  table->status=error ? STATUS_NOT_FOUND: 0;
+  return error;
+}
+
+int ha_myisam::index_read_idx(byte * buf, uint index, const byte * key,
+			      uint key_len, enum ha_rkey_function find_flag)
+{
+  statistic_increment(table->in_use->status_var.ha_read_key_count,
+		      &LOCK_status);
+  int error=mi_rkey(file,buf,index, key, key_len, find_flag);
+  table->status=error ? STATUS_NOT_FOUND: 0;
+  return error;
+}
+
+int ha_myisam::index_read_last(byte * buf, const byte * key, uint key_len)
+{
+  DBUG_ENTER("ha_myisam::index_read_last");
+  DBUG_ASSERT(inited==INDEX);
+  statistic_increment(table->in_use->status_var.ha_read_key_count,
+		      &LOCK_status);
+  int error=mi_rkey(file,buf,active_index, key, key_len, HA_READ_PREFIX_LAST);
+  table->status=error ? STATUS_NOT_FOUND: 0;
+  DBUG_RETURN(error);
+}
+
+int ha_myisam::index_next(byte * buf)
+{
+  DBUG_ASSERT(inited==INDEX);
+  statistic_increment(table->in_use->status_var.ha_read_next_count,
+		      &LOCK_status);
+  int error=mi_rnext(file,buf,active_index);
+  table->status=error ? STATUS_NOT_FOUND: 0;
+  return error;
+}
+
+int ha_myisam::index_prev(byte * buf)
+{
+  DBUG_ASSERT(inited==INDEX);
+  statistic_increment(table->in_use->status_var.ha_read_prev_count,
+		      &LOCK_status);
+  int error=mi_rprev(file,buf, active_index);
+  table->status=error ? STATUS_NOT_FOUND: 0;
+  return error;
+}
+
+int ha_myisam::index_first(byte * buf)
+{
+  DBUG_ASSERT(inited==INDEX);
+  statistic_increment(table->in_use->status_var.ha_read_first_count,
+		      &LOCK_status);
+  int error=mi_rfirst(file, buf, active_index);
+  table->status=error ? STATUS_NOT_FOUND: 0;
+  return error;
+}
+
+int ha_myisam::index_last(byte * buf)
+{
+  DBUG_ASSERT(inited==INDEX);
+  statistic_increment(table->in_use->status_var.ha_read_last_count,
+		      &LOCK_status);
+  int error=mi_rlast(file, buf, active_index);
+  table->status=error ? STATUS_NOT_FOUND: 0;
+  return error;
+}
+
+int ha_myisam::index_next_same(byte * buf,
+			       const byte *key __attribute__((unused)),
+			       uint length __attribute__((unused)))
+{
+  DBUG_ASSERT(inited==INDEX);
+  statistic_increment(table->in_use->status_var.ha_read_next_count,
+		      &LOCK_status);
+  int error=mi_rnext_same(file,buf);
+  table->status=error ? STATUS_NOT_FOUND: 0;
+  return error;
+}
+
+
+int ha_myisam::rnd_init(bool scan)
+{
+  if (scan)
+    return mi_scan_init(file);
+  return mi_reset(file);                        // Free buffers
+}
+
+int ha_myisam::rnd_next(byte *buf)
+{
+  statistic_increment(table->in_use->status_var.ha_read_rnd_next_count,
+		      &LOCK_status);
+  int error=mi_scan(file, buf);
+  table->status=error ? STATUS_NOT_FOUND: 0;
+  return error;
+}
+
+int ha_myisam::restart_rnd_next(byte *buf, byte *pos)
+{
+  return rnd_pos(buf,pos);
+}
+
+int ha_myisam::rnd_pos(byte * buf, byte *pos)
+{
+  statistic_increment(table->in_use->status_var.ha_read_rnd_count,
+		      &LOCK_status);
+  int error=mi_rrnd(file, buf, my_get_ptr(pos,ref_length));
+  table->status=error ? STATUS_NOT_FOUND: 0;
+  return error;
+}
+
+void ha_myisam::position(const byte* record)
+{
+  my_off_t position=mi_position(file);
+  my_store_ptr(ref, ref_length, position);
+}
+
+void ha_myisam::info(uint flag)
+{
+  MI_ISAMINFO info;
+  char name_buff[FN_REFLEN];
+
+  (void) mi_status(file,&info,flag);
+  if (flag & HA_STATUS_VARIABLE)
+  {
+    stats.records = info.records;
+    stats.deleted = info.deleted;
+    stats.data_file_length=info.data_file_length;
+    stats.index_file_length=info.index_file_length;
+    stats.delete_length = info.delete_length;
+    stats.check_time  = info.check_time;
+    stats. mean_rec_length=info.mean_reclength;
+  }
+  if (flag & HA_STATUS_CONST)
+  {
+    TABLE_SHARE *share= table->s;
+    stats.max_data_file_length=  info.max_data_file_length;
+    stats.max_index_file_length= info.max_index_file_length;
+    stats.create_time= info.create_time;
+    ref_length= info.reflength;
+    share->db_options_in_use= info.options;
+    stats.block_size= myisam_block_size;        /* record block size */
+
+    /* Update share */
+    if (share->tmp_table == NO_TMP_TABLE)
+      pthread_mutex_lock(&share->mutex);
+    share->keys_in_use.set_prefix(share->keys);
+    share->keys_in_use.intersect_extended(info.key_map);
+    share->keys_for_keyread.intersect(share->keys_in_use);
+    share->db_record_offset= info.record_offset;
+    if (share->key_parts)
+      memcpy((char*) table->key_info[0].rec_per_key,
+	     (char*) info.rec_per_key,
+	     sizeof(table->key_info[0].rec_per_key)*share->key_parts);
+    if (share->tmp_table == NO_TMP_TABLE)
+      pthread_mutex_unlock(&share->mutex);
+
+   /*
+     Set data_file_name and index_file_name to point at the symlink value
+     if table is symlinked (Ie;  Real name is not same as generated name)
+   */
+    data_file_name= index_file_name= 0;
+    fn_format(name_buff, file->filename, "", MI_NAME_DEXT, MY_APPEND_EXT);
+    if (strcmp(name_buff, info.data_file_name))
+      data_file_name=info.data_file_name;
+    fn_format(name_buff, file->filename, "", MI_NAME_IEXT, MY_APPEND_EXT);
+    if (strcmp(name_buff, info.index_file_name))
+      index_file_name=info.index_file_name;
+  }
+  if (flag & HA_STATUS_ERRKEY)
+  {
+    errkey  = info.errkey;
+    my_store_ptr(dup_ref, ref_length, info.dupp_key_pos);
+  }
+  if (flag & HA_STATUS_TIME)
+    stats.update_time = info.update_time;
+  if (flag & HA_STATUS_AUTO)
+    stats.auto_increment_value= info.auto_increment;
+}
+
+
+int ha_myisam::extra(enum ha_extra_function operation)
+{
+  if ((specialflag & SPECIAL_SAFE_MODE) && operation == HA_EXTRA_KEYREAD)
+    return 0;
+  return mi_extra(file, operation, 0);
+}
+
+int ha_myisam::reset(void)
+{
+  return mi_reset(file);
+}
+
+/* To be used with WRITE_CACHE and EXTRA_CACHE */
+
+int ha_myisam::extra_opt(enum ha_extra_function operation, ulong cache_size)
+{
+  if ((specialflag & SPECIAL_SAFE_MODE) && operation == HA_EXTRA_WRITE_CACHE)
+    return 0;
+  return mi_extra(file, operation, (void*) &cache_size);
+}
+
+int ha_myisam::delete_all_rows()
+{
+  return mi_delete_all_rows(file);
+}
+
+int ha_myisam::delete_table(const char *name)
+{
+  return mi_delete_table(name);
+}
+
+
+int ha_myisam::external_lock(THD *thd, int lock_type)
+{
+  return mi_lock_database(file, !table->s->tmp_table ?
+			  lock_type : ((lock_type == F_UNLCK) ?
+				       F_UNLCK : F_EXTRA_LCK));
+}
+
+THR_LOCK_DATA **ha_myisam::store_lock(THD *thd,
+				      THR_LOCK_DATA **to,
+				      enum thr_lock_type lock_type)
+{
+  if (lock_type != TL_IGNORE && file->lock.type == TL_UNLOCK)
+    file->lock.type=lock_type;
+  *to++= &file->lock;
+  return to;
+}
+
+void ha_myisam::update_create_info(HA_CREATE_INFO *create_info)
+{
+  ha_myisam::info(HA_STATUS_AUTO | HA_STATUS_CONST);
+  if (!(create_info->used_fields & HA_CREATE_USED_AUTO))
+  {
+    create_info->auto_increment_value= stats.auto_increment_value;
+  }
+  create_info->data_file_name=data_file_name;
+  create_info->index_file_name=index_file_name;
+}
+
+
+int ha_myisam::create(const char *name, register TABLE *table_arg,
+		      HA_CREATE_INFO *info)
+{
+  int error;
+  uint i,j,recpos,minpos,fieldpos,temp_length,length, create_flags= 0;
+  bool found_real_auto_increment=0;
+  enum ha_base_keytype type;
+  char buff[FN_REFLEN];
+  KEY *pos;
+  MI_KEYDEF *keydef;
+  MI_COLUMNDEF *recinfo,*recinfo_pos;
+  HA_KEYSEG *keyseg;
+  TABLE_SHARE *share= table_arg->s;
+  uint options= share->db_options_in_use;
+  DBUG_ENTER("ha_myisam::create");
+
+  type=HA_KEYTYPE_BINARY;				// Keep compiler happy
+  if (!(my_multi_malloc(MYF(MY_WME),
+			&recinfo,(share->fields*2+2)*
+                        sizeof(MI_COLUMNDEF),
+			&keydef, share->keys*sizeof(MI_KEYDEF),
+			&keyseg,
+			((share->key_parts + share->keys) *
+			 sizeof(HA_KEYSEG)),
+			NullS)))
+    DBUG_RETURN(HA_ERR_OUT_OF_MEM);
+
+  pos=table_arg->key_info;
+  for (i=0; i < share->keys ; i++, pos++)
+  {
+    if (pos->flags & HA_USES_PARSER)
+      create_flags|= HA_CREATE_RELIES_ON_SQL_LAYER;
+    keydef[i].flag= (pos->flags & (HA_NOSAME | HA_FULLTEXT | HA_SPATIAL));
+    keydef[i].key_alg= pos->algorithm == HA_KEY_ALG_UNDEF ? 
+      (pos->flags & HA_SPATIAL ? HA_KEY_ALG_RTREE : HA_KEY_ALG_BTREE) :
+      pos->algorithm;
+    keydef[i].block_length= pos->block_size;
+
+    keydef[i].seg=keyseg;
+    keydef[i].keysegs=pos->key_parts;
+    for (j=0 ; j < pos->key_parts ; j++)
+    {
+      Field *field=pos->key_part[j].field;
+      type=field->key_type();
+      keydef[i].seg[j].flag=pos->key_part[j].key_part_flag;
+
+      if (options & HA_OPTION_PACK_KEYS ||
+	  (pos->flags & (HA_PACK_KEY | HA_BINARY_PACK_KEY |
+			 HA_SPACE_PACK_USED)))
+      {
+	if (pos->key_part[j].length > 8 &&
+	    (type == HA_KEYTYPE_TEXT ||
+	     type == HA_KEYTYPE_NUM ||
+	     (type == HA_KEYTYPE_BINARY && !field->zero_pack())))
+	{
+	  /* No blobs here */
+	  if (j == 0)
+	    keydef[i].flag|=HA_PACK_KEY;
+	  if (!(field->flags & ZEROFILL_FLAG) &&
+	      (field->type() == MYSQL_TYPE_STRING ||
+	       field->type() == MYSQL_TYPE_VAR_STRING ||
+	       ((int) (pos->key_part[j].length - field->decimals()))
+	       >= 4))
+	    keydef[i].seg[j].flag|=HA_SPACE_PACK;
+	}
+	else if (j == 0 && (!(pos->flags & HA_NOSAME) || pos->key_length > 16))
+	  keydef[i].flag|= HA_BINARY_PACK_KEY;
+      }
+      keydef[i].seg[j].type=   (int) type;
+      keydef[i].seg[j].start=  pos->key_part[j].offset;
+      keydef[i].seg[j].length= pos->key_part[j].length;
+      keydef[i].seg[j].bit_start= keydef[i].seg[j].bit_end=
+        keydef[i].seg[j].bit_length= 0;
+      keydef[i].seg[j].bit_pos= 0;
+      keydef[i].seg[j].language= field->charset()->number;
+
+      if (field->null_ptr)
+      {
+	keydef[i].seg[j].null_bit=field->null_bit;
+	keydef[i].seg[j].null_pos= (uint) (field->null_ptr-
+					   (uchar*) table_arg->record[0]);
+      }
+      else
+      {
+	keydef[i].seg[j].null_bit=0;
+	keydef[i].seg[j].null_pos=0;
+      }
+      if (field->type() == FIELD_TYPE_BLOB ||
+	  field->type() == FIELD_TYPE_GEOMETRY)
+      {
+	keydef[i].seg[j].flag|=HA_BLOB_PART;
+	/* save number of bytes used to pack length */
+	keydef[i].seg[j].bit_start= (uint) (field->pack_length() -
+					    share->blob_ptr_size);
+      }
+      else if (field->type() == FIELD_TYPE_BIT)
+      {
+        keydef[i].seg[j].bit_length= ((Field_bit *) field)->bit_len;
+        keydef[i].seg[j].bit_start= ((Field_bit *) field)->bit_ofs;
+        keydef[i].seg[j].bit_pos= (uint) (((Field_bit *) field)->bit_ptr -
+                                          (uchar*) table_arg->record[0]);
+      }
+    }
+    keyseg+=pos->key_parts;
+  }
+
+  if (table_arg->found_next_number_field)
+  {
+    keydef[share->next_number_index].flag|= HA_AUTO_KEY;
+    found_real_auto_increment= share->next_number_key_offset == 0;
+  }
+
+  recpos=0; recinfo_pos=recinfo;
+  while (recpos < (uint) share->reclength)
+  {
+    Field **field,*found=0;
+    minpos= share->reclength;
+    length=0;
+
+    for (field=table_arg->field ; *field ; field++)
+    {
+      if ((fieldpos=(*field)->offset()) >= recpos &&
+	  fieldpos <= minpos)
+      {
+	/* skip null fields */
+	if (!(temp_length= (*field)->pack_length_in_rec()))
+	  continue;				/* Skip null-fields */
+	if (! found || fieldpos < minpos ||
+	    (fieldpos == minpos && temp_length < length))
+	{
+	  minpos=fieldpos; found= *field; length=temp_length;
+	}
+      }
+    }
+    DBUG_PRINT("loop",("found: 0x%lx  recpos: %d  minpos: %d  length: %d",
+		       found,recpos,minpos,length));
+    if (recpos != minpos)
+    {						// Reserved space (Null bits?)
+      bzero((char*) recinfo_pos,sizeof(*recinfo_pos));
+      recinfo_pos->type=(int) FIELD_NORMAL;
+      recinfo_pos++->length= (uint16) (minpos-recpos);
+    }
+    if (! found)
+      break;
+
+    if (found->flags & BLOB_FLAG)
+      recinfo_pos->type= (int) FIELD_BLOB;
+    else if (found->type() == MYSQL_TYPE_VARCHAR)
+      recinfo_pos->type= FIELD_VARCHAR;
+    else if (!(options & HA_OPTION_PACK_RECORD))
+      recinfo_pos->type= (int) FIELD_NORMAL;
+    else if (found->zero_pack())
+      recinfo_pos->type= (int) FIELD_SKIP_ZERO;
+    else
+      recinfo_pos->type= (int) ((length <= 3 ||
+                                 (found->flags & ZEROFILL_FLAG)) ?
+                                FIELD_NORMAL :
+                                found->type() == MYSQL_TYPE_STRING ||
+                                found->type() == MYSQL_TYPE_VAR_STRING ?
+                                FIELD_SKIP_ENDSPACE :
+                                FIELD_SKIP_PRESPACE);
+    if (found->null_ptr)
+    {
+      recinfo_pos->null_bit=found->null_bit;
+      recinfo_pos->null_pos= (uint) (found->null_ptr-
+                                     (uchar*) table_arg->record[0]);
+    }
+    else
+    {
+      recinfo_pos->null_bit=0;
+      recinfo_pos->null_pos=0;
+    }
+    (recinfo_pos++)->length= (uint16) length;
+    recpos=minpos+length;
+    DBUG_PRINT("loop",("length: %d  type: %d",
+		       recinfo_pos[-1].length,recinfo_pos[-1].type));
+
+  }
+  MI_CREATE_INFO create_info;
+  bzero((char*) &create_info,sizeof(create_info));
+  create_info.max_rows= share->max_rows;
+  create_info.reloc_rows= share->min_rows;
+  create_info.with_auto_increment=found_real_auto_increment;
+  create_info.auto_increment=(info->auto_increment_value ?
+			      info->auto_increment_value -1 :
+			      (ulonglong) 0);
+  create_info.data_file_length= ((ulonglong) share->max_rows *
+				 share->avg_row_length);
+  create_info.data_file_name=  info->data_file_name;
+  create_info.index_file_name= info->index_file_name;
+
+  if (info->options & HA_LEX_CREATE_TMP_TABLE)
+    create_flags|= HA_CREATE_TMP_TABLE;
+  if (options & HA_OPTION_PACK_RECORD)
+    create_flags|= HA_PACK_RECORD;
+  if (options & HA_OPTION_CHECKSUM)
+    create_flags|= HA_CREATE_CHECKSUM;
+  if (options & HA_OPTION_DELAY_KEY_WRITE)
+    create_flags|= HA_CREATE_DELAY_KEY_WRITE;
+
+  /* TODO: Check that the following fn_format is really needed */
+  error=mi_create(fn_format(buff,name,"","",MY_UNPACK_FILENAME|MY_APPEND_EXT),
+		  share->keys,keydef,
+		  (uint) (recinfo_pos-recinfo), recinfo,
+		  0, (MI_UNIQUEDEF*) 0,
+		  &create_info, create_flags);
+
+  my_free((gptr) recinfo,MYF(0));
+  DBUG_RETURN(error);
+}
+
+
+int ha_myisam::rename_table(const char * from, const char * to)
+{
+  return mi_rename(from,to);
+}
+
+
+void ha_myisam::get_auto_increment(ulonglong offset, ulonglong increment,
+                                   ulonglong nb_desired_values,
+                                   ulonglong *first_value,
+                                   ulonglong *nb_reserved_values)
+{
+  ulonglong nr;
+  int error;
+  byte key[MI_MAX_KEY_LENGTH];
+
+  if (!table->s->next_number_key_offset)
+  {						// Autoincrement at key-start
+    ha_myisam::info(HA_STATUS_AUTO);
+    *first_value= stats.auto_increment_value;
+    /* MyISAM has only table-level lock, so reserves to +inf */
+    *nb_reserved_values= ULONGLONG_MAX;
+    return;
+  }
+
+  /* it's safe to call the following if bulk_insert isn't on */
+  mi_flush_bulk_insert(file, table->s->next_number_index);
+
+  (void) extra(HA_EXTRA_KEYREAD);
+  key_copy(key, table->record[0],
+           table->key_info + table->s->next_number_index,
+           table->s->next_number_key_offset);
+  error= mi_rkey(file,table->record[1],(int) table->s->next_number_index,
+                 key,table->s->next_number_key_offset,HA_READ_PREFIX_LAST);
+  if (error)
+    nr= 1;
+  else
+  {
+    /* Get data from record[1] */
+    nr= ((ulonglong) table->next_number_field->
+         val_int_offset(table->s->rec_buff_length)+1);
+  }
+  extra(HA_EXTRA_NO_KEYREAD);
+  *first_value= nr;
+  /*
+    MySQL needs to call us for next row: assume we are inserting ("a",null)
+    here, we return 3, and next this statement will want to insert ("b",null):
+    there is no reason why ("b",3+1) would be the good row to insert: maybe it
+    already exists, maybe 3+1 is too large...
+  */
+  *nb_reserved_values= 1;
+}
+
+
+/*
+  Find out how many rows there is in the given range
+
+  SYNOPSIS
+    records_in_range()
+    inx			Index to use
+    min_key		Start of range.  Null pointer if from first key
+    max_key		End of range. Null pointer if to last key
+
+  NOTES
+    min_key.flag can have one of the following values:
+      HA_READ_KEY_EXACT		Include the key in the range
+      HA_READ_AFTER_KEY		Don't include key in range
+
+    max_key.flag can have one of the following values:  
+      HA_READ_BEFORE_KEY	Don't include key in range
+      HA_READ_AFTER_KEY		Include all 'end_key' values in the range
+
+  RETURN
+   HA_POS_ERROR		Something is wrong with the index tree.
+   0			There is no matching keys in the given range
+   number > 0		There is approximately 'number' matching rows in
+			the range.
+*/
+
+ha_rows ha_myisam::records_in_range(uint inx, key_range *min_key,
+                                    key_range *max_key)
+{
+  return (ha_rows) mi_records_in_range(file, (int) inx, min_key, max_key);
+}
+
+
+int ha_myisam::ft_read(byte * buf)
+{
+  int error;
+
+  if (!ft_handler)
+    return -1;
+
+  thread_safe_increment(table->in_use->status_var.ha_read_next_count,
+			&LOCK_status); // why ?
+
+  error=ft_handler->please->read_next(ft_handler,(char*) buf);
+
+  table->status=error ? STATUS_NOT_FOUND: 0;
+  return error;
+}
+
+uint ha_myisam::checksum() const
+{
+  return (uint)file->state->checksum;
+}
+
+
+bool ha_myisam::check_if_incompatible_data(HA_CREATE_INFO *info,
+					   uint table_changes)
+{
+  uint options= table->s->db_options_in_use;
+
+  if (info->auto_increment_value != stats.auto_increment_value ||
+      info->data_file_name != data_file_name ||
+      info->index_file_name != index_file_name ||
+      table_changes == IS_EQUAL_NO ||
+      table_changes & IS_EQUAL_PACK_LENGTH) // Not implemented yet
+    return COMPATIBLE_DATA_NO;
+
+  if ((options & (HA_OPTION_PACK_RECORD | HA_OPTION_CHECKSUM |
+		  HA_OPTION_DELAY_KEY_WRITE)) !=
+      (info->table_options & (HA_OPTION_PACK_RECORD | HA_OPTION_CHECKSUM |
+			      HA_OPTION_DELAY_KEY_WRITE)))
+    return COMPATIBLE_DATA_NO;
+  return COMPATIBLE_DATA_YES;
+}
+
+handlerton myisam_hton;
+
+static int myisam_init()
+{
+  myisam_hton.state=SHOW_OPTION_YES;
+  myisam_hton.db_type=DB_TYPE_MYISAM;
+  myisam_hton.create=myisam_create_handler;
+  myisam_hton.panic=mi_panic;
+  myisam_hton.flags=HTON_CAN_RECREATE;
+  return 0;
+}
+
+struct st_mysql_storage_engine myisam_storage_engine=
+{ MYSQL_HANDLERTON_INTERFACE_VERSION, &myisam_hton };
+
+mysql_declare_plugin(myisam)
+{
+  MYSQL_STORAGE_ENGINE_PLUGIN,
+  &myisam_storage_engine,
+  "MyISAM",
+  "MySQL AB",
+  "Default engine as of MySQL 3.23 with great performance",
+  myisam_init, /* Plugin Init */
+  NULL, /* Plugin Deinit */
+  0x0100, /* 1.0 */
+  0
+}
+mysql_declare_plugin_end;
+
diff --git a/storage/myisam/ha_myisam.h b/storage/myisam/ha_myisam.h
new file mode 100644
index 00000000000..5544e5040b3
--- /dev/null
+++ b/storage/myisam/ha_myisam.h
@@ -0,0 +1,139 @@
+/* Copyright (C) 2000,2004 MySQL AB & MySQL Finland AB & TCX DataKonsult 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 */
+
+
+#ifdef USE_PRAGMA_INTERFACE
+#pragma interface			/* gcc class implementation */
+#endif
+
+/* class for the the myisam handler */
+
+#include <myisam.h>
+#include <ft_global.h>
+
+#define HA_RECOVER_NONE		0	/* No automatic recover */
+#define HA_RECOVER_DEFAULT	1	/* Automatic recover active */
+#define HA_RECOVER_BACKUP	2	/* Make a backupfile on recover */
+#define HA_RECOVER_FORCE	4	/* Recover even if we loose rows */
+#define HA_RECOVER_QUICK	8	/* Don't check rows in data file */
+
+extern ulong myisam_sort_buffer_size;
+extern TYPELIB myisam_recover_typelib;
+extern ulong myisam_recover_options;
+
+class ha_myisam: public handler
+{
+  MI_INFO *file;
+  ulong   int_table_flags;
+  char    *data_file_name, *index_file_name;
+  bool can_enable_indexes;
+  int repair(THD *thd, MI_CHECK &param, bool optimize);
+
+ public:
+  ha_myisam(TABLE_SHARE *table_arg);
+  ~ha_myisam() {}
+  const char *table_type() const { return "MyISAM"; }
+  const char *index_type(uint key_number);
+  const char **bas_ext() const;
+  ulonglong table_flags() const { return int_table_flags; }
+  ulong index_flags(uint inx, uint part, bool all_parts) const
+  {
+    return ((table_share->key_info[inx].algorithm == HA_KEY_ALG_FULLTEXT) ?
+            0 : HA_READ_NEXT | HA_READ_PREV | HA_READ_RANGE |
+            HA_READ_ORDER | HA_KEYREAD_ONLY);
+  }
+  uint max_supported_keys()          const { return MI_MAX_KEY; }
+  uint max_supported_key_length()    const { return MI_MAX_KEY_LENGTH; }
+  uint max_supported_key_part_length() const { return MI_MAX_KEY_LENGTH; }
+  uint checksum() const;
+
+  virtual bool check_if_locking_is_allowed(uint sql_command,
+                                           ulong type, TABLE *table,
+                                           uint count,
+                                           bool called_by_logger_thread);
+  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_read_last(byte * buf, const byte * key, uint key_len);
+  int index_next(byte * buf);
+  int index_prev(byte * buf);
+  int index_first(byte * buf);
+  int index_last(byte * buf);
+  int index_next_same(byte *buf, const byte *key, uint keylen);
+  int ft_init()
+  {
+    if (!ft_handler)
+      return 1;
+    ft_handler->please->reinit_search(ft_handler);
+    return 0;
+  }
+  FT_INFO *ft_init_ext(uint flags, uint inx,String *key)
+  {
+    return ft_init_search(flags,file,inx,
+                          (byte *)key->ptr(), key->length(), key->charset(),
+                          table->record[0]);
+  }
+  int ft_read(byte *buf);
+  int rnd_init(bool scan);
+  int rnd_next(byte *buf);
+  int rnd_pos(byte * buf, byte *pos);
+  int restart_rnd_next(byte *buf, byte *pos);
+  void position(const byte *record);
+  void info(uint);
+  int extra(enum ha_extra_function operation);
+  int extra_opt(enum ha_extra_function operation, ulong cache_size);
+  int reset(void);
+  int external_lock(THD *thd, int lock_type);
+  int delete_all_rows(void);
+  int disable_indexes(uint mode);
+  int enable_indexes(uint mode);
+  int indexes_are_disabled(void);
+  void start_bulk_insert(ha_rows rows);
+  int end_bulk_insert();
+  ha_rows records_in_range(uint inx, key_range *min_key, key_range *max_key);
+  void update_create_info(HA_CREATE_INFO *create_info);
+  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);
+  virtual void get_auto_increment(ulonglong offset, ulonglong increment,
+                                  ulonglong nb_desired_values,
+                                  ulonglong *first_value,
+                                  ulonglong *nb_reserved_values);
+  int rename_table(const char * from, const char * to);
+  int delete_table(const char *name);
+  int check(THD* thd, HA_CHECK_OPT* check_opt);
+  int analyze(THD* thd,HA_CHECK_OPT* check_opt);
+  int repair(THD* thd, HA_CHECK_OPT* check_opt);
+  bool check_and_repair(THD *thd);
+  bool is_crashed() const;
+  bool auto_repair() const { return myisam_recover_options != 0; }
+  int optimize(THD* thd, HA_CHECK_OPT* check_opt);
+  int restore(THD* thd, HA_CHECK_OPT* check_opt);
+  int backup(THD* thd, HA_CHECK_OPT* check_opt);
+  int assign_to_keycache(THD* thd, HA_CHECK_OPT* check_opt);
+  int preload_keys(THD* thd, HA_CHECK_OPT* check_opt);
+  bool check_if_incompatible_data(HA_CREATE_INFO *info, uint table_changes);
+#ifdef HAVE_REPLICATION
+  int dump(THD* thd, int fd);
+  int net_read_dump(NET* net);
+#endif
+};
diff --git a/storage/myisam/plug.in b/storage/myisam/plug.in
new file mode 100644
index 00000000000..3160752182d
--- /dev/null
+++ b/storage/myisam/plug.in
@@ -0,0 +1,6 @@
+MYSQL_STORAGE_ENGINE(myisam,no, [MyISAM Storage Engine],
+        [Traditional non-transactional MySQL tables])
+MYSQL_PLUGIN_DIRECTORY(myisam,  [storage/myisam])
+MYSQL_PLUGIN_STATIC(myisam,     [libmyisam.a])
+MYSQL_PLUGIN_MANDATORY(myisam)  dnl Default
+
diff --git a/storage/myisammrg/CMakeLists.txt b/storage/myisammrg/CMakeLists.txt
index 83168f6c60c..a86eff9d764 100644
--- a/storage/myisammrg/CMakeLists.txt
+++ b/storage/myisammrg/CMakeLists.txt
@@ -1,8 +1,12 @@
 SET(CMAKE_CXX_FLAGS_DEBUG "${CMAKE_CXX_FLAGS_DEBUG} -DSAFEMALLOC -DSAFE_MUTEX")
 SET(CMAKE_C_FLAGS_DEBUG "${CMAKE_C_FLAGS_DEBUG} -DSAFEMALLOC -DSAFE_MUTEX")
 
-INCLUDE_DIRECTORIES(${CMAKE_SOURCE_DIR}/include)
+INCLUDE_DIRECTORIES(${CMAKE_SOURCE_DIR}/include ${CMAKE_SOURCE_DIR}/zlib
+                    ${CMAKE_SOURCE_DIR}/sql
+                    ${CMAKE_SOURCE_DIR}/regex
+                    ${CMAKE_SOURCE_DIR}/extra/yassl/include)
 ADD_LIBRARY(myisammrg myrg_close.c myrg_create.c myrg_delete.c myrg_extra.c myrg_info.c
+				ha_myisammrg.cc
 				myrg_locking.c myrg_open.c myrg_panic.c myrg_queue.c myrg_range.c
 				myrg_rfirst.c myrg_rkey.c myrg_rlast.c myrg_rnext.c myrg_rnext_same.c
 				myrg_rprev.c myrg_rrnd.c myrg_rsame.c myrg_static.c myrg_update.c
diff --git a/storage/myisammrg/Makefile.am b/storage/myisammrg/Makefile.am
index 0402f2730b9..08cd52c363f 100644
--- a/storage/myisammrg/Makefile.am
+++ b/storage/myisammrg/Makefile.am
@@ -14,15 +14,31 @@
 # along with this program; if not, write to the Free Software
 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
-INCLUDES =		-I$(top_builddir)/include -I$(top_srcdir)/include
+MYSQLDATAdir =          $(localstatedir)
+MYSQLSHAREdir =         $(pkgdatadir)
+MYSQLBASEdir=           $(prefix)
+MYSQLLIBdir=            $(pkglibdir)
+INCLUDES =              -I$(top_srcdir)/include -I$(top_builddir)/include \
+			-I$(top_srcdir)/regex \
+			-I$(top_srcdir)/sql \
+                        -I$(srcdir)
+WRAPLIBS=
+
+LDADD =
+
+DEFS =                  @DEFS@
 pkglib_LIBRARIES =	libmyisammrg.a
-noinst_HEADERS =	myrg_def.h
+noinst_HEADERS =	myrg_def.h ha_myisammrg.h
+noinst_LIBRARIES =	libmyisammrg.a
 libmyisammrg_a_SOURCES = myrg_open.c myrg_extra.c myrg_info.c myrg_locking.c \
 			myrg_rrnd.c myrg_update.c myrg_delete.c myrg_rsame.c \
 			myrg_panic.c myrg_close.c myrg_create.c myrg_static.c \
 			myrg_rkey.c myrg_rfirst.c myrg_rlast.c myrg_rnext.c \
 			myrg_rprev.c myrg_queue.c myrg_write.c myrg_range.c \
+			ha_myisammrg.cc \
 			myrg_rnext_same.c
+
+
 EXTRA_DIST =		CMakeLists.txt
 
 # Don't update the files from bitkeeper
diff --git a/storage/myisammrg/ha_myisammrg.cc b/storage/myisammrg/ha_myisammrg.cc
new file mode 100644
index 00000000000..8c767e32b83
--- /dev/null
+++ b/storage/myisammrg/ha_myisammrg.cc
@@ -0,0 +1,578 @@
+/* Copyright (C) 2000 MySQL AB & MySQL Finland AB & TCX DataKonsult 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 */
+
+
+#ifdef USE_PRAGMA_IMPLEMENTATION
+#pragma implementation				// gcc: Class implementation
+#endif
+
+#define MYSQL_SERVER 1
+#include "mysql_priv.h"
+#include <mysql/plugin.h>
+#include <m_ctype.h>
+#include "ha_myisammrg.h"
+#include "myrg_def.h"
+
+
+/*****************************************************************************
+** MyISAM MERGE tables
+*****************************************************************************/
+
+static handler *myisammrg_create_handler(TABLE_SHARE *table,
+                                         MEM_ROOT *mem_root);
+
+/* MyISAM MERGE handlerton */
+
+handlerton myisammrg_hton;
+
+static handler *myisammrg_create_handler(TABLE_SHARE *table,
+                                         MEM_ROOT *mem_root)
+{
+  return new (mem_root) ha_myisammrg(table);
+}
+
+
+ha_myisammrg::ha_myisammrg(TABLE_SHARE *table_arg)
+  :handler(&myisammrg_hton, table_arg), file(0)
+{}
+
+static const char *ha_myisammrg_exts[] = {
+  ".MRG",
+  NullS
+};
+
+const char **ha_myisammrg::bas_ext() const
+{
+  return ha_myisammrg_exts;
+}
+
+
+const char *ha_myisammrg::index_type(uint key_number)
+{
+  return ((table->key_info[key_number].flags & HA_FULLTEXT) ? 
+	  "FULLTEXT" :
+	  (table->key_info[key_number].flags & HA_SPATIAL) ?
+	  "SPATIAL" :
+	  (table->key_info[key_number].algorithm == HA_KEY_ALG_RTREE) ?
+	  "RTREE" :
+	  "BTREE");
+}
+
+
+int ha_myisammrg::open(const char *name, int mode, uint test_if_locked)
+{
+  char name_buff[FN_REFLEN];
+
+  DBUG_PRINT("info", ("ha_myisammrg::open"));
+  if (!(file=myrg_open(fn_format(name_buff,name,"","",
+                                 MY_UNPACK_FILENAME|MY_APPEND_EXT),
+                       mode, test_if_locked)))
+  {
+    DBUG_PRINT("info", ("ha_myisammrg::open exit %d", my_errno));
+    return (my_errno ? my_errno : -1);
+  }
+  DBUG_PRINT("info", ("ha_myisammrg::open myrg_extrafunc..."));
+  myrg_extrafunc(file, query_cache_invalidate_by_MyISAM_filename_ref);
+  if (!(test_if_locked == HA_OPEN_WAIT_IF_LOCKED ||
+	test_if_locked == HA_OPEN_ABORT_IF_LOCKED))
+    myrg_extra(file,HA_EXTRA_NO_WAIT_LOCK,0);
+  info(HA_STATUS_NO_LOCK | HA_STATUS_VARIABLE | HA_STATUS_CONST);
+  if (!(test_if_locked & HA_OPEN_WAIT_IF_LOCKED))
+    myrg_extra(file,HA_EXTRA_WAIT_LOCK,0);
+
+  if (table->s->reclength != stats.mean_rec_length && stats.mean_rec_length)
+  {
+    DBUG_PRINT("error",("reclength: %d  mean_rec_length: %d",
+			table->s->reclength, stats.mean_rec_length));
+    goto err;
+  }
+#if !defined(BIG_TABLES) || SIZEOF_OFF_T == 4
+  /* Merge table has more than 2G rows */
+  if (table->s->crashed)
+    goto err;
+#endif
+  return (0);
+err:
+  myrg_close(file);
+  file=0;
+  return (my_errno= HA_ERR_WRONG_MRG_TABLE_DEF);
+}
+
+int ha_myisammrg::close(void)
+{
+  return myrg_close(file);
+}
+
+int ha_myisammrg::write_row(byte * buf)
+{
+  statistic_increment(table->in_use->status_var.ha_write_count,&LOCK_status);
+
+  if (file->merge_insert_method == MERGE_INSERT_DISABLED || !file->tables)
+    return (HA_ERR_TABLE_READONLY);
+
+  if (table->timestamp_field_type & TIMESTAMP_AUTO_SET_ON_INSERT)
+    table->timestamp_field->set_time();
+  if (table->next_number_field && buf == table->record[0])
+      update_auto_increment();
+  return myrg_write(file,buf);
+}
+
+int ha_myisammrg::update_row(const byte * old_data, byte * new_data)
+{
+  statistic_increment(table->in_use->status_var.ha_update_count,&LOCK_status);
+  if (table->timestamp_field_type & TIMESTAMP_AUTO_SET_ON_UPDATE)
+    table->timestamp_field->set_time();
+  return myrg_update(file,old_data,new_data);
+}
+
+int ha_myisammrg::delete_row(const byte * buf)
+{
+  statistic_increment(table->in_use->status_var.ha_delete_count,&LOCK_status);
+  return myrg_delete(file,buf);
+}
+
+int ha_myisammrg::index_read(byte * buf, const byte * key,
+			  uint key_len, enum ha_rkey_function find_flag)
+{
+  statistic_increment(table->in_use->status_var.ha_read_key_count,
+		      &LOCK_status);
+  int error=myrg_rkey(file,buf,active_index, key, key_len, find_flag);
+  table->status=error ? STATUS_NOT_FOUND: 0;
+  return error;
+}
+
+int ha_myisammrg::index_read_idx(byte * buf, uint index, const byte * key,
+				 uint key_len, enum ha_rkey_function find_flag)
+{
+  statistic_increment(table->in_use->status_var.ha_read_key_count,
+		      &LOCK_status);
+  int error=myrg_rkey(file,buf,index, key, key_len, find_flag);
+  table->status=error ? STATUS_NOT_FOUND: 0;
+  return error;
+}
+
+int ha_myisammrg::index_read_last(byte * buf, const byte * key, uint key_len)
+{
+  statistic_increment(table->in_use->status_var.ha_read_key_count,
+		      &LOCK_status);
+  int error=myrg_rkey(file,buf,active_index, key, key_len,
+		      HA_READ_PREFIX_LAST);
+  table->status=error ? STATUS_NOT_FOUND: 0;
+  return error;
+}
+
+int ha_myisammrg::index_next(byte * buf)
+{
+  statistic_increment(table->in_use->status_var.ha_read_next_count,
+		      &LOCK_status);
+  int error=myrg_rnext(file,buf,active_index);
+  table->status=error ? STATUS_NOT_FOUND: 0;
+  return error;
+}
+
+int ha_myisammrg::index_prev(byte * buf)
+{
+  statistic_increment(table->in_use->status_var.ha_read_prev_count,
+		      &LOCK_status);
+  int error=myrg_rprev(file,buf, active_index);
+  table->status=error ? STATUS_NOT_FOUND: 0;
+  return error;
+}
+
+int ha_myisammrg::index_first(byte * buf)
+{
+  statistic_increment(table->in_use->status_var.ha_read_first_count,
+		      &LOCK_status);
+  int error=myrg_rfirst(file, buf, active_index);
+  table->status=error ? STATUS_NOT_FOUND: 0;
+  return error;
+}
+
+int ha_myisammrg::index_last(byte * buf)
+{
+  statistic_increment(table->in_use->status_var.ha_read_last_count,
+		      &LOCK_status);
+  int error=myrg_rlast(file, buf, active_index);
+  table->status=error ? STATUS_NOT_FOUND: 0;
+  return error;
+}
+
+int ha_myisammrg::index_next_same(byte * buf,
+                                  const byte *key __attribute__((unused)),
+                                  uint length __attribute__((unused)))
+{
+  statistic_increment(table->in_use->status_var.ha_read_next_count,
+		      &LOCK_status);
+  int error=myrg_rnext_same(file,buf);
+  table->status=error ? STATUS_NOT_FOUND: 0;
+  return error;
+}
+
+
+int ha_myisammrg::rnd_init(bool scan)
+{
+  return myrg_reset(file);
+}
+
+
+int ha_myisammrg::rnd_next(byte *buf)
+{
+  statistic_increment(table->in_use->status_var.ha_read_rnd_next_count,
+		      &LOCK_status);
+  int error=myrg_rrnd(file, buf, HA_OFFSET_ERROR);
+  table->status=error ? STATUS_NOT_FOUND: 0;
+  return error;
+}
+
+
+int ha_myisammrg::rnd_pos(byte * buf, byte *pos)
+{
+  statistic_increment(table->in_use->status_var.ha_read_rnd_count,
+		      &LOCK_status);
+  int error=myrg_rrnd(file, buf, my_get_ptr(pos,ref_length));
+  table->status=error ? STATUS_NOT_FOUND: 0;
+  return error;
+}
+
+void ha_myisammrg::position(const byte *record)
+{
+  ulonglong position= myrg_position(file);
+  my_store_ptr(ref, ref_length, (my_off_t) position);
+}
+
+
+ha_rows ha_myisammrg::records_in_range(uint inx, key_range *min_key,
+                                       key_range *max_key)
+{
+  return (ha_rows) myrg_records_in_range(file, (int) inx, min_key, max_key);
+}
+
+
+void ha_myisammrg::info(uint flag)
+{
+  MYMERGE_INFO info;
+  (void) myrg_status(file,&info,flag);
+  /*
+    The following fails if one has not compiled MySQL with -DBIG_TABLES
+    and one has more than 2^32 rows in the merge tables.
+  */
+  stats.records = (ha_rows) info.records;
+  stats.deleted = (ha_rows) info.deleted;
+#if !defined(BIG_TABLES) || SIZEOF_OFF_T == 4
+  if ((info.records >= (ulonglong) 1 << 32) ||
+      (info.deleted >= (ulonglong) 1 << 32))
+    table->s->crashed= 1;
+#endif
+  stats.data_file_length=info.data_file_length;
+  errkey  = info.errkey;
+  table->s->keys_in_use.set_prefix(table->s->keys);
+  table->s->db_options_in_use= info.options;
+  stats.mean_rec_length= info.reclength;
+  
+  /* 
+    The handler::block_size is used all over the code in index scan cost
+    calculations. It is used to get number of disk seeks required to
+    retrieve a number of index tuples.
+    If the merge table has N underlying tables, then (assuming underlying
+    tables have equal size, the only "simple" approach we can use)
+    retrieving X index records from a merge table will require N times more
+    disk seeks compared to doing the same on a MyISAM table with equal
+    number of records.
+    In the edge case (file_tables > myisam_block_size) we'll get
+    block_size==0, and index calculation code will act as if we need one
+    disk seek to retrieve one index tuple.
+
+    TODO: In 5.2 index scan cost calculation will be factored out into a
+    virtual function in class handler and we'll be able to remove this hack.
+  */
+  stats.block_size= 0;
+  if (file->tables)
+    stats.block_size= myisam_block_size / file->tables;
+  
+  stats.update_time= 0;
+#if SIZEOF_OFF_T > 4
+  ref_length=6;					// Should be big enough
+#else
+  ref_length=4;					// Can't be > than my_off_t
+#endif
+  if (flag & HA_STATUS_CONST)
+  {
+    if (table->s->key_parts && info.rec_per_key)
+      memcpy((char*) table->key_info[0].rec_per_key,
+	     (char*) info.rec_per_key,
+	     sizeof(table->key_info[0].rec_per_key)*table->s->key_parts);
+  }
+}
+
+
+int ha_myisammrg::extra(enum ha_extra_function operation)
+{
+  /* As this is just a mapping, we don't have to force the underlying
+     tables to be closed */
+  if (operation == HA_EXTRA_FORCE_REOPEN ||
+      operation == HA_EXTRA_PREPARE_FOR_DELETE)
+    return 0;
+  return myrg_extra(file,operation,0);
+}
+
+int ha_myisammrg::reset(void)
+{
+  return myrg_reset(file);
+}
+
+/* To be used with WRITE_CACHE, EXTRA_CACHE and BULK_INSERT_BEGIN */
+
+int ha_myisammrg::extra_opt(enum ha_extra_function operation, ulong cache_size)
+{
+  if ((specialflag & SPECIAL_SAFE_MODE) && operation == HA_EXTRA_WRITE_CACHE)
+    return 0;
+  return myrg_extra(file, operation, (void*) &cache_size);
+}
+
+int ha_myisammrg::external_lock(THD *thd, int lock_type)
+{
+  return myrg_lock_database(file,lock_type);
+}
+
+uint ha_myisammrg::lock_count(void) const
+{
+  return file->tables;
+}
+
+
+THR_LOCK_DATA **ha_myisammrg::store_lock(THD *thd,
+					 THR_LOCK_DATA **to,
+					 enum thr_lock_type lock_type)
+{
+  MYRG_TABLE *open_table;
+
+  for (open_table=file->open_tables ;
+       open_table != file->end_table ;
+       open_table++)
+  {
+    *(to++)= &open_table->table->lock;
+    if (lock_type != TL_IGNORE && open_table->table->lock.type == TL_UNLOCK)
+      open_table->table->lock.type=lock_type;
+  }
+  return to;
+}
+
+
+/* Find out database name and table name from a filename */
+
+static void split_file_name(const char *file_name,
+			    LEX_STRING *db, LEX_STRING *name)
+{
+  uint dir_length, prefix_length;
+  char buff[FN_REFLEN];
+
+  db->length= 0;
+  strmake(buff, file_name, sizeof(buff)-1);
+  dir_length= dirname_length(buff);
+  if (dir_length > 1)
+  {
+    /* Get database */
+    buff[dir_length-1]= 0;			// Remove end '/'
+    prefix_length= dirname_length(buff);
+    db->str= (char*) file_name+ prefix_length;
+    db->length= dir_length - prefix_length -1;
+  }
+  name->str= (char*) file_name+ dir_length;
+  name->length= (uint) (fn_ext(name->str) - name->str);
+}
+
+
+void ha_myisammrg::update_create_info(HA_CREATE_INFO *create_info)
+{
+  DBUG_ENTER("ha_myisammrg::update_create_info");
+
+  if (!(create_info->used_fields & HA_CREATE_USED_UNION))
+  {
+    MYRG_TABLE *open_table;
+    THD *thd=current_thd;
+
+    create_info->merge_list.next= &create_info->merge_list.first;
+    create_info->merge_list.elements=0;
+
+    for (open_table=file->open_tables ;
+	 open_table != file->end_table ;
+	 open_table++)
+    {
+      TABLE_LIST *ptr;
+      LEX_STRING db, name;
+
+      if (!(ptr = (TABLE_LIST *) thd->calloc(sizeof(TABLE_LIST))))
+	goto err;
+      split_file_name(open_table->table->filename, &db, &name);
+      if (!(ptr->table_name= thd->strmake(name.str, name.length)))
+	goto err;
+      if (db.length && !(ptr->db= thd->strmake(db.str, db.length)))
+	goto err;
+
+      create_info->merge_list.elements++;
+      (*create_info->merge_list.next) = (byte*) ptr;
+      create_info->merge_list.next= (byte**) &ptr->next_local;
+    }
+    *create_info->merge_list.next=0;
+  }
+  if (!(create_info->used_fields & HA_CREATE_USED_INSERT_METHOD))
+  {
+    create_info->merge_insert_method = file->merge_insert_method;
+  }
+  DBUG_VOID_RETURN;
+
+err:
+  create_info->merge_list.elements=0;
+  create_info->merge_list.first=0;
+  DBUG_VOID_RETURN;
+}
+
+
+int ha_myisammrg::create(const char *name, register TABLE *form,
+			 HA_CREATE_INFO *create_info)
+{
+  char buff[FN_REFLEN];
+  const char **table_names, **pos;
+  TABLE_LIST *tables= (TABLE_LIST*) create_info->merge_list.first;
+  THD *thd= current_thd;
+  uint dirlgt= dirname_length(name);
+  DBUG_ENTER("ha_myisammrg::create");
+
+  if (!(table_names= (const char**)
+        thd->alloc((create_info->merge_list.elements+1) * sizeof(char*))))
+    DBUG_RETURN(HA_ERR_OUT_OF_MEM);
+  for (pos= table_names; tables; tables= tables->next_local)
+  {
+    const char *table_name;
+    TABLE *tbl= 0;
+    if (create_info->options & HA_LEX_CREATE_TMP_TABLE)
+      tbl= find_temporary_table(thd, tables);
+    if (!tbl)
+    {
+      /*
+        Construct the path to the MyISAM table. Try to meet two conditions:
+        1.) Allow to include MyISAM tables from different databases, and
+        2.) allow for moving DATADIR around in the file system.
+        The first means that we need paths in the .MRG file. The second
+        means that we should not have absolute paths in the .MRG file.
+        The best, we can do, is to use 'mysql_data_home', which is '.'
+        in mysqld and may be an absolute path in an embedded server.
+        This means that it might not be possible to move the DATADIR of
+        an embedded server without changing the paths in the .MRG file.
+      */
+      uint length= build_table_filename(buff, sizeof(buff),
+                                        tables->db, tables->table_name, "", 0);
+      /*
+        If a MyISAM table is in the same directory as the MERGE table,
+        we use the table name without a path. This means that the
+        DATADIR can easily be moved even for an embedded server as long
+        as the MyISAM tables are from the same database as the MERGE table.
+      */
+      if ((dirname_length(buff) == dirlgt) && ! memcmp(buff, name, dirlgt))
+        table_name= tables->table_name;
+      else
+        if (! (table_name= thd->strmake(buff, length)))
+          DBUG_RETURN(HA_ERR_OUT_OF_MEM);
+    }
+    else
+      table_name= tbl->s->path.str;
+    *pos++= table_name;
+  }
+  *pos=0;
+  DBUG_RETURN(myrg_create(fn_format(buff,name,"","",
+                                    MY_RESOLVE_SYMLINKS|
+                                    MY_UNPACK_FILENAME|MY_APPEND_EXT),
+			  table_names,
+                          create_info->merge_insert_method,
+                          (my_bool) 0));
+}
+
+
+void ha_myisammrg::append_create_info(String *packet)
+{
+  const char *current_db;
+  uint db_length;
+  THD *thd= current_thd;
+  MYRG_TABLE *open_table, *first;
+
+  if (file->merge_insert_method != MERGE_INSERT_DISABLED)
+  {
+    packet->append(STRING_WITH_LEN(" INSERT_METHOD="));
+    packet->append(get_type(&merge_insert_method,file->merge_insert_method-1));
+  }
+  packet->append(STRING_WITH_LEN(" UNION=("));
+
+  current_db= table->s->db.str;
+  db_length=  table->s->db.length;
+
+  for (first=open_table=file->open_tables ;
+       open_table != file->end_table ;
+       open_table++)
+  {
+    LEX_STRING db, name;
+    split_file_name(open_table->table->filename, &db, &name);
+    if (open_table != first)
+      packet->append(',');
+    /* Report database for mapped table if it isn't in current database */
+    if (db.length &&
+	(db_length != db.length ||
+	 strncmp(current_db, db.str, db.length)))
+    {
+      append_identifier(thd, packet, db.str, db.length);
+      packet->append('.');
+    }
+    append_identifier(thd, packet, name.str, name.length);
+  }
+  packet->append(')');
+}
+
+
+bool ha_myisammrg::check_if_incompatible_data(HA_CREATE_INFO *info,
+					      uint table_changes)
+{
+  /*
+    For myisammrg, we should always re-generate the mapping file as this
+    is trivial to do
+  */
+  return COMPATIBLE_DATA_NO;
+}
+
+static int myisammrg_init()
+{
+  myisammrg_hton.state=have_merge_db;
+  myisammrg_hton.db_type=DB_TYPE_MRG_MYISAM;
+  myisammrg_hton.create=myisammrg_create_handler;
+  myisammrg_hton.panic=myrg_panic;
+  myisammrg_hton.flags= HTON_CAN_RECREATE;
+  return 0;
+}
+
+struct st_mysql_storage_engine myisammrg_storage_engine=
+{ MYSQL_HANDLERTON_INTERFACE_VERSION, &myisammrg_hton };
+
+mysql_declare_plugin(myisammrg)
+{
+  MYSQL_STORAGE_ENGINE_PLUGIN,
+  &myisammrg_storage_engine,
+  "MRG_MYISAM",
+  "MySQL AB",
+  "Collection of identical MyISAM tables",
+  myisammrg_init, /* Plugin Init */
+  NULL, /* Plugin Deinit */
+  0x0100, /* 1.0 */
+  0
+}
+mysql_declare_plugin_end;
diff --git a/storage/myisammrg/ha_myisammrg.h b/storage/myisammrg/ha_myisammrg.h
new file mode 100644
index 00000000000..d58a3523c26
--- /dev/null
+++ b/storage/myisammrg/ha_myisammrg.h
@@ -0,0 +1,88 @@
+/* Copyright (C) 2000 MySQL AB & MySQL Finland AB & TCX DataKonsult 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 */
+
+
+#ifdef USE_PRAGMA_INTERFACE
+#pragma interface			/* gcc class implementation */
+#endif
+
+/* class for the the myisam merge handler */
+
+#include <myisammrg.h>
+
+class ha_myisammrg: public handler
+{
+  MYRG_INFO *file;
+
+ public:
+  ha_myisammrg(TABLE_SHARE *table_arg);
+  ~ha_myisammrg() {}
+  const char *table_type() const { return "MRG_MyISAM"; }
+  const char **bas_ext() const;
+  const char *index_type(uint key_number);
+  ulonglong table_flags() const
+  {
+    return (HA_REC_NOT_IN_SEQ | HA_AUTO_PART_KEY | HA_NO_TRANSACTIONS |
+	    HA_NULL_IN_KEY | HA_CAN_INDEX_BLOBS | HA_FILE_BASED |
+            HA_CAN_INSERT_DELAYED | HA_ANY_INDEX_MAY_BE_UNIQUE |
+            HA_CAN_BIT_FIELD | HA_NO_COPY_ON_ALTER);
+  }
+  ulong index_flags(uint inx, uint part, bool all_parts) const
+  {
+    return ((table_share->key_info[inx].algorithm == HA_KEY_ALG_FULLTEXT) ?
+            0 : HA_READ_NEXT | HA_READ_PREV | HA_READ_RANGE |
+            HA_READ_ORDER | HA_KEYREAD_ONLY);
+  }
+  uint max_supported_keys()          const { return MI_MAX_KEY; }
+  uint max_supported_key_length()    const { return MI_MAX_KEY_LENGTH; }
+  uint max_supported_key_part_length() const { return MI_MAX_KEY_LENGTH; }
+  double scan_time()
+  { return ulonglong2double(stats.data_file_length) / IO_SIZE + file->tables; }
+
+  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_read_last(byte * buf, const byte * key, uint key_len);
+  int index_next(byte * buf);
+  int index_prev(byte * buf);
+  int index_first(byte * buf);
+  int index_last(byte * buf);
+  int index_next_same(byte *buf, const byte *key, uint keylen);
+  int rnd_init(bool scan);
+  int rnd_next(byte *buf);
+  int rnd_pos(byte * buf, byte *pos);
+  void position(const byte *record);
+  ha_rows records_in_range(uint inx, key_range *min_key, key_range *max_key);
+  void info(uint);
+  int reset(void);
+  int extra(enum ha_extra_function operation);
+  int extra_opt(enum ha_extra_function operation, ulong cache_size);
+  int external_lock(THD *thd, int lock_type);
+  uint lock_count(void) const;
+  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);
+  void update_create_info(HA_CREATE_INFO *create_info);
+  void append_create_info(String *packet);
+  MYRG_INFO *myrg_info() { return file; }
+  bool check_if_incompatible_data(HA_CREATE_INFO *info, uint table_changes);
+};
diff --git a/storage/myisammrg/plug.in b/storage/myisammrg/plug.in
new file mode 100644
index 00000000000..b4b2af8d984
--- /dev/null
+++ b/storage/myisammrg/plug.in
@@ -0,0 +1,5 @@
+MYSQL_STORAGE_ENGINE(myisammrg,no,[MyISAM MERGE Engine],
+        [Merge multiple MySQL tables into one])
+MYSQL_PLUGIN_DIRECTORY(myisammrg,[storage/myisammrg])
+MYSQL_PLUGIN_STATIC(myisammrg,  [libmyisammrg.a])
+MYSQL_PLUGIN_MANDATORY(myisammrg)
diff --git a/storage/ndb/plug.in b/storage/ndb/plug.in
new file mode 100644
index 00000000000..a7e351417b1
--- /dev/null
+++ b/storage/ndb/plug.in
@@ -0,0 +1,6 @@
+MYSQL_STORAGE_ENGINE(ndbcluster, ndbcluster, [Cluster Storage Engine],
+        [High Availability Clustered tables], [max])
+MYSQL_PLUGIN_DIRECTORY(ndbcluster,[storage/ndb])
+MYSQL_PLUGIN_STATIC(ndbcluster, [[\$(ndbcluster_libs) \$(ndbcluster_system_libs) \$(NDB_SCI_LIBS)]])
+MYSQL_PLUGIN_ACTIONS(ndbcluster,[MYSQL_SETUP_NDBCLUSTER])
+MYSQL_PLUGIN_DEPENDS(ndbcluster, [partition])