MDEV-24224 Gap lock on delete in 10.5 using READ COMMITTEDbb-10.5-MDEV-24224

When MDEV-19544 (commit 1a6f470464171bd1144e4dd6f169bb4018f2e81a)
simplified the initialization of the local variable
set_also_gap_locks, an inadvertent change was included.
Essentially, all code branches that are executed when
set_also_gap_locks hold must also ensure that
trx->isolation_level > TRX_ISO_READ_COMMITTED holds.
This was being violated in a few code paths.

It turns out that there is an even simpler fix: Remove the test
of thd_is_select() completely. In that way, the first part of
UPDATE or DELETE should work exactly like SELECT...FOR UPDATE.

thd_is_select(): Remove.

5 files changed, 78 insertions, 27 deletions

diff --git a/mysql-test/suite/innodb/r/gap_locks.result b/mysql-test/suite/innodb/r/gap_locks.result
new file mode 100644
index 00000000000..cd60b1fab22
--- /dev/null
+++ b/mysql-test/suite/innodb/r/gap_locks.result
@@ -0,0 +1,35 @@
+CREATE TABLE t1(a INT PRIMARY KEY, b VARCHAR(40), c INT, INDEX(b,c))
+ENGINE=InnoDB;
+INSERT INTO t1 VALUES (1,'1',1),(2,'2',1);
+SET @save_locks= @@GLOBAL.innodb_status_output_locks;
+SET GLOBAL INNODB_STATUS_OUTPUT_LOCKS = 'ON';
+SET TRANSACTION ISOLATION LEVEL SERIALIZABLE;
+BEGIN;
+DELETE FROM t1 WHERE b='2' AND c=2;
+SHOW ENGINE INNODB STATUS;
+Type	Name	Status
+InnoDB		2 lock struct(s), 1 row lock(s)
+ROLLBACK;
+SET TRANSACTION ISOLATION LEVEL REPEATABLE READ;
+BEGIN;
+DELETE FROM t1 WHERE b='2' AND c=2;
+SHOW ENGINE INNODB STATUS;
+Type	Name	Status
+InnoDB		2 lock struct(s), 1 row lock(s)
+ROLLBACK;
+SET TRANSACTION ISOLATION LEVEL READ COMMITTED;
+BEGIN;
+DELETE FROM t1 WHERE b='2' AND c=2;
+SHOW ENGINE INNODB STATUS;
+Type	Name	Status
+InnoDB		1 lock struct(s), 0 row lock(s)
+ROLLBACK;
+SET TRANSACTION ISOLATION LEVEL READ UNCOMMITTED;
+BEGIN;
+DELETE FROM t1 WHERE b='2' AND c=2;
+SHOW ENGINE INNODB STATUS;
+Type	Name	Status
+InnoDB		1 lock struct(s), 0 row lock(s)
+ROLLBACK;
+SET GLOBAL INNODB_STATUS_OUTPUT_LOCKS = @save_locks;
+DROP TABLE t1;
diff --git a/mysql-test/suite/innodb/t/gap_locks.test b/mysql-test/suite/innodb/t/gap_locks.test
new file mode 100644
index 00000000000..77ce2c842b1
--- /dev/null
+++ b/mysql-test/suite/innodb/t/gap_locks.test
@@ -0,0 +1,29 @@
+--source include/have_innodb.inc
+
+CREATE TABLE t1(a INT PRIMARY KEY, b VARCHAR(40), c INT, INDEX(b,c))
+ENGINE=InnoDB;
+INSERT INTO t1 VALUES (1,'1',1),(2,'2',1);
+
+SET @save_locks= @@GLOBAL.innodb_status_output_locks;
+SET GLOBAL INNODB_STATUS_OUTPUT_LOCKS = 'ON';
+
+let $isolation= 4;
+while ($isolation) {
+let $tx_isolation= `SELECT CASE $isolation
+WHEN 1 THEN 'READ UNCOMMITTED'
+WHEN 2 THEN 'READ COMMITTED'
+WHEN 3 THEN 'REPEATABLE READ'
+ELSE 'SERIALIZABLE' END`;
+
+eval SET TRANSACTION ISOLATION LEVEL $tx_isolation;
+BEGIN;
+DELETE FROM t1 WHERE b='2' AND c=2;
+--replace_regex /.*(\d+ lock struct...), heap size \d+(, \d+ row lock...).*/\1\2/
+SHOW ENGINE INNODB STATUS;
+ROLLBACK;
+
+dec $isolation;
+}
+
+SET GLOBAL INNODB_STATUS_OUTPUT_LOCKS = @save_locks;
+DROP TABLE t1;
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index 6fd1e99023a..a35b90e91f0 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -1495,7 +1495,7 @@ thd_trx_is_auto_commit(
 	       && !thd_test_options(
 		       thd,
 		       OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN)
-	       && thd_is_select(thd));
+	       && thd_sql_command(thd) == SQLCOM_SELECT);
 }
 
 /******************************************************************//**
@@ -1532,17 +1532,6 @@ thd_query_start_micro(
 }
 
 /******************************************************************//**
-Returns true if the thread is executing a SELECT statement.
-@return true if thd is executing SELECT */
-ibool
-thd_is_select(
-/*==========*/
-	const THD*	thd)	/*!< in: thread handle */
-{
-	return(thd_sql_command(thd) == SQLCOM_SELECT);
-}
-
-/******************************************************************//**
 Returns the lock wait timeout for the current connection.
 @return the lock wait timeout, in seconds */
 ulong
diff --git a/storage/innobase/include/ha_prototypes.h b/storage/innobase/include/ha_prototypes.h
index 35250f99df9..41068e7c56c 100644
--- a/storage/innobase/include/ha_prototypes.h
+++ b/storage/innobase/include/ha_prototypes.h
@@ -191,14 +191,6 @@ innobase_basename(
 	const char*	path_name);
 
 /******************************************************************//**
-Returns true if the thread is executing a SELECT statement.
-@return true if thd is executing SELECT */
-ibool
-thd_is_select(
-/*==========*/
-	const THD*	thd);	/*!< in: thread handle */
-
-/******************************************************************//**
 Converts an identifier to a table name. */
 void
 innobase_convert_from_table_id(
diff --git a/storage/innobase/row/row0sel.cc b/storage/innobase/row/row0sel.cc
index d7a0490db3d..7a859ab8b09 100644
--- a/storage/innobase/row/row0sel.cc
+++ b/storage/innobase/row/row0sel.cc
@@ -4542,12 +4542,11 @@ aborted:
 	      || prebuilt->table->no_rollback()
 	      || srv_read_only_mode);
 
-	/* Do not lock gaps for plain SELECT
-	at READ UNCOMMITTED or READ COMMITTED isolation level */
+	/* Do not lock gaps at READ UNCOMMITTED or READ COMMITTED
+	isolation level */
 	const bool set_also_gap_locks =
 		prebuilt->select_lock_type != LOCK_NONE
-		&& (trx->isolation_level > TRX_ISO_READ_COMMITTED
-		    || !thd_is_select(trx->mysql_thd))
+		&& trx->isolation_level > TRX_ISO_READ_COMMITTED
 #ifdef WITH_WSREP
 		&& !wsrep_thd_skip_locking(trx->mysql_thd)
 #endif /* WITH_WSREP */
@@ -4755,7 +4754,6 @@ rec_loop:
 	if (page_rec_is_supremum(rec)) {
 
 		if (set_also_gap_locks
-		    && trx->isolation_level > TRX_ISO_READ_COMMITTED
 		    && !dict_index_is_spatial(index)) {
 
 			/* Try to place a lock on the index record */
@@ -5020,8 +5018,16 @@ wrong_offs:
 			goto no_gap_lock;
 		}
 
-		if (!set_also_gap_locks
-		    || (unique_search && !rec_get_deleted_flag(rec, comp))
+#ifdef WITH_WSREP
+		if (UNIV_UNLIKELY(!set_also_gap_locks)) {
+			ut_ad(wsrep_thd_skip_locking(trx->mysql_thd));
+			goto no_gap_lock;
+		}
+#else /* WITH_WSREP */
+		ut_ad(set_also_gap_locks);
+#endif /* WITH_WSREP */
+
+		if ((unique_search && !rec_get_deleted_flag(rec, comp))
 		    || dict_index_is_spatial(index)) {
 
 			goto no_gap_lock;