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.

21 files changed, 334 insertions, 184 deletions

diff --git a/storage/archive/ha_archive.cc b/storage/archive/ha_archive.cc
index ef907b035b5..c24fbc21f94 100644
--- a/storage/archive/ha_archive.cc
+++ b/storage/archive/ha_archive.cc
@@ -1650,9 +1650,9 @@ int ha_archive::end_bulk_insert()
   This is done for security reasons. In a later version we will enable this by 
   allowing the user to select a different row format.
 */
-int ha_archive::delete_all_rows()
+int ha_archive::truncate()
 {
-  DBUG_ENTER("ha_archive::delete_all_rows");
+  DBUG_ENTER("ha_archive::truncate");
   DBUG_RETURN(HA_ERR_WRONG_COMMAND);
 }
 
diff --git a/storage/archive/ha_archive.h b/storage/archive/ha_archive.h
index b258b403c3c..fec41c12b81 100644
--- a/storage/archive/ha_archive.h
+++ b/storage/archive/ha_archive.h
@@ -115,7 +115,7 @@ public:
   int close(void);
   int write_row(uchar * buf);
   int real_write_row(uchar *buf, azio_stream *writer);
-  int delete_all_rows();
+  int truncate();
   int rnd_init(bool scan=1);
   int rnd_next(uchar *buf);
   int rnd_pos(uchar * buf, uchar *pos);
diff --git a/storage/blackhole/ha_blackhole.cc b/storage/blackhole/ha_blackhole.cc
index 6591c3a2c78..04560d89be3 100644
--- a/storage/blackhole/ha_blackhole.cc
+++ b/storage/blackhole/ha_blackhole.cc
@@ -87,6 +87,16 @@ int ha_blackhole::create(const char *name, TABLE *table_arg,
   DBUG_RETURN(0);
 }
 
+/*
+  Intended to support partitioning.
+  Allows a particular partition to be truncated.
+*/
+int ha_blackhole::truncate()
+{
+  DBUG_ENTER("ha_blackhole::truncate");
+  DBUG_RETURN(0);
+}
+
 const char *ha_blackhole::index_type(uint key_number)
 {
   DBUG_ENTER("ha_blackhole::index_type");
diff --git a/storage/blackhole/ha_blackhole.h b/storage/blackhole/ha_blackhole.h
index 9de3c22c614..17066b6edce 100644
--- a/storage/blackhole/ha_blackhole.h
+++ b/storage/blackhole/ha_blackhole.h
@@ -76,6 +76,7 @@ public:
   uint max_supported_key_part_length() const { return BLACKHOLE_MAX_KEY_LENGTH; }
   int open(const char *name, int mode, uint test_if_locked);
   int close(void);
+  int truncate();
   int rnd_init(bool scan);
   int rnd_next(uchar *buf);
   int rnd_pos(uchar * buf, uchar *pos);
diff --git a/storage/example/ha_example.cc b/storage/example/ha_example.cc
index 306f8eaeccd..6af471494af 100644
--- a/storage/example/ha_example.cc
+++ b/storage/example/ha_example.cc
@@ -356,14 +356,14 @@ int ha_example::close(void)
   is happening. buf() is a byte array of data. You can use the field
   information to extract the data from the native byte array type.
 
-    @details
+  @details
   Example of this would be:
-    @code
+  @code
   for (Field **field=table->field ; *field ; field++)
   {
     ...
   }
-    @endcode
+  @endcode
 
   See ha_tina.cc for an example of extracting all of the data as strings.
   ha_berekly.cc has an example of how to store it intact by "packing" it
@@ -375,7 +375,7 @@ int ha_example::close(void)
   Called from item_sum.cc, item_sum.cc, sql_acl.cc, sql_insert.cc,
   sql_insert.cc, sql_select.cc, sql_table.cc, sql_udf.cc, and sql_update.cc.
 
-    @see
+  @see
   item_sum.cc, item_sum.cc, sql_acl.cc, sql_insert.cc,
   sql_insert.cc, sql_select.cc, sql_table.cc, sql_udf.cc and sql_update.cc
 */
@@ -400,19 +400,19 @@ int ha_example::write_row(uchar *buf)
   Keep in mind that the server can do updates based on ordering if an ORDER BY
   clause was used. Consecutive ordering is not guaranteed.
 
-    @details
+  @details
   Currently new_data will not have an updated auto_increament record, or
   and updated timestamp field. You can do these for example by doing:
-    @code
+  @code
   if (table->timestamp_field_type & TIMESTAMP_AUTO_SET_ON_UPDATE)
     table->timestamp_field->set_time();
   if (table->next_number_field && record == table->record[0])
     update_auto_increment();
-    @endcode
+  @endcode
 
   Called from sql_select.cc, sql_acl.cc, sql_update.cc, and sql_insert.cc.
 
-    @see
+  @see
   sql_select.cc, sql_acl.cc, sql_update.cc and sql_insert.cc
 */
 int ha_example::update_row(const uchar *old_data, uchar *new_data)
@@ -507,10 +507,10 @@ int ha_example::index_prev(uchar *buf)
   @brief
   index_first() asks for the first key in the index.
 
-    @details
+  @details
   Called from opt_range.cc, opt_sum.cc, sql_handler.cc, and sql_select.cc.
 
-    @see
+  @see
   opt_range.cc, opt_sum.cc, sql_handler.cc and sql_select.cc
 */
 int ha_example::index_first(uchar *buf)
@@ -528,10 +528,10 @@ int ha_example::index_first(uchar *buf)
   @brief
   index_last() asks for the last key in the index.
 
-    @details
+  @details
   Called from opt_range.cc, opt_sum.cc, sql_handler.cc, and sql_select.cc.
 
-    @see
+  @see
   opt_range.cc, opt_sum.cc, sql_handler.cc and sql_select.cc
 */
 int ha_example::index_last(uchar *buf)
@@ -551,11 +551,11 @@ int ha_example::index_last(uchar *buf)
   scan. See the example in the introduction at the top of this file to see when
   rnd_init() is called.
 
-    @details
+  @details
   Called from filesort.cc, records.cc, sql_handler.cc, sql_select.cc, sql_table.cc,
   and sql_update.cc.
 
-    @see
+  @see
   filesort.cc, records.cc, sql_handler.cc, sql_select.cc, sql_table.cc and sql_update.cc
 */
 int ha_example::rnd_init(bool scan)
@@ -578,11 +578,11 @@ int ha_example::rnd_end()
   The Field structure for the table is the key to getting data into buf
   in a manner that will allow the server to understand it.
 
-    @details
+  @details
   Called from filesort.cc, records.cc, sql_handler.cc, sql_select.cc, sql_table.cc,
   and sql_update.cc.
 
-    @see
+  @see
   filesort.cc, records.cc, sql_handler.cc, sql_select.cc, sql_table.cc and sql_update.cc
 */
 int ha_example::rnd_next(uchar *buf)
@@ -602,11 +602,11 @@ int ha_example::rnd_next(uchar *buf)
   position() is called after each call to rnd_next() if the data needs
   to be ordered. You can do something like the following to store
   the position:
-    @code
+  @code
   my_store_ptr(ref, ref_length, current_position);
-    @endcode
+  @endcode
 
-    @details
+  @details
   The server uses ref to store data. ref_length in the above case is
   the size needed to store current_position. ref is just a byte array
   that the server will maintain. If you are using offsets to mark rows, then
@@ -615,7 +615,7 @@ int ha_example::rnd_next(uchar *buf)
 
   Called from filesort.cc, sql_select.cc, sql_delete.cc, and sql_update.cc.
 
-    @see
+  @see
   filesort.cc, sql_select.cc, sql_delete.cc and sql_update.cc
 */
 void ha_example::position(const uchar *record)
@@ -632,10 +632,10 @@ void ha_example::position(const uchar *record)
   ref. You can use ha_get_ptr(pos,ref_length) to retrieve whatever key
   or position you saved when position() was called.
 
-    @details
+  @details
   Called from filesort.cc, records.cc, sql_insert.cc, sql_select.cc, and sql_update.cc.
 
-    @see
+  @see
   filesort.cc, records.cc, sql_insert.cc, sql_select.cc and sql_update.cc
 */
 int ha_example::rnd_pos(uchar *buf, uchar *pos)
@@ -655,15 +655,15 @@ int ha_example::rnd_pos(uchar *buf, uchar *pos)
   ::info() is used to return information to the optimizer. See my_base.h for
   the complete description.
 
-    @details
+  @details
   Currently this table handler doesn't implement most of the fields really needed.
   SHOW also makes use of this data.
 
   You will probably want to have the following in your code:
-    @code
+  @code
   if (records < 2)
     records = 2;
-    @endcode
+  @endcode
   The reason is that the server will optimize for cases of only a single
   record. If, in a table scan, you don't know the number of records, it
   will probably be better to set records to two so you can return as many
@@ -682,7 +682,7 @@ int ha_example::rnd_pos(uchar *buf, uchar *pos)
   sql_select.cc, sql_select.cc, sql_show.cc, sql_show.cc, sql_show.cc, sql_show.cc,
   sql_table.cc, sql_union.cc, and sql_update.cc.
 
-    @see
+  @see
   filesort.cc, ha_heap.cc, item_sum.cc, opt_sum.cc, sql_delete.cc, sql_delete.cc,
   sql_derived.cc, sql_select.cc, sql_select.cc, sql_select.cc, sql_select.cc,
   sql_select.cc, sql_show.cc, sql_show.cc, sql_show.cc, sql_show.cc, sql_table.cc,
@@ -716,14 +716,14 @@ int ha_example::extra(enum ha_extra_function operation)
   Used to delete all rows in a table, including cases of truncate and cases where
   the optimizer realizes that all rows will be removed as a result of an SQL statement.
 
-    @details
+  @details
   Called from item_sum.cc by Item_func_group_concat::clear(),
   Item_sum_count_distinct::clear(), and Item_func_group_concat::clear().
   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().
 
-    @see
+  @see
   Item_func_group_concat::clear(), Item_sum_count_distinct::clear() and
   Item_func_group_concat::clear() in item_sum.cc;
   mysql_delete() in sql_delete.cc;
@@ -739,17 +739,40 @@ int ha_example::delete_all_rows()
 
 /**
   @brief
+  Used for handler specific truncate table.  The table is locked in
+  exclusive mode and handler is responsible for reseting the auto-
+  increment counter.
+
+  @details
+  Called from Truncate_statement::handler_truncate.
+  Not used if the handlerton supports HTON_CAN_RECREATE, unless this
+  engine can be used as a partition. In this case, it is invoked when
+  a particular partition is to be truncated.
+
+  @see
+  Truncate_statement in sql_truncate.cc
+  Remarks in handler::truncate.
+*/
+int ha_example::truncate()
+{
+  DBUG_ENTER("ha_example::truncate");
+  DBUG_RETURN(HA_ERR_WRONG_COMMAND);
+}
+
+
+/**
+  @brief
   This create a lock on the table. If you are implementing a storage engine
   that can handle transacations look at ha_berkely.cc to see how you will
   want to go about doing this. Otherwise you should consider calling flock()
   here. Hint: Read the section "locking functions for mysql" in lock.cc to understand
   this.
 
-    @details
+  @details
   Called from lock.cc by lock_external() and unlock_external(). Also called
   from sql_table.cc by copy_data_between_tables().
 
-    @see
+  @see
   lock.cc by lock_external() and unlock_external() in lock.cc;
   the section "locking functions for mysql" in lock.cc;
   copy_data_between_tables() in sql_table.cc.
@@ -767,7 +790,7 @@ int ha_example::external_lock(THD *thd, int lock_type)
   should be needed for the table. For updates/deletes/inserts we get WRITE
   locks, for SELECT... we get read locks.
 
-    @details
+  @details
   Before adding the lock into the table lock handler (see thr_lock.c),
   mysqld calls store lock with the requested locks. Store lock can now
   modify a write lock to a read lock (or some other lock), ignore the
@@ -790,12 +813,12 @@ int ha_example::external_lock(THD *thd, int lock_type)
 
   Called from lock.cc by get_lock_data().
 
-    @note
+  @note
   In this method one should NEVER rely on table->in_use, it may, in fact,
   refer to a different thread! (this happens if get_lock_data() is called
   from mysql_lock_abort_for_thread() function)
 
-    @see
+  @see
   get_lock_data() in lock.cc
 */
 THR_LOCK_DATA **ha_example::store_lock(THD *thd,
@@ -816,7 +839,7 @@ THR_LOCK_DATA **ha_example::store_lock(THD *thd,
   shared references released). The variable name will just be the name of
   the table. You will need to remove any files you have created at this point.
 
-    @details
+  @details
   If you do not implement this, the default delete_table() is called from
   handler.cc and it will delete all files with the file extensions returned
   by bas_ext().
@@ -825,7 +848,7 @@ THR_LOCK_DATA **ha_example::store_lock(THD *thd,
   during create if the table_flag HA_DROP_BEFORE_CREATE was specified for
   the storage engine.
 
-    @see
+  @see
   delete_table and ha_create_table() in handler.cc
 */
 int ha_example::delete_table(const char *name)
diff --git a/storage/example/ha_example.h b/storage/example/ha_example.h
index 12e088f5f05..5555a10930b 100644
--- a/storage/example/ha_example.h
+++ b/storage/example/ha_example.h
@@ -245,6 +245,7 @@ public:
   int extra(enum ha_extra_function operation);
   int external_lock(THD *thd, int lock_type);                   ///< required
   int delete_all_rows(void);
+  int truncate();
   ha_rows records_in_range(uint inx, key_range *min_key,
                            key_range *max_key);
   int delete_table(const char *from);
diff --git a/storage/federated/ha_federated.cc b/storage/federated/ha_federated.cc
index b1ae276dce8..fb10ce82334 100644
--- a/storage/federated/ha_federated.cc
+++ b/storage/federated/ha_federated.cc
@@ -3042,6 +3042,16 @@ int ha_federated::delete_all_rows()
 
 
 /*
+  Used to manually truncate the table via a delete of all rows in a table.
+*/
+
+int ha_federated::truncate()
+{
+  return delete_all_rows();
+}
+
+
+/*
   The idea with handler::store_lock() is the following:
 
   The statement decided which locks we should need for the table
diff --git a/storage/federated/ha_federated.h b/storage/federated/ha_federated.h
index 0f4c0201bd7..e39bac525b5 100644
--- a/storage/federated/ha_federated.h
+++ b/storage/federated/ha_federated.h
@@ -248,6 +248,7 @@ public:
   int optimize(THD* thd, HA_CHECK_OPT* check_opt);
 
   int delete_all_rows(void);
+  int truncate();
   int create(const char *name, TABLE *form,
              HA_CREATE_INFO *create_info);                      //required
   ha_rows records_in_range(uint inx, key_range *start_key,
diff --git a/storage/heap/ha_heap.cc b/storage/heap/ha_heap.cc
index 481257def1d..fc870d33fb2 100644
--- a/storage/heap/ha_heap.cc
+++ b/storage/heap/ha_heap.cc
@@ -455,6 +455,13 @@ int ha_heap::delete_all_rows()
 }
 
 
+int ha_heap::truncate()
+{
+  int error= delete_all_rows();
+  return error ? error : reset_auto_increment(0);
+}
+
+
 int ha_heap::reset_auto_increment(ulonglong value)
 {
   file->s->auto_increment= value;
diff --git a/storage/heap/ha_heap.h b/storage/heap/ha_heap.h
index 7185fbc7720..5f3d66cd53c 100644
--- a/storage/heap/ha_heap.h
+++ b/storage/heap/ha_heap.h
@@ -99,6 +99,7 @@ public:
   int reset();
   int external_lock(THD *thd, int lock_type);
   int delete_all_rows(void);
+  int truncate();
   int reset_auto_increment(ulonglong value);
   int disable_indexes(uint mode);
   int enable_indexes(uint mode);
diff --git a/storage/ibmdb2i/db2i_constraints.cc b/storage/ibmdb2i/db2i_constraints.cc
index 3afa12032d0..5b9d3600393 100644
--- a/storage/ibmdb2i/db2i_constraints.cc
+++ b/storage/ibmdb2i/db2i_constraints.cc
@@ -494,10 +494,10 @@ int ha_ibmdb2i::get_foreign_key_list(THD *thd, List<FOREIGN_KEY_INFO> *f_key_lis
 
         convFromEbcdic(FKCstDef->CstName.Name, convName,FKCstDef->CstName.Len);
         if (convName[0] == '"')        // If quoted, exclude quotes. 
-          f_key_info.forein_id = thd_make_lex_string(thd, 0,
+          f_key_info.foreign_id = thd_make_lex_string(thd, 0,
                       convName + 1, (uint) (FKCstDef->CstName.Len - 2), 1);
         else                           // Not quoted                         
-          f_key_info.forein_id = thd_make_lex_string(thd, 0,
+          f_key_info.foreign_id = thd_make_lex_string(thd, 0,
                       convName, (uint) FKCstDef->CstName.Len, 1);
  
       /* Process the names of the foreign keys.                                 */
diff --git a/storage/ibmdb2i/ha_ibmdb2i.cc b/storage/ibmdb2i/ha_ibmdb2i.cc
index 947df8ad2fe..e5eccad1573 100644
--- a/storage/ibmdb2i/ha_ibmdb2i.cc
+++ b/storage/ibmdb2i/ha_ibmdb2i.cc
@@ -1806,6 +1806,13 @@ int ha_ibmdb2i::delete_all_rows()
 }
 
 
+int ha_ibmdb2i::truncate()
+{
+  int error = delete_all_rows();
+  return error ? error : reset_auto_increment(0);
+}
+
+
 int ha_ibmdb2i::external_lock(THD *thd, int lock_type)
 {
   int rc = 0;
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index eafc73b4f87..d68fcadffad 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -7081,33 +7081,21 @@ Deletes all rows of an InnoDB table.
 @return	error number */
 UNIV_INTERN
 int
-ha_innobase::delete_all_rows(void)
+ha_innobase::truncate(void)
 /*==============================*/
 {
 	int		error;
 
-	DBUG_ENTER("ha_innobase::delete_all_rows");
+	DBUG_ENTER("ha_innobase::truncate");
 
 	/* Get the transaction associated with the current thd, or create one
 	if not yet created, and update prebuilt->trx */
 
 	update_thd(ha_thd());
 
-	if (thd_sql_command(user_thd) != SQLCOM_TRUNCATE) {
-	fallback:
-		/* We only handle TRUNCATE TABLE t as a special case.
-		DELETE FROM t will have to use ha_innobase::delete_row(),
-		because DELETE is transactional while TRUNCATE is not. */
-		DBUG_RETURN(my_errno=HA_ERR_WRONG_COMMAND);
-	}
-
 	/* Truncate the table in InnoDB */
 
 	error = row_truncate_table_for_mysql(prebuilt->table, prebuilt->trx);
-	if (error == DB_ERROR) {
-		/* Cannot truncate; resort to ha_innobase::delete_row() */
-		goto fallback;
-	}
 
 	error = convert_error_code_to_mysql(error, prebuilt->table->flags,
 					    NULL);
@@ -8315,136 +8303,199 @@ ha_innobase::get_foreign_key_create_info(void)
 }
 
 
+/***********************************************************************//**
+Maps a InnoDB foreign key constraint to a equivalent MySQL foreign key info.
+@return pointer to foreign key info */
+static
+FOREIGN_KEY_INFO*
+get_foreign_key_info(
+/*=================*/
+	THD*			thd,		/*!< in: user thread handle */
+	dict_foreign_t*		foreign)	/*!< in: foreign key constraint */
+{
+	FOREIGN_KEY_INFO	f_key_info;
+	FOREIGN_KEY_INFO*	pf_key_info;
+	uint			i = 0;
+	ulint			len;
+	char			tmp_buff[NAME_LEN+1];
+	char			name_buff[NAME_LEN+1];
+	const char*		ptr;
+	LEX_STRING*		referenced_key_name;
+	LEX_STRING*		name = NULL;
+
+	ptr = dict_remove_db_name(foreign->id);
+	f_key_info.foreign_id = thd_make_lex_string(thd, 0, ptr,
+						    (uint) strlen(ptr), 1);
+
+	/* Name format: database name, '/', table name, '\0' */
+
+	/* Referenced (parent) database name */
+	len = dict_get_db_name_len(foreign->referenced_table_name);
+	ut_a(len < sizeof(tmp_buff));
+	ut_memcpy(tmp_buff, foreign->referenced_table_name, len);
+	tmp_buff[len] = 0;
+
+	len = filename_to_tablename(tmp_buff, name_buff, sizeof(name_buff));
+	f_key_info.referenced_db = thd_make_lex_string(thd, 0, name_buff, len, 1);
+
+	/* Referenced (parent) table name */
+	ptr = dict_remove_db_name(foreign->referenced_table_name);
+	len = filename_to_tablename(ptr, name_buff, sizeof(name));
+	f_key_info.referenced_table = thd_make_lex_string(thd, 0, name_buff, len, 1);
+
+	/* Dependent (child) database name */
+	len = dict_get_db_name_len(foreign->foreign_table_name);
+	ut_a(len < sizeof(tmp_buff));
+	ut_memcpy(tmp_buff, foreign->foreign_table_name, len);
+	tmp_buff[len] = 0;
+
+	len = filename_to_tablename(tmp_buff, name_buff, sizeof(name_buff));
+	f_key_info.foreign_db = thd_make_lex_string(thd, 0, name_buff, len, 1);
+
+	/* Dependent (child) table name */
+	ptr = dict_remove_db_name(foreign->foreign_table_name);
+	len = filename_to_tablename(ptr, name_buff, sizeof(name_buff));
+	f_key_info.foreign_table = thd_make_lex_string(thd, 0, name_buff, len, 1);
+
+	do {
+		ptr = foreign->foreign_col_names[i];
+		name = thd_make_lex_string(thd, name, ptr,
+					   (uint) strlen(ptr), 1);
+		f_key_info.foreign_fields.push_back(name);
+		ptr = foreign->referenced_col_names[i];
+		name = thd_make_lex_string(thd, name, ptr,
+					   (uint) strlen(ptr), 1);
+		f_key_info.referenced_fields.push_back(name);
+	} while (++i < foreign->n_fields);
+
+	if (foreign->type & DICT_FOREIGN_ON_DELETE_CASCADE) {
+		len = 7;
+		ptr = "CASCADE";
+	} else if (foreign->type & DICT_FOREIGN_ON_DELETE_SET_NULL) {
+		len = 8;
+		ptr = "SET NULL";
+	} else if (foreign->type & DICT_FOREIGN_ON_DELETE_NO_ACTION) {
+		len = 9;
+		ptr = "NO ACTION";
+	} else {
+		len = 8;
+		ptr = "RESTRICT";
+	}
+
+	f_key_info.delete_method = thd_make_lex_string(thd,
+						       f_key_info.delete_method,
+						       ptr, len, 1);
+
+	if (foreign->type & DICT_FOREIGN_ON_UPDATE_CASCADE) {
+		len = 7;
+		ptr = "CASCADE";
+	} else if (foreign->type & DICT_FOREIGN_ON_UPDATE_SET_NULL) {
+		len = 8;
+		ptr = "SET NULL";
+	} else if (foreign->type & DICT_FOREIGN_ON_UPDATE_NO_ACTION) {
+		len = 9;
+		ptr = "NO ACTION";
+	} else {
+		len = 8;
+		ptr = "RESTRICT";
+	}
+
+	f_key_info.update_method = thd_make_lex_string(thd,
+						       f_key_info.update_method,
+						       ptr, len, 1);
+
+	if (foreign->referenced_index && foreign->referenced_index->name) {
+		referenced_key_name = thd_make_lex_string(thd,
+					f_key_info.referenced_key_name,
+					foreign->referenced_index->name,
+					 (uint) strlen(foreign->referenced_index->name),
+					1);
+	} else {
+		referenced_key_name = NULL;
+	}
+
+	f_key_info.referenced_key_name = referenced_key_name;
+
+	pf_key_info = (FOREIGN_KEY_INFO *) thd_memdup(thd, &f_key_info,
+						      sizeof(FOREIGN_KEY_INFO));
+
+	return(pf_key_info);
+}
+
+/*******************************************************************//**
+Gets the list of foreign keys in this table.
+@return always 0, that is, always succeeds */
 UNIV_INTERN
 int
-ha_innobase::get_foreign_key_list(THD *thd, List<FOREIGN_KEY_INFO> *f_key_list)
-{
-  dict_foreign_t* foreign;
-
-  DBUG_ENTER("get_foreign_key_list");
-  ut_a(prebuilt != NULL);
-  update_thd(ha_thd());
-  prebuilt->trx->op_info = (char*)"getting list of foreign keys";
-  trx_search_latch_release_if_reserved(prebuilt->trx);
-  mutex_enter(&(dict_sys->mutex));
-  foreign = UT_LIST_GET_FIRST(prebuilt->table->foreign_list);
-
-  while (foreign != NULL) {
-	  uint i;
-	  FOREIGN_KEY_INFO f_key_info;
-	  LEX_STRING *name= 0;
-          uint ulen;
-          char uname[NAME_LEN+1];           /* Unencoded name */
-          char db_name[NAME_LEN+1];
-	  const char *tmp_buff;
-
-	  tmp_buff= foreign->id;
-	  i= 0;
-	  while (tmp_buff[i] != '/')
-		  i++;
-	  tmp_buff+= i + 1;
-	  f_key_info.forein_id = thd_make_lex_string(thd, 0,
-		  tmp_buff, (uint) strlen(tmp_buff), 1);
-	  tmp_buff= foreign->referenced_table_name;
-
-          /* Database name */
-	  i= 0;
-	  while (tmp_buff[i] != '/')
-          {
-            db_name[i]= tmp_buff[i];
-            i++;
-          }
-          db_name[i]= 0;
-          ulen= filename_to_tablename(db_name, uname, sizeof(uname));
-	  f_key_info.referenced_db = thd_make_lex_string(thd, 0,
-		  uname, ulen, 1);
-
-          /* Table name */
-	  tmp_buff+= i + 1;
-          ulen= filename_to_tablename(tmp_buff, uname, sizeof(uname));
-	  f_key_info.referenced_table = thd_make_lex_string(thd, 0,
-		  uname, ulen, 1);
-
-	  for (i= 0;;) {
-		  tmp_buff= foreign->foreign_col_names[i];
-		  name = thd_make_lex_string(thd, name,
-			  tmp_buff, (uint) strlen(tmp_buff), 1);
-		  f_key_info.foreign_fields.push_back(name);
-		  tmp_buff= foreign->referenced_col_names[i];
-		  name = thd_make_lex_string(thd, name,
-			tmp_buff, (uint) strlen(tmp_buff), 1);
-		  f_key_info.referenced_fields.push_back(name);
-		  if (++i >= foreign->n_fields)
-			  break;
-	  }
-
-          ulong length;
-          if (foreign->type & DICT_FOREIGN_ON_DELETE_CASCADE)
-          {
-            length=7;
-            tmp_buff= "CASCADE";
-          }
-          else if (foreign->type & DICT_FOREIGN_ON_DELETE_SET_NULL)
-          {
-            length=8;
-            tmp_buff= "SET NULL";
-          }
-          else if (foreign->type & DICT_FOREIGN_ON_DELETE_NO_ACTION)
-          {
-            length=9;
-            tmp_buff= "NO ACTION";
-          }
-          else
-          {
-            length=8;
-            tmp_buff= "RESTRICT";
-          }
-	  f_key_info.delete_method = thd_make_lex_string(
-		  thd, f_key_info.delete_method, tmp_buff, length, 1);
-
-
-          if (foreign->type & DICT_FOREIGN_ON_UPDATE_CASCADE)
-          {
-            length=7;
-            tmp_buff= "CASCADE";
-          }
-          else if (foreign->type & DICT_FOREIGN_ON_UPDATE_SET_NULL)
-          {
-            length=8;
-            tmp_buff= "SET NULL";
-          }
-          else if (foreign->type & DICT_FOREIGN_ON_UPDATE_NO_ACTION)
-          {
-            length=9;
-            tmp_buff= "NO ACTION";
-          }
-          else
-          {
-            length=8;
-            tmp_buff= "RESTRICT";
-          }
-	  f_key_info.update_method = thd_make_lex_string(
-		  thd, f_key_info.update_method, tmp_buff, length, 1);
-          if (foreign->referenced_index &&
-              foreign->referenced_index->name)
-          {
-	    f_key_info.referenced_key_name = thd_make_lex_string(
-		    thd, f_key_info.referenced_key_name,
-		    foreign->referenced_index->name,
-		    (uint) strlen(foreign->referenced_index->name), 1);
-          }
-          else
-            f_key_info.referenced_key_name= 0;
-
-	  FOREIGN_KEY_INFO *pf_key_info = (FOREIGN_KEY_INFO *)
-		  thd_memdup(thd, &f_key_info, sizeof(FOREIGN_KEY_INFO));
-	  f_key_list->push_back(pf_key_info);
-	  foreign = UT_LIST_GET_NEXT(foreign_list, foreign);
-  }
-  mutex_exit(&(dict_sys->mutex));
-  prebuilt->trx->op_info = (char*)"";
+ha_innobase::get_foreign_key_list(
+/*==============================*/
+	THD*			thd,		/*!< in: user thread handle */
+	List<FOREIGN_KEY_INFO>*	f_key_list)	/*!< out: foreign key list */
+{
+	FOREIGN_KEY_INFO*	pf_key_info;
+	dict_foreign_t*		foreign;
+
+	ut_a(prebuilt != NULL);
+	update_thd(ha_thd());
+
+	prebuilt->trx->op_info = "getting list of foreign keys";
+
+	trx_search_latch_release_if_reserved(prebuilt->trx);
+
+	mutex_enter(&(dict_sys->mutex));
+
+	for (foreign = UT_LIST_GET_FIRST(prebuilt->table->foreign_list);
+	     foreign != NULL;
+	     foreign = UT_LIST_GET_NEXT(referenced_list, foreign)) {
+		pf_key_info = get_foreign_key_info(thd, foreign);
+		if (pf_key_info) {
+			f_key_list->push_back(pf_key_info);
+		}
+	}
+
+	mutex_exit(&(dict_sys->mutex));
+
+	prebuilt->trx->op_info = "";
+
+	return(0);
+}
+
+/*******************************************************************//**
+Gets the set of foreign keys where this table is the referenced table.
+@return always 0, that is, always succeeds */
+UNIV_INTERN
+int
+ha_innobase::get_parent_foreign_key_list(
+/*=====================================*/
+	THD*			thd,		/*!< in: user thread handle */
+	List<FOREIGN_KEY_INFO>*	f_key_list)	/*!< out: foreign key list */
+{
+	FOREIGN_KEY_INFO*	pf_key_info;
+	dict_foreign_t*		foreign;
 
-  DBUG_RETURN(0);
+	ut_a(prebuilt != NULL);
+	update_thd(ha_thd());
+
+	prebuilt->trx->op_info = "getting list of referencing foreign keys";
+
+	trx_search_latch_release_if_reserved(prebuilt->trx);
+
+	mutex_enter(&(dict_sys->mutex));
+
+	for (foreign = UT_LIST_GET_FIRST(prebuilt->table->referenced_list);
+	     foreign != NULL;
+	     foreign = UT_LIST_GET_NEXT(referenced_list, foreign)) {
+		pf_key_info = get_foreign_key_info(thd, foreign);
+		if (pf_key_info) {
+			f_key_list->push_back(pf_key_info);
+		}
+	}
+
+	mutex_exit(&(dict_sys->mutex));
+
+	prebuilt->trx->op_info = "";
+
+	return(0);
 }
 
 /*****************************************************************//**
diff --git a/storage/innobase/handler/ha_innodb.h b/storage/innobase/handler/ha_innodb.h
index 8f118199ad8..4e128178f2b 100644
--- a/storage/innobase/handler/ha_innodb.h
+++ b/storage/innobase/handler/ha_innodb.h
@@ -178,13 +178,15 @@ class ha_innobase: public handler
 	void update_create_info(HA_CREATE_INFO* create_info);
 	int create(const char *name, register TABLE *form,
 					HA_CREATE_INFO *create_info);
-	int delete_all_rows();
+	int truncate();
 	int delete_table(const char *name);
 	int rename_table(const char* from, const char* to);
 	int check(THD* thd, HA_CHECK_OPT* check_opt);
 	char* update_table_comment(const char* comment);
 	char* get_foreign_key_create_info();
 	int get_foreign_key_list(THD *thd, List<FOREIGN_KEY_INFO> *f_key_list);
+	int get_parent_foreign_key_list(THD *thd,
+					List<FOREIGN_KEY_INFO> *f_key_list);
 	bool can_switch_engines();
 	uint referenced_by_foreign_key();
 	void free_foreign_key_create_info(char* str);
diff --git a/storage/innobase/row/row0mysql.c b/storage/innobase/row/row0mysql.c
index c12970b40ed..62f9ab07af9 100644
--- a/storage/innobase/row/row0mysql.c
+++ b/storage/innobase/row/row0mysql.c
@@ -2985,8 +2985,7 @@ next_rec:
 		dict_table_change_id_in_cache(table, new_id);
 	}
 
-	/* MySQL calls ha_innobase::reset_auto_increment() which does
-	the same thing. */
+	/* Reset auto-increment. */
 	dict_table_autoinc_lock(table);
 	dict_table_autoinc_initialize(table, 1);
 	dict_table_autoinc_unlock(table);
diff --git a/storage/myisam/ha_myisam.cc b/storage/myisam/ha_myisam.cc
index 87de58076cd..2ba62d03c6b 100644
--- a/storage/myisam/ha_myisam.cc
+++ b/storage/myisam/ha_myisam.cc
@@ -1788,6 +1788,18 @@ int ha_myisam::delete_all_rows()
   return mi_delete_all_rows(file);
 }
 
+
+/*
+  Intended to support partitioning.
+  Allows a particular partition to be truncated.
+*/
+
+int ha_myisam::truncate()
+{
+  int error= delete_all_rows();
+  return error ? error : reset_auto_increment(0);
+}
+
 int ha_myisam::reset_auto_increment(ulonglong value)
 {
   file->s->state.auto_increment= value;
diff --git a/storage/myisam/ha_myisam.h b/storage/myisam/ha_myisam.h
index 1a8246ae882..46b732512ac 100644
--- a/storage/myisam/ha_myisam.h
+++ b/storage/myisam/ha_myisam.h
@@ -107,6 +107,7 @@ class ha_myisam: public handler
   int reset(void);
   int external_lock(THD *thd, int lock_type);
   int delete_all_rows(void);
+  int truncate();
   int reset_auto_increment(ulonglong value);
   int disable_indexes(uint mode);
   int enable_indexes(uint mode);
diff --git a/storage/myisammrg/ha_myisammrg.cc b/storage/myisammrg/ha_myisammrg.cc
index 26e54d61101..79696cad721 100644
--- a/storage/myisammrg/ha_myisammrg.cc
+++ b/storage/myisammrg/ha_myisammrg.cc
@@ -1226,6 +1226,22 @@ ha_rows ha_myisammrg::records_in_range(uint inx, key_range *min_key,
 }
 
 
+int ha_myisammrg::truncate()
+{
+  int err= 0;
+  MYRG_TABLE *table;
+  DBUG_ENTER("ha_myisammrg::truncate");
+
+  for (table= file->open_tables; table != file->end_table; table++)
+  {
+    if ((err= mi_delete_all_rows(table->table)))
+      break;
+  }
+
+  DBUG_RETURN(err);
+}
+
+
 int ha_myisammrg::info(uint flag)
 {
   MYMERGE_INFO mrg_info;
diff --git a/storage/myisammrg/ha_myisammrg.h b/storage/myisammrg/ha_myisammrg.h
index 4ff24c69071..cae4bd17f50 100644
--- a/storage/myisammrg/ha_myisammrg.h
+++ b/storage/myisammrg/ha_myisammrg.h
@@ -131,6 +131,7 @@ public:
   int rnd_pos(uchar * buf, uchar *pos);
   void position(const uchar *record);
   ha_rows records_in_range(uint inx, key_range *min_key, key_range *max_key);
+  int truncate();
   int info(uint);
   int reset(void);
   int extra(enum ha_extra_function operation);
diff --git a/storage/perfschema/ha_perfschema.cc b/storage/perfschema/ha_perfschema.cc
index e5e324d6c6b..896117397a2 100644
--- a/storage/perfschema/ha_perfschema.cc
+++ b/storage/perfschema/ha_perfschema.cc
@@ -345,6 +345,11 @@ int ha_perfschema::delete_all_rows(void)
   DBUG_RETURN(result);
 }
 
+int ha_perfschema::truncate()
+{
+  return delete_all_rows();
+}
+
 THR_LOCK_DATA **ha_perfschema::store_lock(THD *thd,
                                        THR_LOCK_DATA **to,
                                        enum thr_lock_type lock_type)
diff --git a/storage/perfschema/ha_perfschema.h b/storage/perfschema/ha_perfschema.h
index 2c7b45fbbf7..1a0c16541be 100644
--- a/storage/perfschema/ha_perfschema.h
+++ b/storage/perfschema/ha_perfschema.h
@@ -127,6 +127,8 @@ public:
 
   int delete_all_rows(void);
 
+  int truncate();
+
   int delete_table(const char *from);
 
   int rename_table(const char * from, const char * to);