Bug#42643: InnoDB does not support replication of TRUNCATE TABLE

The problem was that TRUNCATE TABLE didn't take a exclusive
lock on a table if it resorted to truncating via delete of
all rows in the table. Specifically for InnoDB tables, this
could break proper isolation as InnoDB ends up aborting some
granted locks when truncating a table.

The solution is to take a exclusive metadata lock before
TRUNCATE TABLE can proceed. This guarantees that no other
transaction is using the table.

Incompatible change: Truncate via delete no longer fails
if sql_safe_updates is activated (this was a undocumented
side effect).

libmysqld/CMakeLists.txt:
  Add new files to the build list.
libmysqld/Makefile.am:
  Add new files to the build list.
mysql-test/extra/binlog_tests/binlog_truncate.test:
  Add test case for Bug#42643
mysql-test/include/mix1.inc:
  Update test case as TRUNCATE TABLE now grabs a exclusive lock.
  Ensure that TRUNCATE waits for granted locks on the table.
mysql-test/suite/binlog/t/binlog_truncate_innodb.test:
  As with other data modifying statements, TRUNCATE is still not
  possible in a transaction with isolation level READ COMMITTED
  or READ UNCOMMITED. It would be possible to implement so, but
  it is not worth the effort.
mysql-test/suite/binlog/t/binlog_truncate_myisam.test:
  Test under different binlog formats.
mysql-test/suite/binlog/t/disabled.def:
  Re-enable test case.
mysql-test/t/innodb_bug38231.test:
  Truncate no longer works with row-level locks.
mysql-test/t/mdl_sync.test:
  Ensure that a acquired lock is not given up due to a conflict.
mysql-test/t/partition_innodb_semi_consistent.test:
  End transaction as to release metadata locks.
mysql-test/t/truncate.test:
  A metadata lock is now taken before the object is verified.
sql/CMakeLists.txt:
  Add new files to the build list.
sql/Makefile.am:
  Add new files to the build list.
sql/datadict.cc:
  Introduce a new file specific for data dictionary operations.
sql/datadict.h:
  Add header file.
sql/sql_base.cc:
  Rename data dictionary function.
sql/sql_bitmap.h:
  Include dependency.
sql/sql_delete.cc:
  Move away from relying on mysql_delete() to delete all rows of
  a table. Thus, move any bits related to truncate to sql_truncate.cc
sql/sql_delete.h:
  Remove parameter.
sql/sql_parse.cc:
  Add protection against the global read lock -- a intention
  exclusive lock can be acquired in the truncate path.
sql/sql_show.cc:
  Add sync point for testing scenarios where a pending flush
  is ignored.
sql/sql_truncate.cc:
  Acquire a shared metadata lock before accessing table metadata.
  Upgrade the lock to a exclusive one if the table can be re-created.
  Rework binlog rules to better reflect the requirements.
sql/sql_yacc.yy:
  Set appropriate lock types for table to be truncated.
sql/table.h:
  Move to data dictionary header.

19 files changed, 751 insertions, 388 deletions

diff --git a/sql/CMakeLists.txt b/sql/CMakeLists.txt
index 357575c5b47..87461be88f3 100755
--- a/sql/CMakeLists.txt
+++ b/sql/CMakeLists.txt
@@ -75,7 +75,7 @@ SET (SQL_SOURCE
                sql_connect.cc scheduler.cc 
                sql_profile.cc event_parse_data.cc
                sql_signal.cc rpl_handler.cc mdl.cc
-               transaction.cc sys_vars.cc
+               transaction.cc sys_vars.cc sql_truncate.cc datadict.cc
                ${GEN_SOURCES}
                ${MYSYS_LIBWRAP_SOURCE})
 
diff --git a/sql/Makefile.am b/sql/Makefile.am
index 93595a964eb..0fba8d12406 100644
--- a/sql/Makefile.am
+++ b/sql/Makefile.am
@@ -39,7 +39,9 @@ DTRACEFILES =           filesort.o \
                         sql_connect.o \
                         sql_cursor.o \
                         sql_delete.o \
+                        sql_truncate.o \
                         sql_insert.o \
+                        datadict.o \
                         sql_parse.o \
                         sql_prepare.o \
                         sql_select.o \
@@ -56,7 +58,9 @@ DTRACEFILES_DEPEND =    filesort.o \
                         sql_connect.o \
                         sql_cursor.o \
                         sql_delete.o \
+                        sql_truncate.o \
                         sql_insert.o \
+                        datadict.o \
                         sql_parse.o \
                         sql_prepare.o \
                         sql_select.o \
@@ -121,7 +125,8 @@ noinst_HEADERS =	item.h item_func.h item_sum.h item_cmpfunc.h \
 			sql_audit.h \
 			contributors.h sql_servers.h sql_signal.h records.h \
 			sql_prepare.h rpl_handler.h replication.h mdl.h \
-			sql_plist.h transaction.h sys_vars.h
+			sql_plist.h transaction.h sys_vars.h sql_truncate.h \
+			datadict.h
 
 mysqld_SOURCES =	sql_lex.cc sql_handler.cc sql_partition.cc \
 			item.cc item_sum.cc item_buff.cc item_func.cc \
@@ -136,10 +141,10 @@ mysqld_SOURCES =	sql_lex.cc sql_handler.cc sql_partition.cc \
 			sql_connect.cc scheduler.cc sql_parse.cc \
 			keycaches.cc set_var.cc sql_yacc.yy sys_vars.cc \
 			sql_base.cc table.cc sql_select.cc sql_insert.cc \
-			sql_profile.cc \
+			datadict.cc sql_profile.cc \
 			sql_prepare.cc sql_error.cc sql_locale.cc \
 			sql_update.cc sql_delete.cc uniques.cc sql_do.cc \
-			procedure.cc sql_test.cc \
+			procedure.cc sql_test.cc sql_truncate.cc \
 			log.cc init.cc derror.cc sql_acl.cc \
 			unireg.cc des_key_file.cc \
 			log_event.cc rpl_record.cc \
diff --git a/sql/datadict.cc b/sql/datadict.cc
new file mode 100644
index 00000000000..5067c13a255
--- /dev/null
+++ b/sql/datadict.cc
@@ -0,0 +1,161 @@
+/* Copyright (c) 2010, Oracle and/or its affiliates. All rights reserved.
+
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; version 2 of the License.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA */
+
+#include "datadict.h"
+#include "sql_priv.h"
+#include "sql_class.h"
+#include "sql_table.h"
+
+
+/**
+  Check type of .frm if we are not going to parse it.
+
+  @param  path  path to FRM file
+
+  @retval  FRMTYPE_ERROR        error
+  @retval  FRMTYPE_TABLE        table
+  @retval  FRMTYPE_VIEW         view
+*/
+
+frm_type_enum dd_frm_type(THD *thd, char *path, enum legacy_db_type *dbt)
+{
+  File file;
+  uchar header[10];     //"TYPE=VIEW\n" it is 10 characters
+  size_t error;
+  DBUG_ENTER("dd_frm_type");
+
+  *dbt= DB_TYPE_UNKNOWN;
+
+  if ((file= mysql_file_open(key_file_frm, path, O_RDONLY | O_SHARE, MYF(0))) < 0)
+    DBUG_RETURN(FRMTYPE_ERROR);
+  error= mysql_file_read(file, (uchar*) header, sizeof(header), MYF(MY_NABP));
+  mysql_file_close(file, MYF(MY_WME));
+
+  if (error)
+    DBUG_RETURN(FRMTYPE_ERROR);
+  if (!strncmp((char*) header, "TYPE=VIEW\n", sizeof(header)))
+    DBUG_RETURN(FRMTYPE_VIEW);
+
+  /*
+    This is just a check for DB_TYPE. We'll return default unknown type
+    if the following test is true (arg #3). This should not have effect
+    on return value from this function (default FRMTYPE_TABLE)
+  */
+  if (header[0] != (uchar) 254 || header[1] != 1 ||
+      (header[2] != FRM_VER && header[2] != FRM_VER+1 &&
+       (header[2] < FRM_VER+3 || header[2] > FRM_VER+4)))
+    DBUG_RETURN(FRMTYPE_TABLE);
+
+  *dbt= (enum legacy_db_type) (uint) *(header + 3);
+
+  /* Probably a table. */
+  DBUG_RETURN(FRMTYPE_TABLE);
+}
+
+
+/**
+  Given a table name, check if the storage engine for the
+  table referred by this name supports an option 'flag'.
+  Return an error if the table does not exist or is not a
+  base table.
+
+  @pre Any metadata lock on the table.
+
+  @param[in]    thd         The current session.
+  @param[in]    db          Table schema.
+  @param[in]    table_name  Table database.
+  @param[in]    flag        The option to check.
+  @param[out]   yes_no      The result. Undefined if error.
+*/
+
+bool dd_check_storage_engine_flag(THD *thd,
+                                  const char *db, const char *table_name,
+                                  uint32 flag, bool *yes_no)
+{
+  char path[FN_REFLEN + 1];
+  enum legacy_db_type db_type;
+  handlerton *table_type;
+  LEX_STRING db_name = {(char *) db, strlen(db)};
+
+  if (check_db_name(&db_name))
+  {
+    my_error(ER_WRONG_DB_NAME, MYF(0), db_name.str);
+    return TRUE;
+  }
+
+  if (check_table_name(table_name, strlen(table_name)))
+  {
+    my_error(ER_WRONG_TABLE_NAME, MYF(0), table_name);
+    return TRUE;
+  }
+
+  /* There should be at least some lock on the table.  */
+  DBUG_ASSERT(thd->mdl_context.is_lock_owner(MDL_key::TABLE, db,
+                                             table_name, MDL_SHARED));
+
+  (void) build_table_filename(path, sizeof(path) - 1, db,
+                              table_name, reg_ext, 0);
+
+  dd_frm_type(thd, path, &db_type);
+
+  /* Type is unknown if the object is not found or is not a table. */
+  if (db_type == DB_TYPE_UNKNOWN)
+  {
+    my_error(ER_NO_SUCH_TABLE, MYF(0), db, table_name);
+    return TRUE;
+  }
+
+  table_type= ha_resolve_by_legacy_type(thd, db_type);
+  *yes_no= ha_check_storage_engine_flag(table_type, flag);
+
+  return FALSE;
+}
+
+
+/*
+  Regenerate a metadata locked table.
+
+  @param  thd   Thread context.
+  @param  db    Name of the database to which the table belongs to.
+  @param  name  Table name.
+
+  @retval  FALSE  Success.
+  @retval  TRUE   Error.
+*/
+
+bool dd_recreate_table(THD *thd, const char *db, const char *table_name)
+{
+  bool error= TRUE;
+  HA_CREATE_INFO create_info;
+  char path[FN_REFLEN + 1];
+  DBUG_ENTER("dd_recreate_table");
+
+  /* There should be a exclusive metadata lock on the table. */
+  DBUG_ASSERT(thd->mdl_context.is_lock_owner(MDL_key::TABLE, db, table_name,
+                                             MDL_EXCLUSIVE));
+
+  memset(&create_info, 0, sizeof(create_info));
+
+  /* Create a path to the table, but without a extension. */
+  build_table_filename(path, sizeof(path) - 1, db, table_name, "", 0);
+
+  /* Attempt to reconstruct the table. */
+  mysql_mutex_lock(&LOCK_open);
+  error= ha_create_table(thd, path, db, table_name, &create_info, TRUE);
+  mysql_mutex_unlock(&LOCK_open);
+
+  DBUG_RETURN(error);
+}
+
diff --git a/sql/datadict.h b/sql/datadict.h
new file mode 100644
index 00000000000..05b5a9bba4b
--- /dev/null
+++ b/sql/datadict.h
@@ -0,0 +1,40 @@
+#ifndef DATADICT_INCLUDED
+#define DATADICT_INCLUDED
+/* Copyright (c) 2010, Oracle and/or its affiliates. All rights reserved.
+
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; version 2 of the License.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA */
+
+#include "handler.h"
+
+/*
+  Data dictionary API.
+*/
+
+enum frm_type_enum
+{
+  FRMTYPE_ERROR= 0,
+  FRMTYPE_TABLE,
+  FRMTYPE_VIEW
+};
+
+
+frm_type_enum dd_frm_type(THD *thd, char *path, enum legacy_db_type *dbt);
+
+bool dd_check_storage_engine_flag(THD *thd,
+                                  const char *db, const char *table_name,
+                                  uint32 flag,
+                                  bool *yes_no);
+bool dd_recreate_table(THD *thd, const char *db, const char *table_name);
+
+#endif // DATADICT_INCLUDED
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 85017886d24..fa484abf0be 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -27,7 +27,7 @@
 #include "sql_show.h"    // append_identifier
 #include "strfunc.h"     // find_type
 #include "parse_file.h"  // sql_parse_prepare, File_parser
-#include "sql_view.h"    // mysql_frm_type, mysql_make_view, VIEW_ANY_ACL
+#include "sql_view.h"    // mysql_make_view, VIEW_ANY_ACL
 #include "sql_parse.h"   // check_table_access
 #include "sql_insert.h"  // kill_delayed_threads
 #include "sql_acl.h"     // *_ACL, check_grant_all_columns,
@@ -52,6 +52,7 @@
 #include <hash.h>
 #include "rpl_filter.h"
 #include "sql_table.h"                          // build_table_filename
+#include "datadict.h"   // dd_frm_type()
 #ifdef  __WIN__
 #include <io.h>
 #endif
@@ -2678,7 +2679,7 @@ bool open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
         during prelocking process (in this case in theory we still
         should hold shared metadata lock on it).
       */
-      if (mysql_frm_type(thd, path, &not_used) == FRMTYPE_VIEW)
+      if (dd_frm_type(thd, path, &not_used) == FRMTYPE_VIEW)
       {
         if (!tdc_open_view(thd, table_list, alias, key, key_length,
                            mem_root, 0))
diff --git a/sql/sql_bitmap.h b/sql/sql_bitmap.h
index 80a4712dd69..8d00c984d14 100644
--- a/sql/sql_bitmap.h
+++ b/sql/sql_bitmap.h
@@ -22,6 +22,7 @@
 #ifndef SQL_BITMAP_INCLUDED
 #define SQL_BITMAP_INCLUDED
 
+#include <my_sys.h>
 #include <my_bitmap.h>
 
 template <uint default_width> class Bitmap
diff --git a/sql/sql_delete.cc b/sql/sql_delete.cc
index 10bdb8a22a6..730cb711a05 100644
--- a/sql/sql_delete.cc
+++ b/sql/sql_delete.cc
@@ -14,7 +14,7 @@
    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
 
 /*
-  Delete of records and truncate of tables.
+  Delete of records tables.
 
   Multi-table deletes were introduced by Monty and Sinisa
 */
@@ -48,8 +48,7 @@
 */
 
 bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
-                  SQL_LIST *order, ha_rows limit, ulonglong options,
-                  bool reset_auto_increment)
+                  SQL_LIST *order, ha_rows limit, ulonglong options)
 {
   bool          will_batch;
   int		error, loc_error;
@@ -60,17 +59,11 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
   bool		transactional_table, safe_update, const_cond;
   bool          const_cond_result;
   ha_rows	deleted= 0;
-  bool          triggers_applicable;
   uint usable_index= MAX_KEY;
   SELECT_LEX   *select_lex= &thd->lex->select_lex;
   THD::killed_state killed_status= THD::NOT_KILLED;
+  THD::enum_binlog_query_type query_type= THD::ROW_QUERY_TYPE;
   DBUG_ENTER("mysql_delete");
-  bool save_binlog_row_based;
-
-  THD::enum_binlog_query_type query_type=
-    thd->lex->sql_command == SQLCOM_TRUNCATE ?
-    THD::STMT_QUERY_TYPE :
-    THD::ROW_QUERY_TYPE;
 
   if (open_and_lock_tables(thd, table_list, TRUE, 0))
     DBUG_RETURN(TRUE);
@@ -130,25 +123,20 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
     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 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
+    - If the current command is DELETE FROM with no where clause, then:
+      - We should not be binlogging this statement in row-based, and
       - there should be no delete triggers associated with the table.
   */
   if (!using_limit && const_cond_result &&
       !(specialflag & (SPECIAL_NO_NEW_FUNC | SPECIAL_SAFE_MODE)) &&
-      (thd->lex->sql_command == SQLCOM_TRUNCATE ||
        (!thd->is_current_stmt_binlog_format_row() &&
-        !(table->triggers && table->triggers->has_delete_triggers()))))
+        !(table->triggers && table->triggers->has_delete_triggers())))
   {
     /* Update the table->file->stats.records number */
     table->file->info(HA_STATUS_VARIABLE | HA_STATUS_NO_LOCK);
@@ -161,16 +149,14 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
         query in row format, so we have to log it in statement format.
       */
       query_type= THD::STMT_QUERY_TYPE;
-      error= -1;				// ok
+      error= -1;
       deleted= maybe_deleted;
-      save_binlog_row_based= thd->is_current_stmt_binlog_format_row();
       goto cleanup;
     }
     if (error != HA_ERR_WRONG_COMMAND)
     {
       table->file->print_error(error,MYF(0));
       error=0;
-      save_binlog_row_based= thd->is_current_stmt_binlog_format_row();
       goto cleanup;
     }
     /* Handler didn't support fast delete; Delete rows one by one */
@@ -213,11 +199,6 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
     if (thd->is_error())
       DBUG_RETURN(TRUE);
     my_ok(thd, 0);
-    /*
-      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
   }
 
@@ -287,12 +268,7 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
   init_ftfuncs(thd, select_lex, 1);
   thd_proc_info(thd, "updating");
 
-  /* NOTE: TRUNCATE must not invoke triggers. */
-
-  triggers_applicable= table->triggers &&
-                       thd->lex->sql_command != SQLCOM_TRUNCATE;
-
-  if (triggers_applicable &&
+  if (table->triggers &&
       table->triggers->has_triggers(TRG_EVENT_DELETE,
                                     TRG_ACTION_AFTER))
   {
@@ -310,11 +286,6 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
 
   table->mark_columns_needed_for_delete();
 
-  save_binlog_row_based= thd->is_current_stmt_binlog_format_row();
-  if (thd->lex->sql_command == SQLCOM_TRUNCATE &&
-      thd->is_current_stmt_binlog_format_row())
-    thd->clear_current_stmt_binlog_format_row();
-
   while (!(error=info.read_record(&info)) && !thd->killed &&
 	 ! thd->is_error())
   {
@@ -322,7 +293,7 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
     if (!(select && select->skip_record())&& ! thd->is_error() )
     {
 
-      if (triggers_applicable &&
+      if (table->triggers &&
           table->triggers->process_triggers(thd, TRG_EVENT_DELETE,
                                             TRG_ACTION_BEFORE, FALSE))
       {
@@ -333,7 +304,7 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
       if (!(error= table->file->ha_delete_row(table->record[0])))
       {
 	deleted++;
-        if (triggers_applicable &&
+        if (table->triggers &&
             table->triggers->process_triggers(thd, TRG_EVENT_DELETE,
                                               TRG_ACTION_AFTER, FALSE))
         {
@@ -378,21 +349,6 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
   if (options & OPTION_QUICK)
     (void) table->file->extra(HA_EXTRA_NORMAL);
 
-  if (reset_auto_increment && (error < 0))
-  {
-    /*
-      We're really doing a truncate and need to reset the table's
-      auto-increment counter.
-    */
-    int error2= table->file->ha_reset_auto_increment(0);
-
-    if (error2 && (error2 != HA_ERR_WRONG_COMMAND))
-    {
-      table->file->print_error(error2, MYF(0));
-      error= 1;
-    }
-  }
-
 cleanup:
   /*
     Invalidate the table in the query cache if something changed. This must
@@ -413,34 +369,24 @@ cleanup:
   /* See similar binlogging code in sql_update.cc, for comments */
   if ((error < 0) || thd->transaction.stmt.modified_non_trans_table)
   {
-    if (mysql_bin_log.is_open() &&
-        !(thd->lex->sql_command == SQLCOM_TRUNCATE &&
-          thd->is_current_stmt_binlog_format_row() &&
-          find_temporary_table(thd, table_list)))
+    if (mysql_bin_log.is_open())
     {
-      bool const is_trans=
-        thd->lex->sql_command == SQLCOM_TRUNCATE ?
-        FALSE :
-        transactional_table;
-
       int errcode= 0;
       if (error < 0)
         thd->clear_error();
       else
         errcode= query_error_code(thd, killed_status == THD::NOT_KILLED);
-      
+
       /*
         [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.
-
-        Note that TRUNCATE TABLE is not transactional and should
-        therefore be treated as a DDL.
       */
       int log_result= thd->binlog_query(query_type,
                                         thd->query(), thd->query_length(),
-                                        is_trans, FALSE, FALSE, errcode);
+                                        transactional_table, FALSE, FALSE,
+                                        errcode);
 
       if (log_result)
       {
@@ -448,18 +394,12 @@ cleanup:
       }
     }
   }
-  if (save_binlog_row_based)
-    thd->set_current_stmt_binlog_format_row();
   DBUG_ASSERT(transactional_table || !deleted || thd->transaction.stmt.modified_non_trans_table);
   free_underlaid_joins(thd, select_lex);
   if (error < 0 || 
       (thd->lex->ignore && !thd->is_error() && !thd->is_fatal_error))
   {
-    /*
-      If a TRUNCATE TABLE was issued, the number of rows should be reported as
-      zero since the exact number is unknown.
-    */
-    my_ok(thd, reset_auto_increment ? 0 : deleted);
+    my_ok(thd, deleted);
     DBUG_PRINT("info",("%ld records deleted",(long) deleted));
   }
   DBUG_RETURN(error >= 0 || thd->is_error());
@@ -1061,227 +1001,3 @@ bool multi_delete::send_eof()
   return 0;
 }
 
-
-/***************************************************************************
-  TRUNCATE TABLE
-****************************************************************************/
-
-/*
-  Row-by-row truncation if the engine does not support table recreation.
-  Probably a InnoDB table.
-*/
-
-static bool mysql_truncate_by_delete(THD *thd, TABLE_LIST *table_list)
-{
-  bool error;
-  DBUG_ENTER("mysql_truncate_by_delete");
-  table_list->lock_type= TL_WRITE;
-  table_list->mdl_request.set_type(MDL_SHARED_WRITE);
-  mysql_init_select(thd->lex);
-  /* Delete all rows from table */
-  error= mysql_delete(thd, table_list, NULL, NULL, HA_POS_ERROR, LL(0), TRUE);
-  /*
-    All effects of a TRUNCATE TABLE operation are rolled back if a row by row
-    deletion fails. Otherwise, operation is automatically committed at the end.
-  */
-  if (error)
-  {
-    DBUG_ASSERT(thd->stmt_da->is_error());
-    trans_rollback_stmt(thd);
-    trans_rollback(thd);
-  }
-  DBUG_RETURN(error);
-}
-
-
-/*
-  Optimize delete of all rows by doing a full generate of the table
-  This will work even if the .ISM and .ISD tables are destroyed
-
-  dont_send_ok should be set if:
-  - We should always wants to generate the table (even if the table type
-    normally can't safely do this.
-  - We don't want an ok to be sent to the end user.
-  - We don't want to log the truncate command
-  - If we want to keep exclusive metadata lock on the table (obtained by
-    caller) on exit without errors.
-*/
-
-bool mysql_truncate(THD *thd, TABLE_LIST *table_list, bool dont_send_ok)
-{
-  HA_CREATE_INFO create_info;
-  char path[FN_REFLEN + 1];
-  TABLE *table;
-  bool error= TRUE;
-  uint path_length;
-  /*
-    Is set if we're under LOCK TABLES, and used
-    to downgrade the exclusive lock after the
-    table was truncated.
-  */
-  MDL_ticket *mdl_ticket= NULL;
-  bool has_mdl_lock= FALSE;
-  bool is_temporary_table= false;
-  DBUG_ENTER("mysql_truncate");
-
-  bzero((char*) &create_info,sizeof(create_info));
-
-  /* Remove tables from the HANDLER's hash. */
-  mysql_ha_rm_tables(thd, table_list);
-
-  /* If it is a temporary table, close and regenerate it */
-  if (!dont_send_ok && (table= find_temporary_table(thd, table_list)))
-  {
-    is_temporary_table= true;
-    handlerton *table_type= table->s->db_type();
-    TABLE_SHARE *share= table->s;
-    /* Note that a temporary table cannot be partitioned */
-    if (!ha_check_storage_engine_flag(table_type, HTON_CAN_RECREATE))
-      goto trunc_by_del;
-
-    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, share->path.str,
-                                             share->db.str,
-					     share->table_name.str, 1))))
-      (void) rm_temporary_table(table_type, path);
-    else
-      thd->thread_specific_used= TRUE;
-    
-    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 my_ok() to the client.
-    */
-    goto end;
-  }
-
-  path_length= build_table_filename(path, sizeof(path) - 1, table_list->db,
-                                    table_list->table_name, reg_ext, 0);
-
-  if (!dont_send_ok)
-  {
-    enum legacy_db_type table_type;
-    /*
-      FIXME: Code of TRUNCATE breaks the meta-data
-      locking protocol since it tries to find out the table storage
-      engine and therefore accesses table in some way without holding
-      any kind of meta-data lock.
-    */
-    mysql_frm_type(thd, path, &table_type);
-    if (table_type == DB_TYPE_UNKNOWN)
-    {
-      my_error(ER_NO_SUCH_TABLE, MYF(0),
-               table_list->db, table_list->table_name);
-      DBUG_RETURN(TRUE);
-    }
-#ifdef WITH_PARTITION_STORAGE_ENGINE
-    /*
-      TODO: Add support for TRUNCATE PARTITION for NDB and other engines
-      supporting native partitioning
-    */
-    if (table_type != DB_TYPE_PARTITION_DB &&
-        thd->lex->alter_info.flags & ALTER_ADMIN_PARTITION)
-    {
-      my_error(ER_PARTITION_MGMT_ON_NONPARTITIONED, MYF(0));
-      DBUG_RETURN(TRUE);
-    }
-#endif
-    if (!ha_check_storage_engine_flag(ha_resolve_by_legacy_type(thd,
-                                                                table_type),
-                                      HTON_CAN_RECREATE) ||
-        thd->lex->alter_info.flags & ALTER_ADMIN_PARTITION)
-      goto trunc_by_del;
-
-
-    if (thd->locked_tables_mode)
-    {
-      if (!(table= find_table_for_mdl_upgrade(thd->open_tables, table_list->db,
-                                              table_list->table_name, FALSE)))
-        DBUG_RETURN(TRUE);
-      mdl_ticket= table->mdl_ticket;
-      if (wait_while_table_is_used(thd, table, HA_EXTRA_FORCE_REOPEN))
-        goto end;
-      close_all_tables_for_name(thd, table->s, FALSE);
-    }
-    else
-    {
-      MDL_request mdl_global_request, mdl_request;
-      MDL_request_list mdl_requests;
-      /*
-        Even though we could use the previous execution branch
-        here just as well, we must not try to open the table: 
-        MySQL manual documents that TRUNCATE can be used to 
-        repair a damaged table, i.e. a table that can not be
-        fully "opened". In particular MySQL manual says:
-
-        As long as the table format file tbl_name.frm  is valid,
-        the table can be re-created as an empty table with TRUNCATE
-        TABLE, even if the data or index files have become corrupted.
-      */
-
-      mdl_global_request.init(MDL_key::GLOBAL, "", "", MDL_INTENTION_EXCLUSIVE);
-      mdl_request.init(MDL_key::TABLE, table_list->db, table_list->table_name,
-                       MDL_EXCLUSIVE);
-      mdl_requests.push_front(&mdl_request);
-      mdl_requests.push_front(&mdl_global_request);
-
-      if (thd->mdl_context.acquire_locks(&mdl_requests,
-                                         thd->variables.lock_wait_timeout))
-        DBUG_RETURN(TRUE);
-
-      has_mdl_lock= TRUE;
-      mysql_mutex_lock(&LOCK_open);
-      tdc_remove_table(thd, TDC_RT_REMOVE_ALL, table_list->db,
-                       table_list->table_name);
-      mysql_mutex_unlock(&LOCK_open);
-    }
-  }
-
-  /*
-    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;
-  mysql_mutex_lock(&LOCK_open);
-  error= ha_create_table(thd, path, table_list->db, table_list->table_name,
-                         &create_info, 1);
-  mysql_mutex_unlock(&LOCK_open);
-  query_cache_invalidate3(thd, table_list, 0);
-
-end:
-  if (!dont_send_ok)
-  {
-    if (thd->locked_tables_mode && thd->locked_tables_list.reopen_tables(thd))
-      thd->locked_tables_list.unlink_all_closed_tables(thd, NULL, 0);
-    /*
-      Even if we failed to reopen some tables,
-      the operation itself succeeded, write the binlog.
-    */
-    if (!error)
-    {
-      /* In RBR, the statement is not binlogged if the table is temporary. */
-      if (!is_temporary_table || !thd->is_current_stmt_binlog_format_row())
-        error= write_bin_log(thd, TRUE, thd->query(), thd->query_length());
-      if (!error)
-        my_ok(thd);             // This should return record count
-    }
-    if (has_mdl_lock)
-      thd->mdl_context.release_transactional_locks();
-    if (mdl_ticket)
-      mdl_ticket->downgrade_exclusive_lock(MDL_SHARED_NO_READ_WRITE);
-  }
-
-  DBUG_PRINT("exit", ("error: %d", error));
-  DBUG_RETURN(error);
-
-trunc_by_del:
-  error= mysql_truncate_by_delete(thd, table_list);
-  DBUG_RETURN(error);
-}
diff --git a/sql/sql_delete.h b/sql/sql_delete.h
index d1c1b363abd..c718323ce1e 100644
--- a/sql/sql_delete.h
+++ b/sql/sql_delete.h
@@ -27,8 +27,6 @@ typedef struct st_sql_list SQL_LIST;
 
 int mysql_prepare_delete(THD *thd, TABLE_LIST *table_list, Item **conds);
 bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
-                  SQL_LIST *order, ha_rows rows, ulonglong options,
-                  bool reset_auto_increment);
-bool mysql_truncate(THD *thd, TABLE_LIST *table_list, bool dont_send_ok);
+                  SQL_LIST *order, ha_rows rows, ulonglong options);
 
 #endif /* SQL_DELETE_INCLUDED */
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 4c3f44ea75c..adab0b8e3d0 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -49,6 +49,7 @@
                               // mysql_recreate_table,
                               // mysql_backup_table,
                               // mysql_restore_table
+#include "sql_truncate.h"     // mysql_truncate_table
 #include "sql_connect.h"      // check_user,
                               // decrease_user_connections,
                               // thd_init_client_charset, check_mqh,
@@ -253,7 +254,7 @@ void init_update_queries(void)
   sql_command_flags[SQLCOM_ALTER_TABLE]=    CF_CHANGES_DATA | CF_WRITE_LOGS_COMMAND |
                                             CF_AUTO_COMMIT_TRANS | CF_PROTECT_AGAINST_GRL;
   sql_command_flags[SQLCOM_TRUNCATE]=       CF_CHANGES_DATA | CF_WRITE_LOGS_COMMAND |
-                                            CF_AUTO_COMMIT_TRANS;
+                                            CF_AUTO_COMMIT_TRANS | CF_PROTECT_AGAINST_GRL;
   sql_command_flags[SQLCOM_DROP_TABLE]=     CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_LOAD]=           CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
                                             CF_PROTECT_AGAINST_GRL;
@@ -3280,9 +3281,8 @@ end_with_restore_list:
                  ER(ER_LOCK_OR_ACTIVE_TRANSACTION), MYF(0));
       goto error;
     }
-    if (thd->global_read_lock.wait_if_global_read_lock(thd, FALSE, TRUE))
-      goto error;
-    res= mysql_truncate(thd, first_table, 0);
+    if (! (res= mysql_truncate_table(thd, first_table)))
+      my_ok(thd);
     break;
   case SQLCOM_DELETE:
   {
@@ -3295,8 +3295,7 @@ end_with_restore_list:
     MYSQL_DELETE_START(thd->query());
     res = mysql_delete(thd, all_tables, select_lex->where,
                        &select_lex->order_list,
-                       unit->select_limit_cnt, select_lex->options,
-                       FALSE);
+                       unit->select_limit_cnt, select_lex->options);
     MYSQL_DELETE_DONE(res, (ulong) thd->get_row_count_func());
     break;
   }
diff --git a/sql/sql_parse.h b/sql/sql_parse.h
index e1543a09549..ca804d5fab0 100644
--- a/sql/sql_parse.h
+++ b/sql/sql_parse.h
@@ -129,11 +129,6 @@ bool check_simple_select();
 Item *negate_expression(THD *thd, Item *expr);
 bool check_stack_overrun(THD *thd, long margin, uchar *dummy);
 
-bool begin_trans(THD *thd);
-bool end_active_trans(THD *thd);
-int end_trans(THD *thd, enum enum_mysql_completiontype completion);
-
-
 /* Variables */
 
 extern const char* any_db;
diff --git a/sql/sql_rename.cc b/sql/sql_rename.cc
index ea95b59b0c2..130a99a374f 100644
--- a/sql/sql_rename.cc
+++ b/sql/sql_rename.cc
@@ -29,6 +29,7 @@
                         // start_waiting_global_read_lock
 #include "sql_base.h"   // tdc_remove_table
 #include "sql_handler.h"                        // mysql_ha_rm_tables
+#include "datadict.h"
 
 static TABLE_LIST *rename_tables(THD *thd, TABLE_LIST *table_list,
 				 bool skip_error);
@@ -283,7 +284,7 @@ do_rename(THD *thd, TABLE_LIST *ren_table, char *new_db, char *new_table_name,
   build_table_filename(name, sizeof(name) - 1,
                        ren_table->db, old_alias, reg_ext, 0);
 
-  frm_type= mysql_frm_type(thd, name, &table_type);
+  frm_type= dd_frm_type(thd, name, &table_type);
   switch (frm_type)
   {
     case FRMTYPE_TABLE:
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index a33bc5943da..7fe84ce0dd4 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -27,7 +27,6 @@
                                               // primary_key_name,
                                               // build_table_filename
 #include "repl_failsafe.h"
-#include "sql_view.h"                           // mysql_frm_type
 #include "sql_parse.h"             // check_access, check_table_access
 #include "sql_partition.h"         // partition_element
 #include "sql_db.h"     // check_db_dir_existence, load_db_opt_by_name
@@ -50,6 +49,8 @@
 #endif
 #include <my_dir.h>
 #include "lock.h"                           // MYSQL_LOCK_IGNORE_FLUSH
+#include "debug_sync.h"
+#include "datadict.h"   // dd_frm_type()
 
 #define STR_OR_NIL(S) ((S) ? (S) : "<nil>")
 
@@ -2959,6 +2960,9 @@ fill_schema_show_cols_or_idxs(THD *thd, TABLE_LIST *tables,
                                        (can_deadlock ?
                                         MYSQL_OPEN_FAIL_ON_MDL_CONFLICT : 0)));
   lex->sql_command= save_sql_command;
+
+  DEBUG_SYNC(thd, "after_open_table_ignore_flush");
+
   /*
     get_all_tables() returns 1 on failure and 0 on success thus
     return only these and not the result code of ::process_table()
@@ -3018,7 +3022,7 @@ static int fill_schema_table_names(THD *thd, TABLE *table,
     char path[FN_REFLEN + 1];
     (void) build_table_filename(path, sizeof(path) - 1, db_name->str, 
                                 table_name->str, reg_ext, 0);
-    switch (mysql_frm_type(thd, path, &not_used)) {
+    switch (dd_frm_type(thd, path, &not_used)) {
     case FRMTYPE_ERROR:
       table->field[3]->store(STRING_WITH_LEN("ERROR"),
                              system_charset_info);
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index b2a950ca4b0..bad88476c09 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -27,8 +27,8 @@
                         // start_waiting_global_read_lock,
                         // unlock_table_names, mysql_unlock_tables
 #include "strfunc.h"    // find_type2, find_set
-#include "sql_view.h" // mysql_frm_type, view_checksum, mysql_frm_type
-#include "sql_delete.h"                         // mysql_truncate
+#include "sql_view.h" // view_checksum 
+#include "sql_truncate.h"                       // regenerate_locked_table 
 #include "sql_partition.h"                      // mem_alloc_error,
                                                 // generate_partition_syntax,
                                                 // partition_info
@@ -52,6 +52,7 @@
 #include "sql_show.h"
 #include "transaction.h"
 #include "keycaches.h"
+#include "datadict.h"  // dd_frm_type()
 
 #ifdef __WIN__
 #include <io.h>
@@ -2096,7 +2097,7 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
         ((access(path, F_OK) &&
           ha_create_table_from_engine(thd, db, alias)) ||
          (!drop_view &&
-          mysql_frm_type(thd, path, &frm_db_type) != FRMTYPE_TABLE)))
+          dd_frm_type(thd, path, &frm_db_type) != FRMTYPE_TABLE)))
     {
       // Table was not found on disk and table can't be created from engine
       if (if_exists)
@@ -2116,7 +2117,7 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
       */
       if (frm_db_type == DB_TYPE_UNKNOWN)
       {
-        mysql_frm_type(thd, path, &frm_db_type);
+        dd_frm_type(thd, path, &frm_db_type);
         DBUG_PRINT("info", ("frm_db_type %d from %s", frm_db_type, path));
       }
       table_type= ha_resolve_by_legacy_type(thd, frm_db_type);
@@ -4557,12 +4558,18 @@ static int prepare_for_repair(THD *thd, TABLE_LIST *table_list,
 			     "Failed renaming data file");
     goto end;
   }
-  if (mysql_truncate(thd, table_list, 1))
+  if (dd_recreate_table(thd, table_list->db, table_list->table_name))
   {
     error= send_check_errmsg(thd, table_list, "repair",
 			     "Failed generating table from .frm file");
     goto end;
   }
+  /*
+    'FALSE' for 'using_transactions' means don't postpone
+    invalidation till the end of a transaction, but do it
+    immediately.
+  */
+  query_cache_invalidate3(thd, table_list, FALSE);
   if (mysql_file_rename(key_file_misc, tmp, from, MYF(MY_WME)))
   {
     error= send_check_errmsg(thd, table_list, "repair",
@@ -6544,7 +6551,7 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
     into the main table list, like open_tables does).
     This code is wrong and will be removed, please do not copy.
   */
-  frm_type= mysql_frm_type(thd, new_name_buff, &table_type);
+  frm_type= dd_frm_type(thd, new_name_buff, &table_type);
   /* Rename a view */
   /* Sic: there is a race here */
   if (frm_type == FRMTYPE_VIEW && !(alter_info->flags & ~ALTER_RENAME))
diff --git a/sql/sql_truncate.cc b/sql/sql_truncate.cc
new file mode 100644
index 00000000000..47a98769d57
--- /dev/null
+++ b/sql/sql_truncate.cc
@@ -0,0 +1,464 @@
+/* Copyright (c) 2010, Oracle and/or its affiliates. All rights reserved.
+
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; version 2 of the License.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA */
+
+#include "sql_truncate.h"
+#include "sql_priv.h"
+#include "transaction.h"
+#include "debug_sync.h"
+#include "records.h"     // READ_RECORD
+#include "table.h"       // TABLE
+#include "sql_class.h"   // THD
+#include "sql_base.h"    // open_and_lock_tables
+#include "sql_table.h"   // write_bin_log
+#include "sql_handler.h" // mysql_ha_rm_tables
+#include "datadict.h"    // dd_recreate_table()
+#include "lock.h"        // MYSQL_OPEN_TEMPORARY_ONLY
+
+
+/*
+  Delete all rows of a locked table.
+
+  @param  thd           Thread context.
+  @param  table_list    Table list element for the table.
+  @param  rows_deleted  Whether rows might have been deleted.
+
+  @retval  FALSE  Success.
+  @retval  TRUE   Error.
+*/
+
+static bool
+delete_all_rows(THD *thd, TABLE *table)
+{
+  int error;
+  READ_RECORD info;
+  bool is_bulk_delete;
+  bool some_rows_deleted= FALSE;
+  bool save_binlog_row_based= thd->is_current_stmt_binlog_format_row();
+  DBUG_ENTER("delete_all_rows");
+
+  /* Replication of truncate table must be statement based. */
+  thd->clear_current_stmt_binlog_format_row();
+
+  /*
+    Attempt to delete all rows in the table.
+    If it is unsupported, switch to row by row deletion.
+  */
+  if (! (error= table->file->ha_delete_all_rows()))
+    goto end;
+
+  if (error != HA_ERR_WRONG_COMMAND)
+  {
+    /*
+      If a transactional engine fails in the middle of deletion,
+      we expect it to be able to roll it back.  Some reasons
+      for the engine to fail would be media failure or corrupted
+      data dictionary (i.e. in case of a partitioned table). We
+      have sufficiently strong metadata locks to rule out any
+      potential deadlocks.
+
+      If a non-transactional engine fails here (that would
+      not be MyISAM, since MyISAM does TRUNCATE by recreate),
+      and binlog is on, replication breaks, since nothing gets
+      written to the binary log.  (XXX: is this a bug?)
+    */
+    table->file->print_error(error, MYF(0));
+    goto end;
+  }
+
+  /*
+    A workaround for Bug#53696  "Performance schema engine violates the
+    PSEA API by calling my_error()".
+  */
+  if (thd->is_error())
+    goto end;
+
+  /* Handler didn't support fast delete. Delete rows one by one. */
+
+  init_read_record(&info, thd, table, NULL, TRUE, TRUE, FALSE);
+
+  /*
+    Start bulk delete. If the engine does not support it, go on,
+    it's not an error.
+  */
+  is_bulk_delete= ! table->file->start_bulk_delete();
+
+  table->mark_columns_needed_for_delete();
+
+  while (!(error= info.read_record(&info)) && !thd->killed)
+  {
+    if ((error= table->file->ha_delete_row(table->record[0])))
+    {
+      table->file->print_error(error, MYF(0));
+      break;
+    }
+
+    some_rows_deleted= TRUE;
+  }
+
+  /* HA_ERR_END_OF_FILE */
+  if (error == -1)
+    error= 0;
+
+  /* Close down the bulk delete. */
+  if (is_bulk_delete)
+  {
+    int bulk_delete_error= table->file->end_bulk_delete();
+    if (bulk_delete_error && !error)
+    {
+      table->file->print_error(bulk_delete_error, MYF(0));
+      error= bulk_delete_error;
+    }
+  }
+
+  end_read_record(&info);
+
+  /*
+    Regardless of the error status, the query must be written to the
+    binary log if rows of the table is non-transactional.
+  */
+  if (some_rows_deleted && !table->file->has_transactions())
+  {
+    thd->transaction.stmt.modified_non_trans_table= TRUE;
+    thd->transaction.all.modified_non_trans_table= TRUE;
+  }
+
+  if (error || thd->killed)
+    goto end;
+
+  /* Truncate resets the auto-increment counter. */
+  error= table->file->ha_reset_auto_increment(0);
+  if (error)
+  {
+    if (error != HA_ERR_WRONG_COMMAND)
+      table->file->print_error(error, MYF(0));
+    else
+      error= 0;
+  }
+
+end:
+  if (save_binlog_row_based)
+    thd->set_current_stmt_binlog_format_row();
+
+  DBUG_RETURN(error);
+}
+
+
+/*
+  Close and recreate a temporary table. In case of success,
+  write truncate statement into the binary log if in statement
+  mode.
+
+  @param  thd     Thread context.
+  @param  table   The temporary table.
+
+  @retval  FALSE  Success.
+  @retval  TRUE   Error.
+*/
+
+static bool recreate_temporary_table(THD *thd, TABLE *table)
+{
+  bool error= TRUE;
+  TABLE_SHARE *share= table->s;
+  HA_CREATE_INFO create_info;
+  handlerton *table_type= table->s->db_type();
+  DBUG_ENTER("recreate_temporary_table");
+
+  memset(&create_info, 0, sizeof(create_info));
+
+  table->file->info(HA_STATUS_AUTO | HA_STATUS_NO_LOCK);
+
+  /* Don't free share. */
+  close_temporary_table(thd, table, FALSE, FALSE);
+
+  /*
+    We must use share->normalized_path.str since for temporary tables it
+    differs from what dd_recreate_table() would generate based
+    on table and schema names.
+  */
+  ha_create_table(thd, share->normalized_path.str, share->db.str,
+                  share->table_name.str, &create_info, 1);
+
+  if (open_temporary_table(thd, share->path.str, share->db.str,
+                           share->table_name.str, 1))
+  {
+    error= FALSE;
+    thd->thread_specific_used= TRUE;
+  }
+  else
+    rm_temporary_table(table_type, share->path.str);
+
+  free_table_share(share);
+  my_free(table, MYF(0));
+
+  DBUG_RETURN(error);
+}
+
+
+/*
+  Handle opening and locking if a base table for truncate.
+
+  @param[in]  thd               Thread context.
+  @param[in]  table_ref         Table list element for the table to
+                                be truncated.
+  @param[out] hton_can_recreate Set to TRUE if table can be dropped
+                                and recreated.
+  @param[out] ticket_downgrade  Set if a lock must be downgraded after
+                                truncate is done.
+
+  @retval  FALSE  Success.
+  @retval  TRUE   Error.
+*/
+
+static bool open_and_lock_table_for_truncate(THD *thd, TABLE_LIST *table_ref,
+                                             bool *hton_can_recreate,
+                                             MDL_ticket **ticket_downgrade)
+{
+  TABLE *table= NULL;
+  MDL_ticket *mdl_ticket= NULL;
+  DBUG_ENTER("open_and_lock_table_for_truncate");
+
+  /*
+    Before doing anything else, acquire a metadata lock on the table,
+    or ensure we have one.  We don't use open_and_lock_tables()
+    right away because we want to be able to truncate (and recreate)
+    corrupted tables, those that we can't fully open.
+
+    MySQL manual documents that TRUNCATE can be used to repair a
+    damaged table, i.e. a table that can not be fully "opened".
+    In particular MySQL manual says: As long as the table format
+    file tbl_name.frm is valid, the table can be re-created as
+    an empty table with TRUNCATE TABLE, even if the data or index
+    files have become corrupted.
+  */
+  if (thd->locked_tables_mode)
+  {
+    if (!(table= find_table_for_mdl_upgrade(thd->open_tables, table_ref->db,
+                                            table_ref->table_name, FALSE)))
+      DBUG_RETURN(TRUE);
+
+    *hton_can_recreate= ha_check_storage_engine_flag(table->s->db_type(),
+                                                     HTON_CAN_RECREATE);
+  }
+  else
+  {
+    /*
+      Even though we could use the previous execution branch here just as
+      well, we must not try to open the table:
+    */
+    MDL_request mdl_global_request, mdl_request;
+    MDL_request_list mdl_requests;
+
+    mdl_global_request.init(MDL_key::GLOBAL, "", "", MDL_INTENTION_EXCLUSIVE);
+    mdl_request.init(MDL_key::TABLE, table_ref->db, table_ref->table_name,
+                     MDL_SHARED_NO_READ_WRITE);
+    mdl_requests.push_front(&mdl_request);
+    mdl_requests.push_front(&mdl_global_request);
+
+    if (thd->mdl_context.acquire_locks(&mdl_requests,
+                                        thd->variables.lock_wait_timeout))
+      DBUG_RETURN(TRUE);
+
+    mdl_ticket= mdl_request.ticket;
+
+    if (dd_check_storage_engine_flag(thd, table_ref->db, table_ref->table_name,
+                                     HTON_CAN_RECREATE, hton_can_recreate))
+      DBUG_RETURN(TRUE);
+  }
+
+  DEBUG_SYNC(thd, "lock_table_for_truncate");
+
+  if (*hton_can_recreate)
+  {
+    /*
+      Acquire an exclusive lock. The storage engine can recreate the
+      table only if there are no references to it from anywhere, i.e.
+      no cached TABLE in the table cache. To remove the table from the
+      cache we need an exclusive lock.
+    */
+    if (thd->locked_tables_mode)
+    {
+      if (wait_while_table_is_used(thd, table, HA_EXTRA_FORCE_REOPEN))
+        DBUG_RETURN(TRUE);
+      *ticket_downgrade= table->mdl_ticket;
+      close_all_tables_for_name(thd, table->s, FALSE);
+    }
+    else
+    {
+      ulong timeout= thd->variables.lock_wait_timeout;
+      if (thd->mdl_context.upgrade_shared_lock_to_exclusive(mdl_ticket, timeout))
+        DBUG_RETURN(TRUE);
+      mysql_mutex_lock(&LOCK_open);
+      tdc_remove_table(thd, TDC_RT_REMOVE_ALL, table_ref->db,
+                       table_ref->table_name);
+      mysql_mutex_unlock(&LOCK_open);
+    }
+  }
+  else
+  {
+    /*
+      Can't recreate, we must mechanically delete all rows in
+      the table. Our metadata lock guarantees that no transaction
+      is reading or writing into the table. Yet, to open a write
+      cursor we need a thr_lock lock. Use open_and_lock_tables()
+      to do the necessary job.
+    */
+
+    /* Allow to open base tables only. */
+    table_ref->required_type= FRMTYPE_TABLE;
+    /* We don't need to load triggers. */
+    DBUG_ASSERT(table_ref->trg_event_map == 0);
+    /* Work around partition parser rules using alter table's. */
+    if (thd->lex->alter_info.flags & ALTER_ADMIN_PARTITION)
+    {
+      table_ref->lock_type= TL_WRITE;
+      table_ref->mdl_request.set_type(MDL_SHARED_WRITE);
+    }
+    /* Ensure proper lock types (e.g. from the parser). */
+    DBUG_ASSERT(table_ref->lock_type == TL_WRITE);
+    DBUG_ASSERT(table_ref->mdl_request.type == MDL_SHARED_WRITE);
+
+    /*
+      Open the table as it will handle some required preparations.
+      Ignore pending FLUSH TABLES since we don't want to release
+      the MDL lock taken above and otherwise there is no way to
+      wait for FLUSH TABLES in deadlock-free fashion.
+    */
+    if (open_and_lock_tables(thd, table_ref, TL_WRITE,
+                             MYSQL_OPEN_IGNORE_FLUSH |
+                             MYSQL_OPEN_SKIP_TEMPORARY))
+      DBUG_RETURN(TRUE);
+  }
+
+  DBUG_RETURN(FALSE);
+}
+
+
+/*
+  Optimized delete of all rows by doing a full generate of the table.
+
+  @remark Will work even if the .MYI and .MYD files are destroyed.
+          In other words, it works as long as the .FRM is intact and
+          the engine supports re-create.
+
+  @param  thd         Thread context.
+  @param  table_ref   Table list element for the table to be truncated.
+
+  @retval  FALSE  Success.
+  @retval  TRUE   Error.
+*/
+
+bool mysql_truncate_table(THD *thd, TABLE_LIST *table_ref)
+{
+  TABLE *table;
+  bool error= TRUE, binlog_stmt;
+  MDL_ticket *mdl_ticket= NULL;
+  DBUG_ENTER("mysql_truncate_table");
+
+  /* Remove tables from the HANDLER's hash. */
+  mysql_ha_rm_tables(thd, table_ref);
+
+  /* If it is a temporary table, no need to take locks. */
+  if ((table= find_temporary_table(thd, table_ref)))
+  {
+    /* In RBR, the statement is not binlogged if the table is temporary. */
+    binlog_stmt= !thd->is_current_stmt_binlog_format_row();
+
+    /* Note that a temporary table cannot be partitioned. */
+    if (ha_check_storage_engine_flag(table->s->db_type(), HTON_CAN_RECREATE))
+    {
+      if ((error= recreate_temporary_table(thd, table)))
+        binlog_stmt= FALSE; /* No need to binlog failed truncate-by-recreate. */
+
+      DBUG_ASSERT(! thd->transaction.stmt.modified_non_trans_table);
+    }
+    else
+    {
+      /*
+        The engine does not support truncate-by-recreate. Open the
+        table and delete all rows. In such a manner this can in fact
+        open several tables if it's a temporary MyISAMMRG table.
+      */
+      if (open_and_lock_tables(thd, table_ref, TL_WRITE,
+                               MYSQL_OPEN_TEMPORARY_ONLY))
+        DBUG_RETURN(TRUE);
+
+      error= delete_all_rows(thd, table_ref->table);
+    }
+
+    /*
+      No need to invalidate the query cache, queries with temporary
+      tables are not in the cache. No need to write to the binary
+      log a failed row-by-row delete even if under RBR as the table
+      might not exist on the slave.
+    */
+  }
+  else /* It's not a temporary table. */
+  {
+    bool hton_can_recreate;
+
+    if (open_and_lock_table_for_truncate(thd, table_ref,
+                                         &hton_can_recreate, &mdl_ticket))
+      DBUG_RETURN(TRUE);
+
+    if (hton_can_recreate)
+    {
+     /*
+        The storage engine can truncate the table by creating an
+        empty table with the same structure.
+      */
+      error= dd_recreate_table(thd, table_ref->db, table_ref->table_name);
+
+      if (thd->locked_tables_mode && thd->locked_tables_list.reopen_tables(thd))
+          thd->locked_tables_list.unlink_all_closed_tables(thd, NULL, 0);
+
+      /* No need to binlog a failed truncate-by-recreate. */
+      binlog_stmt= !error;
+    }
+    else
+    {
+      error= delete_all_rows(thd, table_ref->table);
+
+      /*
+        Regardless of the error status, the query must be written to the
+        binary log if rows of a non-transactional table were deleted.
+      */
+      binlog_stmt= !error || thd->transaction.stmt.modified_non_trans_table;
+    }
+
+    query_cache_invalidate3(thd, table_ref, FALSE);
+  }
+
+  /* DDL is logged in statement format, regardless of binlog format. */
+  if (binlog_stmt)
+    error|= write_bin_log(thd, !error, thd->query(), thd->query_length());
+
+  /*
+    All effects of a TRUNCATE TABLE operation are rolled back if a row
+    by row deletion fails. Otherwise, it is automatically committed at
+    the end.
+  */
+  if (error)
+  {
+    trans_rollback_stmt(thd);
+    trans_rollback(thd);
+  }
+
+  if (mdl_ticket)
+    mdl_ticket->downgrade_exclusive_lock(MDL_SHARED_NO_READ_WRITE);
+
+  DBUG_PRINT("exit", ("error: %d", error));
+  DBUG_RETURN(test(error));
+}
+
diff --git a/sql/sql_truncate.h b/sql/sql_truncate.h
new file mode 100644
index 00000000000..a9eec384718
--- /dev/null
+++ b/sql/sql_truncate.h
@@ -0,0 +1,23 @@
+#ifndef SQL_TRUNCATE_INCLUDED
+#define SQL_TRUNCATE_INCLUDED
+/* Copyright (c) 2010, Oracle and/or its affiliates. All rights reserved.
+
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; version 2 of the License.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA */
+
+class THD;
+class TABLE_LIST;
+
+bool mysql_truncate_table(THD *thd, TABLE_LIST *table_ref);
+
+#endif
diff --git a/sql/sql_view.cc b/sql/sql_view.cc
index 4eee9502177..3c8de0a253c 100644
--- a/sql/sql_view.cc
+++ b/sql/sql_view.cc
@@ -32,6 +32,7 @@
 #include "sp.h"
 #include "sp_head.h"
 #include "sp_cache.h"
+#include "datadict.h"   // dd_frm_type()
 
 #define MD5_BUFF_LENGTH 33
 
@@ -1663,7 +1664,7 @@ bool mysql_drop_view(THD *thd, TABLE_LIST *views, enum_drop_mode drop_mode)
                          view->db, view->table_name, reg_ext, 0);
 
     if (access(path, F_OK) || 
-        FRMTYPE_VIEW != (type= mysql_frm_type(thd, path, &not_used)))
+        FRMTYPE_VIEW != (type= dd_frm_type(thd, path, &not_used)))
     {
       char name[FN_REFLEN];
       my_snprintf(name, sizeof(name), "%s.%s", view->db, view->table_name);
@@ -1742,54 +1743,6 @@ bool mysql_drop_view(THD *thd, TABLE_LIST *views, enum_drop_mode drop_mode)
 
 
 /*
-  Check type of .frm if we are not going to parse it
-
-  SYNOPSIS
-    mysql_frm_type()
-    path	path to file
-
-  RETURN
-    FRMTYPE_ERROR	error
-    FRMTYPE_TABLE	table
-    FRMTYPE_VIEW	view
-*/
-
-frm_type_enum mysql_frm_type(THD *thd, char *path, enum legacy_db_type *dbt)
-{
-  File file;
-  uchar header[10];	//"TYPE=VIEW\n" it is 10 characters
-  size_t error;
-  DBUG_ENTER("mysql_frm_type");
-
-  *dbt= DB_TYPE_UNKNOWN;
-
-  if ((file= mysql_file_open(key_file_frm,
-                             path, O_RDONLY | O_SHARE, MYF(0))) < 0)
-    DBUG_RETURN(FRMTYPE_ERROR);
-  error= mysql_file_read(file, (uchar*) header, sizeof(header), MYF(MY_NABP));
-  mysql_file_close(file, MYF(MY_WME));
-
-  if (error)
-    DBUG_RETURN(FRMTYPE_ERROR);
-  if (!strncmp((char*) header, "TYPE=VIEW\n", sizeof(header)))
-    DBUG_RETURN(FRMTYPE_VIEW);
-
-  /*
-    This is just a check for DB_TYPE. We'll return default unknown type
-    if the following test is true (arg #3). This should not have effect
-    on return value from this function (default FRMTYPE_TABLE)
-  */
-  if (header[0] != (uchar) 254 || header[1] != 1 ||
-      (header[2] != FRM_VER && header[2] != FRM_VER+1 &&
-       (header[2] < FRM_VER+3 || header[2] > FRM_VER+4)))
-    DBUG_RETURN(FRMTYPE_TABLE);
-
-  *dbt= (enum legacy_db_type) (uint) *(header + 3);
-  DBUG_RETURN(FRMTYPE_TABLE);                   // Is probably a .frm table
-}
-
-
-/*
   check of key (primary or unique) presence in updatable view
 
   SYNOPSIS
diff --git a/sql/sql_view.h b/sql/sql_view.h
index 7d06abb9068..c15ecffccb8 100644
--- a/sql/sql_view.h
+++ b/sql/sql_view.h
@@ -43,8 +43,6 @@ bool check_key_in_view(THD *thd, TABLE_LIST * view);
 
 bool insert_view_fields(THD *thd, List<Item> *list, TABLE_LIST *view);
 
-frm_type_enum mysql_frm_type(THD *thd, char *path, enum legacy_db_type *dbt);
-
 int view_checksum(THD *thd, TABLE_LIST *view);
 
 extern TYPELIB updatable_views_with_limit_typelib;
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index e5875663d4e..fdc8af942d1 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -10685,7 +10685,7 @@ opt_delete_option:
         ;
 
 truncate:
-          TRUNCATE_SYM opt_table_sym table_name
+          TRUNCATE_SYM opt_table_sym
           {
             LEX* lex= Lex;
             lex->sql_command= SQLCOM_TRUNCATE;
@@ -10693,7 +10693,11 @@ truncate:
             lex->select_lex.options= 0;
             lex->select_lex.sql_cache= SELECT_LEX::SQL_CACHE_UNSPECIFIED;
             lex->select_lex.init_order();
+            YYPS->m_lock_type= TL_WRITE;
+            YYPS->m_mdl_type= MDL_SHARED_WRITE;
           }
+          table_name
+          {}
         ;
 
 opt_table_sym:
diff --git a/sql/table.h b/sql/table.h
index ea585208b83..cb773052c8b 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -20,6 +20,7 @@
 #include "sql_plist.h"
 #include "sql_list.h"                           /* Sql_alloc */
 #include "mdl.h"
+#include "datadict.h"
 
 #ifndef MYSQL_CLIENT
 
@@ -305,14 +306,6 @@ enum tmp_table_type
   NO_TMP_TABLE, NON_TRANSACTIONAL_TMP_TABLE, TRANSACTIONAL_TMP_TABLE,
   INTERNAL_TMP_TABLE, SYSTEM_TMP_TABLE
 };
-
-enum frm_type_enum
-{
-  FRMTYPE_ERROR= 0,
-  FRMTYPE_TABLE,
-  FRMTYPE_VIEW
-};
-
 enum release_type { RELEASE_NORMAL, RELEASE_WAIT_FOR_DROP };
 
 typedef struct st_filesort_info