Fix for bug #29928: INSERT ... VALUES(connection_id(), ...) incorrect

restores from mysqlbinlog out

Problem: using "mysqlbinlog | mysql" for recoveries the connection_id() 
result may differ from what was used when issuing the statement.

Fix: if there is a connection_id() in a statement, write to binlog
SET pseudo_thread_id= XXX; before it and use the value later on.


mysql-test/r/mysqlbinlog.result:
  Fix for bug #29928: INSERT ... VALUES(connection_id(), ...) incorrect 
  restores from mysqlbinlog out
    - test result.
mysql-test/t/mysqlbinlog.test:
  Fix for bug #29928: INSERT ... VALUES(connection_id(), ...) incorrect 
  restores from mysqlbinlog out
    - test case.
sql/item_create.cc:
  Fix for bug #29928: INSERT ... VALUES(connection_id(), ...) incorrect 
  restores from mysqlbinlog out
    - set thread_specific_used flag for the connection_id() function.
sql/item_func.cc:
  Fix for bug #29928: INSERT ... VALUES(connection_id(), ...) incorrect 
  restores from mysqlbinlog out
    - always return thd->variables.pseudo_thread_id as a connection_id() 
  result, as it contains a proper value for both master and slave.
sql/log_event.cc:
  Fix for bug #29928: INSERT ... VALUES(connection_id(), ...) incorrect 
  restores from mysqlbinlog out
    - set LOG_EVENT_THREAD_SPECIFIC_F event flag if thread_specific_used
  is set.
sql/sql_class.cc:
  Fix for bug #29928: INSERT ... VALUES(connection_id(), ...) incorrect 
  restores from mysqlbinlog out
    - thd->thread_specific_used introduced, which is set if thread specific 
  value(s) used in a statement.
sql/sql_class.h:
  Fix for bug #29928: INSERT ... VALUES(connection_id(), ...) incorrect 
  restores from mysqlbinlog out
    - thd->thread_specific_used introduced, which is set if thread specific 
  value(s) used in a statement.

mysql-test/r/mysqlbinlog.result

@@ -318,4 +318,11 @@ INSERT INTO t1 VALUES ('0123456789');
 flush logs;
 DROP TABLE t1;
 # 	Query	thread_id=REMOVED	exec_time=REMOVED	error_code=REMOVED
+flush logs;
+create table t1(a int);
+insert into t1 values(connection_id());
+flush logs;
+drop table t1;
+1
+drop table t1;
 End of 5.0 tests

mysql-test/t/mysqlbinlog.test

@@ -216,4 +216,25 @@ flush logs;
 DROP TABLE t1;
 --exec $MYSQL_BINLOG --hexdump --local-load=$MYSQLTEST_VARDIR/tmp/ $MYSQLTEST_VARDIR/log/master-bin.000011 | grep 'Query' | sed 's/[0-9]\{1,\}/REMOVED/g'
 
+#
+# Bug #29928: incorrect connection_id() restoring from mysqlbinlog out
+#
+flush logs;
+create table t1(a int);
+insert into t1 values(connection_id()); 
+let $a= `select a from t1`;
+flush logs;
+--exec $MYSQL_BINLOG $MYSQLTEST_VARDIR/log/master-bin.000013 > $MYSQLTEST_VARDIR/tmp/bug29928.sql
+drop table t1;
+connect (con1, localhost, root, , test);                                            
+connection con1;
+--exec $MYSQL test < $MYSQLTEST_VARDIR/tmp/bug29928.sql
+--remove_file $MYSQLTEST_VARDIR/tmp/bug29928.sql
+let $b= `select a from t1`;
+disconnect con1;
+connection default;
+let $c= `select $a=$b`;
+--echo $c
+drop table t1;
+
 --echo End of 5.0 tests

sql/item_create.cc

@@ -70,7 +70,9 @@ Item *create_func_ceiling(Item* a)
 
 Item *create_func_connection_id(void)
 {
-  current_thd->lex->safe_to_cache_query= 0;
+  THD *thd= current_thd;
+  thd->lex->safe_to_cache_query= 0;
+  thd->thread_specific_used= TRUE;
   return new Item_func_connection_id();
 }
 

sql/item_func.cc

@@ -649,16 +649,7 @@ bool Item_func_connection_id::fix_fields(THD *thd, Item **ref)
 {
   if (Item_int_func::fix_fields(thd, ref))
     return TRUE;
-
+  value= thd->variables.pseudo_thread_id;
-  /*
-    To replicate CONNECTION_ID() properly we should use
-    pseudo_thread_id on slave, which contains the value of thread_id
-    on master.
-  */
-  value= ((thd->slave_thread) ?
-          thd->variables.pseudo_thread_id :
-          thd->thread_id);
-
   return FALSE;
 }
 

sql/log_event.cc

@@ -1303,8 +1303,9 @@ Query_log_event::Query_log_event(THD* thd_arg, const char* query_arg,
 				 ulong query_length, bool using_trans,
 				 bool suppress_use, THD::killed_state killed_status_arg)
   :Log_event(thd_arg,
-	     ((thd_arg->tmp_table_used ? LOG_EVENT_THREAD_SPECIFIC_F : 0)
+             ((thd_arg->tmp_table_used || thd_arg->thread_specific_used) ? 
-	      | (suppress_use          ? LOG_EVENT_SUPPRESS_USE_F    : 0)),
+	        LOG_EVENT_THREAD_SPECIFIC_F : 0) |
+	      (suppress_use ? LOG_EVENT_SUPPRESS_USE_F : 0),
 	     using_trans),
    data_buf(0), query(query_arg), catalog(thd_arg->catalog),
    db(thd_arg->db), q_len((uint32) query_length),
@@ -2689,8 +2690,10 @@ Load_log_event::Load_log_event(THD *thd_arg, sql_exchange *ex,
 			       List<Item> &fields_arg,
 			       enum enum_duplicates handle_dup,
 			       bool ignore, bool using_trans)
-  :Log_event(thd_arg, !thd_arg->tmp_table_used ?
+  :Log_event(thd_arg,
-	     0 : LOG_EVENT_THREAD_SPECIFIC_F, using_trans),
+             (thd_arg->tmp_table_used || thd_arg->thread_specific_used) ?
+               LOG_EVENT_THREAD_SPECIFIC_F : 0,
+             using_trans),
    thread_id(thd_arg->thread_id),
    slave_proxy_id(thd_arg->variables.pseudo_thread_id),
    num_fields(0),fields(0),

sql/sql_class.cc

@@ -197,7 +197,7 @@ THD::THD()
   count_cuted_fields= CHECK_FIELD_IGNORE;
   killed= NOT_KILLED;
   db_length= col_access=0;
-  query_error= tmp_table_used= 0;
+  query_error= tmp_table_used= thread_specific_used= 0;
   next_insert_id=last_insert_id=0;
   hash_clear(&handler_tables_hash);
   tmp_table=0;

sql/sql_class.h

@@ -1457,6 +1457,9 @@ public:
   bool	     in_lock_tables;
   bool       query_error, bootstrap, cleanup_done;
   bool	     tmp_table_used;
+  
+  /**  is set if some thread specific value(s) used in a statement. */
+  bool       thread_specific_used;
   bool	     charset_is_system_charset, charset_is_collation_connection;
   bool       charset_is_character_set_filesystem;
   bool       enable_slow_log;   /* enable slow log for current statement */