merge from mysql-trunk-runtime

27 files changed, 333 insertions, 147 deletions

diff --git a/sql/ha_ndbcluster.cc b/sql/ha_ndbcluster.cc
index 38324f3cf19..2d082cc71f6 100644
--- a/sql/ha_ndbcluster.cc
+++ b/sql/ha_ndbcluster.cc
@@ -4628,7 +4628,7 @@ int ha_ndbcluster::start_statement(THD *thd,
   trans_register_ha(thd, FALSE, ndbcluster_hton);
   if (!thd_ndb->trans)
   {
-    if (thd->in_multi_stmt_transaction())
+    if (thd->in_multi_stmt_transaction_mode())
       trans_register_ha(thd, TRUE, ndbcluster_hton);
     DBUG_PRINT("trans",("Starting transaction"));      
     thd_ndb->trans= ndb->startTransaction();
@@ -4698,7 +4698,7 @@ int ha_ndbcluster::init_handler_for_statement(THD *thd, Thd_ndb *thd_ndb)
   }
 #endif
 
-  if (thd->in_multi_stmt_transaction())
+  if (thd->in_multi_stmt_transaction_mode())
   {
     const void *key= m_table;
     HASH_SEARCH_STATE state;
@@ -4782,7 +4782,7 @@ int ha_ndbcluster::external_lock(THD *thd, int lock_type)
     if (opt_ndb_cache_check_time && m_rows_changed)
     {
       DBUG_PRINT("info", ("Rows has changed and util thread is running"));
-      if (thd->in_multi_stmt_transaction())
+      if (thd->in_multi_stmt_transaction_mode())
       {
         DBUG_PRINT("info", ("Add share to list of tables to be invalidated"));
         /* NOTE push_back allocates memory using transactions mem_root! */
@@ -4801,7 +4801,7 @@ int ha_ndbcluster::external_lock(THD *thd, int lock_type)
       DBUG_PRINT("trans", ("Last external_lock"));
       PRINT_OPTION_FLAGS(thd);
 
-      if (!thd->in_multi_stmt_transaction())
+      if (!thd->in_multi_stmt_transaction_mode())
       {
         if (thd_ndb->trans)
         {
@@ -4911,7 +4911,7 @@ static int ndbcluster_commit(handlerton *hton, THD *thd, bool all)
   PRINT_OPTION_FLAGS(thd);
   DBUG_PRINT("enter", ("Commit %s", (all ? "all" : "stmt")));
   thd_ndb->start_stmt_count= 0;
-  if (trans == NULL || (!all && thd->in_multi_stmt_transaction()))
+  if (trans == NULL || (!all && thd->in_multi_stmt_transaction_mode()))
   {
     /*
       An odditity in the handler interface is that commit on handlerton
@@ -4981,7 +4981,7 @@ static int ndbcluster_rollback(handlerton *hton, THD *thd, bool all)
   DBUG_ASSERT(ndb);
   thd_ndb->start_stmt_count= 0;
   if (trans == NULL || (!all &&
-      thd->in_multi_stmt_transaction()))
+      thd->in_multi_stmt_transaction_mode()))
   {
     /* Ignore end-of-statement until real rollback or commit is called */
     DBUG_PRINT("info", ("Rollback before start or end-of-statement only"));
@@ -8271,7 +8271,7 @@ ndbcluster_cache_retrieval_allowed(THD *thd,
   DBUG_ENTER("ndbcluster_cache_retrieval_allowed");
   DBUG_PRINT("enter", ("dbname: %s, tabname: %s", dbname, tabname));
 
-  if (thd->in_multi_stmt_transaction())
+  if (thd->in_multi_stmt_transaction_mode())
   {
     DBUG_PRINT("exit", ("No, don't use cache in transaction"));
     DBUG_RETURN(FALSE);
@@ -8339,7 +8339,7 @@ ha_ndbcluster::register_query_cache_table(THD *thd,
   DBUG_ENTER("ha_ndbcluster::register_query_cache_table");
   DBUG_PRINT("enter",("dbname: %s, tabname: %s", m_dbname, m_tabname));
 
-  if (thd->in_multi_stmt_transaction())
+  if (thd->in_multi_stmt_transaction_mode())
   {
     DBUG_PRINT("exit", ("Can't register table during transaction"));
     DBUG_RETURN(FALSE);
diff --git a/sql/handler.cc b/sql/handler.cc
index ee02441e7ff..c0a5e2ff55c 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -1245,7 +1245,14 @@ end:
 /**
   @note
   This function does not care about global read lock. A caller should.
+
+  @param[in]  all  Is set in case of explicit commit
+                   (COMMIT statement), or implicit commit
+                   issued by DDL. Is not set when called
+                   at the end of statement, even if
+                   autocommit=1.
 */
+
 int ha_commit_one_phase(THD *thd, bool all)
 {
   int error=0;
@@ -1253,9 +1260,15 @@ int ha_commit_one_phase(THD *thd, bool all)
   /*
     "real" is a nick name for a transaction for which a commit will
     make persistent changes. E.g. a 'stmt' transaction inside a 'all'
-    transation is not 'real': even though it's possible to commit it,
+    transaction is not 'real': even though it's possible to commit it,
     the changes are not durable as they might be rolled back if the
     enclosing 'all' transaction is rolled back.
+    We establish the value of 'is_real_trans' by checking
+    if it's an explicit COMMIT/BEGIN statement, or implicit
+    commit issued by DDL (all == TRUE), or if we're running
+    in autocommit mode (it's only in the autocommit mode
+    ha_commit_one_phase() can be called with an empty
+    transaction.all.ha_list, see why in trans_register_ha()).
   */
   bool is_real_trans=all || thd->transaction.all.ha_list == 0;
   Ha_trx_info *ha_info= trans->ha_list, *ha_info_next;
@@ -1303,9 +1316,15 @@ int ha_rollback_trans(THD *thd, bool all)
   /*
     "real" is a nick name for a transaction for which a commit will
     make persistent changes. E.g. a 'stmt' transaction inside a 'all'
-    transation is not 'real': even though it's possible to commit it,
+    transaction is not 'real': even though it's possible to commit it,
     the changes are not durable as they might be rolled back if the
     enclosing 'all' transaction is rolled back.
+    We establish the value of 'is_real_trans' by checking
+    if it's an explicit COMMIT or BEGIN statement, or implicit
+    commit issued by DDL (in these cases all == TRUE),
+    or if we're running in autocommit mode (it's only in the autocommit mode
+    ha_commit_one_phase() is called with an empty
+    transaction.all.ha_list, see why in trans_register_ha()).
   */
   bool is_real_trans=all || thd->transaction.all.ha_list == 0;
   DBUG_ENTER("ha_rollback_trans");
@@ -1358,7 +1377,7 @@ int ha_rollback_trans(THD *thd, bool all)
     if (all)
       thd->variables.tx_isolation=thd->session_tx_isolation;
   }
-  /* Always cleanup. Even if there nht==0. There may be savepoints. */
+  /* Always cleanup. Even if nht==0. There may be savepoints. */
   if (is_real_trans)
     thd->transaction.cleanup();
   if (all)
diff --git a/sql/log.cc b/sql/log.cc
index 18f812d96bf..6c0ab33e87a 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -1686,7 +1686,7 @@ static int binlog_commit(handlerton *hton, THD *thd, bool all)
   DBUG_PRINT("debug",
              ("all: %d, in_transaction: %s, all.modified_non_trans_table: %s, stmt.modified_non_trans_table: %s",
               all,
-              YESNO(thd->in_multi_stmt_transaction()),
+              YESNO(thd->in_multi_stmt_transaction_mode()),
               YESNO(thd->transaction.all.modified_non_trans_table),
               YESNO(thd->transaction.stmt.modified_non_trans_table)));
 
@@ -4267,7 +4267,7 @@ bool use_trans_cache(const THD* thd, bool is_transactional)
 */
 bool ending_trans(THD* thd, const bool all)
 {
-  return (all || (!all && !thd->in_multi_stmt_transaction()));
+  return (all || (!all && !thd->in_multi_stmt_transaction_mode()));
 }
 
 /**
@@ -4370,7 +4370,7 @@ THD::binlog_start_trans_and_stmt()
       cache_mngr->trx_cache.get_prev_position() == MY_OFF_T_UNDEF)
   {
     this->binlog_set_stmt_begin();
-    if (in_multi_stmt_transaction())
+    if (in_multi_stmt_transaction_mode())
       trans_register_ha(this, TRUE, binlog_hton);
     trans_register_ha(this, FALSE, binlog_hton);
     /*
diff --git a/sql/log_event.cc b/sql/log_event.cc
index 2fdc1da4249..d5ae1c954ff 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -2485,13 +2485,13 @@ Query_log_event::Query_log_event(THD* thd_arg, const char* query_arg,
       implicit_commit= TRUE;
       break;
     case SQLCOM_DROP_TABLE:
-      force_trans= lex->drop_temporary && thd->in_multi_stmt_transaction();
+      force_trans= lex->drop_temporary && thd->in_multi_stmt_transaction_mode();
       implicit_commit= !force_trans;
       break;
     case SQLCOM_ALTER_TABLE:
     case SQLCOM_CREATE_TABLE:
       force_trans= (lex->create_info.options & HA_LEX_CREATE_TMP_TABLE) &&
-                    thd->in_multi_stmt_transaction();
+                    thd->in_multi_stmt_transaction_mode();
       implicit_commit= !force_trans &&
                        !(lex->select_lex.item_list.elements &&
                          thd->is_current_stmt_binlog_format_row());
@@ -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..28633365e28 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -1558,7 +1558,7 @@ void close_thread_tables(THD *thd)
     - If in autocommit mode, or outside a transactional context,
     automatically release metadata locks of the current statement.
   */
-  if (! thd->in_multi_stmt_transaction() &&
+  if (! thd->in_multi_stmt_transaction_mode() &&
       ! (thd->state_flags & Open_tables_state::BACKUPS_AVAIL))
   {
     thd->mdl_context.release_transactional_locks();
@@ -3783,7 +3783,7 @@ end_with_lock_open:
 Open_table_context::Open_table_context(THD *thd, ulong timeout)
   :m_action(OT_NO_ACTION),
    m_start_of_statement_svp(thd->mdl_context.mdl_savepoint()),
-   m_has_locks((thd->in_multi_stmt_transaction() &&
+   m_has_locks((thd->in_multi_stmt_transaction_mode() &&
                 thd->mdl_context.has_locks()) ||
                 thd->mdl_context.trans_sentinel()),
    m_global_mdl_request(NULL),
@@ -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,44 @@ 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().
+  @remark Note that even in prelocked mode it is important to correctly
+          determine lock type value. In this mode lock type is passed to
+          handler::start_stmt() method and can be used by storage engine,
+          for example, to determine what kind of row locks it should acquire
+          when reading data from the table.
 */
 
-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);
+  /*
+    In cases when this function is called for a sub-statement executed in
+    prelocked mode we can't rely on OPTION_BIN_LOG flag in THD::options
+    bitmap to determine that binary logging is turned on as this bit can
+    be cleared before executing sub-statement. So instead we have to look
+    at THD::sql_log_bin_toplevel member.
+  */
+  bool log_on= mysql_bin_log.is_open() && thd->sql_log_bin_toplevel;
   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 ||
+        (thd->locked_tables_mode > LTM_LOCK_TABLES)))
     return TL_READ;
   else
     return TL_READ_NO_INSERT;
@@ -4336,7 +4361,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;
   }
@@ -4989,35 +5014,49 @@ handle_view(THD *thd, Query_tables_list *prelocking_ctx,
 }
 
 
-/*
+/**
   Check that lock is ok for tables; Call start stmt if ok
 
-  SYNOPSIS
-    check_lock_and_start_stmt()
-    thd			Thread handle
-    table_list		Table to check
-    lock_type		Lock used for table
+  @param thd             Thread handle.
+  @param prelocking_ctx  Prelocking context.
+  @param table_list      Table list element for table to be checked.
 
-  RETURN VALUES
-  0	ok
-  1	error
+  @retval FALSE - Ok.
+  @retval TRUE  - Error.
 */
 
-static bool check_lock_and_start_stmt(THD *thd, TABLE *table,
-				      thr_lock_type lock_type)
+static bool check_lock_and_start_stmt(THD *thd,
+                                      Query_tables_list *prelocking_ctx,
+                                      TABLE_LIST *table_list)
 {
   int error;
+  thr_lock_type lock_type;
   DBUG_ENTER("check_lock_and_start_stmt");
 
+  /*
+    TL_WRITE_DEFAULT and TL_READ_DEFAULT are supposed to be parser only
+    types of locks so they should be converted to appropriate other types
+    to be passed to storage engine. The exact lock type passed to the
+    engine is important as, for example, InnoDB uses it to determine
+    what kind of row locks should be acquired when executing statement
+    in prelocked mode or under LOCK TABLES with @@innodb_table_locks = 0.
+  */
+  if (table_list->lock_type == TL_WRITE_DEFAULT)
+    lock_type= thd->update_lock_default;
+  else if (table_list->lock_type == TL_READ_DEFAULT)
+    lock_type= read_lock_type_for_table(thd, prelocking_ctx, table_list);
+  else
+    lock_type= table_list->lock_type;
+
   if ((int) lock_type >= (int) TL_WRITE_ALLOW_READ &&
-      (int) table->reginfo.lock_type < (int) TL_WRITE_ALLOW_READ)
+      (int) table_list->table->reginfo.lock_type < (int) TL_WRITE_ALLOW_READ)
   {
-    my_error(ER_TABLE_NOT_LOCKED_FOR_WRITE, MYF(0),table->alias);
+    my_error(ER_TABLE_NOT_LOCKED_FOR_WRITE, MYF(0), table_list->alias);
     DBUG_RETURN(1);
   }
-  if ((error=table->file->start_stmt(thd, lock_type)))
+  if ((error= table_list->table->file->start_stmt(thd, lock_type)))
   {
-    table->file->print_error(error,MYF(0));
+    table_list->table->file->print_error(error, MYF(0));
     DBUG_RETURN(1);
   }
   DBUG_RETURN(0);
@@ -5162,7 +5201,7 @@ TABLE *open_ltable(THD *thd, TABLE_LIST *table_list, thr_lock_type lock_type,
     table->grant= table_list->grant;
     if (thd->locked_tables_mode)
     {
-      if (check_lock_and_start_stmt(thd, table, lock_type))
+      if (check_lock_and_start_stmt(thd, thd->lex, table_list))
 	table= 0;
     }
     else
@@ -5390,7 +5429,7 @@ bool lock_tables(THD *thd, TABLE_LIST *tables, uint count,
         if (!table->placeholder())
         {
           table->table->query_id= thd->query_id;
-          if (check_lock_and_start_stmt(thd, table->table, table->lock_type))
+          if (check_lock_and_start_stmt(thd, thd->lex, table))
           {
             mysql_unlock_tables(thd, thd->lock);
             thd->lock= 0;
@@ -5444,7 +5483,7 @@ bool lock_tables(THD *thd, TABLE_LIST *tables, uint count,
         }
       }
 
-      if (check_lock_and_start_stmt(thd, table->table, table->lock_type))
+      if (check_lock_and_start_stmt(thd, thd->lex, table))
       {
 	DBUG_RETURN(TRUE);
       }
diff --git a/sql/sql_base.h b/sql/sql_base.h
index a57666afe49..0c16151e43a 100644
--- a/sql/sql_base.h
+++ b/sql/sql_base.h
@@ -56,9 +56,6 @@ enum enum_resolution_type {
   RESOLVED_AGAINST_ALIAS
 };
 
-enum enum_open_table_action {OT_NO_ACTION= 0, OT_BACK_OFF_AND_RETRY,
-                             OT_DISCOVER, OT_REPAIR};
-
 enum find_item_error_report_type {REPORT_ALL_ERRORS, REPORT_EXCEPT_NOT_FOUND,
 				  IGNORE_ERRORS, REPORT_EXCEPT_NON_UNIQUE,
                                   IGNORE_EXCEPT_NON_UNIQUE};
@@ -123,7 +120,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_cache.cc b/sql/sql_cache.cc
index 1e4161dfa1c..92d54c8e71b 100644
--- a/sql/sql_cache.cc
+++ b/sql/sql_cache.cc
@@ -1177,7 +1177,7 @@ void Query_cache::store_query(THD *thd, TABLE_LIST *tables_used)
     DBUG_ASSERT(flags.protocol_type != (unsigned int) Protocol::PROTOCOL_LOCAL);
     flags.more_results_exists= test(thd->server_status &
                                     SERVER_MORE_RESULTS_EXISTS);
-    flags.in_trans= test(thd->server_status & SERVER_STATUS_IN_TRANS);
+    flags.in_trans= thd->in_active_multi_stmt_transaction();
     flags.autocommit= test(thd->server_status & SERVER_STATUS_AUTOCOMMIT);
     flags.pkt_nr= net->pkt_nr;
     flags.character_set_client_num=
@@ -1470,7 +1470,7 @@ Query_cache::send_result_to_client(THD *thd, char *sql, uint query_length)
   flags.protocol_type= (unsigned int) thd->protocol->type();
   flags.more_results_exists= test(thd->server_status &
                                   SERVER_MORE_RESULTS_EXISTS);
-  flags.in_trans= test(thd->server_status & SERVER_STATUS_IN_TRANS);
+  flags.in_trans= thd->in_active_multi_stmt_transaction();
   flags.autocommit= test(thd->server_status & SERVER_STATUS_AUTOCOMMIT);
   flags.pkt_nr= thd->net.pkt_nr;
   flags.character_set_client_num= thd->variables.character_set_client->number;
@@ -1541,7 +1541,7 @@ def_week_frmt: %lu, in_trans: %d, autocommit: %d",
   }
   DBUG_PRINT("qcache", ("Query have result 0x%lx", (ulong) query));
 
-  if (thd->in_multi_stmt_transaction() &&
+  if (thd->in_multi_stmt_transaction_mode() &&
       (query->tables_type() & HA_CACHE_TBL_TRANSACT))
   {
     DBUG_PRINT("qcache",
@@ -1698,7 +1698,7 @@ void Query_cache::invalidate(THD *thd, TABLE_LIST *tables_used,
   if (is_disabled())
     DBUG_VOID_RETURN;
 
-  using_transactions= using_transactions && thd->in_multi_stmt_transaction();
+  using_transactions= using_transactions && thd->in_multi_stmt_transaction_mode();
   for (; tables_used; tables_used= tables_used->next_local)
   {
     DBUG_ASSERT(!using_transactions || tables_used->table!=0);
@@ -1782,7 +1782,7 @@ void Query_cache::invalidate(THD *thd, TABLE *table,
   if (is_disabled())
     DBUG_VOID_RETURN;
 
-  using_transactions= using_transactions && thd->in_multi_stmt_transaction();
+  using_transactions= using_transactions && thd->in_multi_stmt_transaction_mode();
   if (using_transactions && 
       (table->file->table_cache_type() == HA_CACHE_TBL_TRANSACT))
     thd->add_changed_table(table);
@@ -1800,7 +1800,7 @@ void Query_cache::invalidate(THD *thd, const char *key, uint32  key_length,
   if (is_disabled())
    DBUG_VOID_RETURN;
 
-  using_transactions= using_transactions && thd->in_multi_stmt_transaction();
+  using_transactions= using_transactions && thd->in_multi_stmt_transaction_mode();
   if (using_transactions) // used for innodb => has_transactions() is TRUE
     thd->add_changed_table(key, key_length);
   else
@@ -3572,7 +3572,7 @@ Query_cache::is_cacheable(THD *thd, size_t query_len, const char *query,
                                                 tables_type)))
       DBUG_RETURN(0);
 
-    if (thd->in_multi_stmt_transaction() &&
+    if (thd->in_multi_stmt_transaction_mode() &&
 	((*tables_type)&HA_CACHE_TBL_TRANSACT))
     {
       DBUG_PRINT("qcache", ("not in autocommin mode"));
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 9ed6bdcf098..789b01443f7 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -1443,7 +1443,7 @@ void THD::add_changed_table(TABLE *table)
 {
   DBUG_ENTER("THD::add_changed_table(table)");
 
-  DBUG_ASSERT(in_multi_stmt_transaction() && table->file->has_transactions());
+  DBUG_ASSERT(in_multi_stmt_transaction_mode() && table->file->has_transactions());
   add_changed_table(table->s->table_cache_key.str,
                     (long) table->s->table_cache_key.length);
   DBUG_VOID_RETURN;
diff --git a/sql/sql_class.h b/sql/sql_class.h
index d9bf6c9bb98..916b79f8353 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -2352,10 +2352,6 @@ public:
   {
     return limit_found_rows;
   }
-  inline bool active_transaction()
-  {
-    return server_status & SERVER_STATUS_IN_TRANS;
-  }
   /**
     Returns TRUE if session is in a multi-statement transaction mode.
 
@@ -2366,11 +2362,60 @@ public:
     OPTION_BEGIN: Regardless of the autocommit status, a multi-statement
     transaction can be explicitly started with the statements "START
     TRANSACTION", "BEGIN [WORK]", "[COMMIT | ROLLBACK] AND CHAIN", etc.
+
+    Note: this doesn't tell you whether a transaction is active.
+    A session can be in multi-statement transaction mode, and yet
+    have no active transaction, e.g., in case of:
+    set @@autocommit=0;
+    set @a= 3;                                     <-- these statements don't
+    set transaction isolation level serializable;  <-- start an active
+    flush tables;                                  <-- transaction
+
+    I.e. for the above scenario this function returns TRUE, even
+    though no active transaction has begun.
+    @sa in_active_multi_stmt_transaction()
   */
-  inline bool in_multi_stmt_transaction()
+  inline bool in_multi_stmt_transaction_mode()
   {
     return variables.option_bits & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN);
   }
+  /**
+    TRUE if the session is in a multi-statement transaction mode
+    (@sa in_multi_stmt_transaction_mode()) *and* there is an
+    active transaction, i.e. there is an explicit start of a
+    transaction with BEGIN statement, or implicit with a
+    statement that uses a transactional engine.
+
+    For example, these scenarios don't start an active transaction
+    (even though the server is in multi-statement transaction mode):
+
+    set @@autocommit=0;
+    select * from nontrans_table;
+    set @var=TRUE;
+    flush tables;
+
+    Note, that even for a statement that starts a multi-statement
+    transaction (i.e. select * from trans_table), this
+    flag won't be set until we open the statement's tables
+    and the engines register themselves for the transaction
+    (see trans_register_ha()),
+    hence this method is reliable to use only after
+    open_tables() has completed.
+
+    Why do we need a flag?
+    ----------------------
+    We need to maintain a (at first glance redundant)
+    session flag, rather than looking at thd->transaction.all.ha_list
+    because of explicit start of a transaction with BEGIN. 
+
+    I.e. in case of
+    BEGIN;
+    select * from nontrans_t1; <-- in_active_multi_stmt_transaction() is true
+  */
+  inline bool in_active_multi_stmt_transaction()
+  {
+    return server_status & SERVER_STATUS_IN_TRANS;
+  }
   inline bool fill_derived_tables()
   {
     return !stmt_arena->is_stmt_prepare() && !lex->only_view_structure();
diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc
index 860296b4050..1795bc272f1 100644
--- a/sql/sql_lex.cc
+++ b/sql/sql_lex.cc
@@ -377,7 +377,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();
@@ -390,7 +389,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;
@@ -1735,7 +1733,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()
@@ -2246,6 +2243,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;
@@ -2308,8 +2306,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 5230753e374..2ce6bdeed42 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. */
@@ -1920,7 +1922,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;
 
@@ -1932,7 +1933,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;
@@ -2248,11 +2248,21 @@ public:
     yacc_yyss= NULL;
     yacc_yyvs= NULL;
     m_set_signal_info.clear();
+    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().
   */
@@ -2270,6 +2280,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 9bab87e2720..ff9c16d229b 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 c787ee02f28..60b2ffa4179 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -2765,7 +2765,7 @@ end_with_restore_list:
     client thread has locked tables
   */
   if (thd->locked_tables_mode ||
-      thd->active_transaction() || thd->global_read_lock.is_acquired())
+      thd->in_active_multi_stmt_transaction() || thd->global_read_lock.is_acquired())
   {
     my_message(ER_LOCK_OR_ACTIVE_TRANSACTION,
                ER(ER_LOCK_OR_ACTIVE_TRANSACTION), MYF(0));
@@ -3273,7 +3273,7 @@ end_with_restore_list:
       Don't allow this within a transaction because we want to use
       re-generate table
     */
-    if (thd->active_transaction())
+    if (thd->in_active_multi_stmt_transaction())
     {
       my_message(ER_LOCK_OR_ACTIVE_TRANSACTION,
                  ER(ER_LOCK_OR_ACTIVE_TRANSACTION), MYF(0));
@@ -4703,6 +4703,9 @@ finish:
     thd->global_read_lock.start_waiting_global_read_lock(thd);
   }
 
+  DBUG_ASSERT(!thd->in_active_multi_stmt_transaction() ||
+               thd->in_multi_stmt_transaction_mode());
+
   if (stmt_causes_implicit_commit(thd, CF_IMPLICIT_COMMIT_END))
   {
     /* If commit fails, we should be able to reset the OK status. */
@@ -5516,7 +5519,7 @@ void THD::reset_for_next_command()
     OPTION_STATUS_NO_TRANS_UPDATE | OPTION_KEEP_LOG to not get warnings
     in ha_rollback_trans() about some tables couldn't be rolled back.
   */
-  if (!thd->in_multi_stmt_transaction())
+  if (!thd->in_multi_stmt_transaction_mode())
   {
     thd->variables.option_bits&= ~OPTION_KEEP_LOG;
     thd->transaction.all.modified_non_trans_table= FALSE;
@@ -5702,7 +5705,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_prepare.cc b/sql/sql_prepare.cc
index 739b6576a99..c2d3c595d95 100644
--- a/sql/sql_prepare.cc
+++ b/sql/sql_prepare.cc
@@ -1708,7 +1708,7 @@ static bool mysql_test_create_table(Prepared_statement *stmt)
 
     res= select_like_stmt_test(stmt, 0, 0);
 
-    lex->link_first_table_back(create_table, &link_to_local);
+    lex->link_first_table_back(create_table, link_to_local);
   }
   else
   {
@@ -3246,7 +3246,7 @@ bool Prepared_statement::prepare(const char *packet, uint packet_len)
     locks have already been released and our savepoint points
     to ticket which has been released as well.
   */
-  if (thd->in_multi_stmt_transaction())
+  if (thd->in_multi_stmt_transaction_mode())
     thd->mdl_context.rollback_to_savepoint(mdl_savepoint);
   thd->restore_backup_statement(this, &stmt_backup);
   thd->stmt_arena= old_stmt_arena;
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_rename.cc b/sql/sql_rename.cc
index d387010141c..ea95b59b0c2 100644
--- a/sql/sql_rename.cc
+++ b/sql/sql_rename.cc
@@ -54,7 +54,7 @@ bool mysql_rename_tables(THD *thd, TABLE_LIST *table_list, bool silent)
     if the user is trying to to do this in a transcation context
   */
 
-  if (thd->locked_tables_mode || thd->active_transaction())
+  if (thd->locked_tables_mode || thd->in_active_multi_stmt_transaction())
   {
     my_message(ER_LOCK_OR_ACTIVE_TRANSACTION,
                ER(ER_LOCK_OR_ACTIVE_TRANSACTION), MYF(0));
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..2b8e7de3a60 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;
@@ -5018,7 +5024,7 @@ send_result_message:
         /* Clear the ticket released in close_thread_tables(). */
         table->mdl_request.ticket= NULL;
         DEBUG_SYNC(thd, "ha_admin_open_ltable");
-        if (table->table= open_ltable(thd, table, lock_type, 0))
+        if ((table->table= open_ltable(thd, table, lock_type, 0)))
         {
           result_code= table->table->file->ha_analyze(thd, check_opt);
           if (result_code == HA_ADMIN_ALREADY_DONE)
@@ -6553,7 +6559,7 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
       if the user is trying to to do this in a transcation context
     */
 
-    if (thd->locked_tables_mode || thd->active_transaction())
+    if (thd->locked_tables_mode || thd->in_active_multi_stmt_transaction())
     {
       my_message(ER_LOCK_OR_ACTIVE_TRANSACTION,
                  ER(ER_LOCK_OR_ACTIVE_TRANSACTION), MYF(0));
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 fb3a7605a94..300ca1098fb 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -1053,7 +1053,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; }
diff --git a/sql/sys_vars.cc b/sql/sys_vars.cc
index b5df2ae58c1..b8312fc3255 100644
--- a/sql/sys_vars.cc
+++ b/sql/sys_vars.cc
@@ -295,7 +295,7 @@ static bool binlog_format_check(sys_var *self, THD *thd, set_var *var)
   /*
     Make the session variable 'binlog_format' read-only inside a transaction.
   */
-  if (thd->active_transaction())
+  if (thd->in_active_multi_stmt_transaction())
   {
     my_error(ER_INSIDE_TRANSACTION_PREVENTS_SWITCH_BINLOG_FORMAT, MYF(0));
     return true;
@@ -348,7 +348,7 @@ static bool binlog_direct_check(sys_var *self, THD *thd, set_var *var)
      Makes the session variable 'binlog_direct_non_transactional_updates'
      read-only inside a transaction.
    */
-   if (thd->active_transaction())
+   if (thd->in_active_multi_stmt_transaction())
    {
      my_error(ER_INSIDE_TRANSACTION_PREVENTS_SWITCH_BINLOG_DIRECT, MYF(0));
      return true;
@@ -1428,7 +1428,7 @@ static my_bool read_only;
 static bool check_read_only(sys_var *self, THD *thd, set_var *var)
 {
   /* Prevent self dead-lock */
-  if (thd->locked_tables_mode || thd->active_transaction())
+  if (thd->locked_tables_mode || thd->in_active_multi_stmt_transaction())
   {
     my_error(ER_LOCK_OR_ACTIVE_TRANSACTION, MYF(0));
     return true;
@@ -2006,15 +2006,20 @@ static Sys_var_ulong Sys_thread_pool_size(
        VALID_RANGE(1, 16384), DEFAULT(20), BLOCK_SIZE(0));
 #endif
 
-//  Can't change the 'next' tx_isolation if we are already in a transaction
+/**
+  Can't change the 'next' tx_isolation if we are already in a
+  transaction.
+*/
+
 static bool check_tx_isolation(sys_var *self, THD *thd, set_var *var)
 {
-  if (var->type == OPT_DEFAULT && (thd->server_status & SERVER_STATUS_IN_TRANS))
+  if (var->type == OPT_DEFAULT && thd->in_active_multi_stmt_transaction())
   {
+    DBUG_ASSERT(thd->in_multi_stmt_transaction_mode());
     my_error(ER_CANT_CHANGE_TX_ISOLATION, MYF(0));
-    return true;
+    return TRUE;
   }
-  return false;
+  return FALSE;
 }
 
 /*
@@ -2027,6 +2032,7 @@ static bool fix_tx_isolation(sys_var *self, THD *thd, enum_var_type type)
     thd->session_tx_isolation= (enum_tx_isolation)thd->variables.tx_isolation;
   return false;
 }
+
 // NO_CMD_LINE - different name of the option
 static Sys_var_enum Sys_tx_isolation(
        "tx_isolation", "Default transaction isolation level",
diff --git a/sql/transaction.cc b/sql/transaction.cc
index ff4eabc2b0f..5047de1ccdc 100644
--- a/sql/transaction.cc
+++ b/sql/transaction.cc
@@ -169,7 +169,7 @@ bool trans_commit_implicit(THD *thd)
   if (trans_check(thd))
     DBUG_RETURN(TRUE);
 
-  if (thd->in_multi_stmt_transaction() ||
+  if (thd->in_multi_stmt_transaction_mode() ||
       (thd->variables.option_bits & OPTION_TABLE_LOCK))
   {
     /* Safety if one did "drop table" on locked tables */
@@ -305,7 +305,7 @@ bool trans_savepoint(THD *thd, LEX_STRING name)
   SAVEPOINT **sv, *newsv;
   DBUG_ENTER("trans_savepoint");
 
-  if (!(thd->in_multi_stmt_transaction() || thd->in_sub_stmt) ||
+  if (!(thd->in_multi_stmt_transaction_mode() || thd->in_sub_stmt) ||
       !opt_using_transactions)
     DBUG_RETURN(FALSE);
 
@@ -467,7 +467,7 @@ bool trans_xa_start(THD *thd)
     my_error(ER_XAER_INVAL, MYF(0));
   else if (xa_state != XA_NOTR)
     my_error(ER_XAER_RMFAIL, MYF(0), xa_state_names[xa_state]);
-  else if (thd->locked_tables_mode || thd->active_transaction())
+  else if (thd->locked_tables_mode || thd->in_active_multi_stmt_transaction())
     my_error(ER_XAER_OUTSIDE, MYF(0));
   else if (xid_cache_search(thd->lex->xid))
     my_error(ER_XAER_DUPID, MYF(0));