MDEV-515 InnoDB bulk insert10.6-MDEV-515

dict_table_t::bulk_trx_id: Stores the bulk insert
transaction id. Protected by exclusive lock of the table.
It can be used by other connection read operation to
identify whether to read the table

ins_node_t::bulk_insert: Flag to indicate the bulk
insert of the table.

trx_mod_table_time_t::bulk_insert_undo: Enable undo logging
for the consecutive insert during bulk insert operation.

Introduced new undo log record "TRX_UNDO_EMPTY".
It should be first undo log during bulk insert operation.
While rollback, if innodb encounters the undo record
then it should empty the table

dict_table_t::empty_table(): Basically it empties all the
indexes associated with table . This is undo operation
of bulk insert operation. Metadata record should be retained
during this rollback operation

dict_table_t::remove_bulk_trx(): Resets the bulk_trx_id

row_log_table_empty(): If the table is being created during
empty_table() then log the EMPTY operation in the online log

row_log_online_op(): If the secondary index is being created
during empty_table() then log ROW_OP_EMPTY operation
in online log

row_log_table_apply_empty(): Applies the ROW_T_EMPTY to the
table that was being rebuild. It does call empty_table() to
empty the table.

btr_root_page_init(): Initialize the root page of the b-tree.
It is used during dict_index_t::empty() and btr_create().

btr_cur_ins_lock_and_undo(): During first insert of bulk
insert operation, write UNDO_EMPTY undo log of it and
reset DB_TRX_ID and DB_ROLL_PTR.

trx_mark_sql_stat_end(): Set the bulk_insert_undo for
bulk operation of the table. So that consecutive insert
does allow undo operation.

row_search_mvcc(): check whether the current transaction
can view the table records based on bulk_trx_id.

row_merge_read_clustered_index(): Avoid the table read if the
bulk transaction id of the table is not visible within
current transaction read view.

- HA_EXTRA_IGNORE_INSERT flag in ha_innobase::extra() is to
indicate ignore statement. Sets the bulk insert undo
for the transaction.

- Add new insert in many test case to avoid the hang. Because MDEV-515
takes X-lock of the table for first insert. So concurrent insert
will wait for X-lock to be released.

80 files changed, 767 insertions, 157 deletions

diff --git a/include/my_base.h b/include/my_base.h
index dc5b135628f..81d32847b64 100644
--- a/include/my_base.h
+++ b/include/my_base.h
@@ -219,7 +219,9 @@ enum ha_extra_function {
   /** Finish writing rows during ALTER TABLE...ALGORITHM=COPY. */
   HA_EXTRA_END_ALTER_COPY,
   /** Fake the start of a statement after wsrep_load_data_splitting hack */
-  HA_EXTRA_FAKE_START_STMT
+  HA_EXTRA_FAKE_START_STMT,
+  /** IGNORE is being used for the insert statement */
+  HA_EXTRA_IGNORE_INSERT
 };
 
 /* Compatible option, to be deleted in 6.0 */
diff --git a/mysql-test/main/commit.result b/mysql-test/main/commit.result
index f2e012d4782..44d46900d38 100644
--- a/mysql-test/main/commit.result
+++ b/mysql-test/main/commit.result
@@ -247,11 +247,13 @@ COMMIT;
 SET @@completion_type=1;
 COMMIT AND NO CHAIN;
 SET TRANSACTION ISOLATION LEVEL READ COMMITTED;
-START TRANSACTION;
 TRUNCATE TABLE t1;
+INSERT INTO t1 VALUES(100);
+START TRANSACTION;
 INSERT INTO t1 VALUES (1000);
 SELECT * FROM t1;
 s1
+100
 1000
 Should read '1000'
 connection con1;
@@ -260,6 +262,7 @@ COMMIT;
 connection default;
 SELECT * FROM t1;
 s1
+100
 1000
 Should only read the '1000' as this transaction is now in REP READ
 COMMIT AND NO CHAIN;
diff --git a/mysql-test/main/commit.test b/mysql-test/main/commit.test
index 762397dfa23..9f5e077fd2d 100644
--- a/mysql-test/main/commit.test
+++ b/mysql-test/main/commit.test
@@ -284,8 +284,11 @@ COMMIT;
 SET @@completion_type=1;
 COMMIT AND NO CHAIN;
 SET TRANSACTION ISOLATION LEVEL READ COMMITTED;
-START TRANSACTION;
 TRUNCATE TABLE t1;
+# MDEV-515 takes X-lock on the table for the first insert.
+# So concurrent insert won't happen on the table
+INSERT INTO t1 VALUES(100);
+START TRANSACTION;
 INSERT INTO t1 VALUES (1000);
 SELECT * FROM t1;
 --echo Should read '1000'
diff --git a/mysql-test/main/flush_read_lock.result b/mysql-test/main/flush_read_lock.result
index 9a81cae1724..ae57788c7d4 100644
--- a/mysql-test/main/flush_read_lock.result
+++ b/mysql-test/main/flush_read_lock.result
@@ -1372,6 +1372,7 @@ unlock tables;
 connection default;
 # Reap XA COMMIT.
 delete from t3_trans;
+INSERT INTO t3_trans VALUES(100);
 #
 # Check that XA COMMIT / ROLLBACK for prepared transaction from a
 # disconnected session is blocked by active FTWRL in another connection.
diff --git a/mysql-test/main/flush_read_lock.test b/mysql-test/main/flush_read_lock.test
index 205b8b302ea..40f8b2aec3b 100644
--- a/mysql-test/main/flush_read_lock.test
+++ b/mysql-test/main/flush_read_lock.test
@@ -1679,6 +1679,9 @@ connection default;
 --echo # Reap XA COMMIT.
 --reap
 delete from t3_trans;
+# MDEV-515 takes X-lock on the table for the first insert.
+# So concurrent insert won't happen on the table
+INSERT INTO t3_trans VALUES(100);
 --echo #
 --echo # Check that XA COMMIT / ROLLBACK for prepared transaction from a
 --echo # disconnected session is blocked by active FTWRL in another connection.
diff --git a/mysql-test/main/innodb_mysql_lock.result b/mysql-test/main/innodb_mysql_lock.result
index 25cf2882cd0..6ca332018ca 100644
--- a/mysql-test/main/innodb_mysql_lock.result
+++ b/mysql-test/main/innodb_mysql_lock.result
@@ -12,6 +12,8 @@ connect  con3,localhost,root,,;
 connection con1;
 set @@autocommit=0;
 CREATE TABLE t1(s1 INT UNIQUE) ENGINE=innodb;
+INSERT INTO t1 VALUES (100);
+COMMIT;
 INSERT INTO t1 VALUES (1);
 connection con2;
 set @@autocommit=0;
diff --git a/mysql-test/main/innodb_mysql_lock.test b/mysql-test/main/innodb_mysql_lock.test
index 6e746468dba..5ee688b1d6e 100644
--- a/mysql-test/main/innodb_mysql_lock.test
+++ b/mysql-test/main/innodb_mysql_lock.test
@@ -25,6 +25,11 @@ connect (con3,localhost,root,,);
 connection con1;
 set @@autocommit=0;
 CREATE TABLE t1(s1 INT UNIQUE) ENGINE=innodb;
+# MDEV-515 takes X-lock on the table for the first insert.
+# So concurrent DML won't happen on the table
+INSERT INTO t1 VALUES (100);
+COMMIT;
+
 INSERT INTO t1 VALUES (1);
 
 connection con2;
diff --git a/mysql-test/main/xa_prepared_binlog_off.result b/mysql-test/main/xa_prepared_binlog_off.result
index ca19f6cdfaf..ebcb24d3af3 100644
--- a/mysql-test/main/xa_prepared_binlog_off.result
+++ b/mysql-test/main/xa_prepared_binlog_off.result
@@ -3,6 +3,7 @@ connection default;
 CREATE VIEW v_processlist  as SELECT * FROM performance_schema.threads where type = 'FOREGROUND';
 call mtr.add_suppression("Found 10 prepared XA transactions");
 CREATE TABLE t (a INT) ENGINE=innodb;
+INSERT INTO t VALUES(100);
 connect  conn$index$type, 127.0.0.1,root,,test,$MASTER_MYPORT,;
 SET @@sql_log_bin = OFF;
 CREATE TEMPORARY TABLE tmp1 (a int) ENGINE=innodb;
@@ -48,18 +49,21 @@ connect  conn$index$type, 127.0.0.1,root,,test,$MASTER_MYPORT,;
 XA START   'trx1ro';
 SELECT * from t ORDER BY a;
 a
+100
 XA END     'trx1ro';
 XA PREPARE 'trx1ro';
 connect  conn$index$type, 127.0.0.1,root,,test,$MASTER_MYPORT,;
 XA START   'trx2ro';
 SELECT * from t ORDER BY a;
 a
+100
 XA END     'trx2ro';
 XA PREPARE 'trx2ro';
 connect  conn$index$type, 127.0.0.1,root,,test,$MASTER_MYPORT,;
 XA START   'trx3ro';
 SELECT * from t ORDER BY a;
 a
+100
 XA END     'trx3ro';
 XA PREPARE 'trx3ro';
 connection default;
@@ -402,6 +406,7 @@ ERROR XAE08: XAER_DUPID: The XID already exists
 XA ROLLBACK 'trx_19';
 SELECT * FROM t;
 a
+100
 5
 6
 7
@@ -533,6 +538,7 @@ a
 12
 13
 14
+100
 XA END     'trx1ro';
 XA PREPARE 'trx1ro';
 connect  conn$index$type, 127.0.0.1,root,,test,$MASTER_MYPORT,;
@@ -559,6 +565,7 @@ a
 12
 13
 14
+100
 XA END     'trx2ro';
 XA PREPARE 'trx2ro';
 connect  conn$index$type, 127.0.0.1,root,,test,$MASTER_MYPORT,;
@@ -585,6 +592,7 @@ a
 12
 13
 14
+100
 XA END     'trx3ro';
 XA PREPARE 'trx3ro';
 connection default;
@@ -927,6 +935,7 @@ ERROR XAE08: XAER_DUPID: The XID already exists
 XA ROLLBACK 'trx_19';
 SELECT * FROM t;
 a
+100
 5
 6
 7
@@ -1036,7 +1045,7 @@ XA END     'one_phase_trx_4';
 XA COMMIT  'one_phase_trx_4' ONE PHASE;
 SELECT SUM(a) FROM t;
 SUM(a)
-290
+390
 DROP TABLE t;
 DROP VIEW v_processlist;
 All transactions must be completed, to empty-list the following:
diff --git a/mysql-test/suite/binlog/r/binlog_xa_prepared_disconnect.result b/mysql-test/suite/binlog/r/binlog_xa_prepared_disconnect.result
index 9fda8ab3143..10a0fc39266 100644
--- a/mysql-test/suite/binlog/r/binlog_xa_prepared_disconnect.result
+++ b/mysql-test/suite/binlog/r/binlog_xa_prepared_disconnect.result
@@ -3,6 +3,7 @@ RESET MASTER;
 CREATE VIEW v_processlist  as SELECT * FROM performance_schema.threads where type = 'FOREGROUND';
 call mtr.add_suppression("Found 10 prepared XA transactions");
 CREATE TABLE t (a INT) ENGINE=innodb;
+INSERT INTO t VALUES(100);
 connect  conn$index$type, 127.0.0.1,root,,test,$MASTER_MYPORT,;
 SET @@sql_log_bin = OFF;
 CREATE TEMPORARY TABLE tmp1 (a int) ENGINE=innodb;
@@ -48,18 +49,21 @@ connect  conn$index$type, 127.0.0.1,root,,test,$MASTER_MYPORT,;
 XA START   'trx1ro';
 SELECT * from t ORDER BY a;
 a
+100
 XA END     'trx1ro';
 XA PREPARE 'trx1ro';
 connect  conn$index$type, 127.0.0.1,root,,test,$MASTER_MYPORT,;
 XA START   'trx2ro';
 SELECT * from t ORDER BY a;
 a
+100
 XA END     'trx2ro';
 XA PREPARE 'trx2ro';
 connect  conn$index$type, 127.0.0.1,root,,test,$MASTER_MYPORT,;
 XA START   'trx3ro';
 SELECT * from t ORDER BY a;
 a
+100
 XA END     'trx3ro';
 XA PREPARE 'trx3ro';
 connection default;
@@ -402,6 +406,7 @@ ERROR XAE08: XAER_DUPID: The XID already exists
 XA ROLLBACK 'trx_19';
 SELECT * FROM t;
 a
+100
 5
 6
 7
@@ -533,6 +538,7 @@ a
 12
 13
 14
+100
 XA END     'trx1ro';
 XA PREPARE 'trx1ro';
 connect  conn$index$type, 127.0.0.1,root,,test,$MASTER_MYPORT,;
@@ -559,6 +565,7 @@ a
 12
 13
 14
+100
 XA END     'trx2ro';
 XA PREPARE 'trx2ro';
 connect  conn$index$type, 127.0.0.1,root,,test,$MASTER_MYPORT,;
@@ -585,6 +592,7 @@ a
 12
 13
 14
+100
 XA END     'trx3ro';
 XA PREPARE 'trx3ro';
 connection default;
@@ -927,6 +935,7 @@ ERROR XAE08: XAER_DUPID: The XID already exists
 XA ROLLBACK 'trx_19';
 SELECT * FROM t;
 a
+100
 5
 6
 7
@@ -1036,7 +1045,7 @@ XA END     'one_phase_trx_4';
 XA COMMIT  'one_phase_trx_4' ONE PHASE;
 SELECT SUM(a) FROM t;
 SUM(a)
-290
+390
 DROP TABLE t;
 DROP VIEW v_processlist;
 include/show_binlog_events.inc
@@ -1048,6 +1057,9 @@ master-bin.000001	#	Query	#	#	use `mtr`; INSERT INTO test_suppressions (pattern)
 master-bin.000001	#	Query	#	#	COMMIT
 master-bin.000001	#	Gtid	#	#	GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t (a INT) ENGINE=innodb
+master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; INSERT INTO t VALUES(100)
+master-bin.000001	#	Xid	#	#	COMMIT /* XID */
 master-bin.000001	#	Gtid	#	#	XA START X'7472785f30',X'',1 GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; INSERT INTO t SET a=0
 master-bin.000001	#	Query	#	#	XA END X'7472785f30',X'',1
diff --git a/mysql-test/suite/binlog/t/binlog_xa_prepared.inc b/mysql-test/suite/binlog/t/binlog_xa_prepared.inc
index b6306791cf4..e93832dbf08 100644
--- a/mysql-test/suite/binlog/t/binlog_xa_prepared.inc
+++ b/mysql-test/suite/binlog/t/binlog_xa_prepared.inc
@@ -54,6 +54,10 @@ CREATE VIEW v_processlist  as SELECT * FROM performance_schema.threads where typ
 
 CREATE TABLE t (a INT) ENGINE=innodb;
 
+# MDEV-515 takes X-lock on the table for the first insert.
+# So concurrent insert won't happen on the table
+INSERT INTO t VALUES(100);
+
 # Counter is incremented at the end of post restart to
 # reflect number of loops done in correctness computation.
 --let $restart_number = 0
diff --git a/mysql-test/suite/innodb/r/alter_candidate_key.result b/mysql-test/suite/innodb/r/alter_candidate_key.result
index 23989e0da5f..b0b56047abc 100644
--- a/mysql-test/suite/innodb/r/alter_candidate_key.result
+++ b/mysql-test/suite/innodb/r/alter_candidate_key.result
@@ -66,6 +66,7 @@ test.t1	check	status	OK
 DROP TABLE t1;
 SET SQL_MODE= strict_trans_tables;
 CREATE TABLE t1(a INT UNIQUE) ENGINE=InnoDB;
+INSERT INTO t1 VALUES(3);
 SET DEBUG_SYNC='row_log_table_apply1_before SIGNAL dml WAIT_FOR dml_done';
 ALTER TABLE t1 MODIFY COLUMN a INT NOT NULL;
 connection con1;
diff --git a/mysql-test/suite/innodb/r/binlog_consistent.result b/mysql-test/suite/innodb/r/binlog_consistent.result
index 0562a92ef68..cceca394166 100644
--- a/mysql-test/suite/innodb/r/binlog_consistent.result
+++ b/mysql-test/suite/innodb/r/binlog_consistent.result
@@ -5,6 +5,7 @@ connect con3,localhost,root,,;
 connect con4,localhost,root,,;
 connection default;
 CREATE TABLE t1 (a INT, b VARCHAR(100), PRIMARY KEY (a,b)) ENGINE=innodb;
+INSERT INTO t1 VALUES(9, "");
 SHOW MASTER STATUS;
 File	Position	Binlog_Do_DB	Binlog_Ignore_DB
 master-bin.000001	<pos>		
@@ -39,6 +40,7 @@ connection default;
 SELECT * FROM t1 ORDER BY a,b;
 a	b
 0	
+9	
 SHOW STATUS LIKE 'binlog_snapshot_%';
 Variable_name	Value
 Binlog_snapshot_file	master-bin.000001
@@ -61,6 +63,7 @@ connection default;
 SELECT * FROM t1 ORDER BY a,b;
 a	b
 0	
+9	
 SHOW STATUS LIKE 'binlog_snapshot_%';
 Variable_name	Value
 Binlog_snapshot_file	master-bin.000001
@@ -80,6 +83,9 @@ include/show_binlog_events.inc
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
 master-bin.000001	#	Gtid	#	#	GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t1 (a INT, b VARCHAR(100), PRIMARY KEY (a,b)) ENGINE=innodb
+master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; INSERT INTO t1 VALUES(9, "")
+master-bin.000001	#	Xid	#	#	COMMIT /* XID */
 master-bin.000001	#	Gtid	#	#	GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t2 (a INT PRIMARY KEY) ENGINE=myisam
 master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
diff --git a/mysql-test/suite/innodb/r/ddl_purge.result b/mysql-test/suite/innodb/r/ddl_purge.result
index 45f4c99e97b..a1d96de24ca 100644
--- a/mysql-test/suite/innodb/r/ddl_purge.result
+++ b/mysql-test/suite/innodb/r/ddl_purge.result
@@ -1,5 +1,6 @@
 CREATE TABLE t0 (pk INT PRIMARY KEY) ENGINE=InnoDB;
 CREATE TABLE t1 (pk INT PRIMARY KEY, b INT) ENGINE=InnoDB;
+INSERT INTO t0 VALUES(100);
 connect  con1,localhost,root,,test;
 BEGIN;
 INSERT INTO t0 SET pk=1;
diff --git a/mysql-test/suite/innodb/r/group_commit.result b/mysql-test/suite/innodb/r/group_commit.result
index e2fb89767d4..5fd7f9c9155 100644
--- a/mysql-test/suite/innodb/r/group_commit.result
+++ b/mysql-test/suite/innodb/r/group_commit.result
@@ -1,4 +1,5 @@
 CREATE TABLE t1 (a VARCHAR(10) PRIMARY KEY) ENGINE=innodb;
+INSERT INTO t1 VALUES(100);
 SELECT variable_value INTO @commits FROM information_schema.global_status
 WHERE variable_name = 'binlog_commits';
 SELECT variable_value INTO @group_commits FROM information_schema.global_status
@@ -32,11 +33,13 @@ SET DEBUG_SYNC= "now WAIT_FOR group2_con4";
 SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED;
 SELECT * FROM t1 ORDER BY a;
 a
+100
 SET DEBUG_SYNC= "now SIGNAL group2_queued";
 connection con1;
 connection default;
 SELECT * FROM t1 ORDER BY a;
 a
+100
 con1
 connection con5;
 SET DEBUG_SYNC= "commit_before_get_LOCK_after_binlog_sync SIGNAL group3_con5";
@@ -51,11 +54,13 @@ connection default;
 SET DEBUG_SYNC= "now WAIT_FOR group3_con5";
 SELECT * FROM t1 ORDER BY a;
 a
+100
 con1
 SET DEBUG_SYNC= "now SIGNAL group3_committed";
 SET DEBUG_SYNC= "now WAIT_FOR group2_visible";
 SELECT * FROM t1 ORDER BY a;
 a
+100
 con1
 con2
 con3
@@ -69,6 +74,7 @@ connection con6;
 connection default;
 SELECT * FROM t1 ORDER BY a;
 a
+100
 con1
 con2
 con3
diff --git a/mysql-test/suite/innodb/r/group_commit_no_optimize_thread.result b/mysql-test/suite/innodb/r/group_commit_no_optimize_thread.result
index 41e9f6a9d08..e22589ed255 100644
--- a/mysql-test/suite/innodb/r/group_commit_no_optimize_thread.result
+++ b/mysql-test/suite/innodb/r/group_commit_no_optimize_thread.result
@@ -1,4 +1,5 @@
 CREATE TABLE t1 (a VARCHAR(10) PRIMARY KEY) ENGINE=innodb;
+INSERT INTO t1 VALUES("default");
 SELECT variable_value INTO @commits FROM information_schema.global_status
 WHERE variable_name = 'binlog_commits';
 SELECT variable_value INTO @group_commits FROM information_schema.global_status
@@ -32,12 +33,14 @@ SET DEBUG_SYNC= "now WAIT_FOR group2_con4";
 SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED;
 SELECT * FROM t1 ORDER BY a;
 a
+default
 SET DEBUG_SYNC= "now SIGNAL group2_queued";
 connection con1;
 connection default;
 SELECT * FROM t1 ORDER BY a;
 a
 con1
+default
 connection con5;
 SET DEBUG_SYNC= "commit_before_get_LOCK_after_binlog_sync SIGNAL group3_con5";
 SET DEBUG_SYNC= "commit_after_get_LOCK_log SIGNAL con5_leader WAIT_FOR con6_queued";
@@ -52,6 +55,7 @@ SET DEBUG_SYNC= "now WAIT_FOR group3_con5";
 SELECT * FROM t1 ORDER BY a;
 a
 con1
+default
 SET DEBUG_SYNC= "now SIGNAL group3_committed";
 SET DEBUG_SYNC= "now WAIT_FOR group2_visible";
 SELECT * FROM t1 ORDER BY a;
@@ -60,6 +64,7 @@ con1
 con2
 con3
 con4
+default
 SET DEBUG_SYNC= "now SIGNAL group2_checked";
 connection con2;
 connection con3;
@@ -75,6 +80,7 @@ con3
 con4
 con5
 con6
+default
 SELECT variable_value - @commits FROM information_schema.global_status
 WHERE variable_name = 'binlog_commits';
 variable_value - @commits
diff --git a/mysql-test/suite/innodb/r/innodb-lock.result b/mysql-test/suite/innodb/r/innodb-lock.result
index 1fe0d263fef..83bed139da8 100644
--- a/mysql-test/suite/innodb/r/innodb-lock.result
+++ b/mysql-test/suite/innodb/r/innodb-lock.result
@@ -147,6 +147,8 @@ DROP TABLE t1, t2;
 #
 CREATE TABLE t1 (pk INT AUTO_INCREMENT PRIMARY KEY) ENGINE=InnoDB
 PARTITION BY key (pk) PARTITIONS 2;
+INSERT INTO t1 VALUES(100);
+INSERT INTO t1 VALUES(101);
 CREATE TABLE t2 (a INT) ENGINE=InnoDB;
 INSERT INTO t2 VALUES (1),(2),(3),(4),(5),(6);
 CREATE TABLE t3 (b INT) ENGINE=InnoDB;
diff --git a/mysql-test/suite/innodb/t/alter_candidate_key.test b/mysql-test/suite/innodb/t/alter_candidate_key.test
index 7429cd89a1a..979d8fa4fee 100644
--- a/mysql-test/suite/innodb/t/alter_candidate_key.test
+++ b/mysql-test/suite/innodb/t/alter_candidate_key.test
@@ -42,6 +42,9 @@ DROP TABLE t1;
 
 SET SQL_MODE= strict_trans_tables;
 CREATE TABLE t1(a INT UNIQUE) ENGINE=InnoDB;
+# MDEV-515 takes X-lock on the table for the first insert.
+# So concurrent insert won't happen on the table
+INSERT INTO t1 VALUES(3);
 SET DEBUG_SYNC='row_log_table_apply1_before SIGNAL dml WAIT_FOR dml_done';
 --send ALTER TABLE t1 MODIFY COLUMN a INT NOT NULL
 connection con1;
diff --git a/mysql-test/suite/innodb/t/autoinc_persist.test b/mysql-test/suite/innodb/t/autoinc_persist.test
index fd85b45fbfa..6e094b40e02 100644
--- a/mysql-test/suite/innodb/t/autoinc_persist.test
+++ b/mysql-test/suite/innodb/t/autoinc_persist.test
@@ -410,6 +410,17 @@ INSERT INTO mdev6076a SET b=NULL;
 SELECT * FROM mdev6076a;
 INSERT INTO mdev6076b SET b=NULL;
 SELECT * FROM mdev6076b;
+# MDEV-515 resets the PAGE_ROOT_AUTOINC field in
+# root page during rollback.
+--disable_query_log
+BEGIN;
+let $i = 55;
+WHILE ($i) {
+ eval INSERT INTO mdev6076empty SET b=$i;
+ dec $i;
+}
+ROLLBACK;
+--enable_query_log
 INSERT INTO mdev6076empty SET b=NULL;
 SELECT * FROM mdev6076empty;
 DROP TABLE mdev6076a, mdev6076b, mdev6076empty;
diff --git a/mysql-test/suite/innodb/t/binlog_consistent.test b/mysql-test/suite/innodb/t/binlog_consistent.test
index 2a735a30a0e..6a1935f80c4 100644
--- a/mysql-test/suite/innodb/t/binlog_consistent.test
+++ b/mysql-test/suite/innodb/t/binlog_consistent.test
@@ -16,7 +16,10 @@ connect(con4,localhost,root,,);
 connection default;
 
 CREATE TABLE t1 (a INT, b VARCHAR(100), PRIMARY KEY (a,b)) ENGINE=innodb;
-let pos=`select $binlog_start_pos + 254`;
+# MDEV-515 takes X-lock on the table for the first insert.
+# So concurrent insert won't happen on the table
+INSERT INTO t1 VALUES(9, "");
+let pos=`select $binlog_start_pos + 422`;
 --replace_result $pos <pos>
 SHOW MASTER STATUS;
 --replace_result $pos <pos>
@@ -53,10 +56,10 @@ COMMIT;
 
 connection default;
 SELECT * FROM t1 ORDER BY a,b;
-let pos=`select $binlog_start_pos + 788`;
+let pos=`select $binlog_start_pos + 956`;
 --replace_result $pos <pos>
 SHOW STATUS LIKE 'binlog_snapshot_%';
-let pos=`select $binlog_start_pos + 1164`;
+let pos=`select $binlog_start_pos + 1332`;
 --replace_result $pos <pos>
 SHOW MASTER STATUS;
 SELECT * FROM t2 ORDER BY a;
@@ -74,7 +77,7 @@ FLUSH LOGS;
 
 connection default;
 SELECT * FROM t1 ORDER BY a,b;
-let pos=`select $binlog_start_pos + 788`;
+let pos=`select $binlog_start_pos + 956`;
 --replace_result $pos <pos>
 SHOW STATUS LIKE 'binlog_snapshot_%';
 let pos=`select $binlog_start_pos + 131`;
diff --git a/mysql-test/suite/innodb/t/ddl_purge.test b/mysql-test/suite/innodb/t/ddl_purge.test
index 60d17acead8..678cb597c03 100644
--- a/mysql-test/suite/innodb/t/ddl_purge.test
+++ b/mysql-test/suite/innodb/t/ddl_purge.test
@@ -4,7 +4,9 @@
 
 CREATE TABLE t0 (pk INT PRIMARY KEY) ENGINE=InnoDB;
 CREATE TABLE t1 (pk INT PRIMARY KEY, b INT) ENGINE=InnoDB;
-
+# MDEV-515 takes X-lock on the table for the first insert.
+# So concurrent insert won't happen on the table
+INSERT INTO t0 VALUES(100);
 --connect (con1,localhost,root,,test)
 BEGIN;
 INSERT INTO t0 SET pk=1;
diff --git a/mysql-test/suite/innodb/t/group_commit.test b/mysql-test/suite/innodb/t/group_commit.test
index 692e06f38b8..3a2018bec5d 100644
--- a/mysql-test/suite/innodb/t/group_commit.test
+++ b/mysql-test/suite/innodb/t/group_commit.test
@@ -10,6 +10,9 @@
 # to check some edge case for concurrency control.
 
 CREATE TABLE t1 (a VARCHAR(10) PRIMARY KEY) ENGINE=innodb;
+# MDEV-515 takes X-lock on the table for the first insert.
+# So concurrent insert won't happen on the table
+INSERT INTO t1 VALUES(100);
 
 SELECT variable_value INTO @commits FROM information_schema.global_status
  WHERE variable_name = 'binlog_commits';
diff --git a/mysql-test/suite/innodb/t/group_commit_no_optimize_thread.test b/mysql-test/suite/innodb/t/group_commit_no_optimize_thread.test
index 85c0e295424..107fc6f4056 100644
--- a/mysql-test/suite/innodb/t/group_commit_no_optimize_thread.test
+++ b/mysql-test/suite/innodb/t/group_commit_no_optimize_thread.test
@@ -10,6 +10,9 @@
 # to check some edge case for concurrency control.
 
 CREATE TABLE t1 (a VARCHAR(10) PRIMARY KEY) ENGINE=innodb;
+# MDEV-515 takes X-LOCK on the table for the first insert.
+# So concurrent insert won't happen on the table
+INSERT INTO t1 VALUES("default");
 
 SELECT variable_value INTO @commits FROM information_schema.global_status
  WHERE variable_name = 'binlog_commits';
diff --git a/mysql-test/suite/innodb/t/innodb-lock.test b/mysql-test/suite/innodb/t/innodb-lock.test
index 9e5505270be..84219db111e 100644
--- a/mysql-test/suite/innodb/t/innodb-lock.test
+++ b/mysql-test/suite/innodb/t/innodb-lock.test
@@ -199,6 +199,10 @@ DROP TABLE t1, t2;
 
 CREATE TABLE t1 (pk INT AUTO_INCREMENT PRIMARY KEY) ENGINE=InnoDB
 PARTITION BY key (pk) PARTITIONS 2;
+# MDEV-515 takes X-lock on the table for the first insert.
+# So concurrent insert won't happen on the table
+INSERT INTO t1 VALUES(100);
+INSERT INTO t1 VALUES(101);
 
 CREATE TABLE t2 (a INT) ENGINE=InnoDB;
 INSERT INTO t2 VALUES (1),(2),(3),(4),(5),(6);
diff --git a/mysql-test/suite/innodb_fts/r/innodb_fts_plugin.result b/mysql-test/suite/innodb_fts/r/innodb_fts_plugin.result
index ec417af97c5..b1426ece222 100644
--- a/mysql-test/suite/innodb_fts/r/innodb_fts_plugin.result
+++ b/mysql-test/suite/innodb_fts/r/innodb_fts_plugin.result
@@ -139,25 +139,25 @@ INSERT INTO articles (title, body) VALUES
 SELECT * FROM articles WHERE
 MATCH(title, body) AGAINST('MySQL');
 id	title	body
-6	MySQL Tutorial	DBMS stands for MySQL DataBase ...
-7	How To Use MySQL Well	After you went through a ...
-8	Optimizing MySQL	In this tutorial we will show ...
-9	1001 MySQL Tricks	How to use full-text search engine
+1	MySQL Tutorial	DBMS stands for MySQL DataBase ...
+2	How To Use MySQL Well	After you went through a ...
+3	Optimizing MySQL	In this tutorial we will show ...
+4	1001 MySQL Tricks	How to use full-text search engine
 SELECT * FROM articles WHERE
 MATCH(title, body) AGAINST('tutorial');
 id	title	body
-6	MySQL Tutorial	DBMS stands for MySQL DataBase ...
-8	Optimizing MySQL	In this tutorial we will show ...
+1	MySQL Tutorial	DBMS stands for MySQL DataBase ...
+3	Optimizing MySQL	In this tutorial we will show ...
 SELECT * FROM articles WHERE
 MATCH(title, body) AGAINST('Tricks');
 id	title	body
-9	1001 MySQL Tricks	How to use full-text search engine
-10	Go MariaDB Tricks	How to use full text search engine
+4	1001 MySQL Tricks	How to use full-text search engine
+5	Go MariaDB Tricks	How to use full text search engine
 SELECT * FROM articles WHERE
 MATCH(title, body) AGAINST('full text search');
 id	title	body
-10	Go MariaDB Tricks	How to use full text search engine
-9	1001 MySQL Tricks	How to use full-text search engine
+5	Go MariaDB Tricks	How to use full text search engine
+4	1001 MySQL Tricks	How to use full-text search engine
 SELECT COUNT(*) FROM articles;
 COUNT(*)
 5
diff --git a/mysql-test/suite/mariabackup/incremental_backup.result b/mysql-test/suite/mariabackup/incremental_backup.result
index d6a78655a0c..ed67ceee8e2 100644
--- a/mysql-test/suite/mariabackup/incremental_backup.result
+++ b/mysql-test/suite/mariabackup/incremental_backup.result
@@ -1,6 +1,7 @@
 call mtr.add_suppression("InnoDB: New log files created");
 CREATE TABLE t_aria(i INT) ENGINE ARIA;
 CREATE TABLE t(i INT PRIMARY KEY) ENGINE INNODB;
+INSERT INTO t VALUES(100);
 BEGIN;
 INSERT INTO t VALUES(2);
 connect  con1,localhost,root,,;
@@ -17,6 +18,7 @@ SELECT * FROM t;
 i
 1
 2
+100
 # Prepare full backup, apply incremental one
 # Aria log file was updated during applying incremental backup
 disconnect con1;
@@ -29,5 +31,6 @@ SELECT * FROM t;
 i
 1
 2
+100
 DROP TABLE t;
 DROP TABLE t_aria;
diff --git a/mysql-test/suite/mariabackup/incremental_backup.test b/mysql-test/suite/mariabackup/incremental_backup.test
index 88e277fd95a..62cdf9e6cb3 100644
--- a/mysql-test/suite/mariabackup/incremental_backup.test
+++ b/mysql-test/suite/mariabackup/incremental_backup.test
@@ -13,6 +13,9 @@ let incremental_dir=$MYSQLTEST_VARDIR/tmp/backup_inc1;
 
 CREATE TABLE t_aria(i INT) ENGINE ARIA;
 CREATE TABLE t(i INT PRIMARY KEY) ENGINE INNODB;
+# MDEV-515 takes X-lock on the table for the first insert.
+# So concurrent insert won't happen on the table
+INSERT INTO t VALUES(100);
 BEGIN;
 INSERT INTO t VALUES(2);
 connect (con1,localhost,root,,);
diff --git a/mysql-test/suite/multi_source/disabled.def b/mysql-test/suite/multi_source/disabled.def
index e69de29bb2d..c1cfd27d125 100644
--- a/mysql-test/suite/multi_source/disabled.def
+++ b/mysql-test/suite/multi_source/disabled.def
@@ -0,0 +1 @@
+multi_parallel : MDEV-515
diff --git a/mysql-test/suite/parts/inc/partition_auto_increment.inc b/mysql-test/suite/parts/inc/partition_auto_increment.inc
index 4392d04db8a..3721caeb465 100644
--- a/mysql-test/suite/parts/inc/partition_auto_increment.inc
+++ b/mysql-test/suite/parts/inc/partition_auto_increment.inc
@@ -393,6 +393,12 @@ connect(con1, localhost, root,,);
 connection default;
 eval CREATE TABLE t1 (c1 INT NOT NULL AUTO_INCREMENT, PRIMARY KEY (c1))
   ENGINE = $engine;
+if ($engine == "'InnoDB'")
+{
+# MDEV-515 takes X-lock on the table for the first insert.
+# So concurrent insert won't happen on the table
+INSERT INTO t1(c1) VALUES(100);
+}
 START TRANSACTION;
 INSERT INTO t1 (c1) VALUES (2);
 INSERT INTO t1 (c1) VALUES (4);
@@ -434,6 +440,13 @@ eval CREATE TABLE t1 (c1 INT NOT NULL AUTO_INCREMENT, PRIMARY KEY (c1))
   ENGINE = $engine
   PARTITION BY HASH(c1)
   PARTITIONS 2;
+IF ($engine == "'InnoDB'")
+{
+# MDEV-515 takes X-lock on the table for the first insert.
+# So concurrent insert won't happen on the table
+INSERT INTO t1 (c1) VALUES (100);
+INSERT INTO t1 (c1) VALUES (101);
+}
 START TRANSACTION;
 INSERT INTO t1 (c1) VALUES (2);
 INSERT INTO t1 (c1) VALUES (4);
diff --git a/mysql-test/suite/parts/r/partition_auto_increment_innodb.result b/mysql-test/suite/parts/r/partition_auto_increment_innodb.result
index 76f1ddfceae..7b25d4858ff 100644
--- a/mysql-test/suite/parts/r/partition_auto_increment_innodb.result
+++ b/mysql-test/suite/parts/r/partition_auto_increment_innodb.result
@@ -490,6 +490,7 @@ connect con1, localhost, root,,;
 connection default;
 CREATE TABLE t1 (c1 INT NOT NULL AUTO_INCREMENT, PRIMARY KEY (c1))
 ENGINE = 'InnoDB';
+INSERT INTO t1(c1) VALUES(100);
 START TRANSACTION;
 INSERT INTO t1 (c1) VALUES (2);
 INSERT INTO t1 (c1) VALUES (4);
@@ -510,17 +511,19 @@ connection con1;
 INSERT INTO t1 (c1) VALUES (NULL);
 SELECT * FROM t1 ORDER BY c1;
 c1
-5
 10
-22
-23
+100
+101
+104
+105
 COMMIT;
 SELECT * FROM t1 ORDER BY c1;
 c1
-5
 10
-22
-23
+100
+101
+104
+105
 disconnect con1;
 connection default;
 INSERT INTO t1 (c1) VALUES (NULL);
@@ -528,31 +531,33 @@ SELECT * FROM t1 ORDER BY c1;
 c1
 2
 4
-5
 10
-11
-12
 16
 19
 21
-22
-23
-24
+100
+101
+102
+103
+104
+105
+106
 COMMIT;
 SELECT * FROM t1 ORDER BY c1;
 c1
 2
 4
-5
 10
-11
-12
 16
 19
 21
-22
-23
-24
+100
+101
+102
+103
+104
+105
+106
 DROP TABLE t1;
 # Test with two threads + start transaction
 connect con1, localhost, root,,;
@@ -561,6 +566,8 @@ CREATE TABLE t1 (c1 INT NOT NULL AUTO_INCREMENT, PRIMARY KEY (c1))
 ENGINE = 'InnoDB'
 PARTITION BY HASH(c1)
 PARTITIONS 2;
+INSERT INTO t1 (c1) VALUES (100);
+INSERT INTO t1 (c1) VALUES (101);
 START TRANSACTION;
 INSERT INTO t1 (c1) VALUES (2);
 INSERT INTO t1 (c1) VALUES (4);
@@ -578,17 +585,21 @@ connection con1;
 INSERT INTO t1 (c1) VALUES (NULL);
 SELECT * FROM t1 ORDER BY c1;
 c1
-5
 10
-22
-23
+100
+101
+102
+105
+106
 COMMIT;
 SELECT * FROM t1 ORDER BY c1;
 c1
-5
 10
-22
-23
+100
+101
+102
+105
+106
 disconnect con1;
 connection default;
 INSERT INTO t1 (c1) VALUES (NULL);
@@ -596,31 +607,35 @@ SELECT * FROM t1 ORDER BY c1;
 c1
 2
 4
-5
 10
-11
-12
 16
 19
 21
-22
-23
-24
+100
+101
+102
+103
+104
+105
+106
+107
 COMMIT;
 SELECT * FROM t1 ORDER BY c1;
 c1
 2
 4
-5
 10
-11
-12
 16
 19
 21
-22
-23
-24
+100
+101
+102
+103
+104
+105
+106
+107
 DROP TABLE t1;
 # Test with another column after
 CREATE TABLE t1 (
diff --git a/mysql-test/suite/rpl/include/rpl_parallel_gco_wait_kill.inc b/mysql-test/suite/rpl/include/rpl_parallel_gco_wait_kill.inc
index d918b2ea692..415960e563e 100644
--- a/mysql-test/suite/rpl/include/rpl_parallel_gco_wait_kill.inc
+++ b/mysql-test/suite/rpl/include/rpl_parallel_gco_wait_kill.inc
@@ -26,6 +26,10 @@ CHANGE MASTER TO master_use_gtid=slave_pos;
 --connect (con_temp5,127.0.0.1,root,,test,$SERVER_MYPORT_1,)
 ALTER TABLE mysql.gtid_slave_pos ENGINE=InnoDB;
 CREATE TABLE t3 (a INT PRIMARY KEY, b INT) ENGINE=InnoDB;
+# MDEV-515 takes X-lock on the table for the first insert.
+# So concurrent insert won't happen on the table
+INSERT INTO t3 VALUES(100, 100);
+
 --save_master_pos
 
 --connection server_2
diff --git a/mysql-test/suite/rpl/include/rpl_parallel_ignored_errors.inc b/mysql-test/suite/rpl/include/rpl_parallel_ignored_errors.inc
index 7a6a758a508..493385f1ae3 100644
--- a/mysql-test/suite/rpl/include/rpl_parallel_ignored_errors.inc
+++ b/mysql-test/suite/rpl/include/rpl_parallel_ignored_errors.inc
@@ -57,6 +57,9 @@ CALL mtr.add_suppression("Commit failed due to failure of an earlier commit on w
 --connection server_1
 ALTER TABLE mysql.gtid_slave_pos ENGINE=InnoDB;
 CREATE TABLE t1 (a int PRIMARY KEY) ENGINE=InnoDB;
+# MDEV-515 takes X-lock on the table for the first insert.
+# So concurrent insert won't happen on the table
+INSERT INTO t1 VALUES(1);
 --source include/save_master_gtid.inc
 
 --connection server_2
diff --git a/mysql-test/suite/rpl/include/rpl_parallel_slave_bgc_kill.inc b/mysql-test/suite/rpl/include/rpl_parallel_slave_bgc_kill.inc
index a78dbad052f..efb998b0443 100644
--- a/mysql-test/suite/rpl/include/rpl_parallel_slave_bgc_kill.inc
+++ b/mysql-test/suite/rpl/include/rpl_parallel_slave_bgc_kill.inc
@@ -25,6 +25,10 @@ ALTER TABLE mysql.gtid_slave_pos ENGINE=InnoDB;
 CREATE TABLE t1 (a int PRIMARY KEY) ENGINE=MyISAM;
 CREATE TABLE t2 (a int PRIMARY KEY) ENGINE=InnoDB;
 CREATE TABLE t3 (a INT PRIMARY KEY, b INT) ENGINE=InnoDB;
+# MDEV-515 takes X-lock on the table for the first insert.
+# So concurrent insert won't happen on the table
+INSERT INTO t2 VALUES(100);
+INSERT INTO t3 VALUES(100, 100);
 --save_master_pos
 
 --connection server_2
diff --git a/mysql-test/suite/rpl/include/rpl_parallel_stop_slave.inc b/mysql-test/suite/rpl/include/rpl_parallel_stop_slave.inc
index 4eeddc927e0..35879e98e66 100644
--- a/mysql-test/suite/rpl/include/rpl_parallel_stop_slave.inc
+++ b/mysql-test/suite/rpl/include/rpl_parallel_stop_slave.inc
@@ -23,6 +23,10 @@ ALTER TABLE mysql.gtid_slave_pos ENGINE=InnoDB;
 CREATE TABLE t1 (a int PRIMARY KEY) ENGINE=MyISAM;
 CREATE TABLE t2 (a int PRIMARY KEY) ENGINE=InnoDB;
 CREATE TABLE t3 (a INT PRIMARY KEY, b INT) ENGINE=InnoDB;
+# MDEV-515 takes X-lock on the table for the first insert.
+# So concurrent insert won't happen on the table
+INSERT INTO t2 VALUES(100);
+INSERT INTO t3 VALUES(100, 100);
 --save_master_pos
 
 --connection server_2
diff --git a/mysql-test/suite/rpl/r/parallel_backup.result b/mysql-test/suite/rpl/r/parallel_backup.result
index d87c61f2d0f..9d394a220d5 100644
--- a/mysql-test/suite/rpl/r/parallel_backup.result
+++ b/mysql-test/suite/rpl/r/parallel_backup.result
@@ -6,6 +6,7 @@ include/master-slave.inc
 #
 connection master;
 CREATE TABLE t1 (a INT PRIMARY KEY) ENGINE = innodb;
+INSERT INTO t1 VALUES(100);
 connection slave;
 include/stop_slave.inc
 SET @old_parallel_threads= @@GLOBAL.slave_parallel_threads;
diff --git a/mysql-test/suite/rpl/r/rpl_mariadb_slave_capability.result b/mysql-test/suite/rpl/r/rpl_mariadb_slave_capability.result
index d384422f88a..ac846ac6c00 100644
--- a/mysql-test/suite/rpl/r/rpl_mariadb_slave_capability.result
+++ b/mysql-test/suite/rpl/r/rpl_mariadb_slave_capability.result
@@ -78,6 +78,7 @@ slave-relay-bin.000007	#	Query	#	#	COMMIT
 *** MDEV-5754: MySQL 5.5 slaves cannot replicate from MariaDB 10.0 ***
 connection master;
 CREATE TABLE t2 (a INT PRIMARY KEY) ENGINE=InnoDB;
+INSERT INTO t2 VALUES(100);
 connect  con1,127.0.0.1,root,,test,$SERVER_MYPORT_1,;
 SET debug_sync='commit_after_release_LOCK_prepare_ordered SIGNAL master_queued1 WAIT_FOR master_cont1';
 INSERT INTO t2 VALUES (1);
@@ -112,6 +113,7 @@ SELECT * FROM t2 ORDER BY a;
 a
 1
 2
+100
 # Test that slave which cannot tolerate holes in binlog stream but
 # knows the event does not get dummy event
 include/stop_slave.inc
diff --git a/mysql-test/suite/rpl/r/rpl_parallel_gco_wait_kill.result b/mysql-test/suite/rpl/r/rpl_parallel_gco_wait_kill.result
index 4472550c4f2..f12d19442f1 100644
--- a/mysql-test/suite/rpl/r/rpl_parallel_gco_wait_kill.result
+++ b/mysql-test/suite/rpl/r/rpl_parallel_gco_wait_kill.result
@@ -20,6 +20,7 @@ connect  con_temp4,127.0.0.1,root,,test,$SERVER_MYPORT_1,;
 connect  con_temp5,127.0.0.1,root,,test,$SERVER_MYPORT_1,;
 ALTER TABLE mysql.gtid_slave_pos ENGINE=InnoDB;
 CREATE TABLE t3 (a INT PRIMARY KEY, b INT) ENGINE=InnoDB;
+INSERT INTO t3 VALUES(100, 100);
 connection server_2;
 connection server_1;
 SET sql_log_bin=0;
@@ -136,6 +137,7 @@ a	b
 68	68
 69	69
 70	70
+100	100
 SET debug_sync='RESET';
 connection server_2;
 SET debug_sync='now SIGNAL d0_cont';
@@ -161,6 +163,7 @@ a	b
 68	68
 69	69
 70	70
+100	100
 SET debug_sync='RESET';
 SET GLOBAL slave_parallel_threads=0;
 SET GLOBAL slave_parallel_threads=10;
@@ -190,6 +193,7 @@ a	b
 68	68
 69	69
 70	70
+100	100
 SET sql_log_bin=0;
 DROP FUNCTION foo;
 CREATE FUNCTION foo(x INT, d1 VARCHAR(500), d2 VARCHAR(500))
@@ -225,6 +229,7 @@ SELECT * FROM t3 WHERE a >= 80 ORDER BY a;
 a	b
 80	0
 81	10000
+100	100
 connection server_2;
 SET debug_sync='now WAIT_FOR wait_queue_ready';
 KILL THD_ID;
@@ -244,6 +249,7 @@ a	b
 80	0
 81	10000
 82	0
+100	100
 connection server_2;
 include/stop_slave.inc
 SET GLOBAL slave_parallel_mode=@old_parallel_mode;
diff --git a/mysql-test/suite/rpl/r/rpl_parallel_ignored_errors.result b/mysql-test/suite/rpl/r/rpl_parallel_ignored_errors.result
index 3dd5a3ea83c..ce11b814d44 100644
--- a/mysql-test/suite/rpl/r/rpl_parallel_ignored_errors.result
+++ b/mysql-test/suite/rpl/r/rpl_parallel_ignored_errors.result
@@ -13,6 +13,7 @@ include/start_slave.inc
 connection server_1;
 ALTER TABLE mysql.gtid_slave_pos ENGINE=InnoDB;
 CREATE TABLE t1 (a int PRIMARY KEY) ENGINE=InnoDB;
+INSERT INTO t1 VALUES(1);
 include/save_master_gtid.inc
 connection server_2;
 include/sync_with_master_gtid.inc
diff --git a/mysql-test/suite/rpl/r/rpl_parallel_retry.result b/mysql-test/suite/rpl/r/rpl_parallel_retry.result
index b43556815d1..381fec649ff 100644
--- a/mysql-test/suite/rpl/r/rpl_parallel_retry.result
+++ b/mysql-test/suite/rpl/r/rpl_parallel_retry.result
@@ -259,6 +259,7 @@ connection server_1;
 CREATE TABLE t3 (a INT PRIMARY KEY, b INT, KEY b_idx(b)) ENGINE=InnoDB;
 INSERT INTO t3 VALUES (1,NULL), (2,2), (3,NULL), (4,4), (5, NULL), (6, 6);
 CREATE TABLE t4 (a INT PRIMARY KEY, b INT) ENGINE=InnoDB;
+INSERT INTO t4 VALUES(100, 100);
 SET @old_format= @@SESSION.binlog_format;
 SET binlog_format='statement';
 connection server_2;
@@ -344,6 +345,7 @@ DROP function foo;
 connection server_2;
 connection server_1;
 CREATE TABLE t1 (a int PRIMARY KEY, b INT) ENGINE=InnoDB;
+INSERT INTO t1 VALUES(100, 100);
 connection server_2;
 include/stop_slave.inc
 SET @old_parallel_threads=@@GLOBAL.slave_parallel_threads;
diff --git a/mysql-test/suite/rpl/r/rpl_parallel_slave_bgc_kill.result b/mysql-test/suite/rpl/r/rpl_parallel_slave_bgc_kill.result
index 320bf0e49f8..ba131ea094f 100644
--- a/mysql-test/suite/rpl/r/rpl_parallel_slave_bgc_kill.result
+++ b/mysql-test/suite/rpl/r/rpl_parallel_slave_bgc_kill.result
@@ -17,6 +17,8 @@ ALTER TABLE mysql.gtid_slave_pos ENGINE=InnoDB;
 CREATE TABLE t1 (a int PRIMARY KEY) ENGINE=MyISAM;
 CREATE TABLE t2 (a int PRIMARY KEY) ENGINE=InnoDB;
 CREATE TABLE t3 (a INT PRIMARY KEY, b INT) ENGINE=InnoDB;
+INSERT INTO t2 VALUES(100);
+INSERT INTO t3 VALUES(100, 100);
 connection server_2;
 connection server_1;
 SET sql_log_bin=0;
@@ -80,6 +82,7 @@ a	b
 32	32
 33	33
 34	34
+100	100
 SET debug_sync='RESET';
 connection server_2;
 SET sql_log_bin=0;
@@ -98,6 +101,7 @@ STOP SLAVE IO_THREAD;
 SELECT * FROM t3 WHERE a >= 30 ORDER BY a;
 a	b
 31	31
+100	100
 SET debug_sync='RESET';
 SET GLOBAL slave_parallel_threads=0;
 SET GLOBAL slave_parallel_threads=10;
@@ -121,6 +125,7 @@ a	b
 33	33
 34	34
 39	0
+100	100
 SET sql_log_bin=0;
 DROP FUNCTION foo;
 CREATE FUNCTION foo(x INT, d1 VARCHAR(500), d2 VARCHAR(500))
@@ -179,6 +184,7 @@ a	b
 42	42
 43	43
 44	44
+100	100
 SET debug_sync='RESET';
 connection server_2;
 SET debug_sync='now WAIT_FOR t2_query';
@@ -211,6 +217,7 @@ a	b
 43	43
 44	44
 49	0
+100	100
 SET sql_log_bin=0;
 DROP FUNCTION foo;
 CREATE FUNCTION foo(x INT, d1 VARCHAR(500), d2 VARCHAR(500))
@@ -274,6 +281,7 @@ a	b
 52	52
 53	53
 54	54
+100	100
 SET debug_sync='RESET';
 connection server_2;
 SET debug_sync='now WAIT_FOR t2_query';
@@ -286,6 +294,7 @@ include/wait_for_slave_sql_error.inc [errno=1317,1927,1964]
 SELECT * FROM t3 WHERE a >= 50 ORDER BY a;
 a	b
 51	51
+100	100
 SET debug_sync='RESET';
 SET GLOBAL slave_parallel_threads=0;
 SET GLOBAL slave_parallel_threads=10;
@@ -309,6 +318,7 @@ a	b
 53	53
 54	54
 59	0
+100	100
 connection server_2;
 include/stop_slave.inc
 CHANGE MASTER TO master_use_gtid=slave_pos;
diff --git a/mysql-test/suite/rpl/r/rpl_parallel_stop_slave.result b/mysql-test/suite/rpl/r/rpl_parallel_stop_slave.result
index 6c9fd168e73..0c810d2a3f4 100644
--- a/mysql-test/suite/rpl/r/rpl_parallel_stop_slave.result
+++ b/mysql-test/suite/rpl/r/rpl_parallel_stop_slave.result
@@ -16,6 +16,8 @@ ALTER TABLE mysql.gtid_slave_pos ENGINE=InnoDB;
 CREATE TABLE t1 (a int PRIMARY KEY) ENGINE=MyISAM;
 CREATE TABLE t2 (a int PRIMARY KEY) ENGINE=InnoDB;
 CREATE TABLE t3 (a INT PRIMARY KEY, b INT) ENGINE=InnoDB;
+INSERT INTO t2 VALUES(100);
+INSERT INTO t3 VALUES(100, 100);
 connection server_2;
 include/stop_slave.inc
 connection server_1;
@@ -55,9 +57,11 @@ SELECT * FROM t2 WHERE a >= 20 ORDER BY a;
 a
 20
 21
+100
 SELECT * FROM t3 WHERE a >= 20 ORDER BY a;
 a	b
 20	20
+100	100
 include/start_slave.inc
 SELECT * FROM t1 WHERE a >= 20 ORDER BY a;
 a
@@ -66,11 +70,13 @@ SELECT * FROM t2 WHERE a >= 20 ORDER BY a;
 a
 20
 21
+100
 SELECT * FROM t3 WHERE a >= 20 ORDER BY a;
 a	b
 20	20
 21	21
 22	22
+100	100
 connection server_2;
 include/stop_slave.inc
 SET GLOBAL slave_parallel_mode=@old_parallel_mode;
diff --git a/mysql-test/suite/rpl/r/rpl_semisync_ali_issues.result b/mysql-test/suite/rpl/r/rpl_semisync_ali_issues.result
index 9607e8a7998..36a2bac5838 100644
--- a/mysql-test/suite/rpl/r/rpl_semisync_ali_issues.result
+++ b/mysql-test/suite/rpl/r/rpl_semisync_ali_issues.result
@@ -74,6 +74,7 @@ select * from t1;
 a
 1
 truncate table t1;
+INSERT INTO  t1 VALUES (100);
 connection slave;
 connection con1;
 SET DEBUG_SYNC= 'reset';
@@ -107,6 +108,7 @@ SET DEBUG_SYNC= "now WAIT_FOR after_commit_done";
 connection slave;
 select * from t1;
 a
+100
 1
 2
 3
@@ -114,6 +116,7 @@ a
 connection con2;
 select * from t1;
 a
+100
 1
 2
 3
@@ -123,6 +126,7 @@ connection con1;
 connection con1;
 select * from t1;
 a
+100
 1
 2
 3
diff --git a/mysql-test/suite/rpl/r/rpl_xa_survive_disconnect.result b/mysql-test/suite/rpl/r/rpl_xa_survive_disconnect.result
index 0ee7b497077..3ab263b7f35 100644
--- a/mysql-test/suite/rpl/r/rpl_xa_survive_disconnect.result
+++ b/mysql-test/suite/rpl/r/rpl_xa_survive_disconnect.result
@@ -7,6 +7,7 @@ CREATE DATABASE d1;
 CREATE DATABASE d2;
 CREATE TABLE d1.t (a INT) ENGINE=innodb;
 CREATE TABLE d2.t (a INT) ENGINE=innodb;
+INSERT INTO d1.t VALUES(100);
 connect  master_conn1, 127.0.0.1,root,,test,$MASTER_MYPORT,;
 SET @@session.binlog_format= statement;
 XA START '1-stmt';
@@ -45,6 +46,9 @@ master-bin.000001	#	Gtid	#	#	GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE d1.t (a INT) ENGINE=innodb
 master-bin.000001	#	Gtid	#	#	GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE d2.t (a INT) ENGINE=innodb
+master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; INSERT INTO d1.t VALUES(100)
+master-bin.000001	#	Xid	#	#	COMMIT /* XID */
 master-bin.000001	#	Gtid	#	#	XA START X'312d73746d74',X'',1 GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; INSERT INTO d1.t VALUES (1)
 master-bin.000001	#	Query	#	#	XA END X'312d73746d74',X'',1
@@ -90,6 +94,7 @@ connection master2;
 XA START '4';
 SELECT * FROM d1.t;
 a
+100
 1
 2
 XA END '4';
diff --git a/mysql-test/suite/rpl/r/rpl_xa_survive_disconnect_lsu_off.result b/mysql-test/suite/rpl/r/rpl_xa_survive_disconnect_lsu_off.result
index 0ee7b497077..3ab263b7f35 100644
--- a/mysql-test/suite/rpl/r/rpl_xa_survive_disconnect_lsu_off.result
+++ b/mysql-test/suite/rpl/r/rpl_xa_survive_disconnect_lsu_off.result
@@ -7,6 +7,7 @@ CREATE DATABASE d1;
 CREATE DATABASE d2;
 CREATE TABLE d1.t (a INT) ENGINE=innodb;
 CREATE TABLE d2.t (a INT) ENGINE=innodb;
+INSERT INTO d1.t VALUES(100);
 connect  master_conn1, 127.0.0.1,root,,test,$MASTER_MYPORT,;
 SET @@session.binlog_format= statement;
 XA START '1-stmt';
@@ -45,6 +46,9 @@ master-bin.000001	#	Gtid	#	#	GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE d1.t (a INT) ENGINE=innodb
 master-bin.000001	#	Gtid	#	#	GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE d2.t (a INT) ENGINE=innodb
+master-bin.000001	#	Gtid	#	#	BEGIN GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; INSERT INTO d1.t VALUES(100)
+master-bin.000001	#	Xid	#	#	COMMIT /* XID */
 master-bin.000001	#	Gtid	#	#	XA START X'312d73746d74',X'',1 GTID #-#-#
 master-bin.000001	#	Query	#	#	use `test`; INSERT INTO d1.t VALUES (1)
 master-bin.000001	#	Query	#	#	XA END X'312d73746d74',X'',1
@@ -90,6 +94,7 @@ connection master2;
 XA START '4';
 SELECT * FROM d1.t;
 a
+100
 1
 2
 XA END '4';
diff --git a/mysql-test/suite/rpl/t/parallel_backup.test b/mysql-test/suite/rpl/t/parallel_backup.test
index 6ed182c024b..964e2a30309 100644
--- a/mysql-test/suite/rpl/t/parallel_backup.test
+++ b/mysql-test/suite/rpl/t/parallel_backup.test
@@ -10,6 +10,9 @@
 
 --connection master
 CREATE TABLE t1 (a INT PRIMARY KEY) ENGINE = innodb;
+# MDEV-515 takes X-lock on the table for the first insert.
+# So concurrent insert won't happen on the table
+INSERT INTO t1 VALUES(100);
 
 --sync_slave_with_master
 --source include/stop_slave.inc
diff --git a/mysql-test/suite/rpl/t/rpl_mariadb_slave_capability.test b/mysql-test/suite/rpl/t/rpl_mariadb_slave_capability.test
index 046a65f77db..19f2db32cb7 100644
--- a/mysql-test/suite/rpl/t/rpl_mariadb_slave_capability.test
+++ b/mysql-test/suite/rpl/t/rpl_mariadb_slave_capability.test
@@ -84,6 +84,9 @@ let $binlog_limit=7,5;
 
 --connection master
 CREATE TABLE t2 (a INT PRIMARY KEY) ENGINE=InnoDB;
+# MDEV-515 takes X-lock on the table for the first insert.
+# So concurrent insert won't happen on the table
+INSERT INTO t2 VALUES(100);
 let $binlog_file= query_get_value(SHOW MASTER STATUS, File, 1);
 let $binlog_start= query_get_value(SHOW MASTER STATUS, Position, 1);
 
diff --git a/mysql-test/suite/rpl/t/rpl_parallel_no_log_slave_updates.test b/mysql-test/suite/rpl/t/rpl_parallel_no_log_slave_updates.test
index 75db619d225..811d2bf24d9 100644
--- a/mysql-test/suite/rpl/t/rpl_parallel_no_log_slave_updates.test
+++ b/mysql-test/suite/rpl/t/rpl_parallel_no_log_slave_updates.test
@@ -30,6 +30,7 @@ CREATE FUNCTION foo(x INT, d1 VARCHAR(500), d2 VARCHAR(500))
 --delimiter ;
 SET sql_log_bin=1;
 CREATE TABLE t3 (a INT PRIMARY KEY, b INT) ENGINE=InnoDB;
+INSERT INTO t3 VALUES (100, 100);
 --save_master_pos
 
 --connection server_2
diff --git a/mysql-test/suite/rpl/t/rpl_parallel_retry.test b/mysql-test/suite/rpl/t/rpl_parallel_retry.test
index 55da54e3c8c..601999c4fe7 100644
--- a/mysql-test/suite/rpl/t/rpl_parallel_retry.test
+++ b/mysql-test/suite/rpl/t/rpl_parallel_retry.test
@@ -266,6 +266,9 @@ SELECT * FROM t1 WHERE a >= 100 ORDER BY a;
 CREATE TABLE t3 (a INT PRIMARY KEY, b INT, KEY b_idx(b)) ENGINE=InnoDB;
 INSERT INTO t3 VALUES (1,NULL), (2,2), (3,NULL), (4,4), (5, NULL), (6, 6);
 CREATE TABLE t4 (a INT PRIMARY KEY, b INT) ENGINE=InnoDB;
+# MDEV-515 takes X-lock on the table for the first insert.
+# So concurrent insert won't happen on the table
+INSERT INTO t4 VALUES(100, 100);
 
 # We need statement binlog format to be able to inject debug_sync statements
 # on the slave with calls to foo().
@@ -390,7 +393,9 @@ DROP function foo;
 
 --connection server_1
 CREATE TABLE t1 (a int PRIMARY KEY, b INT) ENGINE=InnoDB;
-
+# MDEV-515 takes X-lock on the table for the first insert.
+# So concurrent insert won't happen on the table
+INSERT INTO t1 VALUES(100, 100);
 
 # Replicate create-t1 and prepare to re-start slave in optimistic mode
 --sync_slave_with_master server_2
diff --git a/mysql-test/suite/rpl/t/rpl_semisync_ali_issues.test b/mysql-test/suite/rpl/t/rpl_semisync_ali_issues.test
index 52cd9e31753..8ad3bc76805 100644
--- a/mysql-test/suite/rpl/t/rpl_semisync_ali_issues.test
+++ b/mysql-test/suite/rpl/t/rpl_semisync_ali_issues.test
@@ -77,6 +77,10 @@ connection con1;
 select * from t1;
 truncate table t1;
 
+# MDEV-515 takes X-lock on the table
+# So concurrent DML won't happen on the table
+INSERT INTO  t1 VALUES (100);
+
 sync_slave_with_master;
 
 # Test more threads in one semisync queue
diff --git a/mysql-test/suite/rpl/t/rpl_xa_survive_disconnect.test b/mysql-test/suite/rpl/t/rpl_xa_survive_disconnect.test
index 28d189364d6..e4193a35583 100644
--- a/mysql-test/suite/rpl/t/rpl_xa_survive_disconnect.test
+++ b/mysql-test/suite/rpl/t/rpl_xa_survive_disconnect.test
@@ -30,6 +30,10 @@ CREATE DATABASE d2;
 CREATE TABLE d1.t (a INT) ENGINE=innodb;
 CREATE TABLE d2.t (a INT) ENGINE=innodb;
 
+# MDEV-515 takes X-lock on the table for the first insert.
+# So concurrent DML won't happen on the table
+INSERT INTO d1.t VALUES(100);
+
 connect (master_conn1, 127.0.0.1,root,,test,$MASTER_MYPORT,);
 --let $conn_id=`SELECT connection_id()`
 SET @@session.binlog_format= statement;
diff --git a/mysql-test/suite/sys_vars/r/identity_func.result b/mysql-test/suite/sys_vars/r/identity_func.result
index 47a6fa856c5..2e064532228 100644
--- a/mysql-test/suite/sys_vars/r/identity_func.result
+++ b/mysql-test/suite/sys_vars/r/identity_func.result
@@ -14,6 +14,8 @@ id INT NOT NULL auto_increment,
 PRIMARY KEY (id),
 name VARCHAR(30)
 ) ENGINE = INNODB;
+INSERT INTO t1 VALUES(100, "MDEV-515");
+INSERT INTO t2 VALUES(100, "MDEV-515");
 '#--------------------FN_DYNVARS_035_01-------------------------#'
 ## It should be zero ## 
 SELECT @@identity = 0;
@@ -29,40 +31,47 @@ INSERT into t1(name) values('Record_3');
 ## Verifying total values in t1 ## 
 SELECT @@identity from t1;
 @@identity
-3
-3
-3
+103
+103
+103
+103
 ## Now inserting some data in table t2 ## 
 INSERT into t2(name) values('Record_1');
 ## Verifying total values in t2 ## 
 SELECT @@identity from t2;
 @@identity
-1
+101
+101
 '#--------------------FN_DYNVARS_035_02-------------------------#'
 connect  test_con2, localhost, root,,;
 connection test_con2;
 SELECT * from t1;
 id	name
+100	MDEV-515
 ## Verifying total values in t1 ## 
 SELECT @@identity from t1;
 @@identity
+0
 ## Verifying total values in t2 ## 
 SELECT @@identity from t2;
 @@identity
+0
 ## Inserting some more records in table t1 ##
 INSERT into t1(name) values('Record_1_1');
 INSERT into t1(name) values('Record_1_2');
 ## Verifying total values in t1 ## 
 SELECT @@identity from t1;
 @@identity
-5
-5
+105
+105
+105
 ## Inserting row in table t2 ##
 INSERT into t2(name) values('Record_1_3');
 ## Verifying total values in t2 ## 
 SELECT @@identity from t2;
 @@identity
-2
+102
+102
 '#--------------------FN_DYNVARS_035_03-------------------------#'
 connection test_con1;
 ## Commiting rows added in test_con1 ##
@@ -70,38 +79,43 @@ COMMIT;
 ## Verifying records in both tables ##
 SELECT * from t1;
 id	name
-1	Record_1
-2	Record_2
-3	Record_3
-4	Record_1_1
-5	Record_1_2
+100	MDEV-515
+101	Record_1
+102	Record_2
+103	Record_3
+104	Record_1_1
+105	Record_1_2
 SELECT * from t2;
 id	name
-1	Record_1
-2	Record_1_3
+100	MDEV-515
+101	Record_1
+102	Record_1_3
 ## Verifying total values in t1 after commiting data ## 
 SELECT @@identity from t1;
 @@identity
-1
-1
-1
-1
-1
+101
+101
+101
+101
+101
+101
 ## Verifying total values in t2 after commiting data ## 
 SELECT @@identity from t2;
 @@identity
-1
-1
+101
+101
+101
 INSERT into t1(name) values('Record_4');
 ## Now verifying value of variable after inserting 1 row in this connection ## 
 SELECT @@identity from t1;
 @@identity
-6
-6
-6
-6
-6
-6
+106
+106
+106
+106
+106
+106
+106
 ## Dropping tables t1 & t2 ##
 drop table t1, t2;
 disconnect test_con1;
diff --git a/mysql-test/suite/sys_vars/t/identity_func.test b/mysql-test/suite/sys_vars/t/identity_func.test
index e3bbaa1d3a2..7398b5d0b63 100644
--- a/mysql-test/suite/sys_vars/t/identity_func.test
+++ b/mysql-test/suite/sys_vars/t/identity_func.test
@@ -46,6 +46,11 @@ PRIMARY KEY (id),
 name VARCHAR(30)
 ) ENGINE = INNODB;
 
+# MDEV-515 takes X-lock on the table for the first insert
+# So concurrent insert won't happen on the table
+INSERT INTO t1 VALUES(100, "MDEV-515");
+INSERT INTO t2 VALUES(100, "MDEV-515");
+
 --echo '#--------------------FN_DYNVARS_035_01-------------------------#'
 ###############################################
 #    Verifying initial value of identity.     #     
diff --git a/mysql-test/suite/versioning/r/trx_id.result b/mysql-test/suite/versioning/r/trx_id.result
index e8d550cefa6..22e224edc38 100644
--- a/mysql-test/suite/versioning/r/trx_id.result
+++ b/mysql-test/suite/versioning/r/trx_id.result
@@ -81,6 +81,7 @@ sys_start bigint(20) unsigned as row start invisible,
 sys_end bigint(20) unsigned as row end invisible,
 period for system_time (sys_start, sys_end)
 ) with system versioning;
+INSERT INTO t1 VALUES(100);
 set transaction isolation level read committed;
 start transaction;
 insert into t1 values (1);
@@ -121,25 +122,31 @@ select @ts1 < @ts2, @ts2 < @ts3;
 # MVCC is resolved
 select * from t1 for system_time as of transaction @trx_id1;
 x
+100
 1
 2
 3
 select * from t1 for system_time as of timestamp @ts1;
 x
+100
 3
 select * from t1 for system_time as of transaction @trx_id2;
 x
+100
 2
 3
 select * from t1 for system_time as of timestamp @ts2;
 x
+100
 2
 3
 select * from t1 for system_time as of transaction @trx_id3;
 x
+100
 3
 select * from t1 for system_time as of timestamp @ts3;
 x
+100
 1
 2
 3
diff --git a/mysql-test/suite/versioning/t/trx_id.test b/mysql-test/suite/versioning/t/trx_id.test
index 216f2082bed..2e779e4160c 100644
--- a/mysql-test/suite/versioning/t/trx_id.test
+++ b/mysql-test/suite/versioning/t/trx_id.test
@@ -81,6 +81,10 @@ create or replace table t1 (
   period for system_time (sys_start, sys_end)
 ) with system versioning;
 
+# MDEV-515 takes X-lock on the table for the first insert
+# So concurrent DML won't happen on the table 
+INSERT INTO t1 VALUES(100);
+
 set transaction isolation level read committed;
 start transaction;
 insert into t1 values (1);
diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index 470f59fe15f..291d8f92cb8 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -9146,6 +9146,8 @@ int ha_partition::extra(enum ha_extra_function operation)
   case HA_EXTRA_END_ALTER_COPY:
   case HA_EXTRA_FAKE_START_STMT:
     DBUG_RETURN(loop_partitions(extra_cb, &operation));
+  case HA_EXTRA_IGNORE_INSERT:
+    DBUG_RETURN(loop_partitions(extra_cb, &operation));
   default:
   {
     /* Temporary crash to discover what is wrong */
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index f26ee27df42..697ac87bd53 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -2114,6 +2114,8 @@ int write_record(THD *thd, TABLE *table, COPY_INFO *info, select_result *sink)
     goto after_trg_or_ignored_err;
   }
 
+  if (info->handle_duplicates == DUP_ERROR && info->ignore)
+     table->file->extra(HA_EXTRA_IGNORE_INSERT);
 after_trg_n_copied_inc:
   info->copied++;
   thd->record_first_successful_insert_id_in_cur_stmt(table->file->insert_id_for_cur_row);
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 318c01bb1dc..f1f57471253 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -11499,6 +11499,8 @@ copy_data_between_tables(THD *thd, TABLE *from, TABLE *to,
     }
     else
     {
+      if (ignore)
+        to->file->extra(HA_EXTRA_IGNORE_INSERT);
       DEBUG_SYNC(thd, "copy_data_between_tables_before");
       found_count++;
       mysql_stage_set_work_completed(thd->m_stage_progress_psi, found_count);
diff --git a/storage/innobase/btr/btr0btr.cc b/storage/innobase/btr/btr0btr.cc
index ded701af612..a1929ac9562 100644
--- a/storage/innobase/btr/btr0btr.cc
+++ b/storage/innobase/btr/btr0btr.cc
@@ -975,6 +975,39 @@ btr_free_root_check(
 	return(block);
 }
 
+void
+btr_root_page_init(buf_block_t *block, index_id_t index_id,
+                   dict_index_t *index, mtr_t *mtr)
+{
+  constexpr uint16_t field = PAGE_HEADER + PAGE_INDEX_ID;
+  byte* page_index_id = my_assume_aligned<2>(field + block->frame);
+
+  /* Create a new index page on the allocated segment page */
+  if (UNIV_LIKELY_NULL(block->page.zip.data))
+  {
+    mach_write_to_8(page_index_id, index_id);
+    ut_ad(!page_has_siblings(block->page.zip.data));
+    page_create_zip(block, index, 0, 0, mtr);
+  }
+  else
+  {
+    page_create(block, mtr, index && index->table->not_redundant());
+    if (index && index->is_spatial())
+    {
+      static_assert(((FIL_PAGE_INDEX & 0xff00) | byte(FIL_PAGE_RTREE))
+                    == FIL_PAGE_RTREE, "compatibility");
+      mtr->write<1>(*block, FIL_PAGE_TYPE + 1 + block->frame,
+                    byte(FIL_PAGE_RTREE));
+      if (mach_read_from_8(block->frame + FIL_RTREE_SPLIT_SEQ_NUM))
+        mtr->memset(block, FIL_RTREE_SPLIT_SEQ_NUM, 8, 0);
+    }
+    /* Set the level of the new index page */
+    mtr->write<2,mtr_t::MAYBE_NOP>(
+        *block, PAGE_HEADER + PAGE_LEVEL + block->frame, 0U);
+    mtr->write<8,mtr_t::MAYBE_NOP>(*block, page_index_id, index_id);
+  }
+}
+
 /** Create the root node for a new index tree.
 @param[in]	type			type of the index
 @param[in]	index_id		index id
@@ -1047,36 +1080,7 @@ btr_create(
 
 	ut_ad(!page_has_siblings(block->frame));
 
-	constexpr uint16_t field = PAGE_HEADER + PAGE_INDEX_ID;
-
-	byte* page_index_id = my_assume_aligned<2>(field + block->frame);
-
-	/* Create a new index page on the allocated segment page */
-	if (UNIV_LIKELY_NULL(block->page.zip.data)) {
-		mach_write_to_8(page_index_id, index_id);
-		ut_ad(!page_has_siblings(block->page.zip.data));
-		page_create_zip(block, index, 0, 0, mtr);
-	} else {
-		page_create(block, mtr,
-			    index && index->table->not_redundant());
-		if (index && index->is_spatial()) {
-			static_assert(((FIL_PAGE_INDEX & 0xff00)
-				       | byte(FIL_PAGE_RTREE))
-				      == FIL_PAGE_RTREE, "compatibility");
-			mtr->write<1>(*block, FIL_PAGE_TYPE + 1 + block->frame,
-				      byte(FIL_PAGE_RTREE));
-			if (mach_read_from_8(block->frame
-					     + FIL_RTREE_SPLIT_SEQ_NUM)) {
-				mtr->memset(block, FIL_RTREE_SPLIT_SEQ_NUM,
-					    8, 0);
-			}
-		}
-		/* Set the level of the new index page */
-		mtr->write<2,mtr_t::MAYBE_NOP>(*block, PAGE_HEADER + PAGE_LEVEL
-					       + block->frame, 0U);
-		mtr->write<8,mtr_t::MAYBE_NOP>(*block, page_index_id,
-					       index_id);
-	}
+	btr_root_page_init(block, index_id, index, mtr);
 
 	/* We reset the free bits for the page in a separate
 	mini-transaction to allow creation of several trees in the
@@ -1104,7 +1108,6 @@ btr_create(
 this by calling btr_free_root.
 @param[in,out]	block		root page
 @param[in]	log_mode	mtr logging mode */
-static
 void
 btr_free_but_not_root(
 	buf_block_t*	block,
diff --git a/storage/innobase/btr/btr0cur.cc b/storage/innobase/btr/btr0cur.cc
index 2f4db18ca50..dcc56ceae14 100644
--- a/storage/innobase/btr/btr0cur.cc
+++ b/storage/innobase/btr/btr0cur.cc
@@ -67,6 +67,7 @@ Created 10/16/1994 Heikki Tuuri
 #include "srv0start.h"
 #include "mysql_com.h"
 #include "dict0stats.h"
+#include "row0ins.h"
 
 /** Buffered B-tree operation types, introduced as part of delete buffering. */
 enum btr_op_t {
@@ -3214,7 +3215,6 @@ btr_cur_ins_lock_and_undo(
 	dberr_t		err = DB_SUCCESS;
 	rec_t*		rec;
 	roll_ptr_t	roll_ptr;
-
 	/* Check if we have to wait for a lock: enqueue an explicit lock
 	request if yes */
 
@@ -3225,6 +3225,14 @@ btr_cur_ins_lock_and_undo(
 	      || dict_index_is_clust(index)
 	      || (flags & BTR_CREATE_FLAG));
 	ut_ad(mtr->is_named_space(index->table->space));
+	bool write_empty = thr
+			   && thr->run_node
+			   && que_node_get_type(thr->run_node)
+				== QUE_NODE_INSERT
+			   && static_cast<ins_node_t*>(
+				thr->run_node)->bulk_insert
+			   && !thr_get_trx(thr)->allow_insert_undo(
+				index->table);
 
 	/* Check if there is predicate or GAP lock preventing the insertion */
 	if (!(flags & BTR_NO_LOCKING_FLAG)) {
@@ -3273,6 +3281,20 @@ upd_sys:
 						    NULL, 0, NULL, NULL,
 						    &roll_ptr);
 		if (err == DB_SUCCESS) {
+
+			if (index->table->no_rollback()) {
+			} else if (write_empty) {
+				roll_ptr = roll_ptr_t(1)
+					<< ROLL_PTR_INSERT_FLAG_POS;
+			} else {
+				trx_t* trx = thr_get_trx(thr);
+				dfield_t *r = dtuple_get_nth_field(
+					entry, index->db_trx_id());
+				trx_write_trx_id(
+					static_cast<byte*>(r->data),
+					trx->id);
+			}
+
 			goto upd_sys;
 		}
 	}
@@ -3510,7 +3532,9 @@ fail_err:
 				ut_ad(thr->graph->trx->id
 				      == trx_read_trx_id(
 					      static_cast<const byte*>(
-						      trx_id->data)));
+						      trx_id->data))
+				      || static_cast<ins_node_t*>(
+					     thr->run_node)->bulk_insert);
 			}
 		}
 #endif
@@ -3566,7 +3590,7 @@ fail_err:
 	} else if (entry->info_bits & REC_INFO_MIN_REC_FLAG) {
 		ut_ad(entry->is_metadata());
 		ut_ad(index->is_instant());
-		ut_ad(flags == BTR_NO_LOCKING_FLAG);
+		ut_ad(flags & BTR_NO_LOCKING_FLAG);
 	} else {
 		srw_lock* ahi_latch = btr_search_sys.get_latch(*index);
 		if (!reorg && cursor->flag == BTR_CUR_HASH) {
@@ -5505,7 +5529,13 @@ btr_cur_optimistic_delete_func(
 			if (index->is_instant()) {
 				/* MDEV-17383: free metadata BLOBs! */
 				index->clear_instant_alter();
+				index->table->remove_bulk_trx();
 			}
+
+			if (dict_index_is_clust(index)) {
+				index->table->remove_bulk_trx();
+			}
+
 			page_cur_set_after_last(block,
 						btr_cur_get_page_cur(cursor));
 			goto func_exit;
@@ -5723,6 +5753,11 @@ btr_cur_pessimistic_delete(
 					/* MDEV-17383: free metadata BLOBs! */
 					index->clear_instant_alter();
 				}
+
+				if (dict_index_is_clust(index)) {
+					index->table->remove_bulk_trx();
+				}
+
 				page_cur_set_after_last(
 					block,
 					btr_cur_get_page_cur(cursor));
diff --git a/storage/innobase/dict/dict0mem.cc b/storage/innobase/dict/dict0mem.cc
index c9d86139741..e360a27cc9c 100644
--- a/storage/innobase/dict/dict0mem.cc
+++ b/storage/innobase/dict/dict0mem.cc
@@ -38,6 +38,7 @@ Created 1/8/1996 Heikki Tuuri
 #include "row0row.h"
 #include "sql_string.h"
 #include <iostream>
+#include "row0log.h"
 
 #define	DICT_HEAP_SIZE		100	/*!< initial memory heap size when
 					creating a table or index object */
@@ -1392,3 +1393,39 @@ dict_index_t::vers_history_row(
 	}
 	return(error);
 }
+
+void dict_table_t::empty_table(que_thr_t *thr)
+{
+  mtr_t mtr;
+  bool rebuild= false;
+  for (dict_index_t* index= UT_LIST_GET_FIRST(indexes);
+       index != NULL; index= UT_LIST_GET_NEXT(indexes, index))
+  {
+    if (index->online_status == ONLINE_INDEX_ABORTED
+        || index->online_status == ONLINE_INDEX_ABORTED_DROPPED)
+      continue;
+
+    if (index->type & DICT_FTS)
+      continue;
+
+    if (index->online_status == ONLINE_INDEX_CREATION)
+    {
+      if (dict_index_is_clust(index))
+      {
+        row_log_table_empty(index);
+        rebuild= true;
+      }
+      else if (!rebuild)
+      {
+        mtr.start();
+        mtr_s_lock_index(index, &mtr);
+        row_log_online_op(index, nullptr, 0);
+        mtr.commit();
+      }
+    }
+
+    index->empty(thr);
+  }
+
+  remove_bulk_trx();
+}
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index c4083f8aeb9..80287836ecc 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -15140,6 +15140,14 @@ ha_innobase::extra(
 		trx_register_for_2pc(m_prebuilt->trx);
 		m_prebuilt->sql_stat_start = true;
 		break;
+	case HA_EXTRA_IGNORE_INSERT:
+	{
+		ins_node_t* node = m_prebuilt->ins_node;
+		if (node && node->bulk_insert) {
+			m_prebuilt->trx->set_allow_insert_undo(node->table);
+		}
+	}
+		break;
 	default:/* Do nothing */
 		;
 	}
diff --git a/storage/innobase/handler/handler0alter.cc b/storage/innobase/handler/handler0alter.cc
index 8dceb3a21f5..ba0688e16d3 100644
--- a/storage/innobase/handler/handler0alter.cc
+++ b/storage/innobase/handler/handler0alter.cc
@@ -11463,3 +11463,27 @@ ib_sequence_t::operator++(int) UNIV_NOTHROW
 
 	return(current);
 }
+
+void dict_index_t::empty(que_thr_t *thr)
+{
+  mtr_t mtr;
+  mtr.start();
+  mtr.set_named_space_id(table->space->id);
+  /* Free the indexes */
+  buf_block_t* root_block= buf_page_get(
+    page_id_t(table->space->id, page),
+    table->space->zip_size(), RW_X_LATCH, &mtr);
+  if (root_block)
+    btr_free_but_not_root(root_block, mtr.get_log_mode());
+
+  mtr.memset(root_block, PAGE_HEADER + PAGE_BTR_SEG_LEAF,
+             FSEG_HEADER_SIZE, 0);
+  if (!fseg_create(table->space, PAGE_HEADER + PAGE_BTR_SEG_LEAF,
+                   &mtr, false, root_block))
+  {
+    ut_ad(0);
+  }
+
+  btr_root_page_init(root_block, id, this, &mtr);
+  mtr.commit();
+}
diff --git a/storage/innobase/include/btr0btr.h b/storage/innobase/include/btr0btr.h
index 3d5fd015398..281ea2be7a8 100644
--- a/storage/innobase/include/btr0btr.h
+++ b/storage/innobase/include/btr0btr.h
@@ -313,6 +313,14 @@ btr_node_ptr_get_child_page_no(
 	const rec_offs*	offsets)/*!< in: array returned by rec_get_offsets() */
 	MY_ATTRIBUTE((warn_unused_result));
 
+/** Initialize the root page of the b-tree
+@param[in,out]  block           root block
+@param[in]      index_id        index id
+@param[in]      index           index of root page
+@param[in,out]  mtr             mini-transaction */
+void btr_root_page_init(buf_block_t *block, index_id_t index_id,
+                        dict_index_t *index, mtr_t *mtr);
+
 /** Create the root node for a new index tree.
 @param[in]	type			type of the index
 @param[in,out]	space			tablespace where created
@@ -736,4 +744,11 @@ Global variable controlling if scrubbing should be performed */
 extern my_bool srv_immediate_scrub_data_uncompressed;
 extern Atomic_counter<uint32_t> btr_validate_index_running;
 
+/** Free a B-tree except the root page. The root page MUST be freed after
+this by calling btr_free_root.
+@param[in,out]	block		root page
+@param[in]	log_mode	mtr logging mode */
+void
+btr_free_but_not_root(buf_block_t *block, mtr_log_t log_mode);
+
 #endif
diff --git a/storage/innobase/include/dict0mem.h b/storage/innobase/include/dict0mem.h
index 8ad293f6a9a..660cb24a855 100644
--- a/storage/innobase/include/dict0mem.h
+++ b/storage/innobase/include/dict0mem.h
@@ -1368,6 +1368,9 @@ public:
 	everything in overflow) size of the longest possible row and index
 	of a field which made index records too big to fit on a page.*/
 	inline record_size_info_t record_size_info() const;
+
+	/** Empty the index content and reinitialize the root page */
+	void empty(que_thr_t *thr);
 };
 
 /** Detach a virtual column from an index.
@@ -1950,6 +1953,14 @@ struct dict_table_t {
 			char (&tbl_name)[NAME_LEN + 1],
 			size_t *db_name_len, size_t *tbl_name_len) const;
 
+  /** Empty the table */
+  void empty_table(que_thr_t *thr);
+
+  void remove_bulk_trx()
+  {
+    bulk_trx_id_= 0;
+  }
+
 private:
 	/** Initialize instant->field_map.
 	@param[in]	table	table definition to copy from */
@@ -2299,6 +2310,9 @@ private:
 	itself check the number of open handles at DROP. */
 	Atomic_counter<uint32_t>		n_ref_count;
 
+	/** Trx id of bulk insert operation. This is under the
+	protection of exclusive lock of table object */
+	Atomic_relaxed<trx_id_t>		bulk_trx_id_;
 public:
 	/** List of locks on the table. Protected by lock_sys.mutex. */
 	table_lock_list_t			locks;
@@ -2316,6 +2330,16 @@ public:
 	/** mysql_row_templ_t for base columns used for compute the virtual
 	columns */
 	dict_vcol_templ_t*			vc_templ;
+
+  trx_id_t bulk_trx_id() const
+  {
+    return bulk_trx_id_;
+  }
+
+  void set_bulk_trx_id(trx_id_t trx_id)
+  {
+    bulk_trx_id_= trx_id;
+  }
 };
 
 inline void dict_index_t::set_modified(mtr_t& mtr) const
diff --git a/storage/innobase/include/lock0lock.h b/storage/innobase/include/lock0lock.h
index e197f383201..75c420c61d2 100644
--- a/storage/innobase/include/lock0lock.h
+++ b/storage/innobase/include/lock0lock.h
@@ -438,6 +438,12 @@ lock_table_ix_resurrect(
 	dict_table_t*	table,	/*!< in/out: table */
 	trx_t*		trx);	/*!< in/out: transaction */
 
+/** Creates a table X lock object for a resurrected transaction.
+@param	table	table to be X-locked
+@param	trx	transaction */
+void
+lock_table_x_resurrect(dict_table_t *table, trx_t *trx);
+
 /** Sets a lock on a table based on the given mode.
 @param[in]	table	table to lock
 @param[in,out]	trx	transaction
diff --git a/storage/innobase/include/row0ins.h b/storage/innobase/include/row0ins.h
index 34427dc6dc7..2609f563e9f 100644
--- a/storage/innobase/include/row0ins.h
+++ b/storage/innobase/include/row0ins.h
@@ -174,7 +174,8 @@ struct ins_node_t
 		row(NULL), table(table), select(NULL), values_list(NULL),
 		state(INS_NODE_SET_IX_LOCK), index(NULL),
 		entry_list(), entry(entry_list.end()),
-		trx_id(0), entry_sys_heap(mem_heap_create(128))
+		trx_id(0), entry_sys_heap(mem_heap_create(128)),
+		bulk_insert(false)
 	{
 	}
 	que_common_t common;	 /*!< node type: QUE_NODE_INSERT */
@@ -205,7 +206,10 @@ struct ins_node_t
 				entry_list and sys fields are stored here;
 				if this is NULL, entry list should be created
 				and buffers for sys fields in row allocated */
-        void vers_update_end(row_prebuilt_t *prebuilt, bool history_row);
+	/** Bulk insert enabled for this table */
+	bool		bulk_insert;
+
+	void vers_update_end(row_prebuilt_t *prebuilt, bool history_row);
 };
 
 /** Create an insert object.
diff --git a/storage/innobase/include/row0log.h b/storage/innobase/include/row0log.h
index 5ec4b9c1103..b046a19ac7e 100644
--- a/storage/innobase/include/row0log.h
+++ b/storage/innobase/include/row0log.h
@@ -102,9 +102,8 @@ row_log_online_op(
 /*==============*/
 	dict_index_t*	index,	/*!< in/out: index, S or X latched */
 	const dtuple_t*	tuple,	/*!< in: index tuple */
-	trx_id_t	trx_id)	/*!< in: transaction ID for insert,
+	trx_id_t	trx_id);/*!< in: transaction ID for insert,
 				or 0 for delete */
-	ATTRIBUTE_COLD __attribute__((nonnull));
 
 /******************************************************//**
 Gets the error status of the online index rebuild log.
@@ -258,6 +257,11 @@ row_log_estimate_work(
 	const dict_index_t*	index);
 #endif /* HAVE_PSI_STAGE_INTERFACE */
 
+/** Logs an empty operation of the table which means it should empty
+the table.
+@param index	clustered index */
+void row_log_table_empty(dict_index_t *index);
+
 #include "row0log.ic"
 
 #endif /* row0log.h */
diff --git a/storage/innobase/include/trx0rec.h b/storage/innobase/include/trx0rec.h
index 9aeff6312f6..d802deca75f 100644
--- a/storage/innobase/include/trx0rec.h
+++ b/storage/innobase/include/trx0rec.h
@@ -295,7 +295,9 @@ record */
 					a not delete marked record; also the
 					fields of the record can change */
 #define	TRX_UNDO_DEL_MARK_REC	14	/* delete marking of a record; fields
-					do not change */
+					   do not change */
+#define TRX_UNDO_EMPTY		15      /* Empty the table */
+
 #define	TRX_UNDO_CMPL_INFO_MULT	16U	/* compilation info is multiplied by
 					this and ORed to the type above */
 #define	TRX_UNDO_UPD_EXTERN	128U	/* This bit can be ORed to type_cmpl
diff --git a/storage/innobase/include/trx0trx.h b/storage/innobase/include/trx0trx.h
index 2b06aa78c6e..d785e3634d4 100644
--- a/storage/innobase/include/trx0trx.h
+++ b/storage/innobase/include/trx0trx.h
@@ -575,11 +575,15 @@ class trx_mod_table_time_t
 	table was never modified in a transaction. */
 	static const undo_no_t UNVERSIONED = IB_ID_MAX;
 
+	/** Bulk insert is happening for the table. Allow undo for the
+	consecutive inserts. */
+	bool		bulk_insert_undo;
 public:
 	/** Constructor
 	@param[in]	rows	number of modified rows so far */
 	trx_mod_table_time_t(undo_no_t rows)
-		: first(rows), first_versioned(UNVERSIONED) {}
+		: first(rows), first_versioned(UNVERSIONED),
+		  bulk_insert_undo(false) {}
 
 #ifdef UNIV_DEBUG
 	/** Validation
@@ -618,6 +622,10 @@ public:
 
 		return false;
 	}
+
+	void set_bulk_insert_undo() { bulk_insert_undo= true; }
+
+	bool bulk_insert() { return bulk_insert_undo; }
 };
 
 /** Collection of persistent tables and their first modification
@@ -1095,6 +1103,27 @@ public:
     ut_ad(dict_operation == TRX_DICT_OP_NONE);
   }
 
+  bool allow_insert_undo(dict_table_t *table)
+  {
+    auto it= mod_tables.find(table);
+    if (it == mod_tables.end())
+      return false;
+    return it->second.bulk_insert();
+  }
+
+  void set_allow_insert_undo(dict_table_t *table)
+  {
+    auto it= mod_tables.find(table);
+    if (it != mod_tables.end())
+      it->second.set_bulk_insert_undo();
+  }
+
+  void set_allow_insert_undo()
+  {
+    for (auto it = mod_tables.begin();
+	 it != mod_tables.end(); it++)
+      it->second.set_bulk_insert_undo();
+  }
 
 private:
 	/** Assign a rollback segment for modifying temporary tables.
diff --git a/storage/innobase/lock/lock0lock.cc b/storage/innobase/lock/lock0lock.cc
index 66644865309..fa97728e37d 100644
--- a/storage/innobase/lock/lock0lock.cc
+++ b/storage/innobase/lock/lock0lock.cc
@@ -3604,6 +3604,25 @@ lock_table_ix_resurrect(
 	mutex->wr_unlock();
 }
 
+void
+lock_table_x_resurrect(dict_table_t *table,trx_t *trx)
+{
+  ut_ad(trx->is_recovered);
+  if (lock_table_has(trx, table, LOCK_X))
+    return;
+
+  auto mutex= &trx->mutex;
+  lock_sys.mutex_lock();
+  /* We have to check if the new lock is compatible with any locks
+  other transactions have in the table lock queue. */
+  ut_ad(!lock_table_other_has_incompatible(trx, LOCK_WAIT, table, LOCK_X));
+
+  mutex->wr_lock();
+  lock_table_create(table, LOCK_X, trx);
+  lock_sys.mutex_unlock();
+  mutex->wr_unlock();
+}
+
 /*********************************************************************//**
 Checks if a waiting table lock request still has to wait in a queue.
 @return TRUE if still has to wait */
diff --git a/storage/innobase/row/row0ins.cc b/storage/innobase/row/row0ins.cc
index 9835e6b0afe..d038f660603 100644
--- a/storage/innobase/row/row0ins.cc
+++ b/storage/innobase/row/row0ins.cc
@@ -2552,14 +2552,18 @@ row_ins_clust_index_entry_low(
 	rec_offs	offsets_[REC_OFFS_NORMAL_SIZE];
 	rec_offs*	offsets         = offsets_;
 	rec_offs_init(offsets_);
+	trx_t*		trx	= thr_get_trx(thr);
+	buf_block_t*	block;
 
 	DBUG_ENTER("row_ins_clust_index_entry_low");
 
+	DEBUG_SYNC_C("row_ins_clust_index_entry_low_enter");
+
 	ut_ad(dict_index_is_clust(index));
 	ut_ad(!dict_index_is_unique(index)
 	      || n_uniq == dict_index_get_n_unique(index));
 	ut_ad(!n_uniq || n_uniq == dict_index_get_n_unique(index));
-	ut_ad(!thr_get_trx(thr)->in_rollback);
+	ut_ad(!trx->in_rollback);
 
 	mtr_start(&mtr);
 
@@ -2631,6 +2635,31 @@ row_ins_clust_index_entry_low(
 	}
 #endif /* UNIV_DEBUG */
 
+	block = btr_cur_get_block(cursor);
+
+	if (block->page.id().page_no() == index->page
+	    && !(flags & BTR_NO_UNDO_LOG_FLAG)
+	    && !index->table->is_temporary()
+	    && !entry->is_metadata() && !trx->duplicates
+	    && !trx->ddl && !trx->internal
+	    && page_is_empty(block->frame)) {
+
+		DEBUG_SYNC_C("empty_root_page_insert");
+
+		err = lock_table(0, index->table, LOCK_X, thr);
+
+		if (err != DB_SUCCESS) {
+			mtr_commit(&mtr);
+			trx->error_state = err;
+			goto func_exit;
+		}
+
+		index->table->set_bulk_trx_id(trx->id);
+		ins_node_t *run_node= static_cast<ins_node_t*>(
+					thr->run_node);
+		run_node->bulk_insert= true;
+	}
+
 	if (UNIV_UNLIKELY(entry->info_bits != 0)) {
 		ut_ad(entry->is_metadata());
 		ut_ad(flags == BTR_NO_LOCKING_FLAG);
@@ -2641,7 +2670,7 @@ row_ins_clust_index_entry_low(
 
 		if (rec_get_info_bits(rec, page_rec_is_comp(rec))
 		    & REC_INFO_MIN_REC_FLAG) {
-			thr_get_trx(thr)->error_info = index;
+			trx->error_info = index;
 			err = DB_DUPLICATE_KEY;
 			goto err_exit;
 		}
@@ -2674,7 +2703,7 @@ row_ins_clust_index_entry_low(
 				/* fall through */
 			case DB_SUCCESS_LOCKED_REC:
 			case DB_DUPLICATE_KEY:
-				thr_get_trx(thr)->error_info = cursor->index;
+				trx->error_info = cursor->index;
 			}
 		} else {
 			/* Note that the following may return also
@@ -2758,7 +2787,7 @@ do_insert:
 				log_write_up_to(mtr.commit_lsn(), true););
 			err = row_ins_index_entry_big_rec(
 				entry, big_rec, offsets, &offsets_heap, index,
-				thr_get_trx(thr)->mysql_thd);
+				trx->mysql_thd);
 			dtuple_convert_back_big_rec(index, entry, big_rec);
 		} else {
 			if (err == DB_SUCCESS
@@ -3693,10 +3722,6 @@ row_ins_step(
 		goto do_insert;
 	}
 
-	if (UNIV_LIKELY(!node->table->skip_alter_undo)) {
-		trx_write_trx_id(&node->sys_buf[DATA_ROW_ID_LEN], trx->id);
-	}
-
 	if (node->state == INS_NODE_SET_IX_LOCK) {
 
 		node->state = INS_NODE_ALLOC_ROW_ID;
diff --git a/storage/innobase/row/row0log.cc b/storage/innobase/row/row0log.cc
index e4f7be83940..051b99d928b 100644
--- a/storage/innobase/row/row0log.cc
+++ b/storage/innobase/row/row0log.cc
@@ -54,7 +54,9 @@ enum row_tab_op {
 	/** Update a record in place */
 	ROW_T_UPDATE,
 	/** Delete (purge) a record */
-	ROW_T_DELETE
+	ROW_T_DELETE,
+	/** Empty the table */
+	ROW_T_EMPTY
 };
 
 /** Index record modification operations during online index creation */
@@ -62,7 +64,9 @@ enum row_op {
 	/** Insert a record */
 	ROW_OP_INSERT = 0x61,
 	/** Delete a record */
-	ROW_OP_DELETE
+	ROW_OP_DELETE,
+	/** Empy the index */
+	ROW_OP_EMPTY
 };
 
 /** Size of the modification log entry header, in bytes */
@@ -339,8 +343,8 @@ row_log_online_op(
 	ulint		avail_size;
 	row_log_t*	log;
 
-	ut_ad(dtuple_validate(tuple));
-	ut_ad(dtuple_get_n_fields(tuple) == dict_index_get_n_fields(index));
+	ut_ad(!tuple || dtuple_validate(tuple));
+	ut_ad(!tuple || dtuple_get_n_fields(tuple) == dict_index_get_n_fields(index));
 	ut_ad(index->lock.have_x() || index->lock.have_s());
 
 	if (index->is_corrupted()) {
@@ -353,14 +357,19 @@ row_log_online_op(
 	row_merge_buf_encode(), because here we do not encode
 	extra_size+1 (and reserve 0 as the end-of-chunk marker). */
 
-	size = rec_get_converted_size_temp(
-		index, tuple->fields, tuple->n_fields, &extra_size);
-	ut_ad(size >= extra_size);
-	ut_ad(size <= sizeof log->tail.buf);
+	if (!tuple) {
+		mrec_size = 4;
+		extra_size = 0;
+	} else {
+		size = rec_get_converted_size_temp(
+			index, tuple->fields, tuple->n_fields, &extra_size);
+		ut_ad(size >= extra_size);
+		ut_ad(size <= sizeof log->tail.buf);
 
-	mrec_size = ROW_LOG_HEADER_SIZE
-		+ (extra_size >= 0x80) + size
-		+ (trx_id ? DATA_TRX_ID_LEN : 0);
+		mrec_size = ROW_LOG_HEADER_SIZE
+			+ (extra_size >= 0x80) + size
+			+ (trx_id ? DATA_TRX_ID_LEN : 0);
+	}
 
 	log = index->online_log;
 	mysql_mutex_lock(&log->mutex);
@@ -389,6 +398,8 @@ row_log_online_op(
 		*b++ = ROW_OP_INSERT;
 		trx_write_trx_id(b, trx_id);
 		b += DATA_TRX_ID_LEN;
+	} else if (tuple == nullptr) {
+		*b++ = ROW_OP_EMPTY;
 	} else {
 		*b++ = ROW_OP_DELETE;
 	}
@@ -401,9 +412,15 @@ row_log_online_op(
 		*b++ = (byte) extra_size;
 	}
 
-	rec_convert_dtuple_to_temp(
-		b + extra_size, index, tuple->fields, tuple->n_fields);
-	b += size;
+	if (tuple) {
+		rec_convert_dtuple_to_temp(
+			b + extra_size, index, tuple->fields,
+			tuple->n_fields);
+		b += size;
+	} else {
+		*b++ = 0;
+		*b++ = 0;
+	}
 
 	if (mrec_size >= avail_size) {
 		const os_offset_t	byte_offset
@@ -2387,6 +2404,18 @@ func_exit_committed:
 	goto func_exit;
 }
 
+/** Applies the empty table to a table that was rebuilt.
+@param  index   clustered index
+@retrun success if index gets emptied */
+static
+dberr_t
+row_log_table_apply_empty(dict_index_t* index, que_thr_t *thr)
+{
+  dict_table_t* new_table= index->online_log->table;
+  new_table->empty_table(thr);
+  return DB_SUCCESS;
+}
+
 /******************************************************//**
 Applies an operation to a table that was rebuilt.
 @return NULL on failure (mrec corruption) or when out of data;
@@ -2422,11 +2451,6 @@ row_log_table_apply_op(
 
 	*error = DB_SUCCESS;
 
-	/* 3 = 1 (op type) + 1 (extra_size) + at least 1 byte payload */
-	if (mrec + 3 >= mrec_end) {
-		return(NULL);
-	}
-
 	const bool is_instant = log->is_instant(dup->index);
 	const mrec_t* const mrec_start = mrec;
 
@@ -2657,6 +2681,11 @@ row_log_table_apply_op(
 			thr, new_trx_id_col,
 			mrec, offsets, offsets_heap, heap, dup, old_pk);
 		break;
+	case ROW_T_EMPTY:
+		*error = row_log_table_apply_empty(dup->index, thr);
+		log->head.total += 1;
+		next_mrec = mrec;
+		break;
 	}
 
 	ut_ad(log->head.total <= log->tail.total);
@@ -3438,6 +3467,9 @@ row_log_apply_op_low(
 			}
 
 			goto duplicate;
+		case ROW_OP_EMPTY:
+			ut_ad(0);
+			break;
 		}
 	} else {
 		switch (op) {
@@ -3508,6 +3540,9 @@ insert_the_rec:
 				0, NULL, &mtr);
 			ut_ad(!big_rec);
 			break;
+		case ROW_OP_EMPTY:
+			ut_ad(0);
+			break;
 		}
 		mem_heap_empty(offsets_heap);
 	}
@@ -3582,6 +3617,17 @@ row_log_apply_op(
 		op = static_cast<enum row_op>(*mrec++);
 		trx_id = 0;
 		break;
+	case ROW_OP_EMPTY:
+	{
+		mem_heap_t* heap = mem_heap_create(512);
+		que_fork_t* fork = que_fork_create(
+			NULL, NULL, QUE_FORK_MYSQL_INTERFACE, heap);
+		que_thr_t* thr = que_thr_create(fork, heap, nullptr);
+		index->empty(thr);
+		*error = DB_SUCCESS;
+		mem_heap_free(heap);
+		return mrec + 4;
+	}
 	default:
 corrupted:
 		ut_ad(0);
@@ -4024,3 +4070,14 @@ row_log_apply(
 
 	DBUG_RETURN(error);
 }
+
+void row_log_table_empty(dict_index_t *index)
+{
+  row_log_t* log= index->online_log;
+  ulint	avail_size;
+  if (byte* b = row_log_table_open(log, 1, &avail_size))
+  {
+    *b++ = ROW_T_EMPTY;
+    row_log_table_close(index, b, 1, avail_size);
+  }
+}
diff --git a/storage/innobase/row/row0merge.cc b/storage/innobase/row/row0merge.cc
index 3c9331d9909..be54dcf49e6 100644
--- a/storage/innobase/row/row0merge.cc
+++ b/storage/innobase/row/row0merge.cc
@@ -1750,6 +1750,15 @@ row_merge_read_clustered_index(
 	/* There is no previous tuple yet. */
 	prev_mtuple.fields = NULL;
 
+	if (trx_id_t bulk_trx_id = old_table->bulk_trx_id()) {
+		if (trx->read_view.is_open()
+		    && !trx->read_view.changes_visible(
+			bulk_trx_id, old_table->name)) {
+			trx->op_info="";
+			DBUG_RETURN(DB_SUCCESS);
+		}
+	}
+
 	for (ulint i = 0; i < n_index; i++) {
 		if (index[i]->type & DICT_FTS) {
 
diff --git a/storage/innobase/row/row0mysql.cc b/storage/innobase/row/row0mysql.cc
index d08d5662a69..d476cf76d85 100644
--- a/storage/innobase/row/row0mysql.cc
+++ b/storage/innobase/row/row0mysql.cc
@@ -1096,6 +1096,12 @@ row_get_prebuilt_insert_row(
 		    && prebuilt->ins_node->entry_list.size()
 		    == UT_LIST_GET_LEN(table->indexes)) {
 
+			if (prebuilt->ins_node->bulk_insert
+			    && prebuilt->ins_node->trx_id
+			       != prebuilt->trx->id) {
+				prebuilt->ins_node->bulk_insert= false;
+			}
+
 			return(prebuilt->ins_node->row);
 		}
 
diff --git a/storage/innobase/row/row0purge.cc b/storage/innobase/row/row0purge.cc
index 5ddca6a9d3a..a01b305acd4 100644
--- a/storage/innobase/row/row0purge.cc
+++ b/storage/innobase/row/row0purge.cc
@@ -892,6 +892,7 @@ row_purge_parse_undo_rec(
 	switch (type) {
 	case TRX_UNDO_RENAME_TABLE:
 		return false;
+	case TRX_UNDO_EMPTY:
 	case TRX_UNDO_INSERT_METADATA:
 	case TRX_UNDO_INSERT_REC:
 		/* These records do not store any transaction identifier.
@@ -982,6 +983,9 @@ err_exit:
 	if (type == TRX_UNDO_INSERT_METADATA) {
 		node->ref = &trx_undo_metadata;
 		return(true);
+	} else if (type == TRX_UNDO_EMPTY) {
+		node->ref = nullptr;
+		return true;
 	}
 
 	ptr = trx_undo_rec_get_row_ref(ptr, clust_index, &(node->ref),
@@ -1039,6 +1043,8 @@ row_purge_record_func(
 	ut_ad(!trx_undo_roll_ptr_is_insert(node->roll_ptr));
 
 	switch (node->rec_type) {
+	case TRX_UNDO_EMPTY:
+		break;
 	case TRX_UNDO_DEL_MARK_REC:
 		purged = row_purge_del_mark(node);
 		if (purged) {
diff --git a/storage/innobase/row/row0sel.cc b/storage/innobase/row/row0sel.cc
index 419a2e628aa..9333ac59222 100644
--- a/storage/innobase/row/row0sel.cc
+++ b/storage/innobase/row/row0sel.cc
@@ -4593,6 +4593,17 @@ wait_table_again:
 		}
 	}
 
+	if (trx_id_t bulk_trx_id = index->table->bulk_trx_id()) {
+		if (trx->isolation_level != TRX_ISO_READ_UNCOMMITTED
+		    && trx->read_view.is_open()
+		    && !trx->read_view.changes_visible(
+				bulk_trx_id, index->table->name)) {
+			trx->op_info = "";
+			err = DB_END_OF_INDEX;
+			goto normal_return;
+		}
+	}
+
 	/* Open or restore index cursor position */
 
 	if (UNIV_LIKELY(direction != 0)) {
diff --git a/storage/innobase/row/row0uins.cc b/storage/innobase/row/row0uins.cc
index 492c2f35e62..947d2bbcfe9 100644
--- a/storage/innobase/row/row0uins.cc
+++ b/storage/innobase/row/row0uins.cc
@@ -384,6 +384,7 @@ static bool row_undo_ins_parse_undo_rec(undo_node_t* node, bool dict_locked)
 		goto close_table;
 	case TRX_UNDO_INSERT_METADATA:
 	case TRX_UNDO_INSERT_REC:
+	case TRX_UNDO_EMPTY:
 		break;
 	case TRX_UNDO_RENAME_TABLE:
 		dict_table_t* table = node->table;
@@ -424,6 +425,9 @@ close_table:
 				ptr = trx_undo_rec_get_row_ref(
 					ptr, clust_index, &node->ref,
 					node->heap);
+			} else if (node->rec_type == TRX_UNDO_EMPTY) {
+				node->ref = nullptr;
+				return true;
 			} else {
 				node->ref = &trx_undo_metadata;
 				if (!row_undo_search_clust_to_pcur(node)) {
@@ -596,6 +600,11 @@ row_undo_ins(
 		log_free_check();
 		ut_ad(!node->table->is_temporary());
 		err = row_undo_ins_remove_clust_rec(node);
+		break;
+	case TRX_UNDO_EMPTY:
+		node->table->empty_table(thr);
+		err = DB_SUCCESS;
+		break;
 	}
 
 	dict_table_close(node->table, dict_locked, FALSE);
diff --git a/storage/innobase/row/row0undo.cc b/storage/innobase/row/row0undo.cc
index 375de331255..0ef870714c1 100644
--- a/storage/innobase/row/row0undo.cc
+++ b/storage/innobase/row/row0undo.cc
@@ -363,6 +363,7 @@ static bool row_undo_rec_get(undo_node_t* node)
 
 	switch (trx_undo_rec_get_type(node->undo_rec)) {
 	case TRX_UNDO_INSERT_METADATA:
+	case TRX_UNDO_EMPTY:
 		/* This record type was introduced in MDEV-11369
 		instant ADD COLUMN, which was implemented after
 		MDEV-12288 removed the insert_undo log. There is no
diff --git a/storage/innobase/trx/trx0rec.cc b/storage/innobase/trx/trx0rec.cc
index 2503a2bd3aa..07567f52ef4 100644
--- a/storage/innobase/trx/trx0rec.cc
+++ b/storage/innobase/trx/trx0rec.cc
@@ -38,6 +38,7 @@ Created 3/26/1996 Heikki Tuuri
 #include "trx0rseg.h"
 #include "row0row.h"
 #include "row0mysql.h"
+#include "row0ins.h"
 
 /** The search tuple corresponding to TRX_UNDO_INSERT_METADATA. */
 const dtuple_t trx_undo_metadata = {
@@ -383,7 +384,8 @@ trx_undo_page_report_insert(
 	dict_index_t*	index,		/*!< in: clustered index */
 	const dtuple_t*	clust_entry,	/*!< in: index entry which will be
 					inserted to the clustered index */
-	mtr_t*		mtr)		/*!< in: mtr */
+	mtr_t*		mtr,		/*!< in: mtr */
+	bool		write_empty)
 {
 	ut_ad(index->is_primary());
 	/* MariaDB 10.3.1+ in trx_undo_page_init() always initializes
@@ -411,6 +413,13 @@ trx_undo_page_report_insert(
 	*ptr++ = TRX_UNDO_INSERT_REC;
 	ptr += mach_u64_write_much_compressed(ptr, trx->undo_no);
 	ptr += mach_u64_write_much_compressed(ptr, index->table->id);
+
+	/* Table is in bulk operation */
+	if (write_empty) {
+		undo_block->frame[first_free + 2] = TRX_UNDO_EMPTY;
+		goto done;
+	}
+
 	/*----------------------------------------*/
 	/* Store then the fields required to uniquely determine the record
 	to be inserted in the clustered index */
@@ -488,7 +497,7 @@ trx_undo_rec_get_pars(
 	type_cmpl &= ~TRX_UNDO_UPD_EXTERN;
 	*type = type_cmpl & (TRX_UNDO_CMPL_INFO_MULT - 1);
 	ut_ad(*type >= TRX_UNDO_RENAME_TABLE);
-	ut_ad(*type <= TRX_UNDO_DEL_MARK_REC);
+	ut_ad(*type <= TRX_UNDO_EMPTY);
 	*cmpl_info = type_cmpl / TRX_UNDO_CMPL_INFO_MULT;
 
 	*undo_no = mach_read_next_much_compressed(&ptr);
@@ -2002,7 +2011,12 @@ trx_undo_report_row_operation(
 	buf_block_t*	undo_block = trx_undo_assign_low(trx, rseg, pundo,
 							 &err, &mtr);
 	trx_undo_t*	undo	= *pundo;
-
+	bool		write_empty = !rec && thr && thr->run_node
+				      && que_node_get_type(thr->run_node)
+						== QUE_NODE_INSERT
+				      && static_cast<ins_node_t*>(
+						thr->run_node)->bulk_insert
+				      && !trx->allow_insert_undo(index->table);
 	ut_ad((err == DB_SUCCESS) == (undo_block != NULL));
 	if (UNIV_UNLIKELY(undo_block == NULL)) {
 		goto err_exit;
@@ -2013,7 +2027,8 @@ trx_undo_report_row_operation(
 	do {
 		uint16_t offset = !rec
 			? trx_undo_page_report_insert(
-				undo_block, trx, index, clust_entry, &mtr)
+				undo_block, trx, index, clust_entry, &mtr,
+				write_empty)
 			: trx_undo_page_report_modify(
 				undo_block, trx, index, rec, offsets, update,
 				cmpl_info, clust_entry, &mtr);
@@ -2102,8 +2117,12 @@ trx_undo_report_row_operation(
 				}
 			}
 
-			*roll_ptr = trx_undo_build_roll_ptr(
-				!rec, rseg->id, undo->top_page_no, offset);
+			if (!write_empty) {
+				*roll_ptr = trx_undo_build_roll_ptr(
+					!rec, rseg->id, undo->top_page_no,
+					offset);
+			}
+
 			return(DB_SUCCESS);
 		}
 
diff --git a/storage/innobase/trx/trx0roll.cc b/storage/innobase/trx/trx0roll.cc
index 2967281382d..fe6c77ae85e 100644
--- a/storage/innobase/trx/trx0roll.cc
+++ b/storage/innobase/trx/trx0roll.cc
@@ -536,6 +536,8 @@ trx_savepoint_for_mysql(
 
 	UT_LIST_ADD_LAST(trx->trx_savepoints, savep);
 
+	trx->set_allow_insert_undo();
+
 	return(DB_SUCCESS);
 }
 
diff --git a/storage/innobase/trx/trx0trx.cc b/storage/innobase/trx/trx0trx.cc
index 50a78adcf53..c3b8e7bad21 100644
--- a/storage/innobase/trx/trx0trx.cc
+++ b/storage/innobase/trx/trx0trx.cc
@@ -563,6 +563,7 @@ trx_resurrect_table_locks(
 {
 	mtr_t			mtr;
 	table_id_set		tables;
+	table_id_set		unempty_tables;
 
 	ut_ad(trx_state_eq(trx, TRX_STATE_ACTIVE) ||
 	      trx_state_eq(trx, TRX_STATE_PREPARED));
@@ -597,6 +598,11 @@ trx_resurrect_table_locks(
 		trx_undo_rec_get_pars(
 			undo_rec, &type, &cmpl_info,
 			&updated_extern, &undo_no, &table_id);
+
+		if (type == TRX_UNDO_EMPTY) {
+			unempty_tables.insert(table_id);
+		}
+
 		tables.insert(table_id);
 
 		undo_rec = trx_undo_get_prev_rec(
@@ -623,8 +629,13 @@ trx_resurrect_table_locks(
 					trx_mod_tables_t::value_type(table,
 								     0));
 			}
+
 			lock_table_ix_resurrect(table, trx);
 
+			if (unempty_tables.find(*i) != unempty_tables.end()) {
+				lock_table_x_resurrect(table, trx);
+			}
+
 			DBUG_LOG("ib_trx",
 				 "resurrect " << ib::hex(trx->id)
 				 << " IX lock on " << table->name);
@@ -1744,6 +1755,7 @@ trx_mark_sql_stat_end(
 			fts_savepoint_laststmt_refresh(trx);
 		}
 
+		trx->set_allow_insert_undo();
 		return;
 	}