BUG#4286 - HANDLER tables are closed by FLUSH TABLE(S).

BUG#4335 - one name can be handler open'ed many times.
Reworked the HANDLER functions and interface.
Using a HASH to store information on open tables that
survives FLUSH TABLE.
HANDLER tables alias names must now be unique, though it
is allowed in 4.0 to qualify them with the database name
of the base table.


mysql-test/r/flush_table.result:
  BUG#4286 - HANDLER tables are closed by FLUSH TABLE(S).
  BUG#4335 - one name can be handler open'ed many times.
  Moved pure handler test results to handler.result.
  Added the new test results.
mysql-test/r/handler.result:
  BUG#4286 - HANDLER tables are closed by FLUSH TABLE(S).
  BUG#4335 - one name can be handler open'ed many times.
  Moved pure handler test results from flush_table.result to here.
mysql-test/t/flush_table.test:
  BUG#4286 - HANDLER tables are closed by FLUSH TABLE(S).
  BUG#4335 - one name can be handler open'ed many times.
  Moved pure handler tests to handler.test.
  Added new tests.
mysql-test/t/handler.test:
  BUG#4286 - HANDLER tables are closed by FLUSH TABLE(S).
  BUG#4335 - one name can be handler open'ed many times.
  Moved pure handler tests from flush_table.test to here.
sql/mysql_priv.h:
  BUG#4286 - HANDLER tables are closed by FLUSH TABLE(S).
  BUG#4335 - one name can be handler open'ed many times.
  Reworked the handler interface.
sql/sql_base.cc:
  BUG#4286 - HANDLER tables are closed by FLUSH TABLE(S).
  BUG#4335 - one name can be handler open'ed many times.
  Replaced mysql_ha_close_list() by the better named
  function mysql_ha_flush() with readable options.
sql/sql_class.cc:
  BUG#4286 - HANDLER tables are closed by FLUSH TABLE(S).
  BUG#4335 - one name can be handler open'ed many times.
  Added initialization for the handler tables hash.
  Changed the handler tables clean-up code.
  Unreleted to bug: Changed the order of THD initialization
  to avoid warning messages on Linux with gcc.
sql/sql_class.h:
  BUG#4286 - HANDLER tables are closed by FLUSH TABLE(S).
  BUG#4335 - one name can be handler open'ed many times.
  Added the handler tables HASH to THD.
sql/sql_handler.cc:
  BUG#4286 - HANDLER tables are closed by FLUSH TABLE(S).
  BUG#4335 - one name can be handler open'ed many times.
  Completely reworked the handler functions.
  Added an introducing comment, describing the new functionality.
sql/sql_table.cc:
  BUG#4286 - HANDLER tables are closed by FLUSH TABLE(S).
  BUG#4335 - one name can be handler open'ed many times.
  replaced mysql_ha_close() by the better named
  function mysql_ha_flush() with readable options.

1 files changed, 451 insertions, 167 deletions

diff --git a/sql/sql_handler.cc b/sql/sql_handler.cc
index 272289b6176..f056651919f 100644
--- a/sql/sql_handler.cc
+++ b/sql/sql_handler.cc
@@ -17,10 +17,6 @@
 
 /* HANDLER ... commands - direct access to ISAM */
 
-#include "mysql_priv.h"
-#include "sql_select.h"
-#include <assert.h>
-
 /* TODO:
   HANDLER blabla OPEN [ AS foobar ] [ (column-list) ]
 
@@ -38,183 +34,404 @@
   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 };
+
 #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, int 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.
 
-int mysql_ha_open(THD *thd, TABLE_LIST *tables)
+  RETURN
+    Pointer to the TABLE_LIST struct.
+*/
+
+static char *mysql_ha_hash_get_key(TABLE_LIST *tables, uint *key_len_p,
+                                   my_bool first __attribute__((unused)))
 {
-  HANDLER_TABLES_HACK(thd);
-  int err=open_tables(thd,tables);
-  HANDLER_TABLES_HACK(thd);
-  if (err)
-    return -1;
+  *key_len_p= strlen(tables->alias) + 1 ; /* include '\0' in comparisons */
+  return tables->alias;
+}
 
-  // 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;
-  }
 
-  send_ok(&thd->net);
-  return 0;
+/*
+  Free an hash object.
+
+  SYNOPSIS
+    mysql_ha_hash_free()
+    tables                      Pointer to the hash object.
+
+  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;
+  char          *name;
+  char          *alias;
+  uint          dblen;
+  uint          namelen;
+  uint          aliaslen;
+  int           err;
+  DBUG_ENTER("mysql_ha_open");
+  DBUG_PRINT("enter",("mysql_ha_open: '%s'.'%s' as '%s' reopen %d",
+                      tables->db, tables->real_name, tables->alias, reopen));
+
+  if (! hash_inited(&thd->handler_tables_hash))
+  {
+    /*
+      HASH entries are of type TABLE_LIST.
+    */
+    if (hash_init(&thd->handler_tables_hash, 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 (!dont_lock)
-      VOID(pthread_mutex_lock(&LOCK_open));
-    if (close_thread_table(thd, table_ptr))
+    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",("mysql_ha_open:  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);
+  err=open_tables(thd,tables);
+  HANDLER_TABLES_HACK(thd);
+  if (err)
+    goto err;
+
+  /* There can be only one table in '*tables'. */
+  if (! (tables->table->file->table_flags() & HA_CAN_SQL_HANDLER))
   {
-    my_printf_error(ER_UNKNOWN_TABLE, ER(ER_UNKNOWN_TABLE), MYF(0),
-                    tables->alias, "HANDLER");
-    return -1;
+    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)
+  {
+    /* 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)))
+    {
+      DBUG_PRINT("exit",("mysql_ha_open: malloc ERROR"));
+      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 (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->net);
-  return 0;
+  DBUG_PRINT("exit",("mysql_ha_open: OK"));
+  DBUG_RETURN(0);
+
+err:
+  DBUG_PRINT("exit",("mysql_ha_open: 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;
-
-  if (tables)
+  bool          was_flushed= FALSE;
+  bool          not_opened;
+  DBUG_ENTER("mysql_ha_close");
+  DBUG_PRINT("enter",("mysql_ha_close: '%s'.'%s' as '%s'",
+                      tables->db, tables->real_name, tables->alias));
+
+  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)
+    /*
+      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 MYSQL_VERSION_ID < 40100
+    if (*tables->db && strcmp(hash_tables->db, tables->db))
     {
-      mysql_ha_close(thd, tl_item, /*dont_send_ok*/ 1,
-                     /*dont_lock*/ 1, /*no_alias*/ 1);
+      DBUG_PRINT("info",("mysql_ha_close: wrong db"));
+      hash_tables= NULL;
     }
-  }
-  else
-  {
-    table_ptr= &(thd->handler_tables);
-    while (*table_ptr)
+    else
+#endif
     {
-      if (! flushed || ((*table_ptr)->version != refresh_version))
+      if (*table_ptr)
       {
+        VOID(pthread_mutex_lock(&LOCK_open));
         if (close_thread_table(thd, table_ptr))
         {
           /* Tell threads waiting for refresh that something has happened */
           VOID(pthread_cond_broadcast(&COND_refresh));
         }
-        continue;
+        VOID(pthread_mutex_unlock(&LOCK_open));
       }
-      table_ptr= &((*table_ptr)->next);
+
+      hash_delete(&thd->handler_tables_hash, (byte*) hash_tables);
     }
   }
-  return 0;
+
+  if (! hash_tables)
+  {
+#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
+    DBUG_PRINT("exit",("mysql_ha_close: ERROR"));
+    DBUG_RETURN(-1);
+  }
+
+  send_ok(&thd->net);
+  DBUG_PRINT("exit",("mysql_ha_close: OK"));
+  DBUG_RETURN(0);
 }
 
-static enum enum_ha_read_modes rkey_to_rnext[]=
-    { RNEXT_SAME, RNEXT, RPREV, RNEXT, RPREV, RNEXT, 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;
+  int           err;
+  int           keyno=-1;
+  uint          num_rows;
+  bool          was_flushed;
+  MYSQL_LOCK    *lock;
+  DBUG_ENTER("mysql_ha_read");
+  DBUG_PRINT("enter",("mysql_ha_read: '%s'.'%s' as '%s'",
+                      tables->db, tables->real_name, tables->alias));
+
+  List<Item> list;
+  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",("mysql_ha_read: found 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",("mysql_ha_read: reopen failed"));
+        goto err0;
+      }
+
+      table= hash_tables->table;
+      DBUG_PRINT("info",("mysql_ha_read: 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",("mysql_ha_read: 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))
-    return -1;
+    goto err0;
 
   table->file->init_table_handle_for_HANDLER(); // Only InnoDB requires it
 
@@ -224,24 +441,19 @@ 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->index_init(keyno);
   }
 
-  List<Item> list;
-  list.push_front(new Item_field(NULL,NULL,"*"));
-  List_iterator<Item> it(list);
-  uint num_rows;
-  it++;
-
-  insert_fields(thd,tables,tables->db,tables->alias,&it);
+  if (insert_fields(thd,tables,tables->db,tables->alias,&it))
+    goto err0;
 
   select_limit+=offset_limit;
   send_fields(thd,list,1);
 
   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);
   
   byte *key;
@@ -363,83 +575,155 @@ int mysql_ha_read(THD *thd, TABLE_LIST *tables,
 ok:
   mysql_unlock_tables(thd,lock);
   send_eof(&thd->net);
-  return 0;
+  DBUG_PRINT("exit",("mysql_ha_read: OK"));
+  DBUG_RETURN(0);
 err:
   mysql_unlock_tables(thd,lock);
 err0:
-  return -1;
+  DBUG_PRINT("exit",("mysql_ha_read: 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, int mode_flags)
 {
-  int dblen;
-  TABLE **table_ptr;
-
-  DBUG_ASSERT(db);
-  dblen= *db ? strlen(db)+1 : 0;
-  table_ptr= &(thd->handler_tables);
-  *was_flushed= FALSE;
+  TABLE_LIST    **tmp_tables_p;
+  TABLE_LIST    *tmp_tables;
+  TABLE         **table_ptr;
+  bool          was_flushed;
+  DBUG_ENTER("mysql_ha_flush");
+  DBUG_PRINT("enter",("mysql_ha_flush: tables %p mode_flags 0x%02x",
+                      tables, mode_flags));
 
-  for (TABLE *table=*table_ptr; table ; table=*table_ptr)
+  if (tables)
   {
-    if ((!dblen || !memcmp(table->table_cache_key, db, dblen)) &&
-        !my_strcasecmp((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)
     {
-      if (table->version != refresh_version)
+      DBUG_PRINT("info",("mysql_ha_flush: 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));
-        if (close_thread_table(thd, table_ptr))
+        if ((! *tmp_tables->db ||
+             ! my_strcasecmp((*table_ptr)->table_cache_key, tmp_tables->db)) &&
+            ! my_strcasecmp((*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",("mysql_ha_flush: *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_PRINT("exit",("mysql_ha_flush: OK"));
+  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, int mode_flags)
+{
+  TABLE_LIST    *hash_tables;
+  TABLE         *table= *table_ptr;
+  bool          was_flushed;
+  DBUG_ENTER("mysql_ha_flush_table");
+  DBUG_PRINT("info",("mysql_ha_flush_table: '%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,
+                                        (*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);
     }
-    table_ptr=&(table->next);
+    else
+    {
+      /* Mark table as closed, ready for re-open. */
+      hash_tables->table= NULL;
+    }
+  }    
+
+  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_PRINT("exit",("mysql_ha_flush_table: OK"));
+  DBUG_RETURN(0);
 }