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

General overview:
The logic for switching to row format when binlog_format=MIXED had
numerous flaws. The underlying problem was the lack of a consistent
architecture.
General purpose of this changeset:
This changeset introduces an architecture for switching to row format
when binlog_format=MIXED. It enforces the architecture where it has
to. It leaves some bugs to be fixed later. It adds extensive tests to
verify that unsafe statements work as expected and that appropriate
errors are produced by problems with the selection of binlog format.
It was not practical to split this into smaller pieces of work.

Problem 1:
To determine the logging mode, the code has to take several parameters
into account (namely: (1) the value of binlog_format; (2) the
capabilities of the engines; (3) the type of the current statement:
normal, unsafe, or row injection). These parameters may conflict in
several ways, namely:
 - binlog_format=STATEMENT for a row injection
 - binlog_format=STATEMENT for an unsafe statement
 - binlog_format=STATEMENT for an engine only supporting row logging
 - binlog_format=ROW for an engine only supporting statement logging
 - statement is unsafe and engine does not support row logging
 - row injection in a table that does not support statement logging
 - statement modifies one table that does not support row logging and
   one that does not support statement logging
Several of these conflicts were not detected, or were detected with
an inappropriate error message. The problem of BUG#39934 was that no
appropriate error message was written for the case when an engine
only supporting row logging executed a row injection with
binlog_format=ROW. However, all above cases must be handled.
Fix 1:
Introduce new error codes (sql/share/errmsg.txt). Ensure that all
conditions are detected and handled in decide_logging_format()

Problem 2:
The binlog format shall be determined once per statement, in
decide_logging_format(). It shall not be changed before or after that.
Before decide_logging_format() is called, all information necessary to
determine the logging format must be available. This principle ensures
that all unsafe statements are handled in a consistent way.
However, this principle is not followed:
thd->set_current_stmt_binlog_row_based_if_mixed() is called in several
places, including from code executing UPDATE..LIMIT,
INSERT..SELECT..LIMIT, DELETE..LIMIT, INSERT DELAYED, and
SET @@binlog_format. After Problem 1 was fixed, that caused
inconsistencies where these unsafe statements would not print the
appropriate warnings or errors for some of the conflicts.
Fix 2:
Remove calls to THD::set_current_stmt_binlog_row_based_if_mixed() from
code executed after decide_logging_format(). Compensate by calling the
set_current_stmt_unsafe() at parse time. This way, all unsafe statements
are detected by decide_logging_format().

Problem 3:
INSERT DELAYED is not unsafe: it is logged in statement format even if
binlog_format=MIXED, and no warning is printed even if
binlog_format=STATEMENT. This is BUG#45825.
Fix 3:
Made INSERT DELAYED set itself to unsafe at parse time. This allows
decide_logging_format() to detect that a warning should be printed or
the binlog_format changed.

Problem 4:
LIMIT clause were not marked as unsafe when executed inside stored
functions/triggers/views/prepared statements. This is
BUG#45785.
Fix 4:
Make statements containing the LIMIT clause marked as unsafe at
parse time, instead of at execution time. This allows propagating
unsafe-ness to the view.

30 files changed, 525 insertions, 295 deletions

diff --git a/sql/event_db_repository.cc b/sql/event_db_repository.cc
index 9a253d74546..a8d634d76e1 100644
--- a/sql/event_db_repository.cc
+++ b/sql/event_db_repository.cc
@@ -1034,7 +1034,7 @@ update_timing_fields_for_event(THD *thd,
     Turn off row binlogging of event timing updates. These are not used
     for RBR of events replicated to the slave.
   */
-  if (thd->current_stmt_binlog_row_based)
+  if (thd->is_current_stmt_binlog_format_row())
     thd->clear_current_stmt_binlog_row_based();
 
   DBUG_ASSERT(thd->security_ctx->master_access & SUPER_ACL);
diff --git a/sql/events.cc b/sql/events.cc
index ea935e67bd3..b8c3972d1c3 100644
--- a/sql/events.cc
+++ b/sql/events.cc
@@ -403,7 +403,7 @@ Events::create_event(THD *thd, Event_parse_data *parse_data,
     Turn off row binlogging of this statement and use statement-based 
     so that all supporting tables are updated for CREATE EVENT command.
   */
-  if (thd->current_stmt_binlog_row_based)
+  if (thd->is_current_stmt_binlog_format_row())
     thd->clear_current_stmt_binlog_row_based();
 
   pthread_mutex_lock(&LOCK_event_metadata);
@@ -527,7 +527,7 @@ Events::update_event(THD *thd, Event_parse_data *parse_data,
     Turn off row binlogging of this statement and use statement-based 
     so that all supporting tables are updated for UPDATE EVENT command.
   */
-  if (thd->current_stmt_binlog_row_based)
+  if (thd->is_current_stmt_binlog_format_row())
     thd->clear_current_stmt_binlog_row_based();
 
   pthread_mutex_lock(&LOCK_event_metadata);
@@ -624,7 +624,7 @@ Events::drop_event(THD *thd, LEX_STRING dbname, LEX_STRING name, bool if_exists)
     Turn off row binlogging of this statement and use statement-based so
     that all supporting tables are updated for DROP EVENT command.
   */
-  if (thd->current_stmt_binlog_row_based)
+  if (thd->is_current_stmt_binlog_format_row())
     thd->clear_current_stmt_binlog_row_based();
 
   pthread_mutex_lock(&LOCK_event_metadata);
diff --git a/sql/ha_ndbcluster_binlog.cc b/sql/ha_ndbcluster_binlog.cc
index a0c74b60f84..5346ef1d626 100644
--- a/sql/ha_ndbcluster_binlog.cc
+++ b/sql/ha_ndbcluster_binlog.cc
@@ -304,6 +304,7 @@ static void run_query(THD *thd, char *buf, char *end,
   thd->transaction.all= save_thd_transaction_all;
   thd->transaction.stmt= save_thd_transaction_stmt;
   thd->net= save_thd_net;
+  thd->set_current_stmt_binlog_row_based();
 
   if (thd == injector_thd)
   {
@@ -3649,6 +3650,7 @@ pthread_handler_t ndb_binlog_thread_func(void *arg)
 
   thd= new THD; /* note that contructor of THD uses DBUG_ */
   THD_CHECK_SENTRY(thd);
+  thd->set_current_stmt_binlog_row_based();
 
   /* We need to set thd->thread_id before thd->store_globals, or it will
      set an invalid value for thd->variables.pseudo_thread_id.
diff --git a/sql/ha_partition.cc b/sql/ha_partition.cc
index 67bc3156260..97a8dd99201 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -6188,7 +6188,7 @@ void ha_partition::get_auto_increment(ulonglong offset, ulonglong increment,
     if (!auto_increment_safe_stmt_log_lock &&
         thd->lex->sql_command != SQLCOM_INSERT &&
         mysql_bin_log.is_open() &&
-        !thd->current_stmt_binlog_row_based &&
+        !thd->is_current_stmt_binlog_format_row() &&
         (thd->options & OPTION_BIN_LOG))
     {
       DBUG_PRINT("info", ("locking auto_increment_safe_stmt_log_lock"));
diff --git a/sql/handler.cc b/sql/handler.cc
index 058a43eed8d..c61d012e552 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -2412,7 +2412,7 @@ int handler::update_auto_increment()
                                           variables->auto_increment_increment);
     auto_inc_intervals_count++;
     /* Row-based replication does not need to store intervals in binlog */
-    if (mysql_bin_log.is_open() && !thd->current_stmt_binlog_row_based)
+    if (mysql_bin_log.is_open() && !thd->is_current_stmt_binlog_format_row())
         thd->auto_inc_intervals_in_cur_stmt_for_binlog.append(auto_inc_interval_for_cur_row.minimum(),
                                                               auto_inc_interval_for_cur_row.values(),
                                                               variables->auto_increment_increment);
@@ -4423,7 +4423,7 @@ static bool check_table_binlog_row_based(THD *thd, TABLE *table)
   DBUG_ASSERT(table->s->cached_row_logging_check == 0 ||
               table->s->cached_row_logging_check == 1);
 
-  return (thd->current_stmt_binlog_row_based &&
+  return (thd->is_current_stmt_binlog_format_row() &&
           table->s->cached_row_logging_check &&
           (thd->options & OPTION_BIN_LOG) &&
           mysql_bin_log.is_open());
diff --git a/sql/item_create.cc b/sql/item_create.cc
index bf359b10caa..23a921be91c 100644
--- a/sql/item_create.cc
+++ b/sql/item_create.cc
@@ -2375,6 +2375,7 @@ Create_udf_func::create(THD *thd, udf_func *udf, List<Item> *item_list)
   Item *func= NULL;
   int arg_count= 0;
 
+  DBUG_ENTER("Create_udf_func::create");
   if (item_list != NULL)
     arg_count= item_list->elements;
 
@@ -2462,7 +2463,7 @@ Create_udf_func::create(THD *thd, udf_func *udf, List<Item> *item_list)
   }
   }
   thd->lex->safe_to_cache_query= 0;
-  return func;
+  DBUG_RETURN(func);
 }
 #endif
 
@@ -3363,9 +3364,10 @@ Create_func_found_rows Create_func_found_rows::s_singleton;
 Item*
 Create_func_found_rows::create(THD *thd)
 {
+  DBUG_ENTER("Create_func_found_rows::create");
   thd->lex->set_stmt_unsafe();
   thd->lex->safe_to_cache_query= 0;
-  return new (thd->mem_root) Item_func_found_rows();
+  DBUG_RETURN(new (thd->mem_root) Item_func_found_rows());
 }
 
 
@@ -3791,9 +3793,10 @@ Create_func_load_file Create_func_load_file::s_singleton;
 Item*
 Create_func_load_file::create(THD *thd, Item *arg1)
 {
+  DBUG_ENTER("Create_func_load_file::create");
   thd->lex->set_stmt_unsafe();
   thd->lex->uncacheable(UNCACHEABLE_SIDEEFFECT);
-  return new (thd->mem_root) Item_load_file(arg1);
+  DBUG_RETURN(new (thd->mem_root) Item_load_file(arg1));
 }
 
 
@@ -4260,9 +4263,10 @@ Create_func_row_count Create_func_row_count::s_singleton;
 Item*
 Create_func_row_count::create(THD *thd)
 {
+  DBUG_ENTER("Create_func_row_count::create");
   thd->lex->set_stmt_unsafe();
   thd->lex->safe_to_cache_query= 0;
-  return new (thd->mem_root) Item_func_row_count();
+  DBUG_RETURN(new (thd->mem_root) Item_func_row_count());
 }
 
 
@@ -4569,9 +4573,10 @@ Create_func_uuid Create_func_uuid::s_singleton;
 Item*
 Create_func_uuid::create(THD *thd)
 {
+  DBUG_ENTER("Create_func_uuid::create");
   thd->lex->set_stmt_unsafe();
   thd->lex->safe_to_cache_query= 0;
-  return new (thd->mem_root) Item_func_uuid();
+  DBUG_RETURN(new (thd->mem_root) Item_func_uuid());
 }
 
 
@@ -4580,9 +4585,10 @@ Create_func_uuid_short Create_func_uuid_short::s_singleton;
 Item*
 Create_func_uuid_short::create(THD *thd)
 {
+  DBUG_ENTER("Create_func_uuid_short::create");
   thd->lex->set_stmt_unsafe();
   thd->lex->safe_to_cache_query= 0;
-  return new (thd->mem_root) Item_func_uuid_short();
+  DBUG_RETURN(new (thd->mem_root) Item_func_uuid_short());
 }
 
 
diff --git a/sql/log.cc b/sql/log.cc
index ee7ee48b42c..ef7d5c75f84 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -3734,7 +3734,7 @@ int THD::binlog_write_table_map(TABLE *table, bool is_trans)
                        table->s->table_map_id));
 
   /* Pre-conditions */
-  DBUG_ASSERT(current_stmt_binlog_row_based && mysql_bin_log.is_open());
+  DBUG_ASSERT(is_current_stmt_binlog_format_row() && mysql_bin_log.is_open());
   DBUG_ASSERT(table->s->table_map_id != ULONG_MAX);
 
   Table_map_log_event::flag_set const
@@ -4009,7 +4009,7 @@ bool MYSQL_BIN_LOG::write(Log_event *event_info)
     */
     if (thd)
     {
-      if (!thd->current_stmt_binlog_row_based)
+      if (!thd->is_current_stmt_binlog_format_row())
       {
         if (thd->stmt_depends_on_first_successful_insert_id_in_prev_stmt)
         {
diff --git a/sql/log_event.cc b/sql/log_event.cc
index 7dd05c3f703..1c33fb86290 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -7187,16 +7187,12 @@ int Rows_log_event::do_apply_event(Relay_log_info const *rli)
     */
     thd->transaction.stmt.modified_non_trans_table= FALSE;
     /*
-      Check if the slave is set to use SBR.  If so, it should switch
-      to using RBR until the end of the "statement", i.e., next
-      STMT_END_F or next error.
+      This is a row injection, so we flag the "statement" as
+      such. Note that this code is called both when the slave does row
+      injections and when the BINLOG statement is used to do row
+      injections.
     */
-    if (!thd->current_stmt_binlog_row_based &&
-        mysql_bin_log.is_open() && (thd->options & OPTION_BIN_LOG))
-    {
-      thd->set_current_stmt_binlog_row_based();
-    }
-
+    thd->lex->set_stmt_row_injection();
 
     /*
       There are a few flags that are replicated with each row event.
@@ -7444,6 +7440,13 @@ int Rows_log_event::do_apply_event(Relay_log_info const *rli)
     slave_rows_error_report(ERROR_LEVEL, error, rli, thd, table,
                              get_type_str(),
                              RPL_LOG_NAME, (ulong) log_pos);
+    /*
+      @todo We should probably not call
+      reset_current_stmt_binlog_row_based() from here.
+
+      Note: this applies to log_event_old.cc too.
+      /Sven
+    */
     thd->reset_current_stmt_binlog_row_based();
     const_cast<Relay_log_info*>(rli)->cleanup_context(thd, error);
     thd->is_slave_error= 1;
@@ -7545,6 +7548,16 @@ static int rows_event_stmt_cleanup(Relay_log_info const *rli, THD * thd)
       event flushed.
     */
 
+    /*
+      @todo We should probably not call
+      reset_current_stmt_binlog_row_based() from here.
+
+      Note: this applies to log_event_old.cc too
+
+      Btw, the previous comment about transactional engines does not
+      seem related to anything that happens here.
+      /Sven
+    */
     thd->reset_current_stmt_binlog_row_based();
 
     const_cast<Relay_log_info*>(rli)->cleanup_context(thd, 0);
diff --git a/sql/log_event_old.cc b/sql/log_event_old.cc
index 68cd2bf4673..e76dbd257a7 100644
--- a/sql/log_event_old.cc
+++ b/sql/log_event_old.cc
@@ -66,15 +66,12 @@ Old_rows_log_event::do_apply_event(Old_rows_log_event *ev, const Relay_log_info
     mysql_reset_thd_for_next_command(thd);
 
     /*
-      Check if the slave is set to use SBR.  If so, it should switch
-      to using RBR until the end of the "statement", i.e., next
-      STMT_END_F or next error.
+      This is a row injection, so we flag the "statement" as
+      such. Note that this code is called both when the slave does row
+      injections and when the BINLOG statement is used to do row
+      injections.
     */
-    if (!thd->current_stmt_binlog_row_based &&
-        mysql_bin_log.is_open() && (thd->options & OPTION_BIN_LOG))
-    {
-      thd->set_current_stmt_binlog_row_based();
-    }
+    thd->lex->set_stmt_row_injection();
 
     if (simple_open_n_lock_tables(thd, rli->tables_to_lock))
     {
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index de02abc48e3..71fa55ff7a4 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -1559,7 +1559,6 @@ TABLE *open_n_lock_single_table(THD *thd, TABLE_LIST *table_l,
                                 thr_lock_type lock_type);
 bool open_normal_and_derived_tables(THD *thd, TABLE_LIST *tables, uint flags);
 int lock_tables(THD *thd, TABLE_LIST *tables, uint counter, bool *need_reopen);
-int decide_logging_format(THD *thd, TABLE_LIST *tables);
 TABLE *open_temporary_table(THD *thd, const char *path, const char *db,
 			    const char *table_name, bool link_in_list);
 bool rm_temporary_table(handlerton *base, char *path);
diff --git a/sql/rpl_injector.cc b/sql/rpl_injector.cc
index 684655d1c3b..8cdee89e164 100644
--- a/sql/rpl_injector.cc
+++ b/sql/rpl_injector.cc
@@ -36,8 +36,6 @@ injector::transaction::transaction(MYSQL_BIN_LOG *log, THD *thd)
   m_start_pos.m_file_pos= log_info.pos;
 
   begin_trans(m_thd);
-
-  thd->set_current_stmt_binlog_row_based();
 }
 
 injector::transaction::~transaction()
diff --git a/sql/set_var.cc b/sql/set_var.cc
index 0b89333ce03..846aabbab57 100644
--- a/sql/set_var.cc
+++ b/sql/set_var.cc
@@ -1292,6 +1292,11 @@ bool sys_var_thd_binlog_format::is_readonly() const
 
 void fix_binlog_format_after_update(THD *thd, enum_var_type type)
 {
+  /*
+    @todo This function should be eliminated.  We should not set the
+    current binlog format anywhere else than in decide_logging_format.
+    /Sven
+  */
   thd->reset_current_stmt_binlog_row_based();
 }
 
diff --git a/sql/share/errmsg.txt b/sql/share/errmsg.txt
index 42bca02984d..1f466dd3db6 100644
--- a/sql/share/errmsg.txt
+++ b/sql/share/errmsg.txt
@@ -6076,8 +6076,7 @@ ER_SLAVE_INCIDENT
 ER_NO_PARTITION_FOR_GIVEN_VALUE_SILENT
         eng "Table has no partition for some existing values"
 ER_BINLOG_UNSAFE_STATEMENT
-        eng "Statement may not be safe to log in statement format."
-        swe "Detta är inte säkert att logga i statement-format."
+        eng "Unsafe statement binlogged in statement format since BINLOG_FORMAT = STATEMENT."
 ER_SLAVE_FATAL_ERROR
         eng "Fatal error: %s"
 ER_SLAVE_RELAY_LOG_READ_FAILURE
@@ -6201,3 +6200,18 @@ ER_TEMPORARY_NAME
 ER_RENAMED_NAME
   eng "Renamed"
   swe "Namnändrad"
+
+ER_BINLOG_ROW_ENGINE_AND_STMT_ENGINE
+  eng "Cannot execute statement: binlogging impossible since both row-incapable engines and statement-incapable engines are involved."
+ER_BINLOG_ROW_MODE_AND_STMT_ENGINE
+  eng "Cannot execute statement: binlogging impossible since BINLOG_FORMAT = ROW and at least one table uses a storage engine limited to statement-logging."
+ER_BINLOG_UNSAFE_AND_STMT_ENGINE
+  eng "Unsafe statement binlogged as statement since storage engine is limited to statement-logging."
+ER_BINLOG_ROW_INJECTION_AND_STMT_ENGINE
+  eng "Cannot execute row injection: binlogging impossible since at least one table uses a storage engine limited to statement-logging."
+ER_BINLOG_STMT_MODE_AND_ROW_ENGINE
+  eng "Cannot execute statement: binlogging impossible since BINLOG_FORMAT = STATEMENT and at least one table uses a storage engine limited to row-logging.%s"
+ER_BINLOG_ROW_INJECTION_AND_STMT_MODE
+  eng "Cannot execute row injection: binlogging impossible since BINLOG_FORMAT = STATEMENT."
+ER_BINLOG_MULTIPLE_ENGINES_AND_SELF_LOGGING_ENGINE
+  eng "Cannot execute statement: binlogging impossible since more than one engine is involved and at least one engine is self-logging."
diff --git a/sql/sp.cc b/sql/sp.cc
index 29e228f5e45..a0c34e2bc1b 100644
--- a/sql/sp.cc
+++ b/sql/sp.cc
@@ -939,7 +939,7 @@ sp_create_routine(THD *thd, int type, sp_head *sp)
       /* restore sql_mode when binloging */
       thd->variables.sql_mode= saved_mode;
       /* Such a statement can always go directly to binlog, no trans cache */
-      thd->binlog_query(THD::MYSQL_QUERY_TYPE,
+      thd->binlog_query(THD::STMT_QUERY_TYPE,
                         log_query.c_ptr(), log_query.length(),
                         FALSE, FALSE, 0);
       thd->variables.sql_mode= 0;
@@ -1830,6 +1830,8 @@ sp_cache_routines_and_add_tables_for_triggers(THD *thd, LEX *lex,
 {
   int ret= 0;
 
+  DBUG_ENTER("sp_cache_routines_and_add_tables_for_triggers");
+
   Sroutine_hash_entry **last_cached_routine_ptr=
     (Sroutine_hash_entry **)lex->sroutines_list.next;
 
@@ -1863,7 +1865,7 @@ sp_cache_routines_and_add_tables_for_triggers(THD *thd, LEX *lex,
   ret= sp_cache_routines_and_add_tables_aux(thd, lex,
                                             *last_cached_routine_ptr,
                                             FALSE);
-  return ret;
+  DBUG_RETURN(ret);
 }
 
 
diff --git a/sql/sp_head.cc b/sql/sp_head.cc
index 8a8a5b06cc1..629a24585b3 100644
--- a/sql/sp_head.cc
+++ b/sql/sp_head.cc
@@ -1692,7 +1692,7 @@ sp_head::execute_function(THD *thd, Item **argp, uint argcount,
     each substatement be binlogged its way.
   */
   need_binlog_call= mysql_bin_log.is_open() &&
-    (thd->options & OPTION_BIN_LOG) && !thd->current_stmt_binlog_row_based;
+    (thd->options & OPTION_BIN_LOG) && !thd->is_current_stmt_binlog_format_row();
 
   /*
     Remember the original arguments for unrolled replication of functions
diff --git a/sql/sp_head.h b/sql/sp_head.h
index dd11f8693ac..7faccb7afbb 100644
--- a/sql/sp_head.h
+++ b/sql/sp_head.h
@@ -459,6 +459,7 @@ public:
   */
   void propagate_attributes(LEX *lex)
   {
+    DBUG_ENTER("sp_head::propagate_attributes");
     /*
       If this routine needs row-based binary logging, the entire top statement
       too (we cannot switch from statement-based to row-based only for this
@@ -467,6 +468,7 @@ public:
     */
     if (m_flags & BINLOG_ROW_BASED_IF_MIXED)
       lex->set_stmt_unsafe();
+    DBUG_VOID_RETURN;
   }
 
 
diff --git a/sql/sql_acl.cc b/sql/sql_acl.cc
index 79fc5d816fd..e614df4efcd 100644
--- a/sql/sql_acl.cc
+++ b/sql/sql_acl.cc
@@ -1654,7 +1654,7 @@ bool change_password(THD *thd, const char *host, const char *user,
                   acl_user->host.hostname ? acl_user->host.hostname : "",
                   new_password));
     thd->clear_error();
-    thd->binlog_query(THD::MYSQL_QUERY_TYPE, buff, query_length,
+    thd->binlog_query(THD::STMT_QUERY_TYPE, buff, query_length,
                       FALSE, FALSE, 0);
   }
 end:
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index de93c6609f7..a884a982011 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -1441,7 +1441,7 @@ void close_temporary_tables(THD *thd)
     return;
 
   if (!mysql_bin_log.is_open() || 
-      (thd->current_stmt_binlog_row_based && thd->variables.binlog_format == BINLOG_FORMAT_ROW))
+      (thd->is_current_stmt_binlog_format_row() && thd->variables.binlog_format == BINLOG_FORMAT_ROW))
   {
     TABLE *tmp_next;
     for (table= thd->temporary_tables; table; table= tmp_next)
@@ -4831,7 +4831,7 @@ static bool check_lock_and_start_stmt(THD *thd, TABLE *table,
 
     There may be more differences between open_n_lock_single_table() and
     open_ltable(). One known difference is that open_ltable() does
-    neither call decide_logging_format() nor handle some other logging
+    neither call thd->decide_logging_format() nor handle some other logging
     and locking issues because it does not call lock_tables().
 */
 
@@ -5052,63 +5052,142 @@ static void mark_real_tables_as_free_for_reuse(TABLE_LIST *table)
 
 
 /**
-   Decide on logging format to use for the statement.
+  Decide on logging format to use for the statement and issue errors
+  or warnings as needed.  The decision depends on the following
+  parameters:
 
-   Compute the capabilities vector for the involved storage engines
-   and mask out the flags for the binary log. Right now, the binlog
-   flags only include the capabilities of the storage engines, so this
-   is safe.
+  - The logging mode, i.e., the value of binlog_format.  Can be
+    statement, mixed, or row.
 
-   We now have three alternatives that prevent the statement from
-   being loggable:
+  - The type of statement.  There are three types of statements:
+    "normal" safe statements; unsafe statements; and row injections.
+    An unsafe statement is one that, if logged in statement format,
+    might produce different results when replayed on the slave (e.g.,
+    INSERT DELAYED).  A row injection is either a BINLOG statement, or
+    a row event executed by the slave's SQL thread.
 
-   1. If there are no capabilities left (all flags are clear) it is
-      not possible to log the statement at all, so we roll back the
-      statement and report an error.
+  - The capabilities of tables modified by the statement.  The
+    *capabilities vector* for a table is a set of flags associated
+    with the table.  Currently, it only includes two flags: *row
+    capability flag* and *statement capability flag*.
 
-   2. Statement mode is set, but the capabilities indicate that
-      statement format is not possible.
+    The row capability flag is set if and only if the engine can
+    handle row-based logging. The statement capability flag is set if
+    and only if the table can handle statement-based logging.
 
-   3. Row mode is set, but the capabilities indicate that row
-      format is not possible.
+  Decision table for logging format
+  ---------------------------------
 
-   4. Statement is unsafe, but the capabilities indicate that row
-      format is not possible.
+  The following table summarizes how the format and generated
+  warning/error depends on the tables' capabilities, the statement
+  type, and the current binlog_format.
 
-   If we are in MIXED mode, we then decide what logging format to use:
+     Row capable        N NNNNNNNNN YYYYYYYYY YYYYYYYYY
+     Statement capable  N YYYYYYYYY NNNNNNNNN YYYYYYYYY
 
-   1. If the statement is unsafe, row-based logging is used.
+     Statement type     * SSSUUUIII SSSUUUIII SSSUUUIII
 
-   2. If statement-based logging is not possible, row-based logging is
-      used.
+     binlog_format      * SMRSMRSMR SMRSMRSMR SMRSMRSMR
 
-   3. Otherwise, statement-based logging is used.
+     Logged format      - SS-SS---- -RR-RR-RR SRRSRR-RR
+     Warning/Error      1 --2332444 5--5--6-- ---7--6--
 
-   @param thd    Client thread
-   @param tables Tables involved in the query
- */
+  Legend
+  ------
+
+  Row capable:    N - Some table not row-capable, Y - All tables row-capable
+  Stmt capable:   N - Some table not stmt-capable, Y - All tables stmt-capable
+  Statement type: (S)afe, (U)nsafe, or Row (I)njection
+  binlog_format:  (S)TATEMENT, (M)IXED, or (R)OW
+  Logged format:  (S)tatement or (R)ow
+  Warning/Error:  Warnings and error messages are as follows:
+
+  1. Error: Cannot execute statement: binlogging impossible since both
+     row-incapable engines and statement-incapable engines are
+     involved.
+
+  2. Error: Cannot execute statement: binlogging impossible since
+     BINLOG_FORMAT = ROW and at least one table uses a storage engine
+     limited to statement-logging.
+
+  3. Warning: Unsafe statement binlogged as statement since storage
+     engine is limited to statement-logging.
+
+  4. Error: Cannot execute row injection: binlogging impossible since
+     at least one table uses a storage engine limited to
+     statement-logging.
+
+  5. Error: Cannot execute statement: binlogging impossible since
+     BINLOG_FORMAT = STATEMENT and at least one table uses a storage
+     engine limited to row-logging.
+
+  6. Error: Cannot execute row injection: binlogging impossible since
+     BINLOG_FORMAT = STATEMENT.
+
+  7. Warning: Unsafe statement binlogged in statement format since
+     BINLOG_FORMAT = STATEMENT.
+
+  In addition, we can produce the following error (not depending on
+  the variables of the decision diagram):
+
+  8. Error: Cannot execute statement: binlogging impossible since more
+     than one engine is involved and at least one engine is
+     self-logging.
 
-int decide_logging_format(THD *thd, TABLE_LIST *tables)
+  For each error case above, the statement is prevented from being
+  logged, we report an error, and roll back the statement.  For
+  warnings, we set the thd->binlog_flags variable: the warning will be
+  printed only if the statement is successfully logged.
+
+  @see THD::binlog_query
+
+  @param[in] thd    Client thread
+  @param[in] tables Tables involved in the query
+
+  @retval 0 No error; statement can be logged.
+  @retval -1 One of the error conditions above applies (1, 2, 4, 5, or 6).
+*/
+
+int THD::decide_logging_format(TABLE_LIST *tables)
 {
-  if (mysql_bin_log.is_open() && (thd->options & OPTION_BIN_LOG))
+  DBUG_ENTER("THD::decide_logging_format");
+  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.
-     */
+    */
     handler::Table_flags flags_some_set= 0;
     handler::Table_flags flags_all_set=
       HA_BINLOG_ROW_CAPABLE | HA_BINLOG_STMT_CAPABLE;
 
     my_bool multi_engine= FALSE;
     void* prev_ht= NULL;
+
+#ifndef DBUG_OFF
+    {
+      static const char *prelocked_mode_name[] = {
+        "NON_PRELOCKED",
+        "PRELOCKED",
+        "PRELOCKED_UNDER_LOCK_TABLES",
+      };
+      DBUG_PRINT("debug", ("prelocked_mode: %s",
+                           prelocked_mode_name[prelocked_mode]));
+    }
+#endif
+
+    /*
+      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.)
+    */
     for (TABLE_LIST *table= tables; table; table= table->next_global)
     {
       if (table->placeholder())
         continue;
       if (table->table->s->table_category == TABLE_CATEGORY_PERFORMANCE)
-        thd->lex->set_stmt_unsafe();
+        lex->set_stmt_unsafe();
       if (table->lock_type >= TL_WRITE_ALLOW_WRITE)
       {
         ulonglong const flags= table->table->file->ha_table_flags();
@@ -5130,75 +5209,116 @@ int decide_logging_format(THD *thd, 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", ("thd->variables.binlog_format: %ld",
-                        thd->variables.binlog_format));
+    DBUG_PRINT("info", ("variables.binlog_format: %ld",
+                        variables.binlog_format));
     DBUG_PRINT("info", ("multi_engine: %s",
                         multi_engine ? "TRUE" : "FALSE"));
 
     int error= 0;
-    if (flags_all_set == 0)
-    {
-      my_error((error= ER_BINLOG_LOGGING_IMPOSSIBLE), MYF(0),
-               "Statement cannot be logged to the binary log in"
-               " row-based nor statement-based format");
-    }
-    else if (thd->variables.binlog_format == BINLOG_FORMAT_STMT &&
-             (flags_all_set & HA_BINLOG_STMT_CAPABLE) == 0)
-    {
-      my_error((error= ER_BINLOG_LOGGING_IMPOSSIBLE), MYF(0),
-                "Statement-based format required for this statement,"
-                " but not allowed by this combination of engines");
-    }
-    else if ((thd->variables.binlog_format == BINLOG_FORMAT_ROW ||
-              thd->lex->is_stmt_unsafe()) &&
-             (flags_all_set & HA_BINLOG_ROW_CAPABLE) == 0)
-    {
-      my_error((error= ER_BINLOG_LOGGING_IMPOSSIBLE), MYF(0),
-                "Row-based format required for this statement,"
-                " but not allowed by this combination of engines");
-    }
 
     /*
       If more than one engine is involved in the statement and at
       least one is doing it's own logging (is *self-logging*), the
       statement cannot be logged atomically, so we generate an error
       rather than allowing the binlog to become corrupt.
-     */
+    */
     if (multi_engine &&
         (flags_some_set & HA_HAS_OWN_BINLOGGING))
     {
-      error= ER_BINLOG_LOGGING_IMPOSSIBLE;
-      my_error(error, MYF(0),
-               "Statement cannot be written atomically since more"
-               " than one engine involved and at least one engine"
-               " is self-logging");
+      my_error((error= ER_BINLOG_MULTIPLE_ENGINES_AND_SELF_LOGGING_ENGINE),
+               MYF(0));
     }
 
-    DBUG_PRINT("info", ("error: %d", error));
-
-    if (error)
-      return -1;
-
-    /*
-      We switch to row-based format if we are in mixed mode and one of
-      the following are true:
-
-      1. If the statement is unsafe
-      2. If statement format cannot be used
-
-      Observe that point to cannot be decided before the tables
-      involved in a statement has been checked, i.e., we cannot put
-      this code in reset_current_stmt_binlog_row_based(), it has to be
-      here.
-    */
-    if (thd->lex->is_stmt_unsafe() ||
-        (flags_all_set & HA_BINLOG_STMT_CAPABLE) == 0)
+    /* both statement-only and row-only engines involved */
+    if ((flags_all_set & (HA_BINLOG_STMT_CAPABLE | HA_BINLOG_ROW_CAPABLE)) == 0)
     {
-      thd->set_current_stmt_binlog_row_based_if_mixed();
+      /*
+        1. Error: Binary logging impossible since both row-incapable
+           engines and statement-incapable engines are involved
+      */
+      my_error((error= ER_BINLOG_ROW_ENGINE_AND_STMT_ENGINE), MYF(0));
+    }
+    /* statement-only engines involved */
+    else if ((flags_all_set & HA_BINLOG_ROW_CAPABLE) == 0)
+    {
+      if (lex->is_stmt_row_injection())
+      {
+        /*
+          4. Error: Cannot execute row injection since table uses
+             storage engine limited to statement-logging
+        */
+        my_error((error= ER_BINLOG_ROW_INJECTION_AND_STMT_ENGINE), MYF(0));
+      }
+      else if (variables.binlog_format == BINLOG_FORMAT_ROW)
+      {
+        /*
+          2. Error: Cannot modify table that uses a storage engine
+             limited to statement-logging when BINLOG_FORMAT = ROW
+        */
+        my_error((error= ER_BINLOG_ROW_MODE_AND_STMT_ENGINE), MYF(0));
+      }
+      else if (lex->is_stmt_unsafe())
+      {
+        /*
+          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);
+      }
+      /* log in statement format! */
+    }
+    /* no statement-only engines */
+    else
+    {
+      /* binlog_format = STATEMENT */
+      if (variables.binlog_format == BINLOG_FORMAT_STMT)
+      {
+        if (lex->is_stmt_row_injection())
+        {
+          /*
+            6. Error: Cannot execute row injection since
+               BINLOG_FORMAT = STATEMENT
+          */
+          my_error((error= ER_BINLOG_ROW_INJECTION_AND_STMT_MODE), MYF(0));
+        }
+        else if ((flags_all_set & HA_BINLOG_STMT_CAPABLE) == 0)
+        {
+          /*
+            5. Error: Cannot modify table that uses a storage engine
+               limited to row-logging when binlog_format = STATEMENT
+          */
+          my_error((error= ER_BINLOG_STMT_MODE_AND_ROW_ENGINE), MYF(0), "");
+        }
+        else if (lex->is_stmt_unsafe())
+        {
+          /*
+            7. Warning: Unsafe statement logged as statement due to
+               binlog_format = STATEMENT
+          */
+          binlog_warning_flags|=
+            (1 << BINLOG_WARNING_FLAG_UNSAFE_AND_STMT_MODE);
+        }
+        /* log in statement format! */
+      }
+      /* No statement-only engines and binlog_format != STATEMENT.
+         I.e., nothing prevents us from row logging if needed. */
+      else
+      {
+        if (lex->is_stmt_unsafe() || lex->is_stmt_row_injection()
+            || (flags_all_set & HA_BINLOG_STMT_CAPABLE) == 0)
+        {
+          /* log in row format! */
+          set_current_stmt_binlog_row_based_if_mixed();
+        }
+      }
     }
+
+    if (error)
+      DBUG_RETURN(-1);
   }
 
-  return 0;
+  DBUG_RETURN(0);
 }
 
 /*
@@ -5242,7 +5362,7 @@ int lock_tables(THD *thd, TABLE_LIST *tables, uint count, bool *need_reopen)
   *need_reopen= FALSE;
 
   if (!tables && !thd->lex->requires_prelocking())
-    DBUG_RETURN(decide_logging_format(thd, tables));
+    DBUG_RETURN(thd->decide_logging_format(tables));
 
   /*
     We need this extra check for thd->prelocked_mode because we want to avoid
@@ -5402,7 +5522,7 @@ int lock_tables(THD *thd, TABLE_LIST *tables, uint count, bool *need_reopen)
     }
   }
 
-  DBUG_RETURN(decide_logging_format(thd, tables));
+  DBUG_RETURN(thd->decide_logging_format(tables));
 }
 
 
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index f1ad410b877..145fb8b60d7 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -539,7 +539,7 @@ THD::THD()
    lock_id(&main_lock_id),
    user_time(0), in_sub_stmt(0),
    sql_log_bin_toplevel(false),
-   binlog_table_maps(0), binlog_flags(0UL),
+   binlog_warning_flags(0UL), binlog_table_maps(0),
    table_map_for_update(0),
    arg_of_last_insert_id_function(FALSE),
    first_successful_insert_id_in_prev_stmt(0),
@@ -2973,13 +2973,13 @@ void THD::reset_sub_statement_state(Sub_statement_state *backup,
     first_successful_insert_id_in_cur_stmt;
 
   if ((!lex->requires_prelocking() || is_update_query(lex->sql_command)) &&
-      !current_stmt_binlog_row_based)
+      !is_current_stmt_binlog_format_row())
   {
     options&= ~OPTION_BIN_LOG;
   }
 
   if ((backup->options & OPTION_BIN_LOG) && is_update_query(lex->sql_command)&&
-      !current_stmt_binlog_row_based)
+      !is_current_stmt_binlog_format_row())
     mysql_bin_log.start_union_events(this, this->query_id);
 
   /* Disable result sets */
@@ -3041,7 +3041,7 @@ void THD::restore_sub_statement_state(Sub_statement_state *backup)
     is_fatal_sub_stmt_error= FALSE;
 
   if ((options & OPTION_BIN_LOG) && is_update_query(lex->sql_command) &&
-    !current_stmt_binlog_row_based)
+    !is_current_stmt_binlog_format_row())
     mysql_bin_log.stop_union_events(this);
 
   /*
@@ -3463,7 +3463,7 @@ int THD::binlog_write_row(TABLE* table, bool is_trans,
                           MY_BITMAP const* cols, size_t colcnt, 
                           uchar const *record) 
 { 
-  DBUG_ASSERT(current_stmt_binlog_row_based && mysql_bin_log.is_open());
+  DBUG_ASSERT(is_current_stmt_binlog_format_row() && mysql_bin_log.is_open());
 
   /*
     Pack records into format for transfer. We are allocating more
@@ -3493,7 +3493,7 @@ int THD::binlog_update_row(TABLE* table, bool is_trans,
                            const uchar *before_record,
                            const uchar *after_record)
 { 
-  DBUG_ASSERT(current_stmt_binlog_row_based && mysql_bin_log.is_open());
+  DBUG_ASSERT(is_current_stmt_binlog_format_row() && mysql_bin_log.is_open());
 
   size_t const before_maxlen = max_row_length(table, before_record);
   size_t const after_maxlen  = max_row_length(table, after_record);
@@ -3538,7 +3538,7 @@ int THD::binlog_delete_row(TABLE* table, bool is_trans,
                            MY_BITMAP const* cols, size_t colcnt,
                            uchar const *record)
 { 
-  DBUG_ASSERT(current_stmt_binlog_row_based && mysql_bin_log.is_open());
+  DBUG_ASSERT(is_current_stmt_binlog_format_row() && mysql_bin_log.is_open());
 
   /* 
      Pack records into format for transfer. We are allocating more
@@ -3620,8 +3620,6 @@ show_query_type(THD::enum_binlog_query_type qtype)
     return "ROW";
   case THD::STMT_QUERY_TYPE:
     return "STMT";
-  case THD::MYSQL_QUERY_TYPE:
-    return "MYSQL";
   case THD::QUERY_TYPE_COUNT:
   default:
     DBUG_ASSERT(0 <= qtype && qtype < THD::QUERY_TYPE_COUNT);
@@ -3633,28 +3631,30 @@ show_query_type(THD::enum_binlog_query_type qtype)
 #endif
 
 
-/*
-  Member function that will log query, either row-based or
-  statement-based depending on the value of the 'current_stmt_binlog_row_based'
-  the value of the 'qtype' flag.
+/**
+  Log the current query.
 
-  This function should be called after the all calls to ha_*_row()
-  functions have been issued, but before tables are unlocked and
-  closed.
+  The query will be logged in either row format or statement format
+  depending on the value of @c current_stmt_binlog_row_based field and
+  the value of the @c qtype parameter.
 
-  OBSERVE
-    There shall be no writes to any system table after calling
-    binlog_query(), so these writes has to be moved to before the call
-    of binlog_query() for correct functioning.
+  This function must be called:
 
-    This is necessesary not only for RBR, but the master might crash
-    after binlogging the query but before changing the system tables.
-    This means that the slave and the master are not in the same state
-    (after the master has restarted), so therefore we have to
-    eliminate this problem.
+  - After the all calls to ha_*_row() functions have been issued.
 
-  RETURN VALUE
-    Error code, or 0 if no error.
+  - After any writes to system tables. Rationale: if system tables
+    were written after a call to this function, and the master crashes
+    after the call to this function and before writing the system
+    tables, then the master and slave get out of sync.
+
+  - Before tables are unlocked and closed.
+
+  @see decide_logging_format
+
+  @retval 0 Success
+
+  @retval nonzero If there is a failure when writing the query (e.g.,
+  write failure), then the error code is returned.
 */
 int THD::binlog_query(THD::enum_binlog_query_type qtype, char const *query_arg,
                       ulong query_len, bool is_trans, bool suppress_use,
@@ -3679,50 +3679,69 @@ int THD::binlog_query(THD::enum_binlog_query_type qtype, char const *query_arg,
       DBUG_RETURN(error);
 
   /*
-    If we are in statement mode and trying to log an unsafe statement,
-    we should print a warning.
+    Warnings for unsafe statements logged in statement format are
+    printed here instead of in decide_logging_format().  This is
+    because the warnings should be printed only if the statement is
+    actually logged. When executing decide_logging_format(), we cannot
+    know for sure if the statement will be logged.
   */
-  if (sql_log_bin_toplevel && lex->is_stmt_unsafe() &&
-      variables.binlog_format == BINLOG_FORMAT_STMT)
+  if (sql_log_bin_toplevel)
   {
-   /*
-     A warning can be elevated a error when STRICT sql mode.
-     But we don't want to elevate binlog warning to error here.
-   */
-    push_warning(this, MYSQL_ERROR::WARN_LEVEL_NOTE,
-                 ER_BINLOG_UNSAFE_STATEMENT,
-                 ER(ER_BINLOG_UNSAFE_STATEMENT));
-    if (!(binlog_flags & BINLOG_FLAG_UNSAFE_STMT_PRINTED))
+    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_flags|= BINLOG_FLAG_UNSAFE_STMT_PRINTED;
+      binlog_warning_flags|= 1 << BINLOG_WARNING_FLAG_PRINTED;
     }
   }
 
   switch (qtype) {
+    /*
+      ROW_QUERY_TYPE means that the statement may be logged either in
+      row format or in statement format.  If
+      current_stmt_binlog_row_based is set, it means that the
+      statement has already been logged in row format and hence shall
+      not be logged again.
+    */
   case THD::ROW_QUERY_TYPE:
     DBUG_PRINT("debug",
                ("current_stmt_binlog_row_based: %d",
-                current_stmt_binlog_row_based));
-    if (current_stmt_binlog_row_based)
+                is_current_stmt_binlog_format_row()));
+    if (is_current_stmt_binlog_format_row())
       DBUG_RETURN(0);
-    /* Otherwise, we fall through */
-  case THD::MYSQL_QUERY_TYPE:
-    /*
-      Using this query type is a conveniece hack, since we have been
-      moving back and forth between using RBR for replication of
-      system tables and not using it.
+    /* Fall through */
 
-      Make sure to change in check_table_binlog_row_based() according
-      to how you treat this.
+    /*
+      STMT_QUERY_TYPE means that the query must be logged in statement
+      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.
     */
   case THD::STMT_QUERY_TYPE:
     /*
       The MYSQL_LOG::write() function will set the STMT_END_F flag and
       flush the pending rows event if necessary.
-     */
+    */
     {
       Query_log_event qinfo(this, query_arg, query_len, is_trans, suppress_use,
                             errcode);
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 36e696f2da6..5e069f1cbb6 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -1254,6 +1254,7 @@ public:
   /* Used to execute base64 coded binlog events in MySQL server */
   Relay_log_info* rli_fake;
 
+  void reset_for_next_command();
   /*
     Constant for THD::where initialization in the beginning of every query.
 
@@ -1421,23 +1422,44 @@ 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() {
+    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)
+  */
+  enum_binlog_format current_stmt_binlog_format;
+
+  enum enum_binlog_warning_flag {
+    /* 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
+  };
+  /**
+    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.
+
+    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.
+  */
+  uint32 binlog_warning_flags;
+
+  /*
     Number of outstanding table maps, i.e., table maps in the
     transaction cache.
   */
   uint binlog_table_maps;
-
-  enum enum_binlog_flag {
-    BINLOG_FLAG_UNSAFE_STMT_PRINTED,
-    BINLOG_FLAG_COUNT
-  };
-
-  /**
-     Flags with per-thread information regarding the status of the
-     binary log.
-   */
-  uint32 binlog_flags;
 public:
   uint get_binlog_table_maps() const {
     return binlog_table_maps;
@@ -1749,8 +1771,6 @@ public:
   char	     scramble[SCRAMBLE_LENGTH+1];
 
   bool       slave_thread, one_shot_set;
-  /* tells if current statement should binlog row-based(1) or stmt-based(0) */
-  bool       current_stmt_binlog_row_based;
   bool	     locked, some_tables_deleted;
   bool       last_cuted_field;
   bool	     no_errors, password;
@@ -1902,21 +1922,12 @@ public:
 
 #ifndef MYSQL_CLIENT
   enum enum_binlog_query_type {
-    /*
-      The query can be logged row-based or statement-based
-    */
+    /* The query can be logged in row format or in statement format. */
     ROW_QUERY_TYPE,
     
-    /*
-      The query has to be logged statement-based
-    */
+    /* The query has to be logged in statement format. */
     STMT_QUERY_TYPE,
     
-    /*
-      The query represents a change to a table in the "mysql"
-      database and is currently mapped to ROW_QUERY_TYPE.
-    */
-    MYSQL_QUERY_TYPE,
     QUERY_TYPE_COUNT
   };
   
@@ -2120,8 +2131,13 @@ public:
   void set_n_backup_active_arena(Query_arena *set, Query_arena *backup);
   void restore_active_arena(Query_arena *set, Query_arena *backup);
 
+  /*
+    @todo Make these methods private or remove them completely.  Only
+    decide_logging_format should call them. /Sven
+  */
   inline void set_current_stmt_binlog_row_based_if_mixed()
   {
+    DBUG_ENTER("set_current_stmt_binlog_row_based_if_mixed");
     /*
       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
@@ -2133,18 +2149,25 @@ public:
     */
     if ((variables.binlog_format == BINLOG_FORMAT_MIXED) &&
         (in_sub_stmt == 0))
-      current_stmt_binlog_row_based= TRUE;
+      current_stmt_binlog_format= BINLOG_FORMAT_ROW;
+
+    DBUG_VOID_RETURN;
   }
   inline void set_current_stmt_binlog_row_based()
   {
-    current_stmt_binlog_row_based= TRUE;
+    DBUG_ENTER("set_current_stmt_binlog_row_based");
+    current_stmt_binlog_format= BINLOG_FORMAT_ROW;
+    DBUG_VOID_RETURN;
   }
   inline void clear_current_stmt_binlog_row_based()
   {
-    current_stmt_binlog_row_based= FALSE;
+    DBUG_ENTER("clear_current_stmt_binlog_row_based");
+    current_stmt_binlog_format= BINLOG_FORMAT_STMT;
+    DBUG_VOID_RETURN;
   }
   inline void reset_current_stmt_binlog_row_based()
   {
+    DBUG_ENTER("reset_current_stmt_binlog_row_based");
     /*
       If there are temporary tables, don't reset back to
       statement-based. Indeed it could be that:
@@ -2159,19 +2182,18 @@ public:
       or trigger is decided when it starts executing, depending for example on
       the caller (for a stored function: if caller is SELECT or
       INSERT/UPDATE/DELETE...).
-
-      Don't reset binlog format for NDB binlog injector thread.
     */
     DBUG_PRINT("debug",
                ("temporary_tables: %s, in_sub_stmt: %s, system_thread: %s",
                 YESNO(temporary_tables), YESNO(in_sub_stmt),
                 show_system_thread(system_thread)));
-    if ((temporary_tables == NULL) && (in_sub_stmt == 0) &&
-        (system_thread != SYSTEM_THREAD_NDBCLUSTER_BINLOG))
+    if ((temporary_tables == NULL) && (in_sub_stmt == 0))
     {
-      current_stmt_binlog_row_based= 
-        test(variables.binlog_format == BINLOG_FORMAT_ROW);
+      current_stmt_binlog_format=
+        (variables.binlog_format == BINLOG_FORMAT_ROW) ? 
+        BINLOG_FORMAT_ROW : BINLOG_FORMAT_STMT;
     }
+    DBUG_VOID_RETURN;
   }
 
   /**
@@ -2267,7 +2289,11 @@ public:
   */
   void pop_internal_handler();
 
+  int decide_logging_format(TABLE_LIST *tables);
+
+
 private:
+
   /** The current internal error handler for this thread, or NULL. */
   Internal_error_handler *m_internal_handler;
   /**
diff --git a/sql/sql_delete.cc b/sql/sql_delete.cc
index 277269f3b0d..facf1992f82 100644
--- a/sql/sql_delete.cc
+++ b/sql/sql_delete.cc
@@ -131,7 +131,7 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
   if (!using_limit && const_cond_result &&
       !(specialflag & (SPECIAL_NO_NEW_FUNC | SPECIAL_SAFE_MODE)) &&
       (thd->lex->sql_command == SQLCOM_TRUNCATE ||
-       (!thd->current_stmt_binlog_row_based &&
+       (!thd->is_current_stmt_binlog_format_row() &&
         !(table->triggers && table->triggers->has_delete_triggers()))))
   {
     /* Update the table->file->stats.records number */
@@ -452,19 +452,6 @@ int mysql_prepare_delete(THD *thd, TABLE_LIST *table_list, Item **conds)
   DBUG_ENTER("mysql_prepare_delete");
   List<Item> all_fields;
 
-  /*
-    Statement-based replication of DELETE ... LIMIT is not safe as order of
-    rows is not defined, so in mixed mode we go to row-based.
-
-    Note that we may consider a statement as safe if ORDER BY primary_key
-    is present. However it may confuse users to see very similiar statements
-    replicated differently.
-  */
-  if (thd->lex->current_select->select_limit)
-  {
-    thd->lex->set_stmt_unsafe();
-    thd->set_current_stmt_binlog_row_based_if_mixed();
-  }
   thd->lex->allow_sum_func= 0;
   if (setup_tables_and_check_access(thd, &thd->lex->select_lex.context,
                                     &thd->lex->select_lex.top_join_list,
@@ -1023,7 +1010,7 @@ bool multi_delete::send_eof()
 
 static bool mysql_truncate_by_delete(THD *thd, TABLE_LIST *table_list)
 {
-  bool error, save_binlog_row_based= thd->current_stmt_binlog_row_based;
+  bool error, save_binlog_row_based= thd->is_current_stmt_binlog_format_row();
   DBUG_ENTER("mysql_truncate_by_delete");
   table_list->lock_type= TL_WRITE;
   mysql_init_select(thd->lex);
@@ -1031,7 +1018,8 @@ static bool mysql_truncate_by_delete(THD *thd, TABLE_LIST *table_list)
   error= mysql_delete(thd, table_list, NULL, NULL, HA_POS_ERROR, LL(0), TRUE);
   ha_autocommit_or_rollback(thd, error);
   end_trans(thd, error ? ROLLBACK : COMMIT);
-  thd->current_stmt_binlog_row_based= save_binlog_row_based;
+  if (save_binlog_row_based)
+    thd->set_current_stmt_binlog_row_based();
   DBUG_RETURN(error);
 }
 
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 9de27868d74..2cfa6ae6f59 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -1699,6 +1699,7 @@ public:
      table(0),tables_in_use(0),stacked_inserts(0), status(0), dead(0),
      group_count(0)
   {
+    DBUG_ENTER("Delayed_insert constructor");
     thd.security_ctx->user=thd.security_ctx->priv_user=(char*) delayed_user;
     thd.security_ctx->host=(char*) my_localhost;
     thd.current_tablenr=0;
@@ -1707,8 +1708,18 @@ public:
     thd.lex->current_select= 0; 		// for my_message_sql
     thd.lex->sql_command= SQLCOM_INSERT;        // For innodb::store_lock()
     /*
-      Statement-based replication of INSERT DELAYED has problems with RAND()
-      and user vars, so in mixed mode we go to row-based.
+      Statement-based replication of INSERT DELAYED has problems with
+      RAND() and user variables, so in mixed mode we go to row-based.
+      For normal commands, the unsafe flag is set at parse time.
+      However, since the flag is a member of the THD object, of which
+      the delayed_insert thread has its own copy, we must set the
+      statement to unsafe here and explicitly set row logging mode.
+
+      @todo set_current_stmt_binlog_row_based_if_mixed should not be
+      called by anything else than thd->decide_logging_format().  When
+      we call set_current_blah here, none of the checks in
+      decide_logging_format is made.  We should probably call
+      thd->decide_logging_format() directly instead.  /Sven
     */
     thd.lex->set_stmt_unsafe();
     thd.set_current_stmt_binlog_row_based_if_mixed();
@@ -1726,6 +1737,7 @@ public:
     delayed_lock= global_system_variables.low_priority_updates ?
                                           TL_WRITE_LOW_PRIORITY : TL_WRITE;
     VOID(pthread_mutex_unlock(&LOCK_thread_count));
+    DBUG_VOID_RETURN;
   }
   ~Delayed_insert()
   {
@@ -2305,12 +2317,6 @@ pthread_handler_t handle_delayed_insert(void *arg)
   */
   lex_start(thd);
   thd->lex->sql_command= SQLCOM_INSERT;        // For innodb::store_lock()
-  /*
-    Statement-based replication of INSERT DELAYED has problems with RAND()
-    and user vars, so in mixed mode we go to row-based.
-  */
-  thd->lex->set_stmt_unsafe();
-  thd->set_current_stmt_binlog_row_based_if_mixed();
 
   /* Open table */
   if (!(di->table= open_n_lock_single_table(thd, &di->table_list,
@@ -2756,7 +2762,7 @@ bool Delayed_insert::handle_inserts(void)
 
     TODO: Move the logging to last in the sequence of rows.
    */
-  if (thd.current_stmt_binlog_row_based)
+  if (thd.is_current_stmt_binlog_format_row())
     thd.binlog_flush_pending_rows_event(TRUE);
 
   if ((error=table->file->extra(HA_EXTRA_NO_CACHE)))
@@ -2821,19 +2827,6 @@ bool mysql_insert_select_prepare(THD *thd)
   DBUG_ENTER("mysql_insert_select_prepare");
 
   /*
-    Statement-based replication of INSERT ... SELECT ... LIMIT is not safe
-    as order of rows is not defined, so in mixed mode we go to row-based.
-
-    Note that we may consider a statement as safe if ORDER BY primary_key
-    is present or we SELECT a constant. However it may confuse users to
-    see very similiar statements replicated differently.
-  */
-  if (lex->current_select->select_limit)
-  {
-    lex->set_stmt_unsafe();
-    thd->set_current_stmt_binlog_row_based_if_mixed();
-  }
-  /*
     SELECT_LEX do not belong to INSERT statement, so we can't add WHERE
     clause if table is VIEW
   */
@@ -3286,7 +3279,7 @@ void select_insert::abort() {
           thd->binlog_query(THD::ROW_QUERY_TYPE, thd->query, thd->query_length,
                             transactional_table, FALSE, errcode);
         }
-        if (!thd->current_stmt_binlog_row_based && !can_rollback_data())
+        if (!thd->is_current_stmt_binlog_format_row() && !can_rollback_data())
           thd->transaction.all.modified_non_trans_table= TRUE;
 	if (changed)
 	  query_cache_invalidate3(thd, table, 1);
@@ -3550,11 +3543,11 @@ select_create::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
     virtual int do_postlock(TABLE **tables, uint count)
     {
       THD *thd= const_cast<THD*>(ptr->get_thd());
-      if (int error= decide_logging_format(thd, &all_tables))
+      if (int error= thd->decide_logging_format(&all_tables))
         return error;
 
       TABLE const *const table = *tables;
-      if (thd->current_stmt_binlog_row_based  &&
+      if (thd->is_current_stmt_binlog_format_row()  &&
           !table->s->tmp_table &&
           !ptr->get_create_info()->table_existed)
       {
@@ -3578,7 +3571,7 @@ select_create::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
     temporary table, we need to start a statement transaction.
   */
   if ((thd->lex->create_info.options & HA_LEX_CREATE_TMP_TABLE) == 0 &&
-      thd->current_stmt_binlog_row_based &&
+      thd->is_current_stmt_binlog_format_row() &&
       mysql_bin_log.is_open())
   {
     thd->binlog_start_trans_and_stmt();
@@ -3660,7 +3653,7 @@ select_create::binlog_show_create_table(TABLE **tables, uint count)
     schema that will do a close_thread_tables(), destroying the
     statement transaction cache.
   */
-  DBUG_ASSERT(thd->current_stmt_binlog_row_based);
+  DBUG_ASSERT(thd->is_current_stmt_binlog_format_row());
   DBUG_ASSERT(tables && *tables && count > 0);
 
   char buf[2048];
@@ -3700,7 +3693,7 @@ void select_create::send_error(uint errcode,const char *err)
 
   DBUG_PRINT("info",
              ("Current statement %s row-based",
-              thd->current_stmt_binlog_row_based ? "is" : "is NOT"));
+              thd->is_current_stmt_binlog_format_row() ? "is" : "is NOT"));
   DBUG_PRINT("info",
              ("Current table (at 0x%lu) %s a temporary (or non-existant) table",
               (ulong) table,
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index 22b7d2e359c..9528d8f8f29 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -1044,24 +1044,49 @@ public:
   }
 
   /**
-     Has the parser/scanner detected that this statement is unsafe?
-   */
+    Has the parser/scanner detected that this statement is 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);
   }
 
   /**
-     Flag the current (top-level) statement as unsafe.
+    Is this statement actually a row injection?
 
-     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
+  */
+  inline bool is_stmt_row_injection() const {
+    return binlog_stmt_flags & (1U << BINLOG_STMT_FLAG_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_FLAG_ROW_INJECTION);
+    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;
   }
 
   inline void clear_stmt_unsafe() {
+    DBUG_ENTER("clear_stmt_unsafe");
     binlog_stmt_flags&= ~(1U << BINLOG_STMT_FLAG_UNSAFE);
+    DBUG_VOID_RETURN;
   }
 
   /**
@@ -1072,16 +1097,37 @@ public:
   { return sroutines_list.elements != 0; }
 
 private:
+  /**
+    Flags indicating properties of the statement with respect to
+    logging.
+
+    These are combined in a binary manner; e.g., an unsafe statement
+    has the bit (1<<BINLOG_STMT_FLAG_UNSAFE) set.
+  */
   enum enum_binlog_stmt_flag {
-    BINLOG_STMT_FLAG_UNSAFE,
+    /** The statement is unsafe to log in statement mode. */
+    BINLOG_STMT_FLAG_UNSAFE= 0,
+    /**
+      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
   };
 
-  /*
-    Tells if the parsing stage detected properties of the statement,
-    for example: that some items require row-based binlogging to give
-    a reliable binlog/replication, or if we will use stored functions
-    or triggers which themselves need require row-based binlogging.
+  /**
+    Indicates the type of statement with respect to binlogging.
+
+    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.
+
+    This is a binary combination of one or more bits (1<<flag), where
+    flag is a member of enum_binlog_stmt_flag.
   */
   uint32 binlog_stmt_flags;
 };
@@ -1891,6 +1937,7 @@ typedef struct st_lex : public Query_tables_list
     }
     return FALSE;
   }
+
 } LEX;
 
 
diff --git a/sql/sql_load.cc b/sql/sql_load.cc
index babd02c0008..8bf71d55660 100644
--- a/sql/sql_load.cc
+++ b/sql/sql_load.cc
@@ -518,7 +518,7 @@ int mysql_load(THD *thd,sql_exchange *ex,TABLE_LIST *table_list,
       version for the binary log to mark that table maps are invalid
       after this point.
      */
-    if (thd->current_stmt_binlog_row_based)
+    if (thd->is_current_stmt_binlog_format_row())
       thd->binlog_flush_pending_rows_event(true);
     else
     {
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 83f44bba008..0f89b5ab5eb 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -5625,6 +5625,12 @@ bool my_yyoverflow(short **yyss, YYSTYPE **yyvs, ulong *yystacksize)
 
 void mysql_reset_thd_for_next_command(THD *thd)
 {
+  thd->reset_for_next_command();
+}
+
+void THD::reset_for_next_command()
+{
+  THD *thd= this;
   DBUG_ENTER("mysql_reset_thd_for_next_command");
   DBUG_ASSERT(!thd->spcont); /* not for substatements of routines */
   DBUG_ASSERT(! thd->in_sub_stmt);
@@ -5668,15 +5674,12 @@ void mysql_reset_thd_for_next_command(THD *thd)
   thd->rand_used= 0;
   thd->sent_row_count= thd->examined_row_count= 0;
 
-  /*
-    Because we come here only for start of top-statements, binlog format is
-    constant inside a complex statement (using stored functions) etc.
-  */
   thd->reset_current_stmt_binlog_row_based();
+  thd->binlog_warning_flags= 0;
 
   DBUG_PRINT("debug",
              ("current_stmt_binlog_row_based: %d",
-              thd->current_stmt_binlog_row_based));
+              thd->is_current_stmt_binlog_format_row()));
 
   DBUG_VOID_RETURN;
 }
diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc
index 06c6c022780..19df65bfd51 100644
--- a/sql/sql_repl.cc
+++ b/sql/sql_repl.cc
@@ -1684,7 +1684,7 @@ int log_loaded_block(IO_CACHE* file)
   uchar* buffer= (uchar*) my_b_get_buffer_start(file);
   uint max_event_size= current_thd->variables.max_allowed_packet;
   lf_info= (LOAD_FILE_INFO*) file->arg;
-  if (lf_info->thd->current_stmt_binlog_row_based)
+  if (lf_info->thd->is_current_stmt_binlog_format_row())
     DBUG_RETURN(0);
   if (lf_info->last_pos_in_file != HA_POS_ERROR &&
       lf_info->last_pos_in_file >= my_b_get_pos_in_file(file))
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 0fc1d04a41b..9badd43173c 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -1804,7 +1804,7 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
   LINT_INIT(alias);
   LINT_INIT(path_length);
 
-  if (thd->current_stmt_binlog_row_based && !dont_log_query)
+  if (thd->is_current_stmt_binlog_format_row() && !dont_log_query)
   {
     built_query.set_charset(system_charset_info);
     if (if_exists)
@@ -1882,7 +1882,7 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
       being built.  The string always end in a comma and the comma
       will be chopped off before being written to the binary log.
       */
-    if (thd->current_stmt_binlog_row_based && !dont_log_query)
+    if (thd->is_current_stmt_binlog_format_row() && !dont_log_query)
     {
       non_temp_tables_count++;
       /*
@@ -2009,7 +2009,7 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
     query_cache_invalidate3(thd, tables, 0);
     if (!dont_log_query)
     {
-      if (!thd->current_stmt_binlog_row_based ||
+      if (!thd->is_current_stmt_binlog_format_row() ||
           non_temp_tables_count > 0 && !tmp_table_deleted)
       {
         /*
@@ -2021,7 +2021,7 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
          */
         write_bin_log(thd, !error, thd->query, thd->query_length);
       }
-      else if (thd->current_stmt_binlog_row_based &&
+      else if (thd->is_current_stmt_binlog_format_row() &&
                non_temp_tables_count > 0 &&
                tmp_table_deleted)
       {
@@ -3822,8 +3822,8 @@ bool mysql_create_table_no_lock(THD *thd,
     Otherwise, the statement shall be binlogged.
    */
   if (!internal_tmp_table &&
-      (!thd->current_stmt_binlog_row_based ||
-       (thd->current_stmt_binlog_row_based &&
+      (!thd->is_current_stmt_binlog_format_row() ||
+       (thd->is_current_stmt_binlog_format_row() &&
         !(create_info->options & HA_LEX_CREATE_TMP_TABLE))))
     write_bin_log(thd, TRUE, thd->query, thd->query_length);
   error= FALSE;
@@ -5236,7 +5236,7 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table, TABLE_LIST* src_table,
   /*
     We have to write the query before we unlock the tables.
   */
-  if (thd->current_stmt_binlog_row_based)
+  if (thd->is_current_stmt_binlog_format_row())
   {
     /*
        Since temporary tables are not replicated under row-based
@@ -7143,7 +7143,7 @@ view_err:
     if (rename_temporary_table(thd, new_table, new_db, new_name))
       goto err1;
     /* We don't replicate alter table statement on temporary tables */
-    if (!thd->current_stmt_binlog_row_based)
+    if (!thd->is_current_stmt_binlog_format_row())
       write_bin_log(thd, TRUE, thd->query, thd->query_length);
     goto end_temporary;
   }
@@ -7305,7 +7305,7 @@ view_err:
                       db, table_name);
 
   DBUG_ASSERT(!(mysql_bin_log.is_open() &&
-                thd->current_stmt_binlog_row_based &&
+                thd->is_current_stmt_binlog_format_row() &&
                 (create_info->options & HA_LEX_CREATE_TMP_TABLE)));
   write_bin_log(thd, TRUE, thd->query, thd->query_length);
 
diff --git a/sql/sql_udf.cc b/sql/sql_udf.cc
index c60dac42fb8..a49e0df7124 100644
--- a/sql/sql_udf.cc
+++ b/sql/sql_udf.cc
@@ -437,7 +437,7 @@ int mysql_create_function(THD *thd,udf_func *udf)
     Turn off row binlogging of this statement and use statement-based 
     so that all supporting tables are updated for CREATE FUNCTION command.
   */
-  if (thd->current_stmt_binlog_row_based)
+  if (thd->is_current_stmt_binlog_format_row())
     thd->clear_current_stmt_binlog_row_based();
 
   rw_wrlock(&THR_LOCK_udf);
@@ -540,7 +540,7 @@ int mysql_drop_function(THD *thd,const LEX_STRING *udf_name)
     Turn off row binlogging of this statement and use statement-based
     so that all supporting tables are updated for DROP FUNCTION command.
   */
-  if (thd->current_stmt_binlog_row_based)
+  if (thd->is_current_stmt_binlog_format_row())
     thd->clear_current_stmt_binlog_row_based();
 
   rw_wrlock(&THR_LOCK_udf);  
diff --git a/sql/sql_update.cc b/sql/sql_update.cc
index ab6ba2ba756..638b15df414 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -871,19 +871,6 @@ bool mysql_prepare_update(THD *thd, TABLE_LIST *table_list,
   SELECT_LEX *select_lex= &thd->lex->select_lex;
   DBUG_ENTER("mysql_prepare_update");
 
-  /*
-    Statement-based replication of UPDATE ... LIMIT is not safe as order of
-    rows is not defined, so in mixed mode we go to row-based.
-
-    Note that we may consider a statement as safe if ORDER BY primary_key
-    is present. However it may confuse users to see very similiar statements
-    replicated differently.
-  */
-  if (thd->lex->current_select->select_limit)
-  {
-    thd->lex->set_stmt_unsafe();
-    thd->set_current_stmt_binlog_row_based_if_mixed();
-  }
 #ifndef NO_EMBEDDED_ACCESS_CHECKS
   table_list->grant.want_privilege= table->grant.want_privilege= 
     (SELECT_ACL & ~table->grant.privilege);
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index c40062e5d52..dfb5e9d5fcb 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -8939,7 +8939,7 @@ opt_limit_clause:
         ;
 
 limit_clause:
-          LIMIT limit_options {}
+          LIMIT limit_options { Lex->set_stmt_unsafe(); }
         ;
 
 limit_options:
@@ -9001,6 +9001,7 @@ delete_limit_clause:
           {
             SELECT_LEX *sel= Select;
             sel->select_limit= $2;
+            Lex->set_stmt_unsafe();
             sel->explicit_limit= 1;
           }
         ;
@@ -9454,13 +9455,21 @@ insert_lock_option:
 #endif
           }
         | LOW_PRIORITY  { $$= TL_WRITE_LOW_PRIORITY; }
-        | DELAYED_SYM   { $$= TL_WRITE_DELAYED; }
+        | DELAYED_SYM
+        {
+          $$= TL_WRITE_DELAYED;
+          Lex->set_stmt_unsafe();
+        }
         | HIGH_PRIORITY { $$= TL_WRITE; }
         ;
 
 replace_lock_option:
           opt_low_priority { $$= $1; }
-        | DELAYED_SYM { $$= TL_WRITE_DELAYED; }
+        | DELAYED_SYM
+        {
+          $$= TL_WRITE_DELAYED;
+          Lex->set_stmt_unsafe();
+        }
         ;
 
 insert2: