Bug#34306: Can't make copy of log tables when server binary log is enabled

The problem is that when statement-based replication was enabled,
statements such as INSERT INTO .. SELECT FROM .. and CREATE TABLE
.. SELECT FROM need to grab a read lock on the source table that
does not permit concurrent inserts, which would in turn be denied
if the source table is a log table because log tables can't be
locked exclusively.

The solution is to not take such a lock when the source table is
a log table as it is unsafe to replicate log tables under statement
based replication. Furthermore, the read lock that does not permits
concurrent inserts is now only taken if statement-based replication
is enabled and if the source table is not a log table.

include/thr_lock.h:
  Introduce yet another lock type that my get upgraded depending
  on the binary log format. This is not a optimal solution but
  can be easily improved later.
mysql-test/r/log_tables.result:
  Add test case result for Bug#34306
mysql-test/suite/binlog/r/binlog_stm_row.result:
  Add test case result for Bug#34306
mysql-test/suite/binlog/t/binlog_stm_row.test:
  Add test case for Bug#34306
mysql-test/t/log_tables.test:
  Add test case for Bug#34306
sql/lock.cc:
  Assert that TL_READ_DEFAULT is not a real lock type.
sql/mysql_priv.h:
  Export new function.
sql/mysqld.cc:
  Remove using_update_log.
sql/sql_base.cc:
  Introduce function that returns the appropriate read lock type
  depending on how the statement is going to be replicated. It will
  only take a TL_READ_NO_INSERT log if the binary is enabled and the
  binary log format is statement-based and the table is not a log table.
sql/sql_parse.cc:
  Remove using_update_log.
sql/sql_update.cc:
  Use new function to choose read lock type.
sql/sql_yacc.yy:
  The lock type is now decided at open_tables time. This old behavior was
  actually misleading as the binary log format can be dynamically switched
  and this would not change for statements that have already been parsed
  when the binary log format is changed (ie: prepared statements).

include/thr_lock.h

@@ -29,6 +29,14 @@ extern ulong locks_immediate,locks_waited ;
 
 enum thr_lock_type { TL_IGNORE=-1,
 		     TL_UNLOCK,			/* UNLOCK ANY LOCK */
+                     /*
+                       Parser only! At open_tables() becomes TL_READ or
+                       TL_READ_NO_INSERT depending on the binary log format
+                       (SBR/RBR) and on the table category (log table).
+                       Used for tables that are read by statements which
+                       modify tables.
+                     */
+                     TL_READ_DEFAULT,
 		     TL_READ,			/* Read lock */
 		     TL_READ_WITH_SHARED_LOCKS,
 		     /* High prior. than TL_WRITE. Allow concurrent insert */

mysql-test/r/log_tables.result

@@ -832,6 +832,35 @@ Execute	select '000 001 002 003 004 005 006 007 008 009010 011 012 013 014 015 0
 Query	set global general_log = off
 deallocate prepare long_query;
 set global general_log = @old_general_log_state;
+DROP TABLE IF EXISTS log_count;
+DROP TABLE IF EXISTS slow_log_copy;
+DROP TABLE IF EXISTS general_log_copy;
+CREATE TABLE log_count (count BIGINT(21));
+SET @old_general_log_state = @@global.general_log;
+SET @old_slow_log_state = @@global.slow_query_log;
+SET GLOBAL general_log = ON;
+SET GLOBAL slow_query_log = ON;
+CREATE TABLE slow_log_copy SELECT * FROM mysql.slow_log;
+INSERT INTO slow_log_copy SELECT * FROM mysql.slow_log;
+INSERT INTO log_count (count) VALUES ((SELECT count(*) FROM mysql.slow_log));
+DROP TABLE slow_log_copy;
+CREATE TABLE general_log_copy SELECT * FROM mysql.general_log;
+INSERT INTO general_log_copy SELECT * FROM mysql.general_log;
+INSERT INTO log_count (count) VALUES ((SELECT count(*) FROM mysql.general_log));
+DROP TABLE general_log_copy;
+SET GLOBAL general_log = OFF;
+SET GLOBAL slow_query_log = OFF;
+CREATE TABLE slow_log_copy SELECT * FROM mysql.slow_log;
+INSERT INTO slow_log_copy SELECT * FROM mysql.slow_log;
+INSERT INTO log_count (count) VALUES ((SELECT count(*) FROM mysql.slow_log));
+DROP TABLE slow_log_copy;
+CREATE TABLE general_log_copy SELECT * FROM mysql.general_log;
+INSERT INTO general_log_copy SELECT * FROM mysql.general_log;
+INSERT INTO log_count (count) VALUES ((SELECT count(*) FROM mysql.general_log));
+DROP TABLE general_log_copy;
+SET GLOBAL general_log = @old_general_log_state;
+SET GLOBAL slow_query_log = @old_slow_log_state;
+DROP TABLE log_count;
 SET @old_slow_log_state = @@global.slow_query_log;
 SET SESSION long_query_time = 0;
 SET GLOBAL slow_query_log = ON;

mysql-test/suite/binlog/r/binlog_stm_row.result

+DROP TABLE IF EXISTS t1;
+DROP TABLE IF EXISTS t2;
+SET GLOBAL BINLOG_FORMAT = STATEMENT;
+SET SESSION BINLOG_FORMAT = STATEMENT;
+CREATE TABLE t1 (a INT);
+CREATE TABLE t2 LIKE t1;
+select @@SESSION.BINLOG_FORMAT;
+@@SESSION.BINLOG_FORMAT
+STATEMENT
+INSERT INTO t1 VALUES(1);
+INSERT INTO t2 VALUES(2);
+#
+# Ensure that INSERT INTO .. SELECT FROM under SBR takes a read
+# lock that will prevent the source table from being modified.
+#
+# con1
+SELECT GET_LOCK('Bug#34306', 120);
+GET_LOCK('Bug#34306', 120)
+1
+# con2
+PREPARE stmt FROM "INSERT INTO t1 SELECT * FROM t2 WHERE GET_LOCK('Bug#34306', 120)";
+EXECUTE stmt;;
+# default
+INSERT INTO t2 VALUES (3);;
+# con1
+SELECT RELEASE_LOCK('Bug#34306');
+RELEASE_LOCK('Bug#34306')
+1
+# con2
+SELECT RELEASE_LOCK('Bug#34306');
+RELEASE_LOCK('Bug#34306')
+1
+# default
+#
+# Ensure that INSERT INTO .. SELECT FROM prepared under SBR does
+# not prevent the source table from being modified if under RBR.
+#
+# con2
+SET SESSION BINLOG_FORMAT = ROW;
+# con1
+SELECT GET_LOCK('Bug#34306', 120);
+GET_LOCK('Bug#34306', 120)
+1
+# con2
+EXECUTE stmt;;
+# default
+# con1
+INSERT INTO t2 VALUES (4);
+SELECT RELEASE_LOCK('Bug#34306');
+RELEASE_LOCK('Bug#34306')
+1
+# con2
+# default
+# Show binlog events
+show binlog events from <binlog_start>;
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000001	#	Query	#	#	use `test`; DROP TABLE IF EXISTS t1
+master-bin.000001	#	Query	#	#	use `test`; DROP TABLE IF EXISTS t2
+master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t1 (a INT)
+master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t2 LIKE t1
+master-bin.000001	#	Query	#	#	use `test`; INSERT INTO t1 VALUES(1)
+master-bin.000001	#	Query	#	#	use `test`; INSERT INTO t2 VALUES(2)
+master-bin.000001	#	Query	#	#	use `test`; INSERT INTO t1 SELECT * FROM t2 WHERE GET_LOCK('Bug#34306', 120)
+master-bin.000001	#	Query	#	#	use `test`; INSERT INTO t2 VALUES (3)
+master-bin.000001	#	Query	#	#	use `test`; INSERT INTO t2 VALUES (4)
+master-bin.000001	#	Query	#	#	use `test`; BEGIN
+master-bin.000001	#	Table_map	#	#	table_id: # (test.t1)
+master-bin.000001	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Query	#	#	use `test`; COMMIT
+DROP TABLE t1;
+DROP TABLE t2;

mysql-test/suite/binlog/t/binlog_stm_row.test

+--source include/have_log_bin.inc
+--source include/have_binlog_format_row_or_statement.inc
+
+# Get rid of previous tests binlog
+--disable_query_log
+reset master;
+--enable_query_log
+
+#
+# Bug#34306: Can't make copy of log tables when server binary log is enabled
+#
+# This is an additional test for Bug#34306 in order to ensure that INSERT INTO
+# .. SELECT FROM is properly replicated under SBR and RBR and that the proper
+# read lock type are acquired.
+#
+
+--disable_warnings
+DROP TABLE IF EXISTS t1;
+DROP TABLE IF EXISTS t2;
+--enable_warnings
+
+SET GLOBAL BINLOG_FORMAT = STATEMENT;
+SET SESSION BINLOG_FORMAT = STATEMENT;
+
+CREATE TABLE t1 (a INT);
+CREATE TABLE t2 LIKE t1;
+select @@SESSION.BINLOG_FORMAT;
+INSERT INTO t1 VALUES(1);
+INSERT INTO t2 VALUES(2);
+
+--connect(con1,localhost,root,,)
+--connect(con2,localhost,root,,)
+
+--echo #
+--echo # Ensure that INSERT INTO .. SELECT FROM under SBR takes a read
+--echo # lock that will prevent the source table from being modified.
+--echo #
+
+--connection con1
+--echo # con1
+SELECT GET_LOCK('Bug#34306', 120);
+--connection con2
+--echo # con2
+PREPARE stmt FROM "INSERT INTO t1 SELECT * FROM t2 WHERE GET_LOCK('Bug#34306', 120)";
+--send EXECUTE stmt;
+--connection default
+--echo # default
+let $wait_condition=
+  SELECT COUNT(*) = 1 FROM information_schema.processlist WHERE
+  state = "User lock" AND
+  info = "INSERT INTO t1 SELECT * FROM t2 WHERE GET_LOCK('Bug#34306', 120)";
+--source include/wait_condition.inc
+--send INSERT INTO t2 VALUES (3);
+--connection con1
+--echo # con1
+let $wait_condition=
+  SELECT COUNT(*) = 1 FROM information_schema.processlist WHERE
+  state = "Locked" and info = "INSERT INTO t2 VALUES (3)";
+--source include/wait_condition.inc
+SELECT RELEASE_LOCK('Bug#34306');
+--connection con2
+--echo # con2
+--reap
+SELECT RELEASE_LOCK('Bug#34306');
+--connection default
+--echo # default
+--reap
+
+--echo #
+--echo # Ensure that INSERT INTO .. SELECT FROM prepared under SBR does
+--echo # not prevent the source table from being modified if under RBR.
+--echo #
+
+--connection con2
+--echo # con2
+SET SESSION BINLOG_FORMAT = ROW;
+--connection con1
+--echo # con1
+SELECT GET_LOCK('Bug#34306', 120);
+--connection con2
+--echo # con2
+--send EXECUTE stmt;
+--connection default
+--echo # default
+let $wait_condition=
+  SELECT COUNT(*) = 1 FROM information_schema.processlist WHERE
+  state = "User lock" AND
+  info = "INSERT INTO t1 SELECT * FROM t2 WHERE GET_LOCK('Bug#34306', 120)";
+--source include/wait_condition.inc
+--connection con1
+--echo # con1
+INSERT INTO t2 VALUES (4);
+SELECT RELEASE_LOCK('Bug#34306');
+--connection con2
+--echo # con2
+--reap
+
+--disconnect con1
+--disconnect con2
+--connection default
+--echo # default
+
+--echo # Show binlog events
+source include/show_binlog_events.inc;
+
+DROP TABLE t1;
+DROP TABLE t2;

mysql-test/t/log_tables.test

@@ -936,6 +936,52 @@ select command_type, argument from mysql.general_log where thread_id = @thread_i
 deallocate prepare long_query;
 set global general_log = @old_general_log_state;
 
+#
+# Bug#34306: Can't make copy of log tables when server binary log is enabled
+#
+
+--disable_warnings
+DROP TABLE IF EXISTS log_count;
+DROP TABLE IF EXISTS slow_log_copy;
+DROP TABLE IF EXISTS general_log_copy;
+--enable_warnings
+
+CREATE TABLE log_count (count BIGINT(21));
+
+SET @old_general_log_state = @@global.general_log;
+SET @old_slow_log_state = @@global.slow_query_log;
+
+SET GLOBAL general_log = ON;
+SET GLOBAL slow_query_log = ON;
+
+CREATE TABLE slow_log_copy SELECT * FROM mysql.slow_log;
+INSERT INTO slow_log_copy SELECT * FROM mysql.slow_log;
+INSERT INTO log_count (count) VALUES ((SELECT count(*) FROM mysql.slow_log));
+DROP TABLE slow_log_copy;
+
+CREATE TABLE general_log_copy SELECT * FROM mysql.general_log;
+INSERT INTO general_log_copy SELECT * FROM mysql.general_log;
+INSERT INTO log_count (count) VALUES ((SELECT count(*) FROM mysql.general_log));
+DROP TABLE general_log_copy;
+
+SET GLOBAL general_log = OFF;
+SET GLOBAL slow_query_log = OFF;
+
+CREATE TABLE slow_log_copy SELECT * FROM mysql.slow_log;
+INSERT INTO slow_log_copy SELECT * FROM mysql.slow_log;
+INSERT INTO log_count (count) VALUES ((SELECT count(*) FROM mysql.slow_log));
+DROP TABLE slow_log_copy;
+
+CREATE TABLE general_log_copy SELECT * FROM mysql.general_log;
+INSERT INTO general_log_copy SELECT * FROM mysql.general_log;
+INSERT INTO log_count (count) VALUES ((SELECT count(*) FROM mysql.general_log));
+DROP TABLE general_log_copy;
+
+SET GLOBAL general_log = @old_general_log_state;
+SET GLOBAL slow_query_log = @old_slow_log_state;
+
+DROP TABLE log_count;
+
 #
 # Bug #31700: thd->examined_row_count not incremented for 'const' type queries
 #

sql/lock.cc

@@ -854,7 +854,7 @@ static MYSQL_LOCK *get_lock_data(THD *thd, TABLE **table_ptr, uint count,
     if ((table=table_ptr[i])->s->tmp_table == NON_TRANSACTIONAL_TMP_TABLE)
       continue;
     lock_type= table->reginfo.lock_type;
-    DBUG_ASSERT (lock_type != TL_WRITE_DEFAULT);
+    DBUG_ASSERT(lock_type != TL_WRITE_DEFAULT && lock_type != TL_READ_DEFAULT);
     if (lock_type >= TL_WRITE_ALLOW_WRITE)
     {
       *write_lock_used=table;

sql/mysql_priv.h

@@ -1265,6 +1265,7 @@ bool fix_merge_after_open(TABLE_LIST *old_child_list, TABLE_LIST **old_last,
                           TABLE_LIST *new_child_list, TABLE_LIST **new_last);
 bool reopen_table(TABLE *table);
 bool reopen_tables(THD *thd,bool get_locks,bool in_refresh);
+thr_lock_type read_lock_type_for_table(THD *thd, TABLE *table);
 void close_data_files_and_morph_locks(THD *thd, const char *db,
                                       const char *table_name);
 void close_handle_and_leave_table_as_lock(TABLE *table);
@@ -1938,7 +1939,7 @@ extern bool opt_using_transactions;
 extern bool mysqld_embedded;
 #endif /* MYSQL_SERVER || INNODB_COMPATIBILITY_HOOKS */
 #ifdef MYSQL_SERVER
-extern bool using_update_log, opt_large_files, server_id_supplied;
+extern bool opt_large_files, server_id_supplied;
 extern bool opt_update_log, opt_bin_log, opt_error_log;
 extern my_bool opt_log, opt_slow_log;
 extern ulong log_output_options;

sql/mysqld.cc

@@ -382,7 +382,7 @@ my_bool opt_character_set_client_handshake= 1;
 bool server_id_supplied = 0;
 bool opt_endinfo, using_udf_functions;
 my_bool locked_in_memory;
-bool opt_using_transactions, using_update_log;
+bool opt_using_transactions;
 bool volatile abort_loop;
 bool volatile shutdown_in_progress;
 /**
@@ -3815,12 +3815,6 @@ server.");
     {
       unireg_abort(1);
     }
-
-    /*
-      Used to specify which type of lock we need to use for queries of type
-      INSERT ... SELECT. This will change when we have row level logging.
-    */
-    using_update_log=1;
   }
 
   /* call ha_init_key_cache() on all key caches to init them */
@@ -7431,7 +7425,7 @@ static void mysql_init_variables(void)
   slave_open_temp_tables= 0;
   cached_thread_count= 0;
   opt_endinfo= using_udf_functions= 0;
-  opt_using_transactions= using_update_log= 0;
+  opt_using_transactions= 0;
   abort_loop= select_thread_in_use= signal_thread_in_use= 0;
   ready_to_exit= shutdown_in_progress= grant_option= 0;
   aborted_threads= aborted_connects= 0;

sql/sql_base.cc

@@ -4355,6 +4355,38 @@ bool fix_merge_after_open(TABLE_LIST *old_child_list, TABLE_LIST **old_last,
 }
 
 
+/*
+  Return a appropriate read lock type given a table object.
+
+  @param thd Thread context
+  @param table TABLE object for table to be locked
+
+  @remark Due to a statement-based replication limitation, statements such as
+          INSERT INTO .. SELECT FROM .. and CREATE TABLE .. SELECT FROM need
+          to grab a TL_READ_NO_INSERT lock on the source table in order to
+          prevent the replication of a concurrent statement that modifies the
+          source table. If such a statement gets applied on the slave before
+          the INSERT .. SELECT statement finishes, data on the master could
+          differ from data on the slave and end-up with a discrepancy between
+          the binary log and table state. Furthermore, this does not apply to
+          I_S and log tables as it's always unsafe to replicate such tables
+          under statement-based replication as the table on the slave might
+          contain other data (ie: general_log is enabled on the slave). The
+          statement will be marked as unsafe for SBR in decide_logging_format().
+*/
+
+thr_lock_type read_lock_type_for_table(THD *thd, TABLE *table)
+{
+  bool log_on= mysql_bin_log.is_open() && (thd->options & OPTION_BIN_LOG);
+  ulong binlog_format= thd->variables.binlog_format;
+  if ((log_on == FALSE) || (binlog_format == BINLOG_FORMAT_ROW) ||
+      (table->s->table_category == TABLE_CATEGORY_PERFORMANCE))
+    return TL_READ;
+  else
+    return TL_READ_NO_INSERT;
+}
+
+
 /*
   Open all tables in list
 
@@ -4629,6 +4661,9 @@ int open_tables(THD *thd, TABLE_LIST **start, uint *counter, uint flags)
     {
       if (tables->lock_type == TL_WRITE_DEFAULT)
         tables->table->reginfo.lock_type= thd->update_lock_default;
+      else if (tables->lock_type == TL_READ_DEFAULT)
+        tables->table->reginfo.lock_type=
+          read_lock_type_for_table(thd, tables->table);
       else if (tables->table->s->tmp_table == NO_TMP_TABLE)
         tables->table->reginfo.lock_type= tables->lock_type;
     }
@@ -5036,7 +5071,11 @@ int decide_logging_format(THD *thd, TABLE_LIST *tables)
     void* prev_ht= NULL;
     for (TABLE_LIST *table= tables; table; table= table->next_global)
     {
-      if (!table->placeholder() && table->lock_type >= TL_WRITE_ALLOW_WRITE)
+      if (table->placeholder())
+        continue;
+      if (table->table->s->table_category == TABLE_CATEGORY_PERFORMANCE)
+        thd->lex->set_stmt_unsafe();
+      if (table->lock_type >= TL_WRITE_ALLOW_WRITE)
       {
         ulonglong const flags= table->table->file->ha_table_flags();
         DBUG_PRINT("info", ("table: %s; ha_table_flags: %s%s",

sql/sql_parse.cc

@@ -5628,7 +5628,7 @@ void mysql_init_multi_delete(LEX *lex)
   lex->select_lex.select_limit= 0;
   lex->unit.select_limit_cnt= HA_POS_ERROR;
   lex->select_lex.table_list.save_and_clear(&lex->auxiliary_table_list);
-  lex->lock_option= using_update_log ? TL_READ_NO_INSERT : TL_READ;
+  lex->lock_option= TL_READ_DEFAULT;
   lex->query_tables= 0;
   lex->query_tables_last= &lex->query_tables;
 }

sql/sql_update.cc

@@ -1039,7 +1039,7 @@ reopen_tables:
         correct order of statements. Otherwise, we use a TL_READ lock to
         improve performance.
       */
-      tl->lock_type= using_update_log ? TL_READ_NO_INSERT : TL_READ;
+      tl->lock_type= read_lock_type_for_table(thd, table);
       tl->updating= 0;
       /* Update TABLE::lock_type accordingly. */
       if (!tl->placeholder() && !using_lock_tables)

sql/sql_yacc.yy

@@ -4301,7 +4301,7 @@ create_select:
           SELECT_SYM
           {
             LEX *lex=Lex;
-            lex->lock_option= using_update_log ? TL_READ_NO_INSERT : TL_READ;
+            lex->lock_option= TL_READ_DEFAULT;
             if (lex->sql_command == SQLCOM_INSERT)
               lex->sql_command= SQLCOM_INSERT_SELECT;
             else if (lex->sql_command == SQLCOM_REPLACE)
@@ -9398,7 +9398,7 @@ insert:
             lex->duplicates= DUP_ERROR; 
             mysql_init_select(lex);
             /* for subselects */
-            lex->lock_option= (using_update_log) ? TL_READ_NO_INSERT : TL_READ;
+            lex->lock_option= TL_READ_DEFAULT;
           }
           insert_lock_option
           opt_ignore insert2