MDEV-25910: Aim to make all InnoDB DDL durable

Before any committed DDL operation returns control to the caller,
we must ensure that it will have been durably committed before the
ddl_log state may be changed, no matter if
innodb_flush_log_at_trx_commit=0 is being used.

Operations that would involve deleting files were already safe,
because the durable write of the FILE_DELETE record that would precede
the file deletion would also have made the commit durable.

Furthermore, when we clean up ADD INDEX stubs that were left behind
by a previous ha_innobase::commit_inplace_alter_table(commit=false)
when MDL could not be acquired, we will use the same interface as
for dropping indexes. This safety measure should be dead code,
because ADD FULLTEXT INDEX is not supported online, and dropping indexes
only involves file deletion for FULLTEXT INDEX.

mysql-test/suite/atomic/alter_table.result

@@ -2134,7 +2134,7 @@ t1	CREATE TABLE `t1` (
   KEY `a` (`a`)
 ) ENGINE=InnoDB DEFAULT CHARSET=latin1 COMMENT='new'
 count(*)
-0
+2
 master-bin.000002	#	Query	#	#	use `test`; ALTER TABLE t1 ADD COLUMN c INT, ALGORITHM=copy, COMMENT "new"
 crash point: ddl_log_alter_after_rename_to_backup
 t1.frm

storage/innobase/handler/ha_innodb.cc

@@ -1559,6 +1559,8 @@ static void innodb_drop_database(handlerton*, char *path)
     mem_heap_free(heap);
     for (pfs_os_file_t detached : to_close)
       os_file_close(detached);
+    /* Any changes must be persisted before we return. */
+    log_write_up_to(mtr.commit_lsn(), true);
   }
 
   my_free(namebuf);
@@ -13033,6 +13035,9 @@ ha_innobase::create(
 		row_mysql_unlock_data_dictionary(trx);
 		for (pfs_os_file_t d : deleted) os_file_close(d);
 		error = info.create_table_update_dict();
+		if (!(info.flags2() & DICT_TF2_TEMPORARY)) {
+			log_write_up_to(trx->commit_lsn, true);
+		}
 	}
 
 	if (own_trx) {
@@ -13379,10 +13384,11 @@ int ha_innobase::delete_table(const char *name)
   std::vector<pfs_os_file_t> deleted;
   trx->commit(deleted);
   row_mysql_unlock_data_dictionary(trx);
-  if (trx != parent_trx)
-    trx->free();
   for (pfs_os_file_t d : deleted)
     os_file_close(d);
+  log_write_up_to(trx->commit_lsn, true);
+  if (trx != parent_trx)
+    trx->free();
   if (fts)
     purge_sys.resume_FTS();
   DBUG_RETURN(0);
@@ -13595,6 +13601,7 @@ int ha_innobase::truncate()
 
 		err = create(name, table, &info,
 			     dict_table_is_file_per_table(ib_table), trx);
+		/* On success, create() durably committed trx. */
 		if (fts) {
 			purge_sys.resume_FTS();
 		}
@@ -13690,6 +13697,9 @@ ha_innobase::rename_table(
 	}
 
 	row_mysql_unlock_data_dictionary(trx);
+	if (error == DB_SUCCESS) {
+		log_write_up_to(trx->commit_lsn, true);
+	}
 	trx->free();
 
 	if (error == DB_DUPLICATE_KEY) {
@@ -15308,7 +15318,9 @@ ha_innobase::extra(
 		break;
 	case HA_EXTRA_END_ALTER_COPY:
 		m_prebuilt->table->skip_alter_undo = 0;
-		log_write_up_to(LSN_MAX, true);
+		if (!m_prebuilt->table->is_temporary()) {
+			log_write_up_to(LSN_MAX, true);
+		}
 		break;
 	default:/* Do nothing */
 		;

storage/innobase/handler/handler0alter.cc

@@ -4090,6 +4090,26 @@ online_retry_drop_indexes_low(
 	}
 }
 
+/** After commit, unlock the data dictionary and close any deleted files.
+@param deleted  handles of deleted files
+@param trx      committed transaction */
+static void unlock_and_close_files(const std::vector<pfs_os_file_t> &deleted,
+                                   trx_t *trx)
+{
+  row_mysql_unlock_data_dictionary(trx);
+  for (pfs_os_file_t d : deleted)
+    os_file_close(d);
+  log_write_up_to(trx->commit_lsn, true);
+}
+
+/** Commit a DDL transaction and unlink any deleted files. */
+static void commit_unlock_and_unlink(trx_t *trx)
+{
+  std::vector<pfs_os_file_t> deleted;
+  trx->commit(deleted);
+  unlock_and_close_files(deleted, trx);
+}
+
 /********************************************************************//**
 Drop any indexes that we were not able to free previously due to
 open table handles. */
@@ -4107,8 +4127,7 @@ online_retry_drop_indexes(
 
 		row_mysql_lock_data_dictionary(trx);
 		online_retry_drop_indexes_low(table, trx);
-		trx_commit_for_mysql(trx);
-		row_mysql_unlock_data_dictionary(trx);
+		commit_unlock_and_unlink(trx);
 		trx->free();
 	}
 
@@ -4139,7 +4158,16 @@ online_retry_drop_indexes_with_trx(
 		trx_start_for_ddl(trx);
 
 		online_retry_drop_indexes_low(table, trx);
-		trx_commit_for_mysql(trx);
+		std::vector<pfs_os_file_t> deleted;
+		trx->commit(deleted);
+		/* FIXME: We are holding the data dictionary latch here
+		while waiting for the files to be actually deleted.
+		However, we should never have any deleted files here,
+		because they would be related to ADD FULLTEXT INDEX,
+		and that operation is never supported online. */
+		for (pfs_os_file_t d : deleted) {
+			os_file_close(d);
+		}
 	}
 }
 
@@ -6094,24 +6122,6 @@ create_index_dict(
 	DBUG_RETURN(index);
 }
 
-/** After releasing the data dictionary latch, close deleted files
-@param deleted  handles of deleted files */
-static void close_unlinked_files(const std::vector<pfs_os_file_t> &deleted)
-{
-  dict_sys.assert_not_locked();
-  for (pfs_os_file_t d : deleted)
-    os_file_close(d);
-}
-
-/** Commit a DDL transaction and unlink any deleted files. */
-static void commit_unlock_and_unlink(trx_t *trx)
-{
-  std::vector<pfs_os_file_t> deleted;
-  trx->commit(deleted);
-  row_mysql_unlock_data_dictionary(trx);
-  close_unlinked_files(deleted);
-}
-
 /** Update internal structures with concurrent writes blocked,
 while preparing ALTER TABLE.
 
@@ -11146,9 +11156,8 @@ ha_innobase::commit_inplace_alter_table(
 				ctx->prebuilt->table, altered_table->s);
 		}
 
-		row_mysql_unlock_data_dictionary(trx);
+		unlock_and_close_files(deleted, trx);
 		trx->free();
-		close_unlinked_files(deleted);
 		if (fts_exist) {
 			purge_sys.resume_FTS();
 		}
@@ -11199,9 +11208,8 @@ ha_innobase::commit_inplace_alter_table(
 #endif
 	}
 
-	row_mysql_unlock_data_dictionary(trx);
+	unlock_and_close_files(deleted, trx);
 	trx->free();
-	close_unlinked_files(deleted);
 	if (fts_exist) {
 		purge_sys.resume_FTS();
 	}