BUG#39934: Slave stops for engine that only support row-based logging

This is a post-push fix addressing review requests and
problems with extra warnings.

Problem 1: The sub-statement where an unsafe warning was detected was
printed as part of the warning. This was ok for statements that
were unsafe due to, e.g., calls to UUID(), but did not make
sense for statements that were unsafe because there was more than
one autoincrement column (unsafeness in this case comes from the
combination of several sub-statements).
Fix 1: Instead of printing the sub-statement, print an explanation
of why the statement is unsafe.

Problem 2:
When a recursive construct (i.e., stored proceure, stored
function, trigger, view, prepared statement) contained several
sub-statements, and at least one of them was unsafe, there would be
one unsafeness warning per sub-statement - even for safe
sub-statements.
Fix 2:
Ensure that each type of warning is printed at most once, by
remembering throughout the execution of the statement which types
of warnings have been printed.


mysql-test/extra/rpl_tests/create_recursive_construct.inc:
  - Clarified comment per review request.
  - Added checks for the number of warnings in each invocation.
mysql-test/extra/rpl_tests/rpl_insert_delayed.test:
  Per review request, replaced @@session.binlog_format by
  @@global.binlog_format, since INSERT DELAYED reads the global
  variable. (In this test case, the two variables have the same
  value, so the change is cosmetic.)
mysql-test/r/sp_trans.result:
  updated result file
mysql-test/suite/binlog/r/binlog_statement_insert_delayed.result:
  updated result file
mysql-test/suite/binlog/r/binlog_stm_ps.result:
  updated result file
mysql-test/suite/binlog/r/binlog_stm_unsafe_warning.result:
  updated result file
mysql-test/suite/binlog/r/binlog_unsafe.result:
  Updated result file. Note that duplicate warnings are now gone.
mysql-test/suite/binlog/t/binlog_unsafe.test:
  - Added tests for: (1) a statement that is unsafe in many ways;
    (2) a statement that is unsafe in the same way several times.
  - Use -- style to invoke mysqltest commands.
mysql-test/suite/rpl/r/rpl_stm_found_rows.result:
  updated result file
mysql-test/suite/rpl/r/rpl_stm_loadfile.result:
  updated result file
mysql-test/suite/rpl/t/rpl_mix_found_rows.test:
  Per review request, added comment explaining what the test case
  does (copied from rpl_stm_found_rows.test)
mysql-test/suite/rpl/t/rpl_stm_found_rows.test:
  Clarified grammar in comment.
mysql-test/suite/rpl_ndb/r/rpl_ndb_binlog_format_errors.result:
  Updated result file.
sql/item_create.cc:
  Made set_stmt_unsafe take one parameter, describing the
  type of unsafeness.
sql/sp_head.cc:
  Added unsafe_flags field and made it hold all the unsafe flags.
sql/sp_head.h:
  - Removed the BINLOG_ROW_BASED_IF_MIXED flag from m_flags.
    Instead, we use the new unsafe_flags field to hold the
    unsafeness state of the sp.
  - Made propagate_attributes() copy all unsafe flags.
sql/sql_base.cc:
  - Made LEX::set_stmt_unsafe() take an extra argument.
  - Made binlog_unsafe_warning_flags store the type of unsafeness.
  - Per review requests, clarified comments
  - Added DBUG printouts
sql/sql_class.cc:
  - Made warnings be generated in issue_warnings() and call that from
    binlog_query(). Wrote issue_warnings(), which prints zero or more
    warnings, avoiding to print warnings more than once per statement.
  - Per review request, added @todo so that we remember to assert
    correct behavior in binlog_query.
sql/sql_class.h:
  - Removed BINLOG_WARNING_PRINTED 
  - Use [set|clear]_current_stmt_binlog_row_based() instead of
    modifying the flag directly.
  - added issue_unsafe_warnings() (only called from binlog_unsafe)
  - Per review request, improved some documentation.
sql/sql_insert.cc:
  Added extra argument to LEX::set_stmt_unsafe()
sql/sql_lex.h:
  - Added enum_binlog_stmt_unsafe, listing all types of unsafe
    statements.
  - Per review requests, improved many comments for member
    functions.
  - Added [get|set]_stmt_unsafe_flags(), which return/set all the
    unsafe flags for a statement.
sql/sql_parse.cc:
  - Renamed binlog_warning_flags to binlog_unsafe_warning_flags.
  - Per review requests, improved comment.
sql/sql_view.cc:
  Made views propagate all the new unsafe flags.
sql/sql_yacc.yy:
  Added parameter to set_stmt_unsafe().
storage/innobase/handler/ha_innodb.cc:
  Per review requests, replaced DBUG_EXECUTE_IF() by DBUG_EVALUATE_IF().

11 files changed, 393 insertions, 152 deletions

diff --git a/sql/item_create.cc b/sql/item_create.cc
index 23a921be91c..c124128b3f8 100644
--- a/sql/item_create.cc
+++ b/sql/item_create.cc
@@ -2379,7 +2379,7 @@ Create_udf_func::create(THD *thd, udf_func *udf, List<Item> *item_list)
   if (item_list != NULL)
     arg_count= item_list->elements;
 
-  thd->lex->set_stmt_unsafe();
+  thd->lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_UDF);
 
   DBUG_ASSERT(   (udf->type == UDFTYPE_FUNCTION)
               || (udf->type == UDFTYPE_AGGREGATE));
@@ -3365,7 +3365,7 @@ Item*
 Create_func_found_rows::create(THD *thd)
 {
   DBUG_ENTER("Create_func_found_rows::create");
-  thd->lex->set_stmt_unsafe();
+  thd->lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_FUNCTION);
   thd->lex->safe_to_cache_query= 0;
   DBUG_RETURN(new (thd->mem_root) Item_func_found_rows());
 }
@@ -3794,7 +3794,7 @@ Item*
 Create_func_load_file::create(THD *thd, Item *arg1)
 {
   DBUG_ENTER("Create_func_load_file::create");
-  thd->lex->set_stmt_unsafe();
+  thd->lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_FUNCTION);
   thd->lex->uncacheable(UNCACHEABLE_SIDEEFFECT);
   DBUG_RETURN(new (thd->mem_root) Item_load_file(arg1));
 }
@@ -4264,7 +4264,7 @@ Item*
 Create_func_row_count::create(THD *thd)
 {
   DBUG_ENTER("Create_func_row_count::create");
-  thd->lex->set_stmt_unsafe();
+  thd->lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_FUNCTION);
   thd->lex->safe_to_cache_query= 0;
   DBUG_RETURN(new (thd->mem_root) Item_func_row_count());
 }
@@ -4574,7 +4574,7 @@ Item*
 Create_func_uuid::create(THD *thd)
 {
   DBUG_ENTER("Create_func_uuid::create");
-  thd->lex->set_stmt_unsafe();
+  thd->lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_FUNCTION);
   thd->lex->safe_to_cache_query= 0;
   DBUG_RETURN(new (thd->mem_root) Item_func_uuid());
 }
@@ -4586,7 +4586,7 @@ Item*
 Create_func_uuid_short::create(THD *thd)
 {
   DBUG_ENTER("Create_func_uuid_short::create");
-  thd->lex->set_stmt_unsafe();
+  thd->lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_FUNCTION);
   thd->lex->safe_to_cache_query= 0;
   DBUG_RETURN(new (thd->mem_root) Item_func_uuid_short());
 }
diff --git a/sql/sp_head.cc b/sql/sp_head.cc
index 629a24585b3..3d1782682c1 100644
--- a/sql/sp_head.cc
+++ b/sql/sp_head.cc
@@ -508,7 +508,7 @@ sp_head::operator delete(void *ptr, size_t size) throw()
 
 sp_head::sp_head()
   :Query_arena(&main_mem_root, INITIALIZED_FOR_SP),
-   m_flags(0), m_recursion_level(0), m_next_cached_sp(0),
+   m_flags(0), unsafe_flags(0), m_recursion_level(0), m_next_cached_sp(0),
    m_cont_level(0)
 {
   const LEX_STRING str_reset= { NULL, 0 };
@@ -2104,13 +2104,10 @@ sp_head::restore_lex(THD *thd)
 
   oldlex->trg_table_fields.push_back(&sublex->trg_table_fields);
 
-  /*
-    If this substatement needs row-based, the entire routine does too (we
-    cannot switch from statement-based to row-based only for this
-    substatement).
-  */
-  if (sublex->is_stmt_unsafe())
-    m_flags|= BINLOG_ROW_BASED_IF_MIXED;
+  /* If this substatement is unsafe, the entire routine is too. */
+  DBUG_PRINT("info", ("lex->get_stmt_unsafe_flags: 0x%x",
+                      thd->lex->get_stmt_unsafe_flags()));
+  unsafe_flags|= sublex->get_stmt_unsafe_flags();
 
   /*
     Add routines which are used by statement to respective set for
diff --git a/sql/sp_head.h b/sql/sp_head.h
index 7faccb7afbb..3de9abd6760 100644
--- a/sql/sp_head.h
+++ b/sql/sp_head.h
@@ -165,9 +165,8 @@ public:
     HAS_COMMIT_OR_ROLLBACK= 128,
     LOG_SLOW_STATEMENTS= 256,   // Used by events
     LOG_GENERAL_LOG= 512,        // Used by events
-    BINLOG_ROW_BASED_IF_MIXED= 1024,
-    HAS_SQLCOM_RESET= 2048,
-    HAS_SQLCOM_FLUSH= 4096
+    HAS_SQLCOM_RESET= 1024,
+    HAS_SQLCOM_FLUSH= 2048
   };
 
   /** TYPE_ENUM_FUNCTION, TYPE_ENUM_PROCEDURE or TYPE_ENUM_TRIGGER */
@@ -198,6 +197,11 @@ public:
 
 private:
   Stored_program_creation_ctx *m_creation_ctx;
+  /**
+    Boolean combination of (1<<flag), where flag is a member of
+    LEX::enum_binlog_stmt_unsafe.
+  */
+  uint32 unsafe_flags;
 
 public:
   inline Stored_program_creation_ctx *get_creation_ctx()
@@ -453,9 +457,8 @@ public:
 #endif
 
   /*
-    This method is intended for attributes of a routine which need
-    to propagate upwards to the LEX of the caller (when a property of a
-    sp_head needs to "taint" the caller).
+    This method is intended for attributes of a routine which need to
+    propagate upwards to the LEX of the caller.
   */
   void propagate_attributes(LEX *lex)
   {
@@ -466,8 +469,11 @@ public:
       routine, as in statement-based the top-statement may be binlogged and
       the substatements not).
     */
-    if (m_flags & BINLOG_ROW_BASED_IF_MIXED)
-      lex->set_stmt_unsafe();
+    DBUG_PRINT("info", ("lex->get_stmt_unsafe_flags(): 0x%x",
+                        lex->get_stmt_unsafe_flags()));
+    DBUG_PRINT("info", ("sp_head(0x%p=%s)->unsafe_flags: 0x%x",
+                        this, name(), unsafe_flags));
+    lex->set_stmt_unsafe_flags(unsafe_flags);
     DBUG_VOID_RETURN;
   }
 
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 6359fb7b914..bec50443ee4 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -5152,12 +5152,17 @@ static void mark_real_tables_as_free_for_reuse(TABLE_LIST *table)
 int THD::decide_logging_format(TABLE_LIST *tables)
 {
   DBUG_ENTER("THD::decide_logging_format");
+  DBUG_PRINT("info", ("query: %s", query));
+  DBUG_PRINT("info", ("variables.binlog_format: %ld",
+                      variables.binlog_format));
+  DBUG_PRINT("info", ("lex->get_stmt_unsafe_flags(): 0x%x",
+                      lex->get_stmt_unsafe_flags()));
   if (mysql_bin_log.is_open() && (options & OPTION_BIN_LOG))
   {
     /*
-      Compute the starting vectors for the computations by creating a
-      set with all the capabilities bits set and one with no
-      capabilities bits set.
+      Compute one bit field with the union of all the engine
+      capabilities, and one with the intersection of all the engine
+      capabilities.
     */
     handler::Table_flags flags_some_set= 0;
     handler::Table_flags flags_all_set=
@@ -5180,15 +5185,14 @@ int THD::decide_logging_format(TABLE_LIST *tables)
 
     /*
       Get the capabilities vector for all involved storage engines and
-      mask out the flags for the binary log.  (Currently, the binlog
-      flags only include the capabilities of the storage engines.)
+      mask out the flags for the binary log.
     */
     for (TABLE_LIST *table= tables; table; table= table->next_global)
     {
       if (table->placeholder())
         continue;
       if (table->table->s->table_category == TABLE_CATEGORY_PERFORMANCE)
-        lex->set_stmt_unsafe();
+        lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_SYSTEM_TABLE);
       if (table->lock_type >= TL_WRITE_ALLOW_WRITE)
       {
         ulonglong const flags= table->table->file->ha_table_flags();
@@ -5210,12 +5214,11 @@ int THD::decide_logging_format(TABLE_LIST *tables)
     DBUG_PRINT("info", ("flags_some_set: %s%s",
                         FLAGSTR(flags_some_set, HA_BINLOG_STMT_CAPABLE),
                         FLAGSTR(flags_some_set, HA_BINLOG_ROW_CAPABLE)));
-    DBUG_PRINT("info", ("variables.binlog_format: %ld",
-                        variables.binlog_format));
     DBUG_PRINT("info", ("multi_engine: %s",
                         multi_engine ? "TRUE" : "FALSE"));
 
     int error= 0;
+    int unsafe_flags;
 
     /*
       If more than one engine is involved in the statement and at
@@ -5258,14 +5261,20 @@ int THD::decide_logging_format(TABLE_LIST *tables)
         */
         my_error((error= ER_BINLOG_ROW_MODE_AND_STMT_ENGINE), MYF(0));
       }
-      else if (lex->is_stmt_unsafe())
+      else if ((unsafe_flags= lex->get_stmt_unsafe_flags()) != 0)
       {
         /*
           3. Warning: Unsafe statement binlogged as statement since
              storage engine is limited to statement-logging.
         */
-        binlog_warning_flags|=
-          (1 << BINLOG_WARNING_FLAG_UNSAFE_AND_STMT_ENGINE);
+        binlog_unsafe_warning_flags|=
+          (1 << BINLOG_STMT_WARNING_UNSAFE_AND_STMT_ENGINE) |
+          (unsafe_flags << BINLOG_STMT_WARNING_COUNT);
+        DBUG_PRINT("info", ("Scheduling warning to be issued by "
+                            "binlog_query: %s",
+                            ER(ER_BINLOG_UNSAFE_AND_STMT_ENGINE)));
+        DBUG_PRINT("info", ("binlog_unsafe_warning_flags: 0x%x",
+                            binlog_unsafe_warning_flags));
       }
       /* log in statement format! */
     }
@@ -5291,14 +5300,20 @@ int THD::decide_logging_format(TABLE_LIST *tables)
           */
           my_error((error= ER_BINLOG_STMT_MODE_AND_ROW_ENGINE), MYF(0), "");
         }
-        else if (lex->is_stmt_unsafe())
+        else if ((unsafe_flags= lex->get_stmt_unsafe_flags()) != 0)
         {
           /*
             7. Warning: Unsafe statement logged as statement due to
                binlog_format = STATEMENT
           */
-          binlog_warning_flags|=
-            (1 << BINLOG_WARNING_FLAG_UNSAFE_AND_STMT_MODE);
+          binlog_unsafe_warning_flags|=
+            (1 << BINLOG_STMT_WARNING_UNSAFE_AND_STMT_MODE) |
+            (unsafe_flags << BINLOG_STMT_WARNING_COUNT);
+          DBUG_PRINT("info", ("Scheduling warning to be issued by "
+                              "binlog_query: '%s'",
+                              ER(ER_BINLOG_UNSAFE_STATEMENT)));
+          DBUG_PRINT("info", ("binlog_stmt_flags: 0x%x",
+                              binlog_unsafe_warning_flags));
         }
         /* log in statement format! */
       }
@@ -5315,9 +5330,21 @@ int THD::decide_logging_format(TABLE_LIST *tables)
       }
     }
 
-    if (error)
+    if (error) {
+      DBUG_PRINT("info", ("decision: no logging since an error was generated"));
       DBUG_RETURN(-1);
+    }
+    DBUG_PRINT("info", ("decision: logging in %s format",
+                        is_current_stmt_binlog_format_row() ?
+                        "ROW" : "STATEMENT"));
   }
+#ifndef DBUG_OFF
+  else
+    DBUG_PRINT("info", ("decision: no logging since "
+                        "mysql_bin_log.is_open() = %d "
+                        "and (options & OPTION_BIN_LOG) = 0x%llx",
+                        mysql_bin_log.is_open(), (options & OPTION_BIN_LOG)));
+#endif
 
   DBUG_RETURN(0);
 }
@@ -5401,7 +5428,7 @@ int lock_tables(THD *thd, TABLE_LIST *tables, uint count, bool *need_reopen)
       if (thd->variables.binlog_format == BINLOG_FORMAT_MIXED &&
           has_two_write_locked_tables_with_auto_increment(tables))
       {
-        thd->lex->set_stmt_unsafe();
+        thd->lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_TWO_AUTOINC_COLUMNS);
         thd->set_current_stmt_binlog_row_based_if_mixed();
       }
     }
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index f401daaee9b..4f8f0a30544 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -39,6 +39,7 @@
 #include <io.h>
 #endif
 #include <mysys_err.h>
+#include <limits.h>
 
 #include "sp_rcontext.h"
 #include "sp_cache.h"
@@ -540,7 +541,7 @@ THD::THD()
    lock_id(&main_lock_id),
    user_time(0), in_sub_stmt(0),
    sql_log_bin_toplevel(false),
-   binlog_warning_flags(0UL), binlog_table_maps(0),
+   binlog_unsafe_warning_flags(0), binlog_table_maps(0),
    table_map_for_update(0),
    arg_of_last_insert_id_function(FALSE),
    first_successful_insert_id_in_prev_stmt(0),
@@ -3634,6 +3635,93 @@ show_query_type(THD::enum_binlog_query_type qtype)
 
 
 /**
+  Auxiliary method used by @c binlog_query() to raise warnings.
+
+  @param err An ER_BINLOG_UNSAFE_* constant; the warning to print.
+*/
+void THD::issue_unsafe_warnings()
+{
+  DBUG_ENTER("issue_unsafe_warnings");
+  /*
+    Ensure that binlog_unsafe_warning_flags is big enough to hold all
+    bits.  This is actually a constant expression.
+  */
+  DBUG_ASSERT(BINLOG_STMT_WARNING_COUNT + 2 * LEX::BINLOG_STMT_UNSAFE_COUNT <=
+              sizeof(binlog_unsafe_warning_flags) * CHAR_BIT);
+
+  /**
+    @note The order of the elements of this array must correspond to
+    the order of elements in enum_binlog_stmt_unsafe.
+  */
+  static const char *explanations[LEX::BINLOG_STMT_UNSAFE_COUNT] =
+  {
+    "Statement uses a LIMIT clause. This is unsafe because the set of rows included cannot be predicted.",
+    "Statement uses INSERT DELAYED. This is unsafe because the time when rows are inserted cannot be predicted.",
+    "Statement uses the general_log or slow_log table. This is unsafe because system tables may differ on slave.",
+    "Statement updates two AUTO_INCREMENT columns. This is unsafe because the generated value cannot be predicted by slave.",
+    "Statement uses a UDF. It cannot be determined if the UDF will return the same value on slave.",
+    "Statement uses a system variable whose value may differ on slave.",
+    "Statement uses a system function whose value may differ on slave."
+  };
+  uint32 flags= binlog_unsafe_warning_flags;
+  /* No warnings (yet) for this statement. */
+  if (flags == 0)
+    DBUG_VOID_RETURN;
+
+  /* Get the types of unsafeness that affect the current statement. */
+  uint32 unsafe_type_flags= flags >> BINLOG_STMT_WARNING_COUNT;
+  DBUG_ASSERT((unsafe_type_flags & LEX::BINLOG_STMT_UNSAFE_ALL_FLAGS) != 0);
+  /*
+    Clear (1) bits above BINLOG_STMT_UNSAFE_COUNT; (2) bits for
+    warnings that have been printed already.
+  */
+  unsafe_type_flags &= (LEX::BINLOG_STMT_UNSAFE_ALL_FLAGS ^
+                        (unsafe_type_flags >> LEX::BINLOG_STMT_UNSAFE_COUNT));
+  /* If all warnings have been printed already, return. */
+  if (unsafe_type_flags == 0)
+    DBUG_VOID_RETURN;
+
+  /* Figure out which error code to issue. */
+  int err;
+  if (binlog_unsafe_warning_flags &
+      (1 << BINLOG_STMT_WARNING_UNSAFE_AND_STMT_ENGINE))
+    err= ER_BINLOG_UNSAFE_AND_STMT_ENGINE;
+  else {
+    DBUG_ASSERT(binlog_unsafe_warning_flags &
+                (1 << BINLOG_STMT_WARNING_UNSAFE_AND_STMT_MODE));
+    err= ER_BINLOG_UNSAFE_STATEMENT;
+  }
+
+  DBUG_PRINT("info", ("flags: 0x%x  err: %d", unsafe_type_flags, err));
+
+  /*
+    For each unsafe_type, check if the statement is unsafe in this way
+    and issue a warning.
+  */
+  for (int unsafe_type=0;
+       unsafe_type < LEX::BINLOG_STMT_UNSAFE_COUNT;
+       unsafe_type++)
+  {
+    if ((unsafe_type_flags & (1 << unsafe_type)) != 0)
+    {
+      push_warning_printf(this, MYSQL_ERROR::WARN_LEVEL_NOTE, err,
+                          "%s Reason: %s",
+                          ER(err), explanations[unsafe_type]);
+      sql_print_warning("%s Reason: %s Statement: %s",
+                        ER(err), explanations[unsafe_type], query);
+    }
+  }
+  /*
+    Mark these unsafe types as already printed, to avoid printing
+    warnings for them again.
+  */
+  binlog_unsafe_warning_flags|= unsafe_type_flags <<
+    (BINLOG_STMT_WARNING_COUNT + LEX::BINLOG_STMT_UNSAFE_COUNT);
+  DBUG_VOID_RETURN;
+}
+
+
+/**
   Log the current query.
 
   The query will be logged in either row format or statement format
@@ -3688,34 +3776,7 @@ int THD::binlog_query(THD::enum_binlog_query_type qtype, char const *query_arg,
     know for sure if the statement will be logged.
   */
   if (sql_log_bin_toplevel)
-  {
-    if (binlog_warning_flags &
-        (1 << BINLOG_WARNING_FLAG_UNSAFE_AND_STMT_ENGINE))
-    {
-      push_warning_printf(this, MYSQL_ERROR::WARN_LEVEL_NOTE,
-                          ER_BINLOG_UNSAFE_AND_STMT_ENGINE,
-                          "%s Statement: %.*s",
-                          ER(ER_BINLOG_UNSAFE_AND_STMT_ENGINE),
-                          MYSQL_ERRMSG_SIZE, query_arg);
-      sql_print_warning("%s Statement: %.*s",
-                        ER(ER_BINLOG_UNSAFE_AND_STMT_ENGINE),
-                        MYSQL_ERRMSG_SIZE, query_arg);
-      binlog_warning_flags|= 1 << BINLOG_WARNING_FLAG_PRINTED;
-    }
-    else if (binlog_warning_flags &
-             (1 << BINLOG_WARNING_FLAG_UNSAFE_AND_STMT_MODE))
-    {
-      push_warning_printf(this, MYSQL_ERROR::WARN_LEVEL_NOTE,
-                          ER_BINLOG_UNSAFE_STATEMENT,
-                          "%s Statement: %.*s",
-                          ER(ER_BINLOG_UNSAFE_STATEMENT),
-                          MYSQL_ERRMSG_SIZE, query_arg);
-      sql_print_warning("%s Statement: %.*s",
-                        ER(ER_BINLOG_UNSAFE_STATEMENT),
-                        MYSQL_ERRMSG_SIZE, query_arg);
-      binlog_warning_flags|= 1 << BINLOG_WARNING_FLAG_PRINTED;
-    }
-  }
+    issue_unsafe_warnings();
 
   switch (qtype) {
     /*
@@ -3738,6 +3799,10 @@ int THD::binlog_query(THD::enum_binlog_query_type qtype, char const *query_arg,
       format; it cannot be logged in row format.  This is typically
       used by DDL statements.  It is an error to use this query type
       if current_stmt_binlog_row_based is set.
+
+      @todo Currently there are places that call this method with
+      STMT_QUERY_TYPE and current_stmt_binlog_row_based.  Fix those
+      places and add assert to ensure correct behavior. /Sven
     */
   case THD::STMT_QUERY_TYPE:
     /*
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 5e069f1cbb6..cf008cbf865 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -1422,38 +1422,79 @@ public:
   int binlog_flush_pending_rows_event(bool stmt_end);
   int binlog_remove_pending_rows_event(bool clear_maps);
 
-  int is_current_stmt_binlog_format_row() {
+  /**
+    Determine the binlog format of the current statement.
+
+    @retval 0 if the current statement will be logged in statement
+    format.
+    @retval nonzero if the current statement will be logged in row
+    format.
+   */
+  int is_current_stmt_binlog_format_row() const {
     DBUG_ASSERT(current_stmt_binlog_format == BINLOG_FORMAT_STMT ||
                 current_stmt_binlog_format == BINLOG_FORMAT_ROW);
     return current_stmt_binlog_format == BINLOG_FORMAT_ROW;
   }
 
 private:
-  /*
-    Tells if current statement should binlog row-based(1) or stmt-based(0)
+  /**
+    Indicates the format in which the current statement will be
+    logged.  This can only be set from @c decide_logging_format().
   */
   enum_binlog_format current_stmt_binlog_format;
 
-  enum enum_binlog_warning_flag {
+  /**
+    Enumeration listing binlog-related warnings that a statement can
+    cause.
+  */
+  enum enum_binlog_stmt_warning {
+
     /* ER_BINLOG_UNSAFE_AND_STMT_ENGINE affects current stmt */
-    BINLOG_WARNING_FLAG_UNSAFE_AND_STMT_ENGINE = 0,
-    /* ER_BINLOG_UNSAFE_AND_STMT_MODE affects current stmt */
-    BINLOG_WARNING_FLAG_UNSAFE_AND_STMT_MODE,
-    /* One of the warnings has already been printed */
-    BINLOG_WARNING_FLAG_PRINTED,
-    /* number of elements of this enum; insert new members above */
-    BINLOG_WARNING_FLAG_COUNT
+    BINLOG_STMT_WARNING_UNSAFE_AND_STMT_ENGINE= 0,
+
+    /* ER_BINLOG_UNSAFE_STATEMENT affects current stmt */
+    BINLOG_STMT_WARNING_UNSAFE_AND_STMT_MODE,
+
+    /** The last element of this enumeration type. */
+    BINLOG_STMT_WARNING_COUNT
   };
+
   /**
-    Flags holding the status of binlog-related warnings for the
-    current statement.  This is a binary combination of (1<<flag),
-    where flag is a member of @c enum_binlog_warning_flag.
+    Bit field for the state of binlog warnings.
 
-    The warnings are determined in @c THD::decide_logging_format, but
-    issued only later, after the statement has been written to the
-    binlog.  Hence it must be stored in the @c THD object.
+    There are three groups of bits:
+
+    - The low BINLOG_STMT_WARNING_COUNT bits indicate the type of
+      warning that the current (top-level) statement will issue.  At
+      most one of these bits should be set (this is ensured by the
+      logic in decide_logging_format).
+
+    - The following Lex::BINLOG_STMT_UNSAFE_COUNT bits list all types
+      of unsafeness that the current statement has.
+
+    - The following Lex::BINLOG_STMT_UNSAFE_COUNT bits list all types
+      of unsafeness that the current statement has issued warnings
+      for.
+
+    Hence, this variable must be big enough to hold
+    BINLOG_STMT_WARNING_COUNT + 2 * Lex::BINLOG_STMT_UNSAFE_COUNT
+    bits.  This is asserted in @c issue_unsafe_warnings().
+
+    The first and second groups of bits are set by @c
+    decide_logging_format() when it detects that a warning should be
+    issued.  The third group of bits is set from @c binlog_query()
+    when a warning is issued.  All bits are cleared at the end of the
+    top-level statement.
+
+    This must be a member of THD and not of LEX, because warnings are
+    detected and issued in different places (@c
+    decide_logging_format() and @c binlog_query(), respectively).
+    Between these calls, the THD->lex object may change; e.g., if a
+    stored routine is invoked.  Only THD persists between the calls.
   */
-  uint32 binlog_warning_flags;
+  uint32 binlog_unsafe_warning_flags;
+
+  void issue_unsafe_warnings();
 
   /*
     Number of outstanding table maps, i.e., table maps in the
@@ -2139,6 +2180,14 @@ public:
   {
     DBUG_ENTER("set_current_stmt_binlog_row_based_if_mixed");
     /*
+      This should only be called from decide_logging_format.
+
+      @todo Once we have ensured this, uncomment the following
+      statement, remove the big comment below that, and remove the
+      in_sub_stmt==0 condition from the following 'if'.
+    */
+    /* DBUG_ASSERT(in_sub_stmt == 0); */
+    /*
       If in a stored/function trigger, the caller should already have done the
       change. We test in_sub_stmt to prevent introducing bugs where people
       wouldn't ensure that, and would switch to row-based mode in the middle
@@ -2149,7 +2198,7 @@ public:
     */
     if ((variables.binlog_format == BINLOG_FORMAT_MIXED) &&
         (in_sub_stmt == 0))
-      current_stmt_binlog_format= BINLOG_FORMAT_ROW;
+      set_current_stmt_binlog_row_based();
 
     DBUG_VOID_RETURN;
   }
@@ -2189,9 +2238,10 @@ public:
                 show_system_thread(system_thread)));
     if ((temporary_tables == NULL) && (in_sub_stmt == 0))
     {
-      current_stmt_binlog_format=
-        (variables.binlog_format == BINLOG_FORMAT_ROW) ? 
-        BINLOG_FORMAT_ROW : BINLOG_FORMAT_STMT;
+      if (variables.binlog_format == BINLOG_FORMAT_ROW)
+        set_current_stmt_binlog_row_based();
+      else
+        clear_current_stmt_binlog_row_based();
     }
     DBUG_VOID_RETURN;
   }
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index e48731079eb..3cc4a6ea1a1 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -1745,7 +1745,7 @@ public:
       decide_logging_format is made.  We should probably call
       thd->decide_logging_format() directly instead.  /Sven
     */
-    thd.lex->set_stmt_unsafe();
+    thd.lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_INSERT_DELAYED);
     thd.set_current_stmt_binlog_row_based_if_mixed();
 
     bzero((char*) &thd.net, sizeof(thd.net));		// Safety
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index 9528d8f8f29..17ef4e7abd1 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -1043,49 +1043,143 @@ public:
     }
   }
 
+
+  /**
+    Enumeration listing of all types of unsafe statement.
+
+    @note The order of elements of this enumeration type must
+    correspond to the order of the elements of the @c explanations
+    array defined in the body of @c THD::issue_unsafe_warnings.
+  */
+  enum enum_binlog_stmt_unsafe {
+    /**
+      SELECT..LIMIT is unsafe because the set of rows returned cannot
+      be predicted.
+    */
+    BINLOG_STMT_UNSAFE_LIMIT= 0,
+    /**
+      INSERT DELAYED is unsafe because the time when rows are inserted
+      cannot be predicted.
+    */
+    BINLOG_STMT_UNSAFE_INSERT_DELAYED,
+    /**
+      Access to log tables is unsafe because slave and master probably
+      log different things.
+    */
+    BINLOG_STMT_UNSAFE_SYSTEM_TABLE,
+    /**
+      Update of two autoincrement columns is unsafe.  With one
+      autoincrement column, we store the counter in the binlog so that
+      slave can restore the correct value.  But we can only store one
+      such counter per statement, so updating more than one
+      autoincrement column is not safe.
+    */
+    BINLOG_STMT_UNSAFE_TWO_AUTOINC_COLUMNS,
+    /**
+      Using a UDF (user-defined function) is unsafe.
+    */
+    BINLOG_STMT_UNSAFE_UDF,
+    /**
+      Using most system variables is unsafe, because slave may run
+      with different options than master.
+    */
+    BINLOG_STMT_UNSAFE_SYSTEM_VARIABLE,
+    /**
+      Using some functions is unsafe (e.g., UUID).
+    */
+    BINLOG_STMT_UNSAFE_FUNCTION,
+
+    /* The last element of this enumeration type. */
+    BINLOG_STMT_UNSAFE_COUNT
+  };
+  /**
+    This has all flags from 0 (inclusive) to BINLOG_STMT_FLAG_COUNT
+    (exclusive) set.
+  */
+  static const int BINLOG_STMT_UNSAFE_ALL_FLAGS=
+    ((1 << BINLOG_STMT_UNSAFE_COUNT) - 1);
+
   /**
-    Has the parser/scanner detected that this statement is unsafe?
+    Determine if this statement is marked as unsafe.
 
-    @retval 0 if the statement is not marked as unsafe
-    @retval nonzero if the statement is marked as unsafe
+    @retval 0 if the statement is not marked as unsafe.
+    @retval nonzero if the statement is marked as unsafe.
   */
   inline bool is_stmt_unsafe() const {
-    return binlog_stmt_flags & (1U << BINLOG_STMT_FLAG_UNSAFE);
+    return get_stmt_unsafe_flags() != 0;
   }
 
   /**
-    Is this statement actually a row injection?
+    Flag the current (top-level) statement as unsafe.
+    The flag will be reset after the statement has finished.
 
-    @retval 0 if the statement is not a row injection
-    @retval nonzero if the statement is a row injection
+    @param unsafe_type The type of unsafety: one of the @c
+    BINLOG_STMT_FLAG_UNSAFE_* flags in @c enum_binlog_stmt_flag.
   */
-  inline bool is_stmt_row_injection() const {
-    return binlog_stmt_flags & (1U << BINLOG_STMT_FLAG_ROW_INJECTION);
+  inline void set_stmt_unsafe(enum_binlog_stmt_unsafe unsafe_type) {
+    DBUG_ENTER("set_stmt_unsafe");
+    DBUG_ASSERT(unsafe_type >= 0 && unsafe_type < BINLOG_STMT_UNSAFE_COUNT);
+    binlog_stmt_flags|= (1U << unsafe_type);
+    DBUG_VOID_RETURN;
   }
 
   /**
-    Flag the statement as a row injection.  (A row injection is either
-    a BINLOG statement, or a row event in the relay log executed by
-    the slave SQL thread.)
+    Set the bits of binlog_stmt_flags determining the type of
+    unsafeness of the current statement.  No existing bits will be
+    cleared, but new bits may be set.
+
+    @param flags A binary combination of zero or more bits, (1<<flag)
+    where flag is a member of enum_binlog_stmt_unsafe.
   */
-  inline void set_stmt_row_injection() {
-    DBUG_ENTER("set_stmt_row_injection");
-    binlog_stmt_flags|= (1U << BINLOG_STMT_FLAG_ROW_INJECTION);
+  inline void set_stmt_unsafe_flags(uint32 flags) {
+    DBUG_ENTER("set_stmt_unsafe_flags");
+    DBUG_ASSERT((flags & ~BINLOG_STMT_UNSAFE_ALL_FLAGS) == 0);
+    binlog_stmt_flags|= flags;
     DBUG_VOID_RETURN;
   }
+
   /**
-     Flag the current (top-level) statement as unsafe.
-     The flag will be reset after the statement has finished.
-   */
-  inline void set_stmt_unsafe() {
-    DBUG_ENTER("set_stmt_unsafe");
-    binlog_stmt_flags|= (1U << BINLOG_STMT_FLAG_UNSAFE);
-    DBUG_VOID_RETURN;
+    Return a binary combination of all unsafe warnings for the
+    statement.  If the statement has been marked as unsafe by the
+    'flag' member of enum_binlog_stmt_unsafe, then the return value
+    from this function has bit (1<<flag) set to 1.
+  */
+  inline uint32 get_stmt_unsafe_flags() const {
+    DBUG_ENTER("get_stmt_unsafe_flags");
+    DBUG_RETURN(binlog_stmt_flags & BINLOG_STMT_UNSAFE_ALL_FLAGS);
   }
 
+  /**
+    Mark the current statement as safe; i.e., clear all bits in
+    binlog_stmt_flags that correspond to elements of
+    enum_binlog_stmt_unsafe.
+  */
   inline void clear_stmt_unsafe() {
     DBUG_ENTER("clear_stmt_unsafe");
-    binlog_stmt_flags&= ~(1U << BINLOG_STMT_FLAG_UNSAFE);
+    binlog_stmt_flags&= ~BINLOG_STMT_UNSAFE_ALL_FLAGS;
+    DBUG_VOID_RETURN;
+  }
+
+  /**
+    Determine if this statement is a row injection.
+
+    @retval 0 if the statement is not a row injection
+    @retval nonzero if the statement is a row injection
+  */
+  inline bool is_stmt_row_injection() const {
+    return binlog_stmt_flags &
+      (1U << (BINLOG_STMT_UNSAFE_COUNT + BINLOG_STMT_TYPE_ROW_INJECTION));
+  }
+
+  /**
+    Flag the statement as a row injection.  A row injection is either
+    a BINLOG statement, or a row event in the relay log executed by
+    the slave SQL thread.
+  */
+  inline void set_stmt_row_injection() {
+    DBUG_ENTER("set_stmt_row_injection");
+    binlog_stmt_flags|=
+      (1U << (BINLOG_STMT_UNSAFE_COUNT + BINLOG_STMT_TYPE_ROW_INJECTION));
     DBUG_VOID_RETURN;
   }
 
@@ -1097,37 +1191,37 @@ public:
   { return sroutines_list.elements != 0; }
 
 private:
+
   /**
-    Flags indicating properties of the statement with respect to
-    logging.
+    Enumeration listing special types of statements.
 
-    These are combined in a binary manner; e.g., an unsafe statement
-    has the bit (1<<BINLOG_STMT_FLAG_UNSAFE) set.
+    Currently, the only possible type is ROW_INJECTION.
   */
-  enum enum_binlog_stmt_flag {
-    /** The statement is unsafe to log in statement mode. */
-    BINLOG_STMT_FLAG_UNSAFE= 0,
+  enum enum_binlog_stmt_type {
     /**
       The statement is a row injection (i.e., either a BINLOG
       statement or a row event executed by the slave SQL thread).
     */
-    BINLOG_STMT_FLAG_ROW_INJECTION,
-    /**
-      The last element of this enumeration type.  Insert new members
-      above.
-    */
-    BINLOG_STMT_FLAG_COUNT
+    BINLOG_STMT_TYPE_ROW_INJECTION = 0,
+
+    /** The last element of this enumeration type. */
+    BINLOG_STMT_TYPE_COUNT
   };
 
   /**
-    Indicates the type of statement with respect to binlogging.
+    Bit field indicating the type of statement.
+
+    There are two groups of bits:
+
+    - The low BINLOG_STMT_UNSAFE_COUNT bits indicate the types of
+      unsafeness that the current statement has.
 
-    This is typically zeroed before parsing a statement, set during
-    parsing (depending on the query), and read when deciding the
-    logging format of the current statement.
+    - The next BINLOG_STMT_TYPE_COUNT bits indicate if the statement
+      is of some special type.
 
-    This is a binary combination of one or more bits (1<<flag), where
-    flag is a member of enum_binlog_stmt_flag.
+    This must be a member of LEX, not of THD: each stored procedure
+    needs to remember its unsafeness state between calls and each
+    stored procedure has its own LEX object (but no own THD object).
   */
   uint32 binlog_stmt_flags;
 };
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 0f4812660e7..9bd330f1254 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -5609,19 +5609,18 @@ bool my_yyoverflow(short **yyss, YYSTYPE **yyvs, ulong *yystacksize)
 
 
 /**
- Reset THD part responsible for command processing state.
+  Reset the part of THD responsible for the state of command
+  processing.
 
-   This needs to be called before execution of every statement
-   (prepared or conventional).
-   It is not called by substatements of routines.
+  This needs to be called before execution of every statement
+  (prepared or conventional).  It is not called by substatements of
+  routines.
 
-  @todo
-   Make it a method of THD and align its name with the rest of
-   reset/end/start/init methods.
-  @todo
-   Call it after we use THD for queries, not before.
-*/
+  @todo Remove mysql_reset_thd_for_next_command and only use the
+  member function.
 
+  @todo Call it after we use THD for queries, not before.
+*/
 void mysql_reset_thd_for_next_command(THD *thd)
 {
   thd->reset_for_next_command();
@@ -5674,7 +5673,7 @@ void THD::reset_for_next_command()
   thd->sent_row_count= thd->examined_row_count= 0;
 
   thd->reset_current_stmt_binlog_row_based();
-  thd->binlog_warning_flags= 0;
+  thd->binlog_unsafe_warning_flags= 0;
 
   DBUG_PRINT("debug",
              ("current_stmt_binlog_row_based: %d",
diff --git a/sql/sql_view.cc b/sql/sql_view.cc
index 2a4c5c950fe..c285e40a148 100644
--- a/sql/sql_view.cc
+++ b/sql/sql_view.cc
@@ -1306,8 +1306,8 @@ bool mysql_make_view(THD *thd, File_parser *parser, TABLE_LIST *table,
       If the view's body needs row-based binlogging (e.g. the VIEW is created
       from SELECT UUID()), the top statement also needs it.
     */
-    if (lex->is_stmt_unsafe())
-      old_lex->set_stmt_unsafe();
+    old_lex->set_stmt_unsafe_flags(lex->get_stmt_unsafe_flags());
+
     view_is_mergeable= (table->algorithm != VIEW_ALGORITHM_TMPTABLE &&
                         lex->can_be_merged());
     LINT_INIT(view_main_select_tables);
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index c98fd93c1a7..bf8479894f6 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -7213,7 +7213,7 @@ function_call_keyword:
             $$= new (YYTHD->mem_root) Item_func_current_user(Lex->current_context());
             if ($$ == NULL)
               MYSQL_YYABORT;
-            Lex->set_stmt_unsafe();
+            Lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_FUNCTION);
             Lex->safe_to_cache_query= 0;
           }
         | DATE_SYM '(' expr ')'
@@ -7368,7 +7368,7 @@ function_call_keyword:
             $$= new (YYTHD->mem_root) Item_func_user();
             if ($$ == NULL)
               MYSQL_YYABORT;
-            Lex->set_stmt_unsafe();
+            Lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_FUNCTION);
             Lex->safe_to_cache_query=0;
           }
         | YEAR_SYM '(' expr ')'
@@ -8098,7 +8098,7 @@ 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();
+              Lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_SYSTEM_VARIABLE);
           }
         ;
 
@@ -8944,7 +8944,10 @@ opt_limit_clause:
         ;
 
 limit_clause:
-          LIMIT limit_options { Lex->set_stmt_unsafe(); }
+          LIMIT limit_options
+          {
+            Lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_LIMIT);
+          }
         ;
 
 limit_options:
@@ -9006,7 +9009,7 @@ delete_limit_clause:
           {
             SELECT_LEX *sel= Select;
             sel->select_limit= $2;
-            Lex->set_stmt_unsafe();
+            Lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_LIMIT);
             sel->explicit_limit= 1;
           }
         ;
@@ -9463,7 +9466,7 @@ insert_lock_option:
         | DELAYED_SYM
         {
           $$= TL_WRITE_DELAYED;
-          Lex->set_stmt_unsafe();
+          Lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_INSERT_DELAYED);
         }
         | HIGH_PRIORITY { $$= TL_WRITE; }
         ;
@@ -9473,7 +9476,7 @@ replace_lock_option:
         | DELAYED_SYM
         {
           $$= TL_WRITE_DELAYED;
-          Lex->set_stmt_unsafe();
+          Lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_INSERT_DELAYED);
         }
         ;