Revert "MDEV-19544 Remove innodb_locks_unsafe_for_binlog"

Revert "MDEV-19563 Removed references to deprecated option innodb_locks_unsafe_for_binlog"

also revert part of :
* MDEV-23039 update removed_variables in mysql_upgrade_service for 10.5
* more "removed" mysqld command-line options

also add following tests :
* main.innodb_locks_unsafe_for_binlog, based on main.unsafe_binlog_innodb.
* main.concurrent_innodb_locks_unsafe_for_binlog, based on main.concurrent_innodb_unsafelog

Docs/README-wsrep

@@ -269,6 +269,9 @@ innodb_autoinc_lock_mode=2
    autoinc lock modes 0 and 1 can cause unresolved deadlock, and make
    the system unresponsive.
 
+innodb_locks_unsafe_for_binlog=1
+   This option is required for parallel applying.
+
 5.2 WSREP OPTIONS
 
 All options are optional except for wsrep_provider, wsrep_cluster_address, and

mysql-test/main/concurrent_innodb_locks_unsafe_for_binlog-master.opt

+--loose-innodb_locks_unsafe_for_binlog
+--loose-innodb_lock_wait_timeout=1

mysql-test/main/concurrent_innodb_locks_unsafe_for_binlog.test

+# t/concurrent_innodb_locks_unsafe_for_binlog
+#
+# Concurrent InnoDB tests
+#
+# Last update:
+# 2006-07-26 ML test refactored (MySQL 5.1)
+#               main code t/innodb_concurrent.test -> include/concurrent.inc
+#               new wrapper t/concurrent_innodb.test
+# 2008-06-03 KP test refactored; removed name locks, added comments.
+#               renamed wrapper t/concurrent_innodb.test ->
+#                           t/concurrent_innodb_unsafelog.test
+#               new wrapper t/concurrent_innodb_safelog.test
+#
+
+--source include/have_innodb.inc
+
+let $engine_type= InnoDB;
+
+SET GLOBAL TRANSACTION ISOLATION LEVEL REPEATABLE READ;
+# innodb_locks_unsafe_for_binlog is set fro this test.
+--disable_service_connection
+--source include/concurrent_innodb_locks_unsafe_for_binlog.inc
+--enable_service_connection

mysql-test/main/innodb_locks_unsafe_for_binlog-master.opt

+--loose-innodb_locks_unsafe_for_binlog --loose-innodb_lock_wait_timeout=1

mysql-test/main/innodb_locks_unsafe_for_binlog.result

+drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9,t10;
+create table t1 (id int not null, f_id int not null, f int not null,
+primary key(f_id, id)) engine = InnoDB;
+create table t2 (id int not null,s_id int not null,s varchar(200),
+primary key(id)) engine = InnoDB;
+INSERT INTO t1 VALUES (8, 1, 3);
+INSERT INTO t1 VALUES (1, 2, 1);
+INSERT INTO t2 VALUES (1, 0, '');
+INSERT INTO t2 VALUES (8, 1, '');
+commit;
+DELETE ml.* FROM t1 AS ml LEFT JOIN t2 AS mm ON (mm.id=ml.id)
+WHERE mm.id IS NULL;
+select ml.* from t1 as ml left join t2 as mm on (mm.id=ml.id)
+where mm.id is null lock in share mode;
+id	f_id	f
+drop table t1,t2;
+connect  a,localhost,root,,;
+connect  b,localhost,root,,;
+connection a;
+create table t1(a int not null, b int, primary key(a)) engine = InnoDB;
+insert into t1 values(1,1),(2,2),(3,1),(4,2),(5,1),(6,2),(7,3);
+commit;
+set autocommit = 0;
+select * from t1 lock in share mode;
+a	b
+1	1
+2	2
+3	1
+4	2
+5	1
+6	2
+7	3
+update t1 set b = 5 where b = 1;
+connection b;
+set autocommit = 0;
+select * from t1 where a = 2 and b = 2 for update;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+connection a;
+commit;
+connection b;
+commit;
+drop table t1;
+connection default;
+disconnect a;
+disconnect b;
+connect  a,localhost,root,,;
+connect  b,localhost,root,,;
+connection a;
+create table t1(a int not null, b int, primary key(a)) engine = InnoDB;
+insert into t1 values(1,1),(2,2),(3,1),(4,2),(5,1),(6,2),(7,3);
+commit;
+set autocommit = 0;
+update t1 set b = 5 where b = 1;
+connection b;
+set autocommit = 0;
+select * from t1 where a = 7 and b = 3 for update;
+a	b
+7	3
+commit;
+connection a;
+commit;
+drop table t1;
+connection default;
+disconnect a;
+disconnect b;
+connect  a,localhost,root,,;
+connect  b,localhost,root,,;
+connection a;
+create table t1(a int not null, b int, primary key(a)) engine = InnoDB;
+insert into t1 values (1,2),(5,3),(4,2);
+create table t2(d int not null, e int, primary key(d)) engine = InnoDB;
+insert into t2 values (8,6),(12,1),(3,1);
+commit;
+set autocommit = 0;
+select * from t2 for update;
+d	e
+3	1
+8	6
+12	1
+connection b;
+set autocommit = 0;
+insert into t1 select * from t2;
+update t1 set b = (select e from t2 where a = d);
+create table t3(d int not null, e int, primary key(d)) engine = InnoDB
+select * from t2;
+commit;
+connection a;
+commit;
+connection default;
+disconnect a;
+disconnect b;
+drop table t1, t2, t3;
+connect  a,localhost,root,,;
+connect  b,localhost,root,,;
+connect  c,localhost,root,,;
+connect  d,localhost,root,,;
+SET SESSION DEFAULT_STORAGE_ENGINE = InnoDB;
+connect  e,localhost,root,,;
+connect  f,localhost,root,,;
+connect  g,localhost,root,,;
+SET SESSION DEFAULT_STORAGE_ENGINE = InnoDB;
+connect  h,localhost,root,,;
+connect  i,localhost,root,,;
+connect  j,localhost,root,,;
+SET SESSION DEFAULT_STORAGE_ENGINE = InnoDB;
+connection a;
+create table t1(a int not null, b int, primary key(a)) engine = InnoDB;
+insert into t1 values (1,2),(5,3),(4,2);
+create table t2(a int not null, b int, primary key(a)) engine = InnoDB;
+insert into t2 values (8,6),(12,1),(3,1);
+create table t3(d int not null, b int, primary key(d)) engine = InnoDB;
+insert into t3 values (8,6),(12,1),(3,1);
+create table t5(a int not null, b int, primary key(a)) engine = InnoDB;
+insert into t5 values (1,2),(5,3),(4,2);
+create table t6(d int not null, e int, primary key(d)) engine = InnoDB;
+insert into t6 values (8,6),(12,1),(3,1);
+create table t8(a int not null, b int, primary key(a)) engine = InnoDB;
+insert into t8 values (1,2),(5,3),(4,2);
+create table t9(d int not null, e int, primary key(d)) engine = InnoDB;
+insert into t9 values (8,6),(12,1),(3,1);
+commit;
+set autocommit = 0;
+select * from t2 for update;
+a	b
+3	1
+8	6
+12	1
+connection b;
+set autocommit = 0;
+SET SESSION TRANSACTION ISOLATION LEVEL SERIALIZABLE;
+insert into t1 select * from t2;
+connection c;
+set autocommit = 0;
+SET SESSION TRANSACTION ISOLATION LEVEL SERIALIZABLE;
+update t3 set b = (select b from t2 where a = d);
+connection d;
+set autocommit = 0;
+SET SESSION TRANSACTION ISOLATION LEVEL SERIALIZABLE;
+create table t4(a int not null, b int, primary key(a)) select * from t2;
+connection e;
+set autocommit = 0;
+insert into t5 (select * from t2 lock in share mode);
+connection f;
+set autocommit = 0;
+update t6 set e = (select b from t2 where a = d lock in share mode);
+connection g;
+set autocommit = 0;
+create table t7(a int not null, b int, primary key(a)) select * from t2 lock in share mode;
+connection h;
+set autocommit = 0;
+insert into t8 (select * from t2 for update);
+connection i;
+set autocommit = 0;
+update t9 set e = (select b from t2 where a = d for update);
+connection j;
+set autocommit = 0;
+create table t10(a int not null, b int, primary key(a)) select * from t2 for update;
+connection b;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+connection c;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+connection d;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+connection e;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+connection f;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+connection g;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+connection h;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+connection i;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+connection j;
+ERROR HY000: Lock wait timeout exceeded; try restarting transaction
+connection a;
+commit;
+connection default;
+disconnect a;
+disconnect b;
+disconnect c;
+disconnect d;
+disconnect e;
+disconnect f;
+disconnect g;
+disconnect h;
+disconnect i;
+disconnect j;
+drop table t1, t2, t3, t5, t6, t8, t9;

mysql-test/main/innodb_locks_unsafe_for_binlog.test

+# t/unsafe_binlog_innodb.test
+#
+# Note that this test uses at least in case of InnoDB options
+#     innodb_locks_unsafe_for_binlog = true
+#     innodb_lock_timeout = 5
+#
+# Last update:
+# 2006-08-02 ML test refactored
+#               old name was innodb_unsafe_binlog.test
+#               main code went into include/unsafe_binlog.inc
+#
+
+--source include/have_innodb.inc
+let $engine_type= InnoDB;
+
+--source include/unsafe_binlog.inc

mysql-test/suite/sys_vars/r/innodb_locks_unsafe_for_binlog_basic.result

+'#---------------------BS_STVARS_031_01----------------------#'
+SELECT COUNT(@@GLOBAL.innodb_locks_unsafe_for_binlog);
+COUNT(@@GLOBAL.innodb_locks_unsafe_for_binlog)
+1
+1 Expected
+'#---------------------BS_STVARS_031_02----------------------#'
+SET @@GLOBAL.innodb_locks_unsafe_for_binlog=1;
+ERROR HY000: Variable 'innodb_locks_unsafe_for_binlog' is a read only variable
+Expected error 'Read only variable'
+SELECT COUNT(@@GLOBAL.innodb_locks_unsafe_for_binlog);
+COUNT(@@GLOBAL.innodb_locks_unsafe_for_binlog)
+1
+1 Expected
+'#---------------------BS_STVARS_031_03----------------------#'
+SELECT IF(@@GLOBAL.innodb_locks_unsafe_for_binlog, "ON", "OFF") = VARIABLE_VALUE
+FROM INFORMATION_SCHEMA.GLOBAL_VARIABLES
+WHERE VARIABLE_NAME='innodb_locks_unsafe_for_binlog';
+IF(@@GLOBAL.innodb_locks_unsafe_for_binlog, "ON", "OFF") = VARIABLE_VALUE
+1
+1 Expected
+SELECT COUNT(@@GLOBAL.innodb_locks_unsafe_for_binlog);
+COUNT(@@GLOBAL.innodb_locks_unsafe_for_binlog)
+1
+1 Expected
+SELECT COUNT(VARIABLE_VALUE)
+FROM INFORMATION_SCHEMA.GLOBAL_VARIABLES 
+WHERE VARIABLE_NAME='innodb_locks_unsafe_for_binlog';
+COUNT(VARIABLE_VALUE)
+1
+1 Expected
+'#---------------------BS_STVARS_031_04----------------------#'
+SELECT @@innodb_locks_unsafe_for_binlog = @@GLOBAL.innodb_locks_unsafe_for_binlog;
+@@innodb_locks_unsafe_for_binlog = @@GLOBAL.innodb_locks_unsafe_for_binlog
+1
+1 Expected
+'#---------------------BS_STVARS_031_05----------------------#'
+SELECT COUNT(@@innodb_locks_unsafe_for_binlog);
+COUNT(@@innodb_locks_unsafe_for_binlog)
+1
+1 Expected
+SELECT COUNT(@@local.innodb_locks_unsafe_for_binlog);
+ERROR HY000: Variable 'innodb_locks_unsafe_for_binlog' is a GLOBAL variable
+Expected error 'Variable is a GLOBAL variable'
+SELECT COUNT(@@SESSION.innodb_locks_unsafe_for_binlog);
+ERROR HY000: Variable 'innodb_locks_unsafe_for_binlog' is a GLOBAL variable
+Expected error 'Variable is a GLOBAL variable'
+SELECT COUNT(@@GLOBAL.innodb_locks_unsafe_for_binlog);
+COUNT(@@GLOBAL.innodb_locks_unsafe_for_binlog)
+1
+1 Expected
+SELECT innodb_locks_unsafe_for_binlog = @@SESSION.innodb_locks_unsafe_for_binlog;
+ERROR 42S22: Unknown column 'innodb_locks_unsafe_for_binlog' in 'field list'
+Expected error 'Readonly variable'

mysql-test/suite/sys_vars/r/sysvars_innodb.result

@@ -979,6 +979,18 @@ NUMERIC_BLOCK_SIZE	0
 ENUM_VALUE_LIST	NULL
 READ_ONLY	NO
 COMMAND_LINE_ARGUMENT	REQUIRED
+VARIABLE_NAME	INNODB_LOCKS_UNSAFE_FOR_BINLOG
+SESSION_VALUE	NULL
+DEFAULT_VALUE	OFF
+VARIABLE_SCOPE	GLOBAL
+VARIABLE_TYPE	BOOLEAN
+VARIABLE_COMMENT	Force InnoDB to not use next-key locking, to use only row-level locking.
+NUMERIC_MIN_VALUE	NULL
+NUMERIC_MAX_VALUE	NULL
+NUMERIC_BLOCK_SIZE	NULL
+ENUM_VALUE_LIST	OFF,ON
+READ_ONLY	YES
+COMMAND_LINE_ARGUMENT	NONE
 VARIABLE_NAME	INNODB_LOCK_WAIT_TIMEOUT
 SESSION_VALUE	50
 DEFAULT_VALUE	50

mysql-test/suite/sys_vars/t/innodb_locks_unsafe_for_binlog_basic.test

+
+
+################## mysql-test\t\innodb_locks_unsafe_for_binlog_basic.test #####
+#                                                                             #
+# Variable Name: innodb_locks_unsafe_for_binlog                               #
+# Scope: Global                                                               #
+# Access Type: Static                                                         #
+# Data Type: boolean                                                          #
+#                                                                             #
+#                                                                             #
+# Creation Date: 2008-02-07                                                   #
+# Author : Sharique Abdullah                                                      #
+#                                                                             #
+#                                                                             #
+# Description:Test Cases of Dynamic System Variable innodb_locks_unsafe_for_binlog#
+#             that checks the behavior of this variable in the following ways #
+#              * Value Check                                                  #
+#              * Scope Check                                                  #
+#                                                                             #
+# Reference: http://dev.mysql.com/doc/refman/5.1/en/                          #
+#  server-system-variables.html                                               #
+#                                                                             #
+###############################################################################
+
+--source include/have_innodb.inc
+
+--echo '#---------------------BS_STVARS_031_01----------------------#'
+####################################################################
+#   Displaying default value                                       #
+####################################################################
+SELECT COUNT(@@GLOBAL.innodb_locks_unsafe_for_binlog);
+--echo 1 Expected
+
+
+--echo '#---------------------BS_STVARS_031_02----------------------#'
+####################################################################
+#   Check if Value can set                                         #
+####################################################################
+
+--error ER_INCORRECT_GLOBAL_LOCAL_VAR
+SET @@GLOBAL.innodb_locks_unsafe_for_binlog=1;
+--echo Expected error 'Read only variable'
+
+SELECT COUNT(@@GLOBAL.innodb_locks_unsafe_for_binlog);
+--echo 1 Expected
+
+
+
+
+--echo '#---------------------BS_STVARS_031_03----------------------#'
+#################################################################
+# Check if the value in GLOBAL Table matches value in variable  #
+#################################################################
+
+--disable_warnings
+SELECT IF(@@GLOBAL.innodb_locks_unsafe_for_binlog, "ON", "OFF") = VARIABLE_VALUE
+FROM INFORMATION_SCHEMA.GLOBAL_VARIABLES
+WHERE VARIABLE_NAME='innodb_locks_unsafe_for_binlog';
+--enable_warnings
+--echo 1 Expected
+
+SELECT COUNT(@@GLOBAL.innodb_locks_unsafe_for_binlog);
+--echo 1 Expected
+
+--disable_warnings
+SELECT COUNT(VARIABLE_VALUE)
+FROM INFORMATION_SCHEMA.GLOBAL_VARIABLES 
+WHERE VARIABLE_NAME='innodb_locks_unsafe_for_binlog';
+--enable_warnings
+--echo 1 Expected
+
+
+
+--echo '#---------------------BS_STVARS_031_04----------------------#'
+################################################################################
+#  Check if accessing variable with and without GLOBAL point to same variable  #
+################################################################################
+SELECT @@innodb_locks_unsafe_for_binlog = @@GLOBAL.innodb_locks_unsafe_for_binlog;
+--echo 1 Expected
+
+
+
+--echo '#---------------------BS_STVARS_031_05----------------------#'
+################################################################################
+#   Check if innodb_locks_unsafe_for_binlog can be accessed with and without @@ sign #
+################################################################################
+
+SELECT COUNT(@@innodb_locks_unsafe_for_binlog);
+--echo 1 Expected
+
+--Error ER_INCORRECT_GLOBAL_LOCAL_VAR
+SELECT COUNT(@@local.innodb_locks_unsafe_for_binlog);
+--echo Expected error 'Variable is a GLOBAL variable'
+
+--Error ER_INCORRECT_GLOBAL_LOCAL_VAR
+SELECT COUNT(@@SESSION.innodb_locks_unsafe_for_binlog);
+--echo Expected error 'Variable is a GLOBAL variable'
+
+SELECT COUNT(@@GLOBAL.innodb_locks_unsafe_for_binlog);
+--echo 1 Expected
+
+--Error ER_BAD_FIELD_ERROR
+SELECT innodb_locks_unsafe_for_binlog = @@SESSION.innodb_locks_unsafe_for_binlog;
+--echo Expected error 'Readonly variable'
+
+

plugin/wsrep_info/mysql-test/wsrep_info/my.cnf

@@ -4,6 +4,7 @@
 [mysqld]
 binlog-format=row
 innodb-autoinc-lock-mode=2
+innodb-locks-unsafe-for-binlog=1
 wsrep_provider=@ENV.WSREP_PROVIDER
 
 [mysqld.1]

sql/mysqld.cc

@@ -5350,7 +5350,6 @@ static int init_server_components()
       MARIADB_REMOVED_OPTION("super-large-pages"),
 #endif
       MARIADB_REMOVED_OPTION("innodb-idle-flush-pct"),
-      MARIADB_REMOVED_OPTION("innodb-locks-unsafe-for-binlog"),
       MARIADB_REMOVED_OPTION("innodb-rollback-segments"),
       MARIADB_REMOVED_OPTION("innodb-stats-sample-pages"),
       MARIADB_REMOVED_OPTION("max-long-data-size"),

sql/upgrade_conf_file.cc

@@ -94,7 +94,6 @@ static const char *removed_variables[] =
 "innodb_large_prefix",
 "innodb_lazy_drop_table",
 "innodb_locking_fake_changes",
-"innodb_locks_unsafe_for_binlog",
 "innodb_log_arch_dir",
 "innodb_log_arch_expire_sec",
 "innodb_log_archive",

storage/innobase/handler/ha_innodb.cc

@@ -195,6 +195,7 @@ static char*	innobase_reset_all_monitor_counter;
 stopword table to be used */
 static char*	innobase_server_stopword_table;
 
+static my_bool innobase_locks_unsafe_for_binlog;
 my_bool innobase_rollback_on_timeout;
 static my_bool	innobase_create_status_file;
 my_bool	innobase_stats_on_metadata;
@@ -3986,6 +3987,7 @@ static int innodb_init_params()
 	}
 
 	srv_buf_pool_size = ulint(innobase_buffer_pool_size);
+	srv_locks_unsafe_for_binlog = (ibool) innobase_locks_unsafe_for_binlog;
 
 	if (innobase_open_files < 10) {
 		innobase_open_files = 300;
@@ -8740,7 +8742,7 @@ ha_innobase::delete_row(
 /**********************************************************************//**
 Removes a new lock set on a row, if it was not read optimistically. This can
 be called after a row has been read in the processing of an UPDATE or a DELETE
-query. */
+query, if the option innodb_locks_unsafe_for_binlog is set. */
 
 void
 ha_innobase::unlock_row(void)
@@ -8756,8 +8758,11 @@ ha_innobase::unlock_row(void)
 
 	switch (m_prebuilt->row_read_type) {
 	case ROW_READ_WITH_LOCKS:
-		if (m_prebuilt->trx->isolation_level > TRX_ISO_READ_COMMITTED)
+		if (!srv_locks_unsafe_for_binlog
+		    && m_prebuilt->trx->isolation_level
+		    > TRX_ISO_READ_COMMITTED) {
 			break;
+		}
 		/* fall through */
 	case ROW_READ_TRY_SEMI_CONSISTENT:
 		row_unlock_for_mysql(m_prebuilt, FALSE);
@@ -8784,12 +8789,20 @@ void ha_innobase::try_semi_consistent_read(bool yes)
 {
 	ut_ad(m_prebuilt->trx == thd_to_trx(ha_thd()));
 	/* Row read type is set to semi consistent read if this was
-	requested by the SQL layer and the transaction isolation level is
-	READ UNCOMMITTED or READ COMMITTED. */
-	m_prebuilt->row_read_type = yes
-		&& m_prebuilt->trx->isolation_level <= TRX_ISO_READ_COMMITTED
-		? ROW_READ_TRY_SEMI_CONSISTENT
-		: ROW_READ_WITH_LOCKS;
+	requested by the MySQL and either innodb_locks_unsafe_for_binlog
+	option is used or this session is using READ COMMITTED isolation
+	level. */
+
+	if (yes
+	    && (srv_locks_unsafe_for_binlog
+		|| m_prebuilt->trx->isolation_level
+		<= TRX_ISO_READ_COMMITTED)) {
+
+		m_prebuilt->row_read_type = ROW_READ_TRY_SEMI_CONSISTENT;
+
+	} else {
+		m_prebuilt->row_read_type = ROW_READ_WITH_LOCKS;
+	}
 }
 
 /******************************************************************//**
@@ -16544,7 +16557,9 @@ ha_innobase::store_lock(
 		if (sql_command == SQLCOM_CHECKSUM
 		    || sql_command == SQLCOM_CREATE_SEQUENCE
 		    || (sql_command == SQLCOM_ANALYZE && lock_type == TL_READ)
-		    || (trx->isolation_level <= TRX_ISO_READ_COMMITTED
+		    || ((srv_locks_unsafe_for_binlog
+			|| trx->isolation_level <= TRX_ISO_READ_COMMITTED)
+			&& trx->isolation_level != TRX_ISO_SERIALIZABLE
 			&& (lock_type == TL_READ
 			    || lock_type == TL_READ_NO_INSERT)
 			&& (sql_command == SQLCOM_INSERT_SELECT
@@ -16553,8 +16568,10 @@ ha_innobase::store_lock(
 			    || sql_command == SQLCOM_CREATE_SEQUENCE
 			    || sql_command == SQLCOM_CREATE_TABLE))) {
 
-			/* If the transaction isolation level is
-			READ UNCOMMITTED or READ COMMITTED and we are executing
+			/* If we either have innobase_locks_unsafe_for_binlog
+			option set or this session is using READ COMMITTED
+			isolation level and isolation level of the transaction
+			is not set to serializable and MySQL is doing
 			INSERT INTO...SELECT or REPLACE INTO...SELECT
 			or UPDATE ... = (SELECT ...) or CREATE  ...
 			SELECT... without FOR UPDATE or IN SHARE
@@ -18991,6 +19008,11 @@ static MYSQL_SYSVAR_ENUM(flush_method, srv_file_flush_method,
   NULL, NULL, IF_WIN(SRV_ALL_O_DIRECT_FSYNC, SRV_O_DIRECT),
   &innodb_flush_method_typelib);
 
+static MYSQL_SYSVAR_BOOL(locks_unsafe_for_binlog, innobase_locks_unsafe_for_binlog,
+  PLUGIN_VAR_NOCMDARG | PLUGIN_VAR_READONLY,
+  "Force InnoDB to not use next-key locking, to use only row-level locking.",
+  NULL, NULL, FALSE);
+
 static MYSQL_SYSVAR_STR(log_group_home_dir, srv_log_group_home_dir,
   PLUGIN_VAR_RQCMDARG | PLUGIN_VAR_READONLY,
   "Path to ib_logfile0", NULL, NULL, NULL);
@@ -19887,6 +19909,7 @@ static struct st_mysql_sys_var* innobase_system_variables[]= {
   MYSQL_SYSVAR(ft_min_token_size),
   MYSQL_SYSVAR(ft_num_word_optimize),
   MYSQL_SYSVAR(ft_sort_pll_degree),
+  MYSQL_SYSVAR(locks_unsafe_for_binlog),
   MYSQL_SYSVAR(lock_wait_timeout),
   MYSQL_SYSVAR(deadlock_detect),
   MYSQL_SYSVAR(deadlock_report),

storage/innobase/include/row0mysql.h

@@ -262,8 +262,8 @@ row_update_for_mysql(
 	row_prebuilt_t*		prebuilt)
 	MY_ATTRIBUTE((warn_unused_result));
 
-/** This can only be used when the current transaction is at
-READ COMMITTED or READ UNCOMMITTED isolation level.
+/** This can only be used when srv_locks_unsafe_for_binlog is TRUE or this
+session is using a READ COMMITTED or READ UNCOMMITTED isolation level.
 Before calling this function row_search_mvcc() must have
 initialized prebuilt->new_rec_locks to store the information which new
 record locks really were set. This function removes a newly set
@@ -573,9 +573,8 @@ struct row_prebuilt_t {
 	ulint		row_read_type;	/*!< ROW_READ_WITH_LOCKS if row locks
 					should be the obtained for records
 					under an UPDATE or DELETE cursor.
-					At READ UNCOMMITTED or
-					READ COMMITTED isolation level,
-					this can be set to
+					If innodb_locks_unsafe_for_binlog
+					is TRUE, this can be set to
 					ROW_READ_TRY_SEMI_CONSISTENT, so that
 					if the row under an UPDATE or DELETE
 					cursor was locked by another
@@ -597,7 +596,8 @@ struct row_prebuilt_t {
 					cases; note that this breaks
 					serializability. */
 	ulint		new_rec_locks;	/*!< normally 0; if
-					the session is using READ
+					srv_locks_unsafe_for_binlog is
+					TRUE or session is using READ
 					COMMITTED or READ UNCOMMITTED
 					isolation level, set in
 					row_search_mvcc() if we set a new

storage/innobase/include/srv0srv.h

@@ -168,6 +168,10 @@ extern my_bool	high_level_read_only;
 dictionary tables are in the system tablespace 0 */
 extern my_bool	srv_file_per_table;
 
+/** Place locks to records only i.e. do not use next-key locking except
+on duplicate key checking and foreign key checking */
+extern ibool   srv_locks_unsafe_for_binlog;
+
 /** Sort buffer size in index creation */
 extern ulong	srv_sort_buf_size;
 /** Maximum modification log file size for online index creation */

storage/innobase/lock/lock0lock.cc

@@ -2473,8 +2473,8 @@ lock_rec_inherit_to_gap(hash_cell_t &heir_cell, const page_id_t heir,
 {
   ut_ad(!from_split || heir_heap_no == PAGE_HEAP_NO_SUPREMUM);
 
-  /* At READ UNCOMMITTED or READ COMMITTED isolation level,
-  we do not want locks set
+  /* If srv_locks_unsafe_for_binlog is TRUE or session is using
+  READ COMMITTED isolation level, we do not want locks set
   by an UPDATE or a DELETE to be inherited as gap type locks. But we
   DO want S-locks/X-locks(taken for replace) set by a consistency
   constraint to be inherited also then. */
@@ -2485,7 +2485,8 @@ lock_rec_inherit_to_gap(hash_cell_t &heir_cell, const page_id_t heir,
     trx_t *lock_trx= lock->trx;
     if (!lock->trx->is_not_inheriting_locks() &&
         !lock->is_insert_intention() &&
-        (lock_trx->isolation_level > TRX_ISO_READ_COMMITTED ||
+        ((!srv_locks_unsafe_for_binlog
+          && lock_trx->isolation_level > TRX_ISO_READ_COMMITTED) ||
          /* When we are in a page split (not purge), then we don't set a lock
          on supremum if the donor lock type is LOCK_REC_NOT_GAP. That is, do
          not create bogus gap locks for non-gap locks for READ UNCOMMITTED and

storage/innobase/row/row0mysql.cc

@@ -1746,8 +1746,8 @@ row_update_for_mysql(row_prebuilt_t* prebuilt)
 	DBUG_RETURN(err);
 }
 
-/** This can only be used when the current transaction is at
-READ COMMITTED or READ UNCOMMITTED isolation level.
+/** This can only be used when srv_locks_unsafe_for_binlog is TRUE or this
+session is using a READ COMMITTED or READ UNCOMMITTED isolation level.
 Before calling this function row_search_mvcc() must have
 initialized prebuilt->new_rec_locks to store the information which new
 record locks really were set. This function removes a newly set
@@ -1766,7 +1766,15 @@ row_unlock_for_mysql(
 {
 	if (prebuilt->new_rec_locks == 1 && prebuilt->index->is_clust()) {
 		trx_t* trx = prebuilt->trx;
-		ut_ad(trx->isolation_level <= TRX_ISO_READ_COMMITTED);
+		if (UNIV_UNLIKELY
+		    (!srv_locks_unsafe_for_binlog
+		     && trx->isolation_level > TRX_ISO_READ_COMMITTED)) {
+			ib::error() << "Calling row_unlock_for_mysql though"
+				" innodb_locks_unsafe_for_binlog is FALSE and this"
+				" session is not using READ COMMITTED isolation"
+				" level.";
+			return;
+		}
 		trx->op_info = "unlock_row";
 
 		const rec_t*	rec;

storage/innobase/row/row0sel.cc

@@ -1058,14 +1058,16 @@ row_sel_get_clust_rec(
 		/* Try to place a lock on the index record */
 		trx_t* trx = thr_get_trx(thr);
 
-		/* At READ UNCOMMITTED or READ COMMITTED isolation level
+		/* If innodb_locks_unsafe_for_binlog option is used
+		or this session is using READ COMMITTED or lower isolation level
 		we lock only the record, i.e., next-key locking is
 		not used. */
 		err = lock_clust_rec_read_check_and_lock(
 			0, btr_pcur_get_block(&plan->clust_pcur),
 			clust_rec, index, offsets,
 			node->row_lock_mode,
-			trx->isolation_level <= TRX_ISO_READ_COMMITTED
+			(srv_locks_unsafe_for_binlog
+			    || trx->isolation_level <= TRX_ISO_READ_COMMITTED)
 			? LOCK_REC_NOT_GAP : LOCK_ORDINARY,
 			thr);
 
@@ -1834,10 +1836,13 @@ row_sel(
 						  index->n_core_fields,
 						  ULINT_UNDEFINED, &heap);
 
-			/* At READ UNCOMMITTED or READ COMMITTED
-			isolation level, we lock only the record,
-			i.e., next-key locking is not used. */
-			if (trx->isolation_level <= TRX_ISO_READ_COMMITTED) {
+			/* If innodb_locks_unsafe_for_binlog option is used
+			or this session is using READ COMMITTED or lower isolation
+			level, we lock only the record, i.e., next-key
+			locking is not used. */
+			if (srv_locks_unsafe_for_binlog
+			    || trx->isolation_level
+			    <= TRX_ISO_READ_COMMITTED) {
 				if (page_rec_is_supremum(next_rec)) {
 					goto skip_lock;
 				}
@@ -1894,10 +1899,12 @@ row_sel(
 					  index->n_core_fields,
 					  ULINT_UNDEFINED, &heap);
 
-		/* At READ UNCOMMITTED or READ COMMITTED isolation level,
+		/* If innodb_locks_unsafe_for_binlog option is used
+		or this session is using READ COMMITTED or lower isolation level,
 		we lock only the record, i.e., next-key locking is
 		not used. */
-		if (trx->isolation_level <= TRX_ISO_READ_COMMITTED
+		if (srv_locks_unsafe_for_binlog
+		    || trx->isolation_level <= TRX_ISO_READ_COMMITTED
 		    || dict_index_is_spatial(index)) {
 
 			if (page_rec_is_supremum(rec)) {
@@ -4771,9 +4778,18 @@ row_search_mvcc(
 		pcur->old_rec = nullptr;
 
 		if (index->is_spatial()) {
+			bool    need_pred_lock;
+
+			need_pred_lock = (set_also_gap_locks
+					  && !(srv_locks_unsafe_for_binlog
+					      || trx->isolation_level
+						 <= TRX_ISO_READ_COMMITTED)
+					  && prebuilt->select_lock_type
+						 != LOCK_NONE);
+
 			if (!prebuilt->rtr_info) {
 				prebuilt->rtr_info = rtr_create_rtr_info(
-					set_also_gap_locks, true,
+					need_pred_lock, true,
 					btr_pcur_get_btr_cur(pcur), index);
 				prebuilt->rtr_info->search_tuple = search_tuple;
 				prebuilt->rtr_info->search_mode = mode;
@@ -4782,7 +4798,7 @@ row_search_mvcc(
 			} else {
 				rtr_info_reinit_in_cursor(
 					btr_pcur_get_btr_cur(pcur),
-					index, set_also_gap_locks);
+					index, need_pred_lock);
 				prebuilt->rtr_info->search_tuple = search_tuple;
 				prebuilt->rtr_info->search_mode = mode;
 			}
@@ -4807,6 +4823,9 @@ row_search_mvcc(
 
 		if (!moves_up
 		    && set_also_gap_locks
+		    && !(srv_locks_unsafe_for_binlog
+			 || trx->isolation_level <= TRX_ISO_READ_COMMITTED)
+		    && prebuilt->select_lock_type != LOCK_NONE
 		    && !page_rec_is_supremum(rec)
 		    && !dict_index_is_spatial(index)) {
 
@@ -4927,12 +4946,16 @@ row_search_mvcc(
 	if (page_rec_is_supremum(rec)) {
 
 		if (set_also_gap_locks
+		    && !(srv_locks_unsafe_for_binlog
+			 || trx->isolation_level <= TRX_ISO_READ_COMMITTED)
+		    && prebuilt->select_lock_type != LOCK_NONE
 		    && !dict_index_is_spatial(index)) {
 
 			/* Try to place a lock on the index record */
 
-			/* If the transaction isolation level is
-			READ UNCOMMITTED or READ COMMITTED,
+			/* If innodb_locks_unsafe_for_binlog option is used
+			or this session is using a READ COMMITTED or lower isolation
+			level we do not lock gaps. Supremum record is really
 			we do not lock gaps. Supremum record is really
 			a gap and therefore we do not set locks there. */
 
@@ -5072,7 +5095,17 @@ row_search_mvcc(
 		if (cmp_dtuple_rec(search_tuple, rec, index, offsets)) {
 
 			if (set_also_gap_locks
+			    && !(srv_locks_unsafe_for_binlog
+				 || trx->isolation_level
+				 <= TRX_ISO_READ_COMMITTED)
+			    && prebuilt->select_lock_type != LOCK_NONE
 			    && !dict_index_is_spatial(index)) {
+
+				/* Try to place a gap lock on the index
+				record only if innodb_locks_unsafe_for_binlog
+				option is not set or this session is not
+				using a READ COMMITTED or lower isolation level. */
+
 				err = sel_set_rec_lock(
 					pcur,
 					rec, index, offsets,
@@ -5108,7 +5141,17 @@ row_search_mvcc(
 						 index, offsets)) {
 
 			if (set_also_gap_locks
+			    && !(srv_locks_unsafe_for_binlog
+				 || trx->isolation_level
+				 <= TRX_ISO_READ_COMMITTED)
+			    && prebuilt->select_lock_type != LOCK_NONE
 			    && !dict_index_is_spatial(index)) {
+
+				/* Try to place a gap lock on the index
+				record only if innodb_locks_unsafe_for_binlog
+				option is not set or this session is not
+				using a READ COMMITTED or lower isolation level. */
+
 				err = sel_set_rec_lock(
 					pcur,
 					rec, index, offsets,
@@ -5148,9 +5191,15 @@ row_search_mvcc(
 		is a non-delete marked record, then it is enough to lock its
 		existence with LOCK_REC_NOT_GAP. */
 
+		/* If innodb_locks_unsafe_for_binlog option is used
+		or this session is using a READ COMMITED isolation
+		level we lock only the record, i.e., next-key locking is
+		not used. */
+
 		unsigned lock_type;
 
-		if (trx->isolation_level <= TRX_ISO_READ_COMMITTED) {
+		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. */
@@ -5240,7 +5289,9 @@ row_search_mvcc(
 		switch (err) {
 			const rec_t*	old_vers;
 		case DB_SUCCESS_LOCKED_REC:
-			if (trx->isolation_level <= TRX_ISO_READ_COMMITTED) {
+			if (srv_locks_unsafe_for_binlog
+			    || trx->isolation_level
+			    <= TRX_ISO_READ_COMMITTED) {
 				/* Note that a record of
 				prebuilt->index was locked. */
 				prebuilt->new_rec_locks = 1;
@@ -5539,7 +5590,9 @@ row_search_mvcc(
 			break;
 		case DB_SUCCESS_LOCKED_REC:
 			ut_a(clust_rec != NULL);
-			if (trx->isolation_level <= TRX_ISO_READ_COMMITTED) {
+			if (srv_locks_unsafe_for_binlog
+			    || trx->isolation_level
+			    <= TRX_ISO_READ_COMMITTED) {
 				/* Note that the clustered index record
 				was locked. */
 				prebuilt->new_rec_locks = 2;
@@ -5562,7 +5615,8 @@ row_search_mvcc(
 
 			/* The record is delete marked: we can skip it */
 
-			if (trx->isolation_level <= TRX_ISO_READ_COMMITTED
+			if ((srv_locks_unsafe_for_binlog
+			     || trx->isolation_level <= TRX_ISO_READ_COMMITTED)
 			    && prebuilt->select_lock_type != LOCK_NONE) {
 
 				/* No need to keep a lock on a delete-marked
@@ -5881,7 +5935,8 @@ row_search_mvcc(
 				moves_up, &mtr);
 		}
 
-		if (trx->isolation_level <= TRX_ISO_READ_COMMITTED
+		if ((srv_locks_unsafe_for_binlog
+		     || trx->isolation_level <= TRX_ISO_READ_COMMITTED)
 		    && !same_user_rec) {
 
 			/* Since we were not able to restore the cursor

storage/innobase/srv/srv0srv.cc

@@ -127,6 +127,9 @@ my_bool	srv_file_per_table;
 is greater than SRV_FORCE_NO_TRX_UNDO. */
 my_bool	high_level_read_only;
 
+/** Place locks to records only i.e. do not use next-key locking except
+on duplicate key checking and foreign key checking */
+ibool  srv_locks_unsafe_for_binlog;
 /** Sort buffer size in index creation */
 ulong	srv_sort_buf_size;
 /** Maximum modification log file size for online index creation */