BUG#31168: @@hostname does not replicate

Problem: in mixed and statement mode, a query that refers to a
system variable will use the slave's value when replayed on
slave. So if the value of a system variable is inserted into a
table, the slave will differ from the master.
Fix: mark statements that refer to a system variable as "unsafe",
meaning they will be replicated by row in mixed mode and produce a warning
in statement mode. There are some exceptions: some variables are actually
replicated. Those should *not* be marked as unsafe.
BUG#34732: mysqlbinlog does not print default values for auto_increment variables
Problem: mysqlbinlog does not print default values for some variables,
including auto_increment_increment and others. So if a client executing
the output of mysqlbinlog has different default values, replication will
be wrong.
Fix: Always print default values for all variables that are replicated.
I need to fix the two bugs at the same time, because the test cases would
fail if I only fixed one of them.


include/m_ctype.h:
  Added definition of ILLEGAL_CHARSET_INFO_NUMBER. We just need a symbol
  for a number that will never be used by any charset. ~0U should be safe
  since charset numbers are sequential, starting from 0.
mysql-test/include/commit.inc:
  Upated test to avoid making statements unsafe.
mysql-test/r/commit_1innodb.result:
  Updated test needs updated result file.
mysql-test/r/mysqlbinlog.result:
  Updated result file.
mysql-test/r/mysqlbinlog2.result:
  Updated result file.
mysql-test/r/user_var-binlog.result:
  Updated result file.
mysql-test/suite/binlog/r/binlog_base64_flag.result:
  Updated result file.
mysql-test/suite/binlog/r/binlog_stm_ctype_ucs.result:
  Updated result file.
mysql-test/suite/binlog/r/binlog_unsafe.result:
  Modified test file needs modified result file.
mysql-test/suite/binlog/t/binlog_base64_flag.test:
  Need to filter out pseudo_thread_id from result since it is
  nondeterministic.
mysql-test/suite/binlog/t/binlog_unsafe.test:
  Add tests that using variables is unsafe. The 'CREATE VIEW' tests didn't
  make sense, so I removed them. SHOW WARNINGS is not necessary either,
  because we get warnings for each statement in the result file.
mysql-test/suite/rpl/r/rpl_row_mysqlbinlog.result:
  Updated result file.
mysql-test/suite/rpl/r/rpl_skip_error.result:
  Updated result file.
mysql-test/suite/rpl/t/rpl_skip_error.test:
  The test used @@server_id, which is not safe to replicate, so it would
  have given a warning. The way it used @@server_id was hackish (issue a
  query on master that removes rows only on master), so I replaced it by a
  more robust way to do the same thing (connect to slave and insert the
  rows only there).
  Also clarified what the test case does.
mysql-test/t/mysqlbinlog2.test:
  Use --short-form instead of manually filtering out nondeterministic stuff
  from mysqlbinlog (because we added the nondeterministic @@pseudo_thread_id
  to the output).
sql/item_func.cc:
  Added method of Item_func_get_system_var that indicates whether the given
  system variable will be written to the binlog or not.
sql/item_func.h:
  Added method of Item_func_get_system_var that indicates whether the given
  system variable will be written to the binlog or not.
sql/log_event.cc:
   - auto_increment_offset was not written to the binlog if
  auto_increment_increment=1
   - mysqlbinlog did not output default values for some variables
  (BUG#34732). In st_print_event_info, we remember for each variable whether
  it has been printed or not. This is achieved in different ways for
  different variables:
      - For auto_increment_*, lc_time_names, charset_database_number,
        we set the default values in st_print_event_info to something
        illegal, so that it will look like they have changed the first time
        they are seen.
      - For charset, sql_mode, pseudo_thread_id, we add a flag to
        st_print_event_info which indicates whether the variable has been
        printed.
      - Since pseudo_thread_id is now printed more often, and its value is
        not guaranteed to be constant across different runs of the same
        test script, I replaced it by a constant if --short-form is used.
   - Moved st_print_event_info constructor from log_event.h to log_event.cc,
  since it now depends on ILLEGAL_CHARSET_NUMBER, which is defined in
  m_ctype.h, which is better to include from a .cc file than from a header
  file.
sql/log_event.h:
  Added fields to st_print_event_info that indicate whether some of the
  variables have been written or not. Since the initialization of
  charset_database_number now depends on ILLEGAL_CHARSET_INFO_NUMBER, which
  is defined in a header file, which we'd better not include from this
  header file -- I moved the constructor from here to log_event.cc.
sql/set_var.cc:
  System variables now have a flag binlog_status, which indicates if they
  are written to the binlog. If nothing is specified, all variables are
  marked as not written to the binlog (NOT_IN_BINLOG) when created. In this
  file, the variables that are written to the binlog are marked with
  SESSION_VARIABLE_IN_BINLOG.
sql/set_var.h:
  Added flag binlog_status to class sys_var. Added a getter and a
  constructor parameter that sets it.
  Since I had to change sys_var_thd_enum constructor anyways, I simplified
  it to use default values of arguments instead of three copies of the
  constructor.
sql/sql_yacc.yy:
  Mark statements that refer to a system variable as "unsafe",
  meaning they will be replicated by row in mixed mode. Added comment to
  explain strange piece of code just above.
mysql-test/include/diff_tables.inc:
  New auxiliary test file that tests whether two tables (possibly one on
  master and one on slave) differ.
mysql-test/suite/rpl/r/rpl_variables.result:
  New test case needs new result file.
mysql-test/suite/rpl/r/rpl_variables_stm.result:
  New test file needs new result file.
mysql-test/suite/rpl/t/rpl_variables.test:
  Test that INSERT of @@variables is replicated correctly (by switching to
  row-based mode).
mysql-test/suite/rpl/t/rpl_variables_stm.test:
  Test that replication of @@variables which are replicated explicitly works
  as expected in statement mode (without giving warnings).

7 files changed, 283 insertions, 164 deletions

diff --git a/sql/item_func.cc b/sql/item_func.cc
index 984fa9038b7..12bb0999ffc 100644
--- a/sql/item_func.cc
+++ b/sql/item_func.cc
@@ -4789,6 +4789,12 @@ Item_func_get_system_var::fix_fields(THD *thd, Item **ref)
 }
 
 
+bool Item_func_get_system_var::is_written_to_binlog()
+{
+  return var->is_written_to_binlog(var_type);
+}
+
+
 longlong Item_func_inet_aton::val_int()
 {
   DBUG_ASSERT(fixed == 1);
diff --git a/sql/item_func.h b/sql/item_func.h
index f8eb7ff6200..9ab30a2cf93 100644
--- a/sql/item_func.h
+++ b/sql/item_func.h
@@ -1435,6 +1435,15 @@ public:
   void fix_length_and_dec() { DBUG_ASSERT(0); }
   /* TODO: fix to support views */
   const char *func_name() const { return "get_system_var"; }
+  /**
+    Indicates whether this system variable is written to the binlog or not.
+
+    Variables are written to the binlog as part of "status_vars" in
+    Query_log_event, as an Intvar_log_event, or a Rand_log_event.
+
+    @return true if the variable is written to the binlog, false otherwise.
+  */
+  bool is_written_to_binlog();
 };
 
 
diff --git a/sql/log_event.cc b/sql/log_event.cc
index af802f52611..15f1a957149 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -14,7 +14,11 @@
    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
 
 
-#ifndef MYSQL_CLIENT
+#ifdef MYSQL_CLIENT
+
+#include "mysql_priv.h"
+
+#else
 
 #ifdef USE_PRAGMA_IMPLEMENTATION
 #pragma implementation				// gcc: Class implementation
@@ -28,7 +32,9 @@
 #include "rpl_utility.h"
 #include "rpl_record.h"
 #include <my_dir.h>
+
 #endif /* MYSQL_CLIENT */
+
 #include <base64.h>
 #include <my_bitmap.h>
 
@@ -1589,7 +1595,7 @@ bool Query_log_event::write(IO_CACHE* file)
       recognize Q_CATALOG_CODE and have no problem.
     */
   }
-  if (auto_increment_increment != 1)
+  if (auto_increment_increment != 1 || auto_increment_offset != 1)
   {
     *start++= Q_AUTO_INCREMENT;
     int2store(start, auto_increment_increment);
@@ -2102,9 +2108,17 @@ void Query_log_event::print_query_header(IO_CACHE* file,
   end= strmov(end, print_event_info->delimiter);
   *end++='\n';
   my_b_write(file, (uchar*) buff, (uint) (end-buff));
-  if (flags & LOG_EVENT_THREAD_SPECIFIC_F)
+  if ((!print_event_info->thread_id_printed ||
+       ((flags & LOG_EVENT_THREAD_SPECIFIC_F) &&
+        thread_id != print_event_info->thread_id)))
+  {
+    // If --short-form, print deterministic value instead of pseudo_thread_id.
     my_b_printf(file,"SET @@session.pseudo_thread_id=%lu%s\n",
-                (ulong)thread_id, print_event_info->delimiter);
+                short_form ? 999999999 : (ulong)thread_id,
+                print_event_info->delimiter);
+    print_event_info->thread_id= thread_id;
+    print_event_info->thread_id_printed= 1;
+  }
 
   /*
     If flags2_inited==0, this is an event from 3.23 or 4.0; nothing to
@@ -2151,20 +2165,14 @@ void Query_log_event::print_query_header(IO_CACHE* file,
     gracefully). So this code should always be good.
   */
 
-  if (likely(sql_mode_inited))
+  if (likely(sql_mode_inited) &&
+      (unlikely(print_event_info->sql_mode != sql_mode ||
+                !print_event_info->sql_mode_inited)))
   {
-    if (unlikely(!print_event_info->sql_mode_inited)) /* first Query event */
-    {
-      print_event_info->sql_mode_inited= 1;
-      /* force a difference to force write */
-      print_event_info->sql_mode= ~sql_mode;
-    }
-    if (unlikely(print_event_info->sql_mode != sql_mode))
-    {
-      my_b_printf(file,"SET @@session.sql_mode=%lu%s\n",
-                  (ulong)sql_mode, print_event_info->delimiter);
-      print_event_info->sql_mode= sql_mode;
-    }
+    my_b_printf(file,"SET @@session.sql_mode=%lu%s\n",
+                (ulong)sql_mode, print_event_info->delimiter);
+    print_event_info->sql_mode= sql_mode;
+    print_event_info->sql_mode_inited= 1;
   }
   if (print_event_info->auto_increment_increment != auto_increment_increment ||
       print_event_info->auto_increment_offset != auto_increment_offset)
@@ -2178,33 +2186,28 @@ void Query_log_event::print_query_header(IO_CACHE* file,
 
   /* TODO: print the catalog when we feature SET CATALOG */
 
-  if (likely(charset_inited))
+  if (likely(charset_inited) &&
+      (unlikely(!print_event_info->charset_inited ||
+                bcmp((uchar*) print_event_info->charset, (uchar*) charset, 6))))
   {
-    if (unlikely(!print_event_info->charset_inited)) /* first Query event */
+    CHARSET_INFO *cs_info= get_charset(uint2korr(charset), MYF(MY_WME));
+    if (cs_info)
     {
-      print_event_info->charset_inited= 1;
-      print_event_info->charset[0]= ~charset[0]; // force a difference to force write
-    }
-    if (unlikely(bcmp((uchar*) print_event_info->charset, (uchar*) charset, 6)))
-    {
-      CHARSET_INFO *cs_info= get_charset(uint2korr(charset), MYF(MY_WME));
-      if (cs_info)
-      {
-        /* for mysql client */
-        my_b_printf(file, "/*!\\C %s */%s\n",
-                    cs_info->csname, print_event_info->delimiter);
-      }
-      my_b_printf(file,"SET "
-                  "@@session.character_set_client=%d,"
-                  "@@session.collation_connection=%d,"
-                  "@@session.collation_server=%d"
-                  "%s\n",
-                  uint2korr(charset),
-                  uint2korr(charset+2),
-                  uint2korr(charset+4),
-                  print_event_info->delimiter);
-      memcpy(print_event_info->charset, charset, 6);
+      /* for mysql client */
+      my_b_printf(file, "/*!\\C %s */%s\n",
+                  cs_info->csname, print_event_info->delimiter);
     }
+    my_b_printf(file,"SET "
+                "@@session.character_set_client=%d,"
+                "@@session.collation_connection=%d,"
+                "@@session.collation_server=%d"
+                "%s\n",
+                uint2korr(charset),
+                uint2korr(charset+2),
+                uint2korr(charset+4),
+                print_event_info->delimiter);
+    memcpy(print_event_info->charset, charset, 6);
+    print_event_info->charset_inited= 1;
   }
   if (time_zone_len)
   {
@@ -8631,3 +8634,34 @@ Incident_log_event::write_data_body(IO_CACHE *file)
   DBUG_ENTER("Incident_log_event::write_data_body");
   DBUG_RETURN(write_str(file, m_message.str, m_message.length));
 }
+
+
+#ifdef MYSQL_CLIENT
+/**
+  The default values for these variables should be values that are
+  *incorrect*, i.e., values that cannot occur in an event.  This way,
+  they will always be printed for the first event.
+*/
+st_print_event_info::st_print_event_info()
+  :flags2_inited(0), sql_mode_inited(0),
+   auto_increment_increment(0),auto_increment_offset(0), charset_inited(0),
+   lc_time_names_number(~0),
+   charset_database_number(ILLEGAL_CHARSET_INFO_NUMBER),
+   thread_id(0), thread_id_printed(false),
+   base64_output_mode(BASE64_OUTPUT_UNSPEC), printed_fd_event(FALSE)
+{
+  /*
+    Currently we only use static PRINT_EVENT_INFO objects, so zeroed at
+    program's startup, but these explicit bzero() is for the day someone
+    creates dynamic instances.
+  */
+  bzero(db, sizeof(db));
+  bzero(charset, sizeof(charset));
+  bzero(time_zone_str, sizeof(time_zone_str));
+  delimiter[0]= ';';
+  delimiter[1]= 0;
+  myf const flags = MYF(MY_WME | MY_NABP);
+  open_cached_file(&head_cache, NULL, NULL, 0, flags);
+  open_cached_file(&body_cache, NULL, NULL, 0, flags);
+}
+#endif
diff --git a/sql/log_event.h b/sql/log_event.h
index c46827253a3..2cf69e975f4 100644
--- a/sql/log_event.h
+++ b/sql/log_event.h
@@ -592,8 +592,9 @@ typedef struct st_print_event_info
 {
   /*
     Settings for database, sql_mode etc that comes from the last event
-    that was printed.
-   */
+    that was printed.  We cache these so that we don't have to print
+    them if they are unchanged.
+  */
   // TODO: have the last catalog here ??
   char db[FN_REFLEN+1]; // TODO: make this a LEX_STRING when thd->db is
   bool flags2_inited;
@@ -606,26 +607,10 @@ typedef struct st_print_event_info
   char time_zone_str[MAX_TIME_ZONE_NAME_LENGTH];
   uint lc_time_names_number;
   uint charset_database_number;
-  st_print_event_info()
-    :flags2_inited(0), sql_mode_inited(0),
-     auto_increment_increment(1),auto_increment_offset(1), charset_inited(0),
-     lc_time_names_number(0), charset_database_number(0),
-     base64_output_mode(BASE64_OUTPUT_UNSPEC), printed_fd_event(FALSE)
-    {
-      /*
-        Currently we only use static PRINT_EVENT_INFO objects, so zeroed at
-        program's startup, but these explicit bzero() is for the day someone
-        creates dynamic instances.
-      */
-      bzero(db, sizeof(db));
-      bzero(charset, sizeof(charset));
-      bzero(time_zone_str, sizeof(time_zone_str));
-      delimiter[0]= ';';
-      delimiter[1]= 0;
-      myf const flags = MYF(MY_WME | MY_NABP);
-      open_cached_file(&head_cache, NULL, NULL, 0, flags);
-      open_cached_file(&body_cache, NULL, NULL, 0, flags);
-    }
+  uint thread_id;
+  bool thread_id_printed;
+
+  st_print_event_info();
 
   ~st_print_event_info() {
     close_cached_file(&head_cache);
diff --git a/sql/set_var.cc b/sql/set_var.cc
index 71131df0ce3..0d1d8b92d3b 100644
--- a/sql/set_var.cc
+++ b/sql/set_var.cc
@@ -161,10 +161,14 @@ static void sys_default_slow_log_path(THD *thd, enum_var_type type);
 
 static sys_var_chain vars = { NULL, NULL };
 
-static sys_var_thd_ulong	sys_auto_increment_increment(&vars, "auto_increment_increment",
-                                                     &SV::auto_increment_increment);
-static sys_var_thd_ulong	sys_auto_increment_offset(&vars, "auto_increment_offset",
-                                                  &SV::auto_increment_offset);
+static sys_var_thd_ulong
+sys_auto_increment_increment(&vars, "auto_increment_increment",
+                             &SV::auto_increment_increment, NULL, NULL,
+                             sys_var::SESSION_VARIABLE_IN_BINLOG);
+static sys_var_thd_ulong
+sys_auto_increment_offset(&vars, "auto_increment_offset",
+                          &SV::auto_increment_offset, NULL, NULL,
+                          sys_var::SESSION_VARIABLE_IN_BINLOG);
 
 static sys_var_bool_ptr	sys_automatic_sp_privileges(&vars, "automatic_sp_privileges",
 					      &sp_automatic_privileges);
@@ -176,19 +180,25 @@ static sys_var_thd_binlog_format sys_binlog_format(&vars, "binlog_format",
                                             &SV::binlog_format);
 static sys_var_thd_ulong	sys_bulk_insert_buff_size(&vars, "bulk_insert_buffer_size",
 						  &SV::bulk_insert_buff_size);
-static sys_var_character_set_sv	sys_character_set_server(&vars, "character_set_server",
-                                        &SV::collation_server,
-                                        &default_charset_info);
+static sys_var_character_set_sv
+sys_character_set_server(&vars, "character_set_server",
+                         &SV::collation_server, &default_charset_info, 0,
+                         sys_var::SESSION_VARIABLE_IN_BINLOG);
 sys_var_const_str       sys_charset_system(&vars, "character_set_system",
                                            (char *)my_charset_utf8_general_ci.name);
-static sys_var_character_set_database	sys_character_set_database(&vars, "character_set_database");
-static sys_var_character_set_client sys_character_set_client(&vars, 
-                                        "character_set_client",
-                                        &SV::character_set_client,
-                                        &default_charset_info);
-static sys_var_character_set_sv sys_character_set_connection(&vars, "character_set_connection",
-                                        &SV::collation_connection,
-                                        &default_charset_info);
+static sys_var_character_set_database
+sys_character_set_database(&vars, "character_set_database",
+                           sys_var::SESSION_VARIABLE_IN_BINLOG);
+static sys_var_character_set_client
+sys_character_set_client(&vars, "character_set_client",
+                         &SV::character_set_client,
+                         &default_charset_info,
+                         sys_var::SESSION_VARIABLE_IN_BINLOG);
+static sys_var_character_set_sv
+sys_character_set_connection(&vars, "character_set_connection",
+                             &SV::collation_connection,
+                             &default_charset_info, 0,
+                             sys_var::SESSION_VARIABLE_IN_BINLOG);
 static sys_var_character_set_sv sys_character_set_results(&vars, "character_set_results",
                                         &SV::character_set_results,
                                         &default_charset_info, true);
@@ -199,15 +209,18 @@ static sys_var_thd_ulong	sys_completion_type(&vars, "completion_type",
 					 &SV::completion_type,
 					 check_completion_type,
 					 fix_completion_type);
-static sys_var_collation_sv sys_collation_connection(&vars, "collation_connection",
-                                        &SV::collation_connection,
-                                        &default_charset_info);
-static sys_var_collation_sv sys_collation_database(&vars, "collation_database",
-                                        &SV::collation_database,
-                                        &default_charset_info);
-static sys_var_collation_sv sys_collation_server(&vars, "collation_server",
-                                        &SV::collation_server,
-                                        &default_charset_info);
+static sys_var_collation_sv
+sys_collation_connection(&vars, "collation_connection",
+                         &SV::collation_connection, &default_charset_info,
+                         sys_var::SESSION_VARIABLE_IN_BINLOG);
+static sys_var_collation_sv
+sys_collation_database(&vars, "collation_database", &SV::collation_database,
+                       &default_charset_info,
+                       sys_var::SESSION_VARIABLE_IN_BINLOG);
+static sys_var_collation_sv
+sys_collation_server(&vars, "collation_server", &SV::collation_server,
+                     &default_charset_info,
+                     sys_var::SESSION_VARIABLE_IN_BINLOG);
 static sys_var_long_ptr	sys_concurrent_insert(&vars, "concurrent_insert",
                                               &myisam_concurrent_insert);
 static sys_var_long_ptr	sys_connect_timeout(&vars, "connect_timeout",
@@ -303,9 +316,10 @@ static sys_var_thd_ulong	sys_max_error_count(&vars, "max_error_count",
 					    &SV::max_error_count);
 static sys_var_thd_ulonglong	sys_max_heap_table_size(&vars, "max_heap_table_size",
 						&SV::max_heap_table_size);
-static sys_var_thd_ulong       sys_pseudo_thread_id(&vars, "pseudo_thread_id",
-					     &SV::pseudo_thread_id,
-                                             check_pseudo_thread_id, 0);
+static sys_var_thd_ulong sys_pseudo_thread_id(&vars, "pseudo_thread_id",
+                                              &SV::pseudo_thread_id,
+                                              check_pseudo_thread_id, 0,
+                                              sys_var::SESSION_VARIABLE_IN_BINLOG);
 static sys_var_thd_ha_rows	sys_max_join_size(&vars, "max_join_size",
 					  &SV::max_join_size,
 					  fix_max_join_size);
@@ -436,8 +450,14 @@ static sys_var_long_ptr	sys_slow_launch_time(&vars, "slow_launch_time",
 					     &slow_launch_time);
 static sys_var_thd_ulong	sys_sort_buffer(&vars, "sort_buffer_size",
 					&SV::sortbuff_size);
+/*
+  sql_mode should *not* have binlog_mode=SESSION_VARIABLE_IN_BINLOG:
+  even though it is written to the binlog, the slave ignores the
+  MODE_NO_DIR_IN_CREATE variable, so slave's value differs from
+  master's (see log_event.cc: Query_log_event::do_apply_event()).
+*/
 static sys_var_thd_sql_mode    sys_sql_mode(&vars, "sql_mode",
-                                     &SV::sql_mode);
+                                            &SV::sql_mode);
 #ifdef HAVE_OPENSSL
 extern char *opt_ssl_ca, *opt_ssl_capath, *opt_ssl_cert, *opt_ssl_cipher,
             *opt_ssl_key;
@@ -588,7 +608,8 @@ static sys_var_thd_bit	sys_sql_notes(&vars, "sql_notes", 0,
 					 OPTION_SQL_NOTES);
 static sys_var_thd_bit	sys_auto_is_null(&vars, "sql_auto_is_null", 0,
 					 set_option_bit,
-					 OPTION_AUTO_IS_NULL);
+                                         OPTION_AUTO_IS_NULL, 0,
+                                         sys_var::SESSION_VARIABLE_IN_BINLOG);
 static sys_var_thd_bit	sys_safe_updates(&vars, "sql_safe_updates", 0,
 					 set_option_bit,
 					 OPTION_SAFE_UPDATES);
@@ -601,11 +622,12 @@ static sys_var_thd_bit	sys_quote_show_create(&vars, "sql_quote_show_create", 0,
 static sys_var_thd_bit	sys_foreign_key_checks(&vars, "foreign_key_checks", 0,
 					       set_option_bit,
 					       OPTION_NO_FOREIGN_KEY_CHECKS,
-					       1);
+                                               1, sys_var::SESSION_VARIABLE_IN_BINLOG);
 static sys_var_thd_bit	sys_unique_checks(&vars, "unique_checks", 0,
 					  set_option_bit,
 					  OPTION_RELAXED_UNIQUE_CHECKS,
-					  1);
+                                          1,
+                                          sys_var::SESSION_VARIABLE_IN_BINLOG);
 #if defined(ENABLED_PROFILING) && defined(COMMUNITY_SERVER)
 static sys_var_thd_bit  sys_profiling(&vars, "profiling", NULL, 
                                       set_option_bit,
@@ -618,13 +640,41 @@ static sys_var_thd_ulong	sys_profiling_history_size(&vars, "profiling_history_si
 
 static sys_var_thd_ha_rows	sys_select_limit(&vars, "sql_select_limit",
 						 &SV::select_limit);
-static sys_var_timestamp	sys_timestamp(&vars, "timestamp");
-static sys_var_last_insert_id	sys_last_insert_id(&vars, "last_insert_id");
-static sys_var_last_insert_id	sys_identity(&vars, "identity");
+static sys_var_timestamp sys_timestamp(&vars, "timestamp",
+                                       sys_var::SESSION_VARIABLE_IN_BINLOG);
+static sys_var_last_insert_id
+sys_last_insert_id(&vars, "last_insert_id",
+                   sys_var::SESSION_VARIABLE_IN_BINLOG);
+/*
+  identity is an alias for last_insert_id(), so that we are compatible
+  with Sybase
+*/
+static sys_var_last_insert_id
+sys_identity(&vars, "identity", sys_var::SESSION_VARIABLE_IN_BINLOG);
 
-static sys_var_thd_lc_time_names       sys_lc_time_names(&vars, "lc_time_names");
+static sys_var_thd_lc_time_names
+sys_lc_time_names(&vars, "lc_time_names", sys_var::SESSION_VARIABLE_IN_BINLOG);
 
-static sys_var_insert_id	sys_insert_id(&vars, "insert_id");
+/*
+  insert_id should *not* be marked as written to the binlog (i.e., it
+  should *not* have binlog_status==SESSION_VARIABLE_IN_BINLOG),
+  because we want any statement that refers to insert_id explicitly to
+  be unsafe.  (By "explicitly", we mean using @@session.insert_id,
+  whereas insert_id is used "implicitly" when NULL value is inserted
+  into an auto_increment column).
+
+  We want statements referring explicitly to @@session.insert_id to be
+  unsafe, because insert_id is modified internally by the slave sql
+  thread when NULL values are inserted in an AUTO_INCREMENT column.
+  This modification interfers with the value of the
+  @@session.insert_id variable if @@session.insert_id is referred
+  explicitly by an insert statement (as is seen by executing "SET
+  @@session.insert_id=0; CREATE TABLE t (a INT, b INT KEY
+  AUTO_INCREMENT); INSERT INTO t(a) VALUES (@@session.insert_id);" in
+  statement-based logging mode: t will be different on master and
+  slave).
+*/
+static sys_var_insert_id sys_insert_id(&vars, "insert_id");
 static sys_var_readonly		sys_error_count(&vars, "error_count",
 						OPT_SESSION,
 						SHOW_LONG,
@@ -634,9 +684,10 @@ static sys_var_readonly		sys_warning_count(&vars, "warning_count",
 						  SHOW_LONG,
 						  get_warning_count);
 
-/* alias for last_insert_id() to be compatible with Sybase */
-static sys_var_rand_seed1	sys_rand_seed1(&vars, "rand_seed1");
-static sys_var_rand_seed2	sys_rand_seed2(&vars, "rand_seed2");
+static sys_var_rand_seed1 sys_rand_seed1(&vars, "rand_seed1",
+                                         sys_var::SESSION_VARIABLE_IN_BINLOG);
+static sys_var_rand_seed2 sys_rand_seed2(&vars, "rand_seed2",
+                                         sys_var::SESSION_VARIABLE_IN_BINLOG);
 
 static sys_var_thd_ulong        sys_default_week_format(&vars, "default_week_format",
 					                &SV::default_week_format);
@@ -644,7 +695,8 @@ static sys_var_thd_ulong        sys_default_week_format(&vars, "default_week_for
 sys_var_thd_ulong               sys_group_concat_max_len(&vars, "group_concat_max_len",
                                                          &SV::group_concat_max_len);
 
-sys_var_thd_time_zone            sys_time_zone(&vars, "time_zone");
+sys_var_thd_time_zone sys_time_zone(&vars, "time_zone",
+                                    sys_var::SESSION_VARIABLE_IN_BINLOG);
 
 /* Global read-only variable containing hostname */
 static sys_var_const_str        sys_hostname(&vars, "hostname", glob_hostname);
diff --git a/sql/set_var.h b/sql/set_var.h
index 171158fcf1e..46fe7755baf 100644
--- a/sql/set_var.h
+++ b/sql/set_var.h
@@ -48,6 +48,22 @@ struct sys_var_chain
 class sys_var
 {
 public:
+
+  /**
+    Enumeration type to indicate for a system variable whether it will be written to the binlog or not.
+  */
+  enum Binlog_status_enum
+  {  
+    /* The variable value is not in the binlog. */
+    NOT_IN_BINLOG,
+    /* The value of the @@session variable is in the binlog. */
+    SESSION_VARIABLE_IN_BINLOG
+    /*
+      Currently, no @@global variable is ever in the binlog, so we
+      don't need an enumeration value for that.
+    */
+  };
+
   sys_var *next;
   struct my_option *option_limits;	/* Updated by by set_var_init() */
   uint name_length;			/* Updated by by set_var_init() */
@@ -55,8 +71,9 @@ public:
 
   sys_after_update_func after_update;
   bool no_support_one_shot;
-  sys_var(const char *name_arg,sys_after_update_func func= NULL)
-    :name(name_arg), after_update(func)
+  sys_var(const char *name_arg, sys_after_update_func func= NULL,
+          Binlog_status_enum binlog_status_arg= NOT_IN_BINLOG)
+    :name(name_arg), after_update(func), binlog_status(binlog_status_arg)
     , no_support_one_shot(1)
   {}
   virtual ~sys_var() {}
@@ -71,6 +88,11 @@ public:
   virtual bool check(THD *thd, set_var *var);
   bool check_enum(THD *thd, set_var *var, const TYPELIB *enum_names);
   bool check_set(THD *thd, set_var *var, TYPELIB *enum_names);
+  bool is_written_to_binlog(enum_var_type type)
+  {
+    return (type == OPT_SESSION || type == OPT_DEFAULT) &&
+      (binlog_status == SESSION_VARIABLE_IN_BINLOG);
+  }
   virtual bool update(THD *thd, set_var *var)=0;
   virtual void set_default(THD *thd_arg, enum_var_type type) {}
   virtual SHOW_TYPE show_type() { return SHOW_UNDEF; }
@@ -86,6 +108,9 @@ public:
   virtual bool is_struct() { return 0; }
   virtual bool is_readonly() const { return 0; }
   virtual sys_var_pluginvar *cast_pluginvar() { return 0; }
+
+private:
+  const Binlog_status_enum binlog_status;
 };
 
 
@@ -232,8 +257,9 @@ class sys_var_const_str :public sys_var
 {
 public:
   char *value;					// Pointer to const value
-  sys_var_const_str(sys_var_chain *chain, const char *name_arg, const char *value_arg)
-    :sys_var(name_arg),value((char*) value_arg)
+  sys_var_const_str(sys_var_chain *chain, const char *name_arg,
+                    const char *value_arg)
+    :sys_var(name_arg), value((char*) value_arg)
   { chain_sys_var(chain); }
   bool check(THD *thd, set_var *var)
   {
@@ -328,8 +354,9 @@ class sys_var_thd :public sys_var
 {
 public:
   sys_var_thd(const char *name_arg, 
-              sys_after_update_func func= NULL)
-    :sys_var(name_arg,func)
+              sys_after_update_func func= NULL,
+              Binlog_status_enum binlog_status= NOT_IN_BINLOG)
+    :sys_var(name_arg, func, binlog_status)
   {}
   bool check_type(enum_var_type type) { return 0; }
   bool check_default(enum_var_type type)
@@ -344,12 +371,13 @@ class sys_var_thd_ulong :public sys_var_thd
   sys_check_func check_func;
 public:
   ulong SV::*offset;
-  sys_var_thd_ulong(sys_var_chain *chain, const char *name_arg, ulong SV::*offset_arg)
-    :sys_var_thd(name_arg), check_func(0), offset(offset_arg)
-  { chain_sys_var(chain); }
-  sys_var_thd_ulong(sys_var_chain *chain, const char *name_arg, ulong SV::*offset_arg,
-		   sys_check_func c_func, sys_after_update_func au_func)
-    :sys_var_thd(name_arg,au_func), check_func(c_func), offset(offset_arg)
+  sys_var_thd_ulong(sys_var_chain *chain, const char *name_arg,
+                    ulong SV::*offset_arg,
+                    sys_check_func c_func= NULL,
+                    sys_after_update_func au_func= NULL,
+                    Binlog_status_enum binlog_status_arg= NOT_IN_BINLOG)
+    :sys_var_thd(name_arg, au_func, binlog_status_arg), check_func(c_func),
+    offset(offset_arg)
   { chain_sys_var(chain); }
   bool check(THD *thd, set_var *var);
   bool update(THD *thd, set_var *var);
@@ -440,22 +468,12 @@ protected:
   TYPELIB *enum_names;
   sys_check_func check_func;
 public:
-  sys_var_thd_enum(sys_var_chain *chain, const char *name_arg, ulong SV::*offset_arg,
-		   TYPELIB *typelib)
-    :sys_var_thd(name_arg), offset(offset_arg), enum_names(typelib),
-    check_func(0)
-  { chain_sys_var(chain); }
-  sys_var_thd_enum(sys_var_chain *chain, const char *name_arg, ulong SV::*offset_arg,
-		   TYPELIB *typelib,
-		   sys_after_update_func func)
-    :sys_var_thd(name_arg,func), offset(offset_arg), enum_names(typelib),
-    check_func(0)
-  { chain_sys_var(chain); }
-  sys_var_thd_enum(sys_var_chain *chain, const char *name_arg, ulong SV::*offset_arg,
-		   TYPELIB *typelib, sys_after_update_func func,
-                   sys_check_func check)
-    :sys_var_thd(name_arg,func), offset(offset_arg), enum_names(typelib),
-    check_func(check)
+  sys_var_thd_enum(sys_var_chain *chain, const char *name_arg,
+                   ulong SV::*offset_arg, TYPELIB *typelib,
+                   sys_after_update_func func= NULL,
+                   sys_check_func check= NULL)
+    :sys_var_thd(name_arg, func), offset(offset_arg),
+    enum_names(typelib), check_func(check)
   { chain_sys_var(chain); }
   bool check(THD *thd, set_var *var)
   {
@@ -480,7 +498,7 @@ public:
   sys_var_thd_sql_mode(sys_var_chain *chain, const char *name_arg, 
                        ulong SV::*offset_arg)
     :sys_var_thd_enum(chain, name_arg, offset_arg, &sql_mode_typelib,
-		      fix_sql_mode_var)
+                      fix_sql_mode_var)
   {}
   bool check(THD *thd, set_var *var)
   {
@@ -534,9 +552,10 @@ public:
   bool reverse;
   sys_var_thd_bit(sys_var_chain *chain, const char *name_arg,
                   sys_check_func c_func, sys_update_func u_func,
-                  ulonglong bit, bool reverse_arg=0)
-    :sys_var_thd(name_arg), check_func(c_func), update_func(u_func),
-    bit_flag(bit), reverse(reverse_arg)
+                  ulonglong bit, bool reverse_arg=0,
+                  Binlog_status_enum binlog_status_arg= NOT_IN_BINLOG)
+    :sys_var_thd(name_arg, NULL, binlog_status_arg), check_func(c_func),
+    update_func(u_func), bit_flag(bit), reverse(reverse_arg)
   { chain_sys_var(chain); }
   bool check(THD *thd, set_var *var);
   bool update(THD *thd, set_var *var);
@@ -567,8 +586,9 @@ public:
 class sys_var_timestamp :public sys_var
 {
 public:
-  sys_var_timestamp(sys_var_chain *chain, const char *name_arg)
-    :sys_var(name_arg)
+  sys_var_timestamp(sys_var_chain *chain, const char *name_arg,
+                    Binlog_status_enum binlog_status_arg= NOT_IN_BINLOG)
+    :sys_var(name_arg, NULL, binlog_status_arg)
   { chain_sys_var(chain); }
   bool update(THD *thd, set_var *var);
   void set_default(THD *thd, enum_var_type type);
@@ -582,8 +602,9 @@ public:
 class sys_var_last_insert_id :public sys_var
 {
 public:
-  sys_var_last_insert_id(sys_var_chain *chain, const char *name_arg)
-    :sys_var(name_arg)
+  sys_var_last_insert_id(sys_var_chain *chain, const char *name_arg,
+                         Binlog_status_enum binlog_status_arg= NOT_IN_BINLOG)
+    :sys_var(name_arg, NULL, binlog_status_arg)
   { chain_sys_var(chain); }
   bool update(THD *thd, set_var *var);
   bool check_type(enum_var_type type) { return type == OPT_GLOBAL; }
@@ -608,8 +629,9 @@ public:
 class sys_var_rand_seed1 :public sys_var
 {
 public:
-  sys_var_rand_seed1(sys_var_chain *chain, const char *name_arg)
-    :sys_var(name_arg)
+  sys_var_rand_seed1(sys_var_chain *chain, const char *name_arg,
+                     Binlog_status_enum binlog_status_arg= NOT_IN_BINLOG)
+    :sys_var(name_arg, NULL, binlog_status_arg)
   { chain_sys_var(chain); }
   bool update(THD *thd, set_var *var);
   bool check_type(enum_var_type type) { return type == OPT_GLOBAL; }
@@ -618,8 +640,9 @@ public:
 class sys_var_rand_seed2 :public sys_var
 {
 public:
-  sys_var_rand_seed2(sys_var_chain *chain, const char *name_arg)
-    :sys_var(name_arg)
+  sys_var_rand_seed2(sys_var_chain *chain, const char *name_arg,
+                     Binlog_status_enum binlog_status_arg= NOT_IN_BINLOG)
+    :sys_var(name_arg, NULL, binlog_status_arg)
   { chain_sys_var(chain); }
   bool update(THD *thd, set_var *var);
   bool check_type(enum_var_type type) { return type == OPT_GLOBAL; }
@@ -629,11 +652,12 @@ public:
 class sys_var_collation :public sys_var_thd
 {
 public:
-  sys_var_collation(const char *name_arg)
-    :sys_var_thd(name_arg)
-    {
+  sys_var_collation(const char *name_arg,
+                    Binlog_status_enum binlog_status_arg= NOT_IN_BINLOG)
+    :sys_var_thd(name_arg, NULL, binlog_status_arg)
+  {
     no_support_one_shot= 0;
-    }
+  }
   bool check(THD *thd, set_var *var);
   SHOW_TYPE show_type() { return SHOW_CHAR; }
   bool check_update_type(Item_result type)
@@ -648,8 +672,9 @@ class sys_var_character_set :public sys_var_thd
 {
 public:
   bool nullable;
-  sys_var_character_set(const char *name_arg, bool is_nullable= 0) :
-    sys_var_thd(name_arg), nullable(is_nullable)
+  sys_var_character_set(const char *name_arg, bool is_nullable= 0,
+                        Binlog_status_enum binlog_status_arg= NOT_IN_BINLOG)
+    :sys_var_thd(name_arg, NULL, binlog_status_arg), nullable(is_nullable)
   {
     /*
       In fact only almost all variables derived from sys_var_character_set
@@ -678,8 +703,9 @@ public:
   sys_var_character_set_sv(sys_var_chain *chain, const char *name_arg,
 			   CHARSET_INFO *SV::*offset_arg,
 			   CHARSET_INFO **global_default_arg,
-			   bool is_nullable= 0)
-    : sys_var_character_set(name_arg, is_nullable),
+                           bool is_nullable= 0,
+                           Binlog_status_enum binlog_status_arg= NOT_IN_BINLOG)
+    : sys_var_character_set(name_arg, is_nullable, binlog_status_arg),
     offset(offset_arg), global_default(global_default_arg)
   { chain_sys_var(chain); }
   void set_default(THD *thd, enum_var_type type);
@@ -693,9 +719,9 @@ public:
   sys_var_character_set_client(sys_var_chain *chain, const char *name_arg,
                                CHARSET_INFO *SV::*offset_arg,
                                CHARSET_INFO **global_default_arg,
-                               bool is_nullable= 0)
+                               Binlog_status_enum binlog_status_arg)
     : sys_var_character_set_sv(chain, name_arg, offset_arg, global_default_arg,
-                               is_nullable)
+                               0, binlog_status_arg)
   { }
   bool check(THD *thd, set_var *var);
 };
@@ -704,8 +730,10 @@ public:
 class sys_var_character_set_database :public sys_var_character_set
 {
 public:
-  sys_var_character_set_database(sys_var_chain *chain, const char *name_arg) :
-    sys_var_character_set(name_arg)
+  sys_var_character_set_database(sys_var_chain *chain, const char *name_arg,
+                                 Binlog_status_enum binlog_status_arg=
+                                   NOT_IN_BINLOG)
+    : sys_var_character_set(name_arg, 0, binlog_status_arg)
   { chain_sys_var(chain); }
   void set_default(THD *thd, enum_var_type type);
   CHARSET_INFO **ci_ptr(THD *thd, enum_var_type type);
@@ -718,8 +746,9 @@ class sys_var_collation_sv :public sys_var_collation
 public:
   sys_var_collation_sv(sys_var_chain *chain, const char *name_arg,
 		       CHARSET_INFO *SV::*offset_arg,
-		       CHARSET_INFO **global_default_arg)
-    :sys_var_collation(name_arg),
+                       CHARSET_INFO **global_default_arg,
+                       Binlog_status_enum binlog_status_arg= NOT_IN_BINLOG)
+    :sys_var_collation(name_arg, binlog_status_arg),
     offset(offset_arg), global_default(global_default_arg)
   {
     chain_sys_var(chain);
@@ -946,8 +975,9 @@ public:
 class sys_var_thd_time_zone :public sys_var_thd
 {
 public:
-  sys_var_thd_time_zone(sys_var_chain *chain, const char *name_arg):
-    sys_var_thd(name_arg) 
+  sys_var_thd_time_zone(sys_var_chain *chain, const char *name_arg,
+                        Binlog_status_enum binlog_status_arg= NOT_IN_BINLOG)
+    :sys_var_thd(name_arg, NULL, binlog_status_arg)
   {
     no_support_one_shot= 0;
     chain_sys_var(chain);
@@ -1034,8 +1064,9 @@ public:
 class sys_var_thd_lc_time_names :public sys_var_thd
 {
 public:
-  sys_var_thd_lc_time_names(sys_var_chain *chain, const char *name_arg):
-    sys_var_thd(name_arg)
+  sys_var_thd_lc_time_names(sys_var_chain *chain, const char *name_arg,
+                            Binlog_status_enum binlog_status_arg= NOT_IN_BINLOG)
+    : sys_var_thd(name_arg, NULL, binlog_status_arg)
   {
 #if MYSQL_VERSION_ID < 50000
     no_support_one_shot= 0;
@@ -1079,9 +1110,8 @@ public:
   sys_var_thd_binlog_format(sys_var_chain *chain, const char *name_arg, 
                             ulong SV::*offset_arg)
     :sys_var_thd_enum(chain, name_arg, offset_arg,
-                      &binlog_format_typelib
-                      , fix_binlog_format_after_update
-                      )
+                      &binlog_format_typelib,
+                      fix_binlog_format_after_update)
   {};
   bool is_readonly() const;
 };
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 1bd90e74f47..98434d15d87 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -7389,6 +7389,7 @@ variable_aux:
           }
         | '@' opt_var_ident_type ident_or_text opt_component
           {
+            /* disallow "SELECT @@global.global.variable" */
             if ($3.str && $4.str && check_reserved_words(&$3))
             {
               my_parse_error(ER(ER_SYNTAX_ERROR));
@@ -7396,6 +7397,8 @@ variable_aux:
             }
             if (!($$= get_system_var(YYTHD, $2, $3, $4)))
               MYSQL_YYABORT;
+            if (!((Item_func_get_system_var*) $$)->is_written_to_binlog())
+              Lex->set_stmt_unsafe();
           }
         ;