Patch to handle some bad situations resulting from the fix for BUG#19995.

sql/handler.cc:
  Generating table maps from all locks that can be available: THD::extra_lock,
  THD::lock, and THD::locked_tables.
sql/sql_class.h:
  Adding member Open_tables:state::extra_lock to hold the extra lock used by select_create.
  Removing select_insert::lock.
sql/sql_insert.cc:
  Adding member Open_tables:state::extra_lock to hold the extra lock used by select_create.
  Removing select_insert::lock.

sql/handler.cc

@@ -3263,37 +3263,39 @@ namespace
   int write_locked_table_maps(THD *thd)
   {
     DBUG_ENTER("write_locked_table_maps");
-    DBUG_PRINT("enter", ("thd=%p, thd->lock=%p, thd->locked_tables=%p",
+    DBUG_PRINT("enter", ("thd=%p, thd->lock=%p, thd->locked_tables=%p, thd->extra_lock",
-                         thd, thd->lock, thd->locked_tables));
+                         thd, thd->lock, thd->locked_tables, thd->extra_lock));
 
     if (thd->get_binlog_table_maps() == 0)
     {
-      /*
+      MYSQL_LOCK *const locks[] = {
-        Exactly one table has to be locked, otherwise this code is not
+        thd->extra_lock, thd->lock, thd->locked_tables
-        guaranteed to work.
+      };
-      */
+      for (my_ptrdiff_t i= 0 ; i < sizeof(locks)/sizeof(*locks) ; ++i )
-      DBUG_ASSERT((thd->lock != NULL) + (thd->locked_tables != NULL) == 1);
-
-      MYSQL_LOCK *lock= thd->lock ? thd->lock : thd->locked_tables;
-      DBUG_ASSERT(lock->table_count > 0);
-      TABLE **const end_ptr= lock->table + lock->table_count;
-      for (TABLE **table_ptr= lock->table ; 
-           table_ptr != end_ptr ;
-           ++table_ptr)
       {
-        TABLE *const table= *table_ptr;
+        MYSQL_LOCK const *const lock= locks[i];
-        DBUG_PRINT("info", ("Checking table %s", table->s->table_name));
+        if (lock == NULL)
-        if (table->current_lock == F_WRLCK &&
+          continue;
-            check_table_binlog_row_based(thd, table))
+
+        TABLE **const end_ptr= lock->table + lock->table_count;
+        for (TABLE **table_ptr= lock->table ; 
+             table_ptr != end_ptr ;
+             ++table_ptr)
         {
-          int const has_trans= table->file->has_transactions();
+          TABLE *const table= *table_ptr;
-          int const error= thd->binlog_write_table_map(table, has_trans);
+          DBUG_PRINT("info", ("Checking table %s", table->s->table_name));
-          /*
+          if (table->current_lock == F_WRLCK &&
-            If an error occurs, it is the responsibility of the caller to
+              check_table_binlog_row_based(thd, table))
-            roll back the transaction.
+          {
-          */
+            int const has_trans= table->file->has_transactions();
-          if (unlikely(error))
+            int const error= thd->binlog_write_table_map(table, has_trans);
-            DBUG_RETURN(1);
+            /*
+              If an error occurs, it is the responsibility of the caller to
+              roll back the transaction.
+            */
+            if (unlikely(error))
+              DBUG_RETURN(1);
+          }
         }
       }
     }

sql/sql_class.h

@@ -693,6 +693,14 @@ public:
      THD::prelocked_mode for more info.)
   */
   MYSQL_LOCK *locked_tables;
+
+  /*
+    CREATE-SELECT keeps an extra lock for the table being
+    created. This field is used to keep the extra lock available for
+    lower level routines, which would otherwise miss that lock.
+   */
+  MYSQL_LOCK *extra_lock;
+
   /*
     prelocked_mode_type enum and prelocked_mode member are used for
     indicating whenever "prelocked mode" is on, and what type of
@@ -745,7 +753,7 @@ public:
   void reset_open_tables_state()
   {
     open_tables= temporary_tables= handler_tables= derived_tables= 0;
-    lock= locked_tables= 0;
+    extra_lock= lock= locked_tables= 0;
     prelocked_mode= NON_PRELOCKED;
     state_flags= 0U;
   }
@@ -1591,9 +1599,6 @@ class select_insert :public select_result_interceptor {
   bool send_eof();
   /* not implemented: select_insert is never re-used in prepared statements */
   void cleanup();
-
-protected:
-  MYSQL_LOCK *lock;
 };
 
 

sql/sql_insert.cc

@@ -2188,7 +2188,6 @@ select_insert::select_insert(TABLE_LIST *table_list_par, TABLE *table_par,
                              bool ignore_check_option_errors)
   :table_list(table_list_par), table(table_par), fields(fields_par),
    last_insert_id(0),
-   lock(0),
    insert_into_view(table_list_par && table_list_par->view != 0)
 {
   bzero((char*) &info,sizeof(info));
@@ -2356,6 +2355,7 @@ bool select_insert::send_data(List<Item> &values)
 {
   DBUG_ENTER("select_insert::send_data");
   bool error=0;
+
   if (unit->offset_limit_cnt)
   {						// using limit offset,count
     unit->offset_limit_cnt--;
@@ -2377,34 +2377,8 @@ bool select_insert::send_data(List<Item> &values)
     }
   }
 
-  /*
+  error= write_record(thd, table, &info);
-    The thd->lock lock contain the locks for the select part of the
-    statement and the 'lock' variable contain the write lock for the
-    currently locked table that is being created or inserted
-    into. However, the row-based replication will investigate the
-    thd->lock to decide what table maps are to be written, so this one
-    has to contain the tables locked for writing.  To be able to write
-    table map for the table being created, we temporarily set
-    THD::lock to select_insert::lock while writing the record to the
-    storage engine. We cannot set this elsewhere, since the execution
-    of a stored function inside the select expression might cause the
-    lock structures to be NULL.
-   */
-  
-  {
-    MYSQL_LOCK *saved_lock= NULL;
-    if (lock)
-    {
-      saved_lock= thd->lock;
-      thd->lock= lock;
-    }
-
-    error= write_record(thd, table, &info);
     
-    if (lock)
-      thd->lock= saved_lock;
-  }
-
   if (!error)
   {
     if (table->triggers || info.handle_duplicates == DUP_UPDATE)
@@ -2776,8 +2750,8 @@ select_create::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
 
   unit= u;
   if (!(table= create_table_from_items(thd, create_info, create_table,
-                                       extra_fields, keys, &values, &lock,
+                                       extra_fields, keys, &values,
-                                       hook_ptr)))
+                                       &thd->extra_lock, hook_ptr)))
     DBUG_RETURN(-1);				// abort() deletes table
 
   if (table->s->fields < values.elements)
@@ -2884,13 +2858,13 @@ bool select_create::send_eof()
   {
     table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
     VOID(pthread_mutex_lock(&LOCK_open));
-    mysql_unlock_tables(thd, lock);
+    mysql_unlock_tables(thd, thd->extra_lock);
     if (!table->s->tmp_table)
     {
       if (close_thread_table(thd, &table))
         VOID(pthread_cond_broadcast(&COND_refresh));
     }
-    lock=0;
+    thd->extra_lock=0;
     table=0;
     VOID(pthread_mutex_unlock(&LOCK_open));
   }
@@ -2900,10 +2874,10 @@ bool select_create::send_eof()
 void select_create::abort()
 {
   VOID(pthread_mutex_lock(&LOCK_open));
-  if (lock)
+  if (thd->extra_lock)
   {
-    mysql_unlock_tables(thd, lock);
+    mysql_unlock_tables(thd, thd->extra_lock);
-    lock=0;
+    thd->extra_lock=0;
   }
   if (table)
   {