merge

BUILD/autorun.sh:
  Auto merged
BitKeeper/deleted/.del-var:
  Delete: mysql-test/var
BitKeeper/etc/config:
  Auto merged
configure.in:
  Auto merged
mysql-test/r/information_schema_db.result:
  Auto merged
mysys/Makefile.am:
  Auto merged
mysys/default.c:
  Auto merged
sql/ha_berkeley.cc:
  Auto merged
sql/ha_berkeley.h:
  Auto merged
sql/ha_federated.h:
  Auto merged
sql/ha_innodb.cc:
  Auto merged
sql/ha_innodb.h:
  Auto merged
sql/ha_myisam.cc:
  Auto merged
sql/ha_myisammrg.cc:
  Auto merged
sql/ha_ndbcluster.cc:
  Auto merged
sql/handler.h:
  Auto merged
sql/item.cc:
  Auto merged
sql/lex.h:
  Auto merged
sql/lock.cc:
  Auto merged
sql/log.cc:
  Auto merged
sql/log_event.cc:
  Auto merged
sql/mysql_priv.h:
  Auto merged
sql/mysqld.cc:
  Auto merged
sql/opt_range.cc:
  Auto merged
sql/set_var.cc:
  Auto merged
sql/sp.cc:
  Auto merged
sql/sql_base.cc:
  Auto merged
sql/sql_bitmap.h:
  Auto merged
sql/sql_cache.cc:
  Auto merged
sql/sql_class.h:
  Auto merged
sql/sql_lex.h:
  Auto merged
sql/sql_parse.cc:
  Auto merged
sql/sql_prepare.cc:
  Auto merged
sql/sql_repl.cc:
  Auto merged
sql/sql_select.cc:
  Auto merged
sql/sql_select.h:
  Auto merged
sql/sql_show.cc:
  Auto merged
sql/examples/ha_tina.cc:
  Auto merged
sql/sql_table.cc:
  Auto merged
sql/sql_yacc.yy:
  Auto merged
sql/table.h:
  Auto merged
storage/myisam/mi_check.c:
  Auto merged
storage/myisam/mi_create.c:
  Auto merged
storage/myisam/mi_delete.c:
  Auto merged
storage/myisam/mi_extra.c:
  Auto merged
storage/myisam/mi_open.c:
  Auto merged
storage/myisam/mi_preload.c:
  Auto merged
storage/myisam/mi_rsame.c:
  Auto merged
storage/myisam/mi_rsamepos.c:
  Auto merged
storage/myisam/mi_search.c:
  Auto merged
storage/myisam/mi_update.c:
  Auto merged
storage/myisam/mi_write.c:
  Auto merged
storage/myisam/myisamchk.c:
  Auto merged
storage/myisam/myisamdef.h:
  Auto merged
storage/myisam/myisamlog.c:
  Auto merged
storage/myisam/myisampack.c:
  Auto merged
storage/myisam/sort.c:
  Auto merged
storage/ndb/src/common/portlib/NdbMutex.c:
  Auto merged
storage/ndb/src/common/portlib/NdbThread.c:
  Auto merged
support-files/mysql.spec.sh:
  Auto merged

61 files changed, 2465 insertions, 796 deletions

diff --git a/sql/examples/ha_archive.cc b/sql/examples/ha_archive.cc
index c362985f565..e5c35ae019d 100644
--- a/sql/examples/ha_archive.cc
+++ b/sql/examples/ha_archive.cc
@@ -149,7 +149,8 @@ static handlerton archive_hton = {
   0,       /* prepare */
   0,       /* recover */
   0,       /* commit_by_xid */
-  0        /* rollback_by_xid */
+  0,       /* rollback_by_xid */
+  HTON_NO_FLAGS
 };
 
 
@@ -208,6 +209,15 @@ bool archive_db_end()
   return FALSE;
 }
 
+ha_archive::ha_archive(TABLE *table_arg)
+  :handler(&archive_hton, table_arg), delayed_insert(0), bulk_insert(0)
+{
+  /* Set our original buffer from pre-allocated memory */
+  buffer.set((char *)byte_buffer, IO_SIZE, system_charset_info);
+
+  /* The size of the offset value we will use for position() */
+  ref_length = sizeof(z_off_t);
+}
 
 /*
   This method reads the header of a datafile and returns whether or not it was successful.
diff --git a/sql/examples/ha_archive.h b/sql/examples/ha_archive.h
index 3932b62980c..41835c5fb6f 100644
--- a/sql/examples/ha_archive.h
+++ b/sql/examples/ha_archive.h
@@ -58,14 +58,7 @@ class ha_archive: public handler
   bool bulk_insert;          /* If we are performing a bulk insert */
 
 public:
-  ha_archive(TABLE *table): handler(table), delayed_insert(0), bulk_insert(0)
-  {
-    /* Set our original buffer from pre-allocated memory */
-    buffer.set((char *)byte_buffer, IO_SIZE, system_charset_info);
-
-    /* The size of the offset value we will use for position() */
-    ref_length = sizeof(z_off_t);
-  }
+  ha_archive(TABLE *table_arg);
   ~ha_archive()
   {
   }
diff --git a/sql/examples/ha_example.cc b/sql/examples/ha_example.cc
index 9da297ccd1f..2818c176cd3 100644
--- a/sql/examples/ha_example.cc
+++ b/sql/examples/ha_example.cc
@@ -72,6 +72,24 @@
 #ifdef HAVE_EXAMPLE_DB
 #include "ha_example.h"
 
+
+static handlerton example_hton= {
+  "CSV",
+  0,       /* slot */
+  0,       /* savepoint size. */
+  0,       /* close_connection */
+  0,       /* savepoint */
+  0,       /* rollback to savepoint */
+  0,       /* release savepoint */
+  0,       /* commit */
+  0,       /* rollback */
+  0,       /* prepare */
+  0,       /* recover */
+  0,       /* commit_by_xid */
+  0,       /* rollback_by_xid */
+  HTON_NO_FLAGS
+};
+
 /* Variables for example share methods */
 static HASH example_open_tables; // Hash used to track open tables
 pthread_mutex_t example_mutex;   // This is the mutex we use to init the hash
@@ -179,6 +197,10 @@ static int free_share(EXAMPLE_SHARE *share)
 }
 
 
+ha_example::ha_example(TABLE *table_arg)
+  :handler(&example_hton, table_arg)
+{}
+
 /*
   If frm_error() is called then we will use this to to find out what file extentions
   exist for the storage engine. This is also used by the default rename_table and
diff --git a/sql/examples/ha_example.h b/sql/examples/ha_example.h
index ae72e5bb275..37f38fe5210 100644
--- a/sql/examples/ha_example.h
+++ b/sql/examples/ha_example.h
@@ -45,9 +45,7 @@ class ha_example: public handler
   EXAMPLE_SHARE *share;    /* Shared lock info */
 
 public:
-  ha_example(TABLE *table): handler(table)
-  {
-  }
+  ha_example(TABLE *table_arg);
   ~ha_example()
   {
   }
diff --git a/sql/examples/ha_tina.cc b/sql/examples/ha_tina.cc
index 8680d683a2f..50fec4e2883 100644
--- a/sql/examples/ha_tina.cc
+++ b/sql/examples/ha_tina.cc
@@ -54,6 +54,23 @@ pthread_mutex_t tina_mutex;
 static HASH tina_open_tables;
 static int tina_init= 0;
 
+static handlerton tina_hton= {
+  "CSV",
+  0,       /* slot */
+  0,       /* savepoint size. */
+  0,       /* close_connection */
+  0,       /* savepoint */
+  0,       /* rollback to savepoint */
+  0,       /* release savepoint */
+  0,       /* commit */
+  0,       /* rollback */
+  0,       /* prepare */
+  0,       /* recover */
+  0,       /* commit_by_xid */
+  0,       /* rollback_by_xid */
+  HTON_NO_FLAGS
+};
+
 /*****************************************************************************
  ** TINA tables
  *****************************************************************************/
@@ -229,6 +246,20 @@ byte * find_eoln(byte *data, off_t begin, off_t end)
   return 0;
 }
 
+
+ha_tina::ha_tina(TABLE *table_arg)
+  :handler(&tina_hton, table_arg),
+  /*
+    These definitions are found in hanler.h
+    These are not probably completely right.
+  */
+  current_position(0), next_position(0), chain_alloced(0), chain_size(DEFAULT_CHAIN_LENGTH)
+{
+  /* Set our original buffers from pre-allocated memory */
+  buffer.set(byte_buffer, IO_SIZE, system_charset_info);
+  chain= chain_buffer;
+}
+
 /*
   Encode a buffer into the quoted format.
 */
diff --git a/sql/examples/ha_tina.h b/sql/examples/ha_tina.h
index 22193c01013..5679d77a4dc 100644
--- a/sql/examples/ha_tina.h
+++ b/sql/examples/ha_tina.h
@@ -49,18 +49,8 @@ class ha_tina: public handler
   byte chain_alloced;
   uint32 chain_size;
 
-  public:
-  ha_tina(TABLE *table): handler(table),
-  /* 
-     These definitions are found in hanler.h 
-     Theses are not probably completely right.
-   */
-  current_position(0), next_position(0), chain_alloced(0), chain_size(DEFAULT_CHAIN_LENGTH)
-  {
-    /* Set our original buffers from pre-allocated memory */
-    buffer.set(byte_buffer, IO_SIZE, system_charset_info);
-    chain = chain_buffer;
-  }
+public:
+  ha_tina(TABLE *table_arg);
   ~ha_tina() 
   {
     if (chain_alloced)
diff --git a/sql/ha_berkeley.cc b/sql/ha_berkeley.cc
index 6b283079072..3b9cdbe29f7 100644
--- a/sql/ha_berkeley.cc
+++ b/sql/ha_berkeley.cc
@@ -120,7 +120,8 @@ static handlerton berkeley_hton = {
   NULL, /* prepare */
   NULL, /* recover */
   NULL, /* commit_by_xid */
-  NULL /* rollback_by_xid */
+  NULL, /* rollback_by_xid */
+  HTON_CLOSE_CURSORS_AT_COMMIT
 };
 
 typedef struct st_berkeley_trx_data {
@@ -372,6 +373,17 @@ void berkeley_cleanup_log_files(void)
 /*****************************************************************************
 ** Berkeley DB tables
 *****************************************************************************/
+
+ha_berkeley::ha_berkeley(TABLE *table_arg)
+  :handler(&berkeley_hton, table_arg), alloc_ptr(0), rec_buff(0), file(0),
+  int_table_flags(HA_REC_NOT_IN_SEQ | HA_FAST_KEY_READ |
+                  HA_NULL_IN_KEY | HA_CAN_INDEX_BLOBS | HA_NOT_EXACT_COUNT |
+                  HA_PRIMARY_KEY_IN_READ_INDEX | HA_FILE_BASED |
+                  HA_AUTO_PART_KEY | HA_TABLE_SCAN_ON_INDEX),
+  changed_rows(0), last_dup_key((uint) -1), version(0), using_ignore(0)
+{}
+
+
 static const char *ha_berkeley_exts[] = {
   ha_berkeley_ext,
   NullS
diff --git a/sql/ha_berkeley.h b/sql/ha_berkeley.h
index 596a59b4d43..9c0668b967c 100644
--- a/sql/ha_berkeley.h
+++ b/sql/ha_berkeley.h
@@ -85,12 +85,7 @@ class ha_berkeley: public handler
   DBT *get_pos(DBT *to, byte *pos);
 
  public:
-  ha_berkeley(TABLE *table): handler(table), alloc_ptr(0),rec_buff(0), file(0),
-    int_table_flags(HA_REC_NOT_IN_SEQ | HA_FAST_KEY_READ |
-		    HA_NULL_IN_KEY | HA_CAN_INDEX_BLOBS | HA_NOT_EXACT_COUNT |
-		    HA_PRIMARY_KEY_IN_READ_INDEX | HA_FILE_BASED |
-		    HA_AUTO_PART_KEY | HA_TABLE_SCAN_ON_INDEX),
-    changed_rows(0),last_dup_key((uint) -1),version(0),using_ignore(0) {}
+  ha_berkeley(TABLE *table_arg);
   ~ha_berkeley() {}
   const char *table_type() const { return "BerkeleyDB"; }
   ulong index_flags(uint idx, uint part, bool all_parts) const;
diff --git a/sql/ha_blackhole.cc b/sql/ha_blackhole.cc
index 6abbe983f48..856a053db9e 100644
--- a/sql/ha_blackhole.cc
+++ b/sql/ha_blackhole.cc
@@ -24,6 +24,34 @@
 #include "ha_blackhole.h"
 
 
+/* Blackhole storage engine handlerton */
+
+static handlerton myisam_hton= {
+  "BLACKHOLE",
+  0,       /* slot */
+  0,       /* savepoint size. */
+  0,       /* close_connection */
+  0,       /* savepoint */
+  0,       /* rollback to savepoint */
+  0,       /* release savepoint */
+  0,       /* commit */
+  0,       /* rollback */
+  0,       /* prepare */
+  0,       /* recover */
+  0,       /* commit_by_xid */
+  0,       /* rollback_by_xid */
+  HTON_NO_FLAGS
+};
+
+/*****************************************************************************
+** BLACKHOLE tables
+*****************************************************************************/
+
+ha_blackhole::ha_blackhole(TABLE *table_arg)
+  :handler(&blackhole_hton, table_arg)
+{}
+
+
 static const char *ha_blackhole_exts[] = {
   NullS
 };
diff --git a/sql/ha_blackhole.h b/sql/ha_blackhole.h
index 84a386e17f8..2dccabf17cc 100644
--- a/sql/ha_blackhole.h
+++ b/sql/ha_blackhole.h
@@ -28,9 +28,7 @@ class ha_blackhole: public handler
   THR_LOCK thr_lock;
 
 public:
-  ha_blackhole(TABLE *table): handler(table)
-  {
-  }
+  ha_blackhole(TABLE *table_arg);
   ~ha_blackhole()
   {
   }
diff --git a/sql/ha_federated.cc b/sql/ha_federated.cc
index 1cec6faea04..63454bf88ed 100644
--- a/sql/ha_federated.cc
+++ b/sql/ha_federated.cc
@@ -54,6 +54,7 @@
 
   ***IMPORTANT***
 
+  This is a first release, conceptual release
   Only 'mysql://' is supported at this release.
 
 
@@ -352,7 +353,8 @@
 
 #ifdef HAVE_FEDERATED_DB
 #include "ha_federated.h"
-#define MAX_REMOTE_SIZE IO_SIZE
+
+#include "m_string.h"
 /* Variables for federated share methods */
 static HASH federated_open_tables;              // Hash used to track open
                                                 // tables
@@ -413,13 +415,14 @@ bool federated_db_end()
   return FALSE;
 }
 
-
 /*
  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
@@ -427,23 +430,23 @@ bool federated_db_end()
     table, and if so, does the foreign table exist.
 */
 
-static int check_foreign_data_source(FEDERATED_SHARE *share)
+static int check_foreign_data_source(
+                                     FEDERATED_SHARE *share,
+                                     bool table_create_flag)
 {
-  char escaped_table_base_name[IO_SIZE];
-  MYSQL *mysql;
-  MYSQL_RES *result=0;
+  char escaped_table_name[NAME_LEN*2];
+  char query_buffer[FEDERATED_QUERY_BUFFER_SIZE];
+  char error_buffer[FEDERATED_QUERY_BUFFER_SIZE];
   uint error_code;
-  char query_buffer[IO_SIZE];
-  char error_buffer[IO_SIZE];
   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)))
-  {
+  if (!(mysql= mysql_init(NULL)))
     DBUG_RETURN(HA_ERR_OUT_OF_MEM);
-  }
   /* check if we can connect */
   if (!mysql_real_connect(mysql,
                           share->hostname,
@@ -453,11 +456,18 @@ static int check_foreign_data_source(FEDERATED_SHARE *share)
                           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,
-                "unable to connect to database '%s' on host '%s as user '%s' !",
-               share->database, share->hostname, share->username));
-    error_code= ER_CONNECT_TO_MASTER;
+               (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
@@ -468,48 +478,42 @@ static int check_foreign_data_source(FEDERATED_SHARE *share)
       with transactions
     */
     mysql->reconnect= 1;
-    /* 
+    /*
       Note: I am not using INORMATION_SCHEMA because this needs to work with < 5.0
       if we can connect, then make sure the table exists 
+
+      the query will be: SELECT * FROM `tablename` WHERE 1=0
     */
-    query.append("SHOW TABLES LIKE '");
-    escape_string_for_mysql(&my_charset_bin, (char *)escaped_table_base_name,
-                            sizeof(escaped_table_base_name),
-                            share->table_base_name,
-                            share->table_base_name_length);
-    query.append(escaped_table_base_name);
-    query.append("'");
-
-    error_code= ER_QUERY_ON_MASTER;
+    query.append(FEDERATED_SELECT);
+    query.append(FEDERATED_STAR);
+    query.append(FEDERATED_FROM);
+    query.append(FEDERATED_BTICK);
+    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);
+    query.append(FEDERATED_BTICK);
+    query.append(FEDERATED_WHERE);
+    query.append(FEDERATED_FALSE);
+
+    DBUG_PRINT("info", ("check_foreign_data_source query %s", query.c_ptr_quick()));
     if (mysql_real_query(mysql, query.ptr(), query.length()))
-      goto error;
-
-    result= mysql_store_result(mysql);
-    if (! result)
-      goto error;
-
-    /* if ! mysql_num_rows, the table doesn't exist, send error */
-    if (! mysql_num_rows(result))
     {
-      my_sprintf(error_buffer,
-                 (error_buffer, "foreign table '%s' does not exist!",
-                  share->table_base_name));
+      error_code= table_create_flag ?
+        ER_CANT_CREATE_FEDERATED_TABLE : ER_FOREIGN_DATA_SOURCE_DOESNT_EXIST;
+      my_sprintf(error_buffer, (error_buffer, ": %d : %s",
+                mysql_errno(mysql), mysql_error(mysql)));
+
+      my_error(error_code, MYF(0), error_buffer);
       goto error;
     }
-    mysql_free_result(result);
-    result= 0;
-    mysql_close(mysql);
-
   }
-  DBUG_RETURN(0);
+  error_code=0;
 
 error:
-    if (result)
-      mysql_free_result(result);
     mysql_close(mysql);
-    my_error(error_code, MYF(0), error_buffer);
     DBUG_RETURN(error_code);
-
 }
 
 
@@ -545,22 +549,27 @@ error:
     'password' and 'port' are both optional.
 
   RETURN VALUE
-    0   success
-    1  failure, wrong string format
+    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_CANT_CREATE_TABLE :
-                   ER_CONNECT_TO_MASTER);
+  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;
   share->scheme= my_strdup(table->s->comment, MYF(0));
+  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, "://")))
   {
     share->scheme[share->username - share->scheme]= '\0';
@@ -613,20 +622,20 @@ static int parse_url(FEDERATED_SHARE *share, TABLE *table,
             share->port= atoi(share->sport);
         }
 
-        if ((share->table_base_name= strchr(share->database, '/')))
+        if ((share->table_name= strchr(share->database, '/')))
         {
-          share->database[share->table_base_name - share->database]= '\0';
-          share->table_base_name++;
+          share->database[share->table_name - share->database]= '\0';
+          share->table_name++;
         }
         else
           goto error;
 
-        share->table_base_name_length= strlen(share->table_base_name);
+        share->table_name_length= strlen(share->table_name);
       }
       else
         goto error;
       /* make sure there's not an extra / */
-      if ((strchr(share->table_base_name, '/')))
+      if ((strchr(share->table_name, '/')))
         goto error;
 
       if (share->hostname[0] == '\0')
@@ -645,7 +654,7 @@ static int parse_url(FEDERATED_SHARE *share, TABLE *table,
                   hostname %s port %d database %s tablename %s\n",
                   share->scheme, share->username, share->password,
                   share->hostname, share->port, share->database,
-                  share->table_base_name));
+                  share->table_name));
     }
     else
       goto error;
@@ -656,13 +665,51 @@ static int parse_url(FEDERATED_SHARE *share, TABLE *table,
   DBUG_RETURN(0);
 
 error:
-    my_error(error_num, MYF(0),
-             "connection string is not in the correct format",0);
-    DBUG_RETURN(1);
+  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;
+  }
+  my_error(error_num, MYF(0), table->s->comment);
+  DBUG_RETURN(error_num);
 
 }
 
 
+/* Federated storage engine handlerton */
+
+static handlerton federated_hton= {
+  "FEDERATED",
+  0,       /* slot */
+  0,       /* savepoint size. */
+  0,       /* close_connection */
+  0,       /* savepoint */
+  0,       /* rollback to savepoint */
+  0,       /* release savepoint */
+  0,       /* commit */
+  0,       /* rollback */
+  0,       /* prepare */
+  0,       /* recover */
+  0,       /* commit_by_xid */
+  0,       /* rollback_by_xid */
+  HTON_NO_FLAGS
+};
+
+
+/*****************************************************************************
+** FEDERATED tables
+*****************************************************************************/
+
+ha_federated::ha_federated(TABLE *table_arg)
+  :handler(&federated_hton, table_arg),
+  mysql(0), stored_result(0), scan_flag(0),
+  ref_length(sizeof(MYSQL_ROW_OFFSET)), current_position(0)
+{}
+
+
 /*
   Convert MySQL result set row to handler internal format
 
@@ -684,24 +731,115 @@ error:
 
 uint ha_federated::convert_row_to_internal_format(byte *record, MYSQL_ROW row)
 {
-  ulong *lengths;
   uint num_fields;
-  uint x= 0;
+  ulong *lengths;
+  Field **field;
 
   DBUG_ENTER("ha_federated::convert_row_to_internal_format");
 
-  num_fields= mysql_num_fields(result);
-  lengths= mysql_fetch_lengths(result);
+  num_fields= mysql_num_fields(stored_result);
+  lengths= mysql_fetch_lengths(stored_result);
 
   memset(record, 0, table->s->null_bytes);
 
-  for (Field **field= table->field; *field; field++, x++)
+  for (field= table->field; *field; field++)
   {
+    /*
+      index variable to move us through the row at the
+      same iterative step as the field
+    */
+    int x= field - table->field;
+    my_ptrdiff_t old_ptr;
+    old_ptr= (my_ptrdiff_t) (record - table->record[0]);
+    (*field)->move_field(old_ptr);
     if (!row[x])
       (*field)->set_null();
     else
+    {
+      (*field)->set_notnull();
       (*field)->store(row[x], lengths[x], &my_charset_bin);
+    }
+    (*field)->move_field(-old_ptr);
+  }
+
+  DBUG_DUMP("record", record, table->s->reclength);
+  DBUG_RETURN(0);
+}
+
+static bool emit_key_part_name(String *to, KEY_PART_INFO *part)
+{
+  DBUG_ENTER("emit_key_part_name");
+  if (to->append(FEDERATED_BTICK) ||
+      to->append(part->field->field_name) ||
+      to->append(FEDERATED_BTICK))
+    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(FEDERATED_SQUOTE))
+    DBUG_RETURN(1);
+
+  if (part->type == HA_KEYTYPE_BIT)
+  {
+    char buff[STRING_BUFFER_USUAL_SIZE], *buf= buff;
+
+    *buf++= '0';
+    *buf++= 'x';
+    for (; len; ptr++,len--)
+    {
+      uint tmp= (uint)(uchar) *ptr;
+      *buf++= _dig_vec_upper[tmp >> 4];
+      *buf++= _dig_vec_upper[tmp & 15];
+    }
+    if (to->append(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(FEDERATED_PERCENT))
+    DBUG_RETURN(1);
+
+  if (needs_quotes && to->append(FEDERATED_SQUOTE))
+    DBUG_RETURN(1);
 
   DBUG_RETURN(0);
 }
@@ -716,6 +854,8 @@ uint ha_federated::convert_row_to_internal_format(byte *record, MYSQL_ROW row)
       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,
@@ -726,112 +866,402 @@ uint ha_federated::convert_row_to_internal_format(byte *record, MYSQL_ROW row)
     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
 
-bool ha_federated::create_where_from_key(String *to, KEY *key_info,
-                                         const byte *key, uint key_length)
+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)
 {
-  uint second_loop= 0;
-  KEY_PART_INFO *key_part;
-  bool needs_quotes;
-  String tmp;
+  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 };
 
   DBUG_ENTER("ha_federated::create_where_from_key");
-  for (key_part= key_info->key_part; (int) key_length > 0; key_part++)
-  {
-    Field *field= key_part->field;
-    needs_quotes= field->needs_quotes();
-    uint length= key_part->length;
+  tmp.length(0); 
+  if (start_key == NULL && end_key == NULL)
+    DBUG_RETURN(1);
 
-    if (second_loop++ && to->append(" AND ", 5))
-      DBUG_RETURN(1);
-    if (to->append('`') || to->append(field->field_name) || to->append("` ", 2))
-      DBUG_RETURN(1);                           // Out of memory
+  for (int i= 0; i <= 1; i++)
+  {
+    bool needs_quotes;
+    uint loop_counter= 0;
+    KEY_PART_INFO *key_part;
+    if (ranges[i] == NULL)
+      continue;
+    const byte *key= ranges[i]->key;
+    uint key_length= ranges[i]->length;
 
-    if (key_part->null_bit)
+    if (both_not_null)
     {
-      if (*key++)
-      {
-        if (to->append("IS NULL", 7))
-          DBUG_RETURN(1);
-
-        DBUG_PRINT("info",
-                   ("NULL type %s", to->c_ptr_quick()));
-        key_length-= key_part->store_length;
-        key+= key_part->store_length - 1;
-        continue;
-      }
-      key_length--;
+      if (i > 0)
+        tmp.append(FEDERATED_CONJUNCTION);
+      else
+        tmp.append(FEDERATED_OPENPAREN);
     }
-    if (to->append("= "))
-      DBUG_RETURN(1);
-    if (needs_quotes && to->append("'"))
-      DBUG_RETURN(1);
-    if (key_part->type == HA_KEYTYPE_BIT)
-    {
-      /* This is can be treated as a hex string */
-      Field_bit *field= (Field_bit *) (key_part->field);
-      char buff[64 + 2], *ptr;
-      byte *end= (byte*)(key)+length;
-
-      buff[0]= '0';
-      buff[1]= 'x';
-      for (ptr= buff + 2; key < end; key++)
-      {
-        uint tmp= (uint)(uchar) *key;
-        *ptr++= _dig_vec_upper[tmp >> 4];
-        *ptr++= _dig_vec_upper[tmp & 15];
-      }
-      if (to->append(buff, (uint)(ptr - buff)))
-        DBUG_RETURN(1);
 
-      key_length-= length;
-      continue;
-    }
-    if (key_part->key_part_flag & HA_BLOB_PART)
+    for (key_part= key_info->key_part,
+         remainder= key_info->key_parts,
+         length= ranges[i]->length,
+         ptr= ranges[i]->key; ;
+         remainder--,
+         key_part++)
     {
-      uint blob_length= uint2korr(key);
-      key+= HA_KEY_BLOB_LENGTH;
-      key_length-= HA_KEY_BLOB_LENGTH;
+      Field *field= key_part->field;
+      uint store_length= key_part->store_length;
+      uint part_length= min(store_length, length);
+      needs_quotes= field->needs_quotes();
+      DBUG_DUMP("key, start of loop", (char *) ptr, length);
 
-      tmp.set_quick((char*) key, blob_length, &my_charset_bin);
-      if (append_escaped(to, &tmp))
-        DBUG_RETURN(1);
+      if (key_part->null_bit)
+      {
+        if (*ptr++)
+        {
+          if (emit_key_part_name(&tmp, key_part) ||
+              tmp.append(FEDERATED_ISNULL))
+            DBUG_RETURN(1);
+          continue;
+        }
+      }
 
-      length= key_part->length;
-    }
-    else if (key_part->key_part_flag & HA_VAR_LENGTH_PART)
-    {
-      length= uint2korr(key);
-      key+= HA_KEY_BLOB_LENGTH;
-      tmp.set_quick((char*) key, length, &my_charset_bin);
-      if (append_escaped(to, &tmp))
+      if (tmp.append(FEDERATED_OPENPAREN))
         DBUG_RETURN(1);
-    }
-    else
-    {
-      char buff[MAX_FIELD_WIDTH];
-      String str(buff, sizeof(buff), field->charset()), *res;
 
-      res= field->val_str(&str, (char*) (key));
-      if (field->result_type() == STRING_RESULT)
-      {
-        if (append_escaped(to, res))
+      switch(ranges[i]->flag) {
+      case(HA_READ_KEY_EXACT):
+        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))
+            DBUG_RETURN(1);
+
+          if (records_in_range)
+          {
+            if (tmp.append(FEDERATED_GE))
+              DBUG_RETURN(1);
+          }
+          else
+          {
+            if (tmp.append(FEDERATED_EQ))
+              DBUG_RETURN(1);
+          }
+
+          if (emit_key_part_element(&tmp, key_part, needs_quotes, 0, ptr,
+                                    part_length))
+            DBUG_RETURN(1);
+        }
+        else
+          /* LIKE */
+        {
+          if (emit_key_part_name(&tmp, key_part) ||
+              tmp.append(FEDERATED_LIKE) ||
+              emit_key_part_element(&tmp, key_part, needs_quotes, 1, ptr,
+                                    part_length))
+            DBUG_RETURN(1);
+        }
+        break;
+      case(HA_READ_AFTER_KEY):
+        if (store_length >= length) /* end key */
+        {
+          if (emit_key_part_name(&tmp, key_part))
+            DBUG_RETURN(1);
+
+          if (i > 0) /* end key */
+          {
+            if (tmp.append(FEDERATED_LE))
+              DBUG_RETURN(1);
+          }
+          else /* start key */
+          {
+            if (tmp.append(FEDERATED_GT))
+              DBUG_RETURN(1);
+          }
+
+          if (emit_key_part_element(&tmp, key_part, needs_quotes, 0, ptr,
+                                    part_length))
+          {
+            DBUG_RETURN(1);
+          }
+          break;
+        }
+      case(HA_READ_KEY_OR_NEXT):
+        if (emit_key_part_name(&tmp, key_part) ||
+            tmp.append(FEDERATED_GE) ||
+            emit_key_part_element(&tmp, key_part, needs_quotes, 0, ptr,
+              part_length))
+          DBUG_RETURN(1);
+        break;
+      case(HA_READ_BEFORE_KEY):
+        if (store_length >= length)
+        {
+          if (emit_key_part_name(&tmp, key_part) ||
+              tmp.append(FEDERATED_LT) ||
+              emit_key_part_element(&tmp, key_part, needs_quotes, 0, ptr,
+                                    part_length))
+            DBUG_RETURN(1);
+          break;
+        }
+      case(HA_READ_KEY_OR_PREV):
+        if (emit_key_part_name(&tmp, key_part) ||
+            tmp.append(FEDERATED_LE) ||
+            emit_key_part_element(&tmp, key_part, needs_quotes, 0, ptr,
+                                  part_length))
           DBUG_RETURN(1);
-        res= field->val_str(&str, (char*) (key));
+        break;
+      default:
+        DBUG_PRINT("info",("cannot handle flag %d", ranges[i]->flag));
+        DBUG_RETURN(1);
       }
-      else if (to->append(res->ptr(), res->length()))
+      if (tmp.append(FEDERATED_CLOSEPAREN))
         DBUG_RETURN(1);
+
+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(FEDERATED_AND))
+        DBUG_RETURN(1);
+
+      DBUG_PRINT("info",
+                 ("create_where_from_key WHERE clause: %s",
+                  tmp.c_ptr_quick()));
     }
-    if (needs_quotes && to->append("'"))
-      DBUG_RETURN(1);
-    DBUG_PRINT("info",
-               ("final value for 'to' %s", to->c_ptr_quick()));
-    key+= length;
-    key_length-= length;
-    DBUG_RETURN(0);
   }
-  DBUG_RETURN(1);
+  if (both_not_null)
+    if (tmp.append(FEDERATED_CLOSEPAREN))
+      DBUG_RETURN(1);
+
+  if (to->append(FEDERATED_WHERE))
+    DBUG_RETURN(1);
+
+  if (to->append(tmp))
+    DBUG_RETURN(1);
+
+  DBUG_RETURN(0);
 }
 
 /*
@@ -842,40 +1272,47 @@ bool ha_federated::create_where_from_key(String *to, KEY *key_info,
 
 static FEDERATED_SHARE *get_share(const char *table_name, TABLE *table)
 {
-  FEDERATED_SHARE *share;
-  char query_buffer[IO_SIZE];
+  char *select_query, *tmp_table_name;
+  char query_buffer[FEDERATED_QUERY_BUFFER_SIZE];
+  uint tmp_table_name_length;
+  Field **field;
   String query(query_buffer, sizeof(query_buffer), &my_charset_bin);
+  FEDERATED_SHARE *share;
+  /*
+    In order to use this string, we must first zero it's length,
+    or it will contain garbage
+   */
   query.length(0);
 
-  uint table_name_length, table_base_name_length;
-  char *tmp_table_name, *table_base_name, *select_query;
-
-  /* share->table_name has the file location - we want the table's name!  */
-  table_base_name= (char*) table->s->table_name;
-  DBUG_PRINT("info", ("table_name %s", table_base_name));
-  /*
-    So why does this exist? There is no way currently to init a storage engine.
-    Innodb and BDB both have modifications to the server to allow them to
-    do this. Since you will not want to do this, this is probably the next
-    best method.
-  */
   pthread_mutex_lock(&federated_mutex);
-  table_name_length= (uint) strlen(table_name);
-  table_base_name_length= (uint) strlen(table_base_name);
+  tmp_table_name= (char *)table->s->table_name;
+  tmp_table_name_length= (uint) strlen(tmp_table_name);
 
   if (!(share= (FEDERATED_SHARE *) hash_search(&federated_open_tables,
                                                (byte*) table_name,
-                                               table_name_length)))
+                                               strlen(table_name))))
   {
     query.set_charset(system_charset_info);
-    query.append("SELECT * FROM `");
+    query.append(FEDERATED_SELECT);
+    for (field= table->field; *field; field++)
+    {
+      query.append(FEDERATED_BTICK);
+      query.append((*field)->field_name);
+      query.append(FEDERATED_BTICK);
+      query.append(FEDERATED_COMMA);
+    }
+    query.length(query.length()- strlen(FEDERATED_COMMA));
+    query.append(FEDERATED_FROM);
+    query.append(FEDERATED_BTICK);
+
 
     if (!(share= (FEDERATED_SHARE *)
-          my_multi_malloc(MYF(MY_WME | MY_ZEROFILL),
+          my_multi_malloc(MYF(MY_WME),
                           &share, sizeof(*share),
-                          &tmp_table_name, table_name_length + 1,
-                          &select_query, query.length() +
-                          strlen(table->s->comment) + 1,  NullS)))
+                          &tmp_table_name, tmp_table_name_length+ 1,
+                          &select_query,
+                          query.length()+strlen(table->s->comment)+1,
+                          NullS)))
     {
       pthread_mutex_unlock(&federated_mutex);
       return NULL;
@@ -884,16 +1321,15 @@ static FEDERATED_SHARE *get_share(const char *table_name, TABLE *table)
     if (parse_url(share, table, 0))
       goto error;
 
-    query.append(share->table_base_name);
-    query.append("`");
-    share->use_count= 0;
-    share->table_name_length= table_name_length;
-    share->table_name= tmp_table_name;
+    query.append(share->table_name, share->table_name_length);
+    query.append(FEDERATED_BTICK);
     share->select_query= select_query;
-    strmov(share->table_name, table_name);
     strmov(share->select_query, query.ptr());
+    share->use_count= 0;
+    share->table_name_length= strlen(share->table_name);
     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);
@@ -907,9 +1343,10 @@ static FEDERATED_SHARE *get_share(const char *table_name, TABLE *table)
 error:
   pthread_mutex_unlock(&federated_mutex);
   if (share->scheme)
+  {
     my_free((gptr) share->scheme, MYF(0));
-  my_free((gptr) share, MYF(0));
-
+    share->scheme= 0;
+  }
   return NULL;
 }
 
@@ -922,12 +1359,16 @@ error:
 
 static int free_share(FEDERATED_SHARE *share)
 {
+  DBUG_ENTER("free_share");
   pthread_mutex_lock(&federated_mutex);
 
   if (!--share->use_count)
   {
     if (share->scheme)
+    {
       my_free((gptr) share->scheme, MYF(0));
+      share->scheme= 0;
+    }
 
     hash_delete(&federated_open_tables, (byte*) share);
     thr_lock_delete(&share->lock);
@@ -936,23 +1377,37 @@ static int free_share(FEDERATED_SHARE *share)
   }
   pthread_mutex_unlock(&federated_mutex);
 
-  return 0;
+  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.
 */
-static const char *ha_federated_exts[] = {
-  NullS
-};
 
 const char **ha_federated::bas_ext() const
 {
-  return ha_federated_exts;
+  static const char *ext[]=
+  {
+    NullS
+  };
+  return ext;
 }
 
 
@@ -969,6 +1424,7 @@ const char **ha_federated::bas_ext() const
 
 int ha_federated::open(const char *name, int mode, uint test_if_locked)
 {
+  int rc;
   DBUG_ENTER("ha_federated::open");
 
   if (!(share= get_share(name, table)))
@@ -977,11 +1433,6 @@ int ha_federated::open(const char *name, int mode, uint test_if_locked)
 
   /* Connect to foreign database mysql_real_connect() */
   mysql= mysql_init(0);
-  DBUG_PRINT("info", ("hostname %s", share->hostname));
-  DBUG_PRINT("info", ("username %s", share->username));
-  DBUG_PRINT("info", ("password %s", share->password));
-  DBUG_PRINT("info", ("database %s", share->database));
-  DBUG_PRINT("info", ("port %d", share->port));
   if (!mysql_real_connect(mysql,
                           share->hostname,
                           share->username,
@@ -990,8 +1441,13 @@ int ha_federated::open(const char *name, int mode, uint test_if_locked)
                           share->port,
                           share->socket, 0))
   {
-    my_error(ER_CONNECT_TO_MASTER, MYF(0), mysql_error(mysql));
-    DBUG_RETURN(ER_CONNECT_TO_MASTER);
+    int error_code;
+    char error_buffer[FEDERATED_QUERY_BUFFER_SIZE];
+    error_code= ER_CONNECT_TO_FOREIGN_DATA_SOURCE;
+    my_sprintf(error_buffer, (error_buffer, ": %d : %s",
+                              mysql_errno(mysql), mysql_error(mysql)));
+    my_error(error_code, MYF(0), error_buffer);
+    DBUG_RETURN(error_code);
   }
   /*
     Since we do not support transactions at this version, we can let the client
@@ -1016,19 +1472,21 @@ int ha_federated::open(const char *name, int mode, uint test_if_locked)
 
 int ha_federated::close(void)
 {
+  int retval;
   DBUG_ENTER("ha_federated::close");
 
   /* free the result set */
-  if (result)
+  if (stored_result)
   {
     DBUG_PRINT("info",
-               ("mysql_free_result result at address %lx", result));
-    mysql_free_result(result);
-    result= 0;
+               ("mysql_free_result result at address %lx", stored_result));
+    mysql_free_result(stored_result);
+    stored_result= 0;
   }
   /* Disconnect from mysql */
   mysql_close(mysql);
-  DBUG_RETURN(free_share(share));
+  retval= free_share(share);
+  DBUG_RETURN(retval);
 
 }
 
@@ -1085,61 +1543,71 @@ inline uint field_in_record_is_null(TABLE *table,
 
 int ha_federated::write_row(byte *buf)
 {
-  uint x, num_fields;
+  bool has_fields= 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;
 
-  char insert_buffer[IO_SIZE];
-  char values_buffer[IO_SIZE], insert_field_value_buffer[IO_SIZE];
-
   /* The main insert query string */
   String insert_string(insert_buffer, sizeof(insert_buffer), &my_charset_bin);
-  insert_string.length(0);
   /* The string containing the values to be added to the insert */
   String values_string(values_buffer, sizeof(values_buffer), &my_charset_bin);
-  values_string.length(0);
   /* 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);
+  values_string.length(0);
+  insert_string.length(0);
   insert_field_value_string.length(0);
 
   DBUG_ENTER("ha_federated::write_row");
-  DBUG_PRINT("info", ("table charset name %s csname %s",
-                                         table->s->table_charset->name, table->s->table_charset->csname));
+  DBUG_PRINT("info",
+             ("table charset name %s csname %s",
+              table->s->table_charset->name,
+              table->s->table_charset->csname));
 
   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 off our string */
-  insert_string.append("INSERT INTO `");
-  insert_string.append(share->table_base_name);
-  insert_string.append("`");
-  /* start both our field and field values strings */
-  insert_string.append(" (");
-  values_string.append(" VALUES (");
+  /*
+    start both our field and field values strings
+  */
+  insert_string.append(FEDERATED_INSERT);
+  insert_string.append(FEDERATED_BTICK);
+  insert_string.append(share->table_name, share->table_name_length);
+  insert_string.append(FEDERATED_BTICK);
+  insert_string.append(FEDERATED_OPENPAREN);
+
+  values_string.append(FEDERATED_VALUES);
+  values_string.append(FEDERATED_OPENPAREN);
+
   /*
     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
+
+    You might ask "Why an index variable (has_fields) ?" My answer is that
+    we need to count how many fields we actually need
   */
-  for (num_fields= 0, field= table->field; *field; field++)
+  for (field= table->field; *field; field++)
   {
     /* if there is a query id and if it's equal to the current query id */
     if (ha_get_bit_in_write_set((*field)->fieldnr))
     {
-      num_fields++;
+      /*
+        There are some fields. This will be used later to determine
+        whether to chop off commas and parens.
+      */
+      has_fields= TRUE;
 
       if ((*field)->is_null())
-      {
-        DBUG_PRINT("info", ("column %d field is_null", num_fields-1));
-        insert_field_value_string.append("NULL");
-      }
+        insert_field_value_string.append(FEDERATED_NULL);
       else
       {
-        DBUG_PRINT("info", ("column %d field is not null", num_fields-1));
         (*field)->val_str(&insert_field_value_string);
         /* quote these fields if they require it */
-        (*field)->quote_data(&insert_field_value_string); 
+        (*field)->quote_data(&insert_field_value_string);
       }
       /* append the field name */
       insert_string.append((*field)->field_name);
@@ -1149,34 +1617,35 @@ int ha_federated::write_row(byte *buf)
       insert_field_value_string.length(0);
 
       /* append commas between both fields and fieldnames */
-      insert_string.append(',');
-      values_string.append(',');
-
+      /*
+        unfortunately, we can't use the logic
+        if *(fields + 1) to make the following
+        appends conditional because we may not append
+        if the next field doesn't match the condition:
+        (((*field)->query_id && (*field)->query_id == current_query_id)
+      */
+      insert_string.append(FEDERATED_COMMA);
+      values_string.append(FEDERATED_COMMA);
     }
   }
 
   /*
-    chop of the trailing comma, or if there were no fields, a '('
-    So, "INSERT INTO foo (" becomes "INSERT INTO foo "
-    or, with fields, "INSERT INTO foo (field1, field2," becomes
-    "INSERT INTO foo (field1, field2"
+    remove trailing comma
   */
-  insert_string.chop();
-
+  insert_string.length(insert_string.length() - strlen(FEDERATED_COMMA));
   /*
     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 ();"
   */
-  DBUG_PRINT("info", ("x %d  num fields %d", x, num_fields));
-  if (num_fields > 0)
+  if (has_fields)
   {
     /* chops off leading commas */
-    values_string.chop();
-    insert_string.append(')');
+    values_string.length(values_string.length() - strlen(FEDERATED_COMMA));
+    insert_string.append(FEDERATED_CLOSEPAREN);
   }
   /* we always want to append this, even if there aren't any fields */
-  values_string.append(')');
+  values_string.append(FEDERATED_CLOSEPAREN);
 
   /* add the values */
   insert_string.append(values_string);
@@ -1185,8 +1654,69 @@ int ha_federated::write_row(byte *buf)
 
   if (mysql_real_query(mysql, insert_string.ptr(), insert_string.length()))
   {
-    my_error(ER_QUERY_ON_MASTER, MYF(0), mysql_error(mysql));
-    DBUG_RETURN(ER_QUERY_ON_MASTER);
+    int error_code;
+    char error_buffer[FEDERATED_QUERY_BUFFER_SIZE];
+    error_code= ER_QUERY_ON_FOREIGN_DATA_SOURCE;
+    my_sprintf(error_buffer, (error_buffer, ": %d : %s",
+                              mysql_errno(mysql), mysql_error(mysql)));
+    my_error(error_code, MYF(0), error_buffer);
+    DBUG_RETURN(error_code);
+  }
+
+  DBUG_RETURN(0);
+}
+
+
+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(FEDERATED_OPTIMIZE);
+  query.append(FEDERATED_BTICK);
+  query.append(share->table_name, share->table_name_length);
+  query.append(FEDERATED_BTICK);
+
+  if (mysql_real_query(mysql, query.ptr(), query.length()))
+  {
+    my_error(-1, MYF(0), mysql_error(mysql));
+    DBUG_RETURN(-1);
+  }
+
+  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(FEDERATED_REPAIR);
+  query.append(FEDERATED_BTICK);
+  query.append(share->table_name, share->table_name_length);
+  query.append(FEDERATED_BTICK);
+  if (check_opt->flags & T_QUICK)
+    query.append(FEDERATED_QUICK);
+  if (check_opt->flags & T_EXTEND)
+    query.append(FEDERATED_EXTENDED);
+  if (check_opt->sql_flags & TT_USEFRM)
+    query.append(FEDERATED_USE_FRM);
+      
+  if (mysql_real_query(mysql, query.ptr(), query.length()))
+  {
+    my_error(-1, MYF(0), mysql_error(mysql));
+    DBUG_RETURN(-1);
   }
 
   DBUG_RETURN(0);
@@ -1212,39 +1742,57 @@ int ha_federated::write_row(byte *buf)
 
 int ha_federated::update_row(const byte *old_data, byte *new_data)
 {
-  uint x= 0;
-  uint has_a_primary_key= 0;
-  uint primary_key_field_num;
-  char old_field_value_buffer[IO_SIZE], new_field_value_buffer[IO_SIZE];
-  char update_buffer[IO_SIZE], where_buffer[IO_SIZE];
+  /*
+    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= (table->s->primary_key == 0 ? TRUE : FALSE);
+  /* 
+    buffers for following strings
+  */
+  char error_buffer[FEDERATED_QUERY_BUFFER_SIZE];
+  char old_field_value_buffer[STRING_BUFFER_USUAL_SIZE];
+  char new_field_value_buffer[STRING_BUFFER_USUAL_SIZE];
+  char update_buffer[FEDERATED_QUERY_BUFFER_SIZE];
+  char where_buffer[FEDERATED_QUERY_BUFFER_SIZE];
 
   /* stores the value to be replaced of the field were are updating */
-  String old_field_value(old_field_value_buffer, sizeof(old_field_value_buffer),
+  String old_field_value(old_field_value_buffer,
+                         sizeof(old_field_value_buffer),
                          &my_charset_bin);
   /* stores the new value of the field */
-  String new_field_value(new_field_value_buffer, sizeof(new_field_value_buffer),
+  String new_field_value(new_field_value_buffer,
+                         sizeof(new_field_value_buffer),
                          &my_charset_bin);
   /* stores the update query */
-  String update_string(update_buffer, sizeof(update_buffer), &my_charset_bin);
+  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);
+  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
+  */
   old_field_value.length(0);
   new_field_value.length(0);
   update_string.length(0);
   where_string.length(0);
 
-  has_a_primary_key= (table->s->primary_key == 0 ? 1 : 0);
-  primary_key_field_num= has_a_primary_key ?
-    table->key_info[table->s->primary_key].key_part->fieldnr - 1 : -1;
-  if (has_a_primary_key)
-    DBUG_PRINT("info", ("has a primary key"));
-
-  update_string.append("UPDATE `");
-  update_string.append(share->table_base_name);
-  update_string.append("`");
-  update_string.append(" SET ");
+  update_string.append(FEDERATED_UPDATE);
+  update_string.append(FEDERATED_BTICK);
+  update_string.append(share->table_name);
+  update_string.append(FEDERATED_BTICK);
+  update_string.append(FEDERATED_SET);
 
 /*
   In this loop, we want to match column names to values being inserted
@@ -1256,104 +1804,73 @@ int ha_federated::update_row(const byte *old_data, byte *new_data)
   field=oldvalue
  */
 
-  for (Field **field= table->field; *field; field++, x++)
+  for (Field **field= table->field; *field; field++)
   {
-    /*
-      In all of these tests for 'has_a_primary_key', what I'm trying to
-      accomplish is to only use the primary key in the WHERE clause if the
-      table has a primary key, as opposed to a table without a primary key
-      in which case we have to use all the fields to create a WHERE clause
-      using the old/current values, as well as adding a LIMIT statement
-    */
-    if (has_a_primary_key)
-    {
-      if (x == primary_key_field_num)
-        where_string.append((*field)->field_name);
-    }
-    else
-      where_string.append((*field)->field_name);
-
+    where_string.append((*field)->field_name);
     update_string.append((*field)->field_name);
-    update_string.append('=');
+    update_string.append(FEDERATED_EQ);
 
     if ((*field)->is_null())
-    {
-      DBUG_PRINT("info", ("column %d is NULL", x ));
-      new_field_value.append("NULL");
-    }
+      new_field_value.append(FEDERATED_NULL);
     else
     {
       /* otherwise = */
       (*field)->val_str(&new_field_value);
       (*field)->quote_data(&new_field_value);
 
-      if (has_a_primary_key)
-      {
-        if (x == primary_key_field_num)
-          where_string.append("=");
-      }
-      else if (!field_in_record_is_null(table, *field, (char*) old_data))
-        where_string.append("=");
+      if (!field_in_record_is_null(table, *field, (char*) old_data))
+        where_string.append(FEDERATED_EQ);
     }
 
-    if (has_a_primary_key)
-    {
-      if (x == primary_key_field_num)
-      {
-        (*field)->val_str(&old_field_value,
-                          (char*) (old_data + (*field)->offset()));
-        (*field)->quote_data(&old_field_value);
-        where_string.append(old_field_value);
-      }
-    }
+    if (field_in_record_is_null(table, *field, (char*) old_data))
+      where_string.append(FEDERATED_ISNULL);
     else
     {
-      if (field_in_record_is_null(table, *field, (char*) old_data))
-        where_string.append(" IS NULL ");
-      else
-      {
-        uint o_len;
-        (*field)->val_str(&old_field_value,
-                          (char*) (old_data + (*field)->offset()));
-        o_len= (*field)->pack_length();
-        DBUG_PRINT("info", ("o_len %lu", o_len));
-        (*field)->quote_data(&old_field_value);
-        where_string.append(old_field_value);
-      }
+      uint o_len;
+      (*field)->val_str(&old_field_value,
+                        (char*) (old_data + (*field)->offset()));
+      o_len= (*field)->pack_length();
+      (*field)->quote_data(&old_field_value);
+      where_string.append(old_field_value);
     }
-    DBUG_PRINT("info",
-               ("column %d new value %s old value %s",
-                x, new_field_value.c_ptr_quick(), old_field_value.c_ptr_quick() ));
+
     update_string.append(new_field_value);
     new_field_value.length(0);
 
-    if (x + 1 < table->s->fields)
+    /*
+      Only append conjunctions if we have another field in which
+      to iterate
+    */
+    if (*(field + 1))
     {
-      update_string.append(", ");
-      if (!has_a_primary_key)
-        where_string.append(" AND ");
+      update_string.append(FEDERATED_COMMA);
+      where_string.append(FEDERATED_AND);
     }
     old_field_value.length(0);
   }
-  update_string.append(" WHERE ");
+  update_string.append(FEDERATED_WHERE);
   update_string.append(where_string);
-  if (! has_a_primary_key)
-    update_string.append(" LIMIT 1");
+  /*
+    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(FEDERATED_LIMIT1);
 
-  DBUG_PRINT("info", ("Final update query: %s",
-                                          update_string.c_ptr_quick()));
   if (mysql_real_query(mysql, update_string.ptr(), update_string.length()))
   {
-    my_error(ER_QUERY_ON_MASTER, MYF(0), mysql_error(mysql));
-    DBUG_RETURN(ER_QUERY_ON_MASTER);
+    int error_code= ER_QUERY_ON_FOREIGN_DATA_SOURCE;
+    char error_buffer[FEDERATED_QUERY_BUFFER_SIZE];
+    my_sprintf(error_buffer, (error_buffer, ": %d : %s",
+                              mysql_errno(mysql), mysql_error(mysql)));
+    my_error(error_code, MYF(0), error_buffer);
+    DBUG_RETURN(error_code);
   }
-
-
   DBUG_RETURN(0);
 }
 
 /*
-  This will delete a row. 'buf' will contain a copy of the row to be deleted.
+  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
@@ -1369,49 +1886,60 @@ int ha_federated::update_row(const byte *old_data, byte *new_data)
 
 int ha_federated::delete_row(const byte *buf)
 {
-  char delete_buffer[IO_SIZE];
-  char data_buffer[IO_SIZE];
+  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);
+  delete_string.length(0);
+  data_string.length(0);
+
   DBUG_ENTER("ha_federated::delete_row");
 
-  delete_string.length(0);
-  delete_string.append("DELETE FROM `");
-  delete_string.append(share->table_base_name);
-  delete_string.append("`");
-  delete_string.append(" WHERE ");
+  delete_string.append(FEDERATED_DELETE);
+  delete_string.append(FEDERATED_FROM);
+  delete_string.append(FEDERATED_BTICK);
+  delete_string.append(share->table_name);
+  delete_string.append(FEDERATED_BTICK);
+  delete_string.append(FEDERATED_WHERE);
 
   for (Field **field= table->field; *field; field++)
   {
-    Field *cur_field= *field;
-    data_string.length(0);
-    delete_string.append(cur_field->field_name);
+    delete_string.append((*field)->field_name);
 
-    if (cur_field->is_null_in_record((const uchar*) buf))
+    if ((*field)->is_null())
     {
-      delete_string.append(" IS ");
-      data_string.append("NULL");
+      delete_string.append(FEDERATED_IS);
+      data_string.append(FEDERATED_NULL);
     }
     else
     {
-      delete_string.append("=");
-      cur_field->val_str(&data_string, (char*) buf+ cur_field->offset());
-      cur_field->quote_data(&data_string);
+      delete_string.append(FEDERATED_EQ);
+      (*field)->val_str(&data_string);
+      (*field)->quote_data(&data_string);
     }
 
     delete_string.append(data_string);
-    delete_string.append(" AND ");
+    data_string.length(0);
+
+    if (*(field + 1))
+      delete_string.append(FEDERATED_AND);
   }
 
-  delete_string.length(delete_string.length()-5); // Remove AND
-  delete_string.append(" LIMIT 1");
+  delete_string.append(FEDERATED_LIMIT1);
   DBUG_PRINT("info",
              ("Delete sql: %s", delete_string.c_ptr_quick()));
   if (mysql_real_query(mysql, delete_string.ptr(), delete_string.length()))
   {
-    my_error(ER_QUERY_ON_MASTER, MYF(0), mysql_error(mysql));
-    DBUG_RETURN(ER_QUERY_ON_MASTER);
+    int error_code= ER_QUERY_ON_FOREIGN_DATA_SOURCE;
+    char error_buffer[FEDERATED_QUERY_BUFFER_SIZE];
+    my_error(error_code, MYF(0), error_buffer);
+    DBUG_RETURN(error_code);
   }
+  deleted+= mysql->affected_rows;
+  DBUG_PRINT("info",
+             ("rows deleted %d rows deleted for all time %d",
+             int(mysql->affected_rows), deleted));
 
   DBUG_RETURN(0);
 }
@@ -1425,12 +1953,12 @@ int ha_federated::delete_row(const byte *buf)
 */
 
 int ha_federated::index_read(byte *buf, const byte *key,
-                             uint key_len __attribute__ ((unused)),
-                             enum ha_rkey_function find_flag
-                             __attribute__ ((unused)))
+                             uint key_len, enum ha_rkey_function find_flag)
 {
+  int retval;
   DBUG_ENTER("ha_federated::index_read");
-  DBUG_RETURN(index_read_idx(buf, active_index, key, key_len, find_flag));
+  retval= index_read_idx(buf, active_index, key, key_len, find_flag);
+  DBUG_RETURN(retval);
 }
 
 
@@ -1444,17 +1972,23 @@ int ha_federated::index_read(byte *buf, const byte *key,
 */
 
 int ha_federated::index_read_idx(byte *buf, uint index, const byte *key,
-                                 uint key_len __attribute__ ((unused)),
-                                 enum ha_rkey_function find_flag
-                                 __attribute__ ((unused)))
+                                 uint key_len, enum ha_rkey_function find_flag)
 {
-  char index_value[IO_SIZE];
-  String index_string(index_value, sizeof(index_value), &my_charset_bin);
-  index_string.length(0);
-  uint keylen;
+  int retval;
+  char error_buffer[FEDERATED_QUERY_BUFFER_SIZE];
+  char index_value[STRING_BUFFER_USUAL_SIZE];
+  char key_value[STRING_BUFFER_USUAL_SIZE];
+  char test_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;
 
-  char sql_query_buffer[IO_SIZE];
-  String sql_query(sql_query_buffer, sizeof(sql_query_buffer), &my_charset_bin);
+  index_string.length(0);
   sql_query.length(0);
 
   DBUG_ENTER("ha_federated::index_read_idx");
@@ -1463,10 +1997,14 @@ int ha_federated::index_read_idx(byte *buf, uint index, const byte *key,
                       &LOCK_status);
 
   sql_query.append(share->select_query);
-  sql_query.append(" WHERE ");
 
-  keylen= strlen((char*) (key));
-  create_where_from_key(&index_string, &table->key_info[index], key, keylen);
+  range.key= key;
+  range.length= key_len;
+  range.flag= find_flag;
+  create_where_from_key(&index_string,
+                        &table->key_info[index],
+                        &range,
+                        NULL, 0);
   sql_query.append(index_string);
 
   DBUG_PRINT("info",
@@ -1478,42 +2016,50 @@ int ha_federated::index_read_idx(byte *buf, uint index, const byte *key,
              ("current position %d sql_query %s", current_position,
               sql_query.c_ptr_quick()));
 
-  if (result)
+  if (stored_result)
   {
-    mysql_free_result(result);
-    result= 0;
+    mysql_free_result(stored_result);
+    stored_result= 0;
   }
   if (mysql_real_query(mysql, sql_query.ptr(), sql_query.length()))
   {
-    my_error(ER_QUERY_ON_MASTER, MYF(0), mysql_error(mysql));
-    DBUG_RETURN(ER_QUERY_ON_MASTER);
+    my_sprintf(error_buffer, (error_buffer, ": %d : %s",
+                              mysql_errno(mysql), mysql_error(mysql)));
+    retval= ER_QUERY_ON_FOREIGN_DATA_SOURCE;
+    goto error;
   }
-  result= mysql_store_result(mysql);
+  stored_result= mysql_store_result(mysql);
 
-  if (!result)
+  if (!stored_result)
   {
-    table->status= STATUS_NOT_FOUND;
-    DBUG_RETURN(HA_ERR_END_OF_FILE);
+    retval= HA_ERR_END_OF_FILE;
+    goto error;
   }
+  /*
+   This basically says that the record in table->record[0] is legal,
+   and that it is ok to use this record, for whatever reason, such
+   as with a join (without it, joins will not work)
+ */
+  table->status= 0;
+
+  retval= rnd_next(buf);
+  DBUG_RETURN(retval);
 
-  if (mysql_errno(mysql))
+error:
+  if (stored_result)
   {
-    table->status= STATUS_NOT_FOUND;
-    DBUG_RETURN(mysql_errno(mysql));
+    mysql_free_result(stored_result);
+    stored_result= 0;
   }
-  /* 
-     This basically says that the record in table->record[0] is legal, and that it is
-     ok to use this record, for whatever reason, such as with a join (without it, joins 
-     will not work)
-  */
-  table->status=0;
-
-  DBUG_RETURN(rnd_next(buf));
+  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)
 {
+  int error;
   DBUG_ENTER("ha_federated::index_init");
   DBUG_PRINT("info",
              ("table: '%s'  key: %d", table->s->table_name, keynr));
@@ -1521,14 +2067,90 @@ int ha_federated::index_init(uint keynr, bool sorted)
   DBUG_RETURN(0);
 }
 
+/*
+
+  int read_range_first(const key_range *start_key,
+                               const key_range *end_key,
+                               bool eq_range, bool sorted);
+*/
+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");
+  if (start_key == NULL && end_key == NULL)
+    DBUG_RETURN(0);
+
+  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);
+
+  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)
+  {
+    DBUG_PRINT("info",
+               ("mysql_free_result address %lx", stored_result));
+    mysql_free_result(stored_result);
+    stored_result= 0;
+  }
+  stored_result= mysql_store_result(mysql);
+
+  if (!stored_result)
+  {
+    retval= HA_ERR_END_OF_FILE;
+    goto error;
+  }
+ 
+  /* This was successful, please let it be known! */
+  table->status= 0;
+
+  retval= rnd_next(table->record[0]);
+  DBUG_RETURN(retval);
+
+error:
+    table->status= STATUS_NOT_FOUND;
+    if (stored_result)
+    {
+      DBUG_PRINT("info", ("mysql_free_result address %lx", stored_result));
+      mysql_free_result(stored_result);
+      stored_result= 0;
+    }
+    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)
 {
+  int retval;
   DBUG_ENTER("ha_federated::index_next");
-  DBUG_RETURN(rnd_next(buf));
+  statistic_increment(table->in_use->status_var.ha_read_next_count,
+		      &LOCK_status);
+  retval= rnd_next(buf);
+  DBUG_RETURN(retval);
 }
-
-
 /*
   rnd_init() is called when the system wants the storage engine to do a table
   scan.
@@ -1544,12 +2166,15 @@ int ha_federated::index_next(byte *buf)
 
 int ha_federated::rnd_init(bool scan)
 {
-  DBUG_ENTER("ha_federated::rnd_init");
+  int num_fields, rows;
+  int retval;
+  char error_buffer[FEDERATED_QUERY_BUFFER_SIZE];
 
+  DBUG_ENTER("ha_federated::rnd_init");
   /*
-    This 'scan' flag is incredibly important for this handler to work
-    properly, especially with updates containing WHERE clauses using
-    indexed columns.
+    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
@@ -1584,40 +2209,48 @@ int ha_federated::rnd_init(bool scan)
   if (scan)
   {
     DBUG_PRINT("info", ("share->select_query %s", share->select_query));
-    if (result)
+    if (stored_result)
     {
       DBUG_PRINT("info",
-                 ("mysql_free_result address %lx", result));
-      mysql_free_result(result);
-      result= 0;
+                 ("mysql_free_result address %lx", stored_result));
+      mysql_free_result(stored_result);
+      stored_result= 0;
     }
 
-    if (mysql_real_query
-        (mysql, share->select_query, strlen(share->select_query)))
-    {
-      my_error(ER_QUERY_ON_MASTER, MYF(0), mysql_error(mysql));
-      DBUG_RETURN(ER_QUERY_ON_MASTER);
-    }
-    result= mysql_store_result(mysql);
+    if (mysql_real_query(mysql,
+                         share->select_query,
+                         strlen(share->select_query)))
+      goto error;
 
-    if (mysql_errno(mysql))
-      DBUG_RETURN(mysql_errno(mysql));
+    stored_result= mysql_store_result(mysql);
+    if (!stored_result)
+      goto error;
   }
   DBUG_RETURN(0);
+
+error:
+      retval= ER_QUERY_ON_FOREIGN_DATA_SOURCE;
+      my_sprintf(error_buffer, (error_buffer, ": %d : %s",
+                mysql_errno(mysql), mysql_error(mysql)));
+      my_error(retval, MYF(0), error_buffer);
+      DBUG_PRINT("info",
+                 ("return error code %d", retval));
+      DBUG_RETURN(retval);
 }
 
 int ha_federated::rnd_end()
 {
+  int retval;
   DBUG_ENTER("ha_federated::rnd_end");
-  if (result)
+
+  if (stored_result)
   {
-    DBUG_PRINT("info", ("mysql_free_result address %lx", result));
-    mysql_free_result(result);
-    result= 0;
+    DBUG_PRINT("info", ("mysql_free_result address %lx", stored_result));
+    mysql_free_result(stored_result);
+    stored_result= 0;
   }
-
-  mysql_free_result(result);
-  DBUG_RETURN(index_end());
+  retval= index_end();
+  DBUG_RETURN(retval);
 }
 
 int ha_federated::index_end(void)
@@ -1639,10 +2272,11 @@ int ha_federated::index_end(void)
 
 int ha_federated::rnd_next(byte *buf)
 {
+  int retval;
   MYSQL_ROW row;
   DBUG_ENTER("ha_federated::rnd_next");
 
-  if (result == 0)
+  if (stored_result == 0)
   {
     /*
       Return value of rnd_init is not always checked (see records.cc),
@@ -1653,12 +2287,13 @@ int ha_federated::rnd_next(byte *buf)
   }
  
   /* Fetch a row, insert it back in a row format. */
-  current_position= result->data_cursor;
+  current_position= stored_result->data_cursor;
   DBUG_PRINT("info", ("current position %d", current_position));
-  if (!(row= mysql_fetch_row(result)))
+  if (!(row= mysql_fetch_row(stored_result)))
     DBUG_RETURN(HA_ERR_END_OF_FILE);
 
-  DBUG_RETURN(convert_row_to_internal_format(buf, row));
+  retval= convert_row_to_internal_format(buf, row);
+  DBUG_RETURN(retval);
 }
 
 
@@ -1705,13 +2340,15 @@ int ha_federated::rnd_pos(byte *buf, byte *pos)
   */
   if (scan_flag)
   {
+    int retval;
     statistic_increment(table->in_use->status_var.ha_read_rnd_count,
                         &LOCK_status);
     memcpy_fixed(&current_position, pos, sizeof(MYSQL_ROW_OFFSET));  // pos
     /* is not aligned */
-    result->current_row= 0;
-    result->data_cursor= current_position;
-    DBUG_RETURN(rnd_next(buf));
+    stored_result->current_row= 0;
+    stored_result->data_cursor= current_position;
+    retval= rnd_next(buf);
+    DBUG_RETURN(retval);
   }
   DBUG_RETURN(0);
 }
@@ -1760,12 +2397,91 @@ int ha_federated::rnd_pos(byte *buf, byte *pos)
     sql_update.cc
 
 */
-/* FIX: later version provide better information to the optimizer */
 
 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");
-  records= 10000; // fix later
+
+  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(FEDERATED_INFO);
+    status_query_string.append(FEDERATED_SQUOTE);
+
+    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(FEDERATED_SQUOTE);
+
+    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)
+        records=         (ha_rows) my_strtoll10(row[4], (char**) 0, &error);
+      if (row[5] != NULL)
+        mean_rec_length= (ha_rows) my_strtoll10(row[5], (char**) 0, &error);
+      if (row[12] != NULL)
+        update_time=      (ha_rows) my_strtoll10(row[12], (char**) 0, &error);
+      if (row[13] != NULL)
+        check_time=      (ha_rows) my_strtoll10(row[13], (char**) 0, &error);
+    }
+    if (flag & HA_STATUS_CONST)
+    {
+      TABLE_SHARE *share= table->s;
+      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;
 }
 
@@ -1786,22 +2502,30 @@ int ha_federated::delete_all_rows()
 {
   DBUG_ENTER("ha_federated::delete_all_rows");
 
-  char query_buffer[IO_SIZE];
+  char query_buffer[FEDERATED_QUERY_BUFFER_SIZE];
   String query(query_buffer, sizeof(query_buffer), &my_charset_bin);
   query.length(0);
 
   query.set_charset(system_charset_info);
-  query.append("TRUNCATE `");
-  query.append(share->table_base_name);
-  query.append("`");
+  query.append(FEDERATED_TRUNCATE);
+  query.append(FEDERATED_BTICK);
+  query.append(share->table_name);
+  query.append(FEDERATED_BTICK);
 
+  /*
+    TRUNCATE won't return anything in mysql_affected_rows
+  */
+  deleted+= records;
   if (mysql_real_query(mysql, query.ptr(), query.length()))
   {
-    my_error(ER_QUERY_ON_MASTER, MYF(0), mysql_error(mysql));
-    DBUG_RETURN(ER_QUERY_ON_MASTER);
+    int error_code= ER_QUERY_ON_FOREIGN_DATA_SOURCE;
+    char error_buffer[FEDERATED_QUERY_BUFFER_SIZE];
+    my_sprintf(error_buffer, (error_buffer, ": %d : %s",
+                              mysql_errno(mysql), mysql_error(mysql)));
+    my_error(error_code, MYF(0), error_buffer);
+    DBUG_RETURN(error_code);
   }
-
-  DBUG_RETURN(HA_ERR_WRONG_COMMAND);
+  DBUG_RETURN(0);
 }
 
 
@@ -1849,7 +2573,7 @@ THR_LOCK_DATA **ha_federated::store_lock(THD *thd,
     */
 
     if ((lock_type >= TL_WRITE_CONCURRENT_INSERT &&
-         lock_type <= TL_WRITE) && !thd->in_lock_tables && !thd->tablespace_op)
+         lock_type <= TL_WRITE) && !thd->in_lock_tables)
       lock_type= TL_WRITE_ALLOW_WRITE;
 
     /*
@@ -1879,28 +2603,33 @@ THR_LOCK_DATA **ha_federated::store_lock(THD *thd,
 int ha_federated::create(const char *name, TABLE *table_arg,
                          HA_CREATE_INFO *create_info)
 {
-  int connection_error=0;
-  FEDERATED_SHARE tmp;
+  int retval= 0;
+  /*
+    only a temporary share, to test the url
+  */
+  FEDERATED_SHARE tmp_share;
   DBUG_ENTER("ha_federated::create");
 
-  if (parse_url(&tmp, table_arg, 1))
-  {
-    my_error(ER_CANT_CREATE_TABLE, MYF(0), name, 1);
+  if ((retval= parse_url(&tmp_share, table_arg, 1)))
     goto error;
-  }
-  if ((connection_error= check_foreign_data_source(&tmp)))
-  {
-    my_error(connection_error, MYF(0), name, 1);
+
+  if ((retval= check_foreign_data_source(&tmp_share, 1)))
     goto error;
+
+  if (tmp_share.scheme)
+  {
+    my_free((gptr) tmp_share.scheme, MYF(0));
+    tmp_share.scheme= 0;
   }
-  
-  my_free((gptr) tmp.scheme, MYF(0));
-  DBUG_RETURN(0);
-  
+  DBUG_RETURN(retval);
+
 error:
-  DBUG_PRINT("info", ("errors, returning %d", ER_CANT_CREATE_TABLE));
-  my_free((gptr) tmp.scheme, MYF(0));
-  DBUG_RETURN(ER_CANT_CREATE_TABLE);
+  if (tmp_share.scheme)
+  {
+    my_free((gptr) tmp_share.scheme, MYF(0));
+    tmp_share.scheme= 0;
+  }
+  DBUG_RETURN(retval);
 
 }
 #endif /* HAVE_FEDERATED_DB */
diff --git a/sql/ha_federated.h b/sql/ha_federated.h
index 5908fe5afba..ecaa59d1268 100644
--- a/sql/ha_federated.h
+++ b/sql/ha_federated.h
@@ -26,16 +26,95 @@
 #endif
 
 #include <mysql.h>
-//#include <client.h>
+
+#define FEDERATED_QUERY_BUFFER_SIZE STRING_BUFFER_USUAL_SIZE * 5
+#define FEDERATED_RECORDS_IN_RANGE 2
+
+#define FEDERATED_INFO " SHOW TABLE STATUS LIKE "
+#define FEDERATED_INFO_LEN sizeof(FEDERATED_INFO)
+#define FEDERATED_SELECT "SELECT "
+#define FEDERATED_SELECT_LEN sizeof(FEDERATED_SELECT)
+#define FEDERATED_WHERE " WHERE "
+#define FEDERATED_WHERE_LEN sizeof(FEDERATED_WHERE)
+#define FEDERATED_FROM " FROM "
+#define FEDERATED_FROM_LEN sizeof(FEDERATED_FROM)
+#define FEDERATED_PERCENT "%"
+#define FEDERATED_PERCENT_LEN sizeof(FEDERATED_PERCENT)
+#define FEDERATED_IS " IS "
+#define FEDERATED_IS_LEN sizeof(FEDERATED_IS)
+#define FEDERATED_NULL " NULL "
+#define FEDERATED_NULL_LEN sizeof(FEDERATED_NULL)
+#define FEDERATED_ISNULL " IS NULL "
+#define FEDERATED_ISNULL_LEN sizeof(FEDERATED_ISNULL)
+#define FEDERATED_LIKE " LIKE "
+#define FEDERATED_LIKE_LEN sizeof(FEDERATED_LIKE)
+#define FEDERATED_TRUNCATE "TRUNCATE "
+#define FEDERATED_TRUNCATE_LEN sizeof(FEDERATED_TRUNCATE)
+#define FEDERATED_DELETE "DELETE "
+#define FEDERATED_DELETE_LEN sizeof(FEDERATED_DELETE)
+#define FEDERATED_INSERT "INSERT INTO "
+#define FEDERATED_INSERT_LEN sizeof(FEDERATED_INSERT)
+#define FEDERATED_OPTIMIZE "OPTIMIZE TABLE "
+#define FEDERATED_OPTIMIZE_LEN sizeof(FEDERATED_OPTIMIZE)
+#define FEDERATED_REPAIR "REPAIR TABLE "
+#define FEDERATED_REPAIR_LEN sizeof(FEDERATED_REPAIR)
+#define FEDERATED_QUICK " QUICK"
+#define FEDERATED_QUICK_LEN sizeof(FEDERATED_QUICK)
+#define FEDERATED_EXTENDED " EXTENDED"
+#define FEDERATED_EXTENDED_LEN sizeof(FEDERATED_EXTENDED)
+#define FEDERATED_USE_FRM " USE_FRM"
+#define FEDERATED_USE_FRM_LEN sizeof(FEDERATED_USE_FRM)
+#define FEDERATED_LIMIT1 " LIMIT 1"
+#define FEDERATED_LIMIT1_LEN sizeof(FEDERATED_LIMIT1)
+#define FEDERATED_VALUES "VALUES "
+#define FEDERATED_VALUES_LEN sizeof(FEDERATED_VALUES)
+#define FEDERATED_UPDATE "UPDATE "
+#define FEDERATED_UPDATE_LEN sizeof(FEDERATED_UPDATE)
+#define FEDERATED_SET "SET "
+#define FEDERATED_SET_LEN sizeof(FEDERATED_SET)
+#define FEDERATED_AND " AND "
+#define FEDERATED_AND_LEN sizeof(FEDERATED_AND)
+#define FEDERATED_CONJUNCTION ") AND ("
+#define FEDERATED_CONJUNCTION_LEN sizeof(FEDERATED_CONJUNCTION)
+#define FEDERATED_OR " OR "
+#define FEDERATED_OR_LEN sizeof(FEDERATED_OR)
+#define FEDERATED_NOT " NOT "
+#define FEDERATED_NOT_LEN sizeof(FEDERATED_NOT)
+#define FEDERATED_STAR "* "
+#define FEDERATED_STAR_LEN sizeof(FEDERATED_STAR)
+#define FEDERATED_SPACE " "
+#define FEDERATED_SPACE_LEN sizeof(FEDERATED_SPACE)
+#define FEDERATED_SQUOTE "'"
+#define FEDERATED_SQUOTE_LEN sizeof(FEDERATED_SQUOTE)
+#define FEDERATED_COMMA ", "
+#define FEDERATED_COMMA_LEN sizeof(FEDERATED_COMMA)
+#define FEDERATED_BTICK "`" 
+#define FEDERATED_BTICK_LEN sizeof(FEDERATED_BTICK)
+#define FEDERATED_OPENPAREN " ("
+#define FEDERATED_OPENPAREN_LEN sizeof(FEDERATED_OPENPAREN)
+#define FEDERATED_CLOSEPAREN ") "
+#define FEDERATED_CLOSEPAREN_LEN sizeof(FEDERATED_CLOSEPAREN)
+#define FEDERATED_NE " != "
+#define FEDERATED_NE_LEN sizeof(FEDERATED_NE)
+#define FEDERATED_GT " > "
+#define FEDERATED_GT_LEN sizeof(FEDERATED_GT)
+#define FEDERATED_LT " < "
+#define FEDERATED_LT_LEN sizeof(FEDERATED_LT)
+#define FEDERATED_LE " <= "
+#define FEDERATED_LE_LEN sizeof(FEDERATED_LE)
+#define FEDERATED_GE " >= "
+#define FEDERATED_GE_LEN sizeof(FEDERATED_GE)
+#define FEDERATED_EQ " = "
+#define FEDERATED_EQ_LEN sizeof(FEDERATED_EQ)
+#define FEDERATED_FALSE " 1=0"
+#define FEDERATED_FALSE_LEN sizeof(FEDERATED_FALSE)
 
 /*
   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 {
-  char *table_name;
-  char *table_base_name;
-  /* 
+  /*
     the primary select query to be used in rnd_init
   */
   char *select_query;
@@ -47,11 +126,12 @@ typedef struct st_federated_share {
   char *username;
   char *password;
   char *database;
+  char *table_name;
   char *table;
   char *socket;
   char *sport;
-  int port;
-  uint table_name_length,table_base_name_length,use_count;
+  ushort port;
+  uint table_name_length, use_count;
   pthread_mutex_t mutex;
   THR_LOCK lock;
 } FEDERATED_SHARE;
@@ -63,8 +143,8 @@ class ha_federated: public handler
 {
   THR_LOCK_DATA lock;      /* MySQL lock */
   FEDERATED_SHARE *share;    /* Shared lock info */
-  MYSQL *mysql;
-  MYSQL_RES *result;
+  MYSQL *mysql; /* MySQL connection */
+  MYSQL_RES *stored_result;
   bool scan_flag;
   uint ref_length;
   uint fetch_num; // stores the fetch num
@@ -77,14 +157,12 @@ private:
   */
   uint convert_row_to_internal_format(byte *buf, MYSQL_ROW row);
   bool create_where_from_key(String *to, KEY *key_info, 
-                             const byte *key, uint key_length); 
+                             const key_range *start_key,
+                             const key_range *end_key,
+                             bool records_in_range);
 
 public:
-  ha_federated(TABLE *table): handler(table),
-    mysql(0), result(0), scan_flag(0), 
-    ref_length(sizeof(MYSQL_ROW_OFFSET)), current_position(0)
-  {
-  }
+  ha_federated(TABLE *table_arg);
   ~ha_federated()
   {
   }
@@ -94,20 +172,20 @@ public:
     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
-    Serg: Double check these (Brian)
-    // FIX add blob support
   */
   ulong table_flags() const
   {
-    return (HA_TABLE_SCAN_ON_INDEX | HA_NOT_EXACT_COUNT |
-            HA_PRIMARY_KEY_IN_READ_INDEX | HA_FILE_BASED |
-            HA_AUTO_PART_KEY | HA_CAN_INDEX_BLOBS);
+    /* fix server to be able to get remote server table flags */
+    return (HA_NOT_EXACT_COUNT |
+            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);
   }
   /*
     This is a bitmap of flags that says how the storage engine
@@ -119,29 +197,45 @@ public:
     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);
-    // return (HA_READ_NEXT | HA_ONLY_WHOLE_INDEX);
+    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 1024; }
-  uint max_supported_key_length()    const { return 1024; }
+  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 "
   */
-  virtual double scan_time()
+  double scan_time()
   {
-    DBUG_PRINT("ha_federated::scan_time",
-               ("rows %d", records)); return (double)(records*2); 
+    DBUG_PRINT("info",
+               ("records %d", records));
+    return (double)(records*1000); 
   }
   /*
     The next method will never be called if you do not implement indexes.
   */
-  virtual double read_time(uint index, uint ranges, ha_rows rows) 
-  { return (double) rows /  20.0+1; }
+  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.
 
@@ -151,16 +245,20 @@ public:
   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 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,
+  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,
+  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_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).
@@ -172,13 +270,18 @@ public:
   int rnd_init(bool scan);                                      //required
   int rnd_end();
   int rnd_next(byte *buf);                                      //required
-  int rnd_pos(byte * buf, byte *pos);                           //required
+  int rnd_pos(byte *buf, byte *pos);                            //required
   void position(const byte *record);                            //required
   void info(uint);                                              //required
 
+  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);
 
   THR_LOCK_DATA **store_lock(THD *thd, THR_LOCK_DATA **to,
                              enum thr_lock_type lock_type);     //required
diff --git a/sql/ha_heap.cc b/sql/ha_heap.cc
index 6e609a94be3..92a5fe0ea09 100644
--- a/sql/ha_heap.cc
+++ b/sql/ha_heap.cc
@@ -23,9 +23,33 @@
 #include <myisampack.h>
 #include "ha_heap.h"
 
+static handlerton heap_hton= {
+  "MEMORY",
+  0,       /* slot */
+  0,       /* savepoint size. */
+  0,       /* close_connection */
+  0,       /* savepoint */
+  0,       /* rollback to savepoint */
+  0,       /* release savepoint */
+  0,       /* commit */
+  0,       /* rollback */
+  0,       /* prepare */
+  0,       /* recover */
+  0,       /* commit_by_xid */
+  0,       /* rollback_by_xid */
+  HTON_NO_FLAGS
+};
+
 /*****************************************************************************
 ** HEAP tables
 *****************************************************************************/
+
+ha_heap::ha_heap(TABLE *table_arg)
+  :handler(&heap_hton, table_arg), file(0), records_changed(0),
+  key_stats_ok(0)
+{}
+
+
 static const char *ha_heap_exts[] = {
   NullS
 };
diff --git a/sql/ha_heap.h b/sql/ha_heap.h
index 7a97c727049..f7368436456 100644
--- a/sql/ha_heap.h
+++ b/sql/ha_heap.h
@@ -31,8 +31,7 @@ class ha_heap: public handler
   uint    records_changed;
   bool    key_stats_ok;
 public:
-  ha_heap(TABLE *table): handler(table), file(0), records_changed(0),
-      key_stats_ok(0) {}
+  ha_heap(TABLE *table);
   ~ha_heap() {}
   const char *table_type() const
   {
diff --git a/sql/ha_innodb.cc b/sql/ha_innodb.cc
index 1dfc64c9137..3f86399aacd 100644
--- a/sql/ha_innodb.cc
+++ b/sql/ha_innodb.cc
@@ -215,7 +215,14 @@ static handlerton innobase_hton = {
   innobase_xa_prepare,		/* prepare */
   innobase_xa_recover,		/* recover */
   innobase_commit_by_xid,	/* commit_by_xid */
-  innobase_rollback_by_xid	/* rollback_by_xid */
+  innobase_rollback_by_xid,     /* rollback_by_xid */
+  /*
+    For now when one opens a cursor, MySQL does not create an own
+    InnoDB consistent read view for it, and uses the view of the
+    currently active transaction. Therefore, cursors can not
+    survive COMMIT or ROLLBACK statements, which free this view.
+  */
+  HTON_CLOSE_CURSORS_AT_COMMIT
 };
 
 /*********************************************************************
@@ -765,6 +772,24 @@ check_trx_exists(
 	return(trx);
 }
 
+
+/*************************************************************************
+Construct ha_innobase handler. */
+
+ha_innobase::ha_innobase(TABLE *table_arg)
+  :handler(&innobase_hton, table_arg),
+  int_table_flags(HA_REC_NOT_IN_SEQ |
+                  HA_NULL_IN_KEY |
+                  HA_CAN_INDEX_BLOBS |
+                  HA_CAN_SQL_HANDLER |
+                  HA_NOT_EXACT_COUNT |
+                  HA_PRIMARY_KEY_IN_READ_INDEX |
+                  HA_TABLE_SCAN_ON_INDEX),
+  last_dup_key((uint) -1),
+  start_of_scan(0),
+  num_write_row(0)
+{}
+
 /*************************************************************************
 Updates the user_thd field in a handle and also allocates a new InnoDB
 transaction handle if needed, and updates the transaction fields in the
@@ -5486,7 +5511,7 @@ ha_innobase::update_table_comment(
 	external_lock(). To be safe, update the thd of the current table
 	handle. */
 
-	if(length > 64000 - 3) {
+	if (length > 64000 - 3) {
 		return((char*)comment); /* string too long */
 	}
 
@@ -5524,7 +5549,7 @@ ha_innobase::update_table_comment(
 
 		if (str) {
 			char* pos	= str + length;
-			if(length) {
+			if (length) {
 				memcpy(str, comment, length);
 				*pos++ = ';';
 				*pos++ = ' ';
@@ -5582,7 +5607,7 @@ ha_innobase::get_foreign_key_create_info(void)
 		flen = ftell(file);
 		if (flen < 0) {
 			flen = 0;
-		} else if(flen > 64000 - 1) {
+		} else if (flen > 64000 - 1) {
 			flen = 64000 - 1;
 		}
 
diff --git a/sql/ha_innodb.h b/sql/ha_innodb.h
index 150eae63730..acfe8b950eb 100644
--- a/sql/ha_innodb.h
+++ b/sql/ha_innodb.h
@@ -81,19 +81,7 @@ class ha_innobase: public handler
 
 	/* Init values for the class: */
  public:
-  	ha_innobase(TABLE *table): handler(table),
-	  int_table_flags(HA_REC_NOT_IN_SEQ |
-			  HA_NULL_IN_KEY |
-			  HA_CAN_INDEX_BLOBS |
-			  HA_CAN_SQL_HANDLER |
-			  HA_NOT_EXACT_COUNT |
-			  HA_PRIMARY_KEY_IN_READ_INDEX |
-			  HA_TABLE_SCAN_ON_INDEX),
-	  last_dup_key((uint) -1),
-	  start_of_scan(0),
-	  num_write_row(0)
-  	{
-  	}
+  	ha_innobase(TABLE *table_arg);
   	~ha_innobase() {}
 	/*
 	  Get the row type from the storage engine.  If this method returns
diff --git a/sql/ha_myisam.cc b/sql/ha_myisam.cc
index 27023ba4c64..2595a1cca26 100644
--- a/sql/ha_myisam.cc
+++ b/sql/ha_myisam.cc
@@ -44,6 +44,29 @@ TYPELIB myisam_recover_typelib= {array_elements(myisam_recover_names)-1,"",
 ** MyISAM tables
 *****************************************************************************/
 
+/* MyISAM handlerton */
+
+static handlerton myisam_hton= {
+  "MyISAM",
+  0,       /* slot */
+  0,       /* savepoint size. */
+  0,       /* close_connection */
+  0,       /* savepoint */
+  0,       /* rollback to savepoint */
+  0,       /* release savepoint */
+  0,       /* commit */
+  0,       /* rollback */
+  0,       /* prepare */
+  0,       /* recover */
+  0,       /* commit_by_xid */
+  0,       /* rollback_by_xid */
+  /*
+    MyISAM doesn't support transactions and doesn't have
+    transaction-dependent context: cursors can survive a commit.
+  */
+  HTON_NO_FLAGS
+};
+
 // collect errors printed by mi_check routines
 
 static void mi_check_print_msg(MI_CHECK *param,	const char* msg_type,
@@ -123,6 +146,17 @@ void mi_check_print_warning(MI_CHECK *param, const char *fmt,...)
 
 }
 
+
+ha_myisam::ha_myisam(TABLE *table_arg)
+  :handler(&myisam_hton, table_arg), file(0),
+  int_table_flags(HA_NULL_IN_KEY | HA_CAN_FULLTEXT | HA_CAN_SQL_HANDLER |
+                  HA_DUPP_POS | HA_CAN_INDEX_BLOBS | HA_AUTO_PART_KEY |
+                  HA_FILE_BASED | HA_CAN_GEOMETRY | HA_READ_RND_SAME |
+                  HA_CAN_INSERT_DELAYED | HA_CAN_BIT_FIELD),
+  can_enable_indexes(1)
+{}
+
+
 static const char *ha_myisam_exts[] = {
   ".MYI",
   ".MYD",
@@ -602,7 +636,7 @@ int ha_myisam::repair(THD *thd, MI_CHECK &param, bool optimize)
 	!(share->state.changed & STATE_NOT_OPTIMIZED_KEYS))))
   {
     ulonglong key_map= ((local_testflag & T_CREATE_MISSING_KEYS) ?
-			((ulonglong) 1L << share->base.keys)-1 :
+			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) &&
@@ -903,7 +937,7 @@ int ha_myisam::enable_indexes(uint mode)
 {
   int error;
 
-  if (file->s->state.key_map == set_bits(ulonglong, file->s->base.keys))
+  if (mi_is_all_keys_active(file->s->state.key_map, file->s->base.keys))
   {
     /* All indexes are enabled already. */
     return 0;
@@ -1002,8 +1036,8 @@ void ha_myisam::start_bulk_insert(ha_rows rows)
   if (! rows || (rows > MI_MIN_ROWS_TO_USE_WRITE_CACHE))
     mi_extra(file, HA_EXTRA_WRITE_CACHE, (void*) &size);
 
-  can_enable_indexes= (file->s->state.key_map ==
-                       set_bits(ulonglong, file->s->base.keys));
+  can_enable_indexes= mi_is_all_keys_active(file->s->state.key_map,
+                                            file->s->base.keys);
 
   if (!(specialflag & SPECIAL_SAFE_MODE))
   {
@@ -1256,7 +1290,7 @@ void ha_myisam::info(uint flag)
     share->db_options_in_use= info.options;
     block_size= myisam_block_size;
     share->keys_in_use.set_prefix(share->keys);
-    share->keys_in_use.intersect(info.key_map);
+    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)
diff --git a/sql/ha_myisam.h b/sql/ha_myisam.h
index bbd9721f8e2..ca684463311 100644
--- a/sql/ha_myisam.h
+++ b/sql/ha_myisam.h
@@ -43,13 +43,7 @@ class ha_myisam: public handler
   int repair(THD *thd, MI_CHECK &param, bool optimize);
 
  public:
-  ha_myisam(TABLE *table): handler(table), file(0),
-    int_table_flags(HA_NULL_IN_KEY | HA_CAN_FULLTEXT | HA_CAN_SQL_HANDLER |
-		    HA_DUPP_POS | HA_CAN_INDEX_BLOBS | HA_AUTO_PART_KEY |
-		    HA_FILE_BASED | HA_CAN_GEOMETRY | HA_READ_RND_SAME |
-                    HA_CAN_INSERT_DELAYED | HA_CAN_BIT_FIELD),
-    can_enable_indexes(1)
-  {}
+  ha_myisam(TABLE *table_arg);
   ~ha_myisam() {}
   const char *table_type() const { return "MyISAM"; }
   const char *index_type(uint key_number);
diff --git a/sql/ha_myisammrg.cc b/sql/ha_myisammrg.cc
index 794f1c62dd7..ae17b60b7e1 100644
--- a/sql/ha_myisammrg.cc
+++ b/sql/ha_myisammrg.cc
@@ -32,6 +32,30 @@
 ** MyISAM MERGE tables
 *****************************************************************************/
 
+/* MyISAM MERGE handlerton */
+
+static handlerton myisammrg_hton= {
+  "MRG_MyISAM",
+  0,       /* slot */
+  0,       /* savepoint size. */
+  0,       /* close_connection */
+  0,       /* savepoint */
+  0,       /* rollback to savepoint */
+  0,       /* release savepoint */
+  0,       /* commit */
+  0,       /* rollback */
+  0,       /* prepare */
+  0,       /* recover */
+  0,       /* commit_by_xid */
+  0,       /* rollback_by_xid */
+  HTON_NO_FLAGS
+};
+
+
+ha_myisammrg::ha_myisammrg(TABLE *table_arg)
+  :handler(&myisammrg_hton, table_arg), file(0)
+{}
+
 static const char *ha_myisammrg_exts[] = {
   ".MRG",
   NullS
diff --git a/sql/ha_myisammrg.h b/sql/ha_myisammrg.h
index 7348096b695..c762b7c286e 100644
--- a/sql/ha_myisammrg.h
+++ b/sql/ha_myisammrg.h
@@ -28,7 +28,7 @@ class ha_myisammrg: public handler
   MYRG_INFO *file;
 
  public:
-  ha_myisammrg(TABLE *table): handler(table), file(0) {}
+  ha_myisammrg(TABLE *table_arg);
   ~ha_myisammrg() {}
   const char *table_type() const { return "MRG_MyISAM"; }
   const char **bas_ext() const;
diff --git a/sql/ha_ndbcluster.cc b/sql/ha_ndbcluster.cc
index 1f830342b73..c3882436149 100644
--- a/sql/ha_ndbcluster.cc
+++ b/sql/ha_ndbcluster.cc
@@ -63,7 +63,8 @@ static handlerton ndbcluster_hton = {
   NULL, /* prepare */
   NULL, /* recover */
   NULL, /* commit_by_xid */
-  NULL  /* rollback_by_xid */
+  NULL, /* rollback_by_xid */
+  HTON_NO_FLAGS
 };
 
 #define NDB_HIDDEN_PRIMARY_KEY_LENGTH 8
@@ -4091,7 +4092,7 @@ ulonglong ha_ndbcluster::get_auto_increment()
  */
 
 ha_ndbcluster::ha_ndbcluster(TABLE *table_arg):
-  handler(table_arg),
+  handler(&ndbcluster_hton, table_arg),
   m_active_trans(NULL),
   m_active_cursor(NULL),
   m_table(NULL),
@@ -5372,7 +5373,7 @@ ndb_get_table_statistics(Ndb* ndb, const char * table,
     Uint64 sum_commits= 0;
     Uint64 sum_row_size= 0;
     Uint64 sum_mem= 0;
-    while((check= pOp->nextResult(TRUE, TRUE)) == 0)
+    while ((check= pOp->nextResult(TRUE, TRUE)) == 0)
     {
       sum_rows+= rows;
       sum_commits+= commits;
@@ -5400,7 +5401,7 @@ ndb_get_table_statistics(Ndb* ndb, const char * table,
                         sum_mem, count));
 
     DBUG_RETURN(0);
-  } while(0);
+  } while (0);
 
   if (pTrans)
     ndb->closeTransaction(pTrans);
diff --git a/sql/handler.cc b/sql/handler.cc
index 512363f71d7..c2a138e7013 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -217,19 +217,11 @@ handler *get_new_handler(TABLE *table, enum db_type db_type)
 #ifndef NO_HASH
   case DB_TYPE_HASH:
     file= new ha_hash(table);
-#endif
-#ifdef HAVE_ISAM
-  case DB_TYPE_MRG_ISAM:
-    file= new ha_isammrg(table);
     break;
-  case DB_TYPE_ISAM:
-    file= new ha_isam(table);
-    break;
-#else
+#endif
   case DB_TYPE_MRG_ISAM:
     file= new ha_myisammrg(table);
     break;
-#endif
 #ifdef HAVE_BERKELEY_DB
   case DB_TYPE_BERKELEY_DB:
     file= new ha_berkeley(table);
@@ -697,6 +689,11 @@ int ha_commit_trans(THD *thd, bool all)
       DBUG_RETURN(1);
     }
     DBUG_EXECUTE_IF("crash_commit_before", abort(););
+
+    /* Close all cursors that can not survive COMMIT */
+    if (is_real_trans)                          /* not a statement commit */
+      thd->stmt_map.close_transient_cursors();
+
     if (!trans->no_2pc && trans->nht > 1)
     {
       for (; *ht && !error; ht++)
@@ -798,6 +795,10 @@ int ha_rollback_trans(THD *thd, bool all)
 #ifdef USING_TRANSACTIONS
   if (trans->nht)
   {
+    /* Close all cursors that can not survive ROLLBACK */
+    if (is_real_trans)                          /* not a statement commit */
+      thd->stmt_map.close_transient_cursors();
+
     for (handlerton **ht=trans->ht; *ht; ht++)
     {
       int err;
@@ -2602,6 +2603,7 @@ TYPELIB *ha_known_exts(void)
 
     known_extensions_id= mysys_usage_id;
     found_exts.push_back((char*) triggers_file_ext);
+    found_exts.push_back((char*) trigname_file_ext);
     for (types= sys_table_types; types->type; types++)
     {
       if (*types->value == SHOW_OPTION_YES)
diff --git a/sql/handler.h b/sql/handler.h
index d06ae062fb3..39da4f65272 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -355,8 +355,13 @@ typedef struct
    int  (*recover)(XID *xid_list, uint len);
    int  (*commit_by_xid)(XID *xid);
    int  (*rollback_by_xid)(XID *xid);
+   uint32 flags;                                /* global handler flags */
 } handlerton;
 
+/* Possible flags of a handlerton */
+#define HTON_NO_FLAGS 0
+#define HTON_CLOSE_CURSORS_AT_COMMIT 1
+
 typedef struct st_thd_trans
 {
   /* number of entries in the ht[] */
@@ -683,6 +688,7 @@ class handler :public Sql_alloc
 private:
   virtual int reset() { return extra(HA_EXTRA_RESET); }
 public:
+  const handlerton *ht;                 /* storage engine of this handler */
   byte *ref;				/* Pointer to current row */
   byte *dupp_ref;			/* Pointer to dupp row */
   ulonglong data_file_length;		/* Length off data file */
@@ -726,7 +732,8 @@ public:
   MY_BITMAP *read_set;
   MY_BITMAP *write_set;
 
-  handler(TABLE *table_arg) :table(table_arg),
+  handler(const handlerton *ht_arg, TABLE *table_arg) :table(table_arg),
+    ht(ht_arg),
     ref(0), data_file_length(0), max_data_file_length(0), index_file_length(0),
     delete_length(0), auto_increment_value(0),
     records(0), deleted(0), mean_rec_length(0),
diff --git a/sql/item.cc b/sql/item.cc
index c04d7c346d1..593708688f1 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -638,6 +638,38 @@ Item *Item_num::safe_charset_converter(CHARSET_INFO *tocs)
 }
 
 
+Item *Item_static_int_func::safe_charset_converter(CHARSET_INFO *tocs)
+{
+  Item_string *conv;
+  char buf[64];
+  String *s, tmp(buf, sizeof(buf), &my_charset_bin);
+  s= val_str(&tmp);
+  if ((conv= new Item_static_string_func(func_name, s->ptr(), s->length(),
+                                         s->charset())))
+  {
+    conv->str_value.copy();
+    conv->str_value.mark_as_const();
+  }
+  return conv;
+}
+
+
+Item *Item_static_float_func::safe_charset_converter(CHARSET_INFO *tocs)
+{
+  Item_string *conv;
+  char buf[64];
+  String *s, tmp(buf, sizeof(buf), &my_charset_bin);
+  s= val_str(&tmp);
+  if ((conv= new Item_static_string_func(func_name, s->ptr(), s->length(),
+                                         s->charset())))
+  {
+    conv->str_value.copy();
+    conv->str_value.mark_as_const();
+  }
+  return conv;
+}
+
+
 Item *Item_string::safe_charset_converter(CHARSET_INFO *tocs)
 {
   Item_string *conv;
@@ -663,6 +695,33 @@ Item *Item_string::safe_charset_converter(CHARSET_INFO *tocs)
 }
 
 
+Item *Item_static_string_func::safe_charset_converter(CHARSET_INFO *tocs)
+{
+  Item_string *conv;
+  uint conv_errors;
+  String tmp, cstr, *ostr= val_str(&tmp);
+  cstr.copy(ostr->ptr(), ostr->length(), ostr->charset(), tocs, &conv_errors);
+  if (conv_errors ||
+      !(conv= new Item_static_string_func(func_name,
+                                          cstr.ptr(), cstr.length(),
+                                          cstr.charset(),
+                                          collation.derivation)))
+  {
+    /*
+      Safe conversion is not possible (or EOM).
+      We could not convert a string into the requested character set
+      without data loss. The target charset does not cover all the
+      characters from the string. Operation cannot be done correctly.
+    */
+    return NULL;
+  }
+  conv->str_value.copy();
+  /* Ensure that no one is going to change the result string */
+  conv->str_value.mark_as_const();
+  return conv;
+}
+
+
 bool Item_string::eq(const Item *item, bool binary_cmp) const
 {
   if (type() == item->type() && item->basic_const_item())
diff --git a/sql/item.h b/sql/item.h
index a10934e40fb..5a1cf193806 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -1132,6 +1132,7 @@ public:
   Item_static_int_func(const char *str_arg, longlong i, uint length)
     :Item_int(NullS, i, length), func_name(str_arg)
   {}
+  Item *safe_charset_converter(CHARSET_INFO *tocs);
   void print(String *str) { str->append(func_name); }
 };
 
@@ -1242,6 +1243,7 @@ public:
     :Item_float(NullS, val_arg, decimal_par, length), func_name(str)
   {}
   void print(String *str) { str->append(func_name); }
+  Item *safe_charset_converter(CHARSET_INFO *tocs);
 };
 
 
@@ -1314,6 +1316,7 @@ public:
                           Derivation dv= DERIVATION_COERCIBLE)
     :Item_string(NullS, str, length, cs, dv), func_name(name_par)
   {}
+  Item *safe_charset_converter(CHARSET_INFO *tocs);
   void print(String *str) { str->append(func_name); }
 };
 
@@ -1778,7 +1781,7 @@ public:
 */
 enum trg_action_time_type
 {
-  TRG_ACTION_BEFORE= 0, TRG_ACTION_AFTER= 1
+  TRG_ACTION_BEFORE= 0, TRG_ACTION_AFTER= 1, TRG_ACTION_MAX
 };
 
 /*
@@ -1786,7 +1789,7 @@ enum trg_action_time_type
 */
 enum trg_event_type
 {
-  TRG_EVENT_INSERT= 0 , TRG_EVENT_UPDATE= 1, TRG_EVENT_DELETE= 2
+  TRG_EVENT_INSERT= 0 , TRG_EVENT_UPDATE= 1, TRG_EVENT_DELETE= 2, TRG_EVENT_MAX
 };
 
 class Table_triggers_list;
diff --git a/sql/item_cmpfunc.cc b/sql/item_cmpfunc.cc
index 5ed857319be..d430d0d3c23 100644
--- a/sql/item_cmpfunc.cc
+++ b/sql/item_cmpfunc.cc
@@ -237,32 +237,35 @@ void Item_bool_func2::fix_length_and_dec()
     set_cmp_func();
     return;
   }
-    
-  Item *real_item= args[0]->real_item();
-  if (real_item->type() == FIELD_ITEM)
+
+  if (!thd->is_context_analysis_only())
   {
-    Field *field= ((Item_field*) real_item)->field;
-    if (field->can_be_compared_as_longlong())
+    Item *real_item= args[0]->real_item();
+    if (real_item->type() == FIELD_ITEM)
     {
-      if (convert_constant_item(thd, field,&args[1]))
+      Field *field=((Item_field*) real_item)->field;
+      if (field->can_be_compared_as_longlong())
       {
-	cmp.set_cmp_func(this, tmp_arg, tmp_arg+1,
-			 INT_RESULT);		// Works for all types.
-	return;
+        if (convert_constant_item(thd, field,&args[1]))
+        {
+          cmp.set_cmp_func(this, tmp_arg, tmp_arg+1,
+                           INT_RESULT);		// Works for all types.
+          return;
+        }
       }
     }
-  }
-  real_item= args[1]->real_item();
-  if (real_item->type() == FIELD_ITEM)
-  {
-    Field *field= ((Item_field*) real_item)->field;
-    if (field->can_be_compared_as_longlong())
+    real_item= args[1]->real_item();
+    if (real_item->type() == FIELD_ITEM /* && !real_item->const_item() */)
     {
-      if (convert_constant_item(thd, field,&args[0]))
+      Field *field=((Item_field*) real_item)->field;
+      if (field->can_be_compared_as_longlong())
       {
-	cmp.set_cmp_func(this, tmp_arg, tmp_arg+1,
-			 INT_RESULT); // Works for all types.
-	return;
+        if (convert_constant_item(thd, field,&args[0]))
+        {
+          cmp.set_cmp_func(this, tmp_arg, tmp_arg+1,
+                           INT_RESULT); // Works for all types.
+          return;
+        }
       }
     }
   }
@@ -990,7 +993,8 @@ void Item_func_between::fix_length_and_dec()
   if (args[0]->type() == FIELD_ITEM)
   {
     Field *field=((Item_field*) args[0])->field;
-    if (field->can_be_compared_as_longlong())
+    if (!thd->is_context_analysis_only() &&
+        field->can_be_compared_as_longlong())
     {
       /*
         The following can't be recoded with || as convert_constant_item
@@ -2180,7 +2184,13 @@ void Item_func_in::fix_length_and_dec()
     return;
 
   for (arg=args+1, arg_end=args+arg_count; arg != arg_end ; arg++)
-    const_itm&= arg[0]->const_item();
+  {
+    if (!arg[0]->const_item())
+    {
+      const_itm= 0;
+      break;
+    }
+  }
 
   /*
     Row item with NULLs inside can return NULL or FALSE => 
diff --git a/sql/item_create.cc b/sql/item_create.cc
index b9073a6c0b3..b7d8d50f9b3 100644
--- a/sql/item_create.cc
+++ b/sql/item_create.cc
@@ -299,16 +299,8 @@ Item *create_func_pow(Item* a, Item *b)
 
 Item *create_func_current_user()
 {
-  THD *thd=current_thd;
-  char buff[HOSTNAME_LENGTH+USERNAME_LENGTH+2];
-  uint length;
-
-  thd->lex->safe_to_cache_query= 0;
-  length= (uint) (strxmov(buff, thd->priv_user, "@", thd->priv_host, NullS) -
-		  buff);
-  return new Item_static_string_func("current_user()",
-                                     thd->memdup(buff, length), length,
-                                     system_charset_info);
+  current_thd->lex->safe_to_cache_query= 0;
+  return new Item_func_user(TRUE);
 }
 
 Item *create_func_radians(Item *a)
diff --git a/sql/item_func.cc b/sql/item_func.cc
index 0817282b673..c3bdb11418f 100644
--- a/sql/item_func.cc
+++ b/sql/item_func.cc
@@ -4217,6 +4217,36 @@ void Item_user_var_as_out_param::print(String *str)
 }
 
 
+Item_func_get_system_var::
+Item_func_get_system_var(sys_var *var_arg, enum_var_type var_type_arg,
+                       LEX_STRING *component_arg, const char *name_arg,
+                       size_t name_len_arg)
+  :var(var_arg), var_type(var_type_arg), component(*component_arg)
+{
+  /* set_name() will allocate the name */
+  set_name(name_arg, name_len_arg, system_charset_info);
+}
+
+
+bool
+Item_func_get_system_var::fix_fields(THD *thd, Item **ref)
+{
+  Item *item= var->item(thd, var_type, &component);
+  DBUG_ENTER("Item_func_get_system_var::fix_fields");
+  /*
+    Evaluate the system variable and substitute the result (a basic constant)
+    instead of this item. If the variable can not be evaluated,
+    the error is reported in sys_var::item().
+  */
+  if (item == 0)
+    DBUG_RETURN(1);                             // Impossible
+  item->set_name(name, 0, system_charset_info); // don't allocate a new name
+  thd->change_item_tree(ref, item);
+
+  DBUG_RETURN(0);
+}
+
+
 longlong Item_func_inet_aton::val_int()
 {
   DBUG_ASSERT(fixed == 1);
@@ -4563,22 +4593,21 @@ longlong Item_func_bit_xor::val_int()
     0	error
     #	constant item
 */
-  
+
 
 Item *get_system_var(THD *thd, enum_var_type var_type, LEX_STRING name,
 		     LEX_STRING component)
 {
+  sys_var *var;
+  char buff[MAX_SYS_VAR_LENGTH*2+4+8], *pos;
+  LEX_STRING *base_name, *component_name;
+
   if (component.str == 0 &&
       !my_strcasecmp(system_charset_info, name.str, "VERSION"))
     return new Item_string("@@VERSION", server_version,
 			   (uint) strlen(server_version),
 			   system_charset_info, DERIVATION_SYSCONST);
 
-  Item *item;
-  sys_var *var;
-  char buff[MAX_SYS_VAR_LENGTH*2+4+8], *pos;
-  LEX_STRING *base_name, *component_name;
-
   if (component.str)
   {
     base_name= &component;
@@ -4600,9 +4629,8 @@ Item *get_system_var(THD *thd, enum_var_type var_type, LEX_STRING name,
       return 0;
     }
   }
-  if (!(item=var->item(thd, var_type, component_name)))
-    return 0;					// Impossible
   thd->lex->uncacheable(UNCACHEABLE_SIDEEFFECT);
+
   buff[0]='@';
   buff[1]='@';
   pos=buff+2;
@@ -4623,28 +4651,8 @@ Item *get_system_var(THD *thd, enum_var_type var_type, LEX_STRING name,
   memcpy(pos, base_name->str, base_name->length);
   pos+= base_name->length;
 
-  // set_name() will allocate the name
-  item->set_name(buff,(uint) (pos-buff), system_charset_info);
-  return item;
-}
-
-
-Item *get_system_var(THD *thd, enum_var_type var_type, const char *var_name,
-		     uint length, const char *item_name)
-{
-  Item *item;
-  sys_var *var;
-  LEX_STRING null_lex_string;
-
-  null_lex_string.str= 0;
-
-  var= find_sys_var(var_name, length);
-  DBUG_ASSERT(var != 0);
-  if (!(item=var->item(thd, var_type, &null_lex_string)))
-    return 0;						// Impossible
-  thd->lex->uncacheable(UNCACHEABLE_SIDEEFFECT);
-  item->set_name(item_name, 0, system_charset_info);  // Will use original name
-  return item;
+  return new Item_func_get_system_var(var, var_type, component_name,
+                                      buff, pos - buff);
 }
 
 
diff --git a/sql/item_func.h b/sql/item_func.h
index 3ca37b1961f..43a85e6aa0b 100644
--- a/sql/item_func.h
+++ b/sql/item_func.h
@@ -1199,6 +1199,31 @@ public:
 };
 
 
+/* A system variable */
+
+class Item_func_get_system_var :public Item_func
+{
+  sys_var *var;
+  enum_var_type var_type;
+  LEX_STRING component;
+public:
+  Item_func_get_system_var(sys_var *var_arg, enum_var_type var_type_arg,
+                           LEX_STRING *component_arg, const char *name_arg,
+                           size_t name_len_arg);
+  bool fix_fields(THD *thd, Item **ref);
+  /*
+    Stubs for pure virtual methods. Should never be called: this
+    item is always substituted with a constant in fix_fields().
+  */
+  double val_real()         { DBUG_ASSERT(0); return 0.0; }
+  longlong val_int()        { DBUG_ASSERT(0); return 0; }
+  String* val_str(String*)  { DBUG_ASSERT(0); return 0; }
+  void fix_length_and_dec() { DBUG_ASSERT(0); }
+  /* TODO: fix to support views */
+  const char *func_name() const { return "get_system_var"; }
+};
+
+
 class Item_func_inet_aton : public Item_int_func
 {
 public:
diff --git a/sql/item_strfunc.cc b/sql/item_strfunc.cc
index 2fd0f25e699..6d519c73b13 100644
--- a/sql/item_strfunc.cc
+++ b/sql/item_strfunc.cc
@@ -1552,9 +1552,11 @@ Item *Item_func_sysconst::safe_charset_converter(CHARSET_INFO *tocs)
   uint conv_errors;
   String tmp, cstr, *ostr= val_str(&tmp);
   cstr.copy(ostr->ptr(), ostr->length(), ostr->charset(), tocs, &conv_errors);
-  if (conv_errors || !(conv= new Item_string(cstr.ptr(), cstr.length(),
-                                             cstr.charset(),
-                                             collation.derivation)))
+  if (conv_errors ||
+      !(conv= new Item_static_string_func(fully_qualified_func_name(),
+                                          cstr.ptr(), cstr.length(),
+                                          cstr.charset(),
+                                          collation.derivation)))
   {
     return NULL;
   }
@@ -1584,13 +1586,24 @@ String *Item_func_user::val_str(String *str)
   DBUG_ASSERT(fixed == 1);
   THD          *thd=current_thd;
   CHARSET_INFO *cs= system_charset_info;
-  const char   *host= thd->host_or_ip;
+  const char   *host, *user;
   uint		res_length;
 
+  if (is_current)
+  {
+    user= thd->priv_user;
+    host= thd->priv_host;
+  }
+  else
+  {
+    user= thd->user;
+    host= thd->host_or_ip;
+  }
+
   // For system threads (e.g. replication SQL thread) user may be empty
-  if (!thd->user)
+  if (!user)
     return &my_empty_string;
-  res_length= (strlen(thd->user)+strlen(host)+2) * cs->mbmaxlen;
+  res_length= (strlen(user)+strlen(host)+2) * cs->mbmaxlen;
 
   if (str->alloc(res_length))
   {
@@ -1598,12 +1611,13 @@ String *Item_func_user::val_str(String *str)
     return 0;
   }
   res_length=cs->cset->snprintf(cs, (char*)str->ptr(), res_length, "%s@%s",
-			  thd->user, host);
+			        user, host);
   str->length(res_length);
   str->set_charset(cs);
   return str;
 }
 
+
 void Item_func_soundex::fix_length_and_dec()
 {
   collation.set(args[0]->collation);
@@ -2104,7 +2118,7 @@ String *Item_func_rpad::val_str(String *str)
 			func_name(), current_thd->variables.max_allowed_packet);
     goto err;
   }
-  if(args[2]->null_value || !pad_char_length)
+  if (args[2]->null_value || !pad_char_length)
     goto err;
   res_byte_length= res->length();	/* Must be done before alloc_buffer */
   if (!(res= alloc_buffer(res,str,&tmp_value,byte_count)))
diff --git a/sql/item_strfunc.h b/sql/item_strfunc.h
index d85210984d9..fa098849a43 100644
--- a/sql/item_strfunc.h
+++ b/sql/item_strfunc.h
@@ -357,8 +357,15 @@ public:
   Item_func_sysconst()
   { collation.set(system_charset_info,DERIVATION_SYSCONST); }
   Item *safe_charset_converter(CHARSET_INFO *tocs);
+  /*
+    Used to create correct Item name in new converted item in
+    safe_charset_converter, return string representation of this function
+    call
+  */
+  virtual const char *fully_qualified_func_name() const = 0;
 };
 
+
 class Item_func_database :public Item_func_sysconst
 {
 public:
@@ -370,18 +377,27 @@ public:
     maybe_null=1;
   }
   const char *func_name() const { return "database"; }
+  const char *fully_qualified_func_name() const { return "database()"; }
 };
 
+
 class Item_func_user :public Item_func_sysconst
 {
+  bool is_current;
+
 public:
-  Item_func_user() :Item_func_sysconst() {}
+  Item_func_user(bool is_current_arg)
+    :Item_func_sysconst(), is_current(is_current_arg) {}
   String *val_str(String *);
-  void fix_length_and_dec() 
-  { 
-    max_length= (USERNAME_LENGTH+HOSTNAME_LENGTH+1)*system_charset_info->mbmaxlen;
+  void fix_length_and_dec()
+  {
+    max_length= ((USERNAME_LENGTH + HOSTNAME_LENGTH + 1) *
+                 system_charset_info->mbmaxlen);
   }
-  const char *func_name() const { return "user"; }
+  const char *func_name() const
+    { return is_current ? "current_user" : "user"; }
+  const char *fully_qualified_func_name() const
+    { return is_current ? "current_user()" : "user()"; }
 };
 
 
diff --git a/sql/lex.h b/sql/lex.h
index 59ba6a8e15b..c0e91527f45 100644
--- a/sql/lex.h
+++ b/sql/lex.h
@@ -510,6 +510,7 @@ static SYMBOL symbols[] = {
   { "TRAILING",		SYM(TRAILING)},
   { "TRANSACTION",	SYM(TRANSACTION_SYM)},
   { "TRIGGER",          SYM(TRIGGER_SYM)},
+  { "TRIGGERS",         SYM(TRIGGERS_SYM)},
   { "TRUE",		SYM(TRUE_SYM)},
   { "TRUNCATE",		SYM(TRUNCATE_SYM)},
   { "TYPE",		SYM(TYPE_SYM)},
diff --git a/sql/lock.cc b/sql/lock.cc
index a666c6da17a..dff863ccf56 100644
--- a/sql/lock.cc
+++ b/sql/lock.cc
@@ -103,6 +103,10 @@ MYSQL_LOCK *mysql_lock_tables(THD *thd, TABLE **tables, uint count, uint flags)
 {
   MYSQL_LOCK *sql_lock;
   TABLE *write_lock_used;
+  int rc;
+  /* Map the return value of thr_lock to an error from errmsg.txt */
+  const static int thr_lock_errno_to_mysql[]=
+  { 0, 1, ER_LOCK_WAIT_TIMEOUT, ER_LOCK_DEADLOCK };
   DBUG_ENTER("mysql_lock_tables");
 
   for (;;)
@@ -135,15 +139,24 @@ MYSQL_LOCK *mysql_lock_tables(THD *thd, TABLE **tables, uint count, uint flags)
     {
       my_free((gptr) sql_lock,MYF(0));
       sql_lock=0;
-      thd->proc_info=0;
       break;
     }
     thd->proc_info="Table lock";
     thd->locked=1;
-    if (thr_multi_lock(sql_lock->locks,sql_lock->lock_count))
+    rc= thr_lock_errno_to_mysql[(int) thr_multi_lock(sql_lock->locks,
+                                                     sql_lock->lock_count,
+                                                     thd->lock_id)];
+    if (rc > 1)                                 /* a timeout or a deadlock */
+    {
+      my_error(rc, MYF(0));
+      my_free((gptr) sql_lock,MYF(0));
+      sql_lock= 0;
+      break;
+    }
+    else if (rc == 1)                           /* aborted */
     {
       thd->some_tables_deleted=1;		// Try again
-      sql_lock->lock_count=0;			// Locks are alread freed
+      sql_lock->lock_count= 0;                  // Locks are already freed
     }
     else if (!thd->some_tables_deleted || (flags & MYSQL_LOCK_IGNORE_FLUSH))
     {
diff --git a/sql/log.cc b/sql/log.cc
index 7d7ba90509d..1ef522588ff 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -2652,7 +2652,7 @@ int TC_LOG_MMAP::log(THD *thd, my_xid xid)
   {                                          // somebody's syncing. let's wait
     p->waiters++;
     /*
-      note - it must be while(), not do ... while() here
+      note - it must be while (), not do ... while () here
       as p->state may be not DIRTY when we come here
     */
     while (p->state == DIRTY && syncing)
diff --git a/sql/log_event.cc b/sql/log_event.cc
index 1519cfb6630..9b30989278e 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -243,7 +243,7 @@ static void print_set_option(FILE* file, uint32 bits_changed, uint32 option,
   {
     if (*need_comma)
       fprintf(file,", ");
-    fprintf(file,"%s=%d", name, (bool)(flags & option));
+    fprintf(file,"%s=%d", name, test(flags & option));
     *need_comma= 1;
   }
 }
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index 91287a8bd40..933b2f8efad 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -475,6 +475,11 @@ typedef my_bool (*qc_engine_callback)(THD *thd, char *table_key,
 #include "protocol.h"
 #include "sql_udf.h"
 class user_var_entry;
+enum enum_var_type
+{
+  OPT_DEFAULT= 0, OPT_SESSION, OPT_GLOBAL
+};
+class sys_var;
 #include "item.h"
 extern my_decimal decimal_zero;
 typedef Comp_creator* (*chooser_compare_func_creator)(bool invert);
@@ -1093,6 +1098,7 @@ extern const char **errmesg;			/* Error messages */
 extern const char *myisam_recover_options_str;
 extern const char *in_left_expr_name, *in_additional_cond;
 extern const char * const triggers_file_ext;
+extern const char * const trigname_file_ext;
 extern Eq_creator eq_creator;
 extern Ne_creator ne_creator;
 extern Gt_creator gt_creator;
@@ -1345,12 +1351,9 @@ extern bool sql_cache_init();
 extern void sql_cache_free();
 extern int sql_cache_hit(THD *thd, char *inBuf, uint length);
 
-/* item.cc */
+/* item_func.cc */
 Item *get_system_var(THD *thd, enum_var_type var_type, LEX_STRING name,
 		     LEX_STRING component);
-Item *get_system_var(THD *thd, enum_var_type var_type, const char *var_name,
-		     uint length, const char *item_name);
-/* item_func.cc */
 int get_var_with_binlog(THD *thd, LEX_STRING &name,
                         user_var_entry **out_entry);
 /* log.cc */
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 80670f2a445..412d6175e10 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -686,7 +686,13 @@ static void close_connections(void)
   {
     DBUG_PRINT("quit",("Informing thread %ld that it's time to die",
 		       tmp->thread_id));
-    tmp->killed= THD::KILL_CONNECTION;
+    /*
+      Re: bug 7403 - close_connection will be called mulitple times
+      a wholesale clean up of our network code is a very large project.
+      This will wake up the socket on Windows and prevent the printing of
+      the error message that we are force closing a connection.
+    */
+    close_connection(tmp, 0, 0);
     if (tmp->mysys_var)
     {
       tmp->mysys_var->abort=1;
@@ -883,7 +889,7 @@ static void __cdecl kill_server(int sig_ptr)
     unireg_end();
 
 #ifdef __NETWARE__
-  if(!event_flag)
+  if (!event_flag)
     pthread_join(select_thread, NULL);		// wait for main thread
 #endif /* __NETWARE__ */
 
@@ -4343,7 +4349,8 @@ enum options_mysqld
   OPT_ENABLE_LARGE_PAGES,
   OPT_TIMED_MUTEXES,
   OPT_OLD_STYLE_USER_LIMITS,
-  OPT_LOG_SLOW_ADMIN_STATEMENTS
+  OPT_LOG_SLOW_ADMIN_STATEMENTS,
+  OPT_TABLE_LOCK_WAIT_TIMEOUT
 };
 
 
@@ -5606,6 +5613,11 @@ The minimum value for this variable is 4096.",
    "The number of open tables for all threads.", (gptr*) &table_cache_size,
    (gptr*) &table_cache_size, 0, GET_ULONG, REQUIRED_ARG, 64, 1, 512*1024L,
    0, 1, 0},
+  {"table_lock_wait_timeout", OPT_TABLE_LOCK_WAIT_TIMEOUT, "Timeout in "
+    "seconds to wait for a table level lock before returning an error. Used"
+     " only if the connection has active cursors.",
+   (gptr*) &table_lock_wait_timeout, (gptr*) &table_lock_wait_timeout,
+   0, GET_ULONG, REQUIRED_ARG, 50, 1, 1024 * 1024 * 1024, 0, 1, 0},
   {"thread_cache_size", OPT_THREAD_CACHE_SIZE,
    "How many threads we should keep in a cache for reuse.",
    (gptr*) &thread_cache_size, (gptr*) &thread_cache_size, 0, GET_ULONG,
@@ -5739,6 +5751,7 @@ struct show_var_st status_vars[]= {
   {"Com_show_status",	       (char*) offsetof(STATUS_VAR, com_stat[(uint) SQLCOM_SHOW_STATUS]), SHOW_LONG_STATUS},
   {"Com_show_storage_engines", (char*) offsetof(STATUS_VAR, com_stat[(uint) SQLCOM_SHOW_STORAGE_ENGINES]), SHOW_LONG_STATUS},
   {"Com_show_tables",	       (char*) offsetof(STATUS_VAR, com_stat[(uint) SQLCOM_SHOW_TABLES]), SHOW_LONG_STATUS},
+  {"Com_show_triggers",	       (char*) offsetof(STATUS_VAR, com_stat[(uint) SQLCOM_SHOW_TRIGGERS]), SHOW_LONG_STATUS},
   {"Com_show_variables",       (char*) offsetof(STATUS_VAR, com_stat[(uint) SQLCOM_SHOW_VARIABLES]), SHOW_LONG_STATUS},
   {"Com_show_warnings",        (char*) offsetof(STATUS_VAR, com_stat[(uint) SQLCOM_SHOW_WARNS]), SHOW_LONG_STATUS},
   {"Com_slave_start",	       (char*) offsetof(STATUS_VAR, com_stat[(uint) SQLCOM_SLAVE_START]), SHOW_LONG_STATUS},
@@ -7073,4 +7086,6 @@ template class I_List_iterator<THD>;
 template class I_List<i_string>;
 template class I_List<i_string_pair>;
 template class I_List<NAMED_LIST>;
+template class I_List<Statement>;
+template class I_List_iterator<Statement>;
 #endif
diff --git a/sql/opt_range.cc b/sql/opt_range.cc
index c3e73632c6f..25fcdfc51fd 100644
--- a/sql/opt_range.cc
+++ b/sql/opt_range.cc
@@ -630,7 +630,7 @@ int imerge_list_or_tree(PARAM *param,
 {
   SEL_IMERGE *imerge;
   List_iterator<SEL_IMERGE> it(*im1);
-  while((imerge= it++))
+  while ((imerge= it++))
   {
     if (imerge->or_sel_tree_with_checks(param, tree))
       it.remove();
@@ -993,7 +993,7 @@ int QUICK_ROR_INTERSECT_SELECT::init_ror_merged_scan(bool reuse_handler)
       DBUG_RETURN(1);
     quick->file->extra(HA_EXTRA_KEYREAD_PRESERVE_FIELDS);
   }
-  while((quick= quick_it++))
+  while ((quick= quick_it++))
   {
     if (quick->init_ror_merged_scan(FALSE))
       DBUG_RETURN(1);
@@ -3536,7 +3536,8 @@ static SEL_TREE *get_mm_tree(PARAM *param,COND *cond)
     if (arg->type() != Item::FUNC_ITEM)
       DBUG_RETURN(0);
     cond_func= (Item_func*) arg;
-    if (cond_func->select_optimize() == Item_func::OPTIMIZE_NONE)
+    if (cond_func->functype() != Item_func::BETWEEN &&
+        cond_func->functype() != Item_func::IN_FUNC)
       DBUG_RETURN(0);
     inv= TRUE;
   }
@@ -6947,7 +6948,7 @@ get_best_group_min_max(PARAM *param, SEL_TREE *tree)
   List_iterator<Item> select_items_it(join->fields_list);
 
   /* Check (SA1,SA4) and store the only MIN/MAX argument - the C attribute.*/
-  if(join->make_sum_func_list(join->all_fields, join->fields_list, 1))
+  if (join->make_sum_func_list(join->all_fields, join->fields_list, 1))
     DBUG_RETURN(NULL);
   if (join->sum_funcs[0])
   {
@@ -7273,7 +7274,7 @@ check_group_min_max_predicates(COND *cond, Item_field *min_max_arg_item,
     Item *and_or_arg;
     while ((and_or_arg= li++))
     {
-      if(!check_group_min_max_predicates(and_or_arg, min_max_arg_item,
+      if (!check_group_min_max_predicates(and_or_arg, min_max_arg_item,
                                          image_type))
         DBUG_RETURN(FALSE);
     }
@@ -7355,7 +7356,7 @@ check_group_min_max_predicates(COND *cond, Item_field *min_max_arg_item,
     }
     else if (cur_arg->type() == Item::FUNC_ITEM)
     {
-      if(!check_group_min_max_predicates(cur_arg, min_max_arg_item,
+      if (!check_group_min_max_predicates(cur_arg, min_max_arg_item,
                                          image_type))
         DBUG_RETURN(FALSE);
     }
@@ -7886,19 +7887,19 @@ int QUICK_GROUP_MIN_MAX_SELECT::init()
 
   if (min_max_arg_part)
   {
-    if(my_init_dynamic_array(&min_max_ranges, sizeof(QUICK_RANGE*), 16, 16))
+    if (my_init_dynamic_array(&min_max_ranges, sizeof(QUICK_RANGE*), 16, 16))
       return 1;
 
     if (have_min)
     {
-      if(!(min_functions= new List<Item_sum>))
+      if (!(min_functions= new List<Item_sum>))
         return 1;
     }
     else
       min_functions= NULL;
     if (have_max)
     {
-      if(!(max_functions= new List<Item_sum>))
+      if (!(max_functions= new List<Item_sum>))
         return 1;
     }
     else
@@ -7972,7 +7973,7 @@ bool QUICK_GROUP_MIN_MAX_SELECT::add_range(SEL_ARG *sel_range)
   uint range_flag= sel_range->min_flag | sel_range->max_flag;
 
   /* Skip (-inf,+inf) ranges, e.g. (x < 5 or x > 4). */
-  if((range_flag & NO_MIN_RANGE) && (range_flag & NO_MAX_RANGE))
+  if ((range_flag & NO_MIN_RANGE) && (range_flag & NO_MAX_RANGE))
     return FALSE;
 
   if (!(sel_range->min_flag & NO_MIN_RANGE) &&
diff --git a/sql/parse_file.cc b/sql/parse_file.cc
index 7cc563901d2..abca8736916 100644
--- a/sql/parse_file.cc
+++ b/sql/parse_file.cc
@@ -728,7 +728,7 @@ File_parser::parse(gptr base, MEM_ROOT *mem_root,
 					       sizeof(LEX_STRING))) ||
 		list->push_back(str, mem_root))
 	      goto list_err;
-	    if(!(ptr= parse_quoted_escaped_string(ptr, end, mem_root, str)))
+	    if (!(ptr= parse_quoted_escaped_string(ptr, end, mem_root, str)))
 	      goto list_err_w_message;
 	    switch (*ptr) {
 	    case '\n':
diff --git a/sql/set_var.cc b/sql/set_var.cc
index eb6cdfa37d0..f1ac90ea113 100644
--- a/sql/set_var.cc
+++ b/sql/set_var.cc
@@ -375,6 +375,8 @@ sys_var_thd_ulong	sys_sync_replication_timeout(
 sys_var_bool_ptr	sys_sync_frm("sync_frm", &opt_sync_frm);
 sys_var_long_ptr	sys_table_cache_size("table_cache",
 					     &table_cache_size);
+sys_var_long_ptr	sys_table_lock_wait_timeout("table_lock_wait_timeout",
+                                                    &table_lock_wait_timeout);
 sys_var_long_ptr	sys_thread_cache_size("thread_cache_size",
 					      &thread_cache_size);
 sys_var_thd_enum	sys_tx_isolation("tx_isolation",
@@ -682,6 +684,7 @@ sys_var *sys_variables[]=
 #endif
   &sys_sync_frm,
   &sys_table_cache_size,
+  &sys_table_lock_wait_timeout,
   &sys_table_type,
   &sys_thread_cache_size,
   &sys_time_format,
@@ -973,6 +976,7 @@ struct show_var_st init_vars[]= {
   {"system_time_zone",        system_time_zone,                     SHOW_CHAR},
 #endif
   {"table_cache",             (char*) &table_cache_size,            SHOW_LONG},
+  {"table_lock_wait_timeout", (char*) &table_lock_wait_timeout,     SHOW_LONG },
   {sys_table_type.name,	      (char*) &sys_table_type,	            SHOW_SYS},
   {sys_thread_cache_size.name,(char*) &sys_thread_cache_size,       SHOW_SYS},
 #ifdef HAVE_THR_SETCONCURRENCY
@@ -1653,15 +1657,7 @@ err:
 
 /*
   Return an Item for a variable.  Used with @@[global.]variable_name
-
   If type is not given, return local value if exists, else global
-
-  We have to use netprintf() instead of my_error() here as this is
-  called on the parsing stage.
-
-  TODO:
-    With prepared statements/stored procedures this has to be fixed
-    to create an item that gets the current value at fix_fields() stage.
 */
 
 Item *sys_var::item(THD *thd, enum_var_type var_type, LEX_STRING *base)
diff --git a/sql/set_var.h b/sql/set_var.h
index a6532323b34..a7e680cc7fa 100644
--- a/sql/set_var.h
+++ b/sql/set_var.h
@@ -30,11 +30,6 @@ class set_var;
 typedef struct system_variables SV;
 extern TYPELIB bool_typelib, delay_key_write_typelib, sql_mode_typelib;
 
-enum enum_var_type
-{
-  OPT_DEFAULT= 0, OPT_SESSION, OPT_GLOBAL
-};
-
 typedef int (*sys_check_func)(THD *,  set_var *);
 typedef bool (*sys_update_func)(THD *, set_var *);
 typedef void (*sys_after_update_func)(THD *,enum_var_type);
diff --git a/sql/share/errmsg.txt b/sql/share/errmsg.txt
index 466f1049dc5..b3784b6421d 100644
--- a/sql/share/errmsg.txt
+++ b/sql/share/errmsg.txt
@@ -5370,6 +5370,20 @@ ER_SCALE_BIGGER_THAN_PRECISION 42000 S1009
         eng "Scale may not be larger than the precision (column '%-.64s')."
 ER_WRONG_LOCK_OF_SYSTEM_TABLE
         eng "You can't combine write-locking of system '%-.64s.%-.64s' table with other tables"
+ER_CONNECT_TO_FOREIGN_DATA_SOURCE
+        eng "Unable to connect to foreign data source - database '%s'!"
+ER_QUERY_ON_FOREIGN_DATA_SOURCE
+        eng "There was a problem processing the query on the foreign data source. Data source error: '%-.64s'"
+ER_FOREIGN_DATA_SOURCE_DOESNT_EXIST
+        eng "The foreign data source you are trying to reference does not exist. Data source error : '%-.64s'"
+ER_FOREIGN_DATA_STRING_INVALID_CANT_CREATE
+        eng "Can't create federated table. The data source connection string '%-.64s' is not in the correct format"
+ER_FOREIGN_DATA_STRING_INVALID
+        eng "The data source connection string '%-.64s' is not in the correct format"
+ER_CANT_CREATE_FEDERATED_TABLE
+        eng "Can't create federated table. Foreign data src error : '%-.64s'"
+ER_TRG_IN_WRONG_SCHEMA
+        eng "Trigger in wrong schema"
 ER_PARTITION_REQUIRES_VALUES_ERROR
         eng "%s PARTITIONING requires definition of VALUES %s for each partition"
         swe "%s PARTITIONering kräver definition av VALUES %s för varje partition"
diff --git a/sql/sp.cc b/sql/sp.cc
index cf381762bac..0c9af895fb6 100644
--- a/sql/sp.cc
+++ b/sql/sp.cc
@@ -1442,8 +1442,8 @@ sp_cache_routines_and_add_tables_for_triggers(THD *thd, LEX *lex,
   {
     Sroutine_hash_entry **last_cached_routine_ptr=
                             (Sroutine_hash_entry **)lex->sroutines_list.next;
-    for (int i= 0; i < 3; i++)
-      for (int j= 0; j < 2; j++)
+    for (int i= 0; i < (int)TRG_EVENT_MAX; i++)
+      for (int j= 0; j < (int)TRG_ACTION_MAX; j++)
         if (triggers->bodies[i][j])
         {
           (void)triggers->bodies[i][j]->add_used_tables_to_table_list(thd,
diff --git a/sql/sp_head.cc b/sql/sp_head.cc
index 272456d8c8e..02c006d01ee 100644
--- a/sql/sp_head.cc
+++ b/sql/sp_head.cc
@@ -1023,6 +1023,7 @@ sp_head::reset_lex(THD *thd)
   DBUG_ENTER("sp_head::reset_lex");
   LEX *sublex;
   LEX *oldlex= thd->lex;
+  my_lex_states state= oldlex->next_state; // Keep original next_state
 
   (void)m_lex.push_front(oldlex);
   thd->lex= sublex= new st_lex;
@@ -1030,6 +1031,11 @@ sp_head::reset_lex(THD *thd)
   /* Reset most stuff. The length arguments doesn't matter here. */
   lex_start(thd, oldlex->buf, (ulong) (oldlex->end_of_query - oldlex->ptr));
 
+  /*
+   * next_state is normally the same (0), but it happens that we swap lex in
+   * "mid-sentence", so we must restore it.
+   */
+  sublex->next_state= state;
   /* We must reset ptr and end_of_query again */
   sublex->ptr= oldlex->ptr;
   sublex->end_of_query= oldlex->end_of_query;
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 5d6dd29b998..eed7d749b2c 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -1740,7 +1740,7 @@ static int open_unireg_entry(THD *thd, TABLE *entry, const char *db,
       !my_strcasecmp(system_charset_info, name, "proc"))
     entry->s->system_table= 1;
 
-  if (Table_triggers_list::check_n_load(thd, db, name, entry))
+  if (Table_triggers_list::check_n_load(thd, db, name, entry, 0))
     goto err;
 
   /*
diff --git a/sql/sql_bitmap.h b/sql/sql_bitmap.h
index 00f38895fd6..b2994da71f7 100644
--- a/sql/sql_bitmap.h
+++ b/sql/sql_bitmap.h
@@ -51,6 +51,14 @@ public:
     bitmap_init(&map2, (uint32 *)&map2buff, sizeof(ulonglong)*8, 0);
     bitmap_intersect(&map, &map2);
   }
+  /* Use highest bit for all bits above sizeof(ulonglong)*8. */
+  void intersect_extended(ulonglong map2buff)
+  {
+    intersect(map2buff);
+    if (map.bitmap_size > sizeof(ulonglong))
+      bitmap_set_above(&map, sizeof(ulonglong),
+                       test(map2buff & (LL(1) << (sizeof(ulonglong) * 8 - 1))));
+  }
   void subtract(Bitmap& map2) { bitmap_subtract(&map, &map2.map); }
   void merge(Bitmap& map2) { bitmap_union(&map, &map2.map); }
   my_bool is_set(uint n) const { return bitmap_is_set(&map, n); }
@@ -116,6 +124,7 @@ public:
   void clear_all() { map=(ulonglong)0; }
   void intersect(Bitmap<64>& map2) { map&= map2.map; }
   void intersect(ulonglong map2) { map&= map2; }
+  void intersect_extended(ulonglong map2) { map&= map2; }
   void subtract(Bitmap<64>& map2) { map&= ~map2.map; }
   void merge(Bitmap<64>& map2) { map|= map2.map; }
   my_bool is_set(uint n) const { return test(map & (((ulonglong)1) << n)); }
diff --git a/sql/sql_cache.cc b/sql/sql_cache.cc
index cbee80a8695..e70ab38ee43 100644
--- a/sql/sql_cache.cc
+++ b/sql/sql_cache.cc
@@ -2051,7 +2051,7 @@ my_bool Query_cache::allocate_data_chain(Query_cache_block **result_block,
     */
     data_len= len - new_block->length;
     prev_block= new_block;
-  } while(1);
+  } while (1);
 
   DBUG_RETURN(TRUE);
 }
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index d0ac1a16f6b..89d5b543dfc 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -173,6 +173,7 @@ Open_tables_state::Open_tables_state()
 THD::THD()
   :Statement(CONVENTIONAL_EXECUTION, 0, ALLOC_ROOT_MIN_BLOCK_SIZE, 0),
    Open_tables_state(),
+   lock_id(&main_lock_id),
    user_time(0), global_read_lock(0), is_fatal_error(0),
    rand_used(0), time_zone_used(0),
    last_insert_id_used(0), insert_id_used(0), clear_next_insert_id(0),
@@ -265,6 +266,8 @@ THD::THD()
   tablespace_op=FALSE;
   ulong tmp=sql_rnd_with_mutex();
   randominit(&rand, tmp + (ulong) &rand, tmp + (ulong) ::query_id);
+  thr_lock_info_init(&lock_info); /* safety: will be reset after start */
+  thr_lock_owner_init(&main_lock_id, &lock_info);
 }
 
 
@@ -406,6 +409,8 @@ THD::~THD()
     net_end(&net);
   }
 #endif
+  stmt_map.destroy();                     /* close all prepared statements */
+  DBUG_ASSERT(lock_info.n_cursors == 0);
   if (!cleanup_done)
     cleanup();
 
@@ -518,6 +523,7 @@ bool THD::store_globals()
     if this is the slave SQL thread.
   */
   variables.pseudo_thread_id= thread_id;
+  thr_lock_info_init(&lock_info);
   return 0;
 }
 
@@ -1563,6 +1569,12 @@ void Statement::restore_backup_statement(Statement *stmt, Statement *backup)
 }
 
 
+void Statement::close_cursor()
+{
+  DBUG_ASSERT("Statement::close_cursor()" == "not implemented");
+}
+
+
 void THD::end_statement()
 {
   /* Cleanup SQL processing state to resuse this statement in next query. */
@@ -1683,6 +1695,14 @@ int Statement_map::insert(Statement *statement)
 }
 
 
+void Statement_map::close_transient_cursors()
+{
+  Statement *stmt;
+  while ((stmt= transient_cursor_list.head()))
+    stmt->close_cursor();                 /* deletes itself from the list */
+}
+
+
 bool select_dumpvar::send_data(List<Item> &items)
 {
   List_iterator_fast<Item_func_set_user_var> li(vars);
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 37bfd1656a8..283ae96eebc 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -757,7 +757,7 @@ class Cursor;
   be used explicitly.
 */
 
-class Statement: public Query_arena
+class Statement: public ilink, public Query_arena
 {
   Statement(const Statement &rhs);              /* not implemented: */
   Statement &operator=(const Statement &rhs);   /* non-copyable */
@@ -841,6 +841,8 @@ public:
   void restore_backup_statement(Statement *stmt, Statement *backup);
   /* return class type */
   virtual Type type() const;
+  /* Close the cursor open for this statement, if there is one */
+  virtual void close_cursor();
 };
 
 
@@ -892,15 +894,25 @@ public:
     }
     hash_delete(&st_hash, (byte *) statement);
   }
+  void add_transient_cursor(Statement *stmt)
+  { transient_cursor_list.append(stmt); }
+  void erase_transient_cursor(Statement *stmt) { stmt->unlink(); }
+  /*
+    Close all cursors of this connection that use tables of a storage
+    engine that has transaction-specific state and therefore can not
+    survive COMMIT or ROLLBACK. Currently all but MyISAM cursors are closed.
+  */
+  void close_transient_cursors();
   /* Erase all statements (calls Statement destructor) */
   void reset()
   {
     my_hash_reset(&names_hash);
     my_hash_reset(&st_hash);
+    transient_cursor_list.empty();
     last_found_statement= 0;
   }
 
-  ~Statement_map()
+  void destroy()
   {
     hash_free(&names_hash);
     hash_free(&st_hash);
@@ -908,6 +920,7 @@ public:
 private:
   HASH st_hash;
   HASH names_hash;
+  I_List<Statement> transient_cursor_list;
   Statement *last_found_statement;
 };
 
@@ -1025,8 +1038,7 @@ public:
   a thread/connection descriptor
 */
 
-class THD :public ilink,
-           public Statement,
+class THD :public Statement,
            public Open_tables_state
 {
 public:
@@ -1052,6 +1064,10 @@ public:
   struct  rand_struct rand;		// used for authentication
   struct  system_variables variables;	// Changeable local variables
   struct  system_status_var status_var; // Per thread statistic vars
+  THR_LOCK_INFO lock_info;              // Locking info of this thread
+  THR_LOCK_OWNER main_lock_id;          // To use for conventional queries
+  THR_LOCK_OWNER *lock_id;              // If not main_lock_id, points to
+                                        // the lock_id of a cursor.
   pthread_mutex_t LOCK_delete;		// Locked before thd is deleted
   /* all prepared statements and cursors of this connection */
   Statement_map stmt_map; 
@@ -1475,6 +1491,8 @@ public:
              (variables.sql_mode & MODE_STRICT_ALL_TABLES)));
   }
   void set_status_var_init();
+  bool is_context_analysis_only()
+    { return current_arena->is_stmt_prepare() || lex->view_prepare_mode; }
   bool push_open_tables_state();
   void pop_open_tables_state();
 };
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index edcf4db09a4..ec982703116 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -58,6 +58,7 @@ enum enum_sql_command {
   SQLCOM_SHOW_PROCESSLIST, SQLCOM_SHOW_MASTER_STAT, SQLCOM_SHOW_SLAVE_STAT,
   SQLCOM_SHOW_GRANTS, SQLCOM_SHOW_CREATE, SQLCOM_SHOW_CHARSETS,
   SQLCOM_SHOW_COLLATIONS, SQLCOM_SHOW_CREATE_DB, SQLCOM_SHOW_TABLE_STATUS,
+  SQLCOM_SHOW_TRIGGERS,
 
   SQLCOM_LOAD,SQLCOM_SET_OPTION,SQLCOM_LOCK_TABLES,SQLCOM_UNLOCK_TABLES,
   SQLCOM_GRANT,
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index e0c5338a952..58af959888e 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -2108,6 +2108,7 @@ int prepare_schema_table(THD *thd, LEX *lex, Table_ident *table_ident,
   case SCH_TABLE_NAMES:
   case SCH_TABLES:
   case SCH_VIEWS:
+  case SCH_TRIGGERS:
 #ifdef DONT_ALLOW_SHOW_COMMANDS
     my_message(ER_NOT_ALLOWED_COMMAND,
                ER(ER_NOT_ALLOWED_COMMAND), MYF(0)); /* purecov: inspected */
@@ -2155,7 +2156,7 @@ int prepare_schema_table(THD *thd, LEX *lex, Table_ident *table_ident,
       TABLE_LIST **query_tables_last= lex->query_tables_last;
       sel= new SELECT_LEX();
       sel->init_query();
-      if(!sel->add_table_to_list(thd, table_ident, 0, 0, TL_READ, 
+      if (!sel->add_table_to_list(thd, table_ident, 0, 0, TL_READ, 
                                  (List<String> *) 0, (List<String> *) 0))
         DBUG_RETURN(1);
       lex->query_tables_last= query_tables_last;
@@ -2308,7 +2309,8 @@ mysql_execute_command(THD *thd)
     Don't reset warnings when executing a stored routine.
   */
   if ((all_tables || &lex->select_lex != lex->all_selects_list ||
-       lex->sroutines.records) && !thd->spcont)
+       lex->sroutines.records) && !thd->spcont ||
+      lex->time_zone_tables_used)
     mysql_reset_errors(thd, 0);
 
 #ifdef HAVE_REPLICATION
@@ -2389,10 +2391,12 @@ mysql_execute_command(THD *thd)
     select_result *result=lex->result;
     if (all_tables)
     {
-      res= check_table_access(thd,
-			      lex->exchange ? SELECT_ACL | FILE_ACL :
-			      SELECT_ACL,
-			      all_tables, 0);
+      if (lex->orig_sql_command != SQLCOM_SHOW_STATUS_PROC &&
+          lex->orig_sql_command != SQLCOM_SHOW_STATUS_FUNC)
+        res= check_table_access(thd,
+                                lex->exchange ? SELECT_ACL | FILE_ACL :
+                                SELECT_ACL,
+                                all_tables, 0);
     }
     else
       res= check_access(thd,
@@ -6847,7 +6851,7 @@ bool multi_update_precheck(THD *thd, TABLE_LIST *tables)
   /*
     Is there tables of subqueries?
   */
-  if (&lex->select_lex != lex->all_selects_list)
+  if (&lex->select_lex != lex->all_selects_list || lex->time_zone_tables_used)
   {
     DBUG_PRINT("info",("Checking sub query list"));
     for (table= tables; table; table= table->next_global)
diff --git a/sql/sql_prepare.cc b/sql/sql_prepare.cc
index b4293ef7836..9ec2cff71a1 100644
--- a/sql/sql_prepare.cc
+++ b/sql/sql_prepare.cc
@@ -1757,6 +1757,10 @@ bool mysql_stmt_prepare(THD *thd, char *packet, uint packet_length,
     DBUG_RETURN(TRUE);
   }
 
+  /*
+    alloc_query() uses thd->memroot && thd->query, so we have to call
+    both of backup_statement() and backup_item_area() here.
+  */
   thd->set_n_backup_statement(stmt, &stmt_backup);
   thd->set_n_backup_item_arena(stmt, &stmt_backup);
 
@@ -2016,6 +2020,7 @@ void mysql_stmt_execute(THD *thd, char *packet, uint packet_length)
         DBUG_VOID_RETURN;
       /* If lex->result is set, mysql_execute_command will use it */
       stmt->lex->result= &cursor->result;
+      thd->lock_id= &cursor->lock_id;
     }
   }
 #ifndef EMBEDDED_LIBRARY
@@ -2065,6 +2070,9 @@ void mysql_stmt_execute(THD *thd, char *packet, uint packet_length)
       Cursor::open is buried deep in JOIN::exec of the top level join.
     */
     cursor->init_from_thd(thd);
+
+    if (cursor->close_at_commit)
+      thd->stmt_map.add_transient_cursor(stmt);
   }
   else
   {
@@ -2074,6 +2082,7 @@ void mysql_stmt_execute(THD *thd, char *packet, uint packet_length)
   }
 
   thd->set_statement(&stmt_backup);
+  thd->lock_id= &thd->main_lock_id;
   thd->current_arena= thd;
   DBUG_VOID_RETURN;
 
@@ -2244,10 +2253,12 @@ void mysql_stmt_fetch(THD *thd, char *packet, uint packet_length)
     cleanup_stmt_and_thd_after_use(stmt, thd);
     reset_stmt_params(stmt);
     /*
-      Must be the last, as some momory is still needed for
+      Must be the last, as some memory is still needed for
       the previous calls.
     */
     free_root(cursor->mem_root, MYF(0));
+    if (cursor->close_at_commit)
+      thd->stmt_map.erase_transient_cursor(stmt);
   }
 
   thd->restore_backup_statement(stmt, &stmt_backup);
@@ -2287,14 +2298,6 @@ void mysql_stmt_reset(THD *thd, char *packet)
     DBUG_VOID_RETURN;
 
   stmt->close_cursor();                    /* will reset statement params */
-  cursor= stmt->cursor;
-  if (cursor && cursor->is_open())
-  {
-    thd->change_list= cursor->change_list;
-    cursor->close(FALSE);
-    cleanup_stmt_and_thd_after_use(stmt, thd);
-    free_root(cursor->mem_root, MYF(0));
-  }
 
   stmt->state= Query_arena::PREPARED;
 
@@ -2474,6 +2477,8 @@ void Prepared_statement::close_cursor()
     cursor->close(FALSE);
     cleanup_stmt_and_thd_after_use(this, thd);
     free_root(cursor->mem_root, MYF(0));
+    if (cursor->close_at_commit)
+      thd->stmt_map.erase_transient_cursor(this);
   }
   /*
     Clear parameters from data which could be set by
diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc
index dbefbb34b4a..d376423e990 100644
--- a/sql/sql_repl.cc
+++ b/sql/sql_repl.cc
@@ -1527,7 +1527,7 @@ bool show_binlogs(THD* thd)
     else
     {
       /* this is an old log, open it and find the size */
-      if ((file= my_open(fname+dir_len, O_RDONLY | O_SHARE | O_BINARY,
+      if ((file= my_open(fname, O_RDONLY | O_SHARE | O_BINARY,
                          MYF(0))) >= 0)
       {
         file_length= (ulonglong) my_seek(file, 0L, MY_SEEK_END, MYF(0));
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index c54efe531ad..45460d0da01 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -1194,7 +1194,7 @@ JOIN::exec()
     {
       result->send_fields(fields_list,
                           Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF);
-      if (!having || having->val_int())
+      if (cond_value != Item::COND_FALSE && (!having || having->val_int()))
       {
 	if (do_send_rows && (procedure ? (procedure->send_row(fields_list) ||
                                           procedure->end_of_records())
@@ -1714,10 +1714,12 @@ JOIN::destroy()
 
 Cursor::Cursor(THD *thd)
   :Query_arena(&main_mem_root, INITIALIZED),
-   join(0), unit(0)
+   join(0), unit(0),
+   close_at_commit(FALSE)
 {
   /* We will overwrite it at open anyway. */
   init_sql_alloc(&main_mem_root, ALLOC_ROOT_MIN_BLOCK_SIZE, 0);
+  thr_lock_owner_init(&lock_id, &thd->lock_info);
 }
 
 
@@ -1751,6 +1753,21 @@ Cursor::init_from_thd(THD *thd)
   free_list=	  thd->free_list;
   change_list=    thd->change_list;
   reset_thd(thd);
+  /* Now we have an active cursor and can cause a deadlock */
+  thd->lock_info.n_cursors++;
+
+  close_at_commit= FALSE; /* reset in case we're reusing the cursor */
+  for (TABLE *table= open_tables; table; table= table->next)
+  {
+    const handlerton *ht= table->file->ht;
+    if (ht)
+      close_at_commit|= (ht->flags & HTON_CLOSE_CURSORS_AT_COMMIT);
+    else
+    {
+      close_at_commit= TRUE;     /* handler status is unknown */
+      break;
+    }
+  }
   /*
     XXX: thd->locked_tables is not changed.
     What problems can we have with it if cursor is open?
@@ -1919,6 +1936,7 @@ Cursor::close(bool is_active)
     thd->derived_tables= tmp_derived_tables;
     thd->lock= tmp_lock;
   }
+  thd->lock_info.n_cursors--; /* Decrease the number of active cursors */
   join= 0;
   unit= 0;
   free_items();
@@ -2846,11 +2864,11 @@ add_key_fields(KEY_FIELD **key_fields,uint *and_level,
         cond_func->arguments()[1]->real_item()->type() == Item::FIELD_ITEM &&
 	     !(cond_func->arguments()[0]->used_tables() & OUTER_REF_TABLE_BIT))
         values--;
+      DBUG_ASSERT(cond_func->functype() != Item_func::IN_FUNC ||
+                  cond_func->argument_count() != 2);
       add_key_equal_fields(key_fields, *and_level, cond_func,
                            (Item_field*) (cond_func->key_item()->real_item()),
-                           cond_func->argument_count() == 2 &&
-                           cond_func->functype() == Item_func::IN_FUNC,
-                           values,
+                           0, values,
                            cond_func->argument_count()-1,
                            usable_tables);
     }
@@ -5150,7 +5168,23 @@ inline void add_cond_and_fix(Item **e1, Item *e2)
       (where othertbl is a non-const table and othertbl.field may be NULL)
       and add them to conditions on correspoding tables (othertbl in this
       example).
-      
+
+      Exception from that is the case when referred_tab->join != join.
+      I.e. don't add NOT NULL constraints from any embedded subquery.
+      Consider this query:
+      SELECT A.f2 FROM t1 LEFT JOIN t2 A ON A.f2 = f1
+      WHERE A.f3=(SELECT MIN(f3) FROM  t2 C WHERE A.f4 = C.f4) OR A.f3 IS NULL;
+      Here condition A.f3 IS NOT NULL is going to be added to the WHERE
+      condition of the embedding query.
+      Another example:
+      SELECT * FROM t10, t11 WHERE (t10.a < 10 OR t10.a IS NULL)
+      AND t11.b <=> t10.b AND (t11.a = (SELECT MAX(a) FROM t12
+      WHERE t12.b = t10.a ));
+      Here condition t10.a IS NOT NULL is going to be added.
+      In both cases addition of NOT NULL condition will erroneously reject
+      some rows of the result set.
+      referred_tab->join != join constraint would disallow such additions.
+
       This optimization doesn't affect the choices that ref, range, or join
       optimizer make. This was intentional because this was added after 4.1
       was GA.
@@ -5181,6 +5215,13 @@ static void add_not_null_conds(JOIN *join)
           DBUG_ASSERT(item->type() == Item::FIELD_ITEM);
           Item_field *not_null_item= (Item_field*)item;
           JOIN_TAB *referred_tab= not_null_item->field->table->reginfo.join_tab;
+          /*
+            For UPDATE queries such as:
+            UPDATE t1 SET t1.f2=(SELECT MAX(t2.f4) FROM t2 WHERE t2.f3=t1.f1);
+            not_null_item is the t1.f1, but it's referred_tab is 0.
+          */
+          if (!referred_tab || referred_tab->join != join)
+            continue;
           Item *notnull;
           if (!(notnull= new Item_func_isnotnull(not_null_item)))
             DBUG_VOID_RETURN;
@@ -6712,7 +6753,7 @@ static COND *build_equal_items_for_cond(COND *cond,
        of the condition expression.
     */
     li.rewind();
-    while((item= li++))
+    while ((item= li++))
     { 
       Item *new_item;
       if ((new_item = build_equal_items_for_cond(item, inherited))!= item)
@@ -7521,7 +7562,7 @@ simplify_joins(JOIN *join, List<TABLE_LIST> *join_list, COND *conds, bool top)
     
   /* Flatten nested joins that can be flattened. */
   li.rewind();
-  while((table= li++))
+  while ((table= li++))
   {
     nested_join= table->nested_join;
     if (nested_join && !table->on_expr)
@@ -12134,7 +12175,6 @@ create_distinct_group(THD *thd, Item **ref_pointer_array,
   List_iterator<Item> li(fields);
   Item *item;
   ORDER *order,*group,**prev;
-  uint index= 0;
 
   *all_order_by_fields_used= 1;
   while ((item=li++))
@@ -12171,12 +12211,12 @@ create_distinct_group(THD *thd, Item **ref_pointer_array,
         simple indexing of ref_pointer_array (order in the array and in the
         list are same)
       */
-      ord->item= ref_pointer_array + index;
+      ord->item= ref_pointer_array;
       ord->asc=1;
       *prev=ord;
       prev= &ord->next;
     }
-    index++;
+    ref_pointer_array++;
   }
   *prev=0;
   return group;
@@ -13005,7 +13045,7 @@ static bool change_group_ref(THD *thd, Item_func *expr, ORDER *group_list,
           if (item->eq(*group_tmp->item,0))
           {
             Item *new_item;
-            if(!(new_item= new Item_ref(context, group_tmp->item, 0,
+            if (!(new_item= new Item_ref(context, group_tmp->item, 0,
                                         item->name)))
               return 1;                                 // fatal_error is set
             thd->change_item_tree(arg, new_item);
@@ -13075,7 +13115,7 @@ bool JOIN::rollup_init()
     ORDER *group_tmp;
     for (group_tmp= group_list; group_tmp; group_tmp= group_tmp->next)
     {
-      if (item->eq(*group_tmp->item,0))
+      if (*group_tmp->item == item)
         item->maybe_null= 1;
     }
     if (item->type() == Item::FUNC_ITEM)
@@ -13195,7 +13235,7 @@ bool JOIN::rollup_make_fields(List<Item> &fields_arg, List<Item> &sel_fields,
 	for (group_tmp= start_group, i= pos ;
              group_tmp ; group_tmp= group_tmp->next, i++)
 	{
-          if (item->eq(*group_tmp->item,0))
+          if (*group_tmp->item == item)
 	  {
 	    /*
 	      This is an element that is used by the GROUP BY and should be
diff --git a/sql/sql_select.h b/sql/sql_select.h
index ec424c48366..2b2557de180 100644
--- a/sql/sql_select.h
+++ b/sql/sql_select.h
@@ -393,6 +393,8 @@ class Cursor: public Sql_alloc, public Query_arena
 public:
   Item_change_list change_list;
   select_send result;
+  THR_LOCK_OWNER lock_id;
+  my_bool close_at_commit;
 
   /* Temporary implementation as now we replace THD state by value */
   /* Save THD state into cursor */
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index e786b70114d..d56c14e2836 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -21,6 +21,7 @@
 #include "sql_select.h"                         // For select_describe
 #include "repl_failsafe.h"
 #include "sp_head.h"
+#include "sql_trigger.h"
 #include <my_dir.h>
 
 #ifdef HAVE_BERKELEY_DB
@@ -1718,6 +1719,7 @@ void get_index_field_values(LEX *lex, INDEX_FIELD_VALUES *index_field_values)
     break;
   case SQLCOM_SHOW_TABLES:
   case SQLCOM_SHOW_TABLE_STATUS:
+  case SQLCOM_SHOW_TRIGGERS:
     index_field_values->db_value= lex->current_select->db;
     index_field_values->table_value= wild;
     break;
@@ -1740,7 +1742,7 @@ int make_table_list(THD *thd, SELECT_LEX *sel,
   ident_table.length= strlen(table);
   table_ident= new Table_ident(thd, ident_db, ident_table, 1);
   sel->init_query();
-  if(!sel->add_table_to_list(thd, table_ident, 0, 0, TL_READ,
+  if (!sel->add_table_to_list(thd, table_ident, 0, 0, TL_READ,
                              (List<String> *) 0, (List<String> *) 0))
     return 1;
   return 0;
@@ -2399,6 +2401,7 @@ static int get_schema_column_record(THD *thd, struct st_table_list *tables,
     {
       const char *tmp_buff;
       byte *pos;
+      bool is_blob;
       uint flags=field->flags;
       char tmp[MAX_FIELD_WIDTH];
       char tmp1[MAX_FIELD_WIDTH];
@@ -2477,12 +2480,14 @@ static int get_schema_column_record(THD *thd, struct st_table_list *tables,
                    "NO" : "YES");
       table->field[6]->store((const char*) pos,
                              strlen((const char*) pos), cs);
-      if (field->has_charset())
+      is_blob= (field->type() == FIELD_TYPE_BLOB);
+      if (field->has_charset() || is_blob)
       {
-        table->field[8]->store((longlong) field->field_length/
-                               field->charset()->mbmaxlen);
+        longlong c_octet_len= is_blob ? (longlong) field->max_length() :
+          (longlong) field->max_length()/field->charset()->mbmaxlen;
+        table->field[8]->store(c_octet_len);
         table->field[8]->set_notnull();
-        table->field[9]->store((longlong) field->field_length);
+        table->field[9]->store((longlong) field->max_length());
         table->field[9]->set_notnull();
       }
 
@@ -2510,6 +2515,17 @@ static int get_schema_column_record(THD *thd, struct st_table_list *tables,
         case FIELD_TYPE_LONG:
         case FIELD_TYPE_LONGLONG:
         case FIELD_TYPE_INT24:
+        {
+          table->field[10]->store((longlong) field->max_length() - 1);
+          table->field[10]->set_notnull();
+          break;
+        }
+        case FIELD_TYPE_BIT:
+        {
+          table->field[10]->store((longlong) field->max_length());
+          table->field[10]->set_notnull();
+          break;
+        }
         case FIELD_TYPE_FLOAT:  
         case FIELD_TYPE_DOUBLE:
         {
@@ -2985,6 +3001,73 @@ static int get_schema_constraints_record(THD *thd, struct st_table_list *tables,
 }
 
 
+static bool store_trigger(THD *thd, TABLE *table, const char *db,
+                          const char *tname, LEX_STRING *trigger_name,
+                          enum trg_event_type event,
+                          enum trg_action_time_type timing,
+                          LEX_STRING *trigger_stmt)
+{
+  CHARSET_INFO *cs= system_charset_info;
+  restore_record(table, s->default_values);
+  table->field[1]->store(db, strlen(db), cs);
+  table->field[2]->store(trigger_name->str, trigger_name->length, cs);
+  table->field[3]->store(trg_event_type_names[event].str,
+                         trg_event_type_names[event].length, cs);
+  table->field[5]->store(db, strlen(db), cs);
+  table->field[6]->store(tname, strlen(tname), cs);
+  table->field[9]->store(trigger_stmt->str, trigger_stmt->length, cs);
+  table->field[10]->store("ROW", 3, cs);
+  table->field[11]->store(trg_action_time_type_names[timing].str,
+                          trg_action_time_type_names[timing].length, cs);
+  table->field[14]->store("OLD", 3, cs);
+  table->field[15]->store("NEW", 3, cs);
+  return schema_table_store_record(thd, table);
+}
+
+
+static int get_schema_triggers_record(THD *thd, struct st_table_list *tables,
+				      TABLE *table, bool res,
+				      const char *base_name,
+				      const char *file_name)
+{
+  DBUG_ENTER("get_schema_triggers_record");
+  /*
+    res can be non zero value when processed table is a view or
+    error happened during opening of processed table.
+  */
+  if (res)
+  {
+    if (!tables->view)
+      push_warning(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
+                   thd->net.last_errno, thd->net.last_error);
+    thd->clear_error();
+    DBUG_RETURN(0);
+  }
+  if (!tables->view && tables->table->triggers)
+  {
+    Table_triggers_list *triggers= tables->table->triggers;
+    int event, timing;
+    for (event= 0; event < (int)TRG_EVENT_MAX; event++)
+    {
+      for (timing= 0; timing < (int)TRG_ACTION_MAX; timing++)
+      {
+        LEX_STRING trigger_name;
+        LEX_STRING trigger_stmt;
+        if (triggers->get_trigger_info(thd, (enum trg_event_type) event,
+                                       (enum trg_action_time_type)timing,
+                                       &trigger_name, &trigger_stmt))
+          continue;
+        if (store_trigger(thd, table, base_name, file_name, &trigger_name,
+                         (enum trg_event_type) event,
+                         (enum trg_action_time_type) timing, &trigger_stmt))
+          DBUG_RETURN(1);
+      }
+    }
+  }
+  DBUG_RETURN(0);
+}
+
+
 void store_key_column_usage(TABLE *table, const char*db, const char *tname,
                             const char *key_name, uint key_len, 
                             const char *con_type, uint con_len, longlong idx)
@@ -3869,6 +3952,29 @@ ST_FIELD_INFO open_tables_fields_info[]=
 };
 
 
+ST_FIELD_INFO triggers_fields_info[]=
+{
+  {"TRIGGER_CATALOG", FN_REFLEN, MYSQL_TYPE_STRING, 0, 1, 0},
+  {"TRIGGER_SCHEMA",NAME_LEN, MYSQL_TYPE_STRING, 0, 0, 0},
+  {"TRIGGER_NAME", NAME_LEN, MYSQL_TYPE_STRING, 0, 0, "Trigger"},
+  {"EVENT_MANIPULATION", 6, MYSQL_TYPE_STRING, 0, 0, "Event"},
+  {"EVENT_OBJECT_CATALOG", FN_REFLEN, MYSQL_TYPE_STRING, 0, 1, 0},
+  {"EVENT_OBJECT_SCHEMA",NAME_LEN, MYSQL_TYPE_STRING, 0, 0, 0},
+  {"EVENT_OBJECT_TABLE", NAME_LEN, MYSQL_TYPE_STRING, 0, 0, "Table"},
+  {"ACTION_ORDER", 4, MYSQL_TYPE_LONG, 0, 0, 0},
+  {"ACTION_CONDITION", 65535, MYSQL_TYPE_STRING, 0, 1, 0},
+  {"ACTION_STATEMENT", 65535, MYSQL_TYPE_STRING, 0, 0, "Statement"},
+  {"ACTION_ORIENTATION", 9, MYSQL_TYPE_STRING, 0, 0, 0},
+  {"ACTION_TIMING", 6, MYSQL_TYPE_STRING, 0, 0, "Timing"},
+  {"ACTION_REFERENCE_OLD_TABLE", NAME_LEN, MYSQL_TYPE_STRING, 0, 1, 0},
+  {"ACTION_REFERENCE_NEW_TABLE", NAME_LEN, MYSQL_TYPE_STRING, 0, 1, 0},
+  {"ACTION_REFERENCE_OLD_ROW", 3, MYSQL_TYPE_STRING, 0, 0, 0},
+  {"ACTION_REFERENCE_NEW_ROW", 3, MYSQL_TYPE_STRING, 0, 0, 0},
+  {"CREATED", 0, MYSQL_TYPE_TIMESTAMP, 0, 1, "Created"},
+  {0, 0, MYSQL_TYPE_STRING, 0, 0, 0}
+};
+
+
 ST_FIELD_INFO variables_fields_info[]=
 {
   {"Variable_name", 80, MYSQL_TYPE_STRING, 0, 0, "Variable_name"},
@@ -3919,6 +4025,8 @@ ST_SCHEMA_TABLE schema_tables[]=
    fill_open_tables, make_old_format, 0, -1, -1, 1},
   {"STATUS", variables_fields_info, create_schema_table, fill_status, 
    make_old_format, 0, -1, -1, 1},
+  {"TRIGGERS", triggers_fields_info, create_schema_table, 
+   get_all_tables, make_old_format, get_schema_triggers_record, 5, 6, 0},
   {"VARIABLES", variables_fields_info, create_schema_table, fill_variables,
    make_old_format, 0, -1, -1, 1},
   {0, 0, 0, 0, 0, 0, 0, 0, 0}
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 21348908bf3..7d0691455a0 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -23,6 +23,8 @@
 #include <hash.h>
 #include <myisam.h>
 #include <my_dir.h>
+#include "sp_head.h"
+#include "sql_trigger.h"
 
 #ifdef __WIN__
 #include <io.h>
@@ -291,16 +293,8 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
 	if (!(new_error=my_delete(path,MYF(MY_WME))))
         {
 	  some_tables_deleted=1;
-          /*
-            Destroy triggers for this table if there are any.
-
-            We won't need this as soon as we will have new .FRM format,
-            in which we will store trigger definitions in the same .FRM
-            files as table descriptions.
-          */
-          strmov(end, triggers_file_ext);
-          if (!access(path, F_OK))
-            new_error= my_delete(path, MYF(MY_WME));
+          new_error= Table_triggers_list::drop_all_triggers(thd, db,
+                                                            table->table_name);
         }
         error|= new_error;
       }
diff --git a/sql/sql_trigger.cc b/sql/sql_trigger.cc
index fd79fc8b878..a7aee95197e 100644
--- a/sql/sql_trigger.cc
+++ b/sql/sql_trigger.cc
@@ -39,6 +39,45 @@ static File_option triggers_file_parameters[]=
 
 
 /*
+  Structure representing contents of .TRN file which are used to support
+  database wide trigger namespace.
+*/
+
+struct st_trigname
+{
+  LEX_STRING trigger_table;
+};
+
+static const LEX_STRING trigname_file_type= {(char *)"TRIGGERNAME", 11};
+
+const char * const trigname_file_ext= ".TRN";
+
+static File_option trigname_file_parameters[]=
+{
+  {{(char*)"trigger_table", 15}, offsetof(struct st_trigname, trigger_table),
+   FILE_OPTIONS_ESTRING},
+  {{0, 0}, 0, FILE_OPTIONS_STRING}
+};
+
+
+const LEX_STRING trg_action_time_type_names[]=
+{
+  { (char *) STRING_WITH_LEN("BEFORE") },
+  { (char *) STRING_WITH_LEN("AFTER") }
+};
+
+const LEX_STRING trg_event_type_names[]=
+{
+  { (char *) STRING_WITH_LEN("INSERT") },
+  { (char *) STRING_WITH_LEN("UPDATE") },
+  { (char *) STRING_WITH_LEN("DELETE") }
+};
+
+
+static TABLE_LIST *add_table_for_trigger(THD *thd, sp_name *trig);
+
+
+/*
   Create or drop trigger for table.
 
   SYNOPSIS
@@ -69,6 +108,10 @@ bool mysql_create_or_drop_trigger(THD *thd, TABLE_LIST *tables, bool create)
     But do we want this ?
   */
 
+  if (!create &&
+      !(tables= add_table_for_trigger(thd, thd->lex->spname)))
+    DBUG_RETURN(TRUE);
+
   /* We should have only one table in table list. */
   DBUG_ASSERT(tables->next_global == 0);
 
@@ -174,28 +217,28 @@ bool Table_triggers_list::create_trigger(THD *thd, TABLE_LIST *tables)
 {
   LEX *lex= thd->lex;
   TABLE *table= tables->table;
-  char dir_buff[FN_REFLEN], file_buff[FN_REFLEN];
-  LEX_STRING dir, file;
+  char dir_buff[FN_REFLEN], file_buff[FN_REFLEN], trigname_buff[FN_REFLEN],
+       trigname_path[FN_REFLEN];
+  LEX_STRING dir, file, trigname_file;
   LEX_STRING *trg_def, *name;
   Item_trigger_field *trg_field;
-  List_iterator_fast<LEX_STRING> it(names_list);
+  struct st_trigname trigname;
 
-  /* We don't allow creation of several triggers of the same type yet */
-  if (bodies[lex->trg_chistics.event][lex->trg_chistics.action_time])
+
+  /* Trigger must be in the same schema as target table. */
+  if (my_strcasecmp(system_charset_info, table->s->db,
+                    lex->spname->m_db.str ? lex->spname->m_db.str :
+                                            thd->db))
   {
-    my_message(ER_TRG_ALREADY_EXISTS, ER(ER_TRG_ALREADY_EXISTS), MYF(0));
+    my_error(ER_TRG_IN_WRONG_SCHEMA, MYF(0));
     return 1;
   }
 
-  /* Let us check if trigger with the same name exists */
-  while ((name= it++))
+  /* We don't allow creation of several triggers of the same type yet */
+  if (bodies[lex->trg_chistics.event][lex->trg_chistics.action_time])
   {
-    if (my_strcasecmp(system_charset_info, lex->ident.str,
-                      name->str) == 0)
-    {
-      my_message(ER_TRG_ALREADY_EXISTS, ER(ER_TRG_ALREADY_EXISTS), MYF(0));
-      return 1;
-    }
+    my_message(ER_TRG_ALREADY_EXISTS, ER(ER_TRG_ALREADY_EXISTS), MYF(0));
+    return 1;
   }
 
   /*
@@ -234,6 +277,25 @@ bool Table_triggers_list::create_trigger(THD *thd, TABLE_LIST *tables)
   file.length=  strxnmov(file_buff, FN_REFLEN, tables->table_name,
                          triggers_file_ext, NullS) - file_buff;
   file.str= file_buff;
+  trigname_file.length= strxnmov(trigname_buff, FN_REFLEN,
+                                 lex->spname->m_name.str,
+                                 trigname_file_ext, NullS) - trigname_buff;
+  trigname_file.str= trigname_buff;
+  strxnmov(trigname_path, FN_REFLEN, dir_buff, trigname_buff, NullS);
+
+  /* Use the filesystem to enforce trigger namespace constraints. */
+  if (!access(trigname_path, F_OK))
+  {
+    my_error(ER_TRG_ALREADY_EXISTS, MYF(0));
+    return 1;
+  }
+
+  trigname.trigger_table.str= tables->table_name;
+  trigname.trigger_table.length= tables->table_name_length;
+
+  if (sql_create_definition_file(&dir, &trigname_file, &trigname_file_type,
+                                 (gptr)&trigname, trigname_file_parameters, 0))
+    return 1;
 
   /*
     Soon we will invalidate table object and thus Table_triggers_list object
@@ -246,13 +308,66 @@ bool Table_triggers_list::create_trigger(THD *thd, TABLE_LIST *tables)
   if (!(trg_def= (LEX_STRING *)alloc_root(&table->mem_root,
                                           sizeof(LEX_STRING))) ||
       definitions_list.push_back(trg_def, &table->mem_root))
-    return 1;
+    goto err_with_cleanup;
 
   trg_def->str= thd->query;
   trg_def->length= thd->query_length;
 
-  return sql_create_definition_file(&dir, &file, &triggers_file_type,
-                                    (gptr)this, triggers_file_parameters, 3);
+  if (!sql_create_definition_file(&dir, &file, &triggers_file_type,
+                                  (gptr)this, triggers_file_parameters, 3))
+    return 0;
+
+err_with_cleanup:
+  my_delete(trigname_path, MYF(MY_WME));
+  return 1;
+}
+
+
+/*
+  Deletes the .TRG file for a table
+
+  SYNOPSIS
+    rm_trigger_file()
+      path       - char buffer of size FN_REFLEN to be used
+                   for constructing path to .TRG file.
+      db         - table's database name
+      table_name - table's name
+
+  RETURN VALUE
+    False - success
+    True  - error
+*/
+
+static bool rm_trigger_file(char *path, char *db, char *table_name)
+{
+  strxnmov(path, FN_REFLEN, mysql_data_home, "/", db, "/", table_name,
+           triggers_file_ext, NullS);
+  unpack_filename(path, path);
+  return my_delete(path, MYF(MY_WME));
+}
+
+
+/*
+  Deletes the .TRN file for a trigger
+
+  SYNOPSIS
+    rm_trigname_file()
+      path       - char buffer of size FN_REFLEN to be used
+                   for constructing path to .TRN file.
+      db         - trigger's database name
+      table_name - trigger's name
+
+  RETURN VALUE
+    False - success
+    True  - error
+*/
+
+static bool rm_trigname_file(char *path, char *db, char *trigger_name)
+{
+  strxnmov(path, FN_REFLEN, mysql_data_home, "/", db, "/", trigger_name,
+           trigname_file_ext, NullS);
+  unpack_filename(path, path);
+  return my_delete(path, MYF(MY_WME));
 }
 
 
@@ -275,12 +390,13 @@ bool Table_triggers_list::drop_trigger(THD *thd, TABLE_LIST *tables)
   LEX_STRING *name;
   List_iterator_fast<LEX_STRING> it_name(names_list);
   List_iterator<LEX_STRING>      it_def(definitions_list);
+  char path[FN_REFLEN];
 
   while ((name= it_name++))
   {
     it_def++;
 
-    if (my_strcasecmp(system_charset_info, lex->ident.str,
+    if (my_strcasecmp(system_charset_info, lex->spname->m_name.str,
                       name->str) == 0)
     {
       /*
@@ -291,18 +407,14 @@ bool Table_triggers_list::drop_trigger(THD *thd, TABLE_LIST *tables)
 
       if (definitions_list.is_empty())
       {
-        char path[FN_REFLEN];
-
         /*
           TODO: Probably instead of removing .TRG file we should move
           to archive directory but this should be done as part of
           parse_file.cc functionality (because we will need it
           elsewhere).
         */
-        strxnmov(path, FN_REFLEN, mysql_data_home, "/", tables->db, "/",
-                 tables->table_name, triggers_file_ext, NullS);
-        unpack_filename(path, path);
-        return my_delete(path, MYF(MY_WME));
+        if (rm_trigger_file(path, tables->db, tables->table_name))
+          return 1;
       }
       else
       {
@@ -317,10 +429,15 @@ bool Table_triggers_list::drop_trigger(THD *thd, TABLE_LIST *tables)
                                triggers_file_ext, NullS) - file_buff;
         file.str= file_buff;
 
-        return sql_create_definition_file(&dir, &file, &triggers_file_type,
-                                          (gptr)this,
-                                          triggers_file_parameters, 3);
+        if (sql_create_definition_file(&dir, &file, &triggers_file_type,
+                                       (gptr)this, triggers_file_parameters,
+                                       3))
+          return 1;
       }
+
+      if (rm_trigname_file(path, tables->db, lex->spname->m_name.str))
+        return 1;
+      return 0;
     }
   }
 
@@ -331,8 +448,8 @@ bool Table_triggers_list::drop_trigger(THD *thd, TABLE_LIST *tables)
 
 Table_triggers_list::~Table_triggers_list()
 {
-  for (int i= 0; i < 3; i++)
-    for (int j= 0; j < 2; j++)
+  for (int i= 0; i < (int)TRG_EVENT_MAX; i++)
+    for (int j= 0; j < (int)TRG_ACTION_MAX; j++)
       delete bodies[i][j];
 
   if (record1_field)
@@ -389,13 +506,16 @@ bool Table_triggers_list::prepare_record1_accessors(TABLE *table)
       db         - table's database name
       table_name - table's name
       table      - pointer to table object
+      names_only - stop after loading trigger names
 
   RETURN VALUE
     False - success
     True  - error
 */
+
 bool Table_triggers_list::check_n_load(THD *thd, const char *db,
-                                       const char *table_name, TABLE *table)
+                                       const char *table_name, TABLE *table,
+                                       bool names_only)
 {
   char path_buff[FN_REFLEN];
   LEX_STRING path;
@@ -451,7 +571,7 @@ bool Table_triggers_list::check_n_load(THD *thd, const char *db,
         TODO: This could be avoided if there is no triggers
               for UPDATE and DELETE.
       */
-      if (triggers->prepare_record1_accessors(table))
+      if (!names_only && triggers->prepare_record1_accessors(table))
         DBUG_RETURN(1);
 
       List_iterator_fast<LEX_STRING> it(triggers->definitions_list);
@@ -471,32 +591,20 @@ bool Table_triggers_list::check_n_load(THD *thd, const char *db,
             Free lex associated resources
             QQ: Do we really need all this stuff here ?
           */
-          if (lex.sphead)
-          {
-            delete lex.sphead;
-            lex.sphead= 0;
-          }
+          delete lex.sphead;
           goto err_with_lex_cleanup;
         }
 
         triggers->bodies[lex.trg_chistics.event]
                              [lex.trg_chistics.action_time]= lex.sphead;
-        lex.sphead= 0;
-
-        if (!(trg_name_buff= alloc_root(&table->mem_root,
-                                        sizeof(LEX_STRING) +
-                                        lex.ident.length + 1)))
-          goto err_with_lex_cleanup;
-
-        trg_name_str= (LEX_STRING *)trg_name_buff;
-        trg_name_buff+= sizeof(LEX_STRING);
-        memcpy(trg_name_buff, lex.ident.str,
-               lex.ident.length + 1);
-        trg_name_str->str= trg_name_buff;
-        trg_name_str->length= lex.ident.length;
+        if (triggers->names_list.push_back(&lex.sphead->m_name, &table->mem_root))
+            goto err_with_lex_cleanup;
 
-        if (triggers->names_list.push_back(trg_name_str, &table->mem_root))
-          goto err_with_lex_cleanup;
+        if (names_only)
+        {
+          lex_end(&lex);
+          continue;
+        }
 
         /*
           Let us bind Item_trigger_field objects representing access to fields
@@ -537,3 +645,160 @@ err_with_lex_cleanup:
 
   DBUG_RETURN(1);
 }
+
+
+/*
+  Obtains and returns trigger metadata
+
+  SYNOPSIS
+    get_trigger_info()
+      thd       - current thread context
+      event     - trigger event type
+      time_type - trigger action time
+      name      - returns name of trigger
+      stmt      - returns statement of trigger
+
+  RETURN VALUE
+    False - success
+    True  - error
+*/
+
+bool Table_triggers_list::get_trigger_info(THD *thd, trg_event_type event,
+                                           trg_action_time_type time_type,
+                                           LEX_STRING *trigger_name,
+                                           LEX_STRING *trigger_stmt)
+{
+  sp_head *body;
+  DBUG_ENTER("get_trigger_info");
+  if ((body= bodies[event][time_type]))
+  {
+    *trigger_name= body->m_name;
+    *trigger_stmt= body->m_body;
+    DBUG_RETURN(0);
+  }
+  DBUG_RETURN(1);
+}
+
+
+/*
+  Find trigger's table from trigger identifier and add it to
+  the statement table list.
+
+  SYNOPSIS
+    mysql_table_for_trigger()
+      thd    - current thread context
+      trig   - identifier for trigger
+
+  RETURN VALUE
+    0 - error
+    # - pointer to TABLE_LIST object for the table
+*/
+
+static TABLE_LIST *add_table_for_trigger(THD *thd, sp_name *trig)
+{
+  const char *db= !trig->m_db.str ? thd->db : trig->m_db.str;
+  LEX *lex= thd->lex;
+  char path_buff[FN_REFLEN];
+  LEX_STRING path;
+  File_parser *parser;
+  struct st_trigname trigname;
+  DBUG_ENTER("add_table_for_trigger");
+
+  strxnmov(path_buff, FN_REFLEN, mysql_data_home, "/", db, "/",
+           trig->m_name.str, trigname_file_ext, NullS);
+  path.length= unpack_filename(path_buff, path_buff);
+  path.str= path_buff;
+
+  if (access(path_buff, F_OK))
+  {
+    my_error(ER_TRG_DOES_NOT_EXIST, MYF(0));
+    DBUG_RETURN(0);
+  }
+
+  if (!(parser= sql_parse_prepare(&path, thd->mem_root, 1)))
+    DBUG_RETURN(0);
+
+  if (strncmp(trigname_file_type.str, parser->type()->str,
+              parser->type()->length))
+  {
+    my_error(ER_WRONG_OBJECT, MYF(0), trig->m_name.str, trigname_file_ext,
+             "TRIGGERNAME");
+    DBUG_RETURN(0);
+  }
+
+  if (parser->parse((gptr)&trigname, thd->mem_root,
+                    trigname_file_parameters, 1))
+    DBUG_RETURN(0);
+
+  /* We need to reset statement table list to be PS/SP friendly. */
+  lex->query_tables= 0;
+  lex->query_tables_last= &lex->query_tables;
+  DBUG_RETURN(sp_add_to_query_tables(thd, lex, db,
+                                     trigname.trigger_table.str, TL_WRITE));
+}
+
+
+/*
+  Drop all triggers for table.
+
+  SYNOPSIS
+    drop_all_triggers()
+      thd    - current thread context
+      db     - schema for table
+      name   - name for table
+
+  NOTE
+    The calling thread should hold the LOCK_open mutex;
+
+  RETURN VALUE
+    False - success
+    True  - error
+*/
+
+bool Table_triggers_list::drop_all_triggers(THD *thd, char *db, char *name)
+{
+  TABLE table;
+  char path[FN_REFLEN];
+  bool result= 0;
+  DBUG_ENTER("drop_all_triggers");
+
+  bzero(&table, sizeof(table));
+  init_alloc_root(&table.mem_root, 8192, 0);
+
+  safe_mutex_assert_owner(&LOCK_open);
+
+  if (Table_triggers_list::check_n_load(thd, db, name, &table, 1))
+  {
+    result= 1;
+    goto end;
+  }
+  if (table.triggers)
+  {
+    LEX_STRING *trigger;
+    List_iterator_fast<LEX_STRING> it_name(table.triggers->names_list);
+
+    while ((trigger= it_name++))
+    {
+      if (rm_trigname_file(path, db, trigger->str))
+      {
+        /*
+          Instead of immediately bailing out with error if we were unable
+          to remove .TRN file we will try to drop other files.
+        */
+        result= 1;
+        continue;
+      }
+    }
+
+    if (rm_trigger_file(path, db, name))
+    {
+      result= 1;
+      goto end;
+    }
+  }
+end:
+  if (table.triggers)
+    delete table.triggers;
+  free_root(&table.mem_root, MYF(0));
+  DBUG_RETURN(result);
+}
diff --git a/sql/sql_trigger.h b/sql/sql_trigger.h
index 044219d5ac9..e751741fa34 100644
--- a/sql/sql_trigger.h
+++ b/sql/sql_trigger.h
@@ -23,7 +23,7 @@
 class Table_triggers_list: public Sql_alloc
 {
   /* Triggers as SPs grouped by event, action_time */
-  sp_head           *bodies[3][2];
+  sp_head           *bodies[TRG_EVENT_MAX][TRG_ACTION_MAX];
   /*
     Copy of TABLE::Field array with field pointers set to TABLE::record[1]
     buffer instead of TABLE::record[0] (used for OLD values in on UPDATE
@@ -121,9 +121,13 @@ public:
 
     return res;
   }
+  bool get_trigger_info(THD *thd, trg_event_type event,
+                        trg_action_time_type time_type,
+                        LEX_STRING *trigger_name, LEX_STRING *trigger_stmt);
 
   static bool check_n_load(THD *thd, const char *db, const char *table_name,
-                           TABLE *table);
+                           TABLE *table, bool names_only);
+  static bool drop_all_triggers(THD *thd, char *db, char *table_name);
 
   bool has_delete_triggers()
   {
@@ -143,3 +147,6 @@ public:
 private:
   bool prepare_record1_accessors(TABLE *table);
 };
+
+extern const LEX_STRING trg_action_time_type_names[];
+extern const LEX_STRING trg_event_type_names[];
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index e4259197292..d4cd2bd6600 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -45,7 +45,7 @@ int yylex(void *yylval, void *yythd);
 
 const LEX_STRING null_lex_str={0,0};
 
-#define yyoverflow(A,B,C,D,E,F) {ulong val= *(F); if(my_yyoverflow((B), (D), &val)) { yyerror((char*) (A)); return 2; } else { *(F)= (YYSIZE_T)val; }}
+#define yyoverflow(A,B,C,D,E,F) {ulong val= *(F); if (my_yyoverflow((B), (D), &val)) { yyerror((char*) (A)); return 2; } else { *(F)= (YYSIZE_T)val; }}
 
 #define WARN_DEPRECATED(A,B)                                        \
   push_warning_printf(((THD *)yythd), MYSQL_ERROR::WARN_LEVEL_WARN, \
@@ -610,6 +610,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
 %token  TRAILING
 %token  TRANSACTION_SYM
 %token  TRIGGER_SYM
+%token  TRIGGERS_SYM
 %token  TRIM
 %token  TRUE_SYM
 %token  TRUNCATE_SYM
@@ -1277,7 +1278,7 @@ create:
 	  }
 	  opt_view_list AS select_init check_option
 	  {}
-        | CREATE TRIGGER_SYM ident trg_action_time trg_event 
+        | CREATE TRIGGER_SYM sp_name trg_action_time trg_event 
           ON table_ident FOR_SYM EACH_SYM ROW_SYM
           {
             LEX *lex= Lex;
@@ -1296,6 +1297,7 @@ create:
             
             sp->m_type= TYPE_ENUM_TRIGGER;
             lex->sphead= sp;
+            lex->spname= $3;
             /*
               We have to turn of CLIENT_MULTI_QUERIES while parsing a
               stored procedure, otherwise yylex will chop it into pieces
@@ -1306,7 +1308,7 @@ create:
             
             bzero((char *)&lex->sp_chistics, sizeof(st_sp_chistics));
             lex->sphead->m_chistics= &lex->sp_chistics;
-            lex->sphead->m_body_begin= lex->tok_start;
+            lex->sphead->m_body_begin= lex->ptr;
           }
           sp_proc_stmt
           {
@@ -1314,14 +1316,12 @@ create:
             sp_head *sp= lex->sphead;
             
             lex->sql_command= SQLCOM_CREATE_TRIGGER;
-            sp->init_strings(YYTHD, lex, NULL);
+            sp->init_strings(YYTHD, lex, $3);
             /* Restore flag if it was cleared above */
             if (sp->m_old_cmq)
               YYTHD->client_capabilities |= CLIENT_MULTI_QUERIES;
             sp->restore_thd_mem_root(YYTHD);
 
-            lex->ident= $3;
-
             /*
               We have to do it after parsing trigger body, because some of
               sp_proc_stmt alternatives are not saving/restoring LEX, so
@@ -4455,7 +4455,7 @@ select_into:
 select_from:
 	  FROM join_table_list where_clause group_clause having_clause
 	       opt_order_clause opt_limit_clause procedure_clause
-        | FROM DUAL_SYM opt_limit_clause
+        | FROM DUAL_SYM where_clause opt_limit_clause
           /* oracle compatibility: oracle always requires FROM clause,
              and DUAL is system table without fields.
              Is "SELECT 1 FROM DUAL" any better than "SELECT 1" ?
@@ -4656,9 +4656,25 @@ bool_pri:
 
 predicate:
 	 bit_expr IN_SYM '(' expr_list ')'
-	  { $4->push_front($1); $$= new Item_func_in(*$4); }
+	  { 
+            if ($4->elements == 1)
+              $$= new Item_func_eq($1, $4->head());
+            else
+            {
+              $4->push_front($1);
+              $$= new Item_func_in(*$4);
+            }
+          }
 	| bit_expr not IN_SYM '(' expr_list ')'
-	  { $5->push_front($1); $$= negate_expression(YYTHD, new Item_func_in(*$5)); }
+          {
+            if ($5->elements == 1)
+              $$= new Item_func_ne($1, $5->head());
+            else
+            {
+              $5->push_front($1);
+              $$= negate_expression(YYTHD, new Item_func_in(*$5));
+            }            
+          }
         | bit_expr IN_SYM in_subselect
 	  { $$= new Item_in_subselect($1, $3); }
 	| bit_expr not IN_SYM in_subselect
@@ -5235,7 +5251,7 @@ simple_expr:
 	| UNIX_TIMESTAMP '(' expr ')'
 	  { $$= new Item_func_unix_timestamp($3); }
 	| USER '(' ')'
-	  { $$= new Item_func_user(); Lex->safe_to_cache_query=0; }
+	  { $$= new Item_func_user(FALSE); Lex->safe_to_cache_query=0; }
 	| UTC_DATE_SYM optional_braces
 	  { $$= new Item_func_curdate_utc(); Lex->safe_to_cache_query=0;}
 	| UTC_TIME_SYM optional_braces
@@ -6332,19 +6348,11 @@ drop:
 	    lex->sql_command= SQLCOM_DROP_VIEW;
 	    lex->drop_if_exists= $3;
 	  }
-        | DROP TRIGGER_SYM ident '.' ident
+        | DROP TRIGGER_SYM sp_name
           {
             LEX *lex= Lex;
-
             lex->sql_command= SQLCOM_DROP_TRIGGER;
-            /* QQ: Could we loosen lock type in certain cases ? */
-            if (!lex->select_lex.add_table_to_list(YYTHD,
-                                                   new Table_ident($3),
-                                                   (LEX_STRING*) 0,
-                                                   TL_OPTION_UPDATING,
-                                                   TL_WRITE))
-              YYABORT;
-            lex->ident= $5;
+            lex->spname= $3;
           }
 	;
 
@@ -6709,6 +6717,15 @@ show_param:
              if (prepare_schema_table(YYTHD, lex, 0, SCH_TABLE_NAMES))
                YYABORT;
            }
+         | opt_full TRIGGERS_SYM opt_db wild_and_where
+           {
+             LEX *lex= Lex;
+             lex->sql_command= SQLCOM_SELECT;
+             lex->orig_sql_command= SQLCOM_SHOW_TRIGGERS;
+             lex->select_lex.db= $3;
+             if (prepare_schema_table(YYTHD, lex, 0, SCH_TRIGGERS))
+               YYABORT;
+           }
          | TABLE_SYM STATUS_SYM opt_db wild_and_where
            {
              LEX *lex= Lex;
@@ -8012,6 +8029,7 @@ keyword_sp:
 	| TEXT_SYM		{}
 	| THAN_SYM		{}
 	| TRANSACTION_SYM	{}
+	| TRIGGERS_SYM		{}
 	| TIMESTAMP		{}
 	| TIMESTAMP_ADD		{}
 	| TIMESTAMP_DIFF	{}
diff --git a/sql/table.h b/sql/table.h
index 6ba9453b2f0..8c0040b0fa4 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -289,7 +289,7 @@ enum enum_schema_tables
   SCH_COLLATION_CHARACTER_SET_APPLICABILITY, SCH_PROCEDURES, SCH_STATISTICS,
   SCH_VIEWS, SCH_USER_PRIVILEGES, SCH_SCHEMA_PRIVILEGES, SCH_TABLE_PRIVILEGES,
   SCH_COLUMN_PRIVILEGES, SCH_TABLE_CONSTRAINTS, SCH_KEY_COLUMN_USAGE,
-  SCH_TABLE_NAMES, SCH_OPEN_TABLES, SCH_STATUS, SCH_VARIABLES
+  SCH_TABLE_NAMES, SCH_OPEN_TABLES, SCH_STATUS, SCH_TRIGGERS, SCH_VARIABLES
 };