MDEV-24746 Atomic CREATE TRIGGER

The purpose of this task is to ensure that CREATE TRIGGER is atomic

When a trigger is created, we first create a trigger_name.TRN file and then
create or update the table_name.TRG files.
This is done by creating .TRN~ and .TRG~ files and replacing (or creating)
the result files.

The new logic is

- Log CREATE TRIGGER to DDL log, with a marker if old trigger existsted
- If old .TRN or .TRG files exists, make backup copies of these
- Create the new .TRN and .TRG files as before
- Remove the backups

Crash recovery
- If query has been logged to binary log:
  - delete any left over backup files
- else
   - Delete any old .TRN~ or .TRG~ files
   - If there was orignally some triggers (old .TRG file existed)
      - If we crashed before creating all backup files
         - Delete existing backup files
      - else
         - Restore backup files
      - end
   - Delete .TRN and .TRG file (as there was no triggers before

One benefit of the new code is that CREATE OR REPLACE TRIGGER is now
totally atomic even if there existed an old trigger: Either the old
trigger will be replaced or the old one will be left untouched.

Other things:
- If sql_create_definition_file() would fail, there could be memory leaks
  in CREATE TRIGGER, DROP TRIGGER or CREATE OR REPLACE TRIGGER.  This
  is now fixed.

12 files changed, 928 insertions, 56 deletions

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;
-
-  if (!sql_create_definition_file(NULL, &file, &triggers_file_type,
-                                  (uchar*)this, triggers_file_parameters))
-    DBUG_RETURN(false);
+    goto err;
 
-err_with_cleanup:
-  /* Delete .TRN file */
-  mysql_file_delete(key_file_trn, trigname_buff, MYF(MY_WME));
+  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");
 
-err_without_cleanup:
-  delete trigger;                               // Safety, not critical
+  if (!error)
+    DBUG_RETURN(false);
 
-  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;