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.

8 files changed, 59 insertions, 19 deletions

diff --git a/mysql-test/suite/innodb/r/create_select.result b/mysql-test/suite/innodb/r/create_select.result
new file mode 100644
index 00000000000..35acb2e265a
--- /dev/null
+++ b/mysql-test/suite/innodb/r/create_select.result
@@ -0,0 +1,5 @@
+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;
diff --git a/mysql-test/suite/innodb/t/create_select.test b/mysql-test/suite/innodb/t/create_select.test
new file mode 100644
index 00000000000..8103902e5f6
--- /dev/null
+++ b/mysql-test/suite/innodb/t/create_select.test
@@ -0,0 +1,28 @@
+--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
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index e04d431174a..de9096c8fcb 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/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 */);
 		}
 	}
 
diff --git a/storage/innobase/include/row0mysql.h b/storage/innobase/include/row0mysql.h
index a8a228e6b2b..2069caf7f3f 100644
--- a/storage/innobase/include/row0mysql.h
+++ b/storage/innobase/include/row0mysql.h
@@ -1,6 +1,7 @@
 /*****************************************************************************
 
 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)
diff --git a/storage/innobase/row/row0mysql.cc b/storage/innobase/row/row0mysql.cc
index 966448b502c..52aa43384f7 100644
--- a/storage/innobase/row/row0mysql.cc
+++ b/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;
diff --git a/storage/xtradb/handler/ha_innodb.cc b/storage/xtradb/handler/ha_innodb.cc
index 0cdd5ac2452..495bf729744 100644
--- a/storage/xtradb/handler/ha_innodb.cc
+++ b/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 */);
 		}
 	}
 
diff --git a/storage/xtradb/include/row0mysql.h b/storage/xtradb/include/row0mysql.h
index b71f5e2103e..6580596550e 100644
--- a/storage/xtradb/include/row0mysql.h
+++ b/storage/xtradb/include/row0mysql.h
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 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)
diff --git a/storage/xtradb/row/row0mysql.cc b/storage/xtradb/row/row0mysql.cc
index a5192dcad8f..34a91b9e5fd 100644
--- a/storage/xtradb/row/row0mysql.cc
+++ b/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;