MDEV-13115: Implement SELECT SKIP LOCKED

Adds an implementation for SELECT ... FOR UPDATE SKIP LOCKED /
SELECT ... LOCK IN SHARED MODE SKIP LOCKED

This is implemented only InnoDB at the moment, not in RockDB yet.

This adds a new hander flag HA_CAN_SKIP_LOCKED than
will be used when the storage engine advertises the flag.

When a storage engine indicates this flag it will get
TL_WRITE_SKIP_LOCKED and TL_READ_SKIP_LOCKED transaction types.

The Lex structure has been updated to store both the FOR UPDATE/LOCK IN
SHARE as well as the SKIP LOCKED so the SHOW CREATE VIEW
implementation is simplier.

"SELECT FOR UPDATE ... SKIP LOCKED" combined with CREATE TABLE AS or
INSERT.. SELECT on the result set is not safe for STATEMENT based
replication. MIXED replication will replicate this as row based events."

Thanks to guidance from Facebook commit
193896c466
This helped verify basic test case, and components that need implementing
(even though every part was implemented differently).

Thanks Marko for guidance on simplier InnoDB implementation.

Reviewers: Marko, Monty

include/thr_lock.h

@@ -47,6 +47,8 @@ enum thr_lock_type { TL_IGNORE=-1,
 		     TL_READ_HIGH_PRIORITY,
 		     /* READ, Don't allow concurrent insert */
 		     TL_READ_NO_INSERT,
+		     /* READ, but skip locks if found */
+		     TL_READ_SKIP_LOCKED,
 		     /* 
 			Write lock, but allow other threads to read / write.
 			Used by BDB tables in MySQL to mark that someone is
@@ -67,6 +69,8 @@ enum thr_lock_type { TL_IGNORE=-1,
                      TL_WRITE_DEFAULT,
 		     /* WRITE lock that has lower priority than TL_READ */
 		     TL_WRITE_LOW_PRIORITY,
+		     /* WRITE, but skip locks if found */
+		     TL_WRITE_SKIP_LOCKED,
 		     /* Normal WRITE lock */
 		     TL_WRITE,
 		     /* Abort new lock request with an error */

mysql-test/main/flush_read_lock.result

@@ -1096,6 +1096,14 @@ Success: FTWRL is blocked when 'select f2_base()' is active in another connectio
 Success: Was able to run 'select f2_temp()' under FTWRL.
 Success: Was able to run 'select f2_temp()' with FTWRL active in another connection.
 Success: Was able to run FTWRL while 'select f2_temp()' was active in another connection.
+# 30.f) SELECT ... FOR UPDATE SKIP LOCKED is incompatible with FTWRL.
+Success: Was not able to run 'select count(*) from t1_base for update skip locked' under FTWRL.
+Success: 'select count(*) from t1_base for update skip locked' is blocked by FTWRL active in another connection.
+Success: FTWRL is blocked when 'select count(*) from t1_base for update skip locked' is active in another connection.
+# 30.g) SELECT ... LOCK IN SHARE MODE SKIP LOCKED is compatible with FTWRL.
+Success: Was able to run 'select count(*) from t1_base lock in share mode skip locked' under FTWRL.
+Success: Was able to run 'select count(*) from t1_base lock in share mode skip locked' with FTWRL active in another connection.
+Success: Was able to run FTWRL while 'select count(*) from t1_base lock in share mode skip locked' was active in another connection.
 #
 # 31) Compatibility of SET statement with FTWRL depends on its
 #     expression and on whether it is a special SET statement.

mysql-test/main/flush_read_lock.test

@@ -1323,6 +1323,16 @@ let $statement= select f2_temp();
 let $cleanup_stmt= delete from t1_temp limit 1;
 --source include/check_ftwrl_compatible.inc
 
+--echo # 30.f) SELECT ... FOR UPDATE SKIP LOCKED is incompatible with FTWRL.
+let $statement= select count(*) from t1_base for update skip locked;
+let $cleanup_stmt1= ;
+--source include/check_ftwrl_incompatible.inc
+
+--echo # 30.g) SELECT ... LOCK IN SHARE MODE SKIP LOCKED is compatible with FTWRL.
+let $statement= select count(*) from t1_base lock in share mode skip locked;
+let $cleanup_stmt= ;
+--source include/check_ftwrl_compatible.inc
+
 
 --echo #
 --echo # 31) Compatibility of SET statement with FTWRL depends on its

mysql-test/main/view.result

@@ -6842,4 +6842,34 @@ ERROR 42S22: Unknown column 't1.x' in 'on clause'
 CREATE TABLE t4 AS SELECT * FROM t1 JOIN t2 ON t1.x > t2.b;
 ERROR 42S22: Unknown column 't1.x' in 'on clause'
 DROP TABLE t1,t2,t3;
+#
 # End of 10.4 tests
+#
+#
+# MDEV-13115: SELECT .. SKIP LOCKED - ensure SHOW CREATE VIEW is correct
+#
+CREATE TABLE t1 (id int, foo int);
+CREATE VIEW v1 AS  SELECT id, IFNULL(foo,'') AS foo FROM t1 LOCK IN SHARE MODE;
+CREATE VIEW v2 AS  SELECT id, IFNULL(foo,'') AS foo FROM t1 FOR UPDATE;
+CREATE VIEW v3 AS  SELECT id, IFNULL(foo,'') AS foo FROM t1 LOCK IN SHARE MODE SKIP LOCKED;
+CREATE VIEW v4 AS  SELECT id, IFNULL(foo,'') AS foo FROM t1 FOR UPDATE SKIP LOCKED;
+SHOW CREATE VIEW v1;
+View	Create View	character_set_client	collation_connection
+v1	CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `v1` AS select `t1`.`id` AS `id`,ifnull(`t1`.`foo`,'') AS `foo` from `t1` lock in share mode	latin1	latin1_swedish_ci
+SHOW CREATE VIEW v2;
+View	Create View	character_set_client	collation_connection
+v2	CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `v2` AS select `t1`.`id` AS `id`,ifnull(`t1`.`foo`,'') AS `foo` from `t1` for update	latin1	latin1_swedish_ci
+SHOW CREATE VIEW v3;
+View	Create View	character_set_client	collation_connection
+v3	CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `v3` AS select `t1`.`id` AS `id`,ifnull(`t1`.`foo`,'') AS `foo` from `t1` lock in share mode skip locked	latin1	latin1_swedish_ci
+SHOW CREATE VIEW v4;
+View	Create View	character_set_client	collation_connection
+v4	CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`localhost` SQL SECURITY DEFINER VIEW `v4` AS select `t1`.`id` AS `id`,ifnull(`t1`.`foo`,'') AS `foo` from `t1` for update skip locked	latin1	latin1_swedish_ci
+Drop View v1;
+Drop View v2;
+Drop View v3;
+Drop View v4;
+Drop table t1;
+#
+# End of 10.6 tests
+#

mysql-test/main/view.test

@@ -6562,4 +6562,34 @@ CREATE TABLE t4 AS SELECT * FROM t1 JOIN t2 ON t1.x > t2.b;
 
 DROP TABLE t1,t2,t3;
 
+--echo #
 --echo # End of 10.4 tests
+--echo #
+
+--echo #
+--echo # MDEV-13115: SELECT .. SKIP LOCKED - ensure SHOW CREATE VIEW is correct
+--echo #
+
+# Note: MDEV-10063 - LOCK IN SHARE MODE/FOR UPDATE/SKIP LOCKED ignored in views
+
+CREATE TABLE t1 (id int, foo int);
+CREATE VIEW v1 AS  SELECT id, IFNULL(foo,'') AS foo FROM t1 LOCK IN SHARE MODE;
+CREATE VIEW v2 AS  SELECT id, IFNULL(foo,'') AS foo FROM t1 FOR UPDATE;
+CREATE VIEW v3 AS  SELECT id, IFNULL(foo,'') AS foo FROM t1 LOCK IN SHARE MODE SKIP LOCKED;
+CREATE VIEW v4 AS  SELECT id, IFNULL(foo,'') AS foo FROM t1 FOR UPDATE SKIP LOCKED;
+
+SHOW CREATE VIEW v1;
+SHOW CREATE VIEW v2;
+SHOW CREATE VIEW v3;
+SHOW CREATE VIEW v4;
+
+#Cleanup
+Drop View v1;
+Drop View v2;
+Drop View v3;
+Drop View v4;
+Drop table t1;
+
+--echo #
+--echo # End of 10.6 tests
+--echo #

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

@@ -0,0 +1,72 @@
+connect  con1,localhost,root,,;
+SET SESSION innodb_lock_wait_timeout=1;
+connection default;
+SET SESSION innodb_lock_wait_timeout=1;
+# Case 1: Test primary index - CREATE TABLE .. SELECT .. SKIP LOCKED
+CREATE TABLE t1(
+seat_id    INT,
+state      INT,
+PRIMARY KEY(seat_id)
+) ENGINE=InnoDB;
+INSERT INTO t1 VALUES(1,0), (2,0), (3,0), (4,0);
+BEGIN;
+SELECT * FROM t1 WHERE state = 0 LIMIT 2 LOCK IN SHARE MODE;
+seat_id	state
+1	0
+2	0
+connection con1;
+BEGIN;
+CREATE TEMPORARY TABLE s0 AS SELECT * FROM t1 WHERE state = 0 LIMIT 2 LOCK IN SHARE MODE;
+SELECT * FROM s0;
+seat_id	state
+1	0
+2	0
+CREATE TEMPORARY TABLE s1 AS SELECT * FROM t1 WHERE state = 0 LIMIT 2 FOR UPDATE NOWAIT SKIP LOCKED;
+SELECT * FROM s1;
+seat_id	state
+3	0
+4	0
+connection default;
+CREATE TEMPORARY TABLE s0 AS SELECT * FROM t1 WHERE state = 0 LOCK IN SHARE MODE NOWAIT SKIP LOCKED;
+SELECT * FROM s0;
+seat_id	state
+1	0
+2	0
+COMMIT;
+DROP TABLE s0;
+connection con1;
+COMMIT;
+DROP TABLE s0, s1;
+connection default;
+# Case 2: Test primary index - INSERT .. SELECT .. SKIP LOCKED
+CREATE TABLE t2 LIKE t1;
+BEGIN;
+SELECT * FROM t1 WHERE state = 0 LIMIT 2 LOCK IN SHARE MODE;
+seat_id	state
+1	0
+2	0
+connection con1;
+BEGIN;
+INSERT INTO t2 SELECT * FROM t1 WHERE state = 0 LIMIT 2 LOCK IN SHARE MODE RETURNING seat_id, state;
+seat_id	state
+1	0
+2	0
+CREATE TEMPORARY TABLE t2s LIKE t1;
+INSERT INTO t2s SELECT * FROM t1 WHERE state = 0 LIMIT 2 FOR UPDATE NOWAIT SKIP LOCKED RETURNING seat_id, state;
+seat_id	state
+3	0
+4	0
+connection default;
+CREATE TEMPORARY TABLE t2s LIKE t1;
+INSERT INTO t2s SELECT * FROM t1 WHERE state = 0 LOCK IN SHARE MODE NOWAIT SKIP LOCKED RETURNING seat_id, state;
+seat_id	state
+1	0
+2	0
+COMMIT;
+DROP TABLE t2s;
+connection con1;
+COMMIT;
+DROP TABLE t2s;
+DROP TABLE t2;
+connection default;
+DROP TABLE t1;

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

@@ -0,0 +1,73 @@
+#
+# MDEV-13115 Implement SKIP LOCKED
+#
+--source include/have_innodb.inc
+
+
+connect (con1,localhost,root,,);
+SET SESSION innodb_lock_wait_timeout=1;
+
+connection default;
+SET SESSION innodb_lock_wait_timeout=1;
+
+--echo # Case 1: Test primary index - CREATE TABLE .. SELECT .. SKIP LOCKED
+CREATE TABLE t1(
+        seat_id    INT,
+        state      INT,
+        PRIMARY KEY(seat_id)
+) ENGINE=InnoDB;
+
+INSERT INTO t1 VALUES(1,0), (2,0), (3,0), (4,0);
+
+BEGIN;
+SELECT * FROM t1 WHERE state = 0 LIMIT 2 LOCK IN SHARE MODE;
+
+connection con1;
+BEGIN;
+CREATE TEMPORARY TABLE s0 AS SELECT * FROM t1 WHERE state = 0 LIMIT 2 LOCK IN SHARE MODE;
+SELECT * FROM s0;
+
+CREATE TEMPORARY TABLE s1 AS SELECT * FROM t1 WHERE state = 0 LIMIT 2 FOR UPDATE NOWAIT SKIP LOCKED;
+SELECT * FROM s1;
+
+connection default;
+CREATE TEMPORARY TABLE s0 AS SELECT * FROM t1 WHERE state = 0 LOCK IN SHARE MODE NOWAIT SKIP LOCKED;
+SELECT * FROM s0;
+
+COMMIT;
+DROP TABLE s0;
+
+connection con1;
+COMMIT;
+DROP TABLE s0, s1;
+
+connection default;
+
+--echo # Case 2: Test primary index - INSERT .. SELECT .. SKIP LOCKED
+
+CREATE TABLE t2 LIKE t1;
+
+BEGIN;
+SELECT * FROM t1 WHERE state = 0 LIMIT 2 LOCK IN SHARE MODE;
+
+connection con1;
+BEGIN;
+
+INSERT INTO t2 SELECT * FROM t1 WHERE state = 0 LIMIT 2 LOCK IN SHARE MODE RETURNING seat_id, state;
+
+CREATE TEMPORARY TABLE t2s LIKE t1;
+INSERT INTO t2s SELECT * FROM t1 WHERE state = 0 LIMIT 2 FOR UPDATE NOWAIT SKIP LOCKED RETURNING seat_id, state;
+
+connection default;
+CREATE TEMPORARY TABLE t2s LIKE t1;
+INSERT INTO t2s SELECT * FROM t1 WHERE state = 0 LOCK IN SHARE MODE NOWAIT SKIP LOCKED RETURNING seat_id, state;
+COMMIT;
+DROP TABLE t2s;
+
+connection con1;
+COMMIT;
+DROP TABLE t2s;
+DROP TABLE t2;
+
+connection default;
+DROP TABLE t1;

mysql-test/suite/perfschema/r/start_server_low_digest_sql_length.result

@@ -8,5 +8,5 @@ SELECT 1+1+1+1+1+1+1+1+1+1+1+1+1+1+1+1+1+1+1+1+1+1+1+1+1+1+1+1+1+1+1+1+1+1+1+1+1
 ####################################
 SELECT event_name, digest, digest_text, sql_text FROM events_statements_history_long;
 event_name	digest	digest_text	sql_text
-statement/sql/select	ade774bdfbc132a71810ede8ef469660	SELECT ? + ? + 	SELECT ...
+statement/sql/select	c6c8d8d7523f187bbdfda0d54e15b47a	SELECT ? + ? + 	SELECT ...
-statement/sql/truncate	0f84807fb4a75d0f391f8a93e7c3c182	TRUNCATE TABLE 	truncat...
+statement/sql/truncate	13d7058e7e84d7b7133f0a4f74e0b199	TRUNCATE TABLE 	truncat...

mysql-test/suite/rpl/r/rpl_unsafe_statements.result

@@ -63,4 +63,33 @@ connection slave;
 include/diff_tables.inc [master:t1, slave:t1]
 connection master;
 DROP TABLE t1;
+CREATE TABLE t1(i INT,PRIMARY KEY(i)) ENGINE=INNODB;
+CREATE TABLE t2(i INT,PRIMARY KEY(i)) ENGINE=INNODB;
+INSERT INTO t1 (i) VALUES (1),(2),(3),(4),(5);
+connect  con1, localhost, root,;
+START TRANSACTION;
+SELECT i FROM t1 WHERE i=3 FOR UPDATE;
+i
+3
+connection master;
+INSERT INTO t2 SELECT i FROM t1 LOCK IN SHARE MODE SKIP LOCKED;
+CREATE TABLE t3 AS SELECT i FROM t1 LOCK IN SHARE MODE SKIP LOCKED;
+SELECT * FROM t2 ORDER BY i;
+i
+1
+2
+4
+5
+SELECT * FROM t3 ORDER BY i;
+i
+1
+2
+4
+5
+connection slave;
+include/diff_tables.inc [master:t2, slave:t2]
+include/diff_tables.inc [master:t3, slave:t3]
+disconnect con1;
+connection master;
+DROP TABLE t1, t2, t3;
 include/rpl_end.inc

mysql-test/suite/rpl/t/rpl_unsafe_statements.test

@@ -19,6 +19,7 @@
 #    Case-2: BINLOG_STMT_UNSAFE_WRITE_AUTOINC_SELECT
 #    Case-3: BINLOG_STMT_UNSAFE_AUTOINC_NOT_FIRST
 #    Case-4: BINLOG_STMT_UNSAFE_INSERT_TWO_KEYS
+#    Case-5: BINLOG_STMT_UNSAFE_SKIP_LOCKED
 ################################################################################
 
 --source include/have_innodb.inc
@@ -173,4 +174,42 @@ COMMIT;
 --connection master
 DROP TABLE t1;
 
+# Case-5: BINLOG_STMT_UNSAFE_SKIP_LOCKED
+#  INSERT... ON KEY UPDATE SKIP LOCKED is unsafe Statement
+
+# Step-5.1: Create a table some index
+CREATE TABLE t1(i INT,PRIMARY KEY(i)) ENGINE=INNODB;
+CREATE TABLE t2(i INT,PRIMARY KEY(i)) ENGINE=INNODB;
+
+# Step-5.2: Inserting some values
+INSERT INTO t1 (i) VALUES (1),(2),(3),(4),(5);
+
+# Step-5.3: Lock one of the values
+connect (con1, localhost, root,);
+START TRANSACTION;
+SELECT i FROM t1 WHERE i=3 FOR UPDATE;
+
+# Step-5.4: Create non-deterministic inserts/tables
+--connection master
+INSERT INTO t2 SELECT i FROM t1 LOCK IN SHARE MODE SKIP LOCKED;
+CREATE TABLE t3 AS SELECT i FROM t1 LOCK IN SHARE MODE SKIP LOCKED;
+SELECT * FROM t2 ORDER BY i;
+SELECT * FROM t3 ORDER BY i;
+
+# Step-5.5: Sync slave with master
+--sync_slave_with_master
+
+# Step-5.6: Diff master-replica tables insert statements are in sync
+--let $diff_tables=master:t2, slave:t2
+--source include/diff_tables.inc
+
+# Step-5.7: Diff master-replica tables create select table is in sync
+--let $diff_tables=master:t3, slave:t3
+--source include/diff_tables.inc
+
+# Step-5.8: Cleanup
+--disconnect con1
+--connection master
+DROP TABLE t1, t2, t3;
+
 --source include/rpl_end.inc

sql/handler.h

@@ -353,7 +353,10 @@ enum chf_create_flags {
  */
 #define HA_ONLINE_ANALYZE             (1ULL << 59)
 
-#define HA_LAST_TABLE_FLAG HA_ONLINE_ANALYZE
+/* Implements SELECT ... FOR UPDATE SKIP LOCKED */
+#define HA_CAN_SKIP_LOCKED  (1ULL << 60)
+
+#define HA_LAST_TABLE_FLAG HA_CAN_SKIP_LOCKED
 
 
 /* bits in index_flags(index_number) for what you can do with index */

sql/lex.h

@@ -352,6 +352,7 @@ static SYMBOL symbols[] = {
   { "LOCALTIME",	SYM(NOW_SYM)},
   { "LOCALTIMESTAMP",	SYM(NOW_SYM)},
   { "LOCK",		SYM(LOCK_SYM)},
+  { "LOCKED",		SYM(LOCKED_SYM)},
   { "LOCKS",		SYM(LOCKS_SYM)},
   { "LOGFILE",		SYM(LOGFILE_SYM)},
   { "LOGS",		SYM(LOGS_SYM)},
@@ -584,6 +585,7 @@ static SYMBOL symbols[] = {
   { "SIGNAL",           SYM(SIGNAL_SYM)},
   { "SIGNED",		SYM(SIGNED_SYM)},
   { "SIMPLE",		SYM(SIMPLE_SYM)},
+  { "SKIP",             SYM(SKIP_SYM)},
   { "SLAVE",            SYM(SLAVE)},
   { "SLAVES",           SYM(SLAVES)},
   { "SLAVE_POS",        SYM(SLAVE_POS_SYM)},

sql/share/errmsg-utf8.txt

@@ -7973,3 +7973,5 @@ ER_DATA_WAS_COMMITED_UNDER_ROLLBACK
         eng "Engine %s does not support rollback. Changes were committed during rollback call"
 ER_PK_INDEX_CANT_BE_IGNORED
         eng "A primary key cannot be marked as IGNORE"
+ER_BINLOG_UNSAFE_SKIP_LOCKED
+	eng "SKIP LOCKED makes this statement unsafe"

sql/sql_base.cc

@@ -4421,14 +4421,18 @@ restart:
     /* Set appropriate TABLE::lock_type. */
     if (tbl && tables->lock_type != TL_UNLOCK && !thd->locked_tables_mode)
     {
-      if (tables->lock_type == TL_WRITE_DEFAULT)
+      if (tables->lock_type == TL_WRITE_DEFAULT ||
-        tbl->reginfo.lock_type= thd->update_lock_default;
+          unlikely(tables->lock_type == TL_WRITE_SKIP_LOCKED &&
-      else if (tables->lock_type == TL_READ_DEFAULT)
+           !(tables->table->file->ha_table_flags() & HA_CAN_SKIP_LOCKED)))
-          tbl->reginfo.lock_type=
+          tbl->reginfo.lock_type= thd->update_lock_default;
-            read_lock_type_for_table(thd, thd->lex, tables,
+      else if (likely(tables->lock_type == TL_READ_DEFAULT) ||
-                                     some_routine_modifies_data);
+               (tables->lock_type == TL_READ_SKIP_LOCKED &&
+                !(tables->table->file->ha_table_flags() & HA_CAN_SKIP_LOCKED)))
+          tbl->reginfo.lock_type= read_lock_type_for_table(thd, thd->lex, tables,
+                                                           some_routine_modifies_data);
       else
         tbl->reginfo.lock_type= tables->lock_type;
+      tbl->reginfo.skip_locked= tables->skip_locked;
     }
 #ifdef WITH_WSREP
     /*

sql/sql_lex.cc

@@ -612,7 +612,8 @@ Query_tables_list::binlog_stmt_unsafe_errcode[BINLOG_STMT_UNSAFE_COUNT] =
   ER_BINLOG_UNSAFE_CREATE_SELECT_AUTOINC,
   ER_BINLOG_UNSAFE_UPDATE_IGNORE,
   ER_BINLOG_UNSAFE_INSERT_TWO_KEYS,
-  ER_BINLOG_UNSAFE_AUTOINC_NOT_FIRST
+  ER_BINLOG_UNSAFE_AUTOINC_NOT_FIRST,
+  ER_BINLOG_UNSAFE_SKIP_LOCKED
 };
 
 
@@ -3009,6 +3010,8 @@ void st_select_lex::init_select()
   select_limit= 0;      /* denotes the default limit = HA_POS_ERROR */
   offset_limit= 0;      /* denotes the default offset = 0 */
   is_set_query_expr_tail= false;
+  select_lock= select_lock_type::NONE;
+  skip_locked= false;
   with_sum_func= 0;
   with_all_modifier= 0;
   is_correlated= 0;
@@ -9690,19 +9693,24 @@ void Lex_select_lock::set_to(SELECT_LEX *sel)
       sel->master_unit()->set_lock_to_the_last_select(*this);
     else
     {
+      thr_lock_type lock_type;
       sel->parent_lex->safe_to_cache_query= 0;
-      if (update_lock)
+      if (unlikely(skip_locked))
       {
-        sel->lock_type= TL_WRITE;
+        lock_type= update_lock ? TL_WRITE_SKIP_LOCKED : TL_READ_SKIP_LOCKED;
-        sel->set_lock_for_tables(TL_WRITE, false);
       }
       else
       {
-        sel->lock_type= TL_READ_WITH_SHARED_LOCKS;
+        lock_type= update_lock ? TL_WRITE : TL_READ_WITH_SHARED_LOCKS;
-        sel->set_lock_for_tables(TL_READ_WITH_SHARED_LOCKS, false);
       }
+      sel->lock_type= lock_type;
+      sel->select_lock= (update_lock ? st_select_lex::select_lock_type::FOR_UPDATE :
+                         st_select_lex::select_lock_type::IN_SHARE_MODE);
+      sel->set_lock_for_tables(lock_type, false, skip_locked);
     }
   }
+  else
+    sel->select_lock= st_select_lex::select_lock_type::NONE;
 }
 
 bool Lex_order_limit_lock::set_to(SELECT_LEX *sel)

sql/sql_lex.h

@@ -1298,6 +1298,11 @@ public:
   /* index in the select list of the expression currently being fixed */
   int cur_pos_in_select_list;
 
+  /* SELECT [FOR UPDATE/LOCK IN SHARE MODE] [SKIP LOCKED] */
+  enum select_lock_type {NONE, IN_SHARE_MODE, FOR_UPDATE};
+  enum select_lock_type select_lock;
+  bool skip_locked;
+
   List<udf_func>     udf_list;                  /* udf function calls stack */
 
   /* 
@@ -1410,7 +1415,8 @@ public:
   TABLE_LIST *convert_right_join();
   List<Item>* get_item_list();
   ulong get_table_join_options();
-  void set_lock_for_tables(thr_lock_type lock_type, bool for_update);
+  void set_lock_for_tables(thr_lock_type lock_type, bool for_update,
+                           bool skip_locks);
   /*
     This method created for reiniting LEX in mysql_admin_table() and can be
     used only if you are going remove all SELECT_LEX & units except belonger
@@ -1942,6 +1948,13 @@ public:
     */
     BINLOG_STMT_UNSAFE_AUTOINC_NOT_FIRST,
 
+    /**
+       INSERT .. SELECT ... SKIP LOCKED is unlikely to have the same
+       rows locked on the replica.
+       primary key.
+    */
+    BINLOG_STMT_UNSAFE_SKIP_LOCKED,
+
     /* The last element of this enumeration type. */
     BINLOG_STMT_UNSAFE_COUNT
   };

sql/sql_parse.cc

@@ -8875,7 +8875,8 @@ bool st_select_lex::add_window_spec(THD *thd,
 /**
   Set lock for all tables in current select level.
 
-  @param lock_type			Lock to set for tables
+  @param lock_type		Lock to set for tables
+  @param skip_locked		(SELECT {FOR UPDATE/LOCK IN SHARED MODE} SKIP LOCKED)
 
   @note
     If lock is a write lock, then tables->updating is set 1
@@ -8883,16 +8884,19 @@ bool st_select_lex::add_window_spec(THD *thd,
     query
 */
 
-void st_select_lex::set_lock_for_tables(thr_lock_type lock_type, bool for_update)
+void st_select_lex::set_lock_for_tables(thr_lock_type lock_type, bool for_update,
+					bool skip_locked_arg)
 {
   DBUG_ENTER("set_lock_for_tables");
-  DBUG_PRINT("enter", ("lock_type: %d  for_update: %d", lock_type,
+  DBUG_PRINT("enter", ("lock_type: %d  for_update: %d  skip_locked %d",
-		       for_update));
+                       lock_type, for_update, skip_locked));
+  skip_locked= skip_locked_arg;
   for (TABLE_LIST *tables= table_list.first;
        tables;
        tables= tables->next_local)
   {
     tables->lock_type= lock_type;
+    tables->skip_locked= skip_locked;
     tables->updating=  for_update;
     tables->mdl_request.set_type((lock_type >= TL_FIRST_WRITE) ?
                                  MDL_SHARED_WRITE : MDL_SHARED_READ);

sql/sql_select.cc

@@ -27939,11 +27939,14 @@ void st_select_lex::print(THD *thd, String *str, enum_query_type query_type)
   print_limit(thd, str, query_type);
 
   // lock type
-  if (lock_type == TL_READ_WITH_SHARED_LOCKS)
+  if (select_lock == select_lock_type::IN_SHARE_MODE)
     str->append(" lock in share mode");
-  else if (lock_type == TL_WRITE)
+  else if (select_lock == select_lock_type::FOR_UPDATE)
     str->append(" for update");
 
+  if (unlikely(skip_locked))
+    str->append(" skip locked");
+
   // PROCEDURE unsupported here
 }
 

sql/sql_test.cc

@@ -49,11 +49,13 @@ static const char *lock_descriptions[] =
   /* TL_READ_WITH_SHARED_LOCKS  */  "Shared read lock",
   /* TL_READ_HIGH_PRIORITY      */  "High priority read lock",
   /* TL_READ_NO_INSERT          */  "Read lock without concurrent inserts",
+  /* TL_READ_SKIP_LOCKED        */  "Read lock without blocking if row is locked",
   /* TL_WRITE_ALLOW_WRITE       */  "Write lock that allows other writers",
   /* TL_WRITE_CONCURRENT_INSERT */  "Concurrent insert lock",
   /* TL_WRITE_DELAYED           */  "Lock used by delayed insert",
   /* TL_WRITE_DEFAULT           */  NULL,
   /* TL_WRITE_LOW_PRIORITY      */  "Low priority write lock",
+  /* TL_WRITE_SKIP_LOCKED       */  "Write lock but skip existing locked rows",
   /* TL_WRITE                   */  "High priority write lock",
   /* TL_WRITE_ONLY              */  "Highest priority write lock"
 };

sql/sql_view.cc

@@ -444,7 +444,7 @@ bool mysql_create_view(THD *thd, TABLE_LIST *views,
   */
   if (lex->current_select->lock_type != TL_READ_DEFAULT)
   {
-    lex->current_select->set_lock_for_tables(TL_READ_DEFAULT, false);
+    lex->current_select->set_lock_for_tables(TL_READ_DEFAULT, false, select_lex->skip_locked);
     view->mdl_request.set_type(MDL_EXCLUSIVE);
   }
 

sql/sql_yacc.yy

@@ -902,6 +902,7 @@ End SQL_MODE_ORACLE_SPECIFIC */
 %token  <kwd>  LEVEL_SYM
 %token  <kwd>  LIST_SYM
 %token  <kwd>  LOCAL_SYM                     /* SQL-2003-R */
+%token  <kwd>  LOCKED_SYM
 %token  <kwd>  LOCKS_SYM
 %token  <kwd>  LOGFILE_SYM
 %token  <kwd>  LOGS_SYM
@@ -1062,6 +1063,7 @@ End SQL_MODE_ORACLE_SPECIFIC */
 %token  <kwd>  SHUTDOWN
 %token  <kwd>  SIGNED_SYM
 %token  <kwd>  SIMPLE_SYM                    /* SQL-2003-N */
+%token  <kwd>  SKIP_SYM
 %token  <kwd>  SLAVE
 %token  <kwd>  SLAVES
 %token  <kwd>  SLAVE_POS_SYM
@@ -1378,7 +1380,7 @@ End SQL_MODE_ORACLE_SPECIFIC */
         udf_type opt_local opt_no_write_to_binlog
         opt_temporary all_or_any opt_distinct opt_glimit_clause
         opt_ignore_leaves fulltext_options union_option
-        opt_not
+        opt_not opt_skip_locked
         transaction_access_mode_types
         opt_natural_language_mode opt_query_expansion
         opt_ev_status opt_ev_on_completion ev_on_completion opt_ev_comment
@@ -9130,21 +9132,39 @@ opt_select_lock_type:
         }
         ;
 
+opt_skip_locked:
+        /* empty */
+        {
+          $$= 0;
+        }
+        | SKIP_SYM LOCKED_SYM
+        {
+          Lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_SKIP_LOCKED);
+          $$= 1;
+        }
 
 opt_lock_wait_timeout_new:
         /* empty */
         {
           $$.empty();
         }
-        | WAIT_SYM ulong_num
+        | WAIT_SYM ulong_num opt_skip_locked
         {
           $$.defined_timeout= TRUE;
           $$.timeout= $2;
+          $$.skip_locked= $3;
         }
-        | NOWAIT_SYM
+        | NOWAIT_SYM opt_skip_locked
         {
           $$.defined_timeout= TRUE;
           $$.timeout= 0;
+          $$.skip_locked= $2;
+        }
+        | SKIP_SYM LOCKED_SYM
+        {
+          $$.empty();
+          $$.skip_locked= 1;
+          Lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_SKIP_LOCKED);
         }
       ;
 
@@ -12871,7 +12891,7 @@ insert:
           }
           insert_start insert_lock_option opt_ignore opt_into insert_table
           {
-            Select->set_lock_for_tables($4, true);
+            Select->set_lock_for_tables($4, true, false);
           }
           insert_field_spec opt_insert_update opt_returning
           stmt_end
@@ -12888,7 +12908,7 @@ replace:
           }
           insert_start replace_lock_option opt_into insert_table
           {
-            Select->set_lock_for_tables($4, true);
+            Select->set_lock_for_tables($4, true, false);
           }
           insert_field_spec opt_returning
           stmt_end
@@ -13184,7 +13204,7 @@ update:
               be too pessimistic. We will decrease lock level if possible in
               mysql_multi_update().
             */
-            slex->set_lock_for_tables($3, slex->table_list.elements == 1);
+            slex->set_lock_for_tables($3, slex->table_list.elements == 1, false);
           }
           opt_where_clause opt_order_clause delete_limit_clause
           {
@@ -15695,6 +15715,7 @@ keyword_sp_var_and_label:
         | LESS_SYM
         | LEVEL_SYM
         | LIST_SYM
+        | LOCKED_SYM
         | LOCKS_SYM
         | LOGFILE_SYM
         | LOGS_SYM
@@ -15823,6 +15844,7 @@ keyword_sp_var_and_label:
         | SETVAL_SYM
         | SIMPLE_SYM
         | SHARE_SYM
+        | SKIP_SYM
         | SLAVE_POS_SYM
         | SLOW
         | SNAPSHOT_SYM

sql/structs.h

@@ -177,6 +177,7 @@ struct st_join_table;
 typedef struct st_reginfo {		/* Extra info about reg */
   struct st_join_table *join_tab;	/* Used by SELECT() */
   enum thr_lock_type lock_type;		/* How database is used */
+  bool skip_locked;
   bool not_exists_optimize;
   /*
     TRUE <=> range optimizer found that there is no rows satisfying
@@ -807,13 +808,14 @@ public:
     uint defined_lock:1;
     uint update_lock:1;
     uint defined_timeout:1;
+    uint skip_locked:1;
   };
   ulong timeout;
 
 
   void empty()
   {
-    defined_lock= update_lock= defined_timeout= FALSE;
+    defined_lock= update_lock= defined_timeout= skip_locked= FALSE;
     timeout= 0;
   }
   void set_to(st_select_lex *sel);

sql/table.cc

@@ -4044,6 +4044,7 @@ enum open_frm_error open_table_from_share(THD *thd, TABLE_SHARE *share,
   }
 
   outparam->reginfo.lock_type= TL_UNLOCK;
+  outparam->reginfo.skip_locked= false;
   outparam->current_lock= F_UNLCK;
   records=0;
   if ((db_stat & HA_OPEN_KEYFILE) || (prgflag & DELAYED_OPEN))
@@ -5492,6 +5493,7 @@ void TABLE::init(THD *thd, TABLE_LIST *tl)
   reginfo.impossible_range= 0;
   reginfo.join_tab= NULL;
   reginfo.not_exists_optimize= FALSE;
+  reginfo.skip_locked= false;
   created= TRUE;
   cond_selectivity= 1.0;
   cond_selectivity_sampling_explain= NULL;

sql/table.h

@@ -2456,7 +2456,8 @@ struct TABLE_LIST
   bool          updating;               /* for replicate-do/ignore table */
   bool		force_index;		/* prefer index over table scan */
   bool          ignore_leaves;          /* preload only non-leaf nodes */
-  bool          crashed;                 /* Table was found crashed */
+  bool          crashed;                /* Table was found crashed */
+  bool          skip_locked;            /* Skip locked in view defination */
   table_map     dep_tables;             /* tables the table depends on      */
   table_map     on_expr_dep_tables;     /* tables on expression depends on  */
   struct st_nested_join *nested_join;   /* if the element is a nested join  */

storage/innobase/handler/ha_innodb.cc

@@ -2619,6 +2619,7 @@ ha_innobase::ha_innobase(
                           | HA_CAN_TABLES_WITHOUT_ROLLBACK
                           | HA_CAN_ONLINE_BACKUPS
 			  | HA_CONCURRENT_OPTIMIZE
+			  | HA_CAN_SKIP_LOCKED
 			  |  (srv_force_primary_key ? HA_REQUIRE_PRIMARY_KEY : 0)
 		  ),
 	m_start_of_scan(),
@@ -15226,6 +15227,7 @@ ha_innobase::reset()
 	/* This is a statement level counter. */
 	m_prebuilt->autoinc_last_value = 0;
 
+	m_prebuilt->skip_locked = false;
 	return(0);
 }
 
@@ -15901,6 +15903,7 @@ ha_innobase::store_lock(
 	} else if ((lock_type == TL_READ && in_lock_tables)
 		   || (lock_type == TL_READ_HIGH_PRIORITY && in_lock_tables)
 		   || lock_type == TL_READ_WITH_SHARED_LOCKS
+		   || lock_type == TL_READ_SKIP_LOCKED
 		   || lock_type == TL_READ_NO_INSERT
 		   || (lock_type != TL_IGNORE
 		       && sql_command != SQLCOM_SELECT)) {
@@ -15910,11 +15913,12 @@ ha_innobase::store_lock(
 		are processing a stored procedure or function, or
 		2) (we do not know when TL_READ_HIGH_PRIORITY is used), or
 		3) this is a SELECT ... IN SHARE MODE, or
-		4) we are doing a complex SQL statement like
+		4) this is a SELECT ... IN SHARE MODE SKIP LOCKED, or
+		5) we are doing a complex SQL statement like
 		INSERT INTO ... SELECT ... and the logical logging (MySQL
 		binlog) requires the use of a locking read, or
 		MySQL is doing LOCK TABLES ... READ.
-		5) we let InnoDB do locking reads for all SQL statements that
+		6) we let InnoDB do locking reads for all SQL statements that
 		are not simple SELECTs; note that select_lock_type in this
 		case may get strengthened in ::external_lock() to LOCK_X.
 		Note that we MUST use a locking read in all data modifying
@@ -15960,6 +15964,8 @@ ha_innobase::store_lock(
 		m_prebuilt->select_lock_type = LOCK_NONE;
 		m_prebuilt->stored_select_lock_type = LOCK_NONE;
 	}
+	m_prebuilt->skip_locked= (lock_type == TL_WRITE_SKIP_LOCKED ||
+				  lock_type == TL_READ_SKIP_LOCKED);
 
 	if (!trx_is_started(trx)
 	    && (m_prebuilt->select_lock_type != LOCK_NONE

storage/innobase/include/row0mysql.h

@@ -687,6 +687,7 @@ struct row_prebuilt_t {
 	dtuple_t*	clust_ref;	/*!< prebuilt dtuple used in
 					sel/upd/del */
 	lock_mode	select_lock_type;/*!< LOCK_NONE, LOCK_S, or LOCK_X */
+	bool		skip_locked;	/*!< TL_{READ,WRITE}_SKIP_LOCKED */
 	lock_mode	stored_select_lock_type;/*!< this field is used to
 					remember the original select_lock_type
 					that was decided in ha_innodb.cc,

storage/innobase/row/row0sel.cc

@@ -5129,17 +5129,18 @@ no_gap_lock:
 					!= ROW_READ_TRY_SEMI_CONSISTENT)
 			    || unique_search
 			    || index != clust_index) {
-
+				if (!prebuilt->skip_locked) {
-				goto lock_wait_or_error;
+					goto lock_wait_or_error;
+				}
+			} else {
+				/* The following call returns 'offsets'
+				associated with 'old_vers' */
+				row_sel_build_committed_vers_for_mysql(
+					clust_index, prebuilt, rec,
+					&offsets, &heap, &old_vers,
+					need_vrow ? &vrow : NULL, &mtr);
 			}
 
-			/* The following call returns 'offsets'
-			associated with 'old_vers' */
-			row_sel_build_committed_vers_for_mysql(
-				clust_index, prebuilt, rec,
-				&offsets, &heap, &old_vers, need_vrow ? &vrow : NULL,
-			        &mtr);
-
 			/* Check whether it was a deadlock or not, if not
 			a deadlock and the transaction had to wait then
 			release the lock it is waiting on. */
@@ -5161,7 +5162,16 @@ no_gap_lock:
 			case DB_LOCK_WAIT:
 				ut_ad(!dict_index_is_spatial(index));
 				err = DB_SUCCESS;
+				if (prebuilt->skip_locked) {
+					goto next_rec;
+				}
 				break;
+		        case DB_LOCK_WAIT_TIMEOUT:
+				if (prebuilt->skip_locked) {
+					err = DB_SUCCESS;
+					goto next_rec;
+				}
+				/* fall through */
 			default:
 				ut_error;
 			}
@@ -5181,7 +5191,13 @@ no_gap_lock:
 			} else {
 				goto lock_wait_or_error;
 			}
-
+			break;
+		case DB_LOCK_WAIT_TIMEOUT:
+			if (prebuilt->skip_locked) {
+				err = DB_SUCCESS;
+				goto next_rec;
+			}
+			/* fall through */
 		default:
 
 			goto lock_wait_or_error;
@@ -5356,6 +5372,10 @@ requires_clust_rec:
 						      &offsets, &heap,
 						      need_vrow ? &vrow : NULL,
 						      &mtr);
+		if (prebuilt->skip_locked &&
+		    err == DB_LOCK_WAIT) {
+			err = lock_trx_handle_wait(trx);
+		}
 		switch (err) {
 		case DB_SUCCESS:
 			if (clust_rec == NULL) {
@@ -5375,6 +5395,13 @@ requires_clust_rec:
 			}
 			err = DB_SUCCESS;
 			break;
+		case DB_LOCK_WAIT_TIMEOUT:
+		case DB_LOCK_WAIT:
+			if (prebuilt->skip_locked) {
+				err = DB_SUCCESS;
+				goto next_rec;
+			}
+			/* fall through */
 		default:
 			vrow = NULL;
 			goto lock_wait_or_error;

storage/mroonga/ha_mroonga.cpp

@@ -361,6 +361,9 @@ static const char *mrn_inspect_thr_lock_type(enum thr_lock_type lock_type)
   case TL_READ_NO_INSERT:
     inspected = "TL_READ_NO_INSERT";
     break;
+  case TL_READ_SKIP_LOCKED:
+    inspected = "TL_READ_SKIP_LOCKED";
+    break;
   case TL_WRITE_ALLOW_WRITE:
     inspected = "TL_WRITE_ALLOW_WRITE";
     break;
@@ -386,6 +389,9 @@ static const char *mrn_inspect_thr_lock_type(enum thr_lock_type lock_type)
   case TL_WRITE:
     inspected = "TL_WRITE";
     break;
+  case TL_WRITE_SKIP_LOCKED:
+    inspected = "TL_WRITE_SKIP_LOCKED";
+    break;
   case TL_WRITE_ONLY:
     inspected = "TL_WRITE_ONLY";
     break;