MDEV-20102 Phantom InnoDB table remains after interrupted CREATE...SELECT

This is a regression due to MDEV-16515 that affects some versions in
the MariaDB 10.1 server series starting with 10.1.35, and possibly
all versions starting with 10.2.17, 10.3.8, and 10.4.0.

The idea of MDEV-16515 is to allow DROP TABLE to be interrupted,
in case it was stuck due to some concurrent activity. We already
made some cases of internal DROP TABLE immune to kill in MDEV-18237,
MDEV-16647, MDEV-17470. We must include the cleanup of
CREATE TABLE...SELECT in the list of such internal DROP TABLE.

ha_innobase::delete_table(): Pass create_failed=true if the current
SQL statement is CREATE, so that the table will be dropped.

row_drop_table_for_mysql(): If create_failed=true, do not allow
the operation to be interrupted.

mysql-test/suite/innodb/r/create_select.result

+CREATE TABLE t1 ENGINE=InnoDB SELECT * FROM seq_1_to_100000000;
+KILL QUERY @id;
+ERROR 70100: Query execution was interrupted
+CREATE TABLE t1 (a SERIAL) ENGINE=InnoDB;
+DROP TABLE t1;

mysql-test/suite/innodb/t/create_select.test

+--source include/have_innodb.inc
+--source include/have_sequence.inc
+--source include/count_sessions.inc
+
+let $ID= `SELECT @id := CONNECTION_ID()`;
+
+connect (con1, localhost, root,,);
+let $ignore= `SELECT @id := $ID`;
+
+connection default;
+send CREATE TABLE t1 ENGINE=InnoDB SELECT * FROM seq_1_to_100000000;
+
+connection con1;
+let $wait_condition=
+  select count(*) = 1 from information_schema.processlist
+  where state = 'Sending data'
+  and info = 'CREATE TABLE t1 ENGINE=InnoDB SELECT * FROM seq_1_to_100000000';
+--source include/wait_condition.inc
+KILL QUERY @id;
+disconnect con1;
+
+connection default;
+--error ER_QUERY_INTERRUPTED
+reap;
+
+CREATE TABLE t1 (a SERIAL) ENGINE=InnoDB;
+DROP TABLE t1;
+--source include/wait_until_count_sessions.inc

storage/innobase/handler/ha_innodb.cc

@@ -12645,11 +12645,12 @@ ha_innobase::delete_table(
 	++trx->will_lock;
 	trx->ddl = true;
 
+	const int sqlcom = thd_sql_command(thd);
+
 	/* Drop the table in InnoDB */
 	err = row_drop_table_for_mysql(
-		norm_name, trx, thd_sql_command(thd) == SQLCOM_DROP_DB,
-		FALSE);
-
+		norm_name, trx, sqlcom == SQLCOM_DROP_DB,
+		sqlcom == SQLCOM_CREATE_TABLE /* CREATE TABLE ... SELECT */);
 
 	if (err == DB_TABLE_NOT_FOUND
 	    && innobase_get_lower_case_table_names() == 1) {
@@ -12679,8 +12680,9 @@ ha_innobase::delete_table(
 #endif
 			err = row_drop_table_for_mysql(
 				par_case_name, trx,
-				thd_sql_command(thd) == SQLCOM_DROP_DB,
-				FALSE);
+				sqlcom == SQLCOM_DROP_DB,
+				sqlcom == SQLCOM_CREATE_TABLE
+				/* CREATE TABLE ... SELECT */);
 		}
 	}
 

storage/innobase/include/row0mysql.h

 /*****************************************************************************
 
 Copyright (c) 2000, 2016, Oracle and/or its affiliates. All Rights Reserved.
+Copyright (c) 2015, 2019, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -491,7 +492,7 @@ row_drop_table_for_mysql(
 	const char*	name,	/*!< in: table name */
 	trx_t*		trx,	/*!< in: dictionary transaction handle */
 	bool		drop_db,/*!< in: true=dropping whole database */
-	ibool		create_failed,/*!<in: TRUE=create table failed
+	bool		create_failed,/*!<in: TRUE=create table failed
 				       because e.g. foreign key column
 				       type mismatch. */
 	bool		nonatomic = true)

storage/innobase/row/row0mysql.cc

@@ -3883,7 +3883,7 @@ row_drop_table_for_mysql(
 	const char*	name,	/*!< in: table name */
 	trx_t*		trx,	/*!< in: transaction handle */
 	bool		drop_db,/*!< in: true=dropping whole database */
-	ibool		create_failed,/*!<in: TRUE=create table failed
+	bool		create_failed,/*!<in: TRUE=create table failed
 				       because e.g. foreign key column
 				       type mismatch. */
 	bool		nonatomic)
@@ -4223,12 +4223,13 @@ row_drop_table_for_mysql(
 		calling btr_search_drop_page_hash_index() while we
 		hold the InnoDB dictionary lock, we will drop any
 		adaptive hash index entries upfront. */
-		const bool is_temp = dict_table_is_temporary(table)
+		const bool immune = create_failed
+			|| dict_table_is_temporary(table)
 			|| strncmp(tablename_minus_db, tmp_file_prefix,
 				   tmp_file_prefix_length)
 			|| strncmp(tablename_minus_db, "FTS_", 4);
 		while (buf_LRU_drop_page_hash_for_tablespace(table)) {
-			if ((!is_temp && trx_is_interrupted(trx))
+			if ((!immune && trx_is_interrupted(trx))
 			    || srv_shutdown_state != SRV_SHUTDOWN_NONE) {
 				err = DB_INTERRUPTED;
 				goto funct_exit;

storage/xtradb/handler/ha_innodb.cc

@@ -13219,11 +13219,12 @@ ha_innobase::delete_table(
 	++trx->will_lock;
 	trx->ddl = true;
 
+	const int sqlcom = thd_sql_command(thd);
+
 	/* Drop the table in InnoDB */
 	err = row_drop_table_for_mysql(
-		norm_name, trx, thd_sql_command(thd) == SQLCOM_DROP_DB,
-		FALSE);
-
+		norm_name, trx, sqlcom == SQLCOM_DROP_DB,
+		sqlcom == SQLCOM_CREATE_TABLE /* CREATE TABLE ... SELECT */);
 
 	if (err == DB_TABLE_NOT_FOUND
 	    && innobase_get_lower_case_table_names() == 1) {
@@ -13253,8 +13254,9 @@ ha_innobase::delete_table(
 #endif
 			err = row_drop_table_for_mysql(
 				par_case_name, trx,
-				thd_sql_command(thd) == SQLCOM_DROP_DB,
-				FALSE);
+				sqlcom == SQLCOM_DROP_DB,
+				sqlcom == SQLCOM_CREATE_TABLE
+				/* CREATE TABLE ... SELECT */);
 		}
 	}
 

storage/xtradb/include/row0mysql.h

 /*****************************************************************************
 
 Copyright (c) 2000, 2017, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, MariaDB Corporation.
+Copyright (c) 2017, 2019, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -493,7 +493,7 @@ row_drop_table_for_mysql(
 	const char*	name,	/*!< in: table name */
 	trx_t*		trx,	/*!< in: dictionary transaction handle */
 	bool		drop_db,/*!< in: true=dropping whole database */
-	ibool		create_failed,/*!<in: TRUE=create table failed
+	bool		create_failed,/*!<in: TRUE=create table failed
 				       because e.g. foreign key column
 				       type mismatch. */
 	bool		nonatomic = true)

storage/xtradb/row/row0mysql.cc

@@ -3893,7 +3893,7 @@ row_drop_table_for_mysql(
 	const char*	name,	/*!< in: table name */
 	trx_t*		trx,	/*!< in: transaction handle */
 	bool		drop_db,/*!< in: true=dropping whole database */
-	ibool		create_failed,/*!<in: TRUE=create table failed
+	bool		create_failed,/*!<in: TRUE=create table failed
 				       because e.g. foreign key column
 				       type mismatch. */
 	bool		nonatomic)
@@ -4233,12 +4233,13 @@ row_drop_table_for_mysql(
 		calling btr_search_drop_page_hash_index() while we
 		hold the InnoDB dictionary lock, we will drop any
 		adaptive hash index entries upfront. */
-		const bool is_temp = dict_table_is_temporary(table)
+		const bool immune = create_failed
+			|| dict_table_is_temporary(table)
 			|| strncmp(tablename_minus_db, tmp_file_prefix,
 				   tmp_file_prefix_length)
 			|| strncmp(tablename_minus_db, "FTS_", 4);
 		while (buf_LRU_drop_page_hash_for_tablespace(table)) {
-			if ((!is_temp && trx_is_interrupted(trx))
+			if ((!immune && trx_is_interrupted(trx))
 			    || srv_shutdown_state != SRV_SHUTDOWN_NONE) {
 				err = DB_INTERRUPTED;
 				goto funct_exit;