Bug#49938: Failing assertion: inode or deadlock in fsp/fsp0fsp.c

Bug#54678: InnoDB, TRUNCATE, ALTER, I_S SELECT, crash or deadlock

- Incompatible change: truncate no longer resorts to a row by
row delete if the storage engine does not support the truncate
method. Consequently, the count of affected rows does not, in
any case, reflect the actual number of rows.

- Incompatible change: it is no longer possible to truncate a
table that participates as a parent in a foreign key constraint,
unless it is a self-referencing constraint (both parent and child
are in the same table). To work around this incompatible change
and still be able to truncate such tables, disable foreign checks
with SET foreign_key_checks=0 before truncate. Alternatively, if
foreign key checks are necessary, please use a DELETE statement
without a WHERE condition.

Problem description:

The problem was that for storage engines that do not support
truncate table via a external drop and recreate, such as InnoDB
which implements truncate via a internal drop and recreate, the
delete_all_rows method could be invoked with a shared metadata
lock, causing problems if the engine needed exclusive access
to some internal metadata. This problem originated with the
fact that there is no truncate specific handler method, which
ended up leading to a abuse of the delete_all_rows method that
is primarily used for delete operations without a condition.

Solution:

The solution is to introduce a truncate handler method that is
invoked when the engine does not support truncation via a table
drop and recreate. This method is invoked under a exclusive
metadata lock, so that there is only a single instance of the
table when the method is invoked.

Also, the method is not invoked and a error is thrown if
the table is a parent in a non-self-referencing foreign key
relationship. This was necessary to avoid inconsistency as
some integrity checks are bypassed. This is inline with the
fact that truncate is primarily a DDL operation that was
designed to quickly remove all data from a table.

mysql-test/suite/innodb/t/innodb-truncate.test:
  Add test cases for truncate and foreign key checks.
  Also test that InnoDB resets auto-increment on truncate.
mysql-test/suite/innodb/t/innodb.test:
  FK is not necessary, test is related to auto-increment.
  
  Update error number, truncate is no longer invoked if
  table is parent in a FK relationship.
mysql-test/suite/innodb/t/innodb_mysql.test:
  Update error number, truncate is no longer invoked if
  table is parent in a FK relationship.
  
  Use delete instead of truncate, test is used to check
  the interaction of FKs, triggers and delete.
mysql-test/suite/parts/inc/partition_check.inc:
  Fix typo.
mysql-test/suite/sys_vars/t/foreign_key_checks_func.test:
  Update error number, truncate is no longer invoked if
  table is parent in a FK relationship.
mysql-test/t/mdl_sync.test:
  Modify test case to reflect and ensure that truncate takes
  a exclusive metadata lock.
mysql-test/t/trigger-trans.test:
  Update error number, truncate is no longer invoked if
  table is parent in a FK relationship.
sql/ha_partition.cc:
  Reorganize the various truncate methods. delete_all_rows is now
  passed directly to the underlying engines, so as truncate. The
  code responsible for truncating individual partitions is moved
  to ha_partition::truncate_partition, which is invoked when a
  ALTER TABLE t1 TRUNCATE PARTITION p statement is executed.
  
  Since the partition truncate no longer can be invoked via
  delete, the bitmap operations are not necessary anymore. The
  explicit reset of the auto-increment value is also removed
  as the underlying engines are now responsible for reseting
  the value.
sql/handler.cc:
  Wire up the handler truncate method.
sql/handler.h:
  Introduce and document the truncate handler method. It assumes
  certain use cases of delete_all_rows.
  
  Add method to retrieve the list of foreign keys referencing a
  table. Method is used to avoid truncating tables that are
  parent in a foreign key relationship.
sql/share/errmsg-utf8.txt:
  Add error message for truncate and FK.
sql/sql_lex.h:
  Introduce a flag so that the partition engine can detect when
  a partition is being truncated. Used to give a special error.
sql/sql_parse.cc:
  Function mysql_truncate_table no longer exists.
sql/sql_partition_admin.cc:
  Implement the TRUNCATE PARTITION statement.
sql/sql_truncate.cc:
  Change the truncate table implementation to use the new truncate
  handler method and to not rely on row-by-row delete anymore.
  
  The truncate handler method is always invoked with a exclusive
  metadata lock. Also, it is no longer possible to truncate a
  table that is parent in some non-self-referencing foreign key.
storage/archive/ha_archive.cc:
  Rename method as the description indicates that in the future
  this could be a truncate operation.
storage/blackhole/ha_blackhole.cc:
  Implement truncate as no operation for the blackhole engine in
  order to remain compatible with older releases.
storage/federated/ha_federated.cc:
  Introduce truncate method that invokes delete_all_rows.
  This is required to support partition truncate as this
  form of truncate does not implement the drop and recreate
  protocol.
storage/heap/ha_heap.cc:
  Introduce truncate method that invokes delete_all_rows.
  This is required to support partition truncate as this
  form of truncate does not implement the drop and recreate
  protocol.
storage/ibmdb2i/ha_ibmdb2i.cc:
  Introduce truncate method that invokes delete_all_rows.
  This is required to support partition truncate as this
  form of truncate does not implement the drop and recreate
  protocol.
storage/innobase/handler/ha_innodb.cc:
  Rename delete_all_rows to truncate. InnoDB now does truncate
  under a exclusive metadata lock.
  
  Introduce and reorganize methods used to retrieve the list
  of foreign keys referenced by a or referencing a table.
storage/myisammrg/ha_myisammrg.cc:
  Introduce truncate method that invokes delete_all_rows.
  This is required in order to remain compatible with earlier
  releases where truncate would resort to a row-by-row delete.

15 files changed, 528 insertions, 379 deletions

diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index 8bf35f79ba9..3d8d5ae9eb8 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -3337,113 +3337,123 @@ int ha_partition::delete_row(const uchar *buf)
     Called from sql_delete.cc by mysql_delete().
     Called from sql_select.cc by JOIN::reinit().
     Called from sql_union.cc by st_select_lex_unit::exec().
- 
-     Also used for handle ALTER TABLE t TRUNCATE PARTITION ...
-     NOTE: auto increment value will be truncated in that partition as well!
 */
 
 int ha_partition::delete_all_rows()
 {
   int error;
-  bool truncate= FALSE;
   handler **file;
-  THD *thd= ha_thd();
   DBUG_ENTER("ha_partition::delete_all_rows");
 
-  if (thd->lex->sql_command == SQLCOM_TRUNCATE)
+  file= m_file;
+  do
   {
-    Alter_info *alter_info= &thd->lex->alter_info;
-    /* TRUNCATE also means resetting auto_increment */
-    lock_auto_increment();
-    table_share->ha_part_data->next_auto_inc_val= 0;
-    table_share->ha_part_data->auto_inc_initialized= FALSE;
-    unlock_auto_increment();
-    if (alter_info->flags & ALTER_ADMIN_PARTITION)
-    {
-      /* ALTER TABLE t TRUNCATE PARTITION ... */
-      List_iterator<partition_element> part_it(m_part_info->partitions);
-      int saved_error= 0;
-      uint num_parts= m_part_info->num_parts;
-      uint num_subparts= m_part_info->num_subparts;
-      uint i= 0;
-      uint num_parts_set= alter_info->partition_names.elements;
-      uint num_parts_found= set_part_state(alter_info, m_part_info,
-                                           PART_ADMIN);
-      if (num_parts_set != num_parts_found &&
-          (!(alter_info->flags & ALTER_ALL_PARTITION)))
-        DBUG_RETURN(HA_ERR_NO_PARTITION_FOUND);
+    if ((error= (*file)->ha_delete_all_rows()))
+      DBUG_RETURN(error);
+  } while (*(++file));
+  DBUG_RETURN(0);
+}
+
+
+/**
+  Manually truncate the table.
+
+  @retval  0    Success.
+  @retval  > 0  Error code.
+*/
+
+int ha_partition::truncate()
+{
+  int error;
+  handler **file;
+  DBUG_ENTER("ha_partition::truncate");
+
+  /*
+    TRUNCATE also means resetting auto_increment. Hence, reset
+    it so that it will be initialized again at the next use.
+  */
+  lock_auto_increment();
+  table_share->ha_part_data->next_auto_inc_val= 0;
+  table_share->ha_part_data->auto_inc_initialized= FALSE;
+  unlock_auto_increment();
 
-      /*
-        Cannot return HA_ERR_WRONG_COMMAND here without correct pruning
-        since that whould delete the whole table row by row in sql_delete.cc
-      */
-      bitmap_clear_all(&m_part_info->used_partitions);
-      do
-      {
-        partition_element *part_elem= part_it++;
-        if (part_elem->part_state == PART_ADMIN)
-        {
-          if (m_is_sub_partitioned)
-          {
-            List_iterator<partition_element>
-                                        subpart_it(part_elem->subpartitions);
-            partition_element *sub_elem;
-            uint j= 0, part;
-            do
-            {
-              sub_elem= subpart_it++;
-              part= i * num_subparts + j;
-              bitmap_set_bit(&m_part_info->used_partitions, part);
-              if (!saved_error)
-              {
-                DBUG_PRINT("info", ("truncate subpartition %u (%s)",
-                                    part, sub_elem->partition_name));
-                if ((error= m_file[part]->ha_delete_all_rows()))
-                  saved_error= error;
-                /* If not reset_auto_increment is supported, just accept it */
-                if (!saved_error &&
-                    (error= m_file[part]->ha_reset_auto_increment(0)) &&
-                    error != HA_ERR_WRONG_COMMAND)
-                  saved_error= error;
-              }
-            } while (++j < num_subparts);
-          }
-          else
-          {
-            DBUG_PRINT("info", ("truncate partition %u (%s)", i,
-                                part_elem->partition_name));
-            bitmap_set_bit(&m_part_info->used_partitions, i);
-            if (!saved_error)
-            {
-              if ((error= m_file[i]->ha_delete_all_rows()) && !saved_error)
-                saved_error= error;
-              /* If not reset_auto_increment is supported, just accept it */
-              if (!saved_error &&
-                  (error= m_file[i]->ha_reset_auto_increment(0)) &&
-                  error != HA_ERR_WRONG_COMMAND)
-                saved_error= error;
-            }
-          }
-          part_elem->part_state= PART_NORMAL;
-        }
-      } while (++i < num_parts);
-      DBUG_RETURN(saved_error);
-    }
-    truncate= TRUE;
-  }
   file= m_file;
   do
   {
-    if ((error= (*file)->ha_delete_all_rows()))
+    if ((error= (*file)->ha_truncate()))
       DBUG_RETURN(error);
-    /* Ignore the error */
-    if (truncate)
-      (void) (*file)->ha_reset_auto_increment(0);
   } while (*(++file));
   DBUG_RETURN(0);
 }
 
 
+/**
+  Truncate a set of specific partitions.
+
+  @remark Auto increment value will be truncated in that partition as well!
+
+  ALTER TABLE t TRUNCATE PARTITION ...
+*/
+
+int ha_partition::truncate_partition(Alter_info *alter_info)
+{
+  int error= 0;
+  List_iterator<partition_element> part_it(m_part_info->partitions);
+  uint num_parts= m_part_info->num_parts;
+  uint num_subparts= m_part_info->num_subparts;
+  uint i= 0;
+  uint num_parts_set= alter_info->partition_names.elements;
+  uint num_parts_found= set_part_state(alter_info, m_part_info,
+                                        PART_ADMIN);
+  DBUG_ENTER("ha_partition::truncate_partition");
+
+  /*
+    TRUNCATE also means resetting auto_increment. Hence, reset
+    it so that it will be initialized again at the next use.
+  */
+  lock_auto_increment();
+  table_share->ha_part_data->next_auto_inc_val= 0;
+  table_share->ha_part_data->auto_inc_initialized= FALSE;
+  unlock_auto_increment();
+
+  if (num_parts_set != num_parts_found &&
+      (!(alter_info->flags & ALTER_ALL_PARTITION)))
+    DBUG_RETURN(HA_ERR_NO_PARTITION_FOUND);
+
+  do
+  {
+    partition_element *part_elem= part_it++;
+    if (part_elem->part_state == PART_ADMIN)
+    {
+      if (m_is_sub_partitioned)
+      {
+        List_iterator<partition_element>
+                                    subpart_it(part_elem->subpartitions);
+        partition_element *sub_elem;
+        uint j= 0, part;
+        do
+        {
+          sub_elem= subpart_it++;
+          part= i * num_subparts + j;
+          DBUG_PRINT("info", ("truncate subpartition %u (%s)",
+                              part, sub_elem->partition_name));
+          if ((error= m_file[part]->ha_truncate()))
+            break;
+        } while (++j < num_subparts);
+      }
+      else
+      {
+        DBUG_PRINT("info", ("truncate partition %u (%s)", i,
+                            part_elem->partition_name));
+        error= m_file[i]->ha_truncate();
+      }
+      part_elem->part_state= PART_NORMAL;
+    }
+  } while (!error && (++i < num_parts));
+  DBUG_RETURN(error);
+}
+
+
 /*
   Start a large batch of insert rows
 
@@ -6327,8 +6337,8 @@ void ha_partition::print_error(int error, myf errflag)
   /* Should probably look for my own errors first */
   DBUG_PRINT("enter", ("error: %d", error));
 
-  if (error == HA_ERR_NO_PARTITION_FOUND &&
-      thd->lex->sql_command != SQLCOM_TRUNCATE)
+  if ((error == HA_ERR_NO_PARTITION_FOUND) &&
+      ! (thd->lex->alter_info.flags & ALTER_TRUNCATE_PARTITION))
     m_part_info->print_no_partition_found(table);
   else
   {
diff --git a/sql/ha_partition.h b/sql/ha_partition.h
index 353e0d17159..f1abc0cefe2 100644
--- a/sql/ha_partition.h
+++ b/sql/ha_partition.h
@@ -346,6 +346,7 @@ public:
   virtual int update_row(const uchar * old_data, uchar * new_data);
   virtual int delete_row(const uchar * buf);
   virtual int delete_all_rows(void);
+  virtual int truncate();
   virtual void start_bulk_insert(ha_rows rows);
   virtual int end_bulk_insert();
 private:
@@ -354,6 +355,15 @@ private:
   long estimate_read_buffer_size(long original_size);
 public:
 
+  /*
+    Method for truncating a specific partition.
+    (i.e. ALTER TABLE t1 TRUNCATE PARTITION p).
+
+    @remark This method is a partitioning-specific hook
+            and thus not a member of the general SE API.
+  */
+  int truncate_partition(Alter_info *);
+
   virtual bool is_fatal_error(int error, uint flags)
   {
     if (!handler::is_fatal_error(error, flags) ||
diff --git a/sql/handler.cc b/sql/handler.cc
index 567dbe6ea49..1542dd99ec6 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -3209,6 +3209,21 @@ handler::ha_delete_all_rows()
 
 
 /**
+  Truncate table: public interface.
+
+  @sa handler::truncate()
+*/
+
+int
+handler::ha_truncate()
+{
+  mark_trx_read_write();
+
+  return truncate();
+}
+
+
+/**
   Reset auto increment: public interface.
 
   @sa handler::reset_auto_increment()
diff --git a/sql/handler.h b/sql/handler.h
index b1d64a1114b..325df003215 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -1331,6 +1331,7 @@ public:
   int ha_bulk_update_row(const uchar *old_data, uchar *new_data,
                          uint *dup_key_found);
   int ha_delete_all_rows();
+  int ha_truncate();
   int ha_reset_auto_increment(ulonglong value);
   int ha_optimize(THD* thd, HA_CHECK_OPT* check_opt);
   int ha_analyze(THD* thd, HA_CHECK_OPT* check_opt);
@@ -1644,8 +1645,33 @@ public:
   { return(NULL);}  /* gets tablespace name from handler */
   /** used in ALTER TABLE; 1 if changing storage engine is allowed */
   virtual bool can_switch_engines() { return 1; }
-  /** used in REPLACE; is > 0 if table is referred by a FOREIGN KEY */
-  virtual int get_foreign_key_list(THD *thd, List<FOREIGN_KEY_INFO> *f_key_list)
+  /**
+    Get the list of foreign keys in this table.
+
+    @remark Returns the set of foreign keys where this table is the
+            dependent or child table.
+
+    @param thd  The thread handle.
+    @param f_key_list[out]  The list of foreign keys.
+
+    @return The handler error code or zero for success.
+  */
+  virtual int
+  get_foreign_key_list(THD *thd, List<FOREIGN_KEY_INFO> *f_key_list)
+  { return 0; }
+  /**
+    Get the list of foreign keys referencing this table.
+
+    @remark Returns the set of foreign keys where this table is the
+            referenced or parent table.
+
+    @param thd  The thread handle.
+    @param f_key_list[out]  The list of foreign keys.
+
+    @return The handler error code or zero for success.
+  */
+  virtual int
+  get_parent_foreign_key_list(THD *thd, List<FOREIGN_KEY_INFO> *f_key_list)
   { return 0; }
   virtual uint referenced_by_foreign_key() { return 0;}
   virtual void init_table_handle_for_HANDLER()
@@ -2010,16 +2036,34 @@ private:
     This is called to delete all rows in a table
     If the handler don't support this, then this function will
     return HA_ERR_WRONG_COMMAND and MySQL will delete the rows one
-    by one. It should reset auto_increment if
-    thd->lex->sql_command == SQLCOM_TRUNCATE.
+    by one.
   */
   virtual int delete_all_rows()
   { return (my_errno=HA_ERR_WRONG_COMMAND); }
   /**
+    Quickly remove all rows from a table.
+
+    @remark This method is responsible for implementing MySQL's TRUNCATE
+            TABLE statement, which is a DDL operation. As such, a engine
+            can bypass certain integrity checks and in some cases avoid
+            fine-grained locking (e.g. row locks) which would normally be
+            required for a DELETE statement.
+
+    @remark Typically, truncate is not used if it can result in integrity
+            violation. For example, truncate is not used when a foreign
+            key references the table, but it might be used if foreign key
+            checks are disabled.
+
+    @remark Engine is responsible for resetting the auto-increment counter.
+
+    @remark The table is locked in exclusive mode.
+  */
+  virtual int truncate()
+  { return HA_ERR_WRONG_COMMAND; }
+  /**
     Reset the auto-increment counter to the given value, i.e. the next row
-    inserted will get the given value. This is called e.g. after TRUNCATE
-    is emulated by doing a 'DELETE FROM t'. HA_ERR_WRONG_COMMAND is
-    returned by storage engines that don't support this operation.
+    inserted will get the given value. HA_ERR_WRONG_COMMAND is returned by
+    storage engines that don't support this operation.
   */
   virtual int reset_auto_increment(ulonglong value)
   { return HA_ERR_WRONG_COMMAND; }
diff --git a/sql/share/errmsg-utf8.txt b/sql/share/errmsg-utf8.txt
index 07f5589135d..c7d016b19ca 100644
--- a/sql/share/errmsg-utf8.txt
+++ b/sql/share/errmsg-utf8.txt
@@ -6379,3 +6379,6 @@ ER_SET_PASSWORD_AUTH_PLUGIN
 
 ER_GRANT_PLUGIN_USER_EXISTS
         eng "GRANT with IDENTIFIED WITH is illegal because the user %-.*s already exists"
+
+ER_TRUNCATE_ILLEGAL_FK 42000
+  eng "Cannot truncate a table referenced in a foreign key constraint (%.192s)"
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index 3a559433e2d..0f93275434d 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -958,6 +958,7 @@ inline bool st_select_lex_unit::is_union ()
 #define ALTER_ALL_PARTITION      (1L << 21)
 #define ALTER_REMOVE_PARTITIONING (1L << 22)
 #define ALTER_FOREIGN_KEY        (1L << 23)
+#define ALTER_TRUNCATE_PARTITION (1L << 24)
 
 enum enum_alter_table_change_level
 {
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index e44d2563483..2f8a72ee25c 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -34,7 +34,7 @@
 #include "sql_locale.h"                         // my_locale_en_US
 #include "log.h"                                // flush_error_log
 #include "sql_view.h"         // mysql_create_view, mysql_drop_view
-#include "sql_delete.h"       // mysql_truncate, mysql_delete
+#include "sql_delete.h"       // mysql_delete
 #include "sql_insert.h"       // mysql_insert
 #include "sql_update.h"       // mysql_update, mysql_multi_update
 #include "sql_partition.h"    // struct partition_info
@@ -49,7 +49,6 @@
                               // mysql_recreate_table,
                               // mysql_backup_table,
                               // mysql_restore_table
-#include "sql_truncate.h"     // mysql_truncate_table
 #include "sql_reload.h"       // reload_acl_and_cache
 #include "sql_admin.h"        // mysql_assign_to_keycache
 #include "sql_connect.h"      // check_user,
diff --git a/sql/sql_partition_admin.cc b/sql/sql_partition_admin.cc
index fee33303a04..98750314a4a 100644
--- a/sql/sql_partition_admin.cc
+++ b/sql/sql_partition_admin.cc
@@ -16,10 +16,10 @@
 #include "sql_parse.h"                      // check_one_table_access
 #include "sql_table.h"                      // mysql_alter_table, etc.
 #include "sql_lex.h"                        // Sql_statement
-#include "sql_truncate.h"                   // mysql_truncate_table,
-                                            // Truncate_statement
 #include "sql_admin.h"                      // Analyze/Check/.._table_statement
 #include "sql_partition_admin.h"            // Alter_table_*_partition
+#include "ha_partition.h"                   // ha_partition
+#include "sql_base.h"                       // open_and_lock_tables
 
 #ifndef WITH_PARTITION_STORAGE_ENGINE
 
@@ -46,7 +46,7 @@ bool Alter_table_analyze_partition_statement::execute(THD *thd)
   m_lex->alter_info.flags|= ALTER_ADMIN_PARTITION;
 
   res= Analyze_table_statement::execute(thd);
-    
+
   DBUG_RETURN(res);
 }
 
@@ -104,36 +104,85 @@ bool Alter_table_repair_partition_statement::execute(THD *thd)
 
 bool Alter_table_truncate_partition_statement::execute(THD *thd)
 {
+  int error;
+  ha_partition *partition;
+  ulong timeout= thd->variables.lock_wait_timeout;
   TABLE_LIST *first_table= thd->lex->select_lex.table_list.first;
-  bool res;
-  enum_sql_command original_sql_command;
   DBUG_ENTER("Alter_table_truncate_partition_statement::execute");
 
   /*
-    Execute TRUNCATE PARTITION just like TRUNCATE TABLE.
-    Some storage engines (InnoDB, partition) checks thd_sql_command,
-    so we set it to SQLCOM_TRUNCATE during the execution.
-  */
-  original_sql_command= m_lex->sql_command;
-  m_lex->sql_command= SQLCOM_TRUNCATE;
-  
-  /*
     Flag that it is an ALTER command which administrates partitions, used
     by ha_partition.
   */
-  m_lex->alter_info.flags|= ALTER_ADMIN_PARTITION;
-   
+  m_lex->alter_info.flags|= ALTER_ADMIN_PARTITION |
+                            ALTER_TRUNCATE_PARTITION;
+
+  /* Fix the lock types (not the same as ordinary ALTER TABLE). */
+  first_table->lock_type= TL_WRITE;
+  first_table->mdl_request.set_type(MDL_EXCLUSIVE);
+
   /*
-    Fix the lock types (not the same as ordinary ALTER TABLE).
+    Check table permissions and open it with a exclusive lock.
+    Ensure it is a partitioned table and finally, upcast the
+    handler and invoke the partition truncate method. Lastly,
+    write the statement to the binary log if necessary.
   */
-  first_table->lock_type= TL_WRITE;
-  first_table->mdl_request.set_type(MDL_SHARED_NO_READ_WRITE);
 
-  /* execute as a TRUNCATE TABLE */
-  res= Truncate_statement::execute(thd);
+  if (check_one_table_access(thd, DROP_ACL, first_table))
+    DBUG_RETURN(TRUE);
 
-  m_lex->sql_command= original_sql_command;
-  DBUG_RETURN(res);
+  if (open_and_lock_tables(thd, first_table, FALSE, 0))
+    DBUG_RETURN(TRUE);
+
+  /*
+    TODO: Add support for TRUNCATE PARTITION for NDB and other
+          engines supporting native partitioning.
+  */
+  if (first_table->table->s->db_type() != partition_hton)
+  {
+    my_error(ER_PARTITION_MGMT_ON_NONPARTITIONED, MYF(0));
+    DBUG_RETURN(TRUE);
+  }
+
+  /*
+    Under locked table modes this might still not be an exclusive
+    lock. Hence, upgrade the lock since the handler truncate method
+    mandates an exclusive metadata lock.
+  */
+  MDL_ticket *ticket= first_table->table->mdl_ticket;
+  if (thd->mdl_context.upgrade_shared_lock_to_exclusive(ticket, timeout))
+    DBUG_RETURN(TRUE);
+
+  tdc_remove_table(thd, TDC_RT_REMOVE_NOT_OWN, first_table->db,
+                   first_table->table_name, FALSE);
+
+  partition= (ha_partition *) first_table->table->file;
+
+  /* Invoke the handler method responsible for truncating the partition. */
+  if ((error= partition->truncate_partition(&thd->lex->alter_info)))
+    first_table->table->file->print_error(error, MYF(0));
+
+  /*
+    All effects of a truncate operation are committed even if the
+    operation fails. Thus, the query must be written to the binary
+    log. The only exception is a unimplemented truncate method. Also,
+    it is logged in statement format, regardless of the binlog format.
+  */
+  if (error != HA_ERR_WRONG_COMMAND)
+    error|= write_bin_log(thd, !error, thd->query(), thd->query_length());
+
+  /*
+    A locked table ticket was upgraded to a exclusive lock. After the
+    the query has been written to the binary log, downgrade the lock
+    to a shared one.
+  */
+  if (thd->locked_tables_mode)
+    ticket->downgrade_exclusive_lock(MDL_SHARED_NO_READ_WRITE);
+
+  if (! error)
+    my_ok(thd);
+
+  DBUG_RETURN(error);
 }
 
 #endif /* WITH_PARTITION_STORAGE_ENGINE */
diff --git a/sql/sql_partition_admin.h b/sql/sql_partition_admin.h
index 36bafec4202..564b8676be8 100644
--- a/sql/sql_partition_admin.h
+++ b/sql/sql_partition_admin.h
@@ -210,7 +210,7 @@ public:
 /**
   Class that represents the ALTER TABLE t1 TRUNCATE PARTITION p statement.
 */
-class Alter_table_truncate_partition_statement : public Truncate_statement
+class Alter_table_truncate_partition_statement : public Sql_statement
 {
 public:
   /**
@@ -218,10 +218,10 @@ public:
     @param lex the LEX structure for this statement.
   */
   Alter_table_truncate_partition_statement(LEX *lex)
-    : Truncate_statement(lex)
+    : Sql_statement(lex)
   {}
 
-  ~Alter_table_truncate_partition_statement()
+  virtual ~Alter_table_truncate_partition_statement()
   {}
 
   /**
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 6b24e3db7bc..ba4b0ae7a0a 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -5063,8 +5063,8 @@ static int get_schema_constraints_record(THD *thd, TABLE_LIST *tables,
     while ((f_key_info=it++))
     {
       if (store_constraints(thd, table, db_name, table_name, 
-                            f_key_info->forein_id->str,
-                            strlen(f_key_info->forein_id->str),
+                            f_key_info->foreign_id->str,
+                            strlen(f_key_info->foreign_id->str),
                             "FOREIGN KEY", 11))
         DBUG_RETURN(1);
     }
@@ -5263,8 +5263,8 @@ static int get_schema_key_column_usage_record(THD *thd,
         f_idx++;
         restore_record(table, s->default_values);
         store_key_column_usage(table, db_name, table_name,
-                               f_key_info->forein_id->str,
-                               f_key_info->forein_id->length,
+                               f_key_info->foreign_id->str,
+                               f_key_info->foreign_id->length,
                                f_info->str, f_info->length,
                                (longlong) f_idx);
         table->field[8]->store((longlong) f_idx, TRUE);
@@ -6053,8 +6053,8 @@ get_referential_constraints_record(THD *thd, TABLE_LIST *tables,
       table->field[0]->store(STRING_WITH_LEN("def"), cs);
       table->field[1]->store(db_name->str, db_name->length, cs);
       table->field[9]->store(table_name->str, table_name->length, cs);
-      table->field[2]->store(f_key_info->forein_id->str,
-                             f_key_info->forein_id->length, cs);
+      table->field[2]->store(f_key_info->foreign_id->str,
+                             f_key_info->foreign_id->length, cs);
       table->field[3]->store(STRING_WITH_LEN("def"), cs);
       table->field[4]->store(f_key_info->referenced_db->str, 
                              f_key_info->referenced_db->length, cs);
diff --git a/sql/sql_string.h b/sql/sql_string.h
index d21b5353b76..845b7c280b1 100644
--- a/sql/sql_string.h
+++ b/sql/sql_string.h
@@ -265,8 +265,12 @@ public:
 	    CHARSET_INFO *csto, uint *errors);
   bool append(const String &s);
   bool append(const char *s);
-  bool append(const char *s,uint32 arg_length);
-  bool append(const char *s,uint32 arg_length, CHARSET_INFO *cs);
+  bool append(LEX_STRING *ls)
+  {
+    return append(ls->str, ls->length);
+  }
+  bool append(const char *s, uint32 arg_length);
+  bool append(const char *s, uint32 arg_length, CHARSET_INFO *cs);
   bool append_ulonglong(ulonglong val);
   bool append(IO_CACHE* file, uint32 arg_length);
   bool append_with_prefill(const char *s, uint32 arg_length, 
diff --git a/sql/sql_truncate.cc b/sql/sql_truncate.cc
index c0bc726a188..0cff2875ac8 100644
--- a/sql/sql_truncate.cc
+++ b/sql/sql_truncate.cc
@@ -13,11 +13,8 @@
    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_priv.h"
-#include "transaction.h"
-#include "debug_sync.h"
-#include "records.h"     // READ_RECORD
-#include "table.h"       // TABLE
+#include "debug_sync.h"  // DEBUG_SYNC
+#include "table.h"       // TABLE, FOREIGN_KEY_INFO
 #include "sql_class.h"   // THD
 #include "sql_base.h"    // open_and_lock_tables
 #include "sql_table.h"   // write_bin_log
@@ -29,145 +26,212 @@
 #include "sql_truncate.h"
 
 
-/*
-  Delete all rows of a locked table.
+/**
+  Append a list of field names to a string.
 
-  @param  thd           Thread context.
-  @param  table_list    Table list element for the table.
-  @param  rows_deleted  Whether rows might have been deleted.
+  @param  str     The string.
+  @param  fields  The list of field names.
 
-  @retval  FALSE  Success.
-  @retval  TRUE   Error.
+  @return TRUE on failure, FALSE otherwise.
 */
 
-static bool
-delete_all_rows(THD *thd, TABLE *table)
+static bool fk_info_append_fields(String *str, List<LEX_STRING> *fields)
 {
-  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();
+  bool res= FALSE;
+  LEX_STRING *field;
+  List_iterator_fast<LEX_STRING> it(*fields);
 
-  /*
-    Update handler statistics (e.g. table->file->stats.records).
-    Might be used by the storage engine to aggregate information
-    necessary to allow deletion. Currently, this seems to be
-    meaningful only to the archive storage engine, which uses
-    the info method to set the number of records. Although
-    archive does not support deletion, it becomes necessary in
-    order to return a error if the table is not empty.
-  */
-  error= table->file->info(HA_STATUS_VARIABLE | HA_STATUS_NO_LOCK);
-  if (error && error != HA_ERR_WRONG_COMMAND)
+  while ((field= it++))
   {
-    table->file->print_error(error, MYF(0));
-    goto end;
+    res|= str->append("`");
+    res|= str->append(field);
+    res|= str->append("`, ");
   }
 
-  /*
-    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;
+  str->chop();
+  str->chop();
 
-  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;
-  }
+  return res;
+}
+
+
+/**
+  Generate a foreign key description suitable for a error message.
+
+  @param thd          Thread context.
+  @param fk_info   The foreign key information.
+
+  @return A human-readable string describing the foreign key.
+*/
+
+static const char *fk_info_str(THD *thd, FOREIGN_KEY_INFO *fk_info)
+{
+  bool res= FALSE;
+  char buffer[STRING_BUFFER_USUAL_SIZE*2];
+  String str(buffer, sizeof(buffer), system_charset_info);
+
+  str.length(0);
 
   /*
-    A workaround for Bug#53696  "Performance schema engine violates the
-    PSEA API by calling my_error()".
+    `db`.`tbl`, CONSTRAINT `id` FOREIGN KEY (`fk`) REFERENCES `db`.`tbl` (`fk`)
   */
-  if (thd->is_error())
-    goto end;
 
-  /* Handler didn't support fast delete. Delete rows one by one. */
+  res|= str.append('`');
+  res|= str.append(fk_info->foreign_db);
+  res|= str.append("`.`");
+  res|= str.append(fk_info->foreign_table);
+  res|= str.append("`, CONSTRAINT `");
+  res|= str.append(fk_info->foreign_id);
+  res|= str.append("` FOREIGN KEY (");
+  res|= fk_info_append_fields(&str, &fk_info->foreign_fields);
+  res|= str.append(") REFERENCES `");
+  res|= str.append(fk_info->referenced_db);
+  res|= str.append("`.`");
+  res|= str.append(fk_info->referenced_table);
+  res|= str.append("` (");
+  res|= fk_info_append_fields(&str, &fk_info->referenced_fields);
+  res|= str.append(')');
+
+  return res ? NULL : thd->strmake(str.ptr(), str.length());
+}
+
+
+/**
+  Check and emit a fatal error if the table which is going to be
+  affected by TRUNCATE TABLE is a parent table in some non-self-
+  referencing foreign key.
+
+  @remark The intention is to allow truncate only for tables that
+          are not dependent on other tables.
+
+  @param  thd    Thread context.
+  @param  table  Table handle.
+
+  @retval FALSE  This table is not parent in a non-self-referencing foreign
+                 key. Statement can proceed.
+  @retval TRUE   This table is parent in a non-self-referencing foreign key,
+                 error was emitted.
+*/
+
+static bool
+fk_truncate_illegal_if_parent(THD *thd, TABLE *table)
+{
+  FOREIGN_KEY_INFO *fk_info;
+  List<FOREIGN_KEY_INFO> fk_list;
+  List_iterator_fast<FOREIGN_KEY_INFO> it;
 
-  init_read_record(&info, thd, table, NULL, TRUE, TRUE, FALSE);
+  /*
+    Bail out early if the table is not referenced by a foreign key.
+    In this case, the table could only be, if at all, a child table.
+  */
+  if (! table->file->referenced_by_foreign_key())
+    return FALSE;
 
   /*
-    Start bulk delete. If the engine does not support it, go on,
-    it's not an error.
+    This table _is_ referenced by a foreign key. At this point, only
+    self-referencing keys are acceptable. For this reason, get the list
+    of foreign keys referencing this table in order to check the name
+    of the child (dependent) tables.
   */
-  is_bulk_delete= ! table->file->start_bulk_delete();
+  table->file->get_parent_foreign_key_list(thd, &fk_list);
 
-  table->mark_columns_needed_for_delete();
+  /* Out of memory when building list. */
+  if (thd->is_error())
+    return TRUE;
 
-  while (!(error= info.read_record(&info)) && !thd->killed)
+  it.init(fk_list);
+
+  /* Loop over the set of foreign keys for which this table is a parent. */
+  while ((fk_info= it++))
   {
-    if ((error= table->file->ha_delete_row(table->record[0])))
-    {
-      table->file->print_error(error, MYF(0));
+    DBUG_ASSERT(!my_strcasecmp(system_charset_info,
+                               fk_info->referenced_db->str,
+                               table->s->db.str));
+
+    DBUG_ASSERT(!my_strcasecmp(system_charset_info,
+                               fk_info->referenced_table->str,
+                               table->s->table_name.str));
+
+    if (my_strcasecmp(system_charset_info, fk_info->foreign_db->str,
+                      table->s->db.str) ||
+        my_strcasecmp(system_charset_info, fk_info->foreign_table->str,
+                      table->s->table_name.str))
       break;
-    }
-
-    some_rows_deleted= TRUE;
   }
 
-  /* HA_ERR_END_OF_FILE */
-  if (error == -1)
-    error= 0;
-
-  /* Close down the bulk delete. */
-  if (is_bulk_delete)
+  /* Table is parent in a non-self-referencing foreign key. */
+  if (fk_info)
   {
-    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;
-    }
+    my_error(ER_TRUNCATE_ILLEGAL_FK, MYF(0), fk_info_str(thd, fk_info));
+    return TRUE;
   }
 
-  end_read_record(&info);
+  return FALSE;
+}
+
+
+/*
+  Open and truncate a locked table.
+
+  @param  thd           Thread context.
+  @param  table_ref     Table list element for the table to be truncated.
+  @param  is_tmp_table  True if element refers to a temp table.
+
+  @retval  0    Success.
+  @retval  > 0  Error code.
+*/
+
+int Truncate_statement::handler_truncate(THD *thd, TABLE_LIST *table_ref,
+                                         bool is_tmp_table)
+{
+  int error= 0;
+  uint flags;
+  DBUG_ENTER("Truncate_statement::handler_truncate");
 
   /*
-    Regardless of the error status, the query must be written to the
-    binary log if rows of the table is non-transactional.
+    Can't recreate, the engine must mechanically delete all rows
+    in the table. Use open_and_lock_tables() to open a write cursor.
   */
-  if (some_rows_deleted && !table->file->has_transactions())
+
+  /* If it is a temporary table, no need to take locks. */
+  if (is_tmp_table)
+    flags= MYSQL_OPEN_TEMPORARY_ONLY;
+  else
   {
-    thd->transaction.stmt.modified_non_trans_table= TRUE;
-    thd->transaction.all.modified_non_trans_table= TRUE;
+    /* We don't need to load triggers. */
+    DBUG_ASSERT(table_ref->trg_event_map == 0);
+    /*
+      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. Allow to open base tables only.
+    */
+    table_ref->required_type= FRMTYPE_TABLE;
+    /*
+      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.
+    */
+    flags= MYSQL_OPEN_IGNORE_FLUSH | MYSQL_OPEN_SKIP_TEMPORARY;
+    /*
+      Even though we have an MDL lock on the table here, we don't
+      pass MYSQL_OPEN_HAS_MDL_LOCK to open_and_lock_tables
+      since to truncate a MERGE table, we must open and lock
+      merge children, and on those we don't have an MDL lock.
+      Thus clear the ticket to satisfy MDL asserts.
+    */
+    table_ref->mdl_request.ticket= NULL;
   }
 
-  if (error || thd->killed)
-    goto end;
+  /* Open the table as it will handle some required preparations. */
+  if (open_and_lock_tables(thd, table_ref, FALSE, flags))
+    DBUG_RETURN(1);
 
-  /* 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;
-  }
+  /* Whether to truncate regardless of foreign keys. */
+  if (! (thd->variables.option_bits & OPTION_NO_FOREIGN_KEY_CHECKS))
+    error= fk_truncate_illegal_if_parent(thd, table_ref->table);
 
-end:
-  if (save_binlog_row_based)
-    thd->set_current_stmt_binlog_format_row();
+  if (!error && (error= table_ref->table->file->ha_truncate()))
+    table_ref->table->file->print_error(error, MYF(0));
 
   DBUG_RETURN(error);
 }
@@ -225,30 +289,29 @@ static bool recreate_temporary_table(THD *thd, TABLE *table)
 
 
 /*
-  Handle opening and locking if a base table for truncate.
+  Handle locking 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)
+bool Truncate_statement::lock_table(THD *thd, TABLE_LIST *table_ref,
+                                    bool *hton_can_recreate)
 {
   TABLE *table= NULL;
-  handlerton *table_type;
-  DBUG_ENTER("open_and_lock_table_for_truncate");
+  DBUG_ENTER("Truncate_statement::lock_table");
 
+  /* Lock types are set in the parser. */
   DBUG_ASSERT(table_ref->lock_type == TL_WRITE);
-  DBUG_ASSERT(table_ref->mdl_request.type == MDL_SHARED_NO_READ_WRITE);
+  /* The handler truncate protocol dictates a exclusive lock. */
+  DBUG_ASSERT(table_ref->mdl_request.type == MDL_EXCLUSIVE);
+
   /*
     Before doing anything else, acquire a metadata lock on the table,
     or ensure we have one.  We don't use open_and_lock_tables()
@@ -268,103 +331,45 @@ static bool open_and_lock_table_for_truncate(THD *thd, TABLE_LIST *table_ref,
                                             table_ref->table_name, FALSE)))
       DBUG_RETURN(TRUE);
 
-    table_type= table->s->db_type();
-    *hton_can_recreate= ha_check_storage_engine_flag(table_type,
+    *hton_can_recreate= ha_check_storage_engine_flag(table->s->db_type(),
                                                      HTON_CAN_RECREATE);
     table_ref->mdl_request.ticket= table->mdl_ticket;
   }
   else
   {
-    /*
-      Even though we could use the previous execution branch here just as
-      well, we must not try to open the table:
-    */
+    /* Acquire an exclusive lock. */
     DBUG_ASSERT(table_ref->next_global == NULL);
     if (lock_table_names(thd, table_ref, NULL,
                          thd->variables.lock_wait_timeout,
                          MYSQL_OPEN_SKIP_TEMPORARY))
       DBUG_RETURN(TRUE);
 
-    if (dd_frm_storage_engine(thd, table_ref->db, table_ref->table_name,
-                              &table_type))
+    if (dd_check_storage_engine_flag(thd, table_ref->db, table_ref->table_name,
+                                     HTON_CAN_RECREATE, hton_can_recreate))
       DBUG_RETURN(TRUE);
-    *hton_can_recreate= ha_check_storage_engine_flag(table_type,
-                                                     HTON_CAN_RECREATE);
   }
 
-#ifdef WITH_PARTITION_STORAGE_ENGINE
   /*
-    TODO: Add support for TRUNCATE PARTITION for NDB and other engines
-    supporting native partitioning.
+    A storage engine can recreate or truncate the table only if there
+    are no references to it from anywhere, i.e. no cached TABLE in the
+    table cache.
   */
-  if (thd->lex->alter_info.flags & ALTER_ADMIN_PARTITION &&
-      table_type != partition_hton)
-  {
-    my_error(ER_PARTITION_MGMT_ON_NONPARTITIONED, MYF(0));
-    DBUG_RETURN(TRUE);
-  }
-#endif
-  DEBUG_SYNC(thd, "lock_table_for_truncate");
-
-  if (*hton_can_recreate)
+  if (thd->locked_tables_mode)
   {
-    /*
-      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;
+    DEBUG_SYNC(thd, "upgrade_lock_for_truncate");
+    /* To remove the table from the cache we need an exclusive lock. */
+    if (wait_while_table_is_used(thd, table, HA_EXTRA_FORCE_REOPEN))
+      DBUG_RETURN(TRUE);
+    m_ticket_downgrade= table->mdl_ticket;
+    /* Close if table is going to be recreated. */
+    if (*hton_can_recreate)
       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(table_ref->mdl_request.ticket,
-                                           timeout))
-        DBUG_RETURN(TRUE);
-      tdc_remove_table(thd, TDC_RT_REMOVE_ALL, table_ref->db,
-                       table_ref->table_name, FALSE);
-    }
   }
   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);
-    /*
-      Even though we have an MDL lock on the table here, we don't
-      pass MYSQL_OPEN_HAS_MDL_LOCK to open_and_lock_tables
-      since to truncate a MERGE table, we must open and lock
-      merge children, and on those we don't have an MDL lock.
-      Thus clear the ticket to satisfy MDL asserts.
-    */
-    table_ref->mdl_request.ticket= NULL;
-
-    /*
-      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, FALSE,
-                             MYSQL_OPEN_IGNORE_FLUSH |
-                             MYSQL_OPEN_SKIP_TEMPORARY))
-      DBUG_RETURN(TRUE);
+    /* Table is already locked exclusively. Remove cached instances. */
+    tdc_remove_table(thd, TDC_RT_REMOVE_ALL, table_ref->db,
+                     table_ref->table_name, FALSE);
   }
 
   DBUG_RETURN(FALSE);
@@ -385,14 +390,17 @@ static bool open_and_lock_table_for_truncate(THD *thd, TABLE_LIST *table_ref,
   @retval  TRUE   Error.
 */
 
-bool mysql_truncate_table(THD *thd, TABLE_LIST *table_ref)
+bool Truncate_statement::truncate_table(THD *thd, TABLE_LIST *table_ref)
 {
+  int error;
   TABLE *table;
-  bool error= TRUE, binlog_stmt;
-  MDL_ticket *mdl_ticket= NULL;
-  DBUG_ENTER("mysql_truncate_table");
+  bool binlog_stmt;
+  DBUG_ENTER("Truncate_statement::truncate_table");
 
-  /* Remove tables from the HANDLER's hash. */
+  /* Initialize, or reinitialize in case of reexecution (SP). */
+  m_ticket_downgrade= NULL;
+
+  /* Remove table from the HANDLER's hash. */
   mysql_ha_rm_tables(thd, table_ref);
 
   /* If it is a temporary table, no need to take locks. */
@@ -413,14 +421,11 @@ bool mysql_truncate_table(THD *thd, TABLE_LIST *table_ref)
     {
       /*
         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.
+        table and invoke the handler truncate. 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, FALSE,
-                               MYSQL_OPEN_TEMPORARY_ONLY))
-        DBUG_RETURN(TRUE);
-
-      error= delete_all_rows(thd, table_ref->table);
+      error= handler_truncate(thd, table_ref, TRUE);
     }
 
     /*
@@ -434,8 +439,7 @@ bool mysql_truncate_table(THD *thd, TABLE_LIST *table_ref)
   {
     bool hton_can_recreate;
 
-    if (open_and_lock_table_for_truncate(thd, table_ref,
-                                         &hton_can_recreate, &mdl_ticket))
+    if (lock_table(thd, table_ref, &hton_can_recreate))
       DBUG_RETURN(TRUE);
 
     if (hton_can_recreate)
@@ -454,13 +458,18 @@ bool mysql_truncate_table(THD *thd, TABLE_LIST *table_ref)
     }
     else
     {
-      error= delete_all_rows(thd, table_ref->table);
+      /*
+        The engine does not support truncate-by-recreate.
+        Attempt to use the handler truncate method.
+      */
+      error= handler_truncate(thd, table_ref, FALSE);
 
       /*
-        Regardless of the error status, the query must be written to the
-        binary log if rows of a non-transactional table were deleted.
+        All effects of a TRUNCATE TABLE operation are committed even if
+        truncation fails. Thus, the query must be written to the binary
+        log. The only exception is a unimplemented truncate method.
       */
-      binlog_stmt= !error || thd->transaction.stmt.modified_non_trans_table;
+      binlog_stmt= !error || error != HA_ERR_WRONG_COMMAND;
     }
 
     query_cache_invalidate3(thd, table_ref, FALSE);
@@ -471,49 +480,37 @@ bool mysql_truncate_table(THD *thd, TABLE_LIST *table_ref)
     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);
-  }
-
-  /*
     A locked table ticket was upgraded to a exclusive lock. After the
     the query has been written to the binary log, downgrade the lock
     to a shared one.
   */
-  if (mdl_ticket)
-    mdl_ticket->downgrade_exclusive_lock(MDL_SHARED_NO_READ_WRITE);
+  if (m_ticket_downgrade)
+    m_ticket_downgrade->downgrade_exclusive_lock(MDL_SHARED_NO_READ_WRITE);
 
-  DBUG_PRINT("exit", ("error: %d", error));
-  DBUG_RETURN(test(error));
+  DBUG_RETURN(error);
 }
 
 
+/**
+  Execute a TRUNCATE statement at runtime.
+
+  @param  thd   The current thread.
+
+  @return FALSE on success.
+*/
+
 bool Truncate_statement::execute(THD *thd)
 {
-  TABLE_LIST *first_table= thd->lex->select_lex.table_list.first;
   bool res= TRUE;
+  TABLE_LIST *first_table= thd->lex->select_lex.table_list.first;
   DBUG_ENTER("Truncate_statement::execute");
 
   if (check_one_table_access(thd, DROP_ACL, first_table))
-    goto error;
-  /*
-    Don't allow this within a transaction because we want to use
-    re-generate table
-  */
-  if (thd->in_active_multi_stmt_transaction())
-  {
-    my_message(ER_LOCK_OR_ACTIVE_TRANSACTION,
-               ER(ER_LOCK_OR_ACTIVE_TRANSACTION), MYF(0));
-    goto error;
-  }
-  if (! (res= mysql_truncate_table(thd, first_table)))
+    DBUG_RETURN(res);
+
+  if (! (res= truncate_table(thd, first_table)))
     my_ok(thd);
-error:
+
   DBUG_RETURN(res);
 }
+
diff --git a/sql/sql_truncate.h b/sql/sql_truncate.h
index b8b1d3da53d..95a2f35df4f 100644
--- a/sql/sql_truncate.h
+++ b/sql/sql_truncate.h
@@ -18,13 +18,15 @@
 class THD;
 struct TABLE_LIST;
 
-bool mysql_truncate_table(THD *thd, TABLE_LIST *table_ref);
-
 /**
   Truncate_statement represents the TRUNCATE statement.
 */
 class Truncate_statement : public Sql_statement
 {
+private:
+  /* Set if a lock must be downgraded after truncate is done. */
+  MDL_ticket *m_ticket_downgrade;
+
 public:
   /**
     Constructor, used to represent a ALTER TABLE statement.
@@ -34,7 +36,7 @@ public:
     : Sql_statement(lex)
   {}
 
-  ~Truncate_statement()
+  virtual ~Truncate_statement()
   {}
 
   /**
@@ -43,7 +45,20 @@ public:
     @return false on success.
   */
   bool execute(THD *thd);
-};
 
+protected:
+  /** Handle locking a base table for truncate. */
+  bool lock_table(THD *, TABLE_LIST *, bool *);
+
+  /** Truncate table via the handler method. */
+  int handler_truncate(THD *, TABLE_LIST *, bool);
+
+  /**
+    Optimized delete of all rows by doing a full regenerate of the table.
+    Depending on the storage engine, it can be accomplished through a
+    drop and recreate or via the handler truncate method.
+  */
+  bool truncate_table(THD *, TABLE_LIST *);
+};
 
 #endif
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 7c24f18aa6a..2fc39757511 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -10770,7 +10770,7 @@ truncate:
             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_NO_READ_WRITE;
+            YYPS->m_mdl_type= MDL_EXCLUSIVE;
           }
           table_name
           {
diff --git a/sql/table.h b/sql/table.h
index 6723293c1ec..c8e1ad8e658 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -1170,7 +1170,9 @@ enum enum_schema_table_state
 
 typedef struct st_foreign_key_info
 {
-  LEX_STRING *forein_id;
+  LEX_STRING *foreign_id;
+  LEX_STRING *foreign_db;
+  LEX_STRING *foreign_table;
   LEX_STRING *referenced_db;
   LEX_STRING *referenced_table;
   LEX_STRING *update_method;