Approximative fixes for BUG#2610,2611,9100 i.e. WL#2146 binlogging/replication of routines (stored procs and functions).

  Approximative, because it's using our binlogging way (what we call "query"-level) and this is not as good as record-level binlog (5.1) would be. It imposes several
  limitations to routines, and has caveats (which I'll document, and for which the server will try to issue errors but that is not always possible).
  Reason I don't propagate caller info to the binlog as planned is that on master and slave
  users may be different; even with that some caveats would remain.


mysql-test/mysql-test-run.sh:
  In the testsuite we know what we do, we are not creating nasty routines, and breaking binlog is ok except in rpl_sp.
mysql-test/r/blackhole.result:
  Updating results now that 4.1 has been merged
mysql-test/valgrind.supp:
      Some suppressions for Valgrind (useful on my machine Suse 9.1);
      this is just adding to the already existing suppressions of pthread and dl.
sql/item_func.cc:
      Don't binlog the substatements when executing a function. If the function
      is declared to modify data and does not complete, warning "broken binlog".
      Note that SELECT myfunc() will not be binlogged even if myfunc() updates data (will be documented);
      but INSERT INTO t VALUES(myfunc()) will be binlogged (what decides is if the caller
      gets binlogged; the function changes nothing to binlogging).
sql/log_event.cc:
      Just making functions which can be re-used when we binlog more strings
      in status_vars in Query_log_event (e.g. one day "user", "host").
sql/log_event.h:
  comment
sql/mysql_priv.h:
      --log-bin-trust-routine-creators
sql/mysqld.cc:
      --log-bin-trust-routine-creators
sql/set_var.cc:
      --log-bin-trust-routine-creators
sql/share/errmsg.txt:
  error messages to warn about problems with routines and binlog
sql/slave.cc:
      If in a routine, replication table inclusion/exclusion rules always answer "replicate!" (see comment in code).
sql/sp.cc:
      If binlog is on: errors if one wants to create a non-deterministic update routine
      (repeatability problem - note that the test is not perfect for functions) or does not have SUPER (because routines can easily
      be made to destroy slave's data with just CREATE ROUTINE and EXECUTE priv on master).
      --log-bin-trust-routine-creators removes these errors.
      Binlogging of CREATE PROCEDURE|FUNCTION.
sql/sql_acl.cc:
      No thd==0 in tables_ok().
sql/sql_parse.cc:
      Binlogging of CALL (and not of the substatements of the SP).
      If SP returns error, we don't binlog it (see comment); we push warning in this case.
      Binlogging of ALTER|DROP PROCEDURE|FUNCTION with safety messages.

11 files changed, 216 insertions, 40 deletions

diff --git a/sql/item_func.cc b/sql/item_func.cc
index b0fe0a78844..691e34b85a6 100644
--- a/sql/item_func.cc
+++ b/sql/item_func.cc
@@ -4671,9 +4671,24 @@ Item_func_sp::execute(Item **itp)
     DBUG_RETURN(-1);
   }
 #endif
+  /*
+    Like for SPs, we don't binlog the substatements. If the statement which
+    called this function is an update statement, it will be binlogged; but if
+    it's not (e.g. SELECT myfunc()) it won't be binlogged (documented known
+    problem).
+  */
+  tmp_disable_binlog(thd); /* don't binlog the substatements */
 
   res= m_sp->execute_function(thd, args, arg_count, itp);
 
+  reenable_binlog(thd);
+  if (res && mysql_bin_log.is_open() &&
+      (m_sp->m_chistics->daccess == SP_CONTAINS_SQL ||
+       m_sp->m_chistics->daccess == SP_MODIFIES_SQL_DATA))
+    push_warning(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
+                 ER_FAILED_ROUTINE_BREAK_BINLOG,
+		 ER(ER_FAILED_ROUTINE_BREAK_BINLOG));
+
 #ifndef NO_EMBEDDED_ACCESS_CHECKS
   sp_restore_security_context(thd, m_sp, &save_ctx);
 #endif
diff --git a/sql/log_event.cc b/sql/log_event.cc
index 86d31a9c2e8..42134d2b990 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -955,6 +955,18 @@ void Query_log_event::pack_info(Protocol *protocol)
 
 #ifndef MYSQL_CLIENT
 
+/* Utility function for the next method */
+static void write_str_with_code_and_len(char **dst, const char *src,
+                                        int len, uint code)
+{
+  DBUG_ASSERT(src);
+  *((*dst)++)= code;
+  *((*dst)++)= (uchar) len;
+  bmove(*dst, src, len);
+  (*dst)+= len;
+}
+
+
 /*
   Query_log_event::write()
 
@@ -1039,12 +1051,10 @@ bool Query_log_event::write(IO_CACHE* file)
     int8store(start, (ulonglong)sql_mode);
     start+= 8;
   }
-  if (catalog_len) // i.e. "catalog inited" (false for 4.0 events)
+  if (catalog_len) // i.e. this var is inited (false for 4.0 events)
   {
-    *start++= Q_CATALOG_NZ_CODE;
-    *start++= (uchar) catalog_len;
-    bmove(start, catalog, catalog_len);
-    start+= catalog_len;
+    write_str_with_code_and_len((char **)(&start),
+                                catalog, catalog_len, Q_CATALOG_NZ_CODE);
     /*
       In 5.0.x where x<4 masters we used to store the end zero here. This was
       a waste of one byte so we don't do it in x>=4 masters. We change code to
@@ -1176,6 +1186,25 @@ Query_log_event::Query_log_event(THD* thd_arg, const char* query_arg,
 #endif /* MYSQL_CLIENT */
 
 
+/* 2 utility functions for the next method */
+
+static void get_str_len_and_pointer(const char **dst, const char **src, uint *len)
+{
+  if ((*len= **src))
+    *dst= *src + 1;                          // Will be copied later
+  (*src)+= *len+1;
+}
+
+
+static void copy_str_and_move(char **dst, const char **src, uint len)
+{
+  memcpy(*dst, *src, len);
+  *src= *dst;
+  (*dst)+= len;
+  *(*dst)++= 0;
+}
+
+
 /*
   Query_log_event::Query_log_event()
   This is used by the SQL slave thread to prepare the event before execution.
@@ -1264,9 +1293,7 @@ Query_log_event::Query_log_event(const char* buf, uint event_len,
       break;
     }
     case Q_CATALOG_NZ_CODE:
-      if ((catalog_len= *pos))
-        catalog= (char*) pos+1;                 // Will be copied later
-      pos+= catalog_len+1;
+      get_str_len_and_pointer(&catalog, (const char **)(&pos), &catalog_len);
       break;
     case Q_AUTO_INCREMENT:
       auto_increment_increment= uint2korr(pos);
@@ -1282,9 +1309,7 @@ Query_log_event::Query_log_event(const char* buf, uint event_len,
     }
     case Q_TIME_ZONE_CODE:
     {
-      if ((time_zone_len= *pos))
-        time_zone_str= (char *)(pos+1);
-      pos+= time_zone_len+1;
+      get_str_len_and_pointer(&time_zone_str, (const char **)(&pos), &time_zone_len);
       break;
     }
     case Q_CATALOG_CODE: /* for 5.0.x where 0<=x<=3 masters */
@@ -1308,12 +1333,7 @@ Query_log_event::Query_log_event(const char* buf, uint event_len,
   if (catalog_len)                                  // If catalog is given
   {
     if (likely(catalog_nz)) // true except if event comes from 5.0.0|1|2|3.
-    {
-      memcpy(start, catalog, catalog_len);
-      catalog= start;
-      start+= catalog_len;
-      *start++= 0;
-    }
+      copy_str_and_move(&start, &catalog, catalog_len);
     else
     {
       memcpy(start, catalog, catalog_len+1); // copy end 0
@@ -1322,12 +1342,8 @@ Query_log_event::Query_log_event(const char* buf, uint event_len,
     }
   }
   if (time_zone_len)
-  {
-    memcpy(start, time_zone_str, time_zone_len);
-    time_zone_str= start;
-    start+= time_zone_len;
-    *start++= 0;
-  }
+    copy_str_and_move(&start, &time_zone_str, time_zone_len);
+
   /* A 2nd variable part; this is common to all versions */ 
   memcpy((char*) start, end, data_len);          // Copy db and query
   start[data_len]= '\0';              // End query with \0 (For safetly)
diff --git a/sql/log_event.h b/sql/log_event.h
index 2985fcabb50..f8989c4994a 100644
--- a/sql/log_event.h
+++ b/sql/log_event.h
@@ -234,13 +234,12 @@ struct sql_ex_info
 /* these are codes, not offsets; not more than 256 values (1 byte). */
 #define Q_FLAGS2_CODE           0
 #define Q_SQL_MODE_CODE         1
-#ifndef TO_BE_DELETED
 /*
   Q_CATALOG_CODE is catalog with end zero stored; it is used only by MySQL
-  5.0.x where 0<=x<=3.
+  5.0.x where 0<=x<=3. We have to keep it to be able to replicate these
+  old masters.
 */
 #define Q_CATALOG_CODE          2
-#endif
 #define Q_AUTO_INCREMENT	3
 #define Q_CHARSET_CODE          4
 #define Q_TIME_ZONE_CODE        5
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index 06c946114eb..82acd3936f0 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -1088,7 +1088,7 @@ extern my_bool opt_readonly, lower_case_file_system;
 extern my_bool opt_enable_named_pipe, opt_sync_frm, opt_allow_suspicious_udfs;
 extern my_bool opt_secure_auth;
 extern my_bool sp_automatic_privileges;
-extern my_bool opt_old_style_user_limits;
+extern my_bool opt_old_style_user_limits, trust_routine_creators;
 extern uint opt_crash_binlog_innodb;
 extern char *shared_memory_base_name, *mysqld_unix_port;
 extern bool opt_enable_shared_memory;
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 3bfa498e521..16f53101056 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -303,7 +303,7 @@ my_bool opt_log_queries_not_using_indexes= 0;
 my_bool lower_case_file_system= 0;
 my_bool opt_large_pages= 0;
 uint    opt_large_page_size= 0;
-my_bool opt_old_style_user_limits= 0;
+my_bool opt_old_style_user_limits= 0, trust_routine_creators= 0;
 /*
   True if there is at least one per-hour limit for some user, so we should
   check them before each query (and possibly reset counters when hour is
@@ -4170,6 +4170,7 @@ enum options_mysqld
   OPT_INNODB_FAST_SHUTDOWN,
   OPT_INNODB_FILE_PER_TABLE, OPT_CRASH_BINLOG_INNODB,
   OPT_INNODB_LOCKS_UNSAFE_FOR_BINLOG,
+  OPT_LOG_BIN_TRUST_ROUTINE_CREATORS,
   OPT_SAFE_SHOW_DB, OPT_INNODB_SAFE_BINLOG,
   OPT_INNODB, OPT_ISAM,
   OPT_ENGINE_CONDITION_PUSHDOWN,
@@ -4590,6 +4591,17 @@ Disable with --skip-innodb-doublewrite.", (gptr*) &innobase_use_doublewrite,
    "File that holds the names for last binary log files.",
    (gptr*) &opt_binlog_index_name, (gptr*) &opt_binlog_index_name, 0, GET_STR,
    REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
+  /*
+    This option starts with "log-bin" to emphasize that it is specific of
+    binary logging. Hopefully in 5.1 nobody will need it anymore, when we have
+    row-level binlog.
+  */
+  {"log-bin-trust-routine-creators", OPT_LOG_BIN_TRUST_ROUTINE_CREATORS,
+   "If equal to 0 (the default), then when --log-bin is used, creation of "
+   "a routine is allowed only to users having the SUPER privilege and only"
+   "if this routine may not break binary logging",
+   (gptr*) &trust_routine_creators, (gptr*) &trust_routine_creators, 0,
+   GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
   {"log-error", OPT_ERROR_LOG_FILE, "Error log file.",
    (gptr*) &log_error_file_ptr, (gptr*) &log_error_file_ptr, 0, GET_STR,
    OPT_ARG, 0, 0, 0, 0, 0, 0},
diff --git a/sql/set_var.cc b/sql/set_var.cc
index 0e6e36f63a2..99549654e11 100644
--- a/sql/set_var.cc
+++ b/sql/set_var.cc
@@ -203,6 +203,9 @@ sys_var_key_cache_long  sys_key_cache_age_threshold("key_cache_age_threshold",
 							      param_age_threshold));
 sys_var_bool_ptr	sys_local_infile("local_infile",
 					 &opt_local_infile);
+sys_var_bool_ptr       
+sys_trust_routine_creators("log_bin_trust_routine_creators",
+                           &trust_routine_creators);
 sys_var_thd_ulong	sys_log_warnings("log_warnings", &SV::log_warnings);
 sys_var_thd_ulong	sys_long_query_time("long_query_time",
 					     &SV::long_query_time);
@@ -703,6 +706,7 @@ sys_var *sys_variables[]=
   &sys_innodb_thread_sleep_delay,
   &sys_innodb_thread_concurrency,
 #endif  
+  &sys_trust_routine_creators,
   &sys_engine_condition_pushdown,
 #ifdef HAVE_NDBCLUSTER_DB
   &sys_ndb_autoincrement_prefetch_sz,
@@ -842,6 +846,7 @@ struct show_var_st init_vars[]= {
 #endif
   {"log",                     (char*) &opt_log,                     SHOW_BOOL},
   {"log_bin",                 (char*) &opt_bin_log,                 SHOW_BOOL},
+  {sys_trust_routine_creators.name,(char*) &sys_trust_routine_creators, SHOW_SYS},
   {"log_error",               (char*) log_error_file,               SHOW_CHAR},
 #ifdef HAVE_REPLICATION
   {"log_slave_updates",       (char*) &opt_log_slave_updates,       SHOW_MY_BOOL},
diff --git a/sql/share/errmsg.txt b/sql/share/errmsg.txt
index 050bbe86948..388f47bf96f 100644
--- a/sql/share/errmsg.txt
+++ b/sql/share/errmsg.txt
@@ -5346,3 +5346,9 @@ ER_SP_NO_RETSET_IN_FUNC 0A000
 	eng "Not allowed to return a result set from a function"
 ER_CANT_CREATE_GEOMETRY_OBJECT 22003 
 	eng "Cannot get geometry object from data you send to the GEOMETRY field"
+ER_FAILED_ROUTINE_BREAK_BINLOG
+	eng "A routine failed and is declared to modify data and binary logging is enabled; if non-transactional tables were updated, the binary log will miss their changes"
+ER_BINLOG_UNSAFE_ROUTINE
+	eng "This routine is declared to be non-deterministic and to modify data and binary logging is enabled (you *might* want to use the less safe log_bin_trust_routine_creators variable)"
+ER_BINLOG_CREATE_ROUTINE_NEED_SUPER
+	eng "You do not have SUPER privilege and binary logging is enabled (you *might* want to use the less safe log_bin_trust_routine_creators variable)"
diff --git a/sql/slave.cc b/sql/slave.cc
index ebf87660a0e..70803c88c3a 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -852,7 +852,7 @@ static TABLE_RULE_ENT* find_wild(DYNAMIC_ARRAY *a, const char* key, int len)
 
   SYNOPSIS
     tables_ok()
-    thd             thread (SQL slave thread normally)
+    thd             thread (SQL slave thread normally). Mustn't be null.
     tables          list of tables to check
 
   NOTES
@@ -885,6 +885,23 @@ bool tables_ok(THD* thd, TABLE_LIST* tables)
   bool some_tables_updating= 0;
   DBUG_ENTER("tables_ok");
 
+  /*
+    In routine, can't reliably pick and choose substatements, so always
+    replicate.
+    We can't reliably know if one substatement should be executed or not:
+    consider the case of this substatement: a SELECT on a non-replicated
+    constant table; if we don't execute it maybe it was going to fill a
+    variable which was going to be used by the next substatement to update
+    a replicated table? If we execute it maybe the constant non-replicated
+    table does not exist (and so we'll fail) while there was no need to
+    execute this as this SELECT does not influence replicated tables in the
+    rest of the routine? In other words: users are used to replicate-*-table
+    specifying how to handle updates to tables, these options don't say
+    anything about reads to tables; we can't guess.
+  */
+  if (thd->spcont)
+    DBUG_RETURN(1);
+
   for (; tables; tables= tables->next_global)
   {
     char hash_key[2*NAME_LEN+2];
@@ -2791,12 +2808,17 @@ static int init_slave_thread(THD* thd, SLAVE_THD_TYPE thd_type)
   DBUG_ENTER("init_slave_thread");
   thd->system_thread = (thd_type == SLAVE_THD_SQL) ?
     SYSTEM_THREAD_SLAVE_SQL : SYSTEM_THREAD_SLAVE_IO; 
+  /*
+    The two next lines are needed for replication of SP (CREATE PROCEDURE
+    needs a valid user to store in mysql.proc).
+  */
+  thd->priv_user= (char *) "";
+  thd->priv_host[0]= '\0';
   thd->host_or_ip= "";
   thd->client_capabilities = 0;
   my_net_init(&thd->net, 0);
   thd->net.read_timeout = slave_net_timeout;
   thd->master_access= ~0;
-  thd->priv_user = 0;
   thd->slave_thread = 1;
   set_slave_thread_options(thd);
   /* 
diff --git a/sql/sp.cc b/sql/sp.cc
index 1956f32f2c6..81513cb3198 100644
--- a/sql/sp.cc
+++ b/sql/sp.cc
@@ -58,6 +58,9 @@ enum
 
 bool mysql_proc_table_exists= 1;
 
+/* Tells what SP_DEFAULT_ACCESS should be mapped to */
+#define SP_DEFAULT_ACCESS_MAPPING SP_CONTAINS_SQL
+
 /* *opened=true means we opened ourselves */
 static int
 db_find_routine_aux(THD *thd, int type, sp_name *name,
@@ -189,7 +192,7 @@ db_find_routine(THD *thd, int type, sp_name *name, sp_head **sphp)
     chistics.daccess= SP_MODIFIES_SQL_DATA;
     break;
   default:
-    chistics.daccess= SP_CONTAINS_SQL;
+    chistics.daccess= SP_DEFAULT_ACCESS_MAPPING;
   }
 
   if ((ptr= get_field(thd->mem_root,
@@ -425,9 +428,46 @@ db_create_routine(THD *thd, int type, sp_head *sp)
 	store(sp->m_chistics->comment.str, sp->m_chistics->comment.length,
 	      system_charset_info);
 
+    if (!trust_routine_creators && mysql_bin_log.is_open())
+    {
+      if (!sp->m_chistics->detistic)
+      {
+	/*
+	  Note that for a _function_ this test is not enough; one could use
+	  a non-deterministic read-only function in an update statement.
+	*/
+	enum enum_sp_data_access access=
+	  (sp->m_chistics->daccess == SP_DEFAULT_ACCESS) ?
+	  SP_DEFAULT_ACCESS_MAPPING : sp->m_chistics->daccess;
+	if (access == SP_CONTAINS_SQL ||
+	    access == SP_MODIFIES_SQL_DATA)
+	{
+	  my_message(ER_BINLOG_UNSAFE_ROUTINE,
+		     ER(ER_BINLOG_UNSAFE_ROUTINE), MYF(0));
+	  ret= SP_INTERNAL_ERROR;
+	  goto done;
+	}
+      }
+      if (!(thd->master_access & SUPER_ACL))
+      {
+	my_message(ER_BINLOG_CREATE_ROUTINE_NEED_SUPER,
+		   ER(ER_BINLOG_CREATE_ROUTINE_NEED_SUPER), MYF(0));
+	ret= SP_INTERNAL_ERROR;
+	goto done;
+      }
+    }
+
     ret= SP_OK;
     if (table->file->write_row(table->record[0]))
       ret= SP_WRITE_ROW_FAILED;
+    else if (mysql_bin_log.is_open())
+    {
+      thd->clear_error();
+      /* Such a statement can always go directly to binlog, no trans cache */
+      Query_log_event qinfo(thd, thd->query, thd->query_length, 0, FALSE);
+      mysql_bin_log.write(&qinfo);
+    }
+
   }
 
 done:
diff --git a/sql/sql_acl.cc b/sql/sql_acl.cc
index df19c6e55fd..58cbf1996bd 100644
--- a/sql/sql_acl.cc
+++ b/sql/sql_acl.cc
@@ -1515,7 +1515,7 @@ static bool update_user_table(THD *thd, const char *host, const char *user,
     */
     tables.updating= 1;
     /* Thanks to bzero, tables.next==0 */
-    if (!tables_ok(0, &tables))
+    if (!tables_ok(thd, &tables))
       DBUG_RETURN(0);
   }
 #endif
@@ -2699,7 +2699,7 @@ bool mysql_table_grant(THD *thd, TABLE_LIST *table_list,
       account in tests.
     */
     tables[0].updating= tables[1].updating= tables[2].updating= 1;
-    if (!tables_ok(0, tables))
+    if (!tables_ok(thd, tables))
       DBUG_RETURN(FALSE);
   }
 #endif
@@ -2904,7 +2904,7 @@ bool mysql_procedure_grant(THD *thd, TABLE_LIST *table_list,
       account in tests.
     */
     tables[0].updating= tables[1].updating= 1;
-    if (!tables_ok(0, tables))
+    if (!tables_ok(thd, tables))
       DBUG_RETURN(FALSE);
   }
 #endif
@@ -3035,7 +3035,7 @@ bool mysql_grant(THD *thd, const char *db, List <LEX_USER> &list,
       account in tests.
     */
     tables[0].updating= tables[1].updating= 1;
-    if (!tables_ok(0, tables))
+    if (!tables_ok(thd, tables))
       DBUG_RETURN(FALSE);
   }
 #endif
@@ -4245,7 +4245,7 @@ int open_grant_tables(THD *thd, TABLE_LIST *tables)
     */
     tables[0].updating=tables[1].updating=tables[2].updating=
       tables[3].updating=tables[4].updating=1;
-    if (!tables_ok(0, tables))
+    if (!tables_ok(thd, tables))
       DBUG_RETURN(1);
     tables[0].updating=tables[1].updating=tables[2].updating=
       tables[3].updating=tables[4].updating=0;;
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 22febd50035..7b465a0c086 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -4097,7 +4097,43 @@ unsent_create_error:
 	thd->variables.select_limit= HA_POS_ERROR;
 
 	thd->row_count_func= 0;
+        tmp_disable_binlog(thd); /* don't binlog the substatements */
 	res= sp->execute_procedure(thd, &lex->value_list);
+        reenable_binlog(thd);
+
+        /*
+          We write CALL to binlog; on the opposite we didn't write the
+          substatements. That choice is necessary because the substatements
+          may use local vars.
+          Binlogging should happen when all tables are locked. They are locked
+          just above, and unlocked by close_thread_tables(). All tables which
+          are to be updated are locked like with a table-level write lock, and
+          this also applies to InnoDB (I tested - note that it reduces
+          InnoDB's concurrency as we don't use row-level locks). So binlogging
+          below is safe.
+          Note the limitation: if the SP returned an error, but still did some
+          updates, we do NOT binlog it. This is because otherwise "permission
+          denied", "table does not exist" etc would stop the slave quite
+          often. There is no easy way to know if the SP updated something
+          (even no_trans_update is not suitable, as it may be a transactional
+          autocommit update which happened, and no_trans_update covers only
+          INSERT/UPDATE/LOAD).
+        */
+        if (mysql_bin_log.is_open() &&
+            (sp->m_chistics->daccess == SP_CONTAINS_SQL ||
+             sp->m_chistics->daccess == SP_MODIFIES_SQL_DATA))
+        {
+          if (res)
+            push_warning(thd, MYSQL_ERROR::WARN_LEVEL_WARN,
+                         ER_FAILED_ROUTINE_BREAK_BINLOG,
+			 ER(ER_FAILED_ROUTINE_BREAK_BINLOG));
+          else
+          {
+            thd->clear_error();
+            Query_log_event qinfo(thd, thd->query, thd->query_length, 0, FALSE);
+            mysql_bin_log.write(&qinfo);
+          }
+        }
 
 	/*
           If warnings have been cleared, we have to clear total_warn_count
@@ -4153,14 +4189,32 @@ unsent_create_error:
 				  sp->m_name.str, 0))
 	  goto error;
 	memcpy(&lex->sp_chistics, &chistics, sizeof(lex->sp_chistics));
-	if (lex->sql_command == SQLCOM_ALTER_PROCEDURE)
-	  result= sp_update_procedure(thd, lex->spname, &lex->sp_chistics);
-	else
-	  result= sp_update_function(thd, lex->spname, &lex->sp_chistics);
+        if (!trust_routine_creators &&  mysql_bin_log.is_open() &&
+            !sp->m_chistics->detistic &&
+            (chistics.daccess == SP_CONTAINS_SQL ||
+             chistics.daccess == SP_MODIFIES_SQL_DATA))
+        {
+          my_message(ER_BINLOG_UNSAFE_ROUTINE,
+		     ER(ER_BINLOG_UNSAFE_ROUTINE), MYF(0));
+          result= SP_INTERNAL_ERROR;
+        }
+        else
+        {
+          if (lex->sql_command == SQLCOM_ALTER_PROCEDURE)
+            result= sp_update_procedure(thd, lex->spname, &lex->sp_chistics);
+          else
+            result= sp_update_function(thd, lex->spname, &lex->sp_chistics);
+        }
       }
       switch (result)
       {
       case SP_OK:
+        if (mysql_bin_log.is_open())
+        {
+          thd->clear_error();
+          Query_log_event qinfo(thd, thd->query, thd->query_length, 0, FALSE);
+          mysql_bin_log.write(&qinfo);
+        }
 	send_ok(thd);
 	break;
       case SP_KEY_NOT_FOUND:
@@ -4237,6 +4291,12 @@ unsent_create_error:
       switch (result)
       {
       case SP_OK:
+        if (mysql_bin_log.is_open())
+        {
+          thd->clear_error();
+          Query_log_event qinfo(thd, thd->query, thd->query_length, 0, FALSE);
+          mysql_bin_log.write(&qinfo);
+        }
 	send_ok(thd);
 	break;
       case SP_KEY_NOT_FOUND:
@@ -4495,6 +4555,7 @@ unsent_create_error:
     break;
   }
   thd->proc_info="query end";
+  /* Two binlog-related cleanups: */
   if (thd->one_shot_set)
   {
     /*