MDEV-24408 Crash-safe DROP DATABASE

Description of how DROP DATABASE works after this patch

- Collect list of tables
- DDL log tables as they are dropped
- DDL log drop database
- Delete db.opt
- Delete data directory
- Log either DROP TABLE or DROP DATABASE to binary log
- De active ddl log entry

This is in line of how things where before (minus ddl logging) except that
we delete db.opt file last to not loose it if DROP DATABASE fails.

On recovery we have to ensure that all dropped tables are logged in
binary log and that they are properly dropped (as with atomic drop
table).
No new tables be dropped as part of recovery.

Recovery of active drop database ddl log entry:

- If drop database was logged to ddl log but was not found in the binary
  log:
  - drop the db.opt file and database directory.
  - Log DROP DATABASE to binary log
- If drop database was not logged to ddl log
  - Update binary log with DROP TABLE of the dropped tables. If table list
    is longer than max_allowed_packet, then the query will be split into
    multiple DROP TABLE/VIEW queries.

Other things:
- Added DDL_LOG_STATE and 'current database' as arguments to
  mysql_rm_table_no_locks(). This was needed to be able to combine
  ddl logging of DROP DATABASE and DROP TABLE and make the generated
  DROP TABLE statements shorter.
- To make the DROP TABLE statement created by ddl log shorter, I changed
  the binlogged query to use current directory and omit the directory
  part for all tables in the current directory.
- Merged some DROP TABLE and DROP VIEW code in ddl logger.  This was done
  to be able get separate DROP VIEW and DROP TABLE statements in the binary
  log.
- Added a 'recovery_state' variable to remember the state of dropped
  tables and views.
- Moved out code that drops database objects (stored procedures) from
  mysql_rm_db_internal() to drop_database_objects() for better code reuse.
- Made mysql_rm_db_internal() global so that could be used by the ddl
  recovery code.

mysql-test/suite/atomic/drop_db.result

+call mtr.add_suppression("InnoDB: .* does not exist in the InnoDB internal");
+"engine: aria  crash point: ddl_log_drop_before_delete_table  position: 1"
+t1v.frm
+t2.MAD
+t2.MAI
+t2.frm
+Procedure	sql_mode	Create Procedure	character_set_client	collation_connection	Database Collation
+foo	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` PROCEDURE `foo`()
+insert into test.t1 values (42)	latin1	latin1_swedish_ci	latin1_swedish_ci
+master-bin.000002	#	Query	#	#	use `test2`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+"engine: aria  crash point: ddl_log_drop_before_delete_table  position: 2"
+t1v.frm
+t2.MAD
+t2.MAI
+t2.frm
+Procedure	sql_mode	Create Procedure	character_set_client	collation_connection	Database Collation
+foo	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` PROCEDURE `foo`()
+insert into test.t1 values (42)	latin1	latin1_swedish_ci	latin1_swedish_ci
+master-bin.000002	#	Query	#	#	use `test2`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+master-bin.000002	#	Query	#	#	use `test2`; DROP VIEW IF EXISTS `t1v` /* generated by ddl recovery */
+"engine: aria  crash point: ddl_log_drop_before_delete_table  position: 3"
+Procedure	sql_mode	Create Procedure	character_set_client	collation_connection	Database Collation
+foo	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` PROCEDURE `foo`()
+insert into test.t1 values (42)	latin1	latin1_swedish_ci	latin1_swedish_ci
+master-bin.000002	#	Query	#	#	use `test2`; DROP TABLE IF EXISTS `t1`,`t2` /* generated by ddl recovery */
+master-bin.000002	#	Query	#	#	use `test2`; DROP VIEW IF EXISTS `t1v` /* generated by ddl recovery */
+"engine: aria  crash point: ddl_log_drop_after_drop_tables  position: 1"
+Procedure	sql_mode	Create Procedure	character_set_client	collation_connection	Database Collation
+foo	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` PROCEDURE `foo`()
+insert into test.t1 values (42)	latin1	latin1_swedish_ci	latin1_swedish_ci
+master-bin.000002	#	Query	#	#	use `test2`; DROP TABLE IF EXISTS `t1`,`t2` /* generated by ddl recovery */
+master-bin.000002	#	Query	#	#	use `test2`; DROP VIEW IF EXISTS `t1v` /* generated by ddl recovery */
+"engine: aria  crash point: ddl_log_drop_before_ha_drop_database  position: 1"
+master-bin.000002	#	Query	#	#	DROP DATABASE IF EXISTS `test2` /* generated by ddl recovery */
+"engine: aria  crash point: ddl_log_drop_before_drop_db_routines  position: 1"
+master-bin.000002	#	Query	#	#	DROP DATABASE IF EXISTS `test2` /* generated by ddl recovery */
+"engine: aria  crash point: ddl_log_drop_after_drop_db_routines  position: 1"
+master-bin.000002	#	Query	#	#	DROP DATABASE IF EXISTS `test2` /* generated by ddl recovery */
+"engine: aria  crash point: ddl_log_drop_before_drop_option_file  position: 1"
+master-bin.000002	#	Query	#	#	DROP DATABASE IF EXISTS `test2` /* generated by ddl recovery */
+"engine: aria  crash point: ddl_log_drop_before_drop_dir  position: 1"
+master-bin.000002	#	Query	#	#	DROP DATABASE IF EXISTS `test2` /* generated by ddl recovery */
+"engine: aria  crash point: ddl_log_drop_after_drop_dir  position: 1"
+master-bin.000002	#	Query	#	#	DROP DATABASE IF EXISTS `test2` /* generated by ddl recovery */
+"engine: aria  crash point: ddl_log_drop_before_binlog  position: 1"
+master-bin.000002	#	Query	#	#	DROP DATABASE IF EXISTS `test2` /* generated by ddl recovery */
+"engine: aria  crash point: ddl_log_drop_after_binlog  position: 1"
+master-bin.000001	#	Query	#	#	DROP DATABASE test2
+"engine: innodb  crash point: ddl_log_drop_before_delete_table  position: 1"
+t1v.frm
+t2.frm
+t2.ibd
+Procedure	sql_mode	Create Procedure	character_set_client	collation_connection	Database Collation
+foo	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` PROCEDURE `foo`()
+insert into test.t1 values (42)	latin1	latin1_swedish_ci	latin1_swedish_ci
+master-bin.000002	#	Query	#	#	use `test2`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+"engine: innodb  crash point: ddl_log_drop_before_delete_table  position: 2"
+t1v.frm
+t2.frm
+t2.ibd
+Procedure	sql_mode	Create Procedure	character_set_client	collation_connection	Database Collation
+foo	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` PROCEDURE `foo`()
+insert into test.t1 values (42)	latin1	latin1_swedish_ci	latin1_swedish_ci
+master-bin.000002	#	Query	#	#	use `test2`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
+master-bin.000002	#	Query	#	#	use `test2`; DROP VIEW IF EXISTS `t1v` /* generated by ddl recovery */
+"engine: innodb  crash point: ddl_log_drop_before_delete_table  position: 3"
+Procedure	sql_mode	Create Procedure	character_set_client	collation_connection	Database Collation
+foo	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` PROCEDURE `foo`()
+insert into test.t1 values (42)	latin1	latin1_swedish_ci	latin1_swedish_ci
+master-bin.000002	#	Query	#	#	use `test2`; DROP TABLE IF EXISTS `t1`,`t2` /* generated by ddl recovery */
+master-bin.000002	#	Query	#	#	use `test2`; DROP VIEW IF EXISTS `t1v` /* generated by ddl recovery */
+"engine: innodb  crash point: ddl_log_drop_after_drop_tables  position: 1"
+Procedure	sql_mode	Create Procedure	character_set_client	collation_connection	Database Collation
+foo	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` PROCEDURE `foo`()
+insert into test.t1 values (42)	latin1	latin1_swedish_ci	latin1_swedish_ci
+master-bin.000002	#	Query	#	#	use `test2`; DROP TABLE IF EXISTS `t1`,`t2` /* generated by ddl recovery */
+master-bin.000002	#	Query	#	#	use `test2`; DROP VIEW IF EXISTS `t1v` /* generated by ddl recovery */
+"engine: innodb  crash point: ddl_log_drop_before_ha_drop_database  position: 1"
+master-bin.000002	#	Query	#	#	DROP DATABASE IF EXISTS `test2` /* generated by ddl recovery */
+"engine: innodb  crash point: ddl_log_drop_before_drop_db_routines  position: 1"
+master-bin.000002	#	Query	#	#	DROP DATABASE IF EXISTS `test2` /* generated by ddl recovery */
+"engine: innodb  crash point: ddl_log_drop_after_drop_db_routines  position: 1"
+master-bin.000002	#	Query	#	#	DROP DATABASE IF EXISTS `test2` /* generated by ddl recovery */
+"engine: innodb  crash point: ddl_log_drop_before_drop_option_file  position: 1"
+master-bin.000002	#	Query	#	#	DROP DATABASE IF EXISTS `test2` /* generated by ddl recovery */
+"engine: innodb  crash point: ddl_log_drop_before_drop_dir  position: 1"
+master-bin.000002	#	Query	#	#	DROP DATABASE IF EXISTS `test2` /* generated by ddl recovery */
+"engine: innodb  crash point: ddl_log_drop_after_drop_dir  position: 1"
+master-bin.000002	#	Query	#	#	DROP DATABASE IF EXISTS `test2` /* generated by ddl recovery */
+"engine: innodb  crash point: ddl_log_drop_before_binlog  position: 1"
+master-bin.000002	#	Query	#	#	DROP DATABASE IF EXISTS `test2` /* generated by ddl recovery */
+"engine: innodb  crash point: ddl_log_drop_after_binlog  position: 1"
+master-bin.000001	#	Query	#	#	DROP DATABASE test2
+Warnings:
+Note	1008	Can't drop database 'test2'; database doesn't exist

mysql-test/suite/atomic/drop_db.test

+--source include/have_debug.inc
+--source include/have_innodb.inc
+--source include/have_log_bin.inc
+--source include/not_valgrind.inc
+
+#
+# Testing of atomic DROP DATABASE with crashes in a lot of different places
+#
+
+call mtr.add_suppression("InnoDB: .* does not exist in the InnoDB internal");
+let $MYSQLD_DATADIR= `SELECT @@datadir`;
+
+let $engine_count=2;
+let $engines='aria','innodb';
+
+let $crash_count=10;
+let $crash_points='ddl_log_drop_before_delete_table','ddl_log_drop_after_drop_tables','ddl_log_drop_before_ha_drop_database','ddl_log_drop_before_drop_db_routines','ddl_log_drop_after_drop_db_routines','ddl_log_drop_before_drop_option_file','ddl_log_drop_before_drop_dir','ddl_log_drop_after_drop_dir','ddl_log_drop_before_binlog','ddl_log_drop_after_binlog';
+
+let $old_debug=`select @@debug_dbug`;
+
+let $e=0;
+let $keep_include_silent=1;
+let $grep_script=DROP;
+--disable_query_log
+
+while ($e < $engine_count)
+{
+  inc $e;
+  let $engine=`select ELT($e, $engines)`;
+  let $default_engine=$engine;
+  let $extra_option=;
+
+  if ($engine == "aria")
+  {
+    let $extra_option=transactional=1;
+  }
+  if ($engine == "aria_notrans")
+  {
+    let $default_engine="aria";
+    let $extra_option=transactional=0;
+  }
+
+  # Number of tables that should be dropped (we try to crash after each drop)
+  let $drops=3;
+
+  let $c=0;
+  while ($c < $crash_count)
+  {
+    inc $c;
+    let $crash=`select ELT($c, $crash_points)`;
+    let $r=0;
+    while ($r < $drops)
+    {
+      inc $r;
+      create database test2;
+      use test2;
+      --eval set @@default_storage_engine=$default_engine;
+      --eval create table t1 (a int not null) $extra_option;
+      create view t1v as select * from t1;
+      --eval create table t2 (b int not null) $extra_option;
+      create procedure foo()
+        insert into test.t1 values (42);
+
+      flush tables;
+      use test;
+
+      RESET MASTER;
+
+      echo "engine: $engine  crash point: $crash  position: $r";
+     --exec echo "restart" > $MYSQLTEST_VARDIR/tmp/mysqld.1.expect
+      --disable_reconnect
+      --eval set @@debug_dbug="+d,$crash",@debug_crash_counter=$r
+      let $errno=0;
+      --error 0,2013
+      DROP DATABASE test2;
+      let $error=$errno;
+      --enable_reconnect
+      --source include/wait_until_connected_again.inc
+      --disable_query_log
+      --eval set @@debug_dbug="$old_debug"
+
+      if ($error == 0)
+      {
+        echo "No crash!";
+      }
+      use test;
+      # Check which tables still exists
+      --error 0,1
+      --list_files $MYSQLD_DATADIR/test2 t*
+      --error 0,ER_SP_DOES_NOT_EXIST
+      show create procedure test2.foo;
+
+      --let $binlog_file=master-bin.000001
+      --source include/show_binlog_events.inc
+      if ($error)
+      {
+        --let $binlog_file=master-bin.000002
+        --source include/show_binlog_events.inc
+      }
+      # Really drop the tables. The warnings will show what was dropped
+      --disable_warnings
+      --error 0, ER_DB_DROP_EXISTS
+      DROP DATABASE test2;
+      --enable_warnings
+    }
+    # We only need to test drops for all tables for the first crash point
+    let $drops=1;
+  }
+}
+drop database if exists test2;
+
+--enable_query_log

mysql-test/suite/atomic/drop_db_long_names.opt

+--max-allowed-packet=1024 --net-buffer-length=1024

mysql-test/suite/atomic/drop_db_long_names.result

+"engine: aria  crash point: ddl_log_drop_after_drop_tables  position: 1"
+master-bin.000002	#	Query	#	#	use `test2`; DROP TABLE IF EXISTS `tABBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB`,`tACCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCC`,`tADDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDD`,`tAEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEE`,`tAFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF`,`tAGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGG`,`tAHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHH`,`tAIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIII`,`tAJJJJJJJJJJJJJJJJJJJJJJJJJJJJJJJJJJJJJJJJJJJJJJJJJJJJJJJJJJJJ`,`tAKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKK`,`tALLLLLLLLLLLLLLLLLLLLLLLLLLLLLLLLLLLLLLLLLLLLLLLLLLLLLLLLLLLL` /* generated by ddl recovery */
+master-bin.000002	#	Query	#	#	use `test2`; DROP VIEW IF EXISTS `tABBBBBBBBBBBBBBBBBBBBBBBBBBBBBBv`,`tACCCCCCCCCCCCCCCCCCCCCCCCCCCCCCv`,`tADDDDDDDDDDDDDDDDDDDDDDDDDDDDDDv`,`tAEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEv`,`tAFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFv`,`tAGGGGGGGGGGGGGGGGGGGGGGGGGGGGGGv`,`tAHHHHHHHHHHHHHHHHHHHHHHHHHHHHHHv`,`tAIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIv`,`tAJJJJJJJJJJJJJJJJJJJJJJJJJJJJJJv`,`tAKKKKKKKKKKKKKKKKKKKKKKKKKKKKKKv` /* generated by ddl recovery */
+master-bin.000002	#	Query	#	#	use `test2`; DROP TABLE IF EXISTS `tAMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMM`,`tANNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNNN`,`tAOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOO`,`tAPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPP`,`tAQQQQQQQQQQQQQQQQQQQQQQQQQQQQQQQQQQQQQQQQQQQQQQQQQQQQQQQQQQQQ`,`tARRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRR`,`tASSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSSS`,`tATTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTTT`,`tAUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUU`,`tAVVVVVVVVVVVVVVVVVVVVVVVVVVVVVVVVVVVVVVVVVVVVVVVVVVVVVVVVVVVV`,`tAWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWW` /* generated by ddl recovery */
+master-bin.000002	#	Query	#	#	use `test2`; DROP VIEW IF EXISTS `tALLLLLLLLLLLLLLLLLLLLLLLLLLLLLLv`,`tAMMMMMMMMMMMMMMMMMMMMMMMMMMMMMMv`,`tANNNNNNNNNNNNNNNNNNNNNNNNNNNNNNv`,`tAOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOv`,`tAPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPv`,`tAQQQQQQQQQQQQQQQQQQQQQQQQQQQQQQv`,`tARRRRRRRRRRRRRRRRRRRRRRRRRRRRRRv`,`tASSSSSSSSSSSSSSSSSSSSSSSSSSSSSSv`,`tATTTTTTTTTTTTTTTTTTTTTTTTTTTTTTv`,`tAUUUUUUUUUUUUUUUUUUUUUUUUUUUUUUv`,`tAVVVVVVVVVVVVVVVVVVVVVVVVVVVVVVv` /* generated by ddl recovery */
+master-bin.000002	#	Query	#	#	use `test2`; DROP TABLE IF EXISTS `tAXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX`,`tAYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYY`,`tAZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZ`,`tBAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA`,`tBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB`,`tBCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCC`,`tBDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDD`,`tBEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEE` /* generated by ddl recovery */
+master-bin.000002	#	Query	#	#	use `test2`; DROP VIEW IF EXISTS `tAWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWv`,`tAXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXv`,`tAYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYv`,`tAZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZv`,`tBAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAv`,`tBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBv`,`tBCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCv`,`tBDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDv`,`tBEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEv` /* generated by ddl recovery */
+"engine: aria  crash point: ddl_log_drop_before_binlog  position: 1"
+master-bin.000002	#	Query	#	#	DROP DATABASE IF EXISTS `test2` /* generated by ddl recovery */
+Warnings:
+Note	1008	Can't drop database 'test2'; database doesn't exist

mysql-test/suite/atomic/drop_db_long_names.test

+--source include/have_debug.inc
+--source include/have_log_bin.inc
+--source include/not_valgrind.inc
+
+#
+# Testing of atomic DROP DATABASE when the generated query could be too long
+#
+
+let $engine_count=1;
+let $engines='aria';
+
+let $crash_count=2;
+let $crash_points='ddl_log_drop_after_drop_tables','ddl_log_drop_before_binlog';
+
+let $max_tables=30;
+
+let $old_debug=`select @@debug_dbug`;
+
+let $keep_include_silent=1;
+let $grep_script=DROP;
+--disable_query_log
+
+let $e=0;
+while ($e < $engine_count)
+{
+  inc $e;
+  let $engine=`select ELT($e, $engines)`;
+  let $default_engine=$engine;
+  let $extra_option=;
+
+  if ($engine == "aria")
+  {
+    let $extra_option=transactional=1;
+  }
+  if ($engine == "aria_notrans")
+  {
+    let $default_engine="aria";
+    let $extra_option=transactional=0;
+  }
+
+  let $c=0;
+  while ($c < $crash_count)
+  {
+    inc $c;
+    let $crash=`select ELT($c, $crash_points)`;
+    let $r=0;
+    while ($r < 1)
+    {
+      inc $r;
+      create database test2;
+      use test2;
+      --eval set @@default_storage_engine=$default_engine
+      let $t=0;
+      while ($t < $max_tables)
+      {
+        inc $t;
+        let $name=`select concat("t",char(floor(65+$t/26)),repeat(char(65+mod($t,26)),60))`;
+        let $view=`select concat("t",char(floor(65+$t/26)),repeat(char(65+mod($t,26)),30),'v')`;
+        --eval create table $name (a int not null) $extra_option
+        --eval create view $view as select * from $name
+      }
+
+      flush tables;
+      use test;
+
+      RESET MASTER;
+
+      echo "engine: $engine  crash point: $crash  position: $r";
+     --exec echo "restart" > $MYSQLTEST_VARDIR/tmp/mysqld.1.expect
+      --disable_reconnect
+      --eval set @@debug_dbug="+d,$crash",@debug_crash_counter=$r
+      let $errno=0;
+      --error 0,2013
+      DROP DATABASE test2;
+      let $error=$errno;
+      --enable_reconnect
+      --source include/wait_until_connected_again.inc
+      --disable_query_log
+      --eval set @@debug_dbug="$old_debug"
+
+      if ($error == 0)
+      {
+        echo "No crash!";
+      }
+      use test;
+      # Check which tables still exists
+      --error 0,1
+      --list_files $MYSQLD_DATADIR/test2 t*
+      --error 0,ER_SP_DOES_NOT_EXIST
+
+      --let $binlog_file=master-bin.000001
+      --source include/show_binlog_events.inc
+      if ($error)
+      {
+        --let $binlog_file=master-bin.000002
+        --source include/show_binlog_events.inc
+      }
+      # Really drop the tables. The warnings will show what was dropped
+      --disable_warnings
+      --error 0, ER_DB_DROP_EXISTS
+      DROP DATABASE test2;
+      --enable_warnings
+    }
+  }
+}
+drop database if exists test2;
+
+--enable_query_log

mysql-test/suite/atomic/drop_sequence.result

 call mtr.add_suppression("InnoDB: .* does not exist in the InnoDB internal");
+create database test2;
 "engine: aria  crash point: ddl_log_drop_before_delete_table  position: 1"
+ts.MAD
+ts.MAI
+ts.frm
 t2.MAD
 t2.MAI
 t2.TRG
 t2.frm
 t2_trg.TRN
-ts.MAD
-ts.MAI
-ts.frm
-master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1` /* generated by ddl recovery */
+master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
 "engine: aria  crash point: ddl_log_drop_before_delete_table  position: 2"
 ts.MAD
 ts.MAI
 ts.frm
-master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2` /* generated by ddl recovery */
+master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1`,`test2`.`t2` /* generated by ddl recovery */
 "engine: aria  crash point: ddl_log_drop_before_delete_table  position: 3"
-master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2`,`test`.`ts` /* generated by ddl recovery */
+master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1`,`test2`.`t2`,`ts` /* generated by ddl recovery */
 "engine: aria  crash point: ddl_log_drop_after_delete_table  position: 1"
+ts.MAD
+ts.MAI
+ts.frm
 t2.MAD
 t2.MAI
 t2.TRG
 t2.frm
 t2_trg.TRN
-ts.MAD
-ts.MAI
-ts.frm
-master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1` /* generated by ddl recovery */
+master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
 "engine: aria  crash point: ddl_log_drop_after_delete_table  position: 2"
 ts.MAD
 ts.MAI
 ts.frm
-master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2` /* generated by ddl recovery */
+master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1`,`test2`.`t2` /* generated by ddl recovery */
 "engine: aria  crash point: ddl_log_drop_after_delete_table  position: 3"
-master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2`,`test`.`ts` /* generated by ddl recovery */
+master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1`,`test2`.`t2`,`ts` /* generated by ddl recovery */
 "engine: aria  crash point: ddl_log_drop_before_drop_trigger  position: 1"
+ts.MAD
+ts.MAI
+ts.frm
 t2.MAD
 t2.MAI
 t2.TRG
 t2.frm
 t2_trg.TRN
-ts.MAD
-ts.MAI
-ts.frm
-master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1` /* generated by ddl recovery */
+master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
 "engine: aria  crash point: ddl_log_drop_before_drop_trigger  position: 2"
 ts.MAD
 ts.MAI
 ts.frm
-master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2` /* generated by ddl recovery */
+master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1`,`test2`.`t2` /* generated by ddl recovery */
 "engine: aria  crash point: ddl_log_drop_before_drop_trigger  position: 3"
-master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2`,`test`.`ts` /* generated by ddl recovery */
+master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1`,`test2`.`t2`,`ts` /* generated by ddl recovery */
 "engine: aria  crash point: ddl_log_drop_before_drop_trigger2  position: 1"
+ts.MAD
+ts.MAI
+ts.frm
 t2.MAD
 t2.MAI
 t2.TRG
 t2.frm
 t2_trg.TRN
-ts.MAD
-ts.MAI
-ts.frm
-master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1` /* generated by ddl recovery */
+master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
 "engine: aria  crash point: ddl_log_drop_before_drop_trigger2  position: 2"
 ts.MAD
 ts.MAI
 ts.frm
-master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2` /* generated by ddl recovery */
+master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1`,`test2`.`t2` /* generated by ddl recovery */
 "engine: aria  crash point: ddl_log_drop_before_drop_trigger2  position: 3"
-master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2`,`test`.`ts` /* generated by ddl recovery */
+master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1`,`test2`.`t2`,`ts` /* generated by ddl recovery */
 "engine: aria  crash point: ddl_log_drop_after_drop_trigger  position: 1"
+ts.MAD
+ts.MAI
+ts.frm
 t2.MAD
 t2.MAI
 t2.TRG
 t2.frm
 t2_trg.TRN
-ts.MAD
-ts.MAI
-ts.frm
-master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1` /* generated by ddl recovery */
+master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1` /* generated by ddl recovery */
 "engine: aria  crash point: ddl_log_drop_after_drop_trigger  position: 2"
 ts.MAD
 ts.MAI
 ts.frm
-master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2` /* generated by ddl recovery */
+master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1`,`test2`.`t2` /* generated by ddl recovery */
 "engine: aria  crash point: ddl_log_drop_after_drop_trigger  position: 3"
-master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2`,`test`.`ts` /* generated by ddl recovery */
+master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1`,`test2`.`t2`,`ts` /* generated by ddl recovery */
 "engine: aria  crash point: ddl_log_drop_before_binlog  position: 1"
-master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2`,`test`.`ts` /* generated by ddl recovery */
+master-bin.000002	#	Query	#	#	use `test`; DROP TABLE IF EXISTS `t1`,`test2`.`t2`,`ts` /* generated by ddl recovery */
 "engine: aria  crash point: ddl_log_drop_before_binlog  position: 2"
 "No crash!"
-master-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t1`,`t2`,`ts` /* generated by server */
+master-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t1`,`test2`.`t2`,`ts` /* generated by server */
 "engine: aria  crash point: ddl_log_drop_before_binlog  position: 3"
 "No crash!"
-master-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t1`,`t2`,`ts` /* generated by server */
+master-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t1`,`test2`.`t2`,`ts` /* generated by server */
 "engine: aria  crash point: ddl_log_drop_after_binlog  position: 1"
-master-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t1`,`t2`,`ts` /* generated by server */
+master-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t1`,`test2`.`t2`,`ts` /* generated by server */
 "engine: aria  crash point: ddl_log_drop_after_binlog  position: 2"
 "No crash!"
-master-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t1`,`t2`,`ts` /* generated by server */
+master-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t1`,`test2`.`t2`,`ts` /* generated by server */
 "engine: aria  crash point: ddl_log_drop_after_binlog  position: 3"
 "No crash!"
-master-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t1`,`t2`,`ts` /* generated by server */
+master-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t1`,`test2`.`t2`,`ts` /* generated by server */
 Warnings:
-Note	1051	Unknown table 'test.t1,test.t2,test.ts'
+Note	1051	Unknown table 'test.t1,test2.t2,test.ts'

mysql-test/suite/atomic/drop_sequence.test

@@ -4,11 +4,14 @@
 
 #
 # Testing of atomic drop with crashes in a lot of different places
+# We also test having the tables in different databases
 #
 
 call mtr.add_suppression("InnoDB: .* does not exist in the InnoDB internal");
 let $MYSQLD_DATADIR= `SELECT @@datadir`;
 
+create database test2;
+
 let $engine_count=1;
 let $engines='aria';
 
@@ -53,10 +56,10 @@ while ($e < $engine_count)
       inc $r;
      --eval set @@default_storage_engine=$default_engine
      --eval create table t1 (a int not null) $extra_option;
-     --eval create table t2 (b int not null) $extra_option;
+     --eval create table test2.t2 (b int not null) $extra_option;
       create sequence ts;
       insert into t1 values(1);
-      insert into t2 values(2);
+      insert into test2.t2 values(2);
       flush tables;
 
       delimiter |;
@@ -66,7 +69,7 @@ while ($e < $engine_count)
           set new.a:= 1000;
         end if;
       end|
-      create trigger t2_trg before insert on t2 for each row
+      create trigger test2.t2_trg before insert on test2.t2 for each row
       begin
         if isnull(new.b) then
           set new.b:= 2000;
@@ -82,7 +85,7 @@ while ($e < $engine_count)
       --eval set @@debug_dbug="+d,$crash",@debug_crash_counter=$r
       let $errno=0;
       --error 0,2013
-      drop table t1,t2,ts;
+      drop table t1,test2.t2,ts;
       let $error=$errno;
       --enable_reconnect
       --source include/wait_until_connected_again.inc
@@ -95,6 +98,7 @@ while ($e < $engine_count)
       }
       # Check which tables still exists
       --list_files $MYSQLD_DATADIR/test t*
+      --list_files $MYSQLD_DATADIR/test2 t*
 
       --let $binlog_file=master-bin.000001
       --source include/show_binlog_events.inc
@@ -105,11 +109,11 @@ while ($e < $engine_count)
       }
       # Really drop the tables. The warnings will show what was dropped
       --disable_warnings
-      drop table if exists t1,t2,ts;
+      drop table if exists t1,test2.t2,ts;
       --enable_warnings
     }
   }
 }
-drop table if exists t1,t2,ts;
-
+drop table if exists t1,test2.t2,ts;
+drop database test2;
 --enable_query_log

mysql-test/suite/atomic/drop_view.result

 "engine: aria  crash point: ddl_log_drop_before_delete_view  position: 1"
 v2.frm
-master-bin.000002	#	Query	#	#	DROP VIEW IF EXISTS `test`.`v1` /* generated by ddl recovery */
+master-bin.000002	#	Query	#	#	use `test`; DROP VIEW IF EXISTS `v1` /* generated by ddl recovery */
 "engine: aria  crash point: ddl_log_drop_before_delete_view  position: 2"
-master-bin.000002	#	Query	#	#	DROP VIEW IF EXISTS `test`.`v1`,`test`.`v2` /* generated by ddl recovery */
+master-bin.000002	#	Query	#	#	use `test`; DROP VIEW IF EXISTS `v1`,`v2` /* generated by ddl recovery */
 "engine: aria  crash point: ddl_log_drop_after_delete_view  position: 1"
 v2.frm
-master-bin.000002	#	Query	#	#	DROP VIEW IF EXISTS `test`.`v1` /* generated by ddl recovery */
+master-bin.000002	#	Query	#	#	use `test`; DROP VIEW IF EXISTS `v1` /* generated by ddl recovery */
 "engine: aria  crash point: ddl_log_drop_after_delete_view  position: 2"
-master-bin.000002	#	Query	#	#	DROP VIEW IF EXISTS `test`.`v1`,`test`.`v2` /* generated by ddl recovery */
+master-bin.000002	#	Query	#	#	use `test`; DROP VIEW IF EXISTS `v1`,`v2` /* generated by ddl recovery */
 "engine: aria  crash point: ddl_log_drop_before_binlog  position: 1"
-master-bin.000002	#	Query	#	#	DROP VIEW IF EXISTS `test`.`v1`,`test`.`v2` /* generated by ddl recovery */
+master-bin.000002	#	Query	#	#	use `test`; DROP VIEW IF EXISTS `v1`,`v2` /* generated by ddl recovery */
 "engine: aria  crash point: ddl_log_drop_before_binlog  position: 2"
 "No crash!"
 master-bin.000001	#	Query	#	#	use `test`; DROP VIEW v1,v2

sql/ddl_log.h

@@ -77,11 +77,11 @@ enum ddl_log_action_code
   */
   DDL_LOG_RENAME_TABLE_ACTION= 5,
   DDL_LOG_RENAME_VIEW_ACTION= 6,
-  DDL_LOG_DROP_TABLE_INIT_ACTION= 7,
+  DDL_LOG_DROP_INIT_ACTION= 7,
   DDL_LOG_DROP_TABLE_ACTION= 8,
-  DDL_LOG_DROP_VIEW_INIT_ACTION= 9,
-  DDL_LOG_DROP_VIEW_ACTION= 10,
-  DDL_LOG_DROP_TRIGGER_ACTION= 11,
+  DDL_LOG_DROP_VIEW_ACTION= 9,
+  DDL_LOG_DROP_TRIGGER_ACTION= 10,
+  DDL_LOG_DROP_DB_ACTION=11,
   DDL_LOG_LAST_ACTION                          /* End marker */
 };
 
@@ -93,13 +93,15 @@ extern const uchar ddl_log_entry_phases[DDL_LOG_LAST_ACTION];
 enum enum_ddl_log_exchange_phase {
   EXCH_PHASE_NAME_TO_TEMP= 0,
   EXCH_PHASE_FROM_TO_NAME= 1,
-  EXCH_PHASE_TEMP_TO_FROM= 2
+  EXCH_PHASE_TEMP_TO_FROM= 2,
+  EXCH_PHASE_END
 };
 
 enum enum_ddl_log_rename_table_phase {
   DDL_RENAME_PHASE_TRIGGER= 0,
   DDL_RENAME_PHASE_STAT,
   DDL_RENAME_PHASE_TABLE,
+  DDL_RENAME_PHASE_END
 };
 
 enum enum_ddl_log_drop_table_phase {
@@ -110,6 +112,12 @@ enum enum_ddl_log_drop_table_phase {
   DDL_DROP_PHASE_END
 };
 
+enum enum_ddl_log_drop_db_phase {
+  DDL_DROP_DB_PHASE_INIT=0,
+  DDL_DROP_DB_PHASE_LOG,
+  DDL_DROP_DB_PHASE_END
+};
+
 /*
   Setting ddl_log_entry.phase to this has the same effect as setting
   the phase to the maximum phase (..PHASE_END) for an entry.
@@ -218,8 +226,10 @@ bool ddl_log_rename_view(THD *thd, DDL_LOG_STATE *ddl_state,
                          const LEX_CSTRING *new_db,
                          const LEX_CSTRING *new_alias);
 bool ddl_log_drop_table_init(THD *thd, DDL_LOG_STATE *ddl_state,
+                             const LEX_CSTRING *db,
                              const LEX_CSTRING *comment);
-bool ddl_log_drop_view_init(THD *thd, DDL_LOG_STATE *ddl_state);
+bool ddl_log_drop_view_init(THD *thd, DDL_LOG_STATE *ddl_state,
+                            const LEX_CSTRING *db);
 bool ddl_log_drop_table(THD *thd, DDL_LOG_STATE *ddl_state,
                         handlerton *hton,
                         const LEX_CSTRING *path,
@@ -234,5 +244,13 @@ bool ddl_log_drop_trigger(THD *thd, DDL_LOG_STATE *ddl_state,
                           const LEX_CSTRING *table,
                           const LEX_CSTRING *trigger_name,
                           const LEX_CSTRING *query);
+bool ddl_log_drop_view(THD *thd, DDL_LOG_STATE *ddl_state,
+                        const LEX_CSTRING *path,
+                        const LEX_CSTRING *db,
+                        const LEX_CSTRING *table);
+bool ddl_log_drop_view(THD *thd, DDL_LOG_STATE *ddl_state,
+                       const LEX_CSTRING *db);
+bool ddl_log_drop_db(THD *thd, DDL_LOG_STATE *ddl_state,
+                     const LEX_CSTRING *db, const LEX_CSTRING *path);
 extern mysql_mutex_t LOCK_gdl;
 #endif /* DDL_LOG_INCLUDED */

sql/handler.cc

@@ -839,9 +839,10 @@ static my_bool dropdb_handlerton(THD *unused1, plugin_ref plugin,
 }
 
 
-void ha_drop_database(char* path)
+void ha_drop_database(const char* path)
 {
-  plugin_foreach(NULL, dropdb_handlerton, MYSQL_STORAGE_ENGINE_PLUGIN, path);
+  plugin_foreach(NULL, dropdb_handlerton, MYSQL_STORAGE_ENGINE_PLUGIN,
+                 (char*) path);
 }
 
 

sql/handler.h

@@ -5180,7 +5180,7 @@ int ha_panic(enum ha_panic_function flag);
 void ha_close_connection(THD* thd);
 void ha_kill_query(THD* thd, enum thd_kill_levels level);
 bool ha_flush_logs();
-void ha_drop_database(char* path);
+void ha_drop_database(const char* path);
 void ha_checkpoint_state(bool disable);
 void ha_commit_checkpoint_request(void *cookie, void (*pre_hook)(void *));
 int ha_create_table(THD *thd, const char *path, const char *db,

sql/sql_db.h

@@ -47,6 +47,10 @@ void my_dbopt_cleanup(void);
 const char *normalize_db_name(const char *db, char *buffer,
                               size_t buffer_size);
 
+void drop_database_objects(THD *thd, const LEX_CSTRING *path,
+                           const LEX_CSTRING *db,
+                           bool rm_mysql_schema);
+my_bool rm_dir_w_symlink(const char *org_path, my_bool send_error);
 #define MY_DB_OPT_FILE "db.opt"
 
 #endif /* SQL_DB_INCLUDED */

sql/sql_table.cc

@@ -1041,7 +1041,9 @@ bool mysql_rm_table(THD *thd,TABLE_LIST *tables, bool if_exists,
 
   /* mark for close and remove all cached entries */
   thd->push_internal_handler(&err_handler);
-  error= mysql_rm_table_no_locks(thd, tables, if_exists, drop_temporary,
+  error= mysql_rm_table_no_locks(thd, tables, &thd->db, (DDL_LOG_STATE*) 0,
+                                 if_exists,
+                                 drop_temporary,
                                  false, drop_sequence, dont_log_query,
                                  false);
   thd->pop_internal_handler();
@@ -1102,6 +1104,11 @@ static uint32 get_comment(THD *thd, uint32 comment_pos,
 
   @param  thd             Thread handler
   @param  tables          Tables to drop
+  @param  current_db      Current database, used for ddl logs
+  @param  ddl_log_state   DDL log state, for global ddl logging (used by
+                          DROP DATABASE. If not set, an internal ddl log state
+                          will be used.  If set then the caller must call
+                          ddl_log_complete(ddl_log_state);
   @param  if_exists       If set, don't give an error if table doesn't exists.
                           In this case we give an warning of level 'NOTE'
   @param  drop_temporary  Only drop temporary tables
@@ -1130,7 +1137,10 @@ static uint32 get_comment(THD *thd, uint32 comment_pos,
         not all.
 */
 
-int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
+int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables,
+                            const LEX_CSTRING *current_db,
+                            DDL_LOG_STATE *ddl_log_state,
+                            bool if_exists,
                             bool drop_temporary, bool drop_view,
                             bool drop_sequence,
                             bool dont_log_query,
@@ -1140,7 +1150,7 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
   char path[FN_REFLEN + 1];
   LEX_CSTRING alias= null_clex_str;
   StringBuffer<160> unknown_tables(system_charset_info);
-  DDL_LOG_STATE ddl_log_state;
+  DDL_LOG_STATE local_ddl_log_state;
   const char *comment_start;
   uint not_found_errors= 0, table_count= 0, non_temp_tables_count= 0;
   int error= 0;
@@ -1156,6 +1166,12 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
   String built_trans_tmp_query, built_non_trans_tmp_query;
   DBUG_ENTER("mysql_rm_table_no_locks");
 
+  if (!ddl_log_state)
+  {
+    ddl_log_state= &local_ddl_log_state;
+    bzero(ddl_log_state, sizeof(*ddl_log_state));
+  }
+
   unknown_tables.length(0);
   comment_len= get_comment(thd, if_exists ? 17:9, &comment_start);
 
@@ -1209,14 +1225,13 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
     built_non_trans_tmp_query.set_charset(system_charset_info);
     built_non_trans_tmp_query.copy(built_trans_tmp_query);
   }
-  bzero(&ddl_log_state, sizeof(ddl_log_state));
 
   for (table= tables; table; table= table->next_local)
   {
     bool is_trans= 0, temporary_table_was_dropped= 0;
     bool table_creation_was_logged= 0;
     bool wrong_drop_sequence= 0;
-    bool table_dropped= 0;
+    bool table_dropped= 0, res;
     const LEX_CSTRING db= table->db;
     const LEX_CSTRING table_name= table->table_name;
     LEX_CSTRING cpath= {0,0};
@@ -1376,7 +1391,7 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
     if (!table_count++)
     {
       LEX_CSTRING comment= {comment_start, (size_t) comment_len};
-      if (ddl_log_drop_table_init(thd, &ddl_log_state, &comment))
+      if (ddl_log_drop_table_init(thd, ddl_log_state, current_db, &comment))
       {
         error= 1;
         goto err;
@@ -1434,12 +1449,18 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
 
       bool enoent_warning= !dont_log_query && !(hton && hton->discover_table);
 
-      if (ddl_log_drop_table(thd, &ddl_log_state, hton, &cpath, &db,
-                             &table_name))
+      if (was_view)
+        res= ddl_log_drop_view(thd, ddl_log_state, &cpath, &db,
+                               &table_name);
+      else
+        res= ddl_log_drop_table(thd, ddl_log_state, hton, &cpath, &db,
+                                &table_name);
+      if (res)
       {
           error= -1;
           goto err;
       }
+
       debug_crash_here("ddl_log_drop_before_delete_table");
       error= ha_delete_table(thd, hton, path, &db, &table_name,
                              enoent_warning);
@@ -1510,7 +1531,7 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
       int ferror= 0;
       DBUG_ASSERT(!was_view);
 
-      if (ddl_log_drop_table(thd, &ddl_log_state, hton, &cpath, &db,
+      if (ddl_log_drop_table(thd, ddl_log_state, 0, &cpath, &db,
                              &table_name))
       {
         error= -1;
@@ -1548,9 +1569,12 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
     if (thd->replication_flags & OPTION_IF_EXISTS)
       log_if_exists= 1;
 
+    if (!was_view)
+    {
       debug_crash_here("ddl_log_drop_before_drop_trigger");
-    ddl_log_update_phase(&ddl_log_state, DDL_DROP_PHASE_TRIGGER);
+      ddl_log_update_phase(ddl_log_state, DDL_DROP_PHASE_TRIGGER);
       debug_crash_here("ddl_log_drop_before_drop_trigger2");
+    }
 
     if (likely(!error) || non_existing_table_error(error))
     {
@@ -1610,7 +1634,8 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
                                 table_name.str, (uint)table_name.length);
       mysql_audit_drop_table(thd, table);
     }
-    ddl_log_update_phase(&ddl_log_state, DDL_DROP_PHASE_BINLOG);
+    if (!was_view)
+      ddl_log_update_phase(ddl_log_state, DDL_DROP_PHASE_BINLOG);
 
     if (!dont_log_query &&
         (!error || table_dropped || non_existing_table_error(error)))
@@ -1715,7 +1740,7 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
         built_query.append(normal_tables.ptr(), normal_tables.length());
         built_query.append(generated_by_server);
         thd->binlog_xid= thd->query_id;
-        ddl_log_update_xid(&ddl_log_state, thd->binlog_xid);
+        ddl_log_update_xid(ddl_log_state, thd->binlog_xid);
         error |= (thd->binlog_query(THD::STMT_QUERY_TYPE,
                                     built_query.ptr(),
                                     built_query.length(),
@@ -1725,7 +1750,8 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
       debug_crash_here("ddl_log_drop_after_binlog");
     }
   }
-  ddl_log_complete(&ddl_log_state);
+  if (ddl_log_state == &local_ddl_log_state)
+    ddl_log_complete(ddl_log_state);
 
   if (!drop_temporary)
   {
@@ -4234,7 +4260,9 @@ int create_table_impl(THD *thd, const LEX_CSTRING &orig_db,
         */
         (void) trans_rollback_stmt(thd);
         /* Remove normal table without logging. Keep tables locked */
-        if (mysql_rm_table_no_locks(thd, &table_list, 0, 0, 0, 0, 1, 1))
+        if (mysql_rm_table_no_locks(thd, &table_list, &thd->db,
+                                    (DDL_LOG_STATE*) 0,
+                                    0, 0, 0, 0, 1, 1))
           goto err;
 
         /*
@@ -4453,7 +4481,9 @@ int mysql_create_table_no_lock(THD *thd, const LEX_CSTRING *db,
     {
       DBUG_ASSERT(thd->is_error());
       /* Drop the table as it wasn't completely done */
-      if (!mysql_rm_table_no_locks(thd, table_list, 1,
+      if (!mysql_rm_table_no_locks(thd, table_list, &thd->db,
+                                   (DDL_LOG_STATE*) 0,
+                                   1,
                                    create_info->tmp_table(),
                                    false, true /* Sequence*/,
                                    true /* Don't log_query */,

sql/sql_table.h

@@ -37,6 +37,7 @@ typedef struct st_key KEY;
 typedef struct st_key_cache KEY_CACHE;
 typedef struct st_lock_param_type ALTER_PARTITION_PARAM_TYPE;
 typedef struct st_order ORDER;
+typedef struct st_ddl_log_state DDL_LOG_STATE;
 
 enum enum_explain_filename_mode
 {
@@ -177,7 +178,10 @@ bool mysql_checksum_table(THD* thd, TABLE_LIST* table_list,
 bool mysql_rm_table(THD *thd,TABLE_LIST *tables, bool if_exists,
                     bool drop_temporary, bool drop_sequence,
                     bool dont_log_query);
-int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
+int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables,
+                            const LEX_CSTRING *db,
+                            DDL_LOG_STATE *ddl_log_state,
+                            bool if_exists,
                             bool drop_temporary, bool drop_view,
                             bool drop_sequence,
                             bool dont_log_query, bool dont_free_locks);

sql/sql_view.cc

@@ -1870,7 +1870,7 @@ bool mysql_drop_view(THD *thd, TABLE_LIST *views, enum_drop_mode drop_mode)
     }
     if (!view_count++)
     {
-      if (ddl_log_drop_view_init(thd, &ddl_log_state))
+      if (ddl_log_drop_view_init(thd, &ddl_log_state, &thd->db))
         DBUG_RETURN(TRUE);
     }
     if (ddl_log_drop_view(thd, &ddl_log_state, &cpath, &view->db,