Merge bk-internal.mysql.com:/home/bk/mysql-5.1

into  bodhi.(none):/opt/local/work/mysql-5.1-runtime


client/mysqldump.c:
  Auto merged
mysql-test/r/mysqldump.result:
  Auto merged
mysql-test/t/disabled.def:
  Auto merged
mysql-test/t/mysqldump.test:
  Auto merged
sql/handler.cc:
  Auto merged
sql/lock.cc:
  Auto merged
sql/sql_base.cc:
  Auto merged
sql/sql_lex.cc:
  Auto merged
sql/sql_parse.cc:
  Auto merged
sql/sql_table.cc:
  Auto merged
sql/table.cc:
  Auto merged
sql/table.h:
  Auto merged
mysql-test/include/mix1.inc:
  Manual merge.
mysql-test/r/innodb_mysql.result:
  Manual merge.

32 files changed, 990 insertions, 952 deletions

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/handler.cc b/sql/handler.cc
index 55e2e478027..f06ad282efa 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -1516,34 +1516,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.
 
@@ -3687,6 +3659,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..7cf7bfe043d 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -992,6 +992,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 +1023,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 +1038,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 +1594,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 +1728,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/lock.cc b/sql/lock.cc
index 63d0807b975..08ff90ce983 100644
--- a/sql/lock.cc
+++ b/sql/lock.cc
@@ -116,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,
@@ -175,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);
@@ -271,6 +346,7 @@ 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());
@@ -379,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;
@@ -396,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;
 
@@ -445,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);
@@ -704,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..dbb664c9d18 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.
 
@@ -2097,6 +1950,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 +1996,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 +2005,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 +2048,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 +2158,7 @@ bool MYSQL_QUERY_LOG::write(THD *thd, time_t current_time,
       }
     }
   }
+  (void) pthread_mutex_unlock(&LOCK_log);
   DBUG_RETURN(error);
 }
 
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/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/set_var.cc b/sql/set_var.cc
index 7f3e808090d..d21a4c18716 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;
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/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 f1319010270..cf3ef111780 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,
@@ -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;
@@ -3809,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;
     }
   }
@@ -4149,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);
@@ -7182,7 +7198,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"));
@@ -7618,7 +7633,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;
@@ -7685,12 +7700,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_class.h b/sql/sql_class.h
index 71c13e001ee..ff4f3022c2b 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 *temporary_tables;
+  /**
+    List of tables that were opened with HANDLER OPEN and are
+    still in use by this thread.
   */
-  TABLE *open_tables, *temporary_tables, *handler_tables, *derived_tables;
+  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
@@ -2438,6 +2450,7 @@ 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 */
 
diff --git a/sql/sql_delete.cc b/sql/sql_delete.cc
index 7c868092921..7d8fc119cf9 100644
--- a/sql/sql_delete.cc
+++ b/sql/sql_delete.cc
@@ -906,9 +906,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 +958,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 +983,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..dc5db54e128 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -2254,7 +2254,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;
diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc
index 8ab4dee12ac..055dc1e8424 100644
--- a/sql/sql_lex.cc
+++ b/sql/sql_lex.cc
@@ -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;
   }
 }
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 93887db88e1..c98d22d9091 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -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;
@@ -250,6 +250,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 +267,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 +512,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))
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_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..fdd75fbe844 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[];
 
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 54b115af19f..7ffb83bc266 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))
     {
@@ -3966,7 +3966,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 +4014,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 +4083,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,
@@ -4258,7 +4241,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 +4307,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 +4332,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 +4781,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 +5697,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 +5786,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();
 
@@ -6984,7 +6953,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_yacc.yy b/sql/sql_yacc.yy
index d9a808bf8f7..81bb1c2dae9 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; }
diff --git a/sql/table.cc b/sql/table.cc
index e6f9b1086c7..f52c81d7889 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++;
 
@@ -2854,135 +2894,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
 
diff --git a/sql/table.h b/sql/table.h
index a276a9f32fd..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;
 
 
@@ -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();