Added Rand_log_event

sql/item_func.cc

@@ -743,17 +743,31 @@ double Item_func_round::val()
 
 double Item_func_rand::val()
 {
+  THD* thd = current_thd;
   if (arg_count)
   {					// Only use argument once in query
     uint32 tmp= (uint32) (args[0]->val_int());
-    randominit(&current_thd->rand,(uint32) (tmp*0x10001L+55555555L),
+    randominit(&thd->rand,(uint32) (tmp*0x10001L+55555555L),
 	       (uint32) (tmp*0x10000001L));
 #ifdef DELETE_ITEMS
     delete args[0];
 #endif
     arg_count=0;
   }
-  return rnd(&current_thd->rand);
+  else if (!thd->rand_used)
+  {
+    // no need to send a Rand log event if seed was given eg: RAND(seed),
+    // as it will be replicated in the query as such.
+
+    // save the seed only the first time RAND() is used in the query
+
+    // once events are forwarded rather than recreated,
+    // the following can be skipped if inside the slave thread
+    thd->rand_used=1;
+    thd->rand_saved_seed1=thd->rand.seed1;
+    thd->rand_saved_seed2=thd->rand.seed2;
+  }
+  return rnd(&thd->rand);
 }
 
 longlong Item_func_sign::val_int()

sql/log.cc

@@ -1072,6 +1072,13 @@ bool MYSQL_LOG::write(Log_event* event_info)
       if (e.write(file))
 	goto err;
     }
+    if (thd && thd->rand_used)
+    {
+      Rand_log_event e(thd,thd->rand_saved_seed1,thd->rand_saved_seed2);
+      e.set_log_pos(this);
+      if (e.write(file))
+	goto err;
+    }
     if (thd && thd->variables.convert_set)
     {
       char buf[1024] = "SET CHARACTER SET ";

sql/log_event.cc

@@ -348,6 +348,18 @@ void Intvar_log_event::pack_info(String* packet)
   net_store_data(packet, tmp.ptr(), tmp.length());
 }
 
+void Rand_log_event::pack_info(String* packet)
+{
+  char buf1[256], buf[22];
+  String tmp(buf1, sizeof(buf1), system_charset_info);
+  tmp.length(0);
+  tmp.append("randseed1=");
+  tmp.append(llstr(seed1, buf));
+  tmp.append(",randseed2=");
+  tmp.append(llstr(seed2, buf));
+  net_store_data(packet, tmp.ptr(), tmp.length());
+}
+
 void Slave_log_event::pack_info(String* packet)
 {
   char buf1[256], buf[22], *end;
@@ -376,6 +388,9 @@ void Log_event::init_show_field_list(List<Item>* field_list)
   field_list->push_back(new Item_empty_string("Info", 20));
 }
 
+/*
+ * only called by SHOW BINLOG EVENTS
+ */
 int Log_event::net_send(THD* thd, const char* log_name, my_off_t pos)
 {
   String* packet = &thd->packet;
@@ -610,6 +625,9 @@ Log_event* Log_event::read_log_event(const char* buf, int event_len,
   case INTVAR_EVENT:
     ev = new Intvar_log_event(buf, old_format);
     break;
+  case RAND_EVENT:
+    ev = new Rand_log_event(buf, old_format);
+    break;
   default:
     break;
   }
@@ -915,6 +933,41 @@ void Intvar_log_event::print(FILE* file, bool short_form, char* last_db)
 }
 #endif
 
+/*****************************************************************************
+ *
+ *  Rand log event
+ *
+ ****************************************************************************/
+Rand_log_event::Rand_log_event(const char* buf, bool old_format)
+  :Log_event(buf, old_format)
+{
+  buf += (old_format) ? OLD_HEADER_LEN : LOG_EVENT_HEADER_LEN;
+  seed1 = uint8korr(buf+RAND_SEED1_OFFSET);
+  seed2 = uint8korr(buf+RAND_SEED2_OFFSET);
+}
+
+int Rand_log_event::write_data(IO_CACHE* file)
+{
+  char buf[16];
+  int8store(buf + RAND_SEED1_OFFSET, seed1);
+  int8store(buf + RAND_SEED2_OFFSET, seed2);
+  return my_b_safe_write(file, (byte*) buf, sizeof(buf));
+}
+
+#ifdef MYSQL_CLIENT
+void Rand_log_event::print(FILE* file, bool short_form, char* last_db)
+{
+  char llbuff[22];
+  if (!short_form)
+  {
+    print_header(file);
+    fprintf(file, "\tRand\n");
+  }
+  fprintf(file, "SET RAND SEED1=%s;\n", llstr(seed1, llbuff));
+  fprintf(file, "SET RAND SEED2=%s;\n", llstr(seed2, llbuff));
+  fflush(file);
+}
+#endif
 
 int Load_log_event::write_data_header(IO_CACHE* file)
 {
@@ -1926,6 +1979,14 @@ int Intvar_log_event::exec_event(struct st_relay_log_info* rli)
   return 0;
 }
 
+int Rand_log_event::exec_event(struct st_relay_log_info* rli)
+{
+  thd->rand.seed1 = seed1;
+  thd->rand.seed2 = seed2;
+  rli->inc_pending(get_event_len());
+  return 0;
+}
+
 int Slave_log_event::exec_event(struct st_relay_log_info* rli)
 {
   if (mysql_bin_log.is_open())

sql/log_event.h

@@ -152,6 +152,11 @@ struct sql_ex_info
 #define I_TYPE_OFFSET        0
 #define I_VAL_OFFSET         1
 
+/* Rand event post-header */
+
+#define RAND_SEED1_OFFSET 0
+#define RAND_SEED2_OFFSET 8
+
 /* Load event post-header */
 
 #define L_THREAD_ID_OFFSET   0
@@ -199,7 +204,7 @@ enum Log_event_type
   START_EVENT = 1, QUERY_EVENT =2, STOP_EVENT=3, ROTATE_EVENT = 4,
   INTVAR_EVENT=5, LOAD_EVENT=6, SLAVE_EVENT=7, CREATE_FILE_EVENT=8,
   APPEND_BLOCK_EVENT=9, EXEC_LOAD_EVENT=10, DELETE_FILE_EVENT=11,
-  NEW_LOAD_EVENT=12
+  NEW_LOAD_EVENT=12, RAND_EVENT=13
 };
 
 enum Int_event_type
@@ -497,6 +502,34 @@ public:
   bool is_valid() { return 1; }
 };
 
+/*****************************************************************************
+ *
+ *  Rand log event class
+ *
+ ****************************************************************************/
+class Rand_log_event: public Log_event
+{
+ public:
+  ulonglong seed1;
+  ulonglong seed2;
+
+#ifndef MYSQL_CLIENT
+  Rand_log_event(THD* thd_arg, ulonglong seed1_arg, ulonglong seed2_arg)
+    :Log_event(thd_arg),seed1(seed1_arg),seed2(seed2_arg)
+    {}
+  void pack_info(String* packet);
+  int exec_event(struct st_relay_log_info* rli);
+#else
+  void print(FILE* file, bool short_form = 0, char* last_db = 0);
+#endif
+
+  Rand_log_event(const char* buf, bool old_format);
+  ~Rand_log_event() {}
+  Log_event_type get_type_code() { return RAND_EVENT;}
+  int get_data_size() { return sizeof(ulonglong) * 2; }
+  int write_data(IO_CACHE* file);
+  bool is_valid() { return 1; }
+};
 
 class Stop_log_event: public Log_event
 {

sql/sql_class.cc

@@ -81,7 +81,7 @@ static void free_var(user_var_entry *entry)
 
 THD::THD():user_time(0), fatal_error(0),
 	   last_insert_id_used(0),
-	   insert_id_used(0), in_lock_tables(0),
+	   insert_id_used(0), rand_used(0), in_lock_tables(0),
 	   global_read_lock(0), bootstrap(0)
 {
   host=user=priv_user=db=query=ip=0;

sql/sql_class.h

@@ -501,7 +501,8 @@ public:
   bool	     set_query_id,locked,count_cuted_fields,some_tables_deleted;
   bool	     no_errors, allow_sum_func, password;
   bool	     fatal_error;
-  bool	     query_start_used,last_insert_id_used,insert_id_used;
+  bool	     query_start_used,last_insert_id_used,insert_id_used,rand_used;
+  ulonglong  rand_saved_seed1, rand_saved_seed2;
   bool	     system_thread,in_lock_tables,global_read_lock;
   bool       query_error, bootstrap, cleanup_done;
   bool	     safe_to_cache_query;

sql/sql_parse.cc

@@ -2912,6 +2912,7 @@ mysql_init_query(THD *thd)
   thd->total_warn_count=0;			// Warnings for this query
   thd->last_insert_id_used= thd->query_start_used= thd->insert_id_used=0;
   thd->sent_row_count= thd->examined_row_count= 0;
+  thd->rand_used=0;
   thd->safe_to_cache_query= 1;
   thd->lex.param_list.empty();
   DBUG_VOID_RETURN;