Recommit of antonys previous commit.

Bug#21413
"Engine table handler used by multiple threads in REPLACE DELAYED"
When executing a REPLACE DELAYED statement, the storage engine
::extra() method was invoked by a different thread than the thread
which has acquired the handler instance.

This did not cause problems within the current server and with
the current storage engines.
But it has the potential to confuse future storage engines.

Added code to avoid surplus calls to extra() method in case of DELAYED
which avoids calling storage engine from a different thread than
expected.

No test case.
This change does not change behavior in conjunction with current
storage engines. So it cannot be tested by the regression test suite.


sql/sql_insert.cc:
  Bug#21413
  When performing DELAYED operations, we should not call
  the storage engine methods from the 'wrong' context.
  Ensure that the calls to the methods are appropiately guarded
  by either moving calls to sections where the lock_type is checked.
  No need to copy all elements of TABLE_LIST when opening table
  using open_ltable().

sql/sql_insert.cc

@@ -315,8 +315,13 @@ int mysql_insert(THD *thd,TABLE_LIST *table_list,
   error=0;
   id=0;
   thd->proc_info="update";
+#ifndef EMBEDDED_LIBRARY
+  if (lock_type != TL_WRITE_DELAYED)
+#endif /* EMBEDDED_LIBRARY */
+  {
     if (duplic != DUP_ERROR || ignore)
       table->file->extra(HA_EXTRA_IGNORE_DUP_KEY);
+
     /*
       let's *try* to start bulk inserts. It won't necessary
       start them as values_list.elements should be greater than
@@ -325,8 +330,8 @@ int mysql_insert(THD *thd,TABLE_LIST *table_list,
       values_list.elements, and - if nothing else - to initialize
       the code to make the call of end_bulk_insert() below safe.
     */
-  if (lock_type != TL_WRITE_DELAYED)
     table->file->start_bulk_insert(values_list.elements);
+  }
 
   while ((values= its++))
   {
@@ -415,6 +420,9 @@ int mysql_insert(THD *thd,TABLE_LIST *table_list,
     else if (table->next_number_field && info.copied)
       id=table->next_number_field->val_int();	// Return auto_increment value
 
+    if (duplic != DUP_ERROR || ignore)
+      table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
+
     /*
       Invalidate the table in the query cache if something changed.
       For the transactional algorithm to work the invalidation must be
@@ -455,8 +463,6 @@ int mysql_insert(THD *thd,TABLE_LIST *table_list,
   table->next_number_field=0;
   thd->count_cuted_fields= CHECK_FIELD_IGNORE;
   thd->next_insert_id=0;			// Reset this if wrongly used
-  if (duplic != DUP_ERROR || ignore)
-    table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
 
   /* Reset value of LAST_INSERT_ID if no rows where inserted */
   if (!info.copied && thd->insert_id_used)
@@ -561,7 +567,12 @@ int mysql_prepare_insert(THD *thd, TABLE_LIST *table_list,
       DBUG_RETURN(-1);
     }
   }
-  if (duplic == DUP_UPDATE || duplic == DUP_REPLACE)
+  /*
+    Only call extra() handler method if we are not performing a DELAYED
+    operation. It will instead be executed by delayed insert thread.
+  */
+  if ((duplic == DUP_UPDATE || duplic == DUP_REPLACE) &&
+      (insert_table_list->lock_type != TL_WRITE_DELAYED))
     table->file->extra(HA_EXTRA_RETRIEVE_PRIMARY_KEY);
 
   DBUG_RETURN(0);
@@ -930,9 +941,11 @@ static TABLE *delayed_get_table(THD *thd,TABLE_LIST *table_list)
 	my_error(ER_OUT_OF_RESOURCES,MYF(0));
 	goto err1;
       }
-      tmp->table_list= *table_list;			// Needed to open table
+      /* We only need the db and table name to open tables with open_ltable() */
       tmp->table_list.db= tmp->thd.db;
+      tmp->table_list.db_length= table_list->db_length;
       tmp->table_list.alias= tmp->table_list.real_name=tmp->thd.query;
+      tmp->table_list.real_name_length= table_list->real_name_length;
       tmp->lock();
       pthread_mutex_lock(&tmp->mutex);
       if ((error=pthread_create(&tmp->thd.real_id,&connection_attrib,
@@ -1491,6 +1504,9 @@ bool delayed_insert::handle_inserts(void)
 
     info.ignore= row->ignore;
     info.handle_duplicates= row->dup;
+    if (info.handle_duplicates == DUP_UPDATE || 
+        info.handle_duplicates == DUP_REPLACE)
+      table->file->extra(HA_EXTRA_RETRIEVE_PRIMARY_KEY);
     if (info.ignore ||
 	info.handle_duplicates != DUP_ERROR)
     {