MDEV-14589 InnoDB should not lock a delete-marked record

When the transaction isolation level is SERIALIZABLE, or when
a locking read is performed in the REPEATABLE READ isolation level,
InnoDB must lock delete-marked records in order to prevent another
transaction from inserting something.

However, at READ UNCOMMITTED or READ COMMITTED isolation level or
when the parameter innodb_locks_unsafe_for_binlog is set, the
repeatability of the reads does not matter, and there is no need
to lock any records.

row_search_mvcc(): Skip locks on delete-marked committed records upfront,
instead of invoking row_unlock_for_mysql() afterwards. The unlocking
never worked for secondary index records.

3 files changed, 167 insertions, 13 deletions

diff --git a/mysql-test/suite/innodb/r/lock_deleted.result b/mysql-test/suite/innodb/r/lock_deleted.result
new file mode 100644
index 00000000000..0fcb8bd5aa8
--- /dev/null
+++ b/mysql-test/suite/innodb/r/lock_deleted.result
@@ -0,0 +1,57 @@
+connect stop_purge, localhost, root,,;
+START TRANSACTION WITH CONSISTENT SNAPSHOT;
+connect delete, localhost, root,,;
+connection default;
+CREATE TABLE t1(a INT PRIMARY KEY, b INT UNIQUE) ENGINE=InnoDB;
+INSERT INTO t1 VALUES(1,1);
+DELETE FROM t1;
+SET DEBUG_SYNC='row_ins_sec_index_unique SIGNAL inserted WAIT_FOR locked';
+BEGIN;
+INSERT INTO t1 VALUES(1,1);
+connection delete;
+SET DEBUG_SYNC='now WAIT_FOR inserted';
+SET DEBUG_SYNC='innodb_row_search_for_mysql_exit SIGNAL locked';
+SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED;
+BEGIN;
+DELETE FROM t1 WHERE b=1;
+connection default;
+connection delete;
+COMMIT;
+connection default;
+SET DEBUG_SYNC='RESET';
+ROLLBACK;
+SET DEBUG_SYNC='row_ins_sec_index_unique SIGNAL inserted WAIT_FOR locked';
+BEGIN;
+INSERT INTO t1 VALUES(1,1);
+connection delete;
+SET DEBUG_SYNC='now WAIT_FOR inserted';
+SET DEBUG_SYNC='innodb_row_search_for_mysql_exit SIGNAL locked';
+SET SESSION TRANSACTION ISOLATION LEVEL READ UNCOMMITTED;
+BEGIN;
+DELETE FROM t1 WHERE b=1;
+connection default;
+connection delete;
+COMMIT;
+connection default;
+SET DEBUG_SYNC='RESET';
+ROLLBACK;
+SET DEBUG_SYNC='row_ins_sec_index_unique SIGNAL inserted WAIT_FOR locked';
+BEGIN;
+SET innodb_lock_wait_timeout=1;
+INSERT INTO t1 VALUES(1,1);
+connection delete;
+SET DEBUG_SYNC='now WAIT_FOR inserted';
+SET DEBUG_SYNC='innodb_row_search_for_mysql_exit SIGNAL locked';
+SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ;
+BEGIN;
+DELETE FROM t1 WHERE b=1;
+connection default;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+COMMIT;
+SET DEBUG_SYNC='RESET';
+connection delete;
+COMMIT;
+disconnect delete;
+disconnect stop_purge;
+connection default;
+DROP TABLE t1;
diff --git a/mysql-test/suite/innodb/t/lock_deleted.test b/mysql-test/suite/innodb/t/lock_deleted.test
new file mode 100644
index 00000000000..8dbad90d354
--- /dev/null
+++ b/mysql-test/suite/innodb/t/lock_deleted.test
@@ -0,0 +1,72 @@
+--source include/have_innodb.inc
+--source include/have_debug.inc
+--source include/have_debug_sync.inc
+
+--source include/count_sessions.inc
+
+connect(stop_purge, localhost, root,,);
+START TRANSACTION WITH CONSISTENT SNAPSHOT;
+connect(delete, localhost, root,,);
+connection default;
+
+CREATE TABLE t1(a INT PRIMARY KEY, b INT UNIQUE) ENGINE=InnoDB;
+INSERT INTO t1 VALUES(1,1);
+DELETE FROM t1;
+
+let $i=2;
+while ($i) {
+let $iso= `SELECT CASE $i WHEN 1 THEN 'UNCOMMITTED' ELSE 'COMMITTED' END`;
+
+SET DEBUG_SYNC='row_ins_sec_index_unique SIGNAL inserted WAIT_FOR locked';
+BEGIN;
+send INSERT INTO t1 VALUES(1,1);
+
+connection delete;
+SET DEBUG_SYNC='now WAIT_FOR inserted';
+SET DEBUG_SYNC='innodb_row_search_for_mysql_exit SIGNAL locked';
+eval SET SESSION TRANSACTION ISOLATION LEVEL READ $iso;
+BEGIN;
+send DELETE FROM t1 WHERE b=1;
+
+connection default;
+reap;
+connection delete;
+reap;
+COMMIT;
+
+connection default;
+SET DEBUG_SYNC='RESET';
+ROLLBACK;
+
+dec $i;
+}
+
+SET DEBUG_SYNC='row_ins_sec_index_unique SIGNAL inserted WAIT_FOR locked';
+BEGIN;
+SET innodb_lock_wait_timeout=1;
+send INSERT INTO t1 VALUES(1,1);
+
+connection delete;
+SET DEBUG_SYNC='now WAIT_FOR inserted';
+SET DEBUG_SYNC='innodb_row_search_for_mysql_exit SIGNAL locked';
+SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ;
+BEGIN;
+send DELETE FROM t1 WHERE b=1;
+
+connection default;
+--error ER_LOCK_WAIT_TIMEOUT
+reap;
+COMMIT;
+SET DEBUG_SYNC='RESET';
+
+connection delete;
+reap;
+COMMIT;
+
+disconnect delete;
+disconnect stop_purge;
+
+connection default;
+DROP TABLE t1;
+
+--source include/wait_until_count_sessions.inc
diff --git a/storage/innobase/row/row0sel.cc b/storage/innobase/row/row0sel.cc
index 9bf71592f54..9cad3ab7de6 100644
--- a/storage/innobase/row/row0sel.cc
+++ b/storage/innobase/row/row0sel.cc
@@ -4890,9 +4890,44 @@ wrong_offs:
 
 		ulint	lock_type;
 
+		if (srv_locks_unsafe_for_binlog
+		    || trx->isolation_level <= TRX_ISO_READ_COMMITTED) {
+			/* At READ COMMITTED or READ UNCOMMITTED
+			isolation levels, do not lock committed
+			delete-marked records. */
+			if (!rec_get_deleted_flag(rec, comp)) {
+				goto no_gap_lock;
+			}
+			if (index == clust_index) {
+				trx_id_t trx_id = row_get_rec_trx_id(
+					rec, index, offsets);
+				/* In delete-marked records, DB_TRX_ID must
+				always refer to an existing undo log record. */
+				ut_ad(trx_id);
+				if (!trx_rw_is_active(trx_id, NULL, false)) {
+					/* The clustered index record
+					was delete-marked in a committed
+					transaction. Ignore the record. */
+					goto locks_ok_del_marked;
+				}
+			} else if (trx_t* trx = row_vers_impl_x_locked(
+					   rec, index, offsets)) {
+				/* The record belongs to an active
+				transaction. We must acquire a lock. */
+				trx_release_reference(trx);
+			} else {
+				/* The secondary index record does not
+				point to a delete-marked clustered index
+				record that belongs to an active transaction.
+				Ignore the secondary index record, because
+				it is not locked. */
+				goto next_rec;
+			}
+
+			goto no_gap_lock;
+		}
+
 		if (!set_also_gap_locks
-		    || srv_locks_unsafe_for_binlog
-		    || trx->isolation_level <= TRX_ISO_READ_COMMITTED
 		    || (unique_search && !rec_get_deleted_flag(rec, comp))
 		    || dict_index_is_spatial(index)) {
 
@@ -5096,6 +5131,7 @@ locks_ok:
 	page_rec_is_comp() cannot be used! */
 
 	if (rec_get_deleted_flag(rec, comp)) {
+locks_ok_del_marked:
 		/* In delete-marked records, DB_TRX_ID must
 		always refer to an existing undo log record. */
 		ut_ad(index != clust_index
@@ -5103,17 +5139,6 @@ locks_ok:
 
 		/* The record is delete-marked: we can skip it */
 
-		if ((srv_locks_unsafe_for_binlog
-		     || trx->isolation_level <= TRX_ISO_READ_COMMITTED)
-		    && prebuilt->select_lock_type != LOCK_NONE
-		    && !did_semi_consistent_read) {
-
-			/* No need to keep a lock on a delete-marked record
-			if we do not want to use next-key locking. */
-
-			row_unlock_for_mysql(prebuilt, TRUE);
-		}
-
 		/* This is an optimization to skip setting the next key lock
 		on the record that follows this delete-marked record. This
 		optimization works because of the unique search criteria