1 files changed, 203 insertions, 135 deletions

diff --git a/sql/sql_delete.cc b/sql/sql_delete.cc
index af895c172ff..9b66684a463 100644
--- a/sql/sql_delete.cc
+++ b/sql/sql_delete.cc
@@ -20,22 +20,31 @@
 */
 
 #include "mysql_priv.h"
-#include "ha_innodb.h"
 #include "sql_select.h"
 #include "sp_head.h"
 #include "sql_trigger.h"
 
+/**
+  Implement DELETE SQL word.
+
+  @note Like implementations of other DDL/DML in MySQL, this function
+  relies on the caller to close the thread tables. This is done in the
+  end of dispatch_command().
+*/
+
 bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
                   SQL_LIST *order, ha_rows limit, ulonglong options,
                   bool reset_auto_increment)
 {
-  int		error;
+  bool          will_batch;
+  int		error, loc_error;
   TABLE		*table;
   SQL_SELECT	*select=0;
   READ_RECORD	info;
   bool          using_limit=limit != HA_POS_ERROR;
   bool		transactional_table, safe_update, const_cond;
-  ha_rows	deleted;
+  bool          const_cond_result;
+  ha_rows	deleted= 0;
   uint usable_index= MAX_KEY;
   SELECT_LEX   *select_lex= &thd->lex->select_lex;
   THD::killed_state killed_status= THD::NOT_KILLED;
@@ -49,13 +58,7 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
 	     table_list->view_db.str, table_list->view_name.str);
     DBUG_RETURN(TRUE);
   }
-  error= table->file->info(HA_STATUS_VARIABLE | HA_STATUS_NO_LOCK);
-  if (error)
-  {
-    table->file->print_error(error, MYF(0));
-    DBUG_RETURN(error);
-  }
-  thd->proc_info="init";
+  thd_proc_info(thd, "init");
   table->map=1;
 
   if (mysql_prepare_delete(thd, table_list, &conds))
@@ -93,23 +96,45 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
 
   select_lex->no_error= thd->lex->ignore;
 
+  const_cond_result= const_cond && (!conds || conds->val_int());
+  if (thd->is_error())
+  {
+    /* Error evaluating val_int(). */
+    DBUG_RETURN(TRUE);
+  }
   /*
     Test if the user wants to delete all rows and deletion doesn't have
     any side-effects (because of triggers), so we can use optimized
     handler::delete_all_rows() method.
-    We implement fast TRUNCATE for InnoDB even if triggers are present. 
-    TRUNCATE ignores triggers.
+
+    We implement fast TRUNCATE for InnoDB even if triggers are
+    present.  TRUNCATE ignores triggers.
+
+    We can use delete_all_rows() if and only if:
+    - We allow new functions (not using option --skip-new), and are
+      not in safe mode (not using option --safe-mode)
+    - There is no limit clause
+    - The condition is constant
+    - If there is a condition, then it it produces a non-zero value
+    - If the current command is DELETE FROM with no where clause
+      (i.e., not TRUNCATE) then:
+      - We should not be binlogging this statement row-based, and
+      - there should be no delete triggers associated with the table.
   */
-  if (!using_limit && const_cond && (!conds || conds->val_int()) &&
+  if (!using_limit && const_cond_result &&
       !(specialflag & (SPECIAL_NO_NEW_FUNC | SPECIAL_SAFE_MODE)) &&
       (thd->lex->sql_command == SQLCOM_TRUNCATE ||
-       !(table->triggers && table->triggers->has_delete_triggers()))
-     )
+       (!thd->current_stmt_binlog_row_based &&
+        !(table->triggers && table->triggers->has_delete_triggers()))))
   {
-    deleted= table->file->records;
-    if (!(error=table->file->delete_all_rows()))
+    /* Update the table->file->stats.records number */
+    table->file->info(HA_STATUS_VARIABLE | HA_STATUS_NO_LOCK);
+    ha_rows const maybe_deleted= table->file->stats.records;
+    DBUG_PRINT("debug", ("Trying to use delete_all_rows()"));
+    if (!(error=table->file->ha_delete_all_rows()))
     {
       error= -1;				// ok
+      deleted= maybe_deleted;
       goto cleanup;
     }
     if (error != HA_ERR_WRONG_COMMAND)
@@ -120,7 +145,6 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
     }
     /* Handler didn't support fast delete; Delete rows one by one */
   }
-
   if (conds)
   {
     Item::cond_result result;
@@ -129,7 +153,19 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
       limit= 0;
   }
 
-  table->used_keys.clear_all();
+#ifdef WITH_PARTITION_STORAGE_ENGINE
+  if (prune_partitions(thd, table, conds))
+  {
+    free_underlaid_joins(thd, select_lex);
+    thd->row_count_func= 0;
+    my_ok(thd, (ha_rows) thd->row_count_func);  // No matching records
+    DBUG_RETURN(0);
+  }
+#endif
+  /* Update the table->file->stats.records number */
+  table->file->info(HA_STATUS_VARIABLE | HA_STATUS_NO_LOCK);
+
+  table->covering_keys.clear_all();
   table->quick_keys.clear_all();		// Can't use 'only index'
   select=make_select(table, 0, 0, conds, 0, &error);
   if (error)
@@ -139,14 +175,12 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
     delete select;
     free_underlaid_joins(thd, select_lex);
     thd->row_count_func= 0;
-    send_ok(thd,0L);
-
+    my_ok(thd, (ha_rows) thd->row_count_func);
     /*
       We don't need to call reset_auto_increment in this case, because
       mysql_truncate always gives a NULL conds argument, hence we never
       get here.
     */
-
     DBUG_RETURN(0);				// Nothing to delete
   }
 
@@ -183,7 +217,7 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
       if (!(sortorder= make_unireg_sortorder((ORDER*) order->first,
                                              &length, NULL)) ||
 	  (table->sort.found_records = filesort(thd, table, sortorder, length,
-                                                select, HA_POS_ERROR,
+                                                select, HA_POS_ERROR, 1,
                                                 &examined_rows))
 	  == HA_POS_ERROR)
       {
@@ -213,30 +247,31 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
   else
     init_read_record_idx(&info, thd, table, 1, usable_index);
 
-  deleted=0L;
   init_ftfuncs(thd, select_lex, 1);
-  thd->proc_info="updating";
-
-  if (table->triggers)
+  thd_proc_info(thd, "updating");
+  if (table->triggers && 
+      table->triggers->has_triggers(TRG_EVENT_DELETE,
+                                    TRG_ACTION_AFTER))
   {
-    table->triggers->mark_fields_used(thd, TRG_EVENT_DELETE);
-    if (table->triggers->has_triggers(TRG_EVENT_DELETE,
-                                      TRG_ACTION_AFTER))
-    {
-      /*
-	The table has AFTER DELETE triggers that might access to subject table
-	and therefore might need delete to be done immediately. So we turn-off
-	the batching.
-      */
-      (void) table->file->extra(HA_EXTRA_DELETE_CANNOT_BATCH);
-    }
+    /*
+      The table has AFTER DELETE triggers that might access to subject table
+      and therefore might need delete to be done immediately. So we turn-off
+      the batching.
+    */
+    (void) table->file->extra(HA_EXTRA_DELETE_CANNOT_BATCH);
+    will_batch= FALSE;
   }
+  else
+    will_batch= !table->file->start_bulk_delete();
+
+
+  table->mark_columns_needed_for_delete();
 
   while (!(error=info.read_record(&info)) && !thd->killed &&
-	 !thd->net.report_error)
+	 ! thd->is_error())
   {
-    // thd->net.report_error is tested to disallow delete row on error
-    if (!(select && select->skip_record())&& !thd->net.report_error )
+    // thd->is_error() is tested to disallow delete row on error
+    if (!(select && select->skip_record())&& ! thd->is_error() )
     {
 
       if (table->triggers &&
@@ -247,7 +282,7 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
         break;
       }
 
-      if (!(error=table->file->delete_row(table->record[0])))
+      if (!(error= table->file->ha_delete_row(table->record[0])))
       {
 	deleted++;
         if (table->triggers &&
@@ -282,10 +317,16 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
       table->file->unlock_row();  // Row failed selection, release lock on it
   }
   killed_status= thd->killed;
-  error= (killed_status == THD::NOT_KILLED)?  error : 1;
-  thd->proc_info="end";
+  if (killed_status != THD::NOT_KILLED || thd->is_error())
+    error= 1;					// Aborted
+  if (will_batch && (loc_error= table->file->end_bulk_delete()))
+  {
+    if (error != 1)
+      table->file->print_error(loc_error,MYF(0));
+    error=1;
+  }
+  thd_proc_info(thd, "end");
   end_read_record(&info);
-  free_io_cache(table);				// Will not do any harm
   if (options & OPTION_QUICK)
     (void) table->file->extra(HA_EXTRA_NORMAL);
 
@@ -295,7 +336,7 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
       We're really doing a truncate and need to reset the table's
       auto-increment counter.
     */
-    int error2= table->file->reset_auto_increment(0);
+    int error2= table->file->ha_reset_auto_increment(0);
 
     if (error2 && (error2 != HA_ERR_WRONG_COMMAND))
     {
@@ -316,6 +357,7 @@ cleanup:
 
   delete select;
   transactional_table= table->file->has_transactions();
+
   if (!transactional_table && deleted > 0)
     thd->transaction.stmt.modified_non_trans_table= TRUE;
   
@@ -326,34 +368,33 @@ cleanup:
     {
       if (error < 0)
         thd->clear_error();
-      Query_log_event qinfo(thd, thd->query, thd->query_length,
-			    transactional_table, FALSE, killed_status);
-      if (mysql_bin_log.write(&qinfo) && transactional_table)
+      /*
+        [binlog]: If 'handler::delete_all_rows()' was called and the
+        storage engine does not inject the rows itself, we replicate
+        statement-based; otherwise, 'ha_delete_row()' was used to
+        delete specific rows which we might log row-based.
+      */
+      int log_result= thd->binlog_query(THD::ROW_QUERY_TYPE,
+                                        thd->query, thd->query_length,
+                                        transactional_table, FALSE, killed_status);
+
+      if (log_result && transactional_table)
+      {
 	error=1;
+      }
     }
     if (thd->transaction.stmt.modified_non_trans_table)
       thd->transaction.all.modified_non_trans_table= TRUE;
   }
   DBUG_ASSERT(transactional_table || !deleted || thd->transaction.stmt.modified_non_trans_table);
   free_underlaid_joins(thd, select_lex);
-  if (transactional_table)
-  {
-    if (ha_autocommit_or_rollback(thd,error >= 0))
-      error=1;
-  }
-
-  if (thd->lock)
-  {
-    mysql_unlock_tables(thd, thd->lock);
-    thd->lock=0;
-  }
   if (error < 0 || (thd->lex->ignore && !thd->is_fatal_error))
   {
     thd->row_count_func= deleted;
-    send_ok(thd,deleted);
+    my_ok(thd, (ha_rows) thd->row_count_func);
     DBUG_PRINT("info",("%ld records deleted",(long) deleted));
   }
-  DBUG_RETURN(error >= 0 || thd->net.report_error);
+  DBUG_RETURN(error >= 0 || thd->is_error());
 }
 
 
@@ -380,7 +421,7 @@ int mysql_prepare_delete(THD *thd, TABLE_LIST *table_list, Item **conds)
   thd->lex->allow_sum_func= 0;
   if (setup_tables_and_check_access(thd, &thd->lex->select_lex.context,
                                     &thd->lex->select_lex.top_join_list,
-                                    table_list, conds, 
+                                    table_list, 
                                     &select_lex->leaf_tables, FALSE, 
                                     DELETE_ACL, SELECT_ACL) ||
       setup_conds(thd, table_list, select_lex->leaf_tables, conds) ||
@@ -418,7 +459,7 @@ int mysql_prepare_delete(THD *thd, TABLE_LIST *table_list, Item **conds)
 extern "C" int refpos_order_cmp(void* arg, const void *a,const void *b)
 {
   handler *file= (handler*)arg;
-  return file->cmp_ref((const byte*)a, (const byte*)b);
+  return file->cmp_ref((const uchar*)a, (const uchar*)b);
 }
 
 /*
@@ -448,7 +489,7 @@ int mysql_multi_delete_prepare(THD *thd)
   */
   if (setup_tables_and_check_access(thd, &thd->lex->select_lex.context,
                                     &thd->lex->select_lex.top_join_list,
-                                    lex->query_tables, &lex->select_lex.where,
+                                    lex->query_tables,
                                     &lex->select_lex.leaf_tables, FALSE, 
                                     DELETE_ACL, SELECT_ACL))
     DBUG_RETURN(TRUE);
@@ -517,7 +558,7 @@ multi_delete::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
   DBUG_ENTER("multi_delete::prepare");
   unit= u;
   do_delete= 1;
-  thd->proc_info="deleting from main table";
+  thd_proc_info(thd, "deleting from main table");
   DBUG_RETURN(0);
 }
 
@@ -551,25 +592,24 @@ multi_delete::initialize_tables(JOIN *join)
       tbl->no_keyread=1;
       /* Don't use record cache */
       tbl->no_cache= 1;
-      tbl->used_keys.clear_all();
+      tbl->covering_keys.clear_all();
       if (tbl->file->has_transactions())
 	transactional_tables= 1;
       else
 	normal_tables= 1;
-      if (tbl->triggers)
-      {
-        tbl->triggers->mark_fields_used(thd, TRG_EVENT_DELETE);
-	if (tbl->triggers->has_triggers(TRG_EVENT_DELETE,
+      if (tbl->triggers &&
+          tbl->triggers->has_triggers(TRG_EVENT_DELETE,
                                       TRG_ACTION_AFTER))
-	{
-	  /*
-	    The table has AFTER DELETE triggers that might access to subject 
-	    table and therefore might need delete to be done immediately. 
-	    So we turn-off the batching.
-	  */
-	  (void) tbl->file->extra(HA_EXTRA_DELETE_CANNOT_BATCH);
-	}
+      {
+	/*
+          The table has AFTER DELETE triggers that might access to subject 
+          table and therefore might need delete to be done immediately. 
+          So we turn-off the batching.
+        */
+	(void) tbl->file->extra(HA_EXTRA_DELETE_CANNOT_BATCH);
       }
+      tbl->prepare_for_position();
+      tbl->mark_columns_needed_for_delete();
     }
     else if ((tab->type != JT_SYSTEM && tab->type != JT_CONST) &&
              walk == delete_tables)
@@ -609,7 +649,6 @@ multi_delete::~multi_delete()
        table_being_deleted= table_being_deleted->next_local)
   {
     TABLE *table= table_being_deleted->table;
-    free_io_cache(table);                       // Alloced by unique
     table->no_keyread=0;
   }
 
@@ -649,7 +688,7 @@ bool multi_delete::send_data(List<Item> &values)
                                             TRG_ACTION_BEFORE, FALSE))
         DBUG_RETURN(1);
       table->status|= STATUS_DELETED;
-      if (!(error=table->file->delete_row(table->record[0])))
+      if (!(error=table->file->ha_delete_row(table->record[0])))
       {
         deleted++;
         if (!table->file->has_transactions())
@@ -686,6 +725,14 @@ void multi_delete::send_error(uint errcode,const char *err)
   /* First send error what ever it is ... */
   my_message(errcode, err, MYF(0));
 
+  DBUG_VOID_RETURN;
+}
+
+
+void multi_delete::abort()
+{
+  DBUG_ENTER("multi_delete::abort");
+
   /* the error was handled or nothing deleted and no side effects return */
   if (error_handled ||
       !thd->transaction.stmt.modified_non_trans_table && !deleted)
@@ -701,11 +748,9 @@ void multi_delete::send_error(uint errcode,const char *err)
     The same if all tables are transactional, regardless of where we are.
     In all other cases do attempt deletes ...
   */
-  if ((table_being_deleted == delete_tables &&
-       table_being_deleted->table->file->has_transactions()) ||
-      !normal_tables)
-    ha_rollback_stmt(thd);
-  else if (do_delete)
+  if (do_delete && normal_tables &&
+      (table_being_deleted != delete_tables ||
+       !table_being_deleted->table->file->has_transactions()))
   {
     /*
       We have to execute the recorded do_deletes() and write info into the
@@ -724,9 +769,9 @@ void multi_delete::send_error(uint errcode,const char *err)
     */
     if (mysql_bin_log.is_open())
     {
-      Query_log_event qinfo(thd, thd->query, thd->query_length,
-                            transactional_tables, FALSE);
-      mysql_bin_log.write(&qinfo);
+      thd->binlog_query(THD::ROW_QUERY_TYPE,
+                        thd->query, thd->query_length,
+                        transactional_tables, FALSE);
     }
     thd->transaction.all.modified_non_trans_table= true;
   }
@@ -746,7 +791,8 @@ void multi_delete::send_error(uint errcode,const char *err)
 
 int multi_delete::do_deletes()
 {
-  int local_error= 0, counter= 0;
+  int local_error= 0, counter= 0, tmp_error;
+  bool will_batch;
   DBUG_ENTER("do_deletes");
   DBUG_ASSERT(do_delete);
 
@@ -775,6 +821,7 @@ int multi_delete::do_deletes()
       been deleted by foreign key handling
     */
     info.ignore_not_found_rows= 1;
+    will_batch= !table->file->start_bulk_delete();
     while (!(local_error=info.read_record(&info)) && !thd->killed)
     {
       if (table->triggers &&
@@ -784,7 +831,7 @@ int multi_delete::do_deletes()
         local_error= 1;
         break;
       }
-      if ((local_error=table->file->delete_row(table->record[0])))
+      if ((local_error=table->file->ha_delete_row(table->record[0])))
       {
 	table->file->print_error(local_error,MYF(0));
 	break;
@@ -798,6 +845,14 @@ int multi_delete::do_deletes()
         break;
       }
     }
+    if (will_batch && (tmp_error= table->file->end_bulk_delete()))
+    {
+      if (!local_error)
+      {
+        local_error= tmp_error;
+        table->file->print_error(local_error,MYF(0));
+      }
+    }
     if (last_deleted != deleted && !table->file->has_transactions())
       thd->transaction.stmt.modified_non_trans_table= TRUE;
     end_read_record(&info);
@@ -820,7 +875,7 @@ int multi_delete::do_deletes()
 bool multi_delete::send_eof()
 {
   THD::killed_state killed_status= THD::NOT_KILLED;
-  thd->proc_info="deleting from reference tables";
+  thd_proc_info(thd, "deleting from reference tables");
 
   /* Does deletes for the last n - 1 tables, returns 0 if ok */
   int local_error= do_deletes();		// returns 0 if success
@@ -829,7 +884,7 @@ bool multi_delete::send_eof()
   local_error= local_error || error;
   killed_status= (local_error == 0)? THD::NOT_KILLED : thd->killed;
   /* reset used flags */
-  thd->proc_info="end";
+  thd_proc_info(thd, "end");
 
   /*
     We must invalidate the query cache before binlog writing and
@@ -847,10 +902,13 @@ bool multi_delete::send_eof()
     {
       if (local_error == 0)
         thd->clear_error();
-      Query_log_event qinfo(thd, thd->query, thd->query_length,
-			    transactional_tables, FALSE, killed_status);
-      if (mysql_bin_log.write(&qinfo) && !normal_tables)
+      if (thd->binlog_query(THD::ROW_QUERY_TYPE,
+                            thd->query, thd->query_length,
+                            transactional_tables, FALSE, killed_status) &&
+          !normal_tables)
+      {
 	local_error=1;  // Log write failed: roll back the SQL statement
+      }
     }
     if (thd->transaction.stmt.modified_non_trans_table)
       thd->transaction.all.modified_non_trans_table= TRUE;
@@ -858,15 +916,10 @@ bool multi_delete::send_eof()
   if (local_error != 0)
     error_handled= TRUE; // to force early leave from ::send_error()
 
-  /* Commit or rollback the current SQL statement */
-  if (transactional_tables)
-    if (ha_autocommit_or_rollback(thd,local_error > 0))
-      local_error=1;
-
   if (!local_error)
   {
     thd->row_count_func= deleted;
-    ::send_ok(thd, deleted);
+    ::my_ok(thd, (ha_rows) thd->row_count_func);
   }
   return 0;
 }
@@ -892,45 +945,45 @@ bool mysql_truncate(THD *thd, TABLE_LIST *table_list, bool dont_send_ok)
 {
   HA_CREATE_INFO create_info;
   char path[FN_REFLEN];
-  TABLE **table_ptr;
+  TABLE *table;
   bool error;
+  uint path_length;
   DBUG_ENTER("mysql_truncate");
 
   bzero((char*) &create_info,sizeof(create_info));
   /* If it is a temporary table, close and regenerate it */
-  if (!dont_send_ok && (table_ptr=find_temporary_table(thd,table_list->db,
-						       table_list->table_name)))
+  if (!dont_send_ok && (table= find_temporary_table(thd, table_list)))
   {
-    TABLE *table= *table_ptr;
-    table->file->info(HA_STATUS_AUTO | HA_STATUS_NO_LOCK);
-    db_type table_type= table->s->db_type;
+    handlerton *table_type= table->s->db_type();
+    TABLE_SHARE *share= table->s;
     if (!ha_check_storage_engine_flag(table_type, HTON_CAN_RECREATE))
       goto trunc_by_del;
-    strmov(path, table->s->path);
-    *table_ptr= table->next;			// Unlink table from list
-    close_temporary(table,0);
-    if (thd->slave_thread)
-      --slave_open_temp_tables;
-    *fn_ext(path)=0;				// Remove the .frm extension
-    ha_create_table(path, &create_info,1);
+
+    table->file->info(HA_STATUS_AUTO | HA_STATUS_NO_LOCK);
+    
+    close_temporary_table(thd, table, 0, 0);    // Don't free share
+    ha_create_table(thd, share->normalized_path.str,
+                    share->db.str, share->table_name.str, &create_info, 1);
     // We don't need to call invalidate() because this table is not in cache
-    if ((error= (int) !(open_temporary_table(thd, path, table_list->db,
-					     table_list->table_name, 1))))
+    if ((error= (int) !(open_temporary_table(thd, share->path.str,
+                                             share->db.str,
+					     share->table_name.str, 1))))
       (void) rm_temporary_table(table_type, path);
+    free_table_share(share);
+    my_free((char*) table,MYF(0));
     /*
       If we return here we will not have logged the truncation to the bin log
-      and we will not send_ok() to the client.
+      and we will not my_ok() to the client.
     */
     goto end;
   }
 
-  (void) sprintf(path,"%s/%s/%s%s",mysql_data_home,table_list->db,
-		 table_list->table_name,reg_ext);
-  fn_format(path, path, "", "", MY_UNPACK_FILENAME);
+  path_length= build_table_filename(path, sizeof(path), table_list->db,
+                                    table_list->table_name, reg_ext, 0);
 
   if (!dont_send_ok)
   {
-    db_type table_type;
+    enum legacy_db_type table_type;
     mysql_frm_type(thd, path, &table_type);
     if (table_type == DB_TYPE_UNKNOWN)
     {
@@ -938,14 +991,22 @@ bool mysql_truncate(THD *thd, TABLE_LIST *table_list, bool dont_send_ok)
                table_list->db, table_list->table_name);
       DBUG_RETURN(TRUE);
     }
-    if (!ha_check_storage_engine_flag(table_type, HTON_CAN_RECREATE))
+    if (!ha_check_storage_engine_flag(ha_resolve_by_legacy_type(thd, table_type),
+                                      HTON_CAN_RECREATE))
       goto trunc_by_del;
+
     if (lock_and_wait_for_table_name(thd, table_list))
       DBUG_RETURN(TRUE);
   }
 
-  *fn_ext(path)=0;				// Remove the .frm extension
-  error= ha_create_table(path,&create_info,1);
+  // Remove the .frm extension AIX 5.2 64-bit compiler bug (BUG#16155): this
+  // crashes, replacement works.  *(path + path_length - reg_ext_length)=
+  // '\0';
+  path[path_length - reg_ext_length] = 0;
+  VOID(pthread_mutex_lock(&LOCK_open));
+  error= ha_create_table(thd, path, table_list->db, table_list->table_name,
+                         &create_info, 1);
+  VOID(pthread_mutex_unlock(&LOCK_open));
   query_cache_invalidate3(thd, table_list, 0);
 
 end:
@@ -953,14 +1014,12 @@ end:
   {
     if (!error)
     {
-      if (mysql_bin_log.is_open())
-      {
-        thd->clear_error();
-	Query_log_event qinfo(thd, thd->query, thd->query_length,
-			      0, FALSE);
-	mysql_bin_log.write(&qinfo);
-      }
-      send_ok(thd);		// This should return record count
+      /*
+        TRUNCATE must always be statement-based binlogged (not row-based) so
+        we don't test current_stmt_binlog_row_based.
+      */
+      write_bin_log(thd, TRUE, thd->query, thd->query_length);
+      my_ok(thd);		// This should return record count
     }
     VOID(pthread_mutex_lock(&LOCK_open));
     unlock_table_name(thd, table_list);
@@ -974,16 +1033,25 @@ end:
   }
   DBUG_RETURN(error);
 
- trunc_by_del:
+trunc_by_del:
   /* Probably InnoDB table */
   ulonglong save_options= thd->options;
   table_list->lock_type= TL_WRITE;
-  thd->options&= ~(ulong) (OPTION_BEGIN | OPTION_NOT_AUTOCOMMIT);
+  thd->options&= ~(OPTION_BEGIN | OPTION_NOT_AUTOCOMMIT);
   ha_enable_transaction(thd, FALSE);
   mysql_init_select(thd->lex);
+  bool save_binlog_row_based= thd->current_stmt_binlog_row_based;
+  thd->clear_current_stmt_binlog_row_based();
   error= mysql_delete(thd, table_list, (COND*) 0, (SQL_LIST*) 0,
                       HA_POS_ERROR, LL(0), TRUE);
   ha_enable_transaction(thd, TRUE);
+  /*
+    Safety, in case the engine ignored ha_enable_transaction(FALSE)
+    above. Also clears thd->transaction.*.
+  */
+  error= ha_autocommit_or_rollback(thd, error);
+  ha_commit(thd);
   thd->options= save_options;
+  thd->current_stmt_binlog_row_based= save_binlog_row_based;
   DBUG_RETURN(error);
 }