Scrum task 845. Thi is a behaviour change :

now by default, FLUSH, OPTIMIZE, ANALYZE, REPAIR commands are written to the
binlog, unless the new NO_WRITE_TO_BINLOG keyword was used :
OPTIMIZE NO_WRITE_TO_BINLOG table t;

Previously these commands were never written to the binlog, but there are
2 reasons to change this :
- the RENAME TABLE in MERGE table bug (#175) on slave
- the possible "differently optimised queries may lead to different
updates on the master and slave" bug, until we have automatic ORDER BY.

FLUSH LOGS/SLAVE/MASTER/TABLES WITH READ LOCK are never written to the binlog.
New test for the new logging behaviour.
Other small change : reload_acl_and_cache() and reset_slave() don't send their errors themselves,
this is more usual.


mysql-test/mysql-test-run.sh:
  rpl_flush_tables.test generates 'table xx is open on rename'.
  This is normal and done on purpose, so don't report it.
sql/lex.h:
  New keyword NO_WRITE_TO_BINLOG
sql/mysql_priv.h:
  reload_acl_and_cache() now decides if we want to write the FLUSH command
  to the binlog or not (FLUSH MASTER, FLUSH SLAVE, FLUSH TABLES
  WITH READ LOCK, FLUSH LOGS cannot go into the binlog).
sql/mysqld.cc:
  updated for new prototype of reload_acl_and_cache().
sql/sql_lex.h:
  New boolean no_write_to_binlog in the lex structure.
sql/sql_parse.cc:
  reload_acl_and_cache() now does not send its errors itself;
  it saves the error and the caller sends it.
  FLUSH, OPTIMIZE, ANALYZE, REPAIR commands don't write to the binlog
  if the NO_WRITE_TO_BINLOG keyword was used.
sql/sql_repl.cc:
  reset_slave() does not send its errors himself.
sql/sql_yacc.yy:
  New optional keyword NO_WRITE_TO_BINLOG for OPTIMIZE/ANALYZE/REPAIR/FLUSH :
  OPTIMIZE NO_WRITE_TO_BINLOG TABLE t;
  ANALYZE NO_WRITE_TO_BINLOG TABLE t;
  REPAIR NO_WRITE_TO_BINLOG TABLE t;
  FLUSH NO_WRITE_TO_BINLOG TABLE t;

7 files changed, 128 insertions, 45 deletions

diff --git a/sql/lex.h b/sql/lex.h
index e51b3efff87..f85431aadf7 100644
--- a/sql/lex.h
+++ b/sql/lex.h
@@ -412,6 +412,7 @@ static SYMBOL symbols[] = {
   { "WRITE",		SYM(WRITE_SYM),0,0},
   { "WHEN",		SYM(WHEN_SYM),0,0},
   { "WHERE",		SYM(WHERE),0,0},
+  { "NO_WRITE_TO_BINLOG",  SYM(NO_WRITE_TO_BINLOG),0,0},
   { "XOR",		SYM(XOR),0,0},
   { "X509",		SYM(X509_SYM),0,0},
   { "YEAR",		SYM(YEAR_SYM),0,0},
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index 4ecccbf4511..ea6e544a1fd 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -369,7 +369,8 @@ bool do_command(THD *thd);
 bool dispatch_command(enum enum_server_command command, THD *thd,
 		      char* packet, uint packet_length);
 bool check_stack_overrun(THD *thd,char *dummy);
-bool reload_acl_and_cache(THD *thd, ulong options, TABLE_LIST *tables);
+bool reload_acl_and_cache(THD *thd, ulong options, TABLE_LIST *tables, 
+                          bool *write_to_binlog);
 void table_cache_init(void);
 void table_cache_free(void);
 uint cached_tables(void);
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index dae58dd98e8..ba9ecfa1d27 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -1496,7 +1496,7 @@ static void check_data_home(const char *path)
 static void sig_reload(int signo)
 {
  // Flush everything
-  reload_acl_and_cache((THD*) 0,REFRESH_LOG, (TABLE_LIST*) 0);
+  reload_acl_and_cache((THD*) 0,REFRESH_LOG, (TABLE_LIST*) 0, NULL);
   signal(signo, SIG_ACK);
 }
 
@@ -1832,7 +1832,7 @@ extern "C" void *signal_hand(void *arg __attribute__((unused)))
 			     (REFRESH_LOG | REFRESH_TABLES | REFRESH_FAST |
 			      REFRESH_STATUS | REFRESH_GRANT |
 			      REFRESH_THREADS | REFRESH_HOSTS),
-			     (TABLE_LIST*) 0); // Flush logs
+			     (TABLE_LIST*) 0, NULL); // Flush logs
 	mysql_print_status((THD*) 0);		// Send debug some info
       }
       break;
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index e03814bcd2f..f31b3305e07 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -481,7 +481,7 @@ typedef struct st_lex
   uint fk_delete_opt, fk_update_opt, fk_match_option;
   uint param_count;
   bool drop_primary, drop_if_exists, drop_temporary, local_file;
-  bool in_comment, ignore_space, verbose, simple_alter;
+  bool in_comment, ignore_space, verbose, simple_alter, no_write_to_binlog;
   bool derived_tables, describe;
   bool safe_to_cache_query;
   uint slave_thd_opt;
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 0f2764fb1ab..1d2449839d8 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -1391,8 +1391,10 @@ restore_user:
       if (check_global_access(thd,RELOAD_ACL))
 	break;
       mysql_log.write(thd,command,NullS);
-      /* error sending is deferred to reload_acl_and_cache */
-      reload_acl_and_cache(thd, options, (TABLE_LIST*) 0) ;
+      if (reload_acl_and_cache(thd, options, (TABLE_LIST*) 0, NULL))
+        send_error(thd, 0);
+      else
+        send_ok(thd);
       break;
     }
 #ifndef EMBEDDED_LIBRARY
@@ -2164,6 +2166,16 @@ mysql_execute_command(THD *thd)
 	check_table_access(thd,SELECT_ACL | INSERT_ACL, tables))
       goto error; /* purecov: inspected */
     res = mysql_repair_table(thd, tables, &lex->check_opt);
+    /* ! we write after unlocking the table */
+    if (!res && !lex->no_write_to_binlog)
+    {
+      mysql_update_log.write(thd, thd->query, thd->query_length);
+      if (mysql_bin_log.is_open())
+      {
+        Query_log_event qinfo(thd, thd->query, thd->query_length, 0);
+        mysql_bin_log.write(&qinfo);
+      }
+    }
     break;
   }
   case SQLCOM_CHECK:
@@ -2180,6 +2192,16 @@ mysql_execute_command(THD *thd)
 	check_table_access(thd,SELECT_ACL | INSERT_ACL, tables))
       goto error; /* purecov: inspected */
     res = mysql_analyze_table(thd, tables, &lex->check_opt);
+    /* ! we write after unlocking the table */
+    if (!res && !lex->no_write_to_binlog)
+    {
+      mysql_update_log.write(thd, thd->query, thd->query_length);
+      if (mysql_bin_log.is_open())
+      {
+        Query_log_event qinfo(thd, thd->query, thd->query_length, 0);
+        mysql_bin_log.write(&qinfo);
+      }
+    }
     break;
   }
 
@@ -2209,6 +2231,16 @@ mysql_execute_command(THD *thd)
     }
     else
       res = mysql_optimize_table(thd, tables, &lex->check_opt);
+    /* ! we write after unlocking the table */
+    if (!res && !lex->no_write_to_binlog)
+    {
+      mysql_update_log.write(thd, thd->query, thd->query_length);
+      if (mysql_bin_log.is_open())
+      {
+        Query_log_event qinfo(thd, thd->query, thd->query_length, 0);
+        mysql_bin_log.write(&qinfo);
+      }
+    }
     break;
   }
   case SQLCOM_UPDATE:
@@ -2894,13 +2926,42 @@ mysql_execute_command(THD *thd)
     }
     break;
   }
-  case SQLCOM_FLUSH:
   case SQLCOM_RESET:
+    /* 
+       RESET commands are never written to the binary log, so we have to
+       initialize this variable because RESET shares the same code as FLUSH
+    */
+    lex->no_write_to_binlog= 1;
+  case SQLCOM_FLUSH:
+  {
     if (check_global_access(thd,RELOAD_ACL) || check_db_used(thd, tables))
       goto error;
-    /* error sending is deferred to reload_acl_and_cache */
-    reload_acl_and_cache(thd, lex->type, tables);
+    /*
+      reload_acl_and_cache() will tell us if we are allowed to write to the
+      binlog or not.
+    */
+    bool write_to_binlog;
+    if (reload_acl_and_cache(thd, lex->type, tables, &write_to_binlog))
+      send_error(thd, 0);
+    else
+    {
+      /*
+        We WANT to write and we CAN write.
+        ! we write after unlocking the table.
+      */
+      if (!lex->no_write_to_binlog && write_to_binlog)
+      {
+        mysql_update_log.write(thd, thd->query, thd->query_length);
+        if (mysql_bin_log.is_open())
+        {
+          Query_log_event qinfo(thd, thd->query, thd->query_length, 0);
+          mysql_bin_log.write(&qinfo);
+        }
+      }
+      send_ok(thd);
+    }
     break;
+  }
   case SQLCOM_KILL:
     kill_one_thread(thd,lex->thread_id);
     break;
@@ -3957,14 +4018,31 @@ void add_join_natural(TABLE_LIST *a,TABLE_LIST *b)
 
 
 /*
-  Reload/resets privileges and the different caches
+  Reload/resets privileges and the different caches.
+
+  SYNOPSIS
+    reload_acl_and_cache()
+    thd			Thread handler
+    options             What should be reset/reloaded (tables, privileges,
+    slave...)
+    tables              Tables to flush (if any)
+    write_to_binlog     Depending on 'options', it may be very bad to write the
+                        query to the binlog (e.g. FLUSH SLAVE); this is a
+                        pointer where, if it is not NULL, reload_acl_and_cache()
+                        will put 0 if it thinks we really should not write to
+                        the binlog. Otherwise it will put 1.
+
+  RETURN
+    0	 ok
+    !=0  error
 */
 
-bool reload_acl_and_cache(THD *thd, ulong options, TABLE_LIST *tables)
+bool reload_acl_and_cache(THD *thd, ulong options, TABLE_LIST *tables,
+                          bool *write_to_binlog)
 {
   bool result=0;
-  bool error_already_sent=0;
   select_errors=0;				/* Write if more errors */
+  bool tmp_write_to_binlog= 1;
   if (options & REFRESH_GRANT)
   {
     acl_reload(thd);
@@ -3974,6 +4052,12 @@ bool reload_acl_and_cache(THD *thd, ulong options, TABLE_LIST *tables)
   }
   if (options & REFRESH_LOG)
   {
+    /* 
+     Writing this command to the binlog may result in infinite loops when doing
+     mysqlbinlog|mysql, and anyway it does not really make sense to log it
+     automatically (would cause more trouble to users than it would help them)
+    */
+    tmp_write_to_binlog= 0;
     mysql_log.new_file(1);
     mysql_update_log.new_file(1);
     mysql_bin_log.new_file(1);
@@ -4002,10 +4086,16 @@ bool reload_acl_and_cache(THD *thd, ulong options, TABLE_LIST *tables)
     query_cache.flush();			// RESET QUERY CACHE
   }
 #endif /*HAVE_QUERY_CACHE*/
-  if (options & (REFRESH_TABLES | REFRESH_READ_LOCK))
+  /*
+    Note that if REFRESH_READ_LOCK bit is set then REFRESH_TABLES is set too
+    (see sql_yacc.yy)
+  */
+  if (options & (REFRESH_TABLES | REFRESH_READ_LOCK)) 
   {
     if ((options & REFRESH_READ_LOCK) && thd)
     {
+      // writing to the binlog could cause deadlocks, as we don't log UNLOCK TABLES
+      tmp_write_to_binlog= 0;
       if (lock_global_read_lock(thd))
 	return 1;
     }
@@ -4019,8 +4109,11 @@ bool reload_acl_and_cache(THD *thd, ulong options, TABLE_LIST *tables)
     flush_thread_cache();
 #ifndef EMBEDDED_LIBRARY
   if (options & REFRESH_MASTER)
+  {
+    tmp_write_to_binlog= 0;
     if (reset_master(thd))
       result=1;
+  }
 #endif
 #ifdef OPENSSL
    if (options & REFRESH_DES_KEY_FILE)
@@ -4032,32 +4125,17 @@ bool reload_acl_and_cache(THD *thd, ulong options, TABLE_LIST *tables)
 #ifndef EMBEDDED_LIBRARY
  if (options & REFRESH_SLAVE)
  {
+   tmp_write_to_binlog= 0;
    LOCK_ACTIVE_MI;
    if (reset_slave(thd, active_mi))
-   {
      result=1;
-     /*
-       reset_slave() sends error itself.
-       If it didn't, one would either change reset_slave()'s prototype, to
-       pass *errorcode and *errmsg to it when it's called or
-       change reset_slave to use my_error() to register the error.
-     */
-     error_already_sent=1;
-   }
    UNLOCK_ACTIVE_MI;
  }
 #endif
  if (options & REFRESH_USER_RESOURCES)
    reset_mqh(thd,(LEX_USER *) NULL);
-
- if (thd && !error_already_sent)
- {
-   if (result)
-     send_error(thd,0);
-   else
-     send_ok(thd);
- }
-
+ if (write_to_binlog)
+   *write_to_binlog= tmp_write_to_binlog;
  return result;
 }
 
diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc
index 0eb444b85c0..a39541b0fc5 100644
--- a/sql/sql_repl.cc
+++ b/sql/sql_repl.cc
@@ -746,16 +746,9 @@ int stop_slave(THD* thd, MASTER_INFO* mi, bool net_report )
     thd			Thread handler
     mi			Master info for the slave
 
-
-  NOTES
-    We don't send ok in this functions as this is called from
-    reload_acl_and_cache() which may have done other tasks, which may
-    have failed for which we want to send and error.
-
   RETURN
     0	ok
     1	error
-	In this case error is sent to the client with send_error()
 */
 
 
@@ -804,8 +797,8 @@ int reset_slave(THD *thd, MASTER_INFO* mi)
 
 err:
   unlock_slave_threads(mi);
-  if (thd && error) 
-    send_error(thd, sql_errno, errmsg);
+  if (error) 
+    my_error(sql_errno, MYF(0), errmsg);
   DBUG_RETURN(error);
 }
 
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 9314767e7c9..ad60270ccb3 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -381,6 +381,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b,int *yystacksize);
 %token	WHERE
 %token	WITH
 %token	WRITE_SYM
+%token  NO_WRITE_TO_BINLOG
 %token	X509_SYM
 %token	XOR
 %token	COMPRESSED_SYM
@@ -582,7 +583,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b,int *yystacksize);
 %type <num>
 	type int_type real_type order_dir opt_field_spec lock_option
 	udf_type if_exists opt_local opt_table_options table_options
-	table_option opt_if_not_exists opt_var_type opt_var_ident_type
+	table_option opt_if_not_exists opt_no_write_to_binlog opt_var_type opt_var_ident_type
 	delete_option opt_temporary all_or_any opt_distinct
 
 %type <ulong_num>
@@ -1718,10 +1719,11 @@ backup:
         };
 
 repair:
-	REPAIR table_or_tables
+	REPAIR opt_no_write_to_binlog table_or_tables
 	{
 	   LEX *lex=Lex;
 	   lex->sql_command = SQLCOM_REPAIR;
+           lex->no_write_to_binlog= $2;
 	   lex->check_opt.init();
 	}
 	table_list opt_mi_repair_type
@@ -1742,10 +1744,11 @@ mi_repair_type:
         | USE_FRM      { Lex->check_opt.sql_flags|= TT_USEFRM; };
 
 analyze:
-	ANALYZE_SYM table_or_tables
+	ANALYZE_SYM opt_no_write_to_binlog table_or_tables
 	{
 	   LEX *lex=Lex;
 	   lex->sql_command = SQLCOM_ANALYZE;
+           lex->no_write_to_binlog= $2;
 	   lex->check_opt.init();
 	}
 	table_list opt_mi_check_type
@@ -1779,16 +1782,22 @@ mi_check_type:
 	| CHANGED  { Lex->check_opt.flags|= T_CHECK_ONLY_CHANGED; };
 
 optimize:
-	OPTIMIZE table_or_tables
+	OPTIMIZE opt_no_write_to_binlog table_or_tables
 	{
 	   LEX *lex=Lex;
 	   lex->sql_command = SQLCOM_OPTIMIZE;
+           lex->no_write_to_binlog= $2;        
 	   lex->check_opt.init();
 	}
 	table_list opt_mi_check_type
 	{}
 	;
 
+opt_no_write_to_binlog:
+	/* empty */        { $$= 0; }
+	| NO_WRITE_TO_BINLOG  { $$= 1; }
+	;
+
 rename:
 	RENAME table_or_tables
 	{
@@ -3738,10 +3747,11 @@ opt_describe_column:
 /* flush things */
 
 flush:
-	FLUSH_SYM
+	FLUSH_SYM opt_no_write_to_binlog
 	{
 	  LEX *lex=Lex;
 	  lex->sql_command= SQLCOM_FLUSH; lex->type=0;
+          lex->no_write_to_binlog= $2; 
 	}
 	flush_options
 	{}