merge with 4.1

BitKeeper/etc/ignore:
  auto-union
BitKeeper/etc/logging_ok:
  auto-union
BitKeeper/triggers/post-commit:
  Auto merged
Docs/Support/texi2html:
  Auto merged
Makefile.am:
  Auto merged
client/Makefile.am:
  Auto merged
client/mysql.cc:
  Auto merged
client/mysqldump.c:
  Auto merged
include/my_base.h:
  Auto merged
include/my_global.h:
  Auto merged
include/my_pthread.h:
  Auto merged
include/my_sys.h:
  Auto merged
include/my_time.h:
  Auto merged
include/mysql.h:
  Auto merged
include/mysql_com.h:
  Auto merged
innobase/buf/buf0buf.c:
  Auto merged
innobase/include/row0mysql.h:
  Auto merged
innobase/row/row0sel.c:
  Auto merged
libmysql/libmysql.c:
  Auto merged
libmysqld/examples/Makefile.am:
  Auto merged
myisam/mi_check.c:
  Auto merged
mysql-test/include/ps_modify.inc:
  Auto merged
mysql-test/install_test_db.sh:
  Auto merged
mysql-test/r/alter_table.result:
  Auto merged
mysql-test/r/auto_increment.result:
  Auto merged
mysql-test/r/bdb.result:
  Auto merged
mysql-test/r/ctype_latin1_de.result:
  Auto merged
mysql-test/r/ctype_recoding.result:
  Auto merged
mysql-test/r/fulltext.result:
  Auto merged
mysql-test/r/func_gconcat.result:
  Auto merged
mysql-test/r/func_group.result:
  Auto merged
mysql-test/r/func_if.result:
  Auto merged
mysql-test/t/derived.test:
  Auto merged
mysql-test/t/insert.test:
  merge with 4.1
  Fixed test case to not use 'if exists' when it shouldn't
mysql-test/t/range.test:
  merge with 4.1
  Added missing drop table
sql/ha_ndbcluster.cc:
  merge with 4.1
  Simple optimization: use max() instead of ? :
sql/item_func.cc:
  merge with 4.1
  (Added back old variable names for easier merges)
sql/opt_range.cc:
  merge with 4.1
  Removed argument 'parent_alloc' from QUICK_RANGE_SELECT as this was not used
  Added assert if using QUICK_GROUP_MIN_MAX_SELECT with parent_alloc as the init() function can't handle this
  Changed back get_quick_select_for_ref() to use it's own alloc root becasue this function may be called several times for one query
sql/sql_handler.cc:
  merge with 4.1
  change variable 'err' to 'error' as same function had a label named 'err'
sql/sql_update.cc:
  Use multi-update code from 5.0 instead of 4.1
  We will fix the locking code shortly in 5.0 to be faster than in 4.1

1 files changed, 453 insertions, 206 deletions

diff --git a/sql/sql_handler.cc b/sql/sql_handler.cc
index 38bc4756f81..61e57362bd4 100644
--- a/sql/sql_handler.cc
+++ b/sql/sql_handler.cc
@@ -1,5 +1,4 @@
-/* Copyright (C) 2000-2003 MySQL AB
-
+/* Copyright (C) 2000-2004 MySQL AB
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation; either version 2 of the License, or
@@ -17,9 +16,6 @@
 
 /* HANDLER ... commands - direct access to ISAM */
 
-#include "mysql_priv.h"
-#include "sql_select.h"
-
 /* TODO:
   HANDLER blabla OPEN [ AS foobar ] [ (column-list) ]
 
@@ -37,195 +33,391 @@
   all the sql_alloc'ed memory. It's harder to work around...
 */
 
+/*
+  There are two containers holding information about open handler tables.
+  The first is 'thd->handler_tables'. It is a linked list of TABLE objects.
+  It is used like 'thd->open_tables' in the table cache. The trick is to
+  exchange these two lists during open and lock of tables. Thus the normal
+  table cache code can be used.
+  The second container is a HASH. It holds objects of the type TABLE_LIST.
+  Despite its name, no lists of tables but only single structs are hashed
+  (the 'next' pointer is always NULL). The reason for theis second container
+  is, that we want handler tables to survive FLUSH TABLE commands. A table
+  affected by FLUSH TABLE must be closed so that other threads are not
+  blocked by handler tables still in use. Since we use the normal table cache
+  functions with 'thd->handler_tables', the closed tables are removed from
+  this list. Hence we need the original open information for the handler
+  table in the case that it is used again. This information is handed over
+  to mysql_ha_open() as a TABLE_LIST. So we store this information in the
+  second container, where it is not affected by FLUSH TABLE. The second
+  container is implemented as a hash for performance reasons. Consequently,
+  we use it not only for re-opening a handler table, but also for the
+  HANDLER ... READ commands. For this purpose, we store a pointer to the
+  TABLE structure (in the first container) in the TBALE_LIST object in the
+  second container. When the table is flushed, the pointer is cleared.
+*/
+
+#include "mysql_priv.h"
+#include "sql_select.h"
+#include <assert.h>
+
+#define HANDLER_TABLES_HASH_SIZE 120
+
+static enum enum_ha_read_modes rkey_to_rnext[]=
+{ RNEXT_SAME, RNEXT, RPREV, RNEXT, RPREV, RNEXT, RPREV, RPREV };
+
 #define HANDLER_TABLES_HACK(thd) {      \
   TABLE *tmp=thd->open_tables;          \
   thd->open_tables=thd->handler_tables; \
   thd->handler_tables=tmp; }
 
-static TABLE **find_table_ptr_by_name(THD *thd,const char *db,
-                                      const char *table_name,
-                                      bool is_alias, bool dont_lock,
-                                      bool *was_flushed);
+static int mysql_ha_flush_table(THD *thd, TABLE **table_ptr, uint mode_flags);
+
+
+/*
+  Get hash key and hash key length.
+
+  SYNOPSIS
+    mysql_ha_hash_get_key()
+    tables                      Pointer to the hash object.
+    key_len_p   (out)           Pointer to the result for key length.
+    first                       Unused.
+
+  DESCRIPTION
+    The hash object is an TABLE_LIST struct.
+    The hash key is the alias name.
+    The hash key length is the alias name length plus one for the
+    terminateing NUL character.
+
+  RETURN
+    Pointer to the TABLE_LIST struct.
+*/
 
-int mysql_ha_open(THD *thd, TABLE_LIST *tables)
+static char *mysql_ha_hash_get_key(TABLE_LIST *tables, uint *key_len_p,
+                                   my_bool first __attribute__((unused)))
 {
-  HANDLER_TABLES_HACK(thd);
-  uint counter;
+  *key_len_p= strlen(tables->alias) + 1 ; /* include '\0' in comparisons */
+  return tables->alias;
+}
 
-  /* for now HANDLER can be used only for real TABLES */
-  tables->required_type= FRMTYPE_TABLE;
-  int err=open_tables(thd, tables, &counter);
 
-  HANDLER_TABLES_HACK(thd);
-  if (err)
-    return -1;
+/*
+  Free an hash object.
 
-  // there can be only one table in *tables
-  if (!(tables->table->file->table_flags() & HA_CAN_SQL_HANDLER))
-  {
-    my_printf_error(ER_ILLEGAL_HA,ER(ER_ILLEGAL_HA),MYF(0), tables->alias);
-    mysql_ha_close(thd, tables,1);
-    return -1;
-  }
+  SYNOPSIS
+    mysql_ha_hash_free()
+    tables                      Pointer to the hash object.
 
-  send_ok(thd);
-  return 0;
+  DESCRIPTION
+    The hash object is an TABLE_LIST struct.
+
+  RETURN
+    Nothing
+*/
+
+static void mysql_ha_hash_free(TABLE_LIST *tables)
+{
+  my_free((char*) tables, MYF(0));
 }
 
 
 /*
-  Close a HANDLER table.
+  Open a HANDLER table.
 
   SYNOPSIS
-    mysql_ha_close()
+    mysql_ha_open()
     thd                         Thread identifier.
-    tables                      A list of tables with the first entry to close.
-    dont_send_ok                Suppresses the commands' ok message and
-                                error message and error return.
-    dont_lock                   Suppresses the normal locking of LOCK_open.
+    tables                      A list of tables with the first entry to open.
+    reopen                      Re-open a previously opened handler table.
 
   DESCRIPTION
     Though this function takes a list of tables, only the first list entry
-    will be closed. Broadcasts a COND_refresh condition.
-    If mysql_ha_close() is not called from the parser, 'dont_send_ok'
-    must be set.
-    If the caller did already lock LOCK_open, it must set 'dont_lock'.
-
-  IMPLEMENTATION
-    find_table_ptr_by_name() closes the table, if a FLUSH TABLE is outstanding.
-    It returns a NULL pointer in this case, but flags the situation in
-    'was_flushed'. In that case the normal ER_UNKNOWN_TABLE error messages
-    is suppressed.
+    will be opened.
+    'reopen' is set when a handler table is to be re-opened. In this case,
+    'tables' is the pointer to the hashed TABLE_LIST object which has been
+    saved on the original open.
+    'reopen' is also used to suppress the sending of an 'ok' message or
+    error messages.
 
   RETURN
-    0  ok
-    -1 error
+    0    ok
+    != 0 error
 */
 
-int mysql_ha_close(THD *thd, TABLE_LIST *tables,
-                   bool dont_send_ok, bool dont_lock, bool no_alias)
+int mysql_ha_open(THD *thd, TABLE_LIST *tables, bool reopen)
 {
-  TABLE         **table_ptr;
-  bool          was_flushed;
-
-  table_ptr= find_table_ptr_by_name(thd, tables->db, tables->alias,
-                                    !no_alias, dont_lock, &was_flushed);
-  if (*table_ptr)
+  TABLE_LIST    *hash_tables;
+  char          *db, *name, *alias;
+  uint          dblen, namelen, aliaslen, counter;
+  int           error;
+  DBUG_ENTER("mysql_ha_open");
+  DBUG_PRINT("enter",("'%s'.'%s' as '%s'  reopen: %d",
+                      tables->db, tables->real_name, tables->alias,
+                      (int) reopen));
+
+  if (! hash_inited(&thd->handler_tables_hash))
   {
-    (*table_ptr)->file->ha_index_or_rnd_end();
-    if (!dont_lock)
-      VOID(pthread_mutex_lock(&LOCK_open));
-    if (close_thread_table(thd, table_ptr))
+    /*
+      HASH entries are of type TABLE_LIST.
+    */
+    if (hash_init(&thd->handler_tables_hash, &my_charset_latin1,
+                  HANDLER_TABLES_HASH_SIZE, 0, 0,
+                  (hash_get_key) mysql_ha_hash_get_key,
+                  (hash_free_key) mysql_ha_hash_free, 0))
+      goto err;
+  }
+  else if (! reopen) /* Otherwise we have 'tables' already. */
+  {
+    if (hash_search(&thd->handler_tables_hash, (byte*) tables->alias,
+                    strlen(tables->alias) + 1))
     {
-      /* Tell threads waiting for refresh that something has happened */
-      VOID(pthread_cond_broadcast(&COND_refresh));
+      DBUG_PRINT("info",("duplicate '%s'", tables->alias));
+      if (! reopen)
+        my_printf_error(ER_NONUNIQ_TABLE, ER(ER_NONUNIQ_TABLE),
+                        MYF(0), tables->alias);
+      goto err;
     }
-    if (!dont_lock)
-      VOID(pthread_mutex_unlock(&LOCK_open));
   }
-  else if (!was_flushed && !dont_send_ok)
+
+  /*
+    open_tables() will set 'tables->table' if successful.
+    It must be NULL for a real open when calling open_tables().
+  */
+  DBUG_ASSERT(! tables->table);
+  HANDLER_TABLES_HACK(thd);
+
+  /* for now HANDLER can be used only for real TABLES */
+  tables->required_type= FRMTYPE_TABLE;
+  error= open_tables(thd, tables, &counter);
+
+  HANDLER_TABLES_HACK(thd);
+  if (error)
+    goto err;
+
+  /* There can be only one table in '*tables'. */
+  if (! (tables->table->file->table_flags() & HA_CAN_SQL_HANDLER))
+  {
+    if (! reopen)
+      my_printf_error(ER_ILLEGAL_HA,ER(ER_ILLEGAL_HA),MYF(0), tables->alias);
+    mysql_ha_close(thd, tables);
+    goto err;
+  }
+
+  if (! reopen)
   {
-    my_printf_error(ER_UNKNOWN_TABLE, ER(ER_UNKNOWN_TABLE), MYF(0),
-                    tables->alias, "HANDLER");
-    return -1;
+    /* copy the TABLE_LIST struct */
+    dblen= strlen(tables->db) + 1;
+    namelen= strlen(tables->real_name) + 1;
+    aliaslen= strlen(tables->alias) + 1;
+    if (!(my_multi_malloc(MYF(MY_WME),
+                          &hash_tables, sizeof(*hash_tables),
+                          &db, dblen,
+                          &name, namelen,
+                          &alias, aliaslen,
+                          NullS)))
+      goto err;
+    /* structure copy */
+    *hash_tables= *tables;
+    hash_tables->db= db;
+    hash_tables->real_name= name;
+    hash_tables->alias= alias;
+    memcpy(hash_tables->db, tables->db, dblen);
+    memcpy(hash_tables->real_name, tables->real_name, namelen);
+    memcpy(hash_tables->alias, tables->alias, aliaslen);
+
+    /* add to hash */
+    if (my_hash_insert(&thd->handler_tables_hash, (byte*) hash_tables))
+    {
+      mysql_ha_close(thd, tables);
+      goto err;
+    }
   }
-  if (!dont_send_ok)
+
+  if (! reopen)
     send_ok(thd);
-  return 0;
+  DBUG_PRINT("exit",("OK"));
+  DBUG_RETURN(0);
+
+err:
+  DBUG_PRINT("exit",("ERROR"));
+  DBUG_RETURN(-1);
 }
 
 
 /*
-  Close a list of HANDLER tables.
+  Close a HANDLER table.
 
   SYNOPSIS
-    mysql_ha_close_list()
+    mysql_ha_close()
     thd                         Thread identifier.
-    tables                      The list of tables to close. If NULL,
-                                close all HANDLER tables.
-    flushed                     Close only tables which are marked flushed.
-                                Used only if tables is NULL.
+    tables                      A list of tables with the first entry to close.
 
   DESCRIPTION
-    The list of HANDLER tables may be NULL, in which case all HANDLER
-    tables are closed. Broadcasts a COND_refresh condition, for
-    every table closed. If 'tables' is NULL and 'flushed' is set,
-    all HANDLER tables marked for flush are closed.
-    The caller must lock LOCK_open.
-
-  IMPLEMENTATION
-    find_table_ptr_by_name() closes the table, if it is marked for flush.
-    It returns a NULL pointer in this case, but flags the situation in
-    'was_flushed'. In that case the normal ER_UNKNOWN_TABLE error messages
-    is suppressed.
+    Though this function takes a list of tables, only the first list entry
+    will be closed. Broadcasts a COND_refresh condition.
 
   RETURN
-    0  ok
+    0    ok
+    != 0 error
 */
 
-int mysql_ha_close_list(THD *thd, TABLE_LIST *tables, bool flushed)
+int mysql_ha_close(THD *thd, TABLE_LIST *tables)
 {
-  TABLE_LIST    *tl_item;
+  TABLE_LIST    *hash_tables;
   TABLE         **table_ptr;
+  DBUG_ENTER("mysql_ha_close");
+  DBUG_PRINT("enter",("'%s'.'%s' as '%s'",
+                      tables->db, tables->real_name, tables->alias));
 
-  if (tables)
+  if ((hash_tables= (TABLE_LIST*) hash_search(&thd->handler_tables_hash,
+                                              (byte*) tables->alias,
+                                              strlen(tables->alias) + 1)))
   {
-    for (tl_item= tables ; tl_item; tl_item= tl_item->next_local)
+    /*
+      Though we could take the table pointer from hash_tables->table,
+      we must follow the thd->handler_tables chain anyway, as we need the
+      address of the 'next' pointer referencing this table
+      for close_thread_table().
+    */
+    for (table_ptr= &(thd->handler_tables);
+         *table_ptr && (*table_ptr != hash_tables->table);
+           table_ptr= &(*table_ptr)->next)
+      ;
+
+    if (*table_ptr)
     {
-      mysql_ha_close(thd, tl_item, /*dont_send_ok*/ 1,
-                     /*dont_lock*/ 1, /*no_alias*/ 1);
+      (*table_ptr)->file->ha_index_or_rnd_end();
+      VOID(pthread_mutex_lock(&LOCK_open));
+      table->file->ha_index_or_rnd_end();
+      if (close_thread_table(thd, table_ptr))
+      {
+        /* Tell threads waiting for refresh that something has happened */
+        VOID(pthread_cond_broadcast(&COND_refresh));
+      }
+      VOID(pthread_mutex_unlock(&LOCK_open));
     }
+    hash_delete(&thd->handler_tables_hash, (byte*) hash_tables);
   }
   else
   {
-    table_ptr= &(thd->handler_tables);
-    while (*table_ptr)
-    {
-      if (! flushed || ((*table_ptr)->version != refresh_version))
-      {
-	(*table_ptr)->file->ha_index_or_rnd_end();
-        if (close_thread_table(thd, table_ptr))
-        {
-          /* Tell threads waiting for refresh that something has happened */
-          VOID(pthread_cond_broadcast(&COND_refresh));
-        }
-        continue;
-      }
-      table_ptr= &((*table_ptr)->next);
-    }
+    my_printf_error(ER_UNKNOWN_TABLE, ER(ER_UNKNOWN_TABLE), MYF(0),
+                    tables->alias, "HANDLER");
+    DBUG_PRINT("exit",("ERROR"));
+    DBUG_RETURN(-1);
   }
-  return 0;
+
+  send_ok(thd);
+  DBUG_PRINT("exit", ("OK"));
+  DBUG_RETURN(0);
 }
 
-static enum enum_ha_read_modes rkey_to_rnext[]=
-{ RNEXT_SAME, RNEXT, RPREV, RNEXT, RPREV, RNEXT, RPREV, RPREV };
 
+/*
+  Read from a HANDLER table.
+
+  SYNOPSIS
+    mysql_ha_read()
+    thd                         Thread identifier.
+    tables                      A list of tables with the first entry to read.
+    mode
+    keyname
+    key_expr
+    ha_rkey_mode
+    cond
+    select_limit
+    offset_limit
 
+  RETURN
+    0    ok
+    != 0 error
+*/
+ 
 int mysql_ha_read(THD *thd, TABLE_LIST *tables,
                   enum enum_ha_read_modes mode, char *keyname,
                   List<Item> *key_expr,
                   enum ha_rkey_function ha_rkey_mode, Item *cond,
                   ha_rows select_limit,ha_rows offset_limit)
 {
-  int err, keyno=-1;
-  bool was_flushed;
-  TABLE *table= *find_table_ptr_by_name(thd, tables->db, tables->alias,
-                                        /*is_alias*/ 1, /*dont_lock*/ 0,
-                                        &was_flushed);
+  TABLE_LIST    *hash_tables;
+  TABLE         *table;
+  MYSQL_LOCK    *lock;
+  List<Item>	list;
+  Protocol	*protocol= thd->protocol;
+  char		buff[MAX_FIELD_WIDTH];
+  String	buffer(buff, sizeof(buff), system_charset_info);
+  int           error, keyno= -1;
+  uint          num_rows;
+  byte		*key;
+  uint		key_len;
+  DBUG_ENTER("mysql_ha_read");
+  DBUG_PRINT("enter",("'%s'.'%s' as '%s'",
+                      tables->db, tables->real_name, tables->alias));
+
+  LINT_INIT(key);
+  LINT_INIT(key_len);
+
+  list.push_front(new Item_field(NULL,NULL,"*"));
+  List_iterator<Item> it(list);
+  it++;
+
+  if ((hash_tables= (TABLE_LIST*) hash_search(&thd->handler_tables_hash,
+                                              (byte*) tables->alias,
+                                              strlen(tables->alias) + 1)))
+  {
+    table= hash_tables->table;
+    DBUG_PRINT("info-in-hash",("'%s'.'%s' as '%s' tab %p",
+                               hash_tables->db, hash_tables->real_name,
+                               hash_tables->alias, table));
+    if (!table)
+    {
+      /*
+        The handler table has been closed. Re-open it.
+      */
+      if (mysql_ha_open(thd, hash_tables, 1))
+      {
+        DBUG_PRINT("exit",("reopen failed"));
+        goto err0;
+      }
+
+      table= hash_tables->table;
+      DBUG_PRINT("info",("re-opened '%s'.'%s' as '%s' tab %p",
+                         hash_tables->db, hash_tables->real_name,
+                         hash_tables->alias, table));
+    }
+
+#if MYSQL_VERSION_ID < 40100
+    if (*tables->db && strcmp(table->table_cache_key, tables->db))
+    {
+      DBUG_PRINT("info",("wrong db"));
+      table= NULL;
+    }
+#endif
+  }
+  else
+    table= NULL;
+
   if (!table)
   {
-    my_printf_error(ER_UNKNOWN_TABLE,ER(ER_UNKNOWN_TABLE),MYF(0),
-		    tables->alias,"HANDLER");
-    return -1;
+#if MYSQL_VERSION_ID < 40100
+    char buff[MAX_DBKEY_LENGTH];
+    if (*tables->db)
+      strxnmov(buff, sizeof(buff), tables->db, ".", tables->real_name, NullS);
+    else
+      strncpy(buff, tables->alias, sizeof(buff));
+    my_printf_error(ER_UNKNOWN_TABLE, ER(ER_UNKNOWN_TABLE), MYF(0),
+                    buff, "HANDLER");
+#else
+    my_printf_error(ER_UNKNOWN_TABLE, ER(ER_UNKNOWN_TABLE), MYF(0),
+                    tables->alias, "HANDLER");
+#endif
+    goto err0;
   }
   tables->table=table;
 
   if (cond && (cond->fix_fields(thd, tables, &cond) || cond->check_cols(1)))
-    return -1;
-
-  /* InnoDB needs to know that this table handle is used in the HANDLER */
+    goto err0;
 
-  table->file->init_table_handle_for_HANDLER();
+  table->file->init_table_handle_for_HANDLER(); // Only InnoDB requires it
 
   if (keyname)
   {
@@ -233,34 +425,22 @@ int mysql_ha_read(THD *thd, TABLE_LIST *tables,
     {
       my_printf_error(ER_KEY_DOES_NOT_EXITS,ER(ER_KEY_DOES_NOT_EXITS),MYF(0),
                       keyname,tables->alias);
-      return -1;
+      goto err0;
     }
     table->file->ha_index_or_rnd_end();
     table->file->ha_index_init(keyno);
   }
 
-  List<Item> list;
-  list.push_front(new Item_field(NULL,NULL,"*"));
-  List_iterator<Item> it(list);
-  Protocol *protocol= thd->protocol;
-  char buff[MAX_FIELD_WIDTH];
-  String buffer(buff, sizeof(buff), system_charset_info);
-  uint num_rows;
-  byte *key;
-  uint key_len;
-  LINT_INIT(key);
-  LINT_INIT(key_len);
-
-  it++;                                         // Skip first NULL field
-
-  insert_fields(thd, tables, tables->db, tables->alias, &it, 0, 0);
+  if (insert_fields(thd, tables, tables->db, tables->alias, &it, 0, 0))
+    goto err0;
 
   select_limit+=offset_limit;
   protocol->send_fields(&list, Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF);
 
   HANDLER_TABLES_HACK(thd);
-  MYSQL_LOCK *lock=mysql_lock_tables(thd,&tables->table,1);
+  lock= mysql_lock_tables(thd, &tables->table, 1);
   HANDLER_TABLES_HACK(thd);
+
   if (!lock)
     goto err0; // mysql_lock_tables() printed error message already
 
@@ -277,33 +457,33 @@ int mysql_ha_read(THD *thd, TABLE_LIST *tables,
     switch (mode) {
     case RFIRST:
       if (keyname)
-        err=table->file->index_first(table->record[0]);
+        error= table->file->index_first(table->record[0]);
       else
       {
         table->file->ha_index_or_rnd_end();
-	if (!(err=table->file->ha_rnd_init(1)))
-          err=table->file->rnd_next(table->record[0]);
+	if (!(error= table->file->ha_rnd_init(1)))
+          error= table->file->rnd_next(table->record[0]);
       }
       mode=RNEXT;
       break;
     case RLAST:
       DBUG_ASSERT(keyname != 0);
-      err=table->file->index_last(table->record[0]);
+      error= table->file->index_last(table->record[0]);
       mode=RPREV;
       break;
     case RNEXT:
-      err=keyname ?
-	table->file->index_next(table->record[0]) :
-	table->file->rnd_next(table->record[0]);
-        break;
+      error= (keyname ?
+              table->file->index_next(table->record[0]) :
+              table->file->rnd_next(table->record[0]));
+      break;
     case RPREV:
       DBUG_ASSERT(keyname != 0);
-      err=table->file->index_prev(table->record[0]);
+      error= table->file->index_prev(table->record[0]);
       break;
     case RNEXT_SAME:
       /* Continue scan on "(keypart1,keypart2,...)=(c1, c2, ...)  */
       DBUG_ASSERT(keyname != 0);
-      err= table->file->index_next_same(table->record[0], key, key_len);
+      error= table->file->index_next_same(table->record[0], key, key_len);
       break;
     case RKEY:
     {
@@ -338,7 +518,7 @@ int mysql_ha_read(THD *thd, TABLE_LIST *tables,
 	goto err;
       }
       key_copy(key, table->record[0], table->key_info + keyno, key_len);
-      err=table->file->index_read(table->record[0],
+      error= table->file->index_read(table->record[0],
 				  key,key_len,ha_rkey_mode);
       mode=rkey_to_rnext[(int)ha_rkey_mode];
       break;
@@ -348,10 +528,10 @@ int mysql_ha_read(THD *thd, TABLE_LIST *tables,
       goto err;
     }
 
-    if (err == HA_ERR_RECORD_DELETED)
-      continue;
-    if (err)
+    if (error)
     {
+      if (error == HA_ERR_RECORD_DELETED)
+        continue;
       if (err != HA_ERR_KEY_NOT_FOUND && err != HA_ERR_END_OF_FILE)
       {
         sql_print_error("mysql_ha_read: Got error %d when reading table '%s'",
@@ -384,85 +564,152 @@ int mysql_ha_read(THD *thd, TABLE_LIST *tables,
 ok:
   mysql_unlock_tables(thd,lock);
   send_eof(thd);
-  return 0;
+  DBUG_PRINT("exit",("OK"));
+  DBUG_RETURN(0);
+
 err:
   mysql_unlock_tables(thd,lock);
 err0:
-  return -1;
+  DBUG_PRINT("exit",("ERROR"));
+  DBUG_RETURN(-1);
 }
 
 
 /*
-  Find a HANDLER table by name.
+  Flush (close) a list of HANDLER tables.
 
   SYNOPSIS
-    find_table_ptr_by_name()
+    mysql_ha_flush()
     thd                         Thread identifier.
-    db                          Database (schema) name.
-    table_name                  Table name ;-).
-    is_alias                    Table name may be an alias name.
-    dont_lock                   Suppresses the normal locking of LOCK_open.
+    tables                      The list of tables to close. If NULL,
+                                close all HANDLER tables [marked as flushed].
+    mode_flags                  MYSQL_HA_CLOSE_FINAL finally close the table.
+                                MYSQL_HA_REOPEN_ON_USAGE mark for reopen.
+                                MYSQL_HA_FLUSH_ALL flush all tables, not only
+                                those marked for flush.
 
   DESCRIPTION
-    Find the table 'db'.'table_name' in the list of HANDLER tables of the
-    thread 'thd'. If the table has been marked by FLUSH TABLE(S), close it,
-    flag this situation in '*was_flushed' and broadcast a COND_refresh
-    condition.
-    An empty database (schema) name matches all database (schema) names.
-    If the caller did already lock LOCK_open, it must set 'dont_lock'.
-
-  IMPLEMENTATION
-    Just in case that the table is twice in 'thd->handler_tables' (!?!),
-    the loop does not break when the table was flushed. If another table
-    by that name was found and not flushed, '*was_flushed' is cleared again,
-    since a pointer to an open HANDLER table is returned.
+    The list of HANDLER tables may be NULL, in which case all HANDLER
+    tables are closed (if MYSQL_HA_FLUSH_ALL) is set.
+    If 'tables' is NULL and MYSQL_HA_FLUSH_ALL is not set,
+    all HANDLER tables marked for flush are closed.
+    Broadcasts a COND_refresh condition, for every table closed.
+    The caller must lock LOCK_open.
+
+  NOTE
+    Since mysql_ha_flush() is called when the base table has to be closed,
+    we compare real table names, not aliases. Hence, database names matter.
 
   RETURN
-    *was_flushed                Table has been closed due to FLUSH TABLE.
-    NULL     A HANDLER Table by that name does not exist (any more).
-    != NULL  Pointer to the TABLE structure.
+    0  ok
 */
 
-static TABLE **find_table_ptr_by_name(THD *thd, const char *db,
-                                      const char *table_name,
-                                      bool is_alias, bool dont_lock,
-                                      bool *was_flushed)
+int mysql_ha_flush(THD *thd, TABLE_LIST *tables, uint mode_flags)
 {
-  int dblen;
-  TABLE **table_ptr;
-
-  DBUG_ASSERT(db);
-  dblen= strlen(db);
-  table_ptr= &(thd->handler_tables);
-  *was_flushed= FALSE;
+  TABLE_LIST    *tmp_tables;
+  TABLE         **table_ptr;
+  DBUG_ENTER("mysql_ha_flush");
+  DBUG_PRINT("enter", ("tables: %p  mode_flags: 0x%02x", tables, mode_flags));
 
-  for (TABLE *table= *table_ptr; table ; table= *table_ptr)
+  if (tables)
   {
-    if ((db == any_db || !memcmp(table->table_cache_key, db, dblen)) &&
-        !my_strcasecmp(system_charset_info,
-		       (is_alias ? table->table_name : table->real_name),
-		       table_name))
+    /* Close all tables in the list. */
+    for (tmp_tables= tables ; tmp_tables; tmp_tables= tmp_tables->next_local)
     {
-      if (table->version != refresh_version)
+      DBUG_PRINT("info-in-tables-list",("'%s'.'%s' as '%s'",
+                                        tmp_tables->db, tmp_tables->real_name,
+                                        tmp_tables->alias));
+      /* Close all currently open handler tables with the same base table. */
+      table_ptr= &(thd->handler_tables);
+      while (*table_ptr)
       {
-        if (!dont_lock)
-          VOID(pthread_mutex_lock(&LOCK_open));
-        
-        table->file->ha_index_or_rnd_end();
-        if (close_thread_table(thd, table_ptr))
+        if ((! *tmp_tables->db ||
+             ! my_strcasecmp(&my_charset_latin1, (*table_ptr)->table_cache_key,
+                             tmp_tables->db)) &&
+            ! my_strcasecmp(&my_charset_latin1, (*table_ptr)->real_name,
+                            tmp_tables->real_name))
         {
-          /* Tell threads waiting for refresh that something has happened */
-          VOID(pthread_cond_broadcast(&COND_refresh));
+          DBUG_PRINT("info",("*table_ptr '%s'.'%s' as '%s'",
+                             (*table_ptr)->table_cache_key,
+                             (*table_ptr)->real_name,
+                             (*table_ptr)->table_name));
+          mysql_ha_flush_table(thd, table_ptr, mode_flags);
+          continue;
         }
-        if (!dont_lock)
-          VOID(pthread_mutex_unlock(&LOCK_open));
-        *was_flushed= TRUE;
+        table_ptr= &(*table_ptr)->next;
+      }
+      /* end of handler_tables list */
+    }
+    /* end of flush tables list */
+  }
+  else
+  {
+    /* Close all currently open tables [which are marked for flush]. */
+    table_ptr= &(thd->handler_tables);
+    while (*table_ptr)
+    {
+      if ((mode_flags & MYSQL_HA_FLUSH_ALL) ||
+          ((*table_ptr)->version != refresh_version))
+      {
+        mysql_ha_flush_table(thd, table_ptr, mode_flags);
         continue;
       }
-      *was_flushed= FALSE;
-      break;
+      table_ptr= &(*table_ptr)->next;
+    }
+  }
+
+  DBUG_RETURN(0);
+}
+
+/*
+  Flush (close) a table.
+
+  SYNOPSIS
+    mysql_ha_flush_table()
+    thd                         Thread identifier.
+    table                       The table to close.
+    mode_flags                  MYSQL_HA_CLOSE_FINAL finally close the table.
+                                MYSQL_HA_REOPEN_ON_USAGE mark for reopen.
+
+  DESCRIPTION
+    Broadcasts a COND_refresh condition, for every table closed.
+    The caller must lock LOCK_open.
+
+  RETURN
+    0  ok
+*/
+
+static int mysql_ha_flush_table(THD *thd, TABLE **table_ptr, uint mode_flags)
+{
+  TABLE_LIST    *hash_tables;
+  TABLE         *table= *table_ptr;
+  DBUG_ENTER("mysql_ha_flush_table");
+  DBUG_PRINT("enter",("'%s'.'%s' as '%s'  flags: 0x%02x",
+                      table->table_cache_key, table->real_name,
+                      table->table_name, mode_flags));
+
+  if ((hash_tables= (TABLE_LIST*) hash_search(&thd->handler_tables_hash,
+                                        (byte*) (*table_ptr)->table_name,
+                                        strlen((*table_ptr)->table_name) + 1)))
+  {
+    if (! (mode_flags & MYSQL_HA_REOPEN_ON_USAGE))
+    {
+      /* This is a final close. Remove from hash. */
+      hash_delete(&thd->handler_tables_hash, (byte*) hash_tables);
+    }
+    else
+    {
+      /* Mark table as closed, ready for re-open. */
+      hash_tables->table= NULL;
     }
-    table_ptr= &(table->next);
+  }    
+
+  (*table_ptr)->file->ha_index_or_rnd_end();
+  if (close_thread_table(thd, table_ptr))
+  {
+    /* Tell threads waiting for refresh that something has happened */
+    VOID(pthread_cond_broadcast(&COND_refresh));
   }
-  return table_ptr;
+
+  DBUG_RETURN(0);
 }