MDEV-32453 Bulk insert fails to apply when trigger does insert operation

Reason:
=======
- InnoDB fails to apply the buffered insert operation if the
after insert trigger does change the same table. This behaviour
leads to empty table for the subsequent insert operation
and server abort.

Solution:
========
- InnoDB should apply buffered insert operation if
"after insert" trigger changes the same table.

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

@@ -467,3 +467,15 @@ DROP TABLE t1;
 CREATE TABLE t (a CHAR CHARACTER SET utf8) ENGINE=InnoDB ROW_FORMAT=REDUNDANT;
 INSERT t SELECT left(seq,1) FROM seq_1_to_43691;
 DROP TABLE t;
+#
+#  MDEV-32453  Bulk insert fails to apply when trigger
+#                  does insert operation
+#
+CREATE TABLE t(c INT)ENGINE=InnoDB;
+CREATE TRIGGER t2_ai AFTER INSERT ON t FOR EACH ROW SET @a:=(SELECT * FROM t);
+BEGIN;
+INSERT INTO t VALUES (1),(1);
+ERROR 21000: Subquery returns more than 1 row
+COMMIT;
+DROP TABLE t;
+# End of 10.11 tests

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

@@ -495,3 +495,16 @@ DROP TABLE t1;
 CREATE TABLE t (a CHAR CHARACTER SET utf8) ENGINE=InnoDB ROW_FORMAT=REDUNDANT;
 INSERT t SELECT left(seq,1) FROM seq_1_to_43691;
 DROP TABLE t;
+
+--echo #
+--echo #  MDEV-32453  Bulk insert fails to apply when trigger
+--echo #                  does insert operation
+--echo #
+CREATE TABLE t(c INT)ENGINE=InnoDB;
+CREATE TRIGGER t2_ai AFTER INSERT ON t FOR EACH ROW SET @a:=(SELECT * FROM t);
+BEGIN;
+--error ER_SUBQUERY_NO_1_ROW
+INSERT INTO t VALUES (1),(1);
+COMMIT;
+DROP TABLE t;
+--echo # End of 10.11 tests

storage/innobase/handler/ha_innodb.cc

@@ -15824,7 +15824,7 @@ ha_innobase::start_stmt(
 		}
 		/* fall through */
 	default:
-		trx->end_bulk_insert(*m_prebuilt->table);
+		trx->bulk_insert_apply_for_table(m_prebuilt->table);
 		if (!trx->bulk_insert) {
 			break;
 		}
@@ -16018,7 +16018,7 @@ ha_innobase::external_lock(
 		}
 		/* fall through */
 	default:
-		trx->end_bulk_insert(*m_prebuilt->table);
+		trx->bulk_insert_apply_for_table(m_prebuilt->table);
 		if (!trx->bulk_insert) {
 			break;
 		}

storage/innobase/include/trx0trx.h

@@ -1188,10 +1188,16 @@ struct trx_t : ilist_node<>
     return UNIV_UNLIKELY(bulk_insert) ? bulk_insert_apply_low(): DB_SUCCESS;
   }
 
+  /** Do the bulk insert for the buffered insert operation of a table.
+  @param table bulk insert operation
+  @return DB_SUCCESS or error code. */
+  dberr_t bulk_insert_apply_for_table(dict_table_t *table);
 private:
   /** Apply the buffered bulk inserts. */
   dberr_t bulk_insert_apply_low();
 
+  /** Rollback the bulk insert operation for the transaction */
+  void bulk_rollback_low();
   /** Assign a rollback segment for modifying temporary tables.
   @return the assigned rollback segment */
   trx_rseg_t *assign_temp_rseg();

storage/innobase/row/row0merge.cc

@@ -5351,18 +5351,8 @@ dberr_t trx_mod_table_time_t::write_bulk(dict_table_t *table, trx_t *trx)
   return err;
 }
 
-dberr_t trx_t::bulk_insert_apply_low()
+void trx_t::bulk_rollback_low()
 {
-  ut_ad(bulk_insert);
-  ut_ad(!check_unique_secondary);
-  ut_ad(!check_foreigns);
-  dberr_t err;
-  for (auto& t : mod_tables)
-    if (t.second.is_bulk_insert())
-      if ((err= t.second.write_bulk(t.first, this)) != DB_SUCCESS)
-        goto bulk_rollback;
-  return DB_SUCCESS;
-bulk_rollback:
   undo_no_t low_limit= UINT64_MAX;
   for (auto& t : mod_tables)
   {
@@ -5376,5 +5366,28 @@ dberr_t trx_t::bulk_insert_apply_low()
   }
   trx_savept_t bulk_save{low_limit};
   rollback(&bulk_save);
+}
+
+dberr_t trx_t::bulk_insert_apply_for_table(dict_table_t *table)
+{
+  auto t= check_bulk_buffer(table);
+  if (!t || !t->is_bulk_insert())
+    return DB_SUCCESS;
+  dberr_t err= t->write_bulk(table, this);
+  if (err != DB_SUCCESS)
+    bulk_rollback_low();
   return err;
 }
+
+dberr_t trx_t::bulk_insert_apply_low()
+{
+  ut_ad(bulk_insert);
+  for (auto& t : mod_tables)
+    if (t.second.is_bulk_insert())
+      if (dberr_t err= t.second.write_bulk(t.first, this))
+      {
+        bulk_rollback_low();
+        return err;
+      }
+  return DB_SUCCESS;
+}