MDEV-23844 Atomic DROP TABLE (single table)

Logging logic:
- Log tables, just before they are dropped, to the ddl log
- After the last table for the statement is dropped, log an xid for the
  whole ddl log event

In case of crash:
- Remove first any active DROP TABLE events from the ddl log that matches
  xids found in binary log (this mean the drop was successful and was
  propery logged).
- Loop over all active DROP TABLE events
  - Ensure that the table is completely dropped
- Write a DROP TABLE entry to the binary log with the dropped tables.

Other things:
- Added code to ha_drop_table() to be able to tell the difference if
  a get_new_handler() failed because of out-of-memory or because the
  handler refused/was not able to create a a handler. This was needed
  to get sequences to work as sequences needs a share object to be passed
  to get_new_handler()
- TC_LOG_BINLOG::recover() was changed to always collect Xid's from the
  binary log and always call ddl_log_close_binlogged_events(). This was
  needed to be able to collect DROP TABLE events with embedded Xid's
  (used by ddl log).
- Added a new variable "$grep_script" to binlog filter to be able to find
  only rows that matches a regexp.
- Had to adjust some test that changed because drop statements are a bit
  larger in the binary log than before (as we have to store the xid)

Other things:
- MDEV-25588 Atomic DDL: Binlog query event written upon recovery is corrupt
  fixed (in the original commit).

15 files changed, 1136 insertions, 51 deletions

diff --git a/mysql-test/include/filter_file.inc b/mysql-test/include/filter_file.inc
index bfe53896710..708e917a644 100644
--- a/mysql-test/include/filter_file.inc
+++ b/mysql-test/include/filter_file.inc
@@ -56,6 +56,9 @@
 #
 #   $filter_script
 #     If set, rows matching this regexp will be filtered out
+#
+#   $grep_script
+#     If set, only include rows matching this regexp
 
 --let $include_filename= filter_file.inc
 --source include/begin_include_file.inc
@@ -71,6 +74,7 @@ if ($rpl_debug)
 --let _FF_PRE_SCRIPT= $pre_script
 --let _FF_SCRIPT= $script
 --let _FF_FILTER_SCRIPT= $filter_script
+--let _FF_GREP_SCRIPT= $grep_script
 --let _FF_INPUT_FILE= $input_file
 --let _FF_OUTPUT_FILE= $output_file
 --let _FF_SELECT_COLUMNS= $select_columns
@@ -85,6 +89,7 @@ perl;
   $pre_script =~ s/DOLLAR/\$/g;
   my $script = $ENV{'_FF_SCRIPT'};
   my $filter_script = $ENV{'_FF_FILTER_SCRIPT'};
+  my $grep_script = $ENV{'_FF_GREP_SCRIPT'};
   $script =~ s/DOLLAR/\$/g;
   my $input_file = $ENV{'_FF_INPUT_FILE'};
   my $output_file = $ENV{'_FF_OUTPUT_FILE'};
@@ -129,7 +134,8 @@ perl;
     {
       ' . $script . '
     }
-    if (!$filter_script || ! m/$filter_script/)
+    if ((!$filter_script || ! m/$filter_script/) &&
+       (!$grep_script || m/$grep_script/))
     {
       $filtered_contents .= $_."\n";
     }
diff --git a/mysql-test/suite/atomic/drop_sequence.result b/mysql-test/suite/atomic/drop_sequence.result
new file mode 100644
index 00000000000..6eb7d68961c
--- /dev/null
+++ b/mysql-test/suite/atomic/drop_sequence.result
@@ -0,0 +1,104 @@
+call mtr.add_suppression("InnoDB: .* does not exist in the InnoDB internal");
+"engine: aria  crash point: ddl_log_drop_before_delete_table  position: 1"
+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 */
+"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 */
+"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 */
+"engine: aria  crash point: ddl_log_drop_after_delete_table  position: 1"
+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 */
+"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 */
+"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 */
+"engine: aria  crash point: ddl_log_drop_before_drop_trigger  position: 1"
+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 */
+"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 */
+"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 */
+"engine: aria  crash point: ddl_log_drop_before_drop_trigger2  position: 1"
+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 */
+"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 */
+"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 */
+"engine: aria  crash point: ddl_log_drop_after_drop_trigger  position: 1"
+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 */
+"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 */
+"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 */
+"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 */
+"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 */
+"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 */
+"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 */
+"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 */
+"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 */
+Warnings:
+Note	1051	Unknown table 'test.t1,test.t2,test.ts'
diff --git a/mysql-test/suite/atomic/drop_sequence.test b/mysql-test/suite/atomic/drop_sequence.test
new file mode 100644
index 00000000000..b2fd60e8649
--- /dev/null
+++ b/mysql-test/suite/atomic/drop_sequence.test
@@ -0,0 +1,115 @@
+--source include/have_debug.inc
+--source include/have_log_bin.inc
+--source include/not_valgrind.inc
+
+#
+# Testing of atomic drop 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=1;
+let $engines='aria';
+
+let $crash_count=7;
+let $crash_points='ddl_log_drop_before_delete_table', 'ddl_log_drop_after_delete_table', 'ddl_log_drop_before_drop_trigger', 'ddl_log_drop_before_drop_trigger2', 'ddl_log_drop_after_drop_trigger', 'ddl_log_drop_before_binlog', 'ddl_log_drop_after_binlog';
+
+# Number of drops in the tested statement
+let $drops=3;
+
+let $old_debug=`select @@debug_dbug`;
+
+let $e=0;
+let $keep_include_silent=1;
+let $grep_script=DROP TABLE;
+--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;
+  }
+
+  let $c=0;
+  while ($c < $crash_count)
+  {
+    inc $c;
+    let $crash=`select ELT($c, $crash_points)`;
+    let $r=0;
+    while ($r < $drops)
+    {
+      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;
+      create sequence ts;
+      insert into t1 values(1);
+      insert into t2 values(2);
+      flush tables;
+
+      delimiter |;
+      create trigger t1_trg before insert on t1 for each row
+      begin
+        if isnull(new.a) then
+          set new.a:= 1000;
+        end if;
+      end|
+      create trigger t2_trg before insert on t2 for each row
+      begin
+        if isnull(new.b) then
+          set new.b:= 2000;
+        end if;
+      end|
+      delimiter ;|
+
+      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 table t1,t2,ts;
+      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!";
+      }
+      # Check which tables still exists
+      --list_files $MYSQLD_DATADIR/test t*
+
+      --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
+      drop table if exists t1,t2,ts;
+      --enable_warnings
+    }
+  }
+}
+drop table if exists t1,t2,ts;
+
+--enable_query_log
diff --git a/mysql-test/suite/atomic/drop_table.result b/mysql-test/suite/atomic/drop_table.result
new file mode 100644
index 00000000000..50c9708341f
--- /dev/null
+++ b/mysql-test/suite/atomic/drop_table.result
@@ -0,0 +1,273 @@
+call mtr.add_suppression("InnoDB: .* does not exist in the InnoDB internal");
+"engine: myisam  crash point: ddl_log_drop_before_delete_table  position: 1"
+t2.MYD
+t2.MYI
+t2.TRG
+t2.frm
+t2_trg.TRN
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1` /* generated by ddl recovery */
+"engine: myisam  crash point: ddl_log_drop_before_delete_table  position: 2"
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2` /* generated by ddl recovery */
+"engine: myisam  crash point: ddl_log_drop_after_delete_table  position: 1"
+t2.MYD
+t2.MYI
+t2.TRG
+t2.frm
+t2_trg.TRN
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1` /* generated by ddl recovery */
+"engine: myisam  crash point: ddl_log_drop_after_delete_table  position: 2"
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2` /* generated by ddl recovery */
+"engine: myisam  crash point: ddl_log_drop_before_drop_trigger  position: 1"
+t2.MYD
+t2.MYI
+t2.TRG
+t2.frm
+t2_trg.TRN
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1` /* generated by ddl recovery */
+"engine: myisam  crash point: ddl_log_drop_before_drop_trigger  position: 2"
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2` /* generated by ddl recovery */
+"engine: myisam  crash point: ddl_log_drop_before_drop_trigger2  position: 1"
+t2.MYD
+t2.MYI
+t2.TRG
+t2.frm
+t2_trg.TRN
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1` /* generated by ddl recovery */
+"engine: myisam  crash point: ddl_log_drop_before_drop_trigger2  position: 2"
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2` /* generated by ddl recovery */
+"engine: myisam  crash point: ddl_log_drop_after_drop_trigger  position: 1"
+t2.MYD
+t2.MYI
+t2.TRG
+t2.frm
+t2_trg.TRN
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1` /* generated by ddl recovery */
+"engine: myisam  crash point: ddl_log_drop_after_drop_trigger  position: 2"
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2` /* generated by ddl recovery */
+"engine: myisam  crash point: ddl_log_drop_before_binlog  position: 1"
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2` /* generated by ddl recovery */
+"engine: myisam  crash point: ddl_log_drop_before_binlog  position: 2"
+"No crash!"
+master-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t1`,`t2` /* generated by server */
+"engine: myisam  crash point: ddl_log_drop_after_binlog  position: 1"
+master-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t1`,`t2` /* generated by server */
+"engine: myisam  crash point: ddl_log_drop_after_binlog  position: 2"
+"No crash!"
+master-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t1`,`t2` /* generated by server */
+"engine: aria  crash point: ddl_log_drop_before_delete_table  position: 1"
+t2.MAD
+t2.MAI
+t2.TRG
+t2.frm
+t2_trg.TRN
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1` /* generated by ddl recovery */
+"engine: aria  crash point: ddl_log_drop_before_delete_table  position: 2"
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2` /* generated by ddl recovery */
+"engine: aria  crash point: ddl_log_drop_after_delete_table  position: 1"
+t2.MAD
+t2.MAI
+t2.TRG
+t2.frm
+t2_trg.TRN
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1` /* generated by ddl recovery */
+"engine: aria  crash point: ddl_log_drop_after_delete_table  position: 2"
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2` /* generated by ddl recovery */
+"engine: aria  crash point: ddl_log_drop_before_drop_trigger  position: 1"
+t2.MAD
+t2.MAI
+t2.TRG
+t2.frm
+t2_trg.TRN
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1` /* generated by ddl recovery */
+"engine: aria  crash point: ddl_log_drop_before_drop_trigger  position: 2"
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2` /* generated by ddl recovery */
+"engine: aria  crash point: ddl_log_drop_before_drop_trigger2  position: 1"
+t2.MAD
+t2.MAI
+t2.TRG
+t2.frm
+t2_trg.TRN
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1` /* generated by ddl recovery */
+"engine: aria  crash point: ddl_log_drop_before_drop_trigger2  position: 2"
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2` /* generated by ddl recovery */
+"engine: aria  crash point: ddl_log_drop_after_drop_trigger  position: 1"
+t2.MAD
+t2.MAI
+t2.TRG
+t2.frm
+t2_trg.TRN
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1` /* generated by ddl recovery */
+"engine: aria  crash point: ddl_log_drop_after_drop_trigger  position: 2"
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2` /* 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` /* 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` /* generated by server */
+"engine: aria  crash point: ddl_log_drop_after_binlog  position: 1"
+master-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t1`,`t2` /* 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` /* generated by server */
+"engine: aria_notrans  crash point: ddl_log_drop_before_delete_table  position: 1"
+t2.MAD
+t2.MAI
+t2.TRG
+t2.frm
+t2_trg.TRN
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1` /* generated by ddl recovery */
+"engine: aria_notrans  crash point: ddl_log_drop_before_delete_table  position: 2"
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2` /* generated by ddl recovery */
+"engine: aria_notrans  crash point: ddl_log_drop_after_delete_table  position: 1"
+t2.MAD
+t2.MAI
+t2.TRG
+t2.frm
+t2_trg.TRN
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1` /* generated by ddl recovery */
+"engine: aria_notrans  crash point: ddl_log_drop_after_delete_table  position: 2"
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2` /* generated by ddl recovery */
+"engine: aria_notrans  crash point: ddl_log_drop_before_drop_trigger  position: 1"
+t2.MAD
+t2.MAI
+t2.TRG
+t2.frm
+t2_trg.TRN
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1` /* generated by ddl recovery */
+"engine: aria_notrans  crash point: ddl_log_drop_before_drop_trigger  position: 2"
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2` /* generated by ddl recovery */
+"engine: aria_notrans  crash point: ddl_log_drop_before_drop_trigger2  position: 1"
+t2.MAD
+t2.MAI
+t2.TRG
+t2.frm
+t2_trg.TRN
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1` /* generated by ddl recovery */
+"engine: aria_notrans  crash point: ddl_log_drop_before_drop_trigger2  position: 2"
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2` /* generated by ddl recovery */
+"engine: aria_notrans  crash point: ddl_log_drop_after_drop_trigger  position: 1"
+t2.MAD
+t2.MAI
+t2.TRG
+t2.frm
+t2_trg.TRN
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1` /* generated by ddl recovery */
+"engine: aria_notrans  crash point: ddl_log_drop_after_drop_trigger  position: 2"
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2` /* generated by ddl recovery */
+"engine: aria_notrans  crash point: ddl_log_drop_before_binlog  position: 1"
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2` /* generated by ddl recovery */
+"engine: aria_notrans  crash point: ddl_log_drop_before_binlog  position: 2"
+"No crash!"
+master-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t1`,`t2` /* generated by server */
+"engine: aria_notrans  crash point: ddl_log_drop_after_binlog  position: 1"
+master-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t1`,`t2` /* generated by server */
+"engine: aria_notrans  crash point: ddl_log_drop_after_binlog  position: 2"
+"No crash!"
+master-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t1`,`t2` /* generated by server */
+"engine: innodb  crash point: ddl_log_drop_before_delete_table  position: 1"
+t2.TRG
+t2.frm
+t2.ibd
+t2_trg.TRN
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1` /* generated by ddl recovery */
+"engine: innodb  crash point: ddl_log_drop_before_delete_table  position: 2"
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2` /* generated by ddl recovery */
+"engine: innodb  crash point: ddl_log_drop_after_delete_table  position: 1"
+t2.TRG
+t2.frm
+t2.ibd
+t2_trg.TRN
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1` /* generated by ddl recovery */
+"engine: innodb  crash point: ddl_log_drop_after_delete_table  position: 2"
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2` /* generated by ddl recovery */
+"engine: innodb  crash point: ddl_log_drop_before_drop_trigger  position: 1"
+t2.TRG
+t2.frm
+t2.ibd
+t2_trg.TRN
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1` /* generated by ddl recovery */
+"engine: innodb  crash point: ddl_log_drop_before_drop_trigger  position: 2"
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2` /* generated by ddl recovery */
+"engine: innodb  crash point: ddl_log_drop_before_drop_trigger2  position: 1"
+t2.TRG
+t2.frm
+t2.ibd
+t2_trg.TRN
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1` /* generated by ddl recovery */
+"engine: innodb  crash point: ddl_log_drop_before_drop_trigger2  position: 2"
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2` /* generated by ddl recovery */
+"engine: innodb  crash point: ddl_log_drop_after_drop_trigger  position: 1"
+t2.TRG
+t2.frm
+t2.ibd
+t2_trg.TRN
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1` /* generated by ddl recovery */
+"engine: innodb  crash point: ddl_log_drop_after_drop_trigger  position: 2"
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2` /* generated by ddl recovery */
+"engine: innodb  crash point: ddl_log_drop_before_binlog  position: 1"
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2` /* generated by ddl recovery */
+"engine: innodb  crash point: ddl_log_drop_before_binlog  position: 2"
+"No crash!"
+master-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t1`,`t2` /* generated by server */
+"engine: innodb  crash point: ddl_log_drop_after_binlog  position: 1"
+master-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t1`,`t2` /* generated by server */
+"engine: innodb  crash point: ddl_log_drop_after_binlog  position: 2"
+"No crash!"
+master-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t1`,`t2` /* generated by server */
+"engine: csv  crash point: ddl_log_drop_before_delete_table  position: 1"
+t2.CSM
+t2.CSV
+t2.TRG
+t2.frm
+t2_trg.TRN
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1` /* generated by ddl recovery */
+"engine: csv  crash point: ddl_log_drop_before_delete_table  position: 2"
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2` /* generated by ddl recovery */
+"engine: csv  crash point: ddl_log_drop_after_delete_table  position: 1"
+t2.CSM
+t2.CSV
+t2.TRG
+t2.frm
+t2_trg.TRN
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1` /* generated by ddl recovery */
+"engine: csv  crash point: ddl_log_drop_after_delete_table  position: 2"
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2` /* generated by ddl recovery */
+"engine: csv  crash point: ddl_log_drop_before_drop_trigger  position: 1"
+t2.CSM
+t2.CSV
+t2.TRG
+t2.frm
+t2_trg.TRN
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1` /* generated by ddl recovery */
+"engine: csv  crash point: ddl_log_drop_before_drop_trigger  position: 2"
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2` /* generated by ddl recovery */
+"engine: csv  crash point: ddl_log_drop_before_drop_trigger2  position: 1"
+t2.CSM
+t2.CSV
+t2.TRG
+t2.frm
+t2_trg.TRN
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1` /* generated by ddl recovery */
+"engine: csv  crash point: ddl_log_drop_before_drop_trigger2  position: 2"
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2` /* generated by ddl recovery */
+"engine: csv  crash point: ddl_log_drop_after_drop_trigger  position: 1"
+t2.CSM
+t2.CSV
+t2.TRG
+t2.frm
+t2_trg.TRN
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1` /* generated by ddl recovery */
+"engine: csv  crash point: ddl_log_drop_after_drop_trigger  position: 2"
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2` /* generated by ddl recovery */
+"engine: csv  crash point: ddl_log_drop_before_binlog  position: 1"
+master-bin.000002	#	Query	#	#	DROP TABLE IF EXISTS `test`.`t1`,`test`.`t2` /* generated by ddl recovery */
+"engine: csv  crash point: ddl_log_drop_before_binlog  position: 2"
+"No crash!"
+master-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t1`,`t2` /* generated by server */
+"engine: csv  crash point: ddl_log_drop_after_binlog  position: 1"
+master-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t1`,`t2` /* generated by server */
+"engine: csv  crash point: ddl_log_drop_after_binlog  position: 2"
+"No crash!"
+master-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t1`,`t2` /* generated by server */
+Warnings:
+Note	1051	Unknown table 'test.t1,test.t2'
diff --git a/mysql-test/suite/atomic/drop_table.test b/mysql-test/suite/atomic/drop_table.test
new file mode 100644
index 00000000000..975e790f5d8
--- /dev/null
+++ b/mysql-test/suite/atomic/drop_table.test
@@ -0,0 +1,116 @@
+--source include/have_debug.inc
+--source include/have_innodb.inc
+--source include/have_csv.inc
+--source include/have_log_bin.inc
+--source include/not_valgrind.inc
+
+#
+# Testing of atomic drop 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=5;
+let $engines='myisam','aria','aria_notrans','innodb','csv';
+
+let $crash_count=7;
+let $crash_points='ddl_log_drop_before_delete_table', 'ddl_log_drop_after_delete_table', 'ddl_log_drop_before_drop_trigger', 'ddl_log_drop_before_drop_trigger2', 'ddl_log_drop_after_drop_trigger', 'ddl_log_drop_before_binlog', 'ddl_log_drop_after_binlog';
+
+# Number of drops in the tested statement
+let $drops=2;
+
+let $old_debug=`select @@debug_dbug`;
+
+let $e=0;
+let $keep_include_silent=1;
+let $grep_script=DROP TABLE;
+--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;
+  }
+
+  let $c=0;
+  while ($c < $crash_count)
+  {
+    inc $c;
+    let $crash=`select ELT($c, $crash_points)`;
+    let $r=0;
+    while ($r < $drops)
+    {
+      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;
+      insert into t1 values(1);
+      insert into t2 values(2);
+      flush tables;
+
+      delimiter |;
+      create trigger t1_trg before insert on t1 for each row
+      begin
+        if isnull(new.a) then
+          set new.a:= 1000;
+        end if;
+      end|
+      create trigger t2_trg before insert on t2 for each row
+      begin
+        if isnull(new.b) then
+          set new.b:= 2000;
+        end if;
+      end|
+      delimiter ;|
+
+      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 table t1,t2;
+      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!";
+      }
+      # Check which tables still exists
+      --list_files $MYSQLD_DATADIR/test t*
+
+      --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
+      drop table if exists t1,t2;
+      --enable_warnings
+    }
+  }
+}
+drop table if exists t1,t2;
+
+--enable_query_log
diff --git a/mysql-test/suite/atomic/drop_view.result b/mysql-test/suite/atomic/drop_view.result
new file mode 100644
index 00000000000..b2009fa41b6
--- /dev/null
+++ b/mysql-test/suite/atomic/drop_view.result
@@ -0,0 +1,20 @@
+"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 */
+"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 */
+"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 */
+"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 */
+"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 */
+"engine: aria  crash point: ddl_log_drop_before_binlog  position: 2"
+"No crash!"
+master-bin.000001	#	Query	#	#	use `test`; DROP VIEW v1,v2
+"engine: aria  crash point: ddl_log_drop_after_binlog  position: 1"
+master-bin.000001	#	Query	#	#	use `test`; DROP VIEW v1,v2
+"engine: aria  crash point: ddl_log_drop_after_binlog  position: 2"
+"No crash!"
+master-bin.000001	#	Query	#	#	use `test`; DROP VIEW v1,v2
diff --git a/mysql-test/suite/atomic/drop_view.test b/mysql-test/suite/atomic/drop_view.test
new file mode 100644
index 00000000000..f7cfdab963a
--- /dev/null
+++ b/mysql-test/suite/atomic/drop_view.test
@@ -0,0 +1,98 @@
+--source include/have_debug.inc
+--source include/have_log_bin.inc
+--source include/not_valgrind.inc
+
+#
+# Testing of atomic drop of view with crashes in a lot of different places
+#
+
+let $MYSQLD_DATADIR= `SELECT @@datadir`;
+let $engine_count=1;
+let $engines='aria';
+
+let $crash_count=4;
+let $crash_points='ddl_log_drop_before_delete_view', 'ddl_log_drop_after_delete_view', 'ddl_log_drop_before_binlog', 'ddl_log_drop_after_binlog';
+
+# Number of drops in the tested statement
+let $drops=2;
+
+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;
+  }
+
+  --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;
+  insert into t1 values(1);
+  insert into t2 values(2);
+  flush tables;
+
+  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 view v1 as select * from t1;
+      create view v2 as select * from t1;
+      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 VIEW v1,v2;
+      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!";
+      }
+      # Check which tables still exists
+      --list_files $MYSQLD_DATADIR/test v*
+      --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 views
+      --disable_warnings
+      drop view if exists v1,v2;
+      --enable_warnings
+    }
+  }
+  drop table t1,t2;
+}
+
+--enable_query_log
diff --git a/mysql-test/suite/binlog/r/binlog_stm_binlog.result b/mysql-test/suite/binlog/r/binlog_stm_binlog.result
index ad14a31381d..1ef3975176c 100644
--- a/mysql-test/suite/binlog/r/binlog_stm_binlog.result
+++ b/mysql-test/suite/binlog/r/binlog_stm_binlog.result
@@ -506,6 +506,7 @@ master-bin.000001	#	Query	#	#	COMMIT
 master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
 master-bin.000001	#	Query	#	#	use `mysql`; DELETE FROM db WHERE host='localhost' AND user='@#@'
 master-bin.000001	#	Query	#	#	COMMIT
+master-bin.000001	#	Rotate	#	#	master-bin.000002;pos=POS
 drop table t1,t2,t3,tt1;
 reset master;
 create table t1 (a int not null auto_increment, primary key (a)) engine=myisam;
diff --git a/mysql-test/suite/multi_source/reset_slave.test b/mysql-test/suite/multi_source/reset_slave.test
index 63a1f9c3490..d267ecb0cbf 100644
--- a/mysql-test/suite/multi_source/reset_slave.test
+++ b/mysql-test/suite/multi_source/reset_slave.test
@@ -39,9 +39,9 @@ stop slave 'master1';
 
 --let $datadir = `SELECT @@datadir`
 
-let read_master_log_pos=`select $binlog_start_pos + 590`;
-let relay_log_pos=`select 2*$binlog_start_pos + 634`;
-let relay_log_space=`select 3*$binlog_start_pos + 696`;
+let read_master_log_pos=`select $binlog_start_pos + 599`;
+let relay_log_pos=`select 2*$binlog_start_pos + 643`;
+let relay_log_space=`select 3*$binlog_start_pos + 705`;
 --replace_result $SERVER_MYPORT_1 MYPORT_1 $read_master_log_pos <read_master_log_pos> $relay_log_pos <relay_log_pos> $relay_log_space <relay_log_space>
 show slave 'master1' status;
 --list_files $datadir mysqld*
diff --git a/sql/ddl_log.cc b/sql/ddl_log.cc
index 1da6b5e52d5..c7e579f8a2e 100644
--- a/sql/ddl_log.cc
+++ b/sql/ddl_log.cc
@@ -26,6 +26,7 @@
 #include "sql_statistics.h"                     // rename_table_in_stats_tables
 #include "sql_view.h"                           // mysql_rename_view()
 #include "strfunc.h"                            // strconvert
+#include "sql_show.h"                           // append_identifier()
 #include <mysys_err.h>                          // EE_LINK
 
 
@@ -81,17 +82,19 @@ uchar ddl_log_file_magic[]=
 
 /* Action names for ddl_log_action_code */
 
-const char *ddl_log_action_name[DDL_LOG_LAST_ACTION]=
+const char *ddl_log_action_name[]=
 {
   "Unknown", "partitioning delete", "partitioning rename",
   "partitioning replace", "partitioning exchange",
-  "rename table", "rename view"
+  "rename table", "rename view",
+  "initialize drop table", "drop table",
+  "initialize drop view", "drop view"
 };
 
 /* Number of phases per entry */
 const uchar ddl_log_entry_phases[DDL_LOG_LAST_ACTION]=
 {
-  1, 1, 2, 3, 4, 1
+  0, 1, 1, 2, 3, 4, 1, 1, 3, 1, 1
 };
 
 
@@ -109,6 +112,7 @@ struct st_global_ddl_log
 };
 
 st_global_ddl_log global_ddl_log;
+String ddl_drop_query;                      // Used during startup recovery
 
 mysql_mutex_t LOCK_gdl;
 
@@ -285,6 +289,19 @@ static bool update_phase(uint entry_pos, uchar phase)
 }
 
 
+static bool update_next_entry_pos(uint entry_pos, uint next_entry)
+{
+  uchar buff[4];
+  DBUG_ENTER("update_next_entry_pos");
+
+  int4store(buff, next_entry);
+  DBUG_RETURN(mysql_file_pwrite(global_ddl_log.file_id, buff, sizeof(buff),
+                                global_ddl_log.io_size * entry_pos +
+                                DDL_LOG_NEXT_ENTRY_POS,
+                                MYF(MY_WME | MY_NABP)));
+}
+
+
 static bool update_xid(uint entry_pos, ulonglong xid)
 {
   uchar buff[8];
@@ -1130,6 +1147,122 @@ static int ddl_log_execute_action(THD *thd, MEM_ROOT *mem_root,
     (void) update_phase(entry_pos, DDL_LOG_FINAL_PHASE);
   }
   break;
+  case DDL_LOG_DROP_TABLE_INIT_ACTION:
+  {
+    LEX_CSTRING *comment= &ddl_log_entry->tmp_name;
+    ddl_drop_query.length(0);
+    ddl_drop_query.set_charset(system_charset_info);
+    ddl_drop_query.append(STRING_WITH_LEN("DROP TABLE IF EXISTS "));
+    if (comment->length)
+    {
+      ddl_drop_query.append(comment);
+      ddl_drop_query.append(' ');
+    }
+    /* We don't increment phase as we want to retry this in case of crash */
+    break;
+  }
+  case DDL_LOG_DROP_TABLE_ACTION:
+  {
+    LEX_CSTRING db, table, path;
+    db=     ddl_log_entry->db;
+    table=  ddl_log_entry->name;
+    /* Note that path is without .frm extension */
+    path=   ddl_log_entry->tmp_name;
+
+    switch (ddl_log_entry->phase) {
+    case DDL_DROP_PHASE_TABLE:
+      if (hton)
+      {
+        if ((error= hton->drop_table(hton, path.str)))
+        {
+          if (!non_existing_table_error(error))
+            break;
+          error= -1;
+        }
+      }
+      else
+        error= ha_delete_table_force(thd, path.str, &db, &table);
+      if (error <= 0)
+      {
+        /* Not found or already deleted. Delete .frm if it exists */
+        strxnmov(to_path, sizeof(to_path)-1, path.str, reg_ext, NullS);
+        mysql_file_delete(key_file_frm, to_path, MYF(MY_WME|MY_IGNORE_ENOENT));
+      }
+      if (ddl_log_increment_phase_no_lock(entry_pos))
+        break;
+      (void) ddl_log_sync_no_lock();
+      /* Fall through */
+    case DDL_DROP_PHASE_TRIGGER:
+      Table_triggers_list::drop_all_triggers(thd, &db, &table,
+                                             MYF(MY_WME | MY_IGNORE_ENOENT));
+      if (ddl_log_increment_phase_no_lock(entry_pos))
+        break;
+      (void) ddl_log_sync_no_lock();
+      /* Fall through */
+
+    case DDL_DROP_PHASE_BINLOG:
+      append_identifier(thd, &ddl_drop_query, &db);
+      ddl_drop_query.append('.');
+      append_identifier(thd, &ddl_drop_query, &table);
+      ddl_drop_query.append(',');
+      /* We don't increment phase as we want to retry this in case of crash */
+
+      if (!ddl_log_entry->next_entry && mysql_bin_log.is_open())
+      {
+        /* Last drop table. Write query to binlog */
+        LEX_CSTRING end_comment=
+          { STRING_WITH_LEN(" /* generated by ddl recovery */")};
+        ddl_drop_query.length(ddl_drop_query.length()-1);
+        ddl_drop_query.append(&end_comment);
+
+        mysql_mutex_unlock(&LOCK_gdl);
+        (void) thd->binlog_query(THD::STMT_QUERY_TYPE, ddl_drop_query.ptr(),
+                                 ddl_drop_query.length(), TRUE, FALSE,
+                                 FALSE, 0);
+        mysql_mutex_lock(&LOCK_gdl);
+      }
+      break;
+    }
+    break;
+  }
+  case DDL_LOG_DROP_VIEW_INIT_ACTION:
+  {
+    ddl_drop_query.length(0);
+    ddl_drop_query.set_charset(system_charset_info);
+    ddl_drop_query.append(STRING_WITH_LEN("DROP VIEW IF EXISTS "));
+    /* We don't increment phase as we want to retry this in case of crash */
+    break;
+  }
+  case DDL_LOG_DROP_VIEW_ACTION:
+  {
+    LEX_CSTRING db, table, path;
+    db=     ddl_log_entry->db;
+    table=  ddl_log_entry->name;
+    /* Note that for views path is WITH .frm extension */
+    path=   ddl_log_entry->tmp_name;
+
+    mysql_file_delete(key_file_frm, path.str, MYF(MY_WME|MY_IGNORE_ENOENT));
+    append_identifier(thd, &ddl_drop_query, &db);
+    ddl_drop_query.append('.');
+    append_identifier(thd, &ddl_drop_query, &table);
+    ddl_drop_query.append(',');
+
+    if (!ddl_log_entry->next_entry)
+    {
+      /* Last drop view. Write query to binlog */
+      LEX_CSTRING end_comment=
+        { STRING_WITH_LEN(" /* generated by ddl recovery */")};
+      ddl_drop_query.length(ddl_drop_query.length()-1);
+      ddl_drop_query.append(&end_comment);
+
+      mysql_mutex_unlock(&LOCK_gdl);
+      (void) thd->binlog_query(THD::STMT_QUERY_TYPE, ddl_drop_query.ptr(),
+                               ddl_drop_query.length(), TRUE, FALSE,
+                               FALSE, 0);
+      mysql_mutex_lock(&LOCK_gdl);
+    }
+    break;
+  }
   default:
     DBUG_ASSERT(0);
     break;
@@ -1298,9 +1431,14 @@ bool ddl_log_write_entry(DDL_LOG_ENTRY *ddl_log_entry,
   uchar *pos, *end;
   DBUG_ENTER("ddl_log_write_entry");
 
+  *active_entry= 0;
   mysql_mutex_assert_owner(&LOCK_gdl);
-  if (!global_ddl_log.open)
+  DBUG_ASSERT(global_ddl_log.open);
+  if (unlikely(!global_ddl_log.open))
+  {
+    my_error(ER_INTERNAL_ERROR, MYF(0), "ddl log not initialized");
     DBUG_RETURN(TRUE);
+  }
 
   ddl_log_entry->entry_type= DDL_LOG_ENTRY_CODE;
   set_global_from_ddl_log_entry(ddl_log_entry);
@@ -1383,7 +1521,7 @@ bool ddl_log_write_execute_entry(uint first_entry,
     if (ddl_log_get_free_entry(active_entry))
       DBUG_RETURN(TRUE);
     got_free_entry= TRUE;
-  }
+ }
   if (write_ddl_log_file_entry((*active_entry)->entry_pos))
   {
     if (got_free_entry)
@@ -1509,6 +1647,7 @@ bool ddl_log_close_binlogged_events(HASH *xids)
   {
     if (read_ddl_log_entry(i, &ddl_log_entry))
       break;                                    // Read error. Stop reading
+    DBUG_PRINT("xid",("xid: %llu", ddl_log_entry.xid));
     if (ddl_log_entry.entry_type == DDL_LOG_EXECUTE_CODE &&
         ddl_log_entry.xid != 0 &&
         my_hash_search(xids, (uchar*) &ddl_log_entry.xid,
@@ -1562,6 +1701,7 @@ int ddl_log_execute_recovery()
   thd->store_globals();
   thd->init();                                  // Needed for error messages
   thd->log_all_errors= (global_system_variables.log_warnings >= 3);
+  ddl_drop_query.free();
 
   thd->set_query(recover_query_string, strlen(recover_query_string));
 
@@ -1599,6 +1739,7 @@ int ddl_log_execute_recovery()
       count++;
     }
   }
+  ddl_drop_query.free();
   close_ddl_log();
   mysql_mutex_unlock(&LOCK_gdl);
   thd->reset_query();
@@ -1682,6 +1823,7 @@ void ddl_log_release_entries(DDL_LOG_STATE *ddl_log_state)
     next= log_entry->next_active_log_entry;
     ddl_log_release_memory_entry(log_entry);
   }
+  ddl_log_state->list= 0;
 
   if (ddl_log_state->execute_entry)
   {
@@ -1779,6 +1921,33 @@ bool ddl_log_update_xid(DDL_LOG_STATE *state, ulonglong xid)
 }
 
 
+/*
+  Write ddl_log_entry and write or update ddl_execute_entry
+*/
+
+static bool ddl_log_write(DDL_LOG_STATE *ddl_state,
+                          DDL_LOG_ENTRY *ddl_log_entry)
+{
+  int error;
+  DDL_LOG_MEMORY_ENTRY *log_entry;
+  DBUG_ENTER("ddl_log_write");
+
+  mysql_mutex_lock(&LOCK_gdl);
+  error= ((ddl_log_write_entry(ddl_log_entry, &log_entry)) ||
+          ddl_log_write_execute_entry(log_entry->entry_pos,
+                                      &ddl_state->execute_entry));
+  mysql_mutex_unlock(&LOCK_gdl);
+  if (error)
+  {
+    if (log_entry)
+      ddl_log_release_memory_entry(log_entry);
+    DBUG_RETURN(1);
+  }
+  add_log_entry(ddl_state, log_entry);
+  DBUG_RETURN(0);
+}
+
+
 /**
    Logging of rename table
 */
@@ -1791,13 +1960,10 @@ bool ddl_log_rename_table(THD *thd, DDL_LOG_STATE *ddl_state,
                           const LEX_CSTRING *new_alias)
 {
   DDL_LOG_ENTRY ddl_log_entry;
-  DDL_LOG_MEMORY_ENTRY *log_entry;
   DBUG_ENTER("ddl_log_rename_file");
 
   bzero(&ddl_log_entry, sizeof(ddl_log_entry));
 
-  mysql_mutex_lock(&LOCK_gdl);
-
   ddl_log_entry.action_type=  DDL_LOG_RENAME_TABLE_ACTION;
   ddl_log_entry.next_entry=   ddl_state->list ? ddl_state->list->entry_pos : 0;
   lex_string_set(&ddl_log_entry.handler_name,
@@ -1808,20 +1974,7 @@ bool ddl_log_rename_table(THD *thd, DDL_LOG_STATE *ddl_state,
   ddl_log_entry.from_name=    *const_cast<LEX_CSTRING*>(org_alias);
   ddl_log_entry.phase=        DDL_RENAME_PHASE_TABLE;
 
-  if (ddl_log_write_entry(&ddl_log_entry, &log_entry))
-    goto error;
-
-  if (ddl_log_write_execute_entry(log_entry->entry_pos,
-                                  &ddl_state->execute_entry))
-    goto error;
-
-  add_log_entry(ddl_state, log_entry);
-  mysql_mutex_unlock(&LOCK_gdl);
-  DBUG_RETURN(0);
-
-error:
-  mysql_mutex_unlock(&LOCK_gdl);
-  DBUG_RETURN(1);
+  DBUG_RETURN(ddl_log_write(ddl_state, &ddl_log_entry));
 }
 
 /*
@@ -1835,13 +1988,10 @@ bool ddl_log_rename_view(THD *thd, DDL_LOG_STATE *ddl_state,
                          const LEX_CSTRING *new_alias)
 {
   DDL_LOG_ENTRY ddl_log_entry;
-  DDL_LOG_MEMORY_ENTRY *log_entry;
   DBUG_ENTER("ddl_log_rename_file");
 
   bzero(&ddl_log_entry, sizeof(ddl_log_entry));
 
-  mysql_mutex_lock(&LOCK_gdl);
-
   ddl_log_entry.action_type=  DDL_LOG_RENAME_VIEW_ACTION;
   ddl_log_entry.next_entry=   ddl_state->list ? ddl_state->list->entry_pos : 0;
   ddl_log_entry.db=           *const_cast<LEX_CSTRING*>(new_db);
@@ -1849,18 +1999,117 @@ bool ddl_log_rename_view(THD *thd, DDL_LOG_STATE *ddl_state,
   ddl_log_entry.from_db=      *const_cast<LEX_CSTRING*>(org_db);
   ddl_log_entry.from_name=    *const_cast<LEX_CSTRING*>(org_alias);
 
+  DBUG_RETURN(ddl_log_write(ddl_state, &ddl_log_entry));
+}
+
+
+/**
+   Logging of DROP TABLE and DROP VIEW
+
+   Note that in contrast to rename, which are re-done in reverse order,
+   deletes are stored in a linked list according to delete order. This
+   is to ensure that the tables, for the query generated for binlog,
+   is in original delete order.
+*/
+
+static bool ddl_log_drop_init(THD *thd, DDL_LOG_STATE *ddl_state,
+                              ddl_log_action_code action_code,
+                              const LEX_CSTRING *comment)
+{
+  DDL_LOG_ENTRY ddl_log_entry;
+  DBUG_ENTER("ddl_log_drop_file");
+
+  bzero(&ddl_log_entry, sizeof(ddl_log_entry));
+
+  ddl_log_entry.action_type=  action_code;
+  ddl_log_entry.next_entry=   0;
+  ddl_log_entry.tmp_name=     *const_cast<LEX_CSTRING*>(comment);
+  ddl_log_entry.phase=        0;
+
+  DBUG_RETURN(ddl_log_write(ddl_state, &ddl_log_entry));
+}
+
+
+bool ddl_log_drop_table_init(THD *thd, DDL_LOG_STATE *ddl_state,
+                             const LEX_CSTRING *comment)
+{
+  return ddl_log_drop_init(thd, ddl_state, DDL_LOG_DROP_TABLE_INIT_ACTION,
+                           comment);
+}
+
+bool ddl_log_drop_view_init(THD *thd, DDL_LOG_STATE *ddl_state)
+{
+  LEX_CSTRING comment= {0,0};
+  return ddl_log_drop_init(thd, ddl_state, DDL_LOG_DROP_VIEW_INIT_ACTION,
+                           &comment);
+}
+
+static bool ddl_log_drop(THD *thd, DDL_LOG_STATE *ddl_state,
+                         ddl_log_action_code action_code,
+                         uint phase,
+                         handlerton *hton,
+                         const LEX_CSTRING *path,
+                         const LEX_CSTRING *db,
+                         const LEX_CSTRING *table)
+{
+  DDL_LOG_ENTRY ddl_log_entry;
+  DDL_LOG_MEMORY_ENTRY *log_entry;
+  DBUG_ENTER("ddl_log_drop");
+
+  DBUG_ASSERT(ddl_state->list);
+  bzero(&ddl_log_entry, sizeof(ddl_log_entry));
+
+  ddl_log_entry.action_type=  action_code;
+  if (hton)
+    lex_string_set(&ddl_log_entry.handler_name,
+                   ha_resolve_storage_engine_name(hton));
+  ddl_log_entry.db=           *const_cast<LEX_CSTRING*>(db);
+  ddl_log_entry.name=         *const_cast<LEX_CSTRING*>(table);
+  ddl_log_entry.tmp_name=     *const_cast<LEX_CSTRING*>(path);
+  ddl_log_entry.phase=        (uchar) phase;
+
+  mysql_mutex_lock(&LOCK_gdl);
   if (ddl_log_write_entry(&ddl_log_entry, &log_entry))
     goto error;
 
-  if (ddl_log_write_execute_entry(log_entry->entry_pos,
-                                  &ddl_state->execute_entry))
+  (void) ddl_log_sync_no_lock();
+  if (update_next_entry_pos(ddl_state->list->entry_pos,
+                            log_entry->entry_pos))
+  {
+    ddl_log_release_memory_entry(log_entry);
     goto error;
+  }
 
-  add_log_entry(ddl_state, log_entry);
   mysql_mutex_unlock(&LOCK_gdl);
+  add_log_entry(ddl_state, log_entry);
   DBUG_RETURN(0);
 
 error:
   mysql_mutex_unlock(&LOCK_gdl);
   DBUG_RETURN(1);
 }
+
+
+bool ddl_log_drop_table(THD *thd, DDL_LOG_STATE *ddl_state,
+                        handlerton *hton,
+                        const LEX_CSTRING *path,
+                        const LEX_CSTRING *db,
+                        const LEX_CSTRING *table)
+{
+  DBUG_ENTER("ddl_log_drop_table");
+  DBUG_RETURN(ddl_log_drop(thd, ddl_state,
+                           DDL_LOG_DROP_TABLE_ACTION, DDL_DROP_PHASE_TABLE,
+                           hton, path, db, table));
+}
+
+
+bool ddl_log_drop_view(THD *thd, DDL_LOG_STATE *ddl_state,
+                        const LEX_CSTRING *path,
+                        const LEX_CSTRING *db,
+                        const LEX_CSTRING *table)
+{
+  DBUG_ENTER("ddl_log_drop_view");
+  DBUG_RETURN(ddl_log_drop(thd, ddl_state,
+                           DDL_LOG_DROP_VIEW_ACTION, 0,
+                           (handlerton*) 0, path, db, table));
+}
diff --git a/sql/ddl_log.h b/sql/ddl_log.h
index ec91900609f..7dfcf2a0ba7 100644
--- a/sql/ddl_log.h
+++ b/sql/ddl_log.h
@@ -46,8 +46,8 @@ enum ddl_log_entry_code
 
 
 /*
-  When adding things below, also add an entry to ddl_log_entry_phases in
-  ddl_log.cc
+  When adding things below, also add an entry to ddl_log_action_names and
+  ddl_log_entry_phases in ddl_log.cc
 */
 
 enum ddl_log_action_code
@@ -77,7 +77,11 @@ enum ddl_log_action_code
   */
   DDL_LOG_RENAME_TABLE_ACTION= 5,
   DDL_LOG_RENAME_VIEW_ACTION= 6,
-  DDL_LOG_LAST_ACTION                           /* End marker */
+  DDL_LOG_DROP_TABLE_INIT_ACTION= 7,
+  DDL_LOG_DROP_TABLE_ACTION= 8,
+  DDL_LOG_DROP_VIEW_INIT_ACTION= 9,
+  DDL_LOG_DROP_VIEW_ACTION= 10,
+  DDL_LOG_LAST_ACTION                          /* End marker */
 };
 
 
@@ -97,6 +101,14 @@ enum enum_ddl_log_rename_table_phase {
   DDL_RENAME_PHASE_TABLE,
 };
 
+enum enum_ddl_log_drop_table_phase {
+  DDL_DROP_PHASE_TABLE=0,
+  DDL_DROP_PHASE_TRIGGER,
+  DDL_DROP_PHASE_BINLOG,
+  DDL_DROP_PHASE_RESET,  /* Reset found list of dropped tables */
+  DDL_DROP_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.
@@ -204,6 +216,17 @@ bool ddl_log_rename_view(THD *thd, DDL_LOG_STATE *ddl_state,
                          const LEX_CSTRING *org_alias,
                          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 *comment);
+bool ddl_log_drop_view_init(THD *thd, DDL_LOG_STATE *ddl_state);
+bool ddl_log_drop_table(THD *thd, DDL_LOG_STATE *ddl_state,
+                        handlerton *hton,
+                        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 *path,
+                        const LEX_CSTRING *db,
+                        const LEX_CSTRING *table);
 extern mysql_mutex_t LOCK_gdl;
 #endif /* DDL_LOG_INCLUDED */
diff --git a/sql/handler.cc b/sql/handler.cc
index 2b86e926b20..e7fa79b793f 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -566,7 +566,13 @@ static int hton_drop_table(handlerton *hton, const char *path)
   char tmp_path[FN_REFLEN];
   handler *file= get_new_handler(nullptr, current_thd->mem_root, hton);
   if (!file)
-    return ENOMEM;
+  {
+    /*
+      If file is not defined it means that the engine can't create a
+      handler if share is not set or we got an out of memory error
+    */
+    return my_errno == ENOMEM ? ENOMEM : ENOENT;
+  }
   path= get_canonical_filename(file, path, tmp_path);
   int error= file->delete_table(path);
   delete file;
diff --git a/sql/log.cc b/sql/log.cc
index 519dc3e63b3..f1fef44e05f 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -9982,6 +9982,7 @@ int TC_LOG::using_heuristic_recover()
 int TC_LOG_BINLOG::open(const char *opt_name)
 {
   int      error= 1;
+  DBUG_ENTER("TC_LOG_BINLOG::open");
 
   DBUG_ASSERT(total_ha_2pc > 1);
   DBUG_ASSERT(opt_name);
@@ -9991,7 +9992,7 @@ int TC_LOG_BINLOG::open(const char *opt_name)
   {
     /* There was a failure to open the index file, can't open the binlog */
     cleanup();
-    return 1;
+    DBUG_RETURN(1);
   }
 
   if (using_heuristic_recover())
@@ -10001,12 +10002,12 @@ int TC_LOG_BINLOG::open(const char *opt_name)
     open(opt_name, 0, 0, WRITE_CACHE, max_binlog_size, 0, TRUE);
     mysql_mutex_unlock(&LOCK_log);
     cleanup();
-    return 1;
+    DBUG_RETURN(1);
   }
 
   error= do_binlog_recovery(opt_name, true);
   binlog_state_recover_done= true;
-  return error;
+  DBUG_RETURN(error);
 }
 
 /** This is called on shutdown, after ha_panic. */
@@ -10536,6 +10537,7 @@ int TC_LOG_BINLOG::recover(LOG_INFO *linfo, const char *last_log_name,
         Query_log_event *query_ev= (Query_log_event*) ev;
         if (query_ev->xid)
         {
+          DBUG_PRINT("QQ", ("xid: %llu xid"));
           DBUG_ASSERT(sizeof(query_ev->xid) == sizeof(my_xid));
           uchar *x= (uchar *) memdup_root(&mem_root,
                                           (uchar*) &query_ev->xid,
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 36823f5c8a5..82692be8e5f 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -56,6 +56,7 @@
 #include "tztime.h"
 #include "sql_insert.h"                        // binlog_drop_table
 #include "ddl_log.h"
+#include "debug_sync.h"                         // debug_crash_here()
 #include <algorithm>
 
 #ifdef __WIN__
@@ -1139,9 +1140,11 @@ 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);
-  uint not_found_errors= 0;
+  DDL_LOG_STATE ddl_log_state;
+  const char *comment_start;
+  uint not_found_errors= 0, table_count= 0, non_temp_tables_count= 0;
   int error= 0;
-  int non_temp_tables_count= 0;
+  uint32 comment_len;
   bool trans_tmp_table_deleted= 0, non_trans_tmp_table_deleted= 0;
   bool non_tmp_table_deleted= 0;
   bool is_drop_tmp_if_exists_added= 0;
@@ -1154,6 +1157,8 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
   DBUG_ENTER("mysql_rm_table_no_locks");
 
   unknown_tables.length(0);
+  comment_len= get_comment(thd, if_exists ? 17:9, &comment_start);
+
   /*
     Prepares the drop statements that will be written into the binary
     log as follows:
@@ -1204,6 +1209,7 @@ 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)
   {
@@ -1213,6 +1219,7 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
     bool table_dropped= 0;
     const LEX_CSTRING db= table->db;
     const LEX_CSTRING table_name= table->table_name;
+    LEX_CSTRING cpath= {0,0};
     handlerton *hton= 0;
     Table_type table_type;
     size_t path_length= 0;
@@ -1347,6 +1354,8 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
       continue;
     }
 
+    lex_string_set3(&cpath, path, (size_t) (path_end - path));
+
     {
       char engine_buf[NAME_CHAR_LEN + 1];
       LEX_CSTRING engine= { engine_buf, 0 };
@@ -1363,6 +1372,17 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
 
     thd->replication_flags= 0;
     was_view= table_type == TABLE_TYPE_VIEW;
+
+    if (!table_count++)
+    {
+      LEX_CSTRING comment= {comment_start, (size_t) comment_len};
+      if (ddl_log_drop_table_init(thd, &ddl_log_state, &comment))
+      {
+        error= 1;
+        goto err;
+      }
+    }
+
     if ((table_type == TABLE_TYPE_UNKNOWN) || (was_view && !drop_view) ||
         (table_type != TABLE_TYPE_SEQUENCE && drop_sequence))
     {
@@ -1376,6 +1396,8 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
       was_table|= wrong_drop_sequence;
       error= table_type == TABLE_TYPE_UNKNOWN ? ENOENT : -1;
       tdc_remove_table(thd, db.str, table_name.str);
+      if (wrong_drop_sequence)
+        goto report_error;
     }
     else
     {
@@ -1411,7 +1433,17 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
         log_if_exists= 1;
 
       bool enoent_warning= !dont_log_query && !(hton && hton->discover_table);
-      error= ha_delete_table(thd, hton, path, &db, &table_name, enoent_warning);
+
+      if (ddl_log_drop_table(thd, &ddl_log_state, hton, &cpath, &db,
+                             &table_name))
+      {
+        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);
+      debug_crash_here("ddl_log_drop_after_delete_table");
 
       if (!error)
         table_dropped= 1;
@@ -1473,11 +1505,18 @@ int mysql_rm_table_no_locks(THD *thd, TABLE_LIST *tables, bool if_exists,
       scan all engines try to drop the table from there.
       This is to ensure we don't have any partial table files left.
     */
-    if (non_existing_table_error(error) && !wrong_drop_sequence)
+    if (non_existing_table_error(error))
     {
       int ferror= 0;
       DBUG_ASSERT(!was_view);
 
+      if (ddl_log_drop_table(thd, &ddl_log_state, hton, &cpath, &db,
+                             &table_name))
+      {
+        error= -1;
+        goto err;
+      }
+
       /* Remove extension for delete */
       *path_end= '\0';
       ferror= ha_delete_table_force(thd, path, &db, &table_name);
@@ -1509,13 +1548,19 @@ 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;
 
+    debug_crash_here("ddl_log_drop_before_drop_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))
     {
       if (Table_triggers_list::drop_all_triggers(thd, &db, &table_name,
                                                MYF(MY_WME | MY_IGNORE_ENOENT)))
         error= error ? error : -1;
     }
+    debug_crash_here("ddl_log_drop_after_drop_trigger");
 
+report_error:
     if (error)
     {
       StringBuffer<FN_REFLEN> tbl_name(system_charset_info);
@@ -1565,6 +1610,7 @@ 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 (!dont_log_query &&
         (!error || table_dropped || non_existing_table_error(error)))
@@ -1621,6 +1667,7 @@ err:
     query_cache_invalidate3(thd, tables, 0);
     if (!dont_log_query && mysql_bin_log.is_open())
     {
+      debug_crash_here("ddl_log_drop_before_binlog");
       if (non_trans_tmp_table_deleted)
       {
           /* Chop of the last comma */
@@ -1648,8 +1695,6 @@ err:
       if (non_tmp_table_deleted)
       {
         String built_query;
-        const char *comment_start;
-        uint32 comment_len;
 
         built_query.set_charset(thd->charset());
         built_query.append(STRING_WITH_LEN("DROP "));
@@ -1659,8 +1704,7 @@ err:
           built_query.append(STRING_WITH_LEN("IF EXISTS "));
 
         /* Preserve comment in original query */
-        if ((comment_len= get_comment(thd, if_exists ? 17:9,
-                                      &comment_start)))
+        if (comment_len)
         {
           built_query.append(comment_start, comment_len);
           built_query.append(' ');
@@ -1670,13 +1714,18 @@ err:
         normal_tables.chop();
         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);
         error |= (thd->binlog_query(THD::STMT_QUERY_TYPE,
                                     built_query.ptr(),
                                     built_query.length(),
                                     TRUE, FALSE, FALSE, 0) > 0);
+        thd->binlog_xid= 0;
       }
+      debug_crash_here("ddl_log_drop_after_binlog");
     }
   }
+  ddl_log_complete(&ddl_log_state);
 
   if (!drop_temporary)
   {
diff --git a/sql/sql_view.cc b/sql/sql_view.cc
index e6d726b30d7..e1d9ffb9cdc 100644
--- a/sql/sql_view.cc
+++ b/sql/sql_view.cc
@@ -36,6 +36,7 @@
 #include "sql_derived.h"
 #include "sql_cte.h"    // check_dependencies_in_with_clauses()
 #include "opt_trace.h"
+#include "ddl_log.h"
 #include "wsrep_mysqld.h"
 #include "debug_sync.h" // debug_crash_here
 
@@ -1817,9 +1818,12 @@ bool mysql_drop_view(THD *thd, TABLE_LIST *views, enum_drop_mode drop_mode)
   bool delete_error= FALSE, wrong_object_name= FALSE;
   bool some_views_deleted= FALSE;
   bool something_wrong= FALSE;
-  uint not_exists_count= 0;
+  uint not_exists_count= 0, view_count= 0;
+  DDL_LOG_STATE ddl_log_state;
   DBUG_ENTER("mysql_drop_view");
 
+  bzero(&ddl_log_state, sizeof(ddl_log_state));
+
   /*
     We can't allow dropping of unlocked view under LOCK TABLES since this
     might lead to deadlock. But since we can't really lock view with LOCK
@@ -1838,9 +1842,12 @@ bool mysql_drop_view(THD *thd, TABLE_LIST *views, enum_drop_mode drop_mode)
 
   for (view= views; view; view= view->next_local)
   {
+    LEX_CSTRING cpath;
     bool not_exist;
-    build_table_filename(path, sizeof(path) - 1,
-                         view->db.str, view->table_name.str, reg_ext, 0);
+    size_t length;
+    length= build_table_filename(path, sizeof(path) - 1,
+                                 view->db.str, view->table_name.str, reg_ext, 0);
+    lex_string_set3(&cpath, path, length);
 
     if ((not_exist= my_access(path, F_OK)) || !dd_frm_is_view(thd, path))
     {
@@ -1861,8 +1868,18 @@ bool mysql_drop_view(THD *thd, TABLE_LIST *views, enum_drop_mode drop_mode)
         not_exists_count++;
       continue;
     }
+    if (!view_count++)
+    {
+      if (ddl_log_drop_view_init(thd, &ddl_log_state))
+        DBUG_RETURN(TRUE);
+    }
+    if (ddl_log_drop_view(thd, &ddl_log_state, &cpath, &view->db,
+                          &view->table_name))
+      DBUG_RETURN(TRUE);
+    debug_crash_here("ddl_log_drop_before_delete_view");
     if (unlikely(mysql_file_delete(key_file_frm, path, MYF(MY_WME))))
       delete_error= TRUE;
+    debug_crash_here("ddl_log_drop_after_delete_view");
 
     some_views_deleted= TRUE;
 
@@ -1890,10 +1907,16 @@ bool mysql_drop_view(THD *thd, TABLE_LIST *views, enum_drop_mode drop_mode)
     /* if something goes wrong, bin-log with possible error code,
        otherwise bin-log with error code cleared.
      */
+    debug_crash_here("ddl_log_drop_before_binlog");
+    thd->binlog_xid= thd->query_id;
+    ddl_log_update_xid(&ddl_log_state, thd->binlog_xid);
     if (unlikely(write_bin_log(thd, !something_wrong, thd->query(),
                                thd->query_length())))
       something_wrong= 1;
+    thd->binlog_xid= 0;
+    debug_crash_here("ddl_log_drop_after_binlog");
   }
+  ddl_log_complete(&ddl_log_state);
 
   if (unlikely(something_wrong))
   {