MDEV-36504 Memory leak after CREATE TABLE..SELECT

Problem:
========
- After commit cc8eefb0dc (MDEV-33087),
InnoDB does use bulk insert operation for ALTER TABLE.. ALGORITHM=COPY
and CREATE TABLE..SELECT as well. InnoDB fails to clear the bulk
buffer when it encounters error during CREATE..SELECT. Problem
is that while transaction cleanup, InnoDB fails to identify
the bulk insert for DDL operation.

Fix:
====
- Represent bulk_insert in trx by 2 bits. By doing that, InnoDB
can distinguish between TRX_DML_BULK, TRX_DDL_BULK. During DDL,
set bulk insert value for transaction to TRX_DDL_BULK.

- Introduce a parameter HA_EXTRA_ABORT_ALTER_COPY which rollbacks
only TRX_DDL_BULK transaction.

- bulk_insert_apply() happens for TRX_DDL_BULK transaction happens
only during HA_EXTRA_END_ALTER_COPY extra() call.

include/my_base.h

@@ -219,7 +219,10 @@ enum ha_extra_function {
   /** Start writing rows during ALTER TABLE...ALGORITHM=COPY. */
   HA_EXTRA_BEGIN_ALTER_COPY,
   /** Finish writing rows during ALTER TABLE...ALGORITHM=COPY. */
-  HA_EXTRA_END_ALTER_COPY
+  HA_EXTRA_END_ALTER_COPY,
+  /** Abort of writing rows during ALTER TABLE..ALGORITHM=COPY or
+  CREATE..SELCT */
+  HA_EXTRA_ABORT_ALTER_COPY
 };
 
 /* Compatible option, to be deleted in 6.0 */

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

@@ -91,3 +91,24 @@ INSERT INTO t1 VALUES
 ALTER TABLE t1 FORCE, ALGORITHM=COPY;
 DROP TABLE t1;
 SET GLOBAL innodb_stats_persistent=@default_stats_persistent;
+#
+#  MDEV-36504 Memory leak after insert into empty table
+#
+CREATE TABLE t1 (k INT PRIMARY KEY)ENGINE=InnoDB;
+INSERT INTO t1 SET k= 1;
+START TRANSACTION;
+INSERT INTO t1 SET k= 2;
+SELECT COUNT(*) > 0 FROM mysql.innodb_index_stats LOCK IN SHARE MODE;
+COUNT(*) > 0
+1
+connect con1,localhost,root,,,;
+SET innodb_lock_wait_timeout=0;
+CREATE TABLE t2(f1 INT DEFAULT 1 PRIMARY KEY)
+STATS_PERSISTENT= 1 ENGINE=InnoDB as SELECT k FROM t1;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+disconnect con1;
+connection default;
+SET innodb_lock_wait_timeout=default;
+DROP TABLE t1;
+DROP TABLE IF EXISTS t2;
+# restart

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

@@ -109,3 +109,24 @@ INSERT INTO t1 VALUES
 ALTER TABLE t1 FORCE, ALGORITHM=COPY;
 DROP TABLE t1;
 SET GLOBAL innodb_stats_persistent=@default_stats_persistent;
+
+--echo #
+--echo #  MDEV-36504 Memory leak after insert into empty table
+--echo #
+CREATE TABLE t1 (k INT PRIMARY KEY)ENGINE=InnoDB;
+INSERT INTO t1 SET k= 1;
+START TRANSACTION;
+INSERT INTO t1 SET k= 2;
+SELECT COUNT(*) > 0 FROM mysql.innodb_index_stats LOCK IN SHARE MODE;
+
+connect(con1,localhost,root,,,);
+SET innodb_lock_wait_timeout=0;
+--error ER_LOCK_WAIT_TIMEOUT
+CREATE TABLE t2(f1 INT DEFAULT 1 PRIMARY KEY)
+  STATS_PERSISTENT= 1 ENGINE=InnoDB as SELECT k FROM t1;
+disconnect con1;
+connection default;
+SET innodb_lock_wait_timeout=default;
+DROP TABLE t1;
+DROP TABLE IF EXISTS t2;
+--source include/restart_mysqld.inc

sql/ha_partition.cc

@@ -2141,7 +2141,9 @@ int ha_partition::change_partitions(HA_CREATE_INFO *create_info,
     m_added_file[i]->extra(HA_EXTRA_BEGIN_ALTER_COPY);
   error= copy_partitions(copied, deleted);
   for (i= 0; i < part_count; i++)
-    m_added_file[i]->extra(HA_EXTRA_END_ALTER_COPY);
+    m_added_file[i]->extra(error
+                           ? HA_EXTRA_ABORT_ALTER_COPY
+                           : HA_EXTRA_END_ALTER_COPY);
   if (unlikely(error))
   {
     /*
@@ -9467,6 +9469,7 @@ int ha_partition::extra(enum ha_extra_function operation)
   case HA_EXTRA_STARTING_ORDERED_INDEX_SCAN:
   case HA_EXTRA_BEGIN_ALTER_COPY:
   case HA_EXTRA_END_ALTER_COPY:
+  case HA_EXTRA_ABORT_ALTER_COPY:
     DBUG_RETURN(loop_partitions(extra_cb, &operation));
   default:
   {

sql/sql_insert.cc

@@ -4532,7 +4532,7 @@ void select_insert::abort_result_set()
       table->file->ha_rnd_end();
     table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
     table->file->extra(HA_EXTRA_WRITE_CANNOT_REPLACE);
-
+    table->file->extra(HA_EXTRA_ABORT_ALTER_COPY);
     /*
       If at least one row has been inserted/modified and will stay in
       the table (the table doesn't have transactions) we must write to

sql/sql_table.cc

@@ -12378,6 +12378,7 @@ copy_data_between_tables(THD *thd, TABLE *from, TABLE *to, bool ignore,
     if (alt_error > 0)
     {
       error= alt_error;
+      to->file->extra(HA_EXTRA_ABORT_ALTER_COPY);
       copy_data_error_ignore(error, false, to, thd, alter_ctx);
     }
   }

storage/innobase/handler/ha_innodb.cc

@@ -3652,7 +3652,7 @@ ha_innobase::init_table_handle_for_HANDLER(void)
 	m_prebuilt->used_in_HANDLER = TRUE;
 
 	reset_template();
-	m_prebuilt->trx->bulk_insert = false;
+	m_prebuilt->trx->bulk_insert &= TRX_DDL_BULK;
 }
 
 /*********************************************************************//**
@@ -4501,7 +4501,7 @@ static bool end_of_statement(trx_t *trx) noexcept
   undo_no_t savept= 0;
   trx->rollback(&savept);
   /* MariaDB will roll back the entire transaction. */
-  trx->bulk_insert= false;
+  trx->bulk_insert&= TRX_DDL_BULK;
   trx->last_stmt_start= 0;
   return true;
 }
@@ -15875,7 +15875,7 @@ ha_innobase::extra(
 	stmt_boundary:
 		trx->bulk_insert_apply();
 		trx->end_bulk_insert(*m_prebuilt->table);
-		trx->bulk_insert = false;
+		trx->bulk_insert &= TRX_DDL_BULK;
 		break;
 	case HA_EXTRA_NO_KEYREAD:
 		(void)check_trx_exists(ha_thd());
@@ -15941,7 +15941,7 @@ ha_innobase::extra(
 			break;
 		}
 		m_prebuilt->table->skip_alter_undo = 0;
-		if (dberr_t err= trx->bulk_insert_apply()) {
+		if (dberr_t err= trx->bulk_insert_apply<TRX_DDL_BULK>()) {
 			m_prebuilt->table->skip_alter_undo = 0;
 			return convert_error_code_to_mysql(
 				 err, m_prebuilt->table->flags,
@@ -15949,7 +15949,7 @@ ha_innobase::extra(
 		}
 
 		trx->end_bulk_insert(*m_prebuilt->table);
-		trx->bulk_insert = false;
+		trx->bulk_insert &= TRX_DDL_BULK;
 		if (!m_prebuilt->table->is_temporary()
 		    && !high_level_read_only) {
 			/* During copy_data_between_tables(), InnoDB only
@@ -15968,6 +15968,13 @@ ha_innobase::extra(
 			log_buffer_flush_to_disk();
 		}
 		break;
+	case HA_EXTRA_ABORT_ALTER_COPY:
+		if (m_prebuilt->table->skip_alter_undo) {
+			trx = check_trx_exists(ha_thd());
+			m_prebuilt->table->skip_alter_undo = 0;
+			trx->rollback();
+		}
+		break;
 	default:/* Do nothing */
 		;
 	}
@@ -16062,7 +16069,8 @@ ha_innobase::start_stmt(
 			break;
 		}
 
-		trx->bulk_insert = false;
+		ut_ad(trx->bulk_insert != TRX_DDL_BULK);
+		trx->bulk_insert = TRX_NO_BULK;
 		trx->last_stmt_start = trx->undo_no;
 	}
 
@@ -16270,7 +16278,7 @@ ha_innobase::external_lock(
 		if (!trx->bulk_insert) {
 			break;
 		}
-		trx->bulk_insert = false;
+		trx->bulk_insert &= TRX_DDL_BULK;
 		trx->last_stmt_start = trx->undo_no;
 	}
 

storage/innobase/handler/handler0alter.cc

@@ -5544,6 +5544,7 @@ static bool innodb_insert_sys_columns(
 	DBUG_EXECUTE_IF("instant_insert_fail",
 			my_error(ER_INTERNAL_ERROR, MYF(0),
 				 "InnoDB: Insert into SYS_COLUMNS failed");
+			mem_heap_free(info->heap);
 			return true;);
 
 	if (DB_SUCCESS != que_eval_sql(

storage/innobase/include/trx0trx.h

@@ -809,8 +809,13 @@ public:
   /** normally set; "SET unique_checks=0, foreign_key_checks=0"
   enables bulk insert into an empty table */
   unsigned check_unique_secondary:1;
-  /** whether an insert into an empty table is active */
+  /** whether an insert into an empty table is active
-  unsigned bulk_insert:1;
+  Possible states are
+  TRX_NO_BULK
+  TRX_DML_BULK
+  TRX_DDL_BULK
+  @see trx_bulk_insert in trx0types.h */
+  unsigned bulk_insert:2;
 	/*------------------------------*/
 	/* MySQL has a transaction coordinator to coordinate two phase
 	commit between multiple storage engines and the binary log. When
@@ -1117,6 +1122,7 @@ public:
     ut_ad(!is_not_inheriting_locks());
     ut_ad(check_foreigns);
     ut_ad(check_unique_secondary);
+    ut_ad(bulk_insert == TRX_NO_BULK);
   }
 
   /** This has to be invoked on SAVEPOINT or at the end of a statement.
@@ -1142,6 +1148,8 @@ public:
   rollback to the start of a statement will work. */
   void end_bulk_insert()
   {
+    if (bulk_insert == TRX_DDL_BULK)
+      return;
     for (auto& t : mod_tables)
       t.second.end_bulk_insert();
   }
@@ -1149,7 +1157,15 @@ public:
   /** @return whether a bulk insert into empty table is in progress */
   bool is_bulk_insert() const
   {
-    if (!bulk_insert || check_unique_secondary || check_foreigns)
+    switch (bulk_insert) {
+    case TRX_NO_BULK:
+      return false;
+    case TRX_DDL_BULK:
+      return true;
+    default:
+      ut_ad(bulk_insert == TRX_DML_BULK);
+    }
+    if (check_unique_secondary || check_foreigns)
       return false;
     for (const auto& t : mod_tables)
       if (t.second.is_bulk_insert())
@@ -1179,9 +1195,11 @@ public:
   /** Do the bulk insert for the buffered insert operation
   for the transaction.
   @return DB_SUCCESS or error code */
+  template<trx_bulk_insert type= TRX_DML_BULK>
   dberr_t bulk_insert_apply()
   {
-    return UNIV_UNLIKELY(bulk_insert) ? bulk_insert_apply_low(): DB_SUCCESS;
+    static_assert(type != TRX_NO_BULK, "");
+    return bulk_insert == type ? bulk_insert_apply_low(): DB_SUCCESS;
   }
 
 private:

storage/innobase/include/trx0types.h

@@ -65,6 +65,15 @@ enum trx_state_t {
 	TRX_STATE_COMMITTED_IN_MEMORY
 };
 
+/** Transaction bulk insert operation @see trx_t::bulk_insert */
+enum trx_bulk_insert {
+    TRX_NO_BULK,
+    /** bulk insert is being executed during DML */
+    TRX_DML_BULK,
+    /** bulk insert is being executed in copy_data_between_tables() */
+    TRX_DDL_BULK
+};
+
 /** Memory objects */
 /* @{ */
 /** Transaction */

storage/innobase/row/row0ins.cc

@@ -2775,16 +2775,16 @@ err_exit:
 		    && !index->table->has_spatial_index()) {
 
 			ut_ad(!index->table->skip_alter_undo);
-			trx->bulk_insert = true;
+			trx->bulk_insert = TRX_DML_BULK;
 			err = lock_table(index->table, NULL, LOCK_X, thr);
 			if (err != DB_SUCCESS) {
 				trx->error_state = err;
-				trx->bulk_insert = false;
+				trx->bulk_insert = TRX_NO_BULK;
 				goto err_exit;
 			}
 			if (index->table->n_rec_locks) {
 avoid_bulk:
-				trx->bulk_insert = false;
+				trx->bulk_insert = TRX_NO_BULK;
 				goto row_level_insert;
 			}
 #ifdef WITH_WSREP
@@ -2844,7 +2844,7 @@ avoid_bulk:
 			bulk buffer and doesn't check for constraint
 			validity of foreign key relationship. */
 			trx_start_if_not_started(trx, true);
-			trx->bulk_insert = true;
+			trx->bulk_insert = TRX_DDL_BULK;
 			auto m = trx->mod_tables.emplace(index->table, 0);
 			m.first->second.start_bulk_insert(index->table, true);
 			err = m.first->second.bulk_insert_buffered(

storage/innobase/row/row0mysql.cc

@@ -687,8 +687,12 @@ handle_new_error:
 			/* MariaDB will roll back the latest SQL statement */
 			break;
 		}
-		/* MariaDB will roll back the entire transaction. */
+		/* For DML, InnoDB does partial rollback and clear
-		trx->bulk_insert = false;
+		bulk buffer in row_mysql_handle_errors().
+		For ALTER TABLE ALGORITHM=COPY & CREATE TABLE...SELECT,
+		the bulk insert transaction will be rolled back inside
+		ha_innobase::extra(HA_EXTRA_ABORT_ALTER_COPY) */
+		trx->bulk_insert &= TRX_DDL_BULK;
 		trx->last_stmt_start = 0;
 		break;
 	case DB_LOCK_WAIT:

storage/innobase/trx/trx0trx.cc

@@ -134,8 +134,6 @@ trx_init(
 
 	trx->will_lock = false;
 
-	trx->bulk_insert = false;
-
 	trx->apply_online_log = false;
 
 	ut_d(trx->start_file = 0);
@@ -1513,6 +1511,7 @@ bool trx_t::commit_cleanup() noexcept
   *detailed_error= '\0';
   mod_tables.clear();
 
+  bulk_insert= TRX_NO_BULK;
   check_foreigns= true;
   check_unique_secondary= true;
   assert_freed();

storage/mroonga/ha_mroonga.cpp

@@ -558,6 +558,9 @@ static const char *mrn_inspect_extra_function(enum ha_extra_function operation)
   case HA_EXTRA_END_ALTER_COPY:
     inspected = "HA_EXTRA_END_ALTER_COPY";
     break;
+  case HA_EXTRA_ABORT_ALTER_COPY:
+    inspected = "HA_EXTRA_ABORT_ALTER_COPY";
+    break;
 #ifdef MRN_HAVE_HA_EXTRA_EXPORT
   case HA_EXTRA_EXPORT:
     inspected = "HA_EXTRA_EXPORT";
@@ -593,6 +596,11 @@ static const char *mrn_inspect_extra_function(enum ha_extra_function operation)
     inspected = "HA_EXTRA_END_ALTER_COPY";
     break;
 #endif
+#ifdef MRN_HAVE_HA_EXTRA_ABORT_ALTER_COPY
+  case HA_EXTRA_ABORT_ALTER_COPY:
+    inspected = "HA_EXTRA_ABORT_ALTER_COPY";
+    break;
+#endif
 #ifdef MRN_HAVE_HA_EXTRA_NO_AUTOINC_LOCKING
   case HA_EXTRA_NO_AUTOINC_LOCKING:
     inspected = "HA_EXTRA_NO_AUTOINC_LOCKING";