MDEV-33868 Assertion `trx->bulk_insert' failed in innodb_prepare_comm…

…it_versioned

- This issue is caused by commit 188c5da (MDEV-32453).
InnoDB fails to end the bulk insert for the table after
applying the bulk insert operation. This leads to assertion
during commit process.

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

@@ -478,4 +478,14 @@ INSERT INTO t VALUES (1),(1);
 ERROR 21000: Subquery returns more than 1 row
 COMMIT;
 DROP TABLE t;
+#
+#  MDEV-33868 Assertion `trx->bulk_insert' failed in
+#             innodb_prepare_commit_versioned
+#
+CREATE TABLE t (id INT) ENGINE=InnoDB;
+select 1 into outfile "VARDIR/tmp/t.outfile";
+BEGIN;
+LOAD DATA INFILE 'VARDIR/tmp/t.outfile' INTO TABLE t;
+COMMIT;
+DROP TABLE t;
 # End of 10.11 tests

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

@@ -507,4 +507,20 @@ BEGIN;
 INSERT INTO t VALUES (1),(1);
 COMMIT;
 DROP TABLE t;
+
+--echo #
+--echo #  MDEV-33868 Assertion `trx->bulk_insert' failed in
+--echo #             innodb_prepare_commit_versioned
+--echo #
+CREATE TABLE t (id INT) ENGINE=InnoDB;
+--replace_result $MYSQLTEST_VARDIR VARDIR
+--disable_ps2_protocol
+eval select 1 into outfile "$MYSQLTEST_VARDIR/tmp/t.outfile";
+--enable_ps2_protocol
+BEGIN;
+--replace_result $MYSQLTEST_VARDIR VARDIR
+eval LOAD DATA INFILE '$MYSQLTEST_VARDIR/tmp/t.outfile' INTO TABLE t;
+COMMIT;
+DROP TABLE t;
+--remove_file $MYSQLTEST_VARDIR/tmp/t.outfile
 --echo # End of 10.11 tests

storage/innobase/row/row0merge.cc

@@ -5362,6 +5362,7 @@ void trx_t::bulk_rollback_low()
         low_limit= t.second.get_first();
       delete t.second.bulk_store;
       t.second.bulk_store= nullptr;
+      t.second.end_bulk_insert();
     }
   }
   trx_savept_t bulk_save{low_limit};
@@ -5370,13 +5371,19 @@ void trx_t::bulk_rollback_low()
 
 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())
+  if (UNIV_UNLIKELY(!bulk_insert))
     return DB_SUCCESS;
-  dberr_t err= t->write_bulk(table, this);
-  if (err != DB_SUCCESS)
-    bulk_rollback_low();
-  return err;
+  ut_ad(!check_unique_secondary);
+  ut_ad(!check_foreigns);
+  auto it= mod_tables.find(table);
+  if (it != mod_tables.end() && it->second.bulk_store)
+    if (dberr_t err= it->second.write_bulk(table, this))
+    {
+      bulk_rollback_low();
+      return err;
+    }
+  it->second.end_bulk_insert();
+  return DB_SUCCESS;
 }
 
 dberr_t trx_t::bulk_insert_apply_low()