WL#5419 "LOCK_open scalability: make tdc_refresh_version

an atomic counter"

Split the large LOCK_open section in open_table(). 
Do not call open_table_from_share() under LOCK_open.
Remove thd->version.

This fixes
Bug#50589 "Server hang on a query evaluated using a temporary 
table"
Bug#51557 "LOCK_open and kernel_mutex are not happy together"
Bug#49463 "LOCK_table and innodb are not nice when handler 
instances are created".

This patch has effect on storage engines that rely on
ha_open() PSEA method being called under LOCK_open.
In particular:

1) NDB is broken and left unfixed. NDB relies on LOCK_open
being kept as part of ha_open(), since it uses auto-discovery.
While previously the NDB open code was race-prone, now
it simply fails on asserts.

2) HEAP engine had a race in ha_heap::open() when
a share for the same table could be added twice
to the list of shares, or a dangling reference to a share
stored in HEAP handler. This patch aims to address this
problem by 'pinning' the newly created share in the 
internal HEAP engine share list until at least one
handler instance is created using that share.


include/heap.h:
  Add members to HP_CREATE_INFO.
  Declare heap_release_share().
sql/lock.cc:
  Remove thd->version, use thd->open_tables->s->version instead.
sql/repl_failsafe.cc:
  Remove thd->version.
sql/sql_base.cc:
  - close_thread_table(): move handler cleanup code outside the critical section protected by LOCK_open.
  - remove thd->version
  - split the large critical section in open_table() that
  opens a new table from share and is protected by LOCK_open
  into 2 critical sections, thus reducing the critical path.
  - make check_if_table_exists() acquire LOCK_open internally.
  - use thd->open_tables->s->version instead of thd->refresh_version to make sure that all tables in
  thd->open_tables are in the same refresh series.
sql/sql_base.h:
  Add declaration for check_if_table_exists().
sql/sql_class.cc:
  Remove init_open_tables_state(), it's now equal to
  reset_open_tables_state().
sql/sql_class.h:
  Remove thd->version, THD::init_open_tables_state().
sql/sql_plugin.cc:
  Use table->m_needs_reopen to mark the table as stale
  rather than manipulate with thd->version, which is no more.
sql/sql_udf.cc:
  Use table->m_needs_reopen to mark the table as stale
  rather than manipulate with thd->version, which is no more.
sql/table.h:
  Remove an unused variable.
sql/tztime.cc:
  Use table->m_needs_reopen to mark the table as stale
  rather than manipulate with thd->version, which is no more.
storage/heap/CMakeLists.txt:
  Add heap tests to cmake build files.
storage/heap/ha_heap.cc:
  Fix a race when ha_heap::ha_open() could insert two 
  HP_SHARE objects into the internal share list or store
  a dangling reference to a share in ha_heap instance,
  or wrongly set implicit_emptied.
storage/heap/hp_create.c:
  Optionally pin a newly created share in the list of shares
  by increasing its open_count. This is necessary to 
  make sure that a newly created share doesn't disappear while
  a HP_INFO object is being created to reference it.
storage/heap/hp_open.c:
  When adding a new HP_INFO object to the list of objects
  in the heap share, make sure the open_count is not increased
  twice.
storage/heap/hp_test1.c:
  Adjust the test to new function signatures.
storage/heap/hp_test2.c:
  Adjust the test to new function signatures.

include/heap.h

@@ -184,12 +184,22 @@ typedef struct st_heap_info
 
 typedef struct st_heap_create_info
 {
+  HP_KEYDEF *keydef;
+  ulong max_records;
+  ulong min_records;
   uint auto_key;                        /* keynr [1 - maxkey] for auto key */
   uint auto_key_type;
+  uint keys;
+  uint reclength;
   ulonglong max_table_size;
   ulonglong auto_increment;
   my_bool with_auto_increment;
   my_bool internal_table;
+  /*
+    TRUE if heap_create should 'pin' the created share by setting
+    open_count to 1. Is only looked at if not internal_table.
+  */
+  my_bool pin_share;
 } HP_CREATE_INFO;
 
 	/* Prototypes for heap-functions */
@@ -197,6 +207,7 @@ typedef struct st_heap_create_info
 extern HP_INFO *heap_open(const char *name, int mode);
 extern HP_INFO *heap_open_from_share(HP_SHARE *share, int mode);
 extern HP_INFO *heap_open_from_share_and_register(HP_SHARE *share, int mode);
+extern void heap_release_share(HP_SHARE *share, my_bool internal_table);
 extern int heap_close(HP_INFO *info);
 extern int heap_write(HP_INFO *info,const uchar *buff);
 extern int heap_update(HP_INFO *info,const uchar *old,const uchar *newdata);
@@ -205,9 +216,9 @@ extern int heap_scan_init(HP_INFO *info);
 extern int heap_scan(register HP_INFO *info, uchar *record);
 extern int heap_delete(HP_INFO *info,const uchar *buff);
 extern int heap_info(HP_INFO *info,HEAPINFO *x,int flag);
-extern int heap_create(const char *name, uint keys, HP_KEYDEF *keydef,
-		       uint reclength, ulong max_records, ulong min_records,
-		       HP_CREATE_INFO *create_info, HP_SHARE **share);
+extern int heap_create(const char *name,
+                       HP_CREATE_INFO *create_info, HP_SHARE **share,
+                       my_bool *created_new_share);
 extern int heap_delete_table(const char *name);
 extern void heap_drop_table(HP_INFO *info);
 extern int heap_extra(HP_INFO *info,enum ha_extra_function function);

sql/lock.cc

@@ -1298,7 +1298,8 @@ wait_if_global_read_lock(THD *thd, bool abort_on_refresh,
     old_message=thd->enter_cond(&COND_global_read_lock, &LOCK_global_read_lock,
 				"Waiting for release of readlock");
     while (must_wait && ! thd->killed &&
-	   (!abort_on_refresh || thd->version == refresh_version))
+	   (!abort_on_refresh || !thd->open_tables ||
+            thd->open_tables->s->version == refresh_version))
     {
       DBUG_PRINT("signal", ("Waiting for COND_global_read_lock"));
       mysql_cond_wait(&COND_global_read_lock, &LOCK_global_read_lock);

sql/repl_failsafe.cc

@@ -102,7 +102,6 @@ static int init_failsafe_rpl_thread(THD* thd)
 
   thd->mem_root->free= thd->mem_root->used= 0;
   thd_proc_info(thd, "Thread initialized");
-  thd->version=refresh_version;
   thd->set_time();
   DBUG_RETURN(0);
 }

sql/sql_base.cc

@@ -1589,10 +1589,18 @@ bool close_thread_table(THD *thd, TABLE **table_ptr)
   *table_ptr=table->next;
   mysql_mutex_unlock(&thd->LOCK_thd_data);
 
+  if (! table->needs_reopen())
+  {
+    /* Avoid having MERGE tables with attached children in unused_tables. */
+    table->file->extra(HA_EXTRA_DETACH_CHILDREN);
+    /* Free memory and reset for next loop. */
+    free_field_buffers_larger_than(table, MAX_TDC_BLOB_SIZE);
+    table->file->ha_reset();
+  }
+
   mysql_mutex_lock(&LOCK_open);
 
-  if (table->s->needs_reopen() ||
-      thd->version != refresh_version || table->needs_reopen() ||
+  if (table->s->needs_reopen() || table->needs_reopen() ||
       table_def_shutdown_in_progress)
   {
     free_cache_entry(table);
@@ -1600,14 +1608,7 @@ bool close_thread_table(THD *thd, TABLE **table_ptr)
   }
   else
   {
-    /* Avoid to have MERGE tables with attached children in unused_tables. */
     DBUG_ASSERT(table->file);
-    table->file->extra(HA_EXTRA_DETACH_CHILDREN);
-
-    /* Free memory and reset for next loop */
-    free_field_buffers_larger_than(table,MAX_TDC_BLOB_SIZE);
-
-    table->file->ha_reset();
     table_def_unuse_table(table);
     /*
       We free the least used table, not the subject table,
@@ -2305,7 +2306,7 @@ void wait_for_condition(THD *thd, mysql_mutex_t *mutex, mysql_cond_t *cond)
     @param[out]  exists  Out parameter which is set to TRUE if table
                          exists and to FALSE otherwise.
 
-    @note This function assumes that caller owns LOCK_open mutex.
+    @note This function acquires LOCK_open internally.
           It also assumes that the fact that there are no exclusive
           metadata locks on the table was checked beforehand.
 
@@ -2313,28 +2314,30 @@ void wait_for_condition(THD *thd, mysql_mutex_t *mutex, mysql_cond_t *cond)
           of engines (e.g. it was created on another node of NDB cluster)
           this function will fetch and create proper .FRM file for it.
 
-    @retval  TRUE   Some error occured
+    @retval  TRUE   Some error occurred
     @retval  FALSE  No error. 'exists' out parameter set accordingly.
 */
 
 bool check_if_table_exists(THD *thd, TABLE_LIST *table, bool *exists)
 {
   char path[FN_REFLEN + 1];
-  int rc;
+  int rc= 0;
   DBUG_ENTER("check_if_table_exists");
 
-  mysql_mutex_assert_owner(&LOCK_open);
+  mysql_mutex_assert_not_owner(&LOCK_open);
 
   *exists= TRUE;
 
+  mysql_mutex_lock(&LOCK_open);
+
   if (get_cached_table_share(table->db, table->table_name))
-    DBUG_RETURN(FALSE);
+    goto end;
 
   build_table_filename(path, sizeof(path) - 1, table->db, table->table_name,
                        reg_ext, 0);
 
   if (!access(path, F_OK))
-    DBUG_RETURN(FALSE);
+    goto end;
 
   /* .FRM file doesn't exist. Check if some engine can provide it. */
 
@@ -2344,19 +2347,17 @@ bool check_if_table_exists(THD *thd, TABLE_LIST *table, bool *exists)
   {
     /* Table does not exists in engines as well. */
     *exists= FALSE;
-    DBUG_RETURN(FALSE);
-  }
-  else if (!rc)
-  {
-    /* Table exists in some engine and .FRM for it was created. */
-    DBUG_RETURN(FALSE);
+    rc= 0;
   }
-  else /* (rc > 0) */
+  else if (rc)
   {
     my_printf_error(ER_UNKNOWN_ERROR, "Failed to open '%-.64s', error while "
                     "unpacking from engine", MYF(0), table->table_name);
-    DBUG_RETURN(TRUE);
   }
+
+end:
+  mysql_mutex_unlock(&LOCK_open);
+  DBUG_RETURN(test(rc));
 }
 
 
@@ -2651,7 +2652,7 @@ bool open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
     if (thd->global_read_lock.wait_if_global_read_lock(thd, 1, 1))
       DBUG_RETURN(TRUE);
 
-    if (thd->version != refresh_version)
+    if (thd->open_tables && thd->open_tables->s->version != refresh_version)
     {
       (void) ot_ctx->request_backoff_action(Open_table_context::OT_WAIT_TDC,
                                             NULL);
@@ -2848,48 +2849,24 @@ bool open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
   }
 
   hash_value= my_calc_hash(&table_def_cache, (uchar*) key, key_length);
-  mysql_mutex_lock(&LOCK_open);
 
-  /*
-    If it's the first table from a list of tables used in a query,
-    remember refresh_version (the version of open_cache state).
-    If the version changes while we're opening the remaining tables,
-    we will have to back off, close all the tables opened-so-far,
-    and try to reopen them.
-    Note: refresh_version is currently changed only during FLUSH TABLES.
-  */
-  if (!thd->open_tables)
-    thd->version=refresh_version;
-  else if ((thd->version != refresh_version) &&
-           ! (flags & MYSQL_OPEN_IGNORE_FLUSH))
-  {
-    /* Someone did a refresh while thread was opening tables */
-    mysql_mutex_unlock(&LOCK_open);
-    (void) ot_ctx->request_backoff_action(Open_table_context::OT_WAIT_TDC,
-                                          NULL);
-    DBUG_RETURN(TRUE);
-  }
 
   if (table_list->open_strategy == TABLE_LIST::OPEN_IF_EXISTS)
   {
     bool exists;
 
     if (check_if_table_exists(thd, table_list, &exists))
-      goto err_unlock2;
+      DBUG_RETURN(TRUE);
 
     if (!exists)
-    {
-      mysql_mutex_unlock(&LOCK_open);
       DBUG_RETURN(FALSE);
-    }
+
     /* Table exists. Let us try to open it. */
   }
   else if (table_list->open_strategy == TABLE_LIST::OPEN_STUB)
-  {
-    mysql_mutex_unlock(&LOCK_open);
     DBUG_RETURN(FALSE);
-  }
 
+  mysql_mutex_lock(&LOCK_open);
 #ifdef DISABLED_UNTIL_GRL_IS_MADE_PART_OF_MDL
   if (!(share= (TABLE_SHARE *) mdl_ticket->get_cached_object()))
 #endif
@@ -2911,7 +2888,7 @@ bool open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
         my_error(ER_WRONG_MRG_TABLE, MYF(0));
         goto err_unlock;
       }
-      
+
       /*
         This table is a view. Validate its metadata version: in particular,
         that it was a view when the statement was prepared.
@@ -2980,7 +2957,15 @@ bool open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
   }
 #endif
 
-  if (share->needs_reopen())
+
+  /*
+    If the version changes while we're opening the tables,
+    we have to back off, close all the tables opened-so-far,
+    and try to reopen them. Note: refresh_version is currently
+    changed only during FLUSH TABLES.
+  */
+  if (share->needs_reopen() ||
+      (thd->open_tables && thd->open_tables->s->version != share->version))
   {
     if (!(flags & MYSQL_OPEN_IGNORE_FLUSH))
     {
@@ -3000,8 +2985,6 @@ bool open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
                                             NULL);
       DBUG_RETURN(TRUE);
     }
-    /* Force close at once after usage */
-    thd->version= share->version;
   }
 
   if (!share->free_tables.is_empty())
@@ -3017,9 +3000,11 @@ bool open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
     while (table_cache_count > table_cache_size && unused_tables)
       free_cache_entry(unused_tables);
 
+    mysql_mutex_unlock(&LOCK_open);
+
     /* make a new table */
     if (!(table=(TABLE*) my_malloc(sizeof(*table),MYF(MY_WME))))
-      goto err_unlock;
+      goto err_lock;
 
     error= open_table_from_share(thd, share, alias,
                                  (uint) (HA_OPEN_KEYFILE |
@@ -3041,16 +3026,17 @@ bool open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
         (void) ot_ctx->request_backoff_action(Open_table_context::OT_REPAIR,
                                               table_list);
 
-      goto err_unlock;
+      goto err_lock;
     }
 
     if (open_table_entry_fini(thd, share, table))
     {
       closefrm(table, 0);
       my_free((uchar*)table, MYF(0));
-      goto err_unlock;
+      goto err_lock;
     }
 
+    mysql_mutex_lock(&LOCK_open);
     /* Add table to the share's used tables list. */
     table_def_add_used_table(thd, table);
   }
@@ -3112,6 +3098,8 @@ bool open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
     table->file->extra(HA_EXTRA_DETACH_CHILDREN);
   DBUG_RETURN(FALSE);
 
+err_lock:
+  mysql_mutex_lock(&LOCK_open);
 err_unlock:
   release_table_share(share);
 err_unlock2:

sql/sql_base.h

@@ -267,6 +267,7 @@ TABLE *find_table_for_mdl_upgrade(TABLE *list, const char *db,
                                   const char *table_name,
                                   bool no_error);
 void mark_tmp_table_for_reuse(TABLE *table);
+bool check_if_table_exists(THD *thd, TABLE_LIST *table, bool *exists);
 
 extern uint  table_cache_count;
 extern TABLE *unused_tables;

sql/sql_class.cc

@@ -592,7 +592,7 @@ THD::THD()
   *scramble= '\0';
 
   /* Call to init() below requires fully initialized Open_tables_state. */
-  init_open_tables_state(this, refresh_version);
+  reset_open_tables_state(this);
 
   init();
 #if defined(ENABLED_PROFILING)

sql/sql_class.h

@@ -1006,7 +1006,6 @@ class Open_tables_state
     of the main statement is called.
   */
   enum enum_locked_tables_mode locked_tables_mode;
-  ulong	version;
   uint current_tablenr;
 
   enum enum_flags {
@@ -1025,15 +1024,6 @@ class Open_tables_state
   */
   Open_tables_state() : state_flags(0U) { }
 
-  /**
-     Prepare Open_tables_state instance for operations dealing with tables.
-  */
-  void init_open_tables_state(THD *thd, ulong version_arg)
-  {
-    reset_open_tables_state(thd);
-    version= version_arg;
-  }
-
   void set_open_tables_state(Open_tables_state *state)
   {
     *this= *state;

sql/sql_plugin.cc

@@ -261,11 +261,6 @@ static plugin_ref intern_plugin_lock(LEX *lex, plugin_ref plugin
 static void intern_plugin_unlock(LEX *lex, plugin_ref plugin);
 static void reap_plugins(void);
 
-#ifdef EMBEDDED_LIBRARY
-/* declared in sql_base.cc */
-extern bool check_if_table_exists(THD *thd, TABLE_LIST *table, bool *exists);
-#endif /* EMBEDDED_LIBRARY */
-
 static void report_error(int where_to, uint error, ...)
 {
   va_list args;
@@ -1475,10 +1470,8 @@ static void plugin_load(MEM_ROOT *tmp_root, int *argc, char **argv)
     When building an embedded library, if the mysql.plugin table
     does not exist, we silently ignore the missing table
   */
-  mysql_mutex_lock(&LOCK_open);
   if (check_if_table_exists(new_thd, &tables, &table_exists))
     table_exists= FALSE;
-  mysql_mutex_unlock(&LOCK_open);
   if (!table_exists)
     goto end;
 #endif /* EMBEDDED_LIBRARY */
@@ -1519,7 +1512,7 @@ static void plugin_load(MEM_ROOT *tmp_root, int *argc, char **argv)
   if (error > 0)
     sql_print_error(ER(ER_GET_ERRNO), my_errno);
   end_read_record(&read_record_info);
-  new_thd->version--; // Force close to free memory
+  table->m_needs_reopen= TRUE;                  // Force close to free memory
 end:
   close_thread_tables(new_thd);
   /* Remember that we don't have a THD */

sql/sql_udf.cc

@@ -248,7 +248,7 @@ void udf_init()
   if (error > 0)
     sql_print_error("Got unknown error: %d", my_errno);
   end_read_record(&read_record_info);
-  new_thd->version--;				// Force close to free memory
+  table->m_needs_reopen= TRUE;                  // Force close to free memory
 
 end:
   close_thread_tables(new_thd);

sql/table.h

@@ -590,7 +590,6 @@ struct TABLE_SHARE
   enum enum_ha_unused unused2;
 
   uint ref_count;                       /* How many TABLE objects uses this */
-  uint open_count;			/* Number of tables in open list */
   uint blob_ptr_size;			/* 4 or 8 */
   uint key_block_size;			/* create key_block_size, if used */
   uint null_bytes, last_null_bit_pos;

sql/tztime.cc

@@ -1692,7 +1692,11 @@ my_tz_init(THD *org_thd, const char *default_tzname, my_bool bootstrap)
   }
 
   for (TABLE_LIST *tl= tz_tables; tl; tl= tl->next_global)
+  {
     tl->table->use_all_columns();
+    /* Force close at the end of the function to free memory. */
+    tl->table->m_needs_reopen= TRUE;
+  }
 
   /*
     Now we are going to load leap seconds descriptions that are shared
@@ -1781,7 +1785,6 @@ my_tz_init(THD *org_thd, const char *default_tzname, my_bool bootstrap)
 end_with_close:
   if (time_zone_tables_exist)
   {
-    thd->version--; /* Force close to free memory */
     close_thread_tables(thd);
     thd->mdl_context.release_transactional_locks();
   }

storage/heap/CMakeLists.txt

@@ -23,3 +23,9 @@ SET(HEAP_SOURCES  _check.c _rectest.c hp_block.c hp_clear.c hp_close.c hp_create
 				hp_rrnd.c hp_rsame.c hp_scan.c hp_static.c hp_update.c hp_write.c)
 
 MYSQL_ADD_PLUGIN(heap ${HEAP_SOURCES} STORAGE_ENGINE MANDATORY RECOMPILE_FOR_EMBEDDED)
+
+ADD_EXECUTABLE(hp_test1 hp_test1.c)
+TARGET_LINK_LIBRARIES(hp_test1 mysys heap)
+
+ADD_EXECUTABLE(hp_test2 hp_test2.c)
+TARGET_LINK_LIBRARIES(hp_test2 mysys heap)

storage/heap/ha_heap.cc

@@ -29,6 +29,10 @@
 static handler *heap_create_handler(handlerton *hton,
                                     TABLE_SHARE *table, 
                                     MEM_ROOT *mem_root);
+static int
+heap_prepare_hp_create_info(TABLE *table_arg, bool internal_table,
+                            HP_CREATE_INFO *hp_create_info);
+
 
 int heap_panic(handlerton *hton, ha_panic_function flag)
 {
@@ -96,43 +100,48 @@ const char **ha_heap::bas_ext() const
 
 int ha_heap::open(const char *name, int mode, uint test_if_locked)
 {
-  if ((test_if_locked & HA_OPEN_INTERNAL_TABLE) ||
-      (!(file= heap_open(name, mode)) && my_errno == ENOENT))
+  internal_table= test(test_if_locked & HA_OPEN_INTERNAL_TABLE);
+  if (internal_table || (!(file= heap_open(name, mode)) && my_errno == ENOENT))
   {
-    HA_CREATE_INFO create_info;
-    internal_table= test(test_if_locked & HA_OPEN_INTERNAL_TABLE);
-    bzero(&create_info, sizeof(create_info));
+    HP_CREATE_INFO create_info;
+    my_bool created_new_share;
+    int rc;
     file= 0;
-    if (!create(name, table, &create_info))
+    if (heap_prepare_hp_create_info(table, internal_table, &create_info))
+      goto end;
+    create_info.pin_share= TRUE;
+
+    rc= heap_create(name, &create_info, &internal_share, &created_new_share);
+    my_free((uchar*) create_info.keydef, MYF(0));
+    if (rc)
+      goto end;
+
+    implicit_emptied= test(created_new_share);
+    if (internal_table)
+      file= heap_open_from_share(internal_share, mode);
+    else
+      file= heap_open_from_share_and_register(internal_share, mode);
+
+    if (!file)
     {
-        file= internal_table ?
-          heap_open_from_share(internal_share, mode) :
-          heap_open_from_share_and_register(internal_share, mode);
-      if (!file)
-      {
-         /* Couldn't open table; Remove the newly created table */
-        mysql_mutex_lock(&THR_LOCK_heap);
-        hp_free(internal_share);
-        mysql_mutex_unlock(&THR_LOCK_heap);
-      }
-      implicit_emptied= 1;
+      heap_release_share(internal_share, internal_table);
+      goto end;
     }
   }
+
   ref_length= sizeof(HEAP_PTR);
-  if (file)
-  {
-    /* Initialize variables for the opened table */
-    set_keys_for_scanning();
-    /*
-      We cannot run update_key_stats() here because we do not have a
-      lock on the table. The 'records' count might just be changed
-      temporarily at this moment and we might get wrong statistics (Bug
-      #10178). Instead we request for update. This will be done in
-      ha_heap::info(), which is always called before key statistics are
-      used.
+  /* Initialize variables for the opened table */
+  set_keys_for_scanning();
+  /*
+    We cannot run update_key_stats() here because we do not have a
+    lock on the table. The 'records' count might just be changed
+    temporarily at this moment and we might get wrong statistics (Bug
+    #10178). Instead we request for update. This will be done in
+    ha_heap::info(), which is always called before key statistics are
+    used.
     */
-    key_stat_version= file->s->key_stat_version-1;
-  }
+  key_stat_version= file->s->key_stat_version-1;
+end:
   return (file ? 0 : 1);
 }
 
@@ -624,18 +633,20 @@ ha_rows ha_heap::records_in_range(uint inx, key_range *min_key,
 }
 
 
-int ha_heap::create(const char *name, TABLE *table_arg,
-		    HA_CREATE_INFO *create_info)
+static int
+heap_prepare_hp_create_info(TABLE *table_arg, bool internal_table,
+                            HP_CREATE_INFO *hp_create_info)
 {
   uint key, parts, mem_per_row= 0, keys= table_arg->s->keys;
   uint auto_key= 0, auto_key_type= 0;
   ha_rows max_rows;
   HP_KEYDEF *keydef;
   HA_KEYSEG *seg;
-  int error;
   TABLE_SHARE *share= table_arg->s;
   bool found_real_auto_increment= 0;
 
+  bzero(hp_create_info, sizeof(*hp_create_info));
+
   for (key= parts= 0; key < keys; key++)
     parts+= table_arg->key_info[key].key_parts;
 
@@ -715,29 +726,45 @@ int ha_heap::create(const char *name, TABLE *table_arg,
     }
   }
   mem_per_row+= MY_ALIGN(share->reclength + 1, sizeof(char*));
-  max_rows = (ha_rows) (table_arg->in_use->variables.max_heap_table_size /
-			(ulonglong) mem_per_row);
   if (table_arg->found_next_number_field)
   {
     keydef[share->next_number_index].flag|= HA_AUTO_KEY;
     found_real_auto_increment= share->next_number_key_offset == 0;
   }
+  hp_create_info->auto_key= auto_key;
+  hp_create_info->auto_key_type= auto_key_type;
+  hp_create_info->max_table_size=current_thd->variables.max_heap_table_size;
+  hp_create_info->with_auto_increment= found_real_auto_increment;
+  hp_create_info->internal_table= internal_table;
+
+  max_rows= (ha_rows) (hp_create_info->max_table_size / mem_per_row);
+  if (share->max_rows && share->max_rows < max_rows)
+    max_rows= share->max_rows;
+
+  hp_create_info->max_records= (ulong) max_rows;
+  hp_create_info->min_records= (ulong) share->min_rows;
+  hp_create_info->keys= share->keys;
+  hp_create_info->reclength= share->reclength;
+  hp_create_info->keydef= keydef;
+  return 0;
+}
+
+
+int ha_heap::create(const char *name, TABLE *table_arg,
+		    HA_CREATE_INFO *create_info)
+{
+  int error;
+  my_bool created;
   HP_CREATE_INFO hp_create_info;
-  hp_create_info.auto_key= auto_key;
-  hp_create_info.auto_key_type= auto_key_type;
+
+  error= heap_prepare_hp_create_info(table_arg, internal_table,
+                                     &hp_create_info);
+  if (error)
+    return error;
   hp_create_info.auto_increment= (create_info->auto_increment_value ?
 				  create_info->auto_increment_value - 1 : 0);
-  hp_create_info.max_table_size=current_thd->variables.max_heap_table_size;
-  hp_create_info.with_auto_increment= found_real_auto_increment;
-  hp_create_info.internal_table= internal_table;
-  max_rows = (ha_rows) (hp_create_info.max_table_size / mem_per_row);
-  error= heap_create(name,
-		     keys, keydef, share->reclength,
-		     (ulong) ((share->max_rows < max_rows &&
-			       share->max_rows) ? 
-			      share->max_rows : max_rows),
-		     (ulong) share->min_rows, &hp_create_info, &internal_share);
-  my_free((uchar*) keydef, MYF(0));
+  error= heap_create(name, &hp_create_info, &internal_share, &created);
+  my_free((uchar*) hp_create_info.keydef, MYF(0));
   DBUG_ASSERT(file == 0);
   return (error);
 }

storage/heap/hp_create.c

@@ -21,24 +21,30 @@ static void init_block(HP_BLOCK *block,uint reclength,ulong min_records,
 
 /* Create a heap table */
 
-int heap_create(const char *name, uint keys, HP_KEYDEF *keydef,
-		uint reclength, ulong max_records, ulong min_records,
-		HP_CREATE_INFO *create_info, HP_SHARE **res)
+int heap_create(const char *name, HP_CREATE_INFO *create_info,
+                HP_SHARE **res, my_bool *created_new_share)
 {
   uint i, j, key_segs, max_length, length;
   HP_SHARE *share= 0;
   HA_KEYSEG *keyseg;
+  HP_KEYDEF *keydef= create_info->keydef;
+  uint reclength= create_info->reclength;
+  uint keys= create_info->keys;
+  ulong min_records= create_info->min_records;
+  ulong max_records= create_info->max_records;
   DBUG_ENTER("heap_create");
 
   if (!create_info->internal_table)
   {
     mysql_mutex_lock(&THR_LOCK_heap);
-    if ((share= hp_find_named_heap(name)) && share->open_count == 0)
+    share= hp_find_named_heap(name);
+    if (share && share->open_count == 0)
     {
       hp_free(share);
       share= 0;
     }
-  }  
+  }
+  *created_new_share= (share == NULL);
 
   if (!share)
   {
@@ -200,7 +206,11 @@ int heap_create(const char *name, uint keys, HP_KEYDEF *keydef,
       share->delete_on_close= 1;
   }
   if (!create_info->internal_table)
+  {
+    if (create_info->pin_share)
+      ++share->open_count;
     mysql_mutex_unlock(&THR_LOCK_heap);
+  }
 
   *res= share;
   DBUG_RETURN(0);

storage/heap/hp_open.c

@@ -74,12 +74,33 @@ HP_INFO *heap_open_from_share_and_register(HP_SHARE *share, int mode)
   {
     info->open_list.data= (void*) info;
     heap_open_list= list_add(heap_open_list,&info->open_list);
+    /* Unpin the share, it is now pinned by the file. */
+    share->open_count--;
   }
   mysql_mutex_unlock(&THR_LOCK_heap);
   DBUG_RETURN(info);
 }
 
 
+/**
+  Dereference a HEAP share and free it if it's not referenced.
+  We don't check open_count for internal tables since they
+  are always thread-local, i.e. referenced by a single thread.
+*/
+void heap_release_share(HP_SHARE *share, my_bool internal_table)
+{
+  /* Couldn't open table; Remove the newly created table */
+  if (internal_table)
+    hp_free(share);
+  else
+  {
+    mysql_mutex_lock(&THR_LOCK_heap);
+    if (--share->open_count == 0)
+      hp_free(share);
+    mysql_mutex_unlock(&THR_LOCK_heap);
+  }
+}
+
 /*
   Open heap table based on name
 

storage/heap/hp_test1.c

@@ -38,6 +38,7 @@ int main(int argc, char **argv)
   HA_KEYSEG keyseg[4];
   HP_CREATE_INFO hp_create_info;
   HP_SHARE *tmp_share;
+  my_bool unused;
   MY_INIT(argv[0]);
 
   filename= "test1";
@@ -45,6 +46,11 @@ int main(int argc, char **argv)
 
   bzero(&hp_create_info, sizeof(hp_create_info));
   hp_create_info.max_table_size= 1024L*1024L;
+  hp_create_info.keys= 1;
+  hp_create_info.keydef= keyinfo;
+  hp_create_info.reclength= 30;
+  hp_create_info.max_records= (ulong) flag*100000L;
+  hp_create_info.min_records= 10UL;
 
   keyinfo[0].keysegs=1;
   keyinfo[0].seg=keyseg;
@@ -55,13 +61,12 @@ int main(int argc, char **argv)
   keyinfo[0].seg[0].charset= &my_charset_latin1;
   keyinfo[0].seg[0].null_bit= 0;
   keyinfo[0].flag = HA_NOSAME;
-  
+
   deleted=0;
   bzero((uchar*) flags,sizeof(flags));
 
   printf("- Creating heap-file\n");
-  if (heap_create(filename,1,keyinfo,30,(ulong) flag*100000L,10L,
-		  &hp_create_info, &tmp_share) ||
+  if (heap_create(filename, &hp_create_info, &tmp_share, &unused) ||
       !(file= heap_open(filename, 2)))
     goto err;
   printf("- Writing records:s\n");

storage/heap/hp_test2.c

@@ -65,15 +65,21 @@ int main(int argc, char *argv[])
   HEAP_PTR UNINIT_VAR(position);
   HP_CREATE_INFO hp_create_info;
   CHARSET_INFO *cs= &my_charset_latin1;
+  my_bool unused;
   MY_INIT(argv[0]);		/* init my_sys library & pthreads */
 
   filename= "test2";
   filename2= "test2_2";
   file=file2=0;
   get_options(argc,argv);
-  
+
   bzero(&hp_create_info, sizeof(hp_create_info));
   hp_create_info.max_table_size= 1024L*1024L;
+  hp_create_info.keys= keys;
+  hp_create_info.keydef= keyinfo;
+  hp_create_info.reclength= reclength;
+  hp_create_info.max_records= (ulong) flag*100000L;
+  hp_create_info.min_records= (ulong) recant/2;
 
   write_count=update=opt_delete=0;
   key_check=0;
@@ -125,8 +131,7 @@ int main(int argc, char *argv[])
   bzero((char*) key3,sizeof(key3));
 
   printf("- Creating heap-file\n");
-  if (heap_create(filename,keys,keyinfo,reclength,(ulong) flag*100000L, 
-                  (ulong) recant/2, &hp_create_info, &tmp_share) ||
+  if (heap_create(filename, &hp_create_info, &tmp_share, &unused) ||
       !(file= heap_open(filename, 2)))
     goto err;
   signal(SIGINT,endprog);
@@ -563,8 +568,10 @@ int main(int argc, char *argv[])
   heap_close(file2);
 
   printf("- Creating output heap-file 2\n");
-  if (heap_create(filename2, 1, keyinfo, reclength, 0L, 0L, &hp_create_info,
-                  &tmp_share) ||
+  hp_create_info.keys= 1;
+  hp_create_info.max_records= 0;
+  hp_create_info.min_records= 0;
+  if (heap_create(filename2, &hp_create_info, &tmp_share, &unused) ||
       !(file2= heap_open_from_share_and_register(tmp_share, 2)))
     goto err;