diff --git a/include/thr_lock.h b/include/thr_lock.h
index 38dc600951f..d918d6d74f4 100644
--- a/include/thr_lock.h
+++ b/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 */
diff --git a/mysql-test/main/flush_read_lock.result b/mysql-test/main/flush_read_lock.result
index ae57788c7d4..169b68b90c4 100644
--- a/mysql-test/main/flush_read_lock.result
+++ b/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.
diff --git a/mysql-test/main/flush_read_lock.test b/mysql-test/main/flush_read_lock.test
index 40f8b2aec3b..c92d583e3a9 100644
--- a/mysql-test/main/flush_read_lock.test
+++ b/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
diff --git a/mysql-test/main/view.result b/mysql-test/main/view.result
index 584be058318..5d171268624 100644
--- a/mysql-test/main/view.result
+++ b/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
+#
diff --git a/mysql-test/main/view.test b/mysql-test/main/view.test
index e2c9a589294..b504600de91 100644
--- a/mysql-test/main/view.test
+++ b/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 #
diff --git a/mysql-test/suite/innodb/r/create_table_insert_skip_locked.result b/mysql-test/suite/innodb/r/create_table_insert_skip_locked.result
new file mode 100644
index 00000000000..65f61093e34
--- /dev/null
+++ b/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;
diff --git a/mysql-test/suite/innodb/t/create_table_insert_skip_locked.test b/mysql-test/suite/innodb/t/create_table_insert_skip_locked.test
new file mode 100644
index 00000000000..e94c4eacd9b
--- /dev/null
+++ b/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;
diff --git a/mysql-test/suite/perfschema/r/start_server_low_digest_sql_length.result b/mysql-test/suite/perfschema/r/start_server_low_digest_sql_length.result
index c991efce659..ef9a9da6976 100644
--- a/mysql-test/suite/perfschema/r/start_server_low_digest_sql_length.result
+++ b/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/truncate	0f84807fb4a75d0f391f8a93e7c3c182	TRUNCATE TABLE 	truncat...
+statement/sql/select	c6c8d8d7523f187bbdfda0d54e15b47a	SELECT ? + ? + 	SELECT ...
+statement/sql/truncate	13d7058e7e84d7b7133f0a4f74e0b199	TRUNCATE TABLE 	truncat...
diff --git a/mysql-test/suite/rpl/r/rpl_unsafe_statements.result b/mysql-test/suite/rpl/r/rpl_unsafe_statements.result
index 0ce94ca63d0..27065fffaa6 100644
--- a/mysql-test/suite/rpl/r/rpl_unsafe_statements.result
+++ b/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
diff --git a/mysql-test/suite/rpl/t/rpl_unsafe_statements.test b/mysql-test/suite/rpl/t/rpl_unsafe_statements.test
index aa0bd076398..9185e566b9c 100644
--- a/mysql-test/suite/rpl/t/rpl_unsafe_statements.test
+++ b/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
diff --git a/sql/handler.h b/sql/handler.h
index a2dd747a4d7..fc22d8dd8a9 100644
--- a/sql/handler.h
+++ b/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 */
diff --git a/sql/lex.h b/sql/lex.h
index 5a9ec2ec1b3..b40883c4ea3 100644
--- a/sql/lex.h
+++ b/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)},
diff --git a/sql/share/errmsg-utf8.txt b/sql/share/errmsg-utf8.txt
index badaaa17716..555836406fb 100644
--- a/sql/share/errmsg-utf8.txt
+++ b/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"
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index e310f0687f3..35f45cfed21 100644
--- a/sql/sql_base.cc
+++ b/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)
-        tbl->reginfo.lock_type= thd->update_lock_default;
-      else if (tables->lock_type == TL_READ_DEFAULT)
-          tbl->reginfo.lock_type=
-            read_lock_type_for_table(thd, thd->lex, tables,
-                                     some_routine_modifies_data);
+      if (tables->lock_type == TL_WRITE_DEFAULT ||
+          unlikely(tables->lock_type == TL_WRITE_SKIP_LOCKED &&
+           !(tables->table->file->ha_table_flags() & HA_CAN_SKIP_LOCKED)))
+          tbl->reginfo.lock_type= thd->update_lock_default;
+      else if (likely(tables->lock_type == TL_READ_DEFAULT) ||
+               (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
     /*
diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc
index 751ebc1d9fd..a06aa4df988 100644
--- a/sql/sql_lex.cc
+++ b/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;
-        sel->set_lock_for_tables(TL_WRITE, false);
+        lock_type= update_lock ? TL_WRITE_SKIP_LOCKED : TL_READ_SKIP_LOCKED;
       }
       else
       {
-        sel->lock_type= TL_READ_WITH_SHARED_LOCKS;
-        sel->set_lock_for_tables(TL_READ_WITH_SHARED_LOCKS, false);
+        lock_type= update_lock ? TL_WRITE : TL_READ_WITH_SHARED_LOCKS;
       }
+      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)
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index c475c6d3385..02f5b357c50 100644
--- a/sql/sql_lex.h
+++ b/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
   };
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 1ce4b589d7b..cca08c9007f 100644
--- a/sql/sql_parse.cc
+++ b/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,
-		       for_update));
+  DBUG_PRINT("enter", ("lock_type: %d  for_update: %d  skip_locked %d",
+                       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);
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 4509b0a05ef..56a7ac6eb80 100644
--- a/sql/sql_select.cc
+++ b/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
 }
 
diff --git a/sql/sql_test.cc b/sql/sql_test.cc
index 36e4fedf835..c10d869f9b2 100644
--- a/sql/sql_test.cc
+++ b/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"
 };
diff --git a/sql/sql_view.cc b/sql/sql_view.cc
index cfd43bd13ab..3bacc3d1499 100644
--- a/sql/sql_view.cc
+++ b/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);
   }
 
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index c06263adbd2..be7df51ff42 100644
--- a/sql/sql_yacc.yy
+++ b/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
diff --git a/sql/structs.h b/sql/structs.h
index e862890be9c..df362f76f82 100644
--- a/sql/structs.h
+++ b/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);
diff --git a/sql/table.cc b/sql/table.cc
index 71374d6666d..dfaf1881ad0 100644
--- a/sql/table.cc
+++ b/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;
diff --git a/sql/table.h b/sql/table.h
index 5a8ee2c1709..f0499ee63fc 100644
--- a/sql/table.h
+++ b/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  */
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index 20b59b94aa6..4440fe62fa5 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/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
diff --git a/storage/innobase/include/row0mysql.h b/storage/innobase/include/row0mysql.h
index 4d2aadfd5a0..51c588f4a2d 100644
--- a/storage/innobase/include/row0mysql.h
+++ b/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,
diff --git a/storage/innobase/row/row0sel.cc b/storage/innobase/row/row0sel.cc
index e8fc86ee12d..e5e033857ec 100644
--- a/storage/innobase/row/row0sel.cc
+++ b/storage/innobase/row/row0sel.cc
@@ -5129,17 +5129,18 @@ no_gap_lock:
 					!= ROW_READ_TRY_SEMI_CONSISTENT)
 			    || unique_search
 			    || index != clust_index) {
-
-				goto lock_wait_or_error;
+				if (!prebuilt->skip_locked) {
+					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;
diff --git a/storage/mroonga/ha_mroonga.cpp b/storage/mroonga/ha_mroonga.cpp
index 5b481ca9982..0282897f4f3 100644
--- a/storage/mroonga/ha_mroonga.cpp
+++ b/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;