MDEV-27858 Assertion `page_dir_get_n_heap(new_page) == 2U' failed in PageBulk::init

- InnoDB should check whether bulk transaction id set to its own
transaction id before start bulk insert operation.

- When bulk insert failure happens, InnoDB should set the error info
of the transaction.

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

@@ -240,3 +240,13 @@ SELECT length(f1) FROM t1;
 length(f1)
 8459264
 DROP TABLE t1;
+#
+#  MDEV-27858  Assertion `page_dir_get_n_heap(new_page) == 2U' failed in PageBulk::init
+#
+CREATE TABLE t1 (c INT) ENGINE=InnoDB;
+CREATE TABLE t2 (c INT) ENGINE=InnoDB;
+INSERT INTO t2 VALUES (1);
+LOCK TABLES t1 WRITE,t2 WRITE;
+INSERT INTO t1 VALUES (1);
+INSERT INTO t2 VALUES (1);
+DROP TABLE t2, t1;

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

@@ -250,3 +250,14 @@ CREATE TABLE t1(f1 MEDIUMTEXT)ENGINE=InnoDB;
 INSERT INTO t1 VALUES(REPEAT(1, 8459264));
 SELECT length(f1) FROM t1;
 DROP TABLE t1;
+
+--echo #
+--echo #  MDEV-27858  Assertion `page_dir_get_n_heap(new_page) == 2U' failed in PageBulk::init
+--echo #
+CREATE TABLE t1 (c INT) ENGINE=InnoDB;
+CREATE TABLE t2 (c INT) ENGINE=InnoDB;
+INSERT INTO t2 VALUES (1);
+LOCK TABLES t1 WRITE,t2 WRITE;
+INSERT INTO t1 VALUES (1);
+INSERT INTO t2 VALUES (1);
+DROP TABLE t2, t1;

storage/innobase/row/row0merge.cc

@@ -5205,7 +5205,11 @@ dberr_t row_merge_bulk_t::bulk_insert_buffered(const dtuple_t &row,
       row_merge_dup_t dup{index, nullptr, nullptr, 0};
       row_merge_buf_sort(buf, &dup);
       if (dup.n_dup)
-        return DB_DUPLICATE_KEY;
+      {
+        trx->error_info= index;
+        err= DB_DUPLICATE_KEY;
+        goto func_exit;
+      }
     }
     else
       row_merge_buf_sort(buf, NULL);
@@ -5214,7 +5218,10 @@ dberr_t row_merge_bulk_t::bulk_insert_buffered(const dtuple_t &row,
     file->n_rec+= buf->n_tuples;
     err= write_to_tmp_file(i);
     if (err != DB_SUCCESS)
-      return err;
+    {
+      trx->error_info= index;
+      goto func_exit;
+    }
     clean_bulk_buffer(i);
     buf= &m_merge_buf[i];
     goto add_to_buf;
@@ -5243,7 +5250,10 @@ dberr_t row_merge_bulk_t::write_to_index(ulint index_no, trx_t *trx)
     {
       row_merge_buf_sort(&buf, &dup);
       if (dup.n_dup)
-        return DB_DUPLICATE_KEY;
+      {
+        err= DB_DUPLICATE_KEY;
+        goto func_exit;
+      }
     }
     else row_merge_buf_sort(&buf, NULL);
     if (file && file->fd != OS_FILE_CLOSED)
@@ -5276,6 +5286,8 @@ dberr_t row_merge_bulk_t::write_to_index(ulint index_no, trx_t *trx)
         nullptr, &m_blob_file);
 
 func_exit:
+  if (err != DB_SUCCESS)
+    trx->error_info= index;
   err= btr_bulk.finish(err);
   return err;
 }

storage/innobase/trx/trx0rec.cc

@@ -2055,7 +2055,8 @@ trx_undo_report_row_operation(
 	} else if (!m.second || !trx->bulk_insert) {
 		bulk = false;
 	} else if (index->table->is_temporary()) {
-	} else if (trx_has_lock_x(*trx, *index->table)) {
+	} else if (trx_has_lock_x(*trx, *index->table)
+		   && index->table->bulk_trx_id == trx->id) {
 		m.first->second.start_bulk_insert(index->table);
 
 		if (dberr_t err = m.first->second.bulk_insert_buffered(