Merge mysql_cab_desk.:C:/source/c++/mysql-5.1

into  mysql_cab_desk.:C:/source/c++/mysql-5.1-new-rpl-merge

54 files changed, 1600 insertions, 1229 deletions

diff --git a/sql/CMakeLists.txt b/sql/CMakeLists.txt
index a8aa7d70586..c792e10c3af 100644
--- a/sql/CMakeLists.txt
+++ b/sql/CMakeLists.txt
@@ -143,6 +143,10 @@ ADD_CUSTOM_COMMAND(
 )
 ADD_DEPENDENCIES(mysqld${MYSQLD_EXE_SUFFIX} gen_lex_hash)
 
+# Remove the auto-generated files as part of 'Clean Solution'
+SET_DIRECTORY_PROPERTIES(PROPERTIES ADDITIONAL_MAKE_CLEAN_FILES 
+                         "lex_hash.h;message.rc;message.h;sql_yacc.h;sql_yacc.cc")
+
 ADD_LIBRARY(udf_example MODULE udf_example.c udf_example.def)
 ADD_DEPENDENCIES(udf_example strings)
 TARGET_LINK_LIBRARIES(udf_example wsock32)
diff --git a/sql/event_data_objects.cc b/sql/event_data_objects.cc
index 7b58c10079a..144a87e13f6 100644
--- a/sql/event_data_objects.cc
+++ b/sql/event_data_objects.cc
@@ -979,17 +979,18 @@ Event_queue_element::load_from_row(THD *thd, TABLE *table)
     DBUG_RETURN(TRUE);
 
   starts_null= table->field[ET_FIELD_STARTS]->is_null();
+  my_bool not_used= FALSE;
   if (!starts_null)
   {
     table->field[ET_FIELD_STARTS]->get_date(&time, TIME_NO_ZERO_DATE);
-    starts= sec_since_epoch_TIME(&time);
+    starts= my_tz_OFFSET0->TIME_to_gmt_sec(&time,&not_used);
   }
 
   ends_null= table->field[ET_FIELD_ENDS]->is_null();
   if (!ends_null)
   {
     table->field[ET_FIELD_ENDS]->get_date(&time, TIME_NO_ZERO_DATE);
-    ends= sec_since_epoch_TIME(&time);
+    ends= my_tz_OFFSET0->TIME_to_gmt_sec(&time,&not_used);
   }
 
   if (!table->field[ET_FIELD_INTERVAL_EXPR]->is_null())
@@ -1007,7 +1008,7 @@ Event_queue_element::load_from_row(THD *thd, TABLE *table)
     if (table->field[ET_FIELD_EXECUTE_AT]->get_date(&time,
                                                     TIME_NO_ZERO_DATE))
       DBUG_RETURN(TRUE);
-    execute_at= sec_since_epoch_TIME(&time);
+    execute_at= my_tz_OFFSET0->TIME_to_gmt_sec(&time,&not_used);
   }
 
   /*
@@ -1039,7 +1040,7 @@ Event_queue_element::load_from_row(THD *thd, TABLE *table)
   {
     table->field[ET_FIELD_LAST_EXECUTED]->get_date(&time,
                                                    TIME_NO_ZERO_DATE);
-    last_executed= sec_since_epoch_TIME(&time);
+    last_executed= my_tz_OFFSET0->TIME_to_gmt_sec(&time,&not_used);
   }
   last_executed_changed= FALSE;
 
diff --git a/sql/event_db_repository.cc b/sql/event_db_repository.cc
index a16a04739e2..b7bcb6344fd 100644
--- a/sql/event_db_repository.cc
+++ b/sql/event_db_repository.cc
@@ -250,7 +250,7 @@ mysql_event_fill_row(THD *thd,
     if (!et->starts_null)
     {
       MYSQL_TIME time;
-      my_tz_UTC->gmt_sec_to_TIME(&time, et->starts);
+      my_tz_OFFSET0->gmt_sec_to_TIME(&time, et->starts);
 
       fields[ET_FIELD_STARTS]->set_notnull();
       fields[ET_FIELD_STARTS]->store_time(&time, MYSQL_TIMESTAMP_DATETIME);
@@ -259,7 +259,7 @@ mysql_event_fill_row(THD *thd,
     if (!et->ends_null)
     {
       MYSQL_TIME time;
-      my_tz_UTC->gmt_sec_to_TIME(&time, et->ends);
+      my_tz_OFFSET0->gmt_sec_to_TIME(&time, et->ends);
 
       fields[ET_FIELD_ENDS]->set_notnull();
       fields[ET_FIELD_ENDS]->store_time(&time, MYSQL_TIMESTAMP_DATETIME);
@@ -278,7 +278,7 @@ mysql_event_fill_row(THD *thd,
     fields[ET_FIELD_ENDS]->set_null();
 
     MYSQL_TIME time;
-    my_tz_UTC->gmt_sec_to_TIME(&time, et->execute_at);
+    my_tz_OFFSET0->gmt_sec_to_TIME(&time, et->execute_at);
 
     fields[ET_FIELD_EXECUTE_AT]->set_notnull();
     fields[ET_FIELD_EXECUTE_AT]->
@@ -1004,7 +1004,7 @@ update_timing_fields_for_event(THD *thd,
   if (update_last_executed)
   {
     MYSQL_TIME time;
-    my_tz_UTC->gmt_sec_to_TIME(&time, last_executed);
+    my_tz_OFFSET0->gmt_sec_to_TIME(&time, last_executed);
 
     fields[ET_FIELD_LAST_EXECUTED]->set_notnull();
     fields[ET_FIELD_LAST_EXECUTED]->store_time(&time,
diff --git a/sql/event_queue.cc b/sql/event_queue.cc
index 634cc764d74..04449dd48a1 100644
--- a/sql/event_queue.cc
+++ b/sql/event_queue.cc
@@ -740,7 +740,7 @@ Event_queue::dump_internal_status()
   printf("WOC             : %s\n", waiting_on_cond? "YES":"NO");
 
   MYSQL_TIME time;
-  my_tz_UTC->gmt_sec_to_TIME(&time, next_activation_at);
+  my_tz_OFFSET0->gmt_sec_to_TIME(&time, next_activation_at);
   if (time.year != 1970)
     printf("Next activation : %04d-%02d-%02d %02d:%02d:%02d\n",
            time.year, time.month, time.day, time.hour, time.minute, time.second);
diff --git a/sql/field.cc b/sql/field.cc
index 49433deca74..48339f2415f 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -8085,8 +8085,11 @@ int Field_enum::store(longlong nr, bool unsigned_val)
   if ((ulonglong) nr > typelib->count || nr == 0)
   {
     set_warning(MYSQL_ERROR::WARN_LEVEL_WARN, WARN_DATA_TRUNCATED, 1);
-    nr=0;
-    error=1;
+    if (nr != 0 || table->in_use->count_cuted_fields)
+    {
+      nr= 0;
+      error= 1;
+    }
   }
   store_type((ulonglong) (uint) nr);
   return error;
diff --git a/sql/ha_ndbcluster.cc b/sql/ha_ndbcluster.cc
index 0b8ffd4a2fb..48712600e79 100644
--- a/sql/ha_ndbcluster.cc
+++ b/sql/ha_ndbcluster.cc
@@ -4366,7 +4366,7 @@ int ha_ndbcluster::external_lock(THD *thd, int lock_type)
     {
       m_transaction_on= FALSE;
       /* Would be simpler if has_transactions() didn't always say "yes" */
-      thd->no_trans_update.all= thd->no_trans_update.stmt= TRUE;
+      thd->transaction.all.modified_non_trans_table= thd->transaction.stmt.modified_non_trans_table= TRUE;
     }
     else if (!thd->transaction.on)
       m_transaction_on= FALSE;
@@ -4383,7 +4383,10 @@ int ha_ndbcluster::external_lock(THD *thd, int lock_type)
 
         trans= ndb->startTransaction();
         if (trans == NULL)
+        {
+          thd_ndb->lock_count= 0;
           ERR_RETURN(ndb->getNdbError());
+        }
         thd_ndb->init_open_tables();
         thd_ndb->stmt= trans;
 	thd_ndb->query_state&= NDB_QUERY_NORMAL;
@@ -4409,7 +4412,10 @@ int ha_ndbcluster::external_lock(THD *thd, int lock_type)
           
           trans= ndb->startTransaction();
           if (trans == NULL)
+          {
+            thd_ndb->lock_count= 0;
             ERR_RETURN(ndb->getNdbError());
+          }
           thd_ndb->init_open_tables();
           thd_ndb->all= trans; 
 	  thd_ndb->query_state&= NDB_QUERY_NORMAL;
diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index e4924e8e8f2..d46b3a3bb08 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -1142,6 +1142,12 @@ int ha_partition::prepare_new_partition(TABLE *table,
   create_flag= TRUE;
   if ((error= file->ha_open(table, part_name, m_mode, m_open_test_lock)))
     goto error;
+  /*
+    Note: if you plan to add another call that may return failure,
+    better to do it before external_lock() as cleanup_new_partition()
+    assumes that external_lock() is last call that may fail here.
+    Otherwise see description for cleanup_new_partition().
+  */
   if ((error= file->external_lock(current_thd, m_lock_type)))
     goto error;
 
@@ -1164,6 +1170,14 @@ error:
     NONE
 
   DESCRIPTION
+    This function is called immediately after prepare_new_partition() in
+    case the latter fails.
+
+    In prepare_new_partition() last call that may return failure is
+    external_lock(). That means if prepare_new_partition() fails,
+    partition does not have external lock. Thus no need to call
+    external_lock(F_UNLCK) here.
+
   TODO:
     We must ensure that in the case that we get an error during the process
     that we call external_lock with F_UNLCK, close the table and delete the
@@ -1182,7 +1196,6 @@ void ha_partition::cleanup_new_partition(uint part_count)
     m_file= m_added_file;
     m_added_file= NULL;
 
-    external_lock(current_thd, F_UNLCK);
     /* delete_table also needed, a bit more complex */
     close();
 
diff --git a/sql/handler.cc b/sql/handler.cc
index 2d836b30862..e828c9cdde8 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -851,6 +851,9 @@ int ha_rollback_trans(THD *thd, bool all)
     }
   }
 #endif /* USING_TRANSACTIONS */
+  if (all)
+    thd->transaction_rollback_request= FALSE;
+
   /*
     If a non-transactional table was updated, warn; don't warn if this is a
     slave thread (because when a slave thread executes a ROLLBACK, it has
@@ -860,7 +863,7 @@ int ha_rollback_trans(THD *thd, bool all)
     the error log; but we don't want users to wonder why they have this
     message in the error log, so we don't send it.
   */
-  if (is_real_trans && thd->no_trans_update.all &&
+  if (is_real_trans && thd->transaction.all.modified_non_trans_table &&
       !thd->slave_thread && thd->killed != THD::KILL_CONNECTION)
     push_warning(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
                  ER_WARNING_NOT_COMPLETE_ROLLBACK,
@@ -887,6 +890,8 @@ int ha_autocommit_or_rollback(THD *thd, int error)
       if (ha_commit_stmt(thd))
 	error=1;
     }
+    else if (thd->transaction_rollback_request && !thd->in_sub_stmt)
+      (void) ha_rollback(thd);
     else
       (void) ha_rollback_stmt(thd);
 
@@ -1516,34 +1521,6 @@ THD *handler::ha_thd(void) const
 }
 
 
-bool handler::check_if_log_table_locking_is_allowed(uint sql_command,
-                                                    ulong type, TABLE *table)
-{
-  /*
-    Deny locking of the log tables, which is incompatible with
-    concurrent insert. The routine is not called if the table is
-    being locked from a logger THD (general_log_thd or slow_log_thd)
-    or from a privileged thread (see log.cc for details)
-  */
-  if (table->s->log_table &&
-      sql_command != SQLCOM_TRUNCATE &&
-      sql_command != SQLCOM_ALTER_TABLE &&
-      !(sql_command == SQLCOM_FLUSH &&
-        type & REFRESH_LOG) &&
-      (table->reginfo.lock_type >= TL_READ_NO_INSERT))
-  {
-    /*
-      The check  >= TL_READ_NO_INSERT denies all write locks
-      plus the only read lock (TL_READ_NO_INSERT itself)
-    */
-    table->reginfo.lock_type == TL_READ_NO_INSERT ?
-      my_error(ER_CANT_READ_LOCK_LOG_TABLE, MYF(0)) :
-        my_error(ER_CANT_WRITE_LOCK_LOG_TABLE, MYF(0));
-    return FALSE;
-  }
-  return TRUE;
-}
-
 /** @brief
   Open database-handler.
 
@@ -2355,7 +2332,13 @@ static bool update_frm_version(TABLE *table)
   int result= 1;
   DBUG_ENTER("update_frm_version");
 
-  if (table->s->mysql_version != MYSQL_VERSION_ID)
+  /*
+    No need to update frm version in case table was created or checked
+    by server with the same version. This also ensures that we do not
+    update frm version for temporary tables as this code doesn't support
+    temporary tables.
+  */
+  if (table->s->mysql_version == MYSQL_VERSION_ID)
     DBUG_RETURN(0);
 
   strxmov(path, table->s->normalized_path.str, reg_ext, NullS);
@@ -3681,6 +3664,7 @@ int handler::ha_write_row(uchar *buf)
   return 0;
 }
 
+
 int handler::ha_update_row(const uchar *old_data, uchar *new_data)
 {
   int error;
diff --git a/sql/handler.h b/sql/handler.h
index f45b28c55f5..ddea6c3a34c 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -728,6 +728,35 @@ typedef struct st_thd_trans
   bool        no_2pc;
   /* storage engines that registered themselves for this transaction */
   handlerton *ht[MAX_HA];
+  /* 
+    The purpose of this flag is to keep track of non-transactional
+    tables that were modified in scope of:
+    - transaction, when the variable is a member of
+    THD::transaction.all
+    - top-level statement or sub-statement, when the variable is a
+    member of THD::transaction.stmt
+    This member has the following life cycle:
+    * stmt.modified_non_trans_table is used to keep track of
+    modified non-transactional tables of top-level statements. At
+    the end of the previous statement and at the beginning of the session,
+    it is reset to FALSE.  If such functions
+    as mysql_insert, mysql_update, mysql_delete etc modify a
+    non-transactional table, they set this flag to TRUE.  At the
+    end of the statement, the value of stmt.modified_non_trans_table 
+    is merged with all.modified_non_trans_table and gets reset.
+    * all.modified_non_trans_table is reset at the end of transaction
+    
+    * Since we do not have a dedicated context for execution of a
+    sub-statement, to keep track of non-transactional changes in a
+    sub-statement, we re-use stmt.modified_non_trans_table. 
+    At entrance into a sub-statement, a copy of the value of
+    stmt.modified_non_trans_table (containing the changes of the
+    outer statement) is saved on stack. Then 
+    stmt.modified_non_trans_table is reset to FALSE and the
+    substatement is executed. Then the new value is merged with the
+    saved value.
+  */
+  bool modified_non_trans_table;
 } THD_TRANS;
 
 enum enum_tx_isolation { ISO_READ_UNCOMMITTED, ISO_READ_COMMITTED,
@@ -992,6 +1021,7 @@ public:
   uint ref_length;
   FT_INFO *ft_handler;
   enum {NONE=0, INDEX, RND} inited;
+  bool locked;
   bool implicit_emptied;                /* Can be !=0 only if HEAP */
   const COND *pushed_cond;
   /*
@@ -1022,11 +1052,13 @@ public:
     estimation_rows_to_insert(0), ht(ht_arg),
     ref(0), key_used_on_scan(MAX_KEY), active_index(MAX_KEY),
     ref_length(sizeof(my_off_t)),
-    ft_handler(0), inited(NONE), implicit_emptied(0),
+    ft_handler(0), inited(NONE),
+    locked(FALSE), implicit_emptied(0),
     pushed_cond(0), next_insert_id(0), insert_id_for_cur_row(0)
     {}
   virtual ~handler(void)
   {
+    DBUG_ASSERT(locked == FALSE);
     /* TODO: DBUG_ASSERT(inited == NONE); */
   }
   virtual handler *clone(MEM_ROOT *mem_root);
@@ -1035,44 +1067,6 @@ public:
   {
     cached_table_flags= table_flags();
   }
-  /*
-    Check whether a handler allows to lock the table.
-
-    SYNOPSIS
-      check_if_locking_is_allowed()
-        thd     Handler of the thread, trying to lock the table
-        table   Table handler to check
-        count   Total number of tables to be locked
-        current Index of the current table in the list of the tables
-                to be locked.
-        system_count Pointer to the counter of system tables seen thus
-                     far.
-        called_by_privileged_thread TRUE if called from a logger THD
-                                    (general_log_thd or slow_log_thd)
-                                    or by a privileged thread, which
-                                    has the right to lock log tables.
-
-    DESCRIPTION
-      Check whether a handler allows to lock the table. For instance,
-      MyISAM does not allow to lock mysql.proc along with other tables.
-      This limitation stems from the fact that MyISAM does not support
-      row-level locking and we have to add this limitation to avoid
-      deadlocks.
-
-    RETURN
-      TRUE      Locking is allowed
-      FALSE     Locking is not allowed. The error was thrown.
-  */
-  virtual bool check_if_locking_is_allowed(uint sql_command,
-                                           ulong type, TABLE *table,
-                                           uint count, uint current,
-                                           uint *system_count,
-                                           bool called_by_privileged_thread)
-  {
-    return TRUE;
-  }
-  bool check_if_log_table_locking_is_allowed(uint sql_command,
-                                             ulong type, TABLE *table);
   int ha_open(TABLE *table, const char *name, int mode, int test_if_locked);
   void adjust_next_insert_id_after_explicit_value(ulonglong nr);
   int update_auto_increment();
@@ -1629,8 +1623,10 @@ public:
 
   /* lock_count() can be more than one if the table is a MERGE */
   virtual uint lock_count(void) const { return 1; }
-  /*
-    NOTE that one can NOT rely on table->in_use in store_lock().  It may
+  /**
+    Is not invoked for non-transactional temporary tables.
+
+    @note that one can NOT rely on table->in_use in store_lock().  It may
     refer to a different thread if called from mysql_lock_abort_for_thread().
   */
   virtual THR_LOCK_DATA **store_lock(THD *thd,
@@ -1761,6 +1757,29 @@ private:
     overridden by the storage engine class. To call these methods, use
     the corresponding 'ha_*' method above.
   */
+
+  /**
+    Is not invoked for non-transactional temporary tables.
+
+    Tells the storage engine that we intend to read or write data
+    from the table. This call is prefixed with a call to handler::store_lock()
+    and is invoked only for those handler instances that stored the lock.
+
+    Calls to rnd_init/index_init are prefixed with this call. When table
+    IO is complete, we call external_lock(F_UNLCK).
+    A storage engine writer should expect that each call to
+    ::external_lock(F_[RD|WR]LOCK is followed by a call to
+    ::external_lock(F_UNLCK). If it is not, it is a bug in MySQL.
+
+    The name and signature originate from the first implementation
+    in MyISAM, which would call fcntl to set/clear an advisory
+    lock on the data file in this method.
+
+    @param   lock_type    F_RDLCK, F_WRLCK, F_UNLCK
+
+    @return  non-0 in case of failure, 0 in case of success.
+    When lock_type is F_UNLCK, the return value is ignored.
+  */
   virtual int external_lock(THD *thd __attribute__((unused)),
                             int lock_type __attribute__((unused)))
   {
diff --git a/sql/item.cc b/sql/item.cc
index 711a21ecbec..593915ef172 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -334,6 +334,37 @@ int Item::save_date_in_field(Field *field)
 }
 
 
+/*
+  Store the string value in field directly
+
+  SYNOPSIS
+    Item::save_str_value_in_field()
+    field   a pointer to field where to store
+    result  the pointer to the string value to be stored
+
+  DESCRIPTION
+    The method is used by Item_*::save_in_field implementations
+    when we don't need to calculate the value to store
+    See Item_string::save_in_field() implementation for example
+
+  IMPLEMENTATION
+    Check if the Item is null and stores the NULL or the
+    result value in the field accordingly.
+
+  RETURN
+    Nonzero value if error
+*/
+
+int Item::save_str_value_in_field(Field *field, String *result)
+{
+  if (null_value)
+    return set_field_to_null(field);
+  field->set_notnull();
+  return field->store(result->ptr(), result->length(),
+		      collation.collation);
+}
+
+
 Item::Item():
   rsize(0), name(0), orig_name(0), name_length(0), fixed(0),
   is_autogenerated_name(TRUE),
@@ -1051,9 +1082,9 @@ bool Item_sp_variable::is_null()
 Item_splocal::Item_splocal(const LEX_STRING &sp_var_name,
                            uint sp_var_idx,
                            enum_field_types sp_var_type,
-                           uint pos_in_q)
+                           uint pos_in_q, uint len_in_q)
   :Item_sp_variable(sp_var_name.str, sp_var_name.length),
-   m_var_idx(sp_var_idx), pos_in_query(pos_in_q)
+   m_var_idx(sp_var_idx), pos_in_query(pos_in_q), len_in_query(len_in_q)
 {
   maybe_null= TRUE;
 
@@ -3046,16 +3077,6 @@ my_decimal *Item_copy_string::val_decimal(my_decimal *decimal_value)
 }
 
 
-
-int Item_copy_string::save_in_field(Field *field, bool no_conversions)
-{
-  if (null_value)
-    return set_field_to_null(field);
-  field->set_notnull();
-  return field->store(str_value.ptr(),str_value.length(),
-		      collation.collation);
-}
-
 /*
   Functions to convert item to field (for send_fields)
 */
@@ -4508,6 +4529,12 @@ int Item_null::save_safe_in_field(Field *field)
 }
 
 
+/*
+  This implementation can lose str_value content, so if the
+  Item uses str_value to store something, it should
+  reimplement it's ::save_in_field() as Item_string, for example, does
+*/
+
 int Item::save_in_field(Field *field, bool no_conversions)
 {
   int error;
@@ -4565,10 +4592,7 @@ int Item_string::save_in_field(Field *field, bool no_conversions)
 {
   String *result;
   result=val_str(&str_value);
-  if (null_value)
-    return set_field_to_null(field);
-  field->set_notnull();
-  return field->store(result->ptr(),result->length(),collation.collation);
+  return save_str_value_in_field(field, result);
 }
 
 
diff --git a/sql/item.h b/sql/item.h
index 432da6c3a1c..ab39505bf5f 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -653,6 +653,7 @@ public:
 
   int save_time_in_field(Field *field);
   int save_date_in_field(Field *field);
+  int save_str_value_in_field(Field *field, String *result);
 
   virtual Field *get_tmp_table_field() { return 0; }
   /* This is also used to create fields in CREATE ... SELECT: */
@@ -1053,9 +1054,18 @@ public:
     SP variable in query text.
   */
   uint pos_in_query;
+  /*
+    Byte length of SP variable name in the statement (see pos_in_query).
+    The value of this field may differ from the name_length value because
+    name_length contains byte length of UTF8-encoded item name, but
+    the query string (see sp_instr_stmt::m_query) is currently stored with
+    a charset from the SET NAMES statement.
+  */
+  uint len_in_query;
 
   Item_splocal(const LEX_STRING &sp_var_name, uint sp_var_idx,
-               enum_field_types sp_var_type, uint pos_in_q= 0);
+               enum_field_types sp_var_type,
+               uint pos_in_q= 0, uint len_in_q= 0);
 
   bool is_splocal() { return 1; } /* Needed for error checking */
 
@@ -2293,7 +2303,10 @@ public:
   my_decimal *val_decimal(my_decimal *);
   void make_field(Send_field *field) { item->make_field(field); }
   void copy();
-  int save_in_field(Field *field, bool no_conversions);
+  int save_in_field(Field *field, bool no_conversions)
+  {
+    return save_str_value_in_field(field, &str_value);
+  }
   table_map used_tables() const { return (table_map) 1L; }
   bool const_item() const { return 0; }
   bool is_null() { return null_value; }
diff --git a/sql/item_func.cc b/sql/item_func.cc
index f14091b4592..9a26169ad30 100644
--- a/sql/item_func.cc
+++ b/sql/item_func.cc
@@ -649,16 +649,8 @@ bool Item_func_connection_id::fix_fields(THD *thd, Item **ref)
 {
   if (Item_int_func::fix_fields(thd, ref))
     return TRUE;
-
-  /*
-    To replicate CONNECTION_ID() properly we should use
-    pseudo_thread_id on slave, which contains the value of thread_id
-    on master.
-  */
-  value= ((thd->slave_thread) ?
-          thd->variables.pseudo_thread_id :
-          thd->thread_id);
-
+  thd->thread_specific_used= TRUE;
+  value= thd->variables.pseudo_thread_id;
   return FALSE;
 }
 
@@ -5599,10 +5591,20 @@ Item_func_sp::fix_fields(THD *thd, Item **ref)
     
 #endif /* ! NO_EMBEDDED_ACCESS_CHECKS */
   }
+  if (!m_sp->m_chistics->detistic)
+   used_tables_cache |= RAND_TABLE_BIT;
   DBUG_RETURN(res);
 }
 
 
+void Item_func_sp::update_used_tables()
+{
+  Item_func::update_used_tables();
+  if (!m_sp->m_chistics->detistic)
+   used_tables_cache |= RAND_TABLE_BIT;
+}
+
+
 /*
   uuid_short handling.
 
diff --git a/sql/item_func.h b/sql/item_func.h
index 568effb2f63..ea22e35773d 100644
--- a/sql/item_func.h
+++ b/sql/item_func.h
@@ -1472,7 +1472,7 @@ public:
   virtual ~Item_func_sp()
   {}
 
-  table_map used_tables() const { return RAND_TABLE_BIT; }
+  void update_used_tables();
 
   void cleanup();
 
diff --git a/sql/item_strfunc.h b/sql/item_strfunc.h
index 6d2d9c199c9..ea9517976a8 100644
--- a/sql/item_strfunc.h
+++ b/sql/item_strfunc.h
@@ -427,6 +427,10 @@ public:
   }
   const char *func_name() const { return "user"; }
   const char *fully_qualified_func_name() const { return "user()"; }
+  int save_in_field(Field *field, bool no_conversions)
+  {
+    return save_str_value_in_field(field, &str_value);
+  }
 };
 
 
diff --git a/sql/lock.cc b/sql/lock.cc
index 4260a031def..08ff90ce983 100644
--- a/sql/lock.cc
+++ b/sql/lock.cc
@@ -60,6 +60,11 @@
   - When calling UNLOCK TABLES we call mysql_unlock_tables() for all
     tables used in LOCK TABLES
 
+  If table_handler->external_lock(thd, locktype) fails, we call
+  table_handler->external_lock(thd, F_UNLCK) for each table that was locked,
+  excluding one that caused failure. That means handler must cleanup itself
+  in case external_lock() fails.
+
 TODO:
   Change to use my_malloc() ONLY when using LOCK TABLES command or when
   we are forced to use mysql_lock_merge.
@@ -111,16 +116,90 @@ static void print_lock_error(int error, const char *);
 static int thr_lock_errno_to_mysql[]=
 { 0, 1, ER_LOCK_WAIT_TIMEOUT, ER_LOCK_DEADLOCK };
 
+/**
+  Perform semantic checks for mysql_lock_tables.
+  @param thd The current thread
+  @param tables The tables to lock
+  @param count The number of tables to lock
+  @param flags Lock flags
+  @return 0 if all the check passed, non zero if a check failed.
+*/
+int mysql_lock_tables_check(THD *thd, TABLE **tables, uint count, uint flags)
+{
+  bool log_table_write_query;
+  uint system_count;
+  uint i;
+
+  DBUG_ENTER("mysql_lock_tables_check");
+
+  system_count= 0;
+  log_table_write_query= (is_log_table_write_query(thd->lex->sql_command)
+                         || ((flags & MYSQL_LOCK_PERF_SCHEMA) != 0));
+
+  for (i=0 ; i<count; i++)
+  {
+    TABLE *t= tables[i];
+
+    /* Protect against 'fake' partially initialized TABLE_SHARE */
+    DBUG_ASSERT(t->s->table_category != TABLE_UNKNOWN_CATEGORY);
+
+    /*
+      Table I/O to performance schema tables is performed
+      only internally by the server implementation.
+      When a user is requesting a lock, the following
+      constraints are enforced:
+    */
+    if (t->s->require_write_privileges() &&
+        ! log_table_write_query)
+    {
+      /*
+        A user should not be able to prevent writes,
+        or hold any type of lock in a session,
+        since this would be a DOS attack.
+      */
+      if ((t->reginfo.lock_type >= TL_READ_NO_INSERT)
+          || (thd->lex->sql_command == SQLCOM_LOCK_TABLES))
+      {
+        my_error(ER_CANT_LOCK_LOG_TABLE, MYF(0));
+        DBUG_RETURN(1);
+      }
+    }
+
+    if ((t->s->table_category == TABLE_CATEGORY_SYSTEM) &&
+        (t->reginfo.lock_type >= TL_WRITE_ALLOW_WRITE))
+    {
+      system_count++;
+    }
+  }
+
+  /*
+    Locking of system tables is restricted:
+    locking a mix of system and non-system tables in the same lock
+    is prohibited, to prevent contention.
+  */
+  if ((system_count > 0) && (system_count < count))
+  {
+    my_error(ER_WRONG_LOCK_OF_SYSTEM_TABLE, MYF(0));
+    DBUG_RETURN(1);
+  }
+
+  DBUG_RETURN(0);
+}
+
 MYSQL_LOCK *mysql_lock_tables(THD *thd, TABLE **tables, uint count,
                               uint flags, bool *need_reopen)
 {
   MYSQL_LOCK *sql_lock;
   TABLE *write_lock_used;
   int rc;
+
   DBUG_ENTER("mysql_lock_tables");
 
   *need_reopen= FALSE;
 
+  if (mysql_lock_tables_check(thd, tables, count, flags))
+    DBUG_RETURN (NULL);
+
   for (;;)
   {
     if (! (sql_lock= get_lock_data(thd, tables, count, GET_LOCK_STORE_LOCKS,
@@ -170,7 +249,8 @@ MYSQL_LOCK *mysql_lock_tables(THD *thd, TABLE **tables, uint count,
 
     thd->proc_info="System lock";
     DBUG_PRINT("info", ("thd->proc_info %s", thd->proc_info));
-    if (lock_external(thd, tables, count))
+    if (sql_lock->table_count && lock_external(thd, sql_lock->table,
+                                               sql_lock->table_count))
     {
       /* Clear the lock type of all lock data to avoid reusage. */
       reset_lock_data(sql_lock);
@@ -266,11 +346,13 @@ static int lock_external(THD *thd, TABLE **tables, uint count)
 	((*tables)->reginfo.lock_type >= TL_READ &&
 	 (*tables)->reginfo.lock_type <= TL_READ_NO_INSERT))
       lock_type=F_RDLCK;
+
     if ((error=(*tables)->file->ha_external_lock(thd,lock_type)))
     {
       print_lock_error(error, (*tables)->file->table_type());
-      for (; i-- ; tables--)
+      while (--i)
       {
+        tables--;
 	(*tables)->file->ha_external_lock(thd, F_UNLCK);
 	(*tables)->current_lock=F_UNLCK;
       }
@@ -373,10 +455,28 @@ void mysql_unlock_read_tables(THD *thd, MYSQL_LOCK *sql_lock)
 }
 
 
+/**
+  Try to find the table in the list of locked tables.
+  In case of success, unlock the table and remove it from this list.
+
+  @note This function has a legacy side effect: the table is
+  unlocked even if it is not found in the locked list.
+  It's not clear if this side effect is intentional or still
+  desirable. It might lead to unmatched calls to
+  unlock_external(). Moreover, a discrepancy can be left
+  unnoticed by the storage engine, because in
+  unlock_external() we call handler::external_lock(F_UNLCK) only
+  if table->current_lock is not F_UNLCK.
+
+  @param  always_unlock   specify explicitly if the legacy side
+                          effect is desired.
+*/
 
-void mysql_lock_remove(THD *thd, MYSQL_LOCK *locked,TABLE *table)
+void mysql_lock_remove(THD *thd, MYSQL_LOCK *locked,TABLE *table,
+                       bool always_unlock)
 {
-  mysql_unlock_some_tables(thd, &table,1);
+  if (always_unlock == TRUE)
+    mysql_unlock_some_tables(thd, &table, /* table count */ 1);
   if (locked)
   {
     reg1 uint i;
@@ -390,6 +490,10 @@ void mysql_lock_remove(THD *thd, MYSQL_LOCK *locked,TABLE *table)
 
         DBUG_ASSERT(table->lock_position == i);
 
+        /* Unlock if not yet unlocked */
+        if (always_unlock == FALSE)
+          mysql_unlock_some_tables(thd, &table, /* table count */ 1);
+
         /* Decrement table_count in advance, making below expressions easier */
         old_tables= --locked->table_count;
 
@@ -439,7 +543,8 @@ void mysql_lock_downgrade_write(THD *thd, TABLE *table,
 {
   MYSQL_LOCK *locked;
   TABLE *write_lock_used;
-  if ((locked = get_lock_data(thd,&table,1,1,&write_lock_used)))
+  if ((locked = get_lock_data(thd, &table, 1, GET_LOCK_UNLOCK,
+                              &write_lock_used)))
   {
     for (uint i=0; i < locked->lock_count; i++)
       thr_downgrade_write_lock(locked->locks[i], new_lock_type);
@@ -698,25 +803,19 @@ static MYSQL_LOCK *get_lock_data(THD *thd, TABLE **table_ptr, uint count,
   TABLE **to, **table_buf;
   DBUG_ENTER("get_lock_data");
 
+  DBUG_ASSERT((flags == GET_LOCK_UNLOCK) || (flags == GET_LOCK_STORE_LOCKS));
+
   DBUG_PRINT("info", ("count %d", count));
   *write_lock_used=0;
-  uint system_count= 0;
   for (i=tables=lock_count=0 ; i < count ; i++)
   {
-    if (table_ptr[i]->s->tmp_table != NON_TRANSACTIONAL_TMP_TABLE)
+    TABLE *t= table_ptr[i];
+
+    if (t->s->tmp_table != NON_TRANSACTIONAL_TMP_TABLE)
     {
-      tables+=table_ptr[i]->file->lock_count();
+      tables+= t->file->lock_count();
       lock_count++;
     }
-    /*
-      Check if we can lock the table. For some tables we cannot do that
-      beacause of handler-specific locking issues.
-    */
-    if (!table_ptr[i]-> file->
-          check_if_locking_is_allowed(thd->lex->sql_command, thd->lex->type,
-                                      table_ptr[i], count, i, &system_count,
-                                      logger.is_privileged_thread(thd)))
-      DBUG_RETURN(0);
   }
 
   /*
diff --git a/sql/log.cc b/sql/log.cc
index 0bf77d68410..00ef726f037 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -57,6 +57,35 @@ static int binlog_commit(handlerton *hton, THD *thd, bool all);
 static int binlog_rollback(handlerton *hton, THD *thd, bool all);
 static int binlog_prepare(handlerton *hton, THD *thd, bool all);
 
+/**
+  Silence all errors and warnings reported when performing a write
+  to a log table.
+  Errors and warnings are not reported to the client or SQL exception
+  handlers, so that the presence of logging does not interfere and affect
+  the logic of an application.
+*/
+class Silence_log_table_errors : public Internal_error_handler
+{
+public:
+  Silence_log_table_errors()
+  {}
+
+  virtual ~Silence_log_table_errors() {}
+
+  virtual bool handle_error(uint sql_errno,
+                            MYSQL_ERROR::enum_warning_level level,
+                            THD *thd);
+};
+
+bool
+Silence_log_table_errors::handle_error(uint /* sql_errno */,
+                                       MYSQL_ERROR::enum_warning_level /* level */,
+                                       THD * /* thd */)
+{
+  return TRUE;
+}
+
+
 sql_print_message_func sql_print_message_handlers[3] =
 {
   sql_print_information,
@@ -187,6 +216,19 @@ public:
 
 handlerton *binlog_hton;
 
+bool LOGGER::is_log_table_enabled(uint log_table_type)
+{
+  switch (log_table_type) {
+  case QUERY_LOG_SLOW:
+    return (table_log_handler != NULL) && opt_slow_log;
+  case QUERY_LOG_GENERAL:
+    return (table_log_handler != NULL) && opt_log ;
+  default:
+    DBUG_ASSERT(0);
+    return FALSE;                             /* make compiler happy */
+  }
+}
+
 
 /* Check if a given table is opened log table */
 int check_if_log_table(uint db_len, const char *db, uint table_name_len,
@@ -200,211 +242,38 @@ int check_if_log_table(uint db_len, const char *db, uint table_name_len,
     if (table_name_len == 11 && !(lower_case_table_names ?
                                   my_strcasecmp(system_charset_info,
                                                 table_name, "general_log") :
-                                  strcmp(table_name, "general_log")) &&
-        (!check_if_opened || logger.is_log_table_enabled(QUERY_LOG_GENERAL)))
-      return QUERY_LOG_GENERAL;
-    else
-      if (table_name_len == 8 && !(lower_case_table_names ?
-        my_strcasecmp(system_charset_info, table_name, "slow_log") :
-        strcmp(table_name, "slow_log")) &&
-          (!check_if_opened ||logger.is_log_table_enabled(QUERY_LOG_SLOW)))
+                                  strcmp(table_name, "general_log")))
+    {
+      if (!check_if_opened || logger.is_log_table_enabled(QUERY_LOG_GENERAL))
+        return QUERY_LOG_GENERAL;
+      return 0;
+    }
+
+    if (table_name_len == 8 && !(lower_case_table_names ?
+      my_strcasecmp(system_charset_info, table_name, "slow_log") :
+      strcmp(table_name, "slow_log")))
+    {
+      if (!check_if_opened || logger.is_log_table_enabled(QUERY_LOG_SLOW))
         return QUERY_LOG_SLOW;
+      return 0;
+    }
   }
   return 0;
 }
 
 
-/*
-  Open log table of a given type (general or slow log)
-
-  SYNOPSIS
-    open_log_table()
-
-    log_table_type   type of the log table to open: QUERY_LOG_GENERAL
-                     or QUERY_LOG_SLOW
-
-  DESCRIPTION
-
-    The function opens a log table and marks it as such. Log tables are open
-    during the whole time, while server is running. Except for the moments
-    when they have to be reopened: during FLUSH LOGS and TRUNCATE. This
-    function is invoked directly only once during startup. All subsequent
-    calls happen through reopen_log_table(), which performs additional check.
-
-  RETURN
-    FALSE - OK
-    TRUE - error occured
-*/
-
-bool Log_to_csv_event_handler::open_log_table(uint log_table_type)
-{
-  THD *log_thd, *curr= current_thd;
-  TABLE_LIST *table;
-  bool error= FALSE;
-  DBUG_ENTER("open_log_table");
-
-  switch (log_table_type) {
-  case QUERY_LOG_GENERAL:
-    log_thd= general_log_thd;
-    table= &general_log;
-    /* clean up table before reuse/initial usage */
-    bzero((char*) table, sizeof(TABLE_LIST));
-    table->alias= table->table_name= (char*) "general_log";
-    table->table_name_length= 11;
-    break;
-  case QUERY_LOG_SLOW:
-    log_thd= slow_log_thd;
-    table= &slow_log;
-    bzero((char*) table, sizeof(TABLE_LIST));
-    table->alias= table->table_name= (char*) "slow_log";
-    table->table_name_length= 8;
-    break;
-  default:
-    assert(0);                                  // Impossible
-  }
-
-  /*
-    This way we check that appropriate log thd was created ok during
-    initialization. We cannot check "is_log_tables_initialized" var, as
-    the very initialization is not finished until this function is
-    completed in the very first time.
-  */
-  if (!log_thd)
-  {
-    DBUG_PRINT("error",("Cannot initialize log tables"));
-    DBUG_RETURN(TRUE);
-  }
-
-  /*
-    Set THD's thread_stack. This is needed to perform stack overrun
-    check, which is done by some routines (e.g. open_table()).
-    In the case we are called by thread, which already has this parameter
-    set, we use this value. Otherwise we do a wild guess. This won't help
-    to correctly track the stack overrun in these exceptional cases (which
-    could probably happen only during startup and shutdown) but at least
-    lets us to pass asserts.
-    The problem stems from the fact that logger THDs are not real threads.
-  */
-  if (curr)
-    log_thd->thread_stack= curr->thread_stack;
-  else
-    log_thd->thread_stack= (char*) &log_thd;
-
-  log_thd->store_globals();
-
-  table->lock_type= TL_WRITE_CONCURRENT_INSERT;
-  table->db= log_thd->db;
-  table->db_length= log_thd->db_length;
-
-  lex_start(log_thd);
-  log_thd->clear_error();
-  if (simple_open_n_lock_tables(log_thd, table) ||
-      table->table->file->extra(HA_EXTRA_MARK_AS_LOG_TABLE) ||
-      table->table->file->ha_rnd_init(0))
-    error= TRUE;
-  else
-  {
-    table->table->use_all_columns();
-    table->table->locked_by_logger= TRUE;
-    table->table->no_replicate= TRUE;
-
-    /* Honor next number columns if present */
-    table->table->next_number_field= table->table->found_next_number_field;
-  }
-  /* restore thread settings */
-  if (curr)
-    curr->store_globals();
-  else
-  {
-    my_pthread_setspecific_ptr(THR_THD,  0);
-    my_pthread_setspecific_ptr(THR_MALLOC, 0);
-  }
-
-  /*
-    After a log table was opened, we should clear privileged thread
-    flag (which allows locking of a log table by a special thread, usually
-    the one who closed log tables temporarily).
-  */
-  privileged_thread= 0;
-  DBUG_RETURN(error);
-}
-
-
 Log_to_csv_event_handler::Log_to_csv_event_handler()
 {
-  /* init artificial THD's */
-  general_log_thd= new THD;
-  /* logger thread always works with mysql database */
-  general_log_thd->db= my_strdup("mysql", MYF(0));
-  general_log_thd->db_length= 5;
-  general_log.table= 0;
-
-  slow_log_thd= new THD;
-  /* logger thread always works with mysql database */
-  slow_log_thd->db= my_strdup("mysql", MYF(0));;
-  slow_log_thd->db_length= 5;
-  slow_log.table= 0;
-  /* no privileged thread exists at the moment */
-  privileged_thread= 0;
 }
 
 
 Log_to_csv_event_handler::~Log_to_csv_event_handler()
 {
-  /* now cleanup the tables */
-  if (general_log_thd)
-  {
-    delete general_log_thd;
-    general_log_thd= NULL;
-  }
-
-  if (slow_log_thd)
-  {
-    delete slow_log_thd;
-    slow_log_thd= NULL;
-  }
-}
-
-
-/*
-  Reopen log table of a given type
-
-  SYNOPSIS
-    reopen_log_table()
-
-    log_table_type   type of the log table to open: QUERY_LOG_GENERAL
-                     or QUERY_LOG_SLOW
-
-  DESCRIPTION
-
-    The function is a wrapper around open_log_table(). It is used during
-    FLUSH LOGS and TRUNCATE of the log tables (i.e. when we need to close
-    and reopen them). The difference is in the check of the
-    logger.is_log_tables_initialized var, which can't be done in
-    open_log_table(), as it makes no sense during startup.
-
-    NOTE: this code assumes that we have logger mutex locked
-
-  RETURN
-    FALSE - ok
-    TRUE - open_log_table() returned an error
-*/
-
-bool Log_to_csv_event_handler::reopen_log_table(uint log_table_type)
-{
-  /* don't open the log table, if it wasn't enabled during startup */
-  if (!logger.is_log_tables_initialized)
-    return FALSE;
-  return open_log_table(log_table_type);
 }
 
 
 void Log_to_csv_event_handler::cleanup()
 {
-  if (opt_log)
-    close_log_table(QUERY_LOG_GENERAL, FALSE);
-  if (opt_slow_log)
-    close_log_table(QUERY_LOG_SLOW, FALSE);
   logger.is_log_tables_initialized= FALSE;
 }
 
@@ -436,49 +305,88 @@ void Log_to_csv_event_handler::cleanup()
 */
 
 bool Log_to_csv_event_handler::
-  log_general(time_t event_time, const char *user_host,
+  log_general(THD *thd, time_t event_time, const char *user_host,
               uint user_host_len, int thread_id,
               const char *command_type, uint command_type_len,
               const char *sql_text, uint sql_text_len,
               CHARSET_INFO *client_cs)
 {
-  TABLE *table= general_log.table;
+  TABLE_LIST table_list;
+  TABLE *table;
+  bool result= TRUE;
+  bool need_close= FALSE;
+  bool need_pop= FALSE;
+  bool need_rnd_end= FALSE;
   uint field_index;
+  Silence_log_table_errors error_handler;
+  Open_tables_state open_tables_backup;
+  Field_timestamp *field0;
+  ulonglong save_thd_options;
+  bool save_query_start_used;
+  time_t save_start_time;
+  time_t save_time_after_lock;
+  time_t save_user_time;
+  bool save_time_zone_used;
+
+  save_thd_options= thd->options;
+  thd->options&= ~OPTION_BIN_LOG;
+
+  save_query_start_used= thd->query_start_used;
+  save_start_time= thd->start_time;
+  save_time_after_lock= thd->time_after_lock;
+  save_user_time= thd->user_time;
+  save_time_zone_used= thd->time_zone_used;
+
+  bzero(& table_list, sizeof(TABLE_LIST));
+  table_list.alias= table_list.table_name= GENERAL_LOG_NAME.str;
+  table_list.table_name_length= GENERAL_LOG_NAME.length;
+
+  table_list.lock_type= TL_WRITE_CONCURRENT_INSERT;
+
+  table_list.db= MYSQL_SCHEMA_NAME.str;
+  table_list.db_length= MYSQL_SCHEMA_NAME.length;
+
+  table= open_performance_schema_table(thd, & table_list,
+                                       & open_tables_backup);
+  need_close= TRUE;
+
+  if (!table ||
+      table->file->extra(HA_EXTRA_MARK_AS_LOG_TABLE) ||
+      table->file->ha_rnd_init(0))
+    goto err;
+
+  need_rnd_end= TRUE;
+
+  /* Honor next number columns if present */
+  table->next_number_field= table->found_next_number_field;
 
   /*
     "INSERT INTO general_log" can generate warning sometimes.
-    Let's reset warnings from previous queries,
-    otherwise warning list can grow too much,
-    so thd->query gets spoiled as some point in time,
-    and mysql_parse() receives a broken query.
     QQ: this problem needs to be studied in more details.
-    Probably it's better to suppress warnings in logging INSERTs at all.
-    Comment this line and run "cast.test" to see what's happening:
+    Comment this 2 lines and run "cast.test" to see what's happening:
   */
-  mysql_reset_errors(table->in_use, 1);
-
-  /* below should never happen */
-  if (unlikely(!logger.is_log_tables_initialized))
-    return FALSE;
+  thd->push_internal_handler(& error_handler);
+  need_pop= TRUE;
 
   /*
     NOTE: we do not call restore_record() here, as all fields are
     filled by the Logger (=> no need to load default ones).
   */
 
-  /* Set current time. Required for CURRENT_TIMESTAMP to work */
-  general_log_thd->start_time= event_time;
-
   /*
     We do not set a value for table->field[0], as it will use
     default value (which is CURRENT_TIMESTAMP).
   */
 
   /* check that all columns exist */
-  if (!table->field[1] || !table->field[2] || !table->field[3] ||
-      !table->field[4] || !table->field[5])
+  if (table->s->fields < 6)
     goto err;
 
+  DBUG_ASSERT(table->field[0]->type() == MYSQL_TYPE_TIMESTAMP);
+
+  field0= (Field_timestamp*) (table->field[0]);
+  field0->set_time();
+
   /* do a write */
   if (table->field[1]->store(user_host, user_host_len, client_cs) ||
       table->field[2]->store((longlong) thread_id, TRUE) ||
@@ -500,16 +408,39 @@ bool Log_to_csv_event_handler::
     table->field[field_index]->set_default();
   }
 
-  /* log table entries are not replicated at the moment */
-  tmp_disable_binlog(current_thd);
+  /* log table entries are not replicated */
+  if (table->file->ha_write_row(table->record[0]))
+  {
+    struct tm start;
+    localtime_r(&event_time, &start);
 
-  table->file->ha_write_row(table->record[0]);
+    sql_print_error("%02d%02d%02d %2d:%02d:%02d - Failed to write to mysql.general_log",
+                    start.tm_year % 100, start.tm_mon + 1,
+                    start.tm_mday, start.tm_hour,
+                    start.tm_min, start.tm_sec);
+  }
 
-  reenable_binlog(current_thd);
+  result= FALSE;
 
-  return FALSE;
 err:
-  return TRUE;
+  if (need_rnd_end)
+  {
+    table->file->ha_rnd_end();
+    table->file->ha_release_auto_increment();
+  }
+  if (need_pop)
+    thd->pop_internal_handler();
+  if (need_close)
+    close_performance_schema_table(thd, & open_tables_backup);
+
+  thd->options= save_thd_options;
+
+  thd->query_start_used= save_query_start_used;
+  thd->start_time= save_start_time;
+  thd->time_after_lock= save_time_after_lock;
+  thd->user_time= save_user_time;
+  thd->time_zone_used= save_time_zone_used;
+  return result;
 }
 
 
@@ -548,34 +479,61 @@ bool Log_to_csv_event_handler::
            longlong query_time, longlong lock_time, bool is_command,
            const char *sql_text, uint sql_text_len)
 {
-  /* table variables */
-  TABLE *table= slow_log.table;
+  TABLE_LIST table_list;
+  TABLE *table;
+  bool result= TRUE;
+  bool need_close= FALSE;
+  bool need_rnd_end= FALSE;
+  Open_tables_state open_tables_backup;
+  bool save_query_start_used;
+  time_t save_start_time;
+  time_t save_time_after_lock;
+  time_t save_user_time;
+  bool save_time_zone_used;
   CHARSET_INFO *client_cs= thd->variables.character_set_client;
 
-  DBUG_ENTER("log_slow");
+  DBUG_ENTER("Log_to_csv_event_handler::log_slow");
 
-  /* below should never happen */
-  if (unlikely(!logger.is_log_tables_initialized))
-    return FALSE;
+  bzero(& table_list, sizeof(TABLE_LIST));
+  table_list.alias= table_list.table_name= SLOW_LOG_NAME.str;
+  table_list.table_name_length= SLOW_LOG_NAME.length;
+
+  table_list.lock_type= TL_WRITE_CONCURRENT_INSERT;
+
+  table_list.db= MYSQL_SCHEMA_NAME.str;
+  table_list.db_length= MYSQL_SCHEMA_NAME.length;
+
+  save_query_start_used= thd->query_start_used;
+  save_start_time= thd->start_time;
+  save_time_after_lock= thd->time_after_lock;
+  save_user_time= thd->user_time;
+  save_time_zone_used= thd->time_zone_used;
+
+  table= open_performance_schema_table(thd, & table_list,
+                                       & open_tables_backup);
+  need_close= TRUE;
+
+  if (!table ||
+      table->file->extra(HA_EXTRA_MARK_AS_LOG_TABLE) ||
+      table->file->ha_rnd_init(0))
+    goto err;
+
+  need_rnd_end= TRUE;
+
+  /* Honor next number columns if present */
+  table->next_number_field= table->found_next_number_field;
 
-  /*
-     Set start time for CURRENT_TIMESTAMP to the start of the query.
-     This will be default value for the field[0]
-  */
-  slow_log_thd->start_time= query_start_arg;
   restore_record(table, s->default_values);    // Get empty record
 
+  /* check that all columns exist */
+  if (table->s->fields < 11)
+    goto err;
+
   /*
     We do not set a value for table->field[0], as it will use
     default value.
   */
 
-  if (!table->field[1] || !table->field[2] || !table->field[3] ||
-      !table->field[4] || !table->field[5] || !table->field[6] ||
-      !table->field[7] || !table->field[8] || !table->field[9] ||
-      !table->field[10])
-    goto err;
-
   /* store the value */
   if (table->field[1]->store(user_host, user_host_len, client_cs))
     goto err;
@@ -612,7 +570,6 @@ bool Log_to_csv_event_handler::
     table->field[4]->set_null();
     table->field[5]->set_null();
   }
-
   /* fill database field */
   if (thd->db)
   {
@@ -654,17 +611,71 @@ bool Log_to_csv_event_handler::
   if (table->field[10]->store(sql_text,sql_text_len, client_cs))
     goto err;
 
-  /* log table entries are not replicated at the moment */
-  tmp_disable_binlog(current_thd);
+  /* log table entries are not replicated */
+  if (table->file->ha_write_row(table->record[0]))
+  {
+    struct tm start;
+    localtime_r(&current_time, &start);
 
-  /* write the row */
-  table->file->ha_write_row(table->record[0]);
+    sql_print_error("%02d%02d%02d %2d:%02d:%02d - Failed to write to mysql.slow_log",
+                    start.tm_year % 100, start.tm_mon + 1,
+                    start.tm_mday, start.tm_hour,
+                    start.tm_min, start.tm_sec);
+  }
 
-  reenable_binlog(current_thd);
+  result= FALSE;
 
-  DBUG_RETURN(0);
 err:
-  DBUG_RETURN(1);
+  if (need_rnd_end)
+  {
+    table->file->ha_rnd_end();
+    table->file->ha_release_auto_increment();
+  }
+  if (need_close)
+    close_performance_schema_table(thd, & open_tables_backup);
+
+  thd->query_start_used= save_query_start_used;
+  thd->start_time= save_start_time;
+  thd->time_after_lock= save_time_after_lock;
+  thd->user_time= save_user_time;
+  thd->time_zone_used= save_time_zone_used;
+  DBUG_RETURN(result);
+}
+
+int Log_to_csv_event_handler::
+  activate_log(THD *thd, uint log_table_type)
+{
+  TABLE_LIST table_list;
+  TABLE *table;
+  int result;
+  Open_tables_state open_tables_backup;
+
+  DBUG_ENTER("Log_to_csv_event_handler::activate_log");
+
+  bzero(& table_list, sizeof(TABLE_LIST));
+
+  if (log_table_type == QUERY_LOG_GENERAL)
+  {
+    table_list.alias= table_list.table_name= GENERAL_LOG_NAME.str;
+    table_list.table_name_length= GENERAL_LOG_NAME.length;
+  }
+  else
+  {
+    DBUG_ASSERT(log_table_type == QUERY_LOG_SLOW);
+    table_list.alias= table_list.table_name= SLOW_LOG_NAME.str;
+    table_list.table_name_length= SLOW_LOG_NAME.length;
+  }
+
+  table_list.lock_type= TL_WRITE_CONCURRENT_INSERT;
+
+  table_list.db= MYSQL_SCHEMA_NAME.str;
+  table_list.db_length= MYSQL_SCHEMA_NAME.length;
+
+  table= open_performance_schema_table(thd, & table_list,
+                                       & open_tables_backup);
+  result= (table ? 0 : 1);
+  close_performance_schema_table(thd, & open_tables_backup);
+  DBUG_RETURN(result);
 }
 
 bool Log_to_csv_event_handler::
@@ -710,7 +721,7 @@ bool Log_to_file_event_handler::
 */
 
 bool Log_to_file_event_handler::
-  log_general(time_t event_time, const char *user_host,
+  log_general(THD *thd, time_t event_time, const char *user_host,
               uint user_host_len, int thread_id,
               const char *command_type, uint command_type_len,
               const char *sql_text, uint sql_text_len,
@@ -787,7 +798,7 @@ bool LOGGER::error_log_print(enum loglevel level, const char *format,
 void LOGGER::cleanup_base()
 {
   DBUG_ASSERT(inited == 1);
-  (void) pthread_mutex_destroy(&LOCK_logger);
+  rwlock_destroy(&LOCK_logger);
   if (table_log_handler)
   {
     table_log_handler->cleanup();
@@ -806,12 +817,6 @@ void LOGGER::cleanup_end()
 }
 
 
-void LOGGER::close_log_table(uint log_table_type, bool lock_in_use)
-{
-  table_log_handler->close_log_table(log_table_type, lock_in_use);
-}
-
-
 /*
   Perform basic log initialization: create file-based log handler and
   init error log.
@@ -833,7 +838,7 @@ void LOGGER::init_base()
   init_error_log(LOG_FILE);
 
   file_log_handler->init_pthread_objects();
-  (void) pthread_mutex_init(&LOCK_logger, MY_MUTEX_INIT_SLOW);
+  my_rwlock_init(&LOCK_logger, NULL);
 }
 
 
@@ -848,29 +853,6 @@ void LOGGER::init_log_tables()
 }
 
 
-bool LOGGER::reopen_log_table(uint log_table_type)
-{
-  return table_log_handler->reopen_log_table(log_table_type);
-}
-
-bool LOGGER::reopen_log_tables()
-{
-    /*
-      we use | and not || here, to ensure that both reopen_log_table
-      are called, even if the first one fails
-    */
-    if ((opt_slow_log && logger.reopen_log_table(QUERY_LOG_SLOW)) |
-        (opt_log && logger.reopen_log_table(QUERY_LOG_GENERAL)))
-      return TRUE;
-    return FALSE;
-}
-
-
-void LOGGER::tmp_close_log_tables(THD *thd)
-{
-  table_log_handler->tmp_close_log_tables(thd);
-}
-
 bool LOGGER::flush_logs(THD *thd)
 {
   int rc= 0;
@@ -879,19 +861,11 @@ bool LOGGER::flush_logs(THD *thd)
     Now we lock logger, as nobody should be able to use logging routines while
     log tables are closed
   */
-  logger.lock();
-  if (logger.is_log_tables_initialized)
-    table_log_handler->tmp_close_log_tables(thd); // the locking happens here
+  logger.lock_exclusive();
 
   /* reopen log files */
   file_log_handler->flush();
 
-  /* reopen tables in the case they were enabled */
-  if (logger.is_log_tables_initialized)
-  {
-    if (reopen_log_tables())
-      rc= TRUE;
-  }
   /* end of log flush */
   logger.unlock();
   return rc;
@@ -939,7 +913,7 @@ bool LOGGER::slow_log_print(THD *thd, const char *query, uint query_length,
     if (thd->slave_thread)
       return 0;
 
-    lock();
+    lock_shared();
     if (!opt_slow_log)
     {
       unlock();
@@ -1011,7 +985,7 @@ bool LOGGER::general_log_print(THD *thd, enum enum_server_command command,
     else
       id=0;                                     /* Log from connect handler */
 
-    lock();
+    lock_shared();
     if (!opt_log)
     {
       unlock();
@@ -1035,7 +1009,7 @@ bool LOGGER::general_log_print(THD *thd, enum enum_server_command command,
 
     while (*current_handler)
       error+= (*current_handler++)->
-        log_general(current_time, user_host_buff,
+        log_general(thd, current_time, user_host_buff,
                    user_host_len, id,
                    command_name[(uint) command].str,
                    command_name[(uint) command].length,
@@ -1122,35 +1096,51 @@ void LOGGER::init_general_log(uint general_log_printer)
 
 bool LOGGER::activate_log_handler(THD* thd, uint log_type)
 {
-  bool res= 0;
-  lock();
+  MYSQL_QUERY_LOG *file_log;
+  bool res= FALSE;
+  lock_exclusive();
   switch (log_type) {
   case QUERY_LOG_SLOW:
     if (!opt_slow_log)
     {
-      if ((res= reopen_log_table(log_type)))
-        goto err;
-      file_log_handler->get_mysql_slow_log()->
-        open_slow_log(sys_var_slow_log_path.value);
-      init_slow_log(log_output_options);
-      opt_slow_log= TRUE;
+      file_log= file_log_handler->get_mysql_slow_log();
+
+      file_log->open_slow_log(sys_var_slow_log_path.value);
+      if (table_log_handler->activate_log(thd, QUERY_LOG_SLOW))
+      {
+        /* Error printed by open table in activate_log() */
+        res= TRUE;
+        file_log->close(0);
+      }
+      else
+      {
+        init_slow_log(log_output_options);
+        opt_slow_log= TRUE;
+      }
     }
     break;
   case QUERY_LOG_GENERAL:
     if (!opt_log)
     {
-      if ((res= reopen_log_table(log_type)))
-        goto err;
-      file_log_handler->get_mysql_log()->
-        open_query_log(sys_var_general_log_path.value);
-      init_general_log(log_output_options);
-      opt_log= TRUE;
+      file_log= file_log_handler->get_mysql_log();
+
+      file_log->open_query_log(sys_var_general_log_path.value);
+      if (table_log_handler->activate_log(thd, QUERY_LOG_GENERAL))
+      {
+        /* Error printed by open table in activate_log() */
+        res= TRUE;
+        file_log->close(0);
+      }
+      else
+      {
+        init_general_log(log_output_options);
+        opt_log= TRUE;
+      }
     }
     break;
   default:
     DBUG_ASSERT(0);
   }
-err:
   unlock();
   return res;
 }
@@ -1158,23 +1148,17 @@ err:
 
 void LOGGER::deactivate_log_handler(THD *thd, uint log_type)
 {
-  TABLE_LIST *table_list;
   my_bool *tmp_opt= 0;
   MYSQL_LOG *file_log;
-  THD *log_thd;
 
   switch (log_type) {
   case QUERY_LOG_SLOW:
-    table_list= &table_log_handler->slow_log;
     tmp_opt= &opt_slow_log;
     file_log= file_log_handler->get_mysql_slow_log();
-    log_thd= table_log_handler->slow_log_thd;
     break;
   case QUERY_LOG_GENERAL:
-    table_list= &table_log_handler->general_log;
     tmp_opt= &opt_log;
     file_log= file_log_handler->get_mysql_log();
-    log_thd= table_log_handler->general_log_thd;
     break;
   default:
     assert(0);                                  // Impossible
@@ -1183,81 +1167,16 @@ void LOGGER::deactivate_log_handler(THD *thd, uint log_type)
   if (!(*tmp_opt))
     return;
 
-  if (is_log_tables_initialized)
-    lock_and_wait_for_table_name(log_thd, table_list);
-  lock();
-
-  if (is_log_tables_initialized)
-  {
-    VOID(pthread_mutex_lock(&LOCK_open));
-    close_log_table(log_type, TRUE);
-    table_list->table= 0;
-    query_cache_invalidate3(log_thd, table_list, 0);
-    unlock_table_name(log_thd, table_list);
-    VOID(pthread_mutex_unlock(&LOCK_open));
-  }
+  lock_exclusive();
   file_log->close(0);
   *tmp_opt= FALSE;
   unlock();
 }
 
 
-/*
-  Close log tables temporarily. The thread which closed
-  them this way can lock them in any mode it needs.
-  NOTE: one should call logger.lock() before entering this
-  function.
-*/
-void Log_to_csv_event_handler::tmp_close_log_tables(THD *thd)
-{
-  TABLE_LIST close_slow_log, close_general_log;
-
-  /* fill lists, we will need to perform operations on tables */
-  bzero((char*) &close_slow_log, sizeof(TABLE_LIST));
-  close_slow_log.alias= close_slow_log.table_name=(char*) "slow_log";
-  close_slow_log.table_name_length= 8;
-  close_slow_log.db= (char*) "mysql";
-  close_slow_log.db_length= 5;
-
-  bzero((char*) &close_general_log, sizeof(TABLE_LIST));
-  close_general_log.alias= close_general_log.table_name=(char*) "general_log";
-  close_general_log.table_name_length= 11;
-  close_general_log.db= (char*) "mysql";
-  close_general_log.db_length= 5;
-
-  privileged_thread= thd;
-
-  VOID(pthread_mutex_lock(&LOCK_open));
-  /*
-    NOTE: in fact, the first parameter used in query_cache_invalidate3()
-    could be any non-NULL THD, as the underlying code makes certain
-    assumptions about this.
-    Here we use one of the logger handler THD's. Simply because it
-    seems appropriate.
-  */
-  if (opt_log)
-  {
-    close_log_table(QUERY_LOG_GENERAL, TRUE);
-    query_cache_invalidate3(general_log_thd, &close_general_log, 0);
-  }
-  if (opt_slow_log)
-  {
-    close_log_table(QUERY_LOG_SLOW, TRUE);
-    query_cache_invalidate3(general_log_thd, &close_slow_log, 0);
-  }
-  VOID(pthread_mutex_unlock(&LOCK_open));
-}
-
 /* the parameters are unused for the log tables */
 bool Log_to_csv_event_handler::init()
 {
-  /*
-    we use | and not || here, to ensure that both open_log_table
-    are called, even if the first one fails
-  */
-  if ((opt_log && open_log_table(QUERY_LOG_GENERAL)) |
-      (opt_slow_log && open_log_table(QUERY_LOG_SLOW)))
-    return 1;
   return 0;
 }
 
@@ -1268,7 +1187,7 @@ int LOGGER::set_handlers(uint error_log_printer,
   /* error log table is not supported yet */
   DBUG_ASSERT(error_log_printer < LOG_TABLE);
 
-  lock();
+  lock_exclusive();
 
   if ((slow_log_printer & LOG_TABLE || general_log_printer & LOG_TABLE) &&
       !is_log_tables_initialized)
@@ -1290,72 +1209,6 @@ int LOGGER::set_handlers(uint error_log_printer,
 }
 
 
-/*
-  Close log table of a given type (general or slow log)
-
-  SYNOPSIS
-    close_log_table()
-
-    log_table_type   type of the log table to close: QUERY_LOG_GENERAL
-                     or QUERY_LOG_SLOW
-    lock_in_use      Set to TRUE if the caller owns LOCK_open. FALSE otherwise.
-
-  DESCRIPTION
-
-    The function closes a log table. It is invoked (1) when we need to reopen
-    log tables (e.g. FLUSH LOGS or TRUNCATE on the log table is being
-    executed) or (2) during shutdown.
-*/
-
-void Log_to_csv_event_handler::
-  close_log_table(uint log_table_type, bool lock_in_use)
-{
-  THD *log_thd, *curr= current_thd;
-  TABLE_LIST *table;
-
-  if (!logger.is_log_table_enabled(log_table_type))
-    return;                                     /* do nothing */
-
-  switch (log_table_type) {
-  case QUERY_LOG_GENERAL:
-    log_thd= general_log_thd;
-    table= &general_log;
-    break;
-  case QUERY_LOG_SLOW:
-    log_thd= slow_log_thd;
-    table= &slow_log;
-    break;
-  default:
-    assert(0);                                  // Impossible
-  }
-
-  /*
-    Set thread stack start for the logger thread. See comment in
-    open_log_table() for details.
-  */
-  if (curr)
-    log_thd->thread_stack= curr->thread_stack;
-  else
-    log_thd->thread_stack= (char*) &log_thd;
-
-  /* close the table */
-  log_thd->store_globals();
-  table->table->file->ha_rnd_end();
-  table->table->file->ha_release_auto_increment();
-  /* discard logger mark before unlock*/
-  table->table->locked_by_logger= FALSE;
-  close_thread_tables(log_thd, lock_in_use);
-
-  if (curr)
-    curr->store_globals();
-  else
-  {
-    my_pthread_setspecific_ptr(THR_THD,  0);
-    my_pthread_setspecific_ptr(THR_MALLOC, 0);
-  }
-}
-
-
  /*
   Save position of binary log transaction cache.
 
@@ -1609,7 +1462,8 @@ static int binlog_rollback(handlerton *hton, THD *thd, bool all)
     table. Such cases should be rare (updating a
     non-transactional table inside a transaction...)
   */
-  if (unlikely(thd->no_trans_update.all || (thd->options & OPTION_KEEP_LOG)))
+  if (unlikely(thd->transaction.all.modified_non_trans_table || 
+               (thd->options & OPTION_KEEP_LOG)))
   {
     Query_log_event qev(thd, STRING_WITH_LEN("ROLLBACK"), TRUE, FALSE);
     qev.error_code= 0; // see comment in MYSQL_LOG::write(THD, IO_CACHE)
@@ -1663,7 +1517,8 @@ static int binlog_savepoint_rollback(handlerton *hton, THD *thd, void *sv)
     non-transactional table. Otherwise, truncate the binlog cache starting
     from the SAVEPOINT command.
   */
-  if (unlikely(thd->no_trans_update.all || (thd->options & OPTION_KEEP_LOG)))
+  if (unlikely(thd->transaction.all.modified_non_trans_table || 
+               (thd->options & OPTION_KEEP_LOG)))
   {
     int error=
       thd->binlog_query(THD::STMT_QUERY_TYPE,
@@ -2097,6 +1952,8 @@ bool MYSQL_QUERY_LOG::write(time_t event_time, const char *user_host,
   struct tm start;
   uint time_buff_len= 0;
 
+  (void) pthread_mutex_lock(&LOCK_log);
+
   /* Test if someone closed between the is_open test and lock */
   if (is_open())
   {
@@ -2141,6 +1998,7 @@ bool MYSQL_QUERY_LOG::write(time_t event_time, const char *user_host,
       goto err;
   }
 
+  (void) pthread_mutex_unlock(&LOCK_log);
   return FALSE;
 err:
 
@@ -2149,6 +2007,7 @@ err:
     write_error= 1;
     sql_print_error(ER(ER_ERROR_ON_WRITE), name, errno);
   }
+  (void) pthread_mutex_unlock(&LOCK_log);
   return TRUE;
 }
 
@@ -2191,8 +2050,13 @@ bool MYSQL_QUERY_LOG::write(THD *thd, time_t current_time,
   bool error= 0;
   DBUG_ENTER("MYSQL_QUERY_LOG::write");
 
+  (void) pthread_mutex_lock(&LOCK_log);
+
   if (!is_open())
+  {
+    (void) pthread_mutex_unlock(&LOCK_log);
     DBUG_RETURN(0);
+  }
 
   if (is_open())
   {						// Safety agains reopen
@@ -2296,6 +2160,7 @@ bool MYSQL_QUERY_LOG::write(THD *thd, time_t current_time,
       }
     }
   }
+  (void) pthread_mutex_unlock(&LOCK_log);
   DBUG_RETURN(error);
 }
 
@@ -4094,6 +3959,7 @@ bool MYSQL_BIN_LOG::write(THD *thd, IO_CACHE *cache, Log_event *commit_event)
   /* NULL would represent nothing to replicate after ROLLBACK */
   DBUG_ASSERT(commit_event != NULL);
 
+  DBUG_ASSERT(is_open());
   if (likely(is_open()))                       // Should always be true
   {
     /*
diff --git a/sql/log.h b/sql/log.h
index d92e0117bcc..3b1a0950daa 100644
--- a/sql/log.h
+++ b/sql/log.h
@@ -212,6 +212,7 @@ public:
     return open(generate_name(log_name, ".log", 0, buf), LOG_NORMAL, 0,
                 WRITE_CACHE);
   }
+
 private:
   time_t last_time;
 };
@@ -398,7 +399,7 @@ public:
                         const char *sql_text, uint sql_text_len)= 0;
   virtual bool log_error(enum loglevel level, const char *format,
                          va_list args)= 0;
-  virtual bool log_general(time_t event_time, const char *user_host,
+  virtual bool log_general(THD *thd, time_t event_time, const char *user_host,
                            uint user_host_len, int thread_id,
                            const char *command_type, uint command_type_len,
                            const char *sql_text, uint sql_text_len,
@@ -412,27 +413,7 @@ int check_if_log_table(uint db_len, const char *db, uint table_name_len,
 
 class Log_to_csv_event_handler: public Log_event_handler
 {
-  /*
-    We create artificial THD for each of the logs. This is to avoid
-    locking issues: we don't want locks on the log tables reside in the
-    THD's of the query. The reason is the locking order and duration.
-  */
-  THD *general_log_thd, *slow_log_thd;
-  /*
-    This is for the thread, which called tmp_close_log_tables. The thread
-    will be allowed to write-lock the log tables (as it explicitly disabled
-    logging). This is used for such operations as REPAIR, which require
-    exclusive lock on the log tables.
-    NOTE: there can be only one priviliged thread, as one should
-    lock logger with logger.lock() before calling tmp_close_log_tables().
-    So no other thread could get privileged status at the same time.
-  */
-  THD *privileged_thread;
   friend class LOGGER;
-  TABLE_LIST general_log, slow_log;
-
-private:
-  bool open_log_table(uint log_type);
 
 public:
   Log_to_csv_event_handler();
@@ -447,18 +428,13 @@ public:
                         const char *sql_text, uint sql_text_len);
   virtual bool log_error(enum loglevel level, const char *format,
                          va_list args);
-  virtual bool log_general(time_t event_time, const char *user_host,
+  virtual bool log_general(THD *thd, time_t event_time, const char *user_host,
                            uint user_host_len, int thread_id,
                            const char *command_type, uint command_type_len,
                            const char *sql_text, uint sql_text_len,
-                            CHARSET_INFO *client_cs);
-  void tmp_close_log_tables(THD *thd);
-  void close_log_table(uint log_type, bool lock_in_use);
-  bool reopen_log_table(uint log_type);
-  THD* get_privileged_thread()
-  {
-    return privileged_thread;
-  }
+                           CHARSET_INFO *client_cs);
+
+  int activate_log(THD *thd, uint log_type);
 };
 
 
@@ -484,7 +460,7 @@ public:
                         const char *sql_text, uint sql_text_len);
   virtual bool log_error(enum loglevel level, const char *format,
                          va_list args);
-  virtual bool log_general(time_t event_time, const char *user_host,
+  virtual bool log_general(THD *thd, time_t event_time, const char *user_host,
                            uint user_host_len, int thread_id,
                            const char *command_type, uint command_type_len,
                            const char *sql_text, uint sql_text_len,
@@ -499,7 +475,7 @@ public:
 /* Class which manages slow, general and error log event handlers */
 class LOGGER
 {
-  pthread_mutex_t LOCK_logger;
+  rw_lock_t LOCK_logger;
   /* flag to check whether logger mutex is initialized */
   uint inited;
 
@@ -519,21 +495,10 @@ public:
   LOGGER() : inited(0), table_log_handler(NULL),
              file_log_handler(NULL), is_log_tables_initialized(FALSE)
   {}
-  void lock() { (void) pthread_mutex_lock(&LOCK_logger); }
-  void unlock() { (void) pthread_mutex_unlock(&LOCK_logger); }
-  void tmp_close_log_tables(THD *thd);
-  bool is_log_table_enabled(uint log_table_type)
-  {
-    switch (log_table_type) {
-    case QUERY_LOG_SLOW:
-      return table_log_handler && table_log_handler->slow_log.table != 0;
-    case QUERY_LOG_GENERAL:
-      return table_log_handler && table_log_handler->general_log.table != 0;
-    default:
-      DBUG_ASSERT(0);
-      return FALSE;                             /* make compiler happy */
-    }
-  }
+  void lock_shared() { rw_rdlock(&LOCK_logger); }
+  void lock_exclusive() { rw_wrlock(&LOCK_logger); }
+  void unlock() { rw_unlock(&LOCK_logger); }
+  bool is_log_table_enabled(uint log_table_type);
   /*
     We want to initialize all log mutexes as soon as possible,
     but we cannot do it in constructor, as safe_mutex relies on
@@ -543,20 +508,6 @@ public:
   void init_base();
   void init_log_tables();
   bool flush_logs(THD *thd);
-  THD *get_general_log_thd()
-  {
-    if (table_log_handler)
-      return (THD *) table_log_handler->general_log_thd;
-    else
-      return NULL;
-  }
-  THD *get_slow_log_thd()
-  {
-    if (table_log_handler)
-      return (THD *) table_log_handler->slow_log_thd;
-    else
-      return NULL;
-  }
   /* Perform basic logger cleanup. this will leave e.g. error log open. */
   void cleanup_base();
   /* Free memory. Nothing could be logged after this function is called */
@@ -568,10 +519,6 @@ public:
   bool general_log_print(THD *thd,enum enum_server_command command,
                          const char *format, va_list args);
 
-  void close_log_table(uint log_type, bool lock_in_use);
-  bool reopen_log_table(uint log_type);
-  bool reopen_log_tables();
-
   /* we use this function to setup all enabled log event handlers */
   int set_handlers(uint error_log_printer,
                    uint slow_log_printer,
@@ -593,19 +540,6 @@ public:
       return file_log_handler->get_mysql_log();
     return NULL;
   }
-  THD* get_privileged_thread()
-  {
-    if (table_log_handler)
-      return table_log_handler->get_privileged_thread();
-    else
-      return NULL;
-  }
-  bool is_privileged_thread(THD *thd)
-  {
-    return thd == get_general_log_thd() ||
-           thd == get_slow_log_thd() ||
-           thd == get_privileged_thread();
-  }
 };
 
 enum enum_binlog_format {
diff --git a/sql/log_event.cc b/sql/log_event.cc
index 6f63d8bf718..bb8bc33fd82 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -1479,8 +1479,8 @@ Query_log_event::Query_log_event(THD* thd_arg, const char* query_arg,
 				 ulong query_length, bool using_trans,
 				 bool suppress_use, THD::killed_state killed_status_arg)
   :Log_event(thd_arg,
-	     ((thd_arg->tmp_table_used ? LOG_EVENT_THREAD_SPECIFIC_F : 0)
-	      | (suppress_use          ? LOG_EVENT_SUPPRESS_USE_F    : 0)),
+             (thd_arg->thread_specific_used ? LOG_EVENT_THREAD_SPECIFIC_F : 0) |
+               (suppress_use ? LOG_EVENT_SUPPRESS_USE_F : 0),
 	     using_trans),
    data_buf(0), query(query_arg), catalog(thd_arg->catalog),
    db(thd_arg->db), q_len((uint32) query_length),
@@ -2912,8 +2912,9 @@ Load_log_event::Load_log_event(THD *thd_arg, sql_exchange *ex,
 			       List<Item> &fields_arg,
 			       enum enum_duplicates handle_dup,
 			       bool ignore, bool using_trans)
-  :Log_event(thd_arg, !thd_arg->tmp_table_used ?
-	     0 : LOG_EVENT_THREAD_SPECIFIC_F, using_trans),
+  :Log_event(thd_arg,
+             thd_arg->thread_specific_used ? LOG_EVENT_THREAD_SPECIFIC_F : 0,
+             using_trans),
    thread_id(thd_arg->thread_id),
    slave_proxy_id(thd_arg->variables.pseudo_thread_id),
    num_fields(0),fields(0),
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index 7b7bc81957e..2fe70980ea8 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -944,6 +944,7 @@ void mysql_parse(THD *thd, const char *inBuf, uint length,
 
 bool mysql_test_parse_for_slave(THD *thd,char *inBuf,uint length);
 bool is_update_query(enum enum_sql_command command);
+bool is_log_table_write_query(enum enum_sql_command command);
 bool alloc_query(THD *thd, const char *packet, uint packet_length);
 void mysql_init_select(LEX *lex);
 void mysql_reset_thd_for_next_command(THD *thd);
@@ -1123,7 +1124,8 @@ TABLE_SHARE *get_table_share(THD *thd, TABLE_LIST *table_list, char *key,
                              uint key_length, uint db_flags, int *error);
 void release_table_share(TABLE_SHARE *share, enum release_type type);
 TABLE_SHARE *get_cached_table_share(const char *db, const char *table_name);
-TABLE *open_ltable(THD *thd, TABLE_LIST *table_list, thr_lock_type update);
+TABLE *open_ltable(THD *thd, TABLE_LIST *table_list, thr_lock_type update,
+                   uint lock_flags);
 TABLE *open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT* mem,
 		  bool *refresh, uint flags);
 bool reopen_name_locked_table(THD* thd, TABLE_LIST* table_list, bool link_in);
@@ -1231,6 +1233,11 @@ void reset_status_vars();
 
 /* information schema */
 extern LEX_STRING INFORMATION_SCHEMA_NAME;
+/* log tables */
+extern LEX_STRING MYSQL_SCHEMA_NAME;
+extern LEX_STRING GENERAL_LOG_NAME;
+extern LEX_STRING SLOW_LOG_NAME;
+
 extern const LEX_STRING partition_keywords[];
 ST_SCHEMA_TABLE *find_schema_table(THD *thd, const char* table_name);
 ST_SCHEMA_TABLE *get_schema_table(enum enum_schema_tables schema_table_idx);
@@ -1557,6 +1564,10 @@ bool open_system_tables_for_read(THD *thd, TABLE_LIST *table_list,
 void close_system_tables(THD *thd, Open_tables_state *backup);
 TABLE *open_system_table_for_update(THD *thd, TABLE_LIST *one_table);
 
+TABLE *open_performance_schema_table(THD *thd, TABLE_LIST *one_table,
+                                     Open_tables_state *backup);
+void close_performance_schema_table(THD *thd, Open_tables_state *backup);
+
 bool close_cached_tables(THD *thd, bool wait_for_refresh, TABLE_LIST *tables, bool have_lock = FALSE);
 bool close_cached_connection_tables(THD *thd, bool wait_for_refresh,
                                     LEX_STRING *connect_string,
@@ -1891,11 +1902,13 @@ MYSQL_LOCK *mysql_lock_tables(THD *thd, TABLE **table, uint count,
 #define MYSQL_LOCK_NOTIFY_IF_NEED_REOPEN        0x0004
 #define MYSQL_OPEN_TEMPORARY_ONLY               0x0008
 #define MYSQL_LOCK_IGNORE_GLOBAL_READ_ONLY      0x0010
+#define MYSQL_LOCK_PERF_SCHEMA                  0x0020
 
 void mysql_unlock_tables(THD *thd, MYSQL_LOCK *sql_lock);
 void mysql_unlock_read_tables(THD *thd, MYSQL_LOCK *sql_lock);
 void mysql_unlock_some_tables(THD *thd, TABLE **table,uint count);
-void mysql_lock_remove(THD *thd, MYSQL_LOCK *locked,TABLE *table);
+void mysql_lock_remove(THD *thd, MYSQL_LOCK *locked,TABLE *table,
+                       bool always_unlock);
 void mysql_lock_abort(THD *thd, TABLE *table, bool upgrade_lock);
 void mysql_lock_downgrade_write(THD *thd, TABLE *table,
                                 thr_lock_type new_lock_type);
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 6d22047b9db..536d3d95933 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -6320,7 +6320,8 @@ The minimum value for this variable is 4096.",
     (uchar**) &opt_date_time_formats[MYSQL_TIMESTAMP_TIME],
     0, GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
   {"tmp_table_size", OPT_TMP_TABLE_SIZE,
-   "If an in-memory temporary table exceeds this size, MySQL will automatically convert it to an on-disk MyISAM table.",
+   "If an internal in-memory temporary table exceeds this size, MySQL will"
+   " automatically convert it to an on-disk MyISAM table.",
    (uchar**) &global_system_variables.tmp_table_size,
    (uchar**) &max_system_variables.tmp_table_size, 0, GET_ULL,
    REQUIRED_ARG, 16*1024*1024L, 1024, MAX_MEM_TABLE_SIZE, 0, 1, 0},
@@ -7626,11 +7627,16 @@ get_one_option(int optid, const struct my_option *opt __attribute__((unused)),
 #endif
   case OPT_MYISAM_RECOVER:
   {
-    if (!argument || !argument[0])
+    if (!argument)
     {
       myisam_recover_options=    HA_RECOVER_DEFAULT;
       myisam_recover_options_str= myisam_recover_typelib.type_names[0];
     }
+    else if (!argument[0])
+    {
+      myisam_recover_options= HA_RECOVER_NONE;
+      myisam_recover_options_str= "OFF";
+    }
     else
     {
       myisam_recover_options_str=argument;
diff --git a/sql/set_var.cc b/sql/set_var.cc
index 7f3e808090d..a18d45627dc 100644
--- a/sql/set_var.cc
+++ b/sql/set_var.cc
@@ -2053,21 +2053,15 @@ end:
 
 bool sys_var_log_state::update(THD *thd, set_var *var)
 {
-  bool res= 0;
+  bool res;
   pthread_mutex_lock(&LOCK_global_system_variables);
   if (!var->save_result.ulong_value)
-    logger.deactivate_log_handler(thd, log_type);
-  else
   {
-    if ((res= logger.activate_log_handler(thd, log_type)))
-    {
-      my_error(ER_CANT_ACTIVATE_LOG, MYF(0),
-               log_type == QUERY_LOG_GENERAL ? "general" :
-               "slow query");
-      goto err;
-    }
+    logger.deactivate_log_handler(thd, log_type);
+    res= false;
   }
-err:
+  else
+    res= logger.activate_log_handler(thd, log_type);
   pthread_mutex_unlock(&LOCK_global_system_variables);
   return res;
 }
@@ -2143,7 +2137,7 @@ bool update_sys_var_str_path(THD *thd, sys_var_str *var_str,
   }
 
   pthread_mutex_lock(&LOCK_global_system_variables);
-  logger.lock();
+  logger.lock_exclusive();
 
   if (file_log && log_state)
     file_log->close(0);
@@ -2206,7 +2200,7 @@ static void sys_default_slow_log_path(THD *thd, enum_var_type type)
 bool sys_var_log_output::update(THD *thd, set_var *var)
 {
   pthread_mutex_lock(&LOCK_global_system_variables);
-  logger.lock();
+  logger.lock_exclusive();
   logger.init_slow_log(var->save_result.ulong_value);
   logger.init_general_log(var->save_result.ulong_value);
   *value= var->save_result.ulong_value;
@@ -2219,7 +2213,7 @@ bool sys_var_log_output::update(THD *thd, set_var *var)
 void sys_var_log_output::set_default(THD *thd, enum_var_type type)
 {
   pthread_mutex_lock(&LOCK_global_system_variables);
-  logger.lock();
+  logger.lock_exclusive();
   logger.init_slow_log(LOG_FILE);
   logger.init_general_log(LOG_FILE);
   *value= LOG_FILE;
@@ -2564,14 +2558,14 @@ static bool set_option_autocommit(THD *thd, set_var *var)
     {
       /* We changed to auto_commit mode */
       thd->options&= ~(ulonglong) (OPTION_BEGIN | OPTION_KEEP_LOG);
-      thd->no_trans_update.all= FALSE;
+      thd->transaction.all.modified_non_trans_table= FALSE;
       thd->server_status|= SERVER_STATUS_AUTOCOMMIT;
       if (ha_commit(thd))
 	return 1;
     }
     else
     {
-      thd->no_trans_update.all= FALSE;
+      thd->transaction.all.modified_non_trans_table= FALSE;
       thd->server_status&= ~SERVER_STATUS_AUTOCOMMIT;
     }
   }
diff --git a/sql/share/errmsg.txt b/sql/share/errmsg.txt
index 682eee06e02..966e841ccbf 100644
--- a/sql/share/errmsg.txt
+++ b/sql/share/errmsg.txt
@@ -5936,8 +5936,8 @@ ER_WARN_DEPRECATED_SYNTAX_WITH_VER
 ER_CANT_WRITE_LOCK_LOG_TABLE
         eng "You can't write-lock a log table. Only read access is possible"
         ger "Eine Log-Tabelle kann nicht schreibgesperrt werden. Es ist ohnehin nur Lesezugriff möglich"
-ER_CANT_READ_LOCK_LOG_TABLE
-        eng "You can't use usual read lock with log tables. Try READ LOCAL instead"
+ER_CANT_LOCK_LOG_TABLE
+        eng "You can't use locks with log tables."
         ger "Log-Tabellen können nicht mit normalen Lesesperren gesperrt werden. Verwenden Sie statt dessen READ LOCAL"
 ER_FOREIGN_DUPLICATE_KEY 23000 S1009
         eng "Upholding foreign key constraints for table '%.192s', entry '%-.192s', key %d would lead to a duplicate entry"
diff --git a/sql/slave.cc b/sql/slave.cc
index 2e8e3f582de..96a105b06e4 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -977,7 +977,7 @@ static int create_table_from_dump(THD* thd, MYSQL *mysql, const char* db,
 
   thd->proc_info = "Opening master dump table";
   tables.lock_type = TL_WRITE;
-  if (!open_ltable(thd, &tables, TL_WRITE))
+  if (!open_ltable(thd, &tables, TL_WRITE, 0))
   {
     sql_print_error("create_table_from_dump: could not open created table");
     goto err;
diff --git a/sql/sp.cc b/sql/sp.cc
index aed4976f839..372aa9c6780 100644
--- a/sql/sp.cc
+++ b/sql/sp.cc
@@ -406,7 +406,7 @@ db_find_routine(THD *thd, int type, sp_name *name, sp_head **sphp)
   if ((ret= db_find_routine_aux(thd, type, name, table)) != SP_OK)
     goto done;
 
-  if (table->s->fields != MYSQL_PROC_FIELD_COUNT)
+  if (table->s->fields < MYSQL_PROC_FIELD_COUNT)
   {
     ret= SP_GET_FIELD_FAILED;
     goto done;
@@ -695,7 +695,7 @@ sp_create_routine(THD *thd, int type, sp_head *sp)
     strxnmov(definer, sizeof(definer)-1, thd->lex->definer->user.str, "@",
             thd->lex->definer->host.str, NullS);
 
-    if (table->s->fields != MYSQL_PROC_FIELD_COUNT)
+    if (table->s->fields < MYSQL_PROC_FIELD_COUNT)
     {
       ret= SP_GET_FIELD_FAILED;
       goto done;
@@ -1067,7 +1067,7 @@ sp_show_status_routine(THD *thd, int type, const char *name_pattern)
   tables.db= (char*)"mysql";
   tables.table_name= tables.alias= (char*)"proc";
 
-  if (! (table= open_ltable(thd, &tables, TL_READ)))
+  if (! (table= open_ltable(thd, &tables, TL_READ, 0)))
   {
     res= SP_OPEN_TABLE_FAILED;
     goto done;
diff --git a/sql/sp_head.cc b/sql/sp_head.cc
index 9b67a89bed2..7b2073b8de3 100644
--- a/sql/sp_head.cc
+++ b/sql/sp_head.cc
@@ -349,13 +349,13 @@ sp_eval_expr(THD *thd, Field *result_field, Item **expr_item_ptr)
   
   enum_check_fields save_count_cuted_fields= thd->count_cuted_fields;
   bool save_abort_on_warning= thd->abort_on_warning;
-  bool save_no_trans_update_stmt= thd->no_trans_update.stmt;
+  bool save_stmt_modified_non_trans_table= thd->transaction.stmt.modified_non_trans_table;
 
   thd->count_cuted_fields= CHECK_FIELD_ERROR_FOR_NULL;
   thd->abort_on_warning=
     thd->variables.sql_mode &
     (MODE_STRICT_TRANS_TABLES | MODE_STRICT_ALL_TABLES);
-  thd->no_trans_update.stmt= FALSE;
+  thd->transaction.stmt.modified_non_trans_table= FALSE;
 
   /* Save the value in the field. Convert the value if needed. */
 
@@ -363,7 +363,7 @@ sp_eval_expr(THD *thd, Field *result_field, Item **expr_item_ptr)
 
   thd->count_cuted_fields= save_count_cuted_fields;
   thd->abort_on_warning= save_abort_on_warning;
-  thd->no_trans_update.stmt= save_no_trans_update_stmt;
+  thd->transaction.stmt.modified_non_trans_table= save_stmt_modified_non_trans_table;
 
   if (thd->net.report_error)
   {
@@ -858,7 +858,8 @@ int cmp_splocal_locations(Item_splocal * const *a, Item_splocal * const *b)
 
 
 /*
-  Replace thd->query{_length} with a string that one can write to the binlog.
+  Replace thd->query{_length} with a string that one can write to the binlog
+  or the query cache.
  
   SYNOPSIS
     subst_spvars()
@@ -870,7 +871,9 @@ int cmp_splocal_locations(Item_splocal * const *a, Item_splocal * const *b)
   DESCRIPTION
 
   The binlog-suitable string is produced by replacing references to SP local 
-  variables with NAME_CONST('sp_var_name', value) calls.
+  variables with NAME_CONST('sp_var_name', value) calls. To make this string
+  suitable for the query cache this function allocates some additional space
+  for the query cache flags.
  
   RETURN
     FALSE  on success
@@ -883,80 +886,89 @@ static bool
 subst_spvars(THD *thd, sp_instr *instr, LEX_STRING *query_str)
 {
   DBUG_ENTER("subst_spvars");
-  if (thd->prelocked_mode == NON_PRELOCKED && mysql_bin_log.is_open())
-  {
-    Dynamic_array<Item_splocal*> sp_vars_uses;
-    char *pbuf, *cur, buffer[512];
-    String qbuf(buffer, sizeof(buffer), &my_charset_bin);
-    int prev_pos, res;
 
-    /* Find all instances of Item_splocal used in this statement */
-    for (Item *item= instr->free_list; item; item= item->next)
-    {
-      if (item->is_splocal())
-      {
-        Item_splocal *item_spl= (Item_splocal*)item;
-        if (item_spl->pos_in_query)
-          sp_vars_uses.append(item_spl);
-      }
-    }
-    if (!sp_vars_uses.elements())
-      DBUG_RETURN(FALSE);
-      
-    /* Sort SP var refs by their occurences in the query */
-    sp_vars_uses.sort(cmp_splocal_locations);
+  Dynamic_array<Item_splocal*> sp_vars_uses;
+  char *pbuf, *cur, buffer[512];
+  String qbuf(buffer, sizeof(buffer), &my_charset_bin);
+  int prev_pos, res, buf_len;
 
-    /* 
-      Construct a statement string where SP local var refs are replaced
-      with "NAME_CONST(name, value)"
-    */
-    qbuf.length(0);
-    cur= query_str->str;
-    prev_pos= res= 0;
-    for (Item_splocal **splocal= sp_vars_uses.front(); 
-         splocal < sp_vars_uses.back(); splocal++)
+  /* Find all instances of Item_splocal used in this statement */
+  for (Item *item= instr->free_list; item; item= item->next)
+  {
+    if (item->is_splocal())
     {
-      Item *val;
-
-      char str_buffer[STRING_BUFFER_USUAL_SIZE];
-      String str_value_holder(str_buffer, sizeof(str_buffer),
-                              &my_charset_latin1);
-      String *str_value;
-      
-      /* append the text between sp ref occurences */
-      res|= qbuf.append(cur + prev_pos, (*splocal)->pos_in_query - prev_pos);
-      prev_pos= (*splocal)->pos_in_query + (*splocal)->m_name.length;
-      
-      /* append the spvar substitute */
-      res|= qbuf.append(STRING_WITH_LEN(" NAME_CONST('"));
-      res|= qbuf.append((*splocal)->m_name.str, (*splocal)->m_name.length);
-      res|= qbuf.append(STRING_WITH_LEN("',"));
-      res|= (*splocal)->fix_fields(thd, (Item **) splocal);
+      Item_splocal *item_spl= (Item_splocal*)item;
+      if (item_spl->pos_in_query)
+        sp_vars_uses.append(item_spl);
+    }
+  }
+  if (!sp_vars_uses.elements())
+    DBUG_RETURN(FALSE);
+    
+  /* Sort SP var refs by their occurences in the query */
+  sp_vars_uses.sort(cmp_splocal_locations);
 
-      if (res)
-        break;
+  /* 
+    Construct a statement string where SP local var refs are replaced
+    with "NAME_CONST(name, value)"
+  */
+  qbuf.length(0);
+  cur= query_str->str;
+  prev_pos= res= 0;
+  for (Item_splocal **splocal= sp_vars_uses.front(); 
+       splocal < sp_vars_uses.back(); splocal++)
+  {
+    Item *val;
+
+    char str_buffer[STRING_BUFFER_USUAL_SIZE];
+    String str_value_holder(str_buffer, sizeof(str_buffer),
+                            &my_charset_latin1);
+    String *str_value;
+    
+    /* append the text between sp ref occurences */
+    res|= qbuf.append(cur + prev_pos, (*splocal)->pos_in_query - prev_pos);
+    prev_pos= (*splocal)->pos_in_query + (*splocal)->len_in_query;
+    
+    /* append the spvar substitute */
+    res|= qbuf.append(STRING_WITH_LEN(" NAME_CONST('"));
+    res|= qbuf.append((*splocal)->m_name.str, (*splocal)->m_name.length);
+    res|= qbuf.append(STRING_WITH_LEN("',"));
+    res|= (*splocal)->fix_fields(thd, (Item **) splocal);
 
-      val= (*splocal)->this_item();
-      DBUG_PRINT("info", ("print 0x%lx", (long) val));
-      str_value= sp_get_item_value(thd, val, &str_value_holder);
-      if (str_value)
-        res|= qbuf.append(*str_value);
-      else
-        res|= qbuf.append(STRING_WITH_LEN("NULL"));
-      res|= qbuf.append(')');
-      if (res)
-        break;
-    }
-    res|= qbuf.append(cur + prev_pos, query_str->length - prev_pos);
     if (res)
-      DBUG_RETURN(TRUE);
+      break;
 
-    if (!(pbuf= thd->strmake(qbuf.ptr(), qbuf.length())))
-      DBUG_RETURN(TRUE);
+    val= (*splocal)->this_item();
+    DBUG_PRINT("info", ("print 0x%lx", (long) val));
+    str_value= sp_get_item_value(thd, val, &str_value_holder);
+    if (str_value)
+      res|= qbuf.append(*str_value);
+    else
+      res|= qbuf.append(STRING_WITH_LEN("NULL"));
+    res|= qbuf.append(')');
+    if (res)
+      break;
+  }
+  res|= qbuf.append(cur + prev_pos, query_str->length - prev_pos);
+  if (res)
+    DBUG_RETURN(TRUE);
 
-    thd->query= pbuf;
-    thd->query_length= qbuf.length();
+  /*
+    Allocate additional space at the end of the new query string for the
+    query_cache_send_result_to_client function.
+  */
+  buf_len= qbuf.length() + thd->db_length + 1 + QUERY_CACHE_FLAGS_SIZE + 1;
+  if ((pbuf= (char *) alloc_root(thd->mem_root, buf_len)))
+  {
+    memcpy(pbuf, qbuf.ptr(), qbuf.length());
+    pbuf[qbuf.length()]= 0;
   }
+  else
+    DBUG_RETURN(TRUE);
+
+  thd->query= pbuf;
+  thd->query_length= qbuf.length();
+
   DBUG_RETURN(FALSE);
 }
 
@@ -2535,6 +2547,13 @@ sp_lex_keeper::reset_lex_and_exec_core(THD *thd, uint *nextp,
   int res= 0;
   DBUG_ENTER("reset_lex_and_exec_core");
 
+  /* 
+    The flag is saved at the entry to the following substatement.
+    It's reset further in the common code part.
+    It's merged with the saved parent's value at the exit of this func.
+  */
+  bool parent_modified_non_trans_table= thd->transaction.stmt.modified_non_trans_table;
+  thd->transaction.stmt.modified_non_trans_table= FALSE;
   DBUG_ASSERT(!thd->derived_tables);
   DBUG_ASSERT(thd->change_list.is_empty());
   /*
@@ -2604,7 +2623,11 @@ sp_lex_keeper::reset_lex_and_exec_core(THD *thd, uint *nextp,
   /* Update the state of the active arena. */
   thd->stmt_arena->state= Query_arena::EXECUTED;
 
-
+  /*
+    Merge here with the saved parent's values
+    what is needed from the substatement gained
+  */
+  thd->transaction.stmt.modified_non_trans_table |= parent_modified_non_trans_table;
   /*
     Unlike for PS we should not call Item's destructors for newly created
     items after execution of each instruction in stored routine. This is
diff --git a/sql/sp_rcontext.cc b/sql/sp_rcontext.cc
index b94c733cb0a..60a0c962c28 100644
--- a/sql/sp_rcontext.cc
+++ b/sql/sp_rcontext.cc
@@ -37,6 +37,7 @@ sp_rcontext::sp_rcontext(sp_pcontext *root_parsing_ctx,
    m_var_items(0),
    m_return_value_fld(return_value_fld),
    m_return_value_set(FALSE),
+   in_sub_stmt(FALSE),
    m_hcount(0),
    m_hsp(0),
    m_ihsp(0),
@@ -67,6 +68,8 @@ sp_rcontext::~sp_rcontext()
 
 bool sp_rcontext::init(THD *thd)
 {
+  in_sub_stmt= thd->in_sub_stmt;
+
   if (init_var_table(thd) || init_var_items())
     return TRUE;
 
@@ -191,7 +194,7 @@ sp_rcontext::set_return_value(THD *thd, Item **return_value_item)
 */
 
 bool
-sp_rcontext::find_handler(uint sql_errno,
+sp_rcontext::find_handler(THD *thd, uint sql_errno,
                           MYSQL_ERROR::enum_warning_level level)
 {
   if (m_hfound >= 0)
@@ -200,6 +203,15 @@ sp_rcontext::find_handler(uint sql_errno,
   const char *sqlstate= mysql_errno_to_sqlstate(sql_errno);
   int i= m_hcount, found= -1;
 
+  /*
+    If this is a fatal sub-statement error, and this runtime
+    context corresponds to a sub-statement, no CONTINUE/EXIT
+    handlers from this context are applicable: try to locate one
+    in the outer scope.
+  */
+  if (thd->is_fatal_sub_stmt_error && in_sub_stmt)
+    i= 0;
+
   /* Search handlers from the latest (innermost) to the oldest (outermost) */
   while (i--)
   {
@@ -252,7 +264,7 @@ sp_rcontext::find_handler(uint sql_errno,
     */
     if (m_prev_runtime_ctx && IS_EXCEPTION_CONDITION(sqlstate) &&
         level == MYSQL_ERROR::WARN_LEVEL_ERROR)
-      return m_prev_runtime_ctx->find_handler(sql_errno, level);
+      return m_prev_runtime_ctx->find_handler(thd, sql_errno, level);
     return FALSE;
   }
   m_hfound= found;
@@ -298,7 +310,7 @@ sp_rcontext::handle_error(uint sql_errno,
     elevated_level= MYSQL_ERROR::WARN_LEVEL_ERROR;
   }
 
-  if (find_handler(sql_errno, elevated_level))
+  if (find_handler(thd, sql_errno, elevated_level))
   {
     if (elevated_level == MYSQL_ERROR::WARN_LEVEL_ERROR)
     {
diff --git a/sql/sp_rcontext.h b/sql/sp_rcontext.h
index fbf479f52aa..0104b71a648 100644
--- a/sql/sp_rcontext.h
+++ b/sql/sp_rcontext.h
@@ -125,7 +125,7 @@ class sp_rcontext : public Sql_alloc
 
   // Returns 1 if a handler was found, 0 otherwise.
   bool
-  find_handler(uint sql_errno,MYSQL_ERROR::enum_warning_level level);
+  find_handler(THD *thd, uint sql_errno,MYSQL_ERROR::enum_warning_level level);
 
   // If there is an error handler for this error, handle it and return TRUE.
   bool
@@ -236,6 +236,10 @@ private:
     during execution.
   */
   bool m_return_value_set;
+  /**
+    TRUE if the context is created for a sub-statement.
+  */
+  bool in_sub_stmt;
 
   sp_handler_t *m_handler;      // Visible handlers
   uint m_hcount;                // Stack pointer for m_handler
diff --git a/sql/sql_acl.cc b/sql/sql_acl.cc
index 67fa380d313..91f1570f653 100644
--- a/sql/sql_acl.cc
+++ b/sql/sql_acl.cc
@@ -1602,7 +1602,7 @@ bool change_password(THD *thd, const char *host, const char *user,
   }
 #endif
 
-  if (!(table= open_ltable(thd, &tables, TL_WRITE)))
+  if (!(table= open_ltable(thd, &tables, TL_WRITE, 0)))
     DBUG_RETURN(1);
 
   VOID(pthread_mutex_lock(&acl_cache->lock));
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 0cd46a7c6fb..3e594d4da4b 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -902,8 +902,7 @@ bool close_cached_tables(THD *thd, bool if_wait_for_refresh,
     bool found=0;
     for (TABLE_LIST *table= tables; table; table= table->next_local)
     {
-      if ((!table->table || !table->table->s->log_table) &&
-          remove_table_from_cache(thd, table->db, table->table_name,
+      if (remove_table_from_cache(thd, table->db, table->table_name,
                                   RTFC_OWNED_BY_THD_FLAG))
 	found=1;
     }
@@ -951,8 +950,7 @@ bool close_cached_tables(THD *thd, bool if_wait_for_refresh,
           are employed by CREATE TABLE as in this case table simply does not
           exist yet.
         */
-	if (!table->s->log_table &&
-            (table->needs_reopen_or_name_lock() && table->db_stat))
+	if (table->needs_reopen_or_name_lock() && table->db_stat)
 	{
 	  found=1;
           DBUG_PRINT("signal", ("Waiting for COND_refresh"));
@@ -1679,8 +1677,28 @@ TABLE *find_temporary_table(THD *thd, TABLE_LIST *table_list)
 }
 
 
-/*
-  Close temporary table and unlink from thd->temporary tables
+/**
+  Drop a temporary table.
+
+  Try to locate the table in the list of thd->temporary_tables.
+  If the table is found:
+   - if the table is in thd->locked_tables, unlock it and
+     remove it from the list of locked tables. Currently only transactional
+     temporary tables are present in the locked_tables list.
+   - Close the temporary table, remove its .FRM
+   - remove the table from the list of temporary tables
+
+  This function is used to drop user temporary tables, as well as
+  internal tables created in CREATE TEMPORARY TABLE ... SELECT
+  or ALTER TABLE. Even though part of the work done by this function
+  is redundant when the table is internal, as long as we
+  link both internal and user temporary tables into the same
+  thd->temporary_tables list, it's impossible to tell here whether
+  we're dealing with an internal or a user temporary table.
+
+  @retval TRUE   the table was not found in the list of temporary tables
+                 of this thread
+  @retval FALSE  the table was found and dropped successfully.
 */
 
 bool close_temporary_table(THD *thd, TABLE_LIST *table_list)
@@ -1689,6 +1707,11 @@ bool close_temporary_table(THD *thd, TABLE_LIST *table_list)
 
   if (!(table= find_temporary_table(thd, table_list)))
     return 1;
+  /*
+    If LOCK TABLES list is not empty and contains this table,
+    unlock the table and remove the table from this list.
+  */
+  mysql_lock_remove(thd, thd->locked_tables, table, FALSE);
   close_temporary_table(thd, table, 1, 1);
   return 0;
 }
@@ -1835,7 +1858,7 @@ void unlink_open_table(THD *thd, TABLE *find, bool unlock)
 	!memcmp(list->s->table_cache_key.str, key, key_length))
     {
       if (unlock && thd->locked_tables)
-	mysql_lock_remove(thd, thd->locked_tables,list);
+        mysql_lock_remove(thd, thd->locked_tables, list, TRUE);
       VOID(hash_delete(&open_cache,(uchar*) list)); // Close table
     }
     else
@@ -1861,8 +1884,13 @@ void unlink_open_table(THD *thd, TABLE *find, bool unlock)
 
     @note This routine assumes that table to be closed is open only
           by calling thread so we needn't wait until other threads
-          will close the table. It also assumes that table to be
-          dropped is already unlocked.
+          will close the table. Also unless called under implicit or
+          explicit LOCK TABLES mode it assumes that table to be
+          dropped is already unlocked. In the former case it will
+          also remove lock on the table. But one should not rely on
+          this behaviour as it may change in future.
+          Currently, however, this function is never called for a
+          table that was locked with LOCK TABLES.
 */
 
 void drop_open_table(THD *thd, TABLE *table, const char *db_name,
@@ -2277,7 +2305,7 @@ TABLE *open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
 	}
 	table->query_id= thd->query_id;
 	table->clear_query_id= 1;
-	thd->tmp_table_used= 1;
+	thd->thread_specific_used= TRUE;
         DBUG_PRINT("info",("Using temporary table"));
         goto reset;
       }
@@ -2468,8 +2496,7 @@ TABLE *open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
                                  &state))
   {
     /*
-      Here we flush tables marked for flush. However we never flush log
-      tables here. They are flushed only on FLUSH LOGS.
+      Here we flush tables marked for flush.
       Normally, table->s->version contains the value of
       refresh_version from the moment when this table was
       (re-)opened and added to the cache.
@@ -2486,7 +2513,7 @@ TABLE *open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
       c1: name lock t2; -- blocks
       c2: open t1; -- blocks
     */
-    if (table->needs_reopen_or_name_lock() && !table->s->log_table)
+    if (table->needs_reopen_or_name_lock())
     {
       DBUG_PRINT("note",
                  ("Found table '%s.%s' with different refresh version",
@@ -2840,7 +2867,7 @@ void close_data_files_and_morph_locks(THD *thd, const char *db,
 	!strcmp(table->s->db.str, db))
     {
       if (thd->locked_tables)
-        mysql_lock_remove(thd, thd->locked_tables, table);
+        mysql_lock_remove(thd, thd->locked_tables, table, TRUE);
       table->open_placeholder= 1;
       close_handle_and_leave_table_as_lock(table);
     }
@@ -2962,10 +2989,9 @@ void close_old_data_files(THD *thd, TABLE *table, bool morph_locks,
   for (; table ; table=table->next)
   {
     /*
-      Reopen marked for flush. But close log tables. They are flushed only
-      explicitly on FLUSH LOGS
+      Reopen marked for flush.
     */
-    if (table->needs_reopen_or_name_lock() && !table->s->log_table)
+    if (table->needs_reopen_or_name_lock())
     {
       found=1;
       if (table->db_stat)
@@ -2979,7 +3005,7 @@ void close_old_data_files(THD *thd, TABLE *table, bool morph_locks,
             instances of this table.
           */
           mysql_lock_abort(thd, table, TRUE);
-          mysql_lock_remove(thd, thd->locked_tables, table);
+          mysql_lock_remove(thd, thd->locked_tables, table, TRUE);
           /*
             We want to protect the table from concurrent DDL operations
             (like RENAME TABLE) until we will re-open and re-lock it.
@@ -3012,10 +3038,6 @@ void close_old_data_files(THD *thd, TABLE *table, bool morph_locks,
   Wait until all threads has closed the tables in the list
   We have also to wait if there is thread that has a lock on this table even
   if the table is closed
-  NOTE: log tables are handled differently by the logging routines.
-        E.g. general_log is always opened and locked by the logger
-        and the table handler used by the logger, will be skipped by
-        this check.
 */
 
 bool table_is_used(TABLE *table, bool wait_for_name_lock)
@@ -3034,10 +3056,10 @@ bool table_is_used(TABLE *table, bool wait_for_name_lock)
          search= (TABLE*) hash_next(&open_cache, (uchar*) key,
                                     key_length, &state))
     {
-      DBUG_PRINT("info", ("share: 0x%lx  locked_by_logger: %d "
+      DBUG_PRINT("info", ("share: 0x%lx  "
                           "open_placeholder: %d  locked_by_name: %d "
                           "db_stat: %u  version: %lu",
-                          (ulong) search->s, search->locked_by_logger,
+                          (ulong) search->s,
                           search->open_placeholder, search->locked_by_name,
                           search->db_stat,
                           search->s->version));
@@ -3049,12 +3071,9 @@ bool table_is_used(TABLE *table, bool wait_for_name_lock)
         - If we are in flush table and we didn't execute the flush
         - If the table engine is open and it's an old version
         (We must wait until all engines are shut down to use the table)
-        However we fo not wait if we encountered a table, locked by the logger.
-        Log tables are managed separately by logging routines.
       */
-      if (!search->locked_by_logger &&
-          (search->locked_by_name && wait_for_name_lock ||
-           (search->is_name_opened() && search->needs_reopen_or_name_lock())))
+      if ( (search->locked_by_name && wait_for_name_lock) ||
+           (search->is_name_opened() && search->needs_reopen_or_name_lock()))
         DBUG_RETURN(1);
     }
   } while ((table=table->next));
@@ -3131,7 +3150,7 @@ TABLE *drop_locked_tables(THD *thd,const char *db, const char *table_name)
     if (!strcmp(table->s->table_name.str, table_name) &&
 	!strcmp(table->s->db.str, db))
     {
-      mysql_lock_remove(thd, thd->locked_tables,table);
+      mysql_lock_remove(thd, thd->locked_tables, table, TRUE);
       if (!found)
       {
         found= table;
@@ -3766,6 +3785,7 @@ static bool check_lock_and_start_stmt(THD *thd, TABLE *table,
     thd			Thread handler
     table_list		Table to open is first table in this list
     lock_type		Lock to use for open
+    lock_flags          Flags passed to mysql_lock_table
 
   NOTE
     This function don't do anything like SP/SF/views/triggers analysis done
@@ -3781,7 +3801,8 @@ static bool check_lock_and_start_stmt(THD *thd, TABLE *table,
       table_list->table		table
 */
 
-TABLE *open_ltable(THD *thd, TABLE_LIST *table_list, thr_lock_type lock_type)
+TABLE *open_ltable(THD *thd, TABLE_LIST *table_list, thr_lock_type lock_type,
+                   uint lock_flags)
 {
   TABLE *table;
   bool refresh;
@@ -3797,13 +3818,6 @@ TABLE *open_ltable(THD *thd, TABLE_LIST *table_list, thr_lock_type lock_type)
 
   if (table)
   {
-#if defined( __WIN__)
-    /* Win32 can't drop a file that is open */
-    if (lock_type == TL_WRITE_ALLOW_READ)
-    {
-      lock_type= TL_WRITE;
-    }
-#endif /* __WIN__ */
     table_list->lock_type= lock_type;
     table_list->table=	   table;
     table->grant= table_list->grant;
@@ -3816,8 +3830,8 @@ TABLE *open_ltable(THD *thd, TABLE_LIST *table_list, thr_lock_type lock_type)
     {
       DBUG_ASSERT(thd->lock == 0);	// You must lock everything at once
       if ((table->reginfo.lock_type= lock_type) != TL_UNLOCK)
-	if (! (thd->lock= mysql_lock_tables(thd, &table_list->table, 1, 0,
-                                            &refresh)))
+	if (! (thd->lock= mysql_lock_tables(thd, &table_list->table, 1,
+                                            lock_flags, &refresh)))
 	  table= 0;
     }
   }
@@ -4156,11 +4170,6 @@ int lock_tables(THD *thd, TABLE_LIST *tables, uint count, bool *need_reopen)
     DBUG_ASSERT(thd->lock == 0);	// You must lock everything at once
     TABLE **start,**ptr;
     uint lock_flag= MYSQL_LOCK_NOTIFY_IF_NEED_REOPEN;
-    
-    /* Ignore GLOBAL READ LOCK and GLOBAL READ_ONLY if called from a logger */
-    if (logger.is_privileged_thread(thd))
-      lock_flag|= (MYSQL_LOCK_IGNORE_GLOBAL_READ_LOCK |
-                   MYSQL_LOCK_IGNORE_GLOBAL_READ_ONLY);
 
     if (!(ptr=start=(TABLE**) thd->alloc(sizeof(TABLE*)*count)))
       DBUG_RETURN(-1);
@@ -4505,8 +4514,7 @@ find_field_in_view(THD *thd, TABLE_LIST *table_list,
               table_list->alias, name, item_name, (ulong) ref));
   Field_iterator_view field_it;
   field_it.set(table_list);
-  Query_arena *arena, backup;  
-  LINT_INIT(arena);
+  Query_arena *arena= 0, backup;  
   
   DBUG_ASSERT(table_list->schema_table_reformed ||
               (ref != 0 && table_list->view != 0));
@@ -4515,14 +4523,14 @@ find_field_in_view(THD *thd, TABLE_LIST *table_list,
     if (!my_strcasecmp(system_charset_info, field_it.name(), name))
     {
       // in PS use own arena or data will be freed after prepare
-      if (register_tree_change)
+      if (register_tree_change && thd->stmt_arena->is_stmt_prepare_or_first_sp_execute())
         arena= thd->activate_stmt_arena_if_needed(&backup);
       /*
         create_item() may, or may not create a new Item, depending on
         the column reference. See create_view_field() for details.
       */
       Item *item= field_it.create_item(thd);
-      if (register_tree_change && arena)
+      if (arena)
         thd->restore_active_arena(arena, &backup);
       
       if (!item)
@@ -7189,7 +7197,6 @@ bool remove_table_from_cache(THD *thd, const char *db, const char *table_name,
       else if (in_use != thd)
       {
         DBUG_PRINT("info", ("Table was in use by other thread"));
-        in_use->some_tables_deleted=1;
         if (table->is_name_opened())
         {
           DBUG_PRINT("info", ("Found another active instance of the table"));
@@ -7625,7 +7632,7 @@ open_system_tables_for_read(THD *thd, TABLE_LIST *table_list,
     if (!table)
       goto error;
 
-    DBUG_ASSERT(table->s->system_table);
+    DBUG_ASSERT(table->s->table_category == TABLE_CATEGORY_SYSTEM);
 
     table->use_all_columns();
     table->reginfo.lock_type= tables->lock_type;
@@ -7692,12 +7699,92 @@ open_system_table_for_update(THD *thd, TABLE_LIST *one_table)
 {
   DBUG_ENTER("open_system_table_for_update");
 
-  TABLE *table= open_ltable(thd, one_table, one_table->lock_type);
+  TABLE *table= open_ltable(thd, one_table, one_table->lock_type, 0);
   if (table)
   {
-    DBUG_ASSERT(table->s->system_table);
+    DBUG_ASSERT(table->s->table_category == TABLE_CATEGORY_SYSTEM);
     table->use_all_columns();
   }
 
   DBUG_RETURN(table);
 }
+
+/**
+  Open a performance schema table.
+  Opening such tables is performed internally in the server
+  implementation, and is a 'nested' open, since some tables
+  might be already opened by the current thread.
+  The thread context before this call is saved, and is restored
+  when calling close_performance_schema_table().
+  @param thd The current thread
+  @param one_table Performance schema table to open
+  @param backup [out] Temporary storage used to save the thread context
+*/
+TABLE *
+open_performance_schema_table(THD *thd, TABLE_LIST *one_table,
+                              Open_tables_state *backup)
+{
+  uint flags= ( MYSQL_LOCK_IGNORE_GLOBAL_READ_LOCK
+              | MYSQL_LOCK_IGNORE_GLOBAL_READ_ONLY
+              | MYSQL_LOCK_PERF_SCHEMA );
+
+  DBUG_ENTER("open_performance_schema_table");
+
+  thd->reset_n_backup_open_tables_state(backup);
+
+  TABLE *table= open_ltable(thd, one_table, one_table->lock_type, flags);
+  if (table)
+  {
+    DBUG_ASSERT(table->s->table_category == TABLE_CATEGORY_PERFORMANCE);
+    /* Make sure all columns get assigned to a default value */
+    table->use_all_columns();
+    table->no_replicate= 1;
+  }
+
+  DBUG_RETURN(table);
+}
+
+/**
+  Close a performance schema table.
+  The last table opened by open_performance_schema_table()
+  is closed, then the thread context is restored.
+  @param thd The current thread
+  @param backup [in] the context to restore.
+*/
+void close_performance_schema_table(THD *thd, Open_tables_state *backup)
+{
+  bool found_old_table;
+
+  if (thd->lock)
+  {
+    /*
+      Note:
+      We do not create explicitly a separate transaction for the
+      performance table I/O, but borrow the current transaction.
+      lock + unlock will autocommit the change done in the
+      performance schema table: this is the expected result.
+      The current transaction should not be affected by this code.
+      TODO: Note that if a transactional engine is used for log tables,
+      this code will need to be revised, as a separate transaction
+      might be needed.
+    */
+    mysql_unlock_tables(thd, thd->lock);
+    thd->lock= 0;
+  }
+
+  safe_mutex_assert_not_owner(&LOCK_open);
+
+  pthread_mutex_lock(&LOCK_open);
+
+  found_old_table= false;
+  while (thd->open_tables)
+    found_old_table|= close_thread_table(thd, &thd->open_tables);
+
+  if (found_old_table)
+    broadcast_refresh();
+
+  pthread_mutex_unlock(&LOCK_open);
+
+  thd->restore_backup_open_tables_state(backup);
+}
+
diff --git a/sql/sql_cache.cc b/sql/sql_cache.cc
index cab9ef94d6d..39a7aebcc5d 100644
--- a/sql/sql_cache.cc
+++ b/sql/sql_cache.cc
@@ -878,10 +878,14 @@ ulong Query_cache::resize(ulong query_cache_size_arg)
   query_cache_size= query_cache_size_arg;
   ulong new_query_cache_size= init_cache();
 
-  DBUG_EXECUTE("check_querycache",check_integrity(0););
-
   STRUCT_LOCK(&structure_guard_mutex);
   m_cache_status= Query_cache::NO_FLUSH_IN_PROGRESS;
+  /*
+    Must not call check_integrity() with
+    m_cache_status != Query_cache::NO_FLUSH_IN_PROGRESS.
+    It would wait forever.
+  */
+  DBUG_EXECUTE("check_querycache",check_integrity(1););
   pthread_cond_signal(&COND_cache_status_changed);
   STRUCT_UNLOCK(&structure_guard_mutex);
 
@@ -4025,6 +4029,10 @@ my_bool Query_cache::check_integrity(bool locked)
   Query_cache_block * block = first_block;
   do
   {
+    /* When checking at system start, there is no block. */
+    if (!block)
+      break;
+
     DBUG_PRINT("qcache", ("block 0x%lx, type %u...", 
 			  (ulong) block, (uint) block->type));  
     // Check allignment
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index da975ee3103..0135acddd8c 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -367,6 +367,8 @@ THD::THD()
    stmt_depends_on_first_successful_insert_id_in_prev_stmt(FALSE),
    global_read_lock(0),
    is_fatal_error(0),
+   transaction_rollback_request(0),
+   is_fatal_sub_stmt_error(0),
    rand_used(0),
    time_zone_used(0),
    in_lock_tables(0),
@@ -394,12 +396,13 @@ THD::THD()
   count_cuted_fields= CHECK_FIELD_IGNORE;
   killed= NOT_KILLED;
   db_length= col_access=0;
-  query_error= tmp_table_used= 0;
+  query_error= thread_specific_used= FALSE;
   hash_clear(&handler_tables_hash);
   tmp_table=0;
   used_tables=0;
   cuted_fields= sent_row_count= row_count= 0L;
   limit_found_rows= 0;
+  row_count_func= -1;
   statement_id_counter= 0UL;
 #ifdef ERROR_INJECT_SUPPORT
   error_inject_value= 0UL;
@@ -582,7 +585,7 @@ void THD::init(void)
   if (variables.sql_mode & MODE_NO_BACKSLASH_ESCAPES)
     server_status|= SERVER_STATUS_NO_BACKSLASH_ESCAPES;
   options= thd_startup_options;
-  no_trans_update.stmt= no_trans_update.all= FALSE;
+  transaction.all.modified_non_trans_table= transaction.stmt.modified_non_trans_table= FALSE;
   open_options=ha_open_options;
   update_lock_default= (variables.low_priority_updates ?
 			TL_WRITE_LOW_PRIORITY :
@@ -1303,7 +1306,7 @@ void select_send::abort()
 {
   DBUG_ENTER("select_send::abort");
   if (status && thd->spcont &&
-      thd->spcont->find_handler(thd->net.last_errno,
+      thd->spcont->find_handler(thd, thd->net.last_errno,
                                 MYSQL_ERROR::WARN_LEVEL_ERROR))
   {
     /*
@@ -2573,13 +2576,18 @@ extern "C" int thd_slave_thread(const MYSQL_THD thd)
 
 extern "C" int thd_non_transactional_update(const MYSQL_THD thd)
 {
-  return(thd->no_trans_update.all);
+  return(thd->transaction.all.modified_non_trans_table);
 }
 
 extern "C" int thd_binlog_format(const MYSQL_THD thd)
 {
   return (int) thd->variables.binlog_format;
 }
+
+extern "C" void thd_mark_transaction_to_rollback(MYSQL_THD thd, bool all)
+{
+  mark_transaction_to_rollback(thd, all);
+}
 #endif // INNODB_COMPATIBILITY_HOOKS */
 
 /****************************************************************************
@@ -2681,6 +2689,13 @@ void THD::restore_sub_statement_state(Sub_statement_state *backup)
   limit_found_rows= backup->limit_found_rows;
   sent_row_count=   backup->sent_row_count;
   client_capabilities= backup->client_capabilities;
+  /*
+    If we've left sub-statement mode, reset the fatal error flag.
+    Otherwise keep the current value, to propagate it up the sub-statement
+    stack.
+  */
+  if (!in_sub_stmt)
+    is_fatal_sub_stmt_error= FALSE;
 
   if ((options & OPTION_BIN_LOG) && is_update_query(lex->sql_command) &&
     !current_stmt_binlog_row_based)
@@ -2695,6 +2710,18 @@ void THD::restore_sub_statement_state(Sub_statement_state *backup)
 }
 
 
+/**
+  Mark transaction to rollback and mark error as fatal to a sub-statement.
+
+  @param  thd   Thread handle
+  @param  all   TRUE <=> rollback main transaction.
+*/
+
+void mark_transaction_to_rollback(THD *thd, bool all)
+{
+  thd->is_fatal_sub_stmt_error= TRUE;
+  thd->transaction_rollback_request= all;
+}
 /***************************************************************************
   Handling of XA id cacheing
 ***************************************************************************/
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 71c13e001ee..996009f3f5d 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -765,13 +765,25 @@ enum prelocked_mode_type {NON_PRELOCKED= 0, PRELOCKED= 1,
 class Open_tables_state
 {
 public:
-  /*
-    open_tables - list of regular tables in use by this thread
-    temporary_tables - list of temp tables in use by this thread
-    handler_tables - list of tables that were opened with HANDLER OPEN
-     and are still in use by this thread
+  /**
+    List of regular tables in use by this thread. Contains temporary and
+    base tables that were opened with @see open_tables().
+  */
+  TABLE *open_tables;
+  /**
+    List of temporary tables used by this thread. Contains user-level
+    temporary tables, created with CREATE TEMPORARY TABLE, and
+    internal temporary tables, created, e.g., to resolve a SELECT,
+    or for an intermediate table used in ALTER.
+    XXX Why are internal temporary tables added to this list?
   */
-  TABLE *open_tables, *temporary_tables, *handler_tables, *derived_tables;
+  TABLE *temporary_tables;
+  /**
+    List of tables that were opened with HANDLER OPEN and are
+    still in use by this thread.
+  */
+  TABLE *handler_tables;
+  TABLE *derived_tables;
   /*
     During a MySQL session, one can lock tables in two modes: automatic
     or manual. In automatic mode all necessary tables are locked just before
@@ -1399,20 +1411,44 @@ public:
   bool       current_stmt_binlog_row_based;
   bool	     locked, some_tables_deleted;
   bool       last_cuted_field;
-  bool	     no_errors, password, is_fatal_error;
+  bool	     no_errors, password;
+  /**
+    Set to TRUE if execution of the current compound statement
+    can not continue. In particular, disables activation of
+    CONTINUE or EXIT handlers of stored routines.
+    Reset in the end of processing of the current user request, in
+    @see mysql_reset_thd_for_next_command().
+  */
+  bool is_fatal_error;
+  /**
+    Set by a storage engine to request the entire
+    transaction (that possibly spans multiple engines) to
+    rollback. Reset in ha_rollback.
+  */
+  bool       transaction_rollback_request;
+  /**
+    TRUE if we are in a sub-statement and the current error can
+    not be safely recovered until we left the sub-statement mode.
+    In particular, disables activation of CONTINUE and EXIT
+    handlers inside sub-statements. E.g. if it is a deadlock
+    error and requires a transaction-wide rollback, this flag is
+    raised (traditionally, MySQL first has to close all the reads
+    via @see handler::ha_index_or_rnd_end() and only then perform
+    the rollback).
+    Reset to FALSE when we leave the sub-statement mode.
+  */
+  bool       is_fatal_sub_stmt_error;
   bool	     query_start_used, rand_used, time_zone_used;
   /* for IS NULL => = last_insert_id() fix in remove_eq_conds() */
   bool       substitute_null_with_insert_id;
   bool	     in_lock_tables;
   bool       query_error, bootstrap, cleanup_done;
-  bool	     tmp_table_used;
+  
+  /**  is set if some thread specific value(s) used in a statement. */
+  bool       thread_specific_used;
   bool	     charset_is_system_charset, charset_is_collation_connection;
   bool       charset_is_character_set_filesystem;
   bool       enable_slow_log;   /* enable slow log for current statement */
-  struct {
-    bool all:1;
-    bool stmt:1;
-  } no_trans_update;
   bool	     abort_on_warning;
   bool 	     got_warning;       /* Set on call to push_warning() */
   bool	     no_warnings_for_error; /* no warnings on call to my_error() */
@@ -1699,7 +1735,7 @@ public:
   inline bool really_abort_on_warning()
   {
     return (abort_on_warning &&
-            (!no_trans_update.stmt ||
+            (!transaction.stmt.modified_non_trans_table ||
              (variables.sql_mode & MODE_STRICT_ALL_TABLES)));
   }
   void set_status_var_init();
@@ -2438,10 +2474,13 @@ public:
 #define CF_HAS_ROW_COUNT	2
 #define CF_STATUS_COMMAND	4
 #define CF_SHOW_TABLE_COMMAND	8
+#define CF_WRITE_LOGS_COMMAND  16
 
 /* Functions in sql_class.cc */
 
 void add_to_status(STATUS_VAR *to_var, STATUS_VAR *from_var);
 void add_diff_to_status(STATUS_VAR *to_var, STATUS_VAR *from_var,
                         STATUS_VAR *dec_var);
+void mark_transaction_to_rollback(THD *thd, bool all);
+
 #endif /* MYSQL_SERVER */
diff --git a/sql/sql_delete.cc b/sql/sql_delete.cc
index 7c868092921..37f1325228d 100644
--- a/sql/sql_delete.cc
+++ b/sql/sql_delete.cc
@@ -341,6 +341,9 @@ cleanup:
   delete select;
   transactional_table= table->file->has_transactions();
 
+  if (!transactional_table && deleted > 0)
+    thd->transaction.stmt.modified_non_trans_table= TRUE;
+  
   /* See similar binlogging code in sql_update.cc, for comments */
   if ((error < 0) || (deleted && !transactional_table))
   {
@@ -364,9 +367,10 @@ cleanup:
 	error=1;
       }
     }
-    if (!transactional_table)
-      thd->no_trans_update.all= TRUE;
+    if (thd->transaction.stmt.modified_non_trans_table)
+      thd->transaction.all.modified_non_trans_table= TRUE;
   }
+  DBUG_ASSERT(transactional_table || !deleted || thd->transaction.stmt.modified_non_trans_table);
   free_underlaid_joins(thd, select_lex);
   if (transactional_table)
   {
@@ -677,20 +681,22 @@ bool multi_delete::send_data(List<Item> &values)
       if (table->triggers &&
           table->triggers->process_triggers(thd, TRG_EVENT_DELETE,
                                             TRG_ACTION_BEFORE, FALSE))
-	DBUG_RETURN(1);
+        DBUG_RETURN(1);
       table->status|= STATUS_DELETED;
       if (!(error=table->file->ha_delete_row(table->record[0])))
       {
-	deleted++;
+        deleted++;
+        if (!table->file->has_transactions())
+          thd->transaction.stmt.modified_non_trans_table= TRUE;
         if (table->triggers &&
             table->triggers->process_triggers(thd, TRG_EVENT_DELETE,
                                               TRG_ACTION_AFTER, FALSE))
-	  DBUG_RETURN(1);
+          DBUG_RETURN(1);
       }
       else
       {
-	table->file->print_error(error,MYF(0));
-	DBUG_RETURN(1);
+        table->file->print_error(error,MYF(0));
+        DBUG_RETURN(1);
       }
     }
     else
@@ -740,6 +746,7 @@ void multi_delete::send_error(uint errcode,const char *err)
     error= 1;
     send_eof();
   }
+  DBUG_ASSERT(!normal_tables || !deleted || thd->transaction.stmt.modified_non_trans_table);
   DBUG_VOID_RETURN;
 }
 
@@ -768,6 +775,7 @@ int multi_delete::do_deletes()
   for (; table_being_deleted;
        table_being_deleted= table_being_deleted->next_local, counter++)
   { 
+    ha_rows last_deleted= deleted;
     TABLE *table = table_being_deleted->table;
     if (tempfiles[counter]->get(table))
     {
@@ -814,6 +822,8 @@ int multi_delete::do_deletes()
         table->file->print_error(local_error,MYF(0));
       }
     }
+    if (last_deleted != deleted && !table->file->has_transactions())
+      thd->transaction.stmt.modified_non_trans_table= TRUE;
     end_read_record(&info);
     if (thd->killed && !local_error)
       local_error= 1;
@@ -852,7 +862,6 @@ bool multi_delete::send_eof()
   {
     query_cache_invalidate3(thd, delete_tables, 1);
   }
-
   if ((local_error == 0) || (deleted && normal_tables))
   {
     if (mysql_bin_log.is_open())
@@ -867,9 +876,11 @@ bool multi_delete::send_eof()
 	local_error=1;  // Log write failed: roll back the SQL statement
       }
     }
-    if (!transactional_tables)
-      thd->no_trans_update.all= TRUE;
+    if (thd->transaction.stmt.modified_non_trans_table)
+      thd->transaction.all.modified_non_trans_table= TRUE;
   }
+  DBUG_ASSERT(!normal_tables || !deleted || thd->transaction.stmt.modified_non_trans_table);
+
   /* Commit or rollback the current SQL statement */
   if (transactional_tables)
     if (ha_autocommit_or_rollback(thd,local_error > 0))
@@ -906,9 +917,7 @@ bool mysql_truncate(THD *thd, TABLE_LIST *table_list, bool dont_send_ok)
   char path[FN_REFLEN];
   TABLE *table;
   bool error;
-  uint closed_log_tables= 0, lock_logger= 0;
   uint path_length;
-  uint log_type;
   DBUG_ENTER("mysql_truncate");
 
   bzero((char*) &create_info,sizeof(create_info));
@@ -960,18 +969,6 @@ bool mysql_truncate(THD *thd, TABLE_LIST *table_list, bool dont_send_ok)
       DBUG_RETURN(TRUE);
   }
 
-  log_type= check_if_log_table(table_list->db_length, table_list->db,
-                               table_list->table_name_length,
-                               table_list->table_name, 1);
-  /* close log tables in use */
-  if (log_type)
-  {
-    lock_logger= 1;
-    logger.lock();
-    logger.close_log_table(log_type, FALSE);
-    closed_log_tables= closed_log_tables | log_type;
-  }
-
   // Remove the .frm extension AIX 5.2 64-bit compiler bug (BUG#16155): this
   // crashes, replacement works.  *(path + path_length - reg_ext_length)=
   // '\0';
@@ -997,14 +994,6 @@ end:
     VOID(pthread_mutex_lock(&LOCK_open));
     unlock_table_name(thd, table_list);
     VOID(pthread_mutex_unlock(&LOCK_open));
-
-    if (opt_slow_log && (closed_log_tables & QUERY_LOG_SLOW))
-      logger.reopen_log_table(QUERY_LOG_SLOW);
-
-    if (opt_log && (closed_log_tables & QUERY_LOG_GENERAL))
-      logger.reopen_log_table(QUERY_LOG_GENERAL);
-    if (lock_logger)
-      logger.unlock();
   }
   else if (error)
   {
diff --git a/sql/sql_error.cc b/sql/sql_error.cc
index 5ebeba6109a..8bdb2e59ed5 100644
--- a/sql/sql_error.cc
+++ b/sql/sql_error.cc
@@ -137,6 +137,9 @@ MYSQL_ERROR *push_warning(THD *thd, MYSQL_ERROR::enum_warning_level level,
     level= MYSQL_ERROR::WARN_LEVEL_ERROR;
   }
 
+  if (thd->handle_error(code, level))
+    DBUG_RETURN(NULL);
+
   if (thd->spcont &&
       thd->spcont->handle_error(code, level, thd))
   {
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index b747c706f75..6c6a4721369 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -549,6 +549,7 @@ bool mysql_insert(THD *thd,TABLE_LIST *table_list,
   int error, res;
   bool transactional_table, joins_freed= FALSE;
   bool changed;
+  bool was_insert_delayed= (table_list->lock_type ==  TL_WRITE_DELAYED);
   uint value_count;
   ulong counter = 1;
   ulonglong id;
@@ -712,7 +713,6 @@ bool mysql_insert(THD *thd,TABLE_LIST *table_list,
   if (lock_type != TL_WRITE_DELAYED && !thd->prelocked_mode)
     table->file->ha_start_bulk_insert(values_list.elements);
 
-  thd->no_trans_update.stmt= FALSE;
   thd->abort_on_warning= (!ignore && (thd->variables.sql_mode &
                                        (MODE_STRICT_TRANS_TABLES |
                                         MODE_STRICT_ALL_TABLES)));
@@ -832,14 +832,16 @@ bool mysql_insert(THD *thd,TABLE_LIST *table_list,
     }
     transactional_table= table->file->has_transactions();
 
-    if ((changed= (info.copied || info.deleted || info.updated)))
+    if ((changed= (info.copied || info.deleted || info.updated)) ||
+        was_insert_delayed)
     {
       /*
         Invalidate the table in the query cache if something changed.
         For the transactional algorithm to work the invalidation must be
         before binlog writing and ha_autocommit_or_rollback
       */
-      query_cache_invalidate3(thd, table_list, 1);
+      if (changed)
+        query_cache_invalidate3(thd, table_list, 1);
       if (error <= 0 || !transactional_table)
       {
         if (mysql_bin_log.is_open())
@@ -880,10 +882,12 @@ bool mysql_insert(THD *thd,TABLE_LIST *table_list,
             error=1;
           }
         }
-        if (!transactional_table)
-          thd->no_trans_update.all= TRUE;
+        if (thd->transaction.stmt.modified_non_trans_table)
+          thd->transaction.all.modified_non_trans_table= TRUE;
       }
     }
+    DBUG_ASSERT(transactional_table || !changed || 
+                thd->transaction.stmt.modified_non_trans_table);
     if (transactional_table)
       error=ha_autocommit_or_rollback(thd,error);
     
@@ -1293,7 +1297,7 @@ static int last_uniq_key(TABLE *table,uint keynr)
     then both on update triggers will work instead. Similarly both on
     delete triggers will be invoked if we will delete conflicting records.
 
-    Sets thd->no_trans_update.stmt to TRUE if table which is updated didn't have
+    Sets thd->transaction.stmt.modified_non_trans_table to TRUE if table which is updated didn't have
     transactions.
 
   RETURN VALUE
@@ -1513,7 +1517,7 @@ int write_record(THD *thd, TABLE *table,COPY_INFO *info)
             goto err;
           info->deleted++;
           if (!table->file->has_transactions())
-            thd->no_trans_update.stmt= TRUE;
+            thd->transaction.stmt.modified_non_trans_table= TRUE;
           if (table->triggers &&
               table->triggers->process_triggers(thd, TRG_EVENT_DELETE,
                                                 TRG_ACTION_AFTER, TRUE))
@@ -1554,7 +1558,7 @@ ok_or_after_trg_err:
   if (key)
     my_safe_afree(key,table->s->max_unique_length,MAX_KEY_LENGTH);
   if (!table->file->has_transactions())
-    thd->no_trans_update.stmt= TRUE;
+    thd->transaction.stmt.modified_non_trans_table= TRUE;
   DBUG_RETURN(trg_error);
 
 err:
@@ -2254,7 +2258,7 @@ pthread_handler_t handle_delayed_insert(void *arg)
   }
 
   /* open table */
-  if (!(di->table=open_ltable(thd,&di->table_list,TL_WRITE_DELAYED)))
+  if (!(di->table=open_ltable(thd, &di->table_list, TL_WRITE_DELAYED, 0)))
   {
     thd->fatal_error();				// Abort waiting inserts
     goto err;
@@ -2919,7 +2923,6 @@ select_insert::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
     table->file->extra(HA_EXTRA_WRITE_CAN_REPLACE);
   if (info.handle_duplicates == DUP_UPDATE)
     table->file->extra(HA_EXTRA_INSERT_WITH_UPDATE);
-  thd->no_trans_update.stmt= FALSE;
   thd->abort_on_warning= (!info.ignore &&
                           (thd->variables.sql_mode &
                            (MODE_STRICT_TRANS_TABLES |
@@ -3068,6 +3071,7 @@ bool select_insert::send_eof()
   int error;
   bool const trans_table= table->file->has_transactions();
   ulonglong id;
+  bool changed;
   DBUG_ENTER("select_insert::send_eof");
   DBUG_PRINT("enter", ("trans_table=%d, table_type='%s'",
                        trans_table, table->file->table_type()));
@@ -3076,24 +3080,18 @@ bool select_insert::send_eof()
   table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
   table->file->extra(HA_EXTRA_WRITE_CANNOT_REPLACE);
 
-  if (info.copied || info.deleted || info.updated)
+  if (changed= (info.copied || info.deleted || info.updated))
   {
     /*
       We must invalidate the table in the query cache before binlog writing
       and ha_autocommit_or_rollback.
     */
     query_cache_invalidate3(thd, table, 1);
-    /*
-      Mark that we have done permanent changes if all of the below is true
-      - Table doesn't support transactions
-      - It's a normal (not temporary) table. (Changes to temporary tables
-        are not logged in RBR)
-      - We are using statement based replication
-    */
-    if (!trans_table &&
-        (!table->s->tmp_table || !thd->current_stmt_binlog_row_based))
-      thd->no_trans_update.all= TRUE;
-   }
+    if (thd->transaction.stmt.modified_non_trans_table)
+      thd->transaction.all.modified_non_trans_table= TRUE;
+  }
+  DBUG_ASSERT(trans_table || !changed || 
+              thd->transaction.stmt.modified_non_trans_table);
 
   /*
     Write to binlog before commiting transaction.  No statement will
@@ -3190,7 +3188,7 @@ void select_insert::abort() {
                             table->file->has_transactions(), FALSE);
         if (!thd->current_stmt_binlog_row_based && !table->s->tmp_table &&
             !can_rollback_data())
-          thd->no_trans_update.all= TRUE;
+          thd->transaction.all.modified_non_trans_table= TRUE;
         query_cache_invalidate3(thd, table, 1);
       }
     }
@@ -3514,7 +3512,6 @@ select_create::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
     table->file->extra(HA_EXTRA_INSERT_WITH_UPDATE);
   if (!thd->prelocked_mode)
     table->file->ha_start_bulk_insert((ha_rows) 0);
-  thd->no_trans_update.stmt= FALSE;
   thd->abort_on_warning= (!info.ignore &&
                           (thd->variables.sql_mode &
                            (MODE_STRICT_TRANS_TABLES |
diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc
index e493dc05047..055dc1e8424 100644
--- a/sql/sql_lex.cc
+++ b/sql/sql_lex.cc
@@ -1770,7 +1770,7 @@ TABLE_LIST *st_select_lex_node::add_table_to_list (THD *thd, Table_ident *table,
 						  LEX_STRING *alias,
 						  ulong table_join_options,
 						  thr_lock_type flags,
-						  List<index_hint> *hints,
+						  List<Index_hint> *hints,
                                                   LEX_STRING *option)
 {
   return 0;
@@ -2367,12 +2367,129 @@ void st_select_lex_unit::set_limit(SELECT_LEX *sl)
 
 
 /**
-  Update the parsed tree with information about triggers that
-  may be fired when executing this statement.
+  @brief Set the initial purpose of this TABLE_LIST object in the list of used
+    tables.
+
+  We need to track this information on table-by-table basis, since when this
+  table becomes an element of the pre-locked list, it's impossible to identify
+  which SQL sub-statement it has been originally used in.
+
+  E.g.:
+
+  User request:                 SELECT * FROM t1 WHERE f1();
+  FUNCTION f1():                DELETE FROM t2; RETURN 1;
+  BEFORE DELETE trigger on t2:  INSERT INTO t3 VALUES (old.a);
+
+  For this user request, the pre-locked list will contain t1, t2, t3
+  table elements, each needed for different DML.
+
+  The trigger event map is updated to reflect INSERT, UPDATE, DELETE,
+  REPLACE, LOAD DATA, CREATE TABLE .. SELECT, CREATE TABLE ..
+  REPLACE SELECT statements, and additionally ON DUPLICATE KEY UPDATE
+  clause.
 */
 
 void st_lex::set_trg_event_type_for_tables()
 {
+  uint8 new_trg_event_map= 0;
+
+  /*
+    Some auxiliary operations
+    (e.g. GRANT processing) create TABLE_LIST instances outside
+    the parser. Additionally, some commands (e.g. OPTIMIZE) change
+    the lock type for a table only after parsing is done. Luckily,
+    these do not fire triggers and do not need to pre-load them.
+    For these TABLE_LISTs set_trg_event_type is never called, and
+    trg_event_map is always empty. That means that the pre-locking
+    algorithm will ignore triggers defined on these tables, if
+    any, and the execution will either fail with an assert in
+    sql_trigger.cc or with an error that a used table was not
+    pre-locked, in case of a production build.
+
+    TODO: this usage pattern creates unnecessary module dependencies
+    and should be rewritten to go through the parser.
+    Table list instances created outside the parser in most cases
+    refer to mysql.* system tables. It is not allowed to have
+    a trigger on a system table, but keeping track of
+    initialization provides extra safety in case this limitation
+    is circumvented.
+  */
+
+  switch (sql_command) {
+  case SQLCOM_LOCK_TABLES:
+  /*
+    On a LOCK TABLE, all triggers must be pre-loaded for this TABLE_LIST
+    when opening an associated TABLE.
+  */
+    new_trg_event_map= static_cast<uint8>
+                        (1 << static_cast<int>(TRG_EVENT_INSERT)) |
+                      static_cast<uint8>
+                        (1 << static_cast<int>(TRG_EVENT_UPDATE)) |
+                      static_cast<uint8>
+                        (1 << static_cast<int>(TRG_EVENT_DELETE));
+    break;
+  /*
+    Basic INSERT. If there is an additional ON DUPLIATE KEY UPDATE
+    clause, it will be handled later in this method.
+  */
+  case SQLCOM_INSERT:                           /* fall through */
+  case SQLCOM_INSERT_SELECT:
+  /*
+    LOAD DATA ... INFILE is expected to fire BEFORE/AFTER INSERT
+    triggers.
+    If the statement also has REPLACE clause, it will be
+    handled later in this method.
+  */
+  case SQLCOM_LOAD:                             /* fall through */
+  /*
+    REPLACE is semantically equivalent to INSERT. In case
+    of a primary or unique key conflict, it deletes the old
+    record and inserts a new one. So we also may need to
+    fire ON DELETE triggers. This functionality is handled
+    later in this method.
+  */
+  case SQLCOM_REPLACE:                          /* fall through */
+  case SQLCOM_REPLACE_SELECT:
+  /*
+    CREATE TABLE ... SELECT defaults to INSERT if the table or
+    view already exists. REPLACE option of CREATE TABLE ...
+    REPLACE SELECT is handled later in this method.
+  */
+  case SQLCOM_CREATE_TABLE:
+    new_trg_event_map|= static_cast<uint8>
+                          (1 << static_cast<int>(TRG_EVENT_INSERT));
+    break;
+  /* Basic update and multi-update */
+  case SQLCOM_UPDATE:                           /* fall through */
+  case SQLCOM_UPDATE_MULTI:
+    new_trg_event_map|= static_cast<uint8>
+                          (1 << static_cast<int>(TRG_EVENT_UPDATE));
+    break;
+  /* Basic delete and multi-delete */
+  case SQLCOM_DELETE:                           /* fall through */
+  case SQLCOM_DELETE_MULTI:
+    new_trg_event_map|= static_cast<uint8>
+                          (1 << static_cast<int>(TRG_EVENT_DELETE));
+    break;
+  default:
+    break;
+  }
+
+  switch (duplicates) {
+  case DUP_UPDATE:
+    new_trg_event_map|= static_cast<uint8>
+                          (1 << static_cast<int>(TRG_EVENT_UPDATE));
+    break;
+  case DUP_REPLACE:
+    new_trg_event_map|= static_cast<uint8>
+                          (1 << static_cast<int>(TRG_EVENT_DELETE));
+    break;
+  case DUP_ERROR:
+  default:
+    break;
+  }
+
+
   /*
     Do not iterate over sub-selects, only the tables in the outermost
     SELECT_LEX can be modified, if any.
@@ -2381,7 +2498,17 @@ void st_lex::set_trg_event_type_for_tables()
 
   while (tables)
   {
-    tables->set_trg_event_type(this);
+    /*
+      This is a fast check to filter out statements that do
+      not change data, or tables  on the right side, in case of
+      INSERT .. SELECT, CREATE TABLE .. SELECT and so on.
+      Here we also filter out OPTIMIZE statement and non-updateable
+      views, for which lock_type is TL_UNLOCK or TL_READ after
+      parsing.
+    */
+    if (static_cast<int>(tables->lock_type) >=
+        static_cast<int>(TL_WRITE_ALLOW_WRITE))
+      tables->trg_event_map= new_trg_event_map;
     tables= tables->next_local;
   }
 }
@@ -2723,7 +2850,7 @@ void st_select_lex::set_index_hint_type(enum index_hint_type type,
 
 void st_select_lex::alloc_index_hints (THD *thd)
 { 
-  index_hints= new (thd->mem_root) List<index_hint>(); 
+  index_hints= new (thd->mem_root) List<Index_hint>(); 
 }
 
 
@@ -2744,7 +2871,7 @@ void st_select_lex::alloc_index_hints (THD *thd)
 bool st_select_lex::add_index_hint (THD *thd, char *str, uint length)
 {
   return index_hints->push_front (new (thd->mem_root) 
-                                 index_hint(current_index_hint_type,
+                                 Index_hint(current_index_hint_type,
                                             current_index_hint_clause,
                                             str, length));
 }
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index 9ac7f2835f0..b769930958e 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -236,7 +236,7 @@ typedef uchar index_clause_map;
                              INDEX_HINT_MASK_ORDER)
 
 /* Single element of an USE/FORCE/IGNORE INDEX list specified as a SQL hint  */
-class index_hint : public Sql_alloc
+class Index_hint : public Sql_alloc
 {
 public:
   /* The type of the hint : USE/FORCE/IGNORE */
@@ -249,7 +249,7 @@ public:
   */ 
   LEX_STRING key_name;
 
-  index_hint (enum index_hint_type type_arg, index_clause_map clause_arg,
+  Index_hint (enum index_hint_type type_arg, index_clause_map clause_arg,
               char *str, uint length) :
     type(type_arg), clause(clause_arg)
   {
@@ -441,7 +441,7 @@ public:
 					LEX_STRING *alias,
 					ulong table_options,
 					thr_lock_type flags= TL_UNLOCK,
-					List<index_hint> *hints= 0,
+					List<Index_hint> *hints= 0,
                                         LEX_STRING *option= 0);
   virtual void set_lock_for_tables(thr_lock_type lock_type) {}
 
@@ -719,7 +719,7 @@ public:
 				LEX_STRING *alias,
 				ulong table_options,
 				thr_lock_type flags= TL_UNLOCK,
-				List<index_hint> *hints= 0,
+				List<Index_hint> *hints= 0,
                                 LEX_STRING *option= 0);
   TABLE_LIST* get_table_list();
   bool init_nested_join(THD *thd);
@@ -779,9 +779,9 @@ public:
   /* make a list to hold index hints */
   void alloc_index_hints (THD *thd);
   /* read and clear the index hints */
-  List<index_hint>* pop_index_hints(void) 
+  List<Index_hint>* pop_index_hints(void) 
   {
-    List<index_hint> *hints= index_hints;
+    List<Index_hint> *hints= index_hints;
     index_hints= NULL;
     return hints;
   }
@@ -793,7 +793,7 @@ private:
   enum index_hint_type current_index_hint_type;
   index_clause_map current_index_hint_clause;
   /* a list of USE/FORCE/IGNORE INDEX */
-  List<index_hint> *index_hints;
+  List<Index_hint> *index_hints;
 };
 typedef class st_select_lex SELECT_LEX;
 
diff --git a/sql/sql_load.cc b/sql/sql_load.cc
index 3ffbdf83815..8bbe1e413b3 100644
--- a/sql/sql_load.cc
+++ b/sql/sql_load.cc
@@ -376,7 +376,6 @@ bool mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
       table->file->ha_start_bulk_insert((ha_rows) 0);
     table->copy_blobs=1;
 
-    thd->no_trans_update.stmt= FALSE;
     thd->abort_on_warning= (!ignore &&
                             (thd->variables.sql_mode &
                              (MODE_STRICT_TRANS_TABLES |
@@ -410,7 +409,6 @@ bool mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
     ha_autocommit_...
   */
   query_cache_invalidate3(thd, table_list, 0);
-
   if (error)
   {
     if (read_file_from_client)
@@ -469,8 +467,8 @@ bool mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
   sprintf(name, ER(ER_LOAD_INFO), (ulong) info.records, (ulong) info.deleted,
 	  (ulong) (info.records - info.copied), (ulong) thd->cuted_fields);
 
-  if (!transactional_table)
-    thd->no_trans_update.all= TRUE;
+  if (thd->transaction.stmt.modified_non_trans_table)
+    thd->transaction.all.modified_non_trans_table= TRUE;
 #ifndef EMBEDDED_LIBRARY
   if (mysql_bin_log.is_open())
   {
@@ -506,6 +504,8 @@ bool mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
   /* ok to client sent only after binlog write and engine commit */
   send_ok(thd, info.copied + info.deleted, 0L, name);
 err:
+  DBUG_ASSERT(transactional_table || !(info.copied || info.deleted) ||
+              thd->transaction.stmt.modified_non_trans_table);
   table->file->ha_release_auto_increment();
   if (thd->lock)
   {
@@ -552,7 +552,7 @@ read_fixed_length(THD *thd, COPY_INFO &info, TABLE_LIST *table_list,
   Item_field *sql_field;
   TABLE *table= table_list->table;
   ulonglong id;
-  bool no_trans_update_stmt, err;
+  bool err;
   DBUG_ENTER("read_fixed_length");
 
   id= 0;
@@ -580,7 +580,6 @@ read_fixed_length(THD *thd, COPY_INFO &info, TABLE_LIST *table_list,
 #ifdef HAVE_purify
     read_info.row_end[0]=0;
 #endif
-    no_trans_update_stmt= !table->file->has_transactions();
 
     restore_record(table, s->default_values);
     /*
@@ -650,7 +649,6 @@ read_fixed_length(THD *thd, COPY_INFO &info, TABLE_LIST *table_list,
     table->auto_increment_field_not_null= FALSE;
     if (err)
       DBUG_RETURN(1);
-    thd->no_trans_update.stmt= no_trans_update_stmt;
    
     /*
       We don't need to reset auto-increment field since we are restoring
@@ -685,12 +683,11 @@ read_sep_field(THD *thd, COPY_INFO &info, TABLE_LIST *table_list,
   TABLE *table= table_list->table;
   uint enclosed_length;
   ulonglong id;
-  bool no_trans_update_stmt, err;
+  bool err;
   DBUG_ENTER("read_sep_field");
 
   enclosed_length=enclosed.length();
   id= 0;
-  no_trans_update_stmt= !table->file->has_transactions();
 
   for (;;it.rewind())
   {
@@ -827,7 +824,6 @@ read_sep_field(THD *thd, COPY_INFO &info, TABLE_LIST *table_list,
       We don't need to reset auto-increment field since we are restoring
       its default value at the beginning of each loop iteration.
     */
-    thd->no_trans_update.stmt= no_trans_update_stmt;
     if (read_info.next_line())			// Skip to next line
       break;
     if (read_info.line_cuted)
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index f2a61b7f7c5..737ec320a69 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -120,7 +120,7 @@ bool end_active_trans(THD *thd)
       error=1;
   }
   thd->options&= ~(OPTION_BEGIN | OPTION_KEEP_LOG);
-  thd->no_trans_update.all= FALSE;
+  thd->transaction.all.modified_non_trans_table= FALSE;
   DBUG_RETURN(error);
 }
 
@@ -197,8 +197,8 @@ void init_update_queries(void)
 
   sql_command_flags[SQLCOM_CREATE_TABLE]=   CF_CHANGES_DATA;
   sql_command_flags[SQLCOM_CREATE_INDEX]=   CF_CHANGES_DATA;
-  sql_command_flags[SQLCOM_ALTER_TABLE]=    CF_CHANGES_DATA;
-  sql_command_flags[SQLCOM_TRUNCATE]=       CF_CHANGES_DATA;
+  sql_command_flags[SQLCOM_ALTER_TABLE]=    CF_CHANGES_DATA | CF_WRITE_LOGS_COMMAND;
+  sql_command_flags[SQLCOM_TRUNCATE]=       CF_CHANGES_DATA | CF_WRITE_LOGS_COMMAND;
   sql_command_flags[SQLCOM_DROP_TABLE]=     CF_CHANGES_DATA;
   sql_command_flags[SQLCOM_LOAD]=           CF_CHANGES_DATA;
   sql_command_flags[SQLCOM_CREATE_DB]=      CF_CHANGES_DATA;
@@ -211,7 +211,7 @@ void init_update_queries(void)
   sql_command_flags[SQLCOM_DROP_VIEW]=      CF_CHANGES_DATA;
   sql_command_flags[SQLCOM_CREATE_EVENT]=   CF_CHANGES_DATA;
   sql_command_flags[SQLCOM_ALTER_EVENT]=    CF_CHANGES_DATA;
-  sql_command_flags[SQLCOM_DROP_EVENT]=     CF_CHANGES_DATA;  
+  sql_command_flags[SQLCOM_DROP_EVENT]=     CF_CHANGES_DATA;
 
   sql_command_flags[SQLCOM_UPDATE]=	    CF_CHANGES_DATA | CF_HAS_ROW_COUNT;
   sql_command_flags[SQLCOM_UPDATE_MULTI]=   CF_CHANGES_DATA | CF_HAS_ROW_COUNT;
@@ -223,7 +223,6 @@ void init_update_queries(void)
   sql_command_flags[SQLCOM_REPLACE_SELECT]= CF_CHANGES_DATA | CF_HAS_ROW_COUNT;
 
   sql_command_flags[SQLCOM_SHOW_STATUS_PROC]= CF_STATUS_COMMAND;
-  sql_command_flags[SQLCOM_SHOW_STATUS_FUNC]= CF_STATUS_COMMAND;
   sql_command_flags[SQLCOM_SHOW_STATUS]=      CF_STATUS_COMMAND;
   sql_command_flags[SQLCOM_SHOW_DATABASES]=   CF_STATUS_COMMAND;
   sql_command_flags[SQLCOM_SHOW_TRIGGERS]=    CF_STATUS_COMMAND;
@@ -235,10 +234,36 @@ void init_update_queries(void)
   sql_command_flags[SQLCOM_SHOW_VARIABLES]=   CF_STATUS_COMMAND;
   sql_command_flags[SQLCOM_SHOW_CHARSETS]=    CF_STATUS_COMMAND;
   sql_command_flags[SQLCOM_SHOW_COLLATIONS]=  CF_STATUS_COMMAND;
-  sql_command_flags[SQLCOM_SHOW_STATUS_PROC]= CF_STATUS_COMMAND;
-
-  sql_command_flags[SQLCOM_SHOW_TABLES]=       (CF_STATUS_COMMAND |
-                                                CF_SHOW_TABLE_COMMAND);
+  sql_command_flags[SQLCOM_SHOW_NEW_MASTER]= CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_BINLOGS]= CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_SLAVE_HOSTS]= CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_BINLOG_EVENTS]= CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_COLUMN_TYPES]= CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_STORAGE_ENGINES]= CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_AUTHORS]= CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_CONTRIBUTORS]= CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_PRIVILEGES]= CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_WARNS]= CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_ERRORS]= CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_ENGINE_STATUS]= CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_ENGINE_MUTEX]= CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_ENGINE_LOGS]= CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_PROCESSLIST]= CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_GRANTS]=  CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_CREATE_DB]=  CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_CREATE]=  CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_MASTER_STAT]=  CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_SLAVE_STAT]=  CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_CREATE_PROC]=  CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_CREATE_FUNC]=  CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_CREATE_TRIGGER]=  CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_STATUS_FUNC]=  CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_PROC_CODE]=  CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_FUNC_CODE]=  CF_STATUS_COMMAND;
+  sql_command_flags[SQLCOM_SHOW_CREATE_EVENT]=  CF_STATUS_COMMAND;
+
+   sql_command_flags[SQLCOM_SHOW_TABLES]=       (CF_STATUS_COMMAND |
+                                               CF_SHOW_TABLE_COMMAND);
   sql_command_flags[SQLCOM_SHOW_TABLE_STATUS]= (CF_STATUS_COMMAND |
                                                 CF_SHOW_TABLE_COMMAND);
 
@@ -250,6 +275,14 @@ void init_update_queries(void)
   */
   sql_command_flags[SQLCOM_CALL]= 		CF_HAS_ROW_COUNT;
   sql_command_flags[SQLCOM_EXECUTE]= 		CF_HAS_ROW_COUNT;
+
+  /*
+    The following admin table operations are allowed
+    on log tables.
+  */
+  sql_command_flags[SQLCOM_REPAIR]=           CF_WRITE_LOGS_COMMAND;
+  sql_command_flags[SQLCOM_OPTIMIZE]=         CF_WRITE_LOGS_COMMAND;
+  sql_command_flags[SQLCOM_ANALYZE]=          CF_WRITE_LOGS_COMMAND;
 }
 
 
@@ -259,6 +292,17 @@ bool is_update_query(enum enum_sql_command command)
   return (sql_command_flags[command] & CF_CHANGES_DATA) != 0;
 }
 
+/**
+  Check if a sql command is allowed to write to log tables.
+  @param command The SQL command
+  @return true if writing is allowed
+*/
+bool is_log_table_write_query(enum enum_sql_command command)
+{
+  DBUG_ASSERT(command >= 0 && command <= SQLCOM_END);
+  return (sql_command_flags[command] & CF_WRITE_LOGS_COMMAND) != 0;
+}
+
 void execute_init_command(THD *thd, sys_var_str *init_command_var,
 			  rw_lock_t *var_mutex)
 {
@@ -493,7 +537,7 @@ int mysql_table_dump(THD *thd, LEX_STRING *db, char *tbl_name)
   if (lower_case_table_names)
     my_casedn_str(files_charset_info, tbl_name);
 
-  if (!(table=open_ltable(thd, table_list, TL_READ_NO_INSERT)))
+  if (!(table=open_ltable(thd, table_list, TL_READ_NO_INSERT, 0)))
     DBUG_RETURN(1);
 
   if (check_one_table_access(thd, SELECT_ACL, table_list))
@@ -553,7 +597,7 @@ int end_trans(THD *thd, enum enum_mysql_completiontype completion)
     thd->server_status&= ~SERVER_STATUS_IN_TRANS;
     res= ha_commit(thd);
     thd->options&= ~(ulong) (OPTION_BEGIN | OPTION_KEEP_LOG);
-    thd->no_trans_update.all= FALSE;
+    thd->transaction.all.modified_non_trans_table= FALSE;
     break;
   case COMMIT_RELEASE:
     do_release= 1; /* fall through */
@@ -571,7 +615,7 @@ int end_trans(THD *thd, enum enum_mysql_completiontype completion)
     if (ha_rollback(thd))
       res= -1;
     thd->options&= ~(ulong) (OPTION_BEGIN | OPTION_KEEP_LOG);
-    thd->no_trans_update.all= FALSE;
+    thd->transaction.all.modified_non_trans_table= FALSE;
     if (!res && (completion == ROLLBACK_AND_CHAIN))
       res= begin_trans(thd);
     break;
@@ -1323,7 +1367,8 @@ void log_slow_statement(THD *thd)
 	thd->variables.long_query_time ||
 	((thd->server_status &
 	  (SERVER_QUERY_NO_INDEX_USED | SERVER_QUERY_NO_GOOD_INDEX_USED)) &&
-	 opt_log_queries_not_using_indexes))
+         opt_log_queries_not_using_indexes &&
+          !(sql_command_flags[thd->lex->sql_command] & CF_STATUS_COMMAND)))
     {
       thd->status_var.long_query_count++;
       slow_log_print(thd, thd->query, thd->query_length, start_of_query);
@@ -1780,6 +1825,8 @@ mysql_execute_command(THD *thd)
 #endif
   status_var_increment(thd->status_var.com_stat[lex->sql_command]);
 
+  DBUG_ASSERT(thd->transaction.stmt.modified_non_trans_table == FALSE);
+  
   switch (lex->sql_command) {
   case SQLCOM_SHOW_EVENTS:
     if ((res= check_access(thd, EVENT_ACL, thd->lex->select_lex.db, 0, 0, 0,
@@ -3594,7 +3641,8 @@ end_with_restore_list:
         res= TRUE; // cannot happen
       else
       {
-        if (((thd->options & OPTION_KEEP_LOG) || thd->no_trans_update.all) &&
+        if (((thd->options & OPTION_KEEP_LOG) || 
+             thd->transaction.all.modified_non_trans_table) &&
             !thd->slave_thread)
           push_warning(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
                        ER_WARNING_NOT_COMPLETE_ROLLBACK,
@@ -4174,8 +4222,8 @@ create_sp_error:
     thd->transaction.xid_state.xa_state=XA_ACTIVE;
     thd->transaction.xid_state.xid.set(thd->lex->xid);
     xid_cache_insert(&thd->transaction.xid_state);
+    thd->transaction.all.modified_non_trans_table= FALSE;
     thd->options= ((thd->options & ~(OPTION_KEEP_LOG)) | OPTION_BEGIN);
-    thd->no_trans_update.all= FALSE;
     thd->server_status|= SERVER_STATUS_IN_TRANS;
     send_ok(thd);
     break;
@@ -4269,7 +4317,7 @@ create_sp_error:
       break;
     }
     thd->options&= ~(OPTION_BEGIN | OPTION_KEEP_LOG);
-    thd->no_trans_update.all= FALSE;
+    thd->transaction.all.modified_non_trans_table= FALSE;
     thd->server_status&= ~SERVER_STATUS_IN_TRANS;
     xid_cache_delete(&thd->transaction.xid_state);
     thd->transaction.xid_state.xa_state=XA_NOTR;
@@ -4300,7 +4348,7 @@ create_sp_error:
     else
       send_ok(thd);
     thd->options&= ~(OPTION_BEGIN | OPTION_KEEP_LOG);
-    thd->no_trans_update.all= FALSE;
+    thd->transaction.all.modified_non_trans_table= FALSE;
     thd->server_status&= ~SERVER_STATUS_IN_TRANS;
     xid_cache_delete(&thd->transaction.xid_state);
     thd->transaction.xid_state.xa_state=XA_NOTR;
@@ -5042,7 +5090,11 @@ bool check_merge_table_access(THD *thd, char *db,
 
 #ifndef EMBEDDED_LIBRARY
 
-#define used_stack(A,B) (long)(A > B ? A - B : B - A)
+#if STACK_DIRECTION < 0
+#define used_stack(A,B) (long) (A - B)
+#else
+#define used_stack(A,B) (long) (B - A)
+#endif
 
 #ifndef DBUG_OFF
 long max_stack_used;
@@ -5147,10 +5199,10 @@ void mysql_reset_thd_for_next_command(THD *thd)
   if (!(thd->options & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN)))
   {
     thd->options&= ~OPTION_KEEP_LOG;
-    thd->no_trans_update.all= FALSE;
+    thd->transaction.all.modified_non_trans_table= FALSE;
   }
   DBUG_ASSERT(thd->security_ctx== &thd->main_security_ctx);
-  thd->tmp_table_used= 0;
+  thd->thread_specific_used= FALSE;
   if (!thd->in_sub_stmt)
   {
     if (opt_bin_log)
@@ -5636,7 +5688,7 @@ TABLE_LIST *st_select_lex::add_table_to_list(THD *thd,
 					     LEX_STRING *alias,
 					     ulong table_options,
 					     thr_lock_type lock_type,
-					     List<index_hint> *index_hints_arg,
+					     List<Index_hint> *index_hints_arg,
                                              LEX_STRING *option)
 {
   register TABLE_LIST *ptr;
diff --git a/sql/sql_plugin.cc b/sql/sql_plugin.cc
index a44e20b8daf..0d088063462 100644
--- a/sql/sql_plugin.cc
+++ b/sql/sql_plugin.cc
@@ -1617,7 +1617,7 @@ bool mysql_install_plugin(THD *thd, const LEX_STRING *name, const LEX_STRING *dl
     DBUG_RETURN(TRUE);
 
   /* need to open before acquiring LOCK_plugin or it will deadlock */
-  if (! (table = open_ltable(thd, &tables, TL_WRITE)))
+  if (! (table = open_ltable(thd, &tables, TL_WRITE, 0)))
     DBUG_RETURN(TRUE);
 
   pthread_mutex_lock(&LOCK_plugin);
@@ -1674,7 +1674,7 @@ bool mysql_uninstall_plugin(THD *thd, const LEX_STRING *name)
   tables.table_name= tables.alias= (char *)"plugin";
 
   /* need to open before acquiring LOCK_plugin or it will deadlock */
-  if (! (table= open_ltable(thd, &tables, TL_WRITE)))
+  if (! (table= open_ltable(thd, &tables, TL_WRITE, 0)))
     DBUG_RETURN(TRUE);
 
   pthread_mutex_lock(&LOCK_plugin);
diff --git a/sql/sql_rename.cc b/sql/sql_rename.cc
index f5e1b8988f3..750bcd50479 100644
--- a/sql/sql_rename.cc
+++ b/sql/sql_rename.cc
@@ -37,7 +37,6 @@ bool mysql_rename_tables(THD *thd, TABLE_LIST *table_list, bool silent)
   TABLE_LIST *ren_table= 0;
   int to_table;
   char *rename_log_table[2]= {NULL, NULL};
-  int disable_logs= 0;
   DBUG_ENTER("mysql_rename_tables");
 
   /*
@@ -80,12 +79,6 @@ bool mysql_rename_tables(THD *thd, TABLE_LIST *table_list, bool silent)
                               ren_table->table_name, 1)))
       {
         /*
-          Log table encoutered we will need to disable and lock logs
-          for duration of rename.
-        */
-        disable_logs= TRUE;
-
-        /*
           as we use log_table_rename as an array index, we need it to start
           with 0, while QUERY_LOG_SLOW == 1 and QUERY_LOG_GENERAL == 2.
           So, we shift the value to start with 0;
@@ -136,12 +129,6 @@ bool mysql_rename_tables(THD *thd, TABLE_LIST *table_list, bool silent)
                  rename_log_table[1]);
       DBUG_RETURN(1);
     }
-
-    if (disable_logs)
-    {
-      logger.lock();
-      logger.tmp_close_log_tables(thd);
-    }
   }
 
   pthread_mutex_lock(&LOCK_open);
@@ -200,13 +187,6 @@ bool mysql_rename_tables(THD *thd, TABLE_LIST *table_list, bool silent)
   pthread_mutex_unlock(&LOCK_open);
 
 err:
-  /* enable logging back if needed */
-  if (disable_logs)
-  {
-    if (logger.reopen_log_tables())
-      error= TRUE;
-    logger.unlock();
-  }
   start_waiting_global_read_lock(thd);
   DBUG_RETURN(error);
 }
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index be6d1f74852..9ea67af1bf9 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -1134,6 +1134,7 @@ JOIN::optimize()
     order=0;					// The output has only one row
     simple_order=1;
     select_distinct= 0;                       // No need in distinct for 1 row
+    group_optimized_away= 1;
   }
 
   calc_group_buffer(this, group_list);
@@ -2454,7 +2455,7 @@ make_join_statistics(JOIN *join, TABLE_LIST *tables, COND *conds,
          no_partitions_used) &&
 	!s->dependent &&
 	(table->file->ha_table_flags() & HA_STATS_RECORDS_IS_EXACT) &&
-        !table->fulltext_searched)
+        !table->fulltext_searched && !join->no_const_tables)
     {
       set_position(join,const_count++,s,(KEYUSE*) 0);
     }
@@ -11742,7 +11743,8 @@ end_send_group(JOIN *join, JOIN_TAB *join_tab __attribute__((unused)),
   if (!join->first_record || end_of_records ||
       (idx=test_if_group_changed(join->group_fields)) >= 0)
   {
-    if (join->first_record || (end_of_records && !join->group))
+    if (join->first_record || 
+        (end_of_records && !join->group && !join->group_optimized_away))
     {
       if (join->procedure)
 	join->procedure->end_group();
@@ -12289,6 +12291,7 @@ static int test_if_order_by_key(ORDER *order, TABLE *table, uint idx,
   key_part_end=key_part+table->key_info[idx].key_parts;
   key_part_map const_key_parts=table->const_key_parts[idx];
   int reverse=0;
+  my_bool on_primary_key= FALSE;
   DBUG_ENTER("test_if_order_by_key");
 
   for (; order ; order=order->next, const_key_parts>>=1)
@@ -12303,7 +12306,30 @@ static int test_if_order_by_key(ORDER *order, TABLE *table, uint idx,
     for (; const_key_parts & 1 ; const_key_parts>>= 1)
       key_part++; 
 
-    if (key_part == key_part_end || key_part->field != field)
+    if (key_part == key_part_end)
+    {
+      /* 
+        We are at the end of the key. Check if the engine has the primary
+        key as a suffix to the secondary keys. If it has continue to check
+        the primary key as a suffix.
+      */
+      if (!on_primary_key &&
+          (table->file->ha_table_flags() & HA_PRIMARY_KEY_IN_READ_INDEX) &&
+          table->s->primary_key != MAX_KEY)
+      {
+        on_primary_key= TRUE;
+        key_part= table->key_info[table->s->primary_key].key_part;
+        key_part_end=key_part+table->key_info[table->s->primary_key].key_parts;
+        const_key_parts=table->const_key_parts[table->s->primary_key];
+
+        for (; const_key_parts & 1 ; const_key_parts>>= 1)
+          key_part++; 
+      }
+      else
+        DBUG_RETURN(0);
+    }
+
+    if (key_part->field != field)
       DBUG_RETURN(0);
 
     /* set flag to 1 if we can use read-next on key, else to -1 */
@@ -12314,7 +12340,8 @@ static int test_if_order_by_key(ORDER *order, TABLE *table, uint idx,
     reverse=flag;				// Remember if reverse
     key_part++;
   }
-  *used_key_parts= (uint) (key_part - table->key_info[idx].key_part);
+  *used_key_parts= on_primary_key ? table->key_info[idx].key_parts :
+    (uint) (key_part - table->key_info[idx].key_part);
   if (reverse == -1 && !(table->file->index_flags(idx, *used_key_parts-1, 1) &
                          HA_READ_PREV))
     reverse= 0;                                 // Index can't be used
@@ -13030,7 +13057,7 @@ remove_duplicates(JOIN *join, TABLE *entry,List<Item> &fields, Item *having)
       field_count++;
   }
 
-  if (!field_count && !(join->select_options & OPTION_FOUND_ROWS)) 
+  if (!field_count && !(join->select_options & OPTION_FOUND_ROWS) && !having) 
   {                    // only const items with no OPTION_FOUND_ROWS
     join->unit->select_limit_cnt= 1;		// Only send first row
     DBUG_RETURN(0);
@@ -13307,7 +13334,8 @@ static int
 join_init_cache(THD *thd,JOIN_TAB *tables,uint table_count)
 {
   reg1 uint i;
-  uint length,blobs,size;
+  uint length, blobs;
+  size_t size;
   CACHE_FIELD *copy,**blob_ptr;
   JOIN_CACHE  *cache;
   JOIN_TAB *join_tab;
@@ -13423,7 +13451,7 @@ store_record_in_cache(JOIN_CACHE *cache)
   length=cache->length;
   if (cache->blobs)
     length+=used_blob_length(cache->blob_ptr);
-  if ((last_record=(length+cache->length > (uint) (cache->end - pos))))
+  if ((last_record= (length + cache->length > (size_t) (cache->end - pos))))
     cache->ptr_record=cache->records;
 
   /*
@@ -13469,7 +13497,7 @@ store_record_in_cache(JOIN_CACHE *cache)
     }
   }
   cache->pos=pos;
-  return last_record || (uint) (cache->end -pos) < cache->length;
+  return last_record || (size_t) (cache->end - pos) < cache->length;
 }
 
 
diff --git a/sql/sql_select.h b/sql/sql_select.h
index 98f2a7829dd..d1acad1c5d6 100644
--- a/sql/sql_select.h
+++ b/sql/sql_select.h
@@ -316,11 +316,27 @@ public:
   SELECT_LEX_UNIT *unit;
   // select that processed
   SELECT_LEX *select_lex;
+  /* 
+    TRUE <=> optimizer must not mark any table as a constant table.
+    This is needed for subqueries in form "a IN (SELECT .. UNION SELECT ..):
+    when we optimize the select that reads the results of the union from a
+    temporary table, we must not mark the temp. table as constant because
+    the number of rows in it may vary from one subquery execution to another.
+  */
+  bool no_const_tables; 
   
   JOIN *tmp_join; // copy of this JOIN to be used with temporary tables
   ROLLUP rollup;				// Used with rollup
 
   bool select_distinct;				// Set if SELECT DISTINCT
+  /*
+    If we have the GROUP BY statement in the query,
+    but the group_list was emptied by optimizer, this
+    flag is TRUE.
+    It happens when fields in the GROUP BY are from
+    constant table
+  */
+  bool group_optimized_away;
 
   /*
     simple_xxxxx is set if ORDER/GROUP BY doesn't include any references
@@ -429,6 +445,7 @@ public:
     zero_result_cause= 0;
     optimized= 0;
     cond_equal= 0;
+    group_optimized_away= 0;
 
     all_fields= fields_arg;
     fields_list= fields_arg;
@@ -436,6 +453,8 @@ public:
     tmp_table_param.init();
     tmp_table_param.end_write_records= HA_POS_ERROR;
     rollup.state= ROLLUP::STATE_NONE;
+
+    no_const_tables= FALSE;
   }
 
   int prepare(Item ***rref_pointer_array, TABLE_LIST *tables, uint wind_num,
diff --git a/sql/sql_servers.cc b/sql/sql_servers.cc
index ac5ea6f4ac4..911372d5f4e 100644
--- a/sql/sql_servers.cc
+++ b/sql/sql_servers.cc
@@ -366,7 +366,7 @@ insert_server(THD *thd, FOREIGN_SERVER *server)
   tables.alias= tables.table_name= (char*) "servers";
 
   /* need to open before acquiring THR_LOCK_plugin or it will deadlock */
-  if (! (table= open_ltable(thd, &tables, TL_WRITE)))
+  if (! (table= open_ltable(thd, &tables, TL_WRITE, 0)))
     goto end;
 
   /* insert the server into the table */
@@ -588,7 +588,7 @@ int drop_server(THD *thd, LEX_SERVER_OPTIONS *server_options)
   if ((error= delete_server_record_in_cache(server_options)))
     goto end;
 
-  if (! (table= open_ltable(thd, &tables, TL_WRITE)))
+  if (! (table= open_ltable(thd, &tables, TL_WRITE, 0)))
   {
     error= my_errno;
     goto end;
@@ -705,7 +705,7 @@ int update_server(THD *thd, FOREIGN_SERVER *existing, FOREIGN_SERVER *altered)
   tables.db= (char*)"mysql";
   tables.alias= tables.table_name= (char*)"servers";
 
-  if (!(table= open_ltable(thd, &tables, TL_WRITE)))
+  if (!(table= open_ltable(thd, &tables, TL_WRITE, 0)))
   {
     error= my_errno;
     goto end;
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 5b8cb93baab..a0c3355b49a 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -2176,9 +2176,6 @@ void calc_sum_of_all_status(STATUS_VAR *to)
 }
 
 
-/* INFORMATION_SCHEMA name */
-LEX_STRING INFORMATION_SCHEMA_NAME= { C_STRING_WITH_LEN("information_schema")};
-
 /* This is only used internally, but we need it here as a forward reference */
 extern ST_SCHEMA_TABLE schema_tables[];
 
@@ -3670,7 +3667,7 @@ static int get_schema_views_record(THD *thd, TABLE_LIST *tables,
       Item *item;
       Item_field *field;
       /*
-        chech that at least one coulmn in view is updatable
+        check that at least one column in view is updatable
       */
       while ((item= it++))
       {
@@ -3681,6 +3678,8 @@ static int get_schema_views_record(THD *thd, TABLE_LIST *tables,
           break;
         }
       }
+      if (updatable_view && !tables->view->can_be_merged())
+        updatable_view= 0;
     }
     if (updatable_view)
       table->field[5]->store(STRING_WITH_LEN("YES"), cs);
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index dc3e72554fa..dd92c47d29b 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -1513,7 +1513,7 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
       table->db_type= share->db_type();
 
     /* Disable drop of enabled log tables */
-    if (share && share->log_table &&
+    if (share && (share->table_category == TABLE_CATEGORY_PERFORMANCE) &&
         check_if_log_table(table->db_length, table->db,
                            table->table_name_length, table->table_name, 1))
     {
@@ -1656,7 +1656,7 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
     on the table name.
   */
   pthread_mutex_unlock(&LOCK_open);
-  thd->tmp_table_used= tmp_table_deleted;
+  thd->thread_specific_used|= tmp_table_deleted;
   error= 0;
   if (wrong_tables.length())
   {
@@ -2789,6 +2789,8 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
 			  length);
 	      push_warning(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
 			   ER_TOO_LONG_KEY, warn_buff);
+              /* Align key length to multibyte char boundary */
+              length-= length % sql_field->charset->mbmaxlen;
 	    }
 	    else
 	    {
@@ -2819,8 +2821,6 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
       if (length > file->max_key_part_length() && key->type != Key::FULLTEXT)
       {
         length= file->max_key_part_length();
-        /* Align key length to multibyte char boundary */
-        length-= length % sql_field->charset->mbmaxlen;
 	if (key->type == Key::MULTIPLE)
 	{
 	  /* not a critical problem */
@@ -2829,6 +2829,8 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
 		      length);
 	  push_warning(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
 		       ER_TOO_LONG_KEY, warn_buff);
+          /* Align key length to multibyte char boundary */
+          length-= length % sql_field->charset->mbmaxlen;
 	}
 	else
 	{
@@ -3406,7 +3408,7 @@ bool mysql_create_table_no_lock(THD *thd,
       (void) rm_temporary_table(create_info->db_type, path);
       goto unlock_and_end;
     }
-    thd->tmp_table_used= 1;
+    thd->thread_specific_used= TRUE;
   }
 
   /*
@@ -3966,7 +3968,7 @@ static bool mysql_admin_table(THD* thd, TABLE_LIST* tables,
   Item *item;
   Protocol *protocol= thd->protocol;
   LEX *lex= thd->lex;
-  int result_code, disable_logs= 0;
+  int result_code;
   DBUG_ENTER("mysql_admin_table");
 
   if (end_active_trans(thd))
@@ -4014,22 +4016,6 @@ static bool mysql_admin_table(THD* thd, TABLE_LIST* tables,
       if (view_operator_func == NULL)
         table->required_type=FRMTYPE_TABLE;
 
-      /*
-        If we want to perform an admin operation on the log table
-        (E.g. rename) and lock_type >= TL_READ_NO_INSERT disable
-        log tables
-      */
-
-      if (check_if_log_table(table->db_length, table->db,
-                             table->table_name_length,
-                             table->table_name, 1) &&
-          lock_type >= TL_READ_NO_INSERT)
-      {
-        disable_logs= 1;
-        logger.lock();
-        logger.tmp_close_log_tables(thd);
-      }
-
       open_and_lock_tables(thd, table);
       thd->no_warnings_for_error= 0;
       table->next_global= save_next_global;
@@ -4099,8 +4085,7 @@ static bool mysql_admin_table(THD* thd, TABLE_LIST* tables,
     }
 
     /* Close all instances of the table to allow repair to rename files */
-    if (lock_type == TL_WRITE && table->table->s->version &&
-        !table->table->s->log_table)
+    if (lock_type == TL_WRITE && table->table->s->version)
     {
       pthread_mutex_lock(&LOCK_open);
       const char *old_message=thd->enter_cond(&COND_refresh, &LOCK_open,
@@ -4130,7 +4115,8 @@ static bool mysql_admin_table(THD* thd, TABLE_LIST* tables,
         goto err;
     }
 
-    if (operator_func == &handler::ha_repair)
+    if (operator_func == &handler::ha_repair &&
+        !(check_opt->sql_flags & TT_USEFRM))
     {
       if ((table->table->file->check_old_types() == HA_ADMIN_NEEDS_ALTER) ||
           (table->table->file->ha_check_for_upgrade(check_opt) ==
@@ -4258,7 +4244,7 @@ send_result_message:
       close_thread_tables(thd);
       if (!result_code) // recreation went ok
       {
-        if ((table->table= open_ltable(thd, table, lock_type)) &&
+        if ((table->table= open_ltable(thd, table, lock_type, 0)) &&
             ((result_code= table->table->file->analyze(thd, check_opt)) > 0))
           result_code= 0; // analyze went ok
       }
@@ -4324,10 +4310,9 @@ send_result_message:
     }
     if (table->table)
     {
-      /* in the below check we do not refresh the log tables */
       if (fatal_error)
         table->table->s->version=0;               // Force close of table
-      else if (open_for_modify && !table->table->s->log_table)
+      else if (open_for_modify)
       {
         if (table->table->s->tmp_table)
           table->table->file->info(HA_STATUS_CONST);
@@ -4350,24 +4335,11 @@ send_result_message:
   }
 
   send_eof(thd);
-  if (disable_logs)
-  {
-    if (logger.reopen_log_tables())
-      my_error(ER_CANT_ACTIVATE_LOG, MYF(0));
-    logger.unlock();
-  }
   DBUG_RETURN(FALSE);
 
  err:
   ha_autocommit_or_rollback(thd, 1);
   close_thread_tables(thd);			// Shouldn't be needed
-  /* enable logging back if needed */
-  if (disable_logs)
-  {
-    if (logger.reopen_log_tables())
-      my_error(ER_CANT_ACTIVATE_LOG, MYF(0));
-    logger.unlock();
-  }
   if (table)
     table->table=0;
   DBUG_RETURN(TRUE);
@@ -4812,7 +4784,7 @@ mysql_discard_or_import_tablespace(THD *thd,
    not complain when we lock the table
  */
   thd->tablespace_op= TRUE;
-  if (!(table=open_ltable(thd,table_list,TL_WRITE)))
+  if (!(table=open_ltable(thd, table_list, TL_WRITE, 0)))
   {
     thd->tablespace_op=FALSE;
     DBUG_RETURN(-1);
@@ -5728,7 +5700,7 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
 #ifdef WITH_PARTITION_STORAGE_ENGINE
       if (alter_info->flags & ALTER_PARTITION)
       {
-	my_error(ER_WRONG_USAGE, MYF(0), "PARTITION", "log table");
+        my_error(ER_WRONG_USAGE, MYF(0), "PARTITION", "log table");
         DBUG_RETURN(TRUE);
       }
 #endif
@@ -5817,7 +5789,7 @@ view_err:
     start_waiting_global_read_lock(thd);
     DBUG_RETURN(error);
   }
-  if (!(table=open_ltable(thd,table_list,TL_WRITE_ALLOW_READ)))
+  if (!(table=open_ltable(thd, table_list, TL_WRITE_ALLOW_READ, 0)))
     DBUG_RETURN(TRUE);
   table->use_all_columns();
 
@@ -6333,11 +6305,9 @@ view_err:
   {
     VOID(pthread_mutex_lock(&LOCK_open));
     wait_while_table_is_used(thd, table, HA_EXTRA_FORCE_REOPEN);
-    table->file->ha_external_lock(thd, F_WRLCK);
+    VOID(pthread_mutex_unlock(&LOCK_open));
     alter_table_manage_keys(table, table->file->indexes_are_disabled(),
                             alter_info->keys_onoff);
-    table->file->ha_external_lock(thd, F_UNLCK);
-    VOID(pthread_mutex_unlock(&LOCK_open));
     error= ha_commit_stmt(thd);
     if (ha_commit(thd))
       error= 1;
@@ -6768,7 +6738,6 @@ copy_data_between_tables(TABLE *from,TABLE *to,
   alter_table_manage_keys(to, from->file->indexes_are_disabled(), keys_onoff);
 
   /* We can abort alter table for any table type */
-  thd->no_trans_update.stmt= FALSE;
   thd->abort_on_warning= !ignore && test(thd->variables.sql_mode &
                                          (MODE_STRICT_TRANS_TABLES |
                                           MODE_STRICT_ALL_TABLES));
@@ -6986,7 +6955,7 @@ bool mysql_checksum_table(THD *thd, TABLE_LIST *tables,
 
     strxmov(table_name, table->db ,".", table->table_name, NullS);
 
-    t= table->table= open_ltable(thd, table, TL_READ);
+    t= table->table= open_ltable(thd, table, TL_READ, 0);
     thd->clear_error();			// these errors shouldn't get client
 
     protocol->prepare_for_resend();
diff --git a/sql/sql_udf.cc b/sql/sql_udf.cc
index 8361fc64f33..10bb7844d88 100644
--- a/sql/sql_udf.cc
+++ b/sql/sql_udf.cc
@@ -472,7 +472,7 @@ int mysql_create_function(THD *thd,udf_func *udf)
   tables.db= (char*) "mysql";
   tables.table_name= tables.alias= (char*) "func";
   /* Allow creation of functions even if we can't open func table */
-  if (!(table = open_ltable(thd,&tables,TL_WRITE)))
+  if (!(table = open_ltable(thd, &tables, TL_WRITE, 0)))
     goto err;
   table->use_all_columns();
   restore_record(table, s->default_values);	// Default values for fields
@@ -547,7 +547,7 @@ int mysql_drop_function(THD *thd,const LEX_STRING *udf_name)
   bzero((char*) &tables,sizeof(tables));
   tables.db=(char*) "mysql";
   tables.table_name= tables.alias= (char*) "func";
-  if (!(table = open_ltable(thd,&tables,TL_WRITE)))
+  if (!(table = open_ltable(thd, &tables, TL_WRITE, 0)))
     goto err;
   table->use_all_columns();
   table->field[0]->store(exact_name_str, exact_name_len, &my_charset_bin);
diff --git a/sql/sql_union.cc b/sql/sql_union.cc
index c872d3cb241..5da4b97cc5d 100644
--- a/sql/sql_union.cc
+++ b/sql/sql_union.cc
@@ -547,6 +547,10 @@ bool st_select_lex_unit::exec()
 	/*
 	  allocate JOIN for fake select only once (prevent
 	  mysql_select automatic allocation)
+          TODO: The above is nonsense. mysql_select() will not allocate the
+          join if one already exists. There must be some other reason why we
+          don't let it allocate the join. Perhaps this is because we need
+          some special parameter values passed to join constructor?
 	*/
 	if (!(fake_select_lex->join= new JOIN(thd, item_list,
 					      fake_select_lex->options, result)))
@@ -554,33 +558,52 @@ bool st_select_lex_unit::exec()
 	  fake_select_lex->table_list.empty();
 	  DBUG_RETURN(TRUE);
 	}
+        fake_select_lex->join->no_const_tables= TRUE;
 
 	/*
 	  Fake st_select_lex should have item list for correctref_array
 	  allocation.
 	*/
 	fake_select_lex->item_list= item_list;
+        saved_error= mysql_select(thd, &fake_select_lex->ref_pointer_array,
+                              &result_table_list,
+                              0, item_list, NULL,
+                              global_parameters->order_list.elements,
+                              (ORDER*)global_parameters->order_list.first,
+                              (ORDER*) NULL, NULL, (ORDER*) NULL,
+                              fake_select_lex->options | SELECT_NO_UNLOCK,
+                              result, this, fake_select_lex);
       }
       else
       {
-	JOIN_TAB *tab,*end;
-	for (tab=join->join_tab, end=tab+join->tables ;
-	     tab && tab != end ;
-	     tab++)
-	{
-	  delete tab->select;
-	  delete tab->quick;
-	}
-	join->init(thd, item_list, fake_select_lex->options, result);
+        if (describe)
+        {
+          /*
+            In EXPLAIN command, constant subqueries that do not use any
+            tables are executed two times:
+             - 1st time is a real evaluation to get the subquery value
+             - 2nd time is to produce EXPLAIN output rows.
+            1st execution sets certain members (e.g. select_result) to perform
+            subquery execution rather than EXPLAIN line production. In order 
+            to reset them back, we re-do all of the actions (yes it is ugly):
+          */
+	  join->init(thd, item_list, fake_select_lex->options, result);
+          saved_error= mysql_select(thd, &fake_select_lex->ref_pointer_array,
+                                &result_table_list,
+                                0, item_list, NULL,
+                                global_parameters->order_list.elements,
+                                (ORDER*)global_parameters->order_list.first,
+                                (ORDER*) NULL, NULL, (ORDER*) NULL,
+                                fake_select_lex->options | SELECT_NO_UNLOCK,
+                                result, this, fake_select_lex);
+        }
+        else
+        {
+          join->examined_rows= 0;
+          saved_error= join->reinit();
+          join->exec();
+        }
       }
-      saved_error= mysql_select(thd, &fake_select_lex->ref_pointer_array,
-                            &result_table_list,
-                            0, item_list, NULL,
-                            global_parameters->order_list.elements,
-                            (ORDER*)global_parameters->order_list.first,
-                            (ORDER*) NULL, NULL, (ORDER*) NULL,
-                            fake_select_lex->options | SELECT_NO_UNLOCK,
-                            result, this, fake_select_lex);
 
       fake_select_lex->table_list.empty();
       if (!saved_error)
diff --git a/sql/sql_update.cc b/sql/sql_update.cc
index 1016a23b5ae..cb3f2fece89 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -530,7 +530,6 @@ int mysql_update(THD *thd,
   thd->proc_info="Updating";
 
   transactional_table= table->file->has_transactions();
-  thd->no_trans_update.stmt= FALSE;
   thd->abort_on_warning= test(!ignore &&
                               (thd->variables.sql_mode &
                                (MODE_STRICT_TRANS_TABLES |
@@ -641,7 +640,6 @@ int mysql_update(THD *thd,
             updated++;
           else
             error= 0;
-          thd->no_trans_update.stmt= !transactional_table;
 
           if (table->triggers &&
               table->triggers->process_triggers(thd, TRG_EVENT_UPDATE,
@@ -717,6 +715,10 @@ int mysql_update(THD *thd,
   }
   dup_key_found= 0;
 
+  if (!transactional_table && updated > 0)
+    thd->transaction.stmt.modified_non_trans_table= TRUE;
+
+
   /*
     todo bug#27571: to avoid asynchronization of `error' and
     `error_code' of binlog event constructor
@@ -809,9 +811,10 @@ int mysql_update(THD *thd,
         error=1;				// Rollback update
       }
     }
-    if (!transactional_table)
-      thd->no_trans_update.all= TRUE;
+    if (thd->transaction.stmt.modified_non_trans_table)
+      thd->transaction.all.modified_non_trans_table= TRUE;
   }
+  DBUG_ASSERT(transactional_table || !updated || thd->transaction.stmt.modified_non_trans_table);
   free_underlaid_joins(thd, select_lex);
   if (transactional_table)
   {
@@ -1176,7 +1179,6 @@ bool mysql_multi_update(THD *thd,
 				 handle_duplicates, ignore)))
     DBUG_RETURN(TRUE);
 
-  thd->no_trans_update.stmt= FALSE;
   thd->abort_on_warning= test(thd->variables.sql_mode &
                               (MODE_STRICT_TRANS_TABLES |
                                MODE_STRICT_ALL_TABLES));
@@ -1549,9 +1551,8 @@ multi_update::~multi_update()
   if (copy_field)
     delete [] copy_field;
   thd->count_cuted_fields= CHECK_FIELD_IGNORE;		// Restore this setting
-  if (!trans_safe)  // todo: remove since redundant
-    thd->no_trans_update.all= TRUE;
-  DBUG_ASSERT(trans_safe || thd->no_trans_update.all);
+  DBUG_ASSERT(trans_safe || !updated || 
+              thd->transaction.all.modified_non_trans_table);
 }
 
 
@@ -1655,7 +1656,7 @@ bool multi_update::send_data(List<Item> &not_used_values)
           else
           {
             trans_safe= 0;
-            thd->no_trans_update.stmt= TRUE;
+            thd->transaction.stmt.modified_non_trans_table= TRUE;
           }
           if (table->triggers &&
               table->triggers->process_triggers(thd, TRG_EVENT_UPDATE,
@@ -1718,7 +1719,6 @@ void multi_update::send_error(uint errcode,const char *err)
 
   /* Something already updated so we have to invalidate cache */
   query_cache_invalidate3(thd, update_tables, 1);
-
   /*
     If all tables that has been updated are trans safe then just do rollback.
     If not attempt to do remaining updates.
@@ -1731,7 +1731,7 @@ void multi_update::send_error(uint errcode,const char *err)
   }
   else
   { 
-    DBUG_ASSERT(thd->no_trans_update.stmt);
+    DBUG_ASSERT(thd->transaction.stmt.modified_non_trans_table);
     if (do_update && table_count > 1)
     {
       /* Add warning here */
@@ -1742,7 +1742,7 @@ void multi_update::send_error(uint errcode,const char *err)
       VOID(do_updates(0));
     }
   }
-  if (thd->no_trans_update.stmt)
+  if (thd->transaction.stmt.modified_non_trans_table)
   {
     /*
       The query has to binlog because there's a modified non-transactional table
@@ -1755,9 +1755,9 @@ void multi_update::send_error(uint errcode,const char *err)
                         transactional_tables, FALSE);
     }
     if (!trans_safe)
-      thd->no_trans_update.all= TRUE;
+      thd->transaction.all.modified_non_trans_table= TRUE;
   }
-  DBUG_ASSERT(trans_safe || !updated || thd->no_trans_update.stmt);
+  DBUG_ASSERT(trans_safe || !updated || thd->transaction.stmt.modified_non_trans_table);
   
   if (transactional_tables)
   {
@@ -1900,7 +1900,7 @@ int multi_update::do_updates(bool from_send_error)
       else
       {
         trans_safe= 0;				// Can't do safe rollback
-        thd->no_trans_update.stmt= TRUE;
+        thd->transaction.stmt.modified_non_trans_table= TRUE;
       }
     }
     (void) table->file->ha_rnd_end();
@@ -1934,7 +1934,7 @@ err2:
     else
     {
       trans_safe= 0;
-      thd->no_trans_update.stmt= TRUE;
+      thd->transaction.stmt.modified_non_trans_table= TRUE;
     }
   }
   DBUG_RETURN(1);
@@ -1961,7 +1961,6 @@ bool multi_update::send_eof()
   {
     query_cache_invalidate3(thd, update_tables, 1);
   }
-
   /*
     Write the SQL statement to the binlog if we updated
     rows and we succeeded or if we updated some non
@@ -1971,8 +1970,9 @@ bool multi_update::send_eof()
     either from the query's list or via a stored routine: bug#13270,23333
   */
 
-  DBUG_ASSERT(trans_safe || !updated || thd->no_trans_update.stmt);
-  if (local_error == 0 || thd->no_trans_update.stmt)
+  DBUG_ASSERT(trans_safe || !updated || 
+              thd->transaction.stmt.modified_non_trans_table);
+  if (local_error == 0 || thd->transaction.stmt.modified_non_trans_table)
   {
     if (mysql_bin_log.is_open())
     {
@@ -1988,8 +1988,8 @@ bool multi_update::send_eof()
 	local_error= 1;				// Rollback update
       }
     }
-    if (!trans_safe)
-     thd->no_trans_update.all= TRUE;
+    if (thd->transaction.stmt.modified_non_trans_table)
+      thd->transaction.all.modified_non_trans_table= TRUE;
   }
 
   if (transactional_tables)
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index d9a808bf8f7..591f2c46d4d 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -7277,7 +7277,7 @@ join_table:
           so that [INNER | CROSS] JOIN is properly nested as other
           left-associative joins.
         */
-        table_ref %prec TABLE_REF_PRIORITY normal_join table_ref
+        table_ref normal_join table_ref %prec TABLE_REF_PRIORITY
           { MYSQL_YYABORT_UNLESS($1 && ($$=$3)); }
 	| table_ref STRAIGHT_JOIN table_factor
 	  { MYSQL_YYABORT_UNLESS($1 && ($$=$3)); $3->straight=1; }
@@ -8729,6 +8729,8 @@ show_param:
 	    LEX *lex=Lex;
 	    lex->sql_command= SQLCOM_SHOW_STORAGE_ENGINES;
 	    WARN_DEPRECATED(yythd, "5.2", "SHOW TABLE TYPES", "'SHOW [STORAGE] ENGINES'");
+            if (prepare_schema_table(YYTHD, lex, 0, SCH_ENGINES))
+              MYSQL_YYABORT;
 	  }
 	| opt_storage ENGINES_SYM
 	  {
@@ -9503,7 +9505,8 @@ simple_ident:
             Item_splocal *splocal;
             splocal= new Item_splocal($1, spv->offset, spv->type,
                                       lip->get_tok_start_prev() -
-                                      lex->sphead->m_tmp_query);
+                                      lex->sphead->m_tmp_query,
+                                      lip->get_tok_end() - lip->get_tok_start_prev());
 #ifndef DBUG_OFF
             if (splocal)
               splocal->m_sp= lex->sphead;
diff --git a/sql/table.cc b/sql/table.cc
index a58f59d3a75..5e036abe6b1 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -21,6 +21,18 @@
 #include <m_ctype.h>
 #include "my_md5.h"
 
+/* INFORMATION_SCHEMA name */
+LEX_STRING INFORMATION_SCHEMA_NAME= {C_STRING_WITH_LEN("information_schema")};
+
+/* MYSQL_SCHEMA name */
+LEX_STRING MYSQL_SCHEMA_NAME= {C_STRING_WITH_LEN("mysql")};
+
+/* GENERAL_LOG name */
+LEX_STRING GENERAL_LOG_NAME= {C_STRING_WITH_LEN("general_log")};
+
+/* SLOW_LOG name */
+LEX_STRING SLOW_LOG_NAME= {C_STRING_WITH_LEN("slow_log")};
+
 	/* Functions defined in this file */
 
 void open_table_error(TABLE_SHARE *share, int error, int db_errno,
@@ -31,6 +43,7 @@ static void fix_type_pointers(const char ***array, TYPELIB *point_to_type,
 			      uint types, char **names);
 static uint find_field(Field **fields, uchar *record, uint start, uint length);
 
+inline bool is_system_table_name(const char *name, uint length);
 
 /**************************************************************************
   Object_creation_ctx implementation.
@@ -192,6 +205,49 @@ char *fn_rext(char *name)
   return name + strlen(name);
 }
 
+TABLE_CATEGORY get_table_category(const LEX_STRING *db, const LEX_STRING *name)
+{
+  DBUG_ASSERT(db != NULL);
+  DBUG_ASSERT(name != NULL);
+
+  if ((db->length == INFORMATION_SCHEMA_NAME.length) &&
+      (my_strcasecmp(system_charset_info,
+                    INFORMATION_SCHEMA_NAME.str,
+                    db->str) == 0))
+  {
+    return TABLE_CATEGORY_INFORMATION;
+  }
+
+  if ((db->length == MYSQL_SCHEMA_NAME.length) &&
+      (my_strcasecmp(system_charset_info,
+                    MYSQL_SCHEMA_NAME.str,
+                    db->str) == 0))
+  {
+    if (is_system_table_name(name->str, name->length))
+    {
+      return TABLE_CATEGORY_SYSTEM;
+    }
+
+    if ((name->length == GENERAL_LOG_NAME.length) &&
+        (my_strcasecmp(system_charset_info,
+                      GENERAL_LOG_NAME.str,
+                      name->str) == 0))
+    {
+      return TABLE_CATEGORY_PERFORMANCE;
+    }
+
+    if ((name->length == SLOW_LOG_NAME.length) &&
+        (my_strcasecmp(system_charset_info,
+                      SLOW_LOG_NAME.str,
+                      name->str) == 0))
+    {
+      return TABLE_CATEGORY_PERFORMANCE;
+    }
+  }
+
+  return TABLE_CATEGORY_USER;
+}
+
 
 /*
   Allocate a setup TABLE_SHARE structure
@@ -297,7 +353,8 @@ void init_tmp_table_share(TABLE_SHARE *share, const char *key,
 
   bzero((char*) share, sizeof(*share));
   init_sql_alloc(&share->mem_root, TABLE_ALLOC_BLOCK_SIZE, 0);
-  share->tmp_table=  	         INTERNAL_TMP_TABLE;
+  share->table_category=         TABLE_CATEGORY_TEMPORARY;
+  share->tmp_table=              INTERNAL_TMP_TABLE;
   share->db.str=                 (char*) key;
   share->db.length=		 strlen(key);
   share->table_cache_key.str=    (char*) key;
@@ -544,28 +601,11 @@ int open_table_def(THD *thd, TABLE_SHARE *share, uint db_flags)
     *root_ptr= &share->mem_root;
     error= open_binary_frm(thd, share, head, file);
     *root_ptr= old_root;
-
-    if (share->db.length == 5 && !(lower_case_table_names ?
-        my_strcasecmp(system_charset_info, share->db.str, "mysql") :
-        strcmp(share->db.str, "mysql")))
-    {
-      /*
-        We can't mark all tables in 'mysql' database as system since we don't
-        allow to lock such tables for writing with any other tables (even with
-        other system tables) and some privilege tables need this.
-      */
-      share->system_table= is_system_table_name(share->table_name.str,
-                                                share->table_name.length);
-      if (!share->system_table)
-      {
-        share->log_table= check_if_log_table(share->db.length, share->db.str,
-                                             share->table_name.length,
-                                             share->table_name.str, 0);
-      }
-    }
     error_given= 1;
   }
 
+  share->table_category= get_table_category(& share->db, & share->table_name);
+
   if (!error)
     thd->status_var.opened_shares++;
 
@@ -1349,7 +1389,11 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
             the primary key, then we can use any key to find this column
           */
           if (ha_option & HA_PRIMARY_KEY_IN_READ_INDEX)
+          {
             field->part_of_key= share->keys_in_use;
+            if (field->part_of_sortkey.is_set(key))
+              field->part_of_sortkey= share->keys_in_use;
+          }
         }
         if (field->key_length() != key_part->length)
         {
@@ -2849,135 +2893,6 @@ void st_table::reset_item_list(List<Item> *item_list) const
   }
 }
 
-
-/**
-  Set the initial purpose of this TABLE_LIST object in the list of
-  used tables. We need to track this information on table-by-
-  table basis, since when this table becomes an element of the
-  pre-locked list, it's impossible to identify which SQL
-  sub-statement it has been originally used in.
-
-  E.g.:
-
-  User request:                 SELECT * FROM t1 WHERE f1();
-  FUNCTION f1():                DELETE FROM t2; RETURN 1;
-  BEFORE DELETE trigger on t2:  INSERT INTO t3 VALUES (old.a);
-
-  For this user request, the pre-locked list will contain t1, t2, t3
-  table elements, each needed for different DML.
-
-  This method is called immediately after parsing for tables
-  of the table list of the top-level select lex.
-
-  The trigger event map is updated to reflect INSERT, UPDATE, DELETE,
-  REPLACE, LOAD DATA, CREATE TABLE .. SELECT, CREATE TABLE ..
-  REPLACE SELECT statements, and additionally ON DUPLICATE KEY UPDATE
-  clause.
-*/
-
-void
-TABLE_LIST::set_trg_event_type(const st_lex *lex)
-{
-  enum trg_event_type trg_event;
-
-  /*
-    Some auxiliary operations
-    (e.g. GRANT processing) create TABLE_LIST instances outside
-    the parser. Additionally, some commands (e.g. OPTIMIZE) change
-    the lock type for a table only after parsing is done. Luckily,
-    these do not fire triggers and do not need to pre-load them.
-    For these TABLE_LISTs set_trg_event_type is never called, and
-    trg_event_map is always empty. That means that the pre-locking
-    algorithm will ignore triggers defined on these tables, if
-    any, and the execution will either fail with an assert in
-    sql_trigger.cc or with an error that a used table was not
-    pre-locked, in case of a production build.
-
-    TODO: this usage pattern creates unnecessary module dependencies
-    and should be rewritten to go through the parser.
-    Table list instances created outside the parser in most cases
-    refer to mysql.* system tables. It is not allowed to have
-    a trigger on a system table, but keeping track of
-    initialization provides extra safety in case this limitation
-    is circumvented.
-  */
-
-  /*
-    This is a fast check to filter out statements that do
-    not change data, or tables  on the right side, in case of
-    INSERT .. SELECT, CREATE TABLE .. SELECT and so on.
-    Here we also filter out OPTIMIZE statement and non-updateable
-    views, for which lock_type is TL_UNLOCK or TL_READ after
-    parsing.
-  */
-  if (static_cast<int>(lock_type) < static_cast<int>(TL_WRITE_ALLOW_WRITE))
-    return;
-
-  switch (lex->sql_command) {
-  /*
-    Basic INSERT. If there is an additional ON DUPLIATE KEY UPDATE
-    clause, it will be handled later in this method.
-  */
-  case SQLCOM_INSERT:                           /* fall through */
-  case SQLCOM_INSERT_SELECT:
-  /*
-    LOAD DATA ... INFILE is expected to fire BEFORE/AFTER INSERT
-    triggers.
-    If the statement also has REPLACE clause, it will be
-    handled later in this method.
-  */
-  case SQLCOM_LOAD:                             /* fall through */
-  /*
-    REPLACE is semantically equivalent to INSERT. In case
-    of a primary or unique key conflict, it deletes the old
-    record and inserts a new one. So we also may need to
-    fire ON DELETE triggers. This functionality is handled
-    later in this method.
-  */
-  case SQLCOM_REPLACE:                          /* fall through */
-  case SQLCOM_REPLACE_SELECT:
-  /*
-    CREATE TABLE ... SELECT defaults to INSERT if the table or
-    view already exists. REPLACE option of CREATE TABLE ...
-    REPLACE SELECT is handled later in this method.
-  */
-  case SQLCOM_CREATE_TABLE:
-    trg_event= TRG_EVENT_INSERT;
-    break;
-  /* Basic update and multi-update */
-  case SQLCOM_UPDATE:                           /* fall through */
-  case SQLCOM_UPDATE_MULTI:
-    trg_event= TRG_EVENT_UPDATE;
-    break;
-  /* Basic delete and multi-delete */
-  case SQLCOM_DELETE:                           /* fall through */
-  case SQLCOM_DELETE_MULTI:
-    trg_event= TRG_EVENT_DELETE;
-    break;
-  default:
-    /*
-      OK to return, since value of 'duplicates' is irrelevant
-      for non-updating commands.
-    */
-    return;
-  }
-  trg_event_map|= static_cast<uint8>(1 << static_cast<int>(trg_event));
-
-  switch (lex->duplicates) {
-  case DUP_UPDATE:
-    trg_event= TRG_EVENT_UPDATE;
-    break;
-  case DUP_REPLACE:
-    trg_event= TRG_EVENT_DELETE;
-    break;
-  case DUP_ERROR:
-  default:
-    return;
-  }
-  trg_event_map|= static_cast<uint8>(1 << static_cast<int>(trg_event));
-}
-
-
 /*
   calculate md5 of query
 
@@ -4642,11 +4557,11 @@ bool TABLE_LIST::process_index_hints(TABLE *table)
     key_map index_join[INDEX_HINT_FORCE + 1];
     key_map index_order[INDEX_HINT_FORCE + 1];
     key_map index_group[INDEX_HINT_FORCE + 1];
-    index_hint *hint;
+    Index_hint *hint;
     int type;
     bool have_empty_use_join= FALSE, have_empty_use_order= FALSE, 
          have_empty_use_group= FALSE;
-    List_iterator <index_hint> iter(*index_hints);
+    List_iterator <Index_hint> iter(*index_hints);
 
     /* initialize temporary variables used to collect hints of each kind */
     for (type= INDEX_HINT_IGNORE; type <= INDEX_HINT_FORCE; type++)
diff --git a/sql/table.h b/sql/table.h
index 494b74d564c..e46036f3ee9 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -140,6 +140,100 @@ class Field_timestamp;
 class Field_blob;
 class Table_triggers_list;
 
+/**
+  Category of table found in the table share.
+*/
+enum enum_table_category
+{
+  /**
+    Unknown value.
+  */
+  TABLE_UNKNOWN_CATEGORY=0,
+
+  /**
+    Temporary table.
+    The table is visible only in the session.
+    Therefore,
+    - FLUSH TABLES WITH READ LOCK
+    - SET GLOBAL READ_ONLY = ON
+    do not apply to this table.
+    Note that LOCK TABLE <t> FOR READ/WRITE
+    can be used on temporary tables.
+    Temporary tables are not part of the table cache.
+  */
+  TABLE_CATEGORY_TEMPORARY=1,
+
+  /**
+    User table.
+    These tables do honor:
+    - LOCK TABLE <t> FOR READ/WRITE
+    - FLUSH TABLES WITH READ LOCK
+    - SET GLOBAL READ_ONLY = ON
+    User tables are cached in the table cache.
+  */
+  TABLE_CATEGORY_USER=2,
+
+  /**
+    System table, maintained by the server.
+    These tables do honor:
+    - LOCK TABLE <t> FOR READ/WRITE
+    - FLUSH TABLES WITH READ LOCK
+    - SET GLOBAL READ_ONLY = ON
+    Typically, writes to system tables are performed by
+    the server implementation, not explicitly be a user.
+    System tables are cached in the table cache.
+  */
+  TABLE_CATEGORY_SYSTEM=3,
+
+  /**
+    Information schema tables.
+    These tables are an interface provided by the system
+    to inspect the system metadata.
+    These tables do *not* honor:
+    - LOCK TABLE <t> FOR READ/WRITE
+    - FLUSH TABLES WITH READ LOCK
+    - SET GLOBAL READ_ONLY = ON
+    as there is no point in locking explicitely
+    an INFORMATION_SCHEMA table.
+    Nothing is directly written to information schema tables.
+    Note that this value is not used currently,
+    since information schema tables are not shared,
+    but implemented as session specific temporary tables.
+  */
+  /*
+    TODO: Fixing the performance issues of I_S will lead
+    to I_S tables in the table cache, which should use
+    this table type.
+  */
+  TABLE_CATEGORY_INFORMATION=4,
+
+  /**
+    Performance schema tables.
+    These tables are an interface provided by the system
+    to inspect the system performance data.
+    These tables do *not* honor:
+    - LOCK TABLE <t> FOR READ/WRITE
+    - FLUSH TABLES WITH READ LOCK
+    - SET GLOBAL READ_ONLY = ON
+    as there is no point in locking explicitely
+    a PERFORMANCE_SCHEMA table.
+    An example of PERFORMANCE_SCHEMA tables are:
+    - mysql.slow_log
+    - mysql.general_log,
+    which *are* updated even when there is either
+    a GLOBAL READ LOCK or a GLOBAL READ_ONLY in effect.
+    User queries do not write directly to these tables
+    (there are exceptions for log tables).
+    The server implementation perform writes.
+    Performance tables are cached in the table cache.
+  */
+  TABLE_CATEGORY_PERFORMANCE=5
+};
+typedef enum enum_table_category TABLE_CATEGORY;
+
+TABLE_CATEGORY get_table_category(const LEX_STRING *db,
+                                  const LEX_STRING *name);
+
 /*
   This structure is shared between different table objects. There is one
   instance of table share per one table in the database.
@@ -148,6 +242,10 @@ class Table_triggers_list;
 typedef struct st_table_share
 {
   st_table_share() {}                    /* Remove gcc warning */
+
+  /** Category of this table. */
+  TABLE_CATEGORY table_category;
+
   /* hash of field names (contains pointers to elements of field array) */
   HASH	name_hash;			/* hash of field names */
   MEM_ROOT mem_root;
@@ -259,18 +357,6 @@ typedef struct st_table_share
   */
   int cached_row_logging_check;
 
-  /*
-    TRUE if this is a system table like 'mysql.proc', which we want to be
-    able to open and lock even when we already have some tables open and
-    locked. To avoid deadlocks we have to put certain restrictions on
-    locking of this table for writing. FALSE - otherwise.
-  */
-  bool system_table;
-  /*
-    This flag is set for the log tables. Used during FLUSH instances to skip
-    log tables, while closing tables (since logs must be always available)
-  */
-  bool log_table;
 #ifdef WITH_PARTITION_STORAGE_ENGINE
   bool auto_partitioned;
   const char *partition_info;
@@ -334,6 +420,16 @@ typedef struct st_table_share
     set_table_cache_key(key_buff, key_length);
   }
 
+  inline bool honor_global_locks()
+  {
+    return ((table_category == TABLE_CATEGORY_USER)
+            || (table_category == TABLE_CATEGORY_SYSTEM));
+  }
+
+  inline bool require_write_privileges()
+  {
+    return (table_category == TABLE_CATEGORY_PERFORMANCE);
+  }
 } TABLE_SHARE;
 
 
@@ -754,7 +850,7 @@ public:
          (TABLE_LIST::join_using_fields != NULL)
 */
 
-class index_hint;
+class Index_hint;
 struct TABLE_LIST
 {
   TABLE_LIST() {}                          /* Remove gcc warning */
@@ -826,7 +922,7 @@ struct TABLE_LIST
   */
   TABLE_LIST *next_name_resolution_table;
   /* Index names in a "... JOIN ... USE/IGNORE INDEX ..." clause. */
-  List<index_hint> *index_hints;
+  List<Index_hint> *index_hints;
   TABLE        *table;                          /* opened table */
   uint          table_id; /* table id (from binlog) for opened table */
   /*
@@ -1052,7 +1148,6 @@ struct TABLE_LIST
   */
   bool process_index_hints(TABLE *table);
 
-  void set_trg_event_type(const st_lex *lex);
 private:
   bool prep_check_option(THD *thd, uint8 check_opt_type);
   bool prep_where(THD *thd, Item **conds, bool no_where_clause);
diff --git a/sql/tztime.cc b/sql/tztime.cc
index 0c717dd2ece..14192d06978 100644
--- a/sql/tztime.cc
+++ b/sql/tztime.cc
@@ -807,19 +807,6 @@ sec_since_epoch(int year, int mon, int mday, int hour, int min ,int sec)
          SECS_PER_MIN + sec;
 }
 
-
- /*
-  Works like sec_since_epoch but expects MYSQL_TIME structure as parameter.
-*/
-
-my_time_t
-sec_since_epoch_TIME(MYSQL_TIME *t)
-{
-  return sec_since_epoch(t->year, t->month, t->day,
-                         t->hour, t->minute, t->second);
-}
-
-
 /*
   Converts local time in broken down MYSQL_TIME representation to my_time_t
   representation.
@@ -1425,7 +1412,9 @@ Time_zone_offset::get_name() const
 
 static Time_zone_utc tz_UTC;
 static Time_zone_system tz_SYSTEM;
+static Time_zone_offset tz_OFFSET0(0);
 
+Time_zone *my_tz_OFFSET0= &tz_OFFSET0;
 Time_zone *my_tz_UTC= &tz_UTC;
 Time_zone *my_tz_SYSTEM= &tz_SYSTEM;
 
diff --git a/sql/tztime.h b/sql/tztime.h
index f7cc7042d79..ddd80b88bf2 100644
--- a/sql/tztime.h
+++ b/sql/tztime.h
@@ -59,6 +59,7 @@ public:
 
 extern Time_zone * my_tz_UTC;
 extern Time_zone * my_tz_SYSTEM;
+extern Time_zone * my_tz_OFFSET0;
 extern Time_zone * my_tz_find(THD *thd, const String *name);
 extern my_bool     my_tz_init(THD *org_thd, const char *default_tzname, my_bool bootstrap);
 extern void        my_tz_free();