diff --git a/mysql-test/suite/atomic/README.txt b/mysql-test/suite/atomic/README.txt
new file mode 100644
index 00000000000..b1453d8a4f8
--- /dev/null
+++ b/mysql-test/suite/atomic/README.txt
@@ -0,0 +1,19 @@
+To debug a the ddl_recovery code in a failing ddl_recovery test one could do
+the following:
+
+- Add # before --exec echo "restart" ...
+- Force $e (engine), $c (crash point) and $r (crash position) to the values
+  where things goes wrong. See comments in alter_table.test for how to do this.
+- start mariadbd in a debugger
+
+run the following in the debugger
+(Replace 'atomic.create_trigger' with the failing test case)
+
+#break ha_recover
+#break MYSQL_BIN_LOG::recover
+#break MYSQL_BIN_LOG::open
+
+break ddl_log_close_binlogged_events
+break ddl_log_execute_action
+break ddl_log_execute_recovery
+run --datadir=/my/maria-10.6/mysql-test/var/log/atomic.create_trigger/mysqld.1/data --log-basename=master --log-bin-index=mysqld-bin.index --debug --log-bin
diff --git a/mysql-test/suite/atomic/create_trigger.result b/mysql-test/suite/atomic/create_trigger.result
new file mode 100644
index 00000000000..ec37377231e
--- /dev/null
+++ b/mysql-test/suite/atomic/create_trigger.result
@@ -0,0 +1,323 @@
+"engine: aria  crash point: ddl_log_create_before_create_trigger  position: 1"
+"engine: aria  crash point: ddl_log_create_before_create_trigger  position: 2"
+t1.TRG
+t1_trg.TRN
+Trigger	sql_mode	SQL Original Statement	character_set_client	collation_connection	Database Collation	Created
+t1_trg	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` TRIGGER t1_trg before insert on t1 for each row
+begin
+if isnull(new.a) then
+set new.a:= 1000;
+end if;
+end	latin1	latin1_swedish_ci	latin1_swedish_ci	#
+master-bin.000001	#	Query	#	#	use `test`; CREATE DEFINER=`root`@`localhost` TRIGGER t1_trg before insert on t1 for each row
+begin
+if isnull(new.a) then
+set new.a:= 1000;
+end if;
+end
+"engine: aria  crash point: ddl_log_create_before_create_trigger  position: 3"
+t1.TRG
+t1_trg.TRN
+t2_trg.TRN
+Trigger	sql_mode	SQL Original Statement	character_set_client	collation_connection	Database Collation	Created
+t1_trg	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` TRIGGER t1_trg before insert on t1 for each row
+begin
+if isnull(new.a) then
+set new.a:= 1000;
+end if;
+end	latin1	latin1_swedish_ci	latin1_swedish_ci	#
+Trigger	sql_mode	SQL Original Statement	character_set_client	collation_connection	Database Collation	Created
+t2_trg	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` TRIGGER t2_trg before insert on t1 for each row
+begin
+if isnull(new.b) then
+set new.b:= 2000;
+end if;
+end	latin1	latin1_swedish_ci	latin1_swedish_ci	#
+master-bin.000001	#	Query	#	#	use `test`; CREATE DEFINER=`root`@`localhost` TRIGGER t1_trg before insert on t1 for each row
+begin
+if isnull(new.a) then
+set new.a:= 1000;
+end if;
+end
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE DEFINER=`root`@`localhost` TRIGGER t2_trg before insert on t1 for each row
+begin
+if isnull(new.b) then
+set new.b:= 2000;
+end if;
+end
+"engine: aria  crash point: ddl_log_create_after_create_trigger  position: 1"
+"engine: aria  crash point: ddl_log_create_after_create_trigger  position: 2"
+t1.TRG
+t1_trg.TRN
+Trigger	sql_mode	SQL Original Statement	character_set_client	collation_connection	Database Collation	Created
+t1_trg	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` TRIGGER t1_trg before insert on t1 for each row
+begin
+if isnull(new.a) then
+set new.a:= 1000;
+end if;
+end	latin1	latin1_swedish_ci	latin1_swedish_ci	#
+master-bin.000001	#	Query	#	#	use `test`; CREATE DEFINER=`root`@`localhost` TRIGGER t1_trg before insert on t1 for each row
+begin
+if isnull(new.a) then
+set new.a:= 1000;
+end if;
+end
+"engine: aria  crash point: ddl_log_create_after_create_trigger  position: 3"
+t1.TRG
+t1_trg.TRN
+t2_trg.TRN
+Trigger	sql_mode	SQL Original Statement	character_set_client	collation_connection	Database Collation	Created
+t1_trg	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` TRIGGER t1_trg before insert on t1 for each row
+begin
+if isnull(new.a) then
+set new.a:= 1000;
+end if;
+end	latin1	latin1_swedish_ci	latin1_swedish_ci	#
+Trigger	sql_mode	SQL Original Statement	character_set_client	collation_connection	Database Collation	Created
+t2_trg	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` TRIGGER t2_trg before insert on t1 for each row
+begin
+if isnull(new.b) then
+set new.b:= 2000;
+end if;
+end	latin1	latin1_swedish_ci	latin1_swedish_ci	#
+master-bin.000001	#	Query	#	#	use `test`; CREATE DEFINER=`root`@`localhost` TRIGGER t1_trg before insert on t1 for each row
+begin
+if isnull(new.a) then
+set new.a:= 1000;
+end if;
+end
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE DEFINER=`root`@`localhost` TRIGGER t2_trg before insert on t1 for each row
+begin
+if isnull(new.b) then
+set new.b:= 2000;
+end if;
+end
+"engine: aria  crash point: definition_file_after_create  position: 1"
+"engine: aria  crash point: definition_file_after_create  position: 2"
+"engine: aria  crash point: definition_file_after_create  position: 3"
+t1.TRG
+t1_trg.TRN
+Trigger	sql_mode	SQL Original Statement	character_set_client	collation_connection	Database Collation	Created
+t1_trg	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` TRIGGER t1_trg before insert on t1 for each row
+begin
+if isnull(new.a) then
+set new.a:= 1000;
+end if;
+end	latin1	latin1_swedish_ci	latin1_swedish_ci	#
+master-bin.000001	#	Query	#	#	use `test`; CREATE DEFINER=`root`@`localhost` TRIGGER t1_trg before insert on t1 for each row
+begin
+if isnull(new.a) then
+set new.a:= 1000;
+end if;
+end
+"engine: aria  crash point: ddl_log_drop_before_binlog  position: 1"
+"engine: aria  crash point: ddl_log_drop_before_binlog  position: 2"
+t1.TRG
+t1_trg.TRN
+Trigger	sql_mode	SQL Original Statement	character_set_client	collation_connection	Database Collation	Created
+t1_trg	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` TRIGGER t1_trg before insert on t1 for each row
+begin
+if isnull(new.a) then
+set new.a:= 1000;
+end if;
+end	latin1	latin1_swedish_ci	latin1_swedish_ci	#
+master-bin.000001	#	Query	#	#	use `test`; CREATE DEFINER=`root`@`localhost` TRIGGER t1_trg before insert on t1 for each row
+begin
+if isnull(new.a) then
+set new.a:= 1000;
+end if;
+end
+"engine: aria  crash point: ddl_log_drop_before_binlog  position: 3"
+t1.TRG
+t1_trg.TRN
+t2_trg.TRN
+Trigger	sql_mode	SQL Original Statement	character_set_client	collation_connection	Database Collation	Created
+t1_trg	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` TRIGGER t1_trg before insert on t1 for each row
+begin
+if isnull(new.a) then
+set new.a:= 1000;
+end if;
+end	latin1	latin1_swedish_ci	latin1_swedish_ci	#
+Trigger	sql_mode	SQL Original Statement	character_set_client	collation_connection	Database Collation	Created
+t2_trg	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` TRIGGER t2_trg before insert on t1 for each row
+begin
+if isnull(new.b) then
+set new.b:= 2000;
+end if;
+end	latin1	latin1_swedish_ci	latin1_swedish_ci	#
+master-bin.000001	#	Query	#	#	use `test`; CREATE DEFINER=`root`@`localhost` TRIGGER t1_trg before insert on t1 for each row
+begin
+if isnull(new.a) then
+set new.a:= 1000;
+end if;
+end
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE DEFINER=`root`@`localhost` TRIGGER t2_trg before insert on t1 for each row
+begin
+if isnull(new.b) then
+set new.b:= 2000;
+end if;
+end
+"engine: aria  crash point: ddl_log_drop_after_binlog  position: 1"
+t1.TRG
+t1_trg.TRN
+Trigger	sql_mode	SQL Original Statement	character_set_client	collation_connection	Database Collation	Created
+t1_trg	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` TRIGGER t1_trg before insert on t1 for each row
+begin
+if isnull(new.a) then
+set new.a:= 1000;
+end if;
+end	latin1	latin1_swedish_ci	latin1_swedish_ci	#
+master-bin.000001	#	Query	#	#	use `test`; CREATE DEFINER=`root`@`localhost` TRIGGER t1_trg before insert on t1 for each row
+begin
+if isnull(new.a) then
+set new.a:= 1000;
+end if;
+end
+"engine: aria  crash point: ddl_log_drop_after_binlog  position: 2"
+t1.TRG
+t1_trg.TRN
+t2_trg.TRN
+Trigger	sql_mode	SQL Original Statement	character_set_client	collation_connection	Database Collation	Created
+t1_trg	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` TRIGGER t1_trg before insert on t1 for each row
+begin
+if isnull(new.a) then
+set new.a:= 1000;
+end if;
+end	latin1	latin1_swedish_ci	latin1_swedish_ci	#
+Trigger	sql_mode	SQL Original Statement	character_set_client	collation_connection	Database Collation	Created
+t2_trg	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` TRIGGER t2_trg before insert on t1 for each row
+begin
+if isnull(new.b) then
+set new.b:= 2000;
+end if;
+end	latin1	latin1_swedish_ci	latin1_swedish_ci	#
+master-bin.000001	#	Query	#	#	use `test`; CREATE DEFINER=`root`@`localhost` TRIGGER t1_trg before insert on t1 for each row
+begin
+if isnull(new.a) then
+set new.a:= 1000;
+end if;
+end
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE DEFINER=`root`@`localhost` TRIGGER t2_trg before insert on t1 for each row
+begin
+if isnull(new.b) then
+set new.b:= 2000;
+end if;
+end
+"engine: aria  crash point: ddl_log_drop_after_binlog  position: 3"
+t1.TRG
+t1_trg.TRN
+t2_trg.TRN
+Trigger	sql_mode	SQL Original Statement	character_set_client	collation_connection	Database Collation	Created
+t1_trg	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` TRIGGER t1_trg before insert on t1 for each row
+begin
+if isnull(new.a) then
+set new.a:= 1000;
+end if;
+end	latin1	latin1_swedish_ci	latin1_swedish_ci	#
+Trigger	sql_mode	SQL Original Statement	character_set_client	collation_connection	Database Collation	Created
+t2_trg	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` TRIGGER t2_trg before insert on t1 for each row
+begin
+if isnull(new.b) then
+set new.b:= 3000;
+end if;
+end	latin1	latin1_swedish_ci	latin1_swedish_ci	#
+master-bin.000001	#	Query	#	#	use `test`; CREATE DEFINER=`root`@`localhost` TRIGGER t1_trg before insert on t1 for each row
+begin
+if isnull(new.a) then
+set new.a:= 1000;
+end if;
+end
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE DEFINER=`root`@`localhost` TRIGGER t2_trg before insert on t1 for each row
+begin
+if isnull(new.b) then
+set new.b:= 2000;
+end if;
+end
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE DEFINER=`root`@`localhost` TRIGGER t2_trg before insert on t1 for each row
+begin
+if isnull(new.b) then
+set new.b:= 3000;
+end if;
+end
+"engine: aria  crash point: ddl_log_drop_before_delete_tmp  position: 1"
+t1.TRG
+t1_trg.TRN
+Trigger	sql_mode	SQL Original Statement	character_set_client	collation_connection	Database Collation	Created
+t1_trg	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` TRIGGER t1_trg before insert on t1 for each row
+begin
+if isnull(new.a) then
+set new.a:= 1000;
+end if;
+end	latin1	latin1_swedish_ci	latin1_swedish_ci	#
+master-bin.000001	#	Query	#	#	use `test`; CREATE DEFINER=`root`@`localhost` TRIGGER t1_trg before insert on t1 for each row
+begin
+if isnull(new.a) then
+set new.a:= 1000;
+end if;
+end
+"engine: aria  crash point: ddl_log_drop_before_delete_tmp  position: 2"
+t1.TRG
+t1_trg.TRN
+t2_trg.TRN
+Trigger	sql_mode	SQL Original Statement	character_set_client	collation_connection	Database Collation	Created
+t1_trg	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` TRIGGER t1_trg before insert on t1 for each row
+begin
+if isnull(new.a) then
+set new.a:= 1000;
+end if;
+end	latin1	latin1_swedish_ci	latin1_swedish_ci	#
+Trigger	sql_mode	SQL Original Statement	character_set_client	collation_connection	Database Collation	Created
+t2_trg	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` TRIGGER t2_trg before insert on t1 for each row
+begin
+if isnull(new.b) then
+set new.b:= 2000;
+end if;
+end	latin1	latin1_swedish_ci	latin1_swedish_ci	#
+master-bin.000001	#	Query	#	#	use `test`; CREATE DEFINER=`root`@`localhost` TRIGGER t1_trg before insert on t1 for each row
+begin
+if isnull(new.a) then
+set new.a:= 1000;
+end if;
+end
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE DEFINER=`root`@`localhost` TRIGGER t2_trg before insert on t1 for each row
+begin
+if isnull(new.b) then
+set new.b:= 2000;
+end if;
+end
+"engine: aria  crash point: ddl_log_drop_before_delete_tmp  position: 3"
+t1.TRG
+t1_trg.TRN
+t2_trg.TRN
+Trigger	sql_mode	SQL Original Statement	character_set_client	collation_connection	Database Collation	Created
+t1_trg	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` TRIGGER t1_trg before insert on t1 for each row
+begin
+if isnull(new.a) then
+set new.a:= 1000;
+end if;
+end	latin1	latin1_swedish_ci	latin1_swedish_ci	#
+Trigger	sql_mode	SQL Original Statement	character_set_client	collation_connection	Database Collation	Created
+t2_trg	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` TRIGGER t2_trg before insert on t1 for each row
+begin
+if isnull(new.b) then
+set new.b:= 3000;
+end if;
+end	latin1	latin1_swedish_ci	latin1_swedish_ci	#
+master-bin.000001	#	Query	#	#	use `test`; CREATE DEFINER=`root`@`localhost` TRIGGER t1_trg before insert on t1 for each row
+begin
+if isnull(new.a) then
+set new.a:= 1000;
+end if;
+end
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE DEFINER=`root`@`localhost` TRIGGER t2_trg before insert on t1 for each row
+begin
+if isnull(new.b) then
+set new.b:= 2000;
+end if;
+end
+master-bin.000001	#	Query	#	#	use `test`; CREATE OR REPLACE DEFINER=`root`@`localhost` TRIGGER t2_trg before insert on t1 for each row
+begin
+if isnull(new.b) then
+set new.b:= 3000;
+end if;
+end
diff --git a/mysql-test/suite/atomic/create_trigger.test b/mysql-test/suite/atomic/create_trigger.test
new file mode 100644
index 00000000000..2072fdb444a
--- /dev/null
+++ b/mysql-test/suite/atomic/create_trigger.test
@@ -0,0 +1,131 @@
+--source include/have_debug.inc
+--source include/have_log_bin.inc
+--source include/not_valgrind.inc
+
+#
+# Testing of atomic CREATE TRIGGER with crashes in a lot of different places
+#
+
+let $MYSQLD_DATADIR= `SELECT @@datadir`;
+
+let $engine_count=1;
+let $engines='aria';
+
+let $crash_count=6;
+let $crash_points='ddl_log_create_before_create_trigger', 'ddl_log_create_after_create_trigger', 'definition_file_after_create', 'ddl_log_drop_before_binlog', 'ddl_log_drop_after_binlog','ddl_log_drop_before_delete_tmp';
+
+let $old_debug=`select @@debug_dbug`;
+
+let $e=0;
+let $keep_include_silent=1;
+let $grep_script=CREATE.*TRIGGER;
+let $drops=3;
+--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, b int not null) $extra_option;
+  insert into t1 values(1,1);
+  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;
+
+      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;
+      delimiter |;
+      --error 0,2013
+      CREATE TRIGGER t1_trg before insert on t1 for each row
+      begin
+        if isnull(new.a) then
+          set new.a:= 1000;
+        end if;
+      end|
+      delimiter ;|
+      let $error=$errno;
+      if ($error == 0)
+      {
+        delimiter |;
+        --error 0,2013
+        CREATE OR REPLACE TRIGGER t2_trg before insert on t1 for each row
+        begin
+          if isnull(new.b) then
+            set new.b:= 2000;
+          end if;
+        end|
+        delimiter ;|
+        let $error=$errno;
+      }
+      if ($error == 0)
+      {
+        delimiter |;
+        --error 0,2013
+        CREATE OR REPLACE TRIGGER t2_trg before insert on t1 for each row
+        begin
+          if isnull(new.b) then
+            set new.b:= 3000;
+          end if;
+        end|
+        delimiter ;|
+        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 *TR*
+      --list_files $MYSQLD_DATADIR/test *sql*
+
+      --replace_column 7 #
+      --error 0,ER_TRG_DOES_NOT_EXIST
+      SHOW CREATE TRIGGER t1_trg;
+      --replace_column 7 #
+      --error 0,ER_TRG_DOES_NOT_EXIST
+      SHOW CREATE TRIGGER t2_trg;
+
+      --let $binlog_file=master-bin.000001
+      --source include/show_binlog_events.inc
+      --disable_warnings
+      drop trigger if exists t1_trg;
+      drop trigger if exists t2_trg;
+      --enable_warnings
+    }
+  }
+}
+
+drop table t1;
+
+--enable_query_log
diff --git a/mysql-test/suite/atomic/create_trigger2.result b/mysql-test/suite/atomic/create_trigger2.result
new file mode 100644
index 00000000000..d239ac3b839
--- /dev/null
+++ b/mysql-test/suite/atomic/create_trigger2.result
@@ -0,0 +1,94 @@
+"position: 1"
+"position: 2"
+"position: 3"
+t1.TRG
+t1_trg.TRN
+Trigger	sql_mode	SQL Original Statement	character_set_client	collation_connection	Database Collation	Created
+t1_trg	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` trigger t1_trg before insert on t1 for each row
+begin
+if isnull(new.a) then
+set new.a:= 1000;
+end if;
+end	latin1	latin1_swedish_ci	latin1_swedish_ci	#
+"position: 4"
+t1.TRG
+t1_trg.TRN
+Trigger	sql_mode	SQL Original Statement	character_set_client	collation_connection	Database Collation	Created
+t1_trg	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` trigger t1_trg before insert on t1 for each row
+begin
+if isnull(new.a) then
+set new.a:= 1000;
+end if;
+end	latin1	latin1_swedish_ci	latin1_swedish_ci	#
+"position: 5"
+t1.TRG
+t1_trg.TRN
+t2_trg.TRN
+Trigger	sql_mode	SQL Original Statement	character_set_client	collation_connection	Database Collation	Created
+t1_trg	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` trigger t1_trg before insert on t1 for each row
+begin
+if isnull(new.a) then
+set new.a:= 1000;
+end if;
+end	latin1	latin1_swedish_ci	latin1_swedish_ci	#
+Trigger	sql_mode	SQL Original Statement	character_set_client	collation_connection	Database Collation	Created
+t2_trg	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` trigger t2_trg before insert on t1 for each row
+begin
+if isnull(new.b) then
+set new.b:= 2000;
+end if;
+end	latin1	latin1_swedish_ci	latin1_swedish_ci	#
+"position: 6"
+t1.TRG
+t1_trg.TRN
+t2_trg.TRN
+Trigger	sql_mode	SQL Original Statement	character_set_client	collation_connection	Database Collation	Created
+t1_trg	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` trigger t1_trg before insert on t1 for each row
+begin
+if isnull(new.a) then
+set new.a:= 1000;
+end if;
+end	latin1	latin1_swedish_ci	latin1_swedish_ci	#
+Trigger	sql_mode	SQL Original Statement	character_set_client	collation_connection	Database Collation	Created
+t2_trg	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` trigger t2_trg before insert on t1 for each row
+begin
+if isnull(new.b) then
+set new.b:= 2000;
+end if;
+end	latin1	latin1_swedish_ci	latin1_swedish_ci	#
+"position: 7"
+t1.TRG
+t1_trg.TRN
+t2_trg.TRN
+Trigger	sql_mode	SQL Original Statement	character_set_client	collation_connection	Database Collation	Created
+t1_trg	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` trigger t1_trg before insert on t1 for each row
+begin
+if isnull(new.a) then
+set new.a:= 1000;
+end if;
+end	latin1	latin1_swedish_ci	latin1_swedish_ci	#
+Trigger	sql_mode	SQL Original Statement	character_set_client	collation_connection	Database Collation	Created
+t2_trg	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` trigger t2_trg before insert on t1 for each row
+begin
+if isnull(new.b) then
+set new.b:= 2000;
+end if;
+end	latin1	latin1_swedish_ci	latin1_swedish_ci	#
+"position: 8"
+t1.TRG
+t1_trg.TRN
+t2_trg.TRN
+Trigger	sql_mode	SQL Original Statement	character_set_client	collation_connection	Database Collation	Created
+t1_trg	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` trigger t1_trg before insert on t1 for each row
+begin
+if isnull(new.a) then
+set new.a:= 1000;
+end if;
+end	latin1	latin1_swedish_ci	latin1_swedish_ci	#
+Trigger	sql_mode	SQL Original Statement	character_set_client	collation_connection	Database Collation	Created
+t2_trg	STRICT_TRANS_TABLES,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION	CREATE DEFINER=`root`@`localhost` trigger t2_trg before insert on t1 for each row
+begin
+if isnull(new.b) then
+set new.b:= 3000;
+end if;
+end	latin1	latin1_swedish_ci	latin1_swedish_ci	#
diff --git a/mysql-test/suite/atomic/create_trigger2.test b/mysql-test/suite/atomic/create_trigger2.test
new file mode 100644
index 00000000000..b694207bbfc
--- /dev/null
+++ b/mysql-test/suite/atomic/create_trigger2.test
@@ -0,0 +1,89 @@
+--source include/have_debug.inc
+
+#
+# Testing of atomic CREATE TRIGGER when write fails in create_definition_file
+#
+
+let $MYSQLD_DATADIR= `SELECT @@datadir`;
+
+let $engine_count=1;
+let $engines='aria';
+let $old_debug=`select @@debug_dbug`;
+
+let $e=0;
+--disable_query_log
+
+create table t1 (a int not null, b int not null);
+insert into t1 values(1,1);
+flush tables;
+
+# sql_create_definition_file is called twice per CREATE TRIGGER and 1 more
+# in case we drop an existing trigger, so we need to test 3*2 +1 failures
+# and also when there is no failures (= 8)
+let $try_count=8;
+
+let $r=0;
+while ($r < $try_count)
+{
+  inc $r;
+
+  echo "position: $r";
+  --eval set @@debug_dbug="+d,definition_file_simulate_write_error",@debug_error_counter=$r;
+  let $errno=0;
+  delimiter |;
+  --error 0,3
+  create trigger t1_trg before insert on t1 for each row
+  begin
+    if isnull(new.a) then
+      set new.a:= 1000;
+    end if;
+  end|
+  delimiter ;|
+  let $error=$errno;
+  if ($error == 0)
+  {
+    delimiter |;
+    --error 0,3
+    create or replace trigger t2_trg before insert on t1 for each row
+    begin
+      if isnull(new.b) then
+        set new.b:= 2000;
+      end if;
+    end|
+    delimiter ;|
+    let $error=$errno;
+  }
+  if ($error == 0)
+  {
+    delimiter |;
+    --error 0,3
+    create or replace trigger t2_trg before insert on t1 for each row
+    begin
+      if isnull(new.b) then
+        set new.b:= 3000;
+      end if;
+    end|
+    delimiter ;|
+    let $error=$errno;
+  }
+  --eval set @@debug_dbug="$old_debug"
+
+  # Check which tables still exists
+  --list_files $MYSQLD_DATADIR/test *TR*
+  --list_files $MYSQLD_DATADIR/test *sql*
+
+  --replace_column 7 #
+  --error 0,ER_TRG_DOES_NOT_EXIST
+  SHOW CREATE TRIGGER t1_trg;
+  --replace_column 7 #
+  --error 0,ER_TRG_DOES_NOT_EXIST
+  SHOW CREATE TRIGGER t2_trg;
+  --disable_warnings
+  drop trigger if exists t1_trg;
+  drop trigger if exists t2_trg;
+  --enable_warnings
+}
+
+drop table t1;
+
+--enable_query_log
diff --git a/sql/ddl_log.cc b/sql/ddl_log.cc
index 49fdb861f8d..7eae243771b 100644
--- a/sql/ddl_log.cc
+++ b/sql/ddl_log.cc
@@ -90,7 +90,7 @@ const char *ddl_log_action_name[DDL_LOG_LAST_ACTION]=
   "rename table", "rename view",
   "initialize drop table", "drop table",
   "drop view", "drop trigger", "drop db", "create table", "create view",
-  "delete tmp file",
+  "delete tmp file", "create trigger",
 };
 
 /* Number of phases per entry */
@@ -100,7 +100,7 @@ const uchar ddl_log_entry_phases[DDL_LOG_LAST_ACTION]=
   (uchar) EXCH_PHASE_END, (uchar) DDL_RENAME_PHASE_END, 1, 1,
   (uchar) DDL_DROP_PHASE_END, 1, 1,
   (uchar) DDL_DROP_DB_PHASE_END, (uchar) DDL_CREATE_TABLE_PHASE_END,
-  (uchar) DDL_CREATE_VIEW_PHASE_END, 0,
+  (uchar) DDL_CREATE_VIEW_PHASE_END, 0, (uchar) DDL_CREATE_TRIGGER_PHASE_END,
 };
 
 
@@ -1637,7 +1637,96 @@ static int ddl_log_execute_action(THD *thd, MEM_ROOT *mem_root,
     (void) update_phase(entry_pos, DDL_LOG_FINAL_PHASE);
     break;
   }
-  break;
+  case DDL_LOG_CREATE_TRIGGER_ACTION:
+  {
+    LEX_CSTRING db, table, trigger;
+    db=      ddl_log_entry->db;
+    table=   ddl_log_entry->name;
+    trigger= ddl_log_entry->tmp_name;
+
+    /* Delete backup .TRG (trigger file) if it exists */
+    (void) build_filename_and_delete_tmp_file(to_path, sizeof(to_path) - 1,
+                                              &db, &table,
+                                              TRG_EXT,
+                                              key_file_fileparser);
+    (void) build_filename_and_delete_tmp_file(to_path, sizeof(to_path) - 1,
+                                              &db, &trigger,
+                                              TRN_EXT,
+                                              key_file_fileparser);
+    switch (ddl_log_entry->phase) {
+    case DDL_CREATE_TRIGGER_PHASE_DELETE_COPY:
+    {
+      size_t length;
+      /* Delete copy of .TRN and .TRG files */
+      length= build_table_filename(to_path, sizeof(to_path) - 1,
+                                   db.str, table.str, TRG_EXT, 0);
+      to_path[length]= '-';
+      to_path[length+1]= 0;
+      mysql_file_delete(key_file_fileparser, to_path,
+                        MYF(MY_WME|MY_IGNORE_ENOENT));
+
+      length= build_table_filename(to_path, sizeof(to_path) - 1,
+                                   db.str, trigger.str, TRN_EXT, 0);
+      to_path[length]= '-';
+      to_path[length+1]= 0;
+      mysql_file_delete(key_file_fileparser, to_path,
+                        MYF(MY_WME|MY_IGNORE_ENOENT));
+    }
+    /* Nothing else to do */
+    (void) update_phase(entry_pos, DDL_LOG_FINAL_PHASE);
+    break;
+    case DDL_CREATE_TRIGGER_PHASE_OLD_COPIED:
+    {
+      LEX_CSTRING path= {to_path, 0};
+      size_t length;
+      /* Restore old version if the .TRN and .TRG files */
+      length= build_table_filename(to_path, sizeof(to_path) - 1,
+                                   db.str, table.str, TRG_EXT, 0);
+      to_path[length]='-';
+      to_path[length+1]= 0;
+      path.length= length+1;
+      /* an old TRN file only exist in the case if REPLACE was used */
+      if (!access(to_path, F_OK))
+        sql_restore_definition_file(&path);
+
+      length= build_table_filename(to_path, sizeof(to_path) - 1,
+                                   db.str, trigger.str, TRN_EXT, 0);
+      to_path[length]='-';
+      to_path[length+1]= 0;
+      path.length= length+1;
+      if (!access(to_path, F_OK))
+        sql_restore_definition_file(&path);
+      else
+      {
+        /*
+          There was originally no .TRN for this trigger.
+          Delete the newly created one.
+        */
+        to_path[length]= 0;
+        mysql_file_delete(key_file_fileparser, to_path,
+                          MYF(MY_WME|MY_IGNORE_ENOENT));
+      }
+      (void) update_phase(entry_pos, DDL_LOG_FINAL_PHASE);
+      break;
+    }
+    case DDL_CREATE_TRIGGER_PHASE_NO_OLD_TRIGGER:
+    {
+      /* No old trigger existed. We can just delete the .TRN and .TRG files */
+      build_table_filename(to_path, sizeof(to_path) - 1,
+                           db.str, table.str, TRG_EXT, 0);
+      mysql_file_delete(key_file_fileparser, to_path,
+                        MYF(MY_WME|MY_IGNORE_ENOENT));
+      build_table_filename(to_path, sizeof(to_path) - 1,
+                           db.str, trigger.str, TRN_EXT, 0);
+      mysql_file_delete(key_file_fileparser, to_path,
+                        MYF(MY_WME|MY_IGNORE_ENOENT));
+      (void) update_phase(entry_pos, DDL_LOG_FINAL_PHASE);
+      break;
+    }
+    }
+    break;
+  }
+  }
   default:
     DBUG_ASSERT(0);
     break;
@@ -2633,3 +2722,25 @@ bool ddl_log_delete_tmp_file(THD *thd, DDL_LOG_STATE *ddl_state,
     ddl_log_entry.unique_id= depending_state->execute_entry->entry_pos;
   DBUG_RETURN(ddl_log_write(ddl_state, &ddl_log_entry));
 }
+
+
+/**
+   Log CREATE TRIGGER
+*/
+
+bool ddl_log_create_trigger(THD *thd, DDL_LOG_STATE *ddl_state,
+                            const LEX_CSTRING *db, const LEX_CSTRING *table,
+                            const LEX_CSTRING *trigger_name,
+                            enum_ddl_log_create_trigger_phase phase)
+{
+  DDL_LOG_ENTRY ddl_log_entry;
+  DBUG_ENTER("ddl_log_create_view");
+
+  bzero(&ddl_log_entry, sizeof(ddl_log_entry));
+  ddl_log_entry.action_type=  DDL_LOG_CREATE_TRIGGER_ACTION;
+  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*>(trigger_name);
+  ddl_log_entry.phase=        (uchar) phase;
+  DBUG_RETURN(ddl_log_write(ddl_state, &ddl_log_entry));
+}
diff --git a/sql/ddl_log.h b/sql/ddl_log.h
index 38c0129f4bf..3aabde31733 100644
--- a/sql/ddl_log.h
+++ b/sql/ddl_log.h
@@ -85,6 +85,7 @@ enum ddl_log_action_code
   DDL_LOG_CREATE_TABLE_ACTION=12,
   DDL_LOG_CREATE_VIEW_ACTION=13,
   DDL_LOG_DELETE_TMP_FILE_ACTION=14,
+  DDL_LOG_CREATE_TRIGGER_ACTION=15,
   DDL_LOG_LAST_ACTION                          /* End marker */
 };
 
@@ -134,6 +135,13 @@ enum enum_ddl_log_create_view_phase {
   DDL_CREATE_VIEW_PHASE_END
 };
 
+enum enum_ddl_log_create_trigger_phase {
+  DDL_CREATE_TRIGGER_PHASE_NO_OLD_TRIGGER,
+  DDL_CREATE_TRIGGER_PHASE_DELETE_COPY,
+  DDL_CREATE_TRIGGER_PHASE_OLD_COPIED,
+  DDL_CREATE_TRIGGER_PHASE_END
+};
+
 
 /*
   Setting ddl_log_entry.phase to this has the same effect as setting
@@ -282,5 +290,9 @@ bool ddl_log_create_view(THD *thd, DDL_LOG_STATE *ddl_state,
 bool ddl_log_delete_tmp_file(THD *thd, DDL_LOG_STATE *ddl_state,
                              const LEX_CSTRING *path,
                              DDL_LOG_STATE *depending_state);
+bool ddl_log_create_trigger(THD *thd, DDL_LOG_STATE *ddl_state,
+                            const LEX_CSTRING *db, const LEX_CSTRING *table,
+                            const LEX_CSTRING *trigger_name,
+                            enum_ddl_log_create_trigger_phase phase);
 extern mysql_mutex_t LOCK_gdl;
 #endif /* DDL_LOG_INCLUDED */
diff --git a/sql/parse_file.cc b/sql/parse_file.cc
index aaf36970702..8ea00e96f33 100644
--- a/sql/parse_file.cc
+++ b/sql/parse_file.cc
@@ -347,6 +347,48 @@ err_w_file:
   DBUG_RETURN(TRUE);
 }
 
+
+/*
+  Make a copy of a definition file with '-' added to the name
+
+  @param org_name   Original file name
+  @param new_name   Pointer to a buff of FN_REFLEN. Will be updated to name of
+                    backup file
+  @return 0 ok
+  @return 1 error
+*/
+
+int sql_backup_definition_file(const LEX_CSTRING *org_name,
+                               LEX_CSTRING *new_name)
+{
+  char *new_name_buff= (char*) new_name->str;
+  new_name->length= org_name->length+1;
+
+  memcpy(new_name_buff, org_name->str, org_name->length+1);
+  new_name_buff[org_name->length]= '-';
+  new_name_buff[org_name->length+1]= 0;
+  return my_copy(org_name->str, new_name->str, MYF(MY_WME));
+}
+
+/*
+  Restore copy of a definition file
+
+  @param org_name   Name of backup file (ending with '-' or '~')
+
+  @return 0 ok
+  @return 1 error
+*/
+
+int sql_restore_definition_file(const LEX_CSTRING *name)
+{
+  char new_name[FN_REFLEN+1];
+  memcpy(new_name, name->str, name->length-1);
+  new_name[name->length-1]= 0;
+  return mysql_file_rename(key_file_fileparser, name->str, new_name,
+                           MYF(MY_WME));
+}
+
+
 /**
   Renames a frm file (including backups) in same schema.
 
diff --git a/sql/parse_file.h b/sql/parse_file.h
index cbd41d16cbc..cd26ffec91a 100644
--- a/sql/parse_file.h
+++ b/sql/parse_file.h
@@ -96,6 +96,10 @@ my_bool rename_in_schema_file(THD *thd,
                               const char *schema, const char *old_name,
                               const char *new_db, const char *new_name);
 
+int sql_backup_definition_file(const LEX_CSTRING *org_name,
+                                   LEX_CSTRING *new_name);
+int sql_restore_definition_file(const LEX_CSTRING *name);
+
 class File_parser: public Sql_alloc
 {
   char *start, *end;
diff --git a/sql/sql_trigger.cc b/sql/sql_trigger.cc
index f2e8bf49dc0..35d615d325a 100644
--- a/sql/sql_trigger.cc
+++ b/sql/sql_trigger.cc
@@ -402,12 +402,13 @@ bool mysql_create_or_drop_trigger(THD *thd, TABLE_LIST *tables, bool create)
   MDL_ticket *mdl_ticket= NULL;
   MDL_request mdl_request_for_trn;
   Query_tables_list backup;
-  DDL_LOG_STATE ddl_log_state;
+  DDL_LOG_STATE ddl_log_state, ddl_log_state_tmp_file;
   DBUG_ENTER("mysql_create_or_drop_trigger");
 
   /* Charset of the buffer for statement must be system one. */
   stmt_query.set_charset(system_charset_info);
   bzero(&ddl_log_state, sizeof(ddl_log_state));
+  bzero(&ddl_log_state_tmp_file, sizeof(ddl_log_state_tmp_file));
 
   /*
     QQ: This function could be merged in mysql_alter_table() function
@@ -608,7 +609,9 @@ bool mysql_create_or_drop_trigger(THD *thd, TABLE_LIST *tables, bool create)
 #endif /* WITH_WSREP */
 
   if (create)
-    result= table->triggers->create_trigger(thd, tables, &stmt_query);
+    result= table->triggers->create_trigger(thd, tables, &stmt_query,
+                                            &ddl_log_state,
+                                            &ddl_log_state_tmp_file);
   else
   {
     result= table->triggers->drop_trigger(thd, tables,
@@ -644,7 +647,9 @@ end:
     debug_crash_here("ddl_log_drop_after_binlog");
   }
   ddl_log_complete(&ddl_log_state);
-
+  debug_crash_here("ddl_log_drop_before_delete_tmp");
+  /* delete any created log files */
+  ddl_log_revert(thd, &ddl_log_state_tmp_file);
   /*
     If we are under LOCK TABLES we should restore original state of
     meta-data locks. Otherwise all locks will be released along
@@ -797,18 +802,23 @@ static void build_trig_stmt_query(THD *thd, TABLE_LIST *tables,
 */
 
 bool Table_triggers_list::create_trigger(THD *thd, TABLE_LIST *tables,
-                                         String *stmt_query)
+                                         String *stmt_query,
+                                         DDL_LOG_STATE *ddl_log_state,
+                                         DDL_LOG_STATE *ddl_log_state_tmp_file)
 {
   LEX *lex= thd->lex;
   TABLE *table= tables->table;
   char file_buff[FN_REFLEN], trigname_buff[FN_REFLEN];
-  LEX_CSTRING file, trigname_file;
+  char backup_file_buff[FN_REFLEN];
   char trg_definer_holder[USER_HOST_BUFF_SIZE];
+  LEX_CSTRING backup_name= { backup_file_buff, 0 };
+  LEX_CSTRING file, trigname_file;
   Item_trigger_field *trg_field;
   struct st_trigname trigname;
   String trigger_definition;
   Trigger *trigger= 0;
-  bool trigger_dropped= 0;
+  int error;
+  bool trigger_exists;
   DBUG_ENTER("create_trigger");
 
   if (check_for_broken_triggers())
@@ -882,11 +892,34 @@ bool Table_triggers_list::create_trigger(THD *thd, TABLE_LIST *tables,
   trigname_file.str= trigname_buff;
 
   /* Use the filesystem to enforce trigger namespace constraints. */
-  if (!access(trigname_buff, F_OK))
+  trigger_exists= !access(trigname_file.str, F_OK);
+
+  ddl_log_create_trigger(thd, ddl_log_state, &tables->db, &tables->table_name,
+                         &lex->spname->m_name,
+                         trigger_exists || table->triggers->count ?
+                         DDL_CREATE_TRIGGER_PHASE_DELETE_COPY :
+                         DDL_CREATE_TRIGGER_PHASE_NO_OLD_TRIGGER);
+
+  /* Make a backup of the .TRG file that we can restore in case of crash */
+  if (table->triggers->count &&
+      (sql_backup_definition_file(&file, &backup_name) ||
+       ddl_log_delete_tmp_file(thd, ddl_log_state_tmp_file, &backup_name,
+                               ddl_log_state)))
+    DBUG_RETURN(true);
+
+  if (trigger_exists)
   {
     if (lex->create_info.or_replace())
     {
       LEX_CSTRING *sp_name= &thd->lex->spname->m_name; // alias
+
+      /* Make a backup of the .TRN file that we can restore in case of crash */
+      if (sql_backup_definition_file(&trigname_file, &backup_name) ||
+          ddl_log_delete_tmp_file(thd, ddl_log_state_tmp_file, &backup_name,
+                                  ddl_log_state))
+        DBUG_RETURN(true);
+      ddl_log_update_phase(ddl_log_state, DDL_CREATE_TRIGGER_PHASE_OLD_COPIED);
+
       /*
         The following can fail if the trigger is for another table or
         there exists a .TRN file but there was no trigger for it in
@@ -922,6 +955,11 @@ bool Table_triggers_list::create_trigger(THD *thd, TABLE_LIST *tables,
       DBUG_RETURN(true);
     }
   }
+  else
+  {
+    if (table->triggers->count)
+      ddl_log_update_phase(ddl_log_state, DDL_CREATE_TRIGGER_PHASE_OLD_COPIED);
+  }
 
   trigname.trigger_table= tables->table_name;
 
@@ -930,12 +968,16 @@ bool Table_triggers_list::create_trigger(THD *thd, TABLE_LIST *tables,
     going to access lex->sphead later in build_trig_stmt_query()
   */
   if (!(trigger= new (&table->mem_root) Trigger(this, 0)))
-    goto err_without_cleanup;
+    goto err;
 
   /* Create trigger_name.TRN file to ensure trigger name is unique */
   if (sql_create_definition_file(NULL, &trigname_file, &trigname_file_type,
                                  (uchar*)&trigname, trigname_file_parameters))
-    goto err_without_cleanup;
+  {
+    delete trigger;
+    trigger= 0;
+    goto err;
+  }
 
   /* Populate the trigger object */
 
@@ -957,6 +999,7 @@ bool Table_triggers_list::create_trigger(THD *thd, TABLE_LIST *tables,
   trigger->client_cs_name= thd->charset()->cs_name;
   trigger->connection_cl_name= thd->variables.collation_connection->coll_name;
   trigger->db_cl_name= get_default_db_collation(thd, tables->db.str)->coll_name;
+  trigger->name= lex->spname->m_name;
 
   /* Add trigger in it's correct place */
   add_trigger(lex->trg_chistics.event,
@@ -967,30 +1010,31 @@ bool Table_triggers_list::create_trigger(THD *thd, TABLE_LIST *tables,
 
   /* Create trigger definition file .TRG */
   if (unlikely(create_lists_needed_for_files(thd->mem_root)))
-    goto err_with_cleanup;
+    goto err;
 
-  if (!sql_create_definition_file(NULL, &file, &triggers_file_type,
-                                  (uchar*)this, triggers_file_parameters))
+  debug_crash_here("ddl_log_create_before_create_trigger");
+  error= sql_create_definition_file(NULL, &file, &triggers_file_type,
+                                    (uchar*)this, triggers_file_parameters);
+  debug_crash_here("ddl_log_create_after_create_trigger");
+
+  if (!error)
     DBUG_RETURN(false);
 
-err_with_cleanup:
-  /* Delete .TRN file */
-  mysql_file_delete(key_file_trn, trigname_buff, MYF(MY_WME));
-
-err_without_cleanup:
-  delete trigger;                               // Safety, not critical
-
-  if (trigger_dropped)
+err:
+  DBUG_PRINT("error",("create trigger failed"));
+  if (trigger)
   {
-    String drop_trg_query;
-    drop_trg_query.append(STRING_WITH_LEN("DROP TRIGGER /* generated by failed CREATE TRIGGER */ "));
-    drop_trg_query.append(&lex->spname->m_name);
-    /*
-      We dropped an existing trigger and was not able to recreate it because
-      of an internal error. Ensure it's also dropped on the slave.
-    */
-    write_bin_log(thd, FALSE, drop_trg_query.ptr(), drop_trg_query.length());
+    my_debug_put_break_here();
+    /* Delete trigger from trigger list if it exists */
+    find_trigger(&trigger->name, 1);
+    /* Free trigger memory */
+    delete trigger;
   }
+
+  /* Recover the old .TRN and .TRG files & delete backup files */
+  ddl_log_revert(thd, ddl_log_state);
+  /* All backup files are now deleted */
+  ddl_log_complete(ddl_log_state_tmp_file);
   DBUG_RETURN(true);
 }
 
@@ -1113,15 +1157,17 @@ bool Table_triggers_list::save_trigger_file(THD *thd, const LEX_CSTRING *db,
 {
   char file_buff[FN_REFLEN];
   LEX_CSTRING file;
+  DBUG_ENTER("Table_triggers_list::save_trigger_file");
 
   if (create_lists_needed_for_files(thd->mem_root))
-    return true;
+    DBUG_RETURN(true);
 
   file.length= build_table_filename(file_buff, FN_REFLEN - 1, db->str, table_name->str,
                                     TRG_EXT, 0);
   file.str= file_buff;
-  return sql_create_definition_file(NULL, &file, &triggers_file_type,
-                                    (uchar*) this, triggers_file_parameters);
+  DBUG_RETURN(sql_create_definition_file(NULL, &file, &triggers_file_type,
+                                         (uchar*) this,
+                                         triggers_file_parameters));
 }
 
 
@@ -1191,6 +1237,7 @@ bool Table_triggers_list::drop_trigger(THD *thd, TABLE_LIST *tables,
 {
   char path[FN_REFLEN];
   Trigger *trigger;
+  DBUG_ENTER("Table_triggers_list::drop_trigger");
 
   if (stmt_query)
     stmt_query->set(thd->query(), thd->query_length(), stmt_query->charset());
@@ -1200,9 +1247,11 @@ bool Table_triggers_list::drop_trigger(THD *thd, TABLE_LIST *tables,
   {
     my_message(ER_TRG_DOES_NOT_EXIST, ER_THD(thd, ER_TRG_DOES_NOT_EXIST),
                MYF(0));
-    return 1;
+    DBUG_RETURN(1);
   }
+  delete trigger;
 
+  if (ddl_log_state)
   {
     LEX_CSTRING query= {0,0};
     if (stmt_query)
@@ -1210,41 +1259,40 @@ bool Table_triggers_list::drop_trigger(THD *thd, TABLE_LIST *tables,
       /* This code is executed in case of DROP TRIGGER */
       lex_string_set3(&query, thd->query(), thd->query_length());
     }
-
-    if (ddl_log_state)
-      if (ddl_log_drop_trigger(thd, ddl_log_state,
-                               &tables->db, &tables->table_name,
-                               sp_name, &query))
-        return 1;
+    if (ddl_log_drop_trigger(thd, ddl_log_state,
+                             &tables->db, &tables->table_name,
+                             sp_name, &query))
+      goto err;
   }
   debug_crash_here("ddl_log_drop_before_drop_trigger");
 
   if (!count)                                   // If no more triggers
   {
     /*
-      TODO: Probably instead of removing .TRG file we should move
-      to archive directory but this should be done as part of
-      parse_file.cc functionality (because we will need it
-      elsewhere).
+      It is safe to remove the trigger file.  If something goes wrong during
+      drop or create ddl_log recovery will ensure that all related
+      trigger files are deleted or the original ones are restored.
     */
     if (rm_trigger_file(path, &tables->db, &tables->table_name, MYF(MY_WME)))
-      return 1;
+      goto err;
   }
   else
   {
     if (save_trigger_file(thd, &tables->db, &tables->table_name))
-      return 1;
+      goto err;
   }
 
   debug_crash_here("ddl_log_drop_before_drop_trn");
 
   if (rm_trigname_file(path, &tables->db, sp_name, MYF(MY_WME)))
-    return 1;
+    goto err;
 
   debug_crash_here("ddl_log_drop_after_drop_trigger");
 
-  delete trigger;
-  return 0;
+  DBUG_RETURN(0);
+
+err:
+  DBUG_RETURN(1);
 }
 
 
diff --git a/sql/sql_trigger.h b/sql/sql_trigger.h
index f0eae8c348f..98e527a50c0 100644
--- a/sql/sql_trigger.h
+++ b/sql/sql_trigger.h
@@ -220,7 +220,9 @@ public:
   }
   ~Table_triggers_list();
 
-  bool create_trigger(THD *thd, TABLE_LIST *table, String *stmt_query);
+  bool create_trigger(THD *thd, TABLE_LIST *table, String *stmt_query,
+                      DDL_LOG_STATE *ddl_log_state,
+                      DDL_LOG_STATE *ddl_log_state_tmp_file);
   bool drop_trigger(THD *thd, TABLE_LIST *table,
                     LEX_CSTRING *sp_name,
                     String *stmt_query, DDL_LOG_STATE *ddl_log_state);
diff --git a/sql/sql_view.cc b/sql/sql_view.cc
index 59b8962a84a..c72d31d7472 100644
--- a/sql/sql_view.cc
+++ b/sql/sql_view.cc
@@ -1178,11 +1178,8 @@ loop_out:
 
   if (old_view_exists)
   {
-    /* Make a backup that we can restore in case of crash */
-    memcpy(backup_file_name, path.str, path.length);
-    backup_file_name[path.length]='-';
-    backup_file_name[path.length+1]= 0;
-    if (my_copy(path.str, backup_file_name, MYF(MY_WME)))
+    LEX_CSTRING backup_name= { backup_file_name, 0 };
+    if (sql_backup_definition_file(&path, &backup_name))
     {
       error= 1;
       goto err;