Committing on behalf or Dmitry Lenev:

Fix for bug #46947 "Embedded SELECT without FOR UPDATE is
causing a lock", with after-review fixes.

SELECT statements with subqueries referencing InnoDB tables
were acquiring shared locks on rows in these tables when they
were executed in REPEATABLE-READ mode and with statement or
mixed mode binary logging turned on.

This was a regression which were introduced when fixing
bug 39843.

The problem was that for tables belonging to subqueries
parser set TL_READ_DEFAULT as a lock type. In cases when
statement/mixed binary logging at open_tables() time this
type of lock was converted to TL_READ_NO_INSERT lock at
open_tables() time and caused InnoDB engine to acquire
shared locks on reads from these tables. Although in some
cases such behavior was correct (e.g. for subqueries in
DELETE) in case of SELECT it has caused unnecessary locking.

This patch tries to solve this problem by rethinking our
approach to how we handle locking for SELECT and subqueries.
Now we always set TL_READ_DEFAULT lock type for all cases
when we read data. When at open_tables() time this lock
is interpreted as TL_READ_NO_INSERT or TL_READ depending
on whether this statement as a whole or call to function
which uses particular table should be written to the
binary log or not (if yes then statement should be properly
serialized with concurrent statements and stronger lock
should be acquired).

Test coverage is added for both InnoDB and MyISAM.

This patch introduces an "incompatible" change in locking
scheme for subqueries used in SELECT ... FOR UPDATE and
SELECT .. IN SHARE MODE.
In 4.1 the server would use a snapshot InnoDB read for 
subqueries in SELECT FOR UPDATE and SELECT .. IN SHARE MODE
statements, regardless of whether the binary log is on or off.
If the user required a different type of read (i.e. locking read),
he/she could request so explicitly by providing FOR UPDATE/IN SHARE MODE
clause for each individual subquery.
On of the patches for 5.0 broke this behaviour (which was not documented
or tested), and started to use locking reads fora all subqueries in SELECT ... 
FOR UPDATE/IN SHARE MODE. This patch restored 4.1 behaviour.

mysql-test/include/check_concurrent_insert.inc:
  Added auxiliary script which allows to check if statement
  reading table allows concurrent inserts in it.
mysql-test/include/check_no_concurrent_insert.inc:
  Added auxiliary script which allows to check that statement
  reading table doesn't allow concurrent inserts in it.
mysql-test/include/check_no_row_lock.inc:
  Added auxiliary script which allows to check if statement
  reading table doesn't take locks on its rows.
mysql-test/include/check_shared_row_lock.inc:
  Added auxiliary script which allows to check if statement
  reading table takes shared locks on some of its rows.
mysql-test/r/bug39022.result:
  After bug #46947 'Embedded SELECT without FOR UPDATE is
  causing a lock' was fixed test case for bug 39022 has to
  be adjusted in order to trigger execution path on which
  original problem was encountered.
mysql-test/r/innodb_mysql_lock2.result:
  Added coverage for handling of locking in various cases when
  we read data from InnoDB tables (includes test case for
  bug #46947 'Embedded SELECT without FOR UPDATE is causing a
  lock').
mysql-test/r/lock_sync.result:
  Added coverage for handling of locking in various cases when
  we read data from MyISAM tables.
mysql-test/t/bug39022.test:
  After bug #46947 'Embedded SELECT without FOR UPDATE is
  causing a lock' was fixed test case for bug 39022 has to
  be adjusted in order to trigger execution path on which
  original problem was encountered.
mysql-test/t/innodb_mysql_lock2.test:
  Added coverage for handling of locking in various cases when
  we read data from InnoDB tables (includes test case for
  bug #46947 'Embedded SELECT without FOR UPDATE is causing a
  lock').
mysql-test/t/lock_sync.test:
  Added coverage for handling of locking in various cases when
  we read data from MyISAM tables.
sql/log_event.cc:
  Since LEX::lock_option member was removed we no longer can
  rely on its value in Load_log_event::print_query() to
  determine that log event correponds to LOAD DATA CONCURRENT
  statement (this was not correct in all situations anyway).
  A new Load_log_event's member was introduced as a replacement.
  It is initialized at event object construction time and
  explicitly indicates whether LOAD DATA was concurrent.
sql/log_event.h:
  Since LEX::lock_option member was removed we no longer can
  rely on its value in Load_log_event::print_query() to
  determine that log event correponds to LOAD DATA CONCURRENT
  statement (this was not correct in all situations anyway).
  A new Load_log_event's member was introduced as a replacement.
  It is initialized at event object construction time and
  explicitly indicates whether LOAD DATA was concurrent.
sql/sp_head.cc:
  sp_head::reset_lex():
    Before parsing substatement reset part of parser state
    which needs this (e.g. set Yacc_state::m_lock_type to
    default value).
sql/sql_acl.cc:
  Since LEX::reset_n_backup_query_tables_list() now also
  resets LEX::sql_command member (as it became part of
  Query_tables_list class) we have to restore it in cases
  when while working with proxy Query_table_list we assume
  that LEX::sql_command still corresponds to original SQL
  command being executed (for example, when we are logging
  statement to the binary log while having Query_tables_list
  reset and backed up).
sql/sql_base.cc:
  Changed read_lock_type_for_table() to return a weak TL_READ
  type of lock in cases when we are executing statement which
  won't update tables directly and table doesn't belong to
  statement's prelocking list and thus can't be used by a
  stored function. It is OK to do so since in this case table
  won't be used by statement or function call which will be
  written to the binary log, so serializability requirements
  for it can be relaxed.
  One of results from this change is that SELECTs on InnoDB
  tables no longer takes shared row locks for tables which
  are used in subqueries (i.e. bug #46947 is fixed).
  Another result is that for similar SELECTs on MyISAM tables
  concurrent inserts are allowed.
  In order to implement this change signature of
  read_lock_type_for_table() function was changed to take
  pointers to Query_tables_list and TABLE_LIST objects.
sql/sql_base.h:
  - Function read_lock_type_for_table() now takes pointers
    to Query_tables_list and TABLE_LIST elements as its
    arguments since to correctly determine lock type it needs
    to know what statement is being performed and whether table
    element for which lock type to be determined belongs to
    prelocking list.
sql/sql_lex.cc:
  - Removed LEX::lock_option and st_select_lex::lock_option
    members. Places in parser that were using them now use
    Yacc_state::m_lock_type instead.
  - To emphasize that LEX::sql_command member is used during
    process of opening and locking of tables it was moved to
    Query_tables_list class. It is now reset by
    Query_tables_list::reset_query_tables_list() method.
sql/sql_lex.h:
  - Removed st_select_lex::lock_option member as there is no
    real need for per-SELECT lock type (HIGH_PRIORITY option
    should apply to the whole statement. FOR UPDATE/LOCK IN
    SHARE MODE clauses can be handled without this member).
    The main effect which was achieved by introduction of this
    member, i.e. using TL_READ_DEFAULT lock type for
    subqueries, is now achieved by setting LEX::lock_option
    (or rather its replacement - Yacc_state::m_lock_type) to
    TL_READ_DEFAULT in almost all cases.
  - To emphasize that LEX::sql_command member is used during
    process of opening and locking of tables it was moved to
    Query_tables_list class.
  - Replaced LEX::lock_option with Yacc_state::m_lock_type
    in order to emphasize that this value is relevant only
    during parsing. Unlike for LEX::lock_option the default
    value for Yacc_state::m_lock_type is TL_READ_DEFAULT.
    Note that for cases when it is OK to take a "weak" read
    lock (e.g. simple SELECT) this lock type will be converted
    to TL_READ at open_tables() time. So this change won't
    cause negative change in behavior for such statements.
    OTOH this change ensures that, for example, for SELECTs
    which are used in stored functions TL_READ_NO_INSERT lock
    is taken when necessary and as result calls to such stored
    functions can be written to the binary log with correct
    serialization.
sql/sql_load.cc:
  Load_log_event constructor now requires a parameter that
  indicates whether LOAD DATA is concurrent.
sql/sql_parse.cc:
  LEX::lock_option was replaced with Yacc_state::m_lock_type.
  And instead of resetting the latter implicitly in
  mysql_init_multi_delete() we do it explicitly in the
  places in parser which call this function.
sql/sql_priv.h:
  - To be able more easily distinguish high-priority SELECTs
    in st_select_lex::print() method added flag for
    HIGH_PRIORITY option.
sql/sql_select.cc:
  Changed code not to rely on LEX::lock_option to determine
  that it is high-priority SELECT. It was replaced with
  Yacc_state::m_lock_type which is accessible only at
  parse time. So instead of LEX::lock_option we now rely
  on a newly introduced flag for st_select_lex::options -
  SELECT_HIGH_PRIORITY.
sql/sql_show.cc:
  Since LEX::reset_n_backup_query_tables_list() now also
  resets LEX::sql_command member (as it became part of
  Query_tables_list class) we have to restore it in cases
  when while working with proxy Query_table_list we assume
  that LEX::sql_command still corresponds to original SQL
  command being executed.
sql/sql_table.cc:
  Since LEX::reset_query_tables_list() now also resets
  LEX::sql_command member (as it became part of
  Query_tables_list class) we have to restore value of this
  member when this method is called by mysql_admin_table(),
  to make this code safe for re-execution.
sql/sql_trigger.cc:
  Since LEX::reset_n_backup_query_tables_list() now also
  resets LEX::sql_command member (as it became part of
  Query_tables_list class) we have to restore it in cases
  when while working with proxy Query_table_list we assume
  that LEX::sql_command still corresponds to original SQL
  command being executed (for example, when we are logging
  statement to the binary log while having Query_tables_list
  reset and backed up).
sql/sql_update.cc:
  Function read_lock_type_for_table() now takes pointers
  to Query_tables_list and TABLE_LIST elements as its
  arguments since to correctly determine lock type it needs
  to know what statement is being performed and whether table
  element for which lock type to be determined belongs to
  prelocking list.
sql/sql_yacc.yy:
  - Removed st_select_lex::lock_option member as there is no
    real need for per-SELECT lock type (HIGH_PRIORITY option
    should apply to the whole statement. FOR UPDATE/LOCK IN
    SHARE MODE clauses can be handled without this member).
    The main effect which was achieved by introduction of this
    member, i.e. using TL_READ_DEFAULT lock type for
    subqueries, is now achieved by setting LEX::lock_option
    (or rather its replacement - Yacc_state::m_lock_type) to
    TL_READ_DEFAULT in almost all cases.
  - Replaced LEX::lock_option with Yacc_state::m_lock_type
    in order to emphasize that this value is relevant only
    during parsing. Unlike for LEX::lock_option the default
    value for Yacc_state::m_lock_type is TL_READ_DEFAULT.
    Note that for cases when it is OK to take a "weak" read
    lock (e.g. simple SELECT) this lock type will be converted
    to TL_READ at open_tables() time. So this change won't
    cause negative change in behavior for such statements.
    OTOH this change ensures that, for example, for SELECTs
    which are used in stored functions TL_READ_NO_INSERT lock
    is taken when necessary and as result calls to such stored
    functions can be written to the binary log with correct
    serialization.
  - To be able more easily distinguish high-priority SELECTs
    in st_select_lex::print() method we now use new flag
    in st_select_lex::options bit-field.

17 files changed, 165 insertions, 77 deletions

diff --git a/sql/log_event.cc b/sql/log_event.cc
index 3a52b72909a..3a3d6c74e8e 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -4226,7 +4226,7 @@ void Load_log_event::print_query(bool need_db, const char *cs, char *buf,
 
   pos= strmov(pos, "LOAD DATA ");
 
-  if (thd->lex->lock_option == TL_WRITE_CONCURRENT_INSERT)
+  if (is_concurrent)
     pos= strmov(pos, "CONCURRENT ");
 
   if (fn_start)
@@ -4368,6 +4368,7 @@ bool Load_log_event::write_data_body(IO_CACHE* file)
 Load_log_event::Load_log_event(THD *thd_arg, sql_exchange *ex,
 			       const char *db_arg, const char *table_name_arg,
 			       List<Item> &fields_arg,
+                               bool is_concurrent_arg,
 			       enum enum_duplicates handle_dup,
 			       bool ignore, bool using_trans)
   :Log_event(thd_arg,
@@ -4378,7 +4379,8 @@ Load_log_event::Load_log_event(THD *thd_arg, sql_exchange *ex,
    num_fields(0),fields(0),
    field_lens(0),field_block_len(0),
    table_name(table_name_arg ? table_name_arg : ""),
-   db(db_arg), fname(ex->file_name), local_fname(FALSE)
+   db(db_arg), fname(ex->file_name), local_fname(FALSE),
+   is_concurrent(is_concurrent_arg)
 {
   time_t end_time;
   time(&end_time);
@@ -4459,7 +4461,13 @@ Load_log_event::Load_log_event(const char *buf, uint event_len,
                                const Format_description_log_event *description_event)
   :Log_event(buf, description_event), num_fields(0), fields(0),
    field_lens(0),field_block_len(0),
-   table_name(0), db(0), fname(0), local_fname(FALSE)
+   table_name(0), db(0), fname(0), local_fname(FALSE),
+   /*
+     Load_log_event which comes from the binary log does not contain
+     information about the type of insert which was used on the master.
+     Assume that it was an ordinary, non-concurrent LOAD DATA.
+    */
+   is_concurrent(FALSE)
 {
   DBUG_ENTER("Load_log_event");
   /*
@@ -6149,11 +6157,14 @@ int Stop_log_event::do_update_pos(Relay_log_info *rli)
 Create_file_log_event::
 Create_file_log_event(THD* thd_arg, sql_exchange* ex,
 		      const char* db_arg, const char* table_name_arg,
-		      List<Item>& fields_arg, enum enum_duplicates handle_dup,
+                      List<Item>& fields_arg,
+                      bool is_concurrent_arg,
+                      enum enum_duplicates handle_dup,
                       bool ignore,
 		      uchar* block_arg, uint block_len_arg, bool using_trans)
-  :Load_log_event(thd_arg,ex,db_arg,table_name_arg,fields_arg,handle_dup, ignore,
-		  using_trans),
+  :Load_log_event(thd_arg, ex, db_arg, table_name_arg, fields_arg,
+                  is_concurrent_arg,
+                  handle_dup, ignore, using_trans),
    fake_base(0), block(block_arg), event_buf(0), block_len(block_len_arg),
    file_id(thd_arg->file_id = mysql_bin_log.next_file_id())
 {
diff --git a/sql/log_event.h b/sql/log_event.h
index 36397c427e5..e281fd6e206 100644
--- a/sql/log_event.h
+++ b/sql/log_event.h
@@ -2069,6 +2069,17 @@ public:
   uint32 skip_lines;
   sql_ex_info sql_ex;
   bool local_fname;
+  /**
+    Indicates that this event corresponds to LOAD DATA CONCURRENT,
+
+    @note Since Load_log_event event coming from the binary log
+          lacks information whether LOAD DATA on master was concurrent
+          or not, this flag is only set to TRUE for an auxiliary
+          Load_log_event object which is used in mysql_load() to
+          re-construct LOAD DATA statement from function parameters,
+          for logging.
+  */
+  bool is_concurrent;
 
   /* fname doesn't point to memory inside Log_event::temp_buf  */
   void set_fname_outside_temp_buf(const char *afname, uint alen)
@@ -2089,7 +2100,9 @@ public:
 
   Load_log_event(THD* thd, sql_exchange* ex, const char* db_arg,
 		 const char* table_name_arg,
-		 List<Item>& fields_arg, enum enum_duplicates handle_dup, bool ignore,
+		 List<Item>& fields_arg,
+                 bool is_concurrent_arg,
+                 enum enum_duplicates handle_dup, bool ignore,
 		 bool using_trans);
   void set_fields(const char* db, List<Item> &fields_arg,
                   Name_resolution_context *context);
@@ -2708,6 +2721,7 @@ public:
   Create_file_log_event(THD* thd, sql_exchange* ex, const char* db_arg,
 			const char* table_name_arg,
 			List<Item>& fields_arg,
+                        bool is_concurrent_arg,
 			enum enum_duplicates handle_dup, bool ignore,
 			uchar* block_arg, uint block_len_arg,
 			bool using_trans);
diff --git a/sql/sp_head.cc b/sql/sp_head.cc
index c91ba2a68b4..2e66aec91e5 100644
--- a/sql/sp_head.cc
+++ b/sql/sp_head.cc
@@ -2127,6 +2127,9 @@ sp_head::reset_lex(THD *thd)
   sublex->interval_list.empty();
   sublex->type= 0;
 
+  /* Reset part of parser state which needs this. */
+  thd->m_parser_state->m_yacc.reset_before_substatement();
+
   DBUG_RETURN(FALSE);
 }
 
diff --git a/sql/sql_acl.cc b/sql/sql_acl.cc
index f8be3ff6d4a..55d83f49245 100644
--- a/sql/sql_acl.cc
+++ b/sql/sql_acl.cc
@@ -3159,6 +3159,12 @@ int mysql_table_grant(THD *thd, TABLE_LIST *table_list,
   */
   Query_tables_list backup;
   thd->lex->reset_n_backup_query_tables_list(&backup);
+  /*
+    Restore Query_tables_list::sql_command value, which was reset
+    above, as the code writing query to the binary log assumes that
+    this value corresponds to the statement being executed.
+  */
+  thd->lex->sql_command= backup.sql_command;
   if (open_and_lock_tables(thd, tables, FALSE, MYSQL_LOCK_IGNORE_TIMEOUT))
   {						// Should never happen
     close_thread_tables(thd);			/* purecov: deadcode */
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index baf13431d77..1d7ad87eae0 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -3963,7 +3963,8 @@ recover_from_failed_open(THD *thd, MDL_request *mdl_request,
   Return a appropriate read lock type given a table object.
 
   @param thd Thread context
-  @param table TABLE object for table to be locked
+  @param prelocking_ctx Prelocking context.
+  @param table_list     Table list element for table to be locked.
 
   @remark Due to a statement-based replication limitation, statements such as
           INSERT INTO .. SELECT FROM .. and CREATE TABLE .. SELECT FROM need
@@ -3972,20 +3973,31 @@ recover_from_failed_open(THD *thd, MDL_request *mdl_request,
           source table. If such a statement gets applied on the slave before
           the INSERT .. SELECT statement finishes, data on the master could
           differ from data on the slave and end-up with a discrepancy between
-          the binary log and table state. Furthermore, this does not apply to
-          I_S and log tables as it's always unsafe to replicate such tables
-          under statement-based replication as the table on the slave might
-          contain other data (ie: general_log is enabled on the slave). The
-          statement will be marked as unsafe for SBR in decide_logging_format().
+          the binary log and table state.
+          This also applies to SELECT/SET/DO statements which use stored
+          functions. Calls to such functions are going to be logged as a
+          whole and thus should be serialized against concurrent changes
+          to tables used by those functions. This can be avoided if functions
+          only read data but doing so requires more complex analysis than it
+          is done now.
+          Furthermore, this does not apply to I_S and log tables as it's
+          always unsafe to replicate such tables under statement-based
+          replication as the table on the slave might contain other data
+          (ie: general_log is enabled on the slave). The statement will
+          be marked as unsafe for SBR in decide_logging_format().
 */
 
-thr_lock_type read_lock_type_for_table(THD *thd, TABLE *table)
+thr_lock_type read_lock_type_for_table(THD *thd,
+                                       Query_tables_list *prelocking_ctx,
+                                       TABLE_LIST *table_list)
 {
   bool log_on= mysql_bin_log.is_open() && (thd->variables.option_bits & OPTION_BIN_LOG);
   ulong binlog_format= thd->variables.binlog_format;
   if ((log_on == FALSE) || (binlog_format == BINLOG_FORMAT_ROW) ||
-      (table->s->table_category == TABLE_CATEGORY_LOG) ||
-      (table->s->table_category == TABLE_CATEGORY_PERFORMANCE))
+      (table_list->table->s->table_category == TABLE_CATEGORY_LOG) ||
+      (table_list->table->s->table_category == TABLE_CATEGORY_PERFORMANCE) ||
+      !(is_update_query(prelocking_ctx->sql_command) ||
+        table_list->prelocking_placeholder))
     return TL_READ;
   else
     return TL_READ_NO_INSERT;
@@ -4336,7 +4348,7 @@ open_and_process_table(THD *thd, LEX *lex, TABLE_LIST *tables,
       tables->table->reginfo.lock_type= thd->update_lock_default;
     else if (tables->lock_type == TL_READ_DEFAULT)
       tables->table->reginfo.lock_type=
-        read_lock_type_for_table(thd, tables->table);
+        read_lock_type_for_table(thd, lex, tables);
     else
       tables->table->reginfo.lock_type= tables->lock_type;
   }
diff --git a/sql/sql_base.h b/sql/sql_base.h
index a57666afe49..77fbc7458ca 100644
--- a/sql/sql_base.h
+++ b/sql/sql_base.h
@@ -123,7 +123,9 @@ TABLE *open_temporary_table(THD *thd, const char *path, const char *db,
 TABLE *find_locked_table(TABLE *list, const char *db, const char *table_name);
 TABLE *find_write_locked_table(TABLE *list, const char *db,
                                const char *table_name);
-thr_lock_type read_lock_type_for_table(THD *thd, TABLE *table);
+thr_lock_type read_lock_type_for_table(THD *thd,
+                                       Query_tables_list *prelocking_ctx,
+                                       TABLE_LIST *table_list);
 
 my_bool mysql_rm_tmp_tables(void);
 bool rm_temporary_table(handlerton *base, char *path);
diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc
index 13f85b24299..6bd6a374883 100644
--- a/sql/sql_lex.cc
+++ b/sql/sql_lex.cc
@@ -350,7 +350,6 @@ void lex_start(THD *thd)
   lex->subqueries= FALSE;
   lex->view_prepare_mode= FALSE;
   lex->derived_tables= 0;
-  lex->lock_option= TL_READ;
   lex->safe_to_cache_query= 1;
   lex->leaf_tables_insert= 0;
   lex->parsing_options.reset();
@@ -363,7 +362,6 @@ void lex_start(THD *thd)
   lex->select_lex.ftfunc_list= &lex->select_lex.ftfunc_list_alloc;
   lex->select_lex.group_list.empty();
   lex->select_lex.order_list.empty();
-  lex->sql_command= SQLCOM_END;
   lex->duplicates= DUP_ERROR;
   lex->ignore= 0;
   lex->spname= NULL;
@@ -1708,7 +1706,6 @@ void st_select_lex::init_query()
   exclude_from_table_unique_test= no_wrap_view_item= FALSE;
   nest_level= 0;
   link_next= 0;
-  lock_option= TL_READ_DEFAULT;
 }
 
 void st_select_lex::init_select()
@@ -2219,6 +2216,7 @@ void LEX::cleanup_lex_after_parse_error(THD *thd)
 
 void Query_tables_list::reset_query_tables_list(bool init)
 {
+  sql_command= SQLCOM_END;
   if (!init && query_tables)
   {
     TABLE_LIST *table= query_tables;
@@ -2281,8 +2279,7 @@ void Query_tables_list::destroy_query_tables_list()
 */
 
 LEX::LEX()
-  :result(0),
-   sql_command(SQLCOM_END), option_type(OPT_DEFAULT), is_lex_started(0)
+  :result(0), option_type(OPT_DEFAULT), is_lex_started(0)
 {
 
   my_init_dynamic_array2(&plugins, sizeof(plugin_ref),
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index 85ae2697f21..6f7acc4a609 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -739,14 +739,6 @@ public:
 
   List<udf_func>     udf_list;                  /* udf function calls stack */
 
-  /**
-    Per sub-query locking strategy.
-    Note: This variable might interfer with the corresponding statement-level
-    variable Lex::lock_option because on how different parser rules depend
-    on eachother.
-  */
-  thr_lock_type lock_option;
-
   /* 
     This is a copy of the original JOIN USING list that comes from
     the parser. The parser :
@@ -1005,8 +997,11 @@ extern const LEX_STRING empty_lex_str;
 
 
 /*
-  Class representing list of all tables used by statement.
-  It also contains information about stored functions used by statement
+  Class representing list of all tables used by statement and other
+  information which is necessary for opening and locking its tables,
+  like SQL command for this statement.
+
+  Also contains information about stored functions used by statement
   since during its execution we may have to add all tables used by its
   stored functions/triggers to this list in order to pre-open and lock
   them.
@@ -1018,6 +1013,13 @@ extern const LEX_STRING empty_lex_str;
 class Query_tables_list
 {
 public:
+  /**
+    SQL command for this statement. Part of this class since the
+    process of opening and locking tables for the statement needs
+    this information to determine correct type of lock for some of
+    the tables.
+  */
+  enum_sql_command sql_command;
   /* Global list of all tables used by this statement */
   TABLE_LIST *query_tables;
   /* Pointer to next_global member of last element in the previous list. */
@@ -1918,7 +1920,6 @@ struct LEX: public Query_tables_list
     the variable can contain 0 or 1 for each nest level.
   */
   nesting_map allow_sum_func;
-  enum_sql_command sql_command;
 
   Sql_statement *m_stmt;
 
@@ -1930,7 +1931,6 @@ struct LEX: public Query_tables_list
   */
   bool expr_allows_subselect;
 
-  thr_lock_type lock_option;
   enum SSL_type ssl_type;			/* defined in violite.h */
   enum enum_duplicates duplicates;
   enum enum_tx_isolation tx_isolation;
@@ -2237,12 +2237,21 @@ class Yacc_state
 {
 public:
   Yacc_state()
-    : yacc_yyss(NULL), yacc_yyvs(NULL)
+    : yacc_yyss(NULL), yacc_yyvs(NULL), m_lock_type(TL_READ_DEFAULT)
   {}
 
   ~Yacc_state();
 
   /**
+    Reset part of the state which needs resetting before parsing
+    substatement.
+  */
+  void reset_before_substatement()
+  {
+    m_lock_type= TL_READ_DEFAULT;
+  }
+
+  /**
     Bison internal state stack, yyss, when dynamically allocated using
     my_yyoverflow().
   */
@@ -2260,6 +2269,25 @@ public:
   */
   Set_signal_information m_set_signal_info;
 
+  /**
+    Type of lock to be used for tables being added to the statement's
+    table list in table_factor, table_alias_ref, single_multi and
+    table_wild_one rules.
+    Statements which use these rules but require lock type different
+    from one specified by this member have to override it by using
+    st_select_lex::set_lock_for_tables() method.
+
+    The default value of this member is TL_READ_DEFAULT. The only two
+    cases in which we change it are:
+    - When parsing SELECT HIGH_PRIORITY.
+    - Rule for DELETE. In which we use this member to pass information
+      about type of lock from delete to single_multi part of rule.
+
+    We should try to avoid introducing new use cases as we would like
+    to get rid of this member eventually.
+  */
+  thr_lock_type m_lock_type;
+
   /*
     TODO: move more attributes from the LEX structure here.
   */
diff --git a/sql/sql_load.cc b/sql/sql_load.cc
index 87a347b9f98..be047a8711f 100644
--- a/sql/sql_load.cc
+++ b/sql/sql_load.cc
@@ -132,6 +132,7 @@ static int read_xml_field(THD *thd, COPY_INFO &info, TABLE_LIST *table_list,
 static bool write_execute_load_query_log_event(THD *thd, sql_exchange* ex,
                                                const char* db_arg, /* table's database */
                                                const char* table_name_arg,
+                                               bool is_concurrent,
                                                enum enum_duplicates duplicates,
                                                bool ignore,
                                                bool transactional_table,
@@ -184,6 +185,7 @@ int mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
   char *tdb= thd->db ? thd->db : db;		// Result is never null
   ulong skip_lines= ex->skip_lines;
   bool transactional_table;
+  bool is_concurrent;
   THD::killed_state killed_status= THD::NOT_KILLED;
   DBUG_ENTER("mysql_load");
 
@@ -245,6 +247,7 @@ int mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
 
   table= table_list->table;
   transactional_table= table->file->has_transactions();
+  is_concurrent= (table_list->lock_type == TL_WRITE_CONCURRENT_INSERT);
 
   if (!fields_vars.elements)
   {
@@ -562,6 +565,7 @@ int mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
             (void) write_execute_load_query_log_event(thd, ex,
                                                       table_list->db, 
                                                       table_list->table_name,
+                                                      is_concurrent,
                                                       handle_duplicates, ignore,
                                                       transactional_table,
                                                       errcode);
@@ -610,6 +614,7 @@ int mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
         int errcode= query_error_code(thd, killed_status == THD::NOT_KILLED);
         error= write_execute_load_query_log_event(thd, ex,
                                                   table_list->db, table_list->table_name,
+                                                  is_concurrent,
                                                   handle_duplicates, ignore,
                                                   transactional_table,
                                                   errcode);
@@ -638,6 +643,7 @@ err:
 static bool write_execute_load_query_log_event(THD *thd, sql_exchange* ex,
                                                const char* db_arg,  /* table's database */
                                                const char* table_name_arg,
+                                               bool is_concurrent,
                                                enum enum_duplicates duplicates,
                                                bool ignore,
                                                bool transactional_table,
@@ -673,8 +679,8 @@ static bool write_execute_load_query_log_event(THD *thd, sql_exchange* ex,
     tbl= string_buf.c_ptr_safe();
   }
 
-  Load_log_event       lle(thd, ex, tdb, tbl, fv, duplicates,
-                           ignore, transactional_table);
+  Load_log_event       lle(thd, ex, tdb, tbl, fv, is_concurrent,
+                           duplicates, ignore, transactional_table);
 
   /*
     force in a LOCAL if there was one in the original.
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 85e6866f3db..42777dbc837 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -5710,7 +5710,6 @@ void mysql_init_multi_delete(LEX *lex)
   lex->select_lex.select_limit= 0;
   lex->unit.select_limit_cnt= HA_POS_ERROR;
   lex->select_lex.table_list.save_and_clear(&lex->auxiliary_table_list);
-  lex->lock_option= TL_READ_DEFAULT;
   lex->query_tables= 0;
   lex->query_tables_last= &lex->query_tables;
 }
diff --git a/sql/sql_priv.h b/sql/sql_priv.h
index eeefd3cac04..20893e0caa8 100644
--- a/sql/sql_priv.h
+++ b/sql/sql_priv.h
@@ -129,6 +129,12 @@ extern char err_shared_dir[];
 */
 #define TMP_TABLE_FORCE_MYISAM          (1ULL << 32)
 #define OPTION_PROFILING                (1ULL << 33)
+/**
+  Indicates that this is a HIGH_PRIORITY SELECT.
+  Currently used only for printing of such selects.
+  Type of locks to be acquired is specified directly.
+*/
+#define SELECT_HIGH_PRIORITY            (1ULL << 34)     // SELECT, user
 
 
 /* The rest of the file is included in the server only */
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 62a51a32ca2..d126d0e4ec6 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -17179,8 +17179,7 @@ void st_select_lex::print(THD *thd, String *str, enum_query_type query_type)
   /* First add options */
   if (options & SELECT_STRAIGHT_JOIN)
     str->append(STRING_WITH_LEN("straight_join "));
-  if ((thd->lex->lock_option == TL_READ_HIGH_PRIORITY) &&
-      (this == &thd->lex->select_lex))
+  if (options & SELECT_HIGH_PRIORITY)
     str->append(STRING_WITH_LEN("high_priority "));
   if (options & SELECT_DISTINCT)
     str->append(STRING_WITH_LEN("distinct "));
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 41117650e4a..f1db513d0e2 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -3341,7 +3341,6 @@ int get_all_tables(THD *thd, TABLE_LIST *tables, COND *cond)
   LEX *lex= thd->lex;
   TABLE *table= tables->table;
   SELECT_LEX *old_all_select_lex= lex->all_selects_list;
-  enum_sql_command save_sql_command= lex->sql_command;
   SELECT_LEX *lsel= tables->schema_select_lex;
   ST_SCHEMA_TABLE *schema_table= tables->schema_table;
   SELECT_LEX sel;
@@ -3377,6 +3376,12 @@ int get_all_tables(THD *thd, TABLE_LIST *tables, COND *cond)
 
   lex->view_prepare_mode= TRUE;
   lex->reset_n_backup_query_tables_list(&query_tables_list_backup);
+  /*
+    Restore Query_tables_list::sql_command value, which was reset
+    above, as ST_SCHEMA_TABLE::process_table() functions often rely
+    that this value reflects which SHOW statement is executed.
+  */
+  lex->sql_command= query_tables_list_backup.sql_command;
 
   /*
     We should not introduce deadlocks even if we already have some
@@ -3539,7 +3544,7 @@ int get_all_tables(THD *thd, TABLE_LIST *tables, COND *cond)
                    (MYSQL_OPEN_IGNORE_FLUSH |
                     MYSQL_OPEN_FORCE_SHARED_HIGH_PRIO_MDL |
                     (can_deadlock ? MYSQL_OPEN_FAIL_ON_MDL_CONFLICT : 0)));
-            lex->sql_command= save_sql_command;
+            lex->sql_command= query_tables_list_backup.sql_command;
             /*
               XXX:  show_table_list has a flag i_is_requested,
               and when it's set, open_normal_and_derived_tables()
@@ -3598,7 +3603,6 @@ err:
   lex->derived_tables= derived_tables;
   lex->all_selects_list= old_all_select_lex;
   lex->view_prepare_mode= save_view_prepare_mode;
-  lex->sql_command= save_sql_command;
   DBUG_RETURN(error);
 }
 
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index c752905d14c..b688d706762 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -4808,6 +4808,7 @@ static bool mysql_admin_table(THD* thd, TABLE_LIST* tables,
       /* purecov: begin inspected */
       char buff[FN_REFLEN + MYSQL_ERRMSG_SIZE];
       size_t length;
+      enum_sql_command save_sql_command= lex->sql_command;
       DBUG_PRINT("admin", ("sending error message"));
       protocol->prepare_for_resend();
       protocol->store(table_name, system_charset_info);
@@ -4821,6 +4822,11 @@ static bool mysql_admin_table(THD* thd, TABLE_LIST* tables,
       close_thread_tables(thd);
       thd->mdl_context.release_transactional_locks();
       lex->reset_query_tables_list(FALSE);
+      /*
+        Restore Query_tables_list::sql_command value to make statement
+        safe for re-execution.
+      */
+      lex->sql_command= save_sql_command;
       table->table=0;				// For query cache
       if (protocol->write())
 	goto err;
diff --git a/sql/sql_trigger.cc b/sql/sql_trigger.cc
index 701a2ec93c2..9ce62d9f2a4 100644
--- a/sql/sql_trigger.cc
+++ b/sql/sql_trigger.cc
@@ -411,6 +411,13 @@ bool mysql_create_or_drop_trigger(THD *thd, TABLE_LIST *tables, bool create)
       destructive changes necessary to open the trigger's table.
     */
     thd->lex->reset_n_backup_query_tables_list(&backup);
+    /*
+      Restore Query_tables_list::sql_command, which was
+      reset above, as the code that writes the query to the
+      binary log assumes that this value corresponds to the
+      statement that is being executed.
+    */
+    thd->lex->sql_command= backup.sql_command;
 
     if (add_table_for_trigger(thd, thd->lex->spname, if_exists, & tables))
       goto end;
diff --git a/sql/sql_update.cc b/sql/sql_update.cc
index 41737b33fb6..412a053014f 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -1054,7 +1054,7 @@ int mysql_multi_update_prepare(THD *thd)
         be write-locked (for example, trigger to be invoked might try
         to update this table).
       */
-      tl->lock_type= read_lock_type_for_table(thd, table);
+      tl->lock_type= read_lock_type_for_table(thd, lex, tl);
       tl->updating= 0;
       /* Update TABLE::lock_type accordingly. */
       if (!tl->placeholder() && !using_lock_tables)
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index a0d64e6a378..aa336f3c072 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -29,6 +29,7 @@
 #define YYLEX_PARAM yythd
 #define YYTHD ((THD *)yythd)
 #define YYLIP (& YYTHD->m_parser_state->m_lip)
+#define YYPS (& YYTHD->m_parser_state->m_yacc)
 
 #define MYSQL_YACC
 #define YYINITDEPTH 100
@@ -4937,7 +4938,6 @@ create_select:
           SELECT_SYM
           {
             LEX *lex=Lex;
-            lex->lock_option= TL_READ_DEFAULT;
             if (lex->sql_command == SQLCOM_INSERT)
               lex->sql_command= SQLCOM_INSERT_SELECT;
             else if (lex->sql_command == SQLCOM_REPLACE)
@@ -7302,7 +7302,6 @@ select_lock_type:
           {
             LEX *lex=Lex;
             lex->current_select->set_lock_for_tables(TL_WRITE);
-            lex->current_select->lock_option= TL_WRITE;
             lex->safe_to_cache_query=0;
             lex->protect_against_global_read_lock= TRUE;
           }
@@ -7311,7 +7310,6 @@ select_lock_type:
             LEX *lex=Lex;
             lex->current_select->
               set_lock_for_tables(TL_READ_WITH_SHARED_LOCKS);
-            lex->current_select->lock_option= TL_READ_WITH_SHARED_LOCKS;
             lex->safe_to_cache_query=0;
           }
         ;
@@ -9221,7 +9219,7 @@ table_factor:
           {
             if (!($$= Select->add_table_to_list(YYTHD, $2, $3,
                                                 Select->get_table_join_options(),
-                                                Lex->lock_option,
+                                                YYPS->m_lock_type,
                                                 Select->pop_index_hints())))
               MYSQL_YYABORT;
             Select->add_joined_table($$);
@@ -10278,7 +10276,7 @@ table_alias_ref:
           {
             if (!Select->add_table_to_list(YYTHD, $1, NULL,
                                            TL_OPTION_UPDATING | TL_OPTION_ALIAS,
-                                           Lex->lock_option ))
+                                           YYPS->m_lock_type))
               MYSQL_YYABORT;
           }
         ;
@@ -10303,8 +10301,6 @@ insert:
             lex->sql_command= SQLCOM_INSERT;
             lex->duplicates= DUP_ERROR; 
             mysql_init_select(lex);
-            /* for subselects */
-            lex->lock_option= TL_READ_DEFAULT;
           }
           insert_lock_option
           opt_ignore insert2
@@ -10495,7 +10491,6 @@ update:
             LEX *lex= Lex;
             mysql_init_select(lex);
             lex->sql_command= SQLCOM_UPDATE;
-            lex->lock_option= TL_UNLOCK; /* Will be set later */
             lex->duplicates= DUP_ERROR; 
           }
           opt_low_priority opt_ignore join_table_list
@@ -10562,7 +10557,7 @@ delete:
             LEX *lex= Lex;
             lex->sql_command= SQLCOM_DELETE;
             mysql_init_select(lex);
-            lex->lock_option= TL_WRITE_DEFAULT;
+            YYPS->m_lock_type= TL_WRITE_DEFAULT;
             lex->ignore= 0;
             lex->select_lex.init_order();
           }
@@ -10573,20 +10568,27 @@ single_multi:
           FROM table_ident
           {
             if (!Select->add_table_to_list(YYTHD, $2, NULL, TL_OPTION_UPDATING,
-                                           Lex->lock_option))
+                                           YYPS->m_lock_type))
               MYSQL_YYABORT;
+            YYPS->m_lock_type= TL_READ_DEFAULT;
           }
           where_clause opt_order_clause
           delete_limit_clause {}
         | table_wild_list
-          { mysql_init_multi_delete(Lex); }
+          {
+            mysql_init_multi_delete(Lex);
+            YYPS->m_lock_type= TL_READ_DEFAULT;
+          }
           FROM join_table_list where_clause
           {
             if (multi_delete_set_locks_and_link_aux_tables(Lex))
               MYSQL_YYABORT;
           }
         | FROM table_alias_ref_list
-          { mysql_init_multi_delete(Lex); }
+          {
+            mysql_init_multi_delete(Lex);
+            YYPS->m_lock_type= TL_READ_DEFAULT;
+          }
           USING join_table_list where_clause
           {
             if (multi_delete_set_locks_and_link_aux_tables(Lex))
@@ -10609,7 +10611,7 @@ table_wild_one:
                                            ti,
                                            NULL,
                                            TL_OPTION_UPDATING | TL_OPTION_ALIAS,
-                                           Lex->lock_option))
+                                           YYPS->m_lock_type))
               MYSQL_YYABORT;
           }
         | ident '.' ident opt_wild
@@ -10621,7 +10623,7 @@ table_wild_one:
                                            ti,
                                            NULL,
                                            TL_OPTION_UPDATING | TL_OPTION_ALIAS,
-                                           Lex->lock_option))
+                                           YYPS->m_lock_type))
               MYSQL_YYABORT;
           }
         ;
@@ -10638,7 +10640,7 @@ opt_delete_options:
 
 opt_delete_option:
           QUICK        { Select->options|= OPTION_QUICK; }
-        | LOW_PRIORITY { Lex->lock_option= TL_WRITE_LOW_PRIORITY; }
+        | LOW_PRIORITY { YYPS->m_lock_type= TL_WRITE_LOW_PRIORITY; }
         | IGNORE_SYM   { Lex->ignore= 1; }
         ;
 
@@ -10724,7 +10726,6 @@ show:
           {
             LEX *lex=Lex;
             lex->wild=0;
-            lex->lock_option= TL_READ;
             mysql_init_select(lex);
             lex->current_select->parsing_place= SELECT_LIST;
             bzero((char*) &lex->create_info,sizeof(lex->create_info));
@@ -11077,7 +11078,6 @@ describe:
           describe_command table_ident
           {
             LEX *lex= Lex;
-            lex->lock_option= TL_READ;
             mysql_init_select(lex);
             lex->current_select->parsing_place= SELECT_LIST;
             lex->sql_command= SQLCOM_SHOW_FIELDS;
@@ -11291,7 +11291,6 @@ load:
           {
             LEX *lex=Lex;
             lex->sql_command= SQLCOM_LOAD;
-            lex->lock_option= $4;
             lex->local_file=  $5;
             lex->duplicates= DUP_ERROR;
             lex->ignore= 0;
@@ -11302,7 +11301,7 @@ load:
           {
             LEX *lex=Lex;
             if (!Select->add_table_to_list(YYTHD, $12, NULL, TL_OPTION_UPDATING,
-                                           lex->lock_option))
+                                           $4))
               MYSQL_YYABORT;
             lex->field_list.empty();
             lex->update_list.empty();
@@ -13734,17 +13733,6 @@ subselect_start:
 subselect_end:
           {
             LEX *lex=Lex;
-            /*
-              Set the required lock level for the tables associated with the
-              current sub-select. This will overwrite previous lock options set
-              using st_select_lex::add_table_to_list in any of the following
-              rules: single_multi, table_wild_one, load_data, table_alias_ref,
-              table_factor.
-              The default lock level is TL_READ_DEFAULT but it can be modified
-              with query options specific for a certain (sub-)SELECT.
-            */
-            lex->current_select->
-              set_lock_for_tables(lex->current_select->lock_option);
 
             lex->pop_context();
             SELECT_LEX *child= lex->current_select;
@@ -13776,8 +13764,8 @@ query_expression_option:
           {
             if (check_simple_select())
               MYSQL_YYABORT;
-            Lex->lock_option= TL_READ_HIGH_PRIORITY;
-            Lex->current_select->lock_option= TL_READ_HIGH_PRIORITY;
+            YYPS->m_lock_type= TL_READ_HIGH_PRIORITY;
+            Select->options|= SELECT_HIGH_PRIORITY;
           }
         | DISTINCT         { Select->options|= SELECT_DISTINCT; }
         | SQL_SMALL_RESULT { Select->options|= SELECT_SMALL_RESULT; }