Fix to support update + bianry logs with prepared statements (Dynamic query)

sql/item.cc

@@ -542,6 +542,79 @@ String *Item_param::val_str(String* str)
     return (String*) &str_value;
   }
 }
+
+/*
+  Return Param item values in string format, for generating the dynamic 
+  query used in update/binary logs
+*/
+
+String *Item_param::query_val_str(String* str) 
+{ 
+  switch (item_result_type) {
+  case INT_RESULT:
+    str->set(int_value, default_charset());
+    break;
+  case REAL_RESULT:
+    set->set(real_value, 2, default_charset());
+    break;
+  default:
+    str->set("'", 1, default_charset());
+    
+    if (!item_is_time)
+    {
+      str->append(str_value);
+      const char *from= str->ptr(); 
+      uint32 length= 1;
+      
+      // Escape misc cases
+      char *to= (char *)from, *end= (char *)to+str->length(); 
+      for (to++; to != end ; length++, to++)
+      {
+        switch(*to) {
+          case '\'':
+          case '"':  
+          case '\r':
+          case '\n':
+          case '\\': // TODO: Add remaining ..
+            str->replace(length,0,"\\",1); 
+            to++; end++; length++;
+            break;
+          default:     
+            break;
+        }
+      }
+    }
+    else
+    {
+      char buff[25];
+      
+      switch (ltime.time_type)  {
+        case TIMESTAMP_NONE:
+          break;
+        case TIMESTAMP_DATE:
+          sprintf(buff, "%04d-%02d-%02d", 
+                        ltime.year,ltime.month,ltime.day);
+          str->append(buff, 10);
+          break;
+        case TIMESTAMP_FULL:
+          sprintf(buff, "%04d-%02d-%02d %02d:%02d:%02d",
+ 	                ltime.year,ltime.month,ltime.day,
+	                ltime.hour,ltime.minute,ltime.second));
+          str->append(buff, 19);
+          break;
+        case TIMESTAMP_TIME:
+        {
+          sprintf(buff, "%02d:%02d:%02d",
+	  	        ltime.hour,ltime.minute,ltime.second));
+          str->append(buff, 8);
+          break;
+        }
+      }
+    }
+    str->append("'");
+  }
+  return str;
+}
 /* End of Item_param related */
 
 

sql/item.h

@@ -239,15 +239,17 @@ class Item_param :public Item
   enum Type item_type;
   enum enum_field_types buffer_type;
   bool item_is_time;
-  my_bool long_data_supplied;
+  bool long_data_supplied;
+  uint pos_in_query;
 
-  Item_param(char *name_par=0)
+  Item_param::Item_param(uint position)
   { 
-    name= name_par ? name_par : (char*) "?";
-    long_data_supplied= false;
+    name= (char*) "?";
+    pos_in_query= position;
     item_type= STRING_ITEM;
     item_result_type = STRING_RESULT;
     item_is_time= false;
+    long_data_supplied= false;
   }
   enum Type type() const { return item_type; }
   double val();
@@ -268,8 +270,9 @@ class Item_param :public Item
   void (*setup_param_func)(Item_param *param, uchar **pos);
   enum Item_result result_type () const
   { return item_result_type; }
+  String *query_val_str(String *str);
   enum_field_types field_type() const { return MYSQL_TYPE_STRING; }
-  Item *new_item() { return new Item_param(name); }
+  Item *new_item() { return new Item_param(pos_in_query); }
 };
 
 class Item_int :public Item

sql/sql_class.h

@@ -324,11 +324,14 @@ typedef struct st_prep_stmt
   Item_param **param;
   Item *free_list;
   MEM_ROOT mem_root;
+  String *query;
   ulong stmt_id;
   uint param_count;
   uint last_errno;
   char last_error[MYSQL_ERRMSG_SIZE];
   bool error_in_prepare, long_data_used;
+  bool log_full_query;
+  bool (*setup_params)(st_prep_stmt *stmt, uchar *pos, uchar *read_pos);
 } PREP_STMT;
 
 

sql/sql_prepare.cc

@@ -74,7 +74,10 @@ Long data handling:
 
 #define IS_PARAM_NULL(pos, param_no) pos[param_no/8] & (1 << param_no & 7)
 
+#define STMT_QUERY_LOG_LENGTH 8192
+
 extern int yyparse(void *thd);
+static String null_string("NULL", 4, default_charset_info);
 
 /*
   Find prepared statement in thd
@@ -129,6 +132,8 @@ int compare_prep_stmt(void *not_used, PREP_STMT *stmt, ulong *key)
 void free_prep_stmt(PREP_STMT *stmt, TREE_FREE mode, void *not_used)
 {     
   my_free((char *)stmt->param, MYF(MY_ALLOW_ZERO_PTR));
+  if (stmt->query)
+    stmt->query->free();
   free_items(stmt->free_list);
   free_root(&stmt->mem_root, MYF(0));
 }
@@ -374,8 +379,8 @@ static void setup_param_functions(Item_param *param, uchar param_type)
     param->setup_param_func= setup_param_date;
     param->item_result_type= STRING_RESULT;
     break;
-  case FIELD_TYPE_DATETIME:
-  case FIELD_TYPE_TIMESTAMP:
+  case MYSQL_TYPE_DATETIME:
+  case MYSQL_TYPE_TIMESTAMP:
     param->setup_param_func= setup_param_datetime;
     param->item_result_type= STRING_RESULT;
     break;
@@ -386,10 +391,82 @@ static void setup_param_functions(Item_param *param, uchar param_type)
 }
 
 /*
-  Update the parameter markers by reading the data           
-  from client ..                                             
+  Update the parameter markers by reading data from client packet 
+  and if binary/update log is set, generate the valid query.
 */
 
+static bool insert_params_withlog(PREP_STMT *stmt, uchar *pos, uchar *read_pos)
+{
+  THD *thd= stmt->thd;
+  List<Item> &params= thd->lex.param_list;
+  List_iterator<Item> param_iterator(params);
+  Item_param *param;
+  DBUG_ENTER("insert_params_withlog"); 
+  
+  String str, *res, *query= new String(stmt->query->alloced_length());  
+  query->copy(*stmt->query);
+  
+  ulong param_no= 0;  
+  uint32 length= 0;
+  
+  while ((param= (Item_param *)param_iterator++))
+  {
+    if (param->long_data_supplied)
+      res= param->query_val_str(&str);       
+    
+    else
+    {
+      if (IS_PARAM_NULL(pos,param_no))
+      {
+        param->maybe_null= param->null_value= 1;
+        res= &null_string;
+      }
+      else
+      {
+        param->maybe_null= param->null_value= 0;
+        param->setup_param_func(param,&read_pos);
+        res= param->query_val_str(&str);
+      }
+    }
+    if (query->replace(param->pos_in_query+length, 1, *res))
+      DBUG_RETURN(1);
+    
+    length+= res->length()-1;
+    param_no++;
+  }
+  if (alloc_query(stmt->thd, (char *)query->ptr(), query->length()+1))
+    DBUG_RETURN(1);
+  
+  query->free();
+  DBUG_RETURN(0);
+}
+
+static bool insert_params(PREP_STMT *stmt, uchar *pos, uchar *read_pos)
+{
+  THD *thd= stmt->thd;
+  List<Item> &params= thd->lex.param_list;
+  List_iterator<Item> param_iterator(params);
+  Item_param *param;
+  DBUG_ENTER("insert_params"); 
+  
+  ulong param_no= 0;  
+  while ((param= (Item_param *)param_iterator++))
+  {
+    if (!param->long_data_supplied)   
+    {
+      if (IS_PARAM_NULL(pos,param_no))
+        param->maybe_null= param->null_value= 1;
+      else
+      {
+        param->maybe_null= param->null_value= 0;
+        param->setup_param_func(param,&read_pos);
+      }
+    }
+    param_no++;
+  }
+  DBUG_RETURN(0);
+}
+
 static bool setup_params_data(PREP_STMT *stmt)
 {                                       
   THD *thd= stmt->thd;
@@ -418,21 +495,7 @@ static bool setup_params_data(PREP_STMT *stmt)
     }
     param_iterator.rewind();
   }    
-  ulong param_no= 0;
-  while ((param= (Item_param *)param_iterator++))
-  {
-    if (!param->long_data_supplied)
-    {
-      if (IS_PARAM_NULL(pos,param_no))
-        param->maybe_null= param->null_value= 1;
-      else
-      {
-        param->maybe_null= param->null_value= 0;
-        param->setup_param_func(param,&read_pos);
-      }
-    }
-    param_no++;
-  }
+  stmt->setup_params(stmt,pos,read_pos);
   DBUG_RETURN(0);
 }
 
@@ -707,21 +770,42 @@ static bool parse_prepare_query(PREP_STMT *stmt,
 
 static bool init_param_items(PREP_STMT *stmt)
 {
-  List<Item> &params= stmt->thd->lex.param_list;
+  THD *thd= stmt->thd;
+  List<Item> &params= thd->lex.param_list;
   Item_param **to;
+  uint32 length= thd->query_length;
  
-  stmt->lex=  stmt->thd->lex;
+  stmt->lex=  thd->lex;
+
+  if (mysql_bin_log.is_open() || mysql_update_log.is_open())
+  {
+    stmt->log_full_query= 1;
+    stmt->setup_params= insert_params_withlog;
+  }
+  else
+    stmt->setup_params= insert_params; // not fully qualified query
+   
   if (!stmt->param_count)
     stmt->param= (Item_param **)0;
   else
-  {
+  {    
     if (!(stmt->param= to= (Item_param **)
           my_malloc(sizeof(Item_param *)*(stmt->param_count+1), 
                     MYF(MY_WME))))
       return 1;
+
+    if (stmt->log_full_query)
+    {
+      length= thd->query_length+(stmt->param_count*2)+1;
+ 
+      if ( length < STMT_QUERY_LOG_LENGTH ) 
+        length= STMT_QUERY_LOG_LENGTH;
+    }
     List_iterator<Item> param_iterator(params);
     while ((*(to++)= (Item_param *)param_iterator++));
-  }
+  }  
+  stmt->query= new String(length);
+  stmt->query->copy(thd->query, thd->query_length, default_charset_info);
   return 0;
 }
 
@@ -741,6 +825,12 @@ static void init_stmt_execute(PREP_STMT *stmt)
   */  
   for (; tables ; tables= tables->next)
     tables->table= 0; //safety - nasty init
+  
+  if (!(stmt->log_full_query && stmt->param_count))
+  {
+    thd->query= stmt->query->c_ptr();
+    thd->query_length= stmt->query->length();
+  }
 }
 
 /*

sql/sql_string.cc

@@ -508,14 +508,20 @@ int String::strrstr(const String &s,uint32 offset)
 
 bool String::replace(uint32 offset,uint32 arg_length,const String &to)
 {
-  long diff = (long) to.length()-(long) arg_length;
+  return replace(offset,arg_length,to.ptr(),to.length());
+}
+
+bool String::replace(uint32 offset,uint32 arg_length,
+                     const char *to,uint32 length)
+{
+  long diff = (long) length-(long) arg_length;
   if (offset+arg_length <= str_length)
   {
     if (diff < 0)
     {
-      if (to.length())
-	memcpy(Ptr+offset,to.ptr(),to.length());
-      bmove(Ptr+offset+to.length(),Ptr+offset+arg_length,
+      if (length)
+	memcpy(Ptr+offset,to,length);
+      bmove(Ptr+offset+length,Ptr+offset+arg_length,
 	    str_length-offset-arg_length);
     }
     else
@@ -527,14 +533,15 @@ bool String::replace(uint32 offset,uint32 arg_length,const String &to)
 	bmove_upp(Ptr+str_length+diff,Ptr+str_length,
 		  str_length-offset-arg_length);
       }
-      if (to.length())
-	memcpy(Ptr+offset,to.ptr(),to.length());
+      if (length)
+	memcpy(Ptr+offset,to,length);
     }
     str_length+=(uint32) diff;
   }
   return FALSE;
 }
 
+
 // added by Holyfoot for "geometry" needs
 int String::reserve(uint32 space_needed, uint32 grow_by)
 {

sql/sql_string.h

@@ -186,6 +186,7 @@ class String
   int strstr(const String &search,uint32 offset=0); // Returns offset to substring or -1
   int strstr_case(const String &s,uint32 offset=0);
   int strrstr(const String &search,uint32 offset=0); // Returns offset to substring or -1
+  bool replace(uint32 offset,uint32 arg_length,const char *to,uint32 length);
   bool replace(uint32 offset,uint32 arg_length,const String &to);
   inline bool append(char chr)
   {

sql/sql_yacc.yy

@@ -3932,7 +3932,7 @@ param_marker:
 	  LEX *lex=Lex;
           if (YYTHD->prepare_command)
           {
-            lex->param_list.push_back($$=new Item_param());
+            lex->param_list.push_back($$=new Item_param((uint)(lex->tok_start-(uchar *)YYTHD->query)));
             lex->param_count++;
           }
           else