MDEV-25315 Crash in SHOW ENGINE INNODB STATUS

In commit 8ea923f5 (MDEV-24818)
when we optimized multi-statement INSERT into an empty table,
we would sometimes wrongly enable bulk insert into a table that
is actually already using row-level locking and undo logging.

trx_has_lock_x(): New predicate, to check if the transaction of
the current thread is holding an exclusive lock on a table.

trx_undo_report_row_operation(): Only invoke
trx_mod_table_time_t::start_bulk_insert() if
trx_has_lock_x() holds.

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

@@ -102,3 +102,18 @@ f1
 7
 DROP TABLE t1;
 SET foreign_key_checks=0;
+#
+# MDEV-25315 Crash in SHOW ENGINE INNODB STATUS
+#
+CREATE TABLE t1 (a INT PRIMARY KEY) ENGINE=InnoDB;
+CREATE TABLE t2 (a INT PRIMARY KEY) ENGINE=InnoDB;
+INSERT INTO t1 VALUES(1);
+BEGIN;
+INSERT INTO t1 VALUES(1);
+ERROR 23000: Duplicate entry '1' for key 'PRIMARY'
+INSERT INTO t2 VALUES(0);
+INSERT INTO t1 VALUES(2), (2);
+ERROR 23000: Duplicate entry '2' for key 'PRIMARY'
+SHOW ENGINE InnoDB STATUS;
+COMMIT;
+DROP TABLE t1,t2;

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

@@ -99,3 +99,23 @@ SELECT * FROM t1;
 DROP TABLE t1;
 
 SET foreign_key_checks=0;
+
+--echo #
+--echo # MDEV-25315 Crash in SHOW ENGINE INNODB STATUS
+--echo #
+
+CREATE TABLE t1 (a INT PRIMARY KEY) ENGINE=InnoDB;
+CREATE TABLE t2 (a INT PRIMARY KEY) ENGINE=InnoDB;
+
+INSERT INTO t1 VALUES(1);
+BEGIN;
+--error ER_DUP_ENTRY
+INSERT INTO t1 VALUES(1);
+INSERT INTO t2 VALUES(0);
+--error ER_DUP_ENTRY
+INSERT INTO t1 VALUES(2), (2);
+--disable_result_log
+SHOW ENGINE InnoDB STATUS;
+--enable_result_log
+COMMIT;
+DROP TABLE t1,t2;

storage/innobase/include/dict0mem.h

@@ -2340,7 +2340,7 @@ struct dict_table_t {
 	/* @} */
 
   /** Number of granted or pending LOCK_S or LOCK_X on the table.
-  Protected by lock_mutex. */
+  Protected by lock_sys.assert_locked(*this). */
   uint32_t n_lock_x_or_s;
 
 	/** FTS specific state variables. */

storage/innobase/trx/trx0rec.cc

@@ -1948,6 +1948,28 @@ dberr_t trx_undo_report_rename(trx_t* trx, const dict_table_t* table)
 	return err;
 }
 
+ATTRIBUTE_COLD ATTRIBUTE_NOINLINE
+/** @return whether the transaction holds an exclusive lock on a table */
+static bool trx_has_lock_x(const trx_t &trx, dict_table_t& table)
+{
+  if (table.is_temporary())
+    return true;
+
+  table.lock_mutex_lock();
+  const auto n= table.n_lock_x_or_s;
+  table.lock_mutex_unlock();
+
+  /* This thread is executing trx. No other thread can modify our table locks
+  (only record locks might be created, in an implicit-to-explicit conversion).
+  Hence, no mutex is needed here. */
+  if (n == 1)
+    for (const lock_t *lock : trx.lock.table_locks)
+      if (lock && lock->type_mode == (LOCK_X | LOCK_TABLE))
+        return true;
+
+  return false;
+}
+
 /***********************************************************************//**
 Writes information to an undo log about an insert, update, or a delete marking
 of a clustered index record. This information is used in a rollback of the
@@ -2014,7 +2036,8 @@ trx_undo_report_row_operation(
 		ut_ad(que_node_get_type(thr->run_node) == QUE_NODE_INSERT);
 		ut_ad(trx->bulk_insert);
 		return DB_SUCCESS;
-	} else if (m.second && trx->bulk_insert) {
+	} else if (m.second && trx->bulk_insert
+		   && trx_has_lock_x(*trx, *index->table)) {
 		m.first->second.start_bulk_insert();
 	} else {
 		bulk = false;