WL1019: complete patch. Reapplied patch to the clean

tree to get rid of multiple typos in CS comments and
unify the patch.

20 files changed, 1576 insertions, 282 deletions

diff --git a/sql/ha_myisam.cc b/sql/ha_myisam.cc
index 8f73f77967a..5ca22fbc7c7 100644
--- a/sql/ha_myisam.cc
+++ b/sql/ha_myisam.cc
@@ -295,6 +295,25 @@ err:
 }
 #endif /* HAVE_REPLICATION */
 
+
+bool ha_myisam::check_if_locking_is_allowed(THD *thd, TABLE *table, uint count)
+{
+  /*
+    To be able to open and lock for reading system tables like 'mysql.proc',
+    when we already have some tables opened and locked, and avoid deadlocks
+    we have to disallow write-locking of these tables with any other tables.
+  */
+  if (table->s->system_table &&
+      table->reginfo.lock_type >= TL_WRITE_ALLOW_WRITE &&
+      count != 1)
+  {
+    my_error(ER_WRONG_LOCK_OF_SYSTEM_TABLE, MYF(0), table->s->db.str,
+             table->s->table_name.str);
+    return FALSE;
+  }
+  return TRUE;
+}
+
 	/* Name is here without an extension */
 
 int ha_myisam::open(const char *name, int mode, uint test_if_locked)
diff --git a/sql/ha_myisam.h b/sql/ha_myisam.h
index eb3ac9db7e4..1247437375f 100644
--- a/sql/ha_myisam.h
+++ b/sql/ha_myisam.h
@@ -60,6 +60,7 @@ class ha_myisam: public handler
   uint max_supported_key_part_length() const { return MI_MAX_KEY_LENGTH; }
   uint checksum() const;
 
+  virtual bool check_if_locking_is_allowed(THD *thd, TABLE *table, uint count);
   int open(const char *name, int mode, uint test_if_locked);
   int close(void);
   int write_row(byte * buf);
diff --git a/sql/handler.h b/sql/handler.h
index da89041d3a0..f85d98cd34e 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -1089,6 +1089,30 @@ public:
   {
     /* TODO: DBUG_ASSERT(inited == NONE); */
   }
+  /*
+    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   Number of locks already granted to the table
+
+    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(THD *thd, TABLE *table, uint count)
+  {
+    return TRUE;
+  }
   virtual int ha_initialise();
   int ha_open(TABLE *table, const char *name, int mode, int test_if_locked);
   bool update_auto_increment();
diff --git a/sql/lock.cc b/sql/lock.cc
index 76c511e4ba0..36648c33ac6 100644
--- a/sql/lock.cc
+++ b/sql/lock.cc
@@ -614,18 +614,12 @@ static MYSQL_LOCK *get_lock_data(THD *thd, TABLE **table_ptr, uint count,
       lock_count++;
     }
     /*
-      To be able to open and lock for reading system tables like 'mysql.proc',
-      when we already have some tables opened and locked, and avoid deadlocks
-      we have to disallow write-locking of these tables with any other tables.
+      Check if we can lock the table. For some tables we cannot do that
+      beacause of handler-specific locking issues.
     */
-    if (table_ptr[i]->s->system_table &&
-        table_ptr[i]->reginfo.lock_type >= TL_WRITE_ALLOW_WRITE &&
-        count != 1)
-    {
-      my_error(ER_WRONG_LOCK_OF_SYSTEM_TABLE, MYF(0), table_ptr[i]->s->db.str,
-               table_ptr[i]->s->table_name.str);
-      DBUG_RETURN(0);
-    }
+    if (!table_ptr[i]->file->check_if_locking_is_allowed(thd, table_ptr[i],
+                                                        count))
+      return 0;
   }
 
   if (!(sql_lock= (MYSQL_LOCK*)
diff --git a/sql/log.cc b/sql/log.cc
index e5da48196c8..b9318f96a84 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -34,7 +34,17 @@
 #include "message.h"
 #endif
 
-MYSQL_LOG mysql_log, mysql_slow_log, mysql_bin_log;
+/* max size of the log message */
+#define MAX_LOG_BUFFER_SIZE 1024
+#define MAX_USER_HOST_SIZE 512
+#define MAX_TIME_SIZE 32
+
+/* we need this for log files intialization */
+extern char *opt_logname, *opt_slow_logname;
+
+LOGGER logger;
+
+MYSQL_LOG mysql_bin_log;
 ulong sync_binlog_counter= 0;
 
 static bool test_if_number(const char *str,
@@ -95,6 +105,882 @@ handlerton binlog_hton = {
 };
 
 
+
+/*
+  Open log table of a given type (general or slow log)
+
+  SYNOPSIS
+    open_log_table()
+
+    log_type   type of the log table to open: LOG_GENERAL or 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_type)
+{
+  THD *log_thd, *curr= current_thd;
+  TABLE_LIST *table;
+  bool error= FALSE;
+  DBUG_ENTER("open_log_table");
+
+  switch (log_type) {
+  case 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 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:
+    DBUG_ASSERT(0);
+  }
+
+  /*
+    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;
+
+  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->locked_by_logger= TRUE;
+
+  /* restore thread settings */
+  if (curr)
+    curr->store_globals();
+  else
+  {
+    my_pthread_setspecific_ptr(THR_THD,  0);
+    my_pthread_setspecific_ptr(THR_MALLOC, 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;
+
+  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;
+}
+
+
+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_type   type of the log table to open: LOG_GENERAL or 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_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_type);
+}
+
+void Log_to_csv_event_handler::cleanup()
+{
+  close_log_table(LOG_GENERAL, FALSE);
+  close_log_table(LOG_SLOW, FALSE);
+  logger.is_log_tables_initialized= FALSE;
+}
+
+/* log event handlers */
+
+/*
+  Log command to the general log table
+
+  SYNOPSIS
+    log_general_to_csv()
+
+    event_time        command start timestamp
+    user_host         the pointer to the string with user@host info
+    user_host_len     length of the user_host string. this is computed once
+                      and passed to all general log event handlers
+    thread_id         Id of the thread, issued a query
+    command_type      the type of the command being logged
+    command_type_len  the length of the string above
+    sql_text          the very text of the query being executed
+    sql_text_len      the length of sql_text string
+
+  DESCRIPTION
+
+   Log given command to the general log table
+
+  RETURN
+    FALSE - OK
+    TRUE - error occured
+*/
+
+bool Log_to_csv_event_handler::
+  log_general(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)
+{
+  TABLE *table= general_log.table;
+
+  /* below should never happen */
+  if (unlikely(!logger.is_log_tables_initialized))
+    return FALSE;
+
+  /* log table entries are not replicated at the moment */
+  tmp_disable_binlog(current_thd);
+
+  general_log_thd->start_time= event_time;
+  /* set default value (which is CURRENT_TIMESTAMP) */
+  table->field[0]->set_null();
+
+  table->field[1]->store(user_host, user_host_len, &my_charset_latin1);
+  table->field[2]->store((longlong) thread_id);
+  table->field[3]->store((longlong) server_id);
+  table->field[4]->store(command_type, command_type_len, &my_charset_latin1);
+  table->field[5]->store(sql_text, sql_text_len, &my_charset_latin1);
+  table->file->ha_write_row(table->record[0]);
+
+  reenable_binlog(current_thd);
+
+  return FALSE;
+}
+
+
+/*
+  Log a query to the slow log table
+
+  SYNOPSIS
+    log_slow_to_csv()
+    thd               THD of the query
+    current_time      current timestamp
+    query_start_arg   command start timestamp
+    user_host         the pointer to the string with user@host info
+    user_host_len     length of the user_host string. this is computed once
+                      and passed to all general log event handlers
+    query_time        Amount of time the query took to execute (in seconds)
+    lock_time         Amount of time the query was locked (in seconds)
+    is_command        The flag, which determines, whether the sql_text is a
+                      query or an administrator command (these are treated
+                      differently by the old logging routines)
+    sql_text          the very text of the query or administrator command
+                      processed
+    sql_text_len      the length of sql_text string
+
+  DESCRIPTION
+
+   Log a query to the slow log table
+
+  RETURN
+    FALSE - OK
+    TRUE - error occured
+*/
+
+bool Log_to_csv_event_handler::
+  log_slow(THD *thd, time_t current_time, time_t query_start_arg,
+           const char *user_host, uint user_host_len,
+           longlong query_time, longlong lock_time, bool is_command,
+           const char *sql_text, uint sql_text_len)
+{
+  /* table variables */
+  TABLE *table= slow_log.table;
+
+  DBUG_ENTER("log_slow_to_csv");
+
+  /* below should never happen */
+  if (unlikely(!logger.is_log_tables_initialized))
+    return FALSE;
+
+  /* log table entries are not replicated at the moment */
+  tmp_disable_binlog(current_thd);
+
+  /*
+     Set start time for CURRENT_TIMESTAMP to the start of the query.
+     This will be default value for the field
+  */
+  slow_log_thd->start_time= query_start_arg;
+
+  /* set default value (which is CURRENT_TIMESTAMP) */
+  table->field[0]->set_null();
+
+  /* store the value */
+  table->field[1]->store(user_host, user_host_len, &my_charset_latin1);
+
+  if (query_start_arg)
+  {
+    /* fill in query_time field */
+    table->field[2]->store(query_time);
+    /* lock_time */
+    table->field[3]->store(lock_time);
+    /* rows_sent */
+    table->field[4]->store((longlong) thd->sent_row_count);
+    /* rows_examined */
+    table->field[5]->store((longlong) thd->examined_row_count);
+  }
+  else
+  {
+    table->field[2]->set_null();
+    table->field[3]->set_null();
+    table->field[4]->set_null();
+    table->field[5]->set_null();
+  }
+
+  if (thd->db)
+    /* fill database field */
+    table->field[6]->store(thd->db, thd->db_length, &my_charset_latin1);
+  else
+    table->field[6]->set_null();
+
+  if (thd->last_insert_id_used)
+    table->field[7]->store((longlong) thd->current_insert_id);
+  else
+    table->field[7]->set_null();
+
+  /* set value if we do an insert on autoincrement column */
+  if (thd->insert_id_used)
+    table->field[8]->store((longlong) thd->last_insert_id);
+  else
+    table->field[8]->set_null();
+
+  table->field[9]->store((longlong) server_id);
+
+  /* sql_text */
+  table->field[10]->store(sql_text,sql_text_len,
+                          &my_charset_latin1);
+
+  /* write the row */
+  table->file->ha_write_row(table->record[0]);
+
+  reenable_binlog(current_thd);
+
+  DBUG_RETURN(0);
+}
+
+bool Log_to_csv_event_handler::
+  log_error(enum loglevel level, const char *format, va_list args)
+{
+  /* No log table is implemented */
+  DBUG_ASSERT(0);
+  return FALSE;
+}
+
+bool Log_to_file_event_handler::
+  log_error(enum loglevel level, const char *format,
+            va_list args)
+{
+  return vprint_msg_to_log(level, format, args);
+}
+
+void Log_to_file_event_handler::init_pthread_objects()
+{
+  mysql_log.init_pthread_objects();
+  mysql_slow_log.init_pthread_objects();
+}
+
+
+/* Wrapper around MYSQL_LOG::write() for slow log */
+
+bool Log_to_file_event_handler::
+  log_slow(THD *thd, time_t current_time, time_t query_start_arg,
+           const char *user_host, uint user_host_len,
+           longlong query_time, longlong lock_time, bool is_command,
+           const char *sql_text, uint sql_text_len)
+{
+  return mysql_slow_log.write(thd, current_time, query_start_arg,
+                              user_host, user_host_len,
+                              query_time, lock_time, is_command,
+                              sql_text, sql_text_len);
+}
+
+
+/*
+   Wrapper around MYSQL_LOG::write() for general log. We need it since we
+   want all log event handlers to have the same signature.
+*/
+
+bool Log_to_file_event_handler::
+  log_general(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)
+{
+  return mysql_log.write(event_time, user_host, user_host_len,
+                         thread_id, command_type, command_type_len,
+                         sql_text, sql_text_len);
+}
+
+
+bool Log_to_file_event_handler::init()
+{
+  if (!is_initialized)
+  {
+    if (opt_slow_log)
+      mysql_slow_log.open_slow_log(opt_slow_logname);
+
+    if (opt_log)
+      mysql_log.open_query_log(opt_logname);
+
+    is_initialized= TRUE;
+  }
+
+  return FALSE;
+}
+
+
+void Log_to_file_event_handler::cleanup()
+{
+  mysql_log.cleanup();
+  mysql_slow_log.cleanup();
+}
+
+void Log_to_file_event_handler::flush()
+{
+  /* reopen log files */
+  mysql_log.new_file(1);
+  mysql_slow_log.new_file(1);
+}
+
+/*
+  Log error with all enabled log event handlers
+
+  SYNOPSIS
+    error_log_print()
+
+    level             The level of the error significance: NOTE,
+                      WARNING or ERROR.
+    format            format string for the error message
+    args              list of arguments for the format string
+
+  RETURN
+    FALSE - OK
+    TRUE - error occured
+*/
+
+bool LOGGER::error_log_print(enum loglevel level, const char *format,
+                             va_list args)
+{
+  bool error= FALSE;
+  Log_event_handler **current_handler= error_log_handler_list;
+
+  /* currently we don't need locking here as there is no error_log table */
+  while (*current_handler)
+    error= (*current_handler++)->log_error(level, format, args) || error;
+
+  return error;
+}
+
+
+void LOGGER::cleanup()
+{
+  DBUG_ASSERT(inited == 1);
+  (void) pthread_mutex_destroy(&LOCK_logger);
+  if (table_log_handler)
+    table_log_handler->cleanup();
+  if (file_log_handler)
+    file_log_handler->cleanup();
+}
+
+
+void LOGGER::close_log_table(uint log_type, bool lock_in_use)
+{
+  table_log_handler->close_log_table(log_type, lock_in_use);
+}
+
+
+/*
+  Perform basic log initialization: create file-based log handler and
+  init error log.
+*/
+void LOGGER::init_base()
+{
+  DBUG_ASSERT(inited == 0);
+  inited= 1;
+
+  /*
+    Here we create file log handler. We don't do it for the table log handler
+    here as it cannot be created so early. The reason is THD initialization,
+    which depends on the system variables (parsed later).
+  */
+  if (!file_log_handler)
+    file_log_handler= new Log_to_file_event_handler;
+
+  /* by default we use traditional error log */
+  init_error_log(LEGACY);
+
+  file_log_handler->init_pthread_objects();
+  (void) pthread_mutex_init(&LOCK_logger, MY_MUTEX_INIT_SLOW);
+}
+
+
+void LOGGER::init_log_tables()
+{
+  if (!table_log_handler)
+    table_log_handler= new Log_to_csv_event_handler;
+
+  if (!is_log_tables_initialized &&
+      !table_log_handler->init() && !file_log_handler->init())
+    is_log_tables_initialized= TRUE;
+}
+
+
+bool LOGGER::reopen_log_table(uint log_type)
+{
+  return table_log_handler->reopen_log_table(log_type);
+}
+
+
+bool LOGGER::flush_logs(THD *thd)
+{
+  TABLE_LIST close_slow_log, close_general_log;
+
+  /* reopen log 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;
+
+  /* reopen log files */
+  file_log_handler->flush();
+
+  /*
+    this will lock and wait for all but the logger thread to release the
+    tables. Then we could reopen log tables. Then release the name locks.
+  */
+  lock_and_wait_for_table_name(thd, &close_slow_log);
+  lock_and_wait_for_table_name(thd, &close_general_log);
+
+  /* deny others from logging to general and slow log, while reopening tables */
+  logger.lock();
+
+  table_log_handler->flush(thd, &close_slow_log, &close_general_log);
+
+  /* end of log tables flush */
+  logger.unlock();
+  return FALSE;
+}
+
+
+/*
+  Log slow query with all enabled log event handlers
+
+  SYNOPSIS
+    slow_log_print()
+
+    thd               THD of the query being logged
+    query             The query being logged
+    query_length      The length of the query string
+    query_start_arg   Query start timestamp
+
+  RETURN
+    FALSE - OK
+    TRUE - error occured
+*/
+
+bool LOGGER::slow_log_print(THD *thd, const char *query, uint query_length,
+                            time_t query_start_arg)
+{
+  bool error= FALSE;
+  Log_event_handler **current_handler= slow_log_handler_list;
+  bool is_command= FALSE;
+
+  char message_buff[MAX_LOG_BUFFER_SIZE];
+  char user_host_buff[MAX_USER_HOST_SIZE];
+
+  my_time_t current_time;
+  Security_context *sctx= thd->security_ctx;
+  uint message_buff_len= 0, user_host_len= 0;
+  longlong query_time= 0, lock_time= 0;
+  longlong last_insert_id= 0, insert_id= 0;
+
+  /*
+    Print the message to the buffer if we have slow log enabled
+  */
+
+  if (*slow_log_handler_list)
+  {
+    current_time= time(NULL);
+
+    if (!(thd->options & OPTION_UPDATE_LOG))
+      return 0;
+
+    lock();
+
+    /* fill in user_host value: the format is "%s[%s] @ %s [%s]" */
+    user_host_len= strxnmov(user_host_buff, MAX_USER_HOST_SIZE,
+                            sctx->priv_user ? sctx->priv_user : "", "[",
+                            sctx->user ? sctx->user : "", "] @ ",
+                            sctx->host ? sctx->host : "", " [",
+                            sctx->ip ? sctx->ip : "", "]", NullS) -
+      user_host_buff;
+
+    if (query_start_arg)
+    {
+      query_time= (longlong) (current_time - query_start_arg);
+      lock_time= (longlong) (thd->time_after_lock - query_start_arg);
+    }
+
+    if (thd->last_insert_id_used)
+      last_insert_id= (longlong) thd->current_insert_id;
+
+    /* set value if we do an insert on autoincrement column */
+    if (thd->insert_id_used)
+      insert_id= (longlong) thd->last_insert_id;
+
+    if (!query)
+    {
+      is_command= TRUE;
+      query= command_name[thd->command].str;
+      query_length= command_name[thd->command].length;
+    }
+
+    while (*current_handler)
+      error= (*current_handler++)->log_slow(thd, current_time, query_start_arg,
+                                            user_host_buff, user_host_len,
+                                            query_time, lock_time, is_command,
+                                            query, query_length) || error;
+
+    unlock();
+  }
+  return error;
+}
+
+bool LOGGER::general_log_print(THD *thd, enum enum_server_command command,
+                               const char *format, va_list args)
+{
+  bool error= FALSE;
+  Log_event_handler **current_handler= general_log_handler_list;
+
+  /*
+    Print the message to the buffer if we have at least one log event handler
+    enabled and want to log this king of commands
+  */
+  if (*general_log_handler_list && (what_to_log & (1L << (uint) command)))
+  {
+    char message_buff[MAX_LOG_BUFFER_SIZE];
+    char user_host_buff[MAX_USER_HOST_SIZE];
+    Security_context *sctx= thd->security_ctx;
+    ulong id;
+    uint message_buff_len= 0, user_host_len= 0;
+
+    if (thd)
+    {                                           /* Normal thread */
+      if ((thd->options & OPTION_LOG_OFF)
+#ifndef NO_EMBEDDED_ACCESS_CHECKS
+          && (sctx->master_access & SUPER_ACL)
+#endif
+         )
+      {
+        return 0;                         /* No logging */
+      }
+      id= thd->thread_id;
+    }
+    else
+      id=0;                                     /* Log from connect handler */
+
+    lock();
+    time_t current_time= time(NULL);
+
+    user_host_len= strxnmov(user_host_buff, MAX_USER_HOST_SIZE,
+                            sctx->priv_user ? sctx->priv_user : "", "[",
+                            sctx->user ? sctx->user : "", "] @ ",
+                            sctx->host ? sctx->host : "", " [",
+                            sctx->ip ? sctx->ip : "", "]", NullS) -
+                                                            user_host_buff;
+
+    /* prepare message */
+    if (format)
+      message_buff_len= my_vsnprintf(message_buff,
+                   sizeof(message_buff), format, args);
+    else
+      message_buff[0]= '\0';
+
+    while (*current_handler)
+      error+= (*current_handler++)->
+        log_general(current_time, user_host_buff,
+                   user_host_len, id,
+                   command_name[(uint) command].str,
+                   command_name[(uint) command].length,
+                   message_buff, message_buff_len) || error;
+    unlock();
+  }
+  return error;
+}
+
+void LOGGER::init_error_log(enum enum_printer error_log_printer)
+{
+  switch (error_log_printer) {
+  case NONE:
+    error_log_handler_list[0]= 0;
+    break;
+  case LEGACY:
+    error_log_handler_list[0]= file_log_handler;
+    error_log_handler_list[1]= 0;
+    break;
+    /* these two are disabled for now */
+  case CSV:
+    DBUG_ASSERT(0);
+    break;
+  case LEGACY_AND_CSV:
+    DBUG_ASSERT(0);
+    break;
+  }
+}
+
+void LOGGER::init_slow_log(enum enum_printer slow_log_printer)
+{
+  switch (slow_log_printer) {
+  case NONE:
+    slow_log_handler_list[0]= 0;
+    break;
+  case LEGACY:
+    slow_log_handler_list[0]= file_log_handler;
+    slow_log_handler_list[1]= 0;
+    break;
+  case CSV:
+    slow_log_handler_list[0]= table_log_handler;
+    slow_log_handler_list[1]= 0;
+    break;
+  case LEGACY_AND_CSV:
+    slow_log_handler_list[0]= file_log_handler;
+    slow_log_handler_list[1]= table_log_handler;
+    slow_log_handler_list[2]= 0;
+    break;
+  }
+}
+
+void LOGGER::init_general_log(enum enum_printer general_log_printer)
+{
+  switch (general_log_printer) {
+  case NONE:
+    general_log_handler_list[0]= 0;
+    break;
+  case LEGACY:
+    general_log_handler_list[0]= file_log_handler;
+    general_log_handler_list[1]= 0;
+    break;
+  case CSV:
+    general_log_handler_list[0]= table_log_handler;
+    general_log_handler_list[1]= 0;
+    break;
+  case LEGACY_AND_CSV:
+    general_log_handler_list[0]= file_log_handler;
+    general_log_handler_list[1]= table_log_handler;
+    general_log_handler_list[2]= 0;
+    break;
+  }
+}
+
+
+bool Log_to_csv_event_handler::flush(THD *thd, TABLE_LIST *close_slow_log,
+                                     TABLE_LIST *close_general_log)
+{
+  VOID(pthread_mutex_lock(&LOCK_open));
+  close_log_table(LOG_GENERAL, TRUE);
+  close_log_table(LOG_SLOW, TRUE);
+  close_general_log->next_local= close_slow_log;
+  query_cache_invalidate3(thd, close_general_log, 0);
+  unlock_table_name(thd, close_slow_log);
+  unlock_table_name(thd, close_general_log);
+  VOID(pthread_mutex_unlock(&LOCK_open));
+  return reopen_log_table(LOG_SLOW) || reopen_log_table(LOG_GENERAL);
+}
+
+/* the parameters are unused for the log tables */
+bool Log_to_csv_event_handler::init()
+{
+  /* we always open log tables. even if the logging is disabled */
+  return (open_log_table(LOG_GENERAL) || open_log_table(LOG_SLOW));
+}
+
+int LOGGER::set_handlers(enum enum_printer error_log_printer,
+                         enum enum_printer slow_log_printer,
+                         enum enum_printer general_log_printer)
+{
+  /* error log table is not supported yet */
+  DBUG_ASSERT(error_log_printer < CSV);
+
+  lock();
+
+  if ((slow_log_printer >= CSV || general_log_printer >= CSV) &&
+      !is_log_tables_initialized)
+  {
+    slow_log_printer= LEGACY;
+    general_log_printer= LEGACY;
+
+    sql_print_error("Failed to initialize log tables. "
+                    "Falling back to the old-fashioned logs");
+  }
+
+  init_error_log(error_log_printer);
+  init_slow_log(slow_log_printer);
+  init_general_log(general_log_printer);
+
+  unlock();
+
+  return 0;
+}
+
+
+/*
+  Close log table of a given type (general or slow log)
+
+  SYNOPSIS
+    close_log_table()
+
+    log_type       type of the log table to close: LOG_GENERAL or 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_type, bool lock_in_use)
+{
+  THD *log_thd, *curr= current_thd;
+  TABLE_LIST *table;
+
+  if (!logger.is_log_tables_initialized)
+    return;                                     /* do nothing */
+
+  switch (log_type) {
+  case LOG_GENERAL:
+    log_thd= general_log_thd;
+    table= &general_log;
+    break;
+  case LOG_SLOW:
+    log_thd= slow_log_thd;
+    table= &slow_log;
+    break;
+  default:
+    DBUG_ASSERT(0);
+  }
+
+  /*
+    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();
+  /* 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);
+  }
+}
+
+
 /*
   this function is mostly a placeholder.
   conceptually, binlog initialization (now mostly done in MYSQL_LOG::open)
@@ -1527,95 +2413,97 @@ err:
 
 
 /*
-  Write to normal (not rotable) log
-  This is the format for the 'normal' log.
+  Write a command to traditional general log file
+
+  SYNOPSIS
+    write()
+
+    event_time        command start timestamp
+    user_host         the pointer to the string with user@host info
+    user_host_len     length of the user_host string. this is computed once
+                      and passed to all general log  event handlers
+    thread_id         Id of the thread, issued a query
+    command_type      the type of the command being logged
+    command_type_len  the length of the string above
+    sql_text          the very text of the query being executed
+    sql_text_len      the length of sql_text string
+
+  DESCRIPTION
+
+   Log given command to to normal (not rotable) log file
+
+  RETURN
+    FASE - OK
+    TRUE - error occured
 */
 
-bool MYSQL_LOG::write(THD *thd,enum enum_server_command command,
-		      const char *format,...)
+bool MYSQL_LOG::write(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)
 {
-  if (is_open() && (what_to_log & (1L << (uint) command)))
+  char buff[32];
+  uint length= 0;
+  char time_buff[MAX_TIME_SIZE];
+  struct tm start;
+  uint time_buff_len= 0;
+
+  /* Test if someone closed between the is_open test and lock */
+  if (is_open())
   {
-    uint length;
-    int error= 0;
-    VOID(pthread_mutex_lock(&LOCK_log));
+    /* Note that my_b_write() assumes it knows the length for this */
+      if (event_time != last_time)
+      {
+        last_time= event_time;
 
-    /* Test if someone closed between the is_open test and lock */
-    if (is_open())
-    {
-      time_t skr;
-      ulong id;
-      va_list args;
-      va_start(args,format);
-      char buff[32];
-
-      if (thd)
-      {						// Normal thread
-	if ((thd->options & OPTION_LOG_OFF)
-#ifndef NO_EMBEDDED_ACCESS_CHECKS
-	    && (thd->security_ctx->master_access & SUPER_ACL)
-#endif
-)
-	{
-	  VOID(pthread_mutex_unlock(&LOCK_log));
-	  return 0;				// No logging
-	}
-	id=thd->thread_id;
-	if (thd->user_time || !(skr=thd->query_start()))
-	  skr=time(NULL);			// Connected
+        localtime_r(&event_time, &start);
+
+        time_buff_len= my_snprintf(time_buff, MAX_TIME_SIZE,
+                                   "%02d%02d%02d %2d:%02d:%02d",
+                                   start.tm_year % 100, start.tm_mon + 1,
+                                   start.tm_mday, start.tm_hour,
+                                   start.tm_min, start.tm_sec);
+
+        if (my_b_write(&log_file, (byte*) &time_buff, time_buff_len))
+          goto err;
       }
       else
-      {						// Log from connect handler
-	skr=time(NULL);
-	id=0;
-      }
-      if (skr != last_time)
-      {
-	last_time=skr;
-	struct tm tm_tmp;
-	struct tm *start;
-	localtime_r(&skr,&tm_tmp);
-	start=&tm_tmp;
-	/* Note that my_b_write() assumes it knows the length for this */
-	sprintf(buff,"%02d%02d%02d %2d:%02d:%02d\t",
-		start->tm_year % 100,
-		start->tm_mon+1,
-		start->tm_mday,
-		start->tm_hour,
-		start->tm_min,
-		start->tm_sec);
-	if (my_b_write(&log_file, (byte*) buff,16))
-	  error=errno;
-      }
-      else if (my_b_write(&log_file, (byte*) "\t\t",2) < 0)
-	error=errno;
-      length=my_sprintf(buff,
-			(buff, "%7ld %-11.11s", id,
-			 command_name[(uint) command]));
-      if (my_b_write(&log_file, (byte*) buff,length))
-	error=errno;
-      if (format)
-      {
-	if (my_b_write(&log_file, (byte*) " ",1) ||
-	    my_b_vprintf(&log_file,format,args) == (uint) -1)
-	  error=errno;
-      }
-      if (my_b_write(&log_file, (byte*) "\n",1) ||
-	  flush_io_cache(&log_file))
-	error=errno;
-      if (error && ! write_error)
-      {
-	write_error=1;
-	sql_print_error(ER(ER_ERROR_ON_WRITE),name,error);
-      }
-      va_end(args);
-    }
-    VOID(pthread_mutex_unlock(&LOCK_log));
-    return error != 0;
+        if (my_b_write(&log_file, (byte*) "\t\t" ,2) < 0)
+          goto err;
+
+    /* command_type, thread_id */
+    length= my_snprintf(buff, 32, "%5ld ", thread_id);
+
+    if (my_b_write(&log_file, (byte*) buff, length))
+      goto err;
+
+    if (my_b_write(&log_file, (byte*) command_type, command_type_len))
+      goto err;
+
+    if (my_b_write(&log_file, (byte*) "\t", 1))
+      goto err;
+
+    /* sql_text */
+    if (my_b_write(&log_file, (byte*) sql_text, sql_text_len))
+      goto err;
+
+    if (my_b_write(&log_file, (byte*) "\n", 1) ||
+        flush_io_cache(&log_file))
+      goto err;
   }
-  return 0;
+
+  return FALSE;
+err:
+
+  if (!write_error)
+  {
+    write_error= 1;
+    sql_print_error(ER(ER_ERROR_ON_WRITE), name, errno);
+  }
+  return TRUE;
 }
 
+
 bool MYSQL_LOG::flush_and_sync()
 {
   int err=0, fd=log_file.file;
@@ -2003,6 +2891,34 @@ err:
   DBUG_RETURN(error);
 }
 
+
+int error_log_print(enum loglevel level, const char *format,
+                    va_list args)
+{
+  return logger.error_log_print(level, format, args);
+}
+
+
+bool slow_log_print(THD *thd, const char *query, uint query_length,
+                    time_t query_start_arg)
+{
+  return logger.slow_log_print(thd, query, query_length, query_start_arg);
+}
+
+
+bool general_log_print(THD *thd, enum enum_server_command command,
+                       const char *format, ...)
+{
+  va_list args;
+  uint error= 0;
+
+  va_start(args, format);
+  error= logger.general_log_print(thd, command, format, args);
+  va_end(args);
+
+  return error;
+}
+
 void MYSQL_LOG::rotate_and_purge(uint flags)
 {
   if (!(flags & RP_LOCK_LOG_IS_ALREADY_LOCKED))
@@ -2147,71 +3063,86 @@ err:
 
 
 /*
-  Write to the slow query log.
+  Log a query to the traditional slow log file
+
+  SYNOPSIS
+    write()
+
+    thd               THD of the query
+    current_time      current timestamp
+    query_start_arg   command start timestamp
+    user_host         the pointer to the string with user@host info
+    user_host_len     length of the user_host string. this is computed once
+                      and passed to all general log event handlers
+    query_time        Amount of time the query took to execute (in seconds)
+    lock_time         Amount of time the query was locked (in seconds)
+    is_command        The flag, which determines, whether the sql_text is a
+                      query or an administrator command.
+    sql_text          the very text of the query or administrator command
+                      processed
+    sql_text_len      the length of sql_text string
+
+  DESCRIPTION
+
+   Log a query to the slow log file.
+
+  RETURN
+    FALSE - OK
+    TRUE - error occured
 */
 
-bool MYSQL_LOG::write(THD *thd,const char *query, uint query_length,
-		      time_t query_start_arg)
+bool MYSQL_LOG::write(THD *thd, time_t current_time, time_t query_start_arg,
+                      const char *user_host, uint user_host_len,
+                      longlong query_time, longlong lock_time, bool is_command,
+                      const char *sql_text, uint sql_text_len)
 {
-  bool error=0;
-  time_t current_time;
-  if (!is_open())
-    return 0;
+  bool error= 0;
   DBUG_ENTER("MYSQL_LOG::write");
 
-  VOID(pthread_mutex_lock(&LOCK_log));
+  if (!is_open())
+    DBUG_RETURN(0);
+
   if (is_open())
   {						// Safety agains reopen
-    int tmp_errno=0;
-    char buff[80],*end;
-    end=buff;
-    if (!(thd->options & OPTION_UPDATE_LOG))
-    {
-      VOID(pthread_mutex_unlock(&LOCK_log));
-      DBUG_RETURN(0);
-    }
-    if (!(specialflag & SPECIAL_SHORT_LOG_FORMAT) || query_start_arg)
+    int tmp_errno= 0;
+    char buff[80], *end;
+    uint buff_len;
+    end= buff;
+
+    if (!(specialflag & SPECIAL_SHORT_LOG_FORMAT))
     {
       Security_context *sctx= thd->security_ctx;
-      current_time=time(NULL);
       if (current_time != last_time)
       {
-        last_time=current_time;
-        struct tm tm_tmp;
-        struct tm *start;
-        localtime_r(&current_time,&tm_tmp);
-        start=&tm_tmp;
+        last_time= current_time;
+        struct tm start;
+        localtime_r(&current_time, &start);
+
+        buff_len= my_snprintf(buff, sizeof buff,
+                              "# Time: %02d%02d%02d %2d:%02d:%02d\n",
+                              start.tm_year % 100, start.tm_mon + 1,
+                              start.tm_mday, start.tm_hour,
+                              start.tm_min, start.tm_sec);
+
         /* Note that my_b_write() assumes it knows the length for this */
-        sprintf(buff,"# Time: %02d%02d%02d %2d:%02d:%02d\n",
-                start->tm_year % 100,
-                start->tm_mon+1,
-                start->tm_mday,
-                start->tm_hour,
-                start->tm_min,
-                start->tm_sec);
-        if (my_b_write(&log_file, (byte*) buff,24))
+        if (my_b_write(&log_file, (byte*) buff, buff_len))
           tmp_errno=errno;
       }
-      if (my_b_printf(&log_file, "# User@Host: %s[%s] @ %s [%s]\n",
-                      sctx->priv_user ?
-                      sctx->priv_user : "",
-                      sctx->user ? sctx->user : "",
-                      sctx->host ? sctx->host : "",
-                      sctx->ip ? sctx->ip : "") ==
-          (uint) -1)
+      if (my_b_printf(&log_file, "# User@Host: ", sizeof("# User@Host: ") - 1))
         tmp_errno=errno;
-    }
-    if (query_start_arg)
-    {
-      /* For slow query log */
-      if (my_b_printf(&log_file,
-                      "# Query_time: %lu  Lock_time: %lu  Rows_sent: %lu  Rows_examined: %lu\n",
-                      (ulong) (current_time - query_start_arg),
-                      (ulong) (thd->time_after_lock - query_start_arg),
-                      (ulong) thd->sent_row_count,
-                      (ulong) thd->examined_row_count) == (uint) -1)
+      if (my_b_printf(&log_file, user_host, user_host_len))
+        tmp_errno=errno;
+      if (my_b_write(&log_file, (byte*) "\n", 1))
         tmp_errno=errno;
     }
+    /* For slow query log */
+    if (my_b_printf(&log_file,
+                    "# Query_time: %lu  Lock_time: %lu"
+                    " Rows_sent: %lu  Rows_examined: %lu\n",
+                    (ulong) query_time, (ulong) lock_time,
+                    (ulong) thd->sent_row_count,
+                    (ulong) thd->examined_row_count) == (uint) -1)
+      tmp_errno=errno;
     if (thd->db && strcmp(thd->db,db))
     {						// Database changed
       if (my_b_printf(&log_file,"use %s;\n",thd->db) == (uint) -1)
@@ -2232,15 +3163,15 @@ bool MYSQL_LOG::write(THD *thd,const char *query, uint query_length,
         end=longlong10_to_str((longlong) thd->last_insert_id,end,-10);
       }
     }
-    if (thd->query_start_used)
-    {
-      if (query_start_arg != thd->query_start())
-      {
-        query_start_arg=thd->query_start();
-        end=strmov(end,",timestamp=");
-        end=int10_to_str((long) query_start_arg,end,10);
-      }
-    }
+
+    /*
+      This info used to show up randomly, depending on whether the query
+      checked the query start time or not. now we always write current
+      timestamp to the slow log
+    */
+    end= strmov(end, ",timestamp=");
+    end= int10_to_str((long) current_time, end, 10);
+
     if (end != buff)
     {
       *end++=';';
@@ -2249,14 +3180,13 @@ bool MYSQL_LOG::write(THD *thd,const char *query, uint query_length,
           my_b_write(&log_file, (byte*) buff+1,(uint) (end-buff)))
         tmp_errno=errno;
     }
-    if (!query)
+    if (is_command)
     {
-      end=strxmov(buff, "# administrator command: ",
-                  command_name[thd->command], NullS);
-      query_length=(ulong) (end-buff);
-      query=buff;
+      end= strxmov(buff, "# administrator command: ", NullS);
+      buff_len= (ulong) (end - buff);
+      my_b_write(&log_file, (byte*) buff, buff_len);
     }
-    if (my_b_write(&log_file, (byte*) query,query_length) ||
+    if (my_b_write(&log_file, (byte*) sql_text, sql_text_len) ||
         my_b_write(&log_file, (byte*) ";\n",2) ||
         flush_io_cache(&log_file))
       tmp_errno=errno;
@@ -2270,7 +3200,6 @@ bool MYSQL_LOG::write(THD *thd,const char *query, uint query_length,
       }
     }
   }
-  VOID(pthread_mutex_unlock(&LOCK_log));
   DBUG_RETURN(error);
 }
 
@@ -2463,6 +3392,7 @@ void print_buffer_to_file(enum loglevel level, const char *buffer)
   skr=time(NULL);
   localtime_r(&skr, &tm_tmp);
   start=&tm_tmp;
+
   fprintf(stderr, "%02d%02d%02d %2d:%02d:%02d [%s] %s\n",
           start->tm_year % 100,
           start->tm_mon+1,
@@ -2649,23 +3579,26 @@ void print_buffer_to_nt_eventlog(enum loglevel level, char *buff,
     to other functions to write that message to other logging sources.
 
   RETURN VALUES
-    void
+    The function always returns 0. The return value is present in the
+    signature to be compatible with other logging routines, which could
+    return an error (e.g. logging to the log tables)
 */
 
-void vprint_msg_to_log(enum loglevel level, const char *format, va_list args)
+int vprint_msg_to_log(enum loglevel level, const char *format, va_list args)
 {
   char   buff[1024];
   uint length;
   DBUG_ENTER("vprint_msg_to_log");
 
-  length= my_vsnprintf(buff, sizeof(buff)-5, format, args);
+  /* "- 5" is because of print_buffer_to_nt_eventlog() */
+  length= my_vsnprintf(buff, sizeof(buff) - 5, format, args);
   print_buffer_to_file(level, buff);
 
 #ifdef __NT__
   print_buffer_to_nt_eventlog(level, buff, length, sizeof(buff));
 #endif
 
-  DBUG_VOID_RETURN;
+  DBUG_RETURN(0);
 }
 
 
@@ -2675,7 +3608,7 @@ void sql_print_error(const char *format, ...)
   DBUG_ENTER("sql_print_error");
 
   va_start(args, format);
-  vprint_msg_to_log(ERROR_LEVEL, format, args);
+  error_log_print(ERROR_LEVEL, format, args);
   va_end(args);
 
   DBUG_VOID_RETURN;
@@ -2688,7 +3621,7 @@ void sql_print_warning(const char *format, ...)
   DBUG_ENTER("sql_print_warning");
 
   va_start(args, format);
-  vprint_msg_to_log(WARNING_LEVEL, format, args);
+  error_log_print(WARNING_LEVEL, format, args);
   va_end(args);
 
   DBUG_VOID_RETURN;
@@ -2701,7 +3634,7 @@ void sql_print_information(const char *format, ...)
   DBUG_ENTER("sql_print_information");
 
   va_start(args, format);
-  vprint_msg_to_log(INFORMATION_LEVEL, format, args);
+  error_log_print(INFORMATION_LEVEL, format, args);
   va_end(args);
 
   DBUG_VOID_RETURN;
diff --git a/sql/log.h b/sql/log.h
index ea2946c2d86..d709a73a391 100644
--- a/sql/log.h
+++ b/sql/log.h
@@ -132,6 +132,21 @@ typedef struct st_log_info
   ~st_log_info() { pthread_mutex_destroy(&lock);}
 } LOG_INFO;
 
+/*
+  Currently we have only 3 kinds of logging functions: old-fashioned
+  logs, stdout and csv logging routines.
+*/
+#define MAX_LOG_HANDLERS_NUM 3
+
+enum enum_printer
+{
+  NONE,
+  LEGACY,
+  CSV,
+  LEGACY_AND_CSV
+};
+
+
 class Log_event;
 class Rows_log_event;
 
@@ -276,10 +291,18 @@ public:
   bool open_index_file(const char *index_file_name_arg,
                        const char *log_name);
   void new_file(bool need_lock);
-  bool write(THD *thd, enum enum_server_command command,
-	     const char *format,...);
-  bool write(THD *thd, const char *query, uint query_length,
-	     time_t query_start=0);
+  /* log a command to the old-fashioned general log */
+  bool write(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);
+
+  /* log a query to the old-fashioned slow query log */
+  bool write(THD *thd, time_t current_time, time_t query_start_arg,
+             const char *user_host, uint user_host_len,
+             longlong query_time, longlong lock_time, bool is_command,
+             const char *sql_text, uint sql_text_len);
+
   bool write(Log_event* event_info); // binary log write
   bool write(THD *thd, IO_CACHE *cache, Log_event *commit_event);
 
@@ -329,4 +352,151 @@ public:
   inline uint32 get_open_count() { return open_count; }
 };
 
+class Log_event_handler
+{
+public:
+  virtual bool init()= 0;
+  virtual void cleanup()= 0;
+
+  virtual bool log_slow(THD *thd, time_t current_time,
+                        time_t query_start_arg, const char *user_host,
+                        uint user_host_len, longlong query_time,
+                        longlong lock_time, bool is_command,
+                        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,
+                           uint user_host_len, int thread_id,
+                           const char *command_type, uint command_type_len,
+                           const char *sql_text, uint sql_text_len)= 0;
+  virtual ~Log_event_handler() {}
+};
+
+
+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;
+  friend class LOGGER;
+  TABLE_LIST general_log, slow_log;
+
+private:
+  bool open_log_table(uint log_type);
+
+public:
+  Log_to_csv_event_handler();
+  ~Log_to_csv_event_handler();
+  virtual bool init();
+  virtual void cleanup();
+
+  virtual bool log_slow(THD *thd, time_t current_time,
+                        time_t query_start_arg, const char *user_host,
+                        uint user_host_len, longlong query_time,
+                        longlong lock_time, bool is_command,
+                        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,
+                           uint user_host_len, int thread_id,
+                           const char *command_type, uint command_type_len,
+                           const char *sql_text, uint sql_text_len);
+  bool flush(THD *thd, TABLE_LIST *close_slow_Log,
+             TABLE_LIST* close_general_log);
+  void close_log_table(uint log_type, bool lock_in_use);
+  bool reopen_log_table(uint log_type);
+};
+
+
+class Log_to_file_event_handler: public Log_event_handler
+{
+  MYSQL_LOG mysql_log, mysql_slow_log;
+  bool is_initialized;
+public:
+  Log_to_file_event_handler(): is_initialized(FALSE)
+  {}
+  virtual bool init();
+  virtual void cleanup();
+
+  virtual bool log_slow(THD *thd, time_t current_time,
+                        time_t query_start_arg, const char *user_host,
+                        uint user_host_len, longlong query_time,
+                        longlong lock_time, bool is_command,
+                        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,
+                           uint user_host_len, int thread_id,
+                           const char *command_type, uint command_type_len,
+                           const char *sql_text, uint sql_text_len);
+  void flush();
+  void init_pthread_objects();
+};
+
+
+/* Class which manages slow, general and error log event handlers */
+class LOGGER
+{
+  pthread_mutex_t LOCK_logger;
+  /* flag to check whether logger mutex is initialized */
+  uint inited;
+
+  /* available log handlers */
+  Log_to_csv_event_handler *table_log_handler;
+  Log_to_file_event_handler *file_log_handler;
+
+  /* NULL-terminated arrays of log handlers */
+  Log_event_handler *error_log_handler_list[MAX_LOG_HANDLERS_NUM + 1];
+  Log_event_handler *slow_log_handler_list[MAX_LOG_HANDLERS_NUM + 1];
+  Log_event_handler *general_log_handler_list[MAX_LOG_HANDLERS_NUM + 1];
+
+public:
+
+  bool is_log_tables_initialized;
+
+  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); }
+  /*
+    We want to initialize all log mutexes as soon as possible,
+    but we cannot do it in constructor, as safe_mutex relies on
+    initialization, performed by MY_INIT(). This why this is done in
+    this function.
+  */
+  void init_base();
+  void init_log_tables();
+  bool flush_logs(THD *thd);
+  THD *get_general_log_thd()
+  {
+    return (THD *) table_log_handler->general_log_thd;
+  }
+  THD *get_slow_log_thd()
+  {
+    return (THD *) table_log_handler->slow_log_thd;
+  }
+  void cleanup();
+  bool error_log_print(enum loglevel level, const char *format,
+                      va_list args);
+  bool slow_log_print(THD *thd, const char *query, uint query_length,
+                      time_t query_start_arg);
+  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);
+
+  /* we use this function to setup all enabled log event handlers */
+  int set_handlers(enum enum_printer error_log_printer,
+                   enum enum_printer slow_log_printer,
+                   enum enum_printer general_log_printer);
+  void init_error_log(enum enum_printer error_log_printer);
+  void init_slow_log(enum enum_printer slow_log_printer);
+  void init_general_log(enum enum_printer general_log_printer);
+ };
+
 #endif /* LOG_H */
diff --git a/sql/log_event.cc b/sql/log_event.cc
index 6249f699b05..c911baf361c 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -1795,7 +1795,7 @@ START SLAVE; . Query: '%s'", expected_error, thd->query);
 
     /* If the query was not ignored, it is printed to the general log */
     if (thd->net.last_errno != ER_SLAVE_IGNORED_TABLE)
-      mysql_log.write(thd,COM_QUERY,"%s",thd->query);
+      general_log_print(thd, COM_QUERY, "%s", thd->query);
 
 compare_errors:
 
@@ -3513,7 +3513,8 @@ void Xid_log_event::print(FILE* file, PRINT_EVENT_INFO* print_event_info)
 int Xid_log_event::exec_event(struct st_relay_log_info* rli)
 {
   /* For a slave Xid_log_event is COMMIT */
-  mysql_log.write(thd,COM_QUERY,"COMMIT /* implicit, from Xid_log_event */");
+  general_log_print(thd, COM_QUERY,
+                    "COMMIT /* implicit, from Xid_log_event */");
   return end_trans(thd, COMMIT) || Log_event::exec_event(rli);
 }
 #endif /* !MYSQL_CLIENT */
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index cabb71e0263..955983981d1 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -1171,12 +1171,23 @@ int key_rec_cmp(void *key_info, byte *a, byte *b);
 bool init_errmessage(void);
 void sql_perror(const char *message);
 
-void vprint_msg_to_log(enum loglevel level, const char *format, va_list args);
+int vprint_msg_to_log(enum loglevel level, const char *format, va_list args);
 void sql_print_error(const char *format, ...);
 void sql_print_warning(const char *format, ...);
 void sql_print_information(const char *format, ...);
 
+/* type of the log table */
+#define LOG_SLOW 1
+#define LOG_GENERAL 2
 
+int error_log_print(enum loglevel level, const char *format,
+                    va_list args);
+
+bool slow_log_print(THD *thd, const char *query, uint query_length,
+                    time_t query_start_arg);
+
+bool general_log_print(THD *thd, enum enum_server_command command,
+                       const char *format,...);
 
 bool fn_format_relative_to_data_home(my_string to, const char *name,
 				     const char *dir, const char *extension);
@@ -1217,7 +1228,7 @@ extern char *mysql_data_home,server_version[SERVER_VERSION_LENGTH],
             def_ft_boolean_syntax[sizeof(ft_boolean_syntax)];
 #define mysql_tmpdir (my_tmpdir(&mysql_tmpdir_list))
 extern MY_TMPDIR mysql_tmpdir_list;
-extern const char *command_name[];
+extern LEX_STRING command_name[];
 extern const char *first_keyword, *my_localhost, *delayed_user, *binary_keyword;
 extern const char **errmesg;			/* Error messages */
 extern const char *myisam_recover_options_str;
@@ -1279,6 +1290,7 @@ extern my_bool locked_in_memory;
 extern bool opt_using_transactions, mysqld_embedded;
 extern bool using_update_log, opt_large_files, server_id_supplied;
 extern bool opt_log, opt_update_log, opt_bin_log, opt_slow_log, opt_error_log;
+extern bool opt_old_log_format;
 extern bool opt_disable_networking, opt_skip_show_db;
 extern my_bool opt_character_set_client_handshake;
 extern bool volatile abort_loop, shutdown_in_progress, grant_option;
@@ -1300,7 +1312,9 @@ extern char *default_tz_name;
 extern my_bool opt_large_pages;
 extern uint opt_large_page_size;
 
-extern MYSQL_LOG mysql_log,mysql_slow_log,mysql_bin_log;
+extern MYSQL_LOG mysql_bin_log;
+extern LOGGER logger;
+extern TABLE_LIST general_log, slow_log;
 extern FILE *bootstrap_file;
 extern int bootstrap_error;
 extern FILE *stderror_file;
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 5a78ab87d00..5e1dfc089ed 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -331,6 +331,9 @@ static my_bool opt_sync_bdb_logs;
 
 bool opt_log, opt_update_log, opt_bin_log, opt_slow_log;
 bool opt_error_log= IF_WIN(1,0);
+#ifdef WITH_CSV_STORAGE_ENGINE
+bool opt_old_log_format, opt_both_log_formats;
+#endif
 bool opt_disable_networking=0, opt_skip_show_db=0;
 my_bool opt_character_set_client_handshake= 1;
 bool server_id_supplied = 0;
@@ -603,6 +606,7 @@ char *opt_relay_logname = 0, *opt_relaylog_index_name=0;
 my_bool master_ssl;
 char *master_ssl_key, *master_ssl_cert;
 char *master_ssl_ca, *master_ssl_capath, *master_ssl_cipher;
+char *opt_logname, *opt_slow_logname;
 
 /* Static variables */
 
@@ -610,8 +614,8 @@ static bool kill_in_progress, segfaulted;
 static my_bool opt_do_pstack, opt_bootstrap, opt_myisam_log;
 static int cleanup_done;
 static ulong opt_specialflag, opt_myisam_block_size;
-static char *opt_logname, *opt_update_logname, *opt_binlog_index_name;
-static char *opt_slow_logname, *opt_tc_heuristic_recover;
+static char *opt_update_logname, *opt_binlog_index_name;
+static char *opt_tc_heuristic_recover;
 static char *mysql_home_ptr, *pidfile_name_ptr;
 static char **defaults_argv;
 static char *opt_bin_logname;
@@ -1138,8 +1142,7 @@ void clean_up(bool print_message)
   if (cleanup_done++)
     return; /* purecov: inspected */
 
-  mysql_log.cleanup();
-  mysql_slow_log.cleanup();
+  logger.cleanup();
   /*
     make sure that handlers finish up
     what they have that is dependent on the binlog
@@ -2389,6 +2392,9 @@ pthread_handler_t signal_hand(void *arg __attribute__((unused)))
 #ifdef EXTRA_DEBUG
       sql_print_information("Got signal %d to shutdown mysqld",sig);
 #endif
+      /* switch to the old log message processing */
+      logger.set_handlers(LEGACY, opt_slow_log ? LEGACY:NONE,
+                          opt_log ? LEGACY:NONE);
       DBUG_PRINT("info",("Got signal: %d  abort_loop: %d",sig,abort_loop));
       if (!abort_loop)
       {
@@ -2416,6 +2422,9 @@ pthread_handler_t signal_hand(void *arg __attribute__((unused)))
 			      REFRESH_THREADS | REFRESH_HOSTS),
 			     (TABLE_LIST*) 0, &not_used); // Flush logs
       }
+      /* reenable logs after the options were reloaded */
+      logger.set_handlers(LEGACY, opt_slow_log ? CSV:NONE,
+                          opt_log ? CSV:NONE);
       break;
 #ifdef USE_ONE_SIGNAL_HAND
     case THR_SERVER_ALARM:
@@ -2680,8 +2689,6 @@ static int init_common_variables(const char *conf_file_name, int argc,
     global MYSQL_LOGs in their constructors, because then they would be inited
     before MY_INIT(). So we do it here.
   */
-  mysql_log.init_pthread_objects();
-  mysql_slow_log.init_pthread_objects();
   mysql_bin_log.init_pthread_objects();
 
   if (gethostname(glob_hostname,sizeof(glob_hostname)-4) < 0)
@@ -3047,9 +3054,48 @@ static int init_server_components()
 #ifdef HAVE_REPLICATION
   init_slave_list();
 #endif
-  /* Setup log files */
-  if (opt_log)
-    mysql_log.open_query_log(opt_logname);
+  /* Setup logs */
+
+  /* enable old-fashioned error log */
+  if (opt_error_log)
+  {
+    if (!log_error_file_ptr[0])
+      fn_format(log_error_file, glob_hostname, mysql_data_home, ".err",
+                MY_REPLACE_EXT); /* replace '.<domain>' by '.err', bug#4997 */
+    else
+      fn_format(log_error_file, log_error_file_ptr, mysql_data_home, ".err",
+                MY_UNPACK_FILENAME | MY_SAFE_PATH);
+    if (!log_error_file[0])
+      opt_error_log= 1;				// Too long file name
+    else
+    {
+#ifndef EMBEDDED_LIBRARY
+      if (freopen(log_error_file, "a+", stdout))
+#endif
+        freopen(log_error_file, "a+", stderr);
+    }
+  }
+
+#ifdef WITH_CSV_STORAGE_ENGINE
+  logger.init_log_tables();
+
+  if (opt_old_log_format || (have_csv_db != SHOW_OPTION_YES))
+    logger.set_handlers(LEGACY, opt_slow_log ? LEGACY:NONE,
+                        opt_log ? LEGACY:NONE);
+  else
+    if (opt_both_log_formats)
+      logger.set_handlers(LEGACY,
+                          opt_slow_log ? LEGACY_AND_CSV:NONE,
+                          opt_log ? LEGACY_AND_CSV:NONE);
+    else
+      /* the default is CSV log tables */
+      logger.set_handlers(LEGACY, opt_slow_log ? CSV:NONE,
+                          opt_log ? CSV:NONE);
+#else
+  logger.set_handlers(LEGACY, opt_slow_log ? LEGACY:NONE,
+                      opt_log ? LEGACY:NONE);
+#endif
+
   if (opt_update_log)
   {
     /*
@@ -3146,9 +3192,6 @@ with --log-bin instead.");
               array_elements(binlog_format_names)-1);
   opt_binlog_format= binlog_format_names[opt_binlog_format_id];
 
-  if (opt_slow_log)
-    mysql_slow_log.open_slow_log(opt_slow_logname);
-
 #ifdef HAVE_REPLICATION
   if (opt_log_slave_updates && replicate_same_server_id)
   {
@@ -3160,25 +3203,6 @@ server.");
   }
 #endif
 
-  if (opt_error_log)
-  {
-    if (!log_error_file_ptr[0])
-      fn_format(log_error_file, glob_hostname, mysql_data_home, ".err",
-                MY_REPLACE_EXT); /* replace '.<domain>' by '.err', bug#4997 */
-    else
-      fn_format(log_error_file, log_error_file_ptr, mysql_data_home, ".err",
-		MY_UNPACK_FILENAME | MY_SAFE_PATH);
-    if (!log_error_file[0])
-      opt_error_log= 1;				// Too long file name
-    else
-    {
-#ifndef EMBEDDED_LIBRARY
-      if (freopen(log_error_file, "a+", stdout))
-#endif
-	stderror_file= freopen(log_error_file, "a+", stderr);
-    }
-  }
-
   if (opt_bin_log)
   {
     char buf[FN_REFLEN];
@@ -3432,6 +3456,12 @@ int main(int argc, char **argv)
 
   MY_INIT(argv[0]);		// init my_sys library & pthreads
 
+  /*
+    Perform basic logger initialization logger. Should be called after
+    MY_INIT, as it initializes mutexes. Log tables are inited later.
+  */
+  logger.init_base();
+
 #ifdef _CUSTOMSTARTUPCONFIG_
   if (_cust_check_startup())
   {
@@ -3577,6 +3607,7 @@ we force server id to 2, but this MySQL server will not act as a slave.");
   */
   error_handler_hook= my_message_sql;
   start_signal_handler();				// Creates pidfile
+
   if (acl_init(opt_noacl) ||
       my_tz_init((THD *)0, default_tz_name, opt_bootstrap))
   {
@@ -3701,7 +3732,7 @@ we force server id to 2, but this MySQL server will not act as a slave.");
   clean_up_mutexes();
   shutdown_events();
   my_end(opt_endinfo ? MY_CHECK_ERROR | MY_GIVE_INFO : 0);
- 
+
   exit(0);
   return(0);					/* purecov: deadcode */
 }
@@ -4639,7 +4670,7 @@ enum options_mysqld
   OPT_REPLICATE_IGNORE_TABLE,  OPT_REPLICATE_WILD_DO_TABLE,
   OPT_REPLICATE_WILD_IGNORE_TABLE, OPT_REPLICATE_SAME_SERVER_ID,
   OPT_DISCONNECT_SLAVE_EVENT_COUNT, OPT_TC_HEURISTIC_RECOVER,
-  OPT_ABORT_SLAVE_EVENT_COUNT,
+  OPT_ABORT_SLAVE_EVENT_COUNT, OPT_OLD_LOG_FORMAT, OPT_BOTH_LOG_FORMATS,
   OPT_INNODB_DATA_HOME_DIR,
   OPT_INNODB_DATA_FILE_PATH,
   OPT_INNODB_LOG_GROUP_HOME_DIR,
@@ -5195,6 +5226,16 @@ Disable with --skip-innodb-doublewrite.", (gptr*) &innobase_use_doublewrite,
     "Log slow queries to this log file. Defaults logging to hostname-slow.log file. Must be enabled to activate other slow log options.",
    (gptr*) &opt_slow_logname, (gptr*) &opt_slow_logname, 0, GET_STR, OPT_ARG,
    0, 0, 0, 0, 0, 0},
+#ifdef WITH_CSV_STORAGE_ENGINE
+  {"old-log-format", OPT_OLD_LOG_FORMAT,
+   "Enable old log file format. (No SELECT * FROM logs)",
+   (gptr*) &opt_old_log_format, 0, 0, GET_BOOL, OPT_ARG,
+   0, 0, 0, 0, 0, 0},
+  {"both-log-formats", OPT_BOTH_LOG_FORMATS,
+   "Enable old log file format along with log tables",
+   (gptr*) &opt_both_log_formats, 0, 0, GET_BOOL, OPT_ARG,
+   0, 0, 0, 0, 0, 0},
+#endif
   {"log-tc", OPT_LOG_TC,
    "Path to transaction coordinator log (used for transactions that affect "
    "more than one storage engine, when binary log is disabled)",
@@ -6886,6 +6927,10 @@ static void mysql_init_variables(void)
   opt_skip_slave_start= opt_reckless_slave = 0;
   mysql_home[0]= pidfile_name[0]= log_error_file[0]= 0;
   opt_log= opt_update_log= opt_slow_log= 0;
+#ifdef WITH_CSV_STORAGE_ENGINE
+  opt_old_log_format= 0;
+  opt_both_log_formats= 0;
+#endif
   opt_bin_log= 0;
   opt_disable_networking= opt_skip_show_db=0;
   opt_logname= opt_update_logname= opt_binlog_index_name= opt_slow_logname= 0;
@@ -7294,8 +7339,16 @@ get_one_option(int optid, const struct my_option *opt __attribute__((unused)),
   }
 #endif /* HAVE_REPLICATION */
   case (int) OPT_SLOW_QUERY_LOG:
-    opt_slow_log=1;
+    opt_slow_log= 1;
     break;
+#ifdef WITH_CSV_STORAGE_ENGINE
+  case (int) OPT_OLD_LOG_FORMAT:
+    opt_old_log_format= 1;
+    break;
+  case (int) OPT_BOTH_LOG_FORMATS:
+    opt_both_log_formats= 1;
+    break;
+#endif
   case (int) OPT_SKIP_NEW:
     opt_specialflag|= SPECIAL_NO_NEW_FUNC;
     delay_key_write_options= (uint) DELAY_KEY_WRITE_NONE;
diff --git a/sql/share/errmsg.txt b/sql/share/errmsg.txt
index 38b1666c236..81bff69b22c 100644
--- a/sql/share/errmsg.txt
+++ b/sql/share/errmsg.txt
@@ -5794,3 +5794,7 @@ ER_EVENT_DATA_TOO_LONG
 ER_DROP_INDEX_FK
         eng "Cannot drop index '%-.64s': needed in a foreign key constraint"
         ger "Kann Index '%-.64s' nicht löschen: wird für einen einen Fremdschlüssel benötigt"
+ER_CANT_WRITE_LOCK_LOG_TABLE
+        eng "You can't write-lock a log table. Only read access is possible."
+ER_CANT_READ_LOCK_LOG_TABLE
+        eng "You can't use usual read lock with log tables. Try READ LOCAL instead."
diff --git a/sql/slave.cc b/sql/slave.cc
index 41a13f2f5c5..edca614159a 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -4354,8 +4354,8 @@ replication resumed in log '%s' at position %s", mi->user,
     else
     {
       change_rpl_status(RPL_IDLE_SLAVE,RPL_ACTIVE_SLAVE);
-      mysql_log.write(thd, COM_CONNECT_OUT, "%s@%s:%d",
-		      mi->user, mi->host, mi->port);
+      general_log_print(thd, COM_CONNECT_OUT, "%s@%s:%d",
+                        mi->user, mi->host, mi->port);
     }
 #ifdef SIGNAL_WITH_VIO_CLOSE
     thd->set_active_vio(mysql->net.vio);
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index c695fe40d7f..44b3a22ec52 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -835,7 +835,8 @@ 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 (remove_table_from_cache(thd, table->db, table->table_name,
+      if ((!table->table || !table->table->s->log_table) &&
+          remove_table_from_cache(thd, table->db, table->table_name,
                                   RTFC_OWNED_BY_THD_FLAG))
 	found=1;
     }
@@ -869,7 +870,8 @@ bool close_cached_tables(THD *thd, bool if_wait_for_refresh,
       for (uint idx=0 ; idx < open_cache.records ; idx++)
       {
 	TABLE *table=(TABLE*) hash_element(&open_cache,idx);
-	if ((table->s->version) < refresh_version && table->db_stat)
+	if (!table->s->log_table &&
+            ((table->s->version) < refresh_version && table->db_stat))
 	{
 	  found=1;
 	  pthread_cond_wait(&COND_refresh,&LOCK_open);
@@ -1852,7 +1854,7 @@ TABLE *open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
   if (!thd->open_tables)
     thd->version=refresh_version;
   else if ((thd->version != refresh_version) &&
-           ! (flags & MYSQL_LOCK_IGNORE_FLUSH))
+           ! (flags & MYSQL_LOCK_IGNORE_FLUSH) && !table->s->log_table)
   {
     /* Someone did a refresh while thread was opening tables */
     if (refresh)
@@ -1873,7 +1875,11 @@ TABLE *open_table(THD *thd, TABLE_LIST *table_list, MEM_ROOT *mem_root,
   {
     if (table->s->version != refresh_version)
     {
-      if (flags & MYSQL_LOCK_IGNORE_FLUSH)
+      /*
+        Don't close tables if we are working with a log table or were
+        asked not to close the table explicitly
+      */
+      if (flags & MYSQL_LOCK_IGNORE_FLUSH || table->s->log_table)
       {
         /* Force close at once after usage */
         thd->version= table->s->version;
@@ -2236,6 +2242,10 @@ void close_old_data_files(THD *thd, TABLE *table, bool abort_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)
@@ -2254,9 +2264,10 @@ bool table_is_used(TABLE *table, bool wait_for_name_lock)
          search= (TABLE*) hash_next(&open_cache, (byte*) key,
                                     key_length, &state))
     {
-      DBUG_PRINT("info", ("share: 0x%lx  locked_by_flush: %d  "
-                          "locked_by_name: %d  db_stat: %u  version: %u",
-                          (ulong) search->s,
+      DBUG_PRINT("info", ("share: 0x%lx  locked_by_logger: %d "
+                          "locked_by_flush: %d  locked_by_name: %d "
+                          "db_stat: %u  version: %u",
+                          (ulong) search->s, search->locked_by_logger,
                           search->locked_by_flush, search->locked_by_name,
                           search->db_stat,
                           search->s->version));
@@ -2267,12 +2278,15 @@ bool table_is_used(TABLE *table, bool wait_for_name_lock)
         - There is an name lock on it (Table is to be deleted or altered)
         - 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)
+        (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_name && wait_for_name_lock ||
-          search->locked_by_flush ||
-          (search->db_stat && search->s->version < refresh_version))
-	return 1;
+      if (!search->locked_by_logger &&
+          (search->locked_by_name && wait_for_name_lock ||
+           search->locked_by_flush ||
+           (search->db_stat && search->s->version < refresh_version)))
+        return 1;
     }
   } while ((table=table->next));
   DBUG_RETURN(0);
@@ -5867,6 +5881,7 @@ bool remove_table_from_cache(THD *thd, const char *db, const char *table_name,
                                    &state))
     {
       THD *in_use;
+
       table->s->version=0L;		/* Free when thread is ready */
       if (!(in_use=table->in_use))
       {
diff --git a/sql/sql_db.cc b/sql/sql_db.cc
index fa01f98d723..5ffa4fd76ed 100644
--- a/sql/sql_db.cc
+++ b/sql/sql_db.cc
@@ -1158,8 +1158,8 @@ bool mysql_change_db(THD *thd, const char *name, bool no_access_check)
                sctx->priv_user,
                sctx->priv_host,
                dbname);
-      mysql_log.write(thd, COM_INIT_DB, ER(ER_DBACCESS_DENIED_ERROR),
-                      sctx->priv_user, sctx->priv_host, dbname);
+      general_log_print(thd, COM_INIT_DB, ER(ER_DBACCESS_DENIED_ERROR),
+                        sctx->priv_user, sctx->priv_host, dbname);
       my_free(dbname,MYF(0));
       DBUG_RETURN(1);
     }
diff --git a/sql/sql_delete.cc b/sql/sql_delete.cc
index 7b30b2d4a3d..996a4c2ac8a 100644
--- a/sql/sql_delete.cc
+++ b/sql/sql_delete.cc
@@ -857,6 +857,8 @@ 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;
+  TABLE_LIST *tmp_table_list;
   uint path_length;
   DBUG_ENTER("mysql_truncate");
 
@@ -905,13 +907,36 @@ bool mysql_truncate(THD *thd, TABLE_LIST *table_list, bool dont_send_ok)
                                       HTON_CAN_RECREATE)
         || thd->lex->sphead)
       goto trunc_by_del;
+
     if (lock_and_wait_for_table_name(thd, table_list))
       DBUG_RETURN(TRUE);
   }
 
-  // Remove the .frm extension
-  // AIX 5.2 64-bit compiler bug (BUG#16155): this crashes, replacement works.
-  //   *(path + path_length - reg_ext_length)= '\0';
+  /* close log tables in use */
+  if (!my_strcasecmp(system_charset_info, table_list->db, "mysql"))
+  {
+    if (!my_strcasecmp(system_charset_info, table_list->table_name,
+                       "general_log"))
+    {
+      lock_logger= 1;
+      logger.lock();
+      logger.close_log_table(LOG_GENERAL, FALSE);
+      closed_log_tables= closed_log_tables | LOG_GENERAL;
+    }
+    else
+      if (!my_strcasecmp(system_charset_info, table_list->table_name,
+                         "slow_log"))
+      {
+        lock_logger= 1;
+        logger.lock();
+        logger.close_log_table(LOG_SLOW, FALSE);
+        closed_log_tables= closed_log_tables | LOG_SLOW;
+      }
+  }
+
+  // Remove the .frm extension AIX 5.2 64-bit compiler bug (BUG#16155): this
+  // crashes, replacement works.  *(path + path_length - reg_ext_length)=
+  // '\0';
   path[path_length - reg_ext_length] = 0;
   error= ha_create_table(thd, path, table_list->db, table_list->table_name,
                          &create_info, 1);
@@ -937,6 +962,14 @@ end:
     VOID(pthread_mutex_lock(&LOCK_open));
     unlock_table_name(thd, table_list);
     VOID(pthread_mutex_unlock(&LOCK_open));
+
+    if (closed_log_tables & LOG_SLOW)
+      logger.reopen_log_table(LOG_SLOW);
+
+    if (closed_log_tables & LOG_GENERAL)
+      logger.reopen_log_table(LOG_GENERAL);
+    if (lock_logger)
+      logger.unlock();
   }
   else if (error)
   {
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index ecde4d01ae1..f7d687a6a98 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -73,14 +73,23 @@ static bool append_file_to_dir(THD *thd, const char **filename_ptr,
 
 const char *any_db="*any*";	// Special symbol for check_access
 
-const char *command_name[]={
-  "Sleep", "Quit", "Init DB", "Query", "Field List", "Create DB",
-  "Drop DB", "Refresh", "Shutdown", "Statistics", "Processlist",
-  "Connect","Kill","Debug","Ping","Time","Delayed insert","Change user",
-  "Binlog Dump","Table Dump",  "Connect Out", "Register Slave",
-  "Prepare", "Execute", "Long Data", "Close stmt",
-  "Reset stmt", "Set option", "Fetch", "Daemon",
-  "Error"					// Last command number
+LEX_STRING command_name[]={
+  STRING_WITH_LEN("Sleep"), STRING_WITH_LEN("Quit"),
+  STRING_WITH_LEN("Init DB"), STRING_WITH_LEN("Query"),
+  STRING_WITH_LEN("Field List"), STRING_WITH_LEN("Create DB"),
+  STRING_WITH_LEN("Drop DB"), STRING_WITH_LEN("Refresh"),
+  STRING_WITH_LEN("Shutdown"), STRING_WITH_LEN("Statistics"),
+  STRING_WITH_LEN("Processlist"), STRING_WITH_LEN("Connect"),
+  STRING_WITH_LEN("Kill"), STRING_WITH_LEN("Debug"),
+  STRING_WITH_LEN("Ping"), STRING_WITH_LEN("Time"),
+  STRING_WITH_LEN("Delayed insert"), STRING_WITH_LEN("Change user"),
+  STRING_WITH_LEN("Binlog Dump"), STRING_WITH_LEN("Table Dump"),
+  STRING_WITH_LEN("Connect Out"), STRING_WITH_LEN("Register Slave"),
+  STRING_WITH_LEN("Prepare"), STRING_WITH_LEN("Execute"),
+  STRING_WITH_LEN("Long Data"), STRING_WITH_LEN("Close stmt"),
+  STRING_WITH_LEN("Reset stmt"), STRING_WITH_LEN("Set option"),
+  STRING_WITH_LEN("Fetch"), STRING_WITH_LEN("Daemon"),
+  STRING_WITH_LEN("Error")  // Last command number
 };
 
 const char *xa_state_names[]={
@@ -322,7 +331,7 @@ int check_user(THD *thd, enum enum_server_command command,
   if (opt_secure_auth_local && passwd_len == SCRAMBLE_LENGTH_323)
   {
     net_printf_error(thd, ER_NOT_SUPPORTED_AUTH_MODE);
-    mysql_log.write(thd, COM_CONNECT, ER(ER_NOT_SUPPORTED_AUTH_MODE));
+    general_log_print(thd, COM_CONNECT, ER(ER_NOT_SUPPORTED_AUTH_MODE));
     DBUG_RETURN(-1);
   }
   if (passwd_len != 0 &&
@@ -356,9 +365,9 @@ int check_user(THD *thd, enum enum_server_command command,
       net_printf_error(thd, ER_SERVER_IS_IN_SECURE_AUTH_MODE,
                        thd->main_security_ctx.user,
                        thd->main_security_ctx.host_or_ip);
-      mysql_log.write(thd, COM_CONNECT, ER(ER_SERVER_IS_IN_SECURE_AUTH_MODE),
-                      thd->main_security_ctx.user,
-                      thd->main_security_ctx.host_or_ip);
+      general_log_print(thd, COM_CONNECT, ER(ER_SERVER_IS_IN_SECURE_AUTH_MODE),
+                        thd->main_security_ctx.user,
+                        thd->main_security_ctx.host_or_ip);
       DBUG_RETURN(-1);
     }
     /* We have to read very specific packet size */
@@ -406,14 +415,14 @@ int check_user(THD *thd, enum enum_server_command command,
       }
 
       /* Why logging is performed before all checks've passed? */
-      mysql_log.write(thd, command,
-                      (thd->main_security_ctx.priv_user ==
-                       thd->main_security_ctx.user ?
-                       (char*) "%s@%s on %s" :
-                       (char*) "%s@%s as anonymous on %s"),
-                      thd->main_security_ctx.user,
-                      thd->main_security_ctx.host_or_ip,
-                      db ? db : (char*) "");
+      general_log_print(thd, command,
+                        (thd->main_security_ctx.priv_user ==
+                         thd->main_security_ctx.user ?
+                         (char*) "%s@%s on %s" :
+                         (char*) "%s@%s as anonymous on %s"),
+                        thd->main_security_ctx.user,
+                        thd->main_security_ctx.host_or_ip,
+                        db ? db : (char*) "");
 
       /*
         This is the default access rights for the current database.  It's
@@ -460,17 +469,17 @@ int check_user(THD *thd, enum enum_server_command command,
   else if (res == 2) // client gave short hash, server has long hash
   {
     net_printf_error(thd, ER_NOT_SUPPORTED_AUTH_MODE);
-    mysql_log.write(thd,COM_CONNECT,ER(ER_NOT_SUPPORTED_AUTH_MODE));
+    general_log_print(thd, COM_CONNECT, ER(ER_NOT_SUPPORTED_AUTH_MODE));
     DBUG_RETURN(-1);
   }
   net_printf_error(thd, ER_ACCESS_DENIED_ERROR,
                    thd->main_security_ctx.user,
                    thd->main_security_ctx.host_or_ip,
                    passwd_len ? ER(ER_YES) : ER(ER_NO));
-  mysql_log.write(thd, COM_CONNECT, ER(ER_ACCESS_DENIED_ERROR),
-                  thd->main_security_ctx.user,
-                  thd->main_security_ctx.host_or_ip,
-                  passwd_len ? ER(ER_YES) : ER(ER_NO));
+  general_log_print(thd, COM_CONNECT, ER(ER_ACCESS_DENIED_ERROR),
+                    thd->main_security_ctx.user,
+                    thd->main_security_ctx.host_or_ip,
+                    passwd_len ? ER(ER_YES) : ER(ER_NO));
   DBUG_RETURN(-1);
 #endif /* NO_EMBEDDED_ACCESS_CHECKS */
 }
@@ -1570,7 +1579,7 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
 			packet, strlen(packet), thd->charset());
     if (!mysql_change_db(thd, tmp.str, FALSE))
     {
-      mysql_log.write(thd,command,"%s",thd->db);
+      general_log_print(thd, command, "%s",thd->db);
       send_ok(thd);
     }
     break;
@@ -1703,7 +1712,7 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
     if (alloc_query(thd, packet, packet_length))
       break;					// fatal error is set
     char *packet_end= thd->query + thd->query_length;
-    mysql_log.write(thd,command,"%s",thd->query);
+    general_log_print(thd, command, "%s", thd->query);
     DBUG_PRINT("query",("%-.4096s",thd->query));
 
     if (!(specialflag & SPECIAL_NO_PRIOR))
@@ -1812,7 +1821,7 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
     thd->query_length= strlen(packet);       // for simplicity: don't optimize
     if (!(thd->query=fields=thd->memdup(packet,thd->query_length+1)))
       break;
-    mysql_log.write(thd,command,"%s %s",table_list.table_name, fields);
+    general_log_print(thd, command, "%s %s", table_list.table_name, fields);
     if (lower_case_table_names)
       my_casedn_str(files_charset_info, table_list.table_name);
     remove_escape(table_list.table_name);	// This can't have wildcards
@@ -1841,7 +1850,7 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
 #endif
   case COM_QUIT:
     /* We don't calculate statistics for this command */
-    mysql_log.write(thd,command,NullS);
+    general_log_print(thd, command, NullS);
     net->error=0;				// Don't give 'abort' message
     error=TRUE;					// End server
     break;
@@ -1861,7 +1870,7 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
       }
       if (check_access(thd,CREATE_ACL,db,0,1,0,is_schema_db(db)))
 	break;
-      mysql_log.write(thd,command,packet);
+      general_log_print(thd, command, packet);
       bzero(&create_info, sizeof(create_info));
       mysql_create_db(thd, (lower_case_table_names == 2 ? alias : db),
                       &create_info, 0);
@@ -1886,7 +1895,7 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
                    ER(ER_LOCK_OR_ACTIVE_TRANSACTION), MYF(0));
 	break;
       }
-      mysql_log.write(thd,command,db);
+      general_log_print(thd, command, db);
       mysql_rm_db(thd, db, 0, 0);
       break;
     }
@@ -1910,7 +1919,7 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
 	kill_zombie_dump_threads(slave_server_id);
       thd->server_id = slave_server_id;
 
-      mysql_log.write(thd, command, "Log: '%s'  Pos: %ld", packet+10,
+      general_log_print(thd, command, "Log: '%s'  Pos: %ld", packet+10,
                       (long) pos);
       mysql_binlog_send(thd, thd->strdup(packet + 10), (my_off_t) pos, flags);
       unregister_slave(thd,1,1);
@@ -1928,7 +1937,7 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
     ulong options= (ulong) (uchar) packet[0];
     if (check_global_access(thd,RELOAD_ACL))
       break;
-    mysql_log.write(thd,command,NullS);
+    general_log_print(thd, command, NullS);
     if (!reload_acl_and_cache(thd, options, (TABLE_LIST*) 0, &not_used))
       send_ok(thd);
     break;
@@ -1956,7 +1965,7 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
       break;
     }
     DBUG_PRINT("quit",("Got shutdown command for level %u", level));
-    mysql_log.write(thd,command,NullS);
+    general_log_print(thd, command, NullS);
     send_eof(thd);
 #ifdef __WIN__
     sleep(1);					// must wait after eof()
@@ -1973,7 +1982,7 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
 #endif
   case COM_STATISTICS:
   {
-    mysql_log.write(thd,command,NullS);
+    general_log_print(thd, command, NullS);
     statistic_increment(thd->status_var.com_stat[SQLCOM_SHOW_STATUS],
 			&LOCK_status);
 #ifndef EMBEDDED_LIBRARY
@@ -2013,7 +2022,7 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
     if (!thd->security_ctx->priv_user[0] &&
         check_global_access(thd, PROCESS_ACL))
       break;
-    mysql_log.write(thd,command,NullS);
+    general_log_print(thd, command, NullS);
     mysqld_list_processes(thd,
 			  thd->security_ctx->master_access & PROCESS_ACL ? 
 			  NullS : thd->security_ctx->priv_user, 0);
@@ -2050,7 +2059,7 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
     if (check_global_access(thd, SUPER_ACL))
       break;					/* purecov: inspected */
     mysql_print_status();
-    mysql_log.write(thd,command,NullS);
+    general_log_print(thd, command, NullS);
     send_eof(thd);
     break;
   case COM_SLEEP:
@@ -2132,7 +2141,7 @@ void log_slow_statement(THD *thd)
 	 (specialflag & SPECIAL_LOG_QUERIES_NOT_USING_INDEXES)))
     {
       thd->status_var.long_query_count++;
-      mysql_slow_log.write(thd, thd->query, thd->query_length, start_of_query);
+      slow_log_print(thd, thd->query, thd->query_length, start_of_query);
     }
   }
 }
@@ -6541,7 +6550,8 @@ bool reload_acl_and_cache(THD *thd, ulong options, TABLE_LIST *tables,
   {
     /*
       Flush the normal query log, the update log, the binary log,
-      the slow query log, and the relay log (if it exists).
+      the slow query log, the relay log (if it exists) and the log
+      tables.
     */
 
     /*
@@ -6551,14 +6561,16 @@ bool reload_acl_and_cache(THD *thd, ulong options, TABLE_LIST *tables,
       than it would help them)
     */
     tmp_write_to_binlog= 0;
-    mysql_log.new_file(1);
-    mysql_slow_log.new_file(1);
     mysql_bin_log.rotate_and_purge(RP_FORCE_ROTATE);
 #ifdef HAVE_REPLICATION
     pthread_mutex_lock(&LOCK_active_mi);
     rotate_relay_log(active_mi);
     pthread_mutex_unlock(&LOCK_active_mi);
 #endif
+
+    /* flush slow and general logs */
+    logger.flush_logs(thd);
+
     if (ha_flush_logs(NULL))
       result=1;
     if (flush_error_log())
diff --git a/sql/sql_prepare.cc b/sql/sql_prepare.cc
index 7b35f057217..5b3aaf2fd51 100644
--- a/sql/sql_prepare.cc
+++ b/sql/sql_prepare.cc
@@ -1866,7 +1866,7 @@ void mysql_stmt_prepare(THD *thd, const char *packet, uint packet_length)
     thd->stmt_map.erase(stmt);
   }
   else
-    mysql_log.write(thd, COM_STMT_PREPARE, "[%lu] %s", stmt->id, packet);
+    general_log_print(thd, COM_STMT_PREPARE, "[%lu] %s", stmt->id, packet);
 
   /* check_prepared_statemnt sends the metadata packet in case of success */
   DBUG_VOID_RETURN;
@@ -2228,7 +2228,7 @@ void mysql_stmt_execute(THD *thd, char *packet_arg, uint packet_length)
   if (!(specialflag & SPECIAL_NO_PRIOR))
     my_pthread_setprio(pthread_self(), WAIT_PRIOR);
   if (error == 0)
-    mysql_log.write(thd, COM_STMT_EXECUTE, "[%lu] %s", stmt->id, thd->query);
+    general_log_print(thd, COM_STMT_EXECUTE, "[%lu] %s", stmt->id, thd->query);
 
   DBUG_VOID_RETURN;
 
@@ -2607,7 +2607,7 @@ void Prepared_statement::setup_set_params()
 {
   /* Setup binary logging */
   if (mysql_bin_log.is_open() && is_update_query(lex->sql_command) ||
-      mysql_log.is_open() || mysql_slow_log.is_open())
+      opt_log || opt_slow_log)
   {
     set_params_from_vars= insert_params_from_vars_with_log;
 #ifndef EMBEDDED_LIBRARY
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index 53b3c26af9f..672f7fe8abe 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -605,8 +605,8 @@ bool mysqld_show_create_db(THD *thd, char *dbname,
   {
     my_error(ER_DBACCESS_DENIED_ERROR, MYF(0),
              sctx->priv_user, sctx->host_or_ip, dbname);
-    mysql_log.write(thd,COM_INIT_DB,ER(ER_DBACCESS_DENIED_ERROR),
-		    sctx->priv_user, sctx->host_or_ip, dbname);
+    general_log_print(thd,COM_INIT_DB,ER(ER_DBACCESS_DENIED_ERROR),
+                      sctx->priv_user, sctx->host_or_ip, dbname);
     DBUG_RETURN(TRUE);
   }
 #endif
@@ -1502,7 +1502,7 @@ void mysqld_list_processes(THD *thd,const char *user, bool verbose)
     if (thd_info->proc_info)
       protocol->store(thd_info->proc_info, system_charset_info);
     else
-      protocol->store(command_name[thd_info->command], system_charset_info);
+      protocol->store(command_name[thd_info->command].str, system_charset_info);
     if (thd_info->start_time)
       protocol->store((uint32) (now - thd_info->start_time));
     else
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index d1ecef06fe0..5ea46ec666c 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -2928,7 +2928,8 @@ 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)
+    if (lock_type == TL_WRITE && table->table->s->version &&
+        !table->table->s->log_table)
     {
       pthread_mutex_lock(&LOCK_open);
       const char *old_message=thd->enter_cond(&COND_refresh, &LOCK_open,
@@ -3098,9 +3099,10 @@ 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)
+      else if (open_for_modify && !table->table->s->log_table)
       {
         pthread_mutex_lock(&LOCK_open);
         remove_table_from_cache(thd, table->table->s->db.str,
diff --git a/sql/table.cc b/sql/table.cc
index 352ca495a31..4640bfe9849 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -310,16 +310,29 @@ int open_table_def(THD *thd, TABLE_SHARE *share, uint db_flags)
     error= open_binary_frm(thd, share, head, file);
     *root_ptr= old_root;
 
-    /*
-      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.
-    */
     if (share->db.length == 5 &&
-        !my_strcasecmp(system_charset_info, share->db.str, "mysql") &&
-        (!my_strcasecmp(system_charset_info, share->table_name.str, "proc") ||
-         !my_strcasecmp(system_charset_info, share->table_name.str, "event")))
-      share->system_table= 1;
+        !my_strcasecmp(system_charset_info, 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.
+      */
+      if (!my_strcasecmp(system_charset_info, share->table_name.str, "proc")
+          || !my_strcasecmp(system_charset_info, share->table_name.str,
+                            "event"))
+        share->system_table= 1;
+      else
+      {
+        if (!my_strcasecmp(system_charset_info, share->table_name.str,
+                           "general_log"))
+          share->log_table= LOG_GENERAL;
+        else
+          if (!my_strcasecmp(system_charset_info, share->table_name.str,
+                             "slow_log"))
+            share->log_table= LOG_SLOW;
+      }
+    }
     error_given= 1;
   }
 
diff --git a/sql/table.h b/sql/table.h
index ac56a7840bf..eb0c0cf98d3 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -198,6 +198,11 @@ typedef struct st_table_share
     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
   const uchar *partition_info;
   uint  partition_info_len;
@@ -286,6 +291,7 @@ struct st_table {
   my_bool distinct,const_table,no_rows;
   my_bool key_read, no_keyread;
   my_bool locked_by_flush;
+  my_bool locked_by_logger;
   my_bool locked_by_name;
   my_bool fulltext_searched;
   my_bool no_cache;